mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
feat(ecs): awsvpc mode — ENI-per-task allocation + guest datapath (Sprint 4d) (#418)
* 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-<taskID>, 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.
This commit is contained in:
@@ -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{}),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
// "<idx>: <ifname>: <flags> ... link/ether <mac> ..."
|
||||
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)
|
||||
}
|
||||
@@ -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: <LOOPBACK,UP> mtu 65536 ... link/loopback 00:00:00:00:00:00 ...
|
||||
2: eth0: <BROADCAST,UP> mtu 1500 ... link/ether 52:54:00:aa:bb:cc ...
|
||||
3: eth1: <BROADCAST,UP> 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())
|
||||
}
|
||||
}
|
||||
@@ -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...))
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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"`
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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 ""
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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"`
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user