diff --git a/cmd/ateapi/internal/controlapi/create_actor.go b/cmd/ateapi/internal/controlapi/create_actor.go index f19ee0d1b..475134b60 100644 --- a/cmd/ateapi/internal/controlapi/create_actor.go +++ b/cmd/ateapi/internal/controlapi/create_actor.go @@ -59,15 +59,9 @@ 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 - } + // Assuming CreateActor() will not become idempotent so we won't override volume state + // if the actor already exists. + initVols := initialActorVolumes(template) actor := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{ @@ -78,12 +72,10 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ ActorTemplateNamespace: templateNamespace, ActorTemplateName: templateName, WorkerSelector: in.GetWorkerSelector(), - ActorVolumes: volumes, + ActorVolumes: initVols, // TODO: validate external volume fields } 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) } diff --git a/cmd/ateapi/internal/controlapi/delete_actor.go b/cmd/ateapi/internal/controlapi/delete_actor.go index 1dc2a1c01..2f9b5f546 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor.go +++ b/cmd/ateapi/internal/controlapi/delete_actor.go @@ -16,14 +16,9 @@ package controlapi import ( "context" - "errors" - "fmt" - "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" "k8s.io/apimachinery/pkg/util/validation/field" ) @@ -34,35 +29,9 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ actorRef := resources.ActorRefFromObjectRef(req.GetActor()) setSpanActorRefAttributes(ctx, actorRef) - actor, err := s.persistence.GetActor(ctx, actorRef) + deleted, err := s.actorWorkflow.DeleteActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName()) if err != nil { - if errors.Is(err, store.ErrNotFound) { - return nil, status.Errorf(codes.NotFound, "Actor %s not found", actorRef) - } - return nil, fmt.Errorf("while fetching actor: %w", err) - } - - // Delete associated volumes - if err := s.deleteActorVolumes(ctx, req.GetActor(), actor.GetActorVolumes()); err != nil { - return nil, status.Errorf(codes.Internal, "while deleting actor volumes: %v", err) - } - - deleted, err := s.persistence.DeleteActor(ctx, actorRef) - if err != nil { - if errors.Is(err, store.ErrNotFound) { - return nil, status.Errorf(codes.NotFound, "Actor %s not found", actorRef) - } - if errors.Is(err, store.ErrFailedPrecondition) { - current, getErr := s.persistence.GetActor(ctx, actorRef) - if getErr == nil { - return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended (status: %v)", actorRef, current.GetStatus()) - } - return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended", actorRef) - } - if errors.Is(err, store.ErrVersionConflict) { - return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry") - } - return nil, fmt.Errorf("while deleting actor from DB: %w", err) + return nil, err } return deleted, nil diff --git a/cmd/ateapi/internal/controlapi/delete_actor_test.go b/cmd/ateapi/internal/controlapi/delete_actor_test.go index 9e78f676e..fdd9f05ec 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor_test.go +++ b/cmd/ateapi/internal/controlapi/delete_actor_test.go @@ -16,9 +16,15 @@ 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/resources" + "github.com/agent-substrate/substrate/internal/volume" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "k8s.io/apimachinery/pkg/util/validation/field" ) @@ -88,3 +94,116 @@ func TestValidateDeleteActorRequest(t *testing.T) { }) } } + +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(), resources.ActorRef{Atespace: testAtespace, Name: "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{} + oldGlobalPlugin := globalVolumePlugin + globalVolumePlugin = plugin + defer func() { globalVolumePlugin = oldGlobalPlugin }() + + 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_STATUS_CREATED}, + {VolumeName: "vol2", StorageVolumeId: "storage-vol-2", Status: ateapipb.ExternalVolume_STATUS_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 4b914d4fd..b23d8ce76 100644 --- a/cmd/ateapi/internal/controlapi/functional_test.go +++ b/cmd/ateapi/internal/controlapi/functional_test.go @@ -32,6 +32,7 @@ import ( "github.com/agent-substrate/substrate/internal/envtestbins" "github.com/agent-substrate/substrate/internal/proto/ateletpb" "github.com/agent-substrate/substrate/internal/resources" + "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" @@ -51,6 +52,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/validation/field" "k8s.io/apimachinery/pkg/util/wait" @@ -433,6 +435,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 @@ -452,6 +469,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}, }, @@ -764,6 +782,484 @@ 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_STATUS_PENDING { + t.Errorf("volume status = %v, want %v", vol.GetStatus(), ateapipb.ExternalVolume_STATUS_PENDING) + } + if vol.GetStorageVolumeId() != "" { + t.Errorf("expected empty storageVolumeId before resume, got %q", vol.GetStorageVolumeId()) + } + + // 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].GetStatus() != ateapipb.ExternalVolume_STATUS_PENDING { + t.Errorf("GetActor status = %v, want %v", getResp.GetActorVolumes()[0].GetStatus(), ateapipb.ExternalVolume_STATUS_PENDING) + } +} + +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_STATUS_PENDING { + t.Fatalf("expected 1 pending volume after CreateActor, 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()) + } + if len(resumeResp.GetActor().GetActorVolumes()) != 1 || resumeResp.GetActor().GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_STATUS_CREATED { + t.Fatalf("expected 1 created volume after ResumeActor, got %v", resumeResp.GetActor().GetActorVolumes()) + } + if resumeResp.GetActor().GetActorVolumes()[0].GetStorageVolumeId() == "" { + t.Fatalf("expected non-empty storageVolumeId after ResumeActor") + } + + // 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) AttachVolume(ctx context.Context, volumeID, node string) error { + return nil +} + +func (f *partialFailVolumePlugin) DetachVolume(ctx context.Context, volumeID, node string) error { + return nil +} + +func (f *partialFailVolumePlugin) DeleteVolume(ctx context.Context, volumeID string) error { + f.deleted = append(f.deleted, volumeID) + return nil +} + +// TestResumeActor_VolumeCreationFailure tests that when volume provisioning fails during ResumeActor, +// successfully created volumes are saved, the actor remains in STATUS_SUSPENDED, +// and that calling DeleteActor on the suspended actor cleans up all partially created volumes. +func TestResumeActor_VolumeCreationFailure(t *testing.T) { + ns := namespaceForTest("ns-resume-vol-fail") + 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"), + }, + }, + }, + } + mounts := []atev1alpha1.VolumeMount{ + {Name: "succ-vol1", MountPath: "/mnt/vol1"}, + {Name: "fail-vol2", MountPath: "/mnt/vol2"}, + } + createTemplateWithVolumes(t, tc, ns, volumes, mounts) + + // Inject a custom partial-failing VolumePlugin into global scope + // TODO this doesn't support parallelism of test cases + plugin := &partialFailVolumePlugin{} + oldGlobalPlugin := globalVolumePlugin + globalVolumePlugin = plugin + defer func() { + globalVolumePlugin = oldGlobalPlugin + }() + + // 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 succeed, got: %v", err) + } + + // Call ResumeActor RPC, which should trigger volume provisioning and fail on fail-vol2 + _, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "fail-actor"}, + }) + if err == nil { + t.Fatalf("expected ResumeActor to fail due to volume creation error, but it succeeded") + } + + // Verify GetActor returns the actor in STATUS_SUSPENDED 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_SUSPENDED { + t.Errorf("actor status = %v, want %v", getResp.GetStatus(), ateapipb.Actor_STATUS_SUSPENDED) + } + + actorUID := getResp.GetMetadata().GetUid() + if actorUID == "" { + t.Fatalf("expected non-empty UID on actor") + } + + // Verify that succ-vol1 was updated to CREATED with a storageVolumeId, and fail-vol2 is still PENDING + if len(getResp.GetActorVolumes()) != 2 { + t.Fatalf("expected 2 volumes on actor, got %d", len(getResp.GetActorVolumes())) + } + volsByName := make(map[string]*ateapipb.ExternalVolume) + for _, v := range getResp.GetActorVolumes() { + volsByName[v.GetVolumeName()] = v + } + if v1, ok := volsByName["succ-vol1"]; !ok || v1.GetStatus() != ateapipb.ExternalVolume_STATUS_CREATED || v1.GetStorageVolumeId() == "" { + t.Errorf("succ-vol1 unexpected state: %v", v1) + } + if v2, ok := volsByName["fail-vol2"]; !ok || v2.GetStatus() != ateapipb.ExternalVolume_STATUS_PENDING { + t.Errorf("fail-vol2 unexpected state: %v", v2) + } + + // Call DeleteActor on the actor in STATUS_SUSPENDED + _, 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) + } +} + +type retrySuccessVolumePlugin struct { + volume.VolumePluginControlPlane + mu sync.Mutex + attempts int + deleted []string +} + +func (r *retrySuccessVolumePlugin) CreateVolume(ctx context.Context, name, capacity, storageClass string) (string, error) { + r.mu.Lock() + defer r.mu.Unlock() + if strings.HasSuffix(name, "retry-vol2") { + r.attempts++ + if r.attempts == 1 { + return "", fmt.Errorf("simulated temporary volume creation failure") + } + } + return "storage-" + name, nil +} + +func (r *retrySuccessVolumePlugin) AttachVolume(ctx context.Context, volumeID, node string) error { + return nil +} + +func (r *retrySuccessVolumePlugin) DetachVolume(ctx context.Context, volumeID, node string) error { + return nil +} + +func (r *retrySuccessVolumePlugin) DeleteVolume(ctx context.Context, volumeID string) error { + r.mu.Lock() + defer r.mu.Unlock() + r.deleted = append(r.deleted, volumeID) + return nil +} + +// TestResumeActor_VolumeCreationRetrySuccess tests that when volume provisioning fails on the first ResumeActor call, +// a subsequent call to ResumeActor retries provisioning only the pending volumes and succeeds. +func TestResumeActor_VolumeCreationRetrySuccess(t *testing.T) { + ns := namespaceForTest("ns-resume-vol-retry") + 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: "retry-vol2", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + Capacity: resource.MustParse("10Gi"), + }, + }, + }, + } + retryMounts := []atev1alpha1.VolumeMount{ + {Name: "succ-vol1", MountPath: "/mnt/vol1"}, + {Name: "retry-vol2", MountPath: "/mnt/vol2"}, + } + createTemplateWithVolumes(t, tc, ns, volumes, retryMounts) + createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1") + + plugin := &retrySuccessVolumePlugin{} + oldGlobalPlugin := globalVolumePlugin + globalVolumePlugin = plugin + defer func() { + globalVolumePlugin = oldGlobalPlugin + }() + + // Call CreateActor RPC directly + _, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{ + Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "retry-actor"}, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + }, + }) + if err != nil { + t.Fatalf("expected CreateActor to succeed, got: %v", err) + } + + // First call to ResumeActor RPC, which should fail on retry-vol2 (attempt 1) + _, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, + }) + if err == nil { + t.Fatalf("expected first ResumeActor to fail due to temporary volume creation error, but it succeeded") + } + + // Verify GetActor returns the actor in STATUS_SUSPENDED status with succ-vol1 created and retry-vol2 pending + getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, + }) + if err != nil { + t.Fatalf("GetActor after first resume failed: %v", err) + } + if getResp.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED { + t.Errorf("actor status after first resume = %v, want %v", getResp.GetStatus(), ateapipb.Actor_STATUS_SUSPENDED) + } + + volsByName := make(map[string]*ateapipb.ExternalVolume) + for _, v := range getResp.GetActorVolumes() { + volsByName[v.GetVolumeName()] = v + } + if v1, ok := volsByName["succ-vol1"]; !ok || v1.GetStatus() != ateapipb.ExternalVolume_STATUS_CREATED || v1.GetStorageVolumeId() == "" { + t.Errorf("succ-vol1 unexpected state after first resume: %v", v1) + } + if v2, ok := volsByName["retry-vol2"]; !ok || v2.GetStatus() != ateapipb.ExternalVolume_STATUS_PENDING { + t.Errorf("retry-vol2 unexpected state after first resume: %v", v2) + } + + // Second call to ResumeActor RPC, which should succeed on retry-vol2 (attempt 2) + _, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, + }) + if err != nil { + t.Fatalf("expected second ResumeActor to succeed, got: %v", err) + } + + // Verify GetActor returns the actor in STATUS_RUNNING status with both volumes CREATED + getResp, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, + }) + if err != nil { + t.Fatalf("GetActor after second resume failed: %v", err) + } + if getResp.GetStatus() != ateapipb.Actor_STATUS_RUNNING { + t.Errorf("actor status after second resume = %v, want %v", getResp.GetStatus(), ateapipb.Actor_STATUS_RUNNING) + } + for _, v := range getResp.GetActorVolumes() { + if v.GetStatus() != ateapipb.ExternalVolume_STATUS_CREATED || v.GetStorageVolumeId() == "" { + t.Errorf("volume %s unexpected state after second resume: %v", v.GetVolumeName(), v) + } + } + + // Clean up by suspending and deleting the actor + _, err = tc.client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, + }) + if err != nil { + t.Fatalf("SuspendActor failed: %v", err) + } + _, err = tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, + }) + if err != nil { + t.Fatalf("DeleteActor failed: %v", 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") @@ -2328,7 +2824,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 test-atespace/id1 is not suspended (status: STATUS_RUNNING)") + assertGrpcError(t, err, codes.FailedPrecondition, "Actor test-atespace/id1 is not in a deletable status (status: STATUS_RUNNING)") } func TestDeleteActor_Crashed(t *testing.T) { @@ -2362,8 +2858,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..944ae441f 100644 --- a/cmd/ateapi/internal/controlapi/service.go +++ b/cmd/ateapi/internal/controlapi/service.go @@ -51,6 +51,5 @@ func NewService( dialer: dialer, actorWorkflow: NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, kubeClient), } - return s } diff --git a/cmd/ateapi/internal/controlapi/volumes.go b/cmd/ateapi/internal/controlapi/volumes.go index 161c74e8c..c14b87996 100644 --- a/cmd/ateapi/internal/controlapi/volumes.go +++ b/cmd/ateapi/internal/controlapi/volumes.go @@ -29,7 +29,7 @@ import ( ) var ( - globalVolumePlugin = volume.NewMockVolumePlugin() + globalVolumePlugin volume.VolumePluginControlPlane = volume.NewMockVolumePlugin() ) // TODO: Replace with actual volume plugin search @@ -37,46 +37,100 @@ 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. - -// 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 PENDING 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_STATUS_PENDING, }) } } - return volumes, nil + return volumes +} + +// createActorVolumes provisions external volumes specified in volumesToCreate using the provided volume plugin. +// 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 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_STATUS_PENDING: + // proceed with volume creation + case ateapipb.ExternalVolume_STATUS_CREATED: + resultVolumes = append(resultVolumes, vol) + continue + case ateapipb.ExternalVolume_STATUS_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) + + plugin := getVolumePlugin() + if plugin == nil { + return resultVolumes, status.Errorf(codes.Internal, "volume plugin is not configured for creating volume %q", volName) + } + storageVolumeID, volErr := plugin.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_STATUS_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 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 == "" { + // If the volume hasn't been successfully created yet, it's possible + // that it doesn't have a storage volume ID. In that case, fallback + // to the original requested volID. + volID = actorVolumeID(actorUID, vol.GetVolumeName()) + } + if err := 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 +143,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 +163,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..50643c48f --- /dev/null +++ b/cmd/ateapi/internal/controlapi/volumes_test.go @@ -0,0 +1,302 @@ +// 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_PendingState(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_STATUS_PENDING, + }, + { + VolumeName: "data-vol-2", + VolumeType: "mock", + Status: ateapipb.ExternalVolume_STATUS_PENDING, + }, + } + + 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_STATUS_PENDING, + }, + { + VolumeName: "vol2", + Status: ateapipb.ExternalVolume_STATUS_DELETING, + }, + { + VolumeName: "vol3", + Status: ateapipb.ExternalVolume_STATUS_PENDING, + }, + }, + wantErr: true, + wantRes: []*ateapipb.ExternalVolume{ + { + VolumeName: "vol1", + StorageVolumeId: "mock-vol-substrate-actor-uid-123-vol1", + VolumeType: "mock", + Status: ateapipb.ExternalVolume_STATUS_CREATED, + }, + { + VolumeName: "vol2", + Status: ateapipb.ExternalVolume_STATUS_DELETING, + }, + { + VolumeName: "vol3", + Status: ateapipb.ExternalVolume_STATUS_PENDING, + }, + }, + }, + { + name: "created volume status succeeds", + tmpl: standardTmpl, + inputVolumes: []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol", + StorageVolumeId: "existing-vol-id", + Status: ateapipb.ExternalVolume_STATUS_CREATED, + }, + }, + wantErr: false, + wantRes: []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol", + StorageVolumeId: "existing-vol-id", + Status: ateapipb.ExternalVolume_STATUS_CREATED, + }, + }, + }, + { + name: "unspecified volume status returns error", + tmpl: standardTmpl, + inputVolumes: []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol", + Status: ateapipb.ExternalVolume_STATUS_UNSPECIFIED, + }, + }, + wantErr: true, + wantRes: []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol", + Status: ateapipb.ExternalVolume_STATUS_UNSPECIFIED, + }, + }, + }, + { + 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_STATUS_PENDING, + }, + }, + wantErr: true, + wantRes: []*ateapipb.ExternalVolume{ + { + VolumeName: "missing-vol", + Status: ateapipb.ExternalVolume_STATUS_PENDING, + }, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + globalVolumePlugin = volume.NewMockVolumePlugin() + res, err := 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_STATUS_CREATED}, + }, + wantDeleted: []string{"substrate-uid-abc-vol1"}, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + plugin := &trackingVolumePlugin{} + oldGlobalPlugin := globalVolumePlugin + globalVolumePlugin = plugin + defer func() { globalVolumePlugin = oldGlobalPlugin }() + err := 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/workflow.go b/cmd/ateapi/internal/controlapi/workflow.go index f06efa7b6..5289670d2 100644 --- a/cmd/ateapi/internal/controlapi/workflow.go +++ b/cmd/ateapi/internal/controlapi/workflow.go @@ -180,6 +180,7 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, actorRef resources.Acto steps := []WorkflowStep[*ResumeInput, *ResumeState]{ &LoadActorForResumeStep{store: w.store, actorTemplateLister: w.actorTemplateLister}, + &CreateVolumesStep{store: w.store}, &AssignWorkerStep{store: w.store, workerCache: w.workerCache, scheduler: w.scheduler}, &AttachVolumesStep{store: w.store}, &CallAteletRestoreStep{store: w.store, dialer: w.dialer, kubeClient: w.kubeClient, secretCache: w.secretCache, workerPoolLister: w.workerPoolLister, sandboxConfigLister: w.sandboxConfigLister, scheduler: w.scheduler}, @@ -249,6 +250,34 @@ func (w *ActorWorkflow) PauseActor(ctx context.Context, actorRef resources.Actor return state.Actor, nil } +// DeleteActor executes the workflow to delete an actor. Idempotent. +func (w *ActorWorkflow) DeleteActor(ctx context.Context, atespace, name string) (*ateapipb.Actor, error) { + actorRef := resources.ActorRef{Atespace: atespace, Name: name} + input := &DeleteInput{ + ActorRef: actorRef, + } + state := &DeleteState{} + + ctx, lock, err := w.acquireActorLock(ctx, actorRef) + if err != nil { + return nil, err + } + defer lock.Close() + + steps := []WorkflowStep[*DeleteInput, *DeleteState]{ + &LoadActorForDeleteStep{store: w.store}, + &MarkDeletingStep{store: w.store}, + &DeleteVolumesStep{store: w.store}, + &FinalizeDeletedStep{store: w.store}, + } + + if err := RunWorkflow(ctx, input, state, steps); err != nil { + return nil, err + } + + return state.DeletedActor, nil +} + func (w *ActorWorkflow) acquireActorLock(ctx context.Context, actorRef resources.ActorRef) (context.Context, *store.Lock, error) { lockKey := "lock:actor:" + actorRef.Atespace + ":" + actorRef.Name diff --git a/cmd/ateapi/internal/controlapi/workflow_delete.go b/cmd/ateapi/internal/controlapi/workflow_delete.go new file mode 100644 index 000000000..761f46c78 --- /dev/null +++ b/cmd/ateapi/internal/controlapi/workflow_delete.go @@ -0,0 +1,160 @@ +// 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" + "errors" + "fmt" + + "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "k8s.io/apimachinery/pkg/util/wait" +) + +// DeleteInput holds the immutable parameters requested by the client. +type DeleteInput struct { + ActorRef resources.ActorRef +} + +// DeleteState holds the mutable state loaded and modified during execution. +type DeleteState struct { + Actor *ateapipb.Actor + DeletedActor *ateapipb.Actor +} + +type LoadActorForDeleteStep struct { + store store.Interface +} + +func (s *LoadActorForDeleteStep) Name() string { return "LoadActorForDelete" } +func (s *LoadActorForDeleteStep) IsComplete(ctx context.Context, input *DeleteInput, state *DeleteState) (bool, error) { + // Always run to get the freshest state + return false, nil +} +func (s *LoadActorForDeleteStep) CheckPrerequisite(ctx context.Context, input *DeleteInput, state *DeleteState) error { + return nil +} +func (s *LoadActorForDeleteStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { + actor, err := s.store.GetActor(ctx, input.ActorRef) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return status.Errorf(codes.NotFound, "Actor %s not found", input.ActorRef) + } + return fmt.Errorf("while fetching actor: %w", err) + } + state.Actor = actor + return nil +} + +func (s *LoadActorForDeleteStep) RetryBackoff() *wait.Backoff { return nil } + +type MarkDeletingStep struct { + store store.Interface +} + +func (s *MarkDeletingStep) Name() string { return "MarkDeleting" } +func (s *MarkDeletingStep) IsComplete(ctx context.Context, input *DeleteInput, state *DeleteState) (bool, error) { + return state.Actor.GetStatus() == ateapipb.Actor_STATUS_DELETING, nil +} +func (s *MarkDeletingStep) CheckPrerequisite(ctx context.Context, input *DeleteInput, state *DeleteState) error { + if state.Actor.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED && + state.Actor.GetStatus() != ateapipb.Actor_STATUS_CRASHED && + state.Actor.GetStatus() != ateapipb.Actor_STATUS_DELETING { + return status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status (status: %v)", input.ActorRef, state.Actor.GetStatus()) + } + return nil +} +func (s *MarkDeletingStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { + state.Actor.Status = ateapipb.Actor_STATUS_DELETING + for _, vol := range state.Actor.GetActorVolumes() { + vol.Status = ateapipb.ExternalVolume_STATUS_DELETING + } + updated, err := s.store.UpdateActor(ctx, state.Actor, state.Actor.GetMetadata().GetVersion()) + if err != nil { + if errors.Is(err, store.ErrVersionConflict) { + return status.Error(codes.Aborted, "concurrent update conflict, please retry") + } + return fmt.Errorf("while setting actor status to DELETING: %w", err) + } + state.Actor = updated + return nil +} + +func (s *MarkDeletingStep) RetryBackoff() *wait.Backoff { return nil } + +type DeleteVolumesStep struct { + store store.Interface +} + +func (s *DeleteVolumesStep) Name() string { return "DeleteVolumes" } +func (s *DeleteVolumesStep) IsComplete(ctx context.Context, input *DeleteInput, state *DeleteState) (bool, error) { + return false, nil +} +func (s *DeleteVolumesStep) CheckPrerequisite(ctx context.Context, input *DeleteInput, state *DeleteState) error { + if state.Actor.GetStatus() != ateapipb.Actor_STATUS_DELETING { + return status.Errorf(codes.FailedPrecondition, "DeleteVolumesStep prerequisite not met for Actor: %s (got: %v, want %s)", input.ActorRef, state.Actor.GetStatus(), ateapipb.Actor_STATUS_DELETING) + } + return nil +} +func (s *DeleteVolumesStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { + if err := deleteActorVolumes(ctx, state.Actor.GetMetadata().GetUid(), state.Actor.GetActorVolumes()); err != nil { + return status.Errorf(codes.Internal, "while deleting actor volumes: %v", err) + } + return nil +} + +func (s *DeleteVolumesStep) RetryBackoff() *wait.Backoff { return nil } + +type FinalizeDeletedStep struct { + store store.Interface +} + +func (s *FinalizeDeletedStep) Name() string { return "FinalizeDeleted" } +func (s *FinalizeDeletedStep) IsComplete(ctx context.Context, input *DeleteInput, state *DeleteState) (bool, error) { + return state.DeletedActor != nil, nil +} +func (s *FinalizeDeletedStep) CheckPrerequisite(ctx context.Context, input *DeleteInput, state *DeleteState) error { + if state.Actor.GetStatus() != ateapipb.Actor_STATUS_DELETING { + return status.Errorf(codes.FailedPrecondition, "FinalizeDeletedStep prerequisite not met for Actor: %s (got: %v, want %s)", input.ActorRef, state.Actor.GetStatus(), ateapipb.Actor_STATUS_DELETING) + } + return nil +} +func (s *FinalizeDeletedStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { + deleted, err := s.store.DeleteActor(ctx, input.ActorRef) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return status.Errorf(codes.NotFound, "Actor %s not found", input.ActorRef) + } + if errors.Is(err, store.ErrFailedPrecondition) { + current, getErr := s.store.GetActor(ctx, input.ActorRef) + if getErr == nil { + return status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status (status: %v)", input.ActorRef, current.GetStatus()) + } + return status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status", input.ActorRef) + } + if errors.Is(err, store.ErrVersionConflict) { + return status.Error(codes.Aborted, "concurrent update conflict, please retry") + } + return fmt.Errorf("while deleting actor from DB: %w", err) + } + state.DeletedActor = deleted + return nil +} + +func (s *FinalizeDeletedStep) RetryBackoff() *wait.Backoff { return nil } diff --git a/cmd/ateapi/internal/controlapi/workflow_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_delete_test.go new file mode 100644 index 000000000..2751470fa --- /dev/null +++ b/cmd/ateapi/internal/controlapi/workflow_delete_test.go @@ -0,0 +1,140 @@ +// 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/agent-substrate/substrate/cmd/ateapi/internal/store/storetest" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func TestDeleteActorWorkflow_ExecutionPaths(t *testing.T) { + tests := []struct { + name string + seedStatus ateapipb.Actor_Status + wantErr bool + wantCode codes.Code + }{ + { + name: "delete suspended actor succeeds", + seedStatus: ateapipb.Actor_STATUS_SUSPENDED, + wantErr: false, + }, + { + name: "delete crashed actor succeeds", + seedStatus: ateapipb.Actor_STATUS_CRASHED, + wantErr: false, + }, + { + name: "delete deleting actor succeeds", + seedStatus: ateapipb.Actor_STATUS_DELETING, + wantErr: false, + }, + { + name: "delete running actor rejected", + seedStatus: ateapipb.Actor_STATUS_RUNNING, + wantErr: true, + wantCode: codes.FailedPrecondition, + }, + { + name: "delete paused actor rejected", + seedStatus: ateapipb.Actor_STATUS_PAUSED, + wantErr: true, + wantCode: codes.FailedPrecondition, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + st, cleanup := storetest.SetupTestStore(t) + defer cleanup() + w := newTestActorWorkflow(t, st, "ns", "tmpl1") + + actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} + seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", tc.seedStatus) + + deleted, err := w.DeleteActor(ctx, "team-a", "id1") + if tc.wantErr { + if got := status.Code(err); got != tc.wantCode { + t.Fatalf("status.Code(err) = %v, want %v (err: %v)", got, tc.wantCode, err) + } + } else { + if err != nil { + t.Fatalf("DeleteActor failed: %v", err) + } + if deleted == nil { + t.Fatalf("expected non-nil deleted actor") + } + if _, err := st.GetActor(ctx, actorRef); err == nil { + t.Errorf("expected actor to be deleted from store, but it still exists") + } + } + }) + } +} + +func TestDeleteSteps_CheckPrerequisite(t *testing.T) { + tests := []struct { + name string + step WorkflowStep[*DeleteInput, *DeleteState] + allowed map[ateapipb.Actor_Status]bool + }{ + { + name: "LoadActorForDeleteStep", + step: &LoadActorForDeleteStep{}, + allowed: nil, + }, + { + name: "MarkDeletingStep", + step: &MarkDeletingStep{}, + allowed: map[ateapipb.Actor_Status]bool{ + ateapipb.Actor_STATUS_SUSPENDED: true, + ateapipb.Actor_STATUS_CRASHED: true, + ateapipb.Actor_STATUS_DELETING: true, + }, + }, + { + name: "DeleteVolumesStep", + step: &DeleteVolumesStep{}, + allowed: map[ateapipb.Actor_Status]bool{ + ateapipb.Actor_STATUS_DELETING: true, + }, + }, + { + name: "FinalizeDeletedStep", + step: &FinalizeDeletedStep{}, + allowed: map[ateapipb.Actor_Status]bool{ + ateapipb.Actor_STATUS_DELETING: true, + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} + for _, st := range allActorStatuses { + err := tc.step.CheckPrerequisite(ctx, &DeleteInput{ActorRef: actorRef}, &DeleteState{Actor: &ateapipb.Actor{Status: st}}) + assertPrerequisiteResult(t, st, err, tc.allowed == nil || tc.allowed[st]) + } + }) + } +} diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index ebb955a37..aee116890 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -124,6 +124,54 @@ func (s *LoadActorForResumeStep) Execute(ctx context.Context, input *ResumeInput func (s *LoadActorForResumeStep) RetryBackoff() *wait.Backoff { return nil } +// CreateVolumesStep provisions any initial actor volumes that are in PENDING state. +type CreateVolumesStep struct { + store store.Interface +} + +func (s *CreateVolumesStep) Name() string { return "CreateVolumes" } + +func (s *CreateVolumesStep) IsComplete(ctx context.Context, input *ResumeInput, state *ResumeState) (bool, error) { + for _, vol := range state.Actor.GetActorVolumes() { + if vol.GetStatus() == ateapipb.ExternalVolume_STATUS_PENDING { + return false, nil + } + } + return true, nil +} + +func (s *CreateVolumesStep) CheckPrerequisite(ctx context.Context, input *ResumeInput, state *ResumeState) error { + if state.Actor == nil { + return fmt.Errorf("actor is required for CreateVolumesStep") + } + if state.ActorTemplate == nil { + return fmt.Errorf("actorTemplate is required for CreateVolumesStep") + } + return nil +} + +func (s *CreateVolumesStep) Execute(ctx context.Context, input *ResumeInput, state *ResumeState) error { + volumes, err := createActorVolumes(ctx, state.Actor.GetMetadata().GetUid(), state.ActorTemplate, state.Actor.GetActorVolumes()) + state.Actor.ActorVolumes = volumes + if err != nil { + // Even if volume creation failed, we still want to persist any updated volume state. + if updated, updateErr := s.store.UpdateActor(ctx, state.Actor, state.Actor.GetMetadata().GetVersion()); updateErr != nil { + slog.ErrorContext(ctx, "failed to update actor volumes on volume creation failure in resume", slog.Any("error", updateErr)) + } else { + state.Actor = updated + } + return err + } + updated, updateErr := s.store.UpdateActor(ctx, state.Actor, state.Actor.GetMetadata().GetVersion()) + if updateErr != nil { + return fmt.Errorf("while updating actor after volume creation: %w", updateErr) + } + state.Actor = updated + return nil +} + +func (s *CreateVolumesStep) RetryBackoff() *wait.Backoff { return nil } + type AssignWorkerStep struct { store store.Interface workerCache *workercache.Cache 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 e5ada1ff8..ae08e54bc 100644 --- a/cmd/ateapi/internal/store/ateredis/ateredis.go +++ b/cmd/ateapi/internal/store/ateredis/ateredis.go @@ -504,8 +504,7 @@ func (s *Persistence) DeleteActor(ctx context.Context, actorRef resources.ActorR 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 931949f94..ee00d8b61 100644 --- a/cmd/ateapi/internal/store/ateredis/ateredis_test.go +++ b/cmd/ateapi/internal/store/ateredis/ateredis_test.go @@ -414,8 +414,9 @@ 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: "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}, } @@ -1365,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 ae7c136a4..674895f0e 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_UNSPECIFIED ExternalVolume_Status = 0 + // Volume creation pending in the storage system. + ExternalVolume_STATUS_PENDING ExternalVolume_Status = 1 + // Volume successfully created in the storage system. + ExternalVolume_STATUS_CREATED ExternalVolume_Status = 2 + // Volume being deleted from the storage system. + ExternalVolume_STATUS_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_UNSPECIFIED", + 1: "STATUS_PENDING", + 2: "STATUS_CREATED", + 3: "STATUS_DELETING", } ExternalVolume_Status_value = map[string]int32{ - "PROVISIONING": 0, - "CREATED": 1, - "DELETING": 2, + "STATUS_UNSPECIFIED": 0, + "STATUS_PENDING": 1, + "STATUS_CREATED": 2, + "STATUS_DELETING": 3, } ) @@ -98,6 +104,7 @@ const ( Actor_STATUS_PAUSING Actor_Status = 5 Actor_STATUS_PAUSED Actor_Status = 6 Actor_STATUS_CRASHED Actor_Status = 7 + Actor_STATUS_DELETING Actor_Status = 8 ) // Enum value maps for Actor_Status. @@ -111,6 +118,7 @@ var ( 5: "STATUS_PAUSING", 6: "STATUS_PAUSED", 7: "STATUS_CRASHED", + 8: "STATUS_DELETING", } Actor_Status_value = map[string]int32{ "STATUS_UNSPECIFIED": 0, @@ -121,6 +129,7 @@ var ( "STATUS_PAUSING": 5, "STATUS_PAUSED": 6, "STATUS_CRASHED": 7, + "STATUS_DELETING": 8, } ) @@ -521,9 +530,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"` @@ -562,9 +572,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 "" } @@ -587,7 +597,7 @@ func (x *ExternalVolume) GetStatus() ExternalVolume_Status { if x != nil { return x.Status } - return ExternalVolume_PROVISIONING + return ExternalVolume_STATUS_UNSPECIFIED } type Actor struct { @@ -617,7 +627,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 } @@ -2408,17 +2418,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\"\x94\x02\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\"]\n" + + "\x06Status\x12\x16\n" + + "\x12STATUS_UNSPECIFIED\x10\x00\x12\x12\n" + + "\x0eSTATUS_PENDING\x10\x01\x12\x12\n" + + "\x0eSTATUS_CREATED\x10\x02\x12\x13\n" + + "\x0fSTATUS_DELETING\x10\x03\"\xd6\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" + @@ -2434,7 +2446,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\"\xc6\x01\n" + "\x06Status\x12\x16\n" + "\x12STATUS_UNSPECIFIED\x10\x00\x12\x13\n" + "\x0fSTATUS_RESUMING\x10\x01\x12\x12\n" + @@ -2443,7 +2455,8 @@ 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_DELETING\x10\b\"@\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 873f994e9..0384bec60 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_UNSPECIFIED = 0; + // Volume creation pending in the storage system. + STATUS_PENDING = 1; + // Volume successfully created in the storage system. + STATUS_CREATED = 2; + // Volume being deleted from the storage system. + STATUS_DELETING = 3; } Status status = 4; } @@ -148,6 +153,7 @@ message Actor { STATUS_PAUSING = 5; STATUS_PAUSED = 6; STATUS_CRASHED = 7; + STATUS_DELETING = 8; } Status status = 4; @@ -175,7 +181,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: