ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2024-12-26 15:29:04 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' of github.com:ClusterCockpit/cc-metric-collector into develop

Browse Source
This commit is contained in:
Thomas Roehl 2022-02-07 18:35:16 +01:00
commit 9586e67f81
1 changed files with 13 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
sinks/influxSink.go
View File

@ -6,8 +6,8 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"log"
cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric" lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
influxdb2 "github.com/influxdata/influxdb-client-go/v2" influxdb2 "github.com/influxdata/influxdb-client-go/v2"
influxdb2Api "github.com/influxdata/influxdb-client-go/v2/api" influxdb2Api "github.com/influxdata/influxdb-client-go/v2/api"
@ -45,9 +45,15 @@ func (s *InfluxSink) connect() error {
} else { } else {
auth = fmt.Sprintf("%s:%s", s.config.User, s.config.Password) auth = fmt.Sprintf("%s:%s", s.config.User, s.config.Password)
} }
log.Print("Using URI ", uri, " Org ", s.config.Organization, " Bucket ", s.config.Database) cclog.ComponentDebug(s.name, "Using URI", uri, "Org", s.config.Organization, "Bucket", s.config.Database)
s.client = influxdb2.NewClientWithOptions(uri, auth, s.client =
influxdb2.DefaultOptions().SetTLSConfig(&tls.Config{InsecureSkipVerify: true})) influxdb2.NewClientWithOptions(
uri,
auth,
influxdb2.DefaultOptions().SetTLSConfig(
&tls.Config{InsecureSkipVerify: true},
),
)
s.writeApi = s.client.WriteAPIBlocking(s.config.Organization, s.config.Database) s.writeApi = s.client.WriteAPIBlocking(s.config.Organization, s.config.Database)
return nil return nil
} }
@ -71,8 +77,7 @@ func (s *InfluxSink) Init(config json.RawMessage) error {
} }
func (s *InfluxSink) Write(point lp.CCMetric) error { func (s *InfluxSink) Write(point lp.CCMetric) error {
tags := map[string]string{} tags := make(map[string]string)
fields := map[string]interface{}{}
for key, value := range point.Tags() { for key, value := range point.Tags() {
tags[key] = value tags[key] = value
} }
@ -81,10 +86,7 @@ func (s *InfluxSink) Write(point lp.CCMetric) error {
tags[key] = value tags[key] = value
} }
} }
for _, f := range point.FieldList() { p := influxdb2.NewPoint(point.Name(), tags, point.Fields(), point.Time())
fields[f.Key] = f.Value
}
p := influxdb2.NewPoint(point.Name(), tags, fields, point.Time())
err := s.writeApi.WritePoint(context.Background(), p) err := s.writeApi.WritePoint(context.Background(), p)
return err return err
} }
@ -94,6 +96,6 @@ func (s *InfluxSink) Flush() error {
} }
func (s *InfluxSink) Close() { func (s *InfluxSink) Close() {
log.Print("Closing InfluxDB connection") cclog.ComponentDebug(s.name, "Closing InfluxDB connection")
s.client.Close() s.client.Close()
} }