diff --git a/Cargo.lock b/Cargo.lock index 9fff7ed32ef3..a8bd0dae3f82 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2519,6 +2519,8 @@ dependencies = [ "async-nats 0.49.1", "async-stream", "async-trait", + "aws-config", + "aws-sdk-s3", "axum 0.8.4", "axum-server", "base64 0.22.1", diff --git a/container/README.md b/container/README.md index fa71921ff21f..45ace4179e90 100644 --- a/container/README.md +++ b/container/README.md @@ -484,6 +484,8 @@ container/run.sh --image dynamo:latest-sglang-xpu-local-dev --device=xpu \ sudo chown -R dynamo:0 /opt/miniforge3/envs/sglang cargo build --locked --features dynamo-llm/block-manager --workspace # 3a. ai_dynamo_runtime (Rust bindings: dynamo._core) +# Add `--features request-trace-s3` to enable the S3 request-trace sink +# (DYN_REQUEST_TRACE_SINKS=s3); it is off by default to keep the local build lean. cd lib/bindings/python && maturin develop --uv && cd - # 3b. ai-dynamo (Python namespace packages: dynamo.frontend, dynamo.sglang, ...) uv pip install --no-deps -e /workspace @@ -544,6 +546,8 @@ etcd --listen-client-urls http://0.0.0.0:2379 --advertise-client-urls http://0.0 # 4. Compile code cargo build --locked --features dynamo-llm/block-manager --workspace +# Add `--features request-trace-s3` to enable the S3 request-trace sink +# (DYN_REQUEST_TRACE_SINKS=s3); it is off by default to keep the local build lean. cd lib/bindings/python && maturin develop --uv && cd - # 5. Sanity check (optional but recommended) diff --git a/container/templates/wheel_builder.Dockerfile b/container/templates/wheel_builder.Dockerfile index 6a56c38f9e74..7ea02cbf9aa6 100644 --- a/container/templates/wheel_builder.Dockerfile +++ b/container/templates/wheel_builder.Dockerfile @@ -524,9 +524,9 @@ RUN --mount=type=secret,id=aws-web-identity-token,target=/run/secrets/aws-token uv build --wheel --out-dir /opt/dynamo/dist && \ cd /opt/dynamo/lib/bindings/python && \ if [ "$ENABLE_MEDIA_FFMPEG" = "true" ]; then \ - maturin build --release --features "media-ffmpeg,kv-indexer,slot-tracker,select-service,mm-routing,aic-forward-pass{% if target == "planner" %},mocker-kvbm-offload{% endif %}" --out /opt/dynamo/dist; \ + maturin build --release --features "media-ffmpeg,kv-indexer,slot-tracker,select-service,mm-routing,aic-forward-pass,request-trace-s3{% if target == "planner" %},mocker-kvbm-offload{% endif %}" --out /opt/dynamo/dist; \ else \ - maturin build --release --features "kv-indexer,slot-tracker,select-service,mm-routing,aic-forward-pass{% if target == "planner" %},mocker-kvbm-offload{% endif %}" --out /opt/dynamo/dist; \ + maturin build --release --features "kv-indexer,slot-tracker,select-service,mm-routing,aic-forward-pass,request-trace-s3{% if target == "planner" %},mocker-kvbm-offload{% endif %}" --out /opt/dynamo/dist; \ fi && \ /tmp/use-sccache.sh show-stats "Dynamo Runtime" diff --git a/docs/fern/reference/observability/environment-variables.mdx b/docs/fern/reference/observability/environment-variables.mdx index e1eecd569845..e265d9e5adb0 100644 --- a/docs/fern/reference/observability/environment-variables.mdx +++ b/docs/fern/reference/observability/environment-variables.mdx @@ -275,7 +275,7 @@ See [Forward Pass Metrics Trace Reference](forward-pass-metrics-tracing.mdx) for - Comma-separated sinks: `file`, `stderr`, `nats`, and `otel`. + Comma-separated sinks: `file`, `stderr`, `nats`, `otel`, and `s3`. @@ -314,6 +314,26 @@ See [Forward Pass Metrics Trace Reference](forward-pass-metrics-tracing.mdx) for Optional gzip roll threshold in records. + + Destination bucket for the `s3` sink. Required when `DYN_REQUEST_TRACE_SINKS` includes `s3`; startup fails if unset. + + + + Region override for the `s3` sink. When unset the AWS SDK resolves the region from the environment, profile, or IMDS. + + + + Object key prefix for the `s3` sink. When unset, records land at the bucket root. Keys are `{prefix}/{yyyy}/{mm}/{dd}/{host}-{HHMMSS}-{run_id}-{seq}.jsonl.gz`. + + + + Batch roll threshold for the `s3` sink in uncompressed bytes. When the pending batch reaches this size it is gzipped and uploaded as one object. + + + + Periodic flush interval for the `s3` sink in milliseconds. A partial batch is uploaded when this elapses so low-volume traces still reach S3. + + Optional ZMQ PULL bind address for harness tool events. Configure it on only one process. diff --git a/docs/fern/reference/observability/request-tracing.mdx b/docs/fern/reference/observability/request-tracing.mdx index 28a8ae7b2085..17bf4dd458b2 100644 --- a/docs/fern/reference/observability/request-tracing.mdx +++ b/docs/fern/reference/observability/request-tracing.mdx @@ -46,6 +46,30 @@ transport settings in this order: Setting `DYN_REQUEST_TRACE_SINKS=stderr` does not enable OTLP export. Include `otel`, for example `file,otel` or `stderr,otel`. +The `s3` sink writes records directly to an S3 bucket as gzipped JSONL objects, one object per rolled +batch. Records are batched in-process and uploaded when the batch reaches +`DYN_REQUEST_TRACE_S3_ROLL_UNCOMPRESSED_BYTES` or when `DYN_REQUEST_TRACE_S3_FLUSH_INTERVAL_MS` +elapses. Object keys are `{prefix}/{yyyy}/{mm}/{dd}/{host}-{HHMMSS}-{run_id}-{seq}.jsonl.gz`, where +`run_id` is a per-process UUID that keeps container restarts and hostname collisions from overwriting +earlier batches. Credentials come from the AWS SDK default provider chain (environment variables, +IMDS, IRSA, Pod Identity, shared profiles); how the frontend pod is credentialed is a deployment +concern. On terminal upload failure the batch is dropped and a warning is logged; no on-disk retry +queue is kept. + +```bash +export DYN_REQUEST_TRACE=1 +export DYN_REQUEST_TRACE_SINKS=s3 +export DYN_REQUEST_TRACE_S3_BUCKET=my-org-request-traces +export DYN_REQUEST_TRACE_S3_REGION=us-west-2 +export DYN_REQUEST_TRACE_S3_PREFIX=frontend-a/prod +``` + + +The `s3` sink is compiled in only when `dynamo-llm` is built with the `request-trace-s3` cargo +feature. Shipped Dynamo Python wheels enable it; local source builds pass +`--features request-trace-s3` to `maturin develop`. + + ## Record Types ### `request_end` diff --git a/lib/bindings/python/Cargo.lock b/lib/bindings/python/Cargo.lock index ef7ab4f8c82e..3f1ec56e90ca 100644 --- a/lib/bindings/python/Cargo.lock +++ b/lib/bindings/python/Cargo.lock @@ -2216,6 +2216,8 @@ dependencies = [ "async-nats 0.49.1", "async-stream", "async-trait", + "aws-config", + "aws-sdk-s3", "axum", "axum-server", "base64 0.22.1", diff --git a/lib/bindings/python/Cargo.toml b/lib/bindings/python/Cargo.toml index 629bab9a8b0d..8b80aa9f163c 100644 --- a/lib/bindings/python/Cargo.toml +++ b/lib/bindings/python/Cargo.toml @@ -42,6 +42,11 @@ nvtx = ["dynamo-runtime/nvtx"] # recipes pass this flag explicitly; the dev container's # `.devcontainer/post-create.sh` also enables it for local workflows. mm-routing = ["dynamo-llm/mm-routing"] +# Enable the native S3 destination for the request-trace sink. Pulls in +# aws-sdk-s3 + aws-config so the default wheel stays lean; production wheel +# builds enable this flag in container/templates/wheel_builder.Dockerfile so +# `DYN_REQUEST_TRACE_SINKS=s3` works on shipped artifacts. +request-trace-s3 = ["dynamo-llm/request-trace-s3"] [dependencies] # AIC perf model: pure-Rust hot-path latency engine. Pinned to the matching diff --git a/lib/llm/Cargo.toml b/lib/llm/Cargo.toml index be5eb3b2c8d2..b027973ddbd4 100644 --- a/lib/llm/Cargo.toml +++ b/lib/llm/Cargo.toml @@ -35,6 +35,8 @@ ckf-diagnostics = [] kv-router-stress = ["dep:clap", "dep:indicatif", "bench"] mm-routing = ["dep:llm-multimodal", "dep:llm-tokenizer"] request-trace-bench = [] +# S3 destination for the request-trace sink. Pulls in aws-sdk-s3 + aws-config. +request-trace-s3 = ["dep:aws-sdk-s3", "dep:aws-config"] [[bench]] name = "tokenizer_simple" @@ -142,6 +144,10 @@ nix = { version = "0.26", optional = true } # media (zlib compression for NIXL metadata) flate2 = { version = "1" } +# request-trace-s3 (optional S3 sink) +aws-sdk-s3 = { version = "1.120.0", optional = true } +aws-config = { version = "1.8.11", optional = true } + # block_manager_bench clap = { version = "4.5.49", features = ["derive"], optional = true } indicatif = { version = "0.18.0", optional = true } diff --git a/lib/llm/src/request_trace/config.rs b/lib/llm/src/request_trace/config.rs index db0209923e1c..6772714c0979 100644 --- a/lib/llm/src/request_trace/config.rs +++ b/lib/llm/src/request_trace/config.rs @@ -21,6 +21,8 @@ const DEFAULT_FILE_PATH: &str = "/tmp/dynamo-request-trace"; const DEFAULT_NATS_SUBJECT: &str = "dynamo.request_trace.v1"; const DEFAULT_LEGACY_AUDIT_NATS_SUBJECT: &str = "dynamo.audit.v1"; const DEFAULT_OTEL_MAX_PAYLOAD_BYTES: usize = 4 * 1024 * 1024; +const DEFAULT_S3_ROLL_UNCOMPRESSED_BYTES: u64 = 64 * 1024 * 1024; +const DEFAULT_S3_FLUSH_INTERVAL_MS: u64 = 10_000; const CAPTURE_UNINITIALIZED: u8 = 0; const CAPTURE_ACTIVE: u8 = 1; @@ -32,6 +34,7 @@ pub enum RequestTraceSinkKind { Stderr, Nats, Otel, + S3, } impl RequestTraceSinkKind { @@ -41,6 +44,7 @@ impl RequestTraceSinkKind { Self::Stderr => "stderr", Self::Nats => "nats", Self::Otel => "otel", + Self::S3 => "s3", } } } @@ -94,6 +98,11 @@ pub struct RequestTracePolicy { pub http_header_capture_list: Vec, pub tool_events_zmq_endpoint: Option, pub tool_events_zmq_topic: Option, + pub s3_bucket: Option, + pub s3_region: Option, + pub s3_prefix: Option, + pub s3_roll_uncompressed_bytes: u64, + pub s3_flush_interval_ms: u64, } impl RequestTracePolicy { @@ -214,6 +223,17 @@ fn load_from_env() -> RequestTracePolicy { .filter(|value| !value.is_empty()) .unwrap_or_else(|| DEFAULT_TOOL_EVENTS_TOPIC.to_string()) }); + let s3_bucket = env_trimmed(env_request_trace::DYN_REQUEST_TRACE_S3_BUCKET); + let s3_region = env_trimmed(env_request_trace::DYN_REQUEST_TRACE_S3_REGION); + let s3_prefix = env_trimmed(env_request_trace::DYN_REQUEST_TRACE_S3_PREFIX); + let s3_roll_uncompressed_bytes = + env_u64(&[env_request_trace::DYN_REQUEST_TRACE_S3_ROLL_UNCOMPRESSED_BYTES]) + .filter(|value| *value > 0) + .unwrap_or(DEFAULT_S3_ROLL_UNCOMPRESSED_BYTES); + let s3_flush_interval_ms = + env_u64(&[env_request_trace::DYN_REQUEST_TRACE_S3_FLUSH_INTERVAL_MS]) + .filter(|value| *value > 0) + .unwrap_or(DEFAULT_S3_FLUSH_INTERVAL_MS); RequestTracePolicy { enabled, @@ -231,6 +251,11 @@ fn load_from_env() -> RequestTracePolicy { http_header_capture_list, tool_events_zmq_endpoint, tool_events_zmq_topic, + s3_bucket, + s3_region, + s3_prefix, + s3_roll_uncompressed_bytes, + s3_flush_interval_ms, } } @@ -314,6 +339,7 @@ fn parse_sink_kind_names( "stderr" => push_sink(&mut sinks, RequestTraceSinkKind::Stderr), "nats" => push_sink(&mut sinks, RequestTraceSinkKind::Nats), "otel" => push_sink(&mut sinks, RequestTraceSinkKind::Otel), + "s3" => push_sink(&mut sinks, RequestTraceSinkKind::S3), "jsonl" => { legacy_jsonl = true; push_sink(&mut sinks, RequestTraceSinkKind::File); diff --git a/lib/llm/src/request_trace/mod.rs b/lib/llm/src/request_trace/mod.rs index 8f09220006e9..36b841b2af5d 100644 --- a/lib/llm/src/request_trace/mod.rs +++ b/lib/llm/src/request_trace/mod.rs @@ -9,6 +9,8 @@ pub mod payload; pub(crate) mod payload_stream; mod record; mod replay; +#[cfg(feature = "request-trace-s3")] +mod s3_sink; pub mod sink; mod tool_relay; pub mod types; diff --git a/lib/llm/src/request_trace/s3_sink.rs b/lib/llm/src/request_trace/s3_sink.rs new file mode 100644 index 000000000000..a4964eef829a --- /dev/null +++ b/lib/llm/src/request_trace/s3_sink.rs @@ -0,0 +1,562 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! S3 destination for request trace records. +//! +//! Records are batched in-process as gzipped JSONL, and each finished batch is +//! uploaded as one object via `PutObject`. Object keys use a simple time-based +//! layout for PR 1 +//! (`{prefix}/{yyyy}/{mm}/{dd}/{host}-{HHMMSS}-{run_id}-{seq}.jsonl.gz`); +//! richer partitioning ships in a follow-up. +//! +//! Credentials come from the AWS SDK default provider chain — env vars, IMDS, +//! IRSA, Pod Identity, and shared profiles are all handled by the SDK. How the +//! frontend pod is credentialed is a deployment concern, not this sink's. +//! +//! # Upload concurrency +//! +//! A single worker task drains the record channel and uploads each finished +//! batch inline (same shape as the OpenTelemetry Rust `BatchLogProcessor` and +//! the local `telemetry::jsonl_gz` writer). While an upload is in flight the +//! worker is not draining, so a slow `PutObject` applies backpressure and +//! `emit` drops records once the channel fills. Drops are counted and surfaced +//! (see [`S3RequestTraceSink::emit`]). Overlapping uploads with a bounded +//! concurrency pool so the drain never stalls is a follow-up; it belongs with +//! the object-layout work where the sink is restructured to carry more context. + +use std::io::Write; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use anyhow::{Context as _, Result}; +use async_trait::async_trait; +use aws_config::{BehaviorVersion, Region, timeout::TimeoutConfig}; +use aws_sdk_s3::primitives::ByteStream; +use flate2::{Compression, write::GzEncoder}; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; +use uuid::Uuid; + +use dynamo_runtime::config::environment_names::llm::request_trace as env_request_trace; + +use super::RequestTraceRecord; +use super::config::RequestTracePolicy; +use super::sink::RequestTraceSink; + +const CHANNEL_CAPACITY: usize = 2048; +const DEFAULT_BUFFER_INITIAL_BYTES: usize = 256 * 1024; +// Bound S3 upload duration so a stalled endpoint or slow network cannot wedge +// the worker task indefinitely. `attempt_timeout` covers a single HTTP attempt; +// `operation_timeout` bounds the full call including SDK retries (three total +// by default). After the operation timeout expires the batch is discarded with +// a warning; a persistent retry queue is deferred to a follow-up PR. +const S3_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(30); +const S3_OPERATION_TIMEOUT: Duration = Duration::from_secs(90); + +pub struct S3RequestTraceSink { + tx: mpsc::Sender, + shutdown: CancellationToken, + worker: Mutex>>, + /// Count of records dropped by `emit` because the batcher channel was full + /// or closed. Surfaced once on the first drop and again as a summary at + /// shutdown, so a slow S3 does not produce one log line per dropped record. + dropped: Arc, +} + +#[derive(Clone)] +struct S3UploadOptions { + bucket: String, + prefix: String, + host: String, + /// Per-process UUID mixed into every object key so that a pod restart + /// (which reuses hostname and can land within the same second) can't + /// overwrite a previous batch, and two frontends sharing a hostname + /// stay disjoint. + run_id: String, +} + +impl S3RequestTraceSink { + pub async fn from_policy(policy: &RequestTracePolicy) -> Result { + let bucket = policy.s3_bucket.clone().ok_or_else(|| { + anyhow::anyhow!( + "{} must be set when {} includes s3", + env_request_trace::DYN_REQUEST_TRACE_S3_BUCKET, + env_request_trace::DYN_REQUEST_TRACE_SINKS, + ) + })?; + let prefix = policy.s3_prefix.clone().unwrap_or_default(); + let host = hostname_or_fallback(); + let run_id = Uuid::new_v4().simple().to_string(); + let roll_uncompressed_bytes = policy.s3_roll_uncompressed_bytes; + let flush_interval = Duration::from_millis(policy.s3_flush_interval_ms.max(1)); + + let timeout_config = TimeoutConfig::builder() + .operation_attempt_timeout(S3_ATTEMPT_TIMEOUT) + .operation_timeout(S3_OPERATION_TIMEOUT) + .build(); + let mut loader = + aws_config::defaults(BehaviorVersion::latest()).timeout_config(timeout_config); + if let Some(region) = policy.s3_region.clone() { + loader = loader.region(Region::new(region)); + } + let sdk_config = loader.load().await; + let client = aws_sdk_s3::Client::new(&sdk_config); + + let (tx, rx) = mpsc::channel(CHANNEL_CAPACITY); + let shutdown = CancellationToken::new(); + let upload_options = S3UploadOptions { + bucket, + prefix, + host, + run_id, + }; + let worker_shutdown = shutdown.clone(); + let worker = tokio::spawn(async move { + run_worker( + client, + upload_options, + rx, + worker_shutdown, + roll_uncompressed_bytes, + flush_interval, + ) + .await; + }); + + Ok(Self { + tx, + shutdown, + worker: Mutex::new(Some(worker)), + dropped: Arc::new(AtomicU64::new(0)), + }) + } + + /// Record one dropped record and, only on the first drop, emit a single + /// warning. Subsequent drops bump the counter silently; the running total + /// is reported once at shutdown. Mirrors the OpenTelemetry Rust + /// `BatchLogProcessor` drop-accounting pattern so a degraded S3 endpoint + /// cannot flood the log with one line per dropped record. Returns `true` + /// when this call emitted the warning (i.e. it was the first drop). + fn note_dropped(&self, reason: &str) -> bool { + if self.dropped.fetch_add(1, Ordering::Relaxed) == 0 { + tracing::warn!( + target: "dynamo_llm::request_trace", + reason, + "request trace s3: dropping records (batcher backpressure); \ + further drops are counted and summarized at shutdown" + ); + true + } else { + false + } + } +} + +#[async_trait] +impl RequestTraceSink for S3RequestTraceSink { + fn name(&self) -> &'static str { + "s3" + } + + async fn emit(&self, record: &RequestTraceRecord) { + if let Err(error) = self.tx.try_send(record.clone()) { + let reason = match error { + mpsc::error::TrySendError::Full(_) => "channel_full", + mpsc::error::TrySendError::Closed(_) => "channel_closed", + }; + self.note_dropped(reason); + } + } + + async fn shutdown(&self) { + self.shutdown.cancel(); + // Recover the guard even if a prior panic poisoned the lock; it only + // guards an `Option`, so a poisoned lock must not turn + // teardown into a second panic. + let worker = self + .worker + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take(); + if let Some(worker) = worker + && let Err(error) = worker.await + { + tracing::warn!( + target: "dynamo_llm::request_trace", + error = %error, + "request trace s3: batcher task join failed during shutdown" + ); + } + let dropped = self.dropped.load(Ordering::Relaxed); + if dropped > 0 { + tracing::warn!( + target: "dynamo_llm::request_trace", + dropped, + "request trace s3: dropped records during the run (batcher backpressure)" + ); + } + } +} + +async fn run_worker( + client: aws_sdk_s3::Client, + options: S3UploadOptions, + mut rx: mpsc::Receiver, + shutdown: CancellationToken, + roll_uncompressed_bytes: u64, + flush_interval: Duration, +) { + let uploader = Arc::new(S3Uploader { client, options }); + let mut batch = JsonlBatch::new(); + let mut seq: u64 = 0; + let mut flush_tick = tokio::time::interval(flush_interval); + flush_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + // Skip the immediate tick that `interval` fires at t=0. + flush_tick.tick().await; + + loop { + tokio::select! { + biased; + _ = shutdown.cancelled() => { + // Close the receiver first so an in-flight `emit()` cannot + // land a record after we start draining. Then `recv()` yields + // every already-enqueued record and returns `None` once empty. + rx.close(); + while let Some(record) = rx.recv().await { + if let Err(error) = batch.push(&record) { + tracing::warn!( + target: "dynamo_llm::request_trace", + %error, + "request trace s3: serialize failed during shutdown" + ); + } else if batch.uncompressed_bytes() >= roll_uncompressed_bytes { + // Enforce the roll threshold during shutdown too, so a + // full channel can't collapse into one oversized PUT + // that loses everything on a single upload failure. + upload_ready_batch(&uploader, &mut batch, &mut seq).await; + } + } + if !batch.is_empty() { + upload_ready_batch(&uploader, &mut batch, &mut seq).await; + } + return; + } + _ = flush_tick.tick() => { + if !batch.is_empty() { + upload_ready_batch(&uploader, &mut batch, &mut seq).await; + } + } + message = rx.recv() => { + match message { + Some(record) => { + if let Err(error) = batch.push(&record) { + tracing::warn!( + target: "dynamo_llm::request_trace", + %error, + "request trace s3: serialize failed; dropping record" + ); + } else if batch.uncompressed_bytes() >= roll_uncompressed_bytes { + upload_ready_batch(&uploader, &mut batch, &mut seq).await; + } + } + None => { + if !batch.is_empty() { + upload_ready_batch(&uploader, &mut batch, &mut seq).await; + } + return; + } + } + } + } + } +} + +async fn upload_ready_batch(uploader: &Arc, batch: &mut JsonlBatch, seq: &mut u64) { + let ready = match batch.take_finished().await { + Ok(bytes) => bytes, + Err(error) => { + tracing::warn!( + target: "dynamo_llm::request_trace", + %error, + "request trace s3: finalize gzip batch failed; discarding" + ); + return; + } + }; + let this_seq = *seq; + *seq = seq.saturating_add(1); + let key = uploader.object_key(SystemTime::now(), this_seq); + let batch_bytes = ready.len(); + if let Err(error) = uploader.put_object(key.clone(), ready).await { + // The SDK exhausted its retries (three total attempts by default, + // bounded by the operation timeout). The batch is dropped here rather + // than requeued; a persistent retry buffer is a follow-up concern + // tracked in the S3 layout PR. + tracing::warn!( + target: "dynamo_llm::request_trace", + key = %key, + batch_bytes, + %error, + "request trace s3: put_object failed after SDK retries; batch discarded" + ); + } +} + +struct S3Uploader { + client: aws_sdk_s3::Client, + options: S3UploadOptions, +} + +impl S3Uploader { + fn object_key(&self, at: SystemTime, seq: u64) -> String { + // Simple time-based layout for PR 1. Richer partitioning + // (model=/date=/hour= Hive style) lands in the follow-up. + // + // `run_id` guarantees per-process uniqueness so that a pod restart + // within the same second (or two frontends sharing a hostname) + // cannot overwrite each other's objects. + let secs = at + .duration_since(UNIX_EPOCH) + .map(|dur| dur.as_secs()) + .unwrap_or_default(); + let (yyyy, mm, dd, hh, mi, ss) = utc_date_parts(secs); + let mut key = String::new(); + let prefix = self.options.prefix.trim_matches('/'); + if !prefix.is_empty() { + key.push_str(prefix); + key.push('/'); + } + key.push_str(&format!( + "{yyyy:04}/{mm:02}/{dd:02}/{host}-{hh:02}{mi:02}{ss:02}-{run_id}-{seq:06}.jsonl.gz", + host = self.options.host, + run_id = self.options.run_id, + )); + key + } + + async fn put_object(&self, key: String, body: Vec) -> Result<()> { + self.client + .put_object() + .bucket(&self.options.bucket) + .key(key) + .content_type("application/gzip") + .body(ByteStream::from(body)) + .send() + .await + .with_context(|| format!("s3 put_object into bucket {}", self.options.bucket))?; + Ok(()) + } +} + +/// Buffers raw JSONL bytes until the batch is ready to flush; gzip +/// compression happens in [`take_finished`], which offloads the CPU work +/// to a blocking pool (matching `telemetry::jsonl_gz`). +struct JsonlBatch { + raw: Vec, + lines: u64, +} + +impl JsonlBatch { + fn new() -> Self { + Self { + raw: Vec::with_capacity(DEFAULT_BUFFER_INITIAL_BYTES), + lines: 0, + } + } + + fn is_empty(&self) -> bool { + self.lines == 0 + } + + fn uncompressed_bytes(&self) -> u64 { + self.raw.len() as u64 + } + + fn push(&mut self, record: &RequestTraceRecord) -> Result<()> { + let mut line = serde_json::to_vec(record).context("serializing request trace record")?; + line.push(b'\n'); + self.raw.extend_from_slice(&line); + self.lines = self.lines.saturating_add(1); + Ok(()) + } + + /// Consume the accumulated JSONL, gzip it on a blocking worker, and + /// leave the batch empty so it can accept the next record. + async fn take_finished(&mut self) -> Result> { + let raw = std::mem::replace( + &mut self.raw, + Vec::with_capacity(DEFAULT_BUFFER_INITIAL_BYTES), + ); + self.lines = 0; + tokio::task::spawn_blocking(move || { + let mut encoder = GzEncoder::new( + Vec::with_capacity(raw.len() / 4 + DEFAULT_BUFFER_INITIAL_BYTES), + Compression::default(), + ); + encoder + .write_all(&raw) + .context("writing request trace batch to gzip encoder")?; + encoder + .finish() + .context("finalizing gzip batch for s3 upload") + }) + .await + .context("gzip encoder task panicked")? + } +} + +fn hostname_or_fallback() -> String { + std::env::var("HOSTNAME") + .ok() + .filter(|value| !value.trim().is_empty()) + .unwrap_or_else(|| "unknown".to_string()) +} + +/// Convert unix seconds to UTC (year, month, day, hour, minute, second). +/// Implemented inline to avoid pulling in chrono for one call site. +fn utc_date_parts(secs: u64) -> (i64, u32, u32, u32, u32, u32) { + // 1970-01-01 is a Thursday. Compute days since epoch, then split. + const SECS_PER_DAY: u64 = 86_400; + let days = (secs / SECS_PER_DAY) as i64; + let time_of_day = secs % SECS_PER_DAY; + let hh = (time_of_day / 3600) as u32; + let mi = ((time_of_day % 3600) / 60) as u32; + let ss = (time_of_day % 60) as u32; + + // Howard Hinnant civil_from_days + let z = days + 719_468; + let era = if z >= 0 { + z / 146_097 + } else { + (z - 146_096) / 146_097 + }; + let doe = (z - era * 146_097) as u64; + let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365; + let y = yoe as i64 + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = doy - (153 * mp + 2) / 5 + 1; + let m = if mp < 10 { mp + 3 } else { mp - 9 }; + let year = if m <= 2 { y + 1 } else { y }; + (year, m as u32, d as u32, hh, mi, ss) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::request_trace::{RequestTraceEventType, RequestTraceSchema}; + + #[test] + fn utc_date_parts_epoch() { + let (y, m, d, h, mi, s) = utc_date_parts(0); + assert_eq!((y, m, d, h, mi, s), (1970, 1, 1, 0, 0, 0)); + } + + #[test] + fn utc_date_parts_known_moment() { + // 2026-07-15T12:34:56Z = 1_784_118_896 (`date -u -d ...`) + let (y, m, d, h, mi, s) = utc_date_parts(1_784_118_896); + assert_eq!((y, m, d, h, mi, s), (2026, 7, 15, 12, 34, 56)); + } + + #[test] + fn object_key_includes_prefix_date_run_and_seq() { + let uploader = S3Uploader { + client: dummy_client(), + options: S3UploadOptions { + bucket: "b".to_string(), + prefix: "traces/".to_string(), + host: "frontend-0".to_string(), + run_id: "cafebabe".to_string(), + }, + }; + let at = UNIX_EPOCH + Duration::from_secs(1_784_118_896); + let key = uploader.object_key(at, 42); + assert_eq!( + key, + "traces/2026/07/15/frontend-0-123456-cafebabe-000042.jsonl.gz" + ); + } + + #[test] + fn object_key_omits_leading_slash_when_prefix_empty() { + let uploader = S3Uploader { + client: dummy_client(), + options: S3UploadOptions { + bucket: "b".to_string(), + prefix: String::new(), + host: "h".to_string(), + run_id: "abc".to_string(), + }, + }; + let key = uploader.object_key(UNIX_EPOCH, 0); + assert!(!key.starts_with('/')); + assert!(key.starts_with("1970/01/01/h-")); + } + + fn dummy_client() -> aws_sdk_s3::Client { + // A no-network client. Any subsequent `send()` call would fail, but + // pure-Rust key generation does not exercise the transport. + let config = aws_sdk_s3::Config::builder() + .behavior_version(BehaviorVersion::latest()) + .region(Region::new("us-east-1")) + .build(); + aws_sdk_s3::Client::from_conf(config) + } + + /// Build a sink whose bounded channel is never drained, so `emit` hits the + /// same backpressure path a stalled uploader would cause. The receiver is + /// returned to the caller and held so the channel stays open (full), not + /// closed. No AWS client or worker task is created. + fn stalled_sink(capacity: usize) -> (S3RequestTraceSink, mpsc::Receiver) { + let (tx, rx) = mpsc::channel(capacity); + let sink = S3RequestTraceSink { + tx, + shutdown: CancellationToken::new(), + worker: Mutex::new(None), + dropped: Arc::new(AtomicU64::new(0)), + }; + (sink, rx) + } + + fn sample_record() -> RequestTraceRecord { + RequestTraceRecord { + schema: RequestTraceSchema::V1, + event_type: RequestTraceEventType::RequestEnd, + event_time_unix_ms: 0, + event_source: None, + agent_context: None, + request: None, + tool: None, + payload: None, + } + } + + #[tokio::test] + async fn emit_drops_and_counts_when_channel_is_full() { + let capacity = 4; + // Hold the receiver so the channel stays open; never drain it. + let (sink, _rx) = stalled_sink(capacity); + + // The channel accepts `capacity` records, then every further emit drops. + let total = capacity + 20; + for _ in 0..total { + sink.emit(&sample_record()).await; + } + + let dropped = sink.dropped.load(Ordering::Relaxed); + assert_eq!(dropped, (total - capacity) as u64); + } + + #[test] + fn note_dropped_warns_only_on_first_drop() { + let (sink, _rx) = stalled_sink(1); + // First drop emits the warning; every subsequent drop is silent. + assert!(sink.note_dropped("channel_full")); + for _ in 0..1000 { + assert!(!sink.note_dropped("channel_full")); + } + assert_eq!(sink.dropped.load(Ordering::Relaxed), 1001); + } +} diff --git a/lib/llm/src/request_trace/sink.rs b/lib/llm/src/request_trace/sink.rs index 7917d30dc33c..076ec082447f 100644 --- a/lib/llm/src/request_trace/sink.rs +++ b/lib/llm/src/request_trace/sink.rs @@ -205,6 +205,19 @@ async fn parse_sinks_from_env() -> anyhow::Result> JsonlGzipRequestTraceSink::from_policy(policy).await?, )), }, + RequestTraceSinkKind::S3 => { + #[cfg(feature = "request-trace-s3")] + { + use super::s3_sink::S3RequestTraceSink; + sinks.push(Arc::new(S3RequestTraceSink::from_policy(policy).await?)); + } + #[cfg(not(feature = "request-trace-s3"))] + { + return Err(anyhow!( + "request trace s3 sink requested but dynamo-llm was built without the \"request-trace-s3\" feature", + )); + } + } } } Ok(sinks) diff --git a/lib/runtime/src/config/environment_names.rs b/lib/runtime/src/config/environment_names.rs index 3fa144e4268d..1e48188d31db 100644 --- a/lib/runtime/src/config/environment_names.rs +++ b/lib/runtime/src/config/environment_names.rs @@ -476,7 +476,8 @@ pub mod llm { /// Master switch. Truthy enables request trace emission. pub const DYN_REQUEST_TRACE: &str = "DYN_REQUEST_TRACE"; - /// Request trace sink selection. Comma-separated values: `file`, `stderr`, `nats`, `otel`. + /// Request trace sink selection. Comma-separated values: `file`, + /// `stderr`, `nats`, `otel`, `s3`. /// /// Legacy values map as follows: `jsonl` => `file` with `jsonl` format, /// `jsonl_gz` => `file` with `jsonl_gz` format, `stderr` => `stderr`, @@ -553,6 +554,30 @@ pub mod llm { /// are recorded unredacted; avoid credential-bearing headers. pub const DYN_REQUEST_TRACE_HTTP_HEADER_CAPTURE_LIST: &str = "DYN_REQUEST_TRACE_HTTP_HEADER_CAPTURE_LIST"; + + /// S3 bucket for the S3 request-trace sink. Required when + /// `DYN_REQUEST_TRACE_SINKS` includes `s3`. + pub const DYN_REQUEST_TRACE_S3_BUCKET: &str = "DYN_REQUEST_TRACE_S3_BUCKET"; + + /// AWS region for the S3 request-trace sink. When unset the AWS SDK + /// default region resolution is used (env, profile, IMDS). + pub const DYN_REQUEST_TRACE_S3_REGION: &str = "DYN_REQUEST_TRACE_S3_REGION"; + + /// Optional object key prefix for the S3 request-trace sink. When unset + /// records land at the bucket root. + pub const DYN_REQUEST_TRACE_S3_PREFIX: &str = "DYN_REQUEST_TRACE_S3_PREFIX"; + + /// S3 batch roll threshold in uncompressed bytes. When the pending + /// batch reaches this size, it is finalized and uploaded. Default + /// `67108864` (64 MiB). + pub const DYN_REQUEST_TRACE_S3_ROLL_UNCOMPRESSED_BYTES: &str = + "DYN_REQUEST_TRACE_S3_ROLL_UNCOMPRESSED_BYTES"; + + /// S3 periodic flush interval in milliseconds. Any partial batch is + /// finalized and uploaded when this elapses, so low-volume traces + /// still land in S3. Default `10000` (10 s). + pub const DYN_REQUEST_TRACE_S3_FLUSH_INTERVAL_MS: &str = + "DYN_REQUEST_TRACE_S3_FLUSH_INTERVAL_MS"; } }