Compress requests and responses for ingest v2 endpoints (#5779)

* Compress requests and responses for ingest v2 endpoints

* Switch to ZSTD

* Test code-generated client/server compression

* Make gRPC ingest client compression configurable

* Document feature and upgrade

Author: guilload

docs/configuration/node-config.md

@@ -178,6 +178,7 @@ 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` |
+| `grpc_compression_algorithm` | Compression algorithm (`gzip` or `zstd`) to use for gRPC traffic between nodes for the ingest service | `None` |
 
 Example:
 
@@ -186,6 +187,7 @@ ingest_api:
   max_queue_memory_usage: 2GiB
   max_queue_disk_usage: 4GiB
   content_length_limit: 10MiB
+  grpc_compression_algorithm: zstd
 ```
 
 ## Searcher configuration

docs/operating/upgrades.md

@@ -23,4 +23,4 @@ No migration is done if `otel-traces-v0_7` already exists. If you want `service_
 
 Quickwit 0.9 introduces a new ingestion service to to power the ingest and bulk APIs (v2). The new ingest is enabled and used by default, even though the legacy one (v1) remains enabled to finish indexing residual data in the legacy write ahead logs. Note that `ingest_api.max_queue_disk_usage` is enforced on both ingest versions separately, which means that the cumulated disk usage might be up to twice this limit.
 
-The control plane should be upgraded first in order to enable the new ingest source (v2) on all existing indexes. Ingested data into previously existing indexes on upgraded indexer nodes will not be picked by the indexing pipelines until the control plane is upgraded. Because the indexing plan is computed differently in 0.9, all pipelines will be restarted when upgrading the control plane. If possible, we recommend avoiding rolling upgrades for indexers. Instead, scale down the number of indexers to zero first, then upgrade the control plane and finally scale the upgraded indexers back up.
+The control plane should be upgraded first in order to enable the new ingest source (v2) on all existing indexes. Ingested data into previously existing indexes on upgraded indexer nodes will not be picked by the indexing pipelines until the control plane is upgraded. Because the indexing plan is computed differently in 0.9, all pipelines will be restarted when upgrading the control plane. If possible, we recommend avoiding rolling upgrades for indexers. Instead, scale down the number of indexers to zero first, then upgrade the control plane and finally scale the upgraded indexers back up. Finally, if you intend to enable compression for the ingest service (`ingest_api.grpc_compression_algorithm`), you must do so in two steps: first, upgrade the indexer nodes with compression disabled, then update the node configuration to enable compression, and finally restart the indexer nodes.

quickwit/Cargo.lock

@@ -10,10 +10,10 @@ members = [
   "quickwit-common",
   "quickwit-config",
   "quickwit-control-plane",
-  "quickwit-index-management",
   "quickwit-datetime",
   "quickwit-directories",
   "quickwit-doc-mapper",
+  "quickwit-index-management",
   "quickwit-indexing",
   "quickwit-ingest",
   "quickwit-integration-tests",
@@ -245,9 +245,10 @@ tokio-stream = { version = "0.1", features = ["sync"] }
 tokio-util = { version = "0.7", features = ["full"] }
 toml = "0.7"
 tonic = { version = "0.13", features = [
-  "gzip",
   "_tls-any",
+  "gzip",
   "tls-native-roots",
+  "zstd",
 ] }
 tonic-build = "0.13"
 tonic-health = "0.13"

quickwit/Cargo.toml

@@ -43,7 +43,7 @@ pub(crate) async fn cluster_grpc_client(
 
     ClusterServiceClient::tower()
         .stack_layer(CLUSTER_GRPC_CLIENT_METRICS_LAYER.clone())
-        .build_from_channel(socket_addr, channel, MAX_MESSAGE_SIZE)
+        .build_from_channel(socket_addr, channel, MAX_MESSAGE_SIZE, None)
 }
 
 pub fn cluster_grpc_server(

quickwit/quickwit-cluster/src/grpc_service.rs

@@ -165,7 +165,9 @@ mod tests {
     use quickwit_common::tower::{BalanceChannel, Change, TimeoutLayer};
     use tokio::sync::mpsc::error::TrySendError;
     use tokio_stream::StreamExt;
+    use tonic::codec::CompressionEncoding;
     use tonic::transport::{Endpoint, Server};
+    use tonic::{Code, Status};
 
     use super::*;
     use crate::hello::MockHello;
@@ -263,9 +265,8 @@ mod tests {
 
     #[tokio::test]
     async fn test_hello_codegen_grpc() {
-        let grpc_server_adapter = HelloGrpcServerAdapter::new(HelloImpl::default());
-        let grpc_server: HelloGrpcServer<HelloGrpcServerAdapter> =
-            HelloGrpcServer::new(grpc_server_adapter);
+        let grpc_server =
+            HelloClient::new(HelloImpl::default()).as_grpc_service(MAX_GRPC_MESSAGE_SIZE);
         let addr: SocketAddr = "127.0.0.1:6666".parse().unwrap();
 
         tokio::spawn({
@@ -281,7 +282,7 @@ mod tests {
             "127.0.0.1:6666".parse().unwrap(),
             Endpoint::from_static("http://127.0.0.1:6666").connect_lazy(),
         );
-        let grpc_client = HelloClient::from_balance_channel(channel, MAX_GRPC_MESSAGE_SIZE);
+        let grpc_client = HelloClient::from_balance_channel(channel, MAX_GRPC_MESSAGE_SIZE, None);
 
         assert_eq!(
             grpc_client
@@ -340,7 +341,8 @@ mod tests {
 
         // The connectivity check fails if there is no client behind the channel.
         let (balance_channel, _): (BalanceChannel<SocketAddr>, _) = BalanceChannel::new();
-        let grpc_client = HelloClient::from_balance_channel(balance_channel, MAX_GRPC_MESSAGE_SIZE);
+        let grpc_client =
+            HelloClient::from_balance_channel(balance_channel, MAX_GRPC_MESSAGE_SIZE, None);
         assert_eq!(
             grpc_client
                 .check_connectivity()
@@ -351,6 +353,146 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_hello_codegen_grpc_with_compression() {
+        #[derive(Debug, Clone)]
+        struct CheckCompression<S> {
+            inner: S,
+        }
+
+        impl<S, ReqBody, ResBody> Service<http::Request<ReqBody>> for CheckCompression<S>
+        where
+            S: Service<http::Request<ReqBody>, Response = http::Response<ResBody>>
+                + Clone
+                + Send
+                + 'static,
+            S::Future: Send + 'static,
+            ReqBody: Send + 'static,
+        {
+            type Response = S::Response;
+            type Error = S::Error;
+            type Future = BoxFuture<Self::Response, Self::Error>;
+
+            fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+                self.inner.poll_ready(cx)
+            }
+
+            fn call(&mut self, request: http::Request<ReqBody>) -> Self::Future {
+                let Some(grpc_encoding) = request.headers().get("grpc-encoding") else {
+                    panic!("request should be compressed");
+                };
+                assert!(grpc_encoding.to_str().unwrap().contains("zstd"));
+
+                let Some(grpc_accept_encoding) = request.headers().get("grpc-accept-encoding")
+                else {
+                    panic!("client should accept compressed responses");
+                };
+                assert!(grpc_accept_encoding.to_str().unwrap().contains("zstd"));
+                let fut = self.inner.call(request);
+
+                Box::pin(async move {
+                    let response = fut.await?;
+
+                    let grpc_status_code = Status::from_header_map(response.headers())
+                        .map(|status| status.code())
+                        .unwrap_or(Code::Ok);
+
+                    if grpc_status_code == Code::Ok {
+                        let Some(grpc_encoding) = response.headers().get("grpc-encoding") else {
+                            panic!("response should be compressed");
+                        };
+                        assert!(grpc_encoding.to_str().unwrap().contains("zstd"));
+                    }
+                    Ok(response)
+                })
+            }
+        }
+
+        #[derive(Debug, Clone)]
+        struct CheckCompressionLayer;
+
+        impl<S> Layer<S> for CheckCompressionLayer {
+            type Service = CheckCompression<S>;
+
+            fn layer(&self, inner: S) -> Self::Service {
+                Self::Service { inner }
+            }
+        }
+
+        let grpc_server =
+            HelloClient::new(HelloImpl::default()).as_grpc_service(MAX_GRPC_MESSAGE_SIZE);
+        let addr: SocketAddr = "127.0.0.1:33333".parse().unwrap();
+
+        tokio::spawn({
+            async move {
+                Server::builder()
+                    .layer(CheckCompressionLayer)
+                    .add_service(grpc_server)
+                    .serve(addr)
+                    .await
+                    .unwrap();
+            }
+        });
+        let channel = BalanceChannel::from_channel(
+            "127.0.0.1:33333".parse().unwrap(),
+            Endpoint::from_static("http://127.0.0.1:33333").connect_lazy(),
+        );
+        let grpc_client = HelloClient::from_balance_channel(
+            channel,
+            MAX_GRPC_MESSAGE_SIZE,
+            Some(CompressionEncoding::Zstd),
+        );
+
+        assert_eq!(
+            grpc_client
+                .hello(HelloRequest {
+                    name: "gRPC client".to_string()
+                })
+                .await
+                .unwrap(),
+            HelloResponse {
+                message: "Hello, gRPC client!".to_string()
+            }
+        );
+
+        assert!(matches!(
+            grpc_client
+                .hello(HelloRequest {
+                    name: "".to_string()
+                })
+                .await
+                .unwrap_err(),
+            HelloError::InvalidArgument(_)
+        ));
+
+        let (ping_stream_tx, ping_stream) = ServiceStream::new_bounded(1);
+        let mut pong_stream = grpc_client.ping(ping_stream).await.unwrap();
+
+        ping_stream_tx
+            .try_send(PingRequest {
+                name: "gRPC client".to_string(),
+            })
+            .unwrap();
+        assert_eq!(
+            pong_stream.next().await.unwrap().unwrap().message,
+            "Pong, gRPC client!"
+        );
+
+        ping_stream_tx
+            .try_send(PingRequest {
+                name: "stop".to_string(),
+            })
+            .unwrap();
+        assert!(pong_stream.next().await.is_none());
+
+        let error = ping_stream_tx
+            .try_send(PingRequest {
+                name: "stop".to_string(),
+            })
+            .unwrap_err();
+        assert!(matches!(error, TrySendError::Closed(_)));
+    }
+
     #[tokio::test]
     async fn test_hello_codegen_actor() {
         #[derive(Debug)]
@@ -646,7 +788,7 @@ mod tests {
             "127.0.0.1:7777".parse().unwrap(),
             Endpoint::from_static("http://127.0.0.1:7777").connect_lazy(),
         );
-        HelloClient::from_balance_channel(balance_channed, MAX_GRPC_MESSAGE_SIZE);
+        HelloClient::from_balance_channel(balance_channed, MAX_GRPC_MESSAGE_SIZE, None);
     }
 
     #[tokio::test]
@@ -734,7 +876,7 @@ mod tests {
             .timeout(Duration::from_millis(100))
             .connect_lazy();
         let max_message_size = ByteSize::mib(1);
-        let grpc_client = HelloClient::from_channel(addr, channel, max_message_size);
+        let grpc_client = HelloClient::from_channel(addr, channel, max_message_size, None);
 
         let error = grpc_client
             .hello(HelloRequest {
@@ -813,7 +955,7 @@ mod tests {
             // this test hangs forever if we comment out the TimeoutLayer, which
             // shows that a request without explicit timeout might hang forever
             .stack_layer(TimeoutLayer::new(Duration::from_secs(3)))
-            .build_from_balance_channel(balance_channel, ByteSize::mib(1));
+            .build_from_balance_channel(balance_channel, ByteSize::mib(1), None);
 
         let response_fut = async move {
             grpc_client