mirror of
https://github.com/ClusterCockpit/cc-metric-collector.git
synced 2026-06-09 21:37:29 +02:00
Compare commits
12 Commits
nvidiaColl
...
cclog_upda
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c9dfca622f | ||
|
|
3b0638e815 | ||
|
|
037b4f1526 | ||
|
|
bed5491068 | ||
|
|
a2eba41150 | ||
|
|
5d55ee7a77 | ||
|
|
5938368a76 | ||
|
|
077204d39f | ||
|
|
dcc9746df4 | ||
|
|
2c51a3ed72 | ||
|
|
656ea73d12 | ||
|
|
330f923596 |
@@ -132,11 +132,11 @@ func mainFunc() int {
|
||||
if len(rcfg.ConfigFile.Interval) > 0 {
|
||||
t, err := time.ParseDuration(rcfg.ConfigFile.Interval)
|
||||
if err != nil {
|
||||
cclog.Error("Configuration value 'interval' no valid duration")
|
||||
cclog.Errorf("Configuration value interval=%s no valid duration", rcfg.ConfigFile.Interval)
|
||||
}
|
||||
rcfg.Interval = t
|
||||
if rcfg.Interval == 0 {
|
||||
cclog.Error("Configuration value 'interval' must be greater than zero")
|
||||
cclog.Errorf("Configuration value interval=%s must be greater than zero", rcfg.ConfigFile.Interval)
|
||||
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' no valid duration")
|
||||
cclog.Error("Configuration value duration=%s no valid duration", rcfg.ConfigFile.Duration)
|
||||
}
|
||||
rcfg.Duration = t
|
||||
if rcfg.Duration == 0 {
|
||||
cclog.Error("Configuration value 'duration' must be greater than zero")
|
||||
cclog.Error("Configuration value duration=%s must be greater than zero", rcfg.ConfigFile.Duration)
|
||||
return 1
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err))
|
||||
"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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err))
|
||||
"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)
|
||||
|
||||
@@ -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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err))
|
||||
"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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Metric (other): Failed to convert str written '%s' to float: %v", m.matches["other"], err))
|
||||
"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)
|
||||
|
||||
@@ -50,6 +50,7 @@ var AvailableCollectors = map[string]MetricCollector{
|
||||
"nfsiostat": new(NfsIOStatCollector),
|
||||
"slurm_cgroup": new(SlurmCgroupCollector),
|
||||
"smartmon": new(SmartMonCollector),
|
||||
"nvidia_gpm": new(NvidiaGPMCollector),
|
||||
}
|
||||
|
||||
// Metric collector manager data structure
|
||||
@@ -99,17 +100,17 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
|
||||
// Initialize configured collectors
|
||||
for collectorName, collectorCfg := range cm.config {
|
||||
if _, found := AvailableCollectors[collectorName]; !found {
|
||||
cclog.ComponentError("CollectorManager", "SKIP unknown collector", collectorName)
|
||||
cclog.ComponentErrorf("CollectorManager", "SKIP unknown collector %s", collectorName)
|
||||
continue
|
||||
}
|
||||
collector := AvailableCollectors[collectorName]
|
||||
|
||||
err := collector.Init(collectorCfg)
|
||||
if err != nil {
|
||||
cclog.ComponentError("CollectorManager", fmt.Sprintf("Collector %s initialization failed: %v", collectorName, err))
|
||||
cclog.ComponentErrorf("CollectorManager", "Collector %s initialization failed: %v", collectorName, err)
|
||||
continue
|
||||
}
|
||||
cclog.ComponentDebug("CollectorManager", "ADD COLLECTOR", collector.Name())
|
||||
cclog.ComponentDebugf("CollectorManager", "ADD COLLECTOR %s", collector.Name())
|
||||
if collector.Parallel() {
|
||||
cm.collectors = append(cm.collectors, collector)
|
||||
} else {
|
||||
@@ -155,7 +156,7 @@ func (cm *collectorManager) Start() {
|
||||
return
|
||||
default:
|
||||
// Read metrics from collector c via goroutine
|
||||
cclog.ComponentDebug("CollectorManager", c.Name(), t)
|
||||
cclog.ComponentDebugf("CollectorManager: Read %s at %v", c.Name(), t)
|
||||
cm.collector_wg.Add(1)
|
||||
go func(myc MetricCollector) {
|
||||
myc.Read(cm.duration, cm.output)
|
||||
@@ -173,7 +174,7 @@ func (cm *collectorManager) Start() {
|
||||
return
|
||||
default:
|
||||
// Read metrics from collector c
|
||||
cclog.ComponentDebug("CollectorManager", c.Name(), t)
|
||||
cclog.ComponentDebugf("CollectorManager: Read %s at %v", c.Name(), t)
|
||||
c.Read(cm.duration, cm.output)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to open file '%s': %v", cpuInfoFile, err))
|
||||
"Read(): Failed to open file '%s': %v", cpuInfoFile, err)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
if err := file.Close(); err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to close file '%s': %v", cpuInfoFile, err))
|
||||
"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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to convert cpu MHz '%s' to float64: %v", lineSplit[1], err))
|
||||
"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 {
|
||||
|
||||
@@ -95,10 +95,7 @@ func (m *CPUFreqCollector) Init(config json.RawMessage) error {
|
||||
}
|
||||
|
||||
// Initialized
|
||||
cclog.ComponentDebug(
|
||||
m.name,
|
||||
"initialized",
|
||||
len(m.topology), "non-hyper-threading CPUs")
|
||||
cclog.ComponentDebugf(m.name, "initialized %d non-hyper-threading CPUs")
|
||||
m.init = true
|
||||
return nil
|
||||
}
|
||||
@@ -116,16 +113,14 @@ func (m *CPUFreqCollector) Read(interval time.Duration, output chan lp.CCMessage
|
||||
// Read current frequency
|
||||
line, err := os.ReadFile(t.scalingCurFreqFile)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to read file '%s': %v", t.scalingCurFreqFile, err))
|
||||
cclog.ComponentErrorf(
|
||||
m.name, "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,
|
||||
fmt.Sprintf("Read(): Failed to convert CPU frequency '%s' to int64: %v", line, err))
|
||||
m.name, "Read(): Failed to convert CPU frequency '%s' to int64: %v", line, err)
|
||||
continue
|
||||
}
|
||||
|
||||
|
||||
@@ -27,6 +27,7 @@ const CPUSTATFILE = `/proc/stat`
|
||||
|
||||
type CpustatCollectorConfig struct {
|
||||
ExcludeMetrics []string `json:"exclude_metrics,omitempty"`
|
||||
excludeNumCPUs bool
|
||||
}
|
||||
|
||||
type CpustatCollector struct {
|
||||
@@ -79,6 +80,7 @@ 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)
|
||||
@@ -95,11 +97,13 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
|
||||
line := scanner.Text()
|
||||
linefields := strings.Fields(line)
|
||||
if strings.Compare(linefields[0], "cpu") == 0 {
|
||||
// Kernel system statistics for all CPUs
|
||||
m.olddata["cpu"] = make(map[string]int64)
|
||||
for k, v := range m.matches {
|
||||
m.olddata["cpu"][k], _ = strconv.ParseInt(linefields[v], 0, 64)
|
||||
}
|
||||
} else if strings.HasPrefix(linefields[0], "cpu") && strings.Compare(linefields[0], "cpu") != 0 {
|
||||
// Kernel system statistics per CPU
|
||||
cpustr := strings.TrimLeft(linefields[0], "cpu")
|
||||
cpu, _ := strconv.Atoi(cpustr)
|
||||
m.cputags[linefields[0]] = map[string]string{
|
||||
@@ -167,15 +171,15 @@ func (m *CpustatCollector) Read(interval time.Duration, output chan lp.CCMessage
|
||||
|
||||
file, err := os.Open(CPUSTATFILE)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to open file '%s': %v", CPUSTATFILE, err))
|
||||
"Read(): Failed to open file '%s': %v", CPUSTATFILE, err)
|
||||
}
|
||||
defer func() {
|
||||
if err := file.Close(); err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err))
|
||||
"Read(): Failed to close file '%s': %v", string(CPUSTATFILE), err)
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -191,9 +195,10 @@ func (m *CpustatCollector) Read(interval time.Duration, output chan lp.CCMessage
|
||||
}
|
||||
}
|
||||
|
||||
num_cpus_metric, err := lp.NewMetric("num_cpus", m.nodetags, m.meta, num_cpus, now)
|
||||
if err == nil {
|
||||
output <- num_cpus_metric
|
||||
if !m.config.excludeNumCPUs {
|
||||
if num_cpus_metric, err := lp.NewMetric("num_cpus", m.nodetags, m.meta, num_cpus, now); err == nil {
|
||||
output <- num_cpus_metric
|
||||
}
|
||||
}
|
||||
|
||||
m.lastTimestamp = now
|
||||
|
||||
@@ -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.ComponentWarn(
|
||||
cclog.ComponentWarnf(
|
||||
m.name,
|
||||
fmt.Sprintf("%s Init(): Execution of command \"%s\" failed: %v", m.name, command.String(), err))
|
||||
"%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,
|
||||
fmt.Sprintf("%s Init(): Reading of file \"%s\" failed: %v", m.name, fileName, err))
|
||||
"%s Init(): Reading of file \"%s\" failed: %v", m.name, fileName, err)
|
||||
continue
|
||||
}
|
||||
m.files = append(m.files, fileName)
|
||||
@@ -100,20 +100,18 @@ 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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to read command output for command \"%s\": %v", command.String(), err),
|
||||
)
|
||||
"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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to decode influx Message: %v", err),
|
||||
)
|
||||
"Read(): Failed to decode influx Message: %v", err)
|
||||
continue
|
||||
}
|
||||
for _, metric := range metrics {
|
||||
@@ -128,20 +126,18 @@ 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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to read file \"%s\": %v\n", filename, err),
|
||||
)
|
||||
"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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to decode influx Message: %v", err),
|
||||
)
|
||||
"Read(): Failed to decode influx Message: %v", err)
|
||||
continue
|
||||
}
|
||||
for _, metric := range metrics {
|
||||
|
||||
@@ -77,16 +77,16 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMessag
|
||||
|
||||
file, err := os.Open(MOUNTFILE)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to open file '%s': %v", MOUNTFILE, err))
|
||||
"Read(): Failed to open file '%s': %v", MOUNTFILE, err)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
if err := file.Close(); err != nil {
|
||||
cclog.ComponentError(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to close file '%s': %v", MOUNTFILE, err))
|
||||
"Read(): Failed to close file '%s': %v", MOUNTFILE, err)
|
||||
}
|
||||
}()
|
||||
|
||||
|
||||
@@ -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.ComponentWarn(m.name, fmt.Sprintf("got error looking for mmpmon binary '%s': %v . This is expected when using sudo, continuing.", m.config.Mmpmon, err))
|
||||
cclog.ComponentWarnf(m.name, "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.ComponentError(m.name, fmt.Sprintf("Read(): Failed to convert return code '%s' to int: %v", key_value["_rc_"], err))
|
||||
cclog.ComponentErrorf(m.name, "Read(): Failed to convert return code '%s' to int: %v", key_value["_rc_"], err)
|
||||
continue
|
||||
}
|
||||
if rc != 0 {
|
||||
cclog.ComponentError(m.name, fmt.Sprintf("Read(): Filesystem '%s' is not ok.", filesystem))
|
||||
cclog.ComponentErrorf(m.name, "Read(): Filesystem '%s' is not ok.", filesystem)
|
||||
continue
|
||||
}
|
||||
|
||||
// timestamp
|
||||
sec, err := strconv.ParseInt(key_value["_t_"], 10, 64)
|
||||
if err != nil {
|
||||
cclog.ComponentError(m.name, fmt.Sprintf("Read(): Failed to convert seconds '%s' to int64: %v", key_value["_t_"], err))
|
||||
cclog.ComponentErrorf(m.name, "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.ComponentError(m.name, fmt.Sprintf("Read(): Failed to convert micro seconds '%s' to int64: %v", key_value["_tu_"], err))
|
||||
cclog.ComponentErrorf(m.name, "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.ComponentError(m.name, fmt.Sprintf("Read(): Failed to convert %s '%s' to int64: %v", metric.desc, key_value[metric.prefix], err))
|
||||
cclog.ComponentErrorf(m.name, "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.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))
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -23,20 +23,29 @@ import (
|
||||
"golang.org/x/sys/unix"
|
||||
)
|
||||
|
||||
const IB_BASEPATH = "/sys/class/infiniband/"
|
||||
// See: https://www.kernel.org/doc/Documentation/ABI/stable/sysfs-class-infiniband
|
||||
const (
|
||||
ibBasePath = "/sys/class/infiniband/"
|
||||
ibDataUnit = "bytes"
|
||||
ibDataRateUnit = ibDataUnit + "/sec"
|
||||
ibPkgUnit = "packets"
|
||||
ibPkgRateUnit = ibPkgUnit + "/sec"
|
||||
)
|
||||
|
||||
type InfinibandCollectorMetric struct {
|
||||
name string
|
||||
path string
|
||||
unit string
|
||||
scale int64
|
||||
addToIBTotal bool
|
||||
addToIBTotalPkgs bool
|
||||
lastState int64
|
||||
name string
|
||||
path string
|
||||
unit string
|
||||
unitRates string
|
||||
scaleByFourLanes bool
|
||||
addToIBTotal bool
|
||||
addToIBTotalPkgs bool
|
||||
lastState uint64
|
||||
lastStateAvailable bool
|
||||
}
|
||||
|
||||
type InfinibandCollectorInfo struct {
|
||||
LID string // IB local Identifier (LID)
|
||||
lid string // IB local Identifier (LID)
|
||||
device string // IB device
|
||||
port string // IB device port
|
||||
portCounterFiles []InfinibandCollectorMetric // mapping counter name -> InfinibandCollectorMetric
|
||||
@@ -56,7 +65,7 @@ type InfinibandCollector struct {
|
||||
lastTimestamp time.Time // Store time stamp of last tick to derive bandwidths
|
||||
}
|
||||
|
||||
// Init initializes the Infiniband collector by walking through files below IB_BASEPATH
|
||||
// Init initializes the Infiniband collector by walking through files below ibBasePath
|
||||
func (m *InfinibandCollector) Init(config json.RawMessage) error {
|
||||
// Check if already initialized
|
||||
if m.init {
|
||||
@@ -87,7 +96,7 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
|
||||
}
|
||||
|
||||
// Loop for all InfiniBand directories
|
||||
globPattern := filepath.Join(IB_BASEPATH, "*", "ports", "*")
|
||||
globPattern := filepath.Join(ibBasePath, "*", "ports", "*")
|
||||
ibDirs, err := filepath.Glob(globPattern)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s Init(): unable to glob files with pattern %s: %w", m.name, globPattern, err)
|
||||
@@ -122,36 +131,42 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
|
||||
countersDir := filepath.Join(path, "counters")
|
||||
portCounterFiles := []InfinibandCollectorMetric{
|
||||
{
|
||||
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), 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_xmit",
|
||||
path: filepath.Join(countersDir, "port_xmit_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,
|
||||
},
|
||||
{
|
||||
// Total number of packets received on all VLs from this port (this may include packets containing Errors.
|
||||
// This is 64 bit counter.
|
||||
name: "ib_recv_pkts",
|
||||
path: filepath.Join(countersDir, "port_rcv_packets"),
|
||||
unit: "packets",
|
||||
scale: 1,
|
||||
unit: ibPkgUnit,
|
||||
unitRates: ibPkgRateUnit,
|
||||
addToIBTotalPkgs: true,
|
||||
lastState: -1,
|
||||
},
|
||||
{
|
||||
// Total number of packets transmitted on all VLs from this port. This may include packets with errors.
|
||||
// This is 64 bit counter.
|
||||
name: "ib_xmit_pkts",
|
||||
path: filepath.Join(countersDir, "port_xmit_packets"),
|
||||
unit: "packets",
|
||||
scale: 1,
|
||||
unit: ibPkgUnit,
|
||||
unitRates: ibPkgRateUnit,
|
||||
addToIBTotalPkgs: true,
|
||||
lastState: -1,
|
||||
},
|
||||
}
|
||||
for _, counter := range portCounterFiles {
|
||||
@@ -163,7 +178,7 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
|
||||
|
||||
m.info = append(m.info,
|
||||
InfinibandCollectorInfo{
|
||||
LID: LID,
|
||||
lid: LID,
|
||||
device: device,
|
||||
port: port,
|
||||
portCounterFiles: portCounterFiles,
|
||||
@@ -184,7 +199,7 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Read reads Infiniband counter files below IB_BASEPATH
|
||||
// Read reads Infiniband counter files below ibBasePath
|
||||
func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMessage) {
|
||||
// Check if already initialized
|
||||
if !m.init {
|
||||
@@ -201,36 +216,42 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMess
|
||||
for i := range m.info {
|
||||
info := &m.info[i]
|
||||
|
||||
var ib_total, ib_total_last_state,
|
||||
ib_total_pkts, ib_total_pkts_last_state int64
|
||||
var ib_total_last_state_available, ib_total_pkts_last_state_available bool
|
||||
var ibTotal, ibTotalPkts uint64 // sum of xmit and recv counters
|
||||
var ibTotalBw, ibTotalPktsBw float64 // sum of xmit and recv rates
|
||||
var ibTotalBwAvailable, ibTotalPktsBwAvailable bool
|
||||
for i := range info.portCounterFiles {
|
||||
counterDef := &info.portCounterFiles[i]
|
||||
|
||||
// Read counter file
|
||||
line, err := os.ReadFile(counterDef.path)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to read from file '%s': %v", counterDef.path, err))
|
||||
"Read(): Failed to read from file '%s': %v", counterDef.path, err)
|
||||
// Current counter can not be saved as last state
|
||||
counterDef.lastStateAvailable = false
|
||||
continue
|
||||
}
|
||||
data := strings.TrimSpace(string(line))
|
||||
|
||||
// convert counter to int64
|
||||
v, err := strconv.ParseInt(data, 10, 64)
|
||||
// convert counter to uint64
|
||||
vRawCounter, err := strconv.ParseUint(data, 10, 64)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to convert Infininiband metrice %s='%s' to int64: %v", counterDef.name, data, err))
|
||||
"Read(): Failed to convert Infininiband metrice %s='%s' to uint64: %v", counterDef.name, data, err)
|
||||
// Current counter can not be saved as last state
|
||||
counterDef.lastStateAvailable = false
|
||||
continue
|
||||
}
|
||||
// Scale raw value
|
||||
v *= counterDef.scale
|
||||
vScaledCounter := vRawCounter
|
||||
if counterDef.scaleByFourLanes {
|
||||
vScaledCounter *= uint64(4)
|
||||
}
|
||||
|
||||
// Send absolut values
|
||||
if m.config.SendAbsoluteValues {
|
||||
if y, err := lp.NewMetric(counterDef.name, info.tagSet, m.meta, v, now); err == nil {
|
||||
if y, err := lp.NewMetric(counterDef.name, info.tagSet, m.meta, vScaledCounter, now); err == nil {
|
||||
y.AddMeta("unit", counterDef.unit)
|
||||
output <- y
|
||||
}
|
||||
@@ -238,63 +259,72 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMess
|
||||
|
||||
// Send derived values
|
||||
if m.config.SendDerivedValues {
|
||||
if counterDef.lastState >= 0 {
|
||||
rate := float64((v - counterDef.lastState)) / timeDiff
|
||||
if counterDef.lastStateAvailable {
|
||||
var rate float64
|
||||
// uint64 subtraction handles wraparound automatically
|
||||
// in case vRawCounter < counterDef.lastState we would compute:
|
||||
// math.MaxUint64 - lastState + vRawCounter + 1
|
||||
// = (2^64 - 1) - lastState + vRawCounter + 1
|
||||
// = 2^64 - lastState + vRawCounter
|
||||
// ≡ vRawCounter - lastState (mod 2^64)
|
||||
rate = float64(vRawCounter-counterDef.lastState) / timeDiff
|
||||
if counterDef.scaleByFourLanes {
|
||||
rate *= float64(4)
|
||||
}
|
||||
if y, err := lp.NewMetric(counterDef.name+"_bw", info.tagSet, m.meta, rate, now); err == nil {
|
||||
y.AddMeta("unit", counterDef.unit+"/sec")
|
||||
y.AddMeta("unit", counterDef.unitRates)
|
||||
output <- y
|
||||
}
|
||||
|
||||
// Sum up total values of last state
|
||||
// Sum up rates for total rates
|
||||
if m.config.SendTotalValues {
|
||||
switch {
|
||||
case counterDef.addToIBTotal:
|
||||
ib_total_last_state += counterDef.lastState
|
||||
ib_total_last_state_available = true
|
||||
ibTotalBw += rate
|
||||
ibTotalBwAvailable = true
|
||||
case counterDef.addToIBTotalPkgs:
|
||||
ib_total_pkts_last_state += counterDef.lastState
|
||||
ib_total_pkts_last_state_available = true
|
||||
ibTotalPktsBw += rate
|
||||
ibTotalPktsBwAvailable = true
|
||||
}
|
||||
}
|
||||
}
|
||||
counterDef.lastState = v
|
||||
counterDef.lastState = vRawCounter
|
||||
counterDef.lastStateAvailable = true
|
||||
}
|
||||
|
||||
// Sum up total values
|
||||
if m.config.SendTotalValues {
|
||||
switch {
|
||||
case counterDef.addToIBTotal:
|
||||
ib_total += v
|
||||
ibTotal += vScaledCounter
|
||||
case counterDef.addToIBTotalPkgs:
|
||||
ib_total_pkts += v
|
||||
ibTotalPkts += vScaledCounter
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Send total values
|
||||
if m.config.SendTotalValues {
|
||||
if y, err := lp.NewMetric("ib_total", info.tagSet, m.meta, ib_total, now); err == nil {
|
||||
y.AddMeta("unit", "bytes")
|
||||
if y, err := lp.NewMetric("ib_total", info.tagSet, m.meta, ibTotal, now); err == nil {
|
||||
y.AddMeta("unit", ibDataUnit)
|
||||
output <- y
|
||||
}
|
||||
|
||||
if y, err := lp.NewMetric("ib_total_pkts", info.tagSet, m.meta, ib_total_pkts, now); err == nil {
|
||||
y.AddMeta("unit", "packets")
|
||||
if y, err := lp.NewMetric("ib_total_pkts", info.tagSet, m.meta, ibTotalPkts, now); err == nil {
|
||||
y.AddMeta("unit", ibPkgUnit)
|
||||
output <- y
|
||||
}
|
||||
|
||||
if m.config.SendDerivedValues && 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")
|
||||
if m.config.SendDerivedValues && ibTotalBwAvailable {
|
||||
if y, err := lp.NewMetric("ib_total_bw", info.tagSet, m.meta, ibTotalBw, now); err == nil {
|
||||
y.AddMeta("unit", ibDataRateUnit)
|
||||
output <- y
|
||||
}
|
||||
}
|
||||
|
||||
if m.config.SendDerivedValues && 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")
|
||||
if m.config.SendDerivedValues && ibTotalPktsBwAvailable {
|
||||
if y, err := lp.NewMetric("ib_total_pkts_bw", info.tagSet, m.meta, ibTotalPktsBw, now); err == nil {
|
||||
y.AddMeta("unit", ibPkgRateUnit)
|
||||
output <- y
|
||||
}
|
||||
}
|
||||
|
||||
@@ -145,16 +145,16 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMessage)
|
||||
|
||||
file, err := os.Open(IOSTATFILE)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to open file '%s': %v", IOSTATFILE, err))
|
||||
"Read(): Failed to open file '%s': %v", IOSTATFILE, err)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
if err := file.Close(); err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to close file '%s': %v", IOSTATFILE, err))
|
||||
"Read(): Failed to close file '%s': %v", IOSTATFILE, err)
|
||||
}
|
||||
}()
|
||||
|
||||
|
||||
@@ -12,6 +12,12 @@ 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"
|
||||
|
||||
@@ -261,12 +267,12 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
|
||||
}
|
||||
for _, metric := range evset.Metrics {
|
||||
// Try to evaluate the metric
|
||||
cclog.ComponentDebug(m.name, "Checking", metric.Name)
|
||||
cclog.ComponentDebugf(m.name, "Checking %s", metric.Name)
|
||||
if !checkMetricType(metric.Type) {
|
||||
cclog.ComponentError(m.name, "Metric", metric.Name, "uses invalid type", metric.Type)
|
||||
cclog.ComponentErrorf(m.name, "Metric %s uses invalid type %s", metric.Name, metric.Type)
|
||||
metric.Calc = ""
|
||||
} else if !testLikwidMetricFormula(metric.Calc, params) {
|
||||
cclog.ComponentError(m.name, "Metric", metric.Name, "cannot be calculated with given counters")
|
||||
cclog.ComponentError(m.name, "Metric %s cannot be calculated with given counters", metric.Name)
|
||||
metric.Calc = ""
|
||||
} else {
|
||||
globalParams = append(globalParams, metric.Name)
|
||||
@@ -281,13 +287,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.ComponentError(m.name, "Metric", metric.Name, "uses invalid type", metric.Type)
|
||||
cclog.ComponentErrorf(m.name, "Metric %s uses invalid type %s", metric.Name, metric.Type)
|
||||
metric.Calc = ""
|
||||
} else if !testLikwidMetricFormula(metric.Calc, globalParams) {
|
||||
cclog.ComponentError(m.name, "Metric", metric.Name, "cannot be calculated with given counters")
|
||||
cclog.ComponentError(m.name, "Metric %s cannot be calculated with given counters", metric.Name)
|
||||
metric.Calc = ""
|
||||
} else if !checkMetricType(metric.Type) {
|
||||
cclog.ComponentError(m.name, "Metric", metric.Name, "has invalid type")
|
||||
cclog.ComponentError(m.name, "Metric %s has invalid type", metric.Name)
|
||||
metric.Calc = ""
|
||||
} else {
|
||||
totalMetrics++
|
||||
@@ -328,7 +334,7 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
|
||||
for _, c := range m.cpulist {
|
||||
m.measureThread.Call(
|
||||
func() {
|
||||
retCode := C.HPMaddThread(C.uint32_t(c))
|
||||
retCode := C.cc_add_hwthread(C.int(c))
|
||||
if retCode != 0 {
|
||||
err := fmt.Errorf("C.HPMaddThread(%v) failed with return code %v", c, retCode)
|
||||
cclog.ComponentError(m.name, err.Error())
|
||||
@@ -375,16 +381,16 @@ func (m *LikwidCollector) takeMeasurement(evidx int, evset LikwidEventsetConfig,
|
||||
// Watch changes for the lock file ()
|
||||
watcher, err := fsnotify.NewWatcher()
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("takeMeasurement(): Failed to create a new fsnotify.Watcher: %v", err))
|
||||
"takeMeasurement(): Failed to create a new fsnotify.Watcher: %v", err)
|
||||
return true, err
|
||||
}
|
||||
defer func() {
|
||||
if err := watcher.Close(); err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("takeMeasurement(): Failed to close fsnotify.Watcher: %v", err))
|
||||
"takeMeasurement(): Failed to close fsnotify.Watcher: %v", err)
|
||||
}
|
||||
}()
|
||||
if len(m.config.LockfilePath) > 0 {
|
||||
@@ -597,7 +603,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.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
|
||||
cclog.ComponentErrorf(m.name, "Calculation for metric %s failed: %s", metric.Name, err.Error())
|
||||
value = 0.0
|
||||
}
|
||||
if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) {
|
||||
@@ -762,7 +768,7 @@ func (m *LikwidCollector) calcGlobalMetrics(groups []LikwidEventsetConfig, inter
|
||||
// Evaluate the metric
|
||||
value, err := agg.EvalFloat64Condition(metric.Calc, params)
|
||||
if err != nil {
|
||||
cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
|
||||
cclog.ComponentErrorf(m.name, "Calculation for metric %s failed: %s", metric.Name, err.Error())
|
||||
value = 0.0
|
||||
}
|
||||
if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) {
|
||||
|
||||
@@ -89,9 +89,9 @@ func (m *LoadavgCollector) Read(interval time.Duration, output chan lp.CCMessage
|
||||
}
|
||||
buffer, err := os.ReadFile(LOADAVGFILE)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to read file '%s': %v", LOADAVGFILE, err))
|
||||
"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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to convert '%s' to float64: %v", ls[i], err))
|
||||
"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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to convert '%s' to float64: %v", lv[i], err))
|
||||
"Read(): Failed to convert '%s' to float64: %v", lv[i], err)
|
||||
continue
|
||||
}
|
||||
if m.proc_skips[i] {
|
||||
|
||||
@@ -72,7 +72,8 @@ func getStats(filename string) map[string]MemstatStats {
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
linefields := strings.Fields(line)
|
||||
if len(linefields) == 3 {
|
||||
switch len(linefields) {
|
||||
case 3:
|
||||
v, err := strconv.ParseFloat(linefields[1], 64)
|
||||
if err == nil {
|
||||
stats[strings.Trim(linefields[0], ":")] = MemstatStats{
|
||||
@@ -80,10 +81,10 @@ func getStats(filename string) map[string]MemstatStats {
|
||||
unit: linefields[2],
|
||||
}
|
||||
}
|
||||
} else if len(linefields) == 5 {
|
||||
case 5:
|
||||
v, err := strconv.ParseFloat(linefields[3], 64)
|
||||
if err == nil {
|
||||
cclog.ComponentDebug("getStats", strings.Trim(linefields[2], ":"), v, linefields[4])
|
||||
cclog.ComponentDebug("MemstatCollector", "getStats %s value %v unit %s", strings.Trim(linefields[2], ":"), v, linefields[4])
|
||||
stats[strings.Trim(linefields[2], ":")] = MemstatStats{
|
||||
value: v,
|
||||
unit: linefields[4],
|
||||
@@ -106,7 +107,10 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
|
||||
return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
|
||||
}
|
||||
}
|
||||
m.meta = map[string]string{"source": m.name, "group": "Memory"}
|
||||
m.meta = map[string]string{
|
||||
"source": m.name,
|
||||
"group": "Memory",
|
||||
}
|
||||
m.stats = make(map[string]int64)
|
||||
m.matches = make(map[string]string)
|
||||
m.tags = map[string]string{"type": "node"}
|
||||
@@ -145,7 +149,7 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
|
||||
"KernelStack": "mem_kernelstack",
|
||||
}
|
||||
for k, v := range matches {
|
||||
if !slices.Contains(m.config.ExcludeMetrics, k) {
|
||||
if !slices.Contains(m.config.ExcludeMetrics, v) {
|
||||
m.matches[k] = v
|
||||
}
|
||||
}
|
||||
@@ -153,7 +157,7 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
|
||||
if !slices.Contains(m.config.ExcludeMetrics, "mem_used") {
|
||||
m.sendMemUsed = true
|
||||
}
|
||||
if len(m.matches) == 0 {
|
||||
if len(m.matches) == 0 && !m.sendMemUsed {
|
||||
return fmt.Errorf("%s Init(): no metrics to collect", m.name)
|
||||
}
|
||||
if err := m.setup(); err != nil {
|
||||
|
||||
@@ -222,16 +222,16 @@ func (m *NetstatCollector) Read(interval time.Duration, output chan lp.CCMessage
|
||||
|
||||
file, err := os.Open(NETSTATFILE)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to open file '%s': %v", NETSTATFILE, err))
|
||||
"Read(): Failed to open file '%s': %v", NETSTATFILE, err)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
if err := file.Close(); err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to close file '%s': %v", NETSTATFILE, err))
|
||||
"Read(): Failed to close file '%s': %v", NETSTATFILE, err)
|
||||
}
|
||||
}()
|
||||
|
||||
|
||||
@@ -125,10 +125,9 @@ func (m *nfsCollector) Read(interval time.Duration, output chan lp.CCMessage) {
|
||||
timestamp := time.Now()
|
||||
|
||||
if err := m.updateStats(); err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): updateStats() failed: %v", err),
|
||||
)
|
||||
"Read(): updateStats() failed: %v", err)
|
||||
return
|
||||
}
|
||||
var prefix string
|
||||
|
||||
@@ -117,7 +117,7 @@ func (m *NUMAStatsCollector) Init(config json.RawMessage) error {
|
||||
}
|
||||
|
||||
// Initialized
|
||||
cclog.ComponentDebug(m.name, "initialized", len(m.topology), "NUMA domains")
|
||||
cclog.ComponentDebugf(m.name, "initialized %d NUMA domains", len(m.topology))
|
||||
m.init = true
|
||||
return nil
|
||||
}
|
||||
|
||||
396
collectors/nvidiaGPMMetric.go
Normal file
396
collectors/nvidiaGPMMetric.go
Normal file
@@ -0,0 +1,396 @@
|
||||
package collectors
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
|
||||
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
|
||||
"github.com/NVIDIA/go-nvml/pkg/nvml"
|
||||
)
|
||||
|
||||
type NvidiaGPMMetricDef struct {
|
||||
name string
|
||||
outname string
|
||||
id nvml.GpmMetricId
|
||||
unit string
|
||||
}
|
||||
|
||||
var NvidiaGPMMetrics []NvidiaGPMMetricDef = []NvidiaGPMMetricDef{
|
||||
{
|
||||
name: "GRAPHICS_UTIL",
|
||||
outname: "nv_gpm_graphics_util",
|
||||
id: nvml.GPM_METRIC_GRAPHICS_UTIL,
|
||||
unit: "%",
|
||||
},
|
||||
{
|
||||
name: "SM_UTIL",
|
||||
outname: "nv_gpm_sm_util",
|
||||
id: nvml.GPM_METRIC_SM_UTIL,
|
||||
unit: "%",
|
||||
},
|
||||
{
|
||||
name: "SM_OCCUPANCY",
|
||||
outname: "nv_gpm_sm_occupancy",
|
||||
id: nvml.GPM_METRIC_SM_OCCUPANCY,
|
||||
unit: "%",
|
||||
},
|
||||
{
|
||||
name: "INTEGER_UTIL",
|
||||
outname: "nv_gpm_integer_util",
|
||||
id: nvml.GPM_METRIC_INTEGER_UTIL,
|
||||
unit: "%",
|
||||
},
|
||||
{
|
||||
name: "ANY_TENSOR_UTIL",
|
||||
outname: "nv_gpm_any_tensor_util",
|
||||
id: nvml.GPM_METRIC_ANY_TENSOR_UTIL,
|
||||
unit: "%",
|
||||
},
|
||||
{
|
||||
name: "DFMA_TENSOR_UTIL",
|
||||
outname: "nv_gpm_dfma_tensor_util",
|
||||
id: nvml.GPM_METRIC_DFMA_TENSOR_UTIL,
|
||||
unit: "%",
|
||||
},
|
||||
{
|
||||
name: "HMMA_TENSOR_UTIL",
|
||||
outname: "nv_gpm_hmma_tensor_util",
|
||||
id: nvml.GPM_METRIC_HMMA_TENSOR_UTIL,
|
||||
unit: "%",
|
||||
},
|
||||
{
|
||||
name: "IMMA_TENSOR_UTIL",
|
||||
outname: "nv_gpm_imma_tensor_util",
|
||||
id: nvml.GPM_METRIC_IMMA_TENSOR_UTIL,
|
||||
unit: "%",
|
||||
},
|
||||
{
|
||||
name: "DRAM_BW_UTIL",
|
||||
outname: "nv_gpm_dram_bw_util",
|
||||
id: nvml.GPM_METRIC_DRAM_BW_UTIL,
|
||||
unit: "%",
|
||||
},
|
||||
{
|
||||
name: "FP64_UTIL",
|
||||
outname: "nv_gpm_fp64_util",
|
||||
id: nvml.GPM_METRIC_FP64_UTIL,
|
||||
unit: "%",
|
||||
},
|
||||
{
|
||||
name: "FP32_UTIL",
|
||||
outname: "nv_gpm_fp32_util",
|
||||
id: nvml.GPM_METRIC_FP32_UTIL,
|
||||
unit: "%",
|
||||
},
|
||||
{
|
||||
name: "FP16_UTIL",
|
||||
outname: "nv_gpm_fp16_util",
|
||||
id: nvml.GPM_METRIC_FP16_UTIL,
|
||||
unit: "%",
|
||||
},
|
||||
}
|
||||
|
||||
type NvidiaGPMCollectorConfig struct {
|
||||
Metrics []string `json:"metrics,omitempty"`
|
||||
ExcludeDevices []string `json:"exclude_devices,omitempty"`
|
||||
AddPciInfoTag bool `json:"add_pci_info_tag,omitempty"`
|
||||
UsePciInfoAsTypeId bool `json:"use_pci_info_as_type_id,omitempty"`
|
||||
AddUuidMeta bool `json:"add_uuid_meta,omitempty"`
|
||||
AddBoardNumberMeta bool `json:"add_board_number_meta,omitempty"`
|
||||
AddSerialMeta bool `json:"add_serial_meta,omitempty"`
|
||||
ProcessMigDevices bool `json:"process_mig_devices,omitempty"`
|
||||
UseUuidForMigDevices bool `json:"use_uuid_for_mig_device,omitempty"`
|
||||
UseSliceForMigDevices bool `json:"use_slice_for_mig_device,omitempty"`
|
||||
}
|
||||
|
||||
type NvidiaGPMCollectorDevice struct {
|
||||
device nvml.Device
|
||||
tags map[string]string
|
||||
meta map[string]string
|
||||
startTime time.Time
|
||||
endTime time.Time
|
||||
measurement nvml.GpmMetricsGetType
|
||||
metricsLookup map[int]NvidiaGPMMetricDef
|
||||
}
|
||||
|
||||
type NvidiaGPMCollector struct {
|
||||
metricCollector
|
||||
|
||||
config NvidiaGPMCollectorConfig
|
||||
gpus []NvidiaGPMCollectorDevice
|
||||
num_gpus int
|
||||
}
|
||||
|
||||
func (m *NvidiaGPMCollector) Init(config json.RawMessage) error {
|
||||
var err error = nil
|
||||
m.name = "NvidiaGPMCollector"
|
||||
m.parallel = true
|
||||
if err := m.setup(); err != nil {
|
||||
return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
|
||||
}
|
||||
if len(config) > 0 {
|
||||
d := json.NewDecoder(strings.NewReader(string(config)))
|
||||
d.DisallowUnknownFields()
|
||||
if err = d.Decode(&m.config); err != nil {
|
||||
return fmt.Errorf("%s Init(): Error decoding JSON config: %w", m.name, err)
|
||||
}
|
||||
}
|
||||
m.meta = map[string]string{
|
||||
"source": m.name,
|
||||
"group": "NvidiaGPM",
|
||||
}
|
||||
|
||||
// Initialize NVIDIA Management Library (NVML)
|
||||
ret := nvml.Init()
|
||||
|
||||
// Error: NVML library not found
|
||||
// (nvml.ErrorString can not be used in this case)
|
||||
if ret == nvml.ERROR_LIBRARY_NOT_FOUND {
|
||||
return fmt.Errorf("%s Init(): NVML library not found", m.name)
|
||||
}
|
||||
if ret != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(ret))
|
||||
return fmt.Errorf("%s Init(): Unable to initialize NVML: %w", m.name, err)
|
||||
}
|
||||
|
||||
// Number of NVIDIA GPUs
|
||||
num_gpus, ret := nvml.DeviceGetCount()
|
||||
if ret != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(ret))
|
||||
return fmt.Errorf("%s Init(): Unable to get device count: %w", m.name, err)
|
||||
}
|
||||
|
||||
// For all GPUs
|
||||
m.gpus = make([]NvidiaGPMCollectorDevice, 0, num_gpus)
|
||||
for i := range num_gpus {
|
||||
|
||||
// Skip excluded devices by ID
|
||||
str_i := strconv.Itoa(i)
|
||||
if slices.Contains(m.config.ExcludeDevices, str_i) {
|
||||
cclog.ComponentDebugf(m.name, "Skipping excluded device %s", str_i)
|
||||
continue
|
||||
}
|
||||
|
||||
// Get device handle
|
||||
device, ret := nvml.DeviceGetHandleByIndex(i)
|
||||
if ret != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(ret))
|
||||
cclog.ComponentErrorf(m.name, "Unable to get device at index %d: %s", i, err.Error())
|
||||
continue
|
||||
}
|
||||
|
||||
supportInfo, ret := nvml.GpmQueryDeviceSupport(device)
|
||||
if ret != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(ret))
|
||||
cclog.ComponentErrorf(m.name, "Unable to query GPM support for device at index %d: %s", i, err.Error())
|
||||
continue
|
||||
} else {
|
||||
if supportInfo.IsSupportedDevice == uint32(nvml.FEATURE_DISABLED) {
|
||||
cclog.ComponentErrorf(m.name, "Device at index %d does not support GPM metrics", i)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
stream, ret := nvml.GpmQueryIfStreamingEnabled(device)
|
||||
if ret != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(ret))
|
||||
cclog.ComponentErrorf(m.name, "Unable to query GPM streaming for device at index %d: %s", i, err.Error())
|
||||
continue
|
||||
} else {
|
||||
if stream == uint32(nvml.FEATURE_DISABLED) {
|
||||
ret = nvml.GpmSetStreamingEnabled(device, uint32(nvml.FEATURE_ENABLED))
|
||||
if ret != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(ret))
|
||||
cclog.ComponentErrorf(m.name, "Unable to set streaming mode for device at index %d: %s", i, err.Error())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Get device's PCI info
|
||||
pciInfo, ret := nvml.DeviceGetPciInfo(device)
|
||||
if ret != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(ret))
|
||||
cclog.ComponentErrorf(m.name, "Unable to get PCI info for device at index %d: %s", i, err.Error())
|
||||
continue
|
||||
}
|
||||
// Create PCI ID in the common format used by the NVML.
|
||||
pci_id := fmt.Sprintf(
|
||||
nvml.DEVICE_PCI_BUS_ID_FMT,
|
||||
pciInfo.Domain,
|
||||
pciInfo.Bus,
|
||||
pciInfo.Device)
|
||||
|
||||
// Skip excluded devices specified by PCI ID
|
||||
if slices.Contains(m.config.ExcludeDevices, pci_id) {
|
||||
cclog.ComponentDebugf(m.name, "Skipping excluded device %s", pci_id)
|
||||
continue
|
||||
}
|
||||
ss, nvmlErr := nvml.GpmSampleAlloc()
|
||||
if nvmlErr != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(ret))
|
||||
cclog.ComponentErrorf(m.name, "Failed to allocate GPM sample for device %d: %s", i, err.Error())
|
||||
continue
|
||||
}
|
||||
es, nvmlErr := nvml.GpmSampleAlloc()
|
||||
if nvmlErr != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(ret))
|
||||
cclog.ComponentErrorf(m.name, "Failed to allocate GPM sample for device %d: %s", i, err.Error())
|
||||
continue
|
||||
}
|
||||
|
||||
// Select which value to use as 'type-id'.
|
||||
// The PCI ID is commonly required in SLURM environments because the
|
||||
// numberic IDs used by SLURM and the ones used by NVML might differ
|
||||
// depending on the job type. The PCI ID is more reliable but is commonly
|
||||
// not recorded for a job, so it must be added manually in prologue or epilogue
|
||||
// e.g. to the comment field
|
||||
tid := str_i
|
||||
if m.config.UsePciInfoAsTypeId {
|
||||
tid = pci_id
|
||||
}
|
||||
|
||||
// Now we got all infos together, populate the device list
|
||||
g := NvidiaGPMCollectorDevice{}
|
||||
|
||||
// Add device handle
|
||||
g.device = device
|
||||
|
||||
// Add tags
|
||||
g.tags = map[string]string{
|
||||
"type": "accelerator",
|
||||
"type-id": tid,
|
||||
}
|
||||
|
||||
// Add PCI info as tag if not already used as 'type-id'
|
||||
if m.config.AddPciInfoTag && !m.config.UsePciInfoAsTypeId {
|
||||
g.tags["pci_identifier"] = pci_id
|
||||
}
|
||||
|
||||
g.meta = map[string]string{
|
||||
"source": m.name,
|
||||
"group": "Nvidia",
|
||||
}
|
||||
|
||||
if m.config.AddBoardNumberMeta {
|
||||
board, ret := nvml.DeviceGetBoardPartNumber(device)
|
||||
if ret != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(ret))
|
||||
cclog.ComponentError(m.name, "Unable to get boart part number for device at index", i, ":", err.Error())
|
||||
} else {
|
||||
g.meta["board_number"] = board
|
||||
}
|
||||
}
|
||||
if m.config.AddSerialMeta {
|
||||
serial, ret := nvml.DeviceGetSerial(device)
|
||||
if ret != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(ret))
|
||||
cclog.ComponentError(m.name, "Unable to get serial number for device at index", i, ":", err.Error())
|
||||
} else {
|
||||
g.meta["serial"] = serial
|
||||
}
|
||||
}
|
||||
if m.config.AddUuidMeta {
|
||||
uuid, ret := nvml.DeviceGetUUID(device)
|
||||
if ret != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(ret))
|
||||
cclog.ComponentError(m.name, "Unable to get UUID for device at index", i, ":", err.Error())
|
||||
} else {
|
||||
g.meta["uuid"] = uuid
|
||||
}
|
||||
}
|
||||
|
||||
g.measurement.Sample1 = ss
|
||||
g.measurement.Sample2 = es
|
||||
g.measurement.Version = nvml.GPM_METRICS_GET_VERSION
|
||||
g.metricsLookup = make(map[int]NvidiaGPMMetricDef)
|
||||
metIdx := 0
|
||||
for _, inmetric := range m.config.Metrics {
|
||||
for _, defmetric := range NvidiaGPMMetrics {
|
||||
if inmetric == defmetric.outname || inmetric == defmetric.name {
|
||||
g.measurement.Metrics[metIdx] = nvml.GpmMetric{
|
||||
MetricId: uint32(defmetric.id),
|
||||
}
|
||||
g.metricsLookup[metIdx] = defmetric
|
||||
metIdx += 1
|
||||
}
|
||||
}
|
||||
}
|
||||
g.measurement.NumMetrics = uint32(metIdx)
|
||||
m.gpus = append(m.gpus, g)
|
||||
}
|
||||
cclog.ComponentDebugf(m.name, "Found %d Nvidia GPUs with GPM support", len(m.gpus))
|
||||
m.num_gpus = len(m.gpus)
|
||||
m.init = true
|
||||
return err
|
||||
}
|
||||
|
||||
func (m *NvidiaGPMCollector) Read(interval time.Duration, output chan lp.CCMessage) {
|
||||
var err error
|
||||
if !m.init {
|
||||
return
|
||||
}
|
||||
for i, gpu := range m.gpus {
|
||||
gpu.startTime = time.Now()
|
||||
nvmlErr := gpu.measurement.Sample1.Get(gpu.device)
|
||||
if nvmlErr != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(nvmlErr))
|
||||
cclog.ComponentError(m.name, "Unable to get start GPM sample for device at index", i, ":", err.Error())
|
||||
continue
|
||||
}
|
||||
}
|
||||
time.Sleep(interval)
|
||||
|
||||
for i, gpu := range m.gpus {
|
||||
gpu.endTime = time.Now()
|
||||
nvmlErr := gpu.measurement.Sample2.Get(gpu.device)
|
||||
if nvmlErr != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(nvmlErr))
|
||||
cclog.ComponentError(m.name, "Unable to get stop GPM sample for device at index", i, ":", err.Error())
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
for i, gpu := range m.gpus {
|
||||
nvmlErr := nvml.GpmMetricsGet(&gpu.measurement)
|
||||
if nvmlErr != nvml.SUCCESS {
|
||||
err = errors.New(nvml.ErrorString(nvmlErr))
|
||||
cclog.ComponentError(m.name, "Unable to get evaluate GPM sample for device at index", i, ":", err.Error())
|
||||
continue
|
||||
}
|
||||
for idx, metricDef := range gpu.metricsLookup {
|
||||
y, err := lp.NewMetric(metricDef.outname, gpu.tags, gpu.meta, gpu.measurement.Metrics[idx].Value, time.Now())
|
||||
if err == nil {
|
||||
y.AddMeta("unit", metricDef.unit)
|
||||
output <- y
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (m *NvidiaGPMCollector) Close() {
|
||||
if m.init {
|
||||
for i, gpu := range m.gpus {
|
||||
ret := gpu.measurement.Sample1.Free()
|
||||
if ret != nvml.SUCCESS {
|
||||
err := errors.New(nvml.ErrorString(ret))
|
||||
cclog.ComponentErrorf(m.name, "Unable to free start sample for device at index %d: %s", i, err.Error())
|
||||
}
|
||||
ret = gpu.measurement.Sample2.Free()
|
||||
if ret != nvml.SUCCESS {
|
||||
err := errors.New(nvml.ErrorString(ret))
|
||||
cclog.ComponentErrorf(m.name, "Unable to free stop sample for device at index %d: %s", i, err.Error())
|
||||
}
|
||||
}
|
||||
if ret := nvml.Shutdown(); ret != nvml.SUCCESS {
|
||||
cclog.ComponentError(m.name, "nvml.Shutdown() not successful")
|
||||
}
|
||||
m.init = false
|
||||
}
|
||||
}
|
||||
54
collectors/nvidiaGPMMetric.md
Normal file
54
collectors/nvidiaGPMMetric.md
Normal file
@@ -0,0 +1,54 @@
|
||||
<!--
|
||||
---
|
||||
title: "Nvidia NVML GPM metric collector"
|
||||
description: Collect metrics for Nvidia GPUs using the NVML GPM interface
|
||||
categories: [cc-metric-collector]
|
||||
tags: ['Admin']
|
||||
weight: 2
|
||||
hugo_path: docs/reference/cc-metric-collector/collectors/nvidiaGPM.md
|
||||
---
|
||||
-->
|
||||
|
||||
## `nvidiaGPM` collector
|
||||
|
||||
```json
|
||||
"nvidia_gpm": {
|
||||
"metrics": [
|
||||
"nv_fb_mem_used",
|
||||
"nv_fan"
|
||||
],
|
||||
"exclude_devices": [
|
||||
"0","1", "0000000:ff:01.0"
|
||||
],
|
||||
|
||||
"process_mig_devices": false,
|
||||
"use_pci_info_as_type_id": true,
|
||||
"add_pci_info_tag": false,
|
||||
"add_uuid_meta": false,
|
||||
"add_board_number_meta": false,
|
||||
"add_serial_meta": false,
|
||||
"use_uuid_for_mig_device": false,
|
||||
"use_slice_for_mig_device": false
|
||||
}
|
||||
```
|
||||
|
||||
The `nvidia_gpm` collector can be configured to leave out specific devices with the `exclude_devices` option. It takes IDs as supplied to the NVML with `nvmlDeviceGetHandleByIndex()` or the PCI address in NVML format (`%08X:%02X:%02X.0`). Commonly only the physical GPUs are monitored. If MIG devices should be analyzed as well, set `process_mig_devices` (adds `stype=mig,stype-id=<mig_index>`). With the options `use_uuid_for_mig_device` and `use_slice_for_mig_device`, the `<mig_index>` can be replaced with the UUID (e.g. `MIG-6a9f7cc8-6d5b-5ce0-92de-750edc4d8849`) or the MIG slice name (e.g. `1g.5gb`).
|
||||
|
||||
The metrics sent by the `nvidia_gpm` collector use `accelerator` as `type` tag. For the `type-id`, it uses the device handle index by default. With the `use_pci_info_as_type_id` option, the PCI ID is used instead. If both values should be added as tags, activate the `add_pci_info_tag` option. It uses the device handle index as `type-id` and adds the PCI ID as separate `pci_identifier` tag.
|
||||
|
||||
Optionally, it is possible to add the UUID, the board part number and the serial to the meta informations. They are not sent to the sinks (if not configured otherwise).
|
||||
|
||||
|
||||
Available Metrics:
|
||||
* `nv_gpm_graphics_util`
|
||||
* `nv_gpm_sm_util`
|
||||
* `nv_gpm_sm_occupancy`
|
||||
* `nv_gpm_integer_util`
|
||||
* `nv_gpm_any_tensor_util`
|
||||
* `nv_gpm_dfma_tensor_util`
|
||||
* `nv_gpm_hmma_tensor_util`
|
||||
* `nv_gpm_imma_tensor_util`
|
||||
* `nv_gpm_dram_bw_util`
|
||||
* `nv_gpm_fp64_util`
|
||||
* `nv_gpm_fp32_util`
|
||||
* `nv_gpm_fp16_util`
|
||||
@@ -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.ComponentDebug(m.name, "Skipping excluded device", str_i)
|
||||
cclog.ComponentDebugf(m.name, "Skipping excluded device %s", 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.ComponentError(m.name, "Unable to get device at index", i, ":", err.Error())
|
||||
cclog.ComponentErrorf(m.name, "Unable to get device at index %d: %s", 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.ComponentError(m.name, "Unable to get PCI info for device at index", i, ":", err.Error())
|
||||
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.
|
||||
@@ -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.ComponentDebug(m.name, "Skipping excluded device", pci_id)
|
||||
cclog.ComponentDebugf(m.name, "Skipping excluded device %s", 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.ComponentError(m.name, "Unable to get boart part number for device at index", i, ":", err.Error())
|
||||
cclog.ComponentErrorf(m.name, "Unable to get boart part number for device at index %d: %s", 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.ComponentError(m.name, "Unable to get serial number for device at index", i, ":", err.Error())
|
||||
cclog.ComponentErrorf(m.name, "Unable to get serial number for device at index %d: %s", 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.ComponentError(m.name, "Unable to get UUID for device at index", i, ":", err.Error())
|
||||
cclog.ComponentErrorf(m.name, "Unable to get UUID for device at index %d: %s", i, err.Error())
|
||||
} else {
|
||||
g.meta["uuid"] = uuid
|
||||
}
|
||||
@@ -1115,31 +1115,6 @@ 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 {
|
||||
@@ -1153,102 +1128,97 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMessage)
|
||||
}
|
||||
err = readMemoryInfo(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readMemoryInfo for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readMemoryInfo for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readUtilization(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readUtilization for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readUtilization for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readTemp(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readTemp for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readTemp for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readFan(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readFan for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readFan for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readEccMode(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readEccMode for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readEccMode for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readPerfState(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readPerfState for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readPerfState for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readPowerUsage(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readPowerUsage for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readPowerUsage for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readEnergyConsumption(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readEnergyConsumption for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readEnergyConsumption for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readClocks(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readClocks for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readClocks for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readMaxClocks(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readMaxClocks for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readMaxClocks for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readEccErrors(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readEccErrors for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readEccErrors for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readPowerLimit(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readPowerLimit for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readPowerLimit for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readEncUtilization(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readEncUtilization for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readEncUtilization for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readDecUtilization(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readDecUtilization for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readDecUtilization for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readRemappedRows(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readRemappedRows for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readRemappedRows for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readBarMemoryInfo(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readBarMemoryInfo for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readBarMemoryInfo for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readProcessCounts(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readProcessCounts for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readProcessCounts for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readViolationStats(device, output)
|
||||
if err != nil {
|
||||
cclog.ComponentDebug(m.name, "readViolationStats for device", name, "failed")
|
||||
cclog.ComponentDebugf(m.name, "readViolationStats for device %s failed", name)
|
||||
}
|
||||
|
||||
err = readNVLinkStats(device, output)
|
||||
if err != nil {
|
||||
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")
|
||||
cclog.ComponentDebugf(m.name, "readNVLinkStats for device %s failed", name)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1274,7 +1244,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMessage)
|
||||
if maxMig == 0 {
|
||||
continue
|
||||
}
|
||||
cclog.ComponentDebug(m.name, "Reading MIG devices for GPU", i)
|
||||
cclog.ComponentDebugf(m.name, "Reading MIG devices for GPU %d", i)
|
||||
|
||||
for j := range maxMig {
|
||||
mdev, ret := nvml.DeviceGetMigDeviceHandleByIndex(m.gpus[i].device, j)
|
||||
@@ -1298,7 +1268,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.ComponentError(m.name, "Unable to get UUID for mig device at index", j, ":", err.Error())
|
||||
cclog.ComponentErrorf(m.name, "Unable to get UUID for mig device at index %d: %s", j, err.Error())
|
||||
} else {
|
||||
migDevice.tags["stype-id"] = uuid
|
||||
}
|
||||
|
||||
@@ -85,6 +85,5 @@ 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>`.
|
||||
|
||||
@@ -208,11 +208,10 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
|
||||
}
|
||||
|
||||
// Initialized
|
||||
cclog.ComponentDebug(
|
||||
cclog.ComponentDebugf(
|
||||
m.name,
|
||||
"initialized",
|
||||
len(m.RAPLZoneInfo),
|
||||
"zones with running average power limit (RAPL) monitoring attributes")
|
||||
"initialized %d zones with running average power limit (RAPL) monitoring attributes",
|
||||
len(m.RAPLZoneInfo))
|
||||
m.init = true
|
||||
|
||||
return err
|
||||
|
||||
@@ -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.ComponentError(m.name, "Unable to get serial number for device at index", i, ":", rocm_smi.StatusStringNoError(ret))
|
||||
cclog.ComponentErrorf(m.name, "Unable to get serial number for device at index %d: %s", 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.ComponentError(m.name, "Unable to get metrics for device at index", dev.index, ":", rocm_smi.StatusStringNoError(ret))
|
||||
cclog.ComponentErrorf(m.name, "Unable to get metrics for device at index %d: %s", dev.index, rocm_smi.StatusStringNoError(ret))
|
||||
continue
|
||||
}
|
||||
|
||||
|
||||
@@ -147,15 +147,15 @@ func (m *SchedstatCollector) Read(interval time.Duration, output chan lp.CCMessa
|
||||
|
||||
file, err := os.Open(SCHEDSTATFILE)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to open file '%s': %v", SCHEDSTATFILE, err))
|
||||
"Read(): Failed to open file '%s': %v", SCHEDSTATFILE, err)
|
||||
}
|
||||
defer func() {
|
||||
if err := file.Close(); err != nil {
|
||||
cclog.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to close file '%s': %v", SCHEDSTATFILE, err))
|
||||
"Read(): Failed to close file '%s': %v", SCHEDSTATFILE, err)
|
||||
}
|
||||
}()
|
||||
|
||||
|
||||
@@ -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.ComponentError(m.name, "Error globbing job directories:", err.Error())
|
||||
cclog.ComponentErrorf(m.name, "Error globbing job directories: %s", 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", jKey, ":", err.Error())
|
||||
cclog.ComponentError(m.name, "Error reading job data for %s: %s", jKey, err.Error())
|
||||
continue
|
||||
}
|
||||
|
||||
|
||||
@@ -228,12 +228,12 @@ func (m *SmartMonCollector) Read(interval time.Duration, output chan lp.CCMessag
|
||||
|
||||
stdout, err := command.Output()
|
||||
if err != nil {
|
||||
cclog.ComponentError(m.name, "cannot read data for device", d.Name)
|
||||
cclog.ComponentErrorf(m.name, "cannot read data for device %s", d.Name)
|
||||
continue
|
||||
}
|
||||
err = json.Unmarshal(stdout, &data)
|
||||
if err != nil {
|
||||
cclog.ComponentError(m.name, "cannot unmarshal data for device", d.Name)
|
||||
cclog.ComponentErrorf(m.name, "cannot unmarshal data for device %s", d.Name)
|
||||
continue
|
||||
}
|
||||
if !m.excludeMetric.temp {
|
||||
|
||||
@@ -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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to read file '%s': %v", sensor.file, err))
|
||||
"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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to convert temperature '%s' to int64: %v", buffer, err))
|
||||
"Read(): Failed to convert temperature '%s' to int64: %v", buffer, err)
|
||||
continue
|
||||
}
|
||||
x /= 1000
|
||||
|
||||
@@ -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.ComponentError(
|
||||
cclog.ComponentErrorf(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to read output from command \"%s\": %v", command.String(), err))
|
||||
"Read(): Failed to read output from command \"%s\": %v", command.String(), err)
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
10
go.mod
10
go.mod
@@ -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.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
|
||||
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
|
||||
)
|
||||
|
||||
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.11.0 // indirect
|
||||
github.com/tklauser/numcpus v0.12.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
21
go.sum
@@ -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.0 h1:Xx/5Ydg9CeBDX/wi4VJqStNtohYjitZhhlHt4h3St1M=
|
||||
github.com/fsnotify/fsnotify v1.10.0/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo=
|
||||
github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx59Ho=
|
||||
github.com/fsnotify/fsnotify v1.10.1/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo=
|
||||
github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
|
||||
github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
|
||||
github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
|
||||
@@ -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.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/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRUgPQU=
|
||||
github.com/tklauser/go-sysconf v0.4.0/go.mod h1:8mTNWyog7H+MpKijp4VmKJAd2bbYQ2zuUwkYRbUArPI=
|
||||
github.com/tklauser/numcpus v0.12.0 h1:NR85qdvHA9pFse3x3weVZ0r0ST8R6l5RHbZrlRaqob4=
|
||||
github.com/tklauser/numcpus v0.12.0/go.mod h1:ABHeXzJnr/qqwguhClkZKT1/8VABcYrsyUiUGobwWJg=
|
||||
github.com/yusufpapurcu/wmi v1.2.3 h1:E1ctvB7uKFMOJw3fdOW32DwGE9I7t++CRUEMKvFoFiw=
|
||||
github.com/yusufpapurcu/wmi v1.2.3/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
|
||||
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
|
||||
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
|
||||
go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
|
||||
go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
|
||||
golang.design/x/thread v0.0.0-20210122121316-335e9adffdf1 h1:P7S/GeHBAFEZIYp0ePPs2kHXoazz8q2KsyxHyQVGCJg=
|
||||
golang.design/x/thread v0.0.0-20210122121316-335e9adffdf1/go.mod h1:9CWpnTUmlQkfdpdutA1nNf4iE5lAVt3QZOu0Z6hahBE=
|
||||
golang.design/x/thread v0.3.2 h1:FmD1glspGrQCe6FuQLmSrT6wz2CSzq7vKVDluyiMnqo=
|
||||
golang.design/x/thread v0.3.2/go.mod h1:6+Hi2rMOgMHZdKDWaqNHyWtoFUx1HxZ06LfHPh5Z/hQ=
|
||||
golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI=
|
||||
golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q=
|
||||
golang.org/x/exp v0.0.0-20231005195138-3e424a577f31 h1:9k5exFQKQglLo+RoP+4zMjOFE14P6+vyR0baDAi0Rcs=
|
||||
@@ -183,9 +183,8 @@ golang.org/x/mod v0.13.0 h1:I/DsJXRlw/8l/0c24sM9yb0T4z9liZTduXvdAWYiysY=
|
||||
golang.org/x/mod v0.13.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
|
||||
golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
|
||||
golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
|
||||
golang.org/x/sys v0.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/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
|
||||
golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
||||
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
|
||||
golang.org/x/tools v0.14.0 h1:jvNa2pY0M4r62jkRQ6RwEZZyPcymeL9XZMLBbV7U2nc=
|
||||
|
||||
@@ -35,18 +35,18 @@ type metricRouterTagConfig struct {
|
||||
|
||||
// Metric router configuration
|
||||
type metricRouterConfig struct {
|
||||
HostnameTagName string `json:"hostname_tag"` // Key name used when adding the hostname to a metric (default 'hostname')
|
||||
AddTags []metricRouterTagConfig `json:"add_tags"` // List of tags that are added when the condition is met
|
||||
DelTags []metricRouterTagConfig `json:"delete_tags"` // List of tags that are removed when the condition is met
|
||||
IntervalAgg []agg.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
|
||||
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
|
||||
MessageProcessor json.RawMessage `json:"process_messages,omitempty"`
|
||||
}
|
||||
|
||||
@@ -297,7 +297,7 @@ func (r *metricRouter) Start() {
|
||||
|
||||
case timestamp := <-timeChan:
|
||||
r.timestamp = timestamp
|
||||
cclog.ComponentDebug("MetricRouter", "Update timestamp", r.timestamp.UnixNano())
|
||||
cclog.ComponentDebugf("MetricRouter", "Update timestamp %d", r.timestamp.UnixNano())
|
||||
|
||||
case p := <-r.coll_input:
|
||||
coll_forward(p)
|
||||
|
||||
Reference in New Issue
Block a user