ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2026-07-18 15:40:38 +02:00
Code Issues Packages Projects Releases Wiki Activity

collectorManager: Make markers opt-in

Browse Source
This commit is contained in:
Michael Panzlaff
2026-07-15 18:03:19 +02:00
parent cdac3c9163
commit 6124971b2e
2 changed files with 32 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cc-metric-collector.go
+10 -7
View File
@@ -29,15 +29,17 @@ import (
)
type CentralConfigFile struct {
Interval string `json:"interval"`
Duration string `json:"duration"`
Interval string `json:"interval"`
Duration string `json:"duration"`
EnableMarkers bool `json:"enable-markers"`
}
type RuntimeConfig struct {
Interval time.Duration
Duration time.Duration
CliArgs map[string]string
ConfigFile CentralConfigFile
Interval time.Duration
Duration time.Duration
EnableMarkers bool
CliArgs map[string]string
ConfigFile CentralConfigFile
MetricRouter mr.MetricRouter
CollectManager collectors.CollectorManager
@@ -157,6 +159,7 @@ func mainFunc() int {
cclog.Error("The interval should be greater than duration")
return 1
}
rcfg.EnableMarkers = rcfg.ConfigFile.EnableMarkers
routerConf := ccconf.GetPackageConfig("router")
if len(routerConf) == 0 {
@@ -199,7 +202,7 @@ func mainFunc() int {
rcfg.MetricRouter.AddOutput(RouterToSinksChannel)
// Create new collector manager
rcfg.CollectManager, err = collectors.New(rcfg.MultiChanTicker, rcfg.Duration, &rcfg.Sync, collectorConf)
rcfg.CollectManager, err = collectors.New(rcfg.MultiChanTicker, rcfg.Duration, rcfg.EnableMarkers, &rcfg.Sync, collectorConf)
if err != nil {
cclog.Error(err.Error())
return 1
collectors/collectorManager.go
+22 -16
View File
@@ -56,21 +56,22 @@ var AvailableCollectors = map[string]MetricCollector{
// Metric collector manager data structure
type collectorManager struct {
collectors []MetricCollector // List of metric collectors to read in parallel
serial []MetricCollector // List of metric collectors to read serially
output chan lp.CCMessage // Output channels
done chan bool // channel to finish / stop metric collector manager
ticker mct.MultiChanTicker // periodically ticking once each interval
duration time.Duration // duration (for metrics that measure over a given duration)
wg *sync.WaitGroup // wait group for all goroutines in cc-metric-collector
config map[string]json.RawMessage // json encoded config for collector manager
collector_wg sync.WaitGroup // internally used wait group for the parallel reading of collector
parallel_run bool // Flag whether the collectors are currently read in parallel
collectors []MetricCollector // List of metric collectors to read in parallel
serial []MetricCollector // List of metric collectors to read serially
output chan lp.CCMessage // Output channels
done chan bool // channel to finish / stop metric collector manager
ticker mct.MultiChanTicker // periodically ticking once each interval
duration time.Duration // duration (for metrics that measure over a given duration)
wg *sync.WaitGroup // wait group for all goroutines in cc-metric-collector
config map[string]json.RawMessage // json encoded config for collector manager
collector_wg sync.WaitGroup // internally used wait group for the parallel reading of collector
parallel_run bool // Flag whether the collectors are currently read in parallel
enableMarkers bool // Send ccmc-{begin,end} metrics
}
// Metric collector manager access functions
type CollectorManager interface {
Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) error
Init(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) error
AddOutput(output chan lp.CCMessage)
Start()
Close()
@@ -83,7 +84,7 @@ type CollectorManager interface {
// * ticker (from variable ticker)
// * configuration (read from config file in variable collectConfigFile)
// Initialization is done for all configured collectors
func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) error {
func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) error {
cm.collectors = make([]MetricCollector, 0)
cm.serial = make([]MetricCollector, 0)
cm.output = nil
@@ -91,6 +92,7 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
cm.wg = wg
cm.ticker = ticker
cm.duration = duration
cm.enableMarkers = enableMarkers
d := json.NewDecoder(bytes.NewReader(collectConfig))
d.DisallowUnknownFields()
@@ -148,7 +150,9 @@ func (cm *collectorManager) Start() {
done()
return
case t := <-tick:
lp.NewMetric("ccmc-begin", map[string]string{"type": "node"}, nil, map[string]interface{}{"value": 0}, time.Now())
if cm.enableMarkers {
lp.NewMetric("ccmc-begin", map[string]string{"type": "node"}, nil, map[string]interface{}{"value": 0}, time.Now())
}
cm.parallel_run = true
for _, c := range cm.collectors {
// Wait for done signal or execute the collector
@@ -180,7 +184,9 @@ func (cm *collectorManager) Start() {
c.Read(cm.duration, cm.output)
}
}
lp.NewMetric("ccmc-end", map[string]string{"type": "node"}, nil, map[string]interface{}{"value": 0}, time.Now())
if cm.enableMarkers {
lp.NewMetric("ccmc-end", map[string]string{"type": "node"}, nil, map[string]interface{}{"value": 0}, time.Now())
}
}
}
})
@@ -203,9 +209,9 @@ func (cm *collectorManager) Close() {
}
// New creates a new initialized metric collector manager
func New(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) (CollectorManager, error) {
func New(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) (CollectorManager, error) {
cm := new(collectorManager)
err := cm.Init(ticker, duration, wg, collectConfig)
err := cm.Init(ticker, duration, enableMarkers, wg, collectConfig)
if err != nil {
return nil, err
}