ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2025-11-04 02:35:07 +01:00
Code Issues Packages Projects Releases Wiki Activity

Split serial and parallel collectors. Read in parallel first

Browse Source
This commit is contained in:
Thomas Roehl
2022-05-13 12:23:34 +02:00
parent 7b804ad7a2
commit 1d1315c08b
1 changed files with 61 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
collectors/collectorManager.go
View File

@@ -14,39 +14,42 @@ import (
// Map of all available metric collectors // Map of all available metric collectors
var AvailableCollectors = map[string]MetricCollector{ var AvailableCollectors = map[string]MetricCollector{
"likwid": new(LikwidCollector), "likwid": new(LikwidCollector),
"loadavg": new(LoadavgCollector), "loadavg": new(LoadavgCollector),
"memstat": new(MemstatCollector), "memstat": new(MemstatCollector),
"netstat": new(NetstatCollector), "netstat": new(NetstatCollector),
"ibstat": new(InfinibandCollector), "ibstat": new(InfinibandCollector),
"lustrestat": new(LustreCollector), "lustrestat": new(LustreCollector),
"cpustat": new(CpustatCollector), "cpustat": new(CpustatCollector),
"topprocs": new(TopProcsCollector), "topprocs": new(TopProcsCollector),
"nvidia": new(NvidiaCollector), "nvidia": new(NvidiaCollector),
"customcmd": new(CustomCmdCollector), "customcmd": new(CustomCmdCollector),
"iostat": new(IOstatCollector), "iostat": new(IOstatCollector),
"diskstat": new(DiskstatCollector), "diskstat": new(DiskstatCollector),
"tempstat": new(TempCollector), "tempstat": new(TempCollector),
"ipmistat": new(IpmiCollector), "ipmistat": new(IpmiCollector),
"gpfs": new(GpfsCollector), "gpfs": new(GpfsCollector),
"cpufreq": new(CPUFreqCollector), "cpufreq": new(CPUFreqCollector),
"cpufreq_cpuinfo": new(CPUFreqCpuInfoCollector), "cpufreq_cpuinfo": new(CPUFreqCpuInfoCollector),
"nfs3stat": new(Nfs3Collector), "nfs3stat": new(Nfs3Collector),
"nfs4stat": new(Nfs4Collector), "nfs4stat": new(Nfs4Collector),
"numastats": new(NUMAStatsCollector), "numastats": new(NUMAStatsCollector),
"beegfs_meta": new(BeegfsMetaCollector), "beegfs_meta": new(BeegfsMetaCollector),
"beegfs_storage": new(BeegfsStorageCollector), "beegfs_storage": new(BeegfsStorageCollector),
} }
// Metric collector manager data structure // Metric collector manager data structure
type collectorManager struct { type collectorManager struct {
collectors []MetricCollector // List of metric collectors to use collectors []MetricCollector // List of metric collectors to read in parallel
output chan lp.CCMetric // Output channels serial []MetricCollector // List of metric collectors to read serially
done chan bool // channel to finish / stop metric collector manager output chan lp.CCMetric // Output channels
ticker mct.MultiChanTicker // periodically ticking once each interval done chan bool // channel to finish / stop metric collector manager
duration time.Duration // duration (for metrics that measure over a given duration) ticker mct.MultiChanTicker // periodically ticking once each interval
wg *sync.WaitGroup // wait group for all goroutines in cc-metric-collector duration time.Duration // duration (for metrics that measure over a given duration)
config map[string]json.RawMessage // json encoded config for collector manager 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
} }
// Metric collector manager access functions // Metric collector manager access functions
@@ -66,6 +69,7 @@ type CollectorManager interface {
// Initialization is done for all configured collectors // Initialization is done for all configured collectors
func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfigFile string) error { func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfigFile string) error {
cm.collectors = make([]MetricCollector, 0) cm.collectors = make([]MetricCollector, 0)
cm.serial = make([]MetricCollector, 0)
cm.output = nil cm.output = nil
cm.done = make(chan bool) cm.done = make(chan bool)
cm.wg = wg cm.wg = wg
@@ -100,7 +104,11 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
continue continue
} }
cclog.ComponentDebug("CollectorManager", "ADD COLLECTOR", collector.Name()) cclog.ComponentDebug("CollectorManager", "ADD COLLECTOR", collector.Name())
cm.collectors = append(cm.collectors, collector) if collector.Parallel() {
cm.collectors = append(cm.collectors, collector)
} else {
cm.serial = append(cm.serial, collector)
}
} }
return nil return nil
} }
@@ -116,6 +124,10 @@ func (cm *collectorManager) Start() {
// Collector manager is done // Collector manager is done
done := func() { done := func() {
// close all metric collectors // close all metric collectors
if cm.parallel_run {
cm.collector_wg.Wait()
cm.parallel_run = false
}
for _, c := range cm.collectors { for _, c := range cm.collectors {
c.Close() c.Close()
} }
@@ -130,7 +142,26 @@ func (cm *collectorManager) Start() {
done() done()
return return
case t := <-tick: case t := <-tick:
cm.parallel_run = true
for _, c := range cm.collectors { for _, c := range cm.collectors {
// Wait for done signal or execute the collector
select {
case <-cm.done:
done()
return
default:
// Read metrics from collector c via goroutine
cclog.ComponentDebug("CollectorManager", c.Name(), t)
cm.collector_wg.Add(1)
go func(myc MetricCollector) {
myc.Read(cm.duration, cm.output)
cm.collector_wg.Done()
}(c)
}
}
cm.collector_wg.Wait()
cm.parallel_run = false
for _, c := range cm.serial {
// Wait for done signal or execute the collector // Wait for done signal or execute the collector
select { select {
case <-cm.done: case <-cm.done: