From 85dbfa1ccf928f5edc6f8313646c498cdc495363 Mon Sep 17 00:00:00 2001 From: Yusuf Ozturk Date: Wed, 15 Jul 2026 07:07:38 +0200 Subject: [PATCH 1/2] Device failover support --- internal/config/case.go | 2 +- internal/runner/runner.go | 153 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 154 insertions(+), 1 deletion(-) diff --git a/internal/config/case.go b/internal/config/case.go index 7a2d1e8..1f4274b 100644 --- a/internal/config/case.go +++ b/internal/config/case.go @@ -1248,7 +1248,7 @@ func (tc *TestCase) validateCluster() error { return fmt.Errorf("case %q: cluster.nodes must be >= 3, got %d", tc.Name, tc.Cluster.Nodes) } switch tc.Cluster.Action { - case "", "restart_follower", "restart_leader", "stop_two_recover", "agentless_failover", "cluster_ip_failover": + case "", "restart_follower", "restart_leader", "stop_two_recover", "agentless_failover", "device_failover", "cluster_ip_failover": default: return fmt.Errorf("case %q: unknown cluster.action %q", tc.Name, tc.Cluster.Action) } diff --git a/internal/runner/runner.go b/internal/runner/runner.go index 7d51d32..271a83f 100644 --- a/internal/runner/runner.go +++ b/internal/runner/runner.go @@ -2993,6 +2993,52 @@ func waitAgentlessCollectingExcluding(ctx context.Context, containers []string, return "", "", false } +// waitDeviceCollectingExcluding waits until some node other than `exclude` logs +// `marker` in its recent window, returning that node's name and readiness. It is +// the device_failover analogue of waitAgentlessCollectingExcluding, but keys on a +// caller-supplied marker ("Starting event collection of") instead of "entries +// from device": DB and file collectors log the collect-start line every cycle +// even when a correct checkpoint resume forwards ZERO new records, so the +// rows-forwarded marker would never fire on a clean resume. +// waitDeviceCollecting waits until the placement device is assigned to a node AND +// that owner logs `marker` in its recent window — the device_failover baseline +// analogue of waitAgentlessCollecting. Returns the owner node name, its +// container, and readiness. Delivery to the receiver is a separate (soft) signal: +// a cluster collector device's route/sender may not be co-located with the +// collecting owner, so cross-node payload transport is a known data-plane gap. +func waitDeviceCollecting(ctx context.Context, subjectName string, containers []string, marker string, deadline time.Time) (string, string, bool) { + for time.Now().Before(deadline) { + if owner := agentlessDeviceOwner(containers); owner != "" { + ownerContainer := fmt.Sprintf("bench-subject-%s-%s", subjectName, owner) + if strings.Contains(dockerLogsSince(ownerContainer, "30s"), marker) { + return owner, ownerContainer, true + } + } + if sleepCtx(ctx, 5*time.Second) != nil { + break + } + } + return agentlessDeviceOwner(containers), "", false +} + +func waitDeviceCollectingExcluding(ctx context.Context, containers []string, exclude, marker string, deadline time.Time) (string, bool) { + for time.Now().Before(deadline) { + for i, c := range containers { + name := strconv.Itoa(i + 1) + if name == exclude { + continue + } + if strings.Contains(dockerLogsSince(c, "30s"), marker) { + return name, true + } + } + if sleepCtx(ctx, 5*time.Second) != nil { + break + } + } + return "", false +} + // clusterReassignedFrom returns the "Reassigned device ... from to Y" // failover line found across all nodes, and whether one exists. It matches the // specific source node (the owner we stopped) so it can NOT be satisfied by the @@ -3253,6 +3299,7 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf isAgentless := tc.Cluster.Action == "agentless_failover" isClusterIP := tc.Cluster.Action == "cluster_ip_failover" + isDeviceFailover := tc.Cluster.Action == "device_failover" var baselineOK bool var owner, ownerContainer string @@ -3265,6 +3312,25 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf errs = append(errs, "agentless device did not deploy/collect on any node during baseline") } r.sampleDelivery(metricsPort, 0) // soft: downstream E2E delivery (see note above) + } else if isDeviceFailover { + // A placement collector device (DB / file). The baseline is the + // director-level proof — the device is assigned to ONE node and that + // owner runs a collect cycle. Receiver delivery is a SOFT signal: a + // cluster collector device's route/sender may not be co-located with + // the collecting owner, so cross-node payload transport is a known + // data-plane gap (the same one agentless_failover soft-logs). The hard + // verdict is single-owner placement + automatic failover + resume. + fmt.Printf(" baseline: waiting for the device to be placed + run a collect cycle (up to %s)…\n", time.Until(drainDeadline).Round(time.Second)) + owner, ownerContainer, baselineOK = waitDeviceCollecting(r.ctx, subject.Name, nodes, "Starting event collection of", drainDeadline) + if baselineOK { + fmt.Printf(" device placed; owner = node %s (%s), collecting ✓\n", owner, ownerContainer) + } else { + errs = append(errs, "device was not placed / did not run a collect cycle on any node during baseline") + } + if rm, qerr := r.queryReceiverMetrics(metricsPort, 10*time.Second); qerr == nil { + finalCount = rm.LinesReceived + fmt.Printf(" (soft) baseline delivery: %s\n", formatCount(finalCount)) + } } else if isClusterIP { // The VIP test's baseline is simply a formed cluster with a leader (already // asserted by waitClusterReady above). Downstream delivery is a SOFT signal @@ -3450,6 +3516,76 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf } r.sampleDelivery(metricsPort, preFailover) } + case "device_failover": + // Generalized placement-device failover (DB / file). Unlike agentless, + // the owner is resolved here (the hard-delivery baseline branch above + // did not set it), and downstream delivery is asserted HARD (below, + // via the receiver's own correctness verdict): a DB/file device + // forwards on the reliable direct path, not the cross-node payload + // store that makes agentless delivery a soft signal. + owner = agentlessDeviceOwner(nodes) + if owner == "" { + actionOK = false + errs = append(errs, "could not determine the placement device owner (no 'Assigned/Reassigned device' log) — failover untestable") + } else { + ownerContainer = fmt.Sprintf("bench-subject-%s-%s", subject.Name, owner) + preFailover := finalCount + fmt.Printf(" placement device owner = node %s (%s); stopping it to force a failover…\n", owner, ownerContainer) + + // STOP (not restart): the owner must stay down past the ~15s + // heartbeat timeout so the leader reassigns; start it again after. + if serr := exec.Command("docker", "stop", "-t", "10", ownerContainer).Run(); serr != nil { + actionOK = false + errs = append(errs, fmt.Sprintf("docker stop %s failed: %v (disruption did not happen)", ownerContainer, serr)) + } + fmt.Printf(" waiting %s for the leader to detect the down node and reassign the device…\n", settle) + if err := sleepCtx(r.ctx, settle); err != nil { + return results.RunResult{}, fmt.Errorf("interrupted: %w", err) + } + + // Reassignment log (informational). The leader logs "Reassigned + // device ... from to Y" when the OWNER was a follower. When + // the owner was ALSO the leader (placement can land there), the + // newly elected leader may log a fresh "Assigned new device ... to Y" + // instead — so the absence of a "from " line is NOT a + // failure. HARD 1 below (a survivor actively collecting the device on + // a node != owner) is the robust, placement-form-independent proof + // that the device failed over. + if line, ok := clusterReassignedFrom(nodes, owner); ok { + fmt.Printf(" failover observed: %s\n", line) + } else { + fmt.Printf(" (note) no 'Reassigned ... from %s to Y' line (owner was likely the leader; relying on the survivor-collecting proof below)\n", owner) + } + // HARD 1: a survivor (node != stopped owner) actually RAN a poll cycle + // on the reassigned device — proving the device failed over AND the + // new owner resumed from the persisted checkpoint (a clean DB/file + // resume forwards 0 new records, so key on the collect-start line, + // not rows-forwarded). Checked while the old owner is still down so + // placement is firmly on the survivor. + collectDeadline := time.Now().Add(settle + 90*time.Second) + if collectDeadline.After(runDeadline) { + collectDeadline = runDeadline + } + if newOwner, ok := waitDeviceCollectingExcluding(r.ctx, nodes, owner, "Starting event collection of", collectDeadline); ok { + fmt.Printf(" survivor node %s re-homed the device and ran a poll cycle ✓\n", newOwner) + } else { + actionOK = false + errs = append(errs, "no survivor ran a collection cycle after failover — device did not fail over") + } + // HARD 2: a leader still exists after the loss (cluster stayed healthy). + if _, ok := leaderExistsNow(nodes); !ok { + actionOK = false + errs = append(errs, "no leader after the owning node went down") + } + + // Restore full strength. + fmt.Printf(" starting node %s again to restore the cluster…\n", ownerContainer) + if serr := exec.Command("docker", "start", ownerContainer).Run(); serr != nil { + actionOK = false + errs = append(errs, fmt.Sprintf("docker start %s failed: %v", ownerContainer, serr)) + } + r.sampleDelivery(metricsPort, preFailover) + } case "cluster_ip_failover": // The elected leader must hold the virtual IP; followers must not. Then // restart the leader and assert the IP migrates to the newly elected leader @@ -3538,6 +3674,23 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf if rm, qerr := r.queryReceiverMetrics(metricsPort, 30*time.Second); qerr == nil { finalCount = rm.LinesReceived + // device_failover: receiver delivery is a SOFT signal. A cluster collector + // device's route/sender may run on a different node than the collecting + // owner, so cross-node payload transport is a known data-plane gap + // (mem:// → not-found, natsobj → evicted) — the same gap agentless_failover + // soft-logs. When delivery DOES happen (owner co-located with the router), + // the receiver's dedup verdict shows the resumed cursor did not re-deliver; + // log it either way, but do not gate the cluster verdict on it. + if tc.Cluster != nil && tc.Cluster.Action == "device_failover" { + switch { + case rm.Passed != nil && !*rm.Passed: + fmt.Printf(" (soft) post-failover delivery correctness: FAILED — known cluster cross-node data-plane gap (%s)\n", strings.Join(rm.Errors, "; ")) + case finalCount > 0: + fmt.Printf(" (soft) post-failover delivery: %s delivered, receiver dedup clean (resumed cursor did not re-deliver)\n", formatCount(finalCount)) + default: + fmt.Println(" (soft) post-failover delivery: 0 at the receiver (owner not co-located with router — known cluster data-plane gap)") + } + } } elapsed := time.Since(startTime).Seconds() From 2b76b3b5ee5cd2dcdacd91f40f84912586215ffc Mon Sep 17 00:00:00 2001 From: Yusuf Ozturk Date: Wed, 22 Jul 2026 23:53:52 +0200 Subject: [PATCH 2/2] Review fixes --- internal/config/case.go | 12 ++++++++++++ internal/runner/runner.go | 13 +++++++------ 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/internal/config/case.go b/internal/config/case.go index 1f4274b..e8f709e 100644 --- a/internal/config/case.go +++ b/internal/config/case.go @@ -1208,6 +1208,18 @@ type ClusterConfig struct { // a node loss is a known director gap). Requires // persistent_storage in the subject config (collector payloads // must be cluster-shared via the NATS object store). + // device_failover — the placement analogue of agentless_failover for a DB/file + // collector device: stop the node that OWNS the device so the + // leader reassigns it to ANOTHER node, then start it again. + // The hard verdict is that a survivor re-homes the device and + // runs a collect cycle resuming from the persisted checkpoint + // (a clean DB/file resume forwards 0 new rows, so it keys on + // the collect-start line, not rows-forwarded), plus a leader + // still exists. Receiver delivery is soft-logged — the same + // cross-node data-plane gap agentless_failover has. Requires + // the subject to define a DB/file placement collector device + // and, like agentless_failover, cluster-shared collector state + // (persistent_storage) so the survivor can resume. // cluster_ip_failover — the leader binds cluster.ip as a virtual IP on its // interface; restart the leader and assert the IP migrates // to the newly elected leader (and the old leader releases diff --git a/internal/runner/runner.go b/internal/runner/runner.go index 271a83f..48281b5 100644 --- a/internal/runner/runner.go +++ b/internal/runner/runner.go @@ -3517,12 +3517,13 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf r.sampleDelivery(metricsPort, preFailover) } case "device_failover": - // Generalized placement-device failover (DB / file). Unlike agentless, - // the owner is resolved here (the hard-delivery baseline branch above - // did not set it), and downstream delivery is asserted HARD (below, - // via the receiver's own correctness verdict): a DB/file device - // forwards on the reliable direct path, not the cross-node payload - // store that makes agentless delivery a soft signal. + // Generalized placement-device failover (DB / file). The baseline branch + // above already set owner/ownerContainer; re-confirm the current owner + // here before the disruption. Receiver delivery is a SOFT signal — a + // DB/file device's route/sender may not be co-located with the collecting + // owner, the same cross-node data-plane gap agentless_failover soft-logs — + // so the verdict is passed = baselineOK && actionOK (below), not gated on + // delivery. owner = agentlessDeviceOwner(nodes) if owner == "" { actionOK = false