From 6c54a6790c5b931e9402f7e94bc91ffd4a9a789b Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Sun, 10 May 2026 10:08:12 +1000 Subject: [PATCH] =?UTF-8?q?feat(daemon):=20DDIL=20Tier=201=20=E2=80=94=20d?= =?UTF-8?q?aemon-local-autonomy=20phase=201=20(#204)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(daemon): extract Describe instance handlers to service (1b-pre phase 1) Collapse the Describe family in daemon_handlers_instance.go to thin Pattern A wrappers over a transport-neutral InstanceServiceImpl. This brings the read path in line with the rest of the codebase and is the prerequisite for 1b's local HTTP API (mulga-siv-14, daemon-local-autonomy plan). Extracted: DescribeInstances, DescribeInstanceTypes, DescribeInstanceAttribute, DescribeStoppedInstances, DescribeTerminatedInstances. Filter helpers (instanceMatchesFilters, matchTagKey/Value, DescribeInstancesValidFilters, IsInstanceVisible) move to the instance package alongside the service methods. InstanceService gains the new methods and two dep interfaces (ResourceCapacityProvider, StoppedInstanceStore) so the daemon's ResourceManager and JetStreamManager can plug in without leaking concrete types. NATSInstanceService returns a clear error for the fan-out describes (DescribeInstances, DescribeInstanceTypes) since cluster-wide aggregation lives in the gateway's scatter-gather helpers, not a single-node NATS request. Behaviour changes: malformed JSON on DescribeInstanceAttribute now responds with ValidationError (matching every other Pattern A handler) instead of ServerInternal. Test updated. Mutators (RunInstances, ModifyInstanceAttribute, StartStoppedInstance, TerminateStoppedInstance, sub-handlers) remain Pattern B for now; follow-up phases of mulga-siv-14 will collapse them. * feat(daemon): local-first instance state persistence (1a) Make /var/lib/spinifex/state/instance-state.json the source of truth for a node's VMs; JetStream KV becomes a best-effort cluster cache. The daemon now survives KV unavailability without losing instance state. - New local_state.go: atomic write (tmp+fsync+rename), schema_version gate, missing-file = fresh install, corrupt = refuse start. - Daemon.WriteState writes the local file under Instances.Mu, then calls jsManager.WriteStateBestEffort with a 1s timeout. KV failures log and continue; clients never see "KV down" during partition. - Daemon.LoadState reads from the local file only — no KV fallback (a fresh node has empty KV anyway, so silent fallback would mask bugs). - restoreInstances now returns error and Start() aborts on corrupt local state rather than silently starting empty. - jsManager.WriteState no longer self-locks (caller holds Mu); added WriteStateBestEffort and shared marshalInstanceState helper. Reconnect-driven KV resync deferred to siv-16 (1c owns the callback wiring). Bead: mulga-siv-15 Plan: docs/development/feature/daemon-local-state-persistence.md * fix(daemon): root local state at DataDir not BaseDir LocalStatePath was falling back to BaseDir when DataDir was empty, putting the file at /var/lib/spinifex/spinifex/state/instance-state.json (under the daemon's own subdir) instead of the documented /var/lib/spinifex/state/instance-state.json (the platform-shared root). Drop the BaseDir fallback. Empty DataDir resolves to DefaultDataDir (/var/lib/spinifex). Tests now set DataDir alongside BaseDir on their config fixtures. * feat(daemon): graceful NATS disconnect/reconnect handling (1c) Daemon now flips a `mode` field (cluster|standalone) on NATS disconnect/reconnect via optional callback hooks added to utils.ConnectNATS. Reconnect bumps natsRetryCount and triggers 1a's full local-state push to KV. NATSRequest/NATSScatterGather fail fast with ErrClusterUnavailable when the connection is down. Gateway short-circuits NATS-bound requests with a 503 ServiceUnavailable carrying "cluster unavailable: NATS disconnected — check daemon /local/status" so AWS clients see the failure mode immediately instead of hanging on per-call timeouts. Plan: docs/development/feature/daemon-autonomy-1c-disconnect.md * fix(gateway): SigV4 middleware short-circuits on NATS disconnect The 1c gateway.Request() IsConnected check (gateway.go:234) was unreachable for any authenticated path because SigV4 auth middleware runs first and calls IAMService.LookupAccessKey over JetStream KV. With NATS down the lookup hangs ~5s on context.DeadlineExceeded and the gateway returns 500 InternalError. AWS SDK retries it twice for ~18s of user-visible latency instead of the <100ms 503 promised by the 1c plan. Mirror the IsConnected check at the top of SigV4AuthMiddleware (after credential parse so malformed auth headers still get the right SigV4 error code, before the IAM lookup). The catch-all in gateway.Request() stays as defence-in-depth for any future unauthed paths. Coverage debt picked up alongside: 1c utils.NATSRequest / NATSScatterGather fail-fast paths, daemon Mode/onNATSDisconnect/ onNATSReconnect/localStatePath/Write/LoadState branches, and 1b-pre InstanceServiceImpl Describe* methods (partial siv-22). Bead: mulga-siv-23 * feat(daemon): read-only /local/* HTTP API (1b) Wires GET /local/instances, GET /local/instances/{id}, GET /local/status onto the existing ClusterManager TLS listener so observers can read this node's instance state and connectivity mode while NATS is unreachable. Also fixes a pre-existing race in WriteState: json.Marshal walked shared *vm.VM pointers from SnapshotMap while concurrent TransitionState calls mutated VM fields under the manager lock. Marshal now runs inside vmMgr.View; WriteLocalStateBytes / WriteStateBytesBestEffort accept the pre-marshalled JSON and commit lock-free. TestTransitionState_Concurrent Transitions is race-clean across 30 iterations. Adds Daemon.stateRevision (atomic.Uint64) bumped on every successful local-state write, surfaced as /local/status.revision so observers can detect changes without diffing payloads. * feat(daemon): two-phase startup — survive NATS down (1d) Split daemon Start() into startLocal() (HTTPS, mgmt bridge, OVS plumber, OOM, local state load — no NATS) and startCluster() (background, infinite retry exp backoff cap 60s for NATS connect, then JetStream + services + restoreInstances + subscribeAll + heartbeat). Daemon never process-exits on NATS failure; instead it stays up in standalone mode and reconnects in the background. Drop ExecStartPre=wait-for-nats.sh from spinifex-daemon.service — local bootstrap no longer requires NATS to be reachable. ConnectNATSWithRetry now supports infinite retry (WithMaxWait(0)), context cancellation (WithContext), per-attempt error callback for the nats_retry_count operator signal on /local/status, and a configurable backoff cap (WithMaxRetryDelay). bead: mulga-siv-17 (closes) * feat(daemon): Tier 1 invariant guard for cluster-scoped service init (1e-audit) Make the DDIL §1e-audit invariant explicit: startLocal must not initialise any NATS-dependent or JetStream-KV-backed handle. Adds assertNoClusterServicesInitialised as a runtime guard at the end of startLocal that fails fast if a future edit hoists a cluster-scoped service into Phase 1, plus comment blocks naming the invariant on both startLocal and startCluster. Code audit confirms the existing 1d split already keeps every cluster-scoped bucket out of Phase 1; this change pins the property so it cannot regress silently. Tier table for the 25 daemon-process KV buckets is finalised in docs/development/feature/daemon-local-autonomy.md §1e-audit. bead: mulga-siv-19 (closes) * feat(daemon): NATS observability hardening (1d-obs) Closes the residual visibility gap left by the 1d non-blocking startup: - Extend /local/status with kv_sync_failures, last_kv_sync_at, last_kv_sync_error so an operator polling the local API sees prolonged divergence between the local file and JetStream KV. - Add a KVSyncObserver interface on JetStreamManager so the daemon can count best-effort write outcomes without coupling the manager to daemon state. - Escalate utils.ConnectNATSWithRetry per-attempt log to slog.Error (rate-limited to once per minute, with disconnected_for context) once attempt count crosses the natsRetryEscalateAttempt threshold (~30 attempts ≈ ~30 min disconnected at the 60s backoff cap), so a stuck NATS config surfaces in error logs instead of routine warn noise. Prometheus metrics export deferred until a system-wide telemetry plan (likely OTel) lands; the structured /local/status fields plus the escalated retry log cover the Tier 1 visibility gap. * feat(daemon): opt-in SPINIFEX_REQUIRE_NATS strict-startup mode (1d-strict) When SPINIFEX_REQUIRE_NATS=1, startCluster's first NATS connect uses a bounded ~30s retry and the daemon aborts on timeout. Default behaviour (env unset) keeps the Tier 1 infinite-retry path so prod / edge nodes still survive a missing NATS at boot. Restores pre-DDIL fail-fast UX for dev / test / single-node deploys without flipping the prod default. mulga-siv-44. * test(scripts): systemd-level smoke for DDIL phase 1 Bash harness exercising daemon-local-autonomy on a real `make deploy` node: baseline cluster mode → kill spinifex-nats → assert standalone + retry counter advance + /local/instances served → restart NATS → assert mode flip + KV resync stamp → SPINIFEX_REQUIRE_NATS=1 bounded-exit. Cleanup trap restores both services. Complements existing unit coverage by validating real systemd wiring on port 8443. * fix(test-ddil): daemon ClusterManager binds 4432 not 8443 8443 is predastore. Daemon's HTTPS ClusterManager (where /local/* lives) binds [nodes.node1.daemon].host = 0.0.0.0:4432 in the deployed config. Plan doc daemon-local-autonomy.md:112 incorrectly said 8443; actual code uses d.config.Daemon.Host. * fix(test-ddil): correct retry-counter site + bypass NATS Wants= for strict test Two bugs in the v1 script: 1) nats_retry_count assertion ran during disconnect — but the counter only bumps on disconnect→reconnect *cycles* (daemon.go:321, onNATSReconnect). Moved assertion to step 3 (restart NATS) where the reconnect callback fires. 2) Step 4 stopped spinifex-nats then ran 'systemctl restart spinifex-daemon' — but the daemon unit's Wants=spinifex-nats.service re-pulled NATS up via the Wants= chain, so the strict-mode 30s timeout never fired. systemctl mask refuses on /etc-installed units, and --runtime mask is shadowed by /etc. Block via a runtime drop-in with ConditionPathExists=! pointing at a sentinel file. Verified: all 5 checks pass on a single-node make deploy box; daemon exits 32s after restart with SPINIFEX_REQUIRE_NATS=1 + NATS blocked. * fix(admin): render data_dir into spinifex.toml so non-root nodes write local state Pseudo-multinode E2E (rootless tf-user, ~/nodeN data roots) hit "mkdir /var/lib/spinifex/state: permission denied" on every RunInstances: Config.DataDir was empty because the template never emitted a data_dir key, and LocalStatePath fell back to DefaultDataDir (/var/lib/spinifex). Surfaced after 8eb9c5fa dropped the BaseDir fallback in LocalStatePath — previously SPINIFEX_BASE_DIR=$DATA_DIR/spinifex/ (set by start-dev.sh) masked the missing data_dir. ConfigSettings.DataDir is already populated to spxRoot at admin init/join; render it under [nodes.X] so it actually reaches the daemon. --- build/systemd/spinifex-daemon.service | 10 +- cmd/spinifex/cmd/templates/spinifex.toml | 1 + scripts/test-ddil-phase1.sh | 121 +++++ spinifex/daemon/daemon.go | 505 +++++++++++++++--- spinifex/daemon/daemon_handlers_instance.go | 47 +- spinifex/daemon/daemon_handlers_test.go | 9 + spinifex/daemon/daemon_test.go | 89 ++- spinifex/daemon/disconnect_test.go | 439 +++++++++++++++ spinifex/daemon/jetstream.go | 97 +++- spinifex/daemon/jetstream_test.go | 90 ++++ spinifex/daemon/local_api.go | 128 +++++ spinifex/daemon/local_api_test.go | 188 +++++++ spinifex/daemon/local_state.go | 155 ++++++ spinifex/daemon/local_state_test.go | 115 ++++ spinifex/daemon/state_test.go | 4 +- spinifex/gateway/auth.go | 13 + spinifex/gateway/auth_test.go | 55 ++ spinifex/gateway/gateway.go | 52 ++ spinifex/gateway/gateway_test.go | 76 ++- spinifex/gateway/ratelimit_test.go | 21 +- .../ec2/instance/instance_filter_test.go | 215 ++++++++ spinifex/handlers/ec2/instance/service.go | 24 +- .../handlers/ec2/instance/service_impl.go | 415 +++++++++++++- .../ec2/instance/service_impl_test.go | 430 +++++++++++++++ .../handlers/ec2/instance/service_mock.go | 20 + .../handlers/ec2/instance/service_nats.go | 33 +- spinifex/utils/nats.go | 144 ++++- spinifex/utils/nats_test.go | 248 +++++++++ 28 files changed, 3576 insertions(+), 168 deletions(-) create mode 100755 scripts/test-ddil-phase1.sh create mode 100644 spinifex/daemon/disconnect_test.go create mode 100644 spinifex/daemon/local_api.go create mode 100644 spinifex/daemon/local_api_test.go create mode 100644 spinifex/daemon/local_state.go create mode 100644 spinifex/daemon/local_state_test.go create mode 100644 spinifex/handlers/ec2/instance/instance_filter_test.go 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") +}