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: