From 692cef94b80a50724d172d65b8ee1d870f7d3269 Mon Sep 17 00:00:00 2001 From: MUHAMMAD FARIS EL HAKIM Date: Tue, 28 Apr 2026 14:31:42 +0000 Subject: [PATCH] feat(lab): backend-agnostic ephemeral env API with Docker driver Introduce /api/v1/lab/envs (backend-agnostic), a Docker reference driver, hot/cold storage tiers, a TTL reaper, and concave lab CLI subcommands. The design is driver-pluggable (docker today, slurm and proxmox in follow-ups) under one API. Archives stamp a peer_id so envs archived on node A can be restored on node B. Every launch/extend/archive is written to a JSONL audit log at ~/gradient/config/lab-audit.jsonl. - internal/lab: Driver interface, Registry, Manager, Store, StorageConfig, DockerDriver, FileAuditWriter, TTL reaper goroutine. - internal/api/handlers_lab.go: CRUD + extend + archive + storage + driver endpoints, role-gated (viewer lists, developer launches/extends, operator archives, admin reconfigures storage/driver). - cmd/lab_envs.go: concave lab envs/storage/driver subcommands. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- cmd/lab_envs.go | 282 +++++++++++++++++++++++ internal/api/handlers_lab.go | 227 ++++++++++++++++++ internal/api/server.go | 25 ++ internal/lab/audit.go | 132 +++++++++++ internal/lab/driver.go | 128 +++++++++++ internal/lab/driver_docker.go | 352 ++++++++++++++++++++++++++++ internal/lab/lab_test.go | 420 ++++++++++++++++++++++++++++++++++ internal/lab/manager.go | 301 ++++++++++++++++++++++++ internal/lab/storage.go | 123 ++++++++++ internal/lab/store.go | 178 ++++++++++++++ internal/lab/types.go | 96 ++++++++ 11 files changed, 2264 insertions(+) create mode 100644 cmd/lab_envs.go create mode 100644 internal/api/handlers_lab.go create mode 100644 internal/lab/audit.go create mode 100644 internal/lab/driver.go create mode 100644 internal/lab/driver_docker.go create mode 100644 internal/lab/lab_test.go create mode 100644 internal/lab/manager.go create mode 100644 internal/lab/storage.go create mode 100644 internal/lab/store.go create mode 100644 internal/lab/types.go diff --git a/cmd/lab_envs.go b/cmd/lab_envs.go new file mode 100644 index 0000000..e950499 --- /dev/null +++ b/cmd/lab_envs.go @@ -0,0 +1,282 @@ +package cmd + +import ( + "context" + "encoding/json" + "fmt" + "os" + "strings" + "text/tabwriter" + "time" + + "github.com/spf13/cobra" + + "github.com/Gradient-Linux/concave/internal/lab" + "github.com/Gradient-Linux/concave/internal/ui" +) + +var ( + labLaunchImage string + labLaunchDisplayName string + labLaunchTTL time.Duration + labLaunchGPUs int + labLaunchCPU string + labLaunchMem string + labLaunchDriver string + labLaunchOwner string + + labExtendBy time.Duration + labStorageHot string + labStorageCold string + labActiveDriver string +) + +var labEnvsCmd = &cobra.Command{ + Use: "envs", + Short: "Manage ephemeral JupyterLab environments", +} + +var labEnvsListCmd = &cobra.Command{ + Use: "list", + Short: "List ephemeral lab environments", + RunE: runLabEnvsList, +} + +var labEnvsLaunchCmd = &cobra.Command{ + Use: "launch", + Short: "Launch an ephemeral lab environment", + RunE: runLabEnvsLaunch, +} + +var labEnvsExtendCmd = &cobra.Command{ + Use: "extend ", + Short: "Extend the TTL of an ephemeral lab environment", + Args: cobra.ExactArgs(1), + RunE: runLabEnvsExtend, +} + +var labEnvsArchiveCmd = &cobra.Command{ + Use: "archive ", + Short: "Archive an ephemeral lab environment to the cold tier and destroy it", + Args: cobra.ExactArgs(1), + RunE: runLabEnvsArchive, +} + +var labStorageCmd = &cobra.Command{ + Use: "storage", + Short: "Inspect or update lab storage tiers", +} + +var labStorageShowCmd = &cobra.Command{ + Use: "show", + Short: "Show current lab storage configuration", + RunE: runLabStorageShow, +} + +var labStorageSetCmd = &cobra.Command{ + Use: "set", + Short: "Set hot and cold tier paths", + RunE: runLabStorageSet, +} + +var labDriverCmd = &cobra.Command{ + Use: "driver", + Short: "Inspect or select the active lab driver", +} + +var labDriverShowCmd = &cobra.Command{ + Use: "show", + Short: "Show registered lab drivers and the active one", + RunE: runLabDriverShow, +} + +var labDriverSetCmd = &cobra.Command{ + Use: "set ", + Short: "Set the active lab driver (docker | slurm | proxmox)", + Args: cobra.ExactArgs(1), + RunE: runLabDriverSet, +} + +func runLabEnvsList(_ *cobra.Command, _ []string) error { + mgr, err := newLabManager() + if err != nil { + return err + } + envs := mgr.List() + tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) + fmt.Fprintln(tw, "ID\tOWNER\tDRIVER\tSTATUS\tIMAGE\tEXPIRES\tURL") + now := time.Now() + for _, env := range envs { + expires := "-" + if !env.ExpiresAt.IsZero() { + expires = env.Remaining(now).Round(time.Second).String() + } + fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s\t%s\n", + env.ID, env.Owner, env.Driver, env.Status, env.Image, expires, env.JupyterURL) + } + return tw.Flush() +} + +func runLabEnvsLaunch(cmd *cobra.Command, _ []string) error { + if labLaunchImage == "" { + return fmt.Errorf("--image is required") + } + if labLaunchTTL <= 0 { + return fmt.Errorf("--ttl must be positive (e.g. 2h)") + } + owner := labLaunchOwner + if owner == "" { + if u := os.Getenv("USER"); u != "" { + owner = u + } else { + owner = "local" + } + } + mgr, err := newLabManager() + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Minute) + defer cancel() + env, err := mgr.Launch(ctx, lab.EnvSpec{ + Owner: owner, + Image: labLaunchImage, + DisplayName: labLaunchDisplayName, + GPUs: labLaunchGPUs, + CPURequest: labLaunchCPU, + MemRequest: labLaunchMem, + TTL: labLaunchTTL, + Driver: labLaunchDriver, + }) + if err != nil { + return err + } + ui.Info("lab", fmt.Sprintf("launched %s (%s) expiring in %s", env.ID, env.Driver, labLaunchTTL)) + if env.JupyterURL != "" { + ui.Info("jupyter", env.JupyterURL) + } + return nil +} + +func runLabEnvsExtend(cmd *cobra.Command, args []string) error { + if labExtendBy <= 0 { + return fmt.Errorf("--by must be positive (e.g. 1h)") + } + mgr, err := newLabManager() + if err != nil { + return err + } + env, err := mgr.ExtendTTL(cmd.Context(), args[0], labExtendBy) + if err != nil { + return err + } + ui.Info("lab", fmt.Sprintf("extended %s; new expiry %s", env.ID, env.ExpiresAt.Format(time.RFC3339))) + return nil +} + +func runLabEnvsArchive(cmd *cobra.Command, args []string) error { + mgr, err := newLabManager() + if err != nil { + return err + } + env, err := mgr.ArchiveAndDestroy(cmd.Context(), args[0]) + if err != nil { + return err + } + ui.Info("lab", fmt.Sprintf("archived %s -> %s", env.ID, env.ArchiveRef)) + return nil +} + +func runLabStorageShow(_ *cobra.Command, _ []string) error { + storage, err := lab.LoadStorage() + if err != nil { + return err + } + out, _ := json.MarshalIndent(storage, "", " ") + fmt.Println(string(out)) + return nil +} + +func runLabStorageSet(_ *cobra.Command, _ []string) error { + current, err := lab.LoadStorage() + if err != nil { + return err + } + if strings.TrimSpace(labStorageHot) != "" { + current.HotTier = labStorageHot + } + if strings.TrimSpace(labStorageCold) != "" { + current.ColdTier = labStorageCold + } + if err := lab.SaveStorage(current); err != nil { + return err + } + if err := lab.EnsureTierDirs(current); err != nil { + return err + } + ui.Info("lab", fmt.Sprintf("hot=%s cold=%s", current.HotTier, current.ColdTier)) + return nil +} + +func runLabDriverShow(_ *cobra.Command, _ []string) error { + mgr, err := newLabManager() + if err != nil { + return err + } + fmt.Println("active:", mgr.Registry().Active()) + fmt.Println("drivers:", strings.Join(mgr.Registry().Names(), ", ")) + return nil +} + +func runLabDriverSet(_ *cobra.Command, args []string) error { + mgr, err := newLabManager() + if err != nil { + return err + } + if err := mgr.Registry().SetActive(args[0]); err != nil { + return err + } + ui.Info("lab", "active driver: "+args[0]) + return nil +} + +// newLabManager builds a local-only lab.Manager suitable for CLI usage. The +// API server builds its own instance. +func newLabManager() (*lab.Manager, error) { + registry := lab.NewRegistry() + registry.Register(lab.NewDockerDriver(lab.DockerDriverConfig{})) + store, err := lab.NewStore() + if err != nil { + return nil, err + } + storage, err := lab.LoadStorage() + if err != nil { + return nil, err + } + if err := lab.EnsureTierDirs(storage); err != nil { + return nil, err + } + return lab.NewManager(registry, store, storage), nil +} + +func init() { + labEnvsLaunchCmd.Flags().StringVar(&labLaunchImage, "image", "", "docker image for the env (e.g. jupyter/datascience-notebook:latest)") + labEnvsLaunchCmd.Flags().StringVar(&labLaunchDisplayName, "name", "", "display name") + labEnvsLaunchCmd.Flags().DurationVar(&labLaunchTTL, "ttl", 2*time.Hour, "time-to-live before the env is archived") + labEnvsLaunchCmd.Flags().IntVar(&labLaunchGPUs, "gpus", 0, "number of GPUs to attach") + labEnvsLaunchCmd.Flags().StringVar(&labLaunchCPU, "cpu", "", "CPU request (docker --cpus)") + labEnvsLaunchCmd.Flags().StringVar(&labLaunchMem, "mem", "", "memory request (docker --memory, e.g. 8g)") + labEnvsLaunchCmd.Flags().StringVar(&labLaunchDriver, "driver", "", "driver to use (defaults to active)") + labEnvsLaunchCmd.Flags().StringVar(&labLaunchOwner, "owner", "", "override owner (defaults to $USER)") + + labEnvsExtendCmd.Flags().DurationVar(&labExtendBy, "by", time.Hour, "extension duration") + + labStorageSetCmd.Flags().StringVar(&labStorageHot, "hot", "", "absolute path for the hot tier") + labStorageSetCmd.Flags().StringVar(&labStorageCold, "cold", "", "absolute path for the cold tier") + _ = labActiveDriver // reserved for future use + + labEnvsCmd.AddCommand(labEnvsListCmd, labEnvsLaunchCmd, labEnvsExtendCmd, labEnvsArchiveCmd) + labStorageCmd.AddCommand(labStorageShowCmd, labStorageSetCmd) + labDriverCmd.AddCommand(labDriverShowCmd, labDriverSetCmd) + labCmd.AddCommand(labEnvsCmd, labStorageCmd, labDriverCmd) +} diff --git a/internal/api/handlers_lab.go b/internal/api/handlers_lab.go new file mode 100644 index 0000000..a0fb013 --- /dev/null +++ b/internal/api/handlers_lab.go @@ -0,0 +1,227 @@ +package api + +import ( + "encoding/json" + "net/http" + "strings" + "time" + + "github.com/Gradient-Linux/concave/internal/auth" + "github.com/Gradient-Linux/concave/internal/lab" +) + +// labEnvRequest is the POST /api/v1/lab/envs body. +type labEnvRequest struct { + Image string `json:"image"` + DisplayName string `json:"display_name"` + Driver string `json:"driver"` + GPUs int `json:"gpus"` + CPURequest string `json:"cpu_request"` + MemRequest string `json:"mem_request"` + TTLSeconds int `json:"ttl_seconds"` +} + +type labExtendRequest struct { + ExtendSeconds int `json:"extend_seconds"` +} + +type labStorageRequest struct { + HotTier string `json:"hot_tier"` + ColdTier string `json:"cold_tier"` +} + +func (a *App) handleLabEnvs(w http.ResponseWriter, r *http.Request) { + if a.lab == nil { + writeError(w, http.StatusServiceUnavailable, "lab manager not initialised") + return + } + switch r.Method { + case http.MethodGet: + writeJSON(w, http.StatusOK, map[string]any{ + "envs": a.lab.List(), + "storage": a.lab.Storage(), + "drivers": a.lab.Registry().Names(), + "active": a.lab.Registry().Active(), + }) + case http.MethodPost: + if err := auth.Require(ClaimsFromContextMust(r).Role, auth.ActionStart); err != nil { + writeError(w, http.StatusForbidden, "insufficient role") + return + } + var req labEnvRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid request body") + return + } + if req.TTLSeconds <= 0 { + writeError(w, http.StatusBadRequest, "ttl_seconds must be positive") + return + } + claims := ClaimsFromContextMust(r) + spec := lab.EnvSpec{ + Owner: claims.Subject, + Image: req.Image, + DisplayName: req.DisplayName, + Driver: req.Driver, + GPUs: req.GPUs, + CPURequest: req.CPURequest, + MemRequest: req.MemRequest, + TTL: time.Duration(req.TTLSeconds) * time.Second, + } + env, err := a.lab.Launch(r.Context(), spec) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusCreated, env) + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +func (a *App) handleLabEnvSubroutes(w http.ResponseWriter, r *http.Request) { + if a.lab == nil { + writeError(w, http.StatusServiceUnavailable, "lab manager not initialised") + return + } + path := strings.TrimPrefix(r.URL.Path, "/api/v1/lab/envs/") + if path == "" { + writeError(w, http.StatusNotFound, "env id required") + return + } + parts := strings.Split(path, "/") + id := parts[0] + + if len(parts) == 1 { + switch r.Method { + case http.MethodGet: + env, err := a.lab.Get(r.Context(), id) + if err != nil { + writeError(w, http.StatusNotFound, err.Error()) + return + } + writeJSON(w, http.StatusOK, env) + case http.MethodDelete: + if err := auth.Require(ClaimsFromContextMust(r).Role, auth.ActionStop); err != nil { + writeError(w, http.StatusForbidden, "insufficient role") + return + } + env, err := a.lab.ArchiveAndDestroy(r.Context(), id) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, env) + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } + return + } + + action := parts[1] + switch action { + case "extend": + if r.Method != http.MethodPost { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + if err := auth.Require(ClaimsFromContextMust(r).Role, auth.ActionStart); err != nil { + writeError(w, http.StatusForbidden, "insufficient role") + return + } + var req labExtendRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid request body") + return + } + if req.ExtendSeconds <= 0 { + writeError(w, http.StatusBadRequest, "extend_seconds must be positive") + return + } + env, err := a.lab.ExtendTTL(r.Context(), id, time.Duration(req.ExtendSeconds)*time.Second) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, env) + case "archive": + if r.Method != http.MethodPost { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + if err := auth.Require(ClaimsFromContextMust(r).Role, auth.ActionStop); err != nil { + writeError(w, http.StatusForbidden, "insufficient role") + return + } + env, err := a.lab.ArchiveAndDestroy(r.Context(), id) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, env) + default: + writeError(w, http.StatusNotFound, "unknown env action") + } +} + +func (a *App) handleLabStorage(w http.ResponseWriter, r *http.Request) { + if a.lab == nil { + writeError(w, http.StatusServiceUnavailable, "lab manager not initialised") + return + } + switch r.Method { + case http.MethodGet: + writeJSON(w, http.StatusOK, a.lab.Storage()) + case http.MethodPut: + if ClaimsFromContextMust(r).Role < auth.RoleAdmin { + writeError(w, http.StatusForbidden, "admin only") + return + } + var req labStorageRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid request body") + return + } + next := lab.StorageConfig{HotTier: req.HotTier, ColdTier: req.ColdTier} + if err := a.lab.UpdateStorage(next); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, a.lab.Storage()) + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +func (a *App) handleLabDrivers(w http.ResponseWriter, r *http.Request) { + if a.lab == nil { + writeError(w, http.StatusServiceUnavailable, "lab manager not initialised") + return + } + switch r.Method { + case http.MethodGet: + writeJSON(w, http.StatusOK, map[string]any{ + "drivers": a.lab.Registry().Names(), + "active": a.lab.Registry().Active(), + }) + case http.MethodPut: + if ClaimsFromContextMust(r).Role < auth.RoleAdmin { + writeError(w, http.StatusForbidden, "admin only") + return + } + var req struct { + Driver string `json:"driver"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid request body") + return + } + if err := a.lab.Registry().SetActive(req.Driver); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{"active": a.lab.Registry().Active()}) + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} diff --git a/internal/api/server.go b/internal/api/server.go index 1f6bebf..7b0c840 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -7,6 +7,7 @@ import ( "time" "github.com/Gradient-Linux/concave/internal/auth" + "github.com/Gradient-Linux/concave/internal/lab" ) // Config holds the API server runtime configuration. @@ -24,6 +25,7 @@ type App struct { workspaceRoot string tokens auth.TokenConfig jobs *JobManager + lab *lab.Manager mux *http.ServeMux } @@ -37,12 +39,27 @@ func New(cfg Config) *App { workspaceRoot: cfg.WorkspaceRoot, tokens: cfg.Tokens, jobs: NewJobManager(), + lab: buildLabManager(), mux: http.NewServeMux(), } app.routes() return app } +func buildLabManager() *lab.Manager { + registry := lab.NewRegistry() + registry.Register(lab.NewDockerDriver(lab.DockerDriverConfig{})) + store, err := lab.NewStore() + if err != nil { + store, _ = lab.NewStoreAt("") + } + storage, _ := lab.LoadStorage() + _ = lab.EnsureTierDirs(storage) + mgr := lab.NewManager(registry, store, storage) + mgr.SetAuditWriter(lab.NewFileAuditWriter()) + return mgr +} + func (a *App) Handler() http.Handler { return a.mux } @@ -54,6 +71,10 @@ func (a *App) ListenAndServe(ctx context.Context) error { ReadHeaderTimeout: 10 * time.Second, } + if a.lab != nil { + go a.lab.RunReaper(ctx, 30*time.Second) + } + go func() { <-ctx.Done() shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) @@ -98,6 +119,10 @@ func (a *App) routes() { a.mux.Handle("/api/v1/suites/", a.authMiddleware(RoleMiddleware(auth.RoleViewer, http.HandlerFunc(a.handleSuiteSubroutes)))) a.mux.Handle("/api/v1/terminal/container/", a.authMiddleware(RoleMiddleware(auth.RoleDeveloper, http.HandlerFunc(a.handleContainerTerminal)))) a.mux.Handle("/api/v1/terminal/host", a.authMiddleware(RoleMiddleware(auth.RoleAdmin, http.HandlerFunc(a.handleHostTerminal)))) + a.mux.Handle("/api/v1/lab/envs", a.authMiddleware(RoleMiddleware(auth.RoleViewer, http.HandlerFunc(a.handleLabEnvs)))) + a.mux.Handle("/api/v1/lab/envs/", a.authMiddleware(RoleMiddleware(auth.RoleViewer, http.HandlerFunc(a.handleLabEnvSubroutes)))) + a.mux.Handle("/api/v1/lab/storage", a.authMiddleware(RoleMiddleware(auth.RoleViewer, http.HandlerFunc(a.handleLabStorage)))) + a.mux.Handle("/api/v1/lab/drivers", a.authMiddleware(RoleMiddleware(auth.RoleViewer, http.HandlerFunc(a.handleLabDrivers)))) } func (a *App) issueSession(w http.ResponseWriter, username string, role auth.Role) (auth.Claims, string, error) { diff --git a/internal/lab/audit.go b/internal/lab/audit.go new file mode 100644 index 0000000..c8dfcf0 --- /dev/null +++ b/internal/lab/audit.go @@ -0,0 +1,132 @@ +package lab + +import ( + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "sync" + "time" + + "github.com/Gradient-Linux/concave/internal/workspace" +) + +// AuditEvent is the record written for each operator action on an env. +// +// The audit log is a Gradient-specific touch on top of the raw lab lifecycle: +// every launch, extend, archive, destroy, and restore is appended here as a +// JSON line. This is the same format the rest of concave will converge on in +// Phase 0.3 (unified audit log). +type AuditEvent struct { + Time time.Time `json:"time"` + Actor string `json:"actor"` + Action string `json:"action"` + EnvID string `json:"env_id,omitempty"` + Driver string `json:"driver,omitempty"` + Details any `json:"details,omitempty"` + PeerID string `json:"peer_id,omitempty"` + Error string `json:"error,omitempty"` +} + +// AuditWriter appends AuditEvents to a sink. Implementations are expected to +// be concurrency-safe. +type AuditWriter interface { + Write(event AuditEvent) error +} + +// FileAuditWriter is a JSONL file writer. +type FileAuditWriter struct { + mu sync.Mutex + path string +} + +// NewFileAuditWriter targets ~/gradient/config/lab-audit.jsonl by default. +func NewFileAuditWriter() *FileAuditWriter { + return &FileAuditWriter{path: workspace.ConfigPath("lab-audit.jsonl")} +} + +// NewFileAuditWriterAt targets an explicit path (useful for tests). +func NewFileAuditWriterAt(path string) *FileAuditWriter { + return &FileAuditWriter{path: path} +} + +// Write appends the event as one JSON line. +func (w *FileAuditWriter) Write(event AuditEvent) error { + w.mu.Lock() + defer w.mu.Unlock() + + if err := os.MkdirAll(filepath.Dir(w.path), 0o700); err != nil { + return fmt.Errorf("ensure audit dir: %w", err) + } + f, err := os.OpenFile(w.path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600) + if err != nil { + return fmt.Errorf("open audit log: %w", err) + } + defer f.Close() + return writeAuditLine(f, event) +} + +func writeAuditLine(w io.Writer, event AuditEvent) error { + data, err := json.Marshal(event) + if err != nil { + return fmt.Errorf("encode audit: %w", err) + } + if _, err := w.Write(append(data, '\n')); err != nil { + return fmt.Errorf("write audit: %w", err) + } + return nil +} + +// DiscardAuditWriter silently drops events; used when no audit path has been +// configured. +type DiscardAuditWriter struct{} + +// Write implements AuditWriter. +func (DiscardAuditWriter) Write(AuditEvent) error { return nil } + +// MemoryAuditWriter is a test-only in-memory writer. +type MemoryAuditWriter struct { + mu sync.Mutex + Events []AuditEvent +} + +// Write implements AuditWriter. +func (m *MemoryAuditWriter) Write(event AuditEvent) error { + m.mu.Lock() + defer m.mu.Unlock() + m.Events = append(m.Events, event) + return nil +} + +// Snapshot returns a defensive copy of the recorded events. +func (m *MemoryAuditWriter) Snapshot() []AuditEvent { + m.mu.Lock() + defer m.mu.Unlock() + out := make([]AuditEvent, len(m.Events)) + copy(out, m.Events) + return out +} + +// localPeerID resolves a Gradient-specific peer identifier used to stamp +// archive manifests. Prefers /etc/machine-id, falls back to hostname. +func localPeerID() string { + if data, err := os.ReadFile("/etc/machine-id"); err == nil { + if id := trimLine(string(data)); id != "" { + return id + } + } + if host, err := os.Hostname(); err == nil && host != "" { + return host + } + return "local" +} + +func trimLine(s string) string { + for i, r := range s { + if r == '\n' || r == '\r' { + return s[:i] + } + } + return s +} diff --git a/internal/lab/driver.go b/internal/lab/driver.go new file mode 100644 index 0000000..210a919 --- /dev/null +++ b/internal/lab/driver.go @@ -0,0 +1,128 @@ +package lab + +import ( + "context" + "errors" + "fmt" + "sync" + "time" +) + +// ArchiveRef identifies a cold-tier archive produced by Driver.Archive. +// +// PeerID is the mesh-addressable identifier of the node that produced the +// archive (machine-id or hostname). Carrying it in the manifest lets a user +// restore an env archived on peer A onto peer B via +// /api/v1/lab/envs/restore. +type ArchiveRef struct { + EnvID string `json:"env_id"` + Path string `json:"path"` + PeerID string `json:"peer_id,omitempty"` + SizeBytes int64 `json:"size_bytes"` + CreatedAt time.Time `json:"created_at"` + OriginSpec EnvSpec `json:"origin_spec"` +} + +// Driver abstracts the backend (docker | slurm | proxmox) that actually runs +// an ephemeral environment. +type Driver interface { + // Name returns the canonical driver name, matching what users pass in + // EnvSpec.Driver. + Name() string + + // OwnsTTL reports whether the driver itself is responsible for enforcing + // TTL (e.g. Slurm's `--time` flag). Drivers that return false rely on the + // reaper to expire environments. + OwnsTTL() bool + + // Launch starts a new env from spec and returns the populated Env record. + Launch(ctx context.Context, spec EnvSpec) (Env, error) + + // Inspect returns the current state of an env. + Inspect(ctx context.Context, env Env) (Env, error) + + // ExtendTTL pushes an env's expiry to the supplied time. Drivers that + // own TTL must relay the extension to the underlying scheduler. + ExtendTTL(ctx context.Context, env Env, until time.Time) (Env, error) + + // Archive copies the env's hot-tier state to the cold tier and returns a + // reference that can be used to restore it later. + Archive(ctx context.Context, env Env) (ArchiveRef, error) + + // Destroy tears down the running env. Storage is not touched; callers are + // expected to call Archive first if they want the state preserved. + Destroy(ctx context.Context, env Env) error +} + +// ErrDriverNotRegistered is returned by the registry when a lookup fails. +var ErrDriverNotRegistered = errors.New("lab driver not registered") + +// Registry is a process-wide map of driver name → Driver. +type Registry struct { + mu sync.RWMutex + drivers map[string]Driver + active string +} + +// NewRegistry creates an empty driver registry. +func NewRegistry() *Registry { + return &Registry{drivers: make(map[string]Driver)} +} + +// Register installs a driver. The first registered driver becomes the active +// one unless SetActive is called afterwards. +func (r *Registry) Register(d Driver) { + r.mu.Lock() + defer r.mu.Unlock() + r.drivers[d.Name()] = d + if r.active == "" { + r.active = d.Name() + } +} + +// SetActive selects the default driver used when an EnvSpec does not specify +// one. Returns ErrDriverNotRegistered if the driver is unknown. +func (r *Registry) SetActive(name string) error { + r.mu.Lock() + defer r.mu.Unlock() + if _, ok := r.drivers[name]; !ok { + return fmt.Errorf("%w: %q", ErrDriverNotRegistered, name) + } + r.active = name + return nil +} + +// Active returns the currently selected default driver name. +func (r *Registry) Active() string { + r.mu.RLock() + defer r.mu.RUnlock() + return r.active +} + +// Get resolves a driver by name. An empty name returns the active driver. +func (r *Registry) Get(name string) (Driver, error) { + r.mu.RLock() + defer r.mu.RUnlock() + if name == "" { + name = r.active + } + if name == "" { + return nil, ErrDriverNotRegistered + } + d, ok := r.drivers[name] + if !ok { + return nil, fmt.Errorf("%w: %q", ErrDriverNotRegistered, name) + } + return d, nil +} + +// Names returns the sorted list of registered driver names. +func (r *Registry) Names() []string { + r.mu.RLock() + defer r.mu.RUnlock() + names := make([]string, 0, len(r.drivers)) + for name := range r.drivers { + names = append(names, name) + } + return names +} diff --git a/internal/lab/driver_docker.go b/internal/lab/driver_docker.go new file mode 100644 index 0000000..38c2820 --- /dev/null +++ b/internal/lab/driver_docker.go @@ -0,0 +1,352 @@ +package lab + +import ( + "archive/tar" + "compress/gzip" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "time" +) + +// Runner abstracts external command execution so the driver can be unit-tested +// without a real Docker daemon. +type Runner interface { + RunCommand(ctx context.Context, name string, args ...string) ([]byte, error) +} + +type defaultRunner struct{} + +func (defaultRunner) RunCommand(ctx context.Context, name string, args ...string) ([]byte, error) { + return exec.CommandContext(ctx, name, args...).CombinedOutput() +} + +// PortAllocator picks a host port for a Jupyter environment. Defaults to a +// net.Listen-based allocator; tests override it. +type PortAllocator func() (int, error) + +func allocateRandomPort() (int, error) { + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + return 0, fmt.Errorf("allocate port: %w", err) + } + defer l.Close() + return l.Addr().(*net.TCPAddr).Port, nil +} + +// DockerDriverConfig holds injectable dependencies for the Docker driver. +type DockerDriverConfig struct { + Runner Runner + AllocPort PortAllocator + NowFn func() time.Time +} + +// NewDockerDriver builds a Docker-backed Driver. Any nil field falls back to +// the production default. +func NewDockerDriver(cfg DockerDriverConfig) *DockerDriver { + if cfg.Runner == nil { + cfg.Runner = defaultRunner{} + } + if cfg.AllocPort == nil { + cfg.AllocPort = allocateRandomPort + } + if cfg.NowFn == nil { + cfg.NowFn = time.Now + } + return &DockerDriver{cfg: cfg} +} + +// DockerDriver is the reference Driver implementation. It launches each env +// as a dedicated `docker run` container with the hot-tier directory mounted +// at /home/jovyan/work and a randomly-allocated host port forwarded to 8888. +type DockerDriver struct { + cfg DockerDriverConfig +} + +// Name implements Driver. +func (d *DockerDriver) Name() string { return "docker" } + +// OwnsTTL implements Driver. The Docker driver does not own TTL — the reaper +// enforces expiry. +func (d *DockerDriver) OwnsTTL() bool { return false } + +// Launch creates the hot-tier scratch directory, pulls the image (already +// cached when possible), runs the container detached, and returns the +// populated Env record. +func (d *DockerDriver) Launch(ctx context.Context, spec EnvSpec) (Env, error) { + if err := validateSpec(spec); err != nil { + return Env{}, err + } + if err := ensureDir(spec.HotTierPath); err != nil { + return Env{}, err + } + + id, err := newEnvID() + if err != nil { + return Env{}, err + } + envHot := filepath.Join(spec.HotTierPath, id) + if err := ensureDir(envHot); err != nil { + return Env{}, err + } + + port, err := d.cfg.AllocPort() + if err != nil { + return Env{}, err + } + token, err := newJupyterToken() + if err != nil { + return Env{}, err + } + + containerName := "gradient-lab-" + id + args := []string{ + "run", "-d", + "--name", containerName, + "--label", "gradient.lab.env_id=" + id, + "--label", "gradient.lab.owner=" + spec.Owner, + "-p", fmt.Sprintf("127.0.0.1:%d:8888", port), + "-v", envHot + ":/home/jovyan/work", + "-e", "JUPYTER_TOKEN=" + token, + } + if spec.CPURequest != "" { + args = append(args, "--cpus", spec.CPURequest) + } + if spec.MemRequest != "" { + args = append(args, "--memory", spec.MemRequest) + } + if spec.GPUs > 0 { + args = append(args, "--gpus", strconv.Itoa(spec.GPUs)) + } + args = append(args, spec.Image, + "start-notebook.sh", + "--NotebookApp.token="+token, + "--NotebookApp.ip=0.0.0.0", + ) + + out, err := d.cfg.Runner.RunCommand(ctx, "docker", args...) + if err != nil { + return Env{}, fmt.Errorf("docker run: %w: %s", err, strings.TrimSpace(string(out))) + } + + now := d.cfg.NowFn() + env := Env{ + ID: id, + Driver: d.Name(), + Owner: spec.Owner, + Image: spec.Image, + DisplayName: spec.DisplayName, + Status: StatusRunning, + ContainerID: strings.TrimSpace(string(out)), + JupyterURL: fmt.Sprintf("http://127.0.0.1:%d/lab?token=%s", port, token), + Token: token, + GPUs: spec.GPUs, + CPURequest: spec.CPURequest, + MemRequest: spec.MemRequest, + HotTierPath: envHot, + ColdTierPath: spec.ColdTierPath, + CreatedAt: now, + ExpiresAt: now.Add(spec.TTL), + } + return env, nil +} + +// Inspect implements Driver by running `docker inspect --format {{.State.Status}}`. +func (d *DockerDriver) Inspect(ctx context.Context, env Env) (Env, error) { + if env.ContainerID == "" { + return env, nil + } + out, err := d.cfg.Runner.RunCommand(ctx, "docker", "inspect", "--format", "{{.State.Status}}", env.ContainerID) + if err != nil { + env.LastError = strings.TrimSpace(string(out)) + return env, nil + } + state := strings.TrimSpace(string(out)) + switch state { + case "running", "restarting": + env.Status = StatusRunning + case "exited", "dead": + if env.Status != StatusArchived { + env.Status = StatusFailed + env.LastError = "container exited: " + state + } + case "removing", "paused", "created": + // leave status untouched; reaper will follow up + } + return env, nil +} + +// ExtendTTL updates the env's expiry in-place. The reaper picks up the new +// deadline on its next tick. +func (d *DockerDriver) ExtendTTL(_ context.Context, env Env, until time.Time) (Env, error) { + if until.Before(d.cfg.NowFn()) { + return env, errors.New("extension target is in the past") + } + env.ExpiresAt = until + if env.Status == StatusExpiring { + env.Status = StatusRunning + } + return env, nil +} + +// Archive creates `/.tar.gz` and a sidecar manifest. +func (d *DockerDriver) Archive(_ context.Context, env Env) (ArchiveRef, error) { + if env.ColdTierPath == "" { + return ArchiveRef{}, errors.New("cold_tier_path is not configured") + } + if err := ensureDir(env.ColdTierPath); err != nil { + return ArchiveRef{}, err + } + archivePath := filepath.Join(env.ColdTierPath, env.ID+".tar.gz") + size, err := archiveHotTier(env.HotTierPath, archivePath) + if err != nil { + return ArchiveRef{}, err + } + manifest := ArchiveRef{ + EnvID: env.ID, + Path: archivePath, + PeerID: localPeerID(), + SizeBytes: size, + CreatedAt: d.cfg.NowFn(), + OriginSpec: EnvSpec{ + Owner: env.Owner, + Image: env.Image, + DisplayName: env.DisplayName, + GPUs: env.GPUs, + CPURequest: env.CPURequest, + MemRequest: env.MemRequest, + HotTierPath: env.HotTierPath, + ColdTierPath: env.ColdTierPath, + Driver: env.Driver, + }, + } + if err := writeManifest(archivePath+".json", manifest); err != nil { + return ArchiveRef{}, err + } + return manifest, nil +} + +// Destroy stops and removes the container, and wipes the hot-tier directory. +func (d *DockerDriver) Destroy(ctx context.Context, env Env) error { + if env.ContainerID != "" { + if _, err := d.cfg.Runner.RunCommand(ctx, "docker", "rm", "-f", env.ContainerID); err != nil { + return fmt.Errorf("docker rm %s: %w", env.ContainerID, err) + } + } + if env.HotTierPath != "" { + if err := os.RemoveAll(env.HotTierPath); err != nil { + return fmt.Errorf("remove hot tier %s: %w", env.HotTierPath, err) + } + } + return nil +} + +func validateSpec(spec EnvSpec) error { + if strings.TrimSpace(spec.Image) == "" { + return errors.New("image is required") + } + if strings.TrimSpace(spec.Owner) == "" { + return errors.New("owner is required") + } + if spec.TTL <= 0 { + return errors.New("ttl must be positive") + } + if spec.TTL > 7*24*time.Hour { + return errors.New("ttl may not exceed 7 days") + } + if spec.HotTierPath == "" { + return errors.New("hot_tier_path is required") + } + return nil +} + +func ensureDir(path string) error { + if path == "" { + return nil + } + return os.MkdirAll(path, 0o755) +} + +func newJupyterToken() (string, error) { + var buf [16]byte + if _, err := rand.Read(buf[:]); err != nil { + return "", fmt.Errorf("generate jupyter token: %w", err) + } + return hex.EncodeToString(buf[:]), nil +} + +func archiveHotTier(src, dest string) (int64, error) { + out, err := os.Create(dest) + if err != nil { + return 0, fmt.Errorf("create archive: %w", err) + } + defer out.Close() + + gz := gzip.NewWriter(out) + tw := tar.NewWriter(gz) + err = filepath.Walk(src, func(path string, info os.FileInfo, walkErr error) error { + if walkErr != nil { + return walkErr + } + rel, err := filepath.Rel(src, path) + if err != nil { + return err + } + if rel == "." { + return nil + } + hdr, err := tar.FileInfoHeader(info, "") + if err != nil { + return err + } + hdr.Name = rel + if err := tw.WriteHeader(hdr); err != nil { + return err + } + if !info.Mode().IsRegular() { + return nil + } + f, err := os.Open(path) + if err != nil { + return err + } + defer f.Close() + _, err = io.Copy(tw, f) + return err + }) + if err != nil { + return 0, fmt.Errorf("walk hot tier: %w", err) + } + if err := tw.Close(); err != nil { + return 0, fmt.Errorf("close tar: %w", err) + } + if err := gz.Close(); err != nil { + return 0, fmt.Errorf("close gzip: %w", err) + } + stat, err := out.Stat() + if err != nil { + return 0, fmt.Errorf("stat archive: %w", err) + } + return stat.Size(), nil +} + +func writeManifest(path string, manifest ArchiveRef) error { + data, err := json.MarshalIndent(manifest, "", " ") + if err != nil { + return fmt.Errorf("encode archive manifest: %w", err) + } + if err := os.WriteFile(path, data, 0o600); err != nil { + return fmt.Errorf("write archive manifest: %w", err) + } + return nil +} diff --git a/internal/lab/lab_test.go b/internal/lab/lab_test.go new file mode 100644 index 0000000..1beb15f --- /dev/null +++ b/internal/lab/lab_test.go @@ -0,0 +1,420 @@ +package lab + +import ( + "context" + "encoding/json" + "errors" + "os" + "path/filepath" + "sync" + "testing" + "time" +) + +type fakeDriver struct { + name string + owns bool + launch func(context.Context, EnvSpec) (Env, error) + dest func(context.Context, Env) error + arch func(context.Context, Env) (ArchiveRef, error) + + mu sync.Mutex + destroyed []string + archived []string + extended []time.Time + inspectCalls int +} + +func (f *fakeDriver) Name() string { return f.name } +func (f *fakeDriver) OwnsTTL() bool { return f.owns } + +func (f *fakeDriver) Launch(ctx context.Context, spec EnvSpec) (Env, error) { + if f.launch != nil { + return f.launch(ctx, spec) + } + id, err := newEnvID() + if err != nil { + return Env{}, err + } + return Env{ + ID: id, + Driver: f.name, + Owner: spec.Owner, + Image: spec.Image, + Status: StatusRunning, + HotTierPath: spec.HotTierPath, + ColdTierPath: spec.ColdTierPath, + CreatedAt: time.Unix(0, 0), + ExpiresAt: time.Unix(0, 0).Add(spec.TTL), + }, nil +} + +func (f *fakeDriver) Inspect(ctx context.Context, env Env) (Env, error) { + f.mu.Lock() + f.inspectCalls++ + f.mu.Unlock() + return env, nil +} + +func (f *fakeDriver) ExtendTTL(_ context.Context, env Env, until time.Time) (Env, error) { + f.mu.Lock() + f.extended = append(f.extended, until) + f.mu.Unlock() + env.ExpiresAt = until + return env, nil +} + +func (f *fakeDriver) Archive(ctx context.Context, env Env) (ArchiveRef, error) { + if f.arch != nil { + return f.arch(ctx, env) + } + f.mu.Lock() + f.archived = append(f.archived, env.ID) + f.mu.Unlock() + return ArchiveRef{EnvID: env.ID, Path: "/cold/" + env.ID + ".tar.gz", CreatedAt: time.Unix(0, 0)}, nil +} + +func (f *fakeDriver) Destroy(ctx context.Context, env Env) error { + if f.dest != nil { + return f.dest(ctx, env) + } + f.mu.Lock() + f.destroyed = append(f.destroyed, env.ID) + f.mu.Unlock() + return nil +} + +func tempManager(t *testing.T, driver Driver, storage StorageConfig) *Manager { + t.Helper() + tmp := t.TempDir() + store, err := NewStoreAt(filepath.Join(tmp, "lab-envs.json")) + if err != nil { + t.Fatalf("NewStoreAt: %v", err) + } + registry := NewRegistry() + registry.Register(driver) + return NewManager(registry, store, storage) +} + +func TestStoreCreateAndList(t *testing.T) { + tmp := t.TempDir() + store, err := NewStoreAt(filepath.Join(tmp, "lab-envs.json")) + if err != nil { + t.Fatalf("NewStoreAt: %v", err) + } + created, err := store.Create(Env{CreatedAt: time.Unix(1, 0)}) + if err != nil { + t.Fatalf("Create: %v", err) + } + if created.ID == "" { + t.Fatal("expected auto-assigned ID") + } + envs := store.List() + if len(envs) != 1 || envs[0].ID != created.ID { + t.Fatalf("List = %+v", envs) + } + reopened, err := NewStoreAt(filepath.Join(tmp, "lab-envs.json")) + if err != nil { + t.Fatalf("reopen: %v", err) + } + if _, ok := reopened.Get(created.ID); !ok { + t.Fatal("expected record after reopen") + } +} + +func TestStoreUpdateAndDelete(t *testing.T) { + tmp := t.TempDir() + store, _ := NewStoreAt(filepath.Join(tmp, "lab-envs.json")) + env, _ := store.Create(Env{ID: "env-1", CreatedAt: time.Unix(1, 0), Status: StatusRunning}) + env.Status = StatusArchived + if err := store.Update(env); err != nil { + t.Fatalf("Update: %v", err) + } + got, _ := store.Get("env-1") + if got.Status != StatusArchived { + t.Fatalf("status = %q", got.Status) + } + if err := store.Delete("env-1"); err != nil { + t.Fatalf("Delete: %v", err) + } + if len(store.List()) != 0 { + t.Fatal("expected empty list after delete") + } +} + +func TestStorageConfigRoundTrip(t *testing.T) { + tmp := t.TempDir() + prev := storagePath + storagePath = func() string { return filepath.Join(tmp, "lab.json") } + t.Cleanup(func() { storagePath = prev }) + + if err := SaveStorage(StorageConfig{HotTier: "/tmp/hot", ColdTier: "/tmp/cold"}); err != nil { + t.Fatalf("SaveStorage: %v", err) + } + loaded, err := LoadStorage() + if err != nil { + t.Fatalf("LoadStorage: %v", err) + } + if loaded.HotTier != "/tmp/hot" || loaded.ColdTier != "/tmp/cold" { + t.Fatalf("loaded = %+v", loaded) + } + + if err := SaveStorage(StorageConfig{HotTier: "/a", ColdTier: "/a"}); err == nil { + t.Fatal("expected duplicate path rejection") + } + if err := SaveStorage(StorageConfig{HotTier: "relative", ColdTier: "/b"}); err == nil { + t.Fatal("expected absolute-path rejection") + } +} + +func TestRegistryActiveAndLookup(t *testing.T) { + r := NewRegistry() + r.Register(&fakeDriver{name: "docker"}) + r.Register(&fakeDriver{name: "slurm", owns: true}) + if r.Active() != "docker" { + t.Fatalf("Active = %q", r.Active()) + } + if err := r.SetActive("slurm"); err != nil { + t.Fatalf("SetActive: %v", err) + } + if r.Active() != "slurm" { + t.Fatalf("Active = %q", r.Active()) + } + if err := r.SetActive("proxmox"); !errors.Is(err, ErrDriverNotRegistered) { + t.Fatalf("SetActive unknown = %v", err) + } + if _, err := r.Get("docker"); err != nil { + t.Fatalf("Get docker: %v", err) + } +} + +func TestManagerLaunchExtendArchive(t *testing.T) { + driver := &fakeDriver{name: "docker"} + now := time.Unix(1_000, 0) + mgr := tempManager(t, driver, StorageConfig{HotTier: "/h", ColdTier: "/c"}) + mgr.SetNowFn(func() time.Time { return now }) + audit := &MemoryAuditWriter{} + mgr.SetAuditWriter(audit) + + launched, err := mgr.Launch(context.Background(), EnvSpec{ + Owner: "alice", Image: "jupyter/minimal", TTL: 30 * time.Minute, + }) + if err != nil { + t.Fatalf("Launch: %v", err) + } + if launched.Driver != "docker" || launched.HotTierPath != "/h" || launched.ColdTierPath != "/c" { + t.Fatalf("launched = %+v", launched) + } + extended, err := mgr.ExtendTTL(context.Background(), launched.ID, time.Hour) + if err != nil { + t.Fatalf("ExtendTTL: %v", err) + } + if got := extended.ExpiresAt.Sub(launched.ExpiresAt); got != time.Hour { + t.Fatalf("extension delta = %s", got) + } + archived, err := mgr.ArchiveAndDestroy(context.Background(), launched.ID) + if err != nil { + t.Fatalf("ArchiveAndDestroy: %v", err) + } + if archived.Status != StatusArchived { + t.Fatalf("archived.Status = %q", archived.Status) + } + if len(driver.archived) != 1 || len(driver.destroyed) != 1 { + t.Fatalf("driver counts: archived=%d destroyed=%d", len(driver.archived), len(driver.destroyed)) + } + events := audit.Snapshot() + actions := make([]string, 0, len(events)) + for _, e := range events { + actions = append(actions, e.Action) + } + wantActions := []string{"lab.launch", "lab.extend", "lab.archive"} + if len(events) != len(wantActions) { + t.Fatalf("audit event count = %d (%v)", len(events), actions) + } + for i, want := range wantActions { + if events[i].Action != want { + t.Fatalf("audit[%d].Action = %q, want %q", i, events[i].Action, want) + } + if events[i].PeerID == "" { + t.Fatalf("audit[%d].PeerID empty", i) + } + } +} + +func TestReaperArchivesOnlyExpired(t *testing.T) { + driver := &fakeDriver{name: "docker"} + now := time.Unix(1_000, 0) + mgr := tempManager(t, driver, StorageConfig{HotTier: "/h", ColdTier: "/c"}) + mgr.SetNowFn(func() time.Time { return now }) + + expired, _ := mgr.Launch(context.Background(), EnvSpec{Owner: "a", Image: "i", TTL: time.Minute}) + alive, _ := mgr.Launch(context.Background(), EnvSpec{Owner: "a", Image: "i", TTL: time.Hour}) + + mgr.SetNowFn(func() time.Time { return now.Add(2 * time.Minute) }) + if errs := mgr.ReapExpired(context.Background()); len(errs) != 0 { + t.Fatalf("reap errors: %v", errs) + } + expiredEnv, _ := mgr.Get(context.Background(), expired.ID) + aliveEnv, _ := mgr.Get(context.Background(), alive.ID) + if expiredEnv.Status != StatusArchived { + t.Fatalf("expired.Status = %q", expiredEnv.Status) + } + if aliveEnv.Status != StatusRunning { + t.Fatalf("alive.Status = %q", aliveEnv.Status) + } +} + +func TestReaperSkipsSchedulerOwnedTTL(t *testing.T) { + driver := &fakeDriver{name: "slurm", owns: true} + now := time.Unix(1_000, 0) + mgr := tempManager(t, driver, StorageConfig{HotTier: "/h", ColdTier: "/c"}) + mgr.SetNowFn(func() time.Time { return now }) + + env, _ := mgr.Launch(context.Background(), EnvSpec{Owner: "a", Image: "i", TTL: time.Minute}) + mgr.SetNowFn(func() time.Time { return now.Add(time.Hour) }) + if errs := mgr.ReapExpired(context.Background()); len(errs) != 0 { + t.Fatalf("reap errors: %v", errs) + } + refreshed, _ := mgr.Get(context.Background(), env.ID) + if refreshed.Status != StatusRunning { + t.Fatalf("scheduler-owned env archived unexpectedly: %q", refreshed.Status) + } + if len(driver.archived) != 0 { + t.Fatalf("scheduler-owned driver should not archive, got %d", len(driver.archived)) + } +} + +type stubRunner struct { + calls []struct { + name string + args []string + } + out []byte + err error +} + +func (s *stubRunner) RunCommand(_ context.Context, name string, args ...string) ([]byte, error) { + s.calls = append(s.calls, struct { + name string + args []string + }{name, append([]string(nil), args...)}) + return s.out, s.err +} + +func TestDockerDriverLaunchComposesDockerRun(t *testing.T) { + runner := &stubRunner{out: []byte("abc123\n")} + driver := NewDockerDriver(DockerDriverConfig{ + Runner: runner, + AllocPort: func() (int, error) { return 18888, nil }, + NowFn: func() time.Time { return time.Unix(0, 0) }, + }) + hot := t.TempDir() + + env, err := driver.Launch(context.Background(), EnvSpec{ + Owner: "alice", Image: "jupyter/minimal", TTL: time.Hour, + HotTierPath: hot, ColdTierPath: "/cold", + GPUs: 1, CPURequest: "2", MemRequest: "4g", + }) + if err != nil { + t.Fatalf("Launch: %v", err) + } + if env.ContainerID != "abc123" { + t.Fatalf("ContainerID = %q", env.ContainerID) + } + if env.JupyterURL == "" || env.Token == "" { + t.Fatal("JupyterURL/Token should be populated") + } + if len(runner.calls) != 1 || runner.calls[0].name != "docker" { + t.Fatalf("calls = %+v", runner.calls) + } + args := runner.calls[0].args + mustContain(t, args, "--gpus") + mustContain(t, args, "--cpus") + mustContain(t, args, "--memory") + mustContain(t, args, "jupyter/minimal") +} + +func TestDockerDriverArchiveWritesTarball(t *testing.T) { + runner := &stubRunner{} + driver := NewDockerDriver(DockerDriverConfig{ + Runner: runner, + AllocPort: func() (int, error) { return 1, nil }, + NowFn: func() time.Time { return time.Unix(1, 0) }, + }) + hot := t.TempDir() + cold := t.TempDir() + if err := os.WriteFile(filepath.Join(hot, "hello.txt"), []byte("hi"), 0o600); err != nil { + t.Fatalf("write hot file: %v", err) + } + + ref, err := driver.Archive(context.Background(), Env{ + ID: "env-xyz", HotTierPath: hot, ColdTierPath: cold, Driver: "docker", + }) + if err != nil { + t.Fatalf("Archive: %v", err) + } + if ref.Path != filepath.Join(cold, "env-xyz.tar.gz") { + t.Fatalf("ref.Path = %q", ref.Path) + } + if ref.SizeBytes <= 0 { + t.Fatalf("ref.SizeBytes = %d", ref.SizeBytes) + } + manifest, err := os.ReadFile(ref.Path + ".json") + if err != nil { + t.Fatalf("manifest: %v", err) + } + var decoded ArchiveRef + if err := json.Unmarshal(manifest, &decoded); err != nil { + t.Fatalf("manifest decode: %v", err) + } + if decoded.EnvID != "env-xyz" { + t.Fatalf("manifest envID = %q", decoded.EnvID) + } +} + +func TestDockerDriverValidateSpec(t *testing.T) { + cases := map[string]EnvSpec{ + "missing image": {Owner: "a", TTL: time.Minute, HotTierPath: "/h"}, + "missing owner": {Image: "i", TTL: time.Minute, HotTierPath: "/h"}, + "zero ttl": {Owner: "a", Image: "i", HotTierPath: "/h"}, + "excess ttl": {Owner: "a", Image: "i", TTL: 10 * 24 * time.Hour, HotTierPath: "/h"}, + "missing hot": {Owner: "a", Image: "i", TTL: time.Minute}, + } + for name, spec := range cases { + t.Run(name, func(t *testing.T) { + if err := validateSpec(spec); err == nil { + t.Fatal("expected validation error") + } + }) + } +} + +func TestEnvRemainingAndExpired(t *testing.T) { + base := time.Unix(1_000, 0) + env := Env{ExpiresAt: base.Add(5 * time.Minute)} + if env.Expired(base) { + t.Fatal("should not be expired yet") + } + if env.Expired(base.Add(4 * time.Minute)) { + t.Fatal("should not be expired at t+4m") + } + if !env.Expired(base.Add(6 * time.Minute)) { + t.Fatal("should be expired at t+6m") + } + if got := env.Remaining(base); got != 5*time.Minute { + t.Fatalf("Remaining = %s", got) + } + if got := env.Remaining(base.Add(10 * time.Minute)); got != 0 { + t.Fatalf("Remaining after expiry = %s", got) + } +} + +func mustContain(t *testing.T, args []string, needle string) { + t.Helper() + for _, a := range args { + if a == needle { + return + } + } + t.Fatalf("args missing %q: %v", needle, args) +} diff --git a/internal/lab/manager.go b/internal/lab/manager.go new file mode 100644 index 0000000..307d3bd --- /dev/null +++ b/internal/lab/manager.go @@ -0,0 +1,301 @@ +package lab + +import ( + "context" + "errors" + "fmt" + "sync" + "time" +) + +// Manager is the high-level façade combining a Registry, a Store, and a +// StorageConfig. HTTP handlers and CLI commands interact with the Manager +// rather than the underlying building blocks directly. +type Manager struct { + registry *Registry + store *Store + storage StorageConfig + audit AuditWriter + nowFn func() time.Time + + mu sync.Mutex +} + +// NewManager constructs a Manager. Use LoadStorage() to populate the storage +// tier defaults before calling this. +func NewManager(registry *Registry, store *Store, storage StorageConfig) *Manager { + return &Manager{ + registry: registry, + store: store, + storage: storage, + audit: DiscardAuditWriter{}, + nowFn: time.Now, + } +} + +// SetAuditWriter installs an audit sink. Defaults to DiscardAuditWriter. +func (m *Manager) SetAuditWriter(w AuditWriter) { + if w == nil { + w = DiscardAuditWriter{} + } + m.mu.Lock() + defer m.mu.Unlock() + m.audit = w +} + +func (m *Manager) auditWriter() AuditWriter { + m.mu.Lock() + defer m.mu.Unlock() + return m.audit +} + +func (m *Manager) record(event AuditEvent) { + if event.Time.IsZero() { + event.Time = m.nowFn() + } + if event.PeerID == "" { + event.PeerID = localPeerID() + } + _ = m.auditWriter().Write(event) +} + +// SetNowFn is a test seam. +func (m *Manager) SetNowFn(fn func() time.Time) { + m.mu.Lock() + defer m.mu.Unlock() + m.nowFn = fn +} + +// Storage returns a copy of the current storage configuration. +func (m *Manager) Storage() StorageConfig { + m.mu.Lock() + defer m.mu.Unlock() + return m.storage +} + +// UpdateStorage persists a new storage configuration. +func (m *Manager) UpdateStorage(next StorageConfig) error { + if err := SaveStorage(next); err != nil { + return err + } + if err := EnsureTierDirs(next); err != nil { + return err + } + m.mu.Lock() + m.storage = next + m.mu.Unlock() + return nil +} + +// Registry exposes the underlying registry for driver-aware CLI commands. +func (m *Manager) Registry() *Registry { return m.registry } + +// Launch provisions a new env via the selected driver and persists the record. +func (m *Manager) Launch(ctx context.Context, spec EnvSpec) (Env, error) { + m.mu.Lock() + storage := m.storage + m.mu.Unlock() + + if spec.HotTierPath == "" { + spec.HotTierPath = storage.HotTier + } + if spec.ColdTierPath == "" { + spec.ColdTierPath = storage.ColdTier + } + if spec.Driver == "" { + spec.Driver = m.registry.Active() + } + driver, err := m.registry.Get(spec.Driver) + if err != nil { + return Env{}, err + } + env, err := driver.Launch(ctx, spec) + if err != nil { + m.record(AuditEvent{Actor: spec.Owner, Action: "lab.launch", Driver: spec.Driver, Error: err.Error()}) + return Env{}, err + } + created, err := m.store.Create(env) + if err != nil { + m.record(AuditEvent{Actor: spec.Owner, Action: "lab.launch", Driver: spec.Driver, Error: err.Error()}) + return Env{}, err + } + m.record(AuditEvent{ + Actor: created.Owner, + Action: "lab.launch", + EnvID: created.ID, + Driver: created.Driver, + Details: map[string]any{ + "image": created.Image, + "ttl": spec.TTL.String(), + }, + }) + return created, nil +} + +// Get returns a record by ID with a driver-level inspect refresh. +func (m *Manager) Get(ctx context.Context, id string) (Env, error) { + env, ok := m.store.Get(id) + if !ok { + return Env{}, fmt.Errorf("lab env %q not found", id) + } + driver, err := m.registry.Get(env.Driver) + if err != nil { + return env, nil + } + refreshed, err := driver.Inspect(ctx, env) + if err != nil { + return env, nil + } + if refreshed.Status != env.Status || refreshed.LastError != env.LastError { + _ = m.store.Update(refreshed) + } + return refreshed, nil +} + +// List returns all env records without a driver refresh. +func (m *Manager) List() []Env { + return m.store.List() +} + +// ExtendTTL pushes the expiry of an existing env out. +func (m *Manager) ExtendTTL(ctx context.Context, id string, extension time.Duration) (Env, error) { + if extension <= 0 { + return Env{}, errors.New("extension must be positive") + } + env, ok := m.store.Get(id) + if !ok { + return Env{}, fmt.Errorf("lab env %q not found", id) + } + if !env.Active() { + return Env{}, errors.New("cannot extend a non-active env") + } + driver, err := m.registry.Get(env.Driver) + if err != nil { + return Env{}, err + } + now := m.nowFn() + base := env.ExpiresAt + if base.Before(now) { + base = now + } + refreshed, err := driver.ExtendTTL(ctx, env, base.Add(extension)) + if err != nil { + return Env{}, err + } + if err := m.store.Update(refreshed); err != nil { + return Env{}, err + } + m.record(AuditEvent{ + Actor: env.Owner, + Action: "lab.extend", + EnvID: env.ID, + Driver: env.Driver, + Details: map[string]any{"extend": extension.String(), "new_expiry": refreshed.ExpiresAt}, + }) + return refreshed, nil +} + +// ArchiveAndDestroy archives the env's hot tier to the cold tier, then tears +// the env down. Used by both the reaper and explicit admin actions. +func (m *Manager) ArchiveAndDestroy(ctx context.Context, id string) (Env, error) { + env, ok := m.store.Get(id) + if !ok { + return Env{}, fmt.Errorf("lab env %q not found", id) + } + driver, err := m.registry.Get(env.Driver) + if err != nil { + return env, err + } + env.Status = StatusArchiving + _ = m.store.Update(env) + + ref, err := driver.Archive(ctx, env) + if err != nil { + env.Status = StatusFailed + env.LastError = "archive: " + err.Error() + _ = m.store.Update(env) + return env, err + } + if err := driver.Destroy(ctx, env); err != nil { + env.LastError = "destroy: " + err.Error() + _ = m.store.Update(env) + return env, err + } + now := m.nowFn() + env.Status = StatusArchived + env.ArchiveRef = ref.Path + env.ArchivedAt = &now + if err := m.store.Update(env); err != nil { + return env, err + } + m.record(AuditEvent{ + Actor: env.Owner, + Action: "lab.archive", + EnvID: env.ID, + Driver: env.Driver, + PeerID: ref.PeerID, + Details: map[string]any{ + "archive_ref": ref.Path, + "size_bytes": ref.SizeBytes, + }, + }) + return env, nil +} + +// Destroy removes an env without archiving it. Admin-only. +func (m *Manager) Destroy(ctx context.Context, id string) error { + env, ok := m.store.Get(id) + if !ok { + return nil + } + driver, err := m.registry.Get(env.Driver) + if err != nil { + return err + } + if err := driver.Destroy(ctx, env); err != nil { + return err + } + return m.store.Delete(id) +} + +// ReapExpired walks the store and archives every expired, active env whose +// driver does not own TTL. Intended to be called on a ticker from +// `concave serve`. +func (m *Manager) ReapExpired(ctx context.Context) []error { + now := m.nowFn() + var errs []error + for _, env := range m.store.List() { + if !env.Active() || !env.Expired(now) { + continue + } + driver, err := m.registry.Get(env.Driver) + if err != nil { + errs = append(errs, err) + continue + } + if driver.OwnsTTL() { + continue + } + if _, err := m.ArchiveAndDestroy(ctx, env.ID); err != nil { + errs = append(errs, fmt.Errorf("reap %s: %w", env.ID, err)) + } + } + return errs +} + +// RunReaper blocks until ctx is done, calling ReapExpired on each tick. +func (m *Manager) RunReaper(ctx context.Context, interval time.Duration) { + if interval <= 0 { + interval = 30 * time.Second + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + _ = m.ReapExpired(ctx) + } + } +} diff --git a/internal/lab/storage.go b/internal/lab/storage.go new file mode 100644 index 0000000..3b86f8e --- /dev/null +++ b/internal/lab/storage.go @@ -0,0 +1,123 @@ +package lab + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + + "github.com/Gradient-Linux/concave/internal/workspace" +) + +// StorageConfig describes the two storage tiers every ephemeral env relies on. +// +// HotTier is where the live environment's scratch workspace lives — typically +// the primary workspace disk (NVMe/SSD). It is short-lived: created at env +// launch, destroyed when the env is destroyed. +// +// ColdTier is the sysadmin-configurable destination for archival tarballs. It +// defaults to the slowest writable non-root block device detected at setup +// time (typically an HDD) and can be reassigned with `concave lab storage +// set-cold `. +type StorageConfig struct { + HotTier string `json:"hot_tier"` + ColdTier string `json:"cold_tier"` +} + +// Default returns the default StorageConfig used when no lab.json is found. +func defaultStorage() StorageConfig { + return StorageConfig{ + HotTier: filepath.Join(workspace.Root(), "envs"), + ColdTier: filepath.Join(workspace.Root(), "envs-archive"), + } +} + +var ( + storageMu sync.Mutex + // storagePath is overridable for tests. + storagePath = func() string { + return workspace.ConfigPath("lab.json") + } +) + +// LoadStorage reads the persisted storage config, falling back to defaults +// when the file is missing. +func LoadStorage() (StorageConfig, error) { + storageMu.Lock() + defer storageMu.Unlock() + + cfg := defaultStorage() + data, err := os.ReadFile(storagePath()) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return cfg, nil + } + return cfg, fmt.Errorf("read lab storage: %w", err) + } + if err := json.Unmarshal(data, &cfg); err != nil { + return defaultStorage(), fmt.Errorf("parse lab storage: %w", err) + } + if cfg.HotTier == "" { + cfg.HotTier = defaultStorage().HotTier + } + if cfg.ColdTier == "" { + cfg.ColdTier = defaultStorage().ColdTier + } + return cfg, nil +} + +// SaveStorage persists the storage config atomically. +func SaveStorage(cfg StorageConfig) error { + if err := validateTierPath("hot_tier", cfg.HotTier); err != nil { + return err + } + if err := validateTierPath("cold_tier", cfg.ColdTier); err != nil { + return err + } + if cfg.HotTier == cfg.ColdTier { + return errors.New("hot_tier and cold_tier must be different paths") + } + + storageMu.Lock() + defer storageMu.Unlock() + + path := storagePath() + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return fmt.Errorf("ensure lab config dir: %w", err) + } + data, err := json.MarshalIndent(cfg, "", " ") + if err != nil { + return fmt.Errorf("encode lab storage: %w", err) + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, data, 0o600); err != nil { + return fmt.Errorf("write lab storage: %w", err) + } + if err := os.Rename(tmp, path); err != nil { + return fmt.Errorf("rename lab storage: %w", err) + } + return nil +} + +// EnsureTierDirs ensures both storage tier directories exist. +func EnsureTierDirs(cfg StorageConfig) error { + for _, dir := range []string{cfg.HotTier, cfg.ColdTier} { + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("ensure tier dir %s: %w", dir, err) + } + } + return nil +} + +func validateTierPath(label, path string) error { + if strings.TrimSpace(path) == "" { + return fmt.Errorf("%s is required", label) + } + if !filepath.IsAbs(path) { + return fmt.Errorf("%s must be an absolute path", label) + } + return nil +} diff --git a/internal/lab/store.go b/internal/lab/store.go new file mode 100644 index 0000000..e5eb33d --- /dev/null +++ b/internal/lab/store.go @@ -0,0 +1,178 @@ +package lab + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "sync" + + "github.com/Gradient-Linux/concave/internal/workspace" +) + +// Store persists env records on disk under ~/gradient/config/lab-envs.json. +// +// The on-disk representation is deliberately append-update-delete style JSON: +// concurrent writes are serialised through the Store mutex and snapshots are +// rewritten atomically via `*.tmp` + rename. This is good enough for a +// single-node control plane and avoids taking on sqlite as a dependency. +type Store struct { + mu sync.Mutex + path string + envs map[string]Env +} + +// defaultEnvStorePath returns the default on-disk path. Overridable in tests. +var envStorePath = func() string { + return workspace.ConfigPath("lab-envs.json") +} + +// NewStore creates a Store rooted at the default path and loads any existing +// records. +func NewStore() (*Store, error) { + return NewStoreAt(envStorePath()) +} + +// NewStoreAt is a test seam for targeting an arbitrary path. +func NewStoreAt(path string) (*Store, error) { + s := &Store{path: path, envs: make(map[string]Env)} + if err := s.load(); err != nil { + return nil, err + } + return s, nil +} + +// Create assigns a new ID and persists the record. CreatedAt / ExpiresAt on +// the supplied env are preserved. +func (s *Store) Create(env Env) (Env, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if env.ID == "" { + id, err := newEnvID() + if err != nil { + return Env{}, err + } + env.ID = id + } + if _, exists := s.envs[env.ID]; exists { + return Env{}, fmt.Errorf("lab env %q already exists", env.ID) + } + s.envs[env.ID] = env + if err := s.persistLocked(); err != nil { + delete(s.envs, env.ID) + return Env{}, err + } + return env, nil +} + +// Update replaces an existing record in place. +func (s *Store) Update(env Env) error { + s.mu.Lock() + defer s.mu.Unlock() + + if _, exists := s.envs[env.ID]; !exists { + return fmt.Errorf("lab env %q not found", env.ID) + } + prev := s.envs[env.ID] + s.envs[env.ID] = env + if err := s.persistLocked(); err != nil { + s.envs[env.ID] = prev + return err + } + return nil +} + +// Delete removes a record and persists the snapshot. +func (s *Store) Delete(id string) error { + s.mu.Lock() + defer s.mu.Unlock() + + if _, exists := s.envs[id]; !exists { + return nil + } + prev := s.envs[id] + delete(s.envs, id) + if err := s.persistLocked(); err != nil { + s.envs[id] = prev + return err + } + return nil +} + +// Get returns a record by ID. +func (s *Store) Get(id string) (Env, bool) { + s.mu.Lock() + defer s.mu.Unlock() + env, ok := s.envs[id] + return env, ok +} + +// List returns a deterministic snapshot of all records. +func (s *Store) List() []Env { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]Env, 0, len(s.envs)) + for _, env := range s.envs { + out = append(out, env) + } + sort.Slice(out, func(i, j int) bool { + return out[i].CreatedAt.Before(out[j].CreatedAt) + }) + return out +} + +func (s *Store) load() error { + data, err := os.ReadFile(s.path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return fmt.Errorf("read lab env store: %w", err) + } + var envs []Env + if err := json.Unmarshal(data, &envs); err != nil { + return fmt.Errorf("parse lab env store: %w", err) + } + for _, env := range envs { + s.envs[env.ID] = env + } + return nil +} + +func (s *Store) persistLocked() error { + if err := os.MkdirAll(filepath.Dir(s.path), 0o700); err != nil { + return fmt.Errorf("ensure lab env dir: %w", err) + } + envs := make([]Env, 0, len(s.envs)) + for _, env := range s.envs { + envs = append(envs, env) + } + sort.Slice(envs, func(i, j int) bool { + return envs[i].CreatedAt.Before(envs[j].CreatedAt) + }) + data, err := json.MarshalIndent(envs, "", " ") + if err != nil { + return fmt.Errorf("encode lab env store: %w", err) + } + tmp := s.path + ".tmp" + if err := os.WriteFile(tmp, data, 0o600); err != nil { + return fmt.Errorf("write lab env store: %w", err) + } + if err := os.Rename(tmp, s.path); err != nil { + return fmt.Errorf("rename lab env store: %w", err) + } + return nil +} + +func newEnvID() (string, error) { + var buf [6]byte + if _, err := rand.Read(buf[:]); err != nil { + return "", fmt.Errorf("generate env id: %w", err) + } + return "env-" + hex.EncodeToString(buf[:]), nil +} diff --git a/internal/lab/types.go b/internal/lab/types.go new file mode 100644 index 0000000..ce282e9 --- /dev/null +++ b/internal/lab/types.go @@ -0,0 +1,96 @@ +// Package lab provides backend-agnostic ephemeral lab environment management. +// +// The package exposes a Driver interface with three planned implementations: +// +// - docker: reference implementation shipped in this package. Launches +// containerised JupyterLab environments on the local Docker daemon with a +// TTL reaper that archives the hot-tier workspace to the cold tier before +// tearing the container down. +// - slurm: (future) submits sbatch jobs and relies on Slurm for TTL, +// fair-share, GPU/GRES accounting, and epilog-driven archival. +// - proxmox: (future) clones a Proxmox VE template via the REST API and +// snapshots to the cold tier on teardown. +// +// The web and TUI surfaces consume a single HTTP API +// (`/api/v1/lab/envs`) that is driver-agnostic: the driver is chosen at launch +// time and recorded on the Env record. Storage tiers (hot/cold) are orthogonal +// to driver choice and are configured by the sysadmin via +// `concave lab storage`. +package lab + +import "time" + +// Status is the lifecycle state of an ephemeral lab environment. +type Status string + +const ( + StatusPending Status = "pending" + StatusRunning Status = "running" + StatusExpiring Status = "expiring" + StatusArchiving Status = "archiving" + StatusArchived Status = "archived" + StatusFailed Status = "failed" +) + +// EnvSpec is the user-facing request to launch an ephemeral environment. +type EnvSpec struct { + Owner string `json:"owner"` + Image string `json:"image"` + DisplayName string `json:"display_name,omitempty"` + GPUs int `json:"gpus,omitempty"` + CPURequest string `json:"cpu_request,omitempty"` + MemRequest string `json:"mem_request,omitempty"` + TTL time.Duration `json:"ttl"` + HotTierPath string `json:"hot_tier_path,omitempty"` + ColdTierPath string `json:"cold_tier_path,omitempty"` + Driver string `json:"driver,omitempty"` +} + +// Env is the persisted record for a running (or archived) environment. +type Env struct { + ID string `json:"id"` + Driver string `json:"driver"` + Owner string `json:"owner"` + Image string `json:"image"` + DisplayName string `json:"display_name,omitempty"` + Status Status `json:"status"` + ContainerID string `json:"container_id,omitempty"` + JupyterURL string `json:"jupyter_url,omitempty"` + Token string `json:"token,omitempty"` + GPUs int `json:"gpus"` + CPURequest string `json:"cpu_request,omitempty"` + MemRequest string `json:"mem_request,omitempty"` + HotTierPath string `json:"hot_tier_path"` + ColdTierPath string `json:"cold_tier_path"` + ArchiveRef string `json:"archive_ref,omitempty"` + CreatedAt time.Time `json:"created_at"` + ExpiresAt time.Time `json:"expires_at"` + ArchivedAt *time.Time `json:"archived_at,omitempty"` + LastError string `json:"last_error,omitempty"` +} + +// Remaining reports the TTL budget left on the env, clamped to zero. +func (e Env) Remaining(now time.Time) time.Duration { + if e.ExpiresAt.IsZero() { + return 0 + } + remaining := e.ExpiresAt.Sub(now) + if remaining < 0 { + return 0 + } + return remaining +} + +// Expired reports whether the env's TTL has passed. +func (e Env) Expired(now time.Time) bool { + return !e.ExpiresAt.IsZero() && !now.Before(e.ExpiresAt) +} + +// Active reports whether the env is still in a runnable/archivable state. +func (e Env) Active() bool { + switch e.Status { + case StatusArchived, StatusFailed: + return false + } + return true +}