[enrichments] Add processor.event for span events (#85)

enrichments/trace/config/config.go

@@ -23,6 +23,7 @@ type Config struct {
 	Scope       ScopeConfig              `mapstructure:"scope"`
 	Transaction ElasticTransactionConfig `mapstructure:"elastic_transaction"`
 	Span        ElasticSpanConfig        `mapstructure:"elastic_span"`
+	SpanEvent   SpanEventConfig          `mapstructure:"span_event"`
 }
 
 // ResourceConfig configures the enrichment of resource attributes.
@@ -40,6 +41,9 @@ type ScopeConfig struct {
 // ElasticTransactionConfig configures the enrichment attributes for the
 // spans which are identified as elastic transaction.
 type ElasticTransactionConfig struct {
+	// TimestampUs is a temporary attribute to enable higher
+	// resolution timestamps in Elasticsearch. For more details see:
+	// https://github.com/elastic/opentelemetry-dev/issues/374.
 	TimestampUs         AttributeConfig `mapstructure:"timestamp_us"`
 	Sampled             AttributeConfig `mapstructure:"sampled"`
 	ID                  AttributeConfig `mapstructure:"id"`
@@ -56,6 +60,9 @@ type ElasticTransactionConfig struct {
 // ElasticSpanConfig configures the enrichment attributes for the spans
 // which are NOT identified as elastic transaction.
 type ElasticSpanConfig struct {
+	// TimestampUs is a temporary attribute to enable higher
+	// resolution timestamps in Elasticsearch. For more details see:
+	// https://github.com/elastic/opentelemetry-dev/issues/374.
 	TimestampUs         AttributeConfig `mapstructure:"timestamp_us"`
 	Name                AttributeConfig `mapstructure:"name"`
 	ProcessorEvent      AttributeConfig `mapstructure:"processor_event"`
@@ -67,6 +74,15 @@ type ElasticSpanConfig struct {
 	DestinationService  AttributeConfig `mapstructure:"destination_service"`
 }
 
+// SpanEventConfig configures enrichment attributes for the span events.
+type SpanEventConfig struct {
+	// TimestampUs is a temporary attribute to enable higher
+	// resolution timestamps in Elasticsearch. For more details see:
+	// https://github.com/elastic/opentelemetry-dev/issues/374.
+	TimestampUs    AttributeConfig `mapstructure:"timestamp_us"`
+	ProcessorEvent AttributeConfig `mapstructure:"processor_event"`
+}
+
 // AttributeConfig is the configuration options for each attribute.
 type AttributeConfig struct {
 	Enabled bool `mapstructure:"enabled"`
@@ -107,5 +123,9 @@ func Enabled() Config {
 			DestinationService:  AttributeConfig{Enabled: true},
 			RepresentativeCount: AttributeConfig{Enabled: true},
 		},
+		SpanEvent: SpanEventConfig{
+			TimestampUs:    AttributeConfig{Enabled: true},
+			ProcessorEvent: AttributeConfig{Enabled: true},
+		},
 	}
 }

enrichments/trace/config/config_test.go

@@ -30,6 +30,7 @@ func TestEnabled(t *testing.T) {
 	assertAllEnabled(t, reflect.ValueOf(config.Scope))
 	assertAllEnabled(t, reflect.ValueOf(config.Transaction))
 	assertAllEnabled(t, reflect.ValueOf(config.Span))
+	assertAllEnabled(t, reflect.ValueOf(config.SpanEvent))
 }
 
 func assertAllEnabled(t *testing.T, cfg reflect.Value) {

enrichments/trace/internal/elastic/span.go

@@ -157,14 +157,20 @@ func (s *spanEnrichmentContext) Enrich(span ptrace.Span, cfg config.Config) {
 	} else {
 		s.enrichSpan(span, cfg.Span)
 	}
+
+	spanEvents := span.Events()
+	for i := 0; i < spanEvents.Len(); i++ {
+		var c spanEventEnrichmentContext
+		c.enrich(spanEvents.At(i), cfg.SpanEvent)
+	}
 }
 
 func (s *spanEnrichmentContext) enrichTransaction(
 	span ptrace.Span,
 	cfg config.ElasticTransactionConfig,
 ) {
 	if cfg.TimestampUs.Enabled {
-		s.setTimestampUs(span)
+		span.Attributes().PutInt(AttributeTimestampUs, getTimestampUs(span.StartTimestamp()))
 	}
 	if cfg.Sampled.Enabled {
 		span.Attributes().PutBool(AttributeTransactionSampled, true)
@@ -204,7 +210,7 @@ func (s *spanEnrichmentContext) enrichSpan(
 	cfg config.ElasticSpanConfig,
 ) {
 	if cfg.TimestampUs.Enabled {
-		s.setTimestampUs(span)
+		span.Attributes().PutInt(AttributeTimestampUs, getTimestampUs(span.StartTimestamp()))
 	}
 	if cfg.Name.Enabled {
 		span.Attributes().PutStr(AttributeSpanName, span.Name())
@@ -241,26 +247,6 @@ func (s *spanEnrichmentContext) normalizeAttributes() {
 	}
 }
 
-// setTimestampUs sets the attribute timestamp.us for span and
-// span events. This is a temporary function to enable higher
-// resolution timestamps in Elasticsearch. For more details see:
-// https://github.com/elastic/opentelemetry-dev/issues/374.
-//
-// TODO (lahsivjar): If more enrichments need to be added
-// for span events then consider having a separate flow for span
-// events enrichment with its own configuration.
-func (s *spanEnrichmentContext) setTimestampUs(span ptrace.Span) {
-	startTsUs := int64(span.StartTimestamp()) / 1000
-	span.Attributes().PutInt(AttributeTimestampUs, startTsUs)
-
-	events := span.Events()
-	for i := 0; i < events.Len(); i++ {
-		event := events.At(i)
-		eventTsUs := int64(event.Timestamp()) / 1000
-		event.Attributes().PutInt(AttributeTimestampUs, eventTsUs)
-	}
-}
-
 func (s *spanEnrichmentContext) setTxnType(span ptrace.Span) {
 	txnType := "unknown"
 	switch {
@@ -430,6 +416,26 @@ func (s *spanEnrichmentContext) setDestinationService(span ptrace.Span) {
 	}
 }
 
+type spanEventEnrichmentContext struct {
+	exception bool
+}
+
+func (s *spanEventEnrichmentContext) enrich(
+	se ptrace.SpanEvent,
+	cfg config.SpanEventConfig,
+) {
+	// Extract top level span event information.
+	s.exception = se.Name() == "exception"
+
+	// Enrich span event attributes.
+	if cfg.TimestampUs.Enabled {
+		se.Attributes().PutInt(AttributeTimestampUs, getTimestampUs(se.Timestamp()))
+	}
+	if cfg.ProcessorEvent.Enabled && s.exception {
+		se.Attributes().PutStr(AttributeProcessorEvent, "error")
+	}
+}
+
 // getRepresentativeCount returns the number of spans represented by an
 // individually sampled span as per the passed tracestate header.
 //
@@ -523,6 +529,10 @@ func getHostPort(urlFull *url.URL, urlDomain string, urlPort int64) string {
 	return ""
 }
 
+func getTimestampUs(ts pcommon.Timestamp) int64 {
+	return int64(ts) / 1000
+}
+
 var standardStatusCodeResults = [...]string{
 	"HTTP 1xx",
 	"HTTP 2xx",

enrichments/trace/internal/elastic/span_test.go

@@ -757,6 +757,61 @@ func TestElasticSpanEnrich(t *testing.T) {
 	}
 }
 
+func TestSpanEventEnrich(t *testing.T) {
+	now := time.Unix(3600, 0)
+	ts := pcommon.NewTimestampFromTime(now)
+	for _, tc := range []struct {
+		name          string
+		input         ptrace.SpanEvent
+		config        config.SpanEventConfig
+		enrichedAttrs map[string]any
+	}{
+		{
+			name: "not_exception",
+			input: func() ptrace.SpanEvent {
+				event := ptrace.NewSpanEvent()
+				event.SetTimestamp(ts)
+				return event
+			}(),
+			config: config.Enabled().SpanEvent,
+			enrichedAttrs: map[string]any{
+				AttributeTimestampUs: ts.AsTime().UnixMicro(),
+			},
+		},
+		{
+			name: "exception",
+			input: func() ptrace.SpanEvent {
+				event := ptrace.NewSpanEvent()
+				event.SetName("exception")
+				event.SetTimestamp(ts)
+				return event
+			}(),
+			config: config.Enabled().SpanEvent,
+			enrichedAttrs: map[string]any{
+				AttributeTimestampUs:    ts.AsTime().UnixMicro(),
+				AttributeProcessorEvent: "error",
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			// Merge existing input attrs with the attrs added
+			// by enrichment to get the expected attributes.
+			expectedAttrs := tc.input.Attributes().AsRaw()
+			for k, v := range tc.enrichedAttrs {
+				expectedAttrs[k] = v
+			}
+
+			span := ptrace.NewSpan()
+			tc.input.MoveTo(span.Events().AppendEmpty())
+			EnrichSpan(span, config.Config{
+				SpanEvent: tc.config,
+			})
+
+			assert.Empty(t, cmp.Diff(expectedAttrs, span.Events().At(0).Attributes().AsRaw()))
+		})
+	}
+}
+
 func TestIsElasticTransaction(t *testing.T) {
 	for _, tc := range []struct {
 		name  string