Compare commits

..

2 Commits

Author SHA1 Message Date
Thomas Roehl 5d55ee7a77 Fix lint errors 2026-06-02 16:35:11 +02:00
Thomas Roehl 5938368a76 Add collector to get Nvidia GPM metrics 2026-06-02 13:52:44 +02:00
19 changed files with 589 additions and 712 deletions
+35 -36
View File
@@ -21,37 +21,36 @@ import (
// Map of all available metric collectors
var AvailableCollectors = map[string]MetricCollector{
"likwid": new(LikwidCollector),
"loadavg": new(LoadavgCollector),
"memstat": new(MemstatCollector),
"netstat": new(NetstatCollector),
"ibstat": new(InfinibandCollector),
"lustrestat": new(LustreCollector),
"cpustat": new(CpustatCollector),
"topprocs": new(TopProcsCollector),
"nvidia": new(NvidiaCollector),
"customcmd": new(CustomCmdCollector),
"iostat": new(IOstatCollector),
"diskstat": new(DiskstatCollector),
"tempstat": new(TempCollector),
"ipmistat": new(IpmiCollector),
"gpfs": new(GpfsCollector),
"cpufreq": new(CPUFreqCollector),
"cpufreq_cpuinfo": new(CPUFreqCpuInfoCollector),
"nfs3stat": new(Nfs3Collector),
"nfs4stat": new(Nfs4Collector),
"numastats": new(NUMAStatsCollector),
"beegfs_meta": new(BeegfsMetaCollector),
"beegfs_storage": new(BeegfsStorageCollector),
"rapl": new(RAPLCollector),
"rocm_smi": new(RocmSmiCollector),
"self": new(SelfCollector),
"schedstat": new(SchedstatCollector),
"nfsiostat": new(NfsIOStatCollector),
"slurm_cgroup": new(SlurmCgroupCollector),
"smartmon": new(SmartMonCollector),
"lenovo_dense_power": new(LenovoDensePowerCollector),
"megware_eureka": new(MegwareEurekaCollector),
"likwid": new(LikwidCollector),
"loadavg": new(LoadavgCollector),
"memstat": new(MemstatCollector),
"netstat": new(NetstatCollector),
"ibstat": new(InfinibandCollector),
"lustrestat": new(LustreCollector),
"cpustat": new(CpustatCollector),
"topprocs": new(TopProcsCollector),
"nvidia": new(NvidiaCollector),
"customcmd": new(CustomCmdCollector),
"iostat": new(IOstatCollector),
"diskstat": new(DiskstatCollector),
"tempstat": new(TempCollector),
"ipmistat": new(IpmiCollector),
"gpfs": new(GpfsCollector),
"cpufreq": new(CPUFreqCollector),
"cpufreq_cpuinfo": new(CPUFreqCpuInfoCollector),
"nfs3stat": new(Nfs3Collector),
"nfs4stat": new(Nfs4Collector),
"numastats": new(NUMAStatsCollector),
"beegfs_meta": new(BeegfsMetaCollector),
"beegfs_storage": new(BeegfsStorageCollector),
"rapl": new(RAPLCollector),
"rocm_smi": new(RocmSmiCollector),
"self": new(SelfCollector),
"schedstat": new(SchedstatCollector),
"nfsiostat": new(NfsIOStatCollector),
"slurm_cgroup": new(SlurmCgroupCollector),
"smartmon": new(SmartMonCollector),
"nvidia_gpm": new(NvidiaGPMCollector),
}
// Metric collector manager data structure
@@ -101,17 +100,17 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
// Initialize configured collectors
for collectorName, collectorCfg := range cm.config {
if _, found := AvailableCollectors[collectorName]; !found {
cclog.ComponentError("CollectorManager", "SKIP unknown collector", collectorName)
cclog.ComponentErrorf("CollectorManager", "SKIP unknown collector %s", collectorName)
continue
}
collector := AvailableCollectors[collectorName]
err := collector.Init(collectorCfg)
if err != nil {
cclog.ComponentError("CollectorManager", fmt.Sprintf("Collector %s initialization failed: %v", collectorName, err))
cclog.ComponentErrorf("CollectorManager", "Collector %s initialization failed: %v", collectorName, err)
continue
}
cclog.ComponentDebug("CollectorManager", "ADD COLLECTOR", collector.Name())
cclog.ComponentDebugf("CollectorManager", "ADD COLLECTOR %s", collector.Name())
if collector.Parallel() {
cm.collectors = append(cm.collectors, collector)
} else {
@@ -157,7 +156,7 @@ func (cm *collectorManager) Start() {
return
default:
// Read metrics from collector c via goroutine
cclog.ComponentDebug("CollectorManager", c.Name(), t)
cclog.ComponentDebugf("CollectorManager: Read %s at %v", c.Name(), t)
cm.collector_wg.Add(1)
go func(myc MetricCollector) {
myc.Read(cm.duration, cm.output)
@@ -175,7 +174,7 @@ func (cm *collectorManager) Start() {
return
default:
// Read metrics from collector c
cclog.ComponentDebug("CollectorManager", c.Name(), t)
cclog.ComponentDebugf("CollectorManager: Read %s at %v", c.Name(), t)
c.Read(cm.duration, cm.output)
}
}
-6
View File
@@ -116,9 +116,6 @@ func (m *CPUFreqCpuInfoCollector) Init(_ json.RawMessage) error {
physicalPackageID = ""
}
}
if err := scanner.Err(); err != nil {
return fmt.Errorf("%s Init(): Call to scanner.Err failed: %w", m.name, err)
}
if err := file.Close(); err != nil {
return fmt.Errorf("%s Init(): Call to file.Close() failed: %w", m.name, err)
@@ -182,9 +179,6 @@ func (m *CPUFreqCpuInfoCollector) Read(interval time.Duration, output chan lp.CC
}
}
}
if err := scanner.Err(); err != nil {
cclog.ComponentErrorf(m.name, "Read(): Call to scanner.Err failed: %s", err.Error())
}
}
func (m *CPUFreqCpuInfoCollector) Close() {
+6 -12
View File
@@ -44,7 +44,7 @@ type CpustatCollector struct {
func (m *CpustatCollector) Init(config json.RawMessage) error {
m.name = "CpustatCollector"
if err := m.setup(); err != nil {
return fmt.Errorf("%s Init(): setup() call failed: %s", m.name, err.Error())
return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
}
m.parallel = true
m.meta = map[string]string{
@@ -58,7 +58,7 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
d := json.NewDecoder(bytes.NewReader(config))
d.DisallowUnknownFields()
if err := d.Decode(&m.config); err != nil {
return fmt.Errorf("%s Init(): Error decoding JSON config: %s", m.name, err.Error())
return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
}
}
matches := map[string]int{
@@ -85,7 +85,7 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
// Check input file
file, err := os.Open(CPUSTATFILE)
if err != nil {
return fmt.Errorf("%s Init(): Failed to open file '%s': %s", m.name, CPUSTATFILE, err.Error())
return fmt.Errorf("%s Init(): Failed to open file '%s': %w", m.name, CPUSTATFILE, err)
}
// Pre-generate tags for all CPUs
@@ -117,13 +117,10 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
num_cpus++
}
}
if err := scanner.Err(); err != nil {
return fmt.Errorf("%s Init(): Call to scanner.Err failed: %s", m.name, err.Error())
}
// Close file
if err := file.Close(); err != nil {
return fmt.Errorf("%s Init(): Failed to close file '%s': %s", m.name, CPUSTATFILE, err.Error())
return fmt.Errorf("%s Init(): Failed to close file '%s': %w", m.name, CPUSTATFILE, err)
}
m.lastTimestamp = time.Now()
@@ -176,13 +173,13 @@ func (m *CpustatCollector) Read(interval time.Duration, output chan lp.CCMessage
if err != nil {
cclog.ComponentError(
m.name,
fmt.Sprintf("Read(): Failed to open file '%s': %v", CPUSTATFILE, err.Error()))
fmt.Sprintf("Read(): Failed to open file '%s': %v", CPUSTATFILE, err))
}
defer func() {
if err := file.Close(); err != nil {
cclog.ComponentError(
m.name,
fmt.Sprintf("Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err.Error()))
fmt.Sprintf("Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err))
}
}()
@@ -197,9 +194,6 @@ func (m *CpustatCollector) Read(interval time.Duration, output chan lp.CCMessage
num_cpus++
}
}
if err := scanner.Err(); err != nil {
cclog.ComponentErrorf(m.name, "Init(): Call to scanner.Err failed: %s", err.Error())
}
if !m.config.excludeNumCPUs {
if num_cpus_metric, err := lp.NewMetric("num_cpus", m.nodetags, m.meta, num_cpus, now); err == nil {
-5
View File
@@ -147,11 +147,6 @@ mountLoop:
}
}
}
if err := scanner.Err(); err != nil {
cclog.ComponentErrorf(
m.name,
"Read(): Call to scanner.Err failed: %s", err.Error())
}
if m.allowedMetrics["part_max_used"] {
y, err := lp.NewMetric("part_max_used", map[string]string{"type": "node"}, m.meta, int(part_max_used), time.Now())
if err == nil {
+63 -93
View File
@@ -23,29 +23,20 @@ import (
"golang.org/x/sys/unix"
)
// 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"
)
const IB_BASEPATH = "/sys/class/infiniband/"
type InfinibandCollectorMetric struct {
name string
path string
unit string
unitRates string
scaleByFourLanes bool
addToIBTotal bool
addToIBTotalPkgs bool
lastState uint64
lastStateAvailable bool
name string
path string
unit string
scale int64
addToIBTotal bool
addToIBTotalPkgs bool
lastState int64
}
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
@@ -65,7 +56,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 ibBasePath
// Init initializes the Infiniband collector by walking through files below IB_BASEPATH
func (m *InfinibandCollector) Init(config json.RawMessage) error {
// Check if already initialized
if m.init {
@@ -96,7 +87,7 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
}
// Loop for all InfiniBand directories
globPattern := filepath.Join(ibBasePath, "*", "ports", "*")
globPattern := filepath.Join(IB_BASEPATH, "*", "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)
@@ -131,42 +122,36 @@ 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: ibDataUnit,
unitRates: ibDataRateUnit,
scaleByFourLanes: true,
addToIBTotal: true,
name: "ib_recv",
path: filepath.Join(countersDir, "port_rcv_data"),
unit: "bytes",
scale: 4,
addToIBTotal: true,
lastState: -1,
},
{
// Total number of data octets, divided by 4 (lanes), transmitted on all VLs.
// This is 64 bit counter
name: "ib_xmit",
path: filepath.Join(countersDir, "port_xmit_data"),
unit: ibDataUnit,
unitRates: ibDataRateUnit,
scaleByFourLanes: true,
addToIBTotal: true,
name: "ib_xmit",
path: filepath.Join(countersDir, "port_xmit_data"),
unit: "bytes",
scale: 4,
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: ibPkgUnit,
unitRates: ibPkgRateUnit,
unit: "packets",
scale: 1,
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: ibPkgUnit,
unitRates: ibPkgRateUnit,
unit: "packets",
scale: 1,
addToIBTotalPkgs: true,
lastState: -1,
},
}
for _, counter := range portCounterFiles {
@@ -178,7 +163,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,
@@ -199,7 +184,7 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
return nil
}
// Read reads Infiniband counter files below ibBasePath
// Read reads Infiniband counter files below IB_BASEPATH
func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMessage) {
// Check if already initialized
if !m.init {
@@ -216,9 +201,9 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMess
for i := range m.info {
info := &m.info[i]
var ibTotal, ibTotalPkts uint64 // sum of xmit and recv counters
var ibTotalBw, ibTotalPktsBw float64 // sum of xmit and recv rates
var ibTotalBwAvailable, ibTotalPktsBwAvailable bool
var ib_total, ib_total_last_state,
ib_total_pkts, ib_total_pkts_last_state int64
var ib_total_last_state_available, ib_total_pkts_last_state_available bool
for i := range info.portCounterFiles {
counterDef := &info.portCounterFiles[i]
@@ -228,30 +213,24 @@ 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 uint64
vRawCounter, err := strconv.ParseUint(data, 10, 64)
// convert counter to int64
v, err := strconv.ParseInt(data, 10, 64)
if err != nil {
cclog.ComponentError(
m.name,
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
fmt.Sprintf("Read(): Failed to convert Infininiband metrice %s='%s' to int64: %v", counterDef.name, data, err))
continue
}
vScaledCounter := vRawCounter
if counterDef.scaleByFourLanes {
vScaledCounter *= uint64(4)
}
// Scale raw value
v *= counterDef.scale
// Send absolut values
if m.config.SendAbsoluteValues {
if y, err := lp.NewMetric(counterDef.name, info.tagSet, m.meta, vScaledCounter, now); err == nil {
if y, err := lp.NewMetric(counterDef.name, info.tagSet, m.meta, v, now); err == nil {
y.AddMeta("unit", counterDef.unit)
output <- y
}
@@ -259,72 +238,63 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMess
// Send derived values
if m.config.SendDerivedValues {
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 counterDef.lastState >= 0 {
rate := float64((v - counterDef.lastState)) / timeDiff
if y, err := lp.NewMetric(counterDef.name+"_bw", info.tagSet, m.meta, rate, now); err == nil {
y.AddMeta("unit", counterDef.unitRates)
y.AddMeta("unit", counterDef.unit+"/sec")
output <- y
}
// Sum up rates for total rates
// Sum up total values of last state
if m.config.SendTotalValues {
switch {
case counterDef.addToIBTotal:
ibTotalBw += rate
ibTotalBwAvailable = true
ib_total_last_state += counterDef.lastState
ib_total_last_state_available = true
case counterDef.addToIBTotalPkgs:
ibTotalPktsBw += rate
ibTotalPktsBwAvailable = true
ib_total_pkts_last_state += counterDef.lastState
ib_total_pkts_last_state_available = true
}
}
}
counterDef.lastState = vRawCounter
counterDef.lastStateAvailable = true
counterDef.lastState = v
}
// Sum up total values
if m.config.SendTotalValues {
switch {
case counterDef.addToIBTotal:
ibTotal += vScaledCounter
ib_total += v
case counterDef.addToIBTotalPkgs:
ibTotalPkts += vScaledCounter
ib_total_pkts += v
}
}
}
// Send total values
if m.config.SendTotalValues {
if y, err := lp.NewMetric("ib_total", info.tagSet, m.meta, ibTotal, now); err == nil {
y.AddMeta("unit", ibDataUnit)
if y, err := lp.NewMetric("ib_total", info.tagSet, m.meta, ib_total, now); err == nil {
y.AddMeta("unit", "bytes")
output <- y
}
if y, err := lp.NewMetric("ib_total_pkts", info.tagSet, m.meta, ibTotalPkts, now); err == nil {
y.AddMeta("unit", ibPkgUnit)
if y, err := lp.NewMetric("ib_total_pkts", info.tagSet, m.meta, ib_total_pkts, now); err == nil {
y.AddMeta("unit", "packets")
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)
if m.config.SendDerivedValues && ib_total_last_state_available {
rate := float64((ib_total - ib_total_last_state)) / timeDiff
if y, err := lp.NewMetric("ib_total_bw", info.tagSet, m.meta, rate, now); err == nil {
y.AddMeta("unit", "bytes/sec")
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)
if m.config.SendDerivedValues && ib_total_pkts_last_state_available {
rate := float64((ib_total_pkts - ib_total_pkts_last_state)) / timeDiff
if y, err := lp.NewMetric("ib_total_pkts_bw", info.tagSet, m.meta, rate, now); err == nil {
y.AddMeta("unit", "packets/sec")
output <- y
}
}
+6 -14
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: %s", m.name, err.Error())
return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
}
if len(config) > 0 {
d := json.NewDecoder(bytes.NewReader(config))
d.DisallowUnknownFields()
if err := d.Decode(&m.config); err != nil {
return fmt.Errorf("%s Init(): Error decoding JSON config: %s", m.name, err.Error())
return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
}
}
// https://www.kernel.org/doc/html/latest/admin-guide/iostats.html
@@ -89,7 +89,7 @@ func (m *IOstatCollector) Init(config json.RawMessage) error {
}
file, err := os.Open(IOSTATFILE)
if err != nil {
return fmt.Errorf("%s Init(): Failed to open file \"%s\": %s", m.name, IOSTATFILE, err.Error())
return fmt.Errorf("%s Init(): Failed to open file \"%s\": %w", m.name, IOSTATFILE, err)
}
scanner := bufio.NewScanner(file)
@@ -130,11 +130,8 @@ func (m *IOstatCollector) Init(config json.RawMessage) error {
lastValues: lastValues,
}
}
if err := scanner.Err(); err != nil {
return fmt.Errorf("%s Init(): Failed to scan content of file %q: %s", m.name, IOSTATFILE, err.Error())
}
if err := file.Close(); err != nil {
return fmt.Errorf("%s Init(): Failed to close file \"%s\": %s", m.name, IOSTATFILE, err.Error())
return fmt.Errorf("%s Init(): Failed to close file \"%s\": %w", m.name, IOSTATFILE, err)
}
m.init = true
@@ -150,14 +147,14 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMessage)
if err != nil {
cclog.ComponentError(
m.name,
fmt.Sprintf("Read(): Failed to open file '%s': %s", IOSTATFILE, err.Error()))
fmt.Sprintf("Read(): Failed to open file '%s': %v", IOSTATFILE, err))
return
}
defer func() {
if err := file.Close(); err != nil {
cclog.ComponentError(
m.name,
fmt.Sprintf("Read(): Failed to close file '%s': %s", IOSTATFILE, err.Error()))
fmt.Sprintf("Read(): Failed to close file '%s': %v", IOSTATFILE, err))
}
}()
@@ -201,11 +198,6 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMessage)
}
m.devices[device] = entry
}
if err := scanner.Err(); err != nil {
cclog.ComponentErrorf(
m.name,
"Read(): Failed to scan content of file %q: %s", IOSTATFILE, err.Error())
}
}
func (m *IOstatCollector) Close() {
+5 -29
View File
@@ -28,15 +28,13 @@ type IpmiCollector struct {
metricCollector
config struct {
IpmitoolPath string `json:"ipmitool_path"`
IpmisensorsPath string `json:"ipmisensors_path"`
Sudo bool `json:"use_sudo"`
IncludeMetrics []string `json:"include_metrics"`
IpmitoolPath string `json:"ipmitool_path"`
IpmisensorsPath string `json:"ipmisensors_path"`
Sudo bool `json:"use_sudo"`
}
ipmitool string
ipmisensors string
includeMetrics map[string]bool
ipmitool string
ipmisensors string
}
func (m *IpmiCollector) Init(config json.RawMessage) error {
@@ -66,15 +64,6 @@ func (m *IpmiCollector) Init(config json.RawMessage) error {
}
}
// Read metrics to include
m.includeMetrics = make(map[string]bool)
for _, metric := range m.config.IncludeMetrics {
metric = strings.ToLower(strings.TrimSpace(metric))
if metric != "" {
m.includeMetrics[metric] = true
}
}
m.ipmitool = m.config.IpmitoolPath
m.ipmisensors = m.config.IpmisensorsPath
@@ -156,11 +145,6 @@ func (m *IpmiCollector) readIpmiTool(output chan lp.CCMessage) error {
continue
}
name := strings.ToLower(strings.ReplaceAll(strings.TrimSpace(lv[0]), " ", "_"))
if len(m.includeMetrics) > 0 && !m.includeMetrics[name] {
continue
}
unit := strings.TrimSpace(lv[2])
switch unit {
case "Volts":
@@ -181,9 +165,6 @@ func (m *IpmiCollector) readIpmiTool(output chan lp.CCMessage) error {
y.AddMeta("unit", unit)
output <- y
}
if err := scanner.Err(); err != nil {
return fmt.Errorf("failed to scan output of command: %s", err.Error())
}
// Wait for command end
if err := command.Wait(); err != nil {
@@ -228,11 +209,6 @@ func (m *IpmiCollector) readIpmiSensors(output chan lp.CCMessage) error {
continue
}
name := strings.ToLower(strings.ReplaceAll(lv[1], " ", "_"))
if len(m.includeMetrics) > 0 && !m.includeMetrics[name] {
continue
}
y, err := lp.NewMetric(name, map[string]string{"type": "node"}, m.meta, v, time.Now())
if err != nil {
cclog.ComponentErrorf(m.name, "Failed to create message: %v", err)
+1 -4
View File
@@ -15,8 +15,7 @@ hugo_path: docs/reference/cc-metric-collector/collectors/ipmi.md
"ipmistat": {
"ipmitool_path": "/path/to/ipmitool",
"ipmisensors_path": "/path/to/ipmi-sensors",
"use_sudo": true,
"include_metrics" : []
"use_sudo": true
}
```
@@ -37,5 +36,3 @@ Defaults: monitoring !log_allowed, !pam_session
monitoring ALL = (root) NOPASSWD:/usr/bin/ipmitool sensor
monitoring ALL = (root) NOPASSWD:/usr/sbin/ipmi-sensors --comma-separated-output --sdr-cache-recreate
```
If `include_ipmi_metrics` contains any entry, ipmistat collector will only submit these metrics. Any other values will get discarded.
-170
View File
@@ -1,170 +0,0 @@
// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
// All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.
// additional authors:
// Michael Panzlaff (NHR@FAU)
package collectors
import (
"bytes"
"encoding/hex"
"encoding/json"
"fmt"
"os/exec"
"strconv"
"strings"
"time"
cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
)
type LenovoDensePowerCollector struct {
metricCollector
config struct {
IpmitoolPath string `json:"ipmitool_path"`
Sudo bool `json:"use_sudo"`
}
ipmitool string
energyValLast float64
energyTimeLast time.Time
}
func (m *LenovoDensePowerCollector) Init(config json.RawMessage) error {
// Check if already initialized
if m.init {
return nil
}
m.name = "LenovoDensePowerCollector"
if err := m.setup(); err != nil {
return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
}
m.meta = map[string]string{
"source": m.name,
"group": "IPMI",
}
m.config.IpmitoolPath = "ipmitool"
if len(config) > 0 {
d := json.NewDecoder(bytes.NewReader(config))
d.DisallowUnknownFields()
if err := d.Decode(&m.config); err != nil {
return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
}
}
m.ipmitool = m.config.IpmitoolPath
energyVal, energyTime, err := m.readEnergy()
if err != nil {
return fmt.Errorf("energy reading test failed: %w", err)
}
m.energyValLast = energyVal
m.energyTimeLast = energyTime
m.init = true
return nil
}
func (m *LenovoDensePowerCollector) readEnergy() (float64, time.Time, error) {
argv := make([]string, 0)
if m.config.Sudo {
argv = append(argv, "sudo", "-n")
}
// The Lenovo hardware wants this magic byte string
lenovoRequestEnergyMsg := []uint8{0x3a, 0x32, 4, 2, 0, 0, 0}
argv = append(argv, m.ipmitool, "raw")
for _, val := range lenovoRequestEnergyMsg {
argv = append(argv, strconv.Itoa(int(val)))
}
cmd := exec.Command(argv[0], argv[1:]...)
var stdout bytes.Buffer
var stderr bytes.Buffer
cmd.Stdout = &stdout
cmd.Stderr = &stderr
err := cmd.Run()
if err != nil {
return 0, time.Unix(0, 0), fmt.Errorf("failed to run ipmitool: %w (stdout=%s stderr=%s)", err, stdout.String(), stderr.String())
}
data, err := hex.DecodeString(strings.ReplaceAll(strings.TrimSpace(stdout.String()), " ", ""))
if err != nil {
return 0, time.Unix(0, 0), fmt.Errorf("unable to decode ipmitool hex output: %w (stdout=%s)", err, stdout.String())
}
if len(data) != 14 {
return 0, time.Unix(0, 0), fmt.Errorf("result must be 14 bytes as specified in the documentation")
}
// data is laid out like this:
// data[0 to 1]: FIFO index of value (for debug only)
// data[2 to 5]: Integer part of energy in J (little endian)
// data[6 to 7]: Fraction part of energy in mJ (little endian)
// data[8 to 11]: Integer part of Unix time of measurement in seconds (little endian)
// data[12 to 13]: Fraction part of Unix time of measurement in milliseconds (little endian)
wholeJoules := (uint32(data[2]) << 0) | (uint32(data[3]) << 8) | (uint32(data[4]) << 16) | (uint32(data[5]) << 24)
milliJoules := uint16(data[6]) | (uint16(data[7]) << 8)
finalJoules := float64(wholeJoules) + float64(milliJoules)*1e-3
wholeSeconds := (uint32(data[8]) << 0) | (uint32(data[9]) << 8) | (uint32(data[10]) << 16) | (uint32(data[11]) << 24)
milliSeconds := uint16(data[12]) | (uint16(data[13]) << 8)
finalTime := time.Unix(int64(wholeSeconds), (int64(milliSeconds) * 1000000))
return finalJoules, finalTime, nil
}
func (m *LenovoDensePowerCollector) Read(interval time.Duration, output chan lp.CCMessage) {
// Check if already initialized
if !m.init {
return
}
energyVal, energyTime, err := m.readEnergy()
if err != nil {
cclog.ComponentErrorf(m.name, "readEnergy failed: %v", err)
return
}
powerVal := 0.0
if energyTime.Compare(m.energyTimeLast) != 0 {
// Check for overflow
energyValDiff := int64(energyVal) - int64(m.energyValLast)
if energyVal < m.energyValLast {
energyValDiff += int64(0x100000000)
}
energyTimeDiff := energyTime.Sub(m.energyTimeLast)
if energyTime.Before(m.energyTimeLast) {
energyTimeDiff = energyTimeDiff + 0x100000000*time.Second
}
powerVal = float64(energyValDiff) / energyTimeDiff.Seconds()
m.energyValLast = energyVal
m.energyTimeLast = energyTime
}
metric, err := lp.NewMetric("lenovo_node_power", map[string]string{"type": "node"}, m.meta, powerVal, energyTime)
if err != nil {
cclog.ComponentErrorf(m.name, "lp.NewMetric failed: %v", err)
return
}
metric.AddMeta("unit", "Watts")
output <- metric
}
func (m *LenovoDensePowerCollector) Close() {
m.init = false
}
-51
View File
@@ -1,51 +0,0 @@
<!--
---
title: Lenovo Dense Power collector
description: Collect power of Lenovo Dense machines using ipmitool raw
categories: [cc-metric-collector]
tags: ['Admin']
weight: 2
hugo_path: docs/reference/cc-metric-collector/collectors/lenovoDensePower.md
---
-->
## `lenovo_dense_power` collector
```json
"lenovo_dense_power": {
"ipmitool_path": "/path/to/ipmitool",
"use_sudo": true,
"include_metrics" : []
}
```
The `lenovo_dense_power` collector reads power from Lenovo Dense machines via `ipmitool` (`ipmitool raw ...`). This collector is known to only work on the following machines. Others are not supported, so use at your *OWN* risk:
System | Compatibility | Power measured
--- | --- | ---
Lenovo ThinkSystem SD650 V2/V3 | untested, but should work | node + riser1 + riser2
Lenovo ThinkSystem SD650-N V2/ SD650-I V3 | untested, but should work | node + riser1 + riser2 + gpus
Lenovo ThinkSystem SD665-N V3 | tested, known to work | node + riser1 + riser2 + gpus
To test if your system *may* be supported, you can run the following command *at your own risk*:
```
$ ipmitool raw 58 50 4 2 0 0 0
c4 00 83 a9 cb d3 a2 03 df db 47 6a d8 01
```
Exactly 14 bytes should be returned.
In case not or another error occurs your hardware is likely not supported.
In addition, `ipmitool` typically require root to run.
In order to run `cc-metric-collector` without root priviliges, you can enable `use_sudo`.
Add a file like this in `/etc/sudoers.d/` to allow `cc-metric-collector` to run the required commands:
```
# Do not log the following sudo commands from monitoring, since this causes a lot of log spam.
# However keep log_denied enabled, to detect failures
Defaults: monitoring !log_allowed, !pam_session
# Allow to use ipmitool for Lenovo power readings
monitoring ALL = (root) NOPASSWD:/usr/bin/ipmitool raw 58 50 4 2 0 0 0
```
+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/runtime/thread"
"golang.design/x/thread"
)
const (
-194
View File
@@ -1,194 +0,0 @@
// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
// All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.
// additional authors:
// Michael Panzlaff (NHR@FAU)
package collectors
import (
"bytes"
"encoding/json"
"fmt"
"os/exec"
"time"
cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
)
// MPS refers to Monolithic Power Systems, from which the MP5922 is used
// to measure telemetry on the Eureka platform.
type mpsData struct {
Cnt int64 `json:"Cnt"` // No idea what this is, ignore
Energy float64 `json:"Energy"`
Vin float64 `json:"Vin"`
Iin float64 `json:"Iin"`
Pin float64 `json:"Pin"`
PinAvg float64 `json:"PinAvg"`
Vout float64 `json:"Vout"`
Iout float64 `json:"Iout"`
Pout float64 `json:"Pout"`
StandbyVout float64 `json:"StandbyVout"`
StandbyIout float64 `json:"StandbyIout"`
StandbyPout float64 `json:"StandbyPout"`
TempBusbar float64 `json:"TempBusbar"`
TempSsd float64 `json:"TempSsd"`
TempMps float64 `json:"TempMps"`
// No idea what the ones below mean exactl. Ignore them for now.
EnergyTime int64 `json:"EnergyTime"`
EnergyAccumulator int64 `json:"EnergyAccumulator"`
EnergyRolloverCnt int64 `json:"EnergyRolloverCnt"`
EnergySampleCntU24 int64 `json:"EnergySampleCntU24"`
Timestamp time.Time
}
type MegwareEurekaCollector struct {
metricCollector
config struct {
U20Path string `json:"u20_path"`
Sudo bool `json:"use_sudo"`
}
u20path string
energyValLast float64
energyTimeLast time.Time
}
func (m *MegwareEurekaCollector) Init(config json.RawMessage) error {
// Check if already initialized
if m.init {
return nil
}
m.name = "MegwareEurekaCollector"
if err := m.setup(); err != nil {
return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
}
m.meta = map[string]string{
"source": m.name,
"group": "U20",
}
m.config.U20Path = "u20"
if len(config) > 0 {
d := json.NewDecoder(bytes.NewReader(config))
d.DisallowUnknownFields()
if err := d.Decode(&m.config); err != nil {
return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
}
}
m.u20path = m.config.U20Path
data, err := m.readMpsData()
if err != nil {
return fmt.Errorf("energy reading test failed: %w", err)
}
m.energyValLast = data.Energy
m.energyTimeLast = time.Now()
m.init = true
return nil
}
func (m *MegwareEurekaCollector) readMpsData() (*mpsData, error) {
argv := make([]string, 0)
if m.config.Sudo {
argv = append(argv, "sudo", "-n")
}
argv = append(argv, m.u20path, "values", "--read", "GET_MPS_POLL_VALUES")
cmd := exec.Command(argv[0], argv[1:]...)
var stdout bytes.Buffer
var stderr bytes.Buffer
cmd.Stdout = &stdout
cmd.Stderr = &stderr
err := cmd.Run()
if err != nil {
return nil, fmt.Errorf("failed to run u20: %w (stdout=%s stderr=%s)", err, stdout.String(), stderr.String())
}
var u20output struct {
GetMpsPollValues mpsData `json:"GET_MPS_POLL_VALUES"`
}
err = json.Unmarshal(stdout.Bytes(), &u20output)
if err != nil {
return nil, fmt.Errorf("unable to decode u20 JSON output: %w (stdout=%s)", err, stdout.String())
}
u20output.GetMpsPollValues.Timestamp = time.Now()
return &u20output.GetMpsPollValues, nil
}
func (m *MegwareEurekaCollector) Read(interval time.Duration, output chan lp.CCMessage) {
// Check if already initialized
if !m.init {
return
}
data, err := m.readMpsData()
if err != nil {
cclog.ComponentErrorf(m.name, "readMpsData failed: %v", err)
return
}
powerVal := 0.0
if data.Timestamp.After(m.energyTimeLast) {
// Important, m.energy comes in Wh, so multiply by 3600 to get Ws (aka Joule)
energyValDiff := data.Energy - m.energyValLast
energyTimeDiff := data.Timestamp.Sub(m.energyTimeLast)
powerVal = energyValDiff * 3600 / energyTimeDiff.Seconds()
m.energyValLast = data.Energy
m.energyTimeLast = data.Timestamp
}
metricNamePrefix := "eureka_"
metricMap := map[string]struct {
value any
unit string
}{
"power": {value: powerVal, unit: "Watts"},
"vin": {value: data.Vin, unit: "Volts"},
"iin": {value: data.Iin, unit: "Amperes"},
"pin": {value: data.Pin, unit: "Watts"},
"pin_avg": {value: data.PinAvg, unit: "Watts"},
"vout": {value: data.Vout, unit: "Volts"},
"iout": {value: data.Iout, unit: "Amperes"},
"pout": {value: data.Pout, unit: "Watts"},
"standby_vout": {value: data.StandbyVout, unit: "Volts"},
"standby_iout": {value: data.StandbyIout, unit: "Amperes"},
"standby_pout": {value: data.StandbyPout, unit: "Watts"},
"temp_busbar": {value: data.TempBusbar, unit: "degC"},
"temp_ssd": {value: data.TempSsd, unit: "degC"},
"temp_mps": {value: data.TempMps, unit: "degC"},
}
for metricName, metricData := range metricMap {
metricName = metricNamePrefix + metricName
metric, err := lp.NewMetric(metricName, map[string]string{"type": "node"}, m.meta, metricData.value, data.Timestamp)
if err != nil {
cclog.ComponentErrorf(m.name, "lp.NewMetric failed: %v", err)
return
}
metric.AddMeta("unit", metricData.unit)
output <- metric
}
}
func (m *MegwareEurekaCollector) Close() {
m.init = false
}
-44
View File
@@ -1,44 +0,0 @@
<!--
---
title: Megware Eureka collector
description: Collect power and other metrics of Megware Eureka machines using u20
categories: [cc-metric-collector]
tags: ['Admin']
weight: 2
hugo_path: docs/reference/cc-metric-collector/collectors/lenovoDensePower.md
---
-->
## `lenovo_dense_power` collector
```json
"megware_eureka": {
"u20_path": "/path/to/ipmitool",
"use_sudo": true
}
```
The `megware_eureka` collector reads power and other metrics from Megware Eureka machines via `u20`.
If the u20 tool is available for your machines, it's possible that this collector is compatible.
If you don't have access to u20, please contact your Megware sales person.
You can test the collector compatibility by running the following command:
```
$ u20 values --read GET_MPS_POLL_VALUES
```
If this returns a JSON with energy, voltage and current readings, you're fine.
In addition, `u20` typically requires root to run.
In order to run `cc-metric-collector` without root priviliges, you can enable `use_sudo`.
Add a file like this in `/etc/sudoers.d/` to allow `cc-metric-collector` to run the required commands:
```
# Do not log the following sudo commands from monitoring, since this causes a lot of log spam.
# However keep log_denied enabled, to detect failures
Defaults: monitoring !log_allowed, !pam_session
# Allow to use u20 for Megware power readings
monitoring ALL = (root) NOPASSWD:/usr/bin/u20 values --read GET_MPS_POLL_VALUES
```
+396
View File
@@ -0,0 +1,396 @@
package collectors
import (
"encoding/json"
"errors"
"fmt"
"slices"
"strconv"
"strings"
"time"
cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
"github.com/NVIDIA/go-nvml/pkg/nvml"
)
type NvidiaGPMMetricDef struct {
name string
outname string
id nvml.GpmMetricId
unit string
}
var NvidiaGPMMetrics []NvidiaGPMMetricDef = []NvidiaGPMMetricDef{
{
name: "GRAPHICS_UTIL",
outname: "nv_gpm_graphics_util",
id: nvml.GPM_METRIC_GRAPHICS_UTIL,
unit: "%",
},
{
name: "SM_UTIL",
outname: "nv_gpm_sm_util",
id: nvml.GPM_METRIC_SM_UTIL,
unit: "%",
},
{
name: "SM_OCCUPANCY",
outname: "nv_gpm_sm_occupancy",
id: nvml.GPM_METRIC_SM_OCCUPANCY,
unit: "%",
},
{
name: "INTEGER_UTIL",
outname: "nv_gpm_integer_util",
id: nvml.GPM_METRIC_INTEGER_UTIL,
unit: "%",
},
{
name: "ANY_TENSOR_UTIL",
outname: "nv_gpm_any_tensor_util",
id: nvml.GPM_METRIC_ANY_TENSOR_UTIL,
unit: "%",
},
{
name: "DFMA_TENSOR_UTIL",
outname: "nv_gpm_dfma_tensor_util",
id: nvml.GPM_METRIC_DFMA_TENSOR_UTIL,
unit: "%",
},
{
name: "HMMA_TENSOR_UTIL",
outname: "nv_gpm_hmma_tensor_util",
id: nvml.GPM_METRIC_HMMA_TENSOR_UTIL,
unit: "%",
},
{
name: "IMMA_TENSOR_UTIL",
outname: "nv_gpm_imma_tensor_util",
id: nvml.GPM_METRIC_IMMA_TENSOR_UTIL,
unit: "%",
},
{
name: "DRAM_BW_UTIL",
outname: "nv_gpm_dram_bw_util",
id: nvml.GPM_METRIC_DRAM_BW_UTIL,
unit: "%",
},
{
name: "FP64_UTIL",
outname: "nv_gpm_fp64_util",
id: nvml.GPM_METRIC_FP64_UTIL,
unit: "%",
},
{
name: "FP32_UTIL",
outname: "nv_gpm_fp32_util",
id: nvml.GPM_METRIC_FP32_UTIL,
unit: "%",
},
{
name: "FP16_UTIL",
outname: "nv_gpm_fp16_util",
id: nvml.GPM_METRIC_FP16_UTIL,
unit: "%",
},
}
type NvidiaGPMCollectorConfig struct {
Metrics []string `json:"metrics,omitempty"`
ExcludeDevices []string `json:"exclude_devices,omitempty"`
AddPciInfoTag bool `json:"add_pci_info_tag,omitempty"`
UsePciInfoAsTypeId bool `json:"use_pci_info_as_type_id,omitempty"`
AddUuidMeta bool `json:"add_uuid_meta,omitempty"`
AddBoardNumberMeta bool `json:"add_board_number_meta,omitempty"`
AddSerialMeta bool `json:"add_serial_meta,omitempty"`
ProcessMigDevices bool `json:"process_mig_devices,omitempty"`
UseUuidForMigDevices bool `json:"use_uuid_for_mig_device,omitempty"`
UseSliceForMigDevices bool `json:"use_slice_for_mig_device,omitempty"`
}
type NvidiaGPMCollectorDevice struct {
device nvml.Device
tags map[string]string
meta map[string]string
startTime time.Time
endTime time.Time
measurement nvml.GpmMetricsGetType
metricsLookup map[int]NvidiaGPMMetricDef
}
type NvidiaGPMCollector struct {
metricCollector
config NvidiaGPMCollectorConfig
gpus []NvidiaGPMCollectorDevice
num_gpus int
}
func (m *NvidiaGPMCollector) Init(config json.RawMessage) error {
var err error = nil
m.name = "NvidiaGPMCollector"
m.parallel = true
if err := m.setup(); err != nil {
return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
}
if len(config) > 0 {
d := json.NewDecoder(strings.NewReader(string(config)))
d.DisallowUnknownFields()
if err = d.Decode(&m.config); err != nil {
return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
}
}
m.meta = map[string]string{
"source": m.name,
"group": "NvidiaGPM",
}
// Initialize NVIDIA Management Library (NVML)
ret := nvml.Init()
// Error: NVML library not found
// (nvml.ErrorString can not be used in this case)
if ret == nvml.ERROR_LIBRARY_NOT_FOUND {
return fmt.Errorf("%s Init(): NVML library not found", m.name)
}
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
return fmt.Errorf("%s Init(): Unable to initialize NVML: %w", m.name, err)
}
// Number of NVIDIA GPUs
num_gpus, ret := nvml.DeviceGetCount()
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
return fmt.Errorf("%s Init(): Unable to get device count: %w", m.name, err)
}
// For all GPUs
m.gpus = make([]NvidiaGPMCollectorDevice, 0, num_gpus)
for i := range num_gpus {
// Skip excluded devices by ID
str_i := strconv.Itoa(i)
if slices.Contains(m.config.ExcludeDevices, str_i) {
cclog.ComponentDebugf(m.name, "Skipping excluded device %s", str_i)
continue
}
// Get device handle
device, ret := nvml.DeviceGetHandleByIndex(i)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to get device at index %d: %s", i, err.Error())
continue
}
supportInfo, ret := nvml.GpmQueryDeviceSupport(device)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to query GPM support for device at index %d: %s", i, err.Error())
continue
} else {
if supportInfo.IsSupportedDevice == uint32(nvml.FEATURE_DISABLED) {
cclog.ComponentErrorf(m.name, "Device at index %d does not support GPM metrics", i)
continue
}
}
stream, ret := nvml.GpmQueryIfStreamingEnabled(device)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to query GPM streaming for device at index %d: %s", i, err.Error())
continue
} else {
if stream == uint32(nvml.FEATURE_DISABLED) {
ret = nvml.GpmSetStreamingEnabled(device, uint32(nvml.FEATURE_ENABLED))
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to set streaming mode for device at index %d: %s", i, err.Error())
}
}
}
// Get device's PCI info
pciInfo, ret := nvml.DeviceGetPciInfo(device)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to get PCI info for device at index %d: %s", i, err.Error())
continue
}
// Create PCI ID in the common format used by the NVML.
pci_id := fmt.Sprintf(
nvml.DEVICE_PCI_BUS_ID_FMT,
pciInfo.Domain,
pciInfo.Bus,
pciInfo.Device)
// Skip excluded devices specified by PCI ID
if slices.Contains(m.config.ExcludeDevices, pci_id) {
cclog.ComponentDebugf(m.name, "Skipping excluded device %s", pci_id)
continue
}
ss, nvmlErr := nvml.GpmSampleAlloc()
if nvmlErr != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Failed to allocate GPM sample for device %d: %s", i, err.Error())
continue
}
es, nvmlErr := nvml.GpmSampleAlloc()
if nvmlErr != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Failed to allocate GPM sample for device %d: %s", i, err.Error())
continue
}
// Select which value to use as 'type-id'.
// The PCI ID is commonly required in SLURM environments because the
// numberic IDs used by SLURM and the ones used by NVML might differ
// depending on the job type. The PCI ID is more reliable but is commonly
// not recorded for a job, so it must be added manually in prologue or epilogue
// e.g. to the comment field
tid := str_i
if m.config.UsePciInfoAsTypeId {
tid = pci_id
}
// Now we got all infos together, populate the device list
g := NvidiaGPMCollectorDevice{}
// Add device handle
g.device = device
// Add tags
g.tags = map[string]string{
"type": "accelerator",
"type-id": tid,
}
// Add PCI info as tag if not already used as 'type-id'
if m.config.AddPciInfoTag && !m.config.UsePciInfoAsTypeId {
g.tags["pci_identifier"] = pci_id
}
g.meta = map[string]string{
"source": m.name,
"group": "Nvidia",
}
if m.config.AddBoardNumberMeta {
board, ret := nvml.DeviceGetBoardPartNumber(device)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentError(m.name, "Unable to get boart part number for device at index", i, ":", err.Error())
} else {
g.meta["board_number"] = board
}
}
if m.config.AddSerialMeta {
serial, ret := nvml.DeviceGetSerial(device)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentError(m.name, "Unable to get serial number for device at index", i, ":", err.Error())
} else {
g.meta["serial"] = serial
}
}
if m.config.AddUuidMeta {
uuid, ret := nvml.DeviceGetUUID(device)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentError(m.name, "Unable to get UUID for device at index", i, ":", err.Error())
} else {
g.meta["uuid"] = uuid
}
}
g.measurement.Sample1 = ss
g.measurement.Sample2 = es
g.measurement.Version = nvml.GPM_METRICS_GET_VERSION
g.metricsLookup = make(map[int]NvidiaGPMMetricDef)
metIdx := 0
for _, inmetric := range m.config.Metrics {
for _, defmetric := range NvidiaGPMMetrics {
if inmetric == defmetric.outname || inmetric == defmetric.name {
g.measurement.Metrics[metIdx] = nvml.GpmMetric{
MetricId: uint32(defmetric.id),
}
g.metricsLookup[metIdx] = defmetric
metIdx += 1
}
}
}
g.measurement.NumMetrics = uint32(metIdx)
m.gpus = append(m.gpus, g)
}
cclog.ComponentDebugf(m.name, "Found %d Nvidia GPUs with GPM support", len(m.gpus))
m.num_gpus = len(m.gpus)
m.init = true
return err
}
func (m *NvidiaGPMCollector) Read(interval time.Duration, output chan lp.CCMessage) {
var err error
if !m.init {
return
}
for i, gpu := range m.gpus {
gpu.startTime = time.Now()
nvmlErr := gpu.measurement.Sample1.Get(gpu.device)
if nvmlErr != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(nvmlErr))
cclog.ComponentError(m.name, "Unable to get start GPM sample for device at index", i, ":", err.Error())
continue
}
}
time.Sleep(interval)
for i, gpu := range m.gpus {
gpu.endTime = time.Now()
nvmlErr := gpu.measurement.Sample2.Get(gpu.device)
if nvmlErr != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(nvmlErr))
cclog.ComponentError(m.name, "Unable to get stop GPM sample for device at index", i, ":", err.Error())
continue
}
}
for i, gpu := range m.gpus {
nvmlErr := nvml.GpmMetricsGet(&gpu.measurement)
if nvmlErr != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(nvmlErr))
cclog.ComponentError(m.name, "Unable to get evaluate GPM sample for device at index", i, ":", err.Error())
continue
}
for idx, metricDef := range gpu.metricsLookup {
y, err := lp.NewMetric(metricDef.outname, gpu.tags, gpu.meta, gpu.measurement.Metrics[idx].Value, time.Now())
if err == nil {
y.AddMeta("unit", metricDef.unit)
output <- y
}
}
}
}
func (m *NvidiaGPMCollector) Close() {
if m.init {
for i, gpu := range m.gpus {
ret := gpu.measurement.Sample1.Free()
if ret != nvml.SUCCESS {
err := errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to free start sample for device at index %d: %s", i, err.Error())
}
ret = gpu.measurement.Sample2.Free()
if ret != nvml.SUCCESS {
err := errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to free stop sample for device at index %d: %s", i, err.Error())
}
}
if ret := nvml.Shutdown(); ret != nvml.SUCCESS {
cclog.ComponentError(m.name, "nvml.Shutdown() not successful")
}
m.init = false
}
}
+54
View File
@@ -0,0 +1,54 @@
<!--
---
title: "Nvidia NVML GPM metric collector"
description: Collect metrics for Nvidia GPUs using the NVML GPM interface
categories: [cc-metric-collector]
tags: ['Admin']
weight: 2
hugo_path: docs/reference/cc-metric-collector/collectors/nvidiaGPM.md
---
-->
## `nvidiaGPM` collector
```json
"nvidia_gpm": {
"metrics": [
"nv_fb_mem_used",
"nv_fan"
],
"exclude_devices": [
"0","1", "0000000:ff:01.0"
],
"process_mig_devices": false,
"use_pci_info_as_type_id": true,
"add_pci_info_tag": false,
"add_uuid_meta": false,
"add_board_number_meta": false,
"add_serial_meta": false,
"use_uuid_for_mig_device": false,
"use_slice_for_mig_device": false
}
```
The `nvidia_gpm` collector can be configured to leave out specific devices with the `exclude_devices` option. It takes IDs as supplied to the NVML with `nvmlDeviceGetHandleByIndex()` or the PCI address in NVML format (`%08X:%02X:%02X.0`). Commonly only the physical GPUs are monitored. If MIG devices should be analyzed as well, set `process_mig_devices` (adds `stype=mig,stype-id=<mig_index>`). With the options `use_uuid_for_mig_device` and `use_slice_for_mig_device`, the `<mig_index>` can be replaced with the UUID (e.g. `MIG-6a9f7cc8-6d5b-5ce0-92de-750edc4d8849`) or the MIG slice name (e.g. `1g.5gb`).
The metrics sent by the `nvidia_gpm` collector use `accelerator` as `type` tag. For the `type-id`, it uses the device handle index by default. With the `use_pci_info_as_type_id` option, the PCI ID is used instead. If both values should be added as tags, activate the `add_pci_info_tag` option. It uses the device handle index as `type-id` and adds the PCI ID as separate `pci_identifier` tag.
Optionally, it is possible to add the UUID, the board part number and the serial to the meta informations. They are not sent to the sinks (if not configured otherwise).
Available Metrics:
* `nv_gpm_graphics_util`
* `nv_gpm_sm_util`
* `nv_gpm_sm_occupancy`
* `nv_gpm_integer_util`
* `nv_gpm_any_tensor_util`
* `nv_gpm_dfma_tensor_util`
* `nv_gpm_hmma_tensor_util`
* `nv_gpm_imma_tensor_util`
* `nv_gpm_dram_bw_util`
* `nv_gpm_fp64_util`
* `nv_gpm_fp32_util`
* `nv_gpm_fp16_util`
+3 -34
View File
@@ -7,7 +7,6 @@ import (
"os/exec"
"os/user"
"path/filepath"
"regexp"
"strconv"
"strings"
"time"
@@ -46,37 +45,6 @@ type SlurmCgroupCollector struct {
const defaultCgroupBase = "/sys/fs/cgroup/system.slice/slurmstepd.scope"
var (
// Slurm cgroup v2 directory layout:
// - Slurm <= 25.11: job_<numeric job id>
// - Slurm >= 26.05: SLUID, encoded as "s" + 13 Crockford Base32 characters
jobIDDirRE = regexp.MustCompile(`^job_[0-9]+$`)
sluidDirRE = regexp.MustCompile(`(?i)^s[0-9A-HJKMNP-TV-Z]{13}$`)
)
func (m *SlurmCgroupCollector) findSlurmJobDirs() ([]string, error) {
entries, err := os.ReadDir(m.cgroupBase)
if err != nil {
return nil, err
}
jobDirs := make([]string, 0)
for _, entry := range entries {
if !entry.IsDir() {
continue
}
name := entry.Name()
if jobIDDirRE.MatchString(name) || sluidDirRE.MatchString(name) {
jobDirs = append(jobDirs, filepath.Join(m.cgroupBase, name))
}
}
return jobDirs, nil
}
func ParseCPUs(cpuset string) ([]int, error) {
var result []int
if cpuset == "" {
@@ -269,9 +237,10 @@ func (m *SlurmCgroupCollector) Read(interval time.Duration, output chan lp.CCMes
delete(m.cpuUsed, k)
}
jobDirs, err := m.findSlurmJobDirs()
globPattern := filepath.Join(m.cgroupBase, "job_*")
jobDirs, err := filepath.Glob(globPattern)
if err != nil {
cclog.ComponentError(m.name, "Error reading job directories:", err.Error())
cclog.ComponentError(m.name, "Error globbing job directories:", err.Error())
return
}
+3 -3
View File
@@ -5,12 +5,12 @@ go 1.25.0
require (
github.com/ClusterCockpit/cc-lib/v2 v2.12.0
github.com/ClusterCockpit/go-rocm-smi v0.4.0
github.com/NVIDIA/go-nvml v0.13.3-1
github.com/NVIDIA/go-nvml v0.13.0-1
github.com/PaesslerAG/gval v1.2.4
github.com/fsnotify/fsnotify v1.10.1
github.com/tklauser/go-sysconf v0.4.0
golang.design/x/runtime v0.3.0
golang.org/x/sys v0.47.0
golang.design/x/thread v0.0.0-20210122121316-335e9adffdf1
golang.org/x/sys v0.45.0
)
require (
+6 -6
View File
@@ -12,9 +12,8 @@ github.com/Microsoft/go-winio v0.6.1 h1:9/kr64B9VUZrLm5YYwbGtUJnMgqWVOdUAXu6Migc
github.com/Microsoft/go-winio v0.6.1/go.mod h1:LRdKpFKfdobln8UmuiYcKPot9D2v6svN5+sAH+4kjUM=
github.com/Microsoft/hcsshim v0.11.4 h1:68vKo2VN8DE9AdN4tnkWnmdhqdbpUFM8OF3Airm7fz8=
github.com/Microsoft/hcsshim v0.11.4/go.mod h1:smjE4dvqPX9Zldna+t5FG3rnoHhaB7QYxPRqGcpAD9w=
github.com/NVIDIA/go-nvml v0.13.0-1 h1:OLX8Jq3dONuPOQPC7rndB6+iDmDakw0XTYgzMxObkEw=
github.com/NVIDIA/go-nvml v0.13.0-1/go.mod h1:+KNA7c7gIBH7SKSJ1ntlwkfN80zdx8ovl4hrK3LmPt4=
github.com/NVIDIA/go-nvml v0.13.3-1 h1:P76U2h88OZSiMtdhRsJjSF5DXyXUqHIXKeDicVAaae0=
github.com/NVIDIA/go-nvml v0.13.3-1/go.mod h1:ahi2psRYoa+wYUBIrZPRO+wJs9lcvMhxSSkjjvsJJNQ=
github.com/PaesslerAG/gval v1.2.4 h1:rhX7MpjJlcxYwL2eTTYIOBUyEKZ+A96T9vQySWkVUiU=
github.com/PaesslerAG/gval v1.2.4/go.mod h1:XRFLwvmkTEdYziLdaCeCa5ImcGVrfQbeNUbVR+C6xac=
github.com/PaesslerAG/jsonpath v0.1.0 h1:gADYeifvlqK3R3i2cR5B4DGgxLXIPb3TRTH1mGi0jPI=
@@ -174,8 +173,8 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
golang.design/x/runtime v0.3.0 h1:8bA+GQfO8A18JXJQlQA56pEc+Sgrbo4cmLzjzmJ58ZI=
golang.design/x/runtime v0.3.0/go.mod h1:PjVKQjImLdSrvjIw4FmrnCPq9BGs5PMBHb8i6b2BN9I=
golang.design/x/thread v0.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.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI=
golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q=
golang.org/x/exp v0.0.0-20231005195138-3e424a577f31 h1:9k5exFQKQglLo+RoP+4zMjOFE14P6+vyR0baDAi0Rcs=
@@ -184,8 +183,9 @@ golang.org/x/mod v0.13.0 h1:I/DsJXRlw/8l/0c24sM9yb0T4z9liZTduXvdAWYiysY=
golang.org/x/mod v0.13.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/sys v0.0.0-20210122093101-04d7465088b8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
golang.org/x/tools v0.14.0 h1:jvNa2pY0M4r62jkRQ6RwEZZyPcymeL9XZMLBbV7U2nc=
+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.ComponentDebugf("MetricCache", "COLLECT %s COND '%s'", f.Name, f.Condition)
cclog.ComponentDebug("MetricCache", "COLLECT", f.Name, "COND", f.Condition)
var valuesFloat64 []float64
var valuesFloat32 []float32
var valuesInt []int
@@ -140,7 +140,7 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
vars["metric"] = m
value, err := f.gvalCond.EvalBool(context.Background(), vars)
if err != nil {
cclog.ComponentErrorf("MetricCache", "COLLECT %s COND '%s' : %s", f.Name, f.Condition, err.Error())
cclog.ComponentError("MetricCache", "COLLECT", f.Name, "COND", f.Condition, ":", err.Error())
continue
}
if value {
@@ -160,7 +160,7 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
case bool:
valuesBool = append(valuesBool, x)
default:
cclog.ComponentErrorf("MetricCache", "COLLECT ADD VALUE %v FAILED", v)
cclog.ComponentError("MetricCache", "COLLECT ADD VALUE", v, "FAILED")
}
}
matches = append(matches, m)
@@ -213,13 +213,13 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
vars["values"] = valuesBool
len_values = len(valuesBool)
}
cclog.ComponentDebugf("MetricCache", "EVALUATE %s METRICS %d CALC '%s'", f.Name, len_values, f.Function)
cclog.ComponentDebug("MetricCache", "EVALUATE", f.Name, "METRICS", len_values, "CALC", f.Function)
vars["metrics"] = matches
if len_values > 0 {
value, err := gval.Evaluate(f.Function, vars, c.language)
if err != nil {
cclog.ComponentErrorf("MetricCache", "EVALUATE %s METRICS %d CALC '%s': %s", f.Name, len_values, f.Function, err.Error())
cclog.ComponentError("MetricCache", "EVALUATE", f.Name, "METRICS", len_values, "CALC", f.Function, ":", err.Error())
break
}
@@ -273,12 +273,12 @@ func (c *metricAggregator) Eval(starttime time.Time, endtime time.Time, metrics
case string:
m, err = lp.NewMessage(f.Name, tags, meta, map[string]any{"value": t}, starttime)
default:
cclog.ComponentErrorf("MetricCache", "Gval returned invalid type %s skipping metric %s", t, f.Name)
cclog.ComponentError("MetricCache", "Gval returned invalid type", t, "skipping metric", f.Name)
}
if err != nil {
cclog.ComponentErrorf("MetricCache", "Cannot create metric from Gval result %v: %s", value, err.Error())
cclog.ComponentError("MetricCache", "Cannot create metric from Gval result", value, ":", err.Error())
}
cclog.ComponentDebugf("MetricCache", "SEND %s", m.ToLineProtocol(nil))
cclog.ComponentDebug("MetricCache", "SEND", m)
select {
case c.output <- m:
default:
@@ -295,12 +295,12 @@ func (c *metricAggregator) AddAggregation(name, function, condition string, tags
newcond := strings.ReplaceAll(condition, "'", "\"")
gvalCond, err := gval.Full(metricCacheLanguage).NewEvaluable(newcond)
if err != nil {
cclog.ComponentErrorf("MetricAggregator", "Cannot add aggregation, invalid if condition '%s': %s", newcond, err.Error())
cclog.ComponentError("MetricAggregator", "Cannot add aggregation, invalid if condition", newcond, ":", err.Error())
return err
}
gvalFunc, err := gval.Full(metricCacheLanguage).NewEvaluable(newfunc)
if err != nil {
cclog.ComponentErrorf("MetricAggregator", "Cannot add aggregation, invalid function condition %s: %s", newfunc, err.Error())
cclog.ComponentError("MetricAggregator", "Cannot add aggregation, invalid function condition", newfunc, ":", err.Error())
return err
}
for _, agg := range c.functions {