implement counter by key

pkg/console/counter/component.go

@@ -22,7 +22,11 @@ import (
 	"math"
 
 	"github.com/apache/dubbo-admin/pkg/core/events"
+	"github.com/apache/dubbo-admin/pkg/core/logger"
+	meshresource "github.com/apache/dubbo-admin/pkg/core/resource/apis/mesh/v1alpha1"
+	resmodel "github.com/apache/dubbo-admin/pkg/core/resource/model"
 	"github.com/apache/dubbo-admin/pkg/core/runtime"
+	"github.com/apache/dubbo-admin/pkg/core/store"
 )
 
 const ComponentType runtime.ComponentType = "counter manager"
@@ -41,7 +45,7 @@ var _ ManagerComponent = &managerComponent{}
 func (c *managerComponent) RequiredDependencies() []runtime.ComponentType {
 	return []runtime.ComponentType{
 		runtime.ResourceStore,
-		runtime.EventBus, // Counter depends on EventBus to subscribe to events
+		runtime.EventBus,
 	}
 }
 
@@ -64,6 +68,19 @@ func (c *managerComponent) Init(runtime.BuilderContext) error {
 }
 
 func (c *managerComponent) Start(rt runtime.Runtime, _ <-chan struct{}) error {
+	storeComponent, err := rt.GetComponent(runtime.ResourceStore)
+	if err != nil {
+		return err
+	}
+	storeRouter, ok := storeComponent.(store.Router)
+	if !ok {
+		return fmt.Errorf("component %s does not implement store.Router", runtime.ResourceStore)
+	}
+
+	if err := c.initializeCountsFromStore(storeRouter); err != nil {
+		logger.Warnf("Failed to initialize counter manager from store: %v", err)
+	}
+
 	component, err := rt.GetComponent(runtime.EventBus)
 	if err != nil {
 		return err
@@ -75,6 +92,73 @@ func (c *managerComponent) Start(rt runtime.Runtime, _ <-chan struct{}) error {
 	return c.manager.Bind(bus)
 }
 
+func (c *managerComponent) initializeCountsFromStore(storeRouter store.Router) error {
+	if err := c.initializeResourceCount(storeRouter, meshresource.InstanceKind); err != nil {
+		return fmt.Errorf("failed to initialize instance count: %w", err)
+	}
+
+	if err := c.initializeResourceCount(storeRouter, meshresource.ApplicationKind); err != nil {
+		return fmt.Errorf("failed to initialize application count: %w", err)
+	}
+
+	if err := c.initializeResourceCount(storeRouter, meshresource.ServiceProviderMetadataKind); err != nil {
+		return fmt.Errorf("failed to initialize service provider metadata count: %w", err)
+	}
+
+	return nil
+}
+
+func (c *managerComponent) initializeResourceCount(storeRouter store.Router, kind resmodel.ResourceKind) error {
+	resourceStore, err := storeRouter.ResourceKindRoute(kind)
+	if err != nil {
+		return err
+	}
+
+	allResources := resourceStore.List()
+	cm := c.manager.(*counterManager)
+
+	for _, obj := range allResources {
+		resource, ok := obj.(resmodel.Resource)
+		if !ok {
+			continue
+		}
+
+		mesh := resource.ResourceMesh()
+		if mesh == "" {
+			mesh = "default"
+		}
+
+		if counter, exists := cm.simpleCounters[kind]; exists {
+			counter.Increment(mesh)
+		}
+
+		if kind == meshresource.InstanceKind {
+			instance, ok := resource.(*meshresource.InstanceResource)
+			if ok && instance.Spec != nil {
+				protocol := instance.Spec.GetProtocol()
+				if protocol != "" {
+					if cfg := cm.getDistributionConfig(kind, ProtocolCounter); cfg != nil {
+						cfg.counter.Increment(mesh, protocol)
+					}
+				}
+
+				releaseVersion := instance.Spec.GetReleaseVersion()
+				if releaseVersion != "" {
+					if cfg := cm.getDistributionConfig(kind, ReleaseCounter); cfg != nil {
+						cfg.counter.Increment(mesh, releaseVersion)
+					}
+				}
+
+				if cfg := cm.getDistributionConfig(kind, DiscoveryCounter); cfg != nil {
+					cfg.counter.Increment(mesh, mesh)
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
 func (c *managerComponent) CounterManager() CounterManager {
 	return c.manager
 }

pkg/console/counter/counter.go

@@ -19,79 +19,149 @@ package counter
 
 import (
 	"sync"
-	"sync/atomic"
 )
 
 type Counter struct {
-	name  string
-	value atomic.Int64
+	name string
+	data map[string]int64
+	mu   sync.RWMutex
 }
 
 func NewCounter(name string) *Counter {
-	return &Counter{name: name}
+	return &Counter{
+		name: name,
+		data: make(map[string]int64),
+	}
 }
 
 func (c *Counter) Get() int64 {
-	return c.value.Load()
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	var sum int64
+	for _, v := range c.data {
+		sum += v
+	}
+	return sum
 }
 
-func (c *Counter) Increment() {
-	c.value.Add(1)
+func (c *Counter) GetByGroup(group string) int64 {
+	if group == "" {
+		group = "default"
+	}
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.data[group]
 }
 
-func (c *Counter) Decrement() {
-	for {
-		current := c.value.Load()
-		if current == 0 {
-			return
-		}
-		if c.value.CompareAndSwap(current, current-1) {
-			return
+func (c *Counter) Increment(group string) {
+	if group == "" {
+		group = "default"
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.data[group]++
+}
+
+func (c *Counter) Decrement(group string) {
+	if group == "" {
+		group = "default"
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if value, ok := c.data[group]; ok {
+		value--
+		if value <= 0 {
+			delete(c.data, group)
+		} else {
+			c.data[group] = value
 		}
 	}
 }
 
 func (c *Counter) Reset() {
-	c.value.Store(0)
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.data = make(map[string]int64)
 }
 
 type DistributionCounter struct {
 	name string
-	data map[string]int64
+	data map[string]map[string]int64
 	mu   sync.RWMutex
 }
 
 func NewDistributionCounter(name string) *DistributionCounter {
 	return &DistributionCounter{
 		name: name,
-		data: make(map[string]int64),
+		data: make(map[string]map[string]int64),
 	}
 }
 
-func (c *DistributionCounter) Increment(key string) {
+func (c *DistributionCounter) Increment(group, key string) {
+	if group == "" {
+		group = "default"
+	}
+	if key == "" {
+		key = "unknown"
+	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	c.data[key]++
+	if c.data[group] == nil {
+		c.data[group] = make(map[string]int64)
+	}
+	c.data[group][key]++
 }
 
-func (c *DistributionCounter) Decrement(key string) {
+func (c *DistributionCounter) Decrement(group, key string) {
+	if group == "" {
+		group = "default"
+	}
+	if key == "" {
+		key = "unknown"
+	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	if value, ok := c.data[key]; ok {
+	groupData, exists := c.data[group]
+	if !exists {
+		return
+	}
+	if value, ok := groupData[key]; ok {
 		value--
 		if value <= 0 {
-			delete(c.data, key)
+			delete(groupData, key)
+			if len(groupData) == 0 {
+				delete(c.data, group)
+			}
 		} else {
-			c.data[key] = value
+			groupData[key] = value
 		}
 	}
 }
 
 func (c *DistributionCounter) GetAll() map[string]int64 {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
-	result := make(map[string]int64, len(c.data))
-	for k, v := range c.data {
+	result := make(map[string]int64)
+	for _, groupData := range c.data {
+		for k, v := range groupData {
+			result[k] += v
+		}
+	}
+	return result
+}
+
+func (c *DistributionCounter) GetByGroup(group string) map[string]int64 {
+	if group == "" {
+		group = "default"
+	}
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	groupData, exists := c.data[group]
+	if !exists {
+		return map[string]int64{}
+	}
+	result := make(map[string]int64, len(groupData))
+	for k, v := range groupData {
 		result[k] = v
 	}
 	return result
@@ -100,5 +170,5 @@ func (c *DistributionCounter) GetAll() map[string]int64 {
 func (c *DistributionCounter) Reset() {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	c.data = make(map[string]int64)
+	c.data = make(map[string]map[string]int64)
 }