feat(ingester): add the OTLP/HTTP listener (RFC0003.13/.14 + .11 HTTP arms) - #135
Conversation
… arms) Sixth green slice of the OTLP receiver (RFC 0003 §6.2). Adds `receiver::http`: an axum POST handler (default `/v1/logs`, configurable) wrapping the IngestPipeline behind a shared Mutex (single-writer WAL ⇒ requests serialize). It dispatches on Content-Type (application/x-protobuf → decode_protobuf; application/json → decode_json; else 415), handles Content-Encoding identity + gzip via flate2 (unsupported → 415; corrupt gzip → 400), 400s a malformed body, 413s an oversize body (DefaultBodyLimit), 400s a tenant-resolution failure, 404s an unconfigured path, and on success returns a 200 ExportLogsServiceResponse (partial_success unset) encoded per the request format. Flips RFC0003.13 (identity + gzip decode to the same request; unsupported encoding → 415) and RFC0003.14 (default path handled; other path 404; operator override). HTTP transport-error arms of RFC0003.11 (malformed 400, bad/missing Content-Type 415, corrupt gzip 400, wrong path 404, oversize 413, no WAL frame appended) land in tests/http_transport_errors; rfc0003_11 itself stays `#[ignore]`'d until its gRPC cancellation arm exists. All HTTP tests run in-process via tower::ServiceExt::oneshot (no socket). `Journal` gains a `Send` bound so the pipeline can be shared state. Adds axum (0.8, http1+tokio) + flate2 (rust_backend) deps; tower as a dev-dep. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
@coderabbitai review |
|
Warning Review limit reached
More reviews will be available in 49 minutes and 54 seconds. Learn how PR review limits work. Your organization has run out of usage credits. Purchase more in the billing tab. ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans include higher PR review limits than trial, open-source, and free plans. In all cases, reviews become available again over time. During sustained high-volume PR review activity, CodeRabbit may temporarily slow when the next review becomes available. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughThis PR implements a complete OTLP/HTTP receiver for the ourios ingester using Axum, featuring configurable routing at ChangesOTLP/HTTP Receiver with RFC Compliance
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 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 |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
Pull request overview
Adds an OTLP/HTTP receiver to ourios-ingester (RFC0003 §6.2), including routing at a configurable path, request decoding by Content-Type/Content-Encoding, controlled HTTP error mapping (RFC0003.11 HTTP arms), and in-process axum router tests that flip RFC0003.13/.14 green.
Changes:
- Introduce
receiver::httpaxum router/handler for OTLP/HTTP (/v1/logsby default) with gzip support and body-size limiting. - Add new acceptance + transport-error tests for RFC0003.13/.14 and RFC0003.11 HTTP error arms, plus shared HTTP test utilities.
- Extend
Journalwith aSendbound and wire new dependencies (axum,flate2,towerdev-dep).
Reviewed changes
Copilot reviewed 9 out of 10 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| crates/ourios-ingester/src/receiver/http.rs | New OTLP/HTTP router + handler implementing RFC-controlled HTTP behavior. |
| crates/ourios-ingester/src/receiver/pipeline.rs | Adds Send bound to Journal for sharing pipeline state in listeners. |
| crates/ourios-ingester/src/receiver.rs | Exposes the new receiver::http module and updates module docs. |
| crates/ourios-ingester/src/lib.rs | Updates crate-level docs to include OTLP/HTTP listener slice. |
| crates/ourios-ingester/tests/ingest_support/mod.rs | Adds HTTP test helpers (router driving, gzip helper, capturing journal). |
| crates/ourios-ingester/tests/rfc0003_13_compression.rs | Flips RFC0003.13 live test (identity vs gzip equivalence + 415 on unsupported). |
| crates/ourios-ingester/tests/rfc0003_14_path_config.rs | Flips RFC0003.14 live test (default path + operator override semantics). |
| crates/ourios-ingester/tests/http_transport_errors.rs | New tests for RFC0003.11 HTTP transport error arms (400/413/415/404, no WAL append). |
| crates/ourios-ingester/Cargo.toml | Adds axum + flate2 deps and tower dev-dependency for in-process router testing. |
| Cargo.lock | Locks new transitive dependencies for the HTTP listener stack. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| // WAL-before-ack ingest. The lock spans only this synchronous call. | ||
| let outcome = pipeline | ||
| .lock() | ||
| .expect("ingest pipeline lock not poisoned") | ||
| .ingest(request); | ||
|
|
There was a problem hiding this comment.
Fixed — the handler now recovers the guard via PoisonError::into_inner instead of expect, so a poisoned lock cannot panic here; the "no panics" promise holds.
| WireFormat::Json => ( | ||
| StatusCode::OK, | ||
| [(header::CONTENT_TYPE, "application/json")], | ||
| serde_json::to_vec(&response).unwrap_or_default(), | ||
| ) | ||
| .into_response(), |
There was a problem hiding this comment.
Fixed — the JSON arm now returns 500 on a serialize failure rather than a 200 with an empty body (encoding the trivial default response should never fail, but a 500 is the honest fallback).
| /// `Send` so the pipeline can live behind an `Arc<tokio::Mutex<_>>` as | ||
| /// shared state in the async HTTP/gRPC listeners. |
There was a problem hiding this comment.
Fixed the doc — the listener uses a std::sync::Mutex (the lock never spans an .await), so the bound doc now says Arc<Mutex<_>>, not tokio::Mutex.
Review fixes: - The handler promised "no panics" but `lock().expect()` panics on a poisoned mutex; recover the guard via `PoisonError::into_inner` so a prior panic can't cascade into a panic here. - `success_response`'s JSON arm used `to_vec().unwrap_or_default()`, which could return a 200 with an empty body on a (theoretical) serialize failure; return 500 instead — never a 200 without the response. - The `Journal: Send` doc said `Arc<tokio::Mutex<_>>`; the listener uses a `std::sync::Mutex` (the lock never spans an await), so the doc now says `Arc<Mutex<_>>`. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
| // WAL-before-ack ingest. The lock spans only this synchronous call. | ||
| // Recover the guard even if a prior holder panicked: a poisoned lock | ||
| // must not turn into a panic here (the handler promises not to), so | ||
| // take the inner guard regardless and let this request proceed. | ||
| let outcome = pipeline | ||
| .lock() | ||
| .unwrap_or_else(std::sync::PoisonError::into_inner) | ||
| .ingest(request); | ||
|
|
There was a problem hiding this comment.
Fixed in a49aaa1 — the blocking ingest (WAL append + fsync) now runs via tokio::task::spawn_blocking, so it no longer stalls the async runtime; a blocking-task panic maps to 500.
| match media_type { | ||
| "application/x-protobuf" => Some(WireFormat::Protobuf), | ||
| "application/json" => Some(WireFormat::Json), | ||
| _ => None, | ||
| } |
There was a problem hiding this comment.
Fixed in a49aaa1 — content_type now lowercases the media type before matching (HTTP media types are case-insensitive). New test asserts Application/X-Protobuf is accepted.
| Some(value) => match value.to_str().ok()?.trim() { | ||
| "" | "identity" => Some(Encoding::Identity), | ||
| "gzip" => Some(Encoding::Gzip), | ||
| _ => None, | ||
| }, |
There was a problem hiding this comment.
Fixed in a49aaa1 — content_encoding now matches case-insensitively, so GZIP/Identity are accepted (covered by the same case-insensitivity test).
| use std::io::Read; | ||
| let mut decoder = flate2::read::GzDecoder::new(bytes); | ||
| let mut out = Vec::new(); | ||
| decoder.read_to_end(&mut out)?; | ||
| Ok(out) |
There was a problem hiding this comment.
Good catch (decompression bomb) — gunzip now caps decompression at the configured body size via Read::take, returning 413 if the body inflates past it (DefaultBodyLimit only bounds the compressed bytes). New test: a ~1 KB gzip that inflates to 1 MB under a 4 KiB cap → 413.
![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
🤖 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 `@crates/ourios-ingester/src/receiver/http.rs`:
- Around line 123-131: The content_type function currently does exact-case
matching which will reject valid headers; update content_type to perform
case-insensitive comparisons by normalizing the media_type (e.g.,
media_type.to_ascii_lowercase()) before matching against
"application/x-protobuf" and "application/json". Do the same for the
Content-Encoding handling (the code block around the content-encoding check,
likely a function or branch that inspects header::CONTENT_ENCODING) by comparing
tokens with eq_ignore_ascii_case or converting to_ascii_lowercase and matching
"gzip" etc., so both Content-Type and Content-Encoding are treated
case-insensitively.
- Around line 84-93: The gzip branch (content_encoding + Encoding::Gzip using
gunzip and reading into an unbounded Vec) can be exploited to decompress huge
data and exhaust memory; update the decompression to enforce an inflated-size
limit (based on config.max_body_bytes or a configured max_inflated_bytes) by
streaming the gzip decoder and stopping/read-checking bytes as you append,
returning StatusCode::PAYLOAD_TOO_LARGE (or BAD_REQUEST per policy) when the
limit is exceeded; implement the bound either inside gunzip or replace its use
in the Encoding::Gzip arm (referencing content_encoding, Encoding::Gzip, gunzip,
and the raw variable) so decompression never grows unbounded.
🪄 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 Plus
Run ID: 2cdb80d5-d031-4568-bba8-a3bc28573715
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (9)
crates/ourios-ingester/Cargo.tomlcrates/ourios-ingester/src/lib.rscrates/ourios-ingester/src/receiver.rscrates/ourios-ingester/src/receiver/http.rscrates/ourios-ingester/src/receiver/pipeline.rscrates/ourios-ingester/tests/http_transport_errors.rscrates/ourios-ingester/tests/ingest_support/mod.rscrates/ourios-ingester/tests/rfc0003_13_compression.rscrates/ourios-ingester/tests/rfc0003_14_path_config.rs
…headers
Second review round on the HTTP listener:
- `ingest` does blocking I/O (WAL append + fsync); run it via
`tokio::task::spawn_blocking` so it doesn't stall the async runtime
(a blocking-task panic maps to 500).
- Bound gzip decompression to the configured body size (DefaultBodyLimit
only bounds the *compressed* body): a body inflating past the cap → 413,
defusing a decompression bomb. New test asserts a tiny gzip that
inflates to 1 MB under a 4 KiB cap → 413.
- Content-Type and Content-Encoding are matched case-insensitively (both
are case-insensitive per HTTP); new test for uppercased values.
The decompressed cap rides on a small `AppState { pipeline,
max_decompressed_bytes }` so the handler sees the limit the body layer
also enforces.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
What
Sixth green slice of the OTLP receiver (RFC 0003 §6.2): the OTLP/HTTP listener (
axum), flipping RFC0003.13/.14 live and covering RFC0003.11's HTTP error arms.Design
receiver::http::routermounts aPOSThandler at a configurable path (default/v1/logs) wrappingIngestPipelinebehind a sharedMutex(the single-writer WAL serializes concurrent requests; the lock never spans an.await, so a plainstd::sync::Mutexsuffices). The handler:application/x-protobuf(decode_protobuf) /application/json(decode_json) / else 415;DefaultBodyLimit), tenant-resolution failure → 400, unconfigured path → 404;ExportLogsServiceResponse(partial_successunset) encoded per the request format.Scenarios
/v1/logshandled; other path 404; operator override path handled and the default then 404s.tests/http_transport_errors, not flippingrfc0003_11) — malformed 400, bad/missing Content-Type 415, corrupt gzip 400, wrong path 404, oversize 413, each asserting noOtlpBatchframe is appended.rfc0003_11stays#[ignore]'d until its gRPC cancellation arm exists (next slice), keeping the §5 acceptance gate honest. All HTTP tests run in-process viatower::ServiceExt::oneshot(no socket).Notes
Journalgains aSendbound (pipeline as shared state). Deps:axum0.8 (http1+tokio) +flate2(rust_backend);towerdev-dep.ingestunder the lock; a production multi-threaded server should offload viaspawn_blocking— wired when the server binary lands (§9 process-model).Verification
cargo test -p ourios-ingester✓ — RFC0003.1/.3–.10/.12/.13/.14 live; 3 ignored (.2/.11/.15). Workspace green.cargo fmt --all --check✓ ·cargo clippy --all-targets --all-features -- -D warnings✓🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Tests