Skip to content

fix: make atelet / ateom use actor UID for its internal bookkeeping #438

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Eric Bishop (ericdbishop) wants to merge 6 commits into
base: main
Choose a base branch
from
Open

fix: make atelet / ateom use actor UID for its internal bookkeeping #438

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b147613
proto gen
ericdbishop Jul 14, 2026
7e2d848
initial refactor
ericdbishop Jul 14, 2026
b7fa8c2
address adversarial review comments
ericdbishop Jul 14, 2026
baf756e
fmt
ericdbishop Jul 14, 2026
feec003
update tests
ericdbishop Jul 14, 2026
4990eea
observability docs update
ericdbishop Jul 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion cmd/ateapi/internal/controlapi/workflow_pause.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,8 @@ func (s *CallAteletPauseStep) Execute(ctx context.Context, input *PauseInput, st
SnapshotPrefix: state.Actor.InProgressSnapshot,
},
},
Scope: toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnPause),
Scope: toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnPause),
ActorUid: state.Actor.GetMetadata().Uid,
}

_, err = client.Checkpoint(ctx, req)
Expand Down
5 changes: 4 additions & 1 deletion cmd/ateapi/internal/controlapi/workflow_resume.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -276,6 +276,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
ActorTemplateNamespace: state.Actor.GetActorTemplateNamespace(),
ActorTemplateName: state.Actor.GetActorTemplateName(),
Spec: workloadSpec,
ActorUid: state.Actor.GetMetadata().Uid,
}
switch d := data.(type) {
case *ateapipb.SnapshotInfo_Local:
Expand Down Expand Up @@ -318,7 +319,8 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
SnapshotUriPrefix: snapshot,
},
},
Scope: toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnCommit),
Scope: toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnCommit),
ActorUid: state.Actor.GetMetadata().Uid,
}
_, err = client.Restore(ctx, req)
return maybeCrashActor(ctx, s.store, input.Atespace, input.ActorName, err, "while creating workload from golden snapshot")
Expand All @@ -341,6 +343,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
ActorTemplateName: state.Actor.GetActorTemplateName(),
SandboxAssets: sandboxAssets,
Spec: workloadSpec,
ActorUid: state.Actor.GetMetadata().Uid,
}
_, err = client.Run(ctx, req)
return maybeCrashActor(ctx, s.store, input.Atespace, input.ActorName, err, "while creating workload from spec")
Expand Down
3 changes: 2 additions & 1 deletion cmd/ateapi/internal/controlapi/workflow_suspend.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,8 @@ func (s *CallAteletSuspendStep) Execute(ctx context.Context, input *SuspendInput
SnapshotUriPrefix: state.Actor.GetInProgressSnapshot(),
},
},
Scope: toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnCommit),
Scope: toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnCommit),
ActorUid: state.Actor.GetMetadata().Uid,
}

_, err = client.Checkpoint(ctx, req)
Expand Down
67 changes: 37 additions & 30 deletions cmd/atelet/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -219,7 +219,7 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (*atele
return nil, status.Error(codes.InvalidArgument, err.Error())
}

atespace, actorName := req.GetAtespace(), req.GetActorName()
actorUID, atespace, actorName := req.GetActorUid(), req.GetAtespace(), req.GetActorName()

sandboxRec, err := recordFromRequest(req.GetSandboxAssets())
if err != nil {
Expand All @@ -230,18 +230,18 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (*atele
return nil, err
}

if err := resetActorDirs(atespace, actorName); err != nil {
if err := resetActorDirs(actorUID); err != nil {
return nil, fmt.Errorf("while resetting actor dirs: %w", err)
}

// Record the sandbox binaries this actor is running so a later Checkpoint
// (whose request no longer carries the sandbox config) can re-fetch the same
// version and pin it into the snapshot manifest.
if err := writeSandboxRecord(atespace, actorName, sandboxRec); err != nil {
if err := writeSandboxRecord(actorUID, sandboxRec); err != nil {
return nil, fmt.Errorf("while recording sandbox assets: %w", err)
}

if err := s.prepareOCIBundles(ctx, atespace, actorName,
if err := s.prepareOCIBundles(ctx, actorUID, actorName,
req.GetSpec(), req.GetTargetAteomUid(),
); err != nil {
return nil, err
Expand All @@ -262,6 +262,7 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (*atele
RunscPath: runscPathFor(assetPaths),
RuntimeAssetPaths: assetPaths,
Spec: buildAteomWorkloadSpec(req.GetSpec()),
ActorUid: actorUID,
}); err != nil {
return nil, fmt.Errorf("while calling ateom.RunWorkload: %w", err)
}
Expand Down Expand Up @@ -310,13 +311,13 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe
return nil, status.Error(codes.InvalidArgument, err.Error())
}

atespace, actorName := req.GetAtespace(), req.GetActorName()
actorUID, atespace, actorName := req.GetActorUid(), req.GetAtespace(), req.GetActorName()

// Checkpoint requests no longer carry the sandbox config; recover the
// version this actor was started with from the on-node record and re-fetch
// it (a cache hit) so ateom can drive runsc, and so we can pin it into the
// snapshot manifest below.
sandboxRec, err := readSandboxRecord(atespace, actorName)
sandboxRec, err := readSandboxRecord(actorUID)
if err != nil {
return nil, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidSandboxAsset, ateerrors.ReasonTerminalFileSystemError)
}
Expand All @@ -325,7 +326,7 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe
return nil, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidSandboxAsset, ateerrors.ReasonTerminalFileSystemError, ateerrors.ReasonFailedGetExternalObject, ateerrors.ReasonInvalidObjectURL)
}

checkpointDir := ateompath.CheckpointStateDir(atespace, actorName)
checkpointDir := ateompath.CheckpointStateDir(actorUID)

client, err := s.dialAteom(ctx, req.GetTargetAteomUid())
if err != nil {
Expand All @@ -344,6 +345,7 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe
RuntimeAssetPaths: assetPaths,
Spec: buildAteomWorkloadSpec(req.GetSpec()),
Scope: toAteomSnapshotScope(req.GetScope()),
ActorUid: actorUID,
})
if err != nil {
// TODO: Ateom should classify checkpoint failures, and set "should-crash"
Expand All @@ -370,7 +372,7 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe
}

// Note: we do not crash the actor if resetting the directory fails.
if err := resetActorDirs(atespace, actorName); err != nil {
if err := resetActorDirs(actorUID); err != nil {
return nil, fmt.Errorf("while resetting actor dirs: %w", err)
}

Expand All @@ -388,7 +390,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.GetAtespace(), req.GetActorName()), req.GetLocalConfig().GetSnapshotPrefix())
localCheckpointPath := filepath.Join(ateompath.LocalCheckpointsDir(req.GetActorUid()), req.GetLocalConfig().GetSnapshotPrefix())
if err := os.MkdirAll(localCheckpointPath, 0o700); err != nil {
return fmt.Errorf("while creating local checkpoint directory: %w", err)
}
Expand Down Expand Up @@ -454,15 +456,15 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
return nil, status.Error(codes.InvalidArgument, err.Error())
}

atespace, actorName := req.GetAtespace(), req.GetActorName()
actorUID, atespace, actorName := req.GetActorUid(), req.GetAtespace(), req.GetActorName()

// Not crashing the actor, because terminal errors here indicate problems with atelet,
// node or the disk itself.
if err := resetActorDirs(atespace, actorName); err != nil {
if err := resetActorDirs(actorUID); err != nil {
return nil, fmt.Errorf("while resetting actor dirs: %w", err)
}

checkpointDir := ateompath.RestoreStateDir(atespace, actorName)
checkpointDir := ateompath.RestoreStateDir(actorUID)

// Per-step timing so we can attribute resume latency between the rustfs
// download/decompress, the OCI image unpack, and ateom's own work. Logged at the end.
Expand All @@ -485,7 +487,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
return nil, ateerrors.CrashIfReason(ctx, fmt.Errorf("while unmarshalling sandbox record: %w", err), ateerrors.ReasonInvalidSandboxAsset)
}
case ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL:
localCheckpointDir := ateompath.LocalCheckpointsDir(atespace, actorName)
localCheckpointDir := ateompath.LocalCheckpointsDir(actorUID)
snapshotPrefix := req.GetLocalConfig().GetSnapshotPrefix()
manifest, err := os.ReadFile(filepath.Join(localCheckpointDir, snapshotPrefix, sandboxManifestName))
if err != nil {
Expand Down Expand Up @@ -518,7 +520,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonFailedGetExternalObject, ateerrors.ReasonInvalidObjectURL, ateerrors.ReasonTerminalFileSystemError)
}
case ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL:
if err := s.copyLocalCheckpoint(gctx, req.GetLocalConfig().GetSnapshotPrefix(), ateompath.LocalCheckpointsDir(atespace, actorName), checkpointDir, sandboxRec.SnapshotFiles); err != nil {
if err := s.copyLocalCheckpoint(gctx, req.GetLocalConfig().GetSnapshotPrefix(), ateompath.LocalCheckpointsDir(actorUID), checkpointDir, sandboxRec.SnapshotFiles); err != nil {
return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonTerminalFileSystemError)
}
}
Expand All @@ -531,7 +533,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonFailedGetExternalObject, ateerrors.ReasonInvalidObjectURL, ateerrors.ReasonTerminalFileSystemError, ateerrors.ReasonInvalidSandboxAsset)
}
t := time.Now()
if err := s.prepareOCIBundles(gctx, atespace, actorName, req.GetSpec(), req.GetTargetAteomUid()); err != nil {
if err := s.prepareOCIBundles(gctx, actorUID, actorName, req.GetSpec(), req.GetTargetAteomUid()); err != nil {
return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonTerminalFileSystemError)
}
dBundles = time.Since(t)
Expand All @@ -558,6 +560,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
RuntimeAssetPaths: assetPaths,
Spec: buildAteomWorkloadSpec(req.GetSpec()),
Scope: toAteomSnapshotScope(req.GetScope()),
ActorUid: req.GetActorUid(),
}); err != nil {
// TODO: classify the errors returned by Ateom and crash the actor if needed.
return nil, fmt.Errorf("while calling ateom.RestoreWorkload: %w", err)
Expand All @@ -566,7 +569,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)

// Record the (manifest-pinned) sandbox binaries on-node so a subsequent
// Checkpoint of this restored actor can re-pin the same version.
if err := writeSandboxRecord(atespace, actorName, sandboxRec); err != nil {
if err := writeSandboxRecord(req.ActorUid, sandboxRec); err != nil {

@HavenXia Haven Xia (HavenXia) Jul 16, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: can just use actorUID

// Note: crash the actor right away, if we cannot write the sandbox record now, we will not be able to checkpoint it later.
return nil, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonTerminalFileSystemError)
}
Expand Down Expand Up @@ -643,14 +646,15 @@ func (s *AteomHerder) downloadExternalCheckpoint(ctx context.Context, snapshotUr
// container and every application container in spec, in parallel.
func (s *AteomHerder) prepareOCIBundles(
ctx context.Context,
atespace, actorName string,
actorUID string,
actorName string,
spec *ateletpb.WorkloadSpec,
targetAteomUid string,
) error {
// Populate the per-actor nameentity directory that gets bind-mounted into
// the application containers. Regenerated on every resume, so it carries
// the correct per-actor name even when restoring from the golden snapshot.
identityDir := ateompath.ActorIdentityDirPath(atespace, actorName)
identityDir := ateompath.ActorIdentityDirPath(actorUID)
if err := os.MkdirAll(identityDir, 0o755); err != nil {
return fmt.Errorf("while creating actor nameentity dir: %w", err)
}
Expand All @@ -663,7 +667,7 @@ func (s *AteomHerder) prepareOCIBundles(
for _, vol := range spec.GetVolumes() {
if vol.GetType() == ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR {
ddVolumes[vol.GetName()] = true
volPath := ateompath.DurableDirVolumeMountPoint(atespace, actorName, vol.GetName())
volPath := ateompath.DurableDirVolumeMountPoint(actorUID, vol.GetName())
if err := os.MkdirAll(volPath, 0o700); err != nil {
return fmt.Errorf("while creating %q: %w", volPath, err)
}
Expand All @@ -684,14 +688,14 @@ func (s *AteomHerder) prepareOCIBundles(
if vol.GetType() == ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR {
annotations["dev.gvisor.spec.mount.durabledir.type"] = "bind"
annotations["dev.gvisor.spec.mount.durabledir.share"] = "container"
annotations["dev.gvisor.spec.mount.durabledir.source"] = ateompath.DurableDirVolumeMountPoint(atespace, actorName, vol.GetName())
annotations["dev.gvisor.spec.mount.durabledir.source"] = ateompath.DurableDirVolumeMountPoint(actorUID, vol.GetName())
}
}

if err := prepareOCIDirectory(
gCtx,
s.pullCache,
atespace, actorName,
actorUID,
"pause",
spec.GetPauseImage(),
[]string{"/pause"},
Expand Down Expand Up @@ -723,7 +727,7 @@ func (s *AteomHerder) prepareOCIBundles(
if err := prepareOCIDirectory(
gCtx,
s.pullCache,
atespace, actorName,
actorUID,
ctr.GetName(),
ctr.GetImage(),
ctr.GetCommand(),
Expand Down Expand Up @@ -838,6 +842,7 @@ func validateRunRequest(req *ateletpb.RunRequest) error {
var errs field.ErrorList
errs = append(errs, resources.ValidateResourceName(req.GetAtespace(), field.NewPath("atespace"))...)
errs = append(errs, resources.ValidateResourceName(req.GetActorName(), field.NewPath("actor_name"))...)
errs = append(errs, resources.ValidateResourceName(req.GetActorUid(), field.NewPath("actor_uid"))...)
for _, msg := range content.IsDNS1123Label(req.GetActorTemplateNamespace()) {
errs = append(errs, field.Invalid(field.NewPath("actor_template_namespace"), req.GetActorTemplateNamespace(), msg))
}
Expand All @@ -862,6 +867,7 @@ func validateCheckpointRequest(req *ateletpb.CheckpointRequest) error {
var errs field.ErrorList
errs = append(errs, resources.ValidateResourceName(req.GetAtespace(), field.NewPath("atespace"))...)
errs = append(errs, resources.ValidateResourceName(req.GetActorName(), field.NewPath("actor_name"))...)
errs = append(errs, resources.ValidateResourceName(req.GetActorUid(), field.NewPath("actor_uid"))...)
for _, msg := range content.IsDNS1123Label(req.GetActorTemplateNamespace()) {
errs = append(errs, field.Invalid(field.NewPath("actor_template_namespace"), req.GetActorTemplateNamespace(), msg))
}
Expand Down Expand Up @@ -906,6 +912,7 @@ func validateRestoreRequest(req *ateletpb.RestoreRequest) error {
var errs field.ErrorList
errs = append(errs, resources.ValidateResourceName(req.GetAtespace(), field.NewPath("atespace"))...)
errs = append(errs, resources.ValidateResourceName(req.GetActorName(), field.NewPath("actor_name"))...)
errs = append(errs, resources.ValidateResourceName(req.GetActorUid(), field.NewPath("actor_uid"))...)
for _, msg := range content.IsDNS1123Label(req.GetActorTemplateNamespace()) {
errs = append(errs, field.Invalid(field.NewPath("actor_template_namespace"), req.GetActorTemplateNamespace(), msg))
}
Expand Down Expand Up @@ -997,42 +1004,42 @@ func writeFileAtomic(path string, data []byte, perm os.FileMode) error {
return dir.Sync()
}

func resetActorDirs(atespace, actorName string) error {
func resetActorDirs(actorUID string) error {
// Explicitly leave runsc logs dir untouched.

bundleDir := ateompath.OCIBundleDir(atespace, actorName)
bundleDir := ateompath.OCIBundleDir(actorUID)
if err := os.RemoveAll(bundleDir); err != nil {
return wrapFileSystemErr("while deleting bundle dir: %w", err)
}
if err := os.MkdirAll(bundleDir, 0o700); err != nil {
return wrapFileSystemErr("while creating bundle dir: %w", err)
}

runscDir := ateompath.RunSCStateDir(atespace, actorName)
runscDir := ateompath.RunSCStateDir(actorUID)
if err := os.RemoveAll(runscDir); err != nil {
return wrapFileSystemErr("while deleting runsc state dir: %w", err)
}
if err := os.MkdirAll(runscDir, 0o700); err != nil {
return wrapFileSystemErr("while creating runsc state dir: %w", err)
}

pidFileDir := ateompath.PIDFileDir(atespace, actorName)
pidFileDir := ateompath.PIDFileDir(actorUID)
if err := os.RemoveAll(pidFileDir); err != nil {
return wrapFileSystemErr("while deleting PID file dir: %w", err)
}
if err := os.MkdirAll(pidFileDir, 0o700); err != nil {
return wrapFileSystemErr("while creating PID file dir: %w", err)
}

checkpointDir := ateompath.CheckpointStateDir(atespace, actorName)
checkpointDir := ateompath.CheckpointStateDir(actorUID)
if err := os.RemoveAll(checkpointDir); err != nil {
return wrapFileSystemErr("while deleting checkpoint-state dir: %w", err)
}
if err := os.MkdirAll(checkpointDir, 0o700); err != nil {
return wrapFileSystemErr("while creating checkpoint-state dir: %w", err)
}

restoreStateDir := ateompath.RestoreStateDir(atespace, actorName)
restoreStateDir := ateompath.RestoreStateDir(actorUID)
if err := os.RemoveAll(restoreStateDir); err != nil {
return wrapFileSystemErr("while deleting restore-state dir: %w", err)
}
Expand All @@ -1042,15 +1049,15 @@ func resetActorDirs(atespace, actorName string) error {

// World-readable (0o755): bind-mounted into the actor, whose workload
// reads it through the gofer.
identityDir := ateompath.ActorIdentityDirPath(atespace, actorName)
identityDir := ateompath.ActorIdentityDirPath(actorUID)
if err := os.RemoveAll(identityDir); err != nil {
return wrapFileSystemErr("while deleting actor nameentity dir: %w", err)
}
if err := os.MkdirAll(identityDir, 0o755); err != nil {
return wrapFileSystemErr("while creating actor nameentity dir: %w", err)
}

durableDirVolumesMountDir := ateompath.DurableDirVolumeMountsDir(atespace, actorName)
durableDirVolumesMountDir := ateompath.DurableDirVolumeMountsDir(actorUID)
if err := os.RemoveAll(durableDirVolumesMountDir); err != nil {
return wrapFileSystemErr("while deleting durable-dir volumes mount dir: %w", err)
}
Expand Down
Loading
Loading