diff --git a/bottlecap/src/config/mod.rs b/bottlecap/src/config/mod.rs index d1a9ed258..a291eaddf 100644 --- a/bottlecap/src/config/mod.rs +++ b/bottlecap/src/config/mod.rs @@ -51,6 +51,12 @@ use datadog_agent_config::{ flush_strategy::FlushStrategy as UpstreamFlushStrategy, }; +/// Default error trace rescue budget in errored traces per second. +/// +/// Matches the shared sampler's default (`ErrorSamplerConfig::default().target_tps`) +/// and the Go agent's `ErrorTPS`. +pub const DEFAULT_APM_ERROR_TPS: f64 = 10.0; + /// Lambda-specific configuration that lives alongside the shared /// `datadog_agent_config::Config` core fields under `config.ext` once the /// migration onto upstream lands. @@ -69,14 +75,23 @@ pub struct LambdaConfig { 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, whenever stats are computed before the backend, by - /// either the tracer or the extension. 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, + /// Deprecated raw value of `DD_SERVERLESS_ERROR_SAMPLER_ENABLED`, kept so + /// the legacy on/off switch still resolves. Read only through + /// [`LambdaConfig::error_sampler_target_tps`] and + /// [`LambdaConfig::legacy_error_sampler_warning`]; set `DD_APM_ERROR_TPS` + /// instead. + pub serverless_error_sampler_enabled: Option, + + /// `DD_APM_ERROR_TPS`: approximate error trace rescue budget per execution + /// environment, in traces per second, shared across trace signatures. + /// `None` uses [`DEFAULT_APM_ERROR_TPS`]; `<= 0` disables rescue. Rescue + /// runs independently of where stats are computed; rescued traces carry + /// `_dd.errors_sr` for backend retention. + /// + /// Uses 5s buckets over a 30s window, with no hard ceiling. Freeze/thaw + /// expires counters but preserves throttled rates, which recover by at most + /// 20% per recomputation. The lowest stored rate also caps new signatures. + pub apm_error_tps: Option, pub span_dedup_timeout: Option, pub api_key_secret_reload_interval: Option, @@ -144,7 +159,8 @@ impl Default for LambdaConfig { capture_lambda_payload: false, capture_lambda_payload_max_depth: 10, lambda_extension_compute_stats: false, - serverless_error_sampler_enabled: false, + serverless_error_sampler_enabled: None, + apm_error_tps: None, span_dedup_timeout: None, api_key_secret_reload_interval: None, serverless_appsec_enabled: false, @@ -165,6 +181,24 @@ impl Default for LambdaConfig { } } +/// Deserialize an optional `f64`, discarding non-finite values. +/// +/// `figment` accepts non-finite floats from YAML (`.inf`, `.nan`) and from +/// dotted exponent strings (`1.0e999`), so filter them out here. +fn deserialize_optional_finite_f64<'de, D>(deserializer: D) -> Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + let value = deser_opt_lossless::<_, f64>(deserializer)?; + match value { + Some(v) if !v.is_finite() => { + tracing::warn!("Ignoring non-finite DD_APM_ERROR_TPS (apm_error_tps) value {v}"); + Ok(None) + } + other => Ok(other), + } +} + /// Intermediate deserialization type shared by env-var and YAML loading. /// /// `#[serde(default)]` and the forgiving per-field deserializers are required @@ -213,6 +247,11 @@ pub struct LambdaConfigSource { #[serde(deserialize_with = "deser_opt_bool")] pub serverless_error_sampler_enabled: Option, + /// `DD_APM_ERROR_TPS`: error trace rescue budget in errored traces per + /// second. + #[serde(deserialize_with = "deserialize_optional_finite_f64")] + pub apm_error_tps: Option, + #[serde(deserialize_with = "deser_dur_secs_ignore_zero")] pub span_dedup_timeout: Option, #[serde(deserialize_with = "deser_dur_secs_ignore_zero")] @@ -294,7 +333,6 @@ 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, @@ -303,7 +341,7 @@ impl DatadogConfigExtension for LambdaConfig { lambda_durable_function_log_buffer_size, trace_experimental_features_enabled, ], - option: [span_dedup_timeout, api_key_secret_reload_interval, appsec_rules, dsm_exchange_name, dsm_kafka_group], + option: [span_dedup_timeout, api_key_secret_reload_interval, appsec_rules, serverless_error_sampler_enabled, apm_error_tps, dsm_exchange_name, dsm_kafka_group], ); // data_streams_enabled (source / DD_DATA_STREAMS_ENABLED) → @@ -353,6 +391,38 @@ impl DatadogConfigExtension for LambdaConfig { } impl LambdaConfig { + /// Effective error trace rescue budget in errored traces per second. + /// + /// `DD_APM_ERROR_TPS` wins when set. Otherwise the deprecated + /// `DD_SERVERLESS_ERROR_SAMPLER_ENABLED=false` is a kill switch, and + /// everything else falls back to [`DEFAULT_APM_ERROR_TPS`]. + #[must_use] + pub fn error_sampler_target_tps(&self) -> f64 { + match (self.apm_error_tps, self.serverless_error_sampler_enabled) { + (Some(tps), _) => tps, + (None, Some(false)) => 0.0, + _ => DEFAULT_APM_ERROR_TPS, + } + } + + /// Deprecation warning for `DD_SERVERLESS_ERROR_SAMPLER_ENABLED`, or `None` + /// when it is unset. + #[must_use] + pub fn legacy_error_sampler_warning(&self) -> Option { + match (self.apm_error_tps, self.serverless_error_sampler_enabled) { + (_, None) => None, + (Some(tps), Some(enabled)) => Some(format!( + "DD_SERVERLESS_ERROR_SAMPLER_ENABLED (serverless_error_sampler_enabled) is deprecated; its value {enabled} is ignored because DD_APM_ERROR_TPS (apm_error_tps) is set to {tps}" + )), + (None, Some(true)) => Some(format!( + "DD_SERVERLESS_ERROR_SAMPLER_ENABLED (serverless_error_sampler_enabled) is deprecated and has no effect: error trace rescue uses a default budget of {DEFAULT_APM_ERROR_TPS} traces per second; use DD_APM_ERROR_TPS (apm_error_tps) to change the budget" + )), + (None, Some(false)) => Some(String::from( + "DD_SERVERLESS_ERROR_SAMPLER_ENABLED (serverless_error_sampler_enabled) is deprecated; set DD_APM_ERROR_TPS (apm_error_tps) to 0 to disable error trace rescue instead", + )), + } + } + /// Drop `additional_metric_tags` / `additional_metric_tags_cardinality_limit` unless /// `trace_experimental_features_enabled` is set, matching the Serverless Compatibility /// Layer (`datadog-trace-agent`). @@ -715,30 +785,346 @@ 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); + // ---- error trace rescue budget (DD_APM_ERROR_TPS) ---- + + const TPS: &str = "DD_APM_ERROR_TPS"; + const LEGACY: &str = "DD_SERVERLESS_ERROR_SAMPLER_ENABLED"; + + /// One row in the error-sampler resolution table. + struct ErrorSamplerCase { + name: &'static str, + yaml: Option<&'static str>, + env: &'static [(&'static str, &'static str)], + target: f64, + /// Substrings the legacy-flag warning must contain; `None` expects no + /// warning. + warning_contains: Option<&'static [&'static str]>, + /// Whether the raw `apm_error_tps` is unset (missing or rejected). + tps_is_none: bool, } #[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); + #[allow(clippy::too_many_lines)] + #[allow(clippy::float_cmp)] + fn error_sampler_target_tps_resolution_table() { + let cases = [ + ErrorSamplerCase { + name: "default", + yaml: None, + env: &[], + target: DEFAULT_APM_ERROR_TPS, + warning_contains: None, + tps_is_none: true, + }, + ErrorSamplerCase { + name: "env int", + yaml: None, + env: &[(TPS, "5")], + target: 5.0, + warning_contains: None, + tps_is_none: false, + }, + ErrorSamplerCase { + name: "env float", + yaml: None, + env: &[(TPS, "2.5")], + target: 2.5, + warning_contains: None, + tps_is_none: false, + }, + ErrorSamplerCase { + name: "env padded", + yaml: None, + env: &[(TPS, " 7 ")], + target: 7.0, + warning_contains: None, + tps_is_none: false, + }, + ErrorSamplerCase { + name: "yaml", + yaml: Some("apm_error_tps: 3\n"), + env: &[], + target: 3.0, + warning_contains: None, + tps_is_none: false, + }, + ErrorSamplerCase { + name: "env over yaml", + yaml: Some("apm_error_tps: 3\n"), + env: &[(TPS, "4")], + target: 4.0, + warning_contains: None, + tps_is_none: false, + }, + ErrorSamplerCase { + name: "zero", + yaml: None, + env: &[(TPS, "0")], + target: 0.0, + warning_contains: None, + tps_is_none: false, + }, + ErrorSamplerCase { + name: "negative", + yaml: None, + env: &[(TPS, "-1")], + target: -1.0, + warning_contains: None, + tps_is_none: false, + }, + ErrorSamplerCase { + name: "large", + yaml: None, + env: &[(TPS, "1000000000")], + target: 1_000_000_000.0, + warning_contains: None, + tps_is_none: false, + }, + ErrorSamplerCase { + name: "exponent without dot", + yaml: None, + env: &[(TPS, "1e3")], + target: DEFAULT_APM_ERROR_TPS, + warning_contains: None, + tps_is_none: true, + }, + ErrorSamplerCase { + name: "exponent with dot", + yaml: None, + env: &[(TPS, "1.0e3")], + target: 1000.0, + warning_contains: None, + tps_is_none: false, + }, + ErrorSamplerCase { + name: "invalid env: empty", + yaml: None, + env: &[(TPS, "")], + target: DEFAULT_APM_ERROR_TPS, + warning_contains: None, + tps_is_none: true, + }, + ErrorSamplerCase { + name: "invalid env: text", + yaml: None, + env: &[(TPS, "abc")], + target: DEFAULT_APM_ERROR_TPS, + warning_contains: None, + tps_is_none: true, + }, + ErrorSamplerCase { + name: "invalid env: inf", + yaml: None, + env: &[(TPS, "inf")], + target: DEFAULT_APM_ERROR_TPS, + warning_contains: None, + tps_is_none: true, + }, + ErrorSamplerCase { + name: "invalid env: nan", + yaml: None, + env: &[(TPS, "nan")], + target: DEFAULT_APM_ERROR_TPS, + warning_contains: None, + tps_is_none: true, + }, + ErrorSamplerCase { + name: "invalid env: overflow literal", + yaml: None, + env: &[(TPS, "1e400")], + target: DEFAULT_APM_ERROR_TPS, + warning_contains: None, + tps_is_none: true, + }, + ErrorSamplerCase { + name: "non-finite float", + yaml: None, + env: &[(TPS, "1.0e999")], + target: DEFAULT_APM_ERROR_TPS, + warning_contains: None, + tps_is_none: true, + }, + ErrorSamplerCase { + name: "invalid yaml: inf", + yaml: Some("apm_error_tps: .inf\n"), + env: &[], + target: DEFAULT_APM_ERROR_TPS, + warning_contains: None, + tps_is_none: true, + }, + ErrorSamplerCase { + name: "invalid yaml: nan", + yaml: Some("apm_error_tps: .nan\n"), + env: &[], + target: DEFAULT_APM_ERROR_TPS, + warning_contains: None, + tps_is_none: true, + }, + // `lambda_extension_compute_stats` does not participate in this + // resolution; both sources and both values are still exercised + // across these two cases. + ErrorSamplerCase { + name: "legacy env true, compute stats false", + yaml: None, + env: &[ + (LEGACY, "true"), + ("DD_LAMBDA_EXTENSION_COMPUTE_STATS", "false"), + ], + target: DEFAULT_APM_ERROR_TPS, + warning_contains: Some(&["deprecated"]), + tps_is_none: true, + }, + ErrorSamplerCase { + name: "legacy yaml true, compute stats true", + yaml: Some( + "serverless_error_sampler_enabled: true\nlambda_extension_compute_stats: true\n", + ), + env: &[], + target: DEFAULT_APM_ERROR_TPS, + warning_contains: Some(&["deprecated"]), + tps_is_none: true, + }, + ErrorSamplerCase { + name: "legacy false", + yaml: None, + env: &[(LEGACY, "false")], + target: 0.0, + warning_contains: Some(&["set DD_APM_ERROR_TPS (apm_error_tps) to 0"]), + tps_is_none: true, + }, + ErrorSamplerCase { + name: "legacy yaml false", + yaml: Some("serverless_error_sampler_enabled: false\n"), + env: &[], + target: 0.0, + warning_contains: Some(&[ + "DD_SERVERLESS_ERROR_SAMPLER_ENABLED (serverless_error_sampler_enabled)", + "set DD_APM_ERROR_TPS (apm_error_tps) to 0", + ]), + tps_is_none: true, + }, + ErrorSamplerCase { + name: "both, tps 5 + legacy false", + yaml: None, + env: &[(TPS, "5"), (LEGACY, "false")], + target: 5.0, + warning_contains: Some(&[ + "DD_APM_ERROR_TPS (apm_error_tps) is set to 5", + "its value false is ignored", + ]), + tps_is_none: false, + }, + ErrorSamplerCase { + name: "both, tps 0 + legacy true", + yaml: None, + env: &[(TPS, "0"), (LEGACY, "true")], + target: 0.0, + warning_contains: Some(&[ + "DD_APM_ERROR_TPS (apm_error_tps) is set to 0", + "its value true is ignored", + ]), + tps_is_none: false, + }, + ErrorSamplerCase { + name: "cross-source", + yaml: Some("apm_error_tps: 5\n"), + env: &[(LEGACY, "false")], + target: 5.0, + warning_contains: Some(&[ + "DD_APM_ERROR_TPS (apm_error_tps) is set to 5", + "its value false is ignored", + ]), + tps_is_none: false, + }, + ErrorSamplerCase { + name: "invalid tps + legacy false", + yaml: None, + env: &[(TPS, "abc"), (LEGACY, "false")], + target: 0.0, + warning_contains: Some(&["set DD_APM_ERROR_TPS (apm_error_tps) to 0"]), + tps_is_none: true, + }, + ErrorSamplerCase { + name: "invalid env over valid yaml + legacy false", + yaml: Some("apm_error_tps: 5\n"), + env: &[(TPS, "abc"), (LEGACY, "false")], + target: 5.0, + warning_contains: Some(&[ + "DD_APM_ERROR_TPS (apm_error_tps) is set to 5", + "its value false is ignored", + ]), + tps_is_none: false, + }, + ]; + + for case in cases { + let name = case.name; + let config = load(|jail| { + if let Some(yaml) = case.yaml { + jail.create_file("datadog.yaml", yaml)?; + } + for (key, value) in case.env { + jail.set_env(key, value); + } + Ok(()) + }); + + assert_eq!( + config.ext.error_sampler_target_tps(), + case.target, + "case {name:?}: unexpected effective target TPS" + ); + assert_eq!( + config.ext.apm_error_tps.is_none(), + case.tps_is_none, + "case {name:?}: unexpected raw apm_error_tps {:?}", + config.ext.apm_error_tps + ); + + match ( + case.warning_contains, + config.ext.legacy_error_sampler_warning(), + ) { + (None, None) => {} + (Some(needles), Some(message)) => { + assert!( + !message.contains("removed"), + "case {name:?}: deprecation warning must not promise removal: {message}" + ); + for needle in needles { + assert!( + message.contains(needle), + "case {name:?}: warning {message:?} must contain {needle:?}" + ); + } + if config.ext.serverless_error_sampler_enabled == Some(true) + && config.ext.apm_error_tps.is_none() + { + assert_eq!( + message, + format!( + "DD_SERVERLESS_ERROR_SAMPLER_ENABLED (serverless_error_sampler_enabled) is deprecated and has no effect: error trace rescue uses a default budget of {DEFAULT_APM_ERROR_TPS} traces per second; use DD_APM_ERROR_TPS (apm_error_tps) to change the budget" + ), + "case {name:?}: warning must describe the default rescue budget" + ); + } + } + (expected, actual) => { + panic!( + "case {name:?}: expected warning containing {expected:?}, got {actual:?}" + ) + } + } + } } #[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); + #[allow(clippy::float_cmp)] + fn default_apm_error_tps_matches_shared_sampler_default() { + assert_eq!( + DEFAULT_APM_ERROR_TPS, + datadog_agent_trace_sampler::ErrorSamplerConfig::default().target_tps + ); } // ---- APM features (DD_APM_FEATURES, error_rare_sample_tracer_drop) ---- @@ -751,6 +1137,7 @@ mod lambda_config_tests { } #[test] + #[allow(clippy::float_cmp)] fn apm_features_error_rare_sample_tracer_drop_from_env() { let config = load(|jail| { jail.set_env("DD_APM_FEATURES", "error_rare_sample_tracer_drop"); @@ -758,9 +1145,9 @@ mod lambda_config_tests { }); assert_eq!(config.apm_features, ["error_rare_sample_tracer_drop"]); assert!(crate::traces::trace_processor::error_rare_sample_tracer_drop_enabled(&config)); - // The feature only widens rescue eligibility; enabling it must not - // enable the sampler. - assert!(!config.ext.serverless_error_sampler_enabled); + // The feature only widens rescue eligibility; it must not change the + // rescue budget. + assert_eq!(config.ext.error_sampler_target_tps(), DEFAULT_APM_ERROR_TPS); } #[test] @@ -779,6 +1166,22 @@ mod lambda_config_tests { assert!(crate::traces::trace_processor::error_rare_sample_tracer_drop_enabled(&config)); } + #[test] + fn apm_features_space_separated_list_from_env() { + let config = load(|jail| { + jail.set_env( + "DD_APM_FEATURES", + "some_other_feature error_rare_sample_tracer_drop", + ); + Ok(()) + }); + assert_eq!( + config.apm_features, + ["some_other_feature", "error_rare_sample_tracer_drop"] + ); + assert!(crate::traces::trace_processor::error_rare_sample_tracer_drop_enabled(&config)); + } + #[test] fn apm_features_unrelated_feature_does_not_enable_error_rare_sample_tracer_drop() { let config = load(|jail| { diff --git a/bottlecap/src/lifecycle/invocation/processor.rs b/bottlecap/src/lifecycle/invocation/processor.rs index 26a9fe417..89714d457 100644 --- a/bottlecap/src/lifecycle/invocation/processor.rs +++ b/bottlecap/src/lifecycle/invocation/processor.rs @@ -2293,7 +2293,7 @@ mod tests { appsec: None, processor: Arc::new(trace_processor::ServerlessTraceProcessor { obfuscation_config: Arc::new(ObfuscationConfig::default()), - error_sampler: trace_processor::enabled_error_sampler(), + error_sampler: trace_processor::default_error_sampler(), error_rare_sample_tracer_drop: false, }), trace_tx: tokio::sync::mpsc::channel(1).0, @@ -2403,7 +2403,7 @@ mod tests { appsec: None, processor: Arc::new(trace_processor::ServerlessTraceProcessor { obfuscation_config: Arc::new(ObfuscationConfig::default()), - error_sampler: trace_processor::enabled_error_sampler(), + error_sampler: trace_processor::default_error_sampler(), error_rare_sample_tracer_drop: false, }), trace_tx: tokio::sync::mpsc::channel(1).0, @@ -3041,7 +3041,7 @@ mod tests { appsec: None, processor: Arc::new(trace_processor::ServerlessTraceProcessor { obfuscation_config: Arc::new(ObfuscationConfig::default()), - error_sampler: trace_processor::enabled_error_sampler(), + error_sampler: trace_processor::default_error_sampler(), error_rare_sample_tracer_drop: false, }), trace_tx: tokio::sync::mpsc::channel(1).0, @@ -3622,7 +3622,7 @@ mod tests { appsec: None, processor: Arc::new(trace_processor::ServerlessTraceProcessor { obfuscation_config: Arc::new(ObfuscationConfig::default()), - error_sampler: trace_processor::enabled_error_sampler(), + error_sampler: trace_processor::default_error_sampler(), error_rare_sample_tracer_drop: false, }), trace_tx, diff --git a/bottlecap/src/lifecycle/invocation/processor_service.rs b/bottlecap/src/lifecycle/invocation/processor_service.rs index a77e888bd..6f592b45e 100644 --- a/bottlecap/src/lifecycle/invocation/processor_service.rs +++ b/bottlecap/src/lifecycle/invocation/processor_service.rs @@ -831,7 +831,7 @@ mod tests { use std::collections::HashMap; use tokio::time::{Duration, timeout}; - let error_sampler = trace_processor::new_error_sampler(false); + let error_sampler = trace_processor::new_error_sampler(0.0); let timeout_dur = Duration::from_millis(500); diff --git a/bottlecap/src/startup.rs b/bottlecap/src/startup.rs index cf12b0939..425613393 100644 --- a/bottlecap/src/startup.rs +++ b/bottlecap/src/startup.rs @@ -12,7 +12,7 @@ //! together trace, stats, proxy, lifecycle, tags, appsec, and flushing //! pieces: it is orchestration, not a trace-domain API. -use std::{env, sync::Arc}; +use std::sync::Arc; use dogstatsd::api_key::ApiKeyFactory; use libdd_common::regex_engine::{Regex, Replacer}; @@ -157,29 +157,24 @@ pub fn build_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" - ); + if let Some(message) = config.ext.legacy_error_sampler_warning() { + warn!("{message}"); } + let error_tps = config.ext.error_sampler_target_tps(); // One-time startup warning: the feature only widens rescue eligibility, it - // does not enable the sampler. + // does not change the rescue budget. let error_rare_sample_tracer_drop = trace_processor::error_rare_sample_tracer_drop_enabled(config); - if error_rare_sample_tracer_drop && !config.ext.serverless_error_sampler_enabled { + if error_rare_sample_tracer_drop && error_tps <= 0.0 { warn!( - "DD_APM_FEATURES includes error_rare_sample_tracer_drop, but DD_SERVERLESS_ERROR_SAMPLER_ENABLED is false: errored traces dropped by the tracer will not be rescued" + "DD_APM_FEATURES includes error_rare_sample_tracer_drop, but error trace rescue is disabled (effective DD_APM_ERROR_TPS {error_tps}): errored traces dropped by the tracer will not be rescued" ); } 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, - ), + error_sampler: trace_processor::new_error_sampler(error_tps), error_rare_sample_tracer_drop, }); diff --git a/bottlecap/src/traces/trace_agent.rs b/bottlecap/src/traces/trace_agent.rs index bb7bdf6cf..d08448c06 100644 --- a/bottlecap/src/traces/trace_agent.rs +++ b/bottlecap/src/traces/trace_agent.rs @@ -1145,7 +1145,7 @@ mod tests { tokio::spawn(dedup_svc.run()); let trace_processor = Arc::new(trace_processor::ServerlessTraceProcessor { obfuscation_config: Arc::new(ObfuscationConfig::default()), - error_sampler: trace_processor::new_error_sampler(false), + error_sampler: trace_processor::new_error_sampler(0.0), error_rare_sample_tracer_drop: false, }); let stats_aggregator = Arc::new(Mutex::new( diff --git a/bottlecap/src/traces/trace_processor.rs b/bottlecap/src/traces/trace_processor.rs index 57eb3ca0c..e2a877b66 100644 --- a/bottlecap/src/traces/trace_processor.rs +++ b/bottlecap/src/traces/trace_processor.rs @@ -79,8 +79,8 @@ const USER_DROP_PRIORITY: i32 = -1; /// Returns true when `config.apm_features` contains /// [`ERROR_RARE_SAMPLE_TRACER_DROP`]. /// -/// `DD_APM_FEATURES` is parsed upstream as a comma-separated list, so matching -/// is exact per entry; space-separated values are not supported. +/// `DD_APM_FEATURES` is split on commas when one is present, otherwise on +/// spaces, and matching is exact per entry. #[must_use] pub fn error_rare_sample_tracer_drop_enabled(config: &config::Config) -> bool { config @@ -93,10 +93,10 @@ pub fn error_rare_sample_tracer_drop_enabled(config: &config::Config) -> bool { #[allow(clippy::module_name_repetitions)] pub struct ServerlessTraceProcessor { pub obfuscation_config: Arc, - /// Rescues errored chunks whenever stats are computed before the backend, - /// by either the tracer or the extension. Shared across invocations so a - /// stateful mode (`RateLimited`) would keep its rolling window; a std Mutex - /// rather than tokio because `process_traces` is synchronous. + /// Rescues errored chunks independently of stats ownership. Shared across + /// invocations so the `RateLimited` budget and rolling window span the whole + /// execution environment; a std Mutex rather than tokio because + /// `process_traces` is synchronous. pub error_sampler: Arc>, /// Whether `DD_APM_FEATURES` enables [`ERROR_RARE_SAMPLE_TRACER_DROP`]. When /// true, chunks with a negative priority are rescued like `AutoDrop` chunks @@ -117,30 +117,46 @@ fn span_view(span: &Span) -> SpanView<'_> { } } -/// Builds the error sampler as the extension ships it, enabled or disabled by -/// `serverless_error_sampler_enabled`. +/// Builds the error sampler's configuration with the extension's fixed policy: +/// rate-limited rescue with the given budget and no extra sampling. /// -/// The mode is hardcoded to `AlwaysKeep`: Lambda's per-invocation trace volume -/// is low, and freeze/thaw breaks `RateLimited`'s 30s wall-clock window. There -/// is therefore no rate ceiling when enabled: every errored chunk is rescued. +/// All three fields are set explicitly so a change to the crate's defaults +/// cannot silently change bottlecap behavior. #[must_use] -pub fn new_error_sampler(enabled: bool) -> Arc> { +pub fn error_sampler_config(target_tps: f64) -> ErrorSamplerConfig { + ErrorSamplerConfig { + mode: ErrorSamplerMode::RateLimited, + target_tps, + extra_sample_rate: 1.0, + } +} + +/// Builds the error sampler as the extension ships it: rate-limited rescue with +/// the given budget. +/// +/// The rolling window is driven by the wall-clock seconds passed on each call, +/// so a freeze/thaw gap expires stale counters. Stored per-signature rates are +/// not reset by a gap: they are recomputed only when a rescue candidate arrives +/// in a new 5s bucket, a throttled signature recovers by at most 20% per +/// recomputation, and the lowest stored rate also caps unseen signatures +/// (inherited from the Go agent). A budget `<= 0` disables rescue; see +/// [`crate::config::LambdaConfig::apm_error_tps`] for the caveats. +#[must_use] +pub fn new_error_sampler(target_tps: f64) -> 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, - }, + error_sampler_config(target_tps), ))) } 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. - fn drop_sampled_out_chunks(&self, tracer_payloads: &mut Vec) { + /// Marks rescued error chunks for every stats owner. When `drop_sampled_out` + /// is true, also removes rejected chunks and any payload left without chunks. + /// `SamplerPriority::None` (-128) means no explicit priority was set and is kept. + fn sample_and_filter_chunks( + &self, + tracer_payloads: &mut Vec, + drop_sampled_out: bool, + ) { // Read once up front so non-errored and explicitly-dropped chunks never // touch the lock. Recover through poisoning: the sampler holds only // rolling-window counters, so a partially updated bucket is far cheaper @@ -176,15 +192,18 @@ impl ServerlessTraceProcessor { if chunk.priority < 0 && (!self.error_rare_sample_tracer_drop || Self::is_manual_tracer_drop(chunk)) { - return false; + return !drop_sampled_out; } if !rescue_enabled { - return false; + return !drop_sampled_out; } - self.rescue_error_chunk(chunk, env, now_secs) + // Consult the sampler even when rejected chunks must be forwarded. + self.rescue_error_chunk(chunk, env, now_secs) || !drop_sampled_out }); } - tracer_payloads.retain(|tp| !tp.chunks.is_empty()); + if drop_sampled_out { + tracer_payloads.retain(|tp| !tp.chunks.is_empty()); + } } /// Whether the chunk is an explicit tracer manual drop: priority exactly -1 @@ -634,15 +653,12 @@ impl TraceProcessor for ServerlessTraceProcessor { } }; - // Sampled-out chunks are preserved in payloads_for_stats above, so their - // stats are still counted after they are removed here. Filter whenever - // stats are computed before the backend, by either the tracer or the - // extension: the backend only needs these chunks when it owns stats. - if stats_computed_by != StatsComputedBy::Backend - && let TracerPayloadCollection::V07(ref mut tracer_payloads) = payload - { - self.drop_sampled_out_chunks(tracer_payloads); - if tracer_payloads.is_empty() { + // Rescue is independent of stats ownership. Preserve the pre-rescue stats + // snapshot and existing forwarding behavior when the backend owns stats. + let drop_sampled_out = stats_computed_by != StatsComputedBy::Backend; + if let TracerPayloadCollection::V07(ref mut tracer_payloads) = payload { + self.sample_and_filter_chunks(tracer_payloads, drop_sampled_out); + if drop_sampled_out && tracer_payloads.is_empty() { return (None, payloads_for_stats); } } @@ -792,10 +808,11 @@ impl SendingTraceProcessor { } } -/// Enabled error sampler for constructing `ServerlessTraceProcessor` in tests. +/// Error sampler with the default budget, for constructing +/// `ServerlessTraceProcessor` in tests. #[cfg(test)] -pub(crate) fn enabled_error_sampler() -> Arc> { - new_error_sampler(true) +pub(crate) fn default_error_sampler() -> Arc> { + new_error_sampler(crate::config::DEFAULT_APM_ERROR_TPS) } #[cfg(test)] @@ -964,7 +981,7 @@ mod tests { let trace_processor = ServerlessTraceProcessor { obfuscation_config: Arc::new(ObfuscationConfig::default()), - error_sampler: enabled_error_sampler(), + error_sampler: default_error_sampler(), error_rare_sample_tracer_drop: false, }; let config = create_test_config(); @@ -1412,8 +1429,8 @@ mod tests { /// Verifies that `process_traces` filters sampled-out chunks from the backend /// payload whenever stats are computed before the backend, by either the tracer - /// or the extension, while preserving them in the stats collection. When the - /// backend owns stats, all chunks are kept so the backend can compute stats. + /// or the extension, while preserving them in the stats collection. The + /// backend-owned stats path preserves its existing forwarding behavior. #[test] #[allow(clippy::unwrap_used)] fn test_process_traces_filters_sampled_out_chunks() { @@ -1428,7 +1445,7 @@ mod tests { for (compute_on_extension, client_computed_stats, expected_backend_span_count) in cases { let config = create_compute_stats_config_with(compute_on_extension); let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), false); + create_test_processor(&config, default_error_sampler(), false); let mut header_tags = create_test_header_tags(); header_tags.generic.client_computed_stats = client_computed_stats; @@ -1461,24 +1478,29 @@ mod tests { } } - /// On any stats path where stats are computed before the backend (extension or - /// tracer), 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. Priorities are never rewritten. + /// Error rescue stamps eligible errored P0 chunks for every stats owner. + /// Rejected chunks are forwarded only on the backend-owned stats path. + /// Priorities and the pre-rescue stats snapshot are never rewritten. #[test] + #[allow(clippy::float_cmp)] fn test_error_sampler_rescues_errored_p0_chunks() { - // (owner label, lambda_extension_compute_stats, client_computed_stats) - let cases = [("extension", true, false), ("tracer", false, true)]; + // (owner, extension stats enabled, tracer computed stats, forwarded count) + let cases = [ + ("backend", false, false, 4), + ("extension", true, false, 2), + ("tracer", false, true, 2), + ("tracer-precedence", true, true, 2), + ]; - for (owner, compute_on_extension, client_computed_stats) in cases { + for (owner, compute_on_extension, client_computed_stats, forwarded_count) in cases { let config = create_compute_stats_config_with(compute_on_extension); let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), false); + create_test_processor(&config, default_error_sampler(), false); let mut header_tags = create_test_header_tags(); header_tags.generic.client_computed_stats = client_computed_stats; // trace 1: kept normally (priority 1). trace 2: errored P0 (rescued). - // trace 3: non-errored P0 (dropped). trace 4: errored user drop (dropped). + // Traces 3 (non-errored P0) and 4 (errored user drop) are not rescued. let traces = vec![ vec![make_tracer_drop_span(1, 1.0, 0, None)], vec![make_tracer_drop_span(2, 0.0, 1, None)], @@ -1486,8 +1508,10 @@ mod tests { vec![make_tracer_drop_span(4, -1.0, 1, None)], ]; - let (payload_info, _stats) = + let (payload_info, stats) = processor.process_traces(config, tags_provider, header_tags, traces, 0, None); + assert_eq!(trace_spans(&stats).count(), 4); + assert!(trace_spans(&stats).all(|s| !s.metrics.contains_key("_dd.errors_sr"))); let payload_info = payload_info.expect("expected Some payload"); let backend_send_data = payload_info.builder.build(); let kept: Vec<(u64, f64)> = trace_spans(backend_send_data.get_payloads()) @@ -1495,8 +1519,8 @@ mod tests { .collect(); assert_eq!( kept.len(), - 2, - "kept normal trace + rescued errored trace (owner={owner})" + forwarded_count, + "unexpected forwarded count (owner={owner})" ); assert!( kept.contains(&(1, 1.0)), @@ -1510,27 +1534,40 @@ mod tests { let rescued_root = trace_spans(backend_send_data.get_payloads()) .find(|s| s.trace_id == 2) .expect("rescued trace present"); + assert_eq!( + rescued_root.metrics.get("_dd.errors_sr").copied(), + Some(1.0), + "_dd.errors_sr is exactly 1.0 for a cold sampler at low volume (owner={owner})" + ); assert!( - rescued_root.metrics.contains_key("_dd.errors_sr"), - "_dd.errors_sr stamped on rescued root (owner={owner})" + trace_spans(backend_send_data.get_payloads()) + .filter(|s| s.trace_id != 2) + .all(|s| !s.metrics.contains_key("_dd.errors_sr")) ); + if owner == "backend" { + assert!(kept.contains(&(3, 0.0))); + assert!(kept.contains(&(4, -1.0))); + } } } /// With `error_rare_sample_tracer_drop` enabled, an errored chunk the tracer /// dropped itself (priority -1, no decision maker) is rescued like an - /// `AutoDrop` chunk, matching the Go agent. Runs for both stats owners: the - /// tracer-owned path reaches the rescue only when filtering also applies to - /// tracer-computed stats. + /// `AutoDrop` chunk, matching the Go agent, regardless of stats ownership. #[test] fn test_error_rare_sample_tracer_drop_rescues_errored_user_drop() { // (owner label, lambda_extension_compute_stats, client_computed_stats) - let cases = [("extension", true, false), ("tracer", false, true)]; + let cases = [ + ("backend", false, false), + ("extension", true, false), + ("tracer", false, true), + ("tracer-precedence", true, true), + ]; for (owner, compute_on_extension, client_computed_stats) in cases { let config = create_compute_stats_config_with(compute_on_extension); let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), true); + create_test_processor(&config, default_error_sampler(), true); let mut header_tags = create_test_header_tags(); header_tags.generic.client_computed_stats = client_computed_stats; @@ -1624,6 +1661,177 @@ mod tests { (kept, in_stats) } + #[test] + #[allow(clippy::float_cmp)] + fn error_sampler_config_is_rate_limited_with_given_tps() { + for tps in [10.0, 2.5, 0.0, -1.0] { + let config = error_sampler_config(tps); + assert_eq!(config.mode, ErrorSamplerMode::RateLimited, "tps {tps}"); + assert_eq!(config.target_tps, tps, "tps {tps}"); + assert_eq!(config.extra_sample_rate, 1.0, "tps {tps}"); + } + } + + #[test] + fn new_error_sampler_builds_rate_limited() { + let sampler = new_error_sampler(10.0); + assert!( + matches!( + *sampler.lock().expect("lock"), + ErrorsSampler::RateLimited(_) + ), + "a positive budget must build a RateLimited sampler" + ); + assert!( + !sampler.lock().expect("lock").is_disabled(), + "10 tps must not disable rescue" + ); + for tps in [0.0, -1.0] { + assert!( + new_error_sampler(tps).lock().expect("lock").is_disabled(), + "{tps} tps must disable rescue" + ); + } + } + + /// Samples one single-span errored trace at `now`. Every call uses the same + /// signature; `n` only spreads the trace id. + fn sample_errored_trace( + sampler: &std::sync::Mutex, + now: i64, + n: u64, + ) -> SampleDecision { + let spans = [SpanView { + service: "svc", + name: "op", + resource: "res", + error: true, + http_status_code: None, + error_type: None, + }]; + let trace = TraceView { + env: "test-env", + trace_id: n.wrapping_mul(2_654_435_761), + root_index: 0, + root_global_sample_rate: 1.0, + spans: &spans, + }; + sampler.lock().expect("lock").sample(now, &trace) + } + + /// A backward clock step, a gap longer than the 30s window, and the epoch + /// fallback value must all be survivable; after the gap a cold signature + /// falls back to rate 1.0. + #[test] + #[allow(clippy::float_cmp)] + fn error_sampler_survives_clock_jumps() { + let sampler = new_error_sampler(crate::config::DEFAULT_APM_ERROR_TPS); + sample_errored_trace(&sampler, 1_700_000_000, 1); + // Backward step. + sample_errored_trace(&sampler, 1_699_999_000, 2); + assert_eq!( + sample_errored_trace(&sampler, 1_700_100_000, 3), + SampleDecision::Keep { errors_sr: 1.0 }, + "after a long gap a cold signature falls back to rate 1.0" + ); + // The `UNIX_EPOCH.elapsed()` fallback value. + sample_errored_trace(&sampler, 0, 4); + } + + /// A long freeze/thaw gap clears stale counters, but a throttled + /// signature's stored rate recovers by at most 20% per evaluation, so + /// rescue keeps working after the gap at a rate still below 1.0. + #[test] + fn error_sampler_stays_throttled_after_long_gap() { + let sampler = new_error_sampler(1.0); + let t0 = 1_700_000_000_i64; + let mut n = 0_u64; + for bucket in 0..4_i64 { + for _ in 0..200 { + n += 1; + sample_errored_trace(&sampler, t0 + 5 * bucket, n); + } + } + + // All post-gap calls share one bucket, so the rate stays fixed. Ids are + // deterministic, so this bounded scan is stable. + let after_gap = t0 + 15 + 3600; + let errors_sr = (n + 1..=n + 400) + .find_map(|id| match sample_errored_trace(&sampler, after_gap, id) { + SampleDecision::Keep { errors_sr } => Some(errors_sr), + SampleDecision::Drop => None, + }) + .expect("a throttled signature must still rescue some traces after a long gap"); + assert!( + errors_sr < 1.0, + "a previously throttled signature must stay below 1.0, got {errors_sr}" + ); + // Stored rate 1/40 before the gap, capped at a 20% increase. + assert!( + (errors_sr - 0.025 * 1.2).abs() < 1e-9, + "expected the capped recovery rate 0.03, got {errors_sr}" + ); + } + + /// Runs one batch of 50 errored P0 traces with the same signature (distinct + /// trace ids derived from `call`) through `processor`, returning how many + /// spans were rescued. A batch whose chunks are all dropped returns 0. + fn run_errored_p0_batch( + processor: &ServerlessTraceProcessor, + config: &Arc, + tags_provider: &Arc, + call: usize, + ) -> usize { + let traces: Vec> = (0..50) + .map(|i| { + let trace_id = ((call * 50 + i + 1) as u64).wrapping_mul(2_654_435_761); + vec![make_tracer_drop_span(trace_id, 0.0, 1, None)] + }) + .collect(); + + processor + .process_traces( + config.clone(), + tags_provider.clone(), + create_test_header_tags(), + traces, + 0, + None, + ) + .0 + .map_or(0, |info| { + trace_spans(info.builder.build().get_payloads()).count() + }) + } + + /// The rescue budget lives on the shared sampler, so it spans invocations: + /// ten batches through one processor rescue far fewer traces than ten + /// batches each with a fresh sampler. + #[test] + fn error_sampler_budget_is_shared_across_invocations() { + let config = create_compute_stats_config(); + + let (tags_provider, processor) = + create_test_processor(&config, new_error_sampler(1.0), false); + let shared: usize = (0..10) + .map(|call| run_errored_p0_batch(&processor, &config, &tags_provider, call)) + .sum(); + + let control: usize = (0..10) + .map(|call| { + let (tags_provider, processor) = + create_test_processor(&config, new_error_sampler(1.0), false); + run_errored_p0_batch(&processor, &config, &tags_provider, call) + }) + .sum(); + + assert!(shared > 0, "the shared run must rescue some traces"); + assert!( + shared * 2 < control, + "one shared budget must rescue far fewer traces than a fresh sampler per batch: shared {shared}, control {control}" + ); + } + /// A manual tracer drop (priority -1 with decision maker "-4" in the root /// span meta) is never rescued, even with the feature enabled, matching the /// Go agent's `isManualUserDrop`. @@ -1631,7 +1839,7 @@ mod tests { fn test_error_rare_sample_tracer_drop_excludes_manual_drop() { let config = create_compute_stats_config(); let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), true); + create_test_processor(&config, default_error_sampler(), true); let trace = vec![make_tracer_drop_span(7, -1.0, 1, Some("-4"))]; let (kept, in_stats) = run_single_trace(&processor, &config, tags_provider, trace); @@ -1650,7 +1858,7 @@ mod tests { fn test_error_rare_sample_tracer_drop_manual_drop_on_child_span() { let config = create_compute_stats_config(); let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), true); + create_test_processor(&config, default_error_sampler(), true); // Root is fine and carries no decision maker; the errored child does. let mut root = make_tracer_drop_span(7, -1.0, 0, None); @@ -1672,7 +1880,7 @@ mod tests { fn test_error_rare_sample_tracer_drop_nonmanual_decision_maker_rescued() { let config = create_compute_stats_config(); let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), true); + create_test_processor(&config, default_error_sampler(), true); let trace = vec![make_tracer_drop_span(7, -1.0, 1, Some("-3"))]; let (kept, _) = run_single_trace(&processor, &config, tags_provider, trace); @@ -1689,7 +1897,7 @@ mod tests { fn test_error_rare_sample_tracer_drop_other_negative_priority_rescued() { let config = create_compute_stats_config(); let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), true); + create_test_processor(&config, default_error_sampler(), true); let trace = vec![make_tracer_drop_span(7, -2.0, 1, Some("-4"))]; let (kept, _) = run_single_trace(&processor, &config, tags_provider, trace); @@ -1708,7 +1916,7 @@ mod tests { for (first_dm, second_dm, rescued) in [("-3", "-4", true), ("-4", "-3", false)] { let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), true); + create_test_processor(&config, default_error_sampler(), true); let mut root = make_tracer_drop_span(7, -1.0, 1, Some(first_dm)); root.span_id = 1; let mut child = make_tracer_drop_span(7, -1.0, 0, Some(second_dm)); @@ -1729,7 +1937,7 @@ mod tests { fn test_error_rare_sample_tracer_drop_nonerrored_still_dropped() { let config = create_compute_stats_config(); let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), true); + create_test_processor(&config, default_error_sampler(), true); let trace = vec![make_tracer_drop_span(7, -1.0, 0, None)]; let (kept, in_stats) = run_single_trace(&processor, &config, tags_provider, trace); @@ -1747,7 +1955,7 @@ mod tests { fn test_error_rare_sample_tracer_drop_requires_sampler_enabled() { let config = create_compute_stats_config(); let (tags_provider, processor) = - create_test_processor(&config, new_error_sampler(false), true); + create_test_processor(&config, new_error_sampler(0.0), true); let trace = vec![make_tracer_drop_span(7, -1.0, 1, None)]; let (kept, _) = run_single_trace(&processor, &config, tags_provider, trace); @@ -1764,7 +1972,7 @@ mod tests { fn test_error_sampler_rescues_chunk_with_errored_child_span() { let config = create_compute_stats_config(); let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), false); + create_test_processor(&config, default_error_sampler(), false); let header_tags = create_test_header_tags(); let make_span = |span_id: u64, parent_id: u64, error: i32| -> pb::Span { @@ -1808,35 +2016,59 @@ mod tests { ); } - /// With the error sampler disabled (the shipping default), errored P0 + /// With error trace rescue disabled (`DD_APM_ERROR_TPS <= 0`), errored P0 /// chunks are dropped: no rescue, no `_dd.errors_sr`. #[test] fn test_disabled_error_sampler_drops_errored_p0_chunks() { - let config = create_compute_stats_config(); - let (tags_provider, processor) = - create_test_processor(&config, new_error_sampler(false), false); - let header_tags = create_test_header_tags(); + for tps in [0.0, -1.0] { + let config = create_compute_stats_config(); + let (tags_provider, processor) = + create_test_processor(&config, new_error_sampler(tps), false); + let header_tags = create_test_header_tags(); - 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 traces = vec![vec![make_tracer_drop_span(1, 0.0, 1, None)]]; - 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" - ); + 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 rescue is disabled (tps {tps})" + ); + } + } + + #[test] + fn test_backend_stats_forwards_unrescued_chunks() { + for feature_enabled in [false, true] { + for tps in [0.0, -1.0] { + let config = create_compute_stats_config_with(false); + let (tags_provider, processor) = + create_test_processor(&config, new_error_sampler(tps), feature_enabled); + let traces = vec![ + vec![make_tracer_drop_span(1, 0.0, 1, None)], + vec![make_tracer_drop_span(2, -1.0, 1, Some("-4"))], + vec![make_tracer_drop_span(3, 0.0, 0, None)], + ]; + let (payload_info, stats) = processor.process_traces( + config, + tags_provider, + create_test_header_tags(), + traces, + 0, + None, + ); + let payload = payload_info + .expect("backend path must forward rejected chunks") + .builder + .build(); + assert_eq!(trace_spans(payload.get_payloads()).count(), 3); + assert!( + trace_spans(payload.get_payloads()) + .all(|s| !s.metrics.contains_key("_dd.errors_sr")) + ); + assert_eq!(trace_spans(&stats).count(), 3); + } + } } /// With the error sampler disabled, an errored P0 chunk dropped from the @@ -1846,7 +2078,7 @@ mod tests { fn test_disabled_error_sampler_keeps_dropped_errored_p0_chunk_in_stats() { let config = create_compute_stats_config(); let (tags_provider, processor) = - create_test_processor(&config, new_error_sampler(false), false); + create_test_processor(&config, new_error_sampler(0.0), false); let header_tags = create_test_header_tags(); let make_span = |trace_id: u64, priority: f64, error: i32| -> pb::Span { @@ -1914,7 +2146,7 @@ mod tests { fn test_process_traces_returns_none_when_all_sampled_out() { let config = create_compute_stats_config(); let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), false); + create_test_processor(&config, default_error_sampler(), false); let header_tags = create_test_header_tags(); let make_dropped_span = |trace_id: u64| -> pb::Span { @@ -1965,7 +2197,7 @@ mod tests { let config = create_compute_stats_config(); let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), false); + create_test_processor(&config, default_error_sampler(), false); let header_tags = create_test_header_tags(); let make_span = |trace_id: u64, priority: f64| -> pb::Span { @@ -2049,7 +2281,7 @@ mod tests { ..Config::default() }); let (tags_provider, processor) = - create_test_processor(&config, enabled_error_sampler(), false); + create_test_processor(&config, default_error_sampler(), false); let header_tags = create_test_header_tags(); let span = pb::Span { @@ -2605,7 +2837,7 @@ mod tests { appsec: None, processor: Arc::new(ServerlessTraceProcessor { obfuscation_config: Arc::new(ObfuscationConfig::default()), - error_sampler: enabled_error_sampler(), + error_sampler: default_error_sampler(), error_rare_sample_tracer_drop: false, }), trace_tx, diff --git a/bottlecap/tests/apm_integration_test.rs b/bottlecap/tests/apm_integration_test.rs index d4a50d61d..ed6f7f87d 100644 --- a/bottlecap/tests/apm_integration_test.rs +++ b/bottlecap/tests/apm_integration_test.rs @@ -21,7 +21,7 @@ use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; use bottlecap::LAMBDA_RUNTIME_SLUG; -use bottlecap::config::{Config, get_config}; +use bottlecap::config::{Config, DEFAULT_APM_ERROR_TPS, get_config}; use bottlecap::tags::provider::Provider; use bottlecap::traces::http_client::create_client; use bottlecap::traces::stats_aggregator::StatsAggregator; @@ -375,7 +375,7 @@ async fn run_processor_pipeline_with_settings( appsec: None, processor: Arc::new(ServerlessTraceProcessor { obfuscation_config: Arc::new(ObfuscationConfig::default()), - error_sampler: new_error_sampler(true), + error_sampler: new_error_sampler(DEFAULT_APM_ERROR_TPS), error_rare_sample_tracer_drop, }), trace_tx, @@ -509,13 +509,17 @@ fn sampled_root_span(trace_id: u64, priority: f64, error: i32) -> pb::Span { } } -/// An errored P0 trace is rescued and reaches the intake, while a non-errored P0 trace is -/// dropped. Applies whenever stats are computed before the backend, by either the -/// extension or the tracer. +/// An errored P0 trace is marked as rescued for every stats owner. Non-errored +/// P0 traces are forwarded only when the backend owns stats. #[tokio::test] async fn e2e_error_sampler_rescues_only_errored_p0_traces() { // (owner label, lambda_extension_compute_stats, client_computed_stats) - let cases = [("extension", true, false), ("tracer", false, true)]; + let cases = [ + ("backend", false, false), + ("extension", true, false), + ("tracer", false, true), + ("tracer-precedence", true, true), + ]; for (owner, compute_on_extension, client_computed_stats) in cases { let rescued_trace_id = 1; @@ -533,12 +537,20 @@ async fn e2e_error_sampler_rescues_only_errored_p0_traces() { let captured_spans: Vec<&pb::Span> = captured_trace_spans(&outcome.traces).collect(); assert_eq!( captured_spans.len(), - 1, - "only the rescued trace reaches intake (owner={owner})" + if owner == "backend" { 2 } else { 1 }, + "unexpected forwarded count (owner={owner})" + ); + assert!( + captured_spans + .iter() + .filter(|s| s.trace_id != rescued_trace_id) + .all(|s| !s.metrics.contains_key("_dd.errors_sr")) ); - let rescued_span = captured_spans[0]; - assert_eq!(rescued_span.trace_id, rescued_trace_id); + let rescued_span = captured_spans + .iter() + .find(|s| s.trace_id == rescued_trace_id) + .expect("rescued trace present"); assert!( rescued_span.metrics.contains_key("_dd.errors_sr"), "the rescued root span must carry its error sampling rate (owner={owner})", @@ -547,8 +559,8 @@ async fn e2e_error_sampler_rescues_only_errored_p0_traces() { } /// Sampled-out chunks are filtered from the trace intake whenever stats are computed -/// before the backend, by either the extension or the tracer, while backend-owned -/// stats keep them so the backend can compute stats. +/// before the backend, by either the extension or the tracer. The backend-owned +/// stats path preserves its existing forwarding behavior. #[tokio::test] async fn e2e_sampled_out_chunks_filtered_by_stats_owner() { // (owner label, lambda_extension_compute_stats, client_computed_stats, expected_trace_ids) @@ -670,14 +682,18 @@ fn captured_trace_spans(traces: &[pb::AgentPayload]) -> impl Iterator = captured_trace_spans(&outcome.traces).collect(); assert_eq!( captured_spans.len(), - 2, - "only the rescued tracer drop and the normally kept trace reach intake \ - (owner={owner})" + if owner == "backend" { 4 } else { 2 }, + "unexpected forwarded count (owner={owner})" ); assert!( captured_spans @@ -726,10 +741,17 @@ async fn e2e_error_rare_sample_tracer_drop_rescues_tracer_drops_selectively() { "the rescued root span must carry its error sampling rate (owner={owner})", ); + assert!( + captured_spans + .iter() + .filter(|s| s.trace_id != rescued_trace_id) + .all(|s| !s.metrics.contains_key("_dd.errors_sr")) + ); + // Dropped inputs must still be accounted for in stats, but only when the // extension computes them: with tracer-owned stats the extension // suppresses its own stats generation entirely. - if compute_on_extension { + if compute_on_extension && !client_computed_stats { let grouped = captured_grouped_stats(&outcome.stats); let total_hits: u64 = grouped.iter().map(|g| g.hits).sum(); let total_errors: u64 = grouped.iter().map(|g| g.errors).sum(); @@ -744,7 +766,7 @@ async fn e2e_error_rare_sample_tracer_drop_rescues_tracer_drops_selectively() { } else { assert!( outcome.stats.is_empty(), - "extension must not emit stats when the tracer computed them (owner={owner})" + "extension must not emit stats for another stats owner (owner={owner})" ); } } diff --git a/bottlecap/tests/fault_injection_test.rs b/bottlecap/tests/fault_injection_test.rs index 20a61974a..b1f4af96c 100644 --- a/bottlecap/tests/fault_injection_test.rs +++ b/bottlecap/tests/fault_injection_test.rs @@ -19,10 +19,25 @@ mod fixtures; mod harness; use harness::prelude::*; +use libdd_trace_protobuf::pb; +use prost::Message as _; const TRACES_PATH: &str = "/api/v0.2/traces"; const LOGS_PATH: &str = "/api/v2/logs"; +/// The root span with `trace_id` from an accepted trace request. +fn decoded_root_span(request: &IntakeRequest, trace_id: u64) -> pb::Span { + let body = request.decoded_body().expect("decode intake body"); + let payload = pb::AgentPayload::decode(body.as_slice()).expect("decode AgentPayload"); + payload + .tracer_payloads + .into_iter() + .flat_map(|payload| payload.chunks) + .flat_map(|chunk| chunk.spans) + .find(|span| span.trace_id == trace_id) + .expect("the trace must reach the intake") +} + /// The function log [`end_invocation_and_wait`] sends for `request_id`. fn log_text(request_id: &str) -> String { format!("log for {request_id}") @@ -213,3 +228,86 @@ async fn intake_closing_mid_body_does_not_hold_next() { ); assert!(exit.is_some_and(|e| e.success()), "exit {exit:?}"); } + +/// An errored priority-0 trace is marked for backend retention even when the +/// backend owns stats, without changing its sampling priority. +#[tokio::test(flavor = "multi_thread")] +#[serial_test::serial(bottlecap_ports)] +async fn backend_owned_stats_rescues_errored_sampled_out_trace() { + let s = Scenario::start(FaultMode::Ok).await; + // Backend owns stats: the extension computes none and the tracer sends no + // `Datadog-Client-Computed-Stats` header (`post_trace` sends none). + let mut bc = s + .bottlecap() + .env("DD_LAMBDA_EXTENSION_COMPUTE_STATS", "false") + .env("DD_APM_ERROR_TPS", "10") + .spawn(); + let deadline = s + .invoke("req-1", secs(10), 1, bc.startup_deadline()) + .await + .unwrap_or_else(|| panic!("no /next poll. {}", bc.log_tail(40))); + + let mut traces = fixtures::traces::make_traces(1, 1, false); + let root = &mut traces[0][0]; + root.error = 1; + root.metrics + .insert("_sampling_priority_v1".to_string(), 0.0); + let trace_id = root.trace_id; + let status = post_trace(fixtures::traces::v04_msgpack(&traces), STARTUP_BUDGET) + .await + .unwrap(); + eprintln!("trace agent answered {status} to the errored priority-0 trace"); + + let released = end_invocation_and_wait(&s, "req-1", true, 2, deadline).await; + let accepted = s + .intake + .wait_for( + 1, + |r| { + r.path.contains(TRACES_PATH) + && r.status == 202 + && r.api_key() == Some(TEST_API_KEY) + && r.trace_ids().unwrap().contains(&trace_id) + }, + deadline, + ) + .await; + + s.shutdown_cleanly(&mut bc).await; + + assert!( + released, + "the errored sampled-out trace held /next. {}", + bc.log_tail(40) + ); + assert!( + accepted, + "the errored priority-0 trace never reached the intake with the API key (trace agent answered {status}). Intake:\n{}", + s.intake.summary() + ); + let request = s + .intake + .requests_to(TRACES_PATH) + .into_iter() + .find(|r| r.status == 202 && r.api_key() == Some(TEST_API_KEY)) + .expect("accepted trace request"); + let root = decoded_root_span(&request, trace_id); + assert!( + root.error != 0, + "the decoded root span must still be errored. Intake:\n{}", + s.intake.summary() + ); + let errors_sr = root.metrics.get("_dd.errors_sr").copied(); + assert!( + errors_sr.is_some_and(|rate| rate > 0.0), + "the backend-owned errored trace must be rescued with _dd.errors_sr > 0, got {errors_sr:?}. Intake:\n{}\n{}", + s.intake.summary(), + bc.log_tail(40) + ); + assert_eq!( + root.metrics.get("_sampling_priority_v1").copied(), + Some(0.0), + "the rescue must not change the sampling priority. Intake:\n{}", + s.intake.summary() + ); +}