From 1f133b0a1b054c55c24d046337ee60d997b9a9e7 Mon Sep 17 00:00:00 2001
From: Carson Ip <carson.ip@elastic.co>
Date: Tue, 16 Jan 2024 10:49:29 +0000
Subject: [PATCH] Use resource attribute data_stream.*

---
 input/otlp/logs_test.go |  2 ++
 input/otlp/metadata.go  | 12 ++++++++++++
 2 files changed, 14 insertions(+)

diff --git a/input/otlp/logs_test.go b/input/otlp/logs_test.go
index 645ea556..755dcc53 100644
--- a/input/otlp/logs_test.go
+++ b/input/otlp/logs_test.go
@@ -435,6 +435,8 @@ func TestConsumerConsumeOTelEventLogs(t *testing.T) {
 	resourceLogs := logs.ResourceLogs().AppendEmpty()
 	resourceAttrs := logs.ResourceLogs().At(0).Resource().Attributes()
 	resourceAttrs.PutStr(semconv.AttributeTelemetrySDKLanguage, "swift")
+	resourceAttrs.PutStr("data_stream.dataset", "dataset")
+	resourceAttrs.PutStr("data_stream.namespace", "namespace")
 	scopeLogs := resourceLogs.ScopeLogs().AppendEmpty()
 
 	record1 := newLogRecord("") // no log body
diff --git a/input/otlp/metadata.go b/input/otlp/metadata.go
index c0a57e04..4034ac3d 100644
--- a/input/otlp/metadata.go
+++ b/input/otlp/metadata.go
@@ -290,6 +290,18 @@ func translateResourceMetadata(resource pcommon.Resource, out *modelpb.APMEvent)
 		case "telemetry.sdk.elastic_export_timestamp":
 			// Do nothing.
 
+		// data_stream.*
+		case attributeDataStreamDataset:
+			if out.DataStream == nil {
+				out.DataStream = modelpb.DataStreamFromVTPool()
+			}
+			out.DataStream.Dataset = v.Str()
+		case attributeDataStreamNamespace:
+			if out.DataStream == nil {
+				out.DataStream = modelpb.DataStreamFromVTPool()
+			}
+			out.DataStream.Namespace = v.Str()
+
 		default:
 			if out.Labels == nil {
 				out.Labels = make(modelpb.Labels)