Fix metric timeseries gaps on nodes with many cores

On nodes with >300 cores, one interval's burst of per-hwthread metrics
overran the fixed 200-slot channels. With blocking sends at every hop,
sink back-pressure propagated to the collectors, the collection round
exceeded the interval, and time.Ticker silently dropped the missed
ticks - whole intervals were skipped without any log message.

- multiChanTicker: deliver ticks non-blockingly and warn when a
  consumer misses a tick instead of stalling all consumers; guard the
  channel list with a mutex (data race with AddChannel)
- collectorManager: run the collection round detached from the tick
  loop, skip-and-warn when a round is still running, log per-collector
  and per-round durations at debug level, close serial collectors on
  shutdown
- metricRouter: buffer the interval timestamp channel and drain it
  before stamping, so metrics never carry the previous interval's
  timestamp; warn when the collector input channel is full at tick time
- main: scale the inter-manager channels to max(200, 24*NumCPU),
  overridable with the new optional channel_buffer_size config option
- add first unit tests for ticker, collector manager and router

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.gitignore

@@ -17,3 +17,7 @@
 
 # Local copy of LIKWID headers
 /collectors/likwid
+
+# Local development workspace (build against a local cc-lib checkout)
+go.work
+go.work.sum

README.md

@@ -41,6 +41,8 @@ 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)

cc-metric-collector.go

@@ -13,6 +13,7 @@ import (
 	"flag"
 	"os"
 	"os/signal"
+	"runtime"
 	"sync"
 	"syscall"
 	"time"
@@ -29,8 +30,9 @@ import (
 )
 
 type CentralConfigFile struct {
-	Interval string `json:"interval"`
-	Duration string `json:"duration"`
+	Interval          string `json:"interval"`
+	Duration          string `json:"duration"`
+	ChannelBufferSize int    `json:"channel_buffer_size,omitempty"`
 }
 
 type RuntimeConfig struct {
@@ -158,6 +160,14 @@ 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")
@@ -194,7 +204,7 @@ func mainFunc() int {
 	}
 
 	// Connect metric router to sink manager
-	RouterToSinksChannel := make(chan lp.CCMessage, 200)
+	RouterToSinksChannel := make(chan lp.CCMessage, chanSize)
 	rcfg.SinkManager.AddInput(RouterToSinksChannel)
 	rcfg.MetricRouter.AddOutput(RouterToSinksChannel)
 
@@ -206,7 +216,7 @@ func mainFunc() int {
 	}
 
 	// Connect collector manager to metric router
-	CollectToRouterChannel := make(chan lp.CCMessage, 200)
+	CollectToRouterChannel := make(chan lp.CCMessage, chanSize)
 	rcfg.CollectManager.AddOutput(CollectToRouterChannel)
 	rcfg.MetricRouter.AddCollectorInput(CollectToRouterChannel)
 
@@ -220,7 +230,7 @@ func mainFunc() int {
 		}
 
 		// Connect receive manager to metric router
-		ReceiveToRouterChannel := make(chan lp.CCMessage, 200)
+		ReceiveToRouterChannel := make(chan lp.CCMessage, chanSize)
 		rcfg.ReceiveManager.AddOutput(ReceiveToRouterChannel)
 		rcfg.MetricRouter.AddReceiverInput(ReceiveToRouterChannel)
 		use_recv = true

collectors/collectorManager.go

@@ -12,6 +12,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
@@ -21,49 +22,52 @@ 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),
+	"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),
 }
 
 // Metric collector manager data structure
 type collectorManager struct {
-	collectors   []MetricCollector          // List of metric collectors to read in parallel
-	serial       []MetricCollector          // List of metric collectors to read serially
-	output       chan lp.CCMessage          // Output channels
-	done         chan bool                  // channel to finish / stop metric collector manager
-	ticker       mct.MultiChanTicker        // periodically ticking once each interval
-	duration     time.Duration              // duration (for metrics that measure over a given duration)
-	wg           *sync.WaitGroup            // wait group for all goroutines in cc-metric-collector
-	config       map[string]json.RawMessage // json encoded config for collector manager
-	collector_wg sync.WaitGroup             // internally used wait group for the parallel reading of collector
-	parallel_run bool                       // Flag whether the collectors are currently read in parallel
+	collectors    []MetricCollector          // List of metric collectors to read in parallel
+	serial        []MetricCollector          // List of metric collectors to read serially
+	output        chan lp.CCMessage          // Output channels
+	done          chan bool                  // channel to finish / stop metric collector manager
+	ticker        mct.MultiChanTicker        // periodically ticking once each interval
+	duration      time.Duration              // duration (for metrics that measure over a given duration)
+	wg            *sync.WaitGroup            // wait group for all goroutines in cc-metric-collector
+	config        map[string]json.RawMessage // json encoded config for collector manager
+	collector_wg  sync.WaitGroup             // internally used wait group for the parallel reading of collector
+	round_wg      sync.WaitGroup             // wait group for the currently running collection round
+	round_running atomic.Bool                // Flag whether a collection round is currently running
 }
 
 // Metric collector manager access functions
@@ -119,26 +123,54 @@ 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)
+	tick := make(chan time.Time, 1)
 	cm.ticker.AddChannel(tick)
 
 	cm.wg.Go(func() {
 		// Collector manager is done
 		done := func() {
-			// close all metric collectors
-			if cm.parallel_run {
-				cm.collector_wg.Wait()
-				cm.parallel_run = false
-			}
+			// wait for a still running collection round, then close all metric collectors
+			cm.round_wg.Wait()
 			for _, c := range cm.collectors {
 				c.Close()
 			}
+			for _, c := range cm.serial {
+				c.Close()
+			}
 			close(cm.done)
 			cclog.ComponentDebug("CollectorManager", "DONE")
 		}
 
+		var roundStart time.Time
+
 		// Wait for done signal or timer event
 		for {
 			select {
@@ -146,37 +178,20 @@ func (cm *collectorManager) Start() {
 				done()
 				return
 			case t := <-tick:
-				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.ComponentDebug("CollectorManager", c.Name(), t)
-						cm.collector_wg.Add(1)
-						go func(myc MetricCollector) {
-							myc.Read(cm.duration, cm.output)
-							cm.collector_wg.Done()
-						}(c)
-					}
-				}
-				cm.collector_wg.Wait()
-				cm.parallel_run = false
-				for _, c := range cm.serial {
-					// Wait for done signal or execute the collector
-					select {
-					case <-cm.done:
-						done()
-						return
-					default:
-						// Read metrics from collector c
-						cclog.ComponentDebug("CollectorManager", c.Name(), t)
-						c.Read(cm.duration, cm.output)
-					}
+				// 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.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)
+				}()
 			}
 		}
 	})

collectors/collectorManager_test.go

@@ -0,0 +1,121 @@
+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()
+}

collectors/cpufreqCpuinfoMetric.go

@@ -116,6 +116,9 @@ 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)
@@ -179,6 +182,9 @@ 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() {

collectors/cpustatMetric.go

@@ -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: %w", m.name, err)
+		return fmt.Errorf("%s Init(): setup() call failed: %s", m.name, err.Error())
 	}
 	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: %w", m.name, err)
+			return fmt.Errorf("%s Init(): Error decoding JSON config: %s", m.name, err.Error())
 		}
 	}
 	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': %w", m.name, CPUSTATFILE, err)
+		return fmt.Errorf("%s Init(): Failed to open file '%s': %s", m.name, CPUSTATFILE, err.Error())
 	}
 
 	// Pre-generate tags for all CPUs
@@ -117,10 +117,13 @@ 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': %w", m.name, CPUSTATFILE, err)
+		return fmt.Errorf("%s Init(): Failed to close file '%s': %s", m.name, CPUSTATFILE, err.Error())
 	}
 
 	m.lastTimestamp = time.Now()
@@ -173,13 +176,13 @@ func (m *CpustatCollector) Read(interval time.Duration, output chan lp.CCMessage
 	if err != nil {
 		cclog.ComponentError(
 			m.name,
-			fmt.Sprintf("Read(): Failed to open file '%s': %v", CPUSTATFILE, err))
+			fmt.Sprintf("Read(): Failed to open file '%s': %v", CPUSTATFILE, err.Error()))
 	}
 	defer func() {
 		if err := file.Close(); err != nil {
 			cclog.ComponentError(
 				m.name,
-				fmt.Sprintf("Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err))
+				fmt.Sprintf("Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err.Error()))
 		}
 	}()
 
@@ -194,6 +197,9 @@ 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 {

collectors/diskstatMetric.go

@@ -147,6 +147,11 @@ 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 {

collectors/infinibandMetric.go

@@ -23,20 +23,29 @@ import (
 	"golang.org/x/sys/unix"
 )
 
-const IB_BASEPATH = "/sys/class/infiniband/"
+// See: https://www.kernel.org/doc/Documentation/ABI/stable/sysfs-class-infiniband
+const (
+	ibBasePath     = "/sys/class/infiniband/"
+	ibDataUnit     = "bytes"
+	ibDataRateUnit = ibDataUnit + "/sec"
+	ibPkgUnit      = "packets"
+	ibPkgRateUnit  = ibPkgUnit + "/sec"
+)
 
 type InfinibandCollectorMetric struct {
-	name             string
-	path             string
-	unit             string
-	scale            int64
-	addToIBTotal     bool
-	addToIBTotalPkgs bool
-	lastState        int64
+	name               string
+	path               string
+	unit               string
+	unitRates          string
+	scaleByFourLanes   bool
+	addToIBTotal       bool
+	addToIBTotalPkgs   bool
+	lastState          uint64
+	lastStateAvailable bool
 }
 
 type InfinibandCollectorInfo struct {
-	LID              string                      // IB local Identifier (LID)
+	lid              string                      // IB local Identifier (LID)
 	device           string                      // IB device
 	port             string                      // IB device port
 	portCounterFiles []InfinibandCollectorMetric // mapping counter name -> InfinibandCollectorMetric
@@ -56,7 +65,7 @@ type InfinibandCollector struct {
 	lastTimestamp time.Time // Store time stamp of last tick to derive bandwidths
 }
 
-// Init initializes the Infiniband collector by walking through files below IB_BASEPATH
+// Init initializes the Infiniband collector by walking through files below ibBasePath
 func (m *InfinibandCollector) Init(config json.RawMessage) error {
 	// Check if already initialized
 	if m.init {
@@ -87,7 +96,7 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
 	}
 
 	// Loop for all InfiniBand directories
-	globPattern := filepath.Join(IB_BASEPATH, "*", "ports", "*")
+	globPattern := filepath.Join(ibBasePath, "*", "ports", "*")
 	ibDirs, err := filepath.Glob(globPattern)
 	if err != nil {
 		return fmt.Errorf("%s Init(): unable to glob files with pattern %s: %w", m.name, globPattern, err)
@@ -122,36 +131,42 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
 		countersDir := filepath.Join(path, "counters")
 		portCounterFiles := []InfinibandCollectorMetric{
 			{
-				name:         "ib_recv",
-				path:         filepath.Join(countersDir, "port_rcv_data"),
-				unit:         "bytes",
-				scale:        4,
-				addToIBTotal: true,
-				lastState:    -1,
+				// Total number of data octets, divided by 4 (lanes), received on all VLs.
+				// This is 64 bit counter
+				name:             "ib_recv",
+				path:             filepath.Join(countersDir, "port_rcv_data"),
+				unit:             ibDataUnit,
+				unitRates:        ibDataRateUnit,
+				scaleByFourLanes: true,
+				addToIBTotal:     true,
 			},
 			{
-				name:         "ib_xmit",
-				path:         filepath.Join(countersDir, "port_xmit_data"),
-				unit:         "bytes",
-				scale:        4,
-				addToIBTotal: true,
-				lastState:    -1,
+				// Total number of data octets, divided by 4 (lanes), transmitted on all VLs.
+				// This is 64 bit counter
+				name:             "ib_xmit",
+				path:             filepath.Join(countersDir, "port_xmit_data"),
+				unit:             ibDataUnit,
+				unitRates:        ibDataRateUnit,
+				scaleByFourLanes: true,
+				addToIBTotal:     true,
 			},
 			{
+				// Total number of packets received on all VLs from this port (this may include packets containing Errors.
+				// This is 64 bit counter.
 				name:             "ib_recv_pkts",
 				path:             filepath.Join(countersDir, "port_rcv_packets"),
-				unit:             "packets",
-				scale:            1,
+				unit:             ibPkgUnit,
+				unitRates:        ibPkgRateUnit,
 				addToIBTotalPkgs: true,
-				lastState:        -1,
 			},
 			{
+				// Total number of packets transmitted on all VLs from this port. This may include packets with errors.
+				// This is 64 bit counter.
 				name:             "ib_xmit_pkts",
 				path:             filepath.Join(countersDir, "port_xmit_packets"),
-				unit:             "packets",
-				scale:            1,
+				unit:             ibPkgUnit,
+				unitRates:        ibPkgRateUnit,
 				addToIBTotalPkgs: true,
-				lastState:        -1,
 			},
 		}
 		for _, counter := range portCounterFiles {
@@ -163,7 +178,7 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
 
 		m.info = append(m.info,
 			InfinibandCollectorInfo{
-				LID:              LID,
+				lid:              LID,
 				device:           device,
 				port:             port,
 				portCounterFiles: portCounterFiles,
@@ -184,7 +199,7 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
 	return nil
 }
 
-// Read reads Infiniband counter files below IB_BASEPATH
+// Read reads Infiniband counter files below ibBasePath
 func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMessage) {
 	// Check if already initialized
 	if !m.init {
@@ -201,9 +216,9 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMess
 	for i := range m.info {
 		info := &m.info[i]
 
-		var ib_total, ib_total_last_state,
-			ib_total_pkts, ib_total_pkts_last_state int64
-		var ib_total_last_state_available, ib_total_pkts_last_state_available bool
+		var ibTotal, ibTotalPkts uint64      // sum of xmit and recv counters
+		var ibTotalBw, ibTotalPktsBw float64 // sum of xmit and recv rates
+		var ibTotalBwAvailable, ibTotalPktsBwAvailable bool
 		for i := range info.portCounterFiles {
 			counterDef := &info.portCounterFiles[i]
 
@@ -213,24 +228,30 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMess
 				cclog.ComponentError(
 					m.name,
 					fmt.Sprintf("Read(): Failed to read from file '%s': %v", counterDef.path, err))
+				// Current counter can not be saved as last state
+				counterDef.lastStateAvailable = false
 				continue
 			}
 			data := strings.TrimSpace(string(line))
 
-			// convert counter to int64
-			v, err := strconv.ParseInt(data, 10, 64)
+			// convert counter to uint64
+			vRawCounter, err := strconv.ParseUint(data, 10, 64)
 			if err != nil {
 				cclog.ComponentError(
 					m.name,
-					fmt.Sprintf("Read(): Failed to convert Infininiband metrice %s='%s' to int64: %v", counterDef.name, data, err))
+					fmt.Sprintf("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
 			}
-			// Scale raw value
-			v *= counterDef.scale
+			vScaledCounter := vRawCounter
+			if counterDef.scaleByFourLanes {
+				vScaledCounter *= uint64(4)
+			}
 
 			// Send absolut values
 			if m.config.SendAbsoluteValues {
-				if y, err := lp.NewMetric(counterDef.name, info.tagSet, m.meta, v, now); err == nil {
+				if y, err := lp.NewMetric(counterDef.name, info.tagSet, m.meta, vScaledCounter, now); err == nil {
 					y.AddMeta("unit", counterDef.unit)
 					output <- y
 				}
@@ -238,63 +259,72 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMess
 
 			// Send derived values
 			if m.config.SendDerivedValues {
-				if counterDef.lastState >= 0 {
-					rate := float64((v - counterDef.lastState)) / timeDiff
+				if counterDef.lastStateAvailable {
+					var rate float64
+					// uint64 subtraction handles wraparound automatically
+					// in case vRawCounter < counterDef.lastState we would compute:
+					// math.MaxUint64 - lastState + vRawCounter + 1
+					// = (2^64 - 1) - lastState + vRawCounter + 1
+					// = 2^64 - lastState + vRawCounter
+					// ≡ vRawCounter - lastState (mod 2^64)
+					rate = float64(vRawCounter-counterDef.lastState) / timeDiff
+					if counterDef.scaleByFourLanes {
+						rate *= float64(4)
+					}
 					if y, err := lp.NewMetric(counterDef.name+"_bw", info.tagSet, m.meta, rate, now); err == nil {
-						y.AddMeta("unit", counterDef.unit+"/sec")
+						y.AddMeta("unit", counterDef.unitRates)
 						output <- y
 					}
 
-					// Sum up total values of last state
+					// Sum up rates for total rates
 					if m.config.SendTotalValues {
 						switch {
 						case counterDef.addToIBTotal:
-							ib_total_last_state += counterDef.lastState
-							ib_total_last_state_available = true
+							ibTotalBw += rate
+							ibTotalBwAvailable = true
 						case counterDef.addToIBTotalPkgs:
-							ib_total_pkts_last_state += counterDef.lastState
-							ib_total_pkts_last_state_available = true
+							ibTotalPktsBw += rate
+							ibTotalPktsBwAvailable = true
 						}
 					}
 				}
-				counterDef.lastState = v
+				counterDef.lastState = vRawCounter
+				counterDef.lastStateAvailable = true
 			}
 
 			// Sum up total values
 			if m.config.SendTotalValues {
 				switch {
 				case counterDef.addToIBTotal:
-					ib_total += v
+					ibTotal += vScaledCounter
 				case counterDef.addToIBTotalPkgs:
-					ib_total_pkts += v
+					ibTotalPkts += vScaledCounter
 				}
 			}
 		}
 
 		// Send total values
 		if m.config.SendTotalValues {
-			if y, err := lp.NewMetric("ib_total", info.tagSet, m.meta, ib_total, now); err == nil {
-				y.AddMeta("unit", "bytes")
+			if y, err := lp.NewMetric("ib_total", info.tagSet, m.meta, ibTotal, now); err == nil {
+				y.AddMeta("unit", ibDataUnit)
 				output <- y
 			}
 
-			if y, err := lp.NewMetric("ib_total_pkts", info.tagSet, m.meta, ib_total_pkts, now); err == nil {
-				y.AddMeta("unit", "packets")
+			if y, err := lp.NewMetric("ib_total_pkts", info.tagSet, m.meta, ibTotalPkts, now); err == nil {
+				y.AddMeta("unit", ibPkgUnit)
 				output <- y
 			}
 
-			if m.config.SendDerivedValues && ib_total_last_state_available {
-				rate := float64((ib_total - ib_total_last_state)) / timeDiff
-				if y, err := lp.NewMetric("ib_total_bw", info.tagSet, m.meta, rate, now); err == nil {
-					y.AddMeta("unit", "bytes/sec")
+			if m.config.SendDerivedValues && ibTotalBwAvailable {
+				if y, err := lp.NewMetric("ib_total_bw", info.tagSet, m.meta, ibTotalBw, now); err == nil {
+					y.AddMeta("unit", ibDataRateUnit)
 					output <- y
 				}
 			}
 
-			if m.config.SendDerivedValues && ib_total_pkts_last_state_available {
-				rate := float64((ib_total_pkts - ib_total_pkts_last_state)) / timeDiff
-				if y, err := lp.NewMetric("ib_total_pkts_bw", info.tagSet, m.meta, rate, now); err == nil {
-					y.AddMeta("unit", "packets/sec")
+			if m.config.SendDerivedValues && ibTotalPktsBwAvailable {
+				if y, err := lp.NewMetric("ib_total_pkts_bw", info.tagSet, m.meta, ibTotalPktsBw, now); err == nil {
+					y.AddMeta("unit", ibPkgRateUnit)
 					output <- y
 				}
 			}

collectors/iostatMetric.go

@@ -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: %w", m.name, err)
+		return fmt.Errorf("%s Init(): setup() call failed: %s", m.name, err.Error())
 	}
 	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)
+			return fmt.Errorf("%s Init(): Error decoding JSON config: %s", m.name, err.Error())
 		}
 	}
 	// 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\": %w", m.name, IOSTATFILE, err)
+		return fmt.Errorf("%s Init(): Failed to open file \"%s\": %s", m.name, IOSTATFILE, err.Error())
 	}
 
 	scanner := bufio.NewScanner(file)
@@ -130,8 +130,11 @@ 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\": %w", m.name, IOSTATFILE, err)
+		return fmt.Errorf("%s Init(): Failed to close file \"%s\": %s", m.name, IOSTATFILE, err.Error())
 	}
 
 	m.init = true
@@ -147,14 +150,14 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMessage)
 	if err != nil {
 		cclog.ComponentError(
 			m.name,
-			fmt.Sprintf("Read(): Failed to open file '%s': %v", IOSTATFILE, err))
+			fmt.Sprintf("Read(): Failed to open file '%s': %s", IOSTATFILE, err.Error()))
 		return
 	}
 	defer func() {
 		if err := file.Close(); err != nil {
 			cclog.ComponentError(
 				m.name,
-				fmt.Sprintf("Read(): Failed to close file '%s': %v", IOSTATFILE, err))
+				fmt.Sprintf("Read(): Failed to close file '%s': %s", IOSTATFILE, err.Error()))
 		}
 	}()
 
@@ -198,6 +201,11 @@ 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() {

collectors/ipmiMetric.go

@@ -28,13 +28,15 @@ type IpmiCollector struct {
 	metricCollector
 
 	config struct {
-		IpmitoolPath    string `json:"ipmitool_path"`
-		IpmisensorsPath string `json:"ipmisensors_path"`
-		Sudo            bool   `json:"use_sudo"`
+		IpmitoolPath    string   `json:"ipmitool_path"`
+		IpmisensorsPath string   `json:"ipmisensors_path"`
+		Sudo            bool     `json:"use_sudo"`
+		IncludeMetrics  []string `json:"include_metrics"`
 	}
 
-	ipmitool    string
-	ipmisensors string
+	ipmitool       string
+	ipmisensors    string
+	includeMetrics map[string]bool
 }
 
 func (m *IpmiCollector) Init(config json.RawMessage) error {
@@ -64,6 +66,15 @@ 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
 
@@ -145,6 +156,11 @@ 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":
@@ -165,6 +181,9 @@ 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 {
@@ -209,6 +228,11 @@ 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)

collectors/ipmiMetric.md

@@ -15,7 +15,8 @@ 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
+    "use_sudo": true,
+    "include_metrics" : []
   }
 ```
 
@@ -36,3 +37,5 @@ 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. 

collectors/lenovoDensePower.go

@@ -0,0 +1,170 @@
+// 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
+}

collectors/lenovoDensePower.md

@@ -0,0 +1,51 @@
+<!--
+---
+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
+```

collectors/likwidMetric.go

@@ -38,7 +38,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/thread"
+	"golang.design/x/runtime/thread"
 )
 
 const (

collectors/megwareEureka.go

@@ -0,0 +1,194 @@
+// 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
+}

collectors/megwareEureka.md

@@ -0,0 +1,44 @@
+<!--
+---
+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
+```

collectors/slurmCgroupMetric.go

@@ -7,6 +7,7 @@ import (
 	"os/exec"
 	"os/user"
 	"path/filepath"
+	"regexp"
 	"strconv"
 	"strings"
 	"time"
@@ -45,6 +46,37 @@ 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 == "" {
@@ -237,10 +269,9 @@ func (m *SlurmCgroupCollector) Read(interval time.Duration, output chan lp.CCMes
 		delete(m.cpuUsed, k)
 	}
 
-	globPattern := filepath.Join(m.cgroupBase, "job_*")
-	jobDirs, err := filepath.Glob(globPattern)
+	jobDirs, err := m.findSlurmJobDirs()
 	if err != nil {
-		cclog.ComponentError(m.name, "Error globbing job directories:", err.Error())
+		cclog.ComponentError(m.name, "Error reading job directories:", err.Error())
 		return
 	}
 

docs/configuration.md

@@ -19,6 +19,8 @@ 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:

go.mod

@@ -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.0-1
+	github.com/NVIDIA/go-nvml v0.13.2-0
 	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/thread v0.0.0-20210122121316-335e9adffdf1
+	golang.design/x/runtime v0.3.0
 	golang.org/x/sys v0.45.0
 )
 

go.sum

@@ -12,8 +12,9 @@ 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=
@@ -173,8 +174,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/thread v0.0.0-20210122121316-335e9adffdf1 h1:P7S/GeHBAFEZIYp0ePPs2kHXoazz8q2KsyxHyQVGCJg=
-golang.design/x/thread v0.0.0-20210122121316-335e9adffdf1/go.mod h1:9CWpnTUmlQkfdpdutA1nNf4iE5lAVt3QZOu0Z6hahBE=
+golang.design/x/runtime v0.3.0 h1:8bA+GQfO8A18JXJQlQA56pEc+Sgrbo4cmLzjzmJ58ZI=
+golang.design/x/runtime v0.3.0/go.mod h1:PjVKQjImLdSrvjIw4FmrnCPq9BGs5PMBHb8i6b2BN9I=
 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=
@@ -183,7 +184,6 @@ golang.org/x/mod v0.13.0 h1:I/DsJXRlw/8l/0c24sM9yb0T4z9liZTduXvdAWYiysY=
 golang.org/x/mod v0.13.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
 golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
 golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
-golang.org/x/sys v0.0.0-20210122093101-04d7465088b8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
 golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=

internal/metricAggregator/metricAggregator.go

@@ -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.ComponentDebug("MetricCache", "COLLECT", f.Name, "COND", f.Condition)
+		cclog.ComponentDebugf("MetricCache", "COLLECT %s COND '%s'", f.Name, 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.ComponentError("MetricCache", "COLLECT", f.Name, "COND", f.Condition, ":", err.Error())
+				cclog.ComponentErrorf("MetricCache", "COLLECT %s COND '%s' : %s", f.Name, 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.ComponentError("MetricCache", "COLLECT ADD VALUE", v, "FAILED")
+						cclog.ComponentErrorf("MetricCache", "COLLECT ADD VALUE %v FAILED", v)
 					}
 				}
 				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.ComponentDebug("MetricCache", "EVALUATE", f.Name, "METRICS", len_values, "CALC", f.Function)
+		cclog.ComponentDebugf("MetricCache", "EVALUATE %s METRICS %d CALC '%s'", f.Name, len_values, f.Function)
 
 		vars["metrics"] = matches
 		if len_values > 0 {
 			value, err := gval.Evaluate(f.Function, vars, c.language)
 			if err != nil {
-				cclog.ComponentError("MetricCache", "EVALUATE", f.Name, "METRICS", len_values, "CALC", f.Function, ":", err.Error())
+				cclog.ComponentErrorf("MetricCache", "EVALUATE %s METRICS %d CALC '%s': %s", f.Name, len_values, 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.ComponentError("MetricCache", "Gval returned invalid type", t, "skipping metric", f.Name)
+				cclog.ComponentErrorf("MetricCache", "Gval returned invalid type %s skipping metric %s", t, f.Name)
 			}
 			if err != nil {
-				cclog.ComponentError("MetricCache", "Cannot create metric from Gval result", value, ":", err.Error())
+				cclog.ComponentErrorf("MetricCache", "Cannot create metric from Gval result %v: %s", value, err.Error())
 			}
-			cclog.ComponentDebug("MetricCache", "SEND", m)
+			cclog.ComponentDebugf("MetricCache", "SEND %s", m.ToLineProtocol(nil))
 			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.ComponentError("MetricAggregator", "Cannot add aggregation, invalid if condition", newcond, ":", err.Error())
+		cclog.ComponentErrorf("MetricAggregator", "Cannot add aggregation, invalid if condition '%s': %s", newcond, err.Error())
 		return err
 	}
 	gvalFunc, err := gval.Full(metricCacheLanguage).NewEvaluable(newfunc)
 	if err != nil {
-		cclog.ComponentError("MetricAggregator", "Cannot add aggregation, invalid function condition", newfunc, ":", err.Error())
+		cclog.ComponentErrorf("MetricAggregator", "Cannot add aggregation, invalid function condition %s: %s", newfunc, err.Error())
 		return err
 	}
 	for _, agg := range c.functions {

internal/metricRouter/metricCache.go

@@ -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)
+	c.tickchan = make(chan time.Time, 1)
 	c.ticker.AddChannel(c.tickchan)
 	// Router cache is done
 	done := func() {

internal/metricRouter/metricRouter.go

@@ -228,11 +228,25 @@ func (r *metricRouter) DoAddTags(point lp.CCMessage) {
 func (r *metricRouter) Start() {
 	// start timer if configured
 	r.timestamp = time.Now()
-	timeChan := make(chan time.Time)
+	timeChan := make(chan time.Time, 1)
 	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)
@@ -298,14 +312,19 @@ 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")
+				}
 
 			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)

internal/metricRouter/metricRouter_test.go

@@ -0,0 +1,99 @@
+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()
+}

pkg/multiChanTicker/README.md

@@ -30,8 +30,8 @@ Afterwards, you can add channels:
 
 ```golang
 t := MultiChanTicker(duration)
-c1 := make(chan time.Time)
-c2 := make(chan time.Time)
+c1 := make(chan time.Time, 1)
+c2 := make(chan time.Time, 1)
 t.AddChannel(c1)
 t.AddChannel(c2)
 
@@ -46,3 +46,5 @@ 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.

pkg/multiChanTicker/multiChanTicker.go

@@ -8,6 +8,8 @@
 package multiChanTicker
 
 import (
+	"fmt"
+	"sync"
 	"time"
 
 	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
@@ -15,6 +17,7 @@ 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
 }
@@ -40,21 +43,30 @@ func (t *multiChanTicker) Init(duration time.Duration) {
 				return
 			case ts := <-t.ticker.C:
 				cclog.ComponentDebug("MultiChanTicker", "Tick", ts)
-				for _, c := range t.channels {
+				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
 					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() {

pkg/multiChanTicker/multiChanTicker_test.go

@@ -0,0 +1,51 @@
+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")
+	}
+}