feat(traces): use DD_SERVICE for inferred spans when integration service names are removed

bottlecap/src/config/mod.rs

@@ -75,6 +75,12 @@ pub struct LambdaConfig {
     pub api_security_sample_delay: Duration,
     pub custom_metrics_exclude_tags: Vec<String>,
 
+    /// When true, inferred (synthetic) event-source spans report the function's
+    /// base service (`DD_SERVICE`) instead of the AWS resource/instance
+    /// representation. An explicit `DD_SERVICE_MAPPING` entry still wins.
+    /// Defaults to `false`.
+    pub trace_remove_integration_service_names_enabled: bool,
+
     /// Maximum number of request IDs whose logs are held in `held_logs` waiting for durable
     /// execution context. Set to 0 to disable log holding; logs will be flushed immediately
     /// without durable execution context enrichment. Defaults to 0 until the tracer-side
@@ -120,6 +126,7 @@ impl Default for LambdaConfig {
             api_security_enabled: true,
             api_security_sample_delay: Duration::from_secs(30),
             custom_metrics_exclude_tags: Vec::new(),
+            trace_remove_integration_service_names_enabled: false,
             lambda_durable_function_log_buffer_size: 0,
             dsm_consume_enabled: false,
             dsm_exchange_name: None,
@@ -195,6 +202,12 @@ pub struct LambdaConfigSource {
     #[serde(deserialize_with = "deser_csv")]
     pub lambda_customer_metrics_exclude_tags: Vec<String>,
 
+    /// `DD_TRACE_REMOVE_INTEGRATION_SERVICE_NAMES_ENABLED` — when true, inferred
+    /// (synthetic) event-source spans use `DD_SERVICE` rather than the AWS
+    /// resource/instance name. Defaults to `false`.
+    #[serde(deserialize_with = "deser_opt_bool")]
+    pub trace_remove_integration_service_names_enabled: Option<bool>,
+
     /// `DD_LAMBDA_DURABLE_FUNCTION_LOG_BUFFER_SIZE` — max number of request IDs
     /// whose logs are held waiting for durable execution context. Defaults to
     /// 0 (hold mechanism disabled).
@@ -235,6 +248,7 @@ impl DatadogConfigExtension for LambdaConfig {
                 appsec_waf_timeout,
                 api_security_enabled,
                 api_security_sample_delay,
+                trace_remove_integration_service_names_enabled,
                 lambda_durable_function_log_buffer_size,
             ],
             option: [span_dedup_timeout, api_key_secret_reload_interval, appsec_rules, dsm_exchange_name, dsm_kafka_group],
@@ -577,6 +591,33 @@ mod lambda_config_tests {
         assert!(!config.ext.lambda_extension_compute_stats);
     }
 
+    #[test]
+    fn trace_remove_integration_service_names_defaults_false() {
+        let config = load(|_| Ok(()));
+        assert!(!config.ext.trace_remove_integration_service_names_enabled);
+    }
+
+    #[test]
+    fn trace_remove_integration_service_names_from_env() {
+        let config = load(|jail| {
+            jail.set_env("DD_TRACE_REMOVE_INTEGRATION_SERVICE_NAMES_ENABLED", "true");
+            Ok(())
+        });
+        assert!(config.ext.trace_remove_integration_service_names_enabled);
+    }
+
+    #[test]
+    fn trace_remove_integration_service_names_from_yaml() {
+        let config = load(|jail| {
+            jail.create_file(
+                "datadog.yaml",
+                "trace_remove_integration_service_names_enabled: true\n",
+            )?;
+            Ok(())
+        });
+        assert!(config.ext.trace_remove_integration_service_names_enabled);
+    }
+
     #[test]
     fn dsm_consume_enabled_from_data_streams_env() {
         let config = load(|jail| {

bottlecap/src/lifecycle/invocation/span_inferrer.rs

@@ -23,6 +23,38 @@ use crate::{
 };
 use datadog_opentelemetry::propagation::context::SpanContext;
 
+/// Point an inferred (synthetic) span at the function's base service when
+/// `DD_TRACE_REMOVE_INTEGRATION_SERVICE_NAMES_ENABLED` is set and `DD_SERVICE`
+/// is configured, instead of the AWS resource/instance representation the
+/// trigger resolved. This gives a single setting that consolidates every
+/// event-source span onto the function's service, rather than requiring one
+/// `DD_SERVICE_MAPPING` entry per trigger type.
+///
+/// An explicit `DD_SERVICE_MAPPING` entry still wins, preserving the precedence
+/// in [`Trigger::resolve_service_name`].
+///
+/// The value is lowercased so both spans land on one service. That matches the
+/// invocation span directly when AWS service representation is enabled, since
+/// `processor.rs` lowercases `DD_SERVICE` there too. When it is disabled the
+/// invocation span is initially named `aws.lambda`, and `ChunkProcessor::process`
+/// rewrites any `aws.lambda` span to the lowercased `DD_SERVICE` from the tags
+/// map, so the two still converge. Both paths are pinned by tests.
+fn apply_base_service_override(span: &mut Span, trigger: &dyn Trigger, config: &Config) {
+    if !config.ext.trace_remove_integration_service_names_enabled {
+        return;
+    }
+
+    let Some(service) = config.service.as_deref() else {
+        return;
+    };
+
+    if service.is_empty() || trigger.has_service_mapping_entry(&config.service_mapping) {
+        return;
+    }
+
+    span.service = service.to_lowercase();
+}
+
 /// Per-invocation inference output produced by [`SpanInferrer::infer_span`].
 ///
 /// This lives on each invocation's `Context` (not on the shared `Processor`) so
@@ -111,6 +143,7 @@ impl SpanInferrer {
         )
     }
 
+    #[allow(clippy::too_many_lines)]
     fn get_wrapped_inferred_span(
         identified_trigger: &IdentifiedTrigger,
         inferred_span: &mut Span,
@@ -135,6 +168,11 @@ impl SpanInferrer {
                         &config.service_mapping,
                         config.trace_aws_service_representation_enabled,
                     );
+                    apply_base_service_override(
+                        &mut wrapped_inferred_span,
+                        &wrapped_trigger,
+                        config,
+                    );
                     inferred_span.meta.extend(wrapped_trigger.get_tags());
 
                     wrapped_inferred_span.duration =
@@ -161,6 +199,11 @@ impl SpanInferrer {
                         &config.service_mapping,
                         config.trace_aws_service_representation_enabled,
                     );
+                    apply_base_service_override(
+                        &mut wrapped_inferred_span,
+                        &event_bridge_entity,
+                        config,
+                    );
                     inferred_span.meta.extend(event_bridge_entity.get_tags());
 
                     wrapped_inferred_span.duration =
@@ -193,6 +236,11 @@ impl SpanInferrer {
                         &config.service_mapping,
                         config.trace_aws_service_representation_enabled,
                     );
+                    apply_base_service_override(
+                        &mut wrapped_inferred_span,
+                        &event_bridge_wrapper_message,
+                        config,
+                    );
                     inferred_span
                         .meta
                         .extend(event_bridge_wrapper_message.get_tags());
@@ -270,6 +318,7 @@ impl SpanInferrer {
                     &self.config.service_mapping,
                     self.config.trace_aws_service_representation_enabled,
                 );
+                apply_base_service_override(&mut inferred_span, t.as_ref(), &self.config);
             }
 
             if let Some(dd_resource_key) = t.get_dd_resource_key(&aws_config.region) {
@@ -431,6 +480,7 @@ pub fn extract_generated_span_context(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::config::LambdaConfig;
     use crate::lifecycle::invocation::triggers::test_utils::read_json_file;
     use crate::traces::propagation::DatadogCompositePropagator;
     use datadog_opentelemetry::propagation::TracePropagationStyle;
@@ -756,4 +806,149 @@ mod tests {
             "AppSec JSON should not be added when invocation span has none"
         );
     }
+
+    fn sqs_payload() -> Value {
+        let json = read_json_file("sqs_event.json");
+        serde_json::from_str(&json).expect("Failed to deserialize SQS payload")
+    }
+
+    /// Infer a span from `payload` and return the resolved inferred-span service.
+    fn inferred_service(payload: &Value, config: Config) -> String {
+        let inferrer = SpanInferrer::new(Arc::new(config));
+        inferrer
+            .infer_span(payload, &aws_config("us-east-1"))
+            .inferred_span
+            .expect("Should have inferred a span")
+            .service
+    }
+
+    #[test]
+    fn test_base_service_override_uses_dd_service() {
+        let config = Config {
+            service: Some("my-lambda-service".to_string()),
+            ext: LambdaConfig {
+                trace_remove_integration_service_names_enabled: true,
+                ..LambdaConfig::default()
+            },
+            ..Config::default()
+        };
+
+        assert_eq!(
+            inferred_service(&sqs_payload(), config),
+            "my-lambda-service"
+        );
+    }
+
+    #[test]
+    fn test_base_service_override_disabled_by_default() {
+        let config = Config {
+            service: Some("my-lambda-service".to_string()),
+            ..Config::default()
+        };
+
+        // Default behavior is unchanged: the AWS resource name is preserved.
+        assert_eq!(inferred_service(&sqs_payload(), config), "MyQueue");
+    }
+
+    #[test]
+    fn test_base_service_override_yields_to_service_mapping() {
+        let config = Config {
+            service: Some("my-lambda-service".to_string()),
+            service_mapping: HashMap::from([(
+                "lambda_sqs".to_string(),
+                "remapped-queue".to_string(),
+            )]),
+            ext: LambdaConfig {
+                trace_remove_integration_service_names_enabled: true,
+                ..LambdaConfig::default()
+            },
+            ..Config::default()
+        };
+
+        assert_eq!(inferred_service(&sqs_payload(), config), "remapped-queue");
+    }
+
+    #[test]
+    fn test_base_service_override_noop_without_dd_service() {
+        let config = Config {
+            service: None,
+            ext: LambdaConfig {
+                trace_remove_integration_service_names_enabled: true,
+                ..LambdaConfig::default()
+            },
+            ..Config::default()
+        };
+
+        assert_eq!(inferred_service(&sqs_payload(), config), "MyQueue");
+    }
+
+    #[test]
+    fn test_base_service_override_lowercases_dd_service() {
+        // The invocation span in processor.rs lowercases DD_SERVICE, so the
+        // inferred span must too or the two land on different services.
+        let config = Config {
+            service: Some("MyLambdaService".to_string()),
+            ext: LambdaConfig {
+                trace_remove_integration_service_names_enabled: true,
+                ..LambdaConfig::default()
+            },
+            ..Config::default()
+        };
+
+        assert_eq!(inferred_service(&sqs_payload(), config), "mylambdaservice");
+    }
+
+    /// With AWS service representation disabled the trigger would resolve to the
+    /// generic fallback (`sqs`). The override still applies, and the invocation
+    /// span converges on the same value via `ChunkProcessor::process` — see
+    /// `test_invocation_span_normalized_to_dd_service_when_representation_disabled`.
+    #[test]
+    fn test_base_service_override_applies_when_representation_disabled() {
+        let config = Config {
+            service: Some("my-lambda-service".to_string()),
+            trace_aws_service_representation_enabled: false,
+            ext: LambdaConfig {
+                trace_remove_integration_service_names_enabled: true,
+                ..LambdaConfig::default()
+            },
+            ..Config::default()
+        };
+
+        assert_eq!(
+            inferred_service(&sqs_payload(), config),
+            "my-lambda-service"
+        );
+    }
+
+    #[test]
+    fn test_base_service_override_applies_to_wrapped_span() {
+        let json = read_json_file("sns_sqs_event.json");
+        let payload: Value =
+            serde_json::from_str(&json).expect("Failed to deserialize SNS-in-SQS payload");
+
+        let config = Arc::new(Config {
+            service: Some("my-lambda-service".to_string()),
+            ext: LambdaConfig {
+                trace_remove_integration_service_names_enabled: true,
+                ..LambdaConfig::default()
+            },
+            ..Config::default()
+        });
+
+        let inferrer = SpanInferrer::new(config);
+        let data = inferrer.infer_span(&payload, &aws_config("us-east-1"));
+
+        assert_eq!(
+            data.inferred_span
+                .expect("Should have inferred an SQS span")
+                .service,
+            "my-lambda-service"
+        );
+        assert_eq!(
+            data.wrapped_inferred_span
+                .expect("Should have inferred a wrapped SNS span")
+                .service,
+            "my-lambda-service"
+        );
+    }
 }

bottlecap/src/lifecycle/invocation/triggers/mod.rs

@@ -205,6 +205,15 @@ pub trait Trigger: ServiceNameResolver {
         None
     }
 
+    /// Whether an explicit `DD_SERVICE_MAPPING` entry targets this trigger,
+    /// under either its specific or its generic key. Callers that override the
+    /// resolved service name use this to preserve the precedence established by
+    /// [`Trigger::resolve_service_name`]: an explicit mapping always wins.
+    fn has_service_mapping_entry(&self, service_mapping: &HashMap<String, String>) -> bool {
+        service_mapping.contains_key(&self.get_specific_identifier())
+            || service_mapping.contains_key(self.get_generic_identifier())
+    }
+
     /// Default implementation for service name resolution
     fn resolve_service_name(
         &self,

bottlecap/src/traces/trace_processor.rs

@@ -1674,6 +1674,45 @@ mod tests {
         );
     }
 
+    /// `processor.rs` builds the invocation span with
+    /// `get_default_service_name(.., "aws.lambda", representation_enabled)`, so when
+    /// AWS service representation is disabled the invocation span starts out named
+    /// `aws.lambda` even though `DD_SERVICE` is set. This normalization step is what
+    /// puts it back on `DD_SERVICE`, and it is the reason the inferred-span base
+    /// service override in `span_inferrer.rs` stays consistent with the invocation
+    /// span in that configuration rather than diverging from it.
+    #[test]
+    fn test_invocation_span_normalized_to_dd_service_when_representation_disabled() {
+        let config = Arc::new(Config {
+            service: Some("My-Payments-API".to_string()),
+            trace_aws_service_representation_enabled: false,
+            ..Config::default()
+        });
+        let mut processor = create_chunk_processor(config);
+
+        let invocation_span = pb::Span {
+            name: "aws.lambda".to_string(),
+            service: "aws.lambda".to_string(),
+            resource: "my-function".to_string(),
+            ..create_inferred_span()
+        };
+        let mut chunk = pb::TraceChunk {
+            priority: 1,
+            origin: "lambda".to_string(),
+            spans: vec![invocation_span],
+            tags: HashMap::new(),
+            dropped_trace: false,
+        };
+
+        processor.process(&mut chunk, 0);
+
+        assert_eq!(
+            chunk.spans[0].service, "my-payments-api",
+            "invocation span should be normalized to the lowercased DD_SERVICE, \
+             matching what apply_base_service_override puts on inferred spans"
+        );
+    }
+
     #[test]
     fn test_base_service_not_set_on_non_inferred_spans() {
         let config = Arc::new(Config {