feat: controller and analyzer add state counter

Author: askyrie

server/controller/db/metadb/model/model.go

@@ -117,6 +117,22 @@ type Controller struct {
 	Lcuuid             string    `gorm:"column:lcuuid;type:char(64);not null" json:"LCUUID"`
 }
 
+func (c Controller) GetLcuuid() string {
+	return c.Lcuuid
+}
+
+func (c Controller) GetName() string {
+	return c.Name
+}
+
+func (c Controller) GetIP() string {
+	return c.IP
+}
+
+func (c Controller) GetState() int {
+	return c.State
+}
+
 type AZControllerConnection struct {
 	ID           int    `gorm:"primaryKey;column:id;type:int;not null" json:"ID"`
 	AZ           string `gorm:"column:az;type:char(64);default:ALL" json:"AZ"`
@@ -153,6 +169,22 @@ type Analyzer struct {
 	Lcuuid            string    `gorm:"column:lcuuid;type:char(64);not null" json:"LCUUID"`
 }
 
+func (a Analyzer) GetLcuuid() string {
+	return a.Lcuuid
+}
+
+func (a Analyzer) GetName() string {
+	return a.Name
+}
+
+func (a Analyzer) GetIP() string {
+	return a.IP
+}
+
+func (a Analyzer) GetState() int {
+	return a.State
+}
+
 type AZAnalyzerConnection struct {
 	ID         int    `gorm:"primaryKey;column:id;type:int;not null" json:"ID"`
 	AZ         string `gorm:"column:az;type:char(64);default:ALL" json:"AZ"`

server/controller/monitor/analyzer.go

@@ -30,9 +30,12 @@ import (
 	"github.com/deepflowio/deepflow/server/controller/http/service/rebalance"
 	mconfig "github.com/deepflowio/deepflow/server/controller/monitor/config"
 	"github.com/deepflowio/deepflow/server/controller/trisolaris/refresh"
+	"github.com/deepflowio/deepflow/server/libs/stats"
 )
 
 type AnalyzerCheck struct {
+	stateCounter *SysStateCounter[metadbmodel.Analyzer]
+
 	cCtx                  context.Context
 	cCancel               context.CancelFunc
 	cfg                   mconfig.MonitorConfig
@@ -54,11 +57,18 @@ func NewAnalyzerCheck(cfg *config.ControllerConfig, ctx context.Context) *Analyz
 		ch:                    make(chan dbAndIP, cfg.MonitorCfg.HealthCheckHandleChannelLen),
 		normalAnalyzerDict:    make(map[string]*dfHostCheck),
 		exceptionAnalyzerDict: make(map[string]*dfHostCheck),
+		stateCounter:          NewSysStateCounter[metadbmodel.Analyzer](),
 	}
 }
 
 func (c *AnalyzerCheck) Start(sCtx context.Context) {
 	log.Info("analyzer check start")
+
+	err := stats.RegisterCountableWithModulePrefix("server_", "analyzer", c.stateCounter, stats.OptionStatTags{"type": "analyzer"})
+	if err != nil {
+		log.Error(err)
+	}
+
 	go func() {
 		ticker := time.NewTicker(time.Duration(c.cfg.SyncDefaultORGDataInterval) * time.Second)
 		defer ticker.Stop()
@@ -123,6 +133,7 @@ func (c *AnalyzerCheck) Start(sCtx context.Context) {
 
 func (c *AnalyzerCheck) Stop() {
 	if c.cCancel != nil {
+		c.stateCounter.Closed()
 		c.cCancel()
 	}
 	log.Info("analyzer check stopped")
@@ -231,6 +242,12 @@ func (c *AnalyzerCheck) healthCheck(orgDB *metadb.DB) {
 			}
 		}
 	}
+
+	analyzers = []metadbmodel.Analyzer{}
+	if err := metadb.DefaultDB.Find(&analyzers).Error; err == nil {
+		c.stateCounter.AddSysStateCounters(analyzers)
+	}
+
 	log.Info("analyzer health check end")
 }
 

server/controller/monitor/common.go

@@ -19,6 +19,7 @@ package monitor
 import (
 	"fmt"
 	"net/http"
+	"sync"
 	"time"
 
 	"github.com/deepflowio/deepflow/server/controller/common"
@@ -92,3 +93,56 @@ func getIPMap(hostType string) (map[string]bool, error) {
 	}
 	return res, nil
 }
+
+type SysComponent interface {
+	GetLcuuid() string
+	GetName() string
+	GetIP() string
+	GetState() int
+}
+
+type SysState struct {
+	Name  string `statsd:"name"`
+	IP    string `statsd:"ip"`
+	State uint64 `statsd:"state"`
+}
+
+type SysStateCounter[T SysComponent] struct {
+	mu       sync.Mutex
+	Counters map[string]*SysState
+}
+
+func NewSysStateCounter[T SysComponent]() *SysStateCounter[T] {
+	return &SysStateCounter[T]{
+		Counters: make(map[string]*SysState),
+	}
+}
+
+func (s *SysStateCounter[T]) GetCounter() interface{} {
+	counters := make(map[string]*SysState, len(s.Counters))
+	s.mu.Lock()
+	counters, s.Counters = s.Counters, counters
+	s.mu.Unlock()
+
+	result := make([]*SysState, 0, len(counters))
+	for _, counter := range counters {
+		result = append(result, counter)
+	}
+	return result
+}
+
+func (s *SysStateCounter[T]) AddSysStateCounters(sysComponents []T) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, sysComponent := range sysComponents {
+		s.Counters[sysComponent.GetLcuuid()] = &SysState{
+			Name:  sysComponent.GetName(),
+			IP:    sysComponent.GetIP(),
+			State: uint64(sysComponent.GetState()),
+		}
+	}
+}
+
+func (s *SysStateCounter[T]) Closed() bool {
+	return false
+}

server/controller/monitor/controller.go

@@ -30,6 +30,7 @@ import (
 	"github.com/deepflowio/deepflow/server/controller/model"
 	mconfig "github.com/deepflowio/deepflow/server/controller/monitor/config"
 	"github.com/deepflowio/deepflow/server/controller/trisolaris/refresh"
+	"github.com/deepflowio/deepflow/server/libs/stats"
 )
 
 type dbAndIP struct {
@@ -38,6 +39,8 @@ type dbAndIP struct {
 }
 
 type ControllerCheck struct {
+	stateCounter *SysStateCounter[metadbmodel.Controller]
+
 	cCtx                    context.Context
 	cCancel                 context.CancelFunc
 	cfg                     mconfig.MonitorConfig
@@ -59,11 +62,18 @@ func NewControllerCheck(cfg *config.ControllerConfig, ctx context.Context) *Cont
 		ch:                      make(chan dbAndIP, cfg.MonitorCfg.HealthCheckHandleChannelLen),
 		normalControllerDict:    make(map[string]*dfHostCheck),
 		exceptionControllerDict: make(map[string]*dfHostCheck),
+		stateCounter:            NewSysStateCounter[metadbmodel.Controller](),
 	}
 }
 
 func (c *ControllerCheck) Start(sCtx context.Context) {
 	log.Info("controller check start")
+
+	err := stats.RegisterCountableWithModulePrefix("server_", "controller", c.stateCounter, stats.OptionStatTags{"type": "controller"})
+	if err != nil {
+		log.Error(err)
+	}
+
 	go func() {
 		ticker := time.NewTicker(time.Duration(c.cfg.SyncDefaultORGDataInterval) * time.Second)
 		defer ticker.Stop()
@@ -119,6 +129,7 @@ func (c *ControllerCheck) Start(sCtx context.Context) {
 
 func (c *ControllerCheck) Stop() {
 	if c.cCancel != nil {
+		c.stateCounter.Closed()
 		c.cCancel()
 	}
 	log.Info("controller check stopped")
@@ -240,6 +251,12 @@ func (c *ControllerCheck) healthCheck(orgDB *metadb.DB) {
 		}
 	}
 	c.cleanExceptionControllerData(orgDB, controllerIPs)
+
+	controllers = []metadbmodel.Controller{}
+	if err := metadb.DefaultDB.Find(&controllers).Error; err == nil {
+		c.stateCounter.AddSysStateCounters(controllers)
+	}
+
 	log.Info("controller health check end")
 }