diff --git a/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go b/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go index 2c9e509665..0c93d627af 100644 --- a/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go +++ b/cmd/ateapi/internal/controlapi/functionaltest/actor_test.go @@ -1910,12 +1910,133 @@ func TestResumeActorPassesLiteralEnv(t *testing.T) { } } -// TestResumeActor_NoWorkers tests that resuming an actor fails when no free workers are available. -// Workflow: -// 1. Creates a mock ActorTemplate. -// 2. Creates an actor. -// 3. Calls ResumeActor RPC without creating any workers. -// 4. Verifies that ResumeActor fails with FailedPrecondition status. +// createGoldenDataTemplate creates "tmpl1" like createTemplate, but with +// onCommit DATA and onResume.fromData GOLDEN, so a resumed-after-suspend +// actor takes the DATA_ON_GOLDEN path: its data snapshot combined with the +// template's golden. +func createGoldenDataTemplate(t *testing.T, tc *testContext, ns string) *ateapipb.ActorTemplate { + t.Helper() + ensureDefaultGvisorSandboxConfig(t, tc) + createWorkerPool(t, tc, ns, "pool1", map[string]string{poolLabelKey: ns}) + + created, err := tc.client.CreateActorTemplate(context.Background(), &ateapipb.CreateActorTemplateRequest{ + ActorTemplate: &ateapipb.ActorTemplate{ + Metadata: &ateapipb.ResourceMetadata{ + Atespace: testAtespace, + Name: "tmpl1", + }, + SnapshotsConfig: &ateapipb.SnapshotsConfig{ + StorageLocation: testStorageLocation, + OnPause: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL, + OnCommit: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_DATA, + OnResume: &ateapipb.OnResumeConfig{FromData: ateapipb.ResumeSource_RESUME_SOURCE_GOLDEN}, + }, + SandboxConfig: &ateapipb.SandboxConfig{ + SandboxClass: ateapipb.SandboxClass_SANDBOX_CLASS_GVISOR, + ConfigName: "gvisor-default", + }, + Containers: []*ateapipb.Container{{ + Name: "main", + Image: "main@sha256:abc", + Command: []string{"/main"}, + }}, + WorkerSelector: &ateapipb.Selector{ + MatchLabels: map[string]string{poolLabelKey: ns}, + }, + }, + }) + if err != nil { + t.Fatalf("failed to create actor template: %v", err) + } + updated, err := tc.persistence.UpdateActorTemplate(context.Background(), + resources.ActorTemplateRefFromActorTemplate(created), store.PreconditionFrom(created), + func(dbTemplate *ateapipb.ActorTemplate) error { + dbTemplate.Status = &ateapipb.ActorTemplateStatus{ + GoldenSnapshotStatus: &ateapipb.GoldenSnapshotStatus{ + GoldenSnapshot: &ateapipb.ExternalSnapshot{SnapshotUri: goldenSnapshotURI(t), ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL}, + }, + } + return nil + }) + if err != nil { + t.Fatalf("failed to record the template's golden snapshot: %v", err) + } + return updated +} + +// TestResumeActor_GoldenDataResumeSetsBaseConfig drives the DATA_ON_GOLDEN +// resume end to end and pins the wire request's base snapshot fields: while +// the golden_snapshot_uri -> base_config transition lasts, ateapi sets both +// and they must agree, so ateapi and atelet can roll in either order. +func TestResumeActor_GoldenDataResumeSetsBaseConfig(t *testing.T) { + ns := namespaceForTest("ns-resume-golden-data") + tc := setupTest(t, ns) + defer tc.cleanup() + + tmpl := createGoldenDataTemplate(t, tc, ns) + workerName := createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1") + + const name = "id1" + actorRef := &ateapipb.ObjectRef{Atespace: testAtespace, Name: name} + if _, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: name}, + ActorTemplate: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "tmpl1"}, + }}); err != nil { + t.Fatalf("CreateActor failed: %v", err) + } + + // First resume runs fresh from the golden — a shared snapshot, and the + // request must say so (atelet caches only declared-SHARED sources). + if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{Actor: actorRef}); err != nil { + t.Fatalf("ResumeActor (first) failed: %v", err) + } + golden := tmpl.GetStatus().GetGoldenSnapshotStatus().GetGoldenSnapshot().GetSnapshotUri() + freshReq := tc.fakeAtelet.lastRestoreRequest() + if got := freshReq.GetExternalConfig().GetSnapshotUri(); got != golden { + t.Errorf("fresh resume restore uri = %q, want the template's golden %q", got, golden) + } + if got := freshReq.GetExternalConfig().GetSharing(); got != ateletpb.SnapshotSharing_SNAPSHOT_SHARING_SHARED { + t.Errorf("fresh-from-golden restore sharing = %v, want SNAPSHOT_SHARING_SHARED", got) + } + + // The suspend then commits a DATA snapshot per onCommit. + suspended, err := tc.client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{Actor: actorRef}) + if err != nil { + t.Fatalf("SuspendActor failed: %v", err) + } + waitForWorkerAvailable(t, tc, workerName) + actorSnapshotURI := suspended.GetActor().GetStatus().GetExternalSnapshot().GetSnapshotUri() + if actorSnapshotURI == "" { + t.Fatal("SuspendActor recorded no external snapshot") + } + + // Second resume: the actor's DATA snapshot rides on the template's + // golden. + if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{Actor: actorRef}); err != nil { + t.Fatalf("ResumeActor (second) failed: %v", err) + } + restoreReq := tc.fakeAtelet.lastRestoreRequest() + if restoreReq == nil { + t.Fatal("second resume sent no Restore request to atelet") + } + if got := restoreReq.GetScope(); got != ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN { + t.Fatalf("restore scope = %v, want SNAPSHOT_SCOPE_DATA_ON_GOLDEN", got) + } + if got := restoreReq.GetExternalConfig().GetSnapshotUri(); got != actorSnapshotURI { + t.Errorf("restore config snapshot uri = %q, want the actor's data snapshot %q", got, actorSnapshotURI) + } + // The actor's own data snapshot is private: nothing else ever reads it. + if got := restoreReq.GetExternalConfig().GetSharing(); got != ateletpb.SnapshotSharing_SNAPSHOT_SHARING_PRIVATE { + t.Errorf("data-on-golden restore sharing = %v, want SNAPSHOT_SHARING_PRIVATE", got) + } + if got := restoreReq.GetBaseConfig().GetSnapshotUri(); got != golden { + t.Errorf("restore base_config uri = %q, want the template's golden %q", got, golden) + } + if got := restoreReq.GetGoldenSnapshotUri(); got != golden { + t.Errorf("restore golden_snapshot_uri = %q, want %q (transitional dual-write must match base_config)", got, golden) + } +} + // TestResumeActor_NoWorkers tests that resuming an actor fails when no free workers are available. // Workflow: // 1. Creates a mock ActorTemplate. @@ -2476,6 +2597,9 @@ func TestResumeActor_RepointTemplateBeforeResume(t *testing.T) { if got := restoreReq.GetGoldenSnapshotUri(); got != "" { t.Errorf("restore request to atelet had golden snapshot uri = %q, want empty", got) } + if restoreReq.GetBaseConfig() != nil { + t.Errorf("restore request to atelet had base_config = %v, want unset", restoreReq.GetBaseConfig()) + } }) } } diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index f1bcdbc2a2..7bc6b5b7f1 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -700,6 +700,10 @@ func (w *ActorWorkflow) ensureAteletRestored(ctx context.Context, actorRef resou req.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA case !src.GoldenSnapshotURI.IsZero(): req.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN + // Transitional dual-write: base_config supersedes + // golden_snapshot_uri, but an atelet from before it reads only + // the old field. Dropped once both components have rolled. + req.BaseConfig = &ateletpb.ExternalRestoreConfiguration{SnapshotUri: src.GoldenSnapshotURI.String()} req.GoldenSnapshotUri = src.GoldenSnapshotURI.String() default: req.Scope = actorSnapshotContentScopeToAtelet(actorTemplate.GetSnapshotsConfig().GetOnPause()) @@ -718,16 +722,29 @@ func (w *ActorWorkflow) ensureAteletRestored(ctx context.Context, actorRef resou } var scope ateletpb.SnapshotScope var goldenSnapshotURI string + var baseConfig *ateletpb.ExternalRestoreConfiguration switch { case src.TemplateReplaced: scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA case !src.GoldenSnapshotURI.IsZero(): scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN + // Transitional dual-write: base_config supersedes + // golden_snapshot_uri, but an atelet from before it reads only + // the old field. Dropped once both components have rolled. + baseConfig = &ateletpb.ExternalRestoreConfiguration{SnapshotUri: src.GoldenSnapshotURI.String()} goldenSnapshotURI = src.GoldenSnapshotURI.String() default: scope = actorSnapshotContentScopeToAtelet(src.Scope) } tele.WireSnapshotScope = ateattr.SnapshotScopeValue(scope) + // The control plane owns cacheability: a fresh start reads the + // template's golden snapshot, which is shared across actors and + // immutable once published, while an actor's own snapshot is private + // to it. atelet caches only what is declared SHARED. + sharing := ateletpb.SnapshotSharing_SNAPSHOT_SHARING_PRIVATE + if tele.SnapshotKind == ateattr.SnapshotKindGolden { + sharing = ateletpb.SnapshotSharing_SNAPSHOT_SHARING_SHARED + } req := &ateletpb.RestoreRequest{ TargetAteomUid: assignment.GetWorkerPodUid(), Atespace: actor.GetMetadata().GetAtespace(), @@ -737,12 +754,14 @@ func (w *ActorWorkflow) ensureAteletRestored(ctx context.Context, actorRef resou Spec: workloadSpec, Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL, Config: &ateletpb.RestoreRequest_ExternalConfig{ - ExternalConfig: &ateletpb.ExternalCheckpointConfiguration{ + ExternalConfig: &ateletpb.ExternalRestoreConfiguration{ SnapshotUri: src.SnapshotURI.String(), + Sharing: sharing, }, }, Scope: scope, - // Empty unless this is a Golden data resume. + // Both empty unless this is a Golden data resume. + BaseConfig: baseConfig, GoldenSnapshotUri: goldenSnapshotURI, ActorUid: actor.GetMetadata().Uid, EgressGateway: egressGateway, diff --git a/cmd/atelet/imagegc.go b/cmd/atelet/imagegc.go index 4eb8b63f3d..559d9fea57 100644 --- a/cmd/atelet/imagegc.go +++ b/cmd/atelet/imagegc.go @@ -29,10 +29,7 @@ import ( "errors" "fmt" "log/slog" - "os" - "path/filepath" "runtime/debug" - "strings" "time" "github.com/agent-substrate/substrate/internal/ateompath" @@ -76,7 +73,10 @@ func validateImageCacheGCFlags() error { // future), making just-pulled layers evictable. return fmt.Errorf("--image-cache-min-age %v must be >= 0", *imageCacheMinAge) } - if imageCacheDirOutsideBasePath(*imageCacheDir) { + // Warn-worthy, not an error: a separate cache volume is legitimate + // (recommended for IOPS), but the watermarks then measure a different + // volume than actor state. + if !ateompath.UnderBasePath(*imageCacheDir) { slog.Warn("Image cache dir is outside the ateom base path; its volume watermarks are measured separately from actor state", slog.String("image_cache_dir", *imageCacheDir), slog.String("actors_dir", ateompath.ActorsDir)) @@ -84,18 +84,6 @@ func validateImageCacheGCFlags() error { return nil } -// imageCacheDirOutsideBasePath reports whether the cache dir is outside -// the ateom base path — the watermarks then measure a different volume -// than actor state. Warn-worthy, not an error: a separate cache volume is -// legitimate (recommended for IOPS). -func imageCacheDirOutsideBasePath(dir string) bool { - abs, err := filepath.Abs(dir) - if err != nil { - abs = filepath.Clean(dir) - } - return !strings.HasPrefix(abs, ateompath.BasePath+string(os.PathSeparator)) -} - // imageCacheGCTarget computes the bytes a pass should free: the larger // of the watermark shortfall (kubelet's formula — usage at highPct frees // down to lowPct) and the pool's overage past maxBytes, but never more diff --git a/cmd/atelet/imagegc_test.go b/cmd/atelet/imagegc_test.go index 99df8d3a33..c1d73edf45 100644 --- a/cmd/atelet/imagegc_test.go +++ b/cmd/atelet/imagegc_test.go @@ -25,7 +25,6 @@ import ( "testing" "time" - "github.com/agent-substrate/substrate/internal/ateompath" "github.com/agent-substrate/substrate/internal/imagecache" ) @@ -154,30 +153,6 @@ func TestValidateImageCacheGCFlags(t *testing.T) { } } -func TestImageCacheDirOutsideBasePath(t *testing.T) { - cases := []struct { - name string - dir string - want bool - }{ - {"inside", filepath.Join(ateompath.BasePath, "image-cache"), false}, - {"inside with doubled separator", ateompath.BasePath + "//image-cache", false}, - {"inside via dot-dot", ateompath.BasePath + "/x/../image-cache", false}, - {"base path itself is not inside", ateompath.BasePath, true}, - {"sibling with the base path as name prefix", ateompath.BasePath + "-other/image-cache", true}, - {"outside", "/var/lib/elsewhere/image-cache", true}, - {"dot-dot escaping the base path", ateompath.BasePath + "/../elsewhere/image-cache", true}, - {"relative resolves against the cwd, not the base path", "image-cache", true}, - } - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - if got := imageCacheDirOutsideBasePath(tc.dir); got != tc.want { - t.Errorf("imageCacheDirOutsideBasePath(%q) = %v, want %v", tc.dir, got, tc.want) - } - }) - } -} - type fakeGCStore struct { size int64 sizeErr error diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 904c207d5a..b996773604 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -34,6 +34,7 @@ import ( "sync" "github.com/agent-substrate/substrate/cmd/atelet/internal/ategcs" + "github.com/agent-substrate/substrate/cmd/atelet/internal/filecache" "github.com/agent-substrate/substrate/cmd/atelet/internal/sparsefile" "github.com/agent-substrate/substrate/internal/actorlog" "github.com/agent-substrate/substrate/internal/ateapiauth" @@ -217,6 +218,14 @@ func main() { go newImageCacheGC(imageCache, *imageCacheDir).Run(ctx) } + if err := validateSnapshotCacheFlags(); err != nil { + serverboot.Fatal(ctx, "Invalid snapshot cache flags", err) + } + snapshotCache, err := openSnapshotCache(ctx, *snapshotCacheDir, *snapshotCacheMinAge) + if err != nil { + serverboot.Fatal(ctx, "Failed to open snapshot cache", err) + } + wrappedAnonGCS, err := ategcs.NewGCSClient(ctx, option.WithoutAuthentication()) if err != nil { serverboot.Fatal(ctx, "Failed to create anonymous GCS client", err) @@ -291,6 +300,7 @@ func main() { wrappedAnonGCS, wrappedGCS, imageCache, + snapshotCache, instruments, volPlugins, csiDriverConfigLister, @@ -431,8 +441,12 @@ func drainOnShutdown(ctx context.Context, srv *grpc.Server, readiness *serverboo type AteomHerder struct { ateletpb.UnimplementedAteomHerderServer - ateomDialer *AteomDialer - imageCache *imagecache.Store + ateomDialer *AteomDialer + imageCache *imagecache.Store + // snapshotCache dedupes and retains shared snapshot files across + // restores. nil means caching is disabled (--snapshot-cache-dir=""): + // every restore downloads its snapshot files directly. + snapshotCache *filecache.Store anonGCSClient ategcs.ObjectStorage gcsClient ategcs.ObjectStorage instruments *Instruments @@ -451,6 +465,7 @@ func NewService( anonGCSClient ategcs.ObjectStorage, gcsClient ategcs.ObjectStorage, imageCache *imagecache.Store, + snapshotCache *filecache.Store, instruments *Instruments, volumePlugins map[string]volume.VolumePluginWorkerPlane, csiDriverConfigLister listersv1alpha1.CSIDriverConfigLister, @@ -459,6 +474,7 @@ func NewService( wms := &AteomHerder{ ateomDialer: ateomDialer, imageCache: imageCache, + snapshotCache: snapshotCache, anonGCSClient: anonGCSClient, gcsClient: gcsClient, instruments: instruments, @@ -1072,9 +1088,10 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) // and its pinned sandbox binaries are the ones that will run the restored // guest (the golden snapshot's memory image must be resumed by the binaries // that created it). + baseCfg := restoreBaseConfig(req) var goldenRec *sandboxAssetsRecord if req.GetScope() == ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN { - goldenURI, err := resources.ParseSnapshotURI(req.GetGoldenSnapshotUri()) + goldenURI, err := resources.ParseSnapshotURI(baseCfg.GetSnapshotUri()) if err != nil { return nil, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidObjectURL) } @@ -1137,17 +1154,26 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) dDownload = time.Since(t) downloadErr = err }() + // How cacheable snapshot files may be served depends on the sandbox + // class (see cacheModeFor); for DATA_ON_GOLDEN the golden's class + // matches sandboxRec's (validated above). + classMode := cacheModeFor(sandboxRec.SandboxClass) switch req.GetType() { case ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL: if req.GetScope() == ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN { if goldenRec == nil { return fmt.Errorf("no golden snapshot record for a %s restore", req.GetScope()) } - if err := s.downloadCombinedCheckpoint(gctx, req.GetExternalConfig().GetSnapshotUri(), req.GetGoldenSnapshotUri(), checkpointDir, sandboxRec.SnapshotFiles, goldenRec.SnapshotFiles); err != nil { + if err := s.downloadCombinedCheckpoint(gctx, req.GetExternalConfig().GetSnapshotUri(), baseCfg.GetSnapshotUri(), checkpointDir, sandboxRec.SnapshotFiles, goldenRec.SnapshotFiles, classMode); err != nil { + return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonFailedGetExternalObject, ateerrors.ReasonInvalidObjectURL, ateerrors.ReasonTerminalFileSystemError) + } + } else { + // Cache-eligible only when the control plane declared the + // snapshot shared (a fresh-from-golden start); an actor's + // own snapshot downloads fresh. + if err := s.downloadExternalCheckpoint(gctx, req.GetExternalConfig().GetSnapshotUri(), checkpointDir, sandboxRec.SnapshotFiles, externalConfigCacheMode(req, classMode)); err != nil { return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonFailedGetExternalObject, ateerrors.ReasonInvalidObjectURL, ateerrors.ReasonTerminalFileSystemError) } - } else if err := s.downloadExternalCheckpoint(gctx, req.GetExternalConfig().GetSnapshotUri(), checkpointDir, sandboxRec.SnapshotFiles); err != nil { - return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonFailedGetExternalObject, ateerrors.ReasonInvalidObjectURL, ateerrors.ReasonTerminalFileSystemError) } case ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL: combineWithGolden := req.GetScope() == ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN @@ -1166,7 +1192,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) }) if combineWithGolden { gLocal.Go(func() error { - if err := s.downloadExternalCheckpoint(gLocalCtx, req.GetGoldenSnapshotUri(), checkpointDir, goldenOnlyFiles(sandboxRec.SnapshotFiles, goldenRec.SnapshotFiles)); err != nil { + if err := s.downloadExternalCheckpoint(gLocalCtx, baseCfg.GetSnapshotUri(), checkpointDir, goldenOnlyFiles(sandboxRec.SnapshotFiles, goldenRec.SnapshotFiles), classMode); err != nil { return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonFailedGetExternalObject, ateerrors.ReasonInvalidObjectURL, ateerrors.ReasonTerminalFileSystemError) } return nil @@ -1242,7 +1268,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) // Informational: for DATA_ON_GOLDEN the golden snapshot's files are // already staged into the restore dir by the combined download above; // ateom restores from the shared dir and never fetches this URI. - GoldenSnapshotUri: req.GetGoldenSnapshotUri(), + GoldenSnapshotUri: baseCfg.GetSnapshotUri(), }) dAteom = time.Since(tAteom) if err != nil { @@ -1373,18 +1399,23 @@ func goldenOnlyFiles(actorFiles, goldenFiles []string) []string { // as a single folder: every file of the actor's own snapshot (the durable-dir // data) plus the golden snapshot's files the actor's set does not shadow, so // the result looks like a Full snapshot whose durable-dir data is the actor's. -func (s *AteomHerder) downloadCombinedCheckpoint(ctx context.Context, actorURI, goldenURI, dstDir string, actorFiles, goldenFiles []string) error { +// Only the golden leg is cache-eligible (classMode): the actor's own files +// are used by this one actor and deleted on its next suspend. +func (s *AteomHerder) downloadCombinedCheckpoint(ctx context.Context, actorURI, goldenURI, dstDir string, actorFiles, goldenFiles []string, classMode cacheMode) error { g, gctx := errgroup.WithContext(ctx) g.Go(func() error { - return s.downloadExternalCheckpoint(gctx, actorURI, dstDir, actorFiles) + return s.downloadExternalCheckpoint(gctx, actorURI, dstDir, actorFiles, cacheModeOff) }) g.Go(func() error { - return s.downloadExternalCheckpoint(gctx, goldenURI, dstDir, goldenOnlyFiles(actorFiles, goldenFiles)) + return s.downloadExternalCheckpoint(gctx, goldenURI, dstDir, goldenOnlyFiles(actorFiles, goldenFiles), classMode) }) return g.Wait() } -func (s *AteomHerder) downloadExternalCheckpoint(ctx context.Context, snapshotURI string, dstDir string, files []string) error { +// downloadExternalCheckpoint stages files of one snapshot into dstDir, +// through the snapshot cache per mode (see cacheMode). A caller passing +// anything but cacheModeOff asserts the snapshot is immutable at its URI. +func (s *AteomHerder) downloadExternalCheckpoint(ctx context.Context, snapshotURI string, dstDir string, files []string, mode cacheMode) error { uri, err := resources.ParseSnapshotURI(snapshotURI) if err != nil { return err @@ -1398,7 +1429,7 @@ func (s *AteomHerder) downloadExternalCheckpoint(ctx context.Context, snapshotUR if err != nil { return fmt.Errorf("while addressing %s in GCS: %w", fileName, err) } - if err := ategcs.FetchLocalFileFromGCSWithZstd(gCtx, s.gcsClient, objectURI, local); err != nil { + if err := s.fetchSnapshotObject(gCtx, objectURI, local, mode); err != nil { return fmt.Errorf("while downloading %s from GCS: %w", fileName, err) } return nil @@ -1750,14 +1781,33 @@ func validateRestoreRequest(req *ateletpb.RestoreRequest) error { } // A DATA_ON_GOLDEN restore needs both halves: the actor's data snapshot - // (local pause checkpoint or external commit) and the golden snapshot, - // which is always external. + // (local pause checkpoint or external commit) and the base snapshot, + // which is always external. base_config supersedes golden_snapshot_uri; + // a transitional caller sets both, and they must agree. + base, legacy := req.GetBaseConfig().GetSnapshotUri(), req.GetGoldenSnapshotUri() + if base != "" && legacy != "" && base != legacy { + return fmt.Errorf("base_config.snapshot_uri %q and golden_snapshot_uri %q disagree", base, legacy) + } if req.GetScope() == ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN { - if _, err := resources.ParseSnapshotURI(req.GetGoldenSnapshotUri()); err != nil { - return fmt.Errorf("invalid golden_snapshot_uri: %w", err) + if _, err := resources.ParseSnapshotURI(restoreBaseConfig(req).GetSnapshotUri()); err != nil { + return fmt.Errorf("invalid base snapshot URI: %w", err) } - } else if req.GetGoldenSnapshotUri() != "" { - return fmt.Errorf("golden_snapshot_uri is only valid with snapshot scope %s", ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN) + } else if base != "" || legacy != "" { + return fmt.Errorf("a base snapshot (base_config or golden_snapshot_uri) is only valid with snapshot scope %s", ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN) + } + return nil +} + +// restoreBaseConfig returns the base snapshot source of a DATA_ON_GOLDEN +// restore, preferring base_config over the superseded golden_snapshot_uri +// (still sent by callers that predate it). Nil when the request carries +// neither; proto getters make that safe to read through. +func restoreBaseConfig(req *ateletpb.RestoreRequest) *ateletpb.ExternalRestoreConfiguration { + if req.GetBaseConfig().GetSnapshotUri() != "" { + return req.GetBaseConfig() + } + if uri := req.GetGoldenSnapshotUri(); uri != "" { + return &ateletpb.ExternalRestoreConfiguration{SnapshotUri: uri} } return nil } diff --git a/cmd/atelet/main_test.go b/cmd/atelet/main_test.go index 6d8aa061f7..ea03ba8b8c 100644 --- a/cmd/atelet/main_test.go +++ b/cmd/atelet/main_test.go @@ -227,7 +227,7 @@ func validRestoreRequest() *ateletpb.RestoreRequest { Spec: &ateletpb.WorkloadSpec{Containers: []*ateletpb.Container{{Name: "worker"}}}, Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL, Config: &ateletpb.RestoreRequest_ExternalConfig{ - ExternalConfig: &ateletpb.ExternalCheckpointConfiguration{ + ExternalConfig: &ateletpb.ExternalRestoreConfiguration{ SnapshotUri: testSnapshotURI, }, }, @@ -395,6 +395,29 @@ func TestValidateRestoreRequest(t *testing.T) { {"golden uri with non-data-on-golden scope", makeReq(func(r *ateletpb.RestoreRequest) { r.GoldenSnapshotUri = goldenSnapshotURI }), true}, + {"data-on-golden with base config only", makeReq(func(r *ateletpb.RestoreRequest) { + r.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN + r.BaseConfig = &ateletpb.ExternalRestoreConfiguration{SnapshotUri: goldenSnapshotURI} + }), false}, + // A transitional caller sets base_config and the superseded + // golden_snapshot_uri together; they must name one snapshot. + {"data-on-golden with agreeing base config and golden uri", makeReq(func(r *ateletpb.RestoreRequest) { + r.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN + r.BaseConfig = &ateletpb.ExternalRestoreConfiguration{SnapshotUri: goldenSnapshotURI} + r.GoldenSnapshotUri = goldenSnapshotURI + }), false}, + {"base config and golden uri disagree", makeReq(func(r *ateletpb.RestoreRequest) { + r.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN + r.BaseConfig = &ateletpb.ExternalRestoreConfiguration{SnapshotUri: goldenSnapshotURI} + r.GoldenSnapshotUri = testSnapshotURI + }), true}, + {"data-on-golden with bucketless base config", makeReq(func(r *ateletpb.RestoreRequest) { + r.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN + r.BaseConfig = &ateletpb.ExternalRestoreConfiguration{SnapshotUri: "relative/path"} + }), true}, + {"base config with non-data-on-golden scope", makeReq(func(r *ateletpb.RestoreRequest) { + r.BaseConfig = &ateletpb.ExternalRestoreConfiguration{SnapshotUri: goldenSnapshotURI} + }), true}, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { @@ -405,6 +428,33 @@ func TestValidateRestoreRequest(t *testing.T) { } } +// TestRestoreBaseConfig pins the dual-read precedence during the +// golden_snapshot_uri -> base_config transition: base_config wins when it +// names a snapshot, the legacy field covers callers that predate it, and a +// request with neither yields nil (safe through proto getters). +func TestRestoreBaseConfig(t *testing.T) { + cases := []struct { + name string + base *ateletpb.ExternalRestoreConfiguration + legacy string + wantURI string + }{ + {"neither set", nil, "", ""}, + {"base config only", &ateletpb.ExternalRestoreConfiguration{SnapshotUri: goldenSnapshotURI}, "", goldenSnapshotURI}, + {"legacy only", nil, goldenSnapshotURI, goldenSnapshotURI}, + {"base config preferred over legacy", &ateletpb.ExternalRestoreConfiguration{SnapshotUri: goldenSnapshotURI}, testSnapshotURI, goldenSnapshotURI}, + {"empty base config falls back to legacy", &ateletpb.ExternalRestoreConfiguration{}, goldenSnapshotURI, goldenSnapshotURI}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + req := &ateletpb.RestoreRequest{BaseConfig: tc.base, GoldenSnapshotUri: tc.legacy} + if got := restoreBaseConfig(req).GetSnapshotUri(); got != tc.wantURI { + t.Errorf("restoreBaseConfig().GetSnapshotUri() = %q, want %q", got, tc.wantURI) + } + }) + } +} + // Every valid atelet scope must map to its ateom counterpart; in particular // DATA_ON_GOLDEN must never silently degrade to FULL. func TestToAteomSnapshotScope(t *testing.T) { @@ -939,27 +989,29 @@ func (m mapObjectStorage) GetObject(_ context.Context, bucket, object string) (i func (mapObjectStorage) PutObject(_ context.Context, _, _ string, _ io.Reader) error { return nil } +// zstdBytes compresses s the way snapshot objects are stored, so download +// paths can be tested against mapObjectStorage. +func zstdBytes(t *testing.T, s string) []byte { + t.Helper() + var buf bytes.Buffer + zw, err := zstd.NewWriter(&buf) + if err != nil { + t.Fatalf("zstd.NewWriter: %v", err) + } + if _, err := zw.Write([]byte(s)); err != nil { + t.Fatalf("zstd write: %v", err) + } + if err := zw.Close(); err != nil { + t.Fatalf("zstd close: %v", err) + } + return buf.Bytes() +} + // TestDownloadCombinedCheckpoint verifies a DataOnGolden restore stages one // folder holding the actor snapshot's durable-dir tar and the golden // snapshot's remaining files — and that the golden's own durable-dir tar is // the one that loses the name collision. func TestDownloadCombinedCheckpoint(t *testing.T) { - zstdBytes := func(t *testing.T, s string) []byte { - t.Helper() - var buf bytes.Buffer - zw, err := zstd.NewWriter(&buf) - if err != nil { - t.Fatalf("zstd.NewWriter: %v", err) - } - if _, err := zw.Write([]byte(s)); err != nil { - t.Fatalf("zstd write: %v", err) - } - if err := zw.Close(); err != nil { - t.Fatalf("zstd close: %v", err) - } - return buf.Bytes() - } - store := mapObjectStorage{objects: map[string][]byte{ testSnapshotPath + "/durable-dir.tar.zstd": zstdBytes(t, "actor durable data"), goldenSnapshotPath + "/config.json.zstd": zstdBytes(t, "golden config"), @@ -974,7 +1026,8 @@ func TestDownloadCombinedCheckpoint(t *testing.T) { goldenSnapshotURI, dstDir, []string{"durable-dir.tar"}, - []string{"config.json", "memory-ranges", "durable-dir.tar"}) + []string{"config.json", "memory-ranges", "durable-dir.tar"}, + cacheModeOff) if err != nil { t.Fatalf("downloadCombinedCheckpoint: %v", err) } diff --git a/cmd/atelet/snapshotcache.go b/cmd/atelet/snapshotcache.go new file mode 100644 index 0000000000..229c9928ab --- /dev/null +++ b/cmd/atelet/snapshotcache.go @@ -0,0 +1,170 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +// The snapshot file cache. +// +// A shared snapshot's files (a template's golden snapshot today, tag +// snapshots later) are immutable once published, yet every restore that +// needs them downloads them again into its own per-actor dir. This cache (a +// filecache.Store) makes those files a node-level resource: concurrent +// restores of one snapshot share a single download, and later restores are +// served from disk instead of object storage. +// +// Whether a snapshot is cacheable is the control plane's call, carried on +// the request as ExternalRestoreConfiguration.sharing; atelet only maps +// that property to a serving mode per sandbox class. This file owns the +// cache's lifecycle (flags, validation, opening with the startup debris +// sweep) and the restore path's cached reads. The eviction loop is wired +// separately. + +import ( + "context" + "errors" + "fmt" + "log/slog" + "syscall" + "time" + + "github.com/agent-substrate/substrate/cmd/atelet/internal/ategcs" + "github.com/agent-substrate/substrate/cmd/atelet/internal/filecache" + "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/internal/proto/ateletpb" + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" + "github.com/spf13/pflag" +) + +var ( + snapshotCacheDir = pflag.String("snapshot-cache-dir", ateompath.SnapshotCacheDir, "Directory for the node-local shared snapshot file cache. Empty disables caching (every restore downloads its snapshot files). Must be on the same filesystem mount as the actor state dirs: cache hits are served as hard links into the per-actor restore dirs.") + snapshotCacheMinAge = pflag.Duration("snapshot-cache-min-age", 10*time.Minute, "Cached snapshot files younger than this are never evicted, protecting files fetched but not yet linked into a restore dir.") +) + +func validateSnapshotCacheFlags() error { + if *snapshotCacheMinAge < 0 { + // A negative min-age inverts the veto (the cutoff lands in the + // future), making just-fetched files evictable mid-restore. + return fmt.Errorf("--snapshot-cache-min-age %v must be >= 0", *snapshotCacheMinAge) + } + if *snapshotCacheDir != "" && !ateompath.UnderBasePath(*snapshotCacheDir) { + slog.Warn("Snapshot cache dir is outside the ateom base path; hits cannot be hard-linked into restore dirs across mounts, so every hit degrades to a copy", + slog.String("snapshot_cache_dir", *snapshotCacheDir), + slog.String("actors_dir", ateompath.ActorsDir)) + } + return nil +} + +// openSnapshotCache opens the snapshot cache rooted at dir and clears crash +// debris before the store serves any restore. An empty dir disables +// caching: the returned store is nil and restores download their snapshot +// files directly. +func openSnapshotCache(ctx context.Context, dir string, minAge time.Duration) (*filecache.Store, error) { + if dir == "" { + slog.InfoContext(ctx, "Snapshot cache disabled; every restore downloads its snapshot files") + return nil, nil + } + store, err := filecache.New(dir, filecache.WithMinAge(minAge)) + if err != nil { + return nil, fmt.Errorf("while opening snapshot cache at %s: %w", dir, err) + } + stats, err := store.SweepDebris(ctx) + if err != nil { + // Leftover debris wastes space but affects no lookup, so the store + // is fully usable: log and carry on rather than failing startup. + slog.WarnContext(ctx, "Snapshot cache debris sweep incomplete", slog.Any("err", err)) + } + slog.InfoContext(ctx, "Snapshot cache open", + slog.String("dir", dir), + slog.Int("tmp_removed", stats.TmpRemoved), + slog.Int("retired_removed", stats.RetiredRemoved)) + return store, nil +} + +// cacheMode is how a restore may materialize snapshot files from the +// cache. The mode exists because the fast path — a hard link — shares the +// cached inode with the consumer, which is only safe when the consumer never +// writes the staged file in place. +type cacheMode int + +const ( + // cacheModeOff downloads fresh, bypassing the cache. + cacheModeOff cacheMode = iota + // cacheModeLink hard-links the cached copy: zero-cost hits, but the + // consumer must treat the staged file as read-only. + cacheModeLink + // cacheModeCopy stages a private, hole-preserving copy: costlier per hit + // than a link, but the consumer owns the inode and may mutate it in + // place, and a copy can cross mounts. + cacheModeCopy +) + +// cacheModeFor returns how this sandbox class's restores may use the +// snapshot cache. ateom-gvisor consumes restore-state strictly read-only, +// so it gets hard links. ateom-microvm rewrites config.json in place at +// restore and merges checkpoint deltas into memory-ranges' inode at suspend +// — either would corrupt a shared inode — so it gets private copies: still +// one download per snapshot per node, and its mutations stay its own. +func cacheModeFor(sandboxClass string) cacheMode { + switch atev1alpha1.SandboxClass(sandboxClass) { + case atev1alpha1.SandboxClassGvisor: + return cacheModeLink + case atev1alpha1.SandboxClassMicroVM: + return cacheModeCopy + default: + return cacheModeOff + } +} + +// externalConfigCacheMode returns the cache mode for the request's external +// snapshot: classMode when the control plane declared the snapshot shared, +// cacheModeOff otherwise (private snapshots, and callers that predate the +// sharing field). +func externalConfigCacheMode(req *ateletpb.RestoreRequest, classMode cacheMode) cacheMode { + if req.GetExternalConfig().GetSharing() == ateletpb.SnapshotSharing_SNAPSHOT_SHARING_SHARED { + return classMode + } + return cacheModeOff +} + +// fetchSnapshotObject stages one snapshot object at local — through the +// snapshot cache per mode when the cache is enabled, directly from object +// storage otherwise. A cacheModeLink hit is a hard link, so local stays +// valid regardless of later eviction; a cacheModeCopy hit is a private +// copy. Fetch errors pass through the cache wrapped, keeping ateerrors +// classification intact. +func (s *AteomHerder) fetchSnapshotObject(ctx context.Context, objectURI, local string, mode cacheMode) error { + if mode == cacheModeOff || s.snapshotCache == nil { + return ategcs.FetchLocalFileFromGCSWithZstd(ctx, s.gcsClient, objectURI, local) + } + key := filecache.URIKey("gcs-zstd", objectURI) + fetch := func(ctx context.Context, dst string) error { + return ategcs.FetchLocalFileFromGCSWithZstd(ctx, s.gcsClient, objectURI, dst) + } + if mode == cacheModeCopy { + return s.snapshotCache.GetFileCopyTo(ctx, key, local, fetch) + } + err := s.snapshotCache.GetFileTo(ctx, key, local, fetch) + if err == nil || !errors.Is(err, syscall.EXDEV) { + return err + } + // The cache sits on a different mount than the restore dir, so link-out + // cannot work — but a copy can, and the failed link's flight already + // published the entry, so this is a local read rather than a second + // download. The warning points at the misconfiguration (a cache under + // --snapshot-cache-dir off the base-path mount serves every hit the slow + // way). + slog.WarnContext(ctx, "Snapshot cache is on a different filesystem than the restore dir; serving a copy instead of a hard link", + slog.String("object", objectURI)) + return s.snapshotCache.GetFileCopyTo(ctx, key, local, fetch) +} diff --git a/cmd/atelet/snapshotcache_test.go b/cmd/atelet/snapshotcache_test.go new file mode 100644 index 0000000000..db0408e5ca --- /dev/null +++ b/cmd/atelet/snapshotcache_test.go @@ -0,0 +1,385 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "context" + "fmt" + "io" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/agent-substrate/substrate/cmd/atelet/internal/filecache" + "github.com/agent-substrate/substrate/internal/proto/ateletpb" + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" +) + +func TestValidateSnapshotCacheFlags(t *testing.T) { + origDir, origMinAge := *snapshotCacheDir, *snapshotCacheMinAge + t.Cleanup(func() { *snapshotCacheDir, *snapshotCacheMinAge = origDir, origMinAge }) + + cases := []struct { + name string + dir string + minAge time.Duration + wantErr bool + }{ + {"defaults", origDir, 10 * time.Minute, false}, + {"disabled", "", 10 * time.Minute, false}, + {"zero min-age", origDir, 0, false}, + {"outside base path warns but is valid", "/var/lib/elsewhere/snapshot-cache", 10 * time.Minute, false}, + {"negative min-age inverts the veto", origDir, -time.Second, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + *snapshotCacheDir, *snapshotCacheMinAge = tc.dir, tc.minAge + err := validateSnapshotCacheFlags() + if (err != nil) != tc.wantErr { + t.Errorf("dir=%q minAge=%v: err=%v, wantErr=%v", tc.dir, tc.minAge, err, tc.wantErr) + } + }) + } +} + +func TestOpenSnapshotCacheDisabled(t *testing.T) { + store, err := openSnapshotCache(context.Background(), "", 10*time.Minute) + if err != nil { + t.Fatalf("openSnapshotCache(\"\") error: %v", err) + } + if store != nil { + t.Errorf("openSnapshotCache(\"\") = %v, want nil (disabled)", store) + } +} + +func TestOpenSnapshotCacheCreatesRoot(t *testing.T) { + dir := filepath.Join(t.TempDir(), "nested", "snapshot-cache") + store, err := openSnapshotCache(context.Background(), dir, 10*time.Minute) + if err != nil { + t.Fatalf("openSnapshotCache(%q) error: %v", dir, err) + } + if store == nil { + t.Fatal("openSnapshotCache returned a nil store for a non-empty dir") + } + if fi, err := os.Stat(dir); err != nil || !fi.IsDir() { + t.Errorf("cache root %s not created: %v", dir, err) + } +} + +// TestExternalConfigCacheMode pins the caching gate: only a snapshot the +// control plane declared shared is served from the cache. Private, +// unclassified (an older ateapi), and absent external configs all download +// fresh — atelet never infers cacheability from the URI. +func TestExternalConfigCacheMode(t *testing.T) { + withSharing := func(s ateletpb.SnapshotSharing) *ateletpb.RestoreRequest { + return &ateletpb.RestoreRequest{Config: &ateletpb.RestoreRequest_ExternalConfig{ + ExternalConfig: &ateletpb.ExternalRestoreConfiguration{SnapshotUri: goldenSnapshotURI, Sharing: s}, + }} + } + cases := []struct { + name string + req *ateletpb.RestoreRequest + want cacheMode + }{ + {"shared uses the class mode", withSharing(ateletpb.SnapshotSharing_SNAPSHOT_SHARING_SHARED), cacheModeLink}, + {"private is never cached", withSharing(ateletpb.SnapshotSharing_SNAPSHOT_SHARING_PRIVATE), cacheModeOff}, + {"unspecified (older caller) is never cached", withSharing(ateletpb.SnapshotSharing_SNAPSHOT_SHARING_UNSPECIFIED), cacheModeOff}, + {"no external config", &ateletpb.RestoreRequest{}, cacheModeOff}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := externalConfigCacheMode(tc.req, cacheModeLink); got != tc.want { + t.Errorf("externalConfigCacheMode() = %v, want %v", got, tc.want) + } + }) + } +} + +func TestCacheModeFor(t *testing.T) { + if got := cacheModeFor(string(atev1alpha1.SandboxClassGvisor)); got != cacheModeLink { + t.Errorf("gVisor consumes restore-state read-only; want cacheModeLink, got %v", got) + } + // ateom-microvm rewrites config.json and memory-ranges in place; linking + // its restore files from the cache would corrupt shared copies, so it + // gets private ones. + if got := cacheModeFor(string(atev1alpha1.SandboxClassMicroVM)); got != cacheModeCopy { + t.Errorf("micro-VM mutates restore-state files in place; want cacheModeCopy, got %v", got) + } + if got := cacheModeFor(""); got != cacheModeOff { + t.Errorf("unknown sandbox class: want cacheModeOff, got %v", got) + } +} + +// countingObjectStorage counts GetObject calls, so tests can pin how many +// downloads a fetch sequence actually performed. +type countingObjectStorage struct { + mapObjectStorage + gets atomic.Int32 +} + +func (c *countingObjectStorage) GetObject(ctx context.Context, bucket, object string) (io.ReadCloser, error) { + c.gets.Add(1) + return c.mapObjectStorage.GetObject(ctx, bucket, object) +} + +// newSnapshotCacheHerder returns a herder whose gcsClient serves the given +// content for one golden object, plus the object's URI and a directory (on +// the same filesystem as the cache) for link destinations. +func newSnapshotCacheHerder(t *testing.T, content string) (*AteomHerder, *countingObjectStorage, string, string) { + t.Helper() + base := t.TempDir() + store, err := filecache.New(filepath.Join(base, "cache")) + if err != nil { + t.Fatal(err) + } + dstDir := filepath.Join(base, "restore") + if err := os.MkdirAll(dstDir, 0o755); err != nil { + t.Fatal(err) + } + gcs := &countingObjectStorage{mapObjectStorage: mapObjectStorage{objects: map[string][]byte{ + goldenSnapshotPath + "/memory-ranges.zstd": zstdBytes(t, content), + }}} + s := &AteomHerder{gcsClient: gcs, snapshotCache: store} + return s, gcs, goldenSnapshotURI + "/memory-ranges.zstd", dstDir +} + +func TestFetchSnapshotObjectServesFromCache(t *testing.T) { + s, gcs, objectURI, dstDir := newSnapshotCacheHerder(t, "golden memory") + ctx := context.Background() + + // Two cached fetches: one download, both destinations share the inode. + first := filepath.Join(dstDir, "first") + second := filepath.Join(dstDir, "second") + for _, dst := range []string{first, second} { + if err := s.fetchSnapshotObject(ctx, objectURI, dst, cacheModeLink); err != nil { + t.Fatalf("fetchSnapshotObject(%s): %v", dst, err) + } + if got, err := os.ReadFile(dst); err != nil || string(got) != "golden memory" { + t.Fatalf("staged %s = %q, %v", dst, got, err) + } + } + if n := gcs.gets.Load(); n != 1 { + t.Errorf("two cached fetches performed %d downloads, want 1", n) + } + fi1, err1 := os.Stat(first) + fi2, err2 := os.Stat(second) + if err1 != nil || err2 != nil || !os.SameFile(fi1, fi2) { + t.Errorf("cached destinations do not share an inode (%v, %v)", err1, err2) + } + + // A non-cacheable fetch downloads fresh and shares nothing. + direct := filepath.Join(dstDir, "direct") + if err := s.fetchSnapshotObject(ctx, objectURI, direct, cacheModeOff); err != nil { + t.Fatalf("direct fetchSnapshotObject: %v", err) + } + if n := gcs.gets.Load(); n != 2 { + t.Errorf("direct fetch did not download (gets=%d, want 2)", n) + } + if fi3, err := os.Stat(direct); err != nil || os.SameFile(fi1, fi3) { + t.Errorf("direct destination shares the cache inode (%v)", err) + } +} + +// TestFetchSnapshotObjectCopyMode pins the micro-VM serving mode: one +// download per golden object, every destination a private writable inode, +// and a consumer's in-place mutation never reaches the next consumer. +func TestFetchSnapshotObjectCopyMode(t *testing.T) { + s, gcs, objectURI, dstDir := newSnapshotCacheHerder(t, "golden memory") + ctx := context.Background() + + first := filepath.Join(dstDir, "first") + second := filepath.Join(dstDir, "second") + for _, dst := range []string{first, second} { + if err := s.fetchSnapshotObject(ctx, objectURI, dst, cacheModeCopy); err != nil { + t.Fatalf("fetchSnapshotObject(%s): %v", dst, err) + } + } + if n := gcs.gets.Load(); n != 1 { + t.Errorf("two copy-mode fetches performed %d downloads, want 1", n) + } + fi1, err1 := os.Stat(first) + fi2, err2 := os.Stat(second) + if err1 != nil || err2 != nil { + t.Fatal(err1, err2) + } + if os.SameFile(fi1, fi2) { + t.Error("copy-mode destinations share an inode; each consumer must own its own") + } + + // The consumer may rewrite its staged file (ateom-microvm does, to + // config.json) without poisoning what the next restore receives. + if err := os.WriteFile(first, []byte("rewritten by ateom"), 0o600); err != nil { + t.Fatalf("consumer write to its copy: %v", err) + } + third := filepath.Join(dstDir, "third") + if err := s.fetchSnapshotObject(ctx, objectURI, third, cacheModeCopy); err != nil { + t.Fatal(err) + } + if got, err := os.ReadFile(third); err != nil || string(got) != "golden memory" { + t.Errorf("copy after consumer mutation = %q, %v; the cached bytes were poisoned", got, err) + } + if n := gcs.gets.Load(); n != 1 { + t.Errorf("gets=%d, want 1 (mutation must not invalidate the cache)", n) + } +} + +// TestFetchSnapshotObjectLinkFallsBackToCopyAcrossMounts needs two real +// filesystems, so it runs where one is available (/dev/shm on Linux) and +// skips elsewhere: a link-mode fetch whose destination is on a different +// mount than the cache must degrade to a private copy, not to a download. +func TestFetchSnapshotObjectLinkFallsBackToCopyAcrossMounts(t *testing.T) { + s, gcs, objectURI, dstDir := newSnapshotCacheHerder(t, "golden memory") + ctx := context.Background() + + otherFS, err := os.MkdirTemp("/dev/shm", "snapshot-cache-test-") + if err != nil { + t.Skipf("no second filesystem available for a cross-mount test: %v", err) + } + t.Cleanup(func() { os.RemoveAll(otherFS) }) + // Prove the two dirs really are different mounts; same-mount tmpdirs + // (some CI images) would silently test the plain link path. + probe := filepath.Join(dstDir, "probe") + if err := os.WriteFile(probe, nil, 0o600); err != nil { + t.Fatal(err) + } + if err := os.Link(probe, filepath.Join(otherFS, "probe")); err == nil { + t.Skip("test dirs share a filesystem; cannot exercise EXDEV") + } + + dst := filepath.Join(otherFS, "staged") + if err := s.fetchSnapshotObject(ctx, objectURI, dst, cacheModeLink); err != nil { + t.Fatalf("cross-mount link-mode fetch: %v", err) + } + if got, err := os.ReadFile(dst); err != nil || string(got) != "golden memory" { + t.Fatalf("staged %q, %v", got, err) + } + if n := gcs.gets.Load(); n != 1 { + t.Errorf("cross-mount fallback performed %d downloads, want 1 (the copy must reuse the published entry)", n) + } + + // And the fallback stays cache-served: a second cross-mount fetch is a + // local copy, not a download. + dst2 := filepath.Join(otherFS, "staged-2") + if err := s.fetchSnapshotObject(ctx, objectURI, dst2, cacheModeLink); err != nil { + t.Fatal(err) + } + if n := gcs.gets.Load(); n != 1 { + t.Errorf("second cross-mount fetch downloaded again (gets=%d, want 1)", n) + } +} + +func TestFetchSnapshotObjectWithoutCacheDownloads(t *testing.T) { + s, gcs, objectURI, dstDir := newSnapshotCacheHerder(t, "golden memory") + s.snapshotCache = nil // caching disabled (--snapshot-cache-dir="") + + dst := filepath.Join(dstDir, "out") + if err := s.fetchSnapshotObject(context.Background(), objectURI, dst, cacheModeLink); err != nil { + t.Fatalf("fetchSnapshotObject with nil cache: %v", err) + } + if got, err := os.ReadFile(dst); err != nil || string(got) != "golden memory" { + t.Fatalf("staged content = %q, %v", got, err) + } + if n := gcs.gets.Load(); n != 1 { + t.Errorf("gets=%d, want 1", n) + } +} + +func TestFetchSnapshotObjectFetchErrorReachesCaller(t *testing.T) { + s, _, _, dstDir := newSnapshotCacheHerder(t, "golden memory") + dst := filepath.Join(dstDir, "out") + err := s.fetchSnapshotObject(context.Background(), goldenSnapshotURI+"/missing.zstd", dst, cacheModeLink) + if err == nil { + t.Fatal("fetchSnapshotObject succeeded for a missing object") + } + if _, statErr := os.Stat(dst); !os.IsNotExist(statErr) { + t.Errorf("failed fetch left a destination file: %v", statErr) + } +} + +// TestDownloadExternalCheckpointSharesOneGoldenDownload pins M2's exit +// criterion at the unit level: concurrent restores staging the same golden +// snapshot perform its downloads once. +func TestDownloadExternalCheckpointSharesOneGoldenDownload(t *testing.T) { + s, gcs, _, dstDir := newSnapshotCacheHerder(t, "golden memory") + files := []string{"memory-ranges"} + + const restores = 4 + errs := make([]error, restores) + var wg sync.WaitGroup + for i := range restores { + dir := filepath.Join(dstDir, fmt.Sprintf("actor-%d", i)) + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + wg.Go(func() { + errs[i] = s.downloadExternalCheckpoint(context.Background(), goldenSnapshotURI, dir, files, cacheModeLink) + }) + } + wg.Wait() + for i, err := range errs { + if err != nil { + t.Fatalf("restore %d: %v", i, err) + } + got, err := os.ReadFile(filepath.Join(dstDir, fmt.Sprintf("actor-%d", i), "memory-ranges")) + if err != nil || string(got) != "golden memory" { + t.Fatalf("restore %d staged %q, %v", i, got, err) + } + } + if n := gcs.gets.Load(); n != 1 { + t.Errorf("%d concurrent restores performed %d downloads, want 1", restores, n) + } +} + +// TestOpenSnapshotCacheSweepsDebris pins the startup ordering contract: crash +// debris (unfinished fetches in tmp/, interrupted evictions as .rm-*) is gone +// by the time openSnapshotCache returns, while published entries survive. +func TestOpenSnapshotCacheSweepsDebris(t *testing.T) { + dir := t.TempDir() + // A prior life's layout: one published entry, plus debris of both kinds. + entryDir := filepath.Join(dir, "entries", "aaaa") + for _, d := range []string{ + filepath.Join(dir, "tmp", "aaaa-12345"), + filepath.Join(dir, ".rm-bbbb-xyz"), + entryDir, + } { + if err := os.MkdirAll(d, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(d, "data"), []byte("x"), 0o644); err != nil { + t.Fatal(err) + } + } + + store, err := openSnapshotCache(context.Background(), dir, 10*time.Minute) + if err != nil { + t.Fatalf("openSnapshotCache error: %v", err) + } + if store == nil { + t.Fatal("openSnapshotCache returned a nil store for a non-empty dir") + } + + if children, err := os.ReadDir(filepath.Join(dir, "tmp")); err != nil || len(children) != 0 { + t.Errorf("tmp debris not swept: %d children, %v", len(children), err) + } + if _, err := os.Stat(filepath.Join(dir, ".rm-bbbb-xyz")); !os.IsNotExist(err) { + t.Errorf("retired debris not swept: %v", err) + } + if _, err := os.Stat(filepath.Join(entryDir, "data")); err != nil { + t.Errorf("published entry did not survive the sweep: %v", err) + } +} diff --git a/internal/ateompath/ateompath.go b/internal/ateompath/ateompath.go index 401dd313b6..f15afa07e1 100644 --- a/internal/ateompath/ateompath.go +++ b/internal/ateompath/ateompath.go @@ -16,7 +16,9 @@ package ateompath import ( + "os" "path/filepath" + "strings" ) const ( @@ -40,11 +42,33 @@ var ( // under it. ActorsDir = filepath.Join(BasePath, "actors") + // SnapshotCacheDir is the node-local cache of shared snapshot files — + // golden snapshots today, tag snapshots later (see + // cmd/atelet/internal/filecache). It lives under BasePath because cache + // hits are served as hard links into the per-actor restore dirs under + // ActorsDir, and link(2) requires both ends on one mounted filesystem — + // the same mount, not merely the same disk. + SnapshotCacheDir = filepath.Join(BasePath, "snapshot-cache") + // CredentialBrokerSocket is the node-local atelet socket used by atunnel // to request credentials for the worker's current actor assignment. CredentialBrokerSocket = filepath.Join(BasePath, "credential-broker.sock") ) +// UnderBasePath reports whether dir resolves to a directory strictly under +// BasePath (BasePath itself does not count). A relative dir resolves against +// the process working directory. Callers use it to detect a cache directory +// configured off the shared mount, where consequences are theirs to judge: +// the image cache's GC watermarks then measure a different volume than actor +// state, and the golden cache cannot hard-link hits into restore dirs at all. +func UnderBasePath(dir string) bool { + abs, err := filepath.Abs(dir) + if err != nil { + abs = filepath.Clean(dir) + } + return strings.HasPrefix(abs, BasePath+string(os.PathSeparator)) +} + func RunSCBinaryPath(sha256 string) string { return filepath.Join(StaticFilesDir, "runsc-"+sha256) } diff --git a/internal/ateompath/ateompath_test.go b/internal/ateompath/ateompath_test.go index d4b372ba01..6c6b972e2c 100644 --- a/internal/ateompath/ateompath_test.go +++ b/internal/ateompath/ateompath_test.go @@ -69,6 +69,41 @@ func TestAteletOTLPSocketPath(t *testing.T) { } } +func TestUnderBasePath(t *testing.T) { + cases := []struct { + name string + dir string + want bool + }{ + {"inside", BasePath + "/image-cache", true}, + {"inside with doubled separator", BasePath + "//image-cache", true}, + {"inside via dot-dot", BasePath + "/x/../image-cache", true}, + {"base path itself is not under", BasePath, false}, + {"sibling with the base path as name prefix", BasePath + "-other/image-cache", false}, + {"outside", "/var/lib/elsewhere/image-cache", false}, + {"dot-dot escaping the base path", BasePath + "/../elsewhere/image-cache", false}, + {"relative resolves against the cwd, not the base path", "image-cache", false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := UnderBasePath(tc.dir); got != tc.want { + t.Errorf("UnderBasePath(%q) = %v, want %v", tc.dir, got, tc.want) + } + }) + } +} + +func TestSnapshotCacheDirSharesActorsMount(t *testing.T) { + // The snapshot cache serves hits as hard links into per-actor restore + // dirs, and link(2) requires one filesystem, so both must sit under + // BasePath (one mount in the atelet pod). + for name, dir := range map[string]string{"SnapshotCacheDir": SnapshotCacheDir, "ActorsDir": ActorsDir} { + if !strings.HasPrefix(dir, BasePath+"/") { + t.Errorf("%s = %q, want it under %q so snapshot cache hard links can reach restore dirs", name, dir, BasePath) + } + } +} + func TestAteomPathUniqueness(t *testing.T) { uid1 := "123e4567-e89b-12d3-a456-426614174000" uid2 := "987f6543-e21b-32d1-b654-246614174111" diff --git a/internal/proto/ateletpb/atelet.pb.go b/internal/proto/ateletpb/atelet.pb.go index 9a82d70afa..405bf62601 100644 --- a/internal/proto/ateletpb/atelet.pb.go +++ b/internal/proto/ateletpb/atelet.pb.go @@ -205,6 +205,63 @@ func (SnapshotScope) EnumDescriptor() ([]byte, []int) { return file_atelet_proto_rawDescGZIP(), []int{2} } +// SnapshotSharing is how an external snapshot's storage may be consumed, as +// known by the control plane. +type SnapshotSharing int32 + +const ( + // An older caller that predates the field: treated as PRIVATE. + SnapshotSharing_SNAPSHOT_SHARING_UNSPECIFIED SnapshotSharing = 0 + // Read by a single actor and deleted or replaced by its next suspend. + // Never cached. + SnapshotSharing_SNAPSHOT_SHARING_PRIVATE SnapshotSharing = 1 + // Immutable once published and read by many actors (today: a template's + // golden snapshot). atelet may serve its files from the node-local + // snapshot cache. + SnapshotSharing_SNAPSHOT_SHARING_SHARED SnapshotSharing = 2 +) + +// Enum value maps for SnapshotSharing. +var ( + SnapshotSharing_name = map[int32]string{ + 0: "SNAPSHOT_SHARING_UNSPECIFIED", + 1: "SNAPSHOT_SHARING_PRIVATE", + 2: "SNAPSHOT_SHARING_SHARED", + } + SnapshotSharing_value = map[string]int32{ + "SNAPSHOT_SHARING_UNSPECIFIED": 0, + "SNAPSHOT_SHARING_PRIVATE": 1, + "SNAPSHOT_SHARING_SHARED": 2, + } +) + +func (x SnapshotSharing) Enum() *SnapshotSharing { + p := new(SnapshotSharing) + *p = x + return p +} + +func (x SnapshotSharing) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (SnapshotSharing) Descriptor() protoreflect.EnumDescriptor { + return file_atelet_proto_enumTypes[3].Descriptor() +} + +func (SnapshotSharing) Type() protoreflect.EnumType { + return &file_atelet_proto_enumTypes[3] +} + +func (x SnapshotSharing) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use SnapshotSharing.Descriptor instead. +func (SnapshotSharing) EnumDescriptor() ([]byte, []int) { + return file_atelet_proto_rawDescGZIP(), []int{3} +} + type SetWorkerCapacityRequest struct { state protoimpl.MessageState `protogen:"open.v1"` // What the worker can supply, in the same vocabulary the control plane @@ -2094,6 +2151,66 @@ func (x *ExternalCheckpointConfiguration) GetSnapshotUri() string { return "" } +// ExternalRestoreConfiguration is an external snapshot a restore reads. +// Split from the checkpoint-side ExternalCheckpointConfiguration (a write +// destination) so read-side attributes of a restore source have a home. +type ExternalRestoreConfiguration struct { + state protoimpl.MessageState `protogen:"open.v1"` + // The object storage URI of the snapshot to read. Object names are + // appended to it, so it addresses the snapshot as a whole rather than any + // one object. Must stay field 1: old peers decode this message as + // ExternalCheckpointConfiguration. + SnapshotUri string `protobuf:"bytes,1,opt,name=snapshot_uri,json=snapshotUri,proto3" json:"snapshot_uri,omitempty"` + // How the snapshot at snapshot_uri may be consumed. + Sharing SnapshotSharing `protobuf:"varint,2,opt,name=sharing,proto3,enum=atelet.SnapshotSharing" json:"sharing,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExternalRestoreConfiguration) Reset() { + *x = ExternalRestoreConfiguration{} + mi := &file_atelet_proto_msgTypes[32] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExternalRestoreConfiguration) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExternalRestoreConfiguration) ProtoMessage() {} + +func (x *ExternalRestoreConfiguration) ProtoReflect() protoreflect.Message { + mi := &file_atelet_proto_msgTypes[32] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExternalRestoreConfiguration.ProtoReflect.Descriptor instead. +func (*ExternalRestoreConfiguration) Descriptor() ([]byte, []int) { + return file_atelet_proto_rawDescGZIP(), []int{32} +} + +func (x *ExternalRestoreConfiguration) GetSnapshotUri() string { + if x != nil { + return x.SnapshotUri + } + return "" +} + +func (x *ExternalRestoreConfiguration) GetSharing() SnapshotSharing { + if x != nil { + return x.Sharing + } + return SnapshotSharing_SNAPSHOT_SHARING_UNSPECIFIED +} + type CheckpointRequest struct { state protoimpl.MessageState `protogen:"open.v1"` TargetAteomUid string `protobuf:"bytes,1,opt,name=target_ateom_uid,json=targetAteomUid,proto3" json:"target_ateom_uid,omitempty"` @@ -2122,7 +2239,7 @@ type CheckpointRequest struct { func (x *CheckpointRequest) Reset() { *x = CheckpointRequest{} - mi := &file_atelet_proto_msgTypes[32] + mi := &file_atelet_proto_msgTypes[33] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2134,7 +2251,7 @@ func (x *CheckpointRequest) String() string { func (*CheckpointRequest) ProtoMessage() {} func (x *CheckpointRequest) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[32] + mi := &file_atelet_proto_msgTypes[33] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2147,7 +2264,7 @@ func (x *CheckpointRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use CheckpointRequest.ProtoReflect.Descriptor instead. func (*CheckpointRequest) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{32} + return file_atelet_proto_rawDescGZIP(), []int{33} } func (x *CheckpointRequest) GetTargetAteomUid() string { @@ -2262,7 +2379,7 @@ type CheckpointResponse struct { func (x *CheckpointResponse) Reset() { *x = CheckpointResponse{} - mi := &file_atelet_proto_msgTypes[33] + mi := &file_atelet_proto_msgTypes[34] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2274,7 +2391,7 @@ func (x *CheckpointResponse) String() string { func (*CheckpointResponse) ProtoMessage() {} func (x *CheckpointResponse) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[33] + mi := &file_atelet_proto_msgTypes[34] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2287,7 +2404,7 @@ func (x *CheckpointResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use CheckpointResponse.ProtoReflect.Descriptor instead. func (*CheckpointResponse) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{33} + return file_atelet_proto_rawDescGZIP(), []int{34} } type UploadPausedCheckpointRequest struct { @@ -2315,7 +2432,7 @@ type UploadPausedCheckpointRequest struct { func (x *UploadPausedCheckpointRequest) Reset() { *x = UploadPausedCheckpointRequest{} - mi := &file_atelet_proto_msgTypes[34] + mi := &file_atelet_proto_msgTypes[35] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2327,7 +2444,7 @@ func (x *UploadPausedCheckpointRequest) String() string { func (*UploadPausedCheckpointRequest) ProtoMessage() {} func (x *UploadPausedCheckpointRequest) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[34] + mi := &file_atelet_proto_msgTypes[35] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2340,7 +2457,7 @@ func (x *UploadPausedCheckpointRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use UploadPausedCheckpointRequest.ProtoReflect.Descriptor instead. func (*UploadPausedCheckpointRequest) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{34} + return file_atelet_proto_rawDescGZIP(), []int{35} } func (x *UploadPausedCheckpointRequest) GetAtespace() string { @@ -2407,7 +2524,7 @@ type UploadPausedCheckpointResponse struct { func (x *UploadPausedCheckpointResponse) Reset() { *x = UploadPausedCheckpointResponse{} - mi := &file_atelet_proto_msgTypes[35] + mi := &file_atelet_proto_msgTypes[36] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2419,7 +2536,7 @@ func (x *UploadPausedCheckpointResponse) String() string { func (*UploadPausedCheckpointResponse) ProtoMessage() {} func (x *UploadPausedCheckpointResponse) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[35] + mi := &file_atelet_proto_msgTypes[36] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2432,7 +2549,7 @@ func (x *UploadPausedCheckpointResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use UploadPausedCheckpointResponse.ProtoReflect.Descriptor instead. func (*UploadPausedCheckpointResponse) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{35} + return file_atelet_proto_rawDescGZIP(), []int{36} } type RestoreRequest struct { @@ -2457,12 +2574,10 @@ type RestoreRequest struct { Config isRestoreRequest_Config `protobuf_oneof:"config"` // What content to restore from the checkpoint. Scope SnapshotScope `protobuf:"varint,11,opt,name=scope,proto3,enum=atelet.SnapshotScope" json:"scope,omitempty"` - // The object storage URI of the ActorTemplate's golden snapshot. - // Set only when scope is SNAPSHOT_SCOPE_DATA_ON_GOLDEN: restore combines - // the golden snapshot (memory + full fs delta) with the durable data in - // the snapshot referenced by `config`. A top-level field rather than part - // of the `config` oneof: the actor's snapshot may be local (a pause - // checkpoint) while the golden snapshot is always external. + // Superseded by base_config. During the transition callers set both and + // atelet prefers base_config, so ateapi and atelet can roll in either + // order; removed (and reserved) in a follow-up once both sides have + // rolled. GoldenSnapshotUri string `protobuf:"bytes,12,opt,name=golden_snapshot_uri,json=goldenSnapshotUri,proto3" json:"golden_snapshot_uri,omitempty"` // When absent, actor traffic uses direct egress instead of atunnel. EgressGateway *EgressGateway `protobuf:"bytes,13,opt,name=egress_gateway,json=egressGateway,proto3,oneof" json:"egress_gateway,omitempty"` @@ -2470,15 +2585,19 @@ type RestoreRequest struct { // gVisor and micro-VM DATA-scope restores the sandbox is (re)sized to these; // for a FULL micro-VM restore the size baked into the snapshot wins. Zero // means "unset": keep the runtime default. - CpuMilli int64 `protobuf:"varint,14,opt,name=cpu_milli,json=cpuMilli,proto3" json:"cpu_milli,omitempty"` // CPU limit in millicores (1000 = one core). - MemoryBytes int64 `protobuf:"varint,15,opt,name=memory_bytes,json=memoryBytes,proto3" json:"memory_bytes,omitempty"` // Memory limit in bytes. + CpuMilli int64 `protobuf:"varint,14,opt,name=cpu_milli,json=cpuMilli,proto3" json:"cpu_milli,omitempty"` // CPU limit in millicores (1000 = one core). + MemoryBytes int64 `protobuf:"varint,15,opt,name=memory_bytes,json=memoryBytes,proto3" json:"memory_bytes,omitempty"` // Memory limit in bytes. + // The base guest state (memory + rootfs delta) combined with `config`'s + // durable data when scope is SNAPSHOT_SCOPE_DATA_ON_GOLDEN. Supersedes + // golden_snapshot_uri. + BaseConfig *ExternalRestoreConfiguration `protobuf:"bytes,16,opt,name=base_config,json=baseConfig,proto3" json:"base_config,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } func (x *RestoreRequest) Reset() { *x = RestoreRequest{} - mi := &file_atelet_proto_msgTypes[36] + mi := &file_atelet_proto_msgTypes[37] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2490,7 +2609,7 @@ func (x *RestoreRequest) String() string { func (*RestoreRequest) ProtoMessage() {} func (x *RestoreRequest) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[36] + mi := &file_atelet_proto_msgTypes[37] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2503,7 +2622,7 @@ func (x *RestoreRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use RestoreRequest.ProtoReflect.Descriptor instead. func (*RestoreRequest) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{36} + return file_atelet_proto_rawDescGZIP(), []int{37} } func (x *RestoreRequest) GetTargetAteomUid() string { @@ -2578,7 +2697,7 @@ func (x *RestoreRequest) GetLocalConfig() *LocalCheckpointConfiguration { return nil } -func (x *RestoreRequest) GetExternalConfig() *ExternalCheckpointConfiguration { +func (x *RestoreRequest) GetExternalConfig() *ExternalRestoreConfiguration { if x != nil { if x, ok := x.Config.(*RestoreRequest_ExternalConfig); ok { return x.ExternalConfig @@ -2622,6 +2741,13 @@ func (x *RestoreRequest) GetMemoryBytes() int64 { return 0 } +func (x *RestoreRequest) GetBaseConfig() *ExternalRestoreConfiguration { + if x != nil { + return x.BaseConfig + } + return nil +} + type isRestoreRequest_Config interface { isRestoreRequest_Config() } @@ -2631,7 +2757,7 @@ type RestoreRequest_LocalConfig struct { } type RestoreRequest_ExternalConfig struct { - ExternalConfig *ExternalCheckpointConfiguration `protobuf:"bytes,10,opt,name=external_config,json=externalConfig,proto3,oneof"` + ExternalConfig *ExternalRestoreConfiguration `protobuf:"bytes,10,opt,name=external_config,json=externalConfig,proto3,oneof"` } func (*RestoreRequest_LocalConfig) isRestoreRequest_Config() {} @@ -2646,7 +2772,7 @@ type RestoreResponse struct { func (x *RestoreResponse) Reset() { *x = RestoreResponse{} - mi := &file_atelet_proto_msgTypes[37] + mi := &file_atelet_proto_msgTypes[38] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2658,7 +2784,7 @@ func (x *RestoreResponse) String() string { func (*RestoreResponse) ProtoMessage() {} func (x *RestoreResponse) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[37] + mi := &file_atelet_proto_msgTypes[38] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2671,7 +2797,7 @@ func (x *RestoreResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use RestoreResponse.ProtoReflect.Descriptor instead. func (*RestoreResponse) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{37} + return file_atelet_proto_rawDescGZIP(), []int{38} } var File_atelet_proto protoreflect.FileDescriptor @@ -2811,7 +2937,10 @@ const file_atelet_proto_rawDesc = "" + "\x1cLocalCheckpointConfiguration\x12#\n" + "\rsnapshot_name\x18\x01 \x01(\tR\fsnapshotName\"D\n" + "\x1fExternalCheckpointConfiguration\x12!\n" + - "\fsnapshot_uri\x18\x01 \x01(\tR\vsnapshotUri\"\xa9\x04\n" + + "\fsnapshot_uri\x18\x01 \x01(\tR\vsnapshotUri\"t\n" + + "\x1cExternalRestoreConfiguration\x12!\n" + + "\fsnapshot_uri\x18\x01 \x01(\tR\vsnapshotUri\x121\n" + + "\asharing\x18\x02 \x01(\x0e2\x17.atelet.SnapshotSharingR\asharing\"\xa9\x04\n" + "\x11CheckpointRequest\x12(\n" + "\x10target_ateom_uid\x18\x01 \x01(\tR\x0etargetAteomUid\x12\x1a\n" + "\batespace\x18\x02 \x01(\tR\batespace\x12\x1d\n" + @@ -2838,7 +2967,7 @@ const file_atelet_proto_rawDesc = "" + "\x13local_snapshot_name\x18\x06 \x01(\tR\x11localSnapshotName\x128\n" + "\x18destination_snapshot_uri\x18\a \x01(\tR\x16destinationSnapshotUri\x12:\n" + "\rdesired_scope\x18\b \x01(\x0e2\x15.atelet.SnapshotScopeR\fdesiredScope\" \n" + - "\x1eUploadPausedCheckpointResponse\"\xec\x05\n" + + "\x1eUploadPausedCheckpointResponse\"\xb0\x06\n" + "\x0eRestoreRequest\x12(\n" + "\x10target_ateom_uid\x18\x01 \x01(\tR\x0etargetAteomUid\x12\x1a\n" + "\batespace\x18\x02 \x01(\tR\batespace\x12\x1d\n" + @@ -2849,14 +2978,16 @@ const file_atelet_proto_rawDesc = "" + "\x13actor_template_name\x18\x06 \x01(\tR\x11actorTemplateName\x12(\n" + "\x04spec\x18\a \x01(\v2\x14.atelet.WorkloadSpecR\x04spec\x12*\n" + "\x04type\x18\b \x01(\x0e2\x16.atelet.CheckpointTypeR\x04type\x12I\n" + - "\flocal_config\x18\t \x01(\v2$.atelet.LocalCheckpointConfigurationH\x00R\vlocalConfig\x12R\n" + + "\flocal_config\x18\t \x01(\v2$.atelet.LocalCheckpointConfigurationH\x00R\vlocalConfig\x12O\n" + "\x0fexternal_config\x18\n" + - " \x01(\v2'.atelet.ExternalCheckpointConfigurationH\x00R\x0eexternalConfig\x12+\n" + + " \x01(\v2$.atelet.ExternalRestoreConfigurationH\x00R\x0eexternalConfig\x12+\n" + "\x05scope\x18\v \x01(\x0e2\x15.atelet.SnapshotScopeR\x05scope\x12.\n" + "\x13golden_snapshot_uri\x18\f \x01(\tR\x11goldenSnapshotUri\x12A\n" + "\x0eegress_gateway\x18\r \x01(\v2\x15.atelet.EgressGatewayH\x01R\regressGateway\x88\x01\x01\x12\x1b\n" + "\tcpu_milli\x18\x0e \x01(\x03R\bcpuMilli\x12!\n" + - "\fmemory_bytes\x18\x0f \x01(\x03R\vmemoryBytesB\b\n" + + "\fmemory_bytes\x18\x0f \x01(\x03R\vmemoryBytes\x12E\n" + + "\vbase_config\x18\x10 \x01(\v2$.atelet.ExternalRestoreConfigurationR\n" + + "baseConfigB\b\n" + "\x06configB\x11\n" + "\x0f_egress_gateway\"\x11\n" + "\x0fRestoreResponse*\x9a\x01\n" + @@ -2873,7 +3004,11 @@ const file_atelet_proto_rawDesc = "" + "\x1aSNAPSHOT_SCOPE_UNSPECIFIED\x10\x00\x12\x17\n" + "\x13SNAPSHOT_SCOPE_FULL\x10\x01\x12\x17\n" + "\x13SNAPSHOT_SCOPE_DATA\x10\x02\x12!\n" + - "\x1dSNAPSHOT_SCOPE_DATA_ON_GOLDEN\x10\x032w\n" + + "\x1dSNAPSHOT_SCOPE_DATA_ON_GOLDEN\x10\x03*n\n" + + "\x0fSnapshotSharing\x12 \n" + + "\x1cSNAPSHOT_SHARING_UNSPECIFIED\x10\x00\x12\x1c\n" + + "\x18SNAPSHOT_SHARING_PRIVATE\x10\x01\x12\x1b\n" + + "\x17SNAPSHOT_SHARING_SHARED\x10\x022w\n" + "\x10CredentialBroker\x12c\n" + "\x14MintActorCertificate\x12#.atelet.MintActorCertificateRequest\x1a$.atelet.MintActorCertificateResponse\"\x002l\n" + "\x0eWorkerCapacity\x12Z\n" + @@ -2898,115 +3033,119 @@ func file_atelet_proto_rawDescGZIP() []byte { return file_atelet_proto_rawDescData } -var file_atelet_proto_enumTypes = make([]protoimpl.EnumInfo, 3) -var file_atelet_proto_msgTypes = make([]protoimpl.MessageInfo, 41) +var file_atelet_proto_enumTypes = make([]protoimpl.EnumInfo, 4) +var file_atelet_proto_msgTypes = make([]protoimpl.MessageInfo, 42) var file_atelet_proto_goTypes = []any{ (ActorMetadataField)(0), // 0: atelet.ActorMetadataField (CheckpointType)(0), // 1: atelet.CheckpointType (SnapshotScope)(0), // 2: atelet.SnapshotScope - (*SetWorkerCapacityRequest)(nil), // 3: atelet.SetWorkerCapacityRequest - (*SetWorkerCapacityResponse)(nil), // 4: atelet.SetWorkerCapacityResponse - (*MintActorCertificateRequest)(nil), // 5: atelet.MintActorCertificateRequest - (*MintActorCertificateResponse)(nil), // 6: atelet.MintActorCertificateResponse - (*TerminateRequest)(nil), // 7: atelet.TerminateRequest - (*TerminateResponse)(nil), // 8: atelet.TerminateResponse - (*RunRequest)(nil), // 9: atelet.RunRequest - (*EgressGateway)(nil), // 10: atelet.EgressGateway - (*AssetFile)(nil), // 11: atelet.AssetFile - (*ArchAssets)(nil), // 12: atelet.ArchAssets - (*SandboxAssets)(nil), // 13: atelet.SandboxAssets - (*WorkloadSpec)(nil), // 14: atelet.WorkloadSpec - (*DurableDirVolume)(nil), // 15: atelet.DurableDirVolume - (*ExternalVolumeSource)(nil), // 16: atelet.ExternalVolumeSource - (*ImageVolumeSource)(nil), // 17: atelet.ImageVolumeSource - (*ActorMetadataItem)(nil), // 18: atelet.ActorMetadataItem - (*ActorMetadataDataSource)(nil), // 19: atelet.ActorMetadataDataSource - (*TrustBundleDataSource)(nil), // 20: atelet.TrustBundleDataSource - (*SystemInfoDataSource)(nil), // 21: atelet.SystemInfoDataSource - (*SystemInfoVolume)(nil), // 22: atelet.SystemInfoVolume - (*Volume)(nil), // 23: atelet.Volume - (*VolumeMount)(nil), // 24: atelet.VolumeMount - (*Container)(nil), // 25: atelet.Container - (*SecurityContext)(nil), // 26: atelet.SecurityContext - (*Capabilities)(nil), // 27: atelet.Capabilities - (*ResourceLimits)(nil), // 28: atelet.ResourceLimits - (*EnvEntry)(nil), // 29: atelet.EnvEntry - (*Readyz)(nil), // 30: atelet.Readyz - (*HTTPGetAction)(nil), // 31: atelet.HTTPGetAction - (*RunResponse)(nil), // 32: atelet.RunResponse - (*LocalCheckpointConfiguration)(nil), // 33: atelet.LocalCheckpointConfiguration - (*ExternalCheckpointConfiguration)(nil), // 34: atelet.ExternalCheckpointConfiguration - (*CheckpointRequest)(nil), // 35: atelet.CheckpointRequest - (*CheckpointResponse)(nil), // 36: atelet.CheckpointResponse - (*UploadPausedCheckpointRequest)(nil), // 37: atelet.UploadPausedCheckpointRequest - (*UploadPausedCheckpointResponse)(nil), // 38: atelet.UploadPausedCheckpointResponse - (*RestoreRequest)(nil), // 39: atelet.RestoreRequest - (*RestoreResponse)(nil), // 40: atelet.RestoreResponse - nil, // 41: atelet.ArchAssets.FilesEntry - nil, // 42: atelet.SandboxAssets.AssetsEntry - nil, // 43: atelet.ExternalVolumeSource.VolumeContextEntry - (*ateapipb.WorkerResources)(nil), // 44: ateapi.WorkerResources + (SnapshotSharing)(0), // 3: atelet.SnapshotSharing + (*SetWorkerCapacityRequest)(nil), // 4: atelet.SetWorkerCapacityRequest + (*SetWorkerCapacityResponse)(nil), // 5: atelet.SetWorkerCapacityResponse + (*MintActorCertificateRequest)(nil), // 6: atelet.MintActorCertificateRequest + (*MintActorCertificateResponse)(nil), // 7: atelet.MintActorCertificateResponse + (*TerminateRequest)(nil), // 8: atelet.TerminateRequest + (*TerminateResponse)(nil), // 9: atelet.TerminateResponse + (*RunRequest)(nil), // 10: atelet.RunRequest + (*EgressGateway)(nil), // 11: atelet.EgressGateway + (*AssetFile)(nil), // 12: atelet.AssetFile + (*ArchAssets)(nil), // 13: atelet.ArchAssets + (*SandboxAssets)(nil), // 14: atelet.SandboxAssets + (*WorkloadSpec)(nil), // 15: atelet.WorkloadSpec + (*DurableDirVolume)(nil), // 16: atelet.DurableDirVolume + (*ExternalVolumeSource)(nil), // 17: atelet.ExternalVolumeSource + (*ImageVolumeSource)(nil), // 18: atelet.ImageVolumeSource + (*ActorMetadataItem)(nil), // 19: atelet.ActorMetadataItem + (*ActorMetadataDataSource)(nil), // 20: atelet.ActorMetadataDataSource + (*TrustBundleDataSource)(nil), // 21: atelet.TrustBundleDataSource + (*SystemInfoDataSource)(nil), // 22: atelet.SystemInfoDataSource + (*SystemInfoVolume)(nil), // 23: atelet.SystemInfoVolume + (*Volume)(nil), // 24: atelet.Volume + (*VolumeMount)(nil), // 25: atelet.VolumeMount + (*Container)(nil), // 26: atelet.Container + (*SecurityContext)(nil), // 27: atelet.SecurityContext + (*Capabilities)(nil), // 28: atelet.Capabilities + (*ResourceLimits)(nil), // 29: atelet.ResourceLimits + (*EnvEntry)(nil), // 30: atelet.EnvEntry + (*Readyz)(nil), // 31: atelet.Readyz + (*HTTPGetAction)(nil), // 32: atelet.HTTPGetAction + (*RunResponse)(nil), // 33: atelet.RunResponse + (*LocalCheckpointConfiguration)(nil), // 34: atelet.LocalCheckpointConfiguration + (*ExternalCheckpointConfiguration)(nil), // 35: atelet.ExternalCheckpointConfiguration + (*ExternalRestoreConfiguration)(nil), // 36: atelet.ExternalRestoreConfiguration + (*CheckpointRequest)(nil), // 37: atelet.CheckpointRequest + (*CheckpointResponse)(nil), // 38: atelet.CheckpointResponse + (*UploadPausedCheckpointRequest)(nil), // 39: atelet.UploadPausedCheckpointRequest + (*UploadPausedCheckpointResponse)(nil), // 40: atelet.UploadPausedCheckpointResponse + (*RestoreRequest)(nil), // 41: atelet.RestoreRequest + (*RestoreResponse)(nil), // 42: atelet.RestoreResponse + nil, // 43: atelet.ArchAssets.FilesEntry + nil, // 44: atelet.SandboxAssets.AssetsEntry + nil, // 45: atelet.ExternalVolumeSource.VolumeContextEntry + (*ateapipb.WorkerResources)(nil), // 46: ateapi.WorkerResources } var file_atelet_proto_depIdxs = []int32{ - 44, // 0: atelet.SetWorkerCapacityRequest.capacity:type_name -> ateapi.WorkerResources - 14, // 1: atelet.TerminateRequest.spec:type_name -> atelet.WorkloadSpec - 14, // 2: atelet.RunRequest.spec:type_name -> atelet.WorkloadSpec - 13, // 3: atelet.RunRequest.sandbox_assets:type_name -> atelet.SandboxAssets - 10, // 4: atelet.RunRequest.egress_gateway:type_name -> atelet.EgressGateway - 41, // 5: atelet.ArchAssets.files:type_name -> atelet.ArchAssets.FilesEntry - 42, // 6: atelet.SandboxAssets.assets:type_name -> atelet.SandboxAssets.AssetsEntry - 25, // 7: atelet.WorkloadSpec.containers:type_name -> atelet.Container - 23, // 8: atelet.WorkloadSpec.volumes:type_name -> atelet.Volume - 43, // 9: atelet.ExternalVolumeSource.volume_context:type_name -> atelet.ExternalVolumeSource.VolumeContextEntry + 46, // 0: atelet.SetWorkerCapacityRequest.capacity:type_name -> ateapi.WorkerResources + 15, // 1: atelet.TerminateRequest.spec:type_name -> atelet.WorkloadSpec + 15, // 2: atelet.RunRequest.spec:type_name -> atelet.WorkloadSpec + 14, // 3: atelet.RunRequest.sandbox_assets:type_name -> atelet.SandboxAssets + 11, // 4: atelet.RunRequest.egress_gateway:type_name -> atelet.EgressGateway + 43, // 5: atelet.ArchAssets.files:type_name -> atelet.ArchAssets.FilesEntry + 44, // 6: atelet.SandboxAssets.assets:type_name -> atelet.SandboxAssets.AssetsEntry + 26, // 7: atelet.WorkloadSpec.containers:type_name -> atelet.Container + 24, // 8: atelet.WorkloadSpec.volumes:type_name -> atelet.Volume + 45, // 9: atelet.ExternalVolumeSource.volume_context:type_name -> atelet.ExternalVolumeSource.VolumeContextEntry 0, // 10: atelet.ActorMetadataItem.field:type_name -> atelet.ActorMetadataField - 18, // 11: atelet.ActorMetadataDataSource.items:type_name -> atelet.ActorMetadataItem - 19, // 12: atelet.SystemInfoDataSource.actor_metadata:type_name -> atelet.ActorMetadataDataSource - 20, // 13: atelet.SystemInfoDataSource.trust_bundle:type_name -> atelet.TrustBundleDataSource - 21, // 14: atelet.SystemInfoVolume.data_sources:type_name -> atelet.SystemInfoDataSource - 15, // 15: atelet.Volume.durable_dir:type_name -> atelet.DurableDirVolume - 16, // 16: atelet.Volume.external:type_name -> atelet.ExternalVolumeSource - 22, // 17: atelet.Volume.system_info:type_name -> atelet.SystemInfoVolume - 17, // 18: atelet.Volume.image:type_name -> atelet.ImageVolumeSource - 29, // 19: atelet.Container.env:type_name -> atelet.EnvEntry - 30, // 20: atelet.Container.readyz:type_name -> atelet.Readyz - 24, // 21: atelet.Container.volume_mounts:type_name -> atelet.VolumeMount - 26, // 22: atelet.Container.security_context:type_name -> atelet.SecurityContext - 28, // 23: atelet.Container.resources:type_name -> atelet.ResourceLimits - 27, // 24: atelet.SecurityContext.capabilities:type_name -> atelet.Capabilities - 31, // 25: atelet.Readyz.http_get:type_name -> atelet.HTTPGetAction - 14, // 26: atelet.CheckpointRequest.spec:type_name -> atelet.WorkloadSpec - 1, // 27: atelet.CheckpointRequest.type:type_name -> atelet.CheckpointType - 33, // 28: atelet.CheckpointRequest.local_config:type_name -> atelet.LocalCheckpointConfiguration - 34, // 29: atelet.CheckpointRequest.external_config:type_name -> atelet.ExternalCheckpointConfiguration - 2, // 30: atelet.CheckpointRequest.scope:type_name -> atelet.SnapshotScope - 2, // 31: atelet.UploadPausedCheckpointRequest.desired_scope:type_name -> atelet.SnapshotScope - 14, // 32: atelet.RestoreRequest.spec:type_name -> atelet.WorkloadSpec - 1, // 33: atelet.RestoreRequest.type:type_name -> atelet.CheckpointType - 33, // 34: atelet.RestoreRequest.local_config:type_name -> atelet.LocalCheckpointConfiguration - 34, // 35: atelet.RestoreRequest.external_config:type_name -> atelet.ExternalCheckpointConfiguration - 2, // 36: atelet.RestoreRequest.scope:type_name -> atelet.SnapshotScope - 10, // 37: atelet.RestoreRequest.egress_gateway:type_name -> atelet.EgressGateway - 11, // 38: atelet.ArchAssets.FilesEntry.value:type_name -> atelet.AssetFile - 12, // 39: atelet.SandboxAssets.AssetsEntry.value:type_name -> atelet.ArchAssets - 5, // 40: atelet.CredentialBroker.MintActorCertificate:input_type -> atelet.MintActorCertificateRequest - 3, // 41: atelet.WorkerCapacity.SetWorkerCapacity:input_type -> atelet.SetWorkerCapacityRequest - 9, // 42: atelet.AteomHerder.Run:input_type -> atelet.RunRequest - 35, // 43: atelet.AteomHerder.Checkpoint:input_type -> atelet.CheckpointRequest - 39, // 44: atelet.AteomHerder.Restore:input_type -> atelet.RestoreRequest - 37, // 45: atelet.AteomHerder.UploadPausedCheckpoint:input_type -> atelet.UploadPausedCheckpointRequest - 7, // 46: atelet.AteomHerder.Terminate:input_type -> atelet.TerminateRequest - 6, // 47: atelet.CredentialBroker.MintActorCertificate:output_type -> atelet.MintActorCertificateResponse - 4, // 48: atelet.WorkerCapacity.SetWorkerCapacity:output_type -> atelet.SetWorkerCapacityResponse - 32, // 49: atelet.AteomHerder.Run:output_type -> atelet.RunResponse - 36, // 50: atelet.AteomHerder.Checkpoint:output_type -> atelet.CheckpointResponse - 40, // 51: atelet.AteomHerder.Restore:output_type -> atelet.RestoreResponse - 38, // 52: atelet.AteomHerder.UploadPausedCheckpoint:output_type -> atelet.UploadPausedCheckpointResponse - 8, // 53: atelet.AteomHerder.Terminate:output_type -> atelet.TerminateResponse - 47, // [47:54] is the sub-list for method output_type - 40, // [40:47] is the sub-list for method input_type - 40, // [40:40] is the sub-list for extension type_name - 40, // [40:40] is the sub-list for extension extendee - 0, // [0:40] is the sub-list for field type_name + 19, // 11: atelet.ActorMetadataDataSource.items:type_name -> atelet.ActorMetadataItem + 20, // 12: atelet.SystemInfoDataSource.actor_metadata:type_name -> atelet.ActorMetadataDataSource + 21, // 13: atelet.SystemInfoDataSource.trust_bundle:type_name -> atelet.TrustBundleDataSource + 22, // 14: atelet.SystemInfoVolume.data_sources:type_name -> atelet.SystemInfoDataSource + 16, // 15: atelet.Volume.durable_dir:type_name -> atelet.DurableDirVolume + 17, // 16: atelet.Volume.external:type_name -> atelet.ExternalVolumeSource + 23, // 17: atelet.Volume.system_info:type_name -> atelet.SystemInfoVolume + 18, // 18: atelet.Volume.image:type_name -> atelet.ImageVolumeSource + 30, // 19: atelet.Container.env:type_name -> atelet.EnvEntry + 31, // 20: atelet.Container.readyz:type_name -> atelet.Readyz + 25, // 21: atelet.Container.volume_mounts:type_name -> atelet.VolumeMount + 27, // 22: atelet.Container.security_context:type_name -> atelet.SecurityContext + 29, // 23: atelet.Container.resources:type_name -> atelet.ResourceLimits + 28, // 24: atelet.SecurityContext.capabilities:type_name -> atelet.Capabilities + 32, // 25: atelet.Readyz.http_get:type_name -> atelet.HTTPGetAction + 3, // 26: atelet.ExternalRestoreConfiguration.sharing:type_name -> atelet.SnapshotSharing + 15, // 27: atelet.CheckpointRequest.spec:type_name -> atelet.WorkloadSpec + 1, // 28: atelet.CheckpointRequest.type:type_name -> atelet.CheckpointType + 34, // 29: atelet.CheckpointRequest.local_config:type_name -> atelet.LocalCheckpointConfiguration + 35, // 30: atelet.CheckpointRequest.external_config:type_name -> atelet.ExternalCheckpointConfiguration + 2, // 31: atelet.CheckpointRequest.scope:type_name -> atelet.SnapshotScope + 2, // 32: atelet.UploadPausedCheckpointRequest.desired_scope:type_name -> atelet.SnapshotScope + 15, // 33: atelet.RestoreRequest.spec:type_name -> atelet.WorkloadSpec + 1, // 34: atelet.RestoreRequest.type:type_name -> atelet.CheckpointType + 34, // 35: atelet.RestoreRequest.local_config:type_name -> atelet.LocalCheckpointConfiguration + 36, // 36: atelet.RestoreRequest.external_config:type_name -> atelet.ExternalRestoreConfiguration + 2, // 37: atelet.RestoreRequest.scope:type_name -> atelet.SnapshotScope + 11, // 38: atelet.RestoreRequest.egress_gateway:type_name -> atelet.EgressGateway + 36, // 39: atelet.RestoreRequest.base_config:type_name -> atelet.ExternalRestoreConfiguration + 12, // 40: atelet.ArchAssets.FilesEntry.value:type_name -> atelet.AssetFile + 13, // 41: atelet.SandboxAssets.AssetsEntry.value:type_name -> atelet.ArchAssets + 6, // 42: atelet.CredentialBroker.MintActorCertificate:input_type -> atelet.MintActorCertificateRequest + 4, // 43: atelet.WorkerCapacity.SetWorkerCapacity:input_type -> atelet.SetWorkerCapacityRequest + 10, // 44: atelet.AteomHerder.Run:input_type -> atelet.RunRequest + 37, // 45: atelet.AteomHerder.Checkpoint:input_type -> atelet.CheckpointRequest + 41, // 46: atelet.AteomHerder.Restore:input_type -> atelet.RestoreRequest + 39, // 47: atelet.AteomHerder.UploadPausedCheckpoint:input_type -> atelet.UploadPausedCheckpointRequest + 8, // 48: atelet.AteomHerder.Terminate:input_type -> atelet.TerminateRequest + 7, // 49: atelet.CredentialBroker.MintActorCertificate:output_type -> atelet.MintActorCertificateResponse + 5, // 50: atelet.WorkerCapacity.SetWorkerCapacity:output_type -> atelet.SetWorkerCapacityResponse + 33, // 51: atelet.AteomHerder.Run:output_type -> atelet.RunResponse + 38, // 52: atelet.AteomHerder.Checkpoint:output_type -> atelet.CheckpointResponse + 42, // 53: atelet.AteomHerder.Restore:output_type -> atelet.RestoreResponse + 40, // 54: atelet.AteomHerder.UploadPausedCheckpoint:output_type -> atelet.UploadPausedCheckpointResponse + 9, // 55: atelet.AteomHerder.Terminate:output_type -> atelet.TerminateResponse + 49, // [49:56] is the sub-list for method output_type + 42, // [42:49] is the sub-list for method input_type + 42, // [42:42] is the sub-list for extension type_name + 42, // [42:42] is the sub-list for extension extendee + 0, // [0:42] is the sub-list for field type_name } func init() { file_atelet_proto_init() } @@ -3025,11 +3164,11 @@ func file_atelet_proto_init() { (*Volume_SystemInfo)(nil), (*Volume_Image)(nil), } - file_atelet_proto_msgTypes[32].OneofWrappers = []any{ + file_atelet_proto_msgTypes[33].OneofWrappers = []any{ (*CheckpointRequest_LocalConfig)(nil), (*CheckpointRequest_ExternalConfig)(nil), } - file_atelet_proto_msgTypes[36].OneofWrappers = []any{ + file_atelet_proto_msgTypes[37].OneofWrappers = []any{ (*RestoreRequest_LocalConfig)(nil), (*RestoreRequest_ExternalConfig)(nil), } @@ -3038,8 +3177,8 @@ func file_atelet_proto_init() { File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_atelet_proto_rawDesc), len(file_atelet_proto_rawDesc)), - NumEnums: 3, - NumMessages: 41, + NumEnums: 4, + NumMessages: 42, NumExtensions: 0, NumServices: 3, }, diff --git a/internal/proto/ateletpb/atelet.proto b/internal/proto/ateletpb/atelet.proto index f75b66a03d..945fd34d3d 100644 --- a/internal/proto/ateletpb/atelet.proto +++ b/internal/proto/ateletpb/atelet.proto @@ -360,6 +360,33 @@ enum SnapshotScope { SNAPSHOT_SCOPE_DATA_ON_GOLDEN = 3; } +// SnapshotSharing is how an external snapshot's storage may be consumed, as +// known by the control plane. +enum SnapshotSharing { + // An older caller that predates the field: treated as PRIVATE. + SNAPSHOT_SHARING_UNSPECIFIED = 0; + // Read by a single actor and deleted or replaced by its next suspend. + // Never cached. + SNAPSHOT_SHARING_PRIVATE = 1; + // Immutable once published and read by many actors (today: a template's + // golden snapshot). atelet may serve its files from the node-local + // snapshot cache. + SNAPSHOT_SHARING_SHARED = 2; +} + +// ExternalRestoreConfiguration is an external snapshot a restore reads. +// Split from the checkpoint-side ExternalCheckpointConfiguration (a write +// destination) so read-side attributes of a restore source have a home. +message ExternalRestoreConfiguration { + // The object storage URI of the snapshot to read. Object names are + // appended to it, so it addresses the snapshot as a whole rather than any + // one object. Must stay field 1: old peers decode this message as + // ExternalCheckpointConfiguration. + string snapshot_uri = 1; + // How the snapshot at snapshot_uri may be consumed. + SnapshotSharing sharing = 2; +} + message CheckpointRequest { string target_ateom_uid = 1; @@ -437,18 +464,16 @@ message RestoreRequest { // The checkpoint configuration, depending on the type. oneof config { LocalCheckpointConfiguration local_config = 9; - ExternalCheckpointConfiguration external_config = 10; + ExternalRestoreConfiguration external_config = 10; } // What content to restore from the checkpoint. SnapshotScope scope = 11; - // The object storage URI of the ActorTemplate's golden snapshot. - // Set only when scope is SNAPSHOT_SCOPE_DATA_ON_GOLDEN: restore combines - // the golden snapshot (memory + full fs delta) with the durable data in - // the snapshot referenced by `config`. A top-level field rather than part - // of the `config` oneof: the actor's snapshot may be local (a pause - // checkpoint) while the golden snapshot is always external. + // Superseded by base_config. During the transition callers set both and + // atelet prefers base_config, so ateapi and atelet can roll in either + // order; removed (and reserved) in a follow-up once both sides have + // rolled. string golden_snapshot_uri = 12; // When absent, actor traffic uses direct egress instead of atunnel. @@ -460,6 +485,11 @@ message RestoreRequest { // means "unset": keep the runtime default. int64 cpu_milli = 14; // CPU limit in millicores (1000 = one core). int64 memory_bytes = 15; // Memory limit in bytes. + + // The base guest state (memory + rootfs delta) combined with `config`'s + // durable data when scope is SNAPSHOT_SCOPE_DATA_ON_GOLDEN. Supersedes + // golden_snapshot_uri. + ExternalRestoreConfiguration base_config = 16; } message RestoreResponse {