diff --git a/Cargo.lock b/Cargo.lock index 9f2a7968e..6c3a11349 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -482,6 +482,52 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", +] + [[package]] name = "base64" version = "0.22.1" @@ -770,6 +816,15 @@ dependencies = [ "rustc_version", ] +[[package]] +name = "crc32fast" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" +dependencies = [ + "cfg-if", +] + [[package]] name = "criterion" version = "0.5.1" @@ -1555,6 +1610,7 @@ version = "1.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" dependencies = [ + "crc32fast", "miniz_oxide", "zlib-rs", ] @@ -1846,6 +1902,12 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "humantime" version = "2.3.0" @@ -1866,6 +1928,7 @@ dependencies = [ "http", "http-body", "httparse", + "httpdate", "itoa", "pin-project-lite", "smallvec", @@ -2230,12 +2293,24 @@ dependencies = [ "twox-hash", ] +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + [[package]] name = "memchr" version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -2479,6 +2554,8 @@ dependencies = [ name = "ourios-ingester" version = "0.0.0" dependencies = [ + "axum", + "flate2", "opentelemetry", "opentelemetry-proto", "opentelemetry_sdk", @@ -2493,6 +2570,7 @@ dependencies = [ "serde_json", "tempfile", "tokio", + "tower", ] [[package]] diff --git a/crates/ourios-ingester/Cargo.toml b/crates/ourios-ingester/Cargo.toml index ba9a6f8b0..d0884c2f3 100644 --- a/crates/ourios-ingester/Cargo.toml +++ b/crates/ourios-ingester/Cargo.toml @@ -53,9 +53,20 @@ prost = { version = "0.14", default-features = false, features = ["std"] } # `Deserialize` impls — the `application/json` transport body is an # `ExportLogsServiceRequest` in the proto3-JSON mapping. serde_json = { version = "1", default-features = false, features = ["std"] } +# HTTP transport (RFC 0003 §6.2): the OTLP/HTTP `/v1/logs` listener. +# `tokio` + `http1` so the router can be served; the handler + extractors +# (`State`, `Bytes`, `HeaderMap`) are core. On hyper 1.x, already in the +# tree. +axum = { version = "0.8", default-features = false, features = ["tokio", "http1"] } +# gzip decode for OTLP/HTTP `Content-Encoding: gzip` (an OTLP MUST, +# RFC0003.13). Pure-Rust backend — no zlib C dependency (CLAUDE.md). +flate2 = { version = "1", default-features = false, features = ["rust_backend"] } [dev-dependencies] tempfile = "3" +# Drive the axum `Router` in-process (`ServiceExt::oneshot`) for the +# HTTP-transport acceptance tests — no real socket. +tower = { version = "0.5", default-features = false, features = ["util"] } # Proptest strategy over the OTLP proto value space for the # wire-decode equivalence scenarios (RFC0003.5/.6 per §8). proptest = "1" diff --git a/crates/ourios-ingester/src/lib.rs b/crates/ourios-ingester/src/lib.rs index 52117483e..8f3724589 100644 --- a/crates/ourios-ingester/src/lib.rs +++ b/crates/ourios-ingester/src/lib.rs @@ -11,9 +11,10 @@ //! (§6.2 wire decode — protobuf + OTLP/JSON, RFC0003.5/.6), //! [`receiver::materialize`] (§6.1 `LogRecord` → `OtlpLogRecord`, //! RFC0003.7–.10), [`receiver::tenant`] (per-`ResourceLogs` tenant -//! derivation + fan-out, RFC0003.3/.4), and [`receiver::pipeline`] -//! (§6.5 WAL-before-ack ingest path, RFC0003.1/.12). The live -//! gRPC/HTTP transports wrap the pipeline next. +//! derivation + fan-out, RFC0003.3/.4), [`receiver::pipeline`] +//! (§6.5 WAL-before-ack ingest path, RFC0003.1/.12), and +//! [`receiver::http`] (the OTLP/HTTP listener, RFC0003.11-HTTP/.13/.14). +//! The gRPC listener follows. //! - **WAL-before-ack** (RFC 0008 / `CLAUDE.md` §3.4) — durability //! before acknowledgement, via the shipped `ourios-wal`. Wired into //! the ingest path by [`receiver::pipeline`]: every non-empty batch is diff --git a/crates/ourios-ingester/src/receiver.rs b/crates/ourios-ingester/src/receiver.rs index c2f02d990..945b3db11 100644 --- a/crates/ourios-ingester/src/receiver.rs +++ b/crates/ourios-ingester/src/receiver.rs @@ -19,10 +19,14 @@ //! (RFC0003.3/.4). //! - [`pipeline`] — the §6.5 WAL-before-ack ingest path //! ([`pipeline::IngestPipeline`]): fan out → append one `OtlpBatch` -//! frame → fsync → miner → ack (RFC0003.1/.12). The live gRPC/HTTP -//! transports wrap this layer next. +//! frame → fsync → miner → ack (RFC0003.1/.12). +//! - [`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. pub mod decode; +pub mod http; pub mod materialize; pub mod pipeline; pub mod tenant; diff --git a/crates/ourios-ingester/src/receiver/http.rs b/crates/ourios-ingester/src/receiver/http.rs new file mode 100644 index 000000000..bf0672242 --- /dev/null +++ b/crates/ourios-ingester/src/receiver/http.rs @@ -0,0 +1,222 @@ +//! OTLP/HTTP listener (RFC 0003 §6.2). +//! +//! An `axum` `POST` handler at a configurable path (default `/v1/logs`) +//! that decodes the body per `Content-Type` (`application/x-protobuf` or +//! `application/json`) and `Content-Encoding` (`identity` or `gzip`), +//! hands the decoded `ExportLogsServiceRequest` to the [`IngestPipeline`] +//! (WAL-before-ack), and returns an `ExportLogsServiceResponse`. +//! +//! Transport errors are controlled (RFC0003.11): unsupported media type +//! / encoding → 415, malformed body → 400, oversize → 413, an +//! unconfigured path → 404, tenant-resolution failure → 400. No panics. +//! +//! The pipeline is shared behind a `Mutex` — the WAL is a single writer +//! (RFC 0008 §3.1), so concurrent requests serialize on it. The lock is +//! never held across an `.await`, so a plain `std::sync::Mutex` suffices. + +use std::sync::{Arc, Mutex}; + +use axum::Router; +use axum::body::Bytes; +use axum::extract::{DefaultBodyLimit, State}; +use axum::http::{HeaderMap, StatusCode, header}; +use axum::response::{IntoResponse, Response}; +use axum::routing::post; +use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceResponse; +use prost::Message; + +use crate::receiver::decode::{decode_json, decode_protobuf}; +use crate::receiver::pipeline::{IngestPipeline, ReceiveError}; + +/// The ingest pipeline shared across requests. The single-writer WAL +/// forces serialization; concurrent requests queue on the mutex. +pub type SharedPipeline = Arc>; + +/// OTLP/HTTP listener configuration. +#[derive(Debug, Clone)] +pub struct HttpConfig { + /// The accepted `POST` path (default `/v1/logs`; configurable per + /// §6.2 / RFC0003.14). + pub path: String, + /// Maximum request body size in bytes; a larger body is rejected with + /// 413 (RFC0003.11). + pub max_body_bytes: usize, +} + +impl Default for HttpConfig { + fn default() -> Self { + Self { + path: "/v1/logs".to_owned(), + max_body_bytes: 4 * 1024 * 1024, + } + } +} + +/// Handler state: the shared pipeline plus the decompressed-size cap +/// (`DefaultBodyLimit` only bounds the *compressed* body, so gzip is +/// bounded separately to defuse a decompression bomb). +#[derive(Clone)] +struct AppState { + pipeline: SharedPipeline, + max_decompressed_bytes: usize, +} + +/// Build the OTLP/HTTP router over `pipeline`. +pub fn router(pipeline: SharedPipeline, config: &HttpConfig) -> Router { + let state = AppState { + pipeline, + max_decompressed_bytes: config.max_body_bytes, + }; + Router::new() + .route(&config.path, post(handle_logs)) + .layer(DefaultBodyLimit::max(config.max_body_bytes)) + .with_state(state) +} + +/// The OTLP wire format selected by `Content-Type`. +#[derive(Clone, Copy)] +enum WireFormat { + Protobuf, + Json, +} + +/// The supported request `Content-Encoding`s. +enum Encoding { + Identity, + Gzip, +} + +async fn handle_logs(State(state): State, headers: HeaderMap, body: Bytes) -> Response { + let Some(format) = content_type(&headers) else { + return StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response(); + }; + let raw = match content_encoding(&headers) { + Some(Encoding::Identity) => body.to_vec(), + Some(Encoding::Gzip) => match gunzip(&body, state.max_decompressed_bytes) { + Ok(bytes) => bytes, + // Corrupt gzip is a malformed request (400); a body that + // decompresses past the limit is too large (413) — a + // decompression bomb, since DefaultBodyLimit only bounds the + // compressed bytes. + Err(GunzipError::Corrupt) => return StatusCode::BAD_REQUEST.into_response(), + Err(GunzipError::TooLarge) => return StatusCode::PAYLOAD_TOO_LARGE.into_response(), + }, + None => return StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response(), + }; + let decoded = match format { + WireFormat::Protobuf => decode_protobuf(&raw), + WireFormat::Json => decode_json(&raw), + }; + let Ok(request) = decoded else { + return StatusCode::BAD_REQUEST.into_response(); + }; + + // WAL-before-ack ingest. `ingest` does blocking I/O (WAL append + + // fsync), so run it on a blocking thread rather than stalling the + // async runtime. The lock spans only the synchronous call, and a + // poisoned lock is recovered (the handler promises not to panic). + let pipeline = state.pipeline; + let outcome = tokio::task::spawn_blocking(move || { + pipeline + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .ingest(request) + }) + .await; + + match outcome { + Ok(Ok(_)) => success_response(format), + // A Resource that doesn't resolve to a tenant is a client error + // (the whole batch is rejected, RFC0003.4). + Ok(Err(ReceiveError::TenantResolution(_))) => StatusCode::BAD_REQUEST.into_response(), + // A WAL failure (or a panic in the blocking task) is server-side; + // the batch was not acked (§3.4). + Ok(Err(_)) | Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), + } +} + +/// Map `Content-Type` to a wire format, ignoring any `; charset=…` +/// parameters. `None` = missing or unsupported (→ 415). +fn content_type(headers: &HeaderMap) -> Option { + let value = headers.get(header::CONTENT_TYPE)?.to_str().ok()?; + // Media types are case-insensitive; ignore any `; charset=…` params. + let media_type = value + .split(';') + .next() + .unwrap_or_default() + .trim() + .to_ascii_lowercase(); + match media_type.as_str() { + "application/x-protobuf" => Some(WireFormat::Protobuf), + "application/json" => Some(WireFormat::Json), + _ => None, + } +} + +/// Map `Content-Encoding` to a supported encoding. Absent or `identity` +/// is identity; `gzip` is supported; anything else (`zstd`, `br`, …) is +/// `None` (→ 415; zstd is deferred per §9). +fn content_encoding(headers: &HeaderMap) -> Option { + match headers.get(header::CONTENT_ENCODING) { + None => Some(Encoding::Identity), + // Content-Encoding tokens are case-insensitive. + Some(value) => match value.to_str().ok()?.trim().to_ascii_lowercase().as_str() { + "" | "identity" => Some(Encoding::Identity), + "gzip" => Some(Encoding::Gzip), + _ => None, + }, + } +} + +/// Why a gzip body was rejected. +enum GunzipError { + /// Not a valid gzip stream. + Corrupt, + /// The decompressed size exceeded `max` — a decompression bomb. + TooLarge, +} + +/// Decompress a gzip body, refusing to inflate past `max` bytes +/// (`DefaultBodyLimit` bounds only the compressed body, so an attacker +/// could otherwise expand a tiny upload into an unbounded allocation). +fn gunzip(bytes: &[u8], max: usize) -> Result, GunzipError> { + use std::io::Read; + // Read one byte past the cap so we can distinguish "exactly max" from + // "over the cap". + let cap = max.saturating_add(1) as u64; + let mut decoder = flate2::read::GzDecoder::new(bytes).take(cap); + let mut out = Vec::new(); + decoder + .read_to_end(&mut out) + .map_err(|_| GunzipError::Corrupt)?; + if out.len() > max { + return Err(GunzipError::TooLarge); + } + Ok(out) +} + +/// A 200 carrying an empty `ExportLogsServiceResponse` (`partial_success` +/// unset), encoded in the request's wire format. +fn success_response(format: WireFormat) -> Response { + let response = ExportLogsServiceResponse::default(); + match format { + WireFormat::Protobuf => ( + StatusCode::OK, + [(header::CONTENT_TYPE, "application/x-protobuf")], + response.encode_to_vec(), + ) + .into_response(), + WireFormat::Json => match serde_json::to_vec(&response) { + Ok(body) => ( + StatusCode::OK, + [(header::CONTENT_TYPE, "application/json")], + body, + ) + .into_response(), + // Encoding the (trivial) success response shouldn't fail; if + // it ever did, a 500 is honest — never a 200 with an empty + // body. + Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), + }, + } +} diff --git a/crates/ourios-ingester/src/receiver/pipeline.rs b/crates/ourios-ingester/src/receiver/pipeline.rs index bc87576ab..2329421eb 100644 --- a/crates/ourios-ingester/src/receiver/pipeline.rs +++ b/crates/ourios-ingester/src/receiver/pipeline.rs @@ -24,7 +24,10 @@ use crate::receiver::tenant::{TenantResolutionError, TenantRule, fan_out}; /// ordering can be exercised with a spy that records the `append`/`sync` /// calls (RFC0003.1/.12 per §8). The only production implementation is /// [`Wal`]. -pub trait Journal { +/// +/// `Send` so the pipeline can live behind a shared `Arc>` as +/// state in the async HTTP/gRPC listeners. +pub trait Journal: Send { /// Append one `OtlpBatch` frame carrying `payload` (not yet durable). /// /// # Errors diff --git a/crates/ourios-ingester/tests/http_transport_errors.rs b/crates/ourios-ingester/tests/http_transport_errors.rs new file mode 100644 index 000000000..a5a5bbfa8 --- /dev/null +++ b/crates/ourios-ingester/tests/http_transport_errors.rs @@ -0,0 +1,157 @@ +//! HTTP transport-error arms of RFC0003.11 — controlled status codes, +//! never a panic, and no `OtlpBatch` frame appended on a rejected +//! request. +//! +//! These cover the HTTP side of RFC0003.11; the gRPC arm (client +//! cancellation) and the full `rfc0003_11` acceptance flip land with the +//! gRPC-listener slice, so `rfc0003_11` stays `#[ignore]`'d until both +//! transports' arms exist. + +mod ingest_support; + +use axum::http::StatusCode; +use ingest_support::{capturing_pipeline, gzip, post_request, send}; +use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest; +use ourios_ingester::receiver::http::{HttpConfig, router}; +use prost::Message; + +const PROTOBUF: &str = "application/x-protobuf"; + +/// Send `request` and assert the controlled `expected` status and that +/// nothing was appended to the WAL. +async fn assert_rejected(request: axum::http::Request, expected: StatusCode) { + let (pipeline, captured) = capturing_pipeline(); + let (status, _) = send(router(pipeline, &HttpConfig::default()), request).await; + assert_eq!(status, expected); + assert!( + captured.lock().expect("captured").is_empty(), + "a rejected request appends no OtlpBatch frame", + ); +} + +#[tokio::test] +async fn malformed_protobuf_is_400() { + // 0xFF leads with field 31 / wire-type 7 (invalid) — prost rejects it. + assert_rejected( + post_request( + "/v1/logs", + Some(PROTOBUF), + None, + vec![0xff, 0xff, 0xff, 0xff], + ), + StatusCode::BAD_REQUEST, + ) + .await; +} + +#[tokio::test] +async fn unrecognised_content_type_is_415() { + assert_rejected( + post_request("/v1/logs", Some("text/plain"), None, b"hello".to_vec()), + StatusCode::UNSUPPORTED_MEDIA_TYPE, + ) + .await; +} + +#[tokio::test] +async fn missing_content_type_is_415() { + assert_rejected( + post_request("/v1/logs", None, None, b"hello".to_vec()), + StatusCode::UNSUPPORTED_MEDIA_TYPE, + ) + .await; +} + +#[tokio::test] +async fn corrupt_gzip_is_400() { + assert_rejected( + post_request( + "/v1/logs", + Some(PROTOBUF), + Some("gzip"), + b"not a gzip stream".to_vec(), + ), + StatusCode::BAD_REQUEST, + ) + .await; +} + +#[tokio::test] +async fn wrong_path_is_404() { + assert_rejected( + post_request("/not/the/path", Some(PROTOBUF), None, Vec::new()), + StatusCode::NOT_FOUND, + ) + .await; +} + +#[tokio::test] +async fn gzip_decompression_bomb_is_413() { + // Compresses small (under the body limit) but inflates well past it — + // DefaultBodyLimit bounds only the compressed bytes, so the + // decompressed cap must reject this. + let config = HttpConfig { + max_body_bytes: 4096, + ..HttpConfig::default() + }; + let bomb = gzip(&vec![0u8; 1_000_000]); + assert!( + bomb.len() < config.max_body_bytes, + "the compressed bomb is under the body limit" + ); + let (pipeline, captured) = capturing_pipeline(); + let (status, _) = send( + router(pipeline, &config), + post_request("/v1/logs", Some(PROTOBUF), Some("gzip"), bomb), + ) + .await; + assert_eq!( + status, + StatusCode::PAYLOAD_TOO_LARGE, + "a gzip body inflating past the cap → 413", + ); + assert!(captured.lock().expect("captured").is_empty()); +} + +#[tokio::test] +async fn content_type_and_encoding_are_case_insensitive() { + // Uppercased media type + encoding token must still be accepted (HTTP + // media types / Content-Encoding tokens are case-insensitive). An + // empty request takes the fast path to 200. + let payload = ExportLogsServiceRequest::default().encode_to_vec(); + let (pipeline, _) = capturing_pipeline(); + let (status, _) = send( + router(pipeline, &HttpConfig::default()), + post_request( + "/v1/logs", + Some("Application/X-Protobuf"), + Some("GZIP"), + gzip(&payload), + ), + ) + .await; + assert_eq!( + status, + StatusCode::OK, + "uppercased Content-Type and Content-Encoding are accepted", + ); +} + +#[tokio::test] +async fn oversize_body_is_413() { + let config = HttpConfig { + max_body_bytes: 16, + ..HttpConfig::default() + }; + let (pipeline, captured) = capturing_pipeline(); + let (status, _) = send( + router(pipeline, &config), + post_request("/v1/logs", Some(PROTOBUF), None, vec![0u8; 1024]), + ) + .await; + assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE); + assert!( + captured.lock().expect("captured").is_empty(), + "an oversize request appends no OtlpBatch frame", + ); +} diff --git a/crates/ourios-ingester/tests/ingest_support/mod.rs b/crates/ourios-ingester/tests/ingest_support/mod.rs index bc36cf688..7645bd9e0 100644 --- a/crates/ourios-ingester/tests/ingest_support/mod.rs +++ b/crates/ourios-ingester/tests/ingest_support/mod.rs @@ -148,3 +148,86 @@ pub fn resource_logs_without_scopes(service: &str) -> ResourceLogs { pub fn request(resource_logs: Vec) -> ExportLogsServiceRequest { ExportLogsServiceRequest { resource_logs } } + +// ----- HTTP-listener test support (RFC0003.11/.13/.14) ----- + +use axum::Router; +use axum::body::Body; +use axum::http::{Request, StatusCode, header}; +use ourios_ingester::receiver::http::SharedPipeline; +use tower::ServiceExt; + +/// The `OtlpBatch` payloads a [`capturing_pipeline`] appended, in order. +pub type Captured = Arc>>>; + +/// A `Journal` that records appended payloads (and fsyncs nothing), so a +/// test can recover what the pipeline ingested without a real WAL. +struct CapturingJournal { + captured: Captured, +} + +impl Journal for CapturingJournal { + fn append_batch(&mut self, payload: &[u8]) -> Result<(), ReceiveError> { + self.captured + .lock() + .expect("captured") + .push(payload.to_vec()); + Ok(()) + } + + fn sync(&mut self) -> Result<(), ReceiveError> { + Ok(()) + } +} + +/// A shared pipeline whose `Journal` captures appended payloads, plus the +/// capture handle. +pub fn capturing_pipeline() -> (SharedPipeline, Captured) { + let captured = Captured::default(); + let miner = MinerCluster::new(MinerConfig::default()); + let pipeline = IngestPipeline::new( + Box::new(CapturingJournal { + captured: captured.clone(), + }), + miner, + TenantRule::service_name(), + ); + (Arc::new(Mutex::new(pipeline)), captured) +} + +/// Build a `POST` request with optional `Content-Type`/`Content-Encoding`. +pub fn post_request( + path: &str, + content_type: Option<&str>, + content_encoding: Option<&str>, + body: Vec, +) -> Request { + let mut builder = Request::builder().method("POST").uri(path); + if let Some(value) = content_type { + builder = builder.header(header::CONTENT_TYPE, value); + } + if let Some(value) = content_encoding { + builder = builder.header(header::CONTENT_ENCODING, value); + } + builder.body(Body::from(body)).expect("build request") +} + +/// Drive `router` with `request` in-process (no socket) and return the +/// response status + body bytes. +pub async fn send(router: Router, request: Request) -> (StatusCode, Vec) { + let response = router.oneshot(request).await.expect("oneshot"); + let status = response.status(); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("read response body") + .to_vec(); + (status, bytes) +} + +/// gzip-compress `bytes` (for the `Content-Encoding: gzip` arm). +pub fn gzip(bytes: &[u8]) -> Vec { + use std::io::Write; + let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default()); + encoder.write_all(bytes).expect("gzip write"); + encoder.finish().expect("gzip finish") +} diff --git a/crates/ourios-ingester/tests/rfc0003_13_compression.rs b/crates/ourios-ingester/tests/rfc0003_13_compression.rs index fd3e87b6e..20f530917 100644 --- a/crates/ourios-ingester/tests/rfc0003_13_compression.rs +++ b/crates/ourios-ingester/tests/rfc0003_13_compression.rs @@ -1,17 +1,78 @@ //! RFC0003.13 — Compression over HTTP: identity and gzip MUST be supported. //! -//! Red gate (`specified → red`): `#[ignore]`'d until the receiver -//! lands. +//! The same payload sent with `Content-Encoding: identity` and `gzip` +//! decodes to the same request (proving gzip is decompressed, not just +//! accepted); an unsupported encoding is rejected with 415. + +mod ingest_support; + +use axum::http::StatusCode; +use ingest_support::{capturing_pipeline, gzip, post_request, request, resource_logs, send}; +use ourios_ingester::receiver::decode_protobuf; +use ourios_ingester::receiver::http::{HttpConfig, router}; +use prost::Message; + +const PROTOBUF: &str = "application/x-protobuf"; + +/// Recover the single `OtlpBatch` payload the pipeline captured. +fn only_captured(captured: &ingest_support::Captured) -> Vec { + let frames = captured.lock().expect("captured"); + assert_eq!(frames.len(), 1, "exactly one frame captured"); + frames[0].clone() +} /// Scenario RFC0003.13 — Compression over HTTP: identity and gzip MUST be supported. /// See `docs/rfcs/0003-otlp-receiver.md` §5. -#[ignore = "RFC 0003 red gate — implementation pending (RFC0003.13)"] -#[test] -fn rfc0003_13_identity_and_gzip_decode_equally_unsupported_is_415() { - unimplemented!( - "RFC0003.13 — the same payload sent with Content-Encoding: identity (or \ - absent) and with Content-Encoding: gzip yields equal OtlpLogRecord \ - sequences (both encodings are an OTLP MUST). An unsupported encoding \ - (zstd, br) is rejected with HTTP 415; zstd is deferred per §9." +#[tokio::test] +async fn rfc0003_13_identity_and_gzip_decode_equally_unsupported_is_415() { + let export = request(vec![resource_logs("checkout", &["hello"])]); + let payload = export.encode_to_vec(); + + // identity (explicit) + let (pipeline, captured_identity) = capturing_pipeline(); + let (status, _) = send( + router(pipeline, &HttpConfig::default()), + post_request( + "/v1/logs", + Some(PROTOBUF), + Some("identity"), + payload.clone(), + ), + ) + .await; + assert_eq!(status, StatusCode::OK, "identity is accepted"); + + // gzip + let (pipeline, captured_gzip) = capturing_pipeline(); + let (status, _) = send( + router(pipeline, &HttpConfig::default()), + post_request("/v1/logs", Some(PROTOBUF), Some("gzip"), gzip(&payload)), + ) + .await; + assert_eq!(status, StatusCode::OK, "gzip is accepted"); + + // The two paths decoded to the same request (gzip was decompressed), + // and that request is the original. + let from_identity = + decode_protobuf(&only_captured(&captured_identity)).expect("identity decodes"); + let from_gzip = decode_protobuf(&only_captured(&captured_gzip)).expect("gzip decodes"); + assert_eq!(from_identity, from_gzip, "identity and gzip decode equally"); + assert_eq!(from_identity, export, "and recover the original export"); + + // An unsupported encoding → 415, nothing appended. + let (pipeline, captured_zstd) = capturing_pipeline(); + let (status, _) = send( + router(pipeline, &HttpConfig::default()), + post_request("/v1/logs", Some(PROTOBUF), Some("zstd"), payload), + ) + .await; + assert_eq!( + status, + StatusCode::UNSUPPORTED_MEDIA_TYPE, + "an unsupported Content-Encoding → 415", + ); + assert!( + captured_zstd.lock().expect("captured").is_empty(), + "a rejected request appends nothing", ); } diff --git a/crates/ourios-ingester/tests/rfc0003_14_path_config.rs b/crates/ourios-ingester/tests/rfc0003_14_path_config.rs index e76a1bd0e..16ebe99d2 100644 --- a/crates/ourios-ingester/tests/rfc0003_14_path_config.rs +++ b/crates/ourios-ingester/tests/rfc0003_14_path_config.rs @@ -1,17 +1,74 @@ //! RFC0003.14 — Default `/v1/logs` path with configurable override. //! -//! Red gate (`specified → red`): `#[ignore]`'d until the receiver -//! lands. +//! A `POST` to the default `/v1/logs` is handled; any other path returns +//! 404; and an operator-configured override path replaces `/v1/logs` +//! (the default then 404s) without changing any other behaviour. + +mod ingest_support; + +use axum::http::StatusCode; +use ingest_support::{capturing_pipeline, post_request, send}; +use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest; +use ourios_ingester::receiver::http::{HttpConfig, router}; +use prost::Message; + +const PROTOBUF: &str = "application/x-protobuf"; + +/// An empty, valid OTLP/protobuf body — it takes the empty fast path to a +/// 200, so a handled route is observable without needing a tenant. +fn empty_body() -> Vec { + ExportLogsServiceRequest::default().encode_to_vec() +} /// Scenario RFC0003.14 — Default `/v1/logs` path with configurable override. /// See `docs/rfcs/0003-otlp-receiver.md` §5. -#[ignore = "RFC 0003 red gate — implementation pending (RFC0003.14)"] -#[test] -fn rfc0003_14_default_path_and_configurable_override() { - unimplemented!( - "RFC0003.14 — a POST to the default /v1/logs is handled via the §6.2 HTTP \ - path; a POST to any other path returns HTTP 404; and an operator-\ - configured override path (e.g. /otlp/v1/logs) replaces /v1/logs as the \ - accepted path without changing any other receiver behaviour." +#[tokio::test] +async fn rfc0003_14_default_path_and_configurable_override() { + // Default config: /v1/logs is handled. + let (pipeline, _) = capturing_pipeline(); + let (status, _) = send( + router(pipeline, &HttpConfig::default()), + post_request("/v1/logs", Some(PROTOBUF), None, empty_body()), + ) + .await; + assert_eq!(status, StatusCode::OK, "the default /v1/logs is handled"); + + // An unconfigured path → 404. + let (pipeline, _) = capturing_pipeline(); + let (status, _) = send( + router(pipeline, &HttpConfig::default()), + post_request("/not/the/path", Some(PROTOBUF), None, empty_body()), + ) + .await; + assert_eq!(status, StatusCode::NOT_FOUND, "an unconfigured path → 404"); + + // Operator override: the override path is handled... + let override_config = HttpConfig { + path: "/otlp/v1/logs".to_owned(), + ..HttpConfig::default() + }; + let (pipeline, _) = capturing_pipeline(); + let (status, _) = send( + router(pipeline, &override_config), + post_request("/otlp/v1/logs", Some(PROTOBUF), None, empty_body()), + ) + .await; + assert_eq!( + status, + StatusCode::OK, + "the operator override path is handled" + ); + + // ...and the default path no longer matches under the override. + let (pipeline, _) = capturing_pipeline(); + let (status, _) = send( + router(pipeline, &override_config), + post_request("/v1/logs", Some(PROTOBUF), None, empty_body()), + ) + .await; + assert_eq!( + status, + StatusCode::NOT_FOUND, + "the default path 404s once an override is configured", ); }