ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2024-09-20 07:27:25 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fixing format with gofmt

Browse Source
This commit is contained in:
Thomas Roehl 2021-03-25 17:47:08 +01:00
parent e9a74a4c9c
commit f32ff9d006
9 changed files with 310 additions and 325 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
collectors/infinibandMetric.go
View File

@ -5,8 +5,8 @@ import (
"io/ioutil" "io/ioutil"
"log" "log"
"os/exec" "os/exec"
"strings"
"strconv" "strconv"
"strings"
"time" "time"
) )
@ -17,48 +17,48 @@ type InfinibandCollector struct {
} }
func (m *InfinibandCollector) Init() { func (m *InfinibandCollector) Init() {
m.name = "InfinibandCollector" m.name = "InfinibandCollector"
m.setup() m.setup()
} }
func (m *InfinibandCollector) Read(interval time.Duration){ func (m *InfinibandCollector) Read(interval time.Duration) {
buffer, err := ioutil.ReadFile(string(LIDFILE)) buffer, err := ioutil.ReadFile(string(LIDFILE))
if err != nil { if err != nil {
log.Print(err) log.Print(err)
return return
} }
args := fmt.Sprintf("-r %s 1 0xf000", string(buffer)) args := fmt.Sprintf("-r %s 1 0xf000", string(buffer))
command := exec.Command("/usr/sbin/perfquery", args) command := exec.Command("/usr/sbin/perfquery", args)
command.Wait() command.Wait()
stdout, err := command.Output() stdout, err := command.Output()
if err != nil { if err != nil {
log.Print(err) log.Print(err)
return return
} }
ll := strings.Split(string(stdout), "\n") ll := strings.Split(string(stdout), "\n")
for _, line := range ll { for _, line := range ll {
if strings.HasPrefix(line, "PortRcvData") || strings.HasPrefix(line, "RcvData") { if strings.HasPrefix(line, "PortRcvData") || strings.HasPrefix(line, "RcvData") {
lv := strings.Fields(line) lv := strings.Fields(line)
v, err := strconv.ParseFloat(lv[1], 64) v, err := strconv.ParseFloat(lv[1], 64)
if err == nil { if err == nil {
m.node["ib_recv"] = float64(v) m.node["ib_recv"] = float64(v)
} }
} }
if strings.HasPrefix(line, "PortXmitData") || strings.HasPrefix(line, "XmtData") { if strings.HasPrefix(line, "PortXmitData") || strings.HasPrefix(line, "XmtData") {
lv := strings.Fields(line) lv := strings.Fields(line)
v, err := strconv.ParseFloat(lv[1], 64) v, err := strconv.ParseFloat(lv[1], 64)
if err == nil { if err == nil {
m.node["ib_xmit"] = float64(v) m.node["ib_xmit"] = float64(v)
} }
} }
} }
} }
func (m *InfinibandCollector) Close() { func (m *InfinibandCollector) Close() {
return return
} }

5
collectors/likwid/likwid.go
View File

@ -1,4 +1,5 @@
package main package main
/* /*
#cgo CFLAGS: -I. #cgo CFLAGS: -I.
#cgo LDFLAGS: -L. -llikwid -llikwid-hwloc -lm #cgo LDFLAGS: -L. -llikwid -llikwid-hwloc -lm
@ -11,7 +12,7 @@ import "unsafe"
func main() { func main() {
var topo C.CpuTopology_t var topo C.CpuTopology_t
C.topology_init(); C.topology_init()
topo = C.get_cpuTopology() topo = C.get_cpuTopology()
cpulist := make([]C.int, topo.numHWThreads) cpulist := make([]C.int, topo.numHWThreads)
for a := 0; a < int(topo.numHWThreads); a++ { for a := 0; a < int(topo.numHWThreads); a++ {
@ -27,5 +28,5 @@ func main() {
fmt.Println(v) fmt.Println(v)
C.free(unsafe.Pointer(gstring)) C.free(unsafe.Pointer(gstring))
C.perfmon_finalize() C.perfmon_finalize()
C.topology_finalize(); C.topology_finalize()
} }

225
collectors/likwidMetric.go
View File

@ -1,4 +1,5 @@
package collectors package collectors
/* /*
#cgo CFLAGS: -I./likwid #cgo CFLAGS: -I./likwid
#cgo LDFLAGS: -L./likwid -llikwid -llikwid-hwloc -lm #cgo LDFLAGS: -L./likwid -llikwid -llikwid-hwloc -lm
@ -9,166 +10,162 @@ import "C"
import ( import (
"errors" "errors"
// "io/ioutil" // "io/ioutil"
// "log" // "log"
// "strconv" // "strconv"
"fmt"
"log"
"strings" "strings"
"fmt"
"time" "time"
"unsafe" "unsafe"
"log"
//protocol "github.com/influxdata/line-protocol" //protocol "github.com/influxdata/line-protocol"
) )
type LikwidCollector struct { type LikwidCollector struct {
MetricCollector MetricCollector
cpulist []C.int cpulist []C.int
sock2tid map[int]int sock2tid map[int]int
metrics map[C.int]map[string]int metrics map[C.int]map[string]int
groups map[string]C.int groups map[string]C.int
init bool init bool
} }
type LikwidMetric struct { type LikwidMetric struct {
name string name string
search string search string
socket_scope bool socket_scope bool
group_idx int group_idx int
} }
const GROUPPATH = `/home/unrz139/Work/cc-metric-collector/collectors/likwid/groups` const GROUPPATH = `/home/unrz139/Work/cc-metric-collector/collectors/likwid/groups`
var likwid_metrics = map[string][]LikwidMetric{ var likwid_metrics = map[string][]LikwidMetric{
"MEM_DP": {LikwidMetric{name: "mem_bw", search: "Memory bandwidth [MBytes/s]", socket_scope: true}, "MEM_DP": {LikwidMetric{name: "mem_bw", search: "Memory bandwidth [MBytes/s]", socket_scope: true},
LikwidMetric{name: "pwr1", search: "Power [W]", socket_scope: true}, LikwidMetric{name: "pwr1", search: "Power [W]", socket_scope: true},
LikwidMetric{name: "pwr2", search: "Power DRAM [W]", socket_scope: true}, LikwidMetric{name: "pwr2", search: "Power DRAM [W]", socket_scope: true},
LikwidMetric{name: "flops_dp", search: "DP [MFLOP/s]", socket_scope: false}}, LikwidMetric{name: "flops_dp", search: "DP [MFLOP/s]", socket_scope: false}},
"FLOPS_SP" : {LikwidMetric{name: "clock", search: "Clock [MHz]", socket_scope: false}, "FLOPS_SP": {LikwidMetric{name: "clock", search: "Clock [MHz]", socket_scope: false},
LikwidMetric{name: "cpi", search: "CPI", socket_scope: false}, LikwidMetric{name: "cpi", search: "CPI", socket_scope: false},
LikwidMetric{name: "flops_sp", search: "SP [MFLOP/s]", socket_scope: false}}, LikwidMetric{name: "flops_sp", search: "SP [MFLOP/s]", socket_scope: false}},
} }
func getMetricId(group C.int, search string) (int, error) { func getMetricId(group C.int, search string) (int, error) {
for i := 0; i < int(C.perfmon_getNumberOfMetrics(group)); i++ { for i := 0; i < int(C.perfmon_getNumberOfMetrics(group)); i++ {
mname := C.perfmon_getMetricName(group, C.int(i)) mname := C.perfmon_getMetricName(group, C.int(i))
go_mname := C.GoString(mname) go_mname := C.GoString(mname)
if strings.Contains(go_mname, search) { if strings.Contains(go_mname, search) {
return i, nil return i, nil
} }
} }
return -1, errors.New(fmt.Sprintf("Cannot find metric for search string '%s' in group %d", search, int(group))) return -1, errors.New(fmt.Sprintf("Cannot find metric for search string '%s' in group %d", search, int(group)))
} }
func getSocketCpus() map[C.int]int { func getSocketCpus() map[C.int]int {
slist := SocketList() slist := SocketList()
var cpu C.int var cpu C.int
outmap := make(map[C.int]int) outmap := make(map[C.int]int)
for _, s := range slist { for _, s := range slist {
t := C.CString(fmt.Sprintf("S%d", s)) t := C.CString(fmt.Sprintf("S%d", s))
clen := C.cpustr_to_cpulist(t, &cpu, 1) clen := C.cpustr_to_cpulist(t, &cpu, 1)
if int(clen) == 1 { if int(clen) == 1 {
outmap[cpu] = s outmap[cpu] = s
} }
} }
return outmap return outmap
} }
func (m *LikwidCollector) Init() { func (m *LikwidCollector) Init() {
m.name = "LikwidCollector" m.name = "LikwidCollector"
m.setup() m.setup()
cpulist := CpuList() cpulist := CpuList()
m.cpulist = make([]C.int, len(cpulist)) m.cpulist = make([]C.int, len(cpulist))
slist := getSocketCpus() slist := getSocketCpus()
m.sock2tid = make(map[int]int) m.sock2tid = make(map[int]int)
for i, c := range cpulist { for i, c := range cpulist {
m.cpulist[i] = C.int(c) m.cpulist[i] = C.int(c)
if sid, found := slist[m.cpulist[i]]; found { if sid, found := slist[m.cpulist[i]]; found {
m.sock2tid[sid] = i m.sock2tid[sid] = i
} }
} }
m.metrics = make(map[C.int]map[string]int) m.metrics = make(map[C.int]map[string]int)
C.topology_init() C.topology_init()
C.perfmon_init(C.int(len(m.cpulist)), &m.cpulist[0]) C.perfmon_init(C.int(len(m.cpulist)), &m.cpulist[0])
gpath := C.CString(GROUPPATH) gpath := C.CString(GROUPPATH)
C.config_setGroupPath(gpath) C.config_setGroupPath(gpath)
C.free(unsafe.Pointer(gpath)) C.free(unsafe.Pointer(gpath))
m.init = true m.init = true
m.groups = make(map[string]C.int) m.groups = make(map[string]C.int)
for g, metrics := range likwid_metrics { for g, metrics := range likwid_metrics {
cstr := C.CString(g) cstr := C.CString(g)
gid := C.perfmon_addEventSet(cstr) gid := C.perfmon_addEventSet(cstr)
if gid >= 0 { if gid >= 0 {
m.groups[g] = gid m.groups[g] = gid
for i, metric := range metrics { for i, metric := range metrics {
idx, err := getMetricId(gid, metric.search) idx, err := getMetricId(gid, metric.search)
if err != nil { if err != nil {
log.Print(err) log.Print(err)
} else { } else {
likwid_metrics[g][i].group_idx = idx likwid_metrics[g][i].group_idx = idx
} }
} }
} else { } else {
log.Print("Failed to add events set ", g) log.Print("Failed to add events set ", g)
} }
C.free(unsafe.Pointer(cstr)) C.free(unsafe.Pointer(cstr))
} }
} }
func (m *LikwidCollector) Read(interval time.Duration){ func (m *LikwidCollector) Read(interval time.Duration) {
if m.init { if m.init {
for gname, gid := range m.groups { for gname, gid := range m.groups {
C.perfmon_setupCounters(gid) C.perfmon_setupCounters(gid)
C.perfmon_startCounters() C.perfmon_startCounters()
time.Sleep(interval) time.Sleep(interval)
C.perfmon_stopCounters() C.perfmon_stopCounters()
for _, lmetric := range likwid_metrics[gname] { for _, lmetric := range likwid_metrics[gname] {
if lmetric.socket_scope { if lmetric.socket_scope {
for sid, tid := range m.sock2tid { for sid, tid := range m.sock2tid {
res := C.perfmon_getLastMetric(gid, C.int(lmetric.group_idx), C.int(tid)) res := C.perfmon_getLastMetric(gid, C.int(lmetric.group_idx), C.int(tid))
m.sockets[int(sid)][lmetric.name] = float64(res) m.sockets[int(sid)][lmetric.name] = float64(res)
// log.Print("Metric '", lmetric.name,"' on Socket ",int(sid)," returns ", m.sockets[int(sid)][lmetric.name]) // log.Print("Metric '", lmetric.name,"' on Socket ",int(sid)," returns ", m.sockets[int(sid)][lmetric.name])
} }
} else { } else {
for tid, cpu := range m.cpulist { for tid, cpu := range m.cpulist {
res := C.perfmon_getLastMetric(gid, C.int(lmetric.group_idx), C.int(tid)) res := C.perfmon_getLastMetric(gid, C.int(lmetric.group_idx), C.int(tid))
m.cpus[int(cpu)][lmetric.name] = float64(res) m.cpus[int(cpu)][lmetric.name] = float64(res)
// log.Print("Metric '", lmetric.name,"' on CPU ",int(cpu)," returns ", m.cpus[int(cpu)][lmetric.name]) // log.Print("Metric '", lmetric.name,"' on CPU ",int(cpu)," returns ", m.cpus[int(cpu)][lmetric.name])
} }
} }
} }
for cpu, _ := range m.cpus { for cpu := range m.cpus {
if flops_dp, found := m.cpus[cpu]["flops_dp"]; found { if flops_dp, found := m.cpus[cpu]["flops_dp"]; found {
if flops_sp, found := m.cpus[cpu]["flops_sp"]; found { if flops_sp, found := m.cpus[cpu]["flops_sp"]; found {
m.cpus[cpu]["flops_any"] = flops_dp.(float64) + flops_sp.(float64) m.cpus[cpu]["flops_any"] = flops_dp.(float64) + flops_sp.(float64)
} }
} }
} }
for sid, _ := range m.sockets { for sid := range m.sockets {
if pwr1, found := m.sockets[int(sid)]["pwr1"]; found { if pwr1, found := m.sockets[int(sid)]["pwr1"]; found {
if pwr2, found := m.sockets[int(sid)]["pwr2"]; found { if pwr2, found := m.sockets[int(sid)]["pwr2"]; found {
sum := pwr1.(float64) + pwr2.(float64) sum := pwr1.(float64) + pwr2.(float64)
if sum > 0 { if sum > 0 {
m.sockets[int(sid)]["power"] = sum m.sockets[int(sid)]["power"] = sum
} }
delete(m.sockets[int(sid)], "pwr2") delete(m.sockets[int(sid)], "pwr2")
} }
delete(m.sockets[int(sid)], "pwr1") delete(m.sockets[int(sid)], "pwr1")
} }
} }
} }
} }
} }
func (m *LikwidCollector) Close() { func (m *LikwidCollector) Close() {
C.perfmon_finalize() C.perfmon_finalize()
C.topology_finalize() C.topology_finalize()
m.init = false m.init = false
return return
} }

9
collectors/loadavgMetric.go
View File

@ -3,8 +3,8 @@ package collectors
import ( import (
"io/ioutil" "io/ioutil"
"strconv" "strconv"
"time"
"strings" "strings"
"time"
) )
const LOADAVGFILE = `/proc/loadavg` const LOADAVGFILE = `/proc/loadavg`
@ -13,13 +13,12 @@ type LoadavgCollector struct {
MetricCollector MetricCollector
} }
func (m *LoadavgCollector) Init() { func (m *LoadavgCollector) Init() {
m.name = "LoadavgCollector" m.name = "LoadavgCollector"
m.setup() m.setup()
} }
func (m *LoadavgCollector) Read(interval time.Duration){ func (m *LoadavgCollector) Read(interval time.Duration) {
buffer, err := ioutil.ReadFile(string(LOADAVGFILE)) buffer, err := ioutil.ReadFile(string(LOADAVGFILE))
if err != nil { if err != nil {
@ -36,5 +35,5 @@ func (m *LoadavgCollector) Read(interval time.Duration){
} }
func (m *LoadavgCollector) Close() { func (m *LoadavgCollector) Close() {
return return
} }

51
collectors/lustreMetric.go
View File

@ -1,7 +1,7 @@
package collectors package collectors
import ( import (
// "errors" // "errors"
"io/ioutil" "io/ioutil"
"log" "log"
"strconv" "strconv"
@ -15,13 +15,12 @@ type LustreCollector struct {
MetricCollector MetricCollector
} }
func (m *LustreCollector) Init() { func (m *LustreCollector) Init() {
m.name = "LustreCollector" m.name = "LustreCollector"
m.setup() m.setup()
} }
func (m *LustreCollector) Read(interval time.Duration){ func (m *LustreCollector) Read(interval time.Duration) {
buffer, err := ioutil.ReadFile(string(LUSTREFILE)) buffer, err := ioutil.ReadFile(string(LUSTREFILE))
if err != nil { if err != nil {
@ -32,31 +31,31 @@ func (m *LustreCollector) Read(interval time.Duration){
for _, line := range strings.Split(string(buffer), "\n") { for _, line := range strings.Split(string(buffer), "\n") {
lf := strings.Fields(line) lf := strings.Fields(line)
if len(lf) > 1 { if len(lf) > 1 {
switch lf[0] { switch lf[0] {
case "read_bytes": case "read_bytes":
m.node["read_bytes"], err = strconv.ParseInt(lf[6], 0, 64) m.node["read_bytes"], err = strconv.ParseInt(lf[6], 0, 64)
m.node["read_requests"], err = strconv.ParseInt(lf[1], 0, 64) m.node["read_requests"], err = strconv.ParseInt(lf[1], 0, 64)
case "write_bytes": case "write_bytes":
m.node["write_bytes"], err = strconv.ParseInt(lf[6], 0, 64) m.node["write_bytes"], err = strconv.ParseInt(lf[6], 0, 64)
m.node["write_requests"], err = strconv.ParseInt(lf[1], 0, 64) m.node["write_requests"], err = strconv.ParseInt(lf[1], 0, 64)
case "open": case "open":
m.node["open"], err = strconv.ParseInt(lf[1], 0, 64) m.node["open"], err = strconv.ParseInt(lf[1], 0, 64)
case "close": case "close":
m.node["close"], err = strconv.ParseInt(lf[1], 0, 64) m.node["close"], err = strconv.ParseInt(lf[1], 0, 64)
case "setattr": case "setattr":
m.node["setattr"], err = strconv.ParseInt(lf[1], 0, 64) m.node["setattr"], err = strconv.ParseInt(lf[1], 0, 64)
case "getattr": case "getattr":
m.node["getattr"], err = strconv.ParseInt(lf[1], 0, 64) m.node["getattr"], err = strconv.ParseInt(lf[1], 0, 64)
case "statfs": case "statfs":
m.node["statfs"], err = strconv.ParseInt(lf[1], 0, 64) m.node["statfs"], err = strconv.ParseInt(lf[1], 0, 64)
case "inode_permission": case "inode_permission":
m.node["inode_permission"], err = strconv.ParseInt(lf[1], 0, 64) m.node["inode_permission"], err = strconv.ParseInt(lf[1], 0, 64)
} }
} }
} }
} }
func (m *LustreCollector) Close() { func (m *LustreCollector) Close() {
return return
} }

8
collectors/memstatMetric.go
View File

@ -7,7 +7,6 @@ import (
"strconv" "strconv"
"strings" "strings"
"time" "time"
//protocol "github.com/influxdata/line-protocol" //protocol "github.com/influxdata/line-protocol"
) )
@ -17,13 +16,12 @@ type MemstatCollector struct {
MetricCollector MetricCollector
} }
func (m *MemstatCollector) Init() { func (m *MemstatCollector) Init() {
m.name = "MemstatCollector" m.name = "MemstatCollector"
m.setup() m.setup()
} }
func (m *MemstatCollector) Read(interval time.Duration){ func (m *MemstatCollector) Read(interval time.Duration) {
buffer, err := ioutil.ReadFile(string(MEMSTATFILE)) buffer, err := ioutil.ReadFile(string(MEMSTATFILE))
if err != nil { if err != nil {
@ -53,5 +51,5 @@ func (m *MemstatCollector) Read(interval time.Duration){
} }
func (m *MemstatCollector) Close() { func (m *MemstatCollector) Close() {
return return
} }

109
collectors/metricCollector.go
View File

@ -1,15 +1,15 @@
package collectors package collectors
import ( import (
"time"
"io/ioutil" "io/ioutil"
"strings"
"log" "log"
"strconv" "strconv"
"strings"
"time"
) )
type MetricGetter interface { type MetricGetter interface {
Name() string Name() string
Init() Init()
Read(time.Duration) Read(time.Duration)
Close() Close()
@ -19,14 +19,12 @@ type MetricGetter interface {
} }
type MetricCollector struct { type MetricCollector struct {
name string name string
node map[string]interface{} node map[string]interface{}
sockets map[int]map[string]interface{} sockets map[int]map[string]interface{}
cpus map[int]map[string]interface{} cpus map[int]map[string]interface{}
} }
func (c *MetricCollector) Name() string { func (c *MetricCollector) Name() string {
return c.name return c.name
} }
@ -44,76 +42,75 @@ func (c *MetricCollector) GetCpuMetrics() map[int]map[string]interface{} {
} }
func (c *MetricCollector) setup() error { func (c *MetricCollector) setup() error {
slist := SocketList() slist := SocketList()
clist := CpuList() clist := CpuList()
c.node = make(map[string]interface{}) c.node = make(map[string]interface{})
c.sockets = make(map[int]map[string]interface{}, len(slist)) c.sockets = make(map[int]map[string]interface{}, len(slist))
for _, s := range slist { for _, s := range slist {
c.sockets[s] = make(map[string]interface{}) c.sockets[s] = make(map[string]interface{})
} }
c.cpus = make(map[int]map[string]interface{}, len(clist)) c.cpus = make(map[int]map[string]interface{}, len(clist))
for _, s := range clist { for _, s := range clist {
c.cpus[s] = make(map[string]interface{}) c.cpus[s] = make(map[string]interface{})
} }
return nil return nil
} }
func intArrayContains(array []int, str int) (int, bool) { func intArrayContains(array []int, str int) (int, bool) {
for i, a := range array { for i, a := range array {
if a == str { if a == str {
return i, true return i, true
} }
} }
return -1, false return -1, false
} }
func SocketList() []int { func SocketList() []int {
buffer, err := ioutil.ReadFile("/proc/cpuinfo") buffer, err := ioutil.ReadFile("/proc/cpuinfo")
if err != nil { if err != nil {
log.Print(err) log.Print(err)
return nil return nil
} }
ll := strings.Split(string(buffer), "\n") ll := strings.Split(string(buffer), "\n")
var packs []int var packs []int
for _, line := range ll { for _, line := range ll {
if strings.HasPrefix(line, "physical id") { if strings.HasPrefix(line, "physical id") {
lv := strings.Fields(line) lv := strings.Fields(line)
id, err := strconv.ParseInt(lv[3], 10, 32) id, err := strconv.ParseInt(lv[3], 10, 32)
if err != nil { if err != nil {
log.Print(err) log.Print(err)
return packs return packs
} }
_, found := intArrayContains(packs, int(id)) _, found := intArrayContains(packs, int(id))
if !found { if !found {
packs = append(packs, int(id)) packs = append(packs, int(id))
} }
} }
} }
return packs return packs
} }
func CpuList() []int { func CpuList() []int {
buffer, err := ioutil.ReadFile("/proc/cpuinfo") buffer, err := ioutil.ReadFile("/proc/cpuinfo")
if err != nil { if err != nil {
log.Print(err) log.Print(err)
return nil return nil
} }
ll := strings.Split(string(buffer), "\n") ll := strings.Split(string(buffer), "\n")
var cpulist []int var cpulist []int
for _, line := range ll { for _, line := range ll {
if strings.HasPrefix(line, "processor") { if strings.HasPrefix(line, "processor") {
lv := strings.Fields(line) lv := strings.Fields(line)
id, err := strconv.ParseInt(lv[2], 10, 32) id, err := strconv.ParseInt(lv[2], 10, 32)
if err != nil { if err != nil {
log.Print(err) log.Print(err)
return cpulist return cpulist
} }
_, found := intArrayContains(cpulist, int(id)) _, found := intArrayContains(cpulist, int(id))
if !found { if !found {
cpulist = append(cpulist, int(id)) cpulist = append(cpulist, int(id))
} }
} }
} }
return cpulist return cpulist
} }

13
collectors/netstatMetric.go
View File

@ -1,12 +1,12 @@
package collectors package collectors
import ( import (
"fmt"
"io/ioutil" "io/ioutil"
"log"
"strconv" "strconv"
"strings" "strings"
"time" "time"
"log"
"fmt"
) )
const NETSTATFILE = `/proc/net/dev` const NETSTATFILE = `/proc/net/dev`
@ -15,13 +15,12 @@ type NetstatCollector struct {
MetricCollector MetricCollector
} }
func (m *NetstatCollector) Init() { func (m *NetstatCollector) Init() {
m.name = "NetstatCollector" m.name = "NetstatCollector"
m.setup() m.setup()
} }
func (m *NetstatCollector) Read(interval time.Duration){ func (m *NetstatCollector) Read(interval time.Duration) {
data, err := ioutil.ReadFile(string(NETSTATFILE)) data, err := ioutil.ReadFile(string(NETSTATFILE))
if err != nil { if err != nil {
log.Print(err.Error()) log.Print(err.Error())
@ -51,9 +50,9 @@ func (m *NetstatCollector) Read(interval time.Duration){
} }
} }
} }
} }
func (m *NetstatCollector) Close() { func (m *NetstatCollector) Close() {
return return
} }

175
metric-collector.go
View File

@ -3,23 +3,24 @@ package main
import ( import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"github.com/ClusterCockpit/cc-metric-collector/collectors"
protocol "github.com/influxdata/line-protocol"
"log" "log"
"os" "os"
"os/signal" "os/signal"
"sync" "sync"
"time" "time"
"github.com/ClusterCockpit/cc-metric-collector/collectors"
protocol "github.com/influxdata/line-protocol"
) )
var Collectors = map[string]collectors.MetricGetter{ var Collectors = map[string]collectors.MetricGetter{
"likwid": &collectors.LikwidCollector{}, "likwid": &collectors.LikwidCollector{},
"loadavg": &collectors.LoadavgCollector{}, "loadavg": &collectors.LoadavgCollector{},
"memstat": &collectors.MemstatCollector{}, "memstat": &collectors.MemstatCollector{},
"netstat": &collectors.NetstatCollector{}, "netstat": &collectors.NetstatCollector{},
"ibstat": &collectors.InfinibandCollector{}, "ibstat": &collectors.InfinibandCollector{},
"lustrestat": &collectors.LustreCollector{}, "lustrestat": &collectors.LustreCollector{},
} }
type GlobalConfig struct { type GlobalConfig struct {
Sink struct { Sink struct {
User string `json:"user"` User string `json:"user"`
@ -32,8 +33,6 @@ type GlobalConfig struct {
Collectors []string `json:"collectors"` Collectors []string `json:"collectors"`
} }
func LoadConfiguration(file string, config *GlobalConfig) error { func LoadConfiguration(file string, config *GlobalConfig) error {
configFile, err := os.Open(file) configFile, err := os.Open(file)
defer configFile.Close() defer configFile.Close()
@ -45,8 +44,6 @@ func LoadConfiguration(file string, config *GlobalConfig) error {
return err return err
} }
func shutdown(wg *sync.WaitGroup, config *GlobalConfig) { func shutdown(wg *sync.WaitGroup, config *GlobalConfig) {
sigs := make(chan os.Signal, 1) sigs := make(chan os.Signal, 1)
signal.Notify(sigs, os.Interrupt) signal.Notify(sigs, os.Interrupt)
@ -69,23 +66,23 @@ func main() {
var wg sync.WaitGroup var wg sync.WaitGroup
wg.Add(1) wg.Add(1)
host, err := os.Hostname() host, err := os.Hostname()
if err != nil { if err != nil {
log.Print(err) log.Print(err)
return return
} }
var tags = map[string]string {"host": host} var tags = map[string]string{"host": host}
LoadConfiguration("config.json", &config) LoadConfiguration("config.json", &config)
if config.Interval <= 0 || time.Duration(config.Interval) * time.Second <= 0 { if config.Interval <= 0 || time.Duration(config.Interval)*time.Second <= 0 {
log.Print("Configuration value 'interval' must be greater than zero") log.Print("Configuration value 'interval' must be greater than zero")
return return
} }
if config.Duration <= 0 { if config.Duration <= 0 {
log.Print("Configuration value 'duration' must be greater than zero") log.Print("Configuration value 'duration' must be greater than zero")
return return
} }
shutdown(&wg, &config) shutdown(&wg, &config)
serializer := protocol.NewEncoder(os.Stdout) serializer := protocol.NewEncoder(os.Stdout)
serializer.SetPrecision(time.Second) serializer.SetPrecision(time.Second)
serializer.SetMaxLineBytes(1024) serializer.SetMaxLineBytes(1024)
for _, c := range config.Collectors { for _, c := range config.Collectors {
@ -93,91 +90,89 @@ func main() {
col.Init() col.Init()
log.Print("Start ", col.Name()) log.Print("Start ", col.Name())
} }
log.Print(config.Interval, time.Duration(config.Interval) * time.Second) log.Print(config.Interval, time.Duration(config.Interval)*time.Second)
ticker := time.NewTicker(time.Duration(config.Interval) * time.Second) ticker := time.NewTicker(time.Duration(config.Interval) * time.Second)
done := make(chan bool) done := make(chan bool)
node_fields := make(map[string]interface{}) node_fields := make(map[string]interface{})
slist := collectors.SocketList() slist := collectors.SocketList()
sockets_fields := make(map[int]map[string]interface{}, len(slist)) sockets_fields := make(map[int]map[string]interface{}, len(slist))
for _, s := range slist { for _, s := range slist {
sockets_fields[s] = make(map[string]interface{}) sockets_fields[s] = make(map[string]interface{})
} }
clist := collectors.CpuList() clist := collectors.CpuList()
cpu_fields := make(map[int]map[string]interface{}, len(clist)) cpu_fields := make(map[int]map[string]interface{}, len(clist))
for _, s := range clist { for _, s := range clist {
cpu_fields[s] = make(map[string]interface{}) cpu_fields[s] = make(map[string]interface{})
} }
go func() { go func() {
for { for {
select { select {
case <-done: case <-done:
return return
case t:= <-ticker.C: case t := <-ticker.C:
scount := 0
scount := 0 ccount := 0
ccount := 0 for _, c := range config.Collectors {
for _, c := range config.Collectors { col := Collectors[c]
col := Collectors[c] col.Read(time.Duration(config.Duration))
col.Read(time.Duration(config.Duration)) for key, val := range col.GetNodeMetric() {
for key, val := range col.GetNodeMetric() { node_fields[key] = val
node_fields[key] = val }
} for sid, socket := range col.GetSocketMetrics() {
for sid, socket := range col.GetSocketMetrics() { for key, val := range socket {
for key, val := range socket { sockets_fields[sid][key] = val
sockets_fields[sid][key] = val scount++
scount++ }
} }
} for cid, cpu := range col.GetCpuMetrics() {
for cid, cpu := range col.GetCpuMetrics() { for key, val := range cpu {
for key, val := range cpu { cpu_fields[cid][key] = val
cpu_fields[cid][key] = val ccount++
ccount++ }
} }
}
} }
var CurrentNode protocol.MutableMetric var CurrentNode protocol.MutableMetric
CurrentNode, err = protocol.New("node", tags, node_fields, t) CurrentNode, err = protocol.New("node", tags, node_fields, t)
if err != nil { if err != nil {
log.Print(err) log.Print(err)
} }
_, err := serializer.Encode(CurrentNode) _, err := serializer.Encode(CurrentNode)
if err != nil { if err != nil {
log.Print(err) log.Print(err)
} }
if scount > 0 { if scount > 0 {
for sid, socket := range sockets_fields { for sid, socket := range sockets_fields {
var CurrentSocket protocol.MutableMetric var CurrentSocket protocol.MutableMetric
var stags = map[string]string {"socket": fmt.Sprintf("%d", sid), "host": host} var stags = map[string]string{"socket": fmt.Sprintf("%d", sid), "host": host}
CurrentSocket, err = protocol.New("socket", stags, socket, t) CurrentSocket, err = protocol.New("socket", stags, socket, t)
if err != nil { if err != nil {
log.Print(err) log.Print(err)
} }
_, err := serializer.Encode(CurrentSocket) _, err := serializer.Encode(CurrentSocket)
if err != nil { if err != nil {
log.Print(err) log.Print(err)
} }
} }
} }
if ccount > 0 { if ccount > 0 {
for cid, cpu := range cpu_fields { for cid, cpu := range cpu_fields {
var CurrentCpu protocol.MutableMetric var CurrentCpu protocol.MutableMetric
var ctags = map[string]string {"host": host, "cpu": fmt.Sprintf("%d", cid)} var ctags = map[string]string{"host": host, "cpu": fmt.Sprintf("%d", cid)}
CurrentCpu, err = protocol.New("cpu", ctags, cpu, t) CurrentCpu, err = protocol.New("cpu", ctags, cpu, t)
if err != nil { if err != nil {
log.Print(err) log.Print(err)
} }
_, err := serializer.Encode(CurrentCpu) _, err := serializer.Encode(CurrentCpu)
if err != nil { if err != nil {
log.Print(err) log.Print(err)
} }
} }
} }
} }
} }
}() }()
wg.Wait() wg.Wait()
} }