seaweedfs/weed/mq/broker/broker_server.go

146 lines
4.3 KiB
Go
Raw Normal View History

2020-04-18 08:12:01 +00:00
package broker
2020-03-04 08:39:47 +00:00
import (
2023-09-16 22:05:38 +00:00
"fmt"
"github.com/seaweedfs/seaweedfs/weed/glog"
2023-09-15 06:49:05 +00:00
"github.com/seaweedfs/seaweedfs/weed/mq/balancer"
"github.com/seaweedfs/seaweedfs/weed/mq/topic"
"time"
"github.com/seaweedfs/seaweedfs/weed/cluster"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
2020-03-04 08:39:47 +00:00
"google.golang.org/grpc"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
2020-03-04 08:39:47 +00:00
)
2022-07-02 05:44:28 +00:00
type MessageQueueBrokerOption struct {
2022-07-02 06:34:51 +00:00
Masters map[string]pb.ServerAddress
FilerGroup string
2022-07-03 07:29:25 +00:00
DataCenter string
Rack string
2020-03-04 08:39:47 +00:00
DefaultReplication string
MaxMB int
2020-05-05 09:05:28 +00:00
Ip string
2020-03-04 08:39:47 +00:00
Port int
2020-04-17 09:29:00 +00:00
Cipher bool
2020-03-04 08:39:47 +00:00
}
2022-07-02 05:44:28 +00:00
type MessageQueueBroker struct {
2022-07-02 05:43:25 +00:00
mq_pb.UnimplementedSeaweedMessagingServer
option *MessageQueueBrokerOption
grpcDialOption grpc.DialOption
MasterClient *wdclient.MasterClient
filers map[pb.ServerAddress]struct{}
currentFiler pb.ServerAddress
localTopicManager *topic.LocalTopicManager
2023-09-15 06:49:05 +00:00
Balancer *balancer.Balancer
2023-09-16 22:05:38 +00:00
lockAsBalancer *cluster.LiveLock
2023-09-19 21:08:17 +00:00
currentBalancer pb.ServerAddress
2020-03-04 08:39:47 +00:00
}
2022-07-02 06:34:51 +00:00
func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.DialOption) (mqBroker *MessageQueueBroker, err error) {
2020-03-04 08:39:47 +00:00
2022-07-02 06:34:51 +00:00
mqBroker = &MessageQueueBroker{
option: option,
grpcDialOption: grpcDialOption,
2023-09-15 06:49:05 +00:00
MasterClient: wdclient.NewMasterClient(grpcDialOption, option.FilerGroup, cluster.BrokerType, pb.NewServerAddress(option.Ip, option.Port, 0), option.DataCenter, option.Rack, *pb.NewServiceDiscoveryFromMap(option.Masters)),
filers: make(map[pb.ServerAddress]struct{}),
localTopicManager: topic.NewLocalTopicManager(),
2023-09-15 06:49:05 +00:00
Balancer: balancer.NewBalancer(),
2020-03-04 08:39:47 +00:00
}
mqBroker.MasterClient.SetOnPeerUpdateFn(mqBroker.OnBrokerUpdate)
2022-07-02 06:34:51 +00:00
go mqBroker.MasterClient.KeepConnectedToMaster()
2020-03-04 08:39:47 +00:00
2022-07-12 09:00:54 +00:00
existingNodes := cluster.ListExistingPeerUpdates(mqBroker.MasterClient.GetMaster(), grpcDialOption, option.FilerGroup, cluster.FilerType)
for _, newNode := range existingNodes {
mqBroker.OnBrokerUpdate(newNode, time.Now())
}
2023-09-16 22:05:38 +00:00
// keep connecting to balancer
go func() {
for mqBroker.currentFiler == "" {
time.Sleep(time.Millisecond * 237)
}
self := fmt.Sprintf("%s:%d", option.Ip, option.Port)
glog.V(1).Infof("broker %s found filer %s", self, mqBroker.currentFiler)
lockClient := cluster.NewLockClient(grpcDialOption, mqBroker.currentFiler)
mqBroker.lockAsBalancer = lockClient.StartLock(LockBrokerBalancer, self)
for {
err := mqBroker.BrokerConnectToBalancer(self)
if err != nil {
fmt.Printf("BrokerConnectToBalancer: %v\n", err)
}
time.Sleep(time.Second)
}
}()
2022-07-02 06:34:51 +00:00
return mqBroker, nil
2020-03-04 08:39:47 +00:00
}
2022-07-12 09:00:54 +00:00
func (broker *MessageQueueBroker) OnBrokerUpdate(update *master_pb.ClusterNodeUpdate, startFrom time.Time) {
if update.NodeType != cluster.FilerType {
return
}
address := pb.ServerAddress(update.Address)
if update.IsAdd {
broker.filers[address] = struct{}{}
if broker.currentFiler == "" {
broker.currentFiler = address
}
} else {
delete(broker.filers, address)
if broker.currentFiler == address {
for filer := range broker.filers {
2022-07-12 09:00:54 +00:00
broker.currentFiler = filer
break
}
}
}
}
func (broker *MessageQueueBroker) GetFiler() pb.ServerAddress {
return broker.currentFiler
}
2022-07-16 17:49:34 +00:00
func (broker *MessageQueueBroker) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
2020-03-04 08:39:47 +00:00
return pb.WithFilerClient(streamingMode, 0, broker.GetFiler(), broker.grpcDialOption, fn)
2022-07-16 17:49:34 +00:00
}
func (broker *MessageQueueBroker) AdjustedUrl(location *filer_pb.Location) string {
return location.Url
2020-03-04 08:39:47 +00:00
}
func (broker *MessageQueueBroker) GetDataCenter() string {
return ""
}
2022-07-02 05:44:28 +00:00
func (broker *MessageQueueBroker) withMasterClient(streamingMode bool, master pb.ServerAddress, fn func(client master_pb.SeaweedClient) error) error {
2020-03-04 08:39:47 +00:00
return pb.WithMasterClient(streamingMode, master, broker.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
2020-03-04 08:39:47 +00:00
return fn(client)
})
}
2022-07-11 07:20:27 +00:00
func (broker *MessageQueueBroker) withBrokerClient(streamingMode bool, server pb.ServerAddress, fn func(client mq_pb.SeaweedMessagingClient) error) error {
2023-08-28 00:50:59 +00:00
return pb.WithBrokerGrpcClient(streamingMode, server.String(), broker.grpcDialOption, func(client mq_pb.SeaweedMessagingClient) error {
2022-07-11 07:20:27 +00:00
return fn(client)
})
}