mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-19 19:01:42 +00:00
test: speed up slow spinifex suites (#495)
This commit is contained in:
@@ -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()
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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.<id>) 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)
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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 })
|
||||
}
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user