mirror of
https://github.com/ClusterCockpit/cc-metric-collector.git
synced 2026-09-02 20:07:14 +02:00
Compare commits
15
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6df8da8fe9 | ||
|
|
39cae9c79a | ||
|
|
789ba7f4a5 | ||
|
|
9bd9a07810 | ||
|
|
a0aedff202 | ||
|
|
6944454592 | ||
|
|
9e773e2e85 | ||
|
|
c13bb55735 | ||
|
|
6124971b2e | ||
|
|
cdac3c9163 | ||
|
|
70364d084b | ||
|
|
ec0868eba9 | ||
|
|
b8e76e3bf0 | ||
|
|
f028c83a75 | ||
|
|
4a859af358 |
@@ -33,6 +33,7 @@ There is a main configuration file with basic settings that point to the other c
|
||||
"receivers-file" : "receivers.json",
|
||||
"router-file" : "router.json",
|
||||
"main": {
|
||||
"enable-markers": false,
|
||||
"interval": "10s",
|
||||
"duration": "1s"
|
||||
}
|
||||
|
||||
+10
-7
@@ -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
|
||||
|
||||
@@ -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,6 +150,14 @@ func (cm *collectorManager) Start() {
|
||||
done()
|
||||
return
|
||||
case t := <-tick:
|
||||
if cm.enableMarkers {
|
||||
m, err := lp.NewMetric("ccmc-begin", map[string]string{"type": "node"}, nil, 0, time.Now())
|
||||
if err != nil {
|
||||
cclog.ComponentErrorf("CollectorManager", "Unable to create marker metric: %v", err)
|
||||
} else {
|
||||
cm.output <- m
|
||||
}
|
||||
}
|
||||
cm.parallel_run = true
|
||||
for _, c := range cm.collectors {
|
||||
// Wait for done signal or execute the collector
|
||||
@@ -179,6 +189,14 @@ func (cm *collectorManager) Start() {
|
||||
c.Read(cm.duration, cm.output)
|
||||
}
|
||||
}
|
||||
if cm.enableMarkers {
|
||||
m, err := lp.NewMetric("ccmc-end", map[string]string{"type": "node"}, nil, 0, time.Now())
|
||||
if err != nil {
|
||||
cclog.ComponentErrorf("CollectorManager", "Unable to create marker metric: %v", err)
|
||||
} else {
|
||||
cm.output <- m
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
@@ -201,9 +219,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
|
||||
}
|
||||
|
||||
@@ -46,6 +46,7 @@ const (
|
||||
LIKWID_LIB_DL_FLAGS = dl.RTLD_LAZY | dl.RTLD_GLOBAL
|
||||
LIKWID_DEF_ACCESSMODE = "direct"
|
||||
LIKWID_DEF_LOCKFILE = "/var/run/likwid.lock"
|
||||
LIKWID_DEF_WARMUP_DEL = "0s"
|
||||
)
|
||||
|
||||
type LikwidCollectorMetricConfig struct {
|
||||
@@ -83,6 +84,7 @@ type LikwidCollectorConfig struct {
|
||||
DaemonPath string `json:"accessdaemon_path,omitempty"`
|
||||
LibraryPath string `json:"liblikwid_path,omitempty"`
|
||||
LockfilePath string `json:"lockfile_path,omitempty"`
|
||||
WarmupDelay string `json:"warmup_delay"`
|
||||
}
|
||||
|
||||
type LikwidCollector struct {
|
||||
@@ -104,6 +106,7 @@ type LikwidCollector struct {
|
||||
likwidGroups map[C.int]LikwidEventsetConfig
|
||||
lock sync.Mutex
|
||||
measureThread thread.Thread
|
||||
warmupDelay time.Duration
|
||||
}
|
||||
|
||||
type LikwidMetric struct {
|
||||
@@ -206,6 +209,7 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
|
||||
m.config.AccessMode = LIKWID_DEF_ACCESSMODE
|
||||
m.config.LibraryPath = LIKWID_LIB_NAME
|
||||
m.config.LockfilePath = LIKWID_DEF_LOCKFILE
|
||||
m.config.WarmupDelay = LIKWID_DEF_WARMUP_DEL
|
||||
if len(config) > 0 {
|
||||
d := json.NewDecoder(bytes.NewReader(config))
|
||||
d.DisallowUnknownFields()
|
||||
@@ -242,6 +246,17 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
|
||||
m.cpu2tid[c] = i
|
||||
}
|
||||
|
||||
if len(m.config.WarmupDelay) > 0 {
|
||||
t, err := time.ParseDuration(m.config.WarmupDelay)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s Init(): Cannot parse WarmupDelay: %w", m.name, err)
|
||||
}
|
||||
if t < 0 {
|
||||
return fmt.Errorf("%s Init(): WarmupDelay must not be negative", m.name)
|
||||
}
|
||||
m.warmupDelay = t
|
||||
}
|
||||
|
||||
m.likwidGroups = make(map[C.int]LikwidEventsetConfig)
|
||||
|
||||
// This is for the global metrics computation test
|
||||
@@ -495,6 +510,27 @@ func (m *LikwidCollector) takeMeasurement(evidx int, evset LikwidEventsetConfig,
|
||||
if ret != 0 {
|
||||
return true, fmt.Errorf("failed to start events '%s', error %d", evset.go_estr, ret)
|
||||
}
|
||||
|
||||
// warmup measuring
|
||||
if m.warmupDelay > 0 {
|
||||
select {
|
||||
case <-sigchan:
|
||||
ret = -1
|
||||
case e := <-watcher.Events:
|
||||
if e.Op != fsnotify.Chmod {
|
||||
ret = C.perfmon_readCounters()
|
||||
}
|
||||
default:
|
||||
ret = C.perfmon_readCounters()
|
||||
}
|
||||
if ret != 0 {
|
||||
return true, fmt.Errorf("failed to read events '%s', error %d", evset.go_estr, ret)
|
||||
}
|
||||
|
||||
time.Sleep(m.warmupDelay)
|
||||
}
|
||||
|
||||
// begin measuring
|
||||
select {
|
||||
case <-sigchan:
|
||||
ret = -1
|
||||
@@ -512,7 +548,7 @@ func (m *LikwidCollector) takeMeasurement(evidx int, evset LikwidEventsetConfig,
|
||||
// Wait
|
||||
time.Sleep(interval)
|
||||
|
||||
// Read counters
|
||||
// end measuring
|
||||
select {
|
||||
case <-sigchan:
|
||||
ret = -1
|
||||
|
||||
@@ -63,6 +63,7 @@ Additional options:
|
||||
- `accessdaemon_path`: Folder of the accessDaemon `likwid-accessD` (like `/usr/local/sbin`)
|
||||
- `liblikwid_path`: Location of `liblikwid.so` including file name like `/usr/local/lib/liblikwid.so`
|
||||
- `lockfile_path`: Location of LIKWID's lock file if multiple tools should access the hardware counters. Default `/var/run/likwid.lock`
|
||||
- `warmup_delay`: Run an additional measurement of the specified length before the actual measurement. This can be used as a workaround for CPU starvation (during high load) in the measurement thread. Default is disabled (i.e. `0s`).
|
||||
|
||||
### Available metric types
|
||||
|
||||
|
||||
@@ -65,7 +65,7 @@ func (m *MegwareEurekaCollector) Init(config json.RawMessage) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
m.name = "MegwareEureka"
|
||||
m.name = "MegwareEurekaCollector"
|
||||
if err := m.setup(); err != nil {
|
||||
return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
|
||||
}
|
||||
@@ -127,9 +127,6 @@ func (m *MegwareEurekaCollector) readMpsData() (*mpsData, error) {
|
||||
return nil, fmt.Errorf("unable to decode u20 JSON output: %w (stdout=%s)", err, stdout.String())
|
||||
}
|
||||
|
||||
fmt.Printf("string: %+v\n", stdout.String())
|
||||
fmt.Printf("obj: %+v\n", u20output)
|
||||
|
||||
u20output.GetMpsPollValues.Timestamp = time.Now()
|
||||
|
||||
return &u20output.GetMpsPollValues, nil
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
"receivers-file" : "./receivers.json",
|
||||
"router-file" : "./router.json",
|
||||
"main" : {
|
||||
"enable-markers": false,
|
||||
"interval": "10s",
|
||||
"duration": "1s"
|
||||
}
|
||||
|
||||
@@ -5,12 +5,12 @@ go 1.25.0
|
||||
require (
|
||||
github.com/ClusterCockpit/cc-lib/v2 v2.12.0
|
||||
github.com/ClusterCockpit/go-rocm-smi v0.4.0
|
||||
github.com/NVIDIA/go-nvml v0.13.2-0
|
||||
github.com/NVIDIA/go-nvml v0.13.3-1
|
||||
github.com/PaesslerAG/gval v1.2.4
|
||||
github.com/fsnotify/fsnotify v1.10.1
|
||||
github.com/tklauser/go-sysconf v0.4.0
|
||||
golang.design/x/runtime v0.3.0
|
||||
golang.org/x/sys v0.45.0
|
||||
golang.org/x/sys v0.47.0
|
||||
)
|
||||
|
||||
require (
|
||||
|
||||
@@ -13,8 +13,8 @@ github.com/Microsoft/go-winio v0.6.1/go.mod h1:LRdKpFKfdobln8UmuiYcKPot9D2v6svN5
|
||||
github.com/Microsoft/hcsshim v0.11.4 h1:68vKo2VN8DE9AdN4tnkWnmdhqdbpUFM8OF3Airm7fz8=
|
||||
github.com/Microsoft/hcsshim v0.11.4/go.mod h1:smjE4dvqPX9Zldna+t5FG3rnoHhaB7QYxPRqGcpAD9w=
|
||||
github.com/NVIDIA/go-nvml v0.13.0-1/go.mod h1:+KNA7c7gIBH7SKSJ1ntlwkfN80zdx8ovl4hrK3LmPt4=
|
||||
github.com/NVIDIA/go-nvml v0.13.2-0 h1:7M4cFG62wSUHw8i0XSiNU7ejKODytTS6ZrW/vgB2NSI=
|
||||
github.com/NVIDIA/go-nvml v0.13.2-0/go.mod h1:ahi2psRYoa+wYUBIrZPRO+wJs9lcvMhxSSkjjvsJJNQ=
|
||||
github.com/NVIDIA/go-nvml v0.13.3-1 h1:P76U2h88OZSiMtdhRsJjSF5DXyXUqHIXKeDicVAaae0=
|
||||
github.com/NVIDIA/go-nvml v0.13.3-1/go.mod h1:ahi2psRYoa+wYUBIrZPRO+wJs9lcvMhxSSkjjvsJJNQ=
|
||||
github.com/PaesslerAG/gval v1.2.4 h1:rhX7MpjJlcxYwL2eTTYIOBUyEKZ+A96T9vQySWkVUiU=
|
||||
github.com/PaesslerAG/gval v1.2.4/go.mod h1:XRFLwvmkTEdYziLdaCeCa5ImcGVrfQbeNUbVR+C6xac=
|
||||
github.com/PaesslerAG/jsonpath v0.1.0 h1:gADYeifvlqK3R3i2cR5B4DGgxLXIPb3TRTH1mGi0jPI=
|
||||
@@ -184,8 +184,8 @@ golang.org/x/mod v0.13.0 h1:I/DsJXRlw/8l/0c24sM9yb0T4z9liZTduXvdAWYiysY=
|
||||
golang.org/x/mod v0.13.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
|
||||
golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
|
||||
golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
|
||||
golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
|
||||
golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
|
||||
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
||||
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
|
||||
golang.org/x/tools v0.14.0 h1:jvNa2pY0M4r62jkRQ6RwEZZyPcymeL9XZMLBbV7U2nc=
|
||||
|
||||
Reference in New Issue
Block a user