[APMSVLS-469] feat(traces): add error sampler to keep/rescue traces with errors

bottlecap/Cargo.toml

@@ -95,6 +95,7 @@ datadog-opentelemetry = { git = "https://github.com/DataDog/dd-trace-rs", rev =
 dogstatsd = { git = "https://github.com/DataDog/serverless-components", rev = "9daae40afa87f52fad1489f4d7cfd4a579037d2d", default-features = false }
 datadog-fips = { git = "https://github.com/DataDog/serverless-components", rev = "9daae40afa87f52fad1489f4d7cfd4a579037d2d", default-features = false }
 datadog-agent-config = { git = "https://github.com/DataDog/serverless-components", rev = "9daae40afa87f52fad1489f4d7cfd4a579037d2d", default-features = false }
+datadog-agent-trace-sampler = { git = "https://github.com/DataDog/serverless-components", rev = "9daae40afa87f52fad1489f4d7cfd4a579037d2d", default-features = false }
 libddwaf = { version = "1.28.1", git = "https://github.com/DataDog/libddwaf-rust", rev = "d1534a158d976bd4f747bf9fcc58e0712d2d17fc", default-features = false, features = ["serde"] }
 
 [dev-dependencies]

bottlecap/LICENSE-3rdparty.csv

@@ -44,6 +44,7 @@ crossbeam-utils,https://github.com/crossbeam-rs/crossbeam,MIT OR Apache-2.0,The
 crypto-common,https://github.com/RustCrypto/traits,MIT OR Apache-2.0,RustCrypto Developers
 ctor,https://github.com/mmastrac/rust-ctor,Apache-2.0 OR MIT,Matt Mastracci <matthew@mastracci.com>
 datadog-agent-config,https://github.com/DataDog/serverless-components,Apache-2.0,The datadog-agent-config Authors
+datadog-agent-trace-sampler,https://github.com/DataDog/serverless-components,Apache-2.0,The datadog-agent-trace-sampler Authors
 datadog-fips,https://github.com/DataDog/serverless-components,Apache-2.0,The datadog-fips Authors
 datadog-opentelemetry,https://github.com/DataDog/dd-trace-rs/tree/main/datadog-opentelemetry,Apache-2.0,Datadog Inc. <info@datadoghq.com>
 datadog-protos,https://github.com/DataDog/saluki,Apache-2.0,The datadog-protos Authors

bottlecap/src/bin/bottlecap/main.rs

@@ -1223,8 +1223,19 @@ fn start_trace_agent(
         ..Default::default()
     };
 
+    // The Agent's error sampler knob has no effect here: the extension's
+    // sampler is a plain on/off switch, not a TPS budget.
+    if env::var("DD_APM_ERROR_TPS").is_ok_and(|v| !v.trim().is_empty()) {
+        warn!(
+            "DD_APM_ERROR_TPS is not supported by the Lambda extension; error trace rescue is an on/off switch controlled by DD_SERVERLESS_ERROR_SAMPLER_ENABLED"
+        );
+    }
+
     let trace_processor = Arc::new(trace_processor::ServerlessTraceProcessor {
         obfuscation_config: Arc::new(obfuscation_config),
+        error_sampler: trace_processor::new_error_sampler(
+            config.ext.serverless_error_sampler_enabled,
+        ),
     });
 
     let (span_dedup_service, span_dedup_handle) = span_dedup_service::DedupService::new();

bottlecap/src/config/mod.rs

@@ -68,6 +68,16 @@ pub struct LambdaConfig {
     pub capture_lambda_payload: bool,
     pub capture_lambda_payload_max_depth: u32,
     pub lambda_extension_compute_stats: bool,
+
+    /// `DD_SERVERLESS_ERROR_SAMPLER_ENABLED`: rescue errored trace chunks that would
+    /// otherwise be dropped, on the `lambda_extension_compute_stats` path. The
+    /// sampler runs in `AlwaysKeep` mode, so this is a plain on/off switch with
+    /// no volume ceiling: enabled rescues every errored chunk, whatever the
+    /// tracer's sampling rate. A function that errors on most invocations under
+    /// `DD_TRACE_SAMPLE_RATE=0.01` will ingest close to every trace, not 1%.
+    /// See APMSVLS-469.
+    pub serverless_error_sampler_enabled: bool,
+
     pub span_dedup_timeout: Option<Duration>,
     pub api_key_secret_reload_interval: Option<Duration>,
     pub serverless_appsec_enabled: bool,
@@ -134,6 +144,7 @@ impl Default for LambdaConfig {
             capture_lambda_payload: false,
             capture_lambda_payload_max_depth: 10,
             lambda_extension_compute_stats: false,
+            serverless_error_sampler_enabled: false,
             span_dedup_timeout: None,
             api_key_secret_reload_interval: None,
             serverless_appsec_enabled: false,
@@ -199,6 +210,9 @@ pub struct LambdaConfigSource {
     #[serde(deserialize_with = "deser_opt_bool")]
     pub lambda_extension_compute_stats: Option<bool>,
 
+    #[serde(deserialize_with = "deser_opt_bool")]
+    pub serverless_error_sampler_enabled: Option<bool>,
+
     #[serde(deserialize_with = "deser_dur_secs_ignore_zero")]
     pub span_dedup_timeout: Option<Duration>,
     #[serde(deserialize_with = "deser_dur_secs_ignore_zero")]
@@ -280,6 +294,7 @@ impl DatadogConfigExtension for LambdaConfig {
                 capture_lambda_payload,
                 capture_lambda_payload_max_depth,
                 lambda_extension_compute_stats,
+                serverless_error_sampler_enabled,
                 serverless_appsec_enabled,
                 appsec_waf_timeout,
                 api_security_enabled,
@@ -700,6 +715,32 @@ mod lambda_config_tests {
         assert!(!config.ext.dsm_consume_enabled);
     }
 
+    // ---- error sampler (serverless_error_sampler_enabled) ----
+
+    #[test]
+    fn serverless_error_sampler_enabled_defaults_to_false() {
+        let config = load(|_| Ok(()));
+        assert!(!config.ext.serverless_error_sampler_enabled);
+    }
+
+    #[test]
+    fn serverless_error_sampler_enabled_from_env() {
+        let config = load(|jail| {
+            jail.set_env("DD_SERVERLESS_ERROR_SAMPLER_ENABLED", "true");
+            Ok(())
+        });
+        assert!(config.ext.serverless_error_sampler_enabled);
+    }
+
+    #[test]
+    fn serverless_error_sampler_enabled_from_yaml() {
+        let config = load(|jail| {
+            jail.create_file("datadog.yaml", "serverless_error_sampler_enabled: true\n")?;
+            Ok(())
+        });
+        assert!(config.ext.serverless_error_sampler_enabled);
+    }
+
     // ---- Duration fields ----
 
     #[test]

bottlecap/src/lifecycle/invocation/processor.rs

@@ -2293,6 +2293,7 @@ mod tests {
                         appsec: None,
                         processor: Arc::new(trace_processor::ServerlessTraceProcessor {
                             obfuscation_config: Arc::new(ObfuscationConfig::new().expect("Failed to create ObfuscationConfig")),
+                            error_sampler: trace_processor::enabled_error_sampler(),
                         }),
                         trace_tx: tokio::sync::mpsc::channel(1).0,
                         stats_generator: Arc::new(StatsGenerator::new(stats_concentrator_handle)),
@@ -2403,6 +2404,7 @@ mod tests {
                 obfuscation_config: Arc::new(
                     ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"),
                 ),
+                error_sampler: trace_processor::enabled_error_sampler(),
             }),
             trace_tx: tokio::sync::mpsc::channel(1).0,
             stats_generator: Arc::new(StatsGenerator::new(stats_concentrator_handle)),
@@ -3041,6 +3043,7 @@ mod tests {
                 obfuscation_config: Arc::new(
                     ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"),
                 ),
+                error_sampler: trace_processor::enabled_error_sampler(),
             }),
             trace_tx: tokio::sync::mpsc::channel(1).0,
             stats_generator: Arc::new(StatsGenerator::new(stats_concentrator_handle)),
@@ -3622,6 +3625,7 @@ mod tests {
                 obfuscation_config: Arc::new(
                     ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"),
                 ),
+                error_sampler: trace_processor::enabled_error_sampler(),
             }),
             trace_tx,
             stats_generator: Arc::new(StatsGenerator::new(stats_concentrator_handle)),