ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2024-11-10 12:37:25 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add comments

Browse Source
This commit is contained in:
Holger Obermaier 2022-02-09 11:08:50 +01:00
parent b4d7643c25
commit 1d299be3ea
2 changed files with 22 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
sinks/influxAsyncSink.go
View File

@ -1,8 +1,6 @@
package sinks package sinks
import ( import (
// "context"
"crypto/tls" "crypto/tls"
"encoding/json" "encoding/json"
"errors" "errors"
@ -54,10 +52,14 @@ func (s *InfluxAsyncSink) connect() error {
if batch == 0 { if batch == 0 {
batch = 100 batch = 100
} }
s.client = influxdb2.NewClientWithOptions(uri, auth, clientOptions := influxdb2.DefaultOptions()
influxdb2.DefaultOptions().SetBatchSize(batch).SetTLSConfig(&tls.Config{ clientOptions.SetBatchSize(batch)
clientOptions.SetTLSConfig(
&tls.Config{
InsecureSkipVerify: true, InsecureSkipVerify: true,
})) },
)
s.client = influxdb2.NewClientWithOptions(uri, auth, clientOptions)
s.writeApi = s.client.WriteAPI(s.config.Organization, s.config.Database) s.writeApi = s.client.WriteAPI(s.config.Organization, s.config.Database)
return nil return nil
} }
@ -78,7 +80,11 @@ func (s *InfluxAsyncSink) Init(config json.RawMessage) error {
len(s.config.Password) == 0 { len(s.config.Password) == 0 {
return errors.New("not all configuration variables set required by InfluxAsyncSink") return errors.New("not all configuration variables set required by InfluxAsyncSink")
} }
// Connect to InfluxDB server
err := s.connect() err := s.connect()
// Start background: Read from error channel
s.errors = s.writeApi.Errors() s.errors = s.writeApi.Errors()
go func() { go func() {
for err := range s.errors { for err := range s.errors {
@ -90,7 +96,8 @@ func (s *InfluxAsyncSink) Init(config json.RawMessage) error {
func (s *InfluxAsyncSink) Write(m lp.CCMetric) error { func (s *InfluxAsyncSink) Write(m lp.CCMetric) error {
s.writeApi.WritePoint( s.writeApi.WritePoint(
m.ToPoint(s.config.MetaAsTags)) m.ToPoint(s.config.MetaAsTags),
)
return nil return nil
} }

15
sinks/influxSink.go
View File

@ -46,14 +46,13 @@ func (s *InfluxSink) connect() error {
auth = fmt.Sprintf("%s:%s", s.config.User, s.config.Password) 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) cclog.ComponentDebug(s.name, "Using URI", uri, "Org", s.config.Organization, "Bucket", s.config.Database)
s.client = clientOptions := influxdb2.DefaultOptions()
influxdb2.NewClientWithOptions( clientOptions.SetTLSConfig(
uri, &tls.Config{
auth, InsecureSkipVerify: true,
influxdb2.DefaultOptions().SetTLSConfig( },
&tls.Config{InsecureSkipVerify: true},
),
) )
s.client = influxdb2.NewClientWithOptions(uri, auth, clientOptions)
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
} }
@ -73,6 +72,8 @@ func (s *InfluxSink) Init(config json.RawMessage) error {
len(s.config.Password) == 0 { len(s.config.Password) == 0 {
return errors.New("not all configuration variables set required by InfluxSink") return errors.New("not all configuration variables set required by InfluxSink")
} }
// Connect to InfluxDB server
return s.connect() return s.connect()
} }