add tests, changelog, fmt/lint

open-telemetry · dmitryax · Jun 27, 2025 · May 29, 2025 · May 30, 2025 · May 30, 2025
commit 91106bd3174eb2c8e82c7b2daff50ba1d4328e18
@@ -0,0 +1,32 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: 'clickhouseexporter'
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Update exporter to optionally use the new JSON column type for logs and traces
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: []
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  Added a feature gate to enable a JSON pipeline for logs and traces.
+  This feature gate ID is `clickhouse.json`, and will automatically use the new
+  DDL and column type on supported server versions.
+  You may also need to add `enable_json_type=1` to your connection
+  settings, depending on the server version.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]
@@ -6,7 +6,6 @@ package clickhouseexporter // import "github.com/open-telemetry/opentelemetry-co
 import (
 	"context"
 	"fmt"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal/chjson"
 	"time"
 
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
@@ -15,6 +14,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal/chjson"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal/sqltemplates"
 )
 
@@ -31,7 +31,7 @@ type logsJSONExporter struct {
 	spanHexBufferPool            *internal.ExporterStructPool[[]byte]
 }
 
-func newLogsJSONExporter(logger *zap.Logger, cfg *Config) (*logsJSONExporter, error) {
+func newLogsJSONExporter(logger *zap.Logger, cfg *Config) *logsJSONExporter {
 	numConsumers := cfg.QueueSettings.NumConsumers
 
 	newJSONBuffer := func() (*chjson.JSONBuffer, error) {
@@ -56,7 +56,7 @@ func newLogsJSONExporter(logger *zap.Logger, cfg *Config) (*logsJSONExporter, er
 		logAttributesBufferPool:      logAttributesBufferPool,
 		traceHexBufferPool:           traceHexBufferPool,
 		spanHexBufferPool:            spanHexBufferPool,
-	}, nil
+	}
 }
 
 func (e *logsJSONExporter) start(ctx context.Context, _ component.Host) error {

@@ -27,8 +27,7 @@ func testLogsJSONExporter(t *testing.T) {
 }
 
 func newTestLogsJSONExporter(t *testing.T, dsn string, fns ...func(*Config)) *logsJSONExporter {
-	exporter, err := newLogsJSONExporter(zaptest.NewLogger(t), withTestExporterConfig(fns...)(dsn))
-	require.NoError(t, err)
+	exporter := newLogsJSONExporter(zaptest.NewLogger(t), withTestExporterConfig(fns...)(dsn))
 
 	require.NoError(t, exporter.start(context.Background(), nil))
 

@@ -6,7 +6,6 @@ package clickhouseexporter // import "github.com/open-telemetry/opentelemetry-co
 import (
 	"context"
 	"fmt"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal/chjson"
 	"time"
 
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
@@ -15,6 +14,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal/chjson"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal/sqltemplates"
 )
 
@@ -34,7 +34,7 @@ type tracesJSONExporter struct {
 	linksHexBufferPool           *internal.ExporterStructPool[[]byte]
 }
 
-func newTracesJSONExporter(logger *zap.Logger, cfg *Config) (*tracesJSONExporter, error) {
+func newTracesJSONExporter(logger *zap.Logger, cfg *Config) *tracesJSONExporter {
 	numConsumers := cfg.QueueSettings.NumConsumers
 
 	newJSONBuffer := func() (*chjson.JSONBuffer, error) {
@@ -65,7 +65,7 @@ func newTracesJSONExporter(logger *zap.Logger, cfg *Config) (*tracesJSONExporter
 		spanHexBufferPool:            spanHexBufferPool,
 		parentSpanHexBufferPool:      parentSpanHexBufferPool,
 		linksHexBufferPool:           linksHexBufferPool,
-	}, nil
+	}
 }
 
 func (e *tracesJSONExporter) start(ctx context.Context, _ component.Host) error {

@@ -27,8 +27,7 @@ func testTracesJSONExporter(t *testing.T) {
 }
 
 func newTestTracesJSONExporter(t *testing.T, dsn string, fns ...func(*Config)) *tracesJSONExporter {
-	exporter, err := newTracesJSONExporter(zaptest.NewLogger(t), withTestExporterConfig(fns...)(dsn))
-	require.NoError(t, err)
+	exporter := newTracesJSONExporter(zaptest.NewLogger(t), withTestExporterConfig(fns...)(dsn))
 
 	require.NoError(t, exporter.start(context.Background(), nil))
 

@@ -7,14 +7,17 @@ package clickhouseexporter // import "github.com/open-telemetry/opentelemetry-co
 
 import (
 	"context"
-	"fmt"
+
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/featuregate"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal/metadata"
 )
 
+var featureGateJSON = featuregate.GlobalRegistry().MustRegister("clickhouse.json", featuregate.StageAlpha)
+
 // NewFactory creates a factory for the ClickHouse exporter.
 func NewFactory() exporter.Factory {
 	return exporter.NewFactory(
@@ -34,12 +37,24 @@ func createLogsExporter(
 	c := cfg.(*Config)
 	c.collectorVersion = set.BuildInfo.Version
 
-	// TODO: branch on feature flag or config
-	exp, err := newLogsJSONExporter(set.Logger, c)
-	if err != nil {
-		return nil, fmt.Errorf("cannot configure clickhouse json logs exporter: %w", err)
+	if featureGateJSON.IsEnabled() {
+		exp := newLogsJSONExporter(set.Logger, c)
+
+		return exporterhelper.NewLogs(
+			ctx,
+			set,
+			cfg,
+			exp.pushLogsData,
+			exporterhelper.WithStart(exp.start),
+			exporterhelper.WithShutdown(exp.shutdown),
+			exporterhelper.WithTimeout(c.TimeoutSettings),
+			exporterhelper.WithQueue(c.QueueSettings),
+			exporterhelper.WithRetry(c.BackOffConfig),
+		)
 	}
 
+	exp := newLogsExporter(set.Logger, c)
+
 	return exporterhelper.NewLogs(
 		ctx,
 		set,
@@ -61,12 +76,24 @@ func createTracesExporter(
 	c := cfg.(*Config)
 	c.collectorVersion = set.BuildInfo.Version
 
-	// TODO: branch on feature flag or config
-	exp, err := newTracesJSONExporter(set.Logger, c)
-	if err != nil {
-		return nil, fmt.Errorf("cannot configure clickhouse json traces exporter: %w", err)
+	if featureGateJSON.IsEnabled() {
+		exp := newTracesJSONExporter(set.Logger, c)
+
+		return exporterhelper.NewTraces(
+			ctx,
+			set,
+			cfg,
+			exp.pushTraceData,
+			exporterhelper.WithStart(exp.start),
+			exporterhelper.WithShutdown(exp.shutdown),
+			exporterhelper.WithTimeout(c.TimeoutSettings),
+			exporterhelper.WithQueue(c.QueueSettings),
+			exporterhelper.WithRetry(c.BackOffConfig),
+		)
 	}
 
+	exp := newTracesExporter(set.Logger, c)
+
 	return exporterhelper.NewTraces(
 		ctx,
 		set,

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/exporter/exportertest"
+	"go.opentelemetry.io/collector/featuregate"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal/metadata"
 )
@@ -35,6 +36,22 @@ func TestFactory_CreateLogs(t *testing.T) {
 	require.NoError(t, exporter.Shutdown(context.Background()))
 }
 
+func TestFactory_CreateLogsJSON(t *testing.T) {
+	factory := NewFactory()
+	cfg := withDefaultConfig(func(cfg *Config) {
+		cfg.Endpoint = defaultEndpoint
+	})
+	gatePrev := featureGateJSON.IsEnabled()
+	_ = featuregate.GlobalRegistry().Set(featureGateJSON.ID(), true)
+	params := exportertest.NewNopSettings(metadata.Type)
+	exporter, err := factory.CreateLogs(context.Background(), params, cfg)
+	_ = featuregate.GlobalRegistry().Set(featureGateJSON.ID(), gatePrev)
+	require.NoError(t, err)
+	require.NotNil(t, exporter)
+
+	require.NoError(t, exporter.Shutdown(context.Background()))
+}
+
 func TestFactory_CreateTraces(t *testing.T) {
 	factory := NewFactory()
 	cfg := withDefaultConfig(func(cfg *Config) {
@@ -48,6 +65,22 @@ func TestFactory_CreateTraces(t *testing.T) {
 	require.NoError(t, exporter.Shutdown(context.Background()))
 }
 
+func TestFactory_CreateTracesJSON(t *testing.T) {
+	factory := NewFactory()
+	cfg := withDefaultConfig(func(cfg *Config) {
+		cfg.Endpoint = defaultEndpoint
+	})
+	gatePrev := featureGateJSON.IsEnabled()
+	_ = featuregate.GlobalRegistry().Set(featureGateJSON.ID(), true)
+	params := exportertest.NewNopSettings(metadata.Type)
+	exporter, err := factory.CreateTraces(context.Background(), params, cfg)
+	_ = featuregate.GlobalRegistry().Set(featureGateJSON.ID(), gatePrev)
+	require.NoError(t, err)
+	require.NotNil(t, exporter)
+
+	require.NoError(t, exporter.Shutdown(context.Background()))
+}
+
 func TestFactory_CreateMetrics(t *testing.T) {
 	factory := NewFactory()
 	cfg := withDefaultConfig(func(cfg *Config) {

diff --git a/exporter/clickhouseexporter/internal/chjson/json.go b/exporter/clickhouseexporter/internal/chjson/json.go
@@ -6,9 +6,10 @@ package chjson
 import (
 	"encoding/base64"
 	"encoding/hex"
-	"go.opentelemetry.io/collector/pdata/pcommon"
 	"strconv"
 	"strings"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
 )
 
 // AppendSpanIDToHex writes a hex encoded byte slice of SpanID.

diff --git a/exporter/clickhouseexporter/internal/chjson/json_test.go b/exporter/clickhouseexporter/internal/chjson/json_test.go
@@ -4,11 +4,12 @@
 package chjson
 
 import (
+	"testing"
+
 	"github.com/ClickHouse/clickhouse-go/v2/lib/column"
 	"github.com/ClickHouse/clickhouse-go/v2/lib/column/orderedmap"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
-	"testing"
 )
 
 func testMap() pcommon.Map {
@@ -86,7 +87,7 @@ func TestAttributesToJSON(t *testing.T) {
 func TestAppendSpanIDToHex(t *testing.T) {
 	hexBuf := make([]byte, 0, 128)
 
-	var spanID = pcommon.SpanID{
+	spanID := pcommon.SpanID{
 		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
 	}
 
@@ -102,7 +103,7 @@ func TestAppendSpanIDToHex(t *testing.T) {
 func TestAppendTraceIDToHex(t *testing.T) {
 	hexBuf := make([]byte, 0, 128)
 
-	var traceID = pcommon.TraceID{
+	traceID := pcommon.TraceID{
 		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
 		0x88, 0x99, 0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF,
 	}
@@ -119,7 +120,7 @@ func TestAppendTraceIDToHex(t *testing.T) {
 func BenchmarkAppendSpanIDToHex(b *testing.B) {
 	hexBuf := make([]byte, 0, 128)
 
-	var spanID = pcommon.SpanID{
+	spanID := pcommon.SpanID{
 		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
 	}
 
@@ -133,7 +134,7 @@ func BenchmarkAppendSpanIDToHex(b *testing.B) {
 func BenchmarkAppendTraceIDToHex(b *testing.B) {
 	hexBuf := make([]byte, 0, 128)
 
-	var traceID = pcommon.TraceID{
+	traceID := pcommon.TraceID{
 		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
 		0x88, 0x99, 0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF,
 	}