Use maps instead of arrays for custom fields

We are now using maps instead of arrays to store the custom fields in
ClickHouse.

Author: ricoberger

README.md

@@ -27,10 +27,8 @@ CREATE TABLE IF NOT EXISTS logs.logs_local ON CLUSTER `{cluster}`
     `pod_name` LowCardinality(String),
     `container_name` LowCardinality(String),
     `host` LowCardinality(String),
-    `fields_string.key` Array(LowCardinality(String)),
-    `fields_string.value` Array(String) CODEC(ZSTD(1)),
-    `fields_number.key` Array(LowCardinality(String)),
-    `fields_number.value` Array(Float64),
+    `fields_string` Map(LowCardinality(String), String),
+    `fields_number` Map(LowCardinality(String), Float64),
     `log` String CODEC(ZSTD(1))
 )
 ENGINE = ReplicatedMergeTree

cluster/clickhouse-operator/clickhouse-operator.yaml

@@ -62,4 +62,4 @@ spec:
             - name: dockerhub-registry
           containers:
             - name: clickhouse
-              image: yandex/clickhouse-server:21.12.3.32
+              image: clickhouse/clickhouse-server:22.8.1.2097

cluster/clickhouse/clickhouse.yaml

@@ -1,6 +1,6 @@
 ---
 # Setup max number of unavailable pods in StatefulSet
-apiVersion: policy/v1beta1
+apiVersion: policy/v1
 kind: PodDisruptionBudget
 metadata:
   name: zookeeper

cluster/clickhouse/zookeeper-pdb.yaml

@@ -24,7 +24,7 @@ containerdConfigPatches:
     endpoint = ["http://${reg_name}:${reg_port}"]
 nodes:
 - role: control-plane
-  image: kindest/node:v1.20.7
+  image: kindest/node:v1.25.2
 EOF
 
 # connect the registry to the cluster network

cluster/cluster.sh

@@ -18,7 +18,7 @@ spec:
       serviceAccountName: fluent-bit
       serviceAccount: fluent-bit
       containers:
-        - image: fluent/fluent-bit:1.9.1
+        - image: fluent/fluent-bit:1.9.9
           imagePullPolicy: IfNotPresent
           name: fluent-bit
           ports:

cluster/fluent-bit/ingester/fluent-bit-ds.yaml

@@ -57,8 +57,6 @@ data:
         Database              logs
         Username              admin
         Password              admin
-        Write_Timeout         20
-        Read_Timeout          10
         Async_Insert          true
         Wait_For_Async_Insert true
         Batch_Size            1000

cluster/fluent-bit/plugin/fluent-bit-cm.yaml

@@ -1,4 +1,4 @@
-FROM golang:1.17.3-alpine3.14 as build
+FROM golang:1.19.2-alpine3.16 as build
 RUN apk update && apk add git make
 WORKDIR /ingester
 COPY go.mod go.sum ./

cmd/ingester/Dockerfile

@@ -14,8 +14,10 @@ An example Deployment for the ClickHouse ingester can be found in the [ingester.
 | `--clickhouse.database` | `CLICKHOUSE_DATABASE` | ClickHouse database name. | `logs` |
 | `--clickhouse.username` | `CLICKHOUSE_USERNAME` | ClickHouse username for the connection. | |
 | `--clickhouse.password` | `CLICKHOUSE_PASSWORD` | ClickHouse password for the connection. | |
-| `--clickhouse.write-timeout` | `CLICKHOUSE_WRITE_TIMEOUT` | ClickHouse write timeout for the connection. | `10` |
-| `--clickhouse.read-timeout` | `CLICKHOUSE_READ_TIMEOUT` | ClickHouse read timeout for the connection. | `10` |
+| `--clickhouse.dial-timeout` | `CLICKHOUSE_DIAL_TIMEOUT` | ClickHouse dial timeout. | `10s` |
+| `--clickhouse.conn-max-lifetime` | `CLICKHOUSE_CONN_MAX_LIFETIME` | ClickHouse maximum connection lifetime. | `1h` |
+| `--clickhouse.max-idle-conns` | `CLICKHOUSE_MAX_IDLE_CONNS` | ClickHouse maximum number of idle connections. | `5` |
+| `--clickhouse.max-open-conns` | `CLICKHOUSE_MAX_OPEN_CONNS` | ClickHouse maximum number of open connections. | `10` |
 | `--clickhouse.async-insert` | `CLICKHOUSE_ASYNC_INSERT` | Enable async inserts. | `false` |
 | `--clickhouse.wait-for-async-insert` | `CLICKHOUSE_WAIT_FOR_ASYNC_INSERT` | Wait for async inserts. | `false` |
 | `--clickhouse.batch-size` | `CLICKHOUSE_BATCH_SIZE` | The size for how many log lines should be buffered, before they are written to ClickHouse. | `100000` |

cmd/ingester/README.md

@@ -25,8 +25,10 @@ var (
 	clickhouseUsername           string
 	clickhousePassword           string
 	clickhouseDatabase           string
-	clickhouseWriteTimeout       string
-	clickhouseReadTimeout        string
+	clickhouseDialTimeout        string
+	clickhouseConnMaxLifetime    string
+	clickhouseMaxIdleConns       int
+	clickhouseMaxOpenConns       int
 	clickhouseAsyncInsert        bool
 	clickhouseWaitForAsyncInsert bool
 	clickhouseBatchSize          int64
@@ -65,14 +67,32 @@ func init() {
 		defaultClickHouseDatabase = os.Getenv("CLICKHOUSE_DATABASE")
 	}
 
-	defaultClickHouseWriteTimeout := "10"
-	if os.Getenv("CLICKHOUSE_WRITE_TIMEOUT") != "" {
-		defaultClickHouseWriteTimeout = os.Getenv("CLICKHOUSE_WRITE_TIMEOUT")
+	defaultClickHouseDialTimeout := "10s"
+	if os.Getenv("CLICKHOUSE_DIAL_TIMEOUT") != "" {
+		defaultClickHouseDialTimeout = os.Getenv("CLICKHOUSE_DIAL_TIMEOUT")
 	}
 
-	defaultClickHouseReadTimeout := "10"
-	if os.Getenv("CLICKHOUSE_READ_TIMEOUT") != "" {
-		defaultClickHouseReadTimeout = os.Getenv("CLICKHOUSE_READ_TIMEOUT")
+	defaultClickHouseConnMaxLifetime := "1h"
+	if os.Getenv("CLICKHOUSE_CONN_MAX_LIFETIME") != "" {
+		defaultClickHouseConnMaxLifetime = os.Getenv("CLICKHOUSE_CONN_MAX_LIFETIME")
+	}
+
+	defaultClickHouseMaxIdleConns := 5
+	if os.Getenv("CLICKHOUSE_MAX_IDLE_CONNS") != "" {
+		defaultClickHouseMaxIdleConnsString := os.Getenv("CLICKHOUSE_MAX_IDLE_CONNS")
+		defaultClickHouseMaxIdleConnsParsed, err := strconv.Atoi(defaultClickHouseMaxIdleConnsString)
+		if err == nil && defaultClickHouseMaxIdleConnsParsed > 0 {
+			defaultClickHouseMaxIdleConns = defaultClickHouseMaxIdleConnsParsed
+		}
+	}
+
+	defaultClickHouseMaxOpenConns := 10
+	if os.Getenv("CLICKHOUSE_MAX_OPEN_CONNS") != "" {
+		defaultClickHouseMaxOpenConnsString := os.Getenv("CLICKHOUSE_MAX_OPEN_CONNS")
+		defaultClickHouseMaxOpenConnsParsed, err := strconv.Atoi(defaultClickHouseMaxOpenConnsString)
+		if err == nil && defaultClickHouseMaxOpenConnsParsed > 0 {
+			defaultClickHouseMaxOpenConns = defaultClickHouseMaxOpenConnsParsed
+		}
 	}
 
 	defaultClickHouseAsyncInsert := false
@@ -156,8 +176,10 @@ func init() {
 	flag.StringVar(&clickhouseUsername, "clickhouse.username", defaultClickHouseUsername, "ClickHouse username for the connection.")
 	flag.StringVar(&clickhousePassword, "clickhouse.password", defaultClickHousePassword, "ClickHouse password for the connection.")
 	flag.StringVar(&clickhouseDatabase, "clickhouse.database", defaultClickHouseDatabase, "ClickHouse database name.")
-	flag.StringVar(&clickhouseWriteTimeout, "clickhouse.write-timeout", defaultClickHouseWriteTimeout, "ClickHouse write timeout for the connection.")
-	flag.StringVar(&clickhouseReadTimeout, "clickhouse.read-timeout", defaultClickHouseReadTimeout, "ClickHouse read timeout for the connection.")
+	flag.StringVar(&clickhouseDialTimeout, "clickhouse.dial-timeout", defaultClickHouseDialTimeout, "ClickHouse dial timeout.")
+	flag.StringVar(&clickhouseConnMaxLifetime, "clickhouse.conn-max-lifetime", defaultClickHouseConnMaxLifetime, "ClickHouse maximum connection lifetime.")
+	flag.IntVar(&clickhouseMaxIdleConns, "clickhouse.max-idle-conns", defaultClickHouseMaxIdleConns, "ClickHouse maximum number of idle connections.")
+	flag.IntVar(&clickhouseMaxOpenConns, "clickhouse.max-open-conns", defaultClickHouseMaxOpenConns, "ClickHouse maximum number of open connections.")
 	flag.BoolVar(&clickhouseAsyncInsert, "clickhouse.async-insert", defaultClickHouseAsyncInsert, "Enable async inserts.")
 	flag.BoolVar(&clickhouseWaitForAsyncInsert, "clickhouse.wait-for-async-insert", defaultClickHouseWaitForAsyncInsert, "Wait for async inserts.")
 	flag.Int64Var(&clickhouseBatchSize, "clickhouse.batch-size", defaultClickHouseBatchSize, "The size for how many log lines should be buffered, before they are written to ClickHouse.")
@@ -201,7 +223,7 @@ func main() {
 		},
 	}
 
-	logger, err := zapConfig.Build()
+	logger, err := zapConfig.Build(zap.AddCaller(), zap.AddCallerSkip(1))
 	if err != nil {
 		panic(err)
 	}
@@ -225,7 +247,7 @@ func main() {
 
 	log.Info(nil, "Version information", version.Info()...)
 	log.Info(nil, "Build context", version.BuildContext()...)
-	log.Info(nil, "Clickhouse configuration", zap.String("clickhouseAddress", clickhouseAddress), zap.String("clickhouseUsername", clickhouseUsername), zap.String("clickhousePassword", "*****"), zap.String("clickhouseDatabase", clickhouseDatabase), zap.String("clickhouseWriteTimeout", clickhouseWriteTimeout), zap.String("clickhouseReadTimeout", clickhouseReadTimeout), zap.Int64("clickhouseBatchSize", clickhouseBatchSize), zap.Duration("clickhouseFlushInterval", clickhouseFlushInterval))
+	log.Info(nil, "Clickhouse configuration", zap.String("clickhouseAddress", clickhouseAddress), zap.String("clickhouseUsername", clickhouseUsername), zap.String("clickhousePassword", "*****"), zap.String("clickhouseDatabase", clickhouseDatabase), zap.String("clickhouseDialTimeout", clickhouseDialTimeout), zap.String("clickhouseConnMaxLifetime", clickhouseConnMaxLifetime), zap.Int("clickhouseMaxIdleConns", clickhouseMaxIdleConns), zap.Int("clickhouseMaxOpenConns", clickhouseMaxOpenConns), zap.Int64("clickhouseBatchSize", clickhouseBatchSize), zap.Duration("clickhouseFlushInterval", clickhouseFlushInterval))
 	log.Info(nil, "Kafka configuration", zap.String("kafkaBrokers", kafkaBrokers), zap.String("kafkaGroup", kafkaGroup), zap.String("kafkaVersion", kafkaVersion), zap.String("kafkaTopics", kafkaTopics))
 
 	// Create a http server, which can be used for the liveness and readiness probe in Kubernetes. The server also
@@ -246,7 +268,7 @@ func main() {
 	// Create a new client for the configured ClickHouse instance. Then pass the ClickHouse client to the Run function
 	// of the Kafka package, which listens for message in the configured Kafka instance. These messages are then written
 	// to ClickHouse via the created ClickHouse client.
-	client, err := clickhouse.NewClient(clickhouseAddress, clickhouseUsername, clickhousePassword, clickhouseDatabase, clickhouseWriteTimeout, clickhouseReadTimeout, clickhouseAsyncInsert, clickhouseWaitForAsyncInsert)
+	client, err := clickhouse.NewClient(clickhouseAddress, clickhouseUsername, clickhousePassword, clickhouseDatabase, clickhouseDialTimeout, clickhouseConnMaxLifetime, clickhouseMaxIdleConns, clickhouseMaxOpenConns, clickhouseAsyncInsert, clickhouseWaitForAsyncInsert)
 	if err != nil {
 		log.Fatal(nil, "Could not create ClickHouse client", zap.Error(err))
 	}

cmd/ingester/main.go

@@ -1,11 +1,11 @@
-FROM golang:1.17.3 as build
+FROM golang:1.19.2 as build
 WORKDIR /root
 COPY go.mod go.sum ./
 RUN go mod download
 COPY . .
 RUN make build-plugin
 
-FROM fluent/fluent-bit:1.9.1
+FROM fluent/fluent-bit:1.9.9
 COPY --from=build /root/out_clickhouse.so /fluent-bit/bin/
 EXPOSE 2020
 CMD ["/fluent-bit/bin/fluent-bit", "--plugin", "/fluent-bit/bin/out_clickhouse.so", "--config", "/fluent-bit/etc/fluent-bit.conf"]

cmd/plugin/Dockerfile

@@ -14,8 +14,10 @@ An example configuration file can be found in the [fluent-bit-cm.yaml](../../clu
 | `Database` | The name of the database for the logs. | `logs` |
 | `Username` | The username, to authenticate to ClickHouse. | |
 | `Password` | The password, to authenticate to ClickHouse. | |
-| `Write_Timeout` | The write timeout for ClickHouse. | `10` |
-| `Read_Timeout` | The read timeout for ClickHouse. | `10` |
+| `Dial_Timeout` | ClickHouse dial timeout. | `10s` |
+| `Conn_Max_Lifetime` | ClickHouse maximum connection lifetime. | `1h` |
+| `Max_Idle_Conns` | ClickHouse maximum number of idle connections. | `5` |
+| `Max_Open_Conns` | ClickHouse maximum number of open connections. | `10` |
 | `Async_Insert` | Use async inserts to write logs into ClickHouse. | `false` |
 | `Wait_For_Async_Insert` | Wait for the async insert operation. | `false` |
 | `Batch_Size` | The size for how many log lines should be buffered, before they are written to ClickHouse. | `10000` |

cmd/plugin/README.md

@@ -24,11 +24,13 @@ import (
 )
 
 const (
-	defaultDatabase      string        = "logs"
-	defaultWriteTimeout  string        = "10"
-	defaultReadTimeout   string        = "10"
-	defaultBatchSize     int64         = 10000
-	defaultFlushInterval time.Duration = 60 * time.Second
+	defaultDatabase        string        = "logs"
+	defaultDialTimeout     string        = "10s"
+	defaultConnMaxLifetime string        = "1h"
+	defaultMaxIdleConns    int           = 5
+	defaultMaxOpenConns    int           = 10
+	defaultBatchSize       int64         = 10000
+	defaultFlushInterval   time.Duration = 60 * time.Second
 )
 
 var (
@@ -80,7 +82,7 @@ func FLBPluginInit(plugin unsafe.Pointer) int {
 		},
 	}
 
-	logger, err := zapConfig.Build()
+	logger, err := zapConfig.Build(zap.AddCaller(), zap.AddCallerSkip(1))
 	if err != nil {
 		panic(err)
 	}
@@ -101,14 +103,28 @@ func FLBPluginInit(plugin unsafe.Pointer) int {
 
 	password := output.FLBPluginConfigKey(plugin, "password")
 
-	writeTimeout := output.FLBPluginConfigKey(plugin, "write_timeout")
-	if writeTimeout == "" {
-		writeTimeout = defaultReadTimeout
+	dialTimeout := output.FLBPluginConfigKey(plugin, "dial_timeout")
+	if dialTimeout == "" {
+		dialTimeout = defaultDialTimeout
 	}
 
-	readTimeout := output.FLBPluginConfigKey(plugin, "read_timeout")
-	if readTimeout == "" {
-		readTimeout = defaultReadTimeout
+	connMaxLifetime := output.FLBPluginConfigKey(plugin, "conn_max_lifetime")
+	if connMaxLifetime == "" {
+		connMaxLifetime = defaultConnMaxLifetime
+	}
+
+	maxIdleConnsStr := output.FLBPluginConfigKey(plugin, "max_idle_conns")
+	maxIdleConns, err := strconv.Atoi(maxIdleConnsStr)
+	if err != nil || maxIdleConns < 0 {
+		log.Warn(nil, "Could not parse maxIdleConns setting, use default setting", zap.Error(err), zap.Int("default", defaultMaxIdleConns))
+		maxIdleConns = defaultMaxIdleConns
+	}
+
+	maxOpenConnsStr := output.FLBPluginConfigKey(plugin, "max_open_conns")
+	maxOpenConns, err := strconv.Atoi(maxOpenConnsStr)
+	if err != nil || maxOpenConns < 0 {
+		log.Warn(nil, "Could not parse maxOpenConns setting, use default setting", zap.Error(err), zap.Int("default", defaultMaxOpenConns))
+		maxOpenConns = defaultMaxOpenConns
 	}
 
 	asyncInsertStr := output.FLBPluginConfigKey(plugin, "async_insert")
@@ -140,9 +156,9 @@ func FLBPluginInit(plugin unsafe.Pointer) int {
 	forceNumberFieldsStr := output.FLBPluginConfigKey(plugin, "force_number_fields")
 	forceNumberFields = strings.Split(forceNumberFieldsStr, ",")
 
-	log.Info(nil, "Clickhouse configuration", zap.String("clickhouseAddress", address), zap.String("clickhouseUsername", username), zap.String("clickhousePassword", "*****"), zap.String("clickhouseDatabase", database), zap.String("clickhouseWriteTimeout", writeTimeout), zap.String("clickhouseReadTimeout", readTimeout), zap.Int64("clickhouseBatchSize", batchSize), zap.Duration("clickhouseFlushInterval", flushInterval), zap.Strings("forceNumberFields", forceNumberFields))
+	log.Info(nil, "Clickhouse configuration", zap.String("address", address), zap.String("username", username), zap.String("password", "*****"), zap.String("database", database), zap.String("dialTimeout", dialTimeout), zap.String("connMaxLifetime", connMaxLifetime), zap.Int("maxIdleConns", maxIdleConns), zap.Int("maxOpenConns", maxOpenConns), zap.Int64("batchSize", batchSize), zap.Duration("flushInterval", flushInterval))
 
-	clickhouseClient, err := clickhouse.NewClient(address, username, password, database, writeTimeout, readTimeout, asyncInsert, waitForAsyncInsert)
+	clickhouseClient, err := clickhouse.NewClient(address, username, password, database, dialTimeout, connMaxLifetime, maxIdleConns, maxOpenConns, asyncInsert, waitForAsyncInsert)
 	if err != nil {
 		log.Fatal(nil, "Could not create ClickHouse client", zap.Error(err))
 		return output.FLB_ERROR
@@ -187,7 +203,9 @@ func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int
 		}
 
 		row := clickhouse.Row{
-			Timestamp: timestamp,
+			Timestamp:    timestamp,
+			FieldsString: make(map[string]string),
+			FieldsNumber: make(map[string]float64),
 		}
 
 		for k, v := range data {
@@ -260,21 +278,17 @@ func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int
 					row.Log = stringValue
 				default:
 					if isNumber {
-						row.FieldsNumber.Key = append(row.FieldsNumber.Key, k)
-						row.FieldsNumber.Value = append(row.FieldsNumber.Value, numberValue)
+						row.FieldsNumber[k] = numberValue
 					} else {
 						if contains(k, forceNumberFields) {
 							parsedNumber, err := strconv.ParseFloat(stringValue, 64)
 							if err == nil {
-								row.FieldsNumber.Key = append(row.FieldsNumber.Key, k)
-								row.FieldsNumber.Value = append(row.FieldsNumber.Value, parsedNumber)
+								row.FieldsNumber[k] = parsedNumber
 							} else {
-								row.FieldsString.Key = append(row.FieldsString.Key, k)
-								row.FieldsString.Value = append(row.FieldsString.Value, stringValue)
+								row.FieldsString[k] = stringValue
 							}
 						} else {
-							row.FieldsString.Key = append(row.FieldsString.Key, k)
-							row.FieldsString.Value = append(row.FieldsString.Value, stringValue)
+							row.FieldsString[k] = stringValue
 						}
 					}
 				}

cmd/plugin/main.go

@@ -3,7 +3,7 @@ module github.com/kobsio/klogs
 go 1.19
 
 require (
-	github.com/ClickHouse/clickhouse-go v1.5.4
+	github.com/ClickHouse/clickhouse-go/v2 v2.3.0
 	github.com/Shopify/sarama v1.37.2
 	github.com/fluent/fluent-bit-go v0.0.0-20220311094233-780004bf5562
 	github.com/prometheus/client_golang v1.13.0
@@ -13,15 +13,19 @@ require (
 )
 
 require (
+	github.com/ClickHouse/ch-go v0.47.3 // indirect
+	github.com/andybalholm/brotli v1.0.4 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
-	github.com/cloudflare/golz4 v0.0.0-20150217214814-ef862a3cdc58 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/eapache/go-resiliency v1.3.0 // indirect
 	github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 // indirect
 	github.com/eapache/queue v1.1.0 // indirect
+	github.com/go-faster/city v1.0.1 // indirect
+	github.com/go-faster/errors v0.6.1 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
+	github.com/google/uuid v1.3.0 // indirect
 	github.com/hashicorp/errwrap v1.0.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/go-uuid v1.0.3 // indirect
@@ -32,20 +36,25 @@ require (
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
 	github.com/klauspost/compress v1.15.11 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
-	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
+	github.com/paulmach/orb v0.7.1 // indirect
 	github.com/pierrec/lz4/v4 v4.1.17 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.37.0 // indirect
 	github.com/prometheus/procfs v0.8.0 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
 	github.com/rogpeppe/go-internal v1.9.0 // indirect
+	github.com/segmentio/asm v1.2.0 // indirect
+	github.com/shopspring/decimal v1.3.1 // indirect
 	github.com/ugorji/go/codec v1.1.8 // indirect
-	go.uber.org/atomic v1.7.0 // indirect
-	go.uber.org/multierr v1.6.0 // indirect
+	go.opentelemetry.io/otel v1.9.0 // indirect
+	go.opentelemetry.io/otel/trace v1.9.0 // indirect
+	go.uber.org/atomic v1.10.0 // indirect
+	go.uber.org/multierr v1.8.0 // indirect
 	golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa // indirect
 	golang.org/x/net v0.0.0-20220927171203-f486391704dc // indirect
-	golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10 // indirect
+	golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )