diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index 7db50c801..699b97393 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -164,6 +164,13 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) string { return outcomeError } + // A multi-node spread carries the reservation ID the gateway minted once + // for the whole call. reservation is the same pointer every instance + // holds, so this override reaches all of them. + if gatewayReservationID := utils.ReservationIDFromMsg(msg); gatewayReservationID != "" { + reservation.SetReservationId(gatewayReservationID) + } + // PlacementGroupNode is daemon-local identity, set after prepare. for _, instance := range instances { if instance.PlacementGroupName != "" { diff --git a/spinifex/gateway/ec2/instance/DescribeInstances.go b/spinifex/gateway/ec2/instance/DescribeInstances.go index 8106fa6c1..e14098e72 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstances.go +++ b/spinifex/gateway/ec2/instance/DescribeInstances.go @@ -6,6 +6,8 @@ import ( "errors" "fmt" "log/slog" + "sort" + "strings" "sync" "time" @@ -115,25 +117,54 @@ func DescribeInstancesChecked(ctx context.Context, input *ec2.DescribeInstancesI return nil, errors.New(firstClient4xx) } - if len(input.InstanceIds) > 0 && complete { - found := make(map[string]bool) - for _, res := range reservations { - for _, inst := range res.Instances { - if inst != nil && inst.InstanceId != nil { - found[*inst.InstanceId] = true - } - } - } - for _, id := range input.InstanceIds { - if id != nil && !found[*id] { + if len(input.InstanceIds) > 0 { + missing := missingInstanceIDs(reservations, input.InstanceIds) + if len(missing) > 0 { + if complete { return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound) } + // An incomplete sweep cannot prove absence, so a missing id is + // unresolved, not gone. NotFound would be a false negative and an + // empty success would look identical to a real one, so return 503. + return nil, errors.Join( + awserrors.Errorf(awserrors.ErrorServiceUnavailable, + "cannot confirm status of instance(s) %s: describe fan-out did not complete", strings.Join(missing, ", ")), + awserrors.RetryAfter(awserrors.ErrorServiceUnavailable, describeIncompleteRetryAfter), + ) } } return &ec2.DescribeInstancesOutput{Reservations: reservations}, nil } +// describeIncompleteRetryAfter is the suggested backoff surfaced to a caller +// whose named-ID describe hit an incomplete fan-out. Set near the fan-out's +// own deadline, so an immediate retry has a real chance of landing complete. +const describeIncompleteRetryAfter = 3 * time.Second + +// missingInstanceIDs returns the requested instance IDs absent from +// reservations, preserving the caller's requested order. +func missingInstanceIDs(reservations []*ec2.Reservation, instanceIDs []*string) []string { + found := make(map[string]bool) + for _, res := range reservations { + if res == nil { + continue + } + for _, inst := range res.Instances { + if inst != nil && inst.InstanceId != nil { + found[*inst.InstanceId] = true + } + } + } + var missing []string + for _, id := range instanceIDs { + if id != nil && !found[*id] { + missing = append(missing, *id) + } + } + return missing +} + // DescribeInstancesForReconcile is the strict variant, for a caller that may act // only on a provably whole view. complete is true only when the sweep observed // every expected node and both instance buckets; a partial sweep — a node down, @@ -242,11 +273,61 @@ func gatherInstances(ctx context.Context, input *ec2.DescribeInstancesInput, nat kvWg.Wait() allReservations = append(allReservations, bucketReservations...) + allReservations = mergeReservationsByID(allReservations) slog.InfoContext(ctx, "DescribeInstances: Aggregated response", "total_reservations", len(allReservations)) return allReservations, fanoutComplete && bucketsOK, sum.FirstClient4xx, nil } +// mergeReservationsByID merges frames sharing a ReservationId into one, drops +// empty results, and sorts by id so repeated calls answer identically. +func mergeReservationsByID(reservations []*ec2.Reservation) []*ec2.Reservation { + order := make([]string, 0, len(reservations)) + merged := make(map[string]*ec2.Reservation, len(reservations)) + for _, res := range reservations { + if res == nil { + continue + } + id := "" + if res.ReservationId != nil { + id = *res.ReservationId + } + existing, ok := merged[id] + if !ok { + // The slices this function appends to or replaces are copied, so + // merging never writes through to a caller's frame. Instance and + // group elements are shared by pointer and must stay read-only. + copyRes := *res + copyRes.Instances = append([]*ec2.Instance(nil), res.Instances...) + copyRes.Groups = append([]*ec2.GroupIdentifier(nil), res.Groups...) + merged[id] = ©Res + order = append(order, id) + continue + } + existing.Instances = append(existing.Instances, res.Instances...) + if existing.OwnerId == nil { + existing.OwnerId = res.OwnerId + } + if existing.RequesterId == nil { + existing.RequesterId = res.RequesterId + } + if len(existing.Groups) == 0 { + existing.Groups = append([]*ec2.GroupIdentifier(nil), res.Groups...) + } + } + + sort.Strings(order) + result := make([]*ec2.Reservation, 0, len(order)) + for _, id := range order { + res := merged[id] + if len(res.Instances) == 0 { + continue + } + result = append(result, res) + } + return result +} + // instanceSettler tracks which of the requested instance ids have turned up, // across both the running-instance fan-out and the stopped and terminated // bucket queries that run alongside it. Once none are outstanding there is diff --git a/spinifex/gateway/ec2/instance/DescribeInstances_test.go b/spinifex/gateway/ec2/instance/DescribeInstances_test.go index 3a1f1dec7..c1da22351 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstances_test.go +++ b/spinifex/gateway/ec2/instance/DescribeInstances_test.go @@ -101,6 +101,138 @@ func TestDescribeInstances_MultipleNodes(t *testing.T) { assert.Len(t, output.Reservations, 2) } +// TestDescribeInstances_MergesFramesBySameReservationID covers the second +// reservation-grouping defect: frames from different nodes sharing a +// ReservationId must merge into ONE Reservation, not two. +func TestDescribeInstances_MergesFramesBySameReservationID(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + + nc.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) { + data, _ := json.Marshal(&ec2.DescribeInstancesOutput{ + Reservations: []*ec2.Reservation{{ + ReservationId: aws.String("r-shared"), + Instances: []*ec2.Instance{{InstanceId: aws.String("i-a")}}, + }}, + }) + msg.Respond(data) + }) + + nc2, err := nats.Connect(nc.ConnectedUrl()) + require.NoError(t, err) + defer nc2.Close() + + nc2.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) { + data, _ := json.Marshal(&ec2.DescribeInstancesOutput{ + Reservations: []*ec2.Reservation{{ + ReservationId: aws.String("r-shared"), + Instances: []*ec2.Instance{{InstanceId: aws.String("i-b")}}, + }}, + }) + msg.Respond(data) + }) + + nc.Flush() + nc2.Flush() + + input := &ec2.DescribeInstancesInput{} + output, err := DescribeInstances(context.Background(), input, nc, 2, "123456789012") + + require.NoError(t, err) + require.NotNil(t, output) + require.Len(t, output.Reservations, 1, "two frames sharing a ReservationId must merge into one Reservation") + var gotIDs []string + for _, inst := range output.Reservations[0].Instances { + gotIDs = append(gotIDs, aws.StringValue(inst.InstanceId)) + } + assert.ElementsMatch(t, []string{"i-a", "i-b"}, gotIDs) +} + +// TestDescribeInstances_MergedReservationOrderIsDeterministic guards against +// the nondeterminism the merge fixes: repeated calls with the same input +// must answer with reservations in the same order every time. +func TestDescribeInstances_MergedReservationOrderIsDeterministic(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + + nc.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) { + data, _ := json.Marshal(&ec2.DescribeInstancesOutput{ + Reservations: []*ec2.Reservation{ + {ReservationId: aws.String("r-charlie"), Instances: []*ec2.Instance{{InstanceId: aws.String("i-c")}}}, + {ReservationId: aws.String("r-alpha"), Instances: []*ec2.Instance{{InstanceId: aws.String("i-a")}}}, + {ReservationId: aws.String("r-bravo"), Instances: []*ec2.Instance{{InstanceId: aws.String("i-b")}}}, + }, + }) + msg.Respond(data) + }) + + input := &ec2.DescribeInstancesInput{} + var orders [][]string + for range 5 { + output, err := DescribeInstances(context.Background(), input, nc, 1, "123456789012") + require.NoError(t, err) + var order []string + for _, res := range output.Reservations { + order = append(order, aws.StringValue(res.ReservationId)) + } + orders = append(orders, order) + } + + for i := 1; i < len(orders); i++ { + assert.Equal(t, orders[0], orders[i], "repeated calls with the same input must return reservations in the same order") + } + assert.Equal(t, []string{"r-alpha", "r-bravo", "r-charlie"}, orders[0]) +} + +// TestDescribeInstances_MergePreservesReservationLevelFields ensures merging +// two frames of the same reservation does not drop OwnerId, RequesterId, or +// Groups just because only one of the two frames carried them. +func TestDescribeInstances_MergePreservesReservationLevelFields(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + + nc.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) { + data, _ := json.Marshal(&ec2.DescribeInstancesOutput{ + Reservations: []*ec2.Reservation{{ + ReservationId: aws.String("r-shared"), + OwnerId: aws.String("111111111111"), + Instances: []*ec2.Instance{{InstanceId: aws.String("i-a")}}, + }}, + }) + msg.Respond(data) + }) + + nc2, err := nats.Connect(nc.ConnectedUrl()) + require.NoError(t, err) + defer nc2.Close() + + nc2.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) { + data, _ := json.Marshal(&ec2.DescribeInstancesOutput{ + Reservations: []*ec2.Reservation{{ + ReservationId: aws.String("r-shared"), + RequesterId: aws.String("requester-x"), + Groups: []*ec2.GroupIdentifier{{GroupName: aws.String("sg-1")}}, + Instances: []*ec2.Instance{{InstanceId: aws.String("i-b")}}, + }}, + }) + msg.Respond(data) + }) + + nc.Flush() + nc2.Flush() + + input := &ec2.DescribeInstancesInput{} + output, err := DescribeInstances(context.Background(), input, nc, 2, "123456789012") + + require.NoError(t, err) + require.Len(t, output.Reservations, 1) + res := output.Reservations[0] + assert.Equal(t, "111111111111", aws.StringValue(res.OwnerId)) + assert.Equal(t, "requester-x", aws.StringValue(res.RequesterId)) + require.Len(t, res.Groups, 1) + assert.Equal(t, "sg-1", aws.StringValue(res.Groups[0].GroupName)) +} + func TestDescribeInstances_NoSubscribers(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) @@ -217,7 +349,7 @@ func TestDescribeInstances_TimeoutCollection(t *testing.T) { time.Sleep(50 * time.Millisecond) data, _ := json.Marshal(&ec2.DescribeInstancesOutput{ Reservations: []*ec2.Reservation{ - {ReservationId: aws.String("r-delayed")}, + {ReservationId: aws.String("r-delayed"), Instances: []*ec2.Instance{{InstanceId: aws.String("i-delayed")}}}, }, }) msg.Respond(data) @@ -240,7 +372,7 @@ func TestDescribeInstances_EarlyExitWithExpectedNodes(t *testing.T) { nc.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) { data, _ := json.Marshal(&ec2.DescribeInstancesOutput{ Reservations: []*ec2.Reservation{ - {ReservationId: aws.String("r-fast")}, + {ReservationId: aws.String("r-fast"), Instances: []*ec2.Instance{{InstanceId: aws.String("i-fast")}}}, }, }) msg.Respond(data) @@ -386,12 +518,25 @@ func TestDescribeInstancesChecked_FilterOnlyNoMatch_ReturnsEmptyNotNotFound(t *t assert.Empty(t, output.Reservations) } -// TestDescribeInstancesChecked_ExplicitIDNotFound_NodeTimeout verifies the -// subtle third criterion: a node timing out during the fan-out must not -// produce a false NotFound. Only one of two expected nodes answers, so the -// sweep is provably incomplete and the naive found-set check must be -// suppressed. -func TestDescribeInstancesChecked_ExplicitIDNotFound_NodeTimeout(t *testing.T) { +// assertIncompleteSweepUnavailable asserts err is the retryable +// ServiceUnavailable returned when an incomplete fan-out leaves wantIDs +// unresolved: never a false NotFound, but never silent either. +func assertIncompleteSweepUnavailable(t *testing.T, err error, wantIDs ...string) { + t.Helper() + require.Error(t, err, "an incomplete sweep must never assert a false NotFound, but it must not stay silent either") + code, message, ok := awserrors.ResolveErrorDetail(err) + require.True(t, ok, "error must carry a registered awserrors code") + assert.Equal(t, awserrors.ErrorServiceUnavailable, code) + assert.False(t, awserrors.IsTerminal(err), "an incomplete sweep is retryable, not a terminal client fault") + for _, id := range wantIDs { + assert.Contains(t, message, id) + } +} + +// TestDescribeInstancesChecked_ExplicitID_NodeTimeout_ReturnsRetryableUnavailable +// covers a node timeout: only one of two expected nodes answers, so the +// sweep is incomplete and the named id must come back retryable, not NotFound. +func TestDescribeInstancesChecked_ExplicitID_NodeTimeout_ReturnsRetryableUnavailable(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) subscribeEmptyInstanceBuckets(t, nc) @@ -409,17 +554,15 @@ func TestDescribeInstancesChecked_ExplicitIDNotFound_NodeTimeout(t *testing.T) { output, err := DescribeInstancesChecked(context.Background(), input, nc, 2, nil, "123456789012", WithFanoutTimeout(300*time.Millisecond)) - require.NoError(t, err, "an incomplete sweep must never assert a false NotFound") - require.NotNil(t, output) - assert.Empty(t, output.Reservations) + assert.Nil(t, output) + assertIncompleteSweepUnavailable(t, err, "i-on-the-slow-node") } // --- Identity-mode completeness (nodeIDs set) --- -// A missing configured node suppresses NotFound even though the requested ID -// never showed up in what did arrive: the sweep is provably incomplete, so -// silence is the only safe answer. -func TestDescribeInstancesChecked_Identity_MissingNodeSuppressesNotFound(t *testing.T) { +// A missing configured node leaves the requested ID's absence unproven, so +// a retryable 503 replaces the old false NotFound / silent empty success. +func TestDescribeInstancesChecked_Identity_MissingNode_ReturnsRetryableUnavailable(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) subscribeEmptyInstanceBuckets(t, nc) @@ -437,11 +580,8 @@ func TestDescribeInstancesChecked_Identity_MissingNodeSuppressesNotFound(t *test output, err := DescribeInstancesChecked(context.Background(), input, nc, 0, []string{"node-1", "node-2"}, "123456789012", WithFanoutTimeout(300*time.Millisecond)) - // The sweep stays lenient on a partial view — node-1's reservations are - // still returned — it is only the NotFound assertion that a missing node - // must suppress. - require.NoError(t, err, "a node missing from the configured set must never assert a false NotFound") - require.NotNil(t, output) + assert.Nil(t, output) + assertIncompleteSweepUnavailable(t, err, "i-doesnotexist0000000") } // The positive counterpart: every configured node answers, so the sweep is @@ -469,10 +609,9 @@ func TestDescribeInstancesChecked_Identity_AllNodesAnswer_NotFoundAsserted(t *te assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) } -// A node that answers with a decodable error envelope (a 5xx from that node) -// never becomes a valid responder, so it counts the same as a missing node — -// the sweep stays incomplete and NotFound is suppressed. -func TestDescribeInstancesChecked_Identity_ErroringNodeSuppressesNotFound(t *testing.T) { +// A node answering with a decodable error envelope never becomes a valid +// responder, so it counts as missing: the caller gets a retryable 503. +func TestDescribeInstancesChecked_Identity_ErroringNode_ReturnsRetryableUnavailable(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) subscribeEmptyInstanceBuckets(t, nc) @@ -486,16 +625,13 @@ func TestDescribeInstancesChecked_Identity_ErroringNodeSuppressesNotFound(t *tes output, err := DescribeInstancesChecked(context.Background(), input, nc, 0, []string{"node-1", "node-2"}, "123456789012", WithFanoutTimeout(300*time.Millisecond)) - require.NoError(t, err, "an erroring node must suppress NotFound, not be treated as an empty-handed valid responder") - require.NotNil(t, output) - assert.Empty(t, output.Reservations) + assert.Nil(t, output) + assertIncompleteSweepUnavailable(t, err, "i-doesnotexist0000000") } -// A payload that is neither a valid error envelope nor a decodable -// DescribeInstancesOutput (a malformed reply) leaves its node out of -// ValidResponders, so the sweep is incomplete and NotFound is suppressed — -// the same as a node that never answered at all. -func TestDescribeInstancesChecked_Identity_UndecodablePayloadSuppressesNotFound(t *testing.T) { +// An undecodable payload leaves its node out of ValidResponders, the same +// as a node that never answered, so the caller gets a retryable 503. +func TestDescribeInstancesChecked_Identity_UndecodablePayload_ReturnsRetryableUnavailable(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) subscribeEmptyInstanceBuckets(t, nc) @@ -508,9 +644,8 @@ func TestDescribeInstancesChecked_Identity_UndecodablePayloadSuppressesNotFound( output, err := DescribeInstancesChecked(context.Background(), input, nc, 0, []string{"node-1", "node-2"}, "123456789012", WithFanoutTimeout(300*time.Millisecond)) - require.NoError(t, err) - require.NotNil(t, output) - assert.Empty(t, output.Reservations) + assert.Nil(t, output) + assertIncompleteSweepUnavailable(t, err, "i-doesnotexist0000000") } // A node with nil Reservations still decoded successfully, so it is a valid @@ -533,12 +668,10 @@ func TestDescribeInstancesChecked_Identity_NilReservationsIsValidResponder(t *te assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) } -// An empty or unset nodeIDs falls back to the legacy expectedNodes path. -// DescribeInstancesChecked's only production caller always passes -// expectedNodes=0 alongside its node set, so an empty node set (e.g. an -// unconfigured cluster) can never satisfy expectedNodes>0 and NotFound stays -// suppressed — the fallback fails closed rather than trusting a stale count. -func TestDescribeInstancesChecked_Identity_EmptyNodeIDsFallsBackAndStaysSuppressed(t *testing.T) { +// An empty nodeIDs falls back to the legacy expectedNodes path, which the +// only production caller always leaves at 0 — permanently incomplete — so +// a named ID here must come back retryable, never a false NotFound. +func TestDescribeInstancesChecked_Identity_EmptyNodeIDsFallsBack_ReturnsRetryableUnavailable(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) subscribeEmptyInstanceBuckets(t, nc) @@ -550,9 +683,8 @@ func TestDescribeInstancesChecked_Identity_EmptyNodeIDsFallsBackAndStaysSuppress output, err := DescribeInstancesChecked(context.Background(), input, nc, 0, nil, "123456789012", WithFanoutTimeout(300*time.Millisecond)) - require.NoError(t, err) - require.NotNil(t, output) - assert.Empty(t, output.Reservations) + assert.Nil(t, output) + assertIncompleteSweepUnavailable(t, err, "i-doesnotexist0000000") } // mustMarshalDescribeOutput is a t.Helper wrapper so the identity-mode tests diff --git a/spinifex/gateway/ec2/instance/placement.go b/spinifex/gateway/ec2/instance/placement.go index 152bff221..e769ce8c6 100644 --- a/spinifex/gateway/ec2/instance/placement.go +++ b/spinifex/gateway/ec2/instance/placement.go @@ -55,7 +55,10 @@ func distributeInstances(ctx context.Context, input *ec2.RunInstancesInput, nats launchCount := min(maxCount, totalCapacity) allocations := spreadAllocate(nodes, launchCount) - results := launchOnNodes(ctx, allocations, input, natsConn, accountID) + // Minted once here and handed to every node, so a launch that spreads + // across nodes comes back as one reservation instead of one per node. + reservationID := utils.GenerateResourceID("r") + results := launchOnNodes(ctx, allocations, input, natsConn, accountID, reservationID) return aggregateResults(ctx, results, minCount, natsConn, accountID) } @@ -157,8 +160,10 @@ type nodeLaunchResult struct { Err error } -// launchOnNodes sends targeted RunInstances to each node in parallel with MinCount=MaxCount=assigned. -func launchOnNodes(ctx context.Context, allocations []nodeAllocation, input *ec2.RunInstancesInput, natsConn *nats.Conn, accountID string) []nodeLaunchResult { +// launchOnNodes sends targeted RunInstances to each node in parallel with +// MinCount=MaxCount=assigned. A non-empty reservationID is forwarded as a +// header so every node launches into the same reservation. +func launchOnNodes(ctx context.Context, allocations []nodeAllocation, input *ec2.RunInstancesInput, natsConn *nats.Conn, accountID, reservationID string) []nodeLaunchResult { instanceType := aws.StringValue(input.InstanceType) results := make([]nodeLaunchResult, len(allocations)) @@ -174,7 +179,11 @@ func launchOnNodes(ctx context.Context, allocations []nodeAllocation, input *ec2 nodeInput.MaxCount = aws.Int64(int64(a.Assigned)) topic := fmt.Sprintf("ec2.RunInstances.%s.%s", instanceType, a.NodeID) - reservation, err := utils.NATSRequest[ec2.Reservation](ctx, natsConn, topic, &nodeInput, 5*time.Minute, accountID) + var headers []utils.NATSHeader + if reservationID != "" { + headers = append(headers, utils.NATSHeader{Key: utils.ReservationIDHeader, Value: reservationID}) + } + reservation, err := utils.NATSRequest[ec2.Reservation](ctx, natsConn, topic, &nodeInput, 5*time.Minute, accountID, headers...) if err != nil { results[idx] = nodeLaunchResult{NodeID: a.NodeID, Err: fmt.Errorf("launch on %s: %w", a.NodeID, err)} return @@ -269,10 +278,14 @@ func distributeInstancesSpread(ctx context.Context, input *ec2.RunInstancesInput allocations[i] = nodeAllocation{NodeID: nodeID, Assigned: 1} } - results := launchOnNodes(ctx, allocations, input, natsConn, accountID) + // A spread group puts one instance on each node, so without a shared ID + // every node mints its own and one launch comes back as several + // reservations. + spreadReservationID := utils.GenerateResourceID("r") + results := launchOnNodes(ctx, allocations, input, natsConn, accountID, spreadReservationID) var allInstances []*ec2.Instance - var reservationID *string + reservationID := aws.String(spreadReservationID) nodeInstances := make(map[string][]string) var failedNodes []string @@ -289,9 +302,6 @@ func distributeInstancesSpread(ctx context.Context, input *ec2.RunInstancesInput nodeInstances[r.NodeID] = append(nodeInstances[r.NodeID], *inst.InstanceId) } } - if reservationID == nil { - reservationID = r.Reservation.ReservationId - } } } @@ -397,7 +407,7 @@ func distributeInstancesCluster(ctx context.Context, input *ec2.RunInstancesInpu NodeID: targetNode, Assigned: launchCount, }} - results := launchOnNodes(ctx, allocations, input, natsConn, accountID) + results := launchOnNodes(ctx, allocations, input, natsConn, accountID, "") if results[0].Err != nil { if clientErr := extractClientError(results); clientErr != nil { diff --git a/spinifex/gateway/ec2/instance/placement_test.go b/spinifex/gateway/ec2/instance/placement_test.go index cfa945248..847d505c3 100644 --- a/spinifex/gateway/ec2/instance/placement_test.go +++ b/spinifex/gateway/ec2/instance/placement_test.go @@ -500,6 +500,65 @@ func TestDistributeInstances_SuccessfulSpread(t *testing.T) { assert.True(t, ids["i-n2"], "should have instance from node-2") } +// TestDistributeInstances_MultiNodeLaunchSharesOneReservationID checks that +// distributeInstances mints one reservation ID and forwards it to every node +// via the X-Reservation-Id header, so a spread launch shares one reservation. +func TestDistributeInstances_MultiNodeLaunchSharesOneReservationID(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + + statusSub, err := nc.Subscribe("spinifex.node.status", func(msg *nats.Msg) { + for _, resp := range []types.NodeStatusResponse{ + {Node: "node-1", InstanceTypes: []types.InstanceTypeCap{{Name: "t3.micro", Available: 2}}}, + {Node: "node-2", InstanceTypes: []types.InstanceTypeCap{{Name: "t3.micro", Available: 2}}}, + } { + data, _ := json.Marshal(resp) + _ = nc.Publish(msg.Reply, data) + } + }) + require.NoError(t, err) + defer statusSub.Unsubscribe() + + receivedIDs := make(chan string, 2) + makeHandler := func(node string) nats.MsgHandler { + return func(msg *nats.Msg) { + resID := msg.Header.Get(utils.ReservationIDHeader) + receivedIDs <- resID + reservation := ec2.Reservation{ + ReservationId: aws.String(resID), + Instances: []*ec2.Instance{{InstanceId: aws.String(fmt.Sprintf("i-%s", node))}}, + } + data, _ := json.Marshal(reservation) + _ = msg.Respond(data) + } + } + sub1, err := nc.Subscribe("ec2.RunInstances.t3.micro.node-1", makeHandler("node-1")) + require.NoError(t, err) + defer sub1.Unsubscribe() + sub2, err := nc.Subscribe("ec2.RunInstances.t3.micro.node-2", makeHandler("node-2")) + require.NoError(t, err) + defer sub2.Unsubscribe() + + require.NoError(t, nc.Flush()) + + input := &ec2.RunInstancesInput{ + ImageId: aws.String("ami-test"), + InstanceType: aws.String("t3.micro"), + MinCount: aws.Int64(2), + MaxCount: aws.Int64(2), + } + + reservation, err := distributeInstances(context.Background(), input, nc, "test-account", 2) + require.NoError(t, err) + require.NotEmpty(t, aws.StringValue(reservation.ReservationId)) + + id1 := <-receivedIDs + id2 := <-receivedIDs + assert.NotEmpty(t, id1) + assert.Equal(t, id1, id2, "every node in a single RunInstances call must receive the same reservation id") + assert.Equal(t, id1, aws.StringValue(reservation.ReservationId)) +} + func TestDistributeInstances_InsufficientCapacity(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) diff --git a/spinifex/handlers/ec2/image/service_impl.go b/spinifex/handlers/ec2/image/service_impl.go index 30fc2c90e..3c275c8af 100644 --- a/spinifex/handlers/ec2/image/service_impl.go +++ b/spinifex/handlers/ec2/image/service_impl.go @@ -171,6 +171,14 @@ func (s *ImageServiceImpl) describeImages(ctx context.Context, input *ec2.Descri amiOwner := amiMeta.ImageOwnerAlias isSystemAMI := amiOwner != "" && !utils.IsAccountID(amiOwner) + // Resolved up front so the owner filter below compares against the same + // value the OwnerId field reports, rather than the raw alias string a + // system AMI carries instead of a numeric account ID. + ownerID := amiOwner + if isSystemAMI { + ownerID = utils.GlobalAccountID + } + if !callerCanReadAMI(amiMeta, accountID) { continue } @@ -183,13 +191,15 @@ func (s *ImageServiceImpl) describeImages(ctx context.Context, input *ec2.Descri } switch *owner { case "self": - if amiOwner == accountID { + if ownerID == accountID { + found = true + } + case "amazon", "system", "aws-marketplace": + if isSystemAMI { found = true } default: - if amiOwner == *owner { - found = true - } else if isSystemAMI && *owner == utils.GlobalAccountID { + if ownerID == *owner { found = true } } @@ -202,11 +212,6 @@ func (s *ImageServiceImpl) describeImages(ctx context.Context, input *ec2.Descri } } - ownerID := amiOwner - if isSystemAMI { - ownerID = utils.GlobalAccountID - } - image := &ec2.Image{ ImageId: aws.String(amiMeta.ImageID), Name: aws.String(amiMeta.Name), diff --git a/spinifex/handlers/ec2/image/service_impl_test.go b/spinifex/handlers/ec2/image/service_impl_test.go index 2643e6984..e57a2c728 100644 --- a/spinifex/handlers/ec2/image/service_impl_test.go +++ b/spinifex/handlers/ec2/image/service_impl_test.go @@ -1052,6 +1052,90 @@ func TestDescribeImages_FilterBy(t *testing.T) { } } +// TestDescribeImages_OwnerSelf_GlobalAccountIncludesSystemAMI is the direct +// bug fix: describeImages used to compare the raw alias string against the +// caller's account ID, so a system AMI could never match "self" even for the +// global account that conceptually owns it. +func TestDescribeImages_OwnerSelf_GlobalAccountIncludesSystemAMI(t *testing.T) { + svc, store := setupTestImageService(t) + createTestAMIConfigWithOwner(t, store, "ami-system", "system-ami", "amazon") + + out, err := svc.DescribeImages(context.Background(), &ec2.DescribeImagesInput{ + Owners: []*string{aws.String("self")}, + }, utils.GlobalAccountID) + require.NoError(t, err) + + gotIDs := make([]string, len(out.Images)) + for i, img := range out.Images { + gotIDs[i] = aws.StringValue(img.ImageId) + } + assert.ElementsMatch(t, []string{"ami-system"}, gotIDs) +} + +// TestDescribeImages_OwnerSelf_TenantExcludesSystemAMI pins the other half of +// the fix: an ordinary tenant's "self" must still exclude a system AMI, so the +// resolved-ownerID rewrite cannot widen self beyond the caller's own images. +func TestDescribeImages_OwnerSelf_TenantExcludesSystemAMI(t *testing.T) { + svc, store := setupTestImageService(t) + createTestAMIConfigWithOwner(t, store, "ami-selfowned", "self-owned-ami", testAccountID) + createTestAMIConfigWithOwner(t, store, "ami-system", "system-ami", "amazon") + + out, err := svc.DescribeImages(context.Background(), &ec2.DescribeImagesInput{ + Owners: []*string{aws.String("self")}, + }, testAccountID) + require.NoError(t, err) + + gotIDs := make([]string, len(out.Images)) + for i, img := range out.Images { + gotIDs[i] = aws.StringValue(img.ImageId) + } + assert.ElementsMatch(t, []string{"ami-selfowned"}, gotIDs) +} + +// TestDescribeImages_OwnerAlias_MatchesSystemAMIs covers the "amazon" and +// "system" alias values Terraform/CLI callers send for built-in images. +func TestDescribeImages_OwnerAlias_MatchesSystemAMIs(t *testing.T) { + svc, store := setupTestImageService(t) + createTestAMIConfigWithOwner(t, store, "ami-system", "system-ami", "amazon") + createTestAMIConfigWithOwner(t, store, "ami-tenant", "tenant-ami", testAccountID) + + for _, alias := range []string{"amazon", "system"} { + t.Run(alias, func(t *testing.T) { + out, err := svc.DescribeImages(context.Background(), &ec2.DescribeImagesInput{ + Owners: []*string{aws.String(alias)}, + }, testAccountID) + require.NoError(t, err) + + gotIDs := make([]string, len(out.Images)) + for i, img := range out.Images { + gotIDs[i] = aws.StringValue(img.ImageId) + } + assert.ElementsMatch(t, []string{"ami-system"}, gotIDs) + }) + } +} + +// TestDescribeImages_OwnerNumericID_UnaffectedByOrdering is the regression +// guard for the accountteardown image reaper: a numeric owner ID must keep +// matching exactly the caller's own images, not widen to include system AMIs, +// after hoisting the ownerID resolution above the filter. +func TestDescribeImages_OwnerNumericID_UnaffectedByOrdering(t *testing.T) { + svc, store := setupTestImageService(t) + createTestAMIConfigWithOwner(t, store, "ami-tenant", "tenant-ami", testAccountID) + createTestAMIConfigWithOwner(t, store, "ami-system", "system-ami", "amazon") + + out, err := svc.DescribeImages(context.Background(), &ec2.DescribeImagesInput{ + Owners: []*string{aws.String(testAccountID)}, + }, testAccountID) + require.NoError(t, err) + + gotIDs := make([]string, len(out.Images)) + for i, img := range out.Images { + gotIDs[i] = aws.StringValue(img.ImageId) + } + assert.ElementsMatch(t, []string{"ami-tenant"}, gotIDs) +} + func TestDescribeImages_FilterMultipleValues_OR(t *testing.T) { svc, store := setupTestImageService(t) createTestAMIConfigWithName(t, store, "ami-aaa", "debian-13") diff --git a/spinifex/handlers/ec2/instance/projection.go b/spinifex/handlers/ec2/instance/projection.go index be2b3964c..a12fbe611 100644 --- a/spinifex/handlers/ec2/instance/projection.go +++ b/spinifex/handlers/ec2/instance/projection.go @@ -45,6 +45,10 @@ func ProjectInstance(v *vm.VM, cfg InstanceProjection) (inst *ec2.Instance, stat instanceCopy := *v.Instance instanceCopy.State = &ec2.InstanceState{} + // Hoisted so the instance-level PublicDnsName and the primary NIC's + // Association.PublicDnsName (stamped below) cannot drift apart. + var publicDNS string + // Public IP and its derived DNS names exist only while the instance runs; a // stopped instance has released the public IP, so the KV path leaves them // unset. Mirrors the records the control-plane writer publishes to northstar. @@ -53,7 +57,8 @@ func ProjectInstance(v *vm.VM, cfg InstanceProjection) (inst *ec2.Instance, stat instanceCopy.PublicIpAddress = aws.String(v.PublicIP) } - publicDNS, privateDNS := handlers_dns.EC2DNSNames( + var privateDNS string + publicDNS, privateDNS = handlers_dns.EC2DNSNames( cfg.Region, cfg.DNSBaseDomain, cfg.DNSInternalDomain, aws.StringValue(instanceCopy.PublicIpAddress), aws.StringValue(instanceCopy.PrivateIpAddress), ) @@ -131,6 +136,11 @@ func ProjectInstance(v *vm.VM, cfg InstanceProjection) (inst *ec2.Instance, stat } instanceCopy.Monitoring = &ec2.Monitoring{State: aws.String(monitoringState)} + // AWS always reports EnclaveOptions, and Spinifex never enables it. Stamped + // unconditionally, including on stopped instances, since it is launch + // configuration rather than runtime network state. + instanceCopy.EnclaveOptions = &ec2.EnclaveOptions{Enabled: aws.Bool(false)} + // Spot lineage stamped by the post-launch write-back survives a stop, so // project it regardless of runtime state. Both empty for on-demand, so the // fields stay absent there. @@ -157,6 +167,19 @@ func ProjectInstance(v *vm.VM, cfg InstanceProjection) (inst *ec2.Instance, stat } nicCopy := *nic nicCopy.SourceDestCheck = aws.Bool(true) + + // associate_public_ip_address is read off the primary interface and is + // force-new, so leaving Association unset reports false for an instance + // that has a public IP and every plan proposes a destroy/recreate. + isPrimaryNIC := nicCopy.Attachment != nil && nicCopy.Attachment.DeviceIndex != nil && *nicCopy.Attachment.DeviceIndex == 0 + if cfg.IncludeRuntimeNetwork && isPrimaryNIC && v.PublicIP != "" { + nicCopy.Association = &ec2.InstanceNetworkInterfaceAssociation{ + PublicIp: aws.String(v.PublicIP), + PublicDnsName: aws.String(publicDNS), + IpOwnerId: aws.String(v.AccountID), + } + } + nics = append(nics, &nicCopy) } instanceCopy.NetworkInterfaces = nics diff --git a/spinifex/handlers/ec2/instance/projection_test.go b/spinifex/handlers/ec2/instance/projection_test.go index 9e65a939d..c0a00f162 100644 --- a/spinifex/handlers/ec2/instance/projection_test.go +++ b/spinifex/handlers/ec2/instance/projection_test.go @@ -21,6 +21,7 @@ func fullVM(status vm.InstanceState) *vm.VM { InstanceId: aws.String("i-abc123"), PrivateIpAddress: aws.String("10.0.0.5"), }, + AccountID: "123456789012", PublicIP: "203.0.113.7", PlacementGroupName: "pg-1", IamInstanceProfileArn: "arn:aws:iam::123456789012:instance-profile/role", @@ -170,6 +171,7 @@ func TestProjectInstance_DoesNotMutateSource(t *testing.T) { v := fullVM(vm.StateRunning) v.Instance.NetworkInterfaces = []*ec2.InstanceNetworkInterface{{ NetworkInterfaceId: aws.String("eni-1"), + Attachment: &ec2.InstanceNetworkInterfaceAttachment{DeviceIndex: aws.Int64(0)}, }} _, _ = ProjectInstance(v, runningCfg()) @@ -180,6 +182,8 @@ func TestProjectInstance_DoesNotMutateSource(t *testing.T) { assert.Nil(t, v.Instance.SourceDestCheck) assert.Nil(t, v.Instance.NetworkInterfaces[0].SourceDestCheck, "the interfaces are shared with the stored instance, so they must be copied before they are stamped") + assert.Nil(t, v.Instance.NetworkInterfaces[0].Association, + "the interfaces are shared with the stored instance, so the new Association must not be written back onto it") } // The check is always on and cannot be turned off, but the describe never said @@ -223,6 +227,86 @@ func TestProjectInstance_PathsAgreeOnRetainedFields(t *testing.T) { // Runtime-only fields are intentionally path-specific. assert.NotEqual(t, running.PublicIpAddress, stopped.PublicIpAddress) assert.NotEqual(t, running.CapacityReservationId, stopped.CapacityReservationId) + + // EnclaveOptions is launch configuration, not runtime network state, so it + // must be present and identical on both paths. + require.NotNil(t, running.EnclaveOptions) + require.NotNil(t, stopped.EnclaveOptions) + assert.Equal(t, running.EnclaveOptions, stopped.EnclaveOptions) + assert.False(t, aws.BoolValue(stopped.EnclaveOptions.Enabled)) +} + +// TestProjectInstance_PrimaryNICAssociation covers the force-new +// associate_public_ip_address regression: Terraform reads it off the primary +// interface, so an instance with a public IP must report an Association there. +func TestProjectInstance_PrimaryNICAssociation(t *testing.T) { + newNICs := func() []*ec2.InstanceNetworkInterface { + return []*ec2.InstanceNetworkInterface{ + { + NetworkInterfaceId: aws.String("eni-primary"), + Attachment: &ec2.InstanceNetworkInterfaceAttachment{DeviceIndex: aws.Int64(0)}, + }, + { + NetworkInterfaceId: aws.String("eni-secondary"), + Attachment: &ec2.InstanceNetworkInterfaceAttachment{DeviceIndex: aws.Int64(1)}, + }, + } + } + + t.Run("running with public IP stamps the primary NIC only", func(t *testing.T) { + v := fullVM(vm.StateRunning) + v.Instance.NetworkInterfaces = newNICs() + + got, _ := ProjectInstance(v, runningCfg()) + + require.Len(t, got.NetworkInterfaces, 2) + primary := got.NetworkInterfaces[0] + secondary := got.NetworkInterfaces[1] + + require.NotNil(t, primary.Association) + assert.Equal(t, "203.0.113.7", aws.StringValue(primary.Association.PublicIp)) + assert.Equal(t, aws.StringValue(got.PublicDnsName), aws.StringValue(primary.Association.PublicDnsName)) + assert.NotEmpty(t, aws.StringValue(primary.Association.PublicDnsName)) + assert.Equal(t, "123456789012", aws.StringValue(primary.Association.IpOwnerId)) + + assert.Nil(t, secondary.Association) + }) + + t.Run("IncludeRuntimeNetwork off leaves Association unset", func(t *testing.T) { + v := fullVM(vm.StateStopped) + v.Instance.NetworkInterfaces = newNICs() + + got, _ := ProjectInstance(v, stoppedCfg(80, "stopped")) + + require.Len(t, got.NetworkInterfaces, 2) + assert.Nil(t, got.NetworkInterfaces[0].Association) + assert.Nil(t, got.NetworkInterfaces[1].Association) + }) + + t.Run("no public IP leaves Association unset", func(t *testing.T) { + v := fullVM(vm.StateRunning) + v.PublicIP = "" + v.Instance.NetworkInterfaces = newNICs() + + got, _ := ProjectInstance(v, runningCfg()) + + require.Len(t, got.NetworkInterfaces, 2) + assert.Nil(t, got.NetworkInterfaces[0].Association) + }) +} + +// TestProjectInstance_EnclaveOptionsAlwaysDisabled covers the permanent +// enclave_options {} Terraform diff: AWS always reports the block, and +// Spinifex never enables it, on both the running and the stopped/KV paths. +func TestProjectInstance_EnclaveOptionsAlwaysDisabled(t *testing.T) { + running, _ := ProjectInstance(fullVM(vm.StateRunning), runningCfg()) + stopped, _ := ProjectInstance(fullVM(vm.StateStopped), stoppedCfg(80, "stopped")) + + require.NotNil(t, running.EnclaveOptions) + assert.False(t, aws.BoolValue(running.EnclaveOptions.Enabled)) + + require.NotNil(t, stopped.EnclaveOptions) + assert.False(t, aws.BoolValue(stopped.EnclaveOptions.Enabled)) } func TestParseInstanceIDFilter(t *testing.T) { diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index e5b21bce3..f21ad3635 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -1801,16 +1801,6 @@ func (s *InstanceServiceImpl) DescribeInstances(ctx context.Context, input *ec2. resID = *instance.Reservation.ReservationId } - if _, exists := reservationMap[resID]; !exists { - reservation := &ec2.Reservation{} - reservation.SetReservationId(resID) - if instance.Reservation.OwnerId != nil { - reservation.SetOwnerId(*instance.Reservation.OwnerId) - } - reservation.Instances = []*ec2.Instance{} - reservationMap[resID] = reservation - } - // Project every vm.VM-sourced field via the shared projection, the // same call the daemon's running path makes, so both define the field // set once. Running instances include their runtime network and @@ -1833,6 +1823,19 @@ func (s *InstanceServiceImpl) DescribeInstances(ctx context.Context, input *ec2. continue } + // Only create the reservation entry once an instance has survived + // filtering, so a reservation with every instance filtered out + // never appears in the response. + if _, exists := reservationMap[resID]; !exists { + reservation := &ec2.Reservation{} + reservation.SetReservationId(resID) + if instance.Reservation.OwnerId != nil { + reservation.SetOwnerId(*instance.Reservation.OwnerId) + } + reservation.Instances = []*ec2.Instance{} + reservationMap[resID] = reservation + } + reservationMap[resID].Instances = append(reservationMap[resID].Instances, projected) } }) @@ -1932,16 +1935,6 @@ func (s *InstanceServiceImpl) describeInstancesFromKV(ctx context.Context, input resID = *instance.Reservation.ReservationId } - if _, exists := reservationMap[resID]; !exists { - reservation := &ec2.Reservation{} - reservation.SetReservationId(resID) - if instance.Reservation.OwnerId != nil { - reservation.SetOwnerId(*instance.Reservation.OwnerId) - } - reservation.Instances = []*ec2.Instance{} - reservationMap[resID] = reservation - } - // Reuse the shared projection so stopped/terminated instances carry the // same fields as the running path. Runtime network and the capacity // reservation are released on stop, so IncludeRuntimeNetwork stays false; @@ -1957,6 +1950,19 @@ func (s *InstanceServiceImpl) describeInstancesFromKV(ctx context.Context, input continue } + // Only create the reservation entry once an instance has survived + // filtering, so a reservation with every instance filtered out + // never appears in the response. + if _, exists := reservationMap[resID]; !exists { + reservation := &ec2.Reservation{} + reservation.SetReservationId(resID) + if instance.Reservation.OwnerId != nil { + reservation.SetOwnerId(*instance.Reservation.OwnerId) + } + reservation.Instances = []*ec2.Instance{} + reservationMap[resID] = reservation + } + reservationMap[resID].Instances = append(reservationMap[resID].Instances, projected) } diff --git a/spinifex/handlers/ec2/instance/service_impl_test.go b/spinifex/handlers/ec2/instance/service_impl_test.go index a483ea130..f2482ed57 100644 --- a/spinifex/handlers/ec2/instance/service_impl_test.go +++ b/spinifex/handlers/ec2/instance/service_impl_test.go @@ -847,6 +847,46 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { assert.Len(t, out.Reservations[0].Instances, 2) } +// TestDescribeInstances_FilterExcludesOnlyInstance_ReturnsZeroReservations +// checks a reservation whose one instance is filtered out by --filters does +// not appear in the response at all, since AWS never returns an empty one. +func TestDescribeInstances_FilterExcludesOnlyInstance_ReturnsZeroReservations(t *testing.T) { + lonely := &vm.VM{ + ID: "i-lonely", + InstanceType: "t3.micro", + Reservation: &ec2.Reservation{ReservationId: aws.String("r-lonely")}, + Instance: &ec2.Instance{InstanceId: aws.String("i-lonely")}, + } + svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{lonely.ID: lonely}), config: &config.Config{}} + + input := &ec2.DescribeInstancesInput{ + Filters: []*ec2.Filter{{Name: aws.String("instance-type"), Values: []*string{aws.String("t3.nano")}}}, + } + out, err := svc.DescribeInstances(context.Background(), input, utils.GlobalAccountID) + require.NoError(t, err) + assert.Empty(t, out.Reservations, "a reservation with every instance filtered out must not appear in the response") +} + +// TestDescribeInstancesFromKV_FilterExcludesOnlyInstance_ReturnsZeroReservations +// is the KV-path (stopped/terminated) counterpart of the live-path test above. +func TestDescribeInstancesFromKV_FilterExcludesOnlyInstance_ReturnsZeroReservations(t *testing.T) { + lonely := &vm.VM{ + ID: "i-lonely-stopped", + InstanceType: "t3.micro", + Reservation: &ec2.Reservation{ReservationId: aws.String("r-lonely-stopped")}, + Instance: &ec2.Instance{InstanceId: aws.String("i-lonely-stopped")}, + } + svc := &InstanceServiceImpl{config: &config.Config{}} + listFn := func() ([]*vm.VM, error) { return []*vm.VM{lonely}, nil } + + input := &ec2.DescribeInstancesInput{ + Filters: []*ec2.Filter{{Name: aws.String("instance-type"), Values: []*string{aws.String("t3.nano")}}}, + } + out, err := svc.describeInstancesFromKV(context.Background(), input, utils.GlobalAccountID, listFn, 80, "stopped", "DescribeStoppedInstances") + require.NoError(t, err) + assert.Empty(t, out.Reservations, "a reservation with every instance filtered out must not appear in the response") +} + func TestDescribeInstanceAttribute_MissingInstanceID(t *testing.T) { svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{})} _, err := svc.DescribeInstanceAttribute(context.Background(), &ec2.DescribeInstanceAttributeInput{ diff --git a/spinifex/handlers/ec2/vpc/service_impl.go b/spinifex/handlers/ec2/vpc/service_impl.go index 4f602c12d..bf4e29720 100644 --- a/spinifex/handlers/ec2/vpc/service_impl.go +++ b/spinifex/handlers/ec2/vpc/service_impl.go @@ -542,12 +542,18 @@ func (s *VPCServiceImpl) rejectAttachedIGW(ctx context.Context, accountID, vpcID } // describeVpcsValidFilters defines the set of filter names accepted by DescribeVpcs. +// cidr-block and is-default are the spellings an internal caller +// (handlers/rds/network.go) is pinned to; cidr and isDefault are the +// spellings AWS itself sends, so both must be accepted side by side. var describeVpcsValidFilters = map[string]bool{ - "vpc-id": true, - "state": true, - "cidr-block": true, - "is-default": true, - "owner-id": true, + "vpc-id": true, + "state": true, + "cidr-block": true, + "cidr": true, + "is-default": true, + "isDefault": true, + "owner-id": true, + "cidr-block-association.cidr-block": true, } // SupportsDescribeVpcsFilter reports whether DescribeVpcs accepts a filter name. @@ -572,7 +578,7 @@ func (s *VPCServiceImpl) DescribeVpcs(ctx context.Context, input *ec2.DescribeVp parsedFilters, err := filterutil.ParseFilters(input.Filters, describeVpcsValidFilters) if err != nil { slog.WarnContext(ctx, "DescribeVpcs: invalid filter", "err", err) - return nil, errors.New(awserrors.ErrorInvalidParameterValue) + return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "%s", err) } prefix := accountID + "." @@ -1099,9 +1105,13 @@ func vpcMatchesFilters(record *VPCRecord, accountID string, filters map[string][ field = record.VpcId case "state": field = record.State - case "cidr-block": + case "cidr-block", "cidr": field = record.CidrBlock - case "is-default": + case "cidr-block-association.cidr-block": + // VPCRecord stores only the primary CIDR; the association set is + // synthesised per-request, so this can only ever match it. + field = record.CidrBlock + case "is-default", "isDefault": field = strconv.FormatBool(record.IsDefault) case "owner-id": field = accountID diff --git a/spinifex/handlers/ec2/vpc/service_impl_test.go b/spinifex/handlers/ec2/vpc/service_impl_test.go index 0b4340e9d..61e6f9e3c 100644 --- a/spinifex/handlers/ec2/vpc/service_impl_test.go +++ b/spinifex/handlers/ec2/vpc/service_impl_test.go @@ -8,6 +8,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/testutil" "github.com/nats-io/nats.go" @@ -1738,6 +1739,73 @@ func TestDescribeVpcs_FilterUnknownName_Error(t *testing.T) { }, }, testAccountID) require.Error(t, err) + + code, message, ok := awserrors.ResolveErrorDetail(err) + require.True(t, ok, "the AWS code must stay resolvable or the client gets a 500") + assert.Equal(t, awserrors.ErrorInvalidParameterValue, code) + assert.Contains(t, message, "bogus-filter", "the offending filter name must reach the client") +} + +// TestDescribeVpcs_FilterByCidrAlias pins the AWS spelling ("cidr") alongside +// the pre-existing "cidr-block", so terraform's aws_vpc data source (which +// sends "cidr") stops getting InvalidParameterValue. +func TestDescribeVpcs_FilterByCidrAlias(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + createTestVPC(t, svc, "10.0.0.0/16") + createTestVPC(t, svc, "172.16.0.0/16") + + out, err := svc.DescribeVpcs(context.Background(), &ec2.DescribeVpcsInput{ + Filters: []*ec2.Filter{ + {Name: aws.String("cidr"), Values: []*string{aws.String("10.0.0.0/16")}}, + }, + }, testAccountID) + require.NoError(t, err) + assert.Len(t, out.Vpcs, 1) + assert.Equal(t, "10.0.0.0/16", *out.Vpcs[0].CidrBlock) +} + +// TestDescribeVpcs_FilterByIsDefaultAlias pins the AWS spelling ("isDefault") +// alongside the pre-existing "is-default", which an internal RDS caller +// hardcodes and must keep working unchanged. +func TestDescribeVpcs_FilterByIsDefaultAlias(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + createTestVPC(t, svc, "10.0.0.0/16") + + out, err := svc.DescribeVpcs(context.Background(), &ec2.DescribeVpcsInput{ + Filters: []*ec2.Filter{ + {Name: aws.String("isDefault"), Values: []*string{aws.String("false")}}, + }, + }, testAccountID) + require.NoError(t, err) + assert.Len(t, out.Vpcs, 1) + + out, err = svc.DescribeVpcs(context.Background(), &ec2.DescribeVpcsInput{ + Filters: []*ec2.Filter{ + {Name: aws.String("isDefault"), Values: []*string{aws.String("true")}}, + }, + }, testAccountID) + require.NoError(t, err) + assert.Empty(t, out.Vpcs) +} + +// TestDescribeVpcs_FilterByCidrBlockAssociation covers the association-set +// filter against the primary CIDR, the only CIDR VPCRecord stores today. +func TestDescribeVpcs_FilterByCidrBlockAssociation(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + createTestVPC(t, svc, "10.0.0.0/16") + createTestVPC(t, svc, "172.16.0.0/16") + + out, err := svc.DescribeVpcs(context.Background(), &ec2.DescribeVpcsInput{ + Filters: []*ec2.Filter{ + {Name: aws.String("cidr-block-association.cidr-block"), Values: []*string{aws.String("10.0.0.0/16")}}, + }, + }, testAccountID) + require.NoError(t, err) + assert.Len(t, out.Vpcs, 1) + assert.Equal(t, "10.0.0.0/16", *out.Vpcs[0].CidrBlock) } func TestDescribeVpcs_FilterWildcard(t *testing.T) { diff --git a/spinifex/testutil/segscanoracle/segscanoracle.go b/spinifex/testutil/segscanoracle/segscanoracle.go index 7d50f5f04..2a47c0457 100644 --- a/spinifex/testutil/segscanoracle/segscanoracle.go +++ b/spinifex/testutil/segscanoracle/segscanoracle.go @@ -151,7 +151,10 @@ func build(t *testing.T, srcDir string) (string, error) { ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) defer cancel() - cmd := exec.CommandContext(ctx, "go", "build", "-o", bin, ".") + // -buildvcs=false for the same reason as GOWORK=off below: the build must + // not depend on ambient repository state. Stamping fails outright when the + // VCS probe errors, which it does for a synthetic module under a temp dir. + cmd := exec.CommandContext(ctx, "go", "build", "-buildvcs=false", "-o", bin, ".") cmd.Dir = srcDir // GOWORK=off: scripts/segscan is its own module with its own go.mod/go.sum. // An ancestor directory of srcDir may have an unrelated go.work (the mulga diff --git a/spinifex/utils/nats.go b/spinifex/utils/nats.go index b49acdaba..2beac16b2 100644 --- a/spinifex/utils/nats.go +++ b/spinifex/utils/nats.go @@ -239,6 +239,21 @@ const AccountIDHeader = "X-Account-ID" // PrincipalARNHeader carries the caller's resolved IAM principal ARN from gateway to daemon handlers. const PrincipalARNHeader = "X-Principal-ARN" +// ReservationIDHeader carries a gateway-assigned launch-group reservation ID +// for a node-targeted RunInstances request. A node that does not read this +// header (an older build, or a request that never set it) mints its own ID +// as before, so the header is purely additive and never required. +const ReservationIDHeader = "X-Reservation-Id" + +// ReservationIDFromMsg extracts the gateway-assigned reservation ID from a +// NATS message header. Returns "" when absent. +func ReservationIDFromMsg(msg *nats.Msg) string { + if msg == nil || msg.Header == nil { + return "" + } + return msg.Header.Get(ReservationIDHeader) +} + // NATSHeader is an extra request header passed to NATSRequest beyond the // always-set X-Account-ID. type NATSHeader struct{ Key, Value string }