Compare commits

..

1 Commits

Author SHA1 Message Date
Thomas Roehl
0a1c7a9911 Add metric 'nv_util_eff' like nvtop 2026-05-06 18:57:58 +02:00
32 changed files with 275 additions and 728 deletions

View File

@@ -132,11 +132,11 @@ func mainFunc() int {
if len(rcfg.ConfigFile.Interval) > 0 {
t, err := time.ParseDuration(rcfg.ConfigFile.Interval)
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
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
}
}
@@ -145,11 +145,11 @@ func mainFunc() int {
if len(rcfg.ConfigFile.Duration) > 0 {
t, err := time.ParseDuration(rcfg.ConfigFile.Duration)
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
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
}
}

View File

@@ -209,16 +209,16 @@ func (m *BeegfsMetaCollector) Read(interval time.Duration, output chan lp.CCMess
} else {
f1, err := strconv.ParseFloat(m.matches["other"], 32)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}
f2, err := strconv.ParseFloat(split[i], 32)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}
m.matches["beegfs_cstorage_other"] = fmt.Sprintf("%f", f1+f2)

View File

@@ -200,16 +200,16 @@ func (m *BeegfsStorageCollector) Read(interval time.Duration, output chan lp.CCM
} else {
f1, err := strconv.ParseFloat(m.matches["other"], 32)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}
f2, err := strconv.ParseFloat(split[i], 32)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}
m.matches["beegfs_cstorage_other"] = fmt.Sprintf("%f", f1+f2)

View File

@@ -50,7 +50,6 @@ var AvailableCollectors = map[string]MetricCollector{
"nfsiostat": new(NfsIOStatCollector),
"slurm_cgroup": new(SlurmCgroupCollector),
"smartmon": new(SmartMonCollector),
"nvidia_gpm": new(NvidiaGPMCollector),
}
// Metric collector manager data structure
@@ -100,17 +99,17 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
// Initialize configured collectors
for collectorName, collectorCfg := range cm.config {
if _, found := AvailableCollectors[collectorName]; !found {
cclog.ComponentErrorf("CollectorManager", "SKIP unknown collector %s", collectorName)
cclog.ComponentError("CollectorManager", "SKIP unknown collector", collectorName)
continue
}
collector := AvailableCollectors[collectorName]
err := collector.Init(collectorCfg)
if err != nil {
cclog.ComponentErrorf("CollectorManager", "Collector %s initialization failed: %v", collectorName, err)
cclog.ComponentError("CollectorManager", fmt.Sprintf("Collector %s initialization failed: %v", collectorName, err))
continue
}
cclog.ComponentDebugf("CollectorManager", "ADD COLLECTOR %s", collector.Name())
cclog.ComponentDebug("CollectorManager", "ADD COLLECTOR", collector.Name())
if collector.Parallel() {
cm.collectors = append(cm.collectors, collector)
} else {
@@ -156,7 +155,7 @@ func (cm *collectorManager) Start() {
return
default:
// Read metrics from collector c via goroutine
cclog.ComponentDebugf("CollectorManager: Read %s at %v", c.Name(), t)
cclog.ComponentDebug("CollectorManager", c.Name(), t)
cm.collector_wg.Add(1)
go func(myc MetricCollector) {
myc.Read(cm.duration, cm.output)
@@ -174,7 +173,7 @@ func (cm *collectorManager) Start() {
return
default:
// Read metrics from collector c
cclog.ComponentDebugf("CollectorManager: Read %s at %v", c.Name(), t)
cclog.ComponentDebug("CollectorManager", c.Name(), t)
c.Read(cm.duration, cm.output)
}
}

View File

@@ -139,16 +139,16 @@ func (m *CPUFreqCpuInfoCollector) Read(interval time.Duration, output chan lp.CC
const cpuInfoFile = "/proc/cpuinfo"
file, err := os.Open(cpuInfoFile)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to open file '%s': %v", cpuInfoFile, err)
fmt.Sprintf("Read(): Failed to open file '%s': %v", cpuInfoFile, err))
return
}
defer func() {
if err := file.Close(); err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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 {
value, err := strconv.ParseFloat(strings.TrimSpace(lineSplit[1]), 64)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}
if y, err := lp.NewMetric("cpufreq", t.tagSet, m.meta, value, now); err == nil {

View File

@@ -95,7 +95,10 @@ func (m *CPUFreqCollector) Init(config json.RawMessage) error {
}
// 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
return nil
}
@@ -113,14 +116,16 @@ func (m *CPUFreqCollector) Read(interval time.Duration, output chan lp.CCMessage
// Read current frequency
line, err := os.ReadFile(t.scalingCurFreqFile)
if err != nil {
cclog.ComponentErrorf(
m.name, "Read(): Failed to read file '%s': %v", t.scalingCurFreqFile, err)
cclog.ComponentError(
m.name,
fmt.Sprintf("Read(): Failed to read file '%s': %v", t.scalingCurFreqFile, err))
continue
}
cpuFreq, err := strconv.ParseInt(strings.TrimSpace(string(line)), 10, 64)
if err != nil {
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
}

View File

@@ -27,7 +27,6 @@ const CPUSTATFILE = `/proc/stat`
type CpustatCollectorConfig struct {
ExcludeMetrics []string `json:"exclude_metrics,omitempty"`
excludeNumCPUs bool
}
type CpustatCollector struct {
@@ -80,7 +79,6 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
m.matches[match] = index
}
}
m.config.excludeNumCPUs = slices.Contains(m.config.ExcludeMetrics, "num_cpus")
// Check input file
file, err := os.Open(CPUSTATFILE)
@@ -97,13 +95,11 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
line := scanner.Text()
linefields := strings.Fields(line)
if strings.Compare(linefields[0], "cpu") == 0 {
// Kernel system statistics for all CPUs
m.olddata["cpu"] = make(map[string]int64)
for k, v := range m.matches {
m.olddata["cpu"][k], _ = strconv.ParseInt(linefields[v], 0, 64)
}
} else if strings.HasPrefix(linefields[0], "cpu") && strings.Compare(linefields[0], "cpu") != 0 {
// Kernel system statistics per CPU
cpustr := strings.TrimLeft(linefields[0], "cpu")
cpu, _ := strconv.Atoi(cpustr)
m.cputags[linefields[0]] = map[string]string{
@@ -171,15 +167,15 @@ func (m *CpustatCollector) Read(interval time.Duration, output chan lp.CCMessage
file, err := os.Open(CPUSTATFILE)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to open file '%s': %v", CPUSTATFILE, err)
fmt.Sprintf("Read(): Failed to open file '%s': %v", CPUSTATFILE, err))
}
defer func() {
if err := file.Close(); err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err)
fmt.Sprintf("Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err))
}
}()
@@ -195,10 +191,9 @@ func (m *CpustatCollector) Read(interval time.Duration, output chan lp.CCMessage
}
}
if !m.config.excludeNumCPUs {
if num_cpus_metric, err := lp.NewMetric("num_cpus", m.nodetags, m.meta, num_cpus, now); err == nil {
output <- num_cpus_metric
}
num_cpus_metric, err := lp.NewMetric("num_cpus", m.nodetags, m.meta, num_cpus, now)
if err == nil {
output <- num_cpus_metric
}
m.lastTimestamp = now

View File

@@ -64,9 +64,9 @@ func (m *CustomCmdCollector) Init(config json.RawMessage) error {
cmdFields := strings.Fields(c)
command := exec.Command(cmdFields[0], cmdFields[1:]...)
if _, err := command.Output(); err != nil {
cclog.ComponentWarnf(
cclog.ComponentWarn(
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
}
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 {
cclog.ComponentWarn(
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
}
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:]...)
stdout, err := command.Output()
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}
// Read and decode influxDB line-protocol from command output
metrics, err := lp.FromBytes(stdout)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to decode influx Message: %v", err)
fmt.Sprintf("Read(): Failed to decode influx Message: %v", err),
)
continue
}
for _, metric := range metrics {
@@ -126,18 +128,20 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMessa
for _, filename := range m.files {
input, err := os.ReadFile(filename)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}
// Read and decode influxDB line-protocol from file
metrics, err := lp.FromBytes(input)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to decode influx Message: %v", err)
fmt.Sprintf("Read(): Failed to decode influx Message: %v", err),
)
continue
}
for _, metric := range metrics {

View File

@@ -77,16 +77,16 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMessag
file, err := os.Open(MOUNTFILE)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to open file '%s': %v", MOUNTFILE, err)
fmt.Sprintf("Read(): Failed to open file '%s': %v", MOUNTFILE, err))
return
}
defer func() {
if err := file.Close(); err != nil {
cclog.ComponentError(
m.name,
"Read(): Failed to close file '%s': %v", MOUNTFILE, err)
fmt.Sprintf("Read(): Failed to close file '%s': %v", MOUNTFILE, err))
}
}()

View File

@@ -371,7 +371,7 @@ func (m *GpfsCollector) Init(config json.RawMessage) error {
if err != nil {
// 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) {
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
p = m.config.Mmpmon
} else {
@@ -517,23 +517,23 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
// return code
rc, err := strconv.Atoi(key_value["_rc_"])
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
}
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
}
// timestamp
sec, err := strconv.ParseInt(key_value["_t_"], 10, 64)
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
}
msec, err := strconv.ParseInt(key_value["_tu_"], 10, 64)
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
}
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 {
value, err := strconv.ParseInt(key_value[metric.prefix], 10, 64)
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
}
newstate[metric.prefix] = value
@@ -636,7 +636,7 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
}
} else {
// 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))
}
}

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,42 +201,36 @@ 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]
// Read counter file
line, err := os.ReadFile(counterDef.path)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to read from file '%s': %v", counterDef.path, err)
// Current counter can not be saved as last state
counterDef.lastStateAvailable = false
fmt.Sprintf("Read(): Failed to read from file '%s': %v", counterDef.path, err))
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.ComponentErrorf(
cclog.ComponentError(
m.name,
"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
}
}

View File

@@ -145,16 +145,16 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMessage)
file, err := os.Open(IOSTATFILE)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to open file '%s': %v", IOSTATFILE, err)
fmt.Sprintf("Read(): Failed to open file '%s': %v", IOSTATFILE, err))
return
}
defer func() {
if err := file.Close(); err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to close file '%s': %v", IOSTATFILE, err)
fmt.Sprintf("Read(): Failed to close file '%s': %v", IOSTATFILE, err))
}
}()

View File

@@ -12,12 +12,6 @@ package collectors
#cgo LDFLAGS: -Wl,--unresolved-symbols=ignore-in-object-files
#include <stdlib.h>
#include <likwid.h>
int cc_add_hwthread(int cpu_id) {
return HPMaddThread(cpu_id);
}
*/
import "C"
@@ -267,12 +261,12 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
}
for _, metric := range evset.Metrics {
// Try to evaluate the metric
cclog.ComponentDebugf(m.name, "Checking %s", metric.Name)
cclog.ComponentDebug(m.name, "Checking", metric.Name)
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 = ""
} 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 = ""
} else {
globalParams = append(globalParams, metric.Name)
@@ -287,13 +281,13 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
for _, metric := range m.config.Metrics {
// Try to evaluate the global metric
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 = ""
} 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 = ""
} 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 = ""
} else {
totalMetrics++
@@ -334,7 +328,7 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
for _, c := range m.cpulist {
m.measureThread.Call(
func() {
retCode := C.cc_add_hwthread(C.int(c))
retCode := C.HPMaddThread(C.uint32_t(c))
if retCode != 0 {
err := fmt.Errorf("C.HPMaddThread(%v) failed with return code %v", c, retCode)
cclog.ComponentError(m.name, err.Error())
@@ -381,16 +375,16 @@ func (m *LikwidCollector) takeMeasurement(evidx int, evset LikwidEventsetConfig,
// Watch changes for the lock file ()
watcher, err := fsnotify.NewWatcher()
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}
defer func() {
if err := watcher.Close(); err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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 {
@@ -603,7 +597,7 @@ func (m *LikwidCollector) calcEventsetMetrics(evset LikwidEventsetConfig, interv
if tid >= 0 && len(metric.Calc) > 0 {
value, err := agg.EvalFloat64Condition(metric.Calc, evset.results[tid])
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
}
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
value, err := agg.EvalFloat64Condition(metric.Calc, params)
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
}
if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) {

View File

@@ -89,9 +89,9 @@ func (m *LoadavgCollector) Read(interval time.Duration, output chan lp.CCMessage
}
buffer, err := os.ReadFile(LOADAVGFILE)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to read file '%s': %v", LOADAVGFILE, err)
fmt.Sprintf("Read(): Failed to read file '%s': %v", LOADAVGFILE, err))
return
}
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 {
x, err := strconv.ParseFloat(ls[i], 64)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}
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 {
x, err := strconv.ParseInt(lv[i], 10, 64)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}
if m.proc_skips[i] {

View File

@@ -72,8 +72,7 @@ func getStats(filename string) map[string]MemstatStats {
for scanner.Scan() {
line := scanner.Text()
linefields := strings.Fields(line)
switch len(linefields) {
case 3:
if len(linefields) == 3 {
v, err := strconv.ParseFloat(linefields[1], 64)
if err == nil {
stats[strings.Trim(linefields[0], ":")] = MemstatStats{
@@ -81,10 +80,10 @@ func getStats(filename string) map[string]MemstatStats {
unit: linefields[2],
}
}
case 5:
} else if len(linefields) == 5 {
v, err := strconv.ParseFloat(linefields[3], 64)
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{
value: v,
unit: linefields[4],
@@ -107,10 +106,7 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
}
}
m.meta = map[string]string{
"source": m.name,
"group": "Memory",
}
m.meta = map[string]string{"source": m.name, "group": "Memory"}
m.stats = make(map[string]int64)
m.matches = make(map[string]string)
m.tags = map[string]string{"type": "node"}
@@ -149,7 +145,7 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
"KernelStack": "mem_kernelstack",
}
for k, v := range matches {
if !slices.Contains(m.config.ExcludeMetrics, v) {
if !slices.Contains(m.config.ExcludeMetrics, k) {
m.matches[k] = v
}
}
@@ -157,7 +153,7 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
if !slices.Contains(m.config.ExcludeMetrics, "mem_used") {
m.sendMemUsed = true
}
if len(m.matches) == 0 && !m.sendMemUsed {
if len(m.matches) == 0 {
return fmt.Errorf("%s Init(): no metrics to collect", m.name)
}
if err := m.setup(); err != nil {

View File

@@ -222,16 +222,16 @@ func (m *NetstatCollector) Read(interval time.Duration, output chan lp.CCMessage
file, err := os.Open(NETSTATFILE)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to open file '%s': %v", NETSTATFILE, err)
fmt.Sprintf("Read(): Failed to open file '%s': %v", NETSTATFILE, err))
return
}
defer func() {
if err := file.Close(); err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to close file '%s': %v", NETSTATFILE, err)
fmt.Sprintf("Read(): Failed to close file '%s': %v", NETSTATFILE, err))
}
}()

View File

@@ -125,9 +125,10 @@ func (m *nfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
timestamp := time.Now()
if err := m.updateStats(); err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): updateStats() failed: %v", err)
fmt.Sprintf("Read(): updateStats() failed: %v", err),
)
return
}
var prefix string

View File

@@ -117,7 +117,7 @@ func (m *NUMAStatsCollector) Init(config json.RawMessage) error {
}
// 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
return nil
}

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
}
}

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`

View File

@@ -113,7 +113,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
// 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)
cclog.ComponentDebug(m.name, "Skipping excluded device", str_i)
continue
}
@@ -121,7 +121,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
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())
cclog.ComponentError(m.name, "Unable to get device at index", i, ":", err.Error())
continue
}
@@ -129,7 +129,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
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())
cclog.ComponentError(m.name, "Unable to get PCI info for device at index", i, ":", err.Error())
continue
}
// 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
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
}
@@ -183,7 +183,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
if m.config.AddBoardNumberMeta {
board, ret := nvml.DeviceGetBoardPartNumber(device)
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 {
g.meta["board_number"] = board
}
@@ -191,7 +191,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
if m.config.AddSerialMeta {
serial, ret := nvml.DeviceGetSerial(device)
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 {
g.meta["serial"] = serial
}
@@ -199,7 +199,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
if m.config.AddUuidMeta {
uuid, ret := nvml.DeviceGetUUID(device)
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 {
g.meta["uuid"] = uuid
}
@@ -1115,6 +1115,31 @@ func readNVLinkStats(device *NvidiaCollectorDevice, output chan lp.CCMessage) er
return nil
}
func readEfficiency(device *NvidiaCollectorDevice, output chan lp.CCMessage) error {
if !device.excludeMetrics["nv_util_eff"] {
maxPower, ret := nvml.DeviceGetEnforcedPowerLimit(device.device)
if ret == nvml.SUCCESS {
curPower, ret := nvml.DeviceGetPowerUsage(device.device)
if ret == nvml.SUCCESS {
util, ret := nvml.DeviceGetUtilizationRates(device.device)
if ret == nvml.SUCCESS {
factor := float64(curPower) / float64(maxPower)
eff := uint32(float64(util.Gpu) * factor)
if eff > 100 {
eff = 100
}
y, err := lp.NewMetric("nv_util_eff", device.tags, device.meta, eff, time.Now())
if err == nil {
y.AddTag("unit", "percent")
output <- y
}
}
}
}
}
return nil
}
func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMessage) {
var err error
if !m.init {
@@ -1128,97 +1153,102 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMessage)
}
err = readMemoryInfo(device, output)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
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)
if err != nil {
cclog.ComponentDebugf(m.name, "readNVLinkStats for device %s failed", name)
cclog.ComponentDebug(m.name, "readNVLinkStats for device", name, "failed")
}
err = readEfficiency(device, output)
if err != nil {
cclog.ComponentDebug(m.name, "readEfficiency for device", name, "failed")
}
}
@@ -1244,7 +1274,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMessage)
if maxMig == 0 {
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 {
mdev, ret := nvml.DeviceGetMigDeviceHandleByIndex(m.gpus[i].device, j)
@@ -1268,7 +1298,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMessage)
if m.config.UseUuidForMigDevices {
uuid, ret := nvml.DeviceGetUUID(mdev)
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 {
migDevice.tags["stype-id"] = uuid
}

View File

@@ -85,5 +85,6 @@ Metrics:
* `nv_energy`
* `nv_energy_abs`
* `nv_average_power`
* `nv_util_eff` (`nv_util` * (`nv_power_usage` / `nv_power_max_limit`))
Some metrics add the additional sub type tag (`stype`) like the `nv_nvlink_*` metrics set `stype=nvlink,stype-id=<link_number>`.

View File

@@ -208,10 +208,11 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
}
// Initialized
cclog.ComponentDebugf(
cclog.ComponentDebug(
m.name,
"initialized %d zones with running average power limit (RAPL) monitoring attributes",
len(m.RAPLZoneInfo))
"initialized",
len(m.RAPLZoneInfo),
"zones with running average power limit (RAPL) monitoring attributes")
m.init = true
return err

View File

@@ -124,7 +124,7 @@ func (m *RocmSmiCollector) Init(config json.RawMessage) error {
if m.config.AddSerialMeta {
serial, ret := rocm_smi.DeviceGetSerialNumber(device)
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 {
dev.meta["serial"] = serial
}
@@ -152,7 +152,7 @@ func (m *RocmSmiCollector) Read(interval time.Duration, output chan lp.CCMessage
for _, dev := range m.devices {
metrics, ret := rocm_smi.DeviceGetMetrics(dev.device)
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
}

View File

@@ -147,15 +147,15 @@ func (m *SchedstatCollector) Read(interval time.Duration, output chan lp.CCMessa
file, err := os.Open(SCHEDSTATFILE)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to open file '%s': %v", SCHEDSTATFILE, err)
fmt.Sprintf("Read(): Failed to open file '%s': %v", SCHEDSTATFILE, err))
}
defer func() {
if err := file.Close(); err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
m.name,
"Read(): Failed to close file '%s': %v", SCHEDSTATFILE, err)
fmt.Sprintf("Read(): Failed to close file '%s': %v", SCHEDSTATFILE, err))
}
}()

View File

@@ -240,7 +240,7 @@ func (m *SlurmCgroupCollector) Read(interval time.Duration, output chan lp.CCMes
globPattern := filepath.Join(m.cgroupBase, "job_*")
jobDirs, err := filepath.Glob(globPattern)
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
}
@@ -249,7 +249,7 @@ func (m *SlurmCgroupCollector) Read(interval time.Duration, output chan lp.CCMes
jobdata, err := m.ReadJobData(jKey)
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
}

View File

@@ -228,12 +228,12 @@ func (m *SmartMonCollector) Read(interval time.Duration, output chan lp.CCMessag
stdout, err := command.Output()
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
}
err = json.Unmarshal(stdout, &data)
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
}
if !m.excludeMetric.temp {

View File

@@ -188,16 +188,16 @@ func (m *TempCollector) Read(interval time.Duration, output chan lp.CCMessage) {
// Read sensor file
buffer, err := os.ReadFile(sensor.file)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}
x, err := strconv.ParseInt(strings.TrimSpace(string(buffer)), 10, 64)
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}
x /= 1000

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")
stdout, err := command.Output()
if err != nil {
cclog.ComponentErrorf(
cclog.ComponentError(
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
}

10
go.mod
View File

@@ -7,10 +7,10 @@ require (
github.com/ClusterCockpit/go-rocm-smi v0.4.0
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/thread v0.3.2
golang.org/x/sys v0.45.0
github.com/fsnotify/fsnotify v1.10.0
github.com/tklauser/go-sysconf v0.3.16
golang.design/x/thread v0.0.0-20210122121316-335e9adffdf1
golang.org/x/sys v0.43.0
)
require (
@@ -38,7 +38,7 @@ require (
github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 // indirect
github.com/shopspring/decimal v1.4.0 // indirect
github.com/stmcginnis/gofish v0.21.6 // indirect
github.com/tklauser/numcpus v0.12.0 // indirect
github.com/tklauser/numcpus v0.11.0 // indirect
go.yaml.in/yaml/v2 v2.4.4 // indirect
golang.org/x/crypto v0.50.0 // indirect
golang.org/x/net v0.53.0 // indirect

21
go.sum
View File

@@ -53,8 +53,8 @@ github.com/expr-lang/expr v1.17.8 h1:W1loDTT+0PQf5YteHSTpju2qfUfNoBt4yw9+wOEU9VM
github.com/expr-lang/expr v1.17.8/go.mod h1:8/vRC7+7HBzESEqt5kKpYXxrxkr31SaO8r40VO/1IT4=
github.com/frankban/quicktest v1.13.0 h1:yNZif1OkDfNoDfb9zZa9aXIpejNR4F23Wely0c+Qdqk=
github.com/frankban/quicktest v1.13.0/go.mod h1:qLE0fzW0VuyUAJgPU19zByoIr0HtCHN/r/VLSOOIySU=
github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx59Ho=
github.com/fsnotify/fsnotify v1.10.1/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo=
github.com/fsnotify/fsnotify v1.10.0 h1:Xx/5Ydg9CeBDX/wi4VJqStNtohYjitZhhlHt4h3St1M=
github.com/fsnotify/fsnotify v1.10.0/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo=
github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
@@ -163,18 +163,18 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
github.com/testcontainers/testcontainers-go v0.26.0 h1:uqcYdoOHBy1ca7gKODfBd9uTHVK3a7UL848z09MVZ0c=
github.com/testcontainers/testcontainers-go v0.26.0/go.mod h1:ICriE9bLX5CLxL9OFQ2N+2N+f+803LNJ1utJb1+Inx0=
github.com/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRUgPQU=
github.com/tklauser/go-sysconf v0.4.0/go.mod h1:8mTNWyog7H+MpKijp4VmKJAd2bbYQ2zuUwkYRbUArPI=
github.com/tklauser/numcpus v0.12.0 h1:NR85qdvHA9pFse3x3weVZ0r0ST8R6l5RHbZrlRaqob4=
github.com/tklauser/numcpus v0.12.0/go.mod h1:ABHeXzJnr/qqwguhClkZKT1/8VABcYrsyUiUGobwWJg=
github.com/tklauser/go-sysconf v0.3.16 h1:frioLaCQSsF5Cy1jgRBrzr6t502KIIwQ0MArYICU0nA=
github.com/tklauser/go-sysconf v0.3.16/go.mod h1:/qNL9xxDhc7tx3HSRsLWNnuzbVfh3e7gh/BmM179nYI=
github.com/tklauser/numcpus v0.11.0 h1:nSTwhKH5e1dMNsCdVBukSZrURJRoHbSEQjdEbY+9RXw=
github.com/tklauser/numcpus v0.11.0/go.mod h1:z+LwcLq54uWZTX0u/bGobaV34u6V7KNlTZejzM6/3MQ=
github.com/yusufpapurcu/wmi v1.2.3 h1:E1ctvB7uKFMOJw3fdOW32DwGE9I7t++CRUEMKvFoFiw=
github.com/yusufpapurcu/wmi v1.2.3/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
golang.design/x/thread v0.3.2 h1:FmD1glspGrQCe6FuQLmSrT6wz2CSzq7vKVDluyiMnqo=
golang.design/x/thread v0.3.2/go.mod h1:6+Hi2rMOgMHZdKDWaqNHyWtoFUx1HxZ06LfHPh5Z/hQ=
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=
@@ -183,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.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/sys v0.0.0-20210122093101-04d7465088b8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
golang.org/x/sys v0.43.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=

View File

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