Compare commits

..

47 Commits

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

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

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-07 21:18:16 +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 Panzlaff 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 Panzlaff 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
brinkcoder aaabcc639f fix: support SLUID-based Slurm cgroup directories 2026-06-26 12:22:08 +01:00
Michael Schwarz 14d624fd22 Logic error, move creation of map upwards 2026-06-22 15:08:43 +01:00
Michael Schwarz 9677a916be Use map of included metrics instead of iterating through the config again and again 2026-06-22 15:08:43 +01:00
Michael Schwarz 8b373cc2c0 Add filter for ipmi metrics 2026-06-22 15:08:43 +01:00
dependabot[bot] 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 Obermaier 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] 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
dependabot[bot] 077204d39f Bump github.com/tklauser/go-sysconf from 0.3.16 to 0.4.0
Bumps [github.com/tklauser/go-sysconf](https://github.com/tklauser/go-sysconf) from 0.3.16 to 0.4.0.
- [Release notes](https://github.com/tklauser/go-sysconf/releases)
- [Commits](https://github.com/tklauser/go-sysconf/compare/v0.3.16...v0.4.0)

---
updated-dependencies:
- dependency-name: github.com/tklauser/go-sysconf
  dependency-version: 0.4.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2026-05-25 12:17:29 +02:00
dependabot[bot] dcc9746df4 Bump golang.org/x/sys from 0.43.0 to 0.45.0
Bumps [golang.org/x/sys](https://github.com/golang/sys) from 0.43.0 to 0.45.0.
- [Commits](https://github.com/golang/sys/compare/v0.43.0...v0.45.0)

---
updated-dependencies:
- dependency-name: golang.org/x/sys
  dependency-version: 0.45.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2026-05-25 12:10:50 +02:00
dependabot[bot] 2c51a3ed72 Bump github.com/fsnotify/fsnotify from 1.10.0 to 1.10.1
Bumps [github.com/fsnotify/fsnotify](https://github.com/fsnotify/fsnotify) from 1.10.0 to 1.10.1.
- [Release notes](https://github.com/fsnotify/fsnotify/releases)
- [Changelog](https://github.com/fsnotify/fsnotify/blob/main/CHANGELOG.md)
- [Commits](https://github.com/fsnotify/fsnotify/compare/v1.10.0...v1.10.1)

---
updated-dependencies:
- dependency-name: github.com/fsnotify/fsnotify
  dependency-version: 1.10.1
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2026-05-25 12:10:16 +02:00
Holger Obermaier 656ea73d12 Fix: num_cpus could not be excluded 2026-05-07 14:47:23 +02:00
Holger Obermaier 330f923596 Fixed exclude_metrics and check for used metrics 2026-05-07 12:25:07 +02:00
Holger Obermaier 8e58072ff6 Use NewMetric to create a new metric 2026-05-06 13:22:02 +02:00
Holger Obermaier 0f6fee9db4 Do not save current state of infiniband counters, only last state is required 2026-05-06 10:42:06 +02:00
Holger Obermaier 7585ee7289 Add bandwidth metrics for ib_total and ib_total_pkts 2026-05-05 14:13:38 +02:00
Michael Panzlaff 30b2eb69dd Merge pull request #213 from ClusterCockpit/fix/libdrm-ubuntu-deb
CI: Install libdrm-dev for building (required on Ubuntu)
2026-05-04 14:30:44 +02:00
Michael Panzlaff 2a51bd17f3 CI: Install libdrm-dev for building (required on Ubuntu) 2026-05-04 14:17:59 +02:00
dependabot[bot] 34d3d8970e Bump github.com/fsnotify/fsnotify from 1.9.0 to 1.10.0
Bumps [github.com/fsnotify/fsnotify](https://github.com/fsnotify/fsnotify) from 1.9.0 to 1.10.0.
- [Release notes](https://github.com/fsnotify/fsnotify/releases)
- [Changelog](https://github.com/fsnotify/fsnotify/blob/main/CHANGELOG.md)
- [Commits](https://github.com/fsnotify/fsnotify/compare/v1.9.0...v1.10.0)

---
updated-dependencies:
- dependency-name: github.com/fsnotify/fsnotify
  dependency-version: 1.10.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2026-05-04 13:30:59 +02:00
dependabot[bot] 50c7eba192 Bump github.com/ClusterCockpit/cc-lib/v2 from 2.11.0 to 2.12.0
Bumps [github.com/ClusterCockpit/cc-lib/v2](https://github.com/ClusterCockpit/cc-lib) from 2.11.0 to 2.12.0.
- [Release notes](https://github.com/ClusterCockpit/cc-lib/releases)
- [Commits](https://github.com/ClusterCockpit/cc-lib/compare/v2.11.0...v2.12.0)

---
updated-dependencies:
- dependency-name: github.com/ClusterCockpit/cc-lib/v2
  dependency-version: 2.12.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2026-04-27 12:13:32 +02:00
Michael Panzlaff d215cabb3e rpm: The installed binary is not secret and should be rx'able 2026-04-13 12:28:48 +02:00
Michael Panzlaff 86da3c15f7 rpm: The main binary should be owner by root
The system user should not be allowed to modify the ccmc binary.
2026-04-08 16:46:19 +02:00
Michael Panzlaff 93cd397b79 Revert "rpm: chown on /usr/bin/cc-metric-collector is unnecessary"
This reverts commit 65b9c0ea14.
2026-04-08 16:45:57 +02:00
Michael Panzlaff 65b9c0ea14 rpm: chown on /usr/bin/cc-metric-collector is unnecessary
The file belongs to root otherwise. The monitoring user can already
execute it. The monitoring user should not be allowed to change the
file, which is slightly more restricting. However it is in line with
what 99.9% of packages will do.
2026-04-08 15:56:11 +02:00
dependabot[bot] 0ecf06cee7 Bump github.com/ClusterCockpit/go-rocm-smi from 0.3.0 to 0.4.0
Bumps [github.com/ClusterCockpit/go-rocm-smi](https://github.com/ClusterCockpit/go-rocm-smi) from 0.3.0 to 0.4.0.
- [Commits](https://github.com/ClusterCockpit/go-rocm-smi/compare/v0.3...v0.4.0)

---
updated-dependencies:
- dependency-name: github.com/ClusterCockpit/go-rocm-smi
  dependency-version: 0.4.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2026-03-30 13:08:36 +02:00
Thomas Roehl 9eaf77db4f Update README.md 2026-03-24 19:50:51 +01:00
Thomas Roehl 7cb5d1b47a Add/update sudo configuration to all collectors with 'use_sudo' 2026-03-24 19:50:41 +01:00
Thomas Roehl 319e71a853 IpmiCollector: Remove unused configuration 'exclude_devices' 2026-03-24 19:48:34 +01:00
47 changed files with 1367 additions and 415 deletions
+2 -2
View File
@@ -270,7 +270,7 @@ jobs:
- name: Install development packages
run: |
apt update && apt --assume-yes upgrade
apt --assume-yes install build-essential sed git wget bash
apt --assume-yes install build-essential sed git wget bash libdrm-dev
# Checkout git repository and submodules
# fetch-depth must be 0 to use git describe
# See: https://github.com/marketplace/actions/checkout
@@ -321,7 +321,7 @@ jobs:
- name: Install development packages
run: |
apt update && apt --assume-yes upgrade
apt --assume-yes install build-essential sed git wget bash
apt --assume-yes install build-essential sed git wget bash libdrm-dev
# Checkout git repository and submodules
# fetch-depth must be 0 to use git describe
# See: https://github.com/marketplace/actions/checkout
+4
View File
@@ -17,3 +17,7 @@
# Local copy of LIKWID headers
/collectors/likwid
# Local development workspace (build against a local cc-lib checkout)
go.work
go.work.sum
+9 -9
View File
@@ -13,11 +13,7 @@ hugo_path: docs/reference/cc-metric-collector/_index.md
A node agent for measuring, processing and forwarding node level metrics. It is part of the [ClusterCockpit ecosystem](https://clustercockpit.org/docs/overview/).
The metric collector sends (and receives) metric in the [InfluxDB line protocol](https://docs.influxdata.com/influxdb/cloud/reference/syntax/line-protocol/) as it provides flexibility while providing a separation between tags (like index columns in relational databases) and fields (like data columns).
There is a single timer loop that triggers all collectors serially, collects the collectors' data and sends the metrics to the sink. This is done as all data is submitted with a single time stamp. The sinks currently use mostly blocking APIs.
The receiver runs as a go routine side-by-side with the timer loop and asynchronously forwards received metrics to the sink.
The `cc-metric-collector` sends (and maybe receives) metrics in the [InfluxDB line protocol](https://docs.influxdata.com/influxdb/cloud/reference/syntax/line-protocol/) as it provides flexibility while providing a separation between tags (like index columns in relational databases) and fields (like data columns). The `cc-metric-collector` consists of 4 components: collectors, router, sinks and receivers. The collectors read data from the current system and submit metrics to the router. The router can be configured to manipulate the metrics before forwarding them to the sinks. The receivers are also attached to the router like the collectors but they receive data from external source like other `cc-metric-collector` instances.
[![DOI](https://zenodo.org/badge/DOI/10.5281/zenodo.7438287.svg)](https://doi.org/10.5281/zenodo.7438287)
@@ -43,7 +39,9 @@ There is a main configuration file with basic settings that point to the other c
}
```
The `interval` defines how often the metrics should be read and send to the sink. The `duration` tells collectors how long one measurement has to take. This is important for some collectors, like the `likwid` collector. For more information, see [here](./docs/configuration.md).
The `interval` defines how often the metrics should be read and send to the sink(s). The `duration` tells the collectors how long one measurement has to take. This is important for some collectors, like the `likwid` collector. For more information, see [here](./docs/configuration.md).
The optional `channel_buffer_size` sets the capacity of the internal channels between the components. If unset, it defaults to `max(200, 24 * number of CPUs)` so that one interval's burst of per-hwthread metrics fits without back-pressuring the collectors on nodes with many cores.
See the component READMEs for their configuration:
@@ -57,7 +55,7 @@ See the component READMEs for their configuration:
```
$ git clone git@github.com:ClusterCockpit/cc-metric-collector.git
$ make (downloads LIKWID, builds it as static library with 'direct' accessmode and copies all required files for the collector)
$ go get (requires at least golang 1.16)
$ go get
$ make
```
@@ -67,11 +65,13 @@ For more information, see [here](./docs/building.md).
```
$ ./cc-metric-collector --help
Usage of metric-collector:
Usage of ./cc-metric-collector:
-config string
Path to configuration file (default "./config.json")
-log string
Path for logfile (default "stderr")
-loglevel string
Set log level (default "info")
-once
Run all collectors only once
```
@@ -114,7 +114,7 @@ flowchart TD
# Contributing
The ClusterCockpit ecosystem is designed to be used by different HPC computing centers. Since configurations and setups differ between the centers, the centers likely have to put some work into the cc-metric-collector to gather all desired metrics.
The ClusterCockpit ecosystem is designed to be used by different HPC computing centers. Since configurations and setups differ between the centers, the centers likely have to put some work into `cc-metric-collector` to gather all desired metrics.
You are free to open an issue to request a collector but we would also be happy about PRs.
+13 -3
View File
@@ -13,6 +13,7 @@ import (
"flag"
"os"
"os/signal"
"runtime"
"sync"
"syscall"
"time"
@@ -31,6 +32,7 @@ import (
type CentralConfigFile struct {
Interval string `json:"interval"`
Duration string `json:"duration"`
ChannelBufferSize int `json:"channel_buffer_size,omitempty"`
}
type RuntimeConfig struct {
@@ -158,6 +160,14 @@ func mainFunc() int {
return 1
}
// Size the channels between the managers so that one interval's burst of
// per-hwthread metrics fits without back-pressuring the collectors
chanSize := rcfg.ConfigFile.ChannelBufferSize
if chanSize <= 0 {
chanSize = max(200, 24*runtime.NumCPU())
}
cclog.ComponentDebug("main", "channel buffer size", chanSize)
routerConf := ccconf.GetPackageConfig("router")
if len(routerConf) == 0 {
cclog.Error("Metric router configuration file must be set")
@@ -194,7 +204,7 @@ func mainFunc() int {
}
// Connect metric router to sink manager
RouterToSinksChannel := make(chan lp.CCMessage, 200)
RouterToSinksChannel := make(chan lp.CCMessage, chanSize)
rcfg.SinkManager.AddInput(RouterToSinksChannel)
rcfg.MetricRouter.AddOutput(RouterToSinksChannel)
@@ -206,7 +216,7 @@ func mainFunc() int {
}
// Connect collector manager to metric router
CollectToRouterChannel := make(chan lp.CCMessage, 200)
CollectToRouterChannel := make(chan lp.CCMessage, chanSize)
rcfg.CollectManager.AddOutput(CollectToRouterChannel)
rcfg.MetricRouter.AddCollectorInput(CollectToRouterChannel)
@@ -220,7 +230,7 @@ func mainFunc() int {
}
// Connect receive manager to metric router
ReceiveToRouterChannel := make(chan lp.CCMessage, 200)
ReceiveToRouterChannel := make(chan lp.CCMessage, chanSize)
rcfg.ReceiveManager.AddOutput(ReceiveToRouterChannel)
rcfg.MetricRouter.AddReceiverInput(ReceiveToRouterChannel)
use_recv = true
+1 -2
View File
@@ -227,8 +227,7 @@ func (m *BeegfsMetaCollector) Read(interval time.Duration, output chan lp.CCMess
for key, data := range m.matches {
value, _ := strconv.ParseFloat(data, 32)
y, err := lp.NewMessage(key, m.tags, m.meta, map[string]any{"value": value}, time.Now())
if err == nil {
if y, err := lp.NewMetric(key, m.tags, m.meta, value, time.Now()); err == nil {
output <- y
}
}
+1 -2
View File
@@ -218,8 +218,7 @@ func (m *BeegfsStorageCollector) Read(interval time.Duration, output chan lp.CCM
for key, data := range m.matches {
value, _ := strconv.ParseFloat(data, 32)
y, err := lp.NewMessage(key, m.tags, m.meta, map[string]any{"value": value}, time.Now())
if err == nil {
if y, err := lp.NewMetric(key, m.tags, m.meta, value, time.Now()); err == nil {
output <- y
}
}
+52 -37
View File
@@ -12,6 +12,7 @@ import (
"encoding/json"
"fmt"
"sync"
"sync/atomic"
"time"
cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
@@ -50,6 +51,8 @@ var AvailableCollectors = map[string]MetricCollector{
"nfsiostat": new(NfsIOStatCollector),
"slurm_cgroup": new(SlurmCgroupCollector),
"smartmon": new(SmartMonCollector),
"lenovo_dense_power": new(LenovoDensePowerCollector),
"megware_eureka": new(MegwareEurekaCollector),
}
// Metric collector manager data structure
@@ -63,7 +66,8 @@ type collectorManager struct {
wg *sync.WaitGroup // wait group for all goroutines in cc-metric-collector
config map[string]json.RawMessage // json encoded config for collector manager
collector_wg sync.WaitGroup // internally used wait group for the parallel reading of collector
parallel_run bool // Flag whether the collectors are currently read in parallel
round_wg sync.WaitGroup // wait group for the currently running collection round
round_running atomic.Bool // Flag whether a collection round is currently running
}
// Metric collector manager access functions
@@ -119,26 +123,54 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
return nil
}
// runRound executes one collection round: first all parallel collectors
// concurrently, then the serial collectors one by one
func (cm *collectorManager) runRound(t time.Time) {
roundStart := time.Now()
for _, c := range cm.collectors {
// Read metrics from collector c via goroutine
cclog.ComponentDebug("CollectorManager", c.Name(), t)
cm.collector_wg.Add(1)
go func(myc MetricCollector) {
start := time.Now()
myc.Read(cm.duration, cm.output)
cclog.ComponentDebug("CollectorManager", myc.Name(), "took", time.Since(start))
cm.collector_wg.Done()
}(c)
}
cm.collector_wg.Wait()
for _, c := range cm.serial {
// Read metrics from collector c
cclog.ComponentDebug("CollectorManager", c.Name(), t)
start := time.Now()
c.Read(cm.duration, cm.output)
cclog.ComponentDebug("CollectorManager", c.Name(), "took", time.Since(start))
}
cclog.ComponentDebug("CollectorManager", "collection round took", time.Since(roundStart))
}
// Start starts the metric collector manager
func (cm *collectorManager) Start() {
tick := make(chan time.Time)
tick := make(chan time.Time, 1)
cm.ticker.AddChannel(tick)
cm.wg.Go(func() {
// Collector manager is done
done := func() {
// close all metric collectors
if cm.parallel_run {
cm.collector_wg.Wait()
cm.parallel_run = false
}
// wait for a still running collection round, then close all metric collectors
cm.round_wg.Wait()
for _, c := range cm.collectors {
c.Close()
}
for _, c := range cm.serial {
c.Close()
}
close(cm.done)
cclog.ComponentDebug("CollectorManager", "DONE")
}
var roundStart time.Time
// Wait for done signal or timer event
for {
select {
@@ -146,37 +178,20 @@ func (cm *collectorManager) Start() {
done()
return
case t := <-tick:
cm.parallel_run = true
for _, c := range cm.collectors {
// Wait for done signal or execute the collector
select {
case <-cm.done:
done()
return
default:
// Read metrics from collector c via goroutine
cclog.ComponentDebug("CollectorManager", c.Name(), t)
cm.collector_wg.Add(1)
go func(myc MetricCollector) {
myc.Read(cm.duration, cm.output)
cm.collector_wg.Done()
}(c)
}
}
cm.collector_wg.Wait()
cm.parallel_run = false
for _, c := range cm.serial {
// Wait for done signal or execute the collector
select {
case <-cm.done:
done()
return
default:
// Read metrics from collector c
cclog.ComponentDebug("CollectorManager", c.Name(), t)
c.Read(cm.duration, cm.output)
}
// The round runs detached from this loop, so the tick channel
// stays drained even when a round takes longer than the interval
if cm.round_running.Load() {
cclog.ComponentWarn("CollectorManager", "collection round still running after", time.Since(roundStart), "- skipping tick")
continue
}
cm.round_running.Store(true)
roundStart = time.Now()
cm.round_wg.Add(1)
go func() {
defer cm.round_wg.Done()
defer cm.round_running.Store(false)
cm.runRound(t)
}()
}
}
})
+121
View File
@@ -0,0 +1,121 @@
package collectors
import (
"encoding/json"
"sync"
"sync/atomic"
"testing"
"time"
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
)
// Fake ticker that delivers ticks on demand
type fakeTicker struct {
channels []chan time.Time
}
func (t *fakeTicker) Init(duration time.Duration) {}
func (t *fakeTicker) AddChannel(c chan time.Time) {
t.channels = append(t.channels, c)
}
func (t *fakeTicker) Close() {}
func (t *fakeTicker) tick() {
for _, c := range t.channels {
select {
case c <- time.Now():
default:
}
}
}
// Stub collector whose Read blocks until it is released
type stubCollector struct {
metricCollector
readStarted chan struct{}
release chan struct{}
reads atomic.Int32
}
func (c *stubCollector) Init(config json.RawMessage) error {
c.name = "teststub"
c.parallel = true
c.init = true
return nil
}
func (c *stubCollector) Read(duration time.Duration, output chan lp.CCMessage) {
c.reads.Add(1)
c.readStarted <- struct{}{}
<-c.release
}
func (c *stubCollector) Close() {}
func TestOverlongCollectionRoundSkipsTick(t *testing.T) {
stub := &stubCollector{
readStarted: make(chan struct{}, 10),
release: make(chan struct{}),
}
AvailableCollectors["teststub"] = stub
defer delete(AvailableCollectors, "teststub")
ticker := &fakeTicker{}
var wg sync.WaitGroup
cm, err := New(ticker, time.Second, &wg, json.RawMessage(`{"teststub": {}}`))
if err != nil {
t.Fatalf("failed to setup collector manager: %s", err.Error())
}
cm.AddOutput(make(chan lp.CCMessage, 100))
cm.Start()
// First tick starts a collection round that blocks in Read
ticker.tick()
select {
case <-stub.readStarted:
case <-time.After(5 * time.Second):
t.Fatal("collection round did not start on tick")
}
// Further ticks while the round is still running must be skipped,
// not queued up or run concurrently
for range 3 {
ticker.tick()
time.Sleep(20 * time.Millisecond)
}
if got := stub.reads.Load(); got != 1 {
t.Fatalf("expected 1 concurrent collection round, got %d reads", got)
}
// Finish the round, the next tick must start a new one
stub.release <- struct{}{}
deadline := time.After(5 * time.Second)
for stub.reads.Load() < 2 {
ticker.tick()
select {
case <-stub.readStarted:
case <-time.After(20 * time.Millisecond):
case <-deadline:
t.Fatal("no new collection round started after the previous one finished")
}
}
// Shutdown must wait for the running round and terminate cleanly.
// Closing the release channel lets any still running or straggler
// round finish immediately
close(stub.release)
closed := make(chan struct{})
go func() {
cm.Close()
close(closed)
}()
select {
case <-closed:
case <-time.After(5 * time.Second):
t.Fatal("Close() did not terminate")
}
wg.Wait()
}
+7 -1
View File
@@ -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)
@@ -171,7 +174,7 @@ func (m *CPUFreqCpuInfoCollector) Read(interval time.Duration, output chan lp.CC
fmt.Sprintf("Read(): Failed to convert cpu MHz '%s' to float64: %v", lineSplit[1], err))
return
}
if y, err := lp.NewMessage("cpufreq", t.tagSet, m.meta, map[string]any{"value": value}, now); err == nil {
if y, err := lp.NewMetric("cpufreq", t.tagSet, m.meta, value, now); err == nil {
output <- y
}
}
@@ -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() {
+1 -1
View File
@@ -129,7 +129,7 @@ func (m *CPUFreqCollector) Read(interval time.Duration, output chan lp.CCMessage
continue
}
if y, err := lp.NewMessage("cpufreq", t.tagSet, m.meta, map[string]any{"value": cpuFreq}, now); err == nil {
if y, err := lp.NewMetric("cpufreq", t.tagSet, m.meta, cpuFreq, now); err == nil {
output <- y
}
}
+21 -15
View File
@@ -27,6 +27,7 @@ const CPUSTATFILE = `/proc/stat`
type CpustatCollectorConfig struct {
ExcludeMetrics []string `json:"exclude_metrics,omitempty"`
excludeNumCPUs bool
}
type CpustatCollector struct {
@@ -43,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{
@@ -57,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{
@@ -79,11 +80,12 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
m.matches[match] = index
}
}
m.config.excludeNumCPUs = slices.Contains(m.config.ExcludeMetrics, "num_cpus")
// 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
@@ -95,11 +97,13 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
line := scanner.Text()
linefields := strings.Fields(line)
if strings.Compare(linefields[0], "cpu") == 0 {
// Kernel system statistics for all CPUs
m.olddata["cpu"] = make(map[string]int64)
for k, v := range m.matches {
m.olddata["cpu"][k], _ = strconv.ParseInt(linefields[v], 0, 64)
}
} else if strings.HasPrefix(linefields[0], "cpu") && strings.Compare(linefields[0], "cpu") != 0 {
// Kernel system statistics per CPU
cpustr := strings.TrimLeft(linefields[0], "cpu")
cpu, _ := strconv.Atoi(cpustr)
m.cputags[linefields[0]] = map[string]string{
@@ -113,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()
@@ -141,7 +148,7 @@ func (m *CpustatCollector) parseStatLine(linefields []string, tags map[string]st
sum := float64(0)
for name, value := range values {
sum += value
y, err := lp.NewMessage(name, tags, m.meta, map[string]any{"value": value * 100}, now)
y, err := lp.NewMetric(name, tags, m.meta, value*100, now)
if err == nil {
y.AddTag("unit", "Percent")
output <- y
@@ -149,7 +156,7 @@ func (m *CpustatCollector) parseStatLine(linefields []string, tags map[string]st
}
if v, ok := values["cpu_idle"]; ok {
sum -= v
y, err := lp.NewMessage("cpu_used", tags, m.meta, map[string]any{"value": sum * 100}, now)
y, err := lp.NewMetric("cpu_used", tags, m.meta, sum*100, now)
if err == nil {
y.AddTag("unit", "Percent")
output <- y
@@ -169,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()))
}
}()
@@ -190,16 +197,15 @@ 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())
}
num_cpus_metric, err := lp.NewMessage("num_cpus",
m.nodetags,
m.meta,
map[string]any{"value": num_cpus},
now,
)
if err == nil {
if !m.config.excludeNumCPUs {
if num_cpus_metric, err := lp.NewMetric("num_cpus", m.nodetags, m.meta, num_cpus, now); err == nil {
output <- num_cpus_metric
}
}
m.lastTimestamp = now
}
+8 -28
View File
@@ -128,30 +128,14 @@ mountLoop:
tags := map[string]string{"type": "node", "device": linefields[0]}
total := (stat.Blocks * uint64(stat.Bsize)) / uint64(1000_000_000)
if m.allowedMetrics["disk_total"] {
y, err := lp.NewMessage(
"disk_total",
tags,
m.meta,
map[string]any{
"value": total,
},
time.Now())
if err == nil {
if y, err := lp.NewMetric("disk_total", tags, m.meta, total, time.Now()); err == nil {
y.AddMeta("unit", "GBytes")
output <- y
}
}
free := (stat.Bfree * uint64(stat.Bsize)) / uint64(1000_000_000)
if m.allowedMetrics["disk_free"] {
y, err := lp.NewMessage(
"disk_free",
tags,
m.meta,
map[string]any{
"value": free,
},
time.Now())
if err == nil {
if y, err := lp.NewMetric("disk_free", tags, m.meta, free, time.Now()); err == nil {
y.AddMeta("unit", "GBytes")
output <- y
}
@@ -163,17 +147,13 @@ 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.NewMessage(
"part_max_used",
map[string]string{
"type": "node",
},
m.meta,
map[string]any{
"value": int(part_max_used),
},
time.Now())
y, err := lp.NewMetric("part_max_used", map[string]string{"type": "node"}, m.meta, int(part_max_used), time.Now())
if err == nil {
y.AddMeta("unit", "percent")
output <- y
+13
View File
@@ -81,3 +81,16 @@ Metrics:
* `gpfs_metaops_rate` (if `send_total_values == true` and `send_derived_values == true`)
The collector adds a `filesystem` tag to all metrics
`mmpmon` 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 command:
```
# 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 mmpmon
monitoring ALL = (root) NOPASSWD:/absolute/path/to/mmpmon -p -s
```
+97 -70
View File
@@ -23,21 +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
unitRates string
scaleByFourLanes bool
addToIBTotal bool
addToIBTotalPkgs bool
currentState int64
lastState int64
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
@@ -57,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 {
@@ -88,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)
@@ -123,36 +131,42 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
countersDir := filepath.Join(path, "counters")
portCounterFiles := []InfinibandCollectorMetric{
{
// 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: "bytes",
scale: 4,
unit: ibDataUnit,
unitRates: ibDataRateUnit,
scaleByFourLanes: true,
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: "bytes",
scale: 4,
unit: ibDataUnit,
unitRates: ibDataRateUnit,
scaleByFourLanes: true,
addToIBTotal: true,
lastState: -1,
},
{
// 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 {
@@ -164,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,
@@ -185,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 {
@@ -202,7 +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_pkts int64
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]
@@ -212,34 +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
// Save current state
counterDef.currentState = v
vScaledCounter := vRawCounter
if counterDef.scaleByFourLanes {
vScaledCounter *= uint64(4)
}
// Send absolut values
if m.config.SendAbsoluteValues {
if y, err := lp.NewMessage(
counterDef.name,
info.tagSet,
m.meta,
map[string]any{
"value": counterDef.currentState,
},
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
}
@@ -247,60 +259,75 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMess
// Send derived values
if m.config.SendDerivedValues {
if counterDef.lastState >= 0 {
rate := float64((counterDef.currentState - counterDef.lastState)) / timeDiff
if y, err := lp.NewMessage(
counterDef.name+"_bw",
info.tagSet,
m.meta,
map[string]any{
"value": rate,
},
now); err == nil {
y.AddMeta("unit", counterDef.unit+"/sec")
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.unitRates)
output <- y
}
// Sum up rates for total rates
if m.config.SendTotalValues {
switch {
case counterDef.addToIBTotal:
ibTotalBw += rate
ibTotalBwAvailable = true
case counterDef.addToIBTotalPkgs:
ibTotalPktsBw += rate
ibTotalPktsBwAvailable = true
}
}
counterDef.lastState = counterDef.currentState
}
counterDef.lastState = vRawCounter
counterDef.lastStateAvailable = true
}
// Sum up total values
if m.config.SendTotalValues {
switch {
case counterDef.addToIBTotal:
ib_total += counterDef.currentState
ibTotal += vScaledCounter
case counterDef.addToIBTotalPkgs:
ib_total_pkts += counterDef.currentState
ibTotalPkts += vScaledCounter
}
}
}
// Send total values
if m.config.SendTotalValues {
if y, err := lp.NewMessage(
"ib_total",
info.tagSet,
m.meta,
map[string]any{
"value": 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.NewMessage(
"ib_total_pkts",
info.tagSet,
m.meta,
map[string]any{
"value": 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 && 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 && ibTotalPktsBwAvailable {
if y, err := lp.NewMetric("ib_total_pkts_bw", info.tagSet, m.meta, ibTotalPktsBw, now); err == nil {
y.AddMeta("unit", ibPkgRateUnit)
output <- y
}
}
}
}
}
+2
View File
@@ -41,5 +41,7 @@ Metrics:
* `ib_xmit_bw` (if `send_derived_values == true`)
* `ib_recv_pkts_bw` (if `send_derived_values == true`)
* `ib_xmit_pkts_bw` (if `send_derived_values == true`)
* `ib_total_bw` (if `send_total_values == true` and `send_derived_values == true`)
* `ib_total_pkts_bw` (if `send_total_values == true` and `send_derived_values == true`)
The collector adds a `device` tag to all metrics
+14 -6
View File
@@ -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() {
+26 -3
View File
@@ -28,14 +28,15 @@ type IpmiCollector struct {
metricCollector
config struct {
ExcludeDevices []string `json:"exclude_devices"`
IpmitoolPath string `json:"ipmitool_path"`
IpmisensorsPath string `json:"ipmisensors_path"`
Sudo bool `json:"use_sudo"`
IncludeMetrics []string `json:"include_metrics"`
}
ipmitool string
ipmisensors string
includeMetrics map[string]bool
}
func (m *IpmiCollector) Init(config json.RawMessage) error {
@@ -65,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
@@ -146,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":
@@ -158,7 +173,7 @@ func (m *IpmiCollector) readIpmiTool(output chan lp.CCMessage) error {
unit = "Watts"
}
y, err := lp.NewMessage(name, map[string]string{"type": "node"}, m.meta, map[string]any{"value": v}, time.Now())
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)
continue
@@ -166,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 {
@@ -210,7 +228,12 @@ func (m *IpmiCollector) readIpmiSensors(output chan lp.CCMessage) error {
continue
}
name := strings.ToLower(strings.ReplaceAll(lv[1], " ", "_"))
y, err := lp.NewMessage(name, map[string]string{"type": "node"}, m.meta, map[string]any{"value": v}, time.Now())
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)
continue
+7 -4
View File
@@ -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" : []
}
```
@@ -23,9 +24,9 @@ The `ipmistat` collector reads data from `ipmitool` (`ipmitool sensor`) or `ipmi
The metrics depend on the output of the underlying tools but contain temperature, power and energy metrics.
ipmitool and ipmi-sensors typically require root to run.
In order to 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 this command:
`ipmitool` and `ipmi-sensors` 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.
@@ -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.
+170
View File
@@ -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
}
+51
View File
@@ -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
```
+1 -1
View File
@@ -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 (
+2 -2
View File
@@ -109,7 +109,7 @@ func (m *LoadavgCollector) Read(interval time.Duration, output chan lp.CCMessage
if m.load_skips[i] {
continue
}
y, err := lp.NewMessage(name, m.tags, m.meta, map[string]any{"value": x}, now)
y, err := lp.NewMetric(name, m.tags, m.meta, x, now)
if err == nil {
output <- y
}
@@ -128,7 +128,7 @@ func (m *LoadavgCollector) Read(interval time.Duration, output chan lp.CCMessage
if m.proc_skips[i] {
continue
}
y, err := lp.NewMessage(name, m.tags, m.meta, map[string]any{"value": x}, now)
y, err := lp.NewMetric(name, m.tags, m.meta, x, now)
if err == nil {
output <- y
}
+13
View File
@@ -55,3 +55,16 @@ Metrics:
* `lustre_inode_permission_diff` (if `send_diff_values == true`)
This collector adds an `device` tag.
`lctl` 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 command:
```
# 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 lctl
monitoring ALL = (root) NOPASSWD:/absolute/path/to/lctl get_param llite.*.stats
```
+194
View File
@@ -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
}
+44
View File
@@ -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
```
+11 -7
View File
@@ -72,7 +72,8 @@ func getStats(filename string) map[string]MemstatStats {
for scanner.Scan() {
line := scanner.Text()
linefields := strings.Fields(line)
if len(linefields) == 3 {
switch len(linefields) {
case 3:
v, err := strconv.ParseFloat(linefields[1], 64)
if err == nil {
stats[strings.Trim(linefields[0], ":")] = MemstatStats{
@@ -80,7 +81,7 @@ func getStats(filename string) map[string]MemstatStats {
unit: linefields[2],
}
}
} else if len(linefields) == 5 {
case 5:
v, err := strconv.ParseFloat(linefields[3], 64)
if err == nil {
cclog.ComponentDebug("getStats", strings.Trim(linefields[2], ":"), v, linefields[4])
@@ -106,7 +107,10 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
}
}
m.meta = map[string]string{"source": m.name, "group": "Memory"}
m.meta = map[string]string{
"source": m.name,
"group": "Memory",
}
m.stats = make(map[string]int64)
m.matches = make(map[string]string)
m.tags = map[string]string{"type": "node"}
@@ -145,7 +149,7 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
"KernelStack": "mem_kernelstack",
}
for k, v := range matches {
if !slices.Contains(m.config.ExcludeMetrics, k) {
if !slices.Contains(m.config.ExcludeMetrics, v) {
m.matches[k] = v
}
}
@@ -153,7 +157,7 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
if !slices.Contains(m.config.ExcludeMetrics, "mem_used") {
m.sendMemUsed = true
}
if len(m.matches) == 0 {
if len(m.matches) == 0 && !m.sendMemUsed {
return fmt.Errorf("%s Init(): no metrics to collect", m.name)
}
if err := m.setup(); err != nil {
@@ -213,7 +217,7 @@ func (m *MemstatCollector) Read(interval time.Duration, output chan lp.CCMessage
}
}
y, err := lp.NewMessage(name, tags, m.meta, map[string]any{"value": value}, time.Now())
y, err := lp.NewMetric(name, tags, m.meta, value, time.Now())
if err == nil {
if len(unit) > 0 {
y.AddMeta("unit", unit)
@@ -252,7 +256,7 @@ func (m *MemstatCollector) Read(interval time.Duration, output chan lp.CCMessage
}
}
}
y, err := lp.NewMessage("mem_used", tags, m.meta, map[string]any{"value": memUsed}, time.Now())
y, err := lp.NewMetric("mem_used", tags, m.meta, memUsed, time.Now())
if err == nil {
if len(unit) > 0 {
y.AddMeta("unit", unit)
+2 -2
View File
@@ -262,14 +262,14 @@ func (m *NetstatCollector) Read(interval time.Duration, output chan lp.CCMessage
continue
}
if m.config.SendAbsoluteValues {
if y, err := lp.NewMessage(metric.name, metric.tags, metric.meta, map[string]any{"value": v}, now); err == nil {
if y, err := lp.NewMetric(metric.name, metric.tags, metric.meta, v, now); err == nil {
output <- y
}
}
if m.config.SendDerivedValues {
if metric.lastValue >= 0 {
rate := float64(v-metric.lastValue) / timeDiff
if y, err := lp.NewMessage(metric.name+"_bw", metric.tags, metric.meta_rates, map[string]any{"value": rate}, now); err == nil {
if y, err := lp.NewMetric(metric.name+"_bw", metric.tags, metric.meta_rates, rate, now); err == nil {
output <- y
}
}
+3 -4
View File
@@ -146,16 +146,15 @@ func (m *nfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
continue
}
valueMap := make(map[string]any)
if data.current >= 0 && data.last >= 0 {
valueMap["value"] = data.current - data.last
}
y, err := lp.NewMessage(fmt.Sprintf("%s_%s", prefix, name), m.tags, m.meta, valueMap, timestamp)
value := data.current - data.last
y, err := lp.NewMetric(fmt.Sprintf("%s_%s", prefix, name), m.tags, m.meta, value, timestamp)
if err == nil {
y.AddMeta("version", m.version)
output <- y
}
}
}
}
func (m *nfsCollector) Close() {
+2 -9
View File
@@ -145,14 +145,7 @@ func (m *NfsIOStatCollector) Read(interval time.Duration, output chan lp.CCMessa
if old, ok := m.data[mntpoint]; ok {
for name, newVal := range values {
if m.config.SendAbsoluteValues {
msg, err := lp.NewMessage(
"nfsio_"+name,
m.tags,
m.meta,
map[string]any{
"value": newVal,
},
now)
msg, err := lp.NewMetric("nfsio_"+name, m.tags, m.meta, newVal, now)
if err == nil {
msg.AddTag("stype", "filesystem")
msg.AddTag("stype-id", mntpoint)
@@ -161,7 +154,7 @@ func (m *NfsIOStatCollector) Read(interval time.Duration, output chan lp.CCMessa
}
if m.config.SendDerivedValues {
rate := float64(newVal-old[name]) / timeDiff
msg, err := lp.NewMessage(fmt.Sprintf("nfsio_%s_bw", name), m.tags, m.meta, map[string]any{"value": rate}, now)
msg, err := lp.NewMetric(fmt.Sprintf("nfsio_%s_bw", name), m.tags, m.meta, rate, now)
if err == nil {
if strings.HasPrefix(name, "page") {
msg.AddMeta("unit", "4K_pages/s")
+1 -6
View File
@@ -242,12 +242,7 @@ func (m *RAPLCollector) Read(interval time.Duration, output chan lp.CCMessage) {
timeDiff := energyTimestamp.Sub(p.energyTimestamp)
averagePower := float64(energyDiff) / float64(timeDiff.Microseconds())
y, err := lp.NewMessage(
"rapl_average_power",
p.tags,
m.meta,
map[string]any{"value": averagePower},
energyTimestamp)
y, err := lp.NewMetric("rapl_average_power", p.tags, m.meta, averagePower, energyTimestamp)
if err == nil {
output <- y
}
+18 -36
View File
@@ -158,128 +158,110 @@ func (m *RocmSmiCollector) Read(interval time.Duration, output chan lp.CCMessage
if !dev.excludeMetrics["rocm_gfx_util"] {
value := metrics.Average_gfx_activity
y, err := lp.NewMessage("rocm_gfx_util", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_gfx_util", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_umc_util"] {
value := metrics.Average_umc_activity
y, err := lp.NewMessage("rocm_umc_util", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_umc_util", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_mm_util"] {
value := metrics.Average_mm_activity
y, err := lp.NewMessage("rocm_mm_util", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_mm_util", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_avg_power"] {
value := metrics.Average_socket_power
y, err := lp.NewMessage("rocm_avg_power", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_avg_power", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_temp_mem"] {
value := metrics.Temperature_mem
y, err := lp.NewMessage("rocm_temp_mem", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_temp_mem", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_temp_hotspot"] {
value := metrics.Temperature_hotspot
y, err := lp.NewMessage("rocm_temp_hotspot", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_temp_hotspot", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_temp_edge"] {
value := metrics.Temperature_edge
y, err := lp.NewMessage("rocm_temp_edge", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_temp_edge", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_temp_vrgfx"] {
value := metrics.Temperature_vrgfx
y, err := lp.NewMessage("rocm_temp_vrgfx", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_temp_vrgfx", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_temp_vrsoc"] {
value := metrics.Temperature_vrsoc
y, err := lp.NewMessage("rocm_temp_vrsoc", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_temp_vrsoc", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_temp_vrmem"] {
value := metrics.Temperature_vrmem
y, err := lp.NewMessage("rocm_temp_vrmem", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_temp_vrmem", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_gfx_clock"] {
value := metrics.Average_gfxclk_frequency
y, err := lp.NewMessage("rocm_gfx_clock", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_gfx_clock", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_soc_clock"] {
value := metrics.Average_socclk_frequency
y, err := lp.NewMessage("rocm_soc_clock", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_soc_clock", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_u_clock"] {
value := metrics.Average_uclk_frequency
y, err := lp.NewMessage("rocm_u_clock", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_u_clock", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_v0_clock"] {
value := metrics.Average_vclk0_frequency
y, err := lp.NewMessage("rocm_v0_clock", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_v0_clock", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_v1_clock"] {
value := metrics.Average_vclk1_frequency
y, err := lp.NewMessage("rocm_v1_clock", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_v1_clock", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_d0_clock"] {
value := metrics.Average_dclk0_frequency
y, err := lp.NewMessage("rocm_d0_clock", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_d0_clock", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_d1_clock"] {
value := metrics.Average_dclk1_frequency
y, err := lp.NewMessage("rocm_d1_clock", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_d1_clock", dev.tags, dev.meta, value, timestamp); err == nil {
output <- y
}
}
if !dev.excludeMetrics["rocm_temp_hbm"] {
for i := range rocm_smi.NUM_HBM_INSTANCES {
value := metrics.Temperature_hbm[i]
y, err := lp.NewMessage("rocm_temp_hbm", dev.tags, dev.meta, map[string]any{"value": value}, timestamp)
if err == nil {
if y, err := lp.NewMetric("rocm_temp_hbm", dev.tags, dev.meta, value, timestamp); err == nil {
y.AddTag("stype", "device")
y.AddTag("stype-id", strconv.Itoa(i))
output <- y
+34 -3
View File
@@ -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
}
+15
View File
@@ -50,3 +50,18 @@ Metrics:
* `smartmon_errlog_entries`: Error log entries
* `smartmon_warn_temp_time`: Time above the warning temperature threshold
* `smartmon_crit_comp_time`: Time above the critical composite temperature threshold
`smartctl` 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 command:
```
# 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 lctl
monitoring ALL = (root) NOPASSWD:/absolute/path/to/smartctl --json=c --device=* "--all" *
# Or add individual rules for each device
# monitoring ALL = (root) NOPASSWD:/absolute/path/to/smartctl --json=c --device=<device_type> "--all" <device>
```
+3 -21
View File
@@ -201,26 +201,14 @@ func (m *TempCollector) Read(interval time.Duration, output chan lp.CCMessage) {
continue
}
x /= 1000
y, err := lp.NewMessage(
sensor.metricName,
sensor.tags,
m.meta,
map[string]any{"value": x},
time.Now(),
)
y, err := lp.NewMetric(sensor.metricName, sensor.tags, m.meta, x, time.Now())
if err == nil {
output <- y
}
// max temperature
if m.config.ReportMaxTemp && sensor.maxTemp != 0 {
y, err := lp.NewMessage(
sensor.maxTempName,
sensor.tags,
m.meta,
map[string]any{"value": sensor.maxTemp},
time.Now(),
)
y, err := lp.NewMetric(sensor.maxTempName, sensor.tags, m.meta, sensor.maxTemp, time.Now())
if err == nil {
output <- y
}
@@ -228,13 +216,7 @@ func (m *TempCollector) Read(interval time.Duration, output chan lp.CCMessage) {
// critical temperature
if m.config.ReportCriticalTemp && sensor.critTemp != 0 {
y, err := lp.NewMessage(
sensor.critTempName,
sensor.tags,
m.meta,
map[string]any{"value": sensor.critTemp},
time.Now(),
)
y, err := lp.NewMetric(sensor.critTempName, sensor.tags, m.meta, sensor.critTemp, time.Now())
if err == nil {
output <- y
}
+1 -9
View File
@@ -86,15 +86,7 @@ func (m *TopProcsCollector) Read(interval time.Duration, output chan lp.CCMessag
lines := strings.Split(string(stdout), "\n")
for i := 1; i < m.config.Num_procs+1; i++ {
name := fmt.Sprintf("topproc%d", i)
y, err := lp.NewMessage(
name,
m.tags,
m.meta,
map[string]any{
"value": lines[i],
},
time.Now())
if err == nil {
if y, err := lp.NewMetric(name, m.tags, m.meta, lines[i], time.Now()); err == nil {
output <- y
}
}
+2
View File
@@ -19,6 +19,8 @@ The global file contains the paths to the other four files and some global optio
Be aware that the paths are relative to the execution folder of the cc-metric-collector binary, so it is recommended to use absolute paths.
The optional `channel_buffer_size` option sets the capacity of the internal channels between the components (collectors → router → sinks). If unset, it defaults to `max(200, 24 * number of CPUs)` so that one interval's burst of per-hwthread metrics fits without back-pressuring the collectors on nodes with many cores.
## Component configuration
The others are mainly list of of subcomponents: the collectors, the receivers, the router and the sinks. Their role is best shown in a picture:
+16 -14
View File
@@ -3,14 +3,14 @@ module github.com/ClusterCockpit/cc-metric-collector
go 1.25.0
require (
github.com/ClusterCockpit/cc-lib/v2 v2.11.0
github.com/ClusterCockpit/go-rocm-smi v0.3.0
github.com/NVIDIA/go-nvml v0.13.0-1
github.com/ClusterCockpit/cc-lib/v2 v2.12.0
github.com/ClusterCockpit/go-rocm-smi v0.4.0
github.com/NVIDIA/go-nvml v0.13.2-0
github.com/PaesslerAG/gval v1.2.4
github.com/fsnotify/fsnotify v1.9.0
github.com/tklauser/go-sysconf v0.3.16
golang.design/x/thread v0.0.0-20210122121316-335e9adffdf1
golang.org/x/sys v0.42.0
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.45.0
)
require (
@@ -18,27 +18,29 @@ require (
github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/coder/websocket v1.8.14 // indirect
github.com/expr-lang/expr v1.17.8 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/gorilla/mux v1.8.1 // indirect
github.com/influxdata/influxdb-client-go/v2 v2.14.0 // indirect
github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf // indirect
github.com/klauspost/compress v1.18.4 // indirect
github.com/klauspost/compress v1.18.5 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/nats-io/nats.go v1.49.0 // indirect
github.com/nats-io/nats.go v1.51.0 // indirect
github.com/nats-io/nkeys v0.4.15 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
github.com/oapi-codegen/runtime v1.3.0 // indirect
github.com/oapi-codegen/runtime v1.4.0 // indirect
github.com/prometheus/client_golang v1.23.2 // indirect
github.com/prometheus/client_model v0.6.2 // indirect
github.com/prometheus/common v0.67.5 // indirect
github.com/prometheus/procfs v0.20.1 // indirect
github.com/questdb/go-questdb-client/v4 v4.2.0 // indirect
github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 // indirect
github.com/shopspring/decimal v1.4.0 // indirect
github.com/stmcginnis/gofish v0.21.4 // indirect
github.com/tklauser/numcpus v0.11.0 // indirect
github.com/stmcginnis/gofish v0.21.6 // indirect
github.com/tklauser/numcpus v0.12.0 // indirect
go.yaml.in/yaml/v2 v2.4.4 // indirect
golang.org/x/crypto v0.49.0 // indirect
golang.org/x/net v0.52.0 // indirect
golang.org/x/crypto v0.50.0 // indirect
golang.org/x/net v0.53.0 // indirect
google.golang.org/protobuf v1.36.11 // indirect
)
+123 -39
View File
@@ -1,39 +1,71 @@
github.com/ClusterCockpit/cc-lib/v2 v2.11.0 h1:LaLs4J0b7FArIXT8byMUcIcUr55R5obATjVi7qI02r4=
github.com/ClusterCockpit/cc-lib/v2 v2.11.0/go.mod h1:Oj+N2lpFqiBOBzjfrLIGJ2YSWT400TX4M0ii4lNl81A=
dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk=
dario.cat/mergo v1.0.0/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 h1:L/gRVlceqvL25UVaW/CKtUDjefjrs0SPonmDGUVOYP0=
github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E=
github.com/ClusterCockpit/cc-lib/v2 v2.12.0 h1:ZbGD68nDniuvzFjJCdyYawpCBrabdSyWOg5FFSyFbjQ=
github.com/ClusterCockpit/cc-lib/v2 v2.12.0/go.mod h1:ml8xtcYa5WhPM7JDQ+M9/R9ZBxITCR/5xqGJ//GxXJI=
github.com/ClusterCockpit/cc-line-protocol/v2 v2.4.0 h1:hIzxgTBWcmCIHtoDKDkSCsKCOCOwUC34sFsbD2wcW0Q=
github.com/ClusterCockpit/cc-line-protocol/v2 v2.4.0/go.mod h1:y42qUu+YFmu5fdNuUAS4VbbIKxVjxCvbVqFdpdh8ahY=
github.com/ClusterCockpit/go-rocm-smi v0.3.0 h1:1qZnSpG7/NyLtc7AjqnUL9Jb8xtqG1nMVgp69rJfaR8=
github.com/ClusterCockpit/go-rocm-smi v0.3.0/go.mod h1:+I3UMeX3OlizXDf1WpGD43W4KGZZGVSGmny6rTeOnWA=
github.com/NVIDIA/go-nvml v0.11.6-0/go.mod h1:hy7HYeQy335x6nEss0Ne3PYqleRa6Ct+VKD9RQ4nyFs=
github.com/NVIDIA/go-nvml v0.13.0-1 h1:OLX8Jq3dONuPOQPC7rndB6+iDmDakw0XTYgzMxObkEw=
github.com/ClusterCockpit/go-rocm-smi v0.4.0 h1:3+bEPrSkjEJcOtt+qBUX48ugDVlOFaKUnXHTef2Ve2Q=
github.com/ClusterCockpit/go-rocm-smi v0.4.0/go.mod h1:c19u5vBCcgb7DjL4EWTGSGpo6c79d07r4rxD50z25ng=
github.com/Microsoft/go-winio v0.6.1 h1:9/kr64B9VUZrLm5YYwbGtUJnMgqWVOdUAXu6Migciow=
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/go.mod h1:+KNA7c7gIBH7SKSJ1ntlwkfN80zdx8ovl4hrK3LmPt4=
github.com/NVIDIA/go-nvml v0.13.2-0 h1:7M4cFG62wSUHw8i0XSiNU7ejKODytTS6ZrW/vgB2NSI=
github.com/NVIDIA/go-nvml v0.13.2-0/go.mod h1:ahi2psRYoa+wYUBIrZPRO+wJs9lcvMhxSSkjjvsJJNQ=
github.com/PaesslerAG/gval v1.2.4 h1:rhX7MpjJlcxYwL2eTTYIOBUyEKZ+A96T9vQySWkVUiU=
github.com/PaesslerAG/gval v1.2.4/go.mod h1:XRFLwvmkTEdYziLdaCeCa5ImcGVrfQbeNUbVR+C6xac=
github.com/PaesslerAG/jsonpath v0.1.0 h1:gADYeifvlqK3R3i2cR5B4DGgxLXIPb3TRTH1mGi0jPI=
github.com/PaesslerAG/jsonpath v0.1.0/go.mod h1:4BzmtoM/PI8fPO4aQGIusjGxGir2BzcV0grWtFzq1Y8=
github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk=
github.com/antithesishq/antithesis-sdk-go v0.5.0-default-no-op h1:Ucf+QxEKMbPogRO5guBNe5cgd9uZgfoJLOYs8WWhtjM=
github.com/antithesishq/antithesis-sdk-go v0.5.0-default-no-op/go.mod h1:IUpT2DPAKh6i/YhSbt6Gl3v2yvUZjmKncl7U91fup7E=
github.com/antithesishq/antithesis-sdk-go v0.6.0-default-no-op h1:kpBdlEPbRvff0mDD1gk7o9BhI16b9p5yYAXRlidpqJE=
github.com/antithesishq/antithesis-sdk-go v0.6.0-default-no-op/go.mod h1:IUpT2DPAKh6i/YhSbt6Gl3v2yvUZjmKncl7U91fup7E=
github.com/apapsch/go-jsonmerge/v2 v2.0.0 h1:axGnT1gRIfimI7gJifB699GoE/oq+F2MU7Dml6nw9rQ=
github.com/apapsch/go-jsonmerge/v2 v2.0.0/go.mod h1:lvDnEdqiQrp0O42VQGgmlKpxL1AP2+08jFMw88y4klk=
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/bmatcuk/doublestar v1.1.1/go.mod h1:UD6OnuiIn0yFxxA2le/rnRU1G4RaI4UvFv1sNto9p6w=
github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
github.com/containerd/containerd v1.7.12 h1:+KQsnv4VnzyxWcfO9mlxxELaoztsDEjOuCMPAuPqgU0=
github.com/containerd/containerd v1.7.12/go.mod h1:/5OMpE1p0ylxtEUGY8kuCYkDRzJm9NO1TFMWjUpdevk=
github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I=
github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo=
github.com/cpuguy83/dockercfg v0.3.1 h1:/FpZ+JaygUR/lZP2NlFI2DVfrOEMAIKP5wWEJdoYe9E=
github.com/cpuguy83/dockercfg v0.3.1/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/docker/distribution v2.8.2+incompatible h1:T3de5rq0dB1j30rp0sA2rER+m322EBzniBPB6ZIzuh8=
github.com/docker/distribution v2.8.2+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
github.com/docker/docker v24.0.9+incompatible h1:HPGzNmwfLZWdxHqK9/II92pyi1EpYKsAqcl4G0Of9v0=
github.com/docker/docker v24.0.9+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj1Br63c=
github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc=
github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
github.com/expr-lang/expr v1.17.8 h1:W1loDTT+0PQf5YteHSTpju2qfUfNoBt4yw9+wOEU9VM=
github.com/expr-lang/expr v1.17.8/go.mod h1:8/vRC7+7HBzESEqt5kKpYXxrxkr31SaO8r40VO/1IT4=
github.com/frankban/quicktest v1.13.0 h1:yNZif1OkDfNoDfb9zZa9aXIpejNR4F23Wely0c+Qdqk=
github.com/frankban/quicktest v1.13.0/go.mod h1:qLE0fzW0VuyUAJgPU19zByoIr0HtCHN/r/VLSOOIySU=
github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k=
github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx59Ho=
github.com/fsnotify/fsnotify v1.10.1/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo=
github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/google/go-tpm v0.9.7 h1:u89J4tUUeDTlH8xxC3CTW7OHZjbjKoHdQ9W7gCUhtxA=
github.com/google/go-tpm v0.9.7/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY=
@@ -45,32 +77,54 @@ github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf/go.mod h1
github.com/influxdata/line-protocol-corpus v0.0.0-20210922080147-aa28ccfb8937 h1:MHJNQ+p99hFATQm6ORoLmpUCF7ovjwEFshs/NHzAbig=
github.com/influxdata/line-protocol-corpus v0.0.0-20210922080147-aa28ccfb8937/go.mod h1:BKR9c0uHSmRgM/se9JhFHtTT7JTO67X23MtKMHtZcpo=
github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE=
github.com/klauspost/compress v1.18.4 h1:RPhnKRAQ4Fh8zU2FY/6ZFDwTVTxgJ/EMydqSTzE9a2c=
github.com/klauspost/compress v1.18.4/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4=
github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a h1:N9zuLhTvBSRt0gWSiJswwQ2HqDmtX/ZCDJURnKUt1Ik=
github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a/go.mod h1:JKx41uQRwqlTZabZc+kILPrO/3jlKnQ2Z8b7YiVw5cE=
github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY=
github.com/magiconair/properties v1.8.7/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0=
github.com/minio/highwayhash v1.0.4-0.20251030100505-070ab1a87a76 h1:KGuD/pM2JpL9FAYvBrnBBeENKZNh6eNtjqytV6TYjnk=
github.com/minio/highwayhash v1.0.4-0.20251030100505-070ab1a87a76/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
github.com/moby/patternmatcher v0.6.0 h1:GmP9lR19aU5GqSSFko+5pRqHi+Ohk1O69aFiKkVGiPk=
github.com/moby/patternmatcher v0.6.0/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc=
github.com/moby/sys/sequential v0.5.0 h1:OPvI35Lzn9K04PBbCLW0g4LcFAJgHsvXsRyewg5lXtc=
github.com/moby/sys/sequential v0.5.0/go.mod h1:tH2cOOs5V9MlPiXcQzRC+eEyab644PWKGRYaaV5ZZlo=
github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0=
github.com/moby/term v0.5.0/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y=
github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A=
github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
github.com/nats-io/jwt/v2 v2.8.0 h1:K7uzyz50+yGZDO5o772eRE7atlcSEENpL7P+b74JV1g=
github.com/nats-io/jwt/v2 v2.8.0/go.mod h1:me11pOkwObtcBNR8AiMrUbtVOUGkqYjMQZ6jnSdVUIA=
github.com/nats-io/nats-server/v2 v2.12.3 h1:KRv+1n7lddMVgkJPQer+pt36TcO0ENxjilBmeWdjcHs=
github.com/nats-io/nats-server/v2 v2.12.3/go.mod h1:MQXjG9WjyXKz9koWzUc3jYUMKD8x3CLmTNy91IQQz3Y=
github.com/nats-io/nats.go v1.49.0 h1:yh/WvY59gXqYpgl33ZI+XoVPKyut/IcEaqtsiuTJpoE=
github.com/nats-io/nats.go v1.49.0/go.mod h1:fDCn3mN5cY8HooHwE2ukiLb4p4G4ImmzvXyJt+tGwdw=
github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
github.com/nats-io/nats-server/v2 v2.12.7 h1:prQ9cPiWHcnwfT81Wi5lU9LL8TLY+7pxDru6fQYLCQQ=
github.com/nats-io/nats-server/v2 v2.12.7/go.mod h1:dOnmkprKMluTmTF7/QHZioxlau3sKHUM/LBPy9AiBPw=
github.com/nats-io/nats.go v1.51.0 h1:ByW84XTz6W03GSSsygsZcA+xgKK8vPGaa/FCAAEHnAI=
github.com/nats-io/nats.go v1.51.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
github.com/oapi-codegen/runtime v1.3.0 h1:vyK1zc0gDWWXgk2xoQa4+X4RNNc5SL2RbTpJS/4vMYA=
github.com/oapi-codegen/runtime v1.3.0/go.mod h1:kOdeacKy7t40Rclb1je37ZLFboFxh+YLy0zaPCMibPY=
github.com/oapi-codegen/runtime v1.4.0 h1:KLOSFOp7UzkbS7Cs1ms6NBEKYr0WmH2wZG0KKbd2er4=
github.com/oapi-codegen/runtime v1.4.0/go.mod h1:5sw5fxCDmnOzKNYmkVNF8d34kyUeejJEY8HNT2WaPec=
github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
github.com/opencontainers/image-spec v1.1.0-rc5 h1:Ygwkfw9bpDvs+c9E34SdgGOj41dX/cbdlwvlWt0pnFI=
github.com/opencontainers/image-spec v1.1.0-rc5/go.mod h1:X4pATf0uXsnn3g5aiGIsVnJBR4mxhKzfwmvK/B2NTm8=
github.com/opencontainers/runc v1.1.5 h1:L44KXEpKmfWDcS02aeGm8QNTFXTo2D+8MYGDIJ/GDEs=
github.com/opencontainers/runc v1.1.5/go.mod h1:1J5XiS+vdZ3wCyZybsuxXZWGrgSr8fFJHLXuG2PsnNg=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b h1:0LFwY6Q3gMACTjAbMZBjXAqTOzOwFaj2Ld6cjeQ7Rig=
github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o=
github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg=
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
@@ -79,40 +133,70 @@ github.com/prometheus/common v0.67.5 h1:pIgK94WWlQt1WLwAC5j2ynLaBRDiinoAb86HZHTU
github.com/prometheus/common v0.67.5/go.mod h1:SjE/0MzDEEAyrdr5Gqc6G+sXI67maCxzaT3A2+HqjUw=
github.com/prometheus/procfs v0.20.1 h1:XwbrGOIplXW/AU3YhIhLODXMJYyC1isLFfYCsTEycfc=
github.com/prometheus/procfs v0.20.1/go.mod h1:o9EMBZGRyvDrSPH1RqdxhojkuXstoe4UlK79eF5TGGo=
github.com/questdb/go-questdb-client/v4 v4.2.0 h1:+d0HJwCjUWMj7zmY6qmhoqTJzTyoYKl+LSTYGN0T8T8=
github.com/questdb/go-questdb-client/v4 v4.2.0/go.mod h1:/2x93LK1wjM4JX/b5c6q7Yqk22htjWY1lE6p1X8iLbE=
github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 h1:lZUw3E0/J3roVtGQ+SCrUrg3ON6NgVqpn3+iol9aGu4=
github.com/santhosh-tekuri/jsonschema/v5 v5.3.1/go.mod h1:uToXkOrWAZ6/Oc07xWQrPOhJotwFIyu2bBVN41fcDUY=
github.com/shirou/gopsutil/v3 v3.23.12 h1:z90NtUkp3bMtmICZKpC4+WaknU1eXtp5vtbQ11DgpE4=
github.com/shirou/gopsutil/v3 v3.23.12/go.mod h1:1FrWgea594Jp7qmjHUUPlJDTPgcsb9mGnXDxavtikzM=
github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM=
github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ=
github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k=
github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
github.com/spkg/bom v0.0.0-20160624110644-59b7046e48ad/go.mod h1:qLr4V1qq6nMqFKkMo8ZTx3f+BZEkzsRUY10Xsm2mwU0=
github.com/stmcginnis/gofish v0.21.4 h1:daexK8sh31CgeSMkPUNs21HWHHA9ecCPJPyLCTxukCg=
github.com/stmcginnis/gofish v0.21.4/go.mod h1:PzF5i8ecRG9A2ol8XT64npKUunyraJ+7t0kYMpQAtqU=
github.com/stmcginnis/gofish v0.21.6 h1:jK3TGD6VANaAHKHypVNfD6io2nPrU+6eF8X4qARsTlY=
github.com/stmcginnis/gofish v0.21.6/go.mod h1:PzF5i8ecRG9A2ol8XT64npKUunyraJ+7t0kYMpQAtqU=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
github.com/tklauser/go-sysconf v0.3.16 h1:frioLaCQSsF5Cy1jgRBrzr6t502KIIwQ0MArYICU0nA=
github.com/tklauser/go-sysconf v0.3.16/go.mod h1:/qNL9xxDhc7tx3HSRsLWNnuzbVfh3e7gh/BmM179nYI=
github.com/tklauser/numcpus v0.11.0 h1:nSTwhKH5e1dMNsCdVBukSZrURJRoHbSEQjdEbY+9RXw=
github.com/tklauser/numcpus v0.11.0/go.mod h1:z+LwcLq54uWZTX0u/bGobaV34u6V7KNlTZejzM6/3MQ=
github.com/testcontainers/testcontainers-go v0.26.0 h1:uqcYdoOHBy1ca7gKODfBd9uTHVK3a7UL848z09MVZ0c=
github.com/testcontainers/testcontainers-go v0.26.0/go.mod h1:ICriE9bLX5CLxL9OFQ2N+2N+f+803LNJ1utJb1+Inx0=
github.com/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRUgPQU=
github.com/tklauser/go-sysconf v0.4.0/go.mod h1:8mTNWyog7H+MpKijp4VmKJAd2bbYQ2zuUwkYRbUArPI=
github.com/tklauser/numcpus v0.12.0 h1:NR85qdvHA9pFse3x3weVZ0r0ST8R6l5RHbZrlRaqob4=
github.com/tklauser/numcpus v0.12.0/go.mod h1:ABHeXzJnr/qqwguhClkZKT1/8VABcYrsyUiUGobwWJg=
github.com/yusufpapurcu/wmi v1.2.3 h1:E1ctvB7uKFMOJw3fdOW32DwGE9I7t++CRUEMKvFoFiw=
github.com/yusufpapurcu/wmi v1.2.3/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
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.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0=
golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw=
golang.org/x/sys v0.0.0-20210122093101-04d7465088b8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
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=
golang.org/x/exp v0.0.0-20231005195138-3e424a577f31/go.mod h1:S2oDrQGGwySpoQPVqRShND87VCbxmc6bL1Yd2oYrm6k=
golang.org/x/mod v0.13.0 h1:I/DsJXRlw/8l/0c24sM9yb0T4z9liZTduXvdAWYiysY=
golang.org/x/mod v0.13.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
golang.org/x/tools v0.14.0 h1:jvNa2pY0M4r62jkRQ6RwEZZyPcymeL9XZMLBbV7U2nc=
golang.org/x/tools v0.14.0/go.mod h1:uYBEerGOWcJyEORxN+Ek8+TT266gXkNlHdJBwexUsBg=
google.golang.org/genproto/googleapis/rpc v0.0.0-20231002182017-d307bd883b97 h1:6GQBEOdGkX6MMTLT9V+TjtIRZCw9VPD5Z+yHY9wMgS0=
google.golang.org/genproto/googleapis/rpc v0.0.0-20231002182017-d307bd883b97/go.mod h1:v7nGkzlmW8P3n/bKmWBn2WpBjpOEx8Q6gMueudAmKfY=
google.golang.org/grpc v1.58.3 h1:BjnpXut1btbtgN/6sp+brB2Kbm2LjNXnidYujAVbSoQ=
google.golang.org/grpc v1.58.3/go.mod h1:tgX3ZQDlNJGU96V6yHh1T/JeoBQ2TXdr43YbYSsCJk0=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+10 -10
View File
@@ -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 {
+1 -1
View File
@@ -79,7 +79,7 @@ func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker,
// Start starts the metric cache
func (c *metricCache) Start() {
c.tickchan = make(chan time.Time)
c.tickchan = make(chan time.Time, 1)
c.ticker.AddChannel(c.tickchan)
// Router cache is done
done := func() {
+20 -1
View File
@@ -228,11 +228,25 @@ func (r *metricRouter) DoAddTags(point lp.CCMessage) {
func (r *metricRouter) Start() {
// start timer if configured
r.timestamp = time.Now()
timeChan := make(chan time.Time)
timeChan := make(chan time.Time, 1)
if r.config.IntervalStamp {
r.ticker.AddChannel(timeChan)
}
// Drain a pending tick before stamping new metrics, so a new interval's
// metrics never carry the previous interval's timestamp
updateTimestamp := func() {
if !r.config.IntervalStamp {
return
}
select {
case timestamp := <-timeChan:
r.timestamp = timestamp
cclog.ComponentDebug("MetricRouter", "Update timestamp", r.timestamp.UnixNano())
default:
}
}
// Router manager is done
done := func() {
close(r.done)
@@ -298,14 +312,19 @@ func (r *metricRouter) Start() {
case timestamp := <-timeChan:
r.timestamp = timestamp
cclog.ComponentDebug("MetricRouter", "Update timestamp", r.timestamp.UnixNano())
if len(r.coll_input) == cap(r.coll_input) {
cclog.ComponentWarn("MetricRouter", "collector input channel full at tick, sinks may be too slow")
}
case p := <-r.coll_input:
updateTimestamp()
coll_forward(p)
for i := 0; len(r.coll_input) > 0 && i < (r.maxForward-1); i++ {
coll_forward(<-r.coll_input)
}
case p := <-r.recv_input:
updateTimestamp()
recv_forward(p)
for i := 0; len(r.recv_input) > 0 && i < (r.maxForward-1); i++ {
recv_forward(<-r.recv_input)
@@ -0,0 +1,99 @@
package metricRouter
import (
"encoding/json"
"fmt"
"sync"
"testing"
"time"
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
)
// Fake ticker that delivers ticks on demand
type fakeTicker struct {
channels []chan time.Time
}
func (t *fakeTicker) Init(duration time.Duration) {}
func (t *fakeTicker) AddChannel(c chan time.Time) {
t.channels = append(t.channels, c)
}
func (t *fakeTicker) Close() {}
func (t *fakeTicker) tick(ts time.Time) {
for _, c := range t.channels {
c <- ts
}
}
func genMessages(t *testing.T, num int) []lp.CCMessage {
t.Helper()
msgs := make([]lp.CCMessage, 0, num)
tags := map[string]string{"type": "node"}
for i := range num {
m, err := lp.NewMetric(fmt.Sprintf("testmetric%d", i), tags, nil, 42.0, time.Unix(1, 0))
if err != nil {
t.Fatalf("failed to create message: %s", err.Error())
}
msgs = append(msgs, m)
}
return msgs
}
// With interval_timestamp enabled, all metrics forwarded after a tick must
// carry that tick's timestamp, never the previous interval's
func TestIntervalTimestamp(t *testing.T) {
ticker := &fakeTicker{}
var wg sync.WaitGroup
r, err := New(ticker, &wg, json.RawMessage(`{"interval_timestamp": true}`))
if err != nil {
t.Fatalf("failed to setup metric router: %s", err.Error())
}
coll := make(chan lp.CCMessage, 100)
out := make(chan lp.CCMessage, 100)
r.AddCollectorInput(coll)
r.AddOutput(out)
r.Start()
receiveAll := func(num int) []lp.CCMessage {
received := make([]lp.CCMessage, 0, num)
for len(received) < num {
select {
case m := <-out:
received = append(received, m)
case <-time.After(5 * time.Second):
t.Fatalf("received only %d of %d messages", len(received), num)
}
}
return received
}
for interval, tickTime := range []time.Time{time.Unix(1000, 0), time.Unix(1010, 0)} {
ticker.tick(tickTime)
msgs := genMessages(t, 20)
for _, m := range msgs {
coll <- m
}
for i, m := range receiveAll(len(msgs)) {
if !m.Time().Equal(tickTime) {
t.Errorf("interval %d message %d: got timestamp %v, want %v", interval, i, m.Time(), tickTime)
}
}
}
closed := make(chan struct{})
go func() {
r.Close()
close(closed)
}()
select {
case <-closed:
case <-time.After(5 * time.Second):
t.Fatal("Close() did not terminate")
}
wg.Wait()
}
+4 -2
View File
@@ -30,8 +30,8 @@ Afterwards, you can add channels:
```golang
t := MultiChanTicker(duration)
c1 := make(chan time.Time)
c2 := make(chan time.Time)
c1 := make(chan time.Time, 1)
c2 := make(chan time.Time, 1)
t.AddChannel(c1)
t.AddChannel(c2)
@@ -46,3 +46,5 @@ for {
```
The result should be the same `time.Time` output in both channels, notified "simultaneously".
Ticks are delivered with a non-blocking send: a consumer that has not yet read the previous tick does not stall the ticker (which would silently drop `time.Ticker` fires for all consumers); instead, the tick for that consumer is skipped and a warning is logged. Register buffered channels (capacity 1) so a consumer that is briefly busy at tick time does not lose the tick.
+16 -4
View File
@@ -8,6 +8,8 @@
package multiChanTicker
import (
"fmt"
"sync"
"time"
cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
@@ -15,6 +17,7 @@ import (
type multiChanTicker struct {
ticker *time.Ticker
mutex sync.Mutex // protects channels, which is appended to while the tick goroutine iterates it
channels []chan time.Time
done chan bool
}
@@ -40,21 +43,30 @@ func (t *multiChanTicker) Init(duration time.Duration) {
return
case ts := <-t.ticker.C:
cclog.ComponentDebug("MultiChanTicker", "Tick", ts)
for _, c := range t.channels {
t.mutex.Lock()
for i, c := range t.channels {
// Non-blocking send: a consumer that has not yet read the
// previous tick must not stall the ticker, otherwise
// time.Ticker silently drops fires for ALL consumers
select {
case <-t.done:
done()
return
case c <- ts:
default:
cclog.ComponentWarn("MultiChanTicker", fmt.Sprintf("consumer %d did not read previous tick, dropping tick %v", i, ts))
}
}
t.mutex.Unlock()
}
}
}()
}
func (t *multiChanTicker) AddChannel(channel chan time.Time) {
if cap(channel) == 0 {
cclog.ComponentWarn("MultiChanTicker", "unbuffered channel registered, ticks may be dropped if the consumer is not ready")
}
t.mutex.Lock()
t.channels = append(t.channels, channel)
t.mutex.Unlock()
}
func (t *multiChanTicker) Close() {
@@ -0,0 +1,51 @@
package multiChanTicker
import (
"testing"
"time"
)
// A consumer that never reads its channel must not stall the ticker
// or starve the other consumers
func TestStalledConsumerDoesNotStarveOthers(t *testing.T) {
stalled := make(chan time.Time, 1) // never read
fast := make(chan time.Time, 1)
ticker := NewTicker(10 * time.Millisecond)
defer ticker.Close()
ticker.AddChannel(stalled)
ticker.AddChannel(fast)
received := 0
deadline := time.After(5 * time.Second)
for received < 5 {
select {
case <-fast:
received++
case <-deadline:
t.Fatalf("received only %d ticks while another consumer stalled", received)
}
}
}
// Close() must return promptly even if no consumer reads its channel
func TestCloseWithStalledConsumer(t *testing.T) {
stalled := make(chan time.Time, 1) // never read
ticker := NewTicker(10 * time.Millisecond)
ticker.AddChannel(stalled)
// Let some ticks fire and be dropped
time.Sleep(50 * time.Millisecond)
closed := make(chan struct{})
go func() {
ticker.Close()
close(closed)
}()
select {
case <-closed:
case <-time.After(5 * time.Second):
t.Fatal("Close() blocked with a stalled consumer")
}
}
+1 -1
View File
@@ -6,7 +6,7 @@ Installed-Size: {INSTALLED_SIZE}
Architecture: {ARCH}
Maintainer: thomas.gruber@fau.de
Depends: libc6 (>= 2.2.1)
Build-Depends: debhelper-compat (= 13), git, golang-go
Build-Depends: debhelper-compat (= 13), git, golang-go, libdrm-dev
Description: Metric collection daemon from the ClusterCockpit suite
Homepage: https://github.com/ClusterCockpit/cc-metric-collector
Source: cc-metric-collector
+2 -2
View File
@@ -29,7 +29,7 @@ make
%install
install -Dpm 0750 %{name} %{buildroot}%{_bindir}/%{name}
install -Dpm 0755 %{name} %{buildroot}%{_bindir}/%{name}
install -Dpm 0600 example-configs/config.json %{buildroot}%{_sysconfdir}/%{name}/%{name}.json
install -Dpm 0600 example-configs/collectors.json %{buildroot}%{_sysconfdir}/%{name}/collectors.json
install -Dpm 0600 example-configs/sinks.json %{buildroot}%{_sysconfdir}/%{name}/sinks.json
@@ -54,7 +54,7 @@ install -Dpm 0644 scripts/%{name}.sysusers %{buildroot}%{_sysusersdir}/%{name}.c
%files
# Binary
%attr(-,clustercockpit,clustercockpit) %{_bindir}/%{name}
%attr(-,root,root) %{_bindir}/%{name}
# Config
%dir %{_sysconfdir}/%{name}
%attr(0600,clustercockpit,clustercockpit) %config(noreplace) %{_sysconfdir}/%{name}/%{name}.json