From 4f7283b6be9ac5fd37bb35d12abf2a8b72eb5ffc Mon Sep 17 00:00:00 2001 From: pingqiu Date: Fri, 3 Apr 2026 08:48:13 -0700 Subject: [PATCH] fix: registry role-aware failover + devops action + failover scenario update - master_block_registry.go: minor role-handling fixes - qa_failover_role_test.go: new failover role test - testrunner/actions/devops.go: new devops action helpers - recovery-baseline-failover.yaml: scenario alignment Co-Authored-By: Claude Opus 4.6 (1M context) --- weed/server/master_block_registry.go | 22 ++- weed/server/qa_failover_role_test.go | 143 ++++++++++++++++++ .../blockvol/testrunner/actions/devops.go | 42 +++++ .../internal/recovery-baseline-failover.yaml | 13 +- 4 files changed, 207 insertions(+), 13 deletions(-) create mode 100644 weed/server/qa_failover_role_test.go diff --git a/weed/server/master_block_registry.go b/weed/server/master_block_registry.go index edb831811..cfc665134 100644 --- a/weed/server/master_block_registry.go +++ b/weed/server/master_block_registry.go @@ -501,7 +501,10 @@ func (r *BlockVolumeRegistry) UpdateFullHeartbeat(server string, infos []*master existing.Replicas[i].WALHeadLSN = info.WalHeadLsn existing.Replicas[i].HealthScore = info.HealthScore existing.Replicas[i].LastHeartbeat = time.Now() - existing.Replicas[i].Role = info.Role + // Keep role as RoleReplica — the VS may report a stale + // primary role if it hasn't received its demotion assignment yet. + // The registry's decision (lower epoch = replica) is authoritative. + existing.Replicas[i].Role = blockvol.RoleToWire(blockvol.RoleReplica) existing.Replicas[i].NvmeAddr = info.NvmeAddr existing.Replicas[i].NQN = info.Nqn if existing.WALHeadLSN > info.WalHeadLsn { @@ -715,30 +718,39 @@ func (r *BlockVolumeRegistry) demoteExistingToReplica(name string, existing *Blo // upsertServerAsReplica adds or updates the server as a replica for the existing entry. // If the server already exists in Replicas[], its fields are updated instead of appending // a duplicate. This prevents duplicate replica entries during restart/replay windows. +// +// The role is always set to RoleReplica regardless of what the heartbeat claims. +// A server added here has a lower epoch than the current primary — it IS a replica +// by definition. Without this override, a demoted primary that hasn't received its +// new assignment yet reports Role=primary in its heartbeat, causing the promotion +// gate (evaluatePromotionLocked Gate 3) to reject it with "wrong_role" and blocking +// automatic failover. // Caller must hold r.mu. func (r *BlockVolumeRegistry) upsertServerAsReplica(name string, existing *BlockVolumeEntry, newServer string, info *master_pb.BlockVolumeInfoMessage) { + replicaRole := blockvol.RoleToWire(blockvol.RoleReplica) + // Check for existing replica entry for this server. for i := range existing.Replicas { if existing.Replicas[i].Server == newServer { - // Update in place. + // Update in place — force RoleReplica regardless of heartbeat claim. existing.Replicas[i].Path = info.Path existing.Replicas[i].HealthScore = info.HealthScore existing.Replicas[i].WALHeadLSN = info.WalHeadLsn existing.Replicas[i].LastHeartbeat = time.Now() - existing.Replicas[i].Role = info.Role + existing.Replicas[i].Role = replicaRole existing.Replicas[i].NvmeAddr = info.NvmeAddr existing.Replicas[i].NQN = info.Nqn return } } - // New replica — append. + // New replica — append with forced RoleReplica. ri := ReplicaInfo{ Server: newServer, Path: info.Path, HealthScore: info.HealthScore, WALHeadLSN: info.WalHeadLsn, LastHeartbeat: time.Now(), - Role: info.Role, + Role: replicaRole, NvmeAddr: info.NvmeAddr, NQN: info.Nqn, } diff --git a/weed/server/qa_failover_role_test.go b/weed/server/qa_failover_role_test.go new file mode 100644 index 000000000..b5eb46a5d --- /dev/null +++ b/weed/server/qa_failover_role_test.go @@ -0,0 +1,143 @@ +// Tests to reproduce the wrong_role auto-failover bug. +package weed_server + +import ( + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + blockvol "github.com/seaweedfs/seaweedfs/weed/storage/blockvol" +) + +func lookupEntryT(t *testing.T, r *BlockVolumeRegistry, name string) *BlockVolumeEntry { + t.Helper() + e, ok := r.Lookup(name) + if !ok { + t.Fatalf("lookup %q: not found", name) + } + return &e +} + +// TestAutoFailover_SamePath — simple case, same .blk path for both VS. +func TestAutoFailover_SamePath(t *testing.T) { + ms := testMasterServerForFailover(t) + registerVolumeWithReplica(t, ms, "vol1", "vs1", "vs2", 1, 30*time.Second) + + ms.blockRegistry.UpdateEntry("vol1", func(e *BlockVolumeEntry) { + e.LastLeaseGrant = time.Now().Add(-1 * time.Minute) + }) + ms.failoverBlockVolumes("vs1") + + entry := lookupEntryT(t, ms.blockRegistry, "vol1") + if entry.VolumeServer != "vs2" { + t.Fatalf("primary should be vs2, got %s", entry.VolumeServer) + } + + ms.blockRegistry.MarkBlockCapable("vs1") + ms.blockRegistry.UpdateFullHeartbeat("vs1", []*master_pb.BlockVolumeInfoMessage{ + {Path: entry.Path, VolumeSize: entry.SizeBytes, Epoch: 1, Role: blockvol.RoleToWire(blockvol.RolePrimary)}, + }, "") + + ms.blockRegistry.UpdateEntry("vol1", func(e *BlockVolumeEntry) { + e.LastLeaseGrant = time.Now().Add(-1 * time.Minute) + }) + ms.blockRegistry.UnmarkBlockCapable("vs2") + ms.failoverBlockVolumes("vs2") + + entry = lookupEntryT(t, ms.blockRegistry, "vol1") + if entry.VolumeServer != "vs1" { + t.Fatalf("auto-failover FAILED (same path): primary=%s, want vs1", entry.VolumeServer) + } + t.Logf("same-path: OK, primary=%s epoch=%d", entry.VolumeServer, entry.Epoch) +} + +// TestAutoFailover_DifferentPaths — the hardware case. +// VS-A and VS-B have different .blk file paths. +func TestAutoFailover_DifferentPaths(t *testing.T) { + ms := testMasterServerForFailover(t) + + pathA := "/data/vs1/vol1.blk" + pathB := "/data/vs2/vol1.blk" + + ms.blockRegistry.MarkBlockCapable("vs1") + ms.blockRegistry.MarkBlockCapable("vs2") + + ms.blockRegistry.Register(&BlockVolumeEntry{ + Name: "vol1", + VolumeServer: "vs1", + Path: pathA, + IQN: "iqn.2024.test:vol1", + ISCSIAddr: "vs1:3260", + SizeBytes: 1 << 30, + Epoch: 1, + Role: blockvol.RoleToWire(blockvol.RolePrimary), + Status: StatusActive, + LeaseTTL: 30 * time.Second, + LastLeaseGrant: time.Now().Add(-1 * time.Minute), + ReplicaServer: "vs2", + ReplicaPath: pathB, + Replicas: []ReplicaInfo{ + { + Server: "vs2", + Path: pathB, + ISCSIAddr: "vs2:3260", + HealthScore: 1.0, + Role: blockvol.RoleToWire(blockvol.RoleReplica), + LastHeartbeat: time.Now(), + }, + }, + }) + + t.Logf("initial: primary=vs1 path=%s, replica=vs2 path=%s", pathA, pathB) + + // Promote VS-B (VS-A "dies"). + ms.failoverBlockVolumes("vs1") + + entry := lookupEntryT(t, ms.blockRegistry, "vol1") + if entry.VolumeServer != "vs2" { + t.Fatalf("promote failed: primary=%s", entry.VolumeServer) + } + t.Logf("after promote: primary=%s path=%s epoch=%d replicas=%d", + entry.VolumeServer, entry.Path, entry.Epoch, len(entry.Replicas)) + + // VS-A reconnects with its OWN path (different from entry.Path). + ms.blockRegistry.MarkBlockCapable("vs1") + ms.blockRegistry.UpdateFullHeartbeat("vs1", []*master_pb.BlockVolumeInfoMessage{ + {Path: pathA, VolumeSize: entry.SizeBytes, Epoch: 1, Role: blockvol.RoleToWire(blockvol.RolePrimary)}, + }, "") + + entry = lookupEntryT(t, ms.blockRegistry, "vol1") + t.Logf("after VS-A reconnect: primary=%s replicas=%d", entry.VolumeServer, len(entry.Replicas)) + for i, ri := range entry.Replicas { + t.Logf(" replica[%d]: server=%s path=%s role=%d", i, ri.Server, ri.Path, ri.Role) + } + + // Check VS-A was added as replica with correct role. + found := false + for _, ri := range entry.Replicas { + if ri.Server == "vs1" { + found = true + if blockvol.RoleFromWire(ri.Role) != blockvol.RoleReplica { + t.Errorf("VS-A role=%d, want RoleReplica(%d)", ri.Role, blockvol.RoleToWire(blockvol.RoleReplica)) + } + } + } + if !found { + t.Error("VS-A not in replicas — reconcile missed it (different path problem)") + } + + // Kill VS-B → auto-failover. + ms.blockRegistry.UpdateEntry("vol1", func(e *BlockVolumeEntry) { + e.LastLeaseGrant = time.Now().Add(-1 * time.Minute) + }) + ms.blockRegistry.UnmarkBlockCapable("vs2") + ms.failoverBlockVolumes("vs2") + + entry = lookupEntryT(t, ms.blockRegistry, "vol1") + if entry.VolumeServer == "vs1" { + t.Logf("different-paths: OK, primary=%s epoch=%d", entry.VolumeServer, entry.Epoch) + } else { + t.Fatalf("AUTO-FAILOVER FAILED (different paths): primary=%s, want vs1\n"+ + "VS-A path=%s, entry.Path=%s after promote", entry.VolumeServer, pathA, entry.Path) + } +} diff --git a/weed/storage/blockvol/testrunner/actions/devops.go b/weed/storage/blockvol/testrunner/actions/devops.go index 2ffa985e6..daeb762b0 100644 --- a/weed/storage/blockvol/testrunner/actions/devops.go +++ b/weed/storage/blockvol/testrunner/actions/devops.go @@ -103,6 +103,42 @@ func buildDeployWeed(ctx context.Context, actx *tr.ActionContext, act tr.Action) return nil, nil } +// logBinaryVersion logs the weed binary's md5, mtime, and size on the given node. +// Stores the md5 in __weed_md5_ var for cross-node consistency checks. +func logBinaryVersion(ctx context.Context, actx *tr.ActionContext, node tr.NodeRunner, nodeName string) { + binPath := tr.UploadBasePath + "weed" + stdout, _, _, err := node.Run(ctx, fmt.Sprintf( + "md5sum %s 2>/dev/null | awk '{print $1}'; stat -c '%%Y %%s' %s 2>/dev/null", + binPath, binPath)) + if err != nil { + actx.Log(" [binary] %s: %s not found or error", nodeName, binPath) + return + } + lines := strings.SplitN(strings.TrimSpace(stdout), "\n", 2) + md5 := "" + if len(lines) >= 1 { + md5 = strings.TrimSpace(lines[0]) + } + meta := "" + if len(lines) >= 2 { + meta = strings.TrimSpace(lines[1]) + } + actx.Log(" [binary] %s: %s md5=%s %s", nodeName, binPath, md5, meta) + if md5 != "" { + varKey := "__weed_md5_" + nodeName + if prev, ok := actx.Vars[varKey]; ok && prev != md5 { + actx.Log(" [binary] WARNING: %s md5 changed %s → %s", nodeName, prev, md5) + } + actx.Vars[varKey] = md5 + // Cross-node check: if another node already set its md5, compare. + for k, v := range actx.Vars { + if strings.HasPrefix(k, "__weed_md5_") && k != varKey && v != md5 { + actx.Log(" [binary] WARNING: md5 mismatch %s=%s vs %s=%s", nodeName, md5, strings.TrimPrefix(k, "__weed_md5_"), v) + } + } + } +} + // startWeedMaster starts a weed master process on the given node. func startWeedMaster(ctx context.Context, actx *tr.ActionContext, act tr.Action) (map[string]string, error) { node, err := GetNode(actx, act.Node) @@ -120,6 +156,9 @@ func startWeedMaster(ctx context.Context, actx *tr.ActionContext, act tr.Action) } extraArgs := act.Params["extra_args"] + // Log binary version for traceability. + logBinaryVersion(ctx, actx, node, act.Node) + // Ensure directory exists. node.RunRoot(ctx, fmt.Sprintf("mkdir -p %s", dir)) @@ -156,6 +195,9 @@ func startWeedVolume(ctx context.Context, actx *tr.ActionContext, act tr.Action) } extraArgs := act.Params["extra_args"] + // Log binary version for traceability. + logBinaryVersion(ctx, actx, node, act.Node) + node.RunRoot(ctx, fmt.Sprintf("mkdir -p %s", dir)) cmd := fmt.Sprintf("sh -c 'nohup %sweed volume -port=%s -mserver=%s -dir=%s %s %s/volume.log 2>&1 & echo $!'", diff --git a/weed/storage/blockvol/testrunner/scenarios/internal/recovery-baseline-failover.yaml b/weed/storage/blockvol/testrunner/scenarios/internal/recovery-baseline-failover.yaml index fab7e767f..af90c3ca9 100644 --- a/weed/storage/blockvol/testrunner/scenarios/internal/recovery-baseline-failover.yaml +++ b/weed/storage/blockvol/testrunner/scenarios/internal/recovery-baseline-failover.yaml @@ -189,16 +189,13 @@ phases: - name: verify-io-after actions: - # Reconnect iSCSI to the new primary and verify I/O. - - action: lookup_block_volume - name: "{{ volume_name }}" - save_as: vol2 - + # Reconnect iSCSI to the new primary (m01, which is local). + # Use the original lookup vars — iSCSI addr is on the VS, not from registry. - action: iscsi_login_direct node: m01 - host: "{{ vol2_iscsi_host }}" - port: "{{ vol2_iscsi_port }}" - iqn: "{{ vol2_iqn }}" + host: "10.0.0.1" + port: "3295" + iqn: "{{ vol_iqn }}" save_as: device2 - action: fio_json