Use channels, add a metric router, split up configuration and use extended version of Influx line protocol internally

receivers/metricReceiver.go

@@ -2,30 +2,42 @@ package receivers
 
 import (
 	//	"time"
-	s "github.com/ClusterCockpit/cc-metric-collector/sinks"
 	influx "github.com/influxdata/line-protocol"
+	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
+
 )
 
 type ReceiverConfig struct {
 	Addr     string `json:"address"`
 	Port     string `json:"port"`
 	Database string `json:"database"`
+	Organization string `json:"organization", omitempty`
 	Type     string `json:"type"`
 }
 
-type Receiver struct {
+type receiver struct {
 	name         string
 	addr         string
 	port         string
 	database     string
 	organization string
-	sink         s.SinkFuncs
+	sink         chan lp.CCMetric
 }
 
-type ReceiverFuncs interface {
-	Init(config ReceiverConfig, sink s.SinkFuncs) error
+type Receiver interface {
+	Init(config ReceiverConfig) error
 	Start()
 	Close()
+	Name() string
+	SetSink(sink chan lp.CCMetric)
+}
+
+func (r *receiver) Name() string {
+	return r.name
+}
+
+func (r *receiver) SetSink(sink chan lp.CCMetric) {
+    r.sink = sink
 }
 
 func Tags2Map(metric influx.Metric) map[string]string {

receivers/natsReceiver.go

@@ -2,56 +2,68 @@ package receivers
 
 import (
 	"errors"
-	s "github.com/ClusterCockpit/cc-metric-collector/sinks"
-	lp "github.com/influxdata/line-protocol"
+	influx "github.com/influxdata/line-protocol"
+	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
 	nats "github.com/nats-io/nats.go"
 	"log"
 	"time"
+	"fmt"
 )
 
+type NatsReceiverConfig struct {
+    Addr     string `json:"address"`
+	Port     string `json:"port"`
+	Database string `json:"database"`
+}
+
 type NatsReceiver struct {
-	Receiver
+	receiver
 	nc      *nats.Conn
-	handler *lp.MetricHandler
-	parser  *lp.Parser
+	handler *influx.MetricHandler
+	parser  *influx.Parser
+	meta    map[string]string
+	config  ReceiverConfig
 }
 
 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 {
+func (r *NatsReceiver) Init(config ReceiverConfig) error {
+	r.name = "NatsReceiver"
+	r.config = config
+	if len(r.config.Addr) == 0 ||
+		len(r.config.Port) == 0 ||
+		len(r.config.Database) == 0 {
 		return errors.New("Not all configuration variables set required by NatsReceiver")
 	}
-	r.addr = config.Addr
+	r.meta = map[string]string{"source" : r.name}
+	r.addr = r.config.Addr
 	if len(r.addr) == 0 {
 		r.addr = nats.DefaultURL
 	}
-	r.port = config.Port
+	r.port = r.config.Port
 	if len(r.port) == 0 {
 		r.port = "4222"
 	}
-	log.Print("Init NATS Receiver")
-	nc, err := nats.Connect(r.addr)
+	log.Print("[NatsReceiver] INIT")
+	uri := fmt.Sprintf("%s:%s", r.addr, r.port)
+	nc, err := nats.Connect(uri)
 	if err == nil {
-		r.database = config.Database
-		r.sink = sink
+		r.database = r.config.Database
 		r.nc = nc
 	} else {
-		log.Print(err)
 		r.nc = nil
+		return err
 	}
-	r.handler = lp.NewMetricHandler()
-	r.parser = lp.NewParser(r.handler)
+	r.handler = influx.NewMetricHandler()
+	r.parser = influx.NewParser(r.handler)
 	r.parser.SetTimeFunc(DefaultTime)
 	return err
 }
 
 func (r *NatsReceiver) Start() {
-	log.Print("Start NATS Receiver")
+	log.Print("[NatsReceiver] START")
 	r.nc.Subscribe(r.database, r._NatsReceive)
 }
 
@@ -59,9 +71,13 @@ func (r *NatsReceiver) _NatsReceive(m *nats.Msg) {
 	metrics, err := r.parser.Parse(m.Data)
 	if err == nil {
 		for _, m := range metrics {
-			y, err := lp.New(m.Name(), Tags2Map(m), Fields2Map(m), m.Time())
-			if err == nil {
-				r.sink.Write(y)
+		    y := lp.FromInfluxMetric(m)
+		    for k, v := range r.meta {
+		        y.AddMeta(k, v)
+		    }
+			//y, err := lp.New(m.Name(), Tags2Map(m), r.meta, Fields2Map(m), m.Time())
+			if r.sink != nil {
+				r.sink <- y
 			}
 		}
 	}
@@ -69,7 +85,8 @@ func (r *NatsReceiver) _NatsReceive(m *nats.Msg) {
 
 func (r *NatsReceiver) Close() {
 	if r.nc != nil {
-		log.Print("Close NATS Receiver")
+		log.Print("[NatsReceiver] CLOSE")
 		r.nc.Close()
 	}
 }
+

receivers/receiveManager.go

@@ -0,0 +1,156 @@
+package receivers
+
+import (
+    lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
+    "sync"
+    "log"
+    "os"
+    "encoding/json"
+)
+
+var AvailableReceivers = map[string]Receiver{
+	"nats": &NatsReceiver{},
+}
+
+
+type receiveManager struct {
+	inputs      []Receiver
+	output      chan lp.CCMetric
+	done        chan bool
+	wg          *sync.WaitGroup
+	config      []ReceiverConfig
+}
+
+type ReceiveManager interface {
+	Init(wg *sync.WaitGroup, receiverConfigFile string) error
+	AddInput(rawConfig json.RawMessage) error
+	AddOutput(output chan lp.CCMetric)
+	Start()
+	Close()
+}
+
+
+func (rm *receiveManager) Init(wg *sync.WaitGroup, receiverConfigFile string) error {
+    rm.inputs = make([]Receiver, 0)
+    rm.output = nil
+    rm.done = make(chan bool)
+    rm.wg = wg
+    rm.config = make([]ReceiverConfig, 0)
+    configFile, err := os.Open(receiverConfigFile)
+    if err != nil {
+        log.Print(err.Error())
+        return err
+    }
+    defer configFile.Close()
+    jsonParser := json.NewDecoder(configFile)
+    var rawConfigs []json.RawMessage
+	err = jsonParser.Decode(&rawConfigs)
+	if err != nil {
+	    log.Print(err.Error())
+        return err
+    }
+    for _, raw := range rawConfigs {
+        log.Print("[ReceiveManager] ", string(raw))
+        rm.AddInput(raw)
+//        if _, found := AvailableReceivers[k.Type]; !found {
+//            log.Print("[ReceiveManager] SKIP Config specifies unknown receiver 'type': ", k.Type)
+//            continue
+//        }
+//        r := AvailableReceivers[k.Type]
+//        err = r.Init(k)
+//        if err != nil {
+//            log.Print("[ReceiveManager] SKIP Receiver ", k.Type, " cannot be initialized: ", err.Error())
+//            continue
+//        }
+//        rm.inputs = append(rm.inputs, r)
+    }
+    return nil
+}
+
+
+func (rm *receiveManager) Start() {
+    rm.wg.Add(1)
+
+    for _, r := range rm.inputs {
+        log.Print("[ReceiveManager] START ", r.Name())
+        r.Start()
+    }
+    log.Print("[ReceiveManager] STARTED\n")
+//    go func() {
+//        for {
+//ReceiveManagerLoop:
+//            select {
+//            case <- rm.done:
+//                log.Print("ReceiveManager done\n")
+//                rm.wg.Done()
+//                break ReceiveManagerLoop
+//            default:
+//                for _, c := range rm.inputs {
+//ReceiveManagerInputLoop:
+//                    select {
+//                    case <- rm.done:
+//                        log.Print("ReceiveManager done\n")
+//                        rm.wg.Done()
+//                        break ReceiveManagerInputLoop
+//                    case p := <- c:
+//                        log.Print("ReceiveManager: ", p)
+//                        rm.output <- p
+//                    default:
+//                    }
+//                }
+//            }
+//        }
+//    }()
+//    for _, r := range rm.inputs {
+//        r.Close()
+//    }
+}
+
+func (rm *receiveManager) AddInput(rawConfig json.RawMessage) error {
+    var config ReceiverConfig
+    err := json.Unmarshal(rawConfig, &config)
+	if err != nil {
+	    log.Print("[ReceiveManager] SKIP ", config.Type, " JSON config error: ", err.Error())
+	    log.Print(err.Error())
+	    return err
+	}
+    if _, found := AvailableReceivers[config.Type]; !found {
+        log.Print("[ReceiveManager] SKIP ", config.Type, " unknown receiver: ", err.Error())
+	    return err
+    }
+    r := AvailableReceivers[config.Type]
+    err = r.Init(config)
+    if err != nil {
+	    log.Print("[ReceiveManager] SKIP ", r.Name(), " initialization failed: ", err.Error())
+	    return err
+	}
+    rm.inputs = append(rm.inputs, r)
+    rm.config = append(rm.config, config)
+    return nil
+}
+
+func (rm *receiveManager) AddOutput(output chan lp.CCMetric) {
+    rm.output = output
+    for _, r := range rm.inputs {
+        r.SetSink(rm.output)
+    }
+}
+
+func (rm *receiveManager) Close() {
+    for _, r := range rm.inputs {
+        log.Print("[ReceiveManager] CLOSE ", r.Name())
+        r.Close()
+    }
+    rm.wg.Done()
+    log.Print("[ReceiveManager] CLOSE\n")
+    log.Print("[ReceiveManager] EXIT\n")
+}
+
+func New(wg *sync.WaitGroup, receiverConfigFile string) (ReceiveManager, error) {
+    r := &receiveManager{}
+    err := r.Init(wg, receiverConfigFile)
+    if err != nil {
+        return nil, err
+    }
+    return r, err
+}