Merge develop branch into main (#106)

* Add cpu_used (all-cpu_idle) to CpustatCollector

* Update to line-protocol/v2

* Update runonce.yml with Golang 1.20

* Update fsnotify in LIKWID Collector

* Use not a pointer to line-protocol.Encoder

* Simplify Makefile

* Use only as many arguments as required

* Allow sum function to handle non float types

* Allow values to be a slice of type float64, float32, int, int64, int32, bool

* Use generic function to simplify code

* Add missing case for type []int32

* Use generic function to compute minimum

* Use generic function to compute maximum

* Use generic function to compute average

* Add error value to sumAnyType

* Use generic function to compute median

* For older versions of go slices is not part of the installation

* Remove old entries from go.sum

* Use simpler sort function

* Compute metrics ib_total and ib_total_pkts

* Add aggregated metrics.
Add missing units

* Update likwidMetric.go

Fixes a potential bug when `fsnotify.NewWatcher()` fails with an error

* Completly avoid memory allocations in infinibandMetric read()

* Fixed initialization: Initalization and measurements should run in the same thread

---------

Co-authored-by: Holger Obermaier <40787752+ho-ob@users.noreply.github.com>

sinks/httpSink.go

@@ -11,12 +11,12 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/pkg/ccMetric"
-	influx "github.com/influxdata/line-protocol"
+	influx "github.com/influxdata/line-protocol/v2/lineprotocol"
 )
 
 type HttpSinkConfig struct {
 	defaultSinkConfig
-	URL             string `json:"url,omitempty"`
+	URL             string `json:"url"`
 	JWT             string `json:"jwt,omitempty"`
 	Timeout         string `json:"timeout,omitempty"`
 	IdleConnTimeout string `json:"idle_connection_timeout,omitempty"`
@@ -26,10 +26,10 @@ type HttpSinkConfig struct {
 
 type HttpSink struct {
 	sink
-	client          *http.Client
-	encoder         *influx.Encoder
-	lock            sync.Mutex // Flush() runs in another goroutine, so this lock has to protect the buffer
-	buffer          *bytes.Buffer
+	client  *http.Client
+	encoder influx.Encoder
+	lock    sync.Mutex // Flush() runs in another goroutine, so this lock has to protect the buffer
+	//buffer          *bytes.Buffer
 	flushTimer      *time.Timer
 	config          HttpSinkConfig
 	idleConnTimeout time.Duration
@@ -38,15 +38,29 @@ type HttpSink struct {
 }
 
 func (s *HttpSink) Write(m lp.CCMetric) error {
+	var err error = nil
+	var firstWriteOfBatch bool = false
 	p := m.ToPoint(s.meta_as_tags)
 	s.lock.Lock()
-	firstWriteOfBatch := s.buffer.Len() == 0
-	_, err := s.encoder.Encode(p)
-	s.lock.Unlock()
-	if err != nil {
-		cclog.ComponentError(s.name, "encoding failed:", err.Error())
-		return err
+	firstWriteOfBatch = len(s.encoder.Bytes()) == 0
+	v, ok := m.GetField("value")
+	if ok {
+
+		s.encoder.StartLine(p.Name())
+		for _, v := range p.TagList() {
+			s.encoder.AddTag(v.Key, v.Value)
+		}
+
+		s.encoder.AddField("value", influx.MustNewValue(v))
+		s.encoder.EndLine(p.Time())
+		err = s.encoder.Err()
+		if err != nil {
+			cclog.ComponentError(s.name, "encoding failed:", err.Error())
+			s.lock.Unlock()
+			return err
+		}
 	}
+	s.lock.Unlock()
 
 	if s.flushDelay == 0 {
 		return s.Flush()
@@ -70,9 +84,9 @@ func (s *HttpSink) Write(m lp.CCMetric) error {
 func (s *HttpSink) Flush() error {
 	// Own lock for as short as possible: the time it takes to copy the buffer.
 	s.lock.Lock()
-	buf := make([]byte, s.buffer.Len())
-	copy(buf, s.buffer.Bytes())
-	s.buffer.Reset()
+	buf := make([]byte, len(s.encoder.Bytes()))
+	copy(buf, s.encoder.Bytes())
+	s.encoder.Reset()
 	s.lock.Unlock()
 	if len(buf) == 0 {
 		return nil
@@ -134,6 +148,7 @@ func NewHttpSink(name string, config json.RawMessage) (Sink, error) {
 	s.config.Timeout = "5s"
 	s.config.FlushDelay = "5s"
 	s.config.MaxRetries = 3
+	cclog.ComponentDebug(s.name, "init")
 
 	// Read config
 	if len(config) > 0 {
@@ -148,6 +163,7 @@ func NewHttpSink(name string, config json.RawMessage) (Sink, error) {
 	if len(s.config.IdleConnTimeout) > 0 {
 		t, err := time.ParseDuration(s.config.IdleConnTimeout)
 		if err == nil {
+			cclog.ComponentDebug(s.name, "idleConnTimeout", t)
 			s.idleConnTimeout = t
 		}
 	}
@@ -155,12 +171,14 @@ func NewHttpSink(name string, config json.RawMessage) (Sink, error) {
 		t, err := time.ParseDuration(s.config.Timeout)
 		if err == nil {
 			s.timeout = t
+			cclog.ComponentDebug(s.name, "timeout", t)
 		}
 	}
 	if len(s.config.FlushDelay) > 0 {
 		t, err := time.ParseDuration(s.config.FlushDelay)
 		if err == nil {
 			s.flushDelay = t
+			cclog.ComponentDebug(s.name, "flushDelay", t)
 		}
 	}
 	// Create lookup map to use meta infos as tags in the output metric
@@ -173,8 +191,6 @@ func NewHttpSink(name string, config json.RawMessage) (Sink, error) {
 		IdleConnTimeout: s.idleConnTimeout,
 	}
 	s.client = &http.Client{Transport: tr, Timeout: s.timeout}
-	s.buffer = &bytes.Buffer{}
-	s.encoder = influx.NewEncoder(s.buffer)
-	s.encoder.SetPrecision(time.Second)
+	s.encoder.SetPrecision(influx.Second)
 	return s, nil
 }

sinks/influxAsyncSink.go

@@ -186,16 +186,16 @@ func NewInfluxAsyncSink(name string, config json.RawMessage) (Sink, error) {
 		}
 	}
 	if len(s.config.Port) == 0 {
-		return nil, errors.New("Missing port configuration required by InfluxSink")
+		return nil, errors.New("missing port configuration required by InfluxSink")
 	}
 	if len(s.config.Database) == 0 {
-		return nil, errors.New("Missing database configuration required by InfluxSink")
+		return nil, errors.New("missing database configuration required by InfluxSink")
 	}
 	if len(s.config.Organization) == 0 {
-		return nil, errors.New("Missing organization configuration required by InfluxSink")
+		return nil, errors.New("missing organization configuration required by InfluxSink")
 	}
 	if len(s.config.Password) == 0 {
-		return nil, errors.New("Missing password configuration required by InfluxSink")
+		return nil, errors.New("missing password configuration required by InfluxSink")
 	}
 	// Create lookup map to use meta infos as tags in the output metric
 	s.meta_as_tags = make(map[string]bool)

sinks/influxSink.go

@@ -213,19 +213,19 @@ func NewInfluxSink(name string, config json.RawMessage) (Sink, error) {
 	}
 
 	if len(s.config.Host) == 0 {
-		return s, errors.New("Missing host configuration required by InfluxSink")
+		return s, errors.New("missing host configuration required by InfluxSink")
 	}
 	if len(s.config.Port) == 0 {
-		return s, errors.New("Missing port configuration required by InfluxSink")
+		return s, errors.New("missing port configuration required by InfluxSink")
 	}
 	if len(s.config.Database) == 0 {
-		return s, errors.New("Missing database configuration required by InfluxSink")
+		return s, errors.New("missing database configuration required by InfluxSink")
 	}
 	if len(s.config.Organization) == 0 {
-		return s, errors.New("Missing organization configuration required by InfluxSink")
+		return s, errors.New("missing organization configuration required by InfluxSink")
 	}
 	if len(s.config.Password) == 0 {
-		return s, errors.New("Missing password configuration required by InfluxSink")
+		return s, errors.New("missing password configuration required by InfluxSink")
 	}
 
 	// Create lookup map to use meta infos as tags in the output metric

sinks/sinkManager.go

@@ -84,7 +84,7 @@ func (sm *sinkManager) Init(wg *sync.WaitGroup, sinkConfigFile string) error {
 	// Check that at least one sink is running
 	if !(len(sm.sinks) > 0) {
 		cclog.ComponentError("SinkManager", "Found no usable sinks")
-		return fmt.Errorf("Found no usable sinks")
+		return fmt.Errorf("found no usable sinks")
 	}
 
 	return nil