mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
fix(eks): make CP-VPC teardown converge, GC orphaned OVN topology (#510)
A re-driven DeleteCluster whose managed control-plane VPC was already gone (EC2 InvalidVpcID.NotFound) kept looping DependencyViolation: the internal cp-vpc state record (ClusterMeta.ManagedCPVPC) was never cleared unless the *entire* purge succeeded, so any unrelated flaky step retried CP-VPC/SG API calls against a stale VpcId forever instead of converging. purgeClusterInfra now clears ManagedCPVPC (via a new CAS-based ClearClusterManagedCPVPC) as soon as DeleteClusterCPVPC + its SG cleanup both succeed, independent of whatever else in that run still fails. DeleteClusterCPVPC also now republishes vpc.delete for the CP VPC on every call — including the already-gone fast path, falling back to the persisted VpcId when the live tag-indexed describe can no longer name it — so vpcd's topology manager gets another chance to GC an orphaned logical router/subnets/SubnetEgressPriorityDrop policies if the original delete's fire-and-forget event was ever lost.
This commit is contained in:
@@ -300,6 +300,26 @@ func SetClusterHealthState(kv nats.KeyValue, name, issue string, nodeCount int)
|
||||
})
|
||||
}
|
||||
|
||||
// ClearClusterManagedCPVPC clears meta.ManagedCPVPC once the managed CP VPC's
|
||||
// EC2 teardown has converged (including converging from an already-gone VPC).
|
||||
// purgeClusterInfra calls it immediately after a successful DeleteClusterCPVPC
|
||||
// so a re-drive of a wedged DELETING — with some unrelated step still failing —
|
||||
// never retries EC2/OVN calls against a stale VpcId, which is what turns a
|
||||
// single DependencyViolation into a permanent teardown loop. No-op if already
|
||||
// cleared. Returns ErrClusterNotFound if the cluster was deleted concurrently.
|
||||
func ClearClusterManagedCPVPC(kv nats.KeyValue, name string) error {
|
||||
if name == "" {
|
||||
return errors.New("eks: ClearClusterManagedCPVPC empty name")
|
||||
}
|
||||
return casUpdateMeta(kv, name, func(m *ClusterMeta) bool {
|
||||
if m.ManagedCPVPC == nil {
|
||||
return false
|
||||
}
|
||||
m.ManagedCPVPC = nil
|
||||
return true
|
||||
})
|
||||
}
|
||||
|
||||
// SwapControlPlaneMember atomically replaces a lost control-plane member
|
||||
// (deadInstanceID) with a freshly provisioned one in meta.ControlPlaneNodes and
|
||||
// refreshes the scalar [0] mirrors if the primary changed. A no-op (no error)
|
||||
|
||||
@@ -11,6 +11,8 @@ import (
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
"github.com/mulgadc/spinifex/spinifex/awserrors"
|
||||
"github.com/mulgadc/spinifex/spinifex/tags"
|
||||
"github.com/mulgadc/spinifex/spinifex/utils"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// The managed control-plane VPC ("Set B") is the spinifex analogue of AWS EKS's
|
||||
@@ -105,17 +107,21 @@ type CPVPCDeps struct {
|
||||
RT routeTableProvisioner
|
||||
NGW natGatewayProvisioner
|
||||
EIP eipProvisioner
|
||||
// NATSConn republishes vpc.delete for OVN topology GC (see
|
||||
// gcClusterCPVPCTopology). Nil is tolerated (GC is skipped, never blocking).
|
||||
NATSConn *nats.Conn
|
||||
}
|
||||
|
||||
// cpVPCDeps adapts the service's EC2-family deps onto CPVPCDeps for the managed
|
||||
// control-plane VPC build + teardown.
|
||||
func (s *EKSServiceImpl) cpVPCDeps() CPVPCDeps {
|
||||
return CPVPCDeps{
|
||||
VPC: s.deps.VPCMgr,
|
||||
IGW: s.deps.IGW,
|
||||
RT: s.deps.RouteTable,
|
||||
NGW: s.deps.NATGW,
|
||||
EIP: s.deps.EIP,
|
||||
VPC: s.deps.VPCMgr,
|
||||
IGW: s.deps.IGW,
|
||||
RT: s.deps.RouteTable,
|
||||
NGW: s.deps.NATGW,
|
||||
EIP: s.deps.EIP,
|
||||
NATSConn: s.deps.NATSConn,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -423,7 +429,13 @@ func ensureCPNatGateway(ctx context.Context, deps CPVPCDeps, accountID, clusterN
|
||||
// clean delete. Best-effort: every step is attempted and failures are logged;
|
||||
// the final VPC delete failure is returned so the caller can surface a leak.
|
||||
// Safe to call when nothing was provisioned (all describes return empty).
|
||||
func DeleteClusterCPVPC(ctx context.Context, deps CPVPCDeps, accountID, clusterName string) error {
|
||||
//
|
||||
// knownRefs is the caller's last-persisted cp-vpc state record (ClusterMeta's
|
||||
// ManagedCPVPC), used only as an OVN-GC target fallback when the tag-indexed
|
||||
// EC2 VPC is already gone — the live describe can no longer name it, but the
|
||||
// record still can. Pass nil when no such record exists (e.g. the billable
|
||||
// reaper, which acts once the cluster meta itself is gone).
|
||||
func DeleteClusterCPVPC(ctx context.Context, deps CPVPCDeps, accountID, clusterName string, knownRefs *ManagedCPVPC) error {
|
||||
if clusterName == "" {
|
||||
return errors.New("eks: DeleteClusterCPVPC empty cluster name")
|
||||
}
|
||||
@@ -509,7 +521,15 @@ func DeleteClusterCPVPC(ctx context.Context, deps CPVPCDeps, accountID, clusterN
|
||||
if err != nil {
|
||||
return fmt.Errorf("eks: describe cp vpc for delete (%s): %w", clusterName, err)
|
||||
}
|
||||
|
||||
// The tag-indexed VPC is already gone — a prior re-drive completed the EC2
|
||||
// delete, or it was reclaimed out-of-band. Idempotent success: nothing left
|
||||
// to delete here. Its OVN logical router/subnets/egress-drop policies may
|
||||
// still be orphaned if that prior delete's fire-and-forget vpc.delete event
|
||||
// never reached vpcd (e.g. NATS was down), so GC still runs below against
|
||||
// knownRefs — the only surviving reference to its identity.
|
||||
if vpcOut == nil || len(vpcOut.Vpcs) == 0 {
|
||||
gcClusterCPVPCTopology(ctx, deps.NATSConn, clusterName, cpVPCGCTarget(knownRefs))
|
||||
return nil
|
||||
}
|
||||
vpcID := aws.StringValue(vpcOut.Vpcs[0].VpcId)
|
||||
@@ -520,6 +540,38 @@ func DeleteClusterCPVPC(ctx context.Context, deps CPVPCDeps, accountID, clusterN
|
||||
if _, err := deps.VPC.DeleteVpc(ctx, &ec2.DeleteVpcInput{VpcId: aws.String(vpcID)}, accountID); err != nil && !awserrors.IsNotFound(err) {
|
||||
return fmt.Errorf("eks: delete cp vpc %s: %w", vpcID, err)
|
||||
}
|
||||
gcClusterCPVPCTopology(ctx, deps.NATSConn, clusterName, vpcID)
|
||||
slog.InfoContext(ctx, "DeleteClusterCPVPC: managed control-plane VPC removed", "cluster", clusterName, "vpc", vpcID)
|
||||
return nil
|
||||
}
|
||||
|
||||
// cpVPCGCTarget resolves the best-known VpcId for OVN GC when the tag-indexed
|
||||
// EC2 record is already gone: the persisted (internal cp-vpc state record)
|
||||
// VpcId, or "" if none was ever recorded (nil knownRefs).
|
||||
func cpVPCGCTarget(knownRefs *ManagedCPVPC) string {
|
||||
if knownRefs == nil {
|
||||
return ""
|
||||
}
|
||||
return knownRefs.VpcId
|
||||
}
|
||||
|
||||
// gcClusterCPVPCTopology republishes vpc.delete for vpcID so vpcd's topology
|
||||
// manager gets another chance to remove the CP VPC's OVN logical router,
|
||||
// subnets, DHCP options, and SubnetEgressPriorityDrop policies (owned by the
|
||||
// router row, so OVSDB cascades their removal with it). That cleanup normally
|
||||
// runs only as a side effect of a *live* DeleteVpc KV mutation — a re-drive
|
||||
// against an already-gone VPC does not mutate the KV again, so a lost or
|
||||
// never-delivered event would otherwise orphan the OVN state forever.
|
||||
// Best-effort and idempotent: a nil conn or empty vpcID is a no-op, and vpcd
|
||||
// tolerates re-deleting an already-absent router.
|
||||
func gcClusterCPVPCTopology(ctx context.Context, nc *nats.Conn, clusterName, vpcID string) {
|
||||
if nc == nil || vpcID == "" {
|
||||
return
|
||||
}
|
||||
utils.PublishEvent(nc, "vpc.delete", struct {
|
||||
VpcId string `json:"vpc_id"`
|
||||
CidrBlock string `json:"cidr_block"`
|
||||
VNI int64 `json:"vni"`
|
||||
}{VpcId: vpcID})
|
||||
slog.InfoContext(ctx, "DeleteClusterCPVPC: republished vpc.delete for OVN GC", "cluster", clusterName, "vpc", vpcID)
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ package handlers_eks
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -190,7 +191,7 @@ func TestRLC5_DeleteClusterCPVPCReleasesNATGWEIPAfterRoutes(t *testing.T) {
|
||||
tags: map[string]string{clusterEKSClusterTagKey: "alpha", clusterEKSRoleTagKey: clusterEKSRolePrivateRT},
|
||||
}}
|
||||
|
||||
require.NoError(t, DeleteClusterCPVPC(context.Background(), f.svc.cpVPCDeps(), testAccountID, "alpha"))
|
||||
require.NoError(t, DeleteClusterCPVPC(context.Background(), f.svc.cpVPCDeps(), testAccountID, "alpha", nil))
|
||||
|
||||
require.Len(t, f.ngw.gws, 1)
|
||||
assert.Equal(t, "deleted", f.ngw.gws[0].state, "NAT GW must delete after routes are cleared (rule #3 guard not tripped)")
|
||||
@@ -216,7 +217,7 @@ func TestRLC1_DeleteClusterCPVPCToleratesAbsentVPC(t *testing.T) {
|
||||
f.vpcMgr.vpcs = cpVPC
|
||||
f.vpcMgr.deleteVpcErr = errors.New(awserrors.ErrorInvalidVpcIDNotFound)
|
||||
|
||||
require.NoErrorf(t, DeleteClusterCPVPC(context.Background(), f.svc.cpVPCDeps(), testAccountID, "alpha"),
|
||||
require.NoErrorf(t, DeleteClusterCPVPC(context.Background(), f.svc.cpVPCDeps(), testAccountID, "alpha", nil),
|
||||
"RLC rule #1: teardown must tolerate DeleteVpc NotFound (resource already gone), not abort")
|
||||
})
|
||||
|
||||
@@ -225,7 +226,7 @@ func TestRLC1_DeleteClusterCPVPCToleratesAbsentVPC(t *testing.T) {
|
||||
f.vpcMgr.vpcs = cpVPC
|
||||
f.vpcMgr.deleteVpcErr = errors.New(awserrors.ErrorDependencyViolation)
|
||||
|
||||
err := DeleteClusterCPVPC(context.Background(), f.svc.cpVPCDeps(), testAccountID, "alpha")
|
||||
err := DeleteClusterCPVPC(context.Background(), f.svc.cpVPCDeps(), testAccountID, "alpha", nil)
|
||||
require.Error(t, err, "a non-NotFound DeleteVpc failure must surface so the teardown backstop retries")
|
||||
assert.Contains(t, err.Error(), awserrors.ErrorDependencyViolation)
|
||||
})
|
||||
@@ -399,3 +400,114 @@ func TestDeleteCluster_ManagedCPVPCReclaimsCustomerVPCSGs(t *testing.T) {
|
||||
assert.Contains(t, deleted, "sg-cust-cp", "customer-VPC control-plane SG must be reclaimed on teardown")
|
||||
assert.Contains(t, deleted, "sg-cust-ng", "customer-VPC nodegroup SG must be reclaimed on teardown")
|
||||
}
|
||||
|
||||
// TestDeleteCluster_ManagedCPVPCAlreadyGoneClearsStateRecord locks the
|
||||
// mulga-jbb2i teardown-idempotency fix: when the managed CP VPC is already
|
||||
// gone (its tag-indexed EC2 record is absent — the "InvalidVpcID.NotFound"
|
||||
// case confirmed live), DeleteClusterCPVPC converges to success, and the
|
||||
// internal cp-vpc state record (meta.ManagedCPVPC) must be cleared right away —
|
||||
// even while an unrelated step (here, the NLB delete) keeps failing and the
|
||||
// overall teardown still surfaces an error. Without this, every re-drive keeps
|
||||
// retrying SG/VPC API calls against the stale VpcId, which is what turns a
|
||||
// single failure into the reaper's permanent "DependencyViolation" loop. A
|
||||
// later re-drive, once the unrelated failure clears, must then fully converge.
|
||||
func TestDeleteCluster_ManagedCPVPCAlreadyGoneClearsStateRecord(t *testing.T) {
|
||||
f := newDeleteClusterFixture(t, "alpha")
|
||||
|
||||
meta, err := GetClusterMeta(f.kv, "alpha")
|
||||
require.NoError(t, err)
|
||||
meta.ManagedCPVPC = &ManagedCPVPC{VpcId: "vpc-cp-gone", PublicSubnetId: "subnet-cp-pub"}
|
||||
require.NoError(t, PutClusterMeta(f.kv, meta))
|
||||
|
||||
// The CP VPC's own EC2 record is already gone: f.vpcMgr has nothing tagged
|
||||
// for "alpha", so DeleteClusterCPVPC's describe-by-tag finds zero VPCs and
|
||||
// converges to success. An unrelated NLB failure must still surface.
|
||||
lbName := ClusterNLBName("alpha")
|
||||
f.nlb.lbByName[lbName] = &elbv2.LoadBalancer{
|
||||
LoadBalancerArn: aws.String("arn:lb"),
|
||||
LoadBalancerName: aws.String(lbName),
|
||||
}
|
||||
f.nlb.deleteLBErr = errors.New("elbv2 unavailable")
|
||||
|
||||
_, err = f.svc.DeleteCluster(context.Background(), deleteInput("alpha"), testAccountID)
|
||||
require.Error(t, err, "the unrelated NLB failure must still surface")
|
||||
|
||||
stuck, getErr := GetClusterMeta(f.kv, "alpha")
|
||||
require.NoError(t, getErr, "meta must survive for retry")
|
||||
assert.Equal(t, ClusterStatusDeleting, stuck.Status)
|
||||
assert.Nil(t, stuck.ManagedCPVPC,
|
||||
"the internal cp-vpc state record must be cleared once its VPC converges to deleted, independent of the NLB failure")
|
||||
|
||||
// Re-drive: the NLB failure clears, teardown must now fully converge —
|
||||
// and must NOT re-attempt any CP-VPC/SG calls against the stale VpcId,
|
||||
// since the state record is already cleared.
|
||||
f.nlb.deleteLBErr = nil
|
||||
_, err = f.svc.DeleteCluster(context.Background(), deleteInput("alpha"), testAccountID)
|
||||
require.NoError(t, err, "the re-drive must converge to fully deleted")
|
||||
|
||||
_, getErr = GetClusterMeta(f.kv, "alpha")
|
||||
assert.ErrorIs(t, getErr, ErrClusterNotFound, "meta must be swept once teardown converges")
|
||||
}
|
||||
|
||||
// subscribeVPCDelete subscribes to the vpc.delete OVN-GC signal and returns a
|
||||
// getter that blocks for (and unmarshals) the next message's vpc_id.
|
||||
func subscribeVPCDelete(t *testing.T, nc *nats.Conn) func() string {
|
||||
t.Helper()
|
||||
sub, err := nc.SubscribeSync("vpc.delete")
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
return func() string {
|
||||
t.Helper()
|
||||
msg, err := sub.NextMsg(2 * time.Second)
|
||||
require.NoError(t, err, "expected a vpc.delete OVN-GC event")
|
||||
var evt struct {
|
||||
VpcId string `json:"vpc_id"`
|
||||
}
|
||||
require.NoError(t, json.Unmarshal(msg.Data, &evt))
|
||||
return evt.VpcId
|
||||
}
|
||||
}
|
||||
|
||||
// TestDeleteClusterCPVPC_OVNGCSelection locks the OVN-GC half of the
|
||||
// mulga-jbb2i fix: DeleteClusterCPVPC must republish vpc.delete for the CP
|
||||
// VPC so vpcd's topology manager gets another chance to remove the orphaned
|
||||
// logical router/subnets/egress-drop policies — both when the EC2 VPC is
|
||||
// actually deleted this call, and when it was already gone, where the only
|
||||
// surviving reference to its identity is the caller's persisted ManagedCPVPC
|
||||
// record (knownRefs). With no known record at all, there is nothing to GC.
|
||||
func TestDeleteClusterCPVPC_OVNGCSelection(t *testing.T) {
|
||||
t.Run("VPC deleted this call: GC targets the live vpc id", func(t *testing.T) {
|
||||
f := newEKSServiceFixture(t)
|
||||
f.vpcMgr.vpcs = []*fakeCPVPC{{
|
||||
id: "vpc-cp-live",
|
||||
cidr: "10.0.0.0/16",
|
||||
tags: map[string]string{clusterEKSClusterTagKey: "alpha", clusterEKSRoleTagKey: clusterEKSRoleCPVPC},
|
||||
}}
|
||||
nextVPCID := subscribeVPCDelete(t, f.svc.deps.NATSConn)
|
||||
|
||||
require.NoError(t, DeleteClusterCPVPC(context.Background(), f.svc.cpVPCDeps(), testAccountID, "alpha", nil))
|
||||
assert.Equal(t, "vpc-cp-live", nextVPCID())
|
||||
})
|
||||
|
||||
t.Run("VPC already gone: GC falls back to the persisted state record", func(t *testing.T) {
|
||||
f := newEKSServiceFixture(t) // vpcMgr has nothing tagged — already gone
|
||||
nextVPCID := subscribeVPCDelete(t, f.svc.deps.NATSConn)
|
||||
|
||||
knownRefs := &ManagedCPVPC{VpcId: "vpc-cp-orphaned"}
|
||||
require.NoError(t, DeleteClusterCPVPC(context.Background(), f.svc.cpVPCDeps(), testAccountID, "alpha", knownRefs))
|
||||
assert.Equal(t, "vpc-cp-orphaned", nextVPCID())
|
||||
})
|
||||
|
||||
t.Run("VPC already gone, no known record: nothing to GC, no event published", func(t *testing.T) {
|
||||
f := newEKSServiceFixture(t)
|
||||
sub, err := f.svc.deps.NATSConn.SubscribeSync("vpc.delete")
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
|
||||
require.NoError(t, DeleteClusterCPVPC(context.Background(), f.svc.cpVPCDeps(), testAccountID, "alpha", nil))
|
||||
require.NoError(t, f.svc.deps.NATSConn.Flush())
|
||||
|
||||
_, err = sub.NextMsg(50 * time.Millisecond)
|
||||
assert.ErrorIs(t, err, nats.ErrTimeout, "no known vpc id: nothing to GC, must not publish")
|
||||
})
|
||||
}
|
||||
|
||||
@@ -103,7 +103,7 @@ func (r *EKSBillableReaper) reclaimOrphanInfra(ctx context.Context, account, clu
|
||||
if err := DeleteClusterNLB(ctx, r.svc.deps.NLB, account, clusterName); err != nil {
|
||||
slog.WarnContext(ctx, "eks-billable: reclaim orphan NLB failed", "cluster", clusterName, "err", err)
|
||||
}
|
||||
if err := DeleteClusterCPVPC(ctx, r.svc.cpVPCDeps(), account, clusterName); err != nil {
|
||||
if err := DeleteClusterCPVPC(ctx, r.svc.cpVPCDeps(), account, clusterName, nil); err != nil {
|
||||
slog.WarnContext(ctx, "eks-billable: reclaim orphan CP VPC (NAT-GW EIP) failed", "cluster", clusterName, "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1185,8 +1185,9 @@ func (s *EKSServiceImpl) purgeClusterInfra(ctx context.Context, accountID, name
|
||||
// New topology: SGs + IGW + subnets + route tables + NAT GW + VPC all live
|
||||
// in the managed CP VPC under the system account. DeleteClusterCPVPC removes
|
||||
// the IGW + VPC after the SGs, so SG cleanup must precede it.
|
||||
if err := DeleteClusterSGs(ctx, s.deps.VPCSG, infraAcct, name, meta.ManagedCPVPC.VpcId); err != nil {
|
||||
teardownErrs = append(teardownErrs, fmt.Errorf("delete cluster SGs: %w", err))
|
||||
cpSGErr := DeleteClusterSGs(ctx, s.deps.VPCSG, infraAcct, name, meta.ManagedCPVPC.VpcId)
|
||||
if cpSGErr != nil {
|
||||
teardownErrs = append(teardownErrs, fmt.Errorf("delete cluster SGs: %w", cpSGErr))
|
||||
}
|
||||
// launchNodegroupInfra also creates the cluster SGs in the customer VPC for
|
||||
// worker<->control-plane networking; reclaim them here too, or they orphan
|
||||
@@ -1196,8 +1197,21 @@ func (s *EKSServiceImpl) purgeClusterInfra(ctx context.Context, accountID, name
|
||||
teardownErrs = append(teardownErrs, fmt.Errorf("delete customer-VPC cluster SGs: %w", err))
|
||||
}
|
||||
}
|
||||
if err := DeleteClusterCPVPC(ctx, s.cpVPCDeps(), infraAcct, name); err != nil {
|
||||
teardownErrs = append(teardownErrs, fmt.Errorf("delete managed CP VPC: %w", err))
|
||||
cpVPCErr := DeleteClusterCPVPC(ctx, s.cpVPCDeps(), infraAcct, name, meta.ManagedCPVPC)
|
||||
if cpVPCErr != nil {
|
||||
teardownErrs = append(teardownErrs, fmt.Errorf("delete managed CP VPC: %w", cpVPCErr))
|
||||
}
|
||||
if cpSGErr == nil && cpVPCErr == nil {
|
||||
// The CP VPC (including when already gone — tolerated as success)
|
||||
// and its own SGs are both confirmed torn down. Clear the internal
|
||||
// cp-vpc state record right now, independent of whatever else in
|
||||
// this run still fails below: otherwise a re-drive keeps retrying
|
||||
// SG/VPC API calls against a stale VpcId forever, turning a single
|
||||
// DependencyViolation into a permanent DELETING loop.
|
||||
meta.ManagedCPVPC = nil
|
||||
if perr := ClearClusterManagedCPVPC(acctKV, name); perr != nil && !errors.Is(perr, ErrClusterNotFound) {
|
||||
teardownErrs = append(teardownErrs, fmt.Errorf("clear managed CP VPC state: %w", perr))
|
||||
}
|
||||
}
|
||||
} else if meta.ResourcesVpcConfig != nil && meta.ResourcesVpcConfig.VpcId != "" {
|
||||
// Legacy topology: CP lived in the customer VPC; reclaim its SGs + the
|
||||
|
||||
Reference in New Issue
Block a user