From 05bda216af81fa822f2ab0be6979455f1e63f01b Mon Sep 17 00:00:00 2001 From: Yu Yi Date: Thu, 5 Nov 2026 14:53:46 -0500 Subject: [PATCH] api,cli: report task residency, actor UID, and outcome Add residency, actor_uid, and outcome to TaskStatus and show RESIDENCY and OUTCOME in `ax get tasks`, and Residency, Outcome, and Actor UID in `ax describe task`. Residency names where the task's actor is in its Substrate lifecycle: queued, resuming, running, suspending, suspended, reverting, crashed, deleting, or deleted. Unlike phase, it tells a crashed or reverting actor apart from a task AX failed to provision. actor_uid is the actor's ate.actor.uid, which Substrate's logs and telemetry carry. The reconciler records both from the Actor that CreateActor, SuspendActor, and ResumeActor return, and the store keeps them, so each change AX makes is a stored change that state-change telemetry can observe. Reads still overlay the live actor, so a wake by the router or a recreated actor shows up on the next get. Phase is unchanged. outcome is defined but not set yet; command exit detection (#346) will set it to succeeded, failed, or cancelled. --- DESIGN.md | 2 +- README.md | 4 +- cmd/ax/main.go | 27 ++++-- cmd/ax/main_test.go | 69 +++++++++++++ docs/concepts.md | 2 + internal/controller/reconciler.go | 66 ++++++++----- internal/server/actor_status_test.go | 39 ++++++++ internal/server/server.go | 25 ++++- internal/server/server_test.go | 107 +++++++++++++++++++++ internal/server/status_test.go | 5 + internal/store/postgres/store_test.go | 29 ++++++ internal/substrate/client.go | 36 ++++++- internal/substrate/client_test.go | 62 ++++++++++++ internal/substrate/substratetest/server.go | 6 +- pkg/apis/v1alpha1/task.pb.go | 48 ++++++++- pkg/apis/v1alpha1/task.proto | 15 +++ pkg/apis/v1alpha1/types.go | 16 +++ skills/substrate/SKILL.md | 8 +- test/e2e/e2e_test.go | 12 ++- test/e2e/harness_test.go | 24 +++++ 20 files changed, 550 insertions(+), 52 deletions(-) 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})