feat(log-agent): add datadog-log-agent crate and wire into serverless-compat

.gitignore

@@ -1,3 +1,4 @@
 /target
 /.idea
+/CLAUDE.md
 

Cargo.toml

@@ -6,6 +6,9 @@ resolver = "2"
 members = [
   "crates/*",
 ]
+exclude = [
+  "crates/.claude",
+]
 
 [workspace.package]
 edition = "2024"

crates/datadog-agent-config/env.rs

@@ -705,6 +705,7 @@ impl ConfigSource for EnvConfigSource {
 
 #[cfg_attr(coverage_nightly, coverage(off))] // Test modules skew coverage metrics
 #[cfg(test)]
+#[allow(clippy::result_large_err)]
 mod tests {
     use std::time::Duration;
 

crates/datadog-agent-config/mod.rs

@@ -831,6 +831,7 @@ where
 
 #[cfg_attr(coverage_nightly, coverage(off))] // Test modules skew coverage metrics
 #[cfg(test)]
+#[allow(clippy::result_large_err)]
 pub mod tests {
     use libdd_trace_obfuscation::replacer::parse_rules_from_string;
 

crates/datadog-agent-config/yaml.rs

@@ -735,6 +735,7 @@ impl ConfigSource for YamlConfigSource {
 
 #[cfg_attr(coverage_nightly, coverage(off))] // Test modules skew coverage metrics
 #[cfg(test)]
+#[allow(clippy::result_large_err)]
 mod tests {
     use std::path::Path;
     use std::time::Duration;

crates/datadog-log-agent/Cargo.toml

@@ -0,0 +1,35 @@
+# Copyright 2025-Present Datadog, Inc. https://www.datadoghq.com/
+# SPDX-License-Identifier: Apache-2.0
+
+[package]
+name = "datadog-log-agent"
+version = "0.1.0"
+edition.workspace = true
+license.workspace = true
+
+[lib]
+bench = false
+
+[dependencies]
+datadog-fips = { path = "../datadog-fips" }
+reqwest = { version = "0.12.4", features = ["json", "http2"], default-features = false }
+serde = { version = "1.0.197", default-features = false, features = ["derive"] }
+serde_json = { version = "1.0.116", default-features = false, features = ["alloc"] }
+thiserror = { version = "1.0.58", default-features = false }
+hyper = { version = "1", features = ["http1", "server"] }
+http-body-util = "0.1"
+hyper-util = { version = "0.1", features = ["tokio"] }
+futures = { version = "0.3", default-features = false, features = ["alloc"] }
+tokio = { version = "1.37.0", default-features = false, features = ["sync", "net"] }
+tracing = { version = "0.1.40", default-features = false }
+zstd = { version = "0.13.3", default-features = false }
+
+[dev-dependencies]
+http = "1"
+mockito = { version = "1.5.0", default-features = false }
+serde_json = { version = "1.0.116", default-features = false, features = ["alloc"] }
+reqwest = { version = "0.12.4", features = ["json"], default-features = false }
+tokio = { version = "1.37.0", default-features = false, features = ["macros", "rt-multi-thread", "net", "time"] }
+
+[features]
+default = []

crates/datadog-log-agent/examples/send_logs.rs

@@ -0,0 +1,196 @@
+// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
+//! Local test helper: inserts sample log entries and flushes them via the log agent pipeline.
+//!
+//! # Usage
+//!
+//! ## Flush to a local capture server (recommended for local dev)
+//!
+//! The easiest way — runs capture server and example together:
+//!   ./scripts/test-log-intake.sh
+//!
+//! Or manually in two terminals:
+//!
+//! In terminal 1 — start the capture server (handles POST, prints JSON):
+//!   python3 scripts/test-log-intake.sh   # not available standalone
+//!   # Use the script above, or run: python3 -c "$(sed -n '/PYEOF/,/PYEOF/p' scripts/test-log-intake.sh)"
+//!
+//! In terminal 2 — run this example:
+//!   DD_OBSERVABILITY_PIPELINES_WORKER_LOGS_ENABLED=true \
+//!   DD_OBSERVABILITY_PIPELINES_WORKER_LOGS_URL=http://localhost:9999/logs \
+//!   DD_API_KEY=local-test-key \
+//!   cargo run -p datadog-log-agent --example send_logs
+//!
+//! NOTE: `python3 -m http.server` does NOT work — it rejects POST requests.
+//!
+//! ## Flush to a real Datadog endpoint
+//!
+//!   DD_API_KEY=<your-key> \
+//!   DD_SITE=datadoghq.com \
+//!   cargo run -p datadog-log-agent --example send_logs
+//!
+//! ## Configuration via env vars
+//!
+//! | Variable                                         | Default            |
+//! |--------------------------------------------------|--------------------|
+//! | DD_API_KEY                                       | (empty)            |
+//! | DD_SITE                                          | datadoghq.com      |
+//! | DD_LOGS_CONFIG_USE_COMPRESSION                   | true               |
+//! | DD_LOGS_CONFIG_COMPRESSION_LEVEL                 | 3                  |
+//! | DD_OBSERVABILITY_PIPELINES_WORKER_LOGS_ENABLED   | false              |
+//! | DD_OBSERVABILITY_PIPELINES_WORKER_LOGS_URL       | (empty)            |
+//! | LOG_ENTRY_COUNT                                  | 5                  |
+
+use datadog_log_agent::{
+    AggregatorService, Destination, IntakeEntry, LogFlusher, LogFlusherConfig,
+};
+
+#[allow(clippy::disallowed_methods)] // plain reqwest::Client for local testing
+#[tokio::main]
+async fn main() {
+    let entry_count: usize = std::env::var("LOG_ENTRY_COUNT")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(5);
+
+    let config = LogFlusherConfig::from_env();
+
+    // Print effective configuration
+    let (endpoint, compressed) = describe_config(&config);
+    println!("──────────────────────────────────────────");
+    println!("  datadog-log-agent local test");
+    println!("──────────────────────────────────────────");
+    println!("  endpoint    : {endpoint}");
+    println!("  api_key     : {}", mask(&config.api_key));
+    println!("  compressed  : {compressed}");
+    println!("  entries     : {entry_count}");
+    println!("──────────────────────────────────────────");
+
+    // Start aggregator service
+    let (service, handle) = AggregatorService::new();
+    tokio::spawn(service.run());
+
+    // Insert sample entries representing different runtimes
+    let mut entries = Vec::with_capacity(entry_count);
+
+    for i in 0..entry_count {
+        let entry = match i % 3 {
+            0 => lambda_entry(i),
+            1 => azure_entry(i),
+            _ => plain_entry(i),
+        };
+        entries.push(entry);
+    }
+
+    println!("\nInserting {entry_count} log entries...");
+    handle.insert_batch(entries).expect("insert_batch failed");
+
+    // Build HTTP client
+    let client = reqwest::Client::builder()
+        .timeout(config.flush_timeout)
+        .build()
+        .expect("failed to build HTTP client");
+
+    // Flush
+    println!("Flushing to {endpoint}...");
+    let flusher = LogFlusher::new(config, client, handle);
+    let failed = flusher.flush(vec![]).await;
+
+    if failed.is_empty() {
+        println!("\n✓ Flush succeeded");
+    } else {
+        eprintln!("\n✗ Flush failed — check endpoint and API key");
+        std::process::exit(1);
+    }
+}
+
+// ── Sample log entry builders ─────────────────────────────────────────────────
+
+fn lambda_entry(i: usize) -> IntakeEntry {
+    let mut attrs = serde_json::Map::new();
+    attrs.insert(
+        "lambda".to_string(),
+        serde_json::json!({
+            "arn": "arn:aws:lambda:us-east-1:123456789012:function:my-fn",
+            "request_id": format!("req-{i:04}")
+        }),
+    );
+    IntakeEntry {
+        message: format!("[lambda] invocation #{i} completed"),
+        timestamp: now_ms(),
+        hostname: Some("arn:aws:lambda:us-east-1:123456789012:function:my-fn".to_string()),
+        service: Some("my-fn".to_string()),
+        ddsource: Some("lambda".to_string()),
+        ddtags: Some("env:local,runtime:lambda".to_string()),
+        status: Some("info".to_string()),
+        attributes: attrs,
+    }
+}
+
+fn azure_entry(i: usize) -> IntakeEntry {
+    let mut attrs = serde_json::Map::new();
+    attrs.insert(
+        "azure".to_string(),
+        serde_json::json!({
+            "resource_id": "/subscriptions/sub-123/resourceGroups/rg/providers/Microsoft.Web/sites/my-fn",
+            "operation_name": "Microsoft.Web/sites/functions/run/action"
+        }),
+    );
+    IntakeEntry {
+        message: format!("[azure] function triggered #{i}"),
+        timestamp: now_ms(),
+        hostname: Some("my-azure-fn".to_string()),
+        service: Some("payments".to_string()),
+        ddsource: Some("azure-functions".to_string()),
+        ddtags: Some("env:local,runtime:azure".to_string()),
+        status: Some("info".to_string()),
+        attributes: attrs,
+    }
+}
+
+fn plain_entry(i: usize) -> IntakeEntry {
+    IntakeEntry {
+        message: format!("[generic] log message #{i}"),
+        timestamp: now_ms(),
+        hostname: Some("localhost".to_string()),
+        service: Some("test-service".to_string()),
+        ddsource: Some("rust".to_string()),
+        ddtags: Some("env:local".to_string()),
+        status: if i.is_multiple_of(5) {
+            Some("error".to_string())
+        } else {
+            Some("info".to_string())
+        },
+        attributes: serde_json::Map::new(),
+    }
+}
+
+// ── Helpers ───────────────────────────────────────────────────────────────────
+
+fn now_ms() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as i64)
+        .unwrap_or(0)
+}
+
+fn describe_config(config: &LogFlusherConfig) -> (String, bool) {
+    match &config.mode {
+        Destination::Datadog => (
+            format!("https://http-intake.logs.{}/api/v2/logs", config.site),
+            config.use_compression,
+        ),
+        Destination::ObservabilityPipelinesWorker { url } => (url.clone(), false),
+    }
+}
+
+fn mask(s: &str) -> String {
+    if s.is_empty() {
+        return "(not set)".to_string();
+    }
+    if s.len() <= 8 {
+        return "*".repeat(s.len());
+    }
+    format!("{}…{}", &s[..4], &s[s.len() - 4..])
+}