From a588253060f25784a9cc5201f4ae75c3bd8151c5 Mon Sep 17 00:00:00 2001 From: Amp Date: Tue, 27 Oct 2026 02:35:33 +0000 Subject: [PATCH 1/5] Finish deleting an actor whose node is gone instead of retrying its atelet forever - DeleteActor's ensureAteletTerminated dialed the actor's atelet on every reap, and when the node no longer existed the dial failed with Unavailable each time, so the reaper looped on a DELETING actor for good (seen on a customer's EKS cluster after a worker node pool was replaced during an upgrade) - The control-plane atelet dialer now consults a node lister: a dial to an atelet on a node the cluster has lost reports ErrNodeGone, and the delete workflow skips the terminate, since nothing on that node is left to terminate - ate-api-server's ClusterRole gains get, watch and list on nodes for the lister - TestDeleteActor_SkipsTerminateOnGoneNode covers the delete; the dialer test covers the node check Amp-Thread-ID: https://ampcode.com/threads/T-01a178d8-73f0-751b-8cb4-c5a11dd7ef41 Co-authored-by: Hitesh Sagtani --- cmd/ateapi/internal/controlapi/dialer.go | 22 +++++++++++ cmd/ateapi/internal/controlapi/dialer_test.go | 16 ++++++++ .../internal/controlapi/workflow_delete.go | 10 +++++ .../controlapi/workflow_delete_test.go | 39 +++++++++++++++++++ cmd/ateapi/main.go | 5 ++- manifests/ate-install/ate-api-server.yaml | 5 +++ 6 files changed, 96 insertions(+), 1 deletion(-) diff --git a/cmd/ateapi/internal/controlapi/dialer.go b/cmd/ateapi/internal/controlapi/dialer.go index 4b96a60f42..cbb98e43e8 100644 --- a/cmd/ateapi/internal/controlapi/dialer.go +++ b/cmd/ateapi/internal/controlapi/dialer.go @@ -35,6 +35,8 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + corev1listers "k8s.io/client-go/listers/core/v1" "k8s.io/client-go/tools/cache" "k8s.io/utils/lru" ) @@ -44,6 +46,12 @@ import ( // Retryable. var ErrNoAteletOnNode = errors.New("no atelet pod found on node") +// ErrNodeGone reports that the requested node no longer exists in the +// cluster, so no atelet will ever serve on it again: whatever an actor left +// there is gone with the node. Not retryable. Reported only by a dialer built +// WithNodeLister; without one a gone node reports ErrNoAteletOnNode. +var ErrNodeGone = errors.New("node no longer exists") + // ateletConn is a cached connection and the pod IP it was dialed on. type ateletConn struct { ip string @@ -53,6 +61,9 @@ type ateletConn struct { // AteletDialer handles gRPC connections to Atelet pods. type AteletDialer struct { ateletIndexer cache.Indexer + // nodeLister, when set, tells a node that is gone (ErrNodeGone) from one + // whose atelet is merely absent (ErrNoAteletOnNode). + nodeLister corev1listers.NodeLister // mu makes the lookup, stale eviction and insert in DialForAteletOnNode // atomic, so concurrent callers cannot evict each other's fresh conn. mu sync.Mutex @@ -74,6 +85,12 @@ func WithDialCredentials(build func(expectedPodUID string) (credentials.Transpor return func(d *AteletDialer) { d.dialCredentials = build } } +// WithNodeLister lets the dialer report ErrNodeGone for a node the cluster no +// longer has, instead of ErrNoAteletOnNode. +func WithNodeLister(nodes corev1listers.NodeLister) DialerOption { + return func(d *AteletDialer) { d.nodeLister = nodes } +} + // NewAteletDialer creates a new AteletDialer. clientBundlePath and serverCAPath // are used to build the per-atelet mTLS credentials used for every atelet // connection, and ateletSPIFFEID is the identity those credentials expect on @@ -125,6 +142,11 @@ func (d *AteletDialer) DialForAteletOnNode(nodeName string) (*grpc.ClientConn, e } if len(matchingAtelets) == 0 { + if d.nodeLister != nil { + if _, err := d.nodeLister.Get(nodeName); apierrors.IsNotFound(err) { + return nil, fmt.Errorf("%w: %q", ErrNodeGone, nodeName) + } + } return nil, fmt.Errorf("%w: %q", ErrNoAteletOnNode, nodeName) } if len(matchingAtelets) > 1 { diff --git a/cmd/ateapi/internal/controlapi/dialer_test.go b/cmd/ateapi/internal/controlapi/dialer_test.go index 43215e7942..32bf2d6bc0 100644 --- a/cmd/ateapi/internal/controlapi/dialer_test.go +++ b/cmd/ateapi/internal/controlapi/dialer_test.go @@ -41,6 +41,7 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" + corev1listers "k8s.io/client-go/listers/core/v1" "k8s.io/client-go/tools/cache" ) @@ -330,6 +331,21 @@ func TestDialForAteletOnNode(t *testing.T) { } }) + t.Run("no atelet on a node the cluster lost", func(t *testing.T) { + nodes := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + if err := nodes.Add(&corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node2"}}); err != nil { + t.Fatalf("adding node to indexer: %v", err) + } + d := NewAteletDialer(newTestAteletIndexer(t), installdefaults.AteletSPIFFEID(installdefaults.SystemNamespace), "", "", + WithNodeLister(corev1listers.NewNodeLister(nodes))) + if _, err := d.DialForAteletOnNode("node1"); !errors.Is(err, ErrNodeGone) { + t.Fatalf("DialForAteletOnNode(gone node) = %v, want ErrNodeGone", err) + } + if _, err := d.DialForAteletOnNode("node2"); !errors.Is(err, ErrNoAteletOnNode) { + t.Fatalf("DialForAteletOnNode(node without atelet) = %v, want ErrNoAteletOnNode", err) + } + }) + t.Run("more than one atelet on node", func(t *testing.T) { d := NewAteletDialer(newTestAteletIndexer(t, ateletPod("atelet-1", "uid-1", "node1", "10.0.0.1"), diff --git a/cmd/ateapi/internal/controlapi/workflow_delete.go b/cmd/ateapi/internal/controlapi/workflow_delete.go index de65c19a54..f24c55010b 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete.go @@ -155,6 +155,16 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res conn, err := w.dialer.DialForAteletOnNode(nodeName) if err != nil { + // The node went away with everything the actor left on it: its + // workload, local snapshots and mounts. There is nothing to terminate, + // and retrying would hold the actor in DELETING for good (the reaper + // re-deletes a DELETING actor every pass). + if errors.Is(err, ErrNodeGone) { + slog.InfoContext(ctx, "node that hosted the actor no longer exists; skipping atelet terminate request", + slog.String("node", nodeName), + slog.Any("actor", actorRef)) + return nil + } return fmt.Errorf("while connecting to atelet on node %q: %w", nodeName, err) } diff --git a/cmd/ateapi/internal/controlapi/workflow_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_delete_test.go index 9f86bc31d6..a8a56a5e09 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete_test.go @@ -23,12 +23,15 @@ import ( "go.ate.dev/substrate/cmd/ateapi/internal/store" "go.ate.dev/substrate/cmd/ateapi/internal/store/storetest" "go.ate.dev/substrate/internal/apierror" + "go.ate.dev/substrate/internal/installdefaults" "go.ate.dev/substrate/internal/objectstore/objectstoretest" "go.ate.dev/substrate/internal/objectstoreplugin/objectstoreplugintest" "go.ate.dev/substrate/internal/resources" "go.ate.dev/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + corev1listers "k8s.io/client-go/listers/core/v1" + "k8s.io/client-go/tools/cache" ) func TestDeleteActorWorkflow_ExecutionPaths(t *testing.T) { @@ -511,3 +514,39 @@ func TestDeleteActor_CollectsSnapshotsAfterWorkerDelete(t *testing.T) { }) } } + +// A crashed actor keeps the node it ran on as a record. When that node has +// left the cluster, deleting the actor has nothing to terminate there and must +// not wait for an atelet that never comes back. +func TestDeleteActor_SkipsTerminateOnGoneNode(t *testing.T) { + ctx := context.Background() + st, cleanup := storetest.SetupTestStore(t) + defer cleanup() + w := newTestActorWorkflow(t, st, "ns", "tmpl1") + + actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} + seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_CRASHED, func(a *ateapipb.Actor) { + a.Status.AssignedNode = "gone-node" + }) + + // The node is absent from the lister and no atelet is indexed on it. + noNodes := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + w.dialer = NewAteletDialer(newTestAteletIndexer(t), installdefaults.AteletSPIFFEID(installdefaults.SystemNamespace), "", "", + WithNodeLister(corev1listers.NewNodeLister(noNodes))) + + if _, err := w.DeleteActor(ctx, actorRef, false, store.DeletePreconditions{}); err != nil { + t.Fatalf("DeleteActor on a gone node: %v", err) + } + if _, err := st.GetActor(ctx, actorRef); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("GetActor after delete: err = %v, want ErrNotFound", err) + } + + // Without a node lister the same delete keeps retrying, as before. + seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_CRASHED, func(a *ateapipb.Actor) { + a.Status.AssignedNode = "gone-node" + }) + w.dialer = newDanglingDialer() + if _, err := w.DeleteActor(ctx, actorRef, false, store.DeletePreconditions{}); !errors.Is(err, ErrNoAteletOnNode) { + t.Fatalf("DeleteActor without a node lister: err = %v, want ErrNoAteletOnNode", err) + } +} diff --git a/cmd/ateapi/main.go b/cmd/ateapi/main.go index be265399f6..30a77c0358 100644 --- a/cmd/ateapi/main.go +++ b/cmd/ateapi/main.go @@ -284,6 +284,9 @@ func main() { ateletPodInformerFactory, ateletPodInformer := controlapi.AteletInformer(clientset, ateletNamespace) scInformerFactory := informers.NewSharedInformerFactory(clientset, 0) storageClassLister := scInformerFactory.Storage().V1().StorageClasses().Lister() + // Lets the atelet dialer tell a node the cluster lost from one whose + // atelet is restarting, so a delete does not wait forever on the former. + nodeLister := scInformerFactory.Core().V1().Nodes().Lister() stopCh := make(chan struct{}) defer close(stopCh) @@ -322,7 +325,7 @@ func main() { } volPlugins := make(map[string]volume.VolumePluginControlPlane) - ateletDialer := controlapi.NewAteletDialer(ateletPodInformer.GetIndexer(), ateletSPIFFEID, *ateletClientCredBundle, *podIdentityCACerts) + ateletDialer := controlapi.NewAteletDialer(ateletPodInformer.GetIndexer(), ateletSPIFFEID, *ateletClientCredBundle, *podIdentityCACerts, controlapi.WithNodeLister(nodeLister)) actorIDCAPool, err := localca.NewRefreshingPool(*actorIDCAPoolFile) if err != nil { diff --git a/manifests/ate-install/ate-api-server.yaml b/manifests/ate-install/ate-api-server.yaml index b3fcaab851..93b26943cb 100644 --- a/manifests/ate-install/ate-api-server.yaml +++ b/manifests/ate-install/ate-api-server.yaml @@ -22,6 +22,11 @@ rules: - apiGroups: [""] resources: ["pods"] verbs: ["get", "watch", "list"] +# Nodes: to tell a node the cluster lost (nothing left to terminate on it) +# from one whose atelet is restarting. +- apiGroups: [""] + resources: ["nodes"] + verbs: ["get", "watch", "list"] - apiGroups: ["ate.dev"] resources: ["workerpools", "sandboxconfigs", "csidriverconfigs"] verbs: ["get", "watch", "list"] From 515813d4feea4634589dea1249def1000c474f31 Mon Sep 17 00:00:00 2001 From: Amp Date: Tue, 27 Oct 2026 02:35:34 +0000 Subject: [PATCH 2/5] Report transient object-store HTTP failures from the plugin as Unavailable - The object-store plugin reported every storage SDK error whose retries were exhausted as Internal, so a GCS 503 during a checkpoint upload crashed the actor: the control plane retries only Unavailable, Canceled and DeadlineExceeded from atelet - HTTP 408, 429 and 5xx from the storage service now map to Unavailable; 4xx such as 403 stay Internal - Seen during the 2026-10-26 production Substrate upgrade, where five sandboxes were lost to a 503 burst on upload Amp-Thread-ID: https://ampcode.com/threads/T-01a178d8-73f0-751b-8cb4-c5a11dd7ef41 Co-authored-by: Hitesh Sagtani --- internal/objectstoreplugin/status.go | 16 ++++++++++++++++ internal/objectstoreplugin/status_test.go | 8 +++++++- 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/internal/objectstoreplugin/status.go b/internal/objectstoreplugin/status.go index 60ba1382ac..a7fada23a3 100644 --- a/internal/objectstoreplugin/status.go +++ b/internal/objectstoreplugin/status.go @@ -19,6 +19,7 @@ import ( "errors" "fmt" "log/slog" + "net/http" "go.ate.dev/substrate/internal/objectstore" "go.ate.dev/substrate/pkg/objectstorage" @@ -39,12 +40,27 @@ func toStatus(err error) error { case errors.Is(err, context.DeadlineExceeded): return status.Error(codes.DeadlineExceeded, err.Error()) } + // A storage service that is throttling or failing over answers 408, 429 or + // 5xx. The backend client already retried those with a bounded backoff, + // so exhausting it means the outage outlasted the budget, not that the + // request is wrong. Unavailable lets atelet report the call as retryable + // and the control plane retry the operation instead of crashing the actor. + if desc, ok := objectstorage.DescribeServiceError(err); ok && isTransientHTTPStatus(desc.HTTPStatus) { + slog.Info("Object-store plugin hit a transient storage service error; returning Unavailable", internalErrorAttrs(err)...) + return status.Error(codes.Unavailable, err.Error()) + } // The caller sees only Internal, and atelet deliberately drops a plugin's // status text, so this log line is the one record of why the plugin failed. slog.Warn("Object-store plugin failed with an unrecognized error; returning Internal", internalErrorAttrs(err)...) return status.Error(codes.Internal, err.Error()) } +// isTransientHTTPStatus reports whether a storage service response with this +// status is worth retrying later: a timeout, throttling, or a server error. +func isTransientHTTPStatus(httpStatus int) bool { + return httpStatus == http.StatusRequestTimeout || httpStatus == http.StatusTooManyRequests || httpStatus >= 500 +} + // internalErrorAttrs describes err for the log without its text, which may // quote a URI, header or credential: only what the storage backend describes // of the service's response (see objectstorage.DescribeServiceError), and the diff --git a/internal/objectstoreplugin/status_test.go b/internal/objectstoreplugin/status_test.go index e85c55c289..3551b312e0 100644 --- a/internal/objectstoreplugin/status_test.go +++ b/internal/objectstoreplugin/status_test.go @@ -54,6 +54,12 @@ func TestToStatusLogsUnrecognizedErrors(t *testing.T) { []string{`"level":"WARN"`, `"http_status":412`, `"error_type":"*objectstoreplugin.responseError"`}}, {"other error", fmt.Errorf("while reading gs://b/o?sig=%s: %w", secret, errors.New("boom")), codes.Internal, []string{`"level":"WARN"`, `"error_type":"*errors.errorString"`}}, + {"service unavailable", fmt.Errorf("while writing gs://b/o?sig=%s: %w", secret, &responseError{503, "backend error for gs://b/o " + secret}), codes.Unavailable, + []string{`"level":"INFO"`, `"http_status":503`, `"error_type":"*objectstoreplugin.responseError"`}}, + {"throttled", fmt.Errorf("while writing: %w", &responseError{429, "slow down " + secret}), codes.Unavailable, + []string{`"level":"INFO"`, `"http_status":429`}}, + {"forbidden", fmt.Errorf("while writing: %w", &responseError{403, "denied " + secret}), codes.Internal, + []string{`"level":"WARN"`, `"http_status":403`}}, {"not found", fmt.Errorf("while reading: %w", objectstorage.ErrObjectNotFound), codes.NotFound, nil}, {"canceled", context.Canceled, codes.Canceled, nil}, } { @@ -75,7 +81,7 @@ func TestToStatusLogsUnrecognizedErrors(t *testing.T) { t.Errorf("logs missing %s:\n%s", want, logs) } } - for _, leak := range []string{secret, "gs://b/o", "boom", "precondition failed"} { + for _, leak := range []string{secret, "gs://b/o", "boom", "precondition failed", "backend error", "slow down", "denied"} { if strings.Contains(logs, leak) { t.Errorf("logs carry %q:\n%s", leak, logs) } From ea779def24d24c32ac1c3a4cec4c632a9f2c0a89 Mon Sep 17 00:00:00 2001 From: Amp Date: Tue, 27 Oct 2026 02:35:34 +0000 Subject: [PATCH 3/5] Retry an uploading Checkpoint from the snapshot the node already holds - A Checkpoint under UPLOAD_MODE_IMMEDIATE drove worker's CheckpointWorkload, which deletes the workload, and only then uploaded; when the upload failed, or the control plane lost the response, the retried Checkpoint drove worker again against a workload that was gone and the actor crashed (#372 anticipated this) - The checkpoint now moves worker's snapshot to the node's local snapshot directory and records it as LatestLocal before any upload, in both upload modes; a retried Checkpoint of a snapshot this node already committed skips worker and uploads the local copy, or returns the recorded state when it is durable already - The upload from the local copy goes through the promote path, so an object-store outage reaches the control plane as Unavailable and is retried rather than crashing the actor; a checkpoint that cannot keep a local copy still uploads from the checkpoint directory as before - BeginUploadingCheckpoint holds the actor's background uploads off while an immediate checkpoint uploads, so the same snapshot is never uploaded twice at once - TestSuspendUploadFailureRetriesFromLocalCopy covers a 503 on the first upload and the retry; TestFailedCheckpointKeepsLocalSnapshot now covers the move failing Amp-Thread-ID: https://ampcode.com/threads/T-01a178d8-73f0-751b-8cb4-c5a11dd7ef41 Co-authored-by: Hitesh Sagtani --- .../internal/snapshottracker/lifecycle.go | 31 +++ cmd/atelet/lifecycle_test.go | 144 ++++++++---- cmd/atelet/main.go | 221 +++++++++++------- 3 files changed, 271 insertions(+), 125 deletions(-) diff --git a/cmd/atelet/internal/snapshottracker/lifecycle.go b/cmd/atelet/internal/snapshottracker/lifecycle.go index 0144bcc938..a552004e7c 100644 --- a/cmd/atelet/internal/snapshottracker/lifecycle.go +++ b/cmd/atelet/internal/snapshottracker/lifecycle.go @@ -215,6 +215,37 @@ func (t *Tracker) BeginCheckpoint(ctx context.Context, actorUID string, snap *at return t.holdLocked(actorUID, a, &SnapshotRef{Snapshot: proto.CloneOf(snap)}, false), nil } +// BeginUploadingCheckpoint is BeginCheckpoint for a Checkpoint under +// UPLOAD_MODE_IMMEDIATE, which uploads snap itself: it also waits for a +// background upload of snap in progress, and holds the actor's background +// uploads off until the checkpoint ends, as BeginPromote does, so the same +// snapshot is never uploaded twice at once. +func (t *Tracker) BeginUploadingCheckpoint(ctx context.Context, actorUID string, snap *ateletpb.Snapshot) (end func(), err error) { + t.lock(actorUID) + defer t.unlock(actorUID) + a, err := t.waitForDetachOrSnapshotUploadLocked(ctx, actorUID, SnapshotRef{Snapshot: snap}) + if err != nil { + return nil, err + } + return t.holdLocked(actorUID, a, &SnapshotRef{Snapshot: proto.CloneOf(snap)}, true), nil +} + +// Committed reports whether snap is already recorded as the actor's latest +// local snapshot, and whether as its latest durable one. A retried Checkpoint +// of a snapshot an earlier attempt committed, whose response was lost or +// whose upload failed, finds one or both set and must not drive worker again: +// the workload is gone. +func (t *Tracker) Committed(actorUID string, snap *ateletpb.Snapshot) (local, durable bool) { + t.lock(actorUID) + defer t.unlock(actorUID) + a := t.findLocked(actorUID) + if a == nil { + return false, false + } + ref := SnapshotRef{Snapshot: snap} + return sameUUID(a.LatestLocal, &ref), sameUUID(a.LatestDurable, &ref) +} + // BeginPromote records that a PromoteSnapshot of snap is in progress on this // node, once any detach in progress and any background upload of snap have // finished. It fails with FailedPrecondition when snap has a stale generation diff --git a/cmd/atelet/lifecycle_test.go b/cmd/atelet/lifecycle_test.go index 4df2da5f16..3ba0abc191 100644 --- a/cmd/atelet/lifecycle_test.go +++ b/cmd/atelet/lifecycle_test.go @@ -1267,51 +1267,107 @@ func TestImmediateCheckpointWithoutLocalCopy(t *testing.T) { // recorded as the actor's latest local snapshot, so the actor can still be // restored from it. func TestFailedCheckpointKeepsLocalSnapshot(t *testing.T) { - for _, tc := range []struct { - name string - snap string - mode ateletpb.UploadMode - fail func(t *testing.T, a *pluginActor) - }{ - { - name: "suspend upload fails", - snap: "suspend-snap-2", - mode: ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE, - fail: func(t *testing.T, a *pluginActor) { - a.store.putErr = errors.New("object store unavailable") - }, - }, - { - name: "pause move fails", - snap: "pause-snap-2", - mode: ateletpb.UploadMode_UPLOAD_MODE_DEFERRED, - fail: func(t *testing.T, a *pluginActor) { - // A file where the new snapshot's directory belongs fails the move. - if err := os.WriteFile(ateletpath.LocalSnapshotDir(pluginActorUID, "pause-snap-2"), nil, 0o600); err != nil { - t.Fatal(err) - } - }, - }, - } { - t.Run(tc.name, func(t *testing.T) { - a := runPluginActor(t) - if _, err := a.checkpoint(t, a.snapshot("pause-snap-1", 1), ateletpb.UploadMode_UPLOAD_MODE_DEFERRED); err != nil { - t.Fatalf("Checkpoint (pause): %v", err) - } - a.restoreLocal(t, "pause-snap-1", 1) + a := runPluginActor(t) + if _, err := a.checkpoint(t, a.snapshot("pause-snap-1", 1), ateletpb.UploadMode_UPLOAD_MODE_DEFERRED); err != nil { + t.Fatalf("Checkpoint (pause): %v", err) + } + a.restoreLocal(t, "pause-snap-1", 1) - tc.fail(t, a) - if _, err := a.checkpoint(t, a.snapshot(tc.snap, 2), tc.mode); err == nil { - t.Fatal("Checkpoint succeeded, want it to fail") - } - if _, err := os.Stat(ateletpath.LocalSnapshotDir(pluginActorUID, "pause-snap-1")); err != nil { - t.Errorf("pause snapshot after the failed checkpoint: stat = %v, want it kept", err) - } - if got := a.s.snapshots.State(pluginActorUID).GetLatestLocal().GetSnapshot().GetSnapshotUuid(); got != "pause-snap-1" { - t.Errorf("latest_local = %q, want pause-snap-1", got) - } - a.restoreLocal(t, "pause-snap-1", 1) - }) + // A file where the new snapshot's directory belongs fails the move. + if err := os.WriteFile(ateletpath.LocalSnapshotDir(pluginActorUID, "pause-snap-2"), nil, 0o600); err != nil { + t.Fatal(err) + } + if _, err := a.checkpoint(t, a.snapshot("pause-snap-2", 2), ateletpb.UploadMode_UPLOAD_MODE_DEFERRED); err == nil { + t.Fatal("Checkpoint succeeded, want it to fail") + } + if _, err := os.Stat(ateletpath.LocalSnapshotDir(pluginActorUID, "pause-snap-1")); err != nil { + t.Errorf("pause snapshot after the failed checkpoint: stat = %v, want it kept", err) + } + if got := a.s.snapshots.State(pluginActorUID).GetLatestLocal().GetSnapshot().GetSnapshotUuid(); got != "pause-snap-1" { + t.Errorf("latest_local = %q, want pause-snap-1", got) + } + a.restoreLocal(t, "pause-snap-1", 1) +} + +// transientStoreError is a storage service failure that reports a transient +// HTTP status, as a backend SDK's response error does once its retries are +// exhausted. +type transientStoreError struct{ code int } + +func (e *transientStoreError) Error() string { + return fmt.Sprintf("storage service answered %d", e.code) +} +func (e *transientStoreError) HTTPStatusCode() int { return e.code } + +// TestSuspendUploadFailureRetriesFromLocalCopy suspends a resumed actor while +// the object store is failing over: the checkpoint reports the outage as +// Unavailable, so the control plane retries rather than crashes the actor, and +// keeps the snapshot worker wrote on the node, recorded as the actor's latest +// local snapshot. The retried Checkpoint uploads that copy without driving +// worker again (its workload is gone), and records it durable with the +// checkpoint's commit time. A further retry reports the same state without +// uploading again. +func TestSuspendUploadFailureRetriesFromLocalCopy(t *testing.T) { + a := runPluginActor(t) + if _, err := a.checkpoint(t, a.snapshot("pause-snap-1", 1), ateletpb.UploadMode_UPLOAD_MODE_DEFERRED); err != nil { + t.Fatalf("Checkpoint (pause): %v", err) + } + a.restoreLocal(t, "pause-snap-1", 1) + snap := a.snapshot("suspend-snap-2", 2) + + a.store.putErr = &transientStoreError{503} + _, err := a.checkpoint(t, snap, ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE) + if apierror.Code(err) != codes.Unavailable { + t.Fatalf("Checkpoint during the outage = %v, want %s", err, codes.Unavailable) + } + if a.worker.actorDirs["CheckpointWorkload"] == nil { + t.Fatal("worker.CheckpointWorkload not called by the first Checkpoint") + } + if _, err := os.Stat(ateletpath.LocalSnapshotDir(pluginActorUID, "suspend-snap-2")); err != nil { + t.Errorf("snapshot after the failed upload: stat = %v, want it kept", err) + } + failed := a.s.snapshots.State(pluginActorUID) + if got := failed.GetLatestLocal(); !proto.Equal(got.GetSnapshot(), snap) || got.GetLocalCommitTime() == nil { + t.Errorf("latest_local after the failed upload = %v, want %v with its commit time", got, snap) + } + if failed.GetLatestDurable() != nil { + t.Errorf("latest_durable after the failed upload = %v, want unset", failed.GetLatestDurable()) + } + + a.store.putErr = nil + delete(a.worker.actorDirs, "CheckpointWorkload") + resp, err := a.checkpoint(t, snap, ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE) + if err != nil { + t.Fatalf("retried Checkpoint: %v", err) + } + if got := a.worker.actorDirs["CheckpointWorkload"]; got != nil { + t.Errorf("retried Checkpoint called worker.CheckpointWorkload (%v), want the local copy uploaded instead", got) + } + if !slices.Contains(a.store.keys(), manifestKey("suspend-snap-2")) { + t.Errorf("objects = %v, want the snapshot uploaded", a.store.keys()) + } + if _, err := os.Stat(ateletpath.LocalSnapshotDir(pluginActorUID, "suspend-snap-2")); err != nil { + t.Errorf("local snapshot after the retried checkpoint: stat = %v, want it kept", err) + } + st := resp.GetState() + commitTime := failed.GetLatestLocal().GetLocalCommitTime() + if got := st.GetLatestLocal(); !proto.Equal(got.GetSnapshot(), snap) || !proto.Equal(got.GetLocalCommitTime(), commitTime) { + t.Errorf("latest_local = %v, want %v committed at %v", got, snap, commitTime) + } + if got := st.GetLatestDurable(); !proto.Equal(got.GetSnapshot(), snap) || !proto.Equal(got.GetLocalCommitTime(), commitTime) || got.GetDurableCompleteTime() == nil { + t.Errorf("latest_durable = %v, want %v committed at %v, with its upload time", got, snap, commitTime) + } + if got := a.s.snapshots.State(pluginActorUID); !proto.Equal(got, st) { + t.Errorf("tracked state = %v, want %v, as Checkpoint returned", got, st) + } + + a.store.putErr = errors.New("retried suspend should not upload again") + again, err := a.checkpoint(t, snap, ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE) + if err != nil { + t.Fatalf("Checkpoint retried after the upload: %v", err) + } + if got := again.GetState(); !proto.Equal(got, st) { + t.Errorf("state after a second retry = %v, want %v", got, st) } } diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 88dae0a9b3..74b07eadd2 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -918,9 +918,26 @@ func (s *Atelet) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRequest snapshotLogAttrs(attribution, op, checkpointDurationMetric, err, phases)...) }() + mode := req.GetUploadConfig().GetMode() + immediate := false + switch mode { + case ateletpb.UploadMode_UPLOAD_MODE_DEFERRED: + case ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE: + immediate = true + default: + return nil, fmt.Errorf("unexpected upload mode: %v", mode) + } + // A detach in progress finishes first, and none starts until the - // checkpoint returns, so it cannot delete the snapshot being written. - endCheckpoint, err := s.snapshots.BeginCheckpoint(ctx, actorUID, req.GetSnapshot()) + // checkpoint returns, so it cannot delete the snapshot being written. A + // checkpoint that uploads also waits for a background upload of its + // snapshot, and holds further ones off, so it never uploads alongside one. + var endCheckpoint func() + if immediate { + endCheckpoint, err = s.snapshots.BeginUploadingCheckpoint(ctx, actorUID, req.GetSnapshot()) + } else { + endCheckpoint, err = s.snapshots.BeginCheckpoint(ctx, actorUID, req.GetSnapshot()) + } if err != nil { return nil, err } @@ -936,97 +953,139 @@ func (s *Atelet) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRequest } op.sandboxClass = sandboxRec.SandboxClass - tAssets := time.Now() - assetPaths, err := s.ensureSandboxAssets(ctx, sandboxRec, ateletpath.SandboxAssetsDir(actorUID)) - dAssets = time.Since(tAssets) - if err != nil { - return nil, err - } - checkpointDir := ateletpath.CheckpointStateDir(actorUID) + localDir := ateletpath.LocalSnapshotDir(actorUID, req.GetSnapshot().GetSnapshotUuid()) + snap := snapshottracker.SnapshotRef{Snapshot: req.GetSnapshot(), Actor: actorRef} + + // The control plane retries a Checkpoint whose response it lost, or that + // failed after worker wrote the snapshot: on an upload the object store + // refused, or a cleanup step. CheckpointWorkload deleted the workload, so + // the retry cannot take the checkpoint again; it picks up where the earlier + // attempt left off, from the snapshot that attempt committed on this node. + local, durable := s.snapshots.Committed(actorUID, req.GetSnapshot()) + var state *ateletpb.ActorSnapshotState + // uploadFrom is where the snapshot's files sit for an upload: its local + // copy, or the checkpoint directory when no local copy could be kept. + uploadFrom := localDir + if local || durable { + slog.InfoContext(ctx, "Checkpoint retried for a snapshot this node already committed; not driving worker again", + slog.Any("actor", actorRef), slog.String("snapshot_uuid", req.GetSnapshot().GetSnapshotUuid()), + slog.Bool("local", local), slog.Bool("durable", durable)) + } else { + tAssets := time.Now() + assetPaths, err := s.ensureSandboxAssets(ctx, sandboxRec, ateletpath.SandboxAssetsDir(actorUID)) + dAssets = time.Since(tAssets) + if err != nil { + return nil, err + } - client, err := s.dialWorker(ctx, req.GetWorkerPodUid()) - if err != nil { - return nil, err - } - - // Tell worker to take the checkpoint and delete containers. worker reports the - // exact files it wrote so we ship precisely that set (gVisor's image files, - // cloud-hypervisor's snapshot set, ...) rather than a hardcoded list. - spec, err := buildWorkerWorkloadSpec(req.GetSpec(), nil) - if err != nil { - return nil, apierror.InvalidArgument("invalid workload spec: %v", err) - } + client, err := s.dialWorker(ctx, req.GetWorkerPodUid()) + if err != nil { + return nil, err + } - tWorker := time.Now() - resp, err := client.CheckpointWorkload(ctx, &ateworkerpb.CheckpointWorkloadRequest{ - Atespace: actorRef.Atespace, - ActorName: actorRef.Name, - ActorTemplateAtespace: req.GetActorTemplateAtespace(), - ActorTemplateName: req.GetActorTemplateName(), - RunscPath: runscPathFor(assetPaths), - Spec: spec, - Fidelity: toWorkerFidelity(req.GetSnapshot().GetFidelity()), - ActorUid: actorUID, - ActorDirs: ateletpath.ActorDirs(actorUID), - }) - dWorker = time.Since(tWorker) - if err != nil { - // TODO: Worker should classify checkpoint failures, and set "should-crash" - // in the metadata if the error is not retriable. - return nil, fmt.Errorf("while calling worker CheckpointWorkload: %w", err) - } - // The snapshot commits on this node once worker has written it, in either - // upload mode. - snap := snapshottracker.SnapshotRef{Snapshot: req.GetSnapshot(), CommitTime: time.Now(), Actor: actorRef} + // Tell worker to take the checkpoint and delete containers. worker reports the + // exact files it wrote so we ship precisely that set (gVisor's image files, + // cloud-hypervisor's snapshot set, ...) rather than a hardcoded list. + spec, err := buildWorkerWorkloadSpec(req.GetSpec(), nil) + if err != nil { + return nil, apierror.InvalidArgument("invalid workload spec: %v", err) + } - s.systemInfoVolumes.Deregister(actorUID) + tWorker := time.Now() + resp, err := client.CheckpointWorkload(ctx, &ateworkerpb.CheckpointWorkloadRequest{ + Atespace: actorRef.Atespace, + ActorName: actorRef.Name, + ActorTemplateAtespace: req.GetActorTemplateAtespace(), + ActorTemplateName: req.GetActorTemplateName(), + RunscPath: runscPathFor(assetPaths), + Spec: spec, + Fidelity: toWorkerFidelity(req.GetSnapshot().GetFidelity()), + ActorUid: actorUID, + ActorDirs: ateletpath.ActorDirs(actorUID), + }) + dWorker = time.Since(tWorker) + if err != nil { + // TODO: Worker should classify checkpoint failures, and set "should-crash" + // in the metadata if the error is not retriable. + return nil, fmt.Errorf("while calling worker CheckpointWorkload: %w", err) + } + // The snapshot commits on this node once worker has written it, in either + // upload mode. + snap.CommitTime = time.Now() - sandboxRec.SnapshotFiles, sandboxRec.DataSnapshotFiles, err = checkpointSnapshotFiles(resp, shouldHaveSnapshots(req)) - if err != nil { - return nil, err - } - sandboxRec.Atespace = req.GetAtespace() - sandboxRec.ActorName = req.GetActorName() - sandboxRec.ActorUID = req.GetActorUid() - sandboxRec.ActorTemplateAtespace = req.GetActorTemplateAtespace() - sandboxRec.ActorTemplateName = req.GetActorTemplateName() - sandboxRec.Fidelity = ateattr.SnapshotFidelityValue(req.GetSnapshot().GetFidelity()) + s.systemInfoVolumes.Deregister(actorUID) - tPersist := time.Now() - var state *ateletpb.ActorSnapshotState - switch mode := req.GetUploadConfig().GetMode(); mode { - case ateletpb.UploadMode_UPLOAD_MODE_DEFERRED: - if err := s.moveLocalCheckpoint(ctx, req, sandboxRec); err != nil { - dPersist = time.Since(tPersist) - return nil, fmt.Errorf("while moving to local snapshot: %w", err) + sandboxRec.SnapshotFiles, sandboxRec.DataSnapshotFiles, err = checkpointSnapshotFiles(resp, shouldHaveSnapshots(req)) + if err != nil { + return nil, err } - state = s.snapshots.CommitLocal(actorUID, snap) - case ateletpb.UploadMode_UPLOAD_MODE_IMMEDIATE: - // CheckpointWorkload has already deleted the workload, so a retried - // Checkpoint cannot redo this upload. It waits for the object-store plugin - // as long as ctx allows, and a plugin error reaches the control plane as - // Internal, which crashes the actor. - // TODO(#362): Because we do not cache the external snapshot files when upload fails, we have to mark the Actor as CRASHED. - if err := s.uploadExternalCheckpoint(ctx, req, checkpointDir, sandboxRec); err != nil { - dPersist = time.Since(tPersist) - return nil, fmt.Errorf("while uploading external snapshot: %w", err) - } - durable := snap - durable.UploadTime = time.Now() - state = s.snapshots.CommitDurable(actorUID, durable) - // The snapshot is durable now, so the local copy only saves a - // download: failing to keep it must not fail the checkpoint. The prune - // below deletes a partial copy. + sandboxRec.Atespace = req.GetAtespace() + sandboxRec.ActorName = req.GetActorName() + sandboxRec.ActorUID = req.GetActorUid() + sandboxRec.ActorTemplateAtespace = req.GetActorTemplateAtespace() + sandboxRec.ActorTemplateName = req.GetActorTemplateName() + sandboxRec.Fidelity = ateattr.SnapshotFidelityValue(req.GetSnapshot().GetFidelity()) + + // Keep the snapshot on this node and record it before any upload, so a + // Checkpoint that fails from here on can be retried from the local copy. + tMove := time.Now() if err := s.moveLocalCheckpoint(ctx, req, sandboxRec); err != nil { - slog.WarnContext(ctx, "Failed to keep a local copy of the uploaded snapshot", slog.Any("actor", actorRef), slog.Any("err", err)) + dPersist += time.Since(tMove) + if !immediate { + return nil, fmt.Errorf("while moving to local snapshot: %w", err) + } + // An immediate checkpoint can still succeed without a local copy, by + // uploading what worker wrote where it wrote it. That upload is the + // one a retry cannot redo. The prune below deletes a partial copy. + slog.WarnContext(ctx, "Failed to keep a local copy of the snapshot; uploading it from the checkpoint directory", slog.Any("actor", actorRef), slog.Any("err", err)) + uploadFrom = checkpointDir } else { state = s.snapshots.CommitLocal(actorUID, snap) } - default: - return nil, fmt.Errorf("unexpected upload mode: %v", mode) + dPersist += time.Since(tMove) + } + + if immediate && !durable { + tUpload := time.Now() + if uploadFrom == localDir { + // The local copy stays until the upload succeeds, and stays + // recorded as LatestLocal, so a retry uploads it again. Plugin errors + // are reported as uploadLocalCheckpointDir reports them: a plugin + // outage as Unavailable, which the control plane retries. + promoteReq := &ateletpb.PromoteSnapshotRequest{ + Atespace: req.GetAtespace(), ActorName: req.GetActorName(), ActorUid: actorUID, + ActorTemplateAtespace: req.GetActorTemplateAtespace(), ActorTemplateName: req.GetActorTemplateName(), + Snapshot: req.GetSnapshot(), + } + uri, err := resources.ParseSnapshotURI(req.GetSnapshot().GetObject().GetSnapshotUri()) + if err != nil { + return nil, apierror.InvalidArgument("%v", err) + } + _, _, err = s.uploadLocalCheckpointDir(ctx, promoteReq, localDir, uri) + dPersist += time.Since(tUpload) + if err != nil { + return nil, fmt.Errorf("while uploading external snapshot from its local copy: %w", err) + } + snap.UploadTime = time.Now() + state = s.snapshots.CommitPromoted(actorUID, snap) + } else { + // No local copy: this upload waits for the object-store plugin as + // long as ctx allows, and a plugin error reaches the control plane + // as Internal, which crashes the actor, since nothing is left on + // the node to retry from. + err := s.uploadExternalCheckpoint(ctx, req, checkpointDir, sandboxRec) + dPersist += time.Since(tUpload) + if err != nil { + return nil, fmt.Errorf("while uploading external snapshot: %w", err) + } + snap.UploadTime = time.Now() + state = s.snapshots.CommitDurable(actorUID, snap) + } + } + if state == nil { + state = s.snapshots.State(actorUID) } - dPersist = time.Since(tPersist) // Delete the actor's earlier local snapshots only now that this checkpoint // has committed its own. A checkpoint that fails before then leaves the From eb1e0cf1dcb58b968682ad53824adc307d72837e Mon Sep 17 00:00:00 2001 From: Amp Date: Tue, 27 Oct 2026 02:35:34 +0000 Subject: [PATCH 4/5] Give ate-api-server the same shutdown drain budget as atelet - ate-api-server drained for 15 s on SIGTERM, then force-stopped in-flight RPCs; a SuspendActor driving an atelet Checkpoint that streams a multi-GiB snapshot was cut mid-upload during the 2026-10-26 rollout, and the client's retry sent a Checkpoint for a workload worker had already deleted - --drain-timeout is now 5m, atelet's, with terminationGracePeriodSeconds 330 to cover --drain-delay plus the force stop - A Deployment rollout counts a terminating pod as gone, so the longer drain does not hold up the rollout Amp-Thread-ID: https://ampcode.com/threads/T-01a178d8-73f0-751b-8cb4-c5a11dd7ef41 Co-authored-by: Hitesh Sagtani --- manifests/ate-install/ate-api-server.yaml | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/manifests/ate-install/ate-api-server.yaml b/manifests/ate-install/ate-api-server.yaml index 93b26943cb..f4e91d0f76 100644 --- a/manifests/ate-install/ate-api-server.yaml +++ b/manifests/ate-install/ate-api-server.yaml @@ -86,8 +86,12 @@ spec: serviceAccountName: ate-api-server # Budget for the full shutdown sequence: --drain-delay (sized to cover # the readinessProbe below reaching NotReady, plus propagation) + - # --drain-timeout. - terminationGracePeriodSeconds: 40 + # --drain-timeout (5m, atelet's). An in-flight SuspendActor drives + # atelet's Checkpoint, which streams a multi-GiB snapshot to the object + # store; force-stopping it mid-stream makes the client retry a Checkpoint + # whose workload is already gone. The sum must fit within + # terminationGracePeriodSeconds, plus slack for the force Stop(). + terminationGracePeriodSeconds: 330 # The control object-store plugin deletes and copies external snapshots # for ate-api-server, over a Unix socket in a pod-local emptyDir. A native # sidecar starts before ate-api-server and stops after it. ate-api-server @@ -154,7 +158,7 @@ spec: - --default-egress-gateway-address=atenet-egress.ate-system.svc:443 # Graceful shutdown knobs. The sum must fit within terminationGracePeriodSeconds. - --drain-delay=13s - - --drain-timeout=15s + - --drain-timeout=5m - --objectstore-plugin-socket=/run/objectstore-plugin/control.sock env: - name: POD_NAME From 5866fbe65a2ffb75a9b200f7ba52d991b94d68ec Mon Sep 17 00:00:00 2001 From: Amp Date: Tue, 27 Oct 2026 02:40:18 +0000 Subject: [PATCH 5/5] Bound the OTel provider flush at process exit A process past "Shutdown complete" flushed traces and metrics with a background context, so an unreachable collector held the pod until the kubelet's grace period. Give each provider 15s. Amp-Thread-ID: https://ampcode.com/threads/T-01a178d8-73f0-751b-8cb4-c5a11dd7ef41 Co-authored-by: Hitesh Sagtani --- internal/serverboot/serverboot.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/internal/serverboot/serverboot.go b/internal/serverboot/serverboot.go index f261760e4d..61bc93c638 100644 --- a/internal/serverboot/serverboot.go +++ b/internal/serverboot/serverboot.go @@ -427,12 +427,19 @@ func Fatal(ctx context.Context, msg string, err error, additionalArgs ...any) { os.Exit(1) } +// providerShutdownTimeout bounds one OTel provider's final flush at exit. A +// collector that is unreachable while the process is already past "Shutdown +// complete" must not hold the pod until the kubelet's grace period kills it. +const providerShutdownTimeout = 15 * time.Second + // ShutdownProvider invokes the OTel provider's Shutdown and logs any // error. Designed to be deferred from main(): // // defer serverboot.ShutdownProvider("TracerProvider", tp.Shutdown) func ShutdownProvider(name string, shutdown func(context.Context) error) { - if err := shutdown(context.Background()); err != nil { + ctx, cancel := context.WithTimeout(context.Background(), providerShutdownTimeout) + defer cancel() + if err := shutdown(ctx); err != nil { slog.Error("Failed to shutdown "+name, slog.Any("err", err)) } }