[exporter/elasticsearch] add missing scope info in span/log attributes

.chloggen/elasticsearchexporter_add_missing_scope_info_in_span.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: elasticsearchexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "add missing scope info in span attributes"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [27282]
+
+# (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: [user]

cmd/otelcontribcol/exporters_test.go

@@ -83,6 +83,7 @@ func TestDefaultExporters(t *testing.T) {
 				cfg := expFactories["awscloudwatchlogs"].CreateDefaultConfig().(*awscloudwatchlogsexporter.Config)
 				cfg.Endpoint = "http://" + endpoint
 				cfg.Region = "local"
+
 				// disable queue/retry to validate passing the test data synchronously
 				cfg.QueueSettings.Enabled = false
 				cfg.RetrySettings.Enabled = false

exporter/elasticsearchexporter/logs_exporter.go

@@ -83,9 +83,10 @@ func (e *elasticsearchLogsExporter) pushLogsData(ctx context.Context, ld plog.Lo
 		resource := rl.Resource()
 		ills := rl.ScopeLogs()
 		for j := 0; j < ills.Len(); j++ {
+			scope := ills.At(j).Scope()
 			logs := ills.At(j).LogRecords()
 			for k := 0; k < logs.Len(); k++ {
-				if err := e.pushLogRecord(ctx, resource, logs.At(k)); err != nil {
+				if err := e.pushLogRecord(ctx, resource, logs.At(k), scope); err != nil {
 					if cerr := ctx.Err(); cerr != nil {
 						return cerr
 					}
@@ -99,7 +100,7 @@ func (e *elasticsearchLogsExporter) pushLogsData(ctx context.Context, ld plog.Lo
 	return errors.Join(errs...)
 }
 
-func (e *elasticsearchLogsExporter) pushLogRecord(ctx context.Context, resource pcommon.Resource, record plog.LogRecord) error {
+func (e *elasticsearchLogsExporter) pushLogRecord(ctx context.Context, resource pcommon.Resource, record plog.LogRecord, scope pcommon.InstrumentationScope) error {
 	fIndex := e.index
 	if e.dynamicIndex {
 		prefix := getFromBothResourceAndAttribute(indexPrefix, resource, record)
@@ -108,7 +109,7 @@ func (e *elasticsearchLogsExporter) pushLogRecord(ctx context.Context, resource
 		fIndex = fmt.Sprintf("%s%s%s", prefix, fIndex, suffix)
 	}
 
-	document, err := e.model.encodeLog(resource, record)
+	document, err := e.model.encodeLog(resource, record, scope)
 	if err != nil {
 		return fmt.Errorf("Failed to encode log event: %w", err)
 	}

exporter/elasticsearchexporter/logs_exporter_test.go

@@ -403,9 +403,12 @@ func mustSend(t *testing.T, exporter *elasticsearchLogsExporter, contents string
 // send trace with span & resource attributes
 func mustSendLogsWithAttributes(t *testing.T, exporter *elasticsearchLogsExporter, attrMp map[string]string, resMp map[string]string) {
 	logs := newLogsWithAttributeAndResourceMap(attrMp, resMp)
-	resSpans := logs.ResourceLogs().At(0)
-	logRecords := resSpans.ScopeLogs().At(0).LogRecords().At(0)
+	resLogs := logs.ResourceLogs().At(0)
+	logRecords := resLogs.ScopeLogs().At(0).LogRecords().At(0)
 
-	err := exporter.pushLogRecord(context.TODO(), resSpans.Resource(), logRecords)
+	scopeLogs := resLogs.ScopeLogs().AppendEmpty()
+	scope := scopeLogs.Scope()
+
+	err := exporter.pushLogRecord(context.TODO(), resLogs.Resource(), logRecords, scope)
 	require.NoError(t, err)
 }

exporter/elasticsearchexporter/model.go

@@ -17,8 +17,8 @@ import (
 )
 
 type mappingModel interface {
-	encodeLog(pcommon.Resource, plog.LogRecord) ([]byte, error)
-	encodeSpan(pcommon.Resource, ptrace.Span) ([]byte, error)
+	encodeLog(pcommon.Resource, plog.LogRecord, pcommon.InstrumentationScope) ([]byte, error)
+	encodeSpan(pcommon.Resource, ptrace.Span, pcommon.InstrumentationScope) ([]byte, error)
 }
 
 // encodeModel tries to keep the event as close to the original open telemetry semantics as is.
@@ -38,7 +38,7 @@ const (
 	attributeField = "attribute"
 )
 
-func (m *encodeModel) encodeLog(resource pcommon.Resource, record plog.LogRecord) ([]byte, error) {
+func (m *encodeModel) encodeLog(resource pcommon.Resource, record plog.LogRecord, scope pcommon.InstrumentationScope) ([]byte, error) {
 	var document objmodel.Document
 	document.AddTimestamp("@timestamp", record.Timestamp()) // We use @timestamp in order to ensure that we can index if the default data stream logs template is used.
 	document.AddTraceID("TraceId", record.TraceID())
@@ -49,6 +49,7 @@ func (m *encodeModel) encodeLog(resource pcommon.Resource, record plog.LogRecord
 	document.AddAttribute("Body", record.Body())
 	document.AddAttributes("Attributes", record.Attributes())
 	document.AddAttributes("Resource", resource.Attributes())
+	document.AddAttributes("Scope", scopeToAttributes(scope))
 
 	if m.dedup {
 		document.Dedup()
@@ -61,7 +62,7 @@ func (m *encodeModel) encodeLog(resource pcommon.Resource, record plog.LogRecord
 	return buf.Bytes(), err
 }
 
-func (m *encodeModel) encodeSpan(resource pcommon.Resource, span ptrace.Span) ([]byte, error) {
+func (m *encodeModel) encodeSpan(resource pcommon.Resource, span ptrace.Span, scope pcommon.InstrumentationScope) ([]byte, error) {
 	var document objmodel.Document
 	document.AddTimestamp("@timestamp", span.StartTimestamp()) // We use @timestamp in order to ensure that we can index if the default data stream logs template is used.
 	document.AddTimestamp("EndTimestamp", span.EndTimestamp())
@@ -76,6 +77,7 @@ func (m *encodeModel) encodeSpan(resource pcommon.Resource, span ptrace.Span) ([
 	document.AddAttributes("Resource", resource.Attributes())
 	document.AddEvents("Events", span.Events())
 	document.AddInt("Duration", durationAsMicroseconds(span.StartTimestamp().AsTime(), span.EndTimestamp().AsTime())) // unit is microseconds
+	document.AddAttributes("Scope", scopeToAttributes(scope))
 
 	if m.dedup {
 		document.Dedup()
@@ -107,3 +109,13 @@ func spanLinksToString(spanLinkSlice ptrace.SpanLinkSlice) string {
 func durationAsMicroseconds(start, end time.Time) int64 {
 	return (end.UnixNano() - start.UnixNano()) / 1000
 }
+
+func scopeToAttributes(scope pcommon.InstrumentationScope) pcommon.Map {
+	attrs := pcommon.NewMap()
+	attrs.PutStr("name", scope.Name())
+	attrs.PutStr("version", scope.Version())
+	for k, v := range scope.Attributes().AsRaw() {
+		attrs.PutStr(k, v.(string))
+	}
+	return attrs
+}

exporter/elasticsearchexporter/model_test.go

@@ -13,12 +13,12 @@ import (
 	semconv "go.opentelemetry.io/collector/semconv/v1.18.0"
 )
 
-var expectedSpanBody = `{"@timestamp":"2023-04-19T03:04:05.000000006Z","Attributes.service.instance.id":"23","Duration":1000000,"EndTimestamp":"2023-04-19T03:04:06.000000006Z","Events.fooEvent.evnetMockBar":"bar","Events.fooEvent.evnetMockFoo":"foo","Events.fooEvent.time":"2023-04-19T03:04:05.000000006Z","Kind":"SPAN_KIND_CLIENT","Link":"[{\"attribute\":{},\"spanID\":\"\",\"traceID\":\"01020304050607080807060504030200\"}]","Name":"client span","Resource.cloud.platform":"aws_elastic_beanstalk","Resource.cloud.provider":"aws","Resource.deployment.environment":"BETA","Resource.service.instance.id":"23","Resource.service.name":"some-service","Resource.service.version":"env-version-1234","SpanId":"1920212223242526","TraceId":"01020304050607080807060504030201","TraceStatus":0}`
+var expectedSpanBody = `{"@timestamp":"2023-04-19T03:04:05.000000006Z","Attributes.service.instance.id":"23","Duration":1000000,"EndTimestamp":"2023-04-19T03:04:06.000000006Z","Events.fooEvent.evnetMockBar":"bar","Events.fooEvent.evnetMockFoo":"foo","Events.fooEvent.time":"2023-04-19T03:04:05.000000006Z","Kind":"SPAN_KIND_CLIENT","Link":"[{\"attribute\":{},\"spanID\":\"\",\"traceID\":\"01020304050607080807060504030200\"}]","Name":"client span","Resource.cloud.platform":"aws_elastic_beanstalk","Resource.cloud.provider":"aws","Resource.deployment.environment":"BETA","Resource.service.instance.id":"23","Resource.service.name":"some-service","Resource.service.version":"env-version-1234","Scope.lib-foo":"lib-bar","Scope.name":"io.opentelemetry.rabbitmq-2.7","Scope.version":"1.30.0-alpha","SpanId":"1920212223242526","TraceId":"01020304050607080807060504030201","TraceStatus":0}`
 
 func TestEncodeSpan(t *testing.T) {
 	model := &encodeModel{dedup: true, dedot: false}
 	td := mockResourceSpans()
-	spanByte, err := model.encodeSpan(td.ResourceSpans().At(0).Resource(), td.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0))
+	spanByte, err := model.encodeSpan(td.ResourceSpans().At(0).Resource(), td.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0), td.ResourceSpans().At(0).ScopeSpans().At(0).Scope())
 	assert.NoError(t, err)
 	assert.Equal(t, expectedSpanBody, string(spanByte))
 }
@@ -40,6 +40,10 @@ func mockResourceSpans() ptrace.Traces {
 	tEnd := time.Date(2023, 4, 19, 3, 4, 6, 6, time.UTC)
 
 	scopeSpans := resourceSpans.ScopeSpans().AppendEmpty()
+	scopeSpans.Scope().SetName("io.opentelemetry.rabbitmq-2.7")
+	scopeSpans.Scope().SetVersion("1.30.0-alpha")
+	scopeSpans.Scope().Attributes().PutStr("lib-foo", "lib-bar")
+
 	span := scopeSpans.Spans().AppendEmpty()
 	span.SetName("client span")
 	span.SetSpanID([8]byte{0x19, 0x20, 0x21, 0x22, 0x23, 0x24, 0x25, 0x26})

exporter/elasticsearchexporter/trace_exporter.go

@@ -76,9 +76,10 @@ func (e *elasticsearchTracesExporter) pushTraceData(
 		resource := il.Resource()
 		scopeSpans := il.ScopeSpans()
 		for j := 0; j < scopeSpans.Len(); j++ {
+			scope := scopeSpans.At(j).Scope()
 			spans := scopeSpans.At(j).Spans()
 			for k := 0; k < spans.Len(); k++ {
-				if err := e.pushTraceRecord(ctx, resource, spans.At(k)); err != nil {
+				if err := e.pushTraceRecord(ctx, resource, spans.At(k), scope); err != nil {
 					if cerr := ctx.Err(); cerr != nil {
 						return cerr
 					}
@@ -91,7 +92,7 @@ func (e *elasticsearchTracesExporter) pushTraceData(
 	return errors.Join(errs...)
 }
 
-func (e *elasticsearchTracesExporter) pushTraceRecord(ctx context.Context, resource pcommon.Resource, span ptrace.Span) error {
+func (e *elasticsearchTracesExporter) pushTraceRecord(ctx context.Context, resource pcommon.Resource, span ptrace.Span, scope pcommon.InstrumentationScope) error {
 	fIndex := e.index
 	if e.dynamicIndex {
 		prefix := getFromBothResourceAndAttribute(indexPrefix, resource, span)
@@ -100,7 +101,7 @@ func (e *elasticsearchTracesExporter) pushTraceRecord(ctx context.Context, resou
 		fIndex = fmt.Sprintf("%s%s%s", prefix, fIndex, suffix)
 	}
 
-	document, err := e.model.encodeSpan(resource, span)
+	document, err := e.model.encodeSpan(resource, span, scope)
 	if err != nil {
 		return fmt.Errorf("Failed to encode trace record: %w", err)
 	}

exporter/elasticsearchexporter/traces_exporter_test.go

@@ -404,7 +404,8 @@ func mustSendTracesWithAttributes(t *testing.T, exporter *elasticsearchTracesExp
 	traces := newTracesWithAttributeAndResourceMap(attrMp, resMp)
 	resSpans := traces.ResourceSpans().At(0)
 	span := resSpans.ScopeSpans().At(0).Spans().At(0)
+	scope := resSpans.ScopeSpans().At(0).Scope()
 
-	err := exporter.pushTraceRecord(context.TODO(), resSpans.Resource(), span)
+	err := exporter.pushTraceRecord(context.TODO(), resSpans.Resource(), span, scope)
 	require.NoError(t, err)
 }