feat(audit): S3-backed audit sink for request capture and replay - #10903
Closed
YiqiuLiu wants to merge 1 commit into
Closed
feat(audit): S3-backed audit sink for request capture and replay#10903YiqiuLiu wants to merge 1 commit into
YiqiuLiu wants to merge 1 commit into
Conversation
Contributor
|
👋 Hi YiqiuLiu! Thank you for contributing to ai-dynamo/dynamo. Just a reminder: The 🚀 |
github-actions
Bot
added
external-contribution
This comment has been minimized.
This comment has been minimized.
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| // storage (e.g., S3 via IRSA). Whether audit is enabled at runtime | ||
| // is controlled by the DYN_AUDIT_SINKS env var, not by this field. | ||
| // +optional | ||
| Audit *AuditSpec `json:"audit,omitempty"` |
Contributor
There was a problem hiding this comment.
🔴 New Audit field silently lost during API conversion — v1alpha1 is storage version
The PR adds Audit *AuditSpec to DynamoGraphDeploymentSpec in v1beta1 but does not add any conversion handling. Since v1alpha1 is still the storage version (see comment at deploy/operator/api/v1beta1/dynamographdeployment_types.go:141-142), every write through the v1beta1 API triggers v1beta1→v1alpha1 conversion for storage. ConvertToDynamoGraphDeploymentSpec at deploy/operator/api/v1alpha1/dynamographdeployment_conversion.go:407-469 does not save the Audit field into the sparse hub payload (hubSave), and ConvertFromDynamoGraphDeploymentSpec at line 100-179 does not restore it from restored. This means any spec.audit value set via v1beta1 is silently dropped on every write and never round-trips. Additionally, knownV1Beta1ConversionFieldSet in deploy/operator/api/v1alpha1/conversion_field_coverage_test.go:32-161 was not updated, so TestV1Beta1ConversionFieldSetIsAcknowledged will fail. This violates the mandatory rule in deploy/operator/api/AGENTS.md: "Every API type change in any version must update the corresponding conversion code and conversion tests."
Prompt for agents
The new Audit field on DynamoGraphDeploymentSpec (v1beta1) has no conversion handling. Since v1alpha1 is the storage version, this field is silently dropped on every write.
Required changes per deploy/operator/api/CONVERSION.md:
1. In deploy/operator/api/v1alpha1/dynamographdeployment_conversion.go, function ConvertToDynamoGraphDeploymentSpec: save src.Audit into the hubSave sparse payload (save.Audit = src.Audit) so it round-trips through v1alpha1 storage.
2. In ConvertFromDynamoGraphDeploymentSpec: restore dst.Audit from the restored hub payload when the live v1alpha1 source cannot represent it (if restored != nil && restored.Audit != nil, set dst.Audit = restored.Audit).
3. Update knownV1Beta1ConversionFieldSet in conversion_field_coverage_test.go to include the new audit sub-fields (audit.aws_s3.irsaRoleArn, audit.gcp_gcs.gcpServiceAccount, audit.azure_blob.clientId).
4. Add focused round-trip conversion tests for the Audit field.
5. Regenerate zz_generated.deepcopy.go (make generate) so DeepCopy covers AuditSpec and its children.
Was this helpful? React with 👍 or 👎 to provide feedback.
Comment on lines
89
to
134
| Audit *AuditSpec `json:"audit,omitempty"` | ||
| } | ||
|
|
||
| // AuditSpec carries per-cloud identity metadata the operator attaches to | ||
| // the frontend pod's ServiceAccount. Each sub-field maps to a specific | ||
| // cloud's workload-identity annotation. Multiple sub-fields may be set | ||
| // simultaneously for multi-cloud deployments. | ||
| type AuditSpec struct { | ||
| // aws_s3 configures AWS IRSA for the S3 audit sink. | ||
| // +optional | ||
| AwsS3 *AuditAwsS3Spec `json:"aws_s3,omitempty"` | ||
|
|
||
| // gcp_gcs configures GCP Workload Identity for a future GCS audit sink. | ||
| // +optional | ||
| GcpGcs *AuditGcpGcsSpec `json:"gcp_gcs,omitempty"` | ||
|
|
||
| // azure_blob configures Azure Workload Identity for a future Azure Blob audit sink. | ||
| // +optional | ||
| AzureBlob *AuditAzureBlobSpec `json:"azure_blob,omitempty"` | ||
| } | ||
|
|
||
| // AuditAwsS3Spec holds the IAM role ARN for EKS IRSA. The operator | ||
| // translates this into an `eks.amazonaws.com/role-arn` annotation on the | ||
| // per-DGD ServiceAccount. | ||
| type AuditAwsS3Spec struct { | ||
| // irsaRoleArn is the ARN of the IAM role the frontend pod should | ||
| // assume via IRSA to write audit segments to S3. | ||
| // +kubebuilder:validation:Pattern=`^arn:aws:iam::\d{12}:role/.+$` | ||
| IrsaRoleArn string `json:"irsaRoleArn"` | ||
| } | ||
|
|
||
| // AuditGcpGcsSpec holds the GCP service account for GKE Workload Identity. | ||
| // Future use — not implemented in v1. | ||
| type AuditGcpGcsSpec struct { | ||
| // gcpServiceAccount is the GCP service account email the frontend pod | ||
| // should use via GKE Workload Identity. | ||
| // +kubebuilder:validation:Pattern=`^.+@.+\.iam\.gserviceaccount\.com$` | ||
| GcpServiceAccount string `json:"gcpServiceAccount"` | ||
| } | ||
|
|
||
| // AuditAzureBlobSpec holds the Azure client ID for AKS Workload Identity. | ||
| // Future use — not implemented in v1. | ||
| type AuditAzureBlobSpec struct { | ||
| // clientId is the Azure AD application (client) ID. | ||
| ClientId string `json:"clientId"` | ||
| } |
Contributor
There was a problem hiding this comment.
🔴 DeepCopy not regenerated — Audit pointer shared after DeepCopyInto
The zz_generated.deepcopy.go was not regenerated after adding Audit *AuditSpec to DynamoGraphDeploymentSpec. The DeepCopyInto method at deploy/operator/api/v1beta1/zz_generated.deepcopy.go:712-757 does *out = *in (shallow struct copy) and then deep-copies each pointer field — but there is no block for Audit. After DeepCopy(), both the original and copy share the same *AuditSpec pointer. Mutations through one will silently corrupt the other. Furthermore, no DeepCopy/DeepCopyInto methods exist for AuditSpec, AuditAwsS3Spec, AuditGcpGcsSpec, or AuditAzureBlobSpec. In a Kubernetes operator, DeepCopy is used pervasively (status updates, before-mutation snapshots, informer caches), so shared pointer mutation can cause data races and incorrect reconciliation behavior.
Prompt for agents
Run make generate (or controller-gen) to regenerate deploy/operator/api/v1beta1/zz_generated.deepcopy.go. This will add DeepCopyInto/DeepCopy methods for AuditSpec, AuditAwsS3Spec, AuditGcpGcsSpec, AuditAzureBlobSpec, and update DynamoGraphDeploymentSpec.DeepCopyInto to deep-copy the Audit pointer field.
Was this helpful? React with 👍 or 👎 to provide feedback.
Comment on lines
+105
to
108
| // Forced requests bypass head-based sampling so debugging / | ||
| // compliance flags always land in the audit log. | ||
| if !force_logging && sample_rate < 1.0 && !is_sampled(request_id, sample_rate) { | ||
| return None; |
Contributor
There was a problem hiding this comment.
🟡 Sampling applied to store=true requests despite docstring claiming bypass
The sample_rate field docstring at lib/llm/src/audit/config.rs:33-34 states: "Bypassed by force_logging and by request.store == true." However, the implementation at lib/llm/src/audit/handle.rs:107 checks !force_logging — not !force (which is force_logging || store_flag). This means requests with store=true ARE subject to sampling when force_logging=false. When a client explicitly sets store=true to request audit capture and the operator has configured sample_rate < 1.0, those requests can be silently dropped. The test at handle.rs:249-256 confirms this behavior, but it contradicts the documented contract. Either the code should use !force (matching the docstring's stated semantics that store=true bypasses sampling), or the docstring should be corrected to remove the store=true bypass claim.
Suggested change
| // Forced requests bypass head-based sampling so debugging / | |
| // compliance flags always land in the audit log. | |
| if !force_logging && sample_rate < 1.0 && !is_sampled(request_id, sample_rate) { | |
| return None; | |
| // Forced requests bypass head-based sampling so debugging / | |
| // compliance flags always land in the audit log. | |
| if !force && sample_rate < 1.0 && !is_sampled(request_id, sample_rate) { | |
| return None; | |
| } |
Was this helpful? React with 👍 or 👎 to provide feedback.
Comment on lines
+17
to
+60
| pub static AUDIT_S3_SEGMENTS_TOTAL: LazyLock<IntCounterVec> = LazyLock::new(|| { | ||
| let opts = Opts::new( | ||
| "dynamo_audit_s3_segments_total", | ||
| "Total audit segments uploaded to S3", | ||
| ) | ||
| .namespace("dynamo"); | ||
| IntCounterVec::new(opts, &["result"]).expect("audit_s3_segments_total metric") | ||
| }); | ||
|
|
||
| /// Total audit records dropped, labeled by reason: | ||
| /// - "bus_lag": broadcast channel lagged, records were overwritten before the | ||
| /// sink worker could consume them. | ||
| /// - "channel_full": the per-sink mpsc channel was full (try_send failed). | ||
| /// - "serialize_error": serde_json serialization of AuditRecord failed. | ||
| pub static AUDIT_RECORDS_DROPPED_TOTAL: LazyLock<IntCounterVec> = LazyLock::new(|| { | ||
| let opts = Opts::new( | ||
| "dynamo_audit_records_dropped_total", | ||
| "Total audit records dropped before reaching the destination", | ||
| ) | ||
| .namespace("dynamo"); | ||
| IntCounterVec::new(opts, &["reason"]).expect("audit_records_dropped_total metric") | ||
| }); | ||
|
|
||
| /// Histogram of S3 PutObject durations in seconds. | ||
| pub static AUDIT_S3_UPLOAD_DURATION_SECONDS: LazyLock<HistogramVec> = LazyLock::new(|| { | ||
| let opts = HistogramOpts::new( | ||
| "dynamo_audit_s3_upload_duration_seconds", | ||
| "Duration of S3 PutObject calls for audit segments", | ||
| ) | ||
| .namespace("dynamo") | ||
| .buckets(exponential_buckets(0.01, 2.0, 12).expect("valid histogram buckets")); | ||
| HistogramVec::new(opts, &[]).expect("audit_s3_upload_duration_seconds metric") | ||
| }); | ||
|
|
||
| /// Histogram of uploaded segment sizes in bytes (compressed). | ||
| pub static AUDIT_S3_SEGMENT_SIZE_BYTES: LazyLock<HistogramVec> = LazyLock::new(|| { | ||
| let opts = HistogramOpts::new( | ||
| "dynamo_audit_s3_segment_size_bytes", | ||
| "Size of uploaded audit segments in bytes (gzip-compressed)", | ||
| ) | ||
| .namespace("dynamo") | ||
| .buckets(exponential_buckets(1024.0, 4.0, 10).expect("valid histogram buckets")); | ||
| HistogramVec::new(opts, &[]).expect("audit_s3_segment_size_bytes metric") | ||
| }); |
Contributor
There was a problem hiding this comment.
🚩 Prometheus metrics use LazyLock but are never registered with a Registry
The audit metrics in lib/llm/src/audit/metrics.rs use LazyLock for global initialization but are never explicitly registered with Prometheus's default registry or any custom registry. In prometheus 0.14, IntCounterVec::new and HistogramVec::new create metrics but do NOT auto-register them. They must be registered via prometheus::register(...) or registry.register(...) to appear in /metrics scrape output. As written, these metrics are created and incremented correctly in-process, but will never be exposed to Prometheus scrapers. This should be verified — if the codebase has a central registration pattern elsewhere that picks these up, it's fine; otherwise these metrics are silently invisible.
Was this helpful? React with 👍 or 👎 to provide feedback.
Contributor
WalkthroughThis PR adds an S3-backed audit sink to the LLM library: a rotating gzip JSONL writer is refactored around a new ChangesRust LLM Audit S3 Sink
Kubernetes Operator Audit Identity Annotations
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes 🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 golangci-lint (2.12.2)level=error msg="[linters_context] typechecking error: pattern ./...: directory prefix . does not contain main module or its selected dependencies" Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (1)
deploy/operator/api/v1beta1/dynamographdeployment_types.go (1)
96-133: 📐 Maintainability & Code Quality | 🔵 TrivialUse camelCase for the AuditSpec JSON field names to align with Kubernetes API conventions.
The JSON tags
aws_s3,gcp_gcs, andazure_blobuse snake_case while their nested fields (irsaRoleArn,gcpServiceAccount,clientId) use camelCase. This inconsistency conflicts with Kubernetes API naming conventions and will be a breaking change if corrected after release. Since the GCP and Azure specs are marked "Future use — not implemented in v1", renaming now is the safest approach.The controller logic reads only the typed Go fields, not JSON tags, so this change is JSON-only and won't affect the reconciliation code.
♻️ Proposed naming alignment
// aws_s3 configures AWS IRSA for the S3 audit sink. // +optional - AwsS3 *AuditAwsS3Spec `json:"aws_s3,omitempty"` + AwsS3 *AuditAwsS3Spec `json:"awsS3,omitempty"` // gcp_gcs configures GCP Workload Identity for a future GCS audit sink. // +optional - GcpGcs *AuditGcpGcsSpec `json:"gcp_gcs,omitempty"` + GcpGcs *AuditGcpGcsSpec `json:"gcpGcs,omitempty"` // azure_blob configures Azure Workload Identity for a future Azure Blob audit sink. // +optional - AzureBlob *AuditAzureBlobSpec `json:"azure_blob,omitempty"` + AzureBlob *AuditAzureBlobSpec `json:"azureBlob,omitempty"`🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@deploy/operator/api/v1beta1/dynamographdeployment_types.go` around lines 96 - 133, The JSON field tags in the AuditSpec struct use snake_case (aws_s3, gcp_gcs, azure_blob) while their nested field types use camelCase, creating an inconsistency with Kubernetes API conventions. Update the JSON tags on the three pointer fields in AuditSpec (AwsS3, GcpGcs, and AzureBlob) to use camelCase format (awsS3, gcpGcs, azureBlob) to match the naming convention used throughout the nested specifications and maintain consistency with Kubernetes standards.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@deploy/operator/internal/controller/dynamographdeployment_controller.go`:
- Around line 1799-1808: The code silently returns nil when the discovery
ServiceAccount is not found, making it unobservable to operators when
audit-at-runtime is misconfigured. In the error handling block where
errors.IsNotFound(err) is checked and currently returns nil, add logging at info
or warn level before the return statement to inform operators that the IRSA/WI
annotation was skipped due to the ServiceAccount not existing. This log message
should clarify that the absence of the SA (typically due to discovery being
disabled) means the audit pod will not have the necessary annotations for S3
authentication.
In `@lib/llm/src/audit/config.rs`:
- Around line 93-97: The sample_rate calculation in the audit config parsing
chain allows NaN values from the environment variable to bypass the clamp
operation, which causes issues downstream in the is_sampled function. Add a
filter using the is_finite() method in the parsing chain after mapping the
clamped value to reject any non-finite numbers (including NaN, infinity, or
negative infinity) before falling back to DEFAULT_SAMPLE_RATE. This ensures only
valid finite values in the [0.0, 1.0] range are accepted, maintaining the
documented invariant.
- Around line 133-137: The s3_sse match statement accepts any value other than
"none" without validation, which could lead to unsupported encryption modes
being passed to S3. Modify the match expression for
read_string(env_audit::DYN_AUDIT_S3_SSE) to validate that the value is one of
the three supported modes: "AES256", "aws:kms", or "none" (case-insensitive).
For the "none" case, return None; for valid "AES256" and "aws:kms" values,
return Some(v); and for any invalid input or None, default to
Some("AES256".to_string()) to ensure safe encryption defaults.
In `@lib/llm/src/audit/handle.rs`:
- Around line 98-107: The sampling gate logic on line 107 contradicts the
AuditPolicy contract where request.store==true should bypass sampling entirely.
Currently the condition checks !force_logging instead of !force, which means
requests with store=true still get sampled even though they should bypass
sampling. Fix this by changing the condition from !force_logging to !force so
that all forced requests (including those where store_flag is true) bypass the
sampling check and are guaranteed to land in the audit log, while only
non-forced requests undergo the head-based sampling rate check.
In `@lib/llm/src/audit/metrics.rs`:
- Around line 18-23: The metric names being passed to Opts::new() include a
redundant "dynamo_" prefix when the namespace "dynamo" is also being set via the
.namespace("dynamo") call, resulting in doubled namespace prefixes like
"dynamo_dynamo_audit_s3_segments_total". Remove the "dynamo_" prefix from the
metric name strings in all Opts::new() calls (such as in the IntCounterVec
creation and other similar metric definitions) while keeping the
.namespace("dynamo") calls unchanged, so metric names should start with just
"audit_" instead of "dynamo_audit_". Apply this fix to all metric definitions
mentioned in the file.
In `@lib/llm/src/audit/README.md`:
- Around line 117-119: The fenced code block containing the file path format
string is missing a language tag, which violates markdownlint rule MD040. Add
the language identifier "text" to the opening fence (the triple backticks) that
precedes the line with the prefix format containing the pod-name and
startup-uuid8 variables. This will properly label the code block as a text
format example.
In `@lib/llm/src/audit/sink.rs`:
- Around line 212-223: The JsonlGzipWriter::with_segment_sink() constructor call
is not using the configurable s3_channel_capacity parameter from the
AuditPolicy, which means the DYN_AUDIT_S3_CHANNEL_CAPACITY configuration knob
has no effect and a hardcoded channel size is used instead. Add the
s3_channel_capacity field from the policy object as a parameter to the
JsonlGzipWriter::with_segment_sink() call to wire the configuration into the S3
sink queue and allow proper backpressure and memory tuning.
---
Nitpick comments:
In `@deploy/operator/api/v1beta1/dynamographdeployment_types.go`:
- Around line 96-133: The JSON field tags in the AuditSpec struct use snake_case
(aws_s3, gcp_gcs, azure_blob) while their nested field types use camelCase,
creating an inconsistency with Kubernetes API conventions. Update the JSON tags
on the three pointer fields in AuditSpec (AwsS3, GcpGcs, and AzureBlob) to use
camelCase format (awsS3, gcpGcs, azureBlob) to match the naming convention used
throughout the nested specifications and maintain consistency with Kubernetes
standards.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: f5415a5f-216e-4afc-8797-ca04d9f92e71
📒 Files selected for processing (17)
Cargo.tomldeploy/operator/api/v1beta1/dynamographdeployment_types.godeploy/operator/internal/controller/audit_identity_test.godeploy/operator/internal/controller/dynamographdeployment_controller.golib/llm/Cargo.tomllib/llm/src/audit/README.mdlib/llm/src/audit/config.rslib/llm/src/audit/handle.rslib/llm/src/audit/metrics.rslib/llm/src/audit/mod.rslib/llm/src/audit/sink.rslib/llm/src/request_trace/sink.rslib/llm/src/telemetry/jsonl_gz.rslib/llm/src/telemetry/mod.rslib/llm/src/telemetry/s3_segment_sink.rslib/llm/tests/audit_s3_integration.rslib/runtime/src/config/environment_names.rs
Comment on lines
+1799
to
+1808
| // Fetch the SA | ||
| sa := &corev1.ServiceAccount{} | ||
| if err := r.Get(ctx, types.NamespacedName{Name: saName, Namespace: dynamoDeployment.Namespace}, sa); err != nil { | ||
| if errors.IsNotFound(err) { | ||
| // SA hasn't been created yet (e.g., discovery is disabled). | ||
| // Nothing to annotate — skip silently. | ||
| return nil | ||
| } | ||
| return fmt.Errorf("audit identity: failed to get SA %s: %w", saName, err) | ||
| } |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Audit IRSA is silently skipped when discovery is disabled.
When the discovery SA does not exist, the method returns nil and never applies the IRSA/WI annotation. Since audit-at-runtime is gated separately by DYN_AUDIT_SINKS, an operator can enable the S3 audit sink with spec.audit.aws_s3 set while discovery is off, and the frontend pod will then fail to authenticate to S3 with no operator-side signal. Consider logging at info/warn level here (or surfacing a status condition) so this misconfiguration is observable rather than silent.
🔎 Suggested observability
if errors.IsNotFound(err) {
- // SA hasn't been created yet (e.g., discovery is disabled).
- // Nothing to annotate — skip silently.
+ // SA hasn't been created yet (e.g., discovery is disabled).
+ // Nothing to annotate — skip. Warn if audit identity was requested
+ // so the misconfiguration (audit enabled without discovery) is visible.
+ if dynamoDeployment.Spec.Audit != nil {
+ logger.Info("audit identity requested but discovery ServiceAccount is absent; skipping annotation", "sa", saName)
+ }
return nil
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@deploy/operator/internal/controller/dynamographdeployment_controller.go`
around lines 1799 - 1808, The code silently returns nil when the discovery
ServiceAccount is not found, making it unobservable to operators when
audit-at-runtime is misconfigured. In the error handling block where
errors.IsNotFound(err) is checked and currently returns nil, add logging at info
or warn level before the return statement to inform operators that the IRSA/WI
annotation was skipped due to the ServiceAccount not existing. This log message
should clarify that the absence of the SA (typically due to discovery being
disabled) means the audit pod will not have the necessary annotations for S3
authentication.
Comment on lines
+93
to
+97
| let sample_rate = std::env::var(env_audit::DYN_AUDIT_SAMPLE_RATE) | ||
| .ok() | ||
| .and_then(|v| v.parse::<f32>().ok()) | ||
| .map(|v| v.clamp(0.0, 1.0)) | ||
| .unwrap_or(DEFAULT_SAMPLE_RATE); |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major
🧩 Analysis chain
🏁 Script executed:
# First, locate and examine the audit config file
find . -name "config.rs" -path "*/audit/*" | head -5Repository: ai-dynamo/dynamo
Length of output: 184
🏁 Script executed:
# Get the file structure around the relevant lines
ast-grep outline lib/llm/src/audit/config.rs --view expandedRepository: ai-dynamo/dynamo
Length of output: 1642
🏁 Script executed:
# Read the specific section and surrounding context
sed -n '1,120p' lib/llm/src/audit/config.rs | cat -nRepository: ai-dynamo/dynamo
Length of output: 5840
🏁 Script executed:
# Check if f32::clamp actually handles NaN correctly
cat > /tmp/test_nan.rs << 'EOF'
fn main() {
let nan_val = f32::NAN;
let clamped = nan_val.clamp(0.0, 1.0);
println!("NaN clamp result: {}", clamped);
println!("Is finite: {}", clamped.is_finite());
let parsed = "NaN".parse::<f32>().ok();
println!("Parsed NaN: {:?}", parsed);
if let Some(v) = parsed {
println!("Parsed is finite: {}", v.is_finite());
println!("Parsed clamp result: {}", v.clamp(0.0, 1.0));
}
}
EOF
# We'll use python to verify NaN behavior since we can't compile
python3 << 'PY'
import math
nan = float('nan')
print(f"NaN value: {nan}")
print(f"NaN is finite: {math.isfinite(nan)}")
# Simulating clamp behavior
clamped = max(0.0, min(1.0, nan))
print(f"Clamp(0, 1) of NaN: {clamped}")
print(f"Clamped is finite: {math.isfinite(clamped)}")
# String parse
parsed = float("NaN")
print(f"Parsed 'NaN': {parsed}")
PYRepository: ai-dynamo/dynamo
Length of output: 256
🏁 Script executed:
# Check for Linear ticket references in the config.rs file
rg "DIS-|DYN-|OPS-|DEP-" lib/llm/src/audit/config.rs -nRepository: ai-dynamo/dynamo
Length of output: 154
🏁 Script executed:
# Check how sample_rate is used downstream to understand impact
rg "sample_rate" lib/llm/src/audit/ -A 2 -B 2Repository: ai-dynamo/dynamo
Length of output: 5267
Guard sample_rate against non-finite values.
DYN_AUDIT_SAMPLE_RATE=NaN parses successfully and bypasses the clamp operation, passing NaN downstream. In is_sampled(), this causes the threshold calculation to produce 0, blocking all non-forced audit requests. Add a finiteness check before clamping to enforce the [0.0, 1.0] invariant documented in the struct.
Proposed fix
let sample_rate = std::env::var(env_audit::DYN_AUDIT_SAMPLE_RATE)
.ok()
.and_then(|v| v.parse::<f32>().ok())
+ .filter(|v| v.is_finite())
.map(|v| v.clamp(0.0, 1.0))
.unwrap_or(DEFAULT_SAMPLE_RATE);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| let sample_rate = std::env::var(env_audit::DYN_AUDIT_SAMPLE_RATE) | |
| .ok() | |
| .and_then(|v| v.parse::<f32>().ok()) | |
| .map(|v| v.clamp(0.0, 1.0)) | |
| .unwrap_or(DEFAULT_SAMPLE_RATE); | |
| let sample_rate = std::env::var(env_audit::DYN_AUDIT_SAMPLE_RATE) | |
| .ok() | |
| .and_then(|v| v.parse::<f32>().ok()) | |
| .filter(|v| v.is_finite()) | |
| .map(|v| v.clamp(0.0, 1.0)) | |
| .unwrap_or(DEFAULT_SAMPLE_RATE); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/llm/src/audit/config.rs` around lines 93 - 97, The sample_rate
calculation in the audit config parsing chain allows NaN values from the
environment variable to bypass the clamp operation, which causes issues
downstream in the is_sampled function. Add a filter using the is_finite() method
in the parsing chain after mapping the clamped value to reject any non-finite
numbers (including NaN, infinity, or negative infinity) before falling back to
DEFAULT_SAMPLE_RATE. This ensures only valid finite values in the [0.0, 1.0]
range are accepted, maintaining the documented invariant.
Comment on lines
+133
to
+137
| let s3_sse = match read_string(env_audit::DYN_AUDIT_S3_SSE) { | ||
| Some(v) if v.eq_ignore_ascii_case("none") => None, | ||
| Some(v) => Some(v), | ||
| None => Some(DEFAULT_S3_SSE.to_string()), | ||
| }; |
Contributor
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major
🧩 Analysis chain
🏁 Script executed:
# First, examine the specific file and lines mentioned
cat -n lib/llm/src/audit/config.rs | sed -n '125,145p'Repository: ai-dynamo/dynamo
Length of output: 1265
🏁 Script executed:
# Check if there's a reference mapping for DYN_AUDIT_S3_SSE
cat .ai/linear-ticket-refs.md 2>/dev/null || echo "File not found"Repository: ai-dynamo/dynamo
Length of output: 1643
🏁 Script executed:
# Search for DYN_AUDIT_S3_SSE definition and usage
rg "DYN_AUDIT_S3_SSE" --type rustRepository: ai-dynamo/dynamo
Length of output: 154
🏁 Script executed:
# Look for DEFAULT_S3_SSE and understand what the default is
rg "DEFAULT_S3_SSE" --type rust -B 2 -A 2Repository: ai-dynamo/dynamo
Length of output: 154
🏁 Script executed:
# Find the env_audit module definition
fd -t f "env_audit" --type rust || find . -name "*env*audit*" -type fRepository: ai-dynamo/dynamo
Length of output: 326
🏁 Script executed:
# Search more broadly for DEFAULT_S3_SSE
rg "DEFAULT_S3_SSE"Repository: ai-dynamo/dynamo
Length of output: 154
🏁 Script executed:
# Find where env_audit is defined and look for DYN_AUDIT_S3_SSE
rg "DYN_AUDIT_S3_SSE|DEFAULT_S3_SSE" -t rust -t tomlRepository: ai-dynamo/dynamo
Length of output: 154
🏁 Script executed:
# Check the structure of the audit module
ls -la lib/llm/src/audit/Repository: ai-dynamo/dynamo
Length of output: 751
🏁 Script executed:
# Read the entire config.rs to understand the full context
cat -n lib/llm/src/audit/config.rsRepository: ai-dynamo/dynamo
Length of output: 9028
🏁 Script executed:
# Search for where read_string function is defined
rg "fn read_string"Repository: ai-dynamo/dynamo
Length of output: 154
🏁 Script executed:
# Look for env_audit module
rg "mod env_audit|use.*env_audit"Repository: ai-dynamo/dynamo
Length of output: 154
🏁 Script executed:
# Find where env_audit constants are defined
find . -path "*dynamo_runtime*" -name "*.rs" | head -20Repository: ai-dynamo/dynamo
Length of output: 154
🏁 Script executed:
# Search for where these DYN_AUDIT constants are defined
rg "DYN_AUDIT_S3_SSE\s*=" --type-listRepository: ai-dynamo/dynamo
Length of output: 544
🏁 Script executed:
# Look at how S3 SSE is actually used when creating S3 objects
rg -A 5 -B 5 "s3_sse" lib/llm/src/audit/Repository: ai-dynamo/dynamo
Length of output: 2856
🏁 Script executed:
# Check if there are comments or documentation about valid S3 SSE values
rg "AES256|aws:kms|ServerSideEncryption" lib/llm/src/audit/Repository: ai-dynamo/dynamo
Length of output: 321
Validate s3_sse environment variable to supported encryption modes only.
The current code accepts any value other than "none" and passes it directly to S3, which may silently degrade to bucket-default behavior. Per the audit README, only three values are valid: "AES256", "aws:kms", and "none". Normalize accepted values and default to "AES256" for invalid input to prevent accidental encryption downgrades.
Proposed fix
- let s3_sse = match read_string(env_audit::DYN_AUDIT_S3_SSE) {
- Some(v) if v.eq_ignore_ascii_case("none") => None,
- Some(v) => Some(v),
- None => Some(DEFAULT_S3_SSE.to_string()),
- };
+ let s3_sse = match read_string(env_audit::DYN_AUDIT_S3_SSE) {
+ Some(v) if v.eq_ignore_ascii_case("none") => None,
+ Some(v) if v.eq_ignore_ascii_case("aes256") => Some("AES256".to_string()),
+ Some(v) if v.eq_ignore_ascii_case("aws:kms") => Some("aws:kms".to_string()),
+ Some(_) => Some(DEFAULT_S3_SSE.to_string()),
+ None => Some(DEFAULT_S3_SSE.to_string()),
+ };📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| let s3_sse = match read_string(env_audit::DYN_AUDIT_S3_SSE) { | |
| Some(v) if v.eq_ignore_ascii_case("none") => None, | |
| Some(v) => Some(v), | |
| None => Some(DEFAULT_S3_SSE.to_string()), | |
| }; | |
| let s3_sse = match read_string(env_audit::DYN_AUDIT_S3_SSE) { | |
| Some(v) if v.eq_ignore_ascii_case("none") => None, | |
| Some(v) if v.eq_ignore_ascii_case("aes256") => Some("AES256".to_string()), | |
| Some(v) if v.eq_ignore_ascii_case("aws:kms") => Some("aws:kms".to_string()), | |
| Some(_) => Some(DEFAULT_S3_SSE.to_string()), | |
| None => Some(DEFAULT_S3_SSE.to_string()), | |
| }; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/llm/src/audit/config.rs` around lines 133 - 137, The s3_sse match
statement accepts any value other than "none" without validation, which could
lead to unsupported encryption modes being passed to S3. Modify the match
expression for read_string(env_audit::DYN_AUDIT_S3_SSE) to validate that the
value is one of the three supported modes: "AES256", "aws:kms", or "none"
(case-insensitive). For the "none" case, return None; for valid "AES256" and
"aws:kms" values, return Some(v); and for any invalid input or None, default to
Some("AES256".to_string()) to ensure safe encryption defaults.
Comment on lines
+98
to
+107
| let store_flag = req.inner.store.unwrap_or(false); | ||
| let force = force_logging || store_flag; | ||
| // If neither force_logging nor the request `store` flag is set, this | ||
| // request is not eligible for capture at all. | ||
| if !force { | ||
| return None; | ||
| } | ||
| // Forced requests bypass head-based sampling so debugging / | ||
| // compliance flags always land in the audit log. | ||
| if !force_logging && sample_rate < 1.0 && !is_sampled(request_id, sample_rate) { |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Sampling gate currently contradicts the store bypass contract and blocks normal head-sampling.
Line 102-104 drops all requests when force_logging=false and store=false, so regular traffic can never be captured by sampling. Then Line 107 still samples store=true requests, even though AuditPolicy documents request.store==true as a sampling bypass (lib/llm/src/audit/config.rs).
Suggested fix
- let store_flag = req.inner.store.unwrap_or(false);
- let force = force_logging || store_flag;
- // If neither force_logging nor the request `store` flag is set, this
- // request is not eligible for capture at all.
- if !force {
- return None;
- }
- // Forced requests bypass head-based sampling so debugging /
- // compliance flags always land in the audit log.
- if !force_logging && sample_rate < 1.0 && !is_sampled(request_id, sample_rate) {
+ let store_flag = req.inner.store.unwrap_or(false);
+ let forced = force_logging || store_flag;
+ // force_logging and request.store bypass sampling; all other requests
+ // are decided by deterministic head sampling.
+ if !forced && !is_sampled(request_id, sample_rate) {
return None;
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| let store_flag = req.inner.store.unwrap_or(false); | |
| let force = force_logging || store_flag; | |
| // If neither force_logging nor the request `store` flag is set, this | |
| // request is not eligible for capture at all. | |
| if !force { | |
| return None; | |
| } | |
| // Forced requests bypass head-based sampling so debugging / | |
| // compliance flags always land in the audit log. | |
| if !force_logging && sample_rate < 1.0 && !is_sampled(request_id, sample_rate) { | |
| let store_flag = req.inner.store.unwrap_or(false); | |
| let forced = force_logging || store_flag; | |
| // force_logging and request.store bypass sampling; all other requests | |
| // are decided by deterministic head sampling. | |
| if !forced && !is_sampled(request_id, sample_rate) { | |
| return None; | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/llm/src/audit/handle.rs` around lines 98 - 107, The sampling gate logic
on line 107 contradicts the AuditPolicy contract where request.store==true
should bypass sampling entirely. Currently the condition checks !force_logging
instead of !force, which means requests with store=true still get sampled even
though they should bypass sampling. Fix this by changing the condition from
!force_logging to !force so that all forced requests (including those where
store_flag is true) bypass the sampling check and are guaranteed to land in the
audit log, while only non-forced requests undergo the head-based sampling rate
check.
Comment on lines
+18
to
+23
| let opts = Opts::new( | ||
| "dynamo_audit_s3_segments_total", | ||
| "Total audit segments uploaded to S3", | ||
| ) | ||
| .namespace("dynamo"); | ||
| IntCounterVec::new(opts, &["result"]).expect("audit_s3_segments_total metric") |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major
🧩 Analysis chain
🏁 Script executed:
cat -n lib/llm/src/audit/metrics.rsRepository: ai-dynamo/dynamo
Length of output: 3242
🌐 Web query:
prometheus-rs Opts namespace metric name prefix behavior
💡 Result:
In the prometheus-rs library, the Opts struct uses the namespace, subsystem, and name fields to construct the fully-qualified metric name [1][2]. These components are joined together using an underscore () as a separator [1][3]. The behavior for generating the fully-qualified name follows these rules: 1. Components (namespace, subsystem, and name) are concatenated with "" [1][3]. 2. Empty components are ignored during this construction process [3]. 3. Only the name field is mandatory; namespace and subsystem are optional and serve to help structure the metric name [1][4]. 4. The resulting fully-qualified name must be a valid Prometheus metric name [1][5]. For example, if you set the namespace to "my_app", the subsystem to "database", and the name to "connections", the resulting metric name will be "my_app_database_connections" [1][3]. It is important to distinguish this from the Registry, which also supports an optional prefix [6]. When a prefix is defined at the Registry level, it is prepended to the metric names gathered by that registry during the exposition process [6]. This is distinct from the Opts structure, which handles the naming of individual metrics at the time of creation [1][3]. Additionally, newer Prometheus client implementations (such as prometheus_client) offer sub-registry functionality to manage prefixes for groups of metrics more efficiently [7].
Citations:
- 1: https://docs.rs/prometheus/latest/prometheus/core/struct.Opts.html
- 2: https://docs.rs/prometheus/latest/prometheus/struct.Opts.html
- 3: https://docs.rs/prometheus/latest/src/prometheus/metrics.rs.html
- 4: https://docs.rs/prometheus/0.5.0/prometheus/struct.Opts.html
- 5: https://docs.rs/substrate-prometheus-endpoint/latest/substrate_prometheus_endpoint/struct.Opts.html
- 6: https://docs.rs/prometheus/latest/src/prometheus/registry.rs.html
- 7: https://rustdocs.bsx.fi/prometheus_client/registry/struct.Registry.html
Remove redundant namespace prefix from metric names.
Opts::new("dynamo_audit_s3_segments_total", ...).namespace("dynamo") will produce dynamo_dynamo_audit_s3_segments_total. The namespace is automatically prepended to the metric name, so the base names should not include the "dynamo_" prefix.
Suggested fix
- "dynamo_audit_s3_segments_total",
+ "audit_s3_segments_total",
...
- "dynamo_audit_records_dropped_total",
+ "audit_records_dropped_total",
...
- "dynamo_audit_s3_upload_duration_seconds",
+ "audit_s3_upload_duration_seconds",
...
- "dynamo_audit_s3_segment_size_bytes",
+ "audit_s3_segment_size_bytes",Also applies to: 32-37, 42-48, 53-59
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/llm/src/audit/metrics.rs` around lines 18 - 23, The metric names being
passed to Opts::new() include a redundant "dynamo_" prefix when the namespace
"dynamo" is also being set via the .namespace("dynamo") call, resulting in
doubled namespace prefixes like "dynamo_dynamo_audit_s3_segments_total". Remove
the "dynamo_" prefix from the metric name strings in all Opts::new() calls (such
as in the IntCounterVec creation and other similar metric definitions) while
keeping the .namespace("dynamo") calls unchanged, so metric names should start
with just "audit_" instead of "dynamo_audit_". Apply this fix to all metric
definitions mentioned in the file.
Comment on lines
+117
to
+119
| ``` | ||
| <prefix>/[<deployment>/]YYYY/MM/DD/HH/<pod-name>-<startup-uuid8>-<seq>.jsonl.gz | ||
| ``` |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add a language tag to the fenced key-format block.
Line 117 uses an unlabeled fenced code block, which triggers markdownlint MD040.
💡 Suggested fix
-```
+```text
<prefix>/[<deployment>/]YYYY/MM/DD/HH/<pod-name>-<startup-uuid8>-<seq>.jsonl.gz</details>
<!-- suggestion_start -->
<details>
<summary>📝 Committable suggestion</summary>
> ‼️ **IMPORTANT**
> Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
```suggestion
🧰 Tools
🪛 markdownlint-cli2 (0.22.1)
[warning] 117-117: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/llm/src/audit/README.md` around lines 117 - 119, The fenced code block
containing the file path format string is missing a language tag, which violates
markdownlint rule MD040. Add the language identifier "text" to the opening fence
(the triple backticks) that precedes the line with the prefix format containing
the pod-name and startup-uuid8 variables. This will properly label the code
block as a text format example.
Source: Linters/SAST tools
Comment on lines
+212
to
+223
| let writer = JsonlGzipWriter::with_segment_sink( | ||
| segment_sink, | ||
| JsonlGzipSinkOptions { | ||
| buffer_bytes: policy.s3_batch_bytes, | ||
| // The buffer flush cadence keeps in-memory bytes from | ||
| // sitting around between PUTs; segment rotation is | ||
| // controlled by roll_bytes / roll_lines / roll_interval. | ||
| flush_interval: Duration::from_millis(policy.jsonl_flush_interval_ms.max(1)), | ||
| roll_uncompressed_bytes: policy.s3_roll_bytes, | ||
| roll_lines: policy.s3_roll_lines, | ||
| roll_interval: Some(Duration::from_millis(policy.s3_roll_interval_ms.max(1))), | ||
| }, |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
DYN_AUDIT_S3_CHANNEL_CAPACITY is not wired into the S3 sink queue.
AuditPolicy exposes s3_channel_capacity, but this path always constructs JsonlGzipWriter::with_segment_sink(...) without a capacity parameter, and the writer uses a hardcoded channel size (2048). That makes the env/config knob ineffective for backpressure and memory tuning.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/llm/src/audit/sink.rs` around lines 212 - 223, The
JsonlGzipWriter::with_segment_sink() constructor call is not using the
configurable s3_channel_capacity parameter from the AuditPolicy, which means the
DYN_AUDIT_S3_CHANNEL_CAPACITY configuration knob has no effect and a hardcoded
channel size is used instead. Add the s3_channel_capacity field from the policy
object as a parameter to the JsonlGzipWriter::with_segment_sink() call to wire
the configuration into the S3 sink queue and allow proper backpressure and
memory tuning.
YiqiuLiu
force-pushed
the
feat/s3-audit-sink
branch
3 times, most recently
from
June 23, 2026 22:40
d94cd46 to
6a38539
Compare
![dynamo-review-agent[bot]](https://avatars.githubusercontent.com/u/201626793?s=60&v=4){kind=small}
| #[serde(skip_serializing_if = "Option::is_none")] | ||
| pub deployment: Option<String>, | ||
| /// UTC wall-clock unix-millis stamped at `AuditHandle::emit()`. | ||
| pub emitted_at_unix_ms: i64, |
There was a problem hiding this comment.
Adding emitted_at_unix_ms as a required deserialization field breaks replay/reading of existing audit records that were written without it. Fix: give the field a serde default or make it optional.
🤖 AI Fix
In lib/llm/src/audit/handle.rs, add #[serde(default)] above AuditRecord::emitted_at_unix_ms so older audit JSON without this field still deserializes.
| let path = segment_path(&self.base_path, self.current_index); | ||
| let batch = std::mem::take(&mut self.batch); | ||
| let gz_bytes = compress_member(batch).context("compressing gzip jsonl batch")?; | ||
| let seq = self.current_seq; |
There was a problem hiding this comment.
Gzip compression now runs synchronously inside the Tokio async writer task, which can block executor threads for large audit/request-trace batches. Fix: move compress_member(batch) into tokio::task::spawn_blocking before appending to the segment sink.
🤖 AI Fix
In lib/llm/src/telemetry/jsonl_gz.rs, update GzipBatchWriter::flush_batch to run compress_member(batch) via tokio::task::spawn_blocking(move || compress_member(batch)), await it with context/error handling, then pass the resulting bytes to append_to_segment.
| /// Construct a writer that hands each rotated segment to the supplied | ||
| /// [`SegmentSink`]. Use this for cloud destinations such as S3. | ||
| pub fn with_segment_sink( | ||
| segment_sink: Arc<dyn SegmentSink>, |
There was a problem hiding this comment.
The new DYN_AUDIT_S3_CHANNEL_CAPACITY setting is loaded and documented but ignored because JsonlGzipWriter hard-codes its channel capacity to 2048. Fix: thread a channel capacity option through the gzip writer and pass policy.s3_channel_capacity for the S3 sink.
🤖 AI Fix
Add channel_capacity: usize to JsonlGzipSinkOptions with default 2048, use it in JsonlGzipWriter::with_segment_sink when calling mpsc::channel, and in lib/llm/src/audit/sink.rs set channel_capacity: policy.s3_channel_capacity for S3AuditSink::from_policy.
YiqiuLiu
force-pushed
the
feat/s3-audit-sink
branch
from
June 23, 2026 23:34
6a38539 to
6fe5340
Compare
YiqiuLiu
temporarily deployed
to
external_collaborator
June 23, 2026 23:34 — with 
GitHub Actions
Inactive
GitHub Actions
Inactive
YiqiuLiu
force-pushed
the
feat/s3-audit-sink
branch
from
June 23, 2026 23:38
6fe5340 to
0030f63
Compare
YiqiuLiu
temporarily deployed
to
external_collaborator
June 23, 2026 23:38 — with
GitHub Actions
Inactive
GitHub Actions
Inactive
YiqiuLiu
force-pushed
the
feat/s3-audit-sink
branch
from
June 23, 2026 23:40
0030f63 to
020e9e3
Compare
YiqiuLiu
temporarily deployed
to
external_collaborator
June 23, 2026 23:40 — with
GitHub Actions
Inactive
GitHub Actions
Inactive
YiqiuLiu
force-pushed
the
feat/s3-audit-sink
branch
from
June 23, 2026 23:48
020e9e3 to
6777bea
Compare
YiqiuLiu
temporarily deployed
to
external_collaborator
June 23, 2026 23:48 — with
GitHub Actions
Inactive
GitHub Actions
Inactive
YiqiuLiu
force-pushed
the
feat/s3-audit-sink
branch
from
June 23, 2026 23:51
6777bea to
c34e547
Compare
YiqiuLiu
temporarily deployed
to
external_collaborator
June 23, 2026 23:51 — with
GitHub Actions
Inactive
GitHub Actions
Inactive
YiqiuLiu
force-pushed
the
feat/s3-audit-sink
branch
from
June 24, 2026 00:14
c34e547 to
237616e
Compare
YiqiuLiu
temporarily deployed
to
external_collaborator
June 24, 2026 00:14 — with
GitHub Actions
Inactive
GitHub Actions
Inactive
Add a new S3-backed audit sink as a peer of the existing stderr/nats/
jsonl/jsonl_gz sinks. Selected at runtime via DYN_AUDIT_SINKS=s3.
The sink ships per-request audit records (full request + full response,
already in AuditRecord today) to S3 as rotated, gzip-compressed NDJSON
objects. Object keys partition by deployment + date for Athena/Glue and
include a per-process startup uuid so pod restarts cannot collide:
{prefix}/[{deployment}/]YYYY/MM/DD/HH/{instance}-{startup}-{seq:06}.jsonl.gz
Rotation engine
---------------
The existing rotating gzip writer (telemetry/jsonl_gz.rs) is generalized
behind a new SegmentSink trait so it can drive both disk and S3:
trait SegmentSink {
async fn append_to_segment(&self, seq, gz_bytes);
async fn close_segment(&self, seq);
}
FileSegmentSink keeps the existing on-disk behavior used by jsonl_gz
audit and request_trace (no callers changed). S3SegmentSink is a peer
that buffers the segment in memory across appends and uploads on close.
The writer also gains an optional roll_interval for time-based
rotation, default None (unchanged for jsonl_gz). The S3 sink defaults
it to 60s so quiet periods do not leave records in memory.
Hot-path safety
---------------
All I/O is off the request path. emit() pushes serialized records into
a bounded mpsc that an uploader task consumes; PutObject runs there
with the SDK's standard retry policy (3 attempts, exponential backoff
+ jitter). On terminal failure: log + drop the segment, never
propagate. The audit bus already drops the oldest records on
back-pressure, so a slow S3 cannot stall inference.
Sampling
--------
A new global DYN_AUDIT_SAMPLE_RATE (default 1.0) gates handle creation
via a deterministic xxh3 hash of request_id. Bypassed by force_logging
so debug/compliance flags always capture, but not by request.store
alone (store is a normal opt-in; sampling still applies). Unsampled
requests skip every Arc clone of the request body.
AuditRecord schema
------------------
Two new fields:
deployment: Option<String> // skip-if-none
emitted_at_unix_ms: i64 // UTC wall-clock at emit()
The deployment field is auto-detected from DYN_PARENT_DGD_K8S_NAME on
Kubernetes (the Dynamo operator unconditionally injects this env var
on every worker pod with the parent DynamoGraphDeployment CR name).
DYN_AUDIT_DEPLOYMENT remains an explicit override for non-K8s deploys
or test environments.
The per-record gzip envelope timestamp also moves from millis-since-
process-start to UTC unix-millis: process-start-relative timestamps
are not portable across pod lifetimes for an audit log shipped
off-host.
New environment variables
-------------------------
DYN_AUDIT_SAMPLE_RATE
DYN_AUDIT_DEPLOYMENT # optional override; auto = DYN_PARENT_DGD_K8S_NAME
DYN_AUDIT_S3_BUCKET # required when sinks=s3
DYN_AUDIT_S3_PREFIX # default "dynamo-audit"
DYN_AUDIT_S3_REGION
DYN_AUDIT_S3_ENDPOINT_URL # for LocalStack/MinIO
DYN_AUDIT_S3_BATCH_BYTES # default 1 MiB
DYN_AUDIT_S3_ROLL_BYTES # default 64 MiB compressed
DYN_AUDIT_S3_ROLL_INTERVAL_MS # default 60_000
DYN_AUDIT_S3_ROLL_LINES
DYN_AUDIT_S3_SSE # AES256 (default), aws:kms, or "none"
DYN_AUDIT_S3_KMS_KEY_ID
DYN_AUDIT_S3_INSTANCE_ID # override; defaults to POD_NAME / hostname
DYN_AUDIT_S3_CHANNEL_CAPACITY # default 4096
All consts registered in lib/runtime/src/config/environment_names.rs.
Identity resolution
-------------------
S3 object keys identify the writer with full pod name + 8-char startup
uuid. Resolution chain at sink construction:
DYN_AUDIT_S3_INSTANCE_ID -> POD_NAME (K8s downward API) -> HOSTNAME
-> /etc/hostname -> "unknown"
Encryption
----------
Server-side encryption is configurable: AES256 (default), aws:kms with
optional KMS key ARN, or none (LocalStack/MinIO compatibility). No
client-side encryption in v1.
Cargo dependencies
------------------
aws-sdk-s3 1.120.0 and aws-config 1.8.11 added to workspace and to
lib/llm/Cargo.toml (versions match the existing pin in lib/kvbm-engine
to share the SDK compile). No Cargo feature gate, matching the
precedent set by NVIDIA's OTLP audit-sink PR.
Tests
-----
- 4 new unit tests in audit::handle for sampling and deployment.
- 5 new unit tests in telemetry::s3_segment_sink for key formatting
and identity fallback.
- 3 new unit tests in telemetry::jsonl_gz for the SegmentSink trait
(collecting fake) and time-based rotation.
- 1 #[ignore]'d integration test in tests/audit_s3_integration.rs that
drives the full bus -> sink -> S3 PUT path against a LocalStack /
MinIO endpoint and asserts:
* NDJSON lines reach the destination, gzip-decoded correctly
* deployment field is populated in the record envelope
* deployment partition appears in the object key
Verified end-to-end against MinIO at localhost:4566 (see file
module-level docs for setup).
All existing tests stay green: 15 audit, 19 telemetry, 29 request_trace.
Clippy and rustfmt clean.
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
force-pushed
the
feat/s3-audit-sink
branch
from
June 24, 2026 00:39
237616e to
c75a75c
Compare
YiqiuLiu
temporarily deployed
to
external_collaborator
June 24, 2026 00:39 — with
GitHub Actions
Inactive
GitHub Actions
Inactive
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 16, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 16, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 16, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 16, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
10 tasks
Contributor
Author
|
PR outdated, check #11768 for details |
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 16, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 17, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 17, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 21, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 21, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 21, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 24, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 27, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 27, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 27, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 28, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
YiqiuLiu
added a commit
to YiqiuLiu/dynamo
that referenced
this pull request
Jul 29, 2026
Adds an S3 destination for `dynamo.request.trace.v1` records as a
new `RequestTraceSink` implementation, parallel to the existing
OTLP sink. Same record shape as every other sink -- new destination,
no schema change.
Records are batched in-process as gzipped JSONL and each finished
batch is uploaded via a single `PutObject`. Object keys use a simple
time-based layout for this PR (`{prefix}/{yyyy}/{mm}/{dd}/{host}-
{HHMMSS}-{seq}.jsonl.gz`); richer partitioning (model/day/hour Hive
style), tunable roll thresholds, and Prometheus metrics ship 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
documented in a separate PR, not this sink's.
The `request-trace-s3` cargo feature is off by default so the AWS
SDK dependency is opt-in. When the feature is off, selecting
`DYN_REQUEST_TRACE_SINKS=...,s3` fails startup with a clear error.
Refs: ai-dynamo#11768 (supersedes ai-dynamo#10903).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Yiqiu Liu <yiqiuliu@amazon.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Add an
s3audit sink peer to the existingstderr/nats/jsonl/jsonl_gzsinks. Selected at runtime viaDYN_AUDIT_SINKS=s3. Records are uploaded as rotated, gzip-compressed NDJSON segments to S3 (or any S3-compatible endpoint), partitioned by deployment and date.Also includes:
SegmentSinktrait extraction fromJsonlGzipWriterso the rotation engine can drive both disk and cloud destinationsDYN_AUDIT_SAMPLE_RATEgate withforce_loggingbypassdeploymentlabel from operator-injectedDYN_PARENT_DGD_K8S_NAMEspec.audit.aws_s3.irsaRoleArn) for cloud-identity SA managementMotivation
Three offline consumer patterns the existing sinks don't serve:
All three want records in cheap durable storage queryable by Athena/Glue/Spark/jq. S3 is the canonical answer.
Changes
Rust (
lib/llm/)src/telemetry/jsonl_gz.rsSegmentSinktrait +FileSegmentSink. Addroll_intervalfor time-based rotation. Existing callers unchanged.src/telemetry/s3_segment_sink.rsS3SegmentSinkimpl usingaws-sdk-s3. Buffers segments in memory, uploads onclose_segment. Key format:{prefix}/[{deployment}/]YYYY/MM/DD/HH/{pod}-{uuid8}-{seq}.jsonl.gzsrc/audit/sink.rsS3AuditSinktrait impl +"s3"match arm inparse_sinks_from_env. Bus-lag metrics wiring.src/audit/handle.rsAuditRecordgainsdeployment: Option<String>+emitted_at_unix_ms: i64.create_handleadds xxh3-based sampling gate.src/audit/config.rsAuditPolicygains S3 fields +sample_rate+deployment(auto fromDYN_PARENT_DGD_K8S_NAME).src/audit/metrics.rsdynamo_audit_s3_segments_total,dynamo_audit_records_dropped_total,dynamo_audit_s3_upload_duration_seconds,dynamo_audit_s3_segment_size_bytes.src/audit/README.mdterminationGracePeriodSecondsguidance.tests/audit_s3_integration.rs#[ignore]'d end-to-end test against LocalStack/MinIO.Cargo.toml(root + lib/llm)aws-sdk-s3,aws-configworkspace deps.lib/runtime/src/config/environment_names.rsDYN_AUDIT_S3_*+DYN_AUDIT_SAMPLE_RATE+DYN_AUDIT_DEPLOYMENTconsts.Go (
deploy/operator/)api/v1beta1/dynamographdeployment_types.goAuditSpecwith per-cloud sub-blocks (AwsS3,GcpGcs,AzureBlob). Each carries only the cloud-identity field.internal/controller/dynamographdeployment_controller.goreconcileAuditIdentity()— attaches/removes cloud-identity annotations on the existing<dgd>-k8s-service-discoverySA based onspec.audit. Non-fatal on error.internal/controller/audit_identity_test.goDesign decisions
emit()is a non-blockingmpsc::send. PutObject runs on a background task with SDK retry (3 attempts, exponential backoff). On terminal failure: log + drop. The audit bus drops oldest records on lag — S3 can never stall inference.<dgd>-k8s-service-discoverySA rather than creating a separate audit SA. A K8s SA can carry one IRSA-style annotation per cloud (distinct keys) while being bound to multiple K8s Roles.spec.audit.aws_s3.irsaRoleArnis type-safe and extensible. Future clouds add their own sub-blocks without changing the operator's core logic.DYN_AUDIT_S3_*env vars — they change more often than identity and shouldn't require operator reconciliation.xxh3(request_id) % 10000— same ID always produces the same keep/drop decision.force_logging=truebypasses sampling;request.store=truedoes not.Test plan
cargo fmt --checkclean#[ignore]'d integration test passes against MinIO at localhost:4566New environment variables
🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Documentation