feat: describe instance status (#221)

This commit is contained in:
Julian Sommer
2026-05-13 15:55:20 +10:00
committed by GitHub
parent a31fd8098c
commit 97701737cb
16 changed files with 1213 additions and 7 deletions
+1 -1
View File
@@ -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
+1 -1
View File
@@ -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'
+3 -3
View File
@@ -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
+1
View File
@@ -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<br>2. Empty log file returns empty output<br>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<br>2. Get userData from stopped instance<br>3. Get default disableApiTermination<br>4. Invalid attribute name (error)<br>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<br>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)<br>2. Filter by instance ID<br>3. Default excludes stopped instances<br>4. IncludeAllInstances=true includes stopped (not-applicable)<br>5. Cross-account instance excluded<br>6. Unknown filter returns InvalidParameterValue<br>7. Malformed instance ID rejected<br>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<br>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<br>2. Verify monitoring state reverts in describe-instances | **NOT STARTED** |
+2 -1
View File
@@ -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"},
@@ -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)
+56
View File
@@ -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) {
+3
View File
@@ -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)
}),
@@ -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
}
@@ -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)
}
+1 -1
View File
@@ -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",
@@ -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)
@@ -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
}
@@ -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)
}
@@ -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
}
@@ -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)
}