johanels/akvorado
1
0
Fork 0
mirror of https://github.com/akvorado/akvorado.git synced 2025-12-11 22:14:02 +01:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

outlet/clickhouse: make grace period for flushing configurable

Browse Source
This commit is contained in:
Vincent Bernat
2025-11-10 16:20:43 +01:00
parent be364899cc
commit 04eccbe95f
3 changed files with 5 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
console/data/docs/02-configuration.md
View File

@@ -613,6 +613,7 @@ are configurable:
- `maximum-batch-size` defines how many flows to send to ClickHouse in a single batch at most - `maximum-batch-size` defines how many flows to send to ClickHouse in a single batch at most
- `minimum-wait-time` defines how long to wait before sending an incomplete batch - `minimum-wait-time` defines how long to wait before sending an incomplete batch
- `grace-period` defines how long to wait when flushing data to ClickHouse on shutdown
These numbers are per-worker (as defined in the Kafka component). A worker will These numbers are per-worker (as defined in the Kafka component). A worker will
send a batch of size at most `maximum-batch-size` at least every send a batch of size at most `maximum-batch-size` at least every

3
outlet/clickhouse/config.go
View File

@@ -9,6 +9,8 @@ import (
// Configuration describes the configuration for the ClickHouse exporter. // Configuration describes the configuration for the ClickHouse exporter.
type Configuration struct { type Configuration struct {
// GracePeriod defines how long to wait for flushing a batch to ClickHouse on shutdown.
GracePeriod time.Duration `validate:"min=10s"`
// MaximumBatchSize is the maximum number of rows to send to ClickHouse in one batch. // MaximumBatchSize is the maximum number of rows to send to ClickHouse in one batch.
MaximumBatchSize uint `validate:"min=1"` MaximumBatchSize uint `validate:"min=1"`
// MaximumWaitTime is the maximum number of seconds to wait before sending the current batch. // MaximumWaitTime is the maximum number of seconds to wait before sending the current batch.
@@ -22,6 +24,7 @@ const minimumBatchSizeDivider = 10
// DefaultConfiguration represents the default configuration for the ClickHouse exporter. // DefaultConfiguration represents the default configuration for the ClickHouse exporter.
func DefaultConfiguration() Configuration { func DefaultConfiguration() Configuration {
return Configuration{ return Configuration{
GracePeriod: time.Minute,
MaximumBatchSize: 50_000, MaximumBatchSize: 50_000,
MaximumWaitTime: 5 * time.Second, MaximumWaitTime: 5 * time.Second,
} }

2
outlet/clickhouse/worker.go
View File

@@ -137,7 +137,7 @@ func (w *realWorker) Flush(ctx context.Context) {
} }
// Don't use the parent context, it may be too short. // Don't use the parent context, it may be too short.
chCtx, cancel := context.WithTimeout(context.Background(), time.Minute) chCtx, cancel := context.WithTimeout(context.Background(), w.c.config.GracePeriod)
defer cancel() defer cancel()
// Send to ClickHouse in flows_XXXXX_raw. // Send to ClickHouse in flows_XXXXX_raw.