ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2025-04-08 14:35:55 +02:00
Code Issues Packages Projects Releases Wiki Activity

Rename new Ganglia sink to 'libgangliaSink'

Browse Source
This commit is contained in:
Thomas Roehl 2022-02-16 03:04:17 +01:00
parent ef4fec954c
commit 215b954a6b
2 changed files with 106 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

165
sinks/gangliaSink2.go → sinks/libgangliaSink.go
View File

@ -4,6 +4,11 @@ package sinks
#cgo CFLAGS: -DGM_PROTOCOL_GUARD #cgo CFLAGS: -DGM_PROTOCOL_GUARD
#cgo LDFLAGS: -L. -lganglia #cgo LDFLAGS: -L. -lganglia
#include <stdlib.h> #include <stdlib.h>
// This is a copy&paste snippet of ganglia.h (BSD-3 license)
// See https://github.com/ganglia/monitor-core
// for further information
enum ganglia_slope { enum ganglia_slope {
GANGLIA_SLOPE_ZERO = 0, GANGLIA_SLOPE_ZERO = 0,
GANGLIA_SLOPE_POSITIVE, GANGLIA_SLOPE_POSITIVE,
@ -31,12 +36,10 @@ typedef struct Ganglia_metric * Ganglia_metric;
extern "C" { extern "C" {
#endif #endif
Ganglia_gmond_config Ganglia_gmond_config Ganglia_gmond_config_create(char *path, int fallback_to_default);
Ganglia_gmond_config_create(char *path, int fallback_to_default); //void Ganglia_gmond_config_destroy(Ganglia_gmond_config config);
void Ganglia_gmond_config_destroy(Ganglia_gmond_config config);
Ganglia_udp_send_channels Ganglia_udp_send_channels Ganglia_udp_send_channels_create(Ganglia_pool p, Ganglia_gmond_config config);
Ganglia_udp_send_channels_create(Ganglia_pool p, Ganglia_gmond_config config);
void Ganglia_udp_send_channels_destroy(Ganglia_udp_send_channels channels); void Ganglia_udp_send_channels_destroy(Ganglia_udp_send_channels channels);
int Ganglia_udp_send_message(Ganglia_udp_send_channels channels, char *buf, int len ); int Ganglia_udp_send_message(Ganglia_udp_send_channels channels, char *buf, int len );
@ -44,18 +47,18 @@ int Ganglia_udp_send_message(Ganglia_udp_send_channels channels, char *buf, int
Ganglia_metric Ganglia_metric_create( Ganglia_pool parent_pool ); Ganglia_metric Ganglia_metric_create( Ganglia_pool parent_pool );
int Ganglia_metric_set( Ganglia_metric gmetric, char *name, char *value, char *type, char *units, unsigned int slope, unsigned int tmax, unsigned int dmax); int Ganglia_metric_set( Ganglia_metric gmetric, char *name, char *value, char *type, char *units, unsigned int slope, unsigned int tmax, unsigned int dmax);
int Ganglia_metric_send( Ganglia_metric gmetric, Ganglia_udp_send_channels send_channels ); int Ganglia_metric_send( Ganglia_metric gmetric, Ganglia_udp_send_channels send_channels );
int Ganglia_metadata_send( Ganglia_metric gmetric, Ganglia_udp_send_channels send_channels ); //int Ganglia_metadata_send( Ganglia_metric gmetric, Ganglia_udp_send_channels send_channels );
int Ganglia_metadata_send_real( Ganglia_metric gmetric, Ganglia_udp_send_channels send_channels, char *override_string ); //int Ganglia_metadata_send_real( Ganglia_metric gmetric, Ganglia_udp_send_channels send_channels, char *override_string );
void Ganglia_metadata_add( Ganglia_metric gmetric, char *name, char *value ); void Ganglia_metadata_add( Ganglia_metric gmetric, char *name, char *value );
int Ganglia_value_send( Ganglia_metric gmetric, Ganglia_udp_send_channels send_channels ); //int Ganglia_value_send( Ganglia_metric gmetric, Ganglia_udp_send_channels send_channels );
void Ganglia_metric_destroy( Ganglia_metric gmetric ); void Ganglia_metric_destroy( Ganglia_metric gmetric );
Ganglia_pool Ganglia_pool_create( Ganglia_pool parent ); Ganglia_pool Ganglia_pool_create( Ganglia_pool parent );
void Ganglia_pool_destroy( Ganglia_pool pool ); void Ganglia_pool_destroy( Ganglia_pool pool );
ganglia_slope_t cstr_to_slope(const char* str); //ganglia_slope_t cstr_to_slope(const char* str);
const char* slope_to_cstr(unsigned int slope); //const char* slope_to_cstr(unsigned int slope);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif
@ -71,37 +74,35 @@ import (
lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric" lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
) )
const LIBGANGLIA = `libganglia.so` const GMOND_CONFIG_FILE = `/var/ganglia/gmond.conf`
type Ganglia2SinkConfig struct { type LibgangliaSinkConfig struct {
defaultSinkConfig defaultSinkConfig
GmetricPath string `json:"gmetricPath,omitempty"` GmondConfig string `json:"gmond_config,omitempty"`
GmetricConfig string `json:"gmetricConfig,omitempty"`
AddGangliaGroup bool `json:"add_ganglia_group,omitempty"` AddGangliaGroup bool `json:"add_ganglia_group,omitempty"`
AddTagsAsDesc bool `json:"add_tags_as_desc,omitempty"` //AddTagsAsDesc bool `json:"add_tags_as_desc,omitempty"`
AddTypeToName bool `json:"add_type_to_name,omitempty"` AddTypeToName bool `json:"add_type_to_name,omitempty"`
AddUnits bool `json:"add_units,omitempty"` AddUnits bool `json:"add_units,omitempty"`
ClusterName string `json:"cluster_name,omitempty"` ClusterName string `json:"cluster_name,omitempty"`
GangliaLib string `json:"libganglia_path,omitempty"`
ConfuseLib string `json:"libconfuse_path,omitempty"`
} }
type Ganglia2Sink struct { type LibgangliaSink struct {
sink sink
config LibgangliaSinkConfig
config Ganglia2SinkConfig
global_context C.Ganglia_pool global_context C.Ganglia_pool
gmond_config C.Ganglia_gmond_config
send_channels C.Ganglia_udp_send_channels send_channels C.Ganglia_udp_send_channels
constStr map[string]*C.char cstrCache map[string]*C.char
} }
func (s *Ganglia2Sink) Init(config json.RawMessage) error { func (s *LibgangliaSink) Init(config json.RawMessage) error {
var err error = nil var err error = nil
s.name = "Ganglia2Sink" s.name = "LibgangliaSink"
s.config.AddTagsAsDesc = false //s.config.AddTagsAsDesc = false
s.config.AddGangliaGroup = false s.config.AddGangliaGroup = false
s.config.AddTypeToName = false s.config.AddTypeToName = false
s.config.AddUnits = true s.config.AddUnits = true
s.config.GmondConfig = string(GMOND_CONFIG_FILE)
if len(config) > 0 { if len(config) > 0 {
err = json.Unmarshal(config, &s.config) err = json.Unmarshal(config, &s.config)
if err != nil { if err != nil {
@ -109,47 +110,67 @@ func (s *Ganglia2Sink) Init(config json.RawMessage) error {
return err return err
} }
} }
s.constStr = make(map[string]*C.char)
// s.constStr["globals"] = C.CString("globals") // Set up cache for the C strings
s.constStr["conffile"] = C.CString(s.config.GmetricConfig) s.cstrCache = make(map[string]*C.char)
// s.constStr["override_hostname"] = C.CString("override_hostname") // s.cstrCache["globals"] = C.CString("globals")
// s.constStr["override_ip"] = C.CString("override_ip")
s.constStr["GROUP"] = C.CString("GROUP") // s.cstrCache["override_hostname"] = C.CString("override_hostname")
s.constStr["CLUSTER"] = C.CString("CLUSTER") // s.cstrCache["override_ip"] = C.CString("override_ip")
// Add some constant strings
s.cstrCache["GROUP"] = C.CString("GROUP")
s.cstrCache["CLUSTER"] = C.CString("CLUSTER")
s.cstrCache[""] = C.CString("")
// Add cluster name for lookup in Write()
if len(s.config.ClusterName) > 0 { if len(s.config.ClusterName) > 0 {
s.constStr[s.config.ClusterName] = C.CString(s.config.ClusterName) s.cstrCache[s.config.ClusterName] = C.CString(s.config.ClusterName)
} }
s.constStr["double"] = C.CString("double") // Add supported types for later lookup in Write()
s.constStr["int32"] = C.CString("int32") s.cstrCache["double"] = C.CString("double")
s.constStr["string"] = C.CString("string") s.cstrCache["int32"] = C.CString("int32")
s.constStr[""] = C.CString("") s.cstrCache["string"] = C.CString("string")
// Create Ganglia pool
s.global_context = C.Ganglia_pool_create(nil) s.global_context = C.Ganglia_pool_create(nil)
gmond_config := C.Ganglia_gmond_config_create(s.constStr["conffile"], 0) // Load Ganglia configuration
//globals := C.cfg_getsec(gmond_config, s.constStr["globals"]) s.cstrCache[s.config.GmondConfig] = C.CString(s.config.GmondConfig)
//override_hostname := C.cfg_getstr(globals, s.constStr["override_hostname"]) s.gmond_config = C.Ganglia_gmond_config_create(s.cstrCache[s.config.GmondConfig], 0)
//override_ip := C.cfg_getstr(globals, s.constStr["override_ip"]) //globals := C.cfg_getsec(gmond_config, s.cstrCache["globals"])
//override_hostname := C.cfg_getstr(globals, s.cstrCache["override_hostname"])
//override_ip := C.cfg_getstr(globals, s.cstrCache["override_ip"])
s.send_channels = C.Ganglia_udp_send_channels_create(s.global_context, gmond_config) s.send_channels = C.Ganglia_udp_send_channels_create(s.global_context, s.gmond_config)
return nil return nil
} }
func (s *Ganglia2Sink) Write(point lp.CCMetric) error { func (s *LibgangliaSink) Write(point lp.CCMetric) error {
var err error = nil var err error = nil
var c_name *C.char
var c_value *C.char var c_value *C.char
var c_type *C.char var c_type *C.char
var c_unit *C.char var c_unit *C.char
// helper function for looking up C strings in the cache
lookup := func(key string) *C.char { lookup := func(key string) *C.char {
if _, exist := s.constStr[key]; !exist { if _, exist := s.cstrCache[key]; !exist {
s.constStr[key] = C.CString(key) s.cstrCache[key] = C.CString(key)
} }
return s.constStr[key] return s.cstrCache[key]
} }
c_name := lookup(point.Name())
// Get metric name
if s.config.AddTypeToName {
c_name = lookup(gangliaMetricName(point))
} else {
c_name = lookup(point.Name())
}
// Get the value C string and lookup the type string in the cache
value, ok := point.GetField("value") value, ok := point.GetField("value")
if !ok { if !ok {
return fmt.Errorf("metric %s has not 'value' field", point.Name()) return fmt.Errorf("metric %s has no 'value' field", point.Name())
} }
switch real := value.(type) { switch real := value.(type) {
case float64: case float64:
@ -171,19 +192,29 @@ func (s *Ganglia2Sink) Write(point lp.CCMetric) error {
c_value = C.CString(real) c_value = C.CString(real)
c_type = lookup("string") c_type = lookup("string")
default: default:
C.free(unsafe.Pointer(c_name))
return fmt.Errorf("metric %s has invalid 'value' type for %s", point.Name(), s.name) return fmt.Errorf("metric %s has invalid 'value' type for %s", point.Name(), s.name)
} }
if tagunit, tagok := point.GetTag("unit"); tagok {
c_unit = lookup(tagunit) // Add unit
} else if metaunit, metaok := point.GetMeta("unit"); metaok { if s.config.AddUnits {
c_unit = lookup(metaunit) if tagunit, tagok := point.GetTag("unit"); tagok {
c_unit = lookup(tagunit)
} else if metaunit, metaok := point.GetMeta("unit"); metaok {
c_unit = lookup(metaunit)
} else {
c_unit = lookup("")
}
} else { } else {
c_unit = lookup("") c_unit = lookup("")
} }
// Create a new Ganglia metric
gmetric := C.Ganglia_metric_create(s.global_context) gmetric := C.Ganglia_metric_create(s.global_context)
rval := C.int(0) 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'
rval = C.Ganglia_metric_set(gmetric, c_name, c_value, c_type, c_unit, C.GANGLIA_SLOPE_BOTH, 0, 0) rval = C.Ganglia_metric_set(gmetric, c_name, c_value, c_type, c_unit, C.GANGLIA_SLOPE_BOTH, 0, 0)
switch rval { switch rval {
case 1: case 1:
@ -200,31 +231,45 @@ func (s *Ganglia2Sink) Write(point lp.CCMetric) error {
return fmt.Errorf("the value parameter \"%s\" does not represent a number", C.GoString(c_value)) return fmt.Errorf("the value parameter \"%s\" does not represent a number", C.GoString(c_value))
default: default:
} }
// Set the cluster name, otherwise it takes it from the configuration file
if len(s.config.ClusterName) > 0 { if len(s.config.ClusterName) > 0 {
C.Ganglia_metadata_add(gmetric, lookup("CLUSTER"), lookup(s.config.ClusterName)) C.Ganglia_metadata_add(gmetric, lookup("CLUSTER"), lookup(s.config.ClusterName))
} }
if group, ok := point.GetMeta("group"); ok { // Set the group metadata in the Ganglia metric if configured
if group, ok := point.GetMeta("group"); ok && s.config.AddGangliaGroup {
c_group := lookup(group) c_group := lookup(group)
C.Ganglia_metadata_add(gmetric, lookup("GROUP"), c_group) C.Ganglia_metadata_add(gmetric, lookup("GROUP"), c_group)
} }
// Now we send the metric
// gmetric does provide some more options like description and other options
// but they are not provided by the collectors
rval = C.Ganglia_metric_send(gmetric, s.send_channels) rval = C.Ganglia_metric_send(gmetric, s.send_channels)
if rval != 0 { if rval != 0 {
err = fmt.Errorf("there was an error sending metric %s to %d of the send channels ", point.Name(), rval) err = fmt.Errorf("there was an error sending metric %s to %d of the send channels ", point.Name(), rval)
// fall throuph to use Ganglia_metric_destroy from common cleanup // fall throuph to use Ganglia_metric_destroy from common cleanup
} }
// Cleanup Ganglia metric
C.Ganglia_metric_destroy(gmetric) C.Ganglia_metric_destroy(gmetric)
// Free the value C string, the only one not stored in the cache
C.free(unsafe.Pointer(c_value)) C.free(unsafe.Pointer(c_value))
return err return err
} }
func (s *Ganglia2Sink) Flush() error { func (s *LibgangliaSink) Flush() error {
return nil return nil
} }
func (s *Ganglia2Sink) Close() { func (s *LibgangliaSink) Close() {
// Destroy Ganglia configration struct
// (not done by gmetric, I thought I am more clever but no...)
//C.Ganglia_gmond_config_destroy(s.gmond_config)
// Destroy Ganglia pool
C.Ganglia_pool_destroy(s.global_context) C.Ganglia_pool_destroy(s.global_context)
for _, cstr := range s.constStr { // Cleanup C string cache
for _, cstr := range s.cstrCache {
C.free(unsafe.Pointer(cstr)) C.free(unsafe.Pointer(cstr))
} }
} }

2
sinks/sinkManager.go
View File

@ -20,7 +20,7 @@ var AvailableSinks = map[string]Sink{
"http": new(HttpSink), "http": new(HttpSink),
"ganglia": new(GangliaSink), "ganglia": new(GangliaSink),
"influxasync": new(InfluxAsyncSink), "influxasync": new(InfluxAsyncSink),
"libganglia": new(Ganglia2Sink), "libganglia": new(LibgangliaSink),
} }
// Metric collector manager data structure // Metric collector manager data structure