refactor: move ECS agent implementation to the ECS agent tree

cmd/ecs-agent keeps process lifecycle: signal context, the separate New
and Run calls, both failure log lines, the exit and the ldflags-set
build version, now passed in through Config.AgentVersion.
agents/ecs/agent owns Config, LoadConfig and the agent.
This commit is contained in:
joshsiv-mulga
2026-10-07 12:27:11 +10:00
parent 50753e9831
commit 59decbc25c
31 changed files with 122 additions and 82 deletions
+13 -13
View File
@@ -1,13 +1,6 @@
// ecs-agent runs inside a Spinifex ECS container instance (a guest VM booted from
// the ECS-AMI, with containerd baked in). It registers the host with the ECS
// scheduler through the AWS gateway over TLS+SigV4 (never NATS), heartbeats by
// re-registering, polls the gateway for task assignments, runs them through
// containerd, and reports state back over the gateway.
//
// Static config (gateway URL, CA, region, cluster, seeded IAM creds, containerd
// socket) is read from the cloud-init env file /etc/spinifex-ecs/agent.env
// (KEY=value); real env vars override it. Host identity (account, instance, AZ)
// comes from IMDS at boot.
// Command ecs-agent runs inside a Spinifex ECS container instance; the
// implementation is in agents/ecs/agent. This binary owns process lifecycle:
// signal handling, startup and run failure logging, and the exit code.
package main
import (
@@ -18,12 +11,19 @@ import (
"syscall"
_ "github.com/mulgadc/bluebottle/pkg/fipsboot"
ecsagent "github.com/mulgadc/spinifex/spinifex/agents/ecs/agent"
)
func main() {
cfg := loadConfig(defaultEnvFile)
// version is the agent build version, reported in the register message.
// Overridable via -ldflags "-X main.version=...".
var version = "dev"
agent, err := New(cfg)
func main() {
cfg := ecsagent.LoadConfig(ecsagent.DefaultEnvFile)
cfg.AgentVersion = version
agent, err := ecsagent.New(cfg)
if err != nil {
slog.Error("ecs-agent: startup failed", "err", err)
os.Exit(1)
@@ -1,4 +1,4 @@
package main
package agent
import (
"bufio"
@@ -17,17 +17,13 @@ import (
"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
)
// version is the agent build version, reported in the register message.
// Overridable via -ldflags "-X main.version=...".
var version = "dev"
// Agent wires the ecs-agent's runtime seams: a gateway control-plane client, a
// container runtime, and an ECR resolver. It registers the host as a container
// instance at boot, heartbeats (re-registers) while alive, polls the gateway for
// task assignments, and runs them through containerd, reporting state back over
// the gateway. It never connects to NATS — the bus stays host-internal.
type Agent struct {
cfg config
cfg Config
id identity
cp controlPlane
resolver ctrruntime.Resolver
@@ -62,7 +58,7 @@ type Agent struct {
// with fakes; New builds the production seams and delegates here. rt may be nil
// when containerd is unavailable; the assign path then reports the task STOPPED
// rather than crashing, and the watch loop keeps dialling for it.
func newAgent(cfg config, id identity, cp controlPlane, rt ctrruntime.Runtime, resolver ctrruntime.Resolver) *Agent {
func newAgent(cfg Config, id identity, cp controlPlane, rt ctrruntime.Runtime, resolver ctrruntime.Resolver) *Agent {
gate := newRuntimeGate(rt)
a := &Agent{
cfg: cfg,
@@ -87,7 +83,7 @@ func newAgent(cfg config, id identity, cp controlPlane, rt ctrruntime.Runtime, r
// registration and heartbeat still run so the instance is visible while the
// runtime recovers. The ECR gateway client is built lazily on first image pull
// (not here), so a missing or malformed gateway CA does not stop registration.
func New(cfg config) (*Agent, error) {
func New(cfg Config) (*Agent, error) {
imdsClient := &http.Client{Timeout: 5 * time.Second}
meta, err := fetchInstanceMetadata(imdsClient, cfg.IMDSBase)
@@ -103,7 +99,7 @@ func New(cfg config) (*Agent, error) {
InstanceType: meta.InstanceType,
Hostname: host,
Capacity: detectCapacity(discoverNvidiaGPUs(execCommandRunner)),
AgentVersion: version,
AgentVersion: cfg.AgentVersion,
}
creds := credentials.NewIMDSProvider(imdsClient, cfg.IMDSBase)
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -10,7 +10,7 @@ import (
func TestAgent_RunRegistersThenStopsOnContext(t *testing.T) {
cp := &fakeCP{}
cfg := config{Heartbeat: 5 * time.Millisecond, PollInterval: 5 * time.Millisecond}
cfg := Config{Heartbeat: 5 * time.Millisecond, PollInterval: 5 * time.Millisecond}
puller := &ctrruntime.FakePuller{}
a := newAgent(cfg, testIdentity(), cp, puller, nil)
a.closers = append(a.closers, puller.Close)
@@ -1,4 +1,4 @@
package main
package agent
import (
"strconv"
@@ -7,8 +7,11 @@ import (
"github.com/mulgadc/spinifex/internal/guestenv"
)
// DefaultEnvFile is the cloud-init env file the ECS AMI seeds with the
// agent's static configuration.
const DefaultEnvFile = "/etc/spinifex-ecs/agent.env"
const (
defaultEnvFile = "/etc/spinifex-ecs/agent.env"
defaultGatewayCA = "/etc/spinifex-ecs/gateway-ca.pem"
defaultIMDSBase = "http://169.254.169.254/latest"
defaultContainerdSocket = "/run/containerd/containerd.sock"
@@ -16,13 +19,13 @@ const (
defaultPollInterval = 5 * time.Second
)
// config holds the static settings the agent reads at boot. Cluster identity
// Config holds the static settings the agent reads at boot. Cluster identity
// (account ID, instance ID) is discovered at runtime from IMDS, not configured;
// only the cluster *name* the instance was launched into is static. AccessKey /
// SecretKey are legacy seeded creds, optional now that the agent SigV4-signs
// gateway calls with instance-role credentials from IMDS; the agent never holds
// a NATS token.
type config struct {
type Config struct {
GatewayURL string
GatewayCA string
Region string
@@ -36,13 +39,16 @@ type config struct {
RuntimeRetry time.Duration
CredEndpointIP string
CredEndpointPort int
// AgentVersion is the binary's build version, reported in the register
// message; the binary sets it, not the env file.
AgentVersion string
}
// loadConfig reads the cloud-init env file then lets real env vars override.
func loadConfig(envFile string) config {
// LoadConfig reads the cloud-init env file then lets real env vars override.
func LoadConfig(envFile string) Config {
get := guestenv.Load(envFile).Get
cfg := config{
cfg := Config{
GatewayURL: get("ECS_GATEWAY_URL"),
GatewayCA: get("ECS_GATEWAY_CA"),
Region: get("ECS_REGION"),
@@ -1,4 +1,4 @@
package main
package agent
import (
"os"
@@ -14,7 +14,7 @@ func TestLoadConfig_Defaults(t *testing.T) {
"ECS_CONTAINERD_SOCKET", "ECS_HEARTBEAT_INTERVAL", "ECS_POLL_INTERVAL"} {
t.Setenv(k, "")
}
cfg := loadConfig(filepath.Join(t.TempDir(), "absent.env"))
cfg := LoadConfig(filepath.Join(t.TempDir(), "absent.env"))
if cfg.GatewayCA != defaultGatewayCA {
t.Errorf("GatewayCA = %q, want default", cfg.GatewayCA)
}
@@ -45,7 +45,7 @@ func TestLoadConfig_FileThenEnvOverride(t *testing.T) {
t.Setenv("ECS_GATEWAY_URL", "https://gw.env")
t.Setenv("ECS_ACCESS_KEY", "AKIAENV")
cfg := loadConfig(envFile)
cfg := LoadConfig(envFile)
if cfg.GatewayURL != "https://gw.env" {
t.Errorf("GatewayURL = %q, want env override", cfg.GatewayURL)
}
@@ -62,12 +62,12 @@ func TestLoadConfig_FileThenEnvOverride(t *testing.T) {
func TestLoadConfig_HeartbeatOverride(t *testing.T) {
t.Setenv("ECS_HEARTBEAT_INTERVAL", "5s")
if cfg := loadConfig(filepath.Join(t.TempDir(), "absent.env")); cfg.Heartbeat != 5*time.Second {
if cfg := LoadConfig(filepath.Join(t.TempDir(), "absent.env")); cfg.Heartbeat != 5*time.Second {
t.Errorf("Heartbeat = %v, want 5s", cfg.Heartbeat)
}
// Garbage value falls back to the default, not zero.
t.Setenv("ECS_HEARTBEAT_INTERVAL", "not-a-duration")
if cfg := loadConfig(filepath.Join(t.TempDir(), "absent.env")); cfg.Heartbeat != defaultHeartbeat {
if cfg := LoadConfig(filepath.Join(t.TempDir(), "absent.env")); cfg.Heartbeat != defaultHeartbeat {
t.Errorf("Heartbeat = %v, want default on bad value", cfg.Heartbeat)
}
}
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -47,7 +47,7 @@ const pollTimeout = 30 * time.Second
// newGatewayControlPlane builds the SigV4 client signing with the instance-role
// credentials from IMDS (fetched per call so rotation is transparent) + pinned CA.
func newGatewayControlPlane(cfg config, creds credentials.CredentialsProvider) (*gatewayControlPlane, error) {
func newGatewayControlPlane(cfg Config, creds credentials.CredentialsProvider) (*gatewayControlPlane, error) {
client, err := ecsgw.New(cfg.GatewayURL, cfg.GatewayCA, credentialsFunc(creds), cfg.Region, pollTimeout)
if err != nil {
return nil, err
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -23,7 +23,7 @@ func (fixedCredsProvider) Retrieve(context.Context) (credentials.Credentials, er
func newTestGatewayControlPlane(t *testing.T, srv *httptest.Server) *gatewayControlPlane {
t.Helper()
cp, err := newGatewayControlPlane(config{GatewayURL: srv.URL, Region: "us-east-1"}, fixedCredsProvider{})
cp, err := newGatewayControlPlane(Config{GatewayURL: srv.URL, Region: "us-east-1"}, fixedCredsProvider{})
if err != nil {
t.Fatalf("newGatewayControlPlane: %v", err)
}
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -1,4 +1,4 @@
package main
package agent
import (
"testing"
+12
View File
@@ -0,0 +1,12 @@
// Package agent implements ecs-agent, which runs inside a Spinifex ECS
// container instance (a guest VM booted from the ECS-AMI, with containerd baked
// in). It registers the host with the ECS scheduler through the AWS gateway over
// TLS+SigV4 (never NATS), heartbeats by re-registering, polls the gateway for
// task assignments, runs them through containerd, and reports state back over
// the gateway.
//
// Static config (gateway URL, CA, region, cluster, seeded IAM creds, containerd
// socket) is read from the cloud-init env file /etc/spinifex-ecs/agent.env
// (KEY=value); real env vars override it. Host identity (account, instance, AZ)
// comes from IMDS at boot.
package agent
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -1,4 +1,4 @@
package main
package agent
import (
"os/exec"
@@ -1,4 +1,4 @@
package main
package agent
import (
"errors"
@@ -1,4 +1,4 @@
package main
package agent
import (
"fmt"
@@ -1,4 +1,4 @@
package main
package agent
import (
"sync"
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -1,4 +1,4 @@
package main
package agent
import (
"net/http"
@@ -1,4 +1,4 @@
package main
package agent
import (
"fmt"
@@ -1,4 +1,4 @@
package main
package agent
import (
"errors"
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -55,7 +55,7 @@ func TestReconcile_AdoptsRunningLabeledContainer(t *testing.T) {
adoptedContainer("t-001", "web", "default", "t-001", testTaskRole, "52:54:00:de:ad:01", true),
},
}
a := newAgent(config{}, testIdentity(), cp, rt, nil)
a := newAgent(Config{}, testIdentity(), cp, rt, nil)
a.cred = newCredEndpoint(nil, "us-east-1", "https://gw", "", "127.0.0.1", 0, nil)
adopted := a.reconcile(context.Background())
@@ -84,7 +84,7 @@ func TestReconcile_ReportsTaskWhoseContainerDied(t *testing.T) {
adoptedContainer("t-dead", "web", "default", "t-dead", testTaskRole, "52:54:00:de:ad:02", false),
},
}
a := newAgent(config{}, testIdentity(), cp, rt, nil)
a := newAgent(Config{}, testIdentity(), cp, rt, nil)
a.cred = newCredEndpoint(nil, "us-east-1", "https://gw", "", "127.0.0.1", 0, nil)
adopted := a.reconcile(context.Background())
@@ -116,7 +116,7 @@ func TestReconcile_PartiallyDeadTaskStaysRunning(t *testing.T) {
adoptedContainer("t-mixed", "sidecar", "default", "t-mixed", testTaskRole, "52:54:00:de:ad:03", false),
},
}
a := newAgent(config{}, testIdentity(), cp, rt, nil)
a := newAgent(Config{}, testIdentity(), cp, rt, nil)
a.cred = newCredEndpoint(nil, "us-east-1", "https://gw", "", "127.0.0.1", 0, nil)
a.reconcile(context.Background())
@@ -145,7 +145,7 @@ func TestPollAssignments_SeededTaskNotRerun(t *testing.T) {
nil,
}}
rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
a := newAgent(Config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
ctx, cancel := context.WithCancel(context.Background())
go a.pollAssignments(ctx, map[string]bool{"t-001": true})
@@ -197,7 +197,7 @@ func TestReconcile_IgnoresOtherClusterContainers(t *testing.T) {
adoptedContainer("t-ok", "web", "default", "", "", "", true),
},
}
a := newAgent(config{}, testIdentity(), cp, rt, nil)
a := newAgent(Config{}, testIdentity(), cp, rt, nil)
adopted := a.reconcile(context.Background())
@@ -213,13 +213,13 @@ func TestReconcile_IgnoresOtherClusterContainers(t *testing.T) {
// A nil runner or a List error degrades to an empty set without panicking.
func TestReconcile_DegradesGracefully(t *testing.T) {
noRunner := newAgent(config{}, testIdentity(), &fakeCP{}, nil, nil)
noRunner := newAgent(Config{}, testIdentity(), &fakeCP{}, nil, nil)
if got := noRunner.reconcile(context.Background()); len(got) != 0 {
t.Errorf("nil runner: adopted = %v, want empty", got)
}
rt := &ctrruntime.FakePuller{ListErr: errors.New("boom")}
listErr := newAgent(config{}, testIdentity(), &fakeCP{}, rt, nil)
listErr := newAgent(Config{}, testIdentity(), &fakeCP{}, rt, nil)
if got := listErr.reconcile(context.Background()); len(got) != 0 {
t.Errorf("list error: adopted = %v, want empty", got)
}
@@ -1,4 +1,4 @@
package main
package agent
import (
"fmt"
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -1,6 +1,6 @@
//test:in-package — package main cannot be imported by an external test package.
package main
package agent
import (
"context"
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -440,7 +440,7 @@ const (
labelTaskID = "mulga.ecs.taskID"
labelContainerName = "mulga.ecs.containerName"
labelClusterName = "mulga.ecs.clusterName"
labelCredID = "mulga.ecs.credID"
labelCredID = "mulga.ecs.credID" //nolint:gosec // G101: label key name, not a credential value.
labelTaskRoleARN = "mulga.ecs.taskRoleArn"
labelENIMac = "mulga.ecs.eniMac"
// labelStopTimeout carries a container's stopTimeout (seconds) so stopTask
@@ -1,4 +1,4 @@
package main
package agent
import (
"context"
@@ -25,13 +25,13 @@ func testAssign() *bus.Assign {
}
func newRunAgent(cp controlPlane, rt *ctrruntime.FakePuller) *Agent {
return newAgent(config{}, testIdentity(), cp, rt, nil)
return newAgent(Config{}, testIdentity(), cp, rt, nil)
}
// runTask with no runtime reports the task STOPPED instead of crashing.
func TestRunTask_NoRuntimeReportsStopped(t *testing.T) {
cp := &fakeCP{}
a := newAgent(config{}, testIdentity(), cp, nil, nil)
a := newAgent(Config{}, testIdentity(), cp, nil, nil)
a.runTask(context.Background(), testAssign())
st := cp.taskStates()
@@ -252,7 +252,7 @@ func TestRunTask_GPUCarriedToRunSpec(t *testing.T) {
func TestRunTask_GPUPinsAndReportsUUIDs(t *testing.T) {
cp := &fakeCP{}
rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
a := newAgent(config{}, testIdentityWithGPUs("GPU-aaa", "GPU-bbb"), cp, rt, nil)
a := newAgent(Config{}, testIdentityWithGPUs("GPU-aaa", "GPU-bbb"), cp, rt, nil)
as := testAssign()
as.Containers = append(as.Containers, bus.AssignContainer{
Name: "trainer", Image: "registry/trainer:1", GPU: 1,
@@ -296,7 +296,7 @@ func TestRunTask_GPUPinsAndReportsUUIDs(t *testing.T) {
func TestRunTask_GPUIDsReachRunSpec(t *testing.T) {
cp := &fakeCP{}
rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
a := newAgent(config{}, testIdentityWithGPUs("GPU-aaa", "GPU-bbb"), cp, rt, nil)
a := newAgent(Config{}, testIdentityWithGPUs("GPU-aaa", "GPU-bbb"), cp, rt, nil)
as := testAssign()
as.Containers = append(as.Containers, bus.AssignContainer{
Name: "trainer", Image: "registry/trainer:1", GPU: 1,
@@ -361,7 +361,7 @@ func TestStopTask_ReleasesGPUsBackToLedger(t *testing.T) {
"mulga.ecs.taskID": "t-001", "mulga.ecs.containerName": "trainer",
}},
}}
a := newAgent(config{}, testIdentityWithGPUs("GPU-aaa"), cp, rt, nil)
a := newAgent(Config{}, testIdentityWithGPUs("GPU-aaa"), cp, rt, nil)
if _, err := a.gpu.Pin(gpuKey("t-001", "trainer"), 1); err != nil {
t.Fatalf("seed pin: %v", err)
}
@@ -439,7 +439,7 @@ func TestPollAssignments_DispatchesOnceAndAcks(t *testing.T) {
nil, // poll 3: gateway dropped it after the ack
}}
rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
a := newAgent(Config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
ctx, cancel := context.WithCancel(context.Background())
go a.pollAssignments(ctx, map[string]bool{})
@@ -538,7 +538,7 @@ func TestPollAssignments_StopReapsAndSuppressesAssign(t *testing.T) {
rt := &ctrruntime.FakePuller{Containers: []ctrruntime.Container{
{ID: "t-001-web", Running: true, Labels: map[string]string{"mulga.ecs.taskID": "t-001"}},
}}
a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
a := newAgent(Config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
ctx, cancel := context.WithCancel(context.Background())
go a.pollAssignments(ctx, map[string]bool{})
@@ -593,7 +593,7 @@ func countStopAcks(acks [][]string, taskID string) int {
func TestPollAssignments_UnperformableStopIsNotAckedAndRetried(t *testing.T) {
cp := &fakeCP{stopPending: []bus.StopDirective{{TaskID: "t-001", Reason: "bye"}}}
rt := &ctrruntime.FakePuller{ListErr: errors.New("containerd is down")}
a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
a := newAgent(Config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
@@ -628,7 +628,7 @@ func TestPollAssignments_RecoveredStopIsAckedExactlyOnce(t *testing.T) {
{ID: "t-001-web", Running: true, Labels: map[string]string{"mulga.ecs.taskID": "t-001"}},
},
}
a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
a := newAgent(Config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
@@ -667,7 +667,7 @@ func TestPollAssignments_StoppedTaskIsNotRerunByRedeliveredAssign(t *testing.T)
rt := &ctrruntime.FakePuller{Containers: []ctrruntime.Container{
{ID: "t-001-web", Running: true, Labels: map[string]string{"mulga.ecs.taskID": "t-001"}},
}}
a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
a := newAgent(Config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
@@ -696,7 +696,7 @@ func TestPollAssignments_StoppedTaskIsNotRerunByRedeliveredAssign(t *testing.T)
// stopTask with no runtime is a no-op: nothing to reap, no state reported.
func TestStopTask_NilRunnerNoOp(t *testing.T) {
cp := &fakeCP{}
a := newAgent(config{}, testIdentity(), cp, nil, nil)
a := newAgent(Config{}, testIdentity(), cp, nil, nil)
a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "x"})
if len(cp.taskStates()) != 0 {
t.Fatalf("nil runner should report no state, got %+v", cp.taskStates())
@@ -807,7 +807,7 @@ func (sentinelResolver) Authorize(context.Context, string) (string, string, stri
// exec-role resolver is built.
func TestPullResolver(t *testing.T) {
fake := sentinelResolver{}
a := newAgent(config{Region: "us-east-1"}, testIdentity(), &fakeCP{}, nil, fake)
a := newAgent(Config{Region: "us-east-1"}, testIdentity(), &fakeCP{}, nil, fake)
if got := a.pullResolver(&bus.Assign{TaskID: "t1"}); got != fake {
t.Fatalf("no execution role: want instance resolver fallback")
+26
View File
@@ -0,0 +1,26 @@
package agent
import (
"path/filepath"
"testing"
)
// The build version is injected into the binary's main package and handed in
// through Config; New must carry it into the identity the register message sends.
func TestNewReportsConfiguredAgentVersion(t *testing.T) {
srv := metadataStub(t, true)
a, err := New(Config{
IMDSBase: srv.URL + "/latest",
GatewayURL: "https://127.0.0.1:1",
Region: "us-east-1",
ClusterName: "default",
ContainerdSocket: filepath.Join(t.TempDir(), "absent.sock"),
AgentVersion: "1.2.3-test",
})
if err != nil {
t.Fatalf("New: %v", err)
}
if a.id.AgentVersion != "1.2.3-test" {
t.Errorf("identity AgentVersion = %q, want the configured build version", a.id.AgentVersion)
}
}