Update cclog calls in latest IB Collector

Merge branch 'main' into cclog_update
Update cclog calls
2026-07-18 15:40:38 +02:00 · 2026-06-08 14:58:29 +02:00 · 2026-06-08 14:57:31 +02:00 · 2026-06-08 14:52:24 +02:00 · 2026-06-02 16:35:11 +02:00 · 2026-06-02 13:52:44 +02:00
47 changed files with 722 additions and 1156 deletions
@@ -17,7 +17,3 @@

 # Local copy of LIKWID headers
 /collectors/likwid
-
-# Local development workspace (build against a local cc-lib checkout)
-go.work
-go.work.sum
@@ -41,8 +41,6 @@ There is a main configuration file with basic settings that point to the other c

 The `interval` defines how often the metrics should be read and send to the sink(s). The `duration` tells the collectors how long one measurement has to take. This is important for some collectors, like the `likwid` collector. For more information, see [here](./docs/configuration.md).

-The optional `channel_buffer_size` sets the capacity of the internal channels between the components. If unset, it defaults to `max(200, 24 * number of CPUs)` so that one interval's burst of per-hwthread metrics fits without back-pressuring the collectors on nodes with many cores.
-
 See the component READMEs for their configuration:

 * [`collectors`](./collectors/README.md)
@@ -13,7 +13,6 @@ import (
 	"flag"
 	"os"
 	"os/signal"
-	"runtime"
 	"sync"
 	"syscall"
 	"time"
@@ -30,9 +29,8 @@ import (
 )

 type CentralConfigFile struct {
-	Interval          string `json:"interval"`
-	Duration          string `json:"duration"`
-	ChannelBufferSize int    `json:"channel_buffer_size,omitempty"`
+	Interval string `json:"interval"`
+	Duration string `json:"duration"`
 }

 type RuntimeConfig struct {
@@ -134,11 +132,11 @@ func mainFunc() int {
 	if len(rcfg.ConfigFile.Interval) > 0 {
 		t, err := time.ParseDuration(rcfg.ConfigFile.Interval)
 		if err != nil {
-			cclog.Error("Configuration value 'interval' no valid duration")
+			cclog.Errorf("Configuration value interval=%s no valid duration", rcfg.ConfigFile.Interval)
 		}
 		rcfg.Interval = t
 		if rcfg.Interval == 0 {
-			cclog.Error("Configuration value 'interval' must be greater than zero")
+			cclog.Errorf("Configuration value interval=%s must be greater than zero", rcfg.ConfigFile.Interval)
 			return 1
 		}
 	}
@@ -147,11 +145,11 @@ func mainFunc() int {
 	if len(rcfg.ConfigFile.Duration) > 0 {
 		t, err := time.ParseDuration(rcfg.ConfigFile.Duration)
 		if err != nil {
-			cclog.Error("Configuration value 'duration' no valid duration")
+			cclog.Error("Configuration value duration=%s no valid duration", rcfg.ConfigFile.Duration)
 		}
 		rcfg.Duration = t
 		if rcfg.Duration == 0 {
-			cclog.Error("Configuration value 'duration' must be greater than zero")
+			cclog.Error("Configuration value duration=%s must be greater than zero", rcfg.ConfigFile.Duration)
 			return 1
 		}
 	}
@@ -160,14 +158,6 @@ func mainFunc() int {
 		return 1
 	}

-	// Size the channels between the managers so that one interval's burst of
-	// per-hwthread metrics fits without back-pressuring the collectors
-	chanSize := rcfg.ConfigFile.ChannelBufferSize
-	if chanSize <= 0 {
-		chanSize = max(200, 24*runtime.NumCPU())
-	}
-	cclog.ComponentDebug("main", "channel buffer size", chanSize)
-
 	routerConf := ccconf.GetPackageConfig("router")
 	if len(routerConf) == 0 {
 		cclog.Error("Metric router configuration file must be set")
@@ -204,7 +194,7 @@ func mainFunc() int {
 	}

 	// Connect metric router to sink manager
-	RouterToSinksChannel := make(chan lp.CCMessage, chanSize)
+	RouterToSinksChannel := make(chan lp.CCMessage, 200)
 	rcfg.SinkManager.AddInput(RouterToSinksChannel)
 	rcfg.MetricRouter.AddOutput(RouterToSinksChannel)

@@ -216,7 +206,7 @@ func mainFunc() int {
 	}

 	// Connect collector manager to metric router
-	CollectToRouterChannel := make(chan lp.CCMessage, chanSize)
+	CollectToRouterChannel := make(chan lp.CCMessage, 200)
 	rcfg.CollectManager.AddOutput(CollectToRouterChannel)
 	rcfg.MetricRouter.AddCollectorInput(CollectToRouterChannel)

@@ -230,7 +220,7 @@ func mainFunc() int {
 		}

 		// Connect receive manager to metric router
-		ReceiveToRouterChannel := make(chan lp.CCMessage, chanSize)
+		ReceiveToRouterChannel := make(chan lp.CCMessage, 200)
 		rcfg.ReceiveManager.AddOutput(ReceiveToRouterChannel)
 		rcfg.MetricRouter.AddReceiverInput(ReceiveToRouterChannel)
 		use_recv = true
@@ -209,16 +209,16 @@ func (m *BeegfsMetaCollector) Read(interval time.Duration, output chan lp.CCMess
 				} else {
 					f1, err := strconv.ParseFloat(m.matches["other"], 32)
 					if err != nil {
-						cclog.ComponentError(
+						cclog.ComponentErrorf(
 							m.name,
-							fmt.Sprintf("Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err))
+							"Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err)
 						continue
 					}
 					f2, err := strconv.ParseFloat(split[i], 32)
 					if err != nil {
-						cclog.ComponentError(
+						cclog.ComponentErrorf(
 							m.name,
-							fmt.Sprintf("Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err))
+							"Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err)
 						continue
 					}
 					m.matches["beegfs_cstorage_other"] = fmt.Sprintf("%f", f1+f2)
@@ -200,16 +200,16 @@ func (m *BeegfsStorageCollector) Read(interval time.Duration, output chan lp.CCM
 				} else {
 					f1, err := strconv.ParseFloat(m.matches["other"], 32)
 					if err != nil {
-						cclog.ComponentError(
+						cclog.ComponentErrorf(
 							m.name,
-							fmt.Sprintf("Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err))
+							"Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err)
 						continue
 					}
 					f2, err := strconv.ParseFloat(split[i], 32)
 					if err != nil {
-						cclog.ComponentError(
+						cclog.ComponentErrorf(
 							m.name,
-							fmt.Sprintf("Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err))
+							"Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err)
 						continue
 					}
 					m.matches["beegfs_cstorage_other"] = fmt.Sprintf("%f", f1+f2)
@@ -12,7 +12,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"sync"
-	"sync/atomic"
 	"time"

 	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
@@ -22,52 +21,50 @@ import (

 // Map of all available metric collectors
 var AvailableCollectors = map[string]MetricCollector{
-	"likwid":             new(LikwidCollector),
-	"loadavg":            new(LoadavgCollector),
-	"memstat":            new(MemstatCollector),
-	"netstat":            new(NetstatCollector),
-	"ibstat":             new(InfinibandCollector),
-	"lustrestat":         new(LustreCollector),
-	"cpustat":            new(CpustatCollector),
-	"topprocs":           new(TopProcsCollector),
-	"nvidia":             new(NvidiaCollector),
-	"customcmd":          new(CustomCmdCollector),
-	"iostat":             new(IOstatCollector),
-	"diskstat":           new(DiskstatCollector),
-	"tempstat":           new(TempCollector),
-	"ipmistat":           new(IpmiCollector),
-	"gpfs":               new(GpfsCollector),
-	"cpufreq":            new(CPUFreqCollector),
-	"cpufreq_cpuinfo":    new(CPUFreqCpuInfoCollector),
-	"nfs3stat":           new(Nfs3Collector),
-	"nfs4stat":           new(Nfs4Collector),
-	"numastats":          new(NUMAStatsCollector),
-	"beegfs_meta":        new(BeegfsMetaCollector),
-	"beegfs_storage":     new(BeegfsStorageCollector),
-	"rapl":               new(RAPLCollector),
-	"rocm_smi":           new(RocmSmiCollector),
-	"self":               new(SelfCollector),
-	"schedstat":          new(SchedstatCollector),
-	"nfsiostat":          new(NfsIOStatCollector),
-	"slurm_cgroup":       new(SlurmCgroupCollector),
-	"smartmon":           new(SmartMonCollector),
-	"lenovo_dense_power": new(LenovoDensePowerCollector),
-	"megware_eureka":     new(MegwareEurekaCollector),
+	"likwid":          new(LikwidCollector),
+	"loadavg":         new(LoadavgCollector),
+	"memstat":         new(MemstatCollector),
+	"netstat":         new(NetstatCollector),
+	"ibstat":          new(InfinibandCollector),
+	"lustrestat":      new(LustreCollector),
+	"cpustat":         new(CpustatCollector),
+	"topprocs":        new(TopProcsCollector),
+	"nvidia":          new(NvidiaCollector),
+	"customcmd":       new(CustomCmdCollector),
+	"iostat":          new(IOstatCollector),
+	"diskstat":        new(DiskstatCollector),
+	"tempstat":        new(TempCollector),
+	"ipmistat":        new(IpmiCollector),
+	"gpfs":            new(GpfsCollector),
+	"cpufreq":         new(CPUFreqCollector),
+	"cpufreq_cpuinfo": new(CPUFreqCpuInfoCollector),
+	"nfs3stat":        new(Nfs3Collector),
+	"nfs4stat":        new(Nfs4Collector),
+	"numastats":       new(NUMAStatsCollector),
+	"beegfs_meta":     new(BeegfsMetaCollector),
+	"beegfs_storage":  new(BeegfsStorageCollector),
+	"rapl":            new(RAPLCollector),
+	"rocm_smi":        new(RocmSmiCollector),
+	"self":            new(SelfCollector),
+	"schedstat":       new(SchedstatCollector),
+	"nfsiostat":       new(NfsIOStatCollector),
+	"slurm_cgroup":    new(SlurmCgroupCollector),
+	"smartmon":        new(SmartMonCollector),
+	"nvidia_gpm":      new(NvidiaGPMCollector),
 }

 // 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
-	round_wg      sync.WaitGroup             // wait group for the currently running collection round
-	round_running atomic.Bool                // Flag whether a collection round is currently running
+	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
 }

 // Metric collector manager access functions
@@ -103,17 +100,17 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
 	// Initialize configured collectors
 	for collectorName, collectorCfg := range cm.config {
 		if _, found := AvailableCollectors[collectorName]; !found {
-			cclog.ComponentError("CollectorManager", "SKIP unknown collector", collectorName)
+			cclog.ComponentErrorf("CollectorManager", "SKIP unknown collector %s", collectorName)
 			continue
 		}
 		collector := AvailableCollectors[collectorName]

 		err := collector.Init(collectorCfg)
 		if err != nil {
-			cclog.ComponentError("CollectorManager", fmt.Sprintf("Collector %s initialization failed: %v", collectorName, err))
+			cclog.ComponentErrorf("CollectorManager", "Collector %s initialization failed: %v", collectorName, err)
 			continue
 		}
-		cclog.ComponentDebug("CollectorManager", "ADD COLLECTOR", collector.Name())
+		cclog.ComponentDebugf("CollectorManager", "ADD COLLECTOR %s", collector.Name())
 		if collector.Parallel() {
 			cm.collectors = append(cm.collectors, collector)
 		} else {
@@ -123,54 +120,26 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
 	return nil
 }

-// runRound executes one collection round: first all parallel collectors
-// concurrently, then the serial collectors one by one
-func (cm *collectorManager) runRound(t time.Time) {
-	roundStart := time.Now()
-	for _, c := range cm.collectors {
-		// Read metrics from collector c via goroutine
-		cclog.ComponentDebug("CollectorManager", c.Name(), t)
-		cm.collector_wg.Add(1)
-		go func(myc MetricCollector) {
-			start := time.Now()
-			myc.Read(cm.duration, cm.output)
-			cclog.ComponentDebug("CollectorManager", myc.Name(), "took", time.Since(start))
-			cm.collector_wg.Done()
-		}(c)
-	}
-	cm.collector_wg.Wait()
-	for _, c := range cm.serial {
-		// Read metrics from collector c
-		cclog.ComponentDebug("CollectorManager", c.Name(), t)
-		start := time.Now()
-		c.Read(cm.duration, cm.output)
-		cclog.ComponentDebug("CollectorManager", c.Name(), "took", time.Since(start))
-	}
-	cclog.ComponentDebug("CollectorManager", "collection round took", time.Since(roundStart))
-}
-
 // Start starts the metric collector manager
 func (cm *collectorManager) Start() {
-	tick := make(chan time.Time, 1)
+	tick := make(chan time.Time)
 	cm.ticker.AddChannel(tick)

 	cm.wg.Go(func() {
 		// Collector manager is done
 		done := func() {
-			// wait for a still running collection round, then close all metric collectors
-			cm.round_wg.Wait()
-			for _, c := range cm.collectors {
-				c.Close()
+			// close all metric collectors
+			if cm.parallel_run {
+				cm.collector_wg.Wait()
+				cm.parallel_run = false
 			}
-			for _, c := range cm.serial {
+			for _, c := range cm.collectors {
 				c.Close()
 			}
 			close(cm.done)
 			cclog.ComponentDebug("CollectorManager", "DONE")
 		}

-		var roundStart time.Time
-
 		// Wait for done signal or timer event
 		for {
 			select {
@@ -178,20 +147,37 @@ func (cm *collectorManager) Start() {
 				done()
 				return
 			case t := <-tick:
-				// The round runs detached from this loop, so the tick channel
-				// stays drained even when a round takes longer than the interval
-				if cm.round_running.Load() {
-					cclog.ComponentWarn("CollectorManager", "collection round still running after", time.Since(roundStart), "- skipping tick")
-					continue
+				cm.parallel_run = true
+				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.ComponentDebugf("CollectorManager: Read %s at %v", 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
+					select {
+					case <-cm.done:
+						done()
+						return
+					default:
+						// Read metrics from collector c
+						cclog.ComponentDebugf("CollectorManager: Read %s at %v", c.Name(), t)
+						c.Read(cm.duration, cm.output)
+					}
 				}
-				cm.round_running.Store(true)
-				roundStart = time.Now()
-				cm.round_wg.Add(1)
-				go func() {
-					defer cm.round_wg.Done()
-					defer cm.round_running.Store(false)
-					cm.runRound(t)
-				}()
 			}
 		}
 	})
@@ -1,121 +0,0 @@
-package collectors
-
-import (
-	"encoding/json"
-	"sync"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
-)
-
-// Fake ticker that delivers ticks on demand
-type fakeTicker struct {
-	channels []chan time.Time
-}
-
-func (t *fakeTicker) Init(duration time.Duration) {}
-
-func (t *fakeTicker) AddChannel(c chan time.Time) {
-	t.channels = append(t.channels, c)
-}
-
-func (t *fakeTicker) Close() {}
-
-func (t *fakeTicker) tick() {
-	for _, c := range t.channels {
-		select {
-		case c <- time.Now():
-		default:
-		}
-	}
-}
-
-// Stub collector whose Read blocks until it is released
-type stubCollector struct {
-	metricCollector
-	readStarted chan struct{}
-	release     chan struct{}
-	reads       atomic.Int32
-}
-
-func (c *stubCollector) Init(config json.RawMessage) error {
-	c.name = "teststub"
-	c.parallel = true
-	c.init = true
-	return nil
-}
-
-func (c *stubCollector) Read(duration time.Duration, output chan lp.CCMessage) {
-	c.reads.Add(1)
-	c.readStarted <- struct{}{}
-	<-c.release
-}
-
-func (c *stubCollector) Close() {}
-
-func TestOverlongCollectionRoundSkipsTick(t *testing.T) {
-	stub := &stubCollector{
-		readStarted: make(chan struct{}, 10),
-		release:     make(chan struct{}),
-	}
-	AvailableCollectors["teststub"] = stub
-	defer delete(AvailableCollectors, "teststub")
-
-	ticker := &fakeTicker{}
-	var wg sync.WaitGroup
-	cm, err := New(ticker, time.Second, &wg, json.RawMessage(`{"teststub": {}}`))
-	if err != nil {
-		t.Fatalf("failed to setup collector manager: %s", err.Error())
-	}
-	cm.AddOutput(make(chan lp.CCMessage, 100))
-	cm.Start()
-
-	// First tick starts a collection round that blocks in Read
-	ticker.tick()
-	select {
-	case <-stub.readStarted:
-	case <-time.After(5 * time.Second):
-		t.Fatal("collection round did not start on tick")
-	}
-
-	// Further ticks while the round is still running must be skipped,
-	// not queued up or run concurrently
-	for range 3 {
-		ticker.tick()
-		time.Sleep(20 * time.Millisecond)
-	}
-	if got := stub.reads.Load(); got != 1 {
-		t.Fatalf("expected 1 concurrent collection round, got %d reads", got)
-	}
-
-	// Finish the round, the next tick must start a new one
-	stub.release <- struct{}{}
-	deadline := time.After(5 * time.Second)
-	for stub.reads.Load() < 2 {
-		ticker.tick()
-		select {
-		case <-stub.readStarted:
-		case <-time.After(20 * time.Millisecond):
-		case <-deadline:
-			t.Fatal("no new collection round started after the previous one finished")
-		}
-	}
-
-	// Shutdown must wait for the running round and terminate cleanly.
-	// Closing the release channel lets any still running or straggler
-	// round finish immediately
-	close(stub.release)
-	closed := make(chan struct{})
-	go func() {
-		cm.Close()
-		close(closed)
-	}()
-	select {
-	case <-closed:
-	case <-time.After(5 * time.Second):
-		t.Fatal("Close() did not terminate")
-	}
-	wg.Wait()
-}
@@ -116,9 +116,6 @@ func (m *CPUFreqCpuInfoCollector) Init(_ json.RawMessage) error {
 			physicalPackageID = ""
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		return fmt.Errorf("%s Init(): Call to scanner.Err failed: %w", m.name, err)
-	}

 	if err := file.Close(); err != nil {
 		return fmt.Errorf("%s Init(): Call to file.Close() failed: %w", m.name, err)
@@ -142,16 +139,16 @@ func (m *CPUFreqCpuInfoCollector) Read(interval time.Duration, output chan lp.CC
 	const cpuInfoFile = "/proc/cpuinfo"
 	file, err := os.Open(cpuInfoFile)
 	if err != nil {
-		cclog.ComponentError(
+		cclog.ComponentErrorf(
 			m.name,
-			fmt.Sprintf("Read(): Failed to open file '%s': %v", cpuInfoFile, err))
+			"Read(): Failed to open file '%s': %v", cpuInfoFile, err)
 		return
 	}
 	defer func() {
 		if err := file.Close(); err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to close file '%s': %v", cpuInfoFile, err))
+				"Read(): Failed to close file '%s': %v", cpuInfoFile, err)
 		}
 	}()

@@ -169,9 +166,9 @@ func (m *CPUFreqCpuInfoCollector) Read(interval time.Duration, output chan lp.CC
 				if !t.isHT {
 					value, err := strconv.ParseFloat(strings.TrimSpace(lineSplit[1]), 64)
 					if err != nil {
-						cclog.ComponentError(
+						cclog.ComponentErrorf(
 							m.name,
-							fmt.Sprintf("Read(): Failed to convert cpu MHz '%s' to float64: %v", lineSplit[1], err))
+							"Read(): Failed to convert cpu MHz '%s' to float64: %v", lineSplit[1], err)
 						return
 					}
 					if y, err := lp.NewMetric("cpufreq", t.tagSet, m.meta, value, now); err == nil {
@@ -182,9 +179,6 @@ func (m *CPUFreqCpuInfoCollector) Read(interval time.Duration, output chan lp.CC
 			}
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		cclog.ComponentErrorf(m.name, "Read(): Call to scanner.Err failed: %s", err.Error())
-	}
 }

 func (m *CPUFreqCpuInfoCollector) Close() {
@@ -95,10 +95,7 @@ func (m *CPUFreqCollector) Init(config json.RawMessage) error {
 	}

 	// Initialized
-	cclog.ComponentDebug(
-		m.name,
-		"initialized",
-		len(m.topology), "non-hyper-threading CPUs")
+	cclog.ComponentDebugf(m.name, "initialized %d non-hyper-threading CPUs")
 	m.init = true
 	return nil
 }
@@ -116,16 +113,14 @@ func (m *CPUFreqCollector) Read(interval time.Duration, output chan lp.CCMessage
 		// Read current frequency
 		line, err := os.ReadFile(t.scalingCurFreqFile)
 		if err != nil {
-			cclog.ComponentError(
-				m.name,
-				fmt.Sprintf("Read(): Failed to read file '%s': %v", t.scalingCurFreqFile, err))
+			cclog.ComponentErrorf(
+				m.name, "Read(): Failed to read file '%s': %v", t.scalingCurFreqFile, err)
 			continue
 		}
 		cpuFreq, err := strconv.ParseInt(strings.TrimSpace(string(line)), 10, 64)
 		if err != nil {
 			cclog.ComponentError(
-				m.name,
-				fmt.Sprintf("Read(): Failed to convert CPU frequency '%s' to int64: %v", line, err))
+				m.name, "Read(): Failed to convert CPU frequency '%s' to int64: %v", line, err)
 			continue
 		}

@@ -44,7 +44,7 @@ type CpustatCollector struct {
 func (m *CpustatCollector) Init(config json.RawMessage) error {
 	m.name = "CpustatCollector"
 	if err := m.setup(); err != nil {
-		return fmt.Errorf("%s Init(): setup() call failed: %s", m.name, err.Error())
+		return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
 	}
 	m.parallel = true
 	m.meta = map[string]string{
@@ -58,7 +58,7 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
 		d := json.NewDecoder(bytes.NewReader(config))
 		d.DisallowUnknownFields()
 		if err := d.Decode(&m.config); err != nil {
-			return fmt.Errorf("%s Init(): Error decoding JSON config: %s", m.name, err.Error())
+			return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
 		}
 	}
 	matches := map[string]int{
@@ -85,7 +85,7 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
 	// Check input file
 	file, err := os.Open(CPUSTATFILE)
 	if err != nil {
-		return fmt.Errorf("%s Init(): Failed to open file '%s': %s", m.name, CPUSTATFILE, err.Error())
+		return fmt.Errorf("%s Init(): Failed to open file '%s': %w", m.name, CPUSTATFILE, err)
 	}

 	// Pre-generate tags for all CPUs
@@ -117,13 +117,10 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
 			num_cpus++
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		return fmt.Errorf("%s Init(): Call to scanner.Err failed: %s", m.name, err.Error())
-	}

 	// Close file
 	if err := file.Close(); err != nil {
-		return fmt.Errorf("%s Init(): Failed to close file '%s': %s", m.name, CPUSTATFILE, err.Error())
+		return fmt.Errorf("%s Init(): Failed to close file '%s': %w", m.name, CPUSTATFILE, err)
 	}

 	m.lastTimestamp = time.Now()
@@ -174,15 +171,15 @@ func (m *CpustatCollector) Read(interval time.Duration, output chan lp.CCMessage

 	file, err := os.Open(CPUSTATFILE)
 	if err != nil {
-		cclog.ComponentError(
+		cclog.ComponentErrorf(
 			m.name,
-			fmt.Sprintf("Read(): Failed to open file '%s': %v", CPUSTATFILE, err.Error()))
+			"Read(): Failed to open file '%s': %v", CPUSTATFILE, err)
 	}
 	defer func() {
 		if err := file.Close(); err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err.Error()))
+				"Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err)
 		}
 	}()

@@ -197,9 +194,6 @@ func (m *CpustatCollector) Read(interval time.Duration, output chan lp.CCMessage
 			num_cpus++
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		cclog.ComponentErrorf(m.name, "Init(): Call to scanner.Err failed: %s", err.Error())
-	}

 	if !m.config.excludeNumCPUs {
 		if num_cpus_metric, err := lp.NewMetric("num_cpus", m.nodetags, m.meta, num_cpus, now); err == nil {
@@ -64,9 +64,9 @@ func (m *CustomCmdCollector) Init(config json.RawMessage) error {
 		cmdFields := strings.Fields(c)
 		command := exec.Command(cmdFields[0], cmdFields[1:]...)
 		if _, err := command.Output(); err != nil {
-			cclog.ComponentWarn(
+			cclog.ComponentWarnf(
 				m.name,
-				fmt.Sprintf("%s Init(): Execution of command \"%s\" failed: %v", m.name, command.String(), err))
+				"%s Init(): Execution of command \"%s\" failed: %v", m.name, command.String(), err)
 			continue
 		}
 		m.cmdFieldsSlice = append(m.cmdFieldsSlice, cmdFields)
@@ -77,7 +77,7 @@ func (m *CustomCmdCollector) Init(config json.RawMessage) error {
 		if _, err := os.ReadFile(fileName); err != nil {
 			cclog.ComponentWarn(
 				m.name,
-				fmt.Sprintf("%s Init(): Reading of file \"%s\" failed: %v", m.name, fileName, err))
+				"%s Init(): Reading of file \"%s\" failed: %v", m.name, fileName, err)
 			continue
 		}
 		m.files = append(m.files, fileName)
@@ -100,20 +100,18 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMessa
 		command := exec.Command(cmdFields[0], cmdFields[1:]...)
 		stdout, err := command.Output()
 		if err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to read command output for command \"%s\": %v", command.String(), err),
-			)
+				"Read(): Failed to read command output for command \"%s\": %v", command.String(), err)
 			continue
 		}

 		// Read and decode influxDB line-protocol from command output
 		metrics, err := lp.FromBytes(stdout)
 		if err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to decode influx Message: %v", err),
-			)
+				"Read(): Failed to decode influx Message: %v", err)
 			continue
 		}
 		for _, metric := range metrics {
@@ -128,20 +126,18 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMessa
 	for _, filename := range m.files {
 		input, err := os.ReadFile(filename)
 		if err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to read file \"%s\": %v\n", filename, err),
-			)
+				"Read(): Failed to read file \"%s\": %v\n", filename, err)
 			continue
 		}

 		// Read and decode influxDB line-protocol from file
 		metrics, err := lp.FromBytes(input)
 		if err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to decode influx Message: %v", err),
-			)
+				"Read(): Failed to decode influx Message: %v", err)
 			continue
 		}
 		for _, metric := range metrics {
@@ -77,16 +77,16 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMessag

 	file, err := os.Open(MOUNTFILE)
 	if err != nil {
-		cclog.ComponentError(
+		cclog.ComponentErrorf(
 			m.name,
-			fmt.Sprintf("Read(): Failed to open file '%s': %v", MOUNTFILE, err))
+			"Read(): Failed to open file '%s': %v", MOUNTFILE, err)
 		return
 	}
 	defer func() {
 		if err := file.Close(); err != nil {
 			cclog.ComponentError(
 				m.name,
-				fmt.Sprintf("Read(): Failed to close file '%s': %v", MOUNTFILE, err))
+				"Read(): Failed to close file '%s': %v", MOUNTFILE, err)
 		}
 	}()

@@ -147,11 +147,6 @@ mountLoop:
 			}
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		cclog.ComponentErrorf(
-			m.name,
-			"Read(): Call to scanner.Err failed: %s", err.Error())
-	}
 	if m.allowedMetrics["part_max_used"] {
 		y, err := lp.NewMetric("part_max_used", map[string]string{"type": "node"}, m.meta, int(part_max_used), time.Now())
 		if err == nil {
@@ -371,7 +371,7 @@ func (m *GpfsCollector) Init(config json.RawMessage) error {
 	if err != nil {
 		// if using sudo, exec.lookPath will return EACCES (file mode r-x------), this can be ignored
 		if m.config.Sudo && errors.Is(err, syscall.EACCES) {
-			cclog.ComponentWarn(m.name, fmt.Sprintf("got error looking for mmpmon binary '%s': %v . This is expected when using sudo, continuing.", m.config.Mmpmon, err))
+			cclog.ComponentWarnf(m.name, "got error looking for mmpmon binary '%s': %v . This is expected when using sudo, continuing.", m.config.Mmpmon, err)
 			// the file was given in the config, use it
 			p = m.config.Mmpmon
 		} else {
@@ -517,23 +517,23 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
 		// return code
 		rc, err := strconv.Atoi(key_value["_rc_"])
 		if err != nil {
-			cclog.ComponentError(m.name, fmt.Sprintf("Read(): Failed to convert return code '%s' to int: %v", key_value["_rc_"], err))
+			cclog.ComponentErrorf(m.name, "Read(): Failed to convert return code '%s' to int: %v", key_value["_rc_"], err)
 			continue
 		}
 		if rc != 0 {
-			cclog.ComponentError(m.name, fmt.Sprintf("Read(): Filesystem '%s' is not ok.", filesystem))
+			cclog.ComponentErrorf(m.name, "Read(): Filesystem '%s' is not ok.", filesystem)
 			continue
 		}

 		// timestamp
 		sec, err := strconv.ParseInt(key_value["_t_"], 10, 64)
 		if err != nil {
-			cclog.ComponentError(m.name, fmt.Sprintf("Read(): Failed to convert seconds '%s' to int64: %v", key_value["_t_"], err))
+			cclog.ComponentErrorf(m.name, "Read(): Failed to convert seconds '%s' to int64: %v", key_value["_t_"], err)
 			continue
 		}
 		msec, err := strconv.ParseInt(key_value["_tu_"], 10, 64)
 		if err != nil {
-			cclog.ComponentError(m.name, fmt.Sprintf("Read(): Failed to convert micro seconds '%s' to int64: %v", key_value["_tu_"], err))
+			cclog.ComponentErrorf(m.name, "Read(): Failed to convert micro seconds '%s' to int64: %v", key_value["_tu_"], err)
 			continue
 		}
 		timestamp := time.Unix(sec, msec*1000)
@@ -551,7 +551,7 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
 		for _, metric := range GpfsAbsMetrics {
 			value, err := strconv.ParseInt(key_value[metric.prefix], 10, 64)
 			if err != nil {
-				cclog.ComponentError(m.name, fmt.Sprintf("Read(): Failed to convert %s '%s' to int64: %v", metric.desc, key_value[metric.prefix], err))
+				cclog.ComponentErrorf(m.name, "Read(): Failed to convert %s '%s' to int64: %v", metric.desc, key_value[metric.prefix], err)
 				continue
 			}
 			newstate[metric.prefix] = value
@@ -636,7 +636,7 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
 				}
 			} else {
 				// the value could not be computed correctly
-				cclog.ComponentWarn(m.name, fmt.Sprintf("Read(): Could not compute value for filesystem %s of metric %s: vold_ok = %t, vnew_ok = %t", filesystem, metric.name, vold_ok, vnew_ok))
+				cclog.ComponentWarnf(m.name, "Read(): Could not compute value for filesystem %s of metric %s: vold_ok = %t, vnew_ok = %t", filesystem, metric.name, vold_ok, vnew_ok)
 			}
 		}

@@ -225,9 +225,9 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMess
 			// Read counter file
 			line, err := os.ReadFile(counterDef.path)
 			if err != nil {
-				cclog.ComponentError(
+				cclog.ComponentErrorf(
 					m.name,
-					fmt.Sprintf("Read(): Failed to read from file '%s': %v", counterDef.path, err))
+					"Read(): Failed to read from file '%s': %v", counterDef.path, err)
 				// Current counter can not be saved as last state
 				counterDef.lastStateAvailable = false
 				continue
@@ -237,9 +237,9 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMess
 			// convert counter to uint64
 			vRawCounter, err := strconv.ParseUint(data, 10, 64)
 			if err != nil {
-				cclog.ComponentError(
+				cclog.ComponentErrorf(
 					m.name,
-					fmt.Sprintf("Read(): Failed to convert Infininiband metrice %s='%s' to uint64: %v", counterDef.name, data, err))
+					"Read(): Failed to convert Infininiband metrice %s='%s' to uint64: %v", counterDef.name, data, err)
 				// Current counter can not be saved as last state
 				counterDef.lastStateAvailable = false
 				continue
@@ -48,13 +48,13 @@ func (m *IOstatCollector) Init(config json.RawMessage) error {
 	m.parallel = true
 	m.meta = map[string]string{"source": m.name, "group": "Disk"}
 	if err := m.setup(); err != nil {
-		return fmt.Errorf("%s Init(): setup() call failed: %s", m.name, err.Error())
+		return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
 	}
 	if len(config) > 0 {
 		d := json.NewDecoder(bytes.NewReader(config))
 		d.DisallowUnknownFields()
 		if err := d.Decode(&m.config); err != nil {
-			return fmt.Errorf("%s Init(): Error decoding JSON config: %s", m.name, err.Error())
+			return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
 		}
 	}
 	// https://www.kernel.org/doc/html/latest/admin-guide/iostats.html
@@ -89,7 +89,7 @@ func (m *IOstatCollector) Init(config json.RawMessage) error {
 	}
 	file, err := os.Open(IOSTATFILE)
 	if err != nil {
-		return fmt.Errorf("%s Init(): Failed to open file \"%s\": %s", m.name, IOSTATFILE, err.Error())
+		return fmt.Errorf("%s Init(): Failed to open file \"%s\": %w", m.name, IOSTATFILE, err)
 	}

 	scanner := bufio.NewScanner(file)
@@ -130,11 +130,8 @@ func (m *IOstatCollector) Init(config json.RawMessage) error {
 			lastValues:    lastValues,
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		return fmt.Errorf("%s Init(): Failed to scan content of file %q: %s", m.name, IOSTATFILE, err.Error())
-	}
 	if err := file.Close(); err != nil {
-		return fmt.Errorf("%s Init(): Failed to close file \"%s\": %s", m.name, IOSTATFILE, err.Error())
+		return fmt.Errorf("%s Init(): Failed to close file \"%s\": %w", m.name, IOSTATFILE, err)
 	}

 	m.init = true
@@ -148,16 +145,16 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMessage)

 	file, err := os.Open(IOSTATFILE)
 	if err != nil {
-		cclog.ComponentError(
+		cclog.ComponentErrorf(
 			m.name,
-			fmt.Sprintf("Read(): Failed to open file '%s': %s", IOSTATFILE, err.Error()))
+			"Read(): Failed to open file '%s': %v", IOSTATFILE, err)
 		return
 	}
 	defer func() {
 		if err := file.Close(); err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to close file '%s': %s", IOSTATFILE, err.Error()))
+				"Read(): Failed to close file '%s': %v", IOSTATFILE, err)
 		}
 	}()

@@ -201,11 +198,6 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMessage)
 		}
 		m.devices[device] = entry
 	}
-	if err := scanner.Err(); err != nil {
-		cclog.ComponentErrorf(
-			m.name,
-			"Read(): Failed to scan content of file %q: %s", IOSTATFILE, err.Error())
-	}
 }

 func (m *IOstatCollector) Close() {
@@ -28,15 +28,13 @@ type IpmiCollector struct {
 	metricCollector

 	config struct {
-		IpmitoolPath    string   `json:"ipmitool_path"`
-		IpmisensorsPath string   `json:"ipmisensors_path"`
-		Sudo            bool     `json:"use_sudo"`
-		IncludeMetrics  []string `json:"include_metrics"`
+		IpmitoolPath    string `json:"ipmitool_path"`
+		IpmisensorsPath string `json:"ipmisensors_path"`
+		Sudo            bool   `json:"use_sudo"`
 	}

-	ipmitool       string
-	ipmisensors    string
-	includeMetrics map[string]bool
+	ipmitool    string
+	ipmisensors string
 }

 func (m *IpmiCollector) Init(config json.RawMessage) error {
@@ -66,15 +64,6 @@ func (m *IpmiCollector) Init(config json.RawMessage) error {
 		}
 	}

-	// Read metrics to include
-	m.includeMetrics = make(map[string]bool)
-	for _, metric := range m.config.IncludeMetrics {
-		metric = strings.ToLower(strings.TrimSpace(metric))
-		if metric != "" {
-			m.includeMetrics[metric] = true
-		}
-	}
-
 	m.ipmitool = m.config.IpmitoolPath
 	m.ipmisensors = m.config.IpmisensorsPath

@@ -156,11 +145,6 @@ func (m *IpmiCollector) readIpmiTool(output chan lp.CCMessage) error {
 			continue
 		}
 		name := strings.ToLower(strings.ReplaceAll(strings.TrimSpace(lv[0]), " ", "_"))
-
-		if len(m.includeMetrics) > 0 && !m.includeMetrics[name] {
-			continue
-		}
-
 		unit := strings.TrimSpace(lv[2])
 		switch unit {
 		case "Volts":
@@ -181,9 +165,6 @@ func (m *IpmiCollector) readIpmiTool(output chan lp.CCMessage) error {
 		y.AddMeta("unit", unit)
 		output <- y
 	}
-	if err := scanner.Err(); err != nil {
-		return fmt.Errorf("failed to scan output of command: %s", err.Error())
-	}

 	// Wait for command end
 	if err := command.Wait(); err != nil {
@@ -228,11 +209,6 @@ func (m *IpmiCollector) readIpmiSensors(output chan lp.CCMessage) error {
 			continue
 		}
 		name := strings.ToLower(strings.ReplaceAll(lv[1], " ", "_"))
-
-		if len(m.includeMetrics) > 0 && !m.includeMetrics[name] {
-			continue
-		}
-
 		y, err := lp.NewMetric(name, map[string]string{"type": "node"}, m.meta, v, time.Now())
 		if err != nil {
 			cclog.ComponentErrorf(m.name, "Failed to create message: %v", err)
@@ -15,8 +15,7 @@ hugo_path: docs/reference/cc-metric-collector/collectors/ipmi.md
  "ipmistat": {
    "ipmitool_path": "/path/to/ipmitool",
    "ipmisensors_path": "/path/to/ipmi-sensors",
-    "use_sudo": true,
-    "include_metrics" : []
+    "use_sudo": true
  }
 ```

@@ -37,5 +36,3 @@ Defaults: monitoring !log_allowed, !pam_session
 monitoring ALL = (root) NOPASSWD:/usr/bin/ipmitool sensor
 monitoring ALL = (root) NOPASSWD:/usr/sbin/ipmi-sensors --comma-separated-output --sdr-cache-recreate
 ```
-
-If `include_ipmi_metrics` contains any entry, ipmistat collector will only submit these metrics. Any other values will get discarded. 
@@ -1,170 +0,0 @@
-// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
-// All rights reserved.
-// Use of this source code is governed by a MIT-style
-// license that can be found in the LICENSE file.
-// additional authors:
-// Michael Panzlaff (NHR@FAU)
-
-package collectors
-
-import (
-	"bytes"
-	"encoding/hex"
-	"encoding/json"
-	"fmt"
-	"os/exec"
-	"strconv"
-	"strings"
-	"time"
-
-	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
-	lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
-)
-
-type LenovoDensePowerCollector struct {
-	metricCollector
-
-	config struct {
-		IpmitoolPath string `json:"ipmitool_path"`
-		Sudo         bool   `json:"use_sudo"`
-	}
-
-	ipmitool string
-
-	energyValLast  float64
-	energyTimeLast time.Time
-}
-
-func (m *LenovoDensePowerCollector) Init(config json.RawMessage) error {
-	// Check if already initialized
-	if m.init {
-		return nil
-	}
-
-	m.name = "LenovoDensePowerCollector"
-	if err := m.setup(); err != nil {
-		return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
-	}
-	m.meta = map[string]string{
-		"source": m.name,
-		"group":  "IPMI",
-	}
-
-	m.config.IpmitoolPath = "ipmitool"
-
-	if len(config) > 0 {
-		d := json.NewDecoder(bytes.NewReader(config))
-		d.DisallowUnknownFields()
-		if err := d.Decode(&m.config); err != nil {
-			return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
-		}
-	}
-
-	m.ipmitool = m.config.IpmitoolPath
-
-	energyVal, energyTime, err := m.readEnergy()
-	if err != nil {
-		return fmt.Errorf("energy reading test failed: %w", err)
-	}
-
-	m.energyValLast = energyVal
-	m.energyTimeLast = energyTime
-	m.init = true
-
-	return nil
-}
-
-func (m *LenovoDensePowerCollector) readEnergy() (float64, time.Time, error) {
-	argv := make([]string, 0)
-	if m.config.Sudo {
-		argv = append(argv, "sudo", "-n")
-	}
-
-	// The Lenovo hardware wants this magic byte string
-	lenovoRequestEnergyMsg := []uint8{0x3a, 0x32, 4, 2, 0, 0, 0}
-
-	argv = append(argv, m.ipmitool, "raw")
-
-	for _, val := range lenovoRequestEnergyMsg {
-		argv = append(argv, strconv.Itoa(int(val)))
-	}
-
-	cmd := exec.Command(argv[0], argv[1:]...)
-
-	var stdout bytes.Buffer
-	var stderr bytes.Buffer
-	cmd.Stdout = &stdout
-	cmd.Stderr = &stderr
-
-	err := cmd.Run()
-	if err != nil {
-		return 0, time.Unix(0, 0), fmt.Errorf("failed to run ipmitool: %w (stdout=%s stderr=%s)", err, stdout.String(), stderr.String())
-	}
-
-	data, err := hex.DecodeString(strings.ReplaceAll(strings.TrimSpace(stdout.String()), " ", ""))
-	if err != nil {
-		return 0, time.Unix(0, 0), fmt.Errorf("unable to decode ipmitool hex output: %w (stdout=%s)", err, stdout.String())
-	}
-
-	if len(data) != 14 {
-		return 0, time.Unix(0, 0), fmt.Errorf("result must be 14 bytes as specified in the documentation")
-	}
-
-	// data is laid out like this:
-	// data[0 to 1]: FIFO index of value (for debug only)
-	// data[2 to 5]: Integer part of energy in J (little endian)
-	// data[6 to 7]: Fraction part of energy in mJ (little endian)
-	// data[8 to 11]: Integer part of Unix time of measurement in seconds (little endian)
-	// data[12 to 13]: Fraction part of Unix time of measurement in milliseconds (little endian)
-	wholeJoules := (uint32(data[2]) << 0) | (uint32(data[3]) << 8) | (uint32(data[4]) << 16) | (uint32(data[5]) << 24)
-	milliJoules := uint16(data[6]) | (uint16(data[7]) << 8)
-	finalJoules := float64(wholeJoules) + float64(milliJoules)*1e-3
-
-	wholeSeconds := (uint32(data[8]) << 0) | (uint32(data[9]) << 8) | (uint32(data[10]) << 16) | (uint32(data[11]) << 24)
-	milliSeconds := uint16(data[12]) | (uint16(data[13]) << 8)
-	finalTime := time.Unix(int64(wholeSeconds), (int64(milliSeconds) * 1000000))
-
-	return finalJoules, finalTime, nil
-}
-
-func (m *LenovoDensePowerCollector) Read(interval time.Duration, output chan lp.CCMessage) {
-	// Check if already initialized
-	if !m.init {
-		return
-	}
-
-	energyVal, energyTime, err := m.readEnergy()
-	if err != nil {
-		cclog.ComponentErrorf(m.name, "readEnergy failed: %v", err)
-		return
-	}
-
-	powerVal := 0.0
-	if energyTime.Compare(m.energyTimeLast) != 0 {
-		// Check for overflow
-		energyValDiff := int64(energyVal) - int64(m.energyValLast)
-		if energyVal < m.energyValLast {
-			energyValDiff += int64(0x100000000)
-		}
-		energyTimeDiff := energyTime.Sub(m.energyTimeLast)
-		if energyTime.Before(m.energyTimeLast) {
-			energyTimeDiff = energyTimeDiff + 0x100000000*time.Second
-		}
-		powerVal = float64(energyValDiff) / energyTimeDiff.Seconds()
-
-		m.energyValLast = energyVal
-		m.energyTimeLast = energyTime
-	}
-
-	metric, err := lp.NewMetric("lenovo_node_power", map[string]string{"type": "node"}, m.meta, powerVal, energyTime)
-	if err != nil {
-		cclog.ComponentErrorf(m.name, "lp.NewMetric failed: %v", err)
-		return
-	}
-	metric.AddMeta("unit", "Watts")
-	output <- metric
-}
-
-func (m *LenovoDensePowerCollector) Close() {
-	m.init = false
-}
@@ -1,51 +0,0 @@
-<!--
---
-title: Lenovo Dense Power collector
-description: Collect power of Lenovo Dense machines using ipmitool raw
-categories: [cc-metric-collector]
-tags: ['Admin']
-weight: 2
-hugo_path: docs/reference/cc-metric-collector/collectors/lenovoDensePower.md
---
-->
-
-## `lenovo_dense_power` collector
-
-```json
-  "lenovo_dense_power": {
-    "ipmitool_path": "/path/to/ipmitool",
-    "use_sudo": true,
-    "include_metrics" : []
-  }
-```
-
-The `lenovo_dense_power` collector reads power from Lenovo Dense machines via `ipmitool` (`ipmitool raw ...`). This collector is known to only work on the following machines. Others are not supported, so use at your *OWN* risk:
-
-System                                    | Compatibility             | Power measured
---                                       | ---                       | ---
-Lenovo ThinkSystem SD650 V2/V3            | untested, but should work | node + riser1 + riser2
-Lenovo ThinkSystem SD650-N V2/ SD650-I V3 | untested, but should work | node + riser1 + riser2 + gpus
-Lenovo ThinkSystem SD665-N V3             | tested, known to work     | node + riser1 + riser2 + gpus
-
-To test if your system *may* be supported, you can run the following command *at your own risk*:
-
-```
-$ ipmitool raw 58 50 4 2 0 0 0
- c4 00 83 a9 cb d3 a2 03 df db 47 6a d8 01
-```
-
-Exactly 14 bytes should be returned.
-In case not or another error occurs your hardware is likely not supported.
-
-In addition, `ipmitool` typically require root to run.
-In order to run `cc-metric-collector` without root priviliges, you can enable `use_sudo`.
-Add a file like this in `/etc/sudoers.d/` to allow `cc-metric-collector` to run the required commands:
-
-```
-# Do not log the following sudo commands from monitoring, since this causes a lot of log spam.
-# However keep log_denied enabled, to detect failures
-Defaults: monitoring !log_allowed, !pam_session
-
-# Allow to use ipmitool for Lenovo power readings
-monitoring ALL = (root) NOPASSWD:/usr/bin/ipmitool raw 58 50 4 2 0 0 0
-```
@@ -12,6 +12,12 @@ package collectors
 #cgo LDFLAGS: -Wl,--unresolved-symbols=ignore-in-object-files
 #include <stdlib.h>
 #include <likwid.h>
+
+
+int cc_add_hwthread(int cpu_id) {
+    return HPMaddThread(cpu_id);
+}
+
 */
 import "C"

@@ -38,7 +44,7 @@ import (
 	topo "github.com/ClusterCockpit/cc-metric-collector/pkg/ccTopology"
 	"github.com/NVIDIA/go-nvml/pkg/dl"
 	"github.com/fsnotify/fsnotify"
-	"golang.design/x/runtime/thread"
+	"golang.design/x/thread"
 )

 const (
@@ -261,12 +267,12 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
 			}
 			for _, metric := range evset.Metrics {
 				// Try to evaluate the metric
-				cclog.ComponentDebug(m.name, "Checking", metric.Name)
+				cclog.ComponentDebugf(m.name, "Checking %s", metric.Name)
 				if !checkMetricType(metric.Type) {
-					cclog.ComponentError(m.name, "Metric", metric.Name, "uses invalid type", metric.Type)
+					cclog.ComponentErrorf(m.name, "Metric %s uses invalid type %s", metric.Name, metric.Type)
 					metric.Calc = ""
 				} else if !testLikwidMetricFormula(metric.Calc, params) {
-					cclog.ComponentError(m.name, "Metric", metric.Name, "cannot be calculated with given counters")
+					cclog.ComponentError(m.name, "Metric %s cannot be calculated with given counters", metric.Name)
 					metric.Calc = ""
 				} else {
 					globalParams = append(globalParams, metric.Name)
@@ -281,13 +287,13 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
 	for _, metric := range m.config.Metrics {
 		// Try to evaluate the global metric
 		if !checkMetricType(metric.Type) {
-			cclog.ComponentError(m.name, "Metric", metric.Name, "uses invalid type", metric.Type)
+			cclog.ComponentErrorf(m.name, "Metric %s uses invalid type %s", metric.Name, metric.Type)
 			metric.Calc = ""
 		} else if !testLikwidMetricFormula(metric.Calc, globalParams) {
-			cclog.ComponentError(m.name, "Metric", metric.Name, "cannot be calculated with given counters")
+			cclog.ComponentError(m.name, "Metric %s cannot be calculated with given counters", metric.Name)
 			metric.Calc = ""
 		} else if !checkMetricType(metric.Type) {
-			cclog.ComponentError(m.name, "Metric", metric.Name, "has invalid type")
+			cclog.ComponentError(m.name, "Metric %s has invalid type", metric.Name)
 			metric.Calc = ""
 		} else {
 			totalMetrics++
@@ -328,7 +334,7 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
 		for _, c := range m.cpulist {
 			m.measureThread.Call(
 				func() {
-					retCode := C.HPMaddThread(C.uint32_t(c))
+					retCode := C.cc_add_hwthread(C.int(c))
 					if retCode != 0 {
 						err := fmt.Errorf("C.HPMaddThread(%v) failed with return code %v", c, retCode)
 						cclog.ComponentError(m.name, err.Error())
@@ -375,16 +381,16 @@ func (m *LikwidCollector) takeMeasurement(evidx int, evset LikwidEventsetConfig,
 	// Watch changes for the lock file ()
 	watcher, err := fsnotify.NewWatcher()
 	if err != nil {
-		cclog.ComponentError(
+		cclog.ComponentErrorf(
 			m.name,
-			fmt.Sprintf("takeMeasurement(): Failed to create a new fsnotify.Watcher: %v", err))
+			"takeMeasurement(): Failed to create a new fsnotify.Watcher: %v", err)
 		return true, err
 	}
 	defer func() {
 		if err := watcher.Close(); err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("takeMeasurement(): Failed to close fsnotify.Watcher: %v", err))
+				"takeMeasurement(): Failed to close fsnotify.Watcher: %v", err)
 		}
 	}()
 	if len(m.config.LockfilePath) > 0 {
@@ -597,7 +603,7 @@ func (m *LikwidCollector) calcEventsetMetrics(evset LikwidEventsetConfig, interv
 			if tid >= 0 && len(metric.Calc) > 0 {
 				value, err := agg.EvalFloat64Condition(metric.Calc, evset.results[tid])
 				if err != nil {
-					cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
+					cclog.ComponentErrorf(m.name, "Calculation for metric %s failed: %s", metric.Name, err.Error())
 					value = 0.0
 				}
 				if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) {
@@ -762,7 +768,7 @@ func (m *LikwidCollector) calcGlobalMetrics(groups []LikwidEventsetConfig, inter
 				// Evaluate the metric
 				value, err := agg.EvalFloat64Condition(metric.Calc, params)
 				if err != nil {
-					cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
+					cclog.ComponentErrorf(m.name, "Calculation for metric %s failed: %s", metric.Name, err.Error())
 					value = 0.0
 				}
 				if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) {
@@ -89,9 +89,9 @@ func (m *LoadavgCollector) Read(interval time.Duration, output chan lp.CCMessage
 	}
 	buffer, err := os.ReadFile(LOADAVGFILE)
 	if err != nil {
-		cclog.ComponentError(
+		cclog.ComponentErrorf(
 			m.name,
-			fmt.Sprintf("Read(): Failed to read file '%s': %v", LOADAVGFILE, err))
+			"Read(): Failed to read file '%s': %v", LOADAVGFILE, err)
 		return
 	}
 	now := time.Now()
@@ -101,9 +101,9 @@ func (m *LoadavgCollector) Read(interval time.Duration, output chan lp.CCMessage
 	for i, name := range m.load_matches {
 		x, err := strconv.ParseFloat(ls[i], 64)
 		if err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to convert '%s' to float64: %v", ls[i], err))
+				"Read(): Failed to convert '%s' to float64: %v", ls[i], err)
 			continue
 		}
 		if m.load_skips[i] {
@@ -120,9 +120,9 @@ func (m *LoadavgCollector) Read(interval time.Duration, output chan lp.CCMessage
 	for i, name := range m.proc_matches {
 		x, err := strconv.ParseInt(lv[i], 10, 64)
 		if err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to convert '%s' to float64: %v", lv[i], err))
+				"Read(): Failed to convert '%s' to float64: %v", lv[i], err)
 			continue
 		}
 		if m.proc_skips[i] {
@@ -1,194 +0,0 @@
-// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
-// All rights reserved.
-// Use of this source code is governed by a MIT-style
-// license that can be found in the LICENSE file.
-// additional authors:
-// Michael Panzlaff (NHR@FAU)
-
-package collectors
-
-import (
-	"bytes"
-	"encoding/json"
-	"fmt"
-	"os/exec"
-	"time"
-
-	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
-	lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
-)
-
-// MPS refers to Monolithic Power Systems, from which the MP5922 is used
-// to measure telemetry on the Eureka platform.
-type mpsData struct {
-	Cnt         int64   `json:"Cnt"` // No idea what this is, ignore
-	Energy      float64 `json:"Energy"`
-	Vin         float64 `json:"Vin"`
-	Iin         float64 `json:"Iin"`
-	Pin         float64 `json:"Pin"`
-	PinAvg      float64 `json:"PinAvg"`
-	Vout        float64 `json:"Vout"`
-	Iout        float64 `json:"Iout"`
-	Pout        float64 `json:"Pout"`
-	StandbyVout float64 `json:"StandbyVout"`
-	StandbyIout float64 `json:"StandbyIout"`
-	StandbyPout float64 `json:"StandbyPout"`
-	TempBusbar  float64 `json:"TempBusbar"`
-	TempSsd     float64 `json:"TempSsd"`
-	TempMps     float64 `json:"TempMps"`
-	// No idea what the ones below mean exactl. Ignore them for now.
-	EnergyTime         int64 `json:"EnergyTime"`
-	EnergyAccumulator  int64 `json:"EnergyAccumulator"`
-	EnergyRolloverCnt  int64 `json:"EnergyRolloverCnt"`
-	EnergySampleCntU24 int64 `json:"EnergySampleCntU24"`
-
-	Timestamp time.Time
-}
-
-type MegwareEurekaCollector struct {
-	metricCollector
-
-	config struct {
-		U20Path string `json:"u20_path"`
-		Sudo    bool   `json:"use_sudo"`
-	}
-
-	u20path string
-
-	energyValLast  float64
-	energyTimeLast time.Time
-}
-
-func (m *MegwareEurekaCollector) Init(config json.RawMessage) error {
-	// Check if already initialized
-	if m.init {
-		return nil
-	}
-
-	m.name = "MegwareEurekaCollector"
-	if err := m.setup(); err != nil {
-		return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
-	}
-	m.meta = map[string]string{
-		"source": m.name,
-		"group":  "U20",
-	}
-
-	m.config.U20Path = "u20"
-
-	if len(config) > 0 {
-		d := json.NewDecoder(bytes.NewReader(config))
-		d.DisallowUnknownFields()
-		if err := d.Decode(&m.config); err != nil {
-			return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
-		}
-	}
-
-	m.u20path = m.config.U20Path
-
-	data, err := m.readMpsData()
-	if err != nil {
-		return fmt.Errorf("energy reading test failed: %w", err)
-	}
-
-	m.energyValLast = data.Energy
-	m.energyTimeLast = time.Now()
-	m.init = true
-
-	return nil
-}
-
-func (m *MegwareEurekaCollector) readMpsData() (*mpsData, error) {
-	argv := make([]string, 0)
-	if m.config.Sudo {
-		argv = append(argv, "sudo", "-n")
-	}
-
-	argv = append(argv, m.u20path, "values", "--read", "GET_MPS_POLL_VALUES")
-
-	cmd := exec.Command(argv[0], argv[1:]...)
-
-	var stdout bytes.Buffer
-	var stderr bytes.Buffer
-	cmd.Stdout = &stdout
-	cmd.Stderr = &stderr
-
-	err := cmd.Run()
-	if err != nil {
-		return nil, fmt.Errorf("failed to run u20: %w (stdout=%s stderr=%s)", err, stdout.String(), stderr.String())
-	}
-
-	var u20output struct {
-		GetMpsPollValues mpsData `json:"GET_MPS_POLL_VALUES"`
-	}
-
-	err = json.Unmarshal(stdout.Bytes(), &u20output)
-	if err != nil {
-		return nil, fmt.Errorf("unable to decode u20 JSON output: %w (stdout=%s)", err, stdout.String())
-	}
-
-	u20output.GetMpsPollValues.Timestamp = time.Now()
-
-	return &u20output.GetMpsPollValues, nil
-}
-
-func (m *MegwareEurekaCollector) Read(interval time.Duration, output chan lp.CCMessage) {
-	// Check if already initialized
-	if !m.init {
-		return
-	}
-
-	data, err := m.readMpsData()
-	if err != nil {
-		cclog.ComponentErrorf(m.name, "readMpsData failed: %v", err)
-		return
-	}
-
-	powerVal := 0.0
-
-	if data.Timestamp.After(m.energyTimeLast) {
-		// Important, m.energy comes in Wh, so multiply by 3600 to get Ws (aka Joule)
-		energyValDiff := data.Energy - m.energyValLast
-		energyTimeDiff := data.Timestamp.Sub(m.energyTimeLast)
-		powerVal = energyValDiff * 3600 / energyTimeDiff.Seconds()
-
-		m.energyValLast = data.Energy
-		m.energyTimeLast = data.Timestamp
-	}
-
-	metricNamePrefix := "eureka_"
-	metricMap := map[string]struct {
-		value any
-		unit  string
-	}{
-		"power":        {value: powerVal, unit: "Watts"},
-		"vin":          {value: data.Vin, unit: "Volts"},
-		"iin":          {value: data.Iin, unit: "Amperes"},
-		"pin":          {value: data.Pin, unit: "Watts"},
-		"pin_avg":      {value: data.PinAvg, unit: "Watts"},
-		"vout":         {value: data.Vout, unit: "Volts"},
-		"iout":         {value: data.Iout, unit: "Amperes"},
-		"pout":         {value: data.Pout, unit: "Watts"},
-		"standby_vout": {value: data.StandbyVout, unit: "Volts"},
-		"standby_iout": {value: data.StandbyIout, unit: "Amperes"},
-		"standby_pout": {value: data.StandbyPout, unit: "Watts"},
-		"temp_busbar":  {value: data.TempBusbar, unit: "degC"},
-		"temp_ssd":     {value: data.TempSsd, unit: "degC"},
-		"temp_mps":     {value: data.TempMps, unit: "degC"},
-	}
-
-	for metricName, metricData := range metricMap {
-		metricName = metricNamePrefix + metricName
-		metric, err := lp.NewMetric(metricName, map[string]string{"type": "node"}, m.meta, metricData.value, data.Timestamp)
-		if err != nil {
-			cclog.ComponentErrorf(m.name, "lp.NewMetric failed: %v", err)
-			return
-		}
-		metric.AddMeta("unit", metricData.unit)
-		output <- metric
-	}
-}
-
-func (m *MegwareEurekaCollector) Close() {
-	m.init = false
-}
@@ -1,44 +0,0 @@
-<!--
---
-title: Megware Eureka collector
-description: Collect power and other metrics of Megware Eureka machines using u20
-categories: [cc-metric-collector]
-tags: ['Admin']
-weight: 2
-hugo_path: docs/reference/cc-metric-collector/collectors/lenovoDensePower.md
---
-->
-
-## `lenovo_dense_power` collector
-
-```json
-  "megware_eureka": {
-    "u20_path": "/path/to/ipmitool",
-    "use_sudo": true
-  }
-```
-
-The `megware_eureka` collector reads power and other metrics from Megware Eureka machines via `u20`.
-If the u20 tool is available for your machines, it's possible that this collector is compatible.
-If you don't have access to u20, please contact your Megware sales person.
-
-You can test the collector compatibility by running the following command:
-
-```
-$ u20 values --read GET_MPS_POLL_VALUES
-```
-
-If this returns a JSON with energy, voltage and current readings, you're fine.
-
-In addition, `u20` typically requires root to run.
-In order to run `cc-metric-collector` without root priviliges, you can enable `use_sudo`.
-Add a file like this in `/etc/sudoers.d/` to allow `cc-metric-collector` to run the required commands:
-
-```
-# Do not log the following sudo commands from monitoring, since this causes a lot of log spam.
-# However keep log_denied enabled, to detect failures
-Defaults: monitoring !log_allowed, !pam_session
-
-# Allow to use u20 for Megware power readings
-monitoring ALL = (root) NOPASSWD:/usr/bin/u20 values --read GET_MPS_POLL_VALUES
-```
@@ -84,7 +84,7 @@ func getStats(filename string) map[string]MemstatStats {
 		case 5:
 			v, err := strconv.ParseFloat(linefields[3], 64)
 			if err == nil {
-				cclog.ComponentDebug("getStats", strings.Trim(linefields[2], ":"), v, linefields[4])
+				cclog.ComponentDebug("MemstatCollector", "getStats %s value %v unit %s", strings.Trim(linefields[2], ":"), v, linefields[4])
 				stats[strings.Trim(linefields[2], ":")] = MemstatStats{
 					value: v,
 					unit:  linefields[4],
@@ -222,16 +222,16 @@ func (m *NetstatCollector) Read(interval time.Duration, output chan lp.CCMessage

 	file, err := os.Open(NETSTATFILE)
 	if err != nil {
-		cclog.ComponentError(
+		cclog.ComponentErrorf(
 			m.name,
-			fmt.Sprintf("Read(): Failed to open file '%s': %v", NETSTATFILE, err))
+			"Read(): Failed to open file '%s': %v", NETSTATFILE, err)
 		return
 	}
 	defer func() {
 		if err := file.Close(); err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to close file '%s': %v", NETSTATFILE, err))
+				"Read(): Failed to close file '%s': %v", NETSTATFILE, err)
 		}
 	}()

@@ -125,10 +125,9 @@ func (m *nfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
 	timestamp := time.Now()

 	if err := m.updateStats(); err != nil {
-		cclog.ComponentError(
+		cclog.ComponentErrorf(
 			m.name,
-			fmt.Sprintf("Read(): updateStats() failed: %v", err),
-		)
+			"Read(): updateStats() failed: %v", err)
 		return
 	}
 	var prefix string
@@ -117,7 +117,7 @@ func (m *NUMAStatsCollector) Init(config json.RawMessage) error {
 	}

 	// Initialized
-	cclog.ComponentDebug(m.name, "initialized", len(m.topology), "NUMA domains")
+	cclog.ComponentDebugf(m.name, "initialized %d NUMA domains", len(m.topology))
 	m.init = true
 	return nil
 }
@@ -0,0 +1,396 @@
+package collectors
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"slices"
+	"strconv"
+	"strings"
+	"time"
+
+	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
+	lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
+	"github.com/NVIDIA/go-nvml/pkg/nvml"
+)
+
+type NvidiaGPMMetricDef struct {
+	name    string
+	outname string
+	id      nvml.GpmMetricId
+	unit    string
+}
+
+var NvidiaGPMMetrics []NvidiaGPMMetricDef = []NvidiaGPMMetricDef{
+	{
+		name:    "GRAPHICS_UTIL",
+		outname: "nv_gpm_graphics_util",
+		id:      nvml.GPM_METRIC_GRAPHICS_UTIL,
+		unit:    "%",
+	},
+	{
+		name:    "SM_UTIL",
+		outname: "nv_gpm_sm_util",
+		id:      nvml.GPM_METRIC_SM_UTIL,
+		unit:    "%",
+	},
+	{
+		name:    "SM_OCCUPANCY",
+		outname: "nv_gpm_sm_occupancy",
+		id:      nvml.GPM_METRIC_SM_OCCUPANCY,
+		unit:    "%",
+	},
+	{
+		name:    "INTEGER_UTIL",
+		outname: "nv_gpm_integer_util",
+		id:      nvml.GPM_METRIC_INTEGER_UTIL,
+		unit:    "%",
+	},
+	{
+		name:    "ANY_TENSOR_UTIL",
+		outname: "nv_gpm_any_tensor_util",
+		id:      nvml.GPM_METRIC_ANY_TENSOR_UTIL,
+		unit:    "%",
+	},
+	{
+		name:    "DFMA_TENSOR_UTIL",
+		outname: "nv_gpm_dfma_tensor_util",
+		id:      nvml.GPM_METRIC_DFMA_TENSOR_UTIL,
+		unit:    "%",
+	},
+	{
+		name:    "HMMA_TENSOR_UTIL",
+		outname: "nv_gpm_hmma_tensor_util",
+		id:      nvml.GPM_METRIC_HMMA_TENSOR_UTIL,
+		unit:    "%",
+	},
+	{
+		name:    "IMMA_TENSOR_UTIL",
+		outname: "nv_gpm_imma_tensor_util",
+		id:      nvml.GPM_METRIC_IMMA_TENSOR_UTIL,
+		unit:    "%",
+	},
+	{
+		name:    "DRAM_BW_UTIL",
+		outname: "nv_gpm_dram_bw_util",
+		id:      nvml.GPM_METRIC_DRAM_BW_UTIL,
+		unit:    "%",
+	},
+	{
+		name:    "FP64_UTIL",
+		outname: "nv_gpm_fp64_util",
+		id:      nvml.GPM_METRIC_FP64_UTIL,
+		unit:    "%",
+	},
+	{
+		name:    "FP32_UTIL",
+		outname: "nv_gpm_fp32_util",
+		id:      nvml.GPM_METRIC_FP32_UTIL,
+		unit:    "%",
+	},
+	{
+		name:    "FP16_UTIL",
+		outname: "nv_gpm_fp16_util",
+		id:      nvml.GPM_METRIC_FP16_UTIL,
+		unit:    "%",
+	},
+}
+
+type NvidiaGPMCollectorConfig struct {
+	Metrics               []string `json:"metrics,omitempty"`
+	ExcludeDevices        []string `json:"exclude_devices,omitempty"`
+	AddPciInfoTag         bool     `json:"add_pci_info_tag,omitempty"`
+	UsePciInfoAsTypeId    bool     `json:"use_pci_info_as_type_id,omitempty"`
+	AddUuidMeta           bool     `json:"add_uuid_meta,omitempty"`
+	AddBoardNumberMeta    bool     `json:"add_board_number_meta,omitempty"`
+	AddSerialMeta         bool     `json:"add_serial_meta,omitempty"`
+	ProcessMigDevices     bool     `json:"process_mig_devices,omitempty"`
+	UseUuidForMigDevices  bool     `json:"use_uuid_for_mig_device,omitempty"`
+	UseSliceForMigDevices bool     `json:"use_slice_for_mig_device,omitempty"`
+}
+
+type NvidiaGPMCollectorDevice struct {
+	device        nvml.Device
+	tags          map[string]string
+	meta          map[string]string
+	startTime     time.Time
+	endTime       time.Time
+	measurement   nvml.GpmMetricsGetType
+	metricsLookup map[int]NvidiaGPMMetricDef
+}
+
+type NvidiaGPMCollector struct {
+	metricCollector
+
+	config   NvidiaGPMCollectorConfig
+	gpus     []NvidiaGPMCollectorDevice
+	num_gpus int
+}
+
+func (m *NvidiaGPMCollector) Init(config json.RawMessage) error {
+	var err error = nil
+	m.name = "NvidiaGPMCollector"
+	m.parallel = true
+	if err := m.setup(); err != nil {
+		return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
+	}
+	if len(config) > 0 {
+		d := json.NewDecoder(strings.NewReader(string(config)))
+		d.DisallowUnknownFields()
+		if err = d.Decode(&m.config); err != nil {
+			return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
+		}
+	}
+	m.meta = map[string]string{
+		"source": m.name,
+		"group":  "NvidiaGPM",
+	}
+
+	// Initialize NVIDIA Management Library (NVML)
+	ret := nvml.Init()
+
+	// Error: NVML library not found
+	// (nvml.ErrorString can not be used in this case)
+	if ret == nvml.ERROR_LIBRARY_NOT_FOUND {
+		return fmt.Errorf("%s Init(): NVML library not found", m.name)
+	}
+	if ret != nvml.SUCCESS {
+		err = errors.New(nvml.ErrorString(ret))
+		return fmt.Errorf("%s Init(): Unable to initialize NVML: %w", m.name, err)
+	}
+
+	// Number of NVIDIA GPUs
+	num_gpus, ret := nvml.DeviceGetCount()
+	if ret != nvml.SUCCESS {
+		err = errors.New(nvml.ErrorString(ret))
+		return fmt.Errorf("%s Init(): Unable to get device count: %w", m.name, err)
+	}
+
+	// For all GPUs
+	m.gpus = make([]NvidiaGPMCollectorDevice, 0, num_gpus)
+	for i := range num_gpus {
+
+		// Skip excluded devices by ID
+		str_i := strconv.Itoa(i)
+		if slices.Contains(m.config.ExcludeDevices, str_i) {
+			cclog.ComponentDebugf(m.name, "Skipping excluded device %s", str_i)
+			continue
+		}
+
+		// Get device handle
+		device, ret := nvml.DeviceGetHandleByIndex(i)
+		if ret != nvml.SUCCESS {
+			err = errors.New(nvml.ErrorString(ret))
+			cclog.ComponentErrorf(m.name, "Unable to get device at index %d: %s", i, err.Error())
+			continue
+		}
+
+		supportInfo, ret := nvml.GpmQueryDeviceSupport(device)
+		if ret != nvml.SUCCESS {
+			err = errors.New(nvml.ErrorString(ret))
+			cclog.ComponentErrorf(m.name, "Unable to query GPM support for device at index %d: %s", i, err.Error())
+			continue
+		} else {
+			if supportInfo.IsSupportedDevice == uint32(nvml.FEATURE_DISABLED) {
+				cclog.ComponentErrorf(m.name, "Device at index %d does not support GPM metrics", i)
+				continue
+			}
+		}
+
+		stream, ret := nvml.GpmQueryIfStreamingEnabled(device)
+		if ret != nvml.SUCCESS {
+			err = errors.New(nvml.ErrorString(ret))
+			cclog.ComponentErrorf(m.name, "Unable to query GPM streaming for device at index %d: %s", i, err.Error())
+			continue
+		} else {
+			if stream == uint32(nvml.FEATURE_DISABLED) {
+				ret = nvml.GpmSetStreamingEnabled(device, uint32(nvml.FEATURE_ENABLED))
+				if ret != nvml.SUCCESS {
+					err = errors.New(nvml.ErrorString(ret))
+					cclog.ComponentErrorf(m.name, "Unable to set streaming mode for device at index %d: %s", i, err.Error())
+				}
+			}
+		}
+
+		// Get device's PCI info
+		pciInfo, ret := nvml.DeviceGetPciInfo(device)
+		if ret != nvml.SUCCESS {
+			err = errors.New(nvml.ErrorString(ret))
+			cclog.ComponentErrorf(m.name, "Unable to get PCI info for device at index %d: %s", i, err.Error())
+			continue
+		}
+		// Create PCI ID in the common format used by the NVML.
+		pci_id := fmt.Sprintf(
+			nvml.DEVICE_PCI_BUS_ID_FMT,
+			pciInfo.Domain,
+			pciInfo.Bus,
+			pciInfo.Device)
+
+		// Skip excluded devices specified by PCI ID
+		if slices.Contains(m.config.ExcludeDevices, pci_id) {
+			cclog.ComponentDebugf(m.name, "Skipping excluded device %s", pci_id)
+			continue
+		}
+		ss, nvmlErr := nvml.GpmSampleAlloc()
+		if nvmlErr != nvml.SUCCESS {
+			err = errors.New(nvml.ErrorString(ret))
+			cclog.ComponentErrorf(m.name, "Failed to allocate GPM sample for device %d: %s", i, err.Error())
+			continue
+		}
+		es, nvmlErr := nvml.GpmSampleAlloc()
+		if nvmlErr != nvml.SUCCESS {
+			err = errors.New(nvml.ErrorString(ret))
+			cclog.ComponentErrorf(m.name, "Failed to allocate GPM sample for device %d: %s", i, err.Error())
+			continue
+		}
+
+		// Select which value to use as 'type-id'.
+		// The PCI ID is commonly required in SLURM environments because the
+		// numberic IDs used by SLURM and the ones used by NVML might differ
+		// depending on the job type. The PCI ID is more reliable but is commonly
+		// not recorded for a job, so it must be added manually in prologue or epilogue
+		// e.g. to the comment field
+		tid := str_i
+		if m.config.UsePciInfoAsTypeId {
+			tid = pci_id
+		}
+
+		// Now we got all infos together, populate the device list
+		g := NvidiaGPMCollectorDevice{}
+
+		// Add device handle
+		g.device = device
+
+		// Add tags
+		g.tags = map[string]string{
+			"type":    "accelerator",
+			"type-id": tid,
+		}
+
+		// Add PCI info as tag if not already used as 'type-id'
+		if m.config.AddPciInfoTag && !m.config.UsePciInfoAsTypeId {
+			g.tags["pci_identifier"] = pci_id
+		}
+
+		g.meta = map[string]string{
+			"source": m.name,
+			"group":  "Nvidia",
+		}
+
+		if m.config.AddBoardNumberMeta {
+			board, ret := nvml.DeviceGetBoardPartNumber(device)
+			if ret != nvml.SUCCESS {
+				err = errors.New(nvml.ErrorString(ret))
+				cclog.ComponentError(m.name, "Unable to get boart part number for device at index", i, ":", err.Error())
+			} else {
+				g.meta["board_number"] = board
+			}
+		}
+		if m.config.AddSerialMeta {
+			serial, ret := nvml.DeviceGetSerial(device)
+			if ret != nvml.SUCCESS {
+				err = errors.New(nvml.ErrorString(ret))
+				cclog.ComponentError(m.name, "Unable to get serial number for device at index", i, ":", err.Error())
+			} else {
+				g.meta["serial"] = serial
+			}
+		}
+		if m.config.AddUuidMeta {
+			uuid, ret := nvml.DeviceGetUUID(device)
+			if ret != nvml.SUCCESS {
+				err = errors.New(nvml.ErrorString(ret))
+				cclog.ComponentError(m.name, "Unable to get UUID for device at index", i, ":", err.Error())
+			} else {
+				g.meta["uuid"] = uuid
+			}
+		}
+
+		g.measurement.Sample1 = ss
+		g.measurement.Sample2 = es
+		g.measurement.Version = nvml.GPM_METRICS_GET_VERSION
+		g.metricsLookup = make(map[int]NvidiaGPMMetricDef)
+		metIdx := 0
+		for _, inmetric := range m.config.Metrics {
+			for _, defmetric := range NvidiaGPMMetrics {
+				if inmetric == defmetric.outname || inmetric == defmetric.name {
+					g.measurement.Metrics[metIdx] = nvml.GpmMetric{
+						MetricId: uint32(defmetric.id),
+					}
+					g.metricsLookup[metIdx] = defmetric
+					metIdx += 1
+				}
+			}
+		}
+		g.measurement.NumMetrics = uint32(metIdx)
+		m.gpus = append(m.gpus, g)
+	}
+	cclog.ComponentDebugf(m.name, "Found %d Nvidia GPUs with GPM support", len(m.gpus))
+	m.num_gpus = len(m.gpus)
+	m.init = true
+	return err
+}
+
+func (m *NvidiaGPMCollector) Read(interval time.Duration, output chan lp.CCMessage) {
+	var err error
+	if !m.init {
+		return
+	}
+	for i, gpu := range m.gpus {
+		gpu.startTime = time.Now()
+		nvmlErr := gpu.measurement.Sample1.Get(gpu.device)
+		if nvmlErr != nvml.SUCCESS {
+			err = errors.New(nvml.ErrorString(nvmlErr))
+			cclog.ComponentError(m.name, "Unable to get start GPM sample for device at index", i, ":", err.Error())
+			continue
+		}
+	}
+	time.Sleep(interval)
+
+	for i, gpu := range m.gpus {
+		gpu.endTime = time.Now()
+		nvmlErr := gpu.measurement.Sample2.Get(gpu.device)
+		if nvmlErr != nvml.SUCCESS {
+			err = errors.New(nvml.ErrorString(nvmlErr))
+			cclog.ComponentError(m.name, "Unable to get stop GPM sample for device at index", i, ":", err.Error())
+			continue
+		}
+	}
+
+	for i, gpu := range m.gpus {
+		nvmlErr := nvml.GpmMetricsGet(&gpu.measurement)
+		if nvmlErr != nvml.SUCCESS {
+			err = errors.New(nvml.ErrorString(nvmlErr))
+			cclog.ComponentError(m.name, "Unable to get evaluate GPM sample for device at index", i, ":", err.Error())
+			continue
+		}
+		for idx, metricDef := range gpu.metricsLookup {
+			y, err := lp.NewMetric(metricDef.outname, gpu.tags, gpu.meta, gpu.measurement.Metrics[idx].Value, time.Now())
+			if err == nil {
+				y.AddMeta("unit", metricDef.unit)
+				output <- y
+			}
+		}
+	}
+
+}
+
+func (m *NvidiaGPMCollector) Close() {
+	if m.init {
+		for i, gpu := range m.gpus {
+			ret := gpu.measurement.Sample1.Free()
+			if ret != nvml.SUCCESS {
+				err := errors.New(nvml.ErrorString(ret))
+				cclog.ComponentErrorf(m.name, "Unable to free start sample for device at index %d: %s", i, err.Error())
+			}
+			ret = gpu.measurement.Sample2.Free()
+			if ret != nvml.SUCCESS {
+				err := errors.New(nvml.ErrorString(ret))
+				cclog.ComponentErrorf(m.name, "Unable to free stop sample for device at index %d: %s", i, err.Error())
+			}
+		}
+		if ret := nvml.Shutdown(); ret != nvml.SUCCESS {
+			cclog.ComponentError(m.name, "nvml.Shutdown() not successful")
+		}
+		m.init = false
+	}
+}
@@ -0,0 +1,54 @@
+<!--
+---
+title: "Nvidia NVML GPM metric collector"
+description: Collect metrics for Nvidia GPUs using the NVML GPM interface
+categories: [cc-metric-collector]
+tags: ['Admin']
+weight: 2
+hugo_path: docs/reference/cc-metric-collector/collectors/nvidiaGPM.md
+---
+-->
+
+## `nvidiaGPM` collector
+
+```json
+  "nvidia_gpm": {
+    "metrics": [
+      "nv_fb_mem_used",
+      "nv_fan"
+    ],
+    "exclude_devices": [
+      "0","1", "0000000:ff:01.0"
+    ],
+    
+    "process_mig_devices": false,
+    "use_pci_info_as_type_id": true,
+    "add_pci_info_tag": false,
+    "add_uuid_meta": false,
+    "add_board_number_meta": false,
+    "add_serial_meta": false,
+    "use_uuid_for_mig_device": false,
+    "use_slice_for_mig_device": false
+  }
+```
+
+The `nvidia_gpm` collector can be configured to leave out specific devices with the `exclude_devices` option. It takes IDs as supplied to the NVML with `nvmlDeviceGetHandleByIndex()` or the PCI address in NVML format (`%08X:%02X:%02X.0`). Commonly only the physical GPUs are monitored. If MIG devices should be analyzed as well, set `process_mig_devices` (adds `stype=mig,stype-id=<mig_index>`). With the options `use_uuid_for_mig_device` and `use_slice_for_mig_device`, the `<mig_index>` can be replaced with the UUID (e.g. `MIG-6a9f7cc8-6d5b-5ce0-92de-750edc4d8849`) or the MIG slice name (e.g. `1g.5gb`).
+
+The metrics sent by the `nvidia_gpm` collector use `accelerator` as `type` tag. For the `type-id`, it uses the device handle index by default. With the `use_pci_info_as_type_id` option, the PCI ID is used instead. If both values should be added as tags, activate the `add_pci_info_tag` option. It uses the device handle index as `type-id` and adds the PCI ID as separate `pci_identifier` tag.
+
+Optionally, it is possible to add the UUID, the board part number and the serial to the meta informations. They are not sent to the sinks (if not configured otherwise).
+
+
+Available Metrics:
+* `nv_gpm_graphics_util`
+* `nv_gpm_sm_util`
+* `nv_gpm_sm_occupancy`
+* `nv_gpm_integer_util`
+* `nv_gpm_any_tensor_util`
+* `nv_gpm_dfma_tensor_util`
+* `nv_gpm_hmma_tensor_util`
+* `nv_gpm_imma_tensor_util`
+* `nv_gpm_dram_bw_util`
+* `nv_gpm_fp64_util`
+* `nv_gpm_fp32_util`
+* `nv_gpm_fp16_util`
@@ -113,7 +113,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
 		// Skip excluded devices by ID
 		str_i := strconv.Itoa(i)
 		if slices.Contains(m.config.ExcludeDevices, str_i) {
-			cclog.ComponentDebug(m.name, "Skipping excluded device", str_i)
+			cclog.ComponentDebugf(m.name, "Skipping excluded device %s", str_i)
 			continue
 		}

@@ -121,7 +121,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
 		device, ret := nvml.DeviceGetHandleByIndex(i)
 		if ret != nvml.SUCCESS {
 			err = errors.New(nvml.ErrorString(ret))
-			cclog.ComponentError(m.name, "Unable to get device at index", i, ":", err.Error())
+			cclog.ComponentErrorf(m.name, "Unable to get device at index %d: %s", i, err.Error())
 			continue
 		}

@@ -129,7 +129,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
 		pciInfo, ret := nvml.DeviceGetPciInfo(device)
 		if ret != nvml.SUCCESS {
 			err = errors.New(nvml.ErrorString(ret))
-			cclog.ComponentError(m.name, "Unable to get PCI info for device at index", i, ":", err.Error())
+			cclog.ComponentErrorf(m.name, "Unable to get PCI info for device at index %d: %s", i, err.Error())
 			continue
 		}
 		// Create PCI ID in the common format used by the NVML.
@@ -141,7 +141,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {

 		// Skip excluded devices specified by PCI ID
 		if slices.Contains(m.config.ExcludeDevices, pci_id) {
-			cclog.ComponentDebug(m.name, "Skipping excluded device", pci_id)
+			cclog.ComponentDebugf(m.name, "Skipping excluded device %s", pci_id)
 			continue
 		}

@@ -183,7 +183,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
 		if m.config.AddBoardNumberMeta {
 			board, ret := nvml.DeviceGetBoardPartNumber(device)
 			if ret != nvml.SUCCESS {
-				cclog.ComponentError(m.name, "Unable to get boart part number for device at index", i, ":", err.Error())
+				cclog.ComponentErrorf(m.name, "Unable to get boart part number for device at index %d: %s", i, err.Error())
 			} else {
 				g.meta["board_number"] = board
 			}
@@ -191,7 +191,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
 		if m.config.AddSerialMeta {
 			serial, ret := nvml.DeviceGetSerial(device)
 			if ret != nvml.SUCCESS {
-				cclog.ComponentError(m.name, "Unable to get serial number for device at index", i, ":", err.Error())
+				cclog.ComponentErrorf(m.name, "Unable to get serial number for device at index %d: %s", i, err.Error())
 			} else {
 				g.meta["serial"] = serial
 			}
@@ -199,7 +199,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
 		if m.config.AddUuidMeta {
 			uuid, ret := nvml.DeviceGetUUID(device)
 			if ret != nvml.SUCCESS {
-				cclog.ComponentError(m.name, "Unable to get UUID for device at index", i, ":", err.Error())
+				cclog.ComponentErrorf(m.name, "Unable to get UUID for device at index %d: %s", i, err.Error())
 			} else {
 				g.meta["uuid"] = uuid
 			}
@@ -1128,97 +1128,97 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMessage)
 		}
 		err = readMemoryInfo(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readMemoryInfo for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readMemoryInfo for device %s failed", name)
 		}

 		err = readUtilization(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readUtilization for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readUtilization for device %s failed", name)
 		}

 		err = readTemp(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readTemp for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readTemp for device %s failed", name)
 		}

 		err = readFan(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readFan for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readFan for device %s failed", name)
 		}

 		err = readEccMode(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readEccMode for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readEccMode for device %s failed", name)
 		}

 		err = readPerfState(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readPerfState for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readPerfState for device %s failed", name)
 		}

 		err = readPowerUsage(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readPowerUsage for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readPowerUsage for device %s failed", name)
 		}

 		err = readEnergyConsumption(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readEnergyConsumption for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readEnergyConsumption for device %s failed", name)
 		}

 		err = readClocks(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readClocks for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readClocks for device %s failed", name)
 		}

 		err = readMaxClocks(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readMaxClocks for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readMaxClocks for device %s failed", name)
 		}

 		err = readEccErrors(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readEccErrors for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readEccErrors for device %s failed", name)
 		}

 		err = readPowerLimit(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readPowerLimit for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readPowerLimit for device %s failed", name)
 		}

 		err = readEncUtilization(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readEncUtilization for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readEncUtilization for device %s failed", name)
 		}

 		err = readDecUtilization(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readDecUtilization for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readDecUtilization for device %s failed", name)
 		}

 		err = readRemappedRows(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readRemappedRows for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readRemappedRows for device %s failed", name)
 		}

 		err = readBarMemoryInfo(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readBarMemoryInfo for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readBarMemoryInfo for device %s failed", name)
 		}

 		err = readProcessCounts(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readProcessCounts for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readProcessCounts for device %s failed", name)
 		}

 		err = readViolationStats(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readViolationStats for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readViolationStats for device %s failed", name)
 		}

 		err = readNVLinkStats(device, output)
 		if err != nil {
-			cclog.ComponentDebug(m.name, "readNVLinkStats for device", name, "failed")
+			cclog.ComponentDebugf(m.name, "readNVLinkStats for device %s failed", name)
 		}
 	}

@@ -1244,7 +1244,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMessage)
 			if maxMig == 0 {
 				continue
 			}
-			cclog.ComponentDebug(m.name, "Reading MIG devices for GPU", i)
+			cclog.ComponentDebugf(m.name, "Reading MIG devices for GPU %d", i)

 			for j := range maxMig {
 				mdev, ret := nvml.DeviceGetMigDeviceHandleByIndex(m.gpus[i].device, j)
@@ -1268,7 +1268,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMessage)
 				if m.config.UseUuidForMigDevices {
 					uuid, ret := nvml.DeviceGetUUID(mdev)
 					if ret != nvml.SUCCESS {
-						cclog.ComponentError(m.name, "Unable to get UUID for mig device at index", j, ":", err.Error())
+						cclog.ComponentErrorf(m.name, "Unable to get UUID for mig device at index %d: %s", j, err.Error())
 					} else {
 						migDevice.tags["stype-id"] = uuid
 					}
@@ -208,11 +208,10 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
 	}

 	// Initialized
-	cclog.ComponentDebug(
+	cclog.ComponentDebugf(
 		m.name,
-		"initialized",
-		len(m.RAPLZoneInfo),
-		"zones with running average power limit (RAPL) monitoring attributes")
+		"initialized %d zones with running average power limit (RAPL) monitoring attributes",
+		len(m.RAPLZoneInfo))
 	m.init = true

 	return err
@@ -124,7 +124,7 @@ func (m *RocmSmiCollector) Init(config json.RawMessage) error {
 		if m.config.AddSerialMeta {
 			serial, ret := rocm_smi.DeviceGetSerialNumber(device)
 			if ret != rocm_smi.STATUS_SUCCESS {
-				cclog.ComponentError(m.name, "Unable to get serial number for device at index", i, ":", rocm_smi.StatusStringNoError(ret))
+				cclog.ComponentErrorf(m.name, "Unable to get serial number for device at index %d: %s", i, rocm_smi.StatusStringNoError(ret))
 			} else {
 				dev.meta["serial"] = serial
 			}
@@ -152,7 +152,7 @@ func (m *RocmSmiCollector) Read(interval time.Duration, output chan lp.CCMessage
 	for _, dev := range m.devices {
 		metrics, ret := rocm_smi.DeviceGetMetrics(dev.device)
 		if ret != rocm_smi.STATUS_SUCCESS {
-			cclog.ComponentError(m.name, "Unable to get metrics for device at index", dev.index, ":", rocm_smi.StatusStringNoError(ret))
+			cclog.ComponentErrorf(m.name, "Unable to get metrics for device at index %d: %s", dev.index, rocm_smi.StatusStringNoError(ret))
 			continue
 		}

@@ -147,15 +147,15 @@ func (m *SchedstatCollector) Read(interval time.Duration, output chan lp.CCMessa

 	file, err := os.Open(SCHEDSTATFILE)
 	if err != nil {
-		cclog.ComponentError(
+		cclog.ComponentErrorf(
 			m.name,
-			fmt.Sprintf("Read(): Failed to open file '%s': %v", SCHEDSTATFILE, err))
+			"Read(): Failed to open file '%s': %v", SCHEDSTATFILE, err)
 	}
 	defer func() {
 		if err := file.Close(); err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to close file '%s': %v", SCHEDSTATFILE, err))
+				"Read(): Failed to close file '%s': %v", SCHEDSTATFILE, err)
 		}
 	}()

@@ -7,7 +7,6 @@ import (
 	"os/exec"
 	"os/user"
 	"path/filepath"
-	"regexp"
 	"strconv"
 	"strings"
 	"time"
@@ -46,37 +45,6 @@ type SlurmCgroupCollector struct {

 const defaultCgroupBase = "/sys/fs/cgroup/system.slice/slurmstepd.scope"

-var (
-	// Slurm cgroup v2 directory layout:
-	// - Slurm <= 25.11: job_<numeric job id>
-	// - Slurm >= 26.05: SLUID, encoded as "s" + 13 Crockford Base32 characters
-	jobIDDirRE = regexp.MustCompile(`^job_[0-9]+$`)
-	sluidDirRE = regexp.MustCompile(`(?i)^s[0-9A-HJKMNP-TV-Z]{13}$`)
-)
-
-func (m *SlurmCgroupCollector) findSlurmJobDirs() ([]string, error) {
-	entries, err := os.ReadDir(m.cgroupBase)
-	if err != nil {
-		return nil, err
-	}
-
-	jobDirs := make([]string, 0)
-
-	for _, entry := range entries {
-		if !entry.IsDir() {
-			continue
-		}
-
-		name := entry.Name()
-
-		if jobIDDirRE.MatchString(name) || sluidDirRE.MatchString(name) {
-			jobDirs = append(jobDirs, filepath.Join(m.cgroupBase, name))
-		}
-	}
-
-	return jobDirs, nil
-}
-
 func ParseCPUs(cpuset string) ([]int, error) {
 	var result []int
 	if cpuset == "" {
@@ -269,9 +237,10 @@ func (m *SlurmCgroupCollector) Read(interval time.Duration, output chan lp.CCMes
 		delete(m.cpuUsed, k)
 	}

-	jobDirs, err := m.findSlurmJobDirs()
+	globPattern := filepath.Join(m.cgroupBase, "job_*")
+	jobDirs, err := filepath.Glob(globPattern)
 	if err != nil {
-		cclog.ComponentError(m.name, "Error reading job directories:", err.Error())
+		cclog.ComponentErrorf(m.name, "Error globbing job directories: %s", err.Error())
 		return
 	}

@@ -280,7 +249,7 @@ func (m *SlurmCgroupCollector) Read(interval time.Duration, output chan lp.CCMes

 		jobdata, err := m.ReadJobData(jKey)
 		if err != nil {
-			cclog.ComponentError(m.name, "Error reading job data for", jKey, ":", err.Error())
+			cclog.ComponentError(m.name, "Error reading job data for %s: %s", jKey, err.Error())
 			continue
 		}

@@ -228,12 +228,12 @@ func (m *SmartMonCollector) Read(interval time.Duration, output chan lp.CCMessag

 		stdout, err := command.Output()
 		if err != nil {
-			cclog.ComponentError(m.name, "cannot read data for device", d.Name)
+			cclog.ComponentErrorf(m.name, "cannot read data for device %s", d.Name)
 			continue
 		}
 		err = json.Unmarshal(stdout, &data)
 		if err != nil {
-			cclog.ComponentError(m.name, "cannot unmarshal data for device", d.Name)
+			cclog.ComponentErrorf(m.name, "cannot unmarshal data for device %s", d.Name)
 			continue
 		}
 		if !m.excludeMetric.temp {
@@ -188,16 +188,16 @@ func (m *TempCollector) Read(interval time.Duration, output chan lp.CCMessage) {
 		// Read sensor file
 		buffer, err := os.ReadFile(sensor.file)
 		if err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to read file '%s': %v", sensor.file, err))
+				"Read(): Failed to read file '%s': %v", sensor.file, err)
 			continue
 		}
 		x, err := strconv.ParseInt(strings.TrimSpace(string(buffer)), 10, 64)
 		if err != nil {
-			cclog.ComponentError(
+			cclog.ComponentErrorf(
 				m.name,
-				fmt.Sprintf("Read(): Failed to convert temperature '%s' to int64: %v", buffer, err))
+				"Read(): Failed to convert temperature '%s' to int64: %v", buffer, err)
 			continue
 		}
 		x /= 1000
@@ -77,9 +77,9 @@ func (m *TopProcsCollector) Read(interval time.Duration, output chan lp.CCMessag
 	command := exec.Command("ps", "-Ao", "comm", "--sort=-pcpu")
 	stdout, err := command.Output()
 	if err != nil {
-		cclog.ComponentError(
+		cclog.ComponentErrorf(
 			m.name,
-			fmt.Sprintf("Read(): Failed to read output from command \"%s\": %v", command.String(), err))
+			"Read(): Failed to read output from command \"%s\": %v", command.String(), err)
 		return
 	}

@@ -19,8 +19,6 @@ The global file contains the paths to the other four files and some global optio

 Be aware that the paths are relative to the execution folder of the cc-metric-collector binary, so it is recommended to use absolute paths.

-The optional `channel_buffer_size` option sets the capacity of the internal channels between the components (collectors → router → sinks). If unset, it defaults to `max(200, 24 * number of CPUs)` so that one interval's burst of per-hwthread metrics fits without back-pressuring the collectors on nodes with many cores.
-
 ## Component configuration

 The others are mainly list of of subcomponents: the collectors, the receivers, the router and the sinks. Their role is best shown in a picture:
@@ -5,11 +5,11 @@ 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.0-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.design/x/thread v0.3.2
 	golang.org/x/sys v0.45.0
 )

@@ -12,9 +12,8 @@ github.com/Microsoft/go-winio v0.6.1 h1:9/kr64B9VUZrLm5YYwbGtUJnMgqWVOdUAXu6Migc
 github.com/Microsoft/go-winio v0.6.1/go.mod h1:LRdKpFKfdobln8UmuiYcKPot9D2v6svN5+sAH+4kjUM=
 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 h1:OLX8Jq3dONuPOQPC7rndB6+iDmDakw0XTYgzMxObkEw=
 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/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=
@@ -174,8 +173,8 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
 go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
-golang.design/x/runtime v0.3.0 h1:8bA+GQfO8A18JXJQlQA56pEc+Sgrbo4cmLzjzmJ58ZI=
-golang.design/x/runtime v0.3.0/go.mod h1:PjVKQjImLdSrvjIw4FmrnCPq9BGs5PMBHb8i6b2BN9I=
+golang.design/x/thread v0.3.2 h1:FmD1glspGrQCe6FuQLmSrT6wz2CSzq7vKVDluyiMnqo=
+golang.design/x/thread v0.3.2/go.mod h1:6+Hi2rMOgMHZdKDWaqNHyWtoFUx1HxZ06LfHPh5Z/hQ=
 golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI=
 golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q=
 golang.org/x/exp v0.0.0-20231005195138-3e424a577f31 h1:9k5exFQKQglLo+RoP+4zMjOFE14P6+vyR0baDAi0Rcs=
@@ -128,7 +128,7 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
 	vars["starttime"] = starttime
 	vars["endtime"] = endtime
 	for _, f := range c.functions {
-		cclog.ComponentDebugf("MetricCache", "COLLECT %s COND '%s'", f.Name, f.Condition)
+		cclog.ComponentDebug("MetricCache", "COLLECT", f.Name, "COND", f.Condition)
 		var valuesFloat64 []float64
 		var valuesFloat32 []float32
 		var valuesInt []int
@@ -140,7 +140,7 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
 			vars["metric"] = m
 			value, err := f.gvalCond.EvalBool(context.Background(), vars)
 			if err != nil {
-				cclog.ComponentErrorf("MetricCache", "COLLECT %s COND '%s' : %s", f.Name, f.Condition, err.Error())
+				cclog.ComponentError("MetricCache", "COLLECT", f.Name, "COND", f.Condition, ":", err.Error())
 				continue
 			}
 			if value {
@@ -160,7 +160,7 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
 					case bool:
 						valuesBool = append(valuesBool, x)
 					default:
-						cclog.ComponentErrorf("MetricCache", "COLLECT ADD VALUE %v FAILED", v)
+						cclog.ComponentError("MetricCache", "COLLECT ADD VALUE", v, "FAILED")
 					}
 				}
 				matches = append(matches, m)
@@ -213,13 +213,13 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
 			vars["values"] = valuesBool
 			len_values = len(valuesBool)
 		}
-		cclog.ComponentDebugf("MetricCache", "EVALUATE %s METRICS %d CALC '%s'", f.Name, len_values, f.Function)
+		cclog.ComponentDebug("MetricCache", "EVALUATE", f.Name, "METRICS", len_values, "CALC", f.Function)

 		vars["metrics"] = matches
 		if len_values > 0 {
 			value, err := gval.Evaluate(f.Function, vars, c.language)
 			if err != nil {
-				cclog.ComponentErrorf("MetricCache", "EVALUATE %s METRICS %d CALC '%s': %s", f.Name, len_values, f.Function, err.Error())
+				cclog.ComponentError("MetricCache", "EVALUATE", f.Name, "METRICS", len_values, "CALC", f.Function, ":", err.Error())
 				break
 			}

@@ -273,12 +273,12 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
 			case string:
 				m, err = lp.NewMessage(f.Name, tags, meta, map[string]any{"value": t}, starttime)
 			default:
-				cclog.ComponentErrorf("MetricCache", "Gval returned invalid type %s skipping metric %s", t, f.Name)
+				cclog.ComponentError("MetricCache", "Gval returned invalid type", t, "skipping metric", f.Name)
 			}
 			if err != nil {
-				cclog.ComponentErrorf("MetricCache", "Cannot create metric from Gval result %v: %s", value, err.Error())
+				cclog.ComponentError("MetricCache", "Cannot create metric from Gval result", value, ":", err.Error())
 			}
-			cclog.ComponentDebugf("MetricCache", "SEND %s", m.ToLineProtocol(nil))
+			cclog.ComponentDebug("MetricCache", "SEND", m)
 			select {
 			case c.output <- m:
 			default:
@@ -295,12 +295,12 @@ func (c *metricAggregator) AddAggregation(name, function, condition string, tags
 	newcond := strings.ReplaceAll(condition, "'", "\"")
 	gvalCond, err := gval.Full(metricCacheLanguage).NewEvaluable(newcond)
 	if err != nil {
-		cclog.ComponentErrorf("MetricAggregator", "Cannot add aggregation, invalid if condition '%s': %s", newcond, err.Error())
+		cclog.ComponentError("MetricAggregator", "Cannot add aggregation, invalid if condition", newcond, ":", err.Error())
 		return err
 	}
 	gvalFunc, err := gval.Full(metricCacheLanguage).NewEvaluable(newfunc)
 	if err != nil {
-		cclog.ComponentErrorf("MetricAggregator", "Cannot add aggregation, invalid function condition %s: %s", newfunc, err.Error())
+		cclog.ComponentError("MetricAggregator", "Cannot add aggregation, invalid function condition", newfunc, ":", err.Error())
 		return err
 	}
 	for _, agg := range c.functions {
@@ -79,7 +79,7 @@ func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker,

 // Start starts the metric cache
 func (c *metricCache) Start() {
-	c.tickchan = make(chan time.Time, 1)
+	c.tickchan = make(chan time.Time)
 	c.ticker.AddChannel(c.tickchan)
 	// Router cache is done
 	done := func() {
@@ -35,18 +35,18 @@ type metricRouterTagConfig struct {

 // Metric router configuration
 type metricRouterConfig struct {
-	HostnameTagName   string                               `json:"hostname_tag"`        // Key name used when adding the hostname to a metric (default 'hostname')
-	AddTags           []metricRouterTagConfig              `json:"add_tags"`            // List of tags that are added when the condition is met
-	DelTags           []metricRouterTagConfig              `json:"delete_tags"`         // List of tags that are removed when the condition is met
-	IntervalAgg       []agg.MetricAggregatorIntervalConfig `json:"interval_aggregates"` // List of aggregation function processed at the end of an interval
-	DropMetrics       []string                             `json:"drop_metrics"`        // List of metric names to drop. For fine-grained dropping use drop_metrics_if
-	DropMetricsIf     []string                             `json:"drop_metrics_if"`     // List of evaluatable terms to drop metrics
-	RenameMetrics     map[string]string                    `json:"rename_metrics"`      // Map to rename metric name from key to value
-	IntervalStamp     bool                                 `json:"interval_timestamp"`  // Update timestamp periodically by ticker each interval?
-	NumCacheIntervals int                                  `json:"num_cache_intervals"` // Number of intervals of cached metrics for evaluation
-	MaxForward        int                                  `json:"max_forward"`         // Number of maximal forwarded metrics at one select
-	NormalizeUnits    bool                                 `json:"normalize_units"`     // Check unit meta flag and normalize it using cc-units
-	ChangeUnitPrefix  map[string]string                    `json:"change_unit_prefix"`  // Add prefix that should be applied to the metrics
+	HostnameTagName   string                               `json:"hostname_tag,omitempty"`        // Key name used when adding the hostname to a metric (default 'hostname')
+	AddTags           []metricRouterTagConfig              `json:"add_tags,omitempty"`            // List of tags that are added when the condition is met
+	DelTags           []metricRouterTagConfig              `json:"delete_tags,omitempty"`         // List of tags that are removed when the condition is met
+	IntervalAgg       []agg.MetricAggregatorIntervalConfig `json:"interval_aggregates,omitempty"` // List of aggregation function processed at the end of an interval
+	DropMetrics       []string                             `json:"drop_metrics,omitempty"`        // List of metric names to drop. For fine-grained dropping use drop_metrics_if
+	DropMetricsIf     []string                             `json:"drop_metrics_if,omitempty"`     // List of evaluatable terms to drop metrics
+	RenameMetrics     map[string]string                    `json:"rename_metrics,omitempty"`      // Map to rename metric name from key to value
+	IntervalStamp     bool                                 `json:"interval_timestamp,omitempty"`  // Update timestamp periodically by ticker each interval?
+	NumCacheIntervals int                                  `json:"num_cache_intervals,omitempty"` // Number of intervals of cached metrics for evaluation
+	MaxForward        int                                  `json:"max_forward,omitempty"`         // Number of maximal forwarded metrics at one select
+	NormalizeUnits    bool                                 `json:"normalize_units,omitempty"`     // Check unit meta flag and normalize it using cc-units
+	ChangeUnitPrefix  map[string]string                    `json:"change_unit_prefix,omitempty"`  // Add prefix that should be applied to the metrics
 	MessageProcessor  json.RawMessage                      `json:"process_messages,omitempty"`
 }

@@ -228,25 +228,11 @@ func (r *metricRouter) DoAddTags(point lp.CCMessage) {
 func (r *metricRouter) Start() {
 	// start timer if configured
 	r.timestamp = time.Now()
-	timeChan := make(chan time.Time, 1)
+	timeChan := make(chan time.Time)
 	if r.config.IntervalStamp {
 		r.ticker.AddChannel(timeChan)
 	}

-	// Drain a pending tick before stamping new metrics, so a new interval's
-	// metrics never carry the previous interval's timestamp
-	updateTimestamp := func() {
-		if !r.config.IntervalStamp {
-			return
-		}
-		select {
-		case timestamp := <-timeChan:
-			r.timestamp = timestamp
-			cclog.ComponentDebug("MetricRouter", "Update timestamp", r.timestamp.UnixNano())
-		default:
-		}
-	}
-
 	// Router manager is done
 	done := func() {
 		close(r.done)
@@ -311,20 +297,15 @@ func (r *metricRouter) Start() {

 			case timestamp := <-timeChan:
 				r.timestamp = timestamp
-				cclog.ComponentDebug("MetricRouter", "Update timestamp", r.timestamp.UnixNano())
-				if len(r.coll_input) == cap(r.coll_input) {
-					cclog.ComponentWarn("MetricRouter", "collector input channel full at tick, sinks may be too slow")
-				}
+				cclog.ComponentDebugf("MetricRouter", "Update timestamp %d", r.timestamp.UnixNano())

 			case p := <-r.coll_input:
-				updateTimestamp()
 				coll_forward(p)
 				for i := 0; len(r.coll_input) > 0 && i < (r.maxForward-1); i++ {
 					coll_forward(<-r.coll_input)
 				}

 			case p := <-r.recv_input:
-				updateTimestamp()
 				recv_forward(p)
 				for i := 0; len(r.recv_input) > 0 && i < (r.maxForward-1); i++ {
 					recv_forward(<-r.recv_input)
@@ -1,99 +0,0 @@
-package metricRouter
-
-import (
-	"encoding/json"
-	"fmt"
-	"sync"
-	"testing"
-	"time"
-
-	lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
-)
-
-// Fake ticker that delivers ticks on demand
-type fakeTicker struct {
-	channels []chan time.Time
-}
-
-func (t *fakeTicker) Init(duration time.Duration) {}
-
-func (t *fakeTicker) AddChannel(c chan time.Time) {
-	t.channels = append(t.channels, c)
-}
-
-func (t *fakeTicker) Close() {}
-
-func (t *fakeTicker) tick(ts time.Time) {
-	for _, c := range t.channels {
-		c <- ts
-	}
-}
-
-func genMessages(t *testing.T, num int) []lp.CCMessage {
-	t.Helper()
-	msgs := make([]lp.CCMessage, 0, num)
-	tags := map[string]string{"type": "node"}
-	for i := range num {
-		m, err := lp.NewMetric(fmt.Sprintf("testmetric%d", i), tags, nil, 42.0, time.Unix(1, 0))
-		if err != nil {
-			t.Fatalf("failed to create message: %s", err.Error())
-		}
-		msgs = append(msgs, m)
-	}
-	return msgs
-}
-
-// With interval_timestamp enabled, all metrics forwarded after a tick must
-// carry that tick's timestamp, never the previous interval's
-func TestIntervalTimestamp(t *testing.T) {
-	ticker := &fakeTicker{}
-	var wg sync.WaitGroup
-	r, err := New(ticker, &wg, json.RawMessage(`{"interval_timestamp": true}`))
-	if err != nil {
-		t.Fatalf("failed to setup metric router: %s", err.Error())
-	}
-
-	coll := make(chan lp.CCMessage, 100)
-	out := make(chan lp.CCMessage, 100)
-	r.AddCollectorInput(coll)
-	r.AddOutput(out)
-	r.Start()
-
-	receiveAll := func(num int) []lp.CCMessage {
-		received := make([]lp.CCMessage, 0, num)
-		for len(received) < num {
-			select {
-			case m := <-out:
-				received = append(received, m)
-			case <-time.After(5 * time.Second):
-				t.Fatalf("received only %d of %d messages", len(received), num)
-			}
-		}
-		return received
-	}
-
-	for interval, tickTime := range []time.Time{time.Unix(1000, 0), time.Unix(1010, 0)} {
-		ticker.tick(tickTime)
-		msgs := genMessages(t, 20)
-		for _, m := range msgs {
-			coll <- m
-		}
-		for i, m := range receiveAll(len(msgs)) {
-			if !m.Time().Equal(tickTime) {
-				t.Errorf("interval %d message %d: got timestamp %v, want %v", interval, i, m.Time(), tickTime)
-			}
-		}
-	}
-
-	closed := make(chan struct{})
-	go func() {
-		r.Close()
-		close(closed)
-	}()
-	select {
-	case <-closed:
-	case <-time.After(5 * time.Second):
-		t.Fatal("Close() did not terminate")
-	}
-	wg.Wait()
-}
@@ -30,8 +30,8 @@ Afterwards, you can add channels:

 ```golang
 t := MultiChanTicker(duration)
-c1 := make(chan time.Time, 1)
-c2 := make(chan time.Time, 1)
+c1 := make(chan time.Time)
+c2 := make(chan time.Time)
 t.AddChannel(c1)
 t.AddChannel(c2)

@@ -46,5 +46,3 @@ for {
 ```

 The result should be the same `time.Time` output in both channels, notified "simultaneously".
-
-Ticks are delivered with a non-blocking send: a consumer that has not yet read the previous tick does not stall the ticker (which would silently drop `time.Ticker` fires for all consumers); instead, the tick for that consumer is skipped and a warning is logged. Register buffered channels (capacity 1) so a consumer that is briefly busy at tick time does not lose the tick.
@@ -8,8 +8,6 @@
 package multiChanTicker

 import (
-	"fmt"
-	"sync"
 	"time"

 	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
@@ -17,7 +15,6 @@ import (

 type multiChanTicker struct {
 	ticker   *time.Ticker
-	mutex    sync.Mutex // protects channels, which is appended to while the tick goroutine iterates it
 	channels []chan time.Time
 	done     chan bool
 }
@@ -43,30 +40,21 @@ func (t *multiChanTicker) Init(duration time.Duration) {
 				return
 			case ts := <-t.ticker.C:
 				cclog.ComponentDebug("MultiChanTicker", "Tick", ts)
-				t.mutex.Lock()
-				for i, c := range t.channels {
-					// Non-blocking send: a consumer that has not yet read the
-					// previous tick must not stall the ticker, otherwise
-					// time.Ticker silently drops fires for ALL consumers
+				for _, c := range t.channels {
 					select {
+					case <-t.done:
+						done()
+						return
 					case c <- ts:
-					default:
-						cclog.ComponentWarn("MultiChanTicker", fmt.Sprintf("consumer %d did not read previous tick, dropping tick %v", i, ts))
 					}
 				}
-				t.mutex.Unlock()
 			}
 		}
 	}()
 }

 func (t *multiChanTicker) AddChannel(channel chan time.Time) {
-	if cap(channel) == 0 {
-		cclog.ComponentWarn("MultiChanTicker", "unbuffered channel registered, ticks may be dropped if the consumer is not ready")
-	}
-	t.mutex.Lock()
 	t.channels = append(t.channels, channel)
-	t.mutex.Unlock()
 }

 func (t *multiChanTicker) Close() {
@@ -1,51 +0,0 @@
-package multiChanTicker
-
-import (
-	"testing"
-	"time"
-)
-
-// A consumer that never reads its channel must not stall the ticker
-// or starve the other consumers
-func TestStalledConsumerDoesNotStarveOthers(t *testing.T) {
-	stalled := make(chan time.Time, 1) // never read
-	fast := make(chan time.Time, 1)
-
-	ticker := NewTicker(10 * time.Millisecond)
-	defer ticker.Close()
-	ticker.AddChannel(stalled)
-	ticker.AddChannel(fast)
-
-	received := 0
-	deadline := time.After(5 * time.Second)
-	for received < 5 {
-		select {
-		case <-fast:
-			received++
-		case <-deadline:
-			t.Fatalf("received only %d ticks while another consumer stalled", received)
-		}
-	}
-}
-
-// Close() must return promptly even if no consumer reads its channel
-func TestCloseWithStalledConsumer(t *testing.T) {
-	stalled := make(chan time.Time, 1) // never read
-
-	ticker := NewTicker(10 * time.Millisecond)
-	ticker.AddChannel(stalled)
-
-	// Let some ticks fire and be dropped
-	time.Sleep(50 * time.Millisecond)
-
-	closed := make(chan struct{})
-	go func() {
-		ticker.Close()
-		close(closed)
-	}()
-	select {
-	case <-closed:
-	case <-time.After(5 * time.Second):
-		t.Fatal("Close() blocked with a stalled consumer")
-	}
-}
Author	SHA1	Message	Date
Thomas Roehl	c9dfca622f	Update cclog calls in latest IB Collector	2026-06-08 14:58:29 +02:00
Thomas Roehl	3b0638e815	Merge branch 'main' into cclog_update	2026-06-08 14:57:31 +02:00
Thomas Roehl	037b4f1526	Update cclog calls	2026-06-08 14:52:24 +02:00
Thomas Roehl	5d55ee7a77	Fix lint errors	2026-06-02 16:35:11 +02:00
Thomas Roehl	5938368a76	Add collector to get Nvidia GPM metrics	2026-06-02 13:52:44 +02:00