gRPC Health Checking

1.78.0 · active · verified Sat Mar 28

grpcio-health-checking provides a standard Health Checking Service for gRPC servers, implementing the `grpc.health.v1` service API. It allows gRPC clients, load balancers, and orchestration systems to verify the availability and health status of gRPC services. The current version is 1.78.0, with new releases typically occurring every 2-3 months.

Warnings

breaking Mismatched Protobuf Gencode/Runtime versions can lead to `google.protobuf.runtime_version.VersionError`. This occurs when `grpcio` and `grpcio-health-checking` are compiled or installed with incompatible versions of the `protobuf` package.
Fix: Ensure that `grpcio` and `grpcio-health-checking` are installed in a clean environment and that their `protobuf` dependencies are compatible. Consider using a virtual environment and `pip install --upgrade grpcio grpcio-health-checking` to ensure consistent versions. Refer to Protobuf's cross-version runtime guarantee documentation.
gotcha It is crucial to notify the health check service when your gRPC server is shutting down gracefully. Failing to call `HealthServicer.enter_graceful_shutdown()` means connected clients will not be informed that the service is no longer serving, potentially leading to continued (and failed) health checks.
Fix: Implement a shutdown hook (e.g., a `KeyboardInterrupt` handler) to call `health_servicer.enter_graceful_shutdown()` before stopping your gRPC server.
gotcha When a gRPC client performs a health check, if the `Check` or `Watch` RPC call fails with an `UNIMPLEMENTED` status, the client should assume that health checking is not supported by the server for that service and should disable further health checks for it.
Fix: Clients should include logic to catch `grpc.StatusCode.UNIMPLEMENTED` errors and gracefully handle the absence of health checking functionality, typically by ceasing further health check attempts for that service.

Install

pip install grpcio-health-checking grpcio Install with pip

Imports

health
```
from grpc_health.v1 import health
```
health_pb2
```
from grpc_health.v1 import health_pb2
```

health_pb2_grpc

from grpc_health.v1 import health_pb2_grpc

HealthServicer

from grpc_health.v1.health import HealthServicer

Quickstart

This quickstart demonstrates both server-side implementation and client-side health checking. The server initializes a `HealthServicer`, adds it to the gRPC server, and sets the health status for the overall server (empty string) and a hypothetical 'MyService'. The client then uses a `HealthStub` to query these statuses. It includes handling for graceful shutdown notification and `UNIMPLEMENTED` status.

import grpc
import time
from concurrent import futures
from grpc_health.v1 import health_pb2, health_pb2_grpc
from grpc_health.v1.health import HealthServicer

# --- Server Side ---
class MyServiceServicer(object):
    def SayHello(self, request, context):
        return health_pb2.HealthCheckResponse()

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))

    # Add your own gRPC service here if you have one
    # my_service_pb2_grpc.add_MyServiceServicer_to_server(MyServiceServicer(), server)

    health_servicer = HealthServicer()
    health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server)

    # Set initial status for the overall server (empty string) and a specific service
    health_servicer.set('', health_pb2.HealthCheckResponse.ServingStatus.SERVING)
    health_servicer.set('MyService', health_pb2.HealthCheckResponse.ServingStatus.SERVING)

    server.add_insecure_port('[::]:50051')
    server.start()
    print('Server started on port 50051...')
    
    try:
        while True:
            time.sleep(86400) # One day in seconds
    except KeyboardInterrupt:
        health_servicer.enter_graceful_shutdown() # Important for client notification
        server.stop(0)

# --- Client Side ---
def check_health():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = health_pb2_grpc.HealthStub(channel)
        try:
            # Check overall server health
            response_overall = stub.Check(health_pb2.HealthCheckRequest(service=''))
            print(f"Overall Server Health: {health_pb2.HealthCheckResponse.ServingStatus.Name(response_overall.status)}")

            # Check specific service health
            response_my_service = stub.Check(health_pb2.HealthCheckRequest(service='MyService'))
            print(f"'MyService' Health: {health_pb2.HealthCheckResponse.ServingStatus.Name(response_my_service.status)}")

        except grpc.RpcError as e:
            if e.code() == grpc.StatusCode.UNIMPLEMENTED:
                print("Health checking is not implemented by the server.")
            else:
                print(f"Error checking health: {e.details}")

if __name__ == '__main__':
    import threading

    server_thread = threading.Thread(target=serve)
    server_thread.daemon = True # Allow main program to exit even if thread is running
    server_thread.start()

    time.sleep(1) # Give server time to start
    check_health()

view raw JSON →