Limit decompressed body size

Bound the size of decompressed ingest request bodies to prevent
decompression-bomb OOMs. Only the on-the-wire size was capped before, so a
small, highly compressible payload (e.g. a few hundred KiB of zstd) could
inflate to many gigabytes and exhaust the node's memory.

- Add `ingest_api.max_decompressed_body_size` (default 256 MiB), enforced
  *as the body inflates* (`take(limit + 1)`) so an oversized payload is
  never fully decompressed -- bounding both memory and CPU.
- Apply it to the Elasticsearch `_bulk`, OTLP, and native ingest paths.
  OTLP REST is now size-bounded (it was previously unbounded).
- Reject offenders with HTTP 413, a per-endpoint
  `decompression_too_large_total` counter (labeled by endpoint), and a
  rate-limited warning.
- Rename `ingest_api.content_length_limit` to `ingest_api.max_body_size`
  (the wire / pre-decompression cap); the former name is still accepted as
  a config alias.

CHANGELOG.md

@@ -14,12 +14,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - (Jaeger) Query resource attributes when Jaeger request carries tags
 
 ### Changed
+- **Breaking**: ingest requests whose body decompresses to more than `ingest_api.max_decompressed_body_size` (default 256 MiB) are now rejected with HTTP 413 `Payload Too Large`. Previously only the *compressed* request size was bounded, so a small highly-compressible payload could inflate to many gigabytes. Affects the Elasticsearch `_bulk`, OpenTelemetry (OTLP), and native ingest endpoints. Split large batches into smaller requests, or raise the limit.
+- OTLP REST endpoints now enforce a request-size limit (previously unbounded), set to `ingest_api.max_decompressed_body_size` (default 256 MiB).
+- Renamed `ingest_api.content_length_limit` to `ingest_api.max_body_size`; the former name is still accepted as a config alias.
 
 ### Deprecated
 
 ### Removed
 
 ### Security
+- Bound the size of decompressed ingest request bodies (`ingest_api.max_decompressed_body_size`, default 256 MiB) to prevent decompression-bomb out-of-memory crashes. The limit is enforced as the body is inflated, so an oversized payload is never fully decompressed.
 
 --->
 

docs/configuration/node-config.md

@@ -266,7 +266,8 @@ indexer:
 | --- | --- | --- |
 | `max_queue_memory_usage` | Maximum size in bytes of the in-memory Ingest queue. | `2GiB` |
 | `max_queue_disk_usage` | Maximum disk-space in bytes taken by the Ingest queue. The minimum size is at least `256M` and be at least `max_queue_memory_usage`. | `4GiB` |
-| `content_length_limit` | Maximum payload size uncompressed. Increasing this is discouraged, use a [file source](../ingest-data/sqs-files.md) instead. | `10MiB` |
+| `max_body_size` | Maximum size of a request body as received on the wire (i.e. *before* decompression). Increasing this is discouraged, use a [file source](../ingest-data/sqs-files.md) instead. Formerly named `content_length_limit`, which is still accepted as an alias. | `10MiB` |
+| `max_decompressed_body_size` | Maximum size of a request body *after* decompression. A request is rejected with HTTP 413 `Payload Too Large` if its body decompresses beyond this; the limit is enforced as the body inflates, bounding memory and CPU and preventing decompression-bomb out-of-memory crashes. Must be at least `max_body_size`. | `256MiB` |
 | `grpc_compression_algorithm` | Compression algorithm (`gzip` or `zstd`) to use for gRPC traffic between nodes for the ingest service | `None` |
 | `decommission_timeout` | Maximum amount of time to wait for the ingester to finish decommissioning gracefully on shutdown before giving up. Can be overridden with the `QW_INGEST_DECOMMISSION_TIMEOUT` environment variable. | `300s` |
 
@@ -276,7 +277,8 @@ Example:
 ingest_api:
   max_queue_memory_usage: 2GiB
   max_queue_disk_usage: 4GiB
-  content_length_limit: 10MiB
+  max_body_size: 10MiB
+  max_decompressed_body_size: 256MiB
   grpc_compression_algorithm: zstd
   decommission_timeout: 300s
 ```

quickwit/quickwit-config/src/node_config/mod.rs

@@ -717,7 +717,14 @@ pub struct IngestApiConfig {
     /// Deprecated: ingest replication is no longer supported.
     #[serde(default, rename = "replication_factor", skip_serializing)]
     _replication_factor: Option<serde::de::IgnoredAny>,
-    pub content_length_limit: ByteSize,
+    /// Maximum size of a request body as received on the wire, i.e. *before* decompression.
+    /// Accepts the former name `content_length_limit` for backward compatibility.
+    #[serde(alias = "content_length_limit")]
+    pub max_body_size: ByteSize,
+    /// Maximum size of a request body *after* decompression. This bounds the memory (and CPU) a
+    /// single compressed request can consume while being inflated, preventing decompression-bomb
+    /// OOMs where a small compressed body expands to many gigabytes.
+    pub max_decompressed_body_size: ByteSize,
     /// (hidden) Targeted throughput for each shard
     pub shard_throughput_limit: ByteSize,
     /// (hidden) Maximum accumulated throughput capacity for underutilized
@@ -742,7 +749,8 @@ impl Default for IngestApiConfig {
             max_queue_memory_usage: ByteSize::gib(2),
             max_queue_disk_usage: ByteSize::gib(4),
             _replication_factor: None,
-            content_length_limit: ByteSize::mib(10),
+            max_body_size: ByteSize::mib(10),
+            max_decompressed_body_size: ByteSize::mib(256),
             shard_throughput_limit: DEFAULT_SHARD_THROUGHPUT_LIMIT,
             shard_burst_limit: DEFAULT_SHARD_BURST_LIMIT,
             shard_scale_up_factor: DEFAULT_SHARD_SCALE_UP_FACTOR,
@@ -809,12 +817,21 @@ impl IngestApiConfig {
             "shard_throughput_limit ({}) must be within 1mb and 20mb",
             self.shard_throughput_limit.display().si()
         );
+        // A decompressed body is always at least as large as its compressed form, so a decompressed
+        // cap below the wire body-size limit would reject bodies that the body-size filter already
+        // accepted.
+        ensure!(
+            self.max_decompressed_body_size >= self.max_body_size,
+            "max_decompressed_body_size ({}) must be at least max_body_size ({})",
+            self.max_decompressed_body_size.display().si(),
+            self.max_body_size.display().si()
+        );
         // The newline delimited format is persisted as something a bit larger
         // (lines prefixed with their length)
-        let estimated_persist_size = ByteSize::b(3 * self.content_length_limit.as_u64() / 2);
+        let estimated_persist_size = ByteSize::b(3 * self.max_body_size.as_u64() / 2);
         ensure!(
             self.shard_burst_limit >= estimated_persist_size,
-            "shard_burst_limit ({}) must be at least 1.5*content_length_limit ({})",
+            "shard_burst_limit ({}) must be at least 1.5*max_body_size ({})",
             self.shard_burst_limit,
             estimated_persist_size,
         );
@@ -1150,6 +1167,54 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_ingest_api_max_body_size_default_and_alias() {
+        // Default.
+        let default_config = IngestApiConfig::default();
+        assert_eq!(default_config.max_body_size, ByteSize::mib(10));
+
+        // The former name `content_length_limit` is still accepted as an alias.
+        let ingest_api_config: IngestApiConfig = serde_yaml::from_str(
+            r#"
+                content_length_limit: 20M
+            "#,
+        )
+        .unwrap();
+        assert_eq!(ingest_api_config.max_body_size, ByteSize::mb(20));
+        assert!(ingest_api_config.validate().is_ok());
+
+        // The new name works too.
+        let ingest_api_config: IngestApiConfig = serde_yaml::from_str(
+            r#"
+                max_body_size: 20M
+            "#,
+        )
+        .unwrap();
+        assert_eq!(ingest_api_config.max_body_size, ByteSize::mb(20));
+    }
+
+    #[test]
+    fn test_ingest_api_max_decompressed_body_size() {
+        let default_config = IngestApiConfig::default();
+        assert_eq!(
+            default_config.max_decompressed_body_size,
+            ByteSize::mib(256)
+        );
+
+        // A decompressed limit below the wire body-size limit is rejected.
+        let ingest_api_config: IngestApiConfig = serde_yaml::from_str(
+            r#"
+                max_body_size: 10M
+                max_decompressed_body_size: 1M
+            "#,
+        )
+        .unwrap();
+        assert_eq!(
+            ingest_api_config.validate().unwrap_err().to_string(),
+            "max_decompressed_body_size (1.0 MB) must be at least max_body_size (10.0 MB)"
+        );
+    }
+
     #[test]
     fn test_ingest_api_config_decommission_timeout() {
         let default_config = IngestApiConfig::default();

quickwit/quickwit-serve/src/decompression.rs

@@ -16,21 +16,58 @@ use std::io::Read;
 use std::sync::LazyLock;
 
 use bytes::Bytes;
+use bytesize::ByteSize;
 use flate2::read::{MultiGzDecoder, ZlibDecoder};
 use quickwit_common::metrics::IN_FLIGHT_REST_SERVER;
+use quickwit_common::rate_limited_warn;
 use quickwit_common::thread_pool::run_cpu_intensive;
-use quickwit_metrics::GaugeGuard;
+use quickwit_metrics::{GaugeGuard, counter};
 use thiserror::Error;
 use warp::Filter;
 use warp::reject::Reject;
 
 use crate::load_shield::{LoadShield, LoadShieldPermit};
+use crate::metrics::DECOMPRESSION_TOO_LARGE_TOTAL;
 
 fn get_ingest_load_shield() -> &'static LoadShield {
     static LOAD_SHIELD: LazyLock<LoadShield> = LazyLock::new(|| LoadShield::new("ingest"));
     &LOAD_SHIELD
 }
 
+/// Decodes `reader` into memory while enforcing `limit` on the decompressed size.
+///
+/// We read at most `limit + 1` bytes: that is enough to detect an overflow while guaranteeing we
+/// never inflate (and therefore never allocate or spend CPU on) more than the limit, even when the
+/// compressed input would expand to many gigabytes. This is what prevents decompression-bomb OOMs:
+/// the on-the-wire content length limit only caps the *compressed* size, which says nothing about
+/// how much a body expands.
+fn decode_to_limit<R: Read>(
+    reader: R,
+    limit: ByteSize,
+    encoding: &str,
+    compressed_len: usize,
+    endpoint: &'static str,
+) -> Result<Bytes, warp::Rejection> {
+    let mut decompressed = Vec::new();
+    reader
+        .take(limit.as_u64() + 1)
+        .read_to_end(&mut decompressed)
+        .map_err(|_| warp::reject::custom(CorruptedData))?;
+    if decompressed.len() as u64 > limit.as_u64() {
+        counter!(parent: DECOMPRESSION_TOO_LARGE_TOTAL, "endpoint" => endpoint).inc();
+        rate_limited_warn!(
+            limit_per_min = 6,
+            endpoint,
+            encoding,
+            compressed_len,
+            max_decompressed_body_size = limit.as_u64(),
+            "rejecting ingest request: decompressed body exceeds the maximum allowed size"
+        );
+        return Err(warp::reject::custom(DecompressedPayloadTooLarge(limit)));
+    }
+    Ok(Bytes::from(decompressed))
+}
+
 /// There are two ways to decompress the body:
 /// - Stream the body through an async decompressor
 /// - Fetch the body and then decompress the bytes
@@ -38,44 +75,54 @@ fn get_ingest_load_shield() -> &'static LoadShield {
 /// The first approach lowers the latency, while the second approach is more CPU efficient.
 /// Ingesting data is usually CPU bound and there is considerable latency until the data is
 /// searchable, so the second approach is more suitable for this use case.
-async fn decompress_body(encoding: Option<String>, body: Bytes) -> Result<Bytes, warp::Rejection> {
+async fn decompress_body(
+    encoding: Option<String>,
+    body: Bytes,
+    max_decompressed_body_size: ByteSize,
+    endpoint: &'static str,
+) -> Result<Bytes, warp::Rejection> {
     match encoding.as_deref() {
+        // Uncompressed bodies cannot amplify, and are already bounded by the (compressed)
+        // content length limit filter applied upstream of this one, so no extra check is needed.
         Some("identity") => Ok(body),
-        Some("gzip" | "x-gzip") => {
-            let decompressed = run_cpu_intensive(move || {
-                let mut decompressed = Vec::new();
-                let mut decoder = MultiGzDecoder::new(body.as_ref());
-                decoder
-                    .read_to_end(&mut decompressed)
-                    .map_err(|_| warp::reject::custom(CorruptedData))?;
-                Result::<_, warp::Rejection>::Ok(Bytes::from(decompressed))
-            })
-            .await
-            .map_err(|_| warp::reject::custom(CorruptedData))??;
-            Ok(decompressed)
-        }
-        Some("zstd") => {
-            let decompressed = run_cpu_intensive(move || {
-                zstd::decode_all(body.as_ref())
-                    .map(Bytes::from)
-                    .map_err(|_| warp::reject::custom(CorruptedData))
-            })
-            .await
-            .map_err(|_| warp::reject::custom(CorruptedData))??;
-            Ok(decompressed)
-        }
-        Some("deflate" | "x-deflate") => {
-            let decompressed = run_cpu_intensive(move || {
-                let mut decompressed = Vec::new();
-                ZlibDecoder::new(body.as_ref())
-                    .read_to_end(&mut decompressed)
-                    .map_err(|_| warp::reject::custom(CorruptedData))?;
-                Result::<_, warp::Rejection>::Ok(Bytes::from(decompressed))
-            })
-            .await
-            .map_err(|_| warp::reject::custom(CorruptedData))??;
-            Ok(decompressed)
-        }
+        Some("gzip" | "x-gzip") => run_cpu_intensive(move || {
+            let compressed_len = body.len();
+            decode_to_limit(
+                MultiGzDecoder::new(body.as_ref()),
+                max_decompressed_body_size,
+                "gzip",
+                compressed_len,
+                endpoint,
+            )
+        })
+        .await
+        .map_err(|_| warp::reject::custom(CorruptedData))?,
+        Some("zstd") => run_cpu_intensive(move || {
+            let compressed_len = body.len();
+            let decoder = zstd::Decoder::new(body.as_ref())
+                .map_err(|_| warp::reject::custom(CorruptedData))?;
+            decode_to_limit(
+                decoder,
+                max_decompressed_body_size,
+                "zstd",
+                compressed_len,
+                endpoint,
+            )
+        })
+        .await
+        .map_err(|_| warp::reject::custom(CorruptedData))?,
+        Some("deflate" | "x-deflate") => run_cpu_intensive(move || {
+            let compressed_len = body.len();
+            decode_to_limit(
+                ZlibDecoder::new(body.as_ref()),
+                max_decompressed_body_size,
+                "deflate",
+                compressed_len,
+                endpoint,
+            )
+        })
+        .await
+        .map_err(|_| warp::reject::custom(CorruptedData))?,
         Some(encoding) => Err(warp::reject::custom(UnsupportedEncoding(
             encoding.to_string(),
         ))),
@@ -95,13 +142,28 @@ pub(crate) struct UnsupportedEncoding(String);
 
 impl Reject for UnsupportedEncoding {}
 
-/// Custom filter for optional decompression
-pub(crate) fn get_body_bytes() -> impl Filter<Extract = (Body,), Error = warp::Rejection> + Clone {
+#[derive(Debug, Error)]
+#[error(
+    "decompressed request body exceeds the maximum allowed size of {}",
+    self.0.display().si()
+)]
+pub(crate) struct DecompressedPayloadTooLarge(ByteSize);
+
+impl Reject for DecompressedPayloadTooLarge {}
+
+/// Custom filter for optional decompression.
+///
+/// `max_decompressed_body_size` bounds the size of the body *after* decompression, protecting
+/// against decompression bombs (see [`decode_to_limit`]).
+pub(crate) fn get_body_bytes(
+    max_decompressed_body_size: ByteSize,
+    endpoint: &'static str,
+) -> impl Filter<Extract = (Body,), Error = warp::Rejection> + Clone {
     warp::header::optional("content-encoding")
         .and(warp::body::bytes())
-        .and_then(|encoding: Option<String>, body: Bytes| async move {
+        .and_then(move |encoding: Option<String>, body: Bytes| async move {
             let permit = get_ingest_load_shield().acquire_permit().await?;
-            decompress_body(encoding, body)
+            decompress_body(encoding, body, max_decompressed_body_size, endpoint)
                 .await
                 .map(|content| Body::new(content, permit))
         })
@@ -123,3 +185,141 @@ impl Body {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::io::Write;
+
+    use flate2::Compression;
+    use flate2::write::{GzEncoder, ZlibEncoder};
+
+    use super::*;
+
+    // A generous limit for tests that exercise the happy path.
+    const MAX_DECOMPRESSED: ByteSize = ByteSize::mib(256);
+
+    fn gzip(data: &[u8]) -> Vec<u8> {
+        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
+        encoder.write_all(data).unwrap();
+        encoder.finish().unwrap()
+    }
+
+    fn deflate(data: &[u8]) -> Vec<u8> {
+        let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
+        encoder.write_all(data).unwrap();
+        encoder.finish().unwrap()
+    }
+
+    fn zstd_compress(data: &[u8]) -> Vec<u8> {
+        zstd::encode_all(data, 0).unwrap()
+    }
+
+    #[tokio::test]
+    async fn test_decompress_body_roundtrip_under_limit() {
+        let payload = b"hello world, this is a small ingest payload".repeat(100);
+        for (encoding, compressed) in [
+            ("gzip", gzip(&payload)),
+            ("zstd", zstd_compress(&payload)),
+            ("deflate", deflate(&payload)),
+        ] {
+            let decompressed = decompress_body(
+                Some(encoding.to_string()),
+                Bytes::from(compressed),
+                MAX_DECOMPRESSED,
+                "test",
+            )
+            .await
+            .unwrap_or_else(|_| panic!("`{encoding}` body should decompress"));
+            assert_eq!(
+                decompressed.as_ref(),
+                payload.as_slice(),
+                "encoding={encoding}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn test_decompress_body_identity_and_none_passthrough() {
+        let body = Bytes::from_static(b"raw uncompressed body");
+        let out = decompress_body(
+            Some("identity".to_string()),
+            body.clone(),
+            MAX_DECOMPRESSED,
+            "test",
+        )
+        .await
+        .unwrap();
+        assert_eq!(out, body);
+        let out = decompress_body(None, body.clone(), MAX_DECOMPRESSED, "test")
+            .await
+            .unwrap();
+        assert_eq!(out, body);
+    }
+
+    #[tokio::test]
+    async fn test_decompress_body_unsupported_encoding() {
+        let error = decompress_body(
+            Some("br".to_string()),
+            Bytes::from_static(b"x"),
+            MAX_DECOMPRESSED,
+            "test",
+        )
+        .await
+        .unwrap_err();
+        assert!(error.find::<UnsupportedEncoding>().is_some());
+    }
+
+    #[tokio::test]
+    async fn test_decompress_body_rejects_over_limit() {
+        // ~16 KiB decompressed against a 1 KiB limit: every encoding must be rejected, exercising
+        // the full `decompress_body` path (and the `take()` bound) without inflating anything
+        // large.
+        let payload = vec![0u8; 16 * 1024];
+        for (encoding, compressed) in [
+            ("gzip", gzip(&payload)),
+            ("zstd", zstd_compress(&payload)),
+            ("deflate", deflate(&payload)),
+        ] {
+            let error = decompress_body(
+                Some(encoding.to_string()),
+                Bytes::from(compressed),
+                ByteSize::kib(1),
+                "test",
+            )
+            .await
+            .unwrap_err();
+            assert!(
+                error.find::<DecompressedPayloadTooLarge>().is_some(),
+                "encoding={encoding}"
+            );
+        }
+    }
+
+    #[test]
+    fn test_decode_to_limit_accepts_up_to_limit() {
+        let data = vec![0u8; 100];
+        let out = decode_to_limit(
+            data.as_slice(),
+            ByteSize::b(100),
+            "identity",
+            data.len(),
+            "test",
+        )
+        .unwrap();
+        assert_eq!(out.len(), 100);
+    }
+
+    #[test]
+    fn test_decode_to_limit_rejects_overflow() {
+        let data = vec![0u8; 1024];
+        let error = decode_to_limit(
+            data.as_slice(),
+            ByteSize::b(100),
+            "identity",
+            data.len(),
+            "test",
+        )
+        .unwrap_err();
+        assert!(error.find::<DecompressedPayloadTooLarge>().is_some());
+    }
+}

quickwit/quickwit-serve/src/elasticsearch_api/bulk.rs

@@ -37,11 +37,12 @@ use crate::{Body, with_arg};
 pub fn es_compat_bulk_handler(
     ingest_service: IngestServiceClient,
     ingest_router: IngestRouterServiceClient,
-    content_length_limit: ByteSize,
+    max_body_size: ByteSize,
+    max_decompressed_body_size: ByteSize,
     enable_ingest_v1: bool,
     enable_ingest_v2: bool,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
-    elastic_bulk_filter(content_length_limit)
+    elastic_bulk_filter(max_body_size, max_decompressed_body_size)
         .and(with_arg(ingest_service))
         .and(with_arg(ingest_router))
         .then(move |body, bulk_options, ingest_service, ingest_router| {
@@ -64,11 +65,12 @@ pub fn es_compat_bulk_handler(
 pub fn es_compat_index_bulk_handler(
     ingest_service: IngestServiceClient,
     ingest_router: IngestRouterServiceClient,
-    content_length_limit: ByteSize,
+    max_body_size: ByteSize,
+    max_decompressed_body_size: ByteSize,
     enable_ingest_v1: bool,
     enable_ingest_v2: bool,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
-    elastic_index_bulk_filter(content_length_limit)
+    elastic_index_bulk_filter(max_body_size, max_decompressed_body_size)
         .and(with_arg(ingest_service))
         .and(with_arg(ingest_router))
         .then(

quickwit/quickwit-serve/src/elasticsearch_api/bulk_v2.rs

@@ -429,9 +429,9 @@ mod tests {
 
     fn es_compat_bulk_handler_v2(
         ingest_router: IngestRouterServiceClient,
-        content_length_limit: ByteSize,
+        max_body_size: ByteSize,
     ) -> impl Filter<Extract = (impl Reply,), Error = Rejection> + Clone {
-        elastic_bulk_filter(content_length_limit)
+        elastic_bulk_filter(max_body_size, ByteSize::mib(256))
             .and(with_arg(ingest_router))
             .then(|body, bulk_options, ingest_router| {
                 elastic_bulk_ingest_v2(None, body, bulk_options, ingest_router)

quickwit/quickwit-serve/src/elasticsearch_api/filter.rs

@@ -66,14 +66,16 @@ pub(crate) fn elasticsearch_filter()
     )
 )]
 pub(crate) fn elastic_bulk_filter(
-    content_length_limit: ByteSize,
+    max_body_size: ByteSize,
+    max_decompressed_body_size: ByteSize,
 ) -> impl Filter<Extract = (Body, ElasticBulkOptions), Error = Rejection> + Clone {
     warp::path!("_elastic" / "_bulk")
         .and(warp::post().or(warp::put()).unify())
-        .and(warp::body::content_length_limit(
-            content_length_limit.as_u64(),
+        .and(warp::body::content_length_limit(max_body_size.as_u64()))
+        .and(get_body_bytes(
+            max_decompressed_body_size,
+            "elasticsearch_bulk",
         ))
-        .and(get_body_bytes())
         .and(warp::query())
 }
 
@@ -90,14 +92,16 @@ pub(crate) fn elastic_bulk_filter(
     )
 )]
 pub(crate) fn elastic_index_bulk_filter(
-    content_length_limit: ByteSize,
+    max_body_size: ByteSize,
+    max_decompressed_body_size: ByteSize,
 ) -> impl Filter<Extract = (String, Body, ElasticBulkOptions), Error = Rejection> + Clone {
     warp::path!("_elastic" / String / "_bulk")
         .and(warp::post().or(warp::put()).unify())
-        .and(warp::body::content_length_limit(
-            content_length_limit.as_u64(),
+        .and(warp::body::content_length_limit(max_body_size.as_u64()))
+        .and(get_body_bytes(
+            max_decompressed_body_size,
+            "elasticsearch_bulk",
         ))
-        .and(get_body_bytes())
         .and(warp::query::<ElasticBulkOptions>())
 }
 

quickwit/quickwit-serve/src/elasticsearch_api/mod.rs

@@ -68,14 +68,16 @@ pub fn elastic_api_handlers(
     enable_ingest_v1: bool,
     enable_ingest_v2: bool,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
-    let ingest_content_length_limit = node_config.ingest_api_config.content_length_limit;
+    let max_body_size = node_config.ingest_api_config.max_body_size;
+    let max_decompressed_body_size = node_config.ingest_api_config.max_decompressed_body_size;
     es_compat_cluster_info_handler(node_config.clone(), BuildInfo::get())
         .or(es_compat_nodes_handler(node_config.clone()))
         .or(es_compat_search_handler(search_service.clone()))
         .or(es_compat_bulk_handler(
             ingest_service.clone(),
             ingest_router.clone(),
-            ingest_content_length_limit,
+            max_body_size,
+            max_decompressed_body_size,
             enable_ingest_v1,
             enable_ingest_v2,
         ))
@@ -83,7 +85,8 @@ pub fn elastic_api_handlers(
         .or(es_compat_index_bulk_handler(
             ingest_service,
             ingest_router,
-            ingest_content_length_limit,
+            max_body_size,
+            max_decompressed_body_size,
             enable_ingest_v1,
             enable_ingest_v2,
         ))

quickwit/quickwit-serve/src/ingest_api/rest_handler.rs

@@ -95,9 +95,9 @@ fn ingest_filter(
     warp::path!(String / "ingest")
         .and(warp::post())
         .and(warp::body::content_length_limit(
-            config.content_length_limit.as_u64(),
+            config.max_body_size.as_u64(),
         ))
-        .and(get_body_bytes())
+        .and(get_body_bytes(config.max_decompressed_body_size, "ingest"))
         .and(warp::query::<IngestOptions>())
 }
 
@@ -406,8 +406,7 @@ pub(crate) mod tests {
 
     #[tokio::test]
     async fn test_ingest_api_return_413_if_above_content_limit() {
-        let config: IngestApiConfig =
-            serde_json::from_str(r#"{ "content_length_limit": "1" }"#).unwrap();
+        let config: IngestApiConfig = serde_json::from_str(r#"{ "max_body_size": "1" }"#).unwrap();
         let (universe, _temp_dir, ingest_service, _) =
             setup_ingest_v1_service(&["my-index"], &IngestApiConfig::default()).await;
         let ingest_router = IngestRouterServiceClient::mocked();

quickwit/quickwit-serve/src/metrics.rs

@@ -48,3 +48,10 @@ pub(crate) static CIRCUIT_BREAK_TOTAL: LazyCounter = lazy_counter!(
         description: "Circuit breaker counter",
         subsystem: "grpc",
 );
+
+pub(crate) static DECOMPRESSION_TOO_LARGE_TOTAL: LazyCounter = lazy_counter!(
+        name: "decompression_too_large_total",
+        description: "Number of ingest requests rejected because their decompressed body exceeded \
+                      the maximum allowed size, labeled by `endpoint`.",
+        subsystem: "rest",
+);

quickwit/quickwit-serve/src/otlp_api/rest_handler.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use bytesize::ByteSize;
 use quickwit_common::rate_limited_error;
 use quickwit_opentelemetry::otlp::{OtelSignal, OtlpGrpcLogsService, OtlpGrpcTracesService};
 use quickwit_proto::opentelemetry::proto::collector::logs::v1::logs_service_server::LogsService;
@@ -45,11 +46,18 @@ pub struct OtlpApi;
 pub(crate) fn otlp_ingest_api_handlers(
     otlp_logs_service: Option<OtlpGrpcLogsService>,
     otlp_traces_service: Option<OtlpGrpcTracesService>,
+    max_decompressed_body_size: ByteSize,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
-    otlp_default_logs_handler(otlp_logs_service.clone())
-        .or(otlp_default_traces_handler(otlp_traces_service.clone()).recover(recover_fn))
-        .or(otlp_logs_handler(otlp_logs_service).recover(recover_fn))
-        .or(otlp_ingest_traces_handler(otlp_traces_service).recover(recover_fn))
+    otlp_default_logs_handler(otlp_logs_service.clone(), max_decompressed_body_size)
+        .or(
+            otlp_default_traces_handler(otlp_traces_service.clone(), max_decompressed_body_size)
+                .recover(recover_fn),
+        )
+        .or(otlp_logs_handler(otlp_logs_service, max_decompressed_body_size).recover(recover_fn))
+        .or(
+            otlp_ingest_traces_handler(otlp_traces_service, max_decompressed_body_size)
+                .recover(recover_fn),
+        )
         .boxed()
 }
 
@@ -65,6 +73,7 @@ pub(crate) fn otlp_ingest_api_handlers(
 )]
 pub(crate) fn otlp_default_logs_handler(
     otlp_logs_service: Option<OtlpGrpcLogsService>,
+    max_decompressed_body_size: ByteSize,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
     require(otlp_logs_service)
         .and(warp::path!("otlp" / "v1" / "logs"))
@@ -76,7 +85,10 @@ pub(crate) fn otlp_default_logs_handler(
             OtelSignal::Logs.header_name(),
         ))
         .and(warp::post())
-        .and(get_body_bytes())
+        .and(warp::body::content_length_limit(
+            max_decompressed_body_size.as_u64(),
+        ))
+        .and(get_body_bytes(max_decompressed_body_size, "otlp_logs"))
         .then(
             |otlp_logs_service, index_id: Option<String>, body| async move {
                 let index_id =
@@ -100,6 +112,7 @@ pub(crate) fn otlp_default_logs_handler(
 )]
 pub(crate) fn otlp_logs_handler(
     otlp_log_service: Option<OtlpGrpcLogsService>,
+    max_decompressed_body_size: ByteSize,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
     require(otlp_log_service)
         .and(warp::path!(String / "otlp" / "v1" / "logs"))
@@ -108,7 +121,10 @@ pub(crate) fn otlp_logs_handler(
             "application/x-protobuf",
         ))
         .and(warp::post())
-        .and(get_body_bytes())
+        .and(warp::body::content_length_limit(
+            max_decompressed_body_size.as_u64(),
+        ))
+        .and(get_body_bytes(max_decompressed_body_size, "otlp_logs"))
         .then(otlp_ingest_logs)
         .and(with_arg(BodyFormat::default()))
         .map(into_rest_api_response)
@@ -127,6 +143,7 @@ pub(crate) fn otlp_logs_handler(
 )]
 pub(crate) fn otlp_default_traces_handler(
     otlp_traces_service: Option<OtlpGrpcTracesService>,
+    max_decompressed_body_size: ByteSize,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
     require(otlp_traces_service)
         .and(warp::path!("otlp" / "v1" / "traces"))
@@ -138,7 +155,10 @@ pub(crate) fn otlp_default_traces_handler(
             OtelSignal::Traces.header_name(),
         ))
         .and(warp::post())
-        .and(get_body_bytes())
+        .and(warp::body::content_length_limit(
+            max_decompressed_body_size.as_u64(),
+        ))
+        .and(get_body_bytes(max_decompressed_body_size, "otlp_traces"))
         .then(
             |otlp_traces_service, index_id: Option<String>, body| async move {
                 let index_id =
@@ -162,6 +182,7 @@ pub(crate) fn otlp_default_traces_handler(
 )]
 pub(crate) fn otlp_ingest_traces_handler(
     otlp_traces_service: Option<OtlpGrpcTracesService>,
+    max_decompressed_body_size: ByteSize,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
     require(otlp_traces_service)
         .and(warp::path!(String / "otlp" / "v1" / "traces"))
@@ -170,7 +191,10 @@ pub(crate) fn otlp_ingest_traces_handler(
             "application/x-protobuf",
         ))
         .and(warp::post())
-        .and(get_body_bytes())
+        .and(warp::body::content_length_limit(
+            max_decompressed_body_size.as_u64(),
+        ))
+        .and(get_body_bytes(max_decompressed_body_size, "otlp_traces"))
         .then(otlp_ingest_traces)
         .and(with_arg(BodyFormat::default()))
         .map(into_rest_api_response)
@@ -245,6 +269,7 @@ async fn otlp_ingest_traces(
 mod tests {
     use std::io::Write;
 
+    use bytesize::ByteSize;
     use flate2::Compression;
     use flate2::write::GzEncoder;
     use prost::Message;
@@ -353,7 +378,8 @@ mod tests {
         };
         let body = export_logs_request.encode_to_vec();
         let otlp_traces_api_handler =
-            otlp_ingest_api_handlers(Some(logs_service), Some(traces_service)).recover(recover_fn);
+            otlp_ingest_api_handlers(Some(logs_service), Some(traces_service), ByteSize::mib(256))
+                .recover(recover_fn);
         {
             // Test default otlp endpoint
             let resp = warp::test::request()
@@ -499,7 +525,8 @@ mod tests {
         };
         let body = export_trace_request.encode_to_vec();
         let otlp_traces_api_handler =
-            otlp_ingest_api_handlers(Some(logs_service), Some(traces_service)).recover(recover_fn);
+            otlp_ingest_api_handlers(Some(logs_service), Some(traces_service), ByteSize::mib(256))
+                .recover(recover_fn);
         {
             // Test default otlp endpoint
             let resp = warp::test::request()

quickwit/quickwit-serve/src/rest.rs

@@ -45,7 +45,7 @@ use warp::hyper::{Method, StatusCode, http};
 use warp::{Filter, Rejection, Reply, redirect};
 
 use crate::cluster_api::cluster_handler;
-use crate::decompression::{CorruptedData, UnsupportedEncoding};
+use crate::decompression::{CorruptedData, DecompressedPayloadTooLarge, UnsupportedEncoding};
 use crate::delete_task_api::delete_task_api_handlers;
 use crate::developer_api::developer_api_routes;
 use crate::elasticsearch_api::elastic_api_handlers;
@@ -554,6 +554,10 @@ fn api_v1_routes(
         .or(otlp_ingest_api_handlers(
             quickwit_services.otlp_logs_service_opt.clone(),
             quickwit_services.otlp_traces_service_opt.clone(),
+            quickwit_services
+                .node_config
+                .ingest_api_config
+                .max_decompressed_body_size,
         ))
         .boxed()
         .or(index_management_handlers(
@@ -659,6 +663,11 @@ fn get_status_with_error(rejection: Rejection) -> Result<RestApiError, Rejection
             status_code: StatusCode::BAD_REQUEST,
             message: error.to_string(),
         })
+    } else if let Some(error) = rejection.find::<DecompressedPayloadTooLarge>() {
+        Ok(RestApiError {
+            status_code: StatusCode::PAYLOAD_TOO_LARGE,
+            message: error.to_string(),
+        })
     } else if let Some(error) = rejection.find::<warp::reject::InvalidQuery>() {
         Ok(RestApiError {
             status_code: StatusCode::BAD_REQUEST,