diff --git a/README.md b/README.md index 2111fa005..e68758acb 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,7 @@ Some examples require extra dependencies. See each sample's directory for specif * [batch_sliding_window](batch_sliding_window) - Batch processing with a sliding window of child workflows. * [bedrock](bedrock) - Orchestrate a chatbot with Amazon Bedrock. * [cloud_export_to_parquet](cloud_export_to_parquet) - Set up schedule workflow to process exported files on an hourly basis +* [cloud_run_worker](cloud_run_worker) - Run a Temporal Worker inside a Google Cloud Run container with graceful SIGTERM shutdown and optional mTLS for Temporal Cloud. * [context_propagation](context_propagation) - Context propagation through workflows/activities via interceptor. * [custom_converter](custom_converter) - Use a custom payload converter to handle custom types. * [custom_decorator](custom_decorator) - Custom decorator to auto-heartbeat a long-running activity. diff --git a/cloud_run_worker/Dockerfile b/cloud_run_worker/Dockerfile new file mode 100644 index 000000000..afcc14473 --- /dev/null +++ b/cloud_run_worker/Dockerfile @@ -0,0 +1,18 @@ +FROM python:3.12-slim + +WORKDIR /app + +# Install uv for fast dependency installation +COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv + +# Install runtime dependency +RUN uv pip install --system "temporalio>=1.28.0,<2" + +# Copy the cloud_run_worker package +COPY __init__.py activities.py workflows.py worker.py starter.py ./cloud_run_worker/ + +# The worker polls Temporal, and it also serves a tiny health endpoint on $PORT +# (default 8080) so Cloud Run's startup probe passes. See worker.py. +EXPOSE 8080 + +CMD ["python", "-m", "cloud_run_worker.worker"] diff --git a/cloud_run_worker/README.md b/cloud_run_worker/README.md new file mode 100644 index 000000000..cb9465116 --- /dev/null +++ b/cloud_run_worker/README.md @@ -0,0 +1,152 @@ +# Cloud Run Worker + +This sample demonstrates how to run a Temporal Worker inside a +[Google Cloud Run](https://cloud.google.com/run) container. It shows: + +- Reading Temporal connection config from environment variables. +- Handling **SIGTERM gracefully** — Cloud Run sends SIGTERM and allows 10 seconds + for the container to exit; the worker drains in-progress tasks before stopping. +- Optional **mTLS** authentication for [Temporal Cloud](https://temporal.io/cloud) + via base64-encoded certificate/key env vars. + +The sample registers a simple greeting Workflow and Activity. The same pattern +applies to any Workflow/Activity definitions. + +## Prerequisites + +- A [Temporal Cloud](https://temporal.io/cloud) namespace **or** a self-hosted + Temporal server reachable from Cloud Run (e.g. via VPC connector or Cloud Run + direct VPC egress). +- [Docker](https://www.docker.com/) for local testing. +- [Google Cloud SDK (`gcloud`)](https://cloud.google.com/sdk) for deployment. +- Python 3.10+ + +## Files + +| File | Description | +|------|-------------| +| `worker.py` | Entry point — connects to Temporal, registers Workflows/Activities, and handles SIGTERM | +| `workflows.py` | Sample Workflow that executes a greeting Activity | +| `activities.py` | Sample Activity that returns a greeting string | +| `starter.py` | Helper script to start a Workflow execution from a local machine | +| `Dockerfile` | Container definition for Cloud Run | +| `deploy.sh` | Builds the image with Cloud Build and deploys to Cloud Run | +| `pyproject.toml` | Standalone project manifest for this sample | + +## Environment Variables + +| Variable | Required | Description | +|----------|----------|-------------| +| `TEMPORAL_ADDRESS` | Yes | `host:port` of the Temporal frontend (e.g. `..tmprl.cloud:7233`) | +| `TEMPORAL_NAMESPACE` | Yes | Temporal namespace (e.g. `.`) | +| `TEMPORAL_TASK_QUEUE` | No | Task queue name (default: `cloud-run-task-queue`) | +| `TEMPORAL_TLS_CERT` | For Temporal Cloud | Base64-encoded mTLS client certificate | +| `TEMPORAL_TLS_KEY` | For Temporal Cloud | Base64-encoded mTLS client private key | + +For Temporal Cloud, encode your credentials: + +```bash +export TEMPORAL_TLS_CERT=$(base64 -i client.pem) +export TEMPORAL_TLS_KEY=$(base64 -i client.key) +``` + +## Local Development + +### 1. Start a local Temporal dev server + +```bash +temporal server start-dev +``` + +### 2. Run the worker + +```bash +TEMPORAL_ADDRESS=localhost:7233 TEMPORAL_NAMESPACE=default uv run python worker.py +``` + +### 3. Start a workflow + +In a separate terminal, from inside this directory: + +```bash +TEMPORAL_ADDRESS=localhost:7233 TEMPORAL_NAMESPACE=default uv run python starter.py +``` + +## Docker + +### Build + +```bash +docker build -t cloud-run-worker . +``` + +### Run locally (against local dev server) + +```bash +docker run --rm \ + -e TEMPORAL_ADDRESS=host.docker.internal:7233 \ + -e TEMPORAL_NAMESPACE=default \ + cloud-run-worker +``` + +### Run locally (against Temporal Cloud) + +```bash +docker run --rm \ + -e TEMPORAL_ADDRESS=..tmprl.cloud:7233 \ + -e TEMPORAL_NAMESPACE=. \ + -e TEMPORAL_TLS_CERT="$(base64 -i client.pem)" \ + -e TEMPORAL_TLS_KEY="$(base64 -i client.key)" \ + cloud-run-worker +``` + +## Deploy to Cloud Run + +Use the provided `deploy.sh` script (requires `gcloud` and appropriate IAM +permissions to push to Container Registry and deploy Cloud Run services): + +```bash +export TEMPORAL_ADDRESS=..tmprl.cloud:7233 +export TEMPORAL_NAMESPACE=. + +./deploy.sh my-temporal-worker us-central1 my-gcp-project +``` + +The script: +1. Builds the container image with Cloud Build and pushes to Container Registry. +2. Deploys the Cloud Run service with the necessary env vars. +3. Reads `TEMPORAL_TLS_CERT` and `TEMPORAL_TLS_KEY` from Secret Manager secrets + named `-tls-cert` and `-tls-key`. + +Or run `gcloud run deploy` directly: + +```bash +gcloud run deploy my-temporal-worker \ + --image gcr.io/my-gcp-project/my-temporal-worker:latest \ + --region us-central1 \ + --platform managed \ + --no-allow-unauthenticated \ + --set-env-vars "TEMPORAL_ADDRESS=..tmprl.cloud:7233,TEMPORAL_NAMESPACE=." \ + --set-secrets "TEMPORAL_TLS_CERT=my-temporal-worker-tls-cert:latest,TEMPORAL_TLS_KEY=my-temporal-worker-tls-key:latest" \ + --min-instances 1 \ + --no-cpu-throttling +``` + +Setting `--min-instances 1` ensures the worker is always polling; without it +Cloud Run may scale to zero and leave no worker available for task execution. + +### Why a health server? + +Cloud Run services must listen on the port named by `$PORT` (default 8080) or the +deploy's startup probe fails and the service never goes healthy. A Temporal worker +only polls Temporal, so `worker.py` runs a tiny HTTP health endpoint in a background +thread purely to satisfy that contract. The `--no-cpu-throttling` flag keeps CPU +allocated between requests so the worker keeps polling, since Cloud Run otherwise +throttles CPU to near zero when no HTTP request is in flight. + +## Further Reading + +- [Temporal Python SDK documentation](https://python.temporal.io) +- [Temporal Cloud — production worker deployment](https://docs.temporal.io/production-deployment) +- [Cloud Run — container contract](https://cloud.google.com/run/docs/container-contract) +- [Cloud Run — handling signals (SIGTERM)](https://cloud.google.com/run/docs/reference/container-contract#lifecycle) diff --git a/cloud_run_worker/__init__.py b/cloud_run_worker/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/cloud_run_worker/activities.py b/cloud_run_worker/activities.py new file mode 100644 index 000000000..a6a9c8e22 --- /dev/null +++ b/cloud_run_worker/activities.py @@ -0,0 +1,7 @@ +from temporalio import activity + + +@activity.defn +async def hello_activity(name: str) -> str: + activity.logger.info("HelloActivity started with name: %s", name) + return f"Hello, {name}!" diff --git a/cloud_run_worker/deploy.sh b/cloud_run_worker/deploy.sh new file mode 100755 index 000000000..ea337bcb5 --- /dev/null +++ b/cloud_run_worker/deploy.sh @@ -0,0 +1,42 @@ +#!/usr/bin/env bash +# Deploy the Cloud Run worker to Google Cloud Run. +# +# Usage: +# ./deploy.sh +# +# Example: +# ./deploy.sh my-temporal-worker us-central1 my-gcp-project +# +# Required env vars (pass via --set-env-vars or Secret Manager): +# TEMPORAL_ADDRESS - e.g. ..tmprl.cloud:7233 +# TEMPORAL_NAMESPACE - e.g. . +# TEMPORAL_TLS_CERT - base64-encoded mTLS client certificate (Temporal Cloud) +# TEMPORAL_TLS_KEY - base64-encoded mTLS client private key (Temporal Cloud) + +set -euo pipefail + +SERVICE="${1:?Usage: $0 }" +REGION="${2:?Usage: $0 }" +PROJECT="${3:?Usage: $0 }" + +IMAGE="gcr.io/${PROJECT}/${SERVICE}:latest" + +echo "Building and pushing image: ${IMAGE}" +gcloud builds submit --tag "${IMAGE}" --project "${PROJECT}" . + +echo "Deploying Cloud Run service: ${SERVICE}" +gcloud run deploy "${SERVICE}" \ + --image "${IMAGE}" \ + --region "${REGION}" \ + --project "${PROJECT}" \ + --platform managed \ + --no-allow-unauthenticated \ + --set-env-vars "TEMPORAL_ADDRESS=${TEMPORAL_ADDRESS},TEMPORAL_NAMESPACE=${TEMPORAL_NAMESPACE}" \ + --set-secrets "TEMPORAL_TLS_CERT=${SERVICE}-tls-cert:latest,TEMPORAL_TLS_KEY=${SERVICE}-tls-key:latest" \ + --min-instances 1 \ + --max-instances 10 \ + --no-cpu-throttling + +echo "Done. Service URL:" +gcloud run services describe "${SERVICE}" --region "${REGION}" --project "${PROJECT}" \ + --format "value(status.url)" diff --git a/cloud_run_worker/pyproject.toml b/cloud_run_worker/pyproject.toml new file mode 100644 index 000000000..2c5966cca --- /dev/null +++ b/cloud_run_worker/pyproject.toml @@ -0,0 +1,41 @@ +[project] +name = "temporalio-samples-cloud-run-worker" +version = "0.1a1" +description = "Temporal.io Python SDK Cloud Run worker sample" +authors = [{ name = "Temporal Technologies Inc", email = "sdk@temporal.io" }] +requires-python = ">=3.10" +readme = "README.md" +license = "MIT" +dependencies = ["temporalio>=1.28.0,<2"] + +[dependency-groups] +dev = [ + "ruff>=0.5.0,<0.6", + "mypy>=1.4.1,<2", + "poethepoet>=0.36.0", +] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.metadata] +allow-direct-references = true + +[tool.hatch.build.targets.wheel] +packages = ["."] + +[tool.poe.tasks] +format = [ + { cmd = "uv run ruff check --select I --fix" }, + { cmd = "uv run ruff format" }, +] +lint = [ + { cmd = "uv run ruff check --select I" }, + { cmd = "uv run ruff format --check" }, + { ref = "lint-types" }, +] +lint-types = "uv run --all-groups mypy --check-untyped-defs --namespace-packages ." + +[tool.ruff] +target-version = "py310" diff --git a/cloud_run_worker/starter.py b/cloud_run_worker/starter.py new file mode 100644 index 000000000..98afd05a5 --- /dev/null +++ b/cloud_run_worker/starter.py @@ -0,0 +1,53 @@ +""" +Helper script to start a SampleWorkflow execution against the Cloud Run worker. + +Run from the repo root: + + TEMPORAL_ADDRESS=
TEMPORAL_NAMESPACE= \\ + uv run python -m cloud_run_worker.starter +""" + +import asyncio +import os + +from temporalio.client import Client, TLSConfig + +from cloud_run_worker.workflows import TASK_QUEUE, SampleWorkflow + + +def _tls_config() -> TLSConfig | None: + cert_b64 = os.environ.get("TEMPORAL_TLS_CERT") + key_b64 = os.environ.get("TEMPORAL_TLS_KEY") + if cert_b64 and key_b64: + import base64 + + return TLSConfig( + client_cert=base64.b64decode(cert_b64), + client_private_key=base64.b64decode(key_b64), + ) + return None + + +async def main() -> None: + address = os.environ.get("TEMPORAL_ADDRESS", "localhost:7233") + namespace = os.environ.get("TEMPORAL_NAMESPACE", "default") + task_queue = os.environ.get("TEMPORAL_TASK_QUEUE", TASK_QUEUE) + + client = await Client.connect( + address, + namespace=namespace, + tls=_tls_config() or False, + ) + print(f"Connected to Temporal Service at {address}") + + result = await client.execute_workflow( + SampleWorkflow.run, + "Cloud Run Worker!", + id="cloud-run-workflow-id-1", + task_queue=task_queue, + ) + print(f"Workflow result: {result}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/cloud_run_worker/worker.py b/cloud_run_worker/worker.py new file mode 100644 index 000000000..979850449 --- /dev/null +++ b/cloud_run_worker/worker.py @@ -0,0 +1,114 @@ +""" +Temporal Worker for Google Cloud Run. + +Reads connection config from environment variables, registers a sample Workflow +and Activity, and handles SIGTERM gracefully so Cloud Run can shut the container +down within its 10-second termination window. + +Run from the repo root: + + TEMPORAL_ADDRESS=localhost:7233 TEMPORAL_NAMESPACE=default \\ + uv run python -m cloud_run_worker.worker +""" + +import asyncio +import logging +import os +import signal +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer + +from temporalio.client import Client, TLSConfig +from temporalio.worker import Worker + +from cloud_run_worker.activities import hello_activity +from cloud_run_worker.workflows import TASK_QUEUE, SampleWorkflow + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +class _HealthHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: + self.send_response(200) + self.end_headers() + self.wfile.write(b"ok") + + def log_message(self, *_args) -> None: # silence per-request access logs + pass + + +def _start_health_server() -> None: + """Serve HTTP 200 on $PORT so Cloud Run's startup probe passes. + + Cloud Run's container contract requires the container to listen on the + port named by $PORT. A Temporal worker only polls Temporal, so we run a + tiny health endpoint in a daemon thread to satisfy that contract. + """ + port = int(os.environ.get("PORT", "8080")) + server = HTTPServer(("0.0.0.0", port), _HealthHandler) + threading.Thread(target=server.serve_forever, daemon=True).start() + logger.info("Health server listening on :%d", port) + + +def _tls_config() -> TLSConfig | None: + """Return a TLSConfig when mTLS env vars are present, otherwise None.""" + cert_b64 = os.environ.get("TEMPORAL_TLS_CERT") + key_b64 = os.environ.get("TEMPORAL_TLS_KEY") + if cert_b64 and key_b64: + import base64 + + return TLSConfig( + client_cert=base64.b64decode(cert_b64), + client_private_key=base64.b64decode(key_b64), + ) + return None + + +async def run() -> None: + # Open the health port first so Cloud Run's startup probe passes even + # while the Temporal connection is still being established. + _start_health_server() + + address = os.environ.get("TEMPORAL_ADDRESS", "localhost:7233") + namespace = os.environ.get("TEMPORAL_NAMESPACE", "default") + task_queue = os.environ.get("TEMPORAL_TASK_QUEUE", TASK_QUEUE) + + logger.info( + "Connecting to Temporal at %s (namespace=%s, task_queue=%s)", + address, + namespace, + task_queue, + ) + + client = await Client.connect( + address, + namespace=namespace, + tls=_tls_config() or False, + ) + + shutdown_event = asyncio.Event() + + def _handle_sigterm(*_) -> None: # noqa: ANN002 + logger.info("SIGTERM received — initiating graceful shutdown") + shutdown_event.set() + + signal.signal(signal.SIGTERM, _handle_sigterm) + signal.signal(signal.SIGINT, _handle_sigterm) + + async with Worker( + client, + task_queue=task_queue, + workflows=[SampleWorkflow], + activities=[hello_activity], + ): + logger.info( + "Worker running on task queue %r. Waiting for shutdown signal.", task_queue + ) + await shutdown_event.wait() + + logger.info("Worker shut down cleanly.") + + +if __name__ == "__main__": + asyncio.run(run()) diff --git a/cloud_run_worker/workflows.py b/cloud_run_worker/workflows.py new file mode 100644 index 000000000..44a34446e --- /dev/null +++ b/cloud_run_worker/workflows.py @@ -0,0 +1,22 @@ +from datetime import timedelta + +from temporalio import workflow + +TASK_QUEUE = "cloud-run-task-queue" + +with workflow.unsafe.imports_passed_through(): + from cloud_run_worker.activities import hello_activity + + +@workflow.defn +class SampleWorkflow: + @workflow.run + async def run(self, name: str) -> str: + workflow.logger.info("SampleWorkflow started with name: %s", name) + result = await workflow.execute_activity( + hello_activity, + name, + start_to_close_timeout=timedelta(seconds=10), + ) + workflow.logger.info("SampleWorkflow completed with result: %s", result) + return result diff --git a/pyproject.toml b/pyproject.toml index 3ccff7c62..984c9c3ba 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -89,11 +89,11 @@ allow-direct-references = true [tool.hatch.build.targets.sdist] include = ["./**/*.py"] -exclude = ["lambda_worker/**"] +exclude = ["lambda_worker/**", "cloud_run_worker/**"] [tool.hatch.build.targets.wheel] include = ["./**/*.py"] -exclude = ["lambda_worker/**"] +exclude = ["lambda_worker/**", "cloud_run_worker/**"] packages = [ "activity_worker", "bedrock", @@ -159,12 +159,12 @@ log_cli_format = "%(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(linen [tool.ruff] target-version = "py310" -extend-exclude = ["lambda_worker"] +extend-exclude = ["lambda_worker", "cloud_run_worker"] [tool.mypy] ignore_missing_imports = true namespace_packages = true -exclude = ["lambda_worker/"] +exclude = ["lambda_worker/", "cloud_run_worker/"] [[tool.mypy.overrides]] module = "aiohttp.*" diff --git a/tests/cloud_run_worker/__init__.py b/tests/cloud_run_worker/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/cloud_run_worker/worker_test.py b/tests/cloud_run_worker/worker_test.py new file mode 100644 index 000000000..ce6641811 --- /dev/null +++ b/tests/cloud_run_worker/worker_test.py @@ -0,0 +1,26 @@ +import uuid + +from temporalio.client import Client +from temporalio.worker import Worker + +from cloud_run_worker.activities import hello_activity +from cloud_run_worker.workflows import SampleWorkflow + + +async def test_execute_workflow(client: Client): + task_queue_name = str(uuid.uuid4()) + + async with Worker( + client, + task_queue=task_queue_name, + workflows=[SampleWorkflow], + activities=[hello_activity], + ): + result = await client.execute_workflow( + SampleWorkflow.run, + "World", + id=str(uuid.uuid4()), + task_queue=task_queue_name, + ) + + assert result == "Hello, World!"