diff --git a/.gitignore b/.gitignore index 5a4edd14ce..0b20ce159e 100644 --- a/.gitignore +++ b/.gitignore @@ -32,3 +32,4 @@ examples/cxx/exporter_manager.exe examples/cxx/profiling examples/cxx/profiling.exe profile.pprof +*.snap.new diff --git a/Cargo.lock b/Cargo.lock index 4839ed758a..327a84d9e9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -999,6 +999,17 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "console" +version = "0.16.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d64e8af5551369d19cf50138de61f1c42074ab970f74e99be916646777f8fc87" +dependencies = [ + "encode_unicode", + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "console-api" version = "0.9.0" @@ -1713,6 +1724,12 @@ version = "1.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "60b1af1c220855b6ceac025d3f6ecdd2b7c4894bfe9cd9bda4fbb4bc7c0d4cf0" +[[package]] +name = "encode_unicode" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34aa73646ffb006b8f5147f3dc182bd4bcb190227ce861fc4a4844bf8e3cb2c0" + [[package]] name = "encoding_rs" version = "0.8.35" @@ -2670,6 +2687,21 @@ dependencies = [ "serde_core", ] +[[package]] +name = "insta" +version = "1.47.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b4a6248eb93a4401ed2f37dfe8ea592d3cf05b7cf4f8efa867b6895af7e094e" +dependencies = [ + "console", + "once_cell", + "pest", + "pest_derive", + "serde", + "similar", + "tempfile", +] + [[package]] name = "io-lifetimes" version = "1.0.11" @@ -2975,6 +3007,7 @@ dependencies = [ "http", "http-body-util", "httpmock", + "insta", "libdd-capabilities", "libdd-capabilities-impl", "libdd-common", @@ -4084,6 +4117,49 @@ version = "2.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e" +[[package]] +name = "pest" +version = "2.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e0848c601009d37dfa3430c4666e147e49cdcf1b92ecd3e63657d8a5f19da662" +dependencies = [ + "memchr", + "ucd-trie", +] + +[[package]] +name = "pest_derive" +version = "2.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11f486f1ea21e6c10ed15d5a7c77165d0ee443402f0780849d1768e7d9d6fe77" +dependencies = [ + "pest", + "pest_generator", +] + +[[package]] +name = "pest_generator" +version = "2.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8040c4647b13b210a963c1ed407c1ff4fdfa01c31d6d2a098218702e6664f94f" +dependencies = [ + "pest", + "pest_meta", + "proc-macro2", + "quote", + "syn 2.0.87", +] + +[[package]] +name = "pest_meta" +version = "2.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89815c69d36021a140146f26659a81d6c2afa33d216d736dd4be5381a7362220" +dependencies = [ + "pest", + "sha2", +] + [[package]] name = "petgraph" version = "0.8.3" @@ -6162,6 +6238,12 @@ version = "1.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42ff0bf0c66b8238c6f3b578df37d0b7848e55df8577b3f74f92a69acceeb825" +[[package]] +name = "ucd-trie" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971" + [[package]] name = "unarray" version = "0.1.4" diff --git a/LICENSE-3rdparty.csv b/LICENSE-3rdparty.csv index 3ec30833d4..33d4dbcf81 100644 --- a/LICENSE-3rdparty.csv +++ b/LICENSE-3rdparty.csv @@ -85,6 +85,7 @@ colorchoice,https://github.com/rust-cli/anstyle,MIT OR Apache-2.0,The colorchoic colored,https://github.com/mackwic/colored,MPL-2.0,Thomas Wickham combine,https://github.com/Marwes/combine,MIT,Markus Westerlind concurrent-queue,https://github.com/smol-rs/concurrent-queue,Apache-2.0 OR MIT,"Stjepan Glavina , Taiki Endo , John Nunley " +console,https://github.com/console-rs/console,MIT,The console Authors console-api,https://github.com/tokio-rs/console,MIT,"Eliza Weisman , Tokio Contributors " console-subscriber,https://github.com/tokio-rs/console,MIT,"Eliza Weisman , Tokio Contributors " const_format,https://github.com/rodrimati1992/const_format_crates,Zlib,rodrimati1992 @@ -125,6 +126,7 @@ dispatch2,https://github.com/madsmtm/objc2,Zlib OR Apache-2.0 OR MIT,"Mads Marqu displaydoc,https://github.com/yaahc/displaydoc,MIT OR Apache-2.0,Jane Lusby dyn-clone,https://github.com/dtolnay/dyn-clone,MIT OR Apache-2.0,David Tolnay either,https://github.com/rayon-rs/either,MIT OR Apache-2.0,bluss +encode_unicode,https://github.com/tormol/encode_unicode,Apache-2.0 OR MIT,Torbjørn Birch Moltu encoding_rs,https://github.com/hsivonen/encoding_rs,(Apache-2.0 OR MIT) AND BSD-3-Clause,Henri Sivonen enum-as-inner,https://github.com/bluejekyll/enum-as-inner,MIT OR Apache-2.0,Benjamin Fry equivalent,https://github.com/cuviper/equivalent,Apache-2.0 OR MIT,The equivalent Authors @@ -282,6 +284,10 @@ parking_lot_core,https://github.com/Amanieu/parking_lot,MIT OR Apache-2.0,Amanie paste,https://github.com/dtolnay/paste,MIT OR Apache-2.0,David Tolnay path-tree,https://github.com/viz-rs/path-tree,MIT OR Apache-2.0,Fangdun Tsai percent-encoding,https://github.com/servo/rust-url,MIT OR Apache-2.0,The rust-url developers +pest,https://github.com/pest-parser/pest,MIT OR Apache-2.0,Dragoș Tiselice +pest_derive,https://github.com/pest-parser/pest,MIT OR Apache-2.0,Dragoș Tiselice +pest_generator,https://github.com/pest-parser/pest,MIT OR Apache-2.0,Dragoș Tiselice +pest_meta,https://github.com/pest-parser/pest,MIT OR Apache-2.0,Dragoș Tiselice petgraph,https://github.com/petgraph/petgraph,MIT OR Apache-2.0,"bluss, mitchmindtree" pico-args,https://github.com/RazrFalcon/pico-args,MIT,Yevhenii Reizner pin-project,https://github.com/taiki-e/pin-project,Apache-2.0 OR MIT,The pin-project Authors @@ -464,6 +470,7 @@ try-lock,https://github.com/seanmonstar/try-lock,MIT,Sean McArthur typeid,https://github.com/dtolnay/typeid,MIT OR Apache-2.0,David Tolnay typenum,https://github.com/paholg/typenum,MIT OR Apache-2.0,"Paho Lurie-Gregg , Andre Bogus " +ucd-trie,https://github.com/BurntSushi/ucd-generate,MIT OR Apache-2.0,Andrew Gallant unarray,https://github.com/cameron1024/unarray,MIT OR Apache-2.0,The unarray Authors unicase,https://github.com/seanmonstar/unicase,MIT OR Apache-2.0,Sean McArthur unicode-ident,https://github.com/dtolnay/unicode-ident,(MIT OR Apache-2.0) AND Unicode-DFS-2016,David Tolnay diff --git a/libdd-common/src/regex_engine.rs b/libdd-common/src/regex_engine.rs index f3674f6e12..c5fb7d7973 100644 --- a/libdd-common/src/regex_engine.rs +++ b/libdd-common/src/regex_engine.rs @@ -13,7 +13,7 @@ //! regexes requiring Unicode character class support. #[cfg(all(feature = "regex-lite", not(feature = "require-regex-full")))] -pub use regex_lite::{escape, Captures, Regex, RegexBuilder, Replacer}; +pub use regex_lite::{escape, Captures, Error, Regex, RegexBuilder, Replacer}; #[cfg(not(all(feature = "regex-lite", not(feature = "require-regex-full"))))] -pub use regex::{escape, Captures, Regex, RegexBuilder, Replacer}; +pub use regex::{escape, Captures, Error, Regex, RegexBuilder, Replacer}; diff --git a/libdd-data-pipeline/Cargo.toml b/libdd-data-pipeline/Cargo.toml index 192cf04ca5..a6bc2aff54 100644 --- a/libdd-data-pipeline/Cargo.toml +++ b/libdd-data-pipeline/Cargo.toml @@ -80,6 +80,7 @@ tokio = { version = "1.23", features = [ "time", "test-util", ], default-features = false } +insta = { version = "1.47.2", features = ["json", "redactions"] } [features] default = ["https", "telemetry"] diff --git a/libdd-data-pipeline/src/agent_info/fetcher.rs b/libdd-data-pipeline/src/agent_info/fetcher.rs index 43e594caab..131e4c04fc 100644 --- a/libdd-data-pipeline/src/agent_info/fetcher.rs +++ b/libdd-data-pipeline/src/agent_info/fetcher.rs @@ -408,23 +408,43 @@ mod single_threaded_tests { }, "remove_stack_traces": false, "redis": { - "enabled": true, - "remove_all_args": false + "Enabled": true, + "RemoveAllArgs": false }, "memcached": { - "enabled": true, - "keep_command": false + "Enabled": true, + "KeepCommand": false } } }, - "peer_tags": ["db.hostname","http.host","aws.s3.bucket"] + "peer_tags": ["db.hostname","http.host","aws.s3.bucket"], + "obfuscation_version": 1, + "filter_tags": { + "reject": [ + "appsec.events.system_tests_appsec_event.value:tf-reject-exact" + ], + "require": [ + "appsec.events.system_tests_appsec_event.value:tf-require-exact" + ] + }, + "filter_tags_regex": { + "reject": [ + "appsec.events.system_tests_appsec_event.value:tf-reject-regex-.*" + ], + "require": [ + "appsec.events.system_tests_appsec_event.value:tf-require-regex-.*" + ] + }, + "ignore_resources": [ + ".*(stats-unique|StatsUniqueHandler).*" + ] }"#; fn calculate_hash(json: &str) -> String { format!("{:x}", Sha256::digest(json.as_bytes())) } - const TEST_INFO_HASH: &str = "cce54bf6e7d1bf38088a3ec809bfeec160bc52d37f70bd6b581ce3c2f7be5a65"; + const TEST_INFO_HASH: &str = "d0f6dde2c1ef3b7b776a58162d42574346e23f4677c3fafb440f5c7ca83a8a28"; #[cfg_attr(miri, ignore)] #[tokio::test] diff --git a/libdd-data-pipeline/src/agent_info/schema.rs b/libdd-data-pipeline/src/agent_info/schema.rs index f0eedc97e1..341e7b077e 100644 --- a/libdd-data-pipeline/src/agent_info/schema.rs +++ b/libdd-data-pipeline/src/agent_info/schema.rs @@ -40,20 +40,25 @@ pub struct AgentInfoStruct { /// Container tags hash from HTTP response header pub container_tags_hash: Option, /// Exact-match tag filters applied before stats computation (root span only). - pub filter_tags: Option, + #[serde(default)] + pub filter_tags: FilterTagsConfig, /// Regex-match tag filters applied before stats computation (root span only). - pub filter_tags_regex: Option, + #[serde(default)] + pub filter_tags_regex: FilterTagsConfig, /// Regex patterns for root-span resource names; matching traces are excluded from stats. - pub ignore_resources: Option>, + #[serde(default)] + pub ignore_resources: Vec, } /// Require/reject lists for tag-based trace filters exposed by the agent /info endpoint. #[derive(Clone, Serialize, Deserialize, Default, Debug, PartialEq)] pub struct FilterTagsConfig { /// All listed filters must match at least one root-span tag for the trace to be accepted. - pub require: Option>, + #[serde(default)] + pub require: Vec, /// If any listed filter matches a root-span tag the trace is rejected. - pub reject: Option>, + #[serde(default)] + pub reject: Vec, } #[allow(missing_docs)] @@ -99,14 +104,21 @@ pub struct HttpObfuscationConfig { #[allow(missing_docs)] #[derive(Clone, Serialize, Deserialize, Default, Debug, PartialEq)] +#[serde(rename_all = "PascalCase")] pub struct RedisObfuscationConfig { + // Agent sent pascal case fields here in versions <7.79.0 + #[serde(alias = "Enabled")] pub enabled: bool, + #[serde(alias = "RemoveAllArgs")] pub remove_all_args: bool, } #[allow(missing_docs)] #[derive(Clone, Serialize, Deserialize, Default, Debug, PartialEq)] pub struct MemcachedObfuscationConfig { + // Agent sent pascal case fields here in versions <7.79.0 + #[serde(alias = "Enabled")] pub enabled: bool, + #[serde(alias = "KeepCommand")] pub keep_command: bool, } diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs index bd157abe8d..2f7f92add6 100644 --- a/libdd-data-pipeline/src/trace_exporter/builder.rs +++ b/libdd-data-pipeline/src/trace_exporter/builder.rs @@ -1,6 +1,7 @@ // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 +use crate::agent_info::schema::FilterTagsConfig; use crate::agent_info::AgentInfoFetcher; use crate::otlp::config::{OtlpProtocol, DEFAULT_OTLP_TIMEOUT}; use crate::otlp::OtlpTraceConfig; @@ -8,6 +9,7 @@ use crate::otlp::OtlpTraceConfig; use crate::telemetry::TelemetryClientBuilder; use crate::trace_exporter::agent_response::AgentResponsePayloadVersion; use crate::trace_exporter::error::BuilderErrorKind; +use crate::trace_exporter::trace_filter::TraceFilterer; #[cfg(all(not(target_arch = "wasm32"), feature = "telemetry"))] use crate::trace_exporter::TelemetryConfig; #[cfg(not(target_arch = "wasm32"))] @@ -65,6 +67,9 @@ pub struct TraceExporterBuilder { connection_timeout: Option, otlp_endpoint: Option, otlp_headers: Vec<(String, String)>, + filter_tags: FilterTagsConfig, + filter_tags_regex: FilterTagsConfig, + ignore_resources: Vec, } impl TraceExporterBuilder { @@ -286,6 +291,24 @@ impl TraceExporterBuilder { self } + // TODO: doc + pub fn set_filter_tags(&mut self, filter_tags: FilterTagsConfig) -> &mut Self { + self.filter_tags = filter_tags; + self + } + + // TODO: doc + pub fn set_filter_tags_regex(&mut self, filter_tags_regex: FilterTagsConfig) -> &mut Self { + self.filter_tags_regex = filter_tags_regex; + self + } + + // TODO: doc + pub fn set_ignore_resources(&mut self, ignore_resources: Vec) -> &mut Self { + self.ignore_resources = ignore_resources; + self + } + #[allow(missing_docs)] pub fn build( self, @@ -496,6 +519,11 @@ impl TraceExporterBuilder { .agent_rates_payload_version_enabled .then(AgentResponsePayloadVersion::new), otlp_config, + trace_filterer: TraceFilterer::new( + &self.filter_tags, + &self.filter_tags_regex, + &self.ignore_resources, + ), }) } diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index 561bc56e88..3dd1dc501d 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -5,6 +5,7 @@ pub mod builder; pub mod error; pub mod metrics; pub mod stats; +mod trace_filter; mod trace_serializer; // Re-export the builder @@ -236,6 +237,7 @@ pub struct TraceExporter, /// When set, traces are exported via OTLP HTTP/JSON instead of the Datadog agent. otlp_config: Option, + trace_filterer: trace_filter::TraceFilterer, } impl TraceExporter { @@ -382,6 +384,12 @@ impl Tra fn check_agent_info(&self) { if let Some(agent_info) = agent_info::get_agent_info() { if self.has_agent_info_state_changed(&agent_info) { + // FIXME: trace_filterer should only be enabled when CSS is on. (why ?) + self.trace_filterer.update_conf( + &agent_info.info.filter_tags, + &agent_info.info.filter_tags_regex, + &agent_info.info.ignore_resources, + ); match &**self.client_side_stats.status.load() { StatsComputationStatus::Disabled => {} StatsComputationStatus::DisabledByAgent { .. } => { @@ -610,6 +618,11 @@ impl Tra ) -> Result { let mut header_tags: TracerHeaderTags = self.metadata.borrow().into(); + // FIXME: when client_computed_top_level is true, looking twice for the root span here is + // inefficient and just below in process_traces_for_stats. + // Also, only do it when css is on (why ???) + self.trace_filterer.filter_traces(&mut traces); + // Process stats computation and drop non-sampled (p0) chunks. // This must run before the OTLP path so that unsampled spans are not exported. let dropped_p0_stats = stats::process_traces_for_stats( @@ -1888,10 +1901,14 @@ mod tests { #[cfg(test)] mod single_threaded_tests { + use std::collections::HashMap; + use std::sync::Mutex; + use super::*; use crate::agent_info; use httpmock::prelude::*; use libdd_capabilities_impl::NativeCapabilities; + use libdd_trace_protobuf::pb::ClientStatsPayload; use libdd_trace_utils::msgpack_encoder; use libdd_trace_utils::span::v04::SpanBytes; @@ -2190,4 +2207,185 @@ mod single_threaded_tests { "obfuscation must activate when opted in and agent supports" ); } + + #[cfg_attr(miri, ignore)] + #[test] + fn test_trace_filters_snapshot() { + // Clear the agent info cache to ensure test isolation + agent_info::clear_cache_for_test(); + + let server = MockServer::start(); + let captured_stats = Arc::new(Mutex::new(Vec::new())); + + let captured_stats_in = captured_stats.clone(); + + let mock_traces = server.mock(|when, then| { + when.method(POST) + .header("Content-type", "application/msgpack") + .path("/v0.4/traces"); + then.status(200).body(""); + }); + + let mock_stats = server.mock(|when, then| { + when.method(POST) + .header("Content-type", "application/msgpack") + .path("/v0.6/stats") + .is_true(move |req| { + captured_stats_in.lock().unwrap().push(req.body_vec()); + true + }); + then.status(200).body(""); + }); + + let _mock_info = server.mock(|when, then| { + when.method(GET).path("/info"); + then.status(200) + .header("content-type", "application/json") + .header("datadog-agent-state", "1") + .body( + r#"{ + "version":"1", + "client_drop_p0s":true, + "endpoints":["/v0.4/traces","/v0.6/stats"], + "filter_tags": {"reject": ["my_ignore_tag"], "require": ["my_require_tag:true"]}, + "filter_tags_regex": {"reject": ["my_regex_ignore_tag:.*true.*"]}, + "ignore_resources": [".*IGNORED.*"] + }"#, + ); + }); + + let runtime = Arc::new(SharedRuntime::new().unwrap()); + + let mut builder = TraceExporter::::builder(); + builder + .set_url(&server.url("/")) + .set_service("test") + .set_env("staging") + .set_tracer_version("v0.1") + .set_language("nodejs") + .set_language_version("1.0") + .set_language_interpreter("v8") + .set_input_format(TraceExporterInputFormat::V04) + .set_output_format(TraceExporterOutputFormat::V04) + .set_shared_runtime(runtime.clone()) + .enable_stats(Duration::from_secs(10)); + let exporter = builder.build::().unwrap(); + + // Wait for the info fetcher to get the config + while agent_info::get_agent_info().is_none() { + std::thread::sleep(Duration::from_millis(100)); + } + + let result = exporter.send( + msgpack_encoder::v04::to_vec(&[ + vec![SpanBytes { + duration: 10, + resource: "test".into(), + meta: HashMap::from_iter([("my_require_tag".into(), "true".into())]), + ..Default::default() + }], + // This one gets filtered out because it matches an ignore_resources pattern + vec![SpanBytes { + duration: 10, + resource: "test IGNORED resource test".into(), + meta: HashMap::from_iter([("my_require_tag".into(), "true".into())]), + ..Default::default() + }], + // This one gets filtered out because one of its tag matches a reject filter_tag + vec![SpanBytes { + duration: 10, + resource: "test ignored because of reject filter_tag".into(), + meta: HashMap::from_iter([ + ("my_ignore_tag".into(), "".into()), + ("my_require_tag".into(), "true".into()), + ]), + ..Default::default() + }], + // This one gets filtered out because one of its tag matches a reject + // regex_filter_tag + vec![SpanBytes { + duration: 10, + resource: "test ignored because of reject regex_filter_tag".into(), + meta: HashMap::from_iter([ + ( + "my_regex_ignore_tag".into(), + "something-true-something".into(), + ), + ("my_require_tag".into(), "true".into()), + ]), + ..Default::default() + }], + // This one gets filtered out because it doesn't have my_require_tag:true + vec![SpanBytes { + duration: 10, + resource: "test ignored because missing a required filter_tag".into(), + meta: HashMap::from_iter([("a_useless_tag".into(), "true".into())]), + ..Default::default() + }], + // This one gets filtered out because it doesn't have my_require_tag:true + vec![SpanBytes { + duration: 10, + resource: "test ignored because wrong value on filter_tag".into(), + meta: HashMap::from_iter([("my_require_tag".into(), "false".into())]), + ..Default::default() + }], + vec![SpanBytes { + duration: 10, + resource: "test2".into(), + meta: HashMap::from_iter([("my_require_tag".into(), "true".into())]), + ..Default::default() + }], + ]) + .as_ref(), + ); + assert!(result.is_err()); + + // Wait for the stats worker to be active before shutting down to avoid potential flaky + // tests on CI where we shutdown before the stats worker had time to start + let start_time = std::time::Instant::now(); + while !exporter.is_stats_worker_active() { + if start_time.elapsed() > Duration::from_secs(10) { + panic!("Timeout waiting for stats worker to become active"); + } + std::thread::sleep(Duration::from_millis(10)); + } + + runtime.shutdown(None).unwrap(); + + // Wait for the mock server to process the stats + for _ in 0..1000 { + if mock_traces.calls() > 0 && mock_stats.calls() > 0 { + break; + } else { + std::thread::sleep(Duration::from_millis(10)); + } + } + + mock_traces.assert(); + mock_stats.assert(); + + // Verify snapshots matches + let mut captured_stats: Vec = captured_stats + .lock() + .unwrap() + .iter() + .map(|payload| rmp_serde::from_slice(payload).unwrap()) + .collect(); + // Sort for deterministic snapshot output + for payload in &mut captured_stats { + for bucket in &mut payload.stats { + bucket.stats.sort_by(|a, b| a.resource.cmp(&b.resource)); + } + } + insta::assert_json_snapshot!( + "trace_filters", + serde_json::to_value(&captured_stats).unwrap(), + { + "[].RuntimeID" => "[id]", + "[].Stats[].Start" => "[timestamp]", + "[].Stats[].Stats[].OkSummary" => "[sketch]", + "[].Stats[].Stats[].ErrorSummary" => "[sketch]", + } + ); + } } diff --git a/libdd-data-pipeline/src/trace_exporter/snapshots/libdd_data_pipeline__trace_exporter__single_threaded_tests__trace_filters.snap b/libdd-data-pipeline/src/trace_exporter/snapshots/libdd_data_pipeline__trace_exporter__single_threaded_tests__trace_filters.snap new file mode 100644 index 0000000000..cbe5725103 --- /dev/null +++ b/libdd-data-pipeline/src/trace_exporter/snapshots/libdd_data_pipeline__trace_exporter__single_threaded_tests__trace_filters.snap @@ -0,0 +1,150 @@ +--- +source: libdd-data-pipeline/src/trace_exporter/mod.rs +expression: "serde_json::to_value(&captured_stats).unwrap()" +--- +[ + { + "Hostname": "", + "Env": "staging", + "Version": "", + "Stats": [ + { + "Start": "[timestamp]", + "Duration": 10000000000, + "Stats": [ + { + "Service": "", + "Name": "", + "Resource": "test", + "HTTPStatusCode": 0, + "Type": "", + "DBType": "", + "Hits": 1, + "Errors": 0, + "Duration": 10, + "OkSummary": "[sketch]", + "ErrorSummary": "[sketch]", + "Synthetics": false, + "TopLevelHits": 1, + "SpanKind": "", + "PeerTags": [], + "IsTraceRoot": 1, + "GRPCStatusCode": "", + "HTTPMethod": "", + "HTTPEndpoint": "", + "srv_src": "", + "SpanDerivedPrimaryTags": [] + }, + { + "Service": "", + "Name": "", + "Resource": "test2", + "HTTPStatusCode": 0, + "Type": "", + "DBType": "", + "Hits": 1, + "Errors": 0, + "Duration": 10, + "OkSummary": "[sketch]", + "ErrorSummary": "[sketch]", + "Synthetics": false, + "TopLevelHits": 1, + "SpanKind": "", + "PeerTags": [], + "IsTraceRoot": 1, + "GRPCStatusCode": "", + "HTTPMethod": "", + "HTTPEndpoint": "", + "srv_src": "", + "SpanDerivedPrimaryTags": [] + } + ], + "AgentTimeShift": 0 + } + ], + "Lang": "", + "TracerVersion": "", + "RuntimeID": "[id]", + "Sequence": 0, + "AgentAggregation": "", + "Service": "test", + "ContainerID": "", + "Tags": [], + "GitCommitSha": "", + "ImageTag": "", + "ProcessTagsHash": 0, + "ProcessTags": "" + }, + { + "Hostname": "", + "Env": "staging", + "Version": "", + "Stats": [ + { + "Start": "[timestamp]", + "Duration": 10000000000, + "Stats": [ + { + "Service": "", + "Name": "", + "Resource": "test", + "HTTPStatusCode": 0, + "Type": "", + "DBType": "", + "Hits": 1, + "Errors": 0, + "Duration": 10, + "OkSummary": "[sketch]", + "ErrorSummary": "[sketch]", + "Synthetics": false, + "TopLevelHits": 1, + "SpanKind": "", + "PeerTags": [], + "IsTraceRoot": 1, + "GRPCStatusCode": "", + "HTTPMethod": "", + "HTTPEndpoint": "", + "srv_src": "", + "SpanDerivedPrimaryTags": [] + }, + { + "Service": "", + "Name": "", + "Resource": "test2", + "HTTPStatusCode": 0, + "Type": "", + "DBType": "", + "Hits": 1, + "Errors": 0, + "Duration": 10, + "OkSummary": "[sketch]", + "ErrorSummary": "[sketch]", + "Synthetics": false, + "TopLevelHits": 1, + "SpanKind": "", + "PeerTags": [], + "IsTraceRoot": 1, + "GRPCStatusCode": "", + "HTTPMethod": "", + "HTTPEndpoint": "", + "srv_src": "", + "SpanDerivedPrimaryTags": [] + } + ], + "AgentTimeShift": 0 + } + ], + "Lang": "", + "TracerVersion": "", + "RuntimeID": "[id]", + "Sequence": 0, + "AgentAggregation": "", + "Service": "test", + "ContainerID": "", + "Tags": [], + "GitCommitSha": "", + "ImageTag": "", + "ProcessTagsHash": 0, + "ProcessTags": "" + } +] diff --git a/libdd-data-pipeline/src/trace_exporter/trace_filter.rs b/libdd-data-pipeline/src/trace_exporter/trace_filter.rs new file mode 100644 index 0000000000..4b9ea4b845 --- /dev/null +++ b/libdd-data-pipeline/src/trace_exporter/trace_filter.rs @@ -0,0 +1,215 @@ +// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 +use std::{str::FromStr, sync::Arc}; + +use libdd_common::regex_engine; +use libdd_trace_stats::span_concentrator::StatSpan; +use libdd_trace_utils::span::trace_utils::get_root_span_index_v4; +use tracing::{debug, error}; + +#[derive(Debug)] +struct TagFilter { + key: String, + value: Option, +} + +#[derive(Debug)] +struct RegexTagFilter { + key: String, + value: Option, +} + +/// Parsed config +#[derive(Debug)] +struct TraceFilteredConf { + reject: Vec, + reject_regex: Vec, + require: Vec, + require_regex: Vec, + ignore_resources: Vec, +} + +#[derive(Debug)] +pub struct TraceFilterer { + conf: arc_swap::ArcSwap, +} + +impl TagFilter { + fn from_str(tag: &str) -> Self { + if let Some((key, value)) = tag.split_once(":") { + TagFilter { + key: key.to_owned(), + value: Some(value.to_owned()), + } + } else { + TagFilter { + key: tag.to_owned(), + value: None, + } + } + } +} + +impl FromStr for RegexTagFilter { + type Err = regex_engine::Error; + + fn from_str(tag: &str) -> Result { + if let Some((key, value)) = tag.split_once(":") { + let regex = match regex_engine::Regex::new(value) { + Ok(regex) => regex, + Err(err) => { + error!( + ?tag, + ?err, + "Invalid regex pattern in tag filter, skipping it" + ); + return Err(err); + } + }; + Ok(RegexTagFilter { + key: key.to_owned(), + value: Some(regex), + }) + } else { + Ok(RegexTagFilter { + key: tag.to_owned(), + value: None, + }) + } + } +} + +impl TraceFilteredConf { + fn parse( + filter_tags: &crate::agent_info::schema::FilterTagsConfig, + filter_tags_regex: &crate::agent_info::schema::FilterTagsConfig, + ignore_resources: &[String], + ) -> Self { + TraceFilteredConf { + reject: filter_tags + .reject + .iter() + .map(|tag| TagFilter::from_str(tag)) + .collect(), + reject_regex: filter_tags_regex + .reject + .iter() + .filter_map(|regex_tag| RegexTagFilter::from_str(regex_tag).ok()) + .collect(), + require: filter_tags + .require + .iter() + .map(|tag| TagFilter::from_str(tag)) + .collect(), + require_regex: filter_tags_regex + .require + .iter() + .filter_map(|regex_tag| RegexTagFilter::from_str(regex_tag).ok()) + .collect(), + ignore_resources: ignore_resources + .iter() + .filter_map(|regex| { + regex_engine::Regex::new(regex) + .inspect_err(|err| { + error!( + ?regex, + ?err, + "Invalid regex pattern in ignore resources filter, skipping it" + ) + }) + .ok() + }) + .collect(), + } + } +} + +impl TraceFilterer { + pub fn new( + filter_tags: &crate::agent_info::schema::FilterTagsConfig, + filter_tags_regex: &crate::agent_info::schema::FilterTagsConfig, + ignore_resources: &[String], + ) -> Self { + let conf = TraceFilteredConf::parse(filter_tags, filter_tags_regex, ignore_resources); + Self { + conf: arc_swap::ArcSwap::from_pointee(conf), + } + } + + pub fn update_conf( + &self, + filter_tags: &crate::agent_info::schema::FilterTagsConfig, + filter_tags_regex: &crate::agent_info::schema::FilterTagsConfig, + ignore_resources: &[String], + ) { + let new_conf = TraceFilteredConf::parse(filter_tags, filter_tags_regex, ignore_resources); + self.conf.swap(Arc::new(new_conf)); + } + + pub fn filter_traces( + &self, + traces: &mut Vec>>, + ) { + traces.retain(|trace| { + let Ok(root_span_index) = get_root_span_index_v4(trace) else { + // FIXME: in this case it's a distributed trace ? Maybe we should remove the debug + // log in get_root_span_index_v4 then + return true; + }; + let root_span = &trace[root_span_index]; + let should_drop = self.should_drop(root_span); + if should_drop { + debug!("Trace rejected as it fails to meet tag requirements. root: %v"); + } + !should_drop + }); + } + + fn should_drop( + &self, + root_span: &libdd_trace_utils::span::v04::Span, + ) -> bool { + let conf = self.conf.load(); + if conf.reject.iter().any(|tag| { + root_span + .get_meta(&tag.key) + .is_some_and(|value| tag.value.as_ref().is_none_or(|v| v == value)) + }) { + return true; + } + + if conf.reject_regex.iter().any(|tag| { + root_span + .get_meta(&tag.key) + .is_some_and(|value| tag.value.as_ref().is_none_or(|pat| pat.is_match(value))) + }) { + return true; + } + + if !conf.require.iter().all(|tag| { + root_span + .get_meta(&tag.key) + .is_some_and(|value| tag.value.as_ref().is_none_or(|v| v == value)) + }) { + return true; + } + + if !conf.require_regex.iter().all(|tag| { + root_span + .get_meta(&tag.key) + .is_some_and(|value| tag.value.as_ref().is_none_or(|pat| pat.is_match(value))) + }) { + return true; + } + + if conf + .ignore_resources + .iter() + .any(|resource_pattern| resource_pattern.is_match(root_span.resource())) + { + return true; + } + + false + } +} diff --git a/libdd-trace-utils/src/span/trace_utils.rs b/libdd-trace-utils/src/span/trace_utils.rs index 8dd8a03b05..17910320f9 100644 --- a/libdd-trace-utils/src/span/trace_utils.rs +++ b/libdd-trace-utils/src/span/trace_utils.rs @@ -3,8 +3,10 @@ //! Trace-utils functionalities implementation for tinybytes based spans +use tracing::debug; + use super::{v04::Span, SpanText, TraceData}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; /// Span metric the mini agent must set for the backend to recognize top level span const TOP_LEVEL_KEY: &str = "_top_level"; @@ -60,6 +62,50 @@ where } } +// FIXME: duplicated with super::get_root_span_index +pub fn get_root_span_index_v4(trace: &[Span]) -> anyhow::Result +where + T: TraceData, +{ + if trace.is_empty() { + anyhow::bail!("Cannot find root span index in an empty trace."); + } + + // Do a first pass to find if we have an obvious root span (starting from the end) since some + // clients put the root span last. + for (i, span) in trace.iter().enumerate().rev() { + if span.parent_id == 0 { + return Ok(i); + } + } + + let span_ids: HashSet<_> = trace.iter().map(|span| span.span_id).collect(); + + let mut root_span_id = None; + for (i, span) in trace.iter().enumerate() { + // If a span's parent is not in the trace, it is a root + if !span_ids.contains(&span.parent_id) { + if root_span_id.is_some() { + debug!( + trace_id = &trace[0].trace_id, + "trace has multiple root spans" + ); + } + root_span_id = Some(i); + } + } + Ok(match root_span_id { + Some(i) => i, + None => { + debug!( + trace_id = &trace[0].trace_id, + "Could not find the root span for trace" + ); + trace.len() - 1 + } + }) +} + /// Return true if the span has a top level key set pub fn has_top_level(span: &Span) -> bool { span.metrics diff --git a/libdd-trace-utils/src/trace_utils.rs b/libdd-trace-utils/src/trace_utils.rs index cd4d3bfb3f..0289b9db61 100644 --- a/libdd-trace-utils/src/trace_utils.rs +++ b/libdd-trace-utils/src/trace_utils.rs @@ -360,10 +360,7 @@ pub fn get_root_span_index(trace: &[pb::Span]) -> anyhow::Result { } } - let mut span_ids: HashSet = HashSet::with_capacity(trace.len()); - for span in trace.iter() { - span_ids.insert(span.span_id); - } + let span_ids: HashSet<_> = trace.iter().map(|span| span.span_id).collect(); let mut root_span_id = None; for (i, span) in trace.iter().enumerate() {