You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
103 lines
3.1 KiB
103 lines
3.1 KiB
package topic
|
|
|
|
import (
|
|
cmap "github.com/orcaman/concurrent-map/v2"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
|
|
"github.com/shirou/gopsutil/v3/cpu"
|
|
"time"
|
|
)
|
|
|
|
// LocalTopicManager manages topics on local broker
|
|
type LocalTopicManager struct {
|
|
topics cmap.ConcurrentMap[string, *LocalTopic]
|
|
}
|
|
|
|
// NewLocalTopicManager creates a new LocalTopicManager
|
|
func NewLocalTopicManager() *LocalTopicManager {
|
|
return &LocalTopicManager{
|
|
topics: cmap.New[*LocalTopic](),
|
|
}
|
|
}
|
|
|
|
// AddTopicPartition adds a topic to the local topic manager
|
|
func (manager *LocalTopicManager) AddTopicPartition(topic Topic, localPartition *LocalPartition) {
|
|
localTopic, ok := manager.topics.Get(topic.String())
|
|
if !ok {
|
|
localTopic = &LocalTopic{
|
|
Topic: topic,
|
|
Partitions: make([]*LocalPartition, 0),
|
|
}
|
|
}
|
|
if !manager.topics.SetIfAbsent(topic.String(), localTopic) {
|
|
localTopic, _ = manager.topics.Get(topic.String())
|
|
}
|
|
if localTopic.findPartition(localPartition.Partition) != nil {
|
|
return
|
|
}
|
|
localTopic.Partitions = append(localTopic.Partitions, localPartition)
|
|
}
|
|
|
|
// GetTopic gets a topic from the local topic manager
|
|
func (manager *LocalTopicManager) GetTopicPartition(topic Topic, partition Partition) *LocalPartition {
|
|
localTopic, ok := manager.topics.Get(topic.String())
|
|
if !ok {
|
|
return nil
|
|
}
|
|
return localTopic.findPartition(partition)
|
|
}
|
|
|
|
// RemoveTopic removes a topic from the local topic manager
|
|
func (manager *LocalTopicManager) RemoveTopic(topic Topic) {
|
|
manager.topics.Remove(topic.String())
|
|
}
|
|
|
|
func (manager *LocalTopicManager) RemoveTopicPartition(topic Topic, partition Partition) (removed bool) {
|
|
localTopic, ok := manager.topics.Get(topic.String())
|
|
if !ok {
|
|
return false
|
|
}
|
|
return localTopic.removePartition(partition)
|
|
}
|
|
|
|
func (manager *LocalTopicManager) CollectStats(duration time.Duration) *mq_pb.BrokerStats {
|
|
stats := &mq_pb.BrokerStats{
|
|
Stats: make(map[string]*mq_pb.TopicPartitionStats),
|
|
}
|
|
|
|
// collect current broker's cpu usage
|
|
// this needs to be in front, so the following stats can be more accurate
|
|
usages, err := cpu.Percent(duration, false)
|
|
if err == nil && len(usages) > 0 {
|
|
stats.CpuUsagePercent = int32(usages[0])
|
|
}
|
|
|
|
// collect current broker's topics and partitions
|
|
manager.topics.IterCb(func(topic string, localTopic *LocalTopic) {
|
|
for _, localPartition := range localTopic.Partitions {
|
|
topicPartition := &TopicPartition{
|
|
Topic: Topic{Namespace: localTopic.Namespace, Name: localTopic.Name},
|
|
Partition: Partition{
|
|
RingSize: localPartition.RingSize,
|
|
RangeStart: localPartition.RangeStart,
|
|
RangeStop: localPartition.RangeStop,
|
|
},
|
|
}
|
|
stats.Stats[topicPartition.String()] = &mq_pb.TopicPartitionStats{
|
|
Topic: &mq_pb.Topic{
|
|
Namespace: string(localTopic.Namespace),
|
|
Name: localTopic.Name,
|
|
},
|
|
Partition: &mq_pb.Partition{
|
|
RingSize: localPartition.RingSize,
|
|
RangeStart: localPartition.RangeStart,
|
|
RangeStop: localPartition.RangeStop,
|
|
},
|
|
ConsumerCount: localPartition.ConsumerCount,
|
|
}
|
|
// fmt.Printf("collect topic %+v partition %+v\n", topicPartition, localPartition.Partition)
|
|
}
|
|
})
|
|
|
|
return stats
|
|
|
|
}
|