feat(otelcol): allow event extraction from spans in spanlogs (#2427)

CHANGELOG.md

@@ -54,6 +54,8 @@ v1.6.0-rc.1
 
 - Add a new `/-/healthy` endpoint which returns HTTP 500 if one or more components are unhealthy. (@ptodev)
 
+- Add `otelcol.connector.spanlogs` allowing to export span events as logs. (@steve-hb)
+
 ### Enhancements
 
 - Update `prometheus.write.queue` to support v2 for cpu performance. (@mattdurham)

docs/sources/reference/components/otelcol/otelcol.connector.spanlogs.md

@@ -40,8 +40,10 @@ otelcol.connector.spanlogs "LABEL" {
 | `spans`              | `bool`         | Log one line per span.                        | `false` | no       |
 | `roots`              | `bool`         | Log one line for every root span of a trace.  | `false` | no       |
 | `processes`          | `bool`         | Log one line for every process.               | `false` | no       |
+| `events`             | `bool`         | Log one line for every span event.            | `false` | no       |
 | `span_attributes`    | `list(string)` | Additional span attributes to log.            | `[]`    | no       |
 | `process_attributes` | `list(string)` | Additional process attributes to log.         | `[]`    | no       |
+| `event_attributes`   | `list(string)` | Additional event attributes to log.           | `[]`    | no       |
 | `labels`             | `list(string)` | A list of keys that will be logged as labels. | `[]`    | no       |
 
 The values listed in `labels` should be the values of either span or process attributes.

internal/component/otelcol/connector/spanlogs/consumer.go

@@ -19,6 +19,7 @@ const (
 	typeSpan    = "span"
 	typeRoot    = "root"
 	typeProcess = "process"
+	typeEvent   = "event"
 )
 
 type consumer struct {
@@ -30,8 +31,10 @@ type options struct {
 	spans             bool
 	roots             bool
 	processes         bool
+	events            bool
 	spanAttributes    []string
 	processAttributes []string
+	eventAttributes   []string
 	overrides         OverrideConfig
 	labels            map[string]struct{}
 	nextConsumer      otelconsumer.Logs
@@ -67,8 +70,10 @@ func (c *consumer) UpdateOptions(args Arguments, nextConsumer otelconsumer.Logs)
 		spans:             args.Spans,
 		roots:             args.Roots,
 		processes:         args.Processes,
+		events:            args.Events,
 		spanAttributes:    args.SpanAttributes,
 		processAttributes: args.ProcessAttributes,
+		eventAttributes:   args.EventAttributes,
 		overrides:         args.Overrides,
 		labels:            labels,
 		nextConsumer:      nextConsumer,
@@ -126,11 +131,12 @@ func (c *consumer) consumeSpans(serviceName string, ss ptrace.ScopeSpans, rs pco
 		span := ss.Spans().At(k)
 		traceID := span.TraceID().String()
 
+		logEvents := c.opts.events
 		logSpans := c.opts.spans
 		logRoots := c.opts.roots && span.ParentSpanID().IsEmpty()
 		logProcesses := c.opts.processes && lastTraceID != traceID
 
-		if !logSpans && !logRoots && !logProcesses {
+		if !logSpans && !logRoots && !logProcesses && !logEvents {
 			return nil
 		}
 
@@ -175,7 +181,36 @@ func (c *consumer) consumeSpans(serviceName string, ss ptrace.ScopeSpans, rs pco
 				return err
 			}
 		}
+
+		if logEvents {
+			err := c.consumeEvents(keyValues, span.Events(), logRecords)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (c *consumer) consumeEvents(output pcommon.Map, events ptrace.SpanEventSlice, logRecords plog.LogRecordSlice) error {
+	eventsLen := events.Len()
+	for i := 0; i < eventsLen; i++ {
+		event := events.At(i)
+
+		// Can we find a solution without relying on more memory allocation?
+		// Clone output map due to having multiple events in one span otherwise leading to continuous use
+		// of the previous set event keyVals.
+		eventOutput := pcommon.NewMap()
+		output.CopyTo(eventOutput)
+
+		c.eventKeyVals(eventOutput, event)
+
+		err := c.appendLogRecord(typeEvent, eventOutput, logRecords)
+		if err != nil {
+			return err
+		}
 	}
+
 	return nil
 }
 
@@ -242,6 +277,18 @@ func (c *consumer) createLogRecord(kind string, keyValues pcommon.Map) (*plog.Lo
 	return &res, nil
 }
 
+func (c *consumer) eventKeyVals(output pcommon.Map, event ptrace.SpanEvent) {
+	etAtts := event.Attributes()
+
+	for _, name := range c.opts.eventAttributes {
+		att, ok := etAtts.Get(name)
+		if ok {
+			val := output.PutEmpty(name)
+			att.CopyTo(val)
+		}
+	}
+}
+
 func (c *consumer) processKeyVals(output pcommon.Map, resource pcommon.Resource, svc string) {
 	rsAtts := resource.Attributes()
 

internal/component/otelcol/connector/spanlogs/spanlogs.go

@@ -32,8 +32,10 @@ type Arguments struct {
 	Spans             bool           `alloy:"spans,attr,optional"`
 	Roots             bool           `alloy:"roots,attr,optional"`
 	Processes         bool           `alloy:"processes,attr,optional"`
+	Events            bool           `alloy:"events,attr,optional"`
 	SpanAttributes    []string       `alloy:"span_attributes,attr,optional"`
 	ProcessAttributes []string       `alloy:"process_attributes,attr,optional"`
+	EventAttributes   []string       `alloy:"event_attributes,attr,optional"`
 	Overrides         OverrideConfig `alloy:"overrides,block,optional"`
 	Labels            []string       `alloy:"labels,attr,optional"`
 

internal/component/otelcol/connector/spanlogs/spanlogs_test.go

@@ -86,12 +86,33 @@ func Test_ComponentIO(t *testing.T) {
 					{
 						"key": "account_id",
 						"value": { "intValue": "2245" }
+					}],
+					"events": [{
+						"name": "log",
+						"attributes": [{
+							"key": "log.severity",
+							"value": { "stringValue": "INFO" }
+						},
+						{
+							"key": "log.message",
+							"value": { "stringValue": "TestLogMessage" }
+						}]
+					},
+					{
+						"name": "test_event",
+						"attributes": [{
+							"key": "cause",
+							"value": { "stringValue": "call" }
+						},
+						{
+							"key": "ignore",
+							"value": { "stringValue": "ignore" }
+						}]
 					}]
 				}]
 			}]
 		}]
 	}`
-
 	defaultOverrides := spanlogs.OverrideConfig{
 		LogsTag:     "traces",
 		ServiceKey:  "svc",
@@ -695,6 +716,79 @@ func Test_ComponentIO(t *testing.T) {
 				}]
 			}`,
 		},
+		{
+			testName: "Events",
+			cfg: `
+			events = true
+			span_attributes = ["attribute1", "redact_trace", "account_id"]
+			event_attributes = ["log.severity", "log.message"]
+			labels = ["attribute1", "redact_trace", "account_id", "log.severity", "log.message"]
+
+			output {
+				// no-op: will be overridden by test code.
+			}`,
+			expectedUnmarshaledCfg: spanlogs.Arguments{
+				Events:          true,
+				EventAttributes: []string{"log.severity", "log.message"},
+				SpanAttributes:  []string{"attribute1", "redact_trace", "account_id"},
+				Overrides:       defaultOverrides,
+				Labels:          []string{"attribute1", "redact_trace", "account_id", "log.severity", "log.message"},
+				Output:          &otelcol.ConsumerArguments{},
+			},
+			inputTraceJson: defaultInputTrace,
+			expectedOutputLogJson: `{
+				"resourceLogs": [{
+					"scopeLogs": [{
+						"log_records": [{
+							"body": { "stringValue": "span=TestSpan dur=0ns attribute1=78 redact_trace=true account_id=2245 svc=TestSvcName tid=7bba9f33312b3dbb8b2c2c62bb7abe2d log.severity=INFO log.message=TestLogMessage" },
+							"attributes": [{
+								"key": "traces",
+								"value": { "stringValue": "event" }
+							},
+							{
+								"key": "attribute1",
+								"value": { "intValue": "78" }
+							},
+							{
+								"key": "redact_trace", 
+								"value": { "boolValue": true }
+							},
+							{
+								"key": "account_id",
+								"value": { "intValue": "2245" }
+							},
+							{
+								"key": "log.severity",
+								"value": { "stringValue": "INFO" }
+							},
+							{
+								"key": "log.message",
+								"value": { "stringValue": "TestLogMessage" }
+							}]
+						},
+						{
+							"body": { "stringValue": "span=TestSpan dur=0ns attribute1=78 redact_trace=true account_id=2245 svc=TestSvcName tid=7bba9f33312b3dbb8b2c2c62bb7abe2d" },
+							"attributes": [{
+								"key": "traces",
+								"value": { "stringValue": "event" }
+							},
+							{
+								"key": "attribute1",
+								"value": { "intValue": "78" }
+							},
+							{
+								"key": "redact_trace", 
+								"value": { "boolValue": true }
+							},
+							{
+								"key": "account_id",
+								"value": { "intValue": "2245" }
+							}]
+						}]
+					}]
+				}]
+			}`,
+		},
 	}
 
 	for _, tt := range tests {