[connector/exceptions] Add support for exemplars in exceptionsconnect…

…or (#31819)

**Description:** <Describe what has changed.>
Adds support for exemplars fort the generated metrics from exceptions.
It relates the span and trace id with metrics.

The ingestion needs to be enabled via config (disbled by default).

**Link to tracking Issue:**  Resolves #24409

**Documentation:** Added documentation for enabling generation of
exemplars.

---------

Co-authored-by: Juraci Paixão Kröhling <juraci@kroehling.de>

.chloggen/exceptionsconnector_add_exemplars.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: exceptionsconnector
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for exemplars in exceptionsconnector
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [24409]
+
+# (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: []

connector/exceptionsconnector/README.md

@@ -51,6 +51,9 @@ The following settings can be optionally configured:
 
   The provided default config includes `exception.type` and `exception.message` as additional dimensions.
 
+- `exemplars`:  Use to configure how to attach exemplars to metrics.
+  - `enabled` (default: `false`): enabling will add spans as Exemplars.
+
 ## Examples
 
 The following is a simple example usage of the `exceptions` connector.

connector/exceptionsconnector/config.go

@@ -15,6 +15,10 @@ type Dimension struct {
 	Default *string `mapstructure:"default"`
 }
 
+type Exemplars struct {
+	Enabled bool `mapstructure:"enabled"`
+}
+
 // Config defines the configuration options for exceptionsconnector
 type Config struct {
 	// Dimensions defines the list of additional dimensions on top of the provided:
@@ -25,6 +29,8 @@ type Config struct {
 	// The dimensions will be fetched from the span's attributes. Examples of some conventionally used attributes:
 	// https://github.com/open-telemetry/opentelemetry-collector/blob/main/model/semconv/opentelemetry.go.
 	Dimensions []Dimension `mapstructure:"dimensions"`
+	// Exemplars defines the configuration for exemplars.
+	Exemplars Exemplars `mapstructure:"exemplars"`
 }
 
 var _ component.ConfigValidator = (*Config)(nil)

connector/exceptionsconnector/config_test.go

@@ -36,6 +36,9 @@ func TestLoadConfig(t *testing.T) {
 					{Name: exceptionTypeKey},
 					{Name: exceptionMessageKey},
 				},
+				Exemplars: Exemplars{
+					Enabled: false,
+				},
 			},
 		},
 	}

connector/exceptionsconnector/connector_metrics.go

@@ -37,17 +37,18 @@ type metricsConnector struct {
 	component.StartFunc
 	component.ShutdownFunc
 
-	exceptions map[string]*excVal
+	exceptions map[string]*exception
 
 	logger *zap.Logger
 
 	// The starting time of the data points.
 	startTimestamp pcommon.Timestamp
 }
 
-type excVal struct {
-	count int
-	attrs pcommon.Map
+type exception struct {
+	count     int
+	attrs     pcommon.Map
+	exemplars pmetric.ExemplarSlice
 }
 
 func newMetricsConnector(logger *zap.Logger, config component.Config) *metricsConnector {
@@ -59,7 +60,7 @@ func newMetricsConnector(logger *zap.Logger, config component.Config) *metricsCo
 		dimensions:     newDimensions(cfg.Dimensions),
 		keyBuf:         bytes.NewBuffer(make([]byte, 0, 1024)),
 		startTimestamp: pcommon.NewTimestampFromTime(time.Now()),
-		exceptions:     make(map[string]*excVal),
+		exceptions:     make(map[string]*exception),
 	}
 }
 
@@ -95,7 +96,8 @@ func (c *metricsConnector) ConsumeTraces(ctx context.Context, traces ptrace.Trac
 						key := c.keyBuf.String()
 
 						attrs := buildDimensionKVs(c.dimensions, serviceName, span, eventAttrs)
-						c.addException(key, attrs)
+						exc := c.addException(key, attrs)
+						c.addExemplar(exc, span.TraceID(), span.SpanID())
 					}
 				}
 			}
@@ -132,28 +134,45 @@ func (c *metricsConnector) collectExceptions(ilm pmetric.ScopeMetrics) error {
 	dps := mCalls.Sum().DataPoints()
 	dps.EnsureCapacity(len(c.exceptions))
 	timestamp := pcommon.NewTimestampFromTime(time.Now())
-	for _, val := range c.exceptions {
-		dpCalls := dps.AppendEmpty()
-		dpCalls.SetStartTimestamp(c.startTimestamp)
-		dpCalls.SetTimestamp(timestamp)
-
-		dpCalls.SetIntValue(int64(val.count))
-
-		val.attrs.CopyTo(dpCalls.Attributes())
+	for _, exc := range c.exceptions {
+		dp := dps.AppendEmpty()
+		dp.SetStartTimestamp(c.startTimestamp)
+		dp.SetTimestamp(timestamp)
+		dp.SetIntValue(int64(exc.count))
+		for i := 0; i < exc.exemplars.Len(); i++ {
+			exc.exemplars.At(i).SetTimestamp(timestamp)
+		}
+		dp.Exemplars().EnsureCapacity(exc.exemplars.Len())
+		exc.exemplars.CopyTo(dp.Exemplars())
+		exc.attrs.CopyTo(dp.Attributes())
+		// Reset the exemplars for the next batch of spans.
+		exc.exemplars = pmetric.NewExemplarSlice()
 	}
 	return nil
 }
 
-func (c *metricsConnector) addException(excKey string, attrs pcommon.Map) {
+func (c *metricsConnector) addException(excKey string, attrs pcommon.Map) *exception {
 	exc, ok := c.exceptions[excKey]
 	if !ok {
-		c.exceptions[excKey] = &excVal{
-			count: 1,
-			attrs: attrs,
+		c.exceptions[excKey] = &exception{
+			count:     1,
+			attrs:     attrs,
+			exemplars: pmetric.NewExemplarSlice(),
 		}
-		return
+		return c.exceptions[excKey]
 	}
 	exc.count++
+	return exc
+}
+
+func (c *metricsConnector) addExemplar(exc *exception, traceID pcommon.TraceID, spanID pcommon.SpanID) {
+	if !c.config.Exemplars.Enabled || traceID.IsEmpty() {
+		return
+	}
+	e := exc.exemplars.AppendEmpty()
+	e.SetTraceID(traceID)
+	e.SetSpanID(spanID)
+	e.SetDoubleValue(float64(exc.count))
 }
 
 func buildDimensionKVs(dimensions []dimension, serviceName string, span ptrace.Span, eventAttrs pcommon.Map) pcommon.Map {

connector/exceptionsconnector/connector_metrics_test.go

@@ -88,6 +88,25 @@ func TestConnectorConsumeTraces(t *testing.T) {
 			}
 		})
 	}
+	t.Run("Test without exemplars", func(t *testing.T) {
+		msink := &consumertest.MetricsSink{}
+
+		p := newTestMetricsConnector(msink, stringp("defaultNullValue"), zaptest.NewLogger(t))
+		p.config.Exemplars.Enabled = false
+
+		ctx := metadata.NewIncomingContext(context.Background(), nil)
+		err := p.Start(ctx, componenttest.NewNopHost())
+		defer func() { sdErr := p.Shutdown(ctx); require.NoError(t, sdErr) }()
+		require.NoError(t, err)
+
+		err = p.ConsumeTraces(ctx, buildBadSampleTrace())
+		assert.NoError(t, err)
+
+		metrics := msink.AllMetrics()
+		assert.Greater(t, len(metrics), 0)
+		verifyBadMetricsOkay(t, metrics[len(metrics)-1])
+	})
+
 }
 
 func BenchmarkConnectorConsumeTraces(b *testing.B) {
@@ -123,6 +142,9 @@ func newTestMetricsConnector(mcon consumer.Metrics, defaultNullValue *string, lo
 			{exceptionTypeKey, nil},
 			{exceptionMessageKey, nil},
 		},
+		Exemplars: Exemplars{
+			Enabled: true,
+		},
 	}
 	c := newMetricsConnector(logger, cfg)
 	c.metricsConsumer = mcon
@@ -175,6 +197,13 @@ func verifyConsumeMetricsInput(t testing.TB, input pmetric.Metrics, numCumulativ
 		assert.NotZero(t, dp.StartTimestamp(), "StartTimestamp should be set")
 		assert.NotZero(t, dp.Timestamp(), "Timestamp should be set")
 		verifyMetricLabels(dp, t, seenMetricIDs)
+
+		assert.Equal(t, 1, dp.Exemplars().Len())
+		exemplar := dp.Exemplars().At(0)
+		assert.NotZero(t, exemplar.Timestamp())
+		assert.NotZero(t, exemplar.TraceID())
+		assert.NotZero(t, exemplar.SpanID())
+
 	}
 	return true
 }