Merge pull request #76 from tomershafir/exporter-richer-queue-config

Exporter richer queue config

exporter/clickhouseprofileexporter/config.go

@@ -10,30 +10,17 @@ import (
 type Config struct {
 	exporterhelper.TimeoutSettings `mapstructure:",squash"`
 	configretry.BackOffConfig      `mapstructure:"retry_on_failure"`
-	QueueSettings                  `mapstructure:"sending_queue"`
+	exporterhelper.QueueSettings   `mapstructure:"sending_queue"`
 
 	// DSN is the ClickHouse server Data Source Name.
 	// For tcp protocol reference: [ClickHouse/clickhouse-go#dsn](https://github.com/ClickHouse/clickhouse-go#dsn).
 	// For http protocol reference: [mailru/go-clickhouse/#dsn](https://github.com/mailru/go-clickhouse/#dsn).
 	Dsn string `mapstructure:"dsn"`
 }
 
-type QueueSettings struct {
-	// Length of the sending queue
-	QueueSize int `mapstructure:"queue_size"`
-}
-
 var _ component.Config = (*Config)(nil)
 
 // Checks that the receiver configuration is valid
 func (cfg *Config) Validate() error {
 	return nil
 }
-
-func (cfg *Config) enforceQueueSettings() exporterhelper.QueueSettings {
-	return exporterhelper.QueueSettings{
-		Enabled:      true,
-		NumConsumers: 1,
-		QueueSize:    cfg.QueueSettings.QueueSize,
-	}
-}

exporter/clickhouseprofileexporter/factory.go

@@ -19,7 +19,7 @@ const (
 func createDefaultConfig() component.Config {
 	return &Config{
 		TimeoutSettings: exporterhelper.NewDefaultTimeoutSettings(),
-		QueueSettings:   QueueSettings{QueueSize: exporterhelper.NewDefaultQueueSettings().QueueSize},
+		QueueSettings:   exporterhelper.NewDefaultQueueSettings(),
 		BackOffConfig:   configretry.NewDefaultBackOffConfig(),
 		Dsn:             defaultDsn,
 	}
@@ -37,7 +37,7 @@ func createLogsExporter(ctx context.Context, set exporter.CreateSettings, cfg co
 		cfg,
 		exp.send,
 		exporterhelper.WithShutdown(exp.Shutdown),
-		exporterhelper.WithQueue(c.enforceQueueSettings()),
+		exporterhelper.WithQueue(c.QueueSettings),
 		exporterhelper.WithTimeout(c.TimeoutSettings),
 		exporterhelper.WithRetry(c.BackOffConfig),
 	)

exporter/qrynexporter/config.go

@@ -28,9 +28,7 @@ const (
 type Config struct {
 	exporterhelper.TimeoutSettings `mapstructure:",squash"`
 	configretry.BackOffConfig      `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"`
 
 	ClusteredClickhouse bool `mapstructure:"clustered_clickhouse"`
 
@@ -45,12 +43,6 @@ type Config struct {
 	Metrics MetricsConfig `mapstructure:"metrics"`
 }
 
-// QueueSettings is a subset of exporterhelper.QueueSettings.
-type QueueSettings struct {
-	// QueueSize set the length of the sending queue
-	QueueSize int `mapstructure:"queue_size"`
-}
-
 // LogsConfig holds the configuration for log data.
 type LogsConfig struct {
 	// AttributeLabels is the string representing attribute labels.
@@ -74,11 +66,3 @@ var _ component.Config = (*Config)(nil)
 func (cfg *Config) Validate() error {
 	return nil
 }
-
-func (cfg *Config) enforcedQueueSettings() exporterhelper.QueueSettings {
-	return exporterhelper.QueueSettings{
-		Enabled:      true,
-		NumConsumers: 1,
-		QueueSize:    cfg.QueueSettings.QueueSize,
-	}
-}

exporter/qrynexporter/config_test.go

@@ -45,8 +45,11 @@ func TestLoadConfig(t *testing.T) {
 					RandomizationFactor: 0.5,
 					Multiplier:          1.5,
 				},
-				QueueSettings: QueueSettings{
-					QueueSize: 100,
+				QueueSettings: exporterhelper.QueueSettings{
+					Enabled:      true,
+					QueueSize:    100,
+					NumConsumers: 10,
+					StorageID:    nil,
 				},
 			},
 		},

exporter/qrynexporter/factory.go

@@ -42,7 +42,7 @@ func NewFactory() exporter.Factory {
 func createDefaultConfig() component.Config {
 	return &Config{
 		TimeoutSettings: exporterhelper.NewDefaultTimeoutSettings(),
-		QueueSettings:   QueueSettings{QueueSize: exporterhelper.NewDefaultQueueSettings().QueueSize},
+		QueueSettings:   exporterhelper.NewDefaultQueueSettings(),
 		BackOffConfig:   configretry.NewDefaultBackOffConfig(),
 		DSN:             defaultDSN,
 	}
@@ -52,23 +52,23 @@ func createDefaultConfig() component.Config {
 // Traces are directly insert into clickhouse.
 func createTracesExporter(
 	ctx context.Context,
-	params exporter.CreateSettings,
+	set exporter.CreateSettings,
 	cfg component.Config,
 ) (exporter.Traces, error) {
 	c := cfg.(*Config)
-	oce, err := newTracesExporter(params.Logger, c)
+	oce, err := newTracesExporter(set.Logger, c, &set)
 	if err != nil {
 		return nil, fmt.Errorf("cannot configure qryn traces exporter: %w", err)
 	}
 
 	return exporterhelper.NewTracesExporter(
 		ctx,
-		params,
+		set,
 		cfg,
 		oce.pushTraceData,
 		exporterhelper.WithShutdown(oce.Shutdown),
 		exporterhelper.WithTimeout(c.TimeoutSettings),
-		exporterhelper.WithQueue(c.enforcedQueueSettings()),
+		exporterhelper.WithQueue(c.QueueSettings),
 		exporterhelper.WithRetry(c.BackOffConfig),
 	)
 }
@@ -81,7 +81,7 @@ func createLogsExporter(
 	cfg component.Config,
 ) (exporter.Logs, error) {
 	c := cfg.(*Config)
-	exporter, err := newLogsExporter(set.Logger, c)
+	exporter, err := newLogsExporter(set.Logger, c, &set)
 	if err != nil {
 		return nil, fmt.Errorf("cannot configure qryn logs exporter: %w", err)
 	}
@@ -93,7 +93,7 @@ func createLogsExporter(
 		exporter.pushLogsData,
 		exporterhelper.WithShutdown(exporter.Shutdown),
 		exporterhelper.WithTimeout(c.TimeoutSettings),
-		exporterhelper.WithQueue(c.enforcedQueueSettings()),
+		exporterhelper.WithQueue(c.QueueSettings),
 		exporterhelper.WithRetry(c.BackOffConfig),
 	)
 }
@@ -106,7 +106,7 @@ func createMetricsExporter(
 	cfg component.Config,
 ) (exporter.Metrics, error) {
 	c := cfg.(*Config)
-	exporter, err := newMetricsExporter(set.Logger, c)
+	exporter, err := newMetricsExporter(set.Logger, c, &set)
 	if err != nil {
 		return nil, fmt.Errorf("cannot configure qryn logs exporter: %w", err)
 	}
@@ -118,7 +118,7 @@ func createMetricsExporter(
 		exporter.pushMetricsData,
 		exporterhelper.WithShutdown(exporter.Shutdown),
 		exporterhelper.WithTimeout(c.TimeoutSettings),
-		exporterhelper.WithQueue(c.enforcedQueueSettings()),
+		exporterhelper.WithQueue(c.QueueSettings),
 		exporterhelper.WithRetry(c.BackOffConfig),
 	)
 }

exporter/qrynexporter/logs.go

@@ -10,8 +10,10 @@ import (
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/go-logfmt/logfmt"
 	"github.com/prometheus/common/model"
+	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 )
 
@@ -30,6 +32,7 @@ const (
 
 type logsExporter struct {
 	logger *zap.Logger
+	meter  metric.Meter
 
 	db clickhouse.Conn
 
@@ -39,7 +42,7 @@ type logsExporter struct {
 	cluster         bool
 }
 
-func newLogsExporter(logger *zap.Logger, cfg *Config) (*logsExporter, error) {
+func newLogsExporter(logger *zap.Logger, cfg *Config, set *exporter.CreateSettings) (*logsExporter, error) {
 	opts, err := clickhouse.ParseDSN(cfg.DSN)
 	if err != nil {
 		return nil, err
@@ -48,14 +51,20 @@ func newLogsExporter(logger *zap.Logger, cfg *Config) (*logsExporter, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &logsExporter{
+	exp := &logsExporter{
 		logger:          logger,
+		meter:           set.MeterProvider.Meter(typeStr),
 		db:              db,
 		format:          cfg.Logs.Format,
 		attributeLabels: cfg.Logs.AttributeLabels,
 		resourceLabels:  cfg.Logs.ResourceLabels,
 		cluster:         cfg.ClusteredClickhouse,
-	}, nil
+	}
+	if err := initMetrics(exp.meter); err != nil {
+		exp.logger.Error(fmt.Sprintf("failed to init metrics: %s", err.Error()))
+		return exp, err
+	}
+	return exp, nil
 }
 
 // Shutdown will shutdown the exporter.
@@ -433,11 +442,12 @@ func (e *logsExporter) pushLogsData(ctx context.Context, ld plog.Logs) error {
 	}
 
 	if err := batchSamplesAndTimeSeries(context.WithValue(ctx, "cluster", e.cluster), e.db, samples, timeSeries); err != nil {
+		otelcolExporterQrynBatchInsertDurationMillis.Record(ctx, time.Now().UnixMilli()-start.UnixMilli(), metric.WithAttributeSet(*newOtelcolAttrSetBatch(errorCodeError, dataTypeLogs)))
+		e.logger.Error(fmt.Sprintf("failed to insert batch: [%s]", err.Error()))
 		return err
 	}
-
+	otelcolExporterQrynBatchInsertDurationMillis.Record(ctx, time.Now().UnixMilli()-start.UnixMilli(), metric.WithAttributeSet(*newOtelcolAttrSetBatch(errorCodeSuccess, dataTypeLogs)))
 	e.logger.Debug("pushLogsData", zap.Int("samples", len(samples)), zap.Int("timeseries", len(timeSeries)), zap.String("cost", time.Since(start).String()))
-
 	return nil
 }
 

exporter/qrynexporter/metrics.go

@@ -13,9 +13,11 @@ import (
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/model/value"
+	"go.opentelemetry.io/collector/exporter"
 	conventions "go.opentelemetry.io/collector/model/semconv/v1.5.0"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 )
 
@@ -28,14 +30,15 @@ const (
 
 type metricsExporter struct {
 	logger *zap.Logger
+	meter  metric.Meter
 
 	db clickhouse.Conn
 
 	namespace string
 	cluster   bool
 }
 
-func newMetricsExporter(logger *zap.Logger, cfg *Config) (*metricsExporter, error) {
+func newMetricsExporter(logger *zap.Logger, cfg *Config, set *exporter.CreateSettings) (*metricsExporter, error) {
 	opts, err := clickhouse.ParseDSN(cfg.DSN)
 	if err != nil {
 		return nil, err
@@ -44,12 +47,18 @@ func newMetricsExporter(logger *zap.Logger, cfg *Config) (*metricsExporter, erro
 	if err != nil {
 		return nil, err
 	}
-	return &metricsExporter{
+	exp := &metricsExporter{
 		logger:    logger,
+		meter:     set.MeterProvider.Meter(typeStr),
 		db:        db,
 		namespace: cfg.Metrics.Namespace,
 		cluster:   cfg.ClusteredClickhouse,
-	}, nil
+	}
+	if err := initMetrics(exp.meter); err != nil {
+		exp.logger.Error(fmt.Sprintf("failed to init metrics: %s", err.Error()))
+		return exp, err
+	}
+	return exp, nil
 }
 
 // Shutdown will shutdown the exporter.
@@ -483,11 +492,12 @@ func (e *metricsExporter) pushMetricsData(ctx context.Context, md pmetric.Metric
 	}
 
 	if err := batchSamplesAndTimeSeries(context.WithValue(ctx, "cluster", e.cluster), e.db, samples, timeSeries); err != nil {
+		otelcolExporterQrynBatchInsertDurationMillis.Record(ctx, time.Now().UnixMilli()-start.UnixMilli(), metric.WithAttributeSet(*newOtelcolAttrSetBatch(errorCodeError, dataTypeMetrics)))
+		e.logger.Error(fmt.Sprintf("failed to insert batch: [%s]", err.Error()))
 		return err
 	}
-
+	otelcolExporterQrynBatchInsertDurationMillis.Record(ctx, time.Now().UnixMilli()-start.UnixMilli(), metric.WithAttributeSet(*newOtelcolAttrSetBatch(errorCodeSuccess, dataTypeMetrics)))
 	e.logger.Debug("pushMetricsData", zap.Int("samples", len(samples)), zap.Int("timeseries", len(timeSeries)), zap.String("cost", time.Since(start).String()))
-
 	return nil
 }
 

exporter/qrynexporter/self_metrics.go

@@ -0,0 +1,43 @@
+package qrynexporter
+
+import (
+	"fmt"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+const (
+	prefix = "exporter_qryn_"
+
+	errorCodeError   = "1"
+	errorCodeSuccess = ""
+
+	dataTypeLogs    = "logs"
+	dataTypeMetrics = "metrics"
+	dataTypeTraces  = "traces"
+)
+
+var (
+	otelcolExporterQrynBatchInsertDurationMillis metric.Int64Histogram
+)
+
+func initMetrics(meter metric.Meter) error {
+	var err error
+	if otelcolExporterQrynBatchInsertDurationMillis, err = meter.Int64Histogram(
+		fmt.Sprint(prefix, "batch_insert_duration_millis"),
+		metric.WithDescription("Qryn exporter batch insert duration in millis"),
+		metric.WithExplicitBucketBoundaries(0, 5, 10, 20, 50, 100, 200, 500, 1000, 5000),
+	); err != nil {
+		return err
+	}
+	return nil
+}
+
+func newOtelcolAttrSetBatch(errorCode string, dataType string) *attribute.Set {
+	s := attribute.NewSet(
+		attribute.KeyValue{Key: "error_code", Value: attribute.StringValue(errorCode)},
+		attribute.KeyValue{Key: "data_type", Value: attribute.StringValue(dataType)},
+	)
+	return &s
+}