package sinks import ( "context" "crypto/tls" "encoding/json" "errors" "fmt" "strings" "time" cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger" lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric" influxdb2 "github.com/influxdata/influxdb-client-go/v2" influxdb2Api "github.com/influxdata/influxdb-client-go/v2/api" influxdb2ApiHttp "github.com/influxdata/influxdb-client-go/v2/api/http" ) type InfluxAsyncSinkConfig 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 5000 BatchSize uint `json:"batch_size,omitempty"` // Interval, in ms, in which is buffer flushed if it has not been already written (by reaching batch size) . Default 1000ms FlushInterval uint `json:"flush_interval,omitempty"` InfluxRetryInterval string `json:"retry_interval,omitempty"` InfluxExponentialBase uint `json:"retry_exponential_base,omitempty"` InfluxMaxRetries uint `json:"max_retries,omitempty"` InfluxMaxRetryTime string `json:"max_retry_time,omitempty"` CustomFlushInterval string `json:"custom_flush_interval,omitempty"` MaxRetryAttempts uint `json:"max_retry_attempts,omitempty"` } type InfluxAsyncSink struct { sink client influxdb2.Client writeApi influxdb2Api.WriteAPI errors <-chan error config InfluxAsyncSinkConfig influxRetryInterval uint influxMaxRetryTime uint customFlushInterval time.Duration flushTimer *time.Timer } func (s *InfluxAsyncSink) connect() error { var auth string var uri string if s.config.SSL { uri = fmt.Sprintf("https://%s:%s", s.config.Host, s.config.Port) } else { uri = fmt.Sprintf("http://%s:%s", s.config.Host, s.config.Port) } 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, "Using URI", uri, "Org", s.config.Organization, "Bucket", s.config.Database) clientOptions := influxdb2.DefaultOptions() if s.config.BatchSize != 0 { cclog.ComponentDebug(s.name, "Batch size", s.config.BatchSize) clientOptions.SetBatchSize(s.config.BatchSize) } if s.config.FlushInterval != 0 { cclog.ComponentDebug(s.name, "Flush interval", s.config.FlushInterval) clientOptions.SetFlushInterval(s.config.FlushInterval) } if s.influxRetryInterval != 0 { cclog.ComponentDebug(s.name, "MaxRetryInterval", s.influxRetryInterval) clientOptions.SetMaxRetryInterval(s.influxRetryInterval) } if s.influxMaxRetryTime != 0 { cclog.ComponentDebug(s.name, "MaxRetryTime", s.influxMaxRetryTime) clientOptions.SetMaxRetryTime(s.influxMaxRetryTime) } if s.config.InfluxExponentialBase != 0 { cclog.ComponentDebug(s.name, "Exponential Base", s.config.InfluxExponentialBase) clientOptions.SetExponentialBase(s.config.InfluxExponentialBase) } if s.config.InfluxMaxRetries != 0 { cclog.ComponentDebug(s.name, "Max Retries", s.config.InfluxMaxRetries) clientOptions.SetMaxRetries(s.config.InfluxMaxRetries) } clientOptions.SetTLSConfig( &tls.Config{ InsecureSkipVerify: true, }, ).SetPrecision(time.Second) s.client = influxdb2.NewClientWithOptions(uri, auth, clientOptions) s.writeApi = s.client.WriteAPI(s.config.Organization, s.config.Database) ok, err := s.client.Ping(context.Background()) if err != nil { return err } if !ok { return fmt.Errorf("connection to %s not healthy", uri) } s.writeApi.SetWriteFailedCallback(func(batch string, err influxdb2ApiHttp.Error, retryAttempts uint) bool { mlist := strings.Split(batch, "\n") cclog.ComponentError(s.name, fmt.Sprintf("Failed to write batch with %d metrics %d times (max: %d): %s", len(mlist), retryAttempts, s.config.MaxRetryAttempts, err.Error())) return retryAttempts <= s.config.MaxRetryAttempts }) return nil } func (s *InfluxAsyncSink) Write(m lp.CCMetric) error { if s.customFlushInterval != 0 && s.flushTimer == nil { // Run a batched flush for all lines that have arrived in the defined interval s.flushTimer = time.AfterFunc(s.customFlushInterval, func() { if err := s.Flush(); err != nil { cclog.ComponentError(s.name, "flush failed:", err.Error()) } }) } s.writeApi.WritePoint( m.ToPoint(s.meta_as_tags), ) return nil } func (s *InfluxAsyncSink) Flush() error { cclog.ComponentDebug(s.name, "Flushing") s.writeApi.Flush() if s.customFlushInterval != 0 && s.flushTimer != nil { s.flushTimer = nil } return nil } func (s *InfluxAsyncSink) Close() { cclog.ComponentDebug(s.name, "Closing InfluxDB connection") s.writeApi.Flush() s.client.Close() } func NewInfluxAsyncSink(name string, config json.RawMessage) (Sink, error) { s := new(InfluxAsyncSink) s.name = fmt.Sprintf("InfluxSink(%s)", name) // Set default for maximum number of points sent to server in single request. s.config.BatchSize = 0 s.influxRetryInterval = 0 //s.config.InfluxRetryInterval = "1s" s.influxMaxRetryTime = 0 //s.config.InfluxMaxRetryTime = "168h" s.config.InfluxMaxRetries = 0 s.config.InfluxExponentialBase = 0 s.config.FlushInterval = 0 s.config.CustomFlushInterval = "" s.customFlushInterval = time.Duration(0) s.config.MaxRetryAttempts = 1 // Default retry intervals (in seconds) // 1 2 // 2 4 // 4 8 // 8 16 // 16 32 // 32 64 // 64 128 // 128 256 // 256 512 // 512 1024 // 1024 2048 // 2048 4096 // 4096 8192 // 8192 16384 // 16384 32768 // 32768 65536 // 65536 131072 // 131072 262144 // 262144 524288 if len(config) > 0 { err := json.Unmarshal(config, &s.config) if err != nil { return nil, err } } if len(s.config.Port) == 0 { 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") } if len(s.config.Organization) == 0 { 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") } // 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 } toUint := func(duration string, def uint) uint { t, err := time.ParseDuration(duration) if err == nil { return uint(t.Milliseconds()) } return def } s.influxRetryInterval = toUint(s.config.InfluxRetryInterval, s.influxRetryInterval) s.influxMaxRetryTime = toUint(s.config.InfluxMaxRetryTime, s.influxMaxRetryTime) // Use a own timer for calling Flush() if len(s.config.CustomFlushInterval) > 0 { t, err := time.ParseDuration(s.config.CustomFlushInterval) if err != nil { return nil, fmt.Errorf("invalid duration in 'custom_flush_interval': %v", err) } s.customFlushInterval = t } // Connect to InfluxDB server if err := s.connect(); err != nil { return nil, fmt.Errorf("unable to connect: %v", err) } // Start background: Read from error channel s.errors = s.writeApi.Errors() go func() { for err := range s.errors { cclog.ComponentError(s.name, err.Error()) } }() return s, nil }