processor/otel: clone shared labels for log events

changelogs/head.asciidoc

@@ -16,6 +16,7 @@ See <<apm-server-configuration>> for more information.
 - Fix infinite loop in tail-based sampling subscriber causing high CPU and repeated Elasticsearch searches {pull}7211[7211]
 - Do not overwrite `service version` if no transaction/error/... specific `service.name` is givven {pull}7281[7281]
 - Fix panic when processing OpenTelemetry histogram metrics without bounds {pull}7316[7316]
+- Fix mixing of labels across OpenTelemetry log records {pull}7358[7358]
 
 [float]
 ==== Intake API Changes

processor/otel/logs.go

@@ -72,6 +72,7 @@ func (c *Consumer) convertResourceLogs(resourceLogs pdata.ResourceLogs, receiveT
 	resource := resourceLogs.Resource()
 	baseEvent := model.APMEvent{Processor: model.LogProcessor}
 	translateResourceMetadata(resource, &baseEvent)
+
 	if exportTimestamp, ok := exportTimestamp(resource); ok {
 		timeDelta = receiveTimestamp.Sub(exportTimestamp)
 	}
@@ -100,6 +101,7 @@ func (c *Consumer) convertLogRecord(
 	timeDelta time.Duration,
 ) model.APMEvent {
 	event := baseEvent
+	initEventLabels(&event)
 	event.Timestamp = record.Timestamp().AsTime().Add(timeDelta)
 	event.Event.Severity = int64(record.SeverityNumber())
 	event.Event.Action = record.Name()
@@ -126,9 +128,6 @@ func (c *Consumer) convertLogRecord(
 }
 
 func setLabels(m pdata.AttributeMap, event *model.APMEvent) {
-	if event.Labels == nil || event.NumericLabels == nil {
-		initEventLabels(event)
-	}
 	m.Range(func(k string, v pdata.AttributeValue) bool {
 		setLabel(k, event, ifaceAttributeValue(v))
 		return true

processor/otel/logs_test.go

@@ -74,19 +74,19 @@ func TestConsumerConsumeLogs(t *testing.T) {
 			Severity: int64(pdata.SeverityNumberINFO),
 			Action:   "doOperation()",
 		},
-		Log:   model.Log{Level: "Info"},
-		Span:  &model.Span{ID: "0200000000000000"},
-		Trace: model.Trace{ID: "01000000000000000000000000000000"},
-		Labels: model.Labels{
-			"key": model.LabelValue{Value: "value"},
-		},
-		NumericLabels: model.NumericLabels{
-			"numeric_key": model.NumericLabelValue{Value: 1234},
-		},
+		Log:           model.Log{Level: "Info"},
+		Span:          &model.Span{ID: "0200000000000000"},
+		Trace:         model.Trace{ID: "01000000000000000000000000000000"},
+		Labels:        model.Labels{},
+		NumericLabels: model.NumericLabels{},
 	}
 	test := func(name string, body interface{}, expectedMessage string) {
 		t.Run(name, func(t *testing.T) {
-			logs := newLogs(body)
+			logs := pdata.NewLogs()
+			resourceLogs := logs.ResourceLogs().AppendEmpty()
+			logs.ResourceLogs().At(0).Resource().Attributes().InsertString(semconv.AttributeTelemetrySDKLanguage, "go")
+			instrumentationLogs := resourceLogs.InstrumentationLibraryLogs().AppendEmpty()
+			newLogRecord(body).CopyTo(instrumentationLogs.LogRecords().AppendEmpty())
 
 			var processed model.Batch
 			var processor model.ProcessBatchFunc = func(_ context.Context, batch *model.Batch) error {
@@ -113,20 +113,57 @@ func TestConsumerConsumeLogs(t *testing.T) {
 	// TODO(marclop): How to test map body
 }
 
-func newLogs(body interface{}) pdata.Logs {
+func TestConsumerConsumeLogsLabels(t *testing.T) {
 	logs := pdata.NewLogs()
 	resourceLogs := logs.ResourceLogs().AppendEmpty()
-	logs.ResourceLogs().At(0).Resource().Attributes().InsertString(semconv.AttributeTelemetrySDKLanguage, "go")
+	resourceAttrs := logs.ResourceLogs().At(0).Resource().Attributes()
+	resourceAttrs.InsertString(semconv.AttributeTelemetrySDKLanguage, "go")
+	resourceAttrs.InsertString("key0", "zero")
 	instrumentationLogs := resourceLogs.InstrumentationLibraryLogs().AppendEmpty()
-	otelLogRecord := instrumentationLogs.LogRecords().AppendEmpty()
+
+	record1 := newLogRecord("whatever")
+	record1.Attributes().InsertString("key1", "one")
+	record1.CopyTo(instrumentationLogs.LogRecords().AppendEmpty())
+
+	record2 := newLogRecord("andever")
+	record2.Attributes().InsertDouble("key2", 2)
+	record2.CopyTo(instrumentationLogs.LogRecords().AppendEmpty())
+
+	record3 := newLogRecord("amen")
+	record3.Attributes().InsertString("key3", "three")
+	record3.Attributes().InsertInt("key4", 4)
+	record3.CopyTo(instrumentationLogs.LogRecords().AppendEmpty())
+
+	var processed model.Batch
+	var processor model.ProcessBatchFunc = func(_ context.Context, batch *model.Batch) error {
+		if processed != nil {
+			panic("already processes batch")
+		}
+		processed = *batch
+		assert.NotNil(t, processed[0].Timestamp)
+		processed[0].Timestamp = time.Time{}
+		return nil
+	}
+	consumer := otel.Consumer{Processor: processor}
+	assert.NoError(t, consumer.ConsumeLogs(context.Background(), logs))
+
+	assert.Len(t, processed, 3)
+	assert.Equal(t, model.Labels{"key0": {Value: "zero"}, "key1": {Value: "one"}}, processed[0].Labels)
+	assert.Empty(t, processed[0].NumericLabels)
+	assert.Equal(t, model.Labels{"key0": {Value: "zero"}}, processed[1].Labels)
+	assert.Equal(t, model.NumericLabels{"key2": {Value: 2}}, processed[1].NumericLabels)
+	assert.Equal(t, model.Labels{"key0": {Value: "zero"}, "key3": {Value: "three"}}, processed[2].Labels)
+	assert.Equal(t, model.NumericLabels{"key4": {Value: 4}}, processed[2].NumericLabels)
+}
+
+func newLogRecord(body interface{}) pdata.LogRecord {
+	otelLogRecord := pdata.NewLogRecord()
 	otelLogRecord.SetTraceID(pdata.NewTraceID([16]byte{1}))
 	otelLogRecord.SetSpanID(pdata.NewSpanID([8]byte{2}))
 	otelLogRecord.SetName("doOperation()")
 	otelLogRecord.SetSeverityNumber(pdata.SeverityNumberINFO)
 	otelLogRecord.SetSeverityText("Info")
 	otelLogRecord.SetTimestamp(pdata.NewTimestampFromTime(time.Now()))
-	otelLogRecord.Attributes().InsertString("key", "value")
-	otelLogRecord.Attributes().InsertDouble("numeric_key", 1234)
 
 	switch b := body.(type) {
 	case string:
@@ -142,5 +179,5 @@ func newLogs(body interface{}) pdata.Logs {
 		// as a map.
 		// otelLogRecord.Body()
 	}
-	return logs
+	return otelLogRecord
 }