enhancement(vector source, vector sink): enable HTTP/2 adaptive flow-control windows

changelog.d/vector_h2_adaptive_window.enhancement.md

@@ -0,0 +1,3 @@
+The `vector` sink and gRPC-based sources now enable HTTP/2 adaptive flow-control windows. Previously both sides ran with the protocol-default 64 KiB stream/connection windows, which caps a single sink connection at roughly `64 KiB / request-round-trip`. With end-to-end acknowledgements enabled the round trip includes downstream processing on the receiving Vector instance, so a vector-to-vector link would plateau at a few MB/s per connection no matter how `batch` or `request.concurrency` were tuned. Letting hyper/tonic size the windows from the observed bandwidth-delay product removes this ceiling.
+
+authors: Evador

src/sinks/vector/config.rs

@@ -223,7 +223,16 @@ fn new_client(
 ) -> crate::Result<hyper::Client<ProxyConnector<HttpsConnector<HttpConnector>>, BoxBody>> {
     let proxy = build_proxy_connector(tls_settings.clone(), proxy_config)?;
 
-    Ok(hyper::Client::builder().http2_only(true).build(proxy))
+    Ok(hyper::Client::builder()
+        .http2_only(true)
+        // The default HTTP/2 flow-control windows (64 KiB) cap a single
+        // connection at roughly window_size / request_rtt. With end-to-end
+        // acknowledgements the peer only releases window credit once the
+        // request has been fully processed, so an untuned window throttles
+        // the sink to a few MB/s per connection regardless of batch size or
+        // concurrency. Let hyper size the windows from the observed BDP.
+        .http2_adaptive_window(true)
+        .build(proxy))
 }
 
 #[derive(Debug, Clone)]

src/sources/util/grpc/mod.rs

@@ -379,6 +379,10 @@ where
     info!(%address, "Building gRPC server.");
 
     Server::builder()
+        // See the note on `http2_adaptive_window` in the `vector` sink: the
+        // 64 KiB default flow-control windows throttle high-throughput
+        // senders, so size them dynamically from the observed BDP.
+        .http2_adaptive_window(Some(true))
         .layer(MaxConnectionAgeLayer::new())
         .layer(build_grpc_trace_layer(span.clone()))
         // This layer explicitly decompresses payloads, if compressed, and reports the number of message bytes we've
@@ -420,6 +424,10 @@ pub async fn run_grpc_server_with_routes(
     info!(%address, "Building gRPC server.");
 
     Server::builder()
+        // See the note on `http2_adaptive_window` in the `vector` sink: the
+        // 64 KiB default flow-control windows throttle high-throughput
+        // senders, so size them dynamically from the observed BDP.
+        .http2_adaptive_window(Some(true))
         .layer(MaxConnectionAgeLayer::new())
         .layer(build_grpc_trace_layer(span.clone()))
         .layer(DecompressionAndMetricsLayer)