diff --git a/cmd/ecs-agent/agent.go b/cmd/ecs-agent/agent.go index 1d5703104..dcc69c66b 100644 --- a/cmd/ecs-agent/agent.go +++ b/cmd/ecs-agent/agent.go @@ -30,10 +30,13 @@ type Agent struct { cfg config id identity cp controlPlane - puller ctrruntime.ImagePuller - runner ctrruntime.Runner resolver ctrruntime.Resolver + // rt holds the container runtime, which is resolved by a loop rather than + // once at boot. Empty until the containerd socket answers, and emptied again + // if it stops answering. + rt *runtimeGate + reg *registrar hb *heartbeater @@ -48,26 +51,34 @@ type Agent struct { // nvidia-smi discovery. gpu *gpuLedger + // dialRuntime connects to the container runtime. nil in unit tests, which + // drive the gate directly. + dialRuntime func() (ctrruntime.Runtime, error) + closers []func() error } // newAgent assembles an Agent from already-built seams. Tests use this directly -// with fakes; New builds the production seams and delegates here. runner may be -// nil when containerd is unavailable; the assign path then reports the task -// STOPPED rather than crashing. -func newAgent(cfg config, id identity, cp controlPlane, puller ctrruntime.ImagePuller, runner ctrruntime.Runner, resolver ctrruntime.Resolver) *Agent { - return &Agent{ +// 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 { + gate := newRuntimeGate(rt) + a := &Agent{ cfg: cfg, id: id, cp: cp, - puller: puller, - runner: runner, resolver: resolver, - reg: newRegistrar(cp, id), - hb: newHeartbeater(cp, id, cfg.Heartbeat), + rt: gate, netns: newTaskNetns(execNetRunner{}), gpu: newGPULedger(id.Capacity.GPUIDs), } + // Registration reports the host's capacity only while a runtime is there to + // use it, so an agent that cannot run a container advertises none and takes + // no placement. + a.reg = newRegistrar(cp, id, gate.available) + a.hb = newHeartbeater(cp, id, cfg.Heartbeat, gate.available) + return a } // New builds an Agent from config: it resolves the host identity from IMDS, @@ -98,13 +109,12 @@ func New(cfg config) (*Agent, error) { creds := credentials.NewIMDSProvider(imdsClient, cfg.IMDSBase) resolver := newLazyECRResolver(creds, cfg.Region, cfg.GatewayURL, cfg.GatewayCA) - var puller ctrruntime.ImagePuller - var runner ctrruntime.Runner - if rt, perr := ctrruntime.New(cfg.ContainerdSocket); perr != nil { - slog.Warn("ecs-agent: containerd unavailable at boot, image pulls disabled", "err", perr) - } else { - puller = rt - runner = rt + // A dial that fails here is not fatal and not final: the node starts the + // container daemon and the agent back to back, so the socket is often not + // accepting yet. Run's watch loop keeps dialling. + rt, perr := ctrruntime.New(cfg.ContainerdSocket) + if perr != nil { + slog.Warn("ecs-agent: containerd unavailable at boot, will keep retrying", "err", perr) } cp, err := newGatewayControlPlane(cfg, creds) @@ -112,12 +122,10 @@ func New(cfg config) (*Agent, error) { return nil, fmt.Errorf("build gateway control-plane: %w", err) } - a := newAgent(cfg, id, cp, puller, runner, resolver) + a := newAgent(cfg, id, cp, rt, resolver) a.cred = newCredEndpoint(creds, cfg.Region, cfg.GatewayURL, cfg.GatewayCA, cfg.CredEndpointIP, cfg.CredEndpointPort, execNetRunner{}) - if puller != nil { - a.closers = append(a.closers, puller.Close) - } + a.dialRuntime = func() (ctrruntime.Runtime, error) { return ctrruntime.New(cfg.ContainerdSocket) } return a, nil } @@ -138,6 +146,12 @@ func (a *Agent) Run(ctx context.Context) error { go a.hb.Run(ctx) + // Keeps the runtime current for the life of the process: dials while there + // is none, and drops one that stops answering. + if a.dialRuntime != nil { + go a.rt.watch(ctx, a.dialRuntime, a.cfg.RuntimeRetry) + } + // Re-adopt containers still running from before this restart, then poll with // their tasks pre-seeded so re-delivered assignments are acked, not re-run. go a.pollAssignments(ctx, a.reconcile(ctx)) @@ -157,6 +171,7 @@ func (a *Agent) Stop() error { firstErr = err } } + a.rt.clear() return firstErr } diff --git a/cmd/ecs-agent/agent_test.go b/cmd/ecs-agent/agent_test.go index 1c035dddd..ab8c6048a 100644 --- a/cmd/ecs-agent/agent_test.go +++ b/cmd/ecs-agent/agent_test.go @@ -12,7 +12,7 @@ func TestAgent_RunRegistersThenStopsOnContext(t *testing.T) { cp := &fakeCP{} cfg := config{Heartbeat: 5 * time.Millisecond, PollInterval: 5 * time.Millisecond} puller := &ctrruntime.FakePuller{} - a := newAgent(cfg, testIdentity(), cp, puller, puller, nil) + a := newAgent(cfg, testIdentity(), cp, puller, nil) a.closers = append(a.closers, puller.Close) ctx, cancel := context.WithCancel(context.Background()) diff --git a/cmd/ecs-agent/config.go b/cmd/ecs-agent/config.go index 95c7855bc..dac91e554 100644 --- a/cmd/ecs-agent/config.go +++ b/cmd/ecs-agent/config.go @@ -33,6 +33,7 @@ type config struct { ContainerdSocket string Heartbeat time.Duration PollInterval time.Duration + RuntimeRetry time.Duration CredEndpointIP string CredEndpointPort int } @@ -75,6 +76,12 @@ func loadConfig(envFile string) config { cfg.PollInterval = d } } + cfg.RuntimeRetry = defaultRuntimeRetry + if v := get("ECS_RUNTIME_RETRY"); v != "" { + if d, err := time.ParseDuration(v); err == nil && d > 0 { + cfg.RuntimeRetry = d + } + } cfg.CredEndpointIP = get("ECS_CRED_ENDPOINT_IP") if cfg.CredEndpointIP == "" { cfg.CredEndpointIP = defaultCredEndpointIP diff --git a/cmd/ecs-agent/heartbeat.go b/cmd/ecs-agent/heartbeat.go index dc8e67827..e2956a9b0 100644 --- a/cmd/ecs-agent/heartbeat.go +++ b/cmd/ecs-agent/heartbeat.go @@ -14,18 +14,23 @@ type heartbeater struct { cp controlPlane id identity interval time.Duration + // runtimeReady reports whether a container runtime is resolved, so a beat + // advertises capacity only while there is something to run it with. nil + // means always ready. + runtimeReady func() bool } -func newHeartbeater(cp controlPlane, id identity, interval time.Duration) *heartbeater { +func newHeartbeater(cp controlPlane, id identity, interval time.Duration, runtimeReady func() bool) *heartbeater { if interval <= 0 { interval = defaultHeartbeat } - return &heartbeater{cp: cp, id: id, interval: interval} + return &heartbeater{cp: cp, id: id, interval: interval, runtimeReady: runtimeReady} } // beat re-registers the instance once, refreshing its LastSeen. func (h *heartbeater) beat() error { - return h.cp.Register(h.id) + ready := h.runtimeReady == nil || h.runtimeReady() + return h.cp.Register(advertisedIdentity(h.id, ready)) } // Run beats every interval until ctx is cancelled. A failed beat is logged and diff --git a/cmd/ecs-agent/reconcile.go b/cmd/ecs-agent/reconcile.go index 55bedaa04..fb9166ce8 100644 --- a/cmd/ecs-agent/reconcile.go +++ b/cmd/ecs-agent/reconcile.go @@ -24,10 +24,11 @@ import ( // agent falls back to its pre-reconcile behaviour (poll + run). func (a *Agent) reconcile(ctx context.Context) map[string]bool { adopted := map[string]bool{} - if a.runner == nil { + runner := a.rt.get() + if runner == nil { return adopted } - containers, err := a.runner.List(ctx) + containers, err := runner.List(ctx) if err != nil { slog.Warn("ecs-agent: reconcile list failed", "err", err) return adopted @@ -96,9 +97,13 @@ func (a *Agent) reconcile(ctx context.Context) map[string]bool { // is STOPPED, so the scheduler stops counting a task this instance is not // running and its service can place a replacement. func (a *Agent) reportStoppedTask(ctx context.Context, as *bus.Assign, dead []ctrruntime.Container) { + runner := a.rt.get() statuses := make([]bus.ContainerStatus, 0, len(dead)) for _, c := range dead { - if err := a.runner.Remove(ctx, c.ID); err != nil { + if runner == nil { + break + } + if err := runner.Remove(ctx, c.ID); err != nil { slog.Warn("ecs-agent: reconcile remove failed", "task", as.TaskID, "container", c.ID, "err", err) } statuses = append(statuses, bus.ContainerStatus{ diff --git a/cmd/ecs-agent/reconcile_test.go b/cmd/ecs-agent/reconcile_test.go index cabb9a291..104409456 100644 --- a/cmd/ecs-agent/reconcile_test.go +++ b/cmd/ecs-agent/reconcile_test.go @@ -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, 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, 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, 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, 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, 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, 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, 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) } diff --git a/cmd/ecs-agent/registration.go b/cmd/ecs-agent/registration.go index 694627072..8c8cab11c 100644 --- a/cmd/ecs-agent/registration.go +++ b/cmd/ecs-agent/registration.go @@ -39,20 +39,39 @@ func instanceAttributes(id identity) []*ecs.Attribute { return attrs } +// advertisedIdentity is the identity to register with, given whether a +// container runtime is available. A host with no runtime can run nothing, so it +// advertises no capacity: the scheduler places nothing on it, and the control +// plane reads the zero as an involuntary drain. The identity is otherwise +// unchanged, so the instance stays visible rather than disappearing. +func advertisedIdentity(id identity, runtimeReady bool) identity { + if runtimeReady { + return id + } + id.Capacity = bus.InstanceCapacity{} + return id +} + // registrar registers the host as a container instance through the gateway. type registrar struct { cp controlPlane id identity + // runtimeReady reports whether a container runtime is resolved. nil means + // always ready, which is what a test with no gate wants. + runtimeReady func() bool } -func newRegistrar(cp controlPlane, id identity) *registrar { - return ®istrar{cp: cp, id: id} +func newRegistrar(cp controlPlane, id identity, runtimeReady func() bool) *registrar { + return ®istrar{cp: cp, id: id, runtimeReady: runtimeReady} } +// ready reports the runtime state, defaulting to ready when no gate is wired. +func (r *registrar) ready() bool { return r.runtimeReady == nil || r.runtimeReady() } + // Register calls the gateway's RegisterContainerInstance. The scheduler records // (or refreshes) the container instance on receipt. func (r *registrar) Register() error { - if err := r.cp.Register(r.id); err != nil { + if err := r.cp.Register(advertisedIdentity(r.id, r.ready())); err != nil { return fmt.Errorf("register instance %s: %w", r.id.InstanceID, err) } return nil diff --git a/cmd/ecs-agent/registration_test.go b/cmd/ecs-agent/registration_test.go index dc2125230..2135061b0 100644 --- a/cmd/ecs-agent/registration_test.go +++ b/cmd/ecs-agent/registration_test.go @@ -18,15 +18,16 @@ import ( type fakeCP struct { mu sync.Mutex - registers int - states []bus.TaskState - gpuReports []gpuReport - pollAcks [][]string - pollStopAcks [][]string - pollReplies [][]bus.Assign - stopReplies [][]bus.StopDirective - stopPending []bus.StopDirective - pollCalls int + registers int + lastRegistered identity + states []bus.TaskState + gpuReports []gpuReport + pollAcks [][]string + pollStopAcks [][]string + pollReplies [][]bus.Assign + stopReplies [][]bus.StopDirective + stopPending []bus.StopDirective + pollCalls int registerErr bool submitErr bool @@ -41,16 +42,24 @@ type gpuReport struct { var _ controlPlane = (*fakeCP)(nil) -func (f *fakeCP) Register(_ identity) error { +func (f *fakeCP) Register(id identity) error { f.mu.Lock() defer f.mu.Unlock() if f.registerErr { return errors.New("register boom") } f.registers++ + f.lastRegistered = id return nil } +// registered returns the identity of the most recent Register call. +func (f *fakeCP) registered() identity { + f.mu.Lock() + defer f.mu.Unlock() + return f.lastRegistered +} + func (f *fakeCP) SubmitTaskState(st bus.TaskState) error { f.mu.Lock() defer f.mu.Unlock() @@ -150,7 +159,7 @@ func testIdentityWithGPUs(uuids ...string) identity { func TestRegistrar_Register(t *testing.T) { cp := &fakeCP{} - if err := newRegistrar(cp, testIdentity()).Register(); err != nil { + if err := newRegistrar(cp, testIdentity(), nil).Register(); err != nil { t.Fatalf("Register: %v", err) } if cp.registerCount() != 1 { @@ -160,14 +169,14 @@ func TestRegistrar_Register(t *testing.T) { func TestRegistrar_RegisterError(t *testing.T) { cp := &fakeCP{registerErr: true} - if err := newRegistrar(cp, testIdentity()).Register(); err == nil { + if err := newRegistrar(cp, testIdentity(), nil).Register(); err == nil { t.Fatal("expected register error") } } func TestHeartbeater_Beat(t *testing.T) { cp := &fakeCP{} - if err := newHeartbeater(cp, testIdentity(), time.Second).beat(); err != nil { + if err := newHeartbeater(cp, testIdentity(), time.Second, nil).beat(); err != nil { t.Fatalf("beat: %v", err) } if cp.registerCount() != 1 { @@ -176,7 +185,7 @@ func TestHeartbeater_Beat(t *testing.T) { } func TestHeartbeater_DefaultInterval(t *testing.T) { - h := newHeartbeater(&fakeCP{}, testIdentity(), 0) + h := newHeartbeater(&fakeCP{}, testIdentity(), 0, nil) if h.interval != defaultHeartbeat { t.Errorf("interval = %v, want default", h.interval) } @@ -184,7 +193,7 @@ func TestHeartbeater_DefaultInterval(t *testing.T) { func TestHeartbeater_RunStopsOnContext(t *testing.T) { cp := &fakeCP{} - h := newHeartbeater(cp, testIdentity(), 5*time.Millisecond) + h := newHeartbeater(cp, testIdentity(), 5*time.Millisecond, nil) ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{}) @@ -201,3 +210,48 @@ func TestHeartbeater_RunStopsOnContext(t *testing.T) { t.Error("expected at least one re-register before cancel") } } + +// TestRegistrar_NoRuntimeAdvertisesNoCapacity is the placement half of the boot +// bug: an agent with no container runtime kept registering its full CPU and +// memory, so the scheduler placed tasks it could only fail to start. Capacity is +// advertised only while a runtime is there to use it. +func TestRegistrar_NoRuntimeAdvertisesNoCapacity(t *testing.T) { + cp := &fakeCP{} + ready := false + r := newRegistrar(cp, testIdentity(), func() bool { return ready }) + + if err := r.Register(); err != nil { + t.Fatalf("register: %v", err) + } + if got := cp.registered().Capacity; got.CPU != 0 || got.MemoryMiB != 0 { + t.Fatalf("advertised capacity with no runtime: cpu=%d memory=%d", got.CPU, got.MemoryMiB) + } + + ready = true + if err := r.Register(); err != nil { + t.Fatalf("register: %v", err) + } + want := testIdentity().Capacity + if got := cp.registered().Capacity; got.CPU != want.CPU || got.MemoryMiB != want.MemoryMiB { + t.Fatalf("capacity not restored once the runtime resolved: cpu=%d memory=%d", got.CPU, got.MemoryMiB) + } +} + +// TestHeartbeater_NoRuntimeAdvertisesNoCapacity is the same rule on the beat, +// which is how a runtime lost after boot reaches the control plane. +func TestHeartbeater_NoRuntimeAdvertisesNoCapacity(t *testing.T) { + cp := &fakeCP{} + h := newHeartbeater(cp, testIdentity(), time.Second, func() bool { return false }) + + if err := h.beat(); err != nil { + t.Fatalf("beat: %v", err) + } + if got := cp.registered().Capacity; got.CPU != 0 || got.MemoryMiB != 0 { + t.Fatalf("beat advertised capacity with no runtime: cpu=%d memory=%d", got.CPU, got.MemoryMiB) + } + // The instance still identifies itself, so it stays visible as drained rather + // than vanishing from the cluster. + if cp.registered().InstanceID != testIdentity().InstanceID { + t.Error("beat dropped the instance identity") + } +} diff --git a/cmd/ecs-agent/runtimegate.go b/cmd/ecs-agent/runtimegate.go new file mode 100644 index 000000000..dc0454062 --- /dev/null +++ b/cmd/ecs-agent/runtimegate.go @@ -0,0 +1,116 @@ +package main + +import ( + "context" + "log/slog" + "sync" + "time" + + ctrruntime "github.com/mulgadc/spinifex/cmd/ecs-agent/runtime" +) + +// runtimeProbeTimeout bounds the liveness probe against a runtime the gate is +// already holding, so a wedged socket cannot stall the watch loop. +const runtimeProbeTimeout = 5 * time.Second + +// defaultRuntimeRetry is how often the gate dials a runtime it does not have, +// and probes one it does. The node starts containerd and the agent back to +// back, so the first few dials are expected to fail. +const defaultRuntimeRetry = 10 * time.Second + +// runtimeGate holds the container runtime the agent has resolved, or nothing +// when the socket has not answered. The runtime is resolved by a loop rather +// than once at boot: the node's service manager reports the container daemon +// started when the process spawned, not when its socket began accepting, so a +// dial that fails in the first second of boot is a transient condition. +// +// Every read of the runtime goes through the gate, so a runtime that arrives +// after boot is picked up on the next task without restarting the agent. +type runtimeGate struct { + mu sync.RWMutex + rt ctrruntime.Runtime +} + +// newRuntimeGate returns a gate holding rt, which may be nil. +func newRuntimeGate(rt ctrruntime.Runtime) *runtimeGate { + return &runtimeGate{rt: rt} +} + +// get returns the current runtime, or nil when none is resolved. +func (g *runtimeGate) get() ctrruntime.Runtime { + g.mu.RLock() + defer g.mu.RUnlock() + return g.rt +} + +// available reports whether a runtime is resolved. It is what registration +// consults to decide whether the host has any capacity to advertise. +func (g *runtimeGate) available() bool { return g.get() != nil } + +// set replaces the held runtime, closing the one it displaces. +func (g *runtimeGate) set(rt ctrruntime.Runtime) { + g.mu.Lock() + old := g.rt + g.rt = rt + g.mu.Unlock() + if old != nil && old != rt { + _ = old.Close() + } +} + +// clear drops and closes the held runtime, returning whether it held one. +func (g *runtimeGate) clear() bool { + g.mu.Lock() + old := g.rt + g.rt = nil + g.mu.Unlock() + if old == nil { + return false + } + _ = old.Close() + return true +} + +// watch keeps the gate current until ctx is cancelled: it dials while the gate +// is empty, and probes while it is full, clearing the gate when the probe fails +// so the next tick re-dials. A containerd stopped and restarted underneath a +// running agent is therefore picked up the same way a slow boot is. +func (g *runtimeGate) watch(ctx context.Context, dial func() (ctrruntime.Runtime, error), interval time.Duration) { + if interval <= 0 { + interval = defaultRuntimeRetry + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + g.step(ctx, dial) + } + } +} + +// step performs one pass of the watch loop: dial when empty, probe when full. +func (g *runtimeGate) step(ctx context.Context, dial func() (ctrruntime.Runtime, error)) { + if rt := g.get(); rt != nil { + probeCtx, cancel := context.WithTimeout(ctx, runtimeProbeTimeout) + // A container list is what reconcile already asks the runtime at boot and + // the cheapest thing it answers, so it doubles as the liveness probe. + _, err := rt.List(probeCtx) + cancel() + if err != nil && ctx.Err() == nil { + slog.Warn("ecs-agent: container runtime stopped answering, will re-dial", "err", err) + g.clear() + } + return + } + + rt, err := dial() + if err != nil { + slog.Debug("ecs-agent: container runtime still unavailable", "err", err) + return + } + slog.Info("ecs-agent: container runtime available") + g.set(rt) +} diff --git a/cmd/ecs-agent/runtimegate_test.go b/cmd/ecs-agent/runtimegate_test.go new file mode 100644 index 000000000..6f639061b --- /dev/null +++ b/cmd/ecs-agent/runtimegate_test.go @@ -0,0 +1,122 @@ +package main + +import ( + "context" + "errors" + "testing" + + ctrruntime "github.com/mulgadc/spinifex/cmd/ecs-agent/runtime" +) + +// TestRuntimeGate_DialsUntilTheSocketAnswers is the boot case the agent used to +// lose: the node starts containerd and the agent together, so the first dials +// fail. Resolving once left the agent runtime-less for the life of the process. +func TestRuntimeGate_DialsUntilTheSocketAnswers(t *testing.T) { + g := newRuntimeGate(nil) + if g.available() { + t.Fatal("empty gate reported a runtime") + } + + rt := &ctrruntime.FakePuller{} + dials := 0 + dial := func() (ctrruntime.Runtime, error) { + dials++ + if dials < 3 { + return nil, errors.New("dial containerd: no such file or directory") + } + return rt, nil + } + + for range 2 { + g.step(t.Context(), dial) + if g.available() { + t.Fatalf("gate resolved a runtime after %d failed dials", dials) + } + } + + g.step(t.Context(), dial) + if g.get() != rt { + t.Fatalf("gate did not pick up the runtime once the socket answered (dials=%d)", dials) + } +} + +// TestRuntimeGate_DropsARuntimeThatStopsAnswering covers a containerd restarted +// underneath a live agent: the held runtime stops answering, so the gate clears +// it and the next pass re-dials rather than handing out a dead client forever. +func TestRuntimeGate_DropsARuntimeThatStopsAnswering(t *testing.T) { + dead := &ctrruntime.FakePuller{ListErr: errors.New("connection refused")} + fresh := &ctrruntime.FakePuller{} + g := newRuntimeGate(dead) + + dial := func() (ctrruntime.Runtime, error) { return fresh, nil } + + g.step(t.Context(), dial) + if g.available() { + t.Fatal("gate kept a runtime whose probe failed") + } + if !dead.Closed { + t.Error("displaced runtime was not closed") + } + + g.step(t.Context(), dial) + if g.get() != fresh { + t.Fatal("gate did not re-dial after clearing") + } +} + +// TestRuntimeGate_KeepsAHealthyRuntime guards the other direction: a probe that +// succeeds must not churn the runtime, or every pass would drop a working client. +func TestRuntimeGate_KeepsAHealthyRuntime(t *testing.T) { + rt := &ctrruntime.FakePuller{} + g := newRuntimeGate(rt) + + dial := func() (ctrruntime.Runtime, error) { + t.Error("dialled while holding a healthy runtime") + return nil, errors.New("unexpected dial") + } + g.step(t.Context(), dial) + + if g.get() != rt { + t.Fatal("gate dropped a runtime whose probe succeeded") + } + if rt.Closed { + t.Error("healthy runtime was closed") + } +} + +// TestRuntimeGate_ClearIsIdempotent keeps Stop safe to call on an agent that +// never resolved a runtime. +func TestRuntimeGate_ClearIsIdempotent(t *testing.T) { + g := newRuntimeGate(nil) + if g.clear() { + t.Error("clear reported dropping a runtime the gate never held") + } + + rt := &ctrruntime.FakePuller{} + g.set(rt) + if !g.clear() { + t.Error("clear did not report dropping the held runtime") + } + if !rt.Closed { + t.Error("cleared runtime was not closed") + } + if g.clear() { + t.Error("second clear reported dropping a runtime") + } +} + +// TestRuntimeGate_WatchStopsWithContext keeps the watch loop from outliving the +// agent it belongs to. +func TestRuntimeGate_WatchStopsWithContext(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + g := newRuntimeGate(nil) + + done := make(chan struct{}) + go func() { + g.watch(ctx, func() (ctrruntime.Runtime, error) { return nil, errors.New("no socket") }, 0) + close(done) + }() + + cancel() + <-done +} diff --git a/cmd/ecs-agent/taskrun.go b/cmd/ecs-agent/taskrun.go index 6376b6684..cc52e91a0 100644 --- a/cmd/ecs-agent/taskrun.go +++ b/cmd/ecs-agent/taskrun.go @@ -128,11 +128,12 @@ func (l *stopLedger) seen(taskID string) bool { // scheduler releases its capacity. Reports whether it got as far as saying // STOPPED: a caller may only ack the directive when it did. func (a *Agent) stopTask(ctx context.Context, sd bus.StopDirective) bool { - if a.runner == nil { + runner := a.rt.get() + if runner == nil { slog.Warn("ecs-agent: stop deferred, no container runtime", "task", sd.TaskID) return false } - containers, err := a.runner.List(ctx) + containers, err := runner.List(ctx) if err != nil { slog.Warn("ecs-agent: stop list failed", "task", sd.TaskID, "err", err) return false @@ -142,7 +143,7 @@ func (a *Agent) stopTask(ctx context.Context, sd bus.StopDirective) bool { if c.Labels[labelTaskID] != sd.TaskID { continue } - if rerr := a.runner.Stop(ctx, c.ID, containerStopTimeout(c.Labels)); rerr != nil { + if rerr := runner.Stop(ctx, c.ID, containerStopTimeout(c.Labels)); rerr != nil { slog.Warn("ecs-agent: stop failed", "task", sd.TaskID, "container", c.ID, "err", rerr) } statuses = append(statuses, bus.ContainerStatus{ @@ -162,7 +163,8 @@ func (a *Agent) stopTask(ctx context.Context, sd bus.StopDirective) bool { // state on the bus. A missing runtime or any per-container failure reports the // task STOPPED with a reason; success reports RUNNING and waits for exit. func (a *Agent) runTask(ctx context.Context, as *bus.Assign) { - if a.puller == nil || a.runner == nil { + rt := a.rt.get() + if rt == nil { a.reportTaskState(as, bus.TaskStatusStopped, "containerd unavailable on agent", nil) return } @@ -184,7 +186,7 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) { // container's time-to-RUNNING from the same starting point. runCtx, cancel := containerStartCtx(ctx, c.StartTimeout) - if _, err := a.puller.Pull(runCtx, ctrruntime.PullSpec{Ref: c.Image}, resolver); err != nil { + if _, err := rt.Pull(runCtx, ctrruntime.PullSpec{Ref: c.Image}, resolver); err != nil { cancel() slog.Error("ecs-agent: pull failed", "task", as.TaskID, "image", c.Image, "err", err) a.teardownTaskNetns(as) @@ -215,7 +217,7 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) { CapAdd: c.CapAdd, CapDrop: c.CapDrop, } - id, err := a.runner.Run(runCtx, cid, spec) + id, err := rt.Run(runCtx, cid, spec) cancel() if err != nil { slog.Error("ecs-agent: run failed", "task", as.TaskID, "container", c.Name, "err", err) @@ -236,7 +238,12 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) { // waitContainer blocks until a container exits, then reports the task STOPPED // with the exit code. v1 stops the whole task when its first container exits. func (a *Agent) waitContainer(ctx context.Context, as *bus.Assign, name, containerID string) { - status, err := a.runner.Wait(ctx, containerID) + runner := a.rt.get() + if runner == nil { + slog.Warn("ecs-agent: cannot wait on container, no container runtime", "task", as.TaskID, "container", name) + return + } + status, err := runner.Wait(ctx, containerID) if err != nil { slog.Warn("ecs-agent: wait container failed", "task", as.TaskID, "container", name, "err", err) return diff --git a/cmd/ecs-agent/taskrun_test.go b/cmd/ecs-agent/taskrun_test.go index a783b5642..52045c613 100644 --- a/cmd/ecs-agent/taskrun_test.go +++ b/cmd/ecs-agent/taskrun_test.go @@ -25,13 +25,13 @@ func testAssign() *bus.Assign { } func newRunAgent(cp controlPlane, rt *ctrruntime.FakePuller) *Agent { - return newAgent(config{}, testIdentity(), cp, rt, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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") diff --git a/spinifex/handlers/ecs/deployments.go b/spinifex/handlers/ecs/deployments.go index dbf7b5d2a..8617c45b4 100644 --- a/spinifex/handlers/ecs/deployments.go +++ b/spinifex/handlers/ecs/deployments.go @@ -195,6 +195,26 @@ func updateRolloutState(svc *ServiceRecord, primary *Deployment, desired int) { primary.RolloutState = RolloutStateInProgress } +// launchBackoff returns how long to hold off the next launch for a deployment +// with failures behind it, doubling from a base and capped. It reads the same +// counter and the same threshold as the circuit breaker, but applies whether or +// not the breaker is enabled: AWS leaves it off by default, so an unbraked +// service relaunches a task that cannot start on every pass, forever. +// +// Below the threshold there is no hold, so an enabled breaker still sees its +// failures at full speed and trips exactly when it did before. +func launchBackoff(failures int) time.Duration { + if failures < circuitBreakerFailureThreshold { + return 0 + } + // Bounded before the shift so a long-failing deployment cannot overflow it. + d := launchBackoffBase << min(failures-circuitBreakerFailureThreshold, launchBackoffMaxShift) + if d > launchBackoffCap { + return launchBackoffCap + } + return d +} + // tripCircuitBreaker fails a PRIMARY deployment whose task launches have failed // past the threshold when the breaker is enabled, optionally rolling back to the // last-good task definition. Returns true when it acted so the caller can persist diff --git a/spinifex/handlers/ecs/deployments_test.go b/spinifex/handlers/ecs/deployments_test.go index 1a55886b8..2449d14c7 100644 --- a/spinifex/handlers/ecs/deployments_test.go +++ b/spinifex/handlers/ecs/deployments_test.go @@ -3,6 +3,7 @@ package handlers_ecs import ( "context" "testing" + "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ecs" @@ -336,3 +337,73 @@ func TestDeployment_LegacyServiceSynthesizesPrimary(t *testing.T) { assert.Equal(t, "legacy-dep", primary.ID) assert.Equal(t, defaultMinimumHealthyPercent, reloaded.MinimumHealthyPercent) } + +// TestLaunchBackoff_WidensWithFailuresAndCaps is the arithmetic on its own: no +// hold while the circuit breaker is still counting, then doubling, then a cap so +// a long-dead deployment does not wait forever to be retried. +func TestLaunchBackoff_WidensWithFailuresAndCaps(t *testing.T) { + for failures := range circuitBreakerFailureThreshold { + assert.Zero(t, launchBackoff(failures), "held off at %d failures, below the breaker threshold", failures) + } + assert.Equal(t, launchBackoffBase, launchBackoff(circuitBreakerFailureThreshold)) + assert.Equal(t, 2*launchBackoffBase, launchBackoff(circuitBreakerFailureThreshold+1)) + assert.Equal(t, 4*launchBackoffBase, launchBackoff(circuitBreakerFailureThreshold+2)) + assert.Equal(t, launchBackoffCap, launchBackoff(circuitBreakerFailureThreshold+20)) + assert.Equal(t, launchBackoffCap, launchBackoff(1_000_000)) +} + +// A deployment whose tasks keep dying on start stops relaunching on every pass. +// AWS leaves the deployment circuit breaker off by default, so without this a +// service whose node cannot run containers churns tasks forever — which is how +// one env accumulated hundreds of STOPPED tasks at one every thirty seconds. +func TestDeployment_RepeatedStartFailuresBackOffWithoutTheCircuitBreaker(t *testing.T) { + svc, _, kv := serviceTestRig(t) + _, err := svc.CreateService(context.Background(), &ecs.CreateServiceInput{ + Cluster: aws.String("web"), ServiceName: aws.String("web"), + TaskDefinition: aws.String("app"), DesiredCount: aws.Int64(1), + }, testAccountID) + require.NoError(t, err) + + // Fail every task the reconciler launches. The breaker is off, so nothing + // else brakes this. + for range circuitBreakerFailureThreshold { + failPending(t, svc, kv, "web", "web") + require.NoError(t, svc.reconcileService(context.Background(), kv, testAccountID, reloadService(t, kv, "web", "web"))) + } + + rec := reloadService(t, kv, "web", "web") + primary := rec.primaryDeployment() + require.NotNil(t, primary) + require.GreaterOrEqual(t, primary.FailedTasks, circuitBreakerFailureThreshold) + require.False(t, primary.NextLaunchAt.IsZero(), "no backoff recorded after repeated start failures") + assert.True(t, primary.NextLaunchAt.After(time.Now().UTC()), "backoff deadline is already in the past") + + // The service is below its desired count and would otherwise launch, but the + // deadline holds the pass off. + failPending(t, svc, kv, "web", "web") + require.NoError(t, svc.reconcileService(context.Background(), kv, testAccountID, reloadService(t, kv, "web", "web"))) + assert.Zero(t, liveServiceTasks(t, svc, kv, "web", "web"), "reconcile launched a task while the deployment was backed off") + + // Once it passes, launching resumes: the backoff is a brake, not a stop. + rec = reloadService(t, kv, "web", "web") + rec.primaryDeployment().NextLaunchAt = time.Now().UTC().Add(-time.Second) + require.NoError(t, putJSON(t.Context(), kv, ServiceKey("web", "web"), rec)) + require.NoError(t, svc.reconcileService(context.Background(), kv, testAccountID, reloadService(t, kv, "web", "web"))) + + assert.Positive(t, liveServiceTasks(t, svc, kv, "web", "web"), "backoff never released the deployment") +} + +// liveServiceTasks counts a service's tasks that have not stopped, which is what +// the reconciler is trying to hold at the desired count. +func liveServiceTasks(t *testing.T, svc *Service, kv jetstream.KeyValue, cluster, name string) int { + t.Helper() + tasks, err := svc.listServiceTasks(t.Context(), kv, cluster, name) + require.NoError(t, err) + n := 0 + for i := range tasks { + if tasks[i].LastStatus != TaskStatusStopped { + n++ + } + } + return n +} diff --git a/spinifex/handlers/ecs/instance_recovery_test.go b/spinifex/handlers/ecs/instance_recovery_test.go index 332e48982..7ab721a02 100644 --- a/spinifex/handlers/ecs/instance_recovery_test.go +++ b/spinifex/handlers/ecs/instance_recovery_test.go @@ -136,3 +136,89 @@ func TestRegister_DoesNotUndoOperatorDrain(t *testing.T) { after := instanceStatus(t, svc, "web", "i-1") assert.Equal(t, InstanceStatusDraining, after.Status, "operator drain must survive agent re-register") } + +// An agent that cannot resolve a container runtime registers no capacity. That +// is the same condition as an agent that stopped answering, so the instance is +// drained involuntarily and the scheduler stops placing tasks it could only +// fail to start. +func TestRegister_NoCapacityDrainsTheInstance(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.CreateCluster(context.Background(), &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerTaskDef(t, svc, "app", 128, 256) + registerInstance(t, svc, "web", "i-1", 1024, 2048) + + // The runtime goes away: the next registration reports nothing to run with. + registerInstance(t, svc, "web", "i-1", 0, 0) + + drained := instanceStatus(t, svc, "web", "i-1") + assert.Equal(t, InstanceStatusDraining, drained.Status) + assert.True(t, drained.Reaped, "a runtime-less instance is an involuntary drain, not an operator one") + assert.Zero(t, drained.TotalCPU) + assert.Zero(t, drained.TotalMemoryMiB) + + out, err := svc.RunTask(context.Background(), &ecs.RunTaskInput{ + Cluster: aws.String("web"), + TaskDefinition: aws.String("app"), + Count: aws.Int64(1), + }, testAccountID) + require.NoError(t, err) + assert.Empty(t, out.Tasks, "a task was placed on an instance with no runtime") + assert.NotEmpty(t, out.Failures) + + // The instance is visibly drained through the API rather than only in the + // node's log, which is what an operator has to go on. + di, err := svc.DescribeContainerInstances(context.Background(), &ecs.DescribeContainerInstancesInput{ + Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")}, + }, testAccountID) + require.NoError(t, err) + require.Len(t, di.ContainerInstances, 1) + assert.Equal(t, InstanceStatusDraining, aws.StringValue(di.ContainerInstances[0].Status)) + assert.False(t, aws.BoolValue(di.ContainerInstances[0].AgentConnected)) +} + +// The runtime coming back is a registration carrying real capacity again, which +// clears the drain without the agent or the instance being restarted. +func TestRegister_CapacityReturningClearsTheDrain(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.CreateCluster(context.Background(), &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerTaskDef(t, svc, "app", 128, 256) + registerInstance(t, svc, "web", "i-1", 0, 0) + require.Equal(t, InstanceStatusDraining, instanceStatus(t, svc, "web", "i-1").Status) + + registerInstance(t, svc, "web", "i-1", 1024, 2048) + + back := instanceStatus(t, svc, "web", "i-1") + assert.Equal(t, InstanceStatusActive, back.Status) + assert.False(t, back.Reaped) + assert.Equal(t, 1024, back.TotalCPU) + + out, err := svc.RunTask(context.Background(), &ecs.RunTaskInput{ + Cluster: aws.String("web"), + TaskDefinition: aws.String("app"), + Count: aws.Int64(1), + }, testAccountID) + require.NoError(t, err) + assert.Len(t, out.Tasks, 1) +} + +// An operator drain is deliberate, so a capacity-carrying registration must not +// undo it. Only the involuntary drain this path sets is cleared by a register. +func TestRegister_DoesNotClearAnOperatorDrain(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.CreateCluster(context.Background(), &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerInstance(t, svc, "web", "i-1", 1024, 2048) + + _, err = svc.UpdateContainerInstancesState(context.Background(), &ecs.UpdateContainerInstancesStateInput{ + Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")}, + Status: aws.String(InstanceStatusDraining), + }, testAccountID) + require.NoError(t, err) + require.False(t, instanceStatus(t, svc, "web", "i-1").Reaped) + + registerInstance(t, svc, "web", "i-1", 1024, 2048) + + assert.Equal(t, InstanceStatusDraining, instanceStatus(t, svc, "web", "i-1").Status) +} diff --git a/spinifex/handlers/ecs/instances.go b/spinifex/handlers/ecs/instances.go index 555c12a1c..3eb3ec02c 100644 --- a/spinifex/handlers/ecs/instances.go +++ b/spinifex/handlers/ecs/instances.go @@ -35,6 +35,30 @@ func (s *Service) listInstanceRecords(ctx context.Context, kv jetstream.KeyValue return out, nil } +// applyRuntimeCapacity records the capacity a registration reported and derives +// the instance's availability from it. A host reporting no CPU and no memory has +// no container runtime to place on, which is the same condition as an agent that +// stopped answering: it drains involuntarily so the scheduler skips it, and the +// next registration carrying real capacity restores it. An operator drain +// (Reaped false) is never cleared here. +func applyRuntimeCapacity(r *InstanceRecord, cpu, memoryMiB int) { + r.TotalCPU = cpu + r.TotalMemoryMiB = memoryMiB + if cpu == 0 && memoryMiB == 0 { + if r.Status != InstanceStatusDraining { + slog.Warn("ECS: container instance reported no capacity, draining until a runtime is available", + "instance_id", r.InstanceID, "cluster", r.Cluster) + } + r.Status = InstanceStatusDraining + r.Reaped = true + return + } + if r.Status == InstanceStatusDraining && r.Reaped { + r.Status = InstanceStatusActive + r.Reaped = false + } +} + // RegisterContainerInstance is the AWS-API registration path. In 4e the agent // normally registers over the Layer-2 bus; this keeps API parity by writing the // same record shape from an explicit call. @@ -49,12 +73,17 @@ func (s *Service) RegisterContainerInstance(ctx context.Context, input *ecs.Regi return nil, err } rec, err := s.upsertInstance(ctx, kv, accountID, cluster, instanceID, func(r *InstanceRecord) { + // CPU and memory are read together, because it is the pair being zero that + // says the host has no runtime. A registration that omits either leaves the + // previously reported capacity alone. + cpu, memoryMiB := r.TotalCPU, r.TotalMemoryMiB + var sawCPU, sawMemory bool for _, res := range input.TotalResources { switch aws.StringValue(res.Name) { case "CPU": - r.TotalCPU = int(aws.Int64Value(res.IntegerValue)) + cpu, sawCPU = int(aws.Int64Value(res.IntegerValue)), true case "MEMORY": - r.TotalMemoryMiB = int(aws.Int64Value(res.IntegerValue)) + memoryMiB, sawMemory = int(aws.Int64Value(res.IntegerValue)), true case "GPU": // AWS reports GPU as a STRINGSET of device UUIDs; the count is the // capacity, the UUIDs are the placeholder inventory (Epic C3 populates @@ -74,10 +103,12 @@ func (s *Service) RegisterContainerInstance(ctx context.Context, input *ecs.Regi if len(input.Tags) > 0 { r.Tags = tagsToMap(input.Tags) } - // The agent heartbeats by re-registering. A re-register from a reaped - // (involuntarily drained) instance proves the agent is back, so restore - // it to ACTIVE. An operator drain (Reaped=false) is left untouched. - if r.Status == InstanceStatusDraining && r.Reaped { + // The agent heartbeats by re-registering, so the capacity it reports is + // also the liveness signal: real capacity restores a reaped instance, and + // none drains it. + if sawCPU && sawMemory { + applyRuntimeCapacity(r, cpu, memoryMiB) + } else if r.Status == InstanceStatusDraining && r.Reaped { r.Status = InstanceStatusActive r.Reaped = false } @@ -300,11 +331,9 @@ func (s *Service) recordRegister(ctx context.Context, msg *bus.RegisterInstance) r.AZ = msg.AZ r.Hostname = msg.Hostname r.AgentVersion = msg.AgentVersion - r.TotalCPU = msg.Capacity.CPU - r.TotalMemoryMiB = msg.Capacity.MemoryMiB r.TotalGPU = msg.Capacity.GPU r.GPUIDs = msg.Capacity.GPUIDs - r.Status = InstanceStatusActive + applyRuntimeCapacity(r, msg.Capacity.CPU, msg.Capacity.MemoryMiB) }) return err } diff --git a/spinifex/handlers/ecs/records.go b/spinifex/handlers/ecs/records.go index 4ec6513eb..7010954dd 100644 --- a/spinifex/handlers/ecs/records.go +++ b/spinifex/handlers/ecs/records.go @@ -53,6 +53,13 @@ const ( // primary deployment's task launches stop before ever reaching RUNNING. circuitBreakerFailureThreshold = 3 + // launchBackoff bounds: past circuitBreakerFailureThreshold the wait doubles + // from the base on each further failed task, up to the cap. The shift bound + // keeps the doubling inside a duration for a long-failing deployment. + launchBackoffBase = 15 * time.Second + launchBackoffCap = 5 * time.Minute + launchBackoffMaxShift = 16 + CapacityProviderStatusActive = "ACTIVE" CapacityProviderStatusInactive = "INACTIVE" @@ -66,19 +73,23 @@ const ( // one PRIMARY deployment plus zero or more ACTIVE (superseded, draining) ones // while a rolling update is in flight; steady state is a single PRIMARY. type Deployment struct { - ID string `json:"id"` - Status string `json:"status"` - TaskDefARN string `json:"taskDefArn"` - TaskDefFamily string `json:"taskDefFamily"` - TaskDefRevision int `json:"taskDefRevision"` - DesiredCount int `json:"desiredCount"` - RunningCount int `json:"runningCount"` - PendingCount int `json:"pendingCount"` - FailedTasks int `json:"failedTasks"` - RolloutState string `json:"rolloutState"` - RolloutReason string `json:"rolloutStateReason,omitempty"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` + ID string `json:"id"` + Status string `json:"status"` + TaskDefARN string `json:"taskDefArn"` + TaskDefFamily string `json:"taskDefFamily"` + TaskDefRevision int `json:"taskDefRevision"` + DesiredCount int `json:"desiredCount"` + RunningCount int `json:"runningCount"` + PendingCount int `json:"pendingCount"` + FailedTasks int `json:"failedTasks"` + RolloutState string `json:"rolloutState"` + RolloutReason string `json:"rolloutStateReason,omitempty"` + // NextLaunchAt holds off the next launch attempt after tasks have failed to + // start, so a deployment that cannot run anywhere backs off instead of + // relaunching on every reconcile pass. Zero means launch now. + NextLaunchAt time.Time `json:"nextLaunchAt,omitzero"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` } // ServiceEvent is one lifecycle event on a service, mirroring ecs.ServiceEvent. diff --git a/spinifex/handlers/ecs/services.go b/spinifex/handlers/ecs/services.go index 282475922..96495b3bb 100644 --- a/spinifex/handlers/ecs/services.go +++ b/spinifex/handlers/ecs/services.go @@ -379,10 +379,16 @@ func (s *Service) reconcileService(ctx context.Context, kv jetstream.KeyValue, a minCount := ceilPercent(desired, svc.MinimumHealthyPercent) primaryActive := primary.RunningCount + primary.PendingCount - if primaryActive < desired && primary.RolloutState != RolloutStateFailed { + now := time.Now().UTC() + if primaryActive < desired && primary.RolloutState != RolloutStateFailed && !now.Before(primary.NextLaunchAt) { n := min(desired-primaryActive, max(maxCount-(running+pending), 0)) if n > 0 { s.launchDeploymentTasks(ctx, accountID, svc, primary, n) + // Set from the failures already behind this deployment, so the first + // launch is immediate and each repeat start failure widens the gap. + if d := launchBackoff(primary.FailedTasks); d > 0 { + primary.NextLaunchAt = now.Add(d) + } } else { // Wants more tasks than desired but the maximumPercent ceiling leaves // no room this pass (AWS's "unable to place a task"). Unlike the