From 5257bc5ee07a6c7d389ee18a8e990aacf667017e Mon Sep 17 00:00:00 2001 From: Nina Rei Date: Tue, 29 Sep 2026 11:26:02 -0400 Subject: [PATCH 1/4] refactor(compat-inventory): add reporter integration --- crates/datadog-serverless-compat/Cargo.toml | 1 + crates/datadog-serverless-compat/src/main.rs | 21 ++ crates/inventory/Cargo.toml | 4 +- crates/inventory/src/lib.rs | 3 + crates/inventory/src/reporter.rs | 216 +++++++++++++++++++ 5 files changed, 244 insertions(+), 1 deletion(-) create mode 100644 crates/inventory/src/reporter.rs diff --git a/crates/datadog-serverless-compat/Cargo.toml b/crates/datadog-serverless-compat/Cargo.toml index ad8b4950..3162ceeb 100644 --- a/crates/datadog-serverless-compat/Cargo.toml +++ b/crates/datadog-serverless-compat/Cargo.toml @@ -13,6 +13,7 @@ windows-enhanced-metrics = ["datadog-metrics-collector/windows-enhanced-metrics" [dependencies] datadog-logs-agent = { path = "../datadog-logs-agent" } datadog-metrics-collector = { path = "../datadog-metrics-collector" } +datadog-serverless-compat-inventory = { path = "../datadog-serverless-compat-inventory" } datadog-trace-agent = { path = "../datadog-trace-agent" } libdd-trace-utils = { workspace = true, features = ["https"] } datadog-fips = { path = "../datadog-fips", default-features = false } diff --git a/crates/datadog-serverless-compat/src/main.rs b/crates/datadog-serverless-compat/src/main.rs index cdcdcabb..6aec5074 100644 --- a/crates/datadog-serverless-compat/src/main.rs +++ b/crates/datadog-serverless-compat/src/main.rs @@ -28,6 +28,8 @@ use datadog_metrics_collector::azure_cpu::CpuMetricsCollector; use libdd_trace_utils::{config_utils::read_cloud_env, trace_utils::EnvironmentType}; +use datadog_serverless_compat_inventory::run_inventory_reporter; + use datadog_fips::reqwest_adapter::create_reqwest_client_builder; use datadog_logs_agent::{ AggregatorHandle as LogAggregatorHandle, AggregatorService as LogAggregatorService, @@ -161,6 +163,25 @@ pub async fn main() { debug!("Logging subsystem enabled"); + // Inventory runs independently from traces and metrics so failures cannot + // block mini-agent startup or request handling. + if let Some(api_key) = dd_api_key.clone() { + let inventory_site = dd_site.clone(); + let inventory_proxy = https_proxy.clone(); + let inventory_env = env_type.clone(); + tokio::spawn(async move { + run_inventory_reporter( + &api_key, + &inventory_site, + inventory_proxy.as_deref(), + inventory_env, + ) + .await; + }); + } else { + warn!("DD_API_KEY not set, skipping inventory reporter"); + } + let env_verifier = Arc::new(env_verifier::ServerlessEnvVerifier::default()); let config = match config::Config::new() { diff --git a/crates/inventory/Cargo.toml b/crates/inventory/Cargo.toml index a86d35d2..54b96c35 100644 --- a/crates/inventory/Cargo.toml +++ b/crates/inventory/Cargo.toml @@ -9,12 +9,14 @@ license.workspace = true description = "Fleet Automation inventory reporter for the Serverless Compat" [dependencies] +datadog-fips = { path = "../datadog-fips" } libdd-common = { workspace = true } libdd-trace-utils = { workspace = true } reqwest = { version = "0.12", default-features = false } serde_json = { version = "1.0", default-features = false, features = ["alloc"] } -tokio = { version = "1", features = ["macros"] } +tokio = { version = "1", features = ["macros", "rt", "time"] } tracing = "0.1" +uuid = { version = "1", default-features = false, features = ["v4"] } [dev-dependencies] httpmock = "0.8" diff --git a/crates/inventory/src/lib.rs b/crates/inventory/src/lib.rs index 377653e3..5c49a9b9 100644 --- a/crates/inventory/src/lib.rs +++ b/crates/inventory/src/lib.rs @@ -3,6 +3,9 @@ mod payload; mod platform; +mod reporter; + +pub use reporter::run_inventory_reporter; use libdd_common::azure_app_services::QueryEnv; use libdd_trace_utils::trace_utils::EnvironmentType; diff --git a/crates/inventory/src/reporter.rs b/crates/inventory/src/reporter.rs new file mode 100644 index 00000000..b1b522a1 --- /dev/null +++ b/crates/inventory/src/reporter.rs @@ -0,0 +1,216 @@ +// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use crate::{ProcessEnv, build_inventory_report}; +use datadog_fips::reqwest_adapter::create_reqwest_client_builder; +use libdd_common::azure_app_services::QueryEnv; +use libdd_trace_utils::trace_utils::EnvironmentType; +use std::time::Duration; +use tokio::time::interval; +use tracing::{debug, warn}; + +const INVENTORY_INTERVAL: Duration = Duration::from_secs(30 * 60); +const MAX_ATTEMPTS: u32 = 3; + +/// Runs the inventory reporter for the lifetime of the mini-agent. +/// +/// A report is sent at startup and every 30 minutes afterward. Failures are +/// logged and contained so inventory never blocks agent startup or processing. +pub async fn run_inventory_reporter( + api_key: &str, + dd_site: &str, + https_proxy: Option<&str>, + env_type: EnvironmentType, +) { + if !is_inventory_enabled(&ProcessEnv) || !is_supported(&env_type) { + return; + } + + let client = match build_client(https_proxy) { + Ok(client) => client, + Err(error) => { + warn!("inventory: failed to create HTTP client: {error}"); + return; + } + }; + + // Reuse a stable UUID for every report emitted by this process. + let process_id = uuid::Uuid::new_v4().to_string(); + send_report(&client, api_key, dd_site, &env_type, &process_id, "startup").await; + + let mut ticker = interval(INVENTORY_INTERVAL); + ticker.tick().await; // Consume the immediate first tick. + loop { + ticker.tick().await; + send_report( + &client, + api_key, + dd_site, + &env_type, + &process_id, + "periodic", + ) + .await; + } +} + +fn is_inventory_enabled(env: &impl QueryEnv) -> bool { + env.get_var("DD_SERVERLESS_COMPAT_INVENTORY_ENABLED") + .as_deref() + == Some("true") +} + +fn is_supported(env_type: &EnvironmentType) -> bool { + matches!( + env_type, + EnvironmentType::AzureFunction | EnvironmentType::CloudFunction + ) +} + +async fn send_report( + client: &reqwest::Client, + api_key: &str, + dd_site: &str, + env_type: &EnvironmentType, + process_id: &str, + report_reason: &str, +) { + let report = match build_inventory_report(env_type, process_id, report_reason).await { + Ok(Some(report)) => report, + Ok(None) => { + warn!( + "inventory: required cloud identity unavailable, skipping {report_reason} report" + ); + return; + } + Err(error) => { + warn!("inventory: failed to serialize payload: {error}"); + return; + } + }; + + let url = format!("https://api.{dd_site}/api/v1/metadata"); + for attempt in 0..MAX_ATTEMPTS { + match do_send(client, &url, api_key, report.body.clone()).await { + Ok(status) if status < 300 => { + debug!( + "inventory: report sent (report_reason={report_reason}, workload_type={}, resource_id={}, status={status})", + report.workload_type, report.resource_id + ); + return; + } + Ok(status) if should_retry_status(status) && attempt + 1 < MAX_ATTEMPTS => { + let backoff = retry_backoff(attempt); + warn!( + "inventory: transient failure, retrying in {backoff:?} (report_reason={report_reason}, attempt={}, status={status})", + attempt + 1 + ); + tokio::time::sleep(backoff).await; + } + Ok(status) => { + warn!( + "inventory: intake rejected report (report_reason={report_reason}, workload_type={}, status={status})", + report.workload_type + ); + return; + } + Err(error) if attempt + 1 < MAX_ATTEMPTS => { + let backoff = retry_backoff(attempt); + warn!( + "inventory: transport error, retrying in {backoff:?} (report_reason={report_reason}, attempt={}, error={error})", + attempt + 1 + ); + tokio::time::sleep(backoff).await; + } + Err(error) => { + warn!( + "inventory: transport error after {MAX_ATTEMPTS} attempts (report_reason={report_reason}, error={error})" + ); + return; + } + } + } +} + +fn should_retry_status(status: u16) -> bool { + status == 429 || (500..=599).contains(&status) +} + +fn retry_backoff(attempt: u32) -> Duration { + Duration::from_secs(1 << attempt) +} + +async fn do_send( + client: &reqwest::Client, + url: &str, + api_key: &str, + body: Vec, +) -> Result { + let response = client + .post(url) + .header("DD-API-KEY", api_key) + .header("Content-Type", "application/json") + .header( + "User-Agent", + format!("datadog-serverless-compat/{}", env!("CARGO_PKG_VERSION")), + ) + .body(body) + .send() + .await?; + Ok(response.status().as_u16()) +} + +fn build_client(https_proxy: Option<&str>) -> Result> { + let mut builder = create_reqwest_client_builder()?.timeout(Duration::from_secs(10)); + if let Some(proxy) = https_proxy { + builder = builder.proxy(reqwest::Proxy::https(proxy)?); + } + Ok(builder.build()?) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_env::FakeEnv; + + #[test] + fn inventory_gate_is_off_by_default() { + assert!(!is_inventory_enabled(&FakeEnv::default())); + } + + #[test] + fn inventory_gate_requires_exact_true_value() { + assert!(is_inventory_enabled(&FakeEnv::new(&[( + "DD_SERVERLESS_COMPAT_INVENTORY_ENABLED", + "true" + ),]))); + assert!(!is_inventory_enabled(&FakeEnv::new(&[( + "DD_SERVERLESS_COMPAT_INVENTORY_ENABLED", + "TRUE" + ),]))); + } + + #[test] + fn only_compat_inventory_workloads_are_supported() { + assert!(is_supported(&EnvironmentType::AzureFunction)); + assert!(is_supported(&EnvironmentType::CloudFunction)); + assert!(!is_supported(&EnvironmentType::LambdaFunction)); + assert!(!is_supported(&EnvironmentType::AzureSpringApp)); + } + + #[test] + fn retries_rate_limits_and_server_errors() { + assert!(should_retry_status(429)); + assert!(should_retry_status(500)); + assert!(should_retry_status(599)); + assert!(!should_retry_status(400)); + assert!(!should_retry_status(600)); + } + + #[test] + fn retry_backoff_is_bounded_exponential() { + assert_eq!(retry_backoff(0), Duration::from_secs(1)); + assert_eq!(retry_backoff(1), Duration::from_secs(2)); + assert_eq!(retry_backoff(2), Duration::from_secs(4)); + } +} From ed010c1fe31ec34e91a79e881b9581f51bd6894a Mon Sep 17 00:00:00 2001 From: Nina Rei Date: Tue, 29 Sep 2026 16:40:22 -0400 Subject: [PATCH 2/4] fix(compat): harden inventory reporter --- crates/datadog-serverless-compat/src/main.rs | 29 +-- crates/inventory/src/lib.rs | 3 +- crates/inventory/src/payload.rs | 39 ++- crates/inventory/src/reporter.rs | 260 +++++++++++++------ 4 files changed, 230 insertions(+), 101 deletions(-) diff --git a/crates/datadog-serverless-compat/src/main.rs b/crates/datadog-serverless-compat/src/main.rs index 6aec5074..62260b56 100644 --- a/crates/datadog-serverless-compat/src/main.rs +++ b/crates/datadog-serverless-compat/src/main.rs @@ -165,22 +165,19 @@ pub async fn main() { // Inventory runs independently from traces and metrics so failures cannot // block mini-agent startup or request handling. - if let Some(api_key) = dd_api_key.clone() { - let inventory_site = dd_site.clone(); - let inventory_proxy = https_proxy.clone(); - let inventory_env = env_type.clone(); - tokio::spawn(async move { - run_inventory_reporter( - &api_key, - &inventory_site, - inventory_proxy.as_deref(), - inventory_env, - ) - .await; - }); - } else { - warn!("DD_API_KEY not set, skipping inventory reporter"); - } + let inventory_api_key = dd_api_key.clone(); + let inventory_site = dd_site.clone(); + let inventory_proxy = https_proxy.clone(); + let inventory_env = env_type.clone(); + tokio::spawn(async move { + run_inventory_reporter( + inventory_api_key.as_deref(), + &inventory_site, + inventory_proxy.as_deref(), + inventory_env, + ) + .await; + }); let env_verifier = Arc::new(env_verifier::ServerlessEnvVerifier::default()); diff --git a/crates/inventory/src/lib.rs b/crates/inventory/src/lib.rs index 5c49a9b9..d17b6dca 100644 --- a/crates/inventory/src/lib.rs +++ b/crates/inventory/src/lib.rs @@ -39,12 +39,13 @@ pub async fn build_inventory_report( env_type: &EnvironmentType, process_id: &str, report_reason: &str, + dd_site: &str, ) -> Result, serde_json::Error> { let Some(platform) = platform::collect(env_type).await else { return Ok(None); }; - let body = payload::build(process_id, report_reason, &platform)?; + let body = payload::build(process_id, report_reason, &platform, dd_site)?; Ok(Some(InventoryReport { body, resource_id: platform.resource_id, diff --git a/crates/inventory/src/payload.rs b/crates/inventory/src/payload.rs index a8e6f423..61003c67 100644 --- a/crates/inventory/src/payload.rs +++ b/crates/inventory/src/payload.rs @@ -10,20 +10,41 @@ pub(crate) fn build( process_id: &str, report_reason: &str, platform: &PlatformData, + dd_site: &str, ) -> Result, serde_json::Error> { build_with_env( process_id, report_reason, platform, + dd_site, &ProcessEnv, option_env!("DD_SERVERLESS_COMPAT_VERSION"), ) } +pub(crate) fn serverless_compat_version() -> Option { + serverless_compat_version_with_env(&ProcessEnv, option_env!("DD_SERVERLESS_COMPAT_VERSION")) +} + +fn serverless_compat_version_with_env( + env: &impl QueryEnv, + embedded_version: Option<&str>, +) -> Option { + env.get_var("DD_SERVERLESS_COMPAT_VERSION") + .filter(|value| !value.is_empty()) + .or_else(|| { + embedded_version + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_string) + }) +} + fn build_with_env( process_id: &str, report_reason: &str, platform: &PlatformData, + dd_site: &str, env: &impl QueryEnv, embedded_version: Option<&str>, ) -> Result, serde_json::Error> { @@ -54,12 +75,12 @@ fn build_with_env( metadata["serverless_compat_binary_version"] = Value::String(compat_binary_version.to_string()); } + metadata["dd_site"] = Value::String(dd_site.to_string()); for (env_key, metadata_key) in [ ("DD_ENV", "dd_env"), ("DD_SERVICE", "dd_service"), ("DD_VERSION", "dd_version"), - ("DD_SITE", "dd_site"), ] { if let Some(value) = env.get_var(env_key).filter(|value| !value.is_empty()) { metadata[metadata_key] = Value::String(value); @@ -102,7 +123,8 @@ mod tests { "process-id", "startup", &azure_platform(), - &FakeEnv::default(), + "datadoghq.com", + &FakeEnv::new(&[("DD_SITE", "ignored.example")]), Some("1.2.3"), ) .unwrap(); @@ -118,6 +140,7 @@ mod tests { assert_eq!(payload["agent_metadata"]["workload_type"], "azure_function"); assert_eq!(payload["agent_metadata"]["report_reason"], "startup"); assert_eq!(payload["agent_metadata"]["region"], "eastus"); + assert_eq!(payload["agent_metadata"]["dd_site"], "datadoghq.com"); assert_eq!( payload["agent_metadata"]["serverless_compat_binary_version"], "1.2.3" @@ -133,8 +156,15 @@ mod tests { #[test] fn reports_runtime_compat_and_binary_versions_separately() { let env = FakeEnv::new(&[("DD_SERVERLESS_COMPAT_VERSION", "0.18.0")]); - let body = - build_with_env("pid", "startup", &azure_platform(), &env, Some("0.28.0")).unwrap(); + let body = build_with_env( + "pid", + "startup", + &azure_platform(), + "datadoghq.com", + &env, + Some("0.28.0"), + ) + .unwrap(); let payload: Value = serde_json::from_slice(&body).unwrap(); assert_eq!( @@ -154,6 +184,7 @@ mod tests { "pid", "startup", &azure_platform(), + "datadoghq.com", &FakeEnv::default(), None, ) diff --git a/crates/inventory/src/reporter.rs b/crates/inventory/src/reporter.rs index b1b522a1..2de0d95a 100644 --- a/crates/inventory/src/reporter.rs +++ b/crates/inventory/src/reporter.rs @@ -1,23 +1,33 @@ // Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -use crate::{ProcessEnv, build_inventory_report}; +use crate::{ProcessEnv, build_inventory_report, payload}; use datadog_fips::reqwest_adapter::create_reqwest_client_builder; use libdd_common::azure_app_services::QueryEnv; use libdd_trace_utils::trace_utils::EnvironmentType; use std::time::Duration; -use tokio::time::interval; +use tokio::time::{MissedTickBehavior, interval}; use tracing::{debug, warn}; const INVENTORY_INTERVAL: Duration = Duration::from_secs(30 * 60); const MAX_ATTEMPTS: u32 = 3; +struct Reporter { + client: reqwest::Client, + intake_url: reqwest::Url, + api_key: String, + dd_site: String, + user_agent: String, + env_type: EnvironmentType, + process_id: String, +} + /// Runs the inventory reporter for the lifetime of the mini-agent. /// /// A report is sent at startup and every 30 minutes afterward. Failures are /// logged and contained so inventory never blocks agent startup or processing. pub async fn run_inventory_reporter( - api_key: &str, + api_key: Option<&str>, dd_site: &str, https_proxy: Option<&str>, env_type: EnvironmentType, @@ -26,6 +36,19 @@ pub async fn run_inventory_reporter( return; } + let Some(api_key) = api_key else { + warn!("DD_API_KEY not set, skipping inventory reporter"); + return; + }; + + let intake_url = match build_intake_url(dd_site) { + Ok(url) => url, + Err(error) => { + warn!("inventory: invalid DD_SITE, skipping reporter: {error}"); + return; + } + }; + let client = match build_client(https_proxy) { Ok(client) => client, Err(error) => { @@ -34,30 +57,33 @@ pub async fn run_inventory_reporter( } }; - // Reuse a stable UUID for every report emitted by this process. - let process_id = uuid::Uuid::new_v4().to_string(); - send_report(&client, api_key, dd_site, &env_type, &process_id, "startup").await; + let reporter = Reporter { + client, + intake_url, + api_key: api_key.to_string(), + dd_site: dd_site.to_string(), + user_agent: format!( + "datadog-serverless-compat/{}", + payload::serverless_compat_version() + ), + env_type, + // Reuse a stable UUID for every report emitted by this process. + process_id: uuid::Uuid::new_v4().to_string(), + }; + reporter.send_report("startup").await; let mut ticker = interval(INVENTORY_INTERVAL); + ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); ticker.tick().await; // Consume the immediate first tick. loop { ticker.tick().await; - send_report( - &client, - api_key, - dd_site, - &env_type, - &process_id, - "periodic", - ) - .await; + reporter.send_report("periodic").await; } } fn is_inventory_enabled(env: &impl QueryEnv) -> bool { env.get_var("DD_SERVERLESS_COMPAT_INVENTORY_ENABLED") - .as_deref() - == Some("true") + .is_some_and(|value| value.eq_ignore_ascii_case("true")) } fn is_supported(env_type: &EnvironmentType) -> bool { @@ -67,67 +93,76 @@ fn is_supported(env_type: &EnvironmentType) -> bool { ) } -async fn send_report( - client: &reqwest::Client, - api_key: &str, - dd_site: &str, - env_type: &EnvironmentType, - process_id: &str, - report_reason: &str, -) { - let report = match build_inventory_report(env_type, process_id, report_reason).await { - Ok(Some(report)) => report, - Ok(None) => { - warn!( - "inventory: required cloud identity unavailable, skipping {report_reason} report" - ); - return; - } - Err(error) => { - warn!("inventory: failed to serialize payload: {error}"); - return; - } - }; - - let url = format!("https://api.{dd_site}/api/v1/metadata"); - for attempt in 0..MAX_ATTEMPTS { - match do_send(client, &url, api_key, report.body.clone()).await { - Ok(status) if status < 300 => { - debug!( - "inventory: report sent (report_reason={report_reason}, workload_type={}, resource_id={}, status={status})", - report.workload_type, report.resource_id - ); - return; - } - Ok(status) if should_retry_status(status) && attempt + 1 < MAX_ATTEMPTS => { - let backoff = retry_backoff(attempt); - warn!( - "inventory: transient failure, retrying in {backoff:?} (report_reason={report_reason}, attempt={}, status={status})", - attempt + 1 - ); - tokio::time::sleep(backoff).await; - } - Ok(status) => { +impl Reporter { + async fn send_report(&self, report_reason: &str) { + let report = match build_inventory_report( + &self.env_type, + &self.process_id, + report_reason, + &self.dd_site, + ) + .await + { + Ok(Some(report)) => report, + Ok(None) => { warn!( - "inventory: intake rejected report (report_reason={report_reason}, workload_type={}, status={status})", - report.workload_type + "inventory: required cloud identity unavailable, skipping {report_reason} report" ); return; } - Err(error) if attempt + 1 < MAX_ATTEMPTS => { - let backoff = retry_backoff(attempt); - warn!( - "inventory: transport error, retrying in {backoff:?} (report_reason={report_reason}, attempt={}, error={error})", - attempt + 1 - ); - tokio::time::sleep(backoff).await; - } Err(error) => { - warn!( - "inventory: transport error after {MAX_ATTEMPTS} attempts (report_reason={report_reason}, error={error})" - ); + warn!("inventory: failed to serialize payload: {error}"); return; } + }; + + for attempt in 0..MAX_ATTEMPTS { + match do_send( + &self.client, + &self.intake_url, + &self.api_key, + &self.user_agent, + report.body.clone(), + ) + .await + { + Ok(status) if status < 300 => { + debug!( + "inventory: report sent (report_reason={report_reason}, workload_type={}, resource_id={}, status={status})", + report.workload_type, report.resource_id + ); + return; + } + Ok(status) if should_retry_status(status) && attempt + 1 < MAX_ATTEMPTS => { + let backoff = retry_backoff(attempt); + warn!( + "inventory: transient failure, retrying in {backoff:?} (report_reason={report_reason}, attempt={}, status={status})", + attempt + 1 + ); + tokio::time::sleep(backoff).await; + } + Ok(status) => { + warn!( + "inventory: intake rejected report (report_reason={report_reason}, workload_type={}, status={status})", + report.workload_type + ); + return; + } + Err(error) if attempt + 1 < MAX_ATTEMPTS => { + let backoff = retry_backoff(attempt); + warn!( + "inventory: transport error, retrying in {backoff:?} (report_reason={report_reason}, attempt={}, error={error})", + attempt + 1 + ); + tokio::time::sleep(backoff).await; + } + Err(error) => { + warn!( + "inventory: transport error after {MAX_ATTEMPTS} attempts (report_reason={report_reason}, error={error})" + ); + return; + } + } } } } @@ -142,18 +177,16 @@ fn retry_backoff(attempt: u32) -> Duration { async fn do_send( client: &reqwest::Client, - url: &str, + url: &reqwest::Url, api_key: &str, + user_agent: &str, body: Vec, ) -> Result { let response = client - .post(url) + .post(url.clone()) .header("DD-API-KEY", api_key) .header("Content-Type", "application/json") - .header( - "User-Agent", - format!("datadog-serverless-compat/{}", env!("CARGO_PKG_VERSION")), - ) + .header("User-Agent", user_agent) .body(body) .send() .await?; @@ -161,13 +194,28 @@ async fn do_send( } fn build_client(https_proxy: Option<&str>) -> Result> { - let mut builder = create_reqwest_client_builder()?.timeout(Duration::from_secs(10)); + let mut builder = create_reqwest_client_builder()? + .redirect(reqwest::redirect::Policy::none()) + .timeout(Duration::from_secs(10)); if let Some(proxy) = https_proxy { builder = builder.proxy(reqwest::Proxy::https(proxy)?); } Ok(builder.build()?) } +fn build_intake_url(dd_site: &str) -> Result { + if dd_site.is_empty() + || !dd_site + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b':')) + { + return Err(format!("invalid site: {dd_site}")); + } + + reqwest::Url::parse(&format!("https://api.{dd_site}/api/v1/metadata")) + .map_err(|error| error.to_string()) +} + #[cfg(test)] mod tests { use super::*; @@ -179,15 +227,19 @@ mod tests { } #[test] - fn inventory_gate_requires_exact_true_value() { + fn inventory_gate_accepts_case_insensitive_true() { assert!(is_inventory_enabled(&FakeEnv::new(&[( "DD_SERVERLESS_COMPAT_INVENTORY_ENABLED", "true" ),]))); - assert!(!is_inventory_enabled(&FakeEnv::new(&[( + assert!(is_inventory_enabled(&FakeEnv::new(&[( "DD_SERVERLESS_COMPAT_INVENTORY_ENABLED", "TRUE" ),]))); + assert!(!is_inventory_enabled(&FakeEnv::new(&[( + "DD_SERVERLESS_COMPAT_INVENTORY_ENABLED", + "1" + ),]))); } #[test] @@ -213,4 +265,52 @@ mod tests { assert_eq!(retry_backoff(1), Duration::from_secs(2)); assert_eq!(retry_backoff(2), Duration::from_secs(4)); } + + #[test] + fn validates_inventory_intake_site() { + assert_eq!( + build_intake_url("us5.datadoghq.com").unwrap().as_str(), + "https://api.us5.datadoghq.com/api/v1/metadata" + ); + assert!(build_intake_url("datadoghq.com@attacker.example").is_err()); + assert!(build_intake_url("datadoghq.com/path").is_err()); + } + + #[tokio::test] + async fn sends_required_headers_without_following_redirects() { + use httpmock::prelude::*; + + let server = MockServer::start_async().await; + let target = server + .mock_async(|when, then| { + when.method(POST).path("/target"); + then.status(200); + }) + .await; + let redirect = server + .mock_async(|when, then| { + when.method(POST) + .path("/api/v1/metadata") + .header("DD-API-KEY", "test-key") + .header("Content-Type", "application/json") + .header("User-Agent", "datadog-serverless-compat/1.2.3") + .body("{}"); + then.status(307).header("Location", server.url("/target")); + }) + .await; + + let status = do_send( + &build_client(None).unwrap(), + &reqwest::Url::parse(&server.url("/api/v1/metadata")).unwrap(), + "test-key", + "datadog-serverless-compat/1.2.3", + b"{}".to_vec(), + ) + .await + .unwrap(); + + assert_eq!(status, 307); + redirect.assert_async().await; + target.assert_calls_async(0).await; + } } From 5f38be2d30b55d2e0a02913bd614090c94fb85ed Mon Sep 17 00:00:00 2001 From: Nina Rei Date: Mon, 5 Oct 2026 16:15:25 -0400 Subject: [PATCH 3/4] fix(compat): address remaining inventory reporter feedback --- .../build-datadog-serverless-compat.yml | 12 ++++++++++ .github/workflows/publish.yml | 8 +++---- .../release-datadog-serverless-compat.yml | 11 +++++++-- Cargo.lock | 3 +++ crates/datadog-serverless-compat/Cargo.toml | 2 +- crates/datadog-serverless-compat/src/main.rs | 2 +- crates/inventory/src/reporter.rs | 24 +++++++++---------- scripts/resolve-serverless-compat-version.sh | 23 ++++++++++++++++++ 8 files changed, 63 insertions(+), 22 deletions(-) create mode 100755 scripts/resolve-serverless-compat-version.sh diff --git a/.github/workflows/build-datadog-serverless-compat.yml b/.github/workflows/build-datadog-serverless-compat.yml index 070938fe..b80bcd5e 100644 --- a/.github/workflows/build-datadog-serverless-compat.yml +++ b/.github/workflows/build-datadog-serverless-compat.yml @@ -6,6 +6,9 @@ on: runner: required: true type: string + version: + required: false + type: string jobs: setup: @@ -27,6 +30,15 @@ jobs: runs-on: ${{ inputs.runner }} steps: - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 + - name: Embed Serverless Compat version + shell: bash + env: + INPUT_VERSION: ${{ inputs.version }} + run: | + VERSION="$(./scripts/resolve-serverless-compat-version.sh "$INPUT_VERSION")" + if [[ -n "$VERSION" ]]; then + echo "DD_SERVERLESS_COMPAT_VERSION=$VERSION" >> "$GITHUB_ENV" + fi - name: Install Protoc Binary shell: bash run: chmod +x ./scripts/install-protoc.sh && ./scripts/install-protoc.sh $HOME diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 73954695..e8e30430 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -28,6 +28,7 @@ jobs: uses: ./.github/workflows/build-datadog-serverless-compat.yml with: runner: ${{ matrix.runner }} + version: ${{ inputs.version }} package: runs-on: ubuntu-24.04 @@ -65,11 +66,8 @@ jobs: env: INPUT_VERSION: ${{ inputs.version }} run: | - if [[ "$GITHUB_REF" == refs/tags/datadog-serverless-compat/v* ]]; then - VERSION="${GITHUB_REF#refs/tags/datadog-serverless-compat/v}" - elif [[ -n "$INPUT_VERSION" ]]; then - VERSION="$INPUT_VERSION" - else + VERSION="$(./scripts/resolve-serverless-compat-version.sh "$INPUT_VERSION")" + if [[ -z "$VERSION" ]]; then echo "Error: version input is required when not running on a datadog-serverless-compat/v tag" exit 1 fi diff --git a/.github/workflows/release-datadog-serverless-compat.yml b/.github/workflows/release-datadog-serverless-compat.yml index ae17f253..00fbdbe3 100644 --- a/.github/workflows/release-datadog-serverless-compat.yml +++ b/.github/workflows/release-datadog-serverless-compat.yml @@ -1,6 +1,12 @@ name: Release Datadog Serverless Compat -on: workflow_dispatch +on: + workflow_dispatch: + inputs: + version: + description: "Serverless Compat release version" + required: true + type: string permissions: {} @@ -12,7 +18,8 @@ jobs: runner: [ubuntu-24.04, windows-2022] uses: ./.github/workflows/build-datadog-serverless-compat.yml with: - runner: ${{matrix.runner}} + runner: ${{ matrix.runner }} + version: ${{ inputs.version }} release: runs-on: ubuntu-24.04 needs: build diff --git a/Cargo.lock b/Cargo.lock index 0782b8f5..29bca675 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -646,6 +646,7 @@ dependencies = [ "datadog-metrics-collector", "datadog-trace-agent", "dogstatsd", + "inventory", "libdd-trace-utils", "reqwest", "serde_json", @@ -1554,6 +1555,7 @@ checksum = "c8fae54786f62fb2918dcfae3d568594e50eb9b5c25bf04371af6fe7516452fb" name = "inventory" version = "0.1.0" dependencies = [ + "datadog-fips", "httpmock", "libdd-common", "libdd-trace-utils", @@ -1561,6 +1563,7 @@ dependencies = [ "serde_json", "tokio", "tracing", + "uuid", ] [[package]] diff --git a/crates/datadog-serverless-compat/Cargo.toml b/crates/datadog-serverless-compat/Cargo.toml index 3162ceeb..8e199c76 100644 --- a/crates/datadog-serverless-compat/Cargo.toml +++ b/crates/datadog-serverless-compat/Cargo.toml @@ -13,7 +13,7 @@ windows-enhanced-metrics = ["datadog-metrics-collector/windows-enhanced-metrics" [dependencies] datadog-logs-agent = { path = "../datadog-logs-agent" } datadog-metrics-collector = { path = "../datadog-metrics-collector" } -datadog-serverless-compat-inventory = { path = "../datadog-serverless-compat-inventory" } +inventory = { path = "../datadog-serverless-compat-inventory" } datadog-trace-agent = { path = "../datadog-trace-agent" } libdd-trace-utils = { workspace = true, features = ["https"] } datadog-fips = { path = "../datadog-fips", default-features = false } diff --git a/crates/datadog-serverless-compat/src/main.rs b/crates/datadog-serverless-compat/src/main.rs index 62260b56..3a3b2e99 100644 --- a/crates/datadog-serverless-compat/src/main.rs +++ b/crates/datadog-serverless-compat/src/main.rs @@ -28,7 +28,7 @@ use datadog_metrics_collector::azure_cpu::CpuMetricsCollector; use libdd_trace_utils::{config_utils::read_cloud_env, trace_utils::EnvironmentType}; -use datadog_serverless_compat_inventory::run_inventory_reporter; +use inventory::run_inventory_reporter; use datadog_fips::reqwest_adapter::create_reqwest_client_builder; use datadog_logs_agent::{ diff --git a/crates/inventory/src/reporter.rs b/crates/inventory/src/reporter.rs index 2de0d95a..b60c9806 100644 --- a/crates/inventory/src/reporter.rs +++ b/crates/inventory/src/reporter.rs @@ -1,4 +1,4 @@ -// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 use crate::{ProcessEnv, build_inventory_report, payload}; @@ -14,7 +14,6 @@ const MAX_ATTEMPTS: u32 = 3; struct Reporter { client: reqwest::Client, - intake_url: reqwest::Url, api_key: String, dd_site: String, user_agent: String, @@ -41,14 +40,6 @@ pub async fn run_inventory_reporter( return; }; - let intake_url = match build_intake_url(dd_site) { - Ok(url) => url, - Err(error) => { - warn!("inventory: invalid DD_SITE, skipping reporter: {error}"); - return; - } - }; - let client = match build_client(https_proxy) { Ok(client) => client, Err(error) => { @@ -59,12 +50,11 @@ pub async fn run_inventory_reporter( let reporter = Reporter { client, - intake_url, api_key: api_key.to_string(), dd_site: dd_site.to_string(), user_agent: format!( "datadog-serverless-compat/{}", - payload::serverless_compat_version() + payload::serverless_compat_version().unwrap_or_else(|| "unknown".to_string()) ), env_type, // Reuse a stable UUID for every report emitted by this process. @@ -95,6 +85,14 @@ fn is_supported(env_type: &EnvironmentType) -> bool { impl Reporter { async fn send_report(&self, report_reason: &str) { + let intake_url = match build_intake_url(&self.dd_site) { + Ok(url) => url, + Err(error) => { + warn!("inventory: invalid DD_SITE, skipping {report_reason} report: {error}"); + return; + } + }; + let report = match build_inventory_report( &self.env_type, &self.process_id, @@ -119,7 +117,7 @@ impl Reporter { for attempt in 0..MAX_ATTEMPTS { match do_send( &self.client, - &self.intake_url, + &intake_url, &self.api_key, &self.user_agent, report.body.clone(), diff --git a/scripts/resolve-serverless-compat-version.sh b/scripts/resolve-serverless-compat-version.sh new file mode 100755 index 00000000..5ec7c03e --- /dev/null +++ b/scripts/resolve-serverless-compat-version.sh @@ -0,0 +1,23 @@ +#!/usr/bin/env bash +# Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +# SPDX-License-Identifier: Apache-2.0 + +set -euo pipefail + +input_version="${1:-}" +if [[ "${GITHUB_REF:-}" == refs/tags/datadog-serverless-compat/v* ]]; then + version="${GITHUB_REF#refs/tags/datadog-serverless-compat/v}" +else + version="$input_version" +fi + +if [[ -z "$version" ]]; then + exit 0 +fi + +if [[ ! "$version" =~ ^[0-9]+\.[0-9]+\.[0-9]+(-[0-9A-Za-z.-]+)?(\+[0-9A-Za-z.-]+)?$ ]]; then + echo "Invalid Serverless Compat version: $version" >&2 + exit 1 +fi + +printf '%s\n' "$version" From 7b31be289532de12fe9444e02889c2f568dc0da4 Mon Sep 17 00:00:00 2001 From: Nina Rei Date: Tue, 6 Oct 2026 19:29:14 -0400 Subject: [PATCH 4/4] fix(compat): address final inventory reporter feedback --- crates/datadog-serverless-compat/Cargo.toml | 2 +- crates/inventory/Cargo.toml | 2 +- crates/inventory/src/lib.rs | 25 ++++------- crates/inventory/src/payload.rs | 18 ++------ crates/inventory/src/reporter.rs | 47 ++++++++++----------- 5 files changed, 36 insertions(+), 58 deletions(-) diff --git a/crates/datadog-serverless-compat/Cargo.toml b/crates/datadog-serverless-compat/Cargo.toml index 8e199c76..2371ff70 100644 --- a/crates/datadog-serverless-compat/Cargo.toml +++ b/crates/datadog-serverless-compat/Cargo.toml @@ -13,7 +13,7 @@ windows-enhanced-metrics = ["datadog-metrics-collector/windows-enhanced-metrics" [dependencies] datadog-logs-agent = { path = "../datadog-logs-agent" } datadog-metrics-collector = { path = "../datadog-metrics-collector" } -inventory = { path = "../datadog-serverless-compat-inventory" } +inventory = { path = "../inventory" } datadog-trace-agent = { path = "../datadog-trace-agent" } libdd-trace-utils = { workspace = true, features = ["https"] } datadog-fips = { path = "../datadog-fips", default-features = false } diff --git a/crates/inventory/Cargo.toml b/crates/inventory/Cargo.toml index 54b96c35..5c8a14f2 100644 --- a/crates/inventory/Cargo.toml +++ b/crates/inventory/Cargo.toml @@ -14,7 +14,7 @@ libdd-common = { workspace = true } libdd-trace-utils = { workspace = true } reqwest = { version = "0.12", default-features = false } serde_json = { version = "1.0", default-features = false, features = ["alloc"] } -tokio = { version = "1", features = ["macros", "rt", "time"] } +tokio = { version = "1", features = ["macros", "time"] } tracing = "0.1" uuid = { version = "1", default-features = false, features = ["v4"] } diff --git a/crates/inventory/src/lib.rs b/crates/inventory/src/lib.rs index d17b6dca..cdc2f986 100644 --- a/crates/inventory/src/lib.rs +++ b/crates/inventory/src/lib.rs @@ -8,7 +8,6 @@ mod reporter; pub use reporter::run_inventory_reporter; use libdd_common::azure_app_services::QueryEnv; -use libdd_trace_utils::trace_utils::EnvironmentType; use std::env; #[derive(Clone, Copy)] @@ -30,27 +29,19 @@ pub struct InventoryReport { pub workload_type: &'static str, } -/// Builds an inventory report for a supported serverless environment. -/// -/// Returns `None` when the environment is unsupported or its required cloud -/// identity is unavailable. The async API leaves room for platforms such as -/// GCP that may need metadata-server lookups to complete their identity. -pub async fn build_inventory_report( - env_type: &EnvironmentType, +/// Builds an inventory report from the cloud identity collected at startup. +fn build_inventory_report( + platform: &platform::PlatformData, process_id: &str, report_reason: &str, dd_site: &str, -) -> Result, serde_json::Error> { - let Some(platform) = platform::collect(env_type).await else { - return Ok(None); - }; - - let body = payload::build(process_id, report_reason, &platform, dd_site)?; - Ok(Some(InventoryReport { +) -> Result { + let body = payload::build(process_id, report_reason, platform, dd_site)?; + Ok(InventoryReport { body, - resource_id: platform.resource_id, + resource_id: platform.resource_id.clone(), workload_type: platform.workload_type, - })) + }) } #[cfg(test)] diff --git a/crates/inventory/src/payload.rs b/crates/inventory/src/payload.rs index 61003c67..b5072c8c 100644 --- a/crates/inventory/src/payload.rs +++ b/crates/inventory/src/payload.rs @@ -22,22 +22,10 @@ pub(crate) fn build( ) } -pub(crate) fn serverless_compat_version() -> Option { - serverless_compat_version_with_env(&ProcessEnv, option_env!("DD_SERVERLESS_COMPAT_VERSION")) -} - -fn serverless_compat_version_with_env( - env: &impl QueryEnv, - embedded_version: Option<&str>, -) -> Option { - env.get_var("DD_SERVERLESS_COMPAT_VERSION") +pub(crate) fn serverless_compat_binary_version() -> Option<&'static str> { + option_env!("DD_SERVERLESS_COMPAT_VERSION") + .map(str::trim) .filter(|value| !value.is_empty()) - .or_else(|| { - embedded_version - .map(str::trim) - .filter(|value| !value.is_empty()) - .map(str::to_string) - }) } fn build_with_env( diff --git a/crates/inventory/src/reporter.rs b/crates/inventory/src/reporter.rs index b60c9806..1bbf7841 100644 --- a/crates/inventory/src/reporter.rs +++ b/crates/inventory/src/reporter.rs @@ -1,7 +1,7 @@ // Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -use crate::{ProcessEnv, build_inventory_report, payload}; +use crate::{ProcessEnv, build_inventory_report, payload, platform}; use datadog_fips::reqwest_adapter::create_reqwest_client_builder; use libdd_common::azure_app_services::QueryEnv; use libdd_trace_utils::trace_utils::EnvironmentType; @@ -16,8 +16,9 @@ struct Reporter { client: reqwest::Client, api_key: String, dd_site: String, + intake_url: reqwest::Url, user_agent: String, - env_type: EnvironmentType, + platform: platform::PlatformData, process_id: String, } @@ -40,6 +41,19 @@ pub async fn run_inventory_reporter( return; }; + let intake_url = match build_intake_url(dd_site) { + Ok(url) => url, + Err(error) => { + warn!("inventory: invalid DD_SITE, skipping reporter: {error}"); + return; + } + }; + + let Some(platform) = platform::collect(&env_type).await else { + warn!("inventory: required cloud identity unavailable, skipping reporter"); + return; + }; + let client = match build_client(https_proxy) { Ok(client) => client, Err(error) => { @@ -52,11 +66,12 @@ pub async fn run_inventory_reporter( client, api_key: api_key.to_string(), dd_site: dd_site.to_string(), + intake_url, user_agent: format!( "datadog-serverless-compat/{}", - payload::serverless_compat_version().unwrap_or_else(|| "unknown".to_string()) + payload::serverless_compat_binary_version().unwrap_or("unknown") ), - env_type, + platform, // Reuse a stable UUID for every report emitted by this process. process_id: uuid::Uuid::new_v4().to_string(), }; @@ -85,29 +100,13 @@ fn is_supported(env_type: &EnvironmentType) -> bool { impl Reporter { async fn send_report(&self, report_reason: &str) { - let intake_url = match build_intake_url(&self.dd_site) { - Ok(url) => url, - Err(error) => { - warn!("inventory: invalid DD_SITE, skipping {report_reason} report: {error}"); - return; - } - }; - let report = match build_inventory_report( - &self.env_type, + &self.platform, &self.process_id, report_reason, &self.dd_site, - ) - .await - { - Ok(Some(report)) => report, - Ok(None) => { - warn!( - "inventory: required cloud identity unavailable, skipping {report_reason} report" - ); - return; - } + ) { + Ok(report) => report, Err(error) => { warn!("inventory: failed to serialize payload: {error}"); return; @@ -117,7 +116,7 @@ impl Reporter { for attempt in 0..MAX_ATTEMPTS { match do_send( &self.client, - &intake_url, + &self.intake_url, &self.api_key, &self.user_agent, report.body.clone(),