diff --git a/weed/mq/broker/broker_grpc_assign.go b/weed/mq/broker/broker_grpc_assign.go index 264565b7b..48a20e334 100644 --- a/weed/mq/broker/broker_grpc_assign.go +++ b/weed/mq/broker/broker_grpc_assign.go @@ -14,7 +14,6 @@ import ( // AssignTopicPartitions Runs on the assigned broker, to execute the topic partition assignment func (b *MessageQueueBroker) AssignTopicPartitions(c context.Context, request *mq_pb.AssignTopicPartitionsRequest) (*mq_pb.AssignTopicPartitionsResponse, error) { ret := &mq_pb.AssignTopicPartitionsResponse{} - self := pb.ServerAddress(fmt.Sprintf("%s:%d", b.option.Ip, b.option.Port)) // drain existing topic partition subscriptions for _, assignment := range request.BrokerPartitionAssignments { @@ -27,7 +26,7 @@ func (b *MessageQueueBroker) AssignTopicPartitions(c context.Context, request *m } else { var localPartition *topic.LocalPartition if localPartition = b.localTopicManager.GetTopicPartition(t, partition); localPartition == nil { - localPartition = topic.FromPbBrokerPartitionAssignment(self, partition, assignment, b.genLogFlushFunc(t, assignment.Partition), b.genLogOnDiskReadFunc(t, assignment.Partition)) + localPartition = topic.NewLocalPartition(partition, b.genLogFlushFunc(t, assignment.Partition), b.genLogOnDiskReadFunc(t, assignment.Partition)) b.localTopicManager.AddTopicPartition(t, localPartition) } } diff --git a/weed/mq/broker/broker_topic_conf_read_write.go b/weed/mq/broker/broker_topic_conf_read_write.go index 987c60243..dfdf9a708 100644 --- a/weed/mq/broker/broker_topic_conf_read_write.go +++ b/weed/mq/broker/broker_topic_conf_read_write.go @@ -88,7 +88,7 @@ func (b *MessageQueueBroker) genLocalPartitionFromFiler(t topic.Topic, partition self := b.option.BrokerAddress() for _, assignment := range conf.BrokerPartitionAssignments { if assignment.LeaderBroker == string(self) && partition.Equals(topic.FromPbPartition(assignment.Partition)) { - localPartition = topic.FromPbBrokerPartitionAssignment(b.option.BrokerAddress(), partition, assignment, b.genLogFlushFunc(t, assignment.Partition), b.genLogOnDiskReadFunc(t, assignment.Partition)) + localPartition = topic.NewLocalPartition(partition, b.genLogFlushFunc(t, assignment.Partition), b.genLogOnDiskReadFunc(t, assignment.Partition)) b.localTopicManager.AddTopicPartition(t, localPartition) isGenerated = true break diff --git a/weed/mq/pub_balancer/balance_brokers_test.go b/weed/mq/pub_balancer/balance_brokers_test.go index 54667d154..122984f0d 100644 --- a/weed/mq/pub_balancer/balance_brokers_test.go +++ b/weed/mq/pub_balancer/balance_brokers_test.go @@ -21,8 +21,6 @@ func TestBalanceTopicPartitionOnBrokers(t *testing.T) { Topic: topic.Topic{Namespace: "topic1", Name: "topic1"}, Partition: topic.Partition{RangeStart: 0, RangeStop: 512, RingSize: 1024}, }, - ConsumerCount: 1, - IsLeader: true, }) broker2Stats := &BrokerStats{ TopicPartitionCount: 2, @@ -35,16 +33,12 @@ func TestBalanceTopicPartitionOnBrokers(t *testing.T) { Topic: topic.Topic{Namespace: "topic1", Name: "topic1"}, Partition: topic.Partition{RangeStart: 512, RangeStop: 1024, RingSize: 1024}, }, - ConsumerCount: 1, - IsLeader: true, }) broker2Stats.TopicPartitionStats.Set("topic2:0", &TopicPartitionStats{ TopicPartition: topic.TopicPartition{ Topic: topic.Topic{Namespace: "topic2", Name: "topic2"}, Partition: topic.Partition{RangeStart: 0, RangeStop: 1024, RingSize: 1024}, }, - ConsumerCount: 1, - IsLeader: true, }) brokers.Set("broker1", broker1Stats) brokers.Set("broker2", broker2Stats) diff --git a/weed/mq/pub_balancer/broker_stats.go b/weed/mq/pub_balancer/broker_stats.go index b4bb28e42..00f1f80ca 100644 --- a/weed/mq/pub_balancer/broker_stats.go +++ b/weed/mq/pub_balancer/broker_stats.go @@ -9,15 +9,16 @@ import ( type BrokerStats struct { TopicPartitionCount int32 - ConsumerCount int32 + PublisherCount int32 + SubscriberCount int32 CpuUsagePercent int32 TopicPartitionStats cmap.ConcurrentMap[string, *TopicPartitionStats] // key: topic_partition Topics []topic.Topic } type TopicPartitionStats struct { topic.TopicPartition - ConsumerCount int32 - IsLeader bool + PublisherCount int32 + SubscriberCount int32 } func NewBrokerStats() *BrokerStats { @@ -26,15 +27,15 @@ func NewBrokerStats() *BrokerStats { } } func (bs *BrokerStats) String() string { - return fmt.Sprintf("BrokerStats{TopicPartitionCount:%d, ConsumerCount:%d, CpuUsagePercent:%d, Stats:%+v}", - bs.TopicPartitionCount, bs.ConsumerCount, bs.CpuUsagePercent, bs.TopicPartitionStats.Items()) + return fmt.Sprintf("BrokerStats{TopicPartitionCount:%d, Publishers:%d, Subscribers:%d CpuUsagePercent:%d, Stats:%+v}", + bs.TopicPartitionCount, bs.PublisherCount, bs.SubscriberCount, bs.CpuUsagePercent, bs.TopicPartitionStats.Items()) } func (bs *BrokerStats) UpdateStats(stats *mq_pb.BrokerStats) { bs.TopicPartitionCount = int32(len(stats.Stats)) bs.CpuUsagePercent = stats.CpuUsagePercent - var consumerCount int32 + var publisherCount, subscriberCount int32 currentTopicPartitions := bs.TopicPartitionStats.Items() for _, topicPartitionStats := range stats.Stats { tps := &TopicPartitionStats{ @@ -47,10 +48,11 @@ func (bs *BrokerStats) UpdateStats(stats *mq_pb.BrokerStats) { UnixTimeNs: topicPartitionStats.Partition.UnixTimeNs, }, }, - ConsumerCount: topicPartitionStats.ConsumerCount, - IsLeader: topicPartitionStats.IsLeader, + PublisherCount: topicPartitionStats.PublisherCount, + SubscriberCount: topicPartitionStats.SubscriberCount, } - consumerCount += topicPartitionStats.ConsumerCount + publisherCount += topicPartitionStats.PublisherCount + subscriberCount += topicPartitionStats.SubscriberCount key := tps.TopicPartition.String() bs.TopicPartitionStats.Set(key, tps) delete(currentTopicPartitions, key) @@ -59,8 +61,8 @@ func (bs *BrokerStats) UpdateStats(stats *mq_pb.BrokerStats) { for key := range currentTopicPartitions { bs.TopicPartitionStats.Remove(key) } - bs.ConsumerCount = consumerCount - + bs.PublisherCount = publisherCount + bs.SubscriberCount = subscriberCount } func (bs *BrokerStats) RegisterAssignment(t *mq_pb.Topic, partition *mq_pb.Partition, isAdd bool) { @@ -74,8 +76,8 @@ func (bs *BrokerStats) RegisterAssignment(t *mq_pb.Topic, partition *mq_pb.Parti UnixTimeNs: partition.UnixTimeNs, }, }, - ConsumerCount: 0, - IsLeader: true, + PublisherCount: 0, + SubscriberCount: 0, } key := tps.TopicPartition.String() if isAdd { diff --git a/weed/mq/pub_balancer/repair.go b/weed/mq/pub_balancer/repair.go index 0ab1a5ea9..0f307c9eb 100644 --- a/weed/mq/pub_balancer/repair.go +++ b/weed/mq/pub_balancer/repair.go @@ -14,8 +14,7 @@ func (balancer *Balancer) RepairTopics() []BalanceAction { } type TopicPartitionInfo struct { - Leader string - Followers []string + Broker string } // RepairMissingTopicPartitions check the stats of all brokers, @@ -38,11 +37,7 @@ func RepairMissingTopicPartitions(brokers cmap.ConcurrentMap[string, *BrokerStat tpi = &TopicPartitionInfo{} topicPartitionToInfo[topicPartitionStat.Partition] = tpi } - if topicPartitionStat.IsLeader { - tpi.Leader = broker - } else { - tpi.Followers = append(tpi.Followers, broker) - } + tpi.Broker = broker } } diff --git a/weed/mq/topic/local_manager.go b/weed/mq/topic/local_manager.go index aa2eefcdc..02334a27e 100644 --- a/weed/mq/topic/local_manager.go +++ b/weed/mq/topic/local_manager.go @@ -96,8 +96,9 @@ func (manager *LocalTopicManager) CollectStats(duration time.Duration) *mq_pb.Br Namespace: string(localTopic.Namespace), Name: localTopic.Name, }, - Partition: localPartition.Partition.ToPbPartition(), - ConsumerCount: localPartition.ConsumerCount, + Partition: localPartition.Partition.ToPbPartition(), + PublisherCount: int32(localPartition.Publishers.Size()), + SubscriberCount: int32(localPartition.Subscribers.Size()), } // fmt.Printf("collect topic %+v partition %+v\n", topicPartition, localPartition.Partition) } diff --git a/weed/mq/topic/local_partition.go b/weed/mq/topic/local_partition.go index 5df2535a6..8b9970f20 100644 --- a/weed/mq/topic/local_partition.go +++ b/weed/mq/topic/local_partition.go @@ -22,10 +22,7 @@ type LocalPartition struct { ListenersCond *sync.Cond Partition - isLeader bool - FollowerBrokers []pb.ServerAddress LogBuffer *log_buffer.LogBuffer - ConsumerCount int32 Publishers *LocalPartitionPublishers Subscribers *LocalPartitionSubscribers FollowerId int32 @@ -37,11 +34,9 @@ type LocalPartition struct { var TIME_FORMAT = "2006-01-02-15-04-05" -func NewLocalPartition(partition Partition, isLeader bool, followerBrokers []pb.ServerAddress, logFlushFn log_buffer.LogFlushFuncType, readFromDiskFn log_buffer.LogReadFromDiskFuncType) *LocalPartition { +func NewLocalPartition(partition Partition, logFlushFn log_buffer.LogFlushFuncType, readFromDiskFn log_buffer.LogReadFromDiskFuncType) *LocalPartition { lp := &LocalPartition{ Partition: partition, - isLeader: isLeader, - FollowerBrokers: followerBrokers, Publishers: NewLocalPartitionPublishers(), Subscribers: NewLocalPartitionSubscribers(), } @@ -121,15 +116,6 @@ func (p *LocalPartition) GetEarliestInMemoryMessagePosition() log_buffer.Message return p.LogBuffer.GetEarliestPosition() } -func FromPbBrokerPartitionAssignment(self pb.ServerAddress, partition Partition, assignment *mq_pb.BrokerPartitionAssignment, logFlushFn log_buffer.LogFlushFuncType, readFromDiskFn log_buffer.LogReadFromDiskFuncType) *LocalPartition { - isLeader := assignment.LeaderBroker == string(self) - followers := make([]pb.ServerAddress, len(assignment.FollowerBrokers)) - for i, followerBroker := range assignment.FollowerBrokers { - followers[i] = pb.ServerAddress(followerBroker) - } - return NewLocalPartition(partition, isLeader, followers, logFlushFn, readFromDiskFn) -} - func (p *LocalPartition) closePublishers() { p.Publishers.SignalShutdown() } diff --git a/weed/pb/mq.proto b/weed/pb/mq.proto index 45253059f..999985b7d 100644 --- a/weed/pb/mq.proto +++ b/weed/pb/mq.proto @@ -97,8 +97,8 @@ message BrokerStats { message TopicPartitionStats { Topic topic = 1; Partition partition = 2; - int32 consumer_count = 3; - bool is_leader = 4; + int32 publisher_count = 3; + int32 subscriber_count = 4; } diff --git a/weed/pb/mq_pb/mq.pb.go b/weed/pb/mq_pb/mq.pb.go index b5182e6af..8beccd946 100644 --- a/weed/pb/mq_pb/mq.pb.go +++ b/weed/pb/mq_pb/mq.pb.go @@ -476,10 +476,10 @@ type TopicPartitionStats struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - Topic *Topic `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"` - Partition *Partition `protobuf:"bytes,2,opt,name=partition,proto3" json:"partition,omitempty"` - ConsumerCount int32 `protobuf:"varint,3,opt,name=consumer_count,json=consumerCount,proto3" json:"consumer_count,omitempty"` - IsLeader bool `protobuf:"varint,4,opt,name=is_leader,json=isLeader,proto3" json:"is_leader,omitempty"` + Topic *Topic `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"` + Partition *Partition `protobuf:"bytes,2,opt,name=partition,proto3" json:"partition,omitempty"` + PublisherCount int32 `protobuf:"varint,3,opt,name=publisher_count,json=publisherCount,proto3" json:"publisher_count,omitempty"` + SubscriberCount int32 `protobuf:"varint,4,opt,name=subscriber_count,json=subscriberCount,proto3" json:"subscriber_count,omitempty"` } func (x *TopicPartitionStats) Reset() { @@ -528,18 +528,18 @@ func (x *TopicPartitionStats) GetPartition() *Partition { return nil } -func (x *TopicPartitionStats) GetConsumerCount() int32 { +func (x *TopicPartitionStats) GetPublisherCount() int32 { if x != nil { - return x.ConsumerCount + return x.PublisherCount } return 0 } -func (x *TopicPartitionStats) GetIsLeader() bool { +func (x *TopicPartitionStats) GetSubscriberCount() int32 { if x != nil { - return x.IsLeader + return x.SubscriberCount } - return false + return 0 } type PublisherToPubBalancerRequest struct { @@ -2762,7 +2762,7 @@ var file_mq_proto_rawDesc = []byte{ 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x21, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x74, 0x61, 0x74, 0x73, 0x52, 0x05, 0x76, 0x61, 0x6c, - 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xbb, 0x01, 0x0a, 0x13, 0x54, 0x6f, 0x70, 0x69, 0x63, + 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xcb, 0x01, 0x0a, 0x13, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x74, 0x61, 0x74, 0x73, 0x12, 0x29, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x54, 0x6f, 0x70, @@ -2770,11 +2770,12 @@ var file_mq_proto_rawDesc = []byte{ 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x50, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, - 0x12, 0x25, 0x0a, 0x0e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x5f, 0x63, 0x6f, 0x75, - 0x6e, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0d, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6d, - 0x65, 0x72, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x1b, 0x0a, 0x09, 0x69, 0x73, 0x5f, 0x6c, 0x65, - 0x61, 0x64, 0x65, 0x72, 0x18, 0x04, 0x20, 0x01, 0x28, 0x08, 0x52, 0x08, 0x69, 0x73, 0x4c, 0x65, - 0x61, 0x64, 0x65, 0x72, 0x22, 0xd3, 0x01, 0x0a, 0x1d, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, + 0x12, 0x27, 0x0a, 0x0f, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x65, 0x72, 0x5f, 0x63, 0x6f, + 0x75, 0x6e, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0e, 0x70, 0x75, 0x62, 0x6c, 0x69, + 0x73, 0x68, 0x65, 0x72, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x29, 0x0a, 0x10, 0x73, 0x75, 0x62, + 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x72, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x04, 0x20, + 0x01, 0x28, 0x05, 0x52, 0x0f, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x72, 0x43, + 0x6f, 0x75, 0x6e, 0x74, 0x22, 0xd3, 0x01, 0x0a, 0x1d, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x65, 0x72, 0x54, 0x6f, 0x50, 0x75, 0x62, 0x42, 0x61, 0x6c, 0x61, 0x6e, 0x63, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x4d, 0x0a, 0x04, 0x69, 0x6e, 0x69, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x37, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67,