diff --git a/cmd/api/api/instances.go b/cmd/api/api/instances.go index 2c85284b..9a93064b 100644 --- a/cmd/api/api/instances.go +++ b/cmd/api/api/instances.go @@ -620,6 +620,11 @@ func (s *ApiService) RestoreInstance(ctx context.Context, request oapi.RestoreIn Code: "invalid_state", Message: err.Error(), }, nil + case errors.Is(err, instances.ErrInsufficientResources): + return oapi.RestoreInstance409JSONResponse{ + Code: "insufficient_resources", + Message: err.Error(), + }, nil default: log.ErrorContext(ctx, "failed to restore instance", "error", err) return oapi.RestoreInstance500JSONResponse{ @@ -683,6 +688,11 @@ func (s *ApiService) ForkInstance(ctx context.Context, request oapi.ForkInstance Code: "name_conflict", Message: err.Error(), }, nil + case errors.Is(err, instances.ErrInsufficientResources): + return oapi.ForkInstance409JSONResponse{ + Code: "insufficient_resources", + Message: err.Error(), + }, nil case errors.Is(err, instances.ErrNotSupported): return oapi.ForkInstance501JSONResponse{ Code: "not_supported", diff --git a/cmd/api/api/instances_test.go b/cmd/api/api/instances_test.go index 4e7863aa..ecb8ec99 100644 --- a/cmd/api/api/instances_test.go +++ b/cmd/api/api/instances_test.go @@ -1261,6 +1261,43 @@ func TestForkInstance_InvalidRequest(t *testing.T) { assert.Equal(t, "invalid_request", badReq.Code) } +func TestForkInstance_InsufficientResources(t *testing.T) { + t.Parallel() + svc := newTestService(t) + + source := instances.Instance{ + StoredMetadata: instances.StoredMetadata{ + Id: "src-instance", + Name: "src-instance", + Image: "docker.io/library/alpine:latest", + CreatedAt: time.Now(), + HypervisorType: hypervisor.TypeFirecracker, + }, + State: instances.StateStandby, + } + + mockMgr := &captureForkManager{ + Manager: svc.InstanceManager, + err: fmt.Errorf("apply fork target state: %w: insufficient network bandwidth", instances.ErrInsufficientResources), + } + svc.InstanceManager = mockMgr + + resp, err := svc.ForkInstance( + mw.WithResolvedInstance(ctx(), source.Id, source), + oapi.ForkInstanceRequestObject{ + Id: source.Id, + Body: &oapi.ForkInstanceRequest{ + Name: "forked-instance", + }, + }, + ) + require.NoError(t, err) + + conflict, ok := resp.(oapi.ForkInstance409JSONResponse) + require.True(t, ok, "expected 409 response") + assert.Equal(t, "insufficient_resources", conflict.Code) +} + func TestStandbyInstance_InvalidRequest(t *testing.T) { t.Parallel() svc := newTestService(t) diff --git a/cmd/api/api/snapshots.go b/cmd/api/api/snapshots.go index 620d9574..e1b9a460 100644 --- a/cmd/api/api/snapshots.go +++ b/cmd/api/api/snapshots.go @@ -193,6 +193,8 @@ func (s *ApiService) ForkSnapshot(ctx context.Context, request oapi.ForkSnapshot return oapi.ForkSnapshot400JSONResponse{Code: "invalid_request", Message: err.Error()}, nil case errors.Is(err, instances.ErrInvalidState), errors.Is(err, instances.ErrAlreadyExists), errors.Is(err, network.ErrNameExists): return oapi.ForkSnapshot409JSONResponse{Code: "conflict", Message: err.Error()}, nil + case errors.Is(err, instances.ErrInsufficientResources): + return oapi.ForkSnapshot409JSONResponse{Code: "insufficient_resources", Message: err.Error()}, nil case errors.Is(err, instances.ErrNotSupported): return oapi.ForkSnapshot501JSONResponse{Code: "not_supported", Message: err.Error()}, nil default: diff --git a/cmd/api/config/config.go b/cmd/api/config/config.go index e315c7bb..8690ad52 100644 --- a/cmd/api/config/config.go +++ b/cmd/api/config/config.go @@ -198,6 +198,7 @@ type HypervisorConfig struct { FirecrackerBinaryPath string `koanf:"firecracker_binary_path"` FirecrackerSnapshotMemoryBackend string `koanf:"firecracker_snapshot_memory_backend"` FirecrackerUFFDCacheMaxBytes string `koanf:"firecracker_uffd_cache_max_bytes"` + FirecrackerMaxConcurrentRestores int `koanf:"firecracker_max_concurrent_restores"` Memory HypervisorMemoryConfig `koanf:"memory"` } @@ -411,6 +412,7 @@ func defaultConfig() *Config { FirecrackerBinaryPath: "", FirecrackerSnapshotMemoryBackend: "file", FirecrackerUFFDCacheMaxBytes: "4294967296", + FirecrackerMaxConcurrentRestores: 32, Memory: HypervisorMemoryConfig{ Enabled: false, KernelPageInitMode: "hardened", @@ -573,6 +575,9 @@ func (c *Config) Validate() error { if c.Build.Timeout <= 0 { return fmt.Errorf("build.timeout must be positive, got %d", c.Build.Timeout) } + if c.Hypervisor.FirecrackerMaxConcurrentRestores < 0 { + return fmt.Errorf("hypervisor.firecracker_max_concurrent_restores must be >= 0, got %d", c.Hypervisor.FirecrackerMaxConcurrentRestores) + } if c.Instances.LifecycleEventBufferSize <= 0 { return fmt.Errorf("instances.lifecycle_event_buffer_size must be positive, got %d", c.Instances.LifecycleEventBufferSize) } diff --git a/cmd/api/config/config_test.go b/cmd/api/config/config_test.go index f627c628..59e64dc6 100644 --- a/cmd/api/config/config_test.go +++ b/cmd/api/config/config_test.go @@ -61,6 +61,9 @@ func TestDefaultConfigIncludesMetricsSettings(t *testing.T) { if cfg.Hypervisor.FirecrackerUFFDCacheMaxBytes != "4294967296" { t.Fatalf("expected default firecracker uffd cache size to be 4294967296, got %q", cfg.Hypervisor.FirecrackerUFFDCacheMaxBytes) } + if cfg.Hypervisor.FirecrackerMaxConcurrentRestores != 32 { + t.Fatalf("expected default firecracker max concurrent restores to be 32, got %d", cfg.Hypervisor.FirecrackerMaxConcurrentRestores) + } } func TestValidateFirecrackerSnapshotMemoryBackend(t *testing.T) { @@ -86,6 +89,14 @@ func TestValidateFirecrackerSnapshotMemoryBackend(t *testing.T) { } } +func TestValidateFirecrackerMaxConcurrentRestores(t *testing.T) { + cfg := defaultConfig() + cfg.Hypervisor.FirecrackerMaxConcurrentRestores = -1 + if err := cfg.Validate(); err == nil { + t.Fatalf("expected invalid firecracker max concurrent restores validation error") + } +} + func TestLoadEnvOverridesMetricsAndOtelInterval(t *testing.T) { t.Setenv("METRICS__LISTEN_ADDRESS", "0.0.0.0") t.Setenv("METRICS__PORT", "9999") diff --git a/config.example.yaml b/config.example.yaml index fd88ac17..e57733b3 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -26,6 +26,8 @@ data_dir: /var/lib/hypeman # default: cloud-hypervisor # # Optional: use a custom Firecracker binary path instead of the embedded one. # # firecracker_binary_path: /usr/local/bin/firecracker +# # Caps concurrent Firecracker snapshot restore/resume work to avoid host overload. +# # firecracker_max_concurrent_restores: 32 # memory: # enabled: false # # performance: init_on_alloc=0 init_on_free=0 (better density) diff --git a/lib/hypervisor/firecracker/firecracker.go b/lib/hypervisor/firecracker/firecracker.go index e3012aad..3512f653 100644 --- a/lib/hypervisor/firecracker/firecracker.go +++ b/lib/hypervisor/firecracker/firecracker.go @@ -56,15 +56,16 @@ func (f *Firecracker) Capabilities() hypervisor.Capabilities { func capabilities() hypervisor.Capabilities { return hypervisor.Capabilities{ - SupportsSnapshot: true, - SupportsHotplugMemory: false, - SupportsBalloonControl: true, - SupportsPause: true, - SupportsVsock: true, - SupportsGPUPassthrough: false, - SupportsDiskIOLimit: true, - SupportsGracefulVMMShutdown: false, - SupportsSnapshotBaseReuse: true, + SupportsSnapshot: true, + SupportsHotplugMemory: false, + SupportsBalloonControl: true, + SupportsPause: true, + SupportsVsock: true, + SupportsGPUPassthrough: false, + SupportsDiskIOLimit: true, + SupportsGracefulVMMShutdown: false, + SupportsSnapshotBaseReuse: true, + SupportsConcurrentForkPrepare: true, } } diff --git a/lib/hypervisor/firecracker/fork.go b/lib/hypervisor/firecracker/fork.go index 67a70293..673b887e 100644 --- a/lib/hypervisor/firecracker/fork.go +++ b/lib/hypervisor/firecracker/fork.go @@ -2,8 +2,6 @@ package firecracker import ( "context" - "fmt" - "os" "path/filepath" "github.com/kernel/hypeman/lib/hypervisor" @@ -33,6 +31,7 @@ func (s *Starter) PrepareFork(ctx context.Context, req hypervisor.ForkPrepareReq } changed := false + result := hypervisor.ForkPrepareResult{} if req.Network != nil && req.Network.TAPDevice != "" { if len(meta.NetworkOverrides) == 0 { meta.NetworkOverrides = []networkOverride{{ @@ -50,27 +49,12 @@ func (s *Starter) PrepareFork(ctx context.Context, req hypervisor.ForkPrepareReq } } if req.SourceDataDir != "" && req.TargetDataDir != "" && req.SourceDataDir != req.TargetDataDir { - if meta.RetainSnapshotSourceDataDirAlias && meta.SnapshotSourceDataDir != "" { - // Keep the upstream source path for snapshot-derived forks. The retained - // Firecracker base can still reference that path after later diff snapshots. - } else { - retainAlias := false - if _, err := os.Stat(req.SourceDataDir); err != nil { - if os.IsNotExist(err) { - retainAlias = true - } else { - return hypervisor.ForkPrepareResult{}, fmt.Errorf("stat snapshot source data dir %q: %w", req.SourceDataDir, err) - } - } - if meta.SnapshotSourceDataDir != req.SourceDataDir { - meta.SnapshotSourceDataDir = req.SourceDataDir - changed = true - } - if meta.RetainSnapshotSourceDataDirAlias != retainAlias { - meta.RetainSnapshotSourceDataDirAlias = retainAlias - changed = true - } + updated, err := prepareSnapshotDataDirForFork(req.SnapshotConfigPath, meta, req.SourceDataDir, req.TargetDataDir) + if err != nil { + return hypervisor.ForkPrepareResult{}, err } + changed = changed || updated + result.RequiresSnapshotSourceAlias = needsSnapshotSourceDirAlias(meta, req.TargetDataDir) } if changed { @@ -79,5 +63,5 @@ func (s *Starter) PrepareFork(ctx context.Context, req hypervisor.ForkPrepareReq } } - return hypervisor.ForkPrepareResult{}, nil + return result, nil } diff --git a/lib/hypervisor/firecracker/fork_test.go b/lib/hypervisor/firecracker/fork_test.go index 109ae7e7..5ffe5617 100644 --- a/lib/hypervisor/firecracker/fork_test.go +++ b/lib/hypervisor/firecracker/fork_test.go @@ -2,6 +2,7 @@ package firecracker import ( "context" + "encoding/binary" "os" "path/filepath" "testing" @@ -65,6 +66,74 @@ func TestPrepareFork_DoesNotRetainExistingSourceAlias(t *testing.T) { assert.False(t, meta.RetainSnapshotSourceDataDirAlias) } +func TestPrepareFork_RewritesSnapshotStatePathsWithoutAlias(t *testing.T) { + starter := NewStarter() + tmp := t.TempDir() + sourceDir := filepath.Join(tmp, "source-12345678901234567890") + targetDir := filepath.Join(tmp, "target-12345678901234567890") + snapshotDir := filepath.Join(targetDir, "snapshots", "snapshot-latest") + require.NoError(t, os.MkdirAll(sourceDir, 0755)) + require.NoError(t, os.MkdirAll(snapshotDir, 0755)) + require.NoError(t, saveRestoreMetadata(targetDir, []networkInterface{{IfaceID: "eth0", HostDevName: "tap-old"}})) + writeSnapshotStateForTest(t, snapshotStatePath(snapshotDir), []byte("drive="+filepath.Join(sourceDir, "overlay.raw")+" vsock="+filepath.Join(sourceDir, "vsock.sock"))) + + result, err := starter.PrepareFork(context.Background(), hypervisor.ForkPrepareRequest{ + SnapshotConfigPath: filepath.Join(snapshotDir, "config.json"), + SourceDataDir: sourceDir, + TargetDataDir: targetDir, + Network: &hypervisor.ForkNetworkConfig{ + TAPDevice: "tap-new", + }, + }) + require.NoError(t, err) + assert.False(t, result.RequiresSnapshotSourceAlias) + + meta, err := loadRestoreMetadata(targetDir) + require.NoError(t, err) + require.Len(t, meta.NetworkOverrides, 1) + assert.Equal(t, "tap-new", meta.NetworkOverrides[0].HostDevName) + assert.Empty(t, meta.SnapshotSourceDataDir) + assert.False(t, meta.RetainSnapshotSourceDataDirAlias) + + data, err := os.ReadFile(snapshotStatePath(snapshotDir)) + require.NoError(t, err) + assert.Equal(t, uint64(0), firecrackerSnapshotCRC64(data)) + assert.NotContains(t, string(data), sourceDir) + assert.Contains(t, string(data), filepath.Join(targetDir, "overlay.raw")) + assert.Contains(t, string(data), filepath.Join(targetDir, "vsock.sock")) +} + +func TestPrepareFork_FallsBackToAliasWhenSnapshotStatePathLengthDiffers(t *testing.T) { + starter := NewStarter() + tmp := t.TempDir() + sourceDir := filepath.Join(tmp, "source-short") + targetDir := filepath.Join(tmp, "target-much-longer") + snapshotDir := filepath.Join(targetDir, "snapshots", "snapshot-latest") + require.NoError(t, os.MkdirAll(sourceDir, 0755)) + require.NoError(t, os.MkdirAll(snapshotDir, 0755)) + require.NoError(t, saveRestoreMetadata(targetDir, nil)) + writeSnapshotStateForTest(t, snapshotStatePath(snapshotDir), []byte("drive="+filepath.Join(sourceDir, "overlay.raw"))) + + result, err := starter.PrepareFork(context.Background(), hypervisor.ForkPrepareRequest{ + SnapshotConfigPath: filepath.Join(snapshotDir, "config.json"), + SourceDataDir: sourceDir, + TargetDataDir: targetDir, + }) + require.NoError(t, err) + assert.True(t, result.RequiresSnapshotSourceAlias) + + meta, err := loadRestoreMetadata(targetDir) + require.NoError(t, err) + assert.Equal(t, sourceDir, meta.SnapshotSourceDataDir) + assert.False(t, meta.RetainSnapshotSourceDataDirAlias) + + data, err := os.ReadFile(snapshotStatePath(snapshotDir)) + require.NoError(t, err) + assert.Equal(t, uint64(0), firecrackerSnapshotCRC64(data)) + assert.Contains(t, string(data), sourceDir) + assert.NotContains(t, string(data), targetDir) +} + func TestPrepareFork_ReturnsSourceStatErrors(t *testing.T) { starter := NewStarter() tmp := t.TempDir() @@ -81,6 +150,30 @@ func TestPrepareFork_ReturnsSourceStatErrors(t *testing.T) { assert.Contains(t, err.Error(), "stat snapshot source data dir") } +func TestPrepareFork_RetainedAliasSkipsSourceStat(t *testing.T) { + starter := NewStarter() + tmp := t.TempDir() + upstreamDir := filepath.Join(tmp, "upstream") + targetDir := filepath.Join(tmp, "target") + require.NoError(t, os.MkdirAll(targetDir, 0755)) + require.NoError(t, saveRestoreMetadataState(targetDir, &restoreMetadata{ + SnapshotSourceDataDir: upstreamDir, + RetainSnapshotSourceDataDirAlias: true, + })) + + _, err := starter.PrepareFork(context.Background(), hypervisor.ForkPrepareRequest{ + SnapshotConfigPath: filepath.Join(targetDir, "snapshots", "snapshot-latest", "config.json"), + SourceDataDir: filepath.Join(tmp, "source") + "\x00", + TargetDataDir: targetDir, + }) + require.NoError(t, err) + + meta, err := loadRestoreMetadata(targetDir) + require.NoError(t, err) + assert.Equal(t, upstreamDir, meta.SnapshotSourceDataDir) + assert.True(t, meta.RetainSnapshotSourceDataDirAlias) +} + func TestPrepareFork_PreservesRetainedUpstreamAlias(t *testing.T) { starter := NewStarter() tmp := t.TempDir() @@ -137,3 +230,11 @@ func TestPrepareFork_NetworkRewritePreservesRetainedAlias(t *testing.T) { assert.Equal(t, upstreamDir, meta.SnapshotSourceDataDir) assert.True(t, meta.RetainSnapshotSourceDataDirAlias) } + +func writeSnapshotStateForTest(t *testing.T, path string, body []byte) { + t.Helper() + data := make([]byte, len(body)+8) + copy(data, body) + binary.LittleEndian.PutUint64(data[len(data)-8:], firecrackerSnapshotCRC64(body)) + require.NoError(t, os.WriteFile(path, data, 0644)) +} diff --git a/lib/hypervisor/firecracker/process.go b/lib/hypervisor/firecracker/process.go index 43e52913..8d6d30c2 100644 --- a/lib/hypervisor/firecracker/process.go +++ b/lib/hypervisor/firecracker/process.go @@ -8,7 +8,6 @@ import ( "os/exec" "path/filepath" "strings" - "sync" "syscall" "time" @@ -60,8 +59,6 @@ func WithUFFDClient(client UFFDClient) StarterOption { var _ hypervisor.VMStarter = (*Starter)(nil) -var snapshotSourceAliasMu sync.Mutex - func (s *Starter) SocketName() string { return "fc.sock" } @@ -157,13 +154,19 @@ func (s *Starter) RestoreVM(ctx context.Context, p *paths.Paths, version string, createdUFFDSession = resp.SessionID backend = snapshotMemBackend{BackendType: "Uffd", BackendPath: resp.UFFDSocketPath} } - err = func() error { - snapshotSourceAliasMu.Lock() - defer snapshotSourceAliasMu.Unlock() - return withSnapshotSourceDirAlias(meta, filepath.Dir(socketPath), func() error { - return hv.loadSnapshot(ctx, snapshotPath, meta.NetworkOverrides, backend) - }) - }() + targetDataDir := filepath.Dir(socketPath) + loadSnapshot := func() error { + return hv.loadSnapshot(ctx, snapshotPath, meta.NetworkOverrides, backend) + } + if needsSnapshotSourceDirAlias(meta, targetDataDir) { + err = func() error { + unlockAlias := hypervisor.LockSnapshotSourceAliasMutation() + defer unlockAlias() + return withSnapshotSourceDirAlias(meta, targetDataDir, loadSnapshot) + }() + } else { + err = loadSnapshot() + } if err != nil { if createdUFFDSession != "" { _ = s.uffd.CloseSession(context.Background(), createdUFFDSession) @@ -245,6 +248,15 @@ func withSnapshotSourceDirAlias(meta *restoreMetadata, targetDataDir string, run return nil } +func needsSnapshotSourceDirAlias(meta *restoreMetadata, targetDataDir string) bool { + if meta == nil || meta.SnapshotSourceDataDir == "" { + return false + } + sourceDataDir := filepath.Clean(meta.SnapshotSourceDataDir) + targetDataDir = filepath.Clean(targetDataDir) + return sourceDataDir != targetDataDir +} + func (s *Starter) startProcess(_ context.Context, p *paths.Paths, version string, socketPath string) (int, error) { binaryPath, err := s.GetBinaryPath(p, version) if err != nil { diff --git a/lib/hypervisor/firecracker/snapshot_state.go b/lib/hypervisor/firecracker/snapshot_state.go new file mode 100644 index 00000000..77abd5c1 --- /dev/null +++ b/lib/hypervisor/firecracker/snapshot_state.go @@ -0,0 +1,108 @@ +package firecracker + +import ( + "bytes" + "encoding/binary" + "fmt" + "os" + "path/filepath" +) + +const firecrackerSnapshotCRC64JonesReversedPoly uint64 = 0x95ac9329ac4bc9b5 + +type snapshotStatePathRewrite struct { + Source string + Target string +} + +func rewriteSnapshotStatePathsForFork(statePath string, rewrites []snapshotStatePathRewrite) (bool, error) { + if len(rewrites) == 0 { + return false, nil + } + + data, err := os.ReadFile(statePath) + if err != nil { + if os.IsNotExist(err) { + return false, nil + } + return false, fmt.Errorf("read firecracker snapshot state: %w", err) + } + if len(data) < 8 { + return false, fmt.Errorf("firecracker snapshot state is too short") + } + if firecrackerSnapshotCRC64(data) != 0 { + return false, fmt.Errorf("firecracker snapshot state CRC64 validation failed before path rewrite") + } + + body := append([]byte(nil), data[:len(data)-8]...) + replaced := 0 + targetsBySource := make(map[string][]string) + sources := make([]string, 0, len(rewrites)) + seen := make(map[string]struct{}, len(rewrites)) + for _, rewrite := range rewrites { + source := filepath.Clean(rewrite.Source) + target := filepath.Clean(rewrite.Target) + if source == "." || target == "." || source == target { + continue + } + key := source + "\x00" + target + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + if _, ok := targetsBySource[source]; !ok { + sources = append(sources, source) + } + targetsBySource[source] = append(targetsBySource[source], target) + } + + for _, source := range sources { + sourceBytes := []byte(source) + count := bytes.Count(body, sourceBytes) + if count == 0 { + continue + } + var targetBytes []byte + for _, target := range targetsBySource[source] { + candidate := []byte(target) + if len(sourceBytes) == len(candidate) { + targetBytes = candidate + break + } + } + if len(targetBytes) == 0 { + return false, nil + } + body = bytes.ReplaceAll(body, sourceBytes, targetBytes) + replaced += count + } + if replaced == 0 { + return false, nil + } + + out := make([]byte, len(body)+8) + copy(out, body) + binary.LittleEndian.PutUint64(out[len(out)-8:], firecrackerSnapshotCRC64(body)) + if firecrackerSnapshotCRC64(out) != 0 { + return false, fmt.Errorf("firecracker snapshot state CRC64 validation failed after path rewrite") + } + if err := os.WriteFile(statePath, out, 0644); err != nil { + return false, fmt.Errorf("write firecracker snapshot state: %w", err) + } + return true, nil +} + +func firecrackerSnapshotCRC64(data []byte) uint64 { + crc := uint64(0) + for _, b := range data { + crc ^= uint64(b) + for i := 0; i < 8; i++ { + if crc&1 != 0 { + crc = (crc >> 1) ^ firecrackerSnapshotCRC64JonesReversedPoly + } else { + crc >>= 1 + } + } + } + return crc +} diff --git a/lib/hypervisor/firecracker/snapshot_state_fork.go b/lib/hypervisor/firecracker/snapshot_state_fork.go new file mode 100644 index 00000000..b8350a2c --- /dev/null +++ b/lib/hypervisor/firecracker/snapshot_state_fork.go @@ -0,0 +1,91 @@ +package firecracker + +import ( + "fmt" + "os" + "path/filepath" +) + +func prepareSnapshotDataDirForFork(snapshotConfigPath string, meta *restoreMetadata, sourceDir, targetDir string) (bool, error) { + statePath := snapshotStatePath(filepath.Dir(snapshotConfigPath)) + rewritten, err := rewriteSnapshotStatePathsForFork(statePath, snapshotStatePathRewritesForFork(meta, sourceDir, targetDir)) + if err != nil { + return false, err + } + if rewritten { + return clearSnapshotSourceAlias(meta), nil + } + return recordSnapshotSourceAliasFallback(meta, sourceDir) +} + +func clearSnapshotSourceAlias(meta *restoreMetadata) bool { + changed := false + if meta.SnapshotSourceDataDir != "" { + meta.SnapshotSourceDataDir = "" + changed = true + } + if meta.RetainSnapshotSourceDataDirAlias { + meta.RetainSnapshotSourceDataDirAlias = false + changed = true + } + return changed +} + +func recordSnapshotSourceAliasFallback(meta *restoreMetadata, sourceDir string) (bool, error) { + if meta.RetainSnapshotSourceDataDirAlias && meta.SnapshotSourceDataDir != "" { + // Keep the upstream source path for snapshot-derived forks. The retained + // Firecracker base can still reference that path after later diff snapshots. + return false, nil + } + + retainAlias := false + if _, err := os.Stat(sourceDir); err != nil { + if os.IsNotExist(err) { + retainAlias = true + } else { + return false, fmt.Errorf("stat snapshot source data dir %q: %w", sourceDir, err) + } + } + + changed := false + if meta.SnapshotSourceDataDir != sourceDir { + meta.SnapshotSourceDataDir = sourceDir + changed = true + } + if meta.RetainSnapshotSourceDataDirAlias != retainAlias { + meta.RetainSnapshotSourceDataDirAlias = retainAlias + changed = true + } + return changed, nil +} + +func snapshotStatePathRewritesForFork(meta *restoreMetadata, sourceDataDir, targetDataDir string) []snapshotStatePathRewrite { + var rewrites []snapshotStatePathRewrite + add := func(source, target string) { + if source == "" || target == "" { + return + } + rewrites = append(rewrites, snapshotStatePathRewrite{Source: source, Target: target}) + } + + add(sourceDataDir, targetDataDir) + if meta != nil { + add(meta.SnapshotSourceDataDir, targetDataDir) + } + if resolvedTarget, err := filepath.EvalSymlinks(targetDataDir); err == nil { + add(sourceDataDir, resolvedTarget) + if meta != nil { + add(meta.SnapshotSourceDataDir, resolvedTarget) + } + if resolvedSource, err := filepath.EvalSymlinks(sourceDataDir); err == nil { + add(resolvedSource, resolvedTarget) + } + if meta != nil && meta.SnapshotSourceDataDir != "" { + if resolvedSource, err := filepath.EvalSymlinks(meta.SnapshotSourceDataDir); err == nil { + add(resolvedSource, resolvedTarget) + } + } + } + + return rewrites +} diff --git a/lib/hypervisor/hypervisor.go b/lib/hypervisor/hypervisor.go index de1cb698..6d6d9292 100644 --- a/lib/hypervisor/hypervisor.go +++ b/lib/hypervisor/hypervisor.go @@ -166,6 +166,10 @@ type ForkPrepareResult struct { // VsockCIDUpdated indicates whether snapshot state was updated to use // ForkPrepareRequest.VsockCID. VsockCIDUpdated bool + + // RequiresSnapshotSourceAlias indicates the restored fork still depends on + // temporarily aliasing the source data directory during snapshot load. + RequiresSnapshotSourceAlias bool } // Hypervisor defines the interface for VM control operations. @@ -247,6 +251,10 @@ type Capabilities struct { // on-disk base across restore/standby cycles. SupportsSnapshotBaseReuse bool + // SupportsConcurrentForkPrepare indicates stopped/standby forks can prepare + // separate target snapshots concurrently from the same source. + SupportsConcurrentForkPrepare bool + // SupportsDiskResize indicates if live disk resizing (/vm.resize-disk) is available. // Cloud Hypervisor v50.0+ only. SupportsDiskResize bool diff --git a/lib/hypervisor/snapshot_alias_lock.go b/lib/hypervisor/snapshot_alias_lock.go new file mode 100644 index 00000000..73ba5927 --- /dev/null +++ b/lib/hypervisor/snapshot_alias_lock.go @@ -0,0 +1,20 @@ +package hypervisor + +import "sync" + +var snapshotSourceAliasMu sync.RWMutex + +// LockSnapshotSourceAliasMutation serializes the short Firecracker restore +// window where a snapshotted source guest directory is temporarily replaced by +// a symlink to the fork being restored. +func LockSnapshotSourceAliasMutation() func() { + snapshotSourceAliasMu.Lock() + return snapshotSourceAliasMu.Unlock +} + +// LockSnapshotSourceAliasReaders blocks source-directory readers while a +// snapshot source alias mutation is active. +func LockSnapshotSourceAliasReaders() func() { + snapshotSourceAliasMu.RLock() + return snapshotSourceAliasMu.RUnlock +} diff --git a/lib/instances/fork.go b/lib/instances/fork.go index ccea12db..bbafeabf 100644 --- a/lib/instances/fork.go +++ b/lib/instances/fork.go @@ -10,7 +10,6 @@ import ( "strings" "time" - "github.com/kernel/hypeman/lib/forkvm" "github.com/kernel/hypeman/lib/guest" "github.com/kernel/hypeman/lib/hypervisor" "github.com/kernel/hypeman/lib/instances/phasetracking" @@ -25,7 +24,7 @@ import ( // forkInstance creates a new instance by cloning a stopped or standby source // instance. It returns the newly created fork and the requested final target // state; callers apply remaining target state transitions outside the source lock. -func (m *manager) forkInstance(ctx context.Context, id string, req ForkInstanceRequest) (_ *Instance, _ State, retErr error) { +func (m *manager) forkInstance(ctx context.Context, id string, req ForkInstanceRequest) (_ *Instance, _ State, _ bool, retErr error) { log := logger.FromContext(ctx) log.InfoContext(ctx, "forking instance", "source_instance_id", id, "fork_name", req.Name) ctx, span := m.startLifecycleSpan(ctx, "instances.fork", @@ -35,39 +34,39 @@ func (m *manager) forkInstance(ctx context.Context, id string, req ForkInstanceR defer func() { finishInstancesSpan(span, retErr) }() if err := validateForkRequest(req); err != nil { - return nil, "", err + return nil, "", false, err } meta, err := m.loadMetadata(id) if err != nil { - return nil, "", err + return nil, "", false, err } source := m.toInstance(ctx, meta) targetState, err := resolveForkTargetState(req.TargetState, source.State) if err != nil { - return nil, "", err + return nil, "", false, err } switch source.State { case StateRunning: if !req.FromRunning { - return nil, "", fmt.Errorf("%w: cannot fork from state %s (set from_running=true to allow standby+restore flow)", ErrInvalidState, source.State) + return nil, "", false, fmt.Errorf("%w: cannot fork from state %s (set from_running=true to allow standby+restore flow)", ErrInvalidState, source.State) } if err := m.validateForkSupport(ctx, source.HypervisorType); err != nil { - return nil, "", err + return nil, "", false, err } if err := ensureGuestAgentReadyForRunningFork(ctx, &source.StoredMetadata); err != nil { - return nil, "", err + return nil, "", false, err } log.InfoContext(ctx, "fork from running requested; transitioning source to standby", "source_instance_id", id, "hypervisor", source.HypervisorType) if _, err := m.standbyInstance(ctx, id, StandbyInstanceRequest{}, true); err != nil { - return nil, "", fmt.Errorf("standby source instance: %w", err) + return nil, "", false, fmt.Errorf("standby source instance: %w", err) } - forked, forkErr := m.forkInstanceFromStoppedOrStandby(ctx, id, req, true) + forked, _, forkErr := m.forkInstanceFromStoppedOrStandby(ctx, id, req, true) if forkErr == nil && targetState != StateStopped { if err := m.rotateSourceVsockForRestore(ctx, id, forked.Id); err != nil { forkErr = fmt.Errorf("prepare source snapshot for restore: %w", err) @@ -97,30 +96,30 @@ func (m *manager) forkInstance(ctx context.Context, id string, req ForkInstanceR if restoreErr != nil { if forkErr != nil { - return nil, "", fmt.Errorf("fork failed: %v; additionally failed to restore source instance: %w", forkErr, restoreErr) + return nil, "", false, fmt.Errorf("fork failed: %v; additionally failed to restore source instance: %w", forkErr, restoreErr) } - return nil, "", fmt.Errorf("restore source instance after fork: %w", restoreErr) + return nil, "", false, fmt.Errorf("restore source instance after fork: %w", restoreErr) } if restoredSource != nil && !restoredSource.NetworkEnabled && (restoredSource.State == StateRunning || restoredSource.State == StateInitializing) { if err := ensureGuestAgentReadyForForkPhase(ctx, &restoredSource.StoredMetadata, "after restoring running fork source"); err != nil { if forkErr != nil { - return nil, "", fmt.Errorf("fork failed: %v; additionally restored source guest agent was not ready: %w", forkErr, err) + return nil, "", false, fmt.Errorf("fork failed: %v; additionally restored source guest agent was not ready: %w", forkErr, err) } - return nil, "", fmt.Errorf("wait for restored source guest agent readiness: %w", err) + return nil, "", false, fmt.Errorf("wait for restored source guest agent readiness: %w", err) } } if forkErr != nil { - return nil, "", forkErr + return nil, "", false, forkErr } - return forked, targetState, nil + return forked, targetState, false, nil case StateStopped, StateStandby: - forked, err := m.forkInstanceFromStoppedOrStandby(ctx, id, req, false) + forked, restoreNeedsSourceLock, err := m.forkInstanceFromStoppedOrStandby(ctx, id, req, false) if err != nil { - return nil, "", err + return nil, "", false, err } - return forked, targetState, nil + return forked, targetState, restoreNeedsSourceLock, nil default: - return nil, "", fmt.Errorf("%w: cannot fork from state %s (must be Stopped or Standby, or Running with from_running=true)", ErrInvalidState, source.State) + return nil, "", false, fmt.Errorf("%w: cannot fork from state %s (must be Stopped or Standby, or Running with from_running=true)", ErrInvalidState, source.State) } } @@ -202,12 +201,12 @@ func generateForkSourceVsockCID(sourceID, forkID string, current int64) int64 { return cid } -func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id string, req ForkInstanceRequest, supportValidated bool) (*Instance, error) { +func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id string, req ForkInstanceRequest, supportValidated bool) (*Instance, bool, error) { log := logger.FromContext(ctx) meta, err := m.loadMetadata(id) if err != nil { - return nil, err + return nil, false, err } source := m.toInstance(ctx, meta) @@ -217,38 +216,38 @@ func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id strin case StateStopped, StateStandby: // allowed default: - return nil, fmt.Errorf("%w: cannot fork from state %s (must be Stopped or Standby)", ErrInvalidState, source.State) + return nil, false, fmt.Errorf("%w: cannot fork from state %s (must be Stopped or Standby)", ErrInvalidState, source.State) } if !supportValidated { if err := m.validateForkSupport(ctx, stored.HypervisorType); err != nil { - return nil, err + return nil, false, err } } if err := validateForkVolumeSafety(stored.Volumes); err != nil { - return nil, err + return nil, false, err } existsByMetadata, err := m.instanceNameExists(req.Name) if err != nil { - return nil, fmt.Errorf("check instance name availability: %w", err) + return nil, false, fmt.Errorf("check instance name availability: %w", err) } if existsByMetadata { - return nil, fmt.Errorf("%w: instance name '%s' already exists", ErrAlreadyExists, req.Name) + return nil, false, fmt.Errorf("%w: instance name '%s' already exists", ErrAlreadyExists, req.Name) } if stored.NetworkEnabled { exists, err := m.networkManager.NameExists(ctx, req.Name, "") if err != nil { - return nil, fmt.Errorf("check instance name availability: %w", err) + return nil, false, fmt.Errorf("check instance name availability: %w", err) } if exists { - return nil, fmt.Errorf("%w: instance name '%s' already exists in network", ErrAlreadyExists, req.Name) + return nil, false, fmt.Errorf("%w: instance name '%s' already exists in network", ErrAlreadyExists, req.Name) } } forkID := cuid2.Generate() if _, err := m.loadMetadata(forkID); err == nil { - return nil, fmt.Errorf("%w: generated fork id already exists", ErrAlreadyExists) + return nil, false, fmt.Errorf("%w: generated fork id already exists", ErrAlreadyExists) } srcDir := m.paths.InstanceDir(id) @@ -259,22 +258,13 @@ func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id strin }) defer cu.Clean() - if source.State == StateStandby { - if err := m.ensureSnapshotMemoryReady(ctx, m.paths.InstanceSnapshotLatest(id), m.snapshotJobKeyForInstance(id), stored.HypervisorType); err != nil { - return nil, fmt.Errorf("prepare standby snapshot for fork: %w", err) - } - } - - if err := forkvm.CopyGuestDirectory(srcDir, dstDir); err != nil { - if errors.Is(err, forkvm.ErrSparseCopyUnsupported) { - return nil, fmt.Errorf("fork requires sparse-capable filesystem (SEEK_DATA/SEEK_HOLE unsupported): %w", err) - } - return nil, fmt.Errorf("clone guest directory: %w", err) + if err := m.copyForkSourceGuestDirectory(ctx, source.State, id, stored, srcDir, dstDir); err != nil { + return nil, false, err } starter, err := m.getVMStarter(stored.HypervisorType) if err != nil { - return nil, fmt.Errorf("get vm starter: %w", err) + return nil, false, fmt.Errorf("get vm starter: %w", err) } now := time.Now().UTC() @@ -308,9 +298,10 @@ func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id strin forkMeta.Phases.Record(phasetracking.PhaseStopped, now) } - // Keep the original CID for snapshot-based forks. - // Rewriting CID in restored memory snapshots is not reliable across - // hypervisors. + // Keep the original CID for snapshot-based forks. Rewriting CID in restored + // memory snapshots is not reliable across hypervisors. Concurrent standby + // fork prepare is currently enabled only for Firecracker, whose host vsock + // dialer routes through the per-VM UDS path rather than this metadata CID. if source.State == StateStandby { forkMeta.VsockCID = stored.VsockCID } else { @@ -324,13 +315,14 @@ func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id strin forkMeta.MAC = "" } + restoreNeedsSourceLock := false if source.State == StateStandby { snapshotConfigPath := m.paths.InstanceSnapshotConfig(forkID) netCfg := (*hypervisor.ForkNetworkConfig)(nil) if forkMeta.NetworkEnabled { netCfg = &hypervisor.ForkNetworkConfig{TAPDevice: network.GenerateTAPName(forkID)} } - if _, err := starter.PrepareFork(ctx, hypervisor.ForkPrepareRequest{ + prepareResult, err := prepareForkWithAliasReadLock(ctx, starter, hypervisor.ForkPrepareRequest{ SnapshotConfigPath: snapshotConfigPath, SourceDataDir: stored.DataDir, TargetDataDir: forkMeta.DataDir, @@ -338,17 +330,27 @@ func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id strin VsockSocket: forkMeta.VsockSocket, SerialLogPath: m.paths.InstanceAppLog(forkID), Network: netCfg, - }); err != nil { + }) + if err != nil { if errors.Is(err, hypervisor.ErrNotSupported) { - return nil, fmt.Errorf("%w: fork is not supported for hypervisor %s", ErrNotSupported, stored.HypervisorType) + return nil, false, fmt.Errorf("%w: fork is not supported for hypervisor %s", ErrNotSupported, stored.HypervisorType) } - return nil, fmt.Errorf("prepare fork snapshot state: %w", err) + return nil, false, fmt.Errorf("prepare fork snapshot state: %w", err) } + if prepareResult.RequiresSnapshotSourceAlias { + log.WarnContext(ctx, "fork restore requires snapshot source alias fallback", + "hypervisor", stored.HypervisorType, + "source_instance_id", id, + "fork_instance_id", forkID, + "source_data_dir", stored.DataDir, + "target_data_dir", forkMeta.DataDir) + } + restoreNeedsSourceLock = prepareResult.RequiresSnapshotSourceAlias } newMeta := &metadata{StoredMetadata: forkMeta} - if err := m.saveMetadata(newMeta); err != nil { - return nil, fmt.Errorf("save fork metadata: %w", err) + if err := m.saveForkMetadata(ctx, newMeta); err != nil { + return nil, false, err } cu.Release() @@ -358,7 +360,37 @@ func (m *manager) forkInstanceFromStoppedOrStandby(ctx context.Context, id strin "fork_instance_id", forked.Id, "fork_name", forked.Name, "state", forked.State) - return &forked, nil + return &forked, restoreNeedsSourceLock, nil +} + +func (m *manager) saveForkMetadata(ctx context.Context, meta *metadata) error { + m.forkMetadataMu.Lock() + defer m.forkMetadataMu.Unlock() + + // Earlier name checks are advisory so callers can fail before doing fork + // work when possible. This is the serialized admission point for fork + // metadata, so concurrent forks re-check names immediately before save. + name := meta.Name + existsByMetadata, err := m.instanceNameExists(name) + if err != nil { + return fmt.Errorf("check instance name availability: %w", err) + } + if existsByMetadata { + return fmt.Errorf("%w: instance name '%s' already exists", ErrAlreadyExists, name) + } + if meta.NetworkEnabled { + exists, err := m.networkManager.NameExists(ctx, name, "") + if err != nil { + return fmt.Errorf("check instance name availability: %w", err) + } + if exists { + return fmt.Errorf("%w: instance name '%s' already exists in network", ErrAlreadyExists, name) + } + } + if err := m.saveMetadata(meta); err != nil { + return fmt.Errorf("save fork metadata: %w", err) + } + return nil } func (m *manager) validateForkSupport(ctx context.Context, hvType hypervisor.Type) error { diff --git a/lib/instances/fork_test.go b/lib/instances/fork_test.go index 606b7671..8bf7f6cc 100644 --- a/lib/instances/fork_test.go +++ b/lib/instances/fork_test.go @@ -4,12 +4,14 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" "os" "path/filepath" "strings" + "sync" "sync/atomic" "testing" "time" @@ -137,6 +139,277 @@ func TestCleanupForkInstanceOnError(t *testing.T) { assert.ErrorIs(t, err, ErrNotFound) } +func TestGetInstanceWaitsDuringSnapshotSourceAlias(t *testing.T) { + manager, _ := setupTestManager(t) + ctx := context.Background() + + sourceID := "fork-alias-source" + aliasID := "fork-alias-child" + now := time.Now() + for _, meta := range []*metadata{ + {StoredMetadata: StoredMetadata{ + Id: sourceID, + Name: sourceID, + Image: integrationTestImageRef(t, "docker.io/library/alpine:latest"), + CreatedAt: now, + StoppedAt: &now, + HypervisorType: hypervisor.TypeFirecracker, + HypervisorVersion: "test", + SocketPath: paths.New(manager.paths.DataDir()).InstanceSocket(sourceID, "fc.sock"), + DataDir: paths.New(manager.paths.DataDir()).InstanceDir(sourceID), + VsockCID: 42, + VsockSocket: paths.New(manager.paths.DataDir()).InstanceVsockSocket(sourceID), + }}, + {StoredMetadata: StoredMetadata{ + Id: aliasID, + Name: aliasID, + Image: integrationTestImageRef(t, "docker.io/library/alpine:latest"), + CreatedAt: now, + HypervisorType: hypervisor.TypeFirecracker, + HypervisorVersion: "test", + SocketPath: paths.New(manager.paths.DataDir()).InstanceSocket(aliasID, "fc.sock"), + DataDir: paths.New(manager.paths.DataDir()).InstanceDir(aliasID), + VsockCID: 43, + VsockSocket: paths.New(manager.paths.DataDir()).InstanceVsockSocket(aliasID), + }}, + } { + require.NoError(t, manager.ensureDirectories(meta.Id)) + require.NoError(t, manager.saveMetadata(meta)) + } + + sourceDir := manager.paths.InstanceDir(sourceID) + childDir := manager.paths.InstanceDir(aliasID) + backupDir := sourceDir + ".bak" + + unlockAlias := hypervisor.LockSnapshotSourceAliasMutation() + aliasLocked := true + defer func() { + if aliasLocked { + unlockAlias() + } + }() + require.NoError(t, os.Rename(sourceDir, backupDir)) + require.NoError(t, os.Symlink(childDir, sourceDir)) + + type getResult struct { + inst *Instance + err error + } + done := make(chan getResult, 1) + go func() { + inst, err := manager.GetInstance(ctx, sourceID) + done <- getResult{inst: inst, err: err} + }() + + select { + case got := <-done: + t.Fatalf("GetInstance returned during source alias mutation: inst=%v err=%v", got.inst, got.err) + case <-time.After(25 * time.Millisecond): + } + + require.NoError(t, os.Remove(sourceDir)) + require.NoError(t, os.Rename(backupDir, sourceDir)) + unlockAlias() + aliasLocked = false + + got := <-done + require.NoError(t, got.err) + require.NotNil(t, got.inst) + assert.Equal(t, sourceID, got.inst.Id) +} + +func TestForkInstanceStoppedSourceUsesReadLock(t *testing.T) { + manager, _ := setupTestManager(t) + ctx := context.Background() + hvType := hypervisor.Type("concurrent-fork-test") + hypervisor.RegisterCapabilities(hvType, hypervisor.Capabilities{SupportsConcurrentForkPrepare: true}) + manager.vmStarters[hvType] = concurrentForkPrepareTestStarter{} + + sourceID := "fork-stopped-read-lock-source" + createStoppedSnapshotSourceFixture(t, manager, sourceID, sourceID, hvType) + + sourceLock := manager.getInstanceLock(sourceID) + sourceLock.RLock() + defer sourceLock.RUnlock() + + type forkResult struct { + inst *Instance + err error + } + done := make(chan forkResult, 1) + go func() { + inst, err := manager.ForkInstance(ctx, sourceID, ForkInstanceRequest{Name: "fork-stopped-read-lock-copy"}) + done <- forkResult{inst: inst, err: err} + }() + + select { + case got := <-done: + require.NoError(t, got.err) + require.NotNil(t, got.inst) + assert.Equal(t, StateStopped, got.inst.State) + case <-time.After(250 * time.Millisecond): + t.Fatal("ForkInstance blocked behind a source read lock") + } +} + +type concurrentForkPrepareTestStarter struct{} + +func (concurrentForkPrepareTestStarter) SocketName() string { return "concurrent-fork-test.sock" } + +func (concurrentForkPrepareTestStarter) GetBinaryPath(*paths.Paths, string) (string, error) { + return "", nil +} + +func (concurrentForkPrepareTestStarter) GetVersion(*paths.Paths) (string, error) { return "test", nil } + +func (concurrentForkPrepareTestStarter) StartVM(context.Context, *paths.Paths, string, string, hypervisor.VMConfig) (int, hypervisor.Hypervisor, error) { + return 0, nil, nil +} + +func (concurrentForkPrepareTestStarter) RestoreVM(_ context.Context, _ *paths.Paths, _ string, socketPath string, _ string, _ hypervisor.RestoreOptions) (int, hypervisor.Hypervisor, error) { + if err := os.MkdirAll(filepath.Dir(socketPath), 0755); err != nil { + return 0, nil, err + } + if err := os.WriteFile(socketPath, []byte("socket"), 0644); err != nil { + return 0, nil, err + } + return 1, lifecycleNoopHypervisor{state: hypervisor.StateRunning}, nil +} + +func (concurrentForkPrepareTestStarter) PrepareFork(context.Context, hypervisor.ForkPrepareRequest) (hypervisor.ForkPrepareResult, error) { + return hypervisor.ForkPrepareResult{}, nil +} + +func TestForkInstanceStandbyRunningTargetSkipsSourceWriteLockWithoutAlias(t *testing.T) { + manager, _ := setupTestManager(t) + ctx := context.Background() + hvType := hypervisor.Type("concurrent-fork-restore-test") + hypervisor.RegisterCapabilities(hvType, hypervisor.Capabilities{SupportsConcurrentForkPrepare: true}) + hypervisor.RegisterClientFactory(hvType, func(string) (hypervisor.Hypervisor, error) { + return lifecycleNoopHypervisor{state: hypervisor.StateRunning}, nil + }) + manager.vmStarters[hvType] = concurrentForkPrepareTestStarter{} + + sourceID := "fork-standby-running-read-lock-source" + createStandbySnapshotSourceFixture(t, manager, sourceID, sourceID, hvType) + meta, err := manager.loadMetadata(sourceID) + require.NoError(t, err) + meta.StoredMetadata.SkipGuestAgent = true + require.NoError(t, manager.saveMetadata(meta)) + + sourceLock := manager.getInstanceLock(sourceID) + sourceLock.RLock() + defer sourceLock.RUnlock() + + type forkResult struct { + inst *Instance + err error + } + done := make(chan forkResult, 1) + go func() { + inst, err := manager.ForkInstance(ctx, sourceID, ForkInstanceRequest{ + Name: "fork-standby-running-read-lock-copy", + TargetState: StateRunning, + }) + done <- forkResult{inst: inst, err: err} + }() + + select { + case got := <-done: + require.NoError(t, got.err) + require.NotNil(t, got.inst) + assert.Contains(t, []State{StateRunning, StateInitializing}, got.inst.State) + case <-time.After(250 * time.Millisecond): + t.Fatal("ForkInstance blocked behind the source write lock for a no-alias restore") + } +} + +func TestApplyForkTargetStateWithSourceLockRequiresStandbySource(t *testing.T) { + t.Parallel() + + manager, _ := setupTestManager(t) + ctx := context.Background() + + sourceID := "fork-alias-running-source" + forkID := "fork-alias-running-child" + now := time.Now() + require.NoError(t, manager.ensureDirectories(sourceID)) + require.NoError(t, manager.ensureDirectories(forkID)) + + sourceSocket := manager.paths.InstanceSocket(sourceID, "fc.sock") + require.NoError(t, os.MkdirAll(filepath.Dir(sourceSocket), 0755)) + require.NoError(t, os.WriteFile(sourceSocket, []byte("socket"), 0644)) + manager.storeCachedHypervisorState(sourceID, hypervisor.StateRunning) + require.NoError(t, manager.saveMetadata(&metadata{StoredMetadata: StoredMetadata{ + Id: sourceID, + Name: sourceID, + Image: integrationTestImageRef(t, "docker.io/library/alpine:latest"), + CreatedAt: now, + HypervisorType: hypervisor.TypeFirecracker, + HypervisorVersion: "test", + SocketPath: sourceSocket, + DataDir: manager.paths.InstanceDir(sourceID), + VsockCID: 42, + VsockSocket: manager.paths.InstanceVsockSocket(sourceID), + }})) + + forkSnapshotDir := manager.paths.InstanceSnapshotLatest(forkID) + require.NoError(t, os.MkdirAll(forkSnapshotDir, 0755)) + require.NoError(t, os.WriteFile(filepath.Join(forkSnapshotDir, "memory"), []byte("snapshot"), 0644)) + require.NoError(t, manager.saveMetadata(&metadata{StoredMetadata: StoredMetadata{ + Id: forkID, + Name: forkID, + Image: integrationTestImageRef(t, "docker.io/library/alpine:latest"), + CreatedAt: now, + HypervisorType: hypervisor.TypeFirecracker, + HypervisorVersion: "test", + SocketPath: manager.paths.InstanceSocket(forkID, "fc.sock"), + DataDir: manager.paths.InstanceDir(forkID), + VsockCID: 42, + VsockSocket: manager.paths.InstanceVsockSocket(forkID), + SkipGuestAgent: true, + }})) + + _, err := manager.applyForkTargetStateWithSourceLock(ctx, manager.getInstanceLock(sourceID), sourceID, forkID, StateRunning) + require.Error(t, err) + assert.ErrorIs(t, err, ErrInvalidState) + assert.Contains(t, err.Error(), "source fork-alias-running-source is now") +} + +func TestForkInstanceStoppedSourceRequiresWriteLockWithoutConcurrentPrepare(t *testing.T) { + manager, _ := setupTestManager(t) + ctx := context.Background() + + sourceID := "fork-stopped-write-lock-source" + createStoppedSnapshotSourceFixture(t, manager, sourceID, sourceID, hypervisor.TypeQEMU) + + sourceLock := manager.getInstanceLock(sourceID) + sourceLock.RLock() + + type forkResult struct { + inst *Instance + err error + } + done := make(chan forkResult, 1) + go func() { + inst, err := manager.ForkInstance(ctx, sourceID, ForkInstanceRequest{Name: "fork-stopped-write-lock-copy"}) + done <- forkResult{inst: inst, err: err} + }() + + select { + case got := <-done: + sourceLock.RUnlock() + t.Fatalf("ForkInstance returned without exclusive source lock: inst=%v err=%v", got.inst, got.err) + case <-time.After(25 * time.Millisecond): + } + + sourceLock.RUnlock() + got := <-done + require.NoError(t, got.err) + require.NotNil(t, got.inst) + assert.Equal(t, StateStopped, got.inst.State) +} + func TestForkInstance_CleansUpOnTargetTransitionError(t *testing.T) { t.Parallel() manager, _ := setupTestManager(t) @@ -221,6 +494,69 @@ func TestForkInstanceRejectsDuplicateNameForNonNetworkedSource(t *testing.T) { assert.Contains(t, err.Error(), "already exists") } +func TestSaveForkMetadataSerializesNameAdmission(t *testing.T) { + t.Parallel() + manager, _ := setupTestManager(t) + ctx := context.Background() + + now := time.Now() + metas := []*metadata{ + {StoredMetadata: StoredMetadata{ + Id: "fork-concurrent-name-a", + Name: "fork-concurrent-name", + Image: integrationTestImageRef(t, "docker.io/library/alpine:latest"), + CreatedAt: now, + HypervisorType: hypervisor.TypeCloudHypervisor, + HypervisorVersion: "test", + SocketPath: paths.New(manager.paths.DataDir()).InstanceSocket("fork-concurrent-name-a", "cloud-hypervisor.sock"), + DataDir: paths.New(manager.paths.DataDir()).InstanceDir("fork-concurrent-name-a"), + VsockCID: 42, + VsockSocket: paths.New(manager.paths.DataDir()).InstanceVsockSocket("fork-concurrent-name-a"), + }}, + {StoredMetadata: StoredMetadata{ + Id: "fork-concurrent-name-b", + Name: "fork-concurrent-name", + Image: integrationTestImageRef(t, "docker.io/library/alpine:latest"), + CreatedAt: now, + HypervisorType: hypervisor.TypeCloudHypervisor, + HypervisorVersion: "test", + SocketPath: paths.New(manager.paths.DataDir()).InstanceSocket("fork-concurrent-name-b", "cloud-hypervisor.sock"), + DataDir: paths.New(manager.paths.DataDir()).InstanceDir("fork-concurrent-name-b"), + VsockCID: 43, + VsockSocket: paths.New(manager.paths.DataDir()).InstanceVsockSocket("fork-concurrent-name-b"), + }}, + } + for _, meta := range metas { + require.NoError(t, manager.ensureDirectories(meta.Id)) + } + + var wg sync.WaitGroup + errs := make([]error, len(metas)) + for i := range metas { + wg.Add(1) + go func(i int) { + defer wg.Done() + errs[i] = manager.saveForkMetadata(ctx, metas[i]) + }(i) + } + wg.Wait() + + successes := 0 + duplicates := 0 + for _, err := range errs { + switch { + case err == nil: + successes++ + case errors.Is(err, ErrAlreadyExists): + duplicates++ + default: + require.NoError(t, err) + } + } + assert.Equal(t, 1, successes) + assert.Equal(t, 1, duplicates) +} + func TestForkInstanceFromStandbyCancelsCompressionJobAndCopiesRawMemory(t *testing.T) { t.Parallel() @@ -265,7 +601,7 @@ func TestForkInstanceFromStandbyCancelsCompressionJobAndCopiesRawMemory(t *testi } manager.compressionMu.Unlock() - forked, err := manager.forkInstanceFromStoppedOrStandby(ctx, sourceID, ForkInstanceRequest{ + forked, _, err := manager.forkInstanceFromStoppedOrStandby(ctx, sourceID, ForkInstanceRequest{ Name: "fork-standby-compressed-copy", TargetState: StateStopped, }, true) diff --git a/lib/instances/manager.go b/lib/instances/manager.go index fb203965..8e8e25f3 100644 --- a/lib/instances/manager.go +++ b/lib/instances/manager.go @@ -83,11 +83,21 @@ type ResourceLimits struct { // ManagerConfig holds non-resource manager behavior settings. type ManagerConfig struct { - LifecycleEventBufferSize int - FirecrackerSnapshotMemoryBackend string - FirecrackerUFFDCacheMaxBytes int64 + LifecycleEventBufferSize int + FirecrackerSnapshotMemoryBackend string + FirecrackerUFFDCacheMaxBytes int64 + MaxConcurrentRestoresByHypervisor map[hypervisor.Type]int } +const defaultManagerMaxConcurrentRestores = 4 + +type restoreSlotPoolKey struct { + hypervisorType hypervisor.Type + limit int +} + +var restoreSlotPools sync.Map // map[restoreSlotPoolKey]chan struct{} + // Normalize applies defaults to manager config values. func (c ManagerConfig) Normalize() ManagerConfig { if c.LifecycleEventBufferSize <= 0 { @@ -100,9 +110,43 @@ func (c ManagerConfig) Normalize() ManagerConfig { if c.FirecrackerUFFDCacheMaxBytes <= 0 { c.FirecrackerUFFDCacheMaxBytes = 4 << 30 } + if c.MaxConcurrentRestoresByHypervisor == nil { + c.MaxConcurrentRestoresByHypervisor = map[hypervisor.Type]int{ + hypervisor.TypeFirecracker: defaultManagerMaxConcurrentRestores, + } + } else { + limits := make(map[hypervisor.Type]int, len(c.MaxConcurrentRestoresByHypervisor)) + for hvType, limit := range c.MaxConcurrentRestoresByHypervisor { + if limit <= 0 { + limit = defaultManagerMaxConcurrentRestores + } + limits[hvType] = limit + } + c.MaxConcurrentRestoresByHypervisor = limits + } return c } +func sharedRestoreSlots(hvType hypervisor.Type, limit int) chan struct{} { + if limit <= 0 { + limit = defaultManagerMaxConcurrentRestores + } + key := restoreSlotPoolKey{hypervisorType: hvType, limit: limit} + slots, _ := restoreSlotPools.LoadOrStore(key, make(chan struct{}, limit)) + return slots.(chan struct{}) +} + +func sharedRestoreSlotsByHypervisor(limits map[hypervisor.Type]int) map[hypervisor.Type]chan struct{} { + if len(limits) == 0 { + return nil + } + slots := make(map[hypervisor.Type]chan struct{}, len(limits)) + for hvType, limit := range limits { + slots[hvType] = sharedRestoreSlots(hvType, limit) + } + return slots +} + // ResourceValidator validates if resources can be allocated type ResourceValidator interface { // ValidateAllocation checks if the requested resources are available. @@ -126,9 +170,10 @@ type manager struct { limits ResourceLimits resourceValidator ResourceValidator // Optional validator for aggregate resource limits instanceLocks sync.Map // map[string]*sync.RWMutex - per-instance locks - bootMarkerScans sync.Map // map[string]time.Time next allowed boot-marker rescan - hypervisorStateCache sync.Map // map[string]hypervisorStateCacheEntry - last observed hypervisor state per instance - hostTopology *HostTopology // Cached host CPU topology + forkMetadataMu sync.Mutex + bootMarkerScans sync.Map // map[string]time.Time next allowed boot-marker rescan + hypervisorStateCache sync.Map // map[string]hypervisorStateCacheEntry - last observed hypervisor state per instance + hostTopology *HostTopology // Cached host CPU topology metrics *Metrics meter metric.Meter tracer trace.Tracer @@ -141,6 +186,7 @@ type manager struct { snapshotDefaults SnapshotPolicy compressionMu sync.Mutex compressionJobs map[string]*compressionJob + snapshotPrepareLocks sync.Map // map[string]*sync.Mutex - per-snapshot memory prepare locks compressionTimerFactory func(time.Duration) compressionTimer nativeCodecMu sync.Mutex nativeCodecPaths map[string]string @@ -165,6 +211,7 @@ type manager struct { guestMemoryPolicy guestmemory.Policy firecrackerSnapshotMemoryBackend string firecrackerUFFDPager *uffdpager.Supervisor + restoreSlotsByHypervisor map[hypervisor.Type]chan struct{} } // platformStarters is populated by platform-specific init functions. @@ -235,6 +282,7 @@ func NewManagerWithConfigE(p *paths.Paths, imageManager images.Manager, systemMa guestMemoryPolicy: policy, firecrackerSnapshotMemoryBackend: managerConfig.FirecrackerSnapshotMemoryBackend, firecrackerUFFDPager: firecrackerUFFDPager, + restoreSlotsByHypervisor: sharedRestoreSlotsByHypervisor(managerConfig.MaxConcurrentRestoresByHypervisor), snapshotDefaults: snapshotDefaults, compressionJobs: make(map[string]*compressionJob), nativeCodecPaths: make(map[string]string), @@ -318,6 +366,14 @@ func (m *manager) supportsSnapshotBaseReuse(hvType hypervisor.Type) bool { return caps.SupportsSnapshotBaseReuse } +func (m *manager) supportsConcurrentForkPrepare(hvType hypervisor.Type) bool { + caps, ok := hypervisor.CapabilitiesForType(hvType) + if !ok { + return false + } + return caps.SupportsConcurrentForkPrepare +} + // getInstanceLock returns or creates a lock for a specific instance func (m *manager) getInstanceLock(id string) *sync.RWMutex { lock, _ := m.instanceLocks.LoadOrStore(id, &sync.RWMutex{}) @@ -415,16 +471,59 @@ func (m *manager) DeleteSnapshot(ctx context.Context, snapshotID string) error { // ForkInstance creates a forked copy of an instance. func (m *manager) ForkInstance(ctx context.Context, id string, req ForkInstanceRequest) (*Instance, error) { lock := m.getInstanceLock(id) - lock.Lock() - forked, targetState, err := m.forkInstance(ctx, id, req) - lock.Unlock() + useReadLock := false + var sourceState State + // Some hypervisors can safely prepare stopped/standby forks concurrently + // from the same source. Probe under a short read lock first; running sources + // and unsupported hypervisors fall back to the normal exclusive fork path. + lock.RLock() + if meta, err := m.loadMetadata(id); err == nil { + source := m.toInstance(ctx, meta) + sourceState = source.State + useReadLock = m.supportsConcurrentForkPrepare(meta.HypervisorType) && (source.State == StateStopped || source.State == StateStandby) + } + lock.RUnlock() + + var forked *Instance + var targetState State + var targetRestoreNeedsSourceLock bool + var err error + if useReadLock { + // State may have changed after the first probe. Re-load while holding + // the lock used by forkInstance, and only proceed under RLock if the + // source is still in a concurrency-safe state. + lock.RLock() + if meta, loadErr := m.loadMetadata(id); loadErr == nil { + source := m.toInstance(ctx, meta) + sourceState = source.State + useReadLock = m.supportsConcurrentForkPrepare(meta.HypervisorType) && (source.State == StateStopped || source.State == StateStandby) + } else { + useReadLock = false + } + if useReadLock { + forked, targetState, targetRestoreNeedsSourceLock, err = m.forkInstance(ctx, id, req) + } + lock.RUnlock() + } + if !useReadLock { + lock.Lock() + forked, targetState, targetRestoreNeedsSourceLock, err = m.forkInstance(ctx, id, req) + lock.Unlock() + } if err != nil { return nil, err } inst := forked if !forkTargetStateAlreadyApplied(inst, targetState) { - inst, err = m.applyForkTargetState(ctx, forked.Id, targetState) + // Standby -> running forks prepared under RLock can still require the + // source write lock if Firecracker snapshot paths could not be rewritten + // and restore must temporarily alias the source data dir. + if useReadLock && sourceState == StateStandby && targetState == StateRunning && targetRestoreNeedsSourceLock { + inst, err = m.applyForkTargetStateWithSourceLock(ctx, lock, id, forked.Id, targetState) + } 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) @@ -436,6 +535,20 @@ func (m *manager) ForkInstance(ctx context.Context, id string, req ForkInstanceR return inst, nil } +func (m *manager) applyForkTargetStateWithSourceLock(ctx context.Context, lock *sync.RWMutex, sourceID, forkID string, target State) (*Instance, error) { + lock.Lock() + defer lock.Unlock() + sourceMeta, err := m.loadMetadata(sourceID) + if err != nil { + return nil, fmt.Errorf("reload source metadata before aliased fork restore: %w", err) + } + source := m.toInstance(ctx, sourceMeta) + if source.State != StateStandby { + return nil, fmt.Errorf("%w: cannot restore fork %s with snapshot source alias because source %s is now %s", ErrInvalidState, forkID, sourceID, source.State) + } + return m.applyForkTargetState(ctx, forkID, target) +} + func (m *manager) ForkSnapshot(ctx context.Context, snapshotID string, req ForkSnapshotRequest) (*Instance, error) { inst, err := m.forkSnapshot(ctx, snapshotID, req) if err == nil { diff --git a/lib/instances/restore.go b/lib/instances/restore.go index b8c28ba6..13863d30 100644 --- a/lib/instances/restore.go +++ b/lib/instances/restore.go @@ -254,6 +254,26 @@ func (m *manager) restoreInstance( return nil, fmt.Errorf("configure snapshot memory backend: %w", err) } + restoreSlotCtx, restoreSlotSpanEnd := m.startLifecycleStep(ctx, "restore_capacity_wait", + attribute.String("instance_id", id), + attribute.String("hypervisor", string(stored.HypervisorType)), + attribute.String("operation", "restore_capacity_wait"), + ) + releaseRestoreSlot, err := m.acquireRestoreSlot(restoreSlotCtx, stored.HypervisorType) + restoreSlotSpanEnd(err) + if err != nil { + releaseNetwork() + return nil, fmt.Errorf("wait for restore capacity: %w", err) + } + restoreSlotHeld := true + releaseRestoreSlotOnce := func() { + if restoreSlotHeld { + restoreSlotHeld = false + releaseRestoreSlot() + } + } + defer releaseRestoreSlotOnce() + // 5. Transition: Standby → Paused (start hypervisor + restore) restoreCtx, restoreSpanEnd := m.startLifecycleStep(ctx, "restore_from_snapshot", attribute.String("instance_id", id), @@ -321,6 +341,7 @@ func (m *manager) restoreInstance( } reconfigureSpanEnd(nil) } + releaseRestoreSlotOnce() // 8. Delete snapshot after successful restore unless the hypervisor is keeping it // as the base for the next standby snapshot. @@ -409,6 +430,22 @@ func (m *manager) restoreFromSnapshot( return pid, hv, nil } +func (m *manager) acquireRestoreSlot(ctx context.Context, hvType hypervisor.Type) (func(), error) { + if m.restoreSlotsByHypervisor == nil { + return func() {}, nil + } + slots := m.restoreSlotsByHypervisor[hvType] + if slots == nil { + return func() {}, nil + } + select { + case slots <- struct{}{}: + return func() { <-slots }, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + func reconfigureGuestNetwork(ctx context.Context, stored *StoredMetadata, alloc *network.Allocation) error { cfg, err := guestNetworkReconfigureConfig(alloc) if err != nil { diff --git a/lib/instances/restore_capacity_test.go b/lib/instances/restore_capacity_test.go new file mode 100644 index 00000000..cadc6ce7 --- /dev/null +++ b/lib/instances/restore_capacity_test.go @@ -0,0 +1,55 @@ +package instances + +import ( + "context" + "testing" + "time" + + "github.com/kernel/hypeman/lib/hypervisor" +) + +func TestAcquireRestoreSlotLimitsConfiguredHypervisorOnly(t *testing.T) { + m := &manager{restoreSlotsByHypervisor: map[hypervisor.Type]chan struct{}{ + hypervisor.TypeFirecracker: make(chan struct{}, 1), + }} + + release, err := m.acquireRestoreSlot(context.Background(), hypervisor.TypeFirecracker) + if err != nil { + t.Fatalf("first acquire failed: %v", err) + } + defer release() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + if _, err := m.acquireRestoreSlot(ctx, hypervisor.TypeFirecracker); err == nil { + t.Fatalf("expected second firecracker acquire to wait until context expires") + } + + if releaseOther, err := m.acquireRestoreSlot(context.Background(), hypervisor.TypeCloudHypervisor); err != nil { + t.Fatalf("unconfigured hypervisor acquire should not wait: %v", err) + } else { + releaseOther() + } +} + +func TestRestoreSlotsSharedByHypervisorAndLimit(t *testing.T) { + slots := sharedRestoreSlots(hypervisor.TypeFirecracker, 1) + m1 := &manager{restoreSlotsByHypervisor: map[hypervisor.Type]chan struct{}{ + hypervisor.TypeFirecracker: slots, + }} + m2 := &manager{restoreSlotsByHypervisor: map[hypervisor.Type]chan struct{}{ + hypervisor.TypeFirecracker: slots, + }} + + release, err := m1.acquireRestoreSlot(context.Background(), hypervisor.TypeFirecracker) + if err != nil { + t.Fatalf("first acquire failed: %v", err) + } + defer release() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + if _, err := m2.acquireRestoreSlot(ctx, hypervisor.TypeFirecracker); err == nil { + t.Fatalf("expected second manager to share restore capacity") + } +} diff --git a/lib/instances/snapshot.go b/lib/instances/snapshot.go index eac2d008..17f19b98 100644 --- a/lib/instances/snapshot.go +++ b/lib/instances/snapshot.go @@ -367,6 +367,7 @@ func (m *manager) restoreSnapshot(ctx context.Context, id string, snapshotID str } func (m *manager) forkSnapshot(ctx context.Context, snapshotID string, req ForkSnapshotRequest) (*Instance, error) { + log := logger.FromContext(ctx) if err := validateForkSnapshotRequest(req); err != nil { return nil, err } @@ -412,15 +413,8 @@ func (m *manager) forkSnapshot(ctx context.Context, snapshotID string, req ForkS if target != nil && target.State == compressionJobStateRunning { m.recordSnapshotCompressionPreemption(ctx, snapshotCompressionPreemptionForkSnapshot, target.Target) } - if err := m.ensureSnapshotMemoryReady(ctx, m.paths.SnapshotGuestDir(snapshotID), "", rec.StoredMetadata.HypervisorType); err != nil { - return nil, fmt.Errorf("prepare snapshot memory for fork: %w", err) - } - - if err := forkvm.CopyGuestDirectory(m.paths.SnapshotGuestDir(snapshotID), dstDir); err != nil { - if errors.Is(err, forkvm.ErrSparseCopyUnsupported) { - return nil, fmt.Errorf("fork from snapshot requires sparse-capable filesystem (SEEK_DATA/SEEK_HOLE unsupported): %w", err) - } - return nil, fmt.Errorf("clone snapshot payload: %w", err) + if err := m.copySnapshotGuestDirectoryForFork(ctx, snapshotID, rec.StoredMetadata.HypervisorType, dstDir); err != nil { + return nil, err } starter, err := m.getVMStarter(targetHypervisor) @@ -470,7 +464,7 @@ func (m *manager) forkSnapshot(ctx context.Context, snapshotID string, req ForkS if forkMeta.NetworkEnabled { netCfg = &hypervisor.ForkNetworkConfig{TAPDevice: network.GenerateTAPName(forkID)} } - if _, err := starter.PrepareFork(ctx, hypervisor.ForkPrepareRequest{ + prepareResult, err := prepareForkWithAliasReadLock(ctx, starter, hypervisor.ForkPrepareRequest{ SnapshotConfigPath: m.paths.InstanceSnapshotConfig(forkID), SourceDataDir: rec.StoredMetadata.DataDir, TargetDataDir: forkMeta.DataDir, @@ -478,12 +472,21 @@ func (m *manager) forkSnapshot(ctx context.Context, snapshotID string, req ForkS VsockSocket: forkMeta.VsockSocket, SerialLogPath: m.paths.InstanceAppLog(forkID), Network: netCfg, - }); err != nil { + }) + if err != nil { if errors.Is(err, hypervisor.ErrNotSupported) { return nil, fmt.Errorf("%w: snapshot fork is not supported for hypervisor %s", ErrNotSupported, targetHypervisor) } return nil, fmt.Errorf("prepare snapshot fork state: %w", err) } + if prepareResult.RequiresSnapshotSourceAlias { + log.WarnContext(ctx, "snapshot fork restore requires snapshot source alias fallback", + "hypervisor", targetHypervisor, + "snapshot_id", snapshotID, + "fork_instance_id", forkID, + "source_data_dir", rec.StoredMetadata.DataDir, + "target_data_dir", forkMeta.DataDir) + } } if err := m.saveMetadata(&metadata{StoredMetadata: forkMeta}); err != nil { @@ -502,7 +505,8 @@ func (m *manager) forkSnapshot(ctx context.Context, snapshotID string, req ForkS } func (m *manager) copySnapshotPayload(sourceInstanceID, snapshotGuestDir string) error { - if err := forkvm.CopyGuestDirectory(m.paths.InstanceDir(sourceInstanceID), snapshotGuestDir); err != nil { + err := copyGuestDirectoryWithAliasReadLock(m.paths.InstanceDir(sourceInstanceID), snapshotGuestDir) + if err != nil { if errors.Is(err, forkvm.ErrSparseCopyUnsupported) { return fmt.Errorf("snapshot requires sparse-capable filesystem (SEEK_DATA/SEEK_HOLE unsupported): %w", err) } diff --git a/lib/instances/snapshot_alias_lock.go b/lib/instances/snapshot_alias_lock.go new file mode 100644 index 00000000..b6942c95 --- /dev/null +++ b/lib/instances/snapshot_alias_lock.go @@ -0,0 +1,64 @@ +package instances + +import ( + "context" + "errors" + "fmt" + + "github.com/kernel/hypeman/lib/forkvm" + "github.com/kernel/hypeman/lib/hypervisor" +) + +func withSnapshotSourceAliasReadLock(run func() error) error { + unlockAliasReaders := hypervisor.LockSnapshotSourceAliasReaders() + defer unlockAliasReaders() + return run() +} + +func prepareForkWithAliasReadLock(ctx context.Context, starter hypervisor.VMStarter, req hypervisor.ForkPrepareRequest) (hypervisor.ForkPrepareResult, error) { + var result hypervisor.ForkPrepareResult + err := withSnapshotSourceAliasReadLock(func() error { + var err error + result, err = starter.PrepareFork(ctx, req) + return err + }) + return result, err +} + +func copyGuestDirectoryWithAliasReadLock(srcDir, dstDir string) error { + return withSnapshotSourceAliasReadLock(func() error { + return forkvm.CopyGuestDirectory(srcDir, dstDir) + }) +} + +func (m *manager) copyForkSourceGuestDirectory(ctx context.Context, sourceState State, sourceID string, stored *StoredMetadata, srcDir, dstDir string) error { + if sourceState == StateStandby { + if err := m.ensureSnapshotMemoryReady(ctx, m.paths.InstanceSnapshotLatest(sourceID), m.snapshotJobKeyForInstance(sourceID), stored.HypervisorType); err != nil { + return fmt.Errorf("prepare standby snapshot for fork: %w", err) + } + } + return withSnapshotSourceAliasReadLock(func() error { + if err := forkvm.CopyGuestDirectory(srcDir, dstDir); err != nil { + if errors.Is(err, forkvm.ErrSparseCopyUnsupported) { + return fmt.Errorf("fork requires sparse-capable filesystem (SEEK_DATA/SEEK_HOLE unsupported): %w", err) + } + return fmt.Errorf("clone guest directory: %w", err) + } + return nil + }) +} + +func (m *manager) copySnapshotGuestDirectoryForFork(ctx context.Context, snapshotID string, hvType hypervisor.Type, dstDir string) error { + if err := m.ensureSnapshotMemoryReady(ctx, m.paths.SnapshotGuestDir(snapshotID), "", hvType); err != nil { + return fmt.Errorf("prepare snapshot memory for fork: %w", err) + } + return withSnapshotSourceAliasReadLock(func() error { + if err := forkvm.CopyGuestDirectory(m.paths.SnapshotGuestDir(snapshotID), dstDir); err != nil { + if errors.Is(err, forkvm.ErrSparseCopyUnsupported) { + return fmt.Errorf("fork from snapshot requires sparse-capable filesystem (SEEK_DATA/SEEK_HOLE unsupported): %w", err) + } + return fmt.Errorf("clone snapshot payload: %w", err) + } + return nil + }) +} diff --git a/lib/instances/snapshot_compression.go b/lib/instances/snapshot_compression.go index 61d1ebff..9f230a63 100644 --- a/lib/instances/snapshot_compression.go +++ b/lib/instances/snapshot_compression.go @@ -11,6 +11,7 @@ import ( "os/exec" "path/filepath" "strings" + "sync" "syscall" "time" @@ -745,6 +746,12 @@ func (m *manager) cancelAndWaitCompressionJob(ctx context.Context, key string) ( func (m *manager) ensureSnapshotMemoryReady(ctx context.Context, snapshotDir, jobKey string, hvType hypervisor.Type) error { start := time.Now() + return m.withSnapshotMemoryPrepareLock(snapshotDir, func() error { + return m.ensureSnapshotMemoryReadyWithLock(ctx, snapshotDir, jobKey, hvType, start) + }) +} + +func (m *manager) ensureSnapshotMemoryReadyWithLock(ctx context.Context, snapshotDir, jobKey string, hvType hypervisor.Type, start time.Time) error { log := logger.FromContext(ctx) if jobKey != "" { @@ -779,6 +786,15 @@ func (m *manager) ensureSnapshotMemoryReady(ctx context.Context, snapshotDir, jo return nil } +func (m *manager) withSnapshotMemoryPrepareLock(snapshotDir string, run func() error) error { + lockKey := filepath.Clean(snapshotDir) + lock, _ := m.snapshotPrepareLocks.LoadOrStore(lockKey, &sync.Mutex{}) + mu := lock.(*sync.Mutex) + mu.Lock() + defer mu.Unlock() + return run() +} + func (m *manager) updateSnapshotCompressionMetadata(snapshotID, state, compressionError string, cfg *snapshotstore.SnapshotCompressionConfig, compressedSize, uncompressedSize *int64) error { rec, err := m.loadSnapshotRecord(snapshotID) if err != nil { diff --git a/lib/instances/snapshot_compression_test.go b/lib/instances/snapshot_compression_test.go index 86a61e48..540bfbda 100644 --- a/lib/instances/snapshot_compression_test.go +++ b/lib/instances/snapshot_compression_test.go @@ -433,6 +433,51 @@ func newSnapshotCompressionTestManager(t *testing.T) *manager { } } +func TestSnapshotMemoryPrepareLockSerializesSameSnapshotDir(t *testing.T) { + t.Parallel() + + mgr := newSnapshotCompressionTestManager(t) + snapshotDir := filepath.Join(t.TempDir(), "snapshot") + firstEntered := make(chan struct{}) + releaseFirst := make(chan struct{}) + firstDone := make(chan error, 1) + go func() { + firstDone <- mgr.withSnapshotMemoryPrepareLock(snapshotDir, func() error { + close(firstEntered) + <-releaseFirst + return nil + }) + }() + <-firstEntered + + secondStarted := make(chan struct{}) + secondEntered := make(chan struct{}) + secondDone := make(chan error, 1) + go func() { + close(secondStarted) + secondDone <- mgr.withSnapshotMemoryPrepareLock(filepath.Join(snapshotDir, "."), func() error { + close(secondEntered) + return nil + }) + }() + <-secondStarted + + select { + case <-secondEntered: + t.Fatal("second prepare entered while first prepare held the same snapshot lock") + case <-time.After(25 * time.Millisecond): + } + + close(releaseFirst) + require.NoError(t, <-firstDone) + require.NoError(t, <-secondDone) + select { + case <-secondEntered: + default: + t.Fatal("second prepare did not enter after first prepare released the snapshot lock") + } +} + func TestStartCompressionJobDelayedCancellationRecordsSkipped(t *testing.T) { t.Parallel() diff --git a/lib/instances/storage.go b/lib/instances/storage.go index 776c6124..f33a5962 100644 --- a/lib/instances/storage.go +++ b/lib/instances/storage.go @@ -11,6 +11,7 @@ import ( "github.com/kernel/hypeman/lib/autostandby" "github.com/kernel/hypeman/lib/healthcheck" + "github.com/kernel/hypeman/lib/hypervisor" "github.com/kernel/hypeman/lib/images" ) @@ -61,6 +62,9 @@ func (m *manager) ensureDirectories(id string) error { // loadMetadata loads instance metadata from disk func (m *manager) loadMetadata(id string) (*metadata, error) { + unlockAliasReaders := hypervisor.LockSnapshotSourceAliasReaders() + defer unlockAliasReaders() + metaPath := m.paths.InstanceMetadata(id) data, err := os.ReadFile(metaPath) @@ -81,6 +85,9 @@ func (m *manager) loadMetadata(id string) (*metadata, error) { // saveMetadata saves instance metadata to disk func (m *manager) saveMetadata(meta *metadata) error { + unlockAliasReaders := hypervisor.LockSnapshotSourceAliasReaders() + defer unlockAliasReaders() + metaPath := m.paths.InstanceMetadata(meta.Id) data, err := json.MarshalIndent(meta, "", " ") diff --git a/lib/providers/providers.go b/lib/providers/providers.go index 33a909d5..090d2ccf 100644 --- a/lib/providers/providers.go +++ b/lib/providers/providers.go @@ -146,6 +146,9 @@ func ProvideInstanceManager(p *paths.Paths, cfg *config.Config, imageManager ima LifecycleEventBufferSize: cfg.Instances.LifecycleEventBufferSize, FirecrackerSnapshotMemoryBackend: cfg.Hypervisor.FirecrackerSnapshotMemoryBackend, FirecrackerUFFDCacheMaxBytes: int64(firecrackerUFFDCacheMaxBytes), + MaxConcurrentRestoresByHypervisor: map[hypervisor.Type]int{ + hypervisor.TypeFirecracker: cfg.Hypervisor.FirecrackerMaxConcurrentRestores, + }, } return instances.NewManagerWithConfigE(p, imageManager, systemManager, networkManager, deviceManager, volumeManager, limits, defaultHypervisor, snapshotDefaults, managerConfig, meter, tracer, memoryPolicy) } diff --git a/scripts/check-uffd-version.sh b/scripts/check-uffd-version.sh index da3c8eea..b5934583 100755 --- a/scripts/check-uffd-version.sh +++ b/scripts/check-uffd-version.sh @@ -16,7 +16,6 @@ fi changed_runtime_files="$( git diff --name-only "${diff_range}" -- \ lib/uffdpager \ - lib/hypervisor/firecracker/process.go \ lib/hypervisor/firecracker/config.go \ lib/instances/firecracker_uffd.go \ lib/instances/guest_resume_network.go \ @@ -25,6 +24,21 @@ changed_runtime_files="$( grep -Ev '(^lib/uffdpager/VERSION$|(^|/)README\.md$|_test\.go$)' || true )" +# process.go contains both UFFD restore wiring and unrelated Firecracker restore +# orchestration. Only require a pager version bump when changed lines touch the +# UFFD-facing restore contract. +process_uffd_changes="$( + git diff --name-only -G 'uffd|UFFD|SnapshotMemory' "${diff_range}" -- \ + lib/hypervisor/firecracker/process.go || true +)" +if [ -n "${process_uffd_changes}" ]; then + changed_runtime_files="$( + printf '%s\n%s\n' "${changed_runtime_files}" "${process_uffd_changes}" | + sed '/^$/d' | + sort -u + )" +fi + if [ -z "${changed_runtime_files}" ]; then exit 0 fi