fix(apm): Enhance Synthetic Span Service Representation (#751)

<!--- Please remember to review the [contribution
guidelines](https://github.com/DataDog/datadog-lambda-python/blob/main/CONTRIBUTING.md)
if you have not yet done so._ --->

### What does this PR do?
<!--- A brief description of the change being made with this pull
request. --->

Rollout of span naming changes to align serverless product with tracer
to create streamlined Service Representation for Serverless

Key Changes:

- Change service name to match instance name for all managed services
(aws.lambda -> lambda name, etc) (breaking)
- Opt out via `DD_TRACE_AWS_SERVICE_REPRESENTATION_ENABLED`

- Add `span.kind:server` on synthetic spans made via span-inferrer, cold
start and lambda invocation spans

- Remove `_dd.base_service` tags on synthetic spans to avoid
unintentional service override

### Motivation

<!--- What inspired you to submit this pull request? --->

Improve Service Map for Serverless. This allows for synthetic spans to
have their own service on the map which connects with the inferred spans
from the tracer side.

Author: zarirhamza

bottlecap/src/config/env.rs

@@ -178,6 +178,11 @@ pub struct EnvConfig {
     /// <https://docs.datadoghq.com/agent/configuration/dual-shipping/?tab=helm#environment-variable-configuration-1>
     #[serde(deserialize_with = "deserialize_additional_endpoints")]
     pub apm_additional_endpoints: HashMap<String, Vec<String>>,
+    /// @env `DD_TRACE_AWS_SERVICE_REPRESENTATION_ENABLED`
+    ///
+    /// Enable the new AWS-resource naming logic in the tracer.
+    #[serde(deserialize_with = "deserialize_optional_bool_from_anything")]
+    pub trace_aws_service_representation_enabled: Option<bool>,
     //
     // Trace Propagation
     /// @env `DD_TRACE_PROPAGATION_STYLE`
@@ -362,6 +367,7 @@ fn merge_config(config: &mut Config, env_config: &EnvConfig) {
     merge_option_to_value!(config, env_config, apm_config_compression_level);
     merge_vec!(config, env_config, apm_features);
     merge_hashmap!(config, env_config, apm_additional_endpoints);
+    merge_option_to_value!(config, env_config, trace_aws_service_representation_enabled);
 
     // Trace Propagation
     merge_vec!(config, env_config, trace_propagation_style);
@@ -560,6 +566,7 @@ mod tests {
             jail.set_env("DD_TRACE_PROPAGATION_STYLE_EXTRACT", "b3");
             jail.set_env("DD_TRACE_PROPAGATION_EXTRACT_FIRST", "true");
             jail.set_env("DD_TRACE_PROPAGATION_HTTP_BAGGAGE_ENABLED", "true");
+            jail.set_env("DD_TRACE_AWS_SERVICE_REPRESENTATION_ENABLED", "true");
 
             // OTLP
             jail.set_env("DD_OTLP_CONFIG_TRACES_ENABLED", "false");
@@ -709,6 +716,7 @@ mod tests {
                 trace_propagation_style_extract: vec![TracePropagationStyle::B3],
                 trace_propagation_extract_first: true,
                 trace_propagation_http_baggage_enabled: true,
+                trace_aws_service_representation_enabled: true,
                 otlp_config_traces_enabled: false,
                 otlp_config_traces_span_name_as_resource_name: true,
                 otlp_config_traces_span_name_remappings: HashMap::from([(

bottlecap/src/config/mod.rs

@@ -283,6 +283,7 @@ pub struct Config {
     pub trace_propagation_style_extract: Vec<TracePropagationStyle>,
     pub trace_propagation_extract_first: bool,
     pub trace_propagation_http_baggage_enabled: bool,
+    pub trace_aws_service_representation_enabled: bool,
 
     // OTLP
     //
@@ -371,6 +372,7 @@ impl Default for Config {
             apm_config_compression_level: 6,
             apm_features: vec![],
             apm_additional_endpoints: HashMap::new(),
+            trace_aws_service_representation_enabled: true,
             trace_propagation_style: vec![
                 TracePropagationStyle::Datadog,
                 TracePropagationStyle::TraceContext,

bottlecap/src/config/yaml.rs

@@ -64,6 +64,8 @@ pub struct YamlConfig {
     pub apm_config: ApmConfig,
     #[serde(deserialize_with = "deserialize_service_mapping")]
     pub service_mapping: HashMap<String, String>,
+    #[serde(deserialize_with = "deserialize_optional_bool_from_anything")]
+    pub trace_aws_service_representation_enabled: Option<bool>,
     // Trace Propagation
     #[serde(deserialize_with = "deserialize_trace_propagation_style")]
     pub trace_propagation_style: Vec<TracePropagationStyle>,
@@ -451,6 +453,11 @@ fn merge_config(config: &mut Config, yaml_config: &YamlConfig) {
     merge_vec!(config, yaml_config, trace_propagation_style_extract);
     merge_option_to_value!(config, yaml_config, trace_propagation_extract_first);
     merge_option_to_value!(config, yaml_config, trace_propagation_http_baggage_enabled);
+    merge_option_to_value!(
+        config,
+        yaml_config,
+        trace_aws_service_representation_enabled
+    );
 
     // OTLP
     if let Some(otlp_config) = &yaml_config.otlp_config {
@@ -711,6 +718,7 @@ trace_propagation_style: "datadog"
 trace_propagation_style_extract: "b3"
 trace_propagation_extract_first: true
 trace_propagation_http_baggage_enabled: true
+trace_aws_service_representation_enabled: true
 
 # OTLP
 otlp_config:
@@ -839,6 +847,7 @@ extension_version: "compatibility"
                 trace_propagation_style_extract: vec![TracePropagationStyle::B3],
                 trace_propagation_extract_first: true,
                 trace_propagation_http_baggage_enabled: true,
+                trace_aws_service_representation_enabled: true,
                 otlp_config_traces_enabled: false,
                 otlp_config_traces_span_name_as_resource_name: true,
                 otlp_config_traces_span_name_remappings: HashMap::from([(

bottlecap/src/lifecycle/invocation/mod.rs

@@ -32,13 +32,19 @@ pub fn base64_to_string(base64_string: &str) -> Result<String, DecodeError> {
 }
 
 fn create_empty_span(name: String, resource: &str, service: &str) -> Span {
-    Span {
+    let mut span = Span {
         name,
         resource: resource.to_string(),
         service: service.to_string(),
         r#type: String::from("serverless"),
         ..Default::default()
-    }
+    };
+
+    // Add span.kind to the span to enable other server based features for serverless
+    span.meta
+        .insert("span.kind".to_string(), "server".to_string());
+
+    span
 }
 
 #[must_use]

bottlecap/src/lifecycle/invocation/processor.rs

@@ -39,6 +39,8 @@ use crate::{
     },
 };
 
+use crate::lifecycle::invocation::triggers::get_default_service_name;
+
 pub const MS_TO_NS: f64 = 1_000_000.0;
 pub const S_TO_MS: u64 = 1_000;
 pub const S_TO_NS: f64 = 1_000_000_000.0;
@@ -89,16 +91,23 @@ impl Processor {
         aws_config: Arc<AwsConfig>,
         metrics_aggregator: Arc<Mutex<MetricsAggregator>>,
     ) -> Self {
-        let service = config.service.clone().unwrap_or(String::from("aws.lambda"));
         let resource = tags_provider
             .get_canonical_resource_name()
             .unwrap_or(String::from("aws.lambda"));
 
+        let service = get_default_service_name(
+            &config.service.clone().unwrap_or(resource.clone()),
+            "aws.lambda",
+            config.trace_aws_service_representation_enabled,
+        );
         let propagator = DatadogCompositePropagator::new(Arc::clone(&config));
 
         Processor {
             context_buffer: ContextBuffer::default(),
-            inferrer: SpanInferrer::new(config.service_mapping.clone()),
+            inferrer: SpanInferrer::new(
+                config.service_mapping.clone(),
+                config.trace_aws_service_representation_enabled,
+            ),
             propagator,
             enhanced_metrics: EnhancedMetrics::new(metrics_aggregator, Arc::clone(&config)),
             aws_config,
@@ -195,6 +204,7 @@ impl Processor {
 
         self.dynamic_tags
             .insert(String::from("cold_start"), cold_start.to_string());
+
         if proactive_initialization {
             self.dynamic_tags.insert(
                 String::from("proactive_initialization"),
@@ -238,7 +248,6 @@ impl Processor {
         );
         cold_start_span.span_id = generate_span_id();
         cold_start_span.start = start_time;
-
         context.cold_start_span = Some(cold_start_span);
     }
 

bottlecap/src/lifecycle/invocation/span_inferrer.rs

@@ -32,6 +32,7 @@ use crate::{
 #[derive(Default)]
 pub struct SpanInferrer {
     service_mapping: HashMap<String, String>,
+    aws_service_representation_enabled: bool,
     // Span inferred from the Lambda incoming request payload
     pub inferred_span: Option<Span>,
     // Nested span inferred from the Lambda incoming request payload
@@ -50,9 +51,13 @@ pub struct SpanInferrer {
 
 impl SpanInferrer {
     #[must_use]
-    pub fn new(service_mapping: HashMap<String, String>) -> Self {
+    pub fn new(
+        service_mapping: HashMap<String, String>,
+        aws_service_representation_enabled: bool,
+    ) -> Self {
         Self {
             service_mapping,
+            aws_service_representation_enabled,
             inferred_span: None,
             wrapped_inferred_span: None,
             is_async_span: false,
@@ -87,19 +92,31 @@ impl SpanInferrer {
 
         if APIGatewayHttpEvent::is_match(payload_value) {
             if let Some(t) = APIGatewayHttpEvent::new(payload_value.clone()) {
-                t.enrich_span(&mut inferred_span, &self.service_mapping);
+                t.enrich_span(
+                    &mut inferred_span,
+                    &self.service_mapping,
+                    self.aws_service_representation_enabled,
+                );
 
                 trigger = Some(Box::new(t));
             }
         } else if APIGatewayRestEvent::is_match(payload_value) {
             if let Some(t) = APIGatewayRestEvent::new(payload_value.clone()) {
-                t.enrich_span(&mut inferred_span, &self.service_mapping);
+                t.enrich_span(
+                    &mut inferred_span,
+                    &self.service_mapping,
+                    self.aws_service_representation_enabled,
+                );
 
                 trigger = Some(Box::new(t));
             }
         } else if APIGatewayWebSocketEvent::is_match(payload_value) {
             if let Some(t) = APIGatewayWebSocketEvent::new(payload_value.clone()) {
-                t.enrich_span(&mut inferred_span, &self.service_mapping);
+                t.enrich_span(
+                    &mut inferred_span,
+                    &self.service_mapping,
+                    self.aws_service_representation_enabled,
+                );
 
                 trigger = Some(Box::new(t));
             }
@@ -110,19 +127,31 @@ impl SpanInferrer {
             }
         } else if LambdaFunctionUrlEvent::is_match(payload_value) {
             if let Some(t) = LambdaFunctionUrlEvent::new(payload_value.clone()) {
-                t.enrich_span(&mut inferred_span, &self.service_mapping);
+                t.enrich_span(
+                    &mut inferred_span,
+                    &self.service_mapping,
+                    self.aws_service_representation_enabled,
+                );
 
                 trigger = Some(Box::new(t));
             }
         } else if MSKEvent::is_match(payload_value) {
             if let Some(t) = MSKEvent::new(payload_value.clone()) {
-                t.enrich_span(&mut inferred_span, &self.service_mapping);
+                t.enrich_span(
+                    &mut inferred_span,
+                    &self.service_mapping,
+                    self.aws_service_representation_enabled,
+                );
 
                 trigger = Some(Box::new(t));
             }
         } else if SqsRecord::is_match(payload_value) {
             if let Some(t) = SqsRecord::new(payload_value.clone()) {
-                t.enrich_span(&mut inferred_span, &self.service_mapping);
+                t.enrich_span(
+                    &mut inferred_span,
+                    &self.service_mapping,
+                    self.aws_service_representation_enabled,
+                );
 
                 self.generated_span_context = extract_trace_context_from_aws_trace_header(
                     t.attributes.aws_trace_header.clone(),
@@ -140,7 +169,11 @@ impl SpanInferrer {
                         sns: sns_entity,
                         event_subscription_arn: None,
                     };
-                    wt.enrich_span(&mut wrapped_inferred_span, &self.service_mapping);
+                    wt.enrich_span(
+                        &mut wrapped_inferred_span,
+                        &self.service_mapping,
+                        self.aws_service_representation_enabled,
+                    );
                     inferred_span.meta.extend(wt.get_tags());
 
                     wrapped_inferred_span.duration =
@@ -155,8 +188,11 @@ impl SpanInferrer {
                         ..Default::default()
                     };
 
-                    event_bridge_entity
-                        .enrich_span(&mut wrapped_inferred_span, &self.service_mapping);
+                    event_bridge_entity.enrich_span(
+                        &mut wrapped_inferred_span,
+                        &self.service_mapping,
+                        self.aws_service_representation_enabled,
+                    );
                     inferred_span.meta.extend(event_bridge_entity.get_tags());
 
                     wrapped_inferred_span.duration =
@@ -169,7 +205,11 @@ impl SpanInferrer {
             }
         } else if SnsRecord::is_match(payload_value) {
             if let Some(t) = SnsRecord::new(payload_value.clone()) {
-                t.enrich_span(&mut inferred_span, &self.service_mapping);
+                t.enrich_span(
+                    &mut inferred_span,
+                    &self.service_mapping,
+                    self.aws_service_representation_enabled,
+                );
 
                 if let Some(message) = &t.sns.message {
                     if let Ok(event_bridge_wrapper_message) =
@@ -180,8 +220,11 @@ impl SpanInferrer {
                             ..Default::default()
                         };
 
-                        event_bridge_wrapper_message
-                            .enrich_span(&mut wrapped_inferred_span, &self.service_mapping);
+                        event_bridge_wrapper_message.enrich_span(
+                            &mut wrapped_inferred_span,
+                            &self.service_mapping,
+                            self.aws_service_representation_enabled,
+                        );
                         inferred_span
                             .meta
                             .extend(event_bridge_wrapper_message.get_tags());
@@ -197,27 +240,43 @@ impl SpanInferrer {
             }
         } else if DynamoDbRecord::is_match(payload_value) {
             if let Some(t) = DynamoDbRecord::new(payload_value.clone()) {
-                t.enrich_span(&mut inferred_span, &self.service_mapping);
+                t.enrich_span(
+                    &mut inferred_span,
+                    &self.service_mapping,
+                    self.aws_service_representation_enabled,
+                );
                 self.span_pointers = t.get_span_pointers();
 
                 trigger = Some(Box::new(t));
             }
         } else if S3Record::is_match(payload_value) {
             if let Some(t) = S3Record::new(payload_value.clone()) {
-                t.enrich_span(&mut inferred_span, &self.service_mapping);
+                t.enrich_span(
+                    &mut inferred_span,
+                    &self.service_mapping,
+                    self.aws_service_representation_enabled,
+                );
                 self.span_pointers = t.get_span_pointers();
 
                 trigger = Some(Box::new(t));
             }
         } else if EventBridgeEvent::is_match(payload_value) {
             if let Some(t) = EventBridgeEvent::new(payload_value.clone()) {
-                t.enrich_span(&mut inferred_span, &self.service_mapping);
+                t.enrich_span(
+                    &mut inferred_span,
+                    &self.service_mapping,
+                    self.aws_service_representation_enabled,
+                );
 
                 trigger = Some(Box::new(t));
             }
         } else if KinesisRecord::is_match(payload_value) {
             if let Some(t) = KinesisRecord::new(payload_value.clone()) {
-                t.enrich_span(&mut inferred_span, &self.service_mapping);
+                t.enrich_span(
+                    &mut inferred_span,
+                    &self.service_mapping,
+                    self.aws_service_representation_enabled,
+                );
 
                 trigger = Some(Box::new(t));
             }
@@ -282,6 +341,7 @@ impl SpanInferrer {
                 String::from("peer.service"),
                 invocation_span.service.clone(),
             );
+            s.meta.insert("span.kind".to_string(), "server".to_string());
 
             if let Some(ws) = &mut self.wrapped_inferred_span {
                 ws.trace_id = invocation_span.trace_id;