From 9c76df33737c7f54265e21fc99f153bef4550895 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 25 Jun 2026 16:01:14 +1000 Subject: [PATCH] =?UTF-8?q?feat(ecs):=20awsvpc=20mode=20=E2=80=94=20ENI-pe?= =?UTF-8?q?r-task=20allocation=20+=20guest=20datapath=20(Sprint=204d)=20(#?= =?UTF-8?q?418)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(ecs): awsvpc ENI-per-task control plane (Sprint 4d) Scheduler owns the task-ENI lifecycle for awsvpc mode: allocate from the task subnet on placement, hot-plug onto the container instance, plumb ENI identity through the assign payload + task record, and detach+delete on STOPPED (graceful + reaper, idempotent leak guard). bridge/host unchanged. * feat(ecs-agent): awsvpc guest-side task netns from hot-plugged ENI The agent builds a per-task network namespace for awsvpc tasks: it finds the hot-plugged ENI NIC by MAC, moves it into /var/run/netns/ecs-, and DHCPs the OVN-assigned ENI IP. The container then joins that netns via the OCI network namespace path; bridge/host tasks keep the host (VM) netns as before. * build(ecs-node): add iproute2 for awsvpc task netns The ecs-agent builds awsvpc task network namespaces with `ip netns`, which busybox's `ip` applet does not implement. Install iproute2 in the ecs-node AMI; udhcpc (for the in-netns ENI DHCP lease) is already provided by busybox. --- cmd/ecs-agent/agent.go | 4 + cmd/ecs-agent/netns.go | 125 +++++++++ cmd/ecs-agent/netns_test.go | 131 +++++++++ cmd/ecs-agent/runtime/containerd_run.go | 11 +- cmd/ecs-agent/runtime/runtime.go | 4 + cmd/ecs-agent/taskrun.go | 40 ++- cmd/ecs-agent/taskrun_test.go | 40 +++ scripts/images/ecs-agent/manifest.conf | 5 +- spinifex/handlers/ecs/bus/messages.go | 7 + spinifex/handlers/ecs/eni.go | 152 ++++++++++ spinifex/handlers/ecs/eni_test.go | 359 ++++++++++++++++++++++++ spinifex/handlers/ecs/instances.go | 3 +- spinifex/handlers/ecs/netcfg.go | 78 +++++ spinifex/handlers/ecs/records.go | 16 +- spinifex/handlers/ecs/scheduler.go | 10 +- spinifex/handlers/ecs/service.go | 5 +- spinifex/handlers/ecs/tasks.go | 62 ++++ 17 files changed, 1036 insertions(+), 16 deletions(-) create mode 100644 cmd/ecs-agent/netns.go create mode 100644 cmd/ecs-agent/netns_test.go create mode 100644 spinifex/handlers/ecs/eni.go create mode 100644 spinifex/handlers/ecs/eni_test.go create mode 100644 spinifex/handlers/ecs/netcfg.go diff --git a/cmd/ecs-agent/agent.go b/cmd/ecs-agent/agent.go index 1e8fe0402..b95b9e70f 100644 --- a/cmd/ecs-agent/agent.go +++ b/cmd/ecs-agent/agent.go @@ -38,6 +38,9 @@ type Agent struct { reg *registrar hb *heartbeater + // netns builds awsvpc task network namespaces (nil-safe; bridge/host skip it). + netns *taskNetns + nc *nats.Conn closers []func() error } @@ -56,6 +59,7 @@ func newAgent(cfg config, id identity, pub publisher, puller ctrruntime.ImagePul resolver: resolver, reg: newRegistrar(pub, id), hb: newHeartbeater(pub, id, cfg.Heartbeat, nil), + netns: newTaskNetns(execNetRunner{}), } } diff --git a/cmd/ecs-agent/netns.go b/cmd/ecs-agent/netns.go new file mode 100644 index 000000000..17c0cadf4 --- /dev/null +++ b/cmd/ecs-agent/netns.go @@ -0,0 +1,125 @@ +package main + +import ( + "fmt" + "os/exec" + "strings" + "time" +) + +// netnsRunDir is the iproute2 netns mount directory; a name here is the netns +// path containerd joins via the OCI network namespace path. +const netnsRunDir = "/var/run/netns" + +// netCmdRunner executes a host networking command. Abstracted so unit tests can +// assert the ip/netns sequence without touching the kernel. +type netCmdRunner interface { + Run(name string, args ...string) (string, error) +} + +// execNetRunner runs real commands. +type execNetRunner struct{} + +func (execNetRunner) Run(name string, args ...string) (string, error) { + out, err := exec.Command(name, args...).CombinedOutput() + return string(out), err +} + +// taskNetns builds and tears down per-task awsvpc network namespaces. The ENI is +// already hot-plugged into the VM (scheduler did that); the agent finds the NIC +// by MAC, moves it into a task netns, and DHCPs the OVN-assigned ENI IP — the +// same way the primary NIC obtains its address. +type taskNetns struct { + run netCmdRunner + nicWait time.Duration + poll time.Duration +} + +func newTaskNetns(run netCmdRunner) *taskNetns { + return &taskNetns{run: run, nicWait: 30 * time.Second, poll: time.Second} +} + +func netnsName(taskID string) string { return "ecs-" + taskID } +func netnsPathFor(taskID string) string { + return netnsRunDir + "/" + netnsName(taskID) +} + +// Setup creates the task netns, moves the ENI NIC (matched by MAC) into it, and +// brings it up with a DHCP lease. It returns the netns path for containerd. On +// any failure it tears the partial netns down so a retry starts clean. +func (n *taskNetns) Setup(taskID, mac string) (string, error) { + iface, err := n.waitForNIC(mac) + if err != nil { + return "", err + } + name := netnsName(taskID) + if _, err := n.run.Run("ip", "netns", "add", name); err != nil { + return "", fmt.Errorf("netns add %s: %w", name, err) + } + steps := [][]string{ + {"ip", "link", "set", iface, "netns", name}, + {"ip", "-n", name, "link", "set", "lo", "up"}, + {"ip", "-n", name, "link", "set", iface, "up"}, + {"ip", "netns", "exec", name, "udhcpc", "-i", iface, "-q", "-n"}, + } + for _, s := range steps { + if _, err := n.run.Run(s[0], s[1:]...); err != nil { + _ = n.Teardown(taskID) + return "", fmt.Errorf("netns setup %v: %w", s, err) + } + } + return netnsPathFor(taskID), nil +} + +// Teardown deletes the task netns (releasing the moved NIC). Idempotent enough +// for the stop path: a missing netns is reported but not fatal to the caller. +func (n *taskNetns) Teardown(taskID string) error { + name := netnsName(taskID) + if _, err := n.run.Run("ip", "netns", "del", name); err != nil { + return fmt.Errorf("netns del %s: %w", name, err) + } + return nil +} + +// waitForNIC polls until a NIC with the given MAC appears in the host netns, +// returning its interface name. The hot-plug pipeline is async, so the device +// may lag the assign by a few seconds. +func (n *taskNetns) waitForNIC(mac string) (string, error) { + want := strings.ToLower(strings.TrimSpace(mac)) + deadline := time.Now().Add(n.nicWait) + for { + iface, err := n.findNICByMAC(want) + if err == nil { + return iface, nil + } + if time.Now().After(deadline) { + return "", fmt.Errorf("nic with mac %s not found within %s", want, n.nicWait) + } + time.Sleep(n.poll) + } +} + +// findNICByMAC parses `ip -o link` and returns the interface whose link/ether +// matches mac. +func (n *taskNetns) findNICByMAC(mac string) (string, error) { + out, err := n.run.Run("ip", "-o", "link") + if err != nil { + return "", fmt.Errorf("ip -o link: %w", err) + } + for line := range strings.SplitSeq(out, "\n") { + if !strings.Contains(strings.ToLower(line), mac) { + continue + } + // ": : ... link/ether ..." + fields := strings.Fields(line) + if len(fields) < 2 { + continue + } + name := strings.TrimSuffix(strings.TrimSuffix(fields[1], ":"), "@") + if i := strings.IndexByte(name, '@'); i >= 0 { + name = name[:i] + } + return name, nil + } + return "", fmt.Errorf("no interface with mac %s", mac) +} diff --git a/cmd/ecs-agent/netns_test.go b/cmd/ecs-agent/netns_test.go new file mode 100644 index 000000000..cb4a25019 --- /dev/null +++ b/cmd/ecs-agent/netns_test.go @@ -0,0 +1,131 @@ +package main + +import ( + "errors" + "strings" + "sync" + "testing" + "time" +) + +// fakeNetRunner records every command and serves canned `ip -o link` output. A +// failOn substring forces the matching command to error so teardown paths run. +type fakeNetRunner struct { + mu sync.Mutex + calls [][]string + linkOut string + failOn string +} + +func (f *fakeNetRunner) Run(name string, args ...string) (string, error) { + f.mu.Lock() + defer f.mu.Unlock() + cmd := append([]string{name}, args...) + f.calls = append(f.calls, cmd) + joined := strings.Join(cmd, " ") + if f.failOn != "" && strings.Contains(joined, f.failOn) { + return "", errors.New("boom: " + joined) + } + if name == "ip" && len(args) >= 2 && args[0] == "-o" && args[1] == "link" { + return f.linkOut, nil + } + return "", nil +} + +func (f *fakeNetRunner) joined() []string { + f.mu.Lock() + defer f.mu.Unlock() + out := make([]string, len(f.calls)) + for i, c := range f.calls { + out[i] = strings.Join(c, " ") + } + return out +} + +func (f *fakeNetRunner) sawAny(sub string) bool { + for _, j := range f.joined() { + if strings.Contains(j, sub) { + return true + } + } + return false +} + +const linkWithENI = `1: lo: mtu 65536 ... link/loopback 00:00:00:00:00:00 ... +2: eth0: mtu 1500 ... link/ether 52:54:00:aa:bb:cc ... +3: eth1: mtu 1500 ... link/ether 52:54:00:de:ad:01 ...` + +func newTestNetns(f *fakeNetRunner) *taskNetns { + n := newTaskNetns(f) + n.nicWait = 50 * time.Millisecond + n.poll = 5 * time.Millisecond + return n +} + +func TestNetns_FindNICByMAC(t *testing.T) { + n := newTestNetns(&fakeNetRunner{linkOut: linkWithENI}) + iface, err := n.findNICByMAC("52:54:00:de:ad:01") + if err != nil || iface != "eth1" { + t.Fatalf("want eth1, got %q err=%v", iface, err) + } + if _, err := n.findNICByMAC("52:54:00:ff:ff:ff"); err == nil { + t.Fatal("want error for absent mac") + } +} + +func TestNetns_SetupSequence(t *testing.T) { + f := &fakeNetRunner{linkOut: linkWithENI} + n := newTestNetns(f) + path, err := n.Setup("t-001", "52:54:00:DE:AD:01") + if err != nil { + t.Fatalf("setup: %v", err) + } + if path != netnsPathFor("t-001") { + t.Fatalf("want %q, got %q", netnsPathFor("t-001"), path) + } + want := []string{ + "ip netns add ecs-t-001", + "ip link set eth1 netns ecs-t-001", + "ip -n ecs-t-001 link set lo up", + "ip -n ecs-t-001 link set eth1 up", + "ip netns exec ecs-t-001 udhcpc -i eth1 -q -n", + } + for _, w := range want { + if !f.sawAny(w) { + t.Errorf("missing command %q in %v", w, f.joined()) + } + } +} + +func TestNetns_SetupNICNotFound(t *testing.T) { + f := &fakeNetRunner{linkOut: linkWithENI} + n := newTestNetns(f) + if _, err := n.Setup("t-001", "52:54:00:00:00:99"); err == nil { + t.Fatal("want timeout error for absent NIC") + } + if f.sawAny("netns add") { + t.Error("should not create netns when NIC never appears") + } +} + +func TestNetns_SetupTearsDownOnStepFailure(t *testing.T) { + f := &fakeNetRunner{linkOut: linkWithENI, failOn: "udhcpc"} + n := newTestNetns(f) + if _, err := n.Setup("t-001", "52:54:00:de:ad:01"); err == nil { + t.Fatal("want error when udhcpc fails") + } + if !f.sawAny("ip netns del ecs-t-001") { + t.Errorf("expected teardown after step failure, got %v", f.joined()) + } +} + +func TestNetns_Teardown(t *testing.T) { + f := &fakeNetRunner{} + n := newTestNetns(f) + if err := n.Teardown("t-001"); err != nil { + t.Fatalf("teardown: %v", err) + } + if !f.sawAny("ip netns del ecs-t-001") { + t.Errorf("expected netns del, got %v", f.joined()) + } +} diff --git a/cmd/ecs-agent/runtime/containerd_run.go b/cmd/ecs-agent/runtime/containerd_run.go index 42f297e8c..d99f9d6ca 100644 --- a/cmd/ecs-agent/runtime/containerd_run.go +++ b/cmd/ecs-agent/runtime/containerd_run.go @@ -28,10 +28,19 @@ func (p *containerdPuller) Run(ctx context.Context, id string, spec RunSpec) (st specOpts := []oci.SpecOpts{ oci.WithImageConfig(image), - oci.WithHostNamespace(specs.NetworkNamespace), oci.WithHostHostsFile, oci.WithHostResolvconf, } + if spec.NetnsPath != "" { + // awsvpc: join the task ENI netns built by the agent. + specOpts = append(specOpts, oci.WithLinuxNamespace(specs.LinuxNamespace{ + Type: specs.NetworkNamespace, + Path: spec.NetnsPath, + })) + } else { + // bridge/host: share the VM (host) netns. + specOpts = append(specOpts, oci.WithHostNamespace(specs.NetworkNamespace)) + } if len(spec.Command) > 0 { specOpts = append(specOpts, oci.WithProcessArgs(spec.Command...)) } diff --git a/cmd/ecs-agent/runtime/runtime.go b/cmd/ecs-agent/runtime/runtime.go index 87075f650..f27a4a7df 100644 --- a/cmd/ecs-agent/runtime/runtime.go +++ b/cmd/ecs-agent/runtime/runtime.go @@ -44,6 +44,10 @@ type RunSpec struct { Command []string Env map[string]string Labels map[string]string + // NetnsPath, when set, joins the container to a pre-built network namespace + // (awsvpc task ENI). Empty keeps the container in the host (VM) netns — + // bridge/host mode behaviour. + NetnsPath string } // RunStatus is a finished container's outcome. diff --git a/cmd/ecs-agent/taskrun.go b/cmd/ecs-agent/taskrun.go index d3f3c06f4..d8d4e3ef2 100644 --- a/cmd/ecs-agent/taskrun.go +++ b/cmd/ecs-agent/taskrun.go @@ -47,24 +47,35 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) { return } + // awsvpc: build the task netns from the hot-plugged ENI before any container. + netnsPath, err := a.setupTaskNetns(as) + if err != nil { + slog.Error("ecs-agent: task netns setup failed", "task", as.TaskID, "err", err) + a.reportTaskState(as, bus.TaskStatusStopped, "network setup failed: "+err.Error(), 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.teardownTaskNetns(as) 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), + Image: c.Image, + Command: c.Command, + Env: c.Environment, + Labels: taskLabels(as, c.Name), + NetnsPath: netnsPath, } 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.teardownTaskNetns(as) a.reportTaskState(as, bus.TaskStatusStopped, "container start failed: "+err.Error(), statuses) return } @@ -85,11 +96,32 @@ func (a *Agent) waitContainer(ctx context.Context, as *bus.Assign, name, contain return } exit := status.ExitCode + a.teardownTaskNetns(as) 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) } +// setupTaskNetns builds the awsvpc task netns when the assign carries an ENI MAC. +// Returns an empty path (host netns) for bridge/host tasks or when no netns +// manager is configured. +func (a *Agent) setupTaskNetns(as *bus.Assign) (string, error) { + if as.ENIMacAddress == "" || a.netns == nil { + return "", nil + } + return a.netns.Setup(as.TaskID, as.ENIMacAddress) +} + +// teardownTaskNetns removes the awsvpc task netns; a no-op for bridge/host tasks. +func (a *Agent) teardownTaskNetns(as *bus.Assign) { + if as.ENIMacAddress == "" || a.netns == nil { + return + } + if err := a.netns.Teardown(as.TaskID); err != nil { + slog.Warn("ecs-agent: task netns teardown", "task", as.TaskID, "err", err) + } +} + // 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 { diff --git a/cmd/ecs-agent/taskrun_test.go b/cmd/ecs-agent/taskrun_test.go index 72d1c5a34..fda0c0f88 100644 --- a/cmd/ecs-agent/taskrun_test.go +++ b/cmd/ecs-agent/taskrun_test.go @@ -155,6 +155,46 @@ func TestRunTask_ExitReportsStoppedWithCode(t *testing.T) { } } +// awsvpc: an assign carrying an ENI MAC builds the task netns and passes its +// path to the container RunSpec. +func TestRunTask_AwsvpcBuildsNetns(t *testing.T) { + pub := &historyPub{} + rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")} + a := newRunAgent(pub, rt) + f := &fakeNetRunner{linkOut: linkWithENI} + a.netns = newTestNetns(f) + + as := testAssign() + as.ENIMacAddress = "52:54:00:de:ad:01" + a.runTask(context.Background(), as) + + if !f.sawAny("ip netns add ecs-t-001") { + t.Fatalf("expected task netns build, got %v", f.joined()) + } + if len(rt.Runs) != 1 || rt.Runs[0].NetnsPath != netnsPathFor("t-001") { + t.Fatalf("want NetnsPath %q on RunSpec, got %+v", netnsPathFor("t-001"), rt.Runs) + } +} + +// bridge/host: no ENI MAC means no netns is built and the container runs in the +// host (VM) netns (empty NetnsPath). +func TestRunTask_NoMacSkipsNetns(t *testing.T) { + pub := &historyPub{} + rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")} + a := newRunAgent(pub, rt) + f := &fakeNetRunner{linkOut: linkWithENI} + a.netns = newTestNetns(f) + + a.runTask(context.Background(), testAssign()) + + if len(f.joined()) != 0 { + t.Fatalf("expected no netns commands, got %v", f.joined()) + } + if len(rt.Runs) != 1 || rt.Runs[0].NetnsPath != "" { + t.Fatalf("want empty NetnsPath, got %+v", rt.Runs) + } +} + 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/scripts/images/ecs-agent/manifest.conf b/scripts/images/ecs-agent/manifest.conf index 48393a6d3..5ddbb2fe9 100644 --- a/scripts/images/ecs-agent/manifest.conf +++ b/scripts/images/ecs-agent/manifest.conf @@ -11,8 +11,9 @@ IMAGE_SIZE="4G" # containerd-openrc ships the OpenRC service for containerd (the containerd apk # itself has no init script). cni-plugins + the baked bridge conflist ready the # image for Sprint 4d task networking. curl is used by the first-boot CA hook and -# IMDS probes. -APK_PACKAGES="containerd containerd-openrc runc cni-plugins ca-certificates curl openrc" +# IMDS probes. iproute2 supplies the full `ip` (busybox lacks `ip netns`) the +# agent uses to build awsvpc task netns; udhcpc is busybox built-in. +APK_PACKAGES="containerd containerd-openrc runc cni-plugins ca-certificates curl openrc iproute2" # ecs-agent is the only built binary (static, FIPS boot like the host build). BUILD_COMMANDS="CGO_ENABLED=0 GOFIPS140=v1.0.0 go build -ldflags \"-s -w\" -o bin/ecs-agent ./cmd/ecs-agent" diff --git a/spinifex/handlers/ecs/bus/messages.go b/spinifex/handlers/ecs/bus/messages.go index f5a575751..6f153c467 100644 --- a/spinifex/handlers/ecs/bus/messages.go +++ b/spinifex/handlers/ecs/bus/messages.go @@ -79,6 +79,13 @@ type Assign struct { TaskDefFamily string `json:"taskDefFamily"` TaskDefRevision int `json:"taskDefRevision"` Containers []AssignContainer `json:"containers"` + // ENI* describe the awsvpc task ENI the scheduler already created + hot-plugged + // into this instance. The agent finds the NIC by ENIMacAddress, moves it into + // the task netns, and assigns ENIPrivateIP. Empty for bridge/host mode. + ENIID string `json:"eniId,omitempty"` + ENIMacAddress string `json:"eniMac,omitempty"` + ENIPrivateIP string `json:"eniPrivateIp,omitempty"` + ENISubnetID string `json:"eniSubnetId,omitempty"` // CredID + TaskRoleARN are placeholders for the Sprint 4g IAM credential // endpoint; the agent ignores them in 4e. CredID string `json:"credId,omitempty"` diff --git a/spinifex/handlers/ecs/eni.go b/spinifex/handlers/ecs/eni.go new file mode 100644 index 000000000..9fdd2440b --- /dev/null +++ b/spinifex/handlers/ecs/eni.go @@ -0,0 +1,152 @@ +package handlers_ecs + +import ( + "fmt" + "log/slog" + "strings" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// eniRequestTimeout bounds each ENI control-plane NATS round-trip. Attach drives +// the QMP hot-plug pipeline so it gets the same 30s budget the gateway uses. +const eniRequestTimeout = 30 * time.Second + +// taskENIDeviceIndex is the secondary device index every awsvpc task ENI hot-plugs +// onto; the primary NIC (index 0) is the instance's own ENI. +const taskENIDeviceIndex = 1 + +// eniAllocation is the identity of a freshly created task ENI. +type eniAllocation struct { + ENIID string + MacAddress string + PrivateIP string + SubnetID string + AttachmentID string +} + +// eniController is the scheduler's ENI control-plane: it owns create, attach, +// detach, and delete for awsvpc task ENIs (single writer). The agent never calls +// these — it only wires the guest netns once the device is hot-plugged. +type eniController interface { + Allocate(accountID, subnetID string, securityGroups []*string) (eniAllocation, error) + Attach(accountID, instanceID, eniID string) (attachmentID string, err error) + Release(accountID string, rec *TaskRecord) error +} + +// natsENIController drives the EC2 ENI handlers over NATS. Create/Delete are plain +// request/response; Attach/Detach dispatch to the owning daemon via ec2.cmd.{id} +// (the Sprint 3 hot-plug path). +type natsENIController struct { + nc *nats.Conn + timeout time.Duration +} + +var _ eniController = (*natsENIController)(nil) + +func newNATSENIController(nc *nats.Conn) *natsENIController { + return &natsENIController{nc: nc, timeout: eniRequestTimeout} +} + +// Allocate creates an ENI in subnetID with the given security groups. +func (c *natsENIController) Allocate(accountID, subnetID string, securityGroups []*string) (eniAllocation, error) { + in := &ec2.CreateNetworkInterfaceInput{ + SubnetId: aws.String(subnetID), + Groups: securityGroups, + Description: aws.String("ecs-awsvpc-task"), + } + out, err := utils.NATSRequest[ec2.CreateNetworkInterfaceOutput](c.nc, "ec2.CreateNetworkInterface", in, c.timeout, accountID) + if err != nil { + return eniAllocation{}, fmt.Errorf("create task ENI: %w", err) + } + if out.NetworkInterface == nil || aws.StringValue(out.NetworkInterface.NetworkInterfaceId) == "" { + return eniAllocation{}, fmt.Errorf("create task ENI: empty response") + } + ni := out.NetworkInterface + return eniAllocation{ + ENIID: aws.StringValue(ni.NetworkInterfaceId), + MacAddress: strings.ToLower(aws.StringValue(ni.MacAddress)), + PrivateIP: aws.StringValue(ni.PrivateIpAddress), + SubnetID: subnetID, + }, nil +} + +// Attach hot-plugs eniID onto instanceID and returns the attachment ID. +func (c *natsENIController) Attach(accountID, instanceID, eniID string) (string, error) { + cmd := types.EC2InstanceCommand{ + ID: instanceID, + Attributes: types.EC2CommandAttributes{AttachENI: true}, + AttachENIData: &types.AttachENIData{ + NetworkInterfaceID: eniID, + DeviceIndex: taskENIDeviceIndex, + }, + } + out, err := utils.NATSRequest[ec2.AttachNetworkInterfaceOutput](c.nc, eniCmdSubject(instanceID), cmd, c.timeout, accountID) + if err != nil { + return "", fmt.Errorf("attach task ENI %s -> %s: %w", eniID, instanceID, err) + } + return aws.StringValue(out.AttachmentId), nil +} + +// Release detaches (force) then deletes the task ENI. Both steps treat a NotFound +// as success so the graceful-stop and reaper paths can each run idempotently. +func (c *natsENIController) Release(accountID string, rec *TaskRecord) error { + if rec == nil || rec.ENIID == "" { + return nil + } + if rec.ENIAttachmentID != "" && rec.ContainerInstanceID != "" { + cmd := types.EC2InstanceCommand{ + ID: rec.ContainerInstanceID, + Attributes: types.EC2CommandAttributes{DetachENI: true}, + DetachENIData: &types.DetachENIData{ + AttachmentID: rec.ENIAttachmentID, + Force: true, + }, + } + _, err := utils.NATSRequest[ec2.DetachNetworkInterfaceOutput](c.nc, eniCmdSubject(rec.ContainerInstanceID), cmd, c.timeout, accountID) + if err != nil && !isENINotFound(err) { + return fmt.Errorf("detach task ENI %s: %w", rec.ENIID, err) + } + } + + del := &ec2.DeleteNetworkInterfaceInput{NetworkInterfaceId: aws.String(rec.ENIID)} + _, err := utils.NATSRequest[ec2.DeleteNetworkInterfaceOutput](c.nc, "ec2.DeleteNetworkInterface", del, c.timeout, accountID) + if err != nil && !isENINotFound(err) { + return fmt.Errorf("delete task ENI %s: %w", rec.ENIID, err) + } + return nil +} + +func eniCmdSubject(instanceID string) string { + return fmt.Sprintf("ec2.cmd.%s", instanceID) +} + +// reclaimTaskENI releases an awsvpc task's ENI on the single-writer teardown path +// (graceful stop + reaper both reach it). Best effort: a failure is logged and the +// ENI fields stay on the record for a later retry, never blocking the STOPPED +// transition. No-op for non-awsvpc tasks or tasks with no ENI. +func (s *Service) reclaimTaskENI(accountID string, task *TaskRecord) { + if s.eni == nil || task == nil || task.NetworkMode != NetworkModeAwsvpc || task.ENIID == "" { + return + } + if err := s.eni.Release(accountID, task); err != nil { + slog.Error("ECS: task ENI release failed", + "task", task.TaskID, "eni", task.ENIID, "err", err) + } +} + +// isENINotFound reports whether err is an idempotent already-gone signal — a +// missing ENI or attachment means the teardown is already converged. +func isENINotFound(err error) bool { + if err == nil { + return false + } + msg := err.Error() + return strings.Contains(msg, "InvalidNetworkInterfaceID.NotFound") || + strings.Contains(msg, "InvalidAttachmentID.NotFound") +} diff --git a/spinifex/handlers/ecs/eni_test.go b/spinifex/handlers/ecs/eni_test.go new file mode 100644 index 000000000..48f1a921c --- /dev/null +++ b/spinifex/handlers/ecs/eni_test.go @@ -0,0 +1,359 @@ +package handlers_ecs + +import ( + "encoding/json" + "errors" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// stubENI is an in-memory eniController for the awsvpc unit tests. +type stubENI struct { + allocCalls int + attachCalls int + releaseCalls int + allocErr error + attachErr error + lastSubnet string + lastSGs []string + released []string +} + +func (s *stubENI) Allocate(_, subnetID string, sgs []*string) (eniAllocation, error) { + s.allocCalls++ + s.lastSubnet = subnetID + for _, g := range sgs { + s.lastSGs = append(s.lastSGs, aws.StringValue(g)) + } + if s.allocErr != nil { + return eniAllocation{}, s.allocErr + } + return eniAllocation{ + ENIID: "eni-stub", + MacAddress: "02:aa:bb:cc:dd:ee", + PrivateIP: "172.31.0.50", + SubnetID: subnetID, + }, nil +} + +func (s *stubENI) Attach(_, _, _ string) (string, error) { + s.attachCalls++ + if s.attachErr != nil { + return "", s.attachErr + } + return "eni-attach-stub", nil +} + +func (s *stubENI) Release(_ string, rec *TaskRecord) error { + s.releaseCalls++ + s.released = append(s.released, rec.ENIID) + return nil +} + +func registerAwsvpcTaskDef(t *testing.T, svc *Service, family string, cpu, mem int) { + t.Helper() + _, err := svc.RegisterTaskDefinition(&ecs.RegisterTaskDefinitionInput{ + Family: aws.String(family), + NetworkMode: aws.String(NetworkModeAwsvpc), + 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) +} + +func awsvpcRunInput() *ecs.RunTaskInput { + return &ecs.RunTaskInput{ + Cluster: aws.String("web"), + TaskDefinition: aws.String("app"), + Count: aws.Int64(1), + NetworkConfiguration: &ecs.NetworkConfiguration{ + AwsvpcConfiguration: &ecs.AwsVpcConfiguration{ + Subnets: []*string{aws.String("subnet-1")}, + SecurityGroups: []*string{aws.String("sg-1")}, + }, + }, + } +} + +func TestRunTask_Awsvpc_AllocatesAttachesAssigns(t *testing.T) { + svc, nc := newTestService(t) + eni := &stubENI{} + svc.eni = eni + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerAwsvpcTaskDef(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(awsvpcRunInput(), testAccountID) + require.NoError(t, err) + require.Len(t, out.Tasks, 1) + assert.Empty(t, out.Failures) + assert.Equal(t, 1, eni.allocCalls) + assert.Equal(t, 1, eni.attachCalls) + assert.Equal(t, "subnet-1", eni.lastSubnet) + assert.Contains(t, eni.lastSGs, "sg-1") + + // Attachment surfaced on DescribeTasks. + require.Len(t, out.Tasks[0].Attachments, 1) + att := out.Tasks[0].Attachments[0] + assert.Equal(t, "ElasticNetworkInterface", aws.StringValue(att.Type)) + assert.Equal(t, "eni-attach-stub", aws.StringValue(att.Id)) + assert.Equal(t, "eni-stub", detailValue(att, "networkInterfaceId")) + assert.Equal(t, "172.31.0.50", detailValue(att, "privateIPv4Address")) + + // ENI plumbed into the assign payload. + 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, "eni-stub", as.ENIID) + assert.Equal(t, "02:aa:bb:cc:dd:ee", as.ENIMacAddress) + assert.Equal(t, "172.31.0.50", as.ENIPrivateIP) + assert.Equal(t, "subnet-1", as.ENISubnetID) +} + +func TestRunTask_Awsvpc_NoSubnets_Errors(t *testing.T) { + svc, _ := newTestService(t) + eni := &stubENI{} + svc.eni = eni + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerAwsvpcTaskDef(t, svc, "app", 128, 256) + registerInstance(t, svc, "web", "i-1", 1024, 2048) + + // awsvpc task def with no networkConfiguration → request rejected, no ENI. + _, err = svc.RunTask(&ecs.RunTaskInput{ + Cluster: aws.String("web"), TaskDefinition: aws.String("app"), Count: aws.Int64(1), + }, testAccountID) + require.Error(t, err) + assert.Equal(t, 0, eni.allocCalls) +} + +func TestRunTask_Awsvpc_AttachFailure_RollsBack(t *testing.T) { + svc, _ := newTestService(t) + eni := &stubENI{attachErr: errors.New("hot-plug timeout")} + svc.eni = eni + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerAwsvpcTaskDef(t, svc, "app", 128, 256) + registerInstance(t, svc, "web", "i-1", 1024, 2048) + + out, err := svc.RunTask(awsvpcRunInput(), testAccountID) + require.NoError(t, err) + assert.Empty(t, out.Tasks) + require.Len(t, out.Failures, 1) + assert.Equal(t, "RESOURCE:eni", aws.StringValue(out.Failures[0].Reason)) + + // ENI was allocated then released; reservation rolled back. + assert.Equal(t, 1, eni.allocCalls) + assert.Equal(t, 1, eni.releaseCalls) + assert.Contains(t, eni.released, "eni-stub") + + 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)) +} + +func TestRecordTaskState_Awsvpc_ReleasesENIOnStopOnce(t *testing.T) { + svc, _ := newTestService(t) + eni := &stubENI{} + svc.eni = eni + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerAwsvpcTaskDef(t, svc, "app", 128, 256) + registerInstance(t, svc, "web", "i-1", 1024, 2048) + out, err := svc.RunTask(awsvpcRunInput(), testAccountID) + require.NoError(t, err) + taskID := containerInstanceShortID(aws.StringValue(out.Tasks[0].TaskArn)) + + stop := &bus.TaskState{ + AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: taskID, + LastStatus: bus.TaskStatusStopped, Reason: "exited", + } + require.NoError(t, svc.recordTaskState(stop)) + require.NoError(t, svc.recordTaskState(stop)) // re-report STOPPED + + // Released exactly once (the prev!=STOPPED guard makes the re-report a no-op). + assert.Equal(t, 1, eni.releaseCalls) + assert.Equal(t, []string{"eni-stub"}, eni.released) +} + +func TestReaper_Awsvpc_ReleasesENI(t *testing.T) { + svc, nc := newTestService(t) + eni := &stubENI{} + svc.eni = eni + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerAwsvpcTaskDef(t, svc, "app", 128, 256) + registerInstance(t, svc, "web", "i-1", 1024, 2048) + _, err = svc.RunTask(awsvpcRunInput(), testAccountID) + require.NoError(t, err) + + 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()) + + assert.Equal(t, 1, eni.releaseCalls) + assert.Contains(t, eni.released, "eni-stub") +} + +func TestRunTask_BridgeMode_NoENI(t *testing.T) { + svc, _ := newTestService(t) + eni := &stubENI{} + svc.eni = eni + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerTaskDef(t, svc, "app", 128, 256) // no networkMode → bridge default + registerInstance(t, svc, "web", "i-1", 1024, 2048) + + 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.Equal(t, 0, eni.allocCalls) + assert.Empty(t, out.Tasks[0].Attachments) +} + +func TestResolveNetworkMode(t *testing.T) { + assert.Equal(t, NetworkModeAwsvpc, resolveNetworkMode(&TaskDefRecord{NetworkMode: "awsvpc"})) + assert.Equal(t, NetworkModeAwsvpc, resolveNetworkMode(&TaskDefRecord{NetworkMode: "AWSVPC"})) + assert.Equal(t, NetworkModeHost, resolveNetworkMode(&TaskDefRecord{NetworkMode: "host"})) + assert.Equal(t, NetworkModeBridge, resolveNetworkMode(&TaskDefRecord{})) // default + assert.Equal(t, NetworkModeBridge, resolveNetworkMode(&TaskDefRecord{NetworkMode: "garbage"})) +} + +func TestParseAwsvpcConfig(t *testing.T) { + in := awsvpcRunInput() + cfg, err := parseAwsvpcConfig(in, NetworkModeAwsvpc) + require.NoError(t, err) + assert.Equal(t, "subnet-1", cfg.firstSubnet()) + assert.Equal(t, []string{"sg-1"}, cfg.SecurityGroups) + + // awsvpc with no subnets is an error. + _, err = parseAwsvpcConfig(&ecs.RunTaskInput{}, NetworkModeAwsvpc) + require.Error(t, err) + + // bridge with no config is fine (empty). + cfg, err = parseAwsvpcConfig(&ecs.RunTaskInput{}, NetworkModeBridge) + require.NoError(t, err) + assert.Empty(t, cfg.Subnets) +} + +// respond subscribes to subject and replies with the JSON of whatever fn returns +// for each request, modelling the EC2/daemon handlers the controller calls. +func respond(t *testing.T, nc *nats.Conn, subject string, fn func([]byte) any) { + t.Helper() + sub, err := nc.Subscribe(subject, func(msg *nats.Msg) { + data, _ := json.Marshal(fn(msg.Data)) + _ = msg.Respond(data) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) +} + +func TestNATSENIController_AllocateAttachRelease(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + c := newNATSENIController(nc) + + respond(t, nc, "ec2.CreateNetworkInterface", func([]byte) any { + return ec2.CreateNetworkInterfaceOutput{NetworkInterface: &ec2.NetworkInterface{ + NetworkInterfaceId: aws.String("eni-real"), + MacAddress: aws.String("02:11:22:33:44:55"), + PrivateIpAddress: aws.String("172.31.5.9"), + }} + }) + respond(t, nc, "ec2.cmd.i-1", func(req []byte) any { + var cmd types.EC2InstanceCommand + _ = json.Unmarshal(req, &cmd) + if cmd.Attributes.AttachENI { + return ec2.AttachNetworkInterfaceOutput{AttachmentId: aws.String("att-real")} + } + return ec2.DetachNetworkInterfaceOutput{} + }) + respond(t, nc, "ec2.DeleteNetworkInterface", func([]byte) any { + return ec2.DeleteNetworkInterfaceOutput{} + }) + + alloc, err := c.Allocate(testAccountID, "subnet-1", []*string{aws.String("sg-1")}) + require.NoError(t, err) + assert.Equal(t, "eni-real", alloc.ENIID) + assert.Equal(t, "02:11:22:33:44:55", alloc.MacAddress) + assert.Equal(t, "172.31.5.9", alloc.PrivateIP) + + attID, err := c.Attach(testAccountID, "i-1", "eni-real") + require.NoError(t, err) + assert.Equal(t, "att-real", attID) + + require.NoError(t, c.Release(testAccountID, &TaskRecord{ + ENIID: "eni-real", ENIAttachmentID: "att-real", ContainerInstanceID: "i-1", + })) +} + +func TestNATSENIController_Release_NotFoundIsSuccess(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + c := newNATSENIController(nc) + + respond(t, nc, "ec2.cmd.i-1", func([]byte) any { + return json.RawMessage(utils.GenerateErrorPayload("InvalidAttachmentID.NotFound")) + }) + respond(t, nc, "ec2.DeleteNetworkInterface", func([]byte) any { + return json.RawMessage(utils.GenerateErrorPayload("InvalidNetworkInterfaceID.NotFound")) + }) + + // Both legs report already-gone; Release converges without error. + require.NoError(t, c.Release(testAccountID, &TaskRecord{ + ENIID: "eni-x", ENIAttachmentID: "att-x", ContainerInstanceID: "i-1", + })) +} + +func TestNATSENIController_Release_NoENINoop(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + c := newNATSENIController(nc) + require.NoError(t, c.Release(testAccountID, &TaskRecord{})) // no ENIID + require.NoError(t, c.Release(testAccountID, nil)) +} + +func TestIsENINotFound(t *testing.T) { + assert.True(t, isENINotFound(errors.New("InvalidNetworkInterfaceID.NotFound"))) + assert.True(t, isENINotFound(errors.New("delete: InvalidAttachmentID.NotFound"))) + assert.False(t, isENINotFound(errors.New("ServerInternal"))) + assert.False(t, isENINotFound(nil)) +} + +func detailValue(att *ecs.Attachment, name string) string { + for _, d := range att.Details { + if aws.StringValue(d.Name) == name { + return aws.StringValue(d.Value) + } + } + return "" +} diff --git a/spinifex/handlers/ecs/instances.go b/spinifex/handlers/ecs/instances.go index 0e939f9ed..03c74c032 100644 --- a/spinifex/handlers/ecs/instances.go +++ b/spinifex/handlers/ecs/instances.go @@ -225,8 +225,9 @@ func (s *Service) recordTaskState(msg *bus.TaskState) error { return err } - // Release capacity once, on the transition into STOPPED. + // Release capacity + reclaim the task ENI once, on the transition into STOPPED. if msg.LastStatus == TaskStatusStopped && prev != TaskStatusStopped { + s.reclaimTaskENI(msg.AccountID, &task) return s.releaseReservation(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID, task.ReservedCPU, task.ReservedMemoryMiB) } return nil diff --git a/spinifex/handlers/ecs/netcfg.go b/spinifex/handlers/ecs/netcfg.go new file mode 100644 index 000000000..ba7309148 --- /dev/null +++ b/spinifex/handlers/ecs/netcfg.go @@ -0,0 +1,78 @@ +package handlers_ecs + +import ( + "errors" + "strings" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/mulgadc/spinifex/spinifex/awserrors" +) + +// Task network modes (AWS ECS enum). EC2 launch type defaults to bridge when a +// task definition omits networkMode. +const ( + NetworkModeAwsvpc = "awsvpc" + NetworkModeBridge = "bridge" + NetworkModeHost = "host" + NetworkModeNone = "none" + + defaultNetworkMode = NetworkModeBridge +) + +// awsvpcConfig is the resolved subnet + security-group selection for an awsvpc +// task, extracted from RunTask's networkConfiguration. +type awsvpcConfig struct { + Subnets []string + SecurityGroups []string +} + +// resolveNetworkMode normalises a task definition's networkMode, defaulting to +// bridge when unset (AWS EC2-launch-type behaviour). +func resolveNetworkMode(td *TaskDefRecord) string { + mode := strings.ToLower(strings.TrimSpace(td.NetworkMode)) + switch mode { + case NetworkModeAwsvpc, NetworkModeBridge, NetworkModeHost, NetworkModeNone: + return mode + default: + return defaultNetworkMode + } +} + +// parseAwsvpcConfig pulls the awsvpcConfiguration off a RunTask input. It returns +// an error only when the mode is awsvpc and no usable subnet is supplied; for +// other modes it returns an empty config and nil. +func parseAwsvpcConfig(input *ecs.RunTaskInput, mode string) (awsvpcConfig, error) { + var cfg awsvpcConfig + if input != nil && input.NetworkConfiguration != nil && input.NetworkConfiguration.AwsvpcConfiguration != nil { + v := input.NetworkConfiguration.AwsvpcConfiguration + cfg.Subnets = awsStringSlice(v.Subnets) + cfg.SecurityGroups = awsStringSlice(v.SecurityGroups) + } + if mode == NetworkModeAwsvpc && len(cfg.Subnets) == 0 { + return cfg, errors.New(awserrors.ErrorInvalidParameterValue) + } + return cfg, nil +} + +// firstSubnet returns the subnet the task ENI is allocated in. v1 takes the +// first supplied subnet; AZ-affinity to the placed instance is a future refinement. +func (c awsvpcConfig) firstSubnet() string { + if len(c.Subnets) == 0 { + return "" + } + return c.Subnets[0] +} + +// awsStringSliceFromConfig adapts the config security groups to the *string slice +// shape the EC2 CreateNetworkInterface input expects. +func (c awsvpcConfig) securityGroupPtrs() []*string { + if len(c.SecurityGroups) == 0 { + return nil + } + out := make([]*string, 0, len(c.SecurityGroups)) + for _, g := range c.SecurityGroups { + out = append(out, aws.String(g)) + } + return out +} diff --git a/spinifex/handlers/ecs/records.go b/spinifex/handlers/ecs/records.go index 30d3b0fee..6836d109f 100644 --- a/spinifex/handlers/ecs/records.go +++ b/spinifex/handlers/ecs/records.go @@ -143,7 +143,17 @@ type TaskRecord struct { 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"` + // NetworkMode is the resolved task network mode (awsvpc|bridge|host). The + // STOPPED path consults it to decide whether an ENI must be reclaimed. + NetworkMode string `json:"networkMode,omitempty"` + // ENI* hold the per-task elastic network interface for awsvpc mode, allocated + // by the scheduler at placement and reclaimed at STOPPED. Empty otherwise. + ENIID string `json:"eniId,omitempty"` + ENIAttachmentID string `json:"eniAttachmentId,omitempty"` + ENIPrivateIP string `json:"eniPrivateIp,omitempty"` + ENIMacAddress string `json:"eniMac,omitempty"` + ENISubnetID string `json:"eniSubnetId,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 index 04ce88d02..b9e893607 100644 --- a/spinifex/handlers/ecs/scheduler.go +++ b/spinifex/handlers/ecs/scheduler.go @@ -228,7 +228,7 @@ func (sc *Scheduler) reap() { } } -func (sc *Scheduler) reapBucket(kv nats.KeyValue, _ string, now time.Time) { +func (sc *Scheduler) reapBucket(kv nats.KeyValue, accountID string, now time.Time) { keys, err := keysWithPrefix(kv, "clusters/") if err != nil { return @@ -251,12 +251,13 @@ func (sc *Scheduler) reapBucket(kv nats.KeyValue, _ string, now time.Time) { if perr := putJSON(kv, k, &inst); perr != nil { continue } - sc.stopInstanceTasks(kv, inst.Cluster, inst.InstanceID) + sc.stopInstanceTasks(kv, accountID, 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) { +// stopInstanceTasks transitions a reaped instance's non-stopped tasks to STOPPED +// and reclaims each awsvpc task's ENI (leak guard for a dead agent). +func (sc *Scheduler) stopInstanceTasks(kv nats.KeyValue, accountID, cluster, instanceID string) { keys, err := keysWithPrefix(kv, TasksPrefix(cluster)) if err != nil { return @@ -276,6 +277,7 @@ func (sc *Scheduler) stopInstanceTasks(kv nats.KeyValue, cluster, instanceID str if perr := putJSON(kv, k, &task); perr != nil { continue } + sc.svc.reclaimTaskENI(accountID, &task) } } diff --git a/spinifex/handlers/ecs/service.go b/spinifex/handlers/ecs/service.go index 8f0ea9d3b..d3d6a2095 100644 --- a/spinifex/handlers/ecs/service.go +++ b/spinifex/handlers/ecs/service.go @@ -43,6 +43,9 @@ type Service struct { nc *nats.Conn region string suffix string + // eni owns the awsvpc task-ENI control-plane (create/attach/detach/delete). + // Defaults to the NATS-backed controller; tests substitute a stub. + eni eniController } var _ ECSService = (*Service)(nil) @@ -51,7 +54,7 @@ var _ ECSService = (*Service)(nil) // 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} + return &Service{nc: nc, region: region, suffix: suffix, eni: newNATSENIController(nc)} } // defaultCluster is the implicit cluster name when a request omits one. diff --git a/spinifex/handlers/ecs/tasks.go b/spinifex/handlers/ecs/tasks.go index 177467dac..e2baa78b7 100644 --- a/spinifex/handlers/ecs/tasks.go +++ b/spinifex/handlers/ecs/tasks.go @@ -51,6 +51,12 @@ func (s *Service) RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTa strategy := placementStrategyFromAWS(input.PlacementStrategy) cpu, mem := taskDef.reservedCPU(), taskDef.reservedMemory() + mode := resolveNetworkMode(taskDef) + netCfg, err := parseAwsvpcConfig(input, mode) + if err != nil { + return nil, err + } + out := &ecs.RunTaskOutput{} for i := 0; i < count; i++ { taskID := uuid.NewString() @@ -64,6 +70,14 @@ func (s *Service) RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTa } rec := s.newTaskRecord(accountID, cluster, taskID, taskDef, inst, cpu, mem) + rec.NetworkMode = mode + if mode == NetworkModeAwsvpc { + if failure := s.provisionTaskENI(kv, accountID, cluster, rec, netCfg); failure != nil { + out.Failures = append(out.Failures, failure) + continue + } + } + if err := putJSON(kv, TaskKey(cluster, taskID), rec); err != nil { return nil, err } @@ -75,6 +89,37 @@ func (s *Service) RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTa return out, nil } +// provisionTaskENI allocates and hot-plugs an awsvpc task's ENI, stamping its +// identity onto rec. On any failure it rolls back (releases a half-allocated ENI +// and the placement reservation) and returns a RunTask failure for this task — +// the caller skips it without leaking the ENI or the reserved capacity. +func (s *Service) provisionTaskENI(kv nats.KeyValue, accountID, cluster string, rec *TaskRecord, netCfg awsvpcConfig) *ecs.Failure { + rollback := func(reason string, err error) *ecs.Failure { + if rerr := s.releaseReservation(kv, cluster, rec.ContainerInstanceID, rec.TaskID, rec.ReservedCPU, rec.ReservedMemoryMiB); rerr != nil { + slog.Error("ECS RunTask: reservation rollback failed", "task", rec.TaskID, "err", rerr) + } + return &ecs.Failure{Reason: aws.String(reason), Detail: aws.String(err.Error())} + } + + alloc, err := s.eni.Allocate(accountID, netCfg.firstSubnet(), netCfg.securityGroupPtrs()) + if err != nil { + return rollback("RESOURCE:eni", err) + } + rec.ENIID = alloc.ENIID + rec.ENIMacAddress = alloc.MacAddress + rec.ENIPrivateIP = alloc.PrivateIP + rec.ENISubnetID = alloc.SubnetID + + attachmentID, err := s.eni.Attach(accountID, rec.ContainerInstanceID, alloc.ENIID) + if err != nil { + s.reclaimTaskENI(accountID, rec) + rec.ENIID, rec.ENIMacAddress, rec.ENIPrivateIP, rec.ENISubnetID = "", "", "", "" + return rollback("RESOURCE:eni", err) + } + rec.ENIAttachmentID = attachmentID + return 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) { @@ -149,6 +194,10 @@ func (s *Service) publishAssign(accountID, cluster, instanceID string, rec *Task TaskARN: rec.ARN, TaskDefFamily: td.Family, TaskDefRevision: td.Revision, + ENIID: rec.ENIID, + ENIMacAddress: rec.ENIMacAddress, + ENIPrivateIP: rec.ENIPrivateIP, + ENISubnetID: rec.ENISubnetID, AssignedAt: time.Now().UTC(), } for _, c := range td.Containers { @@ -223,6 +272,19 @@ func (s *Service) taskToAWS(accountID string, r *TaskRecord) *ecs.Task { if r.StoppedReason != "" { t.StoppedReason = aws.String(r.StoppedReason) } + if r.ENIID != "" { + t.Attachments = append(t.Attachments, &ecs.Attachment{ + Id: aws.String(r.ENIAttachmentID), + Type: aws.String("ElasticNetworkInterface"), + Status: aws.String(r.LastStatus), + Details: []*ecs.KeyValuePair{ + {Name: aws.String("networkInterfaceId"), Value: aws.String(r.ENIID)}, + {Name: aws.String("privateIPv4Address"), Value: aws.String(r.ENIPrivateIP)}, + {Name: aws.String("macAddress"), Value: aws.String(r.ENIMacAddress)}, + {Name: aws.String("subnetId"), Value: aws.String(r.ENISubnetID)}, + }, + }) + } for _, c := range r.Containers { ctr := &ecs.Container{Name: aws.String(c.Name), LastStatus: aws.String(c.Status)} if c.ExitCode != nil {