From 97701737cbf81d54a25e65f9130b02226a699d63 Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Wed, 13 May 2026 15:55:20 +1000 Subject: [PATCH] feat: describe instance status (#221) --- .github/workflows/e2e-ddil.yml | 2 +- .github/workflows/e2e-nightly.yml | 2 +- .github/workflows/e2e.yml | 6 +- docs/COMMANDS.md | 1 + spinifex/daemon/daemon.go | 3 +- spinifex/daemon/daemon_handlers_instance.go | 5 + spinifex/daemon/daemon_handlers_test.go | 56 +++ spinifex/gateway/ec2.go | 3 + .../ec2/instance/DescribeInstanceStatus.go | 276 +++++++++++ .../instance/DescribeInstanceStatus_test.go | 468 ++++++++++++++++++ spinifex/gateway/gateway_test.go | 2 +- spinifex/handlers/ec2/instance/service.go | 1 + .../handlers/ec2/instance/service_impl.go | 157 ++++++ .../ec2/instance/service_impl_test.go | 228 +++++++++ .../handlers/ec2/instance/service_mock.go | 4 + .../handlers/ec2/instance/service_nats.go | 6 + 16 files changed, 1213 insertions(+), 7 deletions(-) create mode 100644 spinifex/gateway/ec2/instance/DescribeInstanceStatus.go create mode 100644 spinifex/gateway/ec2/instance/DescribeInstanceStatus_test.go diff --git a/.github/workflows/e2e-ddil.yml b/.github/workflows/e2e-ddil.yml index 0b1d5c93c..2665c5df1 100644 --- a/.github/workflows/e2e-ddil.yml +++ b/.github/workflows/e2e-ddil.yml @@ -30,7 +30,7 @@ jobs: TOFU_LOG: /tmp/e2e-tofu-ddil-${{ github.run_id }}.log steps: - name: Clean Workspace - run: rm -rf "$GITHUB_WORKSPACE"/* + run: sudo rm -rf "$GITHUB_WORKSPACE"/* - uses: actions/create-github-app-token@v3 id: app-token diff --git a/.github/workflows/e2e-nightly.yml b/.github/workflows/e2e-nightly.yml index c18697d15..67aa55af3 100644 --- a/.github/workflows/e2e-nightly.yml +++ b/.github/workflows/e2e-nightly.yml @@ -61,7 +61,7 @@ jobs: - name: Clean Workspace if: steps.gate.outputs.selected == 'true' - run: rm -rf "$GITHUB_WORKSPACE"/* + run: sudo rm -rf "$GITHUB_WORKSPACE"/* - uses: actions/create-github-app-token@v3 if: steps.gate.outputs.selected == 'true' diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index 90666cad2..7fbc05d7f 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -34,7 +34,7 @@ jobs: TOFU_LOG: /tmp/e2e-tofu-single-${{ github.run_id }}.log steps: - name: Clean Workspace - run: rm -rf "$GITHUB_WORKSPACE"/* + run: sudo rm -rf "$GITHUB_WORKSPACE"/* - uses: actions/create-github-app-token@v3 id: app-token @@ -103,7 +103,7 @@ jobs: TOFU_LOG: /tmp/e2e-tofu-pseudo-${{ github.run_id }}.log steps: - name: Clean Workspace - run: rm -rf "$GITHUB_WORKSPACE"/* + run: sudo rm -rf "$GITHUB_WORKSPACE"/* - uses: actions/create-github-app-token@v3 id: app-token @@ -167,7 +167,7 @@ jobs: TOFU_LOG: /tmp/e2e-tofu-multi-${{ github.run_id }}.log steps: - name: Clean Workspace - run: rm -rf "$GITHUB_WORKSPACE"/* + run: sudo rm -rf "$GITHUB_WORKSPACE"/* - uses: actions/create-github-app-token@v3 id: app-token diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index f834cb334..5911b136c 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -149,6 +149,7 @@ file, deferred to a later phase. | `get-console-output` | `--instance-id` | `--latest` (always returns latest), `--dry-run` | Instance must be running on a node | Gateway sends NATS `ec2.{instanceId}.GetConsoleOutput` (per-instance topic, routed to owning node) → daemon reads console log file from disk → returns last 64KB base64-encoded with timestamp. Always available regardless of serial console access setting (matches AWS behavior). | 1. Get output from running instance
2. Empty log file returns empty output
3. Instance not found (error: InvalidInstanceID.NotFound) | **DONE** | | `describe-instance-attribute` | `--instance-id`, `--attribute` (instanceType, userData, instanceInitiatedShutdownBehavior, disableApiTermination, disableApiStop, ebsOptimized, enaSupport, sourceDestCheck, rootDeviceName, kernel, ramdisk) | `--dry-run` | Instance must exist (running or stopped) | Gateway validates input → NATS `ec2.DescribeInstanceAttribute` with `spinifex-workers` queue group → daemon checks running instances first (`d.Instances.VMS`), then stopped instances in JetStream KV → returns single attribute per call (matches AWS behavior). Stored attributes (`instanceType`, `userData`) return real values; unstored attributes return AWS defaults (`instanceInitiatedShutdownBehavior`=stop, `disableApiTermination`=false, etc.) | 1. Get instanceType from running instance
2. Get userData from stopped instance
3. Get default disableApiTermination
4. Invalid attribute name (error)
5. Instance not found (error: InvalidInstanceID.NotFound) | **DONE** | | `describe-instance-credit-specifications` | `--instance-ids` | `--filters`, `--max-results`, `--dry-run` | None | Gateway-only stub — returns `CpuCredits: "standard"` for each requested instance ID. No daemon round-trip. T-series credit mode is not persisted. | 1. Get credit spec for T-series instance
2. Multiple instance IDs | **DONE** | +| `describe-instance-status` | `--instance-ids`, `--include-all-instances`, `--filters` (availability-zone, instance-state-code, instance-state-name, tag:\*) | `--max-results`, `--next-token`, `--dry-run` (silently ignored), `event.*` / `instance-status.*` / `system-status.*` filters (rejected with InvalidParameterValue) | None | Gateway fans out NATS `ec2.DescribeInstanceStatus` to all nodes (no queue group). Each daemon validates input (i- prefix, filter names), iterates local VMs and renders InstanceStatus: `Running`→ok/passed, everything-else-included→not-applicable. When `IncludeAllInstances=true`, gateway also queries `ec2.DescribeStoppedInstances` in parallel and transforms reservation entries into InstanceStatus with `not-applicable`. Aggregated results dedup'd by InstanceId (running > stopped). Terminated and error-state instances never returned. No real health probing — static values reflect Mulga's lifecycle state only. | 1. Status of running instance (ok/passed)
2. Filter by instance ID
3. Default excludes stopped instances
4. IncludeAllInstances=true includes stopped (not-applicable)
5. Cross-account instance excluded
6. Unknown filter returns InvalidParameterValue
7. Malformed instance ID rejected
8. Tag filter narrows correctly | **DONE** | | `monitor-instances` | — | `--instance-ids` | Instance must exist | Enable basic monitoring (CPU, disk, network) → store metrics in NATS KV → return monitoring state | 1. Enable monitoring
2. Verify monitoring state in describe-instances | **NOT STARTED** | | `unmonitor-instances` | — | `--instance-ids` | Instance must exist | Disable detailed monitoring → revert to basic monitoring → return monitoring state | 1. Disable monitoring
2. Verify monitoring state reverts in describe-instances | **NOT STARTED** | diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index d37063bc9..ed7773847 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -746,8 +746,9 @@ func (d *Daemon) subscribeAll() error { {"ec2.terminate", d.handleEC2TerminateStoppedInstance, "spinifex-workers"}, {"ec2.DescribeStoppedInstances", d.handleEC2DescribeStoppedInstances, "spinifex-workers"}, {"ec2.DescribeTerminatedInstances", d.handleEC2DescribeTerminatedInstances, "spinifex-workers"}, - // these 2 fan out to all nodes and gateway aggregates the results + // these 3 fan out to all nodes and gateway aggregates the results {"ec2.DescribeInstances", d.handleEC2DescribeInstances, ""}, + {"ec2.DescribeInstanceStatus", d.handleEC2DescribeInstanceStatus, ""}, {"ec2.DescribeInstanceTypes", d.handleEC2DescribeInstanceTypes, ""}, {"ec2.EnableEbsEncryptionByDefault", d.handleEC2EnableEbsEncryptionByDefault, "spinifex-workers"}, {"ec2.DisableEbsEncryptionByDefault", d.handleEC2DisableEbsEncryptionByDefault, "spinifex-workers"}, diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index 88eab3273..e5e03e54c 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -313,6 +313,11 @@ func (d *Daemon) handleEC2DescribeInstanceTypes(msg *nats.Msg) { handleNATSRequest(msg, d.instanceService.DescribeInstanceTypes) } +// handleEC2DescribeInstanceStatus responds with status entries for VMs on this node visible to the caller. +func (d *Daemon) handleEC2DescribeInstanceStatus(msg *nats.Msg) { + handleNATSRequest(msg, d.instanceService.DescribeInstanceStatus) +} + // startStoppedInstanceRequest is the payload for ec2.start topic func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { handleNATSRequest(msg, d.instanceService.StartStoppedInstance) diff --git a/spinifex/daemon/daemon_handlers_test.go b/spinifex/daemon/daemon_handlers_test.go index fe4ba3929..00992a9e9 100644 --- a/spinifex/daemon/daemon_handlers_test.go +++ b/spinifex/daemon/daemon_handlers_test.go @@ -3483,6 +3483,62 @@ func TestHandleEC2DescribeInstances_MalformedInstanceID(t *testing.T) { assert.Contains(t, string(reply.Data), "InvalidInstanceID.Malformed") } +// --- handleEC2DescribeInstanceStatus round-trip --- + +func TestHandleEC2DescribeInstanceStatus_RoundTrip(t *testing.T) { + daemon := createFullTestDaemon(t, sharedNATSURL) + + runningVM := &vm.VM{ + ID: "i-status-001", + Status: vm.StateRunning, + AccountID: testAccountID, + Reservation: &ec2.Reservation{ + ReservationId: aws.String("r-status-001"), + OwnerId: aws.String(testAccountID), + }, + Instance: &ec2.Instance{InstanceId: aws.String("i-status-001")}, + } + daemon.vmMgr.Insert(runningVM) + t.Cleanup(func() { daemon.vmMgr.Delete(runningVM.ID) }) + + sub, err := daemon.natsConn.Subscribe("ec2.DescribeInstanceStatus.rt", daemon.handleEC2DescribeInstanceStatus) + require.NoError(t, err) + defer sub.Unsubscribe() + + reqData, _ := json.Marshal(&ec2.DescribeInstanceStatusInput{}) + reply, err := natsRequest(daemon.natsConn, "ec2.DescribeInstanceStatus.rt", reqData, 5*time.Second) + require.NoError(t, err) + + var output ec2.DescribeInstanceStatusOutput + require.NoError(t, json.Unmarshal(reply.Data, &output)) + + found := false + for _, s := range output.InstanceStatuses { + if s.InstanceId != nil && *s.InstanceId == "i-status-001" { + found = true + assert.Equal(t, "running", *s.InstanceState.Name) + assert.Equal(t, "ok", *s.InstanceStatus.Status) + assert.Equal(t, "ok", *s.SystemStatus.Status) + } + } + assert.True(t, found, "expected running instance in DescribeInstanceStatus output") +} + +func TestHandleEC2DescribeInstanceStatus_MalformedInstanceID(t *testing.T) { + daemon := createFullTestDaemon(t, sharedNATSURL) + + sub, err := daemon.natsConn.Subscribe("ec2.DescribeInstanceStatus.malformed", daemon.handleEC2DescribeInstanceStatus) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.DescribeInstanceStatusInput{InstanceIds: []*string{aws.String("not-an-id")}} + reqData, _ := json.Marshal(input) + reply, err := natsRequest(daemon.natsConn, "ec2.DescribeInstanceStatus.malformed", reqData, 5*time.Second) + require.NoError(t, err) + + assert.Contains(t, string(reply.Data), "InvalidInstanceID.Malformed") +} + // --- Terminated instance handler tests --- func TestHandleEC2DescribeTerminatedInstances_ReturnsTerminatedInstances(t *testing.T) { diff --git a/spinifex/gateway/ec2.go b/spinifex/gateway/ec2.go index a050beff1..6e61a9fb3 100644 --- a/spinifex/gateway/ec2.go +++ b/spinifex/gateway/ec2.go @@ -80,6 +80,9 @@ var ec2Actions = map[string]EC2Handler{ "DescribeInstanceTypes": ec2Handler(func(input *ec2.DescribeInstanceTypesInput, gw *GatewayConfig, accountID string) (any, error) { return gateway_ec2_instance.DescribeInstanceTypes(input, gw.NATSConn, gw.ExpectedNodes) }), + "DescribeInstanceStatus": ec2Handler(func(input *ec2.DescribeInstanceStatusInput, gw *GatewayConfig, accountID string) (any, error) { + return gateway_ec2_instance.DescribeInstanceStatus(input, gw.NATSConn, gw.DiscoverActiveNodes(), accountID, gw.AZ) + }), "GetConsoleOutput": ec2Handler(func(input *ec2.GetConsoleOutputInput, gw *GatewayConfig, accountID string) (any, error) { return gateway_ec2_instance.GetConsoleOutput(input, gw.NATSConn, accountID) }), diff --git a/spinifex/gateway/ec2/instance/DescribeInstanceStatus.go b/spinifex/gateway/ec2/instance/DescribeInstanceStatus.go new file mode 100644 index 000000000..a430acec0 --- /dev/null +++ b/spinifex/gateway/ec2/instance/DescribeInstanceStatus.go @@ -0,0 +1,276 @@ +package gateway_ec2_instance + +import ( + "encoding/json" + "errors" + "fmt" + "log/slog" + "strconv" + "time" + + "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/utils" + "github.com/nats-io/nats.go" +) + +const ( + notApplicable = "not-applicable" + reachabilityDetailName = "reachability" +) + +// DescribeInstanceStatus fans out to every node and, with IncludeAllInstances, +// also augments from the stopped-instance KV bucket. The aggregator propagates +// the first deterministic 4xx only when no data was collected, matching +// DescribeInstances. +func DescribeInstanceStatus(input *ec2.DescribeInstanceStatusInput, natsConn *nats.Conn, expectedNodes int, accountID, az string) (*ec2.DescribeInstanceStatusOutput, error) { + if input == nil { + input = &ec2.DescribeInstanceStatusInput{} + } + + jsonData, err := json.Marshal(input) + if err != nil { + slog.Error("DescribeInstanceStatus: Failed to marshal input", "err", err) + return nil, fmt.Errorf("failed to marshal input: %w", err) + } + + inbox := nats.NewInbox() + sub, err := natsConn.SubscribeSync(inbox) + if err != nil { + slog.Error("DescribeInstanceStatus: Failed to create inbox subscription", "err", err) + return nil, fmt.Errorf("failed to create inbox: %w", err) + } + defer sub.Unsubscribe() + + pubMsg := nats.NewMsg("ec2.DescribeInstanceStatus") + pubMsg.Reply = inbox + pubMsg.Data = jsonData + pubMsg.Header.Set(utils.AccountIDHeader, accountID) + if err := natsConn.PublishMsg(pubMsg); err != nil { + slog.Error("DescribeInstanceStatus: Failed to publish request", "err", err) + return nil, fmt.Errorf("failed to publish request: %w", err) + } + + timeout := 3 * time.Second + deadline := time.Now().Add(timeout) + + var allStatuses []*ec2.InstanceStatus + var clientError string + responsesReceived := 0 + + if expectedNodes <= 0 { + expectedNodes = -1 + slog.Warn("DescribeInstanceStatus: ExpectedNodes not configured, using timeout-only collection") + } + + for time.Now().Before(deadline) { + if expectedNodes > 0 && responsesReceived >= expectedNodes { + slog.Info("DescribeInstanceStatus: Received responses from all expected nodes", "expected", expectedNodes, "received", responsesReceived) + break + } + remaining := time.Until(deadline) + if remaining <= 0 { + break + } + msg, err := sub.NextMsg(remaining) + if err != nil { + if err == nats.ErrTimeout { + break + } + slog.Error("DescribeInstanceStatus: Error receiving message", "err", err) + break + } + responsesReceived++ + + if responseError, perr := utils.ValidateErrorPayload(msg.Data); perr != nil { + code := "" + if responseError.Code != nil { + code = *responseError.Code + } + if clientError == "" && code != "" { + if info, known := awserrors.ErrorLookup[code]; known && info.HTTPCode >= 400 && info.HTTPCode < 500 { + clientError = code + } + } + slog.Warn("DescribeInstanceStatus: Received error from node", "code", code, "responses_received", responsesReceived) + continue + } + + var nodeOutput ec2.DescribeInstanceStatusOutput + if err := json.Unmarshal(msg.Data, &nodeOutput); err != nil { + slog.Error("DescribeInstanceStatus: Failed to unmarshal node response", "err", err) + continue + } + if len(nodeOutput.InstanceStatuses) > 0 { + allStatuses = append(allStatuses, nodeOutput.InstanceStatuses...) + slog.Info("DescribeInstanceStatus: Collected statuses from node", "count", len(nodeOutput.InstanceStatuses), "responses_received", responsesReceived) + } + } + + if aws.BoolValue(input.IncludeAllInstances) { + if stopped := queryStoppedInstancesForStatus(natsConn, input, accountID, az); len(stopped) > 0 { + allStatuses = append(allStatuses, stopped...) + } + } + + finalStatuses := dedupStatuses(allStatuses) + + if clientError != "" && len(finalStatuses) == 0 { + return nil, errors.New(clientError) + } + + slog.Info("DescribeInstanceStatus: Aggregated response", "total_statuses", len(finalStatuses)) + return &ec2.DescribeInstanceStatusOutput{InstanceStatuses: finalStatuses}, nil +} + +// queryStoppedInstancesForStatus projects the input to DescribeInstancesInput +// before forwarding to the stopped-instance KV handler. The handler unmarshals +// with DisallowUnknownFields, so IncludeAllInstances must be dropped. Filters +// only valid for DescribeInstanceStatus (availability-zone, instance-state-code) +// are stripped because the stopped-instance handler would reject them via its +// own valid-filter map; these are applied gateway-side after transform. +func queryStoppedInstancesForStatus(natsConn *nats.Conn, input *ec2.DescribeInstanceStatusInput, accountID, az string) []*ec2.InstanceStatus { + projected := &ec2.DescribeInstancesInput{ + InstanceIds: input.InstanceIds, + Filters: stoppedCompatibleFilters(input.Filters), + } + jsonData, err := json.Marshal(projected) + if err != nil { + slog.Error("DescribeInstanceStatus: failed to marshal stopped query", "err", err) + return nil + } + reservations := queryInstanceBucket(natsConn, "ec2.DescribeStoppedInstances", jsonData, accountID) + var statuses []*ec2.InstanceStatus + for _, res := range reservations { + if res == nil { + continue + } + for _, inst := range res.Instances { + if inst == nil || inst.InstanceId == nil { + continue + } + statuses = append(statuses, buildInstanceStatusFromInstance(inst, az)) + } + } + return filterStoppedStatuses(statuses, input.Filters, az) +} + +// stoppedCompatibleFilters strips filters the stopped-instance handler would +// reject (its valid-filter map differs from DescribeInstanceStatus's). +func stoppedCompatibleFilters(filters []*ec2.Filter) []*ec2.Filter { + if len(filters) == 0 { + return nil + } + out := make([]*ec2.Filter, 0, len(filters)) + for _, f := range filters { + if f == nil || f.Name == nil { + continue + } + switch *f.Name { + case "availability-zone", "instance-state-code": + continue + } + out = append(out, f) + } + return out +} + +// filterStoppedStatuses applies DescribeInstanceStatus-specific filters that +// could not be forwarded to the stopped handler. +func filterStoppedStatuses(in []*ec2.InstanceStatus, filters []*ec2.Filter, az string) []*ec2.InstanceStatus { + if len(in) == 0 || len(filters) == 0 { + return in + } + out := in[:0] + for _, s := range in { + keep := true + for _, f := range filters { + if f == nil || f.Name == nil { + continue + } + switch *f.Name { + case "availability-zone": + if !filterValueMatches(f.Values, az) { + keep = false + } + case "instance-state-code": + code := "" + if s.InstanceState != nil && s.InstanceState.Code != nil { + code = strconv.FormatInt(*s.InstanceState.Code, 10) + } + if !filterValueMatches(f.Values, code) { + keep = false + } + } + if !keep { + break + } + } + if keep { + out = append(out, s) + } + } + return out +} + +func filterValueMatches(values []*string, field string) bool { + for _, v := range values { + if v != nil && *v == field { + return true + } + } + return false +} + +func buildInstanceStatusFromInstance(inst *ec2.Instance, az string) *ec2.InstanceStatus { + state := &ec2.InstanceState{Name: aws.String(ec2.InstanceStateNameStopped)} + if inst.State != nil { + if inst.State.Name != nil { + state.Name = inst.State.Name + } + state.Code = inst.State.Code + } + + return &ec2.InstanceStatus{ + AvailabilityZone: aws.String(az), + InstanceId: inst.InstanceId, + InstanceState: state, + InstanceStatus: &ec2.InstanceStatusSummary{ + Status: aws.String(notApplicable), + Details: []*ec2.InstanceStatusDetails{{ + Name: aws.String(reachabilityDetailName), + Status: aws.String(notApplicable), + }}, + }, + SystemStatus: &ec2.InstanceStatusSummary{ + Status: aws.String(notApplicable), + Details: []*ec2.InstanceStatusDetails{{ + Name: aws.String(reachabilityDetailName), + Status: aws.String(notApplicable), + }}, + }, + } +} + +// First-writer-wins by InstanceId, so a running fan-out entry naturally beats a +// stale stopped-KV entry during stop/start race windows. +func dedupStatuses(in []*ec2.InstanceStatus) []*ec2.InstanceStatus { + if len(in) == 0 { + return nil + } + seen := make(map[string]bool, len(in)) + out := make([]*ec2.InstanceStatus, 0, len(in)) + for _, s := range in { + if s == nil || s.InstanceId == nil { + continue + } + id := *s.InstanceId + if seen[id] { + continue + } + seen[id] = true + out = append(out, s) + } + return out +} diff --git a/spinifex/gateway/ec2/instance/DescribeInstanceStatus_test.go b/spinifex/gateway/ec2/instance/DescribeInstanceStatus_test.go new file mode 100644 index 000000000..fc22058e9 --- /dev/null +++ b/spinifex/gateway/ec2/instance/DescribeInstanceStatus_test.go @@ -0,0 +1,468 @@ +package gateway_ec2_instance + +import ( + "bytes" + "encoding/json" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func runningStatus(id, az string) *ec2.InstanceStatus { + return &ec2.InstanceStatus{ + AvailabilityZone: aws.String(az), + InstanceId: aws.String(id), + InstanceState: &ec2.InstanceState{Code: aws.Int64(16), Name: aws.String("running")}, + InstanceStatus: &ec2.InstanceStatusSummary{ + Status: aws.String("ok"), + Details: []*ec2.InstanceStatusDetails{{ + Name: aws.String("reachability"), Status: aws.String("passed"), + }}, + }, + SystemStatus: &ec2.InstanceStatusSummary{ + Status: aws.String("ok"), + Details: []*ec2.InstanceStatusDetails{{ + Name: aws.String("reachability"), Status: aws.String("passed"), + }}, + }, + } +} + +func respondJSON(t *testing.T, msg *nats.Msg, payload any) { + t.Helper() + data, err := json.Marshal(payload) + require.NoError(t, err) + require.NoError(t, msg.Respond(data)) +} + +func TestDescribeInstanceStatus_SingleNode(t *testing.T) { + _, nc := startTestNATSServer(t) + + _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{ + InstanceStatuses: []*ec2.InstanceStatus{ + runningStatus("i-001", "us-east-1a"), + runningStatus("i-002", "us-east-1a"), + }, + }) + }) + require.NoError(t, err) + + out, err := DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, nc, 1, "123456789012", "us-east-1a") + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 2) +} + +func TestDescribeInstanceStatus_TwoNodesDedup(t *testing.T) { + _, nc := startTestNATSServer(t) + + // First node returns i-001 (running) + _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{ + InstanceStatuses: []*ec2.InstanceStatus{runningStatus("i-001", "us-east-1a")}, + }) + }) + require.NoError(t, err) + + // Second node also reports i-001 — should dedup + nc2, err := nats.Connect(nc.ConnectedUrl()) + require.NoError(t, err) + defer nc2.Close() + + _, err = nc2.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{ + InstanceStatuses: []*ec2.InstanceStatus{runningStatus("i-001", "us-east-1a")}, + }) + }) + require.NoError(t, err) + + require.NoError(t, nc.Flush()) + require.NoError(t, nc2.Flush()) + + out, err := DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, nc, 2, "123456789012", "us-east-1a") + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + assert.Equal(t, "i-001", *out.InstanceStatuses[0].InstanceId) +} + +func TestDescribeInstanceStatus_OneNodeErrorOthersData(t *testing.T) { + _, nc := startTestNATSServer(t) + + // Node 1: data + _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{ + InstanceStatuses: []*ec2.InstanceStatus{runningStatus("i-good", "az-a")}, + }) + }) + require.NoError(t, err) + + // Node 2: error + nc2, err := nats.Connect(nc.ConnectedUrl()) + require.NoError(t, err) + defer nc2.Close() + _, err = nc2.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + require.NoError(t, msg.Respond(utils.GenerateErrorPayload("InvalidParameterValue"))) + }) + require.NoError(t, err) + + require.NoError(t, nc.Flush()) + require.NoError(t, nc2.Flush()) + + out, err := DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, nc, 2, "123456789012", "az-a") + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + assert.Equal(t, "i-good", *out.InstanceStatuses[0].InstanceId) +} + +func TestDescribeInstanceStatus_AllNodesError(t *testing.T) { + _, nc := startTestNATSServer(t) + + _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + require.NoError(t, msg.Respond(utils.GenerateErrorPayload("InvalidParameterValue"))) + }) + require.NoError(t, err) + + _, err = DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, nc, 1, "123456789012", "az-a") + require.Error(t, err) + assert.Equal(t, "InvalidParameterValue", err.Error()) +} + +func TestDescribeInstanceStatus_AllNodesTimeoutReturnsEmpty(t *testing.T) { + _, nc := startTestNATSServer(t) + + out, err := DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, nc, 0, "123456789012", "az-a") + require.NoError(t, err) + require.NotNil(t, out) + assert.Empty(t, out.InstanceStatuses) +} + +func TestDescribeInstanceStatus_IncludeAllAddsStoppedAsNotApplicable(t *testing.T) { + _, nc := startTestNATSServer(t) + + // Fan-out responder: a running instance + _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{ + InstanceStatuses: []*ec2.InstanceStatus{runningStatus("i-running", "az-a")}, + }) + }) + require.NoError(t, err) + + // Stopped KV responder + _, err = nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstancesOutput{ + Reservations: []*ec2.Reservation{{ + ReservationId: aws.String("r-stop"), + Instances: []*ec2.Instance{{ + InstanceId: aws.String("i-stopped"), + State: &ec2.InstanceState{Code: aws.Int64(80), Name: aws.String("stopped")}, + }}, + }}, + }) + }) + require.NoError(t, err) + + input := &ec2.DescribeInstanceStatusInput{IncludeAllInstances: aws.Bool(true)} + out, err := DescribeInstanceStatus(input, nc, 1, "123456789012", "az-a") + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 2) + + statusByID := make(map[string]*ec2.InstanceStatus) + for _, s := range out.InstanceStatuses { + statusByID[*s.InstanceId] = s + } + + running := statusByID["i-running"] + require.NotNil(t, running) + assert.Equal(t, "ok", *running.InstanceStatus.Status) + assert.Equal(t, "passed", *running.InstanceStatus.Details[0].Status) + + stopped := statusByID["i-stopped"] + require.NotNil(t, stopped) + assert.Equal(t, "stopped", *stopped.InstanceState.Name) + assert.Equal(t, "not-applicable", *stopped.InstanceStatus.Status) + assert.Equal(t, "not-applicable", *stopped.SystemStatus.Status) + assert.Equal(t, "not-applicable", *stopped.InstanceStatus.Details[0].Status) + assert.Equal(t, "az-a", *stopped.AvailabilityZone) +} + +func TestDescribeInstanceStatus_RunningWinsOverStoppedDuringRace(t *testing.T) { + _, nc := startTestNATSServer(t) + + // Fan-out: i-001 is running + _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{ + InstanceStatuses: []*ec2.InstanceStatus{runningStatus("i-001", "az-a")}, + }) + }) + require.NoError(t, err) + + // Stopped KV also has i-001 (race: stop transition hasn't fully cleared the KV) + _, err = nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstancesOutput{ + Reservations: []*ec2.Reservation{{ + ReservationId: aws.String("r-1"), + Instances: []*ec2.Instance{{ + InstanceId: aws.String("i-001"), + State: &ec2.InstanceState{Code: aws.Int64(80), Name: aws.String("stopped")}, + }}, + }}, + }) + }) + require.NoError(t, err) + + input := &ec2.DescribeInstanceStatusInput{IncludeAllInstances: aws.Bool(true)} + out, err := DescribeInstanceStatus(input, nc, 1, "123456789012", "az-a") + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + assert.Equal(t, "running", *out.InstanceStatuses[0].InstanceState.Name) + assert.Equal(t, "ok", *out.InstanceStatuses[0].InstanceStatus.Status) +} + +func TestDescribeInstanceStatus_NilInput(t *testing.T) { + _, nc := startTestNATSServer(t) + + _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{}) + }) + require.NoError(t, err) + + out, err := DescribeInstanceStatus(nil, nc, 1, "123456789012", "az-a") + require.NoError(t, err) + require.NotNil(t, out) + assert.Empty(t, out.InstanceStatuses) +} + +func TestDescribeInstanceStatus_ClosedConnection(t *testing.T) { + _, nc := startTestNATSServer(t) + + closed, err := nats.Connect(nc.ConnectedUrl()) + require.NoError(t, err) + closed.Close() + + _, err = DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, closed, 1, "123456789012", "az-a") + require.Error(t, err) +} + +func TestBuildInstanceStatusFromInstance_PopulatesState(t *testing.T) { + in := &ec2.Instance{ + InstanceId: aws.String("i-x"), + State: &ec2.InstanceState{Code: aws.Int64(80), Name: aws.String("stopped")}, + } + got := buildInstanceStatusFromInstance(in, "az-a") + require.NotNil(t, got) + assert.Equal(t, "i-x", *got.InstanceId) + assert.Equal(t, int64(80), *got.InstanceState.Code) + assert.Equal(t, "stopped", *got.InstanceState.Name) + assert.Equal(t, "az-a", *got.AvailabilityZone) + assert.Equal(t, "not-applicable", *got.InstanceStatus.Status) + assert.Equal(t, "not-applicable", *got.SystemStatus.Status) +} + +func TestBuildInstanceStatusFromInstance_FallbackStateName(t *testing.T) { + in := &ec2.Instance{InstanceId: aws.String("i-x")} + got := buildInstanceStatusFromInstance(in, "az-a") + assert.Equal(t, "stopped", *got.InstanceState.Name) +} + +// Regression: the stopped-instance handler unmarshals with DisallowUnknownFields, +// so forwarding the raw DescribeInstanceStatusInput JSON failed with ValidationError +// because of the IncludeAllInstances field. The gateway must project to +// DescribeInstancesInput before publishing. +func TestDescribeInstanceStatus_IncludeAllProjectsInputForStoppedHandler(t *testing.T) { + _, nc := startTestNATSServer(t) + + _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{}) + }) + require.NoError(t, err) + + var capturedReq ec2.DescribeInstancesInput + var unmarshalErr error + _, err = nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { + dec := json.NewDecoder(bytes.NewReader(msg.Data)) + dec.DisallowUnknownFields() + unmarshalErr = dec.Decode(&capturedReq) + respondJSON(t, msg, &ec2.DescribeInstancesOutput{ + Reservations: []*ec2.Reservation{{ + Instances: []*ec2.Instance{{ + InstanceId: aws.String("i-stop"), + State: &ec2.InstanceState{Code: aws.Int64(80), Name: aws.String("stopped")}, + }}, + }}, + }) + }) + require.NoError(t, err) + + input := &ec2.DescribeInstanceStatusInput{ + IncludeAllInstances: aws.Bool(true), + InstanceIds: []*string{aws.String("i-stop")}, + } + out, err := DescribeInstanceStatus(input, nc, 1, "123456789012", "az-a") + require.NoError(t, err) + require.NoError(t, unmarshalErr, "projected input must decode cleanly under DisallowUnknownFields") + require.Len(t, out.InstanceStatuses, 1) + require.NotNil(t, capturedReq.InstanceIds) + require.Len(t, capturedReq.InstanceIds, 1) + assert.Equal(t, "i-stop", *capturedReq.InstanceIds[0]) +} + +func TestStoppedCompatibleFilters_StripsStatusOnlyNames(t *testing.T) { + filters := []*ec2.Filter{ + {Name: aws.String("availability-zone"), Values: []*string{aws.String("az-a")}}, + {Name: aws.String("instance-state-code"), Values: []*string{aws.String("80")}}, + {Name: aws.String("instance-state-name"), Values: []*string{aws.String("stopped")}}, + {Name: aws.String("tag:Name"), Values: []*string{aws.String("foo")}}, + } + got := stoppedCompatibleFilters(filters) + require.Len(t, got, 2) + names := []string{*got[0].Name, *got[1].Name} + assert.ElementsMatch(t, []string{"instance-state-name", "tag:Name"}, names) +} + +func TestStoppedCompatibleFilters_Empty(t *testing.T) { + assert.Nil(t, stoppedCompatibleFilters(nil)) + assert.Nil(t, stoppedCompatibleFilters([]*ec2.Filter{})) +} + +func TestFilterStoppedStatuses_AvailabilityZoneMatch(t *testing.T) { + in := []*ec2.InstanceStatus{ + {InstanceId: aws.String("i-1"), AvailabilityZone: aws.String("az-a"), + InstanceState: &ec2.InstanceState{Code: aws.Int64(80)}}, + } + filters := []*ec2.Filter{ + {Name: aws.String("availability-zone"), Values: []*string{aws.String("az-a")}}, + } + got := filterStoppedStatuses(in, filters, "az-a") + require.Len(t, got, 1) +} + +func TestFilterStoppedStatuses_AvailabilityZoneMiss(t *testing.T) { + in := []*ec2.InstanceStatus{ + {InstanceId: aws.String("i-1"), AvailabilityZone: aws.String("az-a"), + InstanceState: &ec2.InstanceState{Code: aws.Int64(80)}}, + } + filters := []*ec2.Filter{ + {Name: aws.String("availability-zone"), Values: []*string{aws.String("az-b")}}, + } + got := filterStoppedStatuses(in, filters, "az-a") + assert.Empty(t, got) +} + +func TestFilterStoppedStatuses_StateCodeMatch(t *testing.T) { + in := []*ec2.InstanceStatus{ + {InstanceId: aws.String("i-1"), AvailabilityZone: aws.String("az-a"), + InstanceState: &ec2.InstanceState{Code: aws.Int64(80)}}, + } + filters := []*ec2.Filter{ + {Name: aws.String("instance-state-code"), Values: []*string{aws.String("80")}}, + } + got := filterStoppedStatuses(in, filters, "az-a") + require.Len(t, got, 1) +} + +func TestFilterStoppedStatuses_StateCodeMiss(t *testing.T) { + in := []*ec2.InstanceStatus{ + {InstanceId: aws.String("i-1"), AvailabilityZone: aws.String("az-a"), + InstanceState: &ec2.InstanceState{Code: aws.Int64(80)}}, + } + filters := []*ec2.Filter{ + {Name: aws.String("instance-state-code"), Values: []*string{aws.String("16")}}, + } + got := filterStoppedStatuses(in, filters, "az-a") + assert.Empty(t, got) +} + +func TestFilterStoppedStatuses_NoFiltersPassthrough(t *testing.T) { + in := []*ec2.InstanceStatus{{InstanceId: aws.String("i-1")}} + got := filterStoppedStatuses(in, nil, "az-a") + assert.Equal(t, in, got) +} + +func TestFilterStoppedStatuses_UnknownFilterIgnored(t *testing.T) { + in := []*ec2.InstanceStatus{ + {InstanceId: aws.String("i-1"), AvailabilityZone: aws.String("az-a"), + InstanceState: &ec2.InstanceState{Code: aws.Int64(80)}}, + } + // Filter that the stopped-bucket already applies (instance-state-name) + // should not also be enforced gateway-side — pass-through. + filters := []*ec2.Filter{ + {Name: aws.String("instance-state-name"), Values: []*string{aws.String("running")}}, + } + got := filterStoppedStatuses(in, filters, "az-a") + require.Len(t, got, 1) +} + +func TestDescribeInstanceStatus_IncludeAllWithAZFilterMatches(t *testing.T) { + _, nc := startTestNATSServer(t) + + _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{}) + }) + require.NoError(t, err) + + _, err = nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstancesOutput{ + Reservations: []*ec2.Reservation{{Instances: []*ec2.Instance{{ + InstanceId: aws.String("i-stop"), + State: &ec2.InstanceState{Code: aws.Int64(80), Name: aws.String("stopped")}, + }}}}, + }) + }) + require.NoError(t, err) + + input := &ec2.DescribeInstanceStatusInput{ + IncludeAllInstances: aws.Bool(true), + Filters: []*ec2.Filter{ + {Name: aws.String("availability-zone"), Values: []*string{aws.String("az-a")}}, + }, + } + out, err := DescribeInstanceStatus(input, nc, 1, "123456789012", "az-a") + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + assert.Equal(t, "i-stop", *out.InstanceStatuses[0].InstanceId) +} + +func TestDescribeInstanceStatus_IncludeAllWithAZFilterMisses(t *testing.T) { + _, nc := startTestNATSServer(t) + + _, err := nc.Subscribe("ec2.DescribeInstanceStatus", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstanceStatusOutput{}) + }) + require.NoError(t, err) + + _, err = nc.QueueSubscribe("ec2.DescribeStoppedInstances", "spinifex-workers", func(msg *nats.Msg) { + respondJSON(t, msg, &ec2.DescribeInstancesOutput{ + Reservations: []*ec2.Reservation{{Instances: []*ec2.Instance{{ + InstanceId: aws.String("i-stop"), + State: &ec2.InstanceState{Code: aws.Int64(80), Name: aws.String("stopped")}, + }}}}, + }) + }) + require.NoError(t, err) + + input := &ec2.DescribeInstanceStatusInput{ + IncludeAllInstances: aws.Bool(true), + Filters: []*ec2.Filter{ + {Name: aws.String("availability-zone"), Values: []*string{aws.String("az-other")}}, + }, + } + out, err := DescribeInstanceStatus(input, nc, 1, "123456789012", "az-a") + require.NoError(t, err) + assert.Empty(t, out.InstanceStatuses) +} + +func TestDedupStatuses_FirstWins(t *testing.T) { + first := runningStatus("i-001", "az-a") + second := &ec2.InstanceStatus{ + InstanceId: aws.String("i-001"), + InstanceState: &ec2.InstanceState{Code: aws.Int64(80), Name: aws.String("stopped")}, + } + got := dedupStatuses([]*ec2.InstanceStatus{first, second}) + require.Len(t, got, 1) + assert.Equal(t, "running", *got[0].InstanceState.Name) +} diff --git a/spinifex/gateway/gateway_test.go b/spinifex/gateway/gateway_test.go index 2223fd6a6..f0375baf1 100644 --- a/spinifex/gateway/gateway_test.go +++ b/spinifex/gateway/gateway_test.go @@ -981,7 +981,7 @@ func TestSlogRequestLogger_CapturesStatusCode(t *testing.T) { func TestEC2ActionMapCompleteness(t *testing.T) { expectedActions := []string{ "DescribeInstances", "RunInstances", "StartInstances", "StopInstances", - "TerminateInstances", "RebootInstances", "DescribeInstanceTypes", "GetConsoleOutput", + "TerminateInstances", "RebootInstances", "DescribeInstanceTypes", "DescribeInstanceStatus", "GetConsoleOutput", "ModifyInstanceAttribute", "DescribeInstanceAttribute", "CreateKeyPair", "DeleteKeyPair", "DescribeKeyPairs", "ImportKeyPair", "DescribeImages", "CreateImage", "DeregisterImage", "RegisterImage", "CopyImage", diff --git a/spinifex/handlers/ec2/instance/service.go b/spinifex/handlers/ec2/instance/service.go index 5b856e20f..a1f70a47f 100644 --- a/spinifex/handlers/ec2/instance/service.go +++ b/spinifex/handlers/ec2/instance/service.go @@ -10,6 +10,7 @@ import ( type InstanceService interface { RunInstances(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) DescribeInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) + DescribeInstanceStatus(input *ec2.DescribeInstanceStatusInput, accountID string) (*ec2.DescribeInstanceStatusOutput, 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) diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index 9f8c6f08d..dbf34e98e 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -10,6 +10,7 @@ import ( "log/slog" "os" "path/filepath" + "strconv" "strings" "text/template" "time" @@ -1343,6 +1344,16 @@ var DescribeInstancesValidFilters = map[string]bool{ "tag-value": true, } +// DescribeInstanceStatusValidFilters lists the filter names accepted by +// DescribeInstanceStatus. event.*, instance-status.* and system-status.* +// are rejected: Mulga's static-payload health model has no events and a +// single value per status field, so those filters can't usefully narrow. +var DescribeInstanceStatusValidFilters = map[string]bool{ + "availability-zone": true, + "instance-state-code": true, + "instance-state-name": 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 { @@ -2047,3 +2058,149 @@ func (s *InstanceServiceImpl) DescribeInstanceAttribute(input *ec2.DescribeInsta "instanceId", instanceID, "attribute", attribute, "accountID", accountID) return output, nil } + +// Terminated and error states are deliberately never surfaced: terminated +// matches AWS (drops from DescribeInstanceStatus shortly after termination); +// error is a Spinifex-internal state whose name is not a valid AWS state label. +var ( + describeInstanceStatusRunningOnly = map[vm.InstanceState]bool{vm.StateRunning: true} + describeInstanceStatusAllIncluded = map[vm.InstanceState]bool{ + vm.StateRunning: true, + vm.StatePending: true, + vm.StateProvisioning: true, + vm.StateStopping: true, + vm.StateStopped: true, + vm.StateShuttingDown: true, + } +) + +func (s *InstanceServiceImpl) buildInstanceStatus(v *vm.VM) *ec2.InstanceStatus { + state := &ec2.InstanceState{} + if info, ok := vm.EC2StateCodes[v.Status]; ok { + state.SetCode(info.Code) + state.SetName(info.Name) + } else { + state.SetCode(vm.EC2StateCodes[vm.StatePending].Code) + state.SetName(vm.EC2StateCodes[vm.StatePending].Name) + } + + status := instanceStatusNotApplicable + reachability := instanceStatusNotApplicable + if v.Status == vm.StateRunning { + status = instanceStatusOK + reachability = instanceStatusPassed + } + + return &ec2.InstanceStatus{ + AvailabilityZone: aws.String(s.config.AZ), + InstanceId: aws.String(v.ID), + InstanceState: state, + InstanceStatus: &ec2.InstanceStatusSummary{ + Status: aws.String(status), + Details: []*ec2.InstanceStatusDetails{{ + Name: aws.String(reachabilityDetailName), + Status: aws.String(reachability), + }}, + }, + SystemStatus: &ec2.InstanceStatusSummary{ + Status: aws.String(status), + Details: []*ec2.InstanceStatusDetails{{ + Name: aws.String(reachabilityDetailName), + Status: aws.String(reachability), + }}, + }, + } +} + +const ( + instanceStatusOK = "ok" + instanceStatusPassed = "passed" + instanceStatusNotApplicable = "not-applicable" + reachabilityDetailName = "reachability" +) + +func instanceStatusMatchesFilters(v *vm.VM, is *ec2.InstanceStatus, filters map[string][]string) bool { + for name, values := range filters { + if strings.HasPrefix(name, "tag:") { + continue + } + var field string + switch name { + case "availability-zone": + if is.AvailabilityZone != nil { + field = *is.AvailabilityZone + } + case "instance-state-name": + if is.InstanceState != nil && is.InstanceState.Name != nil { + field = *is.InstanceState.Name + } + case "instance-state-code": + if is.InstanceState != nil && is.InstanceState.Code != nil { + field = strconv.FormatInt(*is.InstanceState.Code, 10) + } + default: + return false + } + if !filterutil.MatchesAny(values, field) { + return false + } + } + + if v.Instance != nil { + tags := filterutil.EC2TagsToMap(v.Instance.Tags) + return filterutil.MatchesTags(filters, tags) + } + return filterutil.MatchesTags(filters, nil) +} + +// DescribeInstanceStatus returns per-VM status entries for VMs on this node +// visible to the caller. Stopped instances come from the gateway's KV query, +// not this handler. +func (s *InstanceServiceImpl) DescribeInstanceStatus(input *ec2.DescribeInstanceStatusInput, accountID string) (*ec2.DescribeInstanceStatusOutput, error) { + slog.Info("Processing DescribeInstanceStatus request from this node", "accountID", accountID) + + instanceIDFilter := make(map[string]bool) + for _, id := range input.InstanceIds { + if id == nil || *id == "" { + continue + } + if !strings.HasPrefix(*id, "i-") { + return nil, errors.New(awserrors.ErrorInvalidInstanceIDMalformed) + } + instanceIDFilter[*id] = true + } + + parsedFilters, err := filterutil.ParseFilters(input.Filters, DescribeInstanceStatusValidFilters) + if err != nil { + slog.Warn("DescribeInstanceStatus: invalid filter", "err", err) + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + includedStates := describeInstanceStatusRunningOnly + if aws.BoolValue(input.IncludeAllInstances) { + includedStates = describeInstanceStatusAllIncluded + } + + var statuses []*ec2.InstanceStatus + s.vmMgr.View(func(vms map[string]*vm.VM) { + for _, v := range vms { + if !IsInstanceVisible(accountID, v.AccountID) { + continue + } + if len(instanceIDFilter) > 0 && !instanceIDFilter[v.ID] { + continue + } + if !includedStates[v.Status] { + continue + } + is := s.buildInstanceStatus(v) + if len(parsedFilters) > 0 && !instanceStatusMatchesFilters(v, is, parsedFilters) { + continue + } + statuses = append(statuses, is) + } + }) + + slog.Info("DescribeInstanceStatus completed", "count", len(statuses)) + return &ec2.DescribeInstanceStatusOutput{InstanceStatuses: statuses}, nil +} diff --git a/spinifex/handlers/ec2/instance/service_impl_test.go b/spinifex/handlers/ec2/instance/service_impl_test.go index f84d0dfaa..807f3521b 100644 --- a/spinifex/handlers/ec2/instance/service_impl_test.go +++ b/spinifex/handlers/ec2/instance/service_impl_test.go @@ -2291,3 +2291,231 @@ func TestSetters(t *testing.T) { svc.SetRunInstancesDeps(&fakeAMILoader{}, &fakeKeyValidator{}, nil, nil) require.NotNil(t, svc) } + +// --- DescribeInstanceStatus --- + +func instanceStatusService(t *testing.T, az string, vms map[string]*vm.VM) *InstanceServiceImpl { + t.Helper() + return &InstanceServiceImpl{ + config: &config.Config{AZ: az}, + vmMgr: mgrWith(vms), + } +} + +func runningVM(id, owner string) *vm.VM { + return &vm.VM{ + ID: id, + Status: vm.StateRunning, + AccountID: owner, + Reservation: &ec2.Reservation{ + ReservationId: aws.String("r-" + id), + OwnerId: aws.String(owner), + }, + Instance: &ec2.Instance{InstanceId: aws.String(id)}, + } +} + +func TestDescribeInstanceStatus_Empty(t *testing.T) { + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{}) + out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, utils.GlobalAccountID) + require.NoError(t, err) + require.NotNil(t, out) + assert.Empty(t, out.InstanceStatuses) +} + +func TestDescribeInstanceStatus_RunningInstance(t *testing.T) { + owner := "111122223333" + v := runningVM("i-aaa", owner) + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{v.ID: v}) + + out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, owner) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + + s := out.InstanceStatuses[0] + assert.Equal(t, "i-aaa", *s.InstanceId) + assert.Equal(t, "az-a", *s.AvailabilityZone) + assert.Equal(t, "running", *s.InstanceState.Name) + assert.Equal(t, int64(16), *s.InstanceState.Code) + assert.Equal(t, "ok", *s.InstanceStatus.Status) + assert.Equal(t, "ok", *s.SystemStatus.Status) + require.Len(t, s.InstanceStatus.Details, 1) + assert.Equal(t, "passed", *s.InstanceStatus.Details[0].Status) +} + +func TestDescribeInstanceStatus_AccountFilteringHidesOtherTenant(t *testing.T) { + v := runningVM("i-other", "999988887777") + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{v.ID: v}) + + out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, "111122223333") + require.NoError(t, err) + assert.Empty(t, out.InstanceStatuses) +} + +func TestDescribeInstanceStatus_StoppedExcludedByDefault(t *testing.T) { + owner := "111122223333" + stopped := runningVM("i-stop", owner) + stopped.Status = vm.StateStopped + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{stopped.ID: stopped}) + + out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{}, owner) + require.NoError(t, err) + assert.Empty(t, out.InstanceStatuses) +} + +func TestDescribeInstanceStatus_IncludeAllSurfacesPending(t *testing.T) { + owner := "111122223333" + pend := runningVM("i-pend", owner) + pend.Status = vm.StatePending + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{pend.ID: pend}) + + out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{ + IncludeAllInstances: aws.Bool(true), + }, owner) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + assert.Equal(t, "pending", *out.InstanceStatuses[0].InstanceState.Name) + assert.Equal(t, "not-applicable", *out.InstanceStatuses[0].InstanceStatus.Status) + assert.Equal(t, "not-applicable", *out.InstanceStatuses[0].SystemStatus.Status) +} + +func TestDescribeInstanceStatus_IncludeAllSurfacesStopped(t *testing.T) { + owner := "111122223333" + stopped := runningVM("i-stop", owner) + stopped.Status = vm.StateStopped + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{stopped.ID: stopped}) + + out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{ + IncludeAllInstances: aws.Bool(true), + }, owner) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + assert.Equal(t, "stopped", *out.InstanceStatuses[0].InstanceState.Name) + assert.Equal(t, "not-applicable", *out.InstanceStatuses[0].InstanceStatus.Status) + assert.Equal(t, "not-applicable", *out.InstanceStatuses[0].SystemStatus.Status) +} + +func TestDescribeInstanceStatus_TerminatedNeverReturned(t *testing.T) { + owner := "111122223333" + term := runningVM("i-term", owner) + term.Status = vm.StateTerminated + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{term.ID: term}) + + out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{ + IncludeAllInstances: aws.Bool(true), + }, owner) + require.NoError(t, err) + assert.Empty(t, out.InstanceStatuses) +} + +func TestDescribeInstanceStatus_ErrorStateNeverReturned(t *testing.T) { + owner := "111122223333" + errVM := runningVM("i-err", owner) + errVM.Status = vm.StateError + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{errVM.ID: errVM}) + + out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{ + IncludeAllInstances: aws.Bool(true), + }, owner) + require.NoError(t, err) + assert.Empty(t, out.InstanceStatuses) +} + +func TestDescribeInstanceStatus_InstanceIDFilter(t *testing.T) { + owner := "111122223333" + keep := runningVM("i-keep", owner) + drop := runningVM("i-drop", owner) + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{keep.ID: keep, drop.ID: drop}) + + out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{ + InstanceIds: []*string{aws.String("i-keep")}, + }, owner) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + assert.Equal(t, "i-keep", *out.InstanceStatuses[0].InstanceId) +} + +func TestDescribeInstanceStatus_MalformedInstanceID(t *testing.T) { + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{}) + _, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{ + InstanceIds: []*string{aws.String("not-an-id")}, + }, utils.GlobalAccountID) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDMalformed, err.Error()) +} + +func TestDescribeInstanceStatus_UnknownFilter(t *testing.T) { + owner := "111122223333" + v := runningVM("i-aaa", owner) + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{v.ID: v}) + + _, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{ + Filters: []*ec2.Filter{{ + Name: aws.String("event.code"), + Values: []*string{aws.String("system-reboot")}, + }}, + }, owner) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error()) +} + +func TestDescribeInstanceStatus_StateNameFilter(t *testing.T) { + owner := "111122223333" + run := runningVM("i-run", owner) + stop := runningVM("i-stop", owner) + stop.Status = vm.StateStopped + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{run.ID: run, stop.ID: stop}) + + out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{ + IncludeAllInstances: aws.Bool(true), + Filters: []*ec2.Filter{{ + Name: aws.String("instance-state-name"), + Values: []*string{aws.String("running")}, + }}, + }, owner) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + assert.Equal(t, "i-run", *out.InstanceStatuses[0].InstanceId) +} + +func TestDescribeInstanceStatus_AvailabilityZoneFilter(t *testing.T) { + owner := "111122223333" + v := runningVM("i-aaa", owner) + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{v.ID: v}) + + out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{ + Filters: []*ec2.Filter{{ + Name: aws.String("availability-zone"), + Values: []*string{aws.String("az-a")}, + }}, + }, owner) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + + out, err = svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{ + Filters: []*ec2.Filter{{ + Name: aws.String("availability-zone"), + Values: []*string{aws.String("az-b")}, + }}, + }, owner) + require.NoError(t, err) + assert.Empty(t, out.InstanceStatuses) +} + +func TestDescribeInstanceStatus_TagFilter(t *testing.T) { + owner := "111122223333" + tagged := runningVM("i-tag", owner) + tagged.Instance.Tags = []*ec2.Tag{{Key: aws.String("Name"), Value: aws.String("foo")}} + plain := runningVM("i-plain", owner) + svc := instanceStatusService(t, "az-a", map[string]*vm.VM{tagged.ID: tagged, plain.ID: plain}) + + out, err := svc.DescribeInstanceStatus(&ec2.DescribeInstanceStatusInput{ + Filters: []*ec2.Filter{{ + Name: aws.String("tag:Name"), + Values: []*string{aws.String("foo")}, + }}, + }, owner) + require.NoError(t, err) + require.Len(t, out.InstanceStatuses, 1) + assert.Equal(t, "i-tag", *out.InstanceStatuses[0].InstanceId) +} diff --git a/spinifex/handlers/ec2/instance/service_mock.go b/spinifex/handlers/ec2/instance/service_mock.go index 9b50abc8f..7e002f6a5 100644 --- a/spinifex/handlers/ec2/instance/service_mock.go +++ b/spinifex/handlers/ec2/instance/service_mock.go @@ -44,6 +44,10 @@ func (s *MockInstanceService) DescribeInstanceTypes(_ *ec2.DescribeInstanceTypes return &ec2.DescribeInstanceTypesOutput{}, nil } +func (s *MockInstanceService) DescribeInstanceStatus(_ *ec2.DescribeInstanceStatusInput, _ string) (*ec2.DescribeInstanceStatusOutput, error) { + return &ec2.DescribeInstanceStatusOutput{}, nil +} + func (s *MockInstanceService) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, _ string) (*ec2.DescribeInstanceAttributeOutput, error) { return &ec2.DescribeInstanceAttributeOutput{InstanceId: input.InstanceId}, nil } diff --git a/spinifex/handlers/ec2/instance/service_nats.go b/spinifex/handlers/ec2/instance/service_nats.go index 9792aadbe..67048141d 100644 --- a/spinifex/handlers/ec2/instance/service_nats.go +++ b/spinifex/handlers/ec2/instance/service_nats.go @@ -49,6 +49,12 @@ func (s *NATSInstanceService) DescribeInstanceTypes(_ *ec2.DescribeInstanceTypes return nil, fmt.Errorf("DescribeInstanceTypes is fan-out; use gateway scatter-gather, not NATSInstanceService") } +// DescribeInstanceStatus is a fan-out operation; use the gateway's scatter-gather +// helper instead of this single-node client. +func (s *NATSInstanceService) DescribeInstanceStatus(_ *ec2.DescribeInstanceStatusInput, _ string) (*ec2.DescribeInstanceStatusOutput, error) { + return nil, fmt.Errorf("DescribeInstanceStatus 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) }