[processor/elasticapm] Update processor to handle new supported fields (#867)

* Add missing supported otel sem conv attributes to ecs translation

* Add missing supported APM/ECS sem conv attributes to ecs translation

* Updates `ecs/elastic_span_db` test cases to include all supported OTEL semconv attributes

* Update `ecs_translation.go` to use constants for ecs resource attributes and update test case

Author: isaacaflores2

processor/elasticapmprocessor/internal/ecs/ecs_translation.go

@@ -25,6 +25,48 @@ import (
 	semconv "go.opentelemetry.io/otel/semconv/v1.27.0"
 )
 
+// Supported ECS resource attributes
+const (
+	ecsAttrServiceLanguageName       = "service.language.name"
+	ecsAttrServiceLanguageVersion    = "service.language.version"
+	ecsAttrServiceFrameworkName      = "service.framework.name"
+	ecsAttrServiceFrameworkVersion   = "service.framework.version"
+	ecsAttrServiceRuntimeName        = "service.runtime.name"
+	ecsAttrServiceRuntimeVersion     = "service.runtime.version"
+	ecsAttrServiceOriginID           = "service.origin.id"
+	ecsAttrServiceOriginName         = "service.origin.name"
+	ecsAttrServiceOriginVersion      = "service.origin.version"
+	ecsAttrServiceTargetName         = "service.target.name"
+	ecsAttrServiceTargetType         = "service.target.type"
+	ecsAttrCloudOriginAccountID      = "cloud.origin.account.id"
+	ecsAttrCloudOriginProvider       = "cloud.origin.provider"
+	ecsAttrCloudOriginRegion         = "cloud.origin.region"
+	ecsAttrCloudOriginServiceName    = "cloud.origin.service.name"
+	ecsAttrCloudAccountName          = "cloud.account.name"
+	ecsAttrCloudInstanceID           = "cloud.instance.id"
+	ecsAttrCloudInstanceName         = "cloud.instance.name"
+	ecsAttrCloudMachineType          = "cloud.machine.type"
+	ecsAttrCloudProjectID            = "cloud.project.id"
+	ecsAttrCloudProjectName          = "cloud.project.name"
+	ecsAttrContainerImageTag         = "container.image.tag"
+	ecsAttrHostOSPlatform            = "host.os.platform"
+	ecsAttrProcessRuntimeName        = "process.runtime.name"
+	ecsAttrProcessRuntimeVersion     = "process.runtime.version"
+	ecsAttrDeviceManufacturer        = "device.manufacturer"
+	ecsAttrDataStreamDataset         = "data_stream.dataset"
+	ecsAttrDataStreamNamespace       = "data_stream.namespace"
+	ecsAttrUserDomain                = "user.domain"
+	ecsAttrSourceNATIP               = "source.nat.ip"
+	ecsAttrDestinationIP             = "destination.ip"
+	ecsAttrFaaSTriggerRequestID      = "faas.trigger.request.id"
+	ecsAttrFaaSExecution             = "faas.execution"
+	ecsAttrOpenCensusExporterVersion = "opencensus.exporterversion"
+	ecsAttrAgentName                 = "agent.name"
+	ecsAttrAgentVersion              = "agent.version"
+	ecsAttrAgentEphemeralID          = "agent.ephemeral_id"
+	ecsAttrAgentActivationMethod     = "agent.activation_method"
+)
+
 func TranslateResourceMetadata(resource pcommon.Resource) {
 	attributes := resource.Attributes()
 
@@ -51,18 +93,30 @@ func replaceDots(key string) string {
 	return strings.ReplaceAll(key, ".", "_")
 }
 
-// This is based to what found in the apm-data repo
-// (e.g https://github.com/elastic/apm-data/blob/main/input/otlp/metadata.go)
-// plus other extra fields
+// isSupportedAttribute returns true if the resource attribute is
+// supported by ECS and can be mapped directly.
+// Supported fields can include OTEL SemConv attributes or ECS specific attributes.
+// Fields are based on those found in the below areas:
+// 1. apm-data: https://github.com/elastic/apm-data/blob/main/input/otlp/metadata.go
+// 2. elasticapmintake receiver: https://github.com/elastic/opentelemetry-collector-components/tree/main/receiver/elasticapmintakereceiver/internal/mappers
 func isSupportedAttribute(attr string) bool {
 	switch attr {
 	// service.*
 	case string(semconv.ServiceNameKey),
 		string(semconv.ServiceVersionKey),
 		string(semconv.ServiceInstanceIDKey),
 		string(semconv.ServiceNamespaceKey),
-		"service.language.name",
-		"service.language.version":
+		ecsAttrServiceLanguageName,
+		ecsAttrServiceLanguageVersion,
+		ecsAttrServiceFrameworkName,
+		ecsAttrServiceFrameworkVersion,
+		ecsAttrServiceRuntimeName,
+		ecsAttrServiceRuntimeVersion,
+		ecsAttrServiceOriginID,
+		ecsAttrServiceOriginName,
+		ecsAttrServiceOriginVersion,
+		ecsAttrServiceTargetName,
+		ecsAttrServiceTargetType:
 		return true
 
 	// deployment.*
@@ -80,15 +134,26 @@ func isSupportedAttribute(attr string) bool {
 		string(semconv.CloudAccountIDKey),
 		string(semconv.CloudRegionKey),
 		string(semconv.CloudAvailabilityZoneKey),
-		string(semconv.CloudPlatformKey):
+		string(semconv.CloudPlatformKey),
+		ecsAttrCloudOriginAccountID,
+		ecsAttrCloudOriginProvider,
+		ecsAttrCloudOriginRegion,
+		ecsAttrCloudOriginServiceName,
+		ecsAttrCloudAccountName,
+		ecsAttrCloudInstanceID,
+		ecsAttrCloudInstanceName,
+		ecsAttrCloudMachineType,
+		ecsAttrCloudProjectID,
+		ecsAttrCloudProjectName:
 		return true
 
 	// container.*
 	case string(semconv.ContainerNameKey),
 		string(semconv.ContainerIDKey),
 		string(semconv.ContainerImageNameKey),
-		"container.image.tag",
-		"container.runtime":
+		ecsAttrContainerImageTag,
+		string(semconv.ContainerImageTagsKey),
+		string(semconv.ContainerRuntimeKey):
 		return true
 
 	// k8s.*
@@ -102,16 +167,19 @@ func isSupportedAttribute(attr string) bool {
 	case string(semconv.HostNameKey),
 		string(semconv.HostIDKey),
 		string(semconv.HostTypeKey),
-		"host.arch",
-		string(semconv.HostIPKey):
+		string(semconv.HostArchKey),
+		string(semconv.HostIPKey),
+		ecsAttrHostOSPlatform:
 		return true
 
 	// process.*
 	case string(semconv.ProcessPIDKey),
+		string(semconv.ProcessParentPIDKey),
+		string(semconv.ProcessExecutableNameKey),
 		string(semconv.ProcessCommandLineKey),
 		string(semconv.ProcessExecutablePathKey),
-		"process.runtime.name",
-		"process.runtime.version",
+		ecsAttrProcessRuntimeName,
+		ecsAttrProcessRuntimeVersion,
 		string(semconv.ProcessOwnerKey):
 		return true
 
@@ -126,21 +194,68 @@ func isSupportedAttribute(attr string) bool {
 	case string(semconv.DeviceIDKey),
 		string(semconv.DeviceModelIdentifierKey),
 		string(semconv.DeviceModelNameKey),
-		"device.manufacturer":
+		ecsAttrDeviceManufacturer:
 		return true
 
 	// data_stream.*
-	case "data_stream.dataset",
-		"data_stream.namespace":
+	case ecsAttrDataStreamDataset,
+		ecsAttrDataStreamNamespace:
+		return true
+
+	// user.*
+	case string(semconv.UserIDKey),
+		string(semconv.UserEmailKey),
+		string(semconv.UserNameKey),
+		ecsAttrUserDomain:
+		return true
+
+	// user_agent.*
+	case string(semconv.UserAgentOriginalKey):
+		return true
+
+	// network.*
+	case string(semconv.NetworkConnectionTypeKey),
+		string(semconv.NetworkConnectionSubtypeKey),
+		string(semconv.NetworkCarrierNameKey),
+		string(semconv.NetworkCarrierMccKey),
+		string(semconv.NetworkCarrierMncKey),
+		string(semconv.NetworkCarrierIccKey):
+		return true
+
+	// client.*
+	case string(semconv.ClientAddressKey),
+		string(semconv.ClientPortKey):
+		return true
+
+	// source.*
+	case string(semconv.SourceAddressKey),
+		string(semconv.SourcePortKey),
+		ecsAttrSourceNATIP:
+		return true
+
+	// destination.*
+	case ecsAttrDestinationIP:
+		return true
+
+	// faas.*
+	case string(semconv.FaaSInstanceKey),
+		string(semconv.FaaSNameKey),
+		string(semconv.FaaSVersionKey),
+		string(semconv.FaaSTriggerKey),
+		string(semconv.FaaSColdstartKey),
+		ecsAttrFaaSTriggerRequestID,
+		ecsAttrFaaSExecution:
 		return true
 
 	// Legacy OpenCensus attributes
-	case "opencensus.exporterversion":
+	case ecsAttrOpenCensusExporterVersion:
 		return true
 
 	// APM Agent enrichment
-	case "agent.name",
-		"agent.version":
+	case ecsAttrAgentName,
+		ecsAttrAgentVersion,
+		ecsAttrAgentEphemeralID,
+		ecsAttrAgentActivationMethod:
 		return true
 	}
 

processor/elasticapmprocessor/processor_test.go

@@ -19,22 +19,25 @@ package elasticapmprocessor // import "github.com/elastic/opentelemetry-collecto
 
 import (
 	"context"
+	"flag"
 	"path/filepath"
 	"testing"
 
-	"go.opentelemetry.io/collector/client"
-
-	"github.com/elastic/opentelemetry-collector-components/processor/elasticapmprocessor/internal/metadata"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/ptracetest"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/processor/processortest"
 	"go.uber.org/zap/zapcore"
 	"go.uber.org/zap/zaptest"
+
+	"github.com/elastic/opentelemetry-collector-components/processor/elasticapmprocessor/internal/metadata"
 )
 
+var update = flag.Bool("update", false, "Flag to generate/updated the expected yaml files")
+
 // TestProcessor does some basic tests to check if enrichment is happening.
 // More exhaustive test for the logic are left to the library.
 func TestProcessor(t *testing.T) {
@@ -66,11 +69,18 @@ func TestProcessor(t *testing.T) {
 			dir := filepath.Join("testdata", tc)
 			inputTraces, err := golden.ReadTraces(filepath.Join(dir, "input.yaml"))
 			require.NoError(t, err)
-			expectedTraces, err := golden.ReadTraces(filepath.Join(dir, "output.yaml"))
+
+			outputFile := filepath.Join(dir, "output.yaml")
+			expectedTraces, err := golden.ReadTraces(outputFile)
 			require.NoError(t, err)
 
 			require.NoError(t, tp.ConsumeTraces(ctx, inputTraces))
-			assert.NoError(t, ptracetest.CompareTraces(expectedTraces, next.AllTraces()[0]))
+			actual := next.AllTraces()[0]
+			if *update {
+				err := golden.WriteTraces(t, outputFile, actual)
+				assert.NoError(t, err)
+			}
+			assert.NoError(t, ptracetest.CompareTraces(expectedTraces, actual))
 		})
 	}
 }
@@ -95,9 +105,16 @@ func TestProcessorECS(t *testing.T) {
 
 	inputTraces, err := golden.ReadTraces("testdata/ecs/elastic_span_db/input.yaml")
 	require.NoError(t, err)
-	expectedTraces, err := golden.ReadTraces("testdata/ecs/elastic_span_db/output.yaml")
+
+	outputFile := "testdata/ecs/elastic_span_db/output.yaml"
+	expectedTraces, err := golden.ReadTraces(outputFile)
 	require.NoError(t, err)
 
 	require.NoError(t, tp.ConsumeTraces(ctx, inputTraces))
-	assert.NoError(t, ptracetest.CompareTraces(expectedTraces, next.AllTraces()[0]))
+	actual := next.AllTraces()[0]
+	if *update {
+		err := golden.WriteTraces(t, outputFile, actual)
+		assert.NoError(t, err)
+	}
+	assert.NoError(t, ptracetest.CompareTraces(expectedTraces, actual))
 }