From 7d5b3c7d731c17b333eacfc53d55389bba03531d Mon Sep 17 00:00:00 2001
From: Thomas Roehl <thomas.roehl@fau.de>
Date: Mon, 7 Feb 2022 16:33:46 +0100
Subject: [PATCH] Add configurable batch size

---
 sinks/influxAsyncSink.go | 8 +++++++-
 sinks/influxAsyncSink.md | 4 +++-
 2 files changed, 10 insertions(+), 2 deletions(-)

diff --git a/sinks/influxAsyncSink.go b/sinks/influxAsyncSink.go
index 60df275..0763a4b 100644
--- a/sinks/influxAsyncSink.go
+++ b/sinks/influxAsyncSink.go
@@ -24,6 +24,7 @@ type InfluxAsyncSinkConfig struct {
 	Organization string `json:"organization,omitempty"`
 	SSL          bool   `json:"ssl,omitempty"`
 	RetentionPol string `json:"retention_policy,omitempty"`
+	BatchSize    uint   `json:"batch_size,omitempty"`
 }
 
 type InfluxAsyncSink struct {
@@ -49,8 +50,12 @@ func (s *InfluxAsyncSink) connect() error {
 		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)
+	batch := s.config.BatchSize
+	if batch == 0 {
+		batch = 100
+	}
 	s.client = influxdb2.NewClientWithOptions(uri, auth,
-		influxdb2.DefaultOptions().SetBatchSize(20).SetTLSConfig(&tls.Config{
+		influxdb2.DefaultOptions().SetBatchSize(batch).SetTLSConfig(&tls.Config{
 			InsecureSkipVerify: true,
 		}))
 	s.writeApi = s.client.WriteAPI(s.config.Organization, s.config.Database)
@@ -59,6 +64,7 @@ func (s *InfluxAsyncSink) connect() error {
 
 func (s *InfluxAsyncSink) Init(config json.RawMessage) error {
 	s.name = "InfluxSink"
+	s.config.BatchSize = 100
 	if len(config) > 0 {
 		err := json.Unmarshal(config, &s.config)
 		if err != nil {
diff --git a/sinks/influxAsyncSink.md b/sinks/influxAsyncSink.md
index 0a75eea..286c93c 100644
--- a/sinks/influxAsyncSink.md
+++ b/sinks/influxAsyncSink.md
@@ -17,6 +17,7 @@ The `influxasync` sink uses the official [InfluxDB golang client](https://pkg.go
     "password" : "examplepw",
     "organization": "myorg",
     "ssl": true,
+    "batch_size": 200,
   }
 }
 ```
@@ -29,4 +30,5 @@ The `influxasync` sink uses the official [InfluxDB golang client](https://pkg.go
 - `user`: Username for basic authentification
 - `password`: Password for basic authentification
 - `organization`: Organization in the InfluxDB
-- `ssl`: Use SSL connection
\ No newline at end of file
+- `ssl`: Use SSL connection
+- `batch_size`: batch up metrics internally, default 100
\ No newline at end of file