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/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/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..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) { @@ -1276,40 +1288,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 +1324,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 { 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) { 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 ;;