diff --git a/cmd/ateapi/internal/controlapi/delete_actor.go b/cmd/ateapi/internal/controlapi/delete_actor.go index 601e608c3..f98e5b2ef 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor.go +++ b/cmd/ateapi/internal/controlapi/delete_actor.go @@ -45,7 +45,7 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ actorRef := resources.ActorRefFromObjectRef(req.GetActor()) setSpanActorRefAttributes(ctx, actorRef) - deleted, err = s.actorWorkflow.DeleteActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName()) + deleted, err = s.actorWorkflow.DeleteActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName(), req.GetForce()) if err != nil { return nil, err } diff --git a/cmd/ateapi/internal/controlapi/delete_actor_test.go b/cmd/ateapi/internal/controlapi/delete_actor_test.go index fdd9f05ec..e8694b05c 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor_test.go +++ b/cmd/ateapi/internal/controlapi/delete_actor_test.go @@ -26,6 +26,8 @@ import ( "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/internal/volume" "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" ) @@ -207,3 +209,137 @@ func TestDeleteActor_MultipleVolumeDeletionFailures(t *testing.T) { t.Errorf("expected error message to contain both volume failure details, got: %v", errMsg) } } + +func TestDeleteActor_Force_Success(t *testing.T) { + ns := namespaceForTest("ns-delete-force-succ") + tc := setupTest(t, ns) + defer tc.cleanup() + createTemplate(t, tc, ns) + createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1") + + runningActor := &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{ + Atespace: testAtespace, + Name: "running-actor", + }, + Status: ateapipb.Actor_STATUS_RUNNING, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + WorkerAssignment: &ateapipb.WorkerAssignment{ + WorkerNamespace: ns, + WorkerPod: "worker-1", + WorkerPool: "pool1", + }, + } + if _, err := tc.persistence.CreateActor(context.Background(), runningActor); err != nil { + t.Fatalf("CreateActor: %v", err) + } + + worker, err := tc.persistence.GetWorker(context.Background(), ns, "pool1", "worker-1") + if err != nil { + t.Fatalf("GetWorker: %v", err) + } + worker.Assignment = &ateapipb.Assignment{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "running-actor"}, + } + if err := tc.persistence.UpdateWorker(context.Background(), worker, worker.Version); err != nil { + t.Fatalf("UpdateWorker: %v", err) + } + + deleted, err := tc.service.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "running-actor"}, + Force: true, + }) + if err != nil { + t.Fatalf("DeleteActor with force failed: %v", err) + } + if deleted.GetMetadata().GetName() != "running-actor" { + t.Errorf("deleted actor name = %q, want %q", deleted.GetMetadata().GetName(), "running-actor") + } + + // Verify actor is removed from store + if _, err := tc.persistence.GetActor(context.Background(), resources.ActorRef{Atespace: testAtespace, Name: "running-actor"}); err == nil { + t.Errorf("expected actor to be deleted from store, but it still exists") + } + + // Verify worker assignment is cleared + w, err := tc.persistence.GetWorker(context.Background(), ns, "pool1", "worker-1") + if err != nil { + t.Fatalf("GetWorker after delete failed: %v", err) + } + if w.Assignment != nil { + t.Errorf("expected worker assignment to be nil, got: %v", w.Assignment) + } + + // Verify atelet Terminate was called + if !tc.fakeAtelet.TerminateCalled { + t.Errorf("expected atelet Terminate to be called for force delete") + } +} + +func TestDeleteActor_Force_SuspendedAllowed(t *testing.T) { + ns := namespaceForTest("ns-delete-force-susp") + tc := setupTest(t, ns) + defer tc.cleanup() + createTemplate(t, tc, ns) + + suspendedActor := &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{ + Atespace: testAtespace, + Name: "suspended-actor", + }, + Status: ateapipb.Actor_STATUS_SUSPENDED, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + } + if _, err := tc.persistence.CreateActor(context.Background(), suspendedActor); err != nil { + t.Fatalf("CreateActor: %v", err) + } + + deleted, err := tc.service.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "suspended-actor"}, + Force: true, + }) + if err != nil { + t.Fatalf("expected DeleteActor with force on suspended actor to succeed, got: %v", err) + } + if deleted.GetMetadata().GetName() != "suspended-actor" { + t.Errorf("deleted.Name = %q, want %q", deleted.GetMetadata().GetName(), "suspended-actor") + } +} + +func TestDeleteActor_Force_WorkerNotFound(t *testing.T) { + ns := namespaceForTest("ns-delete-force-noworker") + 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", + WorkerAssignment: &ateapipb.WorkerAssignment{ + WorkerNamespace: ns, + WorkerPod: "non-existent-worker", + WorkerPool: "pool1", + }, + } + 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"}, + Force: true, + }) + if err == nil { + t.Fatalf("expected DeleteActor with force on non-existent worker to fail, but it succeeded") + } + if status.Code(err) != codes.NotFound { + t.Errorf("status.Code(err) = %v, want %v (err: %v)", status.Code(err), codes.NotFound, err) + } +} diff --git a/cmd/ateapi/internal/controlapi/functional_test.go b/cmd/ateapi/internal/controlapi/functional_test.go index c1f82ee72..3245731f0 100644 --- a/cmd/ateapi/internal/controlapi/functional_test.go +++ b/cmd/ateapi/internal/controlapi/functional_test.go @@ -180,6 +180,10 @@ type FakeAteletServer struct { RestoreRequest *ateletpb.RestoreRequest FailRestore error RestoreDelay time.Duration + + TerminateCalled bool + TerminateRequest *ateletpb.TerminateRequest + FailTerminate error } func (f *FakeAteletServer) Reset() { @@ -197,6 +201,23 @@ func (f *FakeAteletServer) Reset() { f.RestoreRequest = nil f.FailRestore = nil f.RestoreDelay = 0 + + f.TerminateCalled = false + f.TerminateRequest = nil + f.FailTerminate = nil +} + +func (f *FakeAteletServer) Terminate(ctx context.Context, req *ateletpb.TerminateRequest) (*ateletpb.TerminateResponse, error) { + f.Lock.Lock() + defer f.Lock.Unlock() + + f.TerminateCalled = true + f.TerminateRequest = proto.Clone(req).(*ateletpb.TerminateRequest) + if f.FailTerminate != nil { + return nil, f.FailTerminate + } + + return &ateletpb.TerminateResponse{}, nil } func (f *FakeAteletServer) Run(ctx context.Context, req *ateletpb.RunRequest) (*ateletpb.RunResponse, error) { @@ -3291,20 +3312,10 @@ func TestDeleteActor_Crashed(t *testing.T) { t.Fatalf("UpdateActor failed: %v", err) } - deleted, err := tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ - Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "id1"}, - }) - if err != nil { - t.Fatalf("DeleteActor of crashed actor failed: %v", err) - } - 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{ + _, err = tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "id1"}, }) - assertGrpcError(t, err, codes.NotFound, "Actor test-atespace/id1 not found") + assertGrpcError(t, err, codes.FailedPrecondition, "Actor test-atespace/id1 is not in a deletable status (status: STATUS_CRASHED)") } func TestDeleteActor_NotFound(t *testing.T) { @@ -3318,6 +3329,88 @@ func TestDeleteActor_NotFound(t *testing.T) { assertGrpcError(t, err, codes.NotFound, "Actor test-atespace/non-existent not found") } +func TestDeleteActor_Force(t *testing.T) { + ns := namespaceForTest("ns-delete-force") + tc := setupTest(t, ns) + defer tc.cleanup() + + createTemplate(t, tc, ns) + createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1") + + // 1. Create and resume actor to running status + _, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "force-actor"}, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + }}) + if err != nil { + t.Fatalf("CreateActor failed: %v", err) + } + + resumeResp, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "force-actor"}, + }) + if err != nil { + t.Fatalf("ResumeActor failed: %v", err) + } + if resumeResp.GetActor().GetStatus() != ateapipb.Actor_STATUS_RUNNING { + t.Fatalf("expected status STATUS_RUNNING, got %v", resumeResp.GetActor().GetStatus()) + } + + // 2. Force delete running actor + deleted, err := tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "force-actor"}, + Force: true, + }) + if err != nil { + t.Fatalf("DeleteActor with force failed: %v", err) + } + if deleted.GetMetadata().GetName() != "force-actor" { + t.Errorf("deleted actor name = %q, want %q", deleted.GetMetadata().GetName(), "force-actor") + } + + // Verify actor is removed from store + _, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "force-actor"}, + }) + assertGrpcError(t, err, codes.NotFound, "Actor test-atespace/force-actor not found") + + // Verify atelet Terminate was invoked + if !tc.fakeAtelet.TerminateCalled { + t.Errorf("expected atelet Terminate to have been called") + } + + // Verify worker is unassigned + w, err := tc.persistence.GetWorker(context.Background(), ns, "pool1", "worker-1") + if err != nil { + t.Fatalf("GetWorker failed: %v", err) + } + if w.Assignment != nil { + t.Errorf("expected worker assignment to be nil, got: %v", w.Assignment) + } + + // 3. Verify force delete on suspended actor succeeds + _, err = tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "susp-actor"}, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + }}) + if err != nil { + t.Fatalf("CreateActor failed: %v", err) + } + + deletedSusp, err := tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "susp-actor"}, + Force: true, + }) + if err != nil { + t.Fatalf("expected DeleteActor with force on suspended actor to succeed, got %v", err) + } + if deletedSusp.GetMetadata().GetName() != "susp-actor" { + t.Errorf("deleted actor name = %q, want %q", deletedSusp.GetMetadata().GetName(), "susp-actor") + } +} + func assertGrpcErrorRegex(t *testing.T, err error, wantCode codes.Code, wantMsg string) { t.Helper() fn := func(got string) (string, bool) { diff --git a/cmd/ateapi/internal/controlapi/workflow.go b/cmd/ateapi/internal/controlapi/workflow.go index a62e954f6..0de67a9b3 100644 --- a/cmd/ateapi/internal/controlapi/workflow.go +++ b/cmd/ateapi/internal/controlapi/workflow.go @@ -281,10 +281,11 @@ func (w *ActorWorkflow) PauseActor(ctx context.Context, actorRef resources.Actor } // DeleteActor executes the workflow to delete an actor. Idempotent. -func (w *ActorWorkflow) DeleteActor(ctx context.Context, atespace, name string) (*ateapipb.Actor, error) { +func (w *ActorWorkflow) DeleteActor(ctx context.Context, atespace, name string, force bool) (*ateapipb.Actor, error) { actorRef := resources.ActorRef{Atespace: atespace, Name: name} input := &DeleteInput{ ActorRef: actorRef, + Force: force, } state := &DeleteState{} @@ -295,7 +296,11 @@ func (w *ActorWorkflow) DeleteActor(ctx context.Context, atespace, name string) defer lock.Close() steps := []WorkflowStep[*DeleteInput, *DeleteState]{ - &LoadActorForDeleteStep{store: w.store}, + &LoadActorForDeleteStep{store: w.store, actorTemplateLister: w.actorTemplateLister}, + &MarkTerminatingStep{store: w.store}, + &CallAteletTerminateStep{store: w.store, dialer: w.dialer}, + &DetachVolumesForDeleteStep{store: w.store}, + &ReleaseWorkerStep{store: w.store}, &MarkDeletingStep{store: w.store}, &DeleteVolumesStep{store: w.store}, &FinalizeDeletedStep{store: w.store}, diff --git a/cmd/ateapi/internal/controlapi/workflow_delete.go b/cmd/ateapi/internal/controlapi/workflow_delete.go index 761f46c78..20699830a 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete.go @@ -18,28 +18,36 @@ import ( "context" "errors" "fmt" + "log/slog" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" + "github.com/agent-substrate/substrate/internal/proto/ateletpb" "github.com/agent-substrate/substrate/internal/resources" + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" + listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + k8serrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/util/wait" ) // DeleteInput holds the immutable parameters requested by the client. type DeleteInput struct { ActorRef resources.ActorRef + Force bool } // DeleteState holds the mutable state loaded and modified during execution. type DeleteState struct { - Actor *ateapipb.Actor - DeletedActor *ateapipb.Actor + Actor *ateapipb.Actor + ActorTemplate *atev1alpha1.ActorTemplate + DeletedActor *ateapipb.Actor } type LoadActorForDeleteStep struct { - store store.Interface + store store.Interface + actorTemplateLister listersv1alpha1.ActorTemplateLister } func (s *LoadActorForDeleteStep) Name() string { return "LoadActorForDelete" } @@ -59,11 +67,71 @@ func (s *LoadActorForDeleteStep) Execute(ctx context.Context, input *DeleteInput return fmt.Errorf("while fetching actor: %w", err) } state.Actor = actor + + if s.actorTemplateLister != nil && actor.GetActorTemplateNamespace() != "" && actor.GetActorTemplateName() != "" { + tmpl, err := s.actorTemplateLister.ActorTemplates(actor.GetActorTemplateNamespace()).Get(actor.GetActorTemplateName()) + if err != nil && !k8serrors.IsNotFound(err) { + return fmt.Errorf("while fetching actor template: %w", err) + } + state.ActorTemplate = tmpl + } return nil } func (s *LoadActorForDeleteStep) RetryBackoff() *wait.Backoff { return nil } +type MarkTerminatingStep struct { + store store.Interface +} + +func (s *MarkTerminatingStep) Name() string { return "MarkTerminating" } + +func (s *MarkTerminatingStep) IsComplete(ctx context.Context, input *DeleteInput, state *DeleteState) (bool, error) { + st := state.Actor.GetStatus() + return st == ateapipb.Actor_STATUS_TERMINATING || st == ateapipb.Actor_STATUS_DELETING, nil +} + +func (s *MarkTerminatingStep) CheckPrerequisite(ctx context.Context, input *DeleteInput, state *DeleteState) error { + if input.Force { + switch state.Actor.GetStatus() { + case ateapipb.Actor_STATUS_RUNNING, + ateapipb.Actor_STATUS_RESUMING, + ateapipb.Actor_STATUS_SUSPENDING, + ateapipb.Actor_STATUS_PAUSING, + ateapipb.Actor_STATUS_PAUSED, + ateapipb.Actor_STATUS_CRASHED, + ateapipb.Actor_STATUS_TERMINATING, + ateapipb.Actor_STATUS_DELETING, + ateapipb.Actor_STATUS_SUSPENDED: + return nil + default: + return status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status (status: %v)", input.ActorRef, state.Actor.GetStatus()) + } + } + switch state.Actor.GetStatus() { + case ateapipb.Actor_STATUS_SUSPENDED, + ateapipb.Actor_STATUS_DELETING: + return nil + default: + return status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status (status: %v)", input.ActorRef, state.Actor.GetStatus()) + } +} + +func (s *MarkTerminatingStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { + state.Actor.Status = ateapipb.Actor_STATUS_TERMINATING + 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 TERMINATING: %w", err) + } + state.Actor = updated + return nil +} + +func (s *MarkTerminatingStep) RetryBackoff() *wait.Backoff { return nil } + type MarkDeletingStep struct { store store.Interface } @@ -73,10 +141,9 @@ func (s *MarkDeletingStep) IsComplete(ctx context.Context, input *DeleteInput, s 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()) + st := state.Actor.GetStatus() + if st != ateapipb.Actor_STATUS_TERMINATING && st != ateapipb.Actor_STATUS_DELETING { + return status.Errorf(codes.FailedPrecondition, "MarkDeletingStep prerequisite not met for Actor: %s (got: %v, want %s or %s)", input.ActorRef, st, ateapipb.Actor_STATUS_TERMINATING, ateapipb.Actor_STATUS_DELETING) } return nil } @@ -98,6 +165,151 @@ func (s *MarkDeletingStep) Execute(ctx context.Context, input *DeleteInput, stat func (s *MarkDeletingStep) RetryBackoff() *wait.Backoff { return nil } +type CallAteletTerminateStep struct { + store store.Interface + dialer *AteletDialer +} + +func (s *CallAteletTerminateStep) Name() string { return "CallAteletTerminate" } + +func (s *CallAteletTerminateStep) IsComplete(ctx context.Context, input *DeleteInput, state *DeleteState) (bool, error) { + return state.Actor.GetStatus() == ateapipb.Actor_STATUS_DELETING, nil +} + +func (s *CallAteletTerminateStep) CheckPrerequisite(ctx context.Context, input *DeleteInput, state *DeleteState) error { + st := state.Actor.GetStatus() + if st != ateapipb.Actor_STATUS_TERMINATING { + return status.Errorf(codes.FailedPrecondition, "CallAteletTerminateStep prerequisite not met for Actor: %s (got: %v, want %s)", input.ActorRef, st, ateapipb.Actor_STATUS_TERMINATING) + } + if state.ActorTemplate == nil { + return status.Errorf(codes.FailedPrecondition, "actor template %s/%s not found for actor %s", state.Actor.GetActorTemplateNamespace(), state.Actor.GetActorTemplateName(), input.ActorRef) + } + return nil +} + +func (s *CallAteletTerminateStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { + // TODO: if a workload has crashed, it's still possible there are resources on the node that need + // to be cleaned up. + assignment := state.Actor.GetWorkerAssignment() + if assignment == nil { + slog.InfoContext(ctx, "actor has no worker assignment, skipping CallAteletTerminateStep", slog.Any("actor", input.ActorRef)) + return nil + } + + workerPodNs := assignment.GetWorkerNamespace() + workerPodName := assignment.GetWorkerPod() + + conn, err := s.dialer.DialForWorker(workerPodNs, workerPodName) + if err != nil { + if errors.Is(err, ErrWorkerPodNotFound) { + return status.Errorf(codes.NotFound, "worker pod %s/%s not found: %v", workerPodNs, workerPodName, err) + } + return fmt.Errorf("while connecting to worker pod %s/%s: %w", workerPodNs, workerPodName, err) + } + + client := ateletpb.NewAteomHerderClient(conn) + + workloadSpec, err := workloadSpecFromActorTemplate(state.ActorTemplate, state.Actor) + if err != nil { + return err + } + + req := &ateletpb.TerminateRequest{ + TargetAteomUid: assignment.GetWorkerPodUid(), + Atespace: state.Actor.GetMetadata().GetAtespace(), + ActorName: state.Actor.GetMetadata().GetName(), + ActorUid: state.Actor.GetMetadata().GetUid(), + ActorTemplateNamespace: state.Actor.GetActorTemplateNamespace(), + ActorTemplateName: state.Actor.GetActorTemplateName(), + Spec: workloadSpec, + } + + if _, err := client.Terminate(ctx, req); err != nil { + return fmt.Errorf("while terminating actor on atelet: %w", err) + } + + return nil +} + +func (s *CallAteletTerminateStep) RetryBackoff() *wait.Backoff { return nil } + +type DetachVolumesForDeleteStep struct { + store store.Interface +} + +func (s *DetachVolumesForDeleteStep) Name() string { return "DetachVolumesForDelete" } + +func (s *DetachVolumesForDeleteStep) IsComplete(ctx context.Context, input *DeleteInput, state *DeleteState) (bool, error) { + return state.Actor.GetStatus() == ateapipb.Actor_STATUS_DELETING, nil +} + +func (s *DetachVolumesForDeleteStep) CheckPrerequisite(ctx context.Context, input *DeleteInput, state *DeleteState) error { + st := state.Actor.GetStatus() + if st != ateapipb.Actor_STATUS_TERMINATING { + return status.Errorf(codes.FailedPrecondition, "DetachVolumesForDeleteStep prerequisite not met for Actor: %s (got: %v, want %s)", input.ActorRef, st, ateapipb.Actor_STATUS_TERMINATING) + } + if state.ActorTemplate == nil { + return status.Errorf(codes.FailedPrecondition, "actor template %s/%s not found for actor %s", state.Actor.GetActorTemplateNamespace(), state.Actor.GetActorTemplateName(), input.ActorRef) + } + return nil +} + +func (s *DetachVolumesForDeleteStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { + return detachActorVolumes(ctx, s.store, state.Actor, state.ActorTemplate, "delete") +} + +func (s *DetachVolumesForDeleteStep) RetryBackoff() *wait.Backoff { return nil } + +type ReleaseWorkerStep struct { + store store.Interface +} + +func (s *ReleaseWorkerStep) Name() string { return "ReleaseWorker" } + +func (s *ReleaseWorkerStep) IsComplete(ctx context.Context, input *DeleteInput, state *DeleteState) (bool, error) { + if state.Actor.GetStatus() == ateapipb.Actor_STATUS_DELETING { + return true, nil + } + return state.Actor.GetWorkerAssignment() == nil, nil +} + +func (s *ReleaseWorkerStep) CheckPrerequisite(ctx context.Context, input *DeleteInput, state *DeleteState) error { + st := state.Actor.GetStatus() + if st != ateapipb.Actor_STATUS_TERMINATING { + return status.Errorf(codes.FailedPrecondition, "ReleaseWorkerStep prerequisite not met for Actor: %s (got: %v, want %s)", input.ActorRef, st, ateapipb.Actor_STATUS_TERMINATING) + } + return nil +} + +func (s *ReleaseWorkerStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { + latestActor, err := s.store.GetActor(ctx, input.ActorRef) + if err != nil { + return err + } + + if latestActor.GetWorkerAssignment() != nil { + if _, err := releaseWorker(ctx, s.store, latestActor); err != nil { + return err + } + + latestActor, err = s.store.GetActor(ctx, input.ActorRef) + if err != nil { + return err + } + + // TODO: can this be done inside releaseWorker()? + latestActor.LocalSnapshotInfo = nil + updatedActor, err := s.store.UpdateActor(ctx, latestActor, latestActor.GetMetadata().GetVersion()) + if err != nil { + return err + } + state.Actor = updatedActor + } + return nil +} + +func (s *ReleaseWorkerStep) RetryBackoff() *wait.Backoff { return nil } + type DeleteVolumesStep struct { store store.Interface } diff --git a/cmd/ateapi/internal/controlapi/workflow_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_delete_test.go index 2751470fa..2dc068a10 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete_test.go @@ -20,6 +20,7 @@ import ( "github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest" "github.com/agent-substrate/substrate/internal/resources" + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -29,36 +30,67 @@ func TestDeleteActorWorkflow_ExecutionPaths(t *testing.T) { tests := []struct { name string seedStatus ateapipb.Actor_Status + force bool wantErr bool wantCode codes.Code }{ { name: "delete suspended actor succeeds", seedStatus: ateapipb.Actor_STATUS_SUSPENDED, + force: false, wantErr: false, }, { - name: "delete crashed actor succeeds", + name: "delete crashed actor rejected when not forced", seedStatus: ateapipb.Actor_STATUS_CRASHED, - wantErr: false, + force: false, + wantErr: true, + wantCode: codes.FailedPrecondition, }, { name: "delete deleting actor succeeds", seedStatus: ateapipb.Actor_STATUS_DELETING, + force: false, wantErr: false, }, { - name: "delete running actor rejected", + name: "delete running actor rejected when not forced", seedStatus: ateapipb.Actor_STATUS_RUNNING, + force: false, wantErr: true, wantCode: codes.FailedPrecondition, }, { - name: "delete paused actor rejected", + name: "delete paused actor rejected when not forced", seedStatus: ateapipb.Actor_STATUS_PAUSED, + force: false, wantErr: true, wantCode: codes.FailedPrecondition, }, + { + name: "force delete suspended actor succeeds", + seedStatus: ateapipb.Actor_STATUS_SUSPENDED, + force: true, + wantErr: false, + }, + { + name: "force delete running actor succeeds", + seedStatus: ateapipb.Actor_STATUS_RUNNING, + force: true, + wantErr: false, + }, + { + name: "force delete paused actor succeeds", + seedStatus: ateapipb.Actor_STATUS_PAUSED, + force: true, + wantErr: false, + }, + { + name: "force delete crashed actor succeeds", + seedStatus: ateapipb.Actor_STATUS_CRASHED, + force: true, + wantErr: false, + }, } for _, tc := range tests { @@ -71,7 +103,7 @@ func TestDeleteActorWorkflow_ExecutionPaths(t *testing.T) { 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") + deleted, err := w.DeleteActor(ctx, "team-a", "id1", tc.force) 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) @@ -95,6 +127,7 @@ func TestDeleteSteps_CheckPrerequisite(t *testing.T) { tests := []struct { name string step WorkflowStep[*DeleteInput, *DeleteState] + force bool allowed map[ateapipb.Actor_Status]bool }{ { @@ -103,14 +136,59 @@ func TestDeleteSteps_CheckPrerequisite(t *testing.T) { allowed: nil, }, { - name: "MarkDeletingStep", - step: &MarkDeletingStep{}, + name: "MarkTerminatingStep_Standard", + step: &MarkTerminatingStep{}, + force: false, allowed: map[ateapipb.Actor_Status]bool{ ateapipb.Actor_STATUS_SUSPENDED: true, - ateapipb.Actor_STATUS_CRASHED: true, ateapipb.Actor_STATUS_DELETING: true, }, }, + { + name: "MarkTerminatingStep_Force", + step: &MarkTerminatingStep{}, + force: true, + allowed: map[ateapipb.Actor_Status]bool{ + ateapipb.Actor_STATUS_RUNNING: true, + ateapipb.Actor_STATUS_RESUMING: true, + ateapipb.Actor_STATUS_SUSPENDING: true, + ateapipb.Actor_STATUS_PAUSING: true, + ateapipb.Actor_STATUS_PAUSED: true, + ateapipb.Actor_STATUS_CRASHED: true, + ateapipb.Actor_STATUS_TERMINATING: true, + ateapipb.Actor_STATUS_DELETING: true, + ateapipb.Actor_STATUS_SUSPENDED: true, + }, + }, + { + name: "MarkDeletingStep", + step: &MarkDeletingStep{}, + allowed: map[ateapipb.Actor_Status]bool{ + ateapipb.Actor_STATUS_TERMINATING: true, + ateapipb.Actor_STATUS_DELETING: true, + }, + }, + { + name: "CallAteletTerminateStep", + step: &CallAteletTerminateStep{}, + allowed: map[ateapipb.Actor_Status]bool{ + ateapipb.Actor_STATUS_TERMINATING: true, + }, + }, + { + name: "DetachVolumesForDeleteStep", + step: &DetachVolumesForDeleteStep{}, + allowed: map[ateapipb.Actor_Status]bool{ + ateapipb.Actor_STATUS_TERMINATING: true, + }, + }, + { + name: "ReleaseWorkerStep", + step: &ReleaseWorkerStep{}, + allowed: map[ateapipb.Actor_Status]bool{ + ateapipb.Actor_STATUS_TERMINATING: true, + }, + }, { name: "DeleteVolumesStep", step: &DeleteVolumesStep{}, @@ -132,7 +210,7 @@ func TestDeleteSteps_CheckPrerequisite(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}}) + err := tc.step.CheckPrerequisite(ctx, &DeleteInput{ActorRef: actorRef, Force: tc.force}, &DeleteState{Actor: &ateapipb.Actor{Status: st}, ActorTemplate: &atev1alpha1.ActorTemplate{}}) assertPrerequisiteResult(t, st, err, tc.allowed == nil || tc.allowed[st]) } }) diff --git a/cmd/ateapi/internal/controlapi/workflow_suspend.go b/cmd/ateapi/internal/controlapi/workflow_suspend.go index cf63b15c7..e6ddac759 100644 --- a/cmd/ateapi/internal/controlapi/workflow_suspend.go +++ b/cmd/ateapi/internal/controlapi/workflow_suspend.go @@ -235,26 +235,9 @@ func (s *FinalizeSuspendedStep) Execute(ctx context.Context, input *SuspendInput } // 1. Free the worker (if it hasn't been freed yet) - if assignment := latestActor.GetWorkerAssignment(); assignment != nil { - workerPod := assignment.GetWorkerPod() - - worker, err := s.store.GetWorker(ctx, assignment.GetWorkerNamespace(), assignment.GetWorkerPool(), workerPod) - if err != nil { - if !errors.Is(err, store.ErrNotFound) { - return fmt.Errorf("while getting worker for release: %w", err) - } - slog.WarnContext(ctx, "Worker already gone during finalize suspend, skipping release", "worker", workerPod) - } else { - // Only free it if it still belongs to us - if wass := worker.Assignment; wass != nil { - if wass.GetActorUid() == latestActor.GetMetadata().GetUid() { - worker.Assignment = nil - err = s.store.UpdateWorker(ctx, worker, worker.Version) - if err != nil { - return err - } - } - } + if latestActor.GetWorkerAssignment() != nil { + if _, err := releaseWorker(ctx, s.store, latestActor); err != nil { + return err } // 2. Clear the actor's assignment, now that the worker is freed diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 95a4e16c4..3df9c0e70 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -748,6 +748,45 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) return &ateletpb.RestoreResponse{}, nil } +// Terminate terminates any running workload on ateom, unmounts external volumes, +// and resets actor directories on the node. +func (s *AteomHerder) Terminate(ctx context.Context, req *ateletpb.TerminateRequest) (*ateletpb.TerminateResponse, error) { + if err := validateTerminateRequest(req); err != nil { + return nil, status.Errorf(codes.InvalidArgument, "%v", err) + } + + actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} + actorUID := req.GetActorUid() + + client, err := s.dialAteom(ctx, req.GetTargetAteomUid()) + if err != nil { + // TODO: handle case where the target ateom is no longer available + return nil, fmt.Errorf("failed to dial ateom for terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) + } + if _, err := client.TerminateWorkload(ctx, &ateompb.TerminateWorkloadRequest{ + Atespace: req.GetAtespace(), + ActorName: req.GetActorName(), + ActorUid: req.GetActorUid(), + ActorTemplateNamespace: req.GetActorTemplateNamespace(), + ActorTemplateName: req.GetActorTemplateName(), + Spec: buildAteomWorkloadSpec(req.GetSpec()), + }); err != nil { + return nil, fmt.Errorf("failed calling ateom.TerminateWorkload (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) + } + + // Unmount external volumes + if err := s.unmountExternalVolumes(ctx, actorUID, req.GetSpec().GetVolumes()); err != nil { + return nil, fmt.Errorf("failed to unmount external volumes during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) + } + + // Reset actor directories on the node + if err := resetActorDirs(actorUID); err != nil { + return nil, fmt.Errorf("failed to reset actor directories during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err) + } + + return &ateletpb.TerminateResponse{}, nil +} + func (s *AteomHerder) copyLocalCheckpoint(ctx context.Context, snapshotPrefix string, srcDir, dstDir string, files []string) error { for _, fileName := range files { if ctx.Err() != nil { @@ -1175,6 +1214,30 @@ func validateRestoreRequest(req *ateletpb.RestoreRequest) error { return nil } +func validateTerminateRequest(req *ateletpb.TerminateRequest) error { + var errs field.ErrorList + errs = append(errs, resources.ValidateResourceName(req.GetAtespace(), field.NewPath("atespace"))...) + errs = append(errs, resources.ValidateResourceName(req.GetActorName(), field.NewPath("actor_name"))...) + errs = append(errs, resources.ValidateResourceName(req.GetActorUid(), field.NewPath("actor_uid"))...) + for _, msg := range content.IsDNS1123Label(req.GetActorTemplateNamespace()) { + errs = append(errs, field.Invalid(field.NewPath("actor_template_namespace"), req.GetActorTemplateNamespace(), msg)) + } + for _, msg := range content.IsDNS1123Subdomain(req.GetActorTemplateName()) { + errs = append(errs, field.Invalid(field.NewPath("actor_template_name"), req.GetActorTemplateName(), msg)) + } + if len(errs) > 0 { + return errs.ToAggregate() + } + if err := resources.ValidateAteomUID(req.GetTargetAteomUid()); err != nil { + return err + } + names := make([]string, 0, len(req.GetSpec().GetContainers())) + for _, ctr := range req.GetSpec().GetContainers() { + names = append(names, ctr.GetName()) + } + return resources.ValidateContainerNames(names) +} + func validateSnapshotScope(scope ateletpb.SnapshotScope) error { switch scope { case ateletpb.SnapshotScope_SNAPSHOT_SCOPE_FULL, diff --git a/cmd/atelet/main_test.go b/cmd/atelet/main_test.go index 88058cffe..915e0fd71 100644 --- a/cmd/atelet/main_test.go +++ b/cmd/atelet/main_test.go @@ -584,6 +584,41 @@ func TestRPCBoundariesReject(t *testing.T) { }) wantInvalidArgument(t, "Restore", err) }) + t.Run("Terminate", func(t *testing.T) { + const okTargetAteomUID = "123e4567-e89b-12d3-a456-426614174001" + t.Run("invalid ateom UID", func(t *testing.T) { + _, err := s.Terminate(ctx, &ateletpb.TerminateRequest{ + Atespace: okAtespace, ActorName: okID, + ActorUid: okActorUID, ActorTemplateNamespace: "default", ActorTemplateName: "template", + TargetAteomUid: badUID, Spec: okSpec, + }) + wantInvalidArgument(t, "Terminate", err) + }) + t.Run("missing template namespace", func(t *testing.T) { + _, err := s.Terminate(ctx, &ateletpb.TerminateRequest{ + Atespace: okAtespace, ActorName: okID, + ActorUid: okActorUID, ActorTemplateName: "template", + TargetAteomUid: okTargetAteomUID, Spec: okSpec, + }) + wantInvalidArgument(t, "Terminate", err) + }) + t.Run("missing template name", func(t *testing.T) { + _, err := s.Terminate(ctx, &ateletpb.TerminateRequest{ + Atespace: okAtespace, ActorName: okID, + ActorUid: okActorUID, ActorTemplateNamespace: "default", + TargetAteomUid: okTargetAteomUID, Spec: okSpec, + }) + wantInvalidArgument(t, "Terminate", err) + }) + t.Run("missing target ateom UID", func(t *testing.T) { + _, err := s.Terminate(ctx, &ateletpb.TerminateRequest{ + Atespace: okAtespace, ActorName: okID, + ActorUid: okActorUID, ActorTemplateNamespace: "default", ActorTemplateName: "template", + Spec: okSpec, + }) + wantInvalidArgument(t, "Terminate", err) + }) + }) } func TestBuildAteomWorkloadSpecForwardsReadyz(t *testing.T) { diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index 752acf2d3..110f92003 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -403,26 +403,7 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec // control server for state/delete calls. Keep this as best-effort cleanup: // atelet resets the actor runsc, bundle, pidfile, and checkpoint // directories after uploading the snapshot. - if err := rcmd.cleanupContainersAfterCheckpoint(ctx, req.GetSpec().GetContainers()); err != nil { - slog.WarnContext(ctx, "Failed to clean up runsc containers after checkpoint", - "actor", actorRef, - "actorUID", req.GetActorUid(), - "err", err) - } - - // Detach the overlay rootfs mounts before atelet wipes the bundle dirs - // (deleting a bundle out from under a live mount in this namespace would - // leave the mount orphaned until the pod restarts). Best-effort, same as - // the container cleanup above. - if err := imagecache.UnmountAllUnder(ateompath.OCIBundleDir(req.GetActorUid())); err != nil { - slog.WarnContext(ctx, "Failed to unmount bundle rootfs overlays after checkpoint", - "actorUID", req.GetActorUid(), - "err", err) - } - - if err := ateomnet.CleanupActorNetwork(ctx, s.interiorNetNS); err != nil { - slog.WarnContext(ctx, "Failed to clean up actor network after checkpoint", slog.Any("err", err)) - } + s.terminateWorkload(ctx, actorRef, req.GetActorUid(), req.GetRunscPath(), req.GetSpec().GetContainers()) // Report exactly the files runsc wrote so atelet ships precisely this set // (checkpoint.img plus any pages images), rather than a hardcoded list. @@ -594,6 +575,48 @@ func (s *AteomService) RestoreWorkload(ctx context.Context, req *ateompb.Restore return &ateompb.RestoreWorkloadResponse{}, nil } +func (s *AteomService) TerminateWorkload(ctx context.Context, req *ateompb.TerminateWorkloadRequest) (*ateompb.TerminateWorkloadResponse, error) { + s.lock.Lock() + defer s.lock.Unlock() + + actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} + + // TODO: consider if this should be a hard failure for the terminate case + s.terminateWorkload(ctx, actorRef, req.GetActorUid(), req.GetRunscPath(), req.GetSpec().GetContainers()) + + s.actorLogger.EmitLifecycleLog("Actor terminated", actorRef, req.GetActorUid(), req.GetActorTemplateNamespace(), req.GetActorTemplateName()) + + return &ateompb.TerminateWorkloadResponse{}, nil +} + +func (s *AteomService) terminateWorkload(ctx context.Context, actorRef resources.ActorRef, actorUID, runscPath string, containers []*ateompb.Container) { + if err := s.deactivateActorNetworking(ctx); err != nil { + slog.WarnContext(ctx, "Failed to deactivate actor networking during terminate", slog.Any("err", err)) + } + + rcmd := &runsc{ + path: runscPath, + actorUID: actorUID, + } + + if err := rcmd.cleanupContainersAfterCheckpoint(ctx, containers); err != nil { + slog.WarnContext(ctx, "Failed to clean up runsc containers during terminate", + "actor", actorRef, + "actorUID", actorUID, + "err", err) + } + + if err := imagecache.UnmountAllUnder(ateompath.OCIBundleDir(actorUID)); err != nil { + slog.WarnContext(ctx, "Failed to unmount bundle rootfs overlays during terminate", + "actorUID", actorUID, + "err", err) + } + + if err := ateomnet.CleanupActorNetwork(ctx, s.interiorNetNS); err != nil { + slog.WarnContext(ctx, "Failed to clean up actor network during terminate", slog.Any("err", err)) + } +} + func (s *AteomService) activateActorNetworking(atespace, actorName string, actorVersion int64, egressGatewayAddress string) error { var egressClient atunnel.EgressDialer if s.atunnelEgress != nil && egressGatewayAddress != "" { diff --git a/cmd/ateom-microvm/checkpoint.go b/cmd/ateom-microvm/checkpoint.go index a63682f0d..db57f1d48 100644 --- a/cmd/ateom-microvm/checkpoint.go +++ b/cmd/ateom-microvm/checkpoint.go @@ -154,14 +154,8 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec // Tear down: the actor returns to "available". Best-effort; the snapshot is // already on disk for atelet to ship. tTeardown := time.Now() - s.teardownActor(ctx, actorUID, ra, client) + s.terminateWorkload(ctx, actorUID) dTeardown := time.Since(tTeardown) - delete(s.running, actorUID) - - // Tear down the per-activation actor network. - if err := ateomnet.CleanupActorNetwork(ctx, s.interiorNetNS); err != nil { - slog.WarnContext(ctx, "Failed to clean up actor network after checkpoint", slog.Any("err", err)) - } s.actorLogger.EmitLifecycleLog("Actor checkpointed", actorRef, actorUID, templateNS, templateName) slog.InfoContext(ctx, "Actor checkpointed", slog.String("id", actorUID), slog.Any("snapshot_files", snapshotFiles), @@ -290,3 +284,39 @@ func (s *AteomService) teardownActor(ctx context.Context, id string, ra *running slog.WarnContext(ctx, "Failed to unmount bundle rootfs overlays", slog.String("actorUID", id), slog.Any("err", err)) } } + +// TerminateWorkload stops the running actor, tears down its VMM, and cleans up +// networking and overlays. +func (s *AteomService) TerminateWorkload(ctx context.Context, req *ateompb.TerminateWorkloadRequest) (*ateompb.TerminateWorkloadResponse, error) { + s.lock.Lock() + defer s.lock.Unlock() + + actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} + actorUID := req.GetActorUid() + + s.terminateWorkload(ctx, actorUID) + + s.actorLogger.EmitLifecycleLog("Actor terminated", actorRef, actorUID, req.GetActorTemplateNamespace(), req.GetActorTemplateName()) + + return &ateompb.TerminateWorkloadResponse{}, nil +} + +func (s *AteomService) terminateWorkload(ctx context.Context, actorUID string) { + if err := s.deactivateActorNetworking(ctx); err != nil { + slog.WarnContext(ctx, "Failed to deactivate actor networking during terminate", slog.Any("err", err)) + } + + ra := s.running[actorUID] + chSocket := kata.CLHSocketPath(actorUID) + if ra != nil && ra.apiSocket != "" { + chSocket = ra.apiSocket + } + client := ch.NewClient(chSocket) + + s.teardownActor(ctx, actorUID, ra, client) + delete(s.running, actorUID) + + if err := ateomnet.CleanupActorNetwork(ctx, s.interiorNetNS); err != nil { + slog.WarnContext(ctx, "Failed to clean up actor network during terminate", slog.Any("err", err)) + } +} diff --git a/demos/claude-code-multiplex/ui/server.go b/demos/claude-code-multiplex/ui/server.go index 0df15b8f5..ffb3b30e7 100644 --- a/demos/claude-code-multiplex/ui/server.go +++ b/demos/claude-code-multiplex/ui/server.go @@ -214,6 +214,10 @@ func actorStatusString(s ateapipb.Actor_Status) string { return "Suspending" case ateapipb.Actor_STATUS_SUSPENDED: return "Suspended" + case ateapipb.Actor_STATUS_TERMINATING: + return "Terminating" + case ateapipb.Actor_STATUS_DELETING: + return "Deleting" default: return "?" } diff --git a/internal/e2e/suites/demo/demo_test.go b/internal/e2e/suites/demo/demo_test.go index ca893a879..098fb3582 100644 --- a/internal/e2e/suites/demo/demo_test.go +++ b/internal/e2e/suites/demo/demo_test.go @@ -29,6 +29,8 @@ import ( "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/fieldmaskpb" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -71,6 +73,10 @@ func TestActorLifecycle(t *testing.T) { name: "SuspendResumeActor", f: suspendActor, }, + { + name: "ForceDeleteActor", + f: forceDeleteActor, + }, } for _, tc := range tests { @@ -796,6 +802,99 @@ func suspendActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj return nil } +func forceDeleteActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj *e2e.Namespace, at *v1alpha1.ActorTemplate) error { + actorName := "force-delete-actor-" + nsObj.Name + + // 1. Creating an actor + t.Logf("Creating Actor %q...", actorName) + if _, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: demoAtespace, Name: actorName}, + ActorTemplateNamespace: nsObj.Name, + ActorTemplateName: at.Name, + }}); err != nil { + t.Fatalf("failed to create Actor: %v", err) + } + waitForActorStatus(ctx, t, clients, actorName, ateapipb.Actor_STATUS_SUSPENDED) + + // Verify that force delete on a SUSPENDED actor succeeds + t.Logf("Attempting force delete on suspended Actor %q (should succeed)...", actorName) + if _, err := clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: demoAtespace, Name: actorName}, + Force: true, + }); err != nil { + t.Fatalf("expected force delete on suspended actor to succeed, got %v", err) + } + + // 2. Re-creating and Resuming the actor + t.Logf("Re-creating Actor %q...", actorName) + if _, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: demoAtespace, Name: actorName}, + ActorTemplateNamespace: nsObj.Name, + ActorTemplateName: at.Name, + }}); err != nil { + t.Fatalf("failed to create Actor: %v", err) + } + waitForActorStatus(ctx, t, clients, actorName, ateapipb.Actor_STATUS_SUSPENDED) + + t.Logf("Resuming Actor %q...", actorName) + if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: demoAtespace, Name: actorName}, + }); err != nil { + t.Fatalf("failed to resume Actor: %v", err) + } + waitForActorStatus(ctx, t, clients, actorName, ateapipb.Actor_STATUS_RUNNING) + + // 3. Call the actor to ensure workload is actively serving on worker + resp, err := callActor(t, resources.ActorRef{Atespace: demoAtespace, Name: actorName}) + if err != nil { + t.Fatalf("failed to call actor: %v", err) + } + validateCounterResponse(t, resp, "after creation", 1, 1) + + // 4. Attempt standard DeleteActor without force on running actor (should fail with FailedPrecondition) + t.Logf("Attempting standard delete on running Actor %q (should fail)...", actorName) + if _, err := clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: demoAtespace, Name: actorName}, + Force: false, + }); status.Code(err) != codes.FailedPrecondition { + t.Fatalf("expected FailedPrecondition when standard deleting running actor, got %v", err) + } + + // 5. Force delete the running actor + t.Logf("Force deleting running Actor %q...", actorName) + if _, err := clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: demoAtespace, Name: actorName}, + Force: true, + }); err != nil { + t.Fatalf("failed to force delete Actor: %v", err) + } + + // 6. Verify deletion in store + if _, err := clients.SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: demoAtespace, Name: actorName}, + }); status.Code(err) != codes.NotFound { + t.Fatalf("expected actor %q to be NotFound after force delete, got err: %v", actorName, err) + } + + // 7. Verify actor name can be immediately reused + t.Logf("Re-creating Actor %q to verify name reuse...", actorName) + if _, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: demoAtespace, Name: actorName}, + ActorTemplateNamespace: nsObj.Name, + ActorTemplateName: at.Name, + }}); err != nil { + t.Fatalf("failed to recreate Actor: %v", err) + } + defer func() { + clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: demoAtespace, Name: actorName}, + }) + }() + waitForActorStatus(ctx, t, clients, actorName, ateapipb.Actor_STATUS_SUSPENDED) + + return nil +} + func createActorTemplateInternal(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj *e2e.Namespace, name string, onCommit, onPause v1alpha1.SnapshotScope, fromData v1alpha1.ResumeSource, modifyTemplate func(*v1alpha1.ActorTemplate)) (*v1alpha1.ActorTemplate, error) { env, err := e2e.CheckEnv("BUCKET_NAME", "KO_DOCKER_REPO") if err != nil { diff --git a/internal/proto/ateletpb/atelet.pb.go b/internal/proto/ateletpb/atelet.pb.go index 5ed205914..4e7adfd3b 100644 --- a/internal/proto/ateletpb/atelet.pb.go +++ b/internal/proto/ateletpb/atelet.pb.go @@ -200,6 +200,134 @@ func (SnapshotScope) EnumDescriptor() ([]byte, []int) { return file_atelet_proto_rawDescGZIP(), []int{2} } +type TerminateRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + TargetAteomUid string `protobuf:"bytes,1,opt,name=target_ateom_uid,json=targetAteomUid,proto3" json:"target_ateom_uid,omitempty"` + Atespace string `protobuf:"bytes,2,opt,name=atespace,proto3" json:"atespace,omitempty"` + ActorName string `protobuf:"bytes,3,opt,name=actor_name,json=actorName,proto3" json:"actor_name,omitempty"` + ActorUid string `protobuf:"bytes,4,opt,name=actor_uid,json=actorUid,proto3" json:"actor_uid,omitempty"` + ActorTemplateNamespace string `protobuf:"bytes,5,opt,name=actor_template_namespace,json=actorTemplateNamespace,proto3" json:"actor_template_namespace,omitempty"` + ActorTemplateName string `protobuf:"bytes,6,opt,name=actor_template_name,json=actorTemplateName,proto3" json:"actor_template_name,omitempty"` + Spec *WorkloadSpec `protobuf:"bytes,7,opt,name=spec,proto3" json:"spec,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *TerminateRequest) Reset() { + *x = TerminateRequest{} + mi := &file_atelet_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *TerminateRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*TerminateRequest) ProtoMessage() {} + +func (x *TerminateRequest) ProtoReflect() protoreflect.Message { + mi := &file_atelet_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use TerminateRequest.ProtoReflect.Descriptor instead. +func (*TerminateRequest) Descriptor() ([]byte, []int) { + return file_atelet_proto_rawDescGZIP(), []int{0} +} + +func (x *TerminateRequest) GetTargetAteomUid() string { + if x != nil { + return x.TargetAteomUid + } + return "" +} + +func (x *TerminateRequest) GetAtespace() string { + if x != nil { + return x.Atespace + } + return "" +} + +func (x *TerminateRequest) GetActorName() string { + if x != nil { + return x.ActorName + } + return "" +} + +func (x *TerminateRequest) GetActorUid() string { + if x != nil { + return x.ActorUid + } + return "" +} + +func (x *TerminateRequest) GetActorTemplateNamespace() string { + if x != nil { + return x.ActorTemplateNamespace + } + return "" +} + +func (x *TerminateRequest) GetActorTemplateName() string { + if x != nil { + return x.ActorTemplateName + } + return "" +} + +func (x *TerminateRequest) GetSpec() *WorkloadSpec { + if x != nil { + return x.Spec + } + return nil +} + +type TerminateResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *TerminateResponse) Reset() { + *x = TerminateResponse{} + mi := &file_atelet_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *TerminateResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*TerminateResponse) ProtoMessage() {} + +func (x *TerminateResponse) ProtoReflect() protoreflect.Message { + mi := &file_atelet_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use TerminateResponse.ProtoReflect.Descriptor instead. +func (*TerminateResponse) Descriptor() ([]byte, []int) { + return file_atelet_proto_rawDescGZIP(), []int{1} +} + type RunRequest struct { state protoimpl.MessageState `protogen:"open.v1"` TargetAteomUid string `protobuf:"bytes,1,opt,name=target_ateom_uid,json=targetAteomUid,proto3" json:"target_ateom_uid,omitempty"` @@ -219,7 +347,7 @@ type RunRequest struct { func (x *RunRequest) Reset() { *x = RunRequest{} - mi := &file_atelet_proto_msgTypes[0] + mi := &file_atelet_proto_msgTypes[2] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -231,7 +359,7 @@ func (x *RunRequest) String() string { func (*RunRequest) ProtoMessage() {} func (x *RunRequest) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[0] + mi := &file_atelet_proto_msgTypes[2] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -244,7 +372,7 @@ func (x *RunRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use RunRequest.ProtoReflect.Descriptor instead. func (*RunRequest) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{0} + return file_atelet_proto_rawDescGZIP(), []int{2} } func (x *RunRequest) GetTargetAteomUid() string { @@ -317,7 +445,7 @@ type AssetFile struct { func (x *AssetFile) Reset() { *x = AssetFile{} - mi := &file_atelet_proto_msgTypes[1] + mi := &file_atelet_proto_msgTypes[3] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -329,7 +457,7 @@ func (x *AssetFile) String() string { func (*AssetFile) ProtoMessage() {} func (x *AssetFile) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[1] + mi := &file_atelet_proto_msgTypes[3] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -342,7 +470,7 @@ func (x *AssetFile) ProtoReflect() protoreflect.Message { // Deprecated: Use AssetFile.ProtoReflect.Descriptor instead. func (*AssetFile) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{1} + return file_atelet_proto_rawDescGZIP(), []int{3} } func (x *AssetFile) GetUrl() string { @@ -370,7 +498,7 @@ type ArchAssets struct { func (x *ArchAssets) Reset() { *x = ArchAssets{} - mi := &file_atelet_proto_msgTypes[2] + mi := &file_atelet_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -382,7 +510,7 @@ func (x *ArchAssets) String() string { func (*ArchAssets) ProtoMessage() {} func (x *ArchAssets) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[2] + mi := &file_atelet_proto_msgTypes[4] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -395,7 +523,7 @@ func (x *ArchAssets) ProtoReflect() protoreflect.Message { // Deprecated: Use ArchAssets.ProtoReflect.Descriptor instead. func (*ArchAssets) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{2} + return file_atelet_proto_rawDescGZIP(), []int{4} } func (x *ArchAssets) GetFiles() map[string]*AssetFile { @@ -420,7 +548,7 @@ type SandboxAssets struct { func (x *SandboxAssets) Reset() { *x = SandboxAssets{} - mi := &file_atelet_proto_msgTypes[3] + mi := &file_atelet_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -432,7 +560,7 @@ func (x *SandboxAssets) String() string { func (*SandboxAssets) ProtoMessage() {} func (x *SandboxAssets) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[3] + mi := &file_atelet_proto_msgTypes[5] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -445,7 +573,7 @@ func (x *SandboxAssets) ProtoReflect() protoreflect.Message { // Deprecated: Use SandboxAssets.ProtoReflect.Descriptor instead. func (*SandboxAssets) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{3} + return file_atelet_proto_rawDescGZIP(), []int{5} } func (x *SandboxAssets) GetSandboxClass() string { @@ -474,7 +602,7 @@ type WorkloadSpec struct { func (x *WorkloadSpec) Reset() { *x = WorkloadSpec{} - mi := &file_atelet_proto_msgTypes[4] + mi := &file_atelet_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -486,7 +614,7 @@ func (x *WorkloadSpec) String() string { func (*WorkloadSpec) ProtoMessage() {} func (x *WorkloadSpec) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[4] + mi := &file_atelet_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -499,7 +627,7 @@ func (x *WorkloadSpec) ProtoReflect() protoreflect.Message { // Deprecated: Use WorkloadSpec.ProtoReflect.Descriptor instead. func (*WorkloadSpec) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{4} + return file_atelet_proto_rawDescGZIP(), []int{6} } func (x *WorkloadSpec) GetContainers() []*Container { @@ -531,7 +659,7 @@ type DurableDirVolume struct { func (x *DurableDirVolume) Reset() { *x = DurableDirVolume{} - mi := &file_atelet_proto_msgTypes[5] + mi := &file_atelet_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -543,7 +671,7 @@ func (x *DurableDirVolume) String() string { func (*DurableDirVolume) ProtoMessage() {} func (x *DurableDirVolume) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[5] + mi := &file_atelet_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -556,7 +684,7 @@ func (x *DurableDirVolume) ProtoReflect() protoreflect.Message { // Deprecated: Use DurableDirVolume.ProtoReflect.Descriptor instead. func (*DurableDirVolume) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{5} + return file_atelet_proto_rawDescGZIP(), []int{7} } type ExternalVolumeSource struct { @@ -569,7 +697,7 @@ type ExternalVolumeSource struct { func (x *ExternalVolumeSource) Reset() { *x = ExternalVolumeSource{} - mi := &file_atelet_proto_msgTypes[6] + mi := &file_atelet_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -581,7 +709,7 @@ func (x *ExternalVolumeSource) String() string { func (*ExternalVolumeSource) ProtoMessage() {} func (x *ExternalVolumeSource) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[6] + mi := &file_atelet_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -594,7 +722,7 @@ func (x *ExternalVolumeSource) ProtoReflect() protoreflect.Message { // Deprecated: Use ExternalVolumeSource.ProtoReflect.Descriptor instead. func (*ExternalVolumeSource) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{6} + return file_atelet_proto_rawDescGZIP(), []int{8} } func (x *ExternalVolumeSource) GetStorageVolumeId() string { @@ -626,7 +754,7 @@ type Volume struct { func (x *Volume) Reset() { *x = Volume{} - mi := &file_atelet_proto_msgTypes[7] + mi := &file_atelet_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -638,7 +766,7 @@ func (x *Volume) String() string { func (*Volume) ProtoMessage() {} func (x *Volume) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[7] + mi := &file_atelet_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -651,7 +779,7 @@ func (x *Volume) ProtoReflect() protoreflect.Message { // Deprecated: Use Volume.ProtoReflect.Descriptor instead. func (*Volume) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{7} + return file_atelet_proto_rawDescGZIP(), []int{9} } func (x *Volume) GetName() string { @@ -719,7 +847,7 @@ type VolumeMount struct { func (x *VolumeMount) Reset() { *x = VolumeMount{} - mi := &file_atelet_proto_msgTypes[8] + mi := &file_atelet_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -731,7 +859,7 @@ func (x *VolumeMount) String() string { func (*VolumeMount) ProtoMessage() {} func (x *VolumeMount) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[8] + mi := &file_atelet_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -744,7 +872,7 @@ func (x *VolumeMount) ProtoReflect() protoreflect.Message { // Deprecated: Use VolumeMount.ProtoReflect.Descriptor instead. func (*VolumeMount) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{8} + return file_atelet_proto_rawDescGZIP(), []int{10} } func (x *VolumeMount) GetName() string { @@ -776,7 +904,7 @@ type Container struct { func (x *Container) Reset() { *x = Container{} - mi := &file_atelet_proto_msgTypes[9] + mi := &file_atelet_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -788,7 +916,7 @@ func (x *Container) String() string { func (*Container) ProtoMessage() {} func (x *Container) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[9] + mi := &file_atelet_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -801,7 +929,7 @@ func (x *Container) ProtoReflect() protoreflect.Message { // Deprecated: Use Container.ProtoReflect.Descriptor instead. func (*Container) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{9} + return file_atelet_proto_rawDescGZIP(), []int{11} } func (x *Container) GetName() string { @@ -863,7 +991,7 @@ type EnvEntry struct { func (x *EnvEntry) Reset() { *x = EnvEntry{} - mi := &file_atelet_proto_msgTypes[10] + mi := &file_atelet_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -875,7 +1003,7 @@ func (x *EnvEntry) String() string { func (*EnvEntry) ProtoMessage() {} func (x *EnvEntry) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[10] + mi := &file_atelet_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -888,7 +1016,7 @@ func (x *EnvEntry) ProtoReflect() protoreflect.Message { // Deprecated: Use EnvEntry.ProtoReflect.Descriptor instead. func (*EnvEntry) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{10} + return file_atelet_proto_rawDescGZIP(), []int{12} } func (x *EnvEntry) GetName() string { @@ -919,7 +1047,7 @@ type Readyz struct { func (x *Readyz) Reset() { *x = Readyz{} - mi := &file_atelet_proto_msgTypes[11] + mi := &file_atelet_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -931,7 +1059,7 @@ func (x *Readyz) String() string { func (*Readyz) ProtoMessage() {} func (x *Readyz) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[11] + mi := &file_atelet_proto_msgTypes[13] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -944,7 +1072,7 @@ func (x *Readyz) ProtoReflect() protoreflect.Message { // Deprecated: Use Readyz.ProtoReflect.Descriptor instead. func (*Readyz) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{11} + return file_atelet_proto_rawDescGZIP(), []int{13} } func (x *Readyz) GetHttpGet() *HTTPGetAction { @@ -974,7 +1102,7 @@ type HTTPGetAction struct { func (x *HTTPGetAction) Reset() { *x = HTTPGetAction{} - mi := &file_atelet_proto_msgTypes[12] + mi := &file_atelet_proto_msgTypes[14] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -986,7 +1114,7 @@ func (x *HTTPGetAction) String() string { func (*HTTPGetAction) ProtoMessage() {} func (x *HTTPGetAction) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[12] + mi := &file_atelet_proto_msgTypes[14] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -999,7 +1127,7 @@ func (x *HTTPGetAction) ProtoReflect() protoreflect.Message { // Deprecated: Use HTTPGetAction.ProtoReflect.Descriptor instead. func (*HTTPGetAction) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{12} + return file_atelet_proto_rawDescGZIP(), []int{14} } func (x *HTTPGetAction) GetPath() string { @@ -1024,7 +1152,7 @@ type RunResponse struct { func (x *RunResponse) Reset() { *x = RunResponse{} - mi := &file_atelet_proto_msgTypes[13] + mi := &file_atelet_proto_msgTypes[15] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1036,7 +1164,7 @@ func (x *RunResponse) String() string { func (*RunResponse) ProtoMessage() {} func (x *RunResponse) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[13] + mi := &file_atelet_proto_msgTypes[15] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1049,7 +1177,7 @@ func (x *RunResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use RunResponse.ProtoReflect.Descriptor instead. func (*RunResponse) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{13} + return file_atelet_proto_rawDescGZIP(), []int{15} } type LocalCheckpointConfiguration struct { @@ -1063,7 +1191,7 @@ type LocalCheckpointConfiguration struct { func (x *LocalCheckpointConfiguration) Reset() { *x = LocalCheckpointConfiguration{} - mi := &file_atelet_proto_msgTypes[14] + mi := &file_atelet_proto_msgTypes[16] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1075,7 +1203,7 @@ func (x *LocalCheckpointConfiguration) String() string { func (*LocalCheckpointConfiguration) ProtoMessage() {} func (x *LocalCheckpointConfiguration) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[14] + mi := &file_atelet_proto_msgTypes[16] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1088,7 +1216,7 @@ func (x *LocalCheckpointConfiguration) ProtoReflect() protoreflect.Message { // Deprecated: Use LocalCheckpointConfiguration.ProtoReflect.Descriptor instead. func (*LocalCheckpointConfiguration) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{14} + return file_atelet_proto_rawDescGZIP(), []int{16} } func (x *LocalCheckpointConfiguration) GetSnapshotPrefix() string { @@ -1117,7 +1245,7 @@ type ExternalCheckpointConfiguration struct { func (x *ExternalCheckpointConfiguration) Reset() { *x = ExternalCheckpointConfiguration{} - mi := &file_atelet_proto_msgTypes[15] + mi := &file_atelet_proto_msgTypes[17] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1129,7 +1257,7 @@ func (x *ExternalCheckpointConfiguration) String() string { func (*ExternalCheckpointConfiguration) ProtoMessage() {} func (x *ExternalCheckpointConfiguration) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[15] + mi := &file_atelet_proto_msgTypes[17] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1142,7 +1270,7 @@ func (x *ExternalCheckpointConfiguration) ProtoReflect() protoreflect.Message { // Deprecated: Use ExternalCheckpointConfiguration.ProtoReflect.Descriptor instead. func (*ExternalCheckpointConfiguration) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{15} + return file_atelet_proto_rawDescGZIP(), []int{17} } func (x *ExternalCheckpointConfiguration) GetSnapshotUriPrefix() string { @@ -1180,7 +1308,7 @@ type CheckpointRequest struct { func (x *CheckpointRequest) Reset() { *x = CheckpointRequest{} - mi := &file_atelet_proto_msgTypes[16] + mi := &file_atelet_proto_msgTypes[18] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1192,7 +1320,7 @@ func (x *CheckpointRequest) String() string { func (*CheckpointRequest) ProtoMessage() {} func (x *CheckpointRequest) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[16] + mi := &file_atelet_proto_msgTypes[18] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1205,7 +1333,7 @@ func (x *CheckpointRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use CheckpointRequest.ProtoReflect.Descriptor instead. func (*CheckpointRequest) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{16} + return file_atelet_proto_rawDescGZIP(), []int{18} } func (x *CheckpointRequest) GetTargetAteomUid() string { @@ -1320,7 +1448,7 @@ type CheckpointResponse struct { func (x *CheckpointResponse) Reset() { *x = CheckpointResponse{} - mi := &file_atelet_proto_msgTypes[17] + mi := &file_atelet_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1332,7 +1460,7 @@ func (x *CheckpointResponse) String() string { func (*CheckpointResponse) ProtoMessage() {} func (x *CheckpointResponse) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[17] + mi := &file_atelet_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1345,7 +1473,7 @@ func (x *CheckpointResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use CheckpointResponse.ProtoReflect.Descriptor instead. func (*CheckpointResponse) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{17} + return file_atelet_proto_rawDescGZIP(), []int{19} } type RestoreRequest struct { @@ -1383,7 +1511,7 @@ type RestoreRequest struct { func (x *RestoreRequest) Reset() { *x = RestoreRequest{} - mi := &file_atelet_proto_msgTypes[18] + mi := &file_atelet_proto_msgTypes[20] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1395,7 +1523,7 @@ func (x *RestoreRequest) String() string { func (*RestoreRequest) ProtoMessage() {} func (x *RestoreRequest) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[18] + mi := &file_atelet_proto_msgTypes[20] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1408,7 +1536,7 @@ func (x *RestoreRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use RestoreRequest.ProtoReflect.Descriptor instead. func (*RestoreRequest) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{18} + return file_atelet_proto_rawDescGZIP(), []int{20} } func (x *RestoreRequest) GetTargetAteomUid() string { @@ -1530,7 +1658,7 @@ type RestoreResponse struct { func (x *RestoreResponse) Reset() { *x = RestoreResponse{} - mi := &file_atelet_proto_msgTypes[19] + mi := &file_atelet_proto_msgTypes[21] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1542,7 +1670,7 @@ func (x *RestoreResponse) String() string { func (*RestoreResponse) ProtoMessage() {} func (x *RestoreResponse) ProtoReflect() protoreflect.Message { - mi := &file_atelet_proto_msgTypes[19] + mi := &file_atelet_proto_msgTypes[21] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1555,14 +1683,24 @@ func (x *RestoreResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use RestoreResponse.ProtoReflect.Descriptor instead. func (*RestoreResponse) Descriptor() ([]byte, []int) { - return file_atelet_proto_rawDescGZIP(), []int{19} + return file_atelet_proto_rawDescGZIP(), []int{21} } var File_atelet_proto protoreflect.FileDescriptor const file_atelet_proto_rawDesc = "" + "\n" + - "\fatelet.proto\x12\x06atelet\"\xe0\x02\n" + + "\fatelet.proto\x12\x06atelet\"\xa8\x02\n" + + "\x10TerminateRequest\x12(\n" + + "\x10target_ateom_uid\x18\x01 \x01(\tR\x0etargetAteomUid\x12\x1a\n" + + "\batespace\x18\x02 \x01(\tR\batespace\x12\x1d\n" + + "\n" + + "actor_name\x18\x03 \x01(\tR\tactorName\x12\x1b\n" + + "\tactor_uid\x18\x04 \x01(\tR\bactorUid\x128\n" + + "\x18actor_template_namespace\x18\x05 \x01(\tR\x16actorTemplateNamespace\x12.\n" + + "\x13actor_template_name\x18\x06 \x01(\tR\x11actorTemplateName\x12(\n" + + "\x04spec\x18\a \x01(\v2\x14.atelet.WorkloadSpecR\x04spec\"\x13\n" + + "\x11TerminateResponse\"\xe0\x02\n" + "\n" + "RunRequest\x12(\n" + "\x10target_ateom_uid\x18\x01 \x01(\tR\x0etargetAteomUid\x12\x1a\n" + @@ -1681,12 +1819,13 @@ const file_atelet_proto_rawDesc = "" + "\x1aSNAPSHOT_SCOPE_UNSPECIFIED\x10\x00\x12\x17\n" + "\x13SNAPSHOT_SCOPE_FULL\x10\x01\x12\x17\n" + "\x13SNAPSHOT_SCOPE_DATA\x10\x02\x12!\n" + - "\x1dSNAPSHOT_SCOPE_DATA_ON_GOLDEN\x10\x032\xc4\x01\n" + + "\x1dSNAPSHOT_SCOPE_DATA_ON_GOLDEN\x10\x032\x88\x02\n" + "\vAteomHerder\x120\n" + "\x03Run\x12\x12.atelet.RunRequest\x1a\x13.atelet.RunResponse\"\x00\x12E\n" + "\n" + "Checkpoint\x12\x19.atelet.CheckpointRequest\x1a\x1a.atelet.CheckpointResponse\"\x00\x12<\n" + - "\aRestore\x12\x16.atelet.RestoreRequest\x1a\x17.atelet.RestoreResponse\"\x00B>ZZ atelet.WorkloadSpec - 6, // 1: atelet.RunRequest.sandbox_assets:type_name -> atelet.SandboxAssets - 23, // 2: atelet.ArchAssets.files:type_name -> atelet.ArchAssets.FilesEntry - 24, // 3: atelet.SandboxAssets.assets:type_name -> atelet.SandboxAssets.AssetsEntry - 12, // 4: atelet.WorkloadSpec.containers:type_name -> atelet.Container - 10, // 5: atelet.WorkloadSpec.volumes:type_name -> atelet.Volume - 0, // 6: atelet.Volume.type:type_name -> atelet.VolumeType - 8, // 7: atelet.Volume.durable_dir:type_name -> atelet.DurableDirVolume - 9, // 8: atelet.Volume.external:type_name -> atelet.ExternalVolumeSource - 13, // 9: atelet.Container.env:type_name -> atelet.EnvEntry - 14, // 10: atelet.Container.readyz:type_name -> atelet.Readyz - 11, // 11: atelet.Container.volume_mounts:type_name -> atelet.VolumeMount - 15, // 12: atelet.Readyz.http_get:type_name -> atelet.HTTPGetAction - 7, // 13: atelet.CheckpointRequest.spec:type_name -> atelet.WorkloadSpec - 1, // 14: atelet.CheckpointRequest.type:type_name -> atelet.CheckpointType - 17, // 15: atelet.CheckpointRequest.local_config:type_name -> atelet.LocalCheckpointConfiguration - 18, // 16: atelet.CheckpointRequest.external_config:type_name -> atelet.ExternalCheckpointConfiguration - 2, // 17: atelet.CheckpointRequest.scope:type_name -> atelet.SnapshotScope - 7, // 18: atelet.RestoreRequest.spec:type_name -> atelet.WorkloadSpec - 1, // 19: atelet.RestoreRequest.type:type_name -> atelet.CheckpointType - 17, // 20: atelet.RestoreRequest.local_config:type_name -> atelet.LocalCheckpointConfiguration - 18, // 21: atelet.RestoreRequest.external_config:type_name -> atelet.ExternalCheckpointConfiguration - 2, // 22: atelet.RestoreRequest.scope:type_name -> atelet.SnapshotScope - 4, // 23: atelet.ArchAssets.FilesEntry.value:type_name -> atelet.AssetFile - 5, // 24: atelet.SandboxAssets.AssetsEntry.value:type_name -> atelet.ArchAssets - 3, // 25: atelet.AteomHerder.Run:input_type -> atelet.RunRequest - 19, // 26: atelet.AteomHerder.Checkpoint:input_type -> atelet.CheckpointRequest - 21, // 27: atelet.AteomHerder.Restore:input_type -> atelet.RestoreRequest - 16, // 28: atelet.AteomHerder.Run:output_type -> atelet.RunResponse - 20, // 29: atelet.AteomHerder.Checkpoint:output_type -> atelet.CheckpointResponse - 22, // 30: atelet.AteomHerder.Restore:output_type -> atelet.RestoreResponse - 28, // [28:31] is the sub-list for method output_type - 25, // [25:28] is the sub-list for method input_type - 25, // [25:25] is the sub-list for extension type_name - 25, // [25:25] is the sub-list for extension extendee - 0, // [0:25] is the sub-list for field type_name + 9, // 0: atelet.TerminateRequest.spec:type_name -> atelet.WorkloadSpec + 9, // 1: atelet.RunRequest.spec:type_name -> atelet.WorkloadSpec + 8, // 2: atelet.RunRequest.sandbox_assets:type_name -> atelet.SandboxAssets + 25, // 3: atelet.ArchAssets.files:type_name -> atelet.ArchAssets.FilesEntry + 26, // 4: atelet.SandboxAssets.assets:type_name -> atelet.SandboxAssets.AssetsEntry + 14, // 5: atelet.WorkloadSpec.containers:type_name -> atelet.Container + 12, // 6: atelet.WorkloadSpec.volumes:type_name -> atelet.Volume + 0, // 7: atelet.Volume.type:type_name -> atelet.VolumeType + 10, // 8: atelet.Volume.durable_dir:type_name -> atelet.DurableDirVolume + 11, // 9: atelet.Volume.external:type_name -> atelet.ExternalVolumeSource + 15, // 10: atelet.Container.env:type_name -> atelet.EnvEntry + 16, // 11: atelet.Container.readyz:type_name -> atelet.Readyz + 13, // 12: atelet.Container.volume_mounts:type_name -> atelet.VolumeMount + 17, // 13: atelet.Readyz.http_get:type_name -> atelet.HTTPGetAction + 9, // 14: atelet.CheckpointRequest.spec:type_name -> atelet.WorkloadSpec + 1, // 15: atelet.CheckpointRequest.type:type_name -> atelet.CheckpointType + 19, // 16: atelet.CheckpointRequest.local_config:type_name -> atelet.LocalCheckpointConfiguration + 20, // 17: atelet.CheckpointRequest.external_config:type_name -> atelet.ExternalCheckpointConfiguration + 2, // 18: atelet.CheckpointRequest.scope:type_name -> atelet.SnapshotScope + 9, // 19: atelet.RestoreRequest.spec:type_name -> atelet.WorkloadSpec + 1, // 20: atelet.RestoreRequest.type:type_name -> atelet.CheckpointType + 19, // 21: atelet.RestoreRequest.local_config:type_name -> atelet.LocalCheckpointConfiguration + 20, // 22: atelet.RestoreRequest.external_config:type_name -> atelet.ExternalCheckpointConfiguration + 2, // 23: atelet.RestoreRequest.scope:type_name -> atelet.SnapshotScope + 6, // 24: atelet.ArchAssets.FilesEntry.value:type_name -> atelet.AssetFile + 7, // 25: atelet.SandboxAssets.AssetsEntry.value:type_name -> atelet.ArchAssets + 5, // 26: atelet.AteomHerder.Run:input_type -> atelet.RunRequest + 21, // 27: atelet.AteomHerder.Checkpoint:input_type -> atelet.CheckpointRequest + 23, // 28: atelet.AteomHerder.Restore:input_type -> atelet.RestoreRequest + 3, // 29: atelet.AteomHerder.Terminate:input_type -> atelet.TerminateRequest + 18, // 30: atelet.AteomHerder.Run:output_type -> atelet.RunResponse + 22, // 31: atelet.AteomHerder.Checkpoint:output_type -> atelet.CheckpointResponse + 24, // 32: atelet.AteomHerder.Restore:output_type -> atelet.RestoreResponse + 4, // 33: atelet.AteomHerder.Terminate:output_type -> atelet.TerminateResponse + 30, // [30:34] is the sub-list for method output_type + 26, // [26:30] is the sub-list for method input_type + 26, // [26:26] is the sub-list for extension type_name + 26, // [26:26] is the sub-list for extension extendee + 0, // [0:26] is the sub-list for field type_name } func init() { file_atelet_proto_init() } @@ -1773,15 +1917,15 @@ func file_atelet_proto_init() { if File_atelet_proto != nil { return } - file_atelet_proto_msgTypes[7].OneofWrappers = []any{ + file_atelet_proto_msgTypes[9].OneofWrappers = []any{ (*Volume_DurableDir)(nil), (*Volume_External)(nil), } - file_atelet_proto_msgTypes[16].OneofWrappers = []any{ + file_atelet_proto_msgTypes[18].OneofWrappers = []any{ (*CheckpointRequest_LocalConfig)(nil), (*CheckpointRequest_ExternalConfig)(nil), } - file_atelet_proto_msgTypes[18].OneofWrappers = []any{ + file_atelet_proto_msgTypes[20].OneofWrappers = []any{ (*RestoreRequest_LocalConfig)(nil), (*RestoreRequest_ExternalConfig)(nil), } @@ -1791,7 +1935,7 @@ func file_atelet_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_atelet_proto_rawDesc), len(file_atelet_proto_rawDesc)), NumEnums: 3, - NumMessages: 22, + NumMessages: 24, NumExtensions: 0, NumServices: 1, }, diff --git a/internal/proto/ateletpb/atelet.proto b/internal/proto/ateletpb/atelet.proto index 1aab6cc3b..3684a395c 100644 --- a/internal/proto/ateletpb/atelet.proto +++ b/internal/proto/ateletpb/atelet.proto @@ -30,6 +30,26 @@ service AteomHerder { // Restore restores a workload from checkpoint onto an ateom. rpc Restore(RestoreRequest) returns (RestoreResponse) {} + + // Terminate tells atelet to terminate/kill any running workload for an actor, + // unmount its volumes, and clean up actor state on the node. + rpc Terminate(TerminateRequest) returns (TerminateResponse) {} +} + +message TerminateRequest { + string target_ateom_uid = 1; + + string atespace = 2; + string actor_name = 3; + string actor_uid = 4; + + string actor_template_namespace = 5; + string actor_template_name = 6; + + WorkloadSpec spec = 7; +} + +message TerminateResponse { } message RunRequest { diff --git a/internal/proto/ateletpb/atelet_grpc.pb.go b/internal/proto/ateletpb/atelet_grpc.pb.go index 4f05a878d..c395f37fe 100644 --- a/internal/proto/ateletpb/atelet_grpc.pb.go +++ b/internal/proto/ateletpb/atelet_grpc.pb.go @@ -36,6 +36,7 @@ const ( AteomHerder_Run_FullMethodName = "/atelet.AteomHerder/Run" AteomHerder_Checkpoint_FullMethodName = "/atelet.AteomHerder/Checkpoint" AteomHerder_Restore_FullMethodName = "/atelet.AteomHerder/Restore" + AteomHerder_Terminate_FullMethodName = "/atelet.AteomHerder/Terminate" ) // AteomHerderClient is the client API for AteomHerder service. @@ -51,6 +52,9 @@ type AteomHerderClient interface { Checkpoint(ctx context.Context, in *CheckpointRequest, opts ...grpc.CallOption) (*CheckpointResponse, error) // Restore restores a workload from checkpoint onto an ateom. Restore(ctx context.Context, in *RestoreRequest, opts ...grpc.CallOption) (*RestoreResponse, error) + // Terminate tells atelet to terminate/kill any running workload for an actor, + // unmount its volumes, and clean up actor state on the node. + Terminate(ctx context.Context, in *TerminateRequest, opts ...grpc.CallOption) (*TerminateResponse, error) } type ateomHerderClient struct { @@ -91,6 +95,16 @@ func (c *ateomHerderClient) Restore(ctx context.Context, in *RestoreRequest, opt return out, nil } +func (c *ateomHerderClient) Terminate(ctx context.Context, in *TerminateRequest, opts ...grpc.CallOption) (*TerminateResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(TerminateResponse) + err := c.cc.Invoke(ctx, AteomHerder_Terminate_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // AteomHerderServer is the server API for AteomHerder service. // All implementations must embed UnimplementedAteomHerderServer // for forward compatibility. @@ -104,6 +118,9 @@ type AteomHerderServer interface { Checkpoint(context.Context, *CheckpointRequest) (*CheckpointResponse, error) // Restore restores a workload from checkpoint onto an ateom. Restore(context.Context, *RestoreRequest) (*RestoreResponse, error) + // Terminate tells atelet to terminate/kill any running workload for an actor, + // unmount its volumes, and clean up actor state on the node. + Terminate(context.Context, *TerminateRequest) (*TerminateResponse, error) mustEmbedUnimplementedAteomHerderServer() } @@ -123,6 +140,9 @@ func (UnimplementedAteomHerderServer) Checkpoint(context.Context, *CheckpointReq func (UnimplementedAteomHerderServer) Restore(context.Context, *RestoreRequest) (*RestoreResponse, error) { return nil, status.Error(codes.Unimplemented, "method Restore not implemented") } +func (UnimplementedAteomHerderServer) Terminate(context.Context, *TerminateRequest) (*TerminateResponse, error) { + return nil, status.Error(codes.Unimplemented, "method Terminate not implemented") +} func (UnimplementedAteomHerderServer) mustEmbedUnimplementedAteomHerderServer() {} func (UnimplementedAteomHerderServer) testEmbeddedByValue() {} @@ -198,6 +218,24 @@ func _AteomHerder_Restore_Handler(srv interface{}, ctx context.Context, dec func return interceptor(ctx, in, info, handler) } +func _AteomHerder_Terminate_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(TerminateRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AteomHerderServer).Terminate(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: AteomHerder_Terminate_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(AteomHerderServer).Terminate(ctx, req.(*TerminateRequest)) + } + return interceptor(ctx, in, info, handler) +} + // AteomHerder_ServiceDesc is the grpc.ServiceDesc for AteomHerder service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -217,6 +255,10 @@ var AteomHerder_ServiceDesc = grpc.ServiceDesc{ MethodName: "Restore", Handler: _AteomHerder_Restore_Handler, }, + { + MethodName: "Terminate", + Handler: _AteomHerder_Terminate_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "atelet.proto", diff --git a/internal/proto/ateompb/ateom.pb.go b/internal/proto/ateompb/ateom.pb.go index 7cd226315..68d0f4fc7 100644 --- a/internal/proto/ateompb/ateom.pb.go +++ b/internal/proto/ateompb/ateom.pb.go @@ -98,6 +98,134 @@ func (SnapshotScope) EnumDescriptor() ([]byte, []int) { return file_ateom_proto_rawDescGZIP(), []int{0} } +type TerminateWorkloadRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Atespace string `protobuf:"bytes,1,opt,name=atespace,proto3" json:"atespace,omitempty"` + ActorName string `protobuf:"bytes,2,opt,name=actor_name,json=actorName,proto3" json:"actor_name,omitempty"` + ActorUid string `protobuf:"bytes,3,opt,name=actor_uid,json=actorUid,proto3" json:"actor_uid,omitempty"` + ActorTemplateNamespace string `protobuf:"bytes,4,opt,name=actor_template_namespace,json=actorTemplateNamespace,proto3" json:"actor_template_namespace,omitempty"` + ActorTemplateName string `protobuf:"bytes,5,opt,name=actor_template_name,json=actorTemplateName,proto3" json:"actor_template_name,omitempty"` + RunscPath string `protobuf:"bytes,6,opt,name=runsc_path,json=runscPath,proto3" json:"runsc_path,omitempty"` + Spec *WorkloadSpec `protobuf:"bytes,7,opt,name=spec,proto3" json:"spec,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *TerminateWorkloadRequest) Reset() { + *x = TerminateWorkloadRequest{} + mi := &file_ateom_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *TerminateWorkloadRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*TerminateWorkloadRequest) ProtoMessage() {} + +func (x *TerminateWorkloadRequest) ProtoReflect() protoreflect.Message { + mi := &file_ateom_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use TerminateWorkloadRequest.ProtoReflect.Descriptor instead. +func (*TerminateWorkloadRequest) Descriptor() ([]byte, []int) { + return file_ateom_proto_rawDescGZIP(), []int{0} +} + +func (x *TerminateWorkloadRequest) GetAtespace() string { + if x != nil { + return x.Atespace + } + return "" +} + +func (x *TerminateWorkloadRequest) GetActorName() string { + if x != nil { + return x.ActorName + } + return "" +} + +func (x *TerminateWorkloadRequest) GetActorUid() string { + if x != nil { + return x.ActorUid + } + return "" +} + +func (x *TerminateWorkloadRequest) GetActorTemplateNamespace() string { + if x != nil { + return x.ActorTemplateNamespace + } + return "" +} + +func (x *TerminateWorkloadRequest) GetActorTemplateName() string { + if x != nil { + return x.ActorTemplateName + } + return "" +} + +func (x *TerminateWorkloadRequest) GetRunscPath() string { + if x != nil { + return x.RunscPath + } + return "" +} + +func (x *TerminateWorkloadRequest) GetSpec() *WorkloadSpec { + if x != nil { + return x.Spec + } + return nil +} + +type TerminateWorkloadResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *TerminateWorkloadResponse) Reset() { + *x = TerminateWorkloadResponse{} + mi := &file_ateom_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *TerminateWorkloadResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*TerminateWorkloadResponse) ProtoMessage() {} + +func (x *TerminateWorkloadResponse) ProtoReflect() protoreflect.Message { + mi := &file_ateom_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use TerminateWorkloadResponse.ProtoReflect.Descriptor instead. +func (*TerminateWorkloadResponse) Descriptor() ([]byte, []int) { + return file_ateom_proto_rawDescGZIP(), []int{1} +} + type RunWorkloadRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Atespace string `protobuf:"bytes,1,opt,name=atespace,proto3" json:"atespace,omitempty"` @@ -123,7 +251,7 @@ type RunWorkloadRequest struct { func (x *RunWorkloadRequest) Reset() { *x = RunWorkloadRequest{} - mi := &file_ateom_proto_msgTypes[0] + mi := &file_ateom_proto_msgTypes[2] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -135,7 +263,7 @@ func (x *RunWorkloadRequest) String() string { func (*RunWorkloadRequest) ProtoMessage() {} func (x *RunWorkloadRequest) ProtoReflect() protoreflect.Message { - mi := &file_ateom_proto_msgTypes[0] + mi := &file_ateom_proto_msgTypes[2] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -148,7 +276,7 @@ func (x *RunWorkloadRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use RunWorkloadRequest.ProtoReflect.Descriptor instead. func (*RunWorkloadRequest) Descriptor() ([]byte, []int) { - return file_ateom_proto_rawDescGZIP(), []int{0} + return file_ateom_proto_rawDescGZIP(), []int{2} } func (x *RunWorkloadRequest) GetAtespace() string { @@ -231,7 +359,7 @@ type WorkloadSpec struct { func (x *WorkloadSpec) Reset() { *x = WorkloadSpec{} - mi := &file_ateom_proto_msgTypes[1] + mi := &file_ateom_proto_msgTypes[3] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -243,7 +371,7 @@ func (x *WorkloadSpec) String() string { func (*WorkloadSpec) ProtoMessage() {} func (x *WorkloadSpec) ProtoReflect() protoreflect.Message { - mi := &file_ateom_proto_msgTypes[1] + mi := &file_ateom_proto_msgTypes[3] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -256,7 +384,7 @@ func (x *WorkloadSpec) ProtoReflect() protoreflect.Message { // Deprecated: Use WorkloadSpec.ProtoReflect.Descriptor instead. func (*WorkloadSpec) Descriptor() ([]byte, []int) { - return file_ateom_proto_rawDescGZIP(), []int{1} + return file_ateom_proto_rawDescGZIP(), []int{3} } func (x *WorkloadSpec) GetContainers() []*Container { @@ -279,7 +407,7 @@ type Container struct { func (x *Container) Reset() { *x = Container{} - mi := &file_ateom_proto_msgTypes[2] + mi := &file_ateom_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -291,7 +419,7 @@ func (x *Container) String() string { func (*Container) ProtoMessage() {} func (x *Container) ProtoReflect() protoreflect.Message { - mi := &file_ateom_proto_msgTypes[2] + mi := &file_ateom_proto_msgTypes[4] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -304,7 +432,7 @@ func (x *Container) ProtoReflect() protoreflect.Message { // Deprecated: Use Container.ProtoReflect.Descriptor instead. func (*Container) Descriptor() ([]byte, []int) { - return file_ateom_proto_rawDescGZIP(), []int{2} + return file_ateom_proto_rawDescGZIP(), []int{4} } func (x *Container) GetName() string { @@ -342,7 +470,7 @@ type DurableDirVolumeMount struct { func (x *DurableDirVolumeMount) Reset() { *x = DurableDirVolumeMount{} - mi := &file_ateom_proto_msgTypes[3] + mi := &file_ateom_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -354,7 +482,7 @@ func (x *DurableDirVolumeMount) String() string { func (*DurableDirVolumeMount) ProtoMessage() {} func (x *DurableDirVolumeMount) ProtoReflect() protoreflect.Message { - mi := &file_ateom_proto_msgTypes[3] + mi := &file_ateom_proto_msgTypes[5] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -367,7 +495,7 @@ func (x *DurableDirVolumeMount) ProtoReflect() protoreflect.Message { // Deprecated: Use DurableDirVolumeMount.ProtoReflect.Descriptor instead. func (*DurableDirVolumeMount) Descriptor() ([]byte, []int) { - return file_ateom_proto_rawDescGZIP(), []int{3} + return file_ateom_proto_rawDescGZIP(), []int{5} } func (x *DurableDirVolumeMount) GetVolumeName() string { @@ -398,7 +526,7 @@ type Readyz struct { func (x *Readyz) Reset() { *x = Readyz{} - mi := &file_ateom_proto_msgTypes[4] + mi := &file_ateom_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -410,7 +538,7 @@ func (x *Readyz) String() string { func (*Readyz) ProtoMessage() {} func (x *Readyz) ProtoReflect() protoreflect.Message { - mi := &file_ateom_proto_msgTypes[4] + mi := &file_ateom_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -423,7 +551,7 @@ func (x *Readyz) ProtoReflect() protoreflect.Message { // Deprecated: Use Readyz.ProtoReflect.Descriptor instead. func (*Readyz) Descriptor() ([]byte, []int) { - return file_ateom_proto_rawDescGZIP(), []int{4} + return file_ateom_proto_rawDescGZIP(), []int{6} } func (x *Readyz) GetHttpGet() *HTTPGetAction { @@ -453,7 +581,7 @@ type HTTPGetAction struct { func (x *HTTPGetAction) Reset() { *x = HTTPGetAction{} - mi := &file_ateom_proto_msgTypes[5] + mi := &file_ateom_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -465,7 +593,7 @@ func (x *HTTPGetAction) String() string { func (*HTTPGetAction) ProtoMessage() {} func (x *HTTPGetAction) ProtoReflect() protoreflect.Message { - mi := &file_ateom_proto_msgTypes[5] + mi := &file_ateom_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -478,7 +606,7 @@ func (x *HTTPGetAction) ProtoReflect() protoreflect.Message { // Deprecated: Use HTTPGetAction.ProtoReflect.Descriptor instead. func (*HTTPGetAction) Descriptor() ([]byte, []int) { - return file_ateom_proto_rawDescGZIP(), []int{5} + return file_ateom_proto_rawDescGZIP(), []int{7} } func (x *HTTPGetAction) GetPath() string { @@ -503,7 +631,7 @@ type RunWorkloadResponse struct { func (x *RunWorkloadResponse) Reset() { *x = RunWorkloadResponse{} - mi := &file_ateom_proto_msgTypes[6] + mi := &file_ateom_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -515,7 +643,7 @@ func (x *RunWorkloadResponse) String() string { func (*RunWorkloadResponse) ProtoMessage() {} func (x *RunWorkloadResponse) ProtoReflect() protoreflect.Message { - mi := &file_ateom_proto_msgTypes[6] + mi := &file_ateom_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -528,7 +656,7 @@ func (x *RunWorkloadResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use RunWorkloadResponse.ProtoReflect.Descriptor instead. func (*RunWorkloadResponse) Descriptor() ([]byte, []int) { - return file_ateom_proto_rawDescGZIP(), []int{6} + return file_ateom_proto_rawDescGZIP(), []int{8} } type CheckpointWorkloadRequest struct { @@ -560,7 +688,7 @@ type CheckpointWorkloadRequest struct { func (x *CheckpointWorkloadRequest) Reset() { *x = CheckpointWorkloadRequest{} - mi := &file_ateom_proto_msgTypes[7] + mi := &file_ateom_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -572,7 +700,7 @@ func (x *CheckpointWorkloadRequest) String() string { func (*CheckpointWorkloadRequest) ProtoMessage() {} func (x *CheckpointWorkloadRequest) ProtoReflect() protoreflect.Message { - mi := &file_ateom_proto_msgTypes[7] + mi := &file_ateom_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -585,7 +713,7 @@ func (x *CheckpointWorkloadRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use CheckpointWorkloadRequest.ProtoReflect.Descriptor instead. func (*CheckpointWorkloadRequest) Descriptor() ([]byte, []int) { - return file_ateom_proto_rawDescGZIP(), []int{7} + return file_ateom_proto_rawDescGZIP(), []int{9} } func (x *CheckpointWorkloadRequest) GetAtespace() string { @@ -670,7 +798,7 @@ type CheckpointWorkloadResponse struct { func (x *CheckpointWorkloadResponse) Reset() { *x = CheckpointWorkloadResponse{} - mi := &file_ateom_proto_msgTypes[8] + mi := &file_ateom_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -682,7 +810,7 @@ func (x *CheckpointWorkloadResponse) String() string { func (*CheckpointWorkloadResponse) ProtoMessage() {} func (x *CheckpointWorkloadResponse) ProtoReflect() protoreflect.Message { - mi := &file_ateom_proto_msgTypes[8] + mi := &file_ateom_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -695,7 +823,7 @@ func (x *CheckpointWorkloadResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use CheckpointWorkloadResponse.ProtoReflect.Descriptor instead. func (*CheckpointWorkloadResponse) Descriptor() ([]byte, []int) { - return file_ateom_proto_rawDescGZIP(), []int{8} + return file_ateom_proto_rawDescGZIP(), []int{10} } func (x *CheckpointWorkloadResponse) GetSnapshotFiles() []string { @@ -736,7 +864,7 @@ type RestoreWorkloadRequest struct { func (x *RestoreWorkloadRequest) Reset() { *x = RestoreWorkloadRequest{} - mi := &file_ateom_proto_msgTypes[9] + mi := &file_ateom_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -748,7 +876,7 @@ func (x *RestoreWorkloadRequest) String() string { func (*RestoreWorkloadRequest) ProtoMessage() {} func (x *RestoreWorkloadRequest) ProtoReflect() protoreflect.Message { - mi := &file_ateom_proto_msgTypes[9] + mi := &file_ateom_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -761,7 +889,7 @@ func (x *RestoreWorkloadRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use RestoreWorkloadRequest.ProtoReflect.Descriptor instead. func (*RestoreWorkloadRequest) Descriptor() ([]byte, []int) { - return file_ateom_proto_rawDescGZIP(), []int{9} + return file_ateom_proto_rawDescGZIP(), []int{11} } func (x *RestoreWorkloadRequest) GetAtespace() string { @@ -863,7 +991,7 @@ type RestoreWorkloadResponse struct { func (x *RestoreWorkloadResponse) Reset() { *x = RestoreWorkloadResponse{} - mi := &file_ateom_proto_msgTypes[10] + mi := &file_ateom_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -875,7 +1003,7 @@ func (x *RestoreWorkloadResponse) String() string { func (*RestoreWorkloadResponse) ProtoMessage() {} func (x *RestoreWorkloadResponse) ProtoReflect() protoreflect.Message { - mi := &file_ateom_proto_msgTypes[10] + mi := &file_ateom_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -888,14 +1016,25 @@ func (x *RestoreWorkloadResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use RestoreWorkloadResponse.ProtoReflect.Descriptor instead. func (*RestoreWorkloadResponse) Descriptor() ([]byte, []int) { - return file_ateom_proto_rawDescGZIP(), []int{10} + return file_ateom_proto_rawDescGZIP(), []int{12} } var File_ateom_proto protoreflect.FileDescriptor const file_ateom_proto_rawDesc = "" + "\n" + - "\vateom.proto\x12\x05ateom\"\xc1\x04\n" + + "\vateom.proto\x12\x05ateom\"\xa4\x02\n" + + "\x18TerminateWorkloadRequest\x12\x1a\n" + + "\batespace\x18\x01 \x01(\tR\batespace\x12\x1d\n" + + "\n" + + "actor_name\x18\x02 \x01(\tR\tactorName\x12\x1b\n" + + "\tactor_uid\x18\x03 \x01(\tR\bactorUid\x128\n" + + "\x18actor_template_namespace\x18\x04 \x01(\tR\x16actorTemplateNamespace\x12.\n" + + "\x13actor_template_name\x18\x05 \x01(\tR\x11actorTemplateName\x12\x1d\n" + + "\n" + + "runsc_path\x18\x06 \x01(\tR\trunscPath\x12'\n" + + "\x04spec\x18\a \x01(\v2\x13.ateom.WorkloadSpecR\x04spec\"\x1b\n" + + "\x19TerminateWorkloadResponse\"\xc1\x04\n" + "\x12RunWorkloadRequest\x12\x1a\n" + "\batespace\x18\x01 \x01(\tR\batespace\x12\x1d\n" + "\n" + @@ -979,11 +1118,12 @@ const file_ateom_proto_rawDesc = "" + "\x1aSNAPSHOT_SCOPE_UNSPECIFIED\x10\x00\x12\x17\n" + "\x13SNAPSHOT_SCOPE_FULL\x10\x01\x12\x17\n" + "\x13SNAPSHOT_SCOPE_DATA\x10\x02\x12!\n" + - "\x1dSNAPSHOT_SCOPE_DATA_ON_GOLDEN\x10\x032\x80\x02\n" + + "\x1dSNAPSHOT_SCOPE_DATA_ON_GOLDEN\x10\x032\xda\x02\n" + "\x05Ateom\x12F\n" + "\vRunWorkload\x12\x19.ateom.RunWorkloadRequest\x1a\x1a.ateom.RunWorkloadResponse\"\x00\x12[\n" + "\x12CheckpointWorkload\x12 .ateom.CheckpointWorkloadRequest\x1a!.ateom.CheckpointWorkloadResponse\"\x00\x12R\n" + - "\x0fRestoreWorkload\x12\x1d.ateom.RestoreWorkloadRequest\x1a\x1e.ateom.RestoreWorkloadResponse\"\x00B=Z;github.com/agent-substrate/substrate/internal/proto/ateompbb\x06proto3" + "\x0fRestoreWorkload\x12\x1d.ateom.RestoreWorkloadRequest\x1a\x1e.ateom.RestoreWorkloadResponse\"\x00\x12X\n" + + "\x11TerminateWorkload\x12\x1f.ateom.TerminateWorkloadRequest\x1a .ateom.TerminateWorkloadResponse\"\x00B=Z;github.com/agent-substrate/substrate/internal/proto/ateompbb\x06proto3" var ( file_ateom_proto_rawDescOnce sync.Once @@ -998,48 +1138,53 @@ func file_ateom_proto_rawDescGZIP() []byte { } var file_ateom_proto_enumTypes = make([]protoimpl.EnumInfo, 1) -var file_ateom_proto_msgTypes = make([]protoimpl.MessageInfo, 14) +var file_ateom_proto_msgTypes = make([]protoimpl.MessageInfo, 16) var file_ateom_proto_goTypes = []any{ (SnapshotScope)(0), // 0: ateom.SnapshotScope - (*RunWorkloadRequest)(nil), // 1: ateom.RunWorkloadRequest - (*WorkloadSpec)(nil), // 2: ateom.WorkloadSpec - (*Container)(nil), // 3: ateom.Container - (*DurableDirVolumeMount)(nil), // 4: ateom.DurableDirVolumeMount - (*Readyz)(nil), // 5: ateom.Readyz - (*HTTPGetAction)(nil), // 6: ateom.HTTPGetAction - (*RunWorkloadResponse)(nil), // 7: ateom.RunWorkloadResponse - (*CheckpointWorkloadRequest)(nil), // 8: ateom.CheckpointWorkloadRequest - (*CheckpointWorkloadResponse)(nil), // 9: ateom.CheckpointWorkloadResponse - (*RestoreWorkloadRequest)(nil), // 10: ateom.RestoreWorkloadRequest - (*RestoreWorkloadResponse)(nil), // 11: ateom.RestoreWorkloadResponse - nil, // 12: ateom.RunWorkloadRequest.RuntimeAssetPathsEntry - nil, // 13: ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntry - nil, // 14: ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntry + (*TerminateWorkloadRequest)(nil), // 1: ateom.TerminateWorkloadRequest + (*TerminateWorkloadResponse)(nil), // 2: ateom.TerminateWorkloadResponse + (*RunWorkloadRequest)(nil), // 3: ateom.RunWorkloadRequest + (*WorkloadSpec)(nil), // 4: ateom.WorkloadSpec + (*Container)(nil), // 5: ateom.Container + (*DurableDirVolumeMount)(nil), // 6: ateom.DurableDirVolumeMount + (*Readyz)(nil), // 7: ateom.Readyz + (*HTTPGetAction)(nil), // 8: ateom.HTTPGetAction + (*RunWorkloadResponse)(nil), // 9: ateom.RunWorkloadResponse + (*CheckpointWorkloadRequest)(nil), // 10: ateom.CheckpointWorkloadRequest + (*CheckpointWorkloadResponse)(nil), // 11: ateom.CheckpointWorkloadResponse + (*RestoreWorkloadRequest)(nil), // 12: ateom.RestoreWorkloadRequest + (*RestoreWorkloadResponse)(nil), // 13: ateom.RestoreWorkloadResponse + nil, // 14: ateom.RunWorkloadRequest.RuntimeAssetPathsEntry + nil, // 15: ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntry + nil, // 16: ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntry } var file_ateom_proto_depIdxs = []int32{ - 2, // 0: ateom.RunWorkloadRequest.spec:type_name -> ateom.WorkloadSpec - 12, // 1: ateom.RunWorkloadRequest.runtime_asset_paths:type_name -> ateom.RunWorkloadRequest.RuntimeAssetPathsEntry - 3, // 2: ateom.WorkloadSpec.containers:type_name -> ateom.Container - 5, // 3: ateom.Container.readyz:type_name -> ateom.Readyz - 4, // 4: ateom.Container.durable_dir_volume_mounts:type_name -> ateom.DurableDirVolumeMount - 6, // 5: ateom.Readyz.http_get:type_name -> ateom.HTTPGetAction - 2, // 6: ateom.CheckpointWorkloadRequest.spec:type_name -> ateom.WorkloadSpec - 13, // 7: ateom.CheckpointWorkloadRequest.runtime_asset_paths:type_name -> ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntry - 0, // 8: ateom.CheckpointWorkloadRequest.scope:type_name -> ateom.SnapshotScope - 2, // 9: ateom.RestoreWorkloadRequest.spec:type_name -> ateom.WorkloadSpec - 14, // 10: ateom.RestoreWorkloadRequest.runtime_asset_paths:type_name -> ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntry - 0, // 11: ateom.RestoreWorkloadRequest.scope:type_name -> ateom.SnapshotScope - 1, // 12: ateom.Ateom.RunWorkload:input_type -> ateom.RunWorkloadRequest - 8, // 13: ateom.Ateom.CheckpointWorkload:input_type -> ateom.CheckpointWorkloadRequest - 10, // 14: ateom.Ateom.RestoreWorkload:input_type -> ateom.RestoreWorkloadRequest - 7, // 15: ateom.Ateom.RunWorkload:output_type -> ateom.RunWorkloadResponse - 9, // 16: ateom.Ateom.CheckpointWorkload:output_type -> ateom.CheckpointWorkloadResponse - 11, // 17: ateom.Ateom.RestoreWorkload:output_type -> ateom.RestoreWorkloadResponse - 15, // [15:18] is the sub-list for method output_type - 12, // [12:15] is the sub-list for method input_type - 12, // [12:12] is the sub-list for extension type_name - 12, // [12:12] is the sub-list for extension extendee - 0, // [0:12] is the sub-list for field type_name + 4, // 0: ateom.TerminateWorkloadRequest.spec:type_name -> ateom.WorkloadSpec + 4, // 1: ateom.RunWorkloadRequest.spec:type_name -> ateom.WorkloadSpec + 14, // 2: ateom.RunWorkloadRequest.runtime_asset_paths:type_name -> ateom.RunWorkloadRequest.RuntimeAssetPathsEntry + 5, // 3: ateom.WorkloadSpec.containers:type_name -> ateom.Container + 7, // 4: ateom.Container.readyz:type_name -> ateom.Readyz + 6, // 5: ateom.Container.durable_dir_volume_mounts:type_name -> ateom.DurableDirVolumeMount + 8, // 6: ateom.Readyz.http_get:type_name -> ateom.HTTPGetAction + 4, // 7: ateom.CheckpointWorkloadRequest.spec:type_name -> ateom.WorkloadSpec + 15, // 8: ateom.CheckpointWorkloadRequest.runtime_asset_paths:type_name -> ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntry + 0, // 9: ateom.CheckpointWorkloadRequest.scope:type_name -> ateom.SnapshotScope + 4, // 10: ateom.RestoreWorkloadRequest.spec:type_name -> ateom.WorkloadSpec + 16, // 11: ateom.RestoreWorkloadRequest.runtime_asset_paths:type_name -> ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntry + 0, // 12: ateom.RestoreWorkloadRequest.scope:type_name -> ateom.SnapshotScope + 3, // 13: ateom.Ateom.RunWorkload:input_type -> ateom.RunWorkloadRequest + 10, // 14: ateom.Ateom.CheckpointWorkload:input_type -> ateom.CheckpointWorkloadRequest + 12, // 15: ateom.Ateom.RestoreWorkload:input_type -> ateom.RestoreWorkloadRequest + 1, // 16: ateom.Ateom.TerminateWorkload:input_type -> ateom.TerminateWorkloadRequest + 9, // 17: ateom.Ateom.RunWorkload:output_type -> ateom.RunWorkloadResponse + 11, // 18: ateom.Ateom.CheckpointWorkload:output_type -> ateom.CheckpointWorkloadResponse + 13, // 19: ateom.Ateom.RestoreWorkload:output_type -> ateom.RestoreWorkloadResponse + 2, // 20: ateom.Ateom.TerminateWorkload:output_type -> ateom.TerminateWorkloadResponse + 17, // [17:21] is the sub-list for method output_type + 13, // [13:17] is the sub-list for method input_type + 13, // [13:13] is the sub-list for extension type_name + 13, // [13:13] is the sub-list for extension extendee + 0, // [0:13] is the sub-list for field type_name } func init() { file_ateom_proto_init() } @@ -1047,15 +1192,15 @@ func file_ateom_proto_init() { if File_ateom_proto != nil { return } - file_ateom_proto_msgTypes[0].OneofWrappers = []any{} - file_ateom_proto_msgTypes[9].OneofWrappers = []any{} + file_ateom_proto_msgTypes[2].OneofWrappers = []any{} + file_ateom_proto_msgTypes[11].OneofWrappers = []any{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_ateom_proto_rawDesc), len(file_ateom_proto_rawDesc)), NumEnums: 1, - NumMessages: 14, + NumMessages: 16, NumExtensions: 0, NumServices: 1, }, diff --git a/internal/proto/ateompb/ateom.proto b/internal/proto/ateompb/ateom.proto index c5f2293bd..13a99f9d2 100644 --- a/internal/proto/ateompb/ateom.proto +++ b/internal/proto/ateompb/ateom.proto @@ -45,6 +45,25 @@ service Ateom { // written by CheckpointWorkload. Ateom will handle downloading the correct // gVisor / runsc version to match the checkpoint. rpc RestoreWorkload(RestoreWorkloadRequest) returns (RestoreWorkloadResponse) {} + + // TerminateWorkload stops and deletes container workloads and cleans up + // network and bundle overlays on ateom. + rpc TerminateWorkload(TerminateWorkloadRequest) returns (TerminateWorkloadResponse) {} +} + +message TerminateWorkloadRequest { + string atespace = 1; + string actor_name = 2; + string actor_uid = 3; + + string actor_template_namespace = 4; + string actor_template_name = 5; + + string runsc_path = 6; + WorkloadSpec spec = 7; +} + +message TerminateWorkloadResponse { } message RunWorkloadRequest { diff --git a/internal/proto/ateompb/ateom_grpc.pb.go b/internal/proto/ateompb/ateom_grpc.pb.go index 68d5bd57a..deed36c4e 100644 --- a/internal/proto/ateompb/ateom_grpc.pb.go +++ b/internal/proto/ateompb/ateom_grpc.pb.go @@ -36,6 +36,7 @@ const ( Ateom_RunWorkload_FullMethodName = "/ateom.Ateom/RunWorkload" Ateom_CheckpointWorkload_FullMethodName = "/ateom.Ateom/CheckpointWorkload" Ateom_RestoreWorkload_FullMethodName = "/ateom.Ateom/RestoreWorkload" + Ateom_TerminateWorkload_FullMethodName = "/ateom.Ateom/TerminateWorkload" ) // AteomClient is the client API for Ateom service. @@ -67,6 +68,9 @@ type AteomClient interface { // written by CheckpointWorkload. Ateom will handle downloading the correct // gVisor / runsc version to match the checkpoint. RestoreWorkload(ctx context.Context, in *RestoreWorkloadRequest, opts ...grpc.CallOption) (*RestoreWorkloadResponse, error) + // TerminateWorkload stops and deletes container workloads and cleans up + // network and bundle overlays on ateom. + TerminateWorkload(ctx context.Context, in *TerminateWorkloadRequest, opts ...grpc.CallOption) (*TerminateWorkloadResponse, error) } type ateomClient struct { @@ -107,6 +111,16 @@ func (c *ateomClient) RestoreWorkload(ctx context.Context, in *RestoreWorkloadRe return out, nil } +func (c *ateomClient) TerminateWorkload(ctx context.Context, in *TerminateWorkloadRequest, opts ...grpc.CallOption) (*TerminateWorkloadResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(TerminateWorkloadResponse) + err := c.cc.Invoke(ctx, Ateom_TerminateWorkload_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // AteomServer is the server API for Ateom service. // All implementations must embed UnimplementedAteomServer // for forward compatibility. @@ -136,6 +150,9 @@ type AteomServer interface { // written by CheckpointWorkload. Ateom will handle downloading the correct // gVisor / runsc version to match the checkpoint. RestoreWorkload(context.Context, *RestoreWorkloadRequest) (*RestoreWorkloadResponse, error) + // TerminateWorkload stops and deletes container workloads and cleans up + // network and bundle overlays on ateom. + TerminateWorkload(context.Context, *TerminateWorkloadRequest) (*TerminateWorkloadResponse, error) mustEmbedUnimplementedAteomServer() } @@ -155,6 +172,9 @@ func (UnimplementedAteomServer) CheckpointWorkload(context.Context, *CheckpointW func (UnimplementedAteomServer) RestoreWorkload(context.Context, *RestoreWorkloadRequest) (*RestoreWorkloadResponse, error) { return nil, status.Error(codes.Unimplemented, "method RestoreWorkload not implemented") } +func (UnimplementedAteomServer) TerminateWorkload(context.Context, *TerminateWorkloadRequest) (*TerminateWorkloadResponse, error) { + return nil, status.Error(codes.Unimplemented, "method TerminateWorkload not implemented") +} func (UnimplementedAteomServer) mustEmbedUnimplementedAteomServer() {} func (UnimplementedAteomServer) testEmbeddedByValue() {} @@ -230,6 +250,24 @@ func _Ateom_RestoreWorkload_Handler(srv interface{}, ctx context.Context, dec fu return interceptor(ctx, in, info, handler) } +func _Ateom_TerminateWorkload_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(TerminateWorkloadRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AteomServer).TerminateWorkload(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Ateom_TerminateWorkload_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(AteomServer).TerminateWorkload(ctx, req.(*TerminateWorkloadRequest)) + } + return interceptor(ctx, in, info, handler) +} + // Ateom_ServiceDesc is the grpc.ServiceDesc for Ateom service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -249,6 +287,10 @@ var Ateom_ServiceDesc = grpc.ServiceDesc{ MethodName: "RestoreWorkload", Handler: _Ateom_RestoreWorkload_Handler, }, + { + MethodName: "TerminateWorkload", + Handler: _Ateom_TerminateWorkload_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "ateom.proto", diff --git a/pkg/proto/ateapipb/ateapi.pb.go b/pkg/proto/ateapipb/ateapi.pb.go index 379eb1fb1..2a49a0d50 100644 --- a/pkg/proto/ateapipb/ateapi.pb.go +++ b/pkg/proto/ateapipb/ateapi.pb.go @@ -205,6 +205,7 @@ const ( Actor_STATUS_PAUSED Actor_Status = 6 Actor_STATUS_CRASHED Actor_Status = 7 Actor_STATUS_DELETING Actor_Status = 8 + Actor_STATUS_TERMINATING Actor_Status = 9 ) // Enum value maps for Actor_Status. @@ -219,6 +220,7 @@ var ( 6: "STATUS_PAUSED", 7: "STATUS_CRASHED", 8: "STATUS_DELETING", + 9: "STATUS_TERMINATING", } Actor_Status_value = map[string]int32{ "STATUS_UNSPECIFIED": 0, @@ -230,6 +232,7 @@ var ( "STATUS_PAUSED": 6, "STATUS_CRASHED": 7, "STATUS_DELETING": 8, + "STATUS_TERMINATING": 9, } ) @@ -1837,6 +1840,7 @@ func (x *ResumeActorResponse) GetResumed() bool { type DeleteActorRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Actor *ObjectRef `protobuf:"bytes,1,opt,name=actor,proto3" json:"actor,omitempty"` + Force bool `protobuf:"varint,2,opt,name=force,proto3" json:"force,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1878,6 +1882,13 @@ func (x *DeleteActorRequest) GetActor() *ObjectRef { return nil } +func (x *DeleteActorRequest) GetForce() bool { + if x != nil { + return x.Force + } + return false +} + type GetActorSnapshotRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Snapshot *ActorSnapshotRef `protobuf:"bytes,1,opt,name=snapshot,proto3" json:"snapshot,omitempty"` @@ -3018,7 +3029,7 @@ const file_ateapi_proto_rawDesc = "" + "\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\"\xef\x06\n" + + "\x0fSTATUS_DELETING\x10\x03\"\x87\a\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" + @@ -3031,7 +3042,7 @@ const file_ateapi_proto_rawDesc = "" + "\x13local_snapshot_info\x18\t \x01(\v2\x19.ateapi.LocalSnapshotInfoR\x11localSnapshotInfo\x12W\n" + ")in_progress_snapshot_source_actor_version\x18\n" + " \x01(\x03R$inProgressSnapshotSourceActorVersion\x12;\n" + - "\ractor_volumes\x18\v \x03(\v2\x16.ateapi.ExternalVolumeR\factorVolumes\"\xc6\x01\n" + + "\ractor_volumes\x18\v \x03(\v2\x16.ateapi.ExternalVolumeR\factorVolumes\"\xde\x01\n" + "\x06Status\x12\x16\n" + "\x12STATUS_UNSPECIFIED\x10\x00\x12\x13\n" + "\x0fSTATUS_RESUMING\x10\x01\x12\x12\n" + @@ -3041,7 +3052,8 @@ const file_ateapi_proto_rawDesc = "" + "\x0eSTATUS_PAUSING\x10\x05\x12\x11\n" + "\rSTATUS_PAUSED\x10\x06\x12\x12\n" + "\x0eSTATUS_CRASHED\x10\a\x12\x13\n" + - "\x0fSTATUS_DELETING\x10\b\"\xc7\x01\n" + + "\x0fSTATUS_DELETING\x10\b\x12\x16\n" + + "\x12STATUS_TERMINATING\x10\t\"\xc7\x01\n" + "\x10WorkerAssignment\x12)\n" + "\x10worker_namespace\x18\x01 \x01(\tR\x0fworkerNamespace\x12\x1f\n" + "\vworker_pool\x18\x02 \x01(\tR\n" + @@ -3107,9 +3119,10 @@ const file_ateapi_proto_rawDesc = "" + "\x04boot\x18\x02 \x01(\bR\x04boot\"T\n" + "\x13ResumeActorResponse\x12#\n" + "\x05actor\x18\x01 \x01(\v2\r.ateapi.ActorR\x05actor\x12\x18\n" + - "\aresumed\x18\x02 \x01(\bR\aresumed\"=\n" + + "\aresumed\x18\x02 \x01(\bR\aresumed\"S\n" + "\x12DeleteActorRequest\x12'\n" + - "\x05actor\x18\x01 \x01(\v2\x11.ateapi.ObjectRefR\x05actor\"O\n" + + "\x05actor\x18\x01 \x01(\v2\x11.ateapi.ObjectRefR\x05actor\x12\x14\n" + + "\x05force\x18\x02 \x01(\bR\x05force\"O\n" + "\x17GetActorSnapshotRequest\x124\n" + "\bsnapshot\x18\x01 \x01(\v2\x18.ateapi.ActorSnapshotRefR\bsnapshot\"s\n" + "\x19ListActorSnapshotsRequest\x12\x1a\n" + diff --git a/pkg/proto/ateapipb/ateapi.proto b/pkg/proto/ateapipb/ateapi.proto index 85106009d..96cef6669 100644 --- a/pkg/proto/ateapipb/ateapi.proto +++ b/pkg/proto/ateapipb/ateapi.proto @@ -173,6 +173,7 @@ message Actor { STATUS_PAUSED = 6; STATUS_CRASHED = 7; STATUS_DELETING = 8; + STATUS_TERMINATING = 9; } Status status = 4; @@ -363,6 +364,7 @@ message ResumeActorResponse { message DeleteActorRequest { ObjectRef actor = 1; + bool force = 2; } message GetActorSnapshotRequest {