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 1bcf391b..b0068fb4 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", @@ -1561,6 +1562,7 @@ checksum = "c8fae54786f62fb2918dcfae3d568594e50eb9b5c25bf04371af6fe7516452fb" name = "inventory" version = "0.1.0" dependencies = [ + "datadog-fips", "httpmock", "libdd-common", "libdd-trace-utils", @@ -1568,6 +1570,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 ad8b4950..2371ff70 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" } +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/datadog-serverless-compat/src/main.rs b/crates/datadog-serverless-compat/src/main.rs index cdcdcabb..3a3b2e99 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 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,22 @@ 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. + 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()); let config = match config::Config::new() { diff --git a/crates/inventory/Cargo.toml b/crates/inventory/Cargo.toml index a86d35d2..5c8a14f2 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", "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..cdc2f986 100644 --- a/crates/inventory/src/lib.rs +++ b/crates/inventory/src/lib.rs @@ -3,9 +3,11 @@ 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; use std::env; #[derive(Clone, Copy)] @@ -27,26 +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, -) -> 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)?; - Ok(Some(InventoryReport { + dd_site: &str, +) -> 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 b6ffc851..865ad060 100644 --- a/crates/inventory/src/payload.rs +++ b/crates/inventory/src/payload.rs @@ -10,20 +10,29 @@ 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_binary_version() -> Option<&'static str> { + option_env!("DD_SERVERLESS_COMPAT_VERSION") + .map(str::trim) + .filter(|value| !value.is_empty()) +} + 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 +63,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 +111,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 +128,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 +144,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 +172,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 new file mode 100644 index 00000000..1bbf7841 --- /dev/null +++ b/crates/inventory/src/reporter.rs @@ -0,0 +1,313 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +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; +use std::time::Duration; +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, + api_key: String, + dd_site: String, + intake_url: reqwest::Url, + user_agent: String, + platform: platform::PlatformData, + 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: Option<&str>, + dd_site: &str, + https_proxy: Option<&str>, + env_type: EnvironmentType, +) { + if !is_inventory_enabled(&ProcessEnv) || !is_supported(&env_type) { + 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 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) => { + warn!("inventory: failed to create HTTP client: {error}"); + return; + } + }; + + let reporter = 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_binary_version().unwrap_or("unknown") + ), + platform, + // 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; + reporter.send_report("periodic").await; + } +} + +fn is_inventory_enabled(env: &impl QueryEnv) -> bool { + env.get_var("DD_SERVERLESS_COMPAT_INVENTORY_ENABLED") + .is_some_and(|value| value.eq_ignore_ascii_case("true")) +} + +fn is_supported(env_type: &EnvironmentType) -> bool { + matches!( + env_type, + EnvironmentType::AzureFunction | EnvironmentType::CloudFunction + ) +} + +impl Reporter { + async fn send_report(&self, report_reason: &str) { + let report = match build_inventory_report( + &self.platform, + &self.process_id, + report_reason, + &self.dd_site, + ) { + Ok(report) => report, + Err(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; + } + } + } + } +} + +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: &reqwest::Url, + api_key: &str, + user_agent: &str, + body: Vec, +) -> Result { + let response = client + .post(url.clone()) + .header("DD-API-KEY", api_key) + .header("Content-Type", "application/json") + .header("User-Agent", user_agent) + .body(body) + .send() + .await?; + Ok(response.status().as_u16()) +} + +fn build_client(https_proxy: Option<&str>) -> Result> { + 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::*; + use crate::test_env::FakeEnv; + + #[test] + fn inventory_gate_is_off_by_default() { + assert!(!is_inventory_enabled(&FakeEnv::default())); + } + + #[test] + 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(&[( + "DD_SERVERLESS_COMPAT_INVENTORY_ENABLED", + "TRUE" + ),]))); + assert!(!is_inventory_enabled(&FakeEnv::new(&[( + "DD_SERVERLESS_COMPAT_INVENTORY_ENABLED", + "1" + ),]))); + } + + #[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)); + } + + #[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; + } +} 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"