mirror of
https://github.com/ClusterCockpit/cc-metric-collector.git
synced 2025-07-19 19:31:41 +02:00
Compare commits
23 Commits
numastats_
...
rapl_colle
Author | SHA1 | Date | |
---|---|---|---|
|
77fceb17d1 | ||
|
b5520efc25 | ||
|
d2b1bad1b8 | ||
|
0e57c8db1c | ||
|
f2f38c81af | ||
|
f9acc51a50 | ||
|
87346e2eae | ||
|
0f92f10b66 | ||
|
6901b06e44 | ||
|
7b343d0bab | ||
|
7d3180b526 | ||
|
70a6afc549 | ||
|
e02a018327 | ||
|
bcecdd033b | ||
|
2645ffeff3 | ||
|
e968aa1991 | ||
|
d2a38e3844 | ||
|
1f35f6d3ca | ||
|
7e6870c7b3 | ||
|
d881093524 | ||
|
c01096c157 | ||
|
3d70c8afc9 | ||
|
7ee85a07dc |
@@ -8,23 +8,21 @@ import (
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
)
|
||||
|
||||
// "log"
|
||||
|
||||
const MOUNTFILE = `/proc/self/mounts`
|
||||
|
||||
type DiskstatCollectorConfig struct {
|
||||
ExcludeMetrics []string `json:"exclude_metrics,omitempty"`
|
||||
ExcludeMounts []string `json:"exclude_mounts,omitempty"`
|
||||
}
|
||||
|
||||
type DiskstatCollector struct {
|
||||
metricCollector
|
||||
//matches map[string]int
|
||||
config IOstatCollectorConfig
|
||||
//devices map[string]IOstatCollectorEntry
|
||||
config DiskstatCollectorConfig
|
||||
allowedMetrics map[string]bool
|
||||
}
|
||||
|
||||
func (m *DiskstatCollector) Init(config json.RawMessage) error {
|
||||
@@ -33,12 +31,21 @@ func (m *DiskstatCollector) Init(config json.RawMessage) error {
|
||||
m.meta = map[string]string{"source": m.name, "group": "Disk"}
|
||||
m.setup()
|
||||
if len(config) > 0 {
|
||||
err := json.Unmarshal(config, &m.config)
|
||||
if err != nil {
|
||||
if err := json.Unmarshal(config, &m.config); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
file, err := os.Open(string(MOUNTFILE))
|
||||
m.allowedMetrics = map[string]bool{
|
||||
"disk_total": true,
|
||||
"disk_free": true,
|
||||
"part_max_used": true,
|
||||
}
|
||||
for _, excl := range m.config.ExcludeMetrics {
|
||||
if _, ok := m.allowedMetrics[excl]; ok {
|
||||
m.allowedMetrics[excl] = false
|
||||
}
|
||||
}
|
||||
file, err := os.Open(MOUNTFILE)
|
||||
if err != nil {
|
||||
cclog.ComponentError(m.name, err.Error())
|
||||
return err
|
||||
@@ -53,7 +60,7 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMessag
|
||||
return
|
||||
}
|
||||
|
||||
file, err := os.Open(string(MOUNTFILE))
|
||||
file, err := os.Open(MOUNTFILE)
|
||||
if err != nil {
|
||||
cclog.ComponentError(m.name, err.Error())
|
||||
return
|
||||
@@ -62,6 +69,7 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMessag
|
||||
|
||||
part_max_used := uint64(0)
|
||||
scanner := bufio.NewScanner(file)
|
||||
mountLoop:
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
if len(line) == 0 {
|
||||
@@ -77,13 +85,17 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMessag
|
||||
if strings.Contains(linefields[1], "boot") {
|
||||
continue
|
||||
}
|
||||
path := strings.Replace(linefields[1], `\040`, " ", -1)
|
||||
stat := syscall.Statfs_t{
|
||||
Blocks: 0,
|
||||
Bsize: 0,
|
||||
Bfree: 0,
|
||||
|
||||
mountPath := strings.Replace(linefields[1], `\040`, " ", -1)
|
||||
|
||||
for _, excl := range m.config.ExcludeMounts {
|
||||
if strings.Contains(mountPath, excl) {
|
||||
continue mountLoop
|
||||
}
|
||||
}
|
||||
err := syscall.Statfs(path, &stat)
|
||||
|
||||
stat := syscall.Statfs_t{}
|
||||
err := syscall.Statfs(mountPath, &stat)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
@@ -92,16 +104,20 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMessag
|
||||
}
|
||||
tags := map[string]string{"type": "node", "device": linefields[0]}
|
||||
total := (stat.Blocks * uint64(stat.Bsize)) / uint64(1000000000)
|
||||
y, err := lp.NewMessage("disk_total", tags, m.meta, map[string]interface{}{"value": total}, time.Now())
|
||||
if err == nil {
|
||||
y.AddMeta("unit", "GBytes")
|
||||
output <- y
|
||||
if m.allowedMetrics["disk_total"] {
|
||||
y, err := lp.NewMessage("disk_total", tags, m.meta, map[string]interface{}{"value": total}, time.Now())
|
||||
if err == nil {
|
||||
y.AddMeta("unit", "GBytes")
|
||||
output <- y
|
||||
}
|
||||
}
|
||||
free := (stat.Bfree * uint64(stat.Bsize)) / uint64(1000000000)
|
||||
y, err = lp.NewMessage("disk_free", tags, m.meta, map[string]interface{}{"value": free}, time.Now())
|
||||
if err == nil {
|
||||
y.AddMeta("unit", "GBytes")
|
||||
output <- y
|
||||
if m.allowedMetrics["disk_free"] {
|
||||
y, err := lp.NewMessage("disk_free", tags, m.meta, map[string]interface{}{"value": free}, time.Now())
|
||||
if err == nil {
|
||||
y.AddMeta("unit", "GBytes")
|
||||
output <- y
|
||||
}
|
||||
}
|
||||
if total > 0 {
|
||||
perc := (100 * (total - free)) / total
|
||||
@@ -110,10 +126,12 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMessag
|
||||
}
|
||||
}
|
||||
}
|
||||
y, err := lp.NewMessage("part_max_used", map[string]string{"type": "node"}, m.meta, map[string]interface{}{"value": int(part_max_used)}, time.Now())
|
||||
if err == nil {
|
||||
y.AddMeta("unit", "percent")
|
||||
output <- y
|
||||
if m.allowedMetrics["part_max_used"] {
|
||||
y, err := lp.NewMessage("part_max_used", map[string]string{"type": "node"}, m.meta, map[string]interface{}{"value": int(part_max_used)}, time.Now())
|
||||
if err == nil {
|
||||
y.AddMeta("unit", "percent")
|
||||
output <- y
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
@@ -6,10 +6,13 @@
|
||||
"exclude_metrics": [
|
||||
"disk_total"
|
||||
],
|
||||
"exclude_mounts": [
|
||||
"slurm-tmpfs"
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
The `diskstat` collector reads data from `/proc/self/mounts` and outputs a handful **node** metrics. If a metric is not required, it can be excluded from forwarding it to the sink.
|
||||
The `diskstat` collector reads data from `/proc/self/mounts` and outputs a handful **node** metrics. If a metric is not required, it can be excluded from forwarding it to the sink. Additionally, any mount point containing one of the strings specified in `exclude_mounts` will be skipped during metric collection.
|
||||
|
||||
Metrics per device (with `device` tag):
|
||||
* `disk_total` (unit `GBytes`)
|
||||
|
@@ -4,8 +4,8 @@ import (
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
"golang.org/x/sys/unix"
|
||||
|
||||
"encoding/json"
|
||||
|
@@ -2,24 +2,24 @@ package collectors
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"os"
|
||||
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
|
||||
|
||||
// "log"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
)
|
||||
|
||||
// Konstante für den Pfad zu /proc/diskstats
|
||||
const IOSTATFILE = `/proc/diskstats`
|
||||
const IOSTAT_SYSFSPATH = `/sys/block`
|
||||
|
||||
type IOstatCollectorConfig struct {
|
||||
ExcludeMetrics []string `json:"exclude_metrics,omitempty"`
|
||||
// Neues Feld zum Ausschließen von Devices per JSON-Konfiguration
|
||||
ExcludeDevices []string `json:"exclude_devices,omitempty"`
|
||||
}
|
||||
|
||||
type IOstatCollectorEntry struct {
|
||||
@@ -76,7 +76,7 @@ func (m *IOstatCollector) Init(config json.RawMessage) error {
|
||||
if len(m.matches) == 0 {
|
||||
return errors.New("no metrics to collect")
|
||||
}
|
||||
file, err := os.Open(string(IOSTATFILE))
|
||||
file, err := os.Open(IOSTATFILE)
|
||||
if err != nil {
|
||||
cclog.ComponentError(m.name, err.Error())
|
||||
return err
|
||||
@@ -87,17 +87,24 @@ func (m *IOstatCollector) Init(config json.RawMessage) error {
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
linefields := strings.Fields(line)
|
||||
if len(linefields) < 3 {
|
||||
continue
|
||||
}
|
||||
device := linefields[2]
|
||||
|
||||
if strings.Contains(device, "loop") {
|
||||
continue
|
||||
}
|
||||
if _, skip := stringArrayContains(m.config.ExcludeDevices, device); skip {
|
||||
continue
|
||||
}
|
||||
values := make(map[string]int64)
|
||||
for m := range m.matches {
|
||||
values[m] = 0
|
||||
}
|
||||
m.devices[device] = IOstatCollectorEntry{
|
||||
tags: map[string]string{
|
||||
"device": linefields[2],
|
||||
"device": device,
|
||||
"type": "node",
|
||||
},
|
||||
lastValues: values,
|
||||
@@ -112,7 +119,7 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMessage)
|
||||
return
|
||||
}
|
||||
|
||||
file, err := os.Open(string(IOSTATFILE))
|
||||
file, err := os.Open(IOSTATFILE)
|
||||
if err != nil {
|
||||
cclog.ComponentError(m.name, err.Error())
|
||||
return
|
||||
@@ -126,10 +133,16 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMessage)
|
||||
continue
|
||||
}
|
||||
linefields := strings.Fields(line)
|
||||
if len(linefields) < 3 {
|
||||
continue
|
||||
}
|
||||
device := linefields[2]
|
||||
if strings.Contains(device, "loop") {
|
||||
continue
|
||||
}
|
||||
if _, skip := stringArrayContains(m.config.ExcludeDevices, device); skip {
|
||||
continue
|
||||
}
|
||||
if _, ok := m.devices[device]; !ok {
|
||||
continue
|
||||
}
|
||||
|
@@ -4,12 +4,17 @@
|
||||
```json
|
||||
"iostat": {
|
||||
"exclude_metrics": [
|
||||
"read_ms"
|
||||
"io_read_ms"
|
||||
],
|
||||
"exclude_devices": [
|
||||
"nvme0n1p1",
|
||||
"nvme0n1p2",
|
||||
"md127"
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
The `iostat` collector reads data from `/proc/diskstats` and outputs a handful **node** metrics. If a metric is not required, it can be excluded from forwarding it to the sink.
|
||||
The `iostat` collector reads data from `/proc/diskstats` and outputs a handful **node** metrics. If a metric or device is not required, it can be excluded from forwarding it to the sink.
|
||||
|
||||
Metrics:
|
||||
* `io_reads`
|
||||
|
@@ -12,8 +12,8 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
)
|
||||
|
||||
const MEMSTATFILE = "/proc/meminfo"
|
||||
|
@@ -9,16 +9,17 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
)
|
||||
|
||||
const NETSTATFILE = "/proc/net/dev"
|
||||
|
||||
type NetstatCollectorConfig struct {
|
||||
IncludeDevices []string `json:"include_devices"`
|
||||
SendAbsoluteValues bool `json:"send_abs_values"`
|
||||
SendDerivedValues bool `json:"send_derived_values"`
|
||||
IncludeDevices []string `json:"include_devices"`
|
||||
SendAbsoluteValues bool `json:"send_abs_values"`
|
||||
SendDerivedValues bool `json:"send_derived_values"`
|
||||
InterfaceAliases map[string][]string `json:"interface_aliases,omitempty"`
|
||||
}
|
||||
|
||||
type NetstatCollectorMetric struct {
|
||||
@@ -32,9 +33,26 @@ type NetstatCollectorMetric struct {
|
||||
|
||||
type NetstatCollector struct {
|
||||
metricCollector
|
||||
config NetstatCollectorConfig
|
||||
matches map[string][]NetstatCollectorMetric
|
||||
lastTimestamp time.Time
|
||||
config NetstatCollectorConfig
|
||||
aliasToCanonical map[string]string
|
||||
matches map[string][]NetstatCollectorMetric
|
||||
lastTimestamp time.Time
|
||||
}
|
||||
|
||||
func (m *NetstatCollector) buildAliasMapping() {
|
||||
m.aliasToCanonical = make(map[string]string)
|
||||
for canon, aliases := range m.config.InterfaceAliases {
|
||||
for _, alias := range aliases {
|
||||
m.aliasToCanonical[alias] = canon
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func getCanonicalName(raw string, aliasToCanonical map[string]string) string {
|
||||
if canon, ok := aliasToCanonical[raw]; ok {
|
||||
return canon
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
func (m *NetstatCollector) Init(config json.RawMessage) error {
|
||||
@@ -77,6 +95,8 @@ func (m *NetstatCollector) Init(config json.RawMessage) error {
|
||||
}
|
||||
}
|
||||
|
||||
m.buildAliasMapping()
|
||||
|
||||
// Check access to net statistic file
|
||||
file, err := os.Open(NETSTATFILE)
|
||||
if err != nil {
|
||||
@@ -97,18 +117,20 @@ func (m *NetstatCollector) Init(config json.RawMessage) error {
|
||||
// Split line into fields
|
||||
f := strings.Fields(l)
|
||||
|
||||
// Get net device entry
|
||||
dev := strings.Trim(f[0], ": ")
|
||||
// Get raw and canonical names
|
||||
raw := strings.Trim(f[0], ": ")
|
||||
canonical := getCanonicalName(raw, m.aliasToCanonical)
|
||||
|
||||
// Check if device is a included device
|
||||
if _, ok := stringArrayContains(m.config.IncludeDevices, dev); ok {
|
||||
tags := map[string]string{"stype": "network", "stype-id": dev, "type": "node"}
|
||||
if _, ok := stringArrayContains(m.config.IncludeDevices, canonical); ok {
|
||||
// Tag will contain original device name (raw).
|
||||
tags := map[string]string{"stype": "network", "stype-id": raw, "type": "node"}
|
||||
meta_unit_byte := map[string]string{"source": m.name, "group": "Network", "unit": "bytes"}
|
||||
meta_unit_byte_per_sec := map[string]string{"source": m.name, "group": "Network", "unit": "bytes/sec"}
|
||||
meta_unit_pkts := map[string]string{"source": m.name, "group": "Network", "unit": "packets"}
|
||||
meta_unit_pkts_per_sec := map[string]string{"source": m.name, "group": "Network", "unit": "packets/sec"}
|
||||
|
||||
m.matches[dev] = []NetstatCollectorMetric{
|
||||
m.matches[canonical] = []NetstatCollectorMetric{
|
||||
{
|
||||
name: "net_bytes_in",
|
||||
index: fieldReceiveBytes,
|
||||
@@ -143,7 +165,6 @@ func (m *NetstatCollector) Init(config json.RawMessage) error {
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
if len(m.matches) == 0 {
|
||||
@@ -164,7 +185,7 @@ func (m *NetstatCollector) Read(interval time.Duration, output chan lp.CCMessage
|
||||
// Save current timestamp
|
||||
m.lastTimestamp = now
|
||||
|
||||
file, err := os.Open(string(NETSTATFILE))
|
||||
file, err := os.Open(NETSTATFILE)
|
||||
if err != nil {
|
||||
cclog.ComponentError(m.name, err.Error())
|
||||
return
|
||||
@@ -183,11 +204,12 @@ func (m *NetstatCollector) Read(interval time.Duration, output chan lp.CCMessage
|
||||
// Split line into fields
|
||||
f := strings.Fields(l)
|
||||
|
||||
// Get net device entry
|
||||
dev := strings.Trim(f[0], ":")
|
||||
// Get raw and canonical names
|
||||
raw := strings.Trim(f[0], ":")
|
||||
canonical := getCanonicalName(raw, m.aliasToCanonical)
|
||||
|
||||
// Check if device is a included device
|
||||
if devmetrics, ok := m.matches[dev]; ok {
|
||||
if devmetrics, ok := m.matches[canonical]; ok {
|
||||
for i := range devmetrics {
|
||||
metric := &devmetrics[i]
|
||||
|
||||
|
@@ -4,14 +4,19 @@
|
||||
```json
|
||||
"netstat": {
|
||||
"include_devices": [
|
||||
"eth0"
|
||||
"eth0",
|
||||
"eno1"
|
||||
],
|
||||
"send_abs_values" : true,
|
||||
"send_derived_values" : true
|
||||
"send_abs_values": true,
|
||||
"send_derived_values": true,
|
||||
"interface_aliases": {
|
||||
"eno1": ["eno1np0", "eno1_alt"],
|
||||
"eth0": ["eth0_alias"]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The `netstat` collector reads data from `/proc/net/dev` and outputs a handful **node** metrics. With the `include_devices` list you can specify which network devices should be measured. **Note**: Most other collectors use an _exclude_ list instead of an include list.
|
||||
The `netstat` collector reads data from `/proc/net/dev` and outputs a handful **node** metrics. With the `include_devices` list you can specify which network devices should be measured. **Note**: Most other collectors use an _exclude_ list instead of an include list. Optionally, you can define an interface_aliases mapping. For each canonical device (as listed in include_devices), you may provide an array of aliases that may be reported by the system. When an alias is detected, it is preferred for matching, while the output tag stype-id always shows the actual system-reported name.
|
||||
|
||||
Metrics:
|
||||
* `net_bytes_in` (`unit=bytes`)
|
||||
@@ -23,5 +28,4 @@ Metrics:
|
||||
* `net_pkts_in_bw` (`unit=packets/sec` if `send_derived_values == true`)
|
||||
* `net_pkts_out_bw` (`unit=packets/sec` if `send_derived_values == true`)
|
||||
|
||||
The device name is added as tag `stype=network,stype-id=<device>`.
|
||||
|
||||
The device name is added as tag `stype=network,stype-id=<device>`.
|
@@ -9,8 +9,8 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
)
|
||||
|
||||
// These are the fields we read from the JSON configuration
|
||||
@@ -18,17 +18,20 @@ type NfsIOStatCollectorConfig struct {
|
||||
ExcludeMetrics []string `json:"exclude_metrics,omitempty"`
|
||||
ExcludeFilesystem []string `json:"exclude_filesystem,omitempty"`
|
||||
UseServerAddressAsSType bool `json:"use_server_as_stype,omitempty"`
|
||||
SendAbsoluteValues bool `json:"send_abs_values"`
|
||||
SendDerivedValues bool `json:"send_derived_values"`
|
||||
}
|
||||
|
||||
// This contains all variables we need during execution and the variables
|
||||
// defined by metricCollector (name, init, ...)
|
||||
type NfsIOStatCollector struct {
|
||||
metricCollector
|
||||
config NfsIOStatCollectorConfig // the configuration structure
|
||||
meta map[string]string // default meta information
|
||||
tags map[string]string // default tags
|
||||
data map[string]map[string]int64 // data storage for difference calculation
|
||||
key string // which device info should be used as subtype ID? 'server' or 'mntpoint', see NfsIOStatCollectorConfig.UseServerAddressAsSType
|
||||
config NfsIOStatCollectorConfig // the configuration structure
|
||||
meta map[string]string // default meta information
|
||||
tags map[string]string // default tags
|
||||
data map[string]map[string]int64 // data storage for difference calculation
|
||||
key string // which device info should be used as subtype ID? 'server' or 'mntpoint'
|
||||
lastTimestamp time.Time
|
||||
}
|
||||
|
||||
var deviceRegex = regexp.MustCompile(`device (?P<server>[^ ]+) mounted on (?P<mntpoint>[^ ]+) with fstype nfs(?P<version>\d*) statvers=[\d\.]+`)
|
||||
@@ -81,7 +84,6 @@ func (m *NfsIOStatCollector) readNfsiostats() map[string]map[string]int64 {
|
||||
data[current[m.key]][name] = val
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
current = nil
|
||||
}
|
||||
@@ -98,6 +100,9 @@ func (m *NfsIOStatCollector) Init(config json.RawMessage) error {
|
||||
m.meta = map[string]string{"source": m.name, "group": "NFS", "unit": "bytes"}
|
||||
m.tags = map[string]string{"type": "node"}
|
||||
m.config.UseServerAddressAsSType = false
|
||||
// Set default configuration
|
||||
m.config.SendAbsoluteValues = true
|
||||
m.config.SendDerivedValues = false
|
||||
if len(config) > 0 {
|
||||
err = json.Unmarshal(config, &m.config)
|
||||
if err != nil {
|
||||
@@ -110,12 +115,15 @@ func (m *NfsIOStatCollector) Init(config json.RawMessage) error {
|
||||
m.key = "server"
|
||||
}
|
||||
m.data = m.readNfsiostats()
|
||||
m.lastTimestamp = time.Now()
|
||||
m.init = true
|
||||
return err
|
||||
}
|
||||
|
||||
func (m *NfsIOStatCollector) Read(interval time.Duration, output chan lp.CCMessage) {
|
||||
timestamp := time.Now()
|
||||
now := time.Now()
|
||||
timeDiff := now.Sub(m.lastTimestamp).Seconds()
|
||||
m.lastTimestamp = now
|
||||
|
||||
// Get the current values for all mountpoints
|
||||
newdata := m.readNfsiostats()
|
||||
@@ -123,21 +131,30 @@ func (m *NfsIOStatCollector) Read(interval time.Duration, output chan lp.CCMessa
|
||||
for mntpoint, values := range newdata {
|
||||
// Was the mount point already present in the last iteration
|
||||
if old, ok := m.data[mntpoint]; ok {
|
||||
// Calculate the difference of old and new values
|
||||
for i := range values {
|
||||
x := values[i] - old[i]
|
||||
y, err := lp.NewMessage(fmt.Sprintf("nfsio_%s", i), m.tags, m.meta, map[string]interface{}{"value": x}, timestamp)
|
||||
if err == nil {
|
||||
if strings.HasPrefix(i, "page") {
|
||||
y.AddMeta("unit", "4K_Pages")
|
||||
for name, newVal := range values {
|
||||
if m.config.SendAbsoluteValues {
|
||||
msg, err := lp.NewMessage(fmt.Sprintf("nfsio_%s", name), m.tags, m.meta, map[string]interface{}{"value": newVal}, now)
|
||||
if err == nil {
|
||||
msg.AddTag("stype", "filesystem")
|
||||
msg.AddTag("stype-id", mntpoint)
|
||||
output <- msg
|
||||
}
|
||||
y.AddTag("stype", "filesystem")
|
||||
y.AddTag("stype-id", mntpoint)
|
||||
// Send it to output channel
|
||||
output <- y
|
||||
}
|
||||
// Update old to the new value for the next iteration
|
||||
old[i] = values[i]
|
||||
if m.config.SendDerivedValues {
|
||||
rate := float64(newVal-old[name]) / timeDiff
|
||||
msg, err := lp.NewMessage(fmt.Sprintf("nfsio_%s_bw", name), m.tags, m.meta, map[string]interface{}{"value": rate}, now)
|
||||
if err == nil {
|
||||
if strings.HasPrefix(name, "page") {
|
||||
msg.AddMeta("unit", "4K_pages/s")
|
||||
} else {
|
||||
msg.AddMeta("unit", "bytes/sec")
|
||||
}
|
||||
msg.AddTag("stype", "filesystem")
|
||||
msg.AddTag("stype-id", mntpoint)
|
||||
output <- msg
|
||||
}
|
||||
}
|
||||
old[name] = newVal
|
||||
}
|
||||
} else {
|
||||
// First time we see this mount point, store all values
|
||||
@@ -157,7 +174,6 @@ func (m *NfsIOStatCollector) Read(interval time.Duration, output chan lp.CCMessa
|
||||
m.data[mntpoint] = nil
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (m *NfsIOStatCollector) Close() {
|
||||
|
@@ -3,16 +3,18 @@
|
||||
```json
|
||||
"nfsiostat": {
|
||||
"exclude_metrics": [
|
||||
"nfsio_oread"
|
||||
"oread", "pageread"
|
||||
],
|
||||
"exclude_filesystems" : [
|
||||
"/mnt",
|
||||
"exclude_filesystems": [
|
||||
"/mnt"
|
||||
],
|
||||
"use_server_as_stype": false
|
||||
"use_server_as_stype": false,
|
||||
"send_abs_values": false,
|
||||
"send_derived_values": true
|
||||
}
|
||||
```
|
||||
|
||||
The `nfsiostat` collector reads data from `/proc/self/mountstats` and outputs a handful **node** metrics for each NFS filesystem. If a metric or filesystem is not required, it can be excluded from forwarding it to the sink.
|
||||
The `nfsiostat` collector reads data from `/proc/self/mountstats` and outputs a handful **node** metrics for each NFS filesystem. If a metric or filesystem is not required, it can be excluded from forwarding it to the sink. **Note:** When excluding metrics, you must provide the base metric name (e.g. pageread) without the nfsio_ prefix. This exclusion applies to both absolute and derived values.
|
||||
|
||||
Metrics:
|
||||
* `nfsio_nread`: Bytes transferred by normal `read()` calls
|
||||
@@ -24,4 +26,9 @@ Metrics:
|
||||
* `nfsio_nfsread`: Bytes transferred for reading from the server
|
||||
* `nfsio_nfswrite`: Pages transferred by writing to the server
|
||||
|
||||
The `nfsiostat` collector adds the mountpoint to the tags as `stype=filesystem,stype-id=<mountpoint>`. If the server address should be used instead of the mountpoint, use the `use_server_as_stype` config setting.
|
||||
For each of these, if derived values are enabled, an additional metric is sent with the `_bw` suffix, which represents the rate:
|
||||
|
||||
* For normal byte metrics: `unit=bytes/sec`
|
||||
* For page metrics: `unit=4K_pages/s`
|
||||
|
||||
The `nfsiostat` collector adds the mountpoint to the tags as `stype=filesystem,stype-id=<mountpoint>`. If the server address should be used instead of the mountpoint, use the `use_server_as_stype` config setting.
|
||||
|
@@ -10,10 +10,15 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
)
|
||||
|
||||
type NUMAStatsCollectorConfig struct {
|
||||
SendAbsoluteValues bool `json:"send_abs_values"`
|
||||
SendDerivedValues bool `json:"send_derived_values"`
|
||||
}
|
||||
|
||||
// Non-Uniform Memory Access (NUMA) policy hit/miss statistics
|
||||
//
|
||||
// numa_hit:
|
||||
@@ -47,13 +52,16 @@ import (
|
||||
//
|
||||
// See: https://www.kernel.org/doc/html/latest/admin-guide/numastat.html
|
||||
type NUMAStatsCollectorTopolgy struct {
|
||||
file string
|
||||
tagSet map[string]string
|
||||
file string
|
||||
tagSet map[string]string
|
||||
previousValues map[string]int64
|
||||
}
|
||||
|
||||
type NUMAStatsCollector struct {
|
||||
metricCollector
|
||||
topology []NUMAStatsCollectorTopolgy
|
||||
topology []NUMAStatsCollectorTopolgy
|
||||
config NUMAStatsCollectorConfig
|
||||
lastTimestamp time.Time
|
||||
}
|
||||
|
||||
func (m *NUMAStatsCollector) Init(config json.RawMessage) error {
|
||||
@@ -86,8 +94,9 @@ func (m *NUMAStatsCollector) Init(config json.RawMessage) error {
|
||||
file := filepath.Join(dir, "numastat")
|
||||
m.topology = append(m.topology,
|
||||
NUMAStatsCollectorTopolgy{
|
||||
file: file,
|
||||
tagSet: map[string]string{"memoryDomain": node},
|
||||
file: file,
|
||||
tagSet: map[string]string{"memoryDomain": node},
|
||||
previousValues: make(map[string]int64),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -102,23 +111,27 @@ func (m *NUMAStatsCollector) Read(interval time.Duration, output chan lp.CCMessa
|
||||
return
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
timeDiff := now.Sub(m.lastTimestamp).Seconds()
|
||||
m.lastTimestamp = now
|
||||
|
||||
for i := range m.topology {
|
||||
// Loop for all NUMA domains
|
||||
t := &m.topology[i]
|
||||
|
||||
now := time.Now()
|
||||
file, err := os.Open(t.file)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to open file '%s': %v", t.file, err))
|
||||
return
|
||||
continue
|
||||
}
|
||||
scanner := bufio.NewScanner(file)
|
||||
|
||||
// Read line by line
|
||||
for scanner.Scan() {
|
||||
split := strings.Fields(scanner.Text())
|
||||
line := scanner.Text()
|
||||
split := strings.Fields(line)
|
||||
if len(split) != 2 {
|
||||
continue
|
||||
}
|
||||
@@ -130,18 +143,38 @@ func (m *NUMAStatsCollector) Read(interval time.Duration, output chan lp.CCMessa
|
||||
fmt.Sprintf("Read(): Failed to convert %s='%s' to int64: %v", key, split[1], err))
|
||||
continue
|
||||
}
|
||||
y, err := lp.NewMessage(
|
||||
"numastats_"+key,
|
||||
t.tagSet,
|
||||
m.meta,
|
||||
map[string]interface{}{"value": value},
|
||||
now,
|
||||
)
|
||||
if err == nil {
|
||||
output <- y
|
||||
|
||||
if m.config.SendAbsoluteValues {
|
||||
msg, err := lp.NewMessage(
|
||||
"numastats_"+key,
|
||||
t.tagSet,
|
||||
m.meta,
|
||||
map[string]interface{}{"value": value},
|
||||
now,
|
||||
)
|
||||
if err == nil {
|
||||
output <- msg
|
||||
}
|
||||
}
|
||||
|
||||
if m.config.SendDerivedValues {
|
||||
prev, ok := t.previousValues[key]
|
||||
if ok {
|
||||
rate := float64(value-prev) / timeDiff
|
||||
msg, err := lp.NewMessage(
|
||||
"numastats_"+key+"_rate",
|
||||
t.tagSet,
|
||||
m.meta,
|
||||
map[string]interface{}{"value": rate},
|
||||
now,
|
||||
)
|
||||
if err == nil {
|
||||
output <- msg
|
||||
}
|
||||
}
|
||||
t.previousValues[key] = value
|
||||
}
|
||||
}
|
||||
|
||||
file.Close()
|
||||
}
|
||||
}
|
||||
|
@@ -2,7 +2,10 @@
|
||||
## `numastat` collector
|
||||
|
||||
```json
|
||||
"numastats": {}
|
||||
"numastats": {
|
||||
"send_abs_values" : true,
|
||||
"send_derived_values" : true
|
||||
}
|
||||
```
|
||||
|
||||
The `numastat` collector reads data from `/sys/devices/system/node/node*/numastat` and outputs a handful **memoryDomain** metrics. See: <https://www.kernel.org/doc/html/latest/admin-guide/numastat.html>
|
||||
@@ -15,3 +18,9 @@ Metrics:
|
||||
* `numastats_local_node`: A process ran on this node's CPU, and got memory from this node.
|
||||
* `numastats_other_node`: A process ran on a different node's CPU, and got memory from this node.
|
||||
* `numastats_interleave_hit`: Interleaving wanted to allocate from this node and succeeded.
|
||||
* `numastats_numa_hit_rate` (if `send_derived_values == true`): Derived rate value per second.
|
||||
* `numastats_numa_miss_rate` (if `send_derived_values == true`): Derived rate value per second.
|
||||
* `numastats_numa_foreign_rate` (if `send_derived_values == true`): Derived rate value per second.
|
||||
* `numastats_local_node_rate` (if `send_derived_values == true`): Derived rate value per second.
|
||||
* `numastats_other_node_rate` (if `send_derived_values == true`): Derived rate value per second.
|
||||
* `numastats_interleave_hit_rate` (if `send_derived_values == true`): Derived rate value per second.
|
||||
|
@@ -9,20 +9,29 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
)
|
||||
|
||||
// running average power limit (RAPL) monitoring attributes for a zone
|
||||
// Only for Intel systems
|
||||
|
||||
type RAPLZoneInfo struct {
|
||||
energy int64 // current reading of the energy counter in micro joules
|
||||
maxEnergyRange int64 // Range of the above energy counter in micro-joules
|
||||
energyTimestamp time.Time // timestamp when energy counter was read
|
||||
energyFilepath string // path to a file containing the zones current energy counter in micro joules
|
||||
shortTermFilepath string // path to short term power limit
|
||||
longTermFilepath string // path to long term power limit
|
||||
enabledFilepath string // path to check whether limits are enabled
|
||||
name string
|
||||
|
||||
// tags describing the RAPL zone:
|
||||
// * zone_name, subzone_name: e.g. psys, dram, core, uncore, package-0
|
||||
// * zone_id: e.g. 0:1 (zone 0 sub zone 1)
|
||||
tags map[string]string
|
||||
energyFilepath string // path to a file containing the zones current energy counter in micro joules
|
||||
energy int64 // current reading of the energy counter in micro joules
|
||||
energyTimestamp time.Time // timestamp when energy counter was read
|
||||
maxEnergyRange int64 // Range of the above energy counter in micro-joules
|
||||
// type=socket for dram, core, uncore, package-* and type=node for psys
|
||||
// type-id=socket id
|
||||
tags map[string]string
|
||||
}
|
||||
|
||||
type RAPLCollector struct {
|
||||
@@ -33,12 +42,40 @@ type RAPLCollector struct {
|
||||
// * 0:1 for zone 0 subzone 1
|
||||
ExcludeByID []string `json:"exclude_device_by_id,omitempty"`
|
||||
// Exclude names for RAPL zones, e.g. psys, dram, core, uncore, package-0
|
||||
ExcludeByName []string `json:"exclude_device_by_name,omitempty"`
|
||||
ExcludeByName []string `json:"exclude_device_by_name,omitempty"`
|
||||
SkipEnergyReading bool `json:"skip_energy_reading,omitempty"`
|
||||
SkipLimitsReading bool `json:"skip_limits_reading,omitempty"`
|
||||
OnlyEnabledLimits bool `json:"only_enabled_limits,omitempty"`
|
||||
}
|
||||
RAPLZoneInfo []RAPLZoneInfo
|
||||
raplZoneInfo []RAPLZoneInfo
|
||||
meta map[string]string // default meta information
|
||||
}
|
||||
|
||||
// Get the path to the power limit file for zone selectable by limit name
|
||||
// Common limit names for Intel systems are
|
||||
// - long_term
|
||||
// - short_term
|
||||
// Does not support AMD as AMD systems do not provide the power limits
|
||||
// through sysfs
|
||||
func ZoneLimitFile(folder string, limit_name string) string {
|
||||
nameGlob := filepath.Join(folder, "constraint_*_name")
|
||||
candidates, err := filepath.Glob(nameGlob)
|
||||
if err == nil {
|
||||
for _, c := range candidates {
|
||||
if v, err := os.ReadFile(c); err == nil {
|
||||
if strings.TrimSpace(string(v)) == limit_name {
|
||||
var i int
|
||||
n, err := fmt.Sscanf(filepath.Base(c), "constraint_%d_name", &i)
|
||||
if err == nil && n == 1 {
|
||||
return filepath.Join(folder, fmt.Sprintf("constraint_%d_power_limit_uw", i))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// Init initializes the running average power limit (RAPL) collector
|
||||
func (m *RAPLCollector) Init(config json.RawMessage) error {
|
||||
|
||||
@@ -58,6 +95,9 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
|
||||
}
|
||||
|
||||
// Read in the JSON configuration
|
||||
m.config.SkipEnergyReading = false
|
||||
m.config.SkipLimitsReading = false
|
||||
m.config.OnlyEnabledLimits = true
|
||||
if len(config) > 0 {
|
||||
err = json.Unmarshal(config, &m.config)
|
||||
if err != nil {
|
||||
@@ -83,50 +123,62 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
|
||||
// readZoneInfo reads RAPL monitoring attributes for a zone given by zonePath
|
||||
// See: https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#monitoring-attributes
|
||||
readZoneInfo := func(zonePath string) (z struct {
|
||||
name string // zones name e.g. psys, dram, core, uncore, package-0
|
||||
energyFilepath string // path to a file containing the zones current energy counter in micro joules
|
||||
energy int64 // current reading of the energy counter in micro joules
|
||||
energyTimestamp time.Time // timestamp when energy counter was read
|
||||
maxEnergyRange int64 // Range of the above energy counter in micro-joules
|
||||
ok bool // Are all information available?
|
||||
name string // zones name e.g. psys, dram, core, uncore, package-0
|
||||
energyFilepath string // path to a file containing the zones current energy counter in micro joules
|
||||
energy int64 // current reading of the energy counter in micro joules
|
||||
energyTimestamp time.Time // timestamp when energy counter was read
|
||||
maxEnergyRange int64 // Range of the above energy counter in micro-joules
|
||||
shortTermFilepath string
|
||||
longTermFilepath string
|
||||
enabledFilepath string
|
||||
}) {
|
||||
// zones name e.g. psys, dram, core, uncore, package-0
|
||||
foundName := false
|
||||
|
||||
if v, err :=
|
||||
os.ReadFile(
|
||||
filepath.Join(zonePath, "name")); err == nil {
|
||||
foundName = true
|
||||
z.name = strings.TrimSpace(string(v))
|
||||
}
|
||||
|
||||
// path to a file containing the zones current energy counter in micro joules
|
||||
z.energyFilepath = filepath.Join(zonePath, "energy_uj")
|
||||
|
||||
// current reading of the energy counter in micro joules
|
||||
foundEnergy := false
|
||||
if v, err := os.ReadFile(z.energyFilepath); err == nil {
|
||||
// timestamp when energy counter was read
|
||||
z.energyTimestamp = time.Now()
|
||||
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
|
||||
foundEnergy = true
|
||||
z.energy = i
|
||||
if os.Geteuid() == 0 && (!m.config.SkipEnergyReading) {
|
||||
// path to a file containing the zones current energy counter in micro joules
|
||||
z.energyFilepath = filepath.Join(zonePath, "energy_uj")
|
||||
// current reading of the energy counter in micro joules
|
||||
if v, err := os.ReadFile(z.energyFilepath); err == nil {
|
||||
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
|
||||
z.energy = i
|
||||
// timestamp when energy counter was read
|
||||
z.energyTimestamp = time.Now()
|
||||
}
|
||||
} else {
|
||||
cclog.ComponentError(m.name, "Cannot read energy file for ", z.name, ":", err.Error())
|
||||
}
|
||||
// Range of the above energy counter in micro-joules
|
||||
if v, err :=
|
||||
os.ReadFile(
|
||||
filepath.Join(zonePath, "max_energy_range_uj")); err == nil {
|
||||
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
|
||||
z.maxEnergyRange = i
|
||||
}
|
||||
}
|
||||
} else {
|
||||
cclog.ComponentInfo(m.name, "Energy readings for", zonePath, "disabled")
|
||||
}
|
||||
|
||||
// Range of the above energy counter in micro-joules
|
||||
foundMaxEnergyRange := false
|
||||
if v, err :=
|
||||
os.ReadFile(
|
||||
filepath.Join(zonePath, "max_energy_range_uj")); err == nil {
|
||||
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
|
||||
foundMaxEnergyRange = true
|
||||
z.maxEnergyRange = i
|
||||
if !m.config.SkipLimitsReading {
|
||||
z.shortTermFilepath = ZoneLimitFile(zonePath, "short_term")
|
||||
if _, err := os.Stat(z.shortTermFilepath); err != nil {
|
||||
z.shortTermFilepath = ""
|
||||
}
|
||||
z.longTermFilepath = ZoneLimitFile(zonePath, "long_term")
|
||||
if _, err := os.Stat(z.longTermFilepath); err != nil {
|
||||
z.longTermFilepath = ""
|
||||
}
|
||||
z.enabledFilepath = filepath.Join(zonePath, "enabled")
|
||||
} else {
|
||||
cclog.ComponentInfo(m.name, "Power limit readings for", zonePath, "disabled")
|
||||
}
|
||||
|
||||
// Are all information available?
|
||||
z.ok = foundName && foundEnergy && foundMaxEnergyRange
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
@@ -143,25 +195,42 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
|
||||
|
||||
for _, zonePath := range zonesPath {
|
||||
zoneID := strings.TrimPrefix(zonePath, zonePrefix)
|
||||
zonetags := make(map[string]string)
|
||||
|
||||
z := readZoneInfo(zonePath)
|
||||
if z.ok &&
|
||||
!isIDExcluded[zoneID] &&
|
||||
if !isIDExcluded[zoneID] &&
|
||||
!isNameExcluded[z.name] {
|
||||
|
||||
si := RAPLZoneInfo{
|
||||
tags: make(map[string]string),
|
||||
energyFilepath: z.energyFilepath,
|
||||
energy: z.energy,
|
||||
energyTimestamp: z.energyTimestamp,
|
||||
maxEnergyRange: z.maxEnergyRange,
|
||||
shortTermFilepath: z.shortTermFilepath,
|
||||
longTermFilepath: z.longTermFilepath,
|
||||
enabledFilepath: z.enabledFilepath,
|
||||
name: z.name,
|
||||
}
|
||||
si.tags["type"] = "node"
|
||||
si.tags["type-id"] = "0"
|
||||
var pid int = 0
|
||||
if strings.HasPrefix(z.name, "package-") {
|
||||
n, err := fmt.Sscanf(z.name, "package-%d", &pid)
|
||||
if err == nil && n == 1 {
|
||||
si.tags["type-id"] = fmt.Sprintf("%d", pid)
|
||||
si.tags["type"] = "socket"
|
||||
}
|
||||
si.name = "pkg"
|
||||
}
|
||||
// Add RAPL monitoring attributes for a zone
|
||||
m.RAPLZoneInfo =
|
||||
append(
|
||||
m.RAPLZoneInfo,
|
||||
RAPLZoneInfo{
|
||||
tags: map[string]string{
|
||||
"id": zoneID,
|
||||
"zone_name": z.name,
|
||||
},
|
||||
energyFilepath: z.energyFilepath,
|
||||
energy: z.energy,
|
||||
energyTimestamp: z.energyTimestamp,
|
||||
maxEnergyRange: z.maxEnergyRange,
|
||||
})
|
||||
if _, ok1 := si.tags["type"]; ok1 {
|
||||
if _, ok2 := si.tags["type-id"]; ok2 {
|
||||
m.raplZoneInfo = append(m.raplZoneInfo, si)
|
||||
zonetags["type"] = si.tags["type"]
|
||||
zonetags["type-id"] = si.tags["type-id"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// find all sub zones for the given zone
|
||||
@@ -174,29 +243,32 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
|
||||
for _, subZonePath := range subZonesPath {
|
||||
subZoneID := strings.TrimPrefix(subZonePath, subZonePrefix)
|
||||
sz := readZoneInfo(subZonePath)
|
||||
|
||||
if len(zoneID) > 0 && len(z.name) > 0 &&
|
||||
sz.ok &&
|
||||
!isIDExcluded[zoneID+":"+subZoneID] &&
|
||||
!isNameExcluded[sz.name] {
|
||||
m.RAPLZoneInfo =
|
||||
append(
|
||||
m.RAPLZoneInfo,
|
||||
RAPLZoneInfo{
|
||||
tags: map[string]string{
|
||||
"id": zoneID + ":" + subZoneID,
|
||||
"zone_name": z.name,
|
||||
"sub_zone_name": sz.name,
|
||||
},
|
||||
energyFilepath: sz.energyFilepath,
|
||||
energy: sz.energy,
|
||||
energyTimestamp: sz.energyTimestamp,
|
||||
maxEnergyRange: sz.maxEnergyRange,
|
||||
})
|
||||
|
||||
si := RAPLZoneInfo{
|
||||
tags: zonetags,
|
||||
energyFilepath: sz.energyFilepath,
|
||||
energy: sz.energy,
|
||||
energyTimestamp: sz.energyTimestamp,
|
||||
maxEnergyRange: sz.maxEnergyRange,
|
||||
shortTermFilepath: sz.shortTermFilepath,
|
||||
longTermFilepath: sz.longTermFilepath,
|
||||
enabledFilepath: sz.enabledFilepath,
|
||||
name: sz.name,
|
||||
}
|
||||
if _, ok1 := si.tags["type"]; ok1 {
|
||||
if _, ok2 := si.tags["type-id"]; ok2 {
|
||||
m.raplZoneInfo = append(m.raplZoneInfo, si)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if m.RAPLZoneInfo == nil {
|
||||
if m.raplZoneInfo == nil {
|
||||
return fmt.Errorf("no running average power limit (RAPL) device found in %s", controlTypePath)
|
||||
|
||||
}
|
||||
@@ -205,7 +277,7 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
|
||||
cclog.ComponentDebug(
|
||||
m.name,
|
||||
"initialized",
|
||||
len(m.RAPLZoneInfo),
|
||||
len(m.raplZoneInfo),
|
||||
"zones with running average power limit (RAPL) monitoring attributes")
|
||||
m.init = true
|
||||
|
||||
@@ -216,40 +288,89 @@ func (m *RAPLCollector) Init(config json.RawMessage) error {
|
||||
// See: https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#monitoring-attributes
|
||||
func (m *RAPLCollector) Read(interval time.Duration, output chan lp.CCMessage) {
|
||||
|
||||
for i := range m.RAPLZoneInfo {
|
||||
p := &m.RAPLZoneInfo[i]
|
||||
for i := range m.raplZoneInfo {
|
||||
p := &m.raplZoneInfo[i]
|
||||
|
||||
// Read current value of the energy counter in micro joules
|
||||
if v, err := os.ReadFile(p.energyFilepath); err == nil {
|
||||
energyTimestamp := time.Now()
|
||||
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
|
||||
energy := i
|
||||
if os.Geteuid() == 0 && (!m.config.SkipEnergyReading) {
|
||||
// Read current value of the energy counter in micro joules
|
||||
if v, err := os.ReadFile(p.energyFilepath); err == nil {
|
||||
energyTimestamp := time.Now()
|
||||
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
|
||||
energy := i
|
||||
|
||||
// Compute average power (Δ energy / Δ time)
|
||||
energyDiff := energy - p.energy
|
||||
if energyDiff < 0 {
|
||||
// Handle overflow:
|
||||
// ( p.maxEnergyRange - p.energy ) + energy
|
||||
// = p.maxEnergyRange + ( energy - p.energy )
|
||||
// = p.maxEnergyRange + diffEnergy
|
||||
energyDiff += p.maxEnergyRange
|
||||
// Compute average power (Δ energy / Δ time)
|
||||
energyDiff := energy - p.energy
|
||||
if energyDiff < 0 {
|
||||
// Handle overflow:
|
||||
// ( p.maxEnergyRange - p.energy ) + energy
|
||||
// = p.maxEnergyRange + ( energy - p.energy )
|
||||
// = p.maxEnergyRange + diffEnergy
|
||||
energyDiff += p.maxEnergyRange
|
||||
}
|
||||
timeDiff := energyTimestamp.Sub(p.energyTimestamp)
|
||||
averagePower := float64(energyDiff) / float64(timeDiff.Microseconds())
|
||||
|
||||
y, err := lp.NewMetric(
|
||||
fmt.Sprintf("rapl_%s_average_power", p.name),
|
||||
p.tags,
|
||||
m.meta,
|
||||
averagePower,
|
||||
energyTimestamp)
|
||||
if err == nil {
|
||||
output <- y
|
||||
}
|
||||
|
||||
e, err := lp.NewMetric(
|
||||
fmt.Sprintf("rapl_%s_energy", p.name),
|
||||
p.tags,
|
||||
m.meta,
|
||||
float64(energyDiff)*1e-3,
|
||||
energyTimestamp)
|
||||
if err == nil {
|
||||
e.AddMeta("unit", "Joules")
|
||||
output <- e
|
||||
}
|
||||
|
||||
// Save current energy counter state
|
||||
p.energy = energy
|
||||
p.energyTimestamp = energyTimestamp
|
||||
}
|
||||
timeDiff := energyTimestamp.Sub(p.energyTimestamp)
|
||||
averagePower := float64(energyDiff) / float64(timeDiff.Microseconds())
|
||||
|
||||
y, err := lp.NewMessage(
|
||||
"rapl_average_power",
|
||||
p.tags,
|
||||
m.meta,
|
||||
map[string]interface{}{"value": averagePower},
|
||||
energyTimestamp)
|
||||
if err == nil {
|
||||
output <- y
|
||||
}
|
||||
}
|
||||
// https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#constraints
|
||||
if !m.config.SkipLimitsReading {
|
||||
skip := false
|
||||
if m.config.OnlyEnabledLimits {
|
||||
if v, err := os.ReadFile(p.enabledFilepath); err == nil {
|
||||
if strings.TrimSpace(string(v)) == "0" {
|
||||
skip = true
|
||||
}
|
||||
}
|
||||
}
|
||||
if !skip {
|
||||
if len(p.shortTermFilepath) > 0 {
|
||||
if v, err := os.ReadFile(p.shortTermFilepath); err == nil {
|
||||
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
|
||||
name := fmt.Sprintf("rapl_%s_limit_short_term", p.name)
|
||||
y, err := lp.NewMetric(name, p.tags, m.meta, i/1e6, time.Now())
|
||||
if err == nil {
|
||||
output <- y
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Save current energy counter state
|
||||
p.energy = energy
|
||||
p.energyTimestamp = energyTimestamp
|
||||
if len(p.longTermFilepath) > 0 {
|
||||
if v, err := os.ReadFile(p.longTermFilepath); err == nil {
|
||||
if i, err := strconv.ParseInt(strings.TrimSpace(string(v)), 10, 64); err == nil {
|
||||
name := fmt.Sprintf("rapl_%s_limit_long_term", p.name)
|
||||
y, err := lp.NewMetric(name, p.tags, m.meta, i/1e6, time.Now())
|
||||
if err == nil {
|
||||
output <- y
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@@ -1,18 +1,23 @@
|
||||
# Running average power limit (RAPL) metric collector
|
||||
## `rapl` collector
|
||||
|
||||
This collector reads running average power limit (RAPL) monitoring attributes to compute average power consumption metrics. See <https://www.kernel.org/doc/html/latest/power/powercap/powercap.html#monitoring-attributes>.
|
||||
|
||||
The Likwid metric collector provides similar functionality.
|
||||
|
||||
## Configuration
|
||||
This collector reads running average power limit (RAPL) monitoring attributes to compute average power consumption metrics. See <https://www.kernel.org/doc/html/latest/power/powercap/powercap.html>.
|
||||
|
||||
```json
|
||||
"rapl": {
|
||||
"exclude_device_by_id": ["0:1", "0:2"],
|
||||
"exclude_device_by_name": ["psys"]
|
||||
"exclude_device_by_name": ["psys"],
|
||||
"skip_energy_reading": false,
|
||||
"skip_limits_reading": false,
|
||||
"only_enabled_limits": true
|
||||
}
|
||||
```
|
||||
|
||||
## Metrics
|
||||
Metrics:
|
||||
* `rapl_<domain>_average_power`: average power consumption in Watt. The average is computed over the entire runtime from the last measurement to the current measurement
|
||||
* `rapl_<domain>_energy`: Difference from the last measurement
|
||||
* `rapl_<domain>_limit_short_term`: Short term powercap setting for the domain
|
||||
* `rapl_<domain>_limit_long_term`: Long term powercap setting for the domain
|
||||
|
||||
* `rapl_average_power`: average power consumption in Watt. The average is computed over the entire runtime from the last measurement to the current measurement
|
||||
Only the `rapl_<domain>_average_power` and `rapl_<domain>_energy` metrics require root-permissions. The limits can be read as user. Some domains have limits available but they are not enabled. By default, only enabled domain limits are collected.
|
||||
|
||||
Energy and power measurments can also be done with the Likwid metric collector.
|
36
go.mod
36
go.mod
@@ -1,11 +1,12 @@
|
||||
module github.com/ClusterCockpit/cc-metric-collector
|
||||
|
||||
go 1.21.1
|
||||
go 1.23.0
|
||||
|
||||
toolchain go1.22.1
|
||||
toolchain go1.23.2
|
||||
|
||||
require (
|
||||
github.com/ClusterCockpit/cc-energy-manager v0.0.0-20240709142550-dd446f7ab900
|
||||
github.com/ClusterCockpit/cc-lib v0.1.0-beta.1
|
||||
github.com/ClusterCockpit/cc-units v0.4.0
|
||||
github.com/ClusterCockpit/go-rocm-smi v0.3.0
|
||||
github.com/NVIDIA/go-nvml v0.12.0-2
|
||||
@@ -13,33 +14,36 @@ require (
|
||||
github.com/expr-lang/expr v1.16.9
|
||||
github.com/fsnotify/fsnotify v1.7.0
|
||||
github.com/gorilla/mux v1.8.1
|
||||
github.com/influxdata/influxdb-client-go/v2 v2.13.0
|
||||
github.com/influxdata/influxdb-client-go/v2 v2.14.0
|
||||
github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf
|
||||
github.com/influxdata/line-protocol/v2 v2.2.1
|
||||
github.com/nats-io/nats.go v1.36.0
|
||||
github.com/prometheus/client_golang v1.19.0
|
||||
github.com/nats-io/nats.go v1.39.0
|
||||
github.com/prometheus/client_golang v1.20.5
|
||||
github.com/stmcginnis/gofish v0.15.0
|
||||
github.com/tklauser/go-sysconf v0.3.13
|
||||
golang.design/x/thread v0.0.0-20210122121316-335e9adffdf1
|
||||
golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8
|
||||
golang.org/x/sys v0.18.0
|
||||
golang.org/x/exp v0.0.0-20250215185904-eff6e970281f
|
||||
golang.org/x/sys v0.28.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/ClusterCockpit/cc-backend v1.4.2 // indirect
|
||||
github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
|
||||
github.com/beorn7/perks v1.0.1 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.2.0 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/google/uuid v1.6.0 // indirect
|
||||
github.com/klauspost/compress v1.17.7 // indirect
|
||||
github.com/nats-io/nkeys v0.4.7 // indirect
|
||||
github.com/klauspost/compress v1.17.9 // indirect
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
|
||||
github.com/nats-io/nkeys v0.4.9 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
github.com/oapi-codegen/runtime v1.1.1 // indirect
|
||||
github.com/prometheus/client_model v0.6.0 // indirect
|
||||
github.com/prometheus/common v0.49.0 // indirect
|
||||
github.com/prometheus/procfs v0.12.0 // indirect
|
||||
github.com/prometheus/client_model v0.6.1 // indirect
|
||||
github.com/prometheus/common v0.55.0 // indirect
|
||||
github.com/prometheus/procfs v0.15.1 // indirect
|
||||
github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 // indirect
|
||||
github.com/shopspring/decimal v1.3.1 // indirect
|
||||
github.com/tklauser/numcpus v0.7.0 // indirect
|
||||
golang.org/x/crypto v0.21.0 // indirect
|
||||
golang.org/x/net v0.22.0 // indirect
|
||||
google.golang.org/protobuf v1.33.0 // indirect
|
||||
golang.org/x/crypto v0.31.0 // indirect
|
||||
golang.org/x/net v0.31.0 // indirect
|
||||
google.golang.org/protobuf v1.35.2 // indirect
|
||||
)
|
||||
|
69
go.sum
69
go.sum
@@ -1,5 +1,9 @@
|
||||
github.com/ClusterCockpit/cc-backend v1.4.2 h1:kTOzqkh9N0564N9nqQThnSs7TAfg8RLgvSm00e5HtIc=
|
||||
github.com/ClusterCockpit/cc-backend v1.4.2/go.mod h1:g8TNHXe4AXej26snu2//jO3mUF980elT93iV/k11O/c=
|
||||
github.com/ClusterCockpit/cc-energy-manager v0.0.0-20240709142550-dd446f7ab900 h1:6+WNav16uWTEDC09hkZKEHfBhtc91p/ZcjgCtyntuIg=
|
||||
github.com/ClusterCockpit/cc-energy-manager v0.0.0-20240709142550-dd446f7ab900/go.mod h1:EbYeC5t+Y0kW1Q1pP2n9zMqbeYEJITG8YGvAUihXVn4=
|
||||
github.com/ClusterCockpit/cc-lib v0.1.0-beta.1 h1:dz9j0g2cod8+SMDjuoIY6ISpiHHeekhX6yQaeiwiwJw=
|
||||
github.com/ClusterCockpit/cc-lib v0.1.0-beta.1/go.mod h1:kXMskla1i5ZSfXW0vVRIHgGeXMU5zu2PzYOYnUaOr80=
|
||||
github.com/ClusterCockpit/cc-units v0.4.0 h1:zP5DOu99GmErW0tCDf0gcLrlWt42RQ9dpoONEOh4cI0=
|
||||
github.com/ClusterCockpit/cc-units v0.4.0/go.mod h1:3S3PAhAayS3pbgcT4q9Vn9VJw22Op51X0YimtG77zBw=
|
||||
github.com/ClusterCockpit/go-rocm-smi v0.3.0 h1:1qZnSpG7/NyLtc7AjqnUL9Jb8xtqG1nMVgp69rJfaR8=
|
||||
@@ -17,8 +21,8 @@ github.com/apapsch/go-jsonmerge/v2 v2.0.0/go.mod h1:lvDnEdqiQrp0O42VQGgmlKpxL1AP
|
||||
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
|
||||
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
|
||||
github.com/bmatcuk/doublestar v1.1.1/go.mod h1:UD6OnuiIn0yFxxA2le/rnRU1G4RaI4UvFv1sNto9p6w=
|
||||
github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
|
||||
github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
@@ -39,8 +43,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY=
|
||||
github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ=
|
||||
github.com/influxdata/influxdb-client-go/v2 v2.13.0 h1:ioBbLmR5NMbAjP4UVA5r9b5xGjpABD7j65pI8kFphDM=
|
||||
github.com/influxdata/influxdb-client-go/v2 v2.13.0/go.mod h1:k+spCbt9hcvqvUiz0sr5D8LolXHqAAOfPw9v/RIRHl4=
|
||||
github.com/influxdata/influxdb-client-go/v2 v2.14.0 h1:AjbBfJuq+QoaXNcrova8smSjwJdUHnwvfjMF71M1iI4=
|
||||
github.com/influxdata/influxdb-client-go/v2 v2.14.0/go.mod h1:Ahpm3QXKMJslpXl3IftVLVezreAUtBOTZssDrjZEFHI=
|
||||
github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf h1:7JTmneyiNEwVBOHSjoMxiWAqB992atOeepeFYegn5RU=
|
||||
github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo=
|
||||
github.com/influxdata/line-protocol-corpus v0.0.0-20210519164801-ca6fa5da0184/go.mod h1:03nmhxzZ7Xk2pdG+lmMd7mHDfeVOYFyhOgwO61qWU98=
|
||||
@@ -51,8 +55,8 @@ github.com/influxdata/line-protocol/v2 v2.1.0/go.mod h1:QKw43hdUBg3GTk2iC3iyCxks
|
||||
github.com/influxdata/line-protocol/v2 v2.2.1 h1:EAPkqJ9Km4uAxtMRgUubJyqAr6zgWM0dznKMLRauQRE=
|
||||
github.com/influxdata/line-protocol/v2 v2.2.1/go.mod h1:DmB3Cnh+3oxmG6LOBIxce4oaL4CPj3OmMPgvauXh+tM=
|
||||
github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE=
|
||||
github.com/klauspost/compress v1.17.7 h1:ehO88t2UGzQK66LMdE8tibEd1ErmzZjNEqWkjLAKQQg=
|
||||
github.com/klauspost/compress v1.17.7/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
|
||||
github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
|
||||
github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
|
||||
github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
|
||||
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
|
||||
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
|
||||
@@ -60,10 +64,14 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
||||
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
|
||||
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
|
||||
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
|
||||
github.com/nats-io/nats.go v1.36.0 h1:suEUPuWzTSse/XhESwqLxXGuj8vGRuPRoG7MoRN/qyU=
|
||||
github.com/nats-io/nats.go v1.36.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
|
||||
github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
|
||||
github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
|
||||
github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
|
||||
github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
|
||||
github.com/nats-io/nats.go v1.39.0 h1:2/yg2JQjiYYKLwDuBzV0FbB2sIV+eFNkEevlRi4n9lI=
|
||||
github.com/nats-io/nats.go v1.39.0/go.mod h1:MgRb8oOdigA6cYpEPhXJuRVH6UE/V4jblJ2jQ27IXYM=
|
||||
github.com/nats-io/nkeys v0.4.9 h1:qe9Faq2Gxwi6RZnZMXfmGMZkg3afLLOtrU+gDZJ35b0=
|
||||
github.com/nats-io/nkeys v0.4.9/go.mod h1:jcMqs+FLG+W5YO36OX6wFIFcmpdAns+w1Wm6D3I/evE=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||
github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
|
||||
@@ -71,16 +79,18 @@ github.com/oapi-codegen/runtime v1.1.1 h1:EXLHh0DXIJnWhdRPN2w4MXAzFyE4CskzhNLUmt
|
||||
github.com/oapi-codegen/runtime v1.1.1/go.mod h1:SK9X900oXmPWilYR5/WKPzt3Kqxn/uS/+lbpREv+eCg=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/prometheus/client_golang v1.19.0 h1:ygXvpU1AoN1MhdzckN+PyD9QJOSD4x7kmXYlnfbA6JU=
|
||||
github.com/prometheus/client_golang v1.19.0/go.mod h1:ZRM9uEAypZakd+q/x7+gmsvXdURP+DABIEIjnmDdp+k=
|
||||
github.com/prometheus/client_model v0.6.0 h1:k1v3CzpSRUTrKMppY35TLwPvxHqBu0bYgxZzqGIgaos=
|
||||
github.com/prometheus/client_model v0.6.0/go.mod h1:NTQHnmxFpouOD0DpvP4XujX3CdOAGQPoaGhyTchlyt8=
|
||||
github.com/prometheus/common v0.49.0 h1:ToNTdK4zSnPVJmh698mGFkDor9wBI/iGaJy5dbH1EgI=
|
||||
github.com/prometheus/common v0.49.0/go.mod h1:Kxm+EULxRbUkjGU6WFsQqo3ORzB4tyKvlWFOE9mB2sE=
|
||||
github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo=
|
||||
github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
|
||||
github.com/prometheus/client_golang v1.20.5 h1:cxppBPuYhUnsO6yo/aoRol4L7q7UFfdm+bR9r+8l63Y=
|
||||
github.com/prometheus/client_golang v1.20.5/go.mod h1:PIEt8X02hGcP8JWbeHyeZ53Y/jReSnHgO035n//V5WE=
|
||||
github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E=
|
||||
github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY=
|
||||
github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G1dc=
|
||||
github.com/prometheus/common v0.55.0/go.mod h1:2SECS4xJG1kd8XF9IcM1gMX6510RAEL65zxzNImwdc8=
|
||||
github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
|
||||
github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
|
||||
github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
|
||||
github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
|
||||
github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 h1:lZUw3E0/J3roVtGQ+SCrUrg3ON6NgVqpn3+iol9aGu4=
|
||||
github.com/santhosh-tekuri/jsonschema/v5 v5.3.1/go.mod h1:uToXkOrWAZ6/Oc07xWQrPOhJotwFIyu2bBVN41fcDUY=
|
||||
github.com/shopspring/decimal v1.3.1 h1:2Usl1nmF/WZucqkFZhnfFYxxxu8LG21F6nPQBE5gKV8=
|
||||
github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
|
||||
github.com/spkg/bom v0.0.0-20160624110644-59b7046e48ad/go.mod h1:qLr4V1qq6nMqFKkMo8ZTx3f+BZEkzsRUY10Xsm2mwU0=
|
||||
@@ -92,26 +102,27 @@ github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpE
|
||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
|
||||
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
|
||||
github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
|
||||
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
|
||||
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
|
||||
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
|
||||
github.com/tklauser/go-sysconf v0.3.13 h1:GBUpcahXSpR2xN01jhkNAbTLRk2Yzgggk8IM08lq3r4=
|
||||
github.com/tklauser/go-sysconf v0.3.13/go.mod h1:zwleP4Q4OehZHGn4CYZDipCgg9usW5IJePewFCGVEa0=
|
||||
github.com/tklauser/numcpus v0.7.0 h1:yjuerZP127QG9m5Zh/mSO4wqurYil27tHrqwRoRjpr4=
|
||||
github.com/tklauser/numcpus v0.7.0/go.mod h1:bb6dMVcj8A42tSE7i32fsIUCbQNllK5iDguyOZRUzAY=
|
||||
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.21.0 h1:X31++rzVUdKhX5sWmSOFZxx8UW/ldWx55cbf08iNAMA=
|
||||
golang.org/x/crypto v0.21.0/go.mod h1:0BP7YvVV9gBbVKyeTG0Gyn+gZm94bibOW5BjDEYAOMs=
|
||||
golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8 h1:yixxcjnhBmY0nkL253HFVIm0JsFHwrHdT3Yh6szTnfY=
|
||||
golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8/go.mod h1:jj3sYF3dwk5D+ghuXyeI3r5MFf+NT2An6/9dOA95KSI=
|
||||
golang.org/x/net v0.22.0 h1:9sGLhx7iRIHEiX0oAJ3MRZMUCElJgy7Br1nO+AMN3Tc=
|
||||
golang.org/x/net v0.22.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
|
||||
golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
|
||||
golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
|
||||
golang.org/x/exp v0.0.0-20250215185904-eff6e970281f h1:oFMYAjX0867ZD2jcNiLBrI9BdpmEkvPyi5YrBGXbamg=
|
||||
golang.org/x/exp v0.0.0-20250215185904-eff6e970281f/go.mod h1:BHOTPb3L19zxehTsLoJXVaTktb06DFgmdW6Wb9s8jqk=
|
||||
golang.org/x/net v0.31.0 h1:68CPQngjLL0r2AlUKiSxtQFKvzRVbnzLwMUn5SzcLHo=
|
||||
golang.org/x/net v0.31.0/go.mod h1:P4fl1q7dY2hnZFxEk4pPSkDHF+QqjitcnDjUQyMM+pM=
|
||||
golang.org/x/sys v0.0.0-20210122093101-04d7465088b8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=
|
||||
golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
|
||||
golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
|
||||
google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
|
||||
google.golang.org/protobuf v1.35.2 h1:8Ar7bF+apOIoThw1EdZl0p1oWvMqTHmpA2fRTyZO8io=
|
||||
google.golang.org/protobuf v1.35.2/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
|
@@ -40,7 +40,7 @@ type metricRouterConfig struct {
|
||||
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
|
||||
// dropMetrics map[string]bool // Internal map for O(1) lookup
|
||||
MessageProcessor json.RawMessage `json:"process_message,omitempty"`
|
||||
MessageProcessor json.RawMessage `json:"process_messages,omitempty"`
|
||||
}
|
||||
|
||||
// Metric router data structure
|
||||
|
@@ -22,27 +22,27 @@ type messageProcessorTagConfig struct {
|
||||
}
|
||||
|
||||
type messageProcessorConfig struct {
|
||||
StageOrder []string `json:"stage_order,omitempty"` // List of stages to execute them in the specified order and to skip unrequired ones
|
||||
DropMessages []string `json:"drop_messages,omitempty"` // List of metric names to drop. For fine-grained dropping use drop_messages_if
|
||||
DropMessagesIf []string `json:"drop_messages_if,omitempty"` // List of evaluatable terms to drop messages
|
||||
RenameMessages map[string]string `json:"rename_messages,omitempty"` // Map of metric names to rename
|
||||
RenameMessagesIf map[string]string `json:"rename_messages_if,omitempty"` // Map to rename metric name based on a condition
|
||||
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 messages
|
||||
AddTagsIf []messageProcessorTagConfig `json:"add_tags_if"` // List of tags that are added when the condition is met
|
||||
DelTagsIf []messageProcessorTagConfig `json:"delete_tags_if"` // List of tags that are removed when the condition is met
|
||||
AddMetaIf []messageProcessorTagConfig `json:"add_meta_if"` // List of meta infos that are added when the condition is met
|
||||
DelMetaIf []messageProcessorTagConfig `json:"delete_meta_if"` // List of meta infos that are removed when the condition is met
|
||||
AddFieldIf []messageProcessorTagConfig `json:"add_field_if"` // List of fields that are added when the condition is met
|
||||
DelFieldIf []messageProcessorTagConfig `json:"delete_field_if"` // List of fields that are removed when the condition is met
|
||||
DropByType []string `json:"drop_by_message_type"` // List of message types that should be dropped
|
||||
MoveTagToMeta []messageProcessorTagConfig `json:"move_tag_to_meta_if"`
|
||||
MoveTagToField []messageProcessorTagConfig `json:"move_tag_to_field_if"`
|
||||
MoveMetaToTag []messageProcessorTagConfig `json:"move_meta_to_tag_if"`
|
||||
MoveMetaToField []messageProcessorTagConfig `json:"move_meta_to_field_if"`
|
||||
MoveFieldToTag []messageProcessorTagConfig `json:"move_field_to_tag_if"`
|
||||
MoveFieldToMeta []messageProcessorTagConfig `json:"move_field_to_meta_if"`
|
||||
AddBaseEnv map[string]interface{} `json:"add_base_env"`
|
||||
StageOrder []string `json:"stage_order,omitempty"` // List of stages to execute them in the specified order and to skip unrequired ones
|
||||
DropMessages []string `json:"drop_messages,omitempty"` // List of metric names to drop. For fine-grained dropping use drop_messages_if
|
||||
DropMessagesIf []string `json:"drop_messages_if,omitempty"` // List of evaluatable terms to drop messages
|
||||
RenameMessages map[string]string `json:"rename_messages,omitempty"` // Map of metric names to rename
|
||||
RenameMessagesIf map[string]string `json:"rename_messages_if,omitempty"` // Map to rename metric name based on a condition
|
||||
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 messages
|
||||
AddTagsIf []messageProcessorTagConfig `json:"add_tags_if,omitempty"` // List of tags that are added when the condition is met
|
||||
DelTagsIf []messageProcessorTagConfig `json:"delete_tags_if,omitempty"` // List of tags that are removed when the condition is met
|
||||
AddMetaIf []messageProcessorTagConfig `json:"add_meta_if,omitempty"` // List of meta infos that are added when the condition is met
|
||||
DelMetaIf []messageProcessorTagConfig `json:"delete_meta_if,omitempty"` // List of meta infos that are removed when the condition is met
|
||||
AddFieldIf []messageProcessorTagConfig `json:"add_field_if,omitempty"` // List of fields that are added when the condition is met
|
||||
DelFieldIf []messageProcessorTagConfig `json:"delete_field_if,omitempty"` // List of fields that are removed when the condition is met
|
||||
DropByType []string `json:"drop_by_message_type,omitempty"` // List of message types that should be dropped
|
||||
MoveTagToMeta []messageProcessorTagConfig `json:"move_tag_to_meta_if,omitempty"`
|
||||
MoveTagToField []messageProcessorTagConfig `json:"move_tag_to_field_if,omitempty"`
|
||||
MoveMetaToTag []messageProcessorTagConfig `json:"move_meta_to_tag_if,omitempty"`
|
||||
MoveMetaToField []messageProcessorTagConfig `json:"move_meta_to_field_if,omitempty"`
|
||||
MoveFieldToTag []messageProcessorTagConfig `json:"move_field_to_tag_if,omitempty"`
|
||||
MoveFieldToMeta []messageProcessorTagConfig `json:"move_field_to_meta_if,omitempty"`
|
||||
AddBaseEnv map[string]interface{} `json:"add_base_env,omitempty"`
|
||||
}
|
||||
|
||||
type messageProcessor struct {
|
||||
|
@@ -13,7 +13,6 @@ import (
|
||||
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
mp "github.com/ClusterCockpit/cc-metric-collector/pkg/messageProcessor"
|
||||
influx "github.com/influxdata/line-protocol/v2/lineprotocol"
|
||||
)
|
||||
|
||||
const HTTP_RECEIVER_PORT = "8080"
|
||||
@@ -151,80 +150,22 @@ func (r *HttpReceiver) ServerHttp(w http.ResponseWriter, req *http.Request) {
|
||||
}
|
||||
}
|
||||
if r.sink != nil {
|
||||
d := influx.NewDecoder(req.Body)
|
||||
for d.Next() {
|
||||
|
||||
// Decode measurement name
|
||||
measurement, err := d.Measurement()
|
||||
buf := make([]byte, 0, req.ContentLength)
|
||||
len, err := req.Body.Read(buf)
|
||||
if err == nil && len > 0 {
|
||||
messages, err := lp.FromBytes(buf)
|
||||
if err != nil {
|
||||
msg := "ServerHttp: Failed to decode measurement: " + err.Error()
|
||||
msg := "ServerHttp: Failed to decode messages: " + err.Error()
|
||||
cclog.ComponentError(r.name, msg)
|
||||
http.Error(w, msg, http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
// Decode tags
|
||||
tags := make(map[string]string)
|
||||
for {
|
||||
key, value, err := d.NextTag()
|
||||
if err != nil {
|
||||
msg := "ServerHttp: Failed to decode tag: " + err.Error()
|
||||
cclog.ComponentError(r.name, msg)
|
||||
http.Error(w, msg, http.StatusInternalServerError)
|
||||
return
|
||||
for _, y := range messages {
|
||||
m, err := r.mp.ProcessMessage(y)
|
||||
if err == nil && m != nil {
|
||||
r.sink <- m
|
||||
}
|
||||
if key == nil {
|
||||
break
|
||||
}
|
||||
tags[string(key)] = string(value)
|
||||
}
|
||||
|
||||
// Decode fields
|
||||
fields := make(map[string]interface{})
|
||||
for {
|
||||
key, value, err := d.NextField()
|
||||
if err != nil {
|
||||
msg := "ServerHttp: Failed to decode field: " + err.Error()
|
||||
cclog.ComponentError(r.name, msg)
|
||||
http.Error(w, msg, http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
if key == nil {
|
||||
break
|
||||
}
|
||||
fields[string(key)] = value.Interface()
|
||||
}
|
||||
|
||||
// Decode time stamp
|
||||
t, err := d.Time(influx.Nanosecond, time.Time{})
|
||||
if err != nil {
|
||||
msg := "ServerHttp: Failed to decode time stamp: " + err.Error()
|
||||
cclog.ComponentError(r.name, msg)
|
||||
http.Error(w, msg, http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
y, _ := lp.NewMessage(
|
||||
string(measurement),
|
||||
tags,
|
||||
nil,
|
||||
fields,
|
||||
t,
|
||||
)
|
||||
|
||||
m, err := r.mp.ProcessMessage(y)
|
||||
if err == nil && m != nil {
|
||||
r.sink <- m
|
||||
}
|
||||
|
||||
}
|
||||
// Check for IO errors
|
||||
err := d.Err()
|
||||
if err != nil {
|
||||
msg := "ServerHttp: Failed to decode: " + err.Error()
|
||||
cclog.ComponentError(r.name, msg)
|
||||
http.Error(w, msg, http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
|
@@ -5,20 +5,18 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
|
||||
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
||||
mp "github.com/ClusterCockpit/cc-metric-collector/pkg/messageProcessor"
|
||||
influx "github.com/influxdata/line-protocol/v2/lineprotocol"
|
||||
nats "github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
type NatsReceiverConfig struct {
|
||||
defaultReceiverConfig
|
||||
Addr string `json:"address"`
|
||||
Port string `json:"port"`
|
||||
Subject string `json:"subject"`
|
||||
Addr string `json:"address"`
|
||||
Port string `json:"port"`
|
||||
Subject string `json:"subject"`
|
||||
User string `json:"user,omitempty"`
|
||||
Password string `json:"password,omitempty"`
|
||||
NkeyFile string `json:"nkey_file,omitempty"`
|
||||
@@ -42,65 +40,14 @@ func (r *NatsReceiver) Start() {
|
||||
func (r *NatsReceiver) _NatsReceive(m *nats.Msg) {
|
||||
|
||||
if r.sink != nil {
|
||||
d := influx.NewDecoderWithBytes(m.Data)
|
||||
for d.Next() {
|
||||
|
||||
// Decode measurement name
|
||||
measurement, err := d.Measurement()
|
||||
if err != nil {
|
||||
msg := "_NatsReceive: Failed to decode measurement: " + err.Error()
|
||||
cclog.ComponentError(r.name, msg)
|
||||
return
|
||||
}
|
||||
|
||||
// Decode tags
|
||||
tags := make(map[string]string)
|
||||
for {
|
||||
key, value, err := d.NextTag()
|
||||
if err != nil {
|
||||
msg := "_NatsReceive: Failed to decode tag: " + err.Error()
|
||||
cclog.ComponentError(r.name, msg)
|
||||
return
|
||||
}
|
||||
if key == nil {
|
||||
break
|
||||
}
|
||||
tags[string(key)] = string(value)
|
||||
}
|
||||
|
||||
// Decode fields
|
||||
fields := make(map[string]interface{})
|
||||
for {
|
||||
key, value, err := d.NextField()
|
||||
if err != nil {
|
||||
msg := "_NatsReceive: Failed to decode field: " + err.Error()
|
||||
cclog.ComponentError(r.name, msg)
|
||||
return
|
||||
}
|
||||
if key == nil {
|
||||
break
|
||||
}
|
||||
fields[string(key)] = value.Interface()
|
||||
}
|
||||
|
||||
// Decode time stamp
|
||||
t, err := d.Time(influx.Nanosecond, time.Time{})
|
||||
if err != nil {
|
||||
msg := "_NatsReceive: Failed to decode time: " + err.Error()
|
||||
cclog.ComponentError(r.name, msg)
|
||||
return
|
||||
}
|
||||
|
||||
y, _ := lp.NewMessage(
|
||||
string(measurement),
|
||||
tags,
|
||||
nil,
|
||||
fields,
|
||||
t,
|
||||
)
|
||||
|
||||
messages, err := lp.FromBytes(m.Data)
|
||||
if err != nil {
|
||||
msg := "_NatsReceive: Failed to decode messages: " + err.Error()
|
||||
cclog.ComponentError(r.name, msg)
|
||||
}
|
||||
for _, y := range messages {
|
||||
m, err := r.mp.ProcessMessage(y)
|
||||
if err == nil && m != nil {
|
||||
if err == nil && m != nil && r.sink != nil {
|
||||
r.sink <- m
|
||||
}
|
||||
}
|
||||
|
175
scripts/generate_docs.sh
Executable file
175
scripts/generate_docs.sh
Executable file
@@ -0,0 +1,175 @@
|
||||
#!/bin/bash -l
|
||||
|
||||
SRCDIR="$(pwd)"
|
||||
DESTDIR="$1"
|
||||
|
||||
if [ -z "$DESTDIR" ]; then
|
||||
echo "Destination folder not provided"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
|
||||
COLLECTORS=$(find "${SRCDIR}/collectors" -name "*Metric.md")
|
||||
SINKS=$(find "${SRCDIR}/sinks" -name "*Sink.md")
|
||||
RECEIVERS=$(find "${SRCDIR}/receivers" -name "*Receiver.md")
|
||||
|
||||
|
||||
|
||||
# Collectors
|
||||
mkdir -p "${DESTDIR}/collectors"
|
||||
for F in $COLLECTORS; do
|
||||
echo "$F"
|
||||
FNAME=$(basename "$F")
|
||||
TITLE=$(grep -E "^##" "$F" | head -n 1 | sed -e 's+## ++g')
|
||||
echo "'${TITLE//\`/}'"
|
||||
if [ "${TITLE}" == "" ]; then continue; fi
|
||||
rm --force "${DESTDIR}/collectors/${FNAME}"
|
||||
cat << EOF >> "${DESTDIR}/collectors/${FNAME}"
|
||||
---
|
||||
title: ${TITLE//\`/}
|
||||
description: >
|
||||
Toplevel ${FNAME/.md/}
|
||||
categories: [cc-metric-collector]
|
||||
tags: [cc-metric-collector, Collector, ${FNAME/Metric.md/}]
|
||||
weight: 2
|
||||
---
|
||||
|
||||
EOF
|
||||
cat "$F" >> "${DESTDIR}/collectors/${FNAME}"
|
||||
done
|
||||
|
||||
if [ -e "${SRCDIR}/collectors/README.md" ]; then
|
||||
cat << EOF > "${DESTDIR}/collectors/_index.md"
|
||||
---
|
||||
title: cc-metric-collector's collectors
|
||||
description: Documentation of cc-metric-collector's collectors
|
||||
categories: [cc-metric-collector]
|
||||
tags: [cc-metric-collector, Collector, General]
|
||||
weight: 40
|
||||
---
|
||||
|
||||
EOF
|
||||
cat "${SRCDIR}/collectors/README.md" >> "${DESTDIR}/collectors/_index.md"
|
||||
fi
|
||||
|
||||
# Sinks
|
||||
mkdir -p "${DESTDIR}/sinks"
|
||||
for F in $SINKS; do
|
||||
echo "$F"
|
||||
FNAME=$(basename "$F")
|
||||
TITLE=$(grep -E "^##" "$F" | head -n 1 | sed -e 's+## ++g')
|
||||
echo "'${TITLE//\`/}'"
|
||||
if [ "${TITLE}" == "" ]; then continue; fi
|
||||
rm --force "${DESTDIR}/sinks/${FNAME}"
|
||||
cat << EOF >> "${DESTDIR}/sinks/${FNAME}"
|
||||
---
|
||||
title: ${TITLE//\`/}
|
||||
description: >
|
||||
Toplevel ${FNAME/.md/}
|
||||
categories: [cc-metric-collector]
|
||||
tags: [cc-metric-collector, Sink, ${FNAME/Sink.md/}]
|
||||
weight: 2
|
||||
---
|
||||
|
||||
EOF
|
||||
cat "$F" >> "${DESTDIR}/sinks/${FNAME}"
|
||||
done
|
||||
|
||||
if [ -e "${SRCDIR}/collectors/README.md" ]; then
|
||||
cat << EOF > "${DESTDIR}/sinks/_index.md"
|
||||
---
|
||||
title: cc-metric-collector's sinks
|
||||
description: Documentation of cc-metric-collector's sinks
|
||||
categories: [cc-metric-collector]
|
||||
tags: [cc-metric-collector, Sink, General]
|
||||
weight: 40
|
||||
---
|
||||
|
||||
EOF
|
||||
cat "${SRCDIR}/sinks/README.md" >> "${DESTDIR}/sinks/_index.md"
|
||||
fi
|
||||
|
||||
|
||||
# Receivers
|
||||
mkdir -p "${DESTDIR}/receivers"
|
||||
for F in $RECEIVERS; do
|
||||
echo "$F"
|
||||
FNAME=$(basename "$F")
|
||||
TITLE=$(grep -E "^##" "$F" | head -n 1 | sed -e 's+## ++g')
|
||||
echo "'${TITLE//\`/}'"
|
||||
if [ "${TITLE}" == "" ]; then continue; fi
|
||||
rm --force "${DESTDIR}/receivers/${FNAME}"
|
||||
cat << EOF >> "${DESTDIR}/receivers/${FNAME}"
|
||||
---
|
||||
title: ${TITLE//\`/}
|
||||
description: >
|
||||
Toplevel ${FNAME/.md/}
|
||||
categories: [cc-metric-collector]
|
||||
tags: [cc-metric-collector, Receiver, ${FNAME/Receiver.md/}]
|
||||
weight: 2
|
||||
---
|
||||
|
||||
EOF
|
||||
cat "$F" >> "${DESTDIR}/receivers/${FNAME}"
|
||||
done
|
||||
|
||||
if [ -e "${SRCDIR}/receivers/README.md" ]; then
|
||||
cat << EOF > "${DESTDIR}/receivers/_index.md"
|
||||
---
|
||||
title: cc-metric-collector's receivers
|
||||
description: Documentation of cc-metric-collector's receivers
|
||||
categories: [cc-metric-collector]
|
||||
tags: [cc-metric-collector, Receiver, General]
|
||||
weight: 40
|
||||
---
|
||||
|
||||
EOF
|
||||
cat "${SRCDIR}/receivers/README.md" >> "${DESTDIR}/receivers/_index.md"
|
||||
fi
|
||||
|
||||
mkdir -p "${DESTDIR}/internal/metricRouter"
|
||||
if [ -e "${SRCDIR}/internal/metricRouter/README.md" ]; then
|
||||
cat << EOF > "${DESTDIR}/internal/metricRouter/_index.md"
|
||||
---
|
||||
title: cc-metric-collector's router
|
||||
description: Documentation of cc-metric-collector's router
|
||||
categories: [cc-metric-collector]
|
||||
tags: [cc-metric-collector, Router, General]
|
||||
weight: 40
|
||||
---
|
||||
|
||||
EOF
|
||||
cat "${SRCDIR}/internal/metricRouter/README.md" >> "${DESTDIR}/internal/metricRouter/_index.md"
|
||||
fi
|
||||
|
||||
if [ -e "${SRCDIR}/README.md" ]; then
|
||||
cat << EOF > "${DESTDIR}/_index.md"
|
||||
---
|
||||
title: cc-metric-collector
|
||||
description: Documentation of cc-metric-collector
|
||||
categories: [cc-metric-collector]
|
||||
tags: [cc-metric-collector, General]
|
||||
weight: 40
|
||||
---
|
||||
|
||||
EOF
|
||||
cat "${SRCDIR}/README.md" >> "${DESTDIR}/_index.md"
|
||||
sed -i -e 's+README.md+_index.md+g' "${DESTDIR}/_index.md"
|
||||
fi
|
||||
|
||||
|
||||
mkdir -p "${DESTDIR}/pkg/messageProcessor"
|
||||
if [ -e "${SRCDIR}/pkg/messageProcessor/README.md" ]; then
|
||||
cat << EOF > "${DESTDIR}/pkg/messageProcessor/_index.md"
|
||||
---
|
||||
title: cc-metric-collector's message processor
|
||||
description: Documentation of cc-metric-collector's message processor
|
||||
categories: [cc-metric-collector]
|
||||
tags: [cc-metric-collector, Message Processor]
|
||||
weight: 40
|
||||
---
|
||||
|
||||
EOF
|
||||
cat "${SRCDIR}/pkg/messageProcessor/README.md" >> "${DESTDIR}/pkg/messageProcessor/_index.md"
|
||||
fi
|
||||
|
Reference in New Issue
Block a user