cc-metric-collector/sinks
Thomas Gruber 195d0794b0
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>
2023-08-29 14:12:49 +02:00
..
gangliaCommon.go Add latest development to main branch (#89) 2022-10-10 12:23:51 +02:00
gangliaSink.go Add latest development to main branch (#89) 2022-10-10 12:23:51 +02:00
gangliaSink.md Sink specific configuration maps (#25) 2022-02-04 18:12:24 +01:00
httpSink.go Merge develop branch into main (#106) 2023-08-29 14:12:49 +02:00
httpSink.md Update httpSink.md 2022-11-04 14:52:05 +01:00
influxAsyncSink.go Merge develop branch into main (#106) 2023-08-29 14:12:49 +02:00
influxAsyncSink.md Add config options for retry intervals of InfluxDB clients (#59) 2022-03-11 13:43:03 +01:00
influxSink.go Merge develop branch into main (#106) 2023-08-29 14:12:49 +02:00
influxSink.md InfluxSink: Use batch&flush logic from HttpSink 2022-04-01 18:37:45 +02:00
libgangliaSink.go Add latest development to main branch (#89) 2022-10-10 12:23:51 +02:00
libgangliaSink.md Ganglia sink using libganglia.so directly (#35) 2022-02-16 18:33:46 +01:00
metricSink.go Add latest development to main branch (#89) 2022-10-10 12:23:51 +02:00
natsSink.go Add latest development to main branch (#89) 2022-10-10 12:23:51 +02:00
natsSink.md Sink specific configuration maps (#25) 2022-02-04 18:12:24 +01:00
prometheusSink.go Add latest development to main branch (#89) 2022-10-10 12:23:51 +02:00
prometheusSink.md Add sink for Prometheus monitoring system (#46) 2022-02-25 14:33:20 +01:00
README.md Update README.md 2022-11-04 14:53:08 +01:00
sampleSink.go Add latest development to main branch (#89) 2022-10-10 12:23:51 +02:00
sinkManager.go Merge develop branch into main (#106) 2023-08-29 14:12:49 +02:00
stdoutSink.go Add latest development to main branch (#89) 2022-10-10 12:23:51 +02:00
stdoutSink.md Sink specific configuration maps (#25) 2022-02-04 18:12:24 +01:00

CCMetric sinks

This folder contains the SinkManager and sink implementations for the cc-metric-collector.

Available sinks:

Configuration

The configuration file for the sinks is a list of configurations. The type field in each specifies which sink to initialize.

[
  "mystdout" : {
    "type" : "stdout",
    "meta_as_tags" : [
    	"unit"
    ]
  },
  "metricstore" : {
    "type" : "http",
    "host" : "localhost",
    "port" : "4123",
    "database" : "ccmetric",
    "password" : "<jwt token>"
  }
]

Contributing own sinks

A sink contains five functions and is derived from the type sink:

  • Init(name string, config json.RawMessage) error
  • Write(point CCMetric) error
  • Flush() error
  • Close()
  • New<Typename>(name string, config json.RawMessage) (Sink, error) (calls the Init() function)

The data structures should be set up in Init() like opening a file or server connection. The Write() function writes/sends the data. For non-blocking sinks, the Flush() method tells the sink to drain its internal buffers. The Close() function should tear down anything created in Init().

Finally, the sink needs to be registered in the sinkManager.go. There is a list of sinks called AvailableSinks which is a map (sink_type_string -> pointer to sink interface). Add a new entry with a descriptive name and the new sink.

Sample sink

package sinks

import (
	"encoding/json"
	"log"
	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
)

type SampleSinkConfig struct {
	defaultSinkConfig  // defines JSON tags for 'name' and 'meta_as_tags'
}

type SampleSink struct {
	sink              // declarate 'name' and 'meta_as_tags'
	config StdoutSinkConfig // entry point to the SampleSinkConfig
}

// Initialize the sink by giving it a name and reading in the config JSON
func (s *SampleSink) Init(name string, config json.RawMessage) error {
	s.name = fmt.Sprintf("SampleSink(%s)", name)   // Always specify a name here
  // Read in the config JSON
	if len(config) > 0 {
		err := json.Unmarshal(config, &s.config)
		if err != nil {
			return err
		}
	}
	return nil
}

// Code to submit a single CCMetric to the sink
func (s *SampleSink) Write(point lp.CCMetric) error {
	log.Print(point)
	return nil
}

// If the sink uses batched sends internally, you can tell to flush its buffers
func (s *SampleSink) Flush() error {
	return nil
}


// Close sink: close network connection, close files, close libraries, ...
func (s *SampleSink) Close() {}


// New function to create a new instance of the sink
func NewSampleSink(name string, config json.RawMessage) (Sink, error) {
	s := new(SampleSink)
	err := s.Init(name, config)
	return s, err
}