[exporter/clickhouseexporter]: persistent queue support

Addresses open-telemetry#27653.
fredthomsen · Oct 26, 2023 · 5065081 · 5065081
1 parent 397dbe6
commit 5065081
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 24 deletions.
diff --git a/exporter/clickhouseexporter/config.go b/exporter/clickhouseexporter/config.go
@@ -18,9 +18,7 @@ import (
 type Config struct {
 	exporterhelper.TimeoutSettings `mapstructure:",squash"`
 	exporterhelper.RetrySettings   `mapstructure:"retry_on_failure"`
-	// QueueSettings is a subset of exporterhelper.QueueSettings,
-	// because only QueueSize is user-settable.
-	QueueSettings QueueSettings `mapstructure:"sending_queue"`
+	exporterhelper.QueueSettings   `mapstructure:"sending_queue"`
 
 	// Endpoint is the clickhouse endpoint.
 	Endpoint string `mapstructure:"endpoint"`
@@ -42,12 +40,6 @@ type Config struct {
 	TTLDays uint `mapstructure:"ttl_days"`
 }
 
-// QueueSettings is a subset of exporterhelper.QueueSettings.
-type QueueSettings struct {
-	// QueueSize set the length of the sending queue
-	QueueSize int `mapstructure:"queue_size"`
-}
-
 const defaultDatabase = "default"
 
 var (
@@ -71,15 +63,11 @@ func (cfg *Config) Validate() (err error) {
 		err = errors.Join(err, e)
 	}
 
-	return err
-}
-
-func (cfg *Config) enforcedQueueSettings() exporterhelper.QueueSettings {
-	return exporterhelper.QueueSettings{
-		Enabled:      true,
-		NumConsumers: 1,
-		QueueSize:    cfg.QueueSettings.QueueSize,
+	if err := cfg.QueueSettings.Validate(); err != nil {
+		err = errors.Join(err, e)
 	}
+
+	return err
 }
 
 func (cfg *Config) buildDSN(database string) (string, error) {

diff --git a/exporter/clickhouseexporter/config_test.go b/exporter/clickhouseexporter/config_test.go
@@ -31,6 +31,8 @@ func TestLoadConfig(t *testing.T) {
 	defaultCfg := createDefaultConfig()
 	defaultCfg.(*Config).Endpoint = defaultEndpoint
 
+	storageID := component.NewIDWithName(component.Type("file_storage"), "clickhouse")
+
 	tests := []struct {
 		id       component.ID
 		expected component.Config
@@ -63,8 +65,11 @@ func TestLoadConfig(t *testing.T) {
 					Multiplier:          backoff.DefaultMultiplier,
 				},
 				ConnectionParams: map[string]string{},
-				QueueSettings: QueueSettings{
-					QueueSize: 100,
+				QueueSettings: exporterhelper.QueueSettings{
+					Enabled:      true,
+					NumConsumers: 10,
+					QueueSize:    100,
+					StorageID:    &storageID,
 				},
 			},
 		},

diff --git a/exporter/clickhouseexporter/factory.go b/exporter/clickhouseexporter/factory.go
@@ -30,7 +30,7 @@ func NewFactory() exporter.Factory {
 func createDefaultConfig() component.Config {
 	return &Config{
 		TimeoutSettings:  exporterhelper.NewDefaultTimeoutSettings(),
-		QueueSettings:    QueueSettings{QueueSize: exporterhelper.NewDefaultQueueSettings().QueueSize},
+		QueueSettings:    exporterhelper.NewDefaultQueueSettings(),
 		RetrySettings:    exporterhelper.NewDefaultRetrySettings(),
 		ConnectionParams: map[string]string{},
 		Database:         defaultDatabase,
@@ -62,7 +62,7 @@ func createLogsExporter(
 		exporterhelper.WithStart(exporter.start),
 		exporterhelper.WithShutdown(exporter.shutdown),
 		exporterhelper.WithTimeout(c.TimeoutSettings),
-		exporterhelper.WithQueue(c.enforcedQueueSettings()),
+		exporterhelper.WithQueue(c.QueueSettings),
 		exporterhelper.WithRetry(c.RetrySettings),
 	)
 }
@@ -88,7 +88,7 @@ func createTracesExporter(
 		exporterhelper.WithStart(exporter.start),
 		exporterhelper.WithShutdown(exporter.shutdown),
 		exporterhelper.WithTimeout(c.TimeoutSettings),
-		exporterhelper.WithQueue(c.enforcedQueueSettings()),
+		exporterhelper.WithQueue(c.QueueSettings),
 		exporterhelper.WithRetry(c.RetrySettings),
 	)
 }
@@ -112,7 +112,7 @@ func createMetricExporter(
 		exporterhelper.WithStart(exporter.start),
 		exporterhelper.WithShutdown(exporter.shutdown),
 		exporterhelper.WithTimeout(c.TimeoutSettings),
-		exporterhelper.WithQueue(c.enforcedQueueSettings()),
+		exporterhelper.WithQueue(c.QueueSettings),
 		exporterhelper.WithRetry(c.RetrySettings),
 	)
 }
diff --git a/exporter/clickhouseexporter/testdata/config.yaml b/exporter/clickhouseexporter/testdata/config.yaml
@@ -15,6 +15,8 @@ clickhouse/full:
     max_interval: 30s
     max_elapsed_time: 300s
   sending_queue:
+    enabled: true
     queue_size: 100
+    storage: file_storage/clickhouse
 clickhouse/invalid-endpoint:
-  endpoint: 127.0.0.1:9000
+  endpoint: 127.0.0.1:9000