Add collectors for custom commands and /proc/diskstat. Per default add a hostname tag to all measurements

collectors/customCmdMetric.go

@@ -0,0 +1,108 @@
+package collectors
+
+import (
+	"fmt"
+	lp "github.com/influxdata/line-protocol"
+	"io/ioutil"
+	"log"
+	"os/exec"
+	"time"
+)
+
+const CUSTOMCMDPATH = `/home/unrz139/Work/cc-metric-collector/collectors/custom`
+
+type CustomCmdCollector struct {
+	MetricCollector
+	handler *lp.MetricHandler
+	parser  *lp.Parser
+}
+
+func (m *CustomCmdCollector) Init() error {
+	m.name = "CustomCmdCollector"
+	m.setup()
+	m.handler = lp.NewMetricHandler()
+	m.parser = lp.NewParser(m.handler)
+	m.parser.SetTimeFunc(DefaultTime)
+	m.init = true
+	return nil
+}
+
+var DefaultTime = func() time.Time {
+	return time.Unix(42, 0)
+}
+
+func (m *CustomCmdCollector) Read(interval time.Duration, out *[]lp.MutableMetric) {
+	files, err := ioutil.ReadDir(string(CUSTOMCMDPATH))
+	if err != nil {
+		log.Print(err)
+		return
+	}
+	for _, file := range files {
+		//		stat, err := os.Stat(file)
+		//		if err != nil {
+		//			log.Print(err)
+		//			continue
+		//		}
+		//		mode := stat.Mode()
+		//		if mode & 0o555 {
+		path := fmt.Sprintf("%s/%s", string(CUSTOMCMDPATH), file.Name())
+		command := exec.Command(path, "")
+		command.Wait()
+		stdout, err := command.Output()
+		if err != nil {
+			log.Print(err)
+			continue
+		}
+		metrics, err := m.parser.Parse(stdout)
+		if err != nil {
+			log.Print(err)
+			continue
+		}
+		for _, m := range metrics {
+			y, err := lp.New(m.Name(), Tags2Map(m), Fields2Map(m), m.Time())
+			if err == nil {
+				*out = append(*out, y)
+			}
+			//				switch m.Name() {
+			//				case "node":
+			//					for k, v := range m.FieldList() {
+			//						m.node[k] = float64(v)
+			//					}
+			//				case "socket":
+			//					tlist := m.TagList()
+			//					if id, found := tlist["socket"]; found {
+			//						for k, v := range m.FieldList() {
+			//							m.socket[id][k] = float64(v)
+			//						}
+			//					}
+			//				case "cpu":
+			//					tlist := m.TagList()
+			//					if id, found := tlist["cpu"]; found {
+			//						for k, v := range m.FieldList() {
+			//							m.cpu[id][k] = float64(v)
+			//						}
+			//					}
+			//				case "network":
+			//					tlist := m.TagList()
+			//					if id, found := tlist["device"]; found {
+			//						for k, v := range m.FieldList() {
+			//							m.network[id][k] = float64(v)
+			//						}
+			//					}
+			//				case "accelerator":
+			//					tlist := m.TagList()
+			//					if id, found := tlist["device"]; found {
+			//						for k, v := range m.FieldList() {
+			//							m.accelerator[id][k] = float64(v)
+			//						}
+			//					}
+			//				}
+		}
+		//		} if file is executable check
+	}
+}
+
+func (m *CustomCmdCollector) Close() {
+	m.init = false
+	return
+}

collectors/diskstatMetric.go

@@ -0,0 +1,90 @@
+package collectors
+
+import (
+	//	"errors"
+	//	"fmt"
+	lp "github.com/influxdata/line-protocol"
+	"io/ioutil"
+	"log"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const DISKSTATFILE = `/proc/diskstats`
+
+type DiskstatCollector struct {
+	MetricCollector
+	matches map[int]string
+}
+
+func (m *DiskstatCollector) Init() error {
+	m.name = "DiskstatCollector"
+	m.setup()
+	// https://www.kernel.org/doc/html/latest/admin-guide/iostats.html
+	m.matches = map[int]string{
+		3:  "reads",
+		4:  "reads_merged",
+		5:  "read_sectors",
+		6:  "read_ms",
+		7:  "writes",
+		8:  "writes_merged",
+		9:  "writes_sectors",
+		10: "writes_ms",
+		11: "ioops",
+		12: "ioops_ms",
+		13: "ioops_weighted_ms",
+		14: "discards",
+		15: "discards_merged",
+		16: "discards_sectors",
+		17: "discards_ms",
+		18: "flushes",
+		19: "flushes_ms",
+	}
+	_, err := ioutil.ReadFile(string(DISKSTATFILE))
+	if err == nil {
+		m.init = true
+	}
+	return err
+}
+
+func (m *DiskstatCollector) Read(interval time.Duration, out *[]lp.MutableMetric) {
+
+	buffer, err := ioutil.ReadFile(string(DISKSTATFILE))
+
+	if err != nil {
+		log.Print(err)
+		return
+	}
+
+	ll := strings.Split(string(buffer), "\n")
+
+	for _, line := range ll {
+		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 {
+			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
+}
+
+func (m *DiskstatCollector) Close() {
+	m.init = false
+	return
+}

metric-collector.go

@@ -28,6 +28,8 @@ var Collectors = map[string]collectors.MetricGetter{
 	"cpustat":    &collectors.CpustatCollector{},
 	"topprocs":   &collectors.TopProcsCollector{},
 	"nvidia":     &collectors.NvidiaCollector{},
+	"customcmd":  &collectors.CustomCmdCollector{},
+	"diskstat":   &collectors.DiskstatCollector{},
 }
 
 var Sinks = map[string]sinks.SinkFuncs{
@@ -226,6 +228,7 @@ func main() {
 		}
 	}
 	config.Collectors = tmp
+	config.DefTags["hostname"] = host
 
 	// Setup up ticker loop
 	log.Print("Running loop every ", time.Duration(config.Interval)*time.Second)