seaweedfs/weed/mq/broker/broker_stats.go

74 lines
2 KiB
Go
Raw Normal View History

2023-09-16 22:05:38 +00:00
package broker
import (
"context"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/glog"
2023-09-24 21:22:11 +00:00
"github.com/seaweedfs/seaweedfs/weed/mq/balancer"
2023-09-16 22:05:38 +00:00
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
"math/rand"
"time"
)
// BrokerConnectToBalancer connects to the broker balancer and sends stats
func (broker *MessageQueueBroker) BrokerConnectToBalancer(self string) error {
// find the lock owner
var brokerBalancer string
err := broker.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.FindLockOwner(context.Background(), &filer_pb.FindLockOwnerRequest{
2023-09-24 21:22:11 +00:00
Name: balancer.LockBrokerBalancer,
2023-09-16 22:05:38 +00:00
})
if err != nil {
return err
}
brokerBalancer = resp.Owner
return nil
})
if err != nil {
return err
}
2023-09-24 21:22:11 +00:00
broker.currentBalancer = pb.ServerAddress(brokerBalancer)
2023-09-16 22:05:38 +00:00
glog.V(1).Infof("broker %s found balancer %s", self, brokerBalancer)
// connect to the lock owner
err = pb.WithBrokerGrpcClient(false, brokerBalancer, broker.grpcDialOption, func(client mq_pb.SeaweedMessagingClient) error {
stream, err := client.ConnectToBalancer(context.Background())
if err != nil {
return fmt.Errorf("connect to balancer %v: %v", brokerBalancer, err)
}
defer stream.CloseSend()
err = stream.Send(&mq_pb.ConnectToBalancerRequest{
Message: &mq_pb.ConnectToBalancerRequest_Init{
Init: &mq_pb.ConnectToBalancerRequest_InitMessage{
Broker: self,
},
},
})
if err != nil {
return fmt.Errorf("send init message: %v", err)
}
for {
stats := broker.localTopicManager.CollectStats(time.Second * 5)
err = stream.Send(&mq_pb.ConnectToBalancerRequest{
Message: &mq_pb.ConnectToBalancerRequest_Stats{
Stats: stats,
},
})
if err != nil {
return fmt.Errorf("send stats message: %v", err)
}
2023-09-25 06:34:31 +00:00
glog.V(3).Infof("sent stats: %+v", stats)
2023-09-16 22:05:38 +00:00
time.Sleep(time.Millisecond*5000 + time.Duration(rand.Intn(1000))*time.Millisecond)
}
return nil
})
return err
}