From e262ee0064c3221f52309f90337e801ba340833e Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Fri, 10 Jul 2026 11:34:39 +1000 Subject: [PATCH] fix(elbv2): reap LBs wedged in provisioning + harden IMDS bootstrap (#496) An ELBv2 LB whose system VM launches but whose in-guest lb-agent never delivers a first heartbeat previously stayed in provisioning forever with an empty StateReason: promotion to active happens only on the first heartbeat and the only provisioning->failed transitions were at create/launch time. A single wedged LB in the multi-node e2e lb suite leaked its listener/target-group and cascaded into ResourceInUse and DependencyViolation on VPC teardown. - Add a lifecycle reaper that marks failed any LB still provisioning past lbProvisioningTimeout (4m) with no heartbeat, reusing markLBFailed, and populate a provisioning StateReason so a stuck LB is self-describing. Started once at daemon startup; stopped on service Close. - Harness: add ErrLBProvisioningTimeout and treat a provisioning-timeout as retryable in createActiveLB; register idempotent LB/listener teardown up-front so no exit path leaks into the shared VPC. - IMDS: bounded retry+backoff on the instance-fact and user-data lookups so a transient gather timeout under multi-node load does not starve the lb-agent bootstrap. --- spinifex/daemon/daemon.go | 2 + spinifex/handlers/elbv2/reaper_test.go | 80 +++++++++++++++++++ spinifex/handlers/elbv2/service_impl.go | 79 +++++++++++++++--- spinifex/handlers/imds/instance_lookup.go | 52 ++++++++++-- .../imds/instance_lookup_retry_test.go | 63 +++++++++++++++ tests/e2e/harness/lb.go | 7 +- tests/e2e/lb/lb_test.go | 9 ++- 7 files changed, 272 insertions(+), 20 deletions(-) create mode 100644 spinifex/handlers/elbv2/reaper_test.go create mode 100644 spinifex/handlers/imds/instance_lookup_retry_test.go diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 08f9fa720..b60697146 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1489,6 +1489,8 @@ func (d *Daemon) startCluster() error { "err", err) } + d.elbv2Service.StartLifecycleReaper(context.Background()) + d.eksService, err = initServiceWithRetry("EKS service", func() (*handlers_eks.EKSServiceImpl, error) { return handlers_eks.NewEKSServiceImpl(d.buildEKSServiceDeps()) }) diff --git a/spinifex/handlers/elbv2/reaper_test.go b/spinifex/handlers/elbv2/reaper_test.go new file mode 100644 index 000000000..1304ce527 --- /dev/null +++ b/spinifex/handlers/elbv2/reaper_test.go @@ -0,0 +1,80 @@ +package handlers_elbv2 + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// testLBRecord builds a minimal, valid LoadBalancerRecord for reaper tests. +func testLBRecord(id, state string, createdAt, lastHeartbeat time.Time) *LoadBalancerRecord { + return &LoadBalancerRecord{ + LoadBalancerArn: "arn:aws:elasticloadbalancing:us-east-1:" + testAccountID + ":loadbalancer/app/reaper-" + id + "/" + id, + LoadBalancerID: id, + Name: "reaper-" + id, + State: state, + AccountID: testAccountID, + CreatedAt: createdAt, + LastHeartbeat: lastHeartbeat, + } +} + +func TestReapStuckProvisioningLBs(t *testing.T) { + svc := setupTestService(t) + now := time.Now().UTC() + + stuck := testLBRecord("stuck0000000000", StateProvisioning, now.Add(-10*time.Minute), time.Time{}) + require.NoError(t, svc.store.PutLoadBalancer(stuck)) + + fresh := testLBRecord("fresh0000000000", StateProvisioning, now, time.Time{}) + require.NoError(t, svc.store.PutLoadBalancer(fresh)) + + activeOld := testLBRecord("active000000000", StateActive, now.Add(-10*time.Minute), time.Time{}) + require.NoError(t, svc.store.PutLoadBalancer(activeOld)) + + heartbeating := testLBRecord("heartbeat0000000", StateProvisioning, now.Add(-10*time.Minute), now.Add(-time.Second)) + require.NoError(t, svc.store.PutLoadBalancer(heartbeating)) + + svc.reapStuckProvisioningLBs(now) + + got, err := svc.store.GetLoadBalancer(stuck.LoadBalancerID) + require.NoError(t, err) + assert.Equal(t, StateFailed, got.State) + assert.NotEmpty(t, got.StateReason) + + got, err = svc.store.GetLoadBalancer(fresh.LoadBalancerID) + require.NoError(t, err) + assert.Equal(t, StateProvisioning, got.State) + + got, err = svc.store.GetLoadBalancer(activeOld.LoadBalancerID) + require.NoError(t, err) + assert.Equal(t, StateActive, got.State) + + got, err = svc.store.GetLoadBalancer(heartbeating.LoadBalancerID) + require.NoError(t, err) + assert.Equal(t, StateProvisioning, got.State) +} + +// TestStartLifecycleReaper drives the ticker goroutine with a tiny interval and +// confirms it reaps a stuck LB, then cancels to exercise the ctx.Done exit. +func TestStartLifecycleReaper(t *testing.T) { + svc := setupTestService(t) + svc.reaperInterval = 5 * time.Millisecond + + stuck := testLBRecord("ticker0000000000", StateProvisioning, time.Now().UTC().Add(-10*time.Minute), time.Time{}) + require.NoError(t, svc.store.PutLoadBalancer(stuck)) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc.StartLifecycleReaper(ctx) + + require.Eventually(t, func() bool { + got, err := svc.store.GetLoadBalancer(stuck.LoadBalancerID) + return err == nil && got != nil && got.State == StateFailed + }, time.Second, 10*time.Millisecond, "reaper goroutine should mark the stuck LB failed") + + cancel() +} diff --git a/spinifex/handlers/elbv2/service_impl.go b/spinifex/handlers/elbv2/service_impl.go index 18db5cd5d..d03d4d5b9 100644 --- a/spinifex/handlers/elbv2/service_impl.go +++ b/spinifex/handlers/elbv2/service_impl.go @@ -40,6 +40,13 @@ const ( // heartbeatPersistInterval is how often a no-op heartbeat writes to KV. heartbeatPersistInterval = 60 * time.Second + // lbProvisioningTimeout bounds how long an LB may sit in provisioning + // waiting for its lb-agent's first heartbeat before the reaper marks it + // failed. Kept below the e2e 5m wait so a wedged LB terminal-fails and the + // harness can retry rather than time out. + lbProvisioningTimeout = 4 * time.Minute + // lbReaperInterval is how often the lifecycle reaper sweeps for stuck LBs. + lbReaperInterval = 30 * time.Second // Health check fields are interpolated into the HAProxy template; restrict // to characters that cannot terminate or inject directives. @@ -118,6 +125,7 @@ type ELBv2ServiceImpl struct { ctx context.Context cancel context.CancelFunc hc *healthChecker + reaperInterval time.Duration // lifecycle reaper tick; 0 falls back to lbReaperInterval recoveryLBIndexMu sync.Mutex recoveryLBIndex map[string]*LoadBalancerRecord @@ -159,15 +167,16 @@ func NewELBv2ServiceImplWithNATS(cfg *config.Config, nc *nats.Conn) (*ELBv2Servi hc := newHealthChecker(store) return &ELBv2ServiceImpl{ - config: cfg, - store: store, - acmStore: acmStore, - nc: nc, - nodeID: nodeID, - region: region, - ctx: ctx, - cancel: cancel, - hc: hc, + config: cfg, + store: store, + acmStore: acmStore, + nc: nc, + nodeID: nodeID, + region: region, + ctx: ctx, + cancel: cancel, + hc: hc, + reaperInterval: lbReaperInterval, }, nil } @@ -1243,6 +1252,7 @@ func (s *ELBv2ServiceImpl) createLoadBalancer(ctx context.Context, input *elbv2. willLaunch := s.InstanceLauncher != nil && len(eniIDs) > 0 && len(subnets) > 0 if willLaunch { state = StateProvisioning + stateReason = "awaiting first lb-agent heartbeat" // Internal LBs without a mgmt return route would hang in provisioning; // mark failed immediately so the broken path is visible. if scheme == SchemeInternal { @@ -1419,6 +1429,57 @@ func (s *ELBv2ServiceImpl) markLBFailed(lbArn, reason string) { } } +// reapStuckProvisioningLBs marks failed any LB that has sat in provisioning past +// lbProvisioningTimeout without a single lb-agent heartbeat. Such an LB launched +// its VM but the agent never reached the daemon, so it would otherwise wedge in +// provisioning forever. Synchronous; caller supplies now for testability. +func (s *ELBv2ServiceImpl) reapStuckProvisioningLBs(now time.Time) { + lbs, err := s.store.ListLoadBalancers() + if err != nil { + slog.Error("ELBv2 lifecycle reaper: failed to list load balancers", "err", err) + return + } + for _, lb := range lbs { + if lb == nil || lb.State != StateProvisioning { + continue + } + if !lb.LastHeartbeat.IsZero() || lb.CreatedAt.IsZero() { + continue + } + if now.Sub(lb.CreatedAt) < lbProvisioningTimeout { + continue + } + reason := fmt.Sprintf("lb-agent did not heartbeat within %s of provisioning", lbProvisioningTimeout) + slog.Warn("ELBv2 lifecycle reaper: LB stuck in provisioning, marking failed", + "lbArn", lb.LoadBalancerArn, "lbId", lb.LoadBalancerID, + "instanceId", lb.InstanceID, "createdAt", lb.CreatedAt) + s.markLBFailed(lb.LoadBalancerArn, reason) + } +} + +// StartLifecycleReaper runs reapStuckProvisioningLBs on a ticker until the +// service context is cancelled (via Close). Call once at daemon startup. +func (s *ELBv2ServiceImpl) StartLifecycleReaper(ctx context.Context) { + interval := s.reaperInterval + if interval <= 0 { + interval = lbReaperInterval + } + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-s.ctx.Done(): + return + case <-ctx.Done(): + return + case <-ticker.C: + s.reapStuckProvisioningLBs(time.Now().UTC()) + } + } + }() +} + func (s *ELBv2ServiceImpl) DeleteLoadBalancer(ctx context.Context, input *elbv2.DeleteLoadBalancerInput, accountID string) (*elbv2.DeleteLoadBalancerOutput, error) { if input.LoadBalancerArn == nil || *input.LoadBalancerArn == "" { return nil, errors.New(awserrors.ErrorMissingParameter) diff --git a/spinifex/handlers/imds/instance_lookup.go b/spinifex/handlers/imds/instance_lookup.go index 11f8f1466..28b3fff1e 100644 --- a/spinifex/handlers/imds/instance_lookup.go +++ b/spinifex/handlers/imds/instance_lookup.go @@ -5,6 +5,7 @@ import ( "encoding/base64" "fmt" "log/slog" + "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" @@ -12,6 +13,39 @@ import ( "github.com/nats-io/nats.go" ) +// imdsLookupRetries bounds transient-gather retries for IMDS instance lookups so +// a momentary NATS gather timeout under load does not starve the guest bootstrap. +const imdsLookupRetries = 3 + +// retryBackoff sleeps a short increasing delay between IMDS lookup attempts, +// returning false if ctx is cancelled first. +func retryBackoff(ctx context.Context, attempt int) bool { + select { + case <-ctx.Done(): + return false + case <-time.After(time.Duration(attempt) * 200 * time.Millisecond): + return true + } +} + +// retryGather retries fn on transient error up to imdsLookupRetries times with a +// short backoff, honouring ctx. Returns the last result and error. +func retryGather[T any](ctx context.Context, label, instanceID string, fn func() (*T, error)) (*T, error) { + var out *T + var err error + for attempt := 1; attempt <= imdsLookupRetries; attempt++ { + out, err = fn() + if err == nil { + return out, nil + } + if attempt == imdsLookupRetries || !retryBackoff(ctx, attempt) { + return out, err + } + slog.WarnContext(ctx, "IMDS: "+label+" failed, retrying", "instance_id", instanceID, "attempt", attempt, "err", err) + } + return out, err +} + // natsInstanceLookup resolves instance-only metadata fields via DescribeInstances fan-out. type natsInstanceLookup struct { nc *nats.Conn @@ -21,9 +55,11 @@ type natsInstanceLookup struct { var _ instanceLookup = (*natsInstanceLookup)(nil) func (l *natsInstanceLookup) describe(ctx context.Context, accountID, instanceID string) (*instanceFacts, error) { - out, err := gateway_ec2_instance.DescribeInstances(ctx, &ec2.DescribeInstancesInput{ - InstanceIds: []*string{aws.String(instanceID)}, - }, l.nc, l.expectedNodes, accountID) + out, err := retryGather(ctx, "describe instance", instanceID, func() (*ec2.DescribeInstancesOutput, error) { + return gateway_ec2_instance.DescribeInstances(ctx, &ec2.DescribeInstancesInput{ + InstanceIds: []*string{aws.String(instanceID)}, + }, l.nc, l.expectedNodes, accountID) + }) if err != nil { return nil, fmt.Errorf("describe instance %s: %w", instanceID, err) } @@ -53,10 +89,12 @@ func (l *natsInstanceLookup) describe(ctx context.Context, accountID, instanceID // userData fetches and decodes the instance's base64 user-data, returning nil on miss or error. func (l *natsInstanceLookup) userData(ctx context.Context, accountID, instanceID string) []byte { - attr, err := gateway_ec2_instance.DescribeInstanceAttribute(ctx, &ec2.DescribeInstanceAttributeInput{ - InstanceId: aws.String(instanceID), - Attribute: aws.String("userData"), - }, l.nc, l.expectedNodes, accountID) + attr, err := retryGather(ctx, "describe instance attribute", instanceID, func() (*ec2.DescribeInstanceAttributeOutput, error) { + return gateway_ec2_instance.DescribeInstanceAttribute(ctx, &ec2.DescribeInstanceAttributeInput{ + InstanceId: aws.String(instanceID), + Attribute: aws.String("userData"), + }, l.nc, l.expectedNodes, accountID) + }) if err != nil || attr == nil || attr.UserData == nil || attr.UserData.Value == nil { return nil } diff --git a/spinifex/handlers/imds/instance_lookup_retry_test.go b/spinifex/handlers/imds/instance_lookup_retry_test.go new file mode 100644 index 000000000..63b4d7b59 --- /dev/null +++ b/spinifex/handlers/imds/instance_lookup_retry_test.go @@ -0,0 +1,63 @@ +package handlers_imds + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestRetryGather_CancelledCtx checks the ctx.Done branch of retryBackoff: a +// cancelled context returns fast after the first failure without exhausting retries. +func TestRetryGather_CancelledCtx(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + calls := 0 + wantErr := errors.New("transient") + out, err := retryGather(ctx, "test", "i-123", func() (*int, error) { + calls++ + return nil, wantErr + }) + + require.ErrorIs(t, err, wantErr) + assert.Nil(t, out) + assert.Equal(t, 1, calls, "cancelled ctx must stop after first attempt") +} + +// TestRetryGather_SucceedsAfterRetries checks the retry-then-succeed path returns +// the successful value and stops calling. +func TestRetryGather_SucceedsAfterRetries(t *testing.T) { + calls := 0 + want := 42 + out, err := retryGather(context.Background(), "test", "i-123", func() (*int, error) { + calls++ + if calls < imdsLookupRetries { + return nil, errors.New("transient") + } + v := want + return &v, nil + }) + + require.NoError(t, err) + require.NotNil(t, out) + assert.Equal(t, want, *out) + assert.Equal(t, imdsLookupRetries, calls) +} + +// TestRetryGather_ExhaustsRetries checks a persistent error is returned after +// exactly imdsLookupRetries attempts. +func TestRetryGather_ExhaustsRetries(t *testing.T) { + calls := 0 + wantErr := errors.New("persistent") + out, err := retryGather(context.Background(), "test", "i-123", func() (*int, error) { + calls++ + return nil, wantErr + }) + + require.ErrorIs(t, err, wantErr) + assert.Nil(t, out) + assert.Equal(t, imdsLookupRetries, calls) +} diff --git a/tests/e2e/harness/lb.go b/tests/e2e/harness/lb.go index 8b73ad131..fea6b583f 100644 --- a/tests/e2e/harness/lb.go +++ b/tests/e2e/harness/lb.go @@ -18,6 +18,11 @@ import ( // the LB VM. Callers can errors.Is on it to retry after capacity reclaim. var ErrLBTerminalFailed = errors.New("LB entered terminal failure state") +// ErrLBProvisioningTimeout marks a load balancer that never left provisioning +// within the wait window (no terminal failure, just stuck). Callers can +// errors.Is on it to tear down and retry, same as a terminal failure. +var ErrLBProvisioningTimeout = errors.New("LB stuck in provisioning past timeout") + // WaitForLBActive polls describe-load-balancers until state=active. Bails // immediately if the LB enters a terminal failure state — no point waiting // the full timeout when provisioning has already given up. Progress emits @@ -54,7 +59,7 @@ func WaitForLBActiveErr(t *testing.T, c *AWSClient, lbArn, label string, timeout } } if time.Now().After(deadline) { - return fmt.Errorf("%s did not become active within %s (last state=%s reason=%s)", label, timeout, lastState, lastReason) + return fmt.Errorf("%s did not become active within %s (last state=%s reason=%s): %w", label, timeout, lastState, lastReason, ErrLBProvisioningTimeout) } if time.Now().After(nextLog) { Step(t, "%s: state=%s, still waiting...", label, lastState) diff --git a/tests/e2e/lb/lb_test.go b/tests/e2e/lb/lb_test.go index f28336167..df9d50f81 100644 --- a/tests/e2e/lb/lb_test.go +++ b/tests/e2e/lb/lb_test.go @@ -861,17 +861,20 @@ func createActiveLB(t *testing.T, c *harness.AWSClient, f *sharedFixture, name, time.Sleep(wait) } lb := createLB(t, c, f, name, lbType, scheme) + // Register idempotent teardown up-front so no exit path (fatal, timeout, + // or success) can leak the LB/listener into the shared VPC. Deletes are + // idempotent server-side, so a retry's explicit teardown below is safe. + t.Cleanup(func() { deleteLB(t, c, lb) }) listener := "" if proto != "" { listener = createListener(t, c, lb.ARN, proto, port, tgArn) + t.Cleanup(func() { deleteListener(t, c, listener) }) } lastErr = harness.WaitForLBActiveErr(t, c, lb.ARN, label, 5*time.Minute) if lastErr == nil { - t.Cleanup(func() { deleteLB(t, c, lb) }) - t.Cleanup(func() { deleteListener(t, c, listener) }) return lb, listener } - if !errors.Is(lastErr, harness.ErrLBTerminalFailed) { + if !errors.Is(lastErr, harness.ErrLBTerminalFailed) && !errors.Is(lastErr, harness.ErrLBProvisioningTimeout) { t.Fatalf("%s: %v", label, lastErr) } t.Logf("%s: attempt %d/%d: %v — tearing down and retrying", label, attempt, lbCreateAttempts, lastErr)