ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2024-12-25 15:09:05 +01:00
Code Issues Packages Projects Releases Wiki Activity

Formatting

Browse Source
This commit is contained in:
Thomas Roehl 2021-11-25 15:11:39 +01:00
parent a88ea4fd43
commit 9232dd9732
15 changed files with 577 additions and 566 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
collectors/Makefile
View File

@ -1,14 +1,20 @@
# Use central installation
CENTRAL_INSTALL = false
CENTRAL_INSTALL = true
# Path to central installation (if CENTRAL_INSTALL=true)
LIKWID_BASE=/usr/local
LIKWID_BASE=/apps/likwid/5.2.0
# LIKWID version (should be same major version as central installation, 5.1.x)
LIKWID_VERSION = 5.1.0
# Target user for LIKWID's accessdaemon (if CENTRAL_INSTALL=false)
DAEMON_USER=root
# Target group for LIKWID's accessdaemon (if CENTRAL_INSTALL=false)
DAEMON_GROUP=root
LIKWID_VERSION = 5.2.0
#######################################################################
# if CENTRAL_INSTALL=false
#######################################################################
# Target user for LIKWID's accessdaemon
DAEMON_USER=unrz139
# Target group for LIKWID's accessdaemon
DAEMON_GROUP=unrz
# Folder for the group files
GROUP_FOLDER=/usr/local/share/likwid/perfgroups/
#################################################
@ -21,7 +27,7 @@ ifneq ($(strip $(CENTRAL_INSTALL)),true)
LIKWID_BASE := $(shell pwd)/$(INSTALL_FOLDER)
DAEMON_BASE := $(LIKWID_BASE)
GROUPS_BASE := $(LIKWID_BASE)/groups
all: $(INSTALL_FOLDER)/liblikwid.a $(INSTALL_FOLDER)/likwid-accessD cleanup prepare_collector
all: $(INSTALL_FOLDER)/liblikwid.a cleanup prepare_collector
else
DAEMON_BASE= $(LIKWID_BASE)/sbin
GROUPS_BASE := $(LIKWID_BASE)/share/likwid/perfgroups
@ -45,6 +51,7 @@ $(BUILD_FOLDER)/likwid-$(LIKWID_VERSION): $(BUILD_FOLDER)/likwid-$(LIKWID_VERSIO
$(INSTALL_FOLDER)/liblikwid.a: $(BUILD_FOLDER)/likwid-$(LIKWID_VERSION) $(INSTALL_FOLDER)
sed -i -e s+"PREFIX ?= .*"+"PREFIX = $(LIKWID_BASE)"+g \
-e s+"SHARED_LIBRARY = .*"+"SHARED_LIBRARY = false"+g \
-e s+"ACCESSMODE = .*"+"ACCESSMODE = accessdaemon"+g \
-e s+"INSTALLED_ACCESSDAEMON = .*"+"INSTALLED_ACCESSDAEMON = $(DAEMON_BASE)/likwid-accessD"+g \
-e s+"LIKWIDGROUPPATH = .*"+"LIKWIDGROUPPATH = $(GROUPS_BASE)"+g \
$(BUILD_FOLDER)/likwid-$(LIKWID_VERSION)/config.mk
@ -64,3 +71,8 @@ prepare_collector: likwidMetric.go
cleanup:
rm -rf $(BUILD_FOLDER)
clean: cleanup
rm -rf likwid
.PHONY: clean

25
collectors/cpustatMetric.go
View File

@ -1,19 +1,19 @@
package collectors
import (
"encoding/json"
"fmt"
lp "github.com/influxdata/line-protocol"
"io/ioutil"
"strconv"
"strings"
"time"
"encoding/json"
)
const CPUSTATFILE = `/proc/stat`
type CpustatCollectorConfig struct {
ExcludeMetrics []string `json:"exclude_metrics, omitempty"`
ExcludeMetrics []string `json:"exclude_metrics, omitempty"`
}
type CpustatCollector struct {
@ -25,22 +25,21 @@ func (m *CpustatCollector) Init(config []byte) error {
m.name = "CpustatCollector"
m.setup()
if len(config) > 0 {
err := json.Unmarshal(config, &m.config)
if err != nil {
return err
}
err := json.Unmarshal(config, &m.config)
if err != nil {
return err
}
}
m.init = true
return nil
}
func ParseStatLine(line string, cpu int, exclude []string, out *[]lp.MutableMetric) {
ls := strings.Fields(line)
matches := []string{"", "cpu_user", "cpu_nice", "cpu_system", "cpu_idle", "cpu_iowait", "cpu_irq", "cpu_softirq", "cpu_steal", "cpu_guest", "cpu_guest_nice"}
for _, ex := range exclude {
matches, _ = RemoveFromStringList(matches, ex)
}
for _, ex := range exclude {
matches, _ = RemoveFromStringList(matches, ex)
}
var tags map[string]string
if cpu < 0 {
@ -62,9 +61,9 @@ func ParseStatLine(line string, cpu int, exclude []string, out *[]lp.MutableMetr
}
func (m *CpustatCollector) Read(interval time.Duration, out *[]lp.MutableMetric) {
if (!m.init) {
return
}
if !m.init {
return
}
buffer, err := ioutil.ReadFile(string(CPUSTATFILE))
if err != nil {

116
collectors/customCmdMetric.go
View File

@ -1,61 +1,61 @@
package collectors
import (
"encoding/json"
"errors"
lp "github.com/influxdata/line-protocol"
"io/ioutil"
"log"
"os/exec"
"time"
"encoding/json"
"strings"
"time"
)
const CUSTOMCMDPATH = `/home/unrz139/Work/cc-metric-collector/collectors/custom`
type CustomCmdCollectorConfig struct {
commands []string `json:"commands"`
files []string `json:"files"`
ExcludeMetrics []string `json:"exclude_metrics"`
commands []string `json:"commands"`
files []string `json:"files"`
ExcludeMetrics []string `json:"exclude_metrics"`
}
type CustomCmdCollector struct {
MetricCollector
handler *lp.MetricHandler
parser *lp.Parser
config CustomCmdCollectorConfig
handler *lp.MetricHandler
parser *lp.Parser
config CustomCmdCollectorConfig
commands []string
files []string
files []string
}
func (m *CustomCmdCollector) Init(config []byte) error {
var err error
var err error
m.name = "CustomCmdCollector"
if len(config) > 0 {
err = json.Unmarshal(config, &m.config)
if err != nil {
log.Print(err.Error())
return err
}
err = json.Unmarshal(config, &m.config)
if err != nil {
log.Print(err.Error())
return err
}
}
m.setup()
for _, c := range m.config.commands {
cmdfields := strings.Fields(c)
command := exec.Command(cmdfields[0], strings.Join(cmdfields[1:], " "))
command.Wait()
_, err = command.Output()
if err != nil {
m.commands = append(m.commands, c)
}
cmdfields := strings.Fields(c)
command := exec.Command(cmdfields[0], strings.Join(cmdfields[1:], " "))
command.Wait()
_, err = command.Output()
if err != nil {
m.commands = append(m.commands, c)
}
}
for _, f := range m.config.files {
_, err = ioutil.ReadFile(f)
if err == nil {
m.files = append(m.files, f)
} else {
log.Print(err.Error())
continue
}
_, err = ioutil.ReadFile(f)
if err == nil {
m.files = append(m.files, f)
} else {
log.Print(err.Error())
continue
}
}
if len(m.files) == 0 && len(m.commands) == 0 {
return errors.New("No metrics to collect")
@ -72,9 +72,9 @@ var DefaultTime = func() time.Time {
}
func (m *CustomCmdCollector) Read(interval time.Duration, out *[]lp.MutableMetric) {
if !m.init {
return
}
if !m.init {
return
}
for _, cmd := range m.commands {
cmdfields := strings.Fields(cmd)
command := exec.Command(cmdfields[0], strings.Join(cmdfields[1:], " "))
@ -89,38 +89,38 @@ func (m *CustomCmdCollector) Read(interval time.Duration, out *[]lp.MutableMetri
log.Print(err)
continue
}
for _, c := range cmdmetrics {
_, skip := stringArrayContains(m.config.ExcludeMetrics, c.Name())
if skip {
continue
}
y, err := lp.New(c.Name(), Tags2Map(c), Fields2Map(c), c.Time())
if err == nil {
*out = append(*out, y)
}
}
for _, c := range cmdmetrics {
_, skip := stringArrayContains(m.config.ExcludeMetrics, c.Name())
if skip {
continue
}
y, err := lp.New(c.Name(), Tags2Map(c), Fields2Map(c), c.Time())
if err == nil {
*out = append(*out, y)
}
}
}
for _, file := range m.files {
buffer, err := ioutil.ReadFile(file)
if err != nil {
log.Print(err)
return
}
fmetrics, err := m.parser.Parse(buffer)
buffer, err := ioutil.ReadFile(file)
if err != nil {
log.Print(err)
return
}
fmetrics, err := m.parser.Parse(buffer)
if err != nil {
log.Print(err)
continue
}
for _, f := range fmetrics {
_, skip := stringArrayContains(m.config.ExcludeMetrics, f.Name())
if skip {
continue
}
y, err := lp.New(f.Name(), Tags2Map(f), Fields2Map(f), f.Time())
if err == nil {
*out = append(*out, y)
}
}
for _, f := range fmetrics {
_, skip := stringArrayContains(m.config.ExcludeMetrics, f.Name())
if skip {
continue
}
y, err := lp.New(f.Name(), Tags2Map(f), Fields2Map(f), f.Time())
if err == nil {
*out = append(*out, y)
}
}
}
}

104
collectors/diskstatMetric.go
View File

@ -3,37 +3,36 @@ package collectors
import (
lp "github.com/influxdata/line-protocol"
"io/ioutil"
// "log"
// "log"
"encoding/json"
"errors"
"strconv"
"strings"
"time"
"encoding/json"
"errors"
)
const DISKSTATFILE = `/proc/diskstats`
const DISKSTAT_SYSFSPATH = `/sys/block`
type DiskstatCollectorConfig struct {
ExcludeMetrics []string `json:"exclude_metrics, omitempty"`
ExcludeMetrics []string `json:"exclude_metrics, omitempty"`
}
type DiskstatCollector struct {
MetricCollector
matches map[int]string
config DiskstatCollectorConfig
config DiskstatCollectorConfig
}
func (m *DiskstatCollector) Init(config []byte) error {
var err error
var err error
m.name = "DiskstatCollector"
m.setup()
if len(config) > 0 {
err = json.Unmarshal(config, &m.config)
if err != nil {
return err
}
err = json.Unmarshal(config, &m.config)
if err != nil {
return err
}
}
// https://www.kernel.org/doc/html/latest/admin-guide/iostats.html
matches := map[int]string{
@ -57,58 +56,57 @@ func (m *DiskstatCollector) Init(config []byte) error {
}
m.matches = make(map[int]string)
for k, v := range matches {
_, skip := stringArrayContains(m.config.ExcludeMetrics, v)
if (!skip) {
m.matches[k] = v
}
_, skip := stringArrayContains(m.config.ExcludeMetrics, v)
if !skip {
m.matches[k] = v
}
}
if len(m.matches) == 0 {
return errors.New("No metrics to collect")
return errors.New("No metrics to collect")
}
_, err = ioutil.ReadFile(string(DISKSTATFILE))
if err == nil {
m.init = true
}
if err == nil {
m.init = true
}
return err
}
func (m *DiskstatCollector) Read(interval time.Duration, out *[]lp.MutableMetric) {
var lines []string
if !m.init {
return
}
var lines []string
if !m.init {
return
}
buffer, err := ioutil.ReadFile(string(DISKSTATFILE))
if err != nil {
return
}
lines = strings.Split(string(buffer), "\n")
buffer, err := ioutil.ReadFile(string(DISKSTATFILE))
if err != nil {
return
}
lines = strings.Split(string(buffer), "\n")
for _, line := range lines {
if len(line) == 0 {
continue
}
f := strings.Fields(line)
if strings.Contains(f[2], "loop") {
continue
}
tags := map[string]string{
"device": f[2],
"type": "node",
}
for idx, name := range m.matches {
if idx < len(f) {
x, err := strconv.ParseInt(f[idx], 0, 64)
if err == nil {
y, err := lp.New(name, tags, map[string]interface{}{"value": int(x)}, time.Now())
if err == nil {
*out = append(*out, y)
}
}
}
}
}
for _, line := range lines {
if len(line) == 0 {
continue
}
f := strings.Fields(line)
if strings.Contains(f[2], "loop") {
continue
}
tags := map[string]string{
"device": f[2],
"type": "node",
}
for idx, name := range m.matches {
if idx < len(f) {
x, err := strconv.ParseInt(f[idx], 0, 64)
if err == nil {
y, err := lp.New(name, tags, map[string]interface{}{"value": int(x)}, time.Now())
if err == nil {
*out = append(*out, y)
}
}
}
}
}
return
}

223
collectors/infinibandMetric.go
View File

@ -6,13 +6,13 @@ import (
"io/ioutil"
"log"
"os/exec"
// "os"
// "os"
"encoding/json"
"errors"
"path/filepath"
"strconv"
"strings"
"time"
"encoding/json"
"errors"
)
const BASEPATH = `/sys/class/infiniband/`
@ -20,67 +20,67 @@ const LIDFILE = `/sys/class/infiniband/mlx4_0/ports/1/lid`
const PERFQUERY = `/usr/sbin/perfquery`
type InfinibandCollectorConfig struct {
ExcludeDevices []string `json:"exclude_devices, omitempty"`
ExcludeDevices []string `json:"exclude_devices, omitempty"`
}
type InfinibandCollector struct {
MetricCollector
tags map[string]string
lids map[string]map[string]string
config NetstatCollectorConfig
tags map[string]string
lids map[string]map[string]string
config NetstatCollectorConfig
use_perfquery bool
}
func (m *InfinibandCollector) Init(config []byte) error {
var err error
var err error
m.name = "InfinibandCollector"
m.use_perfquery = false
m.setup()
m.tags = map[string]string{"type": "node"}
if len(config) > 0 {
err = json.Unmarshal(config, &m.config)
if err != nil {
return err
}
err = json.Unmarshal(config, &m.config)
if err != nil {
return err
}
}
m.lids = make(map[string]map[string]string)
p := fmt.Sprintf("%s/*/ports/*/lid", string(BASEPATH))
files, err := filepath.Glob(p)
for _, f := range(files) {
lid, err := ioutil.ReadFile(f)
if err == nil {
plist := strings.Split(strings.Replace(f, string(BASEPATH), "", -1), "/")
skip := false
for _, d := range m.config.ExcludeDevices {
if d == plist[0] {
skip = true
}
}
if !skip {
m.lids[plist[0]] = make(map[string]string)
m.lids[plist[0]][plist[2]] = string(lid)
}
}
files, err := filepath.Glob(p)
for _, f := range files {
lid, err := ioutil.ReadFile(f)
if err == nil {
plist := strings.Split(strings.Replace(f, string(BASEPATH), "", -1), "/")
skip := false
for _, d := range m.config.ExcludeDevices {
if d == plist[0] {
skip = true
}
}
if !skip {
m.lids[plist[0]] = make(map[string]string)
m.lids[plist[0]][plist[2]] = string(lid)
}
}
}
for _, ports := range m.lids {
for port, lid := range ports {
args := fmt.Sprintf("-r %s %s 0xf000", lid, port)
command := exec.Command(PERFQUERY, args)
command.Wait()
_, err := command.Output()
if (err == nil) {
m.use_perfquery = true
}
break
}
break
for port, lid := range ports {
args := fmt.Sprintf("-r %s %s 0xf000", lid, port)
command := exec.Command(PERFQUERY, args)
command.Wait()
_, err := command.Output()
if err == nil {
m.use_perfquery = true
}
break
}
break
}
if len(m.lids) > 0 {
m.init = true
m.init = true
} else {
err = errors.New("No usable devices")
err = errors.New("No usable devices")
}
return err
@ -88,15 +88,15 @@ func (m *InfinibandCollector) Init(config []byte) error {
func DoPerfQuery(dev string, lid string, port string, tags map[string]string, out *[]lp.MutableMetric) error {
args := fmt.Sprintf("-r %s %s 0xf000", lid, port)
command := exec.Command(PERFQUERY, args)
command.Wait()
stdout, err := command.Output()
if err != nil {
log.Print(err)
return err
}
ll := strings.Split(string(stdout), "\n")
args := fmt.Sprintf("-r %s %s 0xf000", lid, port)
command := exec.Command(PERFQUERY, args)
command.Wait()
stdout, err := command.Output()
if err != nil {
log.Print(err)
return err
}
ll := strings.Split(string(stdout), "\n")
for _, line := range ll {
if strings.HasPrefix(line, "PortRcvData") || strings.HasPrefix(line, "RcvData") {
@ -124,89 +124,88 @@ func DoPerfQuery(dev string, lid string, port string, tags map[string]string, ou
}
func DoSysfsRead(dev string, lid string, port string, tags map[string]string, out *[]lp.MutableMetric) error {
path := fmt.Sprintf("%s/%s/ports/%s/counters/", string(BASEPATH), dev, port)
buffer, err := ioutil.ReadFile(fmt.Sprintf("%s/port_rcv_data", path))
if err == nil {
data := strings.Replace(string(buffer), "\n", "", -1)
v, err := strconv.ParseFloat(data, 64)
path := fmt.Sprintf("%s/%s/ports/%s/counters/", string(BASEPATH), dev, port)
buffer, err := ioutil.ReadFile(fmt.Sprintf("%s/port_rcv_data", path))
if err == nil {
data := strings.Replace(string(buffer), "\n", "", -1)
v, err := strconv.ParseFloat(data, 64)
if err == nil {
y, err := lp.New("ib_recv", tags, map[string]interface{}{"value": float64(v)}, time.Now())
if err == nil {
*out = append(*out, y)
}
}
}
buffer, err = ioutil.ReadFile(fmt.Sprintf("%s/port_xmit_data", path))
if err == nil {
data := strings.Replace(string(buffer), "\n", "", -1)
v, err := strconv.ParseFloat(data, 64)
}
buffer, err = ioutil.ReadFile(fmt.Sprintf("%s/port_xmit_data", path))
if err == nil {
data := strings.Replace(string(buffer), "\n", "", -1)
v, err := strconv.ParseFloat(data, 64)
if err == nil {
y, err := lp.New("ib_xmit", tags, map[string]interface{}{"value": float64(v)}, time.Now())
if err == nil {
*out = append(*out, y)
}
}
}
return nil
}
return nil
}
func (m *InfinibandCollector) Read(interval time.Duration, out *[]lp.MutableMetric) {
if m.init {
for dev, ports := range m.lids {
for port, lid := range ports {
tags := map[string]string{"type" : "node", "device" : dev, "port" : port}
if m.use_perfquery {
DoPerfQuery(dev, lid, port, tags, out)
} else {
DoSysfsRead(dev, lid, port, tags, out)
}
}
}
}
if m.init {
for dev, ports := range m.lids {
for port, lid := range ports {
tags := map[string]string{"type": "node", "device": dev, "port": port}
if m.use_perfquery {
DoPerfQuery(dev, lid, port, tags, out)
} else {
DoSysfsRead(dev, lid, port, tags, out)
}
}
}
}
// buffer, err := ioutil.ReadFile(string(LIDFILE))
// buffer, err := ioutil.ReadFile(string(LIDFILE))
// if err != nil {
// log.Print(err)
// return
// }
// if err != nil {
// log.Print(err)
// return
// }
// args := fmt.Sprintf("-r %s 1 0xf000", string(buffer))
// args := fmt.Sprintf("-r %s 1 0xf000", string(buffer))
// command := exec.Command(PERFQUERY, args)
// command.Wait()
// stdout, err := command.Output()
// if err != nil {
// log.Print(err)
// return
// }
// command := exec.Command(PERFQUERY, args)
// command.Wait()
// stdout, err := command.Output()
// if err != nil {
// log.Print(err)
// return
// }
// ll := strings.Split(string(stdout), "\n")
// ll := strings.Split(string(stdout), "\n")
// for _, line := range ll {
// if strings.HasPrefix(line, "PortRcvData") || strings.HasPrefix(line, "RcvData") {
// lv := strings.Fields(line)
// v, err := strconv.ParseFloat(lv[1], 64)
// if err == nil {
// y, err := lp.New("ib_recv", m.tags, map[string]interface{}{"value": float64(v)}, time.Now())
// if err == nil {
// *out = append(*out, y)
// }
// }
// }
// if strings.HasPrefix(line, "PortXmitData") || strings.HasPrefix(line, "XmtData") {
// lv := strings.Fields(line)
// v, err := strconv.ParseFloat(lv[1], 64)
// if err == nil {
// y, err := lp.New("ib_xmit", m.tags, map[string]interface{}{"value": float64(v)}, time.Now())
// if err == nil {
// *out = append(*out, y)
// }
// }
// }
// }
// for _, line := range ll {
// if strings.HasPrefix(line, "PortRcvData") || strings.HasPrefix(line, "RcvData") {
// lv := strings.Fields(line)
// v, err := strconv.ParseFloat(lv[1], 64)
// if err == nil {
// y, err := lp.New("ib_recv", m.tags, map[string]interface{}{"value": float64(v)}, time.Now())
// if err == nil {
// *out = append(*out, y)
// }
// }
// }
// if strings.HasPrefix(line, "PortXmitData") || strings.HasPrefix(line, "XmtData") {
// lv := strings.Fields(line)
// v, err := strconv.ParseFloat(lv[1], 64)
// if err == nil {
// y, err := lp.New("ib_xmit", m.tags, map[string]interface{}{"value": float64(v)}, time.Now())
// if err == nil {
// *out = append(*out, y)
// }
// }
// }
// }
}
func (m *InfinibandCollector) Close() {

330
collectors/likwidMetric.go
View File

@ -9,50 +9,51 @@ package collectors
import "C"
import (
"encoding/json"
"errors"
"fmt"
lp "github.com/influxdata/line-protocol"
"log"
"strings"
"time"
"os"
"unsafe"
"math"
"encoding/json"
"gopkg.in/Knetic/govaluate.v2"
"io/ioutil"
"log"
"math"
"os"
"strconv"
"strings"
"time"
"unsafe"
)
type LikwidCollectorMetricConfig struct {
Name string `json:"name"`
Calc string `json:"calc"`
Socket_scope bool `json:"socket_scope"`
Publish bool `json:"publish"`
Name string `json:"name"`
Calc string `json:"calc"`
Socket_scope bool `json:"socket_scope"`
Publish bool `json:"publish"`
}
type LikwidCollectorEventsetConfig struct {
Events map[string]string `json:"events"`
Metrics []LikwidCollectorMetricConfig `json:"metrics"`
Events map[string]string `json:"events"`
Metrics []LikwidCollectorMetricConfig `json:"metrics"`
}
type LikwidCollectorConfig struct {
Eventsets []LikwidCollectorEventsetConfig `json:"eventsets"`
Metrics []LikwidCollectorMetricConfig `json:"globalmetrics"`
ExcludeMetrics []string `json:"exclude_metrics"`
Eventsets []LikwidCollectorEventsetConfig `json:"eventsets"`
Metrics []LikwidCollectorMetricConfig `json:"globalmetrics"`
ExcludeMetrics []string `json:"exclude_metrics"`
ForceOverwrite bool `json:"force_overwrite"`
}
type LikwidCollector struct {
MetricCollector
cpulist []C.int
sock2tid map[int]int
metrics map[C.int]map[string]int
groups []C.int
config LikwidCollectorConfig
results map[int]map[int]map[string]interface{}
mresults map[int]map[int]map[string]float64
cpulist []C.int
sock2tid map[int]int
metrics map[C.int]map[string]int
groups []C.int
config LikwidCollectorConfig
results map[int]map[int]map[string]interface{}
mresults map[int]map[int]map[string]float64
gmresults map[int]map[string]float64
basefreq float64
basefreq float64
}
type LikwidMetric struct {
@ -87,30 +88,30 @@ func getMetricId(group C.int, search string) (int, error) {
}
func eventsToEventStr(events map[string]string) string {
elist := make([]string, 0)
for k, v := range events {
elist = append(elist, fmt.Sprintf("%s:%s", v, k))
}
return strings.Join(elist, ",")
elist := make([]string, 0)
for k, v := range events {
elist = append(elist, fmt.Sprintf("%s:%s", v, k))
}
return strings.Join(elist, ",")
}
func getBaseFreq() float64 {
var freq float64 = math.NaN()
C.power_init(0)
info := C.get_powerInfo()
if float64(info.baseFrequency) != 0 {
freq = float64(info.baseFrequency)
} else {
buffer, err := ioutil.ReadFile("/sys/devices/system/cpu/cpu0/cpufreq/bios_limit")
if err == nil {
data := strings.Replace(string(buffer), "\n", "", -1)
x, err := strconv.ParseInt(data, 0, 64)
if err == nil {
freq = float64(x)*1E3
}
}
}
return freq
var freq float64 = math.NaN()
C.power_init(0)
info := C.get_powerInfo()
if float64(info.baseFrequency) != 0 {
freq = float64(info.baseFrequency)
} else {
buffer, err := ioutil.ReadFile("/sys/devices/system/cpu/cpu0/cpufreq/bios_limit")
if err == nil {
data := strings.Replace(string(buffer), "\n", "", -1)
x, err := strconv.ParseInt(data, 0, 64)
if err == nil {
freq = float64(x) * 1e3
}
}
}
return freq
}
func getSocketCpus() map[C.int]int {
@ -131,10 +132,10 @@ func (m *LikwidCollector) Init(config []byte) error {
var ret C.int
m.name = "LikwidCollector"
if len(config) > 0 {
err := json.Unmarshal(config, &m.config)
if err != nil {
return err
}
err := json.Unmarshal(config, &m.config)
if err != nil {
return err
}
}
m.setup()
cpulist := CpuList()
@ -155,28 +156,30 @@ func (m *LikwidCollector) Init(config []byte) error {
if ret != 0 {
return errors.New("Failed to initialize LIKWID topology")
}
os.Setenv("LIKWID_FORCE", "1")
if m.config.ForceOverwrite {
os.Setenv("LIKWID_FORCE", "1")
}
ret = C.perfmon_init(C.int(len(m.cpulist)), &m.cpulist[0])
if ret != 0 {
C.topology_finalize()
return errors.New("Failed to initialize LIKWID topology")
}
for i, evset := range m.config.Eventsets {
estr := eventsToEventStr(evset.Events)
cstr := C.CString(estr)
gid := C.perfmon_addEventSet(cstr)
if gid >= 0 {
m.groups = append(m.groups, gid)
}
C.free(unsafe.Pointer(cstr))
m.results[i] = make(map[int]map[string]interface{})
m.mresults[i] = make(map[int]map[string]float64)
for tid, _ := range m.cpulist {
m.results[i][tid] = make(map[string]interface{})
m.mresults[i][tid] = make(map[string]float64)
m.gmresults[tid] = make(map[string]float64)
}
estr := eventsToEventStr(evset.Events)
cstr := C.CString(estr)
gid := C.perfmon_addEventSet(cstr)
if gid >= 0 {
m.groups = append(m.groups, gid)
}
C.free(unsafe.Pointer(cstr))
m.results[i] = make(map[int]map[string]interface{})
m.mresults[i] = make(map[int]map[string]float64)
for tid, _ := range m.cpulist {
m.results[i][tid] = make(map[string]interface{})
m.mresults[i][tid] = make(map[string]float64)
m.gmresults[tid] = make(map[string]float64)
}
}
if len(m.groups) == 0 {
@ -192,14 +195,14 @@ func (m *LikwidCollector) Init(config []byte) error {
func (m *LikwidCollector) Read(interval time.Duration, out *[]lp.MutableMetric) {
if !m.init {
return
return
}
var ret C.int
for i, gid := range m.groups {
evset := m.config.Eventsets[i]
ret = C.perfmon_setupCounters(gid)
if ret != 0 {
evset := m.config.Eventsets[i]
ret = C.perfmon_setupCounters(gid)
if ret != 0 {
log.Print("Failed to setup performance group ", C.perfmon_getGroupName(gid))
continue
}
@ -216,113 +219,112 @@ func (m *LikwidCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
}
var eidx C.int
for tid, _ := range m.cpulist {
for eidx = 0; int(eidx) < len(evset.Events); eidx++ {
ctr := C.perfmon_getCounterName(gid, eidx)
gctr := C.GoString(ctr)
res := C.perfmon_getLastResult(gid, eidx, C.int(tid))
m.results[i][tid][gctr] = float64(res)
}
m.results[i][tid]["time"] = float64(interval)
m.results[i][tid]["inverseClock"] = float64(1.0/m.basefreq)
for _, metric := range evset.Metrics {
expression, err := govaluate.NewEvaluableExpression(metric.Calc)
if err != nil {
log.Print(err.Error())
continue
}
result, err := expression.Evaluate(m.results[i][tid]);
if err != nil {
log.Print(err.Error())
continue
}
m.mresults[i][tid][metric.Name] = float64(result.(float64))
}
for eidx = 0; int(eidx) < len(evset.Events); eidx++ {
ctr := C.perfmon_getCounterName(gid, eidx)
gctr := C.GoString(ctr)
res := C.perfmon_getLastResult(gid, eidx, C.int(tid))
m.results[i][tid][gctr] = float64(res)
}
m.results[i][tid]["time"] = float64(interval)
m.results[i][tid]["inverseClock"] = float64(1.0 / m.basefreq)
for _, metric := range evset.Metrics {
expression, err := govaluate.NewEvaluableExpression(metric.Calc)
if err != nil {
log.Print(err.Error())
continue
}
result, err := expression.Evaluate(m.results[i][tid])
if err != nil {
log.Print(err.Error())
continue
}
m.mresults[i][tid][metric.Name] = float64(result.(float64))
}
}
}
for _, metric := range m.config.Metrics {
for tid, _ := range m.cpulist {
var params map[string]interface{}
expression, err := govaluate.NewEvaluableExpression(metric.Calc)
if err != nil {
log.Print(err.Error())
continue
}
params = make(map[string]interface{})
for j, _ := range m.groups {
for mname, mres := range m.mresults[j][tid] {
params[mname] = mres
}
}
result, err := expression.Evaluate(params);
if err != nil {
log.Print(err.Error())
continue
}
m.gmresults[tid][metric.Name] = float64(result.(float64))
}
for tid, _ := range m.cpulist {
var params map[string]interface{}
expression, err := govaluate.NewEvaluableExpression(metric.Calc)
if err != nil {
log.Print(err.Error())
continue
}
params = make(map[string]interface{})
for j, _ := range m.groups {
for mname, mres := range m.mresults[j][tid] {
params[mname] = mres
}
}
result, err := expression.Evaluate(params)
if err != nil {
log.Print(err.Error())
continue
}
m.gmresults[tid][metric.Name] = float64(result.(float64))
}
}
for i, _ := range m.groups {
evset := m.config.Eventsets[i]
for _, metric := range evset.Metrics {
_, skip := stringArrayContains(m.config.ExcludeMetrics, metric.Name)
if metric.Publish && !skip {
if metric.Socket_scope {
for sid, tid := range m.sock2tid {
y, err := lp.New(metric.Name,
map[string]string{"type": "socket", "type-id": fmt.Sprintf("%d", int(sid))},
map[string]interface{}{"value": m.mresults[i][tid][metric.Name]},
time.Now())
if err == nil {
*out = append(*out, y)
}
}
} else {
for tid, cpu := range m.cpulist {
y, err := lp.New(metric.Name,
map[string]string{"type": "cpu", "type-id": fmt.Sprintf("%d", int(cpu))},
map[string]interface{}{"value": m.mresults[i][tid][metric.Name]},
time.Now())
evset := m.config.Eventsets[i]
for _, metric := range evset.Metrics {
_, skip := stringArrayContains(m.config.ExcludeMetrics, metric.Name)
if metric.Publish && !skip {
if metric.Socket_scope {
for sid, tid := range m.sock2tid {
y, err := lp.New(metric.Name,
map[string]string{"type": "socket", "type-id": fmt.Sprintf("%d", int(sid))},
map[string]interface{}{"value": m.mresults[i][tid][metric.Name]},
time.Now())
if err == nil {
*out = append(*out, y)
}
}
}
}
}
*out = append(*out, y)
}
}
} else {
for tid, cpu := range m.cpulist {
y, err := lp.New(metric.Name,
map[string]string{"type": "cpu", "type-id": fmt.Sprintf("%d", int(cpu))},
map[string]interface{}{"value": m.mresults[i][tid][metric.Name]},
time.Now())
if err == nil {
*out = append(*out, y)
}
}
}
}
}
}
for _, metric := range m.config.Metrics {
_, skip := stringArrayContains(m.config.ExcludeMetrics, metric.Name)
if metric.Publish && !skip {
if metric.Socket_scope {
for sid, tid := range m.sock2tid {
y, err := lp.New(metric.Name,
map[string]string{"type": "socket", "type-id": fmt.Sprintf("%d", int(sid))},
map[string]interface{}{"value": m.gmresults[tid][metric.Name]},
time.Now())
if err == nil {
*out = append(*out, y)
}
}
} else {
for tid, cpu := range m.cpulist {
y, err := lp.New(metric.Name,
map[string]string{"type": "cpu", "type-id": fmt.Sprintf("%d", int(cpu))},
map[string]interface{}{"value": m.gmresults[tid][metric.Name]},
time.Now())
_, skip := stringArrayContains(m.config.ExcludeMetrics, metric.Name)
if metric.Publish && !skip {
if metric.Socket_scope {
for sid, tid := range m.sock2tid {
y, err := lp.New(metric.Name,
map[string]string{"type": "socket", "type-id": fmt.Sprintf("%d", int(sid))},
map[string]interface{}{"value": m.gmresults[tid][metric.Name]},
time.Now())
if err == nil {
*out = append(*out, y)
}
}
}
}
}
*out = append(*out, y)
}
}
} else {
for tid, cpu := range m.cpulist {
y, err := lp.New(metric.Name,
map[string]string{"type": "cpu", "type-id": fmt.Sprintf("%d", int(cpu))},
map[string]interface{}{"value": m.gmresults[tid][metric.Name]},
time.Now())
if err == nil {
*out = append(*out, y)
}
}
}
}
}
}
func (m *LikwidCollector) Close() {
if m.init {
m.init = false
m.init = false
C.perfmon_finalize()
C.topology_finalize()
}

26
collectors/loadavgMetric.go
View File

@ -1,18 +1,18 @@
package collectors
import (
"encoding/json"
lp "github.com/influxdata/line-protocol"
"io/ioutil"
"strconv"
"strings"
"time"
"encoding/json"
)
const LOADAVGFILE = `/proc/loadavg`
type LoadavgCollectorConfig struct {
ExcludeMetrics []string `json:"exclude_metrics, omitempty"`
ExcludeMetrics []string `json:"exclude_metrics, omitempty"`
}
type LoadavgCollector struct {
@ -20,17 +20,17 @@ type LoadavgCollector struct {
tags map[string]string
load_matches []string
proc_matches []string
config LoadavgCollectorConfig
config LoadavgCollectorConfig
}
func (m *LoadavgCollector) Init(config []byte) error {
m.name = "LoadavgCollector"
m.setup()
if len(config) > 0 {
err := json.Unmarshal(config, &m.config)
if err != nil {
return err
}
err := json.Unmarshal(config, &m.config)
if err != nil {
return err
}
}
m.tags = map[string]string{"type": "node"}
m.load_matches = []string{"load_one", "load_five", "load_fifteen"}
@ -40,10 +40,10 @@ func (m *LoadavgCollector) Init(config []byte) error {
}
func (m *LoadavgCollector) Read(interval time.Duration, out *[]lp.MutableMetric) {
var skip bool
if !m.init {
return
}
var skip bool
if !m.init {
return
}
buffer, err := ioutil.ReadFile(string(LOADAVGFILE))
if err != nil {
@ -54,7 +54,7 @@ func (m *LoadavgCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
for i, name := range m.load_matches {
x, err := strconv.ParseFloat(ls[i], 64)
if err == nil {
_, skip = stringArrayContains(m.config.ExcludeMetrics, name)
_, skip = stringArrayContains(m.config.ExcludeMetrics, name)
y, err := lp.New(name, m.tags, map[string]interface{}{"value": float64(x)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -65,7 +65,7 @@ func (m *LoadavgCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
for i, name := range m.proc_matches {
x, err := strconv.ParseFloat(lv[i], 64)
if err == nil {
_, skip = stringArrayContains(m.config.ExcludeMetrics, name)
_, skip = stringArrayContains(m.config.ExcludeMetrics, name)
y, err := lp.New(name, m.tags, map[string]interface{}{"value": float64(x)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)

98
collectors/lustreMetric.go
View File

@ -1,21 +1,21 @@
package collectors
import (
"encoding/json"
"errors"
lp "github.com/influxdata/line-protocol"
"io/ioutil"
"log"
"strconv"
"strings"
"time"
"encoding/json"
"errors"
)
const LUSTREFILE = `/proc/fs/lustre/llite/lnec-XXXXXX/stats`
type LustreCollectorConfig struct {
procfiles []string `json:"procfiles"`
ExcludeMetrics []string `json:"exclude_metrics"`
procfiles []string `json:"procfiles"`
ExcludeMetrics []string `json:"exclude_metrics"`
}
type LustreCollector struct {
@ -23,17 +23,17 @@ type LustreCollector struct {
tags map[string]string
matches map[string]map[string]int
devices []string
config LustreCollectorConfig
config LustreCollectorConfig
}
func (m *LustreCollector) Init(config []byte) error {
var err error
var err error
m.name = "LustreCollector"
if len(config) > 0 {
err = json.Unmarshal(config, &m.config)
if err != nil {
return err
}
err = json.Unmarshal(config, &m.config)
if err != nil {
return err
}
}
m.setup()
m.tags = map[string]string{"type": "node"}
@ -47,15 +47,15 @@ func (m *LustreCollector) Init(config []byte) error {
"inode_permission": {"inode_permission": 1}}
m.devices = make([]string, 0)
for _, p := range m.config.procfiles {
_, err := ioutil.ReadFile(p)
if err == nil {
m.devices = append(m.devices, p)
} else {
log.Print(err.Error())
continue
}
_, err := ioutil.ReadFile(p)
if err == nil {
m.devices = append(m.devices, p)
} else {
log.Print(err.Error())
continue
}
}
if len(m.devices) == 0 {
return errors.New("No metrics to collect")
}
@ -64,39 +64,39 @@ func (m *LustreCollector) Init(config []byte) error {
}
func (m *LustreCollector) Read(interval time.Duration, out *[]lp.MutableMetric) {
if !m.init {
return
}
for _, p := range m.devices {
buffer, err := ioutil.ReadFile(p)
if !m.init {
return
}
for _, p := range m.devices {
buffer, err := ioutil.ReadFile(p)
if err != nil {
log.Print(err)
return
}
if err != nil {
log.Print(err)
return
}
for _, line := range strings.Split(string(buffer), "\n") {
lf := strings.Fields(line)
if len(lf) > 1 {
for match, fields := range m.matches {
if lf[0] == match {
for name, idx := range fields {
_, skip := stringArrayContains(m.config.ExcludeMetrics, name)
if skip {
continue
}
x, err := strconv.ParseInt(lf[idx], 0, 64)
if err == nil {
y, err := lp.New(name, m.tags, map[string]interface{}{"value": x}, time.Now())
if err == nil {
*out = append(*out, y)
}
}
}
}
}
}
}
for _, line := range strings.Split(string(buffer), "\n") {
lf := strings.Fields(line)
if len(lf) > 1 {
for match, fields := range m.matches {
if lf[0] == match {
for name, idx := range fields {
_, skip := stringArrayContains(m.config.ExcludeMetrics, name)
if skip {
continue
}
x, err := strconv.ParseInt(lf[idx], 0, 64)
if err == nil {
y, err := lp.New(name, m.tags, map[string]interface{}{"value": x}, time.Now())
if err == nil {
*out = append(*out, y)
}
}
}
}
}
}
}
}
}

34
collectors/memstatMetric.go
View File

@ -1,6 +1,7 @@
package collectors
import (
"encoding/json"
"errors"
"fmt"
lp "github.com/influxdata/line-protocol"
@ -9,13 +10,12 @@ import (
"strconv"
"strings"
"time"
"encoding/json"
)
const MEMSTATFILE = `/proc/meminfo`
type MemstatCollectorConfig struct {
ExcludeMetrics []string `json:"exclude_metrics"`
ExcludeMetrics []string `json:"exclude_metrics"`
}
type MemstatCollector struct {
@ -23,17 +23,17 @@ type MemstatCollector struct {
stats map[string]int64
tags map[string]string
matches map[string]string
config MemstatCollectorConfig
config MemstatCollectorConfig
}
func (m *MemstatCollector) Init(config []byte) error {
var err error
var err error
m.name = "MemstatCollector"
if len(config) > 0 {
err = json.Unmarshal(config, &m.config)
if err != nil {
return err
}
err = json.Unmarshal(config, &m.config)
if err != nil {
return err
}
}
m.stats = make(map[string]int64)
m.matches = make(map[string]string)
@ -48,13 +48,13 @@ func (m *MemstatCollector) Init(config []byte) error {
"MemAvailable": "mem_available",
"SwapFree": "swap_free"}
for k, v := range matches {
_, skip := stringArrayContains(m.config.ExcludeMetrics, k)
if (!skip) {
m.matches[k] = v
}
_, skip := stringArrayContains(m.config.ExcludeMetrics, k)
if !skip {
m.matches[k] = v
}
}
if len(m.matches) == 0 {
return errors.New("No metrics to collect")
return errors.New("No metrics to collect")
}
m.setup()
_, err = ioutil.ReadFile(string(MEMSTATFILE))
@ -65,9 +65,9 @@ func (m *MemstatCollector) Init(config []byte) error {
}
func (m *MemstatCollector) Read(interval time.Duration, out *[]lp.MutableMetric) {
if !m.init {
return
}
if !m.init {
return
}
buffer, err := ioutil.ReadFile(string(MEMSTATFILE))
if err != nil {
@ -115,7 +115,7 @@ func (m *MemstatCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
}
}
if _, found := m.stats[`MemShared`]; found {
_, skip := stringArrayContains(m.config.ExcludeMetrics, "mem_shared")
_, skip := stringArrayContains(m.config.ExcludeMetrics, "mem_shared")
y, err := lp.New("mem_shared", m.tags, map[string]interface{}{"value": int(float64(m.stats[`MemShared`]) * 1.0e-3)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)

17
collectors/metricCollector.go
View File

@ -1,13 +1,13 @@
package collectors
import (
"errors"
lp "github.com/influxdata/line-protocol"
"io/ioutil"
"log"
"strconv"
"strings"
"time"
"errors"
)
type MetricGetter interface {
@ -69,7 +69,7 @@ func intArrayContains(array []int, str int) (int, bool) {
}
func stringArrayContains(array []string, str string) (int, bool) {
for i, a := range array {
for i, a := range array {
if a == str {
return i, true
}
@ -144,11 +144,10 @@ func Fields2Map(metric lp.Metric) map[string]interface{} {
}
func RemoveFromStringList(s []string, r string) ([]string, error) {
for i, item := range s {
if r == item {
return append(s[:i], s[i+1:]...), nil
}
}
return s, errors.New("No such string in list")
for i, item := range s {
if r == item {
return append(s[:i], s[i+1:]...), nil
}
}
return s, errors.New("No such string in list")
}

18
collectors/netstatMetric.go
View File

@ -1,24 +1,24 @@
package collectors
import (
"encoding/json"
lp "github.com/influxdata/line-protocol"
"io/ioutil"
"log"
"strconv"
"strings"
"time"
"encoding/json"
)
const NETSTATFILE = `/proc/net/dev`
type NetstatCollectorConfig struct {
ExcludeDevices []string `json:"exclude_devices, omitempty"`
ExcludeDevices []string `json:"exclude_devices, omitempty"`
}
type NetstatCollector struct {
MetricCollector
config NetstatCollectorConfig
config NetstatCollectorConfig
matches map[int]string
}
@ -33,8 +33,8 @@ func (m *NetstatCollector) Init(config []byte) error {
}
err := json.Unmarshal(config, &m.config)
if err != nil {
log.Print(err.Error())
return err
log.Print(err.Error())
return err
}
_, err = ioutil.ReadFile(string(NETSTATFILE))
if err == nil {
@ -59,14 +59,14 @@ func (m *NetstatCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
dev := f[0][0 : len(f[0])-1]
cont := false
for _, d := range m.config.ExcludeDevices {
if d == dev {
cont = true
}
if d == dev {
cont = true
}
}
if cont {
continue
}
tags := map[string]string{"device" : dev, "type": "node"}
tags := map[string]string{"device": dev, "type": "node"}
for i, name := range m.matches {
v, err := strconv.ParseInt(f[i], 10, 0)
if err == nil {

60
collectors/nvidiaMetric.go
View File

@ -1,24 +1,24 @@
package collectors
import (
"encoding/json"
"errors"
"fmt"
"github.com/NVIDIA/go-nvml/pkg/nvml"
lp "github.com/influxdata/line-protocol"
"log"
"time"
"encoding/json"
)
type NvidiaCollectorConfig struct {
ExcludeMetrics []string `json:"exclude_metrics, omitempty"`
ExcludeDevices []string `json:"exclude_devices, omitempty"`
ExcludeMetrics []string `json:"exclude_metrics, omitempty"`
ExcludeDevices []string `json:"exclude_devices, omitempty"`
}
type NvidiaCollector struct {
MetricCollector
num_gpus int
config NvidiaCollectorConfig
config NvidiaCollectorConfig
}
func (m *NvidiaCollector) CatchPanic() error {
@ -32,14 +32,14 @@ func (m *NvidiaCollector) CatchPanic() error {
}
func (m *NvidiaCollector) Init(config []byte) error {
var err error
var err error
m.name = "NvidiaCollector"
m.setup()
if len(config) > 0 {
err = json.Unmarshal(config, &m.config)
if err != nil {
return err
}
err = json.Unmarshal(config, &m.config)
if err != nil {
return err
}
}
m.num_gpus = 0
defer m.CatchPanic()
@ -58,9 +58,9 @@ func (m *NvidiaCollector) Init(config []byte) error {
}
func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric) {
if (!m.init) {
return
}
if !m.init {
return
}
for i := 0; i < m.num_gpus; i++ {
device, ret := nvml.DeviceGetHandleByIndex(i)
if ret != nvml.SUCCESS {
@ -69,13 +69,13 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
}
_, skip := stringArrayContains(m.config.ExcludeDevices, fmt.Sprintf("%d", i))
if skip {
continue
continue
}
tags := map[string]string{"type": "accelerator", "type-id": fmt.Sprintf("%d", i)}
util, ret := nvml.DeviceGetUtilizationRates(device)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "util")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "util")
y, err := lp.New("util", tags, map[string]interface{}{"value": float64(util.Gpu)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -105,7 +105,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
temp, ret := nvml.DeviceGetTemperature(device, nvml.TEMPERATURE_GPU)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "temp")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "temp")
y, err := lp.New("temp", tags, map[string]interface{}{"value": float64(temp)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -114,7 +114,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
fan, ret := nvml.DeviceGetFanSpeed(device)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "fan")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "fan")
y, err := lp.New("fan", tags, map[string]interface{}{"value": float64(fan)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -138,7 +138,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
*out = append(*out, y)
}
} else if ret == nvml.ERROR_NOT_SUPPORTED {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "ecc_mode")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "ecc_mode")
y, err := lp.New("ecc_mode", tags, map[string]interface{}{"value": string("N/A")}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -147,7 +147,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
pstate, ret := nvml.DeviceGetPerformanceState(device)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "perf_state")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "perf_state")
y, err := lp.New("perf_state", tags, map[string]interface{}{"value": fmt.Sprintf("P%d", int(pstate))}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -156,7 +156,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
power, ret := nvml.DeviceGetPowerUsage(device)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "power_usage_report")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "power_usage_report")
y, err := lp.New("power_usage_report", tags, map[string]interface{}{"value": float64(power) / 1000}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -165,7 +165,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
gclk, ret := nvml.DeviceGetClockInfo(device, nvml.CLOCK_GRAPHICS)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "graphics_clock_report")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "graphics_clock_report")
y, err := lp.New("graphics_clock_report", tags, map[string]interface{}{"value": float64(gclk)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -174,7 +174,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
smclk, ret := nvml.DeviceGetClockInfo(device, nvml.CLOCK_SM)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "sm_clock_report")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "sm_clock_report")
y, err := lp.New("sm_clock_report", tags, map[string]interface{}{"value": float64(smclk)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -183,7 +183,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
memclk, ret := nvml.DeviceGetClockInfo(device, nvml.CLOCK_MEM)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "mem_clock_report")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "mem_clock_report")
y, err := lp.New("mem_clock_report", tags, map[string]interface{}{"value": float64(memclk)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -192,7 +192,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
max_gclk, ret := nvml.DeviceGetMaxClockInfo(device, nvml.CLOCK_GRAPHICS)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "max_graphics_clock")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "max_graphics_clock")
y, err := lp.New("max_graphics_clock", tags, map[string]interface{}{"value": float64(max_gclk)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -201,7 +201,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
max_smclk, ret := nvml.DeviceGetClockInfo(device, nvml.CLOCK_SM)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "max_sm_clock")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "max_sm_clock")
y, err := lp.New("max_sm_clock", tags, map[string]interface{}{"value": float64(max_smclk)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -210,7 +210,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
max_memclk, ret := nvml.DeviceGetClockInfo(device, nvml.CLOCK_MEM)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "max_mem_clock")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "max_mem_clock")
y, err := lp.New("max_mem_clock", tags, map[string]interface{}{"value": float64(max_memclk)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -219,7 +219,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
ecc_db, ret := nvml.DeviceGetTotalEccErrors(device, 1, 1)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "ecc_db_error")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "ecc_db_error")
y, err := lp.New("ecc_db_error", tags, map[string]interface{}{"value": float64(ecc_db)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -228,7 +228,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
ecc_sb, ret := nvml.DeviceGetTotalEccErrors(device, 0, 1)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "ecc_sb_error")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "ecc_sb_error")
y, err := lp.New("ecc_sb_error", tags, map[string]interface{}{"value": float64(ecc_sb)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -237,7 +237,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
pwr_limit, ret := nvml.DeviceGetPowerManagementLimit(device)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "power_man_limit")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "power_man_limit")
y, err := lp.New("power_man_limit", tags, map[string]interface{}{"value": float64(pwr_limit)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -246,7 +246,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
enc_util, _, ret := nvml.DeviceGetEncoderUtilization(device)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "encoder_util")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "encoder_util")
y, err := lp.New("encoder_util", tags, map[string]interface{}{"value": float64(enc_util)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)
@ -255,7 +255,7 @@ func (m *NvidiaCollector) Read(interval time.Duration, out *[]lp.MutableMetric)
dec_util, _, ret := nvml.DeviceGetDecoderUtilization(device)
if ret == nvml.SUCCESS {
_, skip = stringArrayContains(m.config.ExcludeMetrics, "decoder_util")
_, skip = stringArrayContains(m.config.ExcludeMetrics, "decoder_util")
y, err := lp.New("decoder_util", tags, map[string]interface{}{"value": float64(dec_util)}, time.Now())
if err == nil && !skip {
*out = append(*out, y)

26
collectors/topprocsMetric.go
View File

@ -1,40 +1,40 @@
package collectors
import (
"encoding/json"
"errors"
"fmt"
lp "github.com/influxdata/line-protocol"
"log"
"os/exec"
"strings"
"time"
"encoding/json"
"errors"
)
const MAX_NUM_PROCS = 10
type TopProcsCollectorConfig struct {
num_procs int `json:"num_procs"`
num_procs int `json:"num_procs"`
}
type TopProcsCollector struct {
MetricCollector
tags map[string]string
tags map[string]string
config TopProcsCollectorConfig
}
func (m *TopProcsCollector) Init(config []byte) error {
var err error
var err error
m.name = "TopProcsCollector"
m.tags = map[string]string{"type": "node"}
if len(config) > 0 {
err = json.Unmarshal(config, &m.config)
if err != nil {
return err
}
err = json.Unmarshal(config, &m.config)
if err != nil {
return err
}
}
if m.config.num_procs <= 0 || m.config.num_procs > MAX_NUM_PROCS {
return errors.New(fmt.Sprintf("num_procs option must be set in 'topprocs' config (range: 1-%d)", MAX_NUM_PROCS))
return errors.New(fmt.Sprintf("num_procs option must be set in 'topprocs' config (range: 1-%d)", MAX_NUM_PROCS))
}
m.setup()
command := exec.Command("ps", "-Ao", "comm", "--sort=-pcpu")
@ -48,9 +48,9 @@ func (m *TopProcsCollector) Init(config []byte) error {
}
func (m *TopProcsCollector) Read(interval time.Duration, out *[]lp.MutableMetric) {
if !m.init {
return
}
if !m.init {
return
}
command := exec.Command("ps", "-Ao", "comm", "--sort=-pcpu")
command.Wait()
stdout, err := command.Output()

2
config.json
View File

@ -42,6 +42,8 @@
]
},
"likwid": {
"accessmode" : "accessdaemon",
"daemon_path" : "/apps/likwid/5.2.0/sbin",
"eventsets": [
{
"events": {

36
metric-collector.go
View File

@ -45,13 +45,13 @@ var Receivers = map[string]receivers.ReceiverFuncs{
// Structure of the configuration file
type GlobalConfig struct {
Sink sinks.SinkConfig `json:"sink"`
Interval int `json:"interval"`
Duration int `json:"duration"`
Collectors []string `json:"collectors"`
Receiver receivers.ReceiverConfig `json:"receiver"`
DefTags map[string]string `json:"default_tags"`
CollectConfigs map[string]json.RawMessage `json:"collect_config"`
Sink sinks.SinkConfig `json:"sink"`
Interval int `json:"interval"`
Duration int `json:"duration"`
Collectors []string `json:"collectors"`
Receiver receivers.ReceiverConfig `json:"receiver"`
DefTags map[string]string `json:"default_tags"`
CollectConfigs map[string]json.RawMessage `json:"collect_config"`
}
// Load JSON configuration file
@ -79,9 +79,9 @@ func ReadCli() map[string]string {
m["logfile"] = *logfile
m["pidfile"] = *pidfile
if *once {
m["once"] = "true"
m["once"] = "true"
} else {
m["once"] = "false"
m["once"] = "false"
}
return m
}
@ -123,7 +123,7 @@ func RemovePidfile(pidfile string) error {
// General shutdown function that gets executed in case of interrupt or graceful shutdown
func shutdown(wg *sync.WaitGroup, collectors []string, sink sinks.SinkFuncs, recv receivers.ReceiverFuncs, pidfile string) {
log.Print("Shutdown...")
log.Print("Shutdown...")
for _, c := range collectors {
col := Collectors[c]
log.Print("Stop ", col.Name())
@ -236,11 +236,11 @@ func main() {
col := Collectors[c]
conf, found := config.CollectConfigs[c]
if !found {
conf = json.RawMessage("")
conf = json.RawMessage("")
}
err = col.Init([]byte(conf))
if err != nil {
log.Print("SKIP ", col.Name(), " (", err.Error(),")")
log.Print("SKIP ", col.Name(), " (", err.Error(), ")")
} else {
log.Print("Start ", col.Name())
tmp = append(tmp, c)
@ -251,10 +251,10 @@ func main() {
// Setup up ticker loop
if clicfg["once"] != "true" {
log.Print("Running loop every ", time.Duration(config.Interval)*time.Second)
} else {
log.Print("Running loop only once")
}
log.Print("Running loop every ", time.Duration(config.Interval)*time.Second)
} else {
log.Print("Running loop only once")
}
ticker := time.NewTicker(time.Duration(config.Interval) * time.Second)
done := make(chan bool)
@ -297,8 +297,8 @@ func main() {
log.Printf("sink error: %s\n", err)
}
if clicfg["once"] == "true" {
shutdown(&wg, config.Collectors, sink, recv, clicfg["pidfile"])
return
shutdown(&wg, config.Collectors, sink, recv, clicfg["pidfile"])
return
}
}
}