Merge pull request #37 from ClusterCockpit/dynload_libganglia

Dynload libganglia

sinks/Makefile

@@ -2,9 +2,10 @@
 all: libganglia.so
 
 libganglia.so:
-	@find /usr -readable -name "libganglia.so*" -print0 | \
+	@find /usr ! -readable -prune -o -type d ! -executable -prune -o -name "$@*" -print0 | \
 	    xargs --null --no-run-if-empty --replace \
-	        ln --symbolic --verbose --force '{}' libganglia.so
+	        ln --symbolic --verbose --force '{}' "$@"
+	@if [[ ! -e "$@" ]]; then touch "$@"; fi
 
 
 clean:

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

@@ -1,6 +1,3 @@
-//go:build ganglia
-// +build ganglia
-
 package sinks
 
 import (
@@ -25,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 {
@@ -85,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:
@@ -96,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:
@@ -105,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/gangliaSink_disabled.go

@@ -1,32 +0,0 @@
-//go:build !ganglia
-// +build !ganglia
-
-package sinks
-
-import (
-	"encoding/json"
-	"errors"
-
-	//	"time"
-
-	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-)
-
-type GangliaSink struct {
-	sink
-}
-
-func (s *GangliaSink) Init(config json.RawMessage) error {
-	return errors.New("sink 'ganglia' not implemented, rebuild with tag 'ganglia'")
-}
-
-func (s *GangliaSink) Write(point lp.CCMetric) error {
-	return errors.New("sink 'ganglia' not implemented, rebuild with tag 'ganglia'")
-}
-
-func (s *GangliaSink) Flush() error {
-	return errors.New("sink 'ganglia' not implemented, rebuild with tag 'ganglia'")
-}
-
-func (s *GangliaSink) Close() {
-}

sinks/libgangliaSink.go

@@ -1,11 +1,8 @@
-//go:build ganglia
-// +build ganglia
-
 package sinks
 
 /*
 #cgo CFLAGS: -DGM_PROTOCOL_GUARD
-#cgo LDFLAGS: -L. -lganglia
+#cgo LDFLAGS: -L. -lganglia -Wl,--unresolved-symbols=ignore-in-object-files
 #include <stdlib.h>
 
 // This is a copy&paste snippet of ganglia.h (BSD-3 license)
@@ -72,22 +69,34 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"strings"
 	"unsafe"
 
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
+	"github.com/NVIDIA/go-nvml/pkg/dl"
 )
 
-const GMOND_CONFIG_FILE = `/etc/ganglia/gmond.conf`
+const (
+	GANGLIA_LIB_NAME     = "libganglia.so"
+	GANGLIA_LIB_DL_FLAGS = dl.RTLD_LAZY | dl.RTLD_GLOBAL
+	GMOND_CONFIG_FILE    = `/etc/ganglia/gmond.conf`
+)
+
+type LibgangliaSinkSpecialMetric struct {
+	MetricName string `json:"metric_name,omitempty"`
+	NewName    string `json:"new_name,omitempty"`
+	Slope      string `json:"slope,omitempty"`
+}
 
 type LibgangliaSinkConfig struct {
 	defaultSinkConfig
+	GangliaLib      string                                 `json:"libganglia_path,omitempty"`
 	GmondConfig     string                                 `json:"gmond_config,omitempty"`
 	AddGangliaGroup bool                                   `json:"add_ganglia_group,omitempty"`
-	//AddTagsAsDesc   bool   `json:"add_tags_as_desc,omitempty"`
 	AddTypeToName   bool                                   `json:"add_type_to_name,omitempty"`
 	AddUnits        bool                                   `json:"add_units,omitempty"`
 	ClusterName     string                                 `json:"cluster_name,omitempty"`
+	SpecialMetrics  map[string]LibgangliaSinkSpecialMetric `json:"rename_metrics,omitempty"` // Map to rename metric name from key to value
+	//AddTagsAsDesc   bool              `json:"add_tags_as_desc,omitempty"`
 }
 
 type LibgangliaSink struct {
@@ -99,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"
@@ -124,6 +116,7 @@ func (s *LibgangliaSink) Init(config json.RawMessage) error {
 	s.config.AddTypeToName = false
 	s.config.AddUnits = true
 	s.config.GmondConfig = string(GMOND_CONFIG_FILE)
+	s.config.GangliaLib = string(GANGLIA_LIB_NAME)
 	if len(config) > 0 {
 		err = json.Unmarshal(config, &s.config)
 		if err != nil {
@@ -131,6 +124,10 @@ func (s *LibgangliaSink) Init(config json.RawMessage) error {
 			return err
 		}
 	}
+	lib := dl.New(s.config.GangliaLib, GANGLIA_LIB_DL_FLAGS)
+	if lib == nil {
+		return fmt.Errorf("error instantiating DynamicLibrary for %s", s.config.GangliaLib)
+	}
 
 	// Set up cache for the C strings
 	s.cstrCache = make(map[string]*C.char)
@@ -182,16 +179,17 @@ func (s *LibgangliaSink) Write(point lp.CCMetric) error {
 	}
 
 	// Get metric name
+	metricname := GangliaMetricRename(point)
 	if s.config.AddTypeToName {
-		c_name = lookup(gangliaMetricName(point))
+		c_name = lookup(GangliaMetricName(point))
 	} else {
-		c_name = lookup(point.Name())
+		c_name = lookup(metricname)
 	}
 
 	// Get the value C string and lookup the type string in the cache
 	value, ok := point.GetField("value")
 	if !ok {
-		return fmt.Errorf("metric %s has no 'value' field", point.Name())
+		return fmt.Errorf("metric %s has no 'value' field", metricname)
 	}
 	switch real := value.(type) {
 	case float64:
@@ -229,14 +227,24 @@ func (s *LibgangliaSink) Write(point lp.CCMetric) error {
 		c_unit = lookup("")
 	}
 
+	// 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 slope {
+	case 0:
+		slope_type = C.GANGLIA_SLOPE_ZERO
+	}
+
 	// Create a new Ganglia metric
 	gmetric := C.Ganglia_metric_create(s.global_context)
-	rval := C.int(0)
 	// Set name, value, type and unit in the Ganglia metric
 	// Since we don't have this information from the collectors,
 	// we assume that the metric value can go up and down (slope),
-	// and their is no maximum for 'dmax' and 'tmax'
+	// and there is no maximum for 'dmax' and 'tmax'.
-	rval = C.Ganglia_metric_set(gmetric, c_name, c_value, c_type, c_unit, C.GANGLIA_SLOPE_BOTH, 0, 0)
+	// Ganglia's collectors set 'tmax' but not 'dmax'
+	rval := C.int(0)
+	rval = C.Ganglia_metric_set(gmetric, c_name, c_value, c_type, c_unit, C.uint(slope_type), 0, 0)
 	switch rval {
 	case 1:
 		C.free(unsafe.Pointer(c_value))

sinks/libgangliaSink_disabled.go

@@ -1,30 +0,0 @@
-//go:build !ganglia
-// +build !ganglia
-
-package sinks
-
-import (
-	"encoding/json"
-	"errors"
-
-	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-)
-
-type LibgangliaSink struct {
-	sink
-}
-
-func (s *LibgangliaSink) Init(config json.RawMessage) error {
-	return errors.New("sink 'libganglia' not implemented, rebuild with tag 'ganglia'")
-}
-
-func (s *LibgangliaSink) Write(point lp.CCMetric) error {
-	return errors.New("sink 'libganglia' not implemented, rebuild with tag 'ganglia'")
-}
-
-func (s *LibgangliaSink) Flush() error {
-	return errors.New("sink 'ganglia' not implemented, rebuild with tag 'ganglia'")
-}
-
-func (s *LibgangliaSink) Close() {
-}