cc-metric-collector/sinks/influxSink.go

476 lines
13 KiB
Go
Raw Normal View History

package sinks
import (
2023-10-13 09:53:49 +02:00
"bytes"
"context"
2021-06-30 16:56:47 +02:00
"crypto/tls"
"encoding/json"
"errors"
"fmt"
"sync"
"time"
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
lp "github.com/ClusterCockpit/cc-metric-collector/pkg/ccMetric"
influxdb2 "github.com/influxdata/influxdb-client-go/v2"
influxdb2Api "github.com/influxdata/influxdb-client-go/v2/api"
2023-10-09 10:12:14 +02:00
influx "github.com/influxdata/line-protocol/v2/lineprotocol"
"golang.org/x/exp/slices"
)
type InfluxSink struct {
Modularize the whole thing (#16) * Use channels, add a metric router, split up configuration and use extended version of Influx line protocol internally * Use central timer for collectors and router. Add expressions to router * Add expression to router config * Update entry points * Start with README * Update README for CCMetric * Formatting * Update README.md * Add README for MultiChanTicker * Add README for MultiChanTicker * Update README.md * Add README to metric router * Update main README * Remove SinkEntity type * Update README for sinks * Update go files * Update README for receivers * Update collectors README * Update collectors README * Use seperate page per collector * Fix for tempstat page * Add docs for customcmd collector * Add docs for ipmistat collector * Add docs for topprocs collector * Update customCmdMetric.md * Use seconds when calculating LIKWID metrics * Add IB metrics ib_recv_pkts and ib_xmit_pkts * Drop domain part of host name * Updated to latest stable version of likwid * Define source code dependencies in Makefile * Add GPFS / IBM Spectrum Scale collector * Add vet and staticcheck make targets * Add vet and staticcheck make targets * Avoid go vet warning: struct field tag `json:"..., omitempty"` not compatible with reflect.StructTag.Get: suspicious space in struct tag value struct field tag `json:"...", omitempty` not compatible with reflect.StructTag.Get: key:"value" pairs not separated by spaces * Add sample collector to README.md * Add CPU frequency collector * Avoid staticcheck warning: redundant return statement * Avoid staticcheck warning: unnecessary assignment to the blank identifier * Simplified code * Add CPUFreqCollectorCpuinfo a metric collector to measure the current frequency of the CPUs as obtained from /proc/cpuinfo Only measure on the first hyperthread * Add collector for NFS clients * Move publication of metrics into Flush() for NatsSink * Update GitHub actions * Refactoring * Avoid vet warning: Println arg list ends with redundant newline * Avoid vet warning struct field commands has json tag but is not exported * Avoid vet warning: return copies lock value. * Corrected typo * Refactoring * Add go sources in internal/... * Bad separator in Makefile * Fix Infiniband collector Co-authored-by: Holger Obermaier <40787752+ho-ob@users.noreply.github.com>
2022-01-25 15:37:43 +01:00
sink
2022-05-06 11:44:57 +02:00
client influxdb2.Client
writeApi influxdb2Api.WriteAPIBlocking
config 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"`
// Maximum number of points sent to server in single request.
// Default: 1000
2022-05-06 11:44:57 +02:00
BatchSize int `json:"batch_size,omitempty"`
2023-10-11 14:55:52 +02:00
// Time interval for delayed sending of metrics.
// If the buffers are already filled before the end of this interval,
// the metrics are sent without further delay.
// Default: 1s
2022-05-06 11:44:57 +02:00
FlushInterval string `json:"flush_delay,omitempty"`
2023-10-11 14:55:52 +02:00
flushDelay time.Duration
2023-09-29 10:36:42 +02:00
// Influx client options:
// HTTP request timeout
HTTPRequestTimeout string `json:"http_request_timeout"`
// Retry interval
InfluxRetryInterval string `json:"retry_interval,omitempty"`
2023-09-29 10:36:42 +02:00
// maximum delay between each retry attempt
InfluxMaxRetryInterval string `json:"max_retry_interval,omitempty"`
2023-09-29 10:36:42 +02:00
// base for the exponential retry delay
InfluxExponentialBase uint `json:"retry_exponential_base,omitempty"`
// maximum count of retry attempts of failed writes
InfluxMaxRetries uint `json:"max_retries,omitempty"`
// maximum total retry timeout
InfluxMaxRetryTime string `json:"max_retry_time,omitempty"`
// Specify whether to use GZip compression in write requests
InfluxUseGzip bool `json:"use_gzip"`
2022-05-06 11:44:57 +02:00
}
2023-10-11 14:55:52 +02:00
2023-10-09 10:12:14 +02:00
// influx line protocol encoder
encoder influx.Encoder
2023-10-11 14:55:52 +02:00
// number of records stored in the encoder
numRecordsInEncoder int
2023-10-09 10:12:14 +02:00
// List of tags and meta data tags which should be used as tags
extended_tag_list []key_value_pair
2023-10-11 14:55:52 +02:00
// Flush() runs in another goroutine and accesses the influx line protocol encoder,
// so this encoderLock has to protect the encoder and numRecordsInEncoder
encoderLock sync.Mutex
// timer to run Flush()
flushTimer *time.Timer
// Lock to assure that only one timer is running at a time
timerLock sync.Mutex
// WaitGroup to ensure only one send operation is running at a time
sendWaitGroup sync.WaitGroup
}
2022-05-04 11:28:06 +02:00
// connect connects to the InfluxDB server
func (s *InfluxSink) connect() error {
2022-05-04 11:28:06 +02:00
// URI options:
// * http://host:port
// * https://host:port
2021-06-30 16:56:47 +02:00
var uri string
if s.config.SSL {
uri = fmt.Sprintf("https://%s:%s", s.config.Host, s.config.Port)
2021-06-30 16:56:47 +02:00
} else {
uri = fmt.Sprintf("http://%s:%s", s.config.Host, s.config.Port)
2021-06-30 16:56:47 +02:00
}
2022-05-04 11:28:06 +02:00
// Authentication options:
// * token
// * username:password
var auth string
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, "connect():",
"Using URI='"+uri+"'",
"Org='"+s.config.Organization+"'",
"Bucket='"+s.config.Database+"'")
2022-05-04 11:28:06 +02:00
// Set influxDB client options
2022-02-09 11:08:50 +01:00
clientOptions := influxdb2.DefaultOptions()
// set HTTP request timeout
if len(s.config.HTTPRequestTimeout) > 0 {
if t, err := time.ParseDuration(s.config.HTTPRequestTimeout); err == nil {
httpRequestTimeout := uint(t.Seconds())
clientOptions.SetHTTPRequestTimeout(httpRequestTimeout)
} else {
cclog.ComponentError(s.name, "connect():", "Failed to parse duration for HTTP RequestTimeout: ", s.config.HTTPRequestTimeout)
}
}
cclog.ComponentDebug(
s.name,
"connect():",
"Influx client options HTTPRequestTimeout:",
time.Second*time.Duration(clientOptions.HTTPRequestTimeout()))
// Set retry interval
if len(s.config.InfluxRetryInterval) > 0 {
if t, err := time.ParseDuration(s.config.InfluxRetryInterval); err == nil {
influxRetryInterval := uint(t.Milliseconds())
clientOptions.SetRetryInterval(influxRetryInterval)
} else {
cclog.ComponentError(s.name, "connect():", "Failed to parse duration for Influx RetryInterval: ", s.config.InfluxRetryInterval)
}
}
cclog.ComponentDebug(
s.name,
"connect():",
"Influx client options RetryInterval:",
time.Millisecond*time.Duration(clientOptions.RetryInterval()))
2023-09-29 10:36:42 +02:00
// Set the maximum delay between each retry attempt
if len(s.config.InfluxMaxRetryInterval) > 0 {
if t, err := time.ParseDuration(s.config.InfluxMaxRetryInterval); err == nil {
influxMaxRetryInterval := uint(t.Milliseconds())
clientOptions.SetMaxRetryInterval(influxMaxRetryInterval)
} else {
cclog.ComponentError(s.name, "connect():", "Failed to parse duration for Influx MaxRetryInterval: ", s.config.InfluxMaxRetryInterval)
2023-09-29 10:36:42 +02:00
}
}
cclog.ComponentDebug(
s.name,
"connect():",
"Influx client options MaxRetryInterval:",
time.Millisecond*time.Duration(clientOptions.MaxRetryInterval()))
2023-09-29 10:36:42 +02:00
// Set the base for the exponential retry delay
if s.config.InfluxExponentialBase != 0 {
clientOptions.SetExponentialBase(s.config.InfluxExponentialBase)
}
cclog.ComponentDebug(
s.name,
"connect():",
"Influx client options ExponentialBase:",
clientOptions.ExponentialBase())
2023-09-29 10:36:42 +02:00
// Set maximum count of retry attempts of failed writes
if s.config.InfluxMaxRetries != 0 {
clientOptions.SetMaxRetries(s.config.InfluxMaxRetries)
}
cclog.ComponentDebug(
s.name,
"connect():",
"Influx client options MaxRetries:",
clientOptions.MaxRetries())
2023-09-29 10:36:42 +02:00
// Set the maximum total retry timeout
if len(s.config.InfluxMaxRetryTime) > 0 {
if t, err := time.ParseDuration(s.config.InfluxMaxRetryTime); err == nil {
influxMaxRetryTime := uint(t.Milliseconds())
cclog.ComponentDebug(s.name, "connect():", "MaxRetryTime", s.config.InfluxMaxRetryTime)
2023-09-29 10:36:42 +02:00
clientOptions.SetMaxRetryTime(influxMaxRetryTime)
} else {
cclog.ComponentError(s.name, "connect():", "Failed to parse duration for Influx MaxRetryInterval: ", s.config.InfluxMaxRetryInterval)
2023-09-29 10:36:42 +02:00
}
}
cclog.ComponentDebug(
s.name,
"connect():",
"Influx client options MaxRetryTime:",
time.Millisecond*time.Duration(clientOptions.MaxRetryTime()))
2023-09-29 10:36:42 +02:00
// Specify whether to use GZip compression in write requests
clientOptions.SetUseGZip(s.config.InfluxUseGzip)
cclog.ComponentDebug(
s.name,
"connect():",
"Influx client options UseGZip:",
clientOptions.UseGZip())
2022-05-04 11:28:06 +02:00
// Do not check InfluxDB certificate
2022-02-09 11:08:50 +01:00
clientOptions.SetTLSConfig(
&tls.Config{
InsecureSkipVerify: true,
},
)
2023-10-09 10:12:14 +02:00
// Set time precision
clientOptions.SetPrecision(time.Nanosecond)
2022-05-04 11:28:06 +02:00
// Create new writeAPI
2022-02-09 11:08:50 +01:00
s.client = influxdb2.NewClientWithOptions(uri, auth, clientOptions)
s.writeApi = s.client.WriteAPIBlocking(s.config.Organization, s.config.Database)
2022-05-04 11:28:06 +02:00
// Check InfluxDB server accessibility
ok, err := s.client.Ping(context.Background())
if err != nil {
return err
}
if !ok {
return fmt.Errorf("connection to %s not healthy", uri)
}
return nil
}
2023-10-11 14:55:52 +02:00
// Write sends metric m in influxDB line protocol
func (s *InfluxSink) Write(m lp.CCMetric) error {
2023-09-26 15:04:39 +02:00
2023-10-11 14:55:52 +02:00
// Lock for encoder usage
s.encoderLock.Lock()
2023-10-09 10:12:14 +02:00
// Encode measurement name
s.encoder.StartLine(m.Name())
// copy tags and meta data which should be used as tags
s.extended_tag_list = s.extended_tag_list[:0]
for key, value := range m.Tags() {
s.extended_tag_list =
append(
s.extended_tag_list,
key_value_pair{
key: key,
value: value,
},
)
}
for _, key := range s.config.MetaAsTags {
if value, ok := m.GetMeta(key); ok {
s.extended_tag_list =
append(
s.extended_tag_list,
key_value_pair{
key: key,
value: value,
},
)
}
}
// Encode tags (they musts be in lexical order)
slices.SortFunc(
s.extended_tag_list,
func(a key_value_pair, b key_value_pair) int {
if a.key < b.key {
return -1
}
if a.key > b.key {
return +1
}
return 0
},
)
for i := range s.extended_tag_list {
s.encoder.AddTag(
s.extended_tag_list[i].key,
s.extended_tag_list[i].value,
)
}
// Encode fields
for key, value := range m.Fields() {
s.encoder.AddField(key, influx.MustNewValue(value))
}
// Encode time stamp
s.encoder.EndLine(m.Time())
2023-10-11 14:55:52 +02:00
// Check for encoder errors
2023-10-09 10:12:14 +02:00
if err := s.encoder.Err(); err != nil {
2023-10-11 14:55:52 +02:00
// Unlock encoder usage
s.encoderLock.Unlock()
2023-10-11 14:55:52 +02:00
return fmt.Errorf("Encoding failed: %v", err)
2023-10-09 10:12:14 +02:00
}
2023-10-11 14:55:52 +02:00
s.numRecordsInEncoder++
2023-10-09 10:12:14 +02:00
2023-10-11 14:55:52 +02:00
if s.config.flushDelay == 0 {
// Unlock encoder usage
s.encoderLock.Unlock()
2023-10-11 14:55:52 +02:00
// Directly flush if no flush delay is configured
return s.Flush()
} else if s.numRecordsInEncoder == s.config.BatchSize {
// Unlock encoder usage
s.encoderLock.Unlock()
// Stop flush timer
if s.flushTimer != nil {
if ok := s.flushTimer.Stop(); ok {
2023-10-11 14:55:52 +02:00
cclog.ComponentDebug(s.name, "Write(): Stopped flush timer. Batch size limit reached before flush delay")
s.timerLock.Unlock()
}
}
2023-10-11 14:55:52 +02:00
// Flush if batch size is reached
2022-05-04 11:28:06 +02:00
return s.Flush()
2023-10-11 14:55:52 +02:00
} else if s.timerLock.TryLock() {
// Setup flush timer when flush delay is configured
// and no other timer is already running
if s.flushTimer != nil {
// Restarting existing flush timer
cclog.ComponentDebug(s.name, "Write(): Restarting flush timer")
s.flushTimer.Reset(s.config.flushDelay)
} else {
// Creating and starting flush timer
cclog.ComponentDebug(s.name, "Write(): Starting new flush timer")
s.flushTimer = time.AfterFunc(
s.config.flushDelay,
func() {
defer s.timerLock.Unlock()
cclog.ComponentDebug(s.name, "Starting flush triggered by flush timer")
if err := s.Flush(); err != nil {
cclog.ComponentError(s.name, "Flush triggered by flush timer: flush failed:", err)
}
})
}
2022-05-04 11:28:06 +02:00
}
2023-10-11 14:55:52 +02:00
// Unlock encoder usage
s.encoderLock.Unlock()
return nil
}
2023-10-11 14:55:52 +02:00
// Flush sends all metrics stored in encoder to InfluxDB server
2021-10-12 13:43:58 +02:00
func (s *InfluxSink) Flush() error {
2022-05-04 11:28:06 +02:00
2023-10-11 14:55:52 +02:00
// Lock for encoder usage
// Own lock for as short as possible: the time it takes to clone the buffer.
s.encoderLock.Lock()
2022-05-04 11:28:06 +02:00
2023-10-11 14:55:52 +02:00
buf := slices.Clone(s.encoder.Bytes())
numRecordsInBuf := s.numRecordsInEncoder
s.encoder.Reset()
s.numRecordsInEncoder = 0
// Unlock encoder usage
s.encoderLock.Unlock()
if len(buf) == 0 {
return nil
}
2022-05-04 11:28:06 +02:00
2023-10-11 14:55:52 +02:00
cclog.ComponentDebug(s.name, "Flush(): Flushing", numRecordsInBuf, "metrics")
2022-05-04 11:28:06 +02:00
2023-10-11 14:55:52 +02:00
// Asynchron send of encoder metrics
s.sendWaitGroup.Add(1)
go func() {
defer s.sendWaitGroup.Done()
startTime := time.Now()
err := s.writeApi.WriteRecord(context.Background(), string(buf))
if err != nil {
cclog.ComponentError(
s.name,
"Flush():",
"Flush failed:", err,
"(number of records =", numRecordsInBuf,
", buffer size =", len(buf),
", send duration =", time.Since(startTime),
")",
)
return
}
}()
2022-05-04 11:28:06 +02:00
2021-10-12 13:43:58 +02:00
return nil
}
func (s *InfluxSink) Close() {
cclog.ComponentDebug(s.name, "Closing InfluxDB connection")
// Stop existing timer and immediately flush
if s.flushTimer != nil {
if ok := s.flushTimer.Stop(); ok {
2023-10-11 14:55:52 +02:00
s.timerLock.Unlock()
}
}
2023-10-11 14:55:52 +02:00
// Flush
if err := s.Flush(); err != nil {
cclog.ComponentError(s.name, "Close():", "Flush failed:", err)
}
// Wait for send operations to finish
2023-10-11 14:55:52 +02:00
s.sendWaitGroup.Wait()
s.client.Close()
}
2022-05-04 11:28:06 +02:00
// NewInfluxSink create a new InfluxDB sink
func NewInfluxSink(name string, config json.RawMessage) (Sink, error) {
s := new(InfluxSink)
s.name = fmt.Sprintf("InfluxSink(%s)", name)
2022-05-04 11:28:06 +02:00
// Set config default values
s.config.BatchSize = 1000
2022-05-06 11:44:57 +02:00
s.config.FlushInterval = "1s"
2022-05-04 11:28:06 +02:00
// Read config
if len(config) > 0 {
2023-10-13 09:53:49 +02:00
d := json.NewDecoder(bytes.NewReader(config))
d.DisallowUnknownFields()
if err := d.Decode(&s.config); err != nil {
cclog.ComponentError(s.name, "Error reading config:", err.Error())
return nil, err
}
}
2022-05-04 11:28:06 +02:00
if len(s.config.Host) == 0 {
2023-07-17 15:20:12 +02:00
return s, errors.New("missing host configuration required by InfluxSink")
}
2022-05-04 11:28:06 +02:00
if len(s.config.Port) == 0 {
2023-07-17 15:20:12 +02:00
return s, errors.New("missing port configuration required by InfluxSink")
2022-05-04 11:28:06 +02:00
}
if len(s.config.Database) == 0 {
2023-07-17 15:20:12 +02:00
return s, errors.New("missing database configuration required by InfluxSink")
2022-05-04 11:28:06 +02:00
}
if len(s.config.Organization) == 0 {
2023-07-17 15:20:12 +02:00
return s, errors.New("missing organization configuration required by InfluxSink")
2022-05-04 11:28:06 +02:00
}
if len(s.config.Password) == 0 {
2023-07-17 15:20:12 +02:00
return s, errors.New("missing password configuration required by InfluxSink")
2022-05-04 11:28:06 +02:00
}
// Create lookup map to use meta infos as tags in the output metric
s.meta_as_tags = make(map[string]bool)
for _, k := range s.config.MetaAsTags {
s.meta_as_tags[k] = true
}
2022-05-04 11:28:06 +02:00
// Configure flush delay duration
2022-05-06 11:44:57 +02:00
if len(s.config.FlushInterval) > 0 {
t, err := time.ParseDuration(s.config.FlushInterval)
if err == nil {
2023-10-11 14:55:52 +02:00
s.config.flushDelay = t
}
}
2022-05-04 11:28:06 +02:00
if !(s.config.BatchSize > 0) {
return s, fmt.Errorf("batch_size=%d in InfluxDB config must be > 0", s.config.BatchSize)
}
// Connect to InfluxDB server
if err := s.connect(); err != nil {
return s, fmt.Errorf("unable to connect: %v", err)
}
2023-10-09 10:12:14 +02:00
// Configure influx line protocol encoder
s.encoder.SetPrecision(influx.Nanosecond)
s.extended_tag_list = make([]key_value_pair, 0)
return s, nil
}