cc-metric-collector/sinks/influxSink.go
2021-10-12 13:43:58 +02:00

82 lines
2.0 KiB
Go

package sinks
import (
"context"
"crypto/tls"
"errors"
"fmt"
influxdb2 "github.com/influxdata/influxdb-client-go/v2"
influxdb2Api "github.com/influxdata/influxdb-client-go/v2/api"
lp "github.com/influxdata/line-protocol"
"log"
)
type InfluxSink struct {
Sink
client influxdb2.Client
writeApi influxdb2Api.WriteAPIBlocking
retPolicy string
}
func (s *InfluxSink) connect() error {
var auth string
var uri string
if s.ssl {
uri = fmt.Sprintf("https://%s:%s", s.host, s.port)
} else {
uri = fmt.Sprintf("http://%s:%s", s.host, s.port)
}
if len(s.user) == 0 {
auth = s.password
} else {
auth = fmt.Sprintf("%s:%s", s.user, s.password)
}
log.Print("Using URI ", uri, " Org ", s.organization, " Bucket ", s.database)
s.client = influxdb2.NewClientWithOptions(uri, auth,
influxdb2.DefaultOptions().SetTLSConfig(&tls.Config{InsecureSkipVerify: true}))
s.writeApi = s.client.WriteAPIBlocking(s.organization, s.database)
return nil
}
func (s *InfluxSink) Init(config SinkConfig) error {
if len(config.Host) == 0 ||
len(config.Port) == 0 ||
len(config.Database) == 0 ||
len(config.Organization) == 0 ||
len(config.Password) == 0 {
return errors.New("Not all configuration variables set required by InfluxSink")
}
s.host = config.Host
s.port = config.Port
s.database = config.Database
s.organization = config.Organization
s.user = config.User
s.password = config.Password
s.ssl = config.SSL
return s.connect()
}
func (s *InfluxSink) Write(point lp.MutableMetric) error {
tags := map[string]string{}
fields := map[string]interface{}{}
for _, t := range point.TagList() {
tags[t.Key] = t.Value
}
for _, f := range point.FieldList() {
fields[f.Key] = f.Value
}
p := influxdb2.NewPoint(point.Name(), tags, fields, point.Time())
err := s.writeApi.WritePoint(context.Background(), p)
return err
}
func (s *InfluxSink) Flush() error {
return nil
}
func (s *InfluxSink) Close() {
log.Print("Closing InfluxDB connection")
s.client.Close()
}