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