diff --git a/internal/elasticattr/attributes.go b/internal/elasticattr/attributes.go index ec6dea77e..7509a33c9 100644 --- a/internal/elasticattr/attributes.go +++ b/internal/elasticattr/attributes.go @@ -35,6 +35,7 @@ const ( CloudProjectName = "cloud.project.name" ContainerImageTag = "container.image.tag" DeviceManufacturer = "device.manufacturer" + DataStreamType = "data_stream.type" DataStreamDataset = "data_stream.dataset" DataStreamNamespace = "data_stream.namespace" DestinationIP = "destination.ip" diff --git a/processor/elasticapmprocessor/README.md b/processor/elasticapmprocessor/README.md index ba181901f..fe0a90fa1 100644 --- a/processor/elasticapmprocessor/README.md +++ b/processor/elasticapmprocessor/README.md @@ -24,6 +24,8 @@ The processor enriches traces, metrics, and logs with elastic specific requireme The processor configuration embeds all configuration options from [opentelemetry-lib enrichments config](https://github.com/elastic/opentelemetry-lib/tree/main/enrichments/config). All opentelemetry-lib enricher configuration fields are available and can be configured directly in the processor configuration. +The resource enrichment config also includes `resource.default_deployment_environment.enabled`. This controls the fallback `deployment.environment=unset` behavior separately from `resource.deployment_environment.enabled`, which still handles copying `deployment.environment.name` to `deployment.environment`. The fallback is only applied for ECS mapping mode. + ### Enable all enrichments ```yaml diff --git a/processor/elasticapmprocessor/go.mod b/processor/elasticapmprocessor/go.mod index f0ecb84bb..c957474c5 100644 --- a/processor/elasticapmprocessor/go.mod +++ b/processor/elasticapmprocessor/go.mod @@ -5,7 +5,7 @@ go 1.25.0 require ( github.com/cespare/xxhash/v2 v2.3.0 github.com/elastic/apm-data v1.19.5 - github.com/elastic/opentelemetry-collector-components/internal/elasticattr v0.37.0 + github.com/elastic/opentelemetry-collector-components/internal/elasticattr v0.38.0 github.com/google/go-cmp v0.7.0 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.147.0 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.147.0 diff --git a/processor/elasticapmprocessor/internal/ecs/ecs_translation.go b/processor/elasticapmprocessor/internal/ecs/ecs_translation.go index db41fd010..3a9d46fee 100644 --- a/processor/elasticapmprocessor/internal/ecs/ecs_translation.go +++ b/processor/elasticapmprocessor/internal/ecs/ecs_translation.go @@ -19,13 +19,12 @@ package ecs // import "github.com/elastic/opentelemetry-collector-components/pro import ( "strconv" - "strings" + "github.com/elastic/opentelemetry-collector-components/internal/elasticattr" + "github.com/elastic/opentelemetry-collector-components/processor/elasticapmprocessor/internal/sanitize" "go.opentelemetry.io/collector/pdata/pcommon" semconv26 "go.opentelemetry.io/otel/semconv/v1.26.0" semconv "go.opentelemetry.io/otel/semconv/v1.27.0" - - "github.com/elastic/opentelemetry-collector-components/internal/elasticattr" ) // Supported ECS resource attributes @@ -42,72 +41,21 @@ func TranslateResourceMetadata(resource pcommon.Resource) { attributes := resource.Attributes() attributes.Range(func(k string, v pcommon.Value) bool { - if isLabelAttribute(k) { - sanitized := sanitizeLabelAttributeKey(k) + if sanitize.IsLabelAttribute(k) { + sanitized := sanitize.HandleLabelAttributeKey(k) if sanitized != k { v.CopyTo(attributes.PutEmpty(sanitized)) attributes.Remove(k) } } else if !isSupportedAttribute(k) { // Other attributes that are not supported by ECS are moved to labels with a "labels." prefix. - setLabelAttributeValue(attributes, sanitizeLabelKey(k), v) + setLabelAttributeValue(attributes, sanitize.HandleAttributeKey(k), v) attributes.Remove(k) } return true }) } -// sanitizeLabelAttributeKey sanitizes the key portion of a label attribute, -// preserving the "labels." or "numeric_labels." prefix. -func sanitizeLabelAttributeKey(attr string) string { - if strings.HasPrefix(attr, "labels.") { - return "labels." + sanitizeLabelKey(strings.TrimPrefix(attr, "labels.")) - } - if strings.HasPrefix(attr, "numeric_labels.") { - return "numeric_labels." + sanitizeLabelKey(strings.TrimPrefix(attr, "numeric_labels.")) - } - return attr -} - -// sanitizeLabelKey sanitizes a label key, replacing the reserved characters -// '.', '*' and '"' with '_'. This matches the apm-server behavior. -// This matches the logic in the apm-data library here: -// https://github.com/elastic/apm-data/blob/e3e170b/model/modeljson/labels.go. -func sanitizeLabelKey(k string) string { - if strings.ContainsAny(k, ".*\"") { - return strings.Map(replaceReservedLabelKeyRune, k) - } - return k -} - -func replaceReservedLabelKeyRune(r rune) rune { - switch r { - case '.', '*', '"': - return '_' - } - return r -} - -// isLabelAttribute returns true if the resource attribute is already a prefixed label. -// The elasticapmintake receiver moves labels and numeric_labels into attributes and -// already prefixes those with "labels." and "numeric_labels." respectively and also does de-dotting. -// So for those, we don't want to double prefix - we just leave them as is. -func isLabelAttribute(attr string) bool { - return strings.HasPrefix(attr, "labels.") || strings.HasPrefix(attr, "numeric_labels.") -} - -// truncate returns s truncated at n runes, and the number of runes in the resulting string (<= n). -func truncate(s string) string { - var j int - for i := range s { - if j == keywordLength { - return s[:i] - } - j++ - } - return s -} - // setLabelAttributeValue maps a value into labels.* / numeric_labels.*. // Elasticsearch label mappings only support flat scalar values and // homogeneous arrays thereof; Map, Bytes, and empty types cannot be @@ -117,7 +65,7 @@ func truncate(s string) string { func setLabelAttributeValue(attributes pcommon.Map, key string, value pcommon.Value) { switch value.Type() { case pcommon.ValueTypeStr: - attributes.PutStr("labels."+key, truncate(value.Str())) + attributes.PutStr("labels."+key, sanitize.Truncate(value.Str(), keywordLength)) case pcommon.ValueTypeBool: attributes.PutStr("labels."+key, strconv.FormatBool(value.Bool())) case pcommon.ValueTypeInt: @@ -135,7 +83,7 @@ func setLabelAttributeValue(attributes pcommon.Map, key string, value pcommon.Va for i := 0; i < slice.Len(); i++ { item := slice.At(i) if item.Type() == pcommon.ValueTypeStr { - target.AppendEmpty().SetStr(truncate(item.Str())) + target.AppendEmpty().SetStr(sanitize.Truncate(item.Str(), keywordLength)) } } case pcommon.ValueTypeBool: diff --git a/processor/elasticapmprocessor/internal/enrichments/config/config.go b/processor/elasticapmprocessor/internal/enrichments/config/config.go index fb35496fa..d612f5b9d 100644 --- a/processor/elasticapmprocessor/internal/enrichments/config/config.go +++ b/processor/elasticapmprocessor/internal/enrichments/config/config.go @@ -30,12 +30,14 @@ type Config struct { // ResourceConfig configures the enrichment of resource attributes. type ResourceConfig struct { - AgentName AttributeConfig `mapstructure:"agent_name"` - AgentVersion AttributeConfig `mapstructure:"agent_version"` - OverrideHostName AttributeConfig `mapstructure:"override_host_name"` - DeploymentEnvironment AttributeConfig `mapstructure:"deployment_environment"` - ServiceInstanceID AttributeConfig `mapstructure:"service_instance_id"` - HostOSType AttributeConfig `mapstructure:"host_os_type"` + AgentName AttributeConfig `mapstructure:"agent_name"` + AgentVersion AttributeConfig `mapstructure:"agent_version"` + OverrideHostName AttributeConfig `mapstructure:"override_host_name"` + DeploymentEnvironment AttributeConfig `mapstructure:"deployment_environment"` + DefaultDeploymentEnvironment AttributeConfig `mapstructure:"default_deployment_environment"` + ServiceInstanceID AttributeConfig `mapstructure:"service_instance_id"` + ServiceName AttributeConfig `mapstructure:"service_name"` + HostOSType AttributeConfig `mapstructure:"host_os_type"` } // ScopeConfig configures the enrichment of scope attributes. diff --git a/processor/elasticapmprocessor/internal/enrichments/config/config_test.go b/processor/elasticapmprocessor/internal/enrichments/config/config_test.go index 6a0860fbe..c009b248e 100644 --- a/processor/elasticapmprocessor/internal/enrichments/config/config_test.go +++ b/processor/elasticapmprocessor/internal/enrichments/config/config_test.go @@ -45,9 +45,11 @@ func assertAttributeConfigDefaults(t *testing.T, cfg reflect.Value, expectDisabl // Fields that are intentionally disabled by default disabledByDefault := map[string]bool{ - "ClearSpanID": true, - "ClearSpanName": true, - "HostOSType": true, + "ClearSpanID": true, + "ClearSpanName": true, + "DefaultDeploymentEnvironment": true, + "ServiceName": true, + "HostOSType": true, } assertAttributeConfigDefaultsRecurse(t, cfg, disabled, disabledByDefault) diff --git a/processor/elasticapmprocessor/internal/enrichments/resource.go b/processor/elasticapmprocessor/internal/enrichments/resource.go index 93fee6a15..0d732feb9 100644 --- a/processor/elasticapmprocessor/internal/enrichments/resource.go +++ b/processor/elasticapmprocessor/internal/enrichments/resource.go @@ -20,6 +20,7 @@ package enrichments // import "github.com/elastic/opentelemetry-collector-compon import ( "fmt" + "github.com/elastic/opentelemetry-collector-components/processor/elasticapmprocessor/internal/sanitize" "go.opentelemetry.io/collector/pdata/pcommon" semconv25 "go.opentelemetry.io/otel/semconv/v1.25.0" semconv "go.opentelemetry.io/otel/semconv/v1.27.0" @@ -49,6 +50,7 @@ type resourceEnrichmentContext struct { deploymentEnvironmentName string serviceInstanceID string + serviceName string containerID string osType string @@ -78,6 +80,8 @@ func (s *resourceEnrichmentContext) Enrich(resource pcommon.Resource, cfg config s.deploymentEnvironmentName = v.Str() case string(semconv25.ServiceInstanceIDKey): s.serviceInstanceID = v.Str() + case string(semconv.ServiceNameKey): + s.serviceName = v.Str() case string(semconv.ContainerIDKey): s.containerID = v.Str() case string(semconv.OSTypeKey): @@ -100,13 +104,21 @@ func (s *resourceEnrichmentContext) Enrich(resource pcommon.Resource, cfg config if cfg.OverrideHostName.Enabled { s.overrideHostNameWithK8sNodeName(resource) } + if cfg.DeploymentEnvironment.Enabled { s.setDeploymentEnvironment(resource) } + if cfg.DefaultDeploymentEnvironment.Enabled { + s.setDefaultDeploymentEnvironment(resource) + } + if cfg.ServiceInstanceID.Enabled { s.setServiceInstanceID(resource) } + if cfg.ServiceName.Enabled { + s.sanitizeServiceName(resource) + } if cfg.HostOSType.Enabled { s.setHostOSType(resource) } @@ -117,13 +129,32 @@ func (s *resourceEnrichmentContext) Enrich(resource pcommon.Resource, cfg config // ES currently doesn't allow aliases with multiple targets, so if the new field name is used (SemConv v1.27+), // we duplicate the value and also send it with the old field name to make the alias work. func (s *resourceEnrichmentContext) setDeploymentEnvironment(resource pcommon.Resource) { - if s.deploymentEnvironmentName != "" && s.deploymentEnvironment == "" { - attribute.PutStr( - resource.Attributes(), - string(semconv25.DeploymentEnvironmentKey), - s.deploymentEnvironmentName, - ) + if s.deploymentEnvironment != "" { + return + } + if s.deploymentEnvironmentName == "" { + return + } + + attribute.PutStr( + resource.Attributes(), + string(semconv25.DeploymentEnvironmentKey), + s.deploymentEnvironmentName, + ) +} + +// setDefaultDeploymentEnvironment sets deployment.environment to "unset" +// when neither deployment environment field is present, matching the +// apm-data/MIS default only where explicitly configured. +func (s *resourceEnrichmentContext) setDefaultDeploymentEnvironment(resource pcommon.Resource) { + if s.deploymentEnvironment != "" || s.deploymentEnvironmentName != "" { + return } + attribute.PutStr( + resource.Attributes(), + string(semconv25.DeploymentEnvironmentKey), + "unset", + ) } func (s *resourceEnrichmentContext) setAgentName(resource pcommon.Resource) { @@ -225,3 +256,13 @@ func (s *resourceEnrichmentContext) setServiceInstanceID(resource pcommon.Resour } attribute.PutStr(resource.Attributes(), string(semconv25.ServiceInstanceIDKey), s.serviceInstanceID) } + +func (s *resourceEnrichmentContext) sanitizeServiceName(resource pcommon.Resource) { + if s.serviceName == "" { + return + } + cleaned := sanitize.CleanServiceName(s.serviceName) + if cleaned != s.serviceName { + resource.Attributes().PutStr(string(semconv.ServiceNameKey), cleaned) + } +} diff --git a/processor/elasticapmprocessor/internal/enrichments/resource_test.go b/processor/elasticapmprocessor/internal/enrichments/resource_test.go index 487a34c07..f65d33b59 100644 --- a/processor/elasticapmprocessor/internal/enrichments/resource_test.go +++ b/processor/elasticapmprocessor/internal/enrichments/resource_test.go @@ -31,9 +31,12 @@ import ( ) // ecsResourceConfig returns a ResourceConfig that mirrors the ECS enricher -// configuration in processor.go: base Enabled() with HostOSType explicitly enabled. +// configuration in processor.go: base Enabled() with ECS-only resource +// enrichments explicitly enabled. func ecsResourceConfig() config.ResourceConfig { c := config.Enabled().Resource + c.DefaultDeploymentEnvironment.Enabled = true + c.ServiceName.Enabled = true c.HostOSType.Enabled = true return c } @@ -53,7 +56,21 @@ func TestResourceEnrich(t *testing.T) { { name: "empty", input: pcommon.NewResource(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), + enrichedAttrs: map[string]any{ + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + }, + }, + { + name: "empty_default_deployment_environment_disabled", + input: pcommon.NewResource(), + config: func() config.ResourceConfig { + c := ecsResourceConfig() + c.DefaultDeploymentEnvironment.Enabled = false + return c + }(), enrichedAttrs: map[string]any{ elasticattr.AgentName: "otlp", elasticattr.AgentVersion: "unknown", @@ -66,10 +83,11 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv.TelemetrySDKNameKey), "customflavor") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "customflavor", - elasticattr.AgentVersion: "unknown", + elasticattr.AgentName: "customflavor", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -80,10 +98,11 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv.TelemetryDistroNameKey), "elastic") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "customflavor/unknown/elastic", - elasticattr.AgentVersion: "unknown", + elasticattr.AgentName: "customflavor/unknown/elastic", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -95,10 +114,11 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv.TelemetryDistroNameKey), "elastic") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "customflavor/cpp/elastic", - elasticattr.AgentVersion: "unknown", + elasticattr.AgentName: "customflavor/cpp/elastic", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -108,10 +128,11 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv.TelemetrySDKLanguageKey), "cpp") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp/cpp", - elasticattr.AgentVersion: "unknown", + elasticattr.AgentName: "otlp/cpp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -122,10 +143,11 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv.TelemetrySDKLanguageKey), "cpp") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "customflavor/cpp", - elasticattr.AgentVersion: "unknown", + elasticattr.AgentName: "customflavor/cpp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -136,10 +158,11 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv.TelemetrySDKVersionKey), "9.999.9") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "customflavor", - elasticattr.AgentVersion: "9.999.9", + elasticattr.AgentName: "customflavor", + elasticattr.AgentVersion: "9.999.9", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -151,10 +174,11 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv.TelemetryDistroNameKey), "elastic") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "customflavor/unknown/elastic", - elasticattr.AgentVersion: "unknown", + elasticattr.AgentName: "customflavor/unknown/elastic", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -167,10 +191,11 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv.TelemetryDistroVersionKey), "1.2.3") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "customflavor/unknown/elastic", - elasticattr.AgentVersion: "1.2.3", + elasticattr.AgentName: "customflavor/unknown/elastic", + elasticattr.AgentVersion: "1.2.3", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -181,13 +206,14 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv.K8SNodeNameKey), "k8s-node") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - string(semconv.HostNameKey): "k8s-node", - string(semconv.K8SNodeNameKey): "k8s-node", - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", - string(semconv25.ServiceInstanceIDKey): string("test-host"), + string(semconv.HostNameKey): "k8s-node", + string(semconv.K8SNodeNameKey): "k8s-node", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.ServiceInstanceIDKey): string("test-host"), + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -197,12 +223,13 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv.K8SNodeNameKey), "k8s-node") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - string(semconv.HostNameKey): "k8s-node", - string(semconv.K8SNodeNameKey): "k8s-node", - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", + string(semconv.HostNameKey): "k8s-node", + string(semconv.K8SNodeNameKey): "k8s-node", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -213,7 +240,7 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv25.DeploymentEnvironmentKey), "prod") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ string(semconv25.DeploymentEnvironmentKey): "prod", elasticattr.AgentName: "otlp", @@ -228,7 +255,7 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv.DeploymentEnvironmentNameKey), "prod") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ // To satisfy aliases defined in ES, we duplicate the value for both fields. string(semconv25.DeploymentEnvironmentKey): "prod", @@ -237,6 +264,25 @@ func TestResourceEnrich(t *testing.T) { elasticattr.AgentVersion: "unknown", }, }, + { + name: "deployment_environment_name_set_default_disabled", + input: func() pcommon.Resource { + res := pcommon.NewResource() + res.Attributes().PutStr(string(semconv.DeploymentEnvironmentNameKey), "prod") + return res + }(), + config: func() config.ResourceConfig { + c := ecsResourceConfig() + c.DefaultDeploymentEnvironment.Enabled = false + return c + }(), + enrichedAttrs: map[string]any{ + string(semconv25.DeploymentEnvironmentKey): "prod", + string(semconv.DeploymentEnvironmentNameKey): "prod", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + }, + }, { // Mixed pre and post SemConv 1.27 versions (should be an edge case, but some EDOTs might do this). name: "deployment_environment_mixed", @@ -246,7 +292,7 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv25.DeploymentEnvironmentKey), "test") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ // If both are set, we don't touch those values and take them as they are. string(semconv25.DeploymentEnvironmentKey): "test", @@ -263,13 +309,14 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv25.HostNameKey), "k8s-node") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - string(semconv25.ServiceInstanceIDKey): "container-id", - string(semconv.ContainerIDKey): "container-id", - string(semconv.HostNameKey): "k8s-node", - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", + string(semconv25.ServiceInstanceIDKey): "container-id", + string(semconv.ContainerIDKey): "container-id", + string(semconv.HostNameKey): "k8s-node", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -279,12 +326,13 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv25.HostNameKey), "k8s-node") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - string(semconv25.ServiceInstanceIDKey): "k8s-node", - string(semconv.HostNameKey): "k8s-node", - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", + string(semconv25.ServiceInstanceIDKey): "k8s-node", + string(semconv.HostNameKey): "k8s-node", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -296,13 +344,86 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv25.HostNameKey), "k8s-node") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), + enrichedAttrs: map[string]any{ + string(semconv25.ServiceInstanceIDKey): "node-name", + string(semconv.ContainerIDKey): "container-id", + string(semconv.HostNameKey): "k8s-node", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + }, + }, + { + name: "service_name_sanitized_overwrites_existing_value", + input: func() pcommon.Resource { + res := pcommon.NewResource() + res.Attributes().PutStr(string(semconv.ServiceNameKey), "my/service") + return res + }(), + config: ecsResourceConfig(), + enrichedAttrs: map[string]any{ + string(semconv.ServiceNameKey): "my_service", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + }, + }, + { + name: "agent_version_disabled", + input: func() pcommon.Resource { + res := pcommon.NewResource() + res.Attributes().PutStr(string(semconv.TelemetrySDKVersionKey), "1.2.3") + return res + }(), + config: func() config.ResourceConfig { + c := ecsResourceConfig() + c.AgentVersion.Enabled = false + return c + }(), + enrichedAttrs: map[string]any{ + elasticattr.AgentName: "otlp", + string(semconv25.DeploymentEnvironmentKey): "unset", + }, + }, + { + name: "agent_version_enabled_defaults_to_unknown", + input: pcommon.NewResource(), + config: ecsResourceConfig(), + enrichedAttrs: map[string]any{ + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + }, + }, + { + name: "agent_version_enabled_uses_sdk_version", + input: func() pcommon.Resource { + res := pcommon.NewResource() + res.Attributes().PutStr(string(semconv.TelemetrySDKVersionKey), "1.2.3") + return res + }(), + config: ecsResourceConfig(), + enrichedAttrs: map[string]any{ + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "1.2.3", + string(semconv25.DeploymentEnvironmentKey): "unset", + }, + }, + { + name: "agent_version_enabled_uses_distro_version_over_sdk_version", + input: func() pcommon.Resource { + res := pcommon.NewResource() + res.Attributes().PutStr(string(semconv.TelemetrySDKVersionKey), "1.2.3") + res.Attributes().PutStr(string(semconv.TelemetryDistroNameKey), "elastic") + res.Attributes().PutStr(string(semconv.TelemetryDistroVersionKey), "4.5.6") + return res + }(), + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - string(semconv25.ServiceInstanceIDKey): "node-name", - string(semconv.ContainerIDKey): "container-id", - string(semconv.HostNameKey): "k8s-node", - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", + elasticattr.AgentName: "otlp/unknown/elastic", + elasticattr.AgentVersion: "4.5.6", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -318,17 +439,18 @@ func TestResourceEnrich(t *testing.T) { res.Attributes().PutStr(string(semconv.HostNameKey), "host-name") return res }(), - config: config.Enabled().Resource, + config: ecsResourceConfig(), enrichedAttrs: map[string]any{ // existing attributes are preserved (not overwritten) elasticattr.AgentName: "existing-agent-name", elasticattr.AgentVersion: "existing-agent-version", string(semconv25.ServiceInstanceIDKey): "existing-service-instance-id", // source attributes remain unchanged - string(semconv.TelemetrySDKNameKey): "customflavor", - string(semconv.TelemetrySDKVersionKey): "9.999.9", - string(semconv25.ContainerIDKey): "container-id", - string(semconv.HostNameKey): "host-name", + string(semconv.TelemetrySDKNameKey): "customflavor", + string(semconv.TelemetrySDKVersionKey): "9.999.9", + string(semconv25.ContainerIDKey): "container-id", + string(semconv.HostNameKey): "host-name", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -340,9 +462,10 @@ func TestResourceEnrich(t *testing.T) { }(), config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", - elasticattr.HostOSType: "windows", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + elasticattr.HostOSType: "windows", }, }, { @@ -354,9 +477,10 @@ func TestResourceEnrich(t *testing.T) { }(), config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", - elasticattr.HostOSType: "linux", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + elasticattr.HostOSType: "linux", }, }, { @@ -368,9 +492,10 @@ func TestResourceEnrich(t *testing.T) { }(), config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", - elasticattr.HostOSType: "macos", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + elasticattr.HostOSType: "macos", }, }, { @@ -382,9 +507,10 @@ func TestResourceEnrich(t *testing.T) { }(), config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", - elasticattr.HostOSType: "unix", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + elasticattr.HostOSType: "unix", }, }, { @@ -396,9 +522,10 @@ func TestResourceEnrich(t *testing.T) { }(), config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", - elasticattr.HostOSType: "unix", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + elasticattr.HostOSType: "unix", }, }, { @@ -410,9 +537,10 @@ func TestResourceEnrich(t *testing.T) { }(), config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", - elasticattr.HostOSType: "unix", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + elasticattr.HostOSType: "unix", }, }, { @@ -424,8 +552,9 @@ func TestResourceEnrich(t *testing.T) { }(), config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -437,9 +566,10 @@ func TestResourceEnrich(t *testing.T) { }(), config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", - elasticattr.HostOSType: "android", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + elasticattr.HostOSType: "android", }, }, { @@ -451,9 +581,10 @@ func TestResourceEnrich(t *testing.T) { }(), config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", - elasticattr.HostOSType: "ios", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + elasticattr.HostOSType: "ios", }, }, { @@ -466,9 +597,10 @@ func TestResourceEnrich(t *testing.T) { }(), config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", - elasticattr.HostOSType: "android", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + elasticattr.HostOSType: "android", }, }, { @@ -480,8 +612,9 @@ func TestResourceEnrich(t *testing.T) { }(), config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", }, }, { @@ -494,9 +627,10 @@ func TestResourceEnrich(t *testing.T) { }(), config: ecsResourceConfig(), enrichedAttrs: map[string]any{ - elasticattr.AgentName: "otlp", - elasticattr.AgentVersion: "unknown", - elasticattr.HostOSType: "custom", + elasticattr.AgentName: "otlp", + elasticattr.AgentVersion: "unknown", + string(semconv25.DeploymentEnvironmentKey): "unset", + elasticattr.HostOSType: "custom", }, }, { diff --git a/processor/elasticapmprocessor/internal/routing/data_stream.go b/processor/elasticapmprocessor/internal/routing/data_stream.go index 691d39961..646c82c56 100644 --- a/processor/elasticapmprocessor/internal/routing/data_stream.go +++ b/processor/elasticapmprocessor/internal/routing/data_stream.go @@ -19,10 +19,10 @@ package routing // import "github.com/elastic/opentelemetry-collector-components import ( "fmt" - "strings" "github.com/elastic/apm-data/model/modelprocessor" "github.com/elastic/opentelemetry-collector-components/internal/elasticattr" + "github.com/elastic/opentelemetry-collector-components/processor/elasticapmprocessor/internal/sanitize" "go.opentelemetry.io/collector/pdata/pcommon" semconv "go.opentelemetry.io/otel/semconv/v1.38.0" ) @@ -51,9 +51,9 @@ func EncodeDataStream(resource pcommon.Resource, dataStreamType string, serviceN 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", NamespaceDefault) + attributes.PutStr(elasticattr.DataStreamType, dataStreamType) + attributes.PutStr(elasticattr.DataStreamDataset, "apm") + attributes.PutStr(elasticattr.DataStreamNamespace, NamespaceDefault) } func encodeDataStreamWithServiceName(resource pcommon.Resource, dataStreamType string) { @@ -64,37 +64,9 @@ func encodeDataStreamWithServiceName(resource pcommon.Resource, dataStreamType s 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 + attributes.PutStr(elasticattr.DataStreamType, dataStreamType) + attributes.PutStr(elasticattr.DataStreamDataset, "apm.app."+sanitize.NormalizeServiceName(serviceName.Str())) + attributes.PutStr(elasticattr.DataStreamNamespace, NamespaceDefault) } // IsErrorEvent checks if a log record or span event represents an APM error event. @@ -123,9 +95,9 @@ func IsErrorEvent(attributes pcommon.Map) bool { // Error logs should always be routed to the "apm.error" dataset regardless of service name. // Error span events should also be routed to the "apm.error" dataset. func EncodeErrorDataStream(attributes pcommon.Map, dataStreamType string) { - attributes.PutStr("data_stream.type", dataStreamType) - attributes.PutStr("data_stream.dataset", "apm.error") - attributes.PutStr("data_stream.namespace", NamespaceDefault) + attributes.PutStr(elasticattr.DataStreamType, dataStreamType) + attributes.PutStr(elasticattr.DataStreamDataset, "apm.error") + attributes.PutStr(elasticattr.DataStreamNamespace, NamespaceDefault) } // hasTransactionSpanContext checks if the attributes contain transaction or span context. @@ -133,21 +105,21 @@ func EncodeErrorDataStream(attributes pcommon.Map, dataStreamType string) { // routed to internal metrics data streams. func hasTransactionSpanContext(attributes pcommon.Map) bool { // Check for transaction-related attributes - if _, ok := attributes.Get("transaction.id"); ok { + if _, ok := attributes.Get(elasticattr.TransactionID); ok { return true } - if _, ok := attributes.Get("transaction.name"); ok { + if _, ok := attributes.Get(elasticattr.TransactionName); ok { return true } - if _, ok := attributes.Get("transaction.type"); ok { + if _, ok := attributes.Get(elasticattr.TransactionType); ok { return true } // Check for span-related attributes - if _, ok := attributes.Get("span.id"); ok { + if _, ok := attributes.Get(elasticattr.SpanID); ok { return true } - if _, ok := attributes.Get("span.name"); ok { + if _, ok := attributes.Get(elasticattr.SpanName); ok { return true } @@ -178,16 +150,16 @@ func isServiceSummary(attributes pcommon.Map) bool { // Using default "apm.internal" data stream func internalMetricDataStream(attributes pcommon.Map, dataStreamType string) { - attributes.PutStr("data_stream.type", dataStreamType) - attributes.PutStr("data_stream.dataset", "apm.internal") - attributes.PutStr("data_stream.namespace", NamespaceDefault) + attributes.PutStr(elasticattr.DataStreamType, dataStreamType) + attributes.PutStr(elasticattr.DataStreamDataset, "apm.internal") + attributes.PutStr(elasticattr.DataStreamNamespace, NamespaceDefault) } // Data stream formatted as: apm.${metricset.name}.${metricset.interval} func internalIntervalMetricDataStream(attributes pcommon.Map, dataStreamType, metricsetName, interval string) { - attributes.PutStr("data_stream.type", dataStreamType) - attributes.PutStr("data_stream.dataset", fmt.Sprintf("apm.%s.%s", metricsetName, interval)) - attributes.PutStr("data_stream.namespace", NamespaceDefault) + attributes.PutStr(elasticattr.DataStreamType, dataStreamType) + attributes.PutStr(elasticattr.DataStreamDataset, fmt.Sprintf("apm.%s.%s", metricsetName, interval)) + attributes.PutStr(elasticattr.DataStreamNamespace, NamespaceDefault) } // EncodeDataStreamMetricDataPoint determines if the metric represents an internal metric diff --git a/processor/elasticapmprocessor/internal/routing/data_stream_test.go b/processor/elasticapmprocessor/internal/routing/data_stream_test.go index babe16320..f7725edf4 100644 --- a/processor/elasticapmprocessor/internal/routing/data_stream_test.go +++ b/processor/elasticapmprocessor/internal/routing/data_stream_test.go @@ -45,24 +45,169 @@ func TestDataStremaEncoderDefault(t *testing.T) { assert.Equal(t, "default", dataStreamNamespace.Str()) } -func TestDataStreamEncoderWithServiceName(t *testing.T) { - resource := pcommon.NewResource() - attributes := resource.Attributes() - attributes.PutStr("service.name", "my-service") +func TestEncodeDataStreamWithServiceName(t *testing.T) { + // The normalizeServiceName function lowercases the input and then applies + // the regex [^a-zA-Z0-9_] to replace disallowed characters with "_". + // This aligns with apm-data's normalizeServiceName: hyphens, spaces, + // dots, and all other non-alphanumeric/non-underscore characters are + // replaced with "_". + tests := []struct { + name string + serviceName string + dataStreamType string + expectedDataset string + }{ + { + name: "simple alphanumeric", + serviceName: "myservice", + dataStreamType: "metrics", + expectedDataset: "apm.app.myservice", + }, + { + name: "hyphen replaced", + serviceName: "my-service", + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service", + }, + { + name: "dot replaced", + serviceName: "my.service", + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service", + }, + { + name: "uppercase lowercased with hyphen replaced", + serviceName: "My-Service", + dataStreamType: "logs", + expectedDataset: "apm.app.my_service", + }, + { + name: "backslash replaced", + serviceName: `my\service`, + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service", + }, + { + name: "forward slash replaced", + serviceName: "my/service", + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service", + }, + { + name: "asterisk replaced", + serviceName: "my*service", + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service", + }, + { + name: "question mark replaced", + serviceName: "my?service", + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service", + }, + { + name: "double quote replaced", + serviceName: `my"service`, + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service", + }, + { + name: "angle brackets replaced", + serviceName: "myname", + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service_name", + }, + { + name: "pipe replaced", + serviceName: "my|service", + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service", + }, + { + name: "space replaced", + serviceName: "my service", + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service", + }, + { + name: "comma replaced", + serviceName: "my,service", + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service", + }, + { + name: "hash replaced", + serviceName: "my#service", + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service", + }, + { + name: "colon replaced", + serviceName: "my:service", + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service", + }, + { + name: "multiple special characters", + serviceName: "My Service.v2-beta/rc#1", + dataStreamType: "logs", + expectedDataset: "apm.app.my_service_v2_beta_rc_1", + }, + { + name: "all disallowed characters replaced", + serviceName: `a\b/c*d?e"fh|i,j#k:l.m`, + dataStreamType: "metrics", + expectedDataset: "apm.app.a_b_c_d_e_f_g_h_i_j_k_l_m", + }, + { + name: "only alphanumeric and underscore preserved", + serviceName: `a b-c_d0`, + dataStreamType: "metrics", + expectedDataset: "apm.app.a_b_c_d0", + }, + { + name: "empty service name falls back to unknown", + serviceName: "", + dataStreamType: "metrics", + expectedDataset: "apm.app.unknown", + }, + { + name: "underscores preserved", + serviceName: "my_service_name", + dataStreamType: "metrics", + expectedDataset: "apm.app.my_service_name", + }, + { + name: "traces type", + serviceName: "my-service", + dataStreamType: "traces", + expectedDataset: "apm.app.my_service", + }, + } - routing.EncodeDataStream(resource, "metrics", true) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + resource := pcommon.NewResource() + attributes := resource.Attributes() + if tt.serviceName != "" { + attributes.PutStr("service.name", tt.serviceName) + } - dataStreamType, ok := attributes.Get("data_stream.type") - assert.True(t, ok) - assert.Equal(t, "metrics", dataStreamType.Str()) + routing.EncodeDataStream(resource, tt.dataStreamType, true) - dataStreamDataset, ok := attributes.Get("data_stream.dataset") - assert.True(t, ok) - assert.Equal(t, "apm.app.my_service", dataStreamDataset.Str()) + dataStreamType, ok := attributes.Get("data_stream.type") + assert.True(t, ok) + assert.Equal(t, tt.dataStreamType, dataStreamType.Str()) - dataStreamNamespace, ok := attributes.Get("data_stream.namespace") - assert.True(t, ok) - assert.Equal(t, "default", dataStreamNamespace.Str()) + dataStreamDataset, ok := attributes.Get("data_stream.dataset") + assert.True(t, ok) + assert.Equal(t, tt.expectedDataset, dataStreamDataset.Str()) + + dataStreamNamespace, ok := attributes.Get("data_stream.namespace") + assert.True(t, ok) + assert.Equal(t, "default", dataStreamNamespace.Str()) + }) + } } func TestIsErrorEvent(t *testing.T) { diff --git a/processor/elasticapmprocessor/internal/sanitize/sanitize.go b/processor/elasticapmprocessor/internal/sanitize/sanitize.go new file mode 100644 index 000000000..1abcd8f3b --- /dev/null +++ b/processor/elasticapmprocessor/internal/sanitize/sanitize.go @@ -0,0 +1,119 @@ +// 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 sanitize // import "github.com/elastic/opentelemetry-collector-components/processor/elasticapmprocessor/internal/sanitize" + +import ( + "regexp" + "strings" +) + +const ( + MaxDataStreamBytes = 100 +) + +var ( + serviceNameInvalidRegexp = regexp.MustCompile("[^a-zA-Z0-9 _-]") +) + +// Truncate returns s truncated at n runes, and the number of runes in the resulting string (<= n). +func Truncate(s string, length uint) string { + var j uint + for i := range s { + if j == length { + return s[:i] + } + j++ + } + return s +} + +func replaceReservedLabelKeyRune(r rune) rune { + switch r { + case '.', '*', '"': + return '_' + } + return r +} + +// HandleAttributeKey sanitizes an attribute key, replacing the reserved characters +// '.', '*' and '"' with '_'. This matches the apm-server behavior. +// This matches the logic in the apm-data library here: +// https://github.com/elastic/apm-data/blob/e3e170b/model/modeljson/labels.go. +func HandleAttributeKey(k string) string { + if strings.ContainsAny(k, ".*\"") { + return strings.Map(replaceReservedLabelKeyRune, k) + } + return k +} + +// HandleLabelAttributeKey sanitizes the suffix portion of a label attribute, +// preserving the "labels." or "numeric_labels." prefix. +func HandleLabelAttributeKey(attr string) string { + if strings.HasPrefix(attr, "labels.") { + return "labels." + HandleAttributeKey(strings.TrimPrefix(attr, "labels.")) + } + if strings.HasPrefix(attr, "numeric_labels.") { + return "numeric_labels." + HandleAttributeKey(strings.TrimPrefix(attr, "numeric_labels.")) + } + return attr +} + +// IsLabelAttribute returns true if the resource attribute is already a prefixed label. +// The elasticapmintake receiver moves labels and numeric_labels into attributes and +// already prefixes those with "labels." and "numeric_labels." respectively and also does de-dotting. +// So for those, we don't want to double prefix - we just leave them as is. +func IsLabelAttribute(attr string) bool { + return strings.HasPrefix(attr, "labels.") || strings.HasPrefix(attr, "numeric_labels.") +} + +// The following is adapted 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 '.': + // added for parity + return '_' + case '-': + // Hyphens are used to separate the data stream type, dataset, + // and namespace. + return '_' + } + return r +} + +// CleanServiceName sanitizes a service name by truncating it to a defined length and replacing invalid characters with "_". +// see https://github.com/elastic/apm-data/blob/34677210900a68d6204cdb79da4ce0d1ee685d9a/input/otlp/metadata.go#L491 +func CleanServiceName(name string) string { + return serviceNameInvalidRegexp.ReplaceAllString(Truncate(name, MaxDataStreamBytes), "_") +} diff --git a/processor/elasticapmprocessor/processor.go b/processor/elasticapmprocessor/processor.go index 075efac81..ec35af754 100644 --- a/processor/elasticapmprocessor/processor.go +++ b/processor/elasticapmprocessor/processor.go @@ -56,9 +56,11 @@ type TraceProcessor struct { func NewTraceProcessor(cfg *Config, next consumer.Traces, logger *zap.Logger) *TraceProcessor { enricherConfig := cfg.Config + ecsEnricherConfig := cfg.Config - ecsEnricherConfig.Resource.DeploymentEnvironment.Enabled = false ecsEnricherConfig.Resource.HostOSType.Enabled = true + ecsEnricherConfig.Resource.ServiceName.Enabled = true + ecsEnricherConfig.Resource.DefaultDeploymentEnvironment.Enabled = true intakeECSEnricherConfig := ecsEnricherConfig // The intake receiver already sets transaction.root; skip re-deriving it @@ -162,10 +164,11 @@ type LogProcessor struct { func newLogProcessor(cfg *Config, next consumer.Logs, logger *zap.Logger) *LogProcessor { enricherConfig := cfg.Config + ecsEnricherConfig := cfg.Config - ecsEnricherConfig.Resource.DeploymentEnvironment.Enabled = false - ecsEnricherConfig.Resource.AgentVersion.Enabled = false ecsEnricherConfig.Resource.HostOSType.Enabled = true + ecsEnricherConfig.Resource.ServiceName.Enabled = true + ecsEnricherConfig.Resource.DefaultDeploymentEnvironment.Enabled = true // disable the transaction result enrichment to avoid deriving a value // when the provided result is empty to match existing apm-data logic ecsEnricherConfig.Transaction.Result.Enabled = false @@ -201,9 +204,11 @@ type MetricProcessor struct { func newMetricProcessor(cfg *Config, next consumer.Metrics, logger *zap.Logger) *MetricProcessor { enricherConfig := cfg.Config + ecsEnricherConfig := cfg.Config - ecsEnricherConfig.Resource.DeploymentEnvironment.Enabled = false ecsEnricherConfig.Resource.HostOSType.Enabled = true + ecsEnricherConfig.Resource.ServiceName.Enabled = true + ecsEnricherConfig.Resource.DefaultDeploymentEnvironment.Enabled = true intakeECSEnricherConfig := ecsEnricherConfig intakeECSEnricherConfig.Resource.HostOSType.Enabled = false diff --git a/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_input.yaml b/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_input.yaml index a0dcccdae..e6049b9da 100644 --- a/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_input.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_input.yaml @@ -3,7 +3,7 @@ resourceLogs: attributes: - key: service.name value: - stringValue: test-service + stringValue: test/service - key: labels.my_value value: stringValue: foo diff --git a/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_otlp_exception_output.yaml b/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_otlp_exception_output.yaml index 2d77dec1f..d6bed4cc3 100644 --- a/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_otlp_exception_output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_otlp_exception_output.yaml @@ -28,6 +28,12 @@ resourceLogs: - key: agent.name value: stringValue: otlp + - key: agent.version + value: + stringValue: unknown + - key: deployment.environment + value: + stringValue: unset scopeLogs: - logRecords: - attributes: diff --git a/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_output.yaml b/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_output.yaml index 0e7647de0..0c8889e70 100644 --- a/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_output.yaml @@ -3,7 +3,7 @@ resourceLogs: attributes: - key: service.name value: - stringValue: test-service + stringValue: test_service - key: labels.my_value value: stringValue: foo @@ -28,6 +28,12 @@ resourceLogs: - key: agent.name value: stringValue: otlp + - key: agent.version + value: + stringValue: unknown + - key: deployment.environment + value: + stringValue: unset scopeLogs: - logRecords: - attributes: diff --git a/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_servicename_input.yaml b/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_servicename_input.yaml index 75200f259..6bc490234 100644 --- a/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_servicename_input.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_servicename_input.yaml @@ -3,7 +3,7 @@ resourceLogs: attributes: - key: service.name value: - stringValue: my-app + stringValue: My App - key: labels.my_value value: stringValue: foo diff --git a/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_servicename_output.yaml b/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_servicename_output.yaml index e6b7671bd..0d441e8d9 100644 --- a/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_servicename_output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/elastic_error/logs_servicename_output.yaml @@ -3,7 +3,7 @@ resourceLogs: attributes: - key: service.name value: - stringValue: my-app + stringValue: My App - key: labels.my_value value: stringValue: foo @@ -28,6 +28,12 @@ resourceLogs: - key: agent.name value: stringValue: otlp + - key: agent.version + value: + stringValue: unknown + - key: deployment.environment + value: + stringValue: unset scopeLogs: - logRecords: - attributes: diff --git a/processor/elasticapmprocessor/testdata/ecs/elastic_hostname/logs_output.yaml b/processor/elasticapmprocessor/testdata/ecs/elastic_hostname/logs_output.yaml index c23ed4cfa..9aaec2f64 100644 --- a/processor/elasticapmprocessor/testdata/ecs/elastic_hostname/logs_output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/elastic_hostname/logs_output.yaml @@ -22,6 +22,12 @@ resourceLogs: - key: agent.name value: stringValue: otlp + - key: agent.version + value: + stringValue: unknown + - key: deployment.environment + value: + stringValue: unset - key: service.instance.id value: stringValue: hostname-1 diff --git a/processor/elasticapmprocessor/testdata/ecs/elastic_hostname/metrics_output.yaml b/processor/elasticapmprocessor/testdata/ecs/elastic_hostname/metrics_output.yaml index 3d246612c..06e8046f9 100644 --- a/processor/elasticapmprocessor/testdata/ecs/elastic_hostname/metrics_output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/elastic_hostname/metrics_output.yaml @@ -16,6 +16,9 @@ resourceMetrics: - key: data_stream.type value: stringValue: metrics + - key: deployment.environment + value: + stringValue: unset - key: host.hostname value: stringValue: hostname-1 diff --git a/processor/elasticapmprocessor/testdata/ecs/elastic_hostname/spans_output.yaml b/processor/elasticapmprocessor/testdata/ecs/elastic_hostname/spans_output.yaml index 4b8533aa5..e2832ca64 100644 --- a/processor/elasticapmprocessor/testdata/ecs/elastic_hostname/spans_output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/elastic_hostname/spans_output.yaml @@ -22,6 +22,9 @@ resourceSpans: - key: agent.version value: stringValue: unknown + - key: deployment.environment + value: + stringValue: unset - key: service.instance.id value: stringValue: hostname-1 diff --git a/processor/elasticapmprocessor/testdata/ecs/elastic_internal_metrics/output.yaml b/processor/elasticapmprocessor/testdata/ecs/elastic_internal_metrics/output.yaml index 3307a747b..cfbe2a3ab 100644 --- a/processor/elasticapmprocessor/testdata/ecs/elastic_internal_metrics/output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/elastic_internal_metrics/output.yaml @@ -16,6 +16,9 @@ resourceMetrics: - key: data_stream.type value: stringValue: metrics + - key: deployment.environment + value: + stringValue: unset - key: host.ip value: stringValue: 1.2.3.4 diff --git a/processor/elasticapmprocessor/testdata/ecs/elastic_log_device_crash_event/output.yaml b/processor/elasticapmprocessor/testdata/ecs/elastic_log_device_crash_event/output.yaml index 47ea13dcd..230eb5b9f 100644 --- a/processor/elasticapmprocessor/testdata/ecs/elastic_log_device_crash_event/output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/elastic_log_device_crash_event/output.yaml @@ -25,6 +25,12 @@ resourceLogs: - key: agent.name value: stringValue: android/java + - key: agent.version + value: + stringValue: unknown + - key: deployment.environment + value: + stringValue: unset scopeLogs: - logRecords: - attributes: diff --git a/processor/elasticapmprocessor/testdata/ecs/elastic_log_device_event/output.yaml b/processor/elasticapmprocessor/testdata/ecs/elastic_log_device_event/output.yaml index 00270d414..eb8e33b8d 100644 --- a/processor/elasticapmprocessor/testdata/ecs/elastic_log_device_event/output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/elastic_log_device_event/output.yaml @@ -25,6 +25,12 @@ resourceLogs: - key: agent.name value: stringValue: android/java + - key: agent.version + value: + stringValue: unknown + - key: deployment.environment + value: + stringValue: unset scopeLogs: - logRecords: - attributes: diff --git a/processor/elasticapmprocessor/testdata/ecs/intake/logs_error_output.yaml b/processor/elasticapmprocessor/testdata/ecs/intake/logs_error_output.yaml index 00cd64214..7902391c3 100644 --- a/processor/elasticapmprocessor/testdata/ecs/intake/logs_error_output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/intake/logs_error_output.yaml @@ -25,6 +25,12 @@ resourceLogs: - key: agent.name value: stringValue: ElasticAPM + - key: agent.version + value: + stringValue: unknown + - key: deployment.environment + value: + stringValue: unset scopeLogs: - logRecords: - attributes: diff --git a/processor/elasticapmprocessor/testdata/ecs/intake/logs_output.yaml b/processor/elasticapmprocessor/testdata/ecs/intake/logs_output.yaml index 4eadf03e3..eb415c731 100644 --- a/processor/elasticapmprocessor/testdata/ecs/intake/logs_output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/intake/logs_output.yaml @@ -25,6 +25,12 @@ resourceLogs: - key: agent.name value: stringValue: ElasticAPM + - key: agent.version + value: + stringValue: unknown + - key: deployment.environment + value: + stringValue: unset scopeLogs: - logRecords: - body: diff --git a/processor/elasticapmprocessor/testdata/ecs/intake/metrics_output.yaml b/processor/elasticapmprocessor/testdata/ecs/intake/metrics_output.yaml index ab788cc8f..5d993af66 100644 --- a/processor/elasticapmprocessor/testdata/ecs/intake/metrics_output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/intake/metrics_output.yaml @@ -16,6 +16,9 @@ resourceMetrics: - key: data_stream.type value: stringValue: metrics + - key: deployment.environment + value: + stringValue: unset - key: host.ip value: stringValue: 1.2.3.4 diff --git a/processor/elasticapmprocessor/testdata/ecs/intake/traces_output.yaml b/processor/elasticapmprocessor/testdata/ecs/intake/traces_output.yaml index 65940fbe6..0c2fdec79 100644 --- a/processor/elasticapmprocessor/testdata/ecs/intake/traces_output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/intake/traces_output.yaml @@ -28,6 +28,9 @@ resourceSpans: - key: agent.version value: stringValue: unknown + - key: deployment.environment + value: + stringValue: unset scopeSpans: - scope: {} spans: diff --git a/processor/elasticapmprocessor/testdata/ecs/intake/traces_txn_db_output.yaml b/processor/elasticapmprocessor/testdata/ecs/intake/traces_txn_db_output.yaml index a725fd94b..a62b664a8 100644 --- a/processor/elasticapmprocessor/testdata/ecs/intake/traces_txn_db_output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/intake/traces_txn_db_output.yaml @@ -28,6 +28,9 @@ resourceSpans: - key: agent.version value: stringValue: unknown + - key: deployment.environment + value: + stringValue: unset scopeSpans: - scope: {} spans: diff --git a/processor/elasticapmprocessor/testdata/ecs/txn_db/output.yaml b/processor/elasticapmprocessor/testdata/ecs/txn_db/output.yaml index d88369a24..581dd7a4c 100644 --- a/processor/elasticapmprocessor/testdata/ecs/txn_db/output.yaml +++ b/processor/elasticapmprocessor/testdata/ecs/txn_db/output.yaml @@ -25,6 +25,9 @@ resourceSpans: - key: agent.version value: stringValue: unknown + - key: deployment.environment + value: + stringValue: unset scopeSpans: - scope: {} spans: diff --git a/processor/elasticapmprocessor/testdata/skip_enrichment/logs_false_ecs_output.yaml b/processor/elasticapmprocessor/testdata/skip_enrichment/logs_false_ecs_output.yaml index 846416598..85c803f1d 100644 --- a/processor/elasticapmprocessor/testdata/skip_enrichment/logs_false_ecs_output.yaml +++ b/processor/elasticapmprocessor/testdata/skip_enrichment/logs_false_ecs_output.yaml @@ -16,6 +16,12 @@ resourceLogs: - key: agent.name value: stringValue: otlp + - key: agent.version + value: + stringValue: unknown + - key: deployment.environment + value: + stringValue: unset scopeLogs: - logRecords: - body: diff --git a/processor/elasticapmprocessor/testdata/skip_enrichment/logs_true_ecs_output.yaml b/processor/elasticapmprocessor/testdata/skip_enrichment/logs_true_ecs_output.yaml index 846416598..85c803f1d 100644 --- a/processor/elasticapmprocessor/testdata/skip_enrichment/logs_true_ecs_output.yaml +++ b/processor/elasticapmprocessor/testdata/skip_enrichment/logs_true_ecs_output.yaml @@ -16,6 +16,12 @@ resourceLogs: - key: agent.name value: stringValue: otlp + - key: agent.version + value: + stringValue: unknown + - key: deployment.environment + value: + stringValue: unset scopeLogs: - logRecords: - body: diff --git a/processor/elasticapmprocessor/testdata/skip_enrichment/metrics_false_ecs_output.yaml b/processor/elasticapmprocessor/testdata/skip_enrichment/metrics_false_ecs_output.yaml index 402253edb..a4137d4de 100644 --- a/processor/elasticapmprocessor/testdata/skip_enrichment/metrics_false_ecs_output.yaml +++ b/processor/elasticapmprocessor/testdata/skip_enrichment/metrics_false_ecs_output.yaml @@ -16,6 +16,9 @@ resourceMetrics: - key: data_stream.type value: stringValue: metrics + - key: deployment.environment + value: + stringValue: unset - key: metricset.name value: stringValue: app diff --git a/processor/elasticapmprocessor/testdata/skip_enrichment/metrics_true_ecs_output.yaml b/processor/elasticapmprocessor/testdata/skip_enrichment/metrics_true_ecs_output.yaml index 402253edb..a4137d4de 100644 --- a/processor/elasticapmprocessor/testdata/skip_enrichment/metrics_true_ecs_output.yaml +++ b/processor/elasticapmprocessor/testdata/skip_enrichment/metrics_true_ecs_output.yaml @@ -16,6 +16,9 @@ resourceMetrics: - key: data_stream.type value: stringValue: metrics + - key: deployment.environment + value: + stringValue: unset - key: metricset.name value: stringValue: app diff --git a/processor/elastictraceprocessor/go.mod b/processor/elastictraceprocessor/go.mod index 55a6c1437..e75554575 100644 --- a/processor/elastictraceprocessor/go.mod +++ b/processor/elastictraceprocessor/go.mod @@ -20,7 +20,7 @@ require ( github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/elastic/apm-data v1.19.5 // indirect - github.com/elastic/opentelemetry-collector-components/internal/elasticattr v0.37.0 // indirect + github.com/elastic/opentelemetry-collector-components/internal/elasticattr v0.38.0 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-viper/mapstructure/v2 v2.5.0 // indirect diff --git a/receiver/elasticapmintakereceiver/go.mod b/receiver/elasticapmintakereceiver/go.mod index 2843ddc24..b01961e16 100644 --- a/receiver/elasticapmintakereceiver/go.mod +++ b/receiver/elasticapmintakereceiver/go.mod @@ -6,7 +6,7 @@ require ( github.com/cespare/xxhash v1.1.0 github.com/elastic/apm-data v1.19.5 github.com/elastic/go-elasticsearch/v8 v8.19.3 - github.com/elastic/opentelemetry-collector-components/internal/elasticattr v0.37.0 + github.com/elastic/opentelemetry-collector-components/internal/elasticattr v0.38.0 github.com/elastic/opentelemetry-collector-components/internal/testutil v0.0.0-20250220144628-323275205ce9 github.com/elastic/opentelemetry-lib v0.33.0 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.147.0