[processor/elasticapm] enhance data_stream.dataset field with service.name (#925)

Add  a configuration knob in the elasticapm processor to populate
the `data_stream.dataset` filed dynamically based on the `service.name` attribute.

This is disabled by default to maintain backwards compatbility.

---------

Signed-off-by: inge4pres <[email protected]>

Author: inge4pres

processor/elasticapmprocessor/config.go

@@ -27,4 +27,9 @@ type Config struct {
 	// the x-elastic-mapping-mode metadata is set to "ecs". Traces are always enriched regardless
 	// of this setting. Defaults to false for backwards compatibility (always enrich).
 	SkipEnrichment bool `mapstructure:"skip_enrichment"`
+
+	// ServiceNameInDataStreamDataset controls whether the service.name attribute
+	// is included in the data_stream.dataset value. If true, the dataset will be
+	// in the format "apm.app.<service.name>". Defaults to false for backwards compatibility.
+	ServiceNameInDataStreamDataset bool `mapstructure:"service_name_in_datastream_dataset"`
 }

processor/elasticapmprocessor/internal/routing/data_stream.go

@@ -17,12 +17,78 @@
 
 package routing // import "github.com/elastic/opentelemetry-collector-components/processor/elasticapmprocessor/internal/routing"
 
-import "go.opentelemetry.io/collector/pdata/pcommon"
+import (
+	"strings"
 
-func EncodeDataStream(resource pcommon.Resource, dataStreamType string) {
+	"go.opentelemetry.io/collector/pdata/pcommon"
+)
+
+// DataStreamType tracks the text associated with a data stream type.
+const (
+	DataStreamTypeLogs    = "logs"
+	DataStreamTypeMetrics = "metrics"
+	DataStreamTypeTraces  = "traces"
+
+	ServiceNameAttributeKey = "service.name"
+
+	NamespaceDefault = "default" //TODO: make this configurable
+
+	ServiceNameUnknownAttributeUnknonw = "unknown"
+)
+
+func EncodeDataStream(resource pcommon.Resource, dataStreamType string, serviceNameInDataset bool) {
+	if serviceNameInDataset {
+		encodeDataStreamWithServiceName(resource, dataStreamType)
+	} else {
+		encodeDataStreamDefault(resource, dataStreamType)
+	}
+}
+
+func encodeDataStreamDefault(resource pcommon.Resource, dataStreamType string) {
 	attributes := resource.Attributes()
 
 	attributes.PutStr("data_stream.type", dataStreamType)
 	attributes.PutStr("data_stream.dataset", "apm")
-	attributes.PutStr("data_stream.namespace", "default") //TODO: make this configurable
+	attributes.PutStr("data_stream.namespace", NamespaceDefault)
+}
+
+func encodeDataStreamWithServiceName(resource pcommon.Resource, dataStreamType string) {
+	attributes := resource.Attributes()
+
+	serviceName, ok := attributes.Get(ServiceNameAttributeKey)
+	if !ok || serviceName.Str() == "" {
+		serviceName = pcommon.NewValueStr(ServiceNameUnknownAttributeUnknonw)
+	}
+
+	attributes.PutStr("data_stream.type", dataStreamType)
+	attributes.PutStr("data_stream.dataset", "apm.app."+normalizeServiceName(serviceName.Str()))
+	attributes.PutStr("data_stream.namespace", NamespaceDefault)
+}
+
+// The follwing is Copied from apm-data
+// https://github.com/elastic/apm-data/blob/46a81347bdbb81a7a308e8d2f58f39c0b1137a77/model/modelprocessor/datastream.go#L186C1-L209C2
+
+// normalizeServiceName translates serviceName into a string suitable
+// for inclusion in a data stream name.
+//
+// Concretely, this function will lowercase the string and replace any
+// reserved characters with "_".
+func normalizeServiceName(s string) string {
+	s = strings.ToLower(s)
+	s = strings.Map(replaceReservedRune, s)
+	return s
+}
+
+func replaceReservedRune(r rune) rune {
+	switch r {
+	case '\\', '/', '*', '?', '"', '<', '>', '|', ' ', ',', '#', ':':
+		// These characters are not permitted in data stream names
+		// by Elasticsearch.
+		return '_'
+	case '-':
+		// Hyphens are used to separate the data stream type, dataset,
+		// and namespace.
+		return '_'
+	}
+	return r
 }

processor/elasticapmprocessor/internal/routing/data_stream_test.go

@@ -0,0 +1,65 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package routing_test
+
+import (
+	"testing"
+
+	"github.com/elastic/opentelemetry-collector-components/processor/elasticapmprocessor/internal/routing"
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+)
+
+func TestDataStremaEncoderDefault(t *testing.T) {
+	resource := pcommon.NewResource()
+	routing.EncodeDataStream(resource, "logs", false)
+
+	attributes := resource.Attributes()
+
+	dataStreamType, ok := attributes.Get("data_stream.type")
+	assert.True(t, ok)
+	assert.Equal(t, "logs", dataStreamType.Str())
+
+	dataStreamDataset, ok := attributes.Get("data_stream.dataset")
+	assert.True(t, ok)
+	assert.Equal(t, "apm", dataStreamDataset.Str())
+
+	dataStreamNamespace, ok := attributes.Get("data_stream.namespace")
+	assert.True(t, ok)
+	assert.Equal(t, "default", dataStreamNamespace.Str())
+}
+
+func TestDataStreamEncoderWithServiceName(t *testing.T) {
+	resource := pcommon.NewResource()
+	attributes := resource.Attributes()
+	attributes.PutStr("service.name", "my-service")
+
+	routing.EncodeDataStream(resource, "metrics", true)
+
+	dataStreamType, ok := attributes.Get("data_stream.type")
+	assert.True(t, ok)
+	assert.Equal(t, "metrics", dataStreamType.Str())
+
+	dataStreamDataset, ok := attributes.Get("data_stream.dataset")
+	assert.True(t, ok)
+	assert.Equal(t, "apm.app.my_service", dataStreamDataset.Str())
+
+	dataStreamNamespace, ok := attributes.Get("data_stream.namespace")
+	assert.True(t, ok)
+	assert.Equal(t, "default", dataStreamNamespace.Str())
+}

processor/elasticapmprocessor/processor.go

@@ -65,7 +65,8 @@ func (p *TraceProcessor) ConsumeTraces(ctx context.Context, td ptrace.Traces) er
 			resourceSpan := resourceSpans.At(i)
 			resource := resourceSpan.Resource()
 			ecs.TranslateResourceMetadata(resource)
-			routing.EncodeDataStream(resource, "traces")
+			// Traces signal never need to be routed to service-specific datasets
+			routing.EncodeDataStream(resource, routing.DataStreamTypeTraces, false)
 			p.enricher.Config.Resource.DeploymentEnvironment.Enabled = false
 		}
 	}
@@ -92,18 +93,20 @@ type LogProcessor struct {
 	component.StartFunc
 	component.ShutdownFunc
 
-	next           consumer.Logs
-	enricher       *enrichments.Enricher
-	logger         *zap.Logger
-	skipEnrichment bool
+	next                   consumer.Logs
+	enricher               *enrichments.Enricher
+	logger                 *zap.Logger
+	skipEnrichment         bool
+	datasetWithServiceName bool
 }
 
 func newLogProcessor(cfg *Config, next consumer.Logs, logger *zap.Logger) *LogProcessor {
 	return &LogProcessor{
-		next:           next,
-		logger:         logger,
-		enricher:       enrichments.NewEnricher(cfg.Config),
-		skipEnrichment: cfg.SkipEnrichment,
+		next:                   next,
+		logger:                 logger,
+		enricher:               enrichments.NewEnricher(cfg.Config),
+		skipEnrichment:         cfg.SkipEnrichment,
+		datasetWithServiceName: cfg.ServiceNameInDataStreamDataset,
 	}
 }
 
@@ -115,18 +118,20 @@ type MetricProcessor struct {
 	component.StartFunc
 	component.ShutdownFunc
 
-	next           consumer.Metrics
-	enricher       *enrichments.Enricher
-	logger         *zap.Logger
-	skipEnrichment bool
+	next                   consumer.Metrics
+	enricher               *enrichments.Enricher
+	logger                 *zap.Logger
+	skipEnrichment         bool
+	datasetWithServiceName bool
 }
 
 func newMetricProcessor(cfg *Config, next consumer.Metrics, logger *zap.Logger) *MetricProcessor {
 	return &MetricProcessor{
-		next:           next,
-		logger:         logger,
-		enricher:       enrichments.NewEnricher(cfg.Config),
-		skipEnrichment: cfg.SkipEnrichment,
+		next:                   next,
+		logger:                 logger,
+		enricher:               enrichments.NewEnricher(cfg.Config),
+		skipEnrichment:         cfg.SkipEnrichment,
+		datasetWithServiceName: cfg.ServiceNameInDataStreamDataset,
 	}
 }
 
@@ -142,7 +147,7 @@ func (p *MetricProcessor) ConsumeMetrics(ctx context.Context, md pmetric.Metrics
 			resourceMetric := resourceMetrics.At(i)
 			resource := resourceMetric.Resource()
 			ecs.TranslateResourceMetadata(resource)
-			routing.EncodeDataStream(resource, "metrics")
+			routing.EncodeDataStream(resource, routing.DataStreamTypeMetrics, p.datasetWithServiceName)
 			p.enricher.Config.Resource.DeploymentEnvironment.Enabled = false
 		}
 	}
@@ -162,7 +167,7 @@ func (p *LogProcessor) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 			resourceLog := resourceLogs.At(i)
 			resource := resourceLog.Resource()
 			ecs.TranslateResourceMetadata(resource)
-			routing.EncodeDataStream(resource, "logs")
+			routing.EncodeDataStream(resource, routing.DataStreamTypeLogs, p.datasetWithServiceName)
 			p.enricher.Config.Resource.AgentVersion.Enabled = false
 			p.enricher.Config.Resource.DeploymentEnvironment.Enabled = false
 		}

processor/elasticapmprocessor/processor_test.go

@@ -254,3 +254,61 @@ func TestSkipEnrichmentMetrics(t *testing.T) {
 		})
 	}
 }
+
+func TestServiceNameInDataStreamDataset_Logs(t *testing.T) {
+	ctx := context.Background()
+	ctx = client.NewContext(ctx, client.Info{
+		Metadata: client.NewMetadata(map[string][]string{"x-elastic-mapping-mode": {"ecs"}}),
+	})
+	cfg := createDefaultConfig().(*Config)
+	cfg.ServiceNameInDataStreamDataset = true
+
+	factory := NewFactory()
+	settings := processortest.NewNopSettings(metadata.Type)
+	settings.TelemetrySettings.Logger = zaptest.NewLogger(t, zaptest.Level(zapcore.DebugLevel))
+	next := &consumertest.LogsSink{}
+
+	lp, err := factory.CreateLogs(ctx, settings, cfg, next)
+	require.NoError(t, err)
+
+	dir := filepath.Join("testdata", "elastic_apm")
+	inputLogs, err := golden.ReadLogs(filepath.Join(dir, "logs_input.yaml"))
+	require.NoError(t, err)
+
+	outputFile := filepath.Join(dir, "logs_output.yaml")
+	require.NoError(t, lp.ConsumeLogs(ctx, inputLogs))
+	actual := next.AllLogs()[0]
+
+	expectedLogs, err := golden.ReadLogs(outputFile)
+	require.NoError(t, err)
+	assert.NoError(t, plogtest.CompareLogs(expectedLogs, actual))
+}
+
+func TestServiceNameInDataStream_Metrics(t *testing.T) {
+	ctx := context.Background()
+	ctx = client.NewContext(ctx, client.Info{
+		Metadata: client.NewMetadata(map[string][]string{"x-elastic-mapping-mode": {"ecs"}}),
+	})
+	cfg := createDefaultConfig().(*Config)
+	cfg.ServiceNameInDataStreamDataset = true
+
+	factory := NewFactory()
+	settings := processortest.NewNopSettings(metadata.Type)
+	settings.TelemetrySettings.Logger = zaptest.NewLogger(t, zaptest.Level(zapcore.DebugLevel))
+	next := &consumertest.MetricsSink{}
+
+	mp, err := factory.CreateMetrics(ctx, settings, cfg, next)
+	require.NoError(t, err)
+
+	dir := filepath.Join("testdata", "elastic_apm")
+	inputMetrics, err := golden.ReadMetrics(filepath.Join(dir, "metrics_input.yaml"))
+	require.NoError(t, err)
+
+	outputFile := filepath.Join(dir, "metrics_output.yaml")
+	require.NoError(t, mp.ConsumeMetrics(ctx, inputMetrics))
+	actual := next.AllMetrics()[0]
+
+	expectedMetrics, err := golden.ReadMetrics(outputFile)
+	require.NoError(t, err)
+	assert.NoError(t, pmetrictest.CompareMetrics(expectedMetrics, actual))
+}

processor/elasticapmprocessor/testdata/elastic_apm/logs_input.yaml

@@ -0,0 +1,12 @@
+resourceLogs:
+  - resource:
+      attributes:
+        - key: service.name
+          value:
+            stringValue: test-service
+    scopeLogs:
+      - logRecords:
+          - body:
+              stringValue: test log message
+            timeUnixNano: "0"
+        scope: {}

processor/elasticapmprocessor/testdata/elastic_apm/logs_output.yaml

@@ -0,0 +1,23 @@
+resourceLogs:
+  - resource:
+      attributes:
+        - key: service.name
+          value:
+            stringValue: test-service
+        - key: data_stream.type
+          value:
+            stringValue: logs
+        - key: data_stream.dataset
+          value:
+            stringValue: apm.app.test_service
+        - key: data_stream.namespace
+          value:
+            stringValue: default
+        - key: agent.name
+          value:
+            stringValue: otlp
+    scopeLogs:
+      - logRecords:
+          - body:
+              stringValue: test log message
+        scope: {}

processor/elasticapmprocessor/testdata/elastic_apm/metrics_input.yaml

@@ -0,0 +1,14 @@
+resourceMetrics:
+  - resource:
+      attributes:
+        - key: service.name
+          value:
+            stringValue: test-service
+    scopeMetrics:
+      - metrics:
+          - name: test_metric
+            gauge:
+              dataPoints:
+                - asInt: 123
+                  timeUnixNano: "0"
+        scope: {}

processor/elasticapmprocessor/testdata/elastic_apm/metrics_output.yaml

@@ -0,0 +1,28 @@
+resourceMetrics:
+  - resource:
+      attributes:
+        - key: agent.name
+          value:
+            stringValue: otlp
+        - key: agent.version
+          value:
+            stringValue: unknown
+        - key: data_stream.dataset
+          value:
+            stringValue: apm.app.test_service
+        - key: data_stream.namespace
+          value:
+            stringValue: default
+        - key: data_stream.type
+          value:
+            stringValue: metrics
+        - key: service.name
+          value:
+            stringValue: test-service
+    scopeMetrics:
+      - metrics:
+          - gauge:
+              dataPoints:
+                - asInt: "123"
+            name: test_metric
+        scope: {}