Add HttpReceiver as counterpart to the HttpSink (#49)

receivers/httpReceiver.go

@@ -0,0 +1,118 @@
+package receivers
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+	"sync"
+
+	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
+	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
+	"github.com/gorilla/mux"
+	influx "github.com/influxdata/line-protocol"
+)
+
+const HTTP_RECEIVER_PORT = "8080"
+
+type HttpReceiverConfig struct {
+	Type string `json:"type"`
+	Addr string `json:"address"`
+	Port string `json:"port"`
+	Path string `json:"path"`
+}
+
+type HttpReceiver struct {
+	receiver
+	handler *influx.MetricHandler
+	parser  *influx.Parser
+	meta    map[string]string
+	config  HttpReceiverConfig
+	router  *mux.Router
+	server  *http.Server
+	wg      sync.WaitGroup
+}
+
+func (r *HttpReceiver) Init(name string, config json.RawMessage) error {
+	r.name = fmt.Sprintf("HttpReceiver(%s)", name)
+	r.config.Port = HTTP_RECEIVER_PORT
+	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.Port) == 0 {
+		return errors.New("not all configuration variables set required by HttpReceiver")
+	}
+	r.meta = map[string]string{"source": r.name}
+	p := r.config.Path
+	if !strings.HasPrefix(p, "/") {
+		p = "/" + p
+	}
+	uri := fmt.Sprintf("%s:%s%s", r.config.Addr, r.config.Port, p)
+	cclog.ComponentDebug(r.name, "INIT", uri)
+	r.handler = influx.NewMetricHandler()
+	r.parser = influx.NewParser(r.handler)
+	r.parser.SetTimeFunc(DefaultTime)
+
+	r.router = mux.NewRouter()
+	r.router.Path(p).HandlerFunc(r.ServerHttp)
+	r.server = &http.Server{Addr: uri, Handler: r.router}
+	return nil
+}
+
+func (r *HttpReceiver) Start() {
+	cclog.ComponentDebug(r.name, "START")
+	r.wg.Add(1)
+	go func() {
+		err := r.server.ListenAndServe()
+		if err != nil && err.Error() != "http: Server closed" {
+			cclog.ComponentError(r.name, err.Error())
+		}
+		r.wg.Done()
+	}()
+}
+
+func (r *HttpReceiver) ServerHttp(w http.ResponseWriter, req *http.Request) {
+	if req.Method != http.MethodPost {
+		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	body, err := ioutil.ReadAll(req.Body)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	metrics, err := r.parser.Parse(body)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+	}
+
+	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
+		}
+	}
+
+	w.WriteHeader(http.StatusOK)
+}
+
+func (r *HttpReceiver) Close() {
+	r.server.Shutdown(context.Background())
+}
+
+func NewHttpReceiver(name string, config json.RawMessage) (Receiver, error) {
+	r := new(HttpReceiver)
+	err := r.Init(name, config)
+	return r, err
+}