diff --git a/spinifex/admin/telemetry_test.go b/spinifex/admin/telemetry_test.go index 9a4bacb06..ff2373d13 100644 --- a/spinifex/admin/telemetry_test.go +++ b/spinifex/admin/telemetry_test.go @@ -63,7 +63,9 @@ func TestSendTelemetry_RespectsTimeout(t *testing.T) { t.Parallel() server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - time.Sleep(10 * time.Second) + // Outlive the 100ms client deadline so SendTelemetry aborts on its own + // context, but return quickly enough that server.Close() does not block. + time.Sleep(300 * time.Millisecond) w.WriteHeader(http.StatusNoContent) })) defer server.Close() diff --git a/spinifex/daemon/eks_worker_launch.go b/spinifex/daemon/eks_worker_launch.go index 27730d9d3..a2f2f7b39 100644 --- a/spinifex/daemon/eks_worker_launch.go +++ b/spinifex/daemon/eks_worker_launch.go @@ -22,6 +22,9 @@ import ( // Compile-time check that Daemon satisfies the EKS worker-launch surface. var _ handlers_eks.WorkerLauncher = (*Daemon)(nil) +// terminateRetrySleep is the backoff seam between NoResponders retries; tests override it. +var terminateRetrySleep = time.Sleep + // RunWorkerInstance launches a nodegroup worker on the local node. func (d *Daemon) RunWorkerInstance(ctx context.Context, input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) { return d.RunWorkerInstanceOnNode(ctx, "", input, accountID) @@ -111,7 +114,7 @@ func (d *Daemon) terminateWorkerInstance(ctx context.Context, instanceID, accoun break } if attempt < 2 { - time.Sleep(time.Duration(attempt+1) * time.Second) + terminateRetrySleep(time.Duration(attempt+1) * time.Second) } } if errors.Is(err, nats.ErrNoResponders) { diff --git a/spinifex/daemon/eks_worker_launch_test.go b/spinifex/daemon/eks_worker_launch_test.go index e8c577ef7..f90eb8f3a 100644 --- a/spinifex/daemon/eks_worker_launch_test.go +++ b/spinifex/daemon/eks_worker_launch_test.go @@ -17,10 +17,20 @@ import ( "github.com/stretchr/testify/require" ) +// noopTerminateRetrySleep replaces the terminate NoResponders backoff with a +// no-op for the duration of the test, so retry paths do not burn real seconds. +func noopTerminateRetrySleep(t *testing.T) { + t.Helper() + prev := terminateRetrySleep + terminateRetrySleep = func(time.Duration) {} + t.Cleanup(func() { terminateRetrySleep = prev }) +} + // A worker with no owner (no responder on ec2.cmd.) is already-gone, so // terminating it is an idempotent no-op success (a retried DeleteNodegroup / // scale-down must not wedge on drained instances). func TestTerminateWorkerInstances_NotFoundIsIdempotent(t *testing.T) { + noopTerminateRetrySleep(t) nc, err := nats.Connect(sharedNATSURL) require.NoError(t, err) t.Cleanup(nc.Close) @@ -105,6 +115,7 @@ func TestTerminateWorkerInstances_OwnerErrorSurfaces(t *testing.T) { // a stopped+wedged node's ENI pins the customer subnet/VPC undeletable and // DeleteCluster wedges in DELETING (mulga-siv-475). func TestTerminateWorkerInstances_StoppedFallbackToEC2Terminate(t *testing.T) { + noopTerminateRetrySleep(t) nc, err := nats.Connect(sharedNATSURL) require.NoError(t, err) t.Cleanup(nc.Close) @@ -133,6 +144,7 @@ func TestTerminateWorkerInstances_StoppedFallbackToEC2Terminate(t *testing.T) { // A NotFound payload from the ec2.terminate fallback (worker already drained) is // idempotent success, not a teardown failure. func TestTerminateWorkerInstances_StoppedFallbackNotFoundIdempotent(t *testing.T) { + noopTerminateRetrySleep(t) nc, err := nats.Connect(sharedNATSURL) require.NoError(t, err) t.Cleanup(nc.Close) @@ -150,6 +162,7 @@ func TestTerminateWorkerInstances_StoppedFallbackNotFoundIdempotent(t *testing.T // A non-NotFound error from the ec2.terminate fallback must surface so the // teardown backstop retries rather than orphaning the worker (and its ENI). func TestTerminateWorkerInstances_StoppedFallbackErrorSurfaces(t *testing.T) { + noopTerminateRetrySleep(t) nc, err := nats.Connect(sharedNATSURL) require.NoError(t, err) t.Cleanup(nc.Close) diff --git a/spinifex/gateway/ec2/instance/TerminateInstances.go b/spinifex/gateway/ec2/instance/TerminateInstances.go index 8573320ad..3020d6d2f 100644 --- a/spinifex/gateway/ec2/instance/TerminateInstances.go +++ b/spinifex/gateway/ec2/instance/TerminateInstances.go @@ -20,6 +20,9 @@ type terminateStoppedInstanceRequest struct { InstanceID string `json:"instance_id"` } +// terminateRetrySleep is the backoff seam between NoResponders retries; tests override it. +var terminateRetrySleep = time.Sleep + func ValidateTerminateInstancesInput(input *ec2.TerminateInstancesInput) error { if input == nil { return errors.New(awserrors.ErrorInvalidParameterValue) @@ -75,7 +78,7 @@ func TerminateInstances(ctx context.Context, input *ec2.TerminateInstancesInput, if attempt < 2 { slog.DebugContext(ctx, "TerminateInstances: No responder on per-instance topic, retrying", "instance_id", instanceID, "attempt", attempt+1) - time.Sleep(time.Duration(attempt+1) * time.Second) + terminateRetrySleep(time.Duration(attempt+1) * time.Second) } } if err != nil { diff --git a/spinifex/gateway/ec2/instance/TerminateInstances_test.go b/spinifex/gateway/ec2/instance/TerminateInstances_test.go index 6c8d51ad5..675ae13db 100644 --- a/spinifex/gateway/ec2/instance/TerminateInstances_test.go +++ b/spinifex/gateway/ec2/instance/TerminateInstances_test.go @@ -109,6 +109,7 @@ func TestTerminateInstances_NilInstanceIdSkipped(t *testing.T) { func TestTerminateInstances_NATSRequestFails(t *testing.T) { _, nc := startTestNATSServer(t) + noopTerminateRetrySleep(t) instanceID := "i-nosubscriber" @@ -125,6 +126,7 @@ func TestTerminateInstances_NATSRequestFails(t *testing.T) { func TestTerminateInstances_MixedSuccessAndFailure(t *testing.T) { _, nc := startTestNATSServer(t) + noopTerminateRetrySleep(t) goodID := "i-good" badID := "i-bad" @@ -171,6 +173,7 @@ func TestTerminateInstances_VerifiesQMPAttributes(t *testing.T) { func TestTerminateInstances_StoppedInstanceFallback(t *testing.T) { _, nc := startTestNATSServer(t) + noopTerminateRetrySleep(t) instanceID := "i-stopped-term" @@ -202,6 +205,7 @@ func TestTerminateInstances_StoppedInstanceFallback(t *testing.T) { func TestTerminateInstances_MixedRunningAndStopped(t *testing.T) { _, nc := startTestNATSServer(t) + noopTerminateRetrySleep(t) runningID := "i-running-mix" stoppedID := "i-stopped-mix" diff --git a/spinifex/gateway/ec2/instance/helpers_test.go b/spinifex/gateway/ec2/instance/helpers_test.go index 4171b9143..606e0fdeb 100644 --- a/spinifex/gateway/ec2/instance/helpers_test.go +++ b/spinifex/gateway/ec2/instance/helpers_test.go @@ -2,6 +2,7 @@ package gateway_ec2_instance import ( "testing" + "time" "github.com/mulgadc/spinifex/spinifex/testutil" "github.com/nats-io/nats-server/v2/server" @@ -13,3 +14,12 @@ func startTestNATSServer(t *testing.T) (*server.Server, *nats.Conn) { t.Helper() return testutil.StartTestNATS(t) } + +// noopTerminateRetrySleep replaces the terminate NoResponders backoff with a +// no-op for the duration of the test, so retry paths do not burn real seconds. +func noopTerminateRetrySleep(t *testing.T) { + t.Helper() + prev := terminateRetrySleep + terminateRetrySleep = func(time.Duration) {} + t.Cleanup(func() { terminateRetrySleep = prev }) +} diff --git a/spinifex/gateway/ec2/instance/placement_test.go b/spinifex/gateway/ec2/instance/placement_test.go index a45912f6e..7a001a702 100644 --- a/spinifex/gateway/ec2/instance/placement_test.go +++ b/spinifex/gateway/ec2/instance/placement_test.go @@ -256,6 +256,7 @@ func TestAggregateResults_PartialSuccessMeetsMinCount(t *testing.T) { func TestAggregateResults_PartialFailureBelowMinCount(t *testing.T) { _, nc := startTestNATSServer(t) + noopTerminateRetrySleep(t) results := []nodeLaunchResult{ { @@ -325,6 +326,7 @@ func TestAggregateResults_TimeoutSurfacedNotMasked(t *testing.T) { // without erroring) keeps the generic capacity code. func TestAggregateResults_ShortWithoutNodeErrors(t *testing.T) { _, nc := startTestNATSServer(t) + noopTerminateRetrySleep(t) results := []nodeLaunchResult{ { NodeID: "node-1", diff --git a/spinifex/utils/nats_test.go b/spinifex/utils/nats_test.go index 25a53c6b9..b4fdd984f 100644 --- a/spinifex/utils/nats_test.go +++ b/spinifex/utils/nats_test.go @@ -300,9 +300,9 @@ func TestNATSRequest_Timeout(t *testing.T) { require.NoError(t, err) defer nc.Close() - // Responder that never responds + // Responder that outlives the 100ms client timeout without replying. _, err = nc.QueueSubscribe("test.slow", "q", func(msg *nats.Msg) { - time.Sleep(5 * time.Second) + time.Sleep(500 * time.Millisecond) }) require.NoError(t, err) diff --git a/spinifex/vpcd/vpcd.go b/spinifex/vpcd/vpcd.go index f9c0d1131..6147bfd96 100644 --- a/spinifex/vpcd/vpcd.go +++ b/spinifex/vpcd/vpcd.go @@ -313,10 +313,16 @@ var externalCIDRFromBridge = func(bridge string) (netip.Prefix, error) { return netip.Prefix{}, fmt.Errorf("no IPv4 address on %q", bridge) } +// externalCIDRRetryDelay is the poll interval between resolve attempts; tests shorten it. +var externalCIDRRetryDelay = 500 * time.Millisecond + +// externalCIDRResolveTimeout bounds startup resolution of the WAN bridge CIDR; tests shorten it. +var externalCIDRResolveTimeout = 30 * time.Second + // resolveExternalCIDR blocks until the WAN bridge has an IPv4 address or timeout elapses. // Guards the boot race where vpcd starts before systemd-networkd or netplan assigns the uplink address. func resolveExternalCIDR(ctx context.Context, bridge string, timeout time.Duration) (netip.Prefix, error) { - const retryDelay = 500 * time.Millisecond + retryDelay := externalCIDRRetryDelay deadline := time.Now().Add(timeout) attempt := 0 for { @@ -348,7 +354,7 @@ func ensureExternalCIDRReady(ctx context.Context, externalMode, bridge string) e if externalMode == "" { return nil } - cidr, err := resolveExternalCIDR(ctx, bridge, 30*time.Second) + cidr, err := resolveExternalCIDR(ctx, bridge, externalCIDRResolveTimeout) if err != nil { slog.Error("vpcd: external CIDR resolution failed", "bridge", bridge, "err", err) return err diff --git a/spinifex/vpcd/vpcd_test.go b/spinifex/vpcd/vpcd_test.go index 986209594..a9044c00e 100644 --- a/spinifex/vpcd/vpcd_test.go +++ b/spinifex/vpcd/vpcd_test.go @@ -360,6 +360,20 @@ type externalCIDRResponse struct { err error } +// stubExternalCIDRTiming shortens the resolve retry delay and startup timeout to +// a few milliseconds so retry/timeout paths do not burn real wall-clock seconds. +func stubExternalCIDRTiming(t *testing.T) { + t.Helper() + origDelay := externalCIDRRetryDelay + origTimeout := externalCIDRResolveTimeout + t.Cleanup(func() { + externalCIDRRetryDelay = origDelay + externalCIDRResolveTimeout = origTimeout + }) + externalCIDRRetryDelay = time.Millisecond + externalCIDRResolveTimeout = 20 * time.Millisecond +} + func TestResolveExternalCIDR_ImmediateSuccess(t *testing.T) { want := netip.MustParsePrefix("192.168.1.10/24") stubExternalCIDR(t, []externalCIDRResponse{{prefix: want}}) @@ -373,6 +387,7 @@ func TestResolveExternalCIDR_ImmediateSuccess(t *testing.T) { } func TestResolveExternalCIDR_RetriesThenSucceeds(t *testing.T) { + stubExternalCIDRTiming(t) want := netip.MustParsePrefix("10.0.0.5/16") stubExternalCIDR(t, []externalCIDRResponse{ {err: fmt.Errorf("no IPv4")}, @@ -389,8 +404,9 @@ func TestResolveExternalCIDR_RetriesThenSucceeds(t *testing.T) { } func TestResolveExternalCIDR_TimeoutFailsStart(t *testing.T) { + stubExternalCIDRTiming(t) stubExternalCIDR(t, []externalCIDRResponse{{err: fmt.Errorf("no IPv4")}}) - _, err := resolveExternalCIDR(context.Background(), "br-wan", 100*time.Millisecond) + _, err := resolveExternalCIDR(context.Background(), "br-wan", 10*time.Millisecond) if err == nil { t.Fatal("expected timeout error") } @@ -452,6 +468,7 @@ func TestEnsureExternalCIDRReady_ResolvesSuccessfully(t *testing.T) { } func TestEnsureExternalCIDRReady_PropagatesError(t *testing.T) { + stubExternalCIDRTiming(t) stubExternalCIDR(t, []externalCIDRResponse{{err: fmt.Errorf("no IPv4")}}) err := ensureExternalCIDRReady(context.Background(), "direct", "br-wan") if err == nil {