Merge current development version into main (#48)

* DiskstatCollector: cast part_max_used metric to int

* Add uint types to GangliaSink and LibgangliaSink

* Use new sink instances to allow multiple of same sink type

* Update sink README and SampleSink

* Use new receiver instances to allow multiple of same receiver type

* Fix metric scope in likwid configuration script

* Mention likwid config script in LikwidCollector README

* Refactor: Embed Init() into New() function

* Refactor: Embed Init() into New() function

* Fix: MetricReceiver uses uninitialized values, when initialization fails

* Use Ganglia configuration (#44)

* Copy all metric configurations from original Ganglia code

* Use metric configurations from Ganglia for some metrics

* Format value string also for known metrics

* Numa-aware memstat collector (#45)

* Add samples for collectors, sinks and receivers

* Ping InfluxDB server after connecting to recognize faulty connections

* Add sink for Prometheus monitoring system (#46)

* Add sink for Prometheus monitoring system

* Add prometheus sink to README

* Add scraper for Prometheus clients (#47)

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

receivers/README.md

@@ -7,17 +7,13 @@ This folder contains the ReceiveManager and receiver implementations for the cc-
 The configuration file for the receivers is a list of configurations. The `type` field in each specifies which receiver to initialize.
 
 ```json
-[
-  {
-    "type": "nats",
-    "address": "nats://my-url",
-    "port" : "4222",
-    "database": "testcluster"
+{
+  "myreceivername" : {
+    <receiver-specific configuration>
   }
-]
+}
 ```
 
-
 ## Type `nats`
 
 ```json
@@ -25,20 +21,20 @@ The configuration file for the receivers is a list of configurations. The `type`
   "type": "nats",
   "address": "<nats-URI or hostname>",
   "port" : "<portnumber>",
-  "database": "<subscribe topic>"
+  "subject": "<subscribe topic>"
 }
 ```
 
 The `nats` receiver subscribes to the topic `database` and listens on `address` and `port` for metrics in the InfluxDB line protocol.
 
 # Contributing own receivers
-A receiver contains three functions and is derived from the type `Receiver` (in `metricReceiver.go`):
-* `Init(config ReceiverConfig) error`
+A receiver contains a few functions and is derived from the type `Receiver` (in `metricReceiver.go`):
 * `Start() error`
 * `Close()`
 * `Name() string`
-* `SetSink(sink chan ccMetric.CCMetric)`
+* `SetSink(sink chan lp.CCMetric)`
+* `New<Typename>(name string, config json.RawMessage)`
 
 The data structures should be set up in `Init()` like opening a file or server connection. The `Start()` function should either start a go routine or issue some other asynchronous mechanism for receiving metrics. The `Close()` function should tear down anything created in `Init()`.
 
-Finally, the receiver needs to be registered in the `receiveManager.go`. There is a list of receivers called `AvailableReceivers` which is a map (`receiver_type_string` -> `pointer to Receiver interface`). Add a new entry with a descriptive name and the new receiver.
+Finally, the receiver needs to be registered in the `receiveManager.go`. There is a list of receivers called `AvailableReceivers` which is a map (`receiver_type_string` -> `pointer to NewReceiver function`). Add a new entry with a descriptive name and the new receiver.

receivers/metricReceiver.go

@@ -1,9 +1,6 @@
 package receivers
 
 import (
-	//	"time"
-	"encoding/json"
-
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
 )
 
@@ -20,13 +17,11 @@ type ReceiverConfig struct {
 }
 
 type receiver struct {
-	typename string
-	name     string
-	sink     chan lp.CCMetric
+	name string
+	sink chan lp.CCMetric
 }
 
 type Receiver interface {
-	Init(name string, config json.RawMessage) error
 	Start()
 	Close()
 	Name() string

receivers/natsReceiver.go

@@ -32,39 +32,6 @@ var DefaultTime = func() time.Time {
 	return time.Unix(42, 0)
 }
 
-func (r *NatsReceiver) Init(name string, config json.RawMessage) error {
-	r.typename = "NatsReceiver"
-	r.name = name
-	r.config.Addr = nats.DefaultURL
-	r.config.Port = "4222"
-	if len(config) > 0 {
-		err := json.Unmarshal(config, &r.config)
-		if err != nil {
-			cclog.ComponentError(r.name, "Error reading config:", err.Error())
-			return err
-		}
-	}
-	if len(r.config.Addr) == 0 ||
-		len(r.config.Port) == 0 ||
-		len(r.config.Subject) == 0 {
-		return errors.New("not all configuration variables set required by NatsReceiver")
-	}
-	r.meta = map[string]string{"source": r.name}
-	uri := fmt.Sprintf("%s:%s", r.config.Addr, r.config.Port)
-	cclog.ComponentDebug(r.name, "INIT", uri, "Subject", r.config.Subject)
-	nc, err := nats.Connect(uri)
-	if err == nil {
-		r.nc = nc
-	} else {
-		r.nc = nil
-		return err
-	}
-	r.handler = influx.NewMetricHandler()
-	r.parser = influx.NewParser(r.handler)
-	r.parser.SetTimeFunc(DefaultTime)
-	return err
-}
-
 func (r *NatsReceiver) Start() {
 	cclog.ComponentDebug(r.name, "START")
 	r.nc.Subscribe(r.config.Subject, r._NatsReceive)
@@ -91,3 +58,35 @@ func (r *NatsReceiver) Close() {
 		r.nc.Close()
 	}
 }
+
+func NewNatsReceiver(name string, config json.RawMessage) (Receiver, error) {
+	r := new(NatsReceiver)
+	r.name = fmt.Sprintf("NatsReceiver(%s)", name)
+	r.config.Addr = nats.DefaultURL
+	r.config.Port = "4222"
+	if len(config) > 0 {
+		err := json.Unmarshal(config, &r.config)
+		if err != nil {
+			cclog.ComponentError(r.name, "Error reading config:", err.Error())
+			return nil, err
+		}
+	}
+	if len(r.config.Addr) == 0 ||
+		len(r.config.Port) == 0 ||
+		len(r.config.Subject) == 0 {
+		return nil, errors.New("not all configuration variables set required by NatsReceiver")
+	}
+	r.meta = map[string]string{"source": r.name}
+	uri := fmt.Sprintf("%s:%s", r.config.Addr, r.config.Port)
+	cclog.ComponentDebug(r.name, "NewNatsReceiver", uri, "Subject", r.config.Subject)
+	if nc, err := nats.Connect(uri); err == nil {
+		r.nc = nc
+	} else {
+		r.nc = nil
+		return nil, err
+	}
+	r.handler = influx.NewMetricHandler()
+	r.parser = influx.NewParser(r.handler)
+	r.parser.SetTimeFunc(DefaultTime)
+	return r, nil
+}

receivers/prometheusReceiver.go

@@ -0,0 +1,122 @@
+package receivers
+
+import (
+	"bufio"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
+	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
+)
+
+type PrometheusReceiverConfig struct {
+	defaultReceiverConfig
+	Addr     string `json:"address"`
+	Port     string `json:"port"`
+	Path     string `json:"path"`
+	Interval string `json:"interval"`
+	SSL      bool   `json:"ssl"`
+}
+
+type PrometheusReceiver struct {
+	receiver
+	meta     map[string]string
+	config   PrometheusReceiverConfig
+	interval time.Duration
+	done     chan bool
+	wg       sync.WaitGroup
+	ticker   *time.Ticker
+	uri      string
+}
+
+func (r *PrometheusReceiver) Start() {
+	cclog.ComponentDebug(r.name, "START", r.uri)
+	r.wg.Add(1)
+
+	r.ticker = time.NewTicker(r.interval)
+	go func() {
+		for {
+			select {
+			case <-r.done:
+				r.wg.Done()
+				return
+			case t := <-r.ticker.C:
+				resp, err := http.Get(r.uri)
+				if err != nil {
+					log.Fatal(err)
+				}
+				defer resp.Body.Close()
+
+				scanner := bufio.NewScanner(resp.Body)
+				for scanner.Scan() {
+					line := scanner.Text()
+					if strings.HasPrefix(line, "#") {
+						continue
+					}
+					lineSplit := strings.Fields(line)
+					// separate metric name from tags (labels in Prometheus)
+					tags := map[string]string{}
+					name := lineSplit[0]
+					if sindex := strings.Index(name, "{"); sindex >= 0 {
+						eindex := strings.Index(name, "}")
+						for _, kv := range strings.Split(name[sindex+1:eindex], ",") {
+							eq := strings.Index(kv, "=")
+							tags[kv[0:eq]] = strings.Trim(kv[eq+1:], "\"")
+						}
+						name = lineSplit[0][0:sindex]
+					}
+					value, err := strconv.ParseFloat(lineSplit[1], 64)
+					if err == nil {
+						y, err := lp.New(name, tags, r.meta, map[string]interface{}{"value": value}, t)
+						if err == nil {
+							r.sink <- y
+						}
+					}
+				}
+			}
+		}
+	}()
+}
+
+func (r *PrometheusReceiver) Close() {
+	cclog.ComponentDebug(r.name, "CLOSE")
+	r.done <- true
+	r.wg.Wait()
+}
+
+func NewPrometheusReceiver(name string, config json.RawMessage) (Receiver, error) {
+	r := new(PrometheusReceiver)
+	r.name = fmt.Sprintf("PrometheusReceiver(%s)", name)
+	if len(config) > 0 {
+		err := json.Unmarshal(config, &r.config)
+		if err != nil {
+			cclog.ComponentError(r.name, "Error reading config:", err.Error())
+			return nil, err
+		}
+	}
+	if len(r.config.Addr) == 0 ||
+		len(r.config.Port) == 0 ||
+		len(r.config.Interval) == 0 {
+		return nil, errors.New("not all configuration variables set required by PrometheusReceiver (address and port)")
+	}
+	if len(r.config.Interval) > 0 {
+		t, err := time.ParseDuration(r.config.Interval)
+		if err == nil {
+			r.interval = t
+		}
+	}
+	r.meta = map[string]string{"source": r.name}
+	proto := "http"
+	if r.config.SSL {
+		proto = "https"
+	}
+	r.uri = fmt.Sprintf("%s://%s:%s/%s", proto, r.config.Addr, r.config.Port, r.config.Path)
+	return r, nil
+}

receivers/prometheusReceiver.md

@@ -0,0 +1,27 @@
+## `prometheus` receiver
+
+The `prometheus` receiver can be used to scrape the metrics of a single `prometheus` client. It does **not** use any official Golang library but making simple HTTP get requests and parse the response.
+
+### Configuration structure
+
+```json
+{
+  "<name>": {
+    "type": "prometheus",
+    "address" : "testpromhost",
+    "port" : "12345",
+    "port" : "/prometheus",
+    "interval": "5s",
+    "ssl" : true,
+  }
+}
+```
+
+- `type`: makes the receiver a `prometheus` receiver
+- `address`: Hostname or IP of the Prometheus agent
+- `port`: Port of Prometheus agent
+- `path`: Path to the Prometheus endpoint
+- `interval`: Scrape the Prometheus endpoint in this interval (default '5s')
+- `ssl`: Use SSL or not
+
+The receiver requests data from `http(s)://<address>:<port>/<path>`.

receivers/receiveManager.go

@@ -9,8 +9,8 @@ import (
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
 )
 
-var AvailableReceivers = map[string]Receiver{
-	"nats": &NatsReceiver{},
+var AvailableReceivers = map[string]func(name string, config json.RawMessage) (Receiver, error){
+	"nats": NewNatsReceiver,
 }
 
 type receiveManager struct {
@@ -30,11 +30,13 @@ type ReceiveManager interface {
 }
 
 func (rm *receiveManager) Init(wg *sync.WaitGroup, receiverConfigFile string) error {
+	// Initialize struct fields
 	rm.inputs = make([]Receiver, 0)
 	rm.output = nil
 	rm.done = make(chan bool)
 	rm.wg = wg
 	rm.config = make([]json.RawMessage, 0)
+
 	configFile, err := os.Open(receiverConfigFile)
 	if err != nil {
 		cclog.ComponentError("ReceiveManager", err.Error())
@@ -51,6 +53,7 @@ func (rm *receiveManager) Init(wg *sync.WaitGroup, receiverConfigFile string) er
 	for name, raw := range rawConfigs {
 		rm.AddInput(name, raw)
 	}
+
 	return nil
 }
 
@@ -75,10 +78,9 @@ func (rm *receiveManager) AddInput(name string, rawConfig json.RawMessage) error
 		cclog.ComponentError("ReceiveManager", "SKIP", config.Type, "unknown receiver:", err.Error())
 		return err
 	}
-	r := AvailableReceivers[config.Type]
-	err = r.Init(name, rawConfig)
+	r, err := AvailableReceivers[config.Type](name, rawConfig)
 	if err != nil {
-		cclog.ComponentError("ReceiveManager", "SKIP", r.Name(), "initialization failed:", err.Error())
+		cclog.ComponentError("ReceiveManager", "SKIP", name, "initialization failed:", err.Error())
 		return err
 	}
 	rm.inputs = append(rm.inputs, r)

receivers/sampleReceiver.go

@@ -0,0 +1,77 @@
+package receivers
+
+import (
+	"encoding/json"
+	"fmt"
+
+	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
+)
+
+type SampleReceiverConfig struct {
+	Type string `json:"type"`
+	Addr string `json:"address"`
+	Port string `json:"port"`
+}
+
+type SampleReceiver struct {
+	receiver
+	config SampleReceiverConfig
+
+	// Storage for static information
+	meta map[string]string
+	// Use in case of own go routine
+	// done chan bool
+	// wg   sync.WaitGroup
+}
+
+func (r *SampleReceiver) Start() {
+	cclog.ComponentDebug(r.name, "START")
+
+	// Start server process like http.ListenAndServe()
+
+	// or use own go routine but always make sure it exits
+	// as soon as it gets the signal of the r.done channel
+	// r.wg.Add(1)
+	// go func() {
+	// 	for {
+	// 		select {
+	// 		case <-r.done:
+	// 			r.wg.Done()
+	// 			return
+	// 		}
+	// 	}
+	// 	r.wg.Done()
+	// }()
+}
+
+func (r *SampleReceiver) Close() {
+	cclog.ComponentDebug(r.name, "CLOSE")
+
+	// Close server like http.Shutdown()
+
+	// in case of own go routine, send the signal and wait
+	// r.done <- true
+	// r.wg.Wait()
+}
+
+func NewSampleReceiver(name string, config json.RawMessage) (Receiver, error) {
+	r := new(SampleReceiver)
+	r.name = fmt.Sprintf("HttpReceiver(%s)", name)
+
+	// Set static information
+	r.meta = map[string]string{"source": r.name}
+
+	// Read the sample receiver specific JSON config
+	if len(config) > 0 {
+		err := json.Unmarshal(config, &r.config)
+		if err != nil {
+			cclog.ComponentError(r.name, "Error reading config:", err.Error())
+			return nil, err
+		}
+	}
+
+	// Check that all required fields in the configuration are set
+	// Use 'if len(r.config.Option) > 0' for strings
+
+	return r, nil
+}