diff --git a/spinifex/daemon/cluster_ready_test.go b/spinifex/daemon/cluster_ready_test.go index eb9ea479a..4bf28d932 100644 --- a/spinifex/daemon/cluster_ready_test.go +++ b/spinifex/daemon/cluster_ready_test.go @@ -62,7 +62,10 @@ func closedPort(t *testing.T) string { // TestWaitForClusterReady_ReadyImmediately asserts the loop returns on the first // pass without sleeping when both dependencies are already reachable. func TestWaitForClusterReady_ReadyImmediately(t *testing.T) { - _, nc := testutil.StartTestNATS(t) + // JetStream, not a bare NATS server: viperblock readiness means the KV the + // mount claims its volume lease from can answer, and a connection to a + // server whose JetStream is still starting reports connected regardless. + _, nc, _ := testutil.StartTestJetStream(t) sleeps := 0 stubClusterReadySleep(t, func(time.Duration) { sleeps++ }) @@ -79,7 +82,7 @@ func TestWaitForClusterReady_ReadyImmediately(t *testing.T) { // while predastore is unreachable and returns once it starts serving, rather // than latching the first not-ready result. func TestWaitForClusterReady_PredastoreBecomesReady(t *testing.T) { - _, nc := testutil.StartTestNATS(t) + _, nc, _ := testutil.StartTestJetStream(t) srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusForbidden) @@ -108,6 +111,31 @@ func TestWaitForClusterReady_PredastoreBecomesReady(t *testing.T) { assert.NotContains(t, logs.String(), "Cluster readiness timeout") } +// TestCheckViperblockReady_ConnectedButJetStreamNotServing is the regression +// guard for a guest lost across a cluster update. A TCP connection to NATS says +// nothing about whether JetStream can serve, so readiness used to pass while the +// volume-lease KV was still unavailable, recovery ran into a lease timeout, and +// the instance was parked in error permanently. +func TestCheckViperblockReady_ConnectedButJetStreamNotServing(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + require.True(t, nc.IsConnected(), "the connection itself must be up, or this proves nothing") + + d := &Daemon{config: &config.Config{}, natsConn: nc} + + assert.False(t, d.checkViperblockReady(), + "connected to a server with no JetStream is not ready: the lease KV cannot answer") +} + +// TestCheckViperblockReady_JetStreamServing is the other half: once JetStream +// answers, readiness must pass rather than holding startup open. +func TestCheckViperblockReady_JetStreamServing(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + + d := &Daemon{config: &config.Config{}, natsConn: nc} + + assert.True(t, d.checkViperblockReady()) +} + // TestWaitForClusterReady_TimesOut asserts the deadline branch: with viperblock // permanently unreachable the loop gives up at maxWait, logs the warning and // returns so recovery proceeds anyway, instead of blocking startup forever. diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 0877859f6..8d9c5d733 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -2350,12 +2350,33 @@ func (d *Daemon) waitForClusterReady() { slog.Warn("Cluster readiness timeout, proceeding with recovery anyway", "max_wait_ms", otelsetup.Millis(maxWait)) } -// checkViperblockReady reports whether viperblock is reachable via NATS. +// checkViperblockReady reports whether viperblock can actually serve a mount, +// which needs more than a live NATS socket: mounting claims a JetStream KV +// lease before it reads any state, so JetStream must be serving too. +// +// Testing only IsConnected was why waitForClusterReady did not prevent the +// race it exists to prevent. The client reconnects as soon as nats-server +// accepts TCP, which is well before JetStream has quorum, so readiness passed +// roughly 30 seconds before the first lease claim could have succeeded — and +// every guest recovering in that window was failed and latched. +// +// AccountInfo is the probe because it is the cheapest round trip that proves +// JetStream is answering, and it changes nothing. func (d *Daemon) checkViperblockReady() bool { - if d.natsConn == nil { + if d.natsConn == nil || !d.natsConn.IsConnected() { return false } - return d.natsConn.IsConnected() + + js, err := jetstream.New(d.natsConn) + if err != nil { + return false + } + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + _, err = js.AccountInfo(ctx) + return err == nil } // predastoreReadinessClient is hoisted to package scope so the 2s readiness @@ -2414,11 +2435,29 @@ func (d *Daemon) LoadState() error { // restoreInstances delegates to vm.Manager.Restore and syncs the local state // file so it matches in-memory state. +// +// It then starts the recovery retry loop in the background. Restore is a single +// pass: any instance whose relaunch lost a race with a still-starting dependency +// is parked in StateError and, before this, stayed there until an operator +// noticed. The loop reconsiders exactly those, once the backing store is +// confirmed ready, and returns as soon as there is nothing left to retry. +// +// Backgrounded because it must not hold up the rest of daemon start: the guests +// it is recovering are already down, and blocking here would keep the API from +// coming up while it waits. func (d *Daemon) restoreInstances() error { d.vmMgr.Restore() if err := d.WriteState(); err != nil { slog.Error("Failed to persist local state after restore", "error", err) } + + go func() { + d.vmMgr.RunRecoveryRetryLoop(d.shuttingDown.Load) + if err := d.WriteState(); err != nil { + slog.Error("Failed to persist local state after recovery retry", "error", err) + } + }() + return nil } diff --git a/spinifex/services/viperblockd/provider_handlers.go b/spinifex/services/viperblockd/provider_handlers.go index 17ac9938d..c606e3548 100644 --- a/spinifex/services/viperblockd/provider_handlers.go +++ b/spinifex/services/viperblockd/provider_handlers.go @@ -1327,10 +1327,17 @@ func constructMountedVB(ctx context.Context, cfg *Config, volumeName string) (*v // mountErrRetryable reports whether a mount failure was caused by the // backing store not yet being ready (a transient state-load gap) rather -// than a permanent condition. Only these two sentinels qualify for the +// than a permanent condition. Only these sentinels qualify for the // recovery-relaunch retry in vm.Manager. +// +// ErrLeaseStoreUnavailable belongs here for the same reason the other two do, +// and its absence is what made a routine cluster update destructive: the lease +// is claimed before any state is read, so a node coming up alongside NATS fails +// there first and never reaches the conditions these sentinels describe. func mountErrRetryable(err error) bool { - return errors.Is(err, viperblock.ErrStateNotFound) || errors.Is(err, viperblock.ErrStateBackendUnavailable) + return errors.Is(err, viperblock.ErrStateNotFound) || + errors.Is(err, viperblock.ErrStateBackendUnavailable) || + errors.Is(err, ErrLeaseStoreUnavailable) } // mountVolume is launchService's ebs.mount body, extracted so the legacy diff --git a/spinifex/services/viperblockd/provider_handlers_test.go b/spinifex/services/viperblockd/provider_handlers_test.go index b3c4f67e9..38f55106d 100644 --- a/spinifex/services/viperblockd/provider_handlers_test.go +++ b/spinifex/services/viperblockd/provider_handlers_test.go @@ -33,9 +33,12 @@ import ( "github.com/stretchr/testify/require" ) -// TestMountErrRetryable locks down the classifier relaunchAll's -// recovery-retry relies on: only the two viperblock state-load sentinels -// count as a transient, retryable mount failure. +// TestMountErrRetryable locks down the classifier relaunchAll's recovery-retry +// relies on: a mount that failed because a dependency was still starting is +// retryable, and nothing else is. A guest lost across a cluster update because +// the lease-store cases below were missing — the mount failed on a JetStream +// that had not finished coming up, was reported as permanent, and the existing +// backoff never engaged. func TestMountErrRetryable(t *testing.T) { tests := []struct { name string @@ -46,7 +49,14 @@ func TestMountErrRetryable(t *testing.T) { {"wrapped ErrStateNotFound", fmt.Errorf("state present but BlockSize=0: %w", viperblock.ErrStateNotFound), true}, {"ErrStateBackendUnavailable", viperblock.ErrStateBackendUnavailable, true}, {"wrapped ErrStateBackendUnavailable", fmt.Errorf("LoadState exhausted 5 retries: %w", viperblock.ErrStateBackendUnavailable), true}, + {"ErrLeaseStoreUnavailable", ErrLeaseStoreUnavailable, true}, + { + "lease claim against a JetStream that is still starting", + fmt.Errorf("claim lease for vol-1: %w: %w", context.DeadlineExceeded, ErrLeaseStoreUnavailable), + true, + }, {"plain error", errors.New("some other mount failure"), false}, + {"lease held by another node is not retryable", errVolumeLeaseHeld, false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { diff --git a/spinifex/services/viperblockd/volume_lease.go b/spinifex/services/viperblockd/volume_lease.go index 0b3561e2d..ef8e9e8bd 100644 --- a/spinifex/services/viperblockd/volume_lease.go +++ b/spinifex/services/viperblockd/volume_lease.go @@ -20,6 +20,35 @@ import ( // exclusion rather than a record of it. const volumeLeaseBucket = "VIPERBLOCK_VOLUME_LEASES" +// ErrLeaseStoreUnavailable marks a lease claim that failed because JetStream +// could not be reached, as distinct from one that failed on its merits. The +// two are opposite conclusions: unreachable means try again shortly, whereas +// a refused claim means somebody else holds the volume. +// +// It exists because they were previously indistinguishable to the caller. A +// node restarting alongside NATS claims a lease before JetStream is serving, +// gets a deadline, and the mount is reported permanent — so a guest that only +// needed a few more seconds is failed and latched out of recovery instead. +var ErrLeaseStoreUnavailable = errors.New("volume lease store unavailable") + +// leaseStoreUnavailable reports whether err means the lease store could not be +// reached at all. Deliberately a closed list of transport failures: anything +// unrecognised stays permanent, because retrying a claim that was genuinely +// refused would keep a volume wedged behind a node that cannot have it. +func leaseStoreUnavailable(err error) bool { + switch { + case errors.Is(err, context.DeadlineExceeded), + errors.Is(err, nats.ErrNoResponders), + errors.Is(err, nats.ErrTimeout), + errors.Is(err, nats.ErrConnectionClosed), + errors.Is(err, nats.ErrConnectionDraining), + errors.Is(err, nats.ErrNoServers), + errors.Is(err, jetstream.ErrBucketNotFound): + return true + } + return false +} + // volumeLeaseTTL is how long a lease outlives the node holding it. A node that // dies mid-mount leaves its entry behind and nothing else may open the volume // until the entry ages out. @@ -174,6 +203,9 @@ func (l *volumeLeases) acquire(ctx context.Context, volumeName string) (*volumeL return nil, err } case err != nil: + if leaseStoreUnavailable(err) { + return nil, fmt.Errorf("claim lease for %s: %w: %w", volumeName, err, ErrLeaseStoreUnavailable) + } return nil, fmt.Errorf("claim lease for %s: %w", volumeName, err) } diff --git a/spinifex/services/viperblockd/volume_lease_test.go b/spinifex/services/viperblockd/volume_lease_test.go index b6c3ea025..292c69327 100644 --- a/spinifex/services/viperblockd/volume_lease_test.go +++ b/spinifex/services/viperblockd/volume_lease_test.go @@ -2,6 +2,7 @@ package viperblockd import ( "context" + "errors" "fmt" "os" "testing" @@ -307,3 +308,61 @@ func TestVolumeLease_SurrenderedLeaseIsNotDeletedOnRelease(t *testing.T) { _, err = leases.kv.Get(t.Context(), volumeName) require.NoError(t, err, "releasing a surrendered lease must leave the entry alone: it may be the successor's now") } + +// TestLeaseStoreUnavailable separates "the store could not answer" from "the +// store answered no". Only the first is worth retrying, and getting it wrong in +// either direction is expensive: treating a held lease as transient would retry +// into a second engine on one encrypted volume, and treating a timeout as +// permanent strands a guest across a cluster restart, which is what happened. +func TestLeaseStoreUnavailable(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {"context deadline on a starting JetStream", context.DeadlineExceeded, true}, + {"wrapped deadline", fmt.Errorf("kv update: %w", context.DeadlineExceeded), true}, + {"no responders", nats.ErrNoResponders, true}, + {"request timeout", nats.ErrTimeout, true}, + {"connection closed", nats.ErrConnectionClosed, true}, + {"connection draining", nats.ErrConnectionDraining, true}, + {"no servers", nats.ErrNoServers, true}, + {"bucket not yet created", jetstream.ErrBucketNotFound, true}, + + // The store was reachable and said no. Retrying cannot change it. + {"key already exists", jetstream.ErrKeyExists, false}, + {"lease held by another owner", errVolumeLeaseHeld, false}, + {"unrelated failure", errors.New("marshal lease record"), false}, + {"nil", nil, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, leaseStoreUnavailable(tt.err)) + }) + } +} + +// TestVolumeLease_UnreachableStoreIsRetryable is the end of the chain that +// matters: an acquire against a store that cannot answer must surface as +// ErrLeaseStoreUnavailable, because that is the sentinel mountErrRetryable +// looks for and the only reason the relaunch backoff engages. +func TestVolumeLease_UnreachableStoreIsRetryable(t *testing.T) { + _, natsURL := setupEmbeddedNATS(t) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + t.Cleanup(nc.Close) + + leases, err := newVolumeLeases(t.Context(), nc, "node-a") + require.NoError(t, err) + + // Close the connection under the bound store, which is the shape of a + // daemon that came back before NATS finished starting. + nc.Close() + + _, err = leases.acquire(t.Context(), "vol-storedown1") + require.Error(t, err) + assert.ErrorIs(t, err, ErrLeaseStoreUnavailable, + "a store that cannot answer must be reported as retryable, not as a refusal") + assert.NotErrorIs(t, err, errVolumeLeaseHeld) +} diff --git a/spinifex/vm/recovery_retry.go b/spinifex/vm/recovery_retry.go new file mode 100644 index 000000000..737a30346 --- /dev/null +++ b/spinifex/vm/recovery_retry.go @@ -0,0 +1,152 @@ +package vm + +import ( + "log/slog" + "time" +) + +// Recovery-failed instances used to stay in StateError until an operator +// intervened, on the reasoning that a failed relaunch needs a human. That holds +// when the relaunch failed on its merits. It does not hold when it failed +// because a dependency was still starting, which is the common case on a +// coordinated cluster update: the daemon comes back before JetStream is serving, +// the mount cannot claim its lease, and a guest that needed a few more seconds +// is parked indefinitely. +// +// So the retry below is deliberately narrow. It reconsiders only the failures a +// not-yet-ready backing store produces, only once the store is confirmed ready, +// and only a bounded number of times — then it stops and leaves the instance for +// an operator, which is the right answer for a failure that is actually real. + +// retryableRecoveryReasons are the MarkRecoveryFailed reasons that describe a +// dependency being unavailable rather than an instance being unlaunchable. +// Reasons absent from this set are never retried automatically: reconnect_failed +// and pre_relaunch_hook_failed can both indicate a guest whose state needs +// looking at, and relaunching those on a timer would hide the problem. +var retryableRecoveryReasons = map[string]bool{ + "recovery_mount_state_unavailable": true, + "recovery_launch_failed": true, +} + +// recoveryRetryMaxPasses bounds how many times the loop reconsiders a given +// instance. Five passes at the interval below is a few minutes of trying, which +// covers a slow cluster start without turning a genuinely broken instance into a +// relaunch loop that never stops. +const recoveryRetryMaxPasses = 5 + +// recoveryRetryInterval is the gap between passes. Package var so tests do not +// sleep real minutes. +var recoveryRetryInterval = 30 * time.Second + +// RetryRecoveryFailed reconsiders instances parked in StateError by a +// dependency-unavailable failure and relaunches them. It is a no-op unless the +// backing store is ready, because retrying against a store that is still down is +// how they got here. +// +// Returns the number of instances it attempted, so a caller can stop polling +// once there is nothing left to do. +func (m *Manager) RetryRecoveryFailed() int { + if !m.backingStoreReady() { + slog.Debug("recovery retry: backing store not ready, deferring") + return 0 + } + + var toRetry []*VM + + for _, instance := range m.Snapshot() { + if m.Status(instance) != StateError { + continue + } + + reason, passes := recoveryFailureReason(instance) + if !retryableRecoveryReasons[reason] { + continue + } + if passes >= recoveryRetryMaxPasses { + continue + } + + // StateError has no outgoing edge in the transition map, so this + // mirrors what classifyRestoredInstances already does for a + // drain-stopped instance: assign the field directly rather than + // widening the state machine for every other caller. + m.UpdateState(instance.ID, func(v *VM) { + v.Status = StatePending + v.recoveryRetries++ + }) + + slog.Info("Retrying recovery-failed instance now the backing store is ready", + "instanceId", instance.ID, "reason", reason, "pass", passes+1) + toRetry = append(toRetry, instance) + } + + if len(toRetry) == 0 { + return 0 + } + + m.relaunchAll(toRetry) + return len(toRetry) +} + +// recoveryFailureReason returns the MarkRecoveryFailed reason recorded on the +// instance and how many automatic retries it has already had. +// A VM with no Instance has no recorded reason, so it is never retryable: the +// nil is normal for an entry that failed before its EC2 view was built. +func recoveryFailureReason(instance *VM) (string, int) { + var reason string + if instance.Instance != nil && instance.Instance.StateReason != nil && + instance.Instance.StateReason.Message != nil { + reason = *instance.Instance.StateReason.Message + } + return reason, instance.recoveryRetries +} + +// RunRecoveryRetryLoop retries dependency-failed instances until none are left +// or the passes are exhausted, then returns. It is not a permanent background +// loop: once the store is up and the backlog is cleared there is nothing for it +// to do, and a loop that runs forever is one more thing to reason about during +// an incident. +// +// stop is checked before every pass so a coordinated shutdown is never delayed. +func (m *Manager) RunRecoveryRetryLoop(stop func() bool) { + for pass := 1; pass <= recoveryRetryMaxPasses; pass++ { + if stop != nil && stop() { + slog.Info("recovery retry loop: aborted by shutdown") + return + } + + if n := m.RetryRecoveryFailed(); n > 0 { + slog.Info("recovery retry pass complete", "pass", pass, "attempted", n) + } + + if !m.hasRetryableRecoveryFailures() { + slog.Debug("recovery retry loop: nothing left to retry") + return + } + + if stop != nil && stop() { + return + } + time.Sleep(recoveryRetryInterval) + } + + if m.hasRetryableRecoveryFailures() { + slog.Warn("recovery retry loop: giving up, instances remain in error for operator action", + "maxPasses", recoveryRetryMaxPasses) + } +} + +// hasRetryableRecoveryFailures reports whether any instance is still parked in +// StateError with a retryable reason and retries remaining. +func (m *Manager) hasRetryableRecoveryFailures() bool { + for _, instance := range m.Snapshot() { + if m.Status(instance) != StateError { + continue + } + reason, passes := recoveryFailureReason(instance) + if retryableRecoveryReasons[reason] && passes < recoveryRetryMaxPasses { + return true + } + } + return false +} diff --git a/spinifex/vm/recovery_retry_test.go b/spinifex/vm/recovery_retry_test.go new file mode 100644 index 000000000..c4173a745 --- /dev/null +++ b/spinifex/vm/recovery_retry_test.go @@ -0,0 +1,219 @@ +//test:in-package — the retry policy is unexported (recoveryRetryMaxPasses, +//recoveryRetryInterval, hasRetryableRecoveryFailures) and the tests drive it +//through the same seams restore_test.go uses: relaunchTestManager, the +//runForRelaunch stub, and VM.recoveryRetries. + +package vm + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// A guest drain-stopped by a cluster update used to be lost for good if its +// relaunch lost a race with a still-starting dependency: MarkRecoveryFailed +// parked it in StateError and nothing ever reconsidered it. These tests cover +// the retry that now does, and the limits on it — an automatic relaunch that +// cannot tell "the store was down" from "this guest will not start" would turn +// one broken instance into a permanent relaunch loop. + +// erroredInstance builds a VM parked in StateError by MarkRecoveryFailed with +// the given reason, which is how the retry decides eligibility. +func erroredInstance(id, reason string) *VM { + return &VM{ + ID: id, + Status: StateError, + InstanceType: "t3.micro", + Instance: &ec2.Instance{ + StateReason: &ec2.StateReason{ + Code: aws.String("Server.RecoveryFailed"), + Message: aws.String(reason), + }, + }, + } +} + +func TestRetryRecoveryFailed(t *testing.T) { + t.Run("relaunches an instance failed by an unavailable backing store", func(t *testing.T) { + m, _, _, _ := relaunchTestManager(t) + m.deps.BackingStoreReady = func() bool { return true } + + origRun := runForRelaunch + t.Cleanup(func() { runForRelaunch = origRun }) + var calls atomic.Int64 + runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error { + calls.Add(1) + return nil + } + + inst := erroredInstance("i-retry-me", "recovery_mount_state_unavailable") + m.Insert(inst) + + attempted := m.RetryRecoveryFailed() + + assert.Equal(t, 1, attempted, "the instance is eligible and must be retried") + assert.EqualValues(t, 1, calls.Load(), "Run must be called for the retried instance") + assert.NotEqual(t, StateError, m.Status(inst), + "a successful retry must take the instance out of StateError") + }) + + t.Run("does nothing while the backing store is still down", func(t *testing.T) { + m, _, _, _ := relaunchTestManager(t) + m.deps.BackingStoreReady = func() bool { return false } + + origRun := runForRelaunch + t.Cleanup(func() { runForRelaunch = origRun }) + var calls atomic.Int64 + runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error { + calls.Add(1) + return nil + } + + inst := erroredInstance("i-store-down", "recovery_launch_failed") + m.Insert(inst) + + assert.Zero(t, m.RetryRecoveryFailed(), + "retrying against a store that is still down is how the instance got here") + assert.Zero(t, calls.Load(), "Run must not be called") + assert.Equal(t, StateError, m.Status(inst), "the instance must stay parked") + }) + + t.Run("leaves failures that are not dependency-related alone", func(t *testing.T) { + m, _, _, _ := relaunchTestManager(t) + m.deps.BackingStoreReady = func() bool { return true } + + origRun := runForRelaunch + t.Cleanup(func() { runForRelaunch = origRun }) + var calls atomic.Int64 + runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error { + calls.Add(1) + return nil + } + + // reconnect_failed and pre_relaunch_hook_failed can both mean the + // guest itself needs looking at. Relaunching those on a timer would + // hide the problem rather than fix it. + for _, reason := range []string{"reconnect_failed", "pre_relaunch_hook_failed", ""} { + m.Insert(erroredInstance("i-"+reason+"-x", reason)) + } + + // An entry that failed before its EC2 view was built has no reason at + // all. That is ordinary, not a crash. + m.Insert(&VM{ID: "i-no-ec2-view", Status: StateError}) + + assert.Zero(t, m.RetryRecoveryFailed(), + "only dependency-unavailable reasons are retried automatically") + assert.Zero(t, calls.Load()) + assert.False(t, m.hasRetryableRecoveryFailures()) + }) + + t.Run("gives up after the retry budget and leaves the instance for an operator", func(t *testing.T) { + m, _, _, _ := relaunchTestManager(t) + m.deps.BackingStoreReady = func() bool { return true } + + origRun := runForRelaunch + t.Cleanup(func() { runForRelaunch = origRun }) + // Not ErrMountRetryable, so relaunchWithRetry returns on the first + // attempt and every pass re-parks the instance in StateError. + runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error { + return assert.AnError + } + + inst := erroredInstance("i-never-starts", "recovery_launch_failed") + m.Insert(inst) + + for pass := 1; pass <= recoveryRetryMaxPasses; pass++ { + require.Equal(t, 1, m.RetryRecoveryFailed(), + "pass %d is within budget and must still be attempted", pass) + require.Equal(t, StateError, m.Status(inst), + "a permanently failing instance must be re-parked each pass") + } + + assert.Zero(t, m.RetryRecoveryFailed(), + "the budget is exhausted, so the instance is left for an operator rather than relaunched forever") + assert.False(t, m.hasRetryableRecoveryFailures(), + "an instance past its budget must not keep the retry loop alive") + }) + + t.Run("running and stopped instances are never touched", func(t *testing.T) { + m, _, _, _ := relaunchTestManager(t) + m.deps.BackingStoreReady = func() bool { return true } + + origRun := runForRelaunch + t.Cleanup(func() { runForRelaunch = origRun }) + var calls atomic.Int64 + runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error { + calls.Add(1) + return nil + } + + for _, st := range []InstanceState{StateRunning, StateStopped, StatePending, StateTerminated} { + m.Insert(&VM{ID: "i-" + string(st), Status: st, Instance: &ec2.Instance{}}) + } + + assert.Zero(t, m.RetryRecoveryFailed(), "only StateError instances are candidates") + assert.Zero(t, calls.Load()) + }) +} + +func TestRunRecoveryRetryLoop(t *testing.T) { + t.Run("returns as soon as there is nothing left to retry", func(t *testing.T) { + m, _, _, _ := relaunchTestManager(t) + m.deps.BackingStoreReady = func() bool { return true } + + origInterval := recoveryRetryInterval + recoveryRetryInterval = time.Millisecond + t.Cleanup(func() { recoveryRetryInterval = origInterval }) + + origRun := runForRelaunch + t.Cleanup(func() { runForRelaunch = origRun }) + runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error { return nil } + + m.Insert(erroredInstance("i-recovers", "recovery_mount_state_unavailable")) + + done := make(chan struct{}) + go func() { m.RunRecoveryRetryLoop(nil); close(done) }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("loop did not return once the backlog was cleared") + } + }) + + t.Run("aborts promptly on shutdown instead of sleeping out its passes", func(t *testing.T) { + m, _, _, _ := relaunchTestManager(t) + m.deps.BackingStoreReady = func() bool { return true } + + origInterval := recoveryRetryInterval + recoveryRetryInterval = time.Hour // a pass that sleeps would hang the test + t.Cleanup(func() { recoveryRetryInterval = origInterval }) + + origRun := runForRelaunch + t.Cleanup(func() { runForRelaunch = origRun }) + var calls atomic.Int64 + runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error { + calls.Add(1) + return assert.AnError + } + + m.Insert(erroredInstance("i-shutting-down", "recovery_launch_failed")) + + done := make(chan struct{}) + go func() { m.RunRecoveryRetryLoop(func() bool { return true }); close(done) }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("shutdown must abort the loop before it sleeps") + } + assert.Zero(t, calls.Load(), "no relaunch may start once shutdown is signalled") + }) +} diff --git a/spinifex/vm/vm.go b/spinifex/vm/vm.go index 9490ae77d..72ee83fe5 100644 --- a/spinifex/vm/vm.go +++ b/spinifex/vm/vm.go @@ -113,6 +113,12 @@ type VM struct { // Set when ownership is released on stop for shared KV storage. LastNode string `json:"last_node,omitempty"` + // recoveryRetries counts automatic retries of a recovery failure caused by + // an unavailable dependency. Unexported and unpersisted on purpose: the + // budget is per daemon run, so a node that restarts gets a fresh one, which + // is right because a restart is new information about the dependency. + recoveryRetries int + // AZ is the availability zone of the node that last ran this instance, // stamped by that node from its own config. Placement observed by the // owner, not something an operator requests. diff --git a/tests/e2e/harness/multinode_cluster_restart.go b/tests/e2e/harness/multinode_cluster_restart.go new file mode 100644 index 000000000..01f792935 --- /dev/null +++ b/tests/e2e/harness/multinode_cluster_restart.go @@ -0,0 +1,201 @@ +//go:build e2e + +package harness + +import ( + "context" + "fmt" + "strings" + "testing" + "time" +) + +// A coordinated cluster shutdown followed by a start is the operation every +// software update performs, and it is the one path that had no e2e coverage: +// StopNode/StartNode simulate a hard outage, which deliberately leaves guests +// running, so nothing exercised a guest that was drained to stopped, had its +// volumes sealed, and then had to relaunch from that sealed state against +// dependencies that were themselves still starting. A guest was lost that way. + +// clusterShutdownPhaseTimeout is passed to `spx admin cluster shutdown` as its +// per-phase budget. The default is 120s; DRAIN legitimately takes longer when +// several guests unmount at once, and a drain that times out is reported as +// success, so allowing the time is cheaper than the ambiguity. +const clusterShutdownPhaseTimeout = 10 * time.Minute + +// spxProcessDrainTimeout bounds the wait for every spx process to exit after +// the units are stopped. Stopping the target returns once the target is +// inactive, which is not the same as every service having exited. +const spxProcessDrainTimeout = 3 * time.Minute + +// ClusterShutdownAndRestart performs the full operator shutdown sequence and +// brings the cluster back: the coordinated phased shutdown, a stop of +// spinifex.target on every node, a confirmed wait for every spx process to exit +// and for systemd to have no spinifex job left queued, then a start of the +// target everywhere. It returns once NATS has reformed and every gateway +// answers, so a caller can immediately assert on guest state. +// +// The spx-process wait is the step that makes this a real restart rather than a +// systemctl call that returned early. spinifex-shutdown.service is ordered +// After= the storage services, so its drain runs while they are still up and +// their stop jobs queue behind it; `systemctl stop` returns as soon as the +// target is inactive, leaving predastore, viperblock and NATS running. +func ClusterShutdownAndRestart(t *testing.T, c *Cluster, env *Env) { + t.Helper() + + if len(c.Nodes) < 2 { + t.Fatalf("cluster restart needs a multi-node cluster, have %d", len(c.Nodes)) + } + + ClusterShutdown(t, c) + + Step(t, "start spinifex.target on all %d nodes", len(c.Nodes)) + for _, n := range c.Nodes { + StartNode(t, n) + } + + // WaitNATSPeers counts remote peers, so a whole cluster is one less than + // the node count. + Step(t, "wait NATS to reform to %d peers", len(c.Nodes)-1) + c.WaitNATSPeers(t, len(c.Nodes)-1, WithTimeout(3*time.Minute), WithPoll(2*time.Second)) + + Step(t, "wait every gateway to answer") + c.WaitGatewayHealthy(t, WithTimeout(3*time.Minute), WithPoll(2*time.Second)) + + if env != nil { + Step(t, "wait every daemon ready") + c.WaitDaemonReady(t, env, WithTimeout(3*time.Minute), WithPoll(2*time.Second)) + } +} + +// ClusterShutdown runs the coordinated shutdown and leaves the cluster down. +// Split out from ClusterShutdownAndRestart so a caller can assert on the +// stopped state, and so a t.Cleanup can restart unconditionally. +func ClusterShutdown(t *testing.T, c *Cluster) { + t.Helper() + + ssh := NewPeerSSH() + leader := c.Nodes[0] + + Step(t, "spx admin cluster shutdown via %s", leader.Name) + ctx, cancel := context.WithTimeout(context.Background(), clusterShutdownPhaseTimeout+2*time.Minute) + out, err := ssh.Run(ctx, leader.Addr, fmt.Sprintf( + "sudo spx admin cluster shutdown --timeout %s --config /etc/spinifex/spinifex.toml", + clusterShutdownPhaseTimeout)) + cancel() + Detail(t, "cluster_shutdown", string(out)) + if err != nil { + // The phases can complete and still lose the SSH channel as NATS goes + // down under it. The per-unit stop below is the authoritative teardown. + t.Logf("cluster shutdown on %s reported %v — proceeding to the per-unit stop", leader.Name, err) + } + + // StopNode is not used here. It stops the service units directly and leaves + // spinifex.target active, which is right for the hard outage it simulates + // and wrong for a real shutdown: the target then reads as satisfied while + // its units are dead, and a later start can race a stop job that is still + // queued. Exactly that lost node3's NATS on the first run — stopped and + // never started again, so the cluster reformed to 2 peers of 3. + Step(t, "stop spinifex.target on all %d nodes", len(c.Nodes)) + for _, n := range c.Nodes { + stopTarget(t, n) + } + + Step(t, "confirm no spx process survives on any node") + for _, n := range c.Nodes { + waitNoSpxProcess(t, n) + } + + Step(t, "confirm systemd has no spinifex jobs still queued") + for _, n := range c.Nodes { + waitNoPendingJobs(t, n) + } +} + +// stopTarget stops spinifex.target, which is what the operator procedure runs +// and what update-nodes.sh runs. Non-fatal: the teardown can racily drop the +// SSH channel, and the waits below are the authoritative gate. +func stopTarget(t *testing.T, node Node) { + t.Helper() + + ssh := NewPeerSSH() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + if _, err := ssh.Run(ctx, node.Addr, "sudo systemctl stop spinifex.target"); err != nil { + t.Logf("stop spinifex.target on %s: %v (proceeding — the waits are the gate)", node.Name, err) + } +} + +// waitNoPendingJobs blocks until systemd has no queued job for any spinifex +// unit. Starting the target with a stop still queued is how a service ends up +// stopped by a job that completes after the start was issued. +func waitNoPendingJobs(t *testing.T, node Node) { + t.Helper() + + ssh := NewPeerSSH() + EventuallyErr(t, func() error { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + out, err := ssh.Run(ctx, node.Addr, + "systemctl list-jobs --no-legend --no-pager | grep -c spinifex || true") + if err != nil { + return fmt.Errorf("%s list-jobs: %w", node.Name, err) + } + if n := strings.TrimSpace(string(out)); n != "0" { + return fmt.Errorf("%s still has %s queued spinifex job(s)", node.Name, n) + } + return nil + }, 2*time.Minute, 2*time.Second) +} + +// stopUnitsHoldingSpx stops whatever units are still running the spx binary, +// found from the processes themselves rather than from a list. +// +// Every service is the same binary — `spx service start` — so the set +// that has to stop is "whoever is still running it", and a hard-coded list goes +// stale silently the moment a service is added. StopNode's list omits +// spinifex-northstar and spinifex-qmp-collector, which is harmless for the hard +// outage it simulates and not harmless here. +func stopUnitsHoldingSpx(t *testing.T, node Node) { + t.Helper() + + ssh := NewPeerSSH() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + + // The unit name comes from each survivor's cgroup, so this needs no + // knowledge of which units exist. + const cmd = `for pid in $(pgrep -x spx || true); do + unit=$(grep -oE 'spinifex-[a-z0-9-]+\.service' "/proc/$pid/cgroup" 2>/dev/null | head -1) + [ -n "$unit" ] && sudo systemctl stop "$unit" +done; true` + if _, err := ssh.Run(ctx, node.Addr, cmd); err != nil { + t.Logf("stopUnitsHoldingSpx %s: %v (the wait below is the gate)", node.Name, err) + } +} + +// waitNoSpxProcess blocks until `pgrep -x spx` finds nothing on node. A +// surviving process means the stop returned before the service did, and +// anything asserted after that is measuring the old process. +func waitNoSpxProcess(t *testing.T, node Node) { + t.Helper() + + ssh := NewPeerSSH() + EventuallyErr(t, func() error { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + // pgrep exits 1 when nothing matches, which is the outcome wanted, so + // the count is read from stdout rather than from the exit status. + out, err := ssh.Run(ctx, node.Addr, "pgrep -x spx | wc -l") + if err != nil { + return fmt.Errorf("%s pgrep spx: %w", node.Name, err) + } + if n := strings.TrimSpace(string(out)); n != "0" { + // A survivor is a unit that was never asked to stop, or one that + // systemd restarted. Ask again rather than only polling. + stopUnitsHoldingSpx(t, node) + return fmt.Errorf("%s still has %s spx process(es)", node.Name, n) + } + return nil + }, spxProcessDrainTimeout, 3*time.Second) +} diff --git a/tests/e2e/multinode/cluster_restart_test.go b/tests/e2e/multinode/cluster_restart_test.go new file mode 100644 index 000000000..e146ce1ca --- /dev/null +++ b/tests/e2e/multinode/cluster_restart_test.go @@ -0,0 +1,111 @@ +//go:build e2e + +package multinode + +import ( + "fmt" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/tests/e2e/harness" + "github.com/stretchr/testify/require" +) + +// runClusterRestartResumesGuests is the software-update path: a coordinated +// shutdown drains every guest to stopped and seals its volumes, then the +// cluster comes back and each guest has to relaunch from that sealed state +// while its dependencies are themselves still starting. +// +// NodeFailure/NodeRecovery do not cover this. They stop the service units +// directly so guests keep running and the target's drain never fires, which +// makes recovery a QMP re-attach to a live process. Here the process is gone +// and the volume has to be re-opened, which is where a guest was lost: the +// relaunch raced a JetStream that had not finished starting, the lease claim +// timed out, and the failure was recorded as permanent. +func runClusterRestartResumesGuests(t *testing.T, fix *Fixture) { + harness.Phase(t, "Multinode — Cluster Restart Resumes Guests") + + require.GreaterOrEqualf(t, len(fix.Cluster.Nodes), 2, + "a cluster restart needs a multi-node cluster, have %d", len(fix.Cluster.Nodes)) + + trio := needInstanceTrio(t, fix) + require.NotEmpty(t, trio, "the restart is only meaningful with guests to resume") + + harness.Step(t, "confirm all %d guests are running before the shutdown", len(trio)) + for _, id := range trio { + harness.WaitForInstanceState(t, fix.AWS, id, "running") + } + + // If an assertion below fails mid-restart the cluster must not be left + // down for every later test in the package. + t.Cleanup(func() { + for _, n := range fix.Cluster.Nodes { + harness.StartNode(t, n) + } + }) + + harness.ClusterShutdownAndRestart(t, fix.Cluster, fix.Env) + + harness.Step(t, "every guest resumes to running without operator action") + var failures []string + for _, id := range trio { + if err := waitInstanceRunningNotErrored(t, fix, id, 5*time.Minute); err != nil { + failures = append(failures, err.Error()) + } + } + require.Emptyf(t, failures, + "guests must resume after a coordinated restart; an instance left in error is a guest an operator has to recover by hand:\n%s", + strings.Join(failures, "\n")) +} + +// waitInstanceRunningNotErrored polls until id is running, failing early and +// with the recorded reason if it lands in error instead. Waiting out the full +// timeout on an instance already parked in error would hide why. +func waitInstanceRunningNotErrored(t *testing.T, fix *Fixture, id string, timeout time.Duration) error { + t.Helper() + + deadline := time.Now().Add(timeout) + var last string + for time.Now().Before(deadline) { + out, err := fix.AWS.EC2.DescribeInstances(&ec2.DescribeInstancesInput{ + InstanceIds: []*string{aws.String(id)}, + }) + if err != nil { + // The gateway can still be settling immediately after the restart. + last = err.Error() + time.Sleep(5 * time.Second) + continue + } + if len(out.Reservations) == 0 || len(out.Reservations[0].Instances) == 0 { + last = "not returned by DescribeInstances" + time.Sleep(5 * time.Second) + continue + } + + inst := out.Reservations[0].Instances[0] + state := aws.StringValue(inst.State.Name) + last = state + switch state { + case "running": + return nil + case "error", "terminated": + return fmt.Errorf("%s reached %q after the restart instead of running: %s", + id, state, stateReason(inst)) + } + time.Sleep(5 * time.Second) + } + return fmt.Errorf("%s did not resume within %s, last state %q", id, timeout, last) +} + +// stateReason renders an instance's StateReason, which carries the recovery +// failure that explains a guest parked in error. +func stateReason(inst *ec2.Instance) string { + if inst.StateReason == nil { + return "no state reason recorded" + } + return fmt.Sprintf("%s: %s", + aws.StringValue(inst.StateReason.Code), aws.StringValue(inst.StateReason.Message)) +} diff --git a/tests/e2e/multinode/tests_test.go b/tests/e2e/multinode/tests_test.go index 215c08ff6..da20927fa 100644 --- a/tests/e2e/multinode/tests_test.go +++ b/tests/e2e/multinode/tests_test.go @@ -73,6 +73,16 @@ func TestMultinodeNodeRecovery(t *testing.T) { runNodeRecovery(t, requireMultiNodeFixture(t)) } +// TestMultinodeClusterRestartResumesGuests is sequential and declared after +// NodeRecovery so the cluster is whole before it takes the whole thing down. +// It is the software-update path — a coordinated drain and restart — which the +// node-failure tests deliberately do not cover: they leave guests running, and +// the failure this catches only appears when a guest has to relaunch from a +// sealed volume against dependencies that are still starting. +func TestMultinodeClusterRestartResumesGuests(t *testing.T) { + runClusterRestartResumesGuests(t, requireMultiNodeFixture(t)) +} + // TestMultinodeOVNRaft is sequential: stops ovn-central on the NB leader to // prove DB failover, restoring it in cleanup before later tests run. func TestMultinodeOVNRaft(t *testing.T) {