diff --git a/build/systemd/spinifex-daemon.service b/build/systemd/spinifex-daemon.service index f5d95b56b..eb895c26d 100644 --- a/build/systemd/spinifex-daemon.service +++ b/build/systemd/spinifex-daemon.service @@ -8,7 +8,15 @@ Wants=spinifex-nats.service Type=simple User=spinifex-daemon Group=spinifex -ExecStartPre=/var/lib/spinifex/wait-for-nats.sh +# DDIL Tier 1 (siv-17): no ExecStartPre wait-for-nats. The daemon starts in +# standalone mode and reconnects to NATS in the background — process-exiting +# on a missing NATS would kill the local VM management plane. +# +# Opt-in strict-startup (§1d-strict, siv-44): set SPINIFEX_REQUIRE_NATS=1 to +# restore pre-DDIL fail-fast UX. First NATS connect uses a ~30s bounded retry +# and the daemon aborts on timeout. Intended for dev/test/single-node deploys +# where silent NATS misconfig is more confusing than helpful. Do NOT set this +# in the shipped unit — it re-introduces the SPOF that Tier 1 removed. ExecStart=/usr/local/bin/spx service spinifex start Restart=on-failure RestartSec=5 diff --git a/cmd/spinifex/cmd/templates/spinifex.toml b/cmd/spinifex/cmd/templates/spinifex.toml index 7a7b40b1a..f341002f1 100644 --- a/cmd/spinifex/cmd/templates/spinifex.toml +++ b/cmd/spinifex/cmd/templates/spinifex.toml @@ -52,6 +52,7 @@ advertise = "{{.AdvertiseIP}}" {{- end}} region = "{{.Region}}" az = "{{.Az}}" +data_dir = "{{.DataDir}}" {{- if .Services}} services = [{{range $i, $s := .Services}}{{if $i}}, {{end}}"{{$s}}"{{end}}] {{- end}} diff --git a/scripts/test-ddil-phase1.sh b/scripts/test-ddil-phase1.sh new file mode 100755 index 000000000..197c224cf --- /dev/null +++ b/scripts/test-ddil-phase1.sh @@ -0,0 +1,121 @@ +#!/bin/bash +# test-ddil-phase1.sh — systemd-level smoke test for daemon-local-autonomy +# Tier 1 (1a / 1b / 1c / 1d / 1d-strict). Run after `make deploy` on a single +# node where spinifex-daemon and spinifex-nats are managed by systemd. +# +# Usage: scripts/test-ddil-phase1.sh +# Requires: sudo, curl, jq +# +# Destructive on the local node — stops/starts spinifex-nats and restarts +# spinifex-daemon. Do NOT run against a host carrying workloads. The cleanup +# trap restores both services on exit. + +set -euo pipefail + +DAEMON_URL="${DAEMON_URL:-https://127.0.0.1:4432}" +DAEMON_SVC=spinifex-daemon +NATS_SVC=spinifex-nats +OVERRIDE=/etc/systemd/system/${DAEMON_SVC}.service.d/99-test-ddil.conf +NATS_BLOCK=/run/spinifex-nats-blocked +NATS_BLOCK_DROPIN=/run/systemd/system/${NATS_SVC}.service.d/99-test-block.conf +CURL=(curl -fsS -k --max-time 5) + +ok() { printf '\033[32mPASS\033[0m %s\n' "$1"; } +fail() { printf '\033[31mFAIL\033[0m %s\n' "$1"; exit 1; } +info() { printf '\033[36m==>\033[0m %s\n' "$1"; } + +cleanup() { + info "cleanup: removing nats block, override, restoring services" + sudo rm -f "$NATS_BLOCK" "$NATS_BLOCK_DROPIN" "$OVERRIDE" + sudo rmdir "$(dirname "$NATS_BLOCK_DROPIN")" 2>/dev/null || true + sudo systemctl daemon-reload + sudo systemctl start "$NATS_SVC" 2>/dev/null || true + sudo systemctl restart "$DAEMON_SVC" 2>/dev/null || true +} +trap cleanup EXIT + +field() { "${CURL[@]}" "${DAEMON_URL}/local/status" | jq -r ".$1 // empty"; } + +wait_field() { + local f=$1 want=$2 timeout=${3:-30} elapsed=0 + while [ "$elapsed" -lt "$timeout" ]; do + [ "$(field "$f" 2>/dev/null || true)" = "$want" ] && return 0 + sleep 1; elapsed=$((elapsed + 1)) + done + return 1 +} + +# 0. baseline +info "baseline: daemon active + cluster mode" +sudo systemctl is-active --quiet "$DAEMON_SVC" || fail "daemon not running — run \`make deploy\` first" +sudo systemctl start "$NATS_SVC" +wait_field mode cluster 30 || fail "never reached mode=cluster (got: $(field mode))" +[ "$(field nats)" = connected ] || fail "nats not connected (got: $(field nats))" +ok "mode=cluster nats=connected" + +# 1. kill NATS → standalone, daemon survives +info "1a/1c: kill NATS" +retry_before=$(field nats_retry_count) +sudo systemctl stop "$NATS_SVC" +wait_field mode standalone 30 || fail "never went standalone (got: $(field mode))" +wait_field nats disconnected 5 || fail "nats not disconnected (got: $(field nats))" +sudo systemctl is-active --quiet "$DAEMON_SVC" || fail "daemon exited — should stay up" +ok "standalone, nats=disconnected, daemon alive" + +# 2. /local/instances reachable during partition +info "1b: /local/instances during partition" +"${CURL[@]}" "${DAEMON_URL}/local/instances" | jq -e . >/dev/null \ + || fail "/local/instances did not return JSON" +ok "/local/instances served while NATS down" + +# 3. restart NATS → cluster mode + retry count advances + KV push observed +# nats_retry_count counts disconnect→reconnect *cycles* (bumped from +# onNATSReconnect, daemon.go:321). It stays flat during disconnect and +# advances by 1 here when the reconnect callback fires. +info "1c reconnect: restart NATS" +before_sync=$(field last_kv_sync_at) +sudo systemctl start "$NATS_SVC" +wait_field mode cluster 60 || fail "did not return to cluster mode" +wait_field nats connected 10 || fail "nats not reconnected" +sleep 5 +retry_after=$(field nats_retry_count) +[ "$retry_after" -gt "$retry_before" ] \ + || fail "nats_retry_count did not advance on reconnect ($retry_before → $retry_after)" +after_sync=$(field last_kv_sync_at) +[ -n "$after_sync" ] && [ "$after_sync" != "$before_sync" ] \ + || fail "last_kv_sync_at did not advance ($before_sync → $after_sync)" +ok "cluster restored, retry $retry_before → $retry_after, KV resync at $after_sync" + +# 4. SPINIFEX_REQUIRE_NATS=1 → daemon exits within ~30s when NATS down. +# Block spinifex-nats via a runtime drop-in with ConditionPathExists=!. This +# defeats the daemon unit's Wants=spinifex-nats.service auto-start, which +# would otherwise pull NATS back up on `systemctl restart spinifex-daemon` +# and let the daemon connect immediately. mask refuses (real unit lives in +# /etc) and --runtime mask is shadowed by /etc — the condition drop-in is +# the only mechanism that works against an /etc-installed unit. +info "1d-strict: SPINIFEX_REQUIRE_NATS=1 exits when NATS down" +sudo mkdir -p "$(dirname "$NATS_BLOCK_DROPIN")" +sudo tee "$NATS_BLOCK_DROPIN" >/dev/null </dev/null <.tmp → fsync → rename. The rename is +// atomic on POSIX so a concurrent reader sees either the old file or the new +// one, never a half-written file. +func WriteLocalState(path string, vms map[string]*vm.VM) error { + data, err := MarshalLocalState(vms) + if err != nil { + return err + } + return WriteLocalStateBytes(path, data) +} + +// WriteLocalStateBytes atomically writes pre-marshalled state JSON to path. +// Used by hot paths that marshal under a short-lived lock and then commit to +// disk lock-free. +func WriteLocalStateBytes(path string, data []byte) error { + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0o750); err != nil { + return fmt.Errorf("mkdir %s: %w", dir, err) + } + + tmp := path + ".tmp" + f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) + if err != nil { + return fmt.Errorf("open tmp %s: %w", tmp, err) + } + + if _, err := f.Write(data); err != nil { + f.Close() + os.Remove(tmp) + return fmt.Errorf("write tmp %s: %w", tmp, err) + } + + if err := f.Sync(); err != nil { + f.Close() + os.Remove(tmp) + return fmt.Errorf("fsync tmp %s: %w", tmp, err) + } + + if err := f.Close(); err != nil { + os.Remove(tmp) + return fmt.Errorf("close tmp %s: %w", tmp, err) + } + + if err := os.Rename(tmp, path); err != nil { + os.Remove(tmp) + return fmt.Errorf("rename %s -> %s: %w", tmp, path, err) + } + + return nil +} + +// ReadLocalState reads the per-node instance state from path. +// +// Returns (nil, nil) if the file does not exist — fresh-install signal, +// caller should start with an empty instance map. +// +// Returns an error on: +// - JSON parse failure (corruption — caller refuses start) +// - Unknown SchemaVersion (caller refuses start; safer than silent migration) +func ReadLocalState(path string) (*LocalState, error) { + data, err := os.ReadFile(path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + slog.Debug("Local state file missing, treating as fresh install", "path", path) + return nil, nil + } + return nil, fmt.Errorf("read local state %s: %w", path, err) + } + + var state LocalState + if err := json.Unmarshal(data, &state); err != nil { + return nil, fmt.Errorf("parse local state %s: %w", path, err) + } + + if state.SchemaVersion != LocalStateSchemaVersion { + return nil, fmt.Errorf("local state %s: unknown schema_version %d (expected %d)", + path, state.SchemaVersion, LocalStateSchemaVersion) + } + + if state.VMS == nil { + state.VMS = make(map[string]*vm.VM) + } + + return &state, nil +} diff --git a/spinifex/daemon/local_state_test.go b/spinifex/daemon/local_state_test.go new file mode 100644 index 000000000..74ed56d2e --- /dev/null +++ b/spinifex/daemon/local_state_test.go @@ -0,0 +1,115 @@ +package daemon + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestLocalStatePath_UsesDataDir(t *testing.T) { + got := LocalStatePath("/data") + assert.Equal(t, "/data/state/instance-state.json", got) +} + +func TestLocalStatePath_DefaultWhenEmpty(t *testing.T) { + got := LocalStatePath("") + assert.Equal(t, "/var/lib/spinifex/state/instance-state.json", got) +} + +func TestWriteLocalState_RoundTrip(t *testing.T) { + path := filepath.Join(t.TempDir(), "state", "instance-state.json") + + vms := map[string]*vm.VM{ + "i-aaa": {ID: "i-aaa", InstanceType: "t3.micro"}, + "i-bbb": {ID: "i-bbb", InstanceType: "m5.large"}, + } + + require.NoError(t, WriteLocalState(path, vms)) + + loaded, err := ReadLocalState(path) + require.NoError(t, err) + require.NotNil(t, loaded) + assert.Equal(t, LocalStateSchemaVersion, loaded.SchemaVersion) + assert.Len(t, loaded.VMS, 2) + assert.Equal(t, "t3.micro", loaded.VMS["i-aaa"].InstanceType) + assert.Equal(t, "m5.large", loaded.VMS["i-bbb"].InstanceType) +} + +func TestWriteLocalState_MkdirParent(t *testing.T) { + path := filepath.Join(t.TempDir(), "deep", "nested", "state", "instance-state.json") + require.NoError(t, WriteLocalState(path, map[string]*vm.VM{})) + _, err := os.Stat(path) + require.NoError(t, err) +} + +func TestWriteLocalState_NoTmpLeftBehind(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "instance-state.json") + require.NoError(t, WriteLocalState(path, map[string]*vm.VM{})) + + entries, err := os.ReadDir(dir) + require.NoError(t, err) + for _, e := range entries { + assert.NotContains(t, e.Name(), ".tmp", "stale tmp file left behind") + } +} + +func TestWriteLocalState_AtomicReplace(t *testing.T) { + path := filepath.Join(t.TempDir(), "instance-state.json") + + require.NoError(t, WriteLocalState(path, map[string]*vm.VM{ + "i-1": {ID: "i-1"}, + })) + + require.NoError(t, WriteLocalState(path, map[string]*vm.VM{ + "i-2": {ID: "i-2"}, + })) + + loaded, err := ReadLocalState(path) + require.NoError(t, err) + require.NotNil(t, loaded) + assert.Len(t, loaded.VMS, 1) + _, ok := loaded.VMS["i-2"] + assert.True(t, ok, "second write should replace, not merge") +} + +func TestReadLocalState_Missing(t *testing.T) { + state, err := ReadLocalState(filepath.Join(t.TempDir(), "absent.json")) + require.NoError(t, err) + assert.Nil(t, state, "missing file is the fresh-install signal") +} + +func TestReadLocalState_CorruptJSON(t *testing.T) { + path := filepath.Join(t.TempDir(), "instance-state.json") + require.NoError(t, os.WriteFile(path, []byte("{not json"), 0o600)) + + _, err := ReadLocalState(path) + require.Error(t, err) + assert.Contains(t, err.Error(), "parse local state") +} + +func TestReadLocalState_UnknownSchema(t *testing.T) { + path := filepath.Join(t.TempDir(), "instance-state.json") + data, _ := json.Marshal(map[string]any{"schema_version": 99, "vms": map[string]any{}}) + require.NoError(t, os.WriteFile(path, data, 0o600)) + + _, err := ReadLocalState(path) + require.Error(t, err) + assert.Contains(t, err.Error(), "unknown schema_version 99") +} + +func TestReadLocalState_NilVMSMapInitialized(t *testing.T) { + path := filepath.Join(t.TempDir(), "instance-state.json") + data, _ := json.Marshal(map[string]any{"schema_version": LocalStateSchemaVersion}) + require.NoError(t, os.WriteFile(path, data, 0o600)) + + state, err := ReadLocalState(path) + require.NoError(t, err) + require.NotNil(t, state) + assert.NotNil(t, state.VMS) +} diff --git a/spinifex/daemon/state_test.go b/spinifex/daemon/state_test.go index d17292acf..b9c10d820 100644 --- a/spinifex/daemon/state_test.go +++ b/spinifex/daemon/state_test.go @@ -28,11 +28,11 @@ func createDaemonWithJetStream(t *testing.T) *Daemon { clusterCfg := &config.ClusterConfig{ Node: "node-1", - Nodes: map[string]config.Config{"node-1": {BaseDir: tmpDir}}, + Nodes: map[string]config.Config{"node-1": {BaseDir: tmpDir, DataDir: tmpDir}}, } daemon, err := NewDaemon(clusterCfg) require.NoError(t, err) - daemon.config = &config.Config{BaseDir: tmpDir} + daemon.config = &config.Config{BaseDir: tmpDir, DataDir: tmpDir} daemon.natsConn = nc daemon.jsManager, err = NewJetStreamManager(nc, 1) diff --git a/spinifex/gateway/auth.go b/spinifex/gateway/auth.go index 384fccbb3..c3549f6ab 100644 --- a/spinifex/gateway/auth.go +++ b/spinifex/gateway/auth.go @@ -105,6 +105,19 @@ func (gw *GatewayConfig) SigV4AuthMiddleware() func(http.Handler) http.Handler { return } + // Fail fast when NATS is down. IAMService.LookupAccessKey runs against + // JetStream KV and waits 5s per attempt before context.DeadlineExceeded; + // the catch-all IsConnected check in gateway.Request() is unreachable + // because this middleware runs first. Mirror it here so authenticated + // paths return the cluster-unavailable 503 promised by 1c. Only fires + // when NATSConn is set but disconnected — production always assigns + // the conn in awsgw.launchService; test helpers leaving it nil get + // the existing pre-1c behaviour. + if gw.NATSConn != nil && !gw.NATSConn.IsConnected() { + gw.writeClusterUnavailable(w, r, service) + return + } + // Lookup access key in IAM KV store if gw.IAMService == nil { slog.Error("SigV4 auth: IAM service not initialized") diff --git a/spinifex/gateway/auth_test.go b/spinifex/gateway/auth_test.go index eb83e8977..007c98dfa 100644 --- a/spinifex/gateway/auth_test.go +++ b/spinifex/gateway/auth_test.go @@ -15,7 +15,9 @@ import ( "github.com/mulgadc/predastore/auth" "github.com/mulgadc/spinifex/spinifex/awserrors" handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" + "github.com/mulgadc/spinifex/spinifex/testutil" "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" ) const ( @@ -2028,3 +2030,56 @@ func TestSignedHeadersIncludeHostAndDate(t *testing.T) { }) } } + +// TestSigV4Auth_NATSDisconnectedShortCircuit verifies that the SigV4 middleware +// returns the cluster-unavailable 503 promised by 1c without ever reaching the +// IAM lookup when the NATS connection is disconnected. With IAMService nil the +// post-lookup path would return 500 InternalError, so a 503 here proves the +// short-circuit fired before the lookup. +func TestSigV4Auth_NATSDisconnectedShortCircuit(t *testing.T) { + ns, _ := testutil.StartTestNATS(t) + nc, err := nats.Connect(ns.ClientURL()) + if err != nil { + t.Fatalf("nats connect: %v", err) + } + nc.Close() // disconnected — IsConnected() returns false + + gw := &GatewayConfig{ + DisableLogging: true, + Region: testRegion, + NATSConn: nc, + IAMService: nil, + } + + r := chi.NewRouter() + r.Use(gw.SigV4AuthMiddleware()) + r.HandleFunc("/*", func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte("OK")) + }) + + authHeader, timestamp := generateTestAuthHeader( + "GET", "/", "", "", + testAccessKey, testSecretKey, testRegion, testService, + ) + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.Host = "localhost:9999" + req.Header.Set("Authorization", authHeader) + req.Header.Set("X-Amz-Date", timestamp) + + resp := doRequest(r, req) + + if resp.StatusCode != http.StatusServiceUnavailable { + t.Errorf("Expected 503, got %d", resp.StatusCode) + } + body, _ := io.ReadAll(resp.Body) + xmlStr := string(body) + if !strings.Contains(xmlStr, "ServiceUnavailable") { + t.Errorf("Expected ServiceUnavailable code in body, got: %s", xmlStr) + } + if !strings.Contains(xmlStr, "cluster unavailable: NATS disconnected") { + t.Errorf("Expected cluster-unavailable message in body, got: %s", xmlStr) + } + if !strings.Contains(xmlStr, "/local/status") { + t.Errorf("Expected /local/status hint in body, got: %s", xmlStr) + } +} diff --git a/spinifex/gateway/gateway.go b/spinifex/gateway/gateway.go index 2483d04b2..9f86a20f9 100644 --- a/spinifex/gateway/gateway.go +++ b/spinifex/gateway/gateway.go @@ -149,6 +149,49 @@ func (gw *GatewayConfig) throttleKeyFuncs() []ratelimit.KeyFunc { } } +// clusterUnavailableMsg is the body returned when the gateway short-circuits +// because the daemon's NATS connection is down. Points operators at the +// daemon's /local/status (1b) for triage instead of letting them watch the +// AWS CLI hang on per-call timeouts. +const clusterUnavailableMsg = "cluster unavailable: NATS disconnected — check daemon /local/status" + +// writeClusterUnavailable returns a 503 ServiceUnavailable for the given +// service-flavoured XML format. The body carries the literal +// clusterUnavailableMsg in — the generic GenerateEC2ErrorResponse +// path drops the message string (see ErrorHandler), so we emit XML directly +// here to make sure the /local/status hint actually reaches the operator. +func (gw *GatewayConfig) writeClusterUnavailable(w http.ResponseWriter, _ *http.Request, svc string) { + requestID := uuid.NewString() + var xmlBody string + if svc == "iam" { + iam := IAMErrorResponse{ + Error: IAMErrorDetail{ + Type: "Sender", + Code: awserrors.ErrorServiceUnavailable, + Message: clusterUnavailableMsg, + }, + RequestID: requestID, + } + out, err := xml.MarshalIndent(iam, "", " ") + if err != nil { + slog.Error("Failed to marshal IAM cluster-unavailable XML", "err", err) + out = []byte(`SenderServiceUnavailable` + clusterUnavailableMsg + `` + requestID + ``) + } + xmlBody = xml.Header + string(out) + } else { + // ec2, elasticloadbalancing, account, spinifex — all share the EC2 envelope. + xmlBody = xml.Header + `` + awserrors.ErrorServiceUnavailable + + `` + clusterUnavailableMsg + `` + + requestID + `` + } + + w.Header().Set("Content-Type", "application/xml") + w.WriteHeader(http.StatusServiceUnavailable) + if _, err := w.Write([]byte(xmlBody)); err != nil { + slog.Error("Failed to write cluster-unavailable response", "err", err) + } +} + // writeThrottleError writes the service-appropriate throttle rejection response. func (gw *GatewayConfig) writeThrottleError(w http.ResponseWriter, r *http.Request) { requestID := uuid.NewString() @@ -190,6 +233,15 @@ func (gw *GatewayConfig) Request(w http.ResponseWriter, r *http.Request) { return } + // Fail fast when NATS is down — every NATS-bound per-service handler would + // otherwise hang until per-call timeout. The body points operators at + // /local/status (1b) for diagnosis. Account is a no-op stub that never + // reaches NATS, so it is exempt. + if svc != "account" && (gw.NATSConn == nil || !gw.NATSConn.IsConnected()) { + gw.writeClusterUnavailable(w, r, svc) + return + } + switch svc { case "ec2": err = gw.EC2_Request(w, r) diff --git a/spinifex/gateway/gateway_test.go b/spinifex/gateway/gateway_test.go index 60618c65c..e3cf06fcf 100644 --- a/spinifex/gateway/gateway_test.go +++ b/spinifex/gateway/gateway_test.go @@ -553,7 +553,9 @@ func TestRequest_MalformedQueryString_EndToEnd(t *testing.T) { } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - gw := &GatewayConfig{DisableLogging: true} + // Connected NATS satisfies the 1c cluster-available gate so the + // request reaches the per-service malformed-query parser. + gw := &GatewayConfig{DisableLogging: true, NATSConn: connectedNATS(t)} req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(tc.body)) ctx := context.WithValue(req.Context(), ctxService, tc.service) ctx = context.WithValue(ctx, ctxAccountID, "123456789012") @@ -571,7 +573,7 @@ func TestRequest_MalformedQueryString_EndToEnd(t *testing.T) { } func TestRequest_EC2MissingAction(t *testing.T) { - gw := &GatewayConfig{DisableLogging: true} + gw := &GatewayConfig{DisableLogging: true, NATSConn: connectedNATS(t)} req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("")) ctx := context.WithValue(req.Context(), ctxService, "ec2") ctx = context.WithValue(ctx, ctxAccountID, "123456789012") @@ -587,7 +589,7 @@ func TestRequest_EC2MissingAction(t *testing.T) { } func TestRequest_IAMNilService(t *testing.T) { - gw := &GatewayConfig{DisableLogging: true, IAMService: nil} + gw := &GatewayConfig{DisableLogging: true, IAMService: nil, NATSConn: connectedNATS(t)} req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("Action=CreateUser&UserName=test")) ctx := context.WithValue(req.Context(), ctxService, "iam") ctx = context.WithValue(ctx, ctxAccountID, "123456789012") @@ -602,6 +604,17 @@ func TestRequest_IAMNilService(t *testing.T) { assert.Contains(t, string(body), "InternalError") } +// connectedNATS returns a live test NATS connection for short-circuit-bypass +// tests that exercise per-service handlers without actually publishing. +func connectedNATS(t *testing.T) *nats.Conn { + t.Helper() + ns, _ := testutil.StartTestNATS(t) + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + t.Cleanup(nc.Close) + return nc +} + func TestRequest_AccountReturns200(t *testing.T) { gw := &GatewayConfig{DisableLogging: true} req := httptest.NewRequest(http.MethodPost, "/", nil) @@ -1268,3 +1281,60 @@ func TestThrottleMiddleware_PerActionIsolation(t *testing.T) { resp = makeReq("RunInstances") assert.Equal(t, 200, resp.StatusCode) } + +func TestRequest_ClusterUnavailableNilConn_EC2(t *testing.T) { + gw := &GatewayConfig{DisableLogging: true} + + req := httptest.NewRequest(http.MethodPost, "/", nil) + ctx := context.WithValue(req.Context(), ctxService, "ec2") + req = req.WithContext(ctx) + + w := httptest.NewRecorder() + gw.Request(w, req) + resp := w.Result() + + assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode) + body, _ := io.ReadAll(resp.Body) + xmlStr := string(body) + assert.Contains(t, xmlStr, "ServiceUnavailable") + assert.Contains(t, xmlStr, "cluster unavailable: NATS disconnected") + assert.Contains(t, xmlStr, "/local/status") + assert.Contains(t, xmlStr, "") // EC2 XML envelope +} + +func TestRequest_ClusterUnavailableNilConn_IAM(t *testing.T) { + gw := &GatewayConfig{DisableLogging: true} + + req := httptest.NewRequest(http.MethodPost, "/", nil) + ctx := context.WithValue(req.Context(), ctxService, "iam") + req = req.WithContext(ctx) + + w := httptest.NewRecorder() + gw.Request(w, req) + resp := w.Result() + + assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode) + body, _ := io.ReadAll(resp.Body) + xmlStr := string(body) + assert.Contains(t, xmlStr, "ServiceUnavailable") + assert.Contains(t, xmlStr, "") // IAM XML envelope +} + +func TestRequest_ClusterUnavailableClosedConn(t *testing.T) { + ns, _ := testutil.StartTestNATS(t) + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + nc.Close() + + gw := &GatewayConfig{DisableLogging: true, NATSConn: nc} + + req := httptest.NewRequest(http.MethodPost, "/", nil) + ctx := context.WithValue(req.Context(), ctxService, "ec2") + req = req.WithContext(ctx) + + w := httptest.NewRecorder() + gw.Request(w, req) + resp := w.Result() + + assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode) +} diff --git a/spinifex/gateway/ratelimit_test.go b/spinifex/gateway/ratelimit_test.go index 57cb8319f..e2d83de0e 100644 --- a/spinifex/gateway/ratelimit_test.go +++ b/spinifex/gateway/ratelimit_test.go @@ -12,6 +12,9 @@ import ( "github.com/go-chi/chi/v5" "github.com/mulgadc/spinifex/spinifex/awserrors" handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/require" ) func TestCheckIP_AllowsUnknownIP(t *testing.T) { @@ -332,8 +335,12 @@ func TestExtractClientIP(t *testing.T) { } // setupTestAppWithRateLimiter creates a test HTTP handler with SigV4 auth and -// the given rate limiter attached. -func setupTestAppWithRateLimiter(accessKey, secretKey string, rl *AuthRateLimiter) http.Handler { +// the given rate limiter attached. A real NATS test connection is attached so +// the SigV4 middleware's cluster-unavailable short-circuit (mulga-siv-23) does +// not fire and mask rate-limit behaviour. +func setupTestAppWithRateLimiter(t *testing.T, accessKey, secretKey string, rl *AuthRateLimiter) http.Handler { + t.Helper() + encryptedSecret, err := handlers_iam.EncryptSecret(secretKey, testMasterKey) if err != nil { panic("failed to encrypt test secret: " + err.Error()) @@ -351,11 +358,17 @@ func setupTestAppWithRateLimiter(accessKey, secretKey string, rl *AuthRateLimite }, } + ns, _ := testutil.StartTestNATS(t) + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + t.Cleanup(nc.Close) + gw := &GatewayConfig{ DisableLogging: true, Region: testRegion, IAMService: mockSvc, RateLimiter: rl, + NATSConn: nc, } r := chi.NewRouter() @@ -371,7 +384,7 @@ func TestRateLimitIntegration_LockedIPGets503(t *testing.T) { rl := NewAuthRateLimiter() defer rl.Stop() - handler := setupTestAppWithRateLimiter(testAccessKey, testSecretKey, rl) + handler := setupTestAppWithRateLimiter(t, testAccessKey, testSecretKey, rl) // Send maxFailures requests with invalid signatures to trigger lockout. for range maxFailures { @@ -405,7 +418,7 @@ func TestRateLimitIntegration_SuccessResetsLockout(t *testing.T) { rl := NewAuthRateLimiter() defer rl.Stop() - handler := setupTestAppWithRateLimiter(testAccessKey, testSecretKey, rl) + handler := setupTestAppWithRateLimiter(t, testAccessKey, testSecretKey, rl) // Accumulate failures below threshold. for range maxFailures - 1 { diff --git a/spinifex/handlers/ec2/instance/instance_filter_test.go b/spinifex/handlers/ec2/instance/instance_filter_test.go new file mode 100644 index 000000000..34b84425d --- /dev/null +++ b/spinifex/handlers/ec2/instance/instance_filter_test.go @@ -0,0 +1,215 @@ +package handlers_ec2_instance + +import ( + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/filterutil" + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestInstanceMatchesFilters_NoFilters(t *testing.T) { + inst := &vm.VM{ID: "i-123", InstanceType: "t3.micro"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-123"), + InstanceType: aws.String("t3.micro"), + State: &ec2.InstanceState{Name: aws.String("running")}, + } + assert.True(t, instanceMatchesFilters(inst, ic, nil)) +} + +func TestInstanceMatchesFilters_SingleFilter_Match(t *testing.T) { + inst := &vm.VM{ID: "i-abc", InstanceType: "t3.micro"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-abc"), + InstanceType: aws.String("t3.micro"), + State: &ec2.InstanceState{Name: aws.String("running")}, + } + filters := map[string][]string{ + "instance-state-name": {"running"}, + } + assert.True(t, instanceMatchesFilters(inst, ic, filters)) +} + +func TestInstanceMatchesFilters_SingleFilter_NoMatch(t *testing.T) { + inst := &vm.VM{ID: "i-abc", InstanceType: "t3.micro"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-abc"), + InstanceType: aws.String("t3.micro"), + State: &ec2.InstanceState{Name: aws.String("running")}, + } + filters := map[string][]string{ + "instance-state-name": {"stopped"}, + } + assert.False(t, instanceMatchesFilters(inst, ic, filters)) +} + +func TestInstanceMatchesFilters_MultipleValues_ORLogic(t *testing.T) { + inst := &vm.VM{ID: "i-abc", InstanceType: "t3.micro"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-abc"), + InstanceType: aws.String("t3.micro"), + State: &ec2.InstanceState{Name: aws.String("running")}, + } + filters := map[string][]string{ + "instance-state-name": {"running", "stopped"}, + } + assert.True(t, instanceMatchesFilters(inst, ic, filters)) +} + +func TestInstanceMatchesFilters_MultipleFilters_ANDLogic(t *testing.T) { + inst := &vm.VM{ID: "i-abc", InstanceType: "t3.micro"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-abc"), + InstanceType: aws.String("t3.micro"), + VpcId: aws.String("vpc-111"), + State: &ec2.InstanceState{Name: aws.String("running")}, + } + filters := map[string][]string{ + "instance-state-name": {"running"}, + "vpc-id": {"vpc-111"}, + } + assert.True(t, instanceMatchesFilters(inst, ic, filters)) + + filters["vpc-id"] = []string{"vpc-999"} + assert.False(t, instanceMatchesFilters(inst, ic, filters)) +} + +func TestInstanceMatchesFilters_InstanceId(t *testing.T) { + inst := &vm.VM{ID: "i-abc"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-abc"), + State: &ec2.InstanceState{Name: aws.String("running")}, + } + filters := map[string][]string{"instance-id": {"i-abc"}} + assert.True(t, instanceMatchesFilters(inst, ic, filters)) + + filters["instance-id"] = []string{"i-other"} + assert.False(t, instanceMatchesFilters(inst, ic, filters)) +} + +func TestInstanceMatchesFilters_InstanceType(t *testing.T) { + inst := &vm.VM{ID: "i-abc", InstanceType: "m5.large"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-abc"), + InstanceType: aws.String("m5.large"), + State: &ec2.InstanceState{Name: aws.String("running")}, + } + filters := map[string][]string{"instance-type": {"m5.large"}} + assert.True(t, instanceMatchesFilters(inst, ic, filters)) +} + +func TestInstanceMatchesFilters_SubnetId(t *testing.T) { + inst := &vm.VM{ID: "i-abc"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-abc"), + SubnetId: aws.String("subnet-123"), + State: &ec2.InstanceState{Name: aws.String("running")}, + } + filters := map[string][]string{"subnet-id": {"subnet-123"}} + assert.True(t, instanceMatchesFilters(inst, ic, filters)) + + filters["subnet-id"] = []string{"subnet-999"} + assert.False(t, instanceMatchesFilters(inst, ic, filters)) +} + +func TestInstanceMatchesFilters_Wildcard(t *testing.T) { + inst := &vm.VM{ID: "i-abc", InstanceType: "t3.micro"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-abc"), + InstanceType: aws.String("t3.micro"), + State: &ec2.InstanceState{Name: aws.String("running")}, + } + filters := map[string][]string{"instance-type": {"t3.*"}} + assert.True(t, instanceMatchesFilters(inst, ic, filters)) + + filters["instance-type"] = []string{"m5.*"} + assert.False(t, instanceMatchesFilters(inst, ic, filters)) +} + +func TestInstanceMatchesFilters_TagFilter(t *testing.T) { + inst := &vm.VM{ID: "i-abc"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-abc"), + State: &ec2.InstanceState{Name: aws.String("running")}, + Tags: []*ec2.Tag{ + {Key: aws.String("Environment"), Value: aws.String("prod")}, + {Key: aws.String("Team"), Value: aws.String("backend")}, + }, + } + + filters := map[string][]string{"tag:Environment": {"prod"}} + assert.True(t, instanceMatchesFilters(inst, ic, filters)) + + filters["tag:Environment"] = []string{"staging"} + assert.False(t, instanceMatchesFilters(inst, ic, filters)) +} + +func TestInstanceMatchesFilters_TagKey(t *testing.T) { + inst := &vm.VM{ID: "i-abc"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-abc"), + State: &ec2.InstanceState{Name: aws.String("running")}, + Tags: []*ec2.Tag{ + {Key: aws.String("Environment"), Value: aws.String("prod")}, + }, + } + filters := map[string][]string{"tag-key": {"Environment"}} + assert.True(t, instanceMatchesFilters(inst, ic, filters)) + + filters["tag-key"] = []string{"MissingKey"} + assert.False(t, instanceMatchesFilters(inst, ic, filters)) +} + +func TestInstanceMatchesFilters_TagValue(t *testing.T) { + inst := &vm.VM{ID: "i-abc"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-abc"), + State: &ec2.InstanceState{Name: aws.String("running")}, + Tags: []*ec2.Tag{ + {Key: aws.String("Environment"), Value: aws.String("prod")}, + }, + } + filters := map[string][]string{"tag-value": {"prod"}} + assert.True(t, instanceMatchesFilters(inst, ic, filters)) + + filters["tag-value"] = []string{"staging"} + assert.False(t, instanceMatchesFilters(inst, ic, filters)) +} + +func TestInstanceMatchesFilters_EmptyResults(t *testing.T) { + inst := &vm.VM{ID: "i-abc", InstanceType: "t3.micro"} + ic := &ec2.Instance{ + InstanceId: aws.String("i-abc"), + InstanceType: aws.String("t3.micro"), + State: &ec2.InstanceState{Name: aws.String("running")}, + } + filters := map[string][]string{"instance-type": {"c5.4xlarge"}} + assert.False(t, instanceMatchesFilters(inst, ic, filters)) +} + +func TestParseFilters_UnknownFilterName(t *testing.T) { + filters := []*ec2.Filter{ + {Name: aws.String("bogus-filter"), Values: []*string{aws.String("val")}}, + } + _, err := filterutil.ParseFilters(filters, DescribeInstancesValidFilters) + require.Error(t, err) + assert.Contains(t, err.Error(), "InvalidParameterValue") +} + +func TestEC2TagsToMap(t *testing.T) { + tags := []*ec2.Tag{ + {Key: aws.String("a"), Value: aws.String("1")}, + {Key: aws.String("b"), Value: aws.String("2")}, + } + m := filterutil.EC2TagsToMap(tags) + assert.Equal(t, "1", m["a"]) + assert.Equal(t, "2", m["b"]) +} + +func TestEC2TagsToMap_Nil(t *testing.T) { + assert.Nil(t, filterutil.EC2TagsToMap(nil)) +} diff --git a/spinifex/handlers/ec2/instance/service.go b/spinifex/handlers/ec2/instance/service.go index 5af6caaf7..0c45e53e7 100644 --- a/spinifex/handlers/ec2/instance/service.go +++ b/spinifex/handlers/ec2/instance/service.go @@ -1,8 +1,30 @@ package handlers_ec2_instance -import "github.com/aws/aws-sdk-go/service/ec2" +import ( + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/vm" +) // InstanceService defines the interface for EC2 instance operations business logic type InstanceService interface { RunInstances(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) + DescribeInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) + DescribeInstanceTypes(input *ec2.DescribeInstanceTypesInput, accountID string) (*ec2.DescribeInstanceTypesOutput, error) + DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) + DescribeStoppedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) + DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) +} + +// ResourceCapacityProvider exposes per-node instance-type availability used by +// DescribeInstanceTypes. Implemented by daemon.ResourceManager. +type ResourceCapacityProvider interface { + GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo +} + +// StoppedInstanceStore provides KV-backed read access to stopped/terminated +// instances. Implemented by daemon.JetStreamManager. +type StoppedInstanceStore interface { + LoadStoppedInstance(instanceID string) (*vm.VM, error) + ListStoppedInstances() ([]*vm.VM, error) + ListTerminatedInstances() ([]*vm.VM, error) } diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index b5e2e2bfa..46be34b67 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -19,6 +19,7 @@ import ( "github.com/kdomanski/iso9660" "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/config" + "github.com/mulgadc/spinifex/spinifex/filterutil" "github.com/mulgadc/spinifex/spinifex/objectstore" spxtypes "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" @@ -232,15 +233,29 @@ type InstanceServiceImpl struct { instanceTypes map[string]*ec2.InstanceTypeInfo natsConn *nats.Conn objectStore objectstore.ObjectStore + vmMgr *vm.Manager + resourceMgr ResourceCapacityProvider + stoppedStore StoppedInstanceStore } // NewInstanceServiceImpl creates a new instance service implementation for daemon use -func NewInstanceServiceImpl(cfg *config.Config, instanceTypes map[string]*ec2.InstanceTypeInfo, nc *nats.Conn, store objectstore.ObjectStore) *InstanceServiceImpl { +func NewInstanceServiceImpl( + cfg *config.Config, + instanceTypes map[string]*ec2.InstanceTypeInfo, + nc *nats.Conn, + store objectstore.ObjectStore, + vmMgr *vm.Manager, + resourceMgr ResourceCapacityProvider, + stoppedStore StoppedInstanceStore, +) *InstanceServiceImpl { return &InstanceServiceImpl{ config: cfg, instanceTypes: instanceTypes, natsConn: nc, objectStore: store, + vmMgr: vmMgr, + resourceMgr: resourceMgr, + stoppedStore: stoppedStore, } } @@ -842,3 +857,401 @@ func generateHostname(instanceID string) string { } return "spinifex-vm-unknown" } + +// DescribeInstancesValidFilters lists the filter names accepted by DescribeInstances +// (and the stopped/terminated variants, which share the same filter shape). +var DescribeInstancesValidFilters = map[string]bool{ + "instance-state-name": true, + "instance-id": true, + "instance-type": true, + "vpc-id": true, + "subnet-id": true, + "tag-key": true, + "tag-value": true, +} + +// IsInstanceVisible reports whether the caller can see this instance. +// Pre-Phase4 instances (empty AccountID) are only visible to root (GlobalAccountID). +func IsInstanceVisible(callerAccountID, ownerAccountID string) bool { + if ownerAccountID == "" { + return callerAccountID == utils.GlobalAccountID + } + return callerAccountID == ownerAccountID +} + +// instanceMatchesFilters checks whether a VM + its built ec2.Instance copy satisfy all parsed filters. +func instanceMatchesFilters(inst *vm.VM, ic *ec2.Instance, filters map[string][]string) bool { + for name, values := range filters { + if strings.HasPrefix(name, "tag:") { + continue + } + + var field string + switch name { + case "instance-state-name": + if ic.State != nil && ic.State.Name != nil { + field = *ic.State.Name + } + case "instance-id": + field = inst.ID + case "instance-type": + field = inst.InstanceType + case "vpc-id": + if ic.VpcId != nil { + field = *ic.VpcId + } + case "subnet-id": + if ic.SubnetId != nil { + field = *ic.SubnetId + } + case "tag-key": + if !matchTagKey(ic.Tags, values) { + return false + } + continue + case "tag-value": + if !matchTagValue(ic.Tags, values) { + return false + } + continue + default: + return false + } + + if !filterutil.MatchesAny(values, field) { + return false + } + } + + tags := filterutil.EC2TagsToMap(ic.Tags) + return filterutil.MatchesTags(filters, tags) +} + +func matchTagKey(tags []*ec2.Tag, values []string) bool { + for _, t := range tags { + if t.Key != nil && filterutil.MatchesAny(values, *t.Key) { + return true + } + } + return false +} + +func matchTagValue(tags []*ec2.Tag, values []string) bool { + for _, t := range tags { + if t.Value != nil && filterutil.MatchesAny(values, *t.Value) { + return true + } + } + return false +} + +// DescribeInstances returns instances on this node visible to the caller's account. +func (s *InstanceServiceImpl) DescribeInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) { + slog.Info("Processing DescribeInstances request from this node", "accountID", accountID) + + instanceIDFilter := make(map[string]bool) + for _, id := range input.InstanceIds { + if id != nil && *id != "" { + if !strings.HasPrefix(*id, "i-") { + return nil, errors.New(awserrors.ErrorInvalidInstanceIDMalformed) + } + instanceIDFilter[*id] = true + } + } + + parsedFilters, err := filterutil.ParseFilters(input.Filters, DescribeInstancesValidFilters) + if err != nil { + slog.Warn("DescribeInstances: invalid filter", "err", err) + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + reservationMap := make(map[string]*ec2.Reservation) + + // Iterate under the manager lock — VM fields (Status, Instance, Reservation, + // PublicIP, PlacementGroupName) are mutated through manager-locked + // Inspect/UpdateState elsewhere, so reading them lock-free would race. + s.vmMgr.View(func(vms map[string]*vm.VM) { + for _, instance := range vms { + if !IsInstanceVisible(accountID, instance.AccountID) { + continue + } + if len(instanceIDFilter) > 0 && !instanceIDFilter[instance.ID] { + continue + } + if instance.Reservation == nil || instance.Instance == nil { + continue + } + + resID := "" + if instance.Reservation.ReservationId != nil { + resID = *instance.Reservation.ReservationId + } + + if _, exists := reservationMap[resID]; !exists { + reservation := &ec2.Reservation{} + reservation.SetReservationId(resID) + if instance.Reservation.OwnerId != nil { + reservation.SetOwnerId(*instance.Reservation.OwnerId) + } + reservation.Instances = []*ec2.Instance{} + reservationMap[resID] = reservation + } + + instanceCopy := *instance.Instance + instanceCopy.State = &ec2.InstanceState{} + + if instance.PublicIP != "" && instanceCopy.PublicIpAddress == nil { + instanceCopy.PublicIpAddress = aws.String(instance.PublicIP) + } + + if info, ok := vm.EC2StateCodes[instance.Status]; ok { + instanceCopy.State.SetCode(info.Code) + instanceCopy.State.SetName(info.Name) + } else { + slog.Warn("Instance has unmapped status, reporting as pending", + "instanceId", instance.ID, "status", string(instance.Status)) + instanceCopy.State.SetCode(0) + instanceCopy.State.SetName("pending") + } + + if instance.PlacementGroupName != "" { + instanceCopy.Placement = &ec2.Placement{ + GroupName: aws.String(instance.PlacementGroupName), + AvailabilityZone: aws.String(s.config.AZ), + } + } + + if len(parsedFilters) > 0 && !instanceMatchesFilters(instance, &instanceCopy, parsedFilters) { + continue + } + + reservationMap[resID].Instances = append(reservationMap[resID].Instances, &instanceCopy) + } + }) + + reservations := make([]*ec2.Reservation, 0, len(reservationMap)) + for _, reservation := range reservationMap { + reservations = append(reservations, reservation) + } + + slog.Info("DescribeInstances completed", "count", len(reservations)) + return &ec2.DescribeInstancesOutput{Reservations: reservations}, nil +} + +// DescribeInstanceTypes returns instance types provisionable on this node. +// The "capacity" filter (when "true") expands each type to one entry per +// available slot so callers can report cluster-wide capacity by aggregating. +func (s *InstanceServiceImpl) DescribeInstanceTypes(input *ec2.DescribeInstanceTypesInput, _ string) (*ec2.DescribeInstanceTypesOutput, error) { + slog.Info("Processing DescribeInstanceTypes request from this node") + + if s.resourceMgr == nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + showCapacity := false + for _, f := range input.Filters { + if f.Name != nil && *f.Name == "capacity" { + for _, v := range f.Values { + if v != nil && *v == "true" { + showCapacity = true + break + } + } + } + } + + filteredTypes := s.resourceMgr.GetAvailableInstanceTypeInfos(showCapacity) + slog.Info("DescribeInstanceTypes completed", "count", len(filteredTypes)) + return &ec2.DescribeInstanceTypesOutput{InstanceTypes: filteredTypes}, nil +} + +// DescribeStoppedInstances returns stopped instances from shared KV. +func (s *InstanceServiceImpl) DescribeStoppedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) { + if s.stoppedStore == nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + return s.describeInstancesFromKV(input, accountID, s.stoppedStore.ListStoppedInstances, 80, "stopped", "DescribeStoppedInstances") +} + +// DescribeTerminatedInstances returns terminated instances from the terminated KV bucket. +func (s *InstanceServiceImpl) DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) { + if s.stoppedStore == nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + return s.describeInstancesFromKV(input, accountID, s.stoppedStore.ListTerminatedInstances, 48, "terminated", "DescribeTerminatedInstances") +} + +func (s *InstanceServiceImpl) describeInstancesFromKV(input *ec2.DescribeInstancesInput, accountID string, listFn func() ([]*vm.VM, error), fallbackCode int64, fallbackName, opName string) (*ec2.DescribeInstancesOutput, error) { + instanceIDFilter := make(map[string]bool) + for _, id := range input.InstanceIds { + if id != nil { + instanceIDFilter[*id] = true + } + } + + parsedFilters, filterErr := filterutil.ParseFilters(input.Filters, DescribeInstancesValidFilters) + if filterErr != nil { + slog.Warn(opName+": invalid filter", "err", filterErr) + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + instances, err := listFn() + if err != nil { + slog.Error(opName+": failed to list instances", "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + reservationMap := make(map[string]*ec2.Reservation) + + for _, instance := range instances { + if !IsInstanceVisible(accountID, instance.AccountID) { + continue + } + if len(instanceIDFilter) > 0 && !instanceIDFilter[instance.ID] { + continue + } + if instance.Reservation == nil || instance.Instance == nil { + slog.Warn(opName+": skipping instance with nil Reservation/Instance (data integrity issue)", + "instanceId", instance.ID) + continue + } + + resID := "" + if instance.Reservation.ReservationId != nil { + resID = *instance.Reservation.ReservationId + } + + if _, exists := reservationMap[resID]; !exists { + reservation := &ec2.Reservation{} + reservation.SetReservationId(resID) + if instance.Reservation.OwnerId != nil { + reservation.SetOwnerId(*instance.Reservation.OwnerId) + } + reservation.Instances = []*ec2.Instance{} + reservationMap[resID] = reservation + } + + instanceCopy := *instance.Instance + instanceCopy.State = &ec2.InstanceState{} + if info, ok := vm.EC2StateCodes[instance.Status]; ok { + instanceCopy.State.SetCode(info.Code) + instanceCopy.State.SetName(info.Name) + } else { + instanceCopy.State.SetCode(fallbackCode) + instanceCopy.State.SetName(fallbackName) + } + + if len(parsedFilters) > 0 && !instanceMatchesFilters(instance, &instanceCopy, parsedFilters) { + continue + } + + reservationMap[resID].Instances = append(reservationMap[resID].Instances, &instanceCopy) + } + + reservations := make([]*ec2.Reservation, 0, len(reservationMap)) + for _, reservation := range reservationMap { + reservations = append(reservations, reservation) + } + + slog.Info(opName+" completed", "count", len(reservations)) + return &ec2.DescribeInstancesOutput{Reservations: reservations}, nil +} + +// DescribeInstanceAttribute returns a single requested attribute for an instance. +// Checks running instances first (in-memory), then falls back to stopped instances in KV. +func (s *InstanceServiceImpl) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) { + if input.InstanceId == nil || *input.InstanceId == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + if input.Attribute == nil || *input.Attribute == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + + instanceID := *input.InstanceId + attribute := *input.Attribute + + var instance *vm.VM + if running, ok := s.vmMgr.Get(instanceID); ok { + instance = running + } + + if instance == nil { + if s.stoppedStore == nil { + slog.Error("DescribeInstanceAttribute: stopped store not available") + return nil, errors.New(awserrors.ErrorServerInternal) + } + stopped, err := s.stoppedStore.LoadStoppedInstance(instanceID) + if err != nil { + slog.Error("DescribeInstanceAttribute: failed to load stopped instance", + "instanceId", instanceID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + instance = stopped + } + + if instance == nil { + slog.Warn("DescribeInstanceAttribute: instance not found", + "instanceId", instanceID) + return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + } + + if !IsInstanceVisible(accountID, instance.AccountID) { + slog.Warn("DescribeInstanceAttribute: instance not visible", + "instanceId", instanceID, "callerAccount", accountID, "ownerAccount", instance.AccountID) + return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + } + + output := &ec2.DescribeInstanceAttributeOutput{ + InstanceId: &instanceID, + } + + switch attribute { + case ec2.InstanceAttributeNameInstanceType: + val := instance.InstanceType + output.InstanceType = &ec2.AttributeValue{Value: &val} + + case ec2.InstanceAttributeNameUserData: + val := instance.UserData + output.UserData = &ec2.AttributeValue{Value: &val} + + case ec2.InstanceAttributeNameDisableApiTermination: + val := false + output.DisableApiTermination = &ec2.AttributeBooleanValue{Value: &val} + + case ec2.InstanceAttributeNameDisableApiStop: + val := false + output.DisableApiStop = &ec2.AttributeBooleanValue{Value: &val} + + case ec2.InstanceAttributeNameInstanceInitiatedShutdownBehavior: + val := ec2.ShutdownBehaviorStop + output.InstanceInitiatedShutdownBehavior = &ec2.AttributeValue{Value: &val} + + case ec2.InstanceAttributeNameEbsOptimized: + val := false + output.EbsOptimized = &ec2.AttributeBooleanValue{Value: &val} + + case ec2.InstanceAttributeNameEnaSupport: + val := true + output.EnaSupport = &ec2.AttributeBooleanValue{Value: &val} + + case ec2.InstanceAttributeNameSourceDestCheck: + val := true + output.SourceDestCheck = &ec2.AttributeBooleanValue{Value: &val} + + case ec2.InstanceAttributeNameGroupSet: + if instance.Instance != nil && len(instance.Instance.SecurityGroups) > 0 { + output.Groups = instance.Instance.SecurityGroups + } else { + output.Groups = []*ec2.GroupIdentifier{} + } + + default: + slog.Warn("DescribeInstanceAttribute: unsupported attribute", + "instanceId", instanceID, "attribute", attribute) + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + slog.Info("DescribeInstanceAttribute: completed", + "instanceId", instanceID, "attribute", attribute, "accountID", accountID) + return output, nil +} diff --git a/spinifex/handlers/ec2/instance/service_impl_test.go b/spinifex/handlers/ec2/instance/service_impl_test.go index 76c09d8be..613ae4f6f 100644 --- a/spinifex/handlers/ec2/instance/service_impl_test.go +++ b/spinifex/handlers/ec2/instance/service_impl_test.go @@ -10,12 +10,43 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/config" + "github.com/mulgadc/spinifex/spinifex/objectstore" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/mulgadc/spinifex/spinifex/vm" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// mgrWith returns a vm.Manager pre-populated with vms. Test fixture for +// services that previously took a *vm.Instances; post-vm.Manager refactor we +// build a Manager and Replace its set rather than inlining a map. +func mgrWith(vms map[string]*vm.VM) *vm.Manager { + m := vm.NewManager() + if len(vms) > 0 { + m.Replace(vms) + } + return m +} + +func TestNewInstanceServiceImpl(t *testing.T) { + cfg := &config.Config{} + instanceTypes := map[string]*ec2.InstanceTypeInfo{ + "t3.micro": {InstanceType: aws.String("t3.micro")}, + } + store := objectstore.NewMemoryObjectStore() + mgr := vm.NewManager() + + svc := NewInstanceServiceImpl(cfg, instanceTypes, nil, store, mgr, nil, nil) + + require.NotNil(t, svc) + assert.Equal(t, cfg, svc.config) + assert.Equal(t, instanceTypes, svc.instanceTypes) + assert.Nil(t, svc.natsConn) + assert.Equal(t, mgr, svc.vmMgr) + assert.Equal(t, store, svc.objectStore) +} + func TestGenerateHostname(t *testing.T) { tests := []struct { name string @@ -606,3 +637,402 @@ func TestCloudInitVolumeNamePerInstance(t *testing.T) { seen[cloudInitName] = true } } + +// --- 1b-pre Describe* coverage (siv-22 parts) ------------------------------ + +type fakeResourceCapacityProvider struct { + types []*ec2.InstanceTypeInfo + gotShowCap bool + calls int +} + +func (f *fakeResourceCapacityProvider) GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo { + f.calls++ + f.gotShowCap = showCapacity + return f.types +} + +type fakeStoppedStore struct { + stopped []*vm.VM + terminated []*vm.VM + loadByID map[string]*vm.VM + listErr error + listTermErr error + loadErr error +} + +func (f *fakeStoppedStore) LoadStoppedInstance(id string) (*vm.VM, error) { + if f.loadErr != nil { + return nil, f.loadErr + } + if v, ok := f.loadByID[id]; ok { + return v, nil + } + return nil, nil +} +func (f *fakeStoppedStore) ListStoppedInstances() ([]*vm.VM, error) { + if f.listErr != nil { + return nil, f.listErr + } + return f.stopped, nil +} +func (f *fakeStoppedStore) ListTerminatedInstances() ([]*vm.VM, error) { + if f.listTermErr != nil { + return nil, f.listTermErr + } + return f.terminated, nil +} + +func TestDescribeInstanceTypes_NilResourceMgr(t *testing.T) { + svc := &InstanceServiceImpl{} + _, err := svc.DescribeInstanceTypes(&ec2.DescribeInstanceTypesInput{}, "") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestDescribeInstanceTypes_ReturnsProviderTypes(t *testing.T) { + prov := &fakeResourceCapacityProvider{ + types: []*ec2.InstanceTypeInfo{ + {InstanceType: aws.String("t3.micro")}, + {InstanceType: aws.String("t3.small")}, + }, + } + svc := &InstanceServiceImpl{resourceMgr: prov} + + out, err := svc.DescribeInstanceTypes(&ec2.DescribeInstanceTypesInput{}, "") + require.NoError(t, err) + require.Len(t, out.InstanceTypes, 2) + assert.False(t, prov.gotShowCap) +} + +func TestDescribeInstanceTypes_CapacityFilterPropagates(t *testing.T) { + prov := &fakeResourceCapacityProvider{} + svc := &InstanceServiceImpl{resourceMgr: prov} + + input := &ec2.DescribeInstanceTypesInput{ + Filters: []*ec2.Filter{ + {Name: aws.String("capacity"), Values: []*string{aws.String("true")}}, + }, + } + _, err := svc.DescribeInstanceTypes(input, "") + require.NoError(t, err) + assert.True(t, prov.gotShowCap, "capacity=true filter must reach the provider") +} + +func TestDescribeInstances_Empty(t *testing.T) { + svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{})} + out, err := svc.DescribeInstances(&ec2.DescribeInstancesInput{}, utils.GlobalAccountID) + require.NoError(t, err) + assert.Empty(t, out.Reservations) +} + +func TestDescribeInstances_OneVisibleInstance(t *testing.T) { + id := "i-aaa111" + resID := "r-1" + owner := "111122223333" + v := &vm.VM{ + ID: id, + InstanceType: "t3.micro", + Status: vm.StateRunning, + AccountID: owner, + Reservation: &ec2.Reservation{ + ReservationId: aws.String(resID), + OwnerId: aws.String(owner), + }, + Instance: &ec2.Instance{InstanceId: aws.String(id), InstanceType: aws.String("t3.micro")}, + } + svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{id: v})} + + out, err := svc.DescribeInstances(&ec2.DescribeInstancesInput{}, owner) + require.NoError(t, err) + require.Len(t, out.Reservations, 1) + assert.Equal(t, resID, *out.Reservations[0].ReservationId) + require.Len(t, out.Reservations[0].Instances, 1) + assert.Equal(t, id, *out.Reservations[0].Instances[0].InstanceId) +} + +func TestDescribeInstances_AccountFilteringHidesOtherTenant(t *testing.T) { + v := &vm.VM{ + ID: "i-other", + AccountID: "999988887777", + Reservation: &ec2.Reservation{ + ReservationId: aws.String("r-other"), + OwnerId: aws.String("999988887777"), + }, + Instance: &ec2.Instance{InstanceId: aws.String("i-other")}, + } + svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{v.ID: v})} + + out, err := svc.DescribeInstances(&ec2.DescribeInstancesInput{}, "111122223333") + require.NoError(t, err) + assert.Empty(t, out.Reservations) +} + +func TestDescribeInstances_MalformedID(t *testing.T) { + svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{})} + input := &ec2.DescribeInstancesInput{ + InstanceIds: []*string{aws.String("not-an-id")}, + } + _, err := svc.DescribeInstances(input, utils.GlobalAccountID) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDMalformed, err.Error()) +} + +func TestDescribeInstances_FilterByInstanceID(t *testing.T) { + keep := &vm.VM{ + ID: "i-keep", + Reservation: &ec2.Reservation{ + ReservationId: aws.String("r-keep"), + }, + Instance: &ec2.Instance{InstanceId: aws.String("i-keep")}, + } + drop := &vm.VM{ + ID: "i-drop", + Reservation: &ec2.Reservation{ + ReservationId: aws.String("r-drop"), + }, + Instance: &ec2.Instance{InstanceId: aws.String("i-drop")}, + } + svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{ + keep.ID: keep, + drop.ID: drop, + })} + + input := &ec2.DescribeInstancesInput{InstanceIds: []*string{aws.String("i-keep")}} + out, err := svc.DescribeInstances(input, utils.GlobalAccountID) + require.NoError(t, err) + require.Len(t, out.Reservations, 1) + assert.Equal(t, "i-keep", *out.Reservations[0].Instances[0].InstanceId) +} + +func TestDescribeInstanceAttribute_MissingInstanceID(t *testing.T) { + svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{})} + _, err := svc.DescribeInstanceAttribute(&ec2.DescribeInstanceAttributeInput{ + Attribute: aws.String(ec2.InstanceAttributeNameInstanceType), + }, "") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorMissingParameter, err.Error()) +} + +func TestDescribeInstanceAttribute_MissingAttribute(t *testing.T) { + svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{})} + _, err := svc.DescribeInstanceAttribute(&ec2.DescribeInstanceAttributeInput{ + InstanceId: aws.String("i-aaa"), + }, "") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorMissingParameter, err.Error()) +} + +func TestDescribeInstanceAttribute_RunningInstance(t *testing.T) { + id := "i-attr1" + owner := utils.GlobalAccountID + v := &vm.VM{ + ID: id, + InstanceType: "t3.large", + AccountID: owner, + UserData: "raw-user-data", + } + svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{id: v})} + + tests := []struct { + name string + attribute string + assertions func(t *testing.T, out *ec2.DescribeInstanceAttributeOutput) + }{ + { + name: "instanceType", + attribute: ec2.InstanceAttributeNameInstanceType, + assertions: func(t *testing.T, out *ec2.DescribeInstanceAttributeOutput) { + require.NotNil(t, out.InstanceType) + assert.Equal(t, "t3.large", *out.InstanceType.Value) + }, + }, + { + name: "userData", + attribute: ec2.InstanceAttributeNameUserData, + assertions: func(t *testing.T, out *ec2.DescribeInstanceAttributeOutput) { + require.NotNil(t, out.UserData) + assert.Equal(t, "raw-user-data", *out.UserData.Value) + }, + }, + { + name: "disableApiTermination", + attribute: ec2.InstanceAttributeNameDisableApiTermination, + assertions: func(t *testing.T, out *ec2.DescribeInstanceAttributeOutput) { + require.NotNil(t, out.DisableApiTermination) + assert.False(t, *out.DisableApiTermination.Value) + }, + }, + { + name: "ebsOptimized", + attribute: ec2.InstanceAttributeNameEbsOptimized, + assertions: func(t *testing.T, out *ec2.DescribeInstanceAttributeOutput) { + require.NotNil(t, out.EbsOptimized) + assert.False(t, *out.EbsOptimized.Value) + }, + }, + { + name: "enaSupport", + attribute: ec2.InstanceAttributeNameEnaSupport, + assertions: func(t *testing.T, out *ec2.DescribeInstanceAttributeOutput) { + require.NotNil(t, out.EnaSupport) + assert.True(t, *out.EnaSupport.Value) + }, + }, + { + name: "sourceDestCheck", + attribute: ec2.InstanceAttributeNameSourceDestCheck, + assertions: func(t *testing.T, out *ec2.DescribeInstanceAttributeOutput) { + require.NotNil(t, out.SourceDestCheck) + assert.True(t, *out.SourceDestCheck.Value) + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + out, err := svc.DescribeInstanceAttribute(&ec2.DescribeInstanceAttributeInput{ + InstanceId: aws.String(id), + Attribute: aws.String(tc.attribute), + }, owner) + require.NoError(t, err) + require.NotNil(t, out) + assert.Equal(t, id, *out.InstanceId) + tc.assertions(t, out) + }) + } +} + +func TestDescribeInstanceAttribute_NotRunning_NoStore(t *testing.T) { + svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{})} + _, err := svc.DescribeInstanceAttribute(&ec2.DescribeInstanceAttributeInput{ + InstanceId: aws.String("i-missing"), + Attribute: aws.String(ec2.InstanceAttributeNameInstanceType), + }, "") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestDescribeInstanceAttribute_FoundInStoppedStore(t *testing.T) { + id := "i-stopped1" + owner := utils.GlobalAccountID + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{ + id: {ID: id, InstanceType: "t3.medium", AccountID: owner}, + }} + svc := &InstanceServiceImpl{ + vmMgr: mgrWith(map[string]*vm.VM{}), + stoppedStore: store, + } + + out, err := svc.DescribeInstanceAttribute(&ec2.DescribeInstanceAttributeInput{ + InstanceId: aws.String(id), + Attribute: aws.String(ec2.InstanceAttributeNameInstanceType), + }, owner) + require.NoError(t, err) + assert.Equal(t, "t3.medium", *out.InstanceType.Value) +} + +func TestDescribeInstanceAttribute_NotFound(t *testing.T) { + svc := &InstanceServiceImpl{ + vmMgr: mgrWith(map[string]*vm.VM{}), + stoppedStore: &fakeStoppedStore{loadByID: map[string]*vm.VM{}}, + } + _, err := svc.DescribeInstanceAttribute(&ec2.DescribeInstanceAttributeInput{ + InstanceId: aws.String("i-ghost"), + Attribute: aws.String(ec2.InstanceAttributeNameInstanceType), + }, utils.GlobalAccountID) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) +} + +func TestDescribeInstanceAttribute_HiddenForOtherAccount(t *testing.T) { + id := "i-other-acct" + v := &vm.VM{ID: id, InstanceType: "t3.micro", AccountID: "999988887777"} + svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{id: v})} + + _, err := svc.DescribeInstanceAttribute(&ec2.DescribeInstanceAttributeInput{ + InstanceId: aws.String(id), + Attribute: aws.String(ec2.InstanceAttributeNameInstanceType), + }, "111122223333") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) +} + +func TestDescribeStoppedInstances_NilStore(t *testing.T) { + svc := &InstanceServiceImpl{} + _, err := svc.DescribeStoppedInstances(&ec2.DescribeInstancesInput{}, "") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestDescribeStoppedInstances_HappyPath(t *testing.T) { + owner := utils.GlobalAccountID + store := &fakeStoppedStore{ + stopped: []*vm.VM{ + { + ID: "i-stop1", + AccountID: owner, + Reservation: &ec2.Reservation{ + ReservationId: aws.String("r-stop1"), + OwnerId: aws.String(owner), + }, + Instance: &ec2.Instance{InstanceId: aws.String("i-stop1")}, + }, + }, + } + svc := &InstanceServiceImpl{stoppedStore: store} + + out, err := svc.DescribeStoppedInstances(&ec2.DescribeInstancesInput{}, owner) + require.NoError(t, err) + require.Len(t, out.Reservations, 1) + assert.Equal(t, "i-stop1", *out.Reservations[0].Instances[0].InstanceId) +} + +func TestDescribeTerminatedInstances_NilStore(t *testing.T) { + svc := &InstanceServiceImpl{} + _, err := svc.DescribeTerminatedInstances(&ec2.DescribeInstancesInput{}, "") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestDescribeTerminatedInstances_HappyPath(t *testing.T) { + owner := utils.GlobalAccountID + store := &fakeStoppedStore{ + terminated: []*vm.VM{ + { + ID: "i-term1", + AccountID: owner, + Reservation: &ec2.Reservation{ + ReservationId: aws.String("r-term1"), + OwnerId: aws.String(owner), + }, + Instance: &ec2.Instance{InstanceId: aws.String("i-term1")}, + }, + }, + } + svc := &InstanceServiceImpl{stoppedStore: store} + + out, err := svc.DescribeTerminatedInstances(&ec2.DescribeInstancesInput{}, owner) + require.NoError(t, err) + require.Len(t, out.Reservations, 1) + assert.Equal(t, "i-term1", *out.Reservations[0].Instances[0].InstanceId) +} + +func TestIsInstanceVisible(t *testing.T) { + tests := []struct { + name string + caller string + owner string + want bool + }{ + {"empty owner, global caller", utils.GlobalAccountID, "", true}, + {"empty owner, non-global caller", "111122223333", "", false}, + {"matching account", "111122223333", "111122223333", true}, + {"different accounts", "111122223333", "999988887777", false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, IsInstanceVisible(tc.caller, tc.owner)) + }) + } +} diff --git a/spinifex/handlers/ec2/instance/service_mock.go b/spinifex/handlers/ec2/instance/service_mock.go index a717eea3a..be98d09f2 100644 --- a/spinifex/handlers/ec2/instance/service_mock.go +++ b/spinifex/handlers/ec2/instance/service_mock.go @@ -35,3 +35,23 @@ func (s *MockInstanceService) RunInstances(input *ec2.RunInstancesInput, account return reservation, nil } + +func (s *MockInstanceService) DescribeInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) { + return &ec2.DescribeInstancesOutput{}, nil +} + +func (s *MockInstanceService) DescribeInstanceTypes(_ *ec2.DescribeInstanceTypesInput, _ string) (*ec2.DescribeInstanceTypesOutput, error) { + return &ec2.DescribeInstanceTypesOutput{}, nil +} + +func (s *MockInstanceService) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, _ string) (*ec2.DescribeInstanceAttributeOutput, error) { + return &ec2.DescribeInstanceAttributeOutput{InstanceId: input.InstanceId}, nil +} + +func (s *MockInstanceService) DescribeStoppedInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) { + return &ec2.DescribeInstancesOutput{}, nil +} + +func (s *MockInstanceService) DescribeTerminatedInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) { + return &ec2.DescribeInstancesOutput{}, nil +} diff --git a/spinifex/handlers/ec2/instance/service_nats.go b/spinifex/handlers/ec2/instance/service_nats.go index 1de8a1b3c..66023c709 100644 --- a/spinifex/handlers/ec2/instance/service_nats.go +++ b/spinifex/handlers/ec2/instance/service_nats.go @@ -10,7 +10,14 @@ import ( "github.com/nats-io/nats.go" ) -// NATSInstanceService handles instance operations via NATS messaging +// NATSInstanceService handles instance operations via NATS messaging. +// +// Note: DescribeInstances and DescribeInstanceTypes subscriptions are +// fan-out (no queue group); a single NATSRequest only captures one +// responder. Cluster-wide aggregation is the gateway's job (see +// spinifex/gateway/ec2/instance). The methods below are kept for symmetry +// with the InstanceService interface and intentionally fail rather than +// silently returning a one-node view. type NATSInstanceService struct { natsConn *nats.Conn } @@ -29,3 +36,27 @@ func (s *NATSInstanceService) RunInstances(input *ec2.RunInstancesInput, account topic := fmt.Sprintf("ec2.RunInstances.%s", aws.StringValue(input.InstanceType)) return utils.NATSRequest[ec2.Reservation](s.natsConn, topic, input, 5*time.Minute, accountID) } + +// DescribeInstances is a fan-out operation; use the gateway's scatter-gather +// helper instead of this single-node client. +func (s *NATSInstanceService) DescribeInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) { + return nil, fmt.Errorf("DescribeInstances is fan-out; use gateway scatter-gather, not NATSInstanceService") +} + +// DescribeInstanceTypes is a fan-out operation; use the gateway's scatter-gather +// helper instead of this single-node client. +func (s *NATSInstanceService) DescribeInstanceTypes(_ *ec2.DescribeInstanceTypesInput, _ string) (*ec2.DescribeInstanceTypesOutput, error) { + return nil, fmt.Errorf("DescribeInstanceTypes is fan-out; use gateway scatter-gather, not NATSInstanceService") +} + +func (s *NATSInstanceService) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) { + return utils.NATSRequest[ec2.DescribeInstanceAttributeOutput](s.natsConn, "ec2.DescribeInstanceAttribute", input, 30*time.Second, accountID) +} + +func (s *NATSInstanceService) DescribeStoppedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) { + return utils.NATSRequest[ec2.DescribeInstancesOutput](s.natsConn, "ec2.DescribeStoppedInstances", input, 30*time.Second, accountID) +} + +func (s *NATSInstanceService) DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) { + return utils.NATSRequest[ec2.DescribeInstancesOutput](s.natsConn, "ec2.DescribeTerminatedInstances", input, 30*time.Second, accountID) +} diff --git a/spinifex/utils/nats.go b/spinifex/utils/nats.go index 3d7feceb3..c5df9d79f 100644 --- a/spinifex/utils/nats.go +++ b/spinifex/utils/nats.go @@ -1,6 +1,7 @@ package utils import ( + "context" "crypto/tls" "crypto/x509" "encoding/json" @@ -21,23 +22,51 @@ var ( ErrCACertParse = errors.New("failed to parse CA cert") ) +// ErrClusterUnavailable is returned by NATS request helpers when the underlying +// connection is not currently connected. Callers (gateway, scatter-gather +// fan-out) use this to fail fast instead of waiting for per-call timeouts. +var ErrClusterUnavailable = errors.New("cluster unavailable: NATS disconnected") + +// natsRetryEscalateAttempt is the attempt count past which ConnectNATSWithRetry +// promotes the per-attempt log line from slog.Warn to slog.Error (rate-limited +// to once a minute). With the 60s backoff cap, ~30 attempts corresponds to +// ~30 minutes of continuous disconnection — long enough to suspect a config +// error rather than a routine NATS restart. +const natsRetryEscalateAttempt = 30 + // ConnectNATS establishes a connection to a NATS server with standard reconnect // handling and logging. If token is non-empty, token authentication is used. // If caCertPath is non-empty, TLS is enabled using the given CA certificate. -func ConnectNATS(host, token, caCertPath string) (*nats.Conn, error) { - opts := []nats.Option{ +// +// Optional callback hooks (WithDisconnectHandler / WithReconnectHandler) wrap +// the default log lines so callers can react to connectivity changes (e.g. the +// daemon flips its cluster/standalone mode field) without losing the existing +// disconnect/reconnect log output. +func ConnectNATS(host, token, caCertPath string, opts ...RetryOption) (*nats.Conn, error) { + cfg := retryConfig{} + for _, o := range opts { + o(&cfg) + } + + natsOpts := []nats.Option{ nats.ReconnectWait(time.Second), nats.MaxReconnects(-1), nats.DisconnectErrHandler(func(nc *nats.Conn, err error) { slog.Warn("NATS disconnected", "err", err) + if cfg.onDisconnect != nil { + cfg.onDisconnect(nc, err) + } }), nats.ReconnectHandler(func(nc *nats.Conn) { slog.Warn("NATS reconnected", "url", nc.ConnectedUrl()) + if cfg.onReconnect != nil { + cfg.onReconnect(nc) + } }), } if token != "" { - opts = append(opts, nats.Token(token)) + natsOpts = append(natsOpts, nats.Token(token)) } if caCertPath != "" { @@ -49,12 +78,12 @@ func ConnectNATS(host, token, caCertPath string) (*nats.Conn, error) { if !pool.AppendCertsFromPEM(caCert) { return nil, fmt.Errorf("%w from %s", ErrCACertParse, caCertPath) } - opts = append(opts, nats.Secure(&tls.Config{ + natsOpts = append(natsOpts, nats.Secure(&tls.Config{ RootCAs: pool, })) } - nc, err := nats.Connect(host, opts...) + nc, err := nats.Connect(host, natsOpts...) if err != nil { return nil, fmt.Errorf("NATS connect failed: %w", err) } @@ -63,13 +92,22 @@ func ConnectNATS(host, token, caCertPath string) (*nats.Conn, error) { return nc, nil } -// retryConfig holds parameters for ConnectNATSWithRetry. +// retryConfig holds parameters for ConnectNATS / ConnectNATSWithRetry. +// +// retry-related fields tune the outer reconnect-with-backoff loop; +// callback fields wrap nats.go's connection-state handlers so callers can +// react to disconnects/reconnects without losing the default log lines. type retryConfig struct { - maxWait time.Duration - retryDelay time.Duration + maxWait time.Duration + retryDelay time.Duration + maxRetryDelay time.Duration + onDisconnect func(*nats.Conn, error) + onReconnect func(*nats.Conn) + onAttemptErr func(err error, attempt int) + ctx context.Context } -// RetryOption configures ConnectNATSWithRetry behavior. +// RetryOption configures ConnectNATS / ConnectNATSWithRetry behavior. type RetryOption func(*retryConfig) // WithMaxWait sets the maximum total time to keep retrying before giving up. @@ -82,22 +120,62 @@ func WithRetryDelay(d time.Duration) RetryOption { return func(c *retryConfig) { c.retryDelay = d } } +// WithMaxRetryDelay overrides the upper bound on the exponential backoff +// (default 10s). +func WithMaxRetryDelay(d time.Duration) RetryOption { + return func(c *retryConfig) { c.maxRetryDelay = d } +} + +// WithDisconnectHandler registers an optional callback invoked after the +// default disconnect log line. Callback runs on a NATS client goroutine; keep +// it non-blocking (atomic stores, channel sends, goroutine spawns). +func WithDisconnectHandler(fn func(*nats.Conn, error)) RetryOption { + return func(c *retryConfig) { c.onDisconnect = fn } +} + +// WithReconnectHandler registers an optional callback invoked after the +// default reconnect log line. Same goroutine constraints as WithDisconnectHandler. +func WithReconnectHandler(fn func(*nats.Conn)) RetryOption { + return func(c *retryConfig) { c.onReconnect = fn } +} + +// WithAttemptErrHandler registers an optional callback invoked after each +// failed connect attempt during ConnectNATSWithRetry's outer loop. Used by the +// daemon to surface initial-connect retries as a counter on /local/status. +func WithAttemptErrHandler(fn func(err error, attempt int)) RetryOption { + return func(c *retryConfig) { c.onAttemptErr = fn } +} + +// WithContext lets callers cancel the retry loop. When ctx is done, +// ConnectNATSWithRetry returns ctx.Err(). +func WithContext(ctx context.Context) RetryOption { + return func(c *retryConfig) { c.ctx = ctx } +} + // ConnectNATSWithRetry calls ConnectNATS in a retry loop with exponential -// backoff. It retries for up to 5 minutes (default) before giving up. TLS +// backoff. It retries for up to 5 minutes (default) before giving up; pass +// WithMaxWait(0) to retry indefinitely (cancel via WithContext). TLS // configuration errors (ErrCACertRead, ErrCACertParse) are permanent and // cause an immediate return without retrying. func ConnectNATSWithRetry(host, token, caCertPath string, opts ...RetryOption) (*nats.Conn, error) { cfg := retryConfig{ - maxWait: 5 * time.Minute, - retryDelay: 500 * time.Millisecond, + maxWait: 5 * time.Minute, + retryDelay: 500 * time.Millisecond, + maxRetryDelay: 10 * time.Second, } for _, o := range opts { o(&cfg) } + if cfg.maxRetryDelay <= 0 { + cfg.maxRetryDelay = 10 * time.Second + } start := time.Now() + attempt := 0 + var lastEscalatedLog time.Time for { - nc, err := ConnectNATS(host, token, caCertPath) + attempt++ + nc, err := ConnectNATS(host, token, caCertPath, opts...) if err == nil { if time.Since(start) > time.Second { slog.Info("NATS connection established", "elapsed", time.Since(start).Round(time.Second)) @@ -110,14 +188,38 @@ func ConnectNATSWithRetry(host, token, caCertPath string, opts ...RetryOption) ( return nil, fmt.Errorf("NATS TLS configuration error: %w", err) } + if cfg.onAttemptErr != nil { + cfg.onAttemptErr(err, attempt) + } + elapsed := time.Since(start) - if elapsed >= cfg.maxWait { + if cfg.maxWait > 0 && elapsed >= cfg.maxWait { return nil, fmt.Errorf("NATS connect failed after %s: %w", elapsed.Round(time.Second), err) } - slog.Warn("NATS not ready, retrying...", "error", err, "elapsed", elapsed.Round(time.Second), "retryIn", cfg.retryDelay) - time.Sleep(cfg.retryDelay) - cfg.retryDelay = min(cfg.retryDelay*2, 10*time.Second) + // Past the retry-escalation threshold (~30 attempts ≈ ~30 min disconnected + // at the 60s backoff cap) the warn-on-every-retry pattern can hide a + // stuck NATS config in routine warning noise. Escalate to slog.Error, + // rate-limited to once a minute, so the operator's error log surfaces it. + if attempt > natsRetryEscalateAttempt { + if lastEscalatedLog.IsZero() || time.Since(lastEscalatedLog) >= time.Minute { + slog.Error("NATS still disconnected", "error", err, "disconnected_for", elapsed.Round(time.Second), "attempt", attempt) + lastEscalatedLog = time.Now() + } + } else { + slog.Warn("NATS not ready, retrying...", "error", err, "elapsed", elapsed.Round(time.Second), "retryIn", cfg.retryDelay, "attempt", attempt) + } + + if cfg.ctx != nil { + select { + case <-cfg.ctx.Done(): + return nil, fmt.Errorf("NATS connect cancelled after %s: %w", elapsed.Round(time.Second), cfg.ctx.Err()) + case <-time.After(cfg.retryDelay): + } + } else { + time.Sleep(cfg.retryDelay) + } + cfg.retryDelay = min(cfg.retryDelay*2, cfg.maxRetryDelay) } } @@ -131,6 +233,10 @@ const AccountIDHeader = "X-Account-ID" // successful response into Out. Handlers can ignore the account ID if the // operation is unscoped (e.g. DescribeInstanceTypes). func NATSRequest[Out any](conn *nats.Conn, subject string, input any, timeout time.Duration, accountID string) (*Out, error) { + if conn == nil || !conn.IsConnected() { + return nil, ErrClusterUnavailable + } + jsonData, err := json.Marshal(input) if err != nil { return nil, fmt.Errorf("failed to marshal input: %w", err) @@ -179,6 +285,10 @@ const ( // error is returned. When expectedNodes > 0, collection exits early once that // many responses have been received. func NATSScatterGather[Out any](conn *nats.Conn, subject string, input any, timeout time.Duration, expectedNodes int, accountID string) (*Out, error) { + if conn == nil || !conn.IsConnected() { + return nil, ErrClusterUnavailable + } + jsonData, err := json.Marshal(input) if err != nil { return nil, fmt.Errorf("failed to marshal input: %w", err) diff --git a/spinifex/utils/nats_test.go b/spinifex/utils/nats_test.go index 71cd4649f..bd16c55d4 100644 --- a/spinifex/utils/nats_test.go +++ b/spinifex/utils/nats_test.go @@ -1,6 +1,7 @@ package utils import ( + "bytes" "crypto/ecdsa" "crypto/elliptic" "crypto/rand" @@ -9,10 +10,13 @@ import ( "crypto/x509/pkix" "encoding/json" "encoding/pem" + "log/slog" "math/big" "net" "os" "path/filepath" + "strconv" + "strings" "testing" "time" @@ -573,3 +577,247 @@ func TestConnectNATSWithRetry_TLSErrorNoRetry(t *testing.T) { assert.Contains(t, err.Error(), "NATS TLS configuration error") assert.Less(t, elapsed, time.Second, "should fail immediately without retrying") } + +// --- Disconnect/Reconnect callback tests --- + +func TestConnectNATS_DisconnectCallbackFires(t *testing.T) { + ns := startTestNATSServer(t) + + disconnected := make(chan struct{}, 1) + nc, err := ConnectNATS(ns.ClientURL(), "", "", + WithDisconnectHandler(func(_ *nats.Conn, _ error) { + select { + case disconnected <- struct{}{}: + default: + } + }), + ) + require.NoError(t, err) + defer nc.Close() + require.True(t, nc.IsConnected()) + + ns.Shutdown() + + select { + case <-disconnected: + case <-time.After(3 * time.Second): + t.Fatal("disconnect callback did not fire") + } +} + +func TestConnectNATS_ReconnectCallbackFires(t *testing.T) { + // Pin the test NATS to a specific port so we can restart it on the same URL. + port := freePort(t) + ns := startTestNATSOnPort(t, port) + + reconnected := make(chan struct{}, 1) + nc, err := ConnectNATS(ns.ClientURL(), "", "", + WithReconnectHandler(func(_ *nats.Conn) { + select { + case reconnected <- struct{}{}: + default: + } + }), + ) + require.NoError(t, err) + defer nc.Close() + require.True(t, nc.IsConnected()) + + ns.Shutdown() + // Wait until the client noticed the drop so the reconnect path runs. + require.Eventually(t, func() bool { return !nc.IsConnected() }, 3*time.Second, 50*time.Millisecond) + + startTestNATSOnPort(t, port) + + select { + case <-reconnected: + case <-time.After(5 * time.Second): + t.Fatal("reconnect callback did not fire") + } +} + +// --- Fast-fail when disconnected --- + +func TestNATSRequest_DisconnectedFastFail(t *testing.T) { + ns := startTestNATSServer(t) + nc, err := ConnectNATS(ns.ClientURL(), "", "") + require.NoError(t, err) + defer nc.Close() + + ns.Shutdown() + require.Eventually(t, func() bool { return !nc.IsConnected() }, 3*time.Second, 50*time.Millisecond) + + start := time.Now() + _, err = NATSRequest[map[string]any](nc, "ec2.Describe", struct{}{}, 5*time.Second, "") + elapsed := time.Since(start) + + require.ErrorIs(t, err, ErrClusterUnavailable) + assert.Less(t, elapsed, 500*time.Millisecond, "should bail before per-call timeout") +} + +func TestNATSScatterGather_DisconnectedFastFail(t *testing.T) { + ns := startTestNATSServer(t) + nc, err := ConnectNATS(ns.ClientURL(), "", "") + require.NoError(t, err) + defer nc.Close() + + ns.Shutdown() + require.Eventually(t, func() bool { return !nc.IsConnected() }, 3*time.Second, 50*time.Millisecond) + + start := time.Now() + _, err = NATSScatterGather[map[string]any](nc, "ec2.Describe", struct{}{}, 5*time.Second, 0, "") + elapsed := time.Since(start) + + require.ErrorIs(t, err, ErrClusterUnavailable) + assert.Less(t, elapsed, 500*time.Millisecond, "should bail before fan-out timeout") +} + +// --- 1c fail-fast tests: NATS request helpers reject when conn is down --- + +func TestNATSRequest_NilConn_ReturnsClusterUnavailable(t *testing.T) { + type Resp struct{} + _, err := NATSRequest[Resp](nil, "test.never", struct{}{}, 50*time.Millisecond, "") + require.Error(t, err) + assert.ErrorIs(t, err, ErrClusterUnavailable) +} + +func TestNATSRequest_ClosedConn_ReturnsClusterUnavailable(t *testing.T) { + ns := startTestNATSServer(t) + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + nc.Close() + + type Resp struct{} + _, err = NATSRequest[Resp](nc, "test.never", struct{}{}, 50*time.Millisecond, "") + require.Error(t, err) + assert.ErrorIs(t, err, ErrClusterUnavailable) +} + +func TestNATSScatterGather_NilConn_ReturnsClusterUnavailable(t *testing.T) { + type Resp struct{} + _, err := NATSScatterGather[Resp](nil, "test.never", struct{}{}, 50*time.Millisecond, 1, "") + require.Error(t, err) + assert.ErrorIs(t, err, ErrClusterUnavailable) +} + +func TestNATSScatterGather_ClosedConn_ReturnsClusterUnavailable(t *testing.T) { + ns := startTestNATSServer(t) + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + nc.Close() + + type Resp struct{} + _, err = NATSScatterGather[Resp](nc, "test.never", struct{}{}, 50*time.Millisecond, 1, "") + require.Error(t, err) + assert.ErrorIs(t, err, ErrClusterUnavailable) +} + +// --- 1c callback hook plumbing: WithDisconnectHandler / WithReconnectHandler --- + +func TestConnectNATS_DisconnectReconnectCallbacks(t *testing.T) { + port := freePort(t) + ns := startTestNATSOnPort(t, port) + + disconnects := make(chan struct{}, 4) + reconnects := make(chan struct{}, 4) + + nc, err := ConnectNATS("nats://127.0.0.1:"+strconv.Itoa(port), "", "", + WithDisconnectHandler(func(_ *nats.Conn, _ error) { disconnects <- struct{}{} }), + WithReconnectHandler(func(_ *nats.Conn) { reconnects <- struct{}{} }), + ) + require.NoError(t, err) + defer nc.Close() + + ns.Shutdown() + select { + case <-disconnects: + case <-time.After(3 * time.Second): + t.Fatal("disconnect callback never fired") + } + + startTestNATSOnPort(t, port) + select { + case <-reconnects: + case <-time.After(5 * time.Second): + t.Fatal("reconnect callback never fired") + } +} + +// --- helpers for restartable NATS server --- + +func freePort(t *testing.T) int { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + addr, ok := l.Addr().(*net.TCPAddr) + require.True(t, ok) + port := addr.Port + require.NoError(t, l.Close()) + return port +} + +func startTestNATSOnPort(t *testing.T, port int) *server.Server { + t.Helper() + opts := &server.Options{ + Host: "127.0.0.1", + Port: port, + NoLog: true, + NoSigs: true, + } + ns, err := server.NewServer(opts) + require.NoError(t, err) + go ns.Start() + require.True(t, ns.ReadyForConnections(5*time.Second)) + t.Cleanup(func() { ns.Shutdown() }) + return ns +} + +// TestConnectNATSWithRetry_LogEscalatesPastThreshold drives ConnectNATSWithRetry +// against a closed port with a tight retry loop until the attempt count exceeds +// natsRetryEscalateAttempt, then asserts that at least one escalated slog.Error +// line was emitted with disconnected_for context (rate-limited to once per +// minute) while earlier attempts stayed at slog.Warn. +func TestConnectNATSWithRetry_LogEscalatesPastThreshold(t *testing.T) { + var buf bytes.Buffer + prev := slog.Default() + t.Cleanup(func() { slog.SetDefault(prev) }) + slog.SetDefault(slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))) + + _, err := ConnectNATSWithRetry("nats://127.0.0.1:1", "", "", + WithRetryDelay(1*time.Millisecond), + WithMaxRetryDelay(1*time.Millisecond), + WithMaxWait(300*time.Millisecond), + ) + require.Error(t, err) + + logs := buf.String() + warnCount := strings.Count(logs, "level=WARN msg=\"NATS not ready, retrying...\"") + errCount := strings.Count(logs, "level=ERROR msg=\"NATS still disconnected\"") + + assert.GreaterOrEqual(t, warnCount, 1, "expect warn logs for first 30 attempts") + assert.GreaterOrEqual(t, errCount, 1, "expect at least one escalated error log past the threshold") + assert.LessOrEqual(t, errCount, 2, "rate-limited to once per minute, so a sub-second test should see at most one or two") + assert.Contains(t, logs, "disconnected_for=", "escalated error should include disconnected_for") +} + +// TestConnectNATSWithRetry_NoEscalation_BelowThreshold keeps the attempt count +// under natsRetryEscalateAttempt and checks that no escalated slog.Error line +// is produced. +func TestConnectNATSWithRetry_NoEscalation_BelowThreshold(t *testing.T) { + var buf bytes.Buffer + prev := slog.Default() + t.Cleanup(func() { slog.SetDefault(prev) }) + slog.SetDefault(slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))) + + // Exponential backoff capped at 5ms × maxWait 50ms = ≲15 attempts < 30. + _, err := ConnectNATSWithRetry("nats://127.0.0.1:1", "", "", + WithRetryDelay(5*time.Millisecond), + WithMaxRetryDelay(5*time.Millisecond), + WithMaxWait(50*time.Millisecond), + ) + require.Error(t, err) + + logs := buf.String() + assert.NotContains(t, logs, "NATS still disconnected", "should not escalate before threshold") + assert.Contains(t, logs, "NATS not ready, retrying...", "should still log warn lines") +}