trending/substrate
1
0
Fork 0
mirror of https://github.com/agent-substrate/substrate.git synced 2026-10-02 03:24:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

Some improvements to the way we allocate and track snapshot ids (#784)

Browse Source 
Fixes #773 

Some improvements to the way we create, track and propagate snapshot
ids.

* Consolidate snapshot id and name into a single concept.
* Use consistent terminology across the whole stack (i.e. remove
confusion about prefix vs id vs name).
* Get rid of internal only storage to hide the snapshot URI.
This commit is contained in:
Julian Gutierrez Oschmann
2026-08-08 12:25:45 -07:00
committed by GitHub
parent ca81df94ec
commit ddeee6c951
33 changed files with 929 additions and 536 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarking/locust/common/ateapi_pb2.py
+112 -113
View File
File diff suppressed because one or more lines are too long
benchmarking/locust/common/ateapi_pb2_grpc.py
+3 -3
View File
@@ -62,7 +62,7 @@ class ControlStub:
self.UpdateActor = channel.unary_unary(
'/ateapi.Control/UpdateActor',
request_serializer=ateapi__pb2.UpdateActorRequest.SerializeToString,
response_deserializer=ateapi__pb2.UpdateActorResponse.FromString,
response_deserializer=ateapi__pb2.Actor.FromString,
_registered_method=True)
self.SuspendActor = channel.unary_unary(
'/ateapi.Control/SuspendActor',
@@ -289,7 +289,7 @@ def add_ControlServicer_to_server(servicer, server):
'UpdateActor': grpc.unary_unary_rpc_method_handler(
servicer.UpdateActor,
request_deserializer=ateapi__pb2.UpdateActorRequest.FromString,
response_serializer=ateapi__pb2.UpdateActorResponse.SerializeToString,
response_serializer=ateapi__pb2.Actor.SerializeToString,
),
'SuspendActor': grpc.unary_unary_rpc_method_handler(
servicer.SuspendActor,
@@ -448,7 +448,7 @@ class Control:
target,
'/ateapi.Control/UpdateActor',
ateapi__pb2.UpdateActorRequest.SerializeToString,
ateapi__pb2.UpdateActorResponse.FromString,
ateapi__pb2.Actor.FromString,
options,
channel_credentials,
insecure,
cmd/ateapi/internal/controlapi/actor_snapshot.go
+10 -11
View File
@@ -47,7 +47,7 @@ func (s *Service) GetActorSnapshot(ctx context.Context, req *ateapipb.GetActorSn
if err := validateActorSnapshotRef(req.GetSnapshot(), "snapshot"); err != nil {
return nil, err
}
snapshot, _, _, _, err := s.getActorSnapshot(ctx, req.GetSnapshot())
snapshot, _, _, err := s.getActorSnapshot(ctx, req.GetSnapshot())
if errors.Is(err, store.ErrNotFound) {
return nil, status.Error(codes.NotFound, "ActorSnapshot not found")
}
@@ -128,7 +128,7 @@ func (s *Service) UpdateActorSnapshotTag(ctx context.Context, req *ateapipb.Upda
}
in := req.GetTag()
atespace, name := in.GetMetadata().GetAtespace(), in.GetMetadata().GetName()
_, _, current, err := s.persistence.GetActorSnapshotByTag(ctx, atespace, name)
_, current, err := s.persistence.GetActorSnapshotByTag(ctx, atespace, name)
if errors.Is(err, store.ErrNotFound) {
return nil, status.Errorf(codes.NotFound, "ActorSnapshot tag %s/%s not found", atespace, name)
}
@@ -202,29 +202,28 @@ func (s *Service) DeleteActorSnapshotTag(ctx context.Context, req *ateapipb.Dele
return tag, nil
}
func (s *Service) getActorSnapshot(ctx context.Context, ref *ateapipb.ActorSnapshotRef) (*ateapipb.ActorSnapshot, string, *ateapipb.ObjectRef, *ateapipb.ActorSnapshotTag, error) {
func (s *Service) getActorSnapshot(ctx context.Context, ref *ateapipb.ActorSnapshotRef) (*ateapipb.ActorSnapshot, *ateapipb.ObjectRef, *ateapipb.ActorSnapshotTag, error) {
var snapshot *ateapipb.ActorSnapshot
var tag *ateapipb.ActorSnapshotTag
var location string
var err error
switch ref.GetReference().(type) {
case *ateapipb.ActorSnapshotRef_Snapshot:
canonical := ref.GetSnapshot()
snapshot, location, err = s.persistence.GetActorSnapshot(ctx, canonical.GetAtespace(), canonical.GetName())
snapshot, err = s.persistence.GetActorSnapshot(ctx, canonical.GetAtespace(), canonical.GetName())
case *ateapipb.ActorSnapshotRef_Tag:
snapshot, location, tag, err = s.persistence.GetActorSnapshotByTag(ctx, ref.GetTag().GetAtespace(), ref.GetTag().GetName())
snapshot, tag, err = s.persistence.GetActorSnapshotByTag(ctx, ref.GetTag().GetAtespace(), ref.GetTag().GetName())
default:
return nil, "", nil, nil, store.ErrNotFound
return nil, nil, nil, store.ErrNotFound
}
if err != nil {
return nil, "", nil, nil, err
return nil, nil, nil, err
}
canonical := &ateapipb.ObjectRef{Atespace: snapshot.GetMetadata().GetAtespace(), Name: snapshot.GetMetadata().GetName()}
return snapshot, location, canonical, tag, nil
return snapshot, canonical, tag, nil
}
func (s *Service) lockActorSnapshot(ctx context.Context, ref *ateapipb.ActorSnapshotRef) (*store.Lock, *ateapipb.ActorSnapshot, *ateapipb.ObjectRef, *ateapipb.ActorSnapshotTag, error) {
_, _, canonical, _, err := s.getActorSnapshot(ctx, ref)
_, canonical, _, err := s.getActorSnapshot(ctx, ref)
if errors.Is(err, store.ErrNotFound) {
return nil, nil, nil, nil, status.Error(codes.NotFound, "ActorSnapshot not found")
}
@@ -238,7 +237,7 @@ func (s *Service) lockActorSnapshot(ctx context.Context, ref *ateapipb.ActorSnap
if err != nil {
return nil, nil, nil, nil, fmt.Errorf("while locking actor snapshot: %w", err)
}
snapshot, _, lockedCanonical, tag, err := s.getActorSnapshot(lock.Context(), ref)
snapshot, lockedCanonical, tag, err := s.getActorSnapshot(lock.Context(), ref)
if err != nil || canonical.GetAtespace() != lockedCanonical.GetAtespace() || canonical.GetName() != lockedCanonical.GetName() {
lock.Close()
if errors.Is(err, store.ErrNotFound) {
cmd/ateapi/internal/controlapi/crash.go
+3 -2
View File
@@ -77,8 +77,9 @@ func crashActor(ctx context.Context, st store.Interface, actorRef resources.Acto
actor.Status = ateapipb.Actor_STATUS_CRASHED
// InProgressSnapshot is kept for debugging; failed workflow
// steps must never promote it to an ActorSnapshot.
// InProgressSnapshotName and InProgressLocalSnapshotName are kept for
// debugging; failed workflow steps must never promote either of them to an
// ActorSnapshot or to LocalSnapshotInfo.
actor.WorkerAssignment = nil
_, err = st.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion())
cmd/ateapi/internal/controlapi/crash_test.go
+6 -6
View File
@@ -46,7 +46,7 @@ func seedActor(t *testing.T, ctx context.Context, st store.Interface, actorRef r
WorkerPodUid: "uid",
WorkerPodIp: "1.2.3.4",
},
InProgressSnapshot: "gs://snapshots/actor-1/reserved",
InProgressSnapshotName: "reserved-snapshot",
}); err != nil {
t.Fatalf("seed actor: %v", err)
}
@@ -85,9 +85,9 @@ func seedWorker(t *testing.T, ctx context.Context, st store.Interface, actorRef
func seedUnboundActor(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef) {
t.Helper()
if _, err := st.CreateActor(ctx, &ateapipb.Actor{
Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Name, Atespace: actorRef.Atespace},
Status: ateapipb.Actor_STATUS_RUNNING,
InProgressSnapshot: "gs://snapshots/actor-1/reserved",
Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Name, Atespace: actorRef.Atespace},
Status: ateapipb.Actor_STATUS_RUNNING,
InProgressSnapshotName: "reserved-snapshot",
}); err != nil {
t.Fatalf("seed unbound actor: %v", err)
}
@@ -105,8 +105,8 @@ func assertCrashed(t *testing.T, ctx context.Context, st store.Interface, actorR
t.Errorf("status = %v, want %v", got.GetStatus(), ateapipb.Actor_STATUS_CRASHED)
}
// Keep the snapshot uri for debugging.
if got.GetInProgressSnapshot() == "" {
t.Error(`InProgressSnapshot = "", want preserved`)
if got.GetInProgressSnapshotName() == "" {
t.Error(`InProgressSnapshotName = "", want preserved`)
}
if got.GetWorkerAssignment() != nil {
t.Errorf("WorkerAssignment = %v, want cleared", got.GetWorkerAssignment())
cmd/ateapi/internal/controlapi/create_actor_test.go
+4 -2
View File
@@ -181,7 +181,8 @@ func TestCreateActor_RejectsDifferentTemplateForDataSnapshot(t *testing.T) {
SourceActor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "source"},
ActorTemplateUid: string(tmpl.GetUID()),
ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_DATA,
}, "gs://snapshots/data")
SnapshotUri: "gs://snapshots/snapshots/" + testAtespace + "/data-snapshot",
})
if err != nil {
t.Fatalf("CreateActorSnapshot: %v", err)
}
@@ -237,7 +238,8 @@ func TestCreateActor_RejectsSnapshotWithExternalVolumes(t *testing.T) {
snapshot, err := tc.persistence.CreateActorSnapshot(context.Background(), &ateapipb.ActorSnapshot{
Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "external-volume-snapshot"},
ActorTemplateUid: string(template.GetUID()),
}, "gs://snapshots/external-volume")
SnapshotUri: "gs://snapshots/snapshots/" + testAtespace + "/external-volume-snapshot",
})
if err != nil {
t.Fatalf("CreateActorSnapshot: %v", err)
}
cmd/ateapi/internal/controlapi/functional_test.go
+11 -11
View File
@@ -459,8 +459,9 @@ func createAtespace(t *testing.T, tc *testContext, name string) {
func createActorSnapshot(t *testing.T, tc *testContext, name string) *ateapipb.ObjectRef {
t.Helper()
if _, err := tc.persistence.CreateActorSnapshot(context.Background(), &ateapipb.ActorSnapshot{
Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: name},
}, "gs://my-bucket/"+name); err != nil {
Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: name},
SnapshotUri: "gs://my-bucket/snapshots/" + testAtespace + "/" + name,
}); err != nil {
t.Fatalf("CreateActorSnapshot(%s) failed: %v", name, err)
}
return &ateapipb.ObjectRef{Atespace: testAtespace, Name: name}
@@ -547,7 +548,8 @@ func createTemplateWithContainersAndVolumes(t *testing.T, tc *testContext, ns st
ActorTemplateName: createdTemplate.GetName(),
ActorTemplateUid: string(createdTemplate.GetUID()),
ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL,
}, "gs://my-bucket/my-folder"); err != nil {
SnapshotUri: "gs://fake-fake-fake/snapshots/" + resources.GoldenActorAtespace + "/" + goldenSnapshot,
}); err != nil {
t.Fatalf("failed to create golden ActorSnapshot: %v", err)
}
createdTemplate.Status = atev1alpha1.ActorTemplateStatus{
@@ -2314,7 +2316,6 @@ func TestPauseActor(t *testing.T) {
ActorTemplateName: "tmpl1",
Status: ateapipb.Actor_STATUS_PAUSED,
LocalSnapshotInfo: &ateapipb.LocalSnapshotInfo{
SnapshotPrefix: name,
NodeVmsWithLocalSnapshots: []string{"node1"},
ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL,
},
@@ -2325,15 +2326,14 @@ func TestPauseActor(t *testing.T) {
ignoreUID,
ignoreVersion,
ignoreTimestamps,
protocmp.FilterField(&ateapipb.LocalSnapshotInfo{}, "snapshot_prefix", cmp.Comparer(func(x, y string) bool {
// The stored prefix is "<actorName>-<timestamp>-<nonce>", so match
// by prefix — in both directions, since Comparers must be
// symmetric (go-cmp probes with swapped arguments).
return strings.HasPrefix(y, x) || strings.HasPrefix(x, y)
})),
protocmp.IgnoreFields(&ateapipb.WorkerAssignment{}, "worker_pod_uid"),
protocmp.IgnoreFields(&ateapipb.LocalSnapshotInfo{}, "snapshot_name"),
); diff != "" {
t.Errorf("GetActor response mismatch (-want +got):\n%s", diff)
}
if getResp.GetLocalSnapshotInfo().GetSnapshotName() == "" {
t.Error("LocalSnapshotInfo.SnapshotName is empty, want the name the pause checkpointed under")
}
}
// TestUpdateActor_Success verifies UpdateActor replaces the actor's
@@ -2530,7 +2530,7 @@ func TestUpdateActorSnapshotTag_Success(t *testing.T) {
t.Errorf("UpdateActorSnapshotTag response mismatch (-want +got):\n%s", diff)
}
_, _, storedTag, err := tc.persistence.GetActorSnapshotByTag(ctx, testAtespace, tagName)
_, storedTag, err := tc.persistence.GetActorSnapshotByTag(ctx, testAtespace, tagName)
if err != nil {
t.Fatalf("GetActorSnapshotByTag failed: %v", err)
}
cmd/ateapi/internal/controlapi/syncer.go
+4 -1
View File
@@ -385,7 +385,10 @@ func (s *WorkerPoolSyncer) releaseActorOnDeadWorker(ctx context.Context, namespa
actor.Status = ateapipb.Actor_STATUS_CRASHED
actor.WorkerAssignment = nil
actor.InProgressSnapshot = ""
// Both in-progress checkpoints die with the worker: the durable one was
// never uploaded, the local one lived on the node that went away.
actor.InProgressSnapshotName = ""
actor.InProgressLocalSnapshotName = ""
_, err = s.persistence.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion())
cmd/ateapi/internal/controlapi/syncer_test.go
+6 -3
View File
@@ -262,8 +262,11 @@ func TestSyncer_DeleteBoundWorker_ClearsActor(t *testing.T) {
WorkerAssignment: &ateapipb.WorkerAssignment{
WorkerNamespace: ns, WorkerPool: pool, WorkerPod: pod, WorkerPodIp: ip,
},
InProgressSnapshot: "gs://snapshots/partial",
LatestSnapshot: &ateapipb.ObjectRef{Atespace: "team-orphan", Name: "last"},
// Both in-progress checkpoints are set so the assertion below covers the
// shared crash path, which cannot know which workflow was in flight.
InProgressSnapshotName: "partial-snapshot",
InProgressLocalSnapshotName: "partial-local-snapshot",
LatestSnapshot: &ateapipb.ObjectRef{Atespace: "team-orphan", Name: "last"},
})
if err != nil {
t.Fatalf("create actor: %v", err)
@@ -297,7 +300,7 @@ func TestSyncer_DeleteBoundWorker_ClearsActor(t *testing.T) {
}); err != nil {
t.Fatalf("actor not reset to CRASHED: %v", err)
}
if got.GetWorkerAssignment() != nil || got.InProgressSnapshot != "" {
if got.GetWorkerAssignment() != nil || got.InProgressSnapshotName != "" || got.InProgressLocalSnapshotName != "" {
t.Errorf("bind fields not cleared: %+v", got)
}
if got.GetLatestSnapshot().GetName() == "" {
cmd/ateapi/internal/controlapi/workflow_pause.go
+8 -10
View File
@@ -16,11 +16,9 @@ package controlapi
import (
"context"
"crypto/rand"
"errors"
"fmt"
"log/slog"
"time"
"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
"github.com/agent-substrate/substrate/internal/ateattr"
@@ -95,7 +93,7 @@ func (s *MarkPausingStep) CheckPrerequisite(ctx context.Context, input *PauseInp
}
func (s *MarkPausingStep) Execute(ctx context.Context, input *PauseInput, state *PauseState) error {
state.Actor.Status = ateapipb.Actor_STATUS_PAUSING
state.Actor.InProgressSnapshot = fmt.Sprintf("%s-%s-%s", state.Actor.GetMetadata().GetName(), time.Now().Format(time.RFC3339), rand.Text())
state.Actor.InProgressLocalSnapshotName = resources.NewSnapshotName()
updatedActor, err := s.store.UpdateActor(ctx, state.Actor, state.Actor.GetMetadata().GetVersion())
if err != nil {
return err
@@ -135,7 +133,7 @@ func (s *CallAteletPauseStep) Execute(ctx context.Context, input *PauseInput, st
ateletConn, err := s.dialer.DialForWorker(assignment.GetWorkerNamespace(), assignment.GetWorkerPod())
if err != nil {
if errors.Is(err, ErrWorkerPodNotFound) {
slog.ErrorContext(ctx, "Worker pod gone before checkpoint, crashing actor", "namespace", assignment.GetWorkerNamespace(), "pod", assignment.GetWorkerPod(), "in_progress_snapshot", state.Actor.GetInProgressSnapshot())
slog.ErrorContext(ctx, "Worker pod gone before checkpoint, crashing actor", "namespace", assignment.GetWorkerNamespace(), "pod", assignment.GetWorkerPod(), "in_progress_local_snapshot_name", state.Actor.GetInProgressLocalSnapshotName())
if err := crashActor(ctx, s.store, input.ActorRef, ateattr.OperationPause, ateattr.ReasonWorkerPodGone); err != nil {
slog.ErrorContext(ctx, "Failed to crash actor", slog.String("err", err.Error()))
}
@@ -163,7 +161,7 @@ func (s *CallAteletPauseStep) Execute(ctx context.Context, input *PauseInput, st
Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL,
Config: &ateletpb.CheckpointRequest_LocalConfig{
LocalConfig: &ateletpb.LocalCheckpointConfiguration{
SnapshotPrefix: state.Actor.InProgressSnapshot,
SnapshotName: state.Actor.InProgressLocalSnapshotName,
},
},
Scope: toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnPause),
@@ -265,17 +263,17 @@ func (s *FinalizePausedStep) Execute(ctx context.Context, input *PauseInput, sta
slog.ErrorContext(ctx, "Node name not found during finalize pause, crashing actor", slog.Any("actor", input.ActorRef))
latestActor.Status = ateapipb.Actor_STATUS_CRASHED
}
// TODO(dberkov) - what if InProgressSnapshot is empty? That shouldn't be possible.
if latestActor.InProgressSnapshot != "" {
// TODO(dberkov) - what if InProgressLocalSnapshotName is empty? That shouldn't be possible.
if latestActor.InProgressLocalSnapshotName != "" {
localInfo := &ateapipb.LocalSnapshotInfo{
SnapshotPrefix: latestActor.InProgressSnapshot,
ContentScope: toActorSnapshotContentScope(state.ActorTemplate.Spec.SnapshotsConfig.OnPause),
SnapshotName: latestActor.InProgressLocalSnapshotName,
ContentScope: toActorSnapshotContentScope(state.ActorTemplate.Spec.SnapshotsConfig.OnPause),
}
if latestActor.Status != ateapipb.Actor_STATUS_CRASHED {
localInfo.NodeVmsWithLocalSnapshots = []string{nodeName}
}
latestActor.LocalSnapshotInfo = localInfo
latestActor.InProgressSnapshot = ""
latestActor.InProgressLocalSnapshotName = ""
}
sandboxClass := ""
if worker != nil {
cmd/ateapi/internal/controlapi/workflow_pause_test.go
+6 -6
View File
@@ -52,7 +52,7 @@ func TestFinalizePausedStep_WorkerGone(t *testing.T) {
WorkerPool: "pool1",
WorkerPod: "worker-pod-1",
},
InProgressSnapshot: "snap-prefix",
InProgressLocalSnapshotName: "local-snap-1",
}
if _, err := st.CreateActor(ctx, actor); err != nil {
t.Fatalf("CreateActor: %v", err)
@@ -120,7 +120,7 @@ func TestFinalizePausedStep_RecordsContentScope(t *testing.T) {
WorkerPool: "pool1",
WorkerPod: "worker-pod-1",
},
InProgressSnapshot: "snap-prefix",
InProgressLocalSnapshotName: "snap-prefix",
})
if err != nil {
t.Fatalf("CreateActor: %v", err)
@@ -309,8 +309,8 @@ func TestCallAteletPauseStep_DanglingWorkerDoesNotRecordPhantomSnapshot(t *testi
WorkerPool: "pool",
WorkerPod: "pod-gone",
},
InProgressSnapshot: "actor-1-never-written",
LatestSnapshot: tt.prevSnapshot,
InProgressLocalSnapshotName: "actor-1-never-written",
LatestSnapshot: tt.prevSnapshot,
}
created, err := persistence.CreateActor(ctx, actor)
if err != nil {
@@ -330,8 +330,8 @@ func TestCallAteletPauseStep_DanglingWorkerDoesNotRecordPhantomSnapshot(t *testi
if stored.GetStatus() != ateapipb.Actor_STATUS_CRASHED {
t.Errorf("status = %v, want CRASHED", stored.GetStatus())
}
if got := stored.GetInProgressSnapshot(); got != "actor-1-never-written" {
t.Errorf("InProgressSnapshot = %q, want preserved for debugging", got)
if got := stored.GetInProgressLocalSnapshotName(); got != "actor-1-never-written" {
t.Errorf("InProgressLocalSnapshotName = %q, want preserved for debugging", got)
}
if tt.prevSnapshot == nil {
if stored.GetLatestSnapshot() != nil {
cmd/ateapi/internal/controlapi/workflow_resume.go
+35 -29
View File
@@ -48,22 +48,22 @@ type ResumeInput struct {
// ResumeState holds the mutable state loaded and modified during execution.
type ResumeState struct {
Actor *ateapipb.Actor
Worker *ateapipb.Worker
ActorTemplate *atev1alpha1.ActorTemplate
WasRunning bool
SnapshotLocation string
SnapshotScope ateapipb.SnapshotContentScope
SnapshotKind string
Actor *ateapipb.Actor
Worker *ateapipb.Worker
ActorTemplate *atev1alpha1.ActorTemplate
WasRunning bool
SnapshotScope ateapipb.SnapshotContentScope
SnapshotKind string
SnapshotURI resources.SnapshotURI
// WireSnapshotScope labels the restore requested, not the stored snapshot's
// SnapshotScope: a data snapshot restored on golden goes out as data_on_golden.
WireSnapshotScope string
// GoldenSnapshotLocation is the storage location of the ActorTemplate's
// GoldenSnapshotURI is the storage location of the ActorTemplate's
// golden snapshot. Populated only when the template's onResume
// configuration selects the golden snapshot as the boot source for the
// pending restore: restore then combines the golden snapshot with the
// actor's data.
GoldenSnapshotLocation string
GoldenSnapshotURI resources.SnapshotURI
}
// validateGoldenSnapshotScope rejects a golden snapshot that does not carry
@@ -113,17 +113,19 @@ func (s *LoadActorForResumeStep) Execute(ctx context.Context, input *ResumeInput
}
state.ActorTemplate = actorTemplate
if ref := actor.GetLatestSnapshot(); ref != nil {
snapshot, location, err := s.store.GetActorSnapshot(ctx, ref.GetAtespace(), ref.GetName())
snapshot, err := s.store.GetActorSnapshot(ctx, ref.GetAtespace(), ref.GetName())
if errors.Is(err, store.ErrNotFound) {
return status.Error(codes.DataLoss, "ActorSnapshot data is missing")
}
if err != nil {
return fmt.Errorf("while getting ActorSnapshot: %w", err)
}
state.SnapshotLocation = location
if state.SnapshotURI, err = resources.ParseSnapshotURI(snapshot.GetSnapshotUri()); err != nil {
return status.Errorf(codes.DataLoss, "ActorSnapshot %s/%s: %v", ref.GetAtespace(), ref.GetName(), err)
}
state.SnapshotScope = snapshot.GetContentScope()
} else if actorTemplate.Status.GoldenSnapshot != "" && !input.Boot {
snapshot, location, err := s.store.GetActorSnapshot(ctx, resources.GoldenActorAtespace, actorTemplate.Status.GoldenSnapshot)
snapshot, err := s.store.GetActorSnapshot(ctx, resources.GoldenActorAtespace, actorTemplate.Status.GoldenSnapshot)
if errors.Is(err, store.ErrNotFound) {
return status.Error(codes.DataLoss, "ActorTemplate golden snapshot data is missing")
}
@@ -133,7 +135,9 @@ func (s *LoadActorForResumeStep) Execute(ctx context.Context, input *ResumeInput
if err := validateGoldenSnapshotScope(snapshot); err != nil {
return err
}
state.SnapshotLocation = location
if state.SnapshotURI, err = resources.ParseSnapshotURI(snapshot.GetSnapshotUri()); err != nil {
return status.Errorf(codes.DataLoss, "golden ActorSnapshot %s: %v", actorTemplate.Status.GoldenSnapshot, err)
}
state.SnapshotScope = snapshot.GetContentScope()
}
@@ -156,7 +160,7 @@ func (s *LoadActorForResumeStep) Execute(ctx context.Context, input *ResumeInput
if actorTemplate.Status.GoldenSnapshot == "" {
return status.Error(codes.FailedPrecondition, "a Golden data resume requires the ActorTemplate golden snapshot, which is not available")
}
goldenSnapshot, goldenLocation, err := s.store.GetActorSnapshot(ctx, resources.GoldenActorAtespace, actorTemplate.Status.GoldenSnapshot)
goldenSnapshot, err := s.store.GetActorSnapshot(ctx, resources.GoldenActorAtespace, actorTemplate.Status.GoldenSnapshot)
if errors.Is(err, store.ErrNotFound) {
return status.Error(codes.DataLoss, "ActorTemplate golden snapshot data is missing")
}
@@ -166,7 +170,9 @@ func (s *LoadActorForResumeStep) Execute(ctx context.Context, input *ResumeInput
if err := validateGoldenSnapshotScope(goldenSnapshot); err != nil {
return err
}
state.GoldenSnapshotLocation = goldenLocation
if state.GoldenSnapshotURI, err = resources.ParseSnapshotURI(goldenSnapshot.GetSnapshotUri()); err != nil {
return status.Errorf(codes.DataLoss, "golden ActorSnapshot %s: %v", actorTemplate.Status.GoldenSnapshot, err)
}
}
}
@@ -581,37 +587,37 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
}
req.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
req.Config = &ateletpb.RestoreRequest_LocalConfig{
LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotPrefix: local.GetSnapshotPrefix()},
LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: local.GetSnapshotName()},
}
// The wire scope describes the restore OPERATION. When the template's
// onResume configuration selected the golden snapshot as the boot
// source, LoadActorForResume resolved the golden location, and the
// source, LoadActorForResume resolved the golden URI, and the
// pause snapshot restores as DATA_ON_GOLDEN — atelet combines the
// golden snapshot's guest state with the actor's data. Otherwise the
// scope mirrors what the pause captured.
req.Scope = toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnPause)
if state.GoldenSnapshotLocation != "" {
if !state.GoldenSnapshotURI.IsZero() {
req.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN
req.GoldenSnapshotUriPrefix = state.GoldenSnapshotLocation
req.GoldenSnapshotUri = state.GoldenSnapshotURI.String()
}
state.WireSnapshotScope = ateattr.SnapshotScopeValue(req.Scope)
_, err = client.Restore(ctx, req)
return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while restoring workload", ateattr.OperationResume)
} else if state.SnapshotLocation != "" {
} else if !state.SnapshotURI.IsZero() {
slog.InfoContext(ctx, "Actor has durable snapshot; Restoring from snapshot")
// Mirrors LoadActorForResume's source resolution: the durable location
// is the actor's own snapshot when one exists, the golden otherwise.
// Mirrors LoadActorForResume's source resolution: the durable URI is
// the actor's own snapshot when one exists, the golden otherwise.
state.SnapshotKind = ateattr.SnapshotKindGolden
if state.Actor.GetLatestSnapshot() != nil {
state.SnapshotKind = ateattr.SnapshotKindLatest
}
// Same wire-scope derivation as the local branch above: the snapshot
// restores as DATA_ON_GOLDEN when the golden location was resolved
// per the template's onResume configuration.
// restores as DATA_ON_GOLDEN when the golden URI was resolved per the
// template's onResume configuration.
scope := actorSnapshotContentScopeToAtelet(state.SnapshotScope)
if state.GoldenSnapshotLocation != "" {
if !state.GoldenSnapshotURI.IsZero() {
scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN
}
state.WireSnapshotScope = ateattr.SnapshotScopeValue(scope)
@@ -625,14 +631,14 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL,
Config: &ateletpb.RestoreRequest_ExternalConfig{
ExternalConfig: &ateletpb.ExternalCheckpointConfiguration{
SnapshotUriPrefix: state.SnapshotLocation,
SnapshotUri: state.SnapshotURI.String(),
},
},
Scope: scope,
// Empty unless this is a Golden data resume.
GoldenSnapshotUriPrefix: state.GoldenSnapshotLocation,
ActorUid: state.Actor.GetMetadata().Uid,
EgressGateway: egressGateway,
GoldenSnapshotUri: state.GoldenSnapshotURI.String(),
ActorUid: state.Actor.GetMetadata().Uid,
EgressGateway: egressGateway,
}
_, err = client.Restore(ctx, req)
return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while restoring durable snapshot", ateattr.OperationResume)
cmd/ateapi/internal/controlapi/workflow_resume_test.go
+16 -13
View File
@@ -806,7 +806,7 @@ func TestCallAteletRestoreStep_CheckPrerequisite_WorkerOwnership(t *testing.T) {
// location into the resume state, and the resume fails early when the golden
// snapshot is unavailable.
func TestLoadActorForResumeStep_OnGoldenDataResume(t *testing.T) {
const goldenLocation = "gs://bucket/ate-golden/snapshots/1/"
const goldenSnapshotURI = "gs://bucket/golden-root/snapshots/ate-golden/golden-1"
actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"}
tests := []struct {
@@ -826,7 +826,7 @@ func TestLoadActorForResumeStep_OnGoldenDataResume(t *testing.T) {
seedGolden bool
goldenScope ateapipb.SnapshotContentScope
wantCode codes.Code
wantGoldenLoc string
wantGoldenURI string
}{
{
name: "resolves golden location for Data durable snapshot",
@@ -836,7 +836,7 @@ func TestLoadActorForResumeStep_OnGoldenDataResume(t *testing.T) {
seedGolden: true,
goldenScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL,
wantCode: codes.OK,
wantGoldenLoc: goldenLocation,
wantGoldenURI: goldenSnapshotURI,
},
{
name: "resolves golden location for paused actor with Data onPause",
@@ -847,7 +847,7 @@ func TestLoadActorForResumeStep_OnGoldenDataResume(t *testing.T) {
seedGolden: true,
goldenScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL,
wantCode: codes.OK,
wantGoldenLoc: goldenLocation,
wantGoldenURI: goldenSnapshotURI,
},
{
// A Full pause snapshot restores from its own content; the policy
@@ -860,7 +860,7 @@ func TestLoadActorForResumeStep_OnGoldenDataResume(t *testing.T) {
seedGolden: true,
goldenScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL,
wantCode: codes.OK,
wantGoldenLoc: "",
wantGoldenURI: "",
},
{
name: "fails when golden snapshot is not Full",
@@ -894,7 +894,7 @@ func TestLoadActorForResumeStep_OnGoldenDataResume(t *testing.T) {
seedGolden: true,
goldenScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL,
wantCode: codes.OK,
wantGoldenLoc: "",
wantGoldenURI: "",
},
{
name: "leaves golden location empty under ColdBoot fromData",
@@ -904,7 +904,7 @@ func TestLoadActorForResumeStep_OnGoldenDataResume(t *testing.T) {
seedGolden: true,
goldenScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL,
wantCode: codes.OK,
wantGoldenLoc: "",
wantGoldenURI: "",
},
}
@@ -917,7 +917,8 @@ func TestLoadActorForResumeStep_OnGoldenDataResume(t *testing.T) {
if _, err := persistence.CreateActorSnapshot(ctx, &ateapipb.ActorSnapshot{
Metadata: &ateapipb.ResourceMetadata{Atespace: resources.GoldenActorAtespace, Name: tt.goldenSnapshot},
ContentScope: tt.goldenScope,
}, goldenLocation); err != nil {
SnapshotUri: goldenSnapshotURI,
}); err != nil {
t.Fatalf("CreateActorSnapshot(golden): %v", err)
}
}
@@ -925,14 +926,15 @@ func TestLoadActorForResumeStep_OnGoldenDataResume(t *testing.T) {
var seedOpts []func(*ateapipb.Actor)
if tt.paused {
seedOpts = append(seedOpts, func(a *ateapipb.Actor) {
a.LocalSnapshotInfo = &ateapipb.LocalSnapshotInfo{SnapshotPrefix: "pause-1"}
a.LocalSnapshotInfo = &ateapipb.LocalSnapshotInfo{SnapshotName: "pause-1"}
})
} else {
snap, err := persistence.CreateActorSnapshot(ctx, &ateapipb.ActorSnapshot{
Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: "snap-1"},
SourceActor: &ateapipb.ObjectRef{Atespace: actorRef.Atespace, Name: actorRef.Name},
ContentScope: tt.contentScope,
}, "gs://bucket/actors/1/snapshots/2/")
SnapshotUri: "gs://bucket/root/snapshots/" + actorRef.Atespace + "/snap-1",
})
if err != nil {
t.Fatalf("CreateActorSnapshot: %v", err)
}
@@ -969,8 +971,8 @@ func TestLoadActorForResumeStep_OnGoldenDataResume(t *testing.T) {
if err != nil {
return
}
if state.GoldenSnapshotLocation != tt.wantGoldenLoc {
t.Errorf("state.GoldenSnapshotLocation = %q, want %q", state.GoldenSnapshotLocation, tt.wantGoldenLoc)
if got := state.GoldenSnapshotURI.String(); got != tt.wantGoldenURI {
t.Errorf("state.GoldenSnapshotURI = %q, want %q", got, tt.wantGoldenURI)
}
if !tt.paused && state.SnapshotScope != tt.contentScope {
t.Errorf("state.SnapshotScope = %v, want %v", state.SnapshotScope, tt.contentScope)
@@ -993,7 +995,8 @@ func TestLoadActorForResumeStep_GoldenFallbackRejectsNonFullGolden(t *testing.T)
if _, err := persistence.CreateActorSnapshot(ctx, &ateapipb.ActorSnapshot{
Metadata: &ateapipb.ResourceMetadata{Atespace: resources.GoldenActorAtespace, Name: "golden-1"},
ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_DATA,
}, "gs://bucket/ate-golden/snapshots/1/"); err != nil {
SnapshotUri: "gs://bucket/golden-root/snapshots/ate-golden/golden-1",
}); err != nil {
t.Fatalf("CreateActorSnapshot(golden): %v", err)
}
seedWorkflowActor(t, ctx, persistence, actorRef, "ns", "tmpl1", ateapipb.Actor_STATUS_SUSPENDED)
cmd/ateapi/internal/controlapi/workflow_suspend.go
+36 -15
View File
@@ -16,12 +16,9 @@ package controlapi
import (
"context"
"crypto/rand"
"errors"
"fmt"
"log/slog"
"strings"
"time"
"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
"github.com/agent-substrate/substrate/internal/ateattr"
@@ -98,11 +95,17 @@ func (s *MarkSuspendingStep) CheckPrerequisite(ctx context.Context, input *Suspe
}
return nil
}
func (s *MarkSuspendingStep) Execute(ctx context.Context, input *SuspendInput, state *SuspendState) error {
state.Actor.Status = ateapipb.Actor_STATUS_SUSPENDING
state.Actor.InProgressSnapshotSourceActorVersion = state.SourceVersion
snapshotID := time.Now().Format(time.RFC3339) + "-" + rand.Text()
state.Actor.InProgressSnapshot = strings.TrimSuffix(state.ActorTemplate.Spec.SnapshotsConfig.Location, "/") + "/snapshots/" + snapshotID
name := resources.NewSnapshotName()
// Fail here rather than at checkpoint time if the template's location
// cannot produce a usable URI: nothing has been written yet.
if _, err := inProgressSnapshotURI(state, input.ActorRef.Atespace, name); err != nil {
return err
}
state.Actor.InProgressSnapshotName = name
updatedActor, err := s.store.UpdateActor(ctx, state.Actor, state.Actor.GetMetadata().GetVersion())
if err != nil {
return err
@@ -153,7 +156,7 @@ func (s *CallAteletSuspendStep) Execute(ctx context.Context, input *SuspendInput
ateletConn, err := s.dialer.DialForWorker(assignment.GetWorkerNamespace(), assignment.GetWorkerPod())
if err != nil {
if errors.Is(err, ErrWorkerPodNotFound) {
slog.ErrorContext(ctx, "Worker pod gone before checkpoint, crashing actor", "namespace", assignment.GetWorkerNamespace(), "pod", assignment.GetWorkerPod(), "in_progress_snapshot", state.Actor.GetInProgressSnapshot())
slog.ErrorContext(ctx, "Worker pod gone before checkpoint, crashing actor", "namespace", assignment.GetWorkerNamespace(), "pod", assignment.GetWorkerPod(), "in_progress_snapshot_name", state.Actor.GetInProgressSnapshotName())
if err := crashActor(ctx, s.store, input.ActorRef, ateattr.OperationSuspend, ateattr.ReasonWorkerPodGone); err != nil {
slog.ErrorContext(ctx, "Failed to crash actor", slog.String("err", err.Error()))
}
@@ -168,6 +171,11 @@ func (s *CallAteletSuspendStep) Execute(ctx context.Context, input *SuspendInput
return err
}
snapshotURI, err := inProgressSnapshotURI(state, state.Actor.GetMetadata().GetAtespace(), state.Actor.GetInProgressSnapshotName())
if err != nil {
return err
}
// Checkpoint does not carry the sandbox config: atelet uses the version the
// actor is currently running (recorded on-node at Run/Restore) and pins it
// into the snapshot manifest.
@@ -181,7 +189,7 @@ func (s *CallAteletSuspendStep) Execute(ctx context.Context, input *SuspendInput
Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL,
Config: &ateletpb.CheckpointRequest_ExternalConfig{
ExternalConfig: &ateletpb.ExternalCheckpointConfiguration{
SnapshotUriPrefix: state.Actor.GetInProgressSnapshot(),
SnapshotUri: snapshotURI.String(),
},
},
Scope: toAteletSnapshotScope(commitSnapshotScope(state.Actor.GetMetadata().GetAtespace(), state.ActorTemplate)),
@@ -193,6 +201,14 @@ func (s *CallAteletSuspendStep) Execute(ctx context.Context, input *SuspendInput
return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while checkpointing workload", ateattr.OperationSuspend)
}
func inProgressSnapshotURI(state *SuspendState, atespace, name string) (resources.SnapshotURI, error) {
uri, err := resources.NewSnapshotURI(state.ActorTemplate.Spec.SnapshotsConfig.Location, atespace, name)
if err != nil {
return resources.SnapshotURI{}, fmt.Errorf("while building the snapshot URI for actor %s/%s: %w", atespace, state.Actor.GetMetadata().GetName(), err)
}
return uri, nil
}
func (s *CallAteletSuspendStep) RetryBackoff() *wait.Backoff { return nil }
type DetachVolumesStep struct {
@@ -266,12 +282,16 @@ func (s *FinalizeSuspendedStep) Execute(ctx context.Context, input *SuspendInput
return err
}
latestActor.Status = ateapipb.Actor_STATUS_SUSPENDED
if latestActor.InProgressSnapshot != "" {
location := latestActor.InProgressSnapshot
prefix := strings.TrimSuffix(state.ActorTemplate.Spec.SnapshotsConfig.Location, "/") + "/snapshots/"
snapshotID := strings.ToLower(strings.NewReplacer(":", "-", "+", "-").Replace(strings.TrimPrefix(location, prefix)))
if latestActor.InProgressSnapshotName != "" {
snapshotName := latestActor.InProgressSnapshotName
// The same inputs CallAteletSuspend used, so the recorded URI is
// where the bytes were actually written.
snapshotURI, err := inProgressSnapshotURI(state, input.ActorRef.Atespace, snapshotName)
if err != nil {
return err
}
snapshot := &ateapipb.ActorSnapshot{
Metadata: &ateapipb.ResourceMetadata{Atespace: input.ActorRef.Atespace, Name: snapshotID},
Metadata: &ateapipb.ResourceMetadata{Atespace: input.ActorRef.Atespace, Name: snapshotName},
SourceActor: input.ActorRef.ToObjectRef(),
SourceActorUid: latestActor.GetMetadata().GetUid(),
SourceActorVersion: state.SourceVersion,
@@ -279,12 +299,13 @@ func (s *FinalizeSuspendedStep) Execute(ctx context.Context, input *SuspendInput
ActorTemplateName: latestActor.GetActorTemplateName(),
ActorTemplateUid: string(state.ActorTemplate.GetUID()),
ContentScope: toActorSnapshotContentScope(commitSnapshotScope(input.ActorRef.Atespace, state.ActorTemplate)),
SnapshotUri: snapshotURI.String(),
}
if _, err := s.store.CreateActorSnapshot(ctx, snapshot, location); err != nil && !errors.Is(err, store.ErrAlreadyExists) {
if _, err := s.store.CreateActorSnapshot(ctx, snapshot); err != nil && !errors.Is(err, store.ErrAlreadyExists) {
return err
}
latestActor.LatestSnapshot = &ateapipb.ObjectRef{Atespace: input.ActorRef.Atespace, Name: snapshotID}
latestActor.InProgressSnapshot = ""
latestActor.LatestSnapshot = &ateapipb.ObjectRef{Atespace: input.ActorRef.Atespace, Name: snapshotName}
latestActor.InProgressSnapshotName = ""
latestActor.InProgressSnapshotSourceActorVersion = 0
}
latestActor.WorkerAssignment = nil
cmd/ateapi/internal/controlapi/workflow_suspend_test.go
+21 -14
View File
@@ -16,7 +16,6 @@ package controlapi
import (
"context"
"strings"
"testing"
"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
@@ -32,7 +31,7 @@ import (
"k8s.io/client-go/tools/cache"
)
func TestMarkSuspendingStep_SnapshotLocation(t *testing.T) {
func TestMarkSuspendingStep_SnapshotName(t *testing.T) {
ctx := context.Background()
persistence := newTestPersistence(t)
actor, err := persistence.CreateActor(ctx, &ateapipb.Actor{
@@ -52,13 +51,21 @@ func TestMarkSuspendingStep_SnapshotLocation(t *testing.T) {
t.Fatalf("Execute: %v", err)
}
const prefix = "gs://bucket/root/snapshots/"
snapshotID, ok := strings.CutPrefix(state.Actor.GetInProgressSnapshot(), prefix)
if !ok {
t.Fatalf("snapshot location = %q, want prefix %q", state.Actor.GetInProgressSnapshot(), prefix)
// The field holds the snapshot's name, not its URI: FinalizeSuspendedStep
// names the ActorSnapshot after it, so it has to be usable as a resource
// name verbatim.
snapshotName := state.Actor.GetInProgressSnapshotName()
if !resources.IsValidResourceName(snapshotName) {
t.Fatalf("in-progress snapshot = %q, want a valid resource name", snapshotName)
}
if snapshotID == "" {
t.Fatal("snapshot ID is empty")
// The URI the later steps rebuild from that name nests under the actor's
// atespace so each tenant gets a distinct storage prefix.
uri, err := resources.NewSnapshotURI(state.ActorTemplate.Spec.SnapshotsConfig.Location, "team-a", snapshotName)
if err != nil {
t.Fatalf("NewSnapshotURI(%q): %v", snapshotName, err)
}
if want := "gs://bucket/root/snapshots/team-a/" + snapshotName; uri.String() != want {
t.Errorf("snapshot URI = %q, want %q", uri, want)
}
}
@@ -259,8 +266,8 @@ func TestCallAteletSuspendStep_DanglingWorkerDoesNotRecordPhantomSnapshot(t *tes
WorkerPool: "pool",
WorkerPod: "pod-gone",
},
InProgressSnapshot: "gs://snapshots/actor-1/never-written",
LatestSnapshot: tt.prevSnapshot,
InProgressSnapshotName: "never-written",
LatestSnapshot: tt.prevSnapshot,
}
created, err := persistence.CreateActor(ctx, actor)
if err != nil {
@@ -280,8 +287,8 @@ func TestCallAteletSuspendStep_DanglingWorkerDoesNotRecordPhantomSnapshot(t *tes
if stored.GetStatus() != ateapipb.Actor_STATUS_CRASHED {
t.Errorf("status = %v, want CRASHED", stored.GetStatus())
}
if got := stored.GetInProgressSnapshot(); got != "gs://snapshots/actor-1/never-written" {
t.Errorf("InProgressSnapshot = %q, want preserved for debugging", got)
if got := stored.GetInProgressSnapshotName(); got != "never-written" {
t.Errorf("InProgressSnapshotName = %q, want preserved for debugging", got)
}
if tt.prevSnapshot == nil {
if stored.GetLatestSnapshot() != nil {
@@ -332,7 +339,7 @@ func TestFinalizeSuspendedStep_ReleasesOnlyOwnWorker(t *testing.T) {
WorkerPool: "pool",
WorkerPod: "pod-1",
},
InProgressSnapshot: "snapshot-1",
InProgressSnapshotName: "snapshot-1",
}
created, err := persistence.CreateActor(ctx, actor)
if err != nil {
@@ -358,7 +365,7 @@ func TestFinalizeSuspendedStep_ReleasesOnlyOwnWorker(t *testing.T) {
step := &FinalizeSuspendedStep{store: persistence}
input := &SuspendInput{ActorRef: resources.ActorRef{Atespace: "team-a", Name: "shared"}}
state := &SuspendState{ActorTemplate: &atev1alpha1.ActorTemplate{Spec: atev1alpha1.ActorTemplateSpec{SnapshotsConfig: atev1alpha1.SnapshotsConfig{Location: "gs://snapshots"}}}}
state := &SuspendState{ActorTemplate: &atev1alpha1.ActorTemplate{Spec: atev1alpha1.ActorTemplateSpec{SnapshotsConfig: atev1alpha1.SnapshotsConfig{Location: "gs://bucket/root"}}}}
if err := step.Execute(ctx, input, state); err != nil {
t.Fatalf("Execute: %v", err)
}
cmd/ateapi/internal/store/ateredis/ateredis.go
+20 -45
View File
@@ -126,31 +126,6 @@ func actorSnapshotTagScanPattern(atespace string) string {
return "actor-snapshot-tag:" + atespace + ":*"
}
type dbActorSnapshot struct {
Snapshot json.RawMessage `json:"snapshot"`
Location string `json:"location"`
}
func marshalActorSnapshot(snapshot *ateapipb.ActorSnapshot, location string) ([]byte, error) {
b, err := protojson.Marshal(snapshot)
if err != nil {
return nil, err
}
return json.Marshal(dbActorSnapshot{Snapshot: b, Location: location})
}
func unmarshalActorSnapshot(b []byte) (*ateapipb.ActorSnapshot, string, error) {
var record dbActorSnapshot
if err := json.Unmarshal(b, &record); err != nil {
return nil, "", err
}
snapshot := &ateapipb.ActorSnapshot{}
if err := protojson.Unmarshal(record.Snapshot, snapshot); err != nil {
return nil, "", err
}
return snapshot, record.Location, nil
}
func atespaceDBKey(name string) string {
return "atespace:" + name
}
@@ -435,11 +410,11 @@ func (s *Persistence) CreateActor(ctx context.Context, actor *ateapipb.Actor) (*
return dbActor, nil
}
func (s *Persistence) CreateActorSnapshot(ctx context.Context, snapshot *ateapipb.ActorSnapshot, location string) (*ateapipb.ActorSnapshot, error) {
func (s *Persistence) CreateActorSnapshot(ctx context.Context, snapshot *ateapipb.ActorSnapshot) (*ateapipb.ActorSnapshot, error) {
dbKey := actorSnapshotDBKey(snapshot.GetMetadata().GetAtespace(), snapshot.GetMetadata().GetName())
dbSnapshot := proto.Clone(snapshot).(*ateapipb.ActorSnapshot)
dbSnapshot.Metadata = newCreateMetadata(snapshot.GetMetadata().GetAtespace(), snapshot.GetMetadata().GetName())
b, err := marshalActorSnapshot(dbSnapshot, location)
b, err := protojson.Marshal(dbSnapshot)
if err != nil {
return nil, fmt.Errorf("while marshaling actor snapshot: %w", err)
}
@@ -453,36 +428,36 @@ func (s *Persistence) CreateActorSnapshot(ctx context.Context, snapshot *ateapip
return dbSnapshot, nil
}
func (s *Persistence) GetActorSnapshot(ctx context.Context, atespace, name string) (*ateapipb.ActorSnapshot, string, error) {
func (s *Persistence) GetActorSnapshot(ctx context.Context, atespace, name string) (*ateapipb.ActorSnapshot, error) {
dbKey := actorSnapshotDBKey(atespace, name)
b, err := s.rdb.Get(ctx, dbKey).Bytes()
if err != nil {
if errors.Is(err, redis.Nil) {
return nil, "", store.ErrNotFound
return nil, store.ErrNotFound
}
return nil, "", fmt.Errorf("while getting actor snapshot key %q: %w", dbKey, err)
return nil, fmt.Errorf("while getting actor snapshot key %q: %w", dbKey, err)
}
snapshot, location, err := unmarshalActorSnapshot(b)
if err != nil {
return nil, "", fmt.Errorf("while unmarshaling actor snapshot: %w", err)
snapshot := &ateapipb.ActorSnapshot{}
if err := protojson.Unmarshal(b, snapshot); err != nil {
return nil, fmt.Errorf("while unmarshaling actor snapshot: %w", err)
}
return snapshot, location, nil
return snapshot, nil
}
func (s *Persistence) GetActorSnapshotByTag(ctx context.Context, atespace, name string) (*ateapipb.ActorSnapshot, string, *ateapipb.ActorSnapshotTag, error) {
func (s *Persistence) GetActorSnapshotByTag(ctx context.Context, atespace, name string) (*ateapipb.ActorSnapshot, *ateapipb.ActorSnapshotTag, error) {
b, err := s.rdb.Get(ctx, actorSnapshotTagDBKey(atespace, name)).Bytes()
if err != nil {
if errors.Is(err, redis.Nil) {
return nil, "", nil, store.ErrNotFound
return nil, nil, store.ErrNotFound
}
return nil, "", nil, fmt.Errorf("while resolving actor snapshot tag %s/%s: %w", atespace, name, err)
return nil, nil, fmt.Errorf("while resolving actor snapshot tag %s/%s: %w", atespace, name, err)
}
tag := &ateapipb.ActorSnapshotTag{}
if err := protojson.Unmarshal(b, tag); err != nil {
return nil, "", nil, fmt.Errorf("while unmarshaling actor snapshot tag %s/%s: %w", atespace, name, err)
return nil, nil, fmt.Errorf("while unmarshaling actor snapshot tag %s/%s: %w", atespace, name, err)
}
snapshot, location, err := s.GetActorSnapshot(ctx, tag.GetSnapshot().GetAtespace(), tag.GetSnapshot().GetName())
return snapshot, location, tag, err
snapshot, err := s.GetActorSnapshot(ctx, tag.GetSnapshot().GetAtespace(), tag.GetSnapshot().GetName())
return snapshot, tag, err
}
func (s *Persistence) ListActorSnapshots(ctx context.Context, atespace string, pageSize int32, pageTokenStr string) ([]*ateapipb.ActorSnapshot, string, error) {
@@ -506,9 +481,9 @@ func (s *Persistence) ListActorSnapshots(ctx context.Context, atespace string, p
if getCmd.Err() != nil {
return 0, fmt.Errorf("while getting actor snapshot: %w", getCmd.Err())
}
snapshot, _, err := unmarshalActorSnapshot([]byte(getCmd.Val()))
if err != nil {
return 0, err
snapshot := &ateapipb.ActorSnapshot{}
if err := protojson.Unmarshal([]byte(getCmd.Val()), snapshot); err != nil {
return 0, fmt.Errorf("while unmarshaling actor snapshot: %w", err)
}
result = append(result, snapshot)
collected++
@@ -522,7 +497,7 @@ func (s *Persistence) ListActorSnapshots(ctx context.Context, atespace string, p
}
func (s *Persistence) TagActorSnapshot(ctx context.Context, atespace, name string, tag *ateapipb.ActorSnapshotTag) (*ateapipb.ActorSnapshotTag, error) {
if _, _, err := s.GetActorSnapshot(ctx, atespace, name); err != nil {
if _, err := s.GetActorSnapshot(ctx, atespace, name); err != nil {
return nil, err
}
dbTag := proto.Clone(tag).(*ateapipb.ActorSnapshotTag)
@@ -601,7 +576,7 @@ func (s *Persistence) UpdateActorSnapshotTag(ctx context.Context, atespace, name
}
func (s *Persistence) DeleteActorSnapshotTag(ctx context.Context, atespace, name string) (*ateapipb.ActorSnapshotTag, error) {
_, _, tag, err := s.GetActorSnapshotByTag(ctx, atespace, name)
_, tag, err := s.GetActorSnapshotByTag(ctx, atespace, name)
if err != nil {
return nil, err
}
cmd/ateapi/internal/store/ateredis/ateredis_test.go
+23 -17
View File
@@ -570,17 +570,20 @@ func TestActorSnapshotLifecycle(t *testing.T) {
SourceActorUid: "actor-uid",
SourceActorVersion: 7,
ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL,
SnapshotUri: "gs://bucket/root/snapshots/" + testAtespace + "/snapshot-1",
}
created, err := s.CreateActorSnapshot(ctx, snapshot, "gs://private/snapshot-1")
created, err := s.CreateActorSnapshot(ctx, snapshot)
if err != nil {
t.Fatalf("CreateActorSnapshot: %v", err)
}
got, location, err := s.GetActorSnapshot(ctx, testAtespace, "snapshot-1")
got, err := s.GetActorSnapshot(ctx, testAtespace, "snapshot-1")
if err != nil {
t.Fatalf("GetActorSnapshot: %v", err)
}
if !proto.Equal(created, got) || location != "gs://private/snapshot-1" {
t.Fatalf("GetActorSnapshot = (%v, %q), want (%v, private location)", got, location, created)
// The store round-trips the whole resource, snapshot_uri included: it is
// an ordinary field now, not a value the store keeps beside the record.
if !proto.Equal(created, got) {
t.Fatalf("GetActorSnapshot = %v, want %v", got, created)
}
tag := &ateapipb.ActorSnapshotTag{
Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "before-upgrade"},
@@ -590,13 +593,14 @@ func TestActorSnapshotLifecycle(t *testing.T) {
if err != nil || tagged.GetSnapshot().GetName() != "snapshot-1" {
t.Fatalf("TagActorSnapshot = (%v, %v), want stable tag", tagged, err)
}
byTag, _, resolvedTag, err := s.GetActorSnapshotByTag(ctx, testAtespace, "before-upgrade")
byTag, resolvedTag, err := s.GetActorSnapshotByTag(ctx, testAtespace, "before-upgrade")
if err != nil || !proto.Equal(created, byTag) || !proto.Equal(tagged, resolvedTag) {
t.Fatalf("GetActorSnapshotByTag = (%v, %v, %v), want tagged snapshot", byTag, resolvedTag, err)
}
if _, err := s.CreateActorSnapshot(ctx, &ateapipb.ActorSnapshot{
Metadata: &ateapipb.ResourceMetadata{Atespace: "other", Name: "snapshot-2"},
}, "gs://private/snapshot-2"); err != nil {
Metadata: &ateapipb.ResourceMetadata{Atespace: "other", Name: "snapshot-2"},
SnapshotUri: "gs://bucket/root/snapshots/other/snapshot-2",
}); err != nil {
t.Fatalf("CreateActorSnapshot second snapshot: %v", err)
}
otherTag := &ateapipb.ActorSnapshotTag{Metadata: &ateapipb.ResourceMetadata{Atespace: "other", Name: "before-upgrade"}, Scope: ateapipb.ActorSnapshotTagScope_ACTOR_SNAPSHOT_TAG_SCOPE_ATESPACE}
@@ -615,7 +619,7 @@ func TestActorSnapshotLifecycle(t *testing.T) {
if err != nil || tagged.GetScope() != ateapipb.ActorSnapshotTagScope_ACTOR_SNAPSHOT_TAG_SCOPE_PUBLISHED {
t.Fatalf("UpdateActorSnapshotTag = (%v, %v), want published", tagged, err)
}
if byTag, _, resolvedTag, err = s.GetActorSnapshotByTag(ctx, testAtespace, "before-upgrade"); err != nil || byTag.GetMetadata().GetUid() != created.GetMetadata().GetUid() || resolvedTag.GetScope() != ateapipb.ActorSnapshotTagScope_ACTOR_SNAPSHOT_TAG_SCOPE_PUBLISHED {
if byTag, resolvedTag, err = s.GetActorSnapshotByTag(ctx, testAtespace, "before-upgrade"); err != nil || byTag.GetMetadata().GetUid() != created.GetMetadata().GetUid() || resolvedTag.GetScope() != ateapipb.ActorSnapshotTagScope_ACTOR_SNAPSHOT_TAG_SCOPE_PUBLISHED {
t.Fatalf("tag after publication = (%v, %v, %v), want same address and snapshot", byTag, resolvedTag, err)
}
listed, _, err := s.ListActorSnapshots(ctx, testAtespace, 10, "")
@@ -627,10 +631,10 @@ func TestActorSnapshotLifecycle(t *testing.T) {
if err != nil || deleted.GetMetadata().GetName() != "before-upgrade" {
t.Fatalf("DeleteActorSnapshotTag = (%v, %v)", deleted, err)
}
if _, _, _, err := s.GetActorSnapshotByTag(ctx, testAtespace, "before-upgrade"); !errors.Is(err, store.ErrNotFound) {
if _, _, err := s.GetActorSnapshotByTag(ctx, testAtespace, "before-upgrade"); !errors.Is(err, store.ErrNotFound) {
t.Fatalf("deleted tag lookup = %v, want ErrNotFound", err)
}
if got, _, err := s.GetActorSnapshot(ctx, testAtespace, "snapshot-1"); err != nil || got.GetMetadata().GetUid() != created.GetMetadata().GetUid() {
if got, err := s.GetActorSnapshot(ctx, testAtespace, "snapshot-1"); err != nil || got.GetMetadata().GetUid() != created.GetMetadata().GetUid() {
t.Fatalf("snapshot after tag deletion = (%v, %v), want retained metadata", got, err)
}
}
@@ -638,8 +642,9 @@ func TestActorSnapshotLifecycle(t *testing.T) {
func TestUpdateActorSnapshotTag_Conflict(t *testing.T) {
_, s, ctx := setupTest(t)
if _, err := s.CreateActorSnapshot(ctx, &ateapipb.ActorSnapshot{
Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "snapshot-1"},
}, "gs://private/snapshot-1"); err != nil {
Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "snapshot-1"},
SnapshotUri: "gs://bucket/root/snapshots/" + testAtespace + "/snapshot-1",
}); err != nil {
t.Fatal(err)
}
if _, err := s.TagActorSnapshot(ctx, testAtespace, "snapshot-1", &ateapipb.ActorSnapshotTag{
@@ -647,11 +652,11 @@ func TestUpdateActorSnapshotTag_Conflict(t *testing.T) {
}); err != nil {
t.Fatal(err)
}
_, _, tag1, err := s.GetActorSnapshotByTag(ctx, testAtespace, "tag-1")
_, tag1, err := s.GetActorSnapshotByTag(ctx, testAtespace, "tag-1")
if err != nil {
t.Fatal(err)
}
_, _, tag2, err := s.GetActorSnapshotByTag(ctx, testAtespace, "tag-1")
_, tag2, err := s.GetActorSnapshotByTag(ctx, testAtespace, "tag-1")
if err != nil {
t.Fatal(err)
}
@@ -1439,8 +1444,9 @@ func TestDeleteAtespace_WithTags_Rejected(t *testing.T) {
t.Fatalf("CreateAtespace: %v", err)
}
if _, err := s.CreateActorSnapshot(ctx, &ateapipb.ActorSnapshot{
Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "snapshot-1"},
}, "gs://private/snapshot-1"); err != nil {
Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "snapshot-1"},
SnapshotUri: "gs://bucket/root/snapshots/team-a/snapshot-1",
}); err != nil {
t.Fatalf("CreateActorSnapshot: %v", err)
}
if _, err := s.TagActorSnapshot(ctx, "team-a", "snapshot-1", &ateapipb.ActorSnapshotTag{
@@ -1452,7 +1458,7 @@ func TestDeleteAtespace_WithTags_Rejected(t *testing.T) {
if _, err := s.DeleteAtespace(ctx, "team-a"); !errors.Is(err, store.ErrFailedPrecondition) {
t.Fatalf("DeleteAtespace = %v, want ErrFailedPrecondition", err)
}
if _, _, _, err := s.GetActorSnapshotByTag(ctx, "team-a", "keep-me"); err != nil {
if _, _, err := s.GetActorSnapshotByTag(ctx, "team-a", "keep-me"); err != nil {
t.Fatalf("GetActorSnapshotByTag after rejected deletion: %v", err)
}
if _, err := s.GetAtespace(ctx, "team-a"); err != nil {
cmd/ateapi/internal/store/store.go
+6 -5
View File
@@ -64,14 +64,15 @@ type Interface interface {
// empty. Returns a page of actors and a next page token.
ListActors(ctx context.Context, atespace string, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error)
// Creates an immutable ActorSnapshot and stores its private physical location.
CreateActorSnapshot(ctx context.Context, snapshot *ateapipb.ActorSnapshot, location string) (*ateapipb.ActorSnapshot, error)
// Creates an immutable ActorSnapshot. The caller sets snapshot_uri; the
// store keeps no location of its own.
CreateActorSnapshot(ctx context.Context, snapshot *ateapipb.ActorSnapshot) (*ateapipb.ActorSnapshot, error)
// Fetches an ActorSnapshot and its private physical location.
GetActorSnapshot(ctx context.Context, atespace, name string) (*ateapipb.ActorSnapshot, string, error)
// Fetches an ActorSnapshot.
GetActorSnapshot(ctx context.Context, atespace, name string) (*ateapipb.ActorSnapshot, error)
// Resolves an Atespace-owned tag to an ActorSnapshot in constant time.
GetActorSnapshotByTag(ctx context.Context, atespace, name string) (*ateapipb.ActorSnapshot, string, *ateapipb.ActorSnapshotTag, error)
GetActorSnapshotByTag(ctx context.Context, atespace, name string) (*ateapipb.ActorSnapshot, *ateapipb.ActorSnapshotTag, error)
// Lists ActorSnapshots in one atespace, or all atespaces when empty.
ListActorSnapshots(ctx context.Context, atespace string, pageSize int32, pageToken string) ([]*ateapipb.ActorSnapshot, string, error)
cmd/atelet/main.go
+65 -34
View File
@@ -28,7 +28,6 @@ import (
"os/signal"
"path/filepath"
"strconv"
"strings"
"syscall"
"time"
@@ -627,7 +626,7 @@ func toAteomSnapshotScope(scope ateletpb.SnapshotScope) ateompb.SnapshotScope {
}
func (s *AteomHerder) moveLocalCheckpoint(ctx context.Context, req *ateletpb.CheckpointRequest, checkpointDir string, rec *sandboxAssetsRecord) error {
localCheckpointPath := filepath.Join(ateompath.LocalCheckpointsDir(req.GetActorUid()), req.GetLocalConfig().GetSnapshotPrefix())
localCheckpointPath := filepath.Join(ateompath.LocalCheckpointsDir(req.GetActorUid()), req.GetLocalConfig().GetSnapshotName())
if err := os.MkdirAll(localCheckpointPath, 0o700); err != nil {
return fmt.Errorf("while creating local checkpoint directory: %w", err)
}
@@ -656,7 +655,10 @@ func (s *AteomHerder) moveLocalCheckpoint(ctx context.Context, req *ateletpb.Che
}
func (s *AteomHerder) uploadExternalCheckpoint(ctx context.Context, req *ateletpb.CheckpointRequest, checkpointDir string, rec *sandboxAssetsRecord) error {
prefix := strings.TrimSuffix(req.GetExternalConfig().GetSnapshotUriPrefix(), "/")
uri, err := resources.ParseSnapshotURI(req.GetExternalConfig().GetSnapshotUri())
if err != nil {
return err
}
// Upload exactly the files ateom reported (each zstd-compressed).
g, gCtx := errgroup.WithContext(ctx)
@@ -665,7 +667,11 @@ func (s *AteomHerder) uploadExternalCheckpoint(ctx context.Context, req *ateletp
local := filepath.Join(checkpointDir, fileName)
recordSnapshotSize(ctx, fileName, local, req.GetActorTemplateNamespace(), req.GetActorTemplateName())
g.Go(func() error {
if err := ategcs.SendLocalFileToGCSWithZstd(gCtx, s.gcsClient, prefix+"/"+fileName+".zstd", local); err != nil {
objectURI, err := uri.ObjectURI(fileName + ".zstd")
if err != nil {
return fmt.Errorf("while addressing %s in GCS: %w", fileName, err)
}
if err := ategcs.SendLocalFileToGCSWithZstd(gCtx, s.gcsClient, objectURI, local); err != nil {
return fmt.Errorf("while uploading %s to GCS: %w", fileName, err)
}
return nil
@@ -680,7 +686,11 @@ func (s *AteomHerder) uploadExternalCheckpoint(ctx context.Context, req *ateletp
if err != nil {
return fmt.Errorf("while marshaling snapshot manifest: %w", err)
}
if err := ategcs.SendBytesToGCS(ctx, s.gcsClient, prefix+"/"+sandboxManifestName, manifest); err != nil {
manifestURI, err := uri.ObjectURI(sandboxManifestName)
if err != nil {
return fmt.Errorf("while addressing snapshot manifest in GCS: %w", err)
}
if err := ategcs.SendBytesToGCS(ctx, s.gcsClient, manifestURI, manifest); err != nil {
return fmt.Errorf("while uploading snapshot manifest: %w", err)
}
return nil
@@ -747,8 +757,15 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
var sandboxRec *sandboxAssetsRecord
switch req.GetType() {
case ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL:
prefix := req.GetExternalConfig().GetSnapshotUriPrefix()
manifest, err := ategcs.FetchFromGCS(ctx, s.gcsClient, strings.TrimSuffix(prefix, "/")+"/"+sandboxManifestName)
uri, err := resources.ParseSnapshotURI(req.GetExternalConfig().GetSnapshotUri())
if err != nil {
return nil, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidObjectURL)
}
manifestURI, err := uri.ObjectURI(sandboxManifestName)
if err != nil {
return nil, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidObjectURL)
}
manifest, err := ategcs.FetchFromGCS(ctx, s.gcsClient, manifestURI)
if err != nil {
return nil, ateerrors.CrashIfReason(ctx, fmt.Errorf("while fetching snapshot manifest: %w", err), ateerrors.ReasonInvalidObjectURL, ateerrors.ReasonFailedGetExternalObject)
}
@@ -757,8 +774,8 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
}
case ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL:
localCheckpointDir := ateompath.LocalCheckpointsDir(actorUID)
snapshotPrefix := req.GetLocalConfig().GetSnapshotPrefix()
manifest, err := os.ReadFile(filepath.Join(localCheckpointDir, snapshotPrefix, sandboxManifestName))
snapshotName := req.GetLocalConfig().GetSnapshotName()
manifest, err := os.ReadFile(filepath.Join(localCheckpointDir, snapshotName, sandboxManifestName))
if err != nil {
if isTerminalFileSystemErr(err) {
return nil, ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonTerminalFileSystemError, ateerrors.ActorCrashedMetadata(), err)
@@ -780,8 +797,15 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
// that created it).
var goldenRec *sandboxAssetsRecord
if req.GetScope() == ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN {
goldenPrefix := req.GetGoldenSnapshotUriPrefix()
manifest, err := ategcs.FetchFromGCS(ctx, s.gcsClient, strings.TrimSuffix(goldenPrefix, "/")+"/"+sandboxManifestName)
goldenURI, err := resources.ParseSnapshotURI(req.GetGoldenSnapshotUri())
if err != nil {
return nil, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidObjectURL)
}
manifestURI, err := goldenURI.ObjectURI(sandboxManifestName)
if err != nil {
return nil, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidObjectURL)
}
manifest, err := ategcs.FetchFromGCS(ctx, s.gcsClient, manifestURI)
if err != nil {
return nil, ateerrors.CrashIfReason(ctx, fmt.Errorf("while fetching golden snapshot manifest: %w", err), ateerrors.ReasonInvalidObjectURL, ateerrors.ReasonFailedGetExternalObject)
}
@@ -835,10 +859,10 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
if goldenRec == nil {
return fmt.Errorf("no golden snapshot record for a %s restore", req.GetScope())
}
if err := s.downloadCombinedCheckpoint(gctx, req.GetExternalConfig().GetSnapshotUriPrefix(), req.GetGoldenSnapshotUriPrefix(), checkpointDir, sandboxRec.SnapshotFiles, goldenRec.SnapshotFiles); err != nil {
if err := s.downloadCombinedCheckpoint(gctx, req.GetExternalConfig().GetSnapshotUri(), req.GetGoldenSnapshotUri(), checkpointDir, sandboxRec.SnapshotFiles, goldenRec.SnapshotFiles); err != nil {
return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonFailedGetExternalObject, ateerrors.ReasonInvalidObjectURL, ateerrors.ReasonTerminalFileSystemError)
}
} else if err := s.downloadExternalCheckpoint(gctx, req.GetExternalConfig().GetSnapshotUriPrefix(), checkpointDir, sandboxRec.SnapshotFiles); err != nil {
} 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:
@@ -851,14 +875,14 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
// the golden's from object storage, concurrently.
gLocal, gLocalCtx := errgroup.WithContext(gctx)
gLocal.Go(func() error {
if err := s.copyLocalCheckpoint(gLocalCtx, req.GetLocalConfig().GetSnapshotPrefix(), ateompath.LocalCheckpointsDir(actorUID), checkpointDir, sandboxRec.SnapshotFiles); err != nil {
if err := s.copyLocalCheckpoint(gLocalCtx, req.GetLocalConfig().GetSnapshotName(), ateompath.LocalCheckpointsDir(actorUID), checkpointDir, sandboxRec.SnapshotFiles); err != nil {
return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonTerminalFileSystemError)
}
return nil
})
if combineWithGolden {
gLocal.Go(func() error {
if err := s.downloadExternalCheckpoint(gLocalCtx, req.GetGoldenSnapshotUriPrefix(), checkpointDir, goldenOnlyFiles(sandboxRec.SnapshotFiles, goldenRec.SnapshotFiles)); err != nil {
if err := s.downloadExternalCheckpoint(gLocalCtx, req.GetGoldenSnapshotUri(), checkpointDir, goldenOnlyFiles(sandboxRec.SnapshotFiles, goldenRec.SnapshotFiles)); err != nil {
return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonFailedGetExternalObject, ateerrors.ReasonInvalidObjectURL, ateerrors.ReasonTerminalFileSystemError)
}
return nil
@@ -921,7 +945,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.
GoldenSnapshotUriPrefix: req.GetGoldenSnapshotUriPrefix(),
GoldenSnapshotUri: req.GetGoldenSnapshotUri(),
})
dAteom = time.Since(tAteom)
if err != nil {
@@ -948,12 +972,12 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
return &ateletpb.RestoreResponse{}, nil
}
func (s *AteomHerder) copyLocalCheckpoint(ctx context.Context, snapshotPrefix string, srcDir, dstDir string, files []string) error {
func (s *AteomHerder) copyLocalCheckpoint(ctx context.Context, snapshotName string, srcDir, dstDir string, files []string) error {
for _, fileName := range files {
if ctx.Err() != nil {
return fmt.Errorf("context cancelled: %w", ctx.Err())
}
src := filepath.Join(srcDir, snapshotPrefix, fileName)
src := filepath.Join(srcDir, snapshotName, fileName)
dst := filepath.Join(dstDir, fileName)
if _, err := copyFile(src, dst); err != nil {
return fmt.Errorf("failed to copy %s to %s: %w", src, dst, err)
@@ -1137,25 +1161,32 @@ 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, actorPrefix, goldenPrefix, dstDir string, actorFiles, goldenFiles []string) error {
func (s *AteomHerder) downloadCombinedCheckpoint(ctx context.Context, actorURI, goldenURI, dstDir string, actorFiles, goldenFiles []string) error {
g, gctx := errgroup.WithContext(ctx)
g.Go(func() error {
return s.downloadExternalCheckpoint(gctx, actorPrefix, dstDir, actorFiles)
return s.downloadExternalCheckpoint(gctx, actorURI, dstDir, actorFiles)
})
g.Go(func() error {
return s.downloadExternalCheckpoint(gctx, goldenPrefix, dstDir, goldenOnlyFiles(actorFiles, goldenFiles))
return s.downloadExternalCheckpoint(gctx, goldenURI, dstDir, goldenOnlyFiles(actorFiles, goldenFiles))
})
return g.Wait()
}
func (s *AteomHerder) downloadExternalCheckpoint(ctx context.Context, snapshotUriPrefix string, dstDir string, files []string) error {
prefix := strings.TrimSuffix(snapshotUriPrefix, "/")
func (s *AteomHerder) downloadExternalCheckpoint(ctx context.Context, snapshotURI string, dstDir string, files []string) error {
uri, err := resources.ParseSnapshotURI(snapshotURI)
if err != nil {
return err
}
g, gCtx := errgroup.WithContext(ctx)
for _, fileName := range files {
fileName := fileName
local := filepath.Join(dstDir, fileName)
g.Go(func() error {
if err := ategcs.FetchLocalFileFromGCSWithZstd(gCtx, s.gcsClient, prefix+"/"+fileName+".zstd", local); err != nil {
objectURI, err := uri.ObjectURI(fileName + ".zstd")
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 {
return fmt.Errorf("while downloading %s from GCS: %w", fileName, err)
}
return nil
@@ -1427,12 +1458,12 @@ func validateCheckpointRequest(req *ateletpb.CheckpointRequest) error {
switch req.GetType() {
case ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL:
if err := resources.ValidateSnapshotURIPrefix(req.GetExternalConfig().GetSnapshotUriPrefix()); err != nil {
if _, err := resources.ParseSnapshotURI(req.GetExternalConfig().GetSnapshotUri()); err != nil {
return err
}
case ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL:
if err := resources.ValidateLocalSnapshotPrefix(req.GetLocalConfig().GetSnapshotPrefix()); err != nil {
return err
if !resources.IsValidResourceName(req.GetLocalConfig().GetSnapshotName()) {
return fmt.Errorf("invalid local snapshot name %q", req.GetLocalConfig().GetSnapshotName())
}
default:
return fmt.Errorf("invalid checkpoint type: %v", req.GetType())
@@ -1479,12 +1510,12 @@ func validateRestoreRequest(req *ateletpb.RestoreRequest) error {
switch req.GetType() {
case ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL:
if err := resources.ValidateSnapshotURIPrefix(req.GetExternalConfig().GetSnapshotUriPrefix()); err != nil {
if _, err := resources.ParseSnapshotURI(req.GetExternalConfig().GetSnapshotUri()); err != nil {
return err
}
case ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL:
if err := resources.ValidateLocalSnapshotPrefix(req.GetLocalConfig().GetSnapshotPrefix()); err != nil {
return err
if !resources.IsValidResourceName(req.GetLocalConfig().GetSnapshotName()) {
return fmt.Errorf("invalid local snapshot name %q", req.GetLocalConfig().GetSnapshotName())
}
default:
return fmt.Errorf("invalid checkpoint type: %v", req.GetType())
@@ -1494,11 +1525,11 @@ func validateRestoreRequest(req *ateletpb.RestoreRequest) error {
// (local pause checkpoint or external commit) and the golden snapshot,
// which is always external.
if req.GetScope() == ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN {
if err := resources.ValidateSnapshotURIPrefix(req.GetGoldenSnapshotUriPrefix()); err != nil {
return fmt.Errorf("invalid golden_snapshot_uri_prefix: %w", err)
if _, err := resources.ParseSnapshotURI(req.GetGoldenSnapshotUri()); err != nil {
return fmt.Errorf("invalid golden_snapshot_uri: %w", err)
}
} else if req.GetGoldenSnapshotUriPrefix() != "" {
return fmt.Errorf("golden_snapshot_uri_prefix is only valid with snapshot scope %s", ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN)
} else if req.GetGoldenSnapshotUri() != "" {
return fmt.Errorf("golden_snapshot_uri is only valid with snapshot scope %s", ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN)
}
return nil
}
cmd/atelet/main_test.go
+32 -24
View File
@@ -221,7 +221,7 @@ func validCheckpointRequest() *ateletpb.CheckpointRequest {
Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL,
Config: &ateletpb.CheckpointRequest_ExternalConfig{
ExternalConfig: &ateletpb.ExternalCheckpointConfiguration{
SnapshotUriPrefix: "gs://bucket/actors/1/snapshots/2/",
SnapshotUri: "gs://bucket/root/snapshots/ate-demo/counter-1-snap",
},
},
Scope: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_FULL,
@@ -240,7 +240,7 @@ func validRestoreRequest() *ateletpb.RestoreRequest {
Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL,
Config: &ateletpb.RestoreRequest_ExternalConfig{
ExternalConfig: &ateletpb.ExternalCheckpointConfiguration{
SnapshotUriPrefix: "gs://bucket/actors/1/snapshots/2/",
SnapshotUri: "gs://bucket/root/snapshots/ate-demo/counter-1-snap",
},
},
Scope: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_FULL,
@@ -275,7 +275,7 @@ func TestValidateRunRequest(t *testing.T) {
}
}
// Checkpoint and Restore must reject a bad snapshot URI prefix even when
// Checkpoint and Restore must reject a bad snapshot URI even when
// every common field is valid.
func TestValidateCheckpointRequest(t *testing.T) {
makeReq := func(opts ...func(*ateletpb.CheckpointRequest)) *ateletpb.CheckpointRequest {
@@ -292,8 +292,8 @@ func TestValidateCheckpointRequest(t *testing.T) {
wantErr bool
}{
{"valid", makeReq(), false},
{"empty snapshot uri", makeReq(func(r *ateletpb.CheckpointRequest) { r.GetExternalConfig().SnapshotUriPrefix = "" }), true},
{"bucketless snapshot uri", makeReq(func(r *ateletpb.CheckpointRequest) { r.GetExternalConfig().SnapshotUriPrefix = "relative/path" }), true},
{"empty snapshot uri", makeReq(func(r *ateletpb.CheckpointRequest) { r.GetExternalConfig().SnapshotUri = "" }), true},
{"bucketless snapshot uri", makeReq(func(r *ateletpb.CheckpointRequest) { r.GetExternalConfig().SnapshotUri = "relative/path" }), true},
{"invalid ateom uid", makeReq(func(r *ateletpb.CheckpointRequest) { r.TargetAteomUid = "../escape" }), true},
{"invalid atespace", makeReq(func(r *ateletpb.CheckpointRequest) { r.Atespace = "../escape" }), true},
{"invalid actor name", makeReq(func(r *ateletpb.CheckpointRequest) { r.ActorName = "../escape" }), true},
@@ -305,15 +305,19 @@ func TestValidateCheckpointRequest(t *testing.T) {
}), true},
{"invalid local snapshot prefix", makeReq(func(r *ateletpb.CheckpointRequest) {
r.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
r.Config = &ateletpb.CheckpointRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotPrefix: ""}}
r.Config = &ateletpb.CheckpointRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: ""}}
}), true},
{"local snapshot name escapes its directory", makeReq(func(r *ateletpb.CheckpointRequest) {
r.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
r.Config = &ateletpb.CheckpointRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "../escape"}}
}), true},
{"nested local snapshot prefix", makeReq(func(r *ateletpb.CheckpointRequest) {
r.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
r.Config = &ateletpb.CheckpointRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotPrefix: "pause/2"}}
r.Config = &ateletpb.CheckpointRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "pause/2"}}
}), true},
{"traversal local snapshot prefix", makeReq(func(r *ateletpb.CheckpointRequest) {
r.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
r.Config = &ateletpb.CheckpointRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotPrefix: ".."}}
r.Config = &ateletpb.CheckpointRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: ".."}}
}), true},
{"unspecified snapshot type", makeReq(func(r *ateletpb.CheckpointRequest) { r.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_UNSPECIFIED }), true},
{"unspecified snapshot scope", makeReq(func(r *ateletpb.CheckpointRequest) { r.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_UNSPECIFIED }), true},
@@ -346,8 +350,8 @@ func TestValidateRestoreRequest(t *testing.T) {
wantErr bool
}{
{"valid", makeReq(), false},
{"empty snapshot uri", makeReq(func(r *ateletpb.RestoreRequest) { r.GetExternalConfig().SnapshotUriPrefix = "" }), true},
{"bucketless snapshot uri", makeReq(func(r *ateletpb.RestoreRequest) { r.GetExternalConfig().SnapshotUriPrefix = "relative/path" }), true},
{"empty snapshot uri", makeReq(func(r *ateletpb.RestoreRequest) { r.GetExternalConfig().SnapshotUri = "" }), true},
{"bucketless snapshot uri", makeReq(func(r *ateletpb.RestoreRequest) { r.GetExternalConfig().SnapshotUri = "relative/path" }), true},
{"invalid ateom uid", makeReq(func(r *ateletpb.RestoreRequest) { r.TargetAteomUid = "../escape" }), true},
{"invalid atespace", makeReq(func(r *ateletpb.RestoreRequest) { r.Atespace = "../escape" }), true},
{"invalid actor name", makeReq(func(r *ateletpb.RestoreRequest) { r.ActorName = "../escape" }), true},
@@ -359,41 +363,45 @@ func TestValidateRestoreRequest(t *testing.T) {
}), true},
{"invalid local snapshot prefix", makeReq(func(r *ateletpb.RestoreRequest) {
r.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
r.Config = &ateletpb.RestoreRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotPrefix: ""}}
r.Config = &ateletpb.RestoreRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: ""}}
}), true},
{"local snapshot name escapes its directory", makeReq(func(r *ateletpb.RestoreRequest) {
r.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
r.Config = &ateletpb.RestoreRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "../escape"}}
}), true},
{"nested local snapshot prefix", makeReq(func(r *ateletpb.RestoreRequest) {
r.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
r.Config = &ateletpb.RestoreRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotPrefix: "pause/2"}}
r.Config = &ateletpb.RestoreRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "pause/2"}}
}), true},
{"traversal local snapshot prefix", makeReq(func(r *ateletpb.RestoreRequest) {
r.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
r.Config = &ateletpb.RestoreRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotPrefix: ".."}}
r.Config = &ateletpb.RestoreRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: ".."}}
}), true},
{"unspecified snapshot type", makeReq(func(r *ateletpb.RestoreRequest) { r.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_UNSPECIFIED }), true},
{"unspecified snapshot scope", makeReq(func(r *ateletpb.RestoreRequest) { r.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_UNSPECIFIED }), true},
{"invalid snapshot scope", makeReq(func(r *ateletpb.RestoreRequest) { r.Scope = ateletpb.SnapshotScope(23) }), true},
{"data-on-golden with golden uri", makeReq(func(r *ateletpb.RestoreRequest) {
r.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN
r.GoldenSnapshotUriPrefix = "gs://bucket/ate-golden/snapshots/1/"
r.GoldenSnapshotUri = "gs://bucket/golden-root/snapshots/ate-golden/golden-1"
}), false},
{"data-on-golden without golden uri", makeReq(func(r *ateletpb.RestoreRequest) {
r.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN
}), true},
{"data-on-golden with bucketless golden uri", makeReq(func(r *ateletpb.RestoreRequest) {
r.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN
r.GoldenSnapshotUriPrefix = "relative/path"
r.GoldenSnapshotUri = "relative/path"
}), true},
// A pause (local) checkpoint may combine with the golden snapshot:
// the golden URI is a top-level field precisely so LOCAL restores
// can carry it.
{"data-on-golden with local checkpoint type", makeReq(func(r *ateletpb.RestoreRequest) {
r.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN
r.GoldenSnapshotUriPrefix = "gs://bucket/ate-golden/snapshots/1/"
r.GoldenSnapshotUri = "gs://bucket/golden-root/snapshots/ate-golden/golden-1"
r.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
r.Config = &ateletpb.RestoreRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotPrefix: "prefix"}}
r.Config = &ateletpb.RestoreRequest_LocalConfig{LocalConfig: &ateletpb.LocalCheckpointConfiguration{SnapshotName: "local-snap-1"}}
}), false},
{"golden uri with non-data-on-golden scope", makeReq(func(r *ateletpb.RestoreRequest) {
r.GoldenSnapshotUriPrefix = "gs://bucket/ate-golden/snapshots/1/"
r.GoldenSnapshotUri = "gs://bucket/golden-root/snapshots/ate-golden/golden-1"
}), true},
}
for _, tc := range tests {
@@ -859,17 +867,17 @@ func TestDownloadCombinedCheckpoint(t *testing.T) {
}
store := mapObjectStorage{objects: map[string][]byte{
"bucket/actors/1/snapshots/2/durable-dir.tar.zstd": zstdBytes(t, "actor durable data"),
"bucket/ate-golden/snapshots/1/config.json.zstd": zstdBytes(t, "golden config"),
"bucket/ate-golden/snapshots/1/memory-ranges.zstd": zstdBytes(t, "golden memory"),
"bucket/ate-golden/snapshots/1/durable-dir.tar.zstd": zstdBytes(t, "golden durable data (must not be downloaded)"),
"bucket/root/snapshots/ate-demo/counter-1-snap/durable-dir.tar.zstd": zstdBytes(t, "actor durable data"),
"bucket/golden-root/snapshots/ate-golden/golden-1/config.json.zstd": zstdBytes(t, "golden config"),
"bucket/golden-root/snapshots/ate-golden/golden-1/memory-ranges.zstd": zstdBytes(t, "golden memory"),
"bucket/golden-root/snapshots/ate-golden/golden-1/durable-dir.tar.zstd": zstdBytes(t, "golden durable data (must not be downloaded)"),
}}
s := &AteomHerder{gcsClient: store}
dstDir := t.TempDir()
err := s.downloadCombinedCheckpoint(context.Background(),
"gs://bucket/actors/1/snapshots/2/",
"gs://bucket/ate-golden/snapshots/1/",
"gs://bucket/root/snapshots/ate-demo/counter-1-snap",
"gs://bucket/golden-root/snapshots/ate-golden/golden-1",
dstDir,
[]string{"durable-dir.tar"},
[]string{"config.json", "memory-ranges", "durable-dir.tar"})
docs/api-guide.md
+32 -2
View File
@@ -143,7 +143,7 @@ The `ActorTemplate` defines the code, environment, and state-management policies
| `containers` | `[]Container` | **Required.** The workload definition — see [Container Fields](#container-fields) below. Each container may also declare an optional `readyz` HTTP probe — see [Container Readiness Probe](#container-readiness-probe-readyz). |
| `sandboxClass` | `string` | Optional. The sandbox runtime family this template's actors require: `gvisor` (default) or `microvm`. Only `WorkerPool`s whose `sandboxClass` matches are eligible. |
| `workerSelector` | `*LabelSelector` | Optional. Gates which `WorkerPool`s actors from this template may use, by matching against each pool's labels. If unset, all pools are eligible (subject to the actor's own `worker_selector`). |
| `snapshotsConfig` | `SnapshotsConfig` | **Required.** GCS bucket and folder where memory snapshots are stored. |
| `snapshotsConfig` | `SnapshotsConfig` | **Required.** The base object-storage location snapshots are written under, plus the pause/commit/resume scopes. See [Snapshot Storage Layout](#snapshot-storage-layout). |
| `pauseImage` | `string` | **Required.** The image used for the sandbox root (e.g. `gcr.io/gke-release/pause`). |
| `volumes` | `[]Volume` | Optional. Volumes the containers may mount, each either a `durableDir` or an `externalVolumeTemplate`. Every declared volume must be mounted by at least one container. A `microvm` template may declare several `durableDir` volumes; a `gvisor` template is limited to one, and `externalVolumeTemplate` is `gvisor`-only. |
@@ -226,9 +226,39 @@ spec:
matchLabels:
workload: secret-agent
snapshotsConfig:
location: gs://my-bucket/snapshots/secret-agent/
location: gs://my-bucket/secret-agent
```
### Snapshot Storage Layout
`snapshotsConfig.location` is a **base prefix**, not the address of any one snapshot. Every snapshot taken from the template lands at:
```
<location>/snapshots/<atespace>/<snapshot name>
```
and the objects of that snapshot (its manifest, memory image, durable-data tar) are named below it. So for the template above, a snapshot named `f47ac10b-…` of an actor in atespace `team-a` is stored at `gs://my-bucket/secret-agent/snapshots/team-a/f47ac10b-…`, and the template's golden snapshot — the golden actor lives in the reserved `ate-golden` atespace — at `gs://my-bucket/secret-agent/snapshots/ate-golden/<name>`.
Each `ActorSnapshot` reports its own address in the **output-only** `snapshotUri` field. It is recorded when the snapshot is written, not recomputed on read, so the layout can change in future versions without stranding existing snapshots. Do not send it on input; parse it only against the scheme above.
An `ActorTemplate` is namespaced but an atespace is the global isolation boundary, so one `location` holds snapshots for many atespaces. The `<atespace>` level exists so that access can be granted per tenant: an object-storage policy can only condition on an **object-name prefix**, and cannot read the identity recorded inside a snapshot's manifest. Binding a per-atespace grant on GCS looks like:
```yaml
# Read-only on team-a's snapshots for this template, and nothing else.
- members: ["serviceAccount:node-runtime@my-project.iam.gserviceaccount.com"]
role: roles/storage.objectViewer
condition:
title: team-a-snapshots
expression: >
resource.name.startsWith(
"projects/_/buckets/my-bucket/objects/secret-agent/snapshots/team-a/")
```
Two consequences worth planning for:
- **A published snapshot is read from the atespace that took it.** Cloning across atespaces via a `PUBLISHED` tag reads the source atespace's prefix, so the reader needs a grant covering it — the target atespace's grant is not enough.
- **A location containing its own `snapshots` segment is legal but confusing.** `gs://my-bucket/snapshots/secret-agent` yields `gs://my-bucket/snapshots/secret-agent/snapshots/<atespace>/<name>`. It parses correctly; it just reads badly in a policy.
---
## 3. SandboxConfig: Sandbox Binaries
internal/proto/ateletpb/atelet.pb.go
+28 -34
View File
@@ -1219,11 +1219,13 @@ func (*RunResponse) Descriptor() ([]byte, []int) {
type LocalCheckpointConfiguration struct {
state protoimpl.MessageState `protogen:"open.v1"`
// A sub-directory on the local filesystem below which the checkpoint data will be
// stored. The structure of the checkpoint should generally be treated as opaque.
SnapshotPrefix string `protobuf:"bytes,1,opt,name=snapshot_prefix,json=snapshotPrefix,proto3" json:"snapshot_prefix,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
// Names a sub-directory on the local filesystem below which the checkpoint
// data will be stored. atelet decides where that directory lives, from the
// actor's UID, so this is a bare name and must not contain a path separator.
// The structure of the checkpoint should generally be treated as opaque.
SnapshotName string `protobuf:"bytes,1,opt,name=snapshot_name,json=snapshotName,proto3" json:"snapshot_name,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *LocalCheckpointConfiguration) Reset() {
@@ -1256,28 +1258,20 @@ func (*LocalCheckpointConfiguration) Descriptor() ([]byte, []int) {
return file_atelet_proto_rawDescGZIP(), []int{17}
}
func (x *LocalCheckpointConfiguration) GetSnapshotPrefix() string {
func (x *LocalCheckpointConfiguration) GetSnapshotName() string {
if x != nil {
return x.SnapshotPrefix
return x.SnapshotName
}
return ""
}
type ExternalCheckpointConfiguration struct {
state protoimpl.MessageState `protogen:"open.v1"`
// An object storage URI prefix below which the checkpoint data will be
// stored.
//
// The structure of the checkpoint should generally be treated as opaque. For
// gVisor, the checkpoint consists of a checkpoint.img file that contains the
// memory, sentry state, and filesystem deltas. ateom will write additional
// files to store the WorkloadSpec, which must be reassembled before gVisor
// can restore a checkpoint.
//
// For example: "gs://bucket/actors/1234/snapshots/5678/"
SnapshotUriPrefix string `protobuf:"bytes,1,opt,name=snapshot_uri_prefix,json=snapshotUriPrefix,proto3" json:"snapshot_uri_prefix,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
// The object storage URI of the snapshot to write. Object names are appended
// to it, so it addresses the snapshot as a whole rather than any one object.
SnapshotUri string `protobuf:"bytes,1,opt,name=snapshot_uri,json=snapshotUri,proto3" json:"snapshot_uri,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ExternalCheckpointConfiguration) Reset() {
@@ -1310,9 +1304,9 @@ func (*ExternalCheckpointConfiguration) Descriptor() ([]byte, []int) {
return file_atelet_proto_rawDescGZIP(), []int{18}
}
func (x *ExternalCheckpointConfiguration) GetSnapshotUriPrefix() string {
func (x *ExternalCheckpointConfiguration) GetSnapshotUri() string {
if x != nil {
return x.SnapshotUriPrefix
return x.SnapshotUri
}
return ""
}
@@ -1535,13 +1529,13 @@ 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 prefix of the ActorTemplate's golden snapshot.
// 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.
GoldenSnapshotUriPrefix string `protobuf:"bytes,12,opt,name=golden_snapshot_uri_prefix,json=goldenSnapshotUriPrefix,proto3" json:"golden_snapshot_uri_prefix,omitempty"`
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"`
unknownFields protoimpl.UnknownFields
@@ -1666,9 +1660,9 @@ func (x *RestoreRequest) GetScope() SnapshotScope {
return SnapshotScope_SNAPSHOT_SCOPE_UNSPECIFIED
}
func (x *RestoreRequest) GetGoldenSnapshotUriPrefix() string {
func (x *RestoreRequest) GetGoldenSnapshotUri() string {
if x != nil {
return x.GoldenSnapshotUriPrefix
return x.GoldenSnapshotUri
}
return ""
}
@@ -1817,11 +1811,11 @@ const file_atelet_proto_rawDesc = "" +
"\rHTTPGetAction\x12\x12\n" +
"\x04path\x18\x01 \x01(\tR\x04path\x12\x12\n" +
"\x04port\x18\x02 \x01(\x05R\x04port\"\r\n" +
"\vRunResponse\"G\n" +
"\x1cLocalCheckpointConfiguration\x12'\n" +
"\x0fsnapshot_prefix\x18\x01 \x01(\tR\x0esnapshotPrefix\"Q\n" +
"\x1fExternalCheckpointConfiguration\x12.\n" +
"\x13snapshot_uri_prefix\x18\x01 \x01(\tR\x11snapshotUriPrefix\"\xab\x04\n" +
"\vRunResponse\"C\n" +
"\x1cLocalCheckpointConfiguration\x12#\n" +
"\rsnapshot_name\x18\x01 \x01(\tR\fsnapshotName\"D\n" +
"\x1fExternalCheckpointConfiguration\x12!\n" +
"\fsnapshot_uri\x18\x01 \x01(\tR\vsnapshotUri\"\xab\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" +
@@ -1837,7 +1831,7 @@ const file_atelet_proto_rawDesc = "" +
" \x01(\v2'.atelet.ExternalCheckpointConfigurationH\x00R\x0eexternalConfig\x12+\n" +
"\x05scope\x18\v \x01(\x0e2\x15.atelet.SnapshotScopeR\x05scopeB\b\n" +
"\x06config\"\x14\n" +
"\x12CheckpointResponse\"\xbb\x05\n" +
"\x12CheckpointResponse\"\xae\x05\n" +
"\x0eRestoreRequest\x12(\n" +
"\x10target_ateom_uid\x18\x01 \x01(\tR\x0etargetAteomUid\x12\x1a\n" +
"\batespace\x18\x02 \x01(\tR\batespace\x12\x1d\n" +
@@ -1851,8 +1845,8 @@ const file_atelet_proto_rawDesc = "" +
"\flocal_config\x18\t \x01(\v2$.atelet.LocalCheckpointConfigurationH\x00R\vlocalConfig\x12R\n" +
"\x0fexternal_config\x18\n" +
" \x01(\v2'.atelet.ExternalCheckpointConfigurationH\x00R\x0eexternalConfig\x12+\n" +
"\x05scope\x18\v \x01(\x0e2\x15.atelet.SnapshotScopeR\x05scope\x12;\n" +
"\x1agolden_snapshot_uri_prefix\x18\f \x01(\tR\x17goldenSnapshotUriPrefix\x12A\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\x01B\b\n" +
"\x06configB\x11\n" +
"\x0f_egress_gateway\"\x11\n" +
internal/proto/ateletpb/atelet.proto
+10 -16
View File
@@ -179,23 +179,17 @@ message RunResponse {
}
message LocalCheckpointConfiguration {
// A sub-directory on the local filesystem below which the checkpoint data will be
// stored. The structure of the checkpoint should generally be treated as opaque.
string snapshot_prefix = 1;
// Names a sub-directory on the local filesystem below which the checkpoint
// data will be stored. atelet decides where that directory lives, from the
// actor's UID, so this is a bare name and must not contain a path separator.
// The structure of the checkpoint should generally be treated as opaque.
string snapshot_name = 1;
}
message ExternalCheckpointConfiguration {
// An object storage URI prefix below which the checkpoint data will be
// stored.
//
// The structure of the checkpoint should generally be treated as opaque. For
// gVisor, the checkpoint consists of a checkpoint.img file that contains the
// memory, sentry state, and filesystem deltas. ateom will write additional
// files to store the WorkloadSpec, which must be reassembled before gVisor
// can restore a checkpoint.
//
// For example: "gs://bucket/actors/1234/snapshots/5678/"
string snapshot_uri_prefix = 1;
// The object storage URI of the snapshot to write. Object names are appended
// to it, so it addresses the snapshot as a whole rather than any one object.
string snapshot_uri = 1;
}
enum CheckpointType {
@@ -282,13 +276,13 @@ message RestoreRequest {
// What content to restore from the checkpoint.
SnapshotScope scope = 11;
// The object storage URI prefix of the ActorTemplate's golden snapshot.
// 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.
string golden_snapshot_uri_prefix = 12;
string golden_snapshot_uri = 12;
// When absent, actor traffic uses direct egress instead of atunnel.
optional EgressGateway egress_gateway = 13;
internal/proto/ateompb/ateom.pb.go
+28 -25
View File
@@ -686,15 +686,17 @@ type CheckpointWorkloadRequest struct {
ActorTemplateName string `protobuf:"bytes,5,opt,name=actor_template_name,json=actorTemplateName,proto3" json:"actor_template_name,omitempty"`
RunscPath string `protobuf:"bytes,6,opt,name=runsc_path,json=runscPath,proto3" json:"runsc_path,omitempty"`
Spec *WorkloadSpec `protobuf:"bytes,7,opt,name=spec,proto3" json:"spec,omitempty"`
// An object storage URI prefix below which the checkpoint data will be
// stored.
// The object storage URI of the snapshot to write. Object names are appended
// to it, so it addresses the snapshot as a whole rather than any one object.
//
// The structure of the checkpoint should generally be treated as opaque. For
// gVisor, the checkpoint consists of a checkpoint.img file that contains the
// memory, sentry state, and filesystem deltas.
//
// For example: "gs://bucket/actors/1234/snapshots/5678/"
SnapshotUriPrefix string `protobuf:"bytes,8,opt,name=snapshot_uri_prefix,json=snapshotUriPrefix,proto3" json:"snapshot_uri_prefix,omitempty"`
// The control plane sends an ActorSnapshot's snapshot_uri, whose layout is
// "<snapshotsConfig.location>/snapshots/<atespace>/<snapshot name>". For
// example: "gs://bucket/root/snapshots/team-a/5678".
SnapshotUri string `protobuf:"bytes,8,opt,name=snapshot_uri,json=snapshotUri,proto3" json:"snapshot_uri,omitempty"`
// runtime_asset_paths maps a runtime asset name to the local on-disk path
// atelet fetched it to (see RunWorkloadRequest). Empty for gVisor.
RuntimeAssetPaths map[string]string `protobuf:"bytes,9,rep,name=runtime_asset_paths,json=runtimeAssetPaths,proto3" json:"runtime_asset_paths,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
@@ -783,9 +785,9 @@ func (x *CheckpointWorkloadRequest) GetSpec() *WorkloadSpec {
return nil
}
func (x *CheckpointWorkloadRequest) GetSnapshotUriPrefix() string {
func (x *CheckpointWorkloadRequest) GetSnapshotUri() string {
if x != nil {
return x.SnapshotUriPrefix
return x.SnapshotUri
}
return ""
}
@@ -860,8 +862,9 @@ type RestoreWorkloadRequest struct {
ActorTemplateName string `protobuf:"bytes,5,opt,name=actor_template_name,json=actorTemplateName,proto3" json:"actor_template_name,omitempty"`
RunscPath string `protobuf:"bytes,6,opt,name=runsc_path,json=runscPath,proto3" json:"runsc_path,omitempty"`
Spec *WorkloadSpec `protobuf:"bytes,7,opt,name=spec,proto3" json:"spec,omitempty"`
// The object storage URI prefix of the snapshot to restore.
SnapshotUriPrefix string `protobuf:"bytes,8,opt,name=snapshot_uri_prefix,json=snapshotUriPrefix,proto3" json:"snapshot_uri_prefix,omitempty"`
// The object storage URI of the snapshot to restore. Object names are
// appended to it; it addresses the snapshot, not any one object.
SnapshotUri string `protobuf:"bytes,8,opt,name=snapshot_uri,json=snapshotUri,proto3" json:"snapshot_uri,omitempty"`
// runtime_asset_paths maps a runtime asset name to the local on-disk path
// atelet fetched it to (see RunWorkloadRequest). Empty for gVisor.
RuntimeAssetPaths map[string]string `protobuf:"bytes,9,rep,name=runtime_asset_paths,json=runtimeAssetPaths,proto3" json:"runtime_asset_paths,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
@@ -869,12 +872,12 @@ type RestoreWorkloadRequest struct {
Scope SnapshotScope `protobuf:"varint,10,opt,name=scope,proto3,enum=ateom.SnapshotScope" json:"scope,omitempty"`
// When absent, actor traffic uses direct egress instead of atunnel.
EgressGateway *EgressGateway `protobuf:"bytes,12,opt,name=egress_gateway,json=egressGateway,proto3,oneof" json:"egress_gateway,omitempty"`
// The object storage URI prefix of the ActorTemplate's golden snapshot.
// The object storage URI of the ActorTemplate's golden snapshot.
// Set only when scope is SNAPSHOT_SCOPE_DATA_ON_GOLDEN. Mirrors the
// snapshot_uri_prefix contract (field 8).
GoldenSnapshotUriPrefix string `protobuf:"bytes,13,opt,name=golden_snapshot_uri_prefix,json=goldenSnapshotUriPrefix,proto3" json:"golden_snapshot_uri_prefix,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
// snapshot_uri contract (field 8).
GoldenSnapshotUri string `protobuf:"bytes,13,opt,name=golden_snapshot_uri,json=goldenSnapshotUri,proto3" json:"golden_snapshot_uri,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *RestoreWorkloadRequest) Reset() {
@@ -956,9 +959,9 @@ func (x *RestoreWorkloadRequest) GetSpec() *WorkloadSpec {
return nil
}
func (x *RestoreWorkloadRequest) GetSnapshotUriPrefix() string {
func (x *RestoreWorkloadRequest) GetSnapshotUri() string {
if x != nil {
return x.SnapshotUriPrefix
return x.SnapshotUri
}
return ""
}
@@ -984,9 +987,9 @@ func (x *RestoreWorkloadRequest) GetEgressGateway() *EgressGateway {
return nil
}
func (x *RestoreWorkloadRequest) GetGoldenSnapshotUriPrefix() string {
func (x *RestoreWorkloadRequest) GetGoldenSnapshotUri() string {
if x != nil {
return x.GoldenSnapshotUriPrefix
return x.GoldenSnapshotUri
}
return ""
}
@@ -1265,7 +1268,7 @@ const file_ateom_proto_rawDesc = "" +
"\rHTTPGetAction\x12\x12\n" +
"\x04path\x18\x01 \x01(\tR\x04path\x12\x12\n" +
"\x04port\x18\x02 \x01(\x05R\x04port\"\x15\n" +
"\x13RunWorkloadResponse\"\xb0\x04\n" +
"\x13RunWorkloadResponse\"\xa3\x04\n" +
"\x19CheckpointWorkloadRequest\x12\x1a\n" +
"\batespace\x18\x01 \x01(\tR\batespace\x12\x1d\n" +
"\n" +
@@ -1275,8 +1278,8 @@ const file_ateom_proto_rawDesc = "" +
"\x13actor_template_name\x18\x05 \x01(\tR\x11actorTemplateName\x12\x1d\n" +
"\n" +
"runsc_path\x18\x06 \x01(\tR\trunscPath\x12'\n" +
"\x04spec\x18\a \x01(\v2\x13.ateom.WorkloadSpecR\x04spec\x12.\n" +
"\x13snapshot_uri_prefix\x18\b \x01(\tR\x11snapshotUriPrefix\x12g\n" +
"\x04spec\x18\a \x01(\v2\x13.ateom.WorkloadSpecR\x04spec\x12!\n" +
"\fsnapshot_uri\x18\b \x01(\tR\vsnapshotUri\x12g\n" +
"\x13runtime_asset_paths\x18\t \x03(\v27.ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntryR\x11runtimeAssetPaths\x12*\n" +
"\x05scope\x18\n" +
" \x01(\x0e2\x14.ateom.SnapshotScopeR\x05scope\x1aD\n" +
@@ -1284,7 +1287,7 @@ const file_ateom_proto_rawDesc = "" +
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"C\n" +
"\x1aCheckpointWorkloadResponse\x12%\n" +
"\x0esnapshot_files\x18\x01 \x03(\tR\rsnapshotFiles\"\xbc\x05\n" +
"\x0esnapshot_files\x18\x01 \x03(\tR\rsnapshotFiles\"\xa2\x05\n" +
"\x16RestoreWorkloadRequest\x12\x1a\n" +
"\batespace\x18\x01 \x01(\tR\batespace\x12\x1d\n" +
"\n" +
@@ -1294,13 +1297,13 @@ const file_ateom_proto_rawDesc = "" +
"\x13actor_template_name\x18\x05 \x01(\tR\x11actorTemplateName\x12\x1d\n" +
"\n" +
"runsc_path\x18\x06 \x01(\tR\trunscPath\x12'\n" +
"\x04spec\x18\a \x01(\v2\x13.ateom.WorkloadSpecR\x04spec\x12.\n" +
"\x13snapshot_uri_prefix\x18\b \x01(\tR\x11snapshotUriPrefix\x12d\n" +
"\x04spec\x18\a \x01(\v2\x13.ateom.WorkloadSpecR\x04spec\x12!\n" +
"\fsnapshot_uri\x18\b \x01(\tR\vsnapshotUri\x12d\n" +
"\x13runtime_asset_paths\x18\t \x03(\v24.ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntryR\x11runtimeAssetPaths\x12*\n" +
"\x05scope\x18\n" +
" \x01(\x0e2\x14.ateom.SnapshotScopeR\x05scope\x12@\n" +
"\x0eegress_gateway\x18\f \x01(\v2\x14.ateom.EgressGatewayH\x00R\regressGateway\x88\x01\x01\x12;\n" +
"\x1agolden_snapshot_uri_prefix\x18\r \x01(\tR\x17goldenSnapshotUriPrefix\x1aD\n" +
"\x0eegress_gateway\x18\f \x01(\v2\x14.ateom.EgressGatewayH\x00R\regressGateway\x88\x01\x01\x12.\n" +
"\x13golden_snapshot_uri\x18\r \x01(\tR\x11goldenSnapshotUri\x1aD\n" +
"\x16RuntimeAssetPathsEntry\x12\x10\n" +
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01B\x11\n" +
internal/proto/ateompb/ateom.proto
+12 -9
View File
@@ -185,15 +185,17 @@ message CheckpointWorkloadRequest {
WorkloadSpec spec = 7;
// An object storage URI prefix below which the checkpoint data will be
// stored.
// The object storage URI of the snapshot to write. Object names are appended
// to it, so it addresses the snapshot as a whole rather than any one object.
//
// The structure of the checkpoint should generally be treated as opaque. For
// gVisor, the checkpoint consists of a checkpoint.img file that contains the
// memory, sentry state, and filesystem deltas.
//
// For example: "gs://bucket/actors/1234/snapshots/5678/"
string snapshot_uri_prefix = 8;
// The control plane sends an ActorSnapshot's snapshot_uri, whose layout is
// "<snapshotsConfig.location>/snapshots/<atespace>/<snapshot name>". For
// example: "gs://bucket/root/snapshots/team-a/5678".
string snapshot_uri = 8;
// runtime_asset_paths maps a runtime asset name to the local on-disk path
// atelet fetched it to (see RunWorkloadRequest). Empty for gVisor.
@@ -221,8 +223,9 @@ message RestoreWorkloadRequest {
WorkloadSpec spec = 7;
// The object storage URI prefix of the snapshot to restore.
string snapshot_uri_prefix = 8;
// The object storage URI of the snapshot to restore. Object names are
// appended to it; it addresses the snapshot, not any one object.
string snapshot_uri = 8;
// runtime_asset_paths maps a runtime asset name to the local on-disk path
// atelet fetched it to (see RunWorkloadRequest). Empty for gVisor.
@@ -234,10 +237,10 @@ message RestoreWorkloadRequest {
// When absent, actor traffic uses direct egress instead of atunnel.
optional EgressGateway egress_gateway = 12;
// The object storage URI prefix of the ActorTemplate's golden snapshot.
// The object storage URI of the ActorTemplate's golden snapshot.
// Set only when scope is SNAPSHOT_SCOPE_DATA_ON_GOLDEN. Mirrors the
// snapshot_uri_prefix contract (field 8).
string golden_snapshot_uri_prefix = 13;
// snapshot_uri contract (field 8).
string golden_snapshot_uri = 13;
}
message RestoreWorkloadResponse {
internal/resources/snapshot.go
+100
View File
@@ -0,0 +1,100 @@
// 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 resources
import (
"fmt"
"net/url"
"strings"
"github.com/google/uuid"
)
const snapshotsPathSegment = "snapshots"
// NewSnapshotName returns a unique name for a new snapshot, durable or
// node-local.
func NewSnapshotName() string {
return uuid.NewString()
}
// SnapshotURI is where one ActorSnapshot's objects live in object storage:
// an ActorTemplate's snapshotsConfig.location, plus /snapshots/<atespace>/<name>.
//
// gs://bucket/root location
// gs://bucket/root/snapshots/team-a/<name> this URI
// gs://bucket/root/snapshots/team-a/<name>/... an object in the snapshot
type SnapshotURI struct {
uri string
location string
atespace string
name string
}
// NewSnapshotURI returns the URI of a snapshot of an actor in a given
// atespace, stored and under an ActorTemplate's snapshotsConfig.location.
func NewSnapshotURI(location, atespace, name string) (SnapshotURI, error) {
if err := ValidateSnapshotLocation(location); err != nil {
return SnapshotURI{}, err
}
if !IsValidResourceName(atespace) {
return SnapshotURI{}, fmt.Errorf("invalid snapshot URI: atespace %q is not a valid resource name", atespace)
}
if !IsValidResourceName(name) {
return SnapshotURI{}, fmt.Errorf("invalid snapshot URI: snapshot name %q is not a valid resource name", name)
}
uri, err := url.JoinPath(location, snapshotsPathSegment, atespace, name)
if err != nil {
return SnapshotURI{}, fmt.Errorf("invalid snapshot URI: %w", err)
}
return SnapshotURI{uri: uri, location: location, atespace: atespace, name: name}, nil
}
// ParseSnapshotURI parses a given snapshot URI.
func ParseSnapshotURI(uri string) (SnapshotURI, error) {
u, err := url.Parse(uri)
if err != nil {
return SnapshotURI{}, fmt.Errorf("invalid snapshot URI %q: %v", uri, err)
}
segments := strings.Split(strings.TrimSuffix(u.Path, "/"), "/")
if len(segments) < 3 || segments[len(segments)-3] != snapshotsPathSegment {
return SnapshotURI{}, fmt.Errorf("invalid snapshot URI %q", uri)
}
atespace, name := segments[len(segments)-2], segments[len(segments)-1]
u.Path = strings.Join(segments[:len(segments)-3], "/")
return NewSnapshotURI(u.String(), atespace, name)
}
// Location returns the ActorTemplate snapshotsConfig.location this snapshot is stored under.
func (u SnapshotURI) Location() string { return u.location }
// Atespace returns the atespace of the actor the snapshot was taken from.
func (u SnapshotURI) Atespace() string { return u.atespace }
// Name returns the snapshot's resource name.
func (u SnapshotURI) Name() string { return u.name }
// IsZero reports whether u is the zero SnapshotURI.
func (u SnapshotURI) IsZero() bool { return u == SnapshotURI{} }
func (u SnapshotURI) String() string { return u.uri }
// ObjectURI returns the address of a single object stored within the
// snapshot.
func (u SnapshotURI) ObjectURI(name string) (string, error) {
return url.JoinPath(u.String(), name)
}
internal/resources/snapshot_test.go
+215
View File
@@ -0,0 +1,215 @@
// 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 resources
import (
"testing"
)
func TestNewSnapshotURI(t *testing.T) {
tests := []struct {
name string
location string
atespace string
snapshot string
want string
wantErr bool
}{
{
name: "no trailing slash", location: "gs://bucket/root", atespace: "team-a", snapshot: "snap-1",
want: "gs://bucket/root/snapshots/team-a/snap-1",
},
{
name: "trailing slash", location: "gs://bucket/root/", atespace: "team-a", snapshot: "snap-1",
want: "gs://bucket/root/snapshots/team-a/snap-1",
},
{
name: "bucket only", location: "gs://bucket", atespace: "team-a", snapshot: "snap-1",
want: "gs://bucket/snapshots/team-a/snap-1",
},
{
name: "location containing a snapshots segment", location: "gs://my-bucket/snapshots/secret-agent", atespace: "team-a", snapshot: "snap-1",
want: "gs://my-bucket/snapshots/secret-agent/snapshots/team-a/snap-1",
},
{name: "empty location", location: "", atespace: "team-a", snapshot: "snap-1", wantErr: true},
{name: "location without a bucket", location: "/root", atespace: "team-a", snapshot: "snap-1", wantErr: true},
{
name: "location with a query", location: "gs://bucket/root?generation=1", atespace: "team-a", snapshot: "snap-1", wantErr: true,
},
{name: "invalid atespace", location: "gs://bucket/root", atespace: "Team_A", snapshot: "snap-1", wantErr: true},
{name: "empty atespace", location: "gs://bucket/root", atespace: "", snapshot: "snap-1", wantErr: true},
{name: "invalid snapshot name", location: "gs://bucket/root", atespace: "team-a", snapshot: "2026-08-05T10:04:05Z", wantErr: true},
{name: "empty snapshot name", location: "gs://bucket/root", atespace: "team-a", snapshot: "", wantErr: true},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
got, err := NewSnapshotURI(tc.location, tc.atespace, tc.snapshot)
if (err != nil) != tc.wantErr {
t.Fatalf("NewSnapshotURI(%q, %q, %q) error = %v, wantErr %t", tc.location, tc.atespace, tc.snapshot, err, tc.wantErr)
}
if got.String() != tc.want {
t.Errorf("NewSnapshotURI(%q, %q, %q) = %q, want %q", tc.location, tc.atespace, tc.snapshot, got, tc.want)
}
if tc.wantErr && !got.IsZero() {
t.Errorf("NewSnapshotURI(%q, %q, %q) returned %q alongside an error, want the zero value", tc.location, tc.atespace, tc.snapshot, got)
}
})
}
}
func TestParseSnapshotURI(t *testing.T) {
tests := []struct {
name string
uri string
wantLocation string
wantAtespace string
wantName string
wantErr bool
}{
{
name: "reads all three parts",
uri: "gs://bucket/root/snapshots/team-a/f47ac10b-58cc-4372-a567-0e02b2c3d479",
wantLocation: "gs://bucket/root",
wantAtespace: "team-a",
wantName: "f47ac10b-58cc-4372-a567-0e02b2c3d479",
},
{
name: "tolerates a trailing slash",
uri: "gs://bucket/root/snapshots/team-a/snap-1/",
wantLocation: "gs://bucket/root",
wantAtespace: "team-a",
wantName: "snap-1",
},
{
name: "bucket-only location",
uri: "gs://bucket/snapshots/team-a/snap-1",
wantLocation: "gs://bucket",
wantAtespace: "team-a",
wantName: "snap-1",
},
{
name: "location containing a snapshots segment",
uri: "gs://my-bucket/snapshots/secret-agent/snapshots/team-a/snap-1",
wantLocation: "gs://my-bucket/snapshots/secret-agent",
wantAtespace: "team-a",
wantName: "snap-1",
},
{
name: "rejects a URI with no atespace segment",
uri: "gs://bucket/root/snapshots/snap-1",
wantErr: true,
},
{
name: "atespace named snapshots",
uri: "gs://bucket/root/snapshots/snapshots/snap-1",
wantLocation: "gs://bucket/root",
wantAtespace: "snapshots",
wantName: "snap-1",
},
{
name: "rejects an object within a snapshot",
uri: "gs://bucket/root/snapshots/team-a/snap-1/manifest.json",
wantErr: true,
},
{name: "rejects a URI with no snapshots segment", uri: "gs://bucket/root/team-a/snap-1", wantErr: true},
{name: "rejects a bare name", uri: "snap-1", wantErr: true},
{name: "rejects an empty URI", uri: "", wantErr: true},
{name: "rejects a missing name", uri: "gs://bucket/root/snapshots/team-a/", wantErr: true},
{name: "rejects a missing bucket", uri: "/root/snapshots/team-a/snap-1", wantErr: true},
{
name: "rejects a name that is not a resource name",
uri: "gs://bucket/root/snapshots/team-a/2026-08-05T10:04:05Z-ABCDEFG",
wantErr: true,
},
{name: "rejects an atespace that is not a resource name", uri: "gs://bucket/root/snapshots/Team_A/snap-1", wantErr: true},
{
name: "rejects a URI with a trailing query",
uri: "gs://bucket/root/snapshots/team-a/snap-1?generation=1",
wantErr: true,
},
{name: "rejects a URI with a trailing fragment", uri: "gs://bucket/root/snapshots/team-a/snap-1#frag", wantErr: true},
{name: "rejects malformed URI syntax", uri: "gs://bucket/root/snapshots/team-a/snap-1%zz", wantErr: true},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
got, err := ParseSnapshotURI(tc.uri)
if (err != nil) != tc.wantErr {
t.Fatalf("ParseSnapshotURI(%q) error = %v, wantErr %t", tc.uri, err, tc.wantErr)
}
if got.Location() != tc.wantLocation {
t.Errorf("ParseSnapshotURI(%q).Location() = %q, want %q", tc.uri, got.Location(), tc.wantLocation)
}
if got.Atespace() != tc.wantAtespace {
t.Errorf("ParseSnapshotURI(%q).Atespace() = %q, want %q", tc.uri, got.Atespace(), tc.wantAtespace)
}
if got.Name() != tc.wantName {
t.Errorf("ParseSnapshotURI(%q).Name() = %q, want %q", tc.uri, got.Name(), tc.wantName)
}
})
}
}
func TestSnapshotURIObject(t *testing.T) {
uri, err := NewSnapshotURI("gs://bucket/root", "team-a", "snap-1")
if err != nil {
t.Fatalf("NewSnapshotURI: %v", err)
}
tests := []struct {
name string
objectName string
want string
wantErr bool
}{
{
name: "manifest", objectName: "manifest.json",
want: "gs://bucket/root/snapshots/team-a/snap-1/manifest.json",
},
{
name: "image file", objectName: "memory.img.zstd",
want: "gs://bucket/root/snapshots/team-a/snap-1/memory.img.zstd",
},
{
name: "malformed escape", objectName: "100%done.img", wantErr: true,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
got, err := uri.ObjectURI(tc.objectName)
if (err != nil) != tc.wantErr {
t.Fatalf("ObjectURI(%q) error = %v, wantErr %t", tc.objectName, err, tc.wantErr)
}
if got != tc.want {
t.Errorf("ObjectURI(%q) = %q, want %q", tc.objectName, got, tc.want)
}
})
}
}
func TestSnapshotURIZeroValue(t *testing.T) {
var zero SnapshotURI
if !zero.IsZero() {
t.Error("the zero SnapshotURI does not report IsZero")
}
if got := zero.String(); got != "" {
t.Errorf("zero SnapshotURI renders as %q, want the empty string", got)
}
uri, err := NewSnapshotURI("gs://bucket/root", "team-a", "snap-1")
if err != nil {
t.Fatalf("NewSnapshotURI: %v", err)
}
if uri.IsZero() {
t.Errorf("%q reports IsZero", uri)
}
}
internal/resources/validate.go
+22 -30
View File
@@ -179,41 +179,33 @@ func ValidateRunscHash(sha256Hash string) error {
return nil
}
// ValidateSnapshotURIPrefix ensures a checkpoint/restore snapshot location is
// a well-formed URI with a bucket, so a bad prefix fails fast at the RPC
// boundary instead of deep inside an object-storage call. It deliberately
// does not restrict the scheme: the storage layer only uses the host (bucket)
// and path, and which schemes are acceptable is a storage-backend policy, not
// a per-RPC one. The local paths used for snapshot upload/download are
// derived from the separately validated actor ref, not from this URI, so this
// is a sanity check rather than a path-traversal guard.
func ValidateSnapshotURIPrefix(prefix string) error {
u, err := url.Parse(prefix)
// ValidateSnapshotLocation ensures an ActorTemplate's snapshotsConfig.location
// is a well-formed URI with a bucket, so a bad location fails fast instead of
// deep inside an object-storage call. It deliberately does not restrict the
// scheme: the storage layer only uses the host (bucket) and path, and which
// schemes are acceptable is a storage-backend policy, not a per-RPC one. The
// local paths used for snapshot upload/download are derived from the
// separately validated actor ref, not from this URI, so this is a sanity check
// rather than a path-traversal guard.
//
// This validates the base that many snapshots share, not any one snapshot's
// URI; SnapshotURI is the type for the latter, and it applies this check when
// it is built.
func ValidateSnapshotLocation(location string) error {
u, err := url.Parse(location)
if err != nil {
return fmt.Errorf("invalid snapshot URI prefix %q: %v", prefix, err)
return fmt.Errorf("invalid snapshot location %q: %v", location, err)
}
if u.Host == "" {
return fmt.Errorf("invalid snapshot URI prefix %q: missing bucket", prefix)
return fmt.Errorf("invalid snapshot location %q: missing bucket", location)
}
// Object names are appended to the prefix by string concatenation. A
// query, fragment, or userinfo component would swallow the appended name
// when the result is re-parsed (the storage layer uses only host and
// path), silently redirecting the upload/download to a different object.
// Snapshot and object names are appended to the location by string
// concatenation. A query, fragment, or userinfo component would swallow
// the appended name when the result is re-parsed (the storage layer uses
// only host and path), silently redirecting the upload/download to a
// different object.
if u.Opaque != "" || u.User != nil || u.RawQuery != "" || u.Fragment != "" {
return fmt.Errorf("invalid snapshot URI prefix %q: must contain only a scheme, bucket, and path", prefix)
}
return nil
}
// ValidateLocalSnapshotPrefix ensures a local snapshot prefix is a single
// path segment: it is joined onto the actor's checkpoint directory, so a
// nested or relative prefix would escape it or nest below it.
func ValidateLocalSnapshotPrefix(prefix string) error {
if prefix == "" {
return fmt.Errorf("snapshot prefix must be non-empty")
}
if prefix == "." || prefix == ".." || strings.ContainsAny(prefix, `/\`) {
return fmt.Errorf("invalid snapshot prefix %q: must be a single path segment", prefix)
return fmt.Errorf("invalid snapshot location %q: must contain only a scheme, bucket, and path", location)
}
return nil
}
internal/resources/validate_test.go
+3 -28
View File
@@ -271,7 +271,7 @@ func TestValidateRunscHash(t *testing.T) {
}
}
func TestValidateSnapshotURIPrefix(t *testing.T) {
func TestValidateSnapshotLocation(t *testing.T) {
tests := []struct {
name string
prefix string
@@ -295,33 +295,8 @@ func TestValidateSnapshotURIPrefix(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if err := ValidateSnapshotURIPrefix(tt.prefix); (err != nil) != tt.wantErr {
t.Errorf("ValidateSnapshotURIPrefix(%q) err = %v, wantErr %v", tt.prefix, err, tt.wantErr)
}
})
}
}
func TestValidateLocalSnapshotPrefix(t *testing.T) {
tests := []struct {
name string
prefix string
wantErr bool
}{
{"valid", "pause", false},
{"valid with dash and digits", "pause-2", false},
{"empty", "", true},
{"dot", ".", true},
{"dotdot", "..", true},
{"nested", "pause/2", true},
{"absolute", "/pause", true},
{"traversal", "../other-actor", true},
{"backslash", `pause\2`, true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if err := ValidateLocalSnapshotPrefix(tt.prefix); (err != nil) != tt.wantErr {
t.Errorf("ValidateLocalSnapshotPrefix(%q) err = %v, wantErr %v", tt.prefix, err, tt.wantErr)
if err := ValidateSnapshotLocation(tt.prefix); (err != nil) != tt.wantErr {
t.Errorf("ValidateSnapshotLocation(%q) err = %v, wantErr %v", tt.prefix, err, tt.wantErr)
}
})
}
manifests/ate-install/generated/ate.dev_actortemplates.yaml
+3 -1
View File
@@ -300,7 +300,9 @@ spec:
description: Snapshots configuration for the actor.
properties:
location:
description: Location to store snapshots in.
description: |-
Location is the base object-storage URI snapshots of this template's
actors are stored under.
minLength: 1
type: string
onCommit:
pkg/api/v1alpha1/actortemplate_types.go
+2 -1
View File
@@ -317,7 +317,8 @@ type OnResumeConfig struct {
// +kubebuilder:validation:XValidation:rule="(has(self.onPause) ? self.onPause : 'Full') == 'Full' || (has(self.onCommit) ? self.onCommit : 'Full') == (has(self.onPause) ? self.onPause : 'Full')",message="onCommit must be a subset of onPause"
type SnapshotsConfig struct {
// Location to store snapshots in.
// Location is the base object-storage URI snapshots of this template's
// actors are stored under.
//
// +required
// +kubebuilder:validation:MinLength=1
pkg/proto/ateapipb/ateapi.pb.go
+40 -22
View File
@@ -356,8 +356,9 @@ func (Worker_State) EnumDescriptor() ([]byte, []int) {
}
type LocalSnapshotInfo struct {
state protoimpl.MessageState `protogen:"open.v1"`
SnapshotPrefix string `protobuf:"bytes,1,opt,name=snapshot_prefix,json=snapshotPrefix,proto3" json:"snapshot_prefix,omitempty"`
state protoimpl.MessageState `protogen:"open.v1"`
// The name of the local checkpoint on each of the nodes below.
SnapshotName string `protobuf:"bytes,1,opt,name=snapshot_name,json=snapshotName,proto3" json:"snapshot_name,omitempty"`
// Node VMs that have local snapshots for this actor, while it's PAUSED.
NodeVmsWithLocalSnapshots []string `protobuf:"bytes,2,rep,name=node_vms_with_local_snapshots,json=nodeVmsWithLocalSnapshots,proto3" json:"node_vms_with_local_snapshots,omitempty"`
// Scope the pause checkpoint captured (the template's onPause at pause
@@ -397,9 +398,9 @@ func (*LocalSnapshotInfo) Descriptor() ([]byte, []int) {
return file_ateapi_proto_rawDescGZIP(), []int{0}
}
func (x *LocalSnapshotInfo) GetSnapshotPrefix() string {
func (x *LocalSnapshotInfo) GetSnapshotName() string {
if x != nil {
return x.SnapshotPrefix
return x.SnapshotName
}
return ""
}
@@ -649,8 +650,8 @@ type Actor struct {
Status Actor_Status `protobuf:"varint,4,opt,name=status,proto3,enum=ateapi.Actor_Status" json:"status,omitempty"`
// worker_assignment points at the worker currently hosting this Actor.
// Unset whenever the Actor has no worker (SUSPENDED, PAUSED, CRASHED).
WorkerAssignment *WorkerAssignment `protobuf:"bytes,5,opt,name=worker_assignment,json=workerAssignment,proto3" json:"worker_assignment,omitempty"`
InProgressSnapshot string `protobuf:"bytes,6,opt,name=in_progress_snapshot,json=inProgressSnapshot,proto3" json:"in_progress_snapshot,omitempty"`
WorkerAssignment *WorkerAssignment `protobuf:"bytes,5,opt,name=worker_assignment,json=workerAssignment,proto3" json:"worker_assignment,omitempty"`
InProgressSnapshotName string `protobuf:"bytes,6,opt,name=in_progress_snapshot_name,json=inProgressSnapshotName,proto3" json:"in_progress_snapshot_name,omitempty"`
// worker_selector is the per-actor placement constraint. The scheduler
// evaluates the AND of this selector and the template's workerSelector to
// find eligible pools. Set at CreateActor; may be updated at any time via
@@ -664,9 +665,10 @@ type Actor struct {
InProgressSnapshotSourceActorVersion int64 `protobuf:"varint,10,opt,name=in_progress_snapshot_source_actor_version,json=inProgressSnapshotSourceActorVersion,proto3" json:"in_progress_snapshot_source_actor_version,omitempty"`
// Volumes attached to the actor. These volumes only live as long as the actor.
// They are deleted when the actor is deleted.
ActorVolumes []*ExternalVolume `protobuf:"bytes,11,rep,name=actor_volumes,json=actorVolumes,proto3" json:"actor_volumes,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
ActorVolumes []*ExternalVolume `protobuf:"bytes,11,rep,name=actor_volumes,json=actorVolumes,proto3" json:"actor_volumes,omitempty"`
InProgressLocalSnapshotName string `protobuf:"bytes,12,opt,name=in_progress_local_snapshot_name,json=inProgressLocalSnapshotName,proto3" json:"in_progress_local_snapshot_name,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Actor) Reset() {
@@ -734,9 +736,9 @@ func (x *Actor) GetWorkerAssignment() *WorkerAssignment {
return nil
}
func (x *Actor) GetInProgressSnapshot() string {
func (x *Actor) GetInProgressSnapshotName() string {
if x != nil {
return x.InProgressSnapshot
return x.InProgressSnapshotName
}
return ""
}
@@ -776,6 +778,13 @@ func (x *Actor) GetActorVolumes() []*ExternalVolume {
return nil
}
func (x *Actor) GetInProgressLocalSnapshotName() string {
if x != nil {
return x.InProgressLocalSnapshotName
}
return ""
}
// WorkerAssignment identifies the Worker currently hosting an Actor.
type WorkerAssignment struct {
state protoimpl.MessageState `protogen:"open.v1"`
@@ -859,8 +868,7 @@ func (x *WorkerAssignment) GetWorkerPodIp() string {
}
// ActorSnapshot is an independently addressable durable Actor snapshot. Its
// contents are immutable. Its physical storage location is private to
// Substrate.
// contents are immutable.
type ActorSnapshot struct {
state protoimpl.MessageState `protogen:"open.v1"`
Metadata *ResourceMetadata `protobuf:"bytes,1,opt,name=metadata,proto3" json:"metadata,omitempty"`
@@ -871,6 +879,7 @@ type ActorSnapshot struct {
ActorTemplateName string `protobuf:"bytes,6,opt,name=actor_template_name,json=actorTemplateName,proto3" json:"actor_template_name,omitempty"`
ActorTemplateUid string `protobuf:"bytes,7,opt,name=actor_template_uid,json=actorTemplateUid,proto3" json:"actor_template_uid,omitempty"`
ContentScope SnapshotContentScope `protobuf:"varint,8,opt,name=content_scope,json=contentScope,proto3,enum=ateapi.SnapshotContentScope" json:"content_scope,omitempty"`
SnapshotUri string `protobuf:"bytes,9,opt,name=snapshot_uri,json=snapshotUri,proto3" json:"snapshot_uri,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -961,6 +970,13 @@ func (x *ActorSnapshot) GetContentScope() SnapshotContentScope {
return SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_UNSPECIFIED
}
func (x *ActorSnapshot) GetSnapshotUri() string {
if x != nil {
return x.SnapshotUri
}
return ""
}
// ActorSnapshotTag is an immutable, Atespace-owned alias and retention pin.
// Its owning Atespace cannot be deleted until the tag is removed.
type ActorSnapshotTag struct {
@@ -3076,9 +3092,9 @@ var File_ateapi_proto protoreflect.FileDescriptor
const file_ateapi_proto_rawDesc = "" +
"\n" +
"\fateapi.proto\x12\x06ateapi\x1a google/protobuf/field_mask.proto\x1a\x1fgoogle/protobuf/timestamp.proto\"\xc1\x01\n" +
"\x11LocalSnapshotInfo\x12'\n" +
"\x0fsnapshot_prefix\x18\x01 \x01(\tR\x0esnapshotPrefix\x12@\n" +
"\fateapi.proto\x12\x06ateapi\x1a google/protobuf/field_mask.proto\x1a\x1fgoogle/protobuf/timestamp.proto\"\xbd\x01\n" +
"\x11LocalSnapshotInfo\x12#\n" +
"\rsnapshot_name\x18\x01 \x01(\tR\fsnapshotName\x12@\n" +
"\x1dnode_vms_with_local_snapshots\x18\x02 \x03(\tR\x19nodeVmsWithLocalSnapshots\x12A\n" +
"\rcontent_scope\x18\x03 \x01(\x0e2\x1c.ateapi.SnapshotContentScopeR\fcontentScope\"\x90\x01\n" +
"\bSelector\x12D\n" +
@@ -3110,20 +3126,21 @@ const file_ateapi_proto_rawDesc = "" +
"\x12STATUS_UNSPECIFIED\x10\x00\x12\x12\n" +
"\x0eSTATUS_PENDING\x10\x01\x12\x12\n" +
"\x0eSTATUS_CREATED\x10\x02\x12\x13\n" +
"\x0fSTATUS_DELETING\x10\x03\"\xef\x06\n" +
"\x0fSTATUS_DELETING\x10\x03\"\xbe\a\n" +
"\x05Actor\x124\n" +
"\bmetadata\x18\x01 \x01(\v2\x18.ateapi.ResourceMetadataR\bmetadata\x128\n" +
"\x18actor_template_namespace\x18\x02 \x01(\tR\x16actorTemplateNamespace\x12.\n" +
"\x13actor_template_name\x18\x03 \x01(\tR\x11actorTemplateName\x12,\n" +
"\x06status\x18\x04 \x01(\x0e2\x14.ateapi.Actor.StatusR\x06status\x12E\n" +
"\x11worker_assignment\x18\x05 \x01(\v2\x18.ateapi.WorkerAssignmentR\x10workerAssignment\x120\n" +
"\x14in_progress_snapshot\x18\x06 \x01(\tR\x12inProgressSnapshot\x129\n" +
"\x11worker_assignment\x18\x05 \x01(\v2\x18.ateapi.WorkerAssignmentR\x10workerAssignment\x129\n" +
"\x19in_progress_snapshot_name\x18\x06 \x01(\tR\x16inProgressSnapshotName\x129\n" +
"\x0fworker_selector\x18\a \x01(\v2\x10.ateapi.SelectorR\x0eworkerSelector\x12:\n" +
"\x0flatest_snapshot\x18\b \x01(\v2\x11.ateapi.ObjectRefR\x0elatestSnapshot\x12I\n" +
"\x13local_snapshot_info\x18\t \x01(\v2\x19.ateapi.LocalSnapshotInfoR\x11localSnapshotInfo\x12W\n" +
")in_progress_snapshot_source_actor_version\x18\n" +
" \x01(\x03R$inProgressSnapshotSourceActorVersion\x12;\n" +
"\ractor_volumes\x18\v \x03(\v2\x16.ateapi.ExternalVolumeR\factorVolumes\"\xc6\x01\n" +
"\ractor_volumes\x18\v \x03(\v2\x16.ateapi.ExternalVolumeR\factorVolumes\x12D\n" +
"\x1fin_progress_local_snapshot_name\x18\f \x01(\tR\x1binProgressLocalSnapshotName\"\xc6\x01\n" +
"\x06Status\x12\x16\n" +
"\x12STATUS_UNSPECIFIED\x10\x00\x12\x13\n" +
"\x0fSTATUS_RESUMING\x10\x01\x12\x12\n" +
@@ -3141,7 +3158,7 @@ const file_ateapi_proto_rawDesc = "" +
"\n" +
"worker_pod\x18\x03 \x01(\tR\tworkerPod\x12$\n" +
"\x0eworker_pod_uid\x18\x04 \x01(\tR\fworkerPodUid\x12\"\n" +
"\rworker_pod_ip\x18\x05 \x01(\tR\vworkerPodIp\"\xb2\x03\n" +
"\rworker_pod_ip\x18\x05 \x01(\tR\vworkerPodIp\"\xd5\x03\n" +
"\rActorSnapshot\x124\n" +
"\bmetadata\x18\x01 \x01(\v2\x18.ateapi.ResourceMetadataR\bmetadata\x124\n" +
"\fsource_actor\x18\x02 \x01(\v2\x11.ateapi.ObjectRefR\vsourceActor\x12(\n" +
@@ -3150,7 +3167,8 @@ const file_ateapi_proto_rawDesc = "" +
"\x18actor_template_namespace\x18\x05 \x01(\tR\x16actorTemplateNamespace\x12.\n" +
"\x13actor_template_name\x18\x06 \x01(\tR\x11actorTemplateName\x12,\n" +
"\x12actor_template_uid\x18\a \x01(\tR\x10actorTemplateUid\x12A\n" +
"\rcontent_scope\x18\b \x01(\x0e2\x1c.ateapi.SnapshotContentScopeR\fcontentScope\"\xac\x01\n" +
"\rcontent_scope\x18\b \x01(\x0e2\x1c.ateapi.SnapshotContentScopeR\fcontentScope\x12!\n" +
"\fsnapshot_uri\x18\t \x01(\tR\vsnapshotUri\"\xac\x01\n" +
"\x10ActorSnapshotTag\x124\n" +
"\bmetadata\x18\x01 \x01(\v2\x18.ateapi.ResourceMetadataR\bmetadata\x12-\n" +
"\bsnapshot\x18\x02 \x01(\v2\x11.ateapi.ObjectRefR\bsnapshot\x123\n" +
pkg/proto/ateapipb/ateapi.proto
+7 -4
View File
@@ -81,7 +81,8 @@ service Control {
}
message LocalSnapshotInfo {
string snapshot_prefix = 1;
// The name of the local checkpoint on each of the nodes below.
string snapshot_name = 1;
// Node VMs that have local snapshots for this actor, while it's PAUSED.
repeated string node_vms_with_local_snapshots = 2;
// Scope the pause checkpoint captured (the template's onPause at pause
@@ -186,7 +187,7 @@ message Actor {
// Unset whenever the Actor has no worker (SUSPENDED, PAUSED, CRASHED).
WorkerAssignment worker_assignment = 5;
string in_progress_snapshot = 6;
string in_progress_snapshot_name = 6;
// worker_selector is the per-actor placement constraint. The scheduler
// evaluates the AND of this selector and the template's workerSelector to
@@ -206,6 +207,8 @@ message Actor {
// Volumes attached to the actor. These volumes only live as long as the actor.
// They are deleted when the actor is deleted.
repeated ExternalVolume actor_volumes = 11;
string in_progress_local_snapshot_name = 12;
}
// WorkerAssignment identifies the Worker currently hosting an Actor.
@@ -227,8 +230,7 @@ message WorkerAssignment {
}
// ActorSnapshot is an independently addressable durable Actor snapshot. Its
// contents are immutable. Its physical storage location is private to
// Substrate.
// contents are immutable.
message ActorSnapshot {
ResourceMetadata metadata = 1;
ObjectRef source_actor = 2;
@@ -238,6 +240,7 @@ message ActorSnapshot {
string actor_template_name = 6;
string actor_template_uid = 7;
SnapshotContentScope content_scope = 8;
string snapshot_uri = 9;
}
// ActorSnapshotTag is an immutable, Atespace-owned alias and retention pin.