mirror of
https://github.com/ClusterCockpit/cc-metric-collector.git
synced 2024-11-10 04:27:25 +01:00
d98076c792
* 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>
93 lines
2.1 KiB
Go
93 lines
2.1 KiB
Go
package receivers
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
|
|
lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
|
|
influx "github.com/influxdata/line-protocol"
|
|
nats "github.com/nats-io/nats.go"
|
|
)
|
|
|
|
type NatsReceiverConfig struct {
|
|
Type string `json:"type"`
|
|
Addr string `json:"address"`
|
|
Port string `json:"port"`
|
|
Subject string `json:"subject"`
|
|
}
|
|
|
|
type NatsReceiver struct {
|
|
receiver
|
|
nc *nats.Conn
|
|
handler *influx.MetricHandler
|
|
parser *influx.Parser
|
|
meta map[string]string
|
|
config NatsReceiverConfig
|
|
}
|
|
|
|
var DefaultTime = func() time.Time {
|
|
return time.Unix(42, 0)
|
|
}
|
|
|
|
func (r *NatsReceiver) Start() {
|
|
cclog.ComponentDebug(r.name, "START")
|
|
r.nc.Subscribe(r.config.Subject, r._NatsReceive)
|
|
}
|
|
|
|
func (r *NatsReceiver) _NatsReceive(m *nats.Msg) {
|
|
metrics, err := r.parser.Parse(m.Data)
|
|
if err == nil {
|
|
for _, m := range metrics {
|
|
y := lp.FromInfluxMetric(m)
|
|
for k, v := range r.meta {
|
|
y.AddMeta(k, v)
|
|
}
|
|
if r.sink != nil {
|
|
r.sink <- y
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (r *NatsReceiver) Close() {
|
|
if r.nc != nil {
|
|
cclog.ComponentDebug(r.name, "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
|
|
}
|