Healthchecks

[PSanetra]

We intent to run pyzeebe workers as Kubernetes pods and would like to provide readiness and liveness probes to Kubernetes. Is there some (recommended) way to implement these probes?

[felicijus]

There is gRPC Health Checking, only if Camunda supports it (really don't know).
This more or less is only helpfull to determine if the gRPC server is healthy. (This could play a role for the readiness and liveness probes)

The only thing I can think of is gRPC interceptors which then generate metrics and could be published via an api.
This might solve your problem, if you wrap your workers in FastAPI and create a /healthz endpoint to signal when metrics above some unwanted threshold, it could be as simple as monitoring gRPC error codes.

These projects could help.
https://github.com/lchenn/py-grpc-prometheus
https://github.com/taranisag/py-async-grpc-prometheus (pyzeebe uses async)

Do you plan on implementing the workers with a Framework like FastAPI ?

[PSanetra]

gRPC interceptors sounds like a nice idea to implement this!

Regarding the api: I think the pyzeebe library should be agnostic regarding the way how to publish this health status. I think some kind of publish-subscribe api (maybe simple callbacks) and a query api (maybe a simple property) should be enough. Then the application, which uses this library can wrap this api how it likes. E.g. creating a file in the file system, providing an http endpoint, open a TCP port or just log the health status on changes.

Edit: I think in my case my current favorite is open a simple TCP port and close it again when the application is not considered healthy anymore.

[dimastbk]

There is gRPC Health Checking, only if Camunda supports it (really don't know).

Yes, Zeebe Gateway supports GRPC Health.

import grpc.aio
from grpc_health.v1 import health_pb2, health_pb2_grpc


async def _health_check_call(stub: health_pb2_grpc.HealthStub) -> bool:
    request = health_pb2.HealthCheckRequest(service="gateway_protocol.Gateway")
    resp = await stub.Check(request)
    if resp.status != health_pb2.HealthCheckResponse.SERVING:
        raise Exception()


async def check_method(self, channel: grpc.aio.Channel) -> None:
    async with channel:
        health_stub = health_pb2_grpc.HealthStub(channel)
        await _health_check_call(health_stub)

But it's the health of Zeebe, not the application.

[felicijus]

But it's the health of Zeebe, not the application.

@dimastbk Yes, that is what I meant before.

@PSanetra Yes it's up to you and not pyzeebe, but maybe if you get it working you can share it here. And after that we could decide if it makes sense to adopt it into pyzeebe, or just provide a guide on how to do it.

[dimastbk]

I fixed worker.zeebe_adapter.connected in #527, now you can check this property

[dimastbk]

Thoughts:

1. we should set zeebe_adapter._current_connection_retries to 0 after successful connection, but, as I understand, it's possible only in interceptors. And grpc.aio doesn't support interception of existing channel (unlike sync), so we should intercept channel on creation
2. we should stop worker before closing the channel. Gracefully via events (like feat: stop worker if channel is closed #526) or via exception like ChannelClosedError in zeebe_adapter._close()?