mirrors
/
seaweedfs
mirror of https://github.com/chrislusf/seaweedfs
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

create topic and report topic

This commit is contained in:
chrislu 2023-09-24 21:19:51 -07:00
parent b3f94feede
commit 3cf9b8d621
10 changed files with 650 additions and 425 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
weed/mq/balancer/allocate.go
View File

@ -6,10 +6,10 @@ import (
"math/rand"
)
func allocateTopicPartitions(brokers cmap.ConcurrentMap[string, *BrokerStats], partitionCount int) (assignments []*mq_pb.BrokerPartitionAssignment) {
func allocateTopicPartitions(brokers cmap.ConcurrentMap[string, *BrokerStats], partitionCount int32) (assignments []*mq_pb.BrokerPartitionAssignment) {
// divide the ring into partitions
rangeSize := MaxPartitionCount / partitionCount
for i := 0; i < partitionCount; i++ {
for i := int32(0); i < partitionCount; i++ {
assignment := &mq_pb.BrokerPartitionAssignment{
Partition: &mq_pb.Partition{
RingSize: MaxPartitionCount,
@ -35,13 +35,13 @@ func allocateTopicPartitions(brokers cmap.ConcurrentMap[string, *BrokerStats], p
// for now: randomly pick brokers
// TODO pick brokers based on the broker stats
func pickBrokers(brokers cmap.ConcurrentMap[string, *BrokerStats], count int) []string {
func pickBrokers(brokers cmap.ConcurrentMap[string, *BrokerStats], count int32) []string {
candidates := make([]string, 0, brokers.Count())
for brokerStatsItem := range brokers.IterBuffered() {
candidates = append(candidates, brokerStatsItem.Key)
}
pickedBrokers := make([]string, 0, count)
for i := 0; i < count; i++ {
for i := int32(0); i < count; i++ {
p := rand.Int() % len(candidates)
if p < 0 {
p = -p

5
weed/mq/balancer/balancer.go
View File

@ -52,6 +52,11 @@ func (bs *BrokerStats) UpdateStats(stats *mq_pb.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.Stats.Items())
}
type TopicPartition struct {
Namespace string
Topic string

2
weed/mq/balancer/lookup.go
View File

@ -47,5 +47,5 @@ func (b *Balancer) LookupOrAllocateTopicPartitions(topic *mq_pb.Topic, publish b
if b.Brokers.IsEmpty() {
return nil, ErrNoBroker
}
return allocateTopicPartitions(b.Brokers, 6), nil
return allocateTopicPartitions(b.Brokers, partitionCount), nil
}

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

@ -43,6 +43,7 @@ func (broker *MessageQueueBroker) ConnectToBalancer(stream mq_pb.SeaweedMessagin
brokerStats.UpdateStats(receivedStats)
glog.V(3).Infof("broker %s stats: %+v", initMessage.Broker, brokerStats)
glog.V(3).Infof("received stats: %+v", receivedStats)
}
}

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

@ -3,6 +3,7 @@ package broker
import (
"context"
"github.com/seaweedfs/seaweedfs/weed/mq/topic"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
@ -27,6 +28,18 @@ func (broker *MessageQueueBroker) CreateTopic(ctx context.Context, request *mq_p
ret := &mq_pb.CreateTopicResponse{}
ret.BrokerPartitionAssignments, err = broker.Balancer.LookupOrAllocateTopicPartitions(request.Topic, true, request.PartitionCount)
for _, bpa := range ret.BrokerPartitionAssignments {
if doCreateErr := broker.withBrokerClient(false, pb.ServerAddress(bpa.LeaderBroker), func(client mq_pb.SeaweedMessagingClient) error {
_, doCreateErr := client.DoCreateTopic(ctx, &mq_pb.DoCreateTopicRequest{
Topic: request.Topic,
Partition: bpa.Partition,
})
return doCreateErr
}); doCreateErr != nil {
return nil, doCreateErr
}
}
return ret, err
}

35
weed/mq/broker/broker_grpc_lookup.go
View File

@ -45,3 +45,38 @@ func (broker *MessageQueueBroker) CheckTopicPartitionsStatus(c context.Context,
ret := &mq_pb.CheckTopicPartitionsStatusResponse{}
return ret, nil
}
func (broker *MessageQueueBroker) ListTopics(ctx context.Context, request *mq_pb.ListTopicsRequest) (resp *mq_pb.ListTopicsResponse, err error) {
if broker.currentBalancer == "" {
return nil, status.Errorf(codes.Unavailable, "no balancer")
}
if !broker.lockAsBalancer.IsLocked() {
proxyErr := broker.withBrokerClient(false, broker.currentBalancer, func(client mq_pb.SeaweedMessagingClient) error {
resp, err = client.ListTopics(ctx, request)
return nil
})
if proxyErr != nil {
return nil, proxyErr
}
return resp, err
}
ret := &mq_pb.ListTopicsResponse{}
knownTopics := make(map[*mq_pb.Topic]struct{})
for brokerStatsItem := range broker.Balancer.Brokers.IterBuffered() {
_, brokerStats := brokerStatsItem.Key, brokerStatsItem.Val
for topicPartitionStatsItem := range brokerStats.Stats.IterBuffered() {
topicPartitionStat := topicPartitionStatsItem.Val
topic := &mq_pb.Topic{
Namespace: topicPartitionStat.TopicPartition.Namespace,
Name: topicPartitionStat.TopicPartition.Topic,
}
if _, found := knownTopics[topic]; found {
continue
}
ret.Topics = append(ret.Topics, topic)
}
}
return ret, nil
}

7
weed/pb/mq.proto
View File

@ -30,6 +30,8 @@ service SeaweedMessaging {
}
rpc DoCreateTopic (DoCreateTopicRequest) returns (DoCreateTopicResponse) {
}
rpc ListTopics (ListTopicsRequest) returns (ListTopicsResponse) {
}
// a pub client will call this to get the topic partitions assignment
rpc RequestTopicPartitions (RequestTopicPartitionsRequest) returns (RequestTopicPartitionsResponse) {
}
@ -145,6 +147,11 @@ message DoCreateTopicRequest {
}
message DoCreateTopicResponse {
}
message ListTopicsRequest {
}
message ListTopicsResponse {
repeated Topic topics = 1;
}
message LookupTopicBrokersRequest {
Topic topic = 1;
bool is_for_publish = 2;

939
weed/pb/mq_pb/mq.pb.go
View File

File diff suppressed because it is too large Load Diff

36
weed/pb/mq_pb/mq_grpc.pb.go
View File

@ -33,6 +33,7 @@ type SeaweedMessagingClient interface {
LookupTopicBrokers(ctx context.Context, in *LookupTopicBrokersRequest, opts ...grpc.CallOption) (*LookupTopicBrokersResponse, error)
CreateTopic(ctx context.Context, in *CreateTopicRequest, opts ...grpc.CallOption) (*CreateTopicResponse, error)
DoCreateTopic(ctx context.Context, in *DoCreateTopicRequest, opts ...grpc.CallOption) (*DoCreateTopicResponse, error)
ListTopics(ctx context.Context, in *ListTopicsRequest, opts ...grpc.CallOption) (*ListTopicsResponse, error)
// a pub client will call this to get the topic partitions assignment
RequestTopicPartitions(ctx context.Context, in *RequestTopicPartitionsRequest, opts ...grpc.CallOption) (*RequestTopicPartitionsResponse, error)
AssignTopicPartitions(ctx context.Context, in *AssignTopicPartitionsRequest, opts ...grpc.CallOption) (*AssignTopicPartitionsResponse, error)
@ -144,6 +145,15 @@ func (c *seaweedMessagingClient) DoCreateTopic(ctx context.Context, in *DoCreate
return out, nil
}
func (c *seaweedMessagingClient) ListTopics(ctx context.Context, in *ListTopicsRequest, opts ...grpc.CallOption) (*ListTopicsResponse, error) {
out := new(ListTopicsResponse)
err := c.cc.Invoke(ctx, "/messaging_pb.SeaweedMessaging/ListTopics", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *seaweedMessagingClient) RequestTopicPartitions(ctx context.Context, in *RequestTopicPartitionsRequest, opts ...grpc.CallOption) (*RequestTopicPartitionsResponse, error) {
out := new(RequestTopicPartitionsResponse)
err := c.cc.Invoke(ctx, "/messaging_pb.SeaweedMessaging/RequestTopicPartitions", in, out, opts...)
@ -249,6 +259,7 @@ type SeaweedMessagingServer interface {
LookupTopicBrokers(context.Context, *LookupTopicBrokersRequest) (*LookupTopicBrokersResponse, error)
CreateTopic(context.Context, *CreateTopicRequest) (*CreateTopicResponse, error)
DoCreateTopic(context.Context, *DoCreateTopicRequest) (*DoCreateTopicResponse, error)
ListTopics(context.Context, *ListTopicsRequest) (*ListTopicsResponse, error)
// a pub client will call this to get the topic partitions assignment
RequestTopicPartitions(context.Context, *RequestTopicPartitionsRequest) (*RequestTopicPartitionsResponse, error)
AssignTopicPartitions(context.Context, *AssignTopicPartitionsRequest) (*AssignTopicPartitionsResponse, error)
@ -287,6 +298,9 @@ func (UnimplementedSeaweedMessagingServer) CreateTopic(context.Context, *CreateT
func (UnimplementedSeaweedMessagingServer) DoCreateTopic(context.Context, *DoCreateTopicRequest) (*DoCreateTopicResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method DoCreateTopic not implemented")
}
func (UnimplementedSeaweedMessagingServer) ListTopics(context.Context, *ListTopicsRequest) (*ListTopicsResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method ListTopics not implemented")
}
func (UnimplementedSeaweedMessagingServer) RequestTopicPartitions(context.Context, *RequestTopicPartitionsRequest) (*RequestTopicPartitionsResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method RequestTopicPartitions not implemented")
}
@ -467,6 +481,24 @@ func _SeaweedMessaging_DoCreateTopic_Handler(srv interface{}, ctx context.Contex
return interceptor(ctx, in, info, handler)
}
func _SeaweedMessaging_ListTopics_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ListTopicsRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(SeaweedMessagingServer).ListTopics(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/messaging_pb.SeaweedMessaging/ListTopics",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(SeaweedMessagingServer).ListTopics(ctx, req.(*ListTopicsRequest))
}
return interceptor(ctx, in, info, handler)
}
func _SeaweedMessaging_RequestTopicPartitions_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(RequestTopicPartitionsRequest)
if err := dec(in); err != nil {
@ -603,6 +635,10 @@ var SeaweedMessaging_ServiceDesc = grpc.ServiceDesc{
MethodName: "DoCreateTopic",
Handler: _SeaweedMessaging_DoCreateTopic_Handler,
},
{
MethodName: "ListTopics",
Handler: _SeaweedMessaging_ListTopics_Handler,
},
{
MethodName: "RequestTopicPartitions",
Handler: _SeaweedMessaging_RequestTopicPartitions_Handler,

29
weed/shell/command_mq_topic_list.go
View File

@ -4,7 +4,9 @@ import (
"context"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/mq/balancer"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
"io"
)
@ -29,21 +31,22 @@ func (c *commandMqTopicList) Do(args []string, commandEnv *CommandEnv, writer io
if err != nil {
return err
}
//pb.WithBrokerGrpcClient(false, brokerBalancer, commandEnv.option.GrpcDialOption, func(client pb.SeaweedMessagingClient) error {
// resp, err := client.ListTopics(context.Background(), &pb.ListTopicsRequest{})
// if err != nil {
// return err
// }
// for _, topic := range resp.Topics {
// fmt.Fprintf(writer, "%s\n", topic)
// }
// return nil
//})
fmt.Fprintf(writer, "current balancer: %s\n", brokerBalancer)
return nil
return pb.WithBrokerGrpcClient(false, brokerBalancer, commandEnv.option.GrpcDialOption, func(client mq_pb.SeaweedMessagingClient) error {
resp, err := client.ListTopics(context.Background(), &mq_pb.ListTopicsRequest{})
if err != nil {
return err
}
if len(resp.Topics) == 0 {
fmt.Fprintf(writer, "no topics found\n")
return nil
}
for _, topic := range resp.Topics {
fmt.Fprintf(writer, " %+v\n", topic)
}
return nil
})
}
func findBrokerBalancer(commandEnv *CommandEnv) (brokerBalancer string, err error) {