enhancement(api)!: replace custom Health RPC with standard gRPC health service (#25139)

* enhancement(api)!: replace custom Health RPC with standard gRPC health service

Replace the custom `ObservabilityService/Health` RPC with the standard
`grpc.health.v1.Health` service on the observability API (port 8686).

This enables native Kubernetes gRPC health probes, `grpc-health-probe`,
and other standard tooling to work out of the box. The empty service
name (`""`) is used for whole-server health, matching what Kubernetes
probes and `grpc-health-probe` query by default.

Key changes:
- Remove Health RPC, HealthRequest, HealthResponse from the proto
- Add tonic_health standard health service to the gRPC server
- Switch vector-api-client to use tonic_health HealthClient; the
  client checks ServingStatus and returns NotServing error if not SERVING
- Flip health to NOT_SERVING in TopologyController::stop before draining
  the topology, so Kubernetes removes the pod from endpoints early
- Update docs, changelog, and add integration test

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix gate

* cleanup

* Update src/topology/controller.rs

Co-authored-by: Thomas <thomas.schneider@datadoghq.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: Thomas <thomas.schneider@datadoghq.com>

Author: 3 people

Cargo.lock

@@ -561,6 +561,7 @@ docker = ["dep:bollard", "dep:dirs-next"]
 api = [
   "dep:base64",
   "dep:tonic",
+  "dep:tonic-health",
   "dep:tonic-reflection",
   "dep:prost",
   "dep:prost-types",

Cargo.toml

@@ -27,8 +27,8 @@ vector top --url http://localhost:8686
   [grpcurl](https://github.com/fullstorydev/grpcurl):
 
 ```bash
-# Check health
-grpcurl -plaintext localhost:8686 vector.observability.v1.ObservabilityService/Health
+# Check health (standard gRPC health check, compatible with Kubernetes gRPC probes)
+grpcurl -plaintext localhost:8686 grpc.health.v1.Health/Check
 
 # List components
 grpcurl -plaintext localhost:8686 vector.observability.v1.ObservabilityService/GetComponents

changelog.d/graphql_to_grpc_api.breaking.md

@@ -9,6 +9,7 @@ license = "MPL-2.0"
 [dependencies]
 # gRPC
 tonic.workspace = true
+tonic-health.workspace = true
 prost.workspace = true
 prost-types.workspace = true
 

lib/vector-api-client/Cargo.toml

@@ -1,24 +1,28 @@
 use http::Uri;
 use tokio_stream::{Stream, StreamExt};
 use tonic::transport::{Channel, Endpoint};
+use tonic_health::pb::{
+    HealthCheckRequest, health_check_response::ServingStatus, health_client::HealthClient,
+};
 
 use crate::{
     error::{Error, Result},
     proto::{
         GetAllocationTracingStatusRequest, GetAllocationTracingStatusResponse,
-        GetComponentsRequest, GetComponentsResponse, GetMetaRequest, GetMetaResponse,
-        HealthRequest, HealthResponse, MetricName, StreamComponentAllocatedBytesRequest,
-        StreamComponentAllocatedBytesResponse, StreamComponentMetricsRequest,
-        StreamComponentMetricsResponse, StreamHeartbeatRequest, StreamHeartbeatResponse,
-        StreamOutputEventsRequest, StreamOutputEventsResponse, StreamUptimeRequest,
-        StreamUptimeResponse, observability_service_client::ObservabilityServiceClient,
+        GetComponentsRequest, GetComponentsResponse, GetMetaRequest, GetMetaResponse, MetricName,
+        StreamComponentAllocatedBytesRequest, StreamComponentAllocatedBytesResponse,
+        StreamComponentMetricsRequest, StreamComponentMetricsResponse, StreamHeartbeatRequest,
+        StreamHeartbeatResponse, StreamOutputEventsRequest, StreamOutputEventsResponse,
+        StreamUptimeRequest, StreamUptimeResponse,
+        observability_service_client::ObservabilityServiceClient,
     },
 };
 
 /// gRPC client for the Vector observability API
 #[derive(Debug, Clone)]
 pub struct Client {
     endpoint: Endpoint,
+    channel: Option<Channel>,
     client: Option<ObservabilityServiceClient<Channel>>,
 }
 
@@ -33,14 +37,16 @@ impl Client {
     pub fn new(uri: Uri) -> Self {
         Self {
             endpoint: Endpoint::from(uri),
+            channel: None,
             client: None,
         }
     }
 
     /// Connect to the gRPC server
     pub async fn connect(&mut self) -> Result<()> {
         let channel = self.endpoint.connect().await?;
-        self.client = Some(ObservabilityServiceClient::new(channel));
+        self.client = Some(ObservabilityServiceClient::new(channel.clone()));
+        self.channel = Some(channel);
         Ok(())
     }
 
@@ -49,13 +55,34 @@ impl Client {
         self.client.as_mut().ok_or(Error::NotConnected)
     }
 
+    /// Get the underlying channel
+    fn channel(&self) -> Result<&Channel> {
+        self.channel.as_ref().ok_or(Error::NotConnected)
+    }
+
     // ========== Unary RPCs ==========
 
-    /// Check if the API server is healthy
-    pub async fn health(&mut self) -> Result<HealthResponse> {
-        let client = self.ensure_connected()?;
-        let response = client.health(HealthRequest {}).await?;
-        Ok(response.into_inner())
+    /// Check if the API server is healthy using the standard gRPC health check
+    /// protocol (grpc.health.v1.Health/Check).
+    ///
+    /// Queries the empty service name (`""`), which represents whole-server
+    /// health. This is the default used by Kubernetes gRPC probes and
+    /// `grpc-health-probe`.
+    ///
+    /// Returns `Ok(())` if the server is `SERVING`, or an error otherwise.
+    pub async fn health(&mut self) -> Result<()> {
+        let channel = self.channel()?.clone();
+        let mut health_client = HealthClient::new(channel);
+        let response = health_client
+            .check(HealthCheckRequest {
+                service: String::new(),
+            })
+            .await?;
+        let status = response.into_inner().status;
+        if status != ServingStatus::Serving as i32 {
+            return Err(Error::NotServing { status });
+        }
+        Ok(())
     }
 
     /// Get metadata about the Vector instance

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

@@ -16,6 +16,9 @@ pub enum Error {
     #[snafu(display("Not connected to gRPC server"))]
     NotConnected,
 
+    #[snafu(display("Server is not serving (status: {})", status))]
+    NotServing { status: i32 },
+
     #[snafu(display("Stream error: {}", message))]
     Stream { message: String },
 }

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

@@ -11,9 +11,9 @@
 //! let mut client = Client::new("http://localhost:9999".parse().unwrap());
 //! client.connect().await?;
 //!
-//! // Check health
-//! let health = client.health().await?;
-//! println!("Healthy: {}", health.healthy);
+//! // Check health (standard gRPC health check)
+//! client.health().await?;
+//! println!("Server is healthy");
 //!
 //! // Get components
 //! let components = client.get_components(0).await?;

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

@@ -8,9 +8,6 @@ import "event.proto";
 service ObservabilityService {
   // ========== Simple Queries ==========
 
-  // Check if the API server is healthy and responding
-  rpc Health(HealthRequest) returns (HealthResponse);
-
   // Get metadata about the Vector instance (version, hostname)
   rpc GetMeta(GetMetaRequest) returns (GetMetaResponse);
 
@@ -42,13 +39,7 @@ service ObservabilityService {
   rpc StreamOutputEvents(StreamOutputEventsRequest) returns (stream StreamOutputEventsResponse);
 }
 
-// ========== Health & Meta Messages ==========
-
-message HealthRequest {}
-
-message HealthResponse {
-  bool healthy = 1;
-}
+// ========== Meta Messages ==========
 
 message GetMetaRequest {}
 

proto/vector/observability.proto

@@ -4,10 +4,7 @@ use std::pin::Pin;
 // (only used in synchronous map updates inside IntervalStream closures), so the
 // cheaper std mutex is correct here. tokio::sync::Mutex is only needed when the
 // critical section itself contains .await.
-use std::sync::{
-    Arc, Mutex,
-    atomic::{AtomicBool, Ordering},
-};
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use futures::{StreamExt as FuturesStreamExt, stream};
@@ -399,30 +396,18 @@ fn ports_to_proto_outputs(
 /// gRPC observability service implementation.
 pub struct ObservabilityService {
     watch_rx: WatchRx,
-    running: Arc<AtomicBool>,
 }
 
 impl ObservabilityService {
-    pub const fn new(watch_rx: WatchRx, running: Arc<AtomicBool>) -> Self {
-        Self { watch_rx, running }
+    pub const fn new(watch_rx: WatchRx) -> Self {
+        Self { watch_rx }
     }
 }
 
 #[tonic::async_trait]
 impl observability::Service for ObservabilityService {
     // ========== Simple Queries ==========
 
-    async fn health(
-        &self,
-        _request: Request<HealthRequest>,
-    ) -> Result<Response<HealthResponse>, Status> {
-        if self.running.load(Ordering::Relaxed) {
-            Ok(Response::new(HealthResponse { healthy: true }))
-        } else {
-            Err(Status::unavailable("Vector is shutting down"))
-        }
-    }
-
     async fn get_meta(
         &self,
         _request: Request<GetMetaRequest>,

src/api/grpc/service.rs

@@ -1,11 +1,8 @@
-use std::{
-    error::Error as StdError,
-    net::SocketAddr,
-    sync::{Arc, atomic::AtomicBool},
-};
+use std::{error::Error as StdError, net::SocketAddr};
 use tokio::sync::oneshot;
 use tokio_stream::wrappers::TcpListenerStream;
 use tonic::transport::Server as TonicServer;
+use tonic_health::server::{HealthReporter, health_reporter};
 use vector_lib::tap::topology::WatchRx;
 
 use super::grpc::ObservabilityService;
@@ -14,6 +11,7 @@ use crate::{config::Config, proto::observability::Server as ObservabilityServer}
 /// gRPC API server for Vector observability.
 pub struct GrpcServer {
     _shutdown: oneshot::Sender<()>,
+    health_reporter: HealthReporter,
     addr: SocketAddr,
 }
 
@@ -25,11 +23,7 @@ impl GrpcServer {
     /// is dropped.
     ///
     /// Returns an error if the server fails to bind to the configured address.
-    pub async fn start(
-        config: &Config,
-        watch_rx: WatchRx,
-        running: Arc<AtomicBool>,
-    ) -> crate::Result<Self> {
+    pub async fn start(config: &Config, watch_rx: WatchRx) -> crate::Result<Self> {
         let addr = config.api.address.ok_or_else(|| {
             crate::Error::from("API address not configured in config.api.address")
         })?;
@@ -46,7 +40,11 @@ impl GrpcServer {
 
         info!("GRPC API server bound to {}.", actual_addr);
 
-        let service = ObservabilityService::new(watch_rx, running);
+        let service = ObservabilityService::new(watch_rx);
+
+        // Create the standard gRPC health service (grpc.health.v1.Health).
+        // The empty service ("") is registered as SERVING by default.
+        let (health_reporter, health_service) = health_reporter();
 
         let (_shutdown, rx) = oneshot::channel();
 
@@ -59,10 +57,12 @@ impl GrpcServer {
                 .register_encoded_file_descriptor_set(
                     crate::proto::observability::FILE_DESCRIPTOR_SET,
                 )
+                .register_encoded_file_descriptor_set(tonic_health::pb::FILE_DESCRIPTOR_SET)
                 .build()
                 .expect("Failed to build reflection service");
 
             let result = TonicServer::builder()
+                .add_service(health_service)
                 .add_service(ObservabilityServer::new(service))
                 .add_service(reflection_service)
                 .serve_with_incoming_shutdown(incoming, async {
@@ -85,10 +85,22 @@ impl GrpcServer {
 
         Ok(Self {
             _shutdown,
+            health_reporter,
             addr: actual_addr,
         })
     }
 
+    /// Signal that the server is no longer serving.
+    ///
+    /// Call this **before** draining the topology so that Kubernetes gRPC
+    /// readiness probes fail early and the pod is removed from endpoints
+    /// before the process exits.
+    pub async fn set_not_serving(&mut self) {
+        self.health_reporter
+            .set_service_status("", tonic_health::ServingStatus::NotServing)
+            .await;
+    }
+
     /// Get the address the server is listening on
     pub const fn addr(&self) -> SocketAddr {
         self.addr