fix(vector source): implement standard gRPC health checking protocol (#24916)

* fix(vector source): implement standard gRPC health checking protocol

Adds support for the standard gRPC health checking protocol (grpc.health.v1.Health)
to the vector source alongside the existing custom health check endpoint. This enables
compatibility with standard health checking tools like grpc-health-probe, which is
commonly used in Kubernetes and other orchestration systems.

Changes:
- Added tonic-health dependency to support standard gRPC health protocol
- Implemented grpc.health.v1.Health service in vector source
- Maintained backward compatibility with existing custom health check
- Added tests for both standard and custom health check endpoints
- Both health services run on the same port using RoutesBuilder

The custom vector.Vector/HealthCheck endpoint continues to work for existing
Vector-to-Vector communication, while the new grpc.health.v1.Health/Check endpoint
provides compatibility with standard tooling.

Fixes: #23657

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* update licenses

* cargo fmt

* changelog edits

---------

Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>
Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>

Author: 3 people

Cargo.lock

@@ -204,6 +204,7 @@ tokio-tungstenite = { version = "0.20.1", default-features = false }
 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"] }
 tonic-build = { version = "0.11", default-features = false, features = ["transport", "prost"] }
+tonic-health = { version = "0.11", default-features = false }
 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"] }
@@ -436,6 +437,7 @@ tokio-tungstenite = { workspace = true, features = ["connect"], optional = true
 toml.workspace = true
 hickory-proto = { workspace = true, optional = true }
 tonic = { workspace = true, optional = true }
+tonic-health = { workspace = true, optional = true }
 thread_local = { version = "1.1.9", default-features = false, optional = true }
 typetag = { version = "0.2.20", default-features = false }
 url.workspace = true
@@ -735,7 +737,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", "protobuf-build"]
+sources-vector = ["dep:prost", "dep:tonic", "dep:tonic-health", "protobuf-build"]
 
 # Transforms
 transforms = ["transforms-logs", "transforms-metrics"]

Cargo.toml

@@ -796,6 +796,7 @@ toml_parser,https://github.com/toml-rs/toml,MIT OR Apache-2.0,The toml_parser Au
 toml_write,https://github.com/toml-rs/toml,MIT OR Apache-2.0,The toml_write Authors
 toml_writer,https://github.com/toml-rs/toml,MIT OR Apache-2.0,The toml_writer Authors
 tonic,https://github.com/hyperium/tonic,MIT,Lucio Franco <luciofranco14@gmail.com>
+tonic-health,https://github.com/hyperium/tonic,MIT,James Nugent <james@jen20.com>
 tower,https://github.com/tower-rs/tower,MIT,Tower Maintainers <team@tower-rs.com>
 tower-http,https://github.com/tower-rs/tower-http,MIT,Tower Maintainers <team@tower-rs.com>
 tower-layer,https://github.com/tower-rs/tower,MIT,Tower Maintainers <team@tower-rs.com>

LICENSE-3rdparty.csv

@@ -0,0 +1,7 @@
+`vector` source: Implement standard gRPC health checking protocol (`grpc.health.v1.Health`)
+alongside the existing custom health check endpoint. This enables compatibility with standard
+tools like `grpc-health-probe` for Kubernetes and other orchestration systems.
+
+Issue: https://github.com/vectordotdev/vector/issues/23657
+
+authors: jpds

changelog.d/23657-grpc-health-protocol.enhancement.md

@@ -159,6 +159,10 @@ async fn healthcheck(
         return Ok(());
     }
 
+    // Use the custom Vector health check
+    // Note: Both custom and standard health checks behave identically - they just
+    // return serving status without actual health validation. The Vector source
+    // implements both protocols now for compatibility.
     let request = service.client.health_check(proto::HealthCheckRequest {});
     match request.await {
         Ok(response) => match proto::ServingStatus::try_from(response.into_inner().status) {

src/sinks/vector/config.rs

@@ -3,7 +3,8 @@ use std::net::SocketAddr;
 
 use chrono::Utc;
 use futures::TryFutureExt;
-use tonic::{Request, Response, Status};
+use tonic::{Request, Response, Status, transport::server::RoutesBuilder};
+use tonic_health::server::health_reporter;
 use vector_lib::{
     EstimatedJsonEncodedSizeOf,
     codecs::NativeDeserializerConfig,
@@ -22,7 +23,7 @@ use crate::{
     internal_events::{EventsReceived, StreamClosedError},
     proto::vector as proto,
     serde::bool_or_struct,
-    sources::{Source, util::grpc::run_grpc_server},
+    sources::{Source, util::grpc::run_grpc_server_with_routes},
     tls::{MaybeTlsSettings, TlsEnableableConfig},
 };
 
@@ -176,7 +177,8 @@ impl SourceConfig for VectorConfig {
         let acknowledgements = cx.do_acknowledgements(self.acknowledgements);
         let log_namespace = cx.log_namespace(self.log_namespace);
 
-        let service = proto::Server::new(Service {
+        // Create the custom Vector service (existing)
+        let vector_service = proto::Server::new(Service {
             pipeline: cx.out,
             acknowledgements,
             log_namespace,
@@ -185,10 +187,25 @@ impl SourceConfig for VectorConfig {
         // Tonic added a default of 4MB in 0.9. This replaces the old behavior.
         .max_decoding_message_size(usize::MAX);
 
+        // Create the standard gRPC health service
+        let (mut health_reporter, health_service) = health_reporter();
+
+        // Register the Vector service as serving in the health reporter
+        health_reporter
+            .set_service_status("vector.Vector", tonic_health::ServingStatus::Serving)
+            .await;
+
+        // Combine both services using RoutesBuilder
+        let mut builder = RoutesBuilder::default();
+        builder
+            .add_service(health_service)
+            .add_service(vector_service);
+
         let source =
-            run_grpc_server(self.address, tls_settings, service, cx.shutdown).map_err(|error| {
-                error!(message = "Source future failed.", %error);
-            });
+            run_grpc_server_with_routes(self.address, tls_settings, builder.routes(), cx.shutdown)
+                .map_err(|error| {
+                    error!(message = "Source future failed.", %error);
+                });
 
         Ok(Box::pin(source))
     }
@@ -337,4 +354,91 @@ mod tests {
         );
         run_test(&config, addr).await;
     }
+
+    #[tokio::test]
+    async fn custom_health_check_works() {
+        use tonic::transport::Channel;
+
+        let (_guard, addr) = test_util::addr::next_addr();
+
+        let config = format!(r#"address = "{addr}""#);
+        let source: VectorConfig = toml::from_str(&config).unwrap();
+
+        let (tx, _rx) = SourceSender::new_test();
+        let server = source
+            .build(SourceContext::new_test(tx, None))
+            .await
+            .unwrap();
+        tokio::spawn(server);
+        test_util::wait_for_tcp(addr).await;
+
+        // Test the custom Vector health check endpoint
+        let endpoint = format!("http://{addr}");
+        let channel = Channel::from_shared(endpoint)
+            .unwrap()
+            .connect()
+            .await
+            .unwrap();
+
+        let mut client = proto::Client::new(channel);
+        let response = client
+            .health_check(proto::HealthCheckRequest {})
+            .await
+            .unwrap();
+
+        assert_eq!(
+            response.into_inner().status,
+            proto::ServingStatus::Serving as i32
+        );
+    }
+
+    #[tokio::test]
+    async fn standard_grpc_health_check_works() {
+        use tonic::transport::Channel;
+        use tonic_health::pb::{HealthCheckRequest, health_client::HealthClient};
+
+        let (_guard, addr) = test_util::addr::next_addr();
+
+        let config = format!(r#"address = "{addr}""#);
+        let source: VectorConfig = toml::from_str(&config).unwrap();
+
+        let (tx, _rx) = SourceSender::new_test();
+        let server = source
+            .build(SourceContext::new_test(tx, None))
+            .await
+            .unwrap();
+        tokio::spawn(server);
+        test_util::wait_for_tcp(addr).await;
+
+        // Test the standard gRPC health check protocol
+        let endpoint = format!("http://{addr}");
+        let channel = Channel::from_shared(endpoint)
+            .unwrap()
+            .connect()
+            .await
+            .unwrap();
+
+        let mut client = HealthClient::new(channel);
+
+        // Check aggregate server health (empty service string)
+        let response = client
+            .check(HealthCheckRequest {
+                service: String::new(),
+            })
+            .await
+            .unwrap();
+
+        use tonic_health::pb::health_check_response::ServingStatus;
+        assert_eq!(response.into_inner().status, ServingStatus::Serving as i32);
+
+        // Check the named Vector service health
+        let response = client
+            .check(HealthCheckRequest {
+                service: "vector.Vector".to_string(),
+            })
+            .await
+            .unwrap();
+
+        assert_eq!(response.into_inner().status, ServingStatus::Serving as i32);
+    }
 }