From 609b4121f0fd72a0bc4d773f4cc43041c963c0ca Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Tue, 1 Sep 2026 11:25:35 +1000 Subject: [PATCH] fix: report an IGW attachment only once the OVN gateway is up (#962) --- spinifex/accountteardown/reapers_ec2.go | 4 + spinifex/accountteardown/reapers_ec2_test.go | 18 ++ spinifex/daemon/daemon_handlers_test.go | 51 ++++ spinifex/daemon/daemon_handlers_vpc.go | 45 ++-- spinifex/handlers/bedrock/fakes_test.go | 16 +- spinifex/handlers/ec2/igw/service_impl.go | 121 ++++++++- .../handlers/ec2/igw/service_impl_test.go | 241 +++++++++++++++++- spinifex/handlers/ec2/vpc/service_impl.go | 67 +++++ .../handlers/ec2/vpc/service_impl_test.go | 35 +++ spinifex/handlers/eks/igw_test.go | 70 ++--- spinifex/handlers/rds/launch_test.go | 19 +- spinifex/handlers/systemvpc/igw.go | 19 +- spinifex/handlers/systemvpc/systemvpc_test.go | 101 ++++++-- spinifex/network/external/types.go | 7 + spinifex/network/reconcile/apply.go | 75 ++++-- spinifex/network/reconcile/apply_test.go | 176 ++++++++++++- spinifex/network/reconcile/drift.go | 7 +- spinifex/network/reconcile/intent.go | 2 + spinifex/network/reconcile/intent_test.go | 46 +++- spinifex/network/reconcile/reconcile.go | 7 + spinifex/vpcd/vpcd.go | 8 + 21 files changed, 996 insertions(+), 139 deletions(-) diff --git a/spinifex/accountteardown/reapers_ec2.go b/spinifex/accountteardown/reapers_ec2.go index 4c730bdb4..1aecaa374 100644 --- a/spinifex/accountteardown/reapers_ec2.go +++ b/spinifex/accountteardown/reapers_ec2.go @@ -396,6 +396,10 @@ func (r *igwReaper) Delete(ctx context.Context, accountID string, resource Resou return err } } + // A gateway whose attach is not yet confirmed reports no attachment, so + // Detail is empty and the delete is refused. DeleteVpc rejects while the + // gateway is attached, so the VPC survives and the next sweep, once the + // attachment is confirmed, takes the branch above. _, err := gateway_ec2_igw.DeleteInternetGateway(ctx, &ec2.DeleteInternetGatewayInput{ InternetGatewayId: aws.String(resource.ID), }, r.nc, accountID) diff --git a/spinifex/accountteardown/reapers_ec2_test.go b/spinifex/accountteardown/reapers_ec2_test.go index ada376331..e32754902 100644 --- a/spinifex/accountteardown/reapers_ec2_test.go +++ b/spinifex/accountteardown/reapers_ec2_test.go @@ -13,6 +13,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/testutil" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" @@ -403,6 +404,23 @@ func TestIGWReaperDeletesADetachedGatewayDirectly(t *testing.T) { assert.Equal(t, []string{"ec2.DeleteInternetGateway"}, cluster.called()) } +// A gateway whose attach is not yet confirmed reports no attachment, so the +// listing carries no VPC and the delete is refused. That is not a wedge: the +// VPC survives because DeleteVpc rejects while the gateway is attached, and the +// stage re-sweeps until the confirmation lands. +func TestIGWReaperReportsARefusedDeleteForAnUnconfirmedAttach(t *testing.T) { + cluster := newFakeCluster(t) + cluster.fail("ec2.DeleteInternetGateway", awserrors.ErrorDependencyViolation) + + reaper := &igwReaper{nc: cluster.nc} + err := reaper.Delete(testCtx(t), "000000000042", + Resource{Kind: "internet-gateway", ID: "igw-1"}, false) + + require.Error(t, err) + assert.Equal(t, []string{"ec2.DeleteInternetGateway"}, cluster.called(), + "the reaper must not guess at the VPC: the next sweep sees the confirmed attachment") +} + // A VPC's main route table is deleted with the VPC and cannot be deleted on // its own, so listing it would guarantee a stuck report every time. func TestRouteTableReaperSkipsTheMainTable(t *testing.T) { diff --git a/spinifex/daemon/daemon_handlers_test.go b/spinifex/daemon/daemon_handlers_test.go index 8318aef30..ed2e65593 100644 --- a/spinifex/daemon/daemon_handlers_test.go +++ b/spinifex/daemon/daemon_handlers_test.go @@ -2300,6 +2300,57 @@ func createVPCTestDaemon(t *testing.T) *Daemon { return daemon } +// Attaching an IGW is asynchronous, so nothing confirms the attachment during +// this test. Both regressions it guards were invisible while the default-VPC +// bootstrap derived "does this VPC have a gateway" from the describe +// projection, which reports only confirmed attachments. +func TestEnsureDefaultVPCInfrastructure_PendingAttachIsNotReDone(t *testing.T) { + daemon := createTestDaemon(t, sharedNATSURL) + + // One JetStream for all three services: CreateRoute validates the gateway + // against the IGW bucket, so a separate store would fail the lookup. + _, nc, _ := testutil.StartTestJetStream(t) + testutil.StubVpcdSGResponder(t, nc) + + vpcSvc, err := handlers_ec2_vpc.NewVPCServiceImplWithNATS(t.Context(), daemon.config, nc) + require.NoError(t, err) + daemon.vpcService = vpcSvc + igwSvc, err := handlers_ec2_igw.NewIGWServiceImplWithNATS(t.Context(), daemon.config, nc) + require.NoError(t, err) + daemon.igwService = igwSvc + rtbSvc, err := handlers_ec2_routetable.NewRouteTableServiceImplWithNATS(t.Context(), daemon.config, nc) + require.NoError(t, err) + daemon.routeTableService = rtbSvc + + const accountID = "000000000042" + _, err = vpcSvc.EnsureDefaultVPC(accountID) + require.NoError(t, err) + + daemon.ensureDefaultVPCInfrastructureFor(t.Context(), accountID) + + // The route must be added in the same call: waiting for confirmation would + // leave a new account's default VPC with no egress until a later restart. + rtbs, err := rtbSvc.DescribeRouteTables(t.Context(), &ec2.DescribeRouteTablesInput{}, accountID) + require.NoError(t, err) + require.Len(t, rtbs.RouteTables, 1) + var defaultRoutes int + for _, r := range rtbs.RouteTables[0].Routes { + if aws.StringValue(r.DestinationCidrBlock) == "0.0.0.0/0" { + defaultRoutes++ + assert.NotEmpty(t, aws.StringValue(r.GatewayId)) + } + } + assert.Equal(t, 1, defaultRoutes, "the default VPC must get its 0.0.0.0/0 route even though the attach is unconfirmed") + + // A restart inside the pending window must reuse the gateway, not add one. + daemon.ensureDefaultVPCInfrastructureFor(t.Context(), accountID) + + igws, err := igwSvc.DescribeInternetGateways(t.Context(), &ec2.DescribeInternetGatewaysInput{}, accountID) + require.NoError(t, err) + assert.Len(t, igws.InternetGateways, 1, + "a second gateway was attached to the same VPC: intent.IGWs keys by VPC, so one of them orphans") +} + func TestDelegateHandlers_VPC(t *testing.T) { daemon := createVPCTestDaemon(t) diff --git a/spinifex/daemon/daemon_handlers_vpc.go b/spinifex/daemon/daemon_handlers_vpc.go index 921263b6e..15f08dba2 100644 --- a/spinifex/daemon/daemon_handlers_vpc.go +++ b/spinifex/daemon/daemon_handlers_vpc.go @@ -145,29 +145,22 @@ func (d *Daemon) ensureDefaultVPCInfrastructureFor(ctx context.Context, accountI return } - // Check if IGW already attached - igwOut, err := d.igwService.DescribeInternetGateways(ctx, &ec2.DescribeInternetGatewaysInput{}, accountID) + // Check if IGW already attached. Uses the intent view: an attach requested + // by an earlier pass but not yet confirmed still counts, or this creates a + // second gateway for the same VPC every time it runs inside that window. + existingIGW, err := d.igwService.AttachmentIntent(ctx, accountID, defaultVpcId) if err != nil { - slog.WarnContext(ctx, "DescribeInternetGateways failed for default VPC infrastructure, retrying", + slog.WarnContext(ctx, "IGW lookup failed for default VPC infrastructure, retrying", "accountID", accountID, "err", err) time.Sleep(500 * time.Millisecond) - igwOut, err = d.igwService.DescribeInternetGateways(ctx, &ec2.DescribeInternetGatewaysInput{}, accountID) + existingIGW, err = d.igwService.AttachmentIntent(ctx, accountID, defaultVpcId) if err != nil { - slog.ErrorContext(ctx, "DescribeInternetGateways failed for default VPC infrastructure after retry", + slog.ErrorContext(ctx, "IGW lookup failed for default VPC infrastructure after retry", "accountID", accountID, "err", err) return } } - hasIGW := false - for _, igw := range igwOut.InternetGateways { - for _, att := range igw.Attachments { - if att.VpcId != nil && *att.VpcId == defaultVpcId { - hasIGW = true - break - } - } - } - if !hasIGW { + if existingIGW == nil { // Create and attach an IGW — use bootstrap ID if available var createOut *ec2.CreateInternetGatewayOutput var err error @@ -228,24 +221,20 @@ func (d *Daemon) ensureDefaultIGWRoute(ctx context.Context, accountID, vpcID str } } - // Find the attached IGW for this VPC - igwOut, err := d.igwService.DescribeInternetGateways(ctx, &ec2.DescribeInternetGatewaysInput{}, accountID) + // Find the IGW for this VPC. Uses the intent view because this runs in the + // same call that attaches it: waiting for confirmation would leave a new + // account's default VPC with no default route until some later restart. + igw, err := d.igwService.AttachmentIntent(ctx, accountID, vpcID) if err != nil { slog.WarnContext(ctx, "Failed to query IGWs for default route", "vpcId", vpcID, "err", err) return } - var igwID string - for _, igw := range igwOut.InternetGateways { - for _, att := range igw.Attachments { - if att.VpcId != nil && *att.VpcId == vpcID { - igwID = *igw.InternetGatewayId - break - } - } - } - if igwID == "" { - return // No IGW attached + if igw == nil { + slog.WarnContext(ctx, "No IGW for default route; VPC has no egress until one is attached", + "vpcId", vpcID, "accountID", accountID) + return } + igwID := *igw.InternetGatewayId // Add the default route dest := "0.0.0.0/0" diff --git a/spinifex/handlers/bedrock/fakes_test.go b/spinifex/handlers/bedrock/fakes_test.go index c70eb3d84..0cc8d16df 100644 --- a/spinifex/handlers/bedrock/fakes_test.go +++ b/spinifex/handlers/bedrock/fakes_test.go @@ -134,20 +134,20 @@ func (f *fakeSystemVPC) DeleteInternetGateway(context.Context, *ec2.DeleteIntern return &ec2.DeleteInternetGatewayOutput{}, nil } -// DescribeInternetGateways answers the post-attach lookup: the builder -// attaches an IGW and then re-reads it, so the pre-create lookup must find -// nothing (or no IGW is ever created) and only the post-attach one answered. -func (f *fakeSystemVPC) DescribeInternetGateways(_ context.Context, in *ec2.DescribeInternetGatewaysInput, _ string) (*ec2.DescribeInternetGatewaysOutput, error) { +// AttachmentIntent answers the post-attach lookup: the builder attaches an IGW +// and then re-reads it, so the pre-create lookup must find nothing (or no IGW +// is ever created) and only the post-attach one answered. +func (f *fakeSystemVPC) AttachmentIntent(_ context.Context, _, vpcID string) (*ec2.InternetGateway, error) { f.mu.Lock() created := f.igwCreated f.mu.Unlock() if !created { - return &ec2.DescribeInternetGatewaysOutput{}, nil + return nil, nil } - return &ec2.DescribeInternetGatewaysOutput{InternetGateways: []*ec2.InternetGateway{{ + return &ec2.InternetGateway{ InternetGatewayId: aws.String("igw-bedrocksys"), - Attachments: []*ec2.InternetGatewayAttachment{{VpcId: in.Filters[0].Values[0]}}, - }}}, nil + Attachments: []*ec2.InternetGatewayAttachment{{VpcId: aws.String(vpcID)}}, + }, nil } // testModelID is the real self-host catalog entry (MinVRAMMiB=5120), used diff --git a/spinifex/handlers/ec2/igw/service_impl.go b/spinifex/handlers/ec2/igw/service_impl.go index cdaf055a5..ef13c2c32 100644 --- a/spinifex/handlers/ec2/igw/service_impl.go +++ b/spinifex/handlers/ec2/igw/service_impl.go @@ -36,10 +36,28 @@ type IGWRecord struct { InternetGatewayId string `json:"internet_gateway_id"` VpcId string `json:"vpc_id,omitempty"` // empty when detached State string `json:"state"` // "available" — AWS attachment.state is "available" when attached + AttachState string `json:"attach_state,omitempty"` Tags map[string]string `json:"tags"` CreatedAt time.Time `json:"created_at"` } +// Observed attachment state, distinct from State. State stays the AWS-facing +// attachment.state and doubles as the reconciler's intent signal; AttachState +// records whether vpcd has actually brought the OVN gateway up. +const ( + // AttachStatePending is exported for the reconciler, which decides from the + // record whether an attachment still needs confirming. + AttachStatePending = "pending" + attachStateAttached = "attached" +) + +// attachmentVisible reports whether the record has an attachment AWS would +// return. A pending attach has been requested but not yet confirmed in OVN; an +// empty AttachState is a record predating attach tracking. +func (r *IGWRecord) attachmentVisible() bool { + return r.VpcId != "" && r.AttachState != AttachStatePending +} + // GatePublisher recomputes per-subnet egress gate/ungate decisions for a VPC. // Wired by the daemon so IGW attach/detach triggers immediate OVN policy updates // rather than waiting for the reconciler's drift tick. @@ -153,9 +171,12 @@ func (s *IGWServiceImpl) DeleteInternetGateway(ctx context.Context, input *ec2.D return nil, errors.New(awserrors.ErrorServerInternal) } - // Cannot delete an attached IGW + // Cannot delete an attached IGW. The VPC is named because a pending + // attachment is hidden from describes, so the caller has no other way to + // learn what it must detach from. if record.VpcId != "" { - return nil, errors.New(awserrors.ErrorDependencyViolation) + return nil, awserrors.Errorf(awserrors.ErrorDependencyViolation, + "the internet gateway is attached to %s and must be detached first", record.VpcId) } if err := s.igwKV.Delete(ctx, key); err != nil { @@ -258,9 +279,12 @@ func igwMatchesFilters(record *IGWRecord, filters map[string][]string) bool { field = record.InternetGatewayId case "attachment.vpc-id": field = record.VpcId + if !record.attachmentVisible() { + field = "" // no reportable attachment means no vpc to match + } case "attachment.state": field = record.State - if record.VpcId == "" { + if !record.attachmentVisible() { field = "" // no attachment means no state to match } default: @@ -315,6 +339,9 @@ func (s *IGWServiceImpl) AttachInternetGateway(ctx context.Context, input *ec2.A record.VpcId = vpcID record.State = "available" + // vpcd attaches the OVN gateway asynchronously, so the attachment is not + // reported until a reconcile pass confirms it. + record.AttachState = AttachStatePending data, err := json.Marshal(record) if err != nil { @@ -376,6 +403,7 @@ func (s *IGWServiceImpl) DetachInternetGateway(ctx context.Context, input *ec2.D record.VpcId = "" record.State = "available" + record.AttachState = "" data, err := json.Marshal(record) if err != nil { @@ -412,12 +440,97 @@ func (s *IGWServiceImpl) DetachInternetGateway(ctx context.Context, input *ec2.D return &ec2.DetachInternetGatewayOutput{}, nil } +// AttachmentIntent returns the IGW whose record names vpcID, or nil if none +// does. Unlike DescribeInternetGateways it reports an attachment that has been +// requested but not yet confirmed, because a caller provisioning or tearing +// down a VPC asks what was asked for, not what OVN has caught up with. The AWS +// surface must not answer that question, so this is the in-process seam for it. +func (s *IGWServiceImpl) AttachmentIntent(ctx context.Context, accountID, vpcID string) (*ec2.InternetGateway, error) { + if vpcID == "" { + return nil, nil + } + + prefix := accountID + "." + keys, err := s.igwKV.Keys(ctx) + if err != nil { + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + slog.ErrorContext(ctx, "AttachmentIntent: IGW key listing failed", "accountID", accountID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + for _, key := range keys { + if key == utils.VersionKey || !strings.HasPrefix(key, prefix) { + continue + } + entry, err := s.igwKV.Get(ctx, key) + if err != nil { + // Fail closed: callers read nil as "no gateway" and build one, so a + // record this could not read must not read as absent. + if errors.Is(err, jetstream.ErrKeyNotFound) { + continue + } + slog.ErrorContext(ctx, "AttachmentIntent: IGW read failed", "key", key, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + var record IGWRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + slog.ErrorContext(ctx, "AttachmentIntent: IGW unmarshal failed", "key", key, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + if record.VpcId != vpcID { + continue + } + // Built directly rather than through recordToEC2, which hides a pending + // attachment: this view exists to show one. + return &ec2.InternetGateway{ + InternetGatewayId: aws.String(record.InternetGatewayId), + Attachments: []*ec2.InternetGatewayAttachment{ + {VpcId: aws.String(record.VpcId), State: aws.String(record.State)}, + }, + Tags: utils.MapToEC2Tags(record.Tags), + }, nil + } + + return nil, nil +} + +// MarkAttached records that vpcd has brought the OVN gateway up for vpcID on the +// IGW at recordKey. vpcID must match: a record key survives detach and re-attach, +// so a pass that converged the previous VPC would otherwise confirm the new one. +// Writing only on the pending transition keeps a pass from waking the drift loop. +func MarkAttached(ctx context.Context, kv jetstream.KeyValue, recordKey, vpcID string) error { + confirmed := false + record, err := kvutil.Update(ctx, kv, recordKey, kvutil.CASConfig{}, func(r *IGWRecord) (bool, error) { + // Reset per attempt: a retried CAS may see a record another writer has + // already moved out of pending. + confirmed = r.VpcId == vpcID && r.AttachState == AttachStatePending + if !confirmed { + return false, nil + } + r.AttachState = attachStateAttached + return true, nil + }) + if err != nil { + return fmt.Errorf("confirm IGW attachment %s: %w", recordKey, err) + } + + if confirmed { + slog.InfoContext(ctx, "IGW attachment confirmed", "internetGatewayId", record.InternetGatewayId, "vpcId", record.VpcId) + } + + return nil +} + func (s *IGWServiceImpl) recordToEC2(record *IGWRecord) *ec2.InternetGateway { igw := &ec2.InternetGateway{ InternetGatewayId: aws.String(record.InternetGatewayId), } - if record.VpcId != "" { + // AWS returns no attachment at all unless one exists, so a requested but + // unconfirmed attach must not be reported as available. + if record.attachmentVisible() { igw.Attachments = []*ec2.InternetGatewayAttachment{ { VpcId: aws.String(record.VpcId), diff --git a/spinifex/handlers/ec2/igw/service_impl_test.go b/spinifex/handlers/ec2/igw/service_impl_test.go index 7f83f9881..ff1a7c2a4 100644 --- a/spinifex/handlers/ec2/igw/service_impl_test.go +++ b/spinifex/handlers/ec2/igw/service_impl_test.go @@ -2,6 +2,7 @@ package handlers_ec2_igw import ( "context" + "encoding/json" "sync" "testing" "time" @@ -61,6 +62,13 @@ func setupTestIGWService(t *testing.T) (*IGWServiceImpl, *nats.Conn) { return svc, nc } +// confirmAttach stands in for the vpcd reconcile pass that observes the OVN +// gateway come up. Without it an attach stays pending and reports no attachment. +func confirmAttach(t *testing.T, svc *IGWServiceImpl, igwID, vpcID string) { + t.Helper() + require.NoError(t, MarkAttached(context.Background(), svc.igwKV, utils.AccountKey(testAccountID, igwID), vpcID)) +} + func createTestIGW(t *testing.T, svc *IGWServiceImpl) string { t.Helper() out, err := svc.CreateInternetGateway(context.Background(), &ec2.CreateInternetGatewayInput{}, testAccountID) @@ -206,17 +214,245 @@ func TestAttachInternetGateway(t *testing.T) { }, testAccountID) require.NoError(t, err) - // Verify attachment via describe + // The attach is only requested here; vpcd brings the OVN gateway up later, + // so AWS-faithfully there is no attachment to report yet. desc, err := svc.DescribeInternetGateways(context.Background(), &ec2.DescribeInternetGatewaysInput{ InternetGatewayIds: []*string{aws.String(igwID)}, }, testAccountID) require.NoError(t, err) require.Len(t, desc.InternetGateways, 1) + assert.Empty(t, desc.InternetGateways[0].Attachments, "a pending attach must report no attachment: EC2 omits the attachment until it exists") + + confirmAttach(t, svc, igwID, "vpc-test123") + + desc, err = svc.DescribeInternetGateways(context.Background(), &ec2.DescribeInternetGatewaysInput{ + InternetGatewayIds: []*string{aws.String(igwID)}, + }, testAccountID) + require.NoError(t, err) + require.Len(t, desc.InternetGateways, 1) require.Len(t, desc.InternetGateways[0].Attachments, 1) assert.Equal(t, "vpc-test123", *desc.InternetGateways[0].Attachments[0].VpcId) assert.Equal(t, "available", *desc.InternetGateways[0].Attachments[0].State) } +// AttachmentIntent is the seam every internal provisioning and teardown caller +// uses, because they must see a requested attachment. Reading the AWS-facing +// describe instead makes them create duplicate gateways and skip teardowns. +func TestAttachmentIntent_SeesPendingAndConfirmedAlike(t *testing.T) { + svc, _ := setupTestIGWService(t) + igwID := createTestIGW(t, svc) + ctx := context.Background() + + got, err := svc.AttachmentIntent(ctx, testAccountID, "vpc-test123") + require.NoError(t, err) + assert.Nil(t, got, "no attach requested yet") + + _, err = svc.AttachInternetGateway(ctx, &ec2.AttachInternetGatewayInput{ + InternetGatewayId: aws.String(igwID), + VpcId: aws.String("vpc-test123"), + }, testAccountID) + require.NoError(t, err) + + got, err = svc.AttachmentIntent(ctx, testAccountID, "vpc-test123") + require.NoError(t, err) + require.NotNil(t, got, "a pending attach is still intent: missing it creates a second gateway") + assert.Equal(t, igwID, *got.InternetGatewayId) + require.Len(t, got.Attachments, 1) + assert.Equal(t, "vpc-test123", *got.Attachments[0].VpcId) + + confirmAttach(t, svc, igwID, "vpc-test123") + got, err = svc.AttachmentIntent(ctx, testAccountID, "vpc-test123") + require.NoError(t, err) + require.NotNil(t, got, "a confirmed attachment is intent too") + assert.Equal(t, igwID, *got.InternetGatewayId) + + _, err = svc.DetachInternetGateway(ctx, &ec2.DetachInternetGatewayInput{ + InternetGatewayId: aws.String(igwID), + VpcId: aws.String("vpc-test123"), + }, testAccountID) + require.NoError(t, err) + + got, err = svc.AttachmentIntent(ctx, testAccountID, "vpc-test123") + require.NoError(t, err) + assert.Nil(t, got, "a detached gateway is no longer intent for that VPC") +} + +// Scoped per account and per VPC: a gateway in another account or on another +// VPC must not answer, or teardown detaches something it does not own. +func TestAttachmentIntent_ScopedByAccountAndVPC(t *testing.T) { + svc, _ := setupTestIGWService(t) + igwID := createTestIGW(t, svc) + ctx := context.Background() + + _, err := svc.AttachInternetGateway(ctx, &ec2.AttachInternetGatewayInput{ + InternetGatewayId: aws.String(igwID), + VpcId: aws.String("vpc-test123"), + }, testAccountID) + require.NoError(t, err) + + got, err := svc.AttachmentIntent(ctx, testAccountID, "vpc-other") + require.NoError(t, err) + assert.Nil(t, got, "another VPC must not match") + + got, err = svc.AttachmentIntent(ctx, "000000009999", "vpc-test123") + require.NoError(t, err) + assert.Nil(t, got, "another account must not match") +} + +// A record written before attach tracking existed has no AttachState. It must +// still report its attachment rather than vanishing until the next drift pass. +func TestDescribeInternetGateways_LegacyRecordReportsAttachment(t *testing.T) { + svc, _ := setupTestIGWService(t) + igwID := createTestIGW(t, svc) + + key := utils.AccountKey(testAccountID, igwID) + entry, err := svc.igwKV.Get(context.Background(), key) + require.NoError(t, err) + var record IGWRecord + require.NoError(t, json.Unmarshal(entry.Value(), &record)) + record.VpcId = "vpc-test123" + record.AttachState = "" + data, err := json.Marshal(record) + require.NoError(t, err) + _, err = svc.igwKV.Put(context.Background(), key, data) + require.NoError(t, err) + + desc, err := svc.DescribeInternetGateways(context.Background(), &ec2.DescribeInternetGatewaysInput{ + InternetGatewayIds: []*string{aws.String(igwID)}, + }, testAccountID) + require.NoError(t, err) + require.Len(t, desc.InternetGateways, 1) + require.Len(t, desc.InternetGateways[0].Attachments, 1, "a record predating attach tracking must keep reporting its attachment") + assert.Equal(t, "vpc-test123", *desc.InternetGateways[0].Attachments[0].VpcId) +} + +// A pending attach must not be reachable through the attachment filters either, +// or a caller filtering on attachment.vpc-id sees a gateway that is not up. +func TestDescribeInternetGateways_PendingAttachDoesNotMatchFilters(t *testing.T) { + svc, _ := setupTestIGWService(t) + igwID := createTestIGW(t, svc) + + _, err := svc.AttachInternetGateway(context.Background(), &ec2.AttachInternetGatewayInput{ + InternetGatewayId: aws.String(igwID), + VpcId: aws.String("vpc-test123"), + }, testAccountID) + require.NoError(t, err) + + for _, f := range []*ec2.Filter{ + {Name: aws.String("attachment.vpc-id"), Values: []*string{aws.String("vpc-test123")}}, + {Name: aws.String("attachment.state"), Values: []*string{aws.String("available")}}, + } { + desc, err := svc.DescribeInternetGateways(context.Background(), &ec2.DescribeInternetGatewaysInput{ + Filters: []*ec2.Filter{f}, + }, testAccountID) + require.NoError(t, err) + assert.Emptyf(t, desc.InternetGateways, "%s must not match a pending attach", *f.Name) + } + + confirmAttach(t, svc, igwID, "vpc-test123") + + desc, err := svc.DescribeInternetGateways(context.Background(), &ec2.DescribeInternetGatewaysInput{ + Filters: []*ec2.Filter{ + {Name: aws.String("attachment.vpc-id"), Values: []*string{aws.String("vpc-test123")}}, + }, + }, testAccountID) + require.NoError(t, err) + assert.Len(t, desc.InternetGateways, 1, "the filter must match once the attachment is confirmed") +} + +// MarkAttached is called on every pass, so it must not rewrite a record that is +// already attached: the drift loop watches this bucket and would re-trigger. +func TestMarkAttached_NoRewriteWhenAlreadyAttached(t *testing.T) { + svc, _ := setupTestIGWService(t) + igwID := createTestIGW(t, svc) + + _, err := svc.AttachInternetGateway(context.Background(), &ec2.AttachInternetGatewayInput{ + InternetGatewayId: aws.String(igwID), + VpcId: aws.String("vpc-test123"), + }, testAccountID) + require.NoError(t, err) + + key := utils.AccountKey(testAccountID, igwID) + confirmAttach(t, svc, igwID, "vpc-test123") + first, err := svc.igwKV.Get(context.Background(), key) + require.NoError(t, err) + + confirmAttach(t, svc, igwID, "vpc-test123") + second, err := svc.igwKV.Get(context.Background(), key) + require.NoError(t, err) + + assert.Equal(t, first.Revision(), second.Revision(), "a second MarkAttached must not write, or every pass wakes the drift loop") +} + +// A record key survives detach and re-attach, so a pass that converged the old +// VPC must not confirm the new one: that reports an attachment to a VPC with no +// gateway, and the confirmation is one-way. +func TestMarkAttached_IgnoresAConfirmationForAnotherVPC(t *testing.T) { + svc, _ := setupTestIGWService(t) + igwID := createTestIGW(t, svc) + ctx := context.Background() + + _, err := svc.AttachInternetGateway(ctx, &ec2.AttachInternetGatewayInput{ + InternetGatewayId: aws.String(igwID), + VpcId: aws.String("vpc-test123"), + }, testAccountID) + require.NoError(t, err) + _, err = svc.DetachInternetGateway(ctx, &ec2.DetachInternetGatewayInput{ + InternetGatewayId: aws.String(igwID), + VpcId: aws.String("vpc-test123"), + }, testAccountID) + require.NoError(t, err) + _, err = svc.AttachInternetGateway(ctx, &ec2.AttachInternetGatewayInput{ + InternetGatewayId: aws.String(igwID), + VpcId: aws.String("vpc-other"), + }, testAccountID) + require.NoError(t, err) + + // The in-flight pass converged vpc-test123, which is no longer the record's VPC. + confirmAttach(t, svc, igwID, "vpc-test123") + + desc, err := svc.DescribeInternetGateways(ctx, &ec2.DescribeInternetGatewaysInput{ + InternetGatewayIds: []*string{aws.String(igwID)}, + }, testAccountID) + require.NoError(t, err) + require.Len(t, desc.InternetGateways, 1) + assert.Empty(t, desc.InternetGateways[0].Attachments, + "a stale confirmation must not report vpc-other as attached: no gateway has come up for it") +} + +// A detach clears the observed state, so a later re-attach starts pending again +// rather than inheriting the previous attachment's confirmation. +func TestDetachThenAttachStartsPending(t *testing.T) { + svc, _ := setupTestIGWService(t) + igwID := createTestIGW(t, svc) + + _, err := svc.AttachInternetGateway(context.Background(), &ec2.AttachInternetGatewayInput{ + InternetGatewayId: aws.String(igwID), + VpcId: aws.String("vpc-test123"), + }, testAccountID) + require.NoError(t, err) + confirmAttach(t, svc, igwID, "vpc-test123") + + _, err = svc.DetachInternetGateway(context.Background(), &ec2.DetachInternetGatewayInput{ + InternetGatewayId: aws.String(igwID), + VpcId: aws.String("vpc-test123"), + }, testAccountID) + require.NoError(t, err) + + _, err = svc.AttachInternetGateway(context.Background(), &ec2.AttachInternetGatewayInput{ + InternetGatewayId: aws.String(igwID), + VpcId: aws.String("vpc-other"), + }, testAccountID) + require.NoError(t, err) + + desc, err := svc.DescribeInternetGateways(context.Background(), &ec2.DescribeInternetGatewaysInput{ + InternetGatewayIds: []*string{aws.String(igwID)}, + }, testAccountID) + require.NoError(t, err) + require.Len(t, desc.InternetGateways, 1) + assert.Empty(t, desc.InternetGateways[0].Attachments, "a re-attach must start pending, not inherit the previous confirmation") +} + func TestAttachInternetGateway_NotFound(t *testing.T) { svc, _ := setupTestIGWService(t) _, err := svc.AttachInternetGateway(context.Background(), &ec2.AttachInternetGatewayInput{ @@ -524,6 +760,7 @@ func TestDescribeInternetGateways_FilterByAttachmentVpcId(t *testing.T) { VpcId: aws.String("vpc-test123"), }, testAccountID) require.NoError(t, err) + confirmAttach(t, svc, igwID, "vpc-test123") desc, err := svc.DescribeInternetGateways(context.Background(), &ec2.DescribeInternetGatewaysInput{ Filters: []*ec2.Filter{ @@ -546,6 +783,7 @@ func TestDescribeInternetGateways_FilterByAttachmentState(t *testing.T) { VpcId: aws.String("vpc-test123"), }, testAccountID) require.NoError(t, err) + confirmAttach(t, svc, igwID, "vpc-test123") desc, err := svc.DescribeInternetGateways(context.Background(), &ec2.DescribeInternetGatewaysInput{ Filters: []*ec2.Filter{ @@ -582,6 +820,7 @@ func TestDescribeInternetGateways_FilterMultipleFilters_AND(t *testing.T) { VpcId: aws.String("vpc-test123"), }, testAccountID) require.NoError(t, err) + confirmAttach(t, svc, igwID, "vpc-test123") // Match both desc, err := svc.DescribeInternetGateways(context.Background(), &ec2.DescribeInternetGatewaysInput{ diff --git a/spinifex/handlers/ec2/vpc/service_impl.go b/spinifex/handlers/ec2/vpc/service_impl.go index 739e06e88..44ec12df1 100644 --- a/spinifex/handlers/ec2/vpc/service_impl.go +++ b/spinifex/handlers/ec2/vpc/service_impl.go @@ -82,6 +82,7 @@ type VPCServiceImpl struct { eniKV jetstream.KeyValue sgKV jetstream.KeyValue rtbKV jetstream.KeyValue // route table bucket for auto-creating main route table + igwKV jetstream.KeyValue // internet gateway bucket for the DeleteVpc dependency check ipam *IPAM // Optional: injected after construction for public IP cleanup in DeleteNetworkInterface. @@ -169,6 +170,13 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn return nil, fmt.Errorf("failed to create KV bucket spinifex-vpc-route-tables: %w", err) } + // Named literally, like the route table bucket above: the IGW package + // imports this one, so its constant cannot be referenced here. + igwKV, err := kvutil.GetOrCreateBucket(ctx, js, "spinifex-igw", 10) + if err != nil { + return nil, fmt.Errorf("failed to create KV bucket spinifex-igw: %w", err) + } + ipam, err := NewIPAM(ctx, js) if err != nil { return nil, fmt.Errorf("failed to initialize IPAM: %w", err) @@ -190,6 +198,7 @@ func NewVPCServiceImplWithNATS(ctx context.Context, cfg *config.Config, natsConn eniKV: eniKV, sgKV: sgKV, rtbKV: rtbKV, + igwKV: igwKV, ipam: ipam, }, nil } @@ -443,6 +452,13 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu } } + // Reject if an internet gateway still names this VPC, as AWS does. Without + // this the gateway outlives the only VPC id that can detach it, and it can + // then never be detached or deleted. + if err := s.rejectAttachedIGW(ctx, accountID, vpcID); err != nil { + return nil, err + } + // Cascade-delete the default SG before removing the VPC record so a vpcd // failure surfaces to the caller and leaves both records intact for retry. if defaultSGId != "" { @@ -480,6 +496,57 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu return &ec2.DeleteVpcOutput{}, nil } +// rejectAttachedIGW returns DependencyViolation when an internet gateway record +// still names vpcID. It reads the raw record rather than a describe projection, +// because an attachment awaiting confirmation is hidden from the AWS surface but +// still blocks the delete. +func (s *VPCServiceImpl) rejectAttachedIGW(ctx context.Context, accountID, vpcID string) error { + if s.igwKV == nil { + return nil + } + + prefix := accountID + "." + keys, err := s.igwKV.Keys(ctx) + if err != nil { + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil + } + slog.WarnContext(ctx, "DeleteVpc: IGW key listing failed", "vpcId", vpcID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + + for _, k := range keys { + if k == utils.VersionKey || !strings.HasPrefix(k, prefix) { + continue + } + entry, err := s.igwKV.Get(ctx, k) + if err != nil { + // Deleted between Keys() and Get() is fine to skip. Any other read + // error is fail-closed: an attachment this cannot see must not let + // DeleteVpc strip the only id that can detach it. + if errors.Is(err, jetstream.ErrKeyNotFound) { + continue + } + slog.WarnContext(ctx, "DeleteVpc: IGW read failed", "key", k, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + var igw struct { + InternetGatewayId string `json:"internet_gateway_id"` + VpcId string `json:"vpc_id"` + } + if err := json.Unmarshal(entry.Value(), &igw); err != nil { + slog.WarnContext(ctx, "DeleteVpc: IGW unmarshal failed", "key", k, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + if igw.VpcId == vpcID { + return awserrors.Errorf(awserrors.ErrorDependencyViolation, + "the VPC has a dependent internet gateway %s that must be detached first", igw.InternetGatewayId) + } + } + + return nil +} + // describeVpcsValidFilters defines the set of filter names accepted by DescribeVpcs. var describeVpcsValidFilters = map[string]bool{ "vpc-id": true, diff --git a/spinifex/handlers/ec2/vpc/service_impl_test.go b/spinifex/handlers/ec2/vpc/service_impl_test.go index 067dd9026..e29f7d8ae 100644 --- a/spinifex/handlers/ec2/vpc/service_impl_test.go +++ b/spinifex/handlers/ec2/vpc/service_impl_test.go @@ -738,6 +738,41 @@ func TestDeleteVpc_RejectsNonMainRouteTable(t *testing.T) { require.NoError(t, err, "non-main route table must not be reaped by a rejected DeleteVpc") } +// An attach awaiting confirmation is hidden from describes, so the gateway +// record is the only place the attachment is visible. Deleting the VPC anyway +// strips the only id that can detach it, leaving a gateway that can never be +// detached or deleted and an account teardown that can never finish. +func TestDeleteVpc_RejectsAttachedInternetGateway(t *testing.T) { + svc := setupTestVPCService(t) + vpcID := createTestVPC(t, svc, "10.64.0.0/16") + + _, err := svc.igwKV.Put(t.Context(), testAccountID+".igw-pending", + []byte(`{"internet_gateway_id":"igw-pending","vpc_id":"`+vpcID+`","state":"available","attach_state":"pending"}`)) + require.NoError(t, err) + + _, err = svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{VpcId: aws.String(vpcID)}, testAccountID) + require.Error(t, err) + assert.ErrorContains(t, err, "DependencyViolation") + assert.ErrorContains(t, err, "igw-pending", "the caller must be told which gateway to detach") + + desc, err := svc.DescribeVpcs(context.Background(), &ec2.DescribeVpcsInput{VpcIds: []*string{aws.String(vpcID)}}, testAccountID) + require.NoError(t, err) + require.Len(t, desc.Vpcs, 1, "VPC must persist so the caller can detach the gateway and retry") +} + +// A gateway attached to some other VPC must not block this one. +func TestDeleteVpc_IgnoresInternetGatewayOnAnotherVPC(t *testing.T) { + svc := setupTestVPCService(t) + vpcID := createTestVPC(t, svc, "10.65.0.0/16") + + _, err := svc.igwKV.Put(t.Context(), testAccountID+".igw-elsewhere", + []byte(`{"internet_gateway_id":"igw-elsewhere","vpc_id":"vpc-somewhere-else","state":"available"}`)) + require.NoError(t, err) + + _, err = svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{VpcId: aws.String(vpcID)}, testAccountID) + require.NoError(t, err) +} + // TestDeleteVpc_RouteTableCheckFailsClosedOnCorruptRTB asserts a corrupt // route table record blocks DeleteVpc rather than being silently skipped — // a transient/corrupt read must never let DeleteVpc orphan a route table it diff --git a/spinifex/handlers/eks/igw_test.go b/spinifex/handlers/eks/igw_test.go index ee058baff..9a1bcab3f 100644 --- a/spinifex/handlers/eks/igw_test.go +++ b/spinifex/handlers/eks/igw_test.go @@ -14,15 +14,15 @@ import ( ) type fakeIGWProvisioner struct { - createCalls []*ec2.CreateInternetGatewayInput - attachCalls []*ec2.AttachInternetGatewayInput - detachCalls []*ec2.DetachInternetGatewayInput - deleteCalls []*ec2.DeleteInternetGatewayInput - describeCalls []*ec2.DescribeInternetGatewaysInput + createCalls []*ec2.CreateInternetGatewayInput + attachCalls []*ec2.AttachInternetGatewayInput + detachCalls []*ec2.DetachInternetGatewayInput + deleteCalls []*ec2.DeleteInternetGatewayInput - // gateways are returned by DescribeInternetGateways (filtered on - // attachment.vpc-id), keyed by IGW id. - gateways map[string]*ec2.InternetGateway + // gateways holds confirmed gateways keyed by IGW id. + gateways map[string]*ec2.InternetGateway + // intent is igwID -> vpcID for attaches not yet confirmed. + intent map[string]string nextID string createErr error } @@ -33,25 +33,6 @@ func newFakeIGWProvisioner() *fakeIGWProvisioner { return &fakeIGWProvisioner{gateways: map[string]*ec2.InternetGateway{}, nextID: "igw-fake0001"} } -func (f *fakeIGWProvisioner) DescribeInternetGateways(_ context.Context, input *ec2.DescribeInternetGatewaysInput, _ string) (*ec2.DescribeInternetGatewaysOutput, error) { - f.describeCalls = append(f.describeCalls, input) - var wantVPC string - for _, flt := range input.Filters { - if aws.StringValue(flt.Name) == "attachment.vpc-id" && len(flt.Values) > 0 { - wantVPC = aws.StringValue(flt.Values[0]) - } - } - var out []*ec2.InternetGateway - for _, igw := range f.gateways { - for _, att := range igw.Attachments { - if aws.StringValue(att.VpcId) == wantVPC { - out = append(out, igw) - } - } - } - return &ec2.DescribeInternetGatewaysOutput{InternetGateways: out}, nil -} - func (f *fakeIGWProvisioner) CreateInternetGateway(_ context.Context, input *ec2.CreateInternetGatewayInput, _ string) (*ec2.CreateInternetGatewayOutput, error) { f.createCalls = append(f.createCalls, input) if f.createErr != nil { @@ -65,22 +46,40 @@ func (f *fakeIGWProvisioner) CreateInternetGateway(_ context.Context, input *ec2 return &ec2.CreateInternetGatewayOutput{InternetGateway: igw}, nil } +// Attaching is asynchronous: the record names the VPC immediately, but the +// describe projection reports no attachment until a reconcile pass confirms it. func (f *fakeIGWProvisioner) AttachInternetGateway(_ context.Context, input *ec2.AttachInternetGatewayInput, _ string) (*ec2.AttachInternetGatewayOutput, error) { f.attachCalls = append(f.attachCalls, input) - if igw, ok := f.gateways[aws.StringValue(input.InternetGatewayId)]; ok { - igw.Attachments = append(igw.Attachments, &ec2.InternetGatewayAttachment{ - VpcId: input.VpcId, - State: aws.String("available"), - }) + if f.intent == nil { + f.intent = map[string]string{} } + f.intent[aws.StringValue(input.InternetGatewayId)] = aws.StringValue(input.VpcId) return &ec2.AttachInternetGatewayOutput{}, nil } +func (f *fakeIGWProvisioner) AttachmentIntent(_ context.Context, _, vpcID string) (*ec2.InternetGateway, error) { + for igwID, attached := range f.intent { + if attached != vpcID { + continue + } + if igw, ok := f.gateways[igwID]; ok { + out := *igw + out.Attachments = []*ec2.InternetGatewayAttachment{ + {VpcId: aws.String(vpcID), State: aws.String("available")}, + } + return &out, nil + } + } + return nil, nil +} + func (f *fakeIGWProvisioner) DetachInternetGateway(_ context.Context, input *ec2.DetachInternetGatewayInput, _ string) (*ec2.DetachInternetGatewayOutput, error) { f.detachCalls = append(f.detachCalls, input) - if igw, ok := f.gateways[aws.StringValue(input.InternetGatewayId)]; ok { + igwID := aws.StringValue(input.InternetGatewayId) + if igw, ok := f.gateways[igwID]; ok { igw.Attachments = nil } + delete(f.intent, igwID) return &ec2.DetachInternetGatewayOutput{}, nil } @@ -100,6 +99,11 @@ func (f *fakeIGWProvisioner) seedAttached(igwID, vpcID string, tagKV ...string) igw.Tags = append(igw.Tags, &ec2.Tag{Key: aws.String(tagKV[i]), Value: aws.String(tagKV[i+1])}) } f.gateways[igwID] = igw + if f.intent == nil { + f.intent = map[string]string{} + } + // A confirmed attachment is intent too. + f.intent[igwID] = vpcID } func TestEnsureClusterIGW_FreshCreatesAndAttaches(t *testing.T) { diff --git a/spinifex/handlers/rds/launch_test.go b/spinifex/handlers/rds/launch_test.go index 40515a7bc..98d3f6431 100644 --- a/spinifex/handlers/rds/launch_test.go +++ b/spinifex/handlers/rds/launch_test.go @@ -156,20 +156,17 @@ func (f *fakeSystemVPC) DeleteInternetGateway(context.Context, *ec2.DeleteIntern return &ec2.DeleteInternetGatewayOutput{}, nil } -// DescribeInternetGateways answers the post-attach lookup: the builder attaches -// an IGW and then re-reads it, so this reports one attached to any VPC asked -// about. -func (f *fakeSystemVPC) DescribeInternetGateways(_ context.Context, in *ec2.DescribeInternetGatewaysInput, _ string) (*ec2.DescribeInternetGatewaysOutput, error) { - // The pre-create lookup must find nothing, or no IGW is ever created; only - // the post-attach one is answered. Distinguish them by whether an IGW has - // been created yet. +// AttachmentIntent answers the post-attach lookup: the builder attaches an IGW +// and then re-reads it, so this reports one for any VPC asked about once +// created. The pre-create lookup must find nothing, or none is ever created. +func (f *fakeSystemVPC) AttachmentIntent(_ context.Context, _, vpcID string) (*ec2.InternetGateway, error) { if !f.igwCreated { - return &ec2.DescribeInternetGatewaysOutput{}, nil + return nil, nil } - return &ec2.DescribeInternetGatewaysOutput{InternetGateways: []*ec2.InternetGateway{{ + return &ec2.InternetGateway{ InternetGatewayId: aws.String("igw-rdssys"), - Attachments: []*ec2.InternetGatewayAttachment{{VpcId: in.Filters[0].Values[0]}}, - }}}, nil + Attachments: []*ec2.InternetGatewayAttachment{{VpcId: aws.String(vpcID)}}, + }, nil } // fakeENIs records the ENIs the launch helper created and deleted, per account, diff --git a/spinifex/handlers/systemvpc/igw.go b/spinifex/handlers/systemvpc/igw.go index d513d5040..9a2c06ef9 100644 --- a/spinifex/handlers/systemvpc/igw.go +++ b/spinifex/handlers/systemvpc/igw.go @@ -17,11 +17,14 @@ import ( // answerable on the wire are built only once an IGW is attached, and a system // VPC has no customer to provision one. type IGWProvisioner interface { - DescribeInternetGateways(ctx context.Context, input *ec2.DescribeInternetGatewaysInput, accountID string) (*ec2.DescribeInternetGatewaysOutput, error) CreateInternetGateway(ctx context.Context, input *ec2.CreateInternetGatewayInput, accountID string) (*ec2.CreateInternetGatewayOutput, error) AttachInternetGateway(ctx context.Context, input *ec2.AttachInternetGatewayInput, accountID string) (*ec2.AttachInternetGatewayOutput, error) DetachInternetGateway(ctx context.Context, input *ec2.DetachInternetGatewayInput, accountID string) (*ec2.DetachInternetGatewayOutput, error) DeleteInternetGateway(ctx context.Context, input *ec2.DeleteInternetGatewayInput, accountID string) (*ec2.DeleteInternetGatewayOutput, error) + // AttachmentIntent is the only lookup offered because attaching is + // asynchronous: DescribeInternetGateways reports confirmed attachments only, + // and using it here creates a second gateway or skips a teardown. + AttachmentIntent(ctx context.Context, accountID, vpcID string) (*ec2.InternetGateway, error) } // EnsureIGW guarantees vpcID has an attached Internet Gateway. An IGW already @@ -109,20 +112,14 @@ func DeleteIGW(ctx context.Context, igwp IGWProvisioner, owner Owner, accountID, } // AttachedIGW returns the Internet Gateway attached to vpcID, or nil if none. +// Includes an attachment still being confirmed: every caller here is asking +// whether this VPC already has a gateway, and a gateway mid-attach does. func AttachedIGW(ctx context.Context, igwp IGWProvisioner, accountID, vpcID string) (*ec2.InternetGateway, error) { - out, err := igwp.DescribeInternetGateways(ctx, &ec2.DescribeInternetGatewaysInput{ - Filters: []*ec2.Filter{{ - Name: aws.String("attachment.vpc-id"), - Values: aws.StringSlice([]string{vpcID}), - }}, - }, accountID) + igw, err := igwp.AttachmentIntent(ctx, accountID, vpcID) if err != nil { return nil, fmt.Errorf("systemvpc: describe IGWs for vpc %s: %w", vpcID, err) } - if out == nil || len(out.InternetGateways) == 0 { - return nil, nil - } - return out.InternetGateways[0], nil + return igw, nil } // ownedBy reports whether igw carries this owner's managed-by + name tags. diff --git a/spinifex/handlers/systemvpc/systemvpc_test.go b/spinifex/handlers/systemvpc/systemvpc_test.go index 1b76c8aae..a45880c86 100644 --- a/spinifex/handlers/systemvpc/systemvpc_test.go +++ b/spinifex/handlers/systemvpc/systemvpc_test.go @@ -31,8 +31,10 @@ type fakeEC2 struct { rts map[string]*ec2.RouteTable ngws map[string]*ec2.NatGateway igws map[string]*ec2.InternetGateway - sgs map[string]*ec2.SecurityGroup - eips map[string]string + // igwIntent is igwID -> vpcID for attaches not yet confirmed. + igwIntent map[string]string + sgs map[string]*ec2.SecurityGroup + eips map[string]string // routes records the default route installed on each route table, keyed by // route-table id, as ":". @@ -52,16 +54,17 @@ var ( func newFakeEC2() *fakeEC2 { return &fakeEC2{ - accs: map[string]int{}, - vpcs: map[string]*ec2.Vpc{}, - subnets: map[string]*ec2.Subnet{}, - rts: map[string]*ec2.RouteTable{}, - ngws: map[string]*ec2.NatGateway{}, - igws: map[string]*ec2.InternetGateway{}, - sgs: map[string]*ec2.SecurityGroup{}, - eips: map[string]string{}, - routes: map[string]string{}, - creates: map[string]int{}, + accs: map[string]int{}, + vpcs: map[string]*ec2.Vpc{}, + subnets: map[string]*ec2.Subnet{}, + rts: map[string]*ec2.RouteTable{}, + ngws: map[string]*ec2.NatGateway{}, + igws: map[string]*ec2.InternetGateway{}, + igwIntent: map[string]string{}, + sgs: map[string]*ec2.SecurityGroup{}, + eips: map[string]string{}, + routes: map[string]string{}, + creates: map[string]int{}, } } @@ -335,23 +338,63 @@ func (f *fakeEC2) CreateInternetGateway(_ context.Context, in *ec2.CreateInterne return &ec2.CreateInternetGatewayOutput{InternetGateway: igw}, nil } +// Attaching is asynchronous in production: the record names the VPC straight +// away, but DescribeInternetGateways reports no attachment until a reconcile +// pass confirms it. The fake models both phases, so code that reads the +// describe projection where it needs intent fails here rather than in a nightly. func (f *fakeEC2) AttachInternetGateway(_ context.Context, in *ec2.AttachInternetGatewayInput, _ string) (*ec2.AttachInternetGatewayOutput, error) { f.mu.Lock() defer f.mu.Unlock() - igw, ok := f.igws[aws.StringValue(in.InternetGatewayId)] - if !ok { - return nil, fmt.Errorf("no such igw %s", aws.StringValue(in.InternetGatewayId)) + igwID := aws.StringValue(in.InternetGatewayId) + if _, ok := f.igws[igwID]; !ok { + return nil, fmt.Errorf("no such igw %s", igwID) } - igw.Attachments = []*ec2.InternetGatewayAttachment{{VpcId: in.VpcId, State: aws.String("available")}} + if f.igwIntent == nil { + f.igwIntent = map[string]string{} + } + f.igwIntent[igwID] = aws.StringValue(in.VpcId) return &ec2.AttachInternetGatewayOutput{}, nil } +// confirmIGW is the reconcile pass: it makes the pending attach observable. +func (f *fakeEC2) confirmIGW(igwID string) { + f.mu.Lock() + defer f.mu.Unlock() + if igw, ok := f.igws[igwID]; ok { + igw.Attachments = []*ec2.InternetGatewayAttachment{ + {VpcId: aws.String(f.igwIntent[igwID]), State: aws.String("available")}, + } + } +} + +func (f *fakeEC2) AttachmentIntent(_ context.Context, _, vpcID string) (*ec2.InternetGateway, error) { + f.mu.Lock() + defer f.mu.Unlock() + for igwID, attached := range f.igwIntent { + if attached != vpcID { + continue + } + igw, ok := f.igws[igwID] + if !ok { + continue + } + out := *igw + out.Attachments = []*ec2.InternetGatewayAttachment{ + {VpcId: aws.String(vpcID), State: aws.String("available")}, + } + return &out, nil + } + return nil, nil +} + func (f *fakeEC2) DetachInternetGateway(_ context.Context, in *ec2.DetachInternetGatewayInput, _ string) (*ec2.DetachInternetGatewayOutput, error) { f.mu.Lock() defer f.mu.Unlock() - if igw, ok := f.igws[aws.StringValue(in.InternetGatewayId)]; ok { + igwID := aws.StringValue(in.InternetGatewayId) + if igw, ok := f.igws[igwID]; ok { igw.Attachments = nil } + delete(f.igwIntent, igwID) return &ec2.DetachInternetGatewayOutput{}, nil } @@ -542,6 +585,30 @@ func TestEnsureBuildsTheFullTopology(t *testing.T) { assert.Equal(t, []string{"000000000000"}, slices.Collect(maps.Keys(f.accs))) } +// Ensure attaches an IGW and then re-reads it in the next breath. Attaching is +// asynchronous, so that lookup has to see the request rather than waiting for a +// reconcile pass, or every first-time EKS/RDS/Bedrock VPC fails to build. +func TestEnsureSucceedsWhileTheAttachIsUnconfirmed(t *testing.T) { + f := newFakeEC2() + + refs, err := Ensure(t.Context(), f.deps(), testSpec("cp-demo"), "000000000000") + require.NoError(t, err, "Ensure must not require the attach to be confirmed first") + require.NotEmpty(t, refs.IGWID) + + // The AWS-facing projection still reports nothing, which is the point: + // Ensure is reading intent, not the confirmed attachment. + out, err := f.DescribeInternetGateways(t.Context(), &ec2.DescribeInternetGatewaysInput{}, "000000000000") + require.NoError(t, err) + require.Len(t, out.InternetGateways, 1) + assert.Empty(t, out.InternetGateways[0].Attachments, "an unconfirmed attach must not be reported") + + f.confirmIGW(refs.IGWID) + out, err = f.DescribeInternetGateways(t.Context(), &ec2.DescribeInternetGatewaysInput{}, "000000000000") + require.NoError(t, err) + require.Len(t, out.InternetGateways, 1) + assert.Len(t, out.InternetGateways[0].Attachments, 1, "confirming makes the attachment observable") +} + func TestEnsureIsIdempotent(t *testing.T) { f := newFakeEC2() spec := testSpec("cp-demo") diff --git a/spinifex/network/external/types.go b/spinifex/network/external/types.go index c8e36f28c..c31a7b1c1 100644 --- a/spinifex/network/external/types.go +++ b/spinifex/network/external/types.go @@ -55,4 +55,11 @@ func (p *ExternalPoolConfig) UsesIfaceMAC() bool { type IGWSpec struct { VPCID string InternetGatewayID string + // RecordKey addresses the control-plane record this spec was loaded from, + // so a successful attach can be reported back against it. Empty when the + // spec did not come from the store. + RecordKey string + // AttachPending is set when the record still awaits confirmation, so a pass + // reports one back only for an attachment that needs it. + AttachPending bool } diff --git a/spinifex/network/reconcile/apply.go b/spinifex/network/reconcile/apply.go index 3a3c3a7b8..44c8e0a6d 100644 --- a/spinifex/network/reconcile/apply.go +++ b/spinifex/network/reconcile/apply.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "github.com/mulgadc/spinifex/spinifex/network/external" "github.com/mulgadc/spinifex/spinifex/network/ovn/nbdb" "github.com/mulgadc/spinifex/spinifex/network/policy" "github.com/mulgadc/spinifex/spinifex/network/topology" @@ -441,7 +442,34 @@ func (r *reconciler) applyIGWs(ctx context.Context, intent IntentState, actual A continue } actual.ExternalSwch[vpcID] = struct{}{} - r.rebindGatewayChassis(ctx, vpcID, eipProbeIP(intent, vpcID), res) + // Confirmed only once the chassis rebind, the SB claim and the datapath + // probe all report converged: AttachIGW returning nil is not proof the + // gateway forwards, and the confirmation is one-way. + if r.rebindGatewayChassis(ctx, vpcID, eipProbeIP(intent, vpcID), res) { + r.reportIGWAttached(ctx, spec) + } + } +} + +// reportIGWAttached confirms the attachment to the control plane so describes +// stop reporting one before it exists. Not a pass failure: the gateway is up, +// and a failed report is retried by the next pass rather than driving backoff. +func (r *reconciler) reportIGWAttached(ctx context.Context, spec external.IGWSpec) { + // Steady state is every pass after the first, so a record already confirmed + // must not cost a KV round trip per pass. + if r.markAttached == nil || !spec.AttachPending { + return + } + // A spec built outside the store carries no address to report against, so + // its attachment would stay unconfirmed with no other signal. + if spec.RecordKey == "" { + slog.Warn("reconcile/apply: IGW spec has no record key; attachment cannot be confirmed", + "vpc_id", spec.VPCID, "igw_id", spec.InternetGatewayID) + return + } + if err := r.markAttached(ctx, spec.RecordKey, spec.VPCID); err != nil { + slog.Warn("reconcile/apply: marking IGW attached failed", + "vpc_id", spec.VPCID, "igw_id", spec.InternetGatewayID, "err", err) } } @@ -459,9 +487,11 @@ func eipProbeIP(intent IntentState, vpcID string) string { } // rebindGatewayChassis re-asserts chassis priority tuples on the gateway LRP. -func (r *reconciler) rebindGatewayChassis(ctx context.Context, vpcID, eipIP string, res *passResult) { +// Reports whether the gateway converged: no chassis to bind is not convergence, +// it is a node with nothing to verify against. +func (r *reconciler) rebindGatewayChassis(ctx context.Context, vpcID, eipIP string, res *passResult) bool { if len(r.chassis) == 0 { - return + return false } gwPortName := topology.GatewayRouterPort(vpcID) lrp, err := r.ovn.GetLogicalRouterPort(ctx, gwPortName) @@ -469,17 +499,20 @@ func (r *reconciler) rebindGatewayChassis(ctx context.Context, vpcID, eipIP stri slog.Warn("reconcile/apply: gateway LRP read failed; skipping chassis rebind and datapath gate", "vpc_id", vpcID, "port", gwPortName, "err", err) res.fail(classIGW, vpcID, err) - return + return false } + bound := true for i, chassis := range r.chassis { priority := max(20-(i*5), 1) if err := r.ovn.SetGatewayChassis(ctx, gwPortName, chassis, priority); err != nil { slog.Warn("reconcile/apply: SetGatewayChassis failed", "vpc_id", vpcID, "chassis", chassis, "err", err) res.fail(classIGW, vpcID, err) + bound = false } } - r.ensureGatewayClaimed(ctx, topology.GatewayChassisRedirectPort(vpcID)) - r.ensureGatewayDatapath(ctx, vpcID, gatewayLRPIP(lrp), eipIP) + claimed := r.ensureGatewayClaimed(ctx, topology.GatewayChassisRedirectPort(vpcID)) + forwarding := r.ensureGatewayDatapath(ctx, vpcID, gatewayLRPIP(lrp), eipIP) + return bound && claimed && forwarding } // gatewayLRPIP returns the bare IPv4 of the gateway router port, parsed from its @@ -510,10 +543,11 @@ func gatewayLRPIP(lrp *nbdb.LogicalRouterPort) string { // without a guest dependency, whereas the gateway LRP IP OVN answers natively and // stays green even when the EIP datapath is dead. Fall back to the LRP IP when the // VPC has no EIP. On a miss repair the uplink + recompute, then re-probe until a -// short deadline. No-op when no verifier is wired or no probe target resolved. -func (r *reconciler) ensureGatewayDatapath(ctx context.Context, vpcID, gwIP, eipIP string) { +// short deadline. Reports whether the datapath was observed forwarding; an +// unwired verifier or unresolved probe target gates the check off and passes. +func (r *reconciler) ensureGatewayDatapath(ctx context.Context, vpcID, gwIP, eipIP string) bool { if r.gwClaim == nil || (gwIP == "" && eipIP == "") { - return + return true } target := eipIP if target == "" { @@ -531,13 +565,13 @@ func (r *reconciler) ensureGatewayDatapath(ctx context.Context, vpcID, gwIP, eip reachable, err := probe() if err != nil { slog.Warn("reconcile/apply: gateway datapath probe failed", "vpc_id", vpcID, "target", target, "err", err) - return + return false } if reachable { if repaired { slog.Info("reconcile/apply: gateway datapath recovered after uplink repair", "vpc_id", vpcID, "target", target) } - return + return true } if !repaired { slog.Warn("reconcile/apply: gateway datapath unreachable despite SB claim; repairing uplink + forcing recompute", @@ -550,11 +584,11 @@ func (r *reconciler) ensureGatewayDatapath(ctx context.Context, vpcID, gwIP, eip if time.Now().After(deadline) { slog.Error("reconcile/apply: gateway datapath did not recover after uplink repair; external connectivity degraded", "vpc_id", vpcID, "target", target, "timeout_ms", otelsetup.Millis(gatewayDatapathTimeout)) - return + return false } select { case <-ctx.Done(): - return + return false case <-time.After(gatewayDatapathInterval): } } @@ -588,10 +622,11 @@ func (r *reconciler) escalateSBReset(ctx context.Context, logKV ...any) bool { // every miss, not once: after a fresh-VPC bring-up or a chassis flap a single early // nudge fires before ovn-controller has processed the gateway_chassis update (or // before the flapped chassis re-registers), so it never binds. Mirrors -// ensureGuestPortDatapath. No-op when no verifier is wired. -func (r *reconciler) ensureGatewayClaimed(ctx context.Context, crPortName string) { +// ensureGuestPortDatapath. Reports whether the binding is claimed; an unwired +// verifier gates the check off and passes. +func (r *reconciler) ensureGatewayClaimed(ctx context.Context, crPortName string) bool { if r.gwClaim == nil { - return + return true } deadline := time.Now().Add(gatewayClaimTimeout) nudged := false @@ -601,13 +636,13 @@ func (r *reconciler) ensureGatewayClaimed(ctx context.Context, crPortName string claimed, err := r.gwClaim.GatewayPortClaimed(ctx, crPortName) if err != nil { slog.Warn("reconcile/apply: gateway SB claim check failed", "port", crPortName, "err", err) - return + return false } if claimed { if nudged { slog.Info("reconcile/apply: gateway SB chassis claim converged after recompute", "port", crPortName) } - return + return true } slog.Warn("reconcile/apply: gateway SB binding unclaimed; nudging ovn-controller recompute", "port", crPortName) if err := r.gwClaim.NudgeRecompute(ctx); err != nil { @@ -621,11 +656,11 @@ func (r *reconciler) ensureGatewayClaimed(ctx context.Context, crPortName string if time.Now().After(deadline) { slog.Error("reconcile/apply: gateway SB chassis claim did not converge; floating IPs may be unreachable", "port", crPortName, "timeout_ms", otelsetup.Millis(gatewayClaimTimeout)) - return + return false } select { case <-ctx.Done(): - return + return false case <-time.After(gatewayClaimInterval): } } diff --git a/spinifex/network/reconcile/apply_test.go b/spinifex/network/reconcile/apply_test.go index dd1ab2135..dbeb33e22 100644 --- a/spinifex/network/reconcile/apply_test.go +++ b/spinifex/network/reconcile/apply_test.go @@ -1171,12 +1171,18 @@ type stubIGW struct { external.IGWManager attachErr error + // attachNoop returns success without building any OVN state, modelling an + // AttachIGW that reports nil on a gateway that does not actually forward. + attachNoop bool } func (s *stubIGW) AttachIGW(ctx context.Context, spec external.IGWSpec) error { if s.attachErr != nil { return s.attachErr } + if s.attachNoop { + return nil + } return s.IGWManager.AttachIGW(ctx, spec) } @@ -1185,10 +1191,178 @@ func (s *stubIGW) AttachIGW(ctx context.Context, spec external.IGWSpec) error { func igwIntent(t *testing.T) IntentState { t.Helper() intent := freshIntent(t) - intent.IGWs["vpc-a"] = external.IGWSpec{VPCID: "vpc-a", InternetGatewayID: "igw-a"} + intent.IGWs["vpc-a"] = external.IGWSpec{ + VPCID: "vpc-a", InternetGatewayID: "igw-a", RecordKey: "acct.igw-a", AttachPending: true, + } return intent } +// The control plane reports an attachment only once a pass confirms it, so the +// hook must fire on success and stay silent on failure — a record marked +// attached after a failed attach is the bug this replaced. +func TestReconcile_MarksIGWAttachedOnlyOnSuccess(t *testing.T) { + ctx := context.Background() + + t.Run("success", func(t *testing.T) { + rec, _ := newTestReconciler(t) + rec.chassis = []string{"chassis-1"} + rec.gwClaim = &fakeClaimVerifier{} + var keys, vpcs []string + rec.markAttached = func(_ context.Context, key, vpcID string) error { + keys = append(keys, key) + vpcs = append(vpcs, vpcID) + return nil + } + if err := rec.Reconcile(ctx, igwIntent(t)); err != nil { + t.Fatalf("Reconcile = %v, want nil", err) + } + if want := []string{"acct.igw-a"}; !slices.Equal(keys, want) { + t.Errorf("markAttached keys = %v, want %v — a converged attach must be reported "+ + "or DescribeInternetGateways never stops calling it pending", keys, want) + } + if want := []string{"vpc-a"}; !slices.Equal(vpcs, want) { + t.Errorf("markAttached vpcs = %v, want %v — the record key survives detach and "+ + "re-attach, so the VPC is what pins the confirmation to this attachment", vpcs, want) + } + }) + + t.Run("failure", func(t *testing.T) { + rec, _ := newTestReconciler(t) + rec.igw = &stubIGW{IGWManager: rec.igw, attachErr: errors.New("dhcp gw-lrp acquire: context deadline exceeded")} + called := false + rec.markAttached = func(context.Context, string, string) error { + called = true + return nil + } + if err := rec.Reconcile(ctx, igwIntent(t)); !errors.Is(err, ErrPassIncomplete) { + t.Fatalf("Reconcile = %v, want ErrPassIncomplete", err) + } + if called { + t.Error("markAttached called after a failed attach: the API would report a gateway that is not up") + } + }) + + // The datapath gate logs at Error and records no pass failure, so a + // confirmation inferred from the failure count would report a gateway that + // demonstrably does not forward — the bug this branch exists to fix. + t.Run("datapath never converges", func(t *testing.T) { + withFastDatapathBounds(t) + rec, _ := newTestReconciler(t) + rec.chassis = []string{"chassis-1"} + rec.gwClaim = &fakeClaimVerifier{reachableAfter: -1} + called := false + rec.markAttached = func(context.Context, string, string) error { + called = true + return nil + } + // A distributed-NAT gateway LRP is link-local and gates the probe off, + // so an EIP is what gives the datapath check a target. + intent := igwIntent(t) + intent.EIPs["10.0.1.5"] = policy.EIPSpec{VPCID: "vpc-a", ExternalIP: "203.0.113.5", LogicalIP: "10.0.1.5"} + if err := rec.Reconcile(ctx, intent); err != nil { + t.Fatalf("Reconcile = %v, want nil", err) + } + if called { + t.Error("markAttached called while the gateway datapath is unreachable: the API would " + + "report an attachment that does not forward, and the confirmation never self-corrects") + } + }) + + // Same shape as the datapath gate: an unclaimed SB binding leaves floating + // IPs dark while every other signal is green. + t.Run("SB claim never converges", func(t *testing.T) { + withFastClaimBounds(t) + rec, _ := newTestReconciler(t) + rec.chassis = []string{"chassis-1"} + rec.gwClaim = &fakeClaimVerifier{claimedAfter: -1} + called := false + rec.markAttached = func(context.Context, string, string) error { + called = true + return nil + } + if err := rec.Reconcile(ctx, igwIntent(t)); err != nil { + t.Fatalf("Reconcile = %v, want nil", err) + } + if called { + t.Error("markAttached called while the SB chassisredirect binding is unclaimed") + } + }) + + // A spec built outside the store has no address to confirm against; sending + // an empty key would have vpcd read key "" on every pass. + t.Run("no record key", func(t *testing.T) { + rec, _ := newTestReconciler(t) + rec.chassis = []string{"chassis-1"} + rec.gwClaim = &fakeClaimVerifier{} + called := false + rec.markAttached = func(context.Context, string, string) error { + called = true + return nil + } + intent := freshIntent(t) + intent.IGWs["vpc-a"] = external.IGWSpec{VPCID: "vpc-a", InternetGatewayID: "igw-a", AttachPending: true} + if err := rec.Reconcile(ctx, intent); err != nil { + t.Fatalf("Reconcile = %v, want nil", err) + } + if called { + t.Error("markAttached called with no record key") + } + }) + + // A record already confirmed must cost nothing: the drift loop watches this + // bucket, so a per-pass read is a per-pass wakeup risk for no work. + t.Run("already confirmed", func(t *testing.T) { + rec, _ := newTestReconciler(t) + rec.chassis = []string{"chassis-1"} + rec.gwClaim = &fakeClaimVerifier{} + called := false + rec.markAttached = func(context.Context, string, string) error { + called = true + return nil + } + intent := igwIntent(t) + spec := intent.IGWs["vpc-a"] + spec.AttachPending = false + intent.IGWs["vpc-a"] = spec + if err := rec.Reconcile(ctx, intent); err != nil { + t.Fatalf("Reconcile = %v, want nil", err) + } + if called { + t.Error("markAttached called for an attachment already confirmed") + } + }) + + // AttachIGW returning nil is not proof the gateway forwards. Confirming + // before the chassis rebind would report a gateway with no bound chassis as + // attached, and the confirmation is one-way so it would never self-correct. + t.Run("chassis rebind failed", func(t *testing.T) { + rec, _ := newTestReconciler(t) + rec.chassis = []string{"chassis-1"} + rec.igw = &stubIGW{IGWManager: rec.igw, attachNoop: true} + called := false + rec.markAttached = func(context.Context, string, string) error { + called = true + return nil + } + if err := rec.Reconcile(ctx, igwIntent(t)); !errors.Is(err, ErrPassIncomplete) { + t.Fatalf("Reconcile = %v, want ErrPassIncomplete", err) + } + if called { + t.Error("markAttached called after a failed chassis rebind: the gateway has no bound " + + "chassis, and the confirmation is one-way so no later pass would revert it") + } + }) + + // vpcd wires the hook, unit callers do not; a nil hook must not panic. + t.Run("nil hook", func(t *testing.T) { + rec, _ := newTestReconciler(t) + rec.markAttached = nil + if err := rec.Reconcile(ctx, igwIntent(t)); err != nil { + t.Fatalf("Reconcile = %v, want nil with no hook wired", err) + } + }) +} + // A failed AttachIGW must surface as ErrPassIncomplete rather than a nil return: // a swallowed failure is indistinguishable from a converged pass, so nothing // retries it until the next drift tick. diff --git a/spinifex/network/reconcile/drift.go b/spinifex/network/reconcile/drift.go index 8673e8ff4..6609bd427 100644 --- a/spinifex/network/reconcile/drift.go +++ b/spinifex/network/reconcile/drift.go @@ -33,9 +33,10 @@ var driftBackoffBase = 5 * time.Second // of a VPC having no external gateway. const driftBackoffFactor = 3 -// intentBuckets names every KV bucket LoadIntentFromKV reads. A write to one of -// them is an intent change and nothing else: none of their writers is periodic, -// and a pass writes OVN rather than KV, so watching them cannot self-trigger. +// intentBuckets names every KV bucket LoadIntentFromKV reads. None of their +// writers is periodic, so watching them cannot self-trigger a loop. The one +// write a pass makes is confirming an IGW attachment, which fires only on the +// pending transition and so costs one extra pass per attach, not a cycle. var intentBuckets = []string{ handlers_ec2_vpc.KVBucketVPCs, handlers_ec2_vpc.KVBucketSubnets, diff --git a/spinifex/network/reconcile/intent.go b/spinifex/network/reconcile/intent.go index 8c0ca9112..b8a4d6584 100644 --- a/spinifex/network/reconcile/intent.go +++ b/spinifex/network/reconcile/intent.go @@ -441,6 +441,8 @@ func loadIGWs(ctx context.Context, js jetstream.JetStream, localVPCs map[string] out[rec.VpcId] = external.IGWSpec{ VPCID: rec.VpcId, InternetGatewayID: rec.InternetGatewayId, + RecordKey: key, + AttachPending: rec.AttachState == handlers_ec2_igw.AttachStatePending, } } return nil diff --git a/spinifex/network/reconcile/intent_test.go b/spinifex/network/reconcile/intent_test.go index cd5bf895c..47b5878f4 100644 --- a/spinifex/network/reconcile/intent_test.go +++ b/spinifex/network/reconcile/intent_test.go @@ -210,14 +210,56 @@ func TestLoadIntentFromKV_IGWAttachedFilter(t *testing.T) { t.Fatalf("LoadIntentFromKV: %v", err) } - if _, ok := intent.IGWs["vpc-local"]; !ok { - t.Errorf("attached IGW missing from intent") + spec, ok := intent.IGWs["vpc-local"] + if !ok { + t.Fatalf("attached IGW missing from intent") + } + // RecordKey is the only address the pass has to confirm the attachment + // against; dropping it silently stops every confirmation forever. + if spec.RecordKey != "acct/igw-attached" { + t.Errorf("RecordKey = %q, want %q — without it no attach is ever confirmed and "+ + "DescribeInternetGateways reports every gateway as detached", spec.RecordKey, "acct/igw-attached") } if len(intent.IGWs) != 1 { t.Errorf("got %d IGWs, want 1 (detached should be excluded)", len(intent.IGWs)) } } +// Intent gates on State, never on AttachState: an IGW enters intent so a pass +// can attach it, and every attach starts pending. Gating intent on the observed +// state instead would mean nothing is ever attached, so nothing is ever +// confirmed — a total IGW outage that no other test would catch. +func TestLoadIntentFromKV_PendingIGWStillEntersIntent(t *testing.T) { + js := startKV(t) + + testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, map[string][]byte{ + "acct/vpc-local": mustJSON(t, handlers_ec2_vpc.VPCRecord{ + VpcId: "vpc-local", CidrBlock: "10.0.0.0/16", AZ: "us-east-1a", + }), + }) + testutil.SeedKV(t, js, handlers_ec2_igw.KVBucketIGW, map[string][]byte{ + "acct/igw-pending": mustJSON(t, handlers_ec2_igw.IGWRecord{ + InternetGatewayId: "igw-pending", VpcId: "vpc-local", + State: "available", AttachState: "pending", + }), + }) + + intent, err := LoadIntentFromKV(context.Background(), js, "us-east-1a") + if err != nil { + t.Fatalf("LoadIntentFromKV: %v", err) + } + + spec, ok := intent.IGWs["vpc-local"] + if !ok { + t.Fatal("a pending IGW was excluded from intent: no pass would attach it, so it would " + + "stay pending forever and no VPC would ever get an external gateway") + } + if !spec.AttachPending { + t.Error("AttachPending not set: the pass skips the confirmation, so the attachment " + + "stays pending and is never reported") + } +} + // TestLoadIntentFromKV_NATGWUsesAssociatedSubnet enforces that loadNATGWs emits // specs for associated *private* subnets, not the NATGW's own public home subnet. // Wrong CIDR corrupts conntrack reverse-NAT and causes 100% packet loss. diff --git a/spinifex/network/reconcile/reconcile.go b/spinifex/network/reconcile/reconcile.go index 47d5bbaa8..f35c3611c 100644 --- a/spinifex/network/reconcile/reconcile.go +++ b/spinifex/network/reconcile/reconcile.go @@ -123,6 +123,11 @@ type Config struct { // Optional: nil leaves the start-of-pass snapshot as the sole liveness source // (unit tests, or callers with no store). FreshIntent func(ctx context.Context) (IntentState, error) + // MarkIGWAttached reports a confirmed IGW attachment back to the control + // plane, so DescribeInternetGateways stops claiming an attachment exists + // before one does. Called with the record key and VPC carried on the IGW + // spec. Optional: nil leaves the record untouched. + MarkIGWAttached func(ctx context.Context, recordKey, vpcID string) error } type reconciler struct { @@ -140,6 +145,7 @@ type reconciler struct { ipsecEnabled bool underlayMTU int reloadIntent func(ctx context.Context) (IntentState, error) + markAttached func(ctx context.Context, recordKey, vpcID string) error // Guest ports that burned their convergence deadline, so a port whose guest // is gone stops paying the full nudge sequence every cycle. @@ -194,6 +200,7 @@ func New(cfg Config) (Reconciler, error) { ipsecEnabled: !cfg.IPSecDisabled, underlayMTU: cfg.UnderlayMTU, reloadIntent: cfg.FreshIntent, + markAttached: cfg.MarkIGWAttached, }, nil } diff --git a/spinifex/vpcd/vpcd.go b/spinifex/vpcd/vpcd.go index f1b2697bd..2ce0c347b 100644 --- a/spinifex/vpcd/vpcd.go +++ b/spinifex/vpcd/vpcd.go @@ -17,6 +17,7 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/admin" + handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw" handlers_imds "github.com/mulgadc/spinifex/spinifex/handlers/imds" "github.com/mulgadc/spinifex/spinifex/network/external" "github.com/mulgadc/spinifex/spinifex/network/external/dhcp" @@ -692,6 +693,13 @@ func launchService(cfg *Config) error { FreshIntent: func(ctx context.Context) (reconcile.IntentState, error) { return reconcile.LoadIntentFromKV(ctx, js, cfg.AZ) }, + MarkIGWAttached: func(ctx context.Context, recordKey, vpcID string) error { + kv, err := js.KeyValue(ctx, handlers_ec2_igw.KVBucketIGW) + if err != nil { + return fmt.Errorf("open %s: %w", handlers_ec2_igw.KVBucketIGW, err) + } + return handlers_ec2_igw.MarkAttached(ctx, kv, recordKey, vpcID) + }, }) if err != nil { return fmt.Errorf("construct reconciler: %w", err)