Initial update to have attributes use the ordered map so the order is maintained into clickhouse

.chloggen/clickhouse-exporter-ordered-map-for-attributes.yaml

@@ -0,0 +1,27 @@
+# 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: "Updated exporter to sort attributes and maintain order of attributes into clickhouse"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [33634]
+
+# (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:
+
+# 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, api]

exporter/clickhouseexporter/exporter_logs.go

@@ -16,6 +16,7 @@ import (
 	conventions "go.opentelemetry.io/collector/semconv/v1.27.0"
 	"go.uber.org/zap"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/traceutil"
 )
 
@@ -77,7 +78,7 @@ func (e *logsExporter) pushLogsData(ctx context.Context, ld plog.Logs) error {
 			logs := ld.ResourceLogs().At(i)
 			res := logs.Resource()
 			resURL := logs.SchemaUrl()
-			resAttr := attributesToMap(res.Attributes())
+			resAttr := internal.OtelAttributesToOrderedMap(res.Attributes())
 			if v, ok := res.Attributes().Get(conventions.AttributeServiceName); ok {
 				serviceName = v.Str()
 			}
@@ -87,7 +88,7 @@ func (e *logsExporter) pushLogsData(ctx context.Context, ld plog.Logs) error {
 				scopeURL := logs.ScopeLogs().At(j).SchemaUrl()
 				scopeName := logs.ScopeLogs().At(j).Scope().Name()
 				scopeVersion := logs.ScopeLogs().At(j).Scope().Version()
-				scopeAttr := attributesToMap(logs.ScopeLogs().At(j).Scope().Attributes())
+				scopeAttr := internal.OtelAttributesToOrderedMap(logs.ScopeLogs().At(j).Scope().Attributes())
 
 				for k := 0; k < rs.Len(); k++ {
 					r := rs.At(k)
@@ -97,7 +98,7 @@ func (e *logsExporter) pushLogsData(ctx context.Context, ld plog.Logs) error {
 						timestamp = r.ObservedTimestamp()
 					}
 
-					logAttr := attributesToMap(r.Attributes())
+					logAttr := internal.OtelAttributesToOrderedMap(r.Attributes())
 					_, err = statement.ExecContext(ctx,
 						timestamp.AsTime(),
 						traceutil.TraceIDToHexOrEmptyString(r.TraceID()),
@@ -129,7 +130,7 @@ func (e *logsExporter) pushLogsData(ctx context.Context, ld plog.Logs) error {
 	return err
 }
 
-func attributesToMap(attributes pcommon.Map) map[string]string {
+func attributesToStringMap(attributes pcommon.Map) map[string]string {
 	m := make(map[string]string, attributes.Len())
 	attributes.Range(func(k string, v pcommon.Value) bool {
 		m[k] = v.AsString()

exporter/clickhouseexporter/exporter_logs_test.go

@@ -18,6 +18,8 @@ import (
 	conventions "go.opentelemetry.io/collector/semconv/v1.27.0"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal"
 )
 
 func TestLogsExporter_New(t *testing.T) {
@@ -94,9 +96,9 @@ func TestExporter_pushLogsData(t *testing.T) {
 		initClickhouseTestServer(t, func(query string, values []driver.Value) error {
 			if strings.HasPrefix(query, "INSERT") {
 				require.Equal(t, "https://opentelemetry.io/schemas/1.4.0", values[8])
-				require.Equal(t, map[string]string{
+				require.Equal(t, internal.NewOrderedMapFromMap(map[string]string{
 					"service.name": "test-service",
-				}, values[9])
+				}), values[9])
 			}
 			return nil
 		})
@@ -109,9 +111,9 @@ func TestExporter_pushLogsData(t *testing.T) {
 				require.Equal(t, "https://opentelemetry.io/schemas/1.7.0", values[10])
 				require.Equal(t, "io.opentelemetry.contrib.clickhouse", values[11])
 				require.Equal(t, "1.0.0", values[12])
-				require.Equal(t, map[string]string{
+				require.Equal(t, internal.NewOrderedMapFromMap(map[string]string{
 					"lib": "clickhouse",
-				}, values[13])
+				}), values[13])
 			}
 			return nil
 		})

exporter/clickhouseexporter/exporter_metrics.go

@@ -63,7 +63,7 @@ func (e *metricsExporter) pushMetricsData(ctx context.Context, md pmetric.Metric
 	metricsMap := internal.NewMetricsModel(e.cfg.MetricsTableName)
 	for i := 0; i < md.ResourceMetrics().Len(); i++ {
 		metrics := md.ResourceMetrics().At(i)
-		resAttr := attributesToMap(metrics.Resource().Attributes())
+		resAttr := attributesToStringMap(metrics.Resource().Attributes())
 		for j := 0; j < metrics.ScopeMetrics().Len(); j++ {
 			rs := metrics.ScopeMetrics().At(j).Metrics()
 			scopeInstr := metrics.ScopeMetrics().At(j).Scope()

exporter/clickhouseexporter/exporter_traces.go

@@ -10,12 +10,14 @@ import (
 	"strings"
 	"time"
 
+	"github.com/ClickHouse/clickhouse-go/v2"
 	_ "github.com/ClickHouse/clickhouse-go/v2" // For register database driver.
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	conventions "go.opentelemetry.io/collector/semconv/v1.27.0"
 	"go.uber.org/zap"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/traceutil"
 )
 
@@ -74,7 +76,7 @@ func (e *tracesExporter) pushTraceData(ctx context.Context, td ptrace.Traces) er
 		for i := 0; i < td.ResourceSpans().Len(); i++ {
 			spans := td.ResourceSpans().At(i)
 			res := spans.Resource()
-			resAttr := attributesToMap(res.Attributes())
+			resAttr := internal.OtelAttributesToOrderedMap(res.Attributes())
 			var serviceName string
 			if v, ok := res.Attributes().Get(conventions.AttributeServiceName); ok {
 				serviceName = v.Str()
@@ -85,7 +87,7 @@ func (e *tracesExporter) pushTraceData(ctx context.Context, td ptrace.Traces) er
 				scopeVersion := spans.ScopeSpans().At(j).Scope().Version()
 				for k := 0; k < rs.Len(); k++ {
 					r := rs.At(k)
-					spanAttr := attributesToMap(r.Attributes())
+					spanAttr := internal.OtelAttributesToOrderedMap(r.Attributes())
 					status := r.Status()
 					eventTimes, eventNames, eventAttrs := convertEvents(r.Events())
 					linksTraceIDs, linksSpanIDs, linksTraceStates, linksAttrs := convertLinks(r.Links())
@@ -127,34 +129,34 @@ func (e *tracesExporter) pushTraceData(ctx context.Context, td ptrace.Traces) er
 	return err
 }
 
-func convertEvents(events ptrace.SpanEventSlice) ([]time.Time, []string, []map[string]string) {
+func convertEvents(events ptrace.SpanEventSlice) ([]time.Time, []string, clickhouse.ArraySet) {
 	var (
 		times []time.Time
 		names []string
-		attrs []map[string]string
+		attrs clickhouse.ArraySet
 	)
 	for i := 0; i < events.Len(); i++ {
 		event := events.At(i)
 		times = append(times, event.Timestamp().AsTime())
 		names = append(names, event.Name())
-		attrs = append(attrs, attributesToMap(event.Attributes()))
+		attrs = append(attrs, internal.OtelAttributesToOrderedMap(event.Attributes()))
 	}
 	return times, names, attrs
 }
 
-func convertLinks(links ptrace.SpanLinkSlice) ([]string, []string, []string, []map[string]string) {
+func convertLinks(links ptrace.SpanLinkSlice) ([]string, []string, []string, clickhouse.ArraySet) {
 	var (
 		traceIDs []string
 		spanIDs  []string
 		states   []string
-		attrs    []map[string]string
+		attrs    clickhouse.ArraySet
 	)
 	for i := 0; i < links.Len(); i++ {
 		link := links.At(i)
 		traceIDs = append(traceIDs, traceutil.TraceIDToHexOrEmptyString(link.TraceID()))
 		spanIDs = append(spanIDs, traceutil.SpanIDToHexOrEmptyString(link.SpanID()))
 		states = append(states, link.TraceState().AsRaw())
-		attrs = append(attrs, attributesToMap(link.Attributes()))
+		attrs = append(attrs, internal.OtelAttributesToOrderedMap(link.Attributes()))
 	}
 	return traceIDs, spanIDs, states, attrs
 }

exporter/clickhouseexporter/exporter_traces_test.go

@@ -10,11 +10,14 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	conventions "go.opentelemetry.io/collector/semconv/v1.27.0"
 	"go.uber.org/zap/zaptest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/clickhouseexporter/internal"
 )
 
 func TestExporter_pushTracesData(t *testing.T) {
@@ -46,6 +49,28 @@ func TestExporter_pushTracesData(t *testing.T) {
 		exporter := newTestTracesExporter(t, defaultEndpoint)
 		mustPushTracesData(t, exporter, simpleTraces(1))
 	})
+
+	t.Run("check OrderMap attributes", func(t *testing.T) {
+		initClickhouseTestServer(t, func(query string, values []driver.Value) error {
+			if strings.HasPrefix(query, "INSERT INTO otel_trace") {
+				require.Equal(t, internal.NewOrderedMapFromMap(map[string]string{
+					"service.name": "test-service",
+				}), values[8])
+				require.Equal(t, internal.NewOrderedMapFromMap(map[string]string{
+					"service.name": "v",
+				}), values[11])
+				require.Equal(t, clickhouse.ArraySet{internal.NewOrderedMapFromMap(map[string]string{
+					"level": "info",
+				})}, values[17])
+				require.Equal(t, clickhouse.ArraySet{internal.NewOrderedMapFromMap(map[string]string{
+					"k": "v",
+				})}, values[21])
+			}
+			return nil
+		})
+		exporter := newTestTracesExporter(t, defaultEndpoint)
+		mustPushTracesData(t, exporter, simpleTraces(1))
+	})
 }
 
 func newTestTracesExporter(t *testing.T, dsn string, fns ...func(*Config)) *tracesExporter {

exporter/clickhouseexporter/internal/exponential_histogram_metrics.go

@@ -143,14 +143,14 @@ func (e *expHistogramMetrics) insert(ctx context.Context, db *sql.DB) error {
 					model.metadata.ResURL,
 					model.metadata.ScopeInstr.Name(),
 					model.metadata.ScopeInstr.Version(),
-					attributesToMap(model.metadata.ScopeInstr.Attributes()),
+					OtelAttributesToOrderedMap(model.metadata.ScopeInstr.Attributes()),
 					model.metadata.ScopeInstr.DroppedAttributesCount(),
 					model.metadata.ScopeURL,
 					serviceName,
 					model.metricName,
 					model.metricDescription,
 					model.metricUnit,
-					attributesToMap(dp.Attributes()),
+					OtelAttributesToOrderedMap(dp.Attributes()),
 					dp.StartTimestamp().AsTime(),
 					dp.Timestamp().AsTime(),
 					dp.Count(),

exporter/clickhouseexporter/internal/gauge_metrics.go

@@ -118,18 +118,18 @@ func (g *gaugeMetrics) insert(ctx context.Context, db *sql.DB) error {
 				dp := model.gauge.DataPoints().At(i)
 				attrs, times, values, traceIDs, spanIDs := convertExemplars(dp.Exemplars())
 				_, err = statement.ExecContext(ctx,
-					model.metadata.ResAttr,
+					NewOrderedMapFromMap(model.metadata.ResAttr),
 					model.metadata.ResURL,
 					model.metadata.ScopeInstr.Name(),
 					model.metadata.ScopeInstr.Version(),
-					attributesToMap(model.metadata.ScopeInstr.Attributes()),
+					OtelAttributesToOrderedMap(model.metadata.ScopeInstr.Attributes()),
 					model.metadata.ScopeInstr.DroppedAttributesCount(),
 					model.metadata.ScopeURL,
 					serviceName,
 					model.metricName,
 					model.metricDescription,
 					model.metricUnit,
-					attributesToMap(dp.Attributes()),
+					OtelAttributesToOrderedMap(dp.Attributes()),
 					dp.StartTimestamp().AsTime(),
 					dp.Timestamp().AsTime(),
 					getValue(dp.IntValue(), dp.DoubleValue(), dp.ValueType()),

exporter/clickhouseexporter/internal/histogram_metrics.go

@@ -130,18 +130,18 @@ func (h *histogramMetrics) insert(ctx context.Context, db *sql.DB) error {
 				dp := model.histogram.DataPoints().At(i)
 				attrs, times, values, traceIDs, spanIDs := convertExemplars(dp.Exemplars())
 				_, err = statement.ExecContext(ctx,
-					model.metadata.ResAttr,
+					NewOrderedMapFromMap(model.metadata.ResAttr),
 					model.metadata.ResURL,
 					model.metadata.ScopeInstr.Name(),
 					model.metadata.ScopeInstr.Version(),
-					attributesToMap(model.metadata.ScopeInstr.Attributes()),
+					OtelAttributesToOrderedMap(model.metadata.ScopeInstr.Attributes()),
 					model.metadata.ScopeInstr.DroppedAttributesCount(),
 					model.metadata.ScopeURL,
 					serviceName,
 					model.metricName,
 					model.metricDescription,
 					model.metricUnit,
-					attributesToMap(dp.Attributes()),
+					OtelAttributesToOrderedMap(dp.Attributes()),
 					dp.StartTimestamp().AsTime(),
 					dp.Timestamp().AsTime(),
 					dp.Count(),

exporter/clickhouseexporter/internal/metrics_model.go

@@ -13,6 +13,7 @@ import (
 	"sync"
 
 	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/ClickHouse/clickhouse-go/v2/lib/column"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.uber.org/zap"
@@ -104,23 +105,35 @@ func InsertMetrics(ctx context.Context, db *sql.DB, metricsMap map[pmetric.Metri
 
 func convertExemplars(exemplars pmetric.ExemplarSlice) (clickhouse.ArraySet, clickhouse.ArraySet, clickhouse.ArraySet, clickhouse.ArraySet, clickhouse.ArraySet) {
 	var (
-		attrs    clickhouse.ArraySet
-		times    clickhouse.ArraySet
-		values   clickhouse.ArraySet
-		traceIDs clickhouse.ArraySet
-		spanIDs  clickhouse.ArraySet
+		attrs         column.IterableOrderedMap
+		exemplarAttrs clickhouse.ArraySet
+		times         clickhouse.ArraySet
+		values        clickhouse.ArraySet
+		traceIDs      clickhouse.ArraySet
+		spanIDs       clickhouse.ArraySet
 	)
+
+	attrs = NewOrderedMap()
+
 	for i := 0; i < exemplars.Len(); i++ {
 		exemplar := exemplars.At(i)
-		attrs = append(attrs, attributesToMap(exemplar.FilteredAttributes()))
+		orderedMap := OtelAttributesToOrderedMap(exemplar.FilteredAttributes())
+		mapIterator := orderedMap.Iterator()
+
+		for mapIterator.Next() {
+			attrs.Put(mapIterator.Key(), mapIterator.Value())
+		}
+
+		exemplarAttrs = append(exemplarAttrs, attrs)
+
 		times = append(times, exemplar.Timestamp().AsTime())
 		values = append(values, getValue(exemplar.IntValue(), exemplar.DoubleValue(), exemplar.ValueType()))
 
 		traceID, spanID := exemplar.TraceID(), exemplar.SpanID()
 		traceIDs = append(traceIDs, hex.EncodeToString(traceID[:]))
 		spanIDs = append(spanIDs, hex.EncodeToString(spanID[:]))
 	}
-	return attrs, times, values, traceIDs, spanIDs
+	return exemplarAttrs, times, values, traceIDs, spanIDs
 }
 
 // https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/metrics/v1/metrics.proto#L358
@@ -159,13 +172,17 @@ func getValue(intValue int64, floatValue float64, dataType any) float64 {
 	}
 }
 
-func attributesToMap(attributes pcommon.Map) map[string]string {
-	m := make(map[string]string, attributes.Len())
+// OtelAttributesToOrderedMap converts Otel attributes to OrderedMap
+// This will Sort the attributes by key
+func OtelAttributesToOrderedMap(attributes pcommon.Map) column.IterableOrderedMap {
+
+	om := NewOrderedMap()
 	attributes.Range(func(k string, v pcommon.Value) bool {
-		m[k] = v.AsString()
+		om.Put(k, v.AsString())
 		return true
 	})
-	return m
+	om.Sort()
+	return om
 }
 
 func convertSliceToArraySet[T any](slice []T) clickhouse.ArraySet {