output/influxdb: use the async periodic flusher

Author: codebien

output/influxdb/config.go

@@ -56,11 +56,20 @@ type Config struct {
 // NewConfig creates a new InfluxDB output config with some default values.
 func NewConfig() Config {
 	c := Config{
-		Addr:             null.NewString("http://localhost:8086", false),
-		DB:               null.NewString("k6", false),
-		TagsAsFields:     []string{"vu", "iter", "url"},
-		ConcurrentWrites: null.NewInt(10, false),
-		PushInterval:     types.NewNullDuration(time.Second, false),
+		Addr:         null.NewString("http://localhost:8086", false),
+		DB:           null.NewString("k6", false),
+		TagsAsFields: []string{"vu", "iter", "url"},
+		PushInterval: types.NewNullDuration(time.Second, false),
+
+		// The minimum value of pow(2, N) for handling a stressful situation
+		// with the default push interval set to 1s.
+		// Concurrency is not expected for the normal use-case,
+		// the response time should be lower than the push interval set value.
+		// In case of spikes, the response time could go around 2s,
+		// higher values will highlight a not sustainable situation
+		// and the user should adjust the executed script
+		// or the configuration based on the environment and rate expected.
+		ConcurrentWrites: null.NewInt(4, false),
 	}
 	return c
 }

output/influxdb/output.go

@@ -58,9 +58,8 @@ type Output struct {
 	Config    Config
 	BatchConf client.BatchPointsConfig
 
-	logger      logrus.FieldLogger
-	semaphoreCh chan struct{}
-	fieldKinds  map[string]FieldKind
+	logger     logrus.FieldLogger
+	fieldKinds map[string]FieldKind
 }
 
 // New returns new influxdb output
@@ -87,11 +86,10 @@ func newOutput(params output.Params) (*Output, error) {
 		logger: params.Logger.WithFields(logrus.Fields{
 			"output": "InfluxDBv1",
 		}),
-		Client:      cl,
-		Config:      conf,
-		BatchConf:   batchConf,
-		semaphoreCh: make(chan struct{}, conf.ConcurrentWrites.Int64),
-		fieldKinds:  fldKinds,
+		Client:     cl,
+		Config:     conf,
+		BatchConf:  batchConf,
+		fieldKinds: fldKinds,
 	}, err
 }
 
@@ -178,12 +176,15 @@ func (o *Output) Start() error {
 	// usually means we're either a non-admin user to an existing DB or connecting over UDP.
 	_, err := o.Client.Query(client.NewQuery("CREATE DATABASE "+o.BatchConf.Database, "", ""))
 	if err != nil {
-		o.logger.WithError(err).Debug("InfluxDB: Couldn't create database; most likely harmless")
+		o.logger.WithError(err).Debug("Couldn't create database; most likely harmless")
 	}
 
-	pf, err := output.NewPeriodicFlusher(time.Duration(o.Config.PushInterval.Duration), o.flushMetrics)
+	pf, err := output.NewAsyncPeriodicFlusher(
+		time.Duration(o.Config.PushInterval.Duration),
+		int(o.Config.ConcurrentWrites.Int64),
+		o.flushMetrics)
 	if err != nil {
-		return err //nolint:wrapcheck
+		return err
 	}
 	o.logger.Debug("Started!")
 	o.periodicFlusher = pf
@@ -201,11 +202,11 @@ func (o *Output) Stop() error {
 
 func (o *Output) flushMetrics() {
 	samples := o.GetBufferedSamples()
+	if len(samples) < 1 {
+		o.logger.Debug("Any buffered samples, skipping the flush operation")
+		return
+	}
 
-	o.semaphoreCh <- struct{}{}
-	defer func() {
-		<-o.semaphoreCh
-	}()
 	o.logger.Debug("Committing...")
 	o.logger.WithField("samples", len(samples)).Debug("Writing...")
 
@@ -219,7 +220,13 @@ func (o *Output) flushMetrics() {
 	startTime := time.Now()
 	if err := o.Client.Write(batch); err != nil {
 		o.logger.WithError(err).Error("Couldn't write stats")
+		return
 	}
 	t := time.Since(startTime)
 	o.logger.WithField("t", t).Debug("Batch written!")
+
+	if t > time.Duration(o.Config.PushInterval.Duration) {
+		o.logger.WithField("t", t).
+			Warn("The flush operation took higher than the expected set push interval. If you see this message multiple times then the setup or configuration need to be adjusted to achieve a sustainable rate.") //nolint:lll
+	}
 }