Fix linting issues

Extend multiChanTicker with GetDuration
Simplify cache, calculate with expr
2026-09-03 04:17:15 +02:00 · 2026-07-31 16:44:43 +02:00 · 2026-07-31 16:36:26 +02:00 · 2026-07-31 16:31:04 +02:00 · 2026-07-21 13:42:11 +02:00 · 2026-07-21 13:40:42 +02:00
29 changed files with 1512 additions and 722 deletions
@@ -33,6 +33,7 @@ There is a main configuration file with basic settings that point to the other c
  "receivers-file" : "receivers.json",
  "router-file" : "router.json",
  "main": {
+    "enable-markers": false,
    "interval": "10s",
    "duration": "1s"
  }
@@ -29,15 +29,17 @@ import (
 )

 type CentralConfigFile struct {
-	Interval string `json:"interval"`
-	Duration string `json:"duration"`
+	Interval      string `json:"interval"`
+	Duration      string `json:"duration"`
+	EnableMarkers bool   `json:"enable-markers"`
 }

 type RuntimeConfig struct {
-	Interval   time.Duration
-	Duration   time.Duration
-	CliArgs    map[string]string
-	ConfigFile CentralConfigFile
+	Interval      time.Duration
+	Duration      time.Duration
+	EnableMarkers bool
+	CliArgs       map[string]string
+	ConfigFile    CentralConfigFile

 	MetricRouter    mr.MetricRouter
 	CollectManager  collectors.CollectorManager
@@ -157,6 +159,7 @@ func mainFunc() int {
 		cclog.Error("The interval should be greater than duration")
 		return 1
 	}
+	rcfg.EnableMarkers = rcfg.ConfigFile.EnableMarkers

 	routerConf := ccconf.GetPackageConfig("router")
 	if len(routerConf) == 0 {
@@ -199,7 +202,7 @@ func mainFunc() int {
 	rcfg.MetricRouter.AddOutput(RouterToSinksChannel)

 	// Create new collector manager
-	rcfg.CollectManager, err = collectors.New(rcfg.MultiChanTicker, rcfg.Duration, &rcfg.Sync, collectorConf)
+	rcfg.CollectManager, err = collectors.New(rcfg.MultiChanTicker, rcfg.Duration, rcfg.EnableMarkers, &rcfg.Sync, collectorConf)
 	if err != nil {
 		cclog.Error(err.Error())
 		return 1
@@ -21,55 +21,57 @@ 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),
-	"nvidia_gpm":      new(NvidiaGPMCollector),
+	"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
+	parallel_run  bool                       // Flag whether the collectors are currently read in parallel
+	enableMarkers bool                       // Send ccmc-{begin,end} metrics
 }

 // Metric collector manager access functions
 type CollectorManager interface {
-	Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) error
+	Init(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) error
 	AddOutput(output chan lp.CCMessage)
 	Start()
 	Close()
@@ -82,7 +84,7 @@ type CollectorManager interface {
 // * ticker (from variable ticker)
 // * configuration (read from config file in variable collectConfigFile)
 // Initialization is done for all configured collectors
-func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) error {
+func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) error {
 	cm.collectors = make([]MetricCollector, 0)
 	cm.serial = make([]MetricCollector, 0)
 	cm.output = nil
@@ -90,6 +92,7 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
 	cm.wg = wg
 	cm.ticker = ticker
 	cm.duration = duration
+	cm.enableMarkers = enableMarkers

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

 		err := collector.Init(collectorCfg)
 		if err != nil {
-			cclog.ComponentErrorf("CollectorManager", "Collector %s initialization failed: %v", collectorName, err)
+			cclog.ComponentError("CollectorManager", fmt.Sprintf("Collector %s initialization failed: %v", collectorName, err))
 			continue
 		}
-		cclog.ComponentDebugf("CollectorManager", "ADD COLLECTOR %s", collector.Name())
+		cclog.ComponentDebug("CollectorManager", "ADD COLLECTOR", collector.Name())
 		if collector.Parallel() {
 			cm.collectors = append(cm.collectors, collector)
 		} else {
@@ -147,6 +150,14 @@ func (cm *collectorManager) Start() {
 				done()
 				return
 			case t := <-tick:
+				if cm.enableMarkers {
+					m, err := lp.NewMetric("ccmc-begin", map[string]string{"type": "node"}, nil, 0, time.Now())
+					if err != nil {
+						cclog.ComponentErrorf("CollectorManager", "Unable to create marker metric: %v", err)
+					} else {
+						cm.output <- m
+					}
+				}
 				cm.parallel_run = true
 				for _, c := range cm.collectors {
 					// Wait for done signal or execute the collector
@@ -156,7 +167,7 @@ func (cm *collectorManager) Start() {
 						return
 					default:
 						// Read metrics from collector c via goroutine
-						cclog.ComponentDebugf("CollectorManager: Read %s at %v", c.Name(), t)
+						cclog.ComponentDebug("CollectorManager", c.Name(), t)
 						cm.collector_wg.Add(1)
 						go func(myc MetricCollector) {
 							myc.Read(cm.duration, cm.output)
@@ -174,10 +185,18 @@ func (cm *collectorManager) Start() {
 						return
 					default:
 						// Read metrics from collector c
-						cclog.ComponentDebugf("CollectorManager: Read %s at %v", c.Name(), t)
+						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
+					}
+				}
 			}
 		}
 	})
@@ -200,9 +219,9 @@ func (cm *collectorManager) Close() {
 }

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

@@ -4,6 +4,7 @@
  "receivers-file" : "./receivers.json",
  "router-file" : "./router.json",
  "main" : {
+    "enable-markers": false,
    "interval": "10s",
    "duration": "1s"
  }
@@ -5,12 +5,12 @@ go 1.25.0
 require (
 	github.com/ClusterCockpit/cc-lib/v2 v2.12.0
 	github.com/ClusterCockpit/go-rocm-smi v0.4.0
-	github.com/NVIDIA/go-nvml v0.13.0-1
+	github.com/NVIDIA/go-nvml v0.13.3-1
 	github.com/PaesslerAG/gval v1.2.4
 	github.com/fsnotify/fsnotify v1.10.1
 	github.com/tklauser/go-sysconf v0.4.0
-	golang.design/x/thread v0.0.0-20210122121316-335e9adffdf1
-	golang.org/x/sys v0.45.0
+	golang.design/x/runtime v0.3.0
+	golang.org/x/sys v0.47.0
 )

 require (
@@ -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.3-1 h1:P76U2h88OZSiMtdhRsJjSF5DXyXUqHIXKeDicVAaae0=
+github.com/NVIDIA/go-nvml v0.13.3-1/go.mod h1:ahi2psRYoa+wYUBIrZPRO+wJs9lcvMhxSSkjjvsJJNQ=
 github.com/PaesslerAG/gval v1.2.4 h1:rhX7MpjJlcxYwL2eTTYIOBUyEKZ+A96T9vQySWkVUiU=
 github.com/PaesslerAG/gval v1.2.4/go.mod h1:XRFLwvmkTEdYziLdaCeCa5ImcGVrfQbeNUbVR+C6xac=
 github.com/PaesslerAG/jsonpath v0.1.0 h1:gADYeifvlqK3R3i2cR5B4DGgxLXIPb3TRTH1mGi0jPI=
@@ -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,9 +184,8 @@ golang.org/x/mod v0.13.0 h1:I/DsJXRlw/8l/0c24sM9yb0T4z9liZTduXvdAWYiysY=
 golang.org/x/mod v0.13.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
 golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
 golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
-golang.org/x/sys v0.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/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
+golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
 golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
 golang.org/x/tools v0.14.0 h1:jvNa2pY0M4r62jkRQ6RwEZZyPcymeL9XZMLBbV7U2nc=
@@ -69,8 +69,8 @@ var metricCacheLanguage = gval.NewLanguage(
 	gval.Function("getNumaCpuList", getCpuListOfNumaDomainFunc),
 	gval.Function("getDieCpuList", getCpuListOfDieFunc),
 	gval.Function("getCoreCpuList", getCpuListOfCoreFunc),
-	gval.Function("getCpuList", getCpuListOfNode),
-	gval.Function("getCpuListOfType", getCpuListOfType),
+	gval.Function("getCpuList", getCpuListOfNodeFunc),
+	gval.Function("getCpuListOfType", getCpuListOfTypeFunc),
 )

 var language gval.Language = gval.NewLanguage(
@@ -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 {
@@ -0,0 +1,449 @@
+package metricAggregator
+
+import (
+	"fmt"
+	"maps"
+	"math"
+	"slices"
+	"strings"
+	"sync"
+	"time"
+
+	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
+	lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
+
+	"github.com/expr-lang/expr"
+	"github.com/expr-lang/expr/vm"
+)
+
+type MetricAggregatorExprIntervalConfig struct {
+	Name                     string            `json:"name"`           // Metric name for the new metric
+	Function                 string            `json:"function"`       // Function to apply on the metric
+	Condition                string            `json:"if"`             // Condition for applying function
+	Tags                     map[string]string `json:"tags,omitempty"` // Tags for the new metric
+	Meta                     map[string]string `json:"meta,omitempty"` // Meta information for the new metric
+	ValueType                string            `json:"value_type,omitempty"`
+	IncludeNumCacheIntervals int               `json:"expr_on_num_intervals,omitempty"`
+	exprCond                 *vm.Program
+	exprFunc                 *vm.Program
+}
+
+type metricAggregatorExpr struct {
+	constants    map[string]any
+	output       chan lp.CCMessage
+	aggregations []MetricAggregatorExprIntervalConfig
+}
+
+var paramMapPool = sync.Pool{
+	New: func() any {
+		return make(map[string]any)
+	},
+}
+
+func sanitizeExprString(key string) string {
+	return strings.ReplaceAll(key, "type-id", "typeid")
+}
+
+// AddAggregation(name, function, condition string, tags, meta map[string]string) error
+// 	DeleteAggregation(name string) error
+// 	Init(output chan lp.CCMessage) error
+// 	Eval(starttime time.Time, endtime time.Time, metrics []lp.CCMessage)
+
+func (m *metricAggregatorExpr) Init(output chan lp.CCMessage) error {
+	m.output = output
+	m.constants = make(map[string]any)
+	m.aggregations = make([]MetricAggregatorExprIntervalConfig, 0)
+	return nil
+}
+
+func GetParamMap(point lp.CCMessage) map[string]any {
+	params := paramMapPool.Get().(map[string]any)
+	clear(params)
+
+	// Put metric name into params map
+	params["name"] = point.Name()
+
+	// Put full message into params map
+	params["message"] = point
+	params["msg"] = point
+
+	// Put timestamp into params map
+	params["timestamp"] = point.Time().Unix()
+	params["time"] = params["timestamp"]
+
+	// Put fields into params map
+	fields := paramMapPool.Get().(map[string]any)
+	clear(fields)
+	for key, value := range point.Fields() {
+		fields[key] = value
+		switch key {
+		case "value":
+			params["messagetype"] = "metric"
+			params["value"] = value
+			params["metric"] = value
+		case "event":
+			params["messagetype"] = "event"
+			params["event"] = value
+		case "control":
+			params["messagetype"] = "control"
+			params["control"] = value
+		case "log":
+			params["messagetype"] = "log"
+			params["log"] = value
+		default:
+			params["messagetype"] = "unknown"
+		}
+	}
+	params["msgtype"] = params["messagetype"]
+	params["fields"] = fields
+	params["field"] = fields
+
+	// Put tags into params map
+	tags := paramMapPool.Get().(map[string]any)
+	clear(tags)
+	for key, value := range point.Tags() {
+		tags[sanitizeExprString(key)] = value
+	}
+	params["tags"] = tags
+	params["tag"] = tags
+
+	// Put meta information into params map
+	meta := paramMapPool.Get().(map[string]any)
+	clear(meta)
+	for key, value := range point.Meta() {
+		meta[sanitizeExprString(key)] = value
+	}
+	params["meta"] = meta
+
+	return params
+}
+
+var baseenv_multi_message = map[string]any{
+	"name":                   "",
+	"starttime":              1234567890,
+	"endtime":                1234567890,
+	"messages":               make([]lp.CCMessage, 0),
+	"Median":                 medianfunc,
+	"Sum":                    sumfunc,
+	"Min":                    minfunc,
+	"Max":                    maxfunc,
+	"Mean":                   avgfunc,
+	"Avg":                    avgfunc,
+	"Match":                  matchfunc,
+	"getCpuCore":             getCpuCoreFunc,
+	"getCpuSocket":           getCpuSocketFunc,
+	"getCpuNumaDomain":       getCpuNumaDomainFunc,
+	"getCpuDie":              getCpuDieFunc,
+	"getCpuListOfCore":       getCpuListOfCoreFunc,
+	"getCpuListOfSocket":     getCpuListOfSocketFunc,
+	"getCpuListOfNumaDomain": getCpuListOfNumaDomainFunc,
+	"getCpuListOfDie":        getCpuListOfDieFunc,
+	"getCpuListOfNode":       getCpuListOfNodeFunc,
+	"getCpuListOfType":       getCpuListOfTypeFunc,
+}
+
+func (m *metricAggregatorExpr) AddAggregationWithType(name, function, condition, valueType string, tags, meta map[string]string) error {
+
+	cond, err := expr.Compile(condition, expr.Env(baseenv_multi_message), expr.AsBool(), expr.AllowUndefinedVariables())
+	if err != nil {
+		err = fmt.Errorf("failed to compile condition for aggregation %s: %s", name, err.Error())
+		return err
+	}
+	var exprOption = expr.AsFloat64()
+	switch valueType {
+	case "int":
+	case "int32":
+		exprOption = expr.AsInt()
+	case "int64":
+		exprOption = expr.AsInt64()
+	case "float32":
+	case "float64":
+		exprOption = expr.AsFloat64()
+	case "bool":
+		exprOption = expr.AsBool()
+	default:
+		err := fmt.Errorf("invalid value type '%s' for aggregation %s", valueType, name)
+		return err
+	}
+	f, err := expr.Compile(function, expr.Env(baseenv_multi_message), exprOption, expr.AllowUndefinedVariables())
+	if err != nil {
+		err = fmt.Errorf("failed to compile function for aggregation %s: %s", name, err.Error())
+		return err
+	}
+	m.aggregations = append(m.aggregations, MetricAggregatorExprIntervalConfig{
+		Name:      name,
+		Condition: condition,
+		Function:  function,
+		Tags:      tags,
+		Meta:      meta,
+		ValueType: valueType,
+		exprCond:  cond,
+		exprFunc:  f,
+	})
+	return nil
+}
+
+func (m *metricAggregatorExpr) AddAggregation(name, function, condition string, tags, meta map[string]string) error {
+	cclog.ComponentDebugf("MetricAggregator", "Adding %s", name)
+	err := m.AddAggregationWithType(name, function, condition, "float64", tags, meta)
+	cclog.ComponentDebugf("MetricAggregator", "Adding %s returned %v", name, err)
+	return err
+}
+
+func (m *metricAggregatorExpr) Eval(starttime, endtime time.Time, metrics []lp.CCMessage) {
+	cclog.ComponentDebugf("MetricAggregator", "Calculating %d expressions", len(m.aggregations))
+	copy_tags := func(tags map[string]string, metrics []lp.CCMessage) map[string]string {
+		out := make(map[string]string)
+		for key, value := range tags {
+			switch value {
+			case "<copy>":
+				for _, m := range metrics {
+					v, err := m.GetTag(key)
+					if err {
+						out[key] = v
+					}
+				}
+			default:
+				out[key] = value
+			}
+		}
+		return out
+	}
+	copy_meta := func(meta map[string]string, metrics []lp.CCMessage) map[string]string {
+		out := make(map[string]string)
+		for key, value := range meta {
+			switch value {
+			case "<copy>":
+				for _, m := range metrics {
+					v, err := m.GetMeta(key)
+					if err {
+						out[key] = v
+					}
+				}
+			default:
+				out[key] = value
+			}
+		}
+		return out
+	}
+
+	for _, aggr := range m.aggregations {
+		selected_metrics := make([]lp.CCMessage, 0)
+		values := make([]float64, 0)
+		aggr_vars := make(map[string]any)
+		maps.Copy(aggr_vars, baseenv_multi_message)
+		maps.Copy(aggr_vars, m.constants)
+		aggr_vars["starttime"] = starttime
+		aggr_vars["endtime"] = endtime
+
+		for _, met := range metrics {
+			met_vars := make(map[string]any)
+			maps.Copy(met_vars, aggr_vars)
+			met_vars["message"] = met
+			maps.Copy(met_vars, GetParamMap(met))
+			res, err := expr.Run(aggr.exprCond, met_vars)
+			if err != nil || res == false {
+				continue
+			}
+			if value, ok := met.GetField("value"); ok {
+				switch v := value.(type) {
+				case float64:
+					values = append(values, v)
+				case float32:
+				case int:
+				case int8:
+				case int16:
+				case int32:
+				case int64:
+				case uint:
+				case uint8:
+				case uint16:
+				case uint32:
+				case uint64:
+					values = append(values, float64(v))
+				case bool:
+					if v {
+						values = append(values, float64(1))
+					} else {
+						values = append(values, float64(0))
+					}
+				default:
+					cclog.ComponentErrorf("MetricAggregator", "Cannot convert value type for %s", met.ToLineProtocol(nil))
+					continue
+				}
+				selected_metrics = append(selected_metrics, met)
+			}
+		}
+		cclog.ComponentDebugf("MetricAggregator", "Collected %d values from %d metrics", len(values), len(selected_metrics))
+		aggr_vars["values"] = values
+		aggr_vars["messages"] = selected_metrics
+		if len(values) > 0 && len(selected_metrics) > 0 {
+			res, err := expr.Run(aggr.exprFunc, aggr_vars)
+			if err == nil {
+				tags := copy_tags(aggr.Tags, selected_metrics)
+				meta := copy_meta(aggr.Meta, selected_metrics)
+				msg, err := lp.NewMetric(aggr.Name, tags, meta, res, time.Now())
+				if err == nil {
+					cclog.ComponentDebugf("MetricAggregator", "Sending %s", msg.ToLineProtocol(nil))
+					select {
+					case m.output <- msg:
+					default:
+					}
+
+				}
+			} else {
+				cclog.ComponentErrorf("MetricAggregator", "Failed to calculate aggregation with name %s: %s", aggr.Name, err.Error())
+			}
+		}
+	}
+
+}
+
+func (c *metricAggregatorExpr) AddConstant(name string, value any) {
+	c.constants[name] = value
+}
+
+func (c *metricAggregatorExpr) DelConstant(name string) {
+	delete(c.constants, name)
+}
+
+func (c *metricAggregatorExpr) DeleteAggregation(name string) error {
+	i := slices.IndexFunc(
+		c.aggregations,
+		func(agg MetricAggregatorExprIntervalConfig) bool {
+			return agg.Name == name
+		})
+	if i == -1 {
+		return fmt.Errorf("no aggregation for metric name %s", name)
+	}
+	copy(c.aggregations[i:], c.aggregations[i+1:])
+	c.aggregations = c.aggregations[:len(c.aggregations)-1]
+	return nil
+}
+
+func NewAggregatorExpr(output chan lp.CCMessage) (MetricAggregator, error) {
+	a := new(metricAggregatorExpr)
+	err := a.Init(output)
+	if err != nil {
+		return nil, err
+	}
+	return a, err
+}
+
+var expr_cached map[string]*vm.Program = make(map[string]*vm.Program)
+var expr_cached_lock sync.Mutex
+
+var baseenv_message = map[string]any{
+	"name":        "",
+	"messagetype": "unknown",
+	"msgtype":     "unknown",
+	"tag": map[string]any{
+		"type":     "unknown",
+		"typeid":   "0",
+		"stype":    "unknown",
+		"stypeid":  "0",
+		"hostname": "localhost",
+		"cluster":  "nocluster",
+	},
+	"tags": map[string]any{
+		"type":     "unknown",
+		"typeid":   "0",
+		"stype":    "unknown",
+		"stypeid":  "0",
+		"hostname": "localhost",
+		"cluster":  "nocluster",
+	},
+	"meta": map[string]any{
+		"unit":   "invalid",
+		"source": "unknown",
+	},
+	"fields": map[string]any{
+		"value":   0,
+		"event":   "",
+		"control": "",
+		"log":     "",
+	},
+	"field": map[string]any{
+		"value":   0,
+		"event":   "",
+		"control": "",
+		"log":     "",
+	},
+	"timestamp": 1234567890,
+	"msg":       lp.EmptyMessage(),
+	"message":   lp.EmptyMessage(),
+}
+
+func EvalBoolConditionExpr(condition string, msg lp.CCMessage) (bool, error) {
+	scond := sanitizeExprString(condition)
+	expr_cached_lock.Lock()
+	evaluable, ok := expr_cached[scond]
+	expr_cached_lock.Unlock()
+	if !ok {
+		newcond := strings.ReplaceAll(
+			strings.ReplaceAll(
+				scond, "'", "\""), "%", "\\")
+		var err error
+		evaluable, err = expr.Compile(newcond, expr.Env(baseenv_message), expr.AsBool())
+		if err != nil {
+			return false, err
+		}
+		expr_cached_lock.Lock()
+		expr_cached[scond] = evaluable
+		expr_cached_lock.Unlock()
+	}
+	vars := GetParamMap(msg)
+
+	res, err := expr.Run(evaluable, vars)
+	if err != nil {
+		return false, err
+	}
+	paramMapPool.Put(vars)
+	return res.(bool), nil
+}
+
+func EvalFloat64ConditionExpr(condition string, msg lp.CCMessage) (float64, error) {
+	scond := sanitizeExprString(condition)
+	expr_cached_lock.Lock()
+	evaluable, ok := expr_cached[scond]
+	expr_cached_lock.Unlock()
+	if !ok {
+		newcond := strings.ReplaceAll(
+			strings.ReplaceAll(
+				scond, "'", "\""), "%", "\\")
+		var err error
+		evaluable, err = expr.Compile(newcond, expr.Env(baseenv_message), expr.AsFloat64())
+		if err != nil {
+			return math.NaN(), err
+		}
+		expr_cached_lock.Lock()
+		expr_cached[scond] = evaluable
+		expr_cached_lock.Unlock()
+	}
+	vars := GetParamMap(msg)
+	res, err := expr.Run(evaluable, vars)
+	paramMapPool.Put(vars)
+	return res.(float64), err
+}
+
+func EvalFloat64Expression(expression string, values map[string]any) (float64, error) {
+	sexpr := sanitizeExprString(expression)
+	expr_cached_lock.Lock()
+	evaluable, ok := expr_cached[sexpr]
+	expr_cached_lock.Unlock()
+	if !ok {
+		newcond := strings.ReplaceAll(
+			strings.ReplaceAll(
+				sexpr, "'", "\""), "%", "\\")
+		var err error
+		evaluable, err = expr.Compile(newcond, expr.Env(baseenv_message), expr.AsFloat64())
+		if err != nil {
+			return math.NaN(), err
+		}
+		expr_cached_lock.Lock()
+		expr_cached[sexpr] = evaluable
+		expr_cached_lock.Unlock()
+	}
+	res, err := expr.Run(evaluable, values)
+	return res.(float64), err
+}
@@ -0,0 +1,97 @@
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
+// All rights reserved. This file is part of cc-lib.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+// additional authors:
+// Holger Obermaier (NHR@KIT)
+
+package metricAggregator
+
+import (
+	"math"
+	"testing"
+	"time"
+
+	lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
+)
+
+func GenMetricNoCheck(value any, tags, meta map[string]string) lp.CCMessage {
+	msg, _ := lp.NewMetric("test", tags, meta, value, time.Now())
+	return msg
+}
+
+type TestBoolConfig struct {
+	msg             lp.CCMessage
+	cond            string
+	expected_result bool
+	should_fail     bool
+}
+
+var testBoolConfig []TestBoolConfig = []TestBoolConfig{
+	{
+		msg:             GenMetricNoCheck(1.0, nil, nil),
+		cond:            "fields.value == 1",
+		expected_result: true,
+		should_fail:     false,
+	},
+	{
+		msg:             GenMetricNoCheck(1.0, map[string]string{"hostname": "testhost"}, nil),
+		cond:            "tags.hostname == 'testhost'",
+		expected_result: true,
+		should_fail:     false,
+	},
+}
+
+func TestEvalBoolConditionExprSimple(t *testing.T) {
+
+	for _, test := range testBoolConfig {
+		res, err := EvalBoolConditionExpr(test.cond, test.msg)
+		if err != nil && !test.should_fail {
+			t.Error(err.Error())
+			return
+		}
+		if !test.should_fail && res != test.expected_result {
+			t.Errorf("Condition '%s' evaluated to %v despite expecting %v", test.cond, res, test.expected_result)
+			return
+		}
+	}
+}
+
+type TestFloat64Config struct {
+	msg             lp.CCMessage
+	cond            string
+	expected_result float64
+	should_fail     bool
+}
+
+var testFloat64Config []TestFloat64Config = []TestFloat64Config{
+	{
+		msg:             GenMetricNoCheck(1.0, nil, nil),
+		cond:            "fields.value + 3.14",
+		expected_result: 4.14,
+		should_fail:     false,
+	},
+	{
+		msg:             GenMetricNoCheck(2.0, nil, nil),
+		cond:            "fields.value * 2",
+		expected_result: 4.0,
+		should_fail:     false,
+	},
+}
+
+const compareFloat64Max = 1e-9
+
+func TestEvalFloat64ConditionExprSimple(t *testing.T) {
+
+	for _, test := range testFloat64Config {
+		res, err := EvalFloat64ConditionExpr(test.cond, test.msg)
+		if err != nil && !test.should_fail {
+			t.Error(err.Error())
+			return
+		}
+		if !test.should_fail && math.Abs(res-test.expected_result) > compareFloat64Max {
+			t.Errorf("Condition '%s' evaluated to %f despite expecting %f", test.cond, res, test.expected_result)
+			return
+		}
+	}
+}
@@ -336,14 +336,14 @@ func getCpuListOfDieFunc(args any) (any, error) {
 }

 // wrapper function to get a list of all cpuids of the node
-func getCpuListOfNode() (any, error) {
+func getCpuListOfNodeFunc() (any, error) {
 	return topo.HwthreadList(), nil
 }

 // helper function to get the cpuid list for a CCMetric type tag set (type and type-id)
 // since there is no access to the metric data in the function, is should be called like
 // `getCpuListOfType()`
-func getCpuListOfType(args ...any) (any, error) {
+func getCpuListOfTypeFunc(args ...any) (any, error) {
 	cpulist := make([]int, 0)
 	switch typ := args[0].(type) {
 	case string:
@@ -9,6 +9,8 @@ package metricRouter

 import (
 	"fmt"
+	"math"
+	"strings"
 	"sync"
 	"time"

@@ -19,57 +21,107 @@ import (
 	mct "github.com/ClusterCockpit/cc-metric-collector/pkg/multiChanTicker"
 )

-type metricCachePeriod struct {
-	startstamp  time.Time
-	stopstamp   time.Time
-	numMetrics  int
-	sizeMetrics int
-	metrics     []lp.CCMessage
+type ccCache struct {
+	periodIdx   int
+	maxPeriods  int
+	periods     [][]lp.CCMessage
+	periodTimes []struct {
+		starttime time.Time
+		endtime   time.Time
+	}
+}
+
+type CCCache interface {
+	Init(numPeriods int) error
+	Add(msg lp.CCMessage) error
+	GetPeriod(offset int) (time.Time, time.Time, []lp.CCMessage)
+	GetAll() []lp.CCMessage
+	NewPeriod()
+}
+
+func (c *ccCache) Init(numPeriods int) error {
+	c.maxPeriods = numPeriods
+	c.periodIdx = 0
+	c.periods = make([][]lp.CCMessage, c.maxPeriods)
+	c.periodTimes = make([]struct {
+		starttime time.Time
+		endtime   time.Time
+	}, c.maxPeriods)
+	return nil
+}
+
+func (c *ccCache) NewPeriod() {
+	c.periodTimes[c.periodIdx].endtime = time.Now()
+	c.periodIdx = (c.periodIdx + 1) % c.maxPeriods
+	fmt.Printf("New period index %d\n", c.periodIdx)
+	c.periods[c.periodIdx] = c.periods[c.periodIdx][:0]
+	c.periodTimes[c.periodIdx].starttime = time.Now()
+	c.periodTimes[c.periodIdx].endtime = c.periodTimes[c.periodIdx].starttime
+}
+
+func (c *ccCache) Add(msg lp.CCMessage) error {
+	c.periods[c.periodIdx] = append(c.periods[c.periodIdx], msg)
+	c.periodTimes[c.periodIdx].endtime = msg.Time()
+	return nil
+}
+
+func (c *ccCache) GetPeriod(offset int) (time.Time, time.Time, []lp.CCMessage) {
+	if offset > c.maxPeriods {
+		offset = offset % c.maxPeriods
+	}
+	poff := int(math.Abs(float64(c.periodIdx - offset)))
+	out := make([]lp.CCMessage, 0, len(c.periods[poff%c.maxPeriods]))
+	out = append(out, c.periods[poff%c.maxPeriods]...)
+
+	return c.periodTimes[poff%c.maxPeriods].starttime, c.periodTimes[poff%c.maxPeriods].endtime, out
+}
+
+func (c *ccCache) GetAll() []lp.CCMessage {
+	out := make([]lp.CCMessage, 0)
+	for _, data := range c.periods {
+		out = append(out, data...)
+	}
+	return out
 }

 // Metric cache data structure
 type metricCache struct {
-	numPeriods int
-	curPeriod  int
-	lock       sync.Mutex
-	intervals  []*metricCachePeriod
+	cache      CCCache
 	wg         *sync.WaitGroup
 	ticker     mct.MultiChanTicker
 	tickchan   chan time.Time
 	done       chan bool
 	output     chan lp.CCMessage
 	aggEngine  agg.MetricAggregator
+	numPeriods int
+	started    bool
 }

 type MetricCache interface {
-	Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, numPeriods int) error
+	Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, interval time.Duration, numPeriods int) error
 	Start()
 	Add(metric lp.CCMessage)
-	GetPeriod(index int) (time.Time, time.Time, []lp.CCMessage)
 	AddAggregation(name, function, condition string, tags, meta map[string]string) error
 	DeleteAggregation(name string) error
 	Close()
 }

-func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, numPeriods int) error {
+func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, interval time.Duration, numPeriods int) error {
 	var err error
 	c.done = make(chan bool)
 	c.wg = wg
 	c.ticker = ticker
 	c.numPeriods = numPeriods
+	c.started = false
+	c.cache = new(ccCache)
 	c.output = output
-	c.intervals = make([]*metricCachePeriod, 0)
-	for i := 0; i < c.numPeriods+1; i++ {
-		p := new(metricCachePeriod)
-		p.numMetrics = 0
-		p.sizeMetrics = 0
-		p.metrics = make([]lp.CCMessage, 0)
-		c.intervals = append(c.intervals, p)
+
+	err = c.cache.Init(numPeriods)
+	if err != nil {
+		return fmt.Errorf("MetricCache: failed to create cache: %w", err)
 	}

-	// Create a new aggregation engine. No separate goroutine at the moment
-	// The code is executed by the MetricCache goroutine
-	c.aggEngine, err = agg.NewAggregator(c.output)
+	c.aggEngine, err = agg.NewAggregatorExpr(c.output)
 	if err != nil {
 		return fmt.Errorf("MetricCache: failed to create aggregator: %w", err)
 	}
@@ -81,68 +133,58 @@ func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker,
 func (c *metricCache) Start() {
 	c.tickchan = make(chan time.Time)
 	c.ticker.AddChannel(c.tickchan)
-	// Router cache is done
-	done := func() {
-		cclog.ComponentDebug("MetricCache", "DONE")
-		close(c.done)
-	}

-	// Rotate cache interval
-	rotate := func(timestamp time.Time) int {
-		oldPeriod := c.curPeriod
-		c.curPeriod = oldPeriod + 1
-		if c.curPeriod >= c.numPeriods {
-			c.curPeriod = 0
-		}
-		c.intervals[oldPeriod].numMetrics = 0
-		c.intervals[oldPeriod].stopstamp = timestamp
-		c.intervals[c.curPeriod].startstamp = timestamp
-		c.intervals[c.curPeriod].stopstamp = timestamp
-		return oldPeriod
-	}
-
-	c.wg.Go(func() {
+	c.wg.Add(1)
+	go func() {
 		for {
 			select {
 			case <-c.done:
-				done()
+				c.wg.Done()
+				close(c.done)
+				cclog.ComponentDebug("MetricCache", "DONE")
+
 				return
 			case tick := <-c.tickchan:
-				c.lock.Lock()
-				old := rotate(tick)
-				// Get the last period and evaluate aggregation metrics
-				starttime, endtime, metrics := c.GetPeriod(old)
-				c.lock.Unlock()
-				if len(metrics) > 0 {
-					c.aggEngine.Eval(starttime, endtime, metrics)
+				cclog.ComponentDebug("MetricCache", "Tick", tick)
+				allmetrics := c.cache.GetAll()
+				c.cache.NewPeriod()
+				mintime := tick
+				maxtime := mintime.AddDate(-1, 0, 0)
+
+				for _, metric := range allmetrics {
+					if metric.Time().Before(mintime) {
+						mintime = metric.Time()
+					}
+					if metric.Time().After(maxtime) {
+						maxtime = metric.Time()
+					}
+				}
+				if len(allmetrics) > 0 {
+					cclog.ComponentDebugf("MetricCache", "Evaluate %d metrics from %v to %v", len(allmetrics), mintime.UnixNano(), maxtime.UnixNano())
+					c.wg.Go(func() {
+						c.aggEngine.Eval(mintime, maxtime, allmetrics)
+					})
 				} else {
 					// This message is also printed in the first interval after startup
 					cclog.ComponentDebug("MetricCache", "EMPTY INTERVAL?")
 				}
+
 			}
 		}
-	})
-	cclog.ComponentDebug("MetricCache", "START")
+	}()
+
+	cclog.ComponentDebug("MetricCache", "STARTED")
 }

 // Add a metric to the cache. The interval is defined by the global timer (rotate() in Start())
 // The intervals list is used as round-robin buffer and the metric list grows dynamically and
 // to avoid reallocations
 func (c *metricCache) Add(metric lp.CCMessage) {
-	if c.curPeriod >= 0 && c.curPeriod < c.numPeriods {
-		c.lock.Lock()
-		p := c.intervals[c.curPeriod]
-		if p.numMetrics < p.sizeMetrics {
-			p.metrics[p.numMetrics] = metric
-			p.numMetrics++
-			p.stopstamp = metric.Time()
-		} else {
-			p.metrics = append(p.metrics, metric)
-			p.numMetrics++
-			p.sizeMetrics++
-			p.stopstamp = metric.Time()
-		}
-		c.lock.Unlock()
+	err := c.cache.Add(metric)
+	if err != nil {
+		s := metric.ToLineProtocol(nil)
+		s = strings.TrimSpace(s)
+		cclog.ComponentErrorf("MetricCache", "Failed to add metric %s", s)
 	}
 }

@@ -154,40 +196,20 @@ func (c *metricCache) DeleteAggregation(name string) error {
 	return c.aggEngine.DeleteAggregation(name)
 }

-// Get all metrics of a interval. The index is the difference to the current interval, so index=0
-// is the current one, index=1 the last interval and so on. Returns and empty array if a wrong index
-// is given (negative index, index larger than configured number of total intervals, ...)
-func (c *metricCache) GetPeriod(index int) (time.Time, time.Time, []lp.CCMessage) {
-	start := time.Now()
-	stop := time.Now()
-	var metrics []lp.CCMessage
-	if index >= 0 && index < c.numPeriods {
-		pindex := c.curPeriod - index
-		if pindex < 0 {
-			pindex = c.numPeriods - pindex
-		}
-		if pindex >= 0 && pindex < c.numPeriods {
-			start = c.intervals[pindex].startstamp
-			stop = c.intervals[pindex].stopstamp
-			metrics = c.intervals[pindex].metrics
-		} else {
-			metrics = make([]lp.CCMessage, 0)
-		}
-	} else {
-		metrics = make([]lp.CCMessage, 0)
-	}
-	return start, stop, metrics
-}
-
 // Close finishes / stops the metric cache
 func (c *metricCache) Close() {
 	cclog.ComponentDebug("MetricCache", "CLOSE")
-	c.done <- true
+	if c.started {
+		c.done <- true
+		c.wg.Wait()
+
+	}
+	cclog.ComponentDebug("MetricCache", "CLOSED")
 }

 func NewCache(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, numPeriods int) (MetricCache, error) {
 	c := new(metricCache)
-	err := c.Init(output, ticker, wg, numPeriods)
+	err := c.Init(output, ticker, wg, ticker.GetDuration(), numPeriods)
 	if err != nil {
 		return nil, err
 	}
@@ -0,0 +1,69 @@
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
+// All rights reserved. This file is part of cc-lib.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+// additional authors:
+// Holger Obermaier (NHR@KIT)
+
+package metricRouter
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
+	mct "github.com/ClusterCockpit/cc-metric-collector/pkg/multiChanTicker"
+)
+
+func TestCache(t *testing.T) {
+	output := make(chan lp.CCMessage, 2000)
+
+	var wg sync.WaitGroup
+	tickTime := time.Second
+	testChan := make(chan time.Time)
+	ticker := mct.NewTicker(tickTime)
+	ticker.AddChannel(testChan)
+	maxIntervals := 100
+
+	c, err := NewCache(output, ticker, &wg, 4)
+	if err != nil {
+		t.Errorf("failed to create new cache: %s", err.Error())
+		return
+	}
+	c.Start()
+	defer c.Close()
+	err = c.AddAggregation("ps_input_power", "Avg(values)", "name == 'ps1_input_power' || name == 'ps2_input_power' || name == 'ps3_input_power'", map[string]string{"hostname": "<copy>", "type": "<copy>"}, map[string]string{"unit": "<copy>"})
+	if err != nil {
+		t.Errorf("failed to add aggregation for %s", "ps_input_power")
+		return
+	}
+
+	for range maxIntervals {
+		timestamp := <-testChan
+		raw_metrics := []string{
+			fmt.Sprintf("ps1_input_power,type=node,hostname=myhost,unit=W value=696.0 %d", timestamp.UnixNano()),
+			fmt.Sprintf("ps2_input_power,type=node,hostname=myhost,unit=W value=732.0 %d", timestamp.UnixNano()),
+			fmt.Sprintf("ps3_input_power,type=node,hostname=myhost,unit=W value=720.0 %d", timestamp.UnixNano()),
+			fmt.Sprintf("cpu_load,type=hwthread,type-id=0,hostname=myhost value=45 %d", timestamp.UnixNano()),
+		}
+		metrics, err := lp.FromBytes([]byte(strings.Join(raw_metrics, "\n")))
+		if err != nil {
+			t.Errorf("failed to generate metrics: %s", err.Error())
+		}
+		for _, m := range metrics {
+			c.Add(m)
+		}
+
+		if len(output) > 0 {
+			for i := 0; i < len(output); i++ {
+				m := <-output
+				t.Log(m.ToLineProtocol(nil))
+			}
+		}
+
+	}
+
+}
@@ -35,19 +35,19 @@ type metricRouterTagConfig struct {

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

 // Metric router data structure
@@ -253,7 +253,7 @@ func (r *metricRouter) Start() {
 		}
 		// even if the metric is dropped, it is stored in the cache for
 		// aggregations
-		if r.config.NumCacheIntervals > 0 {
+		if r.config.NumCacheIntervals > 0 && m != nil {
 			r.cache.Add(m)
 		}
 	}
@@ -17,17 +17,20 @@ type multiChanTicker struct {
 	ticker   *time.Ticker
 	channels []chan time.Time
 	done     chan bool
+	duration time.Duration
 }

 type MultiChanTicker interface {
 	Init(duration time.Duration)
 	AddChannel(channel chan time.Time)
+	GetDuration() time.Duration
 	Close()
 }

 func (t *multiChanTicker) Init(duration time.Duration) {
 	t.ticker = time.NewTicker(duration)
 	t.done = make(chan bool)
+	t.duration = duration
 	go func() {
 		done := func() {
 			close(t.done)
@@ -53,6 +56,10 @@ func (t *multiChanTicker) Init(duration time.Duration) {
 	}()
 }

+func (t *multiChanTicker) GetDuration() time.Duration {
+	return t.duration
+}
+
 func (t *multiChanTicker) AddChannel(channel chan time.Time) {
 	t.channels = append(t.channels, channel)
 }
Author	SHA1	Message	Date
Thomas Gruber	5f9d182903	Fix linting issues	2026-07-31 16:44:43 +02:00
Thomas Gruber	80dbd41b91	Extend multiChanTicker with GetDuration	2026-07-31 16:36:26 +02:00
Thomas Gruber	4c1f810f53	Simplify cache, calculate with expr	2026-07-31 16:31:04 +02:00
Michael Panzlaff	789ba7f4a5	Revert "Prepare for FAU" This reverts commit `9bd9a07810`. This should never have made it into our main repo.	2026-07-21 13:42:11 +02:00
Michael Panzlaff	9bd9a07810	Prepare for FAU	2026-07-21 13:40:42 +02:00
Michael PanzlaffandGitHub	a0aedff202	Merge pull request #235 from ClusterCockpit/marker-commits Add marker metrics	2026-07-21 12:51:37 +02:00
Michael Panzlaff	6944454592	Update configuration documentation for new marker metrics	2026-07-21 12:47:16 +02:00
Michael Panzlaff	9e773e2e85	Fix marker metric creation	2026-07-20 16:16:10 +02:00
Michael Panzlaff	c13bb55735	Actually send the created marker metrics	2026-07-20 15:39:40 +02:00
Michael Panzlaff	6124971b2e	collectorManager: Make markers opt-in	2026-07-15 18:04:56 +02:00
Michael Panzlaff	cdac3c9163	Add ccmc-{begin,end} markers These markers are sent out at the beginning and end of a collection run. This can be used to reconstruct, which metrics belong together and were obtained during the same tick.	2026-07-15 15:45:15 +02:00
dependabot[bot]andThomas Gruber	70364d084b	Bump golang.org/x/sys from 0.45.0 to 0.47.0 Bumps [golang.org/x/sys](https://github.com/golang/sys) from 0.45.0 to 0.47.0. - [Commits](https://github.com/golang/sys/compare/v0.45.0...v0.47.0) --- updated-dependencies: - dependency-name: golang.org/x/sys dependency-version: 0.47.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com>	2026-07-13 12:06:53 +02:00
dependabot[bot]andThomas Gruber	ec0868eba9	Bump github.com/NVIDIA/go-nvml from 0.13.2-0 to 0.13.3-1 Bumps [github.com/NVIDIA/go-nvml](https://github.com/NVIDIA/go-nvml) from 0.13.2-0 to 0.13.3-1. - [Release notes](https://github.com/NVIDIA/go-nvml/releases) - [Commits](https://github.com/NVIDIA/go-nvml/compare/v0.13.2-0...v0.13.3-1) --- updated-dependencies: - dependency-name: github.com/NVIDIA/go-nvml dependency-version: 0.13.3-1 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com>	2026-07-13 12:06:37 +02:00
Michael Panzlaff	b8e76e3bf0	megwareEureka: Fix collector name	2026-07-07 19:03:57 +02:00
Michael Panzlaff	f028c83a75	megwareEureka: Remove debug output This should never have been in the release...	2026-07-07 18:47:02 +02:00
Michael PanzlaffandGitHub	4a859af358	Merge pull request #231 from ClusterCockpit/likwid-lint-fix likwidMetric: Replace deprecated x/thread package	2026-07-07 17:08:46 +02:00
Michael Panzlaff	c4fb350ad2	likwidMetric: Replace deprecated x/thread package	2026-07-07 17:03:50 +02:00
Michael PanzlaffandGitHub	14000c0ff0	Merge pull request #230 from ClusterCockpit/helma-collectors NHR@FAU Helma energy collectors	2026-07-07 16:57:23 +02:00
Michael Panzlaff	f6d33fb29f	Fix golangci-lint checks	2026-07-07 16:51:06 +02:00
Michael Panzlaff	025a5c2861	megwareEureka: JSON fields must be public in order to Unmarshal	2026-07-07 15:42:23 +02:00
Michael Panzlaff	8d1df385ce	make fmt	2026-07-06 14:09:45 +02:00
Michael Panzlaff	bda952f85a	Add MegwareEurekaCollector	2026-07-06 14:06:41 +02:00
Michael Panzlaff	81fa732e79	lenovoDensePower: Improve documentation, fix things Fix energy and time overflow handling. Send the metric as "lenovo_node_power" per default. This should avoid clashes by default. If you want the old behavior, use a router rule to rename the metric.	2026-07-03 18:57:06 +02:00
Michael Panzlaff	93622b72ac	lenovoDensePower: Fix linter checks	2026-07-03 17:45:19 +02:00
Michael Panzlaff	f48eb458e8	make fmt	2026-07-03 17:28:46 +02:00
Michael Panzlaff	9fb3d2300c	lenovoDensePower: Actually mark it as initialized after Init()	2026-07-03 17:20:46 +02:00
Michael Panzlaff	4372cb7994	lenovoDensePower: Strip newlines as well from output	2026-07-03 17:08:40 +02:00
Michael Panzlaff	a38804ab13	Use more compact decimals in lenovo ipmi test command	2026-07-03 17:08:01 +02:00
Michael Panzlaff	852ff18468	Add LenovoDensePower collector	2026-07-03 16:54:50 +02:00
Thomas Roehl	61570f9ecf	Merge branch 'main' of github.com:ClusterCockpit/cc-metric-collector	2026-06-30 12:49:50 +02:00
Thomas Roehl	3098e16ded	Use formating cclog functions	2026-06-30 12:49:45 +02:00
Thomas Roehl	d53ec9233d	Check scanner error and remove %w/%v formats	2026-06-30 12:48:14 +02:00
brinkcoderandThomas Gruber	aaabcc639f	fix: support SLUID-based Slurm cgroup directories	2026-06-26 12:22:08 +01:00
Michael SchwarzandThomas Gruber	14d624fd22	Logic error, move creation of map upwards	2026-06-22 15:08:43 +01:00
Michael SchwarzandThomas Gruber	9677a916be	Use map of included metrics instead of iterating through the config again and again	2026-06-22 15:08:43 +01:00
Michael SchwarzandThomas Gruber	8b373cc2c0	Add filter for ipmi metrics	2026-06-22 15:08:43 +01:00
dependabot[bot]andThomas Gruber	f6260a8fb2	Bump github.com/NVIDIA/go-nvml from 0.13.0-1 to 0.13.2-0 Bumps [github.com/NVIDIA/go-nvml](https://github.com/NVIDIA/go-nvml) from 0.13.0-1 to 0.13.2-0. - [Release notes](https://github.com/NVIDIA/go-nvml/releases) - [Commits](https://github.com/NVIDIA/go-nvml/compare/v0.13.0-1...v0.13.2-0) --- updated-dependencies: - dependency-name: github.com/NVIDIA/go-nvml dependency-version: 0.13.2-0 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com>	2026-06-22 08:42:16 +01:00
Holger ObermaierandGitHub	bed5491068	Fix Overflows in Infiniband collector (#219 ) * Add information about the used infiniband counters * Change datatype from int64 to uint64 * uint64 subtraction handles wraparound automatically * Compute total rates by summing up the xmit and recv rates. This avoids overflows in the raw counters * Check for cases where the current counter can not be saved as last state * Use golang variable naming convention (camelCase)	2026-06-08 14:00:09 +02:00
dependabot[bot]andThomas Gruber	a2eba41150	Bump golang.design/x/thread Bumps [golang.design/x/thread](https://github.com/golang-design/thread) from 0.0.0-20210122121316-335e9adffdf1 to 0.3.2. - [Release notes](https://github.com/golang-design/thread/releases) - [Commits](https://github.com/golang-design/thread/commits/v0.3.2) --- updated-dependencies: - dependency-name: golang.design/x/thread dependency-version: 0.3.2 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com>	2026-06-08 13:10:27 +02:00