feat(ingester): add the OTLP/gRPC LogsService listener (RFC0003.11/.15) - #136
Conversation
Seventh green slice of the OTLP receiver (RFC 0003 §6.2). `receiver::grpc` implements opentelemetry-proto's logs_service_server::LogsService over the shared IngestPipeline: export() hands the tonic-decoded ExportLogsServiceRequest to the WAL-before-ack pipeline (via spawn_blocking, mirroring the HTTP handler; poison-recovered, panic-free) and maps the result — tenant-resolution failure → Status::invalid_argument naming the failing ResourceLogs index + attribute (RFC0003.4), WAL failure → Status::internal, success → an empty ExportLogsServiceResponse. Flips RFC0003.11 fully: HTTP arms in tests/http_transport_errors (prior slice) + gRPC arms here (tenant failure → INVALID_ARGUMENT, never a panic; valid request succeeds). The "gRPC client cancellation mid-decode" arm can't be reproduced by an in-process direct call (tonic decodes before the handler); the test documents this honestly and asserts the testable invariants (panic-free; atomic ingest under the lock leaves no partial WAL state) — flagged as an OTLP/tonic nuance, not faked. Flips RFC0003.15: N concurrent Export calls on one shared pipeline (multi-thread runtime) each ack, and the WAL ends with exactly N durable OtlpBatch frames (the single-writer WAL serializes; each acks after its own append+fsync). Adds the `gen-tonic` feature on opentelemetry-proto + tonic (codegen). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
@coderabbitai review |
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (9)
📝 WalkthroughWalkthroughThis PR adds gRPC receiver support to ourios-ingester by implementing a ChangesgRPC receiver implementation
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 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 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 |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
Pull request overview
Adds the OTLP/gRPC LogsService receiver for the ingester, wiring the existing WAL-before-ack IngestPipeline behind a tonic service implementation and turning previously ignored RFC0003.11/.15 scenarios into green tests.
Changes:
- Introduces
receiver::grpc::LogsReceiverimplementinglogs_service_server::LogsService, offloading ingest tospawn_blockingand mappingReceiveErrorto gRPCStatus. - Enables
tonic/opentelemetry-protocodegen needed for the generatedLogsServicetrait and updates crate docs to include the new transport. - Adds/updates RFC0003 acceptance tests for gRPC error mapping and concurrent WAL durability under concurrent
Exportcalls.
Reviewed changes
Copilot reviewed 7 out of 8 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| crates/ourios-ingester/src/receiver/grpc.rs | New OTLP/gRPC LogsService implementation over the shared WAL-before-ack pipeline. |
| crates/ourios-ingester/src/receiver.rs | Exposes the new grpc module and updates receiver module docs accordingly. |
| crates/ourios-ingester/src/lib.rs | Updates crate-level docs to include the gRPC listener as landed. |
| crates/ourios-ingester/Cargo.toml | Switches opentelemetry-proto to gen-tonic and adds tonic (codegen-only) dependency. |
| crates/ourios-ingester/tests/rfc0003_11_transport_errors.rs | Converts RFC0003.11 gRPC transport-error expectations into executable tests. |
| crates/ourios-ingester/tests/rfc0003_15_concurrent_wal_before_ack.rs | Converts RFC0003.15 concurrency/durability expectations into a multi-threaded tokio test. |
| crates/ourios-ingester/tests/ingest_support/mod.rs | Adds helper to build a shared pipeline over a real WAL for concurrency/durability assertions. |
| Cargo.lock | Locks the new tonic dependency. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| use crate::receiver::http::SharedPipeline; | ||
| use crate::receiver::pipeline::ReceiveError; |
There was a problem hiding this comment.
Fixed — SharedPipeline moved to receiver::pipeline (alongside IngestPipeline) and re-exported from receiver; grpc no longer depends on http for it. Both transports import it from pipeline.
| // A WAL failure (or a panic in the blocking task) is | ||
| // server-side; the batch was not acked (§3.4). | ||
| Ok(Err(_)) | Err(_) => Err(Status::internal("ingest failed")), |
There was a problem hiding this comment.
Fixed — the INTERNAL arm now distinguishes: a WAL append/sync failure surfaces its Display detail via Status::internal(e.to_string()) (AppendError/SyncError gained Display in #134), and a blocking-task join failure (panic/cancel) is a separate Status::internal("ingest task failed"). TenantResolution stays INVALID_ARGUMENT.
…C INTERNAL
Review fixes on the gRPC slice:
- `SharedPipeline` lived in `receiver::http`, so `receiver::grpc` imported
the HTTP module just for it — a wrong dependency direction. Moved the
alias to `receiver::pipeline` (where `IngestPipeline` lives) and
re-export it from `receiver`; both transports import it from there.
- The gRPC `export` collapsed WAL-append, WAL-sync, and join (panic/cancel)
failures into one opaque `INTERNAL("ingest failed")`. Now a WAL failure
surfaces its Display detail via `Status::internal(e.to_string())`, and a
blocking-task join failure is a distinct `INTERNAL("ingest task failed")`.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
| // Assert: a controlled INVALID_ARGUMENT naming the attribute — not a | ||
| // panic — and nothing appended. | ||
| assert_eq!(status.code(), Code::InvalidArgument); | ||
| assert!( | ||
| status.message().contains("service.name"), | ||
| "the Status names the missing attribute, got {:?}", | ||
| status.message(), | ||
| ); |
There was a problem hiding this comment.
Fixed — rfc0003_11 now also asserts the Status message contains ResourceLogs[0] (the failing index), covering the full RFC0003.4/.11 contract (index + attribute), not just the attribute.
| // The blocking ingest task panicked or was cancelled. | ||
| Err(_) => Err(Status::internal("ingest task failed")), |
There was a problem hiding this comment.
Fixed — the join-error arm now includes the JoinError Display: Status::internal(format!("ingest task failed: {join}")). (For spawn_blocking a JoinError is always a panic — blocking tasks aren't cancellable — but the Display still distinguishes it clearly.)
…surface JoinError Review follow-ups: rfc0003_11 now asserts the INVALID_ARGUMENT message names the failing ResourceLogs index (RFC0003.4/.11), not just the attribute; the gRPC join-error arm includes the (short, safe) JoinError Display in the INTERNAL message instead of dropping it. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
| // The blocking ingest task panicked or was cancelled; | ||
| // `JoinError`'s Display is short and safe to surface. |
There was a problem hiding this comment.
Fixed in f4e1c86 — the comment now says the spawn_blocking task panicked (such tasks cannot be cancelled, so a JoinError here is always a panic).
| # (RFC0003.6, lands next). Live gRPC/HTTP listeners (`tonic`/`axum`) are | ||
| # a later transport slice — this slice is the pure decode layer. | ||
| opentelemetry-proto = { version = "0.32", default-features = false, features = ["gen-tonic-messages", "logs", "with-serde"] } | ||
| # `gen-tonic` (not just `-messages`) for the `logs_service_server` | ||
| # `LogsService` trait the gRPC listener implements (RFC0003.11/.15). |
There was a problem hiding this comment.
Fixed in f4e1c86 — dropped the stale "later transport slice / pure decode layer" wording; the comment now just explains gen-tonic supplies the LogsService trait the gRPC listener implements.
| //! - [`http`] — the OTLP/HTTP listener ([`http::router`]) wrapping the | ||
| //! pipeline: `Content-Type`/`Content-Encoding` dispatch, controlled | ||
| //! transport errors, configurable path (RFC0003.11 HTTP arms / .13 / | ||
| //! .14). The gRPC listener (`tonic`) follows. | ||
| //! .14). | ||
| //! - [`grpc`] — the OTLP/gRPC `LogsService` ([`grpc::LogsReceiver`]) |
There was a problem hiding this comment.
Fixed in f4e1c86 — the decode bullet no longer says "No live tonic/axum listener yet"; it now notes the http + grpc transports hand their decoded payload to the shared decode layer.
Three review-surfaced doc/comment staleness fixes: the gRPC join-error comment now says the spawn_blocking task panicked (such tasks cannot be cancelled); the opentelemetry-proto dep comment no longer calls listeners a "later slice"; the receiver decode bullet no longer says "No live tonic/axum listener yet" (http + grpc modules now exist). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
| //! Implements `opentelemetry-proto`'s `LogsService` over the shared | ||
| //! [`IngestPipeline`]: `export` hands the (already tonic-decoded) |
There was a problem hiding this comment.
Fixed in 01c6c3f — these intra-doc links did not resolve from receiver::grpc (it imports SharedPipeline, not IngestPipeline). Made them plain code spans (the http module doc had the same broken link after the SharedPipeline move; fixed there too). cargo doc is now warning-free.
|
|
||
| use crate::receiver::pipeline::{ReceiveError, SharedPipeline}; | ||
|
|
||
| /// The gRPC `LogsService` over a shared [`IngestPipeline`]. |
There was a problem hiding this comment.
Fixed in 01c6c3f — these intra-doc links did not resolve from receiver::grpc (it imports SharedPipeline, not IngestPipeline). Made them plain code spans (the http module doc had the same broken link after the SharedPipeline move; fixed there too). cargo doc is now warning-free.
The grpc + http module docs linked `[`IngestPipeline`]`, which does not resolve from those modules (they import SharedPipeline, not the type). Use plain code spans. Surfaced in review. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
What
Seventh green slice of the OTLP receiver (RFC 0003 §6.2): the OTLP/gRPC
LogsService(tonic), flipping RFC0003.11 fully and RFC0003.15.receiver::grpc::LogsReceiverimplementslogs_service_server::LogsServiceover the sharedIngestPipeline.exporthands the (tonic-decoded) request to the WAL-before-ack pipeline viaspawn_blocking(mirrors the HTTP handler; poison-recovered, panic-free) and maps the result:Status::invalid_argumentnaming the failingResourceLogsindex + attribute (RFC0003.4);Status::internal(not acked, §3.4);ExportLogsServiceResponse.Scenarios
tests/http_transport_errors(prior slice) + gRPC arms here: tenant failure →INVALID_ARGUMENT(never a panic), valid request succeeds. Honest limitation: "gRPC client cancellation mid-decode" can't be reproduced by an in-process direct call (tonicdecodes before the handler); the test documents this and asserts the testable invariants (panic-free; atomic ingest under the lock ⇒ no partial WAL state on a dropped future). The socket-level cancellation path comes with a servedtonicserver (a follow-up) — flagged as an OTLP/tonic nuance, not faked.Exportcalls on one shared pipeline (multi-thread runtime) each ack, and the WAL ends with exactly N durableOtlpBatchframes (single-writer WAL serializes; each acks after its own append+fsync). Stable across repeated runs.Notes
gen-toniconopentelemetry-proto+tonic(codegen). No socket server is stood up in this slice — the service is exercised in-process.Verification
cargo test -p ourios-ingester✓ — only RFC0003.2 (crash-before-ack) remains ignored. Workspace green.cargo fmt --all --check✓ ·cargo clippy --all-targets --all-features -- -D warnings✓🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Tests