Distributed Structured Failure Logs#116
Open
patrickleet wants to merge 1 commit into
Open
Conversation
Implements [[tasks/distributed-structured-failure-logs-1]]
📝 WalkthroughWalkthroughThis PR introduces a ChangesStructured failure-log instrumentation
Estimated code review effort: 4 (Complex) | ~75 minutes Sequence Diagram(s)sequenceDiagram
participant Ingress as HTTP/gRPC/Knative Ingress
participant Service
participant FailureRecord
participant Tracing
Ingress->>Service: dispatch_command / dispatch_message
Service-->>Ingress: HandlerError
Ingress->>FailureRecord: from_handler_error(error, message fields)
FailureRecord->>FailureRecord: sanitize_error_message, classify category/action
FailureRecord->>Tracing: emit() distributed.failure event
Ingress-->>Ingress: return client-facing sanitized response
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
src/microsvc/grpc.rs (1)
288-295: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
session_metadatais duplicated verbatim insrc/microsvc/http.rs(Lines 155-161).Consider hoisting this helper into a shared module to avoid divergence.
🤖 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 `@src/microsvc/grpc.rs` around lines 288 - 295, The session_metadata helper is duplicated in both gRPC and HTTP paths, so hoist the shared logic into a common module and have grpc::session_metadata and the HTTP equivalent call it instead of keeping two verbatim copies. Keep the existing behavior of iterating Session::variables() and cloning key/value pairs, but centralize the implementation so future changes only happen in one place.src/failure_log.rs (1)
637-661: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick win
"[redacted-url]"branch is unreachable.
is_sensitive_token(line 686) already matches any token containing"://", so thesensitivecheck on line 647 is always true for URLs, meaning theelse if lower.contains("://")branch (651-653) that would tag URLs as"[redacted-url]"can never execute — URLs fall through to the generic"[redacted]"tag instead. Not a leak, just dead/misleading code relative to the apparent intent.♻️ Proposed fix
-fn is_sensitive_token(lower: &str) -> bool { - lower.contains("authorization") - || lower == "bearer" - || lower.contains("cookie") - || lower.contains("token") - || lower.contains("secret") - || lower.contains("password") - || lower.contains("passwd") - || lower.contains("api_key") - || lower.contains("apikey") - || lower.contains("access_key") - || lower.contains("session_variables") - || lower.contains("session:") - || lower.contains("x-hasura-") - || lower.contains("://") -} +fn is_sensitive_token(lower: &str) -> bool { + lower.contains("authorization") + || lower == "bearer" + || lower.contains("cookie") + || lower.contains("token") + || lower.contains("secret") + || lower.contains("password") + || lower.contains("passwd") + || lower.contains("api_key") + || lower.contains("apikey") + || lower.contains("access_key") + || lower.contains("session_variables") + || lower.contains("session:") + || lower.contains("x-hasura-") +}🤖 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 `@src/failure_log.rs` around lines 637 - 661, The URL-specific branch in sanitize_error_message is unreachable because is_sensitive_token already treats tokens containing "://" as sensitive, so URLs are always redacted before the later check. Fix this by either removing the dead lower.contains("://") branch or moving the URL handling ahead of the generic sensitive-token logic, and make sure the behavior in sanitize_error_message and is_sensitive_token matches the intended redaction tag for URLs.src/bus/runner.rs (1)
167-253: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicated failure-record construction across Nack/DeadLetter/Park/LogAndAck arms.
record_transport_failure,FailureMessageFields::from_message(received.message()), andemit_transport_failure(...)are repeated almost verbatim in 4 of the 5 arms. The decode-error branch above (lines 75-79) already hoists the equivalent logic once before its innermatch; applying the same structure here removes the duplication without changing behavior (theaction @ Patternbindings simply move to alet action = ...before the match).♻️ Proposed refactor: hoist action/message/record/emit before the inner match
- Err(error) => match options.failure_policy.resolve(&error) { - action @ FailureAction::Nack => { - record_transport_failure(service, transport, error.kind(), action); - let message = FailureMessageFields::from_message(received.message()); - emit_transport_failure(service, transport, message.clone(), &error, action); - let reason = error.to_string(); - settle_and_record( - service, - transport, - kind, - message, - crate::telemetry::transport_outcome::NACK, - crate::telemetry::transport_outcome::NACK, - || received.nack(&reason), - ) - .await?; - } - action @ FailureAction::DeadLetter => { - record_transport_failure(service, transport, error.kind(), action); - let message = FailureMessageFields::from_message(received.message()); - emit_transport_failure(service, transport, message.clone(), &error, action); - let reason = error.to_string(); - settle_and_record( - service, - transport, - kind, - message, - crate::telemetry::transport_outcome::DEAD_LETTER, - crate::telemetry::transport_outcome::DEAD_LETTER, - || received.dead_letter(&reason), - ) - .await?; - } - action @ FailureAction::Park => { - record_transport_failure(service, transport, error.kind(), action); - let message = FailureMessageFields::from_message(received.message()); - emit_transport_failure(service, transport, message.clone(), &error, action); - let reason = error.to_string(); - settle_and_record( - service, - transport, - kind, - message, - crate::telemetry::transport_outcome::PARK, - crate::telemetry::transport_outcome::PARK, - || received.park(&reason), - ) - .await?; - } - FailureAction::LogAndAck => { - record_transport_failure( - service, - transport, - error.kind(), - FailureAction::LogAndAck, - ); - let message = FailureMessageFields::from_message(received.message()); - emit_transport_failure( - service, - transport, - message.clone(), - &error, - FailureAction::LogAndAck, - ); - settle_and_record( - service, - transport, - kind, - message, - crate::telemetry::transport_outcome::ACK, - crate::telemetry::transport_outcome::LOG_AND_ACK, - || received.ack(), - ) - .await?; - } - FailureAction::Stop => { - record_transport_failure(service, transport, error.kind(), FailureAction::Stop); - emit_transport_failure( - service, - transport, - FailureMessageFields::from_message(received.message()), - &error, - FailureAction::Stop, - ); - return Err(error); - } - }, + Err(error) => { + let action = options.failure_policy.resolve(&error); + record_transport_failure(service, transport, error.kind(), action); + let message = FailureMessageFields::from_message(received.message()); + emit_transport_failure(service, transport, message.clone(), &error, action); + match action { + FailureAction::Nack => { + let reason = error.to_string(); + settle_and_record( + service, + transport, + kind, + message, + crate::telemetry::transport_outcome::NACK, + crate::telemetry::transport_outcome::NACK, + || received.nack(&reason), + ) + .await?; + } + FailureAction::DeadLetter => { + let reason = error.to_string(); + settle_and_record( + service, + transport, + kind, + message, + crate::telemetry::transport_outcome::DEAD_LETTER, + crate::telemetry::transport_outcome::DEAD_LETTER, + || received.dead_letter(&reason), + ) + .await?; + } + FailureAction::Park => { + let reason = error.to_string(); + settle_and_record( + service, + transport, + kind, + message, + crate::telemetry::transport_outcome::PARK, + crate::telemetry::transport_outcome::PARK, + || received.park(&reason), + ) + .await?; + } + FailureAction::LogAndAck => { + settle_and_record( + service, + transport, + kind, + message, + crate::telemetry::transport_outcome::ACK, + crate::telemetry::transport_outcome::LOG_AND_ACK, + || received.ack(), + ) + .await?; + } + FailureAction::Stop => return Err(error), + } + }🤖 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 `@src/bus/runner.rs` around lines 167 - 253, The failure-handling branch in runner::handle_error repeats the same transport-failure setup across the Nack, DeadLetter, Park, and LogAndAck arms. Hoist the shared `record_transport_failure`, `FailureMessageFields::from_message(received.message())`, and `emit_transport_failure` work out before the inner match, similar to the earlier decode-error branch, and bind the matched action first so each arm only performs its specific settle/ack/nack/dead-letter/park step. Keep the Stop arm unchanged except for using the shared precomputed message if needed.
🤖 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 `@src/failure_log.rs`:
- Around line 437-456: The feature-gated fields on FailureRecord are triggering
dead_code warnings in normal builds because they are only used when failure-logs
is enabled. Update FailureRecord in failure_log.rs so the unread fields like
message, error_message, transport, http_status_code, grpc_status_code, and the
outbox_* members are also protected by the same feature gate or otherwise marked
to avoid dead_code, keeping the struct clean in non-feature builds.
In `@src/microsvc/service.rs`:
- Around line 626-636: The ingress failure path is emitting duplicate
distributed.failure events because both `service.rs` in
`invoke_with_dispatch_span`/the handler failure flow and the ingress adapters
(`http.rs`, `grpc.rs`, `knative_ingress.rs`) record the same returned error.
Update the `Microsvc` failure emission so only one layer logs/records ingress
errors, either by removing the handler-side `emit_handler_failure` for
ingress-returned errors or by suppressing the adapter-side record in the ingress
wrappers, and keep the change consistent across the `FailureOperation::Dispatch`
and unknown-command handling paths.
---
Nitpick comments:
In `@src/bus/runner.rs`:
- Around line 167-253: The failure-handling branch in runner::handle_error
repeats the same transport-failure setup across the Nack, DeadLetter, Park, and
LogAndAck arms. Hoist the shared `record_transport_failure`,
`FailureMessageFields::from_message(received.message())`, and
`emit_transport_failure` work out before the inner match, similar to the earlier
decode-error branch, and bind the matched action first so each arm only performs
its specific settle/ack/nack/dead-letter/park step. Keep the Stop arm unchanged
except for using the shared precomputed message if needed.
In `@src/failure_log.rs`:
- Around line 637-661: The URL-specific branch in sanitize_error_message is
unreachable because is_sensitive_token already treats tokens containing "://" as
sensitive, so URLs are always redacted before the later check. Fix this by
either removing the dead lower.contains("://") branch or moving the URL handling
ahead of the generic sensitive-token logic, and make sure the behavior in
sanitize_error_message and is_sensitive_token matches the intended redaction tag
for URLs.
In `@src/microsvc/grpc.rs`:
- Around line 288-295: The session_metadata helper is duplicated in both gRPC
and HTTP paths, so hoist the shared logic into a common module and have
grpc::session_metadata and the HTTP equivalent call it instead of keeping two
verbatim copies. Keep the existing behavior of iterating Session::variables()
and cloning key/value pairs, but centralize the implementation so future changes
only happen in one place.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: cfc5b8f6-98c9-4231-aeb6-3b00cc408d3c
📒 Files selected for processing (13)
Cargo.tomldocs/metrics.mddocs/observability.mdsrc/bus/runner.rssrc/failure_log.rssrc/lib.rssrc/microsvc/grpc.rssrc/microsvc/http.rssrc/microsvc/knative_ingress.rssrc/microsvc/service.rssrc/outbox_worker/outbox_dispatch.rssrc/outbox_worker/publish_hook.rssrc/telemetry.rs
Comment on lines
+437
to
+456
| #[derive(Debug, Clone, PartialEq, Eq)] | ||
| pub(crate) struct FailureRecord { | ||
| pub(crate) service_name: String, | ||
| pub(crate) component: FailureComponent, | ||
| pub(crate) operation: FailureOperation, | ||
| pub(crate) category: FailureCategory, | ||
| pub(crate) action: FailureAction, | ||
| pub(crate) retry_class: FailureRetryClass, | ||
| pub(crate) message: FailureMessageFields, | ||
| pub(crate) error_type: &'static str, | ||
| pub(crate) error_message: String, | ||
| pub(crate) http_status_code: Option<u16>, | ||
| pub(crate) grpc_status_code: Option<u16>, | ||
| pub(crate) transport: String, | ||
| pub(crate) outbox_attempt: Option<u32>, | ||
| pub(crate) outbox_max_attempts: Option<u32>, | ||
| pub(crate) outbox_status: String, | ||
| pub(crate) outbox_source_aggregate_type: String, | ||
| pub(crate) outbox_source_sequence: String, | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check whether building without the failure-logs feature warns about unused FailureRecord fields.
rg -n "denied\(warnings\)|deny\(warnings\)" --type=rust -g '!target'Repository: hops-ops/distributed
Length of output: 158
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the file and inspect the relevant symbols.
ast-grep outline src/failure_log.rs --view expanded || true
# Find all uses of FailureRecord and its fields.
rg -n "FailureRecord|message:|error_message|http_status_code|grpc_status_code|transport|outbox_attempt|outbox_max_attempts|outbox_status|outbox_source_aggregate_type|outbox_source_sequence" src tests -g '*.rs' || true
# Inspect the relevant section of src/failure_log.rs.
sed -n '430,820p' src/failure_log.rsRepository: hops-ops/distributed
Length of output: 50376
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Show the two emit_failure implementations and surrounding cfg gates.
sed -n '730,805p' src/failure_log.rs
# Show the FailureRecord impl that populates the fields.
sed -n '458,610p' src/failure_log.rs
# Look for any dead_code-related allowances or crate-level lint settings.
rg -n "dead_code|deny\\(warnings\\)|allow\\(dead_code\\)|expect\\(dead_code\\)" src Cargo.toml -g '!target'Repository: hops-ops/distributed
Length of output: 8196
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the top of src/failure_log.rs for lint attributes and dead_code allowances.
sed -n '1,120p' src/failure_log.rs
# Show the surrounding lines where dead_code is allowed.
sed -n '1,95p' src/failure_log.rs
# Find any crate-level lint configuration that would turn warnings into errors.
rg -n "deny\\(warnings\\)|forbid\\(warnings\\)|deny\\(dead_code\\)|forbid\\(dead_code\\)|allow\\(dead_code\\)" Cargo.toml src -g '!target'Repository: hops-ops/distributed
Length of output: 5573
Guard the feature-gated FailureRecord fields from dead_code warnings.
message, error_message, transport, http_status_code, grpc_status_code, and the outbox_* fields are only read when failure-logs is enabled; in normal builds they stay unread, so this struct will warn unless the fields are gated or explicitly allowed.
🤖 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 `@src/failure_log.rs` around lines 437 - 456, The feature-gated fields on
FailureRecord are triggering dead_code warnings in normal builds because they
are only used when failure-logs is enabled. Update FailureRecord in
failure_log.rs so the unread fields like message, error_message, transport,
http_status_code, grpc_status_code, and the outbox_* members are also protected
by the same feature gate or otherwise marked to avoid dead_code, keeping the
struct clean in non-feature builds.
Comment on lines
+626
to
+636
| let result = self | ||
| .invoke_with_dispatch_span(&message, input, session) | ||
| .await; | ||
| if let Err(error) = &result { | ||
| self.emit_handler_failure( | ||
| FailureOperation::Dispatch, | ||
| FailureMessageFields::from_message(&message), | ||
| error, | ||
| ); | ||
| } | ||
| result |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate the relevant symbols and read the surrounding code.
git ls-files 'src/microsvc/*' | sed -n '1,200p'
printf '\n--- emit_handler_failure references ---\n'
rg -n "emit_handler_failure|FailureRecord::from_handler_error|emit_http_failure|dispatch_command_inner|dispatch_message_inner|dispatch_with_dispatch_span|dispatch_command|dispatch_message" src/microsvc -S
printf '\n--- service.rs around emission sites ---\n'
sed -n '580,740p' src/microsvc/service.rs
printf '\n--- service.rs emit helper ---\n'
sed -n '820,880p' src/microsvc/service.rs
printf '\n--- http.rs relevant section ---\n'
sed -n '1,260p' src/microsvc/http.rs
printf '\n--- grpc.rs relevant section ---\n'
sed -n '1,240p' src/microsvc/grpc.rs
printf '\n--- knative_ingress.rs relevant section ---\n'
sed -n '1,260p' src/microsvc/knative_ingress.rsRepository: hops-ops/distributed
Length of output: 38123
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n--- service.rs emission sites ---\n'
sed -n '596,720p' src/microsvc/service.rs
printf '\n--- service.rs failure helper ---\n'
sed -n '840,870p' src/microsvc/service.rs
printf '\n--- http.rs command handler ---\n'
rg -n "command_handler|emit_http_failure|dispatch_command_inner|dispatch" src/microsvc/http.rs -n -A 40 -B 20
printf '\n--- grpc.rs dispatch path ---\n'
rg -n "FailureRecord::from_handler_error|dispatch_message_inner|dispatch" src/microsvc/grpc.rs -n -A 40 -B 20
printf '\n--- knative_ingress.rs ingress path ---\n'
rg -n "FailureRecord::from_handler_error|dispatch_message_inner|dispatch" src/microsvc/knative_ingress.rs -n -A 40 -B 20Repository: hops-ops/distributed
Length of output: 32747
Avoid duplicate failure events on ingress errors src/microsvc/service.rs:601-636, 678-711 emits FailureComponent::Microsvc for handler and unknown-command errors, and src/microsvc/http.rs, src/microsvc/grpc.rs, and src/microsvc/knative_ingress.rs emit their own ingress record for the same returned error. That produces two distributed.failure events per logical ingress failure and double-counts alerting/log volume. Emit at one layer only.
🤖 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 `@src/microsvc/service.rs` around lines 626 - 636, The ingress failure path is
emitting duplicate distributed.failure events because both `service.rs` in
`invoke_with_dispatch_span`/the handler failure flow and the ingress adapters
(`http.rs`, `grpc.rs`, `knative_ingress.rs`) record the same returned error.
Update the `Microsvc` failure emission so only one layer logs/records ingress
errors, either by removing the handler-side `emit_handler_failure` for
ingress-returned errors or by suppressing the adapter-side record in the ingress
wrappers, and keep the change consistent across the `FailureOperation::Dispatch`
and unknown-command handling paths.
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
failure-logsfeature and a versioneddistributed.failuretracing event schema with bounded sanitization.Acceptance Coverage
Local Verification
cargo fmt --checkcargo test --no-default-features --quietcargo test --features metrics --quietcargo test --features failure-logs,http,grpc --quietcargo check --no-default-features --quietcargo check --features http --quietcargo check --features metrics --quietcargo check --features otel --quietcargo check --features failure-logs --quietgit diff --cached --checkSummary by CodeRabbit
New Features
Bug Fixes
Documentation