2022-03-05 17:30:55 +01:00
|
|
|
package receivers
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"encoding/json"
|
|
|
|
"errors"
|
|
|
|
"fmt"
|
|
|
|
"net/http"
|
|
|
|
"strings"
|
|
|
|
"sync"
|
2023-09-15 15:59:03 +02:00
|
|
|
"time"
|
2022-03-05 17:30:55 +01:00
|
|
|
|
2022-10-10 11:53:11 +02:00
|
|
|
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
|
|
|
lp "github.com/ClusterCockpit/cc-metric-collector/pkg/ccMetric"
|
2022-03-05 17:30:55 +01:00
|
|
|
"github.com/gorilla/mux"
|
2023-09-15 15:59:03 +02:00
|
|
|
influx "github.com/influxdata/line-protocol/v2/lineprotocol"
|
2022-03-05 17:30:55 +01:00
|
|
|
)
|
|
|
|
|
|
|
|
const HTTP_RECEIVER_PORT = "8080"
|
|
|
|
|
|
|
|
type HttpReceiverConfig struct {
|
|
|
|
Type string `json:"type"`
|
|
|
|
Addr string `json:"address"`
|
|
|
|
Port string `json:"port"`
|
|
|
|
Path string `json:"path"`
|
2023-09-20 16:39:03 +02:00
|
|
|
|
|
|
|
// Maximum amount of time to wait for the next request when keep-alives are enabled
|
|
|
|
// should be larger than the measurement interval to keep the connection open
|
|
|
|
IdleTimeout string `json:"idle_timeout"`
|
|
|
|
idleTimeout time.Duration
|
2023-09-20 17:33:08 +02:00
|
|
|
|
|
|
|
// Basic authentication
|
|
|
|
Username string `json:"username"`
|
|
|
|
Password string `json:"password"`
|
|
|
|
useBasicAuth bool
|
2022-03-05 17:30:55 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
type HttpReceiver struct {
|
|
|
|
receiver
|
2023-09-15 15:59:03 +02:00
|
|
|
meta map[string]string
|
|
|
|
config HttpReceiverConfig
|
|
|
|
router *mux.Router
|
|
|
|
server *http.Server
|
|
|
|
wg sync.WaitGroup
|
2022-03-05 17:30:55 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
func (r *HttpReceiver) Init(name string, config json.RawMessage) error {
|
|
|
|
r.name = fmt.Sprintf("HttpReceiver(%s)", name)
|
2023-09-20 16:39:03 +02:00
|
|
|
|
|
|
|
// Set default values
|
2022-03-05 17:30:55 +01:00
|
|
|
r.config.Port = HTTP_RECEIVER_PORT
|
2023-09-20 16:39:03 +02:00
|
|
|
// should be larger than the measurement interval to keep the connection open
|
|
|
|
r.config.IdleTimeout = "120s"
|
|
|
|
|
|
|
|
// Read config
|
2022-03-05 17:30:55 +01:00
|
|
|
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")
|
|
|
|
}
|
2023-09-20 17:33:08 +02:00
|
|
|
|
|
|
|
// Check idle timeout config
|
2023-09-20 16:39:03 +02:00
|
|
|
if len(r.config.IdleTimeout) > 0 {
|
|
|
|
t, err := time.ParseDuration(r.config.IdleTimeout)
|
|
|
|
if err == nil {
|
|
|
|
cclog.ComponentDebug(r.name, "idleTimeout", t)
|
|
|
|
r.config.idleTimeout = t
|
|
|
|
}
|
|
|
|
}
|
2023-09-20 17:33:08 +02:00
|
|
|
|
|
|
|
// Check basic authentication config
|
|
|
|
if len(r.config.Username) > 0 || len(r.config.Password) > 0 {
|
|
|
|
r.config.useBasicAuth = true
|
|
|
|
}
|
|
|
|
if r.config.useBasicAuth && len(r.config.Username) == 0 {
|
|
|
|
return errors.New("basic authentication requires username")
|
|
|
|
}
|
|
|
|
if r.config.useBasicAuth && len(r.config.Password) == 0 {
|
|
|
|
return errors.New("basic authentication requires password")
|
|
|
|
}
|
|
|
|
|
2022-03-05 17:30:55 +01:00
|
|
|
r.meta = map[string]string{"source": r.name}
|
|
|
|
p := r.config.Path
|
|
|
|
if !strings.HasPrefix(p, "/") {
|
|
|
|
p = "/" + p
|
|
|
|
}
|
2023-09-15 14:12:59 +02:00
|
|
|
addr := fmt.Sprintf("%s:%s", r.config.Addr, r.config.Port)
|
|
|
|
uri := addr + p
|
|
|
|
cclog.ComponentDebug(r.name, "INIT", "listen on:", uri)
|
2022-03-05 17:30:55 +01:00
|
|
|
|
2023-09-15 14:12:59 +02:00
|
|
|
// Create new router and register p as path
|
2022-03-05 17:30:55 +01:00
|
|
|
r.router = mux.NewRouter()
|
|
|
|
r.router.Path(p).HandlerFunc(r.ServerHttp)
|
2023-09-15 14:12:59 +02:00
|
|
|
|
|
|
|
// Create http server, with router as handler
|
|
|
|
r.server = &http.Server{
|
2023-09-20 16:39:03 +02:00
|
|
|
Addr: addr,
|
|
|
|
Handler: r.router,
|
|
|
|
IdleTimeout: r.config.idleTimeout,
|
2023-09-15 14:12:59 +02:00
|
|
|
}
|
2022-03-05 17:30:55 +01:00
|
|
|
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) {
|
2023-09-20 17:33:08 +02:00
|
|
|
|
|
|
|
// Check request method, only post method is handled
|
2022-03-05 17:30:55 +01:00
|
|
|
if req.Method != http.MethodPost {
|
|
|
|
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2023-09-20 17:33:08 +02:00
|
|
|
// Check basic authentication
|
|
|
|
if r.config.useBasicAuth {
|
|
|
|
username, password, ok := req.BasicAuth()
|
|
|
|
if !ok || username != r.config.Username || password != r.config.Password {
|
|
|
|
http.Error(w, "Unauthorized", http.StatusUnauthorized)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-09-15 15:59:03 +02:00
|
|
|
d := influx.NewDecoder(req.Body)
|
|
|
|
for d.Next() {
|
|
|
|
|
|
|
|
// Decode measurement name
|
|
|
|
measurement, err := d.Measurement()
|
|
|
|
if err != nil {
|
|
|
|
msg := "ServerHttp: Failed to decode measurement: " + err.Error()
|
|
|
|
cclog.ComponentError(r.name, msg)
|
|
|
|
http.Error(w, msg, http.StatusInternalServerError)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
// Decode tags
|
|
|
|
tags := make(map[string]string)
|
|
|
|
for {
|
|
|
|
key, value, err := d.NextTag()
|
|
|
|
if err != nil {
|
|
|
|
msg := "ServerHttp: Failed to decode tag: " + err.Error()
|
|
|
|
cclog.ComponentError(r.name, msg)
|
|
|
|
http.Error(w, msg, http.StatusInternalServerError)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
if key == nil {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
tags[string(key)] = string(value)
|
|
|
|
}
|
2022-03-05 17:30:55 +01:00
|
|
|
|
2023-09-15 15:59:03 +02:00
|
|
|
// Decode fields
|
|
|
|
fields := make(map[string]interface{})
|
|
|
|
for {
|
|
|
|
key, value, err := d.NextField()
|
|
|
|
if err != nil {
|
|
|
|
msg := "ServerHttp: Failed to decode field: " + err.Error()
|
|
|
|
cclog.ComponentError(r.name, msg)
|
|
|
|
http.Error(w, msg, http.StatusInternalServerError)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
if key == nil {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
fields[string(key)] = value.Interface()
|
2022-03-05 17:30:55 +01:00
|
|
|
}
|
2023-09-15 15:59:03 +02:00
|
|
|
|
2023-09-19 14:48:11 +02:00
|
|
|
// Decode time stamp
|
2023-09-15 15:59:03 +02:00
|
|
|
t, err := d.Time(influx.Nanosecond, time.Time{})
|
|
|
|
if err != nil {
|
2023-09-19 14:48:11 +02:00
|
|
|
msg := "ServerHttp: Failed to decode time stamp: " + err.Error()
|
2023-09-15 15:59:03 +02:00
|
|
|
cclog.ComponentError(r.name, msg)
|
|
|
|
http.Error(w, msg, http.StatusInternalServerError)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
y, _ := lp.New(
|
|
|
|
string(measurement),
|
|
|
|
tags,
|
|
|
|
r.meta,
|
|
|
|
fields,
|
|
|
|
t,
|
|
|
|
)
|
|
|
|
|
2022-03-05 17:30:55 +01:00
|
|
|
if r.sink != nil {
|
|
|
|
r.sink <- y
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-09-15 15:59:03 +02:00
|
|
|
// Check for IO errors
|
|
|
|
err := d.Err()
|
|
|
|
if err != nil {
|
|
|
|
msg := "ServerHttp: Failed to decode: " + err.Error()
|
|
|
|
cclog.ComponentError(r.name, msg)
|
|
|
|
http.Error(w, msg, http.StatusInternalServerError)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2022-03-05 17:30:55 +01:00
|
|
|
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
|
|
|
|
}
|