mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 09:01:31 +00:00
fix: report an IGW attachment only once the OVN gateway is up (#962)
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 "<target-kind>:<target-id>".
|
||||
@@ -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")
|
||||
|
||||
Vendored
+7
@@ -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
|
||||
}
|
||||
|
||||
@@ -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):
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user