diff --git a/bottlecap/Cargo.lock b/bottlecap/Cargo.lock index 7e010800c..c87abf658 100644 --- a/bottlecap/Cargo.lock +++ b/bottlecap/Cargo.lock @@ -485,6 +485,7 @@ dependencies = [ "chrono", "cookie", "datadog-agent-config", + "datadog-agent-trace-sampler", "datadog-fips", "datadog-opentelemetry", "datadog-protos", @@ -809,7 +810,7 @@ dependencies = [ [[package]] name = "datadog-agent-config" version = "0.1.0" -source = "git+https://github.com/DataDog/serverless-components?rev=d0c7f44191445e20d309734675d5e8b91d2a5d51#d0c7f44191445e20d309734675d5e8b91d2a5d51" +source = "git+https://github.com/DataDog/serverless-components?rev=54e570ae60d169acfd48800b1805a9bbdbbeafb2#54e570ae60d169acfd48800b1805a9bbdbbeafb2" dependencies = [ "datadog-opentelemetry", "dogstatsd", @@ -824,10 +825,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "datadog-agent-trace-sampler" +version = "0.1.0" +source = "git+https://github.com/DataDog/serverless-components?rev=54e570ae60d169acfd48800b1805a9bbdbbeafb2#54e570ae60d169acfd48800b1805a9bbdbbeafb2" + [[package]] name = "datadog-fips" version = "0.1.0" -source = "git+https://github.com/DataDog/serverless-components?rev=d0c7f44191445e20d309734675d5e8b91d2a5d51#d0c7f44191445e20d309734675d5e8b91d2a5d51" +source = "git+https://github.com/DataDog/serverless-components?rev=54e570ae60d169acfd48800b1805a9bbdbbeafb2#54e570ae60d169acfd48800b1805a9bbdbbeafb2" dependencies = [ "reqwest", "rustls", @@ -976,7 +982,7 @@ dependencies = [ [[package]] name = "dogstatsd" version = "0.1.0" -source = "git+https://github.com/DataDog/serverless-components?rev=d0c7f44191445e20d309734675d5e8b91d2a5d51#d0c7f44191445e20d309734675d5e8b91d2a5d51" +source = "git+https://github.com/DataDog/serverless-components?rev=54e570ae60d169acfd48800b1805a9bbdbbeafb2#54e570ae60d169acfd48800b1805a9bbdbbeafb2" dependencies = [ "datadog-protos", "ddsketch-agent", diff --git a/bottlecap/Cargo.toml b/bottlecap/Cargo.toml index 563545feb..9623a68ad 100644 --- a/bottlecap/Cargo.toml +++ b/bottlecap/Cargo.toml @@ -82,9 +82,10 @@ libdd-trace-normalization = { git = "https://github.com/DataDog/libdatadog", rev libdd-trace-obfuscation = { git = "https://github.com/DataDog/libdatadog", rev = "85ce322a1dcb1eda7df9bcc021223b2d1a236783", default-features = false } libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog", rev = "85ce322a1dcb1eda7df9bcc021223b2d1a236783", default-features = false } datadog-opentelemetry = { git = "https://github.com/DataDog/dd-trace-rs", rev = "50bfea8755b75e448a80ac04d53fa7edd414eefe", default-features = false, features = ["_unstable_propagation"] } -dogstatsd = { git = "https://github.com/DataDog/serverless-components", rev = "d0c7f44191445e20d309734675d5e8b91d2a5d51", default-features = false } -datadog-fips = { git = "https://github.com/DataDog/serverless-components", rev = "d0c7f44191445e20d309734675d5e8b91d2a5d51", default-features = false } -datadog-agent-config = { git = "https://github.com/DataDog/serverless-components", rev = "d0c7f44191445e20d309734675d5e8b91d2a5d51", default-features = false } +dogstatsd = { git = "https://github.com/DataDog/serverless-components", rev = "54e570ae60d169acfd48800b1805a9bbdbbeafb2", default-features = false } +datadog-fips = { git = "https://github.com/DataDog/serverless-components", rev = "54e570ae60d169acfd48800b1805a9bbdbbeafb2", default-features = false } +datadog-agent-config = { git = "https://github.com/DataDog/serverless-components", rev = "54e570ae60d169acfd48800b1805a9bbdbbeafb2", default-features = false } +datadog-agent-trace-sampler = { git = "https://github.com/DataDog/serverless-components", rev = "54e570ae60d169acfd48800b1805a9bbdbbeafb2", default-features = false } libddwaf = { version = "1.28.1", git = "https://github.com/DataDog/libddwaf-rust", rev = "d1534a158d976bd4f747bf9fcc58e0712d2d17fc", default-features = false, features = ["serde"] } [dev-dependencies] diff --git a/bottlecap/LICENSE-3rdparty.csv b/bottlecap/LICENSE-3rdparty.csv index 9da9c4542..bddc9f975 100644 --- a/bottlecap/LICENSE-3rdparty.csv +++ b/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 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. datadog-protos,https://github.com/DataDog/saluki,Apache-2.0,The datadog-protos Authors diff --git a/bottlecap/src/bin/bottlecap/main.rs b/bottlecap/src/bin/bottlecap/main.rs index 39f352149..8f76fba5d 100644 --- a/bottlecap/src/bin/bottlecap/main.rs +++ b/bottlecap/src/bin/bottlecap/main.rs @@ -1162,6 +1162,7 @@ fn start_trace_agent( let trace_processor = Arc::new(trace_processor::ServerlessTraceProcessor { obfuscation_config: Arc::new(obfuscation_config), + error_sampler: trace_processor::new_error_sampler(config.ext.apm_error_sampler_enabled), }); let (span_dedup_service, span_dedup_handle) = span_dedup_service::DedupService::new(); diff --git a/bottlecap/src/config/mod.rs b/bottlecap/src/config/mod.rs index db0a1f192..648eff8da 100644 --- a/bottlecap/src/config/mod.rs +++ b/bottlecap/src/config/mod.rs @@ -66,6 +66,13 @@ pub struct LambdaConfig { pub capture_lambda_payload: bool, pub capture_lambda_payload_max_depth: u32, pub lambda_extension_compute_stats: bool, + + /// `DD_APM_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: + /// enabled rescues every errored chunk. See APMSVLS-469. + pub apm_error_sampler_enabled: bool, + pub span_dedup_timeout: Option, pub api_key_secret_reload_interval: Option, pub serverless_appsec_enabled: bool, @@ -95,6 +102,7 @@ impl Default for LambdaConfig { capture_lambda_payload: false, capture_lambda_payload_max_depth: 10, lambda_extension_compute_stats: false, + apm_error_sampler_enabled: false, span_dedup_timeout: None, api_key_secret_reload_interval: None, serverless_appsec_enabled: false, @@ -153,6 +161,9 @@ pub struct LambdaConfigSource { #[serde(deserialize_with = "deser_opt_bool")] pub lambda_extension_compute_stats: Option, + #[serde(deserialize_with = "deser_opt_bool")] + pub apm_error_sampler_enabled: Option, + #[serde(deserialize_with = "deser_dur_secs_ignore_zero")] pub span_dedup_timeout: Option, #[serde(deserialize_with = "deser_dur_secs_ignore_zero")] @@ -199,6 +210,7 @@ impl DatadogConfigExtension for LambdaConfig { capture_lambda_payload, capture_lambda_payload_max_depth, lambda_extension_compute_stats, + apm_error_sampler_enabled, serverless_appsec_enabled, appsec_waf_timeout, api_security_enabled, @@ -536,6 +548,32 @@ mod lambda_config_tests { assert!(!config.ext.lambda_extension_compute_stats); } + // ---- error sampler (apm_error_sampler_enabled) ---- + + #[test] + fn apm_error_sampler_enabled_defaults_to_false() { + let config = load(|_| Ok(())); + assert!(!config.ext.apm_error_sampler_enabled); + } + + #[test] + fn apm_error_sampler_enabled_from_env() { + let config = load(|jail| { + jail.set_env("DD_APM_ERROR_SAMPLER_ENABLED", "true"); + Ok(()) + }); + assert!(config.ext.apm_error_sampler_enabled); + } + + #[test] + fn apm_error_sampler_enabled_from_yaml() { + let config = load(|jail| { + jail.create_file("datadog.yaml", "apm_error_sampler_enabled: true\n")?; + Ok(()) + }); + assert!(config.ext.apm_error_sampler_enabled); + } + // ---- Duration fields ---- #[test] diff --git a/bottlecap/src/lifecycle/invocation/processor.rs b/bottlecap/src/lifecycle/invocation/processor.rs index 307484b56..c878d3810 100644 --- a/bottlecap/src/lifecycle/invocation/processor.rs +++ b/bottlecap/src/lifecycle/invocation/processor.rs @@ -1918,6 +1918,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::default_error_sampler(), }), trace_tx: tokio::sync::mpsc::channel(1).0, stats_generator: Arc::new(StatsGenerator::new(stats_concentrator_handle)), @@ -2028,6 +2029,7 @@ mod tests { obfuscation_config: Arc::new( ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), ), + error_sampler: trace_processor::default_error_sampler(), }), trace_tx: tokio::sync::mpsc::channel(1).0, stats_generator: Arc::new(StatsGenerator::new(stats_concentrator_handle)), @@ -2667,6 +2669,7 @@ mod tests { obfuscation_config: Arc::new( ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), ), + error_sampler: trace_processor::default_error_sampler(), }), trace_tx: tokio::sync::mpsc::channel(1).0, stats_generator: Arc::new(StatsGenerator::new(stats_concentrator_handle)), @@ -3176,6 +3179,7 @@ mod tests { obfuscation_config: Arc::new( ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), ), + error_sampler: trace_processor::default_error_sampler(), }), trace_tx, stats_generator: Arc::new(StatsGenerator::new(stats_concentrator_handle)), diff --git a/bottlecap/src/traces/trace_processor.rs b/bottlecap/src/traces/trace_processor.rs index f16eee7bc..e8a59b43b 100644 --- a/bottlecap/src/traces/trace_processor.rs +++ b/bottlecap/src/traces/trace_processor.rs @@ -15,6 +15,9 @@ use crate::traces::{ LAMBDA_RUNTIME_URL_PREFIX, LAMBDA_STATSD_URL_PREFIX, }; use async_trait::async_trait; +use datadog_agent_trace_sampler::{ + ErrorSamplerConfig, ErrorSamplerMode, ErrorsSampler, SampleDecision, SpanView, TraceView, +}; use libdd_common::Endpoint; use libdd_trace_obfuscation::obfuscate::obfuscate_span; use libdd_trace_obfuscation::obfuscation_config; @@ -62,6 +65,141 @@ impl StatsComputedBy { #[allow(clippy::module_name_repetitions)] pub struct ServerlessTraceProcessor { pub obfuscation_config: Arc, + /// Rescues errored `AutoDrop` chunks on the `lambda_extension_compute_stats` + /// path. Shared across invocations so `RateLimited`'s rolling-window rate + /// limiter accumulates; a std Mutex rather than tokio because + /// `process_traces` is synchronous. + pub error_sampler: Arc>, +} + +/// Borrow-only view of a span for the error sampler. +fn span_view(span: &Span) -> SpanView<'_> { + SpanView { + service: &span.service, + name: &span.name, + resource: &span.resource, + error: span.error != 0, + http_status_code: span.meta.get("http.status_code").map(String::as_str), + error_type: span.meta.get("error.type").map(String::as_str), + } +} + +/// Builds the error sampler as the extension ships it, enabled or disabled by +/// `apm_error_sampler_enabled`. +/// +/// The mode is hardcoded to `AlwaysKeep`: Lambda's per-invocation trace volume +/// is low, and freeze/thaw breaks `RateLimited`'s 30s wall-clock window. +#[must_use] +pub fn new_error_sampler(enabled: bool) -> Arc> { + Arc::new(std::sync::Mutex::new(ErrorsSampler::new( + ErrorSamplerConfig { + mode: ErrorSamplerMode::AlwaysKeep, + // `is_disabled()` is `target_tps <= 0.0`; the magnitude only + // matters in RateLimited mode. + target_tps: if enabled { 1.0 } else { 0.0 }, + extra_sample_rate: 1.0, + }, + ))) +} + +impl ServerlessTraceProcessor { + /// Removes sampled-out chunks so they won't be sent to Datadog, then drops + /// any payload left without chunks. `SamplerPriority::None` (-128) means no + /// explicit priority was set and the trace is kept. Only + /// `SamplerPriority::AutoDrop` (0) chunks are rescue candidates; negative + /// priorities are explicit drops and are honored. + fn drop_sampled_out_chunks(&self, tracer_payloads: &mut Vec) { + // Hoisted out of the loop: one lock instead of one per chunk. + let rescue_enabled = !self + .error_sampler + .lock() + .expect("error sampler poisoned") + .is_disabled(); + // Only RateLimited's rolling window reads the clock, and only rescue + // candidates reach it. + let now_secs: i64 = if rescue_enabled { + std::time::UNIX_EPOCH + .elapsed() + .unwrap_or_default() + .as_secs() + .try_into() + .unwrap_or_default() + } else { + 0 + }; + for tp in tracer_payloads.iter_mut() { + // The sampler keys its per-signature rate limits on the env the + // tracer reported for this payload, as the Agent does. + let env = tp.env.as_str(); + tp.chunks.retain_mut(|chunk| { + if chunk.priority > 0 || chunk.priority == SamplerPriority::None as i32 { + return true; + } + // A negative priority is an explicit drop (a tracer sampling rule + // or MANUAL_DROP): honor it and never rescue, as the Agent does. + if chunk.priority < 0 { + return false; + } + if !rescue_enabled { + return false; + } + self.rescue_error_chunk(chunk, env, now_secs) + }); + } + tracer_payloads.retain(|tp| !tp.chunks.is_empty()); + } + + /// Consult the error sampler for a chunk that would otherwise be dropped + /// (`AutoDrop`). Returns `true` to keep (rescue) the chunk. Only errored + /// traces are candidates; on a keep, stamps `_dd.errors_sr` on the root span. + fn rescue_error_chunk(&self, chunk: &mut pb::TraceChunk, env: &str, now_secs: i64) -> bool { + let Ok(root_idx) = trace_utils::get_root_span_index(&chunk.spans) else { + return false; // no identifiable root span + }; + // Only errored traces are rescue candidates (matches the Go agent, which + // only routes error traces through the ErrorTPS ScoreSampler). An error + // anywhere in the chunk counts, not just on the root span. + if !chunk.spans.iter().any(|s| s.error != 0) { + return false; + } + + // Scoped so the views release their borrow of chunk.spans before the + // `_dd.errors_sr` mutation below. + let decision = { + let mut sampler = self.error_sampler.lock().expect("error sampler poisoned"); + let root = &chunk.spans[root_idx]; + // AlwaysKeep only bounds-checks root_index, so building the root + // view alone is enough. RateLimited needs every span to compute + // the trace signature. + let root_view; + let all_views; + let (spans, root_index) = if matches!(*sampler, ErrorsSampler::AlwaysKeep { .. }) { + root_view = [span_view(root)]; + (&root_view[..], 0) + } else { + all_views = chunk.spans.iter().map(span_view).collect::>(); + (&all_views[..], root_idx) + }; + let trace = TraceView { + env, + trace_id: root.trace_id, + root_index, + root_global_sample_rate: root.metrics.get("_sample_rate").copied().unwrap_or(1.0), + spans, + }; + sampler.sample(now_secs, &trace) + }; + + match decision { + SampleDecision::Keep { errors_sr } => { + if let Some(root) = chunk.spans.get_mut(root_idx) { + root.metrics.insert("_dd.errors_sr".to_string(), errors_sr); + } + true + } + SampleDecision::Drop => false, + } + } } struct ChunkProcessor { @@ -437,20 +575,12 @@ impl TraceProcessor for ServerlessTraceProcessor { } }; - // Remove sampled-out chunks so they won't be sent to Datadog. - // Sampled-out chunks are preserved in payloads_for_stats above so their - // stats are still counted. SamplerPriority::None (-128) means no explicit priority - // was set and the trace is kept; drop priorities are SamplerPriority::AutoDrop (0) - // and UserDrop (-1, not represented in SamplerPriority). + // Sampled-out chunks are preserved in payloads_for_stats above, so their + // stats are still counted after they are removed here. if config.ext.lambda_extension_compute_stats && let TracerPayloadCollection::V07(ref mut tracer_payloads) = payload { - for tp in tracer_payloads.iter_mut() { - tp.chunks.retain(|chunk| { - chunk.priority > 0 || chunk.priority == SamplerPriority::None as i32 - }); - } - tracer_payloads.retain(|tp| !tp.chunks.is_empty()); + self.drop_sampled_out_chunks(tracer_payloads); if tracer_payloads.is_empty() { return (None, payloads_for_stats); } @@ -598,6 +728,12 @@ impl SendingTraceProcessor { } } +/// Enabled error sampler for constructing `ServerlessTraceProcessor` in tests. +#[cfg(test)] +pub(crate) fn default_error_sampler() -> Arc> { + new_error_sampler(true) +} + #[cfg(test)] mod tests { use std::{ @@ -723,6 +859,7 @@ mod tests { obfuscation_config: Arc::new( ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), ), + error_sampler: default_error_sampler(), }; let config = create_test_config(); let tags_provider = create_tags_provider(config.clone()); @@ -1201,6 +1338,7 @@ mod tests { obfuscation_config: Arc::new( ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), ), + error_sampler: default_error_sampler(), }; let header_tags = tracer_header_tags::TracerHeaderTags { @@ -1272,6 +1410,254 @@ mod tests { ); } + /// On the compute-stats path, the error sampler rescues errored chunks that + /// would otherwise be dropped (`AutoDrop`), stamping `_dd.errors_sr`, while + /// non-errored P0 chunks and explicit user drops are still dropped. + #[test] + fn test_error_sampler_rescues_errored_p0_chunks() { + use libdd_trace_obfuscation::obfuscation_config::ObfuscationConfig; + + let config = Arc::new(Config { + apm_dd_url: "https://trace.agent.datadoghq.com".to_string(), + ext: crate::config::LambdaConfig { + lambda_extension_compute_stats: true, + ..Default::default() + }, + ..Config::default() + }); + let tags_provider = Arc::new(Provider::new( + config.clone(), + "lambda".to_string(), + &std::collections::HashMap::from([( + "function_arn".to_string(), + "test-arn".to_string(), + )]), + )); + let processor = ServerlessTraceProcessor { + obfuscation_config: Arc::new( + ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), + ), + error_sampler: default_error_sampler(), + }; + + let header_tags = tracer_header_tags::TracerHeaderTags { + lang: "rust", + lang_version: "1.0", + lang_interpreter: "", + lang_vendor: "", + tracer_version: "1.0", + container_id: "", + client_computed_top_level: false, + client_computed_stats: false, + dropped_p0_traces: 0, + dropped_p0_spans: 0, + }; + + let make_span = |trace_id: u64, priority: f64, error: i32| -> pb::Span { + let mut metrics = HashMap::new(); + metrics.insert("_sampling_priority_v1".to_string(), priority); + pb::Span { + trace_id, + span_id: trace_id, + parent_id: 0, + error, + metrics, + service: "svc".to_string(), + name: "op".to_string(), + resource: "res".to_string(), + ..Default::default() + } + }; + + // trace 1: kept normally (priority 1). trace 2: errored P0 (rescued). + // trace 3: non-errored P0 (dropped). trace 4: errored user drop (dropped). + let traces = vec![ + vec![make_span(1, 1.0, 0)], + vec![make_span(2, 0.0, 1)], + vec![make_span(3, 0.0, 0)], + vec![make_span(4, -1.0, 1)], + ]; + + let (payload_info, _stats) = + processor.process_traces(config, tags_provider, header_tags, traces, 0, None); + let payload_info = payload_info.expect("expected Some payload"); + let backend_send_data = payload_info.builder.build(); + let TracerPayloadCollection::V07(backend_payloads) = backend_send_data.get_payloads() + else { + panic!("expected V07"); + }; + + let kept: Vec = backend_payloads + .iter() + .flat_map(|tp| tp.chunks.iter()) + .flat_map(|c| c.spans.iter()) + .map(|s| s.trace_id) + .collect(); + assert_eq!(kept.len(), 2, "kept normal trace + rescued errored trace"); + assert!(kept.contains(&1), "priority-1 trace kept"); + assert!(kept.contains(&2), "errored P0 trace rescued"); + assert!(!kept.contains(&3), "non-errored P0 trace dropped"); + assert!(!kept.contains(&4), "errored user-drop trace not rescued"); + + let rescued_root = backend_payloads + .iter() + .flat_map(|tp| tp.chunks.iter()) + .flat_map(|c| c.spans.iter()) + .find(|s| s.trace_id == 2) + .expect("rescued trace present"); + assert!( + rescued_root.metrics.contains_key("_dd.errors_sr"), + "_dd.errors_sr stamped on rescued root" + ); + } + + /// An error on a child span (root not errored) still makes the chunk a rescue + /// candidate, matching the Go agent's `traceContainsError`. + #[test] + fn test_error_sampler_rescues_chunk_with_errored_child_span() { + use libdd_trace_obfuscation::obfuscation_config::ObfuscationConfig; + + let config = Arc::new(Config { + apm_dd_url: "https://trace.agent.datadoghq.com".to_string(), + ext: crate::config::LambdaConfig { + lambda_extension_compute_stats: true, + ..Default::default() + }, + ..Config::default() + }); + let tags_provider = Arc::new(Provider::new( + config.clone(), + "lambda".to_string(), + &std::collections::HashMap::from([( + "function_arn".to_string(), + "test-arn".to_string(), + )]), + )); + let processor = ServerlessTraceProcessor { + obfuscation_config: Arc::new( + ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), + ), + error_sampler: default_error_sampler(), + }; + + let header_tags = tracer_header_tags::TracerHeaderTags { + lang: "rust", + lang_version: "1.0", + lang_interpreter: "", + lang_vendor: "", + tracer_version: "1.0", + container_id: "", + client_computed_top_level: false, + client_computed_stats: false, + dropped_p0_traces: 0, + dropped_p0_spans: 0, + }; + + let make_span = |span_id: u64, parent_id: u64, error: i32| -> pb::Span { + let mut metrics = HashMap::new(); + metrics.insert("_sampling_priority_v1".to_string(), 0.0); + pb::Span { + trace_id: 1, + span_id, + parent_id, + error, + metrics, + service: "svc".to_string(), + name: "op".to_string(), + resource: "res".to_string(), + ..Default::default() + } + }; + + // P0 trace whose root is fine but whose child failed (e.g. a caught + // downstream call): still an error trace, so it must be rescued. + let traces = vec![vec![make_span(1, 0, 0), make_span(2, 1, 1)]]; + + let (payload_info, _stats) = + processor.process_traces(config, tags_provider, header_tags, traces, 0, None); + let payload_info = payload_info.expect("errored-child P0 trace rescued"); + let backend_send_data = payload_info.builder.build(); + let TracerPayloadCollection::V07(backend_payloads) = backend_send_data.get_payloads() + else { + panic!("expected V07"); + }; + + let root = backend_payloads + .iter() + .flat_map(|tp| tp.chunks.iter()) + .flat_map(|c| c.spans.iter()) + .find(|s| s.span_id == 1) + .expect("rescued trace present"); + assert!( + root.metrics.contains_key("_dd.errors_sr"), + "_dd.errors_sr stamped on rescued root" + ); + } + + /// With the error sampler disabled (the shipping default), errored P0 + /// chunks are dropped: no rescue, no `_dd.errors_sr`. + #[test] + fn test_disabled_error_sampler_drops_errored_p0_chunks() { + use libdd_trace_obfuscation::obfuscation_config::ObfuscationConfig; + + let config = Arc::new(Config { + apm_dd_url: "https://trace.agent.datadoghq.com".to_string(), + ext: crate::config::LambdaConfig { + lambda_extension_compute_stats: true, + ..Default::default() + }, + ..Config::default() + }); + let tags_provider = Arc::new(Provider::new( + config.clone(), + "lambda".to_string(), + &std::collections::HashMap::from([( + "function_arn".to_string(), + "test-arn".to_string(), + )]), + )); + let processor = ServerlessTraceProcessor { + obfuscation_config: Arc::new( + ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), + ), + error_sampler: new_error_sampler(false), + }; + + let header_tags = tracer_header_tags::TracerHeaderTags { + lang: "rust", + lang_version: "1.0", + lang_interpreter: "", + lang_vendor: "", + tracer_version: "1.0", + container_id: "", + client_computed_top_level: false, + client_computed_stats: false, + dropped_p0_traces: 0, + dropped_p0_spans: 0, + }; + + let mut metrics = HashMap::new(); + metrics.insert("_sampling_priority_v1".to_string(), 0.0); + let traces = vec![vec![pb::Span { + trace_id: 1, + span_id: 1, + parent_id: 0, + error: 1, + metrics, + service: "svc".to_string(), + name: "op".to_string(), + resource: "res".to_string(), + ..Default::default() + }]]; + + let (payload_info, _stats) = + processor.process_traces(config, tags_provider, header_tags, traces, 0, None); + assert!( + payload_info.is_none(), + "errored P0 trace must stay dropped when the error sampler is disabled" + ); + } + /// Verifies that `process_traces` returns `None` for the backend payload when all /// traces are sampled out and `lambda_extension_compute_stats` is true. #[test] @@ -1298,6 +1684,7 @@ mod tests { obfuscation_config: Arc::new( ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), ), + error_sampler: default_error_sampler(), }; let header_tags = tracer_header_tags::TracerHeaderTags { lang: "rust", @@ -1379,6 +1766,7 @@ mod tests { obfuscation_config: Arc::new( ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), ), + error_sampler: default_error_sampler(), }; let header_tags = tracer_header_tags::TracerHeaderTags { lang: "rust", @@ -1487,6 +1875,7 @@ mod tests { obfuscation_config: Arc::new( ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), ), + error_sampler: default_error_sampler(), }; let header_tags = tracer_header_tags::TracerHeaderTags { lang: "rust", @@ -1858,6 +2247,7 @@ mod tests { obfuscation_config: Arc::new( ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), ), + error_sampler: default_error_sampler(), }), trace_tx, stats_generator: Arc::new(StatsGenerator::new(concentrator_handle.clone())), diff --git a/bottlecap/tests/apm_integration_test.rs b/bottlecap/tests/apm_integration_test.rs index cfb89f244..63f06ef29 100644 --- a/bottlecap/tests/apm_integration_test.rs +++ b/bottlecap/tests/apm_integration_test.rs @@ -30,7 +30,9 @@ use bottlecap::traces::stats_generator::StatsGenerator; use bottlecap::traces::trace_aggregator::SendDataBuilderInfo; use bottlecap::traces::trace_aggregator_service::AggregatorService; use bottlecap::traces::trace_flusher::TraceFlusher; -use bottlecap::traces::trace_processor::{SendingTraceProcessor, ServerlessTraceProcessor}; +use bottlecap::traces::trace_processor::{ + SendingTraceProcessor, ServerlessTraceProcessor, new_error_sampler, +}; use dogstatsd::api_key::ApiKeyFactory; use libdd_common::Endpoint; use libdd_trace_obfuscation::obfuscation_config::ObfuscationConfig; @@ -339,6 +341,7 @@ async fn run_processor_pipeline_with_traces( obfuscation_config: Arc::new( ObfuscationConfig::new().expect("Failed to create ObfuscationConfig"), ), + error_sampler: new_error_sampler(true), }), trace_tx, stats_generator: Arc::new(StatsGenerator::new(concentrator_handle.clone())),