mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-21 22:56:55 +00:00
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) <noreply@anthropic.com>
This commit is contained in:
pingqiu
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
21ccf06ef3
commit
4f7283b6be
@@ -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,
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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_<node> 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 </dev/null >%s/volume.log 2>&1 & echo $!'",
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user