Compare commits

..
Author SHA1 Message Date
Thomas RoehlandMichael Panzlaff 53fdda341e Update rapl collector with powercap limits 2026-06-18 18:36:25 +02:00
30 changed files with 377 additions and 682 deletions
+4 -4
View File
@@ -132,11 +132,11 @@ func mainFunc() int {
if len(rcfg.ConfigFile.Interval) > 0 { if len(rcfg.ConfigFile.Interval) > 0 {
t, err := time.ParseDuration(rcfg.ConfigFile.Interval) t, err := time.ParseDuration(rcfg.ConfigFile.Interval)
if err != nil { if err != nil {
cclog.Errorf("Configuration value interval=%s no valid duration", rcfg.ConfigFile.Interval) cclog.Error("Configuration value 'interval' no valid duration")
} }
rcfg.Interval = t rcfg.Interval = t
if rcfg.Interval == 0 { if rcfg.Interval == 0 {
cclog.Errorf("Configuration value interval=%s must be greater than zero", rcfg.ConfigFile.Interval) cclog.Error("Configuration value 'interval' must be greater than zero")
return 1 return 1
} }
} }
@@ -145,11 +145,11 @@ func mainFunc() int {
if len(rcfg.ConfigFile.Duration) > 0 { if len(rcfg.ConfigFile.Duration) > 0 {
t, err := time.ParseDuration(rcfg.ConfigFile.Duration) t, err := time.ParseDuration(rcfg.ConfigFile.Duration)
if err != nil { if err != nil {
cclog.Error("Configuration value duration=%s no valid duration", rcfg.ConfigFile.Duration) cclog.Error("Configuration value 'duration' no valid duration")
} }
rcfg.Duration = t rcfg.Duration = t
if rcfg.Duration == 0 { if rcfg.Duration == 0 {
cclog.Error("Configuration value duration=%s must be greater than zero", rcfg.ConfigFile.Duration) cclog.Error("Configuration value 'duration' must be greater than zero")
return 1 return 1
} }
} }
+4 -4
View File
@@ -209,16 +209,16 @@ func (m *BeegfsMetaCollector) Read(interval time.Duration, output chan lp.CCMess
} else { } else {
f1, err := strconv.ParseFloat(m.matches["other"], 32) f1, err := strconv.ParseFloat(m.matches["other"], 32)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err) fmt.Sprintf("Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err))
continue continue
} }
f2, err := strconv.ParseFloat(split[i], 32) f2, err := strconv.ParseFloat(split[i], 32)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err) fmt.Sprintf("Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err))
continue continue
} }
m.matches["beegfs_cstorage_other"] = fmt.Sprintf("%f", f1+f2) m.matches["beegfs_cstorage_other"] = fmt.Sprintf("%f", f1+f2)
+4 -4
View File
@@ -200,16 +200,16 @@ func (m *BeegfsStorageCollector) Read(interval time.Duration, output chan lp.CCM
} else { } else {
f1, err := strconv.ParseFloat(m.matches["other"], 32) f1, err := strconv.ParseFloat(m.matches["other"], 32)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err) fmt.Sprintf("Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err))
continue continue
} }
f2, err := strconv.ParseFloat(split[i], 32) f2, err := strconv.ParseFloat(split[i], 32)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err) fmt.Sprintf("Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err))
continue continue
} }
m.matches["beegfs_cstorage_other"] = fmt.Sprintf("%f", f1+f2) m.matches["beegfs_cstorage_other"] = fmt.Sprintf("%f", f1+f2)
+5 -6
View File
@@ -50,7 +50,6 @@ var AvailableCollectors = map[string]MetricCollector{
"nfsiostat": new(NfsIOStatCollector), "nfsiostat": new(NfsIOStatCollector),
"slurm_cgroup": new(SlurmCgroupCollector), "slurm_cgroup": new(SlurmCgroupCollector),
"smartmon": new(SmartMonCollector), "smartmon": new(SmartMonCollector),
"nvidia_gpm": new(NvidiaGPMCollector),
} }
// Metric collector manager data structure // Metric collector manager data structure
@@ -100,17 +99,17 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
// Initialize configured collectors // Initialize configured collectors
for collectorName, collectorCfg := range cm.config { for collectorName, collectorCfg := range cm.config {
if _, found := AvailableCollectors[collectorName]; !found { if _, found := AvailableCollectors[collectorName]; !found {
cclog.ComponentErrorf("CollectorManager", "SKIP unknown collector %s", collectorName) cclog.ComponentError("CollectorManager", "SKIP unknown collector", collectorName)
continue continue
} }
collector := AvailableCollectors[collectorName] collector := AvailableCollectors[collectorName]
err := collector.Init(collectorCfg) err := collector.Init(collectorCfg)
if err != nil { if err != nil {
cclog.ComponentErrorf("CollectorManager", "Collector %s initialization failed: %v", collectorName, err) cclog.ComponentError("CollectorManager", fmt.Sprintf("Collector %s initialization failed: %v", collectorName, err))
continue continue
} }
cclog.ComponentDebugf("CollectorManager", "ADD COLLECTOR %s", collector.Name()) cclog.ComponentDebug("CollectorManager", "ADD COLLECTOR", collector.Name())
if collector.Parallel() { if collector.Parallel() {
cm.collectors = append(cm.collectors, collector) cm.collectors = append(cm.collectors, collector)
} else { } else {
@@ -156,7 +155,7 @@ func (cm *collectorManager) Start() {
return return
default: default:
// Read metrics from collector c via goroutine // Read metrics from collector c via goroutine
cclog.ComponentDebugf("CollectorManager: Read %s at %v", c.Name(), t) cclog.ComponentDebug("CollectorManager", c.Name(), t)
cm.collector_wg.Add(1) cm.collector_wg.Add(1)
go func(myc MetricCollector) { go func(myc MetricCollector) {
myc.Read(cm.duration, cm.output) myc.Read(cm.duration, cm.output)
@@ -174,7 +173,7 @@ func (cm *collectorManager) Start() {
return return
default: default:
// Read metrics from collector c // Read metrics from collector c
cclog.ComponentDebugf("CollectorManager: Read %s at %v", c.Name(), t) cclog.ComponentDebug("CollectorManager", c.Name(), t)
c.Read(cm.duration, cm.output) c.Read(cm.duration, cm.output)
} }
} }
+6 -6
View File
@@ -139,16 +139,16 @@ func (m *CPUFreqCpuInfoCollector) Read(interval time.Duration, output chan lp.CC
const cpuInfoFile = "/proc/cpuinfo" const cpuInfoFile = "/proc/cpuinfo"
file, err := os.Open(cpuInfoFile) file, err := os.Open(cpuInfoFile)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to open file '%s': %v", cpuInfoFile, err) fmt.Sprintf("Read(): Failed to open file '%s': %v", cpuInfoFile, err))
return return
} }
defer func() { defer func() {
if err := file.Close(); err != nil { if err := file.Close(); err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to close file '%s': %v", cpuInfoFile, err) fmt.Sprintf("Read(): Failed to close file '%s': %v", cpuInfoFile, err))
} }
}() }()
@@ -166,9 +166,9 @@ func (m *CPUFreqCpuInfoCollector) Read(interval time.Duration, output chan lp.CC
if !t.isHT { if !t.isHT {
value, err := strconv.ParseFloat(strings.TrimSpace(lineSplit[1]), 64) value, err := strconv.ParseFloat(strings.TrimSpace(lineSplit[1]), 64)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to convert cpu MHz '%s' to float64: %v", lineSplit[1], err) fmt.Sprintf("Read(): Failed to convert cpu MHz '%s' to float64: %v", lineSplit[1], err))
return return
} }
if y, err := lp.NewMetric("cpufreq", t.tagSet, m.meta, value, now); err == nil { if y, err := lp.NewMetric("cpufreq", t.tagSet, m.meta, value, now); err == nil {
+9 -4
View File
@@ -95,7 +95,10 @@ func (m *CPUFreqCollector) Init(config json.RawMessage) error {
} }
// Initialized // Initialized
cclog.ComponentDebugf(m.name, "initialized %d non-hyper-threading CPUs") cclog.ComponentDebug(
m.name,
"initialized",
len(m.topology), "non-hyper-threading CPUs")
m.init = true m.init = true
return nil return nil
} }
@@ -113,14 +116,16 @@ func (m *CPUFreqCollector) Read(interval time.Duration, output chan lp.CCMessage
// Read current frequency // Read current frequency
line, err := os.ReadFile(t.scalingCurFreqFile) line, err := os.ReadFile(t.scalingCurFreqFile)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, "Read(): Failed to read file '%s': %v", t.scalingCurFreqFile, err) m.name,
fmt.Sprintf("Read(): Failed to read file '%s': %v", t.scalingCurFreqFile, err))
continue continue
} }
cpuFreq, err := strconv.ParseInt(strings.TrimSpace(string(line)), 10, 64) cpuFreq, err := strconv.ParseInt(strings.TrimSpace(string(line)), 10, 64)
if err != nil { if err != nil {
cclog.ComponentError( cclog.ComponentError(
m.name, "Read(): Failed to convert CPU frequency '%s' to int64: %v", line, err) m.name,
fmt.Sprintf("Read(): Failed to convert CPU frequency '%s' to int64: %v", line, err))
continue continue
} }
+4 -4
View File
@@ -171,15 +171,15 @@ func (m *CpustatCollector) Read(interval time.Duration, output chan lp.CCMessage
file, err := os.Open(CPUSTATFILE) file, err := os.Open(CPUSTATFILE)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to open file '%s': %v", CPUSTATFILE, err) fmt.Sprintf("Read(): Failed to open file '%s': %v", CPUSTATFILE, err))
} }
defer func() { defer func() {
if err := file.Close(); err != nil { if err := file.Close(); err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err) fmt.Sprintf("Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err))
} }
}() }()
+15 -11
View File
@@ -64,9 +64,9 @@ func (m *CustomCmdCollector) Init(config json.RawMessage) error {
cmdFields := strings.Fields(c) cmdFields := strings.Fields(c)
command := exec.Command(cmdFields[0], cmdFields[1:]...) command := exec.Command(cmdFields[0], cmdFields[1:]...)
if _, err := command.Output(); err != nil { if _, err := command.Output(); err != nil {
cclog.ComponentWarnf( cclog.ComponentWarn(
m.name, m.name,
"%s Init(): Execution of command \"%s\" failed: %v", m.name, command.String(), err) fmt.Sprintf("%s Init(): Execution of command \"%s\" failed: %v", m.name, command.String(), err))
continue continue
} }
m.cmdFieldsSlice = append(m.cmdFieldsSlice, cmdFields) m.cmdFieldsSlice = append(m.cmdFieldsSlice, cmdFields)
@@ -77,7 +77,7 @@ func (m *CustomCmdCollector) Init(config json.RawMessage) error {
if _, err := os.ReadFile(fileName); err != nil { if _, err := os.ReadFile(fileName); err != nil {
cclog.ComponentWarn( cclog.ComponentWarn(
m.name, m.name,
"%s Init(): Reading of file \"%s\" failed: %v", m.name, fileName, err) fmt.Sprintf("%s Init(): Reading of file \"%s\" failed: %v", m.name, fileName, err))
continue continue
} }
m.files = append(m.files, fileName) m.files = append(m.files, fileName)
@@ -100,18 +100,20 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMessa
command := exec.Command(cmdFields[0], cmdFields[1:]...) command := exec.Command(cmdFields[0], cmdFields[1:]...)
stdout, err := command.Output() stdout, err := command.Output()
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to read command output for command \"%s\": %v", command.String(), err) fmt.Sprintf("Read(): Failed to read command output for command \"%s\": %v", command.String(), err),
)
continue continue
} }
// Read and decode influxDB line-protocol from command output // Read and decode influxDB line-protocol from command output
metrics, err := lp.FromBytes(stdout) metrics, err := lp.FromBytes(stdout)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to decode influx Message: %v", err) fmt.Sprintf("Read(): Failed to decode influx Message: %v", err),
)
continue continue
} }
for _, metric := range metrics { for _, metric := range metrics {
@@ -126,18 +128,20 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMessa
for _, filename := range m.files { for _, filename := range m.files {
input, err := os.ReadFile(filename) input, err := os.ReadFile(filename)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to read file \"%s\": %v\n", filename, err) fmt.Sprintf("Read(): Failed to read file \"%s\": %v\n", filename, err),
)
continue continue
} }
// Read and decode influxDB line-protocol from file // Read and decode influxDB line-protocol from file
metrics, err := lp.FromBytes(input) metrics, err := lp.FromBytes(input)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to decode influx Message: %v", err) fmt.Sprintf("Read(): Failed to decode influx Message: %v", err),
)
continue continue
} }
for _, metric := range metrics { for _, metric := range metrics {
+3 -3
View File
@@ -77,16 +77,16 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMessag
file, err := os.Open(MOUNTFILE) file, err := os.Open(MOUNTFILE)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to open file '%s': %v", MOUNTFILE, err) fmt.Sprintf("Read(): Failed to open file '%s': %v", MOUNTFILE, err))
return return
} }
defer func() { defer func() {
if err := file.Close(); err != nil { if err := file.Close(); err != nil {
cclog.ComponentError( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to close file '%s': %v", MOUNTFILE, err) fmt.Sprintf("Read(): Failed to close file '%s': %v", MOUNTFILE, err))
} }
}() }()
+7 -7
View File
@@ -371,7 +371,7 @@ func (m *GpfsCollector) Init(config json.RawMessage) error {
if err != nil { if err != nil {
// if using sudo, exec.lookPath will return EACCES (file mode r-x------), this can be ignored // if using sudo, exec.lookPath will return EACCES (file mode r-x------), this can be ignored
if m.config.Sudo && errors.Is(err, syscall.EACCES) { if m.config.Sudo && errors.Is(err, syscall.EACCES) {
cclog.ComponentWarnf(m.name, "got error looking for mmpmon binary '%s': %v . This is expected when using sudo, continuing.", m.config.Mmpmon, err) cclog.ComponentWarn(m.name, fmt.Sprintf("got error looking for mmpmon binary '%s': %v . This is expected when using sudo, continuing.", m.config.Mmpmon, err))
// the file was given in the config, use it // the file was given in the config, use it
p = m.config.Mmpmon p = m.config.Mmpmon
} else { } else {
@@ -517,23 +517,23 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
// return code // return code
rc, err := strconv.Atoi(key_value["_rc_"]) rc, err := strconv.Atoi(key_value["_rc_"])
if err != nil { if err != nil {
cclog.ComponentErrorf(m.name, "Read(): Failed to convert return code '%s' to int: %v", key_value["_rc_"], err) cclog.ComponentError(m.name, fmt.Sprintf("Read(): Failed to convert return code '%s' to int: %v", key_value["_rc_"], err))
continue continue
} }
if rc != 0 { if rc != 0 {
cclog.ComponentErrorf(m.name, "Read(): Filesystem '%s' is not ok.", filesystem) cclog.ComponentError(m.name, fmt.Sprintf("Read(): Filesystem '%s' is not ok.", filesystem))
continue continue
} }
// timestamp // timestamp
sec, err := strconv.ParseInt(key_value["_t_"], 10, 64) sec, err := strconv.ParseInt(key_value["_t_"], 10, 64)
if err != nil { if err != nil {
cclog.ComponentErrorf(m.name, "Read(): Failed to convert seconds '%s' to int64: %v", key_value["_t_"], err) cclog.ComponentError(m.name, fmt.Sprintf("Read(): Failed to convert seconds '%s' to int64: %v", key_value["_t_"], err))
continue continue
} }
msec, err := strconv.ParseInt(key_value["_tu_"], 10, 64) msec, err := strconv.ParseInt(key_value["_tu_"], 10, 64)
if err != nil { if err != nil {
cclog.ComponentErrorf(m.name, "Read(): Failed to convert micro seconds '%s' to int64: %v", key_value["_tu_"], err) cclog.ComponentError(m.name, fmt.Sprintf("Read(): Failed to convert micro seconds '%s' to int64: %v", key_value["_tu_"], err))
continue continue
} }
timestamp := time.Unix(sec, msec*1000) timestamp := time.Unix(sec, msec*1000)
@@ -551,7 +551,7 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
for _, metric := range GpfsAbsMetrics { for _, metric := range GpfsAbsMetrics {
value, err := strconv.ParseInt(key_value[metric.prefix], 10, 64) value, err := strconv.ParseInt(key_value[metric.prefix], 10, 64)
if err != nil { if err != nil {
cclog.ComponentErrorf(m.name, "Read(): Failed to convert %s '%s' to int64: %v", metric.desc, key_value[metric.prefix], err) cclog.ComponentError(m.name, fmt.Sprintf("Read(): Failed to convert %s '%s' to int64: %v", metric.desc, key_value[metric.prefix], err))
continue continue
} }
newstate[metric.prefix] = value newstate[metric.prefix] = value
@@ -636,7 +636,7 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
} }
} else { } else {
// the value could not be computed correctly // the value could not be computed correctly
cclog.ComponentWarnf(m.name, "Read(): Could not compute value for filesystem %s of metric %s: vold_ok = %t, vnew_ok = %t", filesystem, metric.name, vold_ok, vnew_ok) cclog.ComponentWarn(m.name, fmt.Sprintf("Read(): Could not compute value for filesystem %s of metric %s: vold_ok = %t, vnew_ok = %t", filesystem, metric.name, vold_ok, vnew_ok))
} }
} }
+4 -4
View File
@@ -225,9 +225,9 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMess
// Read counter file // Read counter file
line, err := os.ReadFile(counterDef.path) line, err := os.ReadFile(counterDef.path)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to read from file '%s': %v", counterDef.path, err) fmt.Sprintf("Read(): Failed to read from file '%s': %v", counterDef.path, err))
// Current counter can not be saved as last state // Current counter can not be saved as last state
counterDef.lastStateAvailable = false counterDef.lastStateAvailable = false
continue continue
@@ -237,9 +237,9 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMess
// convert counter to uint64 // convert counter to uint64
vRawCounter, err := strconv.ParseUint(data, 10, 64) vRawCounter, err := strconv.ParseUint(data, 10, 64)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to convert Infininiband metrice %s='%s' to uint64: %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 // Current counter can not be saved as last state
counterDef.lastStateAvailable = false counterDef.lastStateAvailable = false
continue continue
+4 -4
View File
@@ -145,16 +145,16 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMessage)
file, err := os.Open(IOSTATFILE) file, err := os.Open(IOSTATFILE)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to open file '%s': %v", IOSTATFILE, err) fmt.Sprintf("Read(): Failed to open file '%s': %v", IOSTATFILE, err))
return return
} }
defer func() { defer func() {
if err := file.Close(); err != nil { if err := file.Close(); err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to close file '%s': %v", IOSTATFILE, err) fmt.Sprintf("Read(): Failed to close file '%s': %v", IOSTATFILE, err))
} }
}() }()
+13 -19
View File
@@ -12,12 +12,6 @@ package collectors
#cgo LDFLAGS: -Wl,--unresolved-symbols=ignore-in-object-files #cgo LDFLAGS: -Wl,--unresolved-symbols=ignore-in-object-files
#include <stdlib.h> #include <stdlib.h>
#include <likwid.h> #include <likwid.h>
int cc_add_hwthread(int cpu_id) {
return HPMaddThread(cpu_id);
}
*/ */
import "C" import "C"
@@ -267,12 +261,12 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
} }
for _, metric := range evset.Metrics { for _, metric := range evset.Metrics {
// Try to evaluate the metric // Try to evaluate the metric
cclog.ComponentDebugf(m.name, "Checking %s", metric.Name) cclog.ComponentDebug(m.name, "Checking", metric.Name)
if !checkMetricType(metric.Type) { if !checkMetricType(metric.Type) {
cclog.ComponentErrorf(m.name, "Metric %s uses invalid type %s", metric.Name, metric.Type) cclog.ComponentError(m.name, "Metric", metric.Name, "uses invalid type", metric.Type)
metric.Calc = "" metric.Calc = ""
} else if !testLikwidMetricFormula(metric.Calc, params) { } else if !testLikwidMetricFormula(metric.Calc, params) {
cclog.ComponentError(m.name, "Metric %s cannot be calculated with given counters", metric.Name) cclog.ComponentError(m.name, "Metric", metric.Name, "cannot be calculated with given counters")
metric.Calc = "" metric.Calc = ""
} else { } else {
globalParams = append(globalParams, metric.Name) globalParams = append(globalParams, metric.Name)
@@ -287,13 +281,13 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
for _, metric := range m.config.Metrics { for _, metric := range m.config.Metrics {
// Try to evaluate the global metric // Try to evaluate the global metric
if !checkMetricType(metric.Type) { if !checkMetricType(metric.Type) {
cclog.ComponentErrorf(m.name, "Metric %s uses invalid type %s", metric.Name, metric.Type) cclog.ComponentError(m.name, "Metric", metric.Name, "uses invalid type", metric.Type)
metric.Calc = "" metric.Calc = ""
} else if !testLikwidMetricFormula(metric.Calc, globalParams) { } else if !testLikwidMetricFormula(metric.Calc, globalParams) {
cclog.ComponentError(m.name, "Metric %s cannot be calculated with given counters", metric.Name) cclog.ComponentError(m.name, "Metric", metric.Name, "cannot be calculated with given counters")
metric.Calc = "" metric.Calc = ""
} else if !checkMetricType(metric.Type) { } else if !checkMetricType(metric.Type) {
cclog.ComponentError(m.name, "Metric %s has invalid type", metric.Name) cclog.ComponentError(m.name, "Metric", metric.Name, "has invalid type")
metric.Calc = "" metric.Calc = ""
} else { } else {
totalMetrics++ totalMetrics++
@@ -334,7 +328,7 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
for _, c := range m.cpulist { for _, c := range m.cpulist {
m.measureThread.Call( m.measureThread.Call(
func() { func() {
retCode := C.cc_add_hwthread(C.int(c)) retCode := C.HPMaddThread(C.uint32_t(c))
if retCode != 0 { if retCode != 0 {
err := fmt.Errorf("C.HPMaddThread(%v) failed with return code %v", c, retCode) err := fmt.Errorf("C.HPMaddThread(%v) failed with return code %v", c, retCode)
cclog.ComponentError(m.name, err.Error()) cclog.ComponentError(m.name, err.Error())
@@ -381,16 +375,16 @@ func (m *LikwidCollector) takeMeasurement(evidx int, evset LikwidEventsetConfig,
// Watch changes for the lock file () // Watch changes for the lock file ()
watcher, err := fsnotify.NewWatcher() watcher, err := fsnotify.NewWatcher()
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"takeMeasurement(): Failed to create a new fsnotify.Watcher: %v", err) fmt.Sprintf("takeMeasurement(): Failed to create a new fsnotify.Watcher: %v", err))
return true, err return true, err
} }
defer func() { defer func() {
if err := watcher.Close(); err != nil { if err := watcher.Close(); err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"takeMeasurement(): Failed to close fsnotify.Watcher: %v", err) fmt.Sprintf("takeMeasurement(): Failed to close fsnotify.Watcher: %v", err))
} }
}() }()
if len(m.config.LockfilePath) > 0 { if len(m.config.LockfilePath) > 0 {
@@ -603,7 +597,7 @@ func (m *LikwidCollector) calcEventsetMetrics(evset LikwidEventsetConfig, interv
if tid >= 0 && len(metric.Calc) > 0 { if tid >= 0 && len(metric.Calc) > 0 {
value, err := agg.EvalFloat64Condition(metric.Calc, evset.results[tid]) value, err := agg.EvalFloat64Condition(metric.Calc, evset.results[tid])
if err != nil { if err != nil {
cclog.ComponentErrorf(m.name, "Calculation for metric %s failed: %s", metric.Name, err.Error()) cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
value = 0.0 value = 0.0
} }
if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) { if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) {
@@ -768,7 +762,7 @@ func (m *LikwidCollector) calcGlobalMetrics(groups []LikwidEventsetConfig, inter
// Evaluate the metric // Evaluate the metric
value, err := agg.EvalFloat64Condition(metric.Calc, params) value, err := agg.EvalFloat64Condition(metric.Calc, params)
if err != nil { if err != nil {
cclog.ComponentErrorf(m.name, "Calculation for metric %s failed: %s", metric.Name, err.Error()) cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
value = 0.0 value = 0.0
} }
if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) { if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) {
+6 -6
View File
@@ -89,9 +89,9 @@ func (m *LoadavgCollector) Read(interval time.Duration, output chan lp.CCMessage
} }
buffer, err := os.ReadFile(LOADAVGFILE) buffer, err := os.ReadFile(LOADAVGFILE)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to read file '%s': %v", LOADAVGFILE, err) fmt.Sprintf("Read(): Failed to read file '%s': %v", LOADAVGFILE, err))
return return
} }
now := time.Now() now := time.Now()
@@ -101,9 +101,9 @@ func (m *LoadavgCollector) Read(interval time.Duration, output chan lp.CCMessage
for i, name := range m.load_matches { for i, name := range m.load_matches {
x, err := strconv.ParseFloat(ls[i], 64) x, err := strconv.ParseFloat(ls[i], 64)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to convert '%s' to float64: %v", ls[i], err) fmt.Sprintf("Read(): Failed to convert '%s' to float64: %v", ls[i], err))
continue continue
} }
if m.load_skips[i] { if m.load_skips[i] {
@@ -120,9 +120,9 @@ func (m *LoadavgCollector) Read(interval time.Duration, output chan lp.CCMessage
for i, name := range m.proc_matches { for i, name := range m.proc_matches {
x, err := strconv.ParseInt(lv[i], 10, 64) x, err := strconv.ParseInt(lv[i], 10, 64)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to convert '%s' to float64: %v", lv[i], err) fmt.Sprintf("Read(): Failed to convert '%s' to float64: %v", lv[i], err))
continue continue
} }
if m.proc_skips[i] { if m.proc_skips[i] {
+1 -1
View File
@@ -84,7 +84,7 @@ func getStats(filename string) map[string]MemstatStats {
case 5: case 5:
v, err := strconv.ParseFloat(linefields[3], 64) v, err := strconv.ParseFloat(linefields[3], 64)
if err == nil { if err == nil {
cclog.ComponentDebug("MemstatCollector", "getStats %s value %v unit %s", strings.Trim(linefields[2], ":"), v, linefields[4]) cclog.ComponentDebug("getStats", strings.Trim(linefields[2], ":"), v, linefields[4])
stats[strings.Trim(linefields[2], ":")] = MemstatStats{ stats[strings.Trim(linefields[2], ":")] = MemstatStats{
value: v, value: v,
unit: linefields[4], unit: linefields[4],
+4 -4
View File
@@ -222,16 +222,16 @@ func (m *NetstatCollector) Read(interval time.Duration, output chan lp.CCMessage
file, err := os.Open(NETSTATFILE) file, err := os.Open(NETSTATFILE)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to open file '%s': %v", NETSTATFILE, err) fmt.Sprintf("Read(): Failed to open file '%s': %v", NETSTATFILE, err))
return return
} }
defer func() { defer func() {
if err := file.Close(); err != nil { if err := file.Close(); err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to close file '%s': %v", NETSTATFILE, err) fmt.Sprintf("Read(): Failed to close file '%s': %v", NETSTATFILE, err))
} }
}() }()
+3 -2
View File
@@ -125,9 +125,10 @@ func (m *nfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
timestamp := time.Now() timestamp := time.Now()
if err := m.updateStats(); err != nil { if err := m.updateStats(); err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): updateStats() failed: %v", err) fmt.Sprintf("Read(): updateStats() failed: %v", err),
)
return return
} }
var prefix string var prefix string
+1 -1
View File
@@ -117,7 +117,7 @@ func (m *NUMAStatsCollector) Init(config json.RawMessage) error {
} }
// Initialized // Initialized
cclog.ComponentDebugf(m.name, "initialized %d NUMA domains", len(m.topology)) cclog.ComponentDebug(m.name, "initialized", len(m.topology), "NUMA domains")
m.init = true m.init = true
return nil return nil
} }
-396
View File
@@ -1,396 +0,0 @@
package collectors
import (
"encoding/json"
"errors"
"fmt"
"slices"
"strconv"
"strings"
"time"
cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
"github.com/NVIDIA/go-nvml/pkg/nvml"
)
type NvidiaGPMMetricDef struct {
name string
outname string
id nvml.GpmMetricId
unit string
}
var NvidiaGPMMetrics []NvidiaGPMMetricDef = []NvidiaGPMMetricDef{
{
name: "GRAPHICS_UTIL",
outname: "nv_gpm_graphics_util",
id: nvml.GPM_METRIC_GRAPHICS_UTIL,
unit: "%",
},
{
name: "SM_UTIL",
outname: "nv_gpm_sm_util",
id: nvml.GPM_METRIC_SM_UTIL,
unit: "%",
},
{
name: "SM_OCCUPANCY",
outname: "nv_gpm_sm_occupancy",
id: nvml.GPM_METRIC_SM_OCCUPANCY,
unit: "%",
},
{
name: "INTEGER_UTIL",
outname: "nv_gpm_integer_util",
id: nvml.GPM_METRIC_INTEGER_UTIL,
unit: "%",
},
{
name: "ANY_TENSOR_UTIL",
outname: "nv_gpm_any_tensor_util",
id: nvml.GPM_METRIC_ANY_TENSOR_UTIL,
unit: "%",
},
{
name: "DFMA_TENSOR_UTIL",
outname: "nv_gpm_dfma_tensor_util",
id: nvml.GPM_METRIC_DFMA_TENSOR_UTIL,
unit: "%",
},
{
name: "HMMA_TENSOR_UTIL",
outname: "nv_gpm_hmma_tensor_util",
id: nvml.GPM_METRIC_HMMA_TENSOR_UTIL,
unit: "%",
},
{
name: "IMMA_TENSOR_UTIL",
outname: "nv_gpm_imma_tensor_util",
id: nvml.GPM_METRIC_IMMA_TENSOR_UTIL,
unit: "%",
},
{
name: "DRAM_BW_UTIL",
outname: "nv_gpm_dram_bw_util",
id: nvml.GPM_METRIC_DRAM_BW_UTIL,
unit: "%",
},
{
name: "FP64_UTIL",
outname: "nv_gpm_fp64_util",
id: nvml.GPM_METRIC_FP64_UTIL,
unit: "%",
},
{
name: "FP32_UTIL",
outname: "nv_gpm_fp32_util",
id: nvml.GPM_METRIC_FP32_UTIL,
unit: "%",
},
{
name: "FP16_UTIL",
outname: "nv_gpm_fp16_util",
id: nvml.GPM_METRIC_FP16_UTIL,
unit: "%",
},
}
type NvidiaGPMCollectorConfig struct {
Metrics []string `json:"metrics,omitempty"`
ExcludeDevices []string `json:"exclude_devices,omitempty"`
AddPciInfoTag bool `json:"add_pci_info_tag,omitempty"`
UsePciInfoAsTypeId bool `json:"use_pci_info_as_type_id,omitempty"`
AddUuidMeta bool `json:"add_uuid_meta,omitempty"`
AddBoardNumberMeta bool `json:"add_board_number_meta,omitempty"`
AddSerialMeta bool `json:"add_serial_meta,omitempty"`
ProcessMigDevices bool `json:"process_mig_devices,omitempty"`
UseUuidForMigDevices bool `json:"use_uuid_for_mig_device,omitempty"`
UseSliceForMigDevices bool `json:"use_slice_for_mig_device,omitempty"`
}
type NvidiaGPMCollectorDevice struct {
device nvml.Device
tags map[string]string
meta map[string]string
startTime time.Time
endTime time.Time
measurement nvml.GpmMetricsGetType
metricsLookup map[int]NvidiaGPMMetricDef
}
type NvidiaGPMCollector struct {
metricCollector
config NvidiaGPMCollectorConfig
gpus []NvidiaGPMCollectorDevice
num_gpus int
}
func (m *NvidiaGPMCollector) Init(config json.RawMessage) error {
var err error = nil
m.name = "NvidiaGPMCollector"
m.parallel = true
if err := m.setup(); err != nil {
return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
}
if len(config) > 0 {
d := json.NewDecoder(strings.NewReader(string(config)))
d.DisallowUnknownFields()
if err = d.Decode(&m.config); err != nil {
return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
}
}
m.meta = map[string]string{
"source": m.name,
"group": "NvidiaGPM",
}
// Initialize NVIDIA Management Library (NVML)
ret := nvml.Init()
// Error: NVML library not found
// (nvml.ErrorString can not be used in this case)
if ret == nvml.ERROR_LIBRARY_NOT_FOUND {
return fmt.Errorf("%s Init(): NVML library not found", m.name)
}
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
return fmt.Errorf("%s Init(): Unable to initialize NVML: %w", m.name, err)
}
// Number of NVIDIA GPUs
num_gpus, ret := nvml.DeviceGetCount()
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
return fmt.Errorf("%s Init(): Unable to get device count: %w", m.name, err)
}
// For all GPUs
m.gpus = make([]NvidiaGPMCollectorDevice, 0, num_gpus)
for i := range num_gpus {
// Skip excluded devices by ID
str_i := strconv.Itoa(i)
if slices.Contains(m.config.ExcludeDevices, str_i) {
cclog.ComponentDebugf(m.name, "Skipping excluded device %s", str_i)
continue
}
// Get device handle
device, ret := nvml.DeviceGetHandleByIndex(i)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to get device at index %d: %s", i, err.Error())
continue
}
supportInfo, ret := nvml.GpmQueryDeviceSupport(device)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to query GPM support for device at index %d: %s", i, err.Error())
continue
} else {
if supportInfo.IsSupportedDevice == uint32(nvml.FEATURE_DISABLED) {
cclog.ComponentErrorf(m.name, "Device at index %d does not support GPM metrics", i)
continue
}
}
stream, ret := nvml.GpmQueryIfStreamingEnabled(device)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to query GPM streaming for device at index %d: %s", i, err.Error())
continue
} else {
if stream == uint32(nvml.FEATURE_DISABLED) {
ret = nvml.GpmSetStreamingEnabled(device, uint32(nvml.FEATURE_ENABLED))
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to set streaming mode for device at index %d: %s", i, err.Error())
}
}
}
// Get device's PCI info
pciInfo, ret := nvml.DeviceGetPciInfo(device)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to get PCI info for device at index %d: %s", i, err.Error())
continue
}
// Create PCI ID in the common format used by the NVML.
pci_id := fmt.Sprintf(
nvml.DEVICE_PCI_BUS_ID_FMT,
pciInfo.Domain,
pciInfo.Bus,
pciInfo.Device)
// Skip excluded devices specified by PCI ID
if slices.Contains(m.config.ExcludeDevices, pci_id) {
cclog.ComponentDebugf(m.name, "Skipping excluded device %s", pci_id)
continue
}
ss, nvmlErr := nvml.GpmSampleAlloc()
if nvmlErr != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Failed to allocate GPM sample for device %d: %s", i, err.Error())
continue
}
es, nvmlErr := nvml.GpmSampleAlloc()
if nvmlErr != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Failed to allocate GPM sample for device %d: %s", i, err.Error())
continue
}
// Select which value to use as 'type-id'.
// The PCI ID is commonly required in SLURM environments because the
// numberic IDs used by SLURM and the ones used by NVML might differ
// depending on the job type. The PCI ID is more reliable but is commonly
// not recorded for a job, so it must be added manually in prologue or epilogue
// e.g. to the comment field
tid := str_i
if m.config.UsePciInfoAsTypeId {
tid = pci_id
}
// Now we got all infos together, populate the device list
g := NvidiaGPMCollectorDevice{}
// Add device handle
g.device = device
// Add tags
g.tags = map[string]string{
"type": "accelerator",
"type-id": tid,
}
// Add PCI info as tag if not already used as 'type-id'
if m.config.AddPciInfoTag && !m.config.UsePciInfoAsTypeId {
g.tags["pci_identifier"] = pci_id
}
g.meta = map[string]string{
"source": m.name,
"group": "Nvidia",
}
if m.config.AddBoardNumberMeta {
board, ret := nvml.DeviceGetBoardPartNumber(device)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentError(m.name, "Unable to get boart part number for device at index", i, ":", err.Error())
} else {
g.meta["board_number"] = board
}
}
if m.config.AddSerialMeta {
serial, ret := nvml.DeviceGetSerial(device)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentError(m.name, "Unable to get serial number for device at index", i, ":", err.Error())
} else {
g.meta["serial"] = serial
}
}
if m.config.AddUuidMeta {
uuid, ret := nvml.DeviceGetUUID(device)
if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret))
cclog.ComponentError(m.name, "Unable to get UUID for device at index", i, ":", err.Error())
} else {
g.meta["uuid"] = uuid
}
}
g.measurement.Sample1 = ss
g.measurement.Sample2 = es
g.measurement.Version = nvml.GPM_METRICS_GET_VERSION
g.metricsLookup = make(map[int]NvidiaGPMMetricDef)
metIdx := 0
for _, inmetric := range m.config.Metrics {
for _, defmetric := range NvidiaGPMMetrics {
if inmetric == defmetric.outname || inmetric == defmetric.name {
g.measurement.Metrics[metIdx] = nvml.GpmMetric{
MetricId: uint32(defmetric.id),
}
g.metricsLookup[metIdx] = defmetric
metIdx += 1
}
}
}
g.measurement.NumMetrics = uint32(metIdx)
m.gpus = append(m.gpus, g)
}
cclog.ComponentDebugf(m.name, "Found %d Nvidia GPUs with GPM support", len(m.gpus))
m.num_gpus = len(m.gpus)
m.init = true
return err
}
func (m *NvidiaGPMCollector) Read(interval time.Duration, output chan lp.CCMessage) {
var err error
if !m.init {
return
}
for i, gpu := range m.gpus {
gpu.startTime = time.Now()
nvmlErr := gpu.measurement.Sample1.Get(gpu.device)
if nvmlErr != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(nvmlErr))
cclog.ComponentError(m.name, "Unable to get start GPM sample for device at index", i, ":", err.Error())
continue
}
}
time.Sleep(interval)
for i, gpu := range m.gpus {
gpu.endTime = time.Now()
nvmlErr := gpu.measurement.Sample2.Get(gpu.device)
if nvmlErr != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(nvmlErr))
cclog.ComponentError(m.name, "Unable to get stop GPM sample for device at index", i, ":", err.Error())
continue
}
}
for i, gpu := range m.gpus {
nvmlErr := nvml.GpmMetricsGet(&gpu.measurement)
if nvmlErr != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(nvmlErr))
cclog.ComponentError(m.name, "Unable to get evaluate GPM sample for device at index", i, ":", err.Error())
continue
}
for idx, metricDef := range gpu.metricsLookup {
y, err := lp.NewMetric(metricDef.outname, gpu.tags, gpu.meta, gpu.measurement.Metrics[idx].Value, time.Now())
if err == nil {
y.AddMeta("unit", metricDef.unit)
output <- y
}
}
}
}
func (m *NvidiaGPMCollector) Close() {
if m.init {
for i, gpu := range m.gpus {
ret := gpu.measurement.Sample1.Free()
if ret != nvml.SUCCESS {
err := errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to free start sample for device at index %d: %s", i, err.Error())
}
ret = gpu.measurement.Sample2.Free()
if ret != nvml.SUCCESS {
err := errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to free stop sample for device at index %d: %s", i, err.Error())
}
}
if ret := nvml.Shutdown(); ret != nvml.SUCCESS {
cclog.ComponentError(m.name, "nvml.Shutdown() not successful")
}
m.init = false
}
}
-54
View File
@@ -1,54 +0,0 @@
<!--
---
title: "Nvidia NVML GPM metric collector"
description: Collect metrics for Nvidia GPUs using the NVML GPM interface
categories: [cc-metric-collector]
tags: ['Admin']
weight: 2
hugo_path: docs/reference/cc-metric-collector/collectors/nvidiaGPM.md
---
-->
## `nvidiaGPM` collector
```json
"nvidia_gpm": {
"metrics": [
"nv_fb_mem_used",
"nv_fan"
],
"exclude_devices": [
"0","1", "0000000:ff:01.0"
],
"process_mig_devices": false,
"use_pci_info_as_type_id": true,
"add_pci_info_tag": false,
"add_uuid_meta": false,
"add_board_number_meta": false,
"add_serial_meta": false,
"use_uuid_for_mig_device": false,
"use_slice_for_mig_device": false
}
```
The `nvidia_gpm` collector can be configured to leave out specific devices with the `exclude_devices` option. It takes IDs as supplied to the NVML with `nvmlDeviceGetHandleByIndex()` or the PCI address in NVML format (`%08X:%02X:%02X.0`). Commonly only the physical GPUs are monitored. If MIG devices should be analyzed as well, set `process_mig_devices` (adds `stype=mig,stype-id=<mig_index>`). With the options `use_uuid_for_mig_device` and `use_slice_for_mig_device`, the `<mig_index>` can be replaced with the UUID (e.g. `MIG-6a9f7cc8-6d5b-5ce0-92de-750edc4d8849`) or the MIG slice name (e.g. `1g.5gb`).
The metrics sent by the `nvidia_gpm` collector use `accelerator` as `type` tag. For the `type-id`, it uses the device handle index by default. With the `use_pci_info_as_type_id` option, the PCI ID is used instead. If both values should be added as tags, activate the `add_pci_info_tag` option. It uses the device handle index as `type-id` and adds the PCI ID as separate `pci_identifier` tag.
Optionally, it is possible to add the UUID, the board part number and the serial to the meta informations. They are not sent to the sinks (if not configured otherwise).
Available Metrics:
* `nv_gpm_graphics_util`
* `nv_gpm_sm_util`
* `nv_gpm_sm_occupancy`
* `nv_gpm_integer_util`
* `nv_gpm_any_tensor_util`
* `nv_gpm_dfma_tensor_util`
* `nv_gpm_hmma_tensor_util`
* `nv_gpm_imma_tensor_util`
* `nv_gpm_dram_bw_util`
* `nv_gpm_fp64_util`
* `nv_gpm_fp32_util`
* `nv_gpm_fp16_util`
+28 -28
View File
@@ -113,7 +113,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
// Skip excluded devices by ID // Skip excluded devices by ID
str_i := strconv.Itoa(i) str_i := strconv.Itoa(i)
if slices.Contains(m.config.ExcludeDevices, str_i) { if slices.Contains(m.config.ExcludeDevices, str_i) {
cclog.ComponentDebugf(m.name, "Skipping excluded device %s", str_i) cclog.ComponentDebug(m.name, "Skipping excluded device", str_i)
continue continue
} }
@@ -121,7 +121,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
device, ret := nvml.DeviceGetHandleByIndex(i) device, ret := nvml.DeviceGetHandleByIndex(i)
if ret != nvml.SUCCESS { if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret)) err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to get device at index %d: %s", i, err.Error()) cclog.ComponentError(m.name, "Unable to get device at index", i, ":", err.Error())
continue continue
} }
@@ -129,7 +129,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
pciInfo, ret := nvml.DeviceGetPciInfo(device) pciInfo, ret := nvml.DeviceGetPciInfo(device)
if ret != nvml.SUCCESS { if ret != nvml.SUCCESS {
err = errors.New(nvml.ErrorString(ret)) err = errors.New(nvml.ErrorString(ret))
cclog.ComponentErrorf(m.name, "Unable to get PCI info for device at index %d: %s", i, err.Error()) cclog.ComponentError(m.name, "Unable to get PCI info for device at index", i, ":", err.Error())
continue continue
} }
// Create PCI ID in the common format used by the NVML. // Create PCI ID in the common format used by the NVML.
@@ -141,7 +141,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
// Skip excluded devices specified by PCI ID // Skip excluded devices specified by PCI ID
if slices.Contains(m.config.ExcludeDevices, pci_id) { if slices.Contains(m.config.ExcludeDevices, pci_id) {
cclog.ComponentDebugf(m.name, "Skipping excluded device %s", pci_id) cclog.ComponentDebug(m.name, "Skipping excluded device", pci_id)
continue continue
} }
@@ -183,7 +183,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
if m.config.AddBoardNumberMeta { if m.config.AddBoardNumberMeta {
board, ret := nvml.DeviceGetBoardPartNumber(device) board, ret := nvml.DeviceGetBoardPartNumber(device)
if ret != nvml.SUCCESS { if ret != nvml.SUCCESS {
cclog.ComponentErrorf(m.name, "Unable to get boart part number for device at index %d: %s", i, err.Error()) cclog.ComponentError(m.name, "Unable to get boart part number for device at index", i, ":", err.Error())
} else { } else {
g.meta["board_number"] = board g.meta["board_number"] = board
} }
@@ -191,7 +191,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
if m.config.AddSerialMeta { if m.config.AddSerialMeta {
serial, ret := nvml.DeviceGetSerial(device) serial, ret := nvml.DeviceGetSerial(device)
if ret != nvml.SUCCESS { if ret != nvml.SUCCESS {
cclog.ComponentErrorf(m.name, "Unable to get serial number for device at index %d: %s", i, err.Error()) cclog.ComponentError(m.name, "Unable to get serial number for device at index", i, ":", err.Error())
} else { } else {
g.meta["serial"] = serial g.meta["serial"] = serial
} }
@@ -199,7 +199,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
if m.config.AddUuidMeta { if m.config.AddUuidMeta {
uuid, ret := nvml.DeviceGetUUID(device) uuid, ret := nvml.DeviceGetUUID(device)
if ret != nvml.SUCCESS { if ret != nvml.SUCCESS {
cclog.ComponentErrorf(m.name, "Unable to get UUID for device at index %d: %s", i, err.Error()) cclog.ComponentError(m.name, "Unable to get UUID for device at index", i, ":", err.Error())
} else { } else {
g.meta["uuid"] = uuid g.meta["uuid"] = uuid
} }
@@ -1128,97 +1128,97 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMessage)
} }
err = readMemoryInfo(device, output) err = readMemoryInfo(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readMemoryInfo for device %s failed", name) cclog.ComponentDebug(m.name, "readMemoryInfo for device", name, "failed")
} }
err = readUtilization(device, output) err = readUtilization(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readUtilization for device %s failed", name) cclog.ComponentDebug(m.name, "readUtilization for device", name, "failed")
} }
err = readTemp(device, output) err = readTemp(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readTemp for device %s failed", name) cclog.ComponentDebug(m.name, "readTemp for device", name, "failed")
} }
err = readFan(device, output) err = readFan(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readFan for device %s failed", name) cclog.ComponentDebug(m.name, "readFan for device", name, "failed")
} }
err = readEccMode(device, output) err = readEccMode(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readEccMode for device %s failed", name) cclog.ComponentDebug(m.name, "readEccMode for device", name, "failed")
} }
err = readPerfState(device, output) err = readPerfState(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readPerfState for device %s failed", name) cclog.ComponentDebug(m.name, "readPerfState for device", name, "failed")
} }
err = readPowerUsage(device, output) err = readPowerUsage(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readPowerUsage for device %s failed", name) cclog.ComponentDebug(m.name, "readPowerUsage for device", name, "failed")
} }
err = readEnergyConsumption(device, output) err = readEnergyConsumption(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readEnergyConsumption for device %s failed", name) cclog.ComponentDebug(m.name, "readEnergyConsumption for device", name, "failed")
} }
err = readClocks(device, output) err = readClocks(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readClocks for device %s failed", name) cclog.ComponentDebug(m.name, "readClocks for device", name, "failed")
} }
err = readMaxClocks(device, output) err = readMaxClocks(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readMaxClocks for device %s failed", name) cclog.ComponentDebug(m.name, "readMaxClocks for device", name, "failed")
} }
err = readEccErrors(device, output) err = readEccErrors(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readEccErrors for device %s failed", name) cclog.ComponentDebug(m.name, "readEccErrors for device", name, "failed")
} }
err = readPowerLimit(device, output) err = readPowerLimit(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readPowerLimit for device %s failed", name) cclog.ComponentDebug(m.name, "readPowerLimit for device", name, "failed")
} }
err = readEncUtilization(device, output) err = readEncUtilization(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readEncUtilization for device %s failed", name) cclog.ComponentDebug(m.name, "readEncUtilization for device", name, "failed")
} }
err = readDecUtilization(device, output) err = readDecUtilization(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readDecUtilization for device %s failed", name) cclog.ComponentDebug(m.name, "readDecUtilization for device", name, "failed")
} }
err = readRemappedRows(device, output) err = readRemappedRows(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readRemappedRows for device %s failed", name) cclog.ComponentDebug(m.name, "readRemappedRows for device", name, "failed")
} }
err = readBarMemoryInfo(device, output) err = readBarMemoryInfo(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readBarMemoryInfo for device %s failed", name) cclog.ComponentDebug(m.name, "readBarMemoryInfo for device", name, "failed")
} }
err = readProcessCounts(device, output) err = readProcessCounts(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readProcessCounts for device %s failed", name) cclog.ComponentDebug(m.name, "readProcessCounts for device", name, "failed")
} }
err = readViolationStats(device, output) err = readViolationStats(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readViolationStats for device %s failed", name) cclog.ComponentDebug(m.name, "readViolationStats for device", name, "failed")
} }
err = readNVLinkStats(device, output) err = readNVLinkStats(device, output)
if err != nil { if err != nil {
cclog.ComponentDebugf(m.name, "readNVLinkStats for device %s failed", name) cclog.ComponentDebug(m.name, "readNVLinkStats for device", name, "failed")
} }
} }
@@ -1244,7 +1244,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMessage)
if maxMig == 0 { if maxMig == 0 {
continue continue
} }
cclog.ComponentDebugf(m.name, "Reading MIG devices for GPU %d", i) cclog.ComponentDebug(m.name, "Reading MIG devices for GPU", i)
for j := range maxMig { for j := range maxMig {
mdev, ret := nvml.DeviceGetMigDeviceHandleByIndex(m.gpus[i].device, j) mdev, ret := nvml.DeviceGetMigDeviceHandleByIndex(m.gpus[i].device, j)
@@ -1268,7 +1268,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMessage)
if m.config.UseUuidForMigDevices { if m.config.UseUuidForMigDevices {
uuid, ret := nvml.DeviceGetUUID(mdev) uuid, ret := nvml.DeviceGetUUID(mdev)
if ret != nvml.SUCCESS { if ret != nvml.SUCCESS {
cclog.ComponentErrorf(m.name, "Unable to get UUID for mig device at index %d: %s", j, err.Error()) cclog.ComponentError(m.name, "Unable to get UUID for mig device at index", j, ":", err.Error())
} else { } else {
migDevice.tags["stype-id"] = uuid migDevice.tags["stype-id"] = uuid
} }
+210 -76
View File
@@ -22,15 +22,24 @@ import (
) )
// running average power limit (RAPL) monitoring attributes for a zone // running average power limit (RAPL) monitoring attributes for a zone
// Only for Intel systems
type RAPLZoneInfo struct { type RAPLZoneInfo struct {
energy int64 // current reading of the energy counter in micro joules
maxEnergyRange int64 // Range of the above energy counter in micro-joules
energyTimestamp time.Time // timestamp when energy counter was read
energyFilepath string // path to a file containing the zones current energy counter in micro joules
shortTermFilepath string // path to short term power limit
longTermFilepath string // path to long term power limit
enabledFilepath string // path to check whether limits are enabled
name string
// tags describing the RAPL zone: // tags describing the RAPL zone:
// * zone_name, subzone_name: e.g. psys, dram, core, uncore, package-0 // * zone_name, subzone_name: e.g. psys, dram, core, uncore, package-0
// * zone_id: e.g. 0:1 (zone 0 sub zone 1) // * zone_id: e.g. 0:1 (zone 0 sub zone 1)
tags map[string]string // type=socket for dram, core, uncore, package-* and type=node for psys
energyFilepath string // path to a file containing the zones current energy counter in micro joules // type-id=socket id
energy int64 // current reading of the energy counter in micro joules tags map[string]string
energyTimestamp time.Time // timestamp when energy counter was read
maxEnergyRange int64 // Range of the above energy counter in micro-joules
} }
type RAPLCollector struct { type RAPLCollector struct {
@@ -42,12 +51,40 @@ type RAPLCollector struct {
// * 0:1 for zone 0 subzone 1 // * 0:1 for zone 0 subzone 1
ExcludeByID []string `json:"exclude_device_by_id,omitempty"` ExcludeByID []string `json:"exclude_device_by_id,omitempty"`
// Exclude names for RAPL zones, e.g. psys, dram, core, uncore, package-0 // Exclude names for RAPL zones, e.g. psys, dram, core, uncore, package-0
ExcludeByName []string `json:"exclude_device_by_name,omitempty"` ExcludeByName []string `json:"exclude_device_by_name,omitempty"`
SkipEnergyReading bool `json:"skip_energy_reading,omitempty"`
SkipLimitsReading bool `json:"skip_limits_reading,omitempty"`
OnlyEnabledLimits bool `json:"only_enabled_limits,omitempty"`
} }
RAPLZoneInfo []RAPLZoneInfo raplZoneInfo []RAPLZoneInfo
meta map[string]string // default meta information meta map[string]string // default meta information
} }
// Get the path to the power limit file for zone selectable by limit name
// Common limit names for Intel systems are
// - long_term
// - short_term
// Does not support AMD as AMD systems do not provide the power limits
// through sysfs
func ZoneLimitFile(folder string, limit_name string) string {
nameGlob := filepath.Join(folder, "constraint_*_name")
candidates, err := filepath.Glob(nameGlob)
if err == nil {
for _, c := range candidates {
if v, err := os.ReadFile(c); err == nil {
if strings.TrimSpace(string(v)) == limit_name {
var i int
n, err := fmt.Sscanf(filepath.Base(c), "constraint_%d_name", &i)
if err == nil && n == 1 {
return filepath.Join(folder, fmt.Sprintf("constraint_%d_power_limit_uw", i))
}
}
}
}
}
return ""
}
// Init initializes the running average power limit (RAPL) collector // Init initializes the running average power limit (RAPL) collector
func (m *RAPLCollector) Init(config json.RawMessage) error { func (m *RAPLCollector) Init(config json.RawMessage) error {
// Check if already initialized // Check if already initialized
@@ -67,6 +104,9 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
} }
// Read in the JSON configuration // Read in the JSON configuration
m.config.SkipEnergyReading = false
m.config.SkipLimitsReading = false
m.config.OnlyEnabledLimits = true
if len(config) > 0 { if len(config) > 0 {
d := json.NewDecoder(bytes.NewReader(config)) d := json.NewDecoder(bytes.NewReader(config))
d.DisallowUnknownFields() d.DisallowUnknownFields()
@@ -93,49 +133,63 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
// See: https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#monitoring-attributes // See: https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#monitoring-attributes
readZoneInfo := func(zonePath string) ( readZoneInfo := func(zonePath string) (
z struct { z struct {
name string // zones name e.g. psys, dram, core, uncore, package-0 name string // zones name e.g. psys, dram, core, uncore, package-0
energyFilepath string // path to a file containing the zones current energy counter in micro joules energyFilepath string // path to a file containing the zones current energy counter in micro joules
energy int64 // current reading of the energy counter in micro joules energy int64 // current reading of the energy counter in micro joules
energyTimestamp time.Time // timestamp when energy counter was read energyTimestamp time.Time // timestamp when energy counter was read
maxEnergyRange int64 // Range of the above energy counter in micro-joules maxEnergyRange int64 // Range of the above energy counter in micro-joules
ok bool // Are all information available? shortTermFilepath string
longTermFilepath string
enabledFilepath string
}, },
) { ) {
// zones name e.g. psys, dram, core, uncore, package-0 // zones name e.g. psys, dram, core, uncore, package-0
foundName := false
if v, err := os.ReadFile( if v, err :=
filepath.Join(zonePath, "name")); err == nil { os.ReadFile(
foundName = true filepath.Join(zonePath, "name")); err == nil {
z.name = strings.TrimSpace(string(v)) z.name = strings.TrimSpace(string(v))
} }
// path to a file containing the zones current energy counter in micro joules if !m.config.SkipEnergyReading {
z.energyFilepath = filepath.Join(zonePath, "energy_uj") // path to a file containing the zones current energy counter in micro joules
z.energyFilepath = filepath.Join(zonePath, "energy_uj")
// current reading of the energy counter in micro joules // current reading of the energy counter in micro joules
foundEnergy := false if v, err := os.ReadFile(z.energyFilepath); err == nil {
if v, err := os.ReadFile(z.energyFilepath); err == nil { if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
// timestamp when energy counter was read z.energy = i
z.energyTimestamp = time.Now() // timestamp when energy counter was read
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil { z.energyTimestamp = time.Now()
foundEnergy = true }
z.energy = i } else {
cclog.ComponentErrorf(m.name, "Cannot read energy file for %s: %v", z.name, err.Error())
} }
// Range of the above energy counter in micro-joules
if v, err :=
os.ReadFile(
filepath.Join(zonePath, "max_energy_range_uj")); err == nil {
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
z.maxEnergyRange = i
}
}
} else {
cclog.ComponentDebugf(m.name, "Energy readings for %s disabled", zonePath)
} }
// Range of the above energy counter in micro-joules if !m.config.SkipLimitsReading {
foundMaxEnergyRange := false z.shortTermFilepath = ZoneLimitFile(zonePath, "short_term")
if v, err := os.ReadFile( if _, err := os.Stat(z.shortTermFilepath); err != nil {
filepath.Join(zonePath, "max_energy_range_uj")); err == nil { z.shortTermFilepath = ""
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
foundMaxEnergyRange = true
z.maxEnergyRange = i
} }
z.longTermFilepath = ZoneLimitFile(zonePath, "long_term")
if _, err := os.Stat(z.longTermFilepath); err != nil {
z.longTermFilepath = ""
}
z.enabledFilepath = filepath.Join(zonePath, "enabled")
} else {
cclog.ComponentDebugf(m.name, "Power limit readings for %s disabled", zonePath)
} }
// Are all information available?
z.ok = foundName && foundEnergy && foundMaxEnergyRange
return return
} }
@@ -152,14 +206,36 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
for _, zonePath := range zonesPath { for _, zonePath := range zonesPath {
zoneID := strings.TrimPrefix(zonePath, zonePrefix) zoneID := strings.TrimPrefix(zonePath, zonePrefix)
z := readZoneInfo(zonePath) z := readZoneInfo(zonePath)
if z.ok && if !isIDExcluded[zoneID] &&
!isIDExcluded[zoneID] &&
!isNameExcluded[z.name] { !isNameExcluded[z.name] {
si := RAPLZoneInfo{
tags: make(map[string]string),
energyFilepath: z.energyFilepath,
energy: z.energy,
energyTimestamp: z.energyTimestamp,
maxEnergyRange: z.maxEnergyRange,
shortTermFilepath: z.shortTermFilepath,
longTermFilepath: z.longTermFilepath,
enabledFilepath: z.enabledFilepath,
name: z.name,
}
si.tags["type"] = "node"
si.tags["type-id"] = "0"
var pid int = 0
if strings.HasPrefix(z.name, "package-") {
n, err := fmt.Sscanf(z.name, "package-%d", &pid)
if err == nil && n == 1 {
si.tags["type-id"] = fmt.Sprintf("%d", pid)
si.tags["type"] = "socket"
}
si.name = "pkg"
}
// Add RAPL monitoring attributes for a zone // Add RAPL monitoring attributes for a zone
m.RAPLZoneInfo = append( m.raplZoneInfo = append(
m.RAPLZoneInfo, m.raplZoneInfo,
RAPLZoneInfo{ RAPLZoneInfo{
tags: map[string]string{ tags: map[string]string{
"id": zoneID, "id": zoneID,
@@ -182,36 +258,40 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
for _, subZonePath := range subZonesPath { for _, subZonePath := range subZonesPath {
subZoneID := strings.TrimPrefix(subZonePath, subZonePrefix) subZoneID := strings.TrimPrefix(subZonePath, subZonePrefix)
sz := readZoneInfo(subZonePath) sz := readZoneInfo(subZonePath)
if len(zoneID) > 0 && len(z.name) > 0 && if len(zoneID) > 0 && len(z.name) > 0 &&
sz.ok &&
!isIDExcluded[zoneID+":"+subZoneID] && !isIDExcluded[zoneID+":"+subZoneID] &&
!isNameExcluded[sz.name] { !isNameExcluded[sz.name] {
m.RAPLZoneInfo = append( m.raplZoneInfo = append(
m.RAPLZoneInfo, m.raplZoneInfo,
RAPLZoneInfo{ RAPLZoneInfo{
tags: map[string]string{ tags: map[string]string{
"id": zoneID + ":" + subZoneID, "id": zoneID + ":" + subZoneID,
"zone_name": z.name, "zone_name": z.name,
"sub_zone_name": sz.name, "sub_zone_name": sz.name,
}, },
energyFilepath: sz.energyFilepath, energyFilepath: sz.energyFilepath,
energy: sz.energy, energy: sz.energy,
energyTimestamp: sz.energyTimestamp, energyTimestamp: sz.energyTimestamp,
maxEnergyRange: sz.maxEnergyRange, maxEnergyRange: sz.maxEnergyRange,
shortTermFilepath: sz.shortTermFilepath,
longTermFilepath: sz.longTermFilepath,
enabledFilepath: sz.enabledFilepath,
name: sz.name,
}) })
} }
} }
} }
if m.RAPLZoneInfo == nil { if m.raplZoneInfo == nil {
return fmt.Errorf("no running average power limit (RAPL) device found in %s", controlTypePath) return fmt.Errorf("no running average power limit (RAPL) device found in %s", controlTypePath)
} }
// Initialized // Initialized
cclog.ComponentDebugf( cclog.ComponentDebug(
m.name, m.name,
"initialized %d zones with running average power limit (RAPL) monitoring attributes", "initialized %d zones with running average power limit (RAPL) monitoring attributes",
len(m.RAPLZoneInfo)) len(m.raplZoneInfo))
m.init = true m.init = true
return err return err
@@ -220,35 +300,89 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
// Read reads running average power limit (RAPL) monitoring attributes for all initialized zones // Read reads running average power limit (RAPL) monitoring attributes for all initialized zones
// See: https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#monitoring-attributes // See: https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#monitoring-attributes
func (m *RAPLCollector) Read(interval time.Duration, output chan lp.CCMessage) { func (m *RAPLCollector) Read(interval time.Duration, output chan lp.CCMessage) {
for i := range m.RAPLZoneInfo { for i := range m.raplZoneInfo {
p := &m.RAPLZoneInfo[i] p := &m.raplZoneInfo[i]
// Read current value of the energy counter in micro joules if !m.config.SkipEnergyReading {
if v, err := os.ReadFile(p.energyFilepath); err == nil { // Read current value of the energy counter in micro joules
energyTimestamp := time.Now() if v, err := os.ReadFile(p.energyFilepath); err == nil {
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil { energyTimestamp := time.Now()
energy := i if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
energy := i
// Compute average power (Δ energy / Δ time) // Compute average power (Δ energy / Δ time)
energyDiff := energy - p.energy energyDiff := energy - p.energy
if energyDiff < 0 { if energyDiff < 0 {
// Handle overflow: // Handle overflow:
// ( p.maxEnergyRange - p.energy ) + energy // ( p.maxEnergyRange - p.energy ) + energy
// = p.maxEnergyRange + ( energy - p.energy ) // = p.maxEnergyRange + ( energy - p.energy )
// = p.maxEnergyRange + diffEnergy // = p.maxEnergyRange + diffEnergy
energyDiff += p.maxEnergyRange energyDiff += p.maxEnergyRange
}
timeDiff := energyTimestamp.Sub(p.energyTimestamp)
averagePower := float64(energyDiff) / float64(timeDiff.Microseconds())
y, err := lp.NewMetric(
fmt.Sprintf("rapl_%s_average_power", p.name),
p.tags,
m.meta,
averagePower,
energyTimestamp)
if err == nil {
output <- y
}
e, err := lp.NewMetric(
fmt.Sprintf("rapl_%s_energy", p.name),
p.tags,
m.meta,
float64(energyDiff)*1e-3,
energyTimestamp)
if err == nil {
e.AddMeta("unit", "Joules")
output <- e
}
// Save current energy counter state
p.energy = energy
p.energyTimestamp = energyTimestamp
} }
timeDiff := energyTimestamp.Sub(p.energyTimestamp) }
averagePower := float64(energyDiff) / float64(timeDiff.Microseconds()) }
// https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#constraints
y, err := lp.NewMetric("rapl_average_power", p.tags, m.meta, averagePower, energyTimestamp) if !m.config.SkipLimitsReading {
if err == nil { skip := false
output <- y if m.config.OnlyEnabledLimits {
if v, err := os.ReadFile(p.enabledFilepath); err == nil {
if strings.TrimSpace(string(v)) == "0" {
skip = true
}
}
}
if !skip {
if len(p.shortTermFilepath) > 0 {
if v, err := os.ReadFile(p.shortTermFilepath); err == nil {
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
name := fmt.Sprintf("rapl_%s_limit_short_term", p.name)
y, err := lp.NewMetric(name, p.tags, m.meta, i/1e6, time.Now())
if err == nil {
output <- y
}
}
}
} }
// Save current energy counter state if len(p.longTermFilepath) > 0 {
p.energy = energy if v, err := os.ReadFile(p.longTermFilepath); err == nil {
p.energyTimestamp = energyTimestamp if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
name := fmt.Sprintf("rapl_%s_limit_long_term", p.name)
y, err := lp.NewMetric(name, p.tags, m.meta, i/1e6, time.Now())
if err == nil {
output <- y
}
}
}
}
} }
} }
} }
+13 -5
View File
@@ -11,16 +11,24 @@ hugo_path: docs/reference/cc-metric-collector/collectors/rapl.md
## `rapl` collector ## `rapl` collector
This collector reads running average power limit (RAPL) monitoring attributes to compute average power consumption metrics. See <https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#monitoring-attributes>. This collector reads running average power limit (RAPL) monitoring attributes to compute average power consumption metrics. See <https://www.kernel.org/doc/html/latest/power/powercap/powercap.html>.
The Likwid metric collector provides similar functionality.
```json ```json
"rapl": { "rapl": {
"exclude_device_by_id": ["0:1", "0:2"], "exclude_device_by_id": ["0:1", "0:2"],
"exclude_device_by_name": ["psys"] "exclude_device_by_name": ["psys"],
"skip_energy_reading": false,
"skip_limits_reading": false,
"only_enabled_limits": true
} }
``` ```
Metrics: Metrics:
* `rapl_average_power`: average power consumption in Watt. The average is computed over the entire runtime from the last measurement to the current measurement * `rapl_<domain>_average_power`: average power consumption in Watt. The average is computed over the entire runtime from the last measurement to the current measurement
* `rapl_<domain>_energy`: Difference from the last measurement
* `rapl_<domain>_limit_short_term`: Short term powercap setting for the domain
* `rapl_<domain>_limit_long_term`: Long term powercap setting for the domain
Only the `rapl_<domain>_average_power` and `rapl_<domain>_energy` metrics require root-permissions. The limits can be read as user. Some domains have limits available but they are not enabled. By default, only enabled domain limits are collected.
Energy and power measurments can also be done with the Likwid metric collector.
+2 -2
View File
@@ -124,7 +124,7 @@ func (m *RocmSmiCollector) Init(config json.RawMessage) error {
if m.config.AddSerialMeta { if m.config.AddSerialMeta {
serial, ret := rocm_smi.DeviceGetSerialNumber(device) serial, ret := rocm_smi.DeviceGetSerialNumber(device)
if ret != rocm_smi.STATUS_SUCCESS { if ret != rocm_smi.STATUS_SUCCESS {
cclog.ComponentErrorf(m.name, "Unable to get serial number for device at index %d: %s", i, rocm_smi.StatusStringNoError(ret)) cclog.ComponentError(m.name, "Unable to get serial number for device at index", i, ":", rocm_smi.StatusStringNoError(ret))
} else { } else {
dev.meta["serial"] = serial dev.meta["serial"] = serial
} }
@@ -152,7 +152,7 @@ func (m *RocmSmiCollector) Read(interval time.Duration, output chan lp.CCMessage
for _, dev := range m.devices { for _, dev := range m.devices {
metrics, ret := rocm_smi.DeviceGetMetrics(dev.device) metrics, ret := rocm_smi.DeviceGetMetrics(dev.device)
if ret != rocm_smi.STATUS_SUCCESS { if ret != rocm_smi.STATUS_SUCCESS {
cclog.ComponentErrorf(m.name, "Unable to get metrics for device at index %d: %s", dev.index, rocm_smi.StatusStringNoError(ret)) cclog.ComponentError(m.name, "Unable to get metrics for device at index", dev.index, ":", rocm_smi.StatusStringNoError(ret))
continue continue
} }
+4 -4
View File
@@ -147,15 +147,15 @@ func (m *SchedstatCollector) Read(interval time.Duration, output chan lp.CCMessa
file, err := os.Open(SCHEDSTATFILE) file, err := os.Open(SCHEDSTATFILE)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to open file '%s': %v", SCHEDSTATFILE, err) fmt.Sprintf("Read(): Failed to open file '%s': %v", SCHEDSTATFILE, err))
} }
defer func() { defer func() {
if err := file.Close(); err != nil { if err := file.Close(); err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to close file '%s': %v", SCHEDSTATFILE, err) fmt.Sprintf("Read(): Failed to close file '%s': %v", SCHEDSTATFILE, err))
} }
}() }()
+2 -2
View File
@@ -240,7 +240,7 @@ func (m *SlurmCgroupCollector) Read(interval time.Duration, output chan lp.CCMes
globPattern := filepath.Join(m.cgroupBase, "job_*") globPattern := filepath.Join(m.cgroupBase, "job_*")
jobDirs, err := filepath.Glob(globPattern) jobDirs, err := filepath.Glob(globPattern)
if err != nil { if err != nil {
cclog.ComponentErrorf(m.name, "Error globbing job directories: %s", err.Error()) cclog.ComponentError(m.name, "Error globbing job directories:", err.Error())
return return
} }
@@ -249,7 +249,7 @@ func (m *SlurmCgroupCollector) Read(interval time.Duration, output chan lp.CCMes
jobdata, err := m.ReadJobData(jKey) jobdata, err := m.ReadJobData(jKey)
if err != nil { if err != nil {
cclog.ComponentError(m.name, "Error reading job data for %s: %s", jKey, err.Error()) cclog.ComponentError(m.name, "Error reading job data for", jKey, ":", err.Error())
continue continue
} }
+2 -2
View File
@@ -228,12 +228,12 @@ func (m *SmartMonCollector) Read(interval time.Duration, output chan lp.CCMessag
stdout, err := command.Output() stdout, err := command.Output()
if err != nil { if err != nil {
cclog.ComponentErrorf(m.name, "cannot read data for device %s", d.Name) cclog.ComponentError(m.name, "cannot read data for device", d.Name)
continue continue
} }
err = json.Unmarshal(stdout, &data) err = json.Unmarshal(stdout, &data)
if err != nil { if err != nil {
cclog.ComponentErrorf(m.name, "cannot unmarshal data for device %s", d.Name) cclog.ComponentError(m.name, "cannot unmarshal data for device", d.Name)
continue continue
} }
if !m.excludeMetric.temp { if !m.excludeMetric.temp {
+4 -4
View File
@@ -188,16 +188,16 @@ func (m *TempCollector) Read(interval time.Duration, output chan lp.CCMessage) {
// Read sensor file // Read sensor file
buffer, err := os.ReadFile(sensor.file) buffer, err := os.ReadFile(sensor.file)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to read file '%s': %v", sensor.file, err) fmt.Sprintf("Read(): Failed to read file '%s': %v", sensor.file, err))
continue continue
} }
x, err := strconv.ParseInt(strings.TrimSpace(string(buffer)), 10, 64) x, err := strconv.ParseInt(strings.TrimSpace(string(buffer)), 10, 64)
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to convert temperature '%s' to int64: %v", buffer, err) fmt.Sprintf("Read(): Failed to convert temperature '%s' to int64: %v", buffer, err))
continue continue
} }
x /= 1000 x /= 1000
+2 -2
View File
@@ -77,9 +77,9 @@ func (m *TopProcsCollector) Read(interval time.Duration, output chan lp.CCMessag
command := exec.Command("ps", "-Ao", "comm", "--sort=-pcpu") command := exec.Command("ps", "-Ao", "comm", "--sort=-pcpu")
stdout, err := command.Output() stdout, err := command.Output()
if err != nil { if err != nil {
cclog.ComponentErrorf( cclog.ComponentError(
m.name, m.name,
"Read(): Failed to read output from command \"%s\": %v", command.String(), err) fmt.Sprintf("Read(): Failed to read output from command \"%s\": %v", command.String(), err))
return return
} }
+13 -13
View File
@@ -35,18 +35,18 @@ type metricRouterTagConfig struct {
// Metric router configuration // Metric router configuration
type metricRouterConfig struct { type metricRouterConfig struct {
HostnameTagName string `json:"hostname_tag,omitempty"` // Key name used when adding the hostname to a metric (default 'hostname') HostnameTagName string `json:"hostname_tag"` // Key name used when adding the hostname to a metric (default 'hostname')
AddTags []metricRouterTagConfig `json:"add_tags,omitempty"` // List of tags that are added when the condition is met AddTags []metricRouterTagConfig `json:"add_tags"` // List of tags that are added when the condition is met
DelTags []metricRouterTagConfig `json:"delete_tags,omitempty"` // List of tags that are removed when the condition is met DelTags []metricRouterTagConfig `json:"delete_tags"` // List of tags that are removed when the condition is met
IntervalAgg []agg.MetricAggregatorIntervalConfig `json:"interval_aggregates,omitempty"` // List of aggregation function processed at the end of an interval IntervalAgg []agg.MetricAggregatorIntervalConfig `json:"interval_aggregates"` // List of aggregation function processed at the end of an interval
DropMetrics []string `json:"drop_metrics,omitempty"` // List of metric names to drop. For fine-grained dropping use drop_metrics_if DropMetrics []string `json:"drop_metrics"` // List of metric names to drop. For fine-grained dropping use drop_metrics_if
DropMetricsIf []string `json:"drop_metrics_if,omitempty"` // List of evaluatable terms to drop metrics DropMetricsIf []string `json:"drop_metrics_if"` // List of evaluatable terms to drop metrics
RenameMetrics map[string]string `json:"rename_metrics,omitempty"` // Map to rename metric name from key to value RenameMetrics map[string]string `json:"rename_metrics"` // Map to rename metric name from key to value
IntervalStamp bool `json:"interval_timestamp,omitempty"` // Update timestamp periodically by ticker each interval? IntervalStamp bool `json:"interval_timestamp"` // Update timestamp periodically by ticker each interval?
NumCacheIntervals int `json:"num_cache_intervals,omitempty"` // Number of intervals of cached metrics for evaluation NumCacheIntervals int `json:"num_cache_intervals"` // Number of intervals of cached metrics for evaluation
MaxForward int `json:"max_forward,omitempty"` // Number of maximal forwarded metrics at one select MaxForward int `json:"max_forward"` // Number of maximal forwarded metrics at one select
NormalizeUnits bool `json:"normalize_units,omitempty"` // Check unit meta flag and normalize it using cc-units NormalizeUnits bool `json:"normalize_units"` // Check unit meta flag and normalize it using cc-units
ChangeUnitPrefix map[string]string `json:"change_unit_prefix,omitempty"` // Add prefix that should be applied to the metrics ChangeUnitPrefix map[string]string `json:"change_unit_prefix"` // Add prefix that should be applied to the metrics
MessageProcessor json.RawMessage `json:"process_messages,omitempty"` MessageProcessor json.RawMessage `json:"process_messages,omitempty"`
} }
@@ -297,7 +297,7 @@ func (r *metricRouter) Start() {
case timestamp := <-timeChan: case timestamp := <-timeChan:
r.timestamp = timestamp r.timestamp = timestamp
cclog.ComponentDebugf("MetricRouter", "Update timestamp %d", r.timestamp.UnixNano()) cclog.ComponentDebug("MetricRouter", "Update timestamp", r.timestamp.UnixNano())
case p := <-r.coll_input: case p := <-r.coll_input:
coll_forward(p) coll_forward(p)