package storage import ( "fmt" "net/http" "strings" "github.com/atlasos/calypso/internal/common/cache" "github.com/atlasos/calypso/internal/common/database" "github.com/atlasos/calypso/internal/common/logger" "github.com/atlasos/calypso/internal/tasks" "github.com/gin-gonic/gin" ) // Handler handles storage-related API requests type Handler struct { diskService *DiskService lvmService *LVMService zfsService *ZFSService arcService *ARCService taskEngine *tasks.Engine db *database.DB logger *logger.Logger cache *cache.Cache // Cache for invalidation } // SetCache sets the cache instance for cache invalidation func (h *Handler) SetCache(c *cache.Cache) { h.cache = c } // NewHandler creates a new storage handler func NewHandler(db *database.DB, log *logger.Logger) *Handler { return &Handler{ diskService: NewDiskService(db, log), lvmService: NewLVMService(db, log), zfsService: NewZFSService(db, log), arcService: NewARCService(log), taskEngine: tasks.NewEngine(db, log), db: db, logger: log, } } // ListDisks lists all physical disks func (h *Handler) ListDisks(c *gin.Context) { disks, err := h.diskService.DiscoverDisks(c.Request.Context()) if err != nil { h.logger.Error("Failed to list disks", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list disks"}) return } c.JSON(http.StatusOK, gin.H{"disks": disks}) } // SyncDisks syncs discovered disks to database func (h *Handler) SyncDisks(c *gin.Context) { userID, _ := c.Get("user_id") // Create async task taskID, err := h.taskEngine.CreateTask(c.Request.Context(), tasks.TaskTypeRescan, userID.(string), map[string]interface{}{ "operation": "sync_disks", }) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create task"}) return } // Run sync in background go func() { ctx := c.Request.Context() h.taskEngine.StartTask(ctx, taskID) h.taskEngine.UpdateProgress(ctx, taskID, 50, "Discovering disks...") if err := h.diskService.SyncDisksToDatabase(ctx); err != nil { h.taskEngine.FailTask(ctx, taskID, err.Error()) return } h.taskEngine.UpdateProgress(ctx, taskID, 100, "Disk sync completed") h.taskEngine.CompleteTask(ctx, taskID, "Disks synchronized successfully") }() c.JSON(http.StatusAccepted, gin.H{"task_id": taskID}) } // ListVolumeGroups lists all volume groups func (h *Handler) ListVolumeGroups(c *gin.Context) { vgs, err := h.lvmService.ListVolumeGroups(c.Request.Context()) if err != nil { h.logger.Error("Failed to list volume groups", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list volume groups"}) return } c.JSON(http.StatusOK, gin.H{"volume_groups": vgs}) } // ListRepositories lists all repositories func (h *Handler) ListRepositories(c *gin.Context) { repos, err := h.lvmService.ListRepositories(c.Request.Context()) if err != nil { h.logger.Error("Failed to list repositories", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list repositories"}) return } c.JSON(http.StatusOK, gin.H{"repositories": repos}) } // GetRepository retrieves a repository by ID func (h *Handler) GetRepository(c *gin.Context) { repoID := c.Param("id") repo, err := h.lvmService.GetRepository(c.Request.Context(), repoID) if err != nil { if err.Error() == "repository not found" { c.JSON(http.StatusNotFound, gin.H{"error": "repository not found"}) return } h.logger.Error("Failed to get repository", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get repository"}) return } c.JSON(http.StatusOK, repo) } // CreateRepositoryRequest represents a repository creation request type CreateRepositoryRequest struct { Name string `json:"name" binding:"required"` Description string `json:"description"` VolumeGroup string `json:"volume_group" binding:"required"` SizeGB int64 `json:"size_gb" binding:"required"` } // CreateRepository creates a new repository func (h *Handler) CreateRepository(c *gin.Context) { var req CreateRepositoryRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request"}) return } userID, _ := c.Get("user_id") sizeBytes := req.SizeGB * 1024 * 1024 * 1024 repo, err := h.lvmService.CreateRepository( c.Request.Context(), req.Name, req.VolumeGroup, sizeBytes, userID.(string), ) if err != nil { h.logger.Error("Failed to create repository", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusCreated, repo) } // DeleteRepository deletes a repository func (h *Handler) DeleteRepository(c *gin.Context) { repoID := c.Param("id") if err := h.lvmService.DeleteRepository(c.Request.Context(), repoID); err != nil { if err.Error() == "repository not found" { c.JSON(http.StatusNotFound, gin.H{"error": "repository not found"}) return } h.logger.Error("Failed to delete repository", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "repository deleted successfully"}) } // CreateZPoolRequest represents a ZFS pool creation request type CreateZPoolRequest struct { Name string `json:"name" binding:"required"` Description string `json:"description"` RaidLevel string `json:"raid_level" binding:"required"` // stripe, mirror, raidz, raidz2, raidz3 Disks []string `json:"disks" binding:"required"` // device paths Compression string `json:"compression"` // off, lz4, zstd, gzip Deduplication bool `json:"deduplication"` AutoExpand bool `json:"auto_expand"` } // CreateZPool creates a new ZFS pool func (h *Handler) CreateZPool(c *gin.Context) { var req CreateZPoolRequest if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid request body", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } // Validate required fields if req.Name == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "pool name is required"}) return } if req.RaidLevel == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "raid_level is required"}) return } if len(req.Disks) == 0 { c.JSON(http.StatusBadRequest, gin.H{"error": "at least one disk is required"}) return } userID, exists := c.Get("user_id") if !exists { h.logger.Error("User ID not found in context") c.JSON(http.StatusUnauthorized, gin.H{"error": "authentication required"}) return } userIDStr, ok := userID.(string) if !ok { h.logger.Error("Invalid user ID type", "type", fmt.Sprintf("%T", userID)) c.JSON(http.StatusInternalServerError, gin.H{"error": "invalid user context"}) return } // Set default compression if not provided if req.Compression == "" { req.Compression = "lz4" } h.logger.Info("Creating ZFS pool request", "name", req.Name, "raid_level", req.RaidLevel, "disks", req.Disks, "compression", req.Compression) pool, err := h.zfsService.CreatePool( c.Request.Context(), req.Name, req.RaidLevel, req.Disks, req.Compression, req.Deduplication, req.AutoExpand, userIDStr, ) if err != nil { h.logger.Error("Failed to create ZFS pool", "error", err, "name", req.Name, "raid_level", req.RaidLevel, "disks", req.Disks) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } h.logger.Info("ZFS pool created successfully", "pool_id", pool.ID, "name", pool.Name) c.JSON(http.StatusCreated, pool) } // ListZFSPools lists all ZFS pools func (h *Handler) ListZFSPools(c *gin.Context) { pools, err := h.zfsService.ListPools(c.Request.Context()) if err != nil { h.logger.Error("Failed to list ZFS pools", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list ZFS pools"}) return } c.JSON(http.StatusOK, gin.H{"pools": pools}) } // GetZFSPool retrieves a ZFS pool by ID func (h *Handler) GetZFSPool(c *gin.Context) { poolID := c.Param("id") pool, err := h.zfsService.GetPool(c.Request.Context(), poolID) if err != nil { if err.Error() == "pool not found" { c.JSON(http.StatusNotFound, gin.H{"error": "pool not found"}) return } h.logger.Error("Failed to get ZFS pool", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get ZFS pool"}) return } c.JSON(http.StatusOK, pool) } // DeleteZFSPool deletes a ZFS pool func (h *Handler) DeleteZFSPool(c *gin.Context) { poolID := c.Param("id") if err := h.zfsService.DeletePool(c.Request.Context(), poolID); err != nil { if err.Error() == "pool not found" { c.JSON(http.StatusNotFound, gin.H{"error": "pool not found"}) return } h.logger.Error("Failed to delete ZFS pool", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "ZFS pool deleted successfully"}) } // AddSpareDiskRequest represents a request to add spare disks to a pool type AddSpareDiskRequest struct { Disks []string `json:"disks" binding:"required"` } // AddSpareDisk adds spare disks to a ZFS pool func (h *Handler) AddSpareDisk(c *gin.Context) { poolID := c.Param("id") var req AddSpareDiskRequest if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid add spare disk request", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } if len(req.Disks) == 0 { c.JSON(http.StatusBadRequest, gin.H{"error": "at least one disk must be specified"}) return } if err := h.zfsService.AddSpareDisk(c.Request.Context(), poolID, req.Disks); err != nil { if err.Error() == "pool not found" { c.JSON(http.StatusNotFound, gin.H{"error": "pool not found"}) return } h.logger.Error("Failed to add spare disks", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "Spare disks added successfully"}) } // ListZFSDatasets lists all datasets in a ZFS pool func (h *Handler) ListZFSDatasets(c *gin.Context) { poolID := c.Param("id") // Get pool to get pool name pool, err := h.zfsService.GetPool(c.Request.Context(), poolID) if err != nil { if err.Error() == "pool not found" { c.JSON(http.StatusNotFound, gin.H{"error": "pool not found"}) return } h.logger.Error("Failed to get pool", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get pool"}) return } datasets, err := h.zfsService.ListDatasets(c.Request.Context(), pool.Name) if err != nil { h.logger.Error("Failed to list datasets", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } // Ensure we return an empty array instead of null if datasets == nil { datasets = []*ZFSDataset{} } c.JSON(http.StatusOK, gin.H{"datasets": datasets}) } // CreateZFSDatasetRequest represents a request to create a ZFS dataset type CreateZFSDatasetRequest struct { Name string `json:"name" binding:"required"` // Dataset name (without pool prefix) Type string `json:"type" binding:"required"` // "filesystem" or "volume" Compression string `json:"compression"` // off, lz4, zstd, gzip, etc. Quota int64 `json:"quota"` // -1 for unlimited, >0 for size Reservation int64 `json:"reservation"` // 0 for none MountPoint string `json:"mount_point"` // Optional mount point } // CreateZFSDataset creates a new ZFS dataset in a pool func (h *Handler) CreateZFSDataset(c *gin.Context) { poolID := c.Param("id") // Get pool to get pool name pool, err := h.zfsService.GetPool(c.Request.Context(), poolID) if err != nil { if err.Error() == "pool not found" { c.JSON(http.StatusNotFound, gin.H{"error": "pool not found"}) return } h.logger.Error("Failed to get pool", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get pool"}) return } var req CreateZFSDatasetRequest if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid create dataset request", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } // Validate type if req.Type != "filesystem" && req.Type != "volume" { c.JSON(http.StatusBadRequest, gin.H{"error": "type must be 'filesystem' or 'volume'"}) return } // Validate mount point: volumes cannot have mount points if req.Type == "volume" && req.MountPoint != "" { c.JSON(http.StatusBadRequest, gin.H{"error": "mount point cannot be set for volume datasets (volumes are block devices for iSCSI export)"}) return } // Validate dataset name (should not contain pool name) if strings.Contains(req.Name, "/") { c.JSON(http.StatusBadRequest, gin.H{"error": "dataset name should not contain '/' (pool name is automatically prepended)"}) return } // Create dataset request - CreateDatasetRequest is in the same package (zfs.go) createReq := CreateDatasetRequest{ Name: req.Name, Type: req.Type, Compression: req.Compression, Quota: req.Quota, Reservation: req.Reservation, MountPoint: req.MountPoint, } dataset, err := h.zfsService.CreateDataset(c.Request.Context(), pool.Name, createReq) if err != nil { h.logger.Error("Failed to create dataset", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusCreated, dataset) } // DeleteZFSDataset deletes a ZFS dataset func (h *Handler) DeleteZFSDataset(c *gin.Context) { poolID := c.Param("id") datasetName := c.Param("dataset") // Get pool to get pool name pool, err := h.zfsService.GetPool(c.Request.Context(), poolID) if err != nil { if err.Error() == "pool not found" { c.JSON(http.StatusNotFound, gin.H{"error": "pool not found"}) return } h.logger.Error("Failed to get pool", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get pool"}) return } // Construct full dataset name fullDatasetName := pool.Name + "/" + datasetName // Verify dataset belongs to this pool if !strings.HasPrefix(fullDatasetName, pool.Name+"/") { c.JSON(http.StatusBadRequest, gin.H{"error": "dataset does not belong to this pool"}) return } if err := h.zfsService.DeleteDataset(c.Request.Context(), fullDatasetName); err != nil { if strings.Contains(err.Error(), "does not exist") || strings.Contains(err.Error(), "not found") { c.JSON(http.StatusNotFound, gin.H{"error": "dataset not found"}) return } h.logger.Error("Failed to delete dataset", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } // Invalidate cache for this pool's datasets list if h.cache != nil { // Generate cache key using the same format as cache middleware cacheKey := fmt.Sprintf("http:/api/v1/storage/zfs/pools/%s/datasets:", poolID) h.cache.Delete(cacheKey) // Also invalidate any cached responses with query parameters h.logger.Debug("Cache invalidated for dataset list", "pool_id", poolID, "key", cacheKey) } c.JSON(http.StatusOK, gin.H{"message": "Dataset deleted successfully"}) } // GetARCStats returns ZFS ARC statistics func (h *Handler) GetARCStats(c *gin.Context) { stats, err := h.arcService.GetARCStats(c.Request.Context()) if err != nil { h.logger.Error("Failed to get ARC stats", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get ARC stats: " + err.Error()}) return } c.JSON(http.StatusOK, stats) }