diff --git a/docs/deploy.md b/docs/deploy.md index 9f4c32511..0ab90fcb2 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -680,7 +680,11 @@ dying with the service's cgroup. sandboxd starts no new `cocoon` command once stopped and waits for the running ones, since cocoon dies on its next log line once sandboxd's end of its stderr closes, and a hibernate killed that way leaves the guest paused. A save or hibernate can take 26 minutes, hence -`TimeoutStopSec=27min`; stops normally take seconds. Claimed +`TimeoutStopSec=27min`; stops normally take seconds. If sandboxd dies abruptly during +a capture instead, the next start holds that claim until the `cocoon` command has +exited, then settles it: a hibernate whose snapshot landed stays hibernated and wakes +from it, and any other capture resumes the guest, or marks the claim failed if it +does not answer within 30 minutes. Claimed sandboxes are reaped when their TTL expires (default 5m, capped at 24h). A snapshot save, hibernate, export, restore or clone has 15 minutes. Once canceled, a diff --git a/e2e/fakeengine_test.go b/e2e/fakeengine_test.go index de87f1e05..afb0e23e7 100644 --- a/e2e/fakeengine_test.go +++ b/e2e/fakeengine_test.go @@ -180,6 +180,8 @@ func (f *fakeEngine) VMEvents(ctx context.Context, _ string, _ engine.VMSyncFunc return ctx.Err() } +func (f *fakeEngine) WaitCapture(context.Context, string) error { return nil } + func (f *fakeEngine) Probe(ctx context.Context, vsockSocket string, timeout time.Duration) error { return f.real.Probe(ctx, vsockSocket, timeout) } diff --git a/sandboxd/engine/capture.go b/sandboxd/engine/capture.go new file mode 100644 index 000000000..18ff84871 --- /dev/null +++ b/sandboxd/engine/capture.go @@ -0,0 +1,32 @@ +package engine + +import ( + "bytes" + "context" + "os" + "path/filepath" + "slices" + "time" +) + +const captureExitPoll = time.Second + +func (e *Engine) WaitCapture(ctx context.Context, snap string) error { + for captureRunning(snap) { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(captureExitPoll): + } + } + return nil +} + +func captureRunning(snap string) bool { + procs, _ := os.ReadDir("/proc") + arg := []byte("\x00" + snap + "\x00") + return slices.ContainsFunc(procs, func(proc os.DirEntry) bool { + cmdline, err := os.ReadFile(filepath.Join("/proc", proc.Name(), "cmdline")) + return err == nil && bytes.Contains(cmdline, arg) + }) +} diff --git a/sandboxd/engine/engine_test.go b/sandboxd/engine/engine_test.go index 8f65e83f7..705afbf02 100644 --- a/sandboxd/engine/engine_test.go +++ b/sandboxd/engine/engine_test.go @@ -7,7 +7,9 @@ import ( "io" "net" "os" + "os/exec" "path/filepath" + "runtime" "slices" "strconv" "strings" @@ -328,6 +330,33 @@ func TestCloseWaitsForARunningCommandAndRefusesNewOnes(t *testing.T) { } } +func TestWaitCaptureBlocksOnAnExactArgument(t *testing.T) { + if runtime.GOOS != "linux" { + t.Skip("reads /proc") + } + snap := "sbx-hib-" + strings.ReplaceAll(t.Name(), "/", "-") + cmd := exec.Command("sh", "-c", "sleep 30; :", "--name", snap) + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }) + done := make(chan error, 1) + go func() { done <- newTestEngine().WaitCapture(t.Context(), snap) }() + select { + case err := <-done: + t.Fatalf("WaitCapture returned %v with the process alive", err) + case <-time.After(2 * captureExitPoll): + } + if err := newTestEngine().WaitCapture(t.Context(), snap[:len(snap)-1]); err != nil { + t.Errorf("a prefix of the name: %v", err) + } + _ = cmd.Process.Kill() + _ = cmd.Wait() + if err := <-done; err != nil { + t.Errorf("WaitCapture after exit: %v", err) + } +} + func TestADialOfAGuestThatNeverAnswersFails(t *testing.T) { t.Parallel() path := sockPath(t) diff --git a/sandboxd/pool/capture.go b/sandboxd/pool/capture.go index 8d26c15bb..a1cd3b56b 100644 --- a/sandboxd/pool/capture.go +++ b/sandboxd/pool/capture.go @@ -2,6 +2,8 @@ package pool import ( "context" + "errors" + "fmt" "strings" "github.com/projecteru2/core/log" @@ -22,8 +24,20 @@ func (m *Manager) sourceSnap(ctx context.Context, sb *types.Sandbox) (string, fu return sb.HibernateSnap, func() {}, nil } snap := forkPrefix + strings.TrimPrefix(sb.VMName, vmPrefix) + "-" + randHex(3) + intent, live := m.setIfLive(sb, func() { sb.SaveSnap = snap }) + if !live { + return "", nil, ErrUnknownSandbox + } + if err := m.store.commit(intent); err != nil { + m.setIfLive(sb, func() { sb.SaveSnap = "" }) + return "", nil, fmt.Errorf("save %s: persist intent: %w", sb.ID, err) + } if err := m.eng.SnapshotSave(ctx, sb.VMName, snap); err != nil { - return "", nil, err + _, settleErr := m.settleFailedCapture(ctx, sb) + return "", nil, errors.Join(err, settleErr) + } + if _, err := m.commitIfLive(ctx, sb, func() { sb.SaveSnap = "" }); err != nil { + log.WithFunc("pool.sourceSnap").Warnf(ctx, "clear save intent of %s: %v", sb.ID, err) } return snap, func() { m.dropSnap(ctx, snap) }, nil } diff --git a/sandboxd/pool/claims.go b/sandboxd/pool/claims.go index 195a04d42..50f448160 100644 --- a/sandboxd/pool/claims.go +++ b/sandboxd/pool/claims.go @@ -54,6 +54,7 @@ type claimDTO struct { TAP string `json:"tap,omitempty"` HibernateSnap string `json:"hibernate_snap,omitempty"` PendingSnap string `json:"pending_snap,omitempty"` + SaveSnap string `json:"save_snap,omitempty"` ReaderSafe bool `json:"reader_safe,omitzero"` ArchiveCk string `json:"archive_ck,omitempty"` ArchiveEpoch int64 `json:"archive_epoch,omitzero"` @@ -394,7 +395,7 @@ func dtoOf(sb *types.Sandbox) claimDTO { PolicySource: sb.PolicySource, NoEgress: sb.NoEgress, Tenant: sb.Tenant, EgressClass: sb.EgressClass, ClaimRef: sb.ClaimRef, Metadata: sb.Metadata, OnExpire: sb.OnExpire, Volumes: slices.Clone(sb.Volumes), VolumesWant: volumesWantDTO(sb), Env: sb.Env, VsockSocket: sb.VsockSocket, - TAP: sb.TAP, HibernateSnap: sb.HibernateSnap, PendingSnap: sb.PendingSnap, ReaderSafe: sb.ReaderSafe, + TAP: sb.TAP, HibernateSnap: sb.HibernateSnap, PendingSnap: sb.PendingSnap, SaveSnap: sb.SaveSnap, ReaderSafe: sb.ReaderSafe, ArchiveCk: sb.ArchiveCk, ArchiveEpoch: sb.ArchiveEpoch, ArchivePending: sb.ArchivePending, ReplicaCk: sb.ReplicaCk, ReplicaPending: sb.ReplicaPending, FromCheckpoint: sb.FromCheckpoint, Restarts: sb.Restarts, RestartedAt: sb.RestartedAt, Failed: sb.Failed, MoveOut: sb.MoveOut, MoveIn: sb.MoveIn, } diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index 9033ee49d..319555743 100644 --- a/sandboxd/pool/hibernate.go +++ b/sandboxd/pool/hibernate.go @@ -83,7 +83,7 @@ func (m *Manager) hibernateLocked(ctx context.Context, sb *types.Sandbox) error // cocoon resumes the guest before its fresh tap can be re-locked return ErrNoEgressHibernate } - if _, err := m.settlePendingSnap(ctx, sb); err != nil { + if err := m.settleCapture(ctx, sb); err != nil { return err } if sb.HibernateSnap != "" { @@ -122,17 +122,16 @@ func (m *Manager) hibernateLocked(ctx context.Context, sb *types.Sandbox) error m.mu.Unlock() if hibErr := m.eng.Hibernate(ctx, sb.VMName, snap); hibErr != nil { // The engine can report failure after the snapshot landed - adopted, resolveErr := m.settlePendingSnap(ctx, sb) + adopted, settleErr := m.settleFailedCapture(ctx, sb) if adopted { - m.out.Disarm(sb.ID, true) - if resolveErr != nil { - return fmt.Errorf("hibernate %s: persist claims: %w", sb.ID, resolveErr) + if settleErr != nil { + return fmt.Errorf("hibernate %s: persist claims: %w", sb.ID, settleErr) } m.dropStale(ctx, sb) return nil } - if resolveErr != nil { - return errors.Join(fmt.Errorf("hibernate %s: %w", sb.ID, hibErr), resolveErr) + if settleErr != nil { + return errors.Join(fmt.Errorf("hibernate %s: %w", sb.ID, hibErr), settleErr) } return fmt.Errorf("hibernate %s: %w", sb.ID, hibErr) } @@ -170,7 +169,7 @@ func (m *Manager) wakeOrRead(ctx context.Context, sb *types.Sandbox, readerOK bo if sb.Failed != "" { return "", fmt.Errorf("%w: %s", ErrFailed, sb.Failed) } - if _, err := m.settlePendingSnap(ctx, sb); err != nil { + if err := m.settleCapture(ctx, sb); err != nil { return "", fmt.Errorf("wake %s: %w", sb.ID, err) } if sb.HibernateSnap == "" { diff --git a/sandboxd/pool/hibernate_test.go b/sandboxd/pool/hibernate_test.go index 9ff72a479..7da12a3cc 100644 --- a/sandboxd/pool/hibernate_test.go +++ b/sandboxd/pool/hibernate_test.go @@ -8,6 +8,7 @@ import ( "strings" "testing" "testing/synctest" + "time" "github.com/cocoonstack/sandbox/sandboxd/config" "github.com/cocoonstack/sandbox/sandboxd/types" @@ -109,10 +110,10 @@ func TestHibernateVerifiedFailureClearsIntent(t *testing.T) { if _, g := m.Info(); g.Hibernated != 0 { t.Errorf("hibernated %d, want 0 (still running)", g.Hibernated) } - got := journaled(t, m) - if got[sb.ID] == nil || got[sb.ID].HibernateSnap != "" || got[sb.ID].PendingSnap != "" { - t.Fatalf("journal %+v, want running (no snaps)", got[sb.ID]) - } + waitFor(t, func() bool { + row := journaled(t, m)[sb.ID] + return row != nil && row.HibernateSnap == "" && row.PendingSnap == "" + }) eng.hibernateErr = nil if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { t.Fatalf("Hibernate retry: %v", err) @@ -192,8 +193,9 @@ func TestRetryResolvesDanglingIntent(t *testing.T) { t.Fatalf("intent %q, want the real snapshot %q kept", got[sb.ID].PendingSnap, realSnap) } - eng.hibernateErr = nil - eng.snapListErr = nil + eng.mu.Lock() + eng.hibernateErr, eng.snapListErr = nil, nil + eng.mu.Unlock() if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { t.Fatalf("Hibernate retry: %v", err) } @@ -247,24 +249,23 @@ func TestResolvePendingSnapReleaseRaceDropsOrphan(t *testing.T) { leaveDanglingIntent(t, m, eng, sb) realSnap := eng.hibernates[0] - eng.snapListErr = nil - eng.snapListStall = make(chan struct{}) callsBefore := eng.snapListCalls() - wakeErr := make(chan error, 1) - go func() { - _, _, _, err := m.WakeAgentSocket(t.Context(), sb.ID, sb.Token) - wakeErr <- err - }() - waitFor(t, func() bool { return eng.snapListCalls() > callsBefore }) + eng.mu.Lock() + eng.snapListErr, eng.snapListStall = nil, make(chan struct{}) + eng.mu.Unlock() + time.Sleep(2 * captureResumeRetry) + synctest.Wait() + if eng.snapListCalls() <= callsBefore { + t.Fatal("the settle never re-read the snapshot list") + } if err := m.Release(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { t.Fatalf("release: %v", err) } close(eng.snapListStall) - - if err := <-wakeErr; !errors.Is(err, ErrUnknownSandbox) { + waitFor(t, func() bool { return eng.snapRemoved(realSnap) }) + if _, _, _, err := m.WakeAgentSocket(t.Context(), sb.ID, sb.Token); !errors.Is(err, ErrUnknownSandbox) { t.Fatalf("wake: %v, want ErrUnknownSandbox", err) } - waitFor(t, func() bool { return eng.snapRemoved(realSnap) }) }) } @@ -275,11 +276,12 @@ func TestResolveAdoptBillsWhenPersistFails(t *testing.T) { sb := mustClaim(t, m, testKey) leaveDanglingIntent(t, m, eng, sb) - eng.snapListErr = nil breakStore(t, m) - if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); err == nil { - t.Fatal("Hibernate reported success despite a persist failure") - } + eng.mu.Lock() + eng.snapListErr = nil + eng.mu.Unlock() + time.Sleep(2 * captureResumeRetry) + synctest.Wait() if n := m.Counters().Hibernates; n != 1 { t.Errorf("hibernates billed=%d, want 1 (billed even while the persist lags)", n) } diff --git a/sandboxd/pool/interrupted.go b/sandboxd/pool/interrupted.go new file mode 100644 index 000000000..5b3c8e572 --- /dev/null +++ b/sandboxd/pool/interrupted.go @@ -0,0 +1,152 @@ +package pool + +import ( + "cmp" + "context" + "errors" + "fmt" + "slices" + "time" + + "github.com/projecteru2/core/log" + + "github.com/cocoonstack/sandbox/sandboxd/engine" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +const ( + captureResumeBudget = 30 * time.Minute + captureResumeRetry = 5 * time.Second +) + +func (m *Manager) settleInterrupted(ctx context.Context, sb *types.Sandbox) { + defer sb.Transition.Unlock() + logger := log.WithFunc("pool.settleInterrupted") + m.mu.Lock() + sb.ReaderSafe = false + snap := cmp.Or(sb.PendingSnap, sb.SaveSnap) + m.mu.Unlock() + logger.Warnf(ctx, "%s has an interrupted capture %s; holding it until the capture exits", sb.ID, snap) + if err := m.settleCapture(ctx, sb); err != nil { + logger.Warnf(ctx, "%s keeps its capture intent for the next start: %v", sb.ID, err) + return + } + logger.Infof(ctx, "%s settled its interrupted capture", sb.ID) +} + +func (m *Manager) settleCapture(ctx context.Context, sb *types.Sandbox) (err error) { + m.mu.Lock() + pending, saved, vmName, sock := sb.PendingSnap, sb.SaveSnap, sb.VMName, sb.VsockSocket + m.mu.Unlock() + if pending == "" && saved == "" { + return nil + } + if err = m.eng.WaitCapture(ctx, cmp.Or(pending, saved)); err != nil { + return err + } + var running bool + var snaps []string + err = retryCapture(ctx, func() (bool, error) { + var readErr error + running, snaps, readErr = m.readCapture(ctx, vmName) + return false, readErr + }) + if err != nil { + return err + } + listed := pending != "" && slices.Contains(snaps, pending) + if !listed && running { + if err = m.resumeGuest(ctx, sb, sock); err != nil { + if ctx.Err() != nil || errors.Is(err, engine.ErrStopping) { + return err + } + m.failLocked(ctx, sb, fmt.Sprintf("capture stopped short: guest did not resume: %v", err)) + } + } + _, err = m.finishCapture(ctx, sb, running && listed, saved, snaps) + return err +} + +func (m *Manager) settleFailedCapture(ctx context.Context, sb *types.Sandbox) (adopted bool, err error) { + m.mu.Lock() + pending, saved, vmName := sb.PendingSnap, sb.SaveSnap, sb.VMName + m.mu.Unlock() + running, snaps, err := m.readCapture(ctx, vmName) + if err == nil && pending != "" && slices.Contains(snaps, pending) { + return m.finishCapture(ctx, sb, running, saved, snaps) + } + go func() { + sb.Transition.Lock() + m.settleInterrupted(context.WithoutCancel(ctx), sb) + }() + return false, err +} + +func (m *Manager) readCapture(ctx context.Context, vmName string) (running bool, snaps []string, err error) { + vms, err := m.eng.List(ctx) + if err != nil { + return false, nil, err + } + snaps, err = m.eng.SnapshotList(ctx) + if err != nil { + return false, nil, err + } + return slices.ContainsFunc(vms, func(vm types.VMRecord) bool { + return vm.Config.Name == vmName && vm.State == vmStateRunning + }), snaps, nil +} + +func (m *Manager) finishCapture(ctx context.Context, sb *types.Sandbox, stop bool, saved string, snaps []string) (adopted bool, err error) { + // A listed hibernate snapshot is the guest as it last ran, so it is the wake image in any VMM state + if stop { + if err = m.eng.Stop(ctx, sb.VMName); err != nil { + return false, err + } + } + adopted, err = m.settlePendingSnap(ctx, sb) + if adopted { + m.out.Disarm(sb.ID, true) + } + if saved != "" { + if _, commitErr := m.commitIfLive(ctx, sb, func() { sb.SaveSnap = "" }); commitErr != nil { + err = errors.Join(err, commitErr) + } + if slices.Contains(snaps, saved) { + m.dropSnap(ctx, saved) + } + } + return adopted, err +} + +func (m *Manager) resumeGuest(ctx context.Context, sb *types.Sandbox, sock string) error { + err := retryCapture(ctx, func() (bool, error) { + startErr := m.eng.Start(ctx, sb.VMName) + if startErr == nil { + return false, nil + } + m.mu.Lock() + live := m.claimed[sb.ID] == sb + m.mu.Unlock() + log.WithFunc("pool.resumeGuest").Warnf(ctx, "%s has not resumed: %v", sb.ID, startErr) + return !live || errors.Is(startErr, engine.ErrStopping), startErr + }) + if err != nil { + return err + } + return m.eng.Probe(ctx, sock, coldProbeTimeout) +} + +func retryCapture(ctx context.Context, try func() (final bool, err error)) error { + deadline := time.Now().Add(captureResumeBudget) + for { + final, err := try() + if err == nil || final || time.Now().After(deadline) { + return err + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(captureResumeRetry): + } + } +} diff --git a/sandboxd/pool/interrupted_test.go b/sandboxd/pool/interrupted_test.go new file mode 100644 index 000000000..631c67f5f --- /dev/null +++ b/sandboxd/pool/interrupted_test.go @@ -0,0 +1,446 @@ +package pool + +import ( + "context" + "errors" + "slices" + "strings" + "testing" + "testing/synctest" + "time" + + "github.com/cocoonstack/sandbox/sandboxd/engine" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +const ( + interruptedVM = "sbx-cut-1" + interruptedHib = hibernatePrefix + "cut-1-abc123" + interruptedSave = forkPrefix + "cut-1-abc123" +) + +func TestReconcileSettlesAnInterruptedCapture(t *testing.T) { + for _, tt := range []struct { + name string + pending, save string + listed []string + wantStarts int + wantStops int + wantHibernate bool + }{ + {name: "hibernate persisted, VMM alive", pending: interruptedHib, listed: []string{interruptedHib}, wantStops: 1, wantHibernate: true}, + {name: "hibernate never persisted", pending: interruptedHib, wantStarts: 1}, + {name: "save never adopts a listed snapshot", save: interruptedSave, listed: []string{interruptedSave}, wantStarts: 1}, + {name: "save never persisted", save: interruptedSave, wantStarts: 1}, + } { + t.Run(tt.name, func(t *testing.T) { + eng := newFakeEngine() + m, sb := reconcileInterrupted(t, eng, tt.pending, tt.save, tt.listed) + waitSettled(t, sb) + if want := []string{tt.pending + tt.save}; !slices.Equal(eng.captureWaits, want) { + t.Errorf("waited on %q, want %q", eng.captureWaits, want) + } + if len(eng.starts) != tt.wantStarts || len(eng.stops) != tt.wantStops { + t.Errorf("starts %q stops %q, want %d and %d", eng.starts, eng.stops, tt.wantStarts, tt.wantStops) + } + if got := sb.HibernateSnap == tt.pending && tt.pending != ""; got != tt.wantHibernate { + t.Errorf("hibernate snap %q, want adopted %v", sb.HibernateSnap, tt.wantHibernate) + } + if sb.Failed != "" { + t.Errorf("failed %q", sb.Failed) + } + if tt.wantHibernate && eng.snapRemoved(tt.pending) { + t.Error("the only copy of guest memory was removed") + } + if tt.wantHibernate && !strings.Contains(usageEventsOf(t, m, sb.ID), "hibernate") { + t.Error("the adopted hibernate was not billed") + } + if row := journaled(t, m)[sb.ID]; row.PendingSnap != "" || row.SaveSnap != "" { + t.Errorf("journal keeps intents %q %q after the settle", row.PendingSnap, row.SaveSnap) + } + }) + } +} + +func TestReconcileAdoptsAHibernateTheVMMLeftInError(t *testing.T) { + eng := newFakeEngine() + eng.vmStates = map[string]string{interruptedVM: "error"} + _, sb := reconcileInterrupted(t, eng, interruptedHib, "", []string{interruptedHib}) + if sb.HibernateSnap != interruptedHib || len(eng.starts) != 0 || eng.snapRemoved(interruptedHib) { + t.Errorf("snap %q starts %q, want the snapshot adopted, kept and no start", sb.HibernateSnap, eng.starts) + } +} + +func TestInterruptedCaptureDoesNothingUntilTheCLIExits(t *testing.T) { + eng := newFakeEngine() + eng.captureStall = make(chan struct{}) + m, sb := reconcileInterrupted(t, eng, interruptedHib, "", nil) + if sb.Transition.TryLock() { + t.Fatal("the claim is free before the capture CLI exited") + } + waitFor(t, func() bool { + eng.mu.Lock() + defer eng.mu.Unlock() + return len(eng.captureWaits) == 1 + }) + eng.mu.Lock() + starts, stops := len(eng.starts), len(eng.stops) + eng.mu.Unlock() + if starts != 0 || stops != 0 { + t.Errorf("%d starts and %d stops while the capture CLI may hold the VM", starts, stops) + } + if row := journaled(t, m)[sb.ID]; row.PendingSnap != interruptedHib { + t.Errorf("journal intent %q, want it kept until settled", row.PendingSnap) + } + close(eng.captureStall) + waitSettled(t, sb) + if len(eng.starts) != 1 { + t.Errorf("starts %q after the CLI exited, want one", eng.starts) + } +} + +func TestInterruptedCaptureNeverStartsAVMThatStoppedMeanwhile(t *testing.T) { + eng := newFakeEngine() + eng.captureHook = func() { + eng.mu.Lock() + defer eng.mu.Unlock() + eng.vmStates[interruptedVM] = "stopped" + } + m, sb := reconcileInterrupted(t, eng, interruptedHib, "", nil) + waitSettled(t, sb) + if len(eng.starts) != 0 { + t.Errorf("starts %q on a stopped VM: a silent cold boot", eng.starts) + } + m.mu.Lock() + defer m.mu.Unlock() + if sb.PendingSnap != "" { + t.Errorf("intent %q survived", sb.PendingSnap) + } +} + +func TestInterruptedCaptureDoesNotStopAVMThatStoppedMeanwhile(t *testing.T) { + eng := newFakeEngine() + eng.captureHook = func() { + eng.mu.Lock() + defer eng.mu.Unlock() + eng.vmStates[interruptedVM] = "stopped" + } + _, sb := reconcileInterrupted(t, eng, interruptedHib, "", []string{interruptedHib}) + waitSettled(t, sb) + if len(eng.stops) != 0 || len(eng.starts) != 0 || sb.HibernateSnap != interruptedHib { + t.Errorf("stops %q starts %q snap %q, want the snapshot adopted and the VM left alone", eng.stops, eng.starts, sb.HibernateSnap) + } +} + +func TestInterruptedCaptureDropsASaveSnapshotListedAfterTheSweep(t *testing.T) { + eng := newFakeEngine() + eng.captureHook = func() { + eng.mu.Lock() + defer eng.mu.Unlock() + eng.snapshots = append(eng.snapshots, interruptedSave) + } + _, sb := reconcileInterrupted(t, eng, "", interruptedSave, nil) + waitSettled(t, sb) + if !eng.snapRemoved(interruptedSave) { + t.Error("the interrupted save left its snapshot behind") + } + if len(eng.starts) != 1 || len(eng.stops) != 0 || sb.HibernateSnap != "" { + t.Errorf("starts %q stops %q snap %q: a save's snapshot is never the wake image", eng.starts, eng.stops, sb.HibernateSnap) + } +} + +func TestInterruptedCaptureKeepsTheIntentWhenStopFails(t *testing.T) { + eng := newFakeEngine() + eng.stopErr = errors.New("ops lock busy") + m, sb := reconcileInterrupted(t, eng, interruptedHib, "", []string{interruptedHib}) + waitSettled(t, sb) + m.mu.Lock() + defer m.mu.Unlock() + if sb.PendingSnap != interruptedHib || sb.HibernateSnap != "" || eng.snapRemoved(interruptedHib) { + t.Errorf("intent %q snap %q, want the intent and its snapshot kept, not adopted", sb.PendingSnap, sb.HibernateSnap) + } +} + +func TestAWakeAfterAFailedSettleResumesTheGuestBeforeClearingTheIntent(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + eng.snapListErr = errors.New("cocoon unreachable") + m, sb := reconcileInterrupted(t, eng, interruptedHib, "", nil) + time.Sleep(captureResumeBudget + 2*captureResumeRetry) + synctest.Wait() + if row := journaled(t, m)[sb.ID]; row.PendingSnap != interruptedHib || len(eng.starts) != 0 { + t.Fatalf("intent %q starts %q, want the intent kept after the settle gave up", row.PendingSnap, eng.starts) + } + eng.mu.Lock() + eng.snapListErr = nil + eng.mu.Unlock() + if _, _, release, err := m.WakeAgentSocket(t.Context(), sb.ID, sb.Token); err != nil { + t.Fatalf("wake: %v", err) + } else { + release() + } + if !slices.Equal(eng.starts, []string{interruptedVM}) { + t.Errorf("starts %q, want the paused guest resumed before the intent goes", eng.starts) + } + if row := journaled(t, m)[sb.ID]; row.PendingSnap != "" { + t.Errorf("intent %q kept after the guest resumed", row.PendingSnap) + } + }) +} + +func TestInterruptedCaptureRetriesWhileTheDumpRuns(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + eng.startFails = 3 + _, sb := reconcileInterrupted(t, eng, interruptedHib, "", nil) + time.Sleep(4 * captureResumeRetry) + synctest.Wait() + if len(eng.starts) != 4 || sb.Failed != "" { + t.Errorf("starts %d failed %q, want 4 and not failed", len(eng.starts), sb.Failed) + } + }) +} + +func TestInterruptedCaptureFailsTheClaim(t *testing.T) { + for _, tt := range []struct { + name string + edit func(*fakeEngine) + }{ + {"past its budget", func(eng *fakeEngine) { eng.startFails = 1 << 30 }}, + {"guest silent after the start", func(eng *fakeEngine) { eng.probeErr = errors.New("never ready") }}, + } { + t.Run(tt.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + tt.edit(eng) + _, sb := reconcileInterrupted(t, eng, interruptedHib, "", nil) + time.Sleep(captureResumeBudget + 2*captureResumeRetry) + synctest.Wait() + if !strings.Contains(sb.Failed, "did not resume") { + t.Errorf("failed %q, want the resume failure", sb.Failed) + } + }) + }) + } +} + +func TestInterruptedCaptureKeepsItsIntentWhenStopping(t *testing.T) { + for _, tt := range []struct { + name string + run func(*testing.T, *fakeEngine, context.CancelFunc) + }{ + {"engine stopping", func(_ *testing.T, eng *fakeEngine, _ context.CancelFunc) { eng.startErr = engine.ErrStopping }}, + {"context canceled mid retry", func(_ *testing.T, eng *fakeEngine, cancel context.CancelFunc) { + eng.startFails = 1 << 30 + eng.startHook = cancel + }}, + } { + t.Run(tt.name, func(t *testing.T) { + eng := newFakeEngine() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + tt.run(t, eng, cancel) + m, sb := reconcileInterruptedCtx(ctx, t, eng, interruptedHib, "", nil) + waitSettled(t, sb) + if row := journaled(t, m)[sb.ID]; row.PendingSnap != interruptedHib || sb.Failed != "" { + t.Errorf("journal intent %q failed %q, want the intent for the next start", row.PendingSnap, sb.Failed) + } + }) + } +} + +func TestAFailedHibernateResumesTheGuest(t *testing.T) { + for _, listFails := range []int{0, 3} { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng) + sb := mustClaim(t, m, testKey) + eng.hibernateErr, eng.listFails = errors.New("signal: killed"), listFails + if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); err == nil { + t.Fatal("Hibernate reported success") + } + time.Sleep(time.Duration(listFails+1) * captureResumeRetry) + synctest.Wait() + if !slices.Equal(eng.starts, []string{sb.VMName}) { + t.Errorf("starts %q, want the guest resumed", eng.starts) + } + }) + } +} + +func TestAFailedCaptureAnswersWithoutWaitingForTheGuest(t *testing.T) { + eng := newFakeEngine() + eng.captureStall = make(chan struct{}) + m := newTestManager(t, eng) + sb := mustClaim(t, m, testKey) + eng.hibernateErr = errors.New("signal: killed") + done := make(chan error, 1) + go func() { done <- m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}) }() + select { + case err := <-done: + if err == nil { + t.Fatal("Hibernate reported success") + } + case <-time.After(3 * time.Second): + t.Fatal("the API waited on a capture that is still exiting") + } + close(eng.captureStall) + waitStarted(t, eng, 1) + waitSettled(t, sb) +} + +func TestAFailedHibernateNeverStartsAStoppedVM(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng) + sb := mustClaim(t, m, testKey) + eng.hibernateErr = errors.New("signal: killed") + eng.captureHook = func() { + eng.mu.Lock() + defer eng.mu.Unlock() + eng.stopped[sb.VMName] = true + } + if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); err == nil { + t.Fatal("Hibernate reported success") + } + waitFor(t, func() bool { + eng.mu.Lock() + defer eng.mu.Unlock() + return len(eng.captureWaits) == 1 + }) + waitSettled(t, sb) + if len(eng.starts) != 0 { + t.Errorf("starts %q on a stopped VM: a silent cold boot", eng.starts) + } +} + +func TestAFailedHibernateThatPersistedStopsTheVMMAndAdopts(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng) + sb := mustClaim(t, m, testKey) + eng.hibernateErr, eng.hibernateErrCompletes, eng.hibernateLeavesVMM = errors.New("signal: killed"), true, true + if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { + t.Fatalf("Hibernate: %v", err) + } + if len(eng.starts) != 0 || !slices.Equal(eng.stops, []string{sb.VMName}) || sb.HibernateSnap == "" { + t.Errorf("starts %q stops %q snap %q, want the VMM stopped, the snapshot adopted and no start", eng.starts, eng.stops, sb.HibernateSnap) + } +} + +func TestAFailedCaptureKeepsItsIntentWhileSandboxdStops(t *testing.T) { + for _, tt := range []struct { + name string + capture func(*Manager, *types.Sandbox, *fakeEngine) error + intent func(*types.Sandbox) string + }{ + {"hibernate", func(m *Manager, sb *types.Sandbox, eng *fakeEngine) error { + eng.hibernateErr = errors.New("signal: killed") + return m.Hibernate(context.Background(), sb.ID, Cred{Token: sb.Token}) + }, func(row *types.Sandbox) string { return row.PendingSnap }}, + {"save", func(m *Manager, sb *types.Sandbox, eng *fakeEngine) error { + eng.snapSaveErr = errors.New("signal: killed") + sb.Transition.Lock() + defer sb.Transition.Unlock() + _, _, err := m.sourceSnap(context.Background(), sb) + return err + }, func(row *types.Sandbox) string { return row.SaveSnap }}, + } { + t.Run(tt.name, func(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng) + sb := mustClaim(t, m, testKey) + eng.startErr = engine.ErrStopping + if err := tt.capture(m, sb, eng); err == nil { + t.Fatal("the capture reported success") + } + waitStarted(t, eng, 1) + waitSettled(t, sb) + if err := m.FlushClaims(); err != nil { + t.Fatalf("flush: %v", err) + } + if tt.intent(journaled(t, m)[sb.ID]) == "" || sb.Failed != "" { + t.Errorf("journal lost the intent or failed the claim: %q", sb.Failed) + } + }) + } +} + +func TestASuccessfulSaveClearsItsIntentOnDisk(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng) + sb := mustClaim(t, m, testKey) + sb.Transition.Lock() + defer sb.Transition.Unlock() + if _, _, err := m.sourceSnap(t.Context(), sb); err != nil { + t.Fatalf("sourceSnap: %v", err) + } + if row := journaled(t, m)[sb.ID]; row.SaveSnap != "" { + t.Errorf("journal keeps save intent %q, so every restart settles a healthy guest", row.SaveSnap) + } +} + +func TestASaveJournalsItsIntentAndAFailureResumesTheGuest(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng) + sb := mustClaim(t, m, testKey) + eng.snapSaveStall = make(chan struct{}) + eng.snapSaveErr = errors.New("signal: killed") + done := make(chan error, 1) + go func() { + sb.Transition.Lock() + defer sb.Transition.Unlock() + _, _, err := m.sourceSnap(t.Context(), sb) + done <- err + }() + waitFor(t, func() bool { return journaled(t, m)[sb.ID].SaveSnap != "" }) + close(eng.snapSaveStall) + if err := <-done; err == nil { + t.Fatal("sourceSnap reported success") + } + waitStarted(t, eng, 1) + waitFor(t, func() bool { return journaled(t, m)[sb.ID].SaveSnap == "" }) +} + +func waitStarted(t *testing.T, eng *fakeEngine, n int) { + t.Helper() + waitFor(t, func() bool { + eng.mu.Lock() + defer eng.mu.Unlock() + return len(eng.starts) >= n + }) +} + +func waitSettled(t *testing.T, sb *types.Sandbox) { + t.Helper() + waitFor(t, func() bool { + if !sb.Transition.TryLock() { + return false + } + sb.Transition.Unlock() + return true + }) +} + +func reconcileInterrupted(t *testing.T, eng *fakeEngine, pending, save string, listed []string) (*Manager, *types.Sandbox) { + t.Helper() + return reconcileInterruptedCtx(t.Context(), t, eng, pending, save, listed) +} + +func reconcileInterruptedCtx(ctx context.Context, t *testing.T, eng *fakeEngine, pending, save string, listed []string) (*Manager, *types.Sandbox) { + t.Helper() + eng.vms[interruptedVM] = "/vsock/cut" + if eng.vmStates == nil { + eng.vmStates = map[string]string{} + } + eng.snapshots = append(eng.snapshots, listed...) + dataDir := t.TempDir() + seedClaims(t, dataDir, map[string]*types.Sandbox{ + "sb_cut": {ID: "sb_cut", VMName: interruptedVM, Key: testKey, Token: "tok", VsockSocket: "/vsock/cut", PendingSnap: pending, SaveSnap: save}, + }) + m := newTestManagerAt(t, eng, dataDir) + if err := m.Reconcile(ctx); err != nil { + t.Fatalf("Reconcile: %v", err) + } + m.mu.Lock() + defer m.mu.Unlock() + return m, m.claimed["sb_cut"] +} diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index e724433e0..9319f26cf 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -155,6 +155,7 @@ type Engine interface { Inspect(ctx context.Context, name string) (types.VMRecord, bool, error) VMEvents(ctx context.Context, socket string, onSync engine.VMSyncFunc, onChange engine.VMChangeFunc) error Probe(ctx context.Context, vsockSocket string, timeout time.Duration) error + WaitCapture(ctx context.Context, snap string) error DialGuestPort(ctx context.Context, vsockSocket string, port uint16) (net.Conn, error) InstallCACert(ctx context.Context, vsockSocket string, certPEM []byte) error MarkLane(ctx context.Context, vsockSocket string, lane types.NetRoute) error diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index b4538843b..33e02f9f3 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -1263,6 +1263,14 @@ type fakeEngine struct { runColdErr error probeErr error hibernateErr error + captureWaits []string + vmStates map[string]string + startFails int + stopErr error + hibernateLeavesVMM bool + captureStall chan struct{} + captureHook func() + listFails int restoreErr error restoreStall chan struct{} snapSaveStall chan struct{} @@ -1418,6 +1426,10 @@ func (f *fakeEngine) Start(_ context.Context, name string) error { hook() } f.starts = append(f.starts, name) + if f.startFails > 0 { + f.startFails-- + return errors.New("query vm.info: context deadline exceeded") + } if f.startErr != nil { return f.startErr } @@ -1432,10 +1444,27 @@ func (f *fakeEngine) Stop(_ context.Context, name string) error { f.mu.Lock() defer f.mu.Unlock() f.stops = append(f.stops, name) + if f.stopErr != nil { + return f.stopErr + } f.stopped[name] = true return nil } +func (f *fakeEngine) WaitCapture(_ context.Context, snap string) error { + f.mu.Lock() + f.captureWaits = append(f.captureWaits, snap) + stall, hook := f.captureStall, f.captureHook + f.mu.Unlock() + if stall != nil { + <-stall + } + if hook != nil { + hook() + } + return nil +} + func (f *fakeEngine) ReconcileStaleCreate(ctx context.Context, name string) (engine.StaleCreateOutcome, error) { if err := ctx.Err(); err != nil { return "", err @@ -1568,7 +1597,7 @@ func (f *fakeEngine) Hibernate(_ context.Context, name, snapName string) error { } if completes { f.mu.Lock() - f.stopped[name] = true + f.stopped[name] = !f.hibernateLeavesVMM f.snapshots = append(f.snapshots, snapName) f.mu.Unlock() } @@ -1928,11 +1957,17 @@ func (f *fakeEngine) MigrateAbort(_ context.Context, vmName, series string) erro func (f *fakeEngine) list(_ context.Context) ([]types.VMRecord, error) { f.mu.Lock() defer f.mu.Unlock() + if f.listFails > 0 { + f.listFails-- + return nil, errors.New("cocoon vm list: signal: killed") + } var vms []types.VMRecord for name, sock := range f.vms { sock = f.lateVsock(sock) state := vmStateRunning switch { + case f.vmStates[name] != "": + state = f.vmStates[name] case f.creating[name]: state = vmStateCreating case f.stopped[name]: diff --git a/sandboxd/pool/reconcile.go b/sandboxd/pool/reconcile.go index 2d78af08e..0ea84f961 100644 --- a/sandboxd/pool/reconcile.go +++ b/sandboxd/pool/reconcile.go @@ -39,11 +39,14 @@ func (m *Manager) Reconcile(ctx context.Context) error { owned := map[string]bool{} referenced := map[string]bool{} - var adopted, dropped, resumed []*types.Sandbox + var adopted, dropped, resumed, interrupted []*types.Sandbox m.mu.Lock() for _, sb := range claims { sb.SetEnv(sb.Env) rec, ok := live[sb.VMName] + if !ok || rec.State != vmStateRunning { + sb.SaveSnap = "" + } switch { case sb.ArchiveCk != "": // Archived: no local VM by design, so adopt the stub and let the first exec wake it @@ -51,11 +54,14 @@ func (m *Manager) Reconcile(ctx context.Context) error { continue case ok && rec.State == vmStateRunning: sb.VsockSocket = rec.VsockSocket - // Running with either flag set means a crashed wake or an intent the engine never saw if sb.HibernateSnap != "" { resumed = append(resumed, sb) } - sb.HibernateSnap, sb.PendingSnap = "", "" + sb.HibernateSnap = "" + if sb.PendingSnap != "" || sb.SaveSnap != "" { + referenced[sb.PendingSnap] = true + interrupted = append(interrupted, sb) + } case ok && sb.HibernateSnap != "": // Hibernated: the VM is stopped by design and wakes on demand. sb.PendingSnap = "" @@ -97,6 +103,32 @@ func (m *Manager) Reconcile(ctx context.Context) error { if err := m.tpls.SweepStaging(); err != nil { logger.Error(ctx, err, "sweep template staging") } + m.sweepOrphanGoldens(ctx, pooled) + // A crash mid-export leaves a fork-* staging dir nothing in this life reuses + _ = utils.RemoveDirEntries(m.dataDir, func(name string) bool { return strings.HasPrefix(name, "fork-") }) + _ = utils.RemoveDirEntries(filepath.Join(m.dataDir, readerDir), nil) + _ = utils.RemoveDirEntries(filepath.Join(m.dataDir, movesDir), func(id string) bool { return claims[id] == nil || !claims[id].MoveIn }) + + if fenced { + m.reclaimOrphanArchiveCks(ctx, claims) + } + m.retryArchiveDeletes(ctx) + removed := m.sweepStaleVMs(ctx, live, owned) + + m.sweepOrphanSnaps(ctx, snaps, snapsErr, referenced) + m.resyncEgress(ctx, live, removed) + m.warnDeniedClaims(ctx) + logger.Infof(ctx, "adopted %d claims, %d VMs live", len(m.claimed), len(live)) + m.resumeMoves(context.WithoutCancel(ctx), outs, ins) + for _, sb := range interrupted { + sb.Transition.Lock() + go m.settleInterrupted(ctx, sb) + } + return saveErr +} + +func (m *Manager) sweepOrphanGoldens(ctx context.Context, pooled map[string]bool) { + logger := log.WithFunc("pool.Reconcile") dir := m.goldensDir() goldens, _ := os.ReadDir(dir) for _, e := range goldens { @@ -108,17 +140,10 @@ func (m *Manager) Reconcile(ctx context.Context) error { logger.Infof(ctx, "removed orphan golden %s", name) } } - // A crash mid-export leaves a fork-* staging dir nothing in this life reuses - _ = utils.RemoveDirEntries(m.dataDir, func(name string) bool { return strings.HasPrefix(name, "fork-") }) - _ = utils.RemoveDirEntries(filepath.Join(m.dataDir, readerDir), nil) - _ = utils.RemoveDirEntries(filepath.Join(m.dataDir, movesDir), func(id string) bool { return claims[id] == nil || !claims[id].MoveIn }) - - if fenced { - m.reclaimOrphanArchiveCks(ctx, claims) - } - m.retryArchiveDeletes(ctx) - removed := m.sweepStaleVMs(ctx, live, owned) +} +func (m *Manager) sweepOrphanSnaps(ctx context.Context, snaps []string, snapsErr error, referenced map[string]bool) { + logger := log.WithFunc("pool.Reconcile") // An unreferenced hibernate snapshot is an orphan; fork and golden snapshots never span a restart if snapsErr != nil { logger.Warnf(ctx, "snapshot sweep skipped: %v", snapsErr) @@ -134,11 +159,6 @@ func (m *Manager) Reconcile(ctx context.Context) error { logger.Infof(ctx, "removed orphan snapshot %s", orphans[i]) }).Wait() } - m.resyncEgress(ctx, live, removed) - m.warnDeniedClaims(ctx) - logger.Infof(ctx, "adopted %d claims, %d VMs live", len(m.claimed), len(live)) - m.resumeMoves(context.WithoutCancel(ctx), outs, ins) - return saveErr } // pendingMovesLocked marks the journaled moves of loaded claims: a source keeps its claim gated until the record resolves; callers hold m.mu. diff --git a/sandboxd/types/types.go b/sandboxd/types/types.go index 69372004f..c86cddb3c 100644 --- a/sandboxd/types/types.go +++ b/sandboxd/types/types.go @@ -284,6 +284,7 @@ type Sandbox struct { HibernateSnap string `json:"hibernate_snap,omitempty"` // PendingSnap is the journaled intent of a hibernate in flight, cleared by its commit. PendingSnap string `json:"pending_snap,omitempty"` + SaveSnap string `json:"save_snap,omitempty"` ReaderSafe bool `json:"reader_safe,omitzero"` // ArchiveCk names the store checkpoint holding an archived sandbox; empty means local. ArchiveCk string `json:"archive_ck,omitempty"`