diff --git a/cmd/spinifex/cmd/admin_account_delete.go b/cmd/spinifex/cmd/admin_account_delete.go index dd725de4c..baff38f93 100644 --- a/cmd/spinifex/cmd/admin_account_delete.go +++ b/cmd/spinifex/cmd/admin_account_delete.go @@ -5,7 +5,9 @@ import ( "context" "errors" "fmt" + "maps" "os" + "slices" "strings" "time" @@ -90,7 +92,10 @@ func runAccountDelete(cmd *cobra.Command, args []string) error { node.Predastore.SecretKey, ) - engine, err := accountteardown.NewClusterEngine(ctx, nc, len(cfg.Nodes), svc, buckets) + // The configured node set, in the namespace daemons reply under, is what + // the instance inventory must hear from in full before it reads as empty. + nodeIDs := slices.Collect(maps.Keys(cfg.Nodes)) + engine, err := accountteardown.NewClusterEngine(ctx, nc, len(cfg.Nodes), nodeIDs, svc, buckets) if err != nil { return err } diff --git a/spinifex/accountteardown/cluster.go b/spinifex/accountteardown/cluster.go index c8a424655..02c36237b 100644 --- a/spinifex/accountteardown/cluster.go +++ b/spinifex/accountteardown/cluster.go @@ -21,6 +21,7 @@ func NewClusterEngine( ctx context.Context, nc *nats.Conn, expectedNodes int, + nodeIDs []string, svc handlers_iam.IAMService, buckets BucketStore, ) (*Engine, error) { @@ -64,7 +65,7 @@ func NewClusterEngine( // live spot request launches a replacement for anything already terminated. reapers = append(reapers, ELBv2Reapers(nc)...) reapers = append(reapers, EC2ExtraReapers(nc)...) - reapers = append(reapers, EC2Reapers(nc, expectedNodes)...) + reapers = append(reapers, EC2Reapers(nc, expectedNodes, nodeIDs)...) reapers = append(reapers, S3Reapers(buckets)...) reapers = append(reapers, ACMReapers(nc)...) diff --git a/spinifex/accountteardown/cluster_test.go b/spinifex/accountteardown/cluster_test.go index f132cea2a..481723737 100644 --- a/spinifex/accountteardown/cluster_test.go +++ b/spinifex/accountteardown/cluster_test.go @@ -23,13 +23,13 @@ func TestNewClusterEngineCoversEveryStage(t *testing.T) { svc, err := handlers_iam.NewIAMServiceImpl(t.Context(), nc, masterKey) require.NoError(t, err) - engine, err := NewClusterEngine(t.Context(), nc, 1, svc, newFakeBuckets()) + engine, err := NewClusterEngine(t.Context(), nc, 1, []string{"node-1"}, svc, newFakeBuckets()) require.NoError(t, err) require.NotNil(t, engine.Accounts) // A teardown with no object store would report every stage drained and // still delete the account record, leaving the tenant's objects behind. - _, err = NewClusterEngine(t.Context(), nc, 1, svc, nil) + _, err = NewClusterEngine(t.Context(), nc, 1, []string{"node-1"}, svc, nil) require.Error(t, err) for _, stage := range Stages() { @@ -51,7 +51,7 @@ func TestClusterEnginePrechecks(t *testing.T) { svc, err := handlers_iam.NewIAMServiceImpl(t.Context(), nc, masterKey) require.NoError(t, err) - engine, err := NewClusterEngine(t.Context(), nc, 1, svc, newFakeBuckets()) + engine, err := NewClusterEngine(t.Context(), nc, 1, []string{"node-1"}, svc, newFakeBuckets()) require.NoError(t, err) // Account ids are sequential and the first is the protected super admin. diff --git a/spinifex/accountteardown/reapers_ec2.go b/spinifex/accountteardown/reapers_ec2.go index 3d98a5e69..dfb91aa1f 100644 --- a/spinifex/accountteardown/reapers_ec2.go +++ b/spinifex/accountteardown/reapers_ec2.go @@ -31,9 +31,9 @@ import ( // They drive the same NATS-backed operations the gateway serves a customer // request with, which is what keeps OVN teardown, IPAM release and quota // accounting correct without teardown having to reimplement any of it. -func EC2Reapers(nc *nats.Conn, expectedNodes int) []Reaper { +func EC2Reapers(nc *nats.Conn, expectedNodes int, nodeIDs []string) []Reaper { return []Reaper{ - &instanceReaper{nc: nc, expectedNodes: expectedNodes}, + &instanceReaper{nc: nc, nodeIDs: nodeIDs}, // Snapshots before volumes: a volume with a live snapshot refuses to // delete, because snapshot-backed clones read from its chunk prefix. @@ -67,17 +67,22 @@ func EC2Reapers(nc *nats.Conn, expectedNodes int) []Reaper { var terminalInstanceStates = map[string]bool{"terminated": true, "shutting-down": true} type instanceReaper struct { - nc *nats.Conn - expectedNodes int + nc *nats.Conn + // nodeIDs is the configured node set; all must answer before an empty + // listing counts as drained. + nodeIDs []string + + // describeOpts tunes the inventory fan-out; production leaves it empty. + describeOpts []gateway_ec2_instance.DescribeOption } func (r *instanceReaper) Kind() string { return "instance" } func (r *instanceReaper) Stage() Stage { return StageCompute } func (r *instanceReaper) List(ctx context.Context, accountID string) ([]Resource, error) { - // Checked: a partial answer here would read as "no instances left" and let - // storage deletion start while a guest still holds a volume. - out, err := gateway_ec2_instance.DescribeInstancesChecked(ctx, &ec2.DescribeInstancesInput{}, r.nc, r.expectedNodes, nil, accountID) + // Complete or nothing: a partial answer would read as "no instances left" + // and let storage deletion start while a guest still holds a volume. + out, err := gateway_ec2_instance.DescribeAccountInstancesComplete(ctx, r.nc, r.nodeIDs, accountID, r.describeOpts...) if err != nil { return nil, err } diff --git a/spinifex/accountteardown/reapers_ec2_inventory_test.go b/spinifex/accountteardown/reapers_ec2_inventory_test.go new file mode 100644 index 000000000..1768d7284 --- /dev/null +++ b/spinifex/accountteardown/reapers_ec2_inventory_test.go @@ -0,0 +1,194 @@ +package accountteardown + +//test:in-package — the instance reaper is unexported, and whether its listing +// can be mistaken for an empty account is the safety property under test. + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + gateway_ec2_instance "github.com/mulgadc/spinifex/spinifex/gateway/ec2/instance" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var inventoryNodes = []string{"node-1", "node-2"} + +// testInventoryReaper builds the instance reaper for a two-node cluster, with +// a short fan-out deadline so a silent node costs the test little. +func testInventoryReaper(nc *nats.Conn) *instanceReaper { + return &instanceReaper{nc: nc, nodeIDs: inventoryNodes, describeOpts: []gateway_ec2_instance.DescribeOption{ + gateway_ec2_instance.WithFanoutTimeout(300 * time.Millisecond), + }} +} + +// answerAsNode replies on subject the way a daemon does, carrying the node ID +// header the fan-out attributes frames by. +func answerAsNode(t *testing.T, nc *nats.Conn, subject, nodeID string, payload []byte) { + t.Helper() + sub, err := nc.Subscribe(subject, func(msg *nats.Msg) { + reply := nats.NewMsg(msg.Reply) + reply.Data = payload + if nodeID != "" { + reply.Header.Set(utils.NodeIDHeader, nodeID) + } + _ = msg.RespondMsg(reply) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) +} + +func describeOutput(t *testing.T, instanceIDs ...string) []byte { + t.Helper() + out := ec2.DescribeInstancesOutput{Reservations: []*ec2.Reservation{}} + if len(instanceIDs) > 0 { + reservation := &ec2.Reservation{ReservationId: aws.String("r-" + instanceIDs[0])} + for _, id := range instanceIDs { + reservation.Instances = append(reservation.Instances, &ec2.Instance{ + InstanceId: aws.String(id), + State: &ec2.InstanceState{Name: aws.String("running")}, + }) + } + out.Reservations = append(out.Reservations, reservation) + } + payload, err := json.Marshal(out) + require.NoError(t, err) + return payload +} + +// emptyInstanceBuckets answers the stopped and terminated bucket queries with +// nothing, so only the running-instance fan-out decides completeness. +func emptyInstanceBuckets(t *testing.T, nc *nats.Conn) { + t.Helper() + answerAsNode(t, nc, "ec2.DescribeStoppedInstances", "", describeOutput(t)) + answerAsNode(t, nc, "ec2.DescribeTerminatedInstances", "", describeOutput(t)) +} + +// A node that never answers is the timeout case: its running guests are absent +// from the aggregate, so an empty answer from the rest is not evidence the +// account has no instances. +func TestInstanceReaperFailsClosedWhenANodeDoesNotAnswer(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + emptyInstanceBuckets(t, nc) + answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t)) + require.NoError(t, nc.Flush()) + + found, err := testInventoryReaper(nc).List(testCtx(t), testAccountID) + + require.Error(t, err, "a listing missing a node must not read as an empty account, got %v", found) + code, _, ok := awserrors.ResolveErrorDetail(err) + require.True(t, ok) + assert.Equal(t, awserrors.ErrorServiceUnavailable, code) +} + +// The partial list is the dangerous shape: the reaper would terminate what it +// can see and then report the stage drained on the next, equally partial pass. +func TestInstanceReaperFailsClosedOnAPartialList(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + emptyInstanceBuckets(t, nc) + answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t, "i-on-node-1")) + require.NoError(t, nc.Flush()) + + found, err := testInventoryReaper(nc).List(testCtx(t), testAccountID) + + require.Error(t, err, "a partial listing must not be returned as the inventory, got %v", found) +} + +// An error frame from a node is no better than silence: it says nothing about +// what that node is running. +func TestInstanceReaperFailsClosedWhenANodeAnswersWithAnError(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + emptyInstanceBuckets(t, nc) + answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t)) + answerAsNode(t, nc, "ec2.DescribeInstances", "node-2", utils.GenerateErrorPayload(awserrors.ErrorServerInternal)) + require.NoError(t, nc.Flush()) + + _, err := testInventoryReaper(nc).List(testCtx(t), testAccountID) + + require.Error(t, err) +} + +// Completeness is judged against the configured node set, not a count of +// replies: two frames from one node while another is silent is still missing +// a node. +func TestInstanceReaperFailsClosedWhenOneNodeAnswersTwice(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + emptyInstanceBuckets(t, nc) + answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t)) + answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t)) + require.NoError(t, nc.Flush()) + + _, err := testInventoryReaper(nc).List(testCtx(t), testAccountID) + + require.Error(t, err) +} + +// Stopped instances live in a bucket, not on a node. A failed bucket query +// hides them as surely as a silent node hides running ones. +func TestInstanceReaperFailsClosedWhenTheStoppedBucketFails(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + answerAsNode(t, nc, "ec2.DescribeStoppedInstances", "", utils.GenerateErrorPayload(awserrors.ErrorServerInternal)) + answerAsNode(t, nc, "ec2.DescribeTerminatedInstances", "", describeOutput(t)) + answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t)) + answerAsNode(t, nc, "ec2.DescribeInstances", "node-2", describeOutput(t)) + require.NoError(t, nc.Flush()) + + _, err := testInventoryReaper(nc).List(testCtx(t), testAccountID) + + require.Error(t, err) +} + +// The healthy cluster still lists: every configured node answered, so the +// union is the account's inventory. +func TestInstanceReaperListsEveryNodeWhenAllAnswer(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + emptyInstanceBuckets(t, nc) + answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t, "i-on-node-1")) + answerAsNode(t, nc, "ec2.DescribeInstances", "node-2", describeOutput(t, "i-on-node-2")) + require.NoError(t, nc.Flush()) + + started := time.Now() + found, err := testInventoryReaper(nc).List(testCtx(t), testAccountID) + require.NoError(t, err) + + var ids []string + for _, resource := range found { + ids = append(ids, resource.ID) + } + assert.ElementsMatch(t, []string{"i-on-node-1", "i-on-node-2"}, ids) + assert.Less(t, time.Since(started), 2*time.Second, "a fully answered fan-out must not wait out its deadline") +} + +// End to end through the engine: with a node silent, the compute stage must +// not read as drained and the account record must survive. +func TestTeardownDoesNotReportComputeDrainedWhileANodeIsSilent(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + emptyInstanceBuckets(t, nc) + answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t)) + require.NoError(t, nc.Flush()) + + accounts := newFakeAccounts() + volumes := &fakeReaper{kind: "volume", stage: StageStorage, + resources: []Resource{{Kind: "volume", ID: "vol-held-by-node-2"}}} + engine := quietEngine(accounts, testInventoryReaper(nc), volumes) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + result, err := engine.Teardown(ctx, request()) + + require.Error(t, err, "teardown must not succeed on an incomplete instance inventory") + assert.ErrorContains(t, err, "list instance") + assert.False(t, accounts.deleted, "the account record must survive an incomplete inventory") + for _, stage := range result.Stages { + assert.NotEqual(t, StageCompute, stage.Stage, "the compute stage must not be reported finished") + } + assert.Equal(t, 0, volumes.deletesAttempted, "storage must not be touched before compute is proven drained") +} diff --git a/spinifex/accountteardown/reapers_ec2_test.go b/spinifex/accountteardown/reapers_ec2_test.go index 65c37074f..de3a830c3 100644 --- a/spinifex/accountteardown/reapers_ec2_test.go +++ b/spinifex/accountteardown/reapers_ec2_test.go @@ -53,6 +53,10 @@ func newFakeCluster(t *testing.T) *fakeCluster { return cluster } +// fakeClusterNode is the one node the fake cluster answers as. Replies carry +// its ID because an inventory fan-out only counts attributable frames. +const fakeClusterNode = "node-1" + func (c *fakeCluster) serve(msg *nats.Msg) { c.mu.Lock() c.calls = append(c.calls, msg.Subject) @@ -61,18 +65,24 @@ func (c *fakeCluster) serve(msg *nats.Msg) { reply, hasReply := c.replies[msg.Subject] c.mu.Unlock() + respond := func(data []byte) { + out := nats.NewMsg(msg.Reply) + out.Data = data + out.Header.Set(utils.NodeIDHeader, fakeClusterNode) + _ = msg.RespondMsg(out) + } switch { case isError: - _ = msg.Respond(utils.GenerateErrorPayload(code)) + respond(utils.GenerateErrorPayload(code)) case hasReply: payload, err := json.Marshal(reply) if err != nil { - _ = msg.Respond(utils.GenerateErrorPayload("InternalError")) + respond(utils.GenerateErrorPayload("InternalError")) return } - _ = msg.Respond(payload) + respond(payload) default: - _ = msg.Respond([]byte(`{}`)) + respond([]byte(`{}`)) } } @@ -112,7 +122,7 @@ func testCtx(t *testing.T) context.Context { // The stage graph is what keeps teardown converging, so it is asserted rather // than left to the order the constructor happens to list. func TestEC2ReapersAreOrderedByDependency(t *testing.T) { - reapers := EC2Reapers(nil, 1) + reapers := EC2Reapers(nil, 1, []string{"node-1"}) var kinds []string for _, reaper := range reapers { @@ -157,7 +167,7 @@ func TestInstanceReaperSkipsTerminalStates(t *testing.T) { }}}, }) - reaper := &instanceReaper{nc: cluster.nc, expectedNodes: 1} + reaper := &instanceReaper{nc: cluster.nc, nodeIDs: []string{fakeClusterNode}} found, err := reaper.List(testCtx(t), "000000000042") require.NoError(t, err) @@ -170,7 +180,7 @@ func TestInstanceReaperSkipsTerminalStates(t *testing.T) { // makes a stuck host the case --force exists for. func TestInstanceReaperTerminates(t *testing.T) { cluster := newFakeCluster(t) - reaper := &instanceReaper{nc: cluster.nc, expectedNodes: 1} + reaper := &instanceReaper{nc: cluster.nc, nodeIDs: []string{fakeClusterNode}} require.NoError(t, reaper.Delete(testCtx(t), "000000000042", Resource{Kind: "instance", ID: "i-0abc0123456789def"}, false)) @@ -679,7 +689,7 @@ func TestNetworkInterfaceReaperTreatsAMissingInterfaceAsDeleted(t *testing.T) { // subnet delete is refused while an interface is still in it. func TestEC2ReapersRemoveInterfacesBeforeSubnets(t *testing.T) { var eni, subnet = -1, -1 - for i, reaper := range EC2Reapers(nil, 1) { + for i, reaper := range EC2Reapers(nil, 1, []string{"node-1"}) { switch reaper.Kind() { case "network-interface": eni = i diff --git a/spinifex/accountteardown/reapers_ecs_test.go b/spinifex/accountteardown/reapers_ecs_test.go index 12e345433..6bd1b56d0 100644 --- a/spinifex/accountteardown/reapers_ecs_test.go +++ b/spinifex/accountteardown/reapers_ecs_test.go @@ -178,7 +178,7 @@ func TestECSClusterNameFromARN(t *testing.T) { // stopped first, which is what puts ECS ahead of it in the compute stage. func TestECSReapersRunBeforeTheInstanceReaper(t *testing.T) { reapers := ECSReapers(nil) - reapers = append(reapers, EC2Reapers(nil, 1)...) + reapers = append(reapers, EC2Reapers(nil, 1, []string{"node-1"})...) ecsIndex := indexOfKind(reapers, "ecs-cluster") instanceIndex := indexOfKind(reapers, "instance") @@ -197,7 +197,7 @@ func TestECSReapersRunBeforeTheInstanceReaper(t *testing.T) { // the registration order above survives it. func TestSortReapersKeepsECSAheadOfInstances(t *testing.T) { reapers := ECSReapers(nil) - reapers = append(reapers, EC2Reapers(nil, 1)...) + reapers = append(reapers, EC2Reapers(nil, 1, []string{"node-1"})...) SortReapers(reapers) assert.Less(t, indexOfKind(reapers, "ecs-cluster"), indexOfKind(reapers, "instance")) diff --git a/spinifex/accountteardown/reapers_s3_test.go b/spinifex/accountteardown/reapers_s3_test.go index d7e144ee7..387c82b70 100644 --- a/spinifex/accountteardown/reapers_s3_test.go +++ b/spinifex/accountteardown/reapers_s3_test.go @@ -277,7 +277,7 @@ func TestS3ReaperRunsInTheStorageStage(t *testing.T) { // The engine sorts reapers by stage and the sort is stable, so a bucket reaper // registered after the EC2 storage reapers stays after them. func TestSortReapersKeepsBucketsAfterVolumes(t *testing.T) { - reapers := append(EC2Reapers(nil, 1), S3Reapers(newFakeBuckets())...) + reapers := append(EC2Reapers(nil, 1, []string{"node-1"}), S3Reapers(newFakeBuckets())...) SortReapers(reapers) storage := make([]string, 0, len(reapers)) diff --git a/spinifex/gateway/admin_deleteaccount.go b/spinifex/gateway/admin_deleteaccount.go index 503b747af..1ff84d65c 100644 --- a/spinifex/gateway/admin_deleteaccount.go +++ b/spinifex/gateway/admin_deleteaccount.go @@ -111,7 +111,7 @@ func (gw *GatewayConfig) adminDeleteAccount(ctx context.Context, body []byte) (a return nil, errors.New(awserrors.ErrorInternalError) } - engine, err := accountteardown.NewClusterEngine(ctx, gw.NATSConn, gw.ExpectedNodes, gw.IAMService, gw.BucketStore) + engine, err := accountteardown.NewClusterEngine(ctx, gw.NATSConn, gw.ExpectedNodes, gw.NodeIDs, gw.IAMService, gw.BucketStore) if err != nil { slog.Error("DeleteAccount: failed to build teardown engine", "err", err) return nil, errors.New(awserrors.ErrorServiceUnavailable) diff --git a/spinifex/gateway/admin_deleteaccount_job_test.go b/spinifex/gateway/admin_deleteaccount_job_test.go index f37fdd5aa..959aeb70a 100644 --- a/spinifex/gateway/admin_deleteaccount_job_test.go +++ b/spinifex/gateway/admin_deleteaccount_job_test.go @@ -17,6 +17,7 @@ import ( handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" "github.com/mulgadc/spinifex/spinifex/objectstore" "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "github.com/stretchr/testify/assert" @@ -44,8 +45,13 @@ func newDeleteAccountGateway(t *testing.T) *GatewayConfig { require.NoError(t, err) for _, subject := range teardownSubjects { + // Replies carry a node ID: the instance inventory only counts a frame + // it can attribute to a configured node. sub, err := nc.Subscribe(subject, func(msg *nats.Msg) { - _ = msg.Respond([]byte(`{}`)) + reply := nats.NewMsg(msg.Reply) + reply.Data = []byte(`{}`) + reply.Header.Set(utils.NodeIDHeader, "node-1") + _ = msg.RespondMsg(reply) }) require.NoError(t, err) t.Cleanup(func() { _ = sub.Unsubscribe() }) @@ -57,6 +63,7 @@ func newDeleteAccountGateway(t *testing.T) *GatewayConfig { IAMService: svc, NATSConn: nc, ExpectedNodes: 1, + NodeIDs: []string{"node-1"}, BucketStore: emptyBucketStore{}, } } diff --git a/spinifex/gateway/ec2/instance/DescribeInstances.go b/spinifex/gateway/ec2/instance/DescribeInstances.go index 72ed5e646..f075881ac 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstances.go +++ b/spinifex/gateway/ec2/instance/DescribeInstances.go @@ -177,6 +177,34 @@ func DescribeInstancesForReconcile(ctx context.Context, input *ec2.DescribeInsta return reservations, complete, err } +// DescribeAccountInstancesComplete lists the account's instances, or returns a +// retryable ServiceUnavailable unless every node in nodeIDs and both instance +// buckets answered cleanly. An empty nodeIDs is an unknown set, so never complete. +func DescribeAccountInstancesComplete(ctx context.Context, natsConn *nats.Conn, nodeIDs []string, accountID string, opts ...DescribeOption) (*ec2.DescribeInstancesOutput, error) { + if len(nodeIDs) == 0 { + slog.ErrorContext(ctx, "DescribeInstances: complete listing refused, configured node set unknown") + return nil, incompleteListingError() + } + reservations, complete, _, err := gatherInstances(ctx, &ec2.DescribeInstancesInput{}, natsConn, 0, nodeIDs, accountID, opts...) + if err != nil { + return nil, err + } + if !complete { + slog.WarnContext(ctx, "DescribeInstances: complete listing refused, fan-out or bucket query incomplete", + "configured_nodes", len(nodeIDs)) + return nil, incompleteListingError() + } + return &ec2.DescribeInstancesOutput{Reservations: reservations}, nil +} + +func incompleteListingError() error { + return errors.Join( + awserrors.Errorf(awserrors.ErrorServiceUnavailable, + "cannot list instances: not every node and instance store answered"), + awserrors.RetryAfter(awserrors.ErrorServiceUnavailable, describeIncompleteRetryAfter), + ) +} + // gatherInstances runs the running-instance fan-out plus the stopped/terminated // KV bucket queries and aggregates every reservation. client4xx carries // the first deterministic 4xx for the lenient caller to surface when nothing diff --git a/spinifex/gateway/ec2/instance/DescribeInstances_test.go b/spinifex/gateway/ec2/instance/DescribeInstances_test.go index 6b69700c0..91364519d 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstances_test.go +++ b/spinifex/gateway/ec2/instance/DescribeInstances_test.go @@ -956,3 +956,115 @@ func TestAcquireAbsenceProofSlot_DoesNotBlockWhenExhausted(t *testing.T) { t.Fatal("acquireAbsenceProofSlot blocked; it must refuse immediately so the backlog cannot grow") } } + +// --- Complete account-wide listing --- + +func assertIncompleteListing(t *testing.T, output *ec2.DescribeInstancesOutput, err error) { + t.Helper() + assert.Nil(t, output, "an incomplete listing must not return the partial reservations") + require.Error(t, err) + code, _, 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 listing is retryable") +} + +func TestDescribeAccountInstancesComplete_MissingNodeRefuses(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + subscribeEmptyInstanceBuckets(t, nc) + subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", mustMarshalDescribeOutput(t, &ec2.DescribeInstancesOutput{ + Reservations: []*ec2.Reservation{{ReservationId: aws.String("r-1"), Instances: []*ec2.Instance{{InstanceId: aws.String("i-on-node-1")}}}}, + })) + nc.Flush() + + output, err := DescribeAccountInstancesComplete(context.Background(), nc, []string{"node-1", "node-2"}, "123456789012", + WithFanoutTimeout(300*time.Millisecond)) + + assertIncompleteListing(t, output, err) +} + +func TestDescribeAccountInstancesComplete_AllNodesAnswer(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + subscribeEmptyInstanceBuckets(t, nc) + subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", mustMarshalDescribeOutput(t, &ec2.DescribeInstancesOutput{ + Reservations: []*ec2.Reservation{{ReservationId: aws.String("r-1"), Instances: []*ec2.Instance{{InstanceId: aws.String("i-on-node-1")}}}}, + })) + subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-2", []byte(`{}`)) + nc.Flush() + + output, err := DescribeAccountInstancesComplete(context.Background(), nc, []string{"node-1", "node-2"}, "123456789012", + WithFanoutTimeout(2*time.Second)) + + require.NoError(t, err) + require.Len(t, output.Reservations, 1) + assert.Equal(t, "i-on-node-1", *output.Reservations[0].Instances[0].InstanceId) +} + +// Without the configured node set there is nothing to judge completeness +// against, and counting whoever answered is the failure being prevented. +func TestDescribeAccountInstancesComplete_UnknownNodeSetRefuses(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + subscribeEmptyInstanceBuckets(t, nc) + subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`)) + nc.Flush() + + output, err := DescribeAccountInstancesComplete(context.Background(), nc, nil, "123456789012", + WithFanoutTimeout(300*time.Millisecond)) + + assertIncompleteListing(t, output, err) +} + +// A frame with no node ID cannot be attributed, so it might be standing in +// for the node that went quiet. +func TestDescribeAccountInstancesComplete_UnattributedFrameRefuses(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + subscribeEmptyInstanceBuckets(t, nc) + subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`)) + subscribeAsNode(t, nc, "ec2.DescribeInstances", "", []byte(`{}`)) + nc.Flush() + + output, err := DescribeAccountInstancesComplete(context.Background(), nc, []string{"node-1", "node-2"}, "123456789012", + WithFanoutTimeout(300*time.Millisecond)) + + assertIncompleteListing(t, output, err) +} + +func TestDescribeAccountInstancesComplete_TerminatedBucketFailureRefuses(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + subscribeAsNode(t, nc, "ec2.DescribeStoppedInstances", "", []byte(`{}`)) + subscribeAsNode(t, nc, "ec2.DescribeTerminatedInstances", "", utils.GenerateErrorPayload(awserrors.ErrorServerInternal)) + subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`)) + nc.Flush() + + output, err := DescribeAccountInstancesComplete(context.Background(), nc, []string{"node-1"}, "123456789012", + WithFanoutTimeout(2*time.Second)) + + assertIncompleteListing(t, output, err) +} + +// Pins the lenient by-ID shape: an instance on a silent node comes back as an +// empty success, indistinguishable from absence. Anything that must prove an +// instance gone has to use DescribeInstancesChecked with the node set instead. +func TestDescribeInstances_ExplicitIDOnSilentNode_IsAnEmptySuccess(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + subscribeEmptyInstanceBuckets(t, nc) + subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{"Reservations":[]}`)) + nc.Flush() + + input := &ec2.DescribeInstancesInput{InstanceIds: []*string{aws.String("i-on-the-silent-node")}} + output, err := DescribeInstances(context.Background(), input, nc, 2, "123456789012", + WithFanoutTimeout(300*time.Millisecond)) + + require.NoError(t, err) + assert.Empty(t, output.Reservations) + + _, err = DescribeInstancesChecked(context.Background(), input, nc, 0, []string{"node-1", "node-2"}, "123456789012", + WithFanoutTimeout(300*time.Millisecond)) + assertIncompleteSweepUnavailable(t, err, "i-on-the-silent-node") +}