gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ensure latest stats are reported

Browse Source
This commit is contained in:
chrislu
2023-09-24 23:34:31 -07:00
parent dff2ce5d2f
commit c7e05e4e71
4 changed files with 21 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
weed/mq/broker/broker_grpc_balancer.go
View File

@@ -26,7 +26,9 @@ func (broker *MessageQueueBroker) ConnectToBalancer(stream mq_pb.SeaweedMessagin
brokerStats, found = broker.Balancer.Brokers.Get(initMessage.Broker)
if !found {
brokerStats = balancer.NewBrokerStats()
broker.Balancer.Brokers.Set(initMessage.Broker, brokerStats)
if !broker.Balancer.Brokers.SetIfAbsent(initMessage.Broker, brokerStats) {
brokerStats, _ = broker.Balancer.Brokers.Get(initMessage.Broker)
}
}
} else {
return status.Errorf(codes.InvalidArgument, "balancer init message is empty")

6
weed/mq/broker/broker_grpc_create.go
View File

@@ -43,7 +43,9 @@ func (broker *MessageQueueBroker) CreateTopic(ctx context.Context, request *mq_p
brokerStats, found := broker.Balancer.Brokers.Get(bpa.LeaderBroker)
if !found {
brokerStats = balancer.NewBrokerStats()
broker.Balancer.Brokers.Set(bpa.LeaderBroker, brokerStats)
if !broker.Balancer.Brokers.SetIfAbsent(bpa.LeaderBroker, brokerStats) {
brokerStats, _ = broker.Balancer.Brokers.Get(bpa.LeaderBroker)
}
}
brokerStats.RegisterAssignment(request.Topic, bpa.Partition)
return nil
@@ -52,6 +54,8 @@ func (broker *MessageQueueBroker) CreateTopic(ctx context.Context, request *mq_p
}
}
// TODO revert if some error happens in the middle of the assignments
return ret, err
}

2
weed/mq/broker/broker_stats.go
View File

@@ -61,7 +61,7 @@ func (broker *MessageQueueBroker) BrokerConnectToBalancer(self string) error {
if err != nil {
return fmt.Errorf("send stats message: %v", err)
}
glog.V(4).Infof("sent stats: %+v", stats)
glog.V(3).Infof("sent stats: %+v", stats)
time.Sleep(time.Millisecond*5000 + time.Duration(rand.Intn(1000))*time.Millisecond)
}