feat(data-pipeline)!: add stdout log trace exporter (#2074)

# What does this PR do?

Adds a stdout "log exporter" transport to libdatadog's trace pipeline. When enabled, `TraceExporter` writes traces as newline-delimited JSON (`{"traces":[[...]]}`) to stdout in the format consumed by the Datadog Forwarder, instead of sending them to an agent over HTTP.

New pieces:
- **`libdd-trace-utils::json_log_encoder`** — Forwarder JSON encoder: lowercase hex IDs (incl. 128-bit), integer `error`, ns `start`/`duration`, empty-field omission, greedy 256 KiB line batching, oversize-span drop.
- **`TraceExporterBuilder::set_output_to_log(max_line_size)`** — selects a stdout destination that bypasses agent-info polling, client-side stats, V1 negotiation, and telemetry. Exposed over FFI as `ddog_trace_exporter_config_set_output_to_log`.
- **`libdd_capabilities::LogWriterCapability`** — the write goes through a capability so the transport also works on wasm (where the host, e.g. JavaScript, supplies the sink). The native capability writes to stdout.

# Motivation

Serverless runtimes (primarily AWS Lambda) often have no reachable Datadog agent or Lambda extension. The established fallback is to write traces to stdout, where the Datadog Forwarder tails the logs and submits them to the trace intake. Today each tracer (dd-trace-js/py/go/java) reimplements this independently; this lands a single reusable implementation in libdatadog that native-backed tracers can adopt.

# Additional Notes

- **Breaking (Rust API):** `TraceExporter<C>` (and `TraceExporterBuilder::build`/`build_async`, `trace_buffer::DefaultExport<C>`) now require `C: LogWriterCapability`. Callers using a custom capability bundle must implement `libdd_capabilities::LogWriterCapability`; `NativeCapabilities` already implements it, so callers using `NativeCapabilities` are unaffected. PR title carries the `!` marker accordingly.
- Selection policy stays with the caller — there is no env-reading helper in libdatadog (an earlier `recommended_log_output()` was removed per review; reading env from libdatadog has caused production crashes). The SDK detects Lambda/agent/extension itself and calls `set_output_to_log`.
- `meta_struct` is intentionally omitted (raw msgpack the log intake can't interpret), matching the reference exporters.
- Writes are **synchronous/blocking** — intended for single-threaded serverless runtimes where there is no shared async reactor to stall. Log output takes precedence over an OTLP endpoint if both are configured.
- No new third-party dependencies (`serde_json` already in tree) → no `Cargo.lock` / `LICENSE-3rdparty.csv` churn.

# How to test the change?

```bash
cargo nextest run -p libdd-trace-utils -p libdd-data-pipeline -p libdd-data-pipeline-ffi -E '!test(tracing_integration_tests::)'
cargo +stable clippy -p libdd-capabilities -p libdd-capabilities-impl -p libdd-trace-utils -p libdd-data-pipeline -p libdd-data-pipeline-ffi --all-targets --all-features -- -D warnings
cargo ffi-test   # FFI crate touched
```

Key tests:
- `json_log_encoder` — golden bytes, Forwarder `is_trace` contract, hex (incl. 128-bit), size-cap batching, oversize-drop, non-finite-metric → null, multi-trace flatten, span_links/span_events present-case.
- `trace_exporter::tests::test_log_mode_send_writes_forwarder_json` — full `send(msgpack)` → decode → log branch → capability bytes (via a capturing test capability).
- `trace_exporter::tests::test_log_mode_makes_no_agent_requests` — zero agent calls + `info_fetcher` not spawned in log mode.
- `log_writer` helper tests; FFI `config_output_to_log_test`.


Co-authored-by: ekump <edmund.kump@datadoghq.com>

Author: bengl

libdd-capabilities-impl/src/lib.rs

@@ -15,7 +15,7 @@ use std::time::Duration;
 
 pub use http::NativeHttpClient;
 use libdd_capabilities::{http::HttpError, MaybeSend};
-pub use libdd_capabilities::{HttpClientCapability, SleepCapability};
+pub use libdd_capabilities::{HttpClientCapability, LogWriterCapability, SleepCapability};
 pub use sleep::NativeSleepCapability;
 
 /// Bundle struct for native platform capabilities.
@@ -64,6 +64,18 @@ impl HttpClientCapability for NativeCapabilities {
     }
 }
 
+impl LogWriterCapability for NativeCapabilities {
+    fn write_log_output(&self, bytes: &[u8]) -> std::io::Result<()> {
+        use std::io::Write;
+        // `Stdout` is internally synchronized; lock once so the whole buffer
+        // (one or more newline-terminated JSON lines) is written without
+        // interleaving, then flush.
+        let mut out = std::io::stdout().lock();
+        out.write_all(bytes)?;
+        out.flush()
+    }
+}
+
 impl SleepCapability for NativeCapabilities {
     fn new() -> Self {
         Self {

libdd-capabilities/src/lib.rs

@@ -4,11 +4,13 @@
 //! Portable capability traits for cross-platform libdatadog.
 
 pub mod http;
+pub mod log_output;
 pub mod maybe_send;
 pub mod sleep;
 pub mod spawn;
 
 pub use self::http::{HttpClientCapability, HttpError};
+pub use self::log_output::LogWriterCapability;
 pub use self::sleep::SleepCapability;
 pub use self::spawn::SpawnError;
 pub use ::http::{Request, Response};

libdd-capabilities/src/log_output.rs

@@ -0,0 +1,19 @@
+// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
+//! Log-output capability trait.
+//!
+//! Lets the trace pipeline emit already-encoded log-exporter output to the
+//! platform's log sink. On native targets this writes to process stdout; a wasm
+//! consumer implements it by handing the bytes to the host (e.g. JavaScript),
+//! since wasm cannot write to stdout directly.
+
+/// Capability for writing encoded log-exporter output to the platform log sink.
+pub trait LogWriterCapability {
+    /// Write a buffer of newline-delimited log output.
+    ///
+    /// `bytes` may contain one or more `\n`-terminated JSON lines. Implementations
+    /// should write the whole buffer (so individual lines are not interleaved with
+    /// other writers) and flush before returning.
+    fn write_log_output(&self, bytes: &[u8]) -> std::io::Result<()>;
+}

libdd-data-pipeline-ffi/src/trace_exporter.rs

@@ -88,6 +88,8 @@ pub struct TraceExporterConfig {
     shared_runtime: Option<Arc<ForkSafeRuntime>>,
     otlp_endpoint: Option<String>,
     otlp_protocol: Option<OtlpProtocol>,
+    output_to_log: bool,
+    log_max_line_size: Option<usize>,
 }
 
 #[no_mangle]
@@ -541,6 +543,37 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_otlp_protocol(
     )
 }
 
+/// Configure the exporter to write traces as newline-delimited JSON to stdout (the Datadog
+/// Forwarder "log exporter" path) instead of sending them to a Datadog agent. Used in serverless
+/// environments (e.g. AWS Lambda) when no agent is reachable.
+///
+/// `max_line_size` overrides the per-line byte cap; pass `0` to use the default (256 KiB, the AWS
+/// CloudWatch Logs limit). When enabled, agent-specific behavior (agent-info polling, client-side
+/// stats, V1 negotiation) is bypassed.
+///
+/// In this mode each span's `meta` is serialized to process stdout (and thus captured by CloudWatch
+/// Logs in Lambda); `meta_struct` is excluded because it holds raw msgpack the log intake cannot
+/// interpret.
+///
+/// Writes are synchronous/blocking on stdout, so this mode targets single-threaded / current-thread
+/// serverless runtimes (e.g. AWS Lambda) where a blocking write won't stall a shared async reactor.
+#[no_mangle]
+pub unsafe extern "C" fn ddog_trace_exporter_config_set_output_to_log(
+    config: Option<&mut TraceExporterConfig>,
+    max_line_size: usize,
+) -> Option<Box<ExporterError>> {
+    catch_panic!(
+        if let Some(handle) = config {
+            handle.output_to_log = true;
+            handle.log_max_line_size = (max_line_size != 0).then_some(max_line_size);
+            None
+        } else {
+            gen_error!(ErrorCode::InvalidArgument)
+        },
+        gen_error!(ErrorCode::Panic)
+    )
+}
+
 /// Create a new TraceExporter instance.
 ///
 /// When an OTLP endpoint is configured via `TraceExporterConfig`, the exporter sends traces to
@@ -613,6 +646,10 @@ pub unsafe extern "C" fn ddog_trace_exporter_new(
                 }
             }
 
+            if config.output_to_log {
+                builder.set_output_to_log(config.log_max_line_size);
+            }
+
             match builder.build() {
                 Ok(exporter) => {
                     out_handle.as_ptr().write(Box::new(exporter));
@@ -708,11 +745,37 @@ mod tests {
             assert!(cfg.process_tags.is_none());
             assert!(cfg.test_session_token.is_none());
             assert!(cfg.connection_timeout.is_none());
+            assert!(!cfg.output_to_log);
+            assert_eq!(cfg.log_max_line_size, None);
 
             ddog_trace_exporter_config_free(cfg);
         }
     }
 
+    #[test]
+    fn config_output_to_log_test() {
+        unsafe {
+            // Null config handle -> InvalidArgument.
+            let error = ddog_trace_exporter_config_set_output_to_log(None, 0);
+            assert_eq!(error.as_ref().unwrap().code, ErrorCode::InvalidArgument);
+            ddog_trace_exporter_error_free(error);
+
+            // 0 is a sentinel for "use the default cap" -> None.
+            let mut config = Some(TraceExporterConfig::default());
+            let error = ddog_trace_exporter_config_set_output_to_log(config.as_mut(), 0);
+            assert_eq!(error, None);
+            let cfg = config.unwrap();
+            assert!(cfg.output_to_log);
+            assert_eq!(cfg.log_max_line_size, None);
+
+            // Non-zero cap is stored as-is.
+            let mut config = Some(TraceExporterConfig::default());
+            let error = ddog_trace_exporter_config_set_output_to_log(config.as_mut(), 4096);
+            assert_eq!(error, None);
+            assert_eq!(config.unwrap().log_max_line_size, Some(4096));
+        }
+    }
+
     #[test]
     fn config_url_test() {
         unsafe {

libdd-data-pipeline/src/trace_buffer/mod.rs

@@ -13,7 +13,7 @@ use std::{
     time::{Duration, Instant},
 };
 
-use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability};
+use libdd_capabilities::{HttpClientCapability, LogWriterCapability, MaybeSend, SleepCapability};
 use libdd_shared_runtime::{SharedRuntime, Worker};
 
 use crate::trace_exporter::{
@@ -645,15 +645,15 @@ pub trait Export<T>: Send + Debug {
 #[derive(Debug)]
 pub struct DefaultExport<C, R>
 where
-    C: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static,
+    C: HttpClientCapability + SleepCapability + LogWriterCapability + MaybeSend + Sync + 'static,
     R: SharedRuntime + std::fmt::Debug + Send + Sync + 'static,
 {
     trace_exporter: TraceExporter<C, R>,
 }
 
 impl<C, R> DefaultExport<C, R>
 where
-    C: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static,
+    C: HttpClientCapability + SleepCapability + LogWriterCapability + MaybeSend + Sync + 'static,
     R: SharedRuntime + std::fmt::Debug + Send + Sync + 'static,
 {
     pub fn new(trace_exporter: TraceExporter<C, R>) -> Self {
@@ -663,7 +663,7 @@ where
 
 impl<C, R> Export<libdd_trace_utils::span::v04::SpanBytes> for DefaultExport<C, R>
 where
-    C: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static,
+    C: HttpClientCapability + SleepCapability + LogWriterCapability + MaybeSend + Sync + 'static,
     R: SharedRuntime + std::fmt::Debug + Send + Sync + 'static,
 {
     fn export_trace_chunks(

libdd-data-pipeline/src/trace_exporter/builder.rs

@@ -8,6 +8,7 @@ use crate::otlp::{OtlpMetricsConfig, OtlpResourceInfo, OtlpTraceConfig};
 use crate::telemetry::TelemetryClientBuilder;
 use crate::trace_exporter::agent_response::AgentResponsePayloadVersion;
 use crate::trace_exporter::error::BuilderErrorKind;
+use crate::trace_exporter::log_writer::DEFAULT_LOG_MAX_LINE_SIZE;
 #[cfg(all(not(target_arch = "wasm32"), feature = "telemetry"))]
 use crate::trace_exporter::TelemetryConfig;
 #[cfg(not(target_arch = "wasm32"))]
@@ -18,7 +19,7 @@ use crate::trace_exporter::{
     TracerMetadata, INFO_ENDPOINT,
 };
 use arc_swap::ArcSwap;
-use libdd_capabilities::{HttpClientCapability, MaybeSend, SleepCapability};
+use libdd_capabilities::{HttpClientCapability, LogWriterCapability, MaybeSend, SleepCapability};
 use libdd_common::{parse_uri, tag, Endpoint};
 use libdd_dogstatsd_client::new;
 use libdd_shared_runtime::SharedRuntime;
@@ -90,6 +91,11 @@ pub struct TraceExporterBuilder<R: SharedRuntime> {
     otlp_metrics_headers: Vec<(String, String)>,
     otel_trace_semantics_enabled: bool,
     runtime_id: Option<String>,
+    /// When true, traces are written as newline-delimited JSON to stdout (the
+    /// Datadog Forwarder "log exporter" path) instead of being sent to an agent.
+    output_to_log: bool,
+    /// Optional override for the maximum size of a single emitted log line.
+    log_max_line_size: Option<usize>,
 }
 
 /// Default is impl'd for `R = ForkSafeRuntime` only so that bare
@@ -149,6 +155,8 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
             otlp_metrics_headers: Vec::new(),
             otel_trace_semantics_enabled: false,
             runtime_id: None,
+            output_to_log: false,
+            log_max_line_size: None,
         }
     }
 }
@@ -425,14 +433,41 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
         self.runtime_id = Some(id.to_owned());
         self
     }
+    /// Configure the exporter to write traces as newline-delimited JSON to stdout
+    /// (the Datadog Forwarder "log exporter" path) instead of sending them to a
+    /// Datadog agent. This is the transport used in serverless environments (e.g.
+    /// AWS Lambda) when no agent is reachable.
+    ///
+    /// `max_line_size` overrides the per-line byte cap; `None` (or `Some(0)`,
+    /// which is coerced to the default) uses the default of 256 KiB, the AWS
+    /// CloudWatch Logs per-event limit. When this is set, agent-specific behavior
+    /// (agent-info polling, client-side stats, V1 negotiation) is bypassed.
+    ///
+    /// In this mode each span's `meta` is serialized to process stdout (and thus
+    /// captured by CloudWatch Logs in Lambda); `meta_struct` is excluded because
+    /// it holds raw msgpack the log intake cannot interpret. Writes are
+    /// synchronous/blocking, so this mode targets single-threaded serverless
+    /// runtimes where blocking stdout writes do not stall a shared async reactor.
+    ///
+    /// Takes precedence over an OTLP endpoint: if both this and `set_otlp_endpoint`
+    /// are configured, traces are written to the log output and not sent via OTLP.
+    pub fn set_output_to_log(&mut self, max_line_size: Option<usize>) -> &mut Self {
+        self.output_to_log = true;
+        // Treat `Some(0)` as "use the default" (a 0 cap would drop every span);
+        // keeps parity with the FFI setter's 0-sentinel.
+        self.log_max_line_size = max_line_size.filter(|&n| n != 0);
+        self
+    }
 
     /// Build the [`TraceExporter`] synchronously.
     ///
     /// Sync facade over [`Self::build_async`]; panics inside an existing tokio context.
     /// Requires `R: BlockingRuntime` so the builder can drive setup on the runtime. Not
     /// available on wasm — use [`Self::build_async`] there.
     #[cfg(not(target_arch = "wasm32"))]
-    pub fn build<C: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static>(
+    pub fn build<
+        C: HttpClientCapability + SleepCapability + LogWriterCapability + MaybeSend + Sync + 'static,
+    >(
         mut self,
     ) -> Result<TraceExporter<C, R>, TraceExporterError>
     where
@@ -459,7 +494,7 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
     /// context. If [`set_shared_runtime`](Self::set_shared_runtime) was not called, a new
     /// runtime is constructed via [`SharedRuntime::new`].
     pub async fn build_async<
-        C: HttpClientCapability + SleepCapability + MaybeSend + Sync + 'static,
+        C: HttpClientCapability + SleepCapability + LogWriterCapability + MaybeSend + Sync + 'static,
     >(
         self,
     ) -> Result<TraceExporter<C, R>, TraceExporterError> {
@@ -506,19 +541,28 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
         let info_endpoint = Endpoint::from_url(add_path(&agent_url, INFO_ENDPOINT));
         let (info_fetcher, info_response_observer) =
             AgentInfoFetcher::<C>::new(info_endpoint, Duration::from_secs(5 * 60));
-        let info_fetcher_handle =
-            shared_runtime
-                .spawn_worker(info_fetcher, false)
-                .map_err(|e| {
-                    TraceExporterError::Builder(BuilderErrorKind::InvalidConfiguration(
-                        e.to_string(),
-                    ))
-                })?;
-        // The handle is currently only tracked for shutdown on native; on wasm
-        // it is dropped here (the worker keeps running on the JS event loop
-        // until the page/module is torn down).
+        // TODO(APMSP-3609): consolidate per-mode worker gating (info-fetcher, telemetry,
+        // stats concentrator) off the selected export destination in one place.
+        // In log-export mode there is no agent to poll; skip spawning the worker
+        // entirely so we don't make repeated failing `/info` calls (e.g. in Lambda).
+        let info_fetcher_handle = if self.output_to_log {
+            None
+        } else {
+            Some(
+                shared_runtime
+                    .spawn_worker(info_fetcher, false)
+                    .map_err(|e| {
+                        TraceExporterError::Builder(BuilderErrorKind::InvalidConfiguration(
+                            e.to_string(),
+                        ))
+                    })?,
+            )
+        };
+        // The handle is only tracked for shutdown on native; on wasm the `workers`
+        // field is cfg'd out, so drop it here (the worker keeps running on the JS
+        // event loop until the page/module is torn down).
         #[cfg(target_arch = "wasm32")]
-        let _ = info_fetcher_handle;
+        drop(info_fetcher_handle);
 
         #[allow(unused_mut)]
         let mut stats = StatsComputationStatus::Disabled;
@@ -530,7 +574,13 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
         #[cfg(all(not(target_arch = "wasm32"), feature = "telemetry"))]
         let (telemetry_client, telemetry_handle) = {
             let sessions = self.telemetry_instrumentation_sessions;
-            let telemetry = self.telemetry.map(|telemetry_config| {
+            // Telemetry talks to the agent; disable it in log-export mode.
+            let telemetry = if self.output_to_log {
+                None
+            } else {
+                self.telemetry
+            }
+            .map(|telemetry_config| {
                 let mut builder = TelemetryClientBuilder::default()
                     .set_language(&self.language)
                     .set_language_version(&self.language_version)
@@ -647,6 +697,10 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
             otlp_stats_enabled = true;
         }
 
+        let log_output = self
+            .output_to_log
+            .then(|| self.log_max_line_size.unwrap_or(DEFAULT_LOG_MAX_LINE_SIZE));
+
         Ok(TraceExporter {
             endpoint: Endpoint {
                 url: agent_url,
@@ -708,6 +762,7 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
             otlp_config,
             trace_filterer: ArcSwap::from_pointee(TraceFilterer::with_empty_conf()),
             otlp_stats_enabled,
+            log_output,
         })
     }
 
@@ -734,6 +789,38 @@ mod tests {
     use libdd_capabilities_impl::NativeCapabilities;
     use libdd_shared_runtime::ForkSafeRuntime;
 
+    #[cfg_attr(miri, ignore)]
+    #[test]
+    fn test_log_output_mode() {
+        let mut builder = TraceExporterBuilder::default();
+        builder
+            .set_service("test")
+            .set_input_format(TraceExporterInputFormat::V04)
+            .set_output_to_log(None);
+        let exporter = builder.build::<NativeCapabilities>().unwrap();
+        // Log-output mode is enabled and the agent-info worker is not spawned.
+        // (End-to-end send -> bytes is covered by the capability-injecting test in
+        // `mod.rs` and the `log_writer` unit tests.)
+        assert!(
+            exporter.log_output.is_some(),
+            "log_output should be set in log-output mode"
+        );
+        assert!(
+            exporter.workers.info_fetcher.is_none(),
+            "no agent-info worker should be spawned in log mode"
+        );
+    }
+
+    #[test]
+    fn set_output_to_log_some_zero_uses_default() {
+        // `Some(0)` is coerced to "use the default cap" (a 0 cap would drop every
+        // span), which is represented as `None` on the builder field.
+        let mut builder = TraceExporterBuilder::default();
+        builder.set_output_to_log(Some(0));
+        assert!(builder.output_to_log);
+        assert_eq!(builder.log_max_line_size, None);
+    }
+
     #[cfg_attr(miri, ignore)]
     #[test]
     fn test_new() {