From d32712b7272e009a23f838394700a42903293b9b Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 25 Jun 2026 07:52:07 +1000 Subject: [PATCH] feat: ECS scheduler, RunTask placement, and agent task execution (#416) Add the ECS v1 scheduler control loop and end-to-end RunTask flow: - handlers/ecs: per-account KV-backed Service (clusters, task definitions, container instances, tasks), bin-pack/spread/random placement with CAS capacity reservation, and a leader-elected Scheduler that owns the Layer-2 bus subscriptions (register/heartbeat/task-state) plus a heartbeat reaper. - gateway/ecs: wire CreateCluster, Describe/ListClusters, Register/Describe/ListTaskDefinitions, Register/Describe/ListContainerInstances, RunTask, Describe/ListTasks onto the daemon over NATS. - daemon: construct the Service, start the Scheduler goroutine, subscribe the ECS action subjects. - ecs-agent: subscribe the per-instance assign subject, pull and run each container via containerd, and report task state back on the bus. --- cmd/ecs-agent/agent.go | 26 +- cmd/ecs-agent/agent_test.go | 2 +- cmd/ecs-agent/runtime/containerd.go | 2 +- cmd/ecs-agent/runtime/containerd_run.go | 105 +++++++ cmd/ecs-agent/runtime/runtime.go | 31 ++ cmd/ecs-agent/runtime/runtime_fake.go | 40 ++- cmd/ecs-agent/taskrun.go | 132 ++++++++ cmd/ecs-agent/taskrun_test.go | 162 ++++++++++ spinifex/awserrors/awserrors.go | 12 +- spinifex/daemon/daemon.go | 42 +++ spinifex/daemon/daemon_handlers_ecs.go | 59 ++++ spinifex/gateway/ecs/actions.go | 121 ++++++++ spinifex/gateway/ecs/handler.go | 36 +-- spinifex/gateway/ecs/handler_test.go | 19 +- spinifex/gateway/ecs_test.go | 7 +- spinifex/gateway/gateway.go | 12 +- spinifex/handlers/ecs/bus/messages.go | 68 +++++ spinifex/handlers/ecs/conv.go | 124 ++++++++ spinifex/handlers/ecs/conv_test.go | 83 +++++ spinifex/handlers/ecs/instances.go | 274 +++++++++++++++++ spinifex/handlers/ecs/placement.go | 86 ++++++ spinifex/handlers/ecs/placement_test.go | 79 +++++ spinifex/handlers/ecs/records.go | 149 +++++++++ spinifex/handlers/ecs/scheduler.go | 288 ++++++++++++++++++ spinifex/handlers/ecs/service.go | 384 ++++++++++++++++++++++++ spinifex/handlers/ecs/service_nats.go | 80 +++++ spinifex/handlers/ecs/service_test.go | 292 ++++++++++++++++++ spinifex/handlers/ecs/tasks.go | 244 +++++++++++++++ 28 files changed, 2915 insertions(+), 44 deletions(-) create mode 100644 cmd/ecs-agent/runtime/containerd_run.go create mode 100644 cmd/ecs-agent/taskrun.go create mode 100644 cmd/ecs-agent/taskrun_test.go create mode 100644 spinifex/daemon/daemon_handlers_ecs.go create mode 100644 spinifex/gateway/ecs/actions.go create mode 100644 spinifex/handlers/ecs/conv.go create mode 100644 spinifex/handlers/ecs/conv_test.go create mode 100644 spinifex/handlers/ecs/instances.go create mode 100644 spinifex/handlers/ecs/placement.go create mode 100644 spinifex/handlers/ecs/placement_test.go create mode 100644 spinifex/handlers/ecs/records.go create mode 100644 spinifex/handlers/ecs/scheduler.go create mode 100644 spinifex/handlers/ecs/service.go create mode 100644 spinifex/handlers/ecs/service_nats.go create mode 100644 spinifex/handlers/ecs/service_test.go create mode 100644 spinifex/handlers/ecs/tasks.go diff --git a/cmd/ecs-agent/agent.go b/cmd/ecs-agent/agent.go index 980209d1f..1e8fe0402 100644 --- a/cmd/ecs-agent/agent.go +++ b/cmd/ecs-agent/agent.go @@ -25,12 +25,14 @@ var version = "dev" // Agent wires the ecs-agent's runtime seams: a NATS publisher for the Layer-2 // bus, a container runtime, and an ECR resolver. It registers the host as a -// container instance at boot and heartbeats while alive. Task assignment and -// container lifecycle land in Sprint 4e. +// container instance at boot, heartbeats while alive, and (Sprint 4e) runs +// assigned tasks through containerd, reporting their state back on the bus. type Agent struct { cfg config id identity + pub publisher puller ctrruntime.ImagePuller + runner ctrruntime.Runner resolver ctrruntime.Resolver reg *registrar @@ -41,12 +43,16 @@ type Agent struct { } // newAgent assembles an Agent from already-built seams. Tests use this directly -// with fakes; New builds the production seams and delegates here. -func newAgent(cfg config, id identity, pub publisher, puller ctrruntime.ImagePuller, resolver ctrruntime.Resolver) *Agent { +// with fakes; New builds the production seams and delegates here. runner may be +// nil when containerd is unavailable; the assign path then reports the task +// STOPPED rather than crashing. +func newAgent(cfg config, id identity, pub publisher, puller ctrruntime.ImagePuller, runner ctrruntime.Runner, resolver ctrruntime.Resolver) *Agent { return &Agent{ cfg: cfg, id: id, + pub: pub, puller: puller, + runner: runner, resolver: resolver, reg: newRegistrar(pub, id), hb: newHeartbeater(pub, id, cfg.Heartbeat, nil), @@ -81,10 +87,12 @@ func New(cfg config) (*Agent, error) { resolver := newLazyECRResolver(creds, cfg.Region, cfg.GatewayURL, cfg.GatewayCA) var puller ctrruntime.ImagePuller - if p, perr := ctrruntime.New(cfg.ContainerdSocket); perr != nil { + var runner ctrruntime.Runner + if rt, perr := ctrruntime.New(cfg.ContainerdSocket); perr != nil { slog.Warn("ecs-agent: containerd unavailable at boot, image pulls disabled", "err", perr) } else { - puller = p + puller = rt + runner = rt } nc, err := nats.Connect(cfg.NATSURL, nats.Name("ecs-agent/"+id.InstanceID)) @@ -92,7 +100,7 @@ func New(cfg config) (*Agent, error) { return nil, fmt.Errorf("connect nats %s: %w", cfg.NATSURL, err) } - a := newAgent(cfg, id, nc, puller, resolver) + a := newAgent(cfg, id, nc, puller, runner, resolver) a.nc = nc if puller != nil { a.closers = append(a.closers, puller.Close) @@ -113,6 +121,10 @@ func (a *Agent) Run(ctx context.Context) error { go a.hb.Run(ctx) + if err := a.subscribeAssign(ctx); err != nil { + slog.Error("ecs-agent: assign subscription failed; task assignment disabled", "err", err) + } + <-ctx.Done() return a.Stop() } diff --git a/cmd/ecs-agent/agent_test.go b/cmd/ecs-agent/agent_test.go index a8be7ee17..c15d2489e 100644 --- a/cmd/ecs-agent/agent_test.go +++ b/cmd/ecs-agent/agent_test.go @@ -14,7 +14,7 @@ func TestAgent_RunRegistersThenStopsOnContext(t *testing.T) { cfg := config{Heartbeat: 5 * time.Millisecond} id := testIdentity() puller := &ctrruntime.FakePuller{} - a := newAgent(cfg, id, pub, puller, nil) + a := newAgent(cfg, id, pub, puller, puller, nil) a.closers = append(a.closers, puller.Close) ctx, cancel := context.WithCancel(context.Background()) diff --git a/cmd/ecs-agent/runtime/containerd.go b/cmd/ecs-agent/runtime/containerd.go index 9c08a7305..5c8abeb87 100644 --- a/cmd/ecs-agent/runtime/containerd.go +++ b/cmd/ecs-agent/runtime/containerd.go @@ -31,7 +31,7 @@ var _ ImagePuller = (*containerdPuller)(nil) // returns an error if it is unreachable — otherwise a dead socket would only // surface on the first pull, and the agent's boot-time "containerd unavailable" // log would never fire. -func New(socket string) (ImagePuller, error) { +func New(socket string) (Runtime, error) { client, err := containerd.New(socket, containerd.WithDefaultNamespace(ecsNamespace)) if err != nil { return nil, fmt.Errorf("connect containerd %s: %w", socket, err) diff --git a/cmd/ecs-agent/runtime/containerd_run.go b/cmd/ecs-agent/runtime/containerd_run.go new file mode 100644 index 000000000..42f297e8c --- /dev/null +++ b/cmd/ecs-agent/runtime/containerd_run.go @@ -0,0 +1,105 @@ +package runtime + +import ( + "context" + "fmt" + "sort" + + containerd "github.com/containerd/containerd/v2/client" + "github.com/containerd/containerd/v2/pkg/cio" + "github.com/containerd/containerd/v2/pkg/namespaces" + "github.com/containerd/containerd/v2/pkg/oci" + specs "github.com/opencontainers/runtime-spec/specs-go" +) + +var _ Runner = (*containerdPuller)(nil) + +// Run creates a container from an already-pulled image and starts its task. The +// container joins the host network namespace (bridge/CNI task networking is a +// later sprint) and carries spec.Labels for the reboot reconciler. id must be +// unique within the "ecs" namespace. +func (p *containerdPuller) Run(ctx context.Context, id string, spec RunSpec) (string, error) { + ctx = namespaces.WithNamespace(ctx, ecsNamespace) + + image, err := p.client.GetImage(ctx, spec.Image) + if err != nil { + return "", fmt.Errorf("get image %s: %w", spec.Image, err) + } + + specOpts := []oci.SpecOpts{ + oci.WithImageConfig(image), + oci.WithHostNamespace(specs.NetworkNamespace), + oci.WithHostHostsFile, + oci.WithHostResolvconf, + } + if len(spec.Command) > 0 { + specOpts = append(specOpts, oci.WithProcessArgs(spec.Command...)) + } + if len(spec.Env) > 0 { + specOpts = append(specOpts, oci.WithEnv(envSlice(spec.Env))) + } + + container, err := p.client.NewContainer(ctx, id, + containerd.WithNewSnapshot(id+"-snapshot", image), + containerd.WithNewSpec(specOpts...), + containerd.WithContainerLabels(spec.Labels), + ) + if err != nil { + return "", fmt.Errorf("create container %s: %w", id, err) + } + + task, err := container.NewTask(ctx, cio.NullIO) + if err != nil { + _ = container.Delete(ctx, containerd.WithSnapshotCleanup) + return "", fmt.Errorf("create task %s: %w", id, err) + } + if err := task.Start(ctx); err != nil { + _, _ = task.Delete(ctx) + _ = container.Delete(ctx, containerd.WithSnapshotCleanup) + return "", fmt.Errorf("start task %s: %w", id, err) + } + return id, nil +} + +// Wait blocks until the container's task exits and returns its exit code. +func (p *containerdPuller) Wait(ctx context.Context, containerID string) (RunStatus, error) { + ctx = namespaces.WithNamespace(ctx, ecsNamespace) + container, err := p.client.LoadContainer(ctx, containerID) + if err != nil { + return RunStatus{}, fmt.Errorf("load container %s: %w", containerID, err) + } + task, err := container.Task(ctx, nil) + if err != nil { + return RunStatus{}, fmt.Errorf("attach task %s: %w", containerID, err) + } + statusC, err := task.Wait(ctx) + if err != nil { + return RunStatus{}, fmt.Errorf("wait task %s: %w", containerID, err) + } + status := <-statusC + code, _, err := status.Result() + return RunStatus{ExitCode: int(code)}, err +} + +// Remove kills and deletes the container's task, then the container + snapshot. +func (p *containerdPuller) Remove(ctx context.Context, containerID string) error { + ctx = namespaces.WithNamespace(ctx, ecsNamespace) + container, err := p.client.LoadContainer(ctx, containerID) + if err != nil { + return nil //nolint:nilerr // load failure means the container is already gone + } + if task, terr := container.Task(ctx, nil); terr == nil { + _, _ = task.Delete(ctx, containerd.WithProcessKill) + } + return container.Delete(ctx, containerd.WithSnapshotCleanup) +} + +// envSlice renders an env map as sorted "K=V" entries for deterministic specs. +func envSlice(env map[string]string) []string { + out := make([]string, 0, len(env)) + for k, v := range env { + out = append(out, k+"="+v) + } + sort.Strings(out) + return out +} diff --git a/cmd/ecs-agent/runtime/runtime.go b/cmd/ecs-agent/runtime/runtime.go index 56851c65e..87075f650 100644 --- a/cmd/ecs-agent/runtime/runtime.go +++ b/cmd/ecs-agent/runtime/runtime.go @@ -34,3 +34,34 @@ type ImagePuller interface { Pull(ctx context.Context, spec PullSpec, r Resolver) (Image, error) Close() error } + +// RunSpec describes a single container the agent must create and start. v1 runs +// the container in the host network namespace (bridge/CNI task networking lands +// in a later sprint); Labels carry the mulga.ecs.* task identity for the reboot +// reconciler. +type RunSpec struct { + Image string + Command []string + Env map[string]string + Labels map[string]string +} + +// RunStatus is a finished container's outcome. +type RunStatus struct { + ExitCode int +} + +// Runner creates and starts containers from already-pulled images. id is a +// caller-unique container ID; Wait blocks until the container exits; Remove +// tears down the container + its task. +type Runner interface { + Run(ctx context.Context, id string, spec RunSpec) (containerID string, err error) + Wait(ctx context.Context, containerID string) (RunStatus, error) + Remove(ctx context.Context, containerID string) error +} + +// Runtime is the full container runtime the ecs-agent drives: pull + run. +type Runtime interface { + ImagePuller + Runner +} diff --git a/cmd/ecs-agent/runtime/runtime_fake.go b/cmd/ecs-agent/runtime/runtime_fake.go index cef3fa185..cf09f4ab6 100644 --- a/cmd/ecs-agent/runtime/runtime_fake.go +++ b/cmd/ecs-agent/runtime/runtime_fake.go @@ -16,9 +16,20 @@ type FakePuller struct { Err error Closed bool OnPull func(PullSpec) (Image, error) + + // Run bookkeeping. + Runs []RunSpec + RunErr error + WaitCode int + WaitErr error + Removed []string + OnRun func(string, RunSpec) (string, error) } -var _ ImagePuller = (*FakePuller)(nil) +var ( + _ ImagePuller = (*FakePuller)(nil) + _ Runner = (*FakePuller)(nil) +) // Pull records the spec, exercises the resolver, and returns the programmed // result. If OnPull is set it takes precedence over Result/Err. @@ -57,3 +68,30 @@ func (f *FakePuller) Close() error { f.Closed = true return nil } + +// Run records the spec and returns id (or the programmed result/error). +func (f *FakePuller) Run(_ context.Context, id string, spec RunSpec) (string, error) { + f.mu.Lock() + f.Runs = append(f.Runs, spec) + f.mu.Unlock() + if f.OnRun != nil { + return f.OnRun(id, spec) + } + if f.RunErr != nil { + return "", f.RunErr + } + return id, nil +} + +// Wait returns the programmed exit code/error. +func (f *FakePuller) Wait(_ context.Context, _ string) (RunStatus, error) { + return RunStatus{ExitCode: f.WaitCode}, f.WaitErr +} + +// Remove records the container ID. +func (f *FakePuller) Remove(_ context.Context, containerID string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.Removed = append(f.Removed, containerID) + return nil +} diff --git a/cmd/ecs-agent/taskrun.go b/cmd/ecs-agent/taskrun.go new file mode 100644 index 000000000..d3f3c06f4 --- /dev/null +++ b/cmd/ecs-agent/taskrun.go @@ -0,0 +1,132 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "time" + + "github.com/nats-io/nats.go" + + ctrruntime "github.com/mulgadc/spinifex/cmd/ecs-agent/runtime" + "github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus" +) + +// subscribeAssign wires the per-instance assign subject to the task runner. It is +// a no-op when the agent has no live NATS connection (unit tests drive runTask +// directly). +func (a *Agent) subscribeAssign(ctx context.Context) error { + if a.nc == nil { + return nil + } + subj := bus.AssignSubject(a.id.AccountID, a.id.ClusterName, a.id.InstanceID) + sub, err := a.nc.Subscribe(subj, func(msg *nats.Msg) { + var as bus.Assign + if err := json.Unmarshal(msg.Data, &as); err != nil { + slog.Error("ecs-agent: bad assign payload", "err", err) + return + } + slog.Info("ecs-agent: task assigned", "task", as.TaskID, "containers", len(as.Containers)) + go a.runTask(ctx, &as) + }) + if err != nil { + return fmt.Errorf("subscribe %s: %w", subj, err) + } + a.closers = append(a.closers, sub.Unsubscribe) + slog.Info("ecs-agent: assign subscription active", "subject", subj) + return nil +} + +// 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. +func (a *Agent) runTask(ctx context.Context, as *bus.Assign) { + if a.puller == nil || a.runner == nil { + a.reportTaskState(as, bus.TaskStatusStopped, "containerd unavailable on agent", nil) + return + } + + statuses := make([]bus.ContainerStatus, 0, len(as.Containers)) + for _, c := range as.Containers { + if _, err := a.puller.Pull(ctx, ctrruntime.PullSpec{Ref: c.Image}, a.resolver); err != nil { + slog.Error("ecs-agent: pull failed", "task", as.TaskID, "image", c.Image, "err", err) + a.reportTaskState(as, bus.TaskStatusStopped, "image pull failed: "+err.Error(), statuses) + return + } + + cid := containerID(as.TaskID, c.Name) + spec := ctrruntime.RunSpec{ + Image: c.Image, + Command: c.Command, + Env: c.Environment, + Labels: taskLabels(as, c.Name), + } + id, err := a.runner.Run(ctx, cid, spec) + if err != nil { + slog.Error("ecs-agent: run failed", "task", as.TaskID, "container", c.Name, "err", err) + a.reportTaskState(as, bus.TaskStatusStopped, "container start failed: "+err.Error(), statuses) + return + } + statuses = append(statuses, bus.ContainerStatus{Name: c.Name, Status: bus.TaskStatusRunning, ContainerID: id}) + go a.waitContainer(ctx, as, c.Name, id) + } + + a.reportTaskState(as, bus.TaskStatusRunning, "", statuses) + slog.Info("ecs-agent: task running", "task", as.TaskID, "containers", len(statuses)) +} + +// waitContainer blocks until a container exits, then reports the task STOPPED +// with the exit code. v1 stops the whole task when its first container exits. +func (a *Agent) waitContainer(ctx context.Context, as *bus.Assign, name, containerID string) { + status, err := a.runner.Wait(ctx, containerID) + if err != nil { + slog.Warn("ecs-agent: wait container failed", "task", as.TaskID, "container", name, "err", err) + return + } + exit := status.ExitCode + statuses := []bus.ContainerStatus{{Name: name, Status: bus.TaskStatusStopped, ContainerID: containerID, ExitCode: &exit}} + a.reportTaskState(as, bus.TaskStatusStopped, fmt.Sprintf("container %s exited (%d)", name, exit), statuses) + slog.Info("ecs-agent: task stopped", "task", as.TaskID, "container", name, "exitCode", exit) +} + +// reportTaskState publishes a TaskState message on the task's state subject. +func (a *Agent) reportTaskState(as *bus.Assign, status, reason string, containers []bus.ContainerStatus) { + if a.pub == nil { + return + } + msg := bus.TaskState{ + AccountID: a.id.AccountID, + ClusterName: a.id.ClusterName, + InstanceID: a.id.InstanceID, + TaskID: as.TaskID, + LastStatus: status, + Containers: containers, + Reason: reason, + ReportedAt: time.Now().UTC(), + } + data, err := json.Marshal(&msg) + if err != nil { + slog.Error("ecs-agent: marshal task-state failed", "task", as.TaskID, "err", err) + return + } + subj := bus.TaskStateSubject(a.id.AccountID, a.id.ClusterName, as.TaskID) + if err := a.pub.Publish(subj, data); err != nil { + slog.Error("ecs-agent: publish task-state failed", "task", as.TaskID, "subject", subj, "err", err) + } +} + +// containerID composes a containerd-valid container ID from a task + container. +func containerID(taskID, name string) string { + return fmt.Sprintf("%s-%s", taskID, name) +} + +// taskLabels are the mulga.ecs.* labels stamped on every container so the reboot +// reconciler can re-associate running containers with their task on restart. +func taskLabels(as *bus.Assign, name string) map[string]string { + return map[string]string{ + "mulga.ecs.taskID": as.TaskID, + "mulga.ecs.containerName": name, + "mulga.ecs.clusterName": as.ClusterName, + } +} diff --git a/cmd/ecs-agent/taskrun_test.go b/cmd/ecs-agent/taskrun_test.go new file mode 100644 index 000000000..72d1c5a34 --- /dev/null +++ b/cmd/ecs-agent/taskrun_test.go @@ -0,0 +1,162 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "sync" + "testing" + "time" + + ctrruntime "github.com/mulgadc/spinifex/cmd/ecs-agent/runtime" + "github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus" +) + +// historyPub records every published message in order (capturePub keeps only the +// last per subject, which loses the RUNNING→STOPPED transition on one subject). +type historyPub struct { + mu sync.Mutex + msgs [][]byte +} + +func (h *historyPub) Publish(_ string, data []byte) error { + h.mu.Lock() + defer h.mu.Unlock() + cp := make([]byte, len(data)) + copy(cp, data) + h.msgs = append(h.msgs, cp) + return nil +} + +func (h *historyPub) states() []bus.TaskState { + h.mu.Lock() + defer h.mu.Unlock() + out := make([]bus.TaskState, 0, len(h.msgs)) + for _, raw := range h.msgs { + var ts bus.TaskState + if err := json.Unmarshal(raw, &ts); err == nil { + out = append(out, ts) + } + } + return out +} + +func testAssign() *bus.Assign { + return &bus.Assign{ + AccountID: "123456789012", + ClusterName: "default", + InstanceID: "i-abc123", + TaskID: "t-001", + Containers: []bus.AssignContainer{ + {Name: "web", Image: "registry/web:1", Command: []string{"/bin/true"}, + Environment: map[string]string{"FOO": "bar"}}, + }, + } +} + +func newRunAgent(pub publisher, rt *ctrruntime.FakePuller) *Agent { + return newAgent(config{}, testIdentity(), pub, rt, rt, nil) +} + +// runTask with no runtime reports the task STOPPED instead of crashing. +func TestRunTask_NoRuntimeReportsStopped(t *testing.T) { + pub := &historyPub{} + a := newAgent(config{}, testIdentity(), pub, nil, nil, nil) + a.runTask(context.Background(), testAssign()) + + st := pub.states() + if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped { + t.Fatalf("want one STOPPED state, got %+v", st) + } + if st[0].Reason == "" { + t.Error("STOPPED state missing reason") + } +} + +// Happy path: pull + run reports RUNNING with the container ID; Wait blocking +// (forced error) means no STOPPED overwrite, so we can assert the RUNNING frame. +func TestRunTask_PullRunReportsRunning(t *testing.T) { + pub := &historyPub{} + rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")} + a := newRunAgent(pub, rt) + as := testAssign() + a.runTask(context.Background(), as) + + if len(rt.Pulls) != 1 || rt.Pulls[0].Ref != "registry/web:1" { + t.Fatalf("expected one pull of registry/web:1, got %+v", rt.Pulls) + } + if len(rt.Runs) != 1 || rt.Runs[0].Image != "registry/web:1" { + t.Fatalf("expected one run, got %+v", rt.Runs) + } + if rt.Runs[0].Labels["mulga.ecs.taskID"] != "t-001" { + t.Errorf("missing task label: %+v", rt.Runs[0].Labels) + } + + st := pub.states() + if len(st) == 0 || st[len(st)-1].LastStatus != bus.TaskStatusRunning { + t.Fatalf("want RUNNING final state, got %+v", st) + } + if len(st[len(st)-1].Containers) != 1 || st[len(st)-1].Containers[0].ContainerID != "t-001-web" { + t.Errorf("RUNNING state container wrong: %+v", st[len(st)-1].Containers) + } +} + +// A pull failure reports STOPPED and never starts the container. +func TestRunTask_PullFailureReportsStopped(t *testing.T) { + pub := &historyPub{} + rt := &ctrruntime.FakePuller{Err: errors.New("pull denied")} + a := newRunAgent(pub, rt) + a.runTask(context.Background(), testAssign()) + + if len(rt.Runs) != 0 { + t.Errorf("container should not run after pull failure, got %+v", rt.Runs) + } + st := pub.states() + if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped { + t.Fatalf("want STOPPED, got %+v", st) + } +} + +// A run failure reports STOPPED. +func TestRunTask_RunFailureReportsStopped(t *testing.T) { + pub := &historyPub{} + rt := &ctrruntime.FakePuller{RunErr: errors.New("start boom")} + a := newRunAgent(pub, rt) + a.runTask(context.Background(), testAssign()) + + st := pub.states() + if len(st) != 1 || st[0].LastStatus != bus.TaskStatusStopped { + t.Fatalf("want STOPPED, got %+v", st) + } +} + +// On container exit, waitContainer reports STOPPED with the exit code. +func TestRunTask_ExitReportsStoppedWithCode(t *testing.T) { + pub := &historyPub{} + rt := &ctrruntime.FakePuller{WaitCode: 7} + a := newRunAgent(pub, rt) + a.runTask(context.Background(), testAssign()) + + deadline := time.After(time.Second) + for { + st := pub.states() + if len(st) > 0 && st[len(st)-1].LastStatus == bus.TaskStatusStopped { + last := st[len(st)-1] + if len(last.Containers) != 1 || last.Containers[0].ExitCode == nil || *last.Containers[0].ExitCode != 7 { + t.Fatalf("want exit code 7, got %+v", last.Containers) + } + return + } + select { + case <-deadline: + t.Fatalf("no STOPPED state after exit; got %+v", pub.states()) + case <-time.After(2 * time.Millisecond): + } + } +} + +func TestContainerID(t *testing.T) { + if got := containerID("t-1", "web"); got != "t-1-web" { + t.Errorf("containerID = %q, want t-1-web", got) + } +} diff --git a/spinifex/awserrors/awserrors.go b/spinifex/awserrors/awserrors.go index b3144b040..db87d8d6e 100644 --- a/spinifex/awserrors/awserrors.go +++ b/spinifex/awserrors/awserrors.go @@ -371,9 +371,13 @@ var ( ErrorResourceCountLimitExceeded = "ResourceCountLimitExceeded" ErrorResourceLimitExceeded = "ResourceLimitExceeded" // ErrorResourceNotFound is the ACM not-found code. EKS must use ErrorEKSResourceNotFound — do not cross-wire. - ErrorResourceNotFound = "ResourceNotFound" - ErrorEKSResourceInUse = "ResourceInUseException" - ErrorEKSResourceNotFound = "ResourceNotFoundException" + ErrorResourceNotFound = "ResourceNotFound" + ErrorEKSResourceInUse = "ResourceInUseException" + ErrorEKSResourceNotFound = "ResourceNotFoundException" + // ECS JSON-1.1 exception codes. ECS clients key on the "Exception"-suffixed + // __type; do not cross-wire with the EC2/ACM not-found codes above. + ErrorECSClusterNotFound = "ClusterNotFoundException" + ErrorECSInvalidParameter = "InvalidParameterException" ErrorRetryableError = "RetryableError" ErrorRouteAlreadyExists = "RouteAlreadyExists" ErrorRouteLimitExceeded = "RouteLimitExceeded" @@ -894,6 +898,8 @@ var ErrorLookup = map[string]ErrorMessage{ ErrorResourceNotFound: {HTTPCode: 404, Message: "The specified resource was not found."}, ErrorEKSResourceInUse: {HTTPCode: 409, Message: "A cluster with this name already exists in this account and Region. Delete the existing cluster, or choose a different name."}, ErrorEKSResourceNotFound: {HTTPCode: 404, Message: "The specified cluster could not be found. You can view your available clusters with ListClusters. Amazon EKS clusters are Region specific."}, + ErrorECSClusterNotFound: {HTTPCode: 400, Message: "The specified cluster was not found. You can view your available clusters with ListClusters. Amazon ECS clusters are Region specific."}, + ErrorECSInvalidParameter: {HTTPCode: 400, Message: "The specified parameter is not valid. Review the available parameters for the API request."}, ErrorRetryableError: {HTTPCode: 400, Message: "A request submitted by an AWS service on your behalf could not be completed. The requesting service might automatically retry the request."}, ErrorRouteAlreadyExists: {HTTPCode: 409, Message: "A route for the specified CIDR block already exists in this route table."}, ErrorRouteLimitExceeded: {HTTPCode: 400, Message: "You've reached the limit on the number of routes that you can add to a route table."}, diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 98c535b92..7a2c72427 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -49,6 +49,7 @@ import ( handlers_ec2_volume "github.com/mulgadc/spinifex/spinifex/handlers/ec2/volume" handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" handlers_ecr "github.com/mulgadc/spinifex/spinifex/handlers/ecr" + handlers_ecs "github.com/mulgadc/spinifex/spinifex/handlers/ecs" handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" handlers_elbv2 "github.com/mulgadc/spinifex/spinifex/handlers/elbv2" "github.com/mulgadc/spinifex/spinifex/instancetypes" @@ -149,6 +150,8 @@ type Daemon struct { eipService *handlers_ec2_eip.EIPServiceImpl elbv2Service *handlers_elbv2.ELBv2ServiceImpl eksService *handlers_eks.EKSServiceImpl + ecsService *handlers_ecs.Service + ecsScheduler *handlers_ecs.Scheduler acmService *handlers_acm.ACMServiceImpl ecrMetaService *handlers_ecr.MetaServiceImpl routeTableService *handlers_ec2_routetable.RouteTableServiceImpl @@ -947,6 +950,24 @@ func (d *Daemon) subscribeAll() error { ) } + // ECS gateway → daemon subscriptions (control plane; per-account KV). + if d.ecsService != nil { + subs = append(subs, + natsSub{"ecs.CreateCluster", d.handleECSCreateCluster, "spinifex-workers"}, + natsSub{"ecs.DescribeClusters", d.handleECSDescribeClusters, "spinifex-workers"}, + natsSub{"ecs.ListClusters", d.handleECSListClusters, "spinifex-workers"}, + natsSub{"ecs.RegisterTaskDefinition", d.handleECSRegisterTaskDefinition, "spinifex-workers"}, + natsSub{"ecs.DescribeTaskDefinition", d.handleECSDescribeTaskDefinition, "spinifex-workers"}, + natsSub{"ecs.ListTaskDefinitions", d.handleECSListTaskDefinitions, "spinifex-workers"}, + natsSub{"ecs.RegisterContainerInstance", d.handleECSRegisterContainerInstance, "spinifex-workers"}, + natsSub{"ecs.DescribeContainerInstances", d.handleECSDescribeContainerInstances, "spinifex-workers"}, + natsSub{"ecs.ListContainerInstances", d.handleECSListContainerInstances, "spinifex-workers"}, + natsSub{"ecs.RunTask", d.handleECSRunTask, "spinifex-workers"}, + natsSub{"ecs.DescribeTasks", d.handleECSDescribeTasks, "spinifex-workers"}, + natsSub{"ecs.ListTasks", d.handleECSListTasks, "spinifex-workers"}, + ) + } + // ACM gateway → daemon subscriptions (minimal certificate store). if d.acmService != nil { subs = append(subs, @@ -1377,6 +1398,27 @@ func (d *Daemon) startCluster() error { return fmt.Errorf("failed to initialize EKS service: %w", err) } + // ECS control plane: per-account KV-backed handlers + a leader-elected + // scheduler goroutine that owns the Layer-2 bus subscriptions and heartbeat + // reaper. The scheduler is disabled (handlers still serve) when JetStream is + // unavailable. + d.ecsService = handlers_ecs.NewService(d.natsConn, d.config.Region, d.clusterConfig.AWS.InternalSuffix) + if js, jsErr := d.natsConn.JetStream(); jsErr != nil { + slog.Warn("ECS scheduler disabled: JetStream unavailable", "err", jsErr) + } else if _, lbErr := handlers_ecs.InitLeaderBucket(js); lbErr != nil { + slog.Warn("ECS scheduler disabled: leader bucket init failed", "err", lbErr) + } else { + d.ecsScheduler = handlers_ecs.NewScheduler(d.natsConn, d.ecsService, d.node) + d.shutdownWg.Go(func() { + defer func() { + if r := recover(); r != nil { + slog.Error("ECS scheduler goroutine panicked", "recover", r) + } + }() + d.ecsScheduler.Run(d.ctx) + }) + } + d.acmService, err = initServiceWithRetry("ACM service", func() (*handlers_acm.ACMServiceImpl, error) { return handlers_acm.NewACMServiceImplWithNATS(d.config, d.natsConn) }) diff --git a/spinifex/daemon/daemon_handlers_ecs.go b/spinifex/daemon/daemon_handlers_ecs.go new file mode 100644 index 000000000..b15d9383e --- /dev/null +++ b/spinifex/daemon/daemon_handlers_ecs.go @@ -0,0 +1,59 @@ +package daemon + +import "github.com/nats-io/nats.go" + +// --- Cluster --- + +func (d *Daemon) handleECSCreateCluster(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.CreateCluster) +} + +func (d *Daemon) handleECSDescribeClusters(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.DescribeClusters) +} + +func (d *Daemon) handleECSListClusters(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.ListClusters) +} + +// --- Task definition --- + +func (d *Daemon) handleECSRegisterTaskDefinition(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.RegisterTaskDefinition) +} + +func (d *Daemon) handleECSDescribeTaskDefinition(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.DescribeTaskDefinition) +} + +func (d *Daemon) handleECSListTaskDefinitions(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.ListTaskDefinitions) +} + +// --- Container instance --- + +func (d *Daemon) handleECSRegisterContainerInstance(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.RegisterContainerInstance) +} + +func (d *Daemon) handleECSDescribeContainerInstances(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.DescribeContainerInstances) +} + +func (d *Daemon) handleECSListContainerInstances(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.ListContainerInstances) +} + +// --- Task --- + +func (d *Daemon) handleECSRunTask(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.RunTask) +} + +func (d *Daemon) handleECSDescribeTasks(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.DescribeTasks) +} + +func (d *Daemon) handleECSListTasks(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.ListTasks) +} diff --git a/spinifex/gateway/ecs/actions.go b/spinifex/gateway/ecs/actions.go new file mode 100644 index 000000000..c7e635f88 --- /dev/null +++ b/spinifex/gateway/ecs/actions.go @@ -0,0 +1,121 @@ +package gateway_ecs + +import ( + "encoding/json" + + "github.com/aws/aws-sdk-go/service/ecs" + handlers_ecs "github.com/mulgadc/spinifex/spinifex/handlers/ecs" + "github.com/nats-io/nats.go" +) + +// unmarshalIfBody decodes body into out only when non-empty. +func unmarshalIfBody(body []byte, out any) error { + if len(body) == 0 { + return nil + } + return json.Unmarshal(body, out) +} + +// --- Cluster --- + +func CreateCluster(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.CreateClusterInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).CreateCluster(input, accountID) +} + +func DescribeClusters(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.DescribeClustersInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).DescribeClusters(input, accountID) +} + +func ListClusters(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.ListClustersInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).ListClusters(input, accountID) +} + +// --- Task definition --- + +func RegisterTaskDefinition(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.RegisterTaskDefinitionInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).RegisterTaskDefinition(input, accountID) +} + +func DescribeTaskDefinition(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.DescribeTaskDefinitionInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).DescribeTaskDefinition(input, accountID) +} + +func ListTaskDefinitions(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.ListTaskDefinitionsInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).ListTaskDefinitions(input, accountID) +} + +// --- Container instance --- + +func RegisterContainerInstance(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.RegisterContainerInstanceInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).RegisterContainerInstance(input, accountID) +} + +func DescribeContainerInstances(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.DescribeContainerInstancesInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).DescribeContainerInstances(input, accountID) +} + +func ListContainerInstances(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.ListContainerInstancesInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).ListContainerInstances(input, accountID) +} + +// --- Task --- + +func RunTask(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.RunTaskInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).RunTask(input, accountID) +} + +func DescribeTasks(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.DescribeTasksInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).DescribeTasks(input, accountID) +} + +func ListTasks(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.ListTasksInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).ListTasks(input, accountID) +} diff --git a/spinifex/gateway/ecs/handler.go b/spinifex/gateway/ecs/handler.go index eb838ccaa..b52b9e37f 100644 --- a/spinifex/gateway/ecs/handler.go +++ b/spinifex/gateway/ecs/handler.go @@ -3,9 +3,9 @@ // (AmazonEC2ContainerServiceV20141113.), matching aws-sdk-go's // service/ecs request shape. // -// The full v1 action namespace (ecs-v1.md §1) is registered as the API contract; -// every action resolves to the shared NotImplemented stub until its real handler -// lands in a later Phase 4 sprint. +// The full v1 action namespace (ecs-v1.md §1) is registered as the API contract. +// Cluster, task-definition, container-instance and task actions dispatch to the +// daemon over NATS; the remaining actions resolve to the NotImplemented stub. package gateway_ecs import ( @@ -37,30 +37,30 @@ func NotImplemented(_ *nats.Conn, _ string, _ []byte) (any, error) { } // Actions is the authoritative ECS control-plane action namespace (ecs-v1.md §1). -// Real implementations replace the NotImplemented value as each action lands; the -// key set is the v1 API contract. +// Wired actions point at their handler; unimplemented actions keep the +// NotImplemented value. The key set is the v1 API contract. var Actions = map[string]Handler{ // Cluster. PutClusterCapacityProviders is a no-op stub in v1 (Q15). - "CreateCluster": NotImplemented, - "DescribeClusters": NotImplemented, - "ListClusters": NotImplemented, + "CreateCluster": CreateCluster, + "DescribeClusters": DescribeClusters, + "ListClusters": ListClusters, "DeleteCluster": NotImplemented, "UpdateCluster": NotImplemented, "PutClusterCapacityProviders": NotImplemented, // Task definition. - "RegisterTaskDefinition": NotImplemented, + "RegisterTaskDefinition": RegisterTaskDefinition, "DeregisterTaskDefinition": NotImplemented, - "DescribeTaskDefinition": NotImplemented, - "ListTaskDefinitions": NotImplemented, + "DescribeTaskDefinition": DescribeTaskDefinition, + "ListTaskDefinitions": ListTaskDefinitions, "ListTaskDefinitionFamilies": NotImplemented, - // Task. - "RunTask": NotImplemented, + // Task. StartTask/StopTask land with the 4f service controller + reaper. + "RunTask": RunTask, "StartTask": NotImplemented, "StopTask": NotImplemented, - "DescribeTasks": NotImplemented, - "ListTasks": NotImplemented, + "DescribeTasks": DescribeTasks, + "ListTasks": ListTasks, // Service. ListServicesByNamespace is a no-op stub in v1. "CreateService": NotImplemented, @@ -71,10 +71,10 @@ var Actions = map[string]Handler{ "ListServicesByNamespace": NotImplemented, // Container instance. - "RegisterContainerInstance": NotImplemented, + "RegisterContainerInstance": RegisterContainerInstance, "DeregisterContainerInstance": NotImplemented, - "DescribeContainerInstances": NotImplemented, - "ListContainerInstances": NotImplemented, + "DescribeContainerInstances": DescribeContainerInstances, + "ListContainerInstances": ListContainerInstances, "UpdateContainerInstancesState": NotImplemented, // Account settings (passthrough; no enforcement v1). diff --git a/spinifex/gateway/ecs/handler_test.go b/spinifex/gateway/ecs/handler_test.go index 5e9bf1447..2fa499ced 100644 --- a/spinifex/gateway/ecs/handler_test.go +++ b/spinifex/gateway/ecs/handler_test.go @@ -17,13 +17,26 @@ func TestNotImplemented(t *testing.T) { assert.Equal(t, awserrors.ErrorNotImplemented, err.Error()) } -// Every registered action is the NotImplemented stub in this sprint. -func TestActions_AllNotImplemented(t *testing.T) { +// wiredActions are dispatched to the daemon over NATS; the rest are still the +// NotImplemented stub. Wired actions error here because the nil NATS conn cannot +// reach the daemon — that they no longer return NotImplemented is the assertion. +var wiredActions = map[string]bool{ + "CreateCluster": true, "DescribeClusters": true, "ListClusters": true, + "RegisterTaskDefinition": true, "DescribeTaskDefinition": true, "ListTaskDefinitions": true, + "RegisterContainerInstance": true, "DescribeContainerInstances": true, "ListContainerInstances": true, + "RunTask": true, "DescribeTasks": true, "ListTasks": true, +} + +func TestActions_StubsReturnNotImplemented(t *testing.T) { require.NotEmpty(t, Actions) for action, h := range Actions { _, err := h(nil, "123456789012", []byte("{}")) require.Error(t, err, "action %q", action) - assert.Equal(t, awserrors.ErrorNotImplemented, err.Error(), "action %q", action) + if wiredActions[action] { + assert.NotEqual(t, awserrors.ErrorNotImplemented, err.Error(), "wired action %q should not be a stub", action) + continue + } + assert.Equal(t, awserrors.ErrorNotImplemented, err.Error(), "stub action %q", action) } } diff --git a/spinifex/gateway/ecs_test.go b/spinifex/gateway/ecs_test.go index c1c2164d5..504551ce0 100644 --- a/spinifex/gateway/ecs_test.go +++ b/spinifex/gateway/ecs_test.go @@ -32,7 +32,7 @@ func TestECSActionFromTarget(t *testing.T) { // The Actions map is the v1 API contract (ecs-v1.md §1): every action across the // Cluster / TaskDefinition / Task / Service / ContainerInstance / Account / Tags -// surfaces must be registered, even though each is a 501 stub in this sprint. +// surfaces must be registered, whether wired to the daemon or a 501 stub. func TestECSActionsMap_NamespaceRegistered(t *testing.T) { namespace := []string{ // Cluster @@ -76,12 +76,11 @@ func TestECSRequest_UnknownAction(t *testing.T) { assert.Equal(t, awserrors.ErrorInvalidAction, err.Error()) } -// Every registered action resolves to the 501 stub until its real handler lands -// in a later Phase 4 sprint. +// An action whose real handler has not yet landed still resolves to the 501 stub. func TestECSRequest_KnownActionNotImplemented(t *testing.T) { gw := &GatewayConfig{DisableLogging: true} err := gw.ECS_Request(httptest.NewRecorder(), - setupECSRequest("AmazonEC2ContainerServiceV20141113.RunTask", "{}")) + setupECSRequest("AmazonEC2ContainerServiceV20141113.DeleteCluster", "{}")) require.Error(t, err) assert.Equal(t, awserrors.ErrorNotImplemented, err.Error()) } diff --git a/spinifex/gateway/gateway.go b/spinifex/gateway/gateway.go index e26d57530..6cb7256bc 100644 --- a/spinifex/gateway/gateway.go +++ b/spinifex/gateway/gateway.go @@ -229,8 +229,8 @@ const clusterUnavailableMsg = "cluster unavailable: NATS disconnected — check func (gw *GatewayConfig) writeClusterUnavailable(w http.ResponseWriter, _ *http.Request, svc string) { requestID := uuid.NewString() - // EKS uses AWS REST-JSON 1.1. - if svc == "eks" { + // EKS and ECS use AWS JSON 1.1. + if svc == "eks" || svc == "ecs" { body := GenerateEKSErrorResponse(awserrors.ErrorServiceUnavailable, clusterUnavailableMsg, requestID) w.Header().Set("Content-Type", eksJSONContentType) w.WriteHeader(http.StatusServiceUnavailable) @@ -281,8 +281,8 @@ func (gw *GatewayConfig) writeThrottleError(w http.ResponseWriter, r *http.Reque } errorMsg := awserrors.ErrorLookup[errorCode] - // EKS uses AWS REST-JSON 1.1. - if svc == "eks" { + // EKS and ECS use AWS JSON 1.1. + if svc == "eks" || svc == "ecs" { body := GenerateEKSErrorResponse(errorCode, errorMsg.Message, requestID) w.Header().Set("Content-Type", eksJSONContentType) w.WriteHeader(errorMsg.HTTPCode) @@ -497,8 +497,8 @@ func (gw *GatewayConfig) ErrorHandler(w http.ResponseWriter, r *http.Request, er errorMsg.HTTPCode = 500 } - // EKS, ECR, ACM, and tagging use AWS JSON 1.1; query/XML services fall through. - if svc == "eks" || svc == "ecr" || svc == "acm" || svc == "tagging" { + // EKS, ECR, ACM, ECS, and tagging use AWS JSON 1.1; query/XML services fall through. + if svc == "eks" || svc == "ecr" || svc == "acm" || svc == "ecs" || svc == "tagging" { body := GenerateEKSErrorResponse(err.Error(), errorMsg.Message, requestId) slog.Debug("Generated JSON error response", "service", svc, "error", err.Error(), "json", string(body), "requestId", requestId) w.Header().Set("Content-Type", eksJSONContentType) diff --git a/spinifex/handlers/ecs/bus/messages.go b/spinifex/handlers/ecs/bus/messages.go index a21b55c41..f5a575751 100644 --- a/spinifex/handlers/ecs/bus/messages.go +++ b/spinifex/handlers/ecs/bus/messages.go @@ -39,3 +39,71 @@ const ( StatusActive = "ACTIVE" StatusDraining = "DRAINING" ) + +// Task lifecycle status values, matching the AWS ECS task status enum, reported +// in TaskState.LastStatus and per-container in ContainerStatus.Status. +const ( + TaskStatusPending = "PENDING" + TaskStatusRunning = "RUNNING" + TaskStatusStopped = "STOPPED" +) + +// PortMapping is a container port exposed on the host (bridge mode v1). +type PortMapping struct { + ContainerPort int `json:"containerPort"` + HostPort int `json:"hostPort,omitempty"` + Protocol string `json:"protocol,omitempty"` // tcp | udp (default tcp) +} + +// AssignContainer is one container the agent must pull and run for a task. +type AssignContainer struct { + Name string `json:"name"` + Image string `json:"image"` + CPU int `json:"cpu,omitempty"` + MemoryMiB int `json:"memoryMiB,omitempty"` + Essential bool `json:"essential"` + Command []string `json:"command,omitempty"` + Environment map[string]string `json:"environment,omitempty"` + PortMappings []PortMapping `json:"portMappings,omitempty"` +} + +// Assign is published on AssignSubject when the scheduler places a task on this +// instance (scheduler → agent). The agent pulls each container image, creates and +// starts the containers via containerd, then reports progress on TaskStateSubject. +type Assign struct { + AccountID string `json:"accountId"` + ClusterName string `json:"clusterName"` + InstanceID string `json:"instanceId"` + TaskID string `json:"taskId"` + TaskARN string `json:"taskArn"` + TaskDefFamily string `json:"taskDefFamily"` + TaskDefRevision int `json:"taskDefRevision"` + Containers []AssignContainer `json:"containers"` + // CredID + TaskRoleARN are placeholders for the Sprint 4g IAM credential + // endpoint; the agent ignores them in 4e. + CredID string `json:"credId,omitempty"` + TaskRoleARN string `json:"taskRoleArn,omitempty"` + AssignedAt time.Time `json:"assignedAt"` +} + +// ContainerStatus is a single container's reported lifecycle state. +type ContainerStatus struct { + Name string `json:"name"` + Status string `json:"status"` // PENDING | RUNNING | STOPPED + ContainerID string `json:"containerId,omitempty"` + ExitCode *int `json:"exitCode,omitempty"` +} + +// TaskState is published on TaskStateSubject as the agent drives a task through +// its lifecycle (agent → scheduler). The scheduler updates the task record and +// recomputes the instance's remaining capacity. +type TaskState struct { + AccountID string `json:"accountId"` + ClusterName string `json:"clusterName"` + InstanceID string `json:"instanceId"` + TaskID string `json:"taskId"` + LastStatus string `json:"lastStatus"` // PENDING | RUNNING | STOPPED + Containers []ContainerStatus `json:"containers,omitempty"` + Reason string `json:"reason,omitempty"` + ReportedAt time.Time `json:"reportedAt"` +} diff --git a/spinifex/handlers/ecs/conv.go b/spinifex/handlers/ecs/conv.go new file mode 100644 index 000000000..67708b286 --- /dev/null +++ b/spinifex/handlers/ecs/conv.go @@ -0,0 +1,124 @@ +package handlers_ecs + +import ( + "strings" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus" +) + +// awsStringSlice dereferences a []*string, dropping nils/empties. +func awsStringSlice(in []*string) []string { + out := make([]string, 0, len(in)) + for _, p := range in { + if v := aws.StringValue(p); v != "" { + out = append(out, v) + } + } + return out +} + +// clusterShortName extracts the cluster name from a name or a cluster ARN. +func clusterShortName(ref string) string { + ref = strings.TrimSpace(ref) + if ref == "" { + return defaultCluster + } + if i := strings.LastIndex(ref, "cluster/"); i >= 0 { + return ref[i+len("cluster/"):] + } + return ref +} + +// containerInstanceShortID extracts the container-instance UUID from an ID or ARN. +func containerInstanceShortID(ref string) string { + if i := strings.LastIndexByte(ref, '/'); i >= 0 { + return ref[i+1:] + } + return ref +} + +// containerDefsFromAWS maps SDK container definitions to the persisted subset. +func containerDefsFromAWS(in []*ecs.ContainerDefinition) []ContainerDef { + out := make([]ContainerDef, 0, len(in)) + for _, c := range in { + if c == nil { + continue + } + def := ContainerDef{ + Name: aws.StringValue(c.Name), + Image: aws.StringValue(c.Image), + CPU: int(aws.Int64Value(c.Cpu)), + MemoryMiB: int(aws.Int64Value(c.Memory)), + Essential: aws.BoolValue(c.Essential), + Command: awsStringSlice(c.Command), + } + for _, e := range c.Environment { + if e == nil { + continue + } + if def.Environment == nil { + def.Environment = map[string]string{} + } + def.Environment[aws.StringValue(e.Name)] = aws.StringValue(e.Value) + } + for _, p := range c.PortMappings { + if p == nil { + continue + } + def.PortMappings = append(def.PortMappings, bus.PortMapping{ + ContainerPort: int(aws.Int64Value(p.ContainerPort)), + HostPort: int(aws.Int64Value(p.HostPort)), + Protocol: aws.StringValue(p.Protocol), + }) + } + out = append(out, def) + } + return out +} + +func (c ContainerDef) toAWS() *ecs.ContainerDefinition { + cd := &ecs.ContainerDefinition{ + Name: aws.String(c.Name), + Image: aws.String(c.Image), + Essential: aws.Bool(c.Essential), + } + if c.CPU > 0 { + cd.Cpu = aws.Int64(int64(c.CPU)) + } + if c.MemoryMiB > 0 { + cd.Memory = aws.Int64(int64(c.MemoryMiB)) + } + for _, cmd := range c.Command { + cd.Command = append(cd.Command, aws.String(cmd)) + } + for k, v := range c.Environment { + cd.Environment = append(cd.Environment, &ecs.KeyValuePair{Name: aws.String(k), Value: aws.String(v)}) + } + for _, p := range c.PortMappings { + pm := &ecs.PortMapping{ContainerPort: aws.Int64(int64(p.ContainerPort))} + if p.HostPort > 0 { + pm.HostPort = aws.Int64(int64(p.HostPort)) + } + if p.Protocol != "" { + pm.Protocol = aws.String(p.Protocol) + } + cd.PortMappings = append(cd.PortMappings, pm) + } + return cd +} + +// toAssignContainer maps a persisted container def to its bus assign payload. +func (c ContainerDef) toAssignContainer() bus.AssignContainer { + return bus.AssignContainer{ + Name: c.Name, + Image: c.Image, + CPU: c.CPU, + MemoryMiB: c.MemoryMiB, + Essential: c.Essential, + Command: c.Command, + Environment: c.Environment, + PortMappings: c.PortMappings, + } +} diff --git a/spinifex/handlers/ecs/conv_test.go b/spinifex/handlers/ecs/conv_test.go new file mode 100644 index 000000000..7fed6b93d --- /dev/null +++ b/spinifex/handlers/ecs/conv_test.go @@ -0,0 +1,83 @@ +package handlers_ecs + +import ( + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestARNBuilders(t *testing.T) { + assert.Equal(t, "arn:aws:ecs:ap-southeast-2:123456789012:cluster/web", + ClusterARN("ap-southeast-2", "123456789012", "web")) + assert.Equal(t, "arn:aws:ecs:ap-southeast-2:123456789012:task-definition/nginx:3", + TaskDefARN("ap-southeast-2", "123456789012", "nginx", 3)) + assert.Equal(t, "arn:aws:ecs:ap-southeast-2:123456789012:task/web/t-1", + TaskARN("ap-southeast-2", "123456789012", "web", "t-1")) + assert.Equal(t, "arn:aws:ecs:ap-southeast-2:123456789012:container-instance/web/i-1", + ContainerInstanceARN("ap-southeast-2", "123456789012", "web", "i-1")) +} + +func TestTaskDefReservedSums(t *testing.T) { + td := &TaskDefRecord{Containers: []ContainerDef{ + {CPU: 128, MemoryMiB: 256}, + {CPU: 64, MemoryMiB: 128}, + }} + assert.Equal(t, 192, td.reservedCPU()) + assert.Equal(t, 384, td.reservedMemory()) +} + +func TestClusterShortName(t *testing.T) { + assert.Equal(t, defaultCluster, clusterShortName("")) + assert.Equal(t, "web", clusterShortName("web")) + assert.Equal(t, "web", clusterShortName("arn:aws:ecs:r:a:cluster/web")) +} + +func TestContainerInstanceShortID(t *testing.T) { + assert.Equal(t, "i-1", containerInstanceShortID("i-1")) + assert.Equal(t, "i-1", containerInstanceShortID("arn:aws:ecs:r:a:container-instance/web/i-1")) +} + +func TestAWSStringSlice(t *testing.T) { + got := awsStringSlice([]*string{aws.String("a"), nil, aws.String(""), aws.String("b")}) + assert.Equal(t, []string{"a", "b"}, got) +} + +// containerDefsFromAWS → toAWS / toAssignContainer must preserve image, command, +// env, and port mappings (the fields the agent actually runs on). +func TestContainerDefRoundTrip(t *testing.T) { + in := []*ecs.ContainerDefinition{{ + Name: aws.String("web"), + Image: aws.String("registry/web:1"), + Cpu: aws.Int64(128), + Memory: aws.Int64(256), + Essential: aws.Bool(true), + Command: []*string{aws.String("/bin/sh"), aws.String("-c")}, + Environment: []*ecs.KeyValuePair{{Name: aws.String("FOO"), Value: aws.String("bar")}}, + PortMappings: []*ecs.PortMapping{{ContainerPort: aws.Int64(80), HostPort: aws.Int64(8080), Protocol: aws.String("tcp")}}, + }} + defs := containerDefsFromAWS(in) + require.Len(t, defs, 1) + d := defs[0] + assert.Equal(t, "registry/web:1", d.Image) + assert.Equal(t, 128, d.CPU) + assert.Equal(t, []string{"/bin/sh", "-c"}, d.Command) + assert.Equal(t, "bar", d.Environment["FOO"]) + require.Len(t, d.PortMappings, 1) + assert.Equal(t, 8080, d.PortMappings[0].HostPort) + + back := d.toAWS() + assert.Equal(t, "registry/web:1", aws.StringValue(back.Image)) + assert.Equal(t, int64(80), aws.Int64Value(back.PortMappings[0].ContainerPort)) + + ac := d.toAssignContainer() + assert.Equal(t, "web", ac.Name) + assert.Equal(t, "registry/web:1", ac.Image) + assert.Equal(t, "bar", ac.Environment["FOO"]) +} + +func TestContainerDefsFromAWS_SkipsNil(t *testing.T) { + assert.Empty(t, containerDefsFromAWS([]*ecs.ContainerDefinition{nil})) +} diff --git a/spinifex/handlers/ecs/instances.go b/spinifex/handlers/ecs/instances.go new file mode 100644 index 000000000..6c41a9fbb --- /dev/null +++ b/spinifex/handlers/ecs/instances.go @@ -0,0 +1,274 @@ +package handlers_ecs + +import ( + "encoding/json" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus" + "github.com/nats-io/nats.go" +) + +// listInstanceRecords returns all container-instance records in a cluster. +func (s *Service) listInstanceRecords(kv nats.KeyValue, cluster string) ([]InstanceRecord, error) { + keys, err := keysWithPrefix(kv, InstancesPrefix(cluster)) + if err != nil { + return nil, err + } + out := make([]InstanceRecord, 0, len(keys)) + for _, k := range keys { + var rec InstanceRecord + found, err := getJSON(kv, k, &rec) + if err != nil { + return nil, err + } + if found { + out = append(out, rec) + } + } + return out, nil +} + +// RegisterContainerInstance is the AWS-API registration path. In 4e the agent +// normally registers over the Layer-2 bus; this keeps API parity by writing the +// same record shape from an explicit call. +func (s *Service) RegisterContainerInstance(input *ecs.RegisterContainerInstanceInput, accountID string) (*ecs.RegisterContainerInstanceOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + instanceID := aws.StringValue(input.InstanceIdentityDocument) + if instanceID == "" { + instanceID = "ci-" + time.Now().UTC().Format("20060102150405") + } + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + rec, err := s.upsertInstance(kv, accountID, cluster, instanceID, func(r *InstanceRecord) { + for _, res := range input.TotalResources { + switch aws.StringValue(res.Name) { + case "CPU": + r.TotalCPU = int(aws.Int64Value(res.IntegerValue)) + case "MEMORY": + r.TotalMemoryMiB = int(aws.Int64Value(res.IntegerValue)) + } + } + }) + if err != nil { + return nil, err + } + return &ecs.RegisterContainerInstanceOutput{ContainerInstance: s.instanceToAWS(rec)}, nil +} + +// DescribeContainerInstances returns records for the named container instances. +func (s *Service) DescribeContainerInstances(input *ecs.DescribeContainerInstancesInput, accountID string) (*ecs.DescribeContainerInstancesOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + out := &ecs.DescribeContainerInstancesOutput{} + for _, ref := range awsStringSlice(input.ContainerInstances) { + id := containerInstanceShortID(ref) + var rec InstanceRecord + found, err := getJSON(kv, InstanceKey(cluster, id), &rec) + if err != nil { + return nil, err + } + if found { + out.ContainerInstances = append(out.ContainerInstances, s.instanceToAWS(&rec)) + } else { + out.Failures = append(out.Failures, &ecs.Failure{Arn: aws.String(ref), Reason: aws.String("MISSING")}) + } + } + return out, nil +} + +// ListContainerInstances returns the ARNs of all container instances in a cluster. +func (s *Service) ListContainerInstances(input *ecs.ListContainerInstancesInput, accountID string) (*ecs.ListContainerInstancesOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + recs, err := s.listInstanceRecords(kv, cluster) + if err != nil { + return nil, err + } + out := &ecs.ListContainerInstancesOutput{} + for i := range recs { + out.ContainerInstanceArns = append(out.ContainerInstanceArns, aws.String(recs[i].ARN)) + } + return out, nil +} + +// upsertInstance reads-or-creates the instance record, applies mutate, and writes +// it back. Used by both the AWS register path and the bus register handler. +func (s *Service) upsertInstance(kv nats.KeyValue, accountID, cluster, instanceID string, mutate func(*InstanceRecord)) (*InstanceRecord, error) { + var rec InstanceRecord + found, err := getJSON(kv, InstanceKey(cluster, instanceID), &rec) + if err != nil { + return nil, err + } + now := time.Now().UTC() + if !found { + rec = InstanceRecord{ + InstanceID: instanceID, + ARN: ContainerInstanceARN(s.region, accountID, cluster, instanceID), + Cluster: cluster, + Status: InstanceStatusActive, + RegisteredAt: now, + } + } + rec.LastSeen = now + if mutate != nil { + mutate(&rec) + } + if err := putJSON(kv, InstanceKey(cluster, instanceID), &rec); err != nil { + return nil, err + } + return &rec, nil +} + +func (s *Service) instanceToAWS(r *InstanceRecord) *ecs.ContainerInstance { + registered := []*ecs.Resource{ + {Name: aws.String("CPU"), Type: aws.String("INTEGER"), IntegerValue: aws.Int64(int64(r.TotalCPU))}, + {Name: aws.String("MEMORY"), Type: aws.String("INTEGER"), IntegerValue: aws.Int64(int64(r.TotalMemoryMiB))}, + } + remaining := []*ecs.Resource{ + {Name: aws.String("CPU"), Type: aws.String("INTEGER"), IntegerValue: aws.Int64(int64(r.TotalCPU - r.ReservedCPU))}, + {Name: aws.String("MEMORY"), Type: aws.String("INTEGER"), IntegerValue: aws.Int64(int64(r.TotalMemoryMiB - r.ReservedMemoryMiB))}, + } + return &ecs.ContainerInstance{ + ContainerInstanceArn: aws.String(r.ARN), + Ec2InstanceId: aws.String(r.InstanceID), + Status: aws.String(r.Status), + AgentConnected: aws.Bool(r.Status == InstanceStatusActive), + RunningTasksCount: aws.Int64(int64(len(r.PlacedTasks))), + RegisteredResources: registered, + RemainingResources: remaining, + VersionInfo: &ecs.VersionInfo{AgentVersion: aws.String(r.AgentVersion)}, + } +} + +// --- Layer-2 bus event handlers (called by the scheduler) --- + +// recordRegister upserts a container-instance record from a bus RegisterInstance. +func (s *Service) recordRegister(msg *bus.RegisterInstance) error { + kv, err := s.bucket(msg.AccountID) + if err != nil { + return err + } + _, err = s.upsertInstance(kv, msg.AccountID, msg.ClusterName, msg.InstanceID, func(r *InstanceRecord) { + r.AZ = msg.AZ + r.Hostname = msg.Hostname + r.AgentVersion = msg.AgentVersion + r.TotalCPU = msg.Capacity.CPU + r.TotalMemoryMiB = msg.Capacity.MemoryMiB + r.Status = InstanceStatusActive + }) + return err +} + +// recordHeartbeat refreshes an instance's LastSeen and status from a bus beat. +func (s *Service) recordHeartbeat(msg *bus.Heartbeat) error { + kv, err := s.bucket(msg.AccountID) + if err != nil { + return err + } + var rec InstanceRecord + found, err := getJSON(kv, InstanceKey(msg.ClusterName, msg.InstanceID), &rec) + if err != nil || !found { + return err + } + rec.LastSeen = time.Now().UTC() + if msg.Status != "" { + rec.Status = msg.Status + } + return putJSON(kv, InstanceKey(msg.ClusterName, msg.InstanceID), &rec) +} + +// recordTaskState applies an agent task-state report: it updates the task record +// and, on STOPPED, releases the reserved capacity back to the instance. +func (s *Service) recordTaskState(msg *bus.TaskState) error { + kv, err := s.bucket(msg.AccountID) + if err != nil { + return err + } + var task TaskRecord + found, err := getJSON(kv, TaskKey(msg.ClusterName, msg.TaskID), &task) + if err != nil || !found { + return err + } + + prev := task.LastStatus + task.LastStatus = msg.LastStatus + if len(msg.Containers) > 0 { + task.Containers = task.Containers[:0] + for _, c := range msg.Containers { + task.Containers = append(task.Containers, ContainerState{ + Name: c.Name, Status: c.Status, ContainerID: c.ContainerID, ExitCode: c.ExitCode, + }) + } + } + now := time.Now().UTC() + if msg.LastStatus == TaskStatusRunning && task.StartedAt.IsZero() { + task.StartedAt = now + } + if msg.LastStatus == TaskStatusStopped { + task.StoppedAt = now + if msg.Reason != "" { + task.StoppedReason = msg.Reason + } + } + if err := putJSON(kv, TaskKey(msg.ClusterName, msg.TaskID), &task); err != nil { + return err + } + + // Release capacity once, on the transition into STOPPED. + if msg.LastStatus == TaskStatusStopped && prev != TaskStatusStopped { + return s.releaseReservation(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID, task.ReservedCPU, task.ReservedMemoryMiB) + } + return nil +} + +// releaseReservation returns a stopped task's capacity to its instance under CAS. +func (s *Service) releaseReservation(kv nats.KeyValue, cluster, instanceID, taskID string, cpu, mem int) error { + for range reservePlacementRetries { + entry, err := kv.Get(InstanceKey(cluster, instanceID)) + if err != nil { + return nil //nolint:nilerr // instance gone; nothing to release + } + var rec InstanceRecord + if uerr := json.Unmarshal(entry.Value(), &rec); uerr != nil { + return uerr + } + rec.ReservedCPU = maxZero(rec.ReservedCPU - cpu) + rec.ReservedMemoryMiB = maxZero(rec.ReservedMemoryMiB - mem) + rec.PlacedTasks = removeString(rec.PlacedTasks, taskID) + data, merr := json.Marshal(&rec) + if merr != nil { + return merr + } + if _, uerr := kv.Update(InstanceKey(cluster, instanceID), data, entry.Revision()); uerr == nil { + return nil + } + } + return nil +} + +func maxZero(v int) int { + if v < 0 { + return 0 + } + return v +} + +func removeString(in []string, s string) []string { + out := in[:0] + for _, v := range in { + if v != s { + out = append(out, v) + } + } + return out +} diff --git a/spinifex/handlers/ecs/placement.go b/spinifex/handlers/ecs/placement.go new file mode 100644 index 000000000..668becbd8 --- /dev/null +++ b/spinifex/handlers/ecs/placement.go @@ -0,0 +1,86 @@ +package handlers_ecs + +import ( + "errors" + "sort" + "strings" +) + +// Placement strategy identifiers (ecs-v1.md Q15). Default is binpack:memory. +const ( + StrategyBinpack = "binpack" + StrategySpread = "spread" + StrategyRandom = "random" +) + +// ErrNoCapacity is returned when no ACTIVE instance can fit the task. +var ErrNoCapacity = errors.New("no container instance has capacity for the task") + +// remainingCPU/remainingMemory report an instance's unreserved capacity. +func (r *InstanceRecord) remainingCPU() int { return r.TotalCPU - r.ReservedCPU } +func (r *InstanceRecord) remainingMemory() int { return r.TotalMemoryMiB - r.ReservedMemoryMiB } + +// fits reports whether the instance is ACTIVE and has room for the reservation. +func (r *InstanceRecord) fits(cpu, mem int) bool { + if r.Status != InstanceStatusActive { + return false + } + return r.remainingCPU() >= cpu && r.remainingMemory() >= mem +} + +// placeTask selects a container instance for a task reserving (cpu, mem) using +// the requested strategy. Candidates are filtered to ACTIVE instances that fit; +// ties broken by instance ID for determinism. strategy "" defaults to binpack. +// +// binpack: pick the instance with the LEAST remaining memory that still fits +// (tightest pack). spread: pick the MOST remaining memory (widest spread). +// random: caller-stable first fit by instance ID. +func placeTask(instances []InstanceRecord, cpu, mem int, strategy string) (*InstanceRecord, error) { + candidates := make([]InstanceRecord, 0, len(instances)) + for _, inst := range instances { + if inst.fits(cpu, mem) { + candidates = append(candidates, inst) + } + } + if len(candidates) == 0 { + return nil, ErrNoCapacity + } + + switch normalizeStrategy(strategy) { + case StrategySpread: + sort.Slice(candidates, func(i, j int) bool { + if candidates[i].remainingMemory() != candidates[j].remainingMemory() { + return candidates[i].remainingMemory() > candidates[j].remainingMemory() + } + return candidates[i].InstanceID < candidates[j].InstanceID + }) + case StrategyRandom: + sort.Slice(candidates, func(i, j int) bool { + return candidates[i].InstanceID < candidates[j].InstanceID + }) + default: // binpack + sort.Slice(candidates, func(i, j int) bool { + if candidates[i].remainingMemory() != candidates[j].remainingMemory() { + return candidates[i].remainingMemory() < candidates[j].remainingMemory() + } + return candidates[i].InstanceID < candidates[j].InstanceID + }) + } + + chosen := candidates[0] + return &chosen, nil +} + +// normalizeStrategy maps "binpack:memory"/"binpack:cpu" → "binpack" and lowercases. +func normalizeStrategy(s string) string { + s = strings.ToLower(strings.TrimSpace(s)) + if i := strings.IndexByte(s, ':'); i >= 0 { + s = s[:i] + } + switch s { + case StrategySpread, StrategyRandom: + return s + default: + return StrategyBinpack + } +} diff --git a/spinifex/handlers/ecs/placement_test.go b/spinifex/handlers/ecs/placement_test.go new file mode 100644 index 000000000..9be1e144f --- /dev/null +++ b/spinifex/handlers/ecs/placement_test.go @@ -0,0 +1,79 @@ +package handlers_ecs + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func inst(id string, total, reserved int) InstanceRecord { + return InstanceRecord{ + InstanceID: id, + Status: InstanceStatusActive, + TotalCPU: 10000, + ReservedCPU: 0, + TotalMemoryMiB: total, + ReservedMemoryMiB: reserved, + } +} + +func TestPlaceTask_Binpack_PicksTightest(t *testing.T) { + // remaining mem: a=900, b=300, c=600 → binpack picks b (least remaining). + instances := []InstanceRecord{ + inst("a", 1000, 100), + inst("b", 1000, 700), + inst("c", 1000, 400), + } + got, err := placeTask(instances, 100, 200, "binpack:memory") + require.NoError(t, err) + assert.Equal(t, "b", got.InstanceID) +} + +func TestPlaceTask_Spread_PicksWidest(t *testing.T) { + instances := []InstanceRecord{ + inst("a", 1000, 100), // 900 + inst("b", 1000, 700), // 300 + inst("c", 1000, 400), // 600 + } + got, err := placeTask(instances, 100, 200, StrategySpread) + require.NoError(t, err) + assert.Equal(t, "a", got.InstanceID) +} + +func TestPlaceTask_Random_StableByID(t *testing.T) { + instances := []InstanceRecord{inst("z", 1000, 0), inst("a", 1000, 0), inst("m", 1000, 0)} + got, err := placeTask(instances, 100, 100, StrategyRandom) + require.NoError(t, err) + assert.Equal(t, "a", got.InstanceID) +} + +func TestPlaceTask_NoCapacity(t *testing.T) { + instances := []InstanceRecord{inst("a", 100, 90)} + _, err := placeTask(instances, 0, 50, StrategyBinpack) + assert.ErrorIs(t, err, ErrNoCapacity) +} + +func TestPlaceTask_SkipsDraining(t *testing.T) { + d := inst("drain", 1000, 0) + d.Status = InstanceStatusDraining + _, err := placeTask([]InstanceRecord{d}, 0, 100, StrategyBinpack) + assert.ErrorIs(t, err, ErrNoCapacity) +} + +func TestNormalizeStrategy(t *testing.T) { + assert.Equal(t, StrategyBinpack, normalizeStrategy("")) + assert.Equal(t, StrategyBinpack, normalizeStrategy("binpack:memory")) + assert.Equal(t, StrategyBinpack, normalizeStrategy("BinPack:cpu")) + assert.Equal(t, StrategySpread, normalizeStrategy("spread")) + assert.Equal(t, StrategyRandom, normalizeStrategy("RANDOM")) + assert.Equal(t, StrategyBinpack, normalizeStrategy("bogus")) +} + +func TestInstanceRecord_Fits(t *testing.T) { + r := inst("a", 1000, 600) // remaining mem 400, cpu 10000 + assert.True(t, r.fits(100, 400)) + assert.False(t, r.fits(100, 401)) + r.Status = InstanceStatusDraining + assert.False(t, r.fits(0, 0)) +} diff --git a/spinifex/handlers/ecs/records.go b/spinifex/handlers/ecs/records.go new file mode 100644 index 000000000..30d3b0fee --- /dev/null +++ b/spinifex/handlers/ecs/records.go @@ -0,0 +1,149 @@ +package handlers_ecs + +import ( + "fmt" + "time" + + "github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus" +) + +// Lifecycle status constants matching the AWS ECS enums. Task statuses are +// re-exported from the bus package so the scheduler and the wire payloads agree. +const ( + ClusterStatusActive = "ACTIVE" + + InstanceStatusActive = "ACTIVE" + InstanceStatusDraining = "DRAINING" + + TaskDefStatusActive = "ACTIVE" + + TaskStatusPending = bus.TaskStatusPending + TaskStatusRunning = bus.TaskStatusRunning + TaskStatusStopped = bus.TaskStatusStopped +) + +// ARN builders for the ECS resource shapes (ecs-v1.md §1). Region + accountID +// scope every ARN; the partition is fixed to "aws" to match the rest of the +// stack. +func ClusterARN(region, accountID, name string) string { + return fmt.Sprintf("arn:aws:ecs:%s:%s:cluster/%s", region, accountID, name) +} + +func TaskDefARN(region, accountID, family string, rev int) string { + return fmt.Sprintf("arn:aws:ecs:%s:%s:task-definition/%s:%d", region, accountID, family, rev) +} + +func TaskARN(region, accountID, cluster, taskID string) string { + return fmt.Sprintf("arn:aws:ecs:%s:%s:task/%s/%s", region, accountID, cluster, taskID) +} + +func ContainerInstanceARN(region, accountID, cluster, ciID string) string { + return fmt.Sprintf("arn:aws:ecs:%s:%s:container-instance/%s/%s", region, accountID, cluster, ciID) +} + +// ClusterRecord is the persisted cluster meta at ClusterMetaKey. +type ClusterRecord struct { + Name string `json:"name"` + ARN string `json:"arn"` + Status string `json:"status"` + Tags map[string]string `json:"tags,omitempty"` + CreatedAt time.Time `json:"createdAt"` +} + +// ContainerDef is the persisted subset of an ecs.ContainerDefinition needed to +// pull and run a container (bridge mode v1). +type ContainerDef struct { + Name string `json:"name"` + Image string `json:"image"` + CPU int `json:"cpu,omitempty"` + MemoryMiB int `json:"memoryMiB,omitempty"` + Essential bool `json:"essential"` + Command []string `json:"command,omitempty"` + Environment map[string]string `json:"environment,omitempty"` + PortMappings []bus.PortMapping `json:"portMappings,omitempty"` +} + +// TaskDefRecord is the persisted task definition revision at TaskDefRevKey. +type TaskDefRecord struct { + Family string `json:"family"` + Revision int `json:"revision"` + ARN string `json:"arn"` + NetworkMode string `json:"networkMode,omitempty"` + CPU string `json:"cpu,omitempty"` + Memory string `json:"memory,omitempty"` + Containers []ContainerDef `json:"containers"` + Status string `json:"status"` + RegisteredAt time.Time `json:"registeredAt"` +} + +// reservedCPU/reservedMemory sum the task definition's per-container reservations +// used for bin-pack placement. A taskdef-level cpu/memory is not modelled in v1; +// placement uses the container sums. +func (t *TaskDefRecord) reservedCPU() int { + total := 0 + for _, c := range t.Containers { + total += c.CPU + } + return total +} + +func (t *TaskDefRecord) reservedMemory() int { + total := 0 + for _, c := range t.Containers { + total += c.MemoryMiB + } + return total +} + +// InstanceRecord is the persisted container-instance state at InstanceKey. The +// scheduler writes it from the Layer-2 bus (register/heartbeat) and reserves +// capacity by appending placed task IDs. +type InstanceRecord struct { + InstanceID string `json:"instanceId"` + ARN string `json:"arn"` + Cluster string `json:"cluster"` + AZ string `json:"availabilityZone,omitempty"` + Hostname string `json:"hostname,omitempty"` + Status string `json:"status"` + TotalCPU int `json:"totalCpu"` + TotalMemoryMiB int `json:"totalMemoryMiB"` + // ReservedCPU/ReservedMemoryMiB track capacity committed to placed tasks; + // placement increments them under a KV CAS and the task-state STOPPED path + // releases them. Remaining = Total - Reserved. + ReservedCPU int `json:"reservedCpu"` + ReservedMemoryMiB int `json:"reservedMemoryMiB"` + AgentVersion string `json:"agentVersion,omitempty"` + PlacedTasks []string `json:"placedTasks,omitempty"` + RegisteredAt time.Time `json:"registeredAt"` + LastSeen time.Time `json:"lastSeen"` +} + +// ContainerState is a container's last-reported status within a task record. +type ContainerState struct { + Name string `json:"name"` + Status string `json:"status"` + ContainerID string `json:"containerId,omitempty"` + ExitCode *int `json:"exitCode,omitempty"` +} + +// TaskRecord is the persisted task state at TaskKey; source of truth for +// DescribeTasks and the placement/capacity accounting. +type TaskRecord struct { + TaskID string `json:"taskId"` + ARN string `json:"arn"` + Cluster string `json:"cluster"` + TaskDefFamily string `json:"taskDefFamily"` + TaskDefRevision int `json:"taskDefRevision"` + TaskDefARN string `json:"taskDefArn"` + ContainerInstanceID string `json:"containerInstanceId,omitempty"` + ContainerInstanceARN string `json:"containerInstanceArn,omitempty"` + DesiredStatus string `json:"desiredStatus"` + LastStatus string `json:"lastStatus"` + StoppedReason string `json:"stoppedReason,omitempty"` + ReservedCPU int `json:"reservedCpu"` + ReservedMemoryMiB int `json:"reservedMemoryMiB"` + Containers []ContainerState `json:"containers,omitempty"` + CreatedAt time.Time `json:"createdAt"` + StartedAt time.Time `json:"startedAt,omitzero"` + StoppedAt time.Time `json:"stoppedAt,omitzero"` +} diff --git a/spinifex/handlers/ecs/scheduler.go b/spinifex/handlers/ecs/scheduler.go new file mode 100644 index 000000000..04ce88d02 --- /dev/null +++ b/spinifex/handlers/ecs/scheduler.go @@ -0,0 +1,288 @@ +package handlers_ecs + +import ( + "context" + "encoding/json" + "log/slog" + "strings" + "sync" + "time" + + "github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus" + "github.com/nats-io/nats.go" +) + +// Scheduler timing. The leader lease lives in spinifex-ecs-leader (60s TTL); the +// holder refreshes well inside that. The reaper marks an instance DRAINING after +// three missed 30s heartbeats. +const ( + schedulerLeaderKey = "scheduler" + leaseRefresh = 20 * time.Second + reaperInterval = 30 * time.Second + heartbeatTimeout = 90 * time.Second + stoppedReasonReaped = "ContainerInstance disconnected" +) + +// Scheduler is the per-daemon ECS control loop. A single leader (elected via the +// shared leader bucket) owns the Layer-2 bus subscriptions and the heartbeat +// reaper; losers idle and retry. Placement itself happens synchronously in the +// RunTask handler, so a brief leaderless gap never blocks RunTask. +type Scheduler struct { + nc *nats.Conn + svc *Service + holder string + + mu sync.Mutex + leader bool + subs []*nats.Subscription +} + +// NewScheduler constructs a Scheduler. holder identifies this daemon in the lease. +func NewScheduler(nc *nats.Conn, svc *Service, holder string) *Scheduler { + return &Scheduler{nc: nc, svc: svc, holder: holder} +} + +// Run drives the leadership + reaper loop until ctx is cancelled. It is intended +// to run as a daemon-boot goroutine; panics are the caller's recover concern. +func (sc *Scheduler) Run(ctx context.Context) { + leaseTicker := time.NewTicker(leaseRefresh) + reaperTicker := time.NewTicker(reaperInterval) + defer leaseTicker.Stop() + defer reaperTicker.Stop() + + sc.evaluateLeadership(ctx) + for { + select { + case <-ctx.Done(): + sc.relinquish() + return + case <-leaseTicker.C: + sc.evaluateLeadership(ctx) + case <-reaperTicker.C: + if sc.isLeader() { + sc.reap() + } + } + } +} + +func (sc *Scheduler) isLeader() bool { + sc.mu.Lock() + defer sc.mu.Unlock() + return sc.leader +} + +// evaluateLeadership acquires or refreshes the lease, wiring up (or tearing down) +// the bus subscriptions as leadership changes. +func (sc *Scheduler) evaluateLeadership(ctx context.Context) { + won := sc.acquireOrRefresh() + sc.mu.Lock() + was := sc.leader + sc.leader = won + sc.mu.Unlock() + + switch { + case won && !was: + if err := sc.subscribeBus(ctx); err != nil { + slog.Error("ECS scheduler: bus subscribe failed", "holder", sc.holder, "err", err) + } else { + slog.Info("ECS scheduler: elected leader, bus subscriptions active", "holder", sc.holder) + } + case !won && was: + sc.unsubscribeBus() + slog.Info("ECS scheduler: lost leadership, bus subscriptions dropped", "holder", sc.holder) + } +} + +// acquireOrRefresh tries to claim the scheduler lease, refreshing it (resetting +// the TTL) when we already hold it. Returns true when we are the leader. +func (sc *Scheduler) acquireOrRefresh() bool { + js, err := sc.nc.JetStream() + if err != nil { + return false + } + kv, err := InitLeaderBucket(js) + if err != nil { + return false + } + if _, err := kv.Create(schedulerLeaderKey, []byte(sc.holder)); err == nil { + return true + } + entry, err := kv.Get(schedulerLeaderKey) + if err != nil { + return false + } + if string(entry.Value()) != sc.holder { + return false // someone else holds it + } + // We hold it — refresh to reset the TTL. + if _, err := kv.Put(schedulerLeaderKey, []byte(sc.holder)); err != nil { + return false + } + return true +} + +// relinquish drops subscriptions and deletes the lease key on shutdown. +func (sc *Scheduler) relinquish() { + sc.unsubscribeBus() + js, err := sc.nc.JetStream() + if err != nil { + return + } + kv, err := InitLeaderBucket(js) + if err != nil { + return + } + if entry, gerr := kv.Get(schedulerLeaderKey); gerr == nil && string(entry.Value()) == sc.holder { + _ = kv.Delete(schedulerLeaderKey) + } +} + +// subscribeBus wires the wildcard Layer-2 subscriptions onto the service KV +// writers. All clusters/accounts fan into the leader. +func (sc *Scheduler) subscribeBus(_ context.Context) error { + register, err := sc.nc.Subscribe("ecs.bus.*.*.instance-register.*", sc.onRegister) + if err != nil { + return err + } + heartbeat, err := sc.nc.Subscribe("ecs.bus.*.*.instance-heartbeat.*", sc.onHeartbeat) + if err != nil { + _ = register.Unsubscribe() + return err + } + taskState, err := sc.nc.Subscribe("ecs.bus.*.*.task-state.*", sc.onTaskState) + if err != nil { + _ = register.Unsubscribe() + _ = heartbeat.Unsubscribe() + return err + } + sc.mu.Lock() + sc.subs = []*nats.Subscription{register, heartbeat, taskState} + sc.mu.Unlock() + return nil +} + +func (sc *Scheduler) unsubscribeBus() { + sc.mu.Lock() + subs := sc.subs + sc.subs = nil + sc.mu.Unlock() + for _, s := range subs { + _ = s.Unsubscribe() + } +} + +func (sc *Scheduler) onRegister(msg *nats.Msg) { + var m bus.RegisterInstance + if err := json.Unmarshal(msg.Data, &m); err != nil { + slog.Warn("ECS scheduler: bad register payload", "err", err) + return + } + if err := sc.svc.recordRegister(&m); err != nil { + slog.Error("ECS scheduler: record register failed", "instance", m.InstanceID, "err", err) + return + } + slog.Info("ECS scheduler: container instance registered", "cluster", m.ClusterName, "instance", m.InstanceID) +} + +func (sc *Scheduler) onHeartbeat(msg *nats.Msg) { + var m bus.Heartbeat + if err := json.Unmarshal(msg.Data, &m); err != nil { + return + } + if err := sc.svc.recordHeartbeat(&m); err != nil { + slog.Debug("ECS scheduler: record heartbeat failed", "instance", m.InstanceID, "err", err) + } +} + +func (sc *Scheduler) onTaskState(msg *nats.Msg) { + var m bus.TaskState + if err := json.Unmarshal(msg.Data, &m); err != nil { + return + } + if err := sc.svc.recordTaskState(&m); err != nil { + slog.Error("ECS scheduler: record task-state failed", "task", m.TaskID, "err", err) + return + } + slog.Info("ECS scheduler: task state", "task", m.TaskID, "status", m.LastStatus) +} + +// reap marks instances that have missed their heartbeat window DRAINING and stops +// their tasks, releasing capacity. Iterates every ECS account bucket. +func (sc *Scheduler) reap() { + js, err := sc.nc.JetStream() + if err != nil { + return + } + now := time.Now().UTC() + for bucket := range js.KeyValueStoreNames() { + accountID, ok := accountIDFromBucket(bucket) + if !ok { + continue + } + kv, err := js.KeyValue(bucket) + if err != nil { + continue + } + sc.reapBucket(kv, accountID, now) + } +} + +func (sc *Scheduler) reapBucket(kv nats.KeyValue, _ string, now time.Time) { + keys, err := keysWithPrefix(kv, "clusters/") + if err != nil { + return + } + for _, k := range keys { + if !strings.Contains(k, "/instances/") { + continue + } + var inst InstanceRecord + found, err := getJSON(kv, k, &inst) + if err != nil || !found { + continue + } + if inst.Status != InstanceStatusActive || now.Sub(inst.LastSeen) < heartbeatTimeout { + continue + } + slog.Warn("ECS scheduler: reaping disconnected instance", "cluster", inst.Cluster, "instance", inst.InstanceID, + "lastSeen", inst.LastSeen.Format(time.RFC3339)) + inst.Status = InstanceStatusDraining + if perr := putJSON(kv, k, &inst); perr != nil { + continue + } + sc.stopInstanceTasks(kv, inst.Cluster, inst.InstanceID) + } +} + +// stopInstanceTasks transitions a reaped instance's non-stopped tasks to STOPPED. +func (sc *Scheduler) stopInstanceTasks(kv nats.KeyValue, cluster, instanceID string) { + keys, err := keysWithPrefix(kv, TasksPrefix(cluster)) + if err != nil { + return + } + for _, k := range keys { + var task TaskRecord + found, err := getJSON(kv, k, &task) + if err != nil || !found { + continue + } + if task.ContainerInstanceID != instanceID || task.LastStatus == TaskStatusStopped { + continue + } + task.LastStatus = TaskStatusStopped + task.StoppedReason = stoppedReasonReaped + task.StoppedAt = time.Now().UTC() + if perr := putJSON(kv, k, &task); perr != nil { + continue + } + } +} + +// accountIDFromBucket extracts the account ID from an ECS per-account bucket name. +func accountIDFromBucket(bucket string) (string, bool) { + if !strings.HasPrefix(bucket, KVBucketECSAccountPrefix) { + return "", false + } + return strings.TrimPrefix(bucket, KVBucketECSAccountPrefix), true +} diff --git a/spinifex/handlers/ecs/service.go b/spinifex/handlers/ecs/service.go new file mode 100644 index 000000000..8f0ea9d3b --- /dev/null +++ b/spinifex/handlers/ecs/service.go @@ -0,0 +1,384 @@ +package handlers_ecs + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strconv" + "strings" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/nats-io/nats.go" +) + +// ECSService is the ECS control-plane contract implemented by the daemon and +// adapted onto NATS by NATSECSService. One method per wired AWS ECS action; the +// remaining actions stay NotImplemented at the gateway. +type ECSService interface { + CreateCluster(input *ecs.CreateClusterInput, accountID string) (*ecs.CreateClusterOutput, error) + DescribeClusters(input *ecs.DescribeClustersInput, accountID string) (*ecs.DescribeClustersOutput, error) + ListClusters(input *ecs.ListClustersInput, accountID string) (*ecs.ListClustersOutput, error) + + RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput, accountID string) (*ecs.RegisterTaskDefinitionOutput, error) + DescribeTaskDefinition(input *ecs.DescribeTaskDefinitionInput, accountID string) (*ecs.DescribeTaskDefinitionOutput, error) + ListTaskDefinitions(input *ecs.ListTaskDefinitionsInput, accountID string) (*ecs.ListTaskDefinitionsOutput, error) + + RegisterContainerInstance(input *ecs.RegisterContainerInstanceInput, accountID string) (*ecs.RegisterContainerInstanceOutput, error) + DescribeContainerInstances(input *ecs.DescribeContainerInstancesInput, accountID string) (*ecs.DescribeContainerInstancesOutput, error) + ListContainerInstances(input *ecs.ListContainerInstancesInput, accountID string) (*ecs.ListContainerInstancesOutput, error) + + RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTaskOutput, error) + DescribeTasks(input *ecs.DescribeTasksInput, accountID string) (*ecs.DescribeTasksOutput, error) + ListTasks(input *ecs.ListTasksInput, accountID string) (*ecs.ListTasksOutput, error) +} + +// Service is the daemon-side ECS control-plane implementation, backed by the +// per-account JetStream KV bucket. It publishes task assignments on the Layer-2 +// bus; the scheduler goroutine consumes instance/task-state events. +type Service struct { + nc *nats.Conn + region string + suffix string +} + +var _ ECSService = (*Service)(nil) + +// NewService constructs a Service bound to a NATS connection. region scopes the +// ARNs it mints; suffix is the AWS-parity internal DNS suffix (reserved for ECR +// endpoint composition). +func NewService(nc *nats.Conn, region, suffix string) *Service { + return &Service{nc: nc, region: region, suffix: suffix} +} + +// defaultCluster is the implicit cluster name when a request omits one. +const defaultCluster = "default" + +func (s *Service) js() (nats.JetStreamContext, error) { + if s.nc == nil { + return nil, errors.New("ecs service: nil nats connection") + } + return s.nc.JetStream() +} + +// bucket returns the per-account KV handle, creating it on first use. +func (s *Service) bucket(accountID string) (nats.KeyValue, error) { + js, err := s.js() + if err != nil { + return nil, err + } + return GetOrCreateAccountBucket(js, accountID) +} + +// getJSON reads key into out. Returns (false, nil) when the key is absent. +func getJSON(kv nats.KeyValue, key string, out any) (bool, error) { + entry, err := kv.Get(key) + if errors.Is(err, nats.ErrKeyNotFound) { + return false, nil + } + if err != nil { + return false, err + } + if err := json.Unmarshal(entry.Value(), out); err != nil { + return false, fmt.Errorf("unmarshal %s: %w", key, err) + } + return true, nil +} + +// putJSON marshals v and writes it at key. +func putJSON(kv nats.KeyValue, key string, v any) error { + data, err := json.Marshal(v) + if err != nil { + return err + } + if _, err := kv.Put(key, data); err != nil { + return fmt.Errorf("put %s: %w", key, err) + } + return nil +} + +// keysWithPrefix returns all KV keys under prefix. +func keysWithPrefix(kv nats.KeyValue, prefix string) ([]string, error) { + keys, err := kv.Keys() + if errors.Is(err, nats.ErrNoKeysFound) { + return nil, nil + } + if err != nil { + return nil, err + } + out := make([]string, 0, len(keys)) + for _, k := range keys { + if strings.HasPrefix(k, prefix) { + out = append(out, k) + } + } + sort.Strings(out) + return out, nil +} + +// --- Cluster --- + +// CreateCluster persists a cluster meta record. Idempotent: re-creating an +// existing cluster returns the existing ACTIVE record. +func (s *Service) CreateCluster(input *ecs.CreateClusterInput, accountID string) (*ecs.CreateClusterOutput, error) { + name := aws.StringValue(input.ClusterName) + if name == "" { + name = defaultCluster + } + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + + var rec ClusterRecord + found, err := getJSON(kv, ClusterMetaKey(name), &rec) + if err != nil { + return nil, err + } + if !found { + rec = ClusterRecord{ + Name: name, + ARN: ClusterARN(s.region, accountID, name), + Status: ClusterStatusActive, + CreatedAt: time.Now().UTC(), + } + if len(input.Tags) > 0 { + rec.Tags = map[string]string{} + for _, t := range input.Tags { + rec.Tags[aws.StringValue(t.Key)] = aws.StringValue(t.Value) + } + } + if err := putJSON(kv, ClusterMetaKey(name), &rec); err != nil { + return nil, err + } + } + return &ecs.CreateClusterOutput{Cluster: rec.toAWS()}, nil +} + +// DescribeClusters returns meta for the named clusters; unknown names are +// silently skipped (AWS returns them under "failures", omitted here in v1). +func (s *Service) DescribeClusters(input *ecs.DescribeClustersInput, accountID string) (*ecs.DescribeClustersOutput, error) { + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + names := awsStringSlice(input.Clusters) + if len(names) == 0 { + names = []string{defaultCluster} + } + out := &ecs.DescribeClustersOutput{} + for _, name := range names { + name = clusterShortName(name) + var rec ClusterRecord + found, err := getJSON(kv, ClusterMetaKey(name), &rec) + if err != nil { + return nil, err + } + if found { + out.Clusters = append(out.Clusters, rec.toAWS()) + } + } + return out, nil +} + +// ListClusters returns the ARNs of all clusters in the account. +func (s *Service) ListClusters(_ *ecs.ListClustersInput, accountID string) (*ecs.ListClustersOutput, error) { + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + keys, err := keysWithPrefix(kv, "clusters/") + if err != nil { + return nil, err + } + out := &ecs.ListClustersOutput{} + for _, k := range keys { + if !strings.HasSuffix(k, "/meta") { + continue + } + var rec ClusterRecord + found, err := getJSON(kv, k, &rec) + if err != nil { + return nil, err + } + if found { + out.ClusterArns = append(out.ClusterArns, aws.String(rec.ARN)) + } + } + return out, nil +} + +func (r *ClusterRecord) toAWS() *ecs.Cluster { + return &ecs.Cluster{ + ClusterName: aws.String(r.Name), + ClusterArn: aws.String(r.ARN), + Status: aws.String(r.Status), + } +} + +// --- Task definition --- + +// RegisterTaskDefinition stores a new revision of a family, bumping latest-rev. +func (s *Service) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput, accountID string) (*ecs.RegisterTaskDefinitionOutput, error) { + family := aws.StringValue(input.Family) + if family == "" { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + + rev, err := s.nextRevision(kv, family) + if err != nil { + return nil, err + } + + rec := TaskDefRecord{ + Family: family, + Revision: rev, + ARN: TaskDefARN(s.region, accountID, family, rev), + NetworkMode: aws.StringValue(input.NetworkMode), + CPU: aws.StringValue(input.Cpu), + Memory: aws.StringValue(input.Memory), + Status: TaskDefStatusActive, + RegisteredAt: time.Now().UTC(), + Containers: containerDefsFromAWS(input.ContainerDefinitions), + } + if err := putJSON(kv, TaskDefRevKey(family, rev), &rec); err != nil { + return nil, err + } + if err := putJSON(kv, TaskDefLatestRevKey(family), rev); err != nil { + return nil, err + } + return &ecs.RegisterTaskDefinitionOutput{TaskDefinition: rec.toAWS()}, nil +} + +// nextRevision reads the family's latest-rev and returns latest+1 (1 if absent). +func (s *Service) nextRevision(kv nats.KeyValue, family string) (int, error) { + var latest int + found, err := getJSON(kv, TaskDefLatestRevKey(family), &latest) + if err != nil { + return 0, err + } + if !found { + return 1, nil + } + return latest + 1, nil +} + +// DescribeTaskDefinition resolves "family", "family:rev" or an ARN to a revision. +func (s *Service) DescribeTaskDefinition(input *ecs.DescribeTaskDefinitionInput, accountID string) (*ecs.DescribeTaskDefinitionOutput, error) { + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + rec, err := s.resolveTaskDef(kv, aws.StringValue(input.TaskDefinition)) + if err != nil { + return nil, err + } + return &ecs.DescribeTaskDefinitionOutput{TaskDefinition: rec.toAWS()}, nil +} + +// ListTaskDefinitions returns all revision ARNs, optionally filtered by family. +func (s *Service) ListTaskDefinitions(input *ecs.ListTaskDefinitionsInput, accountID string) (*ecs.ListTaskDefinitionsOutput, error) { + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + prefix := TaskDefFamiliesPrefix() + if fam := aws.StringValue(input.FamilyPrefix); fam != "" { + prefix = TaskDefRevsPrefix(fam) + } + keys, err := keysWithPrefix(kv, prefix) + if err != nil { + return nil, err + } + out := &ecs.ListTaskDefinitionsOutput{} + for _, k := range keys { + if !strings.Contains(k, "/revs/") { + continue + } + var rec TaskDefRecord + found, err := getJSON(kv, k, &rec) + if err != nil { + return nil, err + } + if found { + out.TaskDefinitionArns = append(out.TaskDefinitionArns, aws.String(rec.ARN)) + } + } + return out, nil +} + +// resolveTaskDef loads the TaskDefRecord named by ref ("family", "family:rev", +// or a task-definition ARN). A bare family resolves to its latest revision. +func (s *Service) resolveTaskDef(kv nats.KeyValue, ref string) (*TaskDefRecord, error) { + family, rev := parseTaskDefRef(ref) + if family == "" { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + if rev == 0 { + var latest int + found, err := getJSON(kv, TaskDefLatestRevKey(family), &latest) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + rev = latest + } + var rec TaskDefRecord + found, err := getJSON(kv, TaskDefRevKey(family, rev), &rec) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + return &rec, nil +} + +// parseTaskDefRef splits "family", "family:rev" or an ARN into (family, rev). +// rev is 0 when unspecified (caller resolves to latest). +func parseTaskDefRef(ref string) (string, int) { + ref = strings.TrimSpace(ref) + if i := strings.LastIndex(ref, "task-definition/"); i >= 0 { + ref = ref[i+len("task-definition/"):] + } + family := ref + rev := 0 + if i := strings.LastIndexByte(ref, ':'); i >= 0 { + family = ref[:i] + if n, err := strconv.Atoi(ref[i+1:]); err == nil { + rev = n + } + } + return family, rev +} + +func (r *TaskDefRecord) toAWS() *ecs.TaskDefinition { + td := &ecs.TaskDefinition{ + Family: aws.String(r.Family), + Revision: aws.Int64(int64(r.Revision)), + TaskDefinitionArn: aws.String(r.ARN), + Status: aws.String(r.Status), + } + if r.NetworkMode != "" { + td.NetworkMode = aws.String(r.NetworkMode) + } + if r.CPU != "" { + td.Cpu = aws.String(r.CPU) + } + if r.Memory != "" { + td.Memory = aws.String(r.Memory) + } + for _, c := range r.Containers { + td.ContainerDefinitions = append(td.ContainerDefinitions, c.toAWS()) + } + return td +} diff --git a/spinifex/handlers/ecs/service_nats.go b/spinifex/handlers/ecs/service_nats.go new file mode 100644 index 000000000..f791063c0 --- /dev/null +++ b/spinifex/handlers/ecs/service_nats.go @@ -0,0 +1,80 @@ +package handlers_ecs + +import ( + "time" + + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +const defaultTimeout = 30 * time.Second + +// NATSECSService is the gateway-side adapter that forwards every ECSService +// method as a NATS request to the daemon's matching subscriber. +type NATSECSService struct { + natsConn *nats.Conn +} + +var _ ECSService = (*NATSECSService)(nil) + +// NewNATSECSService returns an ECSService that uses NATS request-response. +func NewNATSECSService(conn *nats.Conn) ECSService { + return &NATSECSService{natsConn: conn} +} + +// --- Cluster --- + +func (s *NATSECSService) CreateCluster(input *ecs.CreateClusterInput, accountID string) (*ecs.CreateClusterOutput, error) { + return utils.NATSRequest[ecs.CreateClusterOutput](s.natsConn, "ecs.CreateCluster", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) DescribeClusters(input *ecs.DescribeClustersInput, accountID string) (*ecs.DescribeClustersOutput, error) { + return utils.NATSRequest[ecs.DescribeClustersOutput](s.natsConn, "ecs.DescribeClusters", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) ListClusters(input *ecs.ListClustersInput, accountID string) (*ecs.ListClustersOutput, error) { + return utils.NATSRequest[ecs.ListClustersOutput](s.natsConn, "ecs.ListClusters", input, defaultTimeout, accountID) +} + +// --- Task definition --- + +func (s *NATSECSService) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput, accountID string) (*ecs.RegisterTaskDefinitionOutput, error) { + return utils.NATSRequest[ecs.RegisterTaskDefinitionOutput](s.natsConn, "ecs.RegisterTaskDefinition", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) DescribeTaskDefinition(input *ecs.DescribeTaskDefinitionInput, accountID string) (*ecs.DescribeTaskDefinitionOutput, error) { + return utils.NATSRequest[ecs.DescribeTaskDefinitionOutput](s.natsConn, "ecs.DescribeTaskDefinition", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) ListTaskDefinitions(input *ecs.ListTaskDefinitionsInput, accountID string) (*ecs.ListTaskDefinitionsOutput, error) { + return utils.NATSRequest[ecs.ListTaskDefinitionsOutput](s.natsConn, "ecs.ListTaskDefinitions", input, defaultTimeout, accountID) +} + +// --- Container instance --- + +func (s *NATSECSService) RegisterContainerInstance(input *ecs.RegisterContainerInstanceInput, accountID string) (*ecs.RegisterContainerInstanceOutput, error) { + return utils.NATSRequest[ecs.RegisterContainerInstanceOutput](s.natsConn, "ecs.RegisterContainerInstance", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) DescribeContainerInstances(input *ecs.DescribeContainerInstancesInput, accountID string) (*ecs.DescribeContainerInstancesOutput, error) { + return utils.NATSRequest[ecs.DescribeContainerInstancesOutput](s.natsConn, "ecs.DescribeContainerInstances", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) ListContainerInstances(input *ecs.ListContainerInstancesInput, accountID string) (*ecs.ListContainerInstancesOutput, error) { + return utils.NATSRequest[ecs.ListContainerInstancesOutput](s.natsConn, "ecs.ListContainerInstances", input, defaultTimeout, accountID) +} + +// --- Task --- + +func (s *NATSECSService) RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTaskOutput, error) { + return utils.NATSRequest[ecs.RunTaskOutput](s.natsConn, "ecs.RunTask", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) DescribeTasks(input *ecs.DescribeTasksInput, accountID string) (*ecs.DescribeTasksOutput, error) { + return utils.NATSRequest[ecs.DescribeTasksOutput](s.natsConn, "ecs.DescribeTasks", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) ListTasks(input *ecs.ListTasksInput, accountID string) (*ecs.ListTasksOutput, error) { + return utils.NATSRequest[ecs.ListTasksOutput](s.natsConn, "ecs.ListTasks", input, defaultTimeout, accountID) +} diff --git a/spinifex/handlers/ecs/service_test.go b/spinifex/handlers/ecs/service_test.go new file mode 100644 index 000000000..1daebe0cb --- /dev/null +++ b/spinifex/handlers/ecs/service_test.go @@ -0,0 +1,292 @@ +package handlers_ecs + +import ( + "encoding/json" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testRegion = "ap-southeast-2" + +func newTestService(t *testing.T) (*Service, *nats.Conn) { + t.Helper() + _, nc, _ := testutil.StartTestJetStream(t) + return NewService(nc, testRegion, "internal"), nc +} + +func TestService_CreateCluster_Idempotent(t *testing.T) { + svc, _ := newTestService(t) + out, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + assert.Equal(t, "ACTIVE", aws.StringValue(out.Cluster.Status)) + assert.Equal(t, ClusterARN(testRegion, testAccountID, "web"), aws.StringValue(out.Cluster.ClusterArn)) + + // Recreate returns the same record, not a duplicate. + _, err = svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + + list, err := svc.ListClusters(&ecs.ListClustersInput{}, testAccountID) + require.NoError(t, err) + assert.Len(t, list.ClusterArns, 1) +} + +func TestService_DescribeClusters_DefaultAndMissing(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.CreateCluster(&ecs.CreateClusterInput{}, testAccountID) // implicit "default" + require.NoError(t, err) + + out, err := svc.DescribeClusters(&ecs.DescribeClustersInput{}, testAccountID) + require.NoError(t, err) + require.Len(t, out.Clusters, 1) + assert.Equal(t, defaultCluster, aws.StringValue(out.Clusters[0].ClusterName)) + + miss, err := svc.DescribeClusters(&ecs.DescribeClustersInput{Clusters: []*string{aws.String("nope")}}, testAccountID) + require.NoError(t, err) + assert.Empty(t, miss.Clusters) +} + +func registerTaskDef(t *testing.T, svc *Service, family string, cpu, mem int) *ecs.RegisterTaskDefinitionOutput { + t.Helper() + out, err := svc.RegisterTaskDefinition(&ecs.RegisterTaskDefinitionInput{ + Family: aws.String(family), + ContainerDefinitions: []*ecs.ContainerDefinition{{ + Name: aws.String("app"), + Image: aws.String("registry/app:1"), + Cpu: aws.Int64(int64(cpu)), + Memory: aws.Int64(int64(mem)), + Essential: aws.Bool(true), + }}, + }, testAccountID) + require.NoError(t, err) + return out +} + +func TestService_RegisterTaskDefinition_RevisionBump(t *testing.T) { + svc, _ := newTestService(t) + r1 := registerTaskDef(t, svc, "nginx", 128, 256) + assert.Equal(t, int64(1), aws.Int64Value(r1.TaskDefinition.Revision)) + r2 := registerTaskDef(t, svc, "nginx", 128, 256) + assert.Equal(t, int64(2), aws.Int64Value(r2.TaskDefinition.Revision)) + + // Bare family resolves to latest. + d, err := svc.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{TaskDefinition: aws.String("nginx")}, testAccountID) + require.NoError(t, err) + assert.Equal(t, int64(2), aws.Int64Value(d.TaskDefinition.Revision)) + + // family:rev resolves the pinned revision. + d1, err := svc.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{TaskDefinition: aws.String("nginx:1")}, testAccountID) + require.NoError(t, err) + assert.Equal(t, int64(1), aws.Int64Value(d1.TaskDefinition.Revision)) + + list, err := svc.ListTaskDefinitions(&ecs.ListTaskDefinitionsInput{}, testAccountID) + require.NoError(t, err) + assert.Len(t, list.TaskDefinitionArns, 2) +} + +func TestService_RegisterTaskDefinition_NoFamily(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.RegisterTaskDefinition(&ecs.RegisterTaskDefinitionInput{}, testAccountID) + require.Error(t, err) +} + +func TestService_DescribeTaskDefinition_Unknown(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{TaskDefinition: aws.String("ghost")}, testAccountID) + require.Error(t, err) +} + +// registerInstance seeds an ACTIVE container instance with capacity via the bus +// register path (the way the agent registers in production). +func registerInstance(t *testing.T, svc *Service, cluster, id string, cpu, mem int) { + t.Helper() + require.NoError(t, svc.recordRegister(&bus.RegisterInstance{ + AccountID: testAccountID, + ClusterName: cluster, + InstanceID: id, + Capacity: bus.InstanceCapacity{CPU: cpu, MemoryMiB: mem}, + })) +} + +func TestService_RunTask_PlacesAndAssigns(t *testing.T) { + svc, nc := newTestService(t) + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerTaskDef(t, svc, "app", 128, 256) + registerInstance(t, svc, "web", "i-1", 1024, 2048) + + sub, err := nc.SubscribeSync(bus.AssignSubject(testAccountID, "web", "i-1")) + require.NoError(t, err) + + out, err := svc.RunTask(&ecs.RunTaskInput{ + Cluster: aws.String("web"), + TaskDefinition: aws.String("app"), + Count: aws.Int64(1), + }, testAccountID) + require.NoError(t, err) + require.Len(t, out.Tasks, 1) + assert.Empty(t, out.Failures) + assert.Equal(t, "PENDING", aws.StringValue(out.Tasks[0].LastStatus)) + + // Assign published to the instance's agent. + msg, err := sub.NextMsg(2 * time.Second) + require.NoError(t, err) + var as bus.Assign + require.NoError(t, json.Unmarshal(msg.Data, &as)) + assert.Equal(t, "i-1", as.InstanceID) + require.Len(t, as.Containers, 1) + assert.Equal(t, "registry/app:1", as.Containers[0].Image) + + // Capacity reserved on the instance. + di, err := svc.DescribeContainerInstances(&ecs.DescribeContainerInstancesInput{ + Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")}, + }, testAccountID) + require.NoError(t, err) + require.Len(t, di.ContainerInstances, 1) + assert.Equal(t, int64(1), aws.Int64Value(di.ContainerInstances[0].RunningTasksCount)) + + // Task visible via Describe/List. + lt, err := svc.ListTasks(&ecs.ListTasksInput{Cluster: aws.String("web")}, testAccountID) + require.NoError(t, err) + assert.Len(t, lt.TaskArns, 1) + assert.Equal(t, as.TaskID, containerInstanceShortID(aws.StringValue(out.Tasks[0].TaskArn))) +} + +func TestService_RunTask_ClusterNotFound(t *testing.T) { + svc, _ := newTestService(t) + registerTaskDef(t, svc, "app", 1, 1) + _, err := svc.RunTask(&ecs.RunTaskInput{Cluster: aws.String("ghost"), TaskDefinition: aws.String("app")}, testAccountID) + require.Error(t, err) +} + +func TestService_RunTask_NoCapacityFails(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerTaskDef(t, svc, "big", 100, 100000) // 100GB memory, won't fit + registerInstance(t, svc, "web", "i-1", 1024, 2048) + + out, err := svc.RunTask(&ecs.RunTaskInput{ + Cluster: aws.String("web"), TaskDefinition: aws.String("big"), Count: aws.Int64(1), + }, testAccountID) + require.NoError(t, err) + assert.Empty(t, out.Tasks) + require.Len(t, out.Failures, 1) +} + +// Task-state RUNNING then STOPPED updates the task and releases capacity. +func TestService_RecordTaskState_ReleasesCapacityOnStop(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerTaskDef(t, svc, "app", 128, 256) + registerInstance(t, svc, "web", "i-1", 1024, 2048) + out, err := svc.RunTask(&ecs.RunTaskInput{Cluster: aws.String("web"), TaskDefinition: aws.String("app")}, testAccountID) + require.NoError(t, err) + taskID := containerInstanceShortID(aws.StringValue(out.Tasks[0].TaskArn)) + + require.NoError(t, svc.recordTaskState(&bus.TaskState{ + AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: taskID, + LastStatus: bus.TaskStatusRunning, + })) + exit := 0 + require.NoError(t, svc.recordTaskState(&bus.TaskState{ + AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: taskID, + LastStatus: bus.TaskStatusStopped, Reason: "exited", + Containers: []bus.ContainerStatus{{Name: "app", Status: bus.TaskStatusStopped, ExitCode: &exit}}, + })) + + dt, err := svc.DescribeTasks(&ecs.DescribeTasksInput{Cluster: aws.String("web"), Tasks: []*string{aws.String(taskID)}}, testAccountID) + require.NoError(t, err) + require.Len(t, dt.Tasks, 1) + assert.Equal(t, "STOPPED", aws.StringValue(dt.Tasks[0].LastStatus)) + + // Capacity fully released: remaining == total. + di, err := svc.DescribeContainerInstances(&ecs.DescribeContainerInstancesInput{ + Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")}, + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, int64(0), aws.Int64Value(di.ContainerInstances[0].RunningTasksCount)) + for _, r := range di.ContainerInstances[0].RemainingResources { + switch aws.StringValue(r.Name) { + case "CPU": + assert.Equal(t, int64(1024), aws.Int64Value(r.IntegerValue)) + case "MEMORY": + assert.Equal(t, int64(2048), aws.Int64Value(r.IntegerValue)) + } + } +} + +func TestService_RecordHeartbeat_UpdatesStatus(t *testing.T) { + svc, _ := newTestService(t) + registerInstance(t, svc, "web", "i-1", 1024, 2048) + require.NoError(t, svc.recordHeartbeat(&bus.Heartbeat{ + AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", Status: bus.StatusDraining, + })) + di, err := svc.DescribeContainerInstances(&ecs.DescribeContainerInstancesInput{ + Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")}, + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, "DRAINING", aws.StringValue(di.ContainerInstances[0].Status)) +} + +// The reaper marks a stale instance DRAINING and stops its tasks. +func TestScheduler_ReapBucket_StopsStaleInstanceTasks(t *testing.T) { + svc, nc := newTestService(t) + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerTaskDef(t, svc, "app", 128, 256) + registerInstance(t, svc, "web", "i-1", 1024, 2048) + out, err := svc.RunTask(&ecs.RunTaskInput{Cluster: aws.String("web"), TaskDefinition: aws.String("app")}, testAccountID) + require.NoError(t, err) + taskID := containerInstanceShortID(aws.StringValue(out.Tasks[0].TaskArn)) + + // Backdate LastSeen beyond the heartbeat timeout. + kv, err := svc.bucket(testAccountID) + require.NoError(t, err) + var rec InstanceRecord + _, err = getJSON(kv, InstanceKey("web", "i-1"), &rec) + require.NoError(t, err) + rec.LastSeen = time.Now().UTC().Add(-2 * heartbeatTimeout) + require.NoError(t, putJSON(kv, InstanceKey("web", "i-1"), &rec)) + + sc := NewScheduler(nc, svc, "test-holder") + sc.reapBucket(kv, testAccountID, time.Now().UTC()) + + di, err := svc.DescribeContainerInstances(&ecs.DescribeContainerInstancesInput{ + Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")}, + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, "DRAINING", aws.StringValue(di.ContainerInstances[0].Status)) + + dt, err := svc.DescribeTasks(&ecs.DescribeTasksInput{Cluster: aws.String("web"), Tasks: []*string{aws.String(taskID)}}, testAccountID) + require.NoError(t, err) + assert.Equal(t, "STOPPED", aws.StringValue(dt.Tasks[0].LastStatus)) + assert.Equal(t, stoppedReasonReaped, aws.StringValue(dt.Tasks[0].StoppedReason)) +} + +// Leader election: first holder wins Create; a second holder is rejected. +func TestScheduler_AcquireLease_SingleLeader(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + svc := NewService(nc, testRegion, "") + a := NewScheduler(nc, svc, "holder-a") + b := NewScheduler(nc, svc, "holder-b") + assert.True(t, a.acquireOrRefresh()) + assert.False(t, b.acquireOrRefresh()) + assert.True(t, a.acquireOrRefresh()) // refresh keeps leadership +} + +func TestAccountIDFromBucket(t *testing.T) { + id, ok := accountIDFromBucket(AccountBucketName(testAccountID)) + assert.True(t, ok) + assert.Equal(t, testAccountID, id) + _, ok = accountIDFromBucket("some-other-bucket") + assert.False(t, ok) +} diff --git a/spinifex/handlers/ecs/tasks.go b/spinifex/handlers/ecs/tasks.go new file mode 100644 index 000000000..177467dac --- /dev/null +++ b/spinifex/handlers/ecs/tasks.go @@ -0,0 +1,244 @@ +package handlers_ecs + +import ( + "encoding/json" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/aws/aws-sdk-go/aws" + "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" +) + +// reservePlacementRetries bounds the CAS retry loop when concurrent RunTask calls +// contend for the same instance's capacity. +const reservePlacementRetries = 5 + +// RunTask places input.Count tasks of a definition onto container instances via +// bin-pack, reserves their capacity in KV, writes PENDING task records, and +// publishes an assign on the Layer-2 bus for each. Placement failures for a task +// are returned as RunTask failures; already-placed tasks in the same call stay. +func (s *Service) RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTaskOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + + var clusterRec ClusterRecord + found, err := getJSON(kv, ClusterMetaKey(cluster), &clusterRec) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSClusterNotFound) + } + + taskDef, err := s.resolveTaskDef(kv, aws.StringValue(input.TaskDefinition)) + if err != nil { + return nil, err + } + + count := int(aws.Int64Value(input.Count)) + if count <= 0 { + count = 1 + } + strategy := placementStrategyFromAWS(input.PlacementStrategy) + cpu, mem := taskDef.reservedCPU(), taskDef.reservedMemory() + + out := &ecs.RunTaskOutput{} + for i := 0; i < count; i++ { + taskID := uuid.NewString() + inst, err := s.reservePlacement(kv, cluster, taskID, cpu, mem, strategy) + if err != nil { + out.Failures = append(out.Failures, &ecs.Failure{ + Reason: aws.String("RESOURCE:placement"), + Detail: aws.String(err.Error()), + }) + continue + } + + rec := s.newTaskRecord(accountID, cluster, taskID, taskDef, inst, cpu, mem) + if err := putJSON(kv, TaskKey(cluster, taskID), rec); err != nil { + return nil, err + } + if err := s.publishAssign(accountID, cluster, inst.InstanceID, rec, taskDef); err != nil { + slog.Error("ECS RunTask: failed to publish assign", "task", taskID, "instance", inst.InstanceID, "err", err) + } + out.Tasks = append(out.Tasks, s.taskToAWS(accountID, rec)) + } + return out, nil +} + +// reservePlacement bin-packs a task onto an ACTIVE instance and commits the +// reservation under a KV CAS, retrying on contention. +func (s *Service) reservePlacement(kv nats.KeyValue, cluster, taskID string, cpu, mem int, strategy string) (*InstanceRecord, error) { + for range reservePlacementRetries { + instances, err := s.listInstanceRecords(kv, cluster) + if err != nil { + return nil, err + } + chosen, err := placeTask(instances, cpu, mem, strategy) + if err != nil { + return nil, err + } + + // Re-read the chosen instance for its current revision, then CAS-update. + entry, err := kv.Get(InstanceKey(cluster, chosen.InstanceID)) + if err != nil { + continue + } + var live InstanceRecord + if uerr := json.Unmarshal(entry.Value(), &live); uerr != nil { + return nil, uerr + } + if !live.fits(cpu, mem) { + continue + } + live.ReservedCPU += cpu + live.ReservedMemoryMiB += mem + live.PlacedTasks = append(live.PlacedTasks, taskID) + data, merr := json.Marshal(&live) + if merr != nil { + return nil, merr + } + if _, uerr := kv.Update(InstanceKey(cluster, live.InstanceID), data, entry.Revision()); uerr != nil { + continue // lost the CAS race; retry placement + } + return &live, nil + } + return nil, fmt.Errorf("placement contended after %d attempts", reservePlacementRetries) +} + +// newTaskRecord builds a PENDING task record for a placed task. +func (s *Service) newTaskRecord(accountID, cluster, taskID string, td *TaskDefRecord, inst *InstanceRecord, cpu, mem int) *TaskRecord { + now := time.Now().UTC() + rec := &TaskRecord{ + TaskID: taskID, + ARN: TaskARN(s.region, accountID, cluster, taskID), + Cluster: cluster, + TaskDefFamily: td.Family, + TaskDefRevision: td.Revision, + TaskDefARN: td.ARN, + ContainerInstanceID: inst.InstanceID, + ContainerInstanceARN: inst.ARN, + DesiredStatus: TaskStatusRunning, + LastStatus: TaskStatusPending, + ReservedCPU: cpu, + ReservedMemoryMiB: mem, + CreatedAt: now, + } + for _, c := range td.Containers { + rec.Containers = append(rec.Containers, ContainerState{Name: c.Name, Status: TaskStatusPending}) + } + return rec +} + +// publishAssign sends the task assignment to the instance's agent over the bus. +func (s *Service) publishAssign(accountID, cluster, instanceID string, rec *TaskRecord, td *TaskDefRecord) error { + msg := bus.Assign{ + AccountID: accountID, + ClusterName: cluster, + InstanceID: instanceID, + TaskID: rec.TaskID, + TaskARN: rec.ARN, + TaskDefFamily: td.Family, + TaskDefRevision: td.Revision, + AssignedAt: time.Now().UTC(), + } + for _, c := range td.Containers { + msg.Containers = append(msg.Containers, c.toAssignContainer()) + } + data, err := json.Marshal(&msg) + if err != nil { + return err + } + subj := bus.AssignSubject(accountID, cluster, instanceID) + return s.nc.Publish(subj, data) +} + +// DescribeTasks returns task records for the named tasks in a cluster. +func (s *Service) DescribeTasks(input *ecs.DescribeTasksInput, accountID string) (*ecs.DescribeTasksOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + out := &ecs.DescribeTasksOutput{} + for _, ref := range awsStringSlice(input.Tasks) { + taskID := containerInstanceShortID(ref) + var rec TaskRecord + found, err := getJSON(kv, TaskKey(cluster, taskID), &rec) + if err != nil { + return nil, err + } + if found { + out.Tasks = append(out.Tasks, s.taskToAWS(accountID, &rec)) + } else { + out.Failures = append(out.Failures, &ecs.Failure{Arn: aws.String(ref), Reason: aws.String("MISSING")}) + } + } + return out, nil +} + +// ListTasks returns the ARNs of all tasks in a cluster. +func (s *Service) ListTasks(input *ecs.ListTasksInput, accountID string) (*ecs.ListTasksOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + keys, err := keysWithPrefix(kv, TasksPrefix(cluster)) + if err != nil { + return nil, err + } + out := &ecs.ListTasksOutput{} + for _, k := range keys { + var rec TaskRecord + found, err := getJSON(kv, k, &rec) + if err != nil { + return nil, err + } + if found { + out.TaskArns = append(out.TaskArns, aws.String(rec.ARN)) + } + } + return out, nil +} + +func (s *Service) taskToAWS(accountID string, r *TaskRecord) *ecs.Task { + t := &ecs.Task{ + TaskArn: aws.String(r.ARN), + ClusterArn: aws.String(ClusterARN(s.region, accountID, r.Cluster)), + TaskDefinitionArn: aws.String(r.TaskDefARN), + DesiredStatus: aws.String(r.DesiredStatus), + LastStatus: aws.String(r.LastStatus), + ContainerInstanceArn: aws.String(r.ContainerInstanceARN), + } + if r.StoppedReason != "" { + t.StoppedReason = aws.String(r.StoppedReason) + } + for _, c := range r.Containers { + ctr := &ecs.Container{Name: aws.String(c.Name), LastStatus: aws.String(c.Status)} + if c.ExitCode != nil { + ctr.ExitCode = aws.Int64(int64(*c.ExitCode)) + } + t.Containers = append(t.Containers, ctr) + } + return t +} + +// placementStrategyFromAWS extracts the first strategy type, defaulting binpack. +func placementStrategyFromAWS(in []*ecs.PlacementStrategy) string { + for _, p := range in { + if p != nil { + return aws.StringValue(p.Type) + } + } + return StrategyBinpack +}