gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

cluster.ps can list brokers

Browse source
This commit is contained in:
chrislu 2022-07-01 23:34:51 -07:00
parent ac66227798
commit 87ccef72db
8 changed files with 55 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
weed/command/command.go
View file

@ -28,12 +28,12 @@ var Commands = []*Command{
cmdFilerSynchronize,
cmdFix,
cmdFuse,
cmdIam,
cmdMaster,
cmdMasterFollower,
cmdMount,
cmdS3,
cmdIam,
cmdMqBroker,
cmdS3,
cmdScaffold,
cmdServer,
cmdShell,

8
weed/command/mq_broker.go
View file

@ -23,6 +23,8 @@ var (
)
type MessageQueueBrokerOptions struct {
masters *string
filerGroup *string
filer *string
ip *string
port *int
@ -32,7 +34,9 @@ type MessageQueueBrokerOptions struct {
func init() {
cmdMqBroker.Run = runMqBroker // break init cycle
mqBrokerStandaloneOptions.masters = cmdMqBroker.Flag.String("master", "localhost:9333", "comma-separated master servers")
mqBrokerStandaloneOptions.filer = cmdMqBroker.Flag.String("filer", "localhost:8888", "filer server address")
mqBrokerStandaloneOptions.filerGroup = cmdMqBroker.Flag.String("filerGroup", "", "share metadata with other filers in the same filerGroup")
mqBrokerStandaloneOptions.ip = cmdMqBroker.Flag.String("ip", util.DetectedHostAddress(), "broker host address")
mqBrokerStandaloneOptions.port = cmdMqBroker.Flag.Int("port", 17777, "broker gRPC listen port")
mqBrokerStandaloneOptions.cpuprofile = cmdMqBroker.Flag.String("cpuprofile", "", "cpu profile output file")
@ -85,7 +89,9 @@ func (mqBrokerOpt *MessageQueueBrokerOptions) startQueueServer() bool {
}
}
qs, err := broker.NewMessageBroker(&broker.MessageBrokerOption{
qs, err := broker.NewMessageBroker(&broker.MessageQueueBrokerOption{
Masters: pb.ServerAddresses(*mqBrokerOpt.masters).ToAddressMap(),
FilerGroup: *mqBrokerOpt.filerGroup,
Filers: []pb.ServerAddress{filerAddress},
DefaultReplication: "",
MaxMB: 0,

1
weed/command/server.go
View file

@ -225,6 +225,7 @@ func runServer(cmd *Command, args []string) bool {
iamOptions.filer = &filerAddress
webdavOptions.filer = &filerAddress
mqBrokerOptions.filer = &filerAddress
mqBrokerOptions.filerGroup = filerOptions.filerGroup
go stats_collect.StartMetricsServer(*serverMetricsHttpPort)

6
weed/mq/broker/broker_grpc_server.go
View file

@ -9,11 +9,11 @@ import (
"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
)
func (broker *MessageBroker) ConfigureTopic(c context.Context, request *mq_pb.ConfigureTopicRequest) (*mq_pb.ConfigureTopicResponse, error) {
func (broker *MessageQueueBroker) ConfigureTopic(c context.Context, request *mq_pb.ConfigureTopicRequest) (*mq_pb.ConfigureTopicResponse, error) {
panic("implement me")
}
func (broker *MessageBroker) DeleteTopic(c context.Context, request *mq_pb.DeleteTopicRequest) (*mq_pb.DeleteTopicResponse, error) {
func (broker *MessageQueueBroker) DeleteTopic(c context.Context, request *mq_pb.DeleteTopicRequest) (*mq_pb.DeleteTopicResponse, error) {
resp := &mq_pb.DeleteTopicResponse{}
dir, entry := genTopicDirEntry(request.Namespace, request.Topic)
if exists, err := filer_pb.Exists(broker, dir, entry, true); err != nil {
@ -24,7 +24,7 @@ func (broker *MessageBroker) DeleteTopic(c context.Context, request *mq_pb.Delet
return resp, nil
}
func (broker *MessageBroker) GetTopicConfiguration(c context.Context, request *mq_pb.GetTopicConfigurationRequest) (*mq_pb.GetTopicConfigurationResponse, error) {
func (broker *MessageQueueBroker) GetTopicConfiguration(c context.Context, request *mq_pb.GetTopicConfigurationRequest) (*mq_pb.GetTopicConfigurationResponse, error) {
panic("implement me")
}

4
weed/mq/broker/broker_grpc_server_discovery.go
View file

@ -26,7 +26,7 @@ If one of the pub or sub connects very late, and the system topo changed quite a
*/
func (broker *MessageBroker) FindBroker(c context.Context, request *mq_pb.FindBrokerRequest) (*mq_pb.FindBrokerResponse, error) {
func (broker *MessageQueueBroker) FindBroker(c context.Context, request *mq_pb.FindBrokerRequest) (*mq_pb.FindBrokerResponse, error) {
t := &mq_pb.FindBrokerResponse{}
var peers []string
@ -61,7 +61,7 @@ func (broker *MessageBroker) FindBroker(c context.Context, request *mq_pb.FindBr
}
func (broker *MessageBroker) checkFilers() {
func (broker *MessageQueueBroker) checkFilers() {
// contact a filer about masters
var masters []pb.ServerAddress

2
weed/mq/broker/broker_grpc_server_publish.go
View file

@ -13,7 +13,7 @@ import (
"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
)
func (broker *MessageBroker) Publish(stream mq_pb.SeaweedMessaging_PublishServer) error {
func (broker *MessageQueueBroker) Publish(stream mq_pb.SeaweedMessaging_PublishServer) error {
// process initial request
in, err := stream.Recv()

19
weed/mq/broker/broker_server.go
View file

@ -2,7 +2,9 @@ package broker
import (
"context"
"github.com/chrislusf/seaweedfs/weed/cluster"
"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
"github.com/chrislusf/seaweedfs/weed/wdclient"
"time"
"google.golang.org/grpc"
@ -14,6 +16,8 @@ import (
)
type MessageQueueBrokerOption struct {
Masters map[string]pb.ServerAddress
FilerGroup string
Filers []pb.ServerAddress
DefaultReplication string
MaxMB int
@ -26,23 +30,26 @@ type MessageQueueBroker struct {
mq_pb.UnimplementedSeaweedMessagingServer
option *MessageQueueBrokerOption
grpcDialOption grpc.DialOption
MasterClient *wdclient.MasterClient
topicManager *TopicManager
}
func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.DialOption) (messageBroker *MessageQueueBroker, err error) {
func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.DialOption) (mqBroker *MessageQueueBroker, err error) {
messageBroker = &MessageQueueBroker{
mqBroker = &MessageQueueBroker{
option: option,
grpcDialOption: grpcDialOption,
MasterClient: wdclient.NewMasterClient(grpcDialOption, option.FilerGroup, cluster.BrokerType, pb.NewServerAddress(option.Ip, option.Port, 0), "", option.Masters),
}
messageBroker.topicManager = NewTopicManager(messageBroker)
mqBroker.topicManager = NewTopicManager(mqBroker)
messageBroker.checkFilers()
mqBroker.checkFilers()
go messageBroker.keepConnectedToOneFiler()
go mqBroker.keepConnectedToOneFiler()
go mqBroker.MasterClient.KeepConnectedToMaster()
return messageBroker, nil
return mqBroker, nil
}
func (broker *MessageQueueBroker) keepConnectedToOneFiler() {

26
weed/shell/command_cluster_ps.go
View file

@ -40,7 +40,9 @@ func (c *commandClusterPs) Do(args []string, commandEnv *CommandEnv, writer io.W
}
var filerNodes []*master_pb.ListClusterNodesResponse_ClusterNode
var mqBrokerNodes []*master_pb.ListClusterNodesResponse_ClusterNode
// get the list of filers
err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
ClientType: cluster.FilerType,
@ -57,6 +59,30 @@ func (c *commandClusterPs) Do(args []string, commandEnv *CommandEnv, writer io.W
return
}
// get the list of message queue brokers
err = commandEnv.MasterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
ClientType: cluster.BrokerType,
FilerGroup: *commandEnv.option.FilerGroup,
})
if err != nil {
return err
}
mqBrokerNodes = resp.ClusterNodes
return err
})
if err != nil {
return
}
if len(mqBrokerNodes) > 0 {
fmt.Fprintf(writer, "* message queue brokers %d\n", len(mqBrokerNodes))
for _, node := range mqBrokerNodes {
fmt.Fprintf(writer, " * %s (%v)\n", node.Address, node.Version)
}
}
fmt.Fprintf(writer, "* filers %d\n", len(filerNodes))
for _, node := range filerNodes {
fmt.Fprintf(writer, " * %s (%v)\n", node.Address, node.Version)