From 7878a0f8ebded5631661cd2a752133c12b0a3b2a Mon Sep 17 00:00:00 2001 From: msau42 Date: Fri, 24 Jul 2026 02:50:42 +0000 Subject: [PATCH] Handle partial creation failures during CreateActor. 1. Adds a new CREATING status that gets persisted before we start to create volumes. Only when all volumes are successfully created the Actor status transitions to SUSPENDED. Currently CreateActor is not idempotent. If volume creation fails, the actor has to be deleted, which will also delete any partially created volumes. 2. Changes volume id to use actor uid instead of atespace+actorname, and add "substrate" prefix 3. Adds a new DELETING status that gets persisted before we start to delete volumes. This shifts the existing DB DELETE precondition checks to the new DELETING status. And now you can only delete the actor in DB from DELETING status. In the future, we will also add a workflow that will do mandatory node cleanup of volumes before starting to delete the volumes. --- .../internal/controlapi/create_actor.go | 40 ++- .../internal/controlapi/create_actor_test.go | 4 +- .../internal/controlapi/delete_actor.go | 28 +- .../internal/controlapi/delete_actor_test.go | 168 ++++++++++ .../internal/controlapi/functional_test.go | 305 +++++++++++++++++- cmd/ateapi/internal/controlapi/service.go | 2 + cmd/ateapi/internal/controlapi/volumes.go | 121 +++++-- .../internal/controlapi/volumes_test.go | 303 +++++++++++++++++ .../internal/controlapi/workload_spec.go | 3 +- .../internal/controlapi/workload_spec_test.go | 2 +- .../internal/store/ateredis/ateredis.go | 3 +- .../internal/store/ateredis/ateredis_test.go | 8 +- pkg/proto/ateapipb/ateapi.pb.go | 69 ++-- pkg/proto/ateapipb/ateapi.proto | 21 +- 14 files changed, 979 insertions(+), 98 deletions(-) create mode 100644 cmd/ateapi/internal/controlapi/volumes_test.go diff --git a/cmd/ateapi/internal/controlapi/create_actor.go b/cmd/ateapi/internal/controlapi/create_actor.go index d022b70d8..f4db7c651 100644 --- a/cmd/ateapi/internal/controlapi/create_actor.go +++ b/cmd/ateapi/internal/controlapi/create_actor.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "log/slog" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/internal/resources" @@ -59,39 +60,48 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ return nil, status.Errorf(codes.FailedPrecondition, "Atespace %s not found", atespace) } - actorRef := &ateapipb.ObjectRef{ - Atespace: atespace, - Name: name, - } - - volumes, err := s.createActorVolumes(ctx, actorRef, template) - if err != nil { - return nil, err - } + // TODO: When CreateActor becomes idempotent, initialActorVolumes() needs to be updated to + // also check for any existing volumes and return those instead of creating new ones. + initVols := initialActorVolumes(template) + // Persist the actor in CREATING state before provisioning volumes. actor := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{ Atespace: atespace, Name: name, }, - Status: ateapipb.Actor_STATUS_SUSPENDED, + Status: ateapipb.Actor_STATUS_CREATING, ActorTemplateNamespace: templateNamespace, ActorTemplateName: templateName, WorkerSelector: in.GetWorkerSelector(), - ActorVolumes: volumes, + ActorVolumes: initVols, } stored, err := s.persistence.CreateActor(ctx, actor) if err != nil { - // Cleanup created volumes if DB write fails - _ = s.deleteActorVolumes(ctx, actorRef, volumes) if errors.Is(err, store.ErrAlreadyExists) { return nil, status.Errorf(codes.AlreadyExists, "Actor %s already exists", name) } return nil, fmt.Errorf("while recording actor: %w", err) } - setSpanActorAttributes(ctx, stored) - return stored, nil + volumes, volErr := s.createActorVolumes(ctx, stored.GetMetadata().GetUid(), template, stored.GetActorVolumes()) + stored.ActorVolumes = volumes + if volErr != nil { + // Even if volume creation failed, we still want to persist any updated volume state. + if _, updateErr := s.persistence.UpdateActor(ctx, stored, stored.GetMetadata().GetVersion()); updateErr != nil { + slog.ErrorContext(ctx, "failed to update actor volumes on volume creation failure", slog.Any("error", updateErr)) + } + return nil, volErr + } + + stored.Status = ateapipb.Actor_STATUS_SUSPENDED + updated, err := s.persistence.UpdateActor(ctx, stored, stored.GetMetadata().GetVersion()) + if err != nil { + return nil, fmt.Errorf("while updating actor: %w", err) + } + + setSpanActorAttributes(ctx, updated) + return updated, nil } func validateCreateActorRequest(req *ateapipb.CreateActorRequest) error { diff --git a/cmd/ateapi/internal/controlapi/create_actor_test.go b/cmd/ateapi/internal/controlapi/create_actor_test.go index 71f9f638d..83a83413f 100644 --- a/cmd/ateapi/internal/controlapi/create_actor_test.go +++ b/cmd/ateapi/internal/controlapi/create_actor_test.go @@ -53,7 +53,7 @@ func TestCreateActor_StampsFullSpanIdentity(t *testing.T) { if v, ok := attrs[ateattr.ActorUIDKey]; !ok || v.Type() != attribute.STRING || v.AsString() == "" { t.Errorf("%s = %v, want non-empty server-assigned uid", ateattr.ActorUIDKey, v.Emit()) } - if v, ok := attrs[ateattr.ActorVersionKey]; !ok || v.Type() != attribute.INT64 || v.AsInt64() != 1 { - t.Errorf("%s = %v, want int64 1", ateattr.ActorVersionKey, v.Emit()) + if v, ok := attrs[ateattr.ActorVersionKey]; !ok || v.Type() != attribute.INT64 || v.AsInt64() != 2 { + t.Errorf("%s = %v, want int64 2", ateattr.ActorVersionKey, v.Emit()) } } diff --git a/cmd/ateapi/internal/controlapi/delete_actor.go b/cmd/ateapi/internal/controlapi/delete_actor.go index df8fac6e3..2655c80fa 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor.go +++ b/cmd/ateapi/internal/controlapi/delete_actor.go @@ -44,8 +44,30 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ return nil, fmt.Errorf("while fetching actor: %w", err) } + if actor.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED && + actor.GetStatus() != ateapipb.Actor_STATUS_CRASHED && + actor.GetStatus() != ateapipb.Actor_STATUS_CREATING && + actor.GetStatus() != ateapipb.Actor_STATUS_DELETING { + return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status (status: %v)", name, actor.GetStatus()) + } + + if actor.GetStatus() != ateapipb.Actor_STATUS_DELETING { + actor.Status = ateapipb.Actor_STATUS_DELETING + for _, vol := range actor.GetActorVolumes() { + vol.Status = ateapipb.ExternalVolume_DELETING + } + updated, err := s.persistence.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion()) + if err != nil { + if errors.Is(err, store.ErrPersistenceRetry) { + return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry") + } + return nil, fmt.Errorf("while setting actor status to DELETING: %w", err) + } + actor = updated + } + // Delete associated volumes - if err := s.deleteActorVolumes(ctx, req.GetActor(), actor.GetActorVolumes()); err != nil { + if err := s.deleteActorVolumes(ctx, actor.GetMetadata().GetUid(), actor.GetActorVolumes()); err != nil { return nil, status.Errorf(codes.Internal, "while deleting actor volumes: %v", err) } @@ -57,9 +79,9 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ if errors.Is(err, store.ErrFailedPrecondition) { current, getErr := s.persistence.GetActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName()) if getErr == nil { - return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended (status: %v)", req.GetActor().GetName(), current.GetStatus()) + return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status (status: %v)", req.GetActor().GetName(), current.GetStatus()) } - return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended", req.GetActor().GetName()) + return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status", req.GetActor().GetName()) } if errors.Is(err, store.ErrPersistenceRetry) { return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry") diff --git a/cmd/ateapi/internal/controlapi/delete_actor_test.go b/cmd/ateapi/internal/controlapi/delete_actor_test.go index 5ba206c92..d648c194a 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor_test.go +++ b/cmd/ateapi/internal/controlapi/delete_actor_test.go @@ -16,9 +16,14 @@ package controlapi import ( "context" + "fmt" + "strings" "testing" + "github.com/google/go-cmp/cmp" + "github.com/agent-substrate/substrate/internal/ateattr" + "github.com/agent-substrate/substrate/internal/volume" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" ) @@ -50,3 +55,166 @@ func TestDeleteActor_StampsRefSpanIdentity(t *testing.T) { assertSpanStr(t, attrs, ateattr.AtespaceKey, testAtespace) assertSpanStr(t, attrs, ateattr.ActorNameKey, testActorID) } + +func TestDeleteActor_StatusCreating(t *testing.T) { + ns := namespaceForTest("ns-delete-creating") + tc := setupTest(t, ns) + defer tc.cleanup() + createTemplate(t, tc, ns) + + plugin := volume.NewMockVolumePlugin() + tc.service.volumePlugin = plugin + defer func() { tc.service.volumePlugin = nil }() + + creatingActor := &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{ + Atespace: testAtespace, + Name: "creating-actor", + }, + Status: ateapipb.Actor_STATUS_CREATING, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + } + createdActor, err := tc.persistence.CreateActor(context.Background(), creatingActor) + if err != nil { + t.Fatalf("CreateActor: %v", err) + } + + volID, err := plugin.CreateVolume(context.Background(), createdActor.GetMetadata().GetUid()+"-vol1", "10Gi", "standard") + if err != nil { + t.Fatalf("CreateVolume: %v", err) + } + + createdActor.ActorVolumes = []*ateapipb.ExternalVolume{ + {VolumeName: "vol1", StorageVolumeId: volID, Status: ateapipb.ExternalVolume_CREATED}, + } + if _, err := tc.persistence.UpdateActor(context.Background(), createdActor, createdActor.GetMetadata().GetVersion()); err != nil { + t.Fatalf("UpdateActor: %v", err) + } + + deleted, err := tc.service.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "creating-actor"}, + }) + if err != nil { + t.Fatalf("DeleteActor on STATUS_CREATING actor failed: %v", err) + } + if len(deleted.GetActorVolumes()) != 1 || deleted.GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_DELETING { + t.Errorf("expected deleted actor volume status to be DELETING, got %v", deleted.GetActorVolumes()) + } + + if _, err := tc.persistence.GetActor(context.Background(), testAtespace, "creating-actor"); err == nil { + t.Errorf("expected actor to be deleted, but it still exists") + } +} + +func TestDeleteActor_StatusDeleting(t *testing.T) { + ns := namespaceForTest("ns-delete-deleting") + tc := setupTest(t, ns) + defer tc.cleanup() + createTemplate(t, tc, ns) + + deletingActor := &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{ + Atespace: testAtespace, + Name: "deleting-actor", + }, + Status: ateapipb.Actor_STATUS_DELETING, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + } + if _, err := tc.persistence.CreateActor(context.Background(), deletingActor); err != nil { + t.Fatalf("CreateActor: %v", err) + } + + if _, err := tc.service.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "deleting-actor"}, + }); err != nil { + t.Fatalf("DeleteActor on STATUS_DELETING actor failed: %v", err) + } + + if _, err := tc.persistence.GetActor(context.Background(), testAtespace, "deleting-actor"); err == nil { + t.Errorf("expected actor to be deleted, but it still exists") + } +} + +func TestDeleteActor_WrongStatus(t *testing.T) { + ns := namespaceForTest("ns-delete-wrong-status") + tc := setupTest(t, ns) + defer tc.cleanup() + createTemplate(t, tc, ns) + + runningActor := &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{ + Atespace: testAtespace, + Name: "running-actor", + }, + Status: ateapipb.Actor_STATUS_RUNNING, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + } + if _, err := tc.persistence.CreateActor(context.Background(), runningActor); err != nil { + t.Fatalf("CreateActor: %v", err) + } + + _, err := tc.service.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "running-actor"}, + }) + if err == nil { + t.Fatalf("expected DeleteActor on STATUS_RUNNING actor to fail, but it succeeded") + } +} + +type failingVolumePlugin struct { + volume.VolumePluginControlPlane + deletedIDs []string +} + +func (f *failingVolumePlugin) DeleteVolume(ctx context.Context, volumeID string) error { + f.deletedIDs = append(f.deletedIDs, volumeID) + return fmt.Errorf("simulated delete error for %s", volumeID) +} + +func TestDeleteActor_MultipleVolumeDeletionFailures(t *testing.T) { + ns := namespaceForTest("ns-delete-multivol-fail") + tc := setupTest(t, ns) + defer tc.cleanup() + createTemplate(t, tc, ns) + + plugin := &failingVolumePlugin{} + tc.service.volumePlugin = plugin + defer func() { tc.service.volumePlugin = nil }() + + actor := &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{ + Atespace: testAtespace, + Name: "multi-vol-actor", + }, + Status: ateapipb.Actor_STATUS_SUSPENDED, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + ActorVolumes: []*ateapipb.ExternalVolume{ + {VolumeName: "vol1", StorageVolumeId: "storage-vol-1", Status: ateapipb.ExternalVolume_CREATED}, + {VolumeName: "vol2", StorageVolumeId: "storage-vol-2", Status: ateapipb.ExternalVolume_CREATED}, + }, + } + if _, err := tc.persistence.CreateActor(context.Background(), actor); err != nil { + t.Fatalf("CreateActor: %v", err) + } + + _, err := tc.service.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "multi-vol-actor"}, + }) + if err == nil { + t.Fatalf("expected DeleteActor to fail when volume deletion fails, but it succeeded") + } + + wantDeleted := []string{"storage-vol-1", "storage-vol-2"} + if diff := cmp.Diff(wantDeleted, plugin.deletedIDs); diff != "" { + t.Errorf("deletedIDs mismatch (-want +got):\n%s", diff) + } + + errMsg := err.Error() + if !strings.Contains(errMsg, "storage-vol-1") || !strings.Contains(errMsg, "storage-vol-2") { + t.Errorf("expected error message to contain both volume failure details, got: %v", errMsg) + } +} diff --git a/cmd/ateapi/internal/controlapi/functional_test.go b/cmd/ateapi/internal/controlapi/functional_test.go index 55429e6aa..c973e6836 100644 --- a/cmd/ateapi/internal/controlapi/functional_test.go +++ b/cmd/ateapi/internal/controlapi/functional_test.go @@ -31,6 +31,7 @@ import ( "github.com/agent-substrate/substrate/internal/ateinterceptors" "github.com/agent-substrate/substrate/internal/envtestbins" "github.com/agent-substrate/substrate/internal/proto/ateletpb" + "github.com/agent-substrate/substrate/internal/volume" atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/client/clientset/versioned" "github.com/agent-substrate/substrate/pkg/client/informers/externalversions" @@ -49,6 +50,7 @@ import ( "google.golang.org/protobuf/types/known/timestamppb" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes" @@ -423,6 +425,21 @@ func createAtespace(t *testing.T, tc *testContext, name string) { const poolLabelKey = "pool" func createTemplateWithContainers(t *testing.T, tc *testContext, ns string, containers []atev1alpha1.Container) { + createTemplateWithContainersAndVolumes(t, tc, ns, containers, nil) +} + +func createTemplateWithVolumes(t *testing.T, tc *testContext, ns string, volumes []atev1alpha1.Volume, mounts []atev1alpha1.VolumeMount) { + createTemplateWithContainersAndVolumes(t, tc, ns, []atev1alpha1.Container{ + { + Name: "main", + Image: "main@sha256:abc", + Command: []string{"/main"}, + VolumeMounts: mounts, + }, + }, volumes) +} + +func createTemplateWithContainersAndVolumes(t *testing.T, tc *testContext, ns string, containers []atev1alpha1.Container, volumes []atev1alpha1.Volume) { t.Helper() // Sandbox binaries now live on a (cluster-scoped) SandboxConfig resolved via @@ -442,6 +459,7 @@ func createTemplateWithContainers(t *testing.T, tc *testContext, ns string, cont Location: "gs://fake-fake-fake", }, Containers: containers, + Volumes: volumes, WorkerSelector: &metav1.LabelSelector{ MatchLabels: map[string]string{poolLabelKey: ns}, }, @@ -723,7 +741,7 @@ func TestCreateActor_Success(t *testing.T) { } want := &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Name: "id1", Atespace: testAtespace, Version: 1}, + Metadata: &ateapipb.ResourceMetadata{Name: "id1", Atespace: testAtespace, Version: 2}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl1", Status: ateapipb.Actor_STATUS_SUSPENDED, @@ -752,6 +770,283 @@ func TestCreateActor_Success(t *testing.T) { } } +func TestCreateActor_WithExternalVolumes(t *testing.T) { + ns := namespaceForTest("ns-create-ext-vols") + tc := setupTest(t, ns) + defer tc.cleanup() + + volumes := []atev1alpha1.Volume{ + { + Name: "ext-vol-1", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + Capacity: resource.MustParse("10Gi"), + }, + }, + }, + } + mounts := []atev1alpha1.VolumeMount{ + { + Name: "ext-vol-1", + MountPath: "/data", + }, + } + createTemplateWithVolumes(t, tc, ns, volumes, mounts) + + createResp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{ + Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "vol-actor-1"}, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + }, + }) + if err != nil { + t.Fatalf("CreateActor failed: %v", err) + } + + if len(createResp.GetActorVolumes()) != 1 { + t.Fatalf("expected 1 volume in CreateActor response, got %d", len(createResp.GetActorVolumes())) + } + vol := createResp.GetActorVolumes()[0] + if vol.GetVolumeName() != "ext-vol-1" { + t.Errorf("volume name = %q, want %q", vol.GetVolumeName(), "ext-vol-1") + } + if vol.GetStatus() != ateapipb.ExternalVolume_CREATED { + t.Errorf("volume status = %v, want %v", vol.GetStatus(), ateapipb.ExternalVolume_CREATED) + } + if vol.GetStorageVolumeId() == "" { + t.Errorf("expected non-empty storageVolumeId") + } + + // Verify GetActor returns the same external volume state + getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "vol-actor-1"}, + }) + if err != nil { + t.Fatalf("GetActor failed: %v", err) + } + if len(getResp.GetActorVolumes()) != 1 { + t.Fatalf("expected 1 volume in GetActor response, got %d", len(getResp.GetActorVolumes())) + } + if getResp.GetActorVolumes()[0].GetStorageVolumeId() != vol.GetStorageVolumeId() { + t.Errorf("GetActor storageVolumeId = %q, want %q", getResp.GetActorVolumes()[0].GetStorageVolumeId(), vol.GetStorageVolumeId()) + } +} + +func TestActorLifecycle_WithExternalVolumes(t *testing.T) { + ns := namespaceForTest("ns-lifecycle-ext-vols") + tc := setupTest(t, ns) + defer tc.cleanup() + + volumes := []atev1alpha1.Volume{ + { + Name: "data-vol", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "fast", + Capacity: resource.MustParse("20Gi"), + }, + }, + }, + } + mounts := []atev1alpha1.VolumeMount{ + { + Name: "data-vol", + MountPath: "/mnt/data", + }, + } + createTemplateWithVolumes(t, tc, ns, volumes, mounts) + createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1") + + // 1. CreateActor + createResp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{ + Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "actor-vol-lc"}, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + }, + }) + if err != nil { + t.Fatalf("CreateActor failed: %v", err) + } + if createResp.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED { + t.Fatalf("expected initial status STATUS_SUSPENDED, got %v", createResp.GetStatus()) + } + if len(createResp.GetActorVolumes()) != 1 || createResp.GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_CREATED { + t.Fatalf("expected 1 created volume, got %v", createResp.GetActorVolumes()) + } + + // 2. ResumeActor + resumeResp, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"}, + }) + if err != nil { + t.Fatalf("ResumeActor failed: %v", err) + } + if resumeResp.GetActor().GetStatus() != ateapipb.Actor_STATUS_RUNNING { + t.Fatalf("expected status STATUS_RUNNING after resume, got %v", resumeResp.GetActor().GetStatus()) + } + + // 3. PauseActor + pauseResp, err := tc.client.PauseActor(context.Background(), &ateapipb.PauseActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"}, + }) + if err != nil { + t.Fatalf("PauseActor failed: %v", err) + } + if pauseResp.GetActor().GetStatus() != ateapipb.Actor_STATUS_PAUSED { + t.Fatalf("expected status STATUS_PAUSED after pause, got %v", pauseResp.GetActor().GetStatus()) + } + + // 4. ResumeActor from paused + resumeResp2, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"}, + }) + if err != nil { + t.Fatalf("ResumeActor from paused failed: %v", err) + } + if resumeResp2.GetActor().GetStatus() != ateapipb.Actor_STATUS_RUNNING { + t.Fatalf("expected status STATUS_RUNNING after second resume, got %v", resumeResp2.GetActor().GetStatus()) + } + + // 5. SuspendActor + suspendResp, err := tc.client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"}, + }) + if err != nil { + t.Fatalf("SuspendActor failed: %v", err) + } + if suspendResp.GetActor().GetStatus() != ateapipb.Actor_STATUS_SUSPENDED { + t.Fatalf("expected status STATUS_SUSPENDED after suspend, got %v", suspendResp.GetActor().GetStatus()) + } + + // 6. DeleteActor + deleteResp, err := tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"}, + }) + if err != nil { + t.Fatalf("DeleteActor failed: %v", err) + } + if deleteResp.GetMetadata().GetName() != "actor-vol-lc" { + t.Errorf("deleted actor name = %q, want %q", deleteResp.GetMetadata().GetName(), "actor-vol-lc") + } + + // Confirm GetActor returns NotFound after deletion + _, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"}, + }) + if status.Code(err) != codes.NotFound { + t.Errorf("GetActor after delete err = %v, want NotFound", err) + } +} + +type partialFailVolumePlugin struct { + volume.VolumePluginControlPlane + deleted []string +} + +func (f *partialFailVolumePlugin) CreateVolume(ctx context.Context, name, capacity, storageClass string) (string, error) { + if strings.HasSuffix(name, "fail-vol2") { + return "", fmt.Errorf("simulated volume creation failure") + } + return "storage-" + name, nil +} + +func (f *partialFailVolumePlugin) DeleteVolume(ctx context.Context, volumeID string) error { + f.deleted = append(f.deleted, volumeID) + return nil +} + +// TestCreateActor_FailedStatus tests that when volume provisioning fails during CreateActor, +// the actor is persisted with STATUS_CREATING and CreateActor returns an error, +// and that calling DeleteActor on the creating actor cleans up all partially created volumes. +func TestCreateActor_FailedStatus(t *testing.T) { + ns := namespaceForTest("ns-create-failed") + tc := setupTest(t, ns) + defer tc.cleanup() + + volumes := []atev1alpha1.Volume{ + { + Name: "succ-vol1", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + Capacity: resource.MustParse("10Gi"), + }, + }, + }, + { + Name: "fail-vol2", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + Capacity: resource.MustParse("10Gi"), + }, + }, + }, + } + createTemplateWithVolumes(t, tc, ns, volumes, nil) + + // Inject a custom partial-failing VolumePlugin into Service + plugin := &partialFailVolumePlugin{} + tc.service.volumePlugin = plugin + defer func() { tc.service.volumePlugin = nil }() + + // Call CreateActor RPC directly + _, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{ + Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "fail-actor"}, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + }, + }) + if err == nil { + t.Fatalf("expected CreateActor to fail, but it succeeded") + } + + // Verify GetActor returns the actor in STATUS_CREATING status + getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "fail-actor"}, + }) + if err != nil { + t.Fatalf("GetActor failed: %v", err) + } + if getResp.GetStatus() != ateapipb.Actor_STATUS_CREATING { + t.Errorf("actor status = %v, want %v", getResp.GetStatus(), ateapipb.Actor_STATUS_CREATING) + } + + actorUID := getResp.GetMetadata().GetUid() + if actorUID == "" { + t.Fatalf("expected non-empty UID on actor in STATUS_CREATING") + } + + // Call DeleteActor on the actor in STATUS_CREATING + _, err = tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "fail-actor"}, + }) + if err != nil { + t.Fatalf("DeleteActor failed: %v", err) + } + + // Verify both volumes were deleted (succ-vol1 via storageID, fail-vol2 via fallback actorVolumeID) + wantDeleted := []string{ + "storage-substrate-" + actorUID + "-succ-vol1", + "substrate-" + actorUID + "-fail-vol2", + } + if diff := cmp.Diff(wantDeleted, plugin.deleted); diff != "" { + t.Errorf("deleted volume IDs mismatch (-want +got):\n%s", diff) + } + + // Confirm GetActor returns NotFound after deletion + _, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "fail-actor"}, + }) + if status.Code(err) != codes.NotFound { + t.Errorf("GetActor after DeleteActor err = %v, want NotFound", err) + } +} + // TestCreateActor_TemplateNotFound tests that creating an actor with a non-existent template fails with FailedPrecondition. func TestCreateActor_TemplateNotFound(t *testing.T) { ns := namespaceForTest("ns-create-notfound") @@ -1698,7 +1993,7 @@ func TestUpdateActor_Success(t *testing.T) { } wantActor := &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Name: "id1", Atespace: testAtespace, Version: 2}, + Metadata: &ateapipb.ResourceMetadata{Name: "id1", Atespace: testAtespace, Version: 3}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl1", Status: ateapipb.Actor_STATUS_SUSPENDED, @@ -2600,7 +2895,7 @@ func TestDeleteActor_NotSuspended(t *testing.T) { _, err = tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "id1"}, }) - assertGrpcError(t, err, codes.FailedPrecondition, "Actor id1 is not suspended (status: STATUS_RUNNING)") + assertGrpcError(t, err, codes.FailedPrecondition, "Actor id1 is not in a deletable status (status: STATUS_RUNNING)") } func TestDeleteActor_Crashed(t *testing.T) { @@ -2634,8 +2929,8 @@ func TestDeleteActor_Crashed(t *testing.T) { if err != nil { t.Fatalf("DeleteActor of crashed actor failed: %v", err) } - if got := deleted.GetStatus(); got != ateapipb.Actor_STATUS_CRASHED { - t.Errorf("deleted actor status = %v, want %v", got, ateapipb.Actor_STATUS_CRASHED) + if got := deleted.GetStatus(); got != ateapipb.Actor_STATUS_DELETING { + t.Errorf("deleted actor status = %v, want %v", got, ateapipb.Actor_STATUS_DELETING) } _, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ diff --git a/cmd/ateapi/internal/controlapi/service.go b/cmd/ateapi/internal/controlapi/service.go index 7ec695b00..1061ec1b7 100644 --- a/cmd/ateapi/internal/controlapi/service.go +++ b/cmd/ateapi/internal/controlapi/service.go @@ -17,6 +17,7 @@ package controlapi import ( "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache" + "github.com/agent-substrate/substrate/internal/volume" listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "k8s.io/client-go/kubernetes" @@ -30,6 +31,7 @@ type Service struct { actorTemplateLister listersv1alpha1.ActorTemplateLister workerPoolLister listersv1alpha1.WorkerPoolLister actorWorkflow *ActorWorkflow + volumePlugin volume.VolumePluginControlPlane } var _ ateapipb.ControlServer = (*Service)(nil) diff --git a/cmd/ateapi/internal/controlapi/volumes.go b/cmd/ateapi/internal/controlapi/volumes.go index 161c74e8c..c4019b236 100644 --- a/cmd/ateapi/internal/controlapi/volumes.go +++ b/cmd/ateapi/internal/controlapi/volumes.go @@ -37,46 +37,105 @@ func getVolumePlugin() volume.VolumePluginControlPlane { return globalVolumePlugin } -// TODO: we should persist creation first so that we can handle background cleanup. -// this probably requires us to add a PROVISIONING actor state. +// TODO: replace with actual volume plugin search. +func (s *Service) getVolumePlugin() volume.VolumePluginControlPlane { + if s != nil && s.volumePlugin != nil { + return s.volumePlugin + } + return getVolumePlugin() +} -// createActorVolumes provisions external volumes specified in the actor template. -// It returns the list of created external volumes, or an error if any creation fails. -// If any volume creation fails, it cleans up any volumes created in this call on a best-effort basis. -func (s *Service) createActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, template *atev1alpha1.ActorTemplate) ([]*ateapipb.ExternalVolume, error) { +// initialActorVolumes constructs initial volume objects in CREATING state before volume creation. +func initialActorVolumes(template *atev1alpha1.ActorTemplate) []*ateapipb.ExternalVolume { var volumes []*ateapipb.ExternalVolume for _, vol := range template.Spec.Volumes { if vol.ExternalVolumeTemplate != nil { - // Use a unique name for the volume to ensure idempotency - uniqueVolName := actorVolumeID(ref, vol.Name) - storageVolumeID, err := getVolumePlugin().CreateVolume(ctx, uniqueVolName, vol.ExternalVolumeTemplate.Capacity.String(), vol.ExternalVolumeTemplate.StorageClassName) - if err != nil { - // TODO: need better system - best effort cleanup of already created volumes - _ = s.deleteActorVolumes(ctx, ref, volumes) - return nil, status.Errorf(codes.Internal, "failed to create volume %q: %v", vol.Name, err) - } volumes = append(volumes, &ateapipb.ExternalVolume{ - ActorVolumeId: uniqueVolName, - StorageVolumeId: storageVolumeID, - VolumeType: "mock", // TODO fix when we support multiple plugins - Status: ateapipb.ExternalVolume_CREATED, + VolumeName: vol.Name, + VolumeType: "mock", // TODO fix when we support multiple plugins + Status: ateapipb.ExternalVolume_CREATING, }) } } - return volumes, nil + return volumes +} + +// createActorVolumes provisions external volumes specified in volumesToCreate. +// It returns the list of external volumes (with updated status and storage IDs), or an error if any creation fails. +// Any volumes processed before or during a failure are returned alongside the error so they can be persisted on the actor. +func (s *Service) createActorVolumes(ctx context.Context, actorUID string, template *atev1alpha1.ActorTemplate, volumesToCreate []*ateapipb.ExternalVolume) (resultVolumes []*ateapipb.ExternalVolume, err error) { + resultVolumes = make([]*ateapipb.ExternalVolume, 0, len(volumesToCreate)) + + var currentIdx int + defer func() { + if err != nil { + // If we encounter an error, append the rest of the volumes to the result with the last + // known state. This allows the caller to persist the state of the volumes. + resultVolumes = append(resultVolumes, volumesToCreate[currentIdx:]...) + } + }() + + for idx, vol := range volumesToCreate { + currentIdx = idx + + var specVol *atev1alpha1.Volume + volName := vol.GetVolumeName() + for i := range template.Spec.Volumes { + if template.Spec.Volumes[i].Name == volName { + specVol = &template.Spec.Volumes[i] + break + } + } + if specVol == nil || specVol.ExternalVolumeTemplate == nil { + return resultVolumes, status.Errorf(codes.NotFound, "volume %q not found in template", volName) + } + + switch vol.GetStatus() { + case ateapipb.ExternalVolume_CREATING: + // proceed with volume creation + case ateapipb.ExternalVolume_CREATED: + resultVolumes = append(resultVolumes, vol) + continue + case ateapipb.ExternalVolume_DELETING: + return resultVolumes, status.Errorf(codes.FailedPrecondition, "cannot create volume %q in DELETING status", volName) + default: + return resultVolumes, status.Errorf(codes.Internal, "unexpected status %s for volume %q", vol.GetStatus(), volName) + } + + actVolID := actorVolumeID(actorUID, volName) + + storageVolumeID, volErr := s.getVolumePlugin().CreateVolume(ctx, actVolID, specVol.ExternalVolumeTemplate.Capacity.String(), specVol.ExternalVolumeTemplate.StorageClassName) + if volErr != nil { + return resultVolumes, status.Errorf(codes.Internal, "failed to create volume %q: %v", specVol.Name, volErr) + } + + resultVolumes = append(resultVolumes, &ateapipb.ExternalVolume{ + VolumeName: volName, + StorageVolumeId: storageVolumeID, + VolumeType: vol.GetVolumeType(), + Status: ateapipb.ExternalVolume_CREATED, + }) + } + return resultVolumes, nil } // deleteActorVolumes deletes all external volumes in the list. -func (s *Service) deleteActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, volumes []*ateapipb.ExternalVolume) error { +func (s *Service) deleteActorVolumes(ctx context.Context, actorUID string, volumes []*ateapipb.ExternalVolume) error { + if actorUID == "" { + return errors.New("actorUID is required") + } var errs []error for _, vol := range volumes { - if err := getVolumePlugin().DeleteVolume(ctx, vol.GetStorageVolumeId()); err != nil { - slog.ErrorContext(ctx, "failed to delete volume", - slog.String("atespace", ref.GetAtespace()), - slog.String("actor_id", ref.GetName()), - slog.String("volume_id", vol.GetStorageVolumeId()), - slog.Any("error", err)) - errs = append(errs, fmt.Errorf("failed to delete volume %q: %w", vol.GetStorageVolumeId(), err)) + volID := vol.GetStorageVolumeId() + if volID == "" { + volID = actorVolumeID(actorUID, vol.GetVolumeName()) + } + if volID == "" { + errs = append(errs, fmt.Errorf("volume ID is empty for volume %q", vol.GetVolumeName())) + continue + } + if err := s.getVolumePlugin().DeleteVolume(ctx, volID); err != nil { + errs = append(errs, fmt.Errorf("failed to delete volume %q: %w", volID, err)) } } return errors.Join(errs...) @@ -89,8 +148,7 @@ func getMountedActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, volume // Find the corresponding volume in the ActorTemplate to check if it's mounted var matchedTemplateVol *atev1alpha1.Volume for _, tVol := range template.Spec.Volumes { - expectedID := actorVolumeID(ref, tVol.Name) - if vol.GetActorVolumeId() == expectedID { + if vol.GetVolumeName() == tVol.Name { matchedTemplateVol = &tVol break } @@ -110,9 +168,8 @@ func getMountedActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, volume return mounted } -func actorVolumeID(ref *ateapipb.ObjectRef, volumeName string) string { - // TODO consider if this should be actor UUID - return fmt.Sprintf("%s-%s-%s", ref.GetAtespace(), ref.GetName(), volumeName) +func actorVolumeID(actorUID string, volumeName string) string { + return fmt.Sprintf("substrate-%s-%s", actorUID, volumeName) } // detachActorVolumes detaches all mounted external volumes for an actor from its worker node. diff --git a/cmd/ateapi/internal/controlapi/volumes_test.go b/cmd/ateapi/internal/controlapi/volumes_test.go new file mode 100644 index 000000000..68f4dad82 --- /dev/null +++ b/cmd/ateapi/internal/controlapi/volumes_test.go @@ -0,0 +1,303 @@ +// 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 controlapi + +import ( + "context" + "testing" + + "github.com/google/go-cmp/cmp" + "google.golang.org/protobuf/testing/protocmp" + + "github.com/agent-substrate/substrate/internal/volume" + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +func TestInitialActorVolumes_CreatingState(t *testing.T) { + tmpl := &atev1alpha1.ActorTemplate{ + Spec: atev1alpha1.ActorTemplateSpec{ + Volumes: []atev1alpha1.Volume{ + { + Name: "data-vol-1", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + }, + }, + }, + { + Name: "scratch-vol", + }, + { + Name: "durable-vol", + VolumeSource: atev1alpha1.VolumeSource{ + DurableDir: &atev1alpha1.DurableDirVolumeSource{}, + }, + }, + { + Name: "data-vol-2", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "fast", + }, + }, + }, + }, + }, + } + + want := []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol-1", + VolumeType: "mock", + Status: ateapipb.ExternalVolume_CREATING, + }, + { + VolumeName: "data-vol-2", + VolumeType: "mock", + Status: ateapipb.ExternalVolume_CREATING, + }, + } + + initVols := initialActorVolumes(tmpl) + if diff := cmp.Diff(want, initVols, protocmp.Transform()); diff != "" { + t.Errorf("initialActorVolumes mismatch (-want +got):\n%s", diff) + } +} + +func TestCreateActorVolumes(t *testing.T) { + ctx := context.Background() + + standardTmpl := &atev1alpha1.ActorTemplate{ + Spec: atev1alpha1.ActorTemplateSpec{ + Volumes: []atev1alpha1.Volume{ + { + Name: "data-vol", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + }, + }, + }, + }, + }, + } + + multiVolTmpl := &atev1alpha1.ActorTemplate{ + Spec: atev1alpha1.ActorTemplateSpec{ + Volumes: []atev1alpha1.Volume{ + { + Name: "vol1", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + }, + }, + }, + { + Name: "vol2", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + }, + }, + }, + { + Name: "vol3", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + }, + }, + }, + }, + }, + } + + tests := []struct { + name string + tmpl *atev1alpha1.ActorTemplate + inputVolumes []*ateapipb.ExternalVolume + wantErr bool + wantRes []*ateapipb.ExternalVolume + }{ + { + name: "partial failure returns error and preserves succeeded, failed, and remaining volumes", + tmpl: multiVolTmpl, + inputVolumes: []*ateapipb.ExternalVolume{ + { + VolumeName: "vol1", + VolumeType: "mock", + Status: ateapipb.ExternalVolume_CREATING, + }, + { + VolumeName: "vol2", + Status: ateapipb.ExternalVolume_DELETING, + }, + { + VolumeName: "vol3", + Status: ateapipb.ExternalVolume_CREATING, + }, + }, + wantErr: true, + wantRes: []*ateapipb.ExternalVolume{ + { + VolumeName: "vol1", + StorageVolumeId: "mock-vol-1", + VolumeType: "mock", + Status: ateapipb.ExternalVolume_CREATED, + }, + { + VolumeName: "vol2", + Status: ateapipb.ExternalVolume_DELETING, + }, + { + VolumeName: "vol3", + Status: ateapipb.ExternalVolume_CREATING, + }, + }, + }, + { + name: "created volume status succeeds", + tmpl: standardTmpl, + inputVolumes: []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol", + StorageVolumeId: "existing-vol-id", + Status: ateapipb.ExternalVolume_CREATED, + }, + }, + wantErr: false, + wantRes: []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol", + StorageVolumeId: "existing-vol-id", + Status: ateapipb.ExternalVolume_CREATED, + }, + }, + }, + { + name: "unknown volume status returns error", + tmpl: standardTmpl, + inputVolumes: []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol", + Status: ateapipb.ExternalVolume_STATUS_UNKNOWN, + }, + }, + wantErr: true, + wantRes: []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol", + Status: ateapipb.ExternalVolume_STATUS_UNKNOWN, + }, + }, + }, + { + name: "volume not found in template returns error", + tmpl: &atev1alpha1.ActorTemplate{ + Spec: atev1alpha1.ActorTemplateSpec{ + Volumes: []atev1alpha1.Volume{}, + }, + }, + inputVolumes: []*ateapipb.ExternalVolume{ + { + VolumeName: "missing-vol", + Status: ateapipb.ExternalVolume_CREATING, + }, + }, + wantErr: true, + wantRes: []*ateapipb.ExternalVolume{ + { + VolumeName: "missing-vol", + Status: ateapipb.ExternalVolume_CREATING, + }, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + svc := &Service{ + volumePlugin: volume.NewMockVolumePlugin(), + } + res, err := svc.createActorVolumes(ctx, "actor-uid-123", tt.tmpl, tt.inputVolumes) + if (err != nil) != tt.wantErr { + t.Errorf("createActorVolumes() error = %v, wantErr %v", err, tt.wantErr) + } + if diff := cmp.Diff(tt.wantRes, res, protocmp.Transform()); diff != "" { + t.Errorf("createActorVolumes() mismatch (-want +got):\n%s", diff) + } + }) + } +} + +type trackingVolumePlugin struct { + volume.VolumePluginControlPlane + deletedIDs []string +} + +func (t *trackingVolumePlugin) DeleteVolume(ctx context.Context, volumeID string) error { + t.deletedIDs = append(t.deletedIDs, volumeID) + return nil +} + +func TestDeleteActorVolumes(t *testing.T) { + ctx := context.Background() + + tests := []struct { + name string + actorUID string + volumes []*ateapipb.ExternalVolume + wantDeleted []string + wantErr bool + }{ + { + name: "uses storage volume ID when present", + actorUID: "uid-abc", + volumes: []*ateapipb.ExternalVolume{ + {VolumeName: "vol1", StorageVolumeId: "storage-vol-123"}, + }, + wantDeleted: []string{"storage-vol-123"}, + wantErr: false, + }, + { + name: "falls back to actorVolumeID when storage volume ID is empty regardless of status", + actorUID: "uid-abc", + volumes: []*ateapipb.ExternalVolume{ + {VolumeName: "vol1", StorageVolumeId: "", Status: ateapipb.ExternalVolume_CREATED}, + }, + wantDeleted: []string{"substrate-uid-abc-vol1"}, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + plugin := &trackingVolumePlugin{} + svc := &Service{volumePlugin: plugin} + + err := svc.deleteActorVolumes(ctx, tt.actorUID, tt.volumes) + if (err != nil) != tt.wantErr { + t.Fatalf("deleteActorVolumes() error = %v, wantErr %v", err, tt.wantErr) + } + + if diff := cmp.Diff(tt.wantDeleted, plugin.deletedIDs); diff != "" { + t.Errorf("deletedIDs mismatch (-want +got):\n%s", diff) + } + }) + } +} diff --git a/cmd/ateapi/internal/controlapi/workload_spec.go b/cmd/ateapi/internal/controlapi/workload_spec.go index f57bb2c2f..cce9e02bf 100644 --- a/cmd/ateapi/internal/controlapi/workload_spec.go +++ b/cmd/ateapi/internal/controlapi/workload_spec.go @@ -128,9 +128,8 @@ func appendExternalVolumes(workloadSpec *ateletpb.WorkloadSpec, template *atev1a var storageVolID string var volType string - expectedID := actorVolumeID(&ateapipb.ObjectRef{Atespace: actor.GetMetadata().GetAtespace(), Name: actor.GetMetadata().GetName()}, vol.Name) for _, dbVol := range actor.GetActorVolumes() { - if dbVol.GetActorVolumeId() == expectedID { + if dbVol.GetVolumeName() == vol.Name { storageVolID = dbVol.GetStorageVolumeId() volType = dbVol.GetVolumeType() break diff --git a/cmd/ateapi/internal/controlapi/workload_spec_test.go b/cmd/ateapi/internal/controlapi/workload_spec_test.go index a41efa984..d55d44562 100644 --- a/cmd/ateapi/internal/controlapi/workload_spec_test.go +++ b/cmd/ateapi/internal/controlapi/workload_spec_test.go @@ -516,7 +516,7 @@ func TestAppendExternalVolumes(t *testing.T) { }, ActorVolumes: []*ateapipb.ExternalVolume{ { - ActorVolumeId: "space-abc-actor-123-vol-1", + VolumeName: "vol-1", StorageVolumeId: "vol-gce-pd-123", VolumeType: "pd-standard", }, diff --git a/cmd/ateapi/internal/store/ateredis/ateredis.go b/cmd/ateapi/internal/store/ateredis/ateredis.go index 161195ca8..6c7def524 100644 --- a/cmd/ateapi/internal/store/ateredis/ateredis.go +++ b/cmd/ateapi/internal/store/ateredis/ateredis.go @@ -500,8 +500,7 @@ func (s *Persistence) DeleteActor(ctx context.Context, atespace, name string) (* return fmt.Errorf("in protojson.Unmarshal: %w", err) } - if currentActor.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED && - currentActor.GetStatus() != ateapipb.Actor_STATUS_CRASHED { + if currentActor.GetStatus() != ateapipb.Actor_STATUS_DELETING { return store.ErrFailedPrecondition } diff --git a/cmd/ateapi/internal/store/ateredis/ateredis_test.go b/cmd/ateapi/internal/store/ateredis/ateredis_test.go index a5a4e7927..2bc6d7d85 100644 --- a/cmd/ateapi/internal/store/ateredis/ateredis_test.go +++ b/cmd/ateapi/internal/store/ateredis/ateredis_test.go @@ -413,8 +413,10 @@ func TestDeleteActor(t *testing.T) { status ateapipb.Actor_Status wantErr error }{ - {name: "suspended", status: ateapipb.Actor_STATUS_SUSPENDED}, - {name: "crashed", status: ateapipb.Actor_STATUS_CRASHED}, + {name: "suspended", status: ateapipb.Actor_STATUS_SUSPENDED, wantErr: store.ErrFailedPrecondition}, + {name: "crashed", status: ateapipb.Actor_STATUS_CRASHED, wantErr: store.ErrFailedPrecondition}, + {name: "creating", status: ateapipb.Actor_STATUS_CREATING, wantErr: store.ErrFailedPrecondition}, + {name: "deleting", status: ateapipb.Actor_STATUS_DELETING}, {name: "running", status: ateapipb.Actor_STATUS_RUNNING, wantErr: store.ErrFailedPrecondition}, {name: "paused", status: ateapipb.Actor_STATUS_PAUSED, wantErr: store.ErrFailedPrecondition}, } @@ -1364,7 +1366,7 @@ func TestDeleteAtespace_EmptyAfterActorsRemoved(t *testing.T) { if _, err := s.CreateAtespace(ctx, newTestAtespace("team-a")); err != nil { t.Fatalf("CreateAtespace failed: %v", err) } - if _, err := s.CreateActor(ctx, &ateapipb.Actor{Metadata: &ateapipb.ResourceMetadata{Name: "id1", Atespace: "team-a"}, Status: ateapipb.Actor_STATUS_SUSPENDED}); err != nil { + if _, err := s.CreateActor(ctx, &ateapipb.Actor{Metadata: &ateapipb.ResourceMetadata{Name: "id1", Atespace: "team-a"}, Status: ateapipb.Actor_STATUS_DELETING}); err != nil { t.Fatalf("CreateActor failed: %v", err) } if _, err := s.DeleteAtespace(ctx, "team-a"); !errors.Is(err, store.ErrFailedPrecondition) { diff --git a/pkg/proto/ateapipb/ateapi.pb.go b/pkg/proto/ateapipb/ateapi.pb.go index 6d5a6e70f..f05c789b4 100644 --- a/pkg/proto/ateapipb/ateapi.pb.go +++ b/pkg/proto/ateapipb/ateapi.pb.go @@ -41,22 +41,28 @@ const ( type ExternalVolume_Status int32 const ( - ExternalVolume_PROVISIONING ExternalVolume_Status = 0 - ExternalVolume_CREATED ExternalVolume_Status = 1 - ExternalVolume_DELETING ExternalVolume_Status = 2 + ExternalVolume_STATUS_UNKNOWN ExternalVolume_Status = 0 + // Volume being created in the storage system. + ExternalVolume_CREATING ExternalVolume_Status = 1 + // Volume successfully created in the storage system. + ExternalVolume_CREATED ExternalVolume_Status = 2 + // Volume being deleted from the storage system. + ExternalVolume_DELETING ExternalVolume_Status = 3 ) // Enum value maps for ExternalVolume_Status. var ( ExternalVolume_Status_name = map[int32]string{ - 0: "PROVISIONING", - 1: "CREATED", - 2: "DELETING", + 0: "STATUS_UNKNOWN", + 1: "CREATING", + 2: "CREATED", + 3: "DELETING", } ExternalVolume_Status_value = map[string]int32{ - "PROVISIONING": 0, - "CREATED": 1, - "DELETING": 2, + "STATUS_UNKNOWN": 0, + "CREATING": 1, + "CREATED": 2, + "DELETING": 3, } ) @@ -98,6 +104,8 @@ const ( Actor_STATUS_PAUSING Actor_Status = 5 Actor_STATUS_PAUSED Actor_Status = 6 Actor_STATUS_CRASHED Actor_Status = 7 + Actor_STATUS_CREATING Actor_Status = 8 + Actor_STATUS_DELETING Actor_Status = 9 ) // Enum value maps for Actor_Status. @@ -111,6 +119,8 @@ var ( 5: "STATUS_PAUSING", 6: "STATUS_PAUSED", 7: "STATUS_CRASHED", + 8: "STATUS_CREATING", + 9: "STATUS_DELETING", } Actor_Status_value = map[string]int32{ "STATUS_UNSPECIFIED": 0, @@ -121,6 +131,8 @@ var ( "STATUS_PAUSING": 5, "STATUS_PAUSED": 6, "STATUS_CRASHED": 7, + "STATUS_CREATING": 8, + "STATUS_DELETING": 9, } ) @@ -472,9 +484,10 @@ func (x *ResourceMetadata) GetUpdateTime() *timestamppb.Timestamp { type ExternalVolume struct { state protoimpl.MessageState `protogen:"open.v1"` - // actor_id + the volume name specified in the actor template. - ActorVolumeId string `protobuf:"bytes,1,opt,name=actor_volume_id,json=actorVolumeId,proto3" json:"actor_volume_id,omitempty"` - // The volume_id returned from the storage system. + // Name of the volume specified in the actor template. + VolumeName string `protobuf:"bytes,1,opt,name=volume_name,json=volumeName,proto3" json:"volume_name,omitempty"` + // The globally unique volume_id returned from the storage system. + // This will be initially empty during volume creation StorageVolumeId string `protobuf:"bytes,2,opt,name=storage_volume_id,json=storageVolumeId,proto3" json:"storage_volume_id,omitempty"` // Internal volume plugin name or CSI driver name. VolumeType string `protobuf:"bytes,3,opt,name=volume_type,json=volumeType,proto3" json:"volume_type,omitempty"` @@ -513,9 +526,9 @@ func (*ExternalVolume) Descriptor() ([]byte, []int) { return file_ateapi_proto_rawDescGZIP(), []int{5} } -func (x *ExternalVolume) GetActorVolumeId() string { +func (x *ExternalVolume) GetVolumeName() string { if x != nil { - return x.ActorVolumeId + return x.VolumeName } return "" } @@ -538,7 +551,7 @@ func (x *ExternalVolume) GetStatus() ExternalVolume_Status { if x != nil { return x.Status } - return ExternalVolume_PROVISIONING + return ExternalVolume_STATUS_UNKNOWN } type Actor struct { @@ -568,7 +581,7 @@ type Actor struct { WorkerPoolName string `protobuf:"bytes,12,opt,name=worker_pool_name,json=workerPoolName,proto3" json:"worker_pool_name,omitempty"` // Volumes attached to the actor. These volumes only live as long as the actor. // They are deleted when the actor is deleted. - ActorVolumes []*ExternalVolume `protobuf:"bytes,16,rep,name=actor_volumes,json=actorVolumes,proto3" json:"actor_volumes,omitempty"` + ActorVolumes []*ExternalVolume `protobuf:"bytes,13,rep,name=actor_volumes,json=actorVolumes,proto3" json:"actor_volumes,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -2351,17 +2364,19 @@ const file_ateapi_proto_rawDesc = "" + "\vcreate_time\x18\x05 \x01(\v2\x1a.google.protobuf.TimestampR\n" + "createTime\x12;\n" + "\vupdate_time\x18\x06 \x01(\v2\x1a.google.protobuf.TimestampR\n" + - "updateTime\"\xf3\x01\n" + - "\x0eExternalVolume\x12&\n" + - "\x0factor_volume_id\x18\x01 \x01(\tR\ractorVolumeId\x12*\n" + + "updateTime\"\xfc\x01\n" + + "\x0eExternalVolume\x12\x1f\n" + + "\vvolume_name\x18\x01 \x01(\tR\n" + + "volumeName\x12*\n" + "\x11storage_volume_id\x18\x02 \x01(\tR\x0fstorageVolumeId\x12\x1f\n" + "\vvolume_type\x18\x03 \x01(\tR\n" + "volumeType\x125\n" + - "\x06status\x18\x04 \x01(\x0e2\x1d.ateapi.ExternalVolume.StatusR\x06status\"5\n" + - "\x06Status\x12\x10\n" + - "\fPROVISIONING\x10\x00\x12\v\n" + - "\aCREATED\x10\x01\x12\f\n" + - "\bDELETING\x10\x02\"\xc1\x06\n" + + "\x06status\x18\x04 \x01(\x0e2\x1d.ateapi.ExternalVolume.StatusR\x06status\"E\n" + + "\x06Status\x12\x12\n" + + "\x0eSTATUS_UNKNOWN\x10\x00\x12\f\n" + + "\bCREATING\x10\x01\x12\v\n" + + "\aCREATED\x10\x02\x12\f\n" + + "\bDELETING\x10\x03\"\xeb\x06\n" + "\x05Actor\x124\n" + "\bmetadata\x18\x01 \x01(\v2\x18.ateapi.ResourceMetadataR\bmetadata\x128\n" + "\x18actor_template_namespace\x18\x02 \x01(\tR\x16actorTemplateNamespace\x12.\n" + @@ -2377,7 +2392,7 @@ const file_ateapi_proto_rawDesc = "" + " \x01(\v2\x14.ateapi.SnapshotInfoR\x12latestSnapshotInfo\x129\n" + "\x0fworker_selector\x18\v \x01(\v2\x10.ateapi.SelectorR\x0eworkerSelector\x12(\n" + "\x10worker_pool_name\x18\f \x01(\tR\x0eworkerPoolName\x12;\n" + - "\ractor_volumes\x18\x10 \x03(\v2\x16.ateapi.ExternalVolumeR\factorVolumes\"\xb1\x01\n" + + "\ractor_volumes\x18\r \x03(\v2\x16.ateapi.ExternalVolumeR\factorVolumes\"\xdb\x01\n" + "\x06Status\x12\x16\n" + "\x12STATUS_UNSPECIFIED\x10\x00\x12\x13\n" + "\x0fSTATUS_RESUMING\x10\x01\x12\x12\n" + @@ -2386,7 +2401,9 @@ const file_ateapi_proto_rawDesc = "" + "\x10STATUS_SUSPENDED\x10\x04\x12\x12\n" + "\x0eSTATUS_PAUSING\x10\x05\x12\x11\n" + "\rSTATUS_PAUSED\x10\x06\x12\x12\n" + - "\x0eSTATUS_CRASHED\x10\a\"@\n" + + "\x0eSTATUS_CRASHED\x10\a\x12\x13\n" + + "\x0fSTATUS_CREATING\x10\b\x12\x13\n" + + "\x0fSTATUS_DELETING\x10\t\"@\n" + "\bAtespace\x124\n" + "\bmetadata\x18\x01 \x01(\v2\x18.ateapi.ResourceMetadataR\bmetadata\";\n" + "\tObjectRef\x12\x1a\n" + diff --git a/pkg/proto/ateapipb/ateapi.proto b/pkg/proto/ateapipb/ateapi.proto index 27c87e552..911b7fa17 100644 --- a/pkg/proto/ateapipb/ateapi.proto +++ b/pkg/proto/ateapipb/ateapi.proto @@ -115,19 +115,24 @@ message ResourceMetadata { } message ExternalVolume { - // actor_id + the volume name specified in the actor template. - string actor_volume_id = 1; + // Name of the volume specified in the actor template. + string volume_name = 1; - // The volume_id returned from the storage system. + // The globally unique volume_id returned from the storage system. + // This will be initially empty during volume creation string storage_volume_id = 2; // Internal volume plugin name or CSI driver name. string volume_type = 3; enum Status { - PROVISIONING = 0; - CREATED = 1; - DELETING = 2; + STATUS_UNKNOWN = 0; + // Volume being created in the storage system. + CREATING = 1; + // Volume successfully created in the storage system. + CREATED = 2; + // Volume being deleted from the storage system. + DELETING = 3; } Status status = 4; } @@ -148,6 +153,8 @@ message Actor { STATUS_PAUSING = 5; STATUS_PAUSED = 6; STATUS_CRASHED = 7; + STATUS_CREATING = 8; + STATUS_DELETING = 9; } Status status = 4; @@ -175,7 +182,7 @@ message Actor { // Volumes attached to the actor. These volumes only live as long as the actor. // They are deleted when the actor is deleted. - repeated ExternalVolume actor_volumes = 16; + repeated ExternalVolume actor_volumes = 13; } // Atespace is the isolation boundary an Actor is created into. Global-scoped: