From dfd7cfaa2b77d09a8af8df1cac825419e5dfb898 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 2 Jul 2026 21:12:30 +1000 Subject: [PATCH] fix: ECS StopTask reaps container before releasing capacity (siv-484) (#470) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: ECS StopTask reaps container before releasing capacity StopTask marked a task lastStatus=STOPPED and released its capacity/ENI in the control plane, but nothing ever told the ecs-agent to kill the container. The container kept holding its host ports, so DescribeTasks reported a STOPPED task whose container was still RUNNING, and a replacement task placed on the freed capacity that rebound the same host port hit EADDRINUSE. Wire the agent-driven cooperative stop the design intended (ecs-v1.md:165): - Add a per-instance stop inbox (StopsPrefix/StopKey) and bus.StopDirective. - requestStopTask marks DesiredStatus=STOPPED and posts a stop directive, leaving lastStatus/capacity/ENI/targets untouched. StopTask, DeleteService, and service scale-in use it; the heartbeat reaper and cluster/instance teardown keep the immediate forceStopTask since no live agent can reap. - PollAssignments returns pending stops; the agent lists containers by mulga.ecs.taskID label, Removes them (kill+delete, freeing ports/netns), then reports STOPPED. recordTaskState performs the single capacity release on the agent's STOPPED report and reclaims the stop inbox. - listServiceTasks excludes DesiredStatus=STOPPED tasks so a cooperatively stopped task drops out of the service's running total immediately. DescribeTasks now keeps the task RUNNING with desiredStatus=STOPPED until the container is actually reaped, matching AWS ordering and eliminating the port race. * test: cover StopTask reap paths — poll stop-inbox drain, unplaced force-stop, agent stopTask edges * ci: retrigger PR checks --- cmd/ecs-agent/controlplane.go | 24 ++-- cmd/ecs-agent/registration_test.go | 30 +++-- cmd/ecs-agent/taskrun.go | 57 +++++++++- cmd/ecs-agent/taskrun_test.go | 145 +++++++++++++++++++++++++ spinifex/handlers/ecs/bus/messages.go | 9 ++ spinifex/handlers/ecs/instances.go | 1 + spinifex/handlers/ecs/poll.go | 40 ++++++- spinifex/handlers/ecs/poll_test.go | 46 ++++++++ spinifex/handlers/ecs/services.go | 13 ++- spinifex/handlers/ecs/services_test.go | 47 +++++++- spinifex/handlers/ecs/store.go | 13 +++ spinifex/handlers/ecs/task_stop.go | 44 +++++++- 12 files changed, 430 insertions(+), 39 deletions(-) create mode 100644 spinifex/handlers/ecs/poll_test.go diff --git a/cmd/ecs-agent/controlplane.go b/cmd/ecs-agent/controlplane.go index 46b7c564b..c72009649 100644 --- a/cmd/ecs-agent/controlplane.go +++ b/cmd/ecs-agent/controlplane.go @@ -25,9 +25,10 @@ type controlPlane interface { Register(id identity) error // SubmitTaskState reports a task's RUNNING/STOPPED transition. SubmitTaskState(st bus.TaskState) error - // PollAssignments drains this instance's assignment inbox, acking the taskIDs - // accepted on the previous poll. Returns the still-pending assignments. - PollAssignments(cluster, instance string, ack []string) ([]bus.Assign, error) + // PollAssignments drains this instance's assignment + stop inboxes, acking the + // assign taskIDs and stop taskIDs accepted on the previous poll. Returns the + // still-pending assignments and stop directives. + PollAssignments(cluster, instance string, ackAssigns, ackStops []string) ([]bus.Assign, []bus.StopDirective, error) } // gatewayControlPlane implements controlPlane over the SigV4 ECS gateway client. @@ -111,19 +112,24 @@ func (g *gatewayControlPlane) SubmitTaskState(st bus.TaskState) error { return nil } -func (g *gatewayControlPlane) PollAssignments(cluster, instance string, ack []string) ([]bus.Assign, error) { - in := &handlers_ecs.PollAssignmentsInput{Cluster: cluster, ContainerInstance: instance, AckTaskIDs: ack} +func (g *gatewayControlPlane) PollAssignments(cluster, instance string, ackAssigns, ackStops []string) ([]bus.Assign, []bus.StopDirective, error) { + in := &handlers_ecs.PollAssignmentsInput{ + Cluster: cluster, + ContainerInstance: instance, + AckTaskIDs: ackAssigns, + AckStopIDs: ackStops, + } body, err := json.Marshal(in) if err != nil { - return nil, fmt.Errorf("marshal poll: %w", err) + return nil, nil, fmt.Errorf("marshal poll: %w", err) } resp, err := g.client.Call("PollAssignments", body) if err != nil { - return nil, fmt.Errorf("poll: %w", err) + return nil, nil, fmt.Errorf("poll: %w", err) } var out handlers_ecs.PollAssignmentsOutput if err := json.Unmarshal(resp, &out); err != nil { - return nil, fmt.Errorf("decode poll: %w", err) + return nil, nil, fmt.Errorf("decode poll: %w", err) } - return out.Assignments, nil + return out.Assignments, out.Stops, nil } diff --git a/cmd/ecs-agent/registration_test.go b/cmd/ecs-agent/registration_test.go index 3b4e34e4e..03bc03956 100644 --- a/cmd/ecs-agent/registration_test.go +++ b/cmd/ecs-agent/registration_test.go @@ -15,11 +15,13 @@ import ( type fakeCP struct { mu sync.Mutex - registers int - states []bus.TaskState - pollAcks [][]string - pollReplies [][]bus.Assign - pollCalls int + registers int + states []bus.TaskState + pollAcks [][]string + pollStopAcks [][]string + pollReplies [][]bus.Assign + stopReplies [][]bus.StopDirective + pollCalls int registerErr bool submitErr bool @@ -47,17 +49,21 @@ func (f *fakeCP) SubmitTaskState(st bus.TaskState) error { return nil } -func (f *fakeCP) PollAssignments(_, _ string, ack []string) ([]bus.Assign, error) { +func (f *fakeCP) PollAssignments(_, _ string, ackAssigns, ackStops []string) ([]bus.Assign, []bus.StopDirective, error) { f.mu.Lock() defer f.mu.Unlock() - cp := append([]string(nil), ack...) - f.pollAcks = append(f.pollAcks, cp) + f.pollAcks = append(f.pollAcks, append([]string(nil), ackAssigns...)) + f.pollStopAcks = append(f.pollStopAcks, append([]string(nil), ackStops...)) var out []bus.Assign if f.pollCalls < len(f.pollReplies) { out = f.pollReplies[f.pollCalls] } + var stops []bus.StopDirective + if f.pollCalls < len(f.stopReplies) { + stops = f.stopReplies[f.pollCalls] + } f.pollCalls++ - return out, nil + return out, stops, nil } func (f *fakeCP) registerCount() int { @@ -78,6 +84,12 @@ func (f *fakeCP) acks() [][]string { return append([][]string(nil), f.pollAcks...) } +func (f *fakeCP) stopAcks() [][]string { + f.mu.Lock() + defer f.mu.Unlock() + return append([][]string(nil), f.pollStopAcks...) +} + func testIdentity() identity { return identity{ AccountID: "123456789012", diff --git a/cmd/ecs-agent/taskrun.go b/cmd/ecs-agent/taskrun.go index 906626262..198f3d9b4 100644 --- a/cmd/ecs-agent/taskrun.go +++ b/cmd/ecs-agent/taskrun.go @@ -22,31 +22,78 @@ func (a *Agent) pollAssignments(ctx context.Context, dispatched map[string]bool) ticker := time.NewTicker(a.cfg.PollInterval) defer ticker.Stop() - var ackNext []string + stopping := map[string]bool{} + var ackAssigns, ackStops []string for { select { case <-ctx.Done(): return case <-ticker.C: - assigns, err := a.cp.PollAssignments(a.id.ClusterName, a.id.InstanceID, ackNext) + assigns, stops, err := a.cp.PollAssignments(a.id.ClusterName, a.id.InstanceID, ackAssigns, ackStops) if err != nil { slog.Warn("ecs-agent: poll assignments failed", "err", err) continue } - ackNext = ackNext[:0] + // Dispatch stops first so a task stopped in the same poll it was assigned + // is never started. + ackStops = ackStops[:0] + for i := range stops { + sd := stops[i] + if !stopping[sd.TaskID] { + stopping[sd.TaskID] = true + slog.Info("ecs-agent: task stop requested", "task", sd.TaskID, "reason", sd.Reason) + go a.stopTask(ctx, sd) + } + ackStops = append(ackStops, sd.TaskID) + } + ackAssigns = ackAssigns[:0] for i := range assigns { as := assigns[i] - if !dispatched[as.TaskID] { + if !dispatched[as.TaskID] && !stopping[as.TaskID] { dispatched[as.TaskID] = true slog.Info("ecs-agent: task assigned", "task", as.TaskID, "containers", len(as.Containers)) go a.runTask(ctx, &as) } - ackNext = append(ackNext, as.TaskID) + ackAssigns = append(ackAssigns, as.TaskID) } } } } +// stopTask reaps a task's containers on a control-plane stop directive: it lists +// the runtime's containers, removes the ones labeled with this task (kill + +// delete, releasing host ports/netns), then reports the task STOPPED with the +// directive's reason. Idempotent — a task with no live containers still reports +// STOPPED so the scheduler releases its capacity. +func (a *Agent) stopTask(ctx context.Context, sd bus.StopDirective) { + if a.runner == nil { + return + } + containers, err := a.runner.List(ctx) + if err != nil { + slog.Warn("ecs-agent: stop list failed", "task", sd.TaskID, "err", err) + return + } + statuses := make([]bus.ContainerStatus, 0) + for _, c := range containers { + if c.Labels[labelTaskID] != sd.TaskID { + continue + } + if rerr := a.runner.Remove(ctx, c.ID); rerr != nil { + slog.Warn("ecs-agent: stop remove failed", "task", sd.TaskID, "container", c.ID, "err", rerr) + } + statuses = append(statuses, bus.ContainerStatus{ + Name: c.Labels[labelContainerName], Status: bus.TaskStatusStopped, ContainerID: c.ID, + }) + } + reason := sd.Reason + if reason == "" { + reason = "Task stopped" + } + a.reportTaskState(&bus.Assign{TaskID: sd.TaskID}, bus.TaskStatusStopped, reason, statuses) + slog.Info("ecs-agent: task reaped", "task", sd.TaskID, "containers", len(statuses)) +} + // runTask pulls each container image, starts the containers, and reports task // state on the bus. A missing runtime or any per-container failure reports the // task STOPPED with a reason; success reports RUNNING and waits for exit. diff --git a/cmd/ecs-agent/taskrun_test.go b/cmd/ecs-agent/taskrun_test.go index bba22854a..d264846cb 100644 --- a/cmd/ecs-agent/taskrun_test.go +++ b/cmd/ecs-agent/taskrun_test.go @@ -172,6 +172,151 @@ func TestPollAssignments_DispatchesOnceAndAcks(t *testing.T) { } } +// stopTask reaps the task's labeled containers and reports STOPPED with the +// directive reason and container status. +func TestStopTask_ReapsLabeledContainersAndReportsStopped(t *testing.T) { + cp := &fakeCP{} + rt := &ctrruntime.FakePuller{Containers: []ctrruntime.Container{ + {ID: "t-001-web", Running: true, Labels: map[string]string{ + "mulga.ecs.taskID": "t-001", "mulga.ecs.containerName": "web", + }}, + {ID: "t-999-other", Running: true, Labels: map[string]string{ + "mulga.ecs.taskID": "t-999", "mulga.ecs.containerName": "web", + }}, + }} + a := newRunAgent(cp, rt) + + a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "bye"}) + + if len(rt.Removed) != 1 || rt.Removed[0] != "t-001-web" { + t.Fatalf("want only t-001-web removed, got %+v", rt.Removed) + } + st := cp.taskStates() + if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped { + t.Fatalf("want one STOPPED report, got %+v", st) + } + if st[0].Reason != "bye" { + t.Errorf("want reason %q, got %q", "bye", st[0].Reason) + } + if len(st[0].Containers) != 1 || st[0].Containers[0].Status != bus.TaskStatusStopped { + t.Errorf("want one STOPPED container, got %+v", st[0].Containers) + } +} + +// A stop for a task with no live containers still reports STOPPED so the +// scheduler releases its capacity. +func TestStopTask_NoContainersStillReportsStopped(t *testing.T) { + cp := &fakeCP{} + rt := &ctrruntime.FakePuller{} + a := newRunAgent(cp, rt) + + a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "gone"}) + + if len(rt.Removed) != 0 { + t.Errorf("nothing to remove, got %+v", rt.Removed) + } + st := cp.taskStates() + if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped { + t.Fatalf("want one STOPPED report, got %+v", st) + } +} + +// A stop delivered by poll is dispatched once, acked in the stop-ack list, and +// suppresses a same-poll assign for the same task. +func TestPollAssignments_StopReapsAndSuppressesAssign(t *testing.T) { + as := testAssign() + cp := &fakeCP{ + pollReplies: [][]bus.Assign{{*as}, {*as}, nil}, + stopReplies: [][]bus.StopDirective{{{TaskID: as.TaskID, Reason: "bye"}}}, + } + rt := &ctrruntime.FakePuller{Containers: []ctrruntime.Container{ + {ID: "t-001-web", Running: true, Labels: map[string]string{"mulga.ecs.taskID": "t-001"}}, + }} + a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, rt, nil) + + ctx, cancel := context.WithCancel(context.Background()) + go a.pollAssignments(ctx, map[string]bool{}) + + deadline := time.After(time.Second) + for { + if stopped := countStatus(cp.taskStates(), as.TaskID, bus.TaskStatusStopped); stopped >= 1 { + break + } + select { + case <-deadline: + t.Fatalf("stop never reaped; states=%+v", cp.taskStates()) + case <-time.After(2 * time.Millisecond): + } + } + time.Sleep(20 * time.Millisecond) + cancel() + + // The assign for the stopping task must never have started a container. + if len(rt.Runs) != 0 { + t.Errorf("assign for a stopping task should not run, got %+v", rt.Runs) + } + if running := countStatus(cp.taskStates(), as.TaskID, bus.TaskStatusRunning); running != 0 { + t.Errorf("stopping task should never report RUNNING, got %d", running) + } + acked := false + for _, ack := range cp.stopAcks() { + for _, id := range ack { + if id == as.TaskID { + acked = true + } + } + } + if !acked { + t.Errorf("stop %s was never acked; stopAcks=%v", as.TaskID, cp.stopAcks()) + } +} + +// stopTask with no runtime is a no-op: nothing to reap, no state reported. +func TestStopTask_NilRunnerNoOp(t *testing.T) { + cp := &fakeCP{} + a := newAgent(config{}, testIdentity(), cp, nil, nil, nil) + a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "x"}) + if len(cp.taskStates()) != 0 { + t.Fatalf("nil runner should report no state, got %+v", cp.taskStates()) + } +} + +// A List failure aborts the reap without reporting a (false) STOPPED state. +func TestStopTask_ListErrorNoReport(t *testing.T) { + cp := &fakeCP{} + rt := &ctrruntime.FakePuller{ListErr: errors.New("list boom")} + a := newRunAgent(cp, rt) + a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "x"}) + if len(cp.taskStates()) != 0 { + t.Fatalf("list error should report no state, got %+v", cp.taskStates()) + } +} + +// A directive with no reason reaps the matching container and reports STOPPED with +// the default reason. +func TestStopTask_EmptyReasonReapsWithDefault(t *testing.T) { + cp := &fakeCP{} + rt := &ctrruntime.FakePuller{Containers: []ctrruntime.Container{{ + ID: "t-001-web", + Labels: map[string]string{ + "mulga.ecs.taskID": "t-001", "mulga.ecs.containerName": "web", + }, + }}} + a := newRunAgent(cp, rt) + a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001"}) + + if len(rt.Removed) != 1 || rt.Removed[0] != "t-001-web" { + t.Fatalf("want container t-001-web reaped, got %+v", rt.Removed) + } + st := cp.taskStates() + if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped { + t.Fatalf("want one STOPPED report, got %+v", st) + } + if st[0].Reason != "Task stopped" { + t.Errorf("want default reason %q, got %q", "Task stopped", st[0].Reason) + } +} + func countStatus(states []bus.TaskState, taskID, status string) int { n := 0 for _, st := range states { diff --git a/spinifex/handlers/ecs/bus/messages.go b/spinifex/handlers/ecs/bus/messages.go index 2d564ad8b..af346d38e 100644 --- a/spinifex/handlers/ecs/bus/messages.go +++ b/spinifex/handlers/ecs/bus/messages.go @@ -101,6 +101,15 @@ type Assign struct { AssignedAt time.Time `json:"assignedAt"` } +// StopDirective is one pending task-stop the scheduler places in an instance's +// stop inbox (scheduler → agent). The agent kills + reaps the task's containers, +// then reports STOPPED with Reason. Delivery is idempotent; the STOPPED +// transition deletes the inbox entry. +type StopDirective struct { + TaskID string `json:"taskId"` + Reason string `json:"reason,omitempty"` +} + // ContainerStatus is a single container's reported lifecycle state. type ContainerStatus struct { Name string `json:"name"` diff --git a/spinifex/handlers/ecs/instances.go b/spinifex/handlers/ecs/instances.go index 799a64218..b9ae90eba 100644 --- a/spinifex/handlers/ecs/instances.go +++ b/spinifex/handlers/ecs/instances.go @@ -251,6 +251,7 @@ func (s *Service) recordTaskState(msg *bus.TaskState) error { s.deregisterServiceTargets(kv, msg.AccountID, &task) s.releaseTaskPublicIP(msg.AccountID, &task) s.reclaimAssignInbox(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID) + s.reclaimStopInbox(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID) s.reclaimTaskENI(msg.AccountID, &task) if perr := putJSON(kv, TaskKey(msg.ClusterName, msg.TaskID), &task); perr != nil { slog.Error("ECS task STOPPED: persist after EIP release failed", "task", msg.TaskID, "err", perr) diff --git a/spinifex/handlers/ecs/poll.go b/spinifex/handlers/ecs/poll.go index 33e535648..d7ad7c664 100644 --- a/spinifex/handlers/ecs/poll.go +++ b/spinifex/handlers/ecs/poll.go @@ -14,11 +14,17 @@ type PollAssignmentsInput struct { Cluster string `json:"cluster"` ContainerInstance string `json:"containerInstance"` AckTaskIDs []string `json:"ackTaskIds,omitempty"` + // AckStopIDs are stop directives the agent reaped on a prior poll; they may now + // be removed. Optional — stop delivery is idempotent and the STOPPED transition + // deletes the entry, so a dropped ack only re-delivers a no-op reap. + AckStopIDs []string `json:"ackStopIds,omitempty"` } -// PollAssignmentsOutput carries the instance's currently pending assignments. +// PollAssignmentsOutput carries the instance's currently pending assignments and +// stop directives. type PollAssignmentsOutput struct { - Assignments []bus.Assign `json:"assignments,omitempty"` + Assignments []bus.Assign `json:"assignments,omitempty"` + Stops []bus.StopDirective `json:"stops,omitempty"` } // PollAssignments acks any completed assignments (deletes their inbox keys) then @@ -40,6 +46,12 @@ func (s *Service) PollAssignments(input *PollAssignmentsInput, accountID string) } _ = kv.Delete(AssignmentKey(cluster, instanceID, taskShortID(taskID))) } + for _, taskID := range input.AckStopIDs { + if taskID == "" { + continue + } + _ = kv.Delete(StopKey(cluster, instanceID, taskShortID(taskID))) + } keys, err := keysWithPrefix(kv, AssignmentsPrefix(cluster, instanceID)) if err != nil { @@ -56,6 +68,21 @@ func (s *Service) PollAssignments(input *PollAssignmentsInput, accountID string) out.Assignments = append(out.Assignments, as) } } + + stopKeys, err := keysWithPrefix(kv, StopsPrefix(cluster, instanceID)) + if err != nil { + return nil, err + } + for _, key := range stopKeys { + var sd bus.StopDirective + found, err := getJSON(kv, key, &sd) + if err != nil { + return nil, err + } + if found { + out.Stops = append(out.Stops, sd) + } + } return out, nil } @@ -67,3 +94,12 @@ func (s *Service) reclaimAssignInbox(kv nats.KeyValue, cluster, instanceID, task } _ = kv.Delete(AssignmentKey(cluster, instanceID, taskID)) } + +// reclaimStopInbox removes a task's stop-inbox entry on the STOPPED transition so +// a reaped task is never re-delivered as a stop. Best-effort. +func (s *Service) reclaimStopInbox(kv nats.KeyValue, cluster, instanceID, taskID string) { + if instanceID == "" || taskID == "" { + return + } + _ = kv.Delete(StopKey(cluster, instanceID, taskID)) +} diff --git a/spinifex/handlers/ecs/poll_test.go b/spinifex/handlers/ecs/poll_test.go new file mode 100644 index 000000000..dd12be615 --- /dev/null +++ b/spinifex/handlers/ecs/poll_test.go @@ -0,0 +1,46 @@ +package handlers_ecs + +import ( + "testing" + + "github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestPollAssignments_DrainsStopsAndAcks covers the agent poll path: a poll +// returns both the pending assignments and the stop directives for the instance, +// and acking their IDs on a follow-up poll deletes the inbox entries so nothing is +// re-delivered. Empty ack IDs are skipped. +func TestPollAssignments_DrainsStopsAndAcks(t *testing.T) { + svc, _, kv := serviceTestRig(t) + + require.NoError(t, putJSON(kv, AssignmentKey("web", "i-1", "t-1"), + &bus.Assign{TaskID: "t-1", ClusterName: "web", InstanceID: "i-1"})) + require.NoError(t, putJSON(kv, StopKey("web", "i-1", "t-2"), + &bus.StopDirective{TaskID: "t-2", Reason: "drain"})) + + out, err := svc.PollAssignments(&PollAssignmentsInput{ + Cluster: "web", ContainerInstance: "i-1", + }, testAccountID) + require.NoError(t, err) + require.Len(t, out.Assignments, 1) + assert.Equal(t, "t-1", out.Assignments[0].TaskID) + require.Len(t, out.Stops, 1) + assert.Equal(t, "t-2", out.Stops[0].TaskID) + assert.Equal(t, "drain", out.Stops[0].Reason) + + // Ack both (with a blank ID that must be skipped) → entries removed, next poll + // drains empty. + out, err = svc.PollAssignments(&PollAssignmentsInput{ + Cluster: "web", ContainerInstance: "i-1", + AckTaskIDs: []string{"t-1", ""}, AckStopIDs: []string{"t-2", ""}, + }, testAccountID) + require.NoError(t, err) + assert.Empty(t, out.Assignments) + assert.Empty(t, out.Stops) + + found, err := getJSON(kv, StopKey("web", "i-1", "t-2"), &bus.StopDirective{}) + require.NoError(t, err) + assert.False(t, found) +} diff --git a/spinifex/handlers/ecs/services.go b/spinifex/handlers/ecs/services.go index f8302a6d7..34656dac8 100644 --- a/spinifex/handlers/ecs/services.go +++ b/spinifex/handlers/ecs/services.go @@ -265,7 +265,7 @@ func (s *Service) DeleteService(input *ecs.DeleteServiceInput, accountID string) return nil, err } for i := range tasks { - s.forceStopTask(kv, accountID, &tasks[i], "Service deleted") + s.requestStopTask(kv, accountID, &tasks[i], "Service deleted") } rec.Status = ServiceStatusInactive rec.RunningCount = 0 @@ -459,7 +459,9 @@ func (s *Service) stopSurplusTasks(kv nats.KeyValue, accountID string, tasks []T } primarySurplus := max(len(primaryPending)+len(primaryRunning)-desired, 0) runBudget := max(runningTotal-minCount, 0) - stop := func(i int, reason string) { s.forceStopTask(kv, accountID, &tasks[i], reason) } + // Cooperative stop: the agent reaps the container, then recordTaskState performs + // the single capacity release. A never-placed task force-stops internally. + stop := func(i int, reason string) { s.requestStopTask(kv, accountID, &tasks[i], reason) } for _, i := range oldPending { stop(i, deploymentSupersededReason) @@ -490,7 +492,10 @@ func (s *Service) stopSurplusTasks(kv nats.KeyValue, accountID string, tasks []T // --- Helpers --- -// listServiceTasks returns a cluster's non-STOPPED tasks owned by the service. +// listServiceTasks returns a cluster's live tasks owned by the service: neither +// STOPPED nor already requested to stop (DesiredStatus=STOPPED). A cooperatively +// stopped task lingers RUNNING until the agent reaps it, but it is on its way out, +// so it must not count toward the service's running total for scaling decisions. func (s *Service) listServiceTasks(kv nats.KeyValue, cluster, name string) ([]TaskRecord, error) { group := serviceTaskGroup(name) all, err := s.listTaskRecords(kv, cluster) @@ -499,7 +504,7 @@ func (s *Service) listServiceTasks(kv nats.KeyValue, cluster, name string) ([]Ta } out := make([]TaskRecord, 0, len(all)) for _, t := range all { - if t.Group == group && t.LastStatus != TaskStatusStopped { + if t.Group == group && t.LastStatus != TaskStatusStopped && t.DesiredStatus != TaskStatusStopped { out = append(out, t) } } diff --git a/spinifex/handlers/ecs/services_test.go b/spinifex/handlers/ecs/services_test.go index aea9323c2..3a84e1de2 100644 --- a/spinifex/handlers/ecs/services_test.go +++ b/spinifex/handlers/ecs/services_test.go @@ -258,20 +258,40 @@ func TestService_StopTask_StopsAndDeregisters(t *testing.T) { task := &TaskRecord{ TaskID: "t-9", Cluster: "web", Group: serviceTaskGroup("web"), - ContainerInstanceID: "i-1", LastStatus: TaskStatusRunning, + ContainerInstanceID: "i-1", LastStatus: TaskStatusRunning, DesiredStatus: TaskStatusRunning, NetworkMode: NetworkModeAwsvpc, ENIPrivateIP: "10.0.1.9", ReservedCPU: 64, ReservedMemoryMiB: 64, } require.NoError(t, putJSON(kv, TaskKey("web", "t-9"), task)) + // Cooperative stop: the task stays RUNNING with desiredStatus=STOPPED and a + // stop directive lands in the instance inbox; nothing is deregistered until the + // agent reaps the container and reports STOPPED. out, err := svc.StopTask(&ecs.StopTaskInput{ Cluster: aws.String("web"), Task: aws.String("t-9"), Reason: aws.String("bye"), }, testAccountID) require.NoError(t, err) - assert.Equal(t, TaskStatusStopped, aws.StringValue(out.Task.LastStatus)) - assert.Equal(t, []string{key(tgARN, "10.0.1.9", 80)}, reg.deregistered) + assert.Equal(t, TaskStatusRunning, aws.StringValue(out.Task.LastStatus)) + assert.Equal(t, TaskStatusStopped, aws.StringValue(out.Task.DesiredStatus)) + assert.Empty(t, reg.deregistered) - // Idempotent: a second stop is a no-op (no extra deregister). + var sd bus.StopDirective + found, err := getJSON(kv, StopKey("web", "i-1", "t-9"), &sd) + require.NoError(t, err) + require.True(t, found) + assert.Equal(t, "bye", sd.Reason) + + // Agent reports STOPPED → deregister + stop-inbox reclaimed. + require.NoError(t, svc.recordTaskState(&bus.TaskState{ + AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", + TaskID: "t-9", LastStatus: bus.TaskStatusStopped, + })) + assert.Equal(t, []string{key(tgARN, "10.0.1.9", 80)}, reg.deregistered) + found, err = getJSON(kv, StopKey("web", "i-1", "t-9"), &sd) + require.NoError(t, err) + assert.False(t, found) + + // Idempotent: a second stop is a no-op (already STOPPED). _, err = svc.StopTask(&ecs.StopTaskInput{Cluster: aws.String("web"), Task: aws.String("t-9")}, testAccountID) require.NoError(t, err) assert.Len(t, reg.deregistered, 1) @@ -283,6 +303,25 @@ func TestService_StopTask_NotFound(t *testing.T) { require.Error(t, err) } +// StopTask on a never-placed task (no container instance) force-stops it directly: +// no agent can reap it, so the control plane transitions it to STOPPED immediately. +func TestService_StopTask_UnplacedForceStops(t *testing.T) { + svc, _, kv := serviceTestRig(t) + task := &TaskRecord{ + TaskID: "t-np", Cluster: "web", + LastStatus: TaskStatusPending, DesiredStatus: TaskStatusRunning, + } + require.NoError(t, putJSON(kv, TaskKey("web", "t-np"), task)) + + out, err := svc.StopTask(&ecs.StopTaskInput{ + Cluster: aws.String("web"), Task: aws.String("t-np"), Reason: aws.String("cancel"), + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, TaskStatusStopped, aws.StringValue(out.Task.LastStatus)) + assert.Equal(t, TaskStatusStopped, aws.StringValue(out.Task.DesiredStatus)) + assert.Equal(t, "cancel", aws.StringValue(out.Task.StoppedReason)) +} + // A service with a target group registers each task's ENI IP on RUNNING and // deregisters it on STOPPED (ecs-v1.md Q8, single-writer). func TestService_TargetGroup_RegisterOnRunningDeregisterOnStopped(t *testing.T) { diff --git a/spinifex/handlers/ecs/store.go b/spinifex/handlers/ecs/store.go index 6b2a6916a..93bb5f714 100644 --- a/spinifex/handlers/ecs/store.go +++ b/spinifex/handlers/ecs/store.go @@ -81,6 +81,19 @@ func AssignmentKey(cluster, instanceID, taskID string) string { return AssignmentsPrefix(cluster, instanceID) + taskID } +// StopsPrefix returns the KV key prefix under which a container instance's pending +// task-stop directives (its "stop inbox") live. A sibling of assignments/ so it is +// never picked up when listing tasks or assignments. The agent drains it by +// polling the gateway; the STOPPED transition removes an entry. +func StopsPrefix(cluster, instanceID string) string { + return fmt.Sprintf("clusters/%s/stops/%s/", cluster, instanceID) +} + +// StopKey returns the KV key for one task's stop directive in an instance inbox. +func StopKey(cluster, instanceID, taskID string) string { + return StopsPrefix(cluster, instanceID) + taskID +} + // ServicesPrefix returns the KV key prefix under which all of a cluster's service // records live. Used by ListServices to enumerate. func ServicesPrefix(cluster string) string { diff --git a/spinifex/handlers/ecs/task_stop.go b/spinifex/handlers/ecs/task_stop.go index 2b94f886e..d3a0dc962 100644 --- a/spinifex/handlers/ecs/task_stop.go +++ b/spinifex/handlers/ecs/task_stop.go @@ -11,15 +11,19 @@ import ( "github.com/aws/aws-sdk-go/service/ecs" "github.com/google/uuid" "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus" "github.com/nats-io/nats.go" ) // defaultStopReason is recorded when StopTask is called without a reason. const defaultStopReason = "Task stopped by user" -// StopTask transitions a task to STOPPED via the control-plane forced-stop path -// (capacity + ENI release, TG deregister). The live container is torn down by the -// agent reconciler diffing KV against containerd (ecs-v1.md Q13). +// StopTask requests a cooperative stop: it marks the task DesiredStatus=STOPPED +// and posts a directive to the instance's stop inbox. The agent reaps the +// container and reports STOPPED, which drives the capacity + ENI + TG release in +// recordTaskState. LastStatus stays RUNNING until the container is actually reaped +// so a freed port is never rebound while the old container still holds it +// (ecs-v1.md:165). func (s *Service) StopTask(input *ecs.StopTaskInput, accountID string) (*ecs.StopTaskOutput, error) { cluster := clusterShortName(aws.StringValue(input.Cluster)) taskID := taskShortID(aws.StringValue(input.Task)) @@ -39,7 +43,7 @@ func (s *Service) StopTask(input *ecs.StopTaskInput, accountID string) (*ecs.Sto if reason == "" { reason = defaultStopReason } - s.forceStopTask(kv, accountID, &task, reason) + s.requestStopTask(kv, accountID, &task, reason) return &ecs.StopTaskOutput{Task: s.taskToAWS(accountID, &task)}, nil } @@ -138,10 +142,37 @@ func (s *Service) reserveOnInstance(kv nats.KeyValue, cluster, instanceID, taskI return nil, errors.New("reservation contended") } +// requestStopTask is the cooperative stop used while the agent is alive: it marks +// the task DesiredStatus=STOPPED and posts a stop directive to the instance's stop +// inbox, but leaves LastStatus, capacity, ENI, targets, and the assign inbox +// untouched. The agent reaps the container and reports STOPPED; recordTaskState +// then performs the single release. No-op once the task is already STOPPED. A task +// with no container instance (never placed) is forced-stopped directly since no +// agent can report it. +func (s *Service) requestStopTask(kv nats.KeyValue, accountID string, task *TaskRecord, reason string) { + if task.LastStatus == TaskStatusStopped { + return + } + if task.ContainerInstanceID == "" { + s.forceStopTask(kv, accountID, task, reason) + return + } + task.DesiredStatus = TaskStatusStopped + if perr := putJSON(kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil { + slog.Error("ECS requestStopTask: persist failed", "task", task.TaskID, "err", perr) + return + } + sd := bus.StopDirective{TaskID: task.TaskID, Reason: reason} + if perr := putJSON(kv, StopKey(task.Cluster, task.ContainerInstanceID, task.TaskID), &sd); perr != nil { + slog.Error("ECS requestStopTask: post stop directive failed", "task", task.TaskID, "err", perr) + } +} + // forceStopTask is the control-plane forced-stop: it transitions a task to // STOPPED, releases its capacity + ENI, drains its assign inbox, and deregisters -// its ELBv2 targets. Idempotent and reused by StopTask, service scale-in, and the -// heartbeat reaper. The mutated record is reflected back into task. +// its ELBv2 targets. Used where no live agent can reap the container (heartbeat +// reaper, cluster/instance teardown). The mutated record is reflected back into +// task. func (s *Service) forceStopTask(kv nats.KeyValue, accountID string, task *TaskRecord, reason string) { if task.LastStatus == TaskStatusStopped { return @@ -160,6 +191,7 @@ func (s *Service) forceStopTask(kv nats.KeyValue, accountID string, task *TaskRe } s.deregisterServiceTargets(kv, accountID, task) s.reclaimAssignInbox(kv, task.Cluster, task.ContainerInstanceID, task.TaskID) + s.reclaimStopInbox(kv, task.Cluster, task.ContainerInstanceID, task.TaskID) s.reclaimTaskENI(accountID, task) if rerr := s.releaseReservation(kv, task.Cluster, task.ContainerInstanceID, task.TaskID, task.ReservedCPU, task.ReservedMemoryMiB); rerr != nil { slog.Error("ECS forceStopTask: release reservation failed", "task", task.TaskID, "err", rerr)