Update rapl collector with powercap limits

README.md

@@ -33,7 +33,6 @@ There is a main configuration file with basic settings that point to the other c
   "receivers-file" : "receivers.json",
   "router-file" : "router.json",
   "main": {
-    "enable-markers": false,
     "interval": "10s",
     "duration": "1s"
   }

cc-metric-collector.go

@@ -29,17 +29,15 @@ import (
 )
 
 type CentralConfigFile struct {
-	Interval      string `json:"interval"`
-	Duration      string `json:"duration"`
-	EnableMarkers bool   `json:"enable-markers"`
+	Interval string `json:"interval"`
+	Duration string `json:"duration"`
 }
 
 type RuntimeConfig struct {
-	Interval      time.Duration
-	Duration      time.Duration
-	EnableMarkers bool
-	CliArgs       map[string]string
-	ConfigFile    CentralConfigFile
+	Interval   time.Duration
+	Duration   time.Duration
+	CliArgs    map[string]string
+	ConfigFile CentralConfigFile
 
 	MetricRouter    mr.MetricRouter
 	CollectManager  collectors.CollectorManager
@@ -159,7 +157,6 @@ 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 {
@@ -202,7 +199,7 @@ func mainFunc() int {
 	rcfg.MetricRouter.AddOutput(RouterToSinksChannel)
 
 	// Create new collector manager
-	rcfg.CollectManager, err = collectors.New(rcfg.MultiChanTicker, rcfg.Duration, rcfg.EnableMarkers, &rcfg.Sync, collectorConf)
+	rcfg.CollectManager, err = collectors.New(rcfg.MultiChanTicker, rcfg.Duration, &rcfg.Sync, collectorConf)
 	if err != nil {
 		cclog.Error(err.Error())
 		return 1

collectors/collectorManager.go

@@ -21,57 +21,54 @@ import (
 
 // Map of all available metric collectors
 var AvailableCollectors = map[string]MetricCollector{
-	"likwid":             new(LikwidCollector),
-	"loadavg":            new(LoadavgCollector),
-	"memstat":            new(MemstatCollector),
-	"netstat":            new(NetstatCollector),
-	"ibstat":             new(InfinibandCollector),
-	"lustrestat":         new(LustreCollector),
-	"cpustat":            new(CpustatCollector),
-	"topprocs":           new(TopProcsCollector),
-	"nvidia":             new(NvidiaCollector),
-	"customcmd":          new(CustomCmdCollector),
-	"iostat":             new(IOstatCollector),
-	"diskstat":           new(DiskstatCollector),
-	"tempstat":           new(TempCollector),
-	"ipmistat":           new(IpmiCollector),
-	"gpfs":               new(GpfsCollector),
-	"cpufreq":            new(CPUFreqCollector),
-	"cpufreq_cpuinfo":    new(CPUFreqCpuInfoCollector),
-	"nfs3stat":           new(Nfs3Collector),
-	"nfs4stat":           new(Nfs4Collector),
-	"numastats":          new(NUMAStatsCollector),
-	"beegfs_meta":        new(BeegfsMetaCollector),
-	"beegfs_storage":     new(BeegfsStorageCollector),
-	"rapl":               new(RAPLCollector),
-	"rocm_smi":           new(RocmSmiCollector),
-	"self":               new(SelfCollector),
-	"schedstat":          new(SchedstatCollector),
-	"nfsiostat":          new(NfsIOStatCollector),
-	"slurm_cgroup":       new(SlurmCgroupCollector),
-	"smartmon":           new(SmartMonCollector),
-	"lenovo_dense_power": new(LenovoDensePowerCollector),
-	"megware_eureka":     new(MegwareEurekaCollector),
+	"likwid":          new(LikwidCollector),
+	"loadavg":         new(LoadavgCollector),
+	"memstat":         new(MemstatCollector),
+	"netstat":         new(NetstatCollector),
+	"ibstat":          new(InfinibandCollector),
+	"lustrestat":      new(LustreCollector),
+	"cpustat":         new(CpustatCollector),
+	"topprocs":        new(TopProcsCollector),
+	"nvidia":          new(NvidiaCollector),
+	"customcmd":       new(CustomCmdCollector),
+	"iostat":          new(IOstatCollector),
+	"diskstat":        new(DiskstatCollector),
+	"tempstat":        new(TempCollector),
+	"ipmistat":        new(IpmiCollector),
+	"gpfs":            new(GpfsCollector),
+	"cpufreq":         new(CPUFreqCollector),
+	"cpufreq_cpuinfo": new(CPUFreqCpuInfoCollector),
+	"nfs3stat":        new(Nfs3Collector),
+	"nfs4stat":        new(Nfs4Collector),
+	"numastats":       new(NUMAStatsCollector),
+	"beegfs_meta":     new(BeegfsMetaCollector),
+	"beegfs_storage":  new(BeegfsStorageCollector),
+	"rapl":            new(RAPLCollector),
+	"rocm_smi":        new(RocmSmiCollector),
+	"self":            new(SelfCollector),
+	"schedstat":       new(SchedstatCollector),
+	"nfsiostat":       new(NfsIOStatCollector),
+	"slurm_cgroup":    new(SlurmCgroupCollector),
+	"smartmon":        new(SmartMonCollector),
 }
 
 // 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
-	enableMarkers bool                       // Send ccmc-{begin,end} metrics
+	collectors   []MetricCollector          // List of metric collectors to read in parallel
+	serial       []MetricCollector          // List of metric collectors to read serially
+	output       chan lp.CCMessage          // Output channels
+	done         chan bool                  // channel to finish / stop metric collector manager
+	ticker       mct.MultiChanTicker        // periodically ticking once each interval
+	duration     time.Duration              // duration (for metrics that measure over a given duration)
+	wg           *sync.WaitGroup            // wait group for all goroutines in cc-metric-collector
+	config       map[string]json.RawMessage // json encoded config for collector manager
+	collector_wg sync.WaitGroup             // internally used wait group for the parallel reading of collector
+	parallel_run bool                       // Flag whether the collectors are currently read in parallel
 }
 
 // Metric collector manager access functions
 type CollectorManager interface {
-	Init(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) error
+	Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) error
 	AddOutput(output chan lp.CCMessage)
 	Start()
 	Close()
@@ -84,7 +81,7 @@ type CollectorManager interface {
 // * ticker (from variable ticker)
 // * configuration (read from config file in variable collectConfigFile)
 // Initialization is done for all configured collectors
-func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) error {
+func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) error {
 	cm.collectors = make([]MetricCollector, 0)
 	cm.serial = make([]MetricCollector, 0)
 	cm.output = nil
@@ -92,7 +89,6 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
 	cm.wg = wg
 	cm.ticker = ticker
 	cm.duration = duration
-	cm.enableMarkers = enableMarkers
 
 	d := json.NewDecoder(bytes.NewReader(collectConfig))
 	d.DisallowUnknownFields()
@@ -150,14 +146,6 @@ 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
@@ -189,14 +177,6 @@ func (cm *collectorManager) Start() {
 						c.Read(cm.duration, cm.output)
 					}
 				}
-				if cm.enableMarkers {
-					m, err := lp.NewMetric("ccmc-end", map[string]string{"type": "node"}, nil, 0, time.Now())
-					if err != nil {
-						cclog.ComponentErrorf("CollectorManager", "Unable to create marker metric: %v", err)
-					} else {
-						cm.output <- m
-					}
-				}
 			}
 		}
 	})
@@ -219,9 +199,9 @@ func (cm *collectorManager) Close() {
 }
 
 // New creates a new initialized metric collector manager
-func New(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) (CollectorManager, error) {
+func New(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) (CollectorManager, error) {
 	cm := new(collectorManager)
-	err := cm.Init(ticker, duration, enableMarkers, wg, collectConfig)
+	err := cm.Init(ticker, duration, wg, collectConfig)
 	if err != nil {
 		return nil, err
 	}

collectors/cpufreqCpuinfoMetric.go

@@ -116,9 +116,6 @@ func (m *CPUFreqCpuInfoCollector) Init(_ json.RawMessage) error {
 			physicalPackageID = ""
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		return fmt.Errorf("%s Init(): Call to scanner.Err failed: %w", m.name, err)
-	}
 
 	if err := file.Close(); err != nil {
 		return fmt.Errorf("%s Init(): Call to file.Close() failed: %w", m.name, err)
@@ -182,9 +179,6 @@ func (m *CPUFreqCpuInfoCollector) Read(interval time.Duration, output chan lp.CC
 			}
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		cclog.ComponentErrorf(m.name, "Read(): Call to scanner.Err failed: %s", err.Error())
-	}
 }
 
 func (m *CPUFreqCpuInfoCollector) Close() {

collectors/cpustatMetric.go

@@ -44,7 +44,7 @@ type CpustatCollector struct {
 func (m *CpustatCollector) Init(config json.RawMessage) error {
 	m.name = "CpustatCollector"
 	if err := m.setup(); err != nil {
-		return fmt.Errorf("%s Init(): setup() call failed: %s", m.name, err.Error())
+		return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
 	}
 	m.parallel = true
 	m.meta = map[string]string{
@@ -58,7 +58,7 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
 		d := json.NewDecoder(bytes.NewReader(config))
 		d.DisallowUnknownFields()
 		if err := d.Decode(&m.config); err != nil {
-			return fmt.Errorf("%s Init(): Error decoding JSON config: %s", m.name, err.Error())
+			return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
 		}
 	}
 	matches := map[string]int{
@@ -85,7 +85,7 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
 	// Check input file
 	file, err := os.Open(CPUSTATFILE)
 	if err != nil {
-		return fmt.Errorf("%s Init(): Failed to open file '%s': %s", m.name, CPUSTATFILE, err.Error())
+		return fmt.Errorf("%s Init(): Failed to open file '%s': %w", m.name, CPUSTATFILE, err)
 	}
 
 	// Pre-generate tags for all CPUs
@@ -117,13 +117,10 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
 			num_cpus++
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		return fmt.Errorf("%s Init(): Call to scanner.Err failed: %s", m.name, err.Error())
-	}
 
 	// Close file
 	if err := file.Close(); err != nil {
-		return fmt.Errorf("%s Init(): Failed to close file '%s': %s", m.name, CPUSTATFILE, err.Error())
+		return fmt.Errorf("%s Init(): Failed to close file '%s': %w", m.name, CPUSTATFILE, err)
 	}
 
 	m.lastTimestamp = time.Now()
@@ -176,13 +173,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.Error()))
+			fmt.Sprintf("Read(): Failed to open file '%s': %v", CPUSTATFILE, err))
 	}
 	defer func() {
 		if err := file.Close(); err != nil {
 			cclog.ComponentError(
 				m.name,
-				fmt.Sprintf("Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err.Error()))
+				fmt.Sprintf("Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err))
 		}
 	}()
 
@@ -197,9 +194,6 @@ func (m *CpustatCollector) Read(interval time.Duration, output chan lp.CCMessage
 			num_cpus++
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		cclog.ComponentErrorf(m.name, "Init(): Call to scanner.Err failed: %s", err.Error())
-	}
 
 	if !m.config.excludeNumCPUs {
 		if num_cpus_metric, err := lp.NewMetric("num_cpus", m.nodetags, m.meta, num_cpus, now); err == nil {

collectors/diskstatMetric.go

@@ -147,11 +147,6 @@ mountLoop:
 			}
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		cclog.ComponentErrorf(
-			m.name,
-			"Read(): Call to scanner.Err failed: %s", err.Error())
-	}
 	if m.allowedMetrics["part_max_used"] {
 		y, err := lp.NewMetric("part_max_used", map[string]string{"type": "node"}, m.meta, int(part_max_used), time.Now())
 		if err == nil {

collectors/iostatMetric.go

@@ -48,13 +48,13 @@ func (m *IOstatCollector) Init(config json.RawMessage) error {
 	m.parallel = true
 	m.meta = map[string]string{"source": m.name, "group": "Disk"}
 	if err := m.setup(); err != nil {
-		return fmt.Errorf("%s Init(): setup() call failed: %s", m.name, err.Error())
+		return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
 	}
 	if len(config) > 0 {
 		d := json.NewDecoder(bytes.NewReader(config))
 		d.DisallowUnknownFields()
 		if err := d.Decode(&m.config); err != nil {
-			return fmt.Errorf("%s Init(): Error decoding JSON config: %s", m.name, err.Error())
+			return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
 		}
 	}
 	// https://www.kernel.org/doc/html/latest/admin-guide/iostats.html
@@ -89,7 +89,7 @@ func (m *IOstatCollector) Init(config json.RawMessage) error {
 	}
 	file, err := os.Open(IOSTATFILE)
 	if err != nil {
-		return fmt.Errorf("%s Init(): Failed to open file \"%s\": %s", m.name, IOSTATFILE, err.Error())
+		return fmt.Errorf("%s Init(): Failed to open file \"%s\": %w", m.name, IOSTATFILE, err)
 	}
 
 	scanner := bufio.NewScanner(file)
@@ -130,11 +130,8 @@ func (m *IOstatCollector) Init(config json.RawMessage) error {
 			lastValues:    lastValues,
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		return fmt.Errorf("%s Init(): Failed to scan content of file %q: %s", m.name, IOSTATFILE, err.Error())
-	}
 	if err := file.Close(); err != nil {
-		return fmt.Errorf("%s Init(): Failed to close file \"%s\": %s", m.name, IOSTATFILE, err.Error())
+		return fmt.Errorf("%s Init(): Failed to close file \"%s\": %w", m.name, IOSTATFILE, err)
 	}
 
 	m.init = true
@@ -150,14 +147,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': %s", IOSTATFILE, err.Error()))
+			fmt.Sprintf("Read(): Failed to open file '%s': %v", IOSTATFILE, err))
 		return
 	}
 	defer func() {
 		if err := file.Close(); err != nil {
 			cclog.ComponentError(
 				m.name,
-				fmt.Sprintf("Read(): Failed to close file '%s': %s", IOSTATFILE, err.Error()))
+				fmt.Sprintf("Read(): Failed to close file '%s': %v", IOSTATFILE, err))
 		}
 	}()
 
@@ -201,11 +198,6 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMessage)
 		}
 		m.devices[device] = entry
 	}
-	if err := scanner.Err(); err != nil {
-		cclog.ComponentErrorf(
-			m.name,
-			"Read(): Failed to scan content of file %q: %s", IOSTATFILE, err.Error())
-	}
 }
 
 func (m *IOstatCollector) Close() {

collectors/ipmiMetric.go

@@ -28,15 +28,13 @@ type IpmiCollector struct {
 	metricCollector
 
 	config struct {
-		IpmitoolPath    string   `json:"ipmitool_path"`
-		IpmisensorsPath string   `json:"ipmisensors_path"`
-		Sudo            bool     `json:"use_sudo"`
-		IncludeMetrics  []string `json:"include_metrics"`
+		IpmitoolPath    string `json:"ipmitool_path"`
+		IpmisensorsPath string `json:"ipmisensors_path"`
+		Sudo            bool   `json:"use_sudo"`
 	}
 
-	ipmitool       string
-	ipmisensors    string
-	includeMetrics map[string]bool
+	ipmitool    string
+	ipmisensors string
 }
 
 func (m *IpmiCollector) Init(config json.RawMessage) error {
@@ -66,15 +64,6 @@ func (m *IpmiCollector) Init(config json.RawMessage) error {
 		}
 	}
 
-	// Read metrics to include
-	m.includeMetrics = make(map[string]bool)
-	for _, metric := range m.config.IncludeMetrics {
-		metric = strings.ToLower(strings.TrimSpace(metric))
-		if metric != "" {
-			m.includeMetrics[metric] = true
-		}
-	}
-
 	m.ipmitool = m.config.IpmitoolPath
 	m.ipmisensors = m.config.IpmisensorsPath
 
@@ -156,11 +145,6 @@ func (m *IpmiCollector) readIpmiTool(output chan lp.CCMessage) error {
 			continue
 		}
 		name := strings.ToLower(strings.ReplaceAll(strings.TrimSpace(lv[0]), " ", "_"))
-
-		if len(m.includeMetrics) > 0 && !m.includeMetrics[name] {
-			continue
-		}
-
 		unit := strings.TrimSpace(lv[2])
 		switch unit {
 		case "Volts":
@@ -181,9 +165,6 @@ func (m *IpmiCollector) readIpmiTool(output chan lp.CCMessage) error {
 		y.AddMeta("unit", unit)
 		output <- y
 	}
-	if err := scanner.Err(); err != nil {
-		return fmt.Errorf("failed to scan output of command: %s", err.Error())
-	}
 
 	// Wait for command end
 	if err := command.Wait(); err != nil {
@@ -228,11 +209,6 @@ func (m *IpmiCollector) readIpmiSensors(output chan lp.CCMessage) error {
 			continue
 		}
 		name := strings.ToLower(strings.ReplaceAll(lv[1], " ", "_"))
-
-		if len(m.includeMetrics) > 0 && !m.includeMetrics[name] {
-			continue
-		}
-
 		y, err := lp.NewMetric(name, map[string]string{"type": "node"}, m.meta, v, time.Now())
 		if err != nil {
 			cclog.ComponentErrorf(m.name, "Failed to create message: %v", err)

collectors/ipmiMetric.md

@@ -15,8 +15,7 @@ hugo_path: docs/reference/cc-metric-collector/collectors/ipmi.md
   "ipmistat": {
     "ipmitool_path": "/path/to/ipmitool",
     "ipmisensors_path": "/path/to/ipmi-sensors",
-    "use_sudo": true,
-    "include_metrics" : []
+    "use_sudo": true
   }
 ```
 
@@ -37,5 +36,3 @@ Defaults: monitoring !log_allowed, !pam_session
 monitoring ALL = (root) NOPASSWD:/usr/bin/ipmitool sensor
 monitoring ALL = (root) NOPASSWD:/usr/sbin/ipmi-sensors --comma-separated-output --sdr-cache-recreate
 ```
-
-If `include_ipmi_metrics` contains any entry, ipmistat collector will only submit these metrics. Any other values will get discarded. 

collectors/lenovoDensePower.go

@@ -1,170 +0,0 @@
-// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
-// All rights reserved.
-// Use of this source code is governed by a MIT-style
-// license that can be found in the LICENSE file.
-// additional authors:
-// Michael Panzlaff (NHR@FAU)
-
-package collectors
-
-import (
-	"bytes"
-	"encoding/hex"
-	"encoding/json"
-	"fmt"
-	"os/exec"
-	"strconv"
-	"strings"
-	"time"
-
-	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
-	lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
-)
-
-type LenovoDensePowerCollector struct {
-	metricCollector
-
-	config struct {
-		IpmitoolPath string `json:"ipmitool_path"`
-		Sudo         bool   `json:"use_sudo"`
-	}
-
-	ipmitool string
-
-	energyValLast  float64
-	energyTimeLast time.Time
-}
-
-func (m *LenovoDensePowerCollector) Init(config json.RawMessage) error {
-	// Check if already initialized
-	if m.init {
-		return nil
-	}
-
-	m.name = "LenovoDensePowerCollector"
-	if err := m.setup(); err != nil {
-		return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
-	}
-	m.meta = map[string]string{
-		"source": m.name,
-		"group":  "IPMI",
-	}
-
-	m.config.IpmitoolPath = "ipmitool"
-
-	if len(config) > 0 {
-		d := json.NewDecoder(bytes.NewReader(config))
-		d.DisallowUnknownFields()
-		if err := d.Decode(&m.config); err != nil {
-			return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
-		}
-	}
-
-	m.ipmitool = m.config.IpmitoolPath
-
-	energyVal, energyTime, err := m.readEnergy()
-	if err != nil {
-		return fmt.Errorf("energy reading test failed: %w", err)
-	}
-
-	m.energyValLast = energyVal
-	m.energyTimeLast = energyTime
-	m.init = true
-
-	return nil
-}
-
-func (m *LenovoDensePowerCollector) readEnergy() (float64, time.Time, error) {
-	argv := make([]string, 0)
-	if m.config.Sudo {
-		argv = append(argv, "sudo", "-n")
-	}
-
-	// The Lenovo hardware wants this magic byte string
-	lenovoRequestEnergyMsg := []uint8{0x3a, 0x32, 4, 2, 0, 0, 0}
-
-	argv = append(argv, m.ipmitool, "raw")
-
-	for _, val := range lenovoRequestEnergyMsg {
-		argv = append(argv, strconv.Itoa(int(val)))
-	}
-
-	cmd := exec.Command(argv[0], argv[1:]...)
-
-	var stdout bytes.Buffer
-	var stderr bytes.Buffer
-	cmd.Stdout = &stdout
-	cmd.Stderr = &stderr
-
-	err := cmd.Run()
-	if err != nil {
-		return 0, time.Unix(0, 0), fmt.Errorf("failed to run ipmitool: %w (stdout=%s stderr=%s)", err, stdout.String(), stderr.String())
-	}
-
-	data, err := hex.DecodeString(strings.ReplaceAll(strings.TrimSpace(stdout.String()), " ", ""))
-	if err != nil {
-		return 0, time.Unix(0, 0), fmt.Errorf("unable to decode ipmitool hex output: %w (stdout=%s)", err, stdout.String())
-	}
-
-	if len(data) != 14 {
-		return 0, time.Unix(0, 0), fmt.Errorf("result must be 14 bytes as specified in the documentation")
-	}
-
-	// data is laid out like this:
-	// data[0 to 1]: FIFO index of value (for debug only)
-	// data[2 to 5]: Integer part of energy in J (little endian)
-	// data[6 to 7]: Fraction part of energy in mJ (little endian)
-	// data[8 to 11]: Integer part of Unix time of measurement in seconds (little endian)
-	// data[12 to 13]: Fraction part of Unix time of measurement in milliseconds (little endian)
-	wholeJoules := (uint32(data[2]) << 0) | (uint32(data[3]) << 8) | (uint32(data[4]) << 16) | (uint32(data[5]) << 24)
-	milliJoules := uint16(data[6]) | (uint16(data[7]) << 8)
-	finalJoules := float64(wholeJoules) + float64(milliJoules)*1e-3
-
-	wholeSeconds := (uint32(data[8]) << 0) | (uint32(data[9]) << 8) | (uint32(data[10]) << 16) | (uint32(data[11]) << 24)
-	milliSeconds := uint16(data[12]) | (uint16(data[13]) << 8)
-	finalTime := time.Unix(int64(wholeSeconds), (int64(milliSeconds) * 1000000))
-
-	return finalJoules, finalTime, nil
-}
-
-func (m *LenovoDensePowerCollector) Read(interval time.Duration, output chan lp.CCMessage) {
-	// Check if already initialized
-	if !m.init {
-		return
-	}
-
-	energyVal, energyTime, err := m.readEnergy()
-	if err != nil {
-		cclog.ComponentErrorf(m.name, "readEnergy failed: %v", err)
-		return
-	}
-
-	powerVal := 0.0
-	if energyTime.Compare(m.energyTimeLast) != 0 {
-		// Check for overflow
-		energyValDiff := int64(energyVal) - int64(m.energyValLast)
-		if energyVal < m.energyValLast {
-			energyValDiff += int64(0x100000000)
-		}
-		energyTimeDiff := energyTime.Sub(m.energyTimeLast)
-		if energyTime.Before(m.energyTimeLast) {
-			energyTimeDiff = energyTimeDiff + 0x100000000*time.Second
-		}
-		powerVal = float64(energyValDiff) / energyTimeDiff.Seconds()
-
-		m.energyValLast = energyVal
-		m.energyTimeLast = energyTime
-	}
-
-	metric, err := lp.NewMetric("lenovo_node_power", map[string]string{"type": "node"}, m.meta, powerVal, energyTime)
-	if err != nil {
-		cclog.ComponentErrorf(m.name, "lp.NewMetric failed: %v", err)
-		return
-	}
-	metric.AddMeta("unit", "Watts")
-	output <- metric
-}
-
-func (m *LenovoDensePowerCollector) Close() {
-	m.init = false
-}

collectors/lenovoDensePower.md

@@ -1,51 +0,0 @@
-<!--
----
-title: Lenovo Dense Power collector
-description: Collect power of Lenovo Dense machines using ipmitool raw
-categories: [cc-metric-collector]
-tags: ['Admin']
-weight: 2
-hugo_path: docs/reference/cc-metric-collector/collectors/lenovoDensePower.md
----
--->
-
-## `lenovo_dense_power` collector
-
-```json
-  "lenovo_dense_power": {
-    "ipmitool_path": "/path/to/ipmitool",
-    "use_sudo": true,
-    "include_metrics" : []
-  }
-```
-
-The `lenovo_dense_power` collector reads power from Lenovo Dense machines via `ipmitool` (`ipmitool raw ...`). This collector is known to only work on the following machines. Others are not supported, so use at your *OWN* risk:
-
-System                                    | Compatibility             | Power measured
----                                       | ---                       | ---
-Lenovo ThinkSystem SD650 V2/V3            | untested, but should work | node + riser1 + riser2
-Lenovo ThinkSystem SD650-N V2/ SD650-I V3 | untested, but should work | node + riser1 + riser2 + gpus
-Lenovo ThinkSystem SD665-N V3             | tested, known to work     | node + riser1 + riser2 + gpus
-
-To test if your system *may* be supported, you can run the following command *at your own risk*:
-
-```
-$ ipmitool raw 58 50 4 2 0 0 0
- c4 00 83 a9 cb d3 a2 03 df db 47 6a d8 01
-```
-
-Exactly 14 bytes should be returned.
-In case not or another error occurs your hardware is likely not supported.
-
-In addition, `ipmitool` typically require root to run.
-In order to run `cc-metric-collector` without root priviliges, you can enable `use_sudo`.
-Add a file like this in `/etc/sudoers.d/` to allow `cc-metric-collector` to run the required commands:
-
-```
-# Do not log the following sudo commands from monitoring, since this causes a lot of log spam.
-# However keep log_denied enabled, to detect failures
-Defaults: monitoring !log_allowed, !pam_session
-
-# Allow to use ipmitool for Lenovo power readings
-monitoring ALL = (root) NOPASSWD:/usr/bin/ipmitool raw 58 50 4 2 0 0 0
-```

collectors/likwidMetric.go

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

collectors/megwareEureka.go

@@ -1,194 +0,0 @@
-// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
-// All rights reserved.
-// Use of this source code is governed by a MIT-style
-// license that can be found in the LICENSE file.
-// additional authors:
-// Michael Panzlaff (NHR@FAU)
-
-package collectors
-
-import (
-	"bytes"
-	"encoding/json"
-	"fmt"
-	"os/exec"
-	"time"
-
-	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
-	lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
-)
-
-// MPS refers to Monolithic Power Systems, from which the MP5922 is used
-// to measure telemetry on the Eureka platform.
-type mpsData struct {
-	Cnt         int64   `json:"Cnt"` // No idea what this is, ignore
-	Energy      float64 `json:"Energy"`
-	Vin         float64 `json:"Vin"`
-	Iin         float64 `json:"Iin"`
-	Pin         float64 `json:"Pin"`
-	PinAvg      float64 `json:"PinAvg"`
-	Vout        float64 `json:"Vout"`
-	Iout        float64 `json:"Iout"`
-	Pout        float64 `json:"Pout"`
-	StandbyVout float64 `json:"StandbyVout"`
-	StandbyIout float64 `json:"StandbyIout"`
-	StandbyPout float64 `json:"StandbyPout"`
-	TempBusbar  float64 `json:"TempBusbar"`
-	TempSsd     float64 `json:"TempSsd"`
-	TempMps     float64 `json:"TempMps"`
-	// No idea what the ones below mean exactl. Ignore them for now.
-	EnergyTime         int64 `json:"EnergyTime"`
-	EnergyAccumulator  int64 `json:"EnergyAccumulator"`
-	EnergyRolloverCnt  int64 `json:"EnergyRolloverCnt"`
-	EnergySampleCntU24 int64 `json:"EnergySampleCntU24"`
-
-	Timestamp time.Time
-}
-
-type MegwareEurekaCollector struct {
-	metricCollector
-
-	config struct {
-		U20Path string `json:"u20_path"`
-		Sudo    bool   `json:"use_sudo"`
-	}
-
-	u20path string
-
-	energyValLast  float64
-	energyTimeLast time.Time
-}
-
-func (m *MegwareEurekaCollector) Init(config json.RawMessage) error {
-	// Check if already initialized
-	if m.init {
-		return nil
-	}
-
-	m.name = "MegwareEurekaCollector"
-	if err := m.setup(); err != nil {
-		return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
-	}
-	m.meta = map[string]string{
-		"source": m.name,
-		"group":  "U20",
-	}
-
-	m.config.U20Path = "u20"
-
-	if len(config) > 0 {
-		d := json.NewDecoder(bytes.NewReader(config))
-		d.DisallowUnknownFields()
-		if err := d.Decode(&m.config); err != nil {
-			return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
-		}
-	}
-
-	m.u20path = m.config.U20Path
-
-	data, err := m.readMpsData()
-	if err != nil {
-		return fmt.Errorf("energy reading test failed: %w", err)
-	}
-
-	m.energyValLast = data.Energy
-	m.energyTimeLast = time.Now()
-	m.init = true
-
-	return nil
-}
-
-func (m *MegwareEurekaCollector) readMpsData() (*mpsData, error) {
-	argv := make([]string, 0)
-	if m.config.Sudo {
-		argv = append(argv, "sudo", "-n")
-	}
-
-	argv = append(argv, m.u20path, "values", "--read", "GET_MPS_POLL_VALUES")
-
-	cmd := exec.Command(argv[0], argv[1:]...)
-
-	var stdout bytes.Buffer
-	var stderr bytes.Buffer
-	cmd.Stdout = &stdout
-	cmd.Stderr = &stderr
-
-	err := cmd.Run()
-	if err != nil {
-		return nil, fmt.Errorf("failed to run u20: %w (stdout=%s stderr=%s)", err, stdout.String(), stderr.String())
-	}
-
-	var u20output struct {
-		GetMpsPollValues mpsData `json:"GET_MPS_POLL_VALUES"`
-	}
-
-	err = json.Unmarshal(stdout.Bytes(), &u20output)
-	if err != nil {
-		return nil, fmt.Errorf("unable to decode u20 JSON output: %w (stdout=%s)", err, stdout.String())
-	}
-
-	u20output.GetMpsPollValues.Timestamp = time.Now()
-
-	return &u20output.GetMpsPollValues, nil
-}
-
-func (m *MegwareEurekaCollector) Read(interval time.Duration, output chan lp.CCMessage) {
-	// Check if already initialized
-	if !m.init {
-		return
-	}
-
-	data, err := m.readMpsData()
-	if err != nil {
-		cclog.ComponentErrorf(m.name, "readMpsData failed: %v", err)
-		return
-	}
-
-	powerVal := 0.0
-
-	if data.Timestamp.After(m.energyTimeLast) {
-		// Important, m.energy comes in Wh, so multiply by 3600 to get Ws (aka Joule)
-		energyValDiff := data.Energy - m.energyValLast
-		energyTimeDiff := data.Timestamp.Sub(m.energyTimeLast)
-		powerVal = energyValDiff * 3600 / energyTimeDiff.Seconds()
-
-		m.energyValLast = data.Energy
-		m.energyTimeLast = data.Timestamp
-	}
-
-	metricNamePrefix := "eureka_"
-	metricMap := map[string]struct {
-		value any
-		unit  string
-	}{
-		"power":        {value: powerVal, unit: "Watts"},
-		"vin":          {value: data.Vin, unit: "Volts"},
-		"iin":          {value: data.Iin, unit: "Amperes"},
-		"pin":          {value: data.Pin, unit: "Watts"},
-		"pin_avg":      {value: data.PinAvg, unit: "Watts"},
-		"vout":         {value: data.Vout, unit: "Volts"},
-		"iout":         {value: data.Iout, unit: "Amperes"},
-		"pout":         {value: data.Pout, unit: "Watts"},
-		"standby_vout": {value: data.StandbyVout, unit: "Volts"},
-		"standby_iout": {value: data.StandbyIout, unit: "Amperes"},
-		"standby_pout": {value: data.StandbyPout, unit: "Watts"},
-		"temp_busbar":  {value: data.TempBusbar, unit: "degC"},
-		"temp_ssd":     {value: data.TempSsd, unit: "degC"},
-		"temp_mps":     {value: data.TempMps, unit: "degC"},
-	}
-
-	for metricName, metricData := range metricMap {
-		metricName = metricNamePrefix + metricName
-		metric, err := lp.NewMetric(metricName, map[string]string{"type": "node"}, m.meta, metricData.value, data.Timestamp)
-		if err != nil {
-			cclog.ComponentErrorf(m.name, "lp.NewMetric failed: %v", err)
-			return
-		}
-		metric.AddMeta("unit", metricData.unit)
-		output <- metric
-	}
-}
-
-func (m *MegwareEurekaCollector) Close() {
-	m.init = false
-}

collectors/megwareEureka.md

@@ -1,44 +0,0 @@
-<!--
----
-title: Megware Eureka collector
-description: Collect power and other metrics of Megware Eureka machines using u20
-categories: [cc-metric-collector]
-tags: ['Admin']
-weight: 2
-hugo_path: docs/reference/cc-metric-collector/collectors/lenovoDensePower.md
----
--->
-
-## `lenovo_dense_power` collector
-
-```json
-  "megware_eureka": {
-    "u20_path": "/path/to/ipmitool",
-    "use_sudo": true
-  }
-```
-
-The `megware_eureka` collector reads power and other metrics from Megware Eureka machines via `u20`.
-If the u20 tool is available for your machines, it's possible that this collector is compatible.
-If you don't have access to u20, please contact your Megware sales person.
-
-You can test the collector compatibility by running the following command:
-
-```
-$ u20 values --read GET_MPS_POLL_VALUES
-```
-
-If this returns a JSON with energy, voltage and current readings, you're fine.
-
-In addition, `u20` typically requires root to run.
-In order to run `cc-metric-collector` without root priviliges, you can enable `use_sudo`.
-Add a file like this in `/etc/sudoers.d/` to allow `cc-metric-collector` to run the required commands:
-
-```
-# Do not log the following sudo commands from monitoring, since this causes a lot of log spam.
-# However keep log_denied enabled, to detect failures
-Defaults: monitoring !log_allowed, !pam_session
-
-# Allow to use u20 for Megware power readings
-monitoring ALL = (root) NOPASSWD:/usr/bin/u20 values --read GET_MPS_POLL_VALUES
-```

collectors/raplMetric.go

@@ -22,15 +22,24 @@ import (
 )
 
 // running average power limit (RAPL) monitoring attributes for a zone
+// Only for Intel systems
+
 type RAPLZoneInfo struct {
+	energy            int64     // current reading of the energy counter in micro joules
+	maxEnergyRange    int64     // Range of the above energy counter in micro-joules
+	energyTimestamp   time.Time // timestamp when energy counter was read
+	energyFilepath    string    // path to a file containing the zones current energy counter in micro joules
+	shortTermFilepath string    // path to short term power limit
+	longTermFilepath  string    // path to long term power limit
+	enabledFilepath   string    // path to check whether limits are enabled
+	name              string
+
 	// tags describing the RAPL zone:
 	// * zone_name, subzone_name: e.g. psys, dram, core, uncore, package-0
 	// * zone_id: e.g. 0:1 (zone 0 sub zone 1)
-	tags            map[string]string
-	energyFilepath  string    // path to a file containing the zones current energy counter in micro joules
-	energy          int64     // current reading of the energy counter in micro joules
-	energyTimestamp time.Time // timestamp when energy counter was read
-	maxEnergyRange  int64     // Range of the above energy counter in micro-joules
+	// type=socket for dram, core, uncore, package-* and type=node for psys
+	// type-id=socket id
+	tags map[string]string
 }
 
 type RAPLCollector struct {
@@ -42,12 +51,40 @@ type RAPLCollector struct {
 		// * 0:1 for zone 0 subzone 1
 		ExcludeByID []string `json:"exclude_device_by_id,omitempty"`
 		// Exclude names for RAPL zones, e.g. psys, dram, core, uncore, package-0
-		ExcludeByName []string `json:"exclude_device_by_name,omitempty"`
+		ExcludeByName     []string `json:"exclude_device_by_name,omitempty"`
+		SkipEnergyReading bool     `json:"skip_energy_reading,omitempty"`
+		SkipLimitsReading bool     `json:"skip_limits_reading,omitempty"`
+		OnlyEnabledLimits bool     `json:"only_enabled_limits,omitempty"`
 	}
-	RAPLZoneInfo []RAPLZoneInfo
+	raplZoneInfo []RAPLZoneInfo
 	meta         map[string]string // default meta information
 }
 
+// Get the path to the power limit file for zone selectable by limit name
+// Common limit names for Intel systems are
+// - long_term
+// - short_term
+// Does not support AMD as AMD systems do not provide the power limits
+// through sysfs
+func ZoneLimitFile(folder string, limit_name string) string {
+	nameGlob := filepath.Join(folder, "constraint_*_name")
+	candidates, err := filepath.Glob(nameGlob)
+	if err == nil {
+		for _, c := range candidates {
+			if v, err := os.ReadFile(c); err == nil {
+				if strings.TrimSpace(string(v)) == limit_name {
+					var i int
+					n, err := fmt.Sscanf(filepath.Base(c), "constraint_%d_name", &i)
+					if err == nil && n == 1 {
+						return filepath.Join(folder, fmt.Sprintf("constraint_%d_power_limit_uw", i))
+					}
+				}
+			}
+		}
+	}
+	return ""
+}
+
 // Init initializes the running average power limit (RAPL) collector
 func (m *RAPLCollector) Init(config json.RawMessage) error {
 	// Check if already initialized
@@ -67,6 +104,9 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
 	}
 
 	// Read in the JSON configuration
+	m.config.SkipEnergyReading = false
+	m.config.SkipLimitsReading = false
+	m.config.OnlyEnabledLimits = true
 	if len(config) > 0 {
 		d := json.NewDecoder(bytes.NewReader(config))
 		d.DisallowUnknownFields()
@@ -93,49 +133,63 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
 	// See: https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#monitoring-attributes
 	readZoneInfo := func(zonePath string) (
 		z struct {
-			name            string    // zones name e.g. psys, dram, core, uncore, package-0
-			energyFilepath  string    // path to a file containing the zones current energy counter in micro joules
-			energy          int64     // current reading of the energy counter in micro joules
-			energyTimestamp time.Time // timestamp when energy counter was read
-			maxEnergyRange  int64     // Range of the above energy counter in micro-joules
-			ok              bool      // Are all information available?
+			name              string    // zones name e.g. psys, dram, core, uncore, package-0
+			energyFilepath    string    // path to a file containing the zones current energy counter in micro joules
+			energy            int64     // current reading of the energy counter in micro joules
+			energyTimestamp   time.Time // timestamp when energy counter was read
+			maxEnergyRange    int64     // Range of the above energy counter in micro-joules
+			shortTermFilepath string
+			longTermFilepath  string
+			enabledFilepath   string
 		},
 	) {
 		// zones name e.g. psys, dram, core, uncore, package-0
-		foundName := false
-		if v, err := os.ReadFile(
-			filepath.Join(zonePath, "name")); err == nil {
-			foundName = true
+
+		if v, err :=
+			os.ReadFile(
+				filepath.Join(zonePath, "name")); err == nil {
 			z.name = strings.TrimSpace(string(v))
 		}
 
-		// path to a file containing the zones current energy counter in micro joules
-		z.energyFilepath = filepath.Join(zonePath, "energy_uj")
-
-		// current reading of the energy counter in micro joules
-		foundEnergy := false
-		if v, err := os.ReadFile(z.energyFilepath); err == nil {
-			// timestamp when energy counter was read
-			z.energyTimestamp = time.Now()
-			if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
-				foundEnergy = true
-				z.energy = i
+		if !m.config.SkipEnergyReading {
+			// path to a file containing the zones current energy counter in micro joules
+			z.energyFilepath = filepath.Join(zonePath, "energy_uj")
+			// current reading of the energy counter in micro joules
+			if v, err := os.ReadFile(z.energyFilepath); err == nil {
+				if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
+					z.energy = i
+					// timestamp when energy counter was read
+					z.energyTimestamp = time.Now()
+				}
+			} else {
+				cclog.ComponentErrorf(m.name, "Cannot read energy file for %s: %v", z.name, err.Error())
 			}
+			// Range of the above energy counter in micro-joules
+			if v, err :=
+				os.ReadFile(
+					filepath.Join(zonePath, "max_energy_range_uj")); err == nil {
+				if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
+					z.maxEnergyRange = i
+				}
+			}
+		} else {
+			cclog.ComponentDebugf(m.name, "Energy readings for %s disabled", zonePath)
 		}
 
-		// Range of the above energy counter in micro-joules
-		foundMaxEnergyRange := false
-		if v, err := os.ReadFile(
-			filepath.Join(zonePath, "max_energy_range_uj")); err == nil {
-			if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
-				foundMaxEnergyRange = true
-				z.maxEnergyRange = i
+		if !m.config.SkipLimitsReading {
+			z.shortTermFilepath = ZoneLimitFile(zonePath, "short_term")
+			if _, err := os.Stat(z.shortTermFilepath); err != nil {
+				z.shortTermFilepath = ""
 			}
+			z.longTermFilepath = ZoneLimitFile(zonePath, "long_term")
+			if _, err := os.Stat(z.longTermFilepath); err != nil {
+				z.longTermFilepath = ""
+			}
+			z.enabledFilepath = filepath.Join(zonePath, "enabled")
+		} else {
+			cclog.ComponentDebugf(m.name, "Power limit readings for %s disabled", zonePath)
 		}
 
-		// Are all information available?
-		z.ok = foundName && foundEnergy && foundMaxEnergyRange
-
 		return
 	}
 
@@ -152,14 +206,36 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
 
 	for _, zonePath := range zonesPath {
 		zoneID := strings.TrimPrefix(zonePath, zonePrefix)
+
 		z := readZoneInfo(zonePath)
-		if z.ok &&
-			!isIDExcluded[zoneID] &&
+		if !isIDExcluded[zoneID] &&
 			!isNameExcluded[z.name] {
 
+			si := RAPLZoneInfo{
+				tags:              make(map[string]string),
+				energyFilepath:    z.energyFilepath,
+				energy:            z.energy,
+				energyTimestamp:   z.energyTimestamp,
+				maxEnergyRange:    z.maxEnergyRange,
+				shortTermFilepath: z.shortTermFilepath,
+				longTermFilepath:  z.longTermFilepath,
+				enabledFilepath:   z.enabledFilepath,
+				name:              z.name,
+			}
+			si.tags["type"] = "node"
+			si.tags["type-id"] = "0"
+			var pid int = 0
+			if strings.HasPrefix(z.name, "package-") {
+				n, err := fmt.Sscanf(z.name, "package-%d", &pid)
+				if err == nil && n == 1 {
+					si.tags["type-id"] = fmt.Sprintf("%d", pid)
+					si.tags["type"] = "socket"
+				}
+				si.name = "pkg"
+			}
 			// Add RAPL monitoring attributes for a zone
-			m.RAPLZoneInfo = append(
-				m.RAPLZoneInfo,
+			m.raplZoneInfo = append(
+				m.raplZoneInfo,
 				RAPLZoneInfo{
 					tags: map[string]string{
 						"id":        zoneID,
@@ -182,37 +258,40 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
 		for _, subZonePath := range subZonesPath {
 			subZoneID := strings.TrimPrefix(subZonePath, subZonePrefix)
 			sz := readZoneInfo(subZonePath)
+
 			if len(zoneID) > 0 && len(z.name) > 0 &&
-				sz.ok &&
 				!isIDExcluded[zoneID+":"+subZoneID] &&
 				!isNameExcluded[sz.name] {
-				m.RAPLZoneInfo = append(
-					m.RAPLZoneInfo,
+				m.raplZoneInfo = append(
+					m.raplZoneInfo,
 					RAPLZoneInfo{
 						tags: map[string]string{
 							"id":            zoneID + ":" + subZoneID,
 							"zone_name":     z.name,
 							"sub_zone_name": sz.name,
 						},
-						energyFilepath:  sz.energyFilepath,
-						energy:          sz.energy,
-						energyTimestamp: sz.energyTimestamp,
-						maxEnergyRange:  sz.maxEnergyRange,
+						energyFilepath:    sz.energyFilepath,
+						energy:            sz.energy,
+						energyTimestamp:   sz.energyTimestamp,
+						maxEnergyRange:    sz.maxEnergyRange,
+						shortTermFilepath: sz.shortTermFilepath,
+						longTermFilepath:  sz.longTermFilepath,
+						enabledFilepath:   sz.enabledFilepath,
+						name:              sz.name,
 					})
 			}
 		}
 	}
 
-	if m.RAPLZoneInfo == nil {
+	if m.raplZoneInfo == nil {
 		return fmt.Errorf("no running average power limit (RAPL) device found in %s", controlTypePath)
 	}
 
 	// Initialized
 	cclog.ComponentDebug(
 		m.name,
-		"initialized",
-		len(m.RAPLZoneInfo),
-		"zones with running average power limit (RAPL) monitoring attributes")
+		"initialized %d zones with running average power limit (RAPL) monitoring attributes",
+		len(m.raplZoneInfo))
 	m.init = true
 
 	return err
@@ -221,35 +300,89 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
 // Read reads running average power limit (RAPL) monitoring attributes for all initialized zones
 // See: https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#monitoring-attributes
 func (m *RAPLCollector) Read(interval time.Duration, output chan lp.CCMessage) {
-	for i := range m.RAPLZoneInfo {
-		p := &m.RAPLZoneInfo[i]
+	for i := range m.raplZoneInfo {
+		p := &m.raplZoneInfo[i]
 
-		// Read current value of the energy counter in micro joules
-		if v, err := os.ReadFile(p.energyFilepath); err == nil {
-			energyTimestamp := time.Now()
-			if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
-				energy := i
+		if !m.config.SkipEnergyReading {
+			// Read current value of the energy counter in micro joules
+			if v, err := os.ReadFile(p.energyFilepath); err == nil {
+				energyTimestamp := time.Now()
+				if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
+					energy := i
 
-				// Compute average power (Δ energy / Δ time)
-				energyDiff := energy - p.energy
-				if energyDiff < 0 {
-					// Handle overflow:
-					// ( p.maxEnergyRange - p.energy ) + energy
-					// = p.maxEnergyRange + ( energy - p.energy )
-					// = p.maxEnergyRange + diffEnergy
-					energyDiff += p.maxEnergyRange
+					// Compute average power (Δ energy / Δ time)
+					energyDiff := energy - p.energy
+					if energyDiff < 0 {
+						// Handle overflow:
+						// ( p.maxEnergyRange - p.energy ) + energy
+						// = p.maxEnergyRange + ( energy - p.energy )
+						// = p.maxEnergyRange + diffEnergy
+						energyDiff += p.maxEnergyRange
+					}
+					timeDiff := energyTimestamp.Sub(p.energyTimestamp)
+					averagePower := float64(energyDiff) / float64(timeDiff.Microseconds())
+
+					y, err := lp.NewMetric(
+						fmt.Sprintf("rapl_%s_average_power", p.name),
+						p.tags,
+						m.meta,
+						averagePower,
+						energyTimestamp)
+					if err == nil {
+						output <- y
+					}
+
+					e, err := lp.NewMetric(
+						fmt.Sprintf("rapl_%s_energy", p.name),
+						p.tags,
+						m.meta,
+						float64(energyDiff)*1e-3,
+						energyTimestamp)
+					if err == nil {
+						e.AddMeta("unit", "Joules")
+						output <- e
+					}
+
+					// Save current energy counter state
+					p.energy = energy
+					p.energyTimestamp = energyTimestamp
 				}
-				timeDiff := energyTimestamp.Sub(p.energyTimestamp)
-				averagePower := float64(energyDiff) / float64(timeDiff.Microseconds())
-
-				y, err := lp.NewMetric("rapl_average_power", p.tags, m.meta, averagePower, energyTimestamp)
-				if err == nil {
-					output <- y
+			}
+		}
+		// https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#constraints
+		if !m.config.SkipLimitsReading {
+			skip := false
+			if m.config.OnlyEnabledLimits {
+				if v, err := os.ReadFile(p.enabledFilepath); err == nil {
+					if strings.TrimSpace(string(v)) == "0" {
+						skip = true
+					}
+				}
+			}
+			if !skip {
+				if len(p.shortTermFilepath) > 0 {
+					if v, err := os.ReadFile(p.shortTermFilepath); err == nil {
+						if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
+							name := fmt.Sprintf("rapl_%s_limit_short_term", p.name)
+							y, err := lp.NewMetric(name, p.tags, m.meta, i/1e6, time.Now())
+							if err == nil {
+								output <- y
+							}
+						}
+					}
 				}
 
-				// Save current energy counter state
-				p.energy = energy
-				p.energyTimestamp = energyTimestamp
+				if len(p.longTermFilepath) > 0 {
+					if v, err := os.ReadFile(p.longTermFilepath); err == nil {
+						if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
+							name := fmt.Sprintf("rapl_%s_limit_long_term", p.name)
+							y, err := lp.NewMetric(name, p.tags, m.meta, i/1e6, time.Now())
+							if err == nil {
+								output <- y
+							}
+						}
+					}
+				}
 			}
 		}
 	}

collectors/raplMetric.md

@@ -11,16 +11,24 @@ hugo_path: docs/reference/cc-metric-collector/collectors/rapl.md
 
 ## `rapl` collector
 
-This collector reads running average power limit (RAPL) monitoring attributes to compute average power consumption metrics. See <https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#monitoring-attributes>.
-
-The Likwid metric collector provides similar functionality.
+This collector reads running average power limit (RAPL) monitoring attributes to compute average power consumption metrics. See <https://www.kernel.org/doc/html/latest/power/powercap/powercap.html>.
 
 ```json
   "rapl": {
     "exclude_device_by_id": ["0:1", "0:2"],
-    "exclude_device_by_name": ["psys"]
+    "exclude_device_by_name": ["psys"],
+    "skip_energy_reading": false,
+    "skip_limits_reading": false,
+    "only_enabled_limits": true
   }
 ```
 
 Metrics:
-* `rapl_average_power`: average power consumption in Watt. The average is computed over the entire runtime from the last measurement to the current measurement
+* `rapl_<domain>_average_power`: average power consumption in Watt. The average is computed over the entire runtime from the last measurement to the current measurement
+* `rapl_<domain>_energy`: Difference from the last measurement
+* `rapl_<domain>_limit_short_term`: Short term powercap setting for the domain
+* `rapl_<domain>_limit_long_term`: Long term powercap setting for the domain
+
+Only the `rapl_<domain>_average_power` and `rapl_<domain>_energy` metrics require root-permissions. The limits can be read as user. Some domains have limits available but they are not enabled. By default, only enabled domain limits are collected.
+
+Energy and power measurments can also be done with the Likwid metric collector.

collectors/slurmCgroupMetric.go

@@ -7,7 +7,6 @@ import (
 	"os/exec"
 	"os/user"
 	"path/filepath"
-	"regexp"
 	"strconv"
 	"strings"
 	"time"
@@ -46,37 +45,6 @@ type SlurmCgroupCollector struct {
 
 const defaultCgroupBase = "/sys/fs/cgroup/system.slice/slurmstepd.scope"
 
-var (
-	// Slurm cgroup v2 directory layout:
-	// - Slurm <= 25.11: job_<numeric job id>
-	// - Slurm >= 26.05: SLUID, encoded as "s" + 13 Crockford Base32 characters
-	jobIDDirRE = regexp.MustCompile(`^job_[0-9]+$`)
-	sluidDirRE = regexp.MustCompile(`(?i)^s[0-9A-HJKMNP-TV-Z]{13}$`)
-)
-
-func (m *SlurmCgroupCollector) findSlurmJobDirs() ([]string, error) {
-	entries, err := os.ReadDir(m.cgroupBase)
-	if err != nil {
-		return nil, err
-	}
-
-	jobDirs := make([]string, 0)
-
-	for _, entry := range entries {
-		if !entry.IsDir() {
-			continue
-		}
-
-		name := entry.Name()
-
-		if jobIDDirRE.MatchString(name) || sluidDirRE.MatchString(name) {
-			jobDirs = append(jobDirs, filepath.Join(m.cgroupBase, name))
-		}
-	}
-
-	return jobDirs, nil
-}
-
 func ParseCPUs(cpuset string) ([]int, error) {
 	var result []int
 	if cpuset == "" {
@@ -269,9 +237,10 @@ func (m *SlurmCgroupCollector) Read(interval time.Duration, output chan lp.CCMes
 		delete(m.cpuUsed, k)
 	}
 
-	jobDirs, err := m.findSlurmJobDirs()
+	globPattern := filepath.Join(m.cgroupBase, "job_*")
+	jobDirs, err := filepath.Glob(globPattern)
 	if err != nil {
-		cclog.ComponentError(m.name, "Error reading job directories:", err.Error())
+		cclog.ComponentError(m.name, "Error globbing job directories:", err.Error())
 		return
 	}
 

example-configs/config.json

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

go.mod

@@ -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.3-1
+	github.com/NVIDIA/go-nvml v0.13.0-1
 	github.com/PaesslerAG/gval v1.2.4
 	github.com/fsnotify/fsnotify v1.10.1
 	github.com/tklauser/go-sysconf v0.4.0
-	golang.design/x/runtime v0.3.0
-	golang.org/x/sys v0.47.0
+	golang.design/x/thread v0.3.2
+	golang.org/x/sys v0.45.0
 )
 
 require (

go.sum

@@ -12,9 +12,8 @@ github.com/Microsoft/go-winio v0.6.1 h1:9/kr64B9VUZrLm5YYwbGtUJnMgqWVOdUAXu6Migc
 github.com/Microsoft/go-winio v0.6.1/go.mod h1:LRdKpFKfdobln8UmuiYcKPot9D2v6svN5+sAH+4kjUM=
 github.com/Microsoft/hcsshim v0.11.4 h1:68vKo2VN8DE9AdN4tnkWnmdhqdbpUFM8OF3Airm7fz8=
 github.com/Microsoft/hcsshim v0.11.4/go.mod h1:smjE4dvqPX9Zldna+t5FG3rnoHhaB7QYxPRqGcpAD9w=
+github.com/NVIDIA/go-nvml v0.13.0-1 h1:OLX8Jq3dONuPOQPC7rndB6+iDmDakw0XTYgzMxObkEw=
 github.com/NVIDIA/go-nvml v0.13.0-1/go.mod h1:+KNA7c7gIBH7SKSJ1ntlwkfN80zdx8ovl4hrK3LmPt4=
-github.com/NVIDIA/go-nvml v0.13.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=
@@ -174,8 +173,8 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
 go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
-golang.design/x/runtime v0.3.0 h1:8bA+GQfO8A18JXJQlQA56pEc+Sgrbo4cmLzjzmJ58ZI=
-golang.design/x/runtime v0.3.0/go.mod h1:PjVKQjImLdSrvjIw4FmrnCPq9BGs5PMBHb8i6b2BN9I=
+golang.design/x/thread v0.3.2 h1:FmD1glspGrQCe6FuQLmSrT6wz2CSzq7vKVDluyiMnqo=
+golang.design/x/thread v0.3.2/go.mod h1:6+Hi2rMOgMHZdKDWaqNHyWtoFUx1HxZ06LfHPh5Z/hQ=
 golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI=
 golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q=
 golang.org/x/exp v0.0.0-20231005195138-3e424a577f31 h1:9k5exFQKQglLo+RoP+4zMjOFE14P6+vyR0baDAi0Rcs=
@@ -184,8 +183,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.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
-golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
+golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
 golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
 golang.org/x/tools v0.14.0 h1:jvNa2pY0M4r62jkRQ6RwEZZyPcymeL9XZMLBbV7U2nc=

internal/metricAggregator/metricAggregator.go

@@ -69,8 +69,8 @@ var metricCacheLanguage = gval.NewLanguage(
 	gval.Function("getNumaCpuList", getCpuListOfNumaDomainFunc),
 	gval.Function("getDieCpuList", getCpuListOfDieFunc),
 	gval.Function("getCoreCpuList", getCpuListOfCoreFunc),
-	gval.Function("getCpuList", getCpuListOfNodeFunc),
-	gval.Function("getCpuListOfType", getCpuListOfTypeFunc),
+	gval.Function("getCpuList", getCpuListOfNode),
+	gval.Function("getCpuListOfType", getCpuListOfType),
 )
 
 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.ComponentDebugf("MetricCache", "COLLECT %s COND '%s'", f.Name, f.Condition)
+		cclog.ComponentDebug("MetricCache", "COLLECT", f.Name, "COND", f.Condition)
 		var valuesFloat64 []float64
 		var valuesFloat32 []float32
 		var valuesInt []int
@@ -140,7 +140,7 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
 			vars["metric"] = m
 			value, err := f.gvalCond.EvalBool(context.Background(), vars)
 			if err != nil {
-				cclog.ComponentErrorf("MetricCache", "COLLECT %s COND '%s' : %s", f.Name, f.Condition, err.Error())
+				cclog.ComponentError("MetricCache", "COLLECT", f.Name, "COND", f.Condition, ":", err.Error())
 				continue
 			}
 			if value {
@@ -160,7 +160,7 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
 					case bool:
 						valuesBool = append(valuesBool, x)
 					default:
-						cclog.ComponentErrorf("MetricCache", "COLLECT ADD VALUE %v FAILED", v)
+						cclog.ComponentError("MetricCache", "COLLECT ADD VALUE", v, "FAILED")
 					}
 				}
 				matches = append(matches, m)
@@ -213,13 +213,13 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
 			vars["values"] = valuesBool
 			len_values = len(valuesBool)
 		}
-		cclog.ComponentDebugf("MetricCache", "EVALUATE %s METRICS %d CALC '%s'", f.Name, len_values, f.Function)
+		cclog.ComponentDebug("MetricCache", "EVALUATE", f.Name, "METRICS", len_values, "CALC", f.Function)
 
 		vars["metrics"] = matches
 		if len_values > 0 {
 			value, err := gval.Evaluate(f.Function, vars, c.language)
 			if err != nil {
-				cclog.ComponentErrorf("MetricCache", "EVALUATE %s METRICS %d CALC '%s': %s", f.Name, len_values, f.Function, err.Error())
+				cclog.ComponentError("MetricCache", "EVALUATE", f.Name, "METRICS", len_values, "CALC", f.Function, ":", err.Error())
 				break
 			}
 
@@ -273,12 +273,12 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
 			case string:
 				m, err = lp.NewMessage(f.Name, tags, meta, map[string]any{"value": t}, starttime)
 			default:
-				cclog.ComponentErrorf("MetricCache", "Gval returned invalid type %s skipping metric %s", t, f.Name)
+				cclog.ComponentError("MetricCache", "Gval returned invalid type", t, "skipping metric", f.Name)
 			}
 			if err != nil {
-				cclog.ComponentErrorf("MetricCache", "Cannot create metric from Gval result %v: %s", value, err.Error())
+				cclog.ComponentError("MetricCache", "Cannot create metric from Gval result", value, ":", err.Error())
 			}
-			cclog.ComponentDebugf("MetricCache", "SEND %s", m.ToLineProtocol(nil))
+			cclog.ComponentDebug("MetricCache", "SEND", m)
 			select {
 			case c.output <- m:
 			default:
@@ -295,12 +295,12 @@ func (c *metricAggregator) AddAggregation(name, function, condition string, tags
 	newcond := strings.ReplaceAll(condition, "'", "\"")
 	gvalCond, err := gval.Full(metricCacheLanguage).NewEvaluable(newcond)
 	if err != nil {
-		cclog.ComponentErrorf("MetricAggregator", "Cannot add aggregation, invalid if condition '%s': %s", newcond, err.Error())
+		cclog.ComponentError("MetricAggregator", "Cannot add aggregation, invalid if condition", newcond, ":", err.Error())
 		return err
 	}
 	gvalFunc, err := gval.Full(metricCacheLanguage).NewEvaluable(newfunc)
 	if err != nil {
-		cclog.ComponentErrorf("MetricAggregator", "Cannot add aggregation, invalid function condition %s: %s", newfunc, err.Error())
+		cclog.ComponentError("MetricAggregator", "Cannot add aggregation, invalid function condition", newfunc, ":", err.Error())
 		return err
 	}
 	for _, agg := range c.functions {

internal/metricAggregator/metricAggregatorExpr.go

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

internal/metricAggregator/metricAggregatorExpr_test.go

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

internal/metricAggregator/metricAggregatorFunctions.go

@@ -336,14 +336,14 @@ func getCpuListOfDieFunc(args any) (any, error) {
 }
 
 // wrapper function to get a list of all cpuids of the node
-func getCpuListOfNodeFunc() (any, error) {
+func getCpuListOfNode() (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 getCpuListOfTypeFunc(args ...any) (any, error) {
+func getCpuListOfType(args ...any) (any, error) {
 	cpulist := make([]int, 0)
 	switch typ := args[0].(type) {
 	case string:

internal/metricRouter/metricCache.go

@@ -9,8 +9,6 @@ package metricRouter
 
 import (
 	"fmt"
-	"math"
-	"strings"
 	"sync"
 	"time"
 
@@ -21,107 +19,57 @@ import (
 	mct "github.com/ClusterCockpit/cc-metric-collector/pkg/multiChanTicker"
 )
 
-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
+type metricCachePeriod struct {
+	startstamp  time.Time
+	stopstamp   time.Time
+	numMetrics  int
+	sizeMetrics int
+	metrics     []lp.CCMessage
 }
 
 // Metric cache data structure
 type metricCache struct {
-	cache      CCCache
+	numPeriods int
+	curPeriod  int
+	lock       sync.Mutex
+	intervals  []*metricCachePeriod
 	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, interval time.Duration, numPeriods int) error
+	Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, 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, interval time.Duration, numPeriods int) error {
+func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, 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
-
-	err = c.cache.Init(numPeriods)
-	if err != nil {
-		return fmt.Errorf("MetricCache: failed to create cache: %w", err)
+	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)
 	}
 
-	c.aggEngine, err = agg.NewAggregatorExpr(c.output)
+	// 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)
 	if err != nil {
 		return fmt.Errorf("MetricCache: failed to create aggregator: %w", err)
 	}
@@ -133,58 +81,68 @@ 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)
+	}
 
-	c.wg.Add(1)
-	go func() {
+	// 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() {
 		for {
 			select {
 			case <-c.done:
-				c.wg.Done()
-				close(c.done)
-				cclog.ComponentDebug("MetricCache", "DONE")
-
+				done()
 				return
 			case tick := <-c.tickchan:
-				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)
-					})
+				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)
 				} else {
 					// This message is also printed in the first interval after startup
 					cclog.ComponentDebug("MetricCache", "EMPTY INTERVAL?")
 				}
-
 			}
 		}
-	}()
-
-	cclog.ComponentDebug("MetricCache", "STARTED")
+	})
+	cclog.ComponentDebug("MetricCache", "START")
 }
 
 // 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) {
-	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)
+	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()
 	}
 }
 
@@ -196,20 +154,40 @@ 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")
-	if c.started {
-		c.done <- true
-		c.wg.Wait()
-
-	}
-	cclog.ComponentDebug("MetricCache", "CLOSED")
+	c.done <- true
 }
 
 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, ticker.GetDuration(), numPeriods)
+	err := c.Init(output, ticker, wg, numPeriods)
 	if err != nil {
 		return nil, err
 	}

internal/metricRouter/metricCache_test.go

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

internal/metricRouter/metricRouter.go

@@ -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.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"`
+	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"`
 }
 
 // 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 && m != nil {
+		if r.config.NumCacheIntervals > 0 {
 			r.cache.Add(m)
 		}
 	}

pkg/multiChanTicker/multiChanTicker.go

@@ -17,20 +17,17 @@ 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)
@@ -56,10 +53,6 @@ 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)
 }