From ccd6750570c4c9f51751fb08fb31625524e8dec1 Mon Sep 17 00:00:00 2001 From: Abhay Joshi Date: Thu, 22 Oct 2026 08:01:24 +0000 Subject: [PATCH] ateapi: skip deleting external volumes when Terminate or volume detach fails DeleteActor called ensureVolumesDeleted even when ensureAteletTerminated or ensureVolumesDetachedForDelete returned an error. When a CSI driver deletes a still-published volume and returns NotFound from NodeUnpublishVolume on the retried Terminate, atelet skips unmounting and resetActorDirs fails with EBUSY on the mounted volume directory. Skip ensureVolumesDeleted whenever ensureAteletTerminated or ensureVolumesDetachedForDelete fails, matching ensureWorkerReleased, so external volumes are deleted only after node termination and volume detachment succeed. Signed-off-by: Abhay Joshi --- .../internal/controlapi/workflow_delete.go | 11 ++- .../controlapi/workflow_delete_test.go | 80 ++++++++++++++++++- 2 files changed, 81 insertions(+), 10 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/workflow_delete.go b/cmd/ateapi/internal/controlapi/workflow_delete.go index 3242503daf..b7f53506a2 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete.go @@ -70,22 +70,21 @@ func (w *ActorWorkflow) DeleteActor(ctx context.Context, actorRef resources.Acto errs = append(errs, volumesDetachedErr) } - // Release worker if atelet termination and volume detachment did not fail. + // Release worker and delete volumes if atelet termination and volume detachment did not fail. if atletTerminatedErr == nil && volumesDetachedErr == nil { if actor, err = w.ensureWorkerReleased(ctx, actorRef, actor); err != nil { errs = append(errs, fmt.Errorf("while releasing worker: %w", err)) } + if err := w.ensureVolumesDeleted(ctx, actor); err != nil { + errs = append(errs, fmt.Errorf("while deleting volumes: %w", err)) + } } else { - slog.WarnContext(ctx, "skipping releasing worker due to atelet termination or volume detachment failure", + slog.WarnContext(ctx, "skipping releasing worker and deleting volumes due to atelet termination or volume detachment failure", slog.Any("actor", actorRef), slog.Any("atletTerminatedErr", atletTerminatedErr), slog.Any("volumesDetachedErr", volumesDetachedErr)) } - if err := w.ensureVolumesDeleted(ctx, actor); err != nil { - errs = append(errs, fmt.Errorf("while deleting volumes: %w", err)) - } - if err := w.ensureExternalSnapshotsReleased(ctx, actor); err != nil { errs = append(errs, fmt.Errorf("while releasing external snapshots: %w", err)) } diff --git a/cmd/ateapi/internal/controlapi/workflow_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_delete_test.go index 9dcfba5258..d3443bae7c 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete_test.go @@ -139,10 +139,16 @@ func TestEnsureAteletTerminated_NodeGone(t *testing.T) { } // detachDeletePlugin is a volume plugin for a whole delete: it records -// detaches like mockDetachVolumePlugin and accepts the volume delete. -type detachDeletePlugin struct{ mockDetachVolumePlugin } +// detaches like mockDetachVolumePlugin and records volume deletes. +type detachDeletePlugin struct { + mockDetachVolumePlugin + deleteCalls []string +} -func (*detachDeletePlugin) DeleteVolume(context.Context, string) error { return nil } +func (p *detachDeletePlugin) DeleteVolume(_ context.Context, volumeID string) error { + p.deleteCalls = append(p.deleteCalls, volumeID) + return nil +} // TestRevertAndDelete_AssignedNodeGone runs the whole revert and delete // workflows for a paused actor with a PENDING upload and a crashed actor whose @@ -185,7 +191,7 @@ func TestRevertAndDelete_AssignedNodeGone(t *testing.T) { } w.dialer = NewAteletDialer(newTestAteletIndexer(t), installdefaults.AteletSPIFFEID(installdefaults.SystemNamespace), "", "", WithNodeGetter(nodeGetter(nil, nodes...))) - plugin := &detachDeletePlugin{mockDetachVolumePlugin{detachErrs: map[string]error{"storage-vol-1": tc.detachErr}}} + plugin := &detachDeletePlugin{mockDetachVolumePlugin: mockDetachVolumePlugin{detachErrs: map[string]error{"storage-vol-1": tc.detachErr}}} w.pluginRegistry = &mockPluginRegistry{plugins: map[string]volume.VolumePluginControlPlane{"mock": plugin}} tmpl := &ateapi.ActorTemplate{ Metadata: &ateapi.ResourceMetadata{Atespace: "ns", Name: "tmpl-vol"}, @@ -243,6 +249,13 @@ func TestRevertAndDelete_AssignedNodeGone(t *testing.T) { if want := []detachCall{{VolumeID: "storage-vol-1", Node: "node-1"}}; !slices.Equal(plugin.detachCalls, want) { t.Errorf("detach calls = %v, want %v", plugin.detachCalls, want) } + var wantDeletes []string + if verb == "delete" && tc.wantErr == nil && !tc.wantDetachErr { + wantDeletes = []string{"storage-vol-1"} + } + if !slices.Equal(plugin.deleteCalls, wantDeletes) { + t.Errorf("delete calls = %v, want %v", plugin.deleteCalls, wantDeletes) + } } got, getErr := st.GetActor(ctx, actorRef) @@ -750,3 +763,62 @@ func TestEnsureWorkerReleased_RecreatedActorKeepsAssignment(t *testing.T) { t.Errorf("status.worker_assignment names %q, want %q", got, wantWorker) } } + +func TestDeleteActor_FailedTerminateSkipsVolumeDeleteUntilRetry(t *testing.T) { + ctx := context.Background() + st, cleanup := storetest.SetupTestStore(t) + t.Cleanup(cleanup) + + w := newTestActorWorkflow(t, st, "ns", "tmpl1") + atelet := &capturingTerminator{err: apierror.Unavailable("atelet unavailable")} + w.dialer = newNodeAteletDialer(t, atelet) + plugin := &detachDeletePlugin{} + w.pluginRegistry = &mockPluginRegistry{plugins: map[string]volume.VolumePluginControlPlane{"mock": plugin}} + if _, err := st.CreateActorTemplate(ctx, &ateapi.ActorTemplate{ + Metadata: &ateapi.ResourceMetadata{Atespace: "ns", Name: "tmpl-vol"}, + SnapshotConfig: &ateapi.SnapshotConfig{StorageLocation: "gs://snapshots"}, + Volumes: []*ateapi.Volume{{Name: "vol1"}}, + Containers: []*ateapi.Container{{VolumeMounts: []*ateapi.VolumeMount{{Name: "vol1"}}}}, + }); err != nil { + t.Fatalf("CreateActorTemplate: %v", err) + } + + actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} + seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl-vol", ateapi.ActorState_ACTOR_STATE_RUNNING, func(a *ateapi.Actor) { + a.Status.AssignedNode = "node-1" + a.Status.WorkerAssignment = wireTestAssignment() + a.Status.ExternalVolumes = &ateapi.ExternalVolumes{Volumes: []*ateapi.ExternalVolumeStatus{ + {Name: "vol1", StorageVolumeId: "storage-vol-1", VolumeType: "mock"}, + }} + }) + actor := mustReadActor(t, ctx, st, actorRef) + seedWireWorker(t, ctx, st, assignmentOf(actor)) + + if _, err := w.DeleteActor(ctx, actorRef, store.DeletePreconditions{}); apierror.Code(err) != codes.Unavailable { + t.Fatalf("first DeleteActor = %v, want Unavailable", err) + } + if len(plugin.deleteCalls) != 0 { + t.Fatalf("delete calls after failed Terminate = %v, want none", plugin.deleteCalls) + } + afterFail := mustReadActor(t, ctx, st, actorRef) + if got := afterFail.GetStatus().GetState(); got != ateapi.ActorState_ACTOR_STATE_DELETING { + t.Errorf("state after failed Terminate = %v, want DELETING", got) + } + if afterFail.GetStatus().GetWorkerAssignment() == nil { + t.Errorf("worker assignment was cleared despite failed Terminate") + } + + atelet.mu.Lock() + atelet.err = nil + atelet.mu.Unlock() + + if _, err := w.DeleteActor(ctx, actorRef, store.DeletePreconditions{}); err != nil { + t.Fatalf("retried DeleteActor: %v", err) + } + if want := []string{"storage-vol-1"}; !slices.Equal(plugin.deleteCalls, want) { + t.Errorf("delete calls after retry = %v, want %v", plugin.deleteCalls, want) + } + if _, err := st.GetActor(ctx, actorRef); !errors.Is(err, store.ErrNotFound) { + t.Errorf("GetActor after retry = %v, want ErrNotFound", err) + } +}