diff --git a/go.mod b/go.mod index 3908213..cb83b40 100644 --- a/go.mod +++ b/go.mod @@ -3,20 +3,21 @@ module github.com/ClusterCockpit/cc-metric-collector go 1.18 require ( - github.com/ClusterCockpit/cc-units v0.3.0 + github.com/ClusterCockpit/cc-units v0.4.0 github.com/ClusterCockpit/go-rocm-smi v0.3.0 github.com/NVIDIA/go-nvml v0.11.6-0 github.com/PaesslerAG/gval v1.2.1 github.com/gorilla/mux v1.8.0 github.com/influxdata/influxdb-client-go/v2 v2.12.1 - github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf + github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 + github.com/influxdata/line-protocol/v2 v2.2.1 github.com/nats-io/nats.go v1.22.1 github.com/prometheus/client_golang v1.14.0 github.com/stmcginnis/gofish v0.13.0 github.com/tklauser/go-sysconf v0.3.11 golang.design/x/thread v0.0.0-20210122121316-335e9adffdf1 golang.org/x/sys v0.3.0 - gopkg.in/fsnotify.v0 v0.9.3 + gopkg.in/fsnotify.v1 v1.4.7 ) require ( @@ -24,6 +25,7 @@ require ( github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.2.0 // indirect github.com/deepmap/oapi-codegen v1.12.4 // indirect + github.com/fsnotify/fsnotify v1.6.0 // indirect github.com/golang/protobuf v1.5.2 // indirect github.com/google/uuid v1.3.0 // indirect github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect diff --git a/sinks/httpSink.go b/sinks/httpSink.go index 49d98b0..3cdae62 100644 --- a/sinks/httpSink.go +++ b/sinks/httpSink.go @@ -11,7 +11,7 @@ import ( cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger" lp "github.com/ClusterCockpit/cc-metric-collector/pkg/ccMetric" - influx "github.com/influxdata/line-protocol" + influx "github.com/influxdata/line-protocol/v2/lineprotocol" ) type HttpSinkConfig struct { @@ -41,7 +41,12 @@ func (s *HttpSink) Write(m lp.CCMetric) error { p := m.ToPoint(s.meta_as_tags) s.lock.Lock() firstWriteOfBatch := s.buffer.Len() == 0 - _, err := s.encoder.Encode(p) + s.encoder.StartLine(p.Name()) + for _, v := range p.TagList() { + s.encoder.AddTag(v.Key, v.Value) + } + s.encoder.EndLine(p.Time()) + err := s.encoder.Err() s.lock.Unlock() if err != nil { cclog.ComponentError(s.name, "encoding failed:", err.Error()) @@ -174,7 +179,7 @@ func NewHttpSink(name string, config json.RawMessage) (Sink, error) { } s.client = &http.Client{Transport: tr, Timeout: s.timeout} s.buffer = &bytes.Buffer{} - s.encoder = influx.NewEncoder(s.buffer) - s.encoder.SetPrecision(time.Second) + s.encoder.SetPrecision(influx.Second) + s.encoder.SetBuffer(s.buffer.Bytes()) return s, nil } diff --git a/sinks/influxAsyncSink.go b/sinks/influxAsyncSink.go index ccf5302..360e6bb 100644 --- a/sinks/influxAsyncSink.go +++ b/sinks/influxAsyncSink.go @@ -186,16 +186,16 @@ func NewInfluxAsyncSink(name string, config json.RawMessage) (Sink, error) { } } if len(s.config.Port) == 0 { - return nil, errors.New("Missing port configuration required by InfluxSink") + return nil, errors.New("missing port configuration required by InfluxSink") } if len(s.config.Database) == 0 { - return nil, errors.New("Missing database configuration required by InfluxSink") + return nil, errors.New("missing database configuration required by InfluxSink") } if len(s.config.Organization) == 0 { - return nil, errors.New("Missing organization configuration required by InfluxSink") + return nil, errors.New("missing organization configuration required by InfluxSink") } if len(s.config.Password) == 0 { - return nil, errors.New("Missing password configuration required by InfluxSink") + return nil, errors.New("missing password configuration required by InfluxSink") } // Create lookup map to use meta infos as tags in the output metric s.meta_as_tags = make(map[string]bool) diff --git a/sinks/influxSink.go b/sinks/influxSink.go index 94704cd..95c28fc 100644 --- a/sinks/influxSink.go +++ b/sinks/influxSink.go @@ -213,19 +213,19 @@ func NewInfluxSink(name string, config json.RawMessage) (Sink, error) { } if len(s.config.Host) == 0 { - return s, errors.New("Missing host configuration required by InfluxSink") + return s, errors.New("missing host configuration required by InfluxSink") } if len(s.config.Port) == 0 { - return s, errors.New("Missing port configuration required by InfluxSink") + return s, errors.New("missing port configuration required by InfluxSink") } if len(s.config.Database) == 0 { - return s, errors.New("Missing database configuration required by InfluxSink") + return s, errors.New("missing database configuration required by InfluxSink") } if len(s.config.Organization) == 0 { - return s, errors.New("Missing organization configuration required by InfluxSink") + return s, errors.New("missing organization configuration required by InfluxSink") } if len(s.config.Password) == 0 { - return s, errors.New("Missing password configuration required by InfluxSink") + return s, errors.New("missing password configuration required by InfluxSink") } // Create lookup map to use meta infos as tags in the output metric diff --git a/sinks/sinkManager.go b/sinks/sinkManager.go index 392346d..eb1a55d 100644 --- a/sinks/sinkManager.go +++ b/sinks/sinkManager.go @@ -84,7 +84,7 @@ func (sm *sinkManager) Init(wg *sync.WaitGroup, sinkConfigFile string) error { // Check that at least one sink is running if !(len(sm.sinks) > 0) { cclog.ComponentError("SinkManager", "Found no usable sinks") - return fmt.Errorf("Found no usable sinks") + return fmt.Errorf("found no usable sinks") } return nil