Add tag filters to /info

Signed-off-by: Bob Weinand <bob.weinand@datadoghq.com>

Author: bwoebi

datadog-ipc/src/platform/mem_handle.rs

@@ -6,8 +6,8 @@ use crate::platform::{mmap_handle, munmap_handle, OwnedFileHandle, PlatformHandl
 #[cfg(feature = "tiny-bytes")]
 use libdd_tinybytes::UnderlyingBytes;
 use serde::{Deserialize, Serialize};
-use std::{ffi::CString, io, ptr::NonNull};
 use std::os::fd::AsRawFd;
+use std::{ffi::CString, io, ptr::NonNull};
 
 #[derive(Clone, Serialize, Deserialize, Debug)]
 pub struct ShmHandle {
@@ -93,7 +93,12 @@ where
         // Use fallocate on Linux to eagerly commit the new pages: ENOSPC at resize time is
         // recoverable; a later SIGBUS mid-execution is not.
         #[cfg(target_os = "linux")]
-        nix::fcntl::fallocate(fd.as_raw_fd(), nix::fcntl::FallocateFlags::empty(), 0, new_size)?;
+        nix::fcntl::fallocate(
+            fd.as_raw_fd(),
+            nix::fcntl::FallocateFlags::empty(),
+            0,
+            new_size,
+        )?;
         #[cfg(not(target_os = "linux"))]
         nix::unistd::ftruncate(&fd, new_size)?;
         Ok(())

datadog-ipc/src/platform/unix/mem_handle.rs

@@ -7,9 +7,9 @@ use crate::platform::{
 use io_lifetimes::OwnedFd;
 use libc::{chmod, off_t};
 use nix::errno::Errno;
-use nix::fcntl::{open, OFlag};
 #[cfg(target_os = "linux")]
 use nix::fcntl::{fallocate, FallocateFlags};
+use nix::fcntl::{open, OFlag};
 use nix::sys::mman::{self, mmap, munmap, MapFlags, ProtFlags};
 use nix::sys::stat::Mode;
 use nix::unistd::{fchown, ftruncate, mkdir, unlink, Uid};

datadog-ipc/src/shm_stats.rs

@@ -853,7 +853,14 @@ mod tests {
         .unwrap();
         c.add_span(&span("svc", "res", 1_000_000));
         c.add_span(&span("svc", "res", 2_000_000));
-        let bytes = c.flush(true, "h", "e", "v", "s", "r");
+        let bytes = c.flush(
+            true,
+            "h".into(),
+            "e".into(),
+            "v".into(),
+            "s".into(),
+            "r".into(),
+        );
         assert!(bytes.is_some());
     }
 
@@ -870,7 +877,14 @@ mod tests {
         .unwrap();
         let worker = ShmSpanConcentrator::open(path.as_c_str()).unwrap();
         worker.add_span(&span("svc2", "res2", 5_000_000));
-        let bytes = creator.flush(true, "h", "", "", "", "r");
+        let bytes = creator.flush(
+            true,
+            "h".into(),
+            "".into(),
+            "".into(),
+            "".into(),
+            "r".into(),
+        );
         assert!(bytes.is_some());
     }
 
@@ -910,7 +924,16 @@ mod tests {
             DEFAULT_STRING_POOL_BYTES,
         )
         .unwrap();
-        assert!(c.flush(false, "h", "e", "v", "s", "r").is_none());
+        assert!(c
+            .flush(
+                false,
+                "h".into(),
+                "e".into(),
+                "v".into(),
+                "s".into(),
+                "r".into()
+            )
+            .is_none());
     }
 
     #[test]

datadog-sidecar/src/service/sidecar_server.rs

@@ -35,7 +35,10 @@ use crate::service::debugger_diagnostics_bookkeeper::{
 };
 use crate::service::exception_hash_rate_limiter::EXCEPTION_HASH_LIMITER;
 use crate::service::remote_configs::{RemoteConfigNotifyTarget, RemoteConfigs};
-use crate::service::stats_flusher::{ensure_stats_concentrator, flush_all_stats_now, get_hostname, stats_endpoint, ConcentratorKey, SpanConcentratorState, StatsConfig};
+use crate::service::stats_flusher::{
+    ensure_stats_concentrator, flush_all_stats_now, get_hostname, stats_endpoint, ConcentratorKey,
+    SpanConcentratorState, StatsConfig,
+};
 use crate::service::tracing::trace_flusher::TraceFlusherStats;
 use crate::tokio_util::run_or_spawn_shared;
 use datadog_live_debugger::sender::{agent_info_supports_debugger_v2_endpoint, DebuggerType};
@@ -681,8 +684,17 @@ impl SidecarInterface for ConnectionSidecarHandler {
         *session.stats_config.lock_or_panic() = Some(StatsConfig {
             endpoint: stats_endpoint(&config.endpoint).unwrap_or_else(|| config.endpoint.clone()),
             flush_interval: config.flush_interval,
-            hostname: if config.hostname.is_empty() { get_hostname() } else { config.hostname.clone() },
-            process_tags: config.process_tags.iter().map(|t| t.to_string()).collect::<Vec<_>>().join(","),
+            hostname: if config.hostname.is_empty() {
+                get_hostname()
+            } else {
+                config.hostname.clone()
+            },
+            process_tags: config
+                .process_tags
+                .iter()
+                .map(|t| t.to_string())
+                .collect::<Vec<_>>()
+                .join(","),
             root_service: config.root_service.clone(),
             language: config.language.clone(),
             tracer_version: config.tracer_version.clone(),
@@ -984,7 +996,11 @@ impl SidecarInterface for ConnectionSidecarHandler {
             .as_ref()
             .map(|c| c.root_service.clone())
             .unwrap_or_default();
-        let map_key = ConcentratorKey { env, version, root_service: service };
+        let map_key = ConcentratorKey {
+            env,
+            version,
+            root_service: service,
+        };
         let guard = self
             .server
             .span_concentrators

datadog-sidecar/src/service/stats_flusher.rs

@@ -11,6 +11,8 @@
 //! triggers creation for a given (env, version, service) is used as the runtime_id in the
 //! stats payload for that key.
 
+use base64::prelude::BASE64_URL_SAFE_NO_PAD;
+use base64::Engine;
 use datadog_ipc::shm_stats::{
     ShmSpanConcentrator, DEFAULT_SLOT_COUNT, DEFAULT_STRING_POOL_BYTES, RELOAD_FILL_RATIO,
 };
@@ -27,8 +29,6 @@ use std::hash::{Hash, Hasher};
 use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering::*};
 use std::sync::{Arc, Mutex, Weak};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
-use base64::Engine;
-use base64::prelude::BASE64_URL_SAFE_NO_PAD;
 use tracing::{error, info, warn};
 use zwohash::ZwoHasher;
 
@@ -55,7 +55,9 @@ pub(crate) fn stats_endpoint(endpoint: &Endpoint) -> Option<Endpoint> {
         return None;
     }
     let mut parts = endpoint.url.clone().into_parts();
-    parts.path_and_query = Some(PathAndQuery::from_static(libdd_trace_stats::stats_exporter::STATS_ENDPOINT_PATH));
+    parts.path_and_query = Some(PathAndQuery::from_static(
+        libdd_trace_stats::stats_exporter::STATS_ENDPOINT_PATH,
+    ));
     Some(Endpoint {
         url: http::Uri::from_parts(parts).ok()?,
         ..endpoint.clone()
@@ -66,7 +68,7 @@ pub(crate) fn stats_endpoint(endpoint: &Endpoint) -> Option<Endpoint> {
 #[derive(Clone)]
 pub(crate) struct StatsConfig {
     /// Stats endpoint with final path already baked in.
-   pub endpoint: Endpoint,
+    pub endpoint: Endpoint,
     pub flush_interval: Duration,
     /// Machine hostname, forwarded to the stats payload `hostname` field.
     pub hostname: String,
@@ -149,20 +151,36 @@ impl SpanConcentratorState {
     /// Used for one-shot flushes (idle-removal, `flush_all_stats_now`).  The retry-accumulator
     /// path in `run_stats_flush_loop` has its own send loop and does not use this.
     async fn send_and_emit(&self, client: &HttpClient, payload: pb::ClientStatsPayload) {
-        let endpoint = self.endpoint.lock().unwrap_or_else(|e| e.into_inner()).clone();
+        let endpoint = self
+            .endpoint
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .clone();
         let spans = spans_in_payload(&payload);
         let buckets = payload.stats.len() as i64;
-        match send_stats(client, &endpoint, &payload, self.language.clone(), self.tracer_version.clone()).await {
-            StatsSendResult::Sent => emit_flush_metrics(&self.dogstatsd, &self.base_tags, spans, 1, buckets, 0),
-            StatsSendResult::Error | StatsSendResult::Network => emit_flush_metrics(&self.dogstatsd, &self.base_tags, 0, 0, 0, 1),
+        match send_stats(
+            client,
+            &endpoint,
+            &payload,
+            self.language.clone(),
+            self.tracer_version.clone(),
+        )
+        .await
+        {
+            StatsSendResult::Sent => {
+                emit_flush_metrics(&self.dogstatsd, &self.base_tags, spans, 1, buckets, 0)
+            }
+            StatsSendResult::Error | StatsSendResult::Network => {
+                emit_flush_metrics(&self.dogstatsd, &self.base_tags, 0, 0, 0, 1)
+            }
         }
     }
 }
 
 /// RAII guard that keeps an (env, version, root-service) concentrator alive.
 ///
-/// Stored in `ActiveApplication`.  When the last guard for a given (env, version, root-service) is dropped,
-/// the flush loop will remove the concentrator after `IDLE_REMOVE_SECS` seconds.
+/// Stored in `ActiveApplication`.  When the last guard for a given (env, version, root-service) is
+/// dropped, the flush loop will remove the concentrator after `IDLE_REMOVE_SECS` seconds.
 pub struct SpanConcentratorGuard {
     ref_count: Arc<AtomicUsize>,
     last_zero_secs: Arc<AtomicU64>,
@@ -198,7 +216,6 @@ pub fn env_stats_shm_path(env: &str, version: &str, service: &str) -> CString {
     CString::new(path).unwrap()
 }
 
-
 /// Result of a single stats payload send attempt.
 #[must_use]
 enum StatsSendResult {
@@ -395,7 +412,15 @@ pub async fn run_stats_flush_loop(
         let mut errors = 0i64;
         let endpoint = s.endpoint.lock().unwrap_or_else(|e| e.into_inner()).clone();
         for p in &pending {
-            match send_stats(&client, &endpoint, &p, s.language.to_owned(), s.tracer_version.to_owned()).await {
+            match send_stats(
+                &client,
+                &endpoint,
+                &p,
+                s.language.to_owned(),
+                s.tracer_version.to_owned(),
+            )
+            .await
+            {
                 StatsSendResult::Sent => {
                     to_drain += 1;
                     payloads_sent += 1;
@@ -413,7 +438,14 @@ pub async fn run_stats_flush_loop(
             }
         }
         pending.drain(..to_drain);
-        emit_flush_metrics(&s.dogstatsd, &s.base_tags, spans_sent, payloads_sent, buckets_sent, errors);
+        emit_flush_metrics(
+            &s.dogstatsd,
+            &s.base_tags,
+            spans_sent,
+            payloads_sent,
+            buckets_sent,
+            errors,
+        );
 
         // Idle-removal check: if no app has held a guard for >= IDLE_REMOVE_SECS, retire this
         // concentrator with a final force-flush.
@@ -428,7 +460,11 @@ pub async fn run_stats_flush_loop(
             };
             let idle_secs = if s.ref_count.load(Acquire) == 0 {
                 let last_zero = s.last_zero_secs.load(Acquire);
-                if last_zero != u64::MAX { now_secs().saturating_sub(last_zero) } else { 0 }
+                if last_zero != u64::MAX {
+                    now_secs().saturating_sub(last_zero)
+                } else {
+                    0
+                }
             } else {
                 0
             };
@@ -455,13 +491,14 @@ pub async fn run_stats_flush_loop(
 /// Create (or look up) the SHM span concentrator for an (env, service, version) pair, increment its
 /// reference count, and return a guard.
 ///
-/// Idempotent with respect to SHM creation: if a concentrator for this (env, service, version) already
-/// exists, only the reference count is incremented.
+/// Idempotent with respect to SHM creation: if a concentrator for this (env, service, version)
+/// already exists, only the reference count is incremented.
 ///
 /// Returns `None` when no `SessionConfig` has been set yet for the calling session (caller should
 /// retry later) or when SHM creation fails.
 ///
-/// - `concentrators`: the global per-(env,version,service) map from `SidecarServer::span_concentrators`
+/// - `concentrators`: the global per-(env,version,service) map from
+///   `SidecarServer::span_concentrators`
 /// - `env`: the environment name
 /// - `version`: the application version
 /// - `service_name`: the root service name reported by `set_universal_service_tags`

libdd-data-pipeline/src/agent_info/schema.rs

@@ -37,6 +37,21 @@ pub struct AgentInfoStruct {
     pub span_kinds_stats_computed: Option<Vec<String>>,
     /// Container tags hash from HTTP response header
     pub container_tags_hash: Option<String>,
+    /// Exact-match tag filters applied before stats computation (root span only).
+    pub filter_tags: Option<FilterTagsConfig>,
+    /// Regex-match tag filters applied before stats computation (root span only).
+    pub filter_tags_regex: Option<FilterTagsConfig>,
+    /// Regex patterns for root-span resource names; matching traces are excluded from stats.
+    pub ignore_resources: Option<Vec<String>>,
+}
+
+/// 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<Vec<String>>,
+    /// If any listed filter matches a root-span tag the trace is rejected.
+    pub reject: Option<Vec<String>>,
 }
 
 #[allow(missing_docs)]