From 979c54f693e0c27def5b6f5d3d0cafc1b73ee697 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 15 Apr 2026 12:29:31 -0700 Subject: [PATCH] fix(wdclient,volume): compare master leader with ServerAddress.Equals (#9089) * fix(wdclient,volume): compare master leader with ServerAddress.Equals Raft leader is advertised as host:httpPort.grpcPort, but clients dial host:httpPort. Raw string comparison against VolumeLocation.Leader / HeartbeatResponse.Leader therefore never matches, causing the masterclient and the volume server heartbeat loop to continuously "redirect" to the already-connected master, tearing down the stream and reconnecting. Use ServerAddress.Equals, which normalizes the grpc-port suffix. * fix(filer,mq): compare ServerAddress via Equals in two more sites filer bootstrap skip (MaybeBootstrapFromOnePeer) and the broker's local partition assignment check both compared a wire-supplied address string against the local self ServerAddress with raw string equality. Both are vulnerable to the same plain-vs-host:port.grpcPort mismatch as the masterclient/volume heartbeat sites: filer would bootstrap from itself, and the broker would fail to claim a partition it was actually assigned. Route both through ServerAddress.Equals. * fix(master,shell): more ServerAddress comparisons via Equals - raft_server_handlers.go HealthzHandler: s.serverAddr == leader would skip the child-lock check on the real leader when the two carry different plain/grpc-suffix forms, returning 200 OK instead of 423. - master_server.go SetRaftServer leader-change callback: the Leader() == Name() guard for ensureTopologyId could disagree with topology.IsLeader() (which already uses Equals), so leader-only initialization could be skipped after an election. - command_volume_merge.go isReplicaServer: the -target guard compared user-supplied host:port against NewServerAddressFromDataNode(...) with ==, letting an existing replica slip through when topology carries the embedded gRPC port. All routed through pb.ServerAddress.Equals. * fix(mq,cluster): more ServerAddress comparisons via Equals - broker_grpc_lookup.go GetTopicPublishers/GetTopicSubscribers: the partition ownership check gated listing on raw LeaderBroker == BrokerAddress().String(), so listings silently omitted partitions hosted locally when the assignment carried the other host:port / host:port.grpcPort form. - lock_client.go: LockHostMovedTo comparison and the seedFiler fallback guard both used raw string equality against configured filer addresses (which may be plain host:port while LockHostMovedTo comes back suffixed), causing spurious host-change churn and blocking the seed-filer fallback. * fix(mq): more ServerAddress comparisons via Equals - pub_balancer/allocate.go EnsureAssignmentsToActiveBrokers: direct activeBrokers.Get() lookup missed brokers when a persisted assignment carried a different address encoding than the registered broker key, triggering a bogus reassignment on every read/write cycle. Added a findActiveBroker helper that falls back to an Equals-based scan and canonicalizes the assignment in place so later writes are stable. - broker_grpc_lookup.go isLockOwner: used raw string equality between LockOwner() and BrokerAddress().String(), so a lock owner could fail to recognize itself and proxy local lookup/config/admin RPCs away. - pub_client/scheduler.go onEachAssignments: reused publisher jobs only on exact LeaderBroker match, so an encoding flip in lookup results tore down and recreated a stream to the same broker. --- weed/cluster/lock_client.go | 4 +-- weed/filer/filer.go | 2 +- weed/mq/broker/broker_grpc_lookup.go | 6 ++-- .../mq/broker/broker_topic_conf_read_write.go | 5 ++-- weed/mq/client/pub_client/scheduler.go | 2 +- weed/mq/pub_balancer/allocate.go | 30 +++++++++++++++++-- weed/server/master_server.go | 2 +- weed/server/raft_server_handlers.go | 2 +- weed/server/volume_grpc_client_to_master.go | 2 +- weed/shell/command_volume_merge.go | 2 +- weed/wdclient/masterclient.go | 6 ++-- 11 files changed, 45 insertions(+), 18 deletions(-) diff --git a/weed/cluster/lock_client.go b/weed/cluster/lock_client.go index 3cc61ac4e..d1ce242f6 100644 --- a/weed/cluster/lock_client.go +++ b/weed/cluster/lock_client.go @@ -267,7 +267,7 @@ func (lock *LiveLock) doLock(lockDuration time.Duration) (errorMessage string, e } if resp != nil { errorMessage = resp.Error - if resp.LockHostMovedTo != "" && resp.LockHostMovedTo != string(previousHostFiler) { + if resp.LockHostMovedTo != "" && !pb.ServerAddress(resp.LockHostMovedTo).Equals(previousHostFiler) { // Only log if the host actually changed glog.V(2).Infof("LOCK: Host changed from %s to %s for key=%s", previousHostFiler, resp.LockHostMovedTo, lock.key) lock.hostFiler = pb.ServerAddress(resp.LockHostMovedTo) @@ -289,7 +289,7 @@ func (lock *LiveLock) doLock(lockDuration time.Duration) (errorMessage string, e return err }) - if err != nil && lock.hostFiler != lock.lc.seedFiler { + if err != nil && !lock.hostFiler.Equals(lock.lc.seedFiler) { lock.consecutiveFailures++ // Fall back to seed filer after 3 consecutive connection failures if lock.consecutiveFailures >= 3 { diff --git a/weed/filer/filer.go b/weed/filer/filer.go index d31c74fbc..50125a0f0 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -108,7 +108,7 @@ func (f *Filer) MaybeBootstrapFromOnePeer(self pb.ServerAddress, existingNodes [ return existingNodes[i].CreatedAtNs < existingNodes[j].CreatedAtNs }) earliestNode := existingNodes[0] - if earliestNode.Address == string(self) { + if pb.ServerAddress(earliestNode.Address).Equals(self) { return } diff --git a/weed/mq/broker/broker_grpc_lookup.go b/weed/mq/broker/broker_grpc_lookup.go index fad10f599..38da0dbb6 100644 --- a/weed/mq/broker/broker_grpc_lookup.go +++ b/weed/mq/broker/broker_grpc_lookup.go @@ -336,7 +336,7 @@ func (b *MessageQueueBroker) GetTopicPublishers(ctx context.Context, request *mq // Collect publishers from each partition that is hosted on this broker for _, assignment := range conf.BrokerPartitionAssignments { // Only collect from partitions where this broker is the leader - if assignment.LeaderBroker == b.option.BrokerAddress().String() { + if pb.ServerAddress(assignment.LeaderBroker).Equals(b.option.BrokerAddress()) { partition := topic.FromPbPartition(assignment.Partition) if localPartition := b.localTopicManager.GetLocalPartition(t, partition); localPartition != nil { // Get publisher information from local partition @@ -390,7 +390,7 @@ func (b *MessageQueueBroker) GetTopicSubscribers(ctx context.Context, request *m // Collect subscribers from each partition that is hosted on this broker for _, assignment := range conf.BrokerPartitionAssignments { // Only collect from partitions where this broker is the leader - if assignment.LeaderBroker == b.option.BrokerAddress().String() { + if pb.ServerAddress(assignment.LeaderBroker).Equals(b.option.BrokerAddress()) { partition := topic.FromPbPartition(assignment.Partition) if localPartition := b.localTopicManager.GetLocalPartition(t, partition); localPartition != nil { // Get subscriber information from local partition @@ -430,5 +430,5 @@ func (b *MessageQueueBroker) GetTopicSubscribers(ctx context.Context, request *m } func (b *MessageQueueBroker) isLockOwner() bool { - return b.lockAsBalancer.LockOwner() == b.option.BrokerAddress().String() + return pb.ServerAddress(b.lockAsBalancer.LockOwner()).Equals(b.option.BrokerAddress()) } diff --git a/weed/mq/broker/broker_topic_conf_read_write.go b/weed/mq/broker/broker_topic_conf_read_write.go index 976efb36c..07322907d 100644 --- a/weed/mq/broker/broker_topic_conf_read_write.go +++ b/weed/mq/broker/broker_topic_conf_read_write.go @@ -13,6 +13,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/mq/logstore" "github.com/seaweedfs/seaweedfs/weed/mq/pub_balancer" "github.com/seaweedfs/seaweedfs/weed/mq/topic" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb" ) @@ -147,12 +148,12 @@ func (b *MessageQueueBroker) genLocalPartitionFromFiler(t topic.Topic, partition for _, assignment := range conf.BrokerPartitionAssignments { assignmentPartition := topic.FromPbPartition(assignment.Partition) glog.V(4).Infof("checking assignment: LeaderBroker=%s, Partition=%s", assignment.LeaderBroker, assignmentPartition) - glog.V(4).Infof("comparing self=%s with LeaderBroker=%s: %v", self, assignment.LeaderBroker, assignment.LeaderBroker == string(self)) + glog.V(4).Infof("comparing self=%s with LeaderBroker=%s: %v", self, assignment.LeaderBroker, pb.ServerAddress(assignment.LeaderBroker).Equals(self)) glog.V(4).Infof("comparing partition=%s with assignmentPartition=%s: %v", partition.String(), assignmentPartition.String(), partition.Equals(assignmentPartition)) glog.V(4).Infof("logical comparison (RangeStart, RangeStop only): %v", partition.LogicalEquals(assignmentPartition)) glog.V(4).Infof("partition details: RangeStart=%d, RangeStop=%d, RingSize=%d, UnixTimeNs=%d", partition.RangeStart, partition.RangeStop, partition.RingSize, partition.UnixTimeNs) glog.V(4).Infof("assignmentPartition details: RangeStart=%d, RangeStop=%d, RingSize=%d, UnixTimeNs=%d", assignmentPartition.RangeStart, assignmentPartition.RangeStop, assignmentPartition.RingSize, assignmentPartition.UnixTimeNs) - if assignment.LeaderBroker == string(self) && partition.LogicalEquals(assignmentPartition) { + if pb.ServerAddress(assignment.LeaderBroker).Equals(self) && partition.LogicalEquals(assignmentPartition) { glog.V(4).Infof("Creating local partition for %s %s", t, partition) localPartition = topic.NewLocalPartition(partition, b.option.LogFlushInterval, b.genLogFlushFunc(t, partition), logstore.GenMergedReadFunc(b, t, partition)) diff --git a/weed/mq/client/pub_client/scheduler.go b/weed/mq/client/pub_client/scheduler.go index 8cb481051..97cbfffee 100644 --- a/weed/mq/client/pub_client/scheduler.go +++ b/weed/mq/client/pub_client/scheduler.go @@ -91,7 +91,7 @@ func (p *TopicPublisher) onEachAssignments(generation int, assignments []*mq_pb. if hasExistingJob { var existingJob *EachPartitionPublishJob existingJob = p.jobs[i] - if existingJob.BrokerPartitionAssignment.LeaderBroker == assignment.LeaderBroker { + if pb.ServerAddress(existingJob.BrokerPartitionAssignment.LeaderBroker).Equals(pb.ServerAddress(assignment.LeaderBroker)) { existingJob.generation = generation jobs = append(jobs, existingJob) continue diff --git a/weed/mq/pub_balancer/allocate.go b/weed/mq/pub_balancer/allocate.go index 09124284b..18f5307ad 100644 --- a/weed/mq/pub_balancer/allocate.go +++ b/weed/mq/pub_balancer/allocate.go @@ -6,10 +6,30 @@ import ( cmap "github.com/orcaman/concurrent-map/v2" "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb" "github.com/seaweedfs/seaweedfs/weed/pb/schema_pb" ) +// findActiveBroker looks up a broker in activeBrokers, tolerating address +// encoding differences (plain host:port vs host:port.grpcPort) by falling +// back to an Equals-based scan when the direct key lookup misses. +func findActiveBroker(activeBrokers cmap.ConcurrentMap[string, *BrokerStats], addr string) (string, bool) { + if addr == "" { + return "", false + } + if _, found := activeBrokers.Get(addr); found { + return addr, true + } + target := pb.ServerAddress(addr) + for item := range activeBrokers.IterBuffered() { + if pb.ServerAddress(item.Key).Equals(target) { + return item.Key, true + } + } + return "", false +} + func AllocateTopicPartitions(brokers cmap.ConcurrentMap[string, *BrokerStats], partitionCount int32) (assignments []*mq_pb.BrokerPartitionAssignment) { // divide the ring into partitions now := time.Now().UnixNano() @@ -91,15 +111,21 @@ func EnsureAssignmentsToActiveBrokers(activeBrokers cmap.ConcurrentMap[string, * count := 0 if assignment.LeaderBroker == "" { count++ - } else if _, found := activeBrokers.Get(assignment.LeaderBroker); !found { + } else if canonical, found := findActiveBroker(activeBrokers, assignment.LeaderBroker); !found { assignment.LeaderBroker = "" count++ + } else if canonical != assignment.LeaderBroker { + assignment.LeaderBroker = canonical + hasChanges = true } if assignment.FollowerBroker == "" { count++ - } else if _, found := activeBrokers.Get(assignment.FollowerBroker); !found { + } else if canonical, found := findActiveBroker(activeBrokers, assignment.FollowerBroker); !found { assignment.FollowerBroker = "" count++ + } else if canonical != assignment.FollowerBroker { + assignment.FollowerBroker = canonical + hasChanges = true } if count > 0 { diff --git a/weed/server/master_server.go b/weed/server/master_server.go index 89634b8f0..bc4eb96f3 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -220,7 +220,7 @@ func (ms *MasterServer) SetRaftServer(raftServer *RaftServer) { if ms.Topo.RaftServer.Leader() != "" { glog.V(0).Infof("[%s] %s becomes leader.", ms.Topo.RaftServer.Name(), ms.Topo.RaftServer.Leader()) ms.Topo.SetLastLeaderChangeTime(time.Now()) - if ms.Topo.RaftServer.Leader() == ms.Topo.RaftServer.Name() { + if pb.ServerAddress(ms.Topo.RaftServer.Leader()).Equals(pb.ServerAddress(ms.Topo.RaftServer.Name())) { go ms.ensureTopologyId() } } diff --git a/weed/server/raft_server_handlers.go b/weed/server/raft_server_handlers.go index 45a63b9e4..0ce7e632b 100644 --- a/weed/server/raft_server_handlers.go +++ b/weed/server/raft_server_handlers.go @@ -36,7 +36,7 @@ func (s *RaftServer) HealthzHandler(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusServiceUnavailable) return } - if s.serverAddr == leader { + if s.serverAddr.Equals(leader) { expBackoff := backoff.NewExponentialBackOff() expBackoff.InitialInterval = 20 * time.Millisecond expBackoff.MaxInterval = 1 * time.Second diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go index 2c484e7ce..6258da9c7 100644 --- a/weed/server/volume_grpc_client_to_master.go +++ b/weed/server/volume_grpc_client_to_master.go @@ -179,7 +179,7 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp } } } - if in.GetLeader() != "" && string(vs.currentMaster) != in.GetLeader() { + if in.GetLeader() != "" && !vs.currentMaster.Equals(pb.ServerAddress(in.GetLeader())) { glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), vs.currentMaster) newLeader = pb.ServerAddress(in.GetLeader()) doneChan <- nil diff --git a/weed/shell/command_volume_merge.go b/weed/shell/command_volume_merge.go index dc41c8480..aa3c0de16 100644 --- a/weed/shell/command_volume_merge.go +++ b/weed/shell/command_volume_merge.go @@ -493,7 +493,7 @@ func ensureVolumeReadonly(commandEnv *CommandEnv, replicas []*VolumeReplica) ([] func isReplicaServer(target pb.ServerAddress, replicas []*VolumeReplica) bool { for _, replica := range replicas { - if pb.NewServerAddressFromDataNode(replica.location.dataNode) == target { + if pb.NewServerAddressFromDataNode(replica.location.dataNode).Equals(target) { return true } } diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go index c22d2ce59..3070019c1 100644 --- a/weed/wdclient/masterclient.go +++ b/weed/wdclient/masterclient.go @@ -264,7 +264,7 @@ func (mc *MasterClient) tryConnectToMaster(ctx context.Context, master pb.Server // check if it is the leader to determine whether to reset the vidMap if resp.VolumeLocation != nil { - if resp.VolumeLocation.Leader != "" && string(master) != resp.VolumeLocation.Leader { + if resp.VolumeLocation.Leader != "" && !master.Equals(pb.ServerAddress(resp.VolumeLocation.Leader)) { glog.V(1).Infof("master %v redirected to leader %v", master, resp.VolumeLocation.Leader) nextHintedLeader = pb.ServerAddress(resp.VolumeLocation.Leader) stats.MasterClientConnectCounter.WithLabelValues(stats.RedirectedToLeader).Inc() @@ -295,8 +295,8 @@ func (mc *MasterClient) tryConnectToMaster(ctx context.Context, master pb.Server if resp.VolumeLocation != nil { // Check for leader change during the stream // If master announces a new leader, reconnect to it - if resp.VolumeLocation.Leader != "" && string(mc.GetMaster(ctx)) != resp.VolumeLocation.Leader { - glog.V(1).Infof("currentMaster %v redirected to leader %v", mc.GetMaster(ctx), resp.VolumeLocation.Leader) + if currentMaster := mc.GetMaster(ctx); resp.VolumeLocation.Leader != "" && !currentMaster.Equals(pb.ServerAddress(resp.VolumeLocation.Leader)) { + glog.V(1).Infof("currentMaster %v redirected to leader %v", currentMaster, resp.VolumeLocation.Leader) nextHintedLeader = pb.ServerAddress(resp.VolumeLocation.Leader) stats.MasterClientConnectCounter.WithLabelValues(stats.RedirectedToLeader).Inc() return nil