diff --git a/DESIGN.md b/DESIGN.md index 20871b04..fd447230 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -4,7 +4,7 @@ Storing millions of short-lived tasks as Kubernetes CRDs pushes etcd past its comfort zone (single-digit GB storage limits, write-rate bottlenecks, control plane degradation). AX stores its state in PostgreSQL and reconciles directly with Agent Substrate under fine-grained distributed locks. -PostgreSQL holds resources as protojson documents alongside the leases that back AX's distributed locks. Lock waiters poll; nothing depends on `LISTEN/NOTIFY`, so any connection pooler works. `ax-server` creates any missing tables when it starts; there are no versioned migrations yet. The bundled manifest runs a single-instance StatefulSet with a persistent volume; any reachable PostgreSQL works. Storage sits behind the `store.Store` and `lock.Locker` interfaces. The store holds only what AX owns: specs, metadata, and the parts of a task's status that describe AX's own work (`Pending`, `Failed`, `Terminating`, the task ID, workspace setup). Everything that belongs to the task's actor (its phase while provisioned, `actor`, `workerIP`, and the `Ready` condition) is read from Agent Substrate whenever a task is fetched or listed, so AX never keeps a copy that can drift from the actor. The split lives in a few helpers in `internal/server/server.go` (`getTask`, `listTasks`, `saveTask`, `updateTask`, `deleteTask`), so storage implementations never need to know about Substrate. +PostgreSQL holds resources as protojson documents alongside the leases that back AX's distributed locks. Lock waiters poll; nothing depends on `LISTEN/NOTIFY`, so any connection pooler works. `ax-server` creates any missing tables when it starts; there are no versioned migrations yet. The bundled manifest runs a single-instance StatefulSet with a persistent volume; any reachable PostgreSQL works. Storage sits behind the `store.Store` and `lock.Locker` interfaces. The store holds only what AX owns: specs, metadata, and the parts of a task's status that describe AX's own work (`Pending`, `Failed`, `Terminating`, the task ID, workspace setup). Everything that belongs to the task's actor (its phase while provisioned, `actor`, `workerIP`, and the `Ready` condition) is read from Agent Substrate whenever a task is fetched or listed, so AX never keeps a copy that can drift from the actor. The exceptions are `residency` and `actorUid`: the store keeps the values Substrate last returned to ax-server, so that each change AX makes is a stored change, while reads still report the actor's current state. The split lives in a few helpers in `internal/server/server.go` (`getTask`, `listTasks`, `saveTask`, `updateTask`, `deleteTask`), so storage implementations never need to know about Substrate. ``` ax apply -f task.yaml diff --git a/README.md b/README.md index bd6ac673..5a57d786 100644 --- a/README.md +++ b/README.md @@ -120,8 +120,8 @@ From your ax checkout: ax apply -f examples/simple.yaml # the smallest useful Task ax resume task simple-task # new tasks start Suspended; resume schedules the actor and starts it ax get tasks -# NAME ATESPACE PHASE ACTOR WORKER-IP AGE -# simple-task default Running simple-task 10.20.3.67 1m +# NAME ATESPACE PHASE RESIDENCY OUTCOME ACTOR WORKER-IP AGE +# simple-task default Running running simple-task 10.20.3.67 1m ax ssh simple-task -- ls -la /workspace # poke around inside the sandbox ax suspend task simple-task # checkpoint and pause diff --git a/cmd/ax/main.go b/cmd/ax/main.go index d07cc4ac..c68bbccb 100644 --- a/cmd/ax/main.go +++ b/cmd/ax/main.go @@ -414,7 +414,7 @@ func runGet(serverURL, atespace string, args []string) error { tasks := resp.Tasks w := tabwriter.NewWriter(os.Stdout, 0, 8, 3, ' ', 0) - fmt.Fprintln(w, "NAME\tATESPACE\tPHASE\tACTOR\tWORKER-IP\tAGE") + fmt.Fprintln(w, "NAME\tATESPACE\tPHASE\tRESIDENCY\tOUTCOME\tACTOR\tWORKER-IP\tAGE") for _, t := range tasks { workerIP := "" actor := "" @@ -426,12 +426,6 @@ func runGet(serverURL, atespace string, args []string) error { phase = t.Status.Phase } } - if workerIP == "" { - workerIP = "" - } - if actor == "" { - actor = "" - } age := "" name := "" tAtespace := "" @@ -442,12 +436,14 @@ func runGet(serverURL, atespace string, args []string) error { age = formatAge(time.Since(t.Metadata.CreateTime.AsTime())) } } - fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\n", + fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n", name, tAtespace, phase, - actor, - workerIP, + orNone(t.GetStatus().GetResidency()), + orNone(t.GetStatus().GetOutcome()), + orNone(actor), + orNone(workerIP), age, ) } @@ -695,7 +691,10 @@ func runDescribe(serverURL, atespace string, args []string) error { fmt.Printf("Name: %s\n", taskName) fmt.Printf("Atespace: %s\n", taskAtespace) fmt.Printf("Phase: %s\n", phase) + fmt.Printf("Residency: %s\n", orNone(task.GetStatus().GetResidency())) + fmt.Printf("Outcome: %s\n", orNone(task.GetStatus().GetOutcome())) fmt.Printf("Actor: %s\n", actor) + fmt.Printf("Actor UID: %s\n", orNone(task.GetStatus().GetActorUid())) fmt.Printf("Worker IP: %s\n", workerIP) if task.Spec != nil { if ref := task.Spec.GetWorkspace(); ref != nil { @@ -730,6 +729,14 @@ func runDescribe(serverURL, atespace string, args []string) error { return nil } +// orNone returns s, or "" when s is empty, for table and describe output. +func orNone(s string) string { + if s == "" { + return "" + } + return s +} + // runDelete removes one resource by kind and name. func runDelete(serverURL, atespace string, args []string) error { if len(args) < 2 { diff --git a/cmd/ax/main_test.go b/cmd/ax/main_test.go index 8641ae22..7952d8b9 100644 --- a/cmd/ax/main_test.go +++ b/cmd/ax/main_test.go @@ -312,6 +312,75 @@ func TestRunGetResourceAliases(t *testing.T) { } } +// captureStdout runs fn with os.Stdout redirected to a file and returns what +// fn wrote. +func captureStdout(t *testing.T, fn func() error) string { + t.Helper() + output, err := os.CreateTemp(t.TempDir(), "stdout") + if err != nil { + t.Fatal(err) + } + defer output.Close() + stdout := os.Stdout + os.Stdout = output + runErr := fn() + os.Stdout = stdout + if runErr != nil { + t.Fatal(runErr) + } + data, err := os.ReadFile(output.Name()) + if err != nil { + t.Fatal(err) + } + return string(data) +} + +func TestRunGetAndDescribeShowResidencyAndOutcome(t *testing.T) { + sub := substratetest.NewServer(t) + srv := server.NewServer(memory.NewStore(), newTestReconciler(sub)) + if _, err := srv.CreateTask(context.Background(), &v1alpha1.CreateTaskRequest{Task: &v1alpha1.Task{ + Metadata: &v1alpha1.ObjectMeta{Name: "resident", Atespace: "default"}, + Spec: &v1alpha1.TaskSpec{Image: "alpine"}, + }}); err != nil { + t.Fatal(err) + } + actor, _ := sub.Actor("default", "resident") + uid := actor.GetMetadata().GetUid() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + grpcServer := srv.GRPCServer() + t.Cleanup(grpcServer.Stop) + go func() { _ = grpcServer.Serve(listener) }() + addr := listener.Addr().String() + + out := captureStdout(t, func() error { return runGet(addr, "default", []string{"tasks"}) }) + rows := strings.Split(strings.TrimSpace(out), "\n") + if len(rows) != 2 { + t.Fatalf("expected a header and one task, got %q", out) + } + header, row := strings.Fields(rows[0]), strings.Fields(rows[1]) + want := map[string]string{"NAME": "resident", "PHASE": "Suspended", "RESIDENCY": "suspended", "OUTCOME": ""} + for i, col := range header { + if w, ok := want[col]; ok && (i >= len(row) || row[i] != w) { + t.Errorf("column %s = %q, want %q; output:\n%s", col, row[i], w, out) + } + delete(want, col) + } + if len(want) != 0 { + t.Errorf("ax get tasks has no columns %v; output:\n%s", want, out) + } + + out = captureStdout(t, func() error { return runDescribe(addr, "default", []string{"task", "resident"}) }) + for _, line := range []string{"Residency: suspended", "Outcome: ", "Actor UID: " + uid} { + if !strings.Contains(out, line+"\n") { + t.Errorf("ax describe task output has no line %q:\n%s", line, out) + } + } +} + // newTestReconciler returns a TaskReconciler that runs tasks on sub. func newTestReconciler(sub *substratetest.Server) *controller.TaskReconciler { return controller.NewTaskReconciler(sub.Client(), "test-template", v1alpha1.DefaultSystemNamespace, "") diff --git a/docs/concepts.md b/docs/concepts.md index 0e19cdd1..cfae72b8 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -17,6 +17,8 @@ The unit is deliberately small. An agent is not one process that runs to complet | `WorkspaceReady` | The workspace has finished setting up. Stays True afterwards. | | `Ready` | The task is running and `WorkspaceReady` is True. This is the one to wait on. | +`status.residency` says where the task's actor is on Agent Substrate: `queued` (no actor yet), `resuming`, `running`, `suspending`, `suspended`, `reverting`, `crashed`, `deleting`, or `deleted`. Unlike the phase, it tells a crashed or reverting actor apart from a task AX failed to provision. `status.actorUid` is the actor's Substrate UID, the `ate.actor.uid` that Substrate's logs and telemetry carry. `status.outcome` will say how the command ended (`succeeded`, `failed`, or `cancelled`); AX does not detect command exit yet ([#346](https://github.com/google/ax/issues/346)), so it is empty for now. `ax get tasks` shows the residency and outcome; `ax describe task` also shows the actor UID. + Two transitions are worth knowing. Suspending a task sets `Ready` to False with reason `Suspended`; resuming sets it back. Deleting a task tears down the sandbox on Agent Substrate and removes the record. `ax delete` blocks until that has completed. ## Workspace diff --git a/internal/controller/reconciler.go b/internal/controller/reconciler.go index 58e85926..666b17df 100644 --- a/internal/controller/reconciler.go +++ b/internal/controller/reconciler.go @@ -32,6 +32,7 @@ import ( "github.com/google/ax/internal/substrate" "github.com/google/ax/pkg/apis/v1alpha1" + ateapipb "go.ate.dev/substrate/api/ateapi/v1" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" @@ -123,11 +124,11 @@ func newProbeHTTPClient() *http.Client { } // ensureSubstrateResources fills in the task's defaults and makes sure its -// atespace, ActorTemplate and actor exist on Substrate. It records only what AX -// stores in the task's status: the task ID, and on failure the reason in a -// False Ready condition. The caller sets the phase, and the server reads the -// rest of the status from the actor. -func (r *TaskReconciler) ensureSubstrateResources(ctx context.Context, task *v1alpha1.Task, ws *v1alpha1.Workspace) error { +// atespace, ActorTemplate and actor exist on Substrate, and returns the actor. +// It records only what AX stores in the task's status: the task ID, and on +// failure the reason in a False Ready condition. The caller sets the phase, +// and the server reads the rest of the status from the actor. +func (r *TaskReconciler) ensureSubstrateResources(ctx context.Context, task *v1alpha1.Task, ws *v1alpha1.Workspace) (*ateapipb.Actor, error) { if task.Metadata == nil { task.Metadata = &v1alpha1.ObjectMeta{} } @@ -155,10 +156,10 @@ func (r *TaskReconciler) ensureSubstrateResources(ctx context.Context, task *v1a if err := r.client.EnsureAtespace(ctx, atespace); err != nil { if substrate.IsTransientError(err) { r.setNotReady(task, v1alpha1.ReasonSubstrateUnavailable, err.Error(), now) - return fmt.Errorf("ensuring atespace: %w", err) + return nil, fmt.Errorf("ensuring atespace: %w", err) } r.setNotReady(task, v1alpha1.ReasonAtespaceCreationFailed, err.Error(), now) - return fmt.Errorf("ensuring atespace: %w", err) + return nil, fmt.Errorf("ensuring atespace: %w", err) } // 2. The actor is always named after the task, so the two can be used @@ -205,7 +206,7 @@ func (r *TaskReconciler) ensureSubstrateResources(ctx context.Context, task *v1a // here as well as at apply time: a task must never run without limits it asked for. if err := v1alpha1.ValidateResources(task.Spec.Resources); err != nil { r.setNotReady(task, v1alpha1.ReasonInvalidResources, err.Error(), now) - return fmt.Errorf("validating resources: %w", err) + return nil, fmt.Errorf("validating resources: %w", err) } resources := substrate.ResourceLimits(task.Spec.Resources) @@ -220,14 +221,14 @@ func (r *TaskReconciler) ensureSubstrateResources(ctx context.Context, task *v1a if err != nil { if substrate.IsTransientError(err) { r.setNotReady(task, v1alpha1.ReasonSubstrateUnavailable, err.Error(), now) - return fmt.Errorf("ensuring actor template: %w", err) + return nil, fmt.Errorf("ensuring actor template: %w", err) } // Falling back to the default template would run a different image than // the user asked for, without the limits it asked for, or fail later with // a misleading "actor template not found". Surface the real error on the // task and stop here. r.setNotReady(task, v1alpha1.ReasonTemplateCreationFailed, substrateErrorMessage(err), now) - return fmt.Errorf("ensuring actor template: %w", err) + return nil, fmt.Errorf("ensuring actor template: %w", err) } if tmpl != nil && tmpl.Metadata != nil { templateAtespace = tmpl.Metadata.Atespace @@ -236,47 +237,67 @@ func (r *TaskReconciler) ensureSubstrateResources(ctx context.Context, task *v1a } } - _, err := r.client.EnsureActor(ctx, atespace, actorName, templateAtespace, templateName) + actor, err := r.client.EnsureActor(ctx, atespace, actorName, templateAtespace, templateName) if err != nil { if substrate.IsTransientError(err) { r.setNotReady(task, v1alpha1.ReasonSubstrateUnavailable, err.Error(), now) - return fmt.Errorf("ensuring actor: %w", err) + return nil, fmt.Errorf("ensuring actor: %w", err) } r.setNotReady(task, v1alpha1.ReasonActorCreationFailed, err.Error(), now) - return fmt.Errorf("ensuring actor: %w", err) + return nil, fmt.Errorf("ensuring actor: %w", err) + } + return actor, nil +} + +// recordActor records the UID and residency of the task's actor, as Substrate +// returned them, in the task's status. It leaves both unchanged when Substrate +// returned no actor. +func recordActor(task *v1alpha1.Task, actor *ateapipb.Actor) { + if uid := actor.GetMetadata().GetUid(); uid != "" { + task.Status.ActorUid = uid + } + if residency := substrate.Residency(actor); residency != "" { + task.Status.Residency = residency } - return nil } -// Suspend runs ensureSubstrateResources, then suspends the task's actor. A -// failure to suspend the actor is returned without changing the task's status. +// Suspend runs ensureSubstrateResources, then suspends the task's actor and +// records the actor's UID and residency in the task's status. A failure to +// suspend the actor is returned without changing the task's status. func (r *TaskReconciler) Suspend(ctx context.Context, task *v1alpha1.Task, ws *v1alpha1.Workspace) error { - if err := r.ensureSubstrateResources(ctx, task, ws); err != nil { + ensured, err := r.ensureSubstrateResources(ctx, task, ws) + if err != nil { return err } atespace, actorName := task.Metadata.Atespace, task.Metadata.Name slog.Info("suspending actor on Substrate", "actor", actorName) - if err := r.client.SuspendActor(ctx, atespace, actorName); err != nil { + suspended, err := r.client.SuspendActor(ctx, atespace, actorName) + if err != nil { return fmt.Errorf("suspending actor: %w", err) } + recordActor(task, ensured) + recordActor(task, suspended) slog.Info("task successfully suspended", "name", task.Metadata.Name, "actor", actorName) return nil } -// Resume runs ensureSubstrateResources, resumes the task's actor, and waits up -// to WorkspaceReadyTimeout for the task's workspace to be set up, recording the +// Resume runs ensureSubstrateResources, resumes the task's actor, records the +// actor's UID and residency in the task's status, and waits up to +// WorkspaceReadyTimeout for the task's workspace to be set up, recording the // result in the WorkspaceReady condition. A failure to resume the actor is // recorded the way ensureSubstrateResources records its own. func (r *TaskReconciler) Resume(ctx context.Context, task *v1alpha1.Task, ws *v1alpha1.Workspace) error { - if err := r.ensureSubstrateResources(ctx, task, ws); err != nil { + ensured, err := r.ensureSubstrateResources(ctx, task, ws) + if err != nil { return err } + recordActor(task, ensured) atespace, actorName := task.Metadata.Atespace, task.Metadata.Name now := time.Now() // Resume the Actor to activate container execution slog.Info("resuming actor on Substrate worker", "actor", actorName) - _, workerIP, err := r.client.ResumeActor(ctx, atespace, actorName) + actor, workerIP, err := r.client.ResumeActor(ctx, atespace, actorName) if err != nil { if substrate.IsTransientError(err) { r.setNotReady(task, v1alpha1.ReasonSubstrateUnavailable, err.Error(), now) @@ -285,6 +306,7 @@ func (r *TaskReconciler) Resume(ctx context.Context, task *v1alpha1.Task, ws *v1 r.setNotReady(task, v1alpha1.ReasonActorResumeFailed, err.Error(), now) return fmt.Errorf("resuming actor: %w", err) } + recordActor(task, actor) // Check if workspace setup inside the actor has completed host := workerIP diff --git a/internal/server/actor_status_test.go b/internal/server/actor_status_test.go index fa37223c..f8e330d5 100644 --- a/internal/server/actor_status_test.go +++ b/internal/server/actor_status_test.go @@ -126,3 +126,42 @@ func TestSetActorStatus_SubstrateErrorReportsUnknown(t *testing.T) { t.Errorf("phase = %q, want %q to be kept", got, v1alpha1.PhaseTerminating) } } + +func TestSetActorStatus_ResidencyAndActorUID(t *testing.T) { + withUID := func(a *ateapipb.Actor, uid string) *ateapipb.Actor { + a.Metadata.Uid = uid + return a + } + recorded := func(name, phase, residency, uid string) *v1alpha1.Task { + task := storedTask(name, phase) + task.Status.Residency, task.Status.ActorUid = residency, uid + return task + } + + tests := []struct { + name string + task *v1alpha1.Task + actor *ateapipb.Actor + err error + wantResidency string + wantUID string + }{ + {"live state wins over the recorded one", recorded("a", "", "suspended", "uid-1"), withUID(testActor("a", ateapipb.ActorState_ACTOR_STATE_RUNNING, ""), "uid-1"), nil, "running", "uid-1"}, + {"crashed", recorded("a", "", "running", "uid-1"), withUID(testActor("a", ateapipb.ActorState_ACTOR_STATE_CRASHED, ""), "uid-1"), nil, "crashed", "uid-1"}, + {"reverting", recorded("a", "", "crashed", "uid-1"), withUID(testActor("a", ateapipb.ActorState_ACTOR_STATE_REVERTING, ""), "uid-1"), nil, "reverting", "uid-1"}, + {"recreated actor has a new UID", recorded("a", "", "crashed", "uid-1"), withUID(testActor("a", ateapipb.ActorState_ACTOR_STATE_SUSPENDED, ""), "uid-2"), nil, "suspended", "uid-2"}, + {"actor without a UID keeps the recorded one", recorded("a", "", "suspended", "uid-1"), testActor("a", ateapipb.ActorState_ACTOR_STATE_RUNNING, ""), nil, "running", "uid-1"}, + {"actor that existed is gone", recorded("a", "", "running", "uid-1"), nil, nil, "deleted", "uid-1"}, + {"actor not created yet", recorded("a", v1alpha1.PhasePending, "queued", ""), nil, nil, "queued", ""}, + {"substrate unreachable keeps the recorded values", recorded("a", "", "suspended", "uid-1"), nil, errors.New("unavailable"), "suspended", "uid-1"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + setActorStatus(tt.task, tt.actor, tt.err) + st := tt.task.GetStatus() + if st.GetResidency() != tt.wantResidency || st.GetActorUid() != tt.wantUID { + t.Errorf("residency=%q actorUID=%q, want %q %q", st.GetResidency(), st.GetActorUid(), tt.wantResidency, tt.wantUID) + } + }) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index f5d61444..db142211 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -198,6 +198,9 @@ func (s *Server) CreateTask(ctx context.Context, req *v1alpha1.CreateTaskRequest } // The task is Pending until its actor exists. task.Status.Phase = v1alpha1.PhasePending + if task.Status.Residency == "" { + task.Status.Residency = v1alpha1.ResidencyQueued + } if err := s.saveTask(ctx, task); err != nil { return nil, status.Errorf(codes.Internal, "saving task: %v", err) } @@ -247,6 +250,7 @@ func (s *Server) DeleteTask(ctx context.Context, req *v1alpha1.DeleteTaskRequest task.Status = &v1alpha1.TaskStatus{} } task.Status.Phase = v1alpha1.PhaseTerminating + task.Status.Residency = v1alpha1.ResidencyDeleting if err := s.updateTask(ctx, atespace, taskName, task.Status); err != nil { return nil, status.Errorf(codes.Internal, "updating task status: %v", err) } @@ -454,10 +458,10 @@ func (s *Server) applyActor(ctx context.Context, task *v1alpha1.Task) { } // setActorStatus overlays the actor's state on the task's status: actor, -// workerIP, and, unless the phase is one of AX's own, the phase and Ready -// condition. actor is nil when the task has no actor; err is set when the actor -// could not be looked up, in which case the task is reported with phase -// Unknown. +// actorUID, residency, workerIP, and, unless the phase is one of AX's own, the +// phase and Ready condition. actor is nil when the task has no actor; err is +// set when the actor could not be looked up, in which case the task is +// reported with phase Unknown and its last recorded residency. func setActorStatus(task *v1alpha1.Task, actor *ateapipb.Actor, err error) { if task.Status == nil { task.Status = &v1alpha1.TaskStatus{} @@ -474,6 +478,11 @@ func setActorStatus(task *v1alpha1.Task, actor *ateapipb.Actor, err error) { setReady(task, "Unknown", v1alpha1.ReasonSubstrateUnavailable, err.Error()) } case actor == nil: + // An actor UID means the actor existed; without one the task is still + // queued. + if st.ActorUid != "" { + st.Residency = v1alpha1.ResidencyDeleted + } // A task that finished provisioning always has an actor, so a missing one // was removed outside AX. if !v1alpha1.IsAXPhase(st.Phase) { @@ -482,6 +491,10 @@ func setActorStatus(task *v1alpha1.Task, actor *ateapipb.Actor, err error) { } default: st.Actor = actor.GetMetadata().GetName() + if uid := actor.GetMetadata().GetUid(); uid != "" { + st.ActorUid = uid + } + st.Residency = substrate.Residency(actor) st.WorkerIp = substrate.WorkerIP(actor) if v1alpha1.IsAXPhase(st.Phase) { return @@ -563,7 +576,9 @@ func setReady(task *v1alpha1.Task, condStatus, reason, message string) { // storedStatus returns the part of a task's status that AX owns and stores. The // actor, its worker, and the phase and Ready condition that follow from the -// actor's state are dropped. st is not modified. +// actor's state are dropped. The actor's UID and residency are kept as +// Substrate last returned them, so that a change to either is a stored change. +// st is not modified. func storedStatus(st *v1alpha1.TaskStatus) *v1alpha1.TaskStatus { out := &v1alpha1.TaskStatus{} if st != nil { diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 52909489..73cbe7f4 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -979,3 +979,110 @@ func TestServer_SuccessClearsStoredFailure(t *testing.T) { t.Errorf("phase after ResumeTask = %q, want Running", task.GetStatus().GetPhase()) } } + +// residencyRecorder records the residency of every task status written to the +// store, in order. +type residencyRecorder struct { + store.Store + written []string +} + +func (r *residencyRecorder) SaveTask(ctx context.Context, task *v1alpha1.Task) error { + r.written = append(r.written, task.GetStatus().GetResidency()) + return r.Store.SaveTask(ctx, task) +} + +func (r *residencyRecorder) UpdateTaskStatus(ctx context.Context, atespace, name string, st *v1alpha1.TaskStatus) error { + r.written = append(r.written, st.GetResidency()) + return r.Store.UpdateTaskStatus(ctx, atespace, name, st) +} + +func TestServer_RecordsResidencyAndActorUID(t *testing.T) { + st := &residencyRecorder{Store: memory.NewStore()} + sub := substratetest.NewServer(t) + srv := server.NewServer(st, newTestReconciler(sub)) + ctx := context.Background() + + task, err := srv.CreateTask(ctx, &v1alpha1.CreateTaskRequest{Task: &v1alpha1.Task{ + Metadata: &v1alpha1.ObjectMeta{Name: "resident"}, + Spec: &v1alpha1.TaskSpec{Image: "alpine"}, + }}) + if err != nil { + t.Fatalf("CreateTask: %v", err) + } + actor, _ := sub.Actor("default", "resident") + uid := actor.GetMetadata().GetUid() + if uid == "" { + t.Fatal("fake Substrate gave the actor no UID") + } + if got := task.GetStatus(); got.GetResidency() != v1alpha1.ResidencySuspended || got.GetActorUid() != uid { + t.Errorf("CreateTask status residency=%q actorUID=%q, want %q %q", got.GetResidency(), got.GetActorUid(), v1alpha1.ResidencySuspended, uid) + } + + if _, err := srv.ResumeTask(ctx, &v1alpha1.ResumeTaskRequest{Name: "resident"}); err != nil { + t.Fatalf("ResumeTask: %v", err) + } + if _, err := srv.SuspendTask(ctx, &v1alpha1.SuspendTaskRequest{Name: "resident"}); err != nil { + t.Fatalf("SuspendTask: %v", err) + } + + stored, err := st.GetTask(ctx, "default", "resident") + if err != nil { + t.Fatalf("store GetTask: %v", err) + } + if got := stored.GetStatus(); got.GetResidency() != v1alpha1.ResidencySuspended || got.GetActorUid() != uid { + t.Errorf("stored residency=%q actorUID=%q, want %q %q", got.GetResidency(), got.GetActorUid(), v1alpha1.ResidencySuspended, uid) + } + + // A request wakes the actor through the router, without AX (#502). Reads + // report it running; the stored residency is AX's last write. + actor, _ = sub.Actor("default", "resident") + actor.Status = &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_RUNNING} + sub.SetActor(actor) + got, err := srv.GetTask(ctx, &v1alpha1.GetTaskRequest{Name: "resident"}) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if got.GetStatus().GetResidency() != v1alpha1.ResidencyRunning { + t.Errorf("residency after a wake on Substrate = %q, want %q", got.GetStatus().GetResidency(), v1alpha1.ResidencyRunning) + } + + if _, err := srv.DeleteTask(ctx, &v1alpha1.DeleteTaskRequest{Name: "resident"}); err != nil { + t.Fatalf("DeleteTask: %v", err) + } + + want := []string{ + v1alpha1.ResidencyQueued, // CreateTask, before the actor exists + v1alpha1.ResidencySuspended, // CreateTask, once the actor is created suspended + v1alpha1.ResidencyRunning, // ResumeTask + v1alpha1.ResidencySuspended, // SuspendTask + v1alpha1.ResidencyDeleting, // DeleteTask, before the actor is deleted + } + if strings.Join(st.written, ",") != strings.Join(want, ",") { + t.Errorf("stored residencies = %v, want %v", st.written, want) + } +} + +func TestServer_ActorDeletedOutsideAXIsResidencyDeleted(t *testing.T) { + sub := substratetest.NewServer(t) + srv := server.NewServer(memory.NewStore(), newTestReconciler(sub)) + ctx := context.Background() + + if _, err := srv.CreateTask(ctx, &v1alpha1.CreateTaskRequest{Task: &v1alpha1.Task{ + Metadata: &v1alpha1.ObjectMeta{Name: "orphaned"}, + Spec: &v1alpha1.TaskSpec{Image: "alpine"}, + }}); err != nil { + t.Fatalf("CreateTask: %v", err) + } + if err := sub.Client().DeleteActor(ctx, "default", "orphaned"); err != nil { + t.Fatalf("DeleteActor: %v", err) + } + + got, err := srv.GetTask(ctx, &v1alpha1.GetTaskRequest{Name: "orphaned"}) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if got.GetStatus().GetResidency() != v1alpha1.ResidencyDeleted || got.GetStatus().GetActorUid() == "" { + t.Errorf("residency=%q actorUID=%q, want %q and the deleted actor's UID", got.GetStatus().GetResidency(), got.GetStatus().GetActorUid(), v1alpha1.ResidencyDeleted) + } +} diff --git a/internal/server/status_test.go b/internal/server/status_test.go index 3a16628a..cfa114fa 100644 --- a/internal/server/status_test.go +++ b/internal/server/status_test.go @@ -44,6 +44,11 @@ func TestStoredStatus(t *testing.T) { t.Errorf("storedStatus modified its input: %v", in) } + in = &v1alpha1.TaskStatus{Phase: "Running", Residency: v1alpha1.ResidencyRunning, ActorUid: "uid-1"} + if got := storedStatus(in); got.Residency != v1alpha1.ResidencyRunning || got.ActorUid != "uid-1" { + t.Errorf("storedStatus(%v) = %v, want residency and actor UID kept", in, got) + } + in = &v1alpha1.TaskStatus{Phase: v1alpha1.PhaseFailed, Conditions: []*v1alpha1.Condition{{Type: "Ready"}}} if got := storedStatus(in); got.Phase != v1alpha1.PhaseFailed || len(got.Conditions) != 1 { t.Errorf("storedStatus(%v) = %v, want AX phase and its conditions kept", in, got) diff --git a/internal/store/postgres/store_test.go b/internal/store/postgres/store_test.go index c74e622e..a38a0ada 100644 --- a/internal/store/postgres/store_test.go +++ b/internal/store/postgres/store_test.go @@ -229,3 +229,32 @@ func TestRestartKeepsTasksAndReleasesLocks(t *testing.T) { } unlock() } + +// TestUpdateTaskStatusKeepsResidencyAndActorUID checks that the status fields +// state-change telemetry compares survive a write and a read. +func TestUpdateTaskStatusKeepsResidencyAndActorUID(t *testing.T) { + s := openTest(t, Options{}) + ctx := context.Background() + task := &v1alpha1.Task{ + Metadata: &v1alpha1.ObjectMeta{Atespace: "default", Name: "resident"}, + Status: &v1alpha1.TaskStatus{Phase: v1alpha1.PhasePending, Residency: v1alpha1.ResidencyQueued}, + } + if err := s.SaveTask(ctx, task); err != nil { + t.Fatalf("SaveTask: %v", err) + } + if got, err := s.GetTask(ctx, "default", "resident"); err != nil || got.GetStatus().GetResidency() != v1alpha1.ResidencyQueued { + t.Fatalf("GetTask after SaveTask = %v, %v; want residency %q", got.GetStatus(), err, v1alpha1.ResidencyQueued) + } + + st := &v1alpha1.TaskStatus{Residency: v1alpha1.ResidencySuspended, ActorUid: "uid-1"} + if err := s.UpdateTaskStatus(ctx, "default", "resident", st); err != nil { + t.Fatalf("UpdateTaskStatus: %v", err) + } + got, err := s.GetTask(ctx, "default", "resident") + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if got.GetStatus().GetResidency() != v1alpha1.ResidencySuspended || got.GetStatus().GetActorUid() != "uid-1" { + t.Errorf("stored status = %v, want residency %q and actor UID uid-1", got.GetStatus(), v1alpha1.ResidencySuspended) + } +} diff --git a/internal/substrate/client.go b/internal/substrate/client.go index b626c7f2..e1cd0c06 100644 --- a/internal/substrate/client.go +++ b/internal/substrate/client.go @@ -551,22 +551,48 @@ func WorkerIP(actor *ateapipb.Actor) string { return "" } -// SuspendActor suspends the specified actor, triggering state checkpointing. -func (c *Client) SuspendActor(ctx context.Context, atespace, actorName string) error { +// Residency returns the task residency that matches an actor's state, or "" +// for a state AX does not recognize. +func Residency(actor *ateapipb.Actor) string { + switch actor.GetStatus().GetState() { + case ateapipb.ActorState_ACTOR_STATE_RESUMING: + return v1alpha1.ResidencyResuming + case ateapipb.ActorState_ACTOR_STATE_RUNNING: + return v1alpha1.ResidencyRunning + case ateapipb.ActorState_ACTOR_STATE_SUSPENDING: + return v1alpha1.ResidencySuspending + case ateapipb.ActorState_ACTOR_STATE_SUSPENDED: + return v1alpha1.ResidencySuspended + case ateapipb.ActorState_ACTOR_STATE_REVERTING: + return v1alpha1.ResidencyReverting + case ateapipb.ActorState_ACTOR_STATE_CRASHED: + return v1alpha1.ResidencyCrashed + case ateapipb.ActorState_ACTOR_STATE_DELETING: + return v1alpha1.ResidencyDeleting + default: + return "" + } +} + +// SuspendActor suspends the specified actor, triggering state checkpointing, +// and returns the actor as Substrate reports it afterwards. +func (c *Client) SuspendActor(ctx context.Context, atespace, actorName string) (*ateapipb.Actor, error) { req := &ateapipb.SuspendActorRequest{ Actor: &ateapipb.ObjectRef{ Atespace: atespace, Name: actorName, }, } + var resp *ateapipb.SuspendActorResponse err := retryOnCodes(ctx, func() error { - _, err := c.control.SuspendActor(ctx, req) + var err error + resp, err = c.control.SuspendActor(ctx, req) return err }, codes.Aborted, codes.Unavailable) if err != nil { - return fmt.Errorf("suspending actor %s/%s: %w", atespace, actorName, err) + return nil, fmt.Errorf("suspending actor %s/%s: %w", atespace, actorName, err) } - return nil + return resp.GetActor(), nil } // GetActor fetches an Actor by name. diff --git a/internal/substrate/client_test.go b/internal/substrate/client_test.go index 30295109..2230958c 100644 --- a/internal/substrate/client_test.go +++ b/internal/substrate/client_test.go @@ -39,6 +39,8 @@ type mockControlClient struct { actorRemaining int // Number of GetActor calls before returning NotFound resumeErrs []error resumeActor *ateapipb.Actor + suspendErrs []error + suspendActor *ateapipb.Actor deleteTemplateErrs []error deleteTemplateCalls int templates map[string]*ateapipb.ActorTemplate @@ -83,6 +85,19 @@ func (m *mockControlClient) ResumeActor(ctx context.Context, req *ateapipb.Resum return &ateapipb.ResumeActorResponse{Actor: m.resumeActor, Resumed: true}, nil } +func (m *mockControlClient) SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRequest, opts ...grpc.CallOption) (*ateapipb.SuspendActorResponse, error) { + m.mu.Lock() + defer m.mu.Unlock() + if len(m.suspendErrs) > 0 { + err := m.suspendErrs[0] + m.suspendErrs = m.suspendErrs[1:] + if err != nil { + return nil, err + } + } + return &ateapipb.SuspendActorResponse{Actor: m.suspendActor}, nil +} + func (m *mockControlClient) DeleteActorTemplate(ctx context.Context, req *ateapipb.DeleteActorTemplateRequest, opts ...grpc.CallOption) (*ateapipb.ActorTemplate, error) { m.mu.Lock() defer m.mu.Unlock() @@ -266,6 +281,53 @@ func TestClient_ResumeActor_WorkerPodIpsAndRetry(t *testing.T) { } } +func TestClient_SuspendActor_ReturnsActorAfterRetry(t *testing.T) { + oldInterval := defaultRetryInterval + defaultRetryInterval = 5 * time.Millisecond + defer func() { defaultRetryInterval = oldInterval }() + + mock := &mockControlClient{ + suspendErrs: []error{status.Error(codes.Aborted, "operation aborted"), nil}, + suspendActor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: "default", Name: "task-1", Uid: "uid-1"}, + Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED}, + }, + } + c := &Client{control: mock} + actor, err := c.SuspendActor(context.Background(), "default", "task-1") + if err != nil { + t.Fatalf("SuspendActor failed: %v", err) + } + if actor.GetMetadata().GetUid() != "uid-1" || Residency(actor) != v1alpha1.ResidencySuspended { + t.Errorf("SuspendActor returned %v, want the suspended actor Substrate returned", actor) + } +} + +func TestResidency(t *testing.T) { + tests := []struct { + state ateapipb.ActorState + want string + }{ + {ateapipb.ActorState_ACTOR_STATE_RESUMING, v1alpha1.ResidencyResuming}, + {ateapipb.ActorState_ACTOR_STATE_RUNNING, v1alpha1.ResidencyRunning}, + {ateapipb.ActorState_ACTOR_STATE_SUSPENDING, v1alpha1.ResidencySuspending}, + {ateapipb.ActorState_ACTOR_STATE_SUSPENDED, v1alpha1.ResidencySuspended}, + {ateapipb.ActorState_ACTOR_STATE_REVERTING, v1alpha1.ResidencyReverting}, + {ateapipb.ActorState_ACTOR_STATE_CRASHED, v1alpha1.ResidencyCrashed}, + {ateapipb.ActorState_ACTOR_STATE_DELETING, v1alpha1.ResidencyDeleting}, + {ateapipb.ActorState_ACTOR_STATE_UNSPECIFIED, ""}, + } + for _, tt := range tests { + actor := &ateapipb.Actor{Status: &ateapipb.ActorStatus{State: tt.state}} + if got := Residency(actor); got != tt.want { + t.Errorf("Residency(%v) = %q, want %q", tt.state, got, tt.want) + } + } + if got := Residency(nil); got != "" { + t.Errorf("Residency(nil) = %q, want empty", got) + } +} + func TestClient_ResumeActor_ResourceExhaustedNotRetried(t *testing.T) { mock := &mockControlClient{ resumeErrs: []error{ diff --git a/internal/substrate/substratetest/server.go b/internal/substrate/substratetest/server.go index db4e6015..e2fae6b9 100644 --- a/internal/substrate/substratetest/server.go +++ b/internal/substrate/substratetest/server.go @@ -18,6 +18,7 @@ package substratetest import ( "context" + "fmt" "net" "sync" "testing" @@ -59,6 +60,7 @@ type Server struct { errs Errors onDeleteActor func() deleteActorCalls int + nextUID int // numbers the UIDs given to created actors } // NewServer starts a Server with nothing in it. It stops when the test ends. @@ -244,6 +246,8 @@ func (s *Server) CreateActor(_ context.Context, req *ateapipb.CreateActorRequest return nil, status.Errorf(codes.AlreadyExists, "actor %s already exists", k) } actor := proto.Clone(req.GetActor()).(*ateapipb.Actor) + s.nextUID++ + actor.Metadata.Uid = fmt.Sprintf("actor-uid-%d", s.nextUID) actor.Status = &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED} s.actors[k] = actor return proto.Clone(actor).(*ateapipb.Actor), nil @@ -260,7 +264,7 @@ func (s *Server) SuspendActor(_ context.Context, req *ateapipb.SuspendActorReque return nil, err } actor.Status = &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED} - return &ateapipb.SuspendActorResponse{}, nil + return &ateapipb.SuspendActorResponse{Actor: proto.Clone(actor).(*ateapipb.Actor)}, nil } func (s *Server) ResumeActor(_ context.Context, req *ateapipb.ResumeActorRequest) (*ateapipb.ResumeActorResponse, error) { diff --git a/pkg/apis/v1alpha1/task.pb.go b/pkg/apis/v1alpha1/task.pb.go index ac7fc149..0f223eba 100644 --- a/pkg/apis/v1alpha1/task.pb.go +++ b/pkg/apis/v1alpha1/task.pb.go @@ -274,8 +274,23 @@ type TaskStatus struct { Id string `protobuf:"bytes,2,opt,name=id,proto3" json:"id,omitempty"` Actor string `protobuf:"bytes,3,opt,name=actor,proto3" json:"actor,omitempty"` // json_name keeps the established `workerIP` spelling in YAML and JSON. - WorkerIp string `protobuf:"bytes,4,opt,name=worker_ip,json=workerIP,proto3" json:"worker_ip,omitempty"` - Conditions []*Condition `protobuf:"bytes,7,rep,name=conditions,proto3" json:"conditions,omitempty"` + WorkerIp string `protobuf:"bytes,4,opt,name=worker_ip,json=workerIP,proto3" json:"worker_ip,omitempty"` + Conditions []*Condition `protobuf:"bytes,7,rep,name=conditions,proto3" json:"conditions,omitempty"` + // residency is where the task's actor is in its lifecycle on Agent + // Substrate: queued (no actor yet), resuming, running, suspending, + // suspended, reverting, crashed, deleting, or deleted. Unlike phase, it + // tells a crashed or reverting actor apart from a failed task. Reads report + // the actor's current state; the stored value is the one Substrate last + // returned to ax-server. + Residency string `protobuf:"bytes,8,opt,name=residency,proto3" json:"residency,omitempty"` + // actor_uid is the UID Agent Substrate assigned to the task's actor, the + // ate.actor.uid that Substrate's telemetry carries. It changes when the + // actor is recreated. + ActorUid string `protobuf:"bytes,9,opt,name=actor_uid,json=actorUid,proto3" json:"actor_uid,omitempty"` + // outcome is how the task's command ended: succeeded, failed, or + // cancelled. It is empty while the command has not ended. AX does not set + // it yet; command exit detection (#346) will. + Outcome string `protobuf:"bytes,10,opt,name=outcome,proto3" json:"outcome,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -345,6 +360,27 @@ func (x *TaskStatus) GetConditions() []*Condition { return nil } +func (x *TaskStatus) GetResidency() string { + if x != nil { + return x.Residency + } + return "" +} + +func (x *TaskStatus) GetActorUid() string { + if x != nil { + return x.ActorUid + } + return "" +} + +func (x *TaskStatus) GetOutcome() string { + if x != nil { + return x.Outcome + } + return "" +} + // Tasks type GetTaskRequest struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -760,7 +796,7 @@ const file_pkg_apis_v1alpha1_task_proto_rawDesc = "" + "\x06config\x18\a \x01(\v2\x17.google.protobuf.StructR\x06config\"2\n" + "\x06EnvVar\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value\"\x9d\x01\n" + + "\x05value\x18\x02 \x01(\tR\x05value\"\xf2\x01\n" + "\n" + "TaskStatus\x12\x14\n" + "\x05phase\x18\x01 \x01(\tR\x05phase\x12\x0e\n" + @@ -769,7 +805,11 @@ const file_pkg_apis_v1alpha1_task_proto_rawDesc = "" + "\tworker_ip\x18\x04 \x01(\tR\bworkerIP\x126\n" + "\n" + "conditions\x18\a \x03(\v2\x16.ax.v1alpha1.ConditionR\n" + - "conditions\"@\n" + + "conditions\x12\x1c\n" + + "\tresidency\x18\b \x01(\tR\tresidency\x12\x1b\n" + + "\tactor_uid\x18\t \x01(\tR\bactorUid\x12\x18\n" + + "\aoutcome\x18\n" + + " \x01(\tR\aoutcome\"@\n" + "\x0eGetTaskRequest\x12\x1a\n" + "\batespace\x18\x01 \x01(\tR\batespace\x12\x12\n" + "\x04name\x18\x02 \x01(\tR\x04name\"\\\n" + diff --git a/pkg/apis/v1alpha1/task.proto b/pkg/apis/v1alpha1/task.proto index 561d8bf0..53c0af6b 100644 --- a/pkg/apis/v1alpha1/task.proto +++ b/pkg/apis/v1alpha1/task.proto @@ -64,6 +64,21 @@ message TaskStatus { // json_name keeps the established `workerIP` spelling in YAML and JSON. string worker_ip = 4 [json_name = "workerIP"]; repeated Condition conditions = 7; + // residency is where the task's actor is in its lifecycle on Agent + // Substrate: queued (no actor yet), resuming, running, suspending, + // suspended, reverting, crashed, deleting, or deleted. Unlike phase, it + // tells a crashed or reverting actor apart from a failed task. Reads report + // the actor's current state; the stored value is the one Substrate last + // returned to ax-server. + string residency = 8; + // actor_uid is the UID Agent Substrate assigned to the task's actor, the + // ate.actor.uid that Substrate's telemetry carries. It changes when the + // actor is recreated. + string actor_uid = 9; + // outcome is how the task's command ended: succeeded, failed, or + // cancelled. It is empty while the command has not ended. AX does not set + // it yet; command exit detection (#346) will. + string outcome = 10; } // --- RPC Request & Response Messages --- diff --git a/pkg/apis/v1alpha1/types.go b/pkg/apis/v1alpha1/types.go index dc3494ec..bc14d01f 100644 --- a/pkg/apis/v1alpha1/types.go +++ b/pkg/apis/v1alpha1/types.go @@ -65,6 +65,22 @@ const ( PhaseUnknown = "Unknown" ) +// Residencies reported in Task status. Each one but ResidencyQueued and +// ResidencyDeleted names a state of the task's actor on Agent Substrate. +const ( + // ResidencyQueued marks a task whose actor has not been created yet. + ResidencyQueued = "queued" + ResidencyResuming = "resuming" + ResidencyRunning = "running" + ResidencySuspending = "suspending" + ResidencySuspended = "suspended" + ResidencyReverting = "reverting" + ResidencyCrashed = "crashed" + ResidencyDeleting = "deleting" + // ResidencyDeleted marks a task whose actor existed and is gone. + ResidencyDeleted = "deleted" +) + // Condition types reported on Task status. const ( // ConditionReady reports whether the task as a whole is ready to do work: diff --git a/skills/substrate/SKILL.md b/skills/substrate/SKILL.md index d236e54d..a1dbc042 100644 --- a/skills/substrate/SKILL.md +++ b/skills/substrate/SKILL.md @@ -37,7 +37,7 @@ All Control API traffic goes through this package. It encapsulates: - **Retry** — `IsTransientError` + `retryOnCodes` for `Unavailable`-class errors. - **Helpers** the reconciler relies on — `EnsureAtespace`, `EnsureActorTemplateWithImage`, `EnsureActor`, `SuspendActor`, `ResumeActor`, `DeleteActor`, `WorkerIP`, - `ResourceLimits`, `BuildActorTemplate`. + `Residency`, `ResourceLimits`, `BuildActorTemplate`. ### Where a new Control API call lives today @@ -95,6 +95,12 @@ instead. No mocking library is in `go.mod`. (`substrate.WorkerIP` in `internal/server/server.go`), and the reconciler uses the IP `ResumeActor` returns only for its readiness probe (`internal/controller/reconciler.go`). +- `TaskStatus.residency` and `actorUid` are the exception to that rule: the + reconciler records them from the actor that `CreateActor`, `SuspendActor` and + `ResumeActor` return (`recordActor` in `internal/controller/reconciler.go`), + the store keeps them, and reads overlay the live actor's values + (`setActorStatus` in `internal/server/server.go`). `substratetest` assigns + each created actor a UID (`actor-uid-N`) for this. - Guest services (process + fs) are mounted by `internal/metadata/server.go` only when `task.Spec.Debug` is true; `ax ssh` (`cmd/ax/main.go`) refuses to connect otherwise. diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index 0962cee2..78ab7b9f 100644 --- a/test/e2e/e2e_test.go +++ b/test/e2e/e2e_test.go @@ -210,6 +210,7 @@ func TestE2E_TaskLifecycleAndSubstrateSync(t *testing.T) { if got := task.GetStatus().GetActor(); got != name { t.Fatalf("task actor = %q, want %q", got, name) } + assertTaskRow(t, env.mustRunAX(t, ns, "get", "tasks"), name, phaseSuspended, v1alpha1.ResidencySuspended, "") // 2. Tasks are immutable: both CreateTask and `ax apply` reject a duplicate. _, err := env.AXClient.CreateTask(ctx, &v1alpha1.CreateTaskRequest{Task: &v1alpha1.Task{ @@ -225,8 +226,13 @@ func TestE2E_TaskLifecycleAndSubstrateSync(t *testing.T) { } assertContains(t, "duplicate 'ax apply'", out, "already exists and is immutable") - // 3. Substrate has a suspended Actor using a per-task ActorTemplate. + // 3. Substrate has a suspended Actor using a per-task ActorTemplate, and + // the task carries the Actor's UID. actor := env.getActor(t, ns, name, ateapipb.ActorState_ACTOR_STATE_SUSPENDED) + actorUID := actor.GetMetadata().GetUid() + if got := task.GetStatus().GetActorUid(); actorUID == "" || got != actorUID { + t.Fatalf("task actorUid = %q, want the Substrate actor's UID %q", got, actorUID) + } tmplName := actor.GetActorTemplate().GetName() if !strings.HasPrefix(tmplName, name+"-tmpl-") { t.Fatalf("actor template = %q, want prefix %q", tmplName, name+"-tmpl-") @@ -260,6 +266,9 @@ func TestE2E_TaskLifecycleAndSubstrateSync(t *testing.T) { if ips := actor.GetStatus().GetWorkerAssignment().GetWorkerPodIps(); len(ips) == 0 || ips[0] != workerIP { t.Fatalf("Substrate WorkerPodIps = %v, want first IP %q", ips, workerIP) } + assertTaskRow(t, env.mustRunAX(t, ns, "get", "tasks"), name, phaseRunning, v1alpha1.ResidencyRunning, "") + assertContains(t, "'ax describe task'", env.mustRunAX(t, ns, "describe", "task", name), + "Residency: running\n", "Outcome: \n", "Actor UID: "+actorUID+"\n") // 5. Deleting the task removes the AX task, the Actor, and its ActorTemplate. assertContains(t, "'ax delete task'", env.mustRunAX(t, ns, "delete", "task", name), "deleted") @@ -482,6 +491,7 @@ func TestE2E_TaskSuspendAndResumePersistence(t *testing.T) { if ip := got.GetStatus().GetWorkerIp(); ip != "" { t.Fatalf("expected empty WorkerIp when suspended, got %q", ip) } + assertTaskRow(t, env.mustRunAX(t, ns, "get", "tasks"), task, phaseSuspended, v1alpha1.ResidencySuspended, "") actor := env.getActor(t, ns, task, ateapipb.ActorState_ACTOR_STATE_SUSPENDED) if len(actor.GetStatus().GetSnapshots()) == 0 && len(actor.GetStatus().ProtoReflect().GetUnknown()) == 0 { t.Fatal("expected non-empty Snapshots on suspended actor") diff --git a/test/e2e/harness_test.go b/test/e2e/harness_test.go index d72f7c92..2e56dbcb 100644 --- a/test/e2e/harness_test.go +++ b/test/e2e/harness_test.go @@ -434,6 +434,30 @@ func assertContains(t *testing.T, what, out string, wants ...string) { } } +// assertTaskRow fails the test unless the `ax get tasks` output has a row for +// name with the given PHASE, RESIDENCY and OUTCOME columns. +func assertTaskRow(t *testing.T, out, name, phase, residency, outcome string) { + t.Helper() + lines := strings.Split(strings.TrimSpace(out), "\n") + header := strings.Fields(lines[0]) + for _, line := range lines[1:] { + row := strings.Fields(line) + if len(row) != len(header) || row[0] != name { + continue + } + got := make(map[string]string, len(header)) + for i, col := range header { + got[col] = row[i] + } + if got["PHASE"] != phase || got["RESIDENCY"] != residency || got["OUTCOME"] != outcome { + t.Fatalf("'ax get tasks' row for %s has PHASE=%q RESIDENCY=%q OUTCOME=%q, want %q %q %q:\n%s", + name, got["PHASE"], got["RESIDENCY"], got["OUTCOME"], phase, residency, outcome, out) + } + return + } + t.Fatalf("'ax get tasks' has no row for %s:\n%s", name, out) +} + func (s *suiteEnv) getTask(t *testing.T, atespace, name string) *v1alpha1.Task { t.Helper() task, err := s.AXClient.GetTask(t.Context(), &v1alpha1.GetTaskRequest{Name: name, Atespace: atespace})