ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2024-12-26 07:29:04 +01:00
Code Issues Packages Projects Releases Wiki Activity

Use new receiver instances to allow multiple of same receiver type

Browse Source
This commit is contained in:
Thomas Roehl 2022-02-22 16:33:38 +01:00
parent 9cfbe10247
commit 3598aed090
4 changed files with 22 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
receivers/README.md
View File

@ -7,14 +7,11 @@ 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. The configuration file for the receivers is a list of configurations. The `type` field in each specifies which receiver to initialize.
```json ```json
[ {
{ "myreceivername" : {
"type": "nats", <receiver-specific configuration>
"address": "nats://my-url",
"port" : "4222",
"database": "testcluster"
} }
] }
``` ```
@ -25,20 +22,21 @@ The configuration file for the receivers is a list of configurations. The `type`
"type": "nats", "type": "nats",
"address": "<nats-URI or hostname>", "address": "<nats-URI or hostname>",
"port" : "<portnumber>", "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. The `nats` receiver subscribes to the topic `database` and listens on `address` and `port` for metrics in the InfluxDB line protocol.
# Contributing own receivers # Contributing own receivers
A receiver contains three functions and is derived from the type `Receiver` (in `metricReceiver.go`): A receiver contains a few functions and is derived from the type `Receiver` (in `metricReceiver.go`):
* `Init(config ReceiverConfig) error` * `Init(name string, config json.RawMessage) error`
* `Start() error` * `Start() error`
* `Close()` * `Close()`
* `Name() string` * `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()`. 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.

5
receivers/metricReceiver.go
View File

@ -20,9 +20,8 @@ type ReceiverConfig struct {
} }
type receiver struct { type receiver struct {
typename string name string
name string sink chan lp.CCMetric
sink chan lp.CCMetric
} }
type Receiver interface { type Receiver interface {

9
receivers/natsReceiver.go
View File

@ -33,8 +33,7 @@ var DefaultTime = func() time.Time {
} }
func (r *NatsReceiver) Init(name string, config json.RawMessage) error { func (r *NatsReceiver) Init(name string, config json.RawMessage) error {
r.typename = "NatsReceiver" r.name = fmt.Sprintf("NatsReceiver(%s)", name)
r.name = name
r.config.Addr = nats.DefaultURL r.config.Addr = nats.DefaultURL
r.config.Port = "4222" r.config.Port = "4222"
if len(config) > 0 { if len(config) > 0 {
@ -91,3 +90,9 @@ func (r *NatsReceiver) Close() {
r.nc.Close() r.nc.Close()
} }
} }
func NewNatsReceiver(name string, config json.RawMessage) (Receiver, error) {
r := new(NatsReceiver)
err := r.Init(name, config)
return r, err
}

7
receivers/receiveManager.go
View File

@ -9,8 +9,8 @@ import (
lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric" lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
) )
var AvailableReceivers = map[string]Receiver{ var AvailableReceivers = map[string]func(name string, config json.RawMessage) (Receiver, error){
"nats": &NatsReceiver{}, "nats": NewNatsReceiver,
} }
type receiveManager struct { type receiveManager struct {
@ -75,8 +75,7 @@ func (rm *receiveManager) AddInput(name string, rawConfig json.RawMessage) error
cclog.ComponentError("ReceiveManager", "SKIP", config.Type, "unknown receiver:", err.Error()) cclog.ComponentError("ReceiveManager", "SKIP", config.Type, "unknown receiver:", err.Error())
return err return err
} }
r := AvailableReceivers[config.Type] r, err := AvailableReceivers[config.Type](name, rawConfig)
err = r.Init(name, rawConfig)
if err != nil { if err != nil {
cclog.ComponentError("ReceiveManager", "SKIP", r.Name(), "initialization failed:", err.Error()) cclog.ComponentError("ReceiveManager", "SKIP", r.Name(), "initialization failed:", err.Error())
return err return err