add function to s3

2026-01-10 05:36:15 +00:00
parent 7b91e0fd24
commit 8a3ff6a12c
19 changed files with 3715 additions and 134 deletions
--- a/backend/internal/object_storage/setup.go
+++ b/backend/internal/object_storage/setup.go
@@ -0,0 +1,511 @@
+package object_storage
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/atlasos/calypso/internal/common/database"
+	"github.com/atlasos/calypso/internal/common/logger"
+)
+
+// SetupService handles object storage setup operations
+type SetupService struct {
+	db     *database.DB
+	logger *logger.Logger
+}
+
+// NewSetupService creates a new setup service
+func NewSetupService(db *database.DB, log *logger.Logger) *SetupService {
+	return &SetupService{
+		db:     db,
+		logger: log,
+	}
+}
+
+// PoolDatasetInfo represents a pool with its datasets
+type PoolDatasetInfo struct {
+	PoolID    string              `json:"pool_id"`
+	PoolName  string              `json:"pool_name"`
+	Datasets  []DatasetInfo       `json:"datasets"`
+}
+
+// DatasetInfo represents a dataset that can be used for object storage
+type DatasetInfo struct {
+	ID         string `json:"id"`
+	Name       string `json:"name"`
+	FullName   string `json:"full_name"` // pool/dataset
+	MountPoint string `json:"mount_point"`
+	Type       string `json:"type"`
+	UsedBytes  int64  `json:"used_bytes"`
+	AvailableBytes int64 `json:"available_bytes"`
+}
+
+// GetAvailableDatasets returns all pools with their datasets that can be used for object storage
+func (s *SetupService) GetAvailableDatasets(ctx context.Context) ([]PoolDatasetInfo, error) {
+	// Get all pools
+	poolsQuery := `
+		SELECT id, name
+		FROM zfs_pools
+		WHERE is_active = true
+		ORDER BY name
+	`
+
+	rows, err := s.db.QueryContext(ctx, poolsQuery)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query pools: %w", err)
+	}
+	defer rows.Close()
+
+	var pools []PoolDatasetInfo
+	for rows.Next() {
+		var pool PoolDatasetInfo
+		if err := rows.Scan(&pool.PoolID, &pool.PoolName); err != nil {
+			s.logger.Warn("Failed to scan pool", "error", err)
+			continue
+		}
+
+		// Get datasets for this pool
+		datasetsQuery := `
+			SELECT id, name, type, mount_point, used_bytes, available_bytes
+			FROM zfs_datasets
+			WHERE pool_name = $1 AND type = 'filesystem'
+			ORDER BY name
+		`
+
+		datasetRows, err := s.db.QueryContext(ctx, datasetsQuery, pool.PoolName)
+		if err != nil {
+			s.logger.Warn("Failed to query datasets", "pool", pool.PoolName, "error", err)
+			pool.Datasets = []DatasetInfo{}
+			pools = append(pools, pool)
+			continue
+		}
+
+		var datasets []DatasetInfo
+		for datasetRows.Next() {
+			var ds DatasetInfo
+			var mountPoint sql.NullString
+
+			if err := datasetRows.Scan(&ds.ID, &ds.Name, &ds.Type, &mountPoint, &ds.UsedBytes, &ds.AvailableBytes); err != nil {
+				s.logger.Warn("Failed to scan dataset", "error", err)
+				continue
+			}
+
+			ds.FullName = fmt.Sprintf("%s/%s", pool.PoolName, ds.Name)
+			if mountPoint.Valid {
+				ds.MountPoint = mountPoint.String
+			} else {
+				ds.MountPoint = ""
+			}
+
+			datasets = append(datasets, ds)
+		}
+		datasetRows.Close()
+
+		pool.Datasets = datasets
+		pools = append(pools, pool)
+	}
+
+	return pools, nil
+}
+
+// SetupRequest represents a request to setup object storage
+type SetupRequest struct {
+	PoolName    string `json:"pool_name" binding:"required"`
+	DatasetName string `json:"dataset_name" binding:"required"`
+	CreateNew   bool   `json:"create_new"` // If true, create new dataset instead of using existing
+}
+
+// SetupResponse represents the response after setup
+type SetupResponse struct {
+	DatasetPath string `json:"dataset_path"`
+	MountPoint  string `json:"mount_point"`
+	Message     string `json:"message"`
+}
+
+// SetupObjectStorage configures MinIO to use a specific ZFS dataset
+func (s *SetupService) SetupObjectStorage(ctx context.Context, req SetupRequest) (*SetupResponse, error) {
+	var datasetPath, mountPoint string
+
+	// Normalize dataset name - if it already contains pool name, use it as-is
+	var fullDatasetName string
+	if strings.HasPrefix(req.DatasetName, req.PoolName+"/") {
+		// Dataset name already includes pool name (e.g., "pool/dataset")
+		fullDatasetName = req.DatasetName
+	} else {
+		// Dataset name is just the name (e.g., "dataset"), combine with pool
+		fullDatasetName = fmt.Sprintf("%s/%s", req.PoolName, req.DatasetName)
+	}
+
+	if req.CreateNew {
+		// Create new dataset for object storage
+		
+		// Check if dataset already exists
+		checkCmd := exec.CommandContext(ctx, "sudo", "zfs", "list", "-H", "-o", "name", fullDatasetName)
+		if err := checkCmd.Run(); err == nil {
+			return nil, fmt.Errorf("dataset %s already exists", fullDatasetName)
+		}
+
+		// Create dataset
+		createCmd := exec.CommandContext(ctx, "sudo", "zfs", "create", fullDatasetName)
+		if output, err := createCmd.CombinedOutput(); err != nil {
+			return nil, fmt.Errorf("failed to create dataset: %s - %w", string(output), err)
+		}
+
+		// Get mount point
+		getMountCmd := exec.CommandContext(ctx, "sudo", "zfs", "get", "-H", "-o", "value", "mountpoint", fullDatasetName)
+		mountOutput, err := getMountCmd.Output()
+		if err != nil {
+			return nil, fmt.Errorf("failed to get mount point: %w", err)
+		}
+		mountPoint = strings.TrimSpace(string(mountOutput))
+
+		datasetPath = fullDatasetName
+		s.logger.Info("Created new dataset for object storage", "dataset", fullDatasetName, "mount_point", mountPoint)
+	} else {
+		// Use existing dataset
+		// fullDatasetName already set above
+		
+		// Verify dataset exists
+		checkCmd := exec.CommandContext(ctx, "sudo", "zfs", "list", "-H", "-o", "name", fullDatasetName)
+		if err := checkCmd.Run(); err != nil {
+			return nil, fmt.Errorf("dataset %s does not exist", fullDatasetName)
+		}
+
+		// Get mount point
+		getMountCmd := exec.CommandContext(ctx, "sudo", "zfs", "get", "-H", "-o", "value", "mountpoint", fullDatasetName)
+		mountOutput, err := getMountCmd.Output()
+		if err != nil {
+			return nil, fmt.Errorf("failed to get mount point: %w", err)
+		}
+		mountPoint = strings.TrimSpace(string(mountOutput))
+
+		datasetPath = fullDatasetName
+		s.logger.Info("Using existing dataset for object storage", "dataset", fullDatasetName, "mount_point", mountPoint)
+	}
+
+	// Ensure mount point directory exists
+	if mountPoint != "none" && mountPoint != "" {
+		if err := os.MkdirAll(mountPoint, 0755); err != nil {
+			return nil, fmt.Errorf("failed to create mount point directory: %w", err)
+		}
+	} else {
+		// If no mount point, use default path
+		mountPoint = filepath.Join("/opt/calypso/data/pool", req.PoolName, req.DatasetName)
+		if err := os.MkdirAll(mountPoint, 0755); err != nil {
+			return nil, fmt.Errorf("failed to create default directory: %w", err)
+		}
+	}
+
+	// Update MinIO configuration to use the selected dataset
+	if err := s.updateMinIOConfig(ctx, mountPoint); err != nil {
+		s.logger.Warn("Failed to update MinIO configuration", "error", err)
+		// Continue anyway, configuration is saved to database
+	}
+
+	// Save configuration to database
+	_, err := s.db.ExecContext(ctx, `
+		INSERT INTO object_storage_config (dataset_path, mount_point, pool_name, dataset_name, created_at, updated_at)
+		VALUES ($1, $2, $3, $4, NOW(), NOW())
+		ON CONFLICT (id) DO UPDATE
+		SET dataset_path = $1, mount_point = $2, pool_name = $3, dataset_name = $4, updated_at = NOW()
+	`, datasetPath, mountPoint, req.PoolName, req.DatasetName)
+
+	if err != nil {
+		// If table doesn't exist, just log warning
+		s.logger.Warn("Failed to save configuration to database (table may not exist)", "error", err)
+	}
+
+	return &SetupResponse{
+		DatasetPath: datasetPath,
+		MountPoint:  mountPoint,
+		Message:     fmt.Sprintf("Object storage configured to use dataset %s at %s. MinIO service needs to be restarted to use the new dataset.", datasetPath, mountPoint),
+	}, nil
+}
+
+// GetCurrentSetup returns the current object storage configuration
+func (s *SetupService) GetCurrentSetup(ctx context.Context) (*SetupResponse, error) {
+	// Check if table exists first
+	var tableExists bool
+	checkQuery := `
+		SELECT EXISTS (
+			SELECT FROM information_schema.tables 
+			WHERE table_schema = 'public' 
+			AND table_name = 'object_storage_config'
+		)
+	`
+	err := s.db.QueryRowContext(ctx, checkQuery).Scan(&tableExists)
+	if err != nil {
+		s.logger.Warn("Failed to check if object_storage_config table exists", "error", err)
+		return nil, nil // Return nil if can't check
+	}
+
+	if !tableExists {
+		s.logger.Debug("object_storage_config table does not exist")
+		return nil, nil // No table, no configuration
+	}
+
+	query := `
+		SELECT dataset_path, mount_point, pool_name, dataset_name
+		FROM object_storage_config
+		ORDER BY updated_at DESC
+		LIMIT 1
+	`
+
+	var resp SetupResponse
+	var poolName, datasetName string
+	err = s.db.QueryRowContext(ctx, query).Scan(&resp.DatasetPath, &resp.MountPoint, &poolName, &datasetName)
+	if err == sql.ErrNoRows {
+		s.logger.Debug("No configuration found in database")
+		return nil, nil // No configuration found
+	}
+	if err != nil {
+		// Check if error is due to table not existing or permission denied
+		errStr := err.Error()
+		if strings.Contains(errStr, "does not exist") || strings.Contains(errStr, "permission denied") {
+			s.logger.Debug("Table does not exist or permission denied, returning nil", "error", errStr)
+			return nil, nil // Return nil instead of error
+		}
+		s.logger.Error("Failed to scan current setup", "error", err)
+		return nil, fmt.Errorf("failed to get current setup: %w", err)
+	}
+
+	s.logger.Debug("Found current setup", "dataset_path", resp.DatasetPath, "mount_point", resp.MountPoint, "pool", poolName, "dataset", datasetName)
+	// Use dataset_path directly since it already contains the full path
+	resp.Message = fmt.Sprintf("Using dataset %s at %s", resp.DatasetPath, resp.MountPoint)
+	return &resp, nil
+}
+
+// UpdateObjectStorage updates the object storage configuration to use a different dataset
+// This will update the configuration but won't migrate existing data
+func (s *SetupService) UpdateObjectStorage(ctx context.Context, req SetupRequest) (*SetupResponse, error) {
+	// First check if there's existing configuration
+	currentSetup, err := s.GetCurrentSetup(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to check current setup: %w", err)
+	}
+
+	if currentSetup == nil {
+		// No existing setup, just do normal setup
+		return s.SetupObjectStorage(ctx, req)
+	}
+
+	// There's existing setup, proceed with update
+	var datasetPath, mountPoint string
+
+	// Normalize dataset name - if it already contains pool name, use it as-is
+	var fullDatasetName string
+	if strings.HasPrefix(req.DatasetName, req.PoolName+"/") {
+		// Dataset name already includes pool name (e.g., "pool/dataset")
+		fullDatasetName = req.DatasetName
+	} else {
+		// Dataset name is just the name (e.g., "dataset"), combine with pool
+		fullDatasetName = fmt.Sprintf("%s/%s", req.PoolName, req.DatasetName)
+	}
+
+	if req.CreateNew {
+		// Create new dataset for object storage
+		
+		// Check if dataset already exists
+		checkCmd := exec.CommandContext(ctx, "sudo", "zfs", "list", "-H", "-o", "name", fullDatasetName)
+		if err := checkCmd.Run(); err == nil {
+			return nil, fmt.Errorf("dataset %s already exists", fullDatasetName)
+		}
+
+		// Create dataset
+		createCmd := exec.CommandContext(ctx, "sudo", "zfs", "create", fullDatasetName)
+		if output, err := createCmd.CombinedOutput(); err != nil {
+			return nil, fmt.Errorf("failed to create dataset: %s - %w", string(output), err)
+		}
+
+		// Get mount point
+		getMountCmd := exec.CommandContext(ctx, "sudo", "zfs", "get", "-H", "-o", "value", "mountpoint", fullDatasetName)
+		mountOutput, err := getMountCmd.Output()
+		if err != nil {
+			return nil, fmt.Errorf("failed to get mount point: %w", err)
+		}
+		mountPoint = strings.TrimSpace(string(mountOutput))
+
+		datasetPath = fullDatasetName
+		s.logger.Info("Created new dataset for object storage update", "dataset", fullDatasetName, "mount_point", mountPoint)
+	} else {
+		// Use existing dataset
+		// fullDatasetName already set above
+		
+		// Verify dataset exists
+		checkCmd := exec.CommandContext(ctx, "sudo", "zfs", "list", "-H", "-o", "name", fullDatasetName)
+		if err := checkCmd.Run(); err != nil {
+			return nil, fmt.Errorf("dataset %s does not exist", fullDatasetName)
+		}
+
+		// Get mount point
+		getMountCmd := exec.CommandContext(ctx, "sudo", "zfs", "get", "-H", "-o", "value", "mountpoint", fullDatasetName)
+		mountOutput, err := getMountCmd.Output()
+		if err != nil {
+			return nil, fmt.Errorf("failed to get mount point: %w", err)
+		}
+		mountPoint = strings.TrimSpace(string(mountOutput))
+
+		datasetPath = fullDatasetName
+		s.logger.Info("Using existing dataset for object storage update", "dataset", fullDatasetName, "mount_point", mountPoint)
+	}
+
+	// Ensure mount point directory exists
+	if mountPoint != "none" && mountPoint != "" {
+		if err := os.MkdirAll(mountPoint, 0755); err != nil {
+			return nil, fmt.Errorf("failed to create mount point directory: %w", err)
+		}
+	} else {
+		// If no mount point, use default path
+		mountPoint = filepath.Join("/opt/calypso/data/pool", req.PoolName, req.DatasetName)
+		if err := os.MkdirAll(mountPoint, 0755); err != nil {
+			return nil, fmt.Errorf("failed to create default directory: %w", err)
+		}
+	}
+
+	// Update configuration in database
+	_, err = s.db.ExecContext(ctx, `
+		UPDATE object_storage_config
+		SET dataset_path = $1, mount_point = $2, pool_name = $3, dataset_name = $4, updated_at = NOW()
+		WHERE id = (SELECT id FROM object_storage_config ORDER BY updated_at DESC LIMIT 1)
+	`, datasetPath, mountPoint, req.PoolName, req.DatasetName)
+
+	if err != nil {
+		// If update fails, try insert
+		_, err = s.db.ExecContext(ctx, `
+			INSERT INTO object_storage_config (dataset_path, mount_point, pool_name, dataset_name, created_at, updated_at)
+			VALUES ($1, $2, $3, $4, NOW(), NOW())
+			ON CONFLICT (dataset_path) DO UPDATE
+			SET mount_point = $2, pool_name = $3, dataset_name = $4, updated_at = NOW()
+		`, datasetPath, mountPoint, req.PoolName, req.DatasetName)
+		if err != nil {
+			s.logger.Warn("Failed to update configuration in database", "error", err)
+		}
+	}
+
+	// Update MinIO configuration to use the selected dataset
+	if err := s.updateMinIOConfig(ctx, mountPoint); err != nil {
+		s.logger.Warn("Failed to update MinIO configuration", "error", err)
+		// Continue anyway, configuration is saved to database
+	} else {
+		// Restart MinIO service to apply new configuration
+		if err := s.restartMinIOService(ctx); err != nil {
+			s.logger.Warn("Failed to restart MinIO service", "error", err)
+			// Continue anyway, user can restart manually
+		}
+	}
+
+	return &SetupResponse{
+		DatasetPath: datasetPath,
+		MountPoint:  mountPoint,
+		Message:     fmt.Sprintf("Object storage updated to use dataset %s at %s. Note: Existing data in previous dataset (%s) is not migrated automatically. MinIO service has been restarted.", datasetPath, mountPoint, currentSetup.DatasetPath),
+	}, nil
+}
+
+// updateMinIOConfig updates MinIO configuration file to use dataset mount point directly
+// Note: MinIO erasure coding requires direct directory paths, not symlinks
+func (s *SetupService) updateMinIOConfig(ctx context.Context, datasetMountPoint string) error {
+	configFile := "/opt/calypso/conf/minio/minio.conf"
+
+	// Ensure dataset mount point directory exists and has correct ownership
+	if err := os.MkdirAll(datasetMountPoint, 0755); err != nil {
+		return fmt.Errorf("failed to create dataset mount point directory: %w", err)
+	}
+
+	// Set ownership to minio-user so MinIO can write to it
+	if err := exec.CommandContext(ctx, "sudo", "chown", "-R", "minio-user:minio-user", datasetMountPoint).Run(); err != nil {
+		s.logger.Warn("Failed to set ownership on dataset mount point", "path", datasetMountPoint, "error", err)
+		// Continue anyway, might already have correct ownership
+	}
+
+	// Set permissions
+	if err := exec.CommandContext(ctx, "sudo", "chmod", "755", datasetMountPoint).Run(); err != nil {
+		s.logger.Warn("Failed to set permissions on dataset mount point", "path", datasetMountPoint, "error", err)
+	}
+
+	s.logger.Info("Prepared dataset mount point for MinIO", "path", datasetMountPoint)
+
+	// Read current config file
+	configContent, err := os.ReadFile(configFile)
+	if err != nil {
+		// If file doesn't exist, create it
+		if os.IsNotExist(err) {
+			configContent = []byte(fmt.Sprintf("MINIO_ROOT_USER=admin\nMINIO_ROOT_PASSWORD=HqBX1IINqFynkWFa\nMINIO_VOLUMES=%s\n", datasetMountPoint))
+		} else {
+			return fmt.Errorf("failed to read MinIO config file: %w", err)
+		}
+	} else {
+		// Update MINIO_VOLUMES in config
+		lines := strings.Split(string(configContent), "\n")
+		updated := false
+		for i, line := range lines {
+			if strings.HasPrefix(strings.TrimSpace(line), "MINIO_VOLUMES=") {
+				lines[i] = fmt.Sprintf("MINIO_VOLUMES=%s", datasetMountPoint)
+				updated = true
+				break
+			}
+		}
+		if !updated {
+			// Add MINIO_VOLUMES if not found
+			lines = append(lines, fmt.Sprintf("MINIO_VOLUMES=%s", datasetMountPoint))
+		}
+		configContent = []byte(strings.Join(lines, "\n"))
+	}
+
+	// Write updated config using sudo
+	// Write temp file to a location we can write to
+	userTempFile := fmt.Sprintf("/tmp/minio.conf.%d.tmp", os.Getpid())
+	if err := os.WriteFile(userTempFile, configContent, 0644); err != nil {
+		return fmt.Errorf("failed to write temp config file: %w", err)
+	}
+	defer os.Remove(userTempFile) // Cleanup
+
+	// Copy temp file to config location with sudo
+	if err := exec.CommandContext(ctx, "sudo", "cp", userTempFile, configFile).Run(); err != nil {
+		return fmt.Errorf("failed to update config file: %w", err)
+	}
+
+	// Set proper ownership and permissions
+	if err := exec.CommandContext(ctx, "sudo", "chown", "minio-user:minio-user", configFile).Run(); err != nil {
+		s.logger.Warn("Failed to set config file ownership", "error", err)
+	}
+	if err := exec.CommandContext(ctx, "sudo", "chmod", "644", configFile).Run(); err != nil {
+		s.logger.Warn("Failed to set config file permissions", "error", err)
+	}
+
+	s.logger.Info("Updated MinIO configuration", "config_file", configFile, "volumes", datasetMountPoint)
+
+	return nil
+}
+
+// restartMinIOService restarts the MinIO service to apply new configuration
+func (s *SetupService) restartMinIOService(ctx context.Context) error {
+	// Restart MinIO service using sudo
+	cmd := exec.CommandContext(ctx, "sudo", "systemctl", "restart", "minio.service")
+	if err := cmd.Run(); err != nil {
+		return fmt.Errorf("failed to restart MinIO service: %w", err)
+	}
+
+	// Wait a moment for service to start
+	time.Sleep(2 * time.Second)
+
+	// Verify service is running
+	checkCmd := exec.CommandContext(ctx, "sudo", "systemctl", "is-active", "minio.service")
+	output, err := checkCmd.Output()
+	if err != nil {
+		return fmt.Errorf("failed to check MinIO service status: %w", err)
+	}
+
+	status := strings.TrimSpace(string(output))
+	if status != "active" {
+		return fmt.Errorf("MinIO service is not active after restart, status: %s", status)
+	}
+
+	s.logger.Info("MinIO service restarted successfully")
+	return nil
+}