Implement warmup delay for LIKWID collector

This reverts commit 9bd9a07810.

This should never have made it into our main repo.

README.md

@@ -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"
   }

cc-metric-collector.go

@@ -29,15 +29,17 @@ import (
 )
 
 type CentralConfigFile struct {
-	Interval string `json:"interval"`
+	Interval      string `json:"interval"`
-	Duration string `json:"duration"`
+	Duration      string `json:"duration"`
+	EnableMarkers bool   `json:"enable-markers"`
 }
 
 type RuntimeConfig struct {
-	Interval   time.Duration
+	Interval      time.Duration
-	Duration   time.Duration
+	Duration      time.Duration
-	CliArgs    map[string]string
+	EnableMarkers bool
-	ConfigFile CentralConfigFile
+	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

@@ -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
+	collectors    []MetricCollector          // List of metric collectors to read in parallel
-	serial       []MetricCollector          // List of metric collectors to read serially
+	serial        []MetricCollector          // List of metric collectors to read serially
-	output       chan lp.CCMessage          // Output channels
+	output        chan lp.CCMessage          // Output channels
-	done         chan bool                  // channel to finish / stop metric collector manager
+	done          chan bool                  // channel to finish / stop metric collector manager
-	ticker       mct.MultiChanTicker        // periodically ticking once each interval
+	ticker        mct.MultiChanTicker        // periodically ticking once each interval
-	duration     time.Duration              // duration (for metrics that measure over a given duration)
+	duration      time.Duration              // duration (for metrics that measure over a given duration)
-	wg           *sync.WaitGroup            // wait group for all goroutines in cc-metric-collector
+	wg            *sync.WaitGroup            // wait group for all goroutines in cc-metric-collector
-	config       map[string]json.RawMessage // json encoded config for collector manager
+	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
+	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
+	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
 	}

collectors/likwidMetric.go

@@ -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,28 @@ 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 {
+		fmt.Printf("Performing warmup measurement for %s\n", m.warmupDelay)
+		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 +549,7 @@ func (m *LikwidCollector) takeMeasurement(evidx int, evset LikwidEventsetConfig,
 	// Wait
 	time.Sleep(interval)
 
-	// Read counters
+	// end measuring
 	select {
 	case <-sigchan:
 		ret = -1

collectors/likwidMetric.md

@@ -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
 

example-configs/config.json

@@ -4,6 +4,7 @@
   "receivers-file" : "./receivers.json",
   "router-file" : "./router.json",
   "main" : {
+    "enable-markers": false,
     "interval": "10s",
     "duration": "1s"
   }

go.mod

@@ -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 (

go.sum

@@ -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.3-1 h1:P76U2h88OZSiMtdhRsJjSF5DXyXUqHIXKeDicVAaae0=
-github.com/NVIDIA/go-nvml v0.13.2-0/go.mod h1:ahi2psRYoa+wYUBIrZPRO+wJs9lcvMhxSSkjjvsJJNQ=
+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.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
-golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+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=