feat(grpc): upgrade to Tonic 0.12 and native HTTP 1 types

Cargo.toml

@@ -168,7 +168,7 @@ arc-swap = { version = "1.8.2", default-features = false }
 async-compression = { version = "0.4.42", default-features = false, features = ["tokio", "gzip"] }
 async-stream = { version = "0.3.6", default-features = false }
 async-trait = { version = "0.1.89", default-features = false }
-axum = { version = "0.6.20", default-features = false }
+axum = { version = "0.7.9", default-features = false }
 base64 = { version = "0.23.0", default-features = false, features = ["std"] }
 bitflags = { version = "2.13.2", default-features = false }
 bytes = { version = "1.11.1", default-features = false, features = ["serde"] }
@@ -209,10 +209,10 @@ pastey = { version = "0.2", default-features = false }
 pin-project = { version = "1.1.11", default-features = false }
 proptest = { version = "1.11" }
 proptest-derive = { version = "0.8.0" }
-prost = { version = "0.12", default-features = false, features = ["std"] }
-prost-build = { version = "0.12", default-features = false }
+prost = { version = "0.13", default-features = false, features = ["std"] }
+prost-build = { version = "0.13", default-features = false }
 prost-reflect = { version = "0.14", features = ["serde"], default-features = false }
-prost-types = { version = "0.12", default-features = false }
+prost-types = { version = "0.13", default-features = false }
 quickcheck = { version = "1.1.0" }
 rand = { version = "0.10.1", default-features = false, features = ["thread_rng"] }
 rand_distr = { version = "0.6.0", default-features = false }
@@ -237,10 +237,10 @@ proc-macro2 = { version = "1", default-features = false }
 quote = { version = "1", default-features = false }
 syn = { version = "3", default-features = false, features = ["full", "visit", "extra-traits"] }
 toml = { version = "0.9.8", default-features = false, features = ["serde", "display", "parse"] }
-tonic = { version = "0.11", default-features = false, features = ["transport", "codegen", "prost", "tls", "tls-roots", "gzip", "zstd"] }
-tonic-build = { version = "0.11", default-features = false, features = ["transport", "prost"] }
-tonic-health = { version = "0.11", default-features = false }
-tonic-reflection = { version = "0.11", default-features = false, features = ["server"] }
+tonic = { version = "0.12.3", default-features = false, features = ["transport", "codegen", "prost", "tls", "tls-native-roots", "gzip", "zstd"] }
+tonic-build = { version = "0.12.3", default-features = false, features = ["transport", "prost"] }
+tonic-health = { version = "0.12.3", default-features = false }
+tonic-reflection = { version = "0.12.3", default-features = false, features = ["server"] }
 tracing = { version = "0.1.44", default-features = false }
 tracing-subscriber = { version = "0.3.22", default-features = false, features = ["fmt"] }
 url = { version = "2.5.4", default-features = false, features = ["serde"] }
@@ -409,7 +409,7 @@ parquet = { version = "59.2.0", default-features = false, features = [
   "lz4",
   "flate2-rust_backened",
 ], optional = true }
-axum = { version = "0.6.20", default-features = false }
+axum = { version = "0.7.9", default-features = false, features = ["http1", "tokio"] }
 base64 = { workspace = true, optional = true }
 bloomy = { version = "1.2.0", default-features = false, optional = true }
 bollard = { version = "0.20", default-features = false, features = ["pipe", "ssl", "chrono"], optional = true }
@@ -484,6 +484,7 @@ roaring = { version = "0.11.2", default-features = false, features = ["std"], op
 rumqttc = { version = "0.25.1", default-features = false, features = ["use-rustls-no-provider"], optional = true }
 # Keep MQTT on Ring alongside Vector's other Rustls users, including in minimal builds.
 rustls = { version = "0.23", default-features = false, features = ["logging", "ring", "std", "tls12"], optional = true }
+rustls-native-certs = { version = "0.8", default-features = false, optional = true }
 seahash = { version = "4.1.0", default-features = false }
 smallvec = { version = "1", default-features = false, features = ["union", "serde"] }
 snap = { version = "1.1.1", default-features = false }
@@ -783,7 +784,7 @@ sources-exec = []
 sources-file = ["vector-lib/file-source"]
 sources-file_descriptor = ["tokio-util/io"]
 sources-fluent = ["dep:base64", "sources-utils-net-tcp", "sources-utils-net-unix", "tokio-util/net", "dep:rmpv", "dep:rmp-serde", "dep:serde_bytes"]
-sources-gcp_pubsub = ["gcp", "dep:h2", "dep:prost", "dep:prost-types", "protobuf-build", "dep:tonic"]
+sources-gcp_pubsub = ["gcp", "dep:h2", "dep:prost", "dep:prost-types", "dep:rustls", "dep:rustls-native-certs", "protobuf-build", "dep:tonic"]
 sources-heroku_logs = ["sources-utils-http", "sources-utils-http-query", "sources-http_server"]
 sources-host_metrics = ["heim/cpu", "heim/host", "heim/memory", "heim/net"]
 sources-http_client = ["sources-utils-http-client"]
@@ -839,7 +840,7 @@ sources-websocket = ["dep:tokio-tungstenite"]
 sources-windows_event_log = ["dep:windows", "dep:quick-xml", "dep:governor"]
 sources-windows_event_log-integration-tests = ["sources-windows_event_log"]
 
-sources-vector = ["dep:prost", "dep:tonic", "dep:tonic-health", "protobuf-build"]
+sources-vector = ["dep:prost", "dep:tonic", "dep:tonic-health", "http-body-util/channel", "protobuf-build", "tower-http-1/trace"]
 
 # Transforms
 transforms = ["transforms-logs", "transforms-metrics"]
@@ -1112,7 +1113,7 @@ databend-integration-tests = ["sinks-databend"]
 datadog-agent-integration-tests = ["sources-datadog_agent"]
 datadog-logs-integration-tests = ["sinks-datadog_logs"]
 datadog-metrics-integration-tests = ["sinks-datadog_metrics", "dep:prost"]
-datadog-traces-integration-tests = ["sources-datadog_agent", "sinks-datadog_traces", "axum/tokio"]
+datadog-traces-integration-tests = ["sources-datadog_agent", "sinks-datadog_traces"]
 docker-logs-integration-tests = ["sources-docker_logs"]
 doris-integration-tests = ["sinks-doris"]
 es-integration-tests = ["sinks-elasticsearch", "aws-core"]

LICENSE-3rdparty.csv

@@ -821,7 +821,6 @@ tinystr,https://github.com/unicode-org/icu4x,Unicode-3.0,The ICU4X Project Devel
 tinyvec,https://github.com/Lokathor/tinyvec,Zlib OR Apache-2.0 OR MIT,Lokathor <zefria@gmail.com>
 tinyvec_macros,https://github.com/Soveu/tinyvec_macros,MIT OR Apache-2.0 OR Zlib,Soveu <marx.tomasz@gmail.com>
 tokio,https://github.com/tokio-rs/tokio,MIT,Tokio Contributors <team@tokio.rs>
-tokio-io-timeout,https://github.com/sfackler/tokio-io-timeout,MIT OR Apache-2.0,Steven Fackler <sfackler@gmail.com>
 tokio-macros,https://github.com/tokio-rs/tokio,MIT,Tokio Contributors <team@tokio.rs>
 tokio-native-tls,https://github.com/tokio-rs/tls,MIT,Tokio Contributors <team@tokio.rs>
 tokio-openssl,https://github.com/tokio-rs/tokio-openssl,MIT OR Apache-2.0,Alex Crichton <alex@alexcrichton.com>

build.rs

@@ -156,7 +156,7 @@ fn main() {
 
         tonic_build::configure()
             .protoc_arg("--experimental_allow_proto3_optional")
-            .compile_with_config(
+            .compile_protos_with_config(
                 prost_build,
                 &[
                     "lib/vector-core/proto/event.proto",

changelog.d/gcp_pubsub_native_reset_retry.fix.md

@@ -0,0 +1,5 @@
+The `gcp_pubsub` source now reconnects immediately, without logging an error, when the server
+resets a streaming pull. Since Vector 0.35.0 it logged an error and waited `retry_delay_secs`.
+It also trusts every certificate in `tls.ca_file`, not only the last one.
+
+authors: thomasqueirozb

changelog.d/grpc_sources_unknown_service.fix.md

@@ -0,0 +1,5 @@
+The `vector` and `opentelemetry` sources now answer gRPC requests for unregistered services with
+gRPC status `UNIMPLEMENTED` instead of HTTP 404. These requests no longer log an error or
+increment `component_errors_total`.
+
+authors: thomasqueirozb

changelog.d/vector_sink_proxy_tls.fix.md

@@ -0,0 +1,4 @@
+The `vector` sink now sends configured credentials to HTTP forward proxies and applies
+`tls.server_name` to TLS connections through CONNECT proxies.
+
+authors: thomasqueirozb

deny.toml

@@ -44,14 +44,7 @@ ignore = [
   { id = "RUSTSEC-2024-0388", reason = "derivative is unmaintained (https://github.com/vectordotdev/vector/issues/24940)" },
   { id = "RUSTSEC-2024-0449", reason = "unsafe-libyaml is unmaintained with no patched release; used by serde_yaml and VRL's serde_yaml_ng, whose backend migration is pending upstream (https://github.com/acatton/serde-yaml-ng/issues/5)" },
   { id = "RUSTSEC-2025-0134", reason = "rustls-pemfile is unmaintained - unpatched crate (https://github.com/bytebeamio/rumqtt/issues/1010) & tonic/reqwest upgrade (https://github.com/vectordotdev/vector/issues/19179)" },
-  { id = "RUSTSEC-2026-0049", reason = "rustls-webpki 0.102 is vulnerable - tonic upgrade (https://github.com/vectordotdev/vector/issues/19179)" },
-  { id = "RUSTSEC-2026-0098", reason = "rustls-webpki 0.102 is vulnerable - tonic upgrade (https://github.com/vectordotdev/vector/issues/19179)" },
-  { id = "RUSTSEC-2026-0099", reason = "rustls-webpki 0.102 is vulnerable - tonic upgrade (https://github.com/vectordotdev/vector/issues/19179)" },
-  { id = "RUSTSEC-2026-0104", reason = "rustls-webpki 0.102 CRL parsing panic - tonic upgrade (https://github.com/vectordotdev/vector/issues/19179); 0.103 already patched to 0.103.13" },
   { id = "RUSTSEC-2026-0173", reason = "proc-macro-error2 is unmaintained - transitive dep via mlua_derive v0.11.0; unblocked when mlua upgrades its derive macro" },
   { id = "RUSTSEC-2026-0235", reason = "rkyv 0.7 is vulnerable but unmaintained; upgrading vector-buffers to rkyv 0.8 requires a breaking serialization migration" },
-  { id = "RUSTSEC-2026-0258", reason = "h2 0.3 via aws-smithy-http-client (hyper 0.14) and tonic 0.11 - http 1.0 / AWS SDK upgrade (https://github.com/vectordotdev/vector/issues/19179)" },
-  { id = "RUSTSEC-2026-0346", reason = "rustls 0.22 is vulnerable - unpatched crate (https://github.com/bytebeamio/rumqtt) & tonic upgrade (https://github.com/vectordotdev/vector/issues/19179)" },
-  { id = "RUSTSEC-2026-0349", reason = "rustls 0.22 is vulnerable - unpatched crate (https://github.com/bytebeamio/rumqtt) & tonic upgrade (https://github.com/vectordotdev/vector/issues/19179)" },
-  { id = "RUSTSEC-2026-0382", reason = "rustls-webpki 0.102 has no patched release; retained by tonic 0.11 and rumqttc 0.24 pending upgrade (https://github.com/vectordotdev/vector/issues/19179); 0.103 already patched to 0.103.16" },
+  { id = "RUSTSEC-2026-0258", reason = "h2 0.3 via aws-smithy-http-client, hyper 0.14, and reqwest 0.11 - legacy HTTP / AWS SDK upgrade (https://github.com/vectordotdev/vector/issues/19179)" },
 ]

lib/codecs/src/encoding/format/protobuf.rs

@@ -2,7 +2,8 @@ use std::path::PathBuf;
 
 use crate::encoding::BuildError;
 use bytes::BytesMut;
-use prost_reflect::{MessageDescriptor, prost::Message as _};
+use prost::Message as _;
+use prost_reflect::MessageDescriptor;
 use tokio_util::codec::Encoder;
 use vector_config_macros::configurable_component;
 use vector_core::{

lib/opentelemetry-proto/build.rs

@@ -22,7 +22,7 @@ fn main() -> Result<()> {
         .build_client(true)
         .build_server(true)
         .file_descriptor_set_path(&descriptor_path)
-        .compile(&proto_paths, &[include_path])?;
+        .compile_protos(&proto_paths, &[include_path])?;
 
     // The vendored OTLP `.proto` files must remain byte-for-byte identical to upstream due to being
     // copied directly out of the upstream git repository. One upstream comment contains an indented
@@ -34,7 +34,7 @@ fn main() -> Result<()> {
     // - `#[doc(hidden)]` hides rendered documentation but does not suppress doctests extracted from
     //   child items in an included generated file.
     // - `[lib] doctest = false` is bypassed because CI explicitly requests doctests with `--doc`.
-    // - tonic-build 0.11's `disable_comments` only controls generated service/RPC comments, not the
+    // - tonic-build's `disable_comments` only controls generated service/RPC comments, not the
     //   protobuf field comments generated by prost.
     //
     // Normalize only the generated Rust copy, leaving the upstream source and its documentation

lib/vector-api-client/Cargo.toml

@@ -24,7 +24,10 @@ tokio-stream = { workspace = true, features = ["sync"] }
 snafu.workspace = true
 
 # Utils
-http = { version = "0.2", default-features = false }
+http = { version = "1", default-features = false, features = ["std"] }
+
+# Internal libs
+vector-common = { path = "../vector-common", default-features = false }
 
 [build-dependencies]
 tonic-build.workspace = true

lib/vector-api-client/build.rs

@@ -4,10 +4,12 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("cargo:rerun-if-changed=../../lib/vector-core/proto/event.proto");
 
     // First, generate event.proto types
-    tonic_build::configure().build_server(false).compile(
-        &["../../lib/vector-core/proto/event.proto"],
-        &["../../lib/vector-core/proto", "../../proto/third-party"],
-    )?;
+    tonic_build::configure()
+        .build_server(false)
+        .compile_protos(
+            &["../../lib/vector-core/proto/event.proto"],
+            &["../../lib/vector-core/proto", "../../proto/third-party"],
+        )?;
 
     // Then, generate observability.proto using extern_path to reference the event types
     let mut prost_config = prost_build::Config::new();
@@ -21,7 +23,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     tonic_build::configure()
         .build_server(false)
-        .compile_with_config(
+        .compile_protos_with_config(
             prost_config,
             &["../../proto/vector/observability.proto"],
             &[

lib/vector-api-client/src/client.rs

@@ -1,9 +1,10 @@
-use http::Uri;
+use http::{Uri, uri::Scheme};
 use tokio_stream::{Stream, StreamExt};
-use tonic::transport::{Channel, Endpoint};
+use tonic::transport::{Channel, ClientTlsConfig, Endpoint};
 use tonic_health::pb::{
     HealthCheckRequest, health_check_response::ServingStatus, health_client::HealthClient,
 };
+use vector_common::constants::GRPC_MAX_HEADER_LIST_SIZE;
 
 use crate::{
     error::{Error, Result},
@@ -36,19 +37,28 @@ impl Client {
     /// * `uri` - The gRPC server URI (e.g., `"http://localhost:9999".parse().unwrap()`)
     pub fn new(uri: Uri) -> Self {
         Self {
-            endpoint: Endpoint::from(uri),
+            endpoint: Endpoint::from(uri).http2_max_header_list_size(GRPC_MAX_HEADER_LIST_SIZE),
             channel: None,
             service: None,
         }
     }
 
     /// Connect to the gRPC server
     ///
+    /// `https` URIs use TLS with the platform's native root certificates.
+    ///
     /// # Errors
     ///
-    /// Returns an error if the connection to the server cannot be established.
+    /// Returns an error if TLS cannot be set up or the connection to the server cannot be
+    /// established.
     pub async fn connect(&mut self) -> Result<()> {
-        let channel = self.endpoint.connect().await?;
+        let endpoint = self.endpoint.clone();
+        let endpoint = if endpoint.uri().scheme() == Some(&Scheme::HTTPS) {
+            endpoint.tls_config(ClientTlsConfig::new().with_enabled_roots())?
+        } else {
+            endpoint
+        };
+        let channel = endpoint.connect().await?;
         self.service = Some(ObservabilityServiceClient::new(channel.clone()));
         self.channel = Some(channel);
         Ok(())
@@ -264,4 +274,23 @@ mod tests {
         assert!(result.is_err());
         assert!(matches!(result.unwrap_err(), Error::NotConnected));
     }
+
+    #[tokio::test]
+    async fn connect_uses_tls_for_https() {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let port = listener.local_addr().unwrap().port();
+        let first_byte = std::thread::spawn(move || {
+            let (mut stream, _) = listener.accept().unwrap();
+            let mut byte = [0u8; 1];
+            std::io::Read::read_exact(&mut stream, &mut byte).unwrap();
+            byte[0]
+        });
+
+        let mut client = Client::new(format!("https://127.0.0.1:{port}").parse().unwrap());
+        // The server closes the connection after the first byte, so the handshake fails.
+        assert!(client.connect().await.is_err());
+
+        // 0x16 is the TLS handshake record type that starts a ClientHello.
+        assert_eq!(first_byte.join().unwrap(), 0x16);
+    }
 }

lib/vector-common/src/constants.rs

@@ -17,3 +17,7 @@ pub const ZLIB_FRAME_OVERHEAD: usize = 6;
 /// Threshold below which zstd's `ZSTD_compressBound` adds extra margin (128 KiB).
 /// See: <https://github.com/facebook/zstd/blob/dev/lib/zstd.h> (`ZSTD_compressBound`)
 pub const ZSTD_SMALL_INPUT_THRESHOLD: usize = 128 << 10;
+
+/// HTTP/2 header list limit for gRPC clients and servers. It keeps the 16 MiB default of the
+/// legacy `h2` stack; Hyper 1 defaults to 16 KiB.
+pub const GRPC_MAX_HEADER_LIST_SIZE: u32 = 16 << 20;

lib/vector-top/Cargo.toml

@@ -20,7 +20,7 @@ unit-prefix = { version = "0.5.2", default-features = false, features = ["std"]
 num-format = { version = "0.4.4", default-features = false, features = ["with-num-bigint"] }
 ratatui = { version = "0.30.0", default-features = false, features = ["crossterm", "layout-cache"] }
 regex.workspace = true
-http = { version = "0.2", default-features = false }
+http = { version = "1", default-features = false, features = ["std"] }
 vector-common = { path = "../vector-common" }
 vector-api-client = { path = "../vector-api-client" }
 

lib/vector-vrl/metrics/Cargo.toml

@@ -16,5 +16,5 @@ vrl.workspace = true
 vector-core = { path = "../../vector-core", default-features = false }
 vector-common = { path = "../../vector-common", default-features = false }
 tokio.workspace = true
-tokio-stream.workspace = true
+tokio-stream = { workspace = true, features = ["time"] }
 vector-vrl-category.workspace = true

src/api/grpc/service.rs

@@ -532,7 +532,7 @@ impl observability::Service for ObservabilityService {
                 component_type: ComponentType::Source as i32,
                 on_type,
                 outputs: ports_to_proto_outputs(ports, &key_str, &sent_events_by_output),
-                metrics: component_metrics_map.get(&key_str).cloned(),
+                metrics: component_metrics_map.get(&key_str).copied(),
             });
         }
 
@@ -556,7 +556,7 @@ impl observability::Service for ObservabilityService {
                 component_type: component_type as i32,
                 on_type,
                 outputs,
-                metrics: component_metrics_map.get(&key_str).cloned(),
+                metrics: component_metrics_map.get(&key_str).copied(),
             });
         }