cc-metric-collector/sinks/httpSink.go
2022-02-08 11:23:19 +01:00

92 lines
1.8 KiB
Go

package sinks
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"net/http"
"time"
lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
influx "github.com/influxdata/line-protocol"
)
type HttpSinkConfig struct {
defaultSinkConfig
Host string `json:"host,omitempty"`
Port string `json:"port,omitempty"`
Database string `json:"database,omitempty"`
JWT string `json:"jwt,omitempty"`
SSL bool `json:"ssl,omitempty"`
}
type HttpSink struct {
sink
client *http.Client
url, jwt string
encoder *influx.Encoder
buffer *bytes.Buffer
config HttpSinkConfig
}
func (s *HttpSink) Init(config json.RawMessage) error {
s.name = "HttpSink"
s.config.SSL = false
if len(config) > 0 {
err := json.Unmarshal(config, &s.config)
if err != nil {
return err
}
}
if len(s.config.Host) == 0 || len(s.config.Port) == 0 || len(s.config.Database) == 0 {
return errors.New("`host`, `port` and `database` config options required for TCP sink")
}
s.client = &http.Client{}
proto := "http"
if s.config.SSL {
proto = "https"
}
s.url = fmt.Sprintf("%s://%s:%s/%s", proto, s.config.Host, s.config.Port, s.config.Database)
s.jwt = s.config.JWT
s.buffer = &bytes.Buffer{}
s.encoder = influx.NewEncoder(s.buffer)
s.encoder.SetPrecision(time.Second)
return nil
}
func (s *HttpSink) Write(m lp.CCMetric) error {
_, err := s.encoder.Encode(m.ToPoint(s.config.MetaAsTags))
return err
}
func (s *HttpSink) Flush() error {
req, err := http.NewRequest(http.MethodPost, s.url, s.buffer)
if err != nil {
return err
}
if len(s.jwt) != 0 {
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", s.jwt))
}
res, err := s.client.Do(req)
s.buffer.Reset()
if err != nil {
return err
}
if res.StatusCode != 200 {
return errors.New(res.Status)
}
return nil
}
func (s *HttpSink) Close() {
s.client.CloseIdleConnections()
}