cc-metric-collector/receivers/httpReceiver.go

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
}