Add Nats receiver

receivers/metricReceiver.go

@@ -0,0 +1,45 @@
+package receivers
+
+import (
+	//	"time"
+	s "github.com/ClusterCockpit/cc-metric-collector/sinks"
+	influx "github.com/influxdata/line-protocol"
+)
+
+type ReceiverConfig struct {
+	Addr     string `json:"host"`
+	Port     string `json:"port"`
+	Database string `json:"database"`
+	Type     string `json:"type"`
+}
+
+type Receiver struct {
+    name         string
+	addr         string
+	port         string
+	database     string
+	organization string
+	sink         s.SinkFuncs
+}
+
+type ReceiverFuncs interface {
+	Init(config ReceiverConfig, sink s.SinkFuncs) error
+	Start()
+	Close()
+}
+
+func Tags2Map(metric influx.Metric) map[string]string {
+	tags := make(map[string]string)
+	for _, t := range metric.TagList() {
+		tags[t.Key] = t.Value
+	}
+	return tags
+}
+
+func Fields2Map(metric influx.Metric) map[string]interface{} {
+	fields := make(map[string]interface{})
+	for _, f := range metric.FieldList() {
+		fields[f.Key] = f.Value
+	}
+	return fields
+}

receivers/natsReceiver.go

@@ -0,0 +1,72 @@
+package receivers
+
+import (
+	s "github.com/ClusterCockpit/cc-metric-collector/sinks"
+	protocol "github.com/influxdata/line-protocol"
+	nats "github.com/nats-io/nats.go"
+	"log"
+	"time"
+	"errors"
+)
+
+type NatsReceiver struct {
+	Receiver
+	nc *nats.Conn
+}
+
+var DefaultTime = func() time.Time {
+	return time.Unix(42, 0)
+}
+
+func (r *NatsReceiver) Init(config ReceiverConfig, sink s.SinkFuncs) error {
+    if len(config.Addr) == 0 ||
+		len(config.Port) == 0 ||
+		len(config.Database) == 0 {
+		return errors.New("Not all configuration variables set required by NatsReceiver")
+	}
+	r.addr = config.Addr
+	if len(r.addr) == 0 {
+		r.addr = nats.DefaultURL
+	}
+	r.port = config.Port
+	if len(r.port) == 0 {
+		r.port = "4222"
+	}
+	log.Print("Init NATS Receiver")
+	nc, err := nats.Connect(r.addr)
+	if err == nil {
+		r.database = config.Database
+		r.sink = sink
+		r.nc = nc
+	} else {
+		log.Print(err)
+		r.nc = nil
+	}
+	return err
+}
+
+func (r *NatsReceiver) Start() {
+	log.Print("Start NATS Receiver")
+	r.nc.Subscribe(r.database, r._NatsReceive)
+}
+
+func (r *NatsReceiver) _NatsReceive(m *nats.Msg) {
+	handler := protocol.NewMetricHandler()
+	parser := protocol.NewParser(handler)
+	parser.SetTimeFunc(DefaultTime)
+	metrics, err := parser.Parse(m.Data)
+	if err == nil {
+		for _, m := range metrics {
+			tags := Tags2Map(m)
+			fields := Fields2Map(m)
+			r.sink.Write(m.Name(), tags, fields, m.Time())
+		}
+	}
+}
+
+func (r *NatsReceiver) Close() {
+	if r.nc != nil {
+		log.Print("Close NATS Receiver")
+		r.nc.Close()
+	}
+}