diff --git a/cmd/installer/firstboot/firstboot.go b/cmd/installer/firstboot/firstboot.go index 93889b4d9..9292efac3 100644 --- a/cmd/installer/firstboot/firstboot.go +++ b/cmd/installer/firstboot/firstboot.go @@ -197,6 +197,9 @@ fi # Enable services to start, on reboot systemctl enable spinifex.target spinifex-banner.service +# WantedBy=timers.target only self-activates once enabled — without this the +# JetStream ENOSPC-latch watchdog never runs and a full disk needs a manual restart. +systemctl enable --now spinifex-nats-watchdog.timer # Mark complete only after every step above has succeeded. Until this point, # any failure (set -e) leaves the marker absent and the next reboot retries diff --git a/cmd/installer/firstboot/firstboot_test.go b/cmd/installer/firstboot/firstboot_test.go index 6702d0ee7..afcfe46ba 100644 --- a/cmd/installer/firstboot/firstboot_test.go +++ b/cmd/installer/firstboot/firstboot_test.go @@ -114,6 +114,25 @@ func readScript(t *testing.T, root string) string { return string(b) } +// TestWriteScriptEnablesNatsWatchdogTimer pins that firstboot activates the +// JetStream ENOSPC-latch watchdog timer. The unit file's WantedBy=timers.target +// only takes effect once enabled — without this line the watchdog is dropped +// onto disk but never runs, and a full disk needs a manual nats restart forever. +func TestWriteScriptEnablesNatsWatchdogTimer(t *testing.T) { + root := t.TempDir() + makeRootDirs(t, root) + + cfg := Config{Hostname: "test-node", ClusterRole: "init"} + if err := Write(root, cfg); err != nil { + t.Fatalf("Write: %v", err) + } + content := readScript(t, root) + + if !strings.Contains(content, "systemctl enable --now spinifex-nats-watchdog.timer") { + t.Error("firstboot script must enable --now spinifex-nats-watchdog.timer") + } +} + func TestWriteScriptCallbackAfterDoneMarker(t *testing.T) { root := t.TempDir() makeRootDirs(t, root) diff --git a/scripts/setup.sh b/scripts/setup.sh index b871f19e6..6fb69fa2f 100755 --- a/scripts/setup.sh +++ b/scripts/setup.sh @@ -667,6 +667,10 @@ EOF fi $SUDO systemctl daemon-reload $SUDO systemctl enable spinifex.target + # The unit file alone does not self-activate (WantedBy=timers.target only + # takes effect once enabled) — without this the JetStream ENOSPC-latch + # watchdog is inert and a full disk requires a manual restart forever. + $SUDO systemctl enable --now spinifex-nats-watchdog.timer info "Systemd units installed and enabled (per-service users)" } diff --git a/spinifex/lbagent/agent.go b/spinifex/lbagent/agent.go index c354119ab..fb4f2a0d0 100644 --- a/spinifex/lbagent/agent.go +++ b/spinifex/lbagent/agent.go @@ -10,6 +10,7 @@ import ( "crypto/tls" "encoding/hex" "encoding/xml" + "errors" "fmt" "io" "log/slog" @@ -25,6 +26,7 @@ import ( "github.com/mulgadc/spinifex/internal/gwsign" "github.com/mulgadc/spinifex/internal/tlsconfig" + "github.com/mulgadc/spinifex/spinifex/awserrors" ) const ( @@ -182,6 +184,15 @@ func (a *Agent) Stop() { // tick runs one heartbeat cycle: send health, check config hash, fetch if changed. func (a *Agent) tick() { + // A stopped agent must never heartbeat again, even if Start's select + // races a pending ticker tick against a just-closed stopCh — the whole + // point of stopping on a deleted LB is zero further noise. + select { + case <-a.stopCh: + return + default: + } + // nginx (NLB) has no per-server stats socket — the agent actively probes // the delivered health targets instead of reading HAProxy stats. var servers []ServerStatus @@ -211,6 +222,21 @@ func (a *Agent) tick() { "gateway", a.gatewayURL, "elapsed", time.Since(a.startedAt).Round(time.Second)) return } + + // LoadBalancerNotFound is a positive response from the gateway (the + // store round-tripped and confirmed no record), not a connectivity + // failure — distinct from a dial error, timeout, or 5xx, all of which + // leave the LB's existence unknown and must keep retrying. Past the + // startup grace window it can no longer be the record's own + // replication race, so it means the LB was deleted: stop heartbeating + // it rather than retrying forever. + var gwErr *gatewayError + if errors.As(err, &gwErr) && gwErr.code == awserrors.ErrorELBv2LoadBalancerNotFound { + slog.Info("Load balancer no longer exists, stopping agent", "lbId", a.lbID, "gateway", a.gatewayURL) + a.Stop() + return + } + slog.Error("Heartbeat failed", "err", err, "gateway", a.gatewayURL, "region", a.region) return } @@ -372,12 +398,43 @@ func (a *Agent) signedPost(params url.Values) ([]byte, error) { } if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("gateway returned %d: %s", resp.StatusCode, string(respBody)) + return nil, &gatewayError{statusCode: resp.StatusCode, code: parseAWSErrorCode(respBody), body: string(respBody)} } return respBody, nil } +// gatewayError wraps a non-200 gateway response, carrying the parsed AWS +// error code (if any) alongside the raw status/body so callers can +// distinguish a specific, unambiguous rejection (e.g. LoadBalancerNotFound) +// from a generic or ambiguous one. +type gatewayError struct { + statusCode int + code string + body string +} + +func (e *gatewayError) Error() string { + return fmt.Sprintf("gateway returned %d: %s", e.statusCode, e.body) +} + +// awsErrorEnvelope is the minimal shape of the query-protocol error response +// needed to recover the AWS error (e.g. "LoadBalancerNotFound"). +type awsErrorEnvelope struct { + XMLName xml.Name `xml:"ErrorResponse"` + Code string `xml:"Error>Code"` +} + +// parseAWSErrorCode extracts the AWS error from a gateway error body. +// Returns "" if the body does not parse as the expected envelope. +func parseAWSErrorCode(body []byte) string { + var env awsErrorEnvelope + if err := xml.Unmarshal(body, &env); err != nil { + return "" + } + return env.Code +} + // writeCertFiles writes each delivered TLS PEM to its path (0600) under certDir. // Paths that escape certDir are rejected to prevent traversal writes. func (a *Agent) writeCertFiles(certDir string, certs []certFile) error { diff --git a/spinifex/lbagent/agent_test.go b/spinifex/lbagent/agent_test.go index d134c9429..d753ff695 100644 --- a/spinifex/lbagent/agent_test.go +++ b/spinifex/lbagent/agent_test.go @@ -695,14 +695,17 @@ func TestHeartbeat_StartupFailureLogsBelowError(t *testing.T) { } func TestHeartbeat_FailureAfterGraceLogsError(t *testing.T) { - gw := notFoundGateway(t) + // An ambiguous/generic failure (not a positive not-found signal) that + // persists past the grace window is the blackhole fingerprint — it must + // escalate to ERROR so the console telemetry catches it. + gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "internal error", http.StatusInternalServerError) + })) defer gw.Close() agent := newTestAgent(t, gw.URL) logs := captureSlog(t) - // Past the grace window with no prior success — the blackhole fingerprint — - // must escalate so the console telemetry catches it. agent.startedAt = time.Now().Add(-2 * registrationGrace) agent.tick() @@ -710,19 +713,71 @@ func TestHeartbeat_FailureAfterGraceLogsError(t *testing.T) { if !strings.Contains(out, "level=ERROR") || !strings.Contains(out, "Heartbeat failed") { t.Errorf("post-grace heartbeat failure did not escalate to ERROR:\n%s", out) } + if isStopped(agent) { + t.Error("an ambiguous/generic failure must not stop the agent") + } +} + +// TestHeartbeat_NotFoundWithinGraceDoesNotStop pins the startup-race carve-out: +// a LoadBalancerNotFound seen before any successful heartbeat and inside the +// grace window is still ambiguous (the record may not have replicated to this +// gateway node yet), so the agent must keep retrying rather than stop. +func TestHeartbeat_NotFoundWithinGraceDoesNotStop(t *testing.T) { + gw := notFoundGateway(t) + defer gw.Close() + + agent := newTestAgent(t, gw.URL) + agent.tick() + + if isStopped(agent) { + t.Error("a LoadBalancerNotFound within the startup grace window must not stop the agent") + } +} + +// TestHeartbeat_NotFoundAfterGraceStopsAgent pins the fix: once the startup +// grace window has passed with no prior success, a LoadBalancerNotFound is no +// longer a replication race — the gateway has given a positive signal that no +// such LB exists — so the agent must stop rather than heartbeat it forever. +func TestHeartbeat_NotFoundAfterGraceStopsAgent(t *testing.T) { + gw := notFoundGateway(t) + defer gw.Close() + + agent := newTestAgent(t, gw.URL) + logs := captureSlog(t) + + agent.startedAt = time.Now().Add(-2 * registrationGrace) + agent.tick() + + if !isStopped(agent) { + t.Error("a LoadBalancerNotFound past the grace window must stop the agent") + } + out := logs() + if strings.Contains(out, "level=ERROR") { + t.Errorf("a confirmed LB deletion is expected teardown, not an error:\n%s", out) + } +} + +// isStopped reports whether agent's poll loop has been signalled to stop. +func isStopped(agent *Agent) bool { + select { + case <-agent.stopCh: + return true + default: + return false + } } func TestHeartbeat_FailureAfterFirstSuccessLogsError(t *testing.T) { - // A gateway that answers healthily once, then only ever 400s. + // A gateway that answers healthily once, then fails ambiguously (a + // transient/unreachable-flavoured error, not a positive not-found). var beats atomic.Int32 - gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/xml") if beats.Add(1) == 1 { fmt.Fprint(w, `active`) return } - w.WriteHeader(http.StatusBadRequest) - fmt.Fprint(w, `LoadBalancerNotFound`) + http.Error(w, "internal error", http.StatusInternalServerError) })) defer gw.Close() @@ -742,4 +797,80 @@ func TestHeartbeat_FailureAfterFirstSuccessLogsError(t *testing.T) { if !strings.Contains(out, "level=ERROR") || !strings.Contains(out, "Heartbeat failed") { t.Errorf("post-success heartbeat failure did not log ERROR:\n%s", out) } + if isStopped(agent) { + t.Error("an ambiguous/transient failure must not stop the agent, even after a prior success") + } +} + +// TestHeartbeat_NotFoundAfterSuccessStopsAgent is the primary bug scenario: +// an LB that heartbeated successfully (proving it once existed and the +// gateway was reachable) is then deleted. The gateway's LoadBalancerNotFound +// is a positive, unambiguous signal — not a connectivity failure — so the +// agent must stop rather than retry indefinitely and fill the log with +// heartbeat noise after teardown. +func TestHeartbeat_NotFoundAfterSuccessStopsAgent(t *testing.T) { + var beats atomic.Int32 + gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/xml") + if beats.Add(1) == 1 { + fmt.Fprint(w, `active`) + return + } + w.WriteHeader(http.StatusBadRequest) + fmt.Fprint(w, `LoadBalancerNotFound`) + })) + defer gw.Close() + + agent := newTestAgent(t, gw.URL) + logs := captureSlog(t) + + agent.tick() // succeeds + agent.tick() // LB now deleted + + if !isStopped(agent) { + t.Fatal("agent must stop once the gateway confirms the LB is gone") + } + + out := logs() + if strings.Contains(out, "level=ERROR") { + t.Errorf("a confirmed LB deletion is expected teardown, not an error:\n%s", out) + } + + // A further tick (as if the poll loop raced one more cycle before + // noticing stopCh) must not re-log a "not found" transition or panic — + // idempotent stop, no additional noise. + preLen := len(logs()) + agent.tick() + if len(logs()) != preLen { + t.Errorf("ticking a stopped agent logged additional heartbeat noise: %q", logs()[preLen:]) + } +} + +// TestHeartbeat_TransientErrorDuringOperationDoesNotStop is the negative +// control the fix depends on: a NATS blip or control-plane restart surfaces +// as a dial failure or 5xx, never as LoadBalancerNotFound, and must never be +// treated as "deleted" — a healthy agent must keep retrying through it. +func TestHeartbeat_TransientErrorDuringOperationDoesNotStop(t *testing.T) { + var beats atomic.Int32 + gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/xml") + if beats.Add(1) == 1 { + fmt.Fprint(w, `active`) + return + } + // Simulates a control-plane restart / NATS blip: unavailable, not deleted. + http.Error(w, "service unavailable", http.StatusServiceUnavailable) + })) + defer gw.Close() + + agent := newTestAgent(t, gw.URL) + + agent.tick() // succeeds + for range 5 { + agent.tick() // repeated transient failures + } + + if isStopped(agent) { + t.Fatal("repeated transient/unreachable errors must never stop the agent") + } }