diff --git a/apps/druid/adapters/cli/worker_pull.go b/apps/druid/adapters/cli/worker_pull.go index 4bdab15..4040629 100644 --- a/apps/druid/adapters/cli/worker_pull.go +++ b/apps/druid/adapters/cli/worker_pull.go @@ -51,6 +51,7 @@ func init() { WorkerPullCommand.Flags().StringVar(&workerPullAction.MountPath, "root", "/scroll", "Mounted runtime root path") WorkerPullCommand.Flags().StringVar(&workerPullAction.CallbackURL, "callback-url", "", "Daemon worker callback URL") WorkerPullCommand.Flags().StringVar(&workerPullAction.TokenFile, "callback-token-file", "", "Projected ServiceAccount token file for callbacks") + WorkerPullCommand.Flags().BoolVar(&workerPullAction.PreserveReleaseManifest, "preserve-release-manifest", false, "Restore the release manifest.json carried by a backup") WorkerPullCommand.Flags().StringVar(&workerPullMode, "mode", string(ports.RuntimeWorkerModeCreate), "Pull mode: create, update, or restore") WorkerPullCommand.MarkFlagRequired("artifact") WorkerPullCommand.MarkFlagRequired("runtime-id") @@ -164,27 +165,101 @@ func pullWorkerUpdate(root string, artifact string, oci ports.OciRegistryInterfa } func pullWorkerRestore(root string, artifact string, oci ports.OciRegistryInterface) error { - tmp, err := os.MkdirTemp("", "druid-worker-restore-*") + if err := os.MkdirAll(root, 0755); err != nil { + return err + } + stage, err := os.MkdirTemp(root, ".druid-worker-restore-stage-*") if err != nil { return err } - defer os.RemoveAll(tmp) - if err := coreservices.MaterializeScrollArtifact(artifact, tmp, oci, true); err != nil { + defer os.RemoveAll(stage) + puller, ok := oci.(interface { + PullSelectiveWithOptions(string, string, bool, *domain.SnapshotProgress, registry.TransferOptions) error + }) + if !ok { + return fmt.Errorf("OCI registry does not support preserve-release-manifest") + } + if err := puller.PullSelectiveWithOptions(stage, artifact, true, nil, registry.TransferOptions{PreserveReleaseManifest: true}); err != nil { return err } - if err := os.MkdirAll(root, 0755); err != nil { + return replaceRestoredRoot(root, stage) +} + +const restoreRootUnsafePrefix = "restore root may be partial:" + +func replaceRestoredRoot(root string, stage string) error { + return replaceRestoredRootWithRename(root, stage, os.Rename) +} + +// replaceRestoredRootWithRename swaps staged artifact entries into a runtime +// root without copying over a live tree. If an entry move fails, it restores +// the original entries before returning. An error with restoreRootUnsafePrefix +// means that rollback itself failed and callers must keep the runtime stopped. +func replaceRestoredRootWithRename(root string, stage string, rename func(string, string) error) error { + rollback, err := os.MkdirTemp(root, ".druid-worker-restore-rollback-*") + if err != nil { return err } + defer os.RemoveAll(rollback) + + stageName := filepath.Base(stage) + rollbackName := filepath.Base(rollback) entries, err := os.ReadDir(root) if err != nil { return err } + original := make([]string, 0, len(entries)) for _, entry := range entries { - if err := os.RemoveAll(filepath.Join(root, entry.Name())); err != nil { - return err + if entry.Name() != stageName && entry.Name() != rollbackName { + original = append(original, entry.Name()) } } - return copyPath(tmp, root) + stagedEntries, err := os.ReadDir(stage) + if err != nil { + return err + } + + restoreOriginal := func(installed []string) error { + for _, name := range installed { + if err := os.RemoveAll(filepath.Join(root, name)); err != nil { + return err + } + } + for _, name := range original { + from := filepath.Join(rollback, name) + if _, err := os.Lstat(from); os.IsNotExist(err) { + continue + } else if err != nil { + return err + } + if err := rename(from, filepath.Join(root, name)); err != nil { + return err + } + } + return nil + } + + for _, entry := range original { + if err := rename(filepath.Join(root, entry), filepath.Join(rollback, entry)); err != nil { + if rollbackErr := restoreOriginal(nil); rollbackErr != nil { + return fmt.Errorf("%s failed to restore original runtime after moving %s: %w", restoreRootUnsafePrefix, entry, rollbackErr) + } + return fmt.Errorf("restore transaction rolled back while moving original entry %s: %w", entry, err) + } + } + + installed := make([]string, 0, len(stagedEntries)) + for _, entry := range stagedEntries { + name := entry.Name() + if err := rename(filepath.Join(stage, name), filepath.Join(root, name)); err != nil { + if rollbackErr := restoreOriginal(installed); rollbackErr != nil { + return fmt.Errorf("%s failed to restore original runtime after staging %s: %w", restoreRootUnsafePrefix, name, rollbackErr) + } + return fmt.Errorf("restore transaction rolled back while staging %s: %w", name, err) + } + installed = append(installed, name) + } + return nil } func collectSkipUpdatePaths(out map[string]bool, parent string, chunks []*domain.Chunks) { diff --git a/apps/druid/adapters/cli/worker_push.go b/apps/druid/adapters/cli/worker_push.go index 5e0b1cd..6e7b754 100644 --- a/apps/druid/adapters/cli/worker_push.go +++ b/apps/druid/adapters/cli/worker_push.go @@ -11,6 +11,7 @@ import ( var workerPushArtifact string var workerPushRoot string +var workerPushPreserveReleaseManifest bool var WorkerPushCommand = &cobra.Command{ Use: "push", @@ -29,7 +30,7 @@ var WorkerPushCommand = &cobra.Command{ } repo, tag := utils.SplitArtifact(workerPushArtifact) oci := registry.NewOciClient(loadWorkerRegistryStore()) - _, err = oci.Push(workerPushRoot, repo, tag, nil, false, &scroll.File) + _, err = oci.PushWithOptions(workerPushRoot, repo, tag, nil, false, &scroll.File, registry.TransferOptions{PreserveReleaseManifest: workerPushPreserveReleaseManifest}) return err }, } @@ -38,5 +39,6 @@ func init() { WorkerCommand.AddCommand(WorkerPushCommand) WorkerPushCommand.Flags().StringVar(&workerPushArtifact, "artifact", "", "OCI artifact to push") WorkerPushCommand.Flags().StringVar(&workerPushRoot, "root", "/scroll", "Mounted runtime root path") + WorkerPushCommand.Flags().BoolVar(&workerPushPreserveReleaseManifest, "preserve-release-manifest", false, "Include the installed release manifest.json in the backup payload") WorkerPushCommand.MarkFlagRequired("artifact") } diff --git a/apps/druid/adapters/cli/worker_test.go b/apps/druid/adapters/cli/worker_test.go index 80a3cb2..ac18897 100644 --- a/apps/druid/adapters/cli/worker_test.go +++ b/apps/druid/adapters/cli/worker_test.go @@ -1,14 +1,17 @@ package cli import ( + "errors" "net/http" "net/http/httptest" "os" "path/filepath" + "strings" "testing" "github.com/highcard-dev/daemon/internal/core/domain" "github.com/highcard-dev/daemon/internal/core/ports" + "github.com/highcard-dev/daemon/internal/core/services/registry" v1 "github.com/opencontainers/image-spec/specs-go/v1" "github.com/spf13/cobra" "oras.land/oras-go/v2/registry/remote" @@ -33,6 +36,18 @@ func TestWorkerPullCommandRequiresRuntimeID(t *testing.T) { } } +func TestWorkerPullCommandSupportsReleaseManifestPreservation(t *testing.T) { + if flag := WorkerPullCommand.Flags().Lookup("preserve-release-manifest"); flag == nil { + t.Fatal("worker pull should expose --preserve-release-manifest") + } +} + +func TestWorkerPushCommandSupportsReleaseManifestPreservation(t *testing.T) { + if flag := WorkerPushCommand.Flags().Lookup("preserve-release-manifest"); flag == nil { + t.Fatal("worker push should expose --preserve-release-manifest") + } +} + func TestReportWorkerResultUsesTokenOnlyWhenProvided(t *testing.T) { tests := []struct { name string @@ -108,6 +123,67 @@ func TestWorkerRestoreStagesBeforeReplacingRoot(t *testing.T) { } } +func TestReplaceRestoredRootRollsBackAfterStagingBegins(t *testing.T) { + root := t.TempDir() + mustWrite(t, filepath.Join(root, "scroll.yaml"), "name: original\n") + mustWrite(t, filepath.Join(root, "data", "world.txt"), "original") + stage, err := os.MkdirTemp(root, ".druid-worker-restore-stage-*") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(stage) + mustWrite(t, filepath.Join(stage, "a-new.txt"), "new") + mustWrite(t, filepath.Join(stage, "b-new.txt"), "new") + + stagedMoves := 0 + err = replaceRestoredRootWithRename(root, stage, func(from string, to string) error { + if strings.HasPrefix(from, stage+string(os.PathSeparator)) { + stagedMoves++ + if stagedMoves == 2 { + return errors.New("injected staging failure") + } + } + return os.Rename(from, to) + }) + if err == nil || !strings.Contains(err.Error(), "restore transaction rolled back") { + t.Fatalf("restore error = %v, want successful rollback error", err) + } + assertFile(t, filepath.Join(root, "scroll.yaml"), "name: original\n") + assertFile(t, filepath.Join(root, "data", "world.txt"), "original") + if _, err := os.Stat(filepath.Join(root, "a-new.txt")); !os.IsNotExist(err) { + t.Fatalf("staged file should be removed during rollback, stat err = %v", err) + } +} + +func TestReplaceRestoredRootMarksFailedRollbackUnsafe(t *testing.T) { + root := t.TempDir() + mustWrite(t, filepath.Join(root, "original.txt"), "original") + stage, err := os.MkdirTemp(root, ".druid-worker-restore-stage-*") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(stage) + mustWrite(t, filepath.Join(stage, "a-new.txt"), "new") + mustWrite(t, filepath.Join(stage, "b-new.txt"), "new") + + stagedMoves := 0 + err = replaceRestoredRootWithRename(root, stage, func(from string, to string) error { + if strings.HasPrefix(from, stage+string(os.PathSeparator)) { + stagedMoves++ + if stagedMoves == 2 { + return errors.New("injected staging failure") + } + } + if strings.Contains(filepath.Base(filepath.Dir(from)), "restore-rollback-") { + return errors.New("injected rollback failure") + } + return os.Rename(from, to) + }) + if err == nil || !strings.HasPrefix(err.Error(), restoreRootUnsafePrefix) { + t.Fatalf("restore error = %v, want unsafe rollback marker", err) + } +} + func TestWorkerCollectSkipUpdatePaths(t *testing.T) { root := filepath.Join(t.TempDir(), "root") mustWrite(t, filepath.Join(root, "scroll.yaml"), `name: skip-test @@ -195,6 +271,16 @@ func (f fakeRestoreOCI) PullSelective(dir string, artifact string, includeData b return nil } +func (f fakeRestoreOCI) PullSelectiveWithOptions(dir string, artifact string, includeData bool, progress *domain.SnapshotProgress, options registry.TransferOptions) error { + if !options.PreserveReleaseManifest { + f.t.Fatal("restore must preserve the release manifest") + } + mustWrite(f.t, filepath.Join(dir, "scroll.yaml"), "name: restored\n") + mustWrite(f.t, filepath.Join(dir, "manifest.json"), `{"digest":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}`) + mustWrite(f.t, filepath.Join(dir, "data", "logs", "latest.log"), "restored") + return nil +} + func (fakeRestoreOCI) FetchFile(string, string) ([]byte, error) { return nil, os.ErrNotExist } diff --git a/apps/druid/core/services/runtime_access.go b/apps/druid/core/services/runtime_access.go index dd2e0b1..f3c6786 100644 --- a/apps/druid/core/services/runtime_access.go +++ b/apps/druid/core/services/runtime_access.go @@ -2,6 +2,8 @@ package services import ( "context" + "fmt" + "strings" "github.com/highcard-dev/daemon/internal/core/domain" "github.com/highcard-dev/daemon/internal/core/ports" @@ -56,18 +58,36 @@ func (s *RuntimeSupervisor) ApplyRouting(id string, assignments []domain.Runtime } func (s *RuntimeSupervisor) Backup(id string, artifact string, registryCredentials []domain.RegistryCredential) (*domain.RuntimeScroll, error) { + unlock := s.lockRuntimeOperation(id) + defer unlock() session, err := s.sessionFor(id) if err != nil { return nil, err } + wasRunning := sessionWasRunning(session) + if err := session.StopRuntimeForMaintenance(); err != nil { + session.markError(err) + return nil, err + } if err := session.Backup(context.Background(), artifact, registryCredentials); err != nil { + if wasRunning { + if _, restartErr := s.startScroll(id); restartErr != nil { + return nil, fmt.Errorf("backup failed: %w; failed to restart prior runtime: %v", err, restartErr) + } + return nil, err + } session.markError(err) return nil, err } + if wasRunning { + return s.startScroll(id) + } return s.store.GetScroll(id) } func (s *RuntimeSupervisor) Restore(id string, artifact string, restart bool, registryCredentials []domain.RegistryCredential) (*domain.RuntimeScroll, error) { + unlock := s.lockRuntimeOperation(id) + defer unlock() session, err := s.sessionFor(id) if err != nil { return nil, err @@ -75,12 +95,19 @@ func (s *RuntimeSupervisor) Restore(id string, artifact string, restart bool, re session.mu.Lock() root := session.runtimeScroll.Root session.mu.Unlock() - if err := session.StopRuntime(); err != nil { + wasRunning := sessionWasRunning(session) + if err := session.StopRuntimeForMaintenance(); err != nil { session.markError(err) return nil, err } materialized, err := s.runPullWorker(context.Background(), s.runtimeBackend, ports.RuntimeWorkerModeRestore, id, artifact, root, registryCredentials, "") if err != nil { + if wasRunning && restoreFailureCanRestart(err) { + if _, restartErr := s.startScroll(id); restartErr != nil { + return nil, fmt.Errorf("restore failed: %w; failed to restart prior runtime: %v", err, restartErr) + } + return nil, err + } session.markError(err) return nil, err } @@ -89,11 +116,21 @@ func (s *RuntimeSupervisor) Restore(id string, artifact string, restart bool, re return nil, err } if restart { - return s.StartScroll(id) + return s.startScroll(id) } return s.store.GetScroll(id) } +func sessionWasRunning(session *RuntimeSession) bool { + session.mu.Lock() + defer session.mu.Unlock() + return session.runtimeScroll.Status == domain.RuntimeScrollStatusRunning +} + +func restoreFailureCanRestart(err error) bool { + return !strings.Contains(err.Error(), "restore root may be partial:") +} + func (s *RuntimeSupervisor) ScrollFile(id string) (*domain.File, error) { session, err := s.sessionFor(id) if err != nil { diff --git a/apps/druid/core/services/runtime_lifecycle.go b/apps/druid/core/services/runtime_lifecycle.go index a0df0ef..5b03664 100644 --- a/apps/druid/core/services/runtime_lifecycle.go +++ b/apps/druid/core/services/runtime_lifecycle.go @@ -7,6 +7,8 @@ func (s *RuntimeSupervisor) Delete(id string) error { } func (s *RuntimeSupervisor) DeleteWithPolicy(id string, purgeData bool) error { + unlock := s.lockRuntimeOperation(id) + defer unlock() s.mu.Lock() session := s.sessions[id] delete(s.sessions, id) @@ -28,10 +30,17 @@ func (s *RuntimeSupervisor) DeleteWithPolicy(id string, purgeData bool) error { } func (s *RuntimeSupervisor) StartScroll(id string) (*domain.RuntimeScroll, error) { + unlock := s.lockRuntimeOperation(id) + defer unlock() + return s.startScroll(id) +} + +func (s *RuntimeSupervisor) startScroll(id string) (*domain.RuntimeScroll, error) { session, err := s.sessionFor(id) if err != nil { return nil, err } + session.Start() if err := session.AutoStartServe(); err != nil { session.markError(err) return nil, err @@ -52,11 +61,13 @@ func (s *RuntimeSupervisor) StartScroll(id string) (*domain.RuntimeScroll, error } func (s *RuntimeSupervisor) Stop(id string) (*domain.RuntimeScroll, error) { + unlock := s.lockRuntimeOperation(id) + defer unlock() session, err := s.detachSession(id) if err != nil { return nil, err } - if err := session.StopRuntime(); err != nil { + if err := session.StopRuntimeForMaintenance(); err != nil { session.markError(err) return nil, err } diff --git a/apps/druid/core/services/runtime_materialization.go b/apps/druid/core/services/runtime_materialization.go index 9db9ce7..fb4c662 100644 --- a/apps/druid/core/services/runtime_materialization.go +++ b/apps/druid/core/services/runtime_materialization.go @@ -30,14 +30,15 @@ func (s *RuntimeSupervisor) runPullWorker(ctx context.Context, runtimeService po defer cancel() callbackURL := s.workerCallbackURL + "/internal/v1/workers/" + runtimeID + "/complete" action := ports.RuntimeWorkerAction{ - Mode: mode, - RuntimeID: runtimeID, - Artifact: artifact, - Storage: storage, - RootRef: root, - MountPath: "/scroll", - CallbackURL: callbackURL, - RegistryCredentials: registryCredentials, + Mode: mode, + RuntimeID: runtimeID, + Artifact: artifact, + Storage: storage, + RootRef: root, + MountPath: "/scroll", + CallbackURL: callbackURL, + PreserveReleaseManifest: mode == ports.RuntimeWorkerModeRestore, + RegistryCredentials: registryCredentials, } workerDone, err := runtimeService.SpawnPullWorker(waitCtx, action) if err != nil { diff --git a/apps/druid/core/services/runtime_session_cache.go b/apps/druid/core/services/runtime_session_cache.go index 3fc7a79..1914542 100644 --- a/apps/druid/core/services/runtime_session_cache.go +++ b/apps/druid/core/services/runtime_session_cache.go @@ -1,11 +1,25 @@ package services import ( + "sync" + "github.com/highcard-dev/daemon/internal/core/domain" "github.com/highcard-dev/daemon/internal/utils/logger" "go.uber.org/zap" ) +func (s *RuntimeSupervisor) lockRuntimeOperation(id string) func() { + s.operationsMu.Lock() + operation := s.operations[id] + if operation == nil { + operation = &sync.Mutex{} + s.operations[id] = operation + } + s.operationsMu.Unlock() + operation.Lock() + return operation.Unlock +} + func (s *RuntimeSupervisor) detachSession(id string) (*RuntimeSession, error) { s.mu.Lock() session := s.sessions[id] diff --git a/apps/druid/core/services/runtime_session_runtime.go b/apps/druid/core/services/runtime_session_runtime.go index 1974724..9fbbede 100644 --- a/apps/druid/core/services/runtime_session_runtime.go +++ b/apps/druid/core/services/runtime_session_runtime.go @@ -77,6 +77,17 @@ func hasRunningProcedure(procedures domain.ProcedureStatusMap) bool { } func (s *RuntimeSession) StopRuntime() error { + return s.stopRuntime(true) +} + +// StopRuntimeForMaintenance keeps the runtime quiescent after the backend has +// confirmed it stopped. Backup and restore use this to avoid a queued serve +// command racing the filesystem operation. +func (s *RuntimeSession) StopRuntimeForMaintenance() error { + return s.stopRuntime(false) +} + +func (s *RuntimeSession) stopRuntime(resumeQueue bool) error { s.mu.Lock() root := s.runtimeScroll.Root started := s.started @@ -111,7 +122,7 @@ func (s *RuntimeSession) StopRuntime() error { s.runtimeScroll.LastError = "" err := s.store.UpdateScroll(s.runtimeScroll) s.mu.Unlock() - if err == nil && started { + if err == nil && started && resumeQueue { s.mu.Lock() s.started = true s.mu.Unlock() @@ -131,7 +142,7 @@ func (s *RuntimeSession) Backup(ctx context.Context, artifact string, registryCr s.mu.Lock() root := s.runtimeScroll.Root s.mu.Unlock() - return s.runtimeBackend.BackupRuntime(ctx, root, artifact, registryCredentials) + return s.runtimeBackend.BackupRuntime(ctx, root, artifact, registryCredentials, true) } func (s *RuntimeSession) ApplyRestore(materialized *ports.RuntimeMaterialization) error { diff --git a/apps/druid/core/services/runtime_supervisor.go b/apps/druid/core/services/runtime_supervisor.go index 6a63a23..6bc64ec 100644 --- a/apps/druid/core/services/runtime_supervisor.go +++ b/apps/druid/core/services/runtime_supervisor.go @@ -27,8 +27,10 @@ type RuntimeSupervisor struct { workerCallbackURL string workerTimeout time.Duration - mu sync.Mutex - sessions map[string]*RuntimeSession + mu sync.Mutex + sessions map[string]*RuntimeSession + operationsMu sync.Mutex + operations map[string]*sync.Mutex } type EnsureOptions struct { @@ -65,6 +67,7 @@ func NewRuntimeSupervisor( manager: manager, workerTimeout: 20 * time.Minute, sessions: map[string]*RuntimeSession{}, + operations: map[string]*sync.Mutex{}, } runtimeBackend, err := backendFactory.Create(supervisor) if err != nil { @@ -179,6 +182,8 @@ func (s *RuntimeSupervisor) createWithOwner(artifact string, name string, ownerI func (s *RuntimeSupervisor) Ensure(options EnsureOptions) (*domain.RuntimeScroll, error) { id := coreservices.RuntimeScrollIDFromName(options.Name) if id != "" { + unlock := s.lockRuntimeOperation(id) + defer unlock() runtimeScroll, err := s.store.GetScroll(id) if err == nil { if options.Namespace != "" && runtimeScroll.Root != "" { diff --git a/apps/druid/core/services/runtime_supervisor_test.go b/apps/druid/core/services/runtime_supervisor_test.go index 46d0119..17fc378 100644 --- a/apps/druid/core/services/runtime_supervisor_test.go +++ b/apps/druid/core/services/runtime_supervisor_test.go @@ -488,6 +488,21 @@ commands: } } +func TestRuntimeSessionStopForMaintenanceKeepsQueueQuiescent(t *testing.T) { + session := newRuntimeSessionForTest(t, map[string]domain.LockStatus{}, updatedScrollYAML("maintenance")) + session.Start() + + if err := session.StopRuntimeForMaintenance(); err != nil { + t.Fatal(err) + } + session.mu.Lock() + started := session.started + session.mu.Unlock() + if started { + t.Fatal("maintenance stop should keep the runtime queue paused") + } +} + func TestRuntimeSupervisorStartDoesNotHydrateStoppedScroll(t *testing.T) { store := newTestStateStore(t) runtimeScroll := &domain.RuntimeScroll{ @@ -919,7 +934,7 @@ func TestRuntimeSupervisorEnsureDoesNotRetryExistingError(t *testing.T) { Artifact: "registry.local/invalid:1.0", Root: store.Root("invalid-scroll"), ScrollName: "invalid-scroll", - ScrollYAML: cachedScrollYAML("start"), + ScrollYAML: updatedScrollYAML("backup-worker"), Status: domain.RuntimeScrollStatusError, LastError: "procedure field mode is unsupported", Procedures: domain.ProcedureStatusMap{}, @@ -953,7 +968,7 @@ func TestRuntimeSupervisorEnsureUpdatesChangedArtifact(t *testing.T) { Artifact: "registry.local/lab:1.0", Root: root, ScrollName: "old-scroll", - ScrollYAML: cachedScrollYAML("start"), + ScrollYAML: updatedScrollYAML("backup-recovery"), Status: domain.RuntimeScrollStatusRunning, Procedures: domain.ProcedureStatusMap{ "start": {"start.0": {Status: domain.ScrollLockStatusDone}}, @@ -1015,7 +1030,7 @@ func TestRuntimeSupervisorUpdateUsesPullWorkerWhenAvailable(t *testing.T) { Artifact: "registry.local/lab:1.0", Root: root, ScrollName: "old-scroll", - ScrollYAML: cachedScrollYAML("start"), + ScrollYAML: updatedScrollYAML("restore-recovery"), Status: domain.RuntimeScrollStatusStopped, Procedures: domain.ProcedureStatusMap{}, } @@ -1120,7 +1135,7 @@ func TestRuntimeSupervisorRestoreUsesPullWorkerResult(t *testing.T) { if backend.stopRoot != root { t.Fatalf("stop root = %s, want %s", backend.stopRoot, root) } - if backend.action.Mode != ports.RuntimeWorkerModeRestore || backend.action.RootRef != root || backend.action.Artifact != "registry.local/backup:1.0" { + if backend.action.Mode != ports.RuntimeWorkerModeRestore || backend.action.RootRef != root || backend.action.Artifact != "registry.local/backup:1.0" || !backend.action.PreserveReleaseManifest { t.Fatalf("worker action = %#v", backend.action) } if restored.Artifact != "registry.local/backup:1.0" || restored.ArtifactDigest != "sha256:restored" || restored.ScrollName != "restored-worker" { @@ -1137,6 +1152,280 @@ func TestRuntimeSupervisorRestoreUsesPullWorkerResult(t *testing.T) { } } +func TestRuntimeSupervisorBackupStopsAndRestartsRunningScroll(t *testing.T) { + store := newTestStateStore(t) + runtimeScroll := &domain.RuntimeScroll{ + ID: "backup-worker", + Artifact: "registry.local/lab:1.0", + Root: "runtime://backup-worker", + ScrollName: "backup-worker", + ScrollYAML: cachedScrollYAML("start"), + Status: domain.RuntimeScrollStatusRunning, + Procedures: domain.ProcedureStatusMap{}, + } + if err := store.CreateScroll(runtimeScroll); err != nil { + t.Fatal(err) + } + backend := &fakeWorkerBackend{} + supervisor := newRuntimeSupervisorForTest(t, store, coreservices.NewRuntimeScrollManager(store), backend) + + backedUp, err := supervisor.Backup("backup-worker", "registry.local/backups:1", nil) + if err != nil { + t.Fatal(err) + } + if backend.stopRoot != runtimeScroll.Root { + t.Fatalf("stop root = %q, want %q", backend.stopRoot, runtimeScroll.Root) + } + if backend.backupRoot != runtimeScroll.Root || backend.backupArtifact != "registry.local/backups:1" || !backend.backupPreserveReleaseManifest { + t.Fatalf("backup = root=%q artifact=%q preserve=%t", backend.backupRoot, backend.backupArtifact, backend.backupPreserveReleaseManifest) + } + if backedUp.Status != domain.RuntimeScrollStatusRunning { + t.Fatalf("backup status = %s, want running", backedUp.Status) + } +} + +func TestRuntimeSupervisorBackupFailureRestartsPriorRunningScroll(t *testing.T) { + store := newTestStateStore(t) + runtimeScroll := &domain.RuntimeScroll{ + ID: "backup-recovery", + Artifact: "registry.local/lab:1.0", + Root: "runtime://backup-recovery", + ScrollName: "backup-recovery", + ScrollYAML: cachedScrollYAML("start"), + Status: domain.RuntimeScrollStatusRunning, + Procedures: domain.ProcedureStatusMap{}, + } + if err := store.CreateScroll(runtimeScroll); err != nil { + t.Fatal(err) + } + backend := &fakeWorkerBackend{backupErr: errors.New("registry unavailable")} + supervisor := newRuntimeSupervisorForTest(t, store, coreservices.NewRuntimeScrollManager(store), backend) + + if _, err := supervisor.Backup("backup-recovery", "registry.local/backups:1", nil); err == nil { + t.Fatal("backup error = nil, want registry error") + } + recovered, err := store.GetScroll("backup-recovery") + if err != nil { + t.Fatal(err) + } + if recovered.Status != domain.RuntimeScrollStatusRunning { + t.Fatalf("recovered status = %s, want running", recovered.Status) + } +} + +func TestRuntimeSupervisorRestoreFailureRestartsPriorRunningScroll(t *testing.T) { + store := newTestStateStore(t) + runtimeScroll := &domain.RuntimeScroll{ + ID: "restore-recovery", + Artifact: "registry.local/lab:1.0", + Root: "runtime://restore-recovery", + ScrollName: "restore-recovery", + ScrollYAML: cachedScrollYAML("start"), + Status: domain.RuntimeScrollStatusRunning, + Procedures: domain.ProcedureStatusMap{}, + } + if err := store.CreateScroll(runtimeScroll); err != nil { + t.Fatal(err) + } + callbacks := NewWorkerCallbackManager() + backend := &fakeWorkerBackend{callbacks: callbacks, workerErr: errors.New("backup pull failed")} + supervisor := newRuntimeSupervisorForTest(t, store, coreservices.NewRuntimeScrollManager(store), backend) + supervisor.SetWorkerCallbacks(callbacks, "http://druid-cli:8083") + + if _, err := supervisor.Restore("restore-recovery", "registry.local/backups:1", true, nil); err == nil { + t.Fatal("restore error = nil, want pull error") + } + recovered, err := store.GetScroll("restore-recovery") + if err != nil { + t.Fatal(err) + } + if recovered.Status != domain.RuntimeScrollStatusRunning { + t.Fatalf("recovered status = %s, want running", recovered.Status) + } +} + +func TestRuntimeSupervisorUnsafeRestoreFailureKeepsRuntimeStopped(t *testing.T) { + store := newTestStateStore(t) + runtimeScroll := &domain.RuntimeScroll{ + ID: "restore-unsafe", + Artifact: "registry.local/lab:1.0", + Root: "runtime://restore-unsafe", + ScrollName: "restore-unsafe", + ScrollYAML: updatedScrollYAML("restore-unsafe"), + Status: domain.RuntimeScrollStatusRunning, + Procedures: domain.ProcedureStatusMap{}, + } + if err := store.CreateScroll(runtimeScroll); err != nil { + t.Fatal(err) + } + callbacks := NewWorkerCallbackManager() + backend := &fakeWorkerBackend{callbacks: callbacks, workerErr: errors.New("restore root may be partial: injected rollback failure")} + supervisor := newRuntimeSupervisorForTest(t, store, coreservices.NewRuntimeScrollManager(store), backend) + supervisor.SetWorkerCallbacks(callbacks, "http://druid-cli:8083") + + if _, err := supervisor.Restore("restore-unsafe", "registry.local/backups:1", true, nil); err == nil { + t.Fatal("restore error = nil, want unsafe restore failure") + } + updated, err := store.GetScroll("restore-unsafe") + if err != nil { + t.Fatal(err) + } + if updated.Status == domain.RuntimeScrollStatusRunning { + t.Fatalf("unsafe restore failure must not restart runtime: %#v", updated) + } + if !strings.Contains(updated.LastError, "restore root may be partial") { + t.Fatalf("last error = %q, want unsafe restore detail", updated.LastError) + } +} + +func TestRuntimeSupervisorSerializesBackupWithStartAndEnsure(t *testing.T) { + store := newTestStateStore(t) + runtimeScroll := &domain.RuntimeScroll{ + ID: "serialized-operation", + Artifact: "registry.local/lab:1.0", + Root: "runtime://serialized-operation", + ScrollName: "serialized-operation", + ScrollYAML: updatedScrollYAML("serialized-operation"), + Status: domain.RuntimeScrollStatusRunning, + Procedures: domain.ProcedureStatusMap{}, + } + if err := store.CreateScroll(runtimeScroll); err != nil { + t.Fatal(err) + } + backupStarted := make(chan struct{}) + backupRelease := make(chan struct{}) + backend := &fakeWorkerBackend{backupStarted: backupStarted, backupRelease: backupRelease} + supervisor := newRuntimeSupervisorForTest(t, store, coreservices.NewRuntimeScrollManager(store), backend) + + backupDone := make(chan error, 1) + go func() { + _, err := supervisor.Backup(runtimeScroll.ID, "registry.local/backups:1", nil) + backupDone <- err + }() + select { + case <-backupStarted: + case <-time.After(time.Second): + t.Fatal("backup did not reach its stopped operation") + } + + startDone := make(chan error, 1) + go func() { + _, err := supervisor.StartScroll(runtimeScroll.ID) + startDone <- err + }() + ensureDone := make(chan error, 1) + go func() { + _, err := supervisor.Ensure(EnsureOptions{Name: runtimeScroll.ID}) + ensureDone <- err + }() + for name, done := range map[string]<-chan error{"start": startDone, "ensure": ensureDone} { + select { + case err := <-done: + t.Fatalf("%s completed during backup: %v", name, err) + case <-time.After(100 * time.Millisecond): + } + } + + close(backupRelease) + for name, done := range map[string]<-chan error{"backup": backupDone, "start": startDone, "ensure": ensureDone} { + select { + case err := <-done: + if err != nil { + t.Fatalf("%s error = %v", name, err) + } + case <-time.After(time.Second): + t.Fatalf("%s did not complete after backup", name) + } + } +} + +func TestRuntimeSupervisorStopKeepsRuntimeQueueQuiescent(t *testing.T) { + store := newTestStateStore(t) + runtimeScroll := &domain.RuntimeScroll{ + ID: "stop-quiescent", + Artifact: "registry.local/lab:1.0", + Root: "runtime://stop-quiescent", + ScrollName: "stop-quiescent", + ScrollYAML: updatedScrollYAML("stop-quiescent"), + Status: domain.RuntimeScrollStatusRunning, + Procedures: domain.ProcedureStatusMap{}, + } + if err := store.CreateScroll(runtimeScroll); err != nil { + t.Fatal(err) + } + var runs atomic.Int32 + backend := &fakeWorkerBackend{runCommand: func(command ports.RuntimeCommand) (*int, error) { + runs.Add(1) + return nil, errors.New("run should not persist in test") + }} + supervisor := newRuntimeSupervisorForTest(t, store, coreservices.NewRuntimeScrollManager(store), backend) + session, err := supervisor.sessionFor(runtimeScroll.ID) + if err != nil { + t.Fatal(err) + } + if err := session.AutoStartServe(); err != nil { + t.Fatal(err) + } + deadline := time.After(time.Second) + for runs.Load() == 0 { + select { + case <-deadline: + t.Fatal("runtime queue did not start") + default: + time.Sleep(10 * time.Millisecond) + } + } + if _, err := supervisor.Stop(runtimeScroll.ID); err != nil { + t.Fatal(err) + } + count := runs.Load() + time.Sleep(100 * time.Millisecond) + if got := runs.Load(); got != count { + t.Fatalf("runtime queue restarted after stop: runs=%d, want %d", got, count) + } +} + +func TestRuntimeSupervisorStartResumesMaintenanceStoppedQueue(t *testing.T) { + store := newTestStateStore(t) + runtimeScroll := &domain.RuntimeScroll{ + ID: "resume-maintenance", + Artifact: "registry.local/lab:1.0", + Root: "runtime://resume-maintenance", + ScrollName: "resume-maintenance", + ScrollYAML: updatedScrollYAML("resume-maintenance"), + Status: domain.RuntimeScrollStatusRunning, + Procedures: domain.ProcedureStatusMap{}, + } + if err := store.CreateScroll(runtimeScroll); err != nil { + t.Fatal(err) + } + var runs atomic.Int32 + backend := &fakeWorkerBackend{runCommand: func(command ports.RuntimeCommand) (*int, error) { + runs.Add(1) + return nil, errors.New("run should not persist in test") + }} + supervisor := newRuntimeSupervisorForTest(t, store, coreservices.NewRuntimeScrollManager(store), backend) + session, err := supervisor.sessionFor(runtimeScroll.ID) + if err != nil { + t.Fatal(err) + } + if err := session.StopRuntimeForMaintenance(); err != nil { + t.Fatal(err) + } + if _, err := supervisor.StartScroll(runtimeScroll.ID); err != nil { + t.Fatal(err) + } + deadline := time.After(time.Second) + for runs.Load() == 0 { + select { + case <-deadline: + t.Fatal("start did not resume the maintenance-stopped runtime queue") + default: + time.Sleep(10 * time.Millisecond) + } + } +} + func TestNewRuntimeSessionRequiresPersistedScrollYAML(t *testing.T) { store := newTestStateStore(t) runtimeScroll := &domain.RuntimeScroll{ @@ -1374,22 +1663,28 @@ func newRuntimeSessionForTest(t *testing.T, commands map[string]domain.LockStatu } type fakeWorkerBackend struct { - callbacks *WorkerCallbackManager - procedureStatusObserver ports.ProcedureStatusObserver - procedureStatusUpdates []ports.ProcedureStatusUpdate - scrollYAML string - digest string - workerErr error - workerDone <-chan error - action ports.RuntimeWorkerAction - stopRoot string - deleteRoot string - spawnCount int - runCommand func(ports.RuntimeCommand) (*int, error) - stopRuntime func(string) error - console io.ReadWriteCloser - consoleRoot string - consoleProcedure string + callbacks *WorkerCallbackManager + procedureStatusObserver ports.ProcedureStatusObserver + procedureStatusUpdates []ports.ProcedureStatusUpdate + scrollYAML string + digest string + workerErr error + workerDone <-chan error + action ports.RuntimeWorkerAction + stopRoot string + deleteRoot string + backupRoot string + backupArtifact string + backupPreserveReleaseManifest bool + backupErr error + backupStarted chan struct{} + backupRelease <-chan struct{} + spawnCount int + runCommand func(ports.RuntimeCommand) (*int, error) + stopRuntime func(string) error + console io.ReadWriteCloser + consoleRoot string + consoleProcedure string } func newRuntimeSupervisorForTest(t *testing.T, store ports.RuntimeScrollStore, manager *coreservices.RuntimeScrollManager, factory ports.RuntimeBackendFactory) *RuntimeSupervisor { @@ -1474,8 +1769,17 @@ func (f *fakeWorkerBackend) DeleteRuntime(root string, purgeData bool) error { return nil } -func (f *fakeWorkerBackend) BackupRuntime(ctx context.Context, root string, artifact string, registryCredentials []domain.RegistryCredential) error { - return nil +func (f *fakeWorkerBackend) BackupRuntime(ctx context.Context, root string, artifact string, registryCredentials []domain.RegistryCredential, preserveReleaseManifest bool) error { + f.backupRoot = root + f.backupArtifact = artifact + f.backupPreserveReleaseManifest = preserveReleaseManifest + if f.backupStarted != nil { + close(f.backupStarted) + } + if f.backupRelease != nil { + <-f.backupRelease + } + return f.backupErr } func (f *fakeWorkerBackend) SpawnPullWorker(ctx context.Context, action ports.RuntimeWorkerAction) (<-chan error, error) { diff --git a/apps/druid/core/services/runtime_update.go b/apps/druid/core/services/runtime_update.go index 2236082..ff2823b 100644 --- a/apps/druid/core/services/runtime_update.go +++ b/apps/druid/core/services/runtime_update.go @@ -12,6 +12,8 @@ import ( ) func (s *RuntimeSupervisor) Update(id string, artifact string, registryCredentials []domain.RegistryCredential) (*domain.RuntimeScroll, error) { + unlock := s.lockRuntimeOperation(id) + defer unlock() runtimeScroll, err := s.store.GetScroll(id) if err != nil { return nil, err @@ -100,7 +102,7 @@ func (s *RuntimeSupervisor) updateExistingScroll(runtimeScroll *domain.RuntimeSc return nil, err } if wasRunning && restartIfRunning { - return s.StartScroll(runtimeScroll.ID) + return s.startScroll(runtimeScroll.ID) } return s.store.GetScroll(runtimeScroll.ID) } diff --git a/internal/core/ports/services_ports.go b/internal/core/ports/services_ports.go index 9967dee..6b16a12 100644 --- a/internal/core/ports/services_ports.go +++ b/internal/core/ports/services_ports.go @@ -41,7 +41,7 @@ type RuntimeBackendInterface interface { StopCommand(root string, command string) error StopRuntime(root string) error DeleteRuntime(root string, purgeData bool) error - BackupRuntime(ctx context.Context, root string, artifact string, registryCredentials []domain.RegistryCredential) error + BackupRuntime(ctx context.Context, root string, artifact string, registryCredentials []domain.RegistryCredential, preserveReleaseManifest bool) error SpawnPullWorker(ctx context.Context, action RuntimeWorkerAction) (<-chan error, error) OpenConsole(ctx context.Context, root string, procedure string) (io.ReadWriteCloser, error) Signal(commandName string, target string, signal string, root string) error @@ -162,15 +162,16 @@ const ( ) type RuntimeWorkerAction struct { - Mode RuntimeWorkerMode - RuntimeID string - Artifact string - Storage string - RootRef string - MountPath string - CallbackURL string - TokenFile string - RegistryCredentials []domain.RegistryCredential + Mode RuntimeWorkerMode + RuntimeID string + Artifact string + Storage string + RootRef string + MountPath string + CallbackURL string + TokenFile string + PreserveReleaseManifest bool + RegistryCredentials []domain.RegistryCredential } type RuntimeWorkerResult struct { diff --git a/internal/runtime/backend_factory_test.go b/internal/runtime/backend_factory_test.go index 8608a1b..15d0bb5 100644 --- a/internal/runtime/backend_factory_test.go +++ b/internal/runtime/backend_factory_test.go @@ -162,7 +162,7 @@ func (f fakeBackend) DeleteRuntime(root string, purgeData bool) error { return nil } -func (f fakeBackend) BackupRuntime(ctx context.Context, root string, artifact string, registryCredentials []domain.RegistryCredential) error { +func (f fakeBackend) BackupRuntime(ctx context.Context, root string, artifact string, registryCredentials []domain.RegistryCredential, preserveReleaseManifest bool) error { return nil } diff --git a/internal/runtime/docker/backend_names_test.go b/internal/runtime/docker/backend_names_test.go index 086957e..5e155dc 100644 --- a/internal/runtime/docker/backend_names_test.go +++ b/internal/runtime/docker/backend_names_test.go @@ -2,7 +2,6 @@ package docker import ( "errors" - "runtime" "strings" "testing" @@ -151,12 +150,6 @@ func TestContainerSpecAddsHostGatewayExtraHost(t *testing.T) { if err != nil { t.Fatal(err) } - if runtime.GOOS != "linux" { - if len(hostConfig.ExtraHosts) != 0 { - t.Fatalf("extra hosts = %#v, want none on %s", hostConfig.ExtraHosts, runtime.GOOS) - } - return - } for _, extraHost := range hostConfig.ExtraHosts { if extraHost == dockerHostGatewayExtraHost { return diff --git a/internal/runtime/docker/names.go b/internal/runtime/docker/names.go index 7b8b8a2..7177146 100644 --- a/internal/runtime/docker/names.go +++ b/internal/runtime/docker/names.go @@ -7,7 +7,6 @@ import ( "os" "path/filepath" "regexp" - "runtime" "strings" "github.com/highcard-dev/daemon/internal/core/domain" @@ -29,10 +28,7 @@ const ( const dockerFailedProcedureRetention = 3 func dockerExtraHosts() []string { - if runtime.GOOS == "linux" { - return []string{dockerHostGatewayExtraHost} - } - return nil + return []string{dockerHostGatewayExtraHost} } func ContainerName(root string, commandName string) string { diff --git a/internal/runtime/docker/workers.go b/internal/runtime/docker/workers.go index dbacedf..7b635ed 100644 --- a/internal/runtime/docker/workers.go +++ b/internal/runtime/docker/workers.go @@ -16,15 +16,19 @@ import ( "github.com/highcard-dev/daemon/internal/core/ports" ) -func (b *Backend) BackupRuntime(ctx context.Context, root string, artifact string, registryCredentials []domain.RegistryCredential) error { +func (b *Backend) BackupRuntime(ctx context.Context, root string, artifact string, registryCredentials []domain.RegistryCredential, preserveReleaseManifest bool) error { if artifact == "" { return fmt.Errorf("backup artifact is required") } - return b.runWorkerRootCommand(ctx, root, []string{ + command := []string{ "worker", "push", "--artifact", artifact, "--root", "/scroll", - }, registryCredentials) + } + if preserveReleaseManifest { + command = append(command, "--preserve-release-manifest") + } + return b.runWorkerRootCommand(ctx, root, command, registryCredentials) } func (b *Backend) runWorkerRootCommand(ctx context.Context, root string, command []string, registryCredentials []domain.RegistryCredential) error { @@ -146,14 +150,7 @@ func (b *Backend) SpawnPullWorker(ctx context.Context, action ports.RuntimeWorke created, err := b.client.ContainerCreate(ctx, &container.Config{ Image: b.config.WorkerImage, Entrypoint: []string{"druid"}, - Cmd: []string{ - "worker", "pull", - "--artifact", artifact, - "--runtime-id", action.RuntimeID, - "--mode", string(action.Mode), - "--root", action.MountPath, - "--callback-url", action.CallbackURL, - }, + Cmd: workerPullCommand(action, artifact), Env: dockerWorkerEnv([]string{ "DRUID_WORKER_TOKEN_FILE=" + action.TokenFile, "DRUID_RUNTIME_REGISTRY_CONFIG_JSON=" + string(registryConfig), @@ -195,3 +192,18 @@ func (b *Backend) SpawnPullWorker(ctx context.Context, action ports.RuntimeWorke }() return done, nil } + +func workerPullCommand(action ports.RuntimeWorkerAction, artifact string) []string { + command := []string{ + "worker", "pull", + "--artifact", artifact, + "--runtime-id", action.RuntimeID, + "--mode", string(action.Mode), + "--root", action.MountPath, + "--callback-url", action.CallbackURL, + } + if action.PreserveReleaseManifest { + command = append(command, "--preserve-release-manifest") + } + return command +} diff --git a/internal/runtime/kubernetes/resources.go b/internal/runtime/kubernetes/resources.go index 5949af7..5d3b40a 100644 --- a/internal/runtime/kubernetes/resources.go +++ b/internal/runtime/kubernetes/resources.go @@ -85,6 +85,9 @@ func workerPullJobSpec(namespace string, jobName string, pvc string, image strin "--root", action.MountPath, "--callback-url", action.CallbackURL, } + if action.PreserveReleaseManifest { + command = append(command, "--preserve-release-manifest") + } job := helperJobSpec(namespace, jobName, pvc, image, command, imagePullSecret, map[string]string{ labelComponent: "worker-pull", labelRuntimeID: runtimeLabel(action.RuntimeID), @@ -128,11 +131,14 @@ func runtimeLabel(runtimeID string) string { return dnsLabel(runtimeID) } -func backupJobSpec(namespace string, jobName string, pvc string, image string, artifact string, imagePullSecret string, registryConfigSecret string, registryPlainHTTP bool) *batchv1.Job { +func backupJobSpec(namespace string, jobName string, pvc string, image string, artifact string, imagePullSecret string, registryConfigSecret string, registryPlainHTTP bool, preserveReleaseManifest bool) *batchv1.Job { command := []string{"druid", "push", artifact, "/scroll"} if registryConfigSecret != "" { command = []string{"sh", "-c", registryConfigScript, "sh", "push", artifact, "/scroll"} } + if preserveReleaseManifest { + command = append(command, "--preserve-release-manifest") + } job := helperJobSpec(namespace, jobName, pvc, image, command, imagePullSecret, map[string]string{ labelComponent: "backup", }) diff --git a/internal/runtime/kubernetes/resources_test.go b/internal/runtime/kubernetes/resources_test.go index ae58d81..766468a 100644 --- a/internal/runtime/kubernetes/resources_test.go +++ b/internal/runtime/kubernetes/resources_test.go @@ -443,18 +443,19 @@ func TestProcedureWorkloadsLabelEveryExpectedPort(t *testing.T) { func TestWorkerPullJobSpecRunsDruidWorkerPull(t *testing.T) { action := ports.RuntimeWorkerAction{ - Mode: ports.RuntimeWorkerModeUpdate, - RuntimeID: "deployment-123", - Artifact: "registry.local/lab:2.0", - MountPath: "/scroll", - CallbackURL: "http://druid-cli:8083/internal/v1/workers/deployment-123/complete", - TokenFile: "token-file", + Mode: ports.RuntimeWorkerModeUpdate, + RuntimeID: "deployment-123", + Artifact: "registry.local/lab:2.0", + MountPath: "/scroll", + CallbackURL: "http://druid-cli:8083/internal/v1/workers/deployment-123/complete", + TokenFile: "token-file", + PreserveReleaseManifest: true, } job := workerPullJobSpec("druid", "worker-pull", "runtime-pvc", "druid-cli:test", action, "pull-secret", "runtime-registry", true, "druid-cli") assertFinishedJobTTL(t, job) container := job.Spec.Template.Spec.Containers[0] command := strings.Join(container.Command, " ") - for _, want := range []string{"druid --config /tmp/druid-registry.json", "worker pull", "--mode update", "--runtime-id deployment-123", "--callback-url", "chown -R 1000:1000"} { + for _, want := range []string{"druid --config /tmp/druid-registry.json", "worker pull", "--mode update", "--runtime-id deployment-123", "--callback-url", "--preserve-release-manifest", "chown -R 1000:1000"} { if !strings.Contains(command, want) { t.Fatalf("command = %#v, want %s", container.Command, want) } @@ -1628,11 +1629,14 @@ func TestDeleteRuntimePurgesServicesAndDataWhenRequested(t *testing.T) { } func TestBackupJobSpecUsesRuntimePVCAndRegistryEnv(t *testing.T) { - backup := backupJobSpec("druid", "backup", "runtime-pvc", "druid-cli:test", "registry.local/scroll:backup", "registry-secret", "", true) + backup := backupJobSpec("druid", "backup", "runtime-pvc", "druid-cli:test", "registry.local/scroll:backup", "registry-secret", "", true, true) assertFinishedJobTTL(t, backup) if backup.Spec.Template.Spec.Containers[0].Command[1] != "push" { t.Fatalf("backup command = %#v", backup.Spec.Template.Spec.Containers[0].Command) } + if !strings.Contains(strings.Join(backup.Spec.Template.Spec.Containers[0].Command, " "), "--preserve-release-manifest") { + t.Fatalf("backup command = %#v, want release manifest preservation", backup.Spec.Template.Spec.Containers[0].Command) + } if got := backup.Spec.Template.Spec.Volumes[0].PersistentVolumeClaim.ClaimName; got != "runtime-pvc" { t.Fatalf("backup PVC = %s, want runtime-pvc", got) } diff --git a/internal/runtime/kubernetes/workers.go b/internal/runtime/kubernetes/workers.go index dbe8d68..d90bd55 100644 --- a/internal/runtime/kubernetes/workers.go +++ b/internal/runtime/kubernetes/workers.go @@ -92,7 +92,7 @@ func (b *Backend) SpawnPullWorker(ctx context.Context, action ports.RuntimeWorke return result, nil } -func (b *Backend) BackupRuntime(ctx context.Context, root string, artifact string, registryCredentials []domain.RegistryCredential) error { +func (b *Backend) BackupRuntime(ctx context.Context, root string, artifact string, registryCredentials []domain.RegistryCredential, preserveReleaseManifest bool) error { if artifact == "" { return fmt.Errorf("backup artifact is required") } @@ -111,7 +111,7 @@ func (b *Backend) BackupRuntime(ctx context.Context, root string, artifact strin return err } defer cleanupRegistryConfig() - job := backupJobSpec(namespace, jobName("backup", root, shortHash(artifact)), pvc, b.config.PullImage, artifact, b.config.RegistrySecret, registryConfigSecret, b.config.RegistryPlainHTTP) + job := backupJobSpec(namespace, jobName("backup", root, shortHash(artifact)), pvc, b.config.PullImage, artifact, b.config.RegistrySecret, registryConfigSecret, b.config.RegistryPlainHTTP, preserveReleaseManifest) if err := b.pinPodToRuntimeNode(ctx, namespace, pvc, &job.Spec.Template.Spec); err != nil { return err } diff --git a/test/integration/docker/docker_cli_test.go b/test/integration/docker/docker_cli_test.go index 71aeb73..2895304 100644 --- a/test/integration/docker/docker_cli_test.go +++ b/test/integration/docker/docker_cli_test.go @@ -147,7 +147,13 @@ func TestDockerBackendVolumeStorageWorkerLifecycleBackupRestore(t *testing.T) { } e2e.UnixJSONRequest(t, socket, http.MethodPost, "/api/v1/scrolls/"+created.ID+"/backup", fmt.Sprintf(`{"artifact":%q}`, backupArtifact)) + if got := e2e.WaitHTTP(t, fmt.Sprintf("http://127.0.0.1:%d/index.txt", fixture.RoutePort)); !strings.Contains(got, "healthy") { + t.Fatalf("backup restart index = %q, want healthy", got) + } writeDockerRootFile(t, volume, "data/public/index.txt", "mutated\n") + if got := readDockerRootFile(t, volume, "data/public/index.txt"); !strings.Contains(got, "mutated") { + t.Fatalf("mutated root index = %q, want mutated", got) + } if got := e2e.WaitHTTP(t, fmt.Sprintf("http://127.0.0.1:%d/index.txt", fixture.RoutePort)); !strings.Contains(got, "mutated") { t.Fatalf("mutated index = %q, want mutated", got) } diff --git a/test/mock/services.go b/test/mock/services.go index cd3883b..90fb284 100644 --- a/test/mock/services.go +++ b/test/mock/services.go @@ -211,17 +211,17 @@ func (m *MockRuntimeBackendInterface) EXPECT() *MockRuntimeBackendInterfaceMockR } // BackupRuntime mocks base method. -func (m *MockRuntimeBackendInterface) BackupRuntime(ctx context.Context, root, artifact string, registryCredentials []domain.RegistryCredential) error { +func (m *MockRuntimeBackendInterface) BackupRuntime(ctx context.Context, root, artifact string, registryCredentials []domain.RegistryCredential, preserveReleaseManifest bool) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "BackupRuntime", ctx, root, artifact, registryCredentials) + ret := m.ctrl.Call(m, "BackupRuntime", ctx, root, artifact, registryCredentials, preserveReleaseManifest) ret0, _ := ret[0].(error) return ret0 } // BackupRuntime indicates an expected call of BackupRuntime. -func (mr *MockRuntimeBackendInterfaceMockRecorder) BackupRuntime(ctx, root, artifact, registryCredentials any) *gomock.Call { +func (mr *MockRuntimeBackendInterfaceMockRecorder) BackupRuntime(ctx, root, artifact, registryCredentials, preserveReleaseManifest any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BackupRuntime", reflect.TypeOf((*MockRuntimeBackendInterface)(nil).BackupRuntime), ctx, root, artifact, registryCredentials) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BackupRuntime", reflect.TypeOf((*MockRuntimeBackendInterface)(nil).BackupRuntime), ctx, root, artifact, registryCredentials, preserveReleaseManifest) } // CreateUIPackageUpload mocks base method.