diff --git a/spinifex/accountteardown/reapers_ec2.go b/spinifex/accountteardown/reapers_ec2.go index dfb91aa1f..c8b9e6264 100644 --- a/spinifex/accountteardown/reapers_ec2.go +++ b/spinifex/accountteardown/reapers_ec2.go @@ -39,7 +39,7 @@ func EC2Reapers(nc *nats.Conn, expectedNodes int, nodeIDs []string) []Reaper { // delete, because snapshot-backed clones read from its chunk prefix. &snapshotReaper{nc: nc}, &imageReaper{nc: nc}, - &volumeReaper{nc: nc, expectedNodes: expectedNodes}, + &volumeReaper{nc: nc, nodeIDs: nodeIDs}, // Reverse of creation order. The VPC cannot go until everything // inside it has, and the gateway enforces that. @@ -114,8 +114,10 @@ func (r *instanceReaper) Delete(ctx context.Context, accountID string, resource } type volumeReaper struct { - nc *nats.Conn - expectedNodes int + nc *nats.Conn + // nodeIDs is the configured node set; every one must answer, by identity, + // before a volume's holder can be proven gone or the delete guard passes. + nodeIDs []string } func (r *volumeReaper) Kind() string { return "volume" } @@ -149,16 +151,16 @@ func (r *volumeReaper) List(ctx context.Context, accountID string) ([]Resource, func (r *volumeReaper) Delete(ctx context.Context, accountID string, resource Resource, force bool) error { _, err := gateway_ec2_volume.DeleteVolume(ctx, &ec2.DeleteVolumeInput{ VolumeId: aws.String(resource.ID), - }, r.nc, r.expectedNodes, accountID) + }, r.nc, r.nodeIDs, accountID) if isAlreadyGone(err) { return nil } if !force { return err } - // Force path: the ordinary delete refuses an attached volume, and an - // instance that will not terminate would otherwise strand it forever. - return forceDeleteVolume(ctx, r.nc, r.expectedNodes, accountID, resource.ID) + // Force path: the ordinary delete refuses an attached volume, and a holder + // that is gone but still on the record would otherwise strand it forever. + return forceDeleteVolume(ctx, r.nc, r.nodeIDs, accountID, resource.ID, err) } type snapshotReaper struct{ nc *nats.Conn } @@ -698,22 +700,120 @@ const forceDetachSubject = "ec2.ForceDetachVolume" // unwell rather than that the work is large. const forceDetachTimeout = 30 * time.Second -// forceDeleteVolume clears the attachment the ordinary delete refuses to -// override, then deletes. -// -// This is the deadlock the force path exists for: a volume attached to an -// instance that will not terminate leaves both undeletable, and the account -// can never be emptied. -func forceDeleteVolume(ctx context.Context, nc *nats.Conn, expectedNodes int, accountID, volumeID string) error { - _, err := utils.NATSRequest[ec2.VolumeAttachment](ctx, nc, forceDetachSubject, +// forceDeleteVolume clears an attachment the ordinary delete refuses to +// override, then deletes, but only once every instance the volume names is +// proven absent. Anything short of proof leaves the volume and its record alone. +func forceDeleteVolume(ctx context.Context, nc *nats.Conn, nodeIDs []string, accountID, volumeID string, refusal error) error { + holders, err := volumeHolders(ctx, nc, accountID, volumeID) + if err != nil { + if isAlreadyGone(err) { + return nil + } + return fmt.Errorf("volume %s retained: cannot read its attachment: %w", volumeID, err) + } + if len(holders) == 0 { + // No holder on record, so the refusal is not an attachment and + // clearing the record would change nothing but the evidence. + return refusal + } + for _, instanceID := range holders { + if err := proveInstanceAbsent(ctx, nc, nodeIDs, accountID, instanceID); err != nil { + return fmt.Errorf("volume %s retained: %w", volumeID, err) + } + } + + _, err = utils.NATSRequest[ec2.VolumeAttachment](ctx, nc, forceDetachSubject, &ec2.DetachVolumeInput{VolumeId: aws.String(volumeID), Force: aws.Bool(true)}, forceDetachTimeout, accountID) if err != nil && !isAlreadyGone(err) { return fmt.Errorf("force detach %s: %w", volumeID, err) } + // The ordinary delete re-checks that the record is unattached and that no + // live instance maps the volume, so a re-attach in between still refuses. _, err = gateway_ec2_volume.DeleteVolume(ctx, &ec2.DeleteVolumeInput{ VolumeId: aws.String(volumeID), - }, nc, expectedNodes, accountID) + }, nc, nodeIDs, accountID) return ignoreAlreadyGone(err) } + +// volumeHolders returns the instances the volume's own record says hold it, +// read fresh rather than from the listing that queued the delete. +func volumeHolders(ctx context.Context, nc *nats.Conn, accountID, volumeID string) ([]string, error) { + out, err := gateway_ec2_volume.DescribeVolumes(ctx, &ec2.DescribeVolumesInput{ + VolumeIds: []*string{aws.String(volumeID)}, + }, nc, accountID) + if err != nil { + return nil, err + } + var holders []string + for _, volume := range out.Volumes { + if volume == nil || aws.StringValue(volume.VolumeId) != volumeID { + continue + } + for _, attachment := range volume.Attachments { + if attachment != nil && aws.StringValue(attachment.InstanceId) != "" { + holders = append(holders, aws.StringValue(attachment.InstanceId)) + } + } + } + return holders, nil +} + +// proveInstanceAbsent accepts only a definitive InvalidInstanceID.NotFound or +// a terminated copy from a describe naming the instance, then confirms both on +// a listing every configured node answered by identity. Anything else is not proof. +func proveInstanceAbsent(ctx context.Context, nc *nats.Conn, nodeIDs []string, accountID, instanceID string) error { + if len(nodeIDs) == 0 { + return fmt.Errorf("cannot prove holding instance %s is gone: the configured node set is unknown", instanceID) + } + named, err := gateway_ec2_instance.DescribeInstancesChecked(ctx, &ec2.DescribeInstancesInput{ + InstanceIds: []*string{aws.String(instanceID)}, + }, nc, 0, nodeIDs, accountID) + if err != nil { + if code, ok := awserrors.ResolveErrorCode(err); !ok || code != awserrors.ErrorInvalidInstanceIDNotFound { + return fmt.Errorf("cannot prove holding instance %s is gone: %w", instanceID, err) + } + } else { + found, err := requireTerminated(named.Reservations, instanceID) + if err != nil { + return err + } + if !found { + return fmt.Errorf("cannot prove holding instance %s is gone: the describe naming it returned neither the instance nor NotFound", instanceID) + } + } + + // The named describe stops collecting once the id turns up, and a lone + // node's 4xx can surface as NotFound, so neither answer is complete alone. + listing, err := gateway_ec2_instance.DescribeAccountInstancesComplete(ctx, nc, nodeIDs, accountID) + if err != nil { + return fmt.Errorf("cannot prove holding instance %s is gone: %w", instanceID, err) + } + _, err = requireTerminated(listing.Reservations, instanceID) + return err +} + +// requireTerminated reports whether instanceID appears in reservations, and +// fails if any copy of it is in a state other than terminated. +func requireTerminated(reservations []*ec2.Reservation, instanceID string) (found bool, err error) { + for _, reservation := range reservations { + if reservation == nil { + continue + } + for _, instance := range reservation.Instances { + if instance == nil || aws.StringValue(instance.InstanceId) != instanceID { + continue + } + found = true + state := "in an unknown state" + if instance.State != nil && aws.StringValue(instance.State.Name) != "" { + state = aws.StringValue(instance.State.Name) + } + if state != ec2.InstanceStateNameTerminated { + return true, fmt.Errorf("holding instance %s is still %s", instanceID, state) + } + } + } + return found, nil +} diff --git a/spinifex/accountteardown/reapers_ec2_test.go b/spinifex/accountteardown/reapers_ec2_test.go index de3a830c3..758a6237d 100644 --- a/spinifex/accountteardown/reapers_ec2_test.go +++ b/spinifex/accountteardown/reapers_ec2_test.go @@ -32,6 +32,7 @@ type fakeCluster struct { requests map[string][]byte replies map[string]any errorCode map[string]string + handlers map[string]func(data []byte) (reply any, errorCode string) } func newFakeCluster(t *testing.T) *fakeCluster { @@ -43,6 +44,7 @@ func newFakeCluster(t *testing.T) *fakeCluster { requests: map[string][]byte{}, replies: map[string]any{}, errorCode: map[string]string{}, + handlers: map[string]func([]byte) (any, string){}, } sub, err := nc.Subscribe("ec2.>", cluster.serve) @@ -63,6 +65,7 @@ func (c *fakeCluster) serve(msg *nats.Msg) { c.requests[msg.Subject] = msg.Data code, isError := c.errorCode[msg.Subject] reply, hasReply := c.replies[msg.Subject] + handler, hasHandler := c.handlers[msg.Subject] c.mu.Unlock() respond := func(data []byte) { @@ -71,6 +74,11 @@ func (c *fakeCluster) serve(msg *nats.Msg) { out.Header.Set(utils.NodeIDHeader, fakeClusterNode) _ = msg.RespondMsg(out) } + if hasHandler && !isError { + reply, code = handler(msg.Data) + isError, hasReply = code != "", true + } + switch { case isError: respond(utils.GenerateErrorPayload(code)) @@ -98,6 +106,29 @@ func (c *fakeCluster) fail(subject, code string) { c.errorCode[subject] = code } +// handle answers subject from state the test owns, for doubles whose answer +// must change as the reaper acts — an attachment a force detach clears. +func (c *fakeCluster) handle(subject string, handler func(data []byte) (any, string)) { + c.mu.Lock() + defer c.mu.Unlock() + c.handlers[subject] = handler +} + +// echoAsNode adds a second reply to every request on subject, attributed to +// node, so one node can answer twice while another stays silent. +func (c *fakeCluster) echoAsNode(t *testing.T, subject, node string) { + t.Helper() + sub, err := c.nc.Subscribe(subject, func(msg *nats.Msg) { + out := nats.NewMsg(msg.Reply) + out.Data = []byte(`{}`) + out.Header.Set(utils.NodeIDHeader, node) + _ = msg.RespondMsg(out) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + require.NoError(t, c.nc.Flush()) +} + func (c *fakeCluster) called() []string { c.mu.Lock() defer c.mu.Unlock() @@ -119,6 +150,13 @@ func testCtx(t *testing.T) context.Context { return ctx } +// silentClusterCtx outlasts the full fan-out deadline a silent node costs. +func silentClusterCtx(t *testing.T) context.Context { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + t.Cleanup(cancel) + return ctx +} + // 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) { @@ -201,7 +239,7 @@ func TestVolumeReaperCarriesItsAttachment(t *testing.T) { }, }) - reaper := &volumeReaper{nc: cluster.nc, expectedNodes: 1} + reaper := newVolumeReaper(cluster.nc, fakeClusterNode) found, err := reaper.List(testCtx(t), "000000000042") require.NoError(t, err) @@ -216,31 +254,250 @@ func TestVolumeReaperReportsRefusalWithoutForce(t *testing.T) { cluster := newFakeCluster(t) cluster.fail("ec2.DeleteVolume", "VolumeInUse") - reaper := &volumeReaper{nc: cluster.nc, expectedNodes: 1} + reaper := newVolumeReaper(cluster.nc, fakeClusterNode) err := reaper.Delete(testCtx(t), "000000000042", Resource{Kind: "volume", ID: "vol-held"}, false) require.Error(t, err) assert.NotContains(t, cluster.called(), forceDetachSubject) } -// The deadlock the force path exists for: the ordinary delete refuses, so the -// attachment is cleared cluster-wide and the delete retried. -func TestVolumeReaperForceClearsTheAttachment(t *testing.T) { +// heldVolume is one volume record as its owner keeps it: the ordinary delete +// refuses while the record names an instance, and a force detach clears it. +type heldVolume struct { + mu sync.Mutex + id string + holder string + deleted bool + detaches int +} + +func newHeldVolume(cluster *fakeCluster, id, holder string) *heldVolume { + volume := &heldVolume{id: id, holder: holder} + cluster.handle("ec2.DescribeVolumes", func(data []byte) (any, string) { + var input ec2.DescribeVolumesInput + _ = json.Unmarshal(data, &input) + volume.mu.Lock() + defer volume.mu.Unlock() + if volume.deleted { + if len(input.VolumeIds) > 0 { + return nil, awserrors.ErrorInvalidVolumeNotFound + } + return ec2.DescribeVolumesOutput{}, "" + } + described := &ec2.Volume{VolumeId: aws.String(volume.id), State: aws.String("available")} + if volume.holder != "" { + described.State = aws.String("in-use") + described.Attachments = []*ec2.VolumeAttachment{ + {VolumeId: aws.String(volume.id), InstanceId: aws.String(volume.holder), State: aws.String("attached")}, + } + } + return ec2.DescribeVolumesOutput{Volumes: []*ec2.Volume{described}}, "" + }) + cluster.handle(forceDetachSubject, func([]byte) (any, string) { + volume.mu.Lock() + defer volume.mu.Unlock() + volume.detaches++ + previous := volume.holder + volume.holder = "" + return ec2.VolumeAttachment{VolumeId: aws.String(volume.id), InstanceId: aws.String(previous)}, "" + }) + cluster.handle("ec2.DeleteVolume", func([]byte) (any, string) { + volume.mu.Lock() + defer volume.mu.Unlock() + switch { + case volume.deleted: + return nil, awserrors.ErrorInvalidVolumeNotFound + case volume.holder != "": + return nil, awserrors.ErrorVolumeInUse + } + volume.deleted = true + return ec2.DeleteVolumeOutput{}, "" + }) + return volume +} + +func (v *heldVolume) state() (deleted bool, holder string, detaches int) { + v.mu.Lock() + defer v.mu.Unlock() + return v.deleted, v.holder, v.detaches +} + +func instancesIn(state string, instanceIDs ...string) ec2.DescribeInstancesOutput { + reservation := &ec2.Reservation{ReservationId: aws.String("r-" + state)} + for _, id := range instanceIDs { + reservation.Instances = append(reservation.Instances, &ec2.Instance{ + InstanceId: aws.String(id), State: &ec2.InstanceState{Name: aws.String(state)}, + }) + } + return ec2.DescribeInstancesOutput{Reservations: []*ec2.Reservation{reservation}} +} + +// Force must not clear an attachment a live guest still holds. The instance +// carries no block device mapping for the volume, so the attachment record is +// the only thing standing between the guest and a deleted disk. +func TestVolumeReaperForceRetainsAVolumeALiveInstanceHolds(t *testing.T) { + for _, state := range []string{"running", "stopped", "shutting-down", "pending"} { + t.Run(state, func(t *testing.T) { + cluster := newFakeCluster(t) + volume := newHeldVolume(cluster, "vol-held", "i-live") + subject := "ec2.DescribeInstances" + if state == "stopped" { + subject = "ec2.DescribeStoppedInstances" + } + cluster.reply(subject, instancesIn(state, "i-live")) + + reaper := newVolumeReaper(cluster.nc, fakeClusterNode) + err := reaper.Delete(testCtx(t), "000000000042", Resource{Kind: "volume", ID: "vol-held"}, true) + + require.Error(t, err) + assert.Contains(t, err.Error(), "vol-held") + assert.Contains(t, err.Error(), "i-live") + assert.Contains(t, err.Error(), state) + deleted, holder, detaches := volume.state() + assert.False(t, deleted, "a volume a live guest holds must survive teardown") + assert.Equal(t, "i-live", holder, "the attachment must not be cleared under a live guest") + assert.Zero(t, detaches) + }) + } +} + +// "Cannot prove gone" is not "gone": every way the proof can fall short keeps +// the volume and its attachment exactly as they were. +func TestVolumeReaperForceRetainsAVolumeWhenTheHolderCannotBeProvenGone(t *testing.T) { + twoNodes := []string{fakeClusterNode, "node-2"} + cases := []struct { + name string + setup func(cluster *fakeCluster) + nodes []string + }{ + {"configured node set unknown", func(*fakeCluster) {}, []string{}}, + {"lone node answers NotFound while another is silent", func(cluster *fakeCluster) { + cluster.fail("ec2.DescribeInstances", awserrors.ErrorInvalidInstanceIDNotFound) + }, twoNodes}, + {"terminated copy found while the holder node is silent", func(cluster *fakeCluster) { + cluster.reply("ec2.DescribeTerminatedInstances", instancesIn("terminated", "i-held")) + }, twoNodes}, + {"node describe fails", func(cluster *fakeCluster) { + cluster.fail("ec2.DescribeInstances", awserrors.ErrorServerInternal) + }, nil}, + {"terminated bucket unreachable", func(cluster *fakeCluster) { + cluster.fail("ec2.DescribeTerminatedInstances", awserrors.ErrorServerInternal) + }, nil}, + {"stopped bucket unreachable", func(cluster *fakeCluster) { + cluster.fail("ec2.DescribeStoppedInstances", awserrors.ErrorServerInternal) + }, nil}, + {"sources disagree", func(cluster *fakeCluster) { + cluster.reply("ec2.DescribeTerminatedInstances", instancesIn("terminated", "i-held")) + cluster.reply("ec2.DescribeInstances", instancesIn("running", "i-held")) + }, nil}, + {"volume describe fails", func(cluster *fakeCluster) { + cluster.fail("ec2.DescribeVolumes", awserrors.ErrorServerInternal) + }, nil}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + cluster := newFakeCluster(t) + volume := newHeldVolume(cluster, "vol-held", "i-held") + tc.setup(cluster) + nodes := tc.nodes + if nodes == nil { + nodes = []string{fakeClusterNode} + } + + reaper := newVolumeReaper(cluster.nc, nodes...) + err := reaper.Delete(silentClusterCtx(t), "000000000042", Resource{Kind: "volume", ID: "vol-held"}, true) + + require.Error(t, err) + assert.Contains(t, err.Error(), "vol-held") + deleted, holder, detaches := volume.state() + assert.False(t, deleted) + assert.Equal(t, "i-held", holder) + assert.Zero(t, detaches) + }) + } +} + +// The deadlock force exists for: the instance is gone but its volume record +// still names it, so the ordinary delete refuses forever. Once the holder is +// proven absent the attachment is cleared and the volume deleted. +func TestVolumeReaperForceDeletesOnceTheHolderIsProvenGone(t *testing.T) { + cases := map[string]func(cluster *fakeCluster){ + "terminated": func(cluster *fakeCluster) { + cluster.reply("ec2.DescribeTerminatedInstances", instancesIn("terminated", "i-gone")) + }, + "not found": func(*fakeCluster) {}, + } + for name, setup := range cases { + t.Run(name, func(t *testing.T) { + cluster := newFakeCluster(t) + volume := newHeldVolume(cluster, "vol-held", "i-gone") + setup(cluster) + + reaper := newVolumeReaper(cluster.nc, fakeClusterNode) + require.NoError(t, reaper.Delete(testCtx(t), "000000000042", + Resource{Kind: "volume", ID: "vol-held"}, true)) + + deleted, holder, detaches := volume.state() + assert.True(t, deleted) + assert.Empty(t, holder) + assert.Equal(t, 1, detaches) + + var input ec2.DetachVolumeInput + cluster.request(t, forceDetachSubject, &input) + assert.Equal(t, "vol-held", aws.StringValue(input.VolumeId)) + }) + } +} + +// With no holder on record, the refusal is about something else — a snapshot, +// a mapping the delete guard found — and clearing the record cannot help. +func TestVolumeReaperForceLeavesAnUnattachedRefusalAlone(t *testing.T) { cluster := newFakeCluster(t) - cluster.fail("ec2.DeleteVolume", "VolumeInUse") + cluster.reply("ec2.DescribeVolumes", ec2.DescribeVolumesOutput{ + Volumes: []*ec2.Volume{{VolumeId: aws.String("vol-snapped"), State: aws.String("available")}}, + }) + cluster.fail("ec2.DeleteVolume", awserrors.ErrorVolumeInUse) - reaper := &volumeReaper{nc: cluster.nc, expectedNodes: 1} - err := reaper.Delete(testCtx(t), "000000000042", Resource{Kind: "volume", ID: "vol-held"}, true) + reaper := newVolumeReaper(cluster.nc, fakeClusterNode) + err := reaper.Delete(testCtx(t), "000000000042", Resource{Kind: "volume", ID: "vol-snapped"}, true) - // The delete still fails here because the double refuses every attempt; - // what matters is that force reached for the detach at all. require.Error(t, err) - assert.Contains(t, cluster.called(), forceDetachSubject) + assert.Contains(t, err.Error(), awserrors.ErrorVolumeInUse) + assert.NotContains(t, cluster.called(), forceDetachSubject) +} - var input ec2.DetachVolumeInput - cluster.request(t, forceDetachSubject, &input) - assert.Equal(t, "vol-held", aws.StringValue(input.VolumeId)) - assert.True(t, aws.BoolValue(input.Force)) +// End to end through the engine: a forced teardown whose guest is still up +// leaves the volume, names why, and keeps the account TERMINATING. +func TestForcedTeardownKeepsTheAccountWhileALiveGuestHoldsAVolume(t *testing.T) { + cluster := newFakeCluster(t) + volume := newHeldVolume(cluster, "vol-held", "i-live") + cluster.reply("ec2.DescribeInstances", instancesIn("running", "i-live")) + + accounts := newFakeAccounts() + engine := quietEngine(accounts, newVolumeReaper(cluster.nc, fakeClusterNode)) + req := request() + req.Force = true + + result, err := engine.Teardown(testCtx(t), req) + + require.ErrorIs(t, err, ErrResourcesStuck) + require.Equal(t, 1, result.StuckCount()) + var stuck Stuck + for _, stage := range result.Stages { + if len(stage.Stuck) > 0 { + stuck = stage.Stuck[0] + } + } + assert.Equal(t, "vol-held", stuck.Resource.ID) + assert.Contains(t, stuck.Reason, "vol-held") + assert.Contains(t, stuck.Reason, "i-live") + assert.Contains(t, stuck.Reason, "running") + assert.False(t, result.AccountDeleted) + assert.False(t, accounts.deleted) + assert.Equal(t, AccountStatusTerminating, accounts.account.Status) + deleted, holder, _ := volume.state() + assert.False(t, deleted) + assert.Equal(t, "i-live", holder) } // A volume that is already gone is a success. Teardown re-runs after a crash @@ -249,7 +506,7 @@ func TestVolumeReaperTreatsAMissingVolumeAsDeleted(t *testing.T) { cluster := newFakeCluster(t) cluster.fail("ec2.DeleteVolume", "InvalidVolume.NotFound") - reaper := &volumeReaper{nc: cluster.nc, expectedNodes: 1} + reaper := newVolumeReaper(cluster.nc, fakeClusterNode) assert.NoError(t, reaper.Delete(testCtx(t), "000000000042", Resource{Kind: "volume", ID: "vol-gone"}, false)) @@ -552,7 +809,7 @@ func TestAListingFailureIsReported(t *testing.T) { cluster := newFakeCluster(t) cluster.fail("ec2.DescribeVolumes", "InternalError") - _, err := (&volumeReaper{nc: cluster.nc, expectedNodes: 1}).List(testCtx(t), "000000000042") + _, err := (newVolumeReaper(cluster.nc, fakeClusterNode)).List(testCtx(t), "000000000042") assert.Error(t, err) } @@ -700,3 +957,60 @@ func TestEC2ReapersRemoveInterfacesBeforeSubnets(t *testing.T) { require.NotEqual(t, -1, eni, "there is no network-interface reaper") require.Less(t, eni, subnet, "interfaces must be reaped before the subnets holding them") } + +func newVolumeReaper(nc *nats.Conn, nodeIDs ...string) *volumeReaper { + return &volumeReaper{nc: nc, nodeIDs: nodeIDs} +} + +// The holder lives on node-2, which never answers, while node-1 answers every +// fan-out twice. A count of replies reads that as complete and the holder as +// gone; only judging by responder identity sees that node-2 is missing. +func TestVolumeReaperForceRetainsAVolumeWhenOneNodeAnswersTwiceAndTheHolderNodeIsSilent(t *testing.T) { + cluster := newFakeCluster(t) + volume := newHeldVolume(cluster, "vol-held", "i-on-silent-node") + cluster.echoAsNode(t, "ec2.DescribeInstances", fakeClusterNode) + + reaper := newVolumeReaper(cluster.nc, fakeClusterNode, "node-2") + err := reaper.Delete(silentClusterCtx(t), "000000000042", Resource{Kind: "volume", ID: "vol-held"}, true) + + require.Error(t, err) + assert.Contains(t, err.Error(), "vol-held") + assert.Contains(t, err.Error(), "i-on-silent-node") + deleted, holder, detaches := volume.state() + assert.False(t, deleted, "a holder on a silent node is not proven gone") + assert.Equal(t, "i-on-silent-node", holder) + assert.Zero(t, detaches) + assert.NotContains(t, cluster.called(), forceDetachSubject) +} + +func TestForcedTeardownKeepsTheAccountWhenOneNodeAnswersTwiceAndTheHolderNodeIsSilent(t *testing.T) { + cluster := newFakeCluster(t) + volume := newHeldVolume(cluster, "vol-held", "i-on-silent-node") + cluster.echoAsNode(t, "ec2.DescribeInstances", fakeClusterNode) + + accounts := newFakeAccounts() + engine := quietEngine(accounts, newVolumeReaper(cluster.nc, fakeClusterNode, "node-2")) + req := request() + req.Force = true + + result, err := engine.Teardown(silentClusterCtx(t), req) + + require.ErrorIs(t, err, ErrResourcesStuck) + require.Equal(t, 1, result.StuckCount()) + var stuck Stuck + for _, stage := range result.Stages { + if len(stage.Stuck) > 0 { + stuck = stage.Stuck[0] + } + } + assert.Equal(t, "vol-held", stuck.Resource.ID) + assert.Contains(t, stuck.Reason, "vol-held") + assert.Contains(t, stuck.Reason, "i-on-silent-node") + assert.False(t, result.AccountDeleted) + assert.False(t, accounts.deleted) + assert.Equal(t, AccountStatusTerminating, accounts.account.Status) + deleted, holder, detaches := volume.state() + assert.False(t, deleted) + assert.Equal(t, "i-on-silent-node", holder) + assert.Zero(t, detaches) +} diff --git a/spinifex/gateway/ec2.go b/spinifex/gateway/ec2.go index fa742ebaa..feec96391 100644 --- a/spinifex/gateway/ec2.go +++ b/spinifex/gateway/ec2.go @@ -350,7 +350,7 @@ var ec2Actions = map[string]ec2Action{ return gateway_ec2_volume.CreateVolume(ctx, input, gw.NATSConn, accountID) }), "DeleteVolume": ec2Handler(func(ctx context.Context, input *ec2.DeleteVolumeInput, gw *GatewayConfig, accountID string) (any, error) { - return gateway_ec2_volume.DeleteVolume(ctx, input, gw.NATSConn, gw.DiscoverActiveNodes(ctx), accountID) + return gateway_ec2_volume.DeleteVolume(ctx, input, gw.NATSConn, gw.NodeIDs, accountID) }), "AttachVolume": ec2Handler(func(ctx context.Context, input *ec2.AttachVolumeInput, gw *GatewayConfig, accountID string) (any, error) { return gateway_ec2_volume.AttachVolume(ctx, input, gw.NATSConn, accountID) diff --git a/spinifex/gateway/ec2/instance/DescribeInstances.go b/spinifex/gateway/ec2/instance/DescribeInstances.go index f075881ac..3cfc8d30c 100644 --- a/spinifex/gateway/ec2/instance/DescribeInstances.go +++ b/spinifex/gateway/ec2/instance/DescribeInstances.go @@ -170,8 +170,8 @@ func missingInstanceIDs(reservations []*ec2.Reservation, instanceIDs []*string) // every expected node and both instance buckets; a partial sweep — a node down, // a timed-out fan-out, or a failed bucket query — returns complete=false so the // caller can decline to act rather than act on a short count. The quota vCPU -// sweep reads the instance record space instead of this; the retype gate and the -// volume delete guard are what still call it. +// sweep reads the instance record space instead of this; the retype gate is +// what still calls it. func DescribeInstancesForReconcile(ctx context.Context, input *ec2.DescribeInstancesInput, natsConn *nats.Conn, expectedNodes int, accountID string) (reservations []*ec2.Reservation, complete bool, err error) { reservations, complete, _, err = gatherInstances(ctx, input, natsConn, expectedNodes, nil, accountID) return reservations, complete, err diff --git a/spinifex/gateway/ec2/volume/DeleteVolume.go b/spinifex/gateway/ec2/volume/DeleteVolume.go index 282bcf1a6..9a1adefa6 100644 --- a/spinifex/gateway/ec2/volume/DeleteVolume.go +++ b/spinifex/gateway/ec2/volume/DeleteVolume.go @@ -61,8 +61,10 @@ func volumeHeldByInstance(reservations []*ec2.Reservation, volumeID string) stri // BlockDeviceMappings: the daemon-side gate trusts persisted volume metadata, // which the attach/detach lifecycle can leave desynced (a running root with // State:"" / AttachedInstance:""), so a drifted-but-live volume could be -// deleted under a running QEMU. The check fails closed on an incomplete view. -func DeleteVolume(ctx context.Context, input *ec2.DeleteVolumeInput, natsConn *nats.Conn, expectedNodes int, accountID string) (ec2.DeleteVolumeOutput, error) { +// deleted under a running QEMU. The check fails closed unless every node in +// nodeIDs answered by identity, so a duplicate reply cannot stand in for a +// silent node. +func DeleteVolume(ctx context.Context, input *ec2.DeleteVolumeInput, natsConn *nats.Conn, nodeIDs []string, accountID string) (ec2.DeleteVolumeOutput, error) { var output ec2.DeleteVolumeOutput err := ValidateDeleteVolumeInput(input) @@ -72,20 +74,20 @@ func DeleteVolume(ctx context.Context, input *ec2.DeleteVolumeInput, natsConn *n volumeID := aws.StringValue(input.VolumeId) - // Strict fan-out: complete is true only when every active node answered and - // both instance buckets queried, so a node that owns the attachment cannot be - // silently missing from the survey. Refuse rather than delete against a - // partial view of the cluster. - reservations, complete, err := gateway_ec2_instance.DescribeInstancesForReconcile(ctx, &ec2.DescribeInstancesInput{}, natsConn, expectedNodes, accountID) + // Every configured node and both instance buckets must answer, judged by + // responder identity, or the node owning the attachment could be the one + // missing. Refuse rather than delete against a partial view of the cluster. + listing, err := gateway_ec2_instance.DescribeAccountInstancesComplete(ctx, natsConn, nodeIDs, accountID) if err != nil { - slog.ErrorContext(ctx, "DeleteVolume: in-use precheck fan-out failed", "volumeId", volumeID, "err", err) + slog.ErrorContext(ctx, "DeleteVolume: refusing delete without a complete instance view", "volumeId", volumeID, "err", err) + // An unanswered node is retryable once the cluster is whole again, so + // keep the ServiceUnavailable and its Retry-After; anything else is ours. + if code, ok := awserrors.ResolveErrorCode(err); ok && code == awserrors.ErrorServiceUnavailable { + return output, err + } return output, errors.New(awserrors.ErrorServerInternal) } - if !complete { - slog.ErrorContext(ctx, "DeleteVolume: refusing delete on incomplete instance view", "volumeId", volumeID) - return output, errors.New(awserrors.ErrorServerInternal) - } - if holder := volumeHeldByInstance(reservations, volumeID); holder != "" { + if holder := volumeHeldByInstance(listing.Reservations, volumeID); holder != "" { slog.ErrorContext(ctx, "DeleteVolume: volume still attached to a live instance", "volumeId", volumeID, "instanceId", holder) return output, errors.New(awserrors.ErrorVolumeInUse) diff --git a/spinifex/gateway/ec2/volume/DeleteVolume_test.go b/spinifex/gateway/ec2/volume/DeleteVolume_test.go index add54b471..33a4fa85e 100644 --- a/spinifex/gateway/ec2/volume/DeleteVolume_test.go +++ b/spinifex/gateway/ec2/volume/DeleteVolume_test.go @@ -1,12 +1,19 @@ package gateway_ec2_volume import ( + "context" + "sync/atomic" "testing" + "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/testutil" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestValidateDeleteVolumeInput(t *testing.T) { @@ -174,3 +181,48 @@ func TestVolumeHeldByInstance(t *testing.T) { }) } } + +// replyAsNode answers every request on subject with data, attributed to node. +func replyAsNode(t *testing.T, nc *nats.Conn, subject, node string, data []byte) { + t.Helper() + sub, err := nc.Subscribe(subject, func(msg *nats.Msg) { + out := nats.NewMsg(msg.Reply) + out.Data = data + if node != "" { + out.Header.Set(utils.NodeIDHeader, node) + } + _ = msg.RespondMsg(out) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) +} + +// node-2 never answers and node-1 answers twice. Counting replies reads that +// as every node heard from, so a volume a guest on node-2 maps would be +// destroyed; the guard must see node-2 is missing and refuse. +func TestDeleteVolume_GuardRefusesWhenOneNodeAnswersTwiceAndAnotherIsSilent(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + replyAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`)) + replyAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`)) + replyAsNode(t, nc, "ec2.DescribeStoppedInstances", "", []byte(`{}`)) + replyAsNode(t, nc, "ec2.DescribeTerminatedInstances", "", []byte(`{}`)) + var deletes atomic.Int32 + sub, err := nc.Subscribe("ec2.DeleteVolume", func(msg *nats.Msg) { + deletes.Add(1) + _ = msg.Respond([]byte(`{}`)) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + require.NoError(t, nc.Flush()) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + _, err = DeleteVolume(ctx, &ec2.DeleteVolumeInput{VolumeId: aws.String("vol-0123456789abcdef0")}, + nc, []string{"node-1", "node-2"}, "000000000042") + + require.Error(t, err) + code, ok := awserrors.ResolveErrorCode(err) + require.True(t, ok) + assert.Equal(t, awserrors.ErrorServiceUnavailable, code, "an unanswered node is retryable, not an internal fault") + assert.Zero(t, deletes.Load(), "the volume service must not be asked to destroy the backing") +} diff --git a/spinifex/gateway/ec2/volume/handler_test.go b/spinifex/gateway/ec2/volume/handler_test.go index 45e8970b5..4070e7a20 100644 --- a/spinifex/gateway/ec2/volume/handler_test.go +++ b/spinifex/gateway/ec2/volume/handler_test.go @@ -29,17 +29,17 @@ func TestCreateVolume_NilNATS(t *testing.T) { } func TestDeleteVolume_ValidationErrors(t *testing.T) { - _, err := DeleteVolume(context.Background(), nil, nil, 1, "") + _, err := DeleteVolume(context.Background(), nil, nil, []string{"node-1"}, "") assert.EqualError(t, err, awserrors.ErrorInvalidParameterValue) - _, err = DeleteVolume(context.Background(), &ec2.DeleteVolumeInput{VolumeId: aws.String("bad")}, nil, 1, "") + _, err = DeleteVolume(context.Background(), &ec2.DeleteVolumeInput{VolumeId: aws.String("bad")}, nil, []string{"node-1"}, "") assert.EqualError(t, err, awserrors.ErrorInvalidVolumeIDMalformed) } func TestDeleteVolume_NilNATS(t *testing.T) { _, err := DeleteVolume(context.Background(), &ec2.DeleteVolumeInput{ VolumeId: aws.String("vol-1234567890abcdef0"), - }, nil, 1, "acct-123") + }, nil, []string{"node-1"}, "acct-123") assert.Error(t, err) } diff --git a/spinifex/gateway/ec2_deletevolume_guard_test.go b/spinifex/gateway/ec2_deletevolume_guard_test.go new file mode 100644 index 000000000..4649cda4a --- /dev/null +++ b/spinifex/gateway/ec2_deletevolume_guard_test.go @@ -0,0 +1,70 @@ +//test:in-package — drives the public DeleteVolume action through EC2_Request +// with the gateway's unexported test helpers. + +package gateway + +import ( + "context" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/awserrors" + "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" +) + +// A configured node that never answers leaves the instance view incomplete, so +// a guest there could still map the volume. The public action must refuse with +// a retryable ServiceUnavailable and never reach the volume service. +func TestDeleteVolumeAction_MissingConfiguredNodeIsRetryableAndDeletesNothing(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + answer := func(subject, node string) { + sub, err := nc.Subscribe(subject, func(msg *nats.Msg) { + out := nats.NewMsg(msg.Reply) + out.Data = []byte(`{}`) + if node != "" { + out.Header.Set(utils.NodeIDHeader, node) + } + _ = msg.RespondMsg(out) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + } + answer("ec2.DescribeInstances", "node-1") + answer("ec2.DescribeStoppedInstances", "") + answer("ec2.DescribeTerminatedInstances", "") + var volumeCalls atomic.Int32 + for _, subject := range []string{"ec2.DeleteVolume", "ec2.ForceDetachVolume"} { + sub, err := nc.Subscribe(subject, func(msg *nats.Msg) { + volumeCalls.Add(1) + _ = msg.Respond([]byte(`{}`)) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + } + require.NoError(t, nc.Flush()) + + gw := &GatewayConfig{ + DisableLogging: true, + NATSConn: nc, + NodeIDs: []string{"node-1", "node-2"}, + IAMService: allowAllIAMService(), + } + req := setupEC2Request("Action=DeleteVolume&VolumeId=vol-0123456789abcdef0", "123456789012") + ctx, cancel := context.WithTimeout(req.Context(), 30*time.Second) + defer cancel() + err := gw.EC2_Request(httptest.NewRecorder(), req.WithContext(ctx)) + + require.Error(t, err) + code, ok := awserrors.ResolveErrorCode(err) + require.True(t, ok) + assert.Equal(t, awserrors.ErrorServiceUnavailable, code) + _, retryable := awserrors.ResolveRetryAfter(err) + assert.True(t, retryable, "the refusal must tell clients when to retry") + assert.Zero(t, volumeCalls.Load(), "the volume must not be detached or deleted") +}