Proper termination of a worker in Python in Kubernetes

Community Support
, ,
13

Thanks, I have rolled out this in prod, tested this multiple times (by doing rolling restart etc), seems to be working fine so far, will keep an eye on it.

For reference, here is the full code, including the main

worker = None


async def handler():
    print("signal received, shutting down worker")
    await worker.shutdown()
    print("finished shutting down worker")


async def main():
    logging.basicConfig(level=logging.INFO)

    with open(settings.TEMPORAL_CLOUD_TLS_CERT, "rb") as f:
        client_cert = f.read()
    with open(settings.TEMPORAL_CLOUD_TLS_KEY, "rb") as f:
        client_private_key = f.read()
    client = await Client.connect(
        settings.TEMPORAL_CLOUD_HOST_PORT,
        namespace=settings.TEMPORAL_CLOUD_NAMESPACE,
        tls=TLSConfig(
            client_cert=client_cert,
            client_private_key=client_private_key,
        ),
        data_converter=pydantic_data_converter,
    )

    with concurrent.futures.ThreadPoolExecutor(
        max_workers=settings.TEMPORAL_WORKER_MAX_THREADS
    ) as activity_executor:
        global worker
        worker = Worker(
            client,
            task_queue=settings.TEMPORAL_CLOUD_TASK_QUEUE,
            workflows=[MyWorkflow],
            activities=[
                my_activity,
            ],
            activity_executor=activity_executor,
            workflow_runner=new_sandbox_runner(),
            graceful_shutdown_timeout=timedelta(seconds=300),
        )
        await asyncio.gather(worker.run())


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    for signame in ("SIGINT", "SIGTERM"):
        loop.add_signal_handler(
            getattr(signal, signame), lambda: asyncio.create_task(handler())
        )
    loop.run_until_complete(main())