From 789ba3a43d2f15bac2cc2464ff2cd1a4b77a5fd2 Mon Sep 17 00:00:00 2001 From: Jeefos Date: Fri, 15 May 2026 14:43:19 -0400 Subject: [PATCH] Daemon mod file injection and bound mounts --- cmd/kleffd/main.go | 19 +- cmd/testprovision/main.go | 2 +- internal/adapters/out/queue/redis.go | 2 + .../adapters/out/runtime/docker/docker.go | 169 ++++++++++++++++-- .../out/runtime/kubernetes/kubernetes.go | 13 ++ internal/app/config/config.go | 3 + internal/application/ports/runtime_adapter.go | 13 ++ internal/application/ports/workload.go | 19 ++ internal/logs/tailer.go | 9 +- internal/workers/jobs/job.go | 2 + internal/workers/mod_worker.go | 74 ++++++++ 11 files changed, 307 insertions(+), 18 deletions(-) create mode 100644 internal/workers/mod_worker.go diff --git a/cmd/kleffd/main.go b/cmd/kleffd/main.go index 8c69151..8dcccf6 100644 --- a/cmd/kleffd/main.go +++ b/cmd/kleffd/main.go @@ -69,6 +69,19 @@ func main() { // --- Repository --- repo := memrepo.NewServerRepository() + // Reseed the in-memory repository from any workloads that were running + // before this daemon process started (e.g. after a daemon restart). + if existing, err := runtime.ListRunning(context.Background()); err != nil { + daemonLog.Warn("Failed to recover running workloads on startup", "error", err) + } else { + for _, srv := range existing { + _ = repo.Save(context.Background(), srv) + } + if len(existing) > 0 { + daemonLog.Info("Recovered running workloads", "count", len(existing)) + } + } + // --- Platform registration + status reporting --- platformClient := platformadapter.NewClient(cfg.PlatformURL, cfg.SharedSecret, cfg.NodeID, daemonLog) if err := platformClient.RegisterNode(context.Background()); err != nil { @@ -84,6 +97,10 @@ func main() { dispatcher.Register(jobs.JobTypeServerDelete, workers.NewDeleteWorker(runtime, repo, daemonLog, platformClient).Handle) dispatcher.Register(jobs.JobTypeServerRestart, workers.NewRestartWorker(runtime, repo, daemonLog, platformClient).Handle) + modWorker := workers.NewModWorker(runtime, daemonLog) + dispatcher.Register(jobs.JobTypeModInstall, modWorker.HandleInstall) + dispatcher.Register(jobs.JobTypeModUninstall, modWorker.HandleUninstall) + daemonLog.Info("Daemon started", "node_id", cfg.NodeID, "grpc_port", cfg.GRPCPort) ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) @@ -124,7 +141,7 @@ func detectRuntime(cfg *config.Config, logger ports.Logger) (ports.RuntimeAdapte } // No Kubernetes — check if Docker is actually reachable before using it. - dockerAdapter, err := dockeradapter.New(cfg.NodeID) + dockerAdapter, err := dockeradapter.New(cfg.NodeID, cfg.StoragePath) if err != nil { return nil, fmt.Errorf("no runtime available: kubernetes not detected, docker client failed: %w", err) } diff --git a/cmd/testprovision/main.go b/cmd/testprovision/main.go index 15eb7df..9cfb951 100644 --- a/cmd/testprovision/main.go +++ b/cmd/testprovision/main.go @@ -79,7 +79,7 @@ func main() { } func runDocker(cleanup bool, serverID string, spec ports.WorkloadSpec) { - adapter, err := dockeradapter.New("test-node") + adapter, err := dockeradapter.New("test-node", "/var/lib/kleffd/servers") if err != nil { log.Fatalf("failed to create docker adapter: %v", err) } diff --git a/internal/adapters/out/queue/redis.go b/internal/adapters/out/queue/redis.go index 2ec5445..8db97a3 100644 --- a/internal/adapters/out/queue/redis.go +++ b/internal/adapters/out/queue/redis.go @@ -141,6 +141,8 @@ func (q *RedisQueue) Retry(jobID string) error { var job jobs.Job json.Unmarshal([]byte(raw), &job) + job.Attempts++ + if job.Attempts >= job.MaxAttempts { job.Status = jobs.JobStatusFailed newRaw, _ := json.Marshal(job) diff --git a/internal/adapters/out/runtime/docker/docker.go b/internal/adapters/out/runtime/docker/docker.go index 85bd323..17a027a 100644 --- a/internal/adapters/out/runtime/docker/docker.go +++ b/internal/adapters/out/runtime/docker/docker.go @@ -6,8 +6,12 @@ import ( "errors" "fmt" "io" + "net/http" + "os" + "path/filepath" "sort" "strings" + "time" "github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/filters" @@ -24,18 +28,43 @@ import ( // All three strategies (agones, statefulset, deployment) map to the same // Docker container lifecycle — the strategy hint is ignored here. type Adapter struct { - client *client.Client - nodeID string + client *client.Client + nodeID string + storageLocalPath string // path inside this container where server data is mounted + storageHostPath string // corresponding host path passed to Docker for bind mounts } var errContainerNotFound = errors.New("container not found") -func New(nodeID string) (*Adapter, error) { +// New creates a Docker adapter. storageLocalPath is the directory inside the +// daemon container where game server data lives (e.g. /var/lib/kleffd/servers). +// The adapter auto-detects the matching host path by inspecting its own container +// mounts so that bind mount sources are always valid host filesystem paths. +func New(nodeID, storageLocalPath string) (*Adapter, error) { c, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) if err != nil { return nil, fmt.Errorf("failed to create docker client: %w", err) } - return &Adapter{client: c, nodeID: nodeID}, nil + + // Default: host path == local path (works when daemon runs directly on host). + storageHostPath := storageLocalPath + if hostname, err := os.Hostname(); err == nil { + if info, err := c.ContainerInspect(context.Background(), hostname); err == nil { + for _, m := range info.Mounts { + if m.Destination == storageLocalPath { + storageHostPath = m.Source + break + } + } + } + } + + return &Adapter{ + client: c, + nodeID: nodeID, + storageLocalPath: storageLocalPath, + storageHostPath: storageHostPath, + }, nil } // Ping checks if the Docker daemon is reachable. @@ -268,7 +297,7 @@ func (a *Adapter) Stop(ctx context.Context, projectID, workloadID string) error return nil } -// Remove stops and removes the container. +// Remove stops and removes the container. Data directory is preserved on disk. func (a *Adapter) Remove(ctx context.Context, projectID, workloadID string) error { containerID, err := a.findContainer(ctx, projectID, workloadID) if err != nil { @@ -385,22 +414,66 @@ func (a *Adapter) Endpoint(ctx context.Context, projectID, workloadID string) (s } // Logs streams the container's stdout/stderr. +// When follow is true, Since is set to the container's last start time so that +// only the logs from the current run are streamed — avoiding replaying the full +// history across restarts before reaching live output. func (a *Adapter) Logs(ctx context.Context, projectID, workloadID string, follow bool) (io.ReadCloser, error) { containerID, err := a.findContainer(ctx, projectID, workloadID) if err != nil { return nil, err } - rc, err := a.client.ContainerLogs(ctx, containerID, container.LogsOptions{ + + opts := container.LogsOptions{ ShowStdout: true, ShowStderr: true, Follow: follow, - }) + } + if follow { + if info, err := a.client.ContainerInspect(ctx, containerID); err == nil && info.State != nil { + if t, err := time.Parse(time.RFC3339Nano, info.State.StartedAt); err == nil && !t.IsZero() { + opts.Since = t.UTC().Format(time.RFC3339Nano) + } + } + } + + rc, err := a.client.ContainerLogs(ctx, containerID, opts) if err != nil { return nil, fmt.Errorf("failed to get logs: %w", err) } return rc, nil } +// ListRunning returns a ServerRecord for each container currently managed by +// this daemon that is in the "running" state. Used to reseed the in-memory +// repository after a daemon restart. +func (a *Adapter) ListRunning(ctx context.Context) ([]*ports.ServerRecord, error) { + containers, err := a.client.ContainerList(ctx, container.ListOptions{ + Filters: filters.NewArgs( + filters.Arg("label", labels.ManagedBy+"="+labels.ManagedByValue), + filters.Arg("status", "running"), + ), + }) + if err != nil { + return nil, fmt.Errorf("list managed containers: %w", err) + } + + records := make([]*ports.ServerRecord, 0, len(containers)) + for _, c := range containers { + wl := labels.FromMap(c.Labels) + if wl.ServerID == "" { + continue + } + records = append(records, &ports.ServerRecord{ + ID: wl.ServerID, + Status: "Running", + NodeID: wl.NodeID, + RuntimeRef: c.ID, + ProjectID: wl.ProjectID, + }) + } + return records, nil +} + // --- Helpers --- func (a *Adapter) createContainer(ctx context.Context, spec ports.WorkloadSpec, scope *ports.ProjectScope) (string, error) { @@ -443,9 +516,17 @@ func (a *Adapter) createContainer(ctx context.Context, spec ports.WorkloadSpec, var mounts []mount.Mount if spec.RuntimeHints.PersistentStorage && spec.RuntimeHints.StoragePath != "" { + localDir := filepath.Join(a.storageLocalPath, projectDataDir(spec.ProjectID, spec.ServerID)) + if err := os.MkdirAll(localDir, 0777); err != nil { + return "", fmt.Errorf("create server storage directory: %w", err) + } + // MkdirAll respects the process umask, so chmod explicitly to ensure + // any user inside the game server container can write to /data. + _ = os.Chmod(localDir, 0777) + hostDir := filepath.Join(a.storageHostPath, projectDataDir(spec.ProjectID, spec.ServerID)) mounts = append(mounts, mount.Mount{ - Type: mount.TypeVolume, - Source: projectVolumeName(spec.ProjectID, spec.ServerID), + Type: mount.TypeBind, + Source: hostDir, Target: spec.RuntimeHints.StoragePath, }) } @@ -549,10 +630,72 @@ func projectNetworkName(projectID string) string { return "kleff_proj_" + shortID(projectID) } -// projectVolumeName namespaces persistent volumes by project so a stray -// workload id collision across projects cannot cross-mount data. -func projectVolumeName(projectID, workloadID string) string { - return fmt.Sprintf("kleff_proj_%s_%s_data", shortID(projectID), workloadID) +// projectDataDir returns a unique directory name for a workload's persistent data. +func projectDataDir(projectID, workloadID string) string { + return fmt.Sprintf("kleff_proj_%s_%s", shortID(projectID), workloadID) +} + +// contentTypeDirs maps a mod content type to its subdirectory inside the storage volume. +var contentTypeDirs = map[string]string{ + "mod": "mods", + "plugin": "plugins", + "datapack": "world/datapacks", + "resourcepack": "resourcepacks", +} + +// InjectFile downloads a file from downloadURL directly into the workload's +// bind-mounted storage directory on the host filesystem. +func (a *Adapter) InjectFile(ctx context.Context, projectID, workloadID, storagePath, contentType, downloadURL, fileName string) error { + subDir, ok := contentTypeDirs[contentType] + if !ok { + return fmt.Errorf("%w: unsupported content type %q", ports.ErrPermanent, contentType) + } + + destDir := filepath.Join(a.storageLocalPath, projectDataDir(projectID, workloadID), subDir) + if err := os.MkdirAll(destDir, 0755); err != nil { + return fmt.Errorf("create mod directory: %w", err) + } + + return downloadFile(ctx, downloadURL, filepath.Join(destDir, fileName)) +} + +// RemoveFile deletes a previously injected file from the workload's storage directory. +func (a *Adapter) RemoveFile(ctx context.Context, projectID, workloadID, storagePath, contentType, fileName string) error { + subDir, ok := contentTypeDirs[contentType] + if !ok { + return fmt.Errorf("%w: unsupported content type %q", ports.ErrPermanent, contentType) + } + + filePath := filepath.Join(a.storageLocalPath, projectDataDir(projectID, workloadID), subDir, fileName) + if err := os.Remove(filePath); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("remove file: %w", err) + } + return nil +} + +func downloadFile(ctx context.Context, url, destPath string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return fmt.Errorf("build request: %w", err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return fmt.Errorf("download: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("download: HTTP %d", resp.StatusCode) + } + f, err := os.Create(destPath) + if err != nil { + return fmt.Errorf("create file: %w", err) + } + defer f.Close() + if _, err := io.Copy(f, resp.Body); err != nil { + os.Remove(destPath) + return fmt.Errorf("write file: %w", err) + } + return nil } func shortID(s string) string { diff --git a/internal/adapters/out/runtime/kubernetes/kubernetes.go b/internal/adapters/out/runtime/kubernetes/kubernetes.go index 5e0eb21..ae28b2e 100644 --- a/internal/adapters/out/runtime/kubernetes/kubernetes.go +++ b/internal/adapters/out/runtime/kubernetes/kubernetes.go @@ -569,3 +569,16 @@ func buildService(spec ports.WorkloadSpec, selectorLabels map[string]string) *co }, } } + + +func (a *Adapter) InjectFile(_ context.Context, _, _, _, _, _, _ string) error { + return fmt.Errorf("mod file injection is not yet supported on Kubernetes") +} + +func (a *Adapter) RemoveFile(_ context.Context, _, _, _, _, _ string) error { + return fmt.Errorf("mod file removal is not yet supported on Kubernetes") +} + +func (a *Adapter) ListRunning(_ context.Context) ([]*ports.ServerRecord, error) { + return nil, nil +} diff --git a/internal/app/config/config.go b/internal/app/config/config.go index 12635b6..312131e 100644 --- a/internal/app/config/config.go +++ b/internal/app/config/config.go @@ -34,6 +34,7 @@ type Config struct { RedisTLS bool `mapstructure:"redis.tls"` PlatformURL string `mapstructure:"platform.url"` SharedSecret string `mapstructure:"shared_secret"` + StoragePath string `mapstructure:"storage.path"` } func (c *Config) Validate() error { @@ -81,6 +82,7 @@ func Load() (*Config, error) { v.SetDefault("redis.tls", false) v.SetDefault("platform.url", "") v.SetDefault("shared_secret", "") + v.SetDefault("storage.path", "/var/lib/kleffd/servers") v.SetEnvPrefix("kleff") v.SetEnvKeyReplacer(strings.NewReplacer(".", "_")) @@ -126,6 +128,7 @@ func Load() (*Config, error) { cfg.RedisTLS = v.GetBool("redis.tls") cfg.PlatformURL = v.GetString("platform.url") cfg.SharedSecret = v.GetString("shared_secret") + cfg.StoragePath = v.GetString("storage.path") if err := cfg.Validate(); err != nil { return nil, fmt.Errorf("configuration validation failed: %w", err) diff --git a/internal/application/ports/runtime_adapter.go b/internal/application/ports/runtime_adapter.go index c980478..77b835e 100644 --- a/internal/application/ports/runtime_adapter.go +++ b/internal/application/ports/runtime_adapter.go @@ -43,4 +43,17 @@ type RuntimeAdapter interface { Endpoint(ctx context.Context, projectID, workloadID string) (string, error) // Logs streams the workload's stdout/stderr. Logs(ctx context.Context, projectID, workloadID string, follow bool) (io.ReadCloser, error) + + // InjectFile downloads a file from downloadURL and writes it into the + // workload's persistent volume under the appropriate content-type subdirectory. + // storagePath is the volume's mount point inside the container (e.g. "/data"). + InjectFile(ctx context.Context, projectID, workloadID, storagePath, contentType, downloadURL, fileName string) error + + // RemoveFile deletes a previously injected file from the workload's volume. + RemoveFile(ctx context.Context, projectID, workloadID, storagePath, contentType, fileName string) error + + // ListRunning returns server records for all workloads currently managed by + // this daemon that are in a running state. Used to reseed the in-memory + // repository after a daemon restart. + ListRunning(ctx context.Context) ([]*ServerRecord, error) } diff --git a/internal/application/ports/workload.go b/internal/application/ports/workload.go index c47a5ea..6e1f2c2 100644 --- a/internal/application/ports/workload.go +++ b/internal/application/ports/workload.go @@ -62,6 +62,25 @@ type RuntimeHints struct { StorageGB int `json:"storage_gb,omitempty"` } +// ModInstallSpec is the payload for server.install_mod jobs. +type ModInstallSpec struct { + ServerID string `json:"server_id"` + ProjectID string `json:"project_id"` + DownloadURL string `json:"download_url"` + FileName string `json:"file_name"` + ContentType string `json:"content_type"` // "mod", "plugin", "datapack", "resourcepack" + StoragePath string `json:"storage_path"` // base mount path, e.g. "/data" +} + +// ModUninstallSpec is the payload for server.uninstall_mod jobs. +type ModUninstallSpec struct { + ServerID string `json:"server_id"` + ProjectID string `json:"project_id"` + FileName string `json:"file_name"` + ContentType string `json:"content_type"` + StoragePath string `json:"storage_path"` +} + // WorkloadHealth is the per-workload status reported in heartbeats. type WorkloadHealth struct { WorkloadID string `json:"workload_id"` diff --git a/internal/logs/tailer.go b/internal/logs/tailer.go index f8dae68..79f2a2e 100644 --- a/internal/logs/tailer.go +++ b/internal/logs/tailer.go @@ -46,7 +46,7 @@ func NewTailer( // Run blocks until ctx is cancelled, reconciling the set of active tailers // every 10 seconds to match the set of running workloads. func (t *Tailer) Run(ctx context.Context) { - ticker := time.NewTicker(10 * time.Second) + ticker := time.NewTicker(3 * time.Second) defer ticker.Stop() // Reconcile immediately on start. t.reconcile(ctx) @@ -140,8 +140,11 @@ func (t *Tailer) tail(ctx context.Context, workloadID, projectID string) { lines := make(chan ports.LogEntry, 256) - go scanStream(ctx, stdoutR, "stdout", lines) - go scanStream(ctx, stderrR, "stderr", lines) + var scanWg sync.WaitGroup + scanWg.Add(2) + go func() { defer scanWg.Done(); scanStream(ctx, stdoutR, "stdout", lines) }() + go func() { defer scanWg.Done(); scanStream(ctx, stderrR, "stderr", lines) }() + go func() { scanWg.Wait(); close(lines) }() for { select { diff --git a/internal/workers/jobs/job.go b/internal/workers/jobs/job.go index e5b28fa..20c64bb 100644 --- a/internal/workers/jobs/job.go +++ b/internal/workers/jobs/job.go @@ -14,6 +14,8 @@ const ( JobTypeServerStop JobType = "server.stop" JobTypeServerDelete JobType = "server.delete" JobTypeServerRestart JobType = "server.restart" + JobTypeModInstall JobType = "server.install_mod" + JobTypeModUninstall JobType = "server.uninstall_mod" ) const ( diff --git a/internal/workers/mod_worker.go b/internal/workers/mod_worker.go new file mode 100644 index 0000000..c7ac115 --- /dev/null +++ b/internal/workers/mod_worker.go @@ -0,0 +1,74 @@ +package workers + +import ( + "context" + "fmt" + + "github.com/kleffio/kleff-daemon/internal/application/ports" + "github.com/kleffio/kleff-daemon/internal/workers/jobs" +) + +type ModWorker struct { + runtime ports.RuntimeAdapter + logger ports.Logger +} + +func NewModWorker(runtime ports.RuntimeAdapter, logger ports.Logger) *ModWorker { + return &ModWorker{runtime: runtime, logger: logger} +} + +func (w *ModWorker) HandleInstall(ctx context.Context, job *jobs.Job) error { + log := w.logger.With(ports.LogKeyJobID, job.JobID, ports.LogKeyWorkerType, "server.install_mod") + + var spec ports.ModInstallSpec + if err := job.UnmarshalPayload(&spec); err != nil { + return fmt.Errorf("%w: invalid payload: %s", ports.ErrPermanent, err) + } + if spec.ServerID == "" || spec.ProjectID == "" { + return fmt.Errorf("%w: server_id and project_id are required", ports.ErrPermanent) + } + if spec.DownloadURL == "" || spec.FileName == "" || spec.ContentType == "" { + return fmt.Errorf("%w: download_url, file_name, and content_type are required", ports.ErrPermanent) + } + if spec.StoragePath == "" { + spec.StoragePath = "/data" + } + + log.Info("Installing mod", "server_id", spec.ServerID, "file", spec.FileName, "type", spec.ContentType) + + if err := w.runtime.InjectFile(ctx, spec.ProjectID, spec.ServerID, spec.StoragePath, spec.ContentType, spec.DownloadURL, spec.FileName); err != nil { + log.Error("Failed to inject mod file", err) + return fmt.Errorf("inject file: %w", err) + } + + log.Info("Mod installed successfully", "server_id", spec.ServerID, "file", spec.FileName) + return nil +} + +func (w *ModWorker) HandleUninstall(ctx context.Context, job *jobs.Job) error { + log := w.logger.With(ports.LogKeyJobID, job.JobID, ports.LogKeyWorkerType, "server.uninstall_mod") + + var spec ports.ModUninstallSpec + if err := job.UnmarshalPayload(&spec); err != nil { + return fmt.Errorf("%w: invalid payload: %s", ports.ErrPermanent, err) + } + if spec.ServerID == "" || spec.ProjectID == "" { + return fmt.Errorf("%w: server_id and project_id are required", ports.ErrPermanent) + } + if spec.FileName == "" || spec.ContentType == "" { + return fmt.Errorf("%w: file_name and content_type are required", ports.ErrPermanent) + } + if spec.StoragePath == "" { + spec.StoragePath = "/data" + } + + log.Info("Uninstalling mod", "server_id", spec.ServerID, "file", spec.FileName, "type", spec.ContentType) + + if err := w.runtime.RemoveFile(ctx, spec.ProjectID, spec.ServerID, spec.StoragePath, spec.ContentType, spec.FileName); err != nil { + log.Error("Failed to remove mod file", err) + return fmt.Errorf("remove file: %w", err) + } + + log.Info("Mod uninstalled successfully", "server_id", spec.ServerID, "file", spec.FileName) + return nil +}