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

@@ -33,7 +33,6 @@ There is a main configuration file with basic settings that point to the other c
   "receivers-file" : "receivers.json",
   "router-file" : "router.json",
   "main": {
-    "enable-markers": false,
     "interval": "10s",
     "duration": "1s"
   }
@@ -42,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,17 +30,16 @@ import (
 )
 
 type CentralConfigFile struct {
-	Interval      string `json:"interval"`
-	Duration      string `json:"duration"`
-	EnableMarkers bool   `json:"enable-markers"`
+	Interval          string `json:"interval"`
+	Duration          string `json:"duration"`
+	ChannelBufferSize int    `json:"channel_buffer_size,omitempty"`
 }
 
 type RuntimeConfig struct {
-	Interval      time.Duration
-	Duration      time.Duration
-	EnableMarkers bool
-	CliArgs       map[string]string
-	ConfigFile    CentralConfigFile
+	Interval   time.Duration
+	Duration   time.Duration
+	CliArgs    map[string]string
+	ConfigFile CentralConfigFile
 
 	MetricRouter    mr.MetricRouter
 	CollectManager  collectors.CollectorManager
@@ -159,7 +159,14 @@ func mainFunc() int {
 		cclog.Error("The interval should be greater than duration")
 		return 1
 	}
-	rcfg.EnableMarkers = rcfg.ConfigFile.EnableMarkers
+
+	// 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 {
@@ -197,19 +204,19 @@ 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)
 
 	// Create new collector manager
-	rcfg.CollectManager, err = collectors.New(rcfg.MultiChanTicker, rcfg.Duration, rcfg.EnableMarkers, &rcfg.Sync, collectorConf)
+	rcfg.CollectManager, err = collectors.New(rcfg.MultiChanTicker, rcfg.Duration, &rcfg.Sync, collectorConf)
 	if err != nil {
 		cclog.Error(err.Error())
 		return 1
 	}
 
 	// 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)
 
@@ -223,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"
@@ -65,13 +66,13 @@ type collectorManager struct {
 	wg            *sync.WaitGroup            // wait group for all goroutines in cc-metric-collector
 	config        map[string]json.RawMessage // json encoded config for collector manager
 	collector_wg  sync.WaitGroup             // internally used wait group for the parallel reading of collector
-	parallel_run  bool                       // Flag whether the collectors are currently read in parallel
-	enableMarkers bool                       // Send ccmc-{begin,end} metrics
+	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
 type CollectorManager interface {
-	Init(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) error
+	Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) error
 	AddOutput(output chan lp.CCMessage)
 	Start()
 	Close()
@@ -84,7 +85,7 @@ type CollectorManager interface {
 // * ticker (from variable ticker)
 // * configuration (read from config file in variable collectConfigFile)
 // Initialization is done for all configured collectors
-func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) error {
+func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) error {
 	cm.collectors = make([]MetricCollector, 0)
 	cm.serial = make([]MetricCollector, 0)
 	cm.output = nil
@@ -92,7 +93,6 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
 	cm.wg = wg
 	cm.ticker = ticker
 	cm.duration = duration
-	cm.enableMarkers = enableMarkers
 
 	d := json.NewDecoder(bytes.NewReader(collectConfig))
 	d.DisallowUnknownFields()
@@ -123,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 {
@@ -150,53 +178,20 @@ func (cm *collectorManager) Start() {
 				done()
 				return
 			case t := <-tick:
-				if cm.enableMarkers {
-					m, err := lp.NewMetric("ccmc-begin", map[string]string{"type": "node"}, nil, 0, time.Now())
-					if err != nil {
-						cclog.ComponentErrorf("CollectorManager", "Unable to create marker metric: %v", err)
-					} else {
-						cm.output <- m
-					}
-				}
-				cm.parallel_run = true
-				for _, c := range cm.collectors {
-					// Wait for done signal or execute the collector
-					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)
-					}
-				}
-				if cm.enableMarkers {
-					m, err := lp.NewMetric("ccmc-end", map[string]string{"type": "node"}, nil, 0, time.Now())
-					if err != nil {
-						cclog.ComponentErrorf("CollectorManager", "Unable to create marker metric: %v", err)
-					} else {
-						cm.output <- m
-					}
+				// 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)
+				}()
 			}
 		}
 	})
@@ -219,9 +214,9 @@ func (cm *collectorManager) Close() {
 }
 
 // New creates a new initialized metric collector manager
-func New(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) (CollectorManager, error) {
+func New(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) (CollectorManager, error) {
 	cm := new(collectorManager)
-	err := cm.Init(ticker, duration, enableMarkers, wg, collectConfig)
+	err := cm.Init(ticker, duration, wg, collectConfig)
 	if err != nil {
 		return nil, err
 	}

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/likwidMetric.go

@@ -46,7 +46,6 @@ const (
 	LIKWID_LIB_DL_FLAGS   = dl.RTLD_LAZY | dl.RTLD_GLOBAL
 	LIKWID_DEF_ACCESSMODE = "direct"
 	LIKWID_DEF_LOCKFILE   = "/var/run/likwid.lock"
-	LIKWID_DEF_WARMUP_DEL = "0s"
 )
 
 type LikwidCollectorMetricConfig struct {
@@ -84,7 +83,6 @@ type LikwidCollectorConfig struct {
 	DaemonPath     string                          `json:"accessdaemon_path,omitempty"`
 	LibraryPath    string                          `json:"liblikwid_path,omitempty"`
 	LockfilePath   string                          `json:"lockfile_path,omitempty"`
-	WarmupDelay    string                          `json:"warmup_delay"`
 }
 
 type LikwidCollector struct {
@@ -106,7 +104,6 @@ type LikwidCollector struct {
 	likwidGroups  map[C.int]LikwidEventsetConfig
 	lock          sync.Mutex
 	measureThread thread.Thread
-	warmupDelay   time.Duration
 }
 
 type LikwidMetric struct {
@@ -209,7 +206,6 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
 	m.config.AccessMode = LIKWID_DEF_ACCESSMODE
 	m.config.LibraryPath = LIKWID_LIB_NAME
 	m.config.LockfilePath = LIKWID_DEF_LOCKFILE
-	m.config.WarmupDelay = LIKWID_DEF_WARMUP_DEL
 	if len(config) > 0 {
 		d := json.NewDecoder(bytes.NewReader(config))
 		d.DisallowUnknownFields()
@@ -246,17 +242,6 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
 		m.cpu2tid[c] = i
 	}
 
-	if len(m.config.WarmupDelay) > 0 {
-		t, err := time.ParseDuration(m.config.WarmupDelay)
-		if err != nil {
-			return fmt.Errorf("%s Init(): Cannot parse WarmupDelay: %w", m.name, err)
-		}
-		if t < 0 {
-			return fmt.Errorf("%s Init(): WarmupDelay must not be negative", m.name)
-		}
-		m.warmupDelay = t
-	}
-
 	m.likwidGroups = make(map[C.int]LikwidEventsetConfig)
 
 	// This is for the global metrics computation test
@@ -510,27 +495,6 @@ func (m *LikwidCollector) takeMeasurement(evidx int, evset LikwidEventsetConfig,
 	if ret != 0 {
 		return true, fmt.Errorf("failed to start events '%s', error %d", evset.go_estr, ret)
 	}
-
-	// warmup measuring
-	if m.warmupDelay > 0 {
-		select {
-		case <-sigchan:
-			ret = -1
-		case e := <-watcher.Events:
-			if e.Op != fsnotify.Chmod {
-				ret = C.perfmon_readCounters()
-			}
-		default:
-			ret = C.perfmon_readCounters()
-		}
-		if ret != 0 {
-			return true, fmt.Errorf("failed to read events '%s', error %d", evset.go_estr, ret)
-		}
-
-		time.Sleep(m.warmupDelay)
-	}
-
-	// begin measuring
 	select {
 	case <-sigchan:
 		ret = -1
@@ -548,7 +512,7 @@ func (m *LikwidCollector) takeMeasurement(evidx int, evset LikwidEventsetConfig,
 	// Wait
 	time.Sleep(interval)
 
-	// end measuring
+	// Read counters
 	select {
 	case <-sigchan:
 		ret = -1

collectors/likwidMetric.md

@@ -63,7 +63,6 @@ Additional options:
 - `accessdaemon_path`: Folder of the accessDaemon `likwid-accessD` (like `/usr/local/sbin`)
 - `liblikwid_path`: Location of `liblikwid.so` including file name like `/usr/local/lib/liblikwid.so`
 - `lockfile_path`: Location of LIKWID's lock file if multiple tools should access the hardware counters. Default `/var/run/likwid.lock`
-- `warmup_delay`: Run an additional measurement of the specified length before the actual measurement. This can be used as a workaround for CPU starvation (during high load) in the measurement thread. Default is disabled (i.e. `0s`).
 
 ### Available metric types
 

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:

example-configs/config.json

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

go.mod

@@ -3,14 +3,14 @@ module github.com/ClusterCockpit/cc-metric-collector
 go 1.25.0
 
 require (
-	github.com/ClusterCockpit/cc-lib/v2 v2.13.0
+	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.4-0
+	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/runtime v0.3.0
-	golang.org/x/sys v0.47.0
+	golang.org/x/sys v0.45.0
 )
 
 require (
@@ -18,28 +18,29 @@ require (
 	github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
-	github.com/coder/websocket v1.8.15 // indirect
+	github.com/coder/websocket v1.8.14 // indirect
 	github.com/expr-lang/expr v1.17.8 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/gorilla/mux v1.8.1 // indirect
 	github.com/influxdata/influxdb-client-go/v2 v2.14.0 // indirect
 	github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf // indirect
-	github.com/klauspost/compress v1.19.2 // indirect
+	github.com/klauspost/compress v1.18.5 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
-	github.com/nats-io/nats.go v1.53.1 // indirect
-	github.com/nats-io/nkeys v0.4.16 // indirect
+	github.com/nats-io/nats.go v1.51.0 // indirect
+	github.com/nats-io/nkeys v0.4.15 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
-	github.com/oapi-codegen/runtime v1.7.0 // indirect
-	github.com/prometheus/client_golang v1.24.1 // indirect
+	github.com/oapi-codegen/runtime v1.4.0 // indirect
+	github.com/prometheus/client_golang v1.23.2 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
-	github.com/prometheus/common v0.70.1 // indirect
-	github.com/prometheus/procfs v0.21.1 // indirect
+	github.com/prometheus/common v0.67.5 // indirect
+	github.com/prometheus/procfs v0.20.1 // indirect
 	github.com/questdb/go-questdb-client/v4 v4.2.0 // indirect
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 // indirect
 	github.com/shopspring/decimal v1.4.0 // indirect
-	github.com/stmcginnis/gofish v0.24.0 // indirect
+	github.com/stmcginnis/gofish v0.21.6 // indirect
 	github.com/tklauser/numcpus v0.12.0 // indirect
-	golang.org/x/crypto v0.55.0 // indirect
-	golang.org/x/net v0.58.0 // indirect
-	google.golang.org/protobuf v1.36.12 // indirect
+	go.yaml.in/yaml/v2 v2.4.4 // indirect
+	golang.org/x/crypto v0.50.0 // indirect
+	golang.org/x/net v0.53.0 // indirect
+	google.golang.org/protobuf v1.36.11 // indirect
 )

go.sum

@@ -2,8 +2,8 @@ dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk=
 dario.cat/mergo v1.0.0/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
 github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 h1:L/gRVlceqvL25UVaW/CKtUDjefjrs0SPonmDGUVOYP0=
 github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E=
-github.com/ClusterCockpit/cc-lib/v2 v2.13.0 h1:Gev5Bl4xuj+u15oTHNBLgpQeGER56e1FVWuTdBrfWLk=
-github.com/ClusterCockpit/cc-lib/v2 v2.13.0/go.mod h1:F7SmuvPGDqZucdm4MdMK///4RePgzWhRP0cyWK1Cons=
+github.com/ClusterCockpit/cc-lib/v2 v2.12.0 h1:ZbGD68nDniuvzFjJCdyYawpCBrabdSyWOg5FFSyFbjQ=
+github.com/ClusterCockpit/cc-lib/v2 v2.12.0/go.mod h1:ml8xtcYa5WhPM7JDQ+M9/R9ZBxITCR/5xqGJ//GxXJI=
 github.com/ClusterCockpit/cc-line-protocol/v2 v2.4.0 h1:hIzxgTBWcmCIHtoDKDkSCsKCOCOwUC34sFsbD2wcW0Q=
 github.com/ClusterCockpit/cc-line-protocol/v2 v2.4.0/go.mod h1:y42qUu+YFmu5fdNuUAS4VbbIKxVjxCvbVqFdpdh8ahY=
 github.com/ClusterCockpit/go-rocm-smi v0.4.0 h1:3+bEPrSkjEJcOtt+qBUX48ugDVlOFaKUnXHTef2Ve2Q=
@@ -13,15 +13,15 @@ github.com/Microsoft/go-winio v0.6.1/go.mod h1:LRdKpFKfdobln8UmuiYcKPot9D2v6svN5
 github.com/Microsoft/hcsshim v0.11.4 h1:68vKo2VN8DE9AdN4tnkWnmdhqdbpUFM8OF3Airm7fz8=
 github.com/Microsoft/hcsshim v0.11.4/go.mod h1:smjE4dvqPX9Zldna+t5FG3rnoHhaB7QYxPRqGcpAD9w=
 github.com/NVIDIA/go-nvml v0.13.0-1/go.mod h1:+KNA7c7gIBH7SKSJ1ntlwkfN80zdx8ovl4hrK3LmPt4=
-github.com/NVIDIA/go-nvml v0.13.4-0 h1:o3jp9u2x1R9ShFE3v+Aesp55XOSIQFMJz/VGNUcJaNE=
-github.com/NVIDIA/go-nvml v0.13.4-0/go.mod h1:id63qwpoDWpFXwnwM6psDCSqW4BmNu6mWpr4YeQtPGo=
+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=
 github.com/PaesslerAG/jsonpath v0.1.0/go.mod h1:4BzmtoM/PI8fPO4aQGIusjGxGir2BzcV0grWtFzq1Y8=
 github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk=
-github.com/antithesishq/antithesis-sdk-go v0.7.2-default-no-op h1:p2zFsAzvhIpFya8AIOHIbWf7NGvO34QpLGclyf7nXj8=
-github.com/antithesishq/antithesis-sdk-go v0.7.2-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
+github.com/antithesishq/antithesis-sdk-go v0.6.0-default-no-op h1:kpBdlEPbRvff0mDD1gk7o9BhI16b9p5yYAXRlidpqJE=
+github.com/antithesishq/antithesis-sdk-go v0.6.0-default-no-op/go.mod h1:IUpT2DPAKh6i/YhSbt6Gl3v2yvUZjmKncl7U91fup7E=
 github.com/apapsch/go-jsonmerge/v2 v2.0.0 h1:axGnT1gRIfimI7gJifB699GoE/oq+F2MU7Dml6nw9rQ=
 github.com/apapsch/go-jsonmerge/v2 v2.0.0/go.mod h1:lvDnEdqiQrp0O42VQGgmlKpxL1AP2+08jFMw88y4klk=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
@@ -31,8 +31,8 @@ github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqy
 github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
-github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA=
-github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
+github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
+github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
 github.com/containerd/containerd v1.7.12 h1:+KQsnv4VnzyxWcfO9mlxxELaoztsDEjOuCMPAuPqgU0=
 github.com/containerd/containerd v1.7.12/go.mod h1:/5OMpE1p0ylxtEUGY8kuCYkDRzJm9NO1TFMWjUpdevk=
 github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I=
@@ -40,6 +40,7 @@ github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3
 github.com/cpuguy83/dockercfg v0.3.1 h1:/FpZ+JaygUR/lZP2NlFI2DVfrOEMAIKP5wWEJdoYe9E=
 github.com/cpuguy83/dockercfg v0.3.1/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/docker/distribution v2.8.2+incompatible h1:T3de5rq0dB1j30rp0sA2rER+m322EBzniBPB6ZIzuh8=
 github.com/docker/distribution v2.8.2+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
@@ -76,10 +77,10 @@ github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf/go.mod h1
 github.com/influxdata/line-protocol-corpus v0.0.0-20210922080147-aa28ccfb8937 h1:MHJNQ+p99hFATQm6ORoLmpUCF7ovjwEFshs/NHzAbig=
 github.com/influxdata/line-protocol-corpus v0.0.0-20210922080147-aa28ccfb8937/go.mod h1:BKR9c0uHSmRgM/se9JhFHtTT7JTO67X23MtKMHtZcpo=
 github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE=
-github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8=
-github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
-github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
-github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
+github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
@@ -88,8 +89,8 @@ github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a h1:N9zuLhTvBSRt0g
 github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a/go.mod h1:JKx41uQRwqlTZabZc+kILPrO/3jlKnQ2Z8b7YiVw5cE=
 github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY=
 github.com/magiconair/properties v1.8.7/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0=
-github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
-github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
+github.com/minio/highwayhash v1.0.4-0.20251030100505-070ab1a87a76 h1:KGuD/pM2JpL9FAYvBrnBBeENKZNh6eNtjqytV6TYjnk=
+github.com/minio/highwayhash v1.0.4-0.20251030100505-070ab1a87a76/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
 github.com/moby/patternmatcher v0.6.0 h1:GmP9lR19aU5GqSSFko+5pRqHi+Ohk1O69aFiKkVGiPk=
 github.com/moby/patternmatcher v0.6.0/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc=
 github.com/moby/sys/sequential v0.5.0 h1:OPvI35Lzn9K04PBbCLW0g4LcFAJgHsvXsRyewg5lXtc=
@@ -100,20 +101,18 @@ github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A=
 github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
-github.com/nats-io/jwt/v2 v2.8.2 h1:XXRgB60MSTnqsRwejQurVDs/hcv2dkt+86GjI+I/bMc=
-github.com/nats-io/jwt/v2 v2.8.2/go.mod h1:Ag/56sq9OblL4JgdYufDd16Egb17Kr/8WwwuO/forVc=
-github.com/nats-io/nats-server/v2 v2.14.5 h1:M6yeo/Xb7khi97RSEVELof3DForDqmYza3P4tHCPFWw=
-github.com/nats-io/nats-server/v2 v2.14.5/go.mod h1:1D3iocrisKvWaD1B/imqarTqmaGrWMqALMLbEDo3v7Q=
-github.com/nats-io/nats.go v1.53.1 h1:Otsq3uLc/kLdjmkNHkXH0jBqwUquwdKFoe3fq6/3/Xo=
-github.com/nats-io/nats.go v1.53.1/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
-github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
-github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
+github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
+github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
+github.com/nats-io/nats-server/v2 v2.12.7 h1:prQ9cPiWHcnwfT81Wi5lU9LL8TLY+7pxDru6fQYLCQQ=
+github.com/nats-io/nats-server/v2 v2.12.7/go.mod h1:dOnmkprKMluTmTF7/QHZioxlau3sKHUM/LBPy9AiBPw=
+github.com/nats-io/nats.go v1.51.0 h1:ByW84XTz6W03GSSsygsZcA+xgKK8vPGaa/FCAAEHnAI=
+github.com/nats-io/nats.go v1.51.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
+github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
+github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
-github.com/oapi-codegen/nullable v1.1.0 h1:eAh8JVc5430VtYVnq00Hrbpag9PFRGWLjxR1/3KntMs=
-github.com/oapi-codegen/nullable v1.1.0/go.mod h1:KUZ3vUzkmEKY90ksAmit2+5juDIhIZhfDl+0PwOQlFY=
-github.com/oapi-codegen/runtime v1.7.0 h1:t7358VYPvNbWJ9gdAkIK/smVeHpBf6yp8VTsaZsb/7k=
-github.com/oapi-codegen/runtime v1.7.0/go.mod h1:GwV7hC2hviaMzj+ITfHVRESK5J2W/GefVwIND/bMGvU=
+github.com/oapi-codegen/runtime v1.4.0 h1:KLOSFOp7UzkbS7Cs1ms6NBEKYr0WmH2wZG0KKbd2er4=
+github.com/oapi-codegen/runtime v1.4.0/go.mod h1:5sw5fxCDmnOzKNYmkVNF8d34kyUeejJEY8HNT2WaPec=
 github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.1.0-rc5 h1:Ygwkfw9bpDvs+c9E34SdgGOj41dX/cbdlwvlWt0pnFI=
@@ -122,19 +121,22 @@ github.com/opencontainers/runc v1.1.5 h1:L44KXEpKmfWDcS02aeGm8QNTFXTo2D+8MYGDIJ/
 github.com/opencontainers/runc v1.1.5/go.mod h1:1J5XiS+vdZ3wCyZybsuxXZWGrgSr8fFJHLXuG2PsnNg=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b h1:0LFwY6Q3gMACTjAbMZBjXAqTOzOwFaj2Ld6cjeQ7Rig=
 github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
-github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU=
-github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE=
+github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o=
+github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg=
 github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
 github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
-github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY=
-github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc=
-github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
-github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
+github.com/prometheus/common v0.67.5 h1:pIgK94WWlQt1WLwAC5j2ynLaBRDiinoAb86HZHTUGI4=
+github.com/prometheus/common v0.67.5/go.mod h1:SjE/0MzDEEAyrdr5Gqc6G+sXI67maCxzaT3A2+HqjUw=
+github.com/prometheus/procfs v0.20.1 h1:XwbrGOIplXW/AU3YhIhLODXMJYyC1isLFfYCsTEycfc=
+github.com/prometheus/procfs v0.20.1/go.mod h1:o9EMBZGRyvDrSPH1RqdxhojkuXstoe4UlK79eF5TGGo=
 github.com/questdb/go-questdb-client/v4 v4.2.0 h1:+d0HJwCjUWMj7zmY6qmhoqTJzTyoYKl+LSTYGN0T8T8=
 github.com/questdb/go-questdb-client/v4 v4.2.0/go.mod h1:/2x93LK1wjM4JX/b5c6q7Yqk22htjWY1lE6p1X8iLbE=
+github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
 github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 h1:lZUw3E0/J3roVtGQ+SCrUrg3ON6NgVqpn3+iol9aGu4=
 github.com/santhosh-tekuri/jsonschema/v5 v5.3.1/go.mod h1:uToXkOrWAZ6/Oc07xWQrPOhJotwFIyu2bBVN41fcDUY=
 github.com/shirou/gopsutil/v3 v3.23.12 h1:z90NtUkp3bMtmICZKpC4+WaknU1eXtp5vtbQ11DgpE4=
@@ -147,8 +149,8 @@ github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+D
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
 github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
 github.com/spkg/bom v0.0.0-20160624110644-59b7046e48ad/go.mod h1:qLr4V1qq6nMqFKkMo8ZTx3f+BZEkzsRUY10Xsm2mwU0=
-github.com/stmcginnis/gofish v0.24.0 h1:zaBBFNtdSFH/+lJju29HMDHU3suIR+YhygqoJYxW+2Q=
-github.com/stmcginnis/gofish v0.24.0/go.mod h1:PzF5i8ecRG9A2ol8XT64npKUunyraJ+7t0kYMpQAtqU=
+github.com/stmcginnis/gofish v0.21.6 h1:jK3TGD6VANaAHKHypVNfD6io2nPrU+6eF8X4qARsTlY=
+github.com/stmcginnis/gofish v0.21.6/go.mod h1:PzF5i8ecRG9A2ol8XT64npKUunyraJ+7t0kYMpQAtqU=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
@@ -158,8 +160,8 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=
-github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/testcontainers/testcontainers-go v0.26.0 h1:uqcYdoOHBy1ca7gKODfBd9uTHVK3a7UL848z09MVZ0c=
 github.com/testcontainers/testcontainers-go v0.26.0/go.mod h1:ICriE9bLX5CLxL9OFQ2N+2N+f+803LNJ1utJb1+Inx0=
 github.com/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRUgPQU=
@@ -172,20 +174,18 @@ 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=
-go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw=
-go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg=
 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.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M=
-golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis=
-golang.org/x/exp v0.0.0-20240404231335-c0f41cb1a7a0 h1:985EYyeCOxTpcgOTJpflJUwOeEz0CQOdPt73OzpE9F8=
-golang.org/x/exp v0.0.0-20240404231335-c0f41cb1a7a0/go.mod h1:/lliqkxwWAhPjf5oSOIJup2XcqJaw8RGS6k3TGEc7GI=
+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=
+golang.org/x/exp v0.0.0-20231005195138-3e424a577f31/go.mod h1:S2oDrQGGwySpoQPVqRShND87VCbxmc6bL1Yd2oYrm6k=
 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.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To=
-golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU=
-golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
-golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
+golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
+golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
+golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
 golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
 golang.org/x/tools v0.14.0 h1:jvNa2pY0M4r62jkRQ6RwEZZyPcymeL9XZMLBbV7U2nc=
@@ -194,8 +194,9 @@ google.golang.org/genproto/googleapis/rpc v0.0.0-20231002182017-d307bd883b97 h1:
 google.golang.org/genproto/googleapis/rpc v0.0.0-20231002182017-d307bd883b97/go.mod h1:v7nGkzlmW8P3n/bKmWBn2WpBjpOEx8Q6gMueudAmKfY=
 google.golang.org/grpc v1.58.3 h1:BjnpXut1btbtgN/6sp+brB2Kbm2LjNXnidYujAVbSoQ=
 google.golang.org/grpc v1.58.3/go.mod h1:tgX3ZQDlNJGU96V6yHh1T/JeoBQ2TXdr43YbYSsCJk0=
-google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=
-google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
+google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
+google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

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")
+	}
+}