Use common functions in both Ganglia sinks

sinks/gangliaCommon.go

@@ -0,0 +1,50 @@
+package sinks
+
+import (
+	"strings"
+
+	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
+)
+
+func GangliaMetricName(point lp.CCMetric) string {
+	name := point.Name()
+	metricType, typeOK := point.GetTag("type")
+	metricTid, tidOk := point.GetTag("type-id")
+	gangliaType := metricType + metricTid
+	if strings.Contains(name, metricType) && tidOk {
+		name = strings.Replace(name, metricType, gangliaType, -1)
+	} else if typeOK && tidOk {
+		name = metricType + metricTid + "_" + name
+	} else if point.HasTag("device") {
+		device, _ := point.GetTag("device")
+		name = name + "_" + device
+	}
+
+	return name
+}
+
+func GangliaMetricRename(point lp.CCMetric) string {
+	name := point.Name()
+	if name == "mem_total" || name == "swap_total" {
+		return name
+	} else if name == "net_bytes_in" {
+		return "bytes_in"
+	} else if name == "net_bytes_out" {
+		return "bytes_out"
+	} else if name == "net_pkts_in" {
+		return "pkts_in"
+	} else if name == "net_pkts_out" {
+		return "pkts_out"
+	} else if name == "cpu_iowait" {
+		return "cpu_wio"
+	}
+	return name
+}
+
+func GangliaSlopeType(point lp.CCMetric) uint {
+	name := point.Name()
+	if name == "mem_total" || name == "swap_total" {
+		return 0
+	}
+	return 3
+}

sinks/gangliaSink.go

@@ -22,6 +22,8 @@ type GangliaSinkConfig struct {
 	GmetricConfig   string `json:"gmetric_config,omitempty"`
 	AddGangliaGroup bool   `json:"add_ganglia_group,omitempty"`
 	AddTagsAsDesc   bool   `json:"add_tags_as_desc,omitempty"`
+	ClusterName     string `json:"cluster_name,omitempty"`
+	AddTypeToName   bool   `json:"add_type_to_name,omitempty"`
 }
 
 type GangliaSink struct {
@@ -82,8 +84,6 @@ func (s *GangliaSink) Write(point lp.CCMetric) error {
 
 	for key, value := range point.Tags() {
 		switch key {
-		case "cluster":
-			argstr = append(argstr, fmt.Sprintf("--cluster=%s", value))
 		case "unit":
 			argstr = append(argstr, fmt.Sprintf("--units=%s", value))
 		default:
@@ -93,8 +93,6 @@ func (s *GangliaSink) Write(point lp.CCMetric) error {
 	if s.config.MetaAsTags {
 		for key, value := range point.Meta() {
 			switch key {
-			case "cluster":
-				argstr = append(argstr, fmt.Sprintf("--cluster=%s", value))
 			case "unit":
 				argstr = append(argstr, fmt.Sprintf("--units=%s", value))
 			default:
@@ -102,13 +100,28 @@ func (s *GangliaSink) Write(point lp.CCMetric) error {
 			}
 		}
 	}
+	if len(s.config.ClusterName) > 0 {
+		argstr = append(argstr, fmt.Sprintf("--cluster=%s", s.config.ClusterName))
+	}
 	if s.config.AddTagsAsDesc && len(tagsstr) > 0 {
 		argstr = append(argstr, fmt.Sprintf("--desc=%q", strings.Join(tagsstr, ",")))
 	}
 	if len(s.gmetric_config) > 0 {
 		argstr = append(argstr, fmt.Sprintf("--conf=%s", s.gmetric_config))
 	}
-	argstr = append(argstr, fmt.Sprintf("--name=%s", point.Name()))
+	name := GangliaMetricRename(point)
+	if s.config.AddTypeToName {
+		argstr = append(argstr, fmt.Sprintf("--name=%s", GangliaMetricName(point)))
+	} else {
+		argstr = append(argstr, fmt.Sprintf("--name=%s", name))
+	}
+	slope := GangliaSlopeType(point)
+	slopeStr := "both"
+	if slope == 0 {
+		slopeStr = "zero"
+	}
+	argstr = append(argstr, fmt.Sprintf("--slope=%s", slopeStr))
+
 	for k, v := range point.Fields() {
 		if k == "value" {
 			switch value := v.(type) {

sinks/libgangliaSink.go

@@ -69,7 +69,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"strings"
 	"unsafe"
 
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
@@ -109,23 +108,6 @@ type LibgangliaSink struct {
 	cstrCache      map[string]*C.char
 }
 
-func gangliaMetricName(point lp.CCMetric) string {
-	name := point.Name()
-	metricType, typeOK := point.GetTag("type")
-	metricTid, tidOk := point.GetTag("type-id")
-	gangliaType := metricType + metricTid
-	if strings.Contains(name, metricType) && tidOk {
-		name = strings.Replace(name, metricType, gangliaType, -1)
-	} else if typeOK && tidOk {
-		name = metricType + metricTid + "_" + name
-	} else if point.HasTag("device") {
-		device, _ := point.GetTag("device")
-		name = name + "_" + device
-	}
-
-	return name
-}
-
 func (s *LibgangliaSink) Init(config json.RawMessage) error {
 	var err error = nil
 	s.name = "LibgangliaSink"
@@ -197,9 +179,9 @@ func (s *LibgangliaSink) Write(point lp.CCMetric) error {
 	}
 
 	// Get metric name
-	metricname := point.Name()
+	metricname := GangliaMetricRename(point)
 	if s.config.AddTypeToName {
-		c_name = lookup(gangliaMetricName(point))
+		c_name = lookup(GangliaMetricName(point))
 	} else {
 		c_name = lookup(metricname)
 	}
@@ -247,11 +229,10 @@ func (s *LibgangliaSink) Write(point lp.CCMetric) error {
 
 	// Determine the slope of the metric. Ganglia's own collector mostly use
 	// 'both' but the mem and swap total uses 'zero'.
+	slope := GangliaSlopeType(point)
 	slope_type := C.GANGLIA_SLOPE_BOTH
-	switch metricname {
-	case "mem_total":
-		slope_type = C.GANGLIA_SLOPE_ZERO
-	case "swap_total":
+	switch slope {
+	case 0:
 		slope_type = C.GANGLIA_SLOPE_ZERO
 	}