cc-metric-collector/sinks/influxAsyncSink.go

127 lines
3.4 KiB
Go
Raw Normal View History

package sinks
import (
"crypto/tls"
"encoding/json"
"errors"
"fmt"
cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
influxdb2 "github.com/influxdata/influxdb-client-go/v2"
influxdb2Api "github.com/influxdata/influxdb-client-go/v2/api"
)
type InfluxAsyncSinkConfig struct {
defaultSinkConfig
Host string `json:"host,omitempty"`
Port string `json:"port,omitempty"`
Database string `json:"database,omitempty"`
User string `json:"user,omitempty"`
Password string `json:"password,omitempty"`
Organization string `json:"organization,omitempty"`
SSL bool `json:"ssl,omitempty"`
RetentionPol string `json:"retention_policy,omitempty"`
2022-02-10 09:43:02 +01:00
// Maximum number of points sent to server in single request. Default 5000
BatchSize uint `json:"batch_size,omitempty"`
// Interval, in ms, in which is buffer flushed if it has not been already written (by reaching batch size) . Default 1000ms
FlushInterval uint `json:"flush_interval,omitempty"`
}
type InfluxAsyncSink struct {
sink
client influxdb2.Client
writeApi influxdb2Api.WriteAPI
retPolicy string
errors <-chan error
config InfluxAsyncSinkConfig
}
func (s *InfluxAsyncSink) connect() error {
var auth string
var uri string
if s.config.SSL {
uri = fmt.Sprintf("https://%s:%s", s.config.Host, s.config.Port)
} else {
uri = fmt.Sprintf("http://%s:%s", s.config.Host, s.config.Port)
}
if len(s.config.User) == 0 {
auth = s.config.Password
} else {
auth = fmt.Sprintf("%s:%s", s.config.User, s.config.Password)
}
cclog.ComponentDebug(s.name, "Using URI", uri, "Org", s.config.Organization, "Bucket", s.config.Database)
2022-02-09 11:08:50 +01:00
clientOptions := influxdb2.DefaultOptions()
2022-02-10 09:43:02 +01:00
if s.config.BatchSize != 0 {
clientOptions.SetBatchSize(s.config.BatchSize)
}
if s.config.FlushInterval != 0 {
clientOptions.SetFlushInterval(s.config.FlushInterval)
}
2022-02-09 11:08:50 +01:00
clientOptions.SetTLSConfig(
&tls.Config{
InsecureSkipVerify: true,
2022-02-09 11:08:50 +01:00
},
)
s.client = influxdb2.NewClientWithOptions(uri, auth, clientOptions)
s.writeApi = s.client.WriteAPI(s.config.Organization, s.config.Database)
return nil
}
func (s *InfluxAsyncSink) Init(name string, config json.RawMessage) error {
s.name = fmt.Sprintf("InfluxSink(%s)", name)
2022-02-10 09:43:02 +01:00
// Set default for maximum number of points sent to server in single request.
s.config.BatchSize = 100
2022-02-10 09:43:02 +01:00
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 ||
len(s.config.Organization) == 0 ||
len(s.config.Password) == 0 {
return errors.New("not all configuration variables set required by InfluxAsyncSink")
}
2022-02-09 11:08:50 +01:00
// Connect to InfluxDB server
err := s.connect()
2022-02-09 11:08:50 +01:00
// Start background: Read from error channel
s.errors = s.writeApi.Errors()
go func() {
for err := range s.errors {
cclog.ComponentError(s.name, err.Error())
}
}()
return err
}
func (s *InfluxAsyncSink) Write(m lp.CCMetric) error {
s.writeApi.WritePoint(
2022-02-09 11:08:50 +01:00
m.ToPoint(s.config.MetaAsTags),
)
return nil
}
func (s *InfluxAsyncSink) Flush() error {
s.writeApi.Flush()
return nil
}
func (s *InfluxAsyncSink) Close() {
cclog.ComponentDebug(s.name, "Closing InfluxDB connection")
s.writeApi.Flush()
s.client.Close()
}
func NewInfluxAsyncSink(name string, config json.RawMessage) (Sink, error) {
s := new(InfluxAsyncSink)
s.Init(name, config)
return s, nil
}