From ac525b2ff2ce915170c9815d3189093b68e86670 Mon Sep 17 00:00:00 2001 From: Sairaj Pokale Date: Tue, 15 Sep 2026 18:51:24 +0000 Subject: [PATCH 1/3] ateapi,atelet: reclaim actor node state when a worker's pod is gone A deleted worker pod is deregistered but its actors are never terminated, so actors/ is orphaned permanently (measured: 55.22 GB over two runs, nothing reclaimed by scale-to-zero, --delete-all, or a reinstall). The worker-delete workflow now sends atelet's existing Terminate before releasing each bound actor: it is the last moment anything knows which node holds the state. It dials by node, since the pod it would dial through is what just disappeared, and is best-effort so an unreachable node cannot wedge deregistration. Terminate failed early for the same reason, tearing the workload down before reaching the directories. A gone ateom (missing socket, no sandbox record, UNAVAILABLE) now skips that step and still reclaims. A reachable ateom that rejects the call still fails: that sandbox is live. Actors holding a local pause snapshot are skipped -- Terminate prunes local checkpoints, and that deletion is unrecoverable. --- cmd/ateapi/internal/controlapi/service.go | 9 +- cmd/ateapi/internal/controlapi/worker_test.go | 2 +- cmd/ateapi/internal/controlapi/workflow.go | 19 +- .../internal/controlapi/workflow_delete.go | 44 ++-- .../controlapi/workflow_delete_test.go | 2 +- .../controlapi/workflow_worker_delete.go | 90 ++++++++ .../controlapi/workflow_worker_delete_test.go | 210 +++++++++++++++++- cmd/atelet/lifecycle_test.go | 101 +++++++++ cmd/atelet/main.go | 112 +++++++--- 9 files changed, 531 insertions(+), 58 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/service.go b/cmd/ateapi/internal/controlapi/service.go index b433515586..a6434183e4 100644 --- a/cmd/ateapi/internal/controlapi/service.go +++ b/cmd/ateapi/internal/controlapi/service.go @@ -102,7 +102,14 @@ func NewRPCService( actorIDCAPool: actorIDCAPool, } s.actorWorkflow = NewActorWorkflow(impl, workerCache, dialer, sandboxConfigLister, storageClassLister, instruments, egressGatewayAddress, s, objectStore) - s.workerWorkflow = NewWorkerWorkflow(impl) + // Converted through an explicit nil check: a nil *AteletDialer assigned + // straight to the interface would be a non-nil interface holding a nil + // pointer, and the worker workflow's nil check would not see it. + var nodeDialer ateletNodeDialer + if dialer != nil { + nodeDialer = dialer + } + s.workerWorkflow = NewWorkerWorkflow(impl, nodeDialer) return s } diff --git a/cmd/ateapi/internal/controlapi/worker_test.go b/cmd/ateapi/internal/controlapi/worker_test.go index 6adff8403d..56062d6f9e 100644 --- a/cmd/ateapi/internal/controlapi/worker_test.go +++ b/cmd/ateapi/internal/controlapi/worker_test.go @@ -96,7 +96,7 @@ func newWorkerAPIService(t *testing.T) (*RPCService, store.Interface) { persistence, cleanup := storetest.SetupTestStore(t) t.Cleanup(cleanup) impl := newServiceImpl(persistence, nil) - return &RPCService{impl: impl, workerWorkflow: NewWorkerWorkflow(persistence)}, persistence + return &RPCService{impl: impl, workerWorkflow: NewWorkerWorkflow(persistence, nil)}, persistence } // seedAPIWorker registers a worker directly through the store and returns it as diff --git a/cmd/ateapi/internal/controlapi/workflow.go b/cmd/ateapi/internal/controlapi/workflow.go index 1057169ba3..b1c097f3b5 100644 --- a/cmd/ateapi/internal/controlapi/workflow.go +++ b/cmd/ateapi/internal/controlapi/workflow.go @@ -32,6 +32,7 @@ import ( "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/trace" + "google.golang.org/grpc" grpcCodes "google.golang.org/grpc/codes" "google.golang.org/grpc/status" storagev1listers "k8s.io/client-go/listers/storage/v1" @@ -170,12 +171,22 @@ type actorWorkflowStore interface { // does from the other side: releasing the Actor bound to a Worker stays // in-process because there is no bind/release RPC. type WorkerWorkflow struct { - store workerWorkflowStore + store workerWorkflowStore + dialer ateletNodeDialer } -// NewWorkerWorkflow creates a new WorkerWorkflow. -func NewWorkerWorkflow(store workerWorkflowStore) *WorkerWorkflow { - return &WorkerWorkflow{store: store} +// ateletNodeDialer reaches the atelet on a named node. Narrower than the +// *AteletDialer the actor workflows hold, because a Worker being deleted has +// lost its pod: the by-pod lookup those use cannot resolve, and the node is +// the only handle left on the atelet still holding the actor's state. +type ateletNodeDialer interface { + DialForAteletOnNode(nodeName string) (*grpc.ClientConn, error) +} + +// NewWorkerWorkflow creates a new WorkerWorkflow. A nil dialer disables the +// node-state reclaim step (tests that assert only on store transitions). +func NewWorkerWorkflow(store workerWorkflowStore, dialer ateletNodeDialer) *WorkerWorkflow { + return &WorkerWorkflow{store: store, dialer: dialer} } // workerWorkflowStore enumerates the exact storage methods needed by diff --git a/cmd/ateapi/internal/controlapi/workflow_delete.go b/cmd/ateapi/internal/controlapi/workflow_delete.go index 7138c04973..9d445a6d0b 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete.go @@ -176,23 +176,7 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res slog.String("actor", actorRef.Name), slog.String("templateAtespace", actor.GetActorTemplate().GetAtespace()), slog.String("templateName", actor.GetActorTemplate().GetName())) - workloadSpec = &ateletpb.WorkloadSpec{} - for _, vol := range actor.GetStatus().GetActorVolumes() { - // StorageVolumeId is only populated once the volume is provisioned. - // Skip volumes that were never created (e.g. failed during PENDING state). - if vol.GetStorageVolumeId() != "" { - workloadSpec.Volumes = append(workloadSpec.Volumes, &ateletpb.Volume{ - Name: vol.GetVolumeName(), - Source: &ateletpb.Volume_External{ - External: &ateletpb.ExternalVolumeSource{ - StorageVolumeId: vol.GetStorageVolumeId(), - VolumeType: vol.GetVolumeType(), - VolumeContext: vol.GetVolumeContext(), - }, - }, - }) - } - } + workloadSpec = fallbackWorkloadSpec(actor) } req := &ateletpb.TerminateRequest{ @@ -216,6 +200,32 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res return nil } +// fallbackWorkloadSpec is the spec to terminate an actor with when its template +// cannot be resolved — deleted, or never readable from the caller in the first +// place. Only the external volumes recorded on the actor are carried, because +// they are the one part of the spec a teardown still acts on: atelet unmounts +// them from the node. A volume with no StorageVolumeId was never provisioned +// (e.g. the actor failed while PENDING), so there is nothing mounted to name. +func fallbackWorkloadSpec(actor *ateapipb.Actor) *ateletpb.WorkloadSpec { + spec := &ateletpb.WorkloadSpec{} + for _, vol := range actor.GetStatus().GetActorVolumes() { + if vol.GetStorageVolumeId() == "" { + continue + } + spec.Volumes = append(spec.Volumes, &ateletpb.Volume{ + Name: vol.GetVolumeName(), + Source: &ateletpb.Volume_External{ + External: &ateletpb.ExternalVolumeSource{ + StorageVolumeId: vol.GetStorageVolumeId(), + VolumeType: vol.GetVolumeType(), + VolumeContext: vol.GetVolumeContext(), + }, + }, + }) + } + return spec +} + // ensureVolumesDetachedForDelete detaches external volumes. func (w *ActorWorkflow) ensureVolumesDetachedForDelete(ctx context.Context, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (err error) { ctx, done := stepSpan(ctx, "DetachVolumesForDelete") diff --git a/cmd/ateapi/internal/controlapi/workflow_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_delete_test.go index fa95d62fea..7e6856da9c 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete_test.go @@ -443,7 +443,7 @@ func TestDeleteActor_CollectsSnapshotsAfterWorkerDelete(t *testing.T) { // The worker's pod goes away with the commit still outstanding, so // the suspend never gets to finish. - if _, err := NewWorkerWorkflow(persistence).DeleteWorker(ctx, workerName, store.DeletePreconditions{}); err != nil { + if _, err := NewWorkerWorkflow(persistence, nil).DeleteWorker(ctx, workerName, store.DeletePreconditions{}); err != nil { t.Fatalf("DeleteWorker: %v", err) } diff --git a/cmd/ateapi/internal/controlapi/workflow_worker_delete.go b/cmd/ateapi/internal/controlapi/workflow_worker_delete.go index 677d85cfbc..ea8614d8a0 100644 --- a/cmd/ateapi/internal/controlapi/workflow_worker_delete.go +++ b/cmd/ateapi/internal/controlapi/workflow_worker_delete.go @@ -22,6 +22,7 @@ import ( "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/internal/ateattr" + "github.com/agent-substrate/substrate/internal/proto/ateletpb" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" @@ -187,6 +188,14 @@ func (w *WorkerWorkflow) releaseBoundActor(ctx context.Context, worker *ateapipb markSkipped(ctx, "actor no longer points at this worker") return nil } + // Reclaim before the state checks below: the disk the actor left on the + // node has to go whether it suspended cleanly or crashed, and this is the + // last moment anything still knows which node that is. Once the record is + // released, the actor names no worker, no worker names a node, and nothing + // ever revisits the actor's UID — the directories are orphaned for the life + // of the node. + w.reclaimActorStateOnNode(ctx, worker, actor) + // If the actor is suspended, it's already been released. if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_SUSPENDED { markSkipped(ctx, "actor suspended cleanly before the pod went away") @@ -238,6 +247,87 @@ func (w *WorkerWorkflow) releaseBoundActor(ctx context.Context, worker *ateapipb return nil } +// reclaimActorStateOnNode asks the atelet on the worker's node to terminate the +// actor, which is what reclaims the actor's state directory +// (/var/lib/ateom-gvisor/actors/: durable dir, checkpoint and restore +// images — gigabytes for a durdir actor) and unmounts its external volumes. +// +// The worker's pod is gone by the time this runs, so the atelet is reached by +// node: the by-pod lookup the actor workflows use resolves through the pod that +// has just disappeared, and returning "pod not found" there is exactly how this +// state came to be orphaned. atelet tolerates the ateom being gone and reclaims +// the directories anyway. +// +// Best-effort by construction: a node that cannot be reached must not wedge the +// deregistration of its workers, and the reclaim has no record of its own to +// retry from once the worker is deleted. What it misses — an unreachable atelet, +// a node that never comes back, an abrupt eviction — is the orphan sweep's to +// collect. +func (w *WorkerWorkflow) reclaimActorStateOnNode(ctx context.Context, worker *ateapipb.Worker, actor *ateapipb.Actor) { + ctx, done := stepSpan(ctx, "ReclaimActorStateOnNode") + defer func() { _ = done(nil) }() + + if w.dialer == nil { + markSkipped(ctx, "no atelet dialer configured") + return + } + nodeName := worker.GetNodeName() + if nodeName == "" { + // Pre-dates the field, or a Worker registered before its pod was + // scheduled. Nothing names the node holding the state. + markSkipped(ctx, "worker records no node") + return + } + // A local snapshot is state this node holds deliberately, and Terminate + // prunes local checkpoints. Leaving it is the conservative choice: a + // wrongly-kept snapshot costs disk the sweep can still reclaim later, a + // wrongly-deleted one cannot be recovered at all. + if actor.GetStatus().GetLocalSnapshotInfo() != nil { + markSkipped(ctx, "actor has a local snapshot pinned to the node") + return + } + + actorRef := resources.ActorRefFromActor(actor) + logAttrs := []any{ + slog.Any("actor", actorRef), + slog.String("actor_uid", actor.GetMetadata().GetUid()), + slog.String("worker", worker.GetMetadata().GetName()), + slog.String("node", nodeName), + } + + conn, err := w.dialer.DialForAteletOnNode(nodeName) + if err != nil { + // Includes ErrNoAteletOnNode: the atelet is restarting, or the node + // itself is gone. Nothing to reclaim against right now. + slog.WarnContext(ctx, "Could not reach the atelet holding a released actor's state; leaving it for the orphan sweep", + append(logAttrs, slog.Any("err", err))...) + return + } + + slog.InfoContext(ctx, "Reclaiming the node state of an actor released from a worker whose pod is gone", logAttrs...) + // The template is not resolvable from this workflow — and would be the + // wrong thing to block on if it were, since a delete can outlive it. The + // fallback spec carries what the teardown acts on: the external volumes to + // unmount. + _, err = ateletpb.NewAteomHerderClient(conn).Terminate(ctx, &ateletpb.TerminateRequest{ + TargetAteomUid: worker.GetWorkerPodUid(), + Atespace: actor.GetMetadata().GetAtespace(), + ActorName: actor.GetMetadata().GetName(), + ActorUid: actor.GetMetadata().GetUid(), + ActorTemplateAtespace: actor.GetActorTemplate().GetAtespace(), + ActorTemplateName: actor.GetActorTemplate().GetName(), + Spec: fallbackWorkloadSpec(actor), + }) + switch { + case err == nil: + case status.Code(err) == codes.NotFound: + slog.InfoContext(ctx, "Actor already terminated on its node", logAttrs...) + default: + slog.WarnContext(ctx, "Failed to reclaim a released actor's node state; leaving it for the orphan sweep", + append(logAttrs, slog.Any("err", err))...) + } +} + // finalizeDeleted removes the worker from the store and returns the deleted // record. The request's guards are carried down as delete preconditions, so a // worker that moved on since the caller read it is reported as a conflict rather diff --git a/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go index 1cfecf6ae0..88c9dd9a96 100644 --- a/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_worker_delete_test.go @@ -17,17 +17,27 @@ package controlapi import ( "context" "errors" + "net" + "slices" "strings" + "sync" "testing" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest" "github.com/agent-substrate/substrate/internal/ateattr" + "github.com/agent-substrate/substrate/internal/proto/ateletpb" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" + "google.golang.org/grpc/test/bufconn" + "google.golang.org/protobuf/proto" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) // newWorkerDeleteWorkflow returns a workflow backed by a real store, which is @@ -37,7 +47,7 @@ func newWorkerDeleteWorkflow(t *testing.T) (*WorkerWorkflow, store.Interface) { t.Helper() persistence, cleanup := storetest.SetupTestStore(t) t.Cleanup(cleanup) - return NewWorkerWorkflow(persistence), persistence + return NewWorkerWorkflow(persistence, nil), persistence } // apiActorRef names the Actor seedAPIActor stores. @@ -87,7 +97,7 @@ func TestDeleteWorkerWorkflow_DrainsBeforeSweeping(t *testing.T) { actor := seedAPIActor(t, ctx, persistence, ateapipb.ActorState_ACTOR_STATE_RUNNING) assignAPIWorker(t, ctx, persistence, apiWorkerName, actor.GetMetadata().GetUid()) - wf := NewWorkerWorkflow(failingUpdateActorStore{Interface: persistence, err: errors.New("release failed")}) + wf := NewWorkerWorkflow(failingUpdateActorStore{Interface: persistence, err: errors.New("release failed")}, nil) if _, err := wf.DeleteWorker(ctx, apiWorkerName, store.DeletePreconditions{}); err == nil { t.Fatal("DeleteWorker() = nil error, want the release failure reported") } @@ -307,7 +317,7 @@ func TestDeleteWorkerWorkflow_FailedReleaseKeepsWorker(t *testing.T) { actor := seedAPIActor(t, ctx, persistence, ateapipb.ActorState_ACTOR_STATE_RUNNING) assignAPIWorker(t, ctx, persistence, apiWorkerName, actor.GetMetadata().GetUid()) - wf := NewWorkerWorkflow(failingUpdateActorStore{Interface: persistence, err: tc.updateErr}) + wf := NewWorkerWorkflow(failingUpdateActorStore{Interface: persistence, err: tc.updateErr}, nil) _, err := wf.DeleteWorker(ctx, apiWorkerName, store.DeletePreconditions{}) if err == nil { t.Fatal("DeleteWorker() = nil error, want the release failure reported") @@ -340,7 +350,7 @@ func TestDeleteWorkerWorkflow_ActorDeletedDuringRelease(t *testing.T) { actor := seedAPIActor(t, ctx, persistence, ateapipb.ActorState_ACTOR_STATE_RUNNING) assignAPIWorker(t, ctx, persistence, apiWorkerName, actor.GetMetadata().GetUid()) - wf := NewWorkerWorkflow(failingUpdateActorStore{Interface: persistence, err: store.ErrNotFound}) + wf := NewWorkerWorkflow(failingUpdateActorStore{Interface: persistence, err: store.ErrNotFound}, nil) if _, err := wf.DeleteWorker(ctx, apiWorkerName, store.DeletePreconditions{}); err != nil { t.Fatalf("DeleteWorker() failed: %v", err) } @@ -376,3 +386,195 @@ type failingUpdateActorStore struct { func (f failingUpdateActorStore) UpdateActor(context.Context, resources.ActorRef, store.Precondition, func(*ateapipb.Actor) error) (*ateapipb.Actor, error) { return nil, f.err } + +// A Worker is deleted because its pod is gone, and the pod took the ateom with +// it but not the actor's state directory on the node — a durdir actor leaves +// gigabytes there. This delete is the last moment anything knows which node +// that is, so it is where the reclaim has to happen. +func TestDeleteWorkerWorkflow_ReclaimsActorNodeState(t *testing.T) { + tests := []struct { + name string + // state the actor is in when its pod disappears. + state ateapipb.ActorState + // seed further shapes the stored actor. + seed func(*ateapipb.Actor) + // terminateErr is what the atelet answers, if anything. + terminateErr error + wantTerminate bool + }{ + { + name: "running actor is reclaimed", + state: ateapipb.ActorState_ACTOR_STATE_RUNNING, + wantTerminate: true, + }, + { + // It saved its state externally and stays resumable, but it is + // resumable somewhere else: what it left on this node is dead + // weight, and the release path skips it for every other purpose. + name: "cleanly suspended actor is reclaimed too", + state: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, + wantTerminate: true, + }, + { + // Terminate prunes local checkpoints, and a local snapshot is the + // one piece of actor state this node holds deliberately. Leaving + // it costs disk a sweep can still reclaim; deleting it is + // unrecoverable. + name: "actor with a local snapshot is left alone", + state: ateapipb.ActorState_ACTOR_STATE_PAUSED, + seed: func(a *ateapipb.Actor) { + a.Status.LocalSnapshotInfo = &ateapipb.LocalSnapshotInfo{ + SnapshotName: "pause-1", + NodeVmsWithLocalSnapshots: []string{"node-1"}, + } + }, + wantTerminate: false, + }, + { + // An unreachable or unhappy node must not wedge deregistration: + // the worker still goes, and the orphan sweep collects what this + // could not. + name: "a failing terminate does not fail the delete", + state: ateapipb.ActorState_ACTOR_STATE_RUNNING, + terminateErr: status.Error(codes.Internal, "atelet is having a bad day"), + wantTerminate: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + persistence, cleanup := storetest.SetupTestStore(t) + t.Cleanup(cleanup) + + atelet := &capturingTerminator{err: tt.terminateErr} + wf := NewWorkerWorkflow(persistence, newNodeAteletDialer(t, atelet)) + + seedAPIWorker(t, ctx, persistence, validWorker(apiWorkerName)) + seeds := []func(*ateapipb.Actor){} + if tt.seed != nil { + seeds = append(seeds, tt.seed) + } + actor := seedAPIActor(t, ctx, persistence, tt.state, seeds...) + assignAPIWorker(t, ctx, persistence, apiWorkerName, actor.GetMetadata().GetUid()) + + if _, err := wf.DeleteWorker(ctx, apiWorkerName, store.DeletePreconditions{}); err != nil { + t.Fatalf("DeleteWorker() failed: %v", err) + } + + got := atelet.requests() + if !tt.wantTerminate { + if len(got) != 0 { + t.Fatalf("atelet received %d Terminate calls, want none: %v", len(got), got) + } + return + } + if len(got) != 1 { + t.Fatalf("atelet received %d Terminate calls, want exactly 1", len(got)) + } + req := got[0] + // The actor the node has to be told about, and the ateom UID it + // was hosted by — which is what atelet resolves the (now absent) + // sandbox through. + if req.GetActorUid() != actor.GetMetadata().GetUid() { + t.Errorf("Terminate actor_uid = %q, want %q", req.GetActorUid(), actor.GetMetadata().GetUid()) + } + if req.GetAtespace() != apiActorRef.Atespace || req.GetActorName() != apiActorRef.Name { + t.Errorf("Terminate named %s/%s, want %s", req.GetAtespace(), req.GetActorName(), apiActorRef) + } + if want := validWorker(apiWorkerName).GetWorkerPodUid(); req.GetTargetAteomUid() != want { + t.Errorf("Terminate target_ateom_uid = %q, want the worker's pod uid %q", req.GetTargetAteomUid(), want) + } + }) + } +} + +// A Worker whose record names no node cannot be reclaimed against: the node is +// the only handle left on the atelet once the pod is gone. The delete carries +// on regardless. +func TestDeleteWorkerWorkflow_ReclaimSkippedWithoutANode(t *testing.T) { + ctx := context.Background() + persistence, cleanup := storetest.SetupTestStore(t) + t.Cleanup(cleanup) + + atelet := &capturingTerminator{} + wf := NewWorkerWorkflow(persistence, newNodeAteletDialer(t, atelet)) + + seedAPIWorker(t, ctx, persistence, validWorker(apiWorkerName, func(w *ateapipb.Worker) { w.NodeName = "" })) + actor := seedAPIActor(t, ctx, persistence, ateapipb.ActorState_ACTOR_STATE_RUNNING) + assignAPIWorker(t, ctx, persistence, apiWorkerName, actor.GetMetadata().GetUid()) + + if _, err := wf.DeleteWorker(ctx, apiWorkerName, store.DeletePreconditions{}); err != nil { + t.Fatalf("DeleteWorker() failed: %v", err) + } + if got := atelet.requests(); len(got) != 0 { + t.Errorf("atelet received %d Terminate calls, want none: %v", len(got), got) + } +} + +// capturingTerminator is an atelet that records the Terminate calls it is sent +// and answers them with err. +type capturingTerminator struct { + ateletpb.UnimplementedAteomHerderServer + err error + + mu sync.Mutex + reqs []*ateletpb.TerminateRequest +} + +func (f *capturingTerminator) Terminate(_ context.Context, req *ateletpb.TerminateRequest) (*ateletpb.TerminateResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.reqs = append(f.reqs, proto.Clone(req).(*ateletpb.TerminateRequest)) + if f.err != nil { + return nil, f.err + } + return &ateletpb.TerminateResponse{}, nil +} + +func (f *capturingTerminator) requests() []*ateletpb.TerminateRequest { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.reqs) +} + +// newNodeAteletDialer resolves node-1's atelet to an in-process fake. The conn +// cache is pre-warmed by the atelet pod's UID, so the by-node lookup under test +// runs for real and only the transport is short-circuited. No worker pod is +// seeded: this is the state the reclaim exists for, where the pod is gone. +func newNodeAteletDialer(t *testing.T, srvImpl ateletpb.AteomHerderServer) *AteletDialer { + t.Helper() + + srv := grpc.NewServer() + ateletpb.RegisterAteomHerderServer(srv, srvImpl) + lis := bufconn.Listen(1 << 20) + go func() { + if err := srv.Serve(lis); err != nil { + t.Logf("fake atelet server exited: %v", err) + } + }() + conn, err := grpc.NewClient("passthrough://bufnet", + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { + return lis.DialContext(ctx) + })) + if err != nil { + t.Fatalf("connecting to the fake atelet: %v", err) + } + t.Cleanup(func() { + conn.Close() + srv.Stop() + }) + + goneWorkerPod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Namespace: "ate-system", Name: "worker-pod-gone", UID: "worker-pod-gone"}, + Spec: corev1.PodSpec{NodeName: "node-1"}, + } + ateletPod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Namespace: ateletNamespace, Name: "atelet-1", UID: "atelet-uid"}, + Spec: corev1.PodSpec{NodeName: "node-1"}, + } + dialer := newDialerForPods(t, goneWorkerPod, ateletPod) + dialer.ateletConns.Add("atelet-uid", conn) + return dialer +} diff --git a/cmd/atelet/lifecycle_test.go b/cmd/atelet/lifecycle_test.go index ed27fa9eb0..52e5c3ef80 100644 --- a/cmd/atelet/lifecycle_test.go +++ b/cmd/atelet/lifecycle_test.go @@ -250,3 +250,104 @@ func TestLocalSnapshotGC(t *testing.T) { t.Errorf("reading actor dir %s: %v", actorDir, err) } } + +// A deleted worker pod takes its ateom with it and leaves the actor's state on +// the node — gigabytes of it for a durdir actor. Terminate is the only path +// that reclaims those directories, and nothing ever revisits a terminated +// actor's UID, so a Terminate that refuses to run without a live ateom orphans +// them for the life of the node. Both ways the sandbox can be absent are +// covered: the pod's socket gone with it, and an actor that never started one. +func TestTerminateReclaimsStateWithoutALiveSandbox(t *testing.T) { + const ( + atespace = "ate-demo" + actorName = "counter" + actorUID = "actor-uid-1" + ateomUID = "ateom-uid-1" + ) + + tests := []struct { + name string + // setup arranges the absence under test and reports it. + setup func(t *testing.T) + }{ + { + name: "ateom socket gone with its pod", + setup: func(t *testing.T) { + orig := ateomSocketPath + missing := filepath.Join(t.TempDir(), "ateom.sock") + ateomSocketPath = func(string) string { return missing } + t.Cleanup(func() { ateomSocketPath = orig }) + }, + }, + { + // The record is written at Run/Restore: without one, no sandbox + // was ever started for this actor here, so there is nothing to + // tear down and no runsc to tear it down with. + name: "ateom reachable but the actor has no sandbox record", + setup: func(t *testing.T) { + serveFakeAteom(t, &fakeAteom{}) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + useTempNodeDirs(t) + ctx := t.Context() + tt.setup(t) + + // The three directories that each hold a full copy of a durdir + // actor's payload, which is what makes one actor cost 3 GiB. + stateDirs := []string{ + ateompath.DurableDirVolumeMountsDir(actorUID), + ateompath.CheckpointStateDir(actorUID), + ateompath.RestoreStateDir(actorUID), + } + for _, dir := range stateDirs { + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatalf("seeding %s: %v", dir, err) + } + if err := os.WriteFile(filepath.Join(dir, "payload"), []byte("actor state"), 0o600); err != nil { + t.Fatalf("seeding %s: %v", dir, err) + } + } + snapshotDir := ateompath.LocalSnapshotDir(actorUID, "pause-snap-1") + if err := os.MkdirAll(snapshotDir, 0o755); err != nil { + t.Fatalf("seeding %s: %v", snapshotDir, err) + } + + s := &AteomHerder{ + ateomDialer: newAteomDialer(1), + systemInfoVolumes: newSystemInfoVolumeRefresher(nil, nil), + } + if _, err := s.Terminate(ctx, &ateletpb.TerminateRequest{ + Atespace: atespace, + ActorName: actorName, + ActorUid: actorUID, + ActorTemplateAtespace: "default", + ActorTemplateName: "counter", + TargetAteomUid: ateomUID, + Spec: &ateletpb.WorkloadSpec{}, + }); err != nil { + t.Fatalf("Terminate: %v", err) + } + + // #1654 made Terminate remove the actor's directory outright + // rather than emptying the state dirs in place, so the seeded + // payloads go with it and the whole tree is the assertion. + if actorDir := ateompath.ActorPath(actorUID); !isGone(actorDir) { + entries, _ := os.ReadDir(actorDir) + t.Errorf("%s survived terminate with %d entries, want the actor's state reclaimed", actorDir, len(entries)) + } + if localDir := ateompath.LocalCheckpointsDir(actorUID); !isGone(localDir) { + t.Errorf("%s survived terminate, want the local snapshots pruned", localDir) + } + }) + } +} + +// isGone reports whether path does not exist. +func isGone(path string) bool { + _, err := os.Stat(path) + return os.IsNotExist(err) +} diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 6155e49aa0..6bc7679a4a 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -1276,40 +1276,15 @@ func (s *AteomHerder) Terminate(ctx context.Context, req *ateletpb.TerminateRequ actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} actorUID := req.GetActorUid() - var assetPaths map[string]string - sandboxRec, err := readSandboxRecord(actorUID) - if err != nil { - return nil, fmt.Errorf("failed to read sandbox record during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) - } - paths, err := s.ensureSandboxAssets(ctx, sandboxRec) - if err != nil { - return nil, fmt.Errorf("failed to ensure sandbox assets during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) - } - assetPaths = paths - - client, err := s.dialAteom(ctx, req.GetTargetAteomUid()) - if err != nil { - return nil, fmt.Errorf("failed to dial ateom for terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) - } - + // Built here rather than next to the RPC it feeds, so a malformed spec is + // still rejected as INVALID_ARGUMENT on the paths that skip that RPC. spec, err := buildAteomWorkloadSpec(req.GetSpec()) if err != nil { return nil, status.Errorf(codes.InvalidArgument, "invalid workload spec: %v", err) } - if _, err := client.TerminateWorkload(ctx, &ateompb.TerminateWorkloadRequest{ - Atespace: req.GetAtespace(), - ActorName: req.GetActorName(), - ActorUid: req.GetActorUid(), - ActorTemplateAtespace: req.GetActorTemplateAtespace(), - ActorTemplateName: req.GetActorTemplateName(), - RunscPath: runscPathFor(assetPaths), - Spec: spec, - }); err != nil { - if status.Code(err) == codes.NotFound { - slog.InfoContext(ctx, "workload not found on ateom during terminate", slog.Any("actor", actorRef), slog.String("actorUID", actorUID)) - } else { - return nil, fmt.Errorf("failed calling ateom.TerminateWorkload (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) - } + + if err := s.terminateWorkloadOnAteom(ctx, req, spec, actorRef, actorUID); err != nil { + return nil, err } // Deregister after teardown succeeds @@ -1337,6 +1312,83 @@ func (s *AteomHerder) Terminate(ctx context.Context, req *ateletpb.TerminateRequ return &ateletpb.TerminateResponse{}, nil } +// terminateWorkloadOnAteom tears the workload down on the ateom hosting it. +// +// A Terminate can legitimately arrive after that ateom is gone: the worker pod +// deleted, evicted, or its node drained, with the actor's state left on the +// node. The teardown is then impossible, but the on-disk state it fronts is +// precisely what the rest of Terminate exists to reclaim — so a missing ateom +// is reported and skipped rather than failing the RPC and stranding tens of GB +// of actor state forever (nothing else ever revisits a terminated actor UID). +// +// An ateom that is present but rejects the call still fails: that is a live +// sandbox this could not tear down, and reclaiming its directories underneath +// it would be worse than leaving them. +func (s *AteomHerder) terminateWorkloadOnAteom(ctx context.Context, req *ateletpb.TerminateRequest, spec *ateompb.WorkloadSpec, actorRef resources.ActorRef, actorUID string) error { + ateomUID := req.GetTargetAteomUid() + // The socket is the ateom's liveness: it lives under the pod's own + // directory, which the ateom creates when it boots and which goes away + // with the pod. Checked before the sandbox record so the common + // pod-is-gone path does no other work — in particular, it never reaches + // ensureSandboxAssets, which would fetch binaries for a sandbox that no + // longer exists. + if _, err := os.Stat(ateomSocketPath(ateomUID)); err != nil { + if !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("failed to stat the ateom socket during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) + } + slog.InfoContext(ctx, "ateom is gone during terminate; reclaiming the actor's node state without a sandbox teardown", + slog.Any("actor", actorRef), slog.String("actorUID", actorUID), slog.String("ateomUID", ateomUID)) + return nil + } + + sandboxRec, err := readSandboxRecord(actorUID) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + // Written at Run/Restore, so its absence means no sandbox was ever + // started for this actor on this node. Nothing to tear down, and + // there is no runsc path to tear it down with. + slog.InfoContext(ctx, "no sandbox record during terminate; reclaiming the actor's node state without a sandbox teardown", + slog.Any("actor", actorRef), slog.String("actorUID", actorUID)) + return nil + } + return fmt.Errorf("failed to read sandbox record during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) + } + assetPaths, err := s.ensureSandboxAssets(ctx, sandboxRec) + if err != nil { + return fmt.Errorf("failed to ensure sandbox assets during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) + } + + client, err := s.dialAteom(ctx, ateomUID) + if err != nil { + return fmt.Errorf("failed to dial ateom for terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) + } + + if _, err := client.TerminateWorkload(ctx, &ateompb.TerminateWorkloadRequest{ + Atespace: req.GetAtespace(), + ActorName: req.GetActorName(), + ActorUid: req.GetActorUid(), + ActorTemplateAtespace: req.GetActorTemplateAtespace(), + ActorTemplateName: req.GetActorTemplateName(), + RunscPath: runscPathFor(assetPaths), + Spec: spec, + }); err != nil { + switch status.Code(err) { + case codes.NotFound: + slog.InfoContext(ctx, "workload not found on ateom during terminate", slog.Any("actor", actorRef), slog.String("actorUID", actorUID)) + case codes.Unavailable: + // The socket outlived the process serving it (an ateom shutting + // down while this RPC was in flight, or a stale socket file). Same + // situation as a missing socket: there is nothing left to tear + // down, and the state on disk still has to go. + slog.InfoContext(ctx, "ateom unreachable during terminate; reclaiming the actor's node state without a sandbox teardown", + slog.Any("actor", actorRef), slog.String("actorUID", actorUID), slog.Any("err", err)) + default: + return fmt.Errorf("failed calling ateom.TerminateWorkload (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) + } + } + return nil +} + func (s *AteomHerder) copyLocalCheckpoint(ctx context.Context, snapshotName string, srcDir, dstDir string, files []string) error { for _, fileName := range files { if ctx.Err() != nil { From f774f591a0e3997c077eadb6ababe82ff317136b Mon Sep 17 00:00:00 2001 From: Sairaj Pokale Date: Tue, 15 Sep 2026 18:58:46 +0000 Subject: [PATCH 2/3] atelet: sweep orphaned actor state directories Evictions, crashes and uninstalls strand actors/ forever: a UID is never revisited and nothing walks the tree. Arm B of the repro showed fix removed none. The sweep runs at startup and on a period (like imagegc.go), reconciling actors/ against the actors the control plane places on this node. Kept rather than swept: everything, if the live set cannot be read in full; actors this atelet is hosting; dirs younger than --actor-gc-min-age; and dirs holding a local pause snapshot, which a PAUSED actor has with no worker assignment to appear in the live set under. Deletion renames out of the UID namespace first, so a crash mid-delete leaves debris the next pass finishes. --- cmd/atelet/actorgc.go | 401 +++++++++++++++++++++++++++++++++ cmd/atelet/actorgc_test.go | 274 ++++++++++++++++++++++ cmd/atelet/main.go | 12 + cmd/atelet/systeminfovolume.go | 18 ++ 4 files changed, 705 insertions(+) create mode 100644 cmd/atelet/actorgc.go create mode 100644 cmd/atelet/actorgc_test.go diff --git a/cmd/atelet/actorgc.go b/cmd/atelet/actorgc.go new file mode 100644 index 0000000000..3523da714d --- /dev/null +++ b/cmd/atelet/actorgc.go @@ -0,0 +1,401 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +// The actor-state orphan sweep. +// +// Terminate reclaims an actor's directory on the graceful path, but no +// graceful path can be guaranteed: a node can be evicted, an atelet can crash +// mid-teardown, a control plane can be uninstalled and reinstalled on top of +// live state. Whatever those leave behind is unreachable forever — actor UIDs +// are unique, so the per-actor reset that Run/Restore/Checkpoint/Terminate +// perform is never invoked for that UID again, and nothing else walks +// actors/. Measured: 109 orphaned directories and 55.22 GB survived a +// workerpool scale-to-zero, a full --delete-all, and a clean reinstall. +// +// The image cache has the same shape of problem and the same shape of answer +// (see imagegc.go): a serialized pass on a fixed period, plus one at startup. +// The root set is the difference: the image cache reads it off the node, while +// only the control plane knows which actors are placed here — so a pass that +// cannot read that set deletes nothing at all. + +import ( + "context" + "errors" + "fmt" + "io/fs" + "log/slog" + "os" + "path/filepath" + "runtime/debug" + "strings" + "time" + + "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/internal/imagecache" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "github.com/spf13/pflag" +) + +var ( + actorGCPeriod = pflag.Duration("actor-gc-period", 5*time.Minute, "How often to sweep orphaned actor state directories. 0 disables the periodic pass (the startup pass still runs).") + actorGCMinAge = pflag.Duration("actor-gc-min-age", 10*time.Minute, "Actor directories younger than this are never swept, which covers the window between atelet creating one and the control plane recording where the actor was placed.") + actorGCDryRun = pflag.Bool("actor-gc-dry-run", false, "Log what the sweep would reclaim without deleting anything.") +) + +// actorGCListPageSize is the page size for the control-plane reads that build +// the live set. The API caps pages at 1000. +const actorGCListPageSize = 1000 + +// retiredActorPrefix marks an actor dir the sweep has renamed aside and is +// about to delete. Actor dirs are named by UID, which never starts with a dot, +// so a retired dir can never be mistaken for a live one — by this sweep, or by +// the image cache's root-set scan, which reads bundle specs out of the same +// tree and treats anything it still finds there as in use. +const retiredActorPrefix = ".rm-" + +func validateActorGCFlags() error { + if *actorGCPeriod < 0 { + // A negative period would silently disable the periodic pass: it + // fails the > 0 guard at the launch site, which also protects the + // ticker. + return fmt.Errorf("--actor-gc-period %v must be >= 0", *actorGCPeriod) + } + if *actorGCMinAge < 0 { + // A negative min-age inverts the veto — the cutoff lands in the + // future, making a directory created moments ago sweepable. + return fmt.Errorf("--actor-gc-min-age %v must be >= 0", *actorGCMinAge) + } + return nil +} + +// liveActorLister reports the UIDs of the actors placed on this node. +type liveActorLister interface { + liveActorUIDs(ctx context.Context) (map[string]bool, error) +} + +// controlPlaneActors answers the live set from the Control API: the actors +// assigned to the workers the control plane records on this node. +// +// The control plane is the only authority for this. atelet's own view is +// in-memory and starts empty (a restarted atelet on a busy node knows about +// none of the actors still running there), which is why it is a supplement to +// this set and never a substitute for it. +type controlPlaneActors struct { + client ateapipb.ControlClient + nodeName string +} + +// liveActorUIDs reads every worker on this node and every actor assigned to +// those workers. +// +// Any failure fails the whole call rather than returning a partial set: a +// half-read set is indistinguishable from "these actors no longer exist", and +// acting on it would delete live actors' state. +func (c *controlPlaneActors) liveActorUIDs(ctx context.Context) (map[string]bool, error) { + live := map[string]bool{} + for token := ""; ; { + page, err := c.client.ListWorkers(ctx, &ateapipb.ListWorkersRequest{ + PageSize: actorGCListPageSize, + PageToken: token, + }) + if err != nil { + return nil, fmt.Errorf("while listing workers: %w", err) + } + for _, w := range page.GetWorkers() { + if w.GetNodeName() != c.nodeName { + continue + } + if err := c.addAssignedActors(ctx, w.GetMetadata().GetName(), live); err != nil { + return nil, err + } + } + if page.GetNextPageToken() == "" { + return live, nil + } + token = page.GetNextPageToken() + } +} + +func (c *controlPlaneActors) addAssignedActors(ctx context.Context, workerName string, live map[string]bool) error { + for token := ""; ; { + page, err := c.client.ListWorkerActorAssignments(ctx, &ateapipb.ListWorkerActorAssignmentsRequest{ + Worker: &ateapipb.ObjectRef{Name: workerName}, + PageSize: actorGCListPageSize, + PageToken: token, + }) + if err != nil { + return fmt.Errorf("while listing the actors assigned to worker %s: %w", workerName, err) + } + for _, assignment := range page.GetActorAssignments() { + if uid := assignment.GetActorUid(); uid != "" { + live[uid] = true + } + } + if page.GetNextPageToken() == "" { + return nil + } + token = page.GetNextPageToken() + } +} + +// actorGC is the sweep's state: its configuration snapshotted from the flags +// at construction, so the pass logic never reads globals and is testable +// without flag juggling. +type actorGC struct { + actorsDir string + live liveActorLister + // resident reports the actors this atelet is currently hosting. Authoritative + // only in the positive direction — it starts empty after a restart — so it + // can protect a directory but never condemn one. + resident func() []string + period time.Duration + minAge time.Duration + dryRun bool +} + +func newActorGC(actorsDir string, live liveActorLister, resident func() []string) *actorGC { + return &actorGC{ + actorsDir: actorsDir, + live: live, + resident: resident, + period: *actorGCPeriod, + minAge: *actorGCMinAge, + dryRun: *actorGCDryRun, + } +} + +// Run sweeps on the configured period until ctx is done. Passes are strictly +// serialized: a slow pass delays the next tick rather than overlapping it. +func (g *actorGC) Run(ctx context.Context) { + // First pass immediately: the debris of the previous atelet's life is on + // disk now, and a node that lost its actors abruptly should not carry + // their state for a full period. + g.runPass(ctx) + + if g.period <= 0 { + return + } + ticker := time.NewTicker(g.period) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + g.runPass(ctx) + } +} + +// actorGCStats counts what one pass did, for its single log line. +type actorGCStats struct { + Reclaimed int + ReclaimedBytes int64 + Live int + Resident int + Fresh int + LocalSnapshot int + Failed int +} + +// runPass performs one sweep. It recovers from panics: this is a background +// janitor, and a bug here — or a directory an operator dropped into the tree — +// must not take atelet down and strand every actor on the node. +func (g *actorGC) runPass(ctx context.Context) { + defer func() { + if r := recover(); r != nil { + slog.ErrorContext(ctx, "Actor GC pass panicked; skipping this pass", + slog.Any("panic", r), slog.String("stack", string(debug.Stack()))) + } + }() + + entries, err := os.ReadDir(g.actorsDir) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + // No actor has ever run here. + return + } + slog.WarnContext(ctx, "Actor GC: listing the actors dir failed; skipping this pass", + slog.String("dir", g.actorsDir), slog.Any("err", err)) + return + } + + // Read before anything is deleted, and required to succeed: a pass with no + // trustworthy root set has nothing to distinguish an orphan from a running + // actor, so it deletes nothing. This is the state during a control-plane + // outage, and the cost of waiting it out is disk that is already spent. + live, err := g.live.liveActorUIDs(ctx) + if err != nil { + slog.WarnContext(ctx, "Actor GC: reading the live actor set failed; skipping this pass (no directory is swept without one)", + slog.Any("err", err)) + return + } + resident := map[string]bool{} + if g.resident != nil { + for _, uid := range g.resident() { + resident[uid] = true + } + } + + tStart := time.Now() + cutoff := tStart.Add(-g.minAge) + var stats actorGCStats + for _, entry := range entries { + if !entry.IsDir() { + continue + } + name := entry.Name() + if strings.HasPrefix(name, retiredActorPrefix) { + // Debris from a pass that died between the rename and the + // delete. It is already unreachable; finish the job. + g.remove(ctx, filepath.Join(g.actorsDir, name), &stats) + continue + } + if live[name] { + stats.Live++ + continue + } + if resident[name] { + // The control plane does not (yet) place this actor here, but + // this atelet is running it: a bind committed after the listing, + // or a release racing a teardown in flight. Either way its + // directory is in use. + stats.Resident++ + continue + } + info, err := entry.Info() + if err != nil { + // Vanished under us, or unreadable. Retention is the safe answer + // and the next pass sees it again. + stats.Failed++ + continue + } + if info.ModTime().After(cutoff) { + // Young enough that the control plane may not have recorded the + // placement yet. + stats.Fresh++ + continue + } + dir := filepath.Join(g.actorsDir, name) + if hasLocalSnapshot(name) { + // A pause snapshot is node-pinned state held deliberately, and a + // PAUSED actor holds no worker assignment, so it is absent from + // the live set by construction. Deleting one is unrecoverable; + // keeping it costs disk that a later delete still reclaims. + stats.LocalSnapshot++ + continue + } + g.retireAndRemove(ctx, dir, name, &stats) + } + + attrs := []any{ + slog.Int("reclaimed_dirs", stats.Reclaimed), + slog.Int64("reclaimed_bytes", stats.ReclaimedBytes), + slog.Int("live_actors", stats.Live), + slog.Int("skipped_resident", stats.Resident), + slog.Int("skipped_fresh", stats.Fresh), + slog.Int("skipped_local_snapshot", stats.LocalSnapshot), + slog.Int("failed", stats.Failed), + slog.Bool("dry_run", g.dryRun), + slog.Duration("took", time.Since(tStart)), + } + if stats.Reclaimed == 0 && stats.Failed == 0 { + // The steady state on a healthy node: nothing to say at INFO every + // period. + slog.DebugContext(ctx, "Actor GC pass complete", attrs...) + return + } + slog.InfoContext(ctx, "Actor GC pass complete", attrs...) +} + +// retireAndRemove reclaims one orphaned actor directory in two phases: a +// rename out of the UID namespace, then the slow delete. A crash in between +// leaves a ".rm-*" dir the next pass finishes, rather than a half-emptied +// directory still named after an actor. +func (g *actorGC) retireAndRemove(ctx context.Context, dir, uid string, stats *actorGCStats) { + size := dirSize(dir) + if g.dryRun { + slog.InfoContext(ctx, "Actor GC would reclaim an orphaned actor directory", + slog.String("actor_uid", uid), slog.Int64("bytes", size)) + stats.Reclaimed++ + stats.ReclaimedBytes += size + return + } + retired := filepath.Join(g.actorsDir, fmt.Sprintf("%s%s-%d", retiredActorPrefix, uid, time.Now().UnixNano())) + if err := os.Rename(dir, retired); err != nil { + if errors.Is(err, os.ErrNotExist) { + return + } + slog.WarnContext(ctx, "Actor GC: retiring an orphaned actor directory failed", + slog.String("actor_uid", uid), slog.Any("err", err)) + stats.Failed++ + return + } + slog.InfoContext(ctx, "Actor GC reclaiming an orphaned actor directory", + slog.String("actor_uid", uid), slog.Int64("bytes", size)) + before := stats.Failed + g.remove(ctx, retired, stats) + if stats.Failed == before { + stats.Reclaimed++ + stats.ReclaimedBytes += size + } +} + +// remove deletes a retired directory. RemoveAllWritable, not os.RemoveAll: a +// bundle's upper dir can hold copied-up image directories carrying the image's +// read-only modes, which atelet cannot remove as plain root without making +// them writable first (same reason resetActorDirs uses it). +func (g *actorGC) remove(ctx context.Context, dir string, stats *actorGCStats) { + if g.dryRun { + return + } + if err := imagecache.RemoveAllWritable(dir); err != nil { + // It is out of the UID namespace already, so it is nothing but bytes; + // the next pass retries. + slog.WarnContext(ctx, "Actor GC: deleting a retired actor directory failed", + slog.String("dir", dir), slog.Any("err", err)) + stats.Failed++ + } +} + +// hasLocalSnapshot reports whether the actor holds at least one local (pause) +// snapshot on this node. +func hasLocalSnapshot(actorUID string) bool { + entries, err := os.ReadDir(ateompath.LocalCheckpointsDir(actorUID)) + if err != nil { + // Missing is the ordinary case. Anything else is unreadable, which + // this reports as "has one" so the directory is kept. + return !errors.Is(err, os.ErrNotExist) + } + return len(entries) > 0 +} + +// dirSize sums the apparent size of a tree, for the reclaimed-bytes figure in +// the pass log. Best-effort: it is telemetry, not a decision input, so an +// unreadable entry is skipped rather than failing the reclaim. +func dirSize(dir string) int64 { + var total int64 + _ = filepath.WalkDir(dir, func(_ string, d fs.DirEntry, err error) error { + if err != nil || d.IsDir() { + return nil + } + if info, err := d.Info(); err == nil { + total += info.Size() + } + return nil + }) + return total +} diff --git a/cmd/atelet/actorgc_test.go b/cmd/atelet/actorgc_test.go new file mode 100644 index 0000000000..62bf3ddd88 --- /dev/null +++ b/cmd/atelet/actorgc_test.go @@ -0,0 +1,274 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "context" + "errors" + "os" + "path/filepath" + "slices" + "testing" + "time" + + "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc" +) + +// fakeLiveActors is a live set, or the failure to read one. +type fakeLiveActors struct { + uids []string + err error +} + +func (f fakeLiveActors) liveActorUIDs(context.Context) (map[string]bool, error) { + if f.err != nil { + return nil, f.err + } + live := map[string]bool{} + for _, uid := range f.uids { + live[uid] = true + } + return live, nil +} + +// seedActorDir creates an actor's state directory with a payload in it, aged +// to look like state left behind rather than state being set up right now. +func seedActorDir(t *testing.T, uid string, age time.Duration) string { + t.Helper() + dir := ateompath.ActorPath(uid) + payload := filepath.Join(dir, "durable-dir", "vol") + if err := os.MkdirAll(payload, 0o755); err != nil { + t.Fatalf("seeding %s: %v", dir, err) + } + if err := os.WriteFile(filepath.Join(payload, "data"), []byte("actor state"), 0o600); err != nil { + t.Fatalf("seeding %s: %v", dir, err) + } + aged := time.Now().Add(-age) + if err := os.Chtimes(dir, aged, aged); err != nil { + t.Fatalf("aging %s: %v", dir, err) + } + return dir +} + +func newTestActorGC(t *testing.T, live liveActorLister, resident func() []string) *actorGC { + t.Helper() + return &actorGC{ + actorsDir: ateompath.ActorsDir, + live: live, + resident: resident, + minAge: 10 * time.Minute, + } +} + +// The sweep is the safety net for every path that cannot terminate +// gracefully — an evicted node, a crashed atelet, an uninstall on top of live +// state — so what it keeps matters as much as what it reclaims. Each row is +// one reason a directory is or is not the sweep's to take. +func TestActorGCSweep(t *testing.T) { + const ( + orphan = "actor-orphan" + assigned = "actor-assigned" + ) + + tests := []struct { + name string + // seed arranges the tree and returns the dirs that must survive the + // pass; every other seeded dir must be gone. + seed func(t *testing.T) (keep []string) + live fakeLiveActors + resident []string + dryRun bool + }{ + { + name: "an orphan is reclaimed", + seed: func(t *testing.T) []string { + seedActorDir(t, orphan, time.Hour) + return nil + }, + }, + { + name: "an actor the control plane placed here is kept", + seed: func(t *testing.T) []string { + return []string{seedActorDir(t, assigned, time.Hour)} + }, + live: fakeLiveActors{uids: []string{assigned}}, + }, + { + // The control plane does not name it, but this atelet is running + // it: a bind that landed after the listing, or a teardown in + // flight. + name: "an actor this atelet is hosting is kept", + seed: func(t *testing.T) []string { + return []string{seedActorDir(t, orphan, time.Hour)} + }, + resident: []string{orphan}, + }, + { + // The window between atelet creating the directory and the + // control plane recording where the actor was placed. + name: "a directory younger than min-age is kept", + seed: func(t *testing.T) []string { + return []string{seedActorDir(t, orphan, time.Minute)} + }, + }, + { + // A PAUSED actor holds no worker assignment, so it is absent from + // the live set by construction; its pause snapshot is node-pinned + // state held deliberately, and deleting it is unrecoverable. + name: "an actor holding a local snapshot is kept", + seed: func(t *testing.T) []string { + dir := seedActorDir(t, orphan, time.Hour) + snap := ateompath.LocalSnapshotDir(orphan, "pause-1") + if err := os.MkdirAll(snap, 0o755); err != nil { + t.Fatalf("seeding %s: %v", snap, err) + } + return []string{dir} + }, + }, + { + // Without a root set an orphan is indistinguishable from a + // running actor, so the pass deletes nothing at all. + name: "nothing is swept when the live set cannot be read", + seed: func(t *testing.T) []string { + return []string{seedActorDir(t, orphan, time.Hour)} + }, + live: fakeLiveActors{err: errors.New("control plane is down")}, + }, + { + name: "a dry run reclaims nothing", + seed: func(t *testing.T) []string { + return []string{seedActorDir(t, orphan, time.Hour)} + }, + dryRun: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + useTempNodeDirs(t) + if err := os.MkdirAll(ateompath.ActorsDir, 0o755); err != nil { + t.Fatalf("creating the actors dir: %v", err) + } + keep := tt.seed(t) + + gc := newTestActorGC(t, tt.live, func() []string { return tt.resident }) + gc.dryRun = tt.dryRun + gc.runPass(t.Context()) + + entries, err := os.ReadDir(ateompath.ActorsDir) + if err != nil { + t.Fatalf("reading the actors dir: %v", err) + } + var left []string + for _, e := range entries { + left = append(left, filepath.Join(ateompath.ActorsDir, e.Name())) + } + slices.Sort(left) + slices.Sort(keep) + if !slices.Equal(left, keep) { + t.Errorf("actors dir holds %v after the pass, want %v", left, keep) + } + }) + } +} + +// A pass that dies between the rename and the delete leaves the tree out of +// the UID namespace but still on disk. It is already unreachable, so the next +// pass finishes it without consulting anything. +func TestActorGCFinishesRetiredDirs(t *testing.T) { + useTempNodeDirs(t) + if err := os.MkdirAll(ateompath.ActorsDir, 0o755); err != nil { + t.Fatalf("creating the actors dir: %v", err) + } + retired := filepath.Join(ateompath.ActorsDir, retiredActorPrefix+"actor-1-12345") + if err := os.MkdirAll(filepath.Join(retired, "durable-dir"), 0o755); err != nil { + t.Fatalf("seeding %s: %v", retired, err) + } + + newTestActorGC(t, fakeLiveActors{}, nil).runPass(t.Context()) + + if _, err := os.Stat(retired); !os.IsNotExist(err) { + t.Errorf("retired dir survived the pass (stat err = %v)", err) + } +} + +// The live set is the actors assigned to the workers on this node, and only +// this node: another node's workers hold actors whose directories live over +// there, and counting them here would say nothing about this disk. +func TestControlPlaneActorsLiveSetIsNodeScoped(t *testing.T) { + client := &fakeControlClient{ + workers: []*ateapipb.Worker{ + {Metadata: &ateapipb.ResourceMetadata{Name: "worker-here"}, NodeName: "node-1"}, + {Metadata: &ateapipb.ResourceMetadata{Name: "worker-elsewhere"}, NodeName: "node-2"}, + }, + assignments: map[string][]string{ + "worker-here": {"actor-a", "actor-b"}, + "worker-elsewhere": {"actor-c"}, + }, + } + + live, err := (&controlPlaneActors{client: client, nodeName: "node-1"}).liveActorUIDs(context.Background()) + if err != nil { + t.Fatalf("liveActorUIDs: %v", err) + } + want := map[string]bool{"actor-a": true, "actor-b": true} + if len(live) != len(want) { + t.Fatalf("live set = %v, want %v", live, want) + } + for uid := range want { + if !live[uid] { + t.Errorf("live set %v is missing %q", live, uid) + } + } +} + +// A listing that fails partway fails the whole call: a half-read set looks +// exactly like a set of actors that no longer exist. +func TestControlPlaneActorsFailsClosedOnAssignmentError(t *testing.T) { + client := &fakeControlClient{ + workers: []*ateapipb.Worker{{Metadata: &ateapipb.ResourceMetadata{Name: "worker-here"}, NodeName: "node-1"}}, + assignmentErr: errors.New("db is down"), + } + + if _, err := (&controlPlaneActors{client: client, nodeName: "node-1"}).liveActorUIDs(context.Background()); err == nil { + t.Fatal("liveActorUIDs() = nil error, want the listing failure reported") + } +} + +// fakeControlClient answers the two Control API reads the live set is built +// from and nothing else. +type fakeControlClient struct { + ateapipb.ControlClient + workers []*ateapipb.Worker + assignments map[string][]string + assignmentErr error +} + +func (f *fakeControlClient) ListWorkers(context.Context, *ateapipb.ListWorkersRequest, ...grpc.CallOption) (*ateapipb.ListWorkersResponse, error) { + return &ateapipb.ListWorkersResponse{Workers: f.workers}, nil +} + +func (f *fakeControlClient) ListWorkerActorAssignments(_ context.Context, req *ateapipb.ListWorkerActorAssignmentsRequest, _ ...grpc.CallOption) (*ateapipb.ListWorkerActorAssignmentsResponse, error) { + if f.assignmentErr != nil { + return nil, f.assignmentErr + } + resp := &ateapipb.ListWorkerActorAssignmentsResponse{} + for _, uid := range f.assignments[req.GetWorker().GetName()] { + resp.ActorAssignments = append(resp.ActorAssignments, &ateapipb.ActorAssignment{ActorUid: uid}) + } + return resp, nil +} diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 6bc7679a4a..178677d3e8 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -352,6 +352,18 @@ func main() { serverboot.Fatal(ctx, "Failed to load atelet Pod identity", fmt.Errorf("credential bundle has no Pod identity")) } + // Started here because the sweep is scoped by node, and the node name is + // what the atelet's own certificate asserts — the same identity the + // control plane authenticates it by. Only the pod-identity signer can mint + // it, so it cannot disagree with where this atelet actually runs. + if err := validateActorGCFlags(); err != nil { + serverboot.Fatal(ctx, "Invalid actor GC flags", err) + } + go newActorGC( + ateompath.ActorsDir, + &controlPlaneActors{client: ateapipb.NewControlClient(ateapiConn), nodeName: ateletIdentity.NodeName}, + systemInfoVolumes.RegisteredActorUIDs, + ).Run(ctx) ateomFacingTLS := tlsCfg.Clone() ateomFacingTLS.VerifyConnection = verifyClientOnSameNode(ateletIdentity) if err := os.Remove(ateompath.AteomSupportSocket); err != nil && !errors.Is(err, os.ErrNotExist) { diff --git a/cmd/atelet/systeminfovolume.go b/cmd/atelet/systeminfovolume.go index fa59db1ec3..6067f3a9b8 100644 --- a/cmd/atelet/systeminfovolume.go +++ b/cmd/atelet/systeminfovolume.go @@ -147,6 +147,24 @@ func (r *systemInfoVolumeRefresher) Deregister(actorUID string) { } } +// RegisteredActorUIDs returns the actors currently registered here, which is +// the set this atelet is hosting: Run and Restore register, Terminate +// deregisters. +// +// Trustworthy in the positive direction only. The registry is in memory and a +// restarted atelet starts with an empty one while the actors it served keep +// running (TODO(#1372)), so a caller may treat a UID here as live but must not +// treat an absent one as dead. +func (r *systemInfoVolumeRefresher) RegisteredActorUIDs() []string { + r.mu.Lock() + defer r.mu.Unlock() + uids := make([]string, 0, len(r.actors)) + for uid := range r.actors { + uids = append(uids, uid) + } + return uids +} + // collectData builds the volume's contents keyed by volume-relative path, // plus each projected bundle's trustBundleHash. func (r *systemInfoVolumeRefresher) collectData(ref resources.ActorRef, actorUID string, si *ateletpb.SystemInfoVolume) (payload map[string][]byte, bundleHashes map[string]string, err error) { From b277cab7b664a7f7e9fd3ee8f5e489de2faf5017 Mon Sep 17 00:00:00 2001 From: Sairaj Pokale Date: Tue, 15 Sep 2026 19:02:03 +0000 Subject: [PATCH 3/3] hack: wipe the node state hostPath on uninstall --delete-all tears down the control plane but never touches the hostPath it wrote to, so the data outlives the software: 55.22 GB across 3 nodes survived it, a reinstall, and fresh atelets starting on top of it. delete_ate_system now runs a short-lived DaemonSet that empties /var/lib/ateom-gvisor on every node, after the atelet DaemonSet is gone so nothing is writing to it. Readiness is the wipe having finished, so the rollout completing means every node is clean. --keep-node-state opts out. --- hack/install-ate.sh | 85 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 85 insertions(+) diff --git a/hack/install-ate.sh b/hack/install-ate.sh index 5a2f411428..67a6f96ea4 100755 --- a/hack/install-ate.sh +++ b/hack/install-ate.sh @@ -77,6 +77,7 @@ function usage() { echo " a bare --setup-csi means nfs; hostpath is Kind only)" echo " --delete-ate-system Delete core system" echo " --delete-all Delete core system and all registered demos" + echo " --keep-node-state Leave /var/lib/ateom-gvisor in place when deleting (default: wipe it on every node)" echo " --atenet-dataplane=envoy|agentgateway Select the atenet ingress and egress dataplane (default: envoy)" echo " --podcert-workers-per-signer N Concurrent workers per podcertificate-controller signer (default: 1)" echo " --rollout-timeout DURATION Per-workload readiness wait timeout, kubectl-style Go duration (default: 60s)" @@ -1347,6 +1348,84 @@ delete_ate_system() { run_kubectl delete --ignore-not-found -f manifests/ate-install/postgres/postgres.yaml run_kubectl delete --ignore-not-found -f manifests/ate-install/generated run_kubectl label nodes -l ate.dev/substrate-version ate.dev/substrate-version- + # Last, and only now: the wipe below needs every atelet stopped, and the + # deletes above are what stop them. + delete_node_state +} + +# delete_node_state empties the hostPath the install writes to on every node. +# +# Deleting the control plane does not touch it -- the data outlives the +# software, so "uninstall" otherwise leaves tens of GB per node behind +# (measured: 55.22 GB across 3 nodes survived --delete-all and a reinstall, +# and fresh atelets started on top of it). Only replacing the node reclaimed +# it. +# +# Runs as a DaemonSet because the state is per-node and there is no other way +# to reach a node's filesystem: atelet is distroless and, by this point, +# deleted. The pod tolerates everything so no node is missed, and lives in +# default rather than ate-system, which this teardown removes. +delete_node_state() { + if [[ "${ATE_KEEP_NODE_STATE:-false}" == "true" ]]; then + log_step "delete_node_state (skipped: --keep-node-state)" + return 0 + fi + log_step "delete_node_state" + + local name="ate-node-state-cleanup" + # Contents, not the directory: it is the pod's own mount point, so removing + # it would fail as busy. What matters is the bytes. + run_kubectl apply -f - </dev/null || true + touch /tmp/done + # Stay up so the rollout can observe readiness; the delete below + # ends it as soon as every node has reported. + sleep 3600 + readinessProbe: + exec: + command: ["/bin/sh", "-c", "test -f /tmp/done"] + initialDelaySeconds: 1 + periodSeconds: 2 + volumeMounts: + - name: node-state + mountPath: /host-state + volumes: + - name: node-state + hostPath: + path: /var/lib/ateom-gvisor + type: DirectoryOrCreate +EOF + # Readiness is the wipe having finished, so the rollout completing means + # every node is clean. A node that cannot be reached is reported rather + # than silently skipped. + if ! run_kubectl rollout status daemonset/"${name}" -n default --timeout=10m; then + echo "warning: node state cleanup did not complete on every node; check daemonset/${name}" >&2 + fi + run_kubectl delete --ignore-not-found daemonset/"${name}" -n default } delete_atenet() { @@ -1441,6 +1520,9 @@ for ((i = 0; i < ${#prescan_args[@]}; i++)); do ATE_ATENET_DATAPLANE="${prescan_args[$((i + 1))]}" ;; --experimental-use-sdsmint) ATE_EXPERIMENTAL_USE_SDSMINT=true ;; + # A modifier on the delete actions, pre-scanned so it can be passed on + # either side of the --delete-* flag it applies to. + --keep-node-state) ATE_KEEP_NODE_STATE=true ;; --experimental-additional-egress-extproc-service=*) ATE_ADDITIONAL_EGRESS_EXTPROC_SERVICE="${prescan_args[i]#*=}" ;; @@ -1591,6 +1673,9 @@ while [[ "$#" -gt 0 ]]; do ;; --delete-ate-system) delete_ate_system ;; --delete-all) delete_all ;; + # Captured in the pre-scan above; consumed here so the unknown-option + # branch does not reject it. + --keep-node-state) ;; --deploy-atelet) deploy_atelet ;; --deploy-ate-apiserver) deploy_ate_apiserver ;;