diff --git a/cmd/api/main.go b/cmd/api/main.go index 320a3adf7..4b4480569 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -365,6 +365,12 @@ func run() error { return fmt.Errorf("reconcile device state: %w", err) } + // A fork that a crash cut short still holds its VMM, memory and vGPU. + // Delete those before the vGPU reconciler counts their claims. + if r, ok := app.InstanceManager.(interface{ RecoverInterruptedForks(context.Context) }); ok { + r.RecoverInterruptedForks(app.Ctx) + } + logger.Info("Reconciling vGPU devices...") if r, ok := app.InstanceManager.(interface{ StartVGPUReconciler(context.Context) }); ok { r.StartVGPUReconciler(ctx) diff --git a/lib/instances/fork.go b/lib/instances/fork.go index 94e2a2132..fcabde435 100644 --- a/lib/instances/fork.go +++ b/lib/instances/fork.go @@ -7,6 +7,7 @@ import ( "fmt" "hash/crc32" "os" + "path/filepath" "strings" "time" @@ -364,6 +365,7 @@ func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id strin restoreNeedsSourceLock = prepareResult.RequiresSnapshotSourceAlias } + forkMeta.ForkPending = currentForkOwner() newMeta := &metadata{StoredMetadata: forkMeta} if err := m.saveForkMetadata(ctx, newMeta); err != nil { return nil, false, err @@ -561,6 +563,84 @@ func forkReturnNeedsGuestAgentReady(inst *Instance, guestReady bool) bool { return inst != nil && (inst.State == StateRunning || inst.State == StateInitializing) && !guestReady } +// completeFork clears the pending mark of a fork that reached its target state. +func (m *manager) completeFork(inst *Instance) error { + lock := m.getInstanceLock(inst.Id) + lock.Lock() + defer lock.Unlock() + + meta, err := m.loadMetadata(inst.Id) + if err != nil { + return fmt.Errorf("load completed fork metadata: %w", err) + } + if meta.ForkPending != nil { + meta.ForkPending = nil + if err := m.saveMetadata(meta); err != nil { + return fmt.Errorf("save completed fork metadata: %w", err) + } + } + inst.ForkPending = nil + return nil +} + +// ForkOwner identifies the API process that is making a fork, across PID reuse +// and host reboots. +type ForkOwner struct { + PID int `json:"pid"` + StartTime uint64 `json:"start_time,omitempty"` // from /proc//stat; 0 = unknown + BootID string `json:"boot_id,omitempty"` // Linux boot ID; empty = unknown +} + +func currentForkOwner() *ForkOwner { + pid := os.Getpid() + return &ForkOwner{PID: pid, StartTime: processStartTime(pid), BootID: hostBootID()} +} + +// gone reports whether the process that was making the fork no longer exists. +// When that cannot be told the owner counts as running, so a fork that another +// live API process is still working on is never deleted. +func (o ForkOwner) gone() bool { + if bootID := hostBootID(); o.BootID != "" && bootID != "" && o.BootID != bootID { + return true + } + if !ProcessExists(o.PID) { + return true + } + // The PID is in use. A different start time means it was recycled. + startTime := processStartTime(o.PID) + return o.StartTime != 0 && startTime != 0 && o.StartTime != startTime +} + +// RecoverInterruptedForks deletes the forks whose API process died before they +// reached their target state. Nothing else does: such a fork looks like an +// ordinary standby or paused instance, and a restore that was cut short keeps +// its VMM, its memory and its vGPU. Call it once at startup, before requests +// are served. +func (m *manager) RecoverInterruptedForks(ctx context.Context) { + log := logger.FromContext(ctx) + files, err := m.listMetadataFiles() + if err != nil { + log.ErrorContext(ctx, "failed to list instances while recovering interrupted forks", "error", err) + return + } + for _, file := range files { + id := filepath.Base(filepath.Dir(file)) + meta, err := m.loadMetadata(id) + if err != nil || meta.ForkPending == nil { + continue + } + if !meta.ForkPending.gone() { + log.WarnContext(ctx, "not deleting pending fork: the process making it is still running", "instance_id", id, "instance_name", meta.Name, "owner_pid", meta.ForkPending.PID) + continue + } + if err := m.cleanupForkInstanceOnError(ctx, id); err != nil { + log.ErrorContext(ctx, "failed to delete interrupted fork; it stays until it is deleted", "instance_id", id, "instance_name", meta.Name, "error", err) + continue + } + log.WarnContext(ctx, "deleted fork interrupted by a restart", "instance_id", id, "instance_name", meta.Name) + } +} + func (m *manager) cleanupForkInstanceOnError(ctx context.Context, forkID string) error { lock := m.getInstanceLock(forkID) lock.Lock() @@ -666,5 +746,8 @@ func cloneStoredMetadataWithoutPendingStandbyCompression(src StoredMetadata) Sto func cloneStoredMetadataForSnapshot(src StoredMetadata) StoredMetadata { dst := cloneStoredMetadataWithoutPendingStandbyCompression(src) dst.ExpiresAt = nil + // The mark belongs to the fork request, not to what is later made from + // a snapshot of its instance. + dst.ForkPending = nil return dst } diff --git a/lib/instances/fork_pending_linux_test.go b/lib/instances/fork_pending_linux_test.go new file mode 100644 index 000000000..856d3965d --- /dev/null +++ b/lib/instances/fork_pending_linux_test.go @@ -0,0 +1,26 @@ +//go:build linux + +package instances + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// A PID alone does not identify the process that was making a fork: it is +// handed out again after the process exits, and after a reboot. +func TestForkOwnerIsGoneWhenItsPIDBelongsToAnotherProcess(t *testing.T) { + current := *currentForkOwner() + require.NotZero(t, current.StartTime) + require.NotEmpty(t, current.BootID) + + recycled := current + recycled.StartTime++ + assert.True(t, recycled.gone(), "same PID, started at another time") + + beforeReboot := current + beforeReboot.BootID = "00000000-0000-0000-0000-000000000000" + assert.True(t, beforeReboot.gone(), "same PID and start time, recorded in an earlier boot") +} diff --git a/lib/instances/fork_pending_test.go b/lib/instances/fork_pending_test.go new file mode 100644 index 000000000..c0e6bc5f8 --- /dev/null +++ b/lib/instances/fork_pending_test.go @@ -0,0 +1,109 @@ +package instances + +import ( + "context" + "os" + "os/exec" + "testing" + "time" + + "github.com/kernel/hypeman/lib/hypervisor" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// exitedProcessOwner names a process that has already exited, the way a fork +// names the API process that crashed while making it. +func exitedProcessOwner(t *testing.T) *ForkOwner { + t.Helper() + process := exec.Command("true") + require.NoError(t, process.Start()) + owner := &ForkOwner{PID: process.Process.Pid, StartTime: processStartTime(process.Process.Pid), BootID: hostBootID()} + require.NoError(t, process.Wait()) + return owner +} + +func TestForkIsPendingOnlyUntilItsRequestSucceeds(t *testing.T) { + mgr, _ := setupTestManager(t) + ctx := context.Background() + + sourceID := "fork-pending-source" + createStoppedSnapshotSourceFixture(t, mgr, sourceID, sourceID, mgr.defaultHypervisor) + + // What a crash between saving the fork and finishing it leaves behind. + interrupted, _, err := mgr.forkInstanceFromStoppedOrStandby(ctx, sourceID, ForkInstanceRequest{Name: "fork-pending-interrupted"}, false) + require.NoError(t, err) + stored, err := mgr.loadMetadata(interrupted.Id) + require.NoError(t, err) + require.NotNil(t, stored.ForkPending) + assert.Equal(t, os.Getpid(), stored.ForkPending.PID, "the mark must name the process making the fork") + + // A snapshot of it must not pass the mark on to what is restored from it. + ofInterrupted, err := mgr.CreateSnapshot(ctx, interrupted.Id, CreateSnapshotRequest{Kind: SnapshotKindStopped, Name: "fork-pending-of-interrupted"}) + require.NoError(t, err) + record, err := mgr.loadSnapshotRecord(ofInterrupted.Id) + require.NoError(t, err) + assert.Nil(t, record.StoredMetadata.ForkPending) + + fromInstance, err := mgr.ForkInstance(ctx, sourceID, ForkInstanceRequest{Name: "fork-pending-from-instance"}) + require.NoError(t, err) + snapshot, err := mgr.CreateSnapshot(ctx, sourceID, CreateSnapshotRequest{Kind: SnapshotKindStopped, Name: "fork-pending-snapshot"}) + require.NoError(t, err) + fromSnapshot, err := mgr.ForkSnapshot(ctx, snapshot.Id, ForkSnapshotRequest{Name: "fork-pending-from-snapshot", TargetState: StateStopped}) + require.NoError(t, err) + + for _, fork := range []*Instance{fromInstance, fromSnapshot} { + assert.Nil(t, fork.ForkPending, fork.Name) + stored, err := mgr.loadMetadata(fork.Id) + require.NoError(t, err) + assert.Nil(t, stored.ForkPending, fork.Name) + } +} + +func TestRecoverInterruptedForksDeletesOnlyForksWhoseProcessIsGone(t *testing.T) { + mgr, _ := setupTestManager(t) + ctx := context.Background() + + save := func(id string, pending *ForkOwner) { + require.NoError(t, mgr.ensureDirectories(id)) + now := time.Now() + require.NoError(t, mgr.saveMetadata(&metadata{StoredMetadata: StoredMetadata{ + Id: id, + Name: id, + Image: integrationTestImageRef(t, "docker.io/library/alpine:latest"), + CreatedAt: now, + StoppedAt: &now, + HypervisorType: hypervisor.TypeCloudHypervisor, + HypervisorVersion: "test", + SocketPath: mgr.paths.InstanceSocket(id, "cloud-hypervisor.sock"), + DataDir: mgr.paths.InstanceDir(id), + VsockCID: generateVsockCID(id), + VsockSocket: mgr.paths.InstanceVsockSocket(id), + ForkPending: pending, + }})) + } + save("fork-recover-interrupted", exitedProcessOwner(t)) + // Another API process on the same data directory is still making this one. + save("fork-recover-in-flight", currentForkOwner()) + save("fork-recover-finished", nil) + + mgr.RecoverInterruptedForks(ctx) + + _, err := mgr.loadMetadata("fork-recover-interrupted") + require.ErrorIs(t, err, ErrNotFound) + assert.NoDirExists(t, mgr.paths.InstanceDir("fork-recover-interrupted")) + + inFlight, err := mgr.loadMetadata("fork-recover-in-flight") + require.NoError(t, err, "a fork whose process is still running must be left alone") + assert.NotNil(t, inFlight.ForkPending) + + _, err = mgr.loadMetadata("fork-recover-finished") + require.NoError(t, err) +} + +func TestForkOwnerIsGoneOnlyWhenItsProcessIs(t *testing.T) { + assert.False(t, currentForkOwner().gone(), "a running process") + assert.True(t, exitedProcessOwner(t).gone(), "an exited process") + // Without a start time or boot ID a live PID is all there is to go by. + assert.False(t, ForkOwner{PID: os.Getpid()}.gone(), "a live PID with nothing else recorded") +} diff --git a/lib/instances/manager.go b/lib/instances/manager.go index ed85f22f4..205a7fc79 100644 --- a/lib/instances/manager.go +++ b/lib/instances/manager.go @@ -556,12 +556,15 @@ func (m *manager) ForkInstance(ctx context.Context, id string, req ForkInstanceR } else { inst, err = m.applyForkTargetState(ctx, forked.Id, targetState) } - if err != nil { - if cleanupErr := m.cleanupForkInstanceOnError(ctx, forked.Id); cleanupErr != nil { - return nil, fmt.Errorf("apply fork target state: %w; additionally failed to cleanup forked instance %s: %v", err, forked.Id, cleanupErr) - } - return nil, fmt.Errorf("apply fork target state: %w", err) + } + if err == nil { + err = m.completeFork(inst) + } + if err != nil { + if cleanupErr := m.cleanupForkInstanceOnError(ctx, forked.Id); cleanupErr != nil { + return nil, fmt.Errorf("apply fork target state: %w; additionally failed to cleanup forked instance %s: %v", err, forked.Id, cleanupErr) } + return nil, fmt.Errorf("apply fork target state: %w", err) } m.notifyLifecycleEvent(ctx, LifecycleEventFork, inst) return inst, nil diff --git a/lib/instances/snapshot.go b/lib/instances/snapshot.go index dd6f6aead..d8bb9327a 100644 --- a/lib/instances/snapshot.go +++ b/lib/instances/snapshot.go @@ -558,6 +558,7 @@ func (m *manager) forkSnapshot(ctx context.Context, snapshotID string, req ForkS } } + forkMeta.ForkPending = currentForkOwner() newMeta := &metadata{StoredMetadata: forkMeta} if forkMeta.MacOS != nil { if err := m.saveMacOSSnapshotFork(ctx, newMeta); err != nil { @@ -569,6 +570,9 @@ func (m *manager) forkSnapshot(ctx context.Context, snapshotID string, req ForkS cu.Release() inst, err := m.applyForkTargetState(ctx, forkID, targetState) + if err == nil { + err = m.completeFork(inst) + } if err != nil { if cleanupErr := m.cleanupForkInstanceOnError(ctx, forkID); cleanupErr != nil { return nil, fmt.Errorf("apply snapshot fork target state: %w; additionally failed to cleanup forked instance %s: %v", err, forkID, cleanupErr) diff --git a/lib/instances/types.go b/lib/instances/types.go index 9c16e6e68..9acd02d15 100644 --- a/lib/instances/types.go +++ b/lib/instances/types.go @@ -97,6 +97,13 @@ type StoredMetadata struct { // IdentityRekeyPending marks a macOS clone whose SSH host keys still come from its // source. The next start rekeys them and clears the flag. IdentityRekeyPending bool `json:"identity_rekey_pending,omitempty"` + // ForkPending marks a fork that has not yet reached the state its caller + // asked for, and names the API process that is bringing it there. It is + // set when the fork's metadata is first saved and cleared when the fork + // request succeeds. A fork still marked once that process is gone was cut + // short by a crash or kill, so its caller never got an answer; the next + // start deletes it instead of leaving it behind holding its resources. + ForkPending *ForkOwner `json:"fork_pending,omitempty"` // Resources (matching Cloud Hypervisor terminology) Size int64 // Base memory in bytes