diff --git a/gcp/__init__.py b/gcp/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/gcp/cloud_run/__init__.py b/gcp/cloud_run/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/gcp/cloud_run/worker_id/.dockerignore b/gcp/cloud_run/worker_id/.dockerignore new file mode 100644 index 00000000..e5cd7d89 --- /dev/null +++ b/gcp/cloud_run/worker_id/.dockerignore @@ -0,0 +1,8 @@ +* +!Dockerfile +!pyproject.toml +!__init__.py +!activities.py +!settings.py +!worker.py +!workflows.py diff --git a/gcp/cloud_run/worker_id/Dockerfile b/gcp/cloud_run/worker_id/Dockerfile new file mode 100644 index 00000000..536ec227 --- /dev/null +++ b/gcp/cloud_run/worker_id/Dockerfile @@ -0,0 +1,60 @@ +# syntax=docker/dockerfile:1 + +# NOTE: the Google Cloud Run metadata helper is not released yet, so +# `pyproject.toml` pins `temporalio` to a local path source +# (../../../../sdk-python-2) that is OUTSIDE this build context and therefore +# unavailable here. Before building the image, either: +# * wait for an SDK release that includes the helper and drop the +# [tool.uv.sources] override so `temporalio` installs from PyPI, or +# * replace that override with a pushed git revision, e.g. +# temporalio = { git = "https://github.com/temporalio/sdk-python", branch = "cloud-run-worker-id" } +# Installing `temporalio` from git compiles the Rust core, which is why the +# builder stage below starts from a Rust toolchain image. Once the helper ships +# on PyPI you can install the released wheel and drop the Rust builder entirely. + +FROM rust:1.91.0-slim-bookworm AS builder + +COPY --from=ghcr.io/astral-sh/uv:0.8.15 /uv /uvx /bin/ + +RUN apt-get update \ + && apt-get install --no-install-recommends --yes \ + build-essential \ + ca-certificates \ + git \ + libprotobuf-dev \ + pkg-config \ + protobuf-compiler \ + python3 \ + python3-dev \ + && rm -rf /var/lib/apt/lists/* + +ENV UV_COMPILE_BYTECODE=1 \ + UV_LINK_MODE=copy \ + UV_PYTHON=/usr/bin/python3 + +WORKDIR /app +COPY pyproject.toml ./ +RUN uv sync --no-dev + +FROM debian:bookworm-slim + +ENV PATH=/app/.venv/bin:$PATH \ + PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +RUN apt-get update \ + && apt-get install --no-install-recommends --yes ca-certificates python3 \ + && rm -rf /var/lib/apt/lists/* \ + && groupadd --system app \ + && useradd --system --gid app --create-home app + +WORKDIR /app +COPY --from=builder /app/.venv /app/.venv +# The build context is this sample directory (flat), so recreate the +# gcp/cloud_run/worker_id package path and its parent package markers. +COPY --chown=app:app __init__.py activities.py settings.py worker.py workflows.py /app/gcp/cloud_run/worker_id/ +RUN touch /app/gcp/__init__.py /app/gcp/cloud_run/__init__.py \ + && chown -R app:app /app/gcp + +USER app +CMD ["python", "-m", "gcp.cloud_run.worker_id.worker"] diff --git a/gcp/cloud_run/worker_id/README.md b/gcp/cloud_run/worker_id/README.md new file mode 100644 index 00000000..81bc4307 --- /dev/null +++ b/gcp/cloud_run/worker_id/README.md @@ -0,0 +1,143 @@ +# Google Cloud Run Worker Identity + +This sample runs a long-lived Temporal Worker in a [Google Cloud Run worker +pool](https://cloud.google.com/run/docs/worker-pools) and uses the +[`temporalio.contrib.gcp.cloud_run.worker_id`](https://python.temporal.io/temporalio.contrib.gcp.cloud_run.worker_id.html) +`WorkerIDPlugin` to derive the worker's identity from Cloud Run instance +metadata. + +Cloud Run runs a long-lived container rather than a per-invocation handler, so +this is a small metadata-driven plugin -- not a worker wrapper. The worker +registers `WorkerIDPlugin()` on the client via `Client.connect(plugins=[...])`. +At connect time the plugin: + +- reads the worker-pool or service name from `CLOUD_RUN_WORKER_POOL` (worker + pools), falling back to `K_SERVICE` (services); +- reads the revision from `CLOUD_RUN_REVISION`, falling back to `K_REVISION`; +- fetches this container's unique instance id from the Cloud Run metadata server. + +From that it sets the client `identity` to `@` (unless you +passed one), so each running container is individually identifiable as a poller. +Client plugins propagate to workers automatically, so there is nothing to wire up +on the `Worker`. The sample registers a simple greeting Workflow and Activity, +but the pattern applies to any Workflow/Activity definitions. + +> **Worker pools vs. services.** A Cloud Run *worker pool* has no HTTP endpoint; +> it is designed for long-running background workloads such as a Temporal +> Worker, which is why it is the primary target here. Worker pools use manual +> scaling and active instances are billed continuously, so remember to scale to +> zero after testing. + +> **This helper is not released yet.** `pyproject.toml` pins `temporalio` to a +> local path source (`../../../../sdk-python-2`) so the sample can be run and +> type-checked locally. Drop that `[tool.uv.sources]` override once an SDK +> release that includes the helper is on PyPI. The local path is not available +> inside a Docker build context, so the container build must use a released or +> git-referenced `temporalio`; see the `Dockerfile`. + +## Files + +| File | Description | +|------|-------------| +| `worker.py` | Long-lived worker: registers `WorkerIDPlugin` to set identity from Cloud Run metadata, then runs until SIGTERM | +| `workflows.py` | Sample Workflow that executes a greeting Activity | +| `activities.py` | Sample Activity that returns a greeting string | +| `settings.py` | Reads `TEMPORAL_*` connection settings from the environment | +| `starter.py` | Helper program to start a Workflow execution from a local machine | +| `Dockerfile` | Builds the worker container image | +| `.dockerignore` | Limits the Docker build context to the worker sources | +| `pyproject.toml` | Standalone dependencies for the sample | + +## Prerequisites + +- A [Temporal Cloud](https://temporal.io/cloud) namespace, or a self-hosted + Temporal cluster reachable from Cloud Run (a plaintext connection is fine). +- A Google Cloud project with billing enabled and the Google Cloud CLI + (`gcloud`) authenticated to it. +- Permission to manage Cloud Run worker pools and Cloud Build. +- Python 3.10+ and [`uv`](https://docs.astral.sh/uv/) to run the starter + locally. + +## Configuration + +The worker and starter read the same environment variables: + +| Variable | Required | Default | Description | +|----------|----------|---------|-------------| +| `TEMPORAL_TASK_QUEUE` | yes | -- | Task queue the worker polls and the starter targets | +| `TEMPORAL_ADDRESS` | no | `localhost:7233` | Temporal frontend address | +| `TEMPORAL_NAMESPACE` | no | `default` | Temporal namespace | +| `TEMPORAL_API_KEY` | no | -- | Set for Temporal Cloud; presence enables TLS | + +## 1. Deploy the worker pool + +Deploy from source; Cloud Build builds the image from the `Dockerfile` and Cloud +Run starts one instance: + +```bash +gcloud run worker-pools deploy temporal-worker \ + --source . \ + --region us-central1 \ + --set-env-vars TEMPORAL_ADDRESS=your-namespace.account-id.tmprl.cloud:7233,TEMPORAL_NAMESPACE=your-namespace.account-id,TEMPORAL_TASK_QUEUE=gcp-cloud-run +``` + +For a self-hosted plaintext server, set `TEMPORAL_ADDRESS` to its +`host:7233` and omit any API key. For Temporal Cloud, provide the API key as a +secret rather than a plaintext env var, for example: + +```bash +gcloud run worker-pools deploy temporal-worker \ + --source . \ + --region us-central1 \ + --set-env-vars TEMPORAL_ADDRESS=your-namespace.account-id.tmprl.cloud:7233,TEMPORAL_NAMESPACE=your-namespace.account-id,TEMPORAL_TASK_QUEUE=gcp-cloud-run \ + --set-secrets TEMPORAL_API_KEY=temporal-api-key:latest +``` + +Cloud Run sets `CLOUD_RUN_WORKER_POOL` and `CLOUD_RUN_REVISION` in the +container, which the helper reads automatically -- you do not set them yourself. + +## 2. Confirm the worker registered + +Check the worker-pool logs for the startup line, which reports the derived +worker identity: + +```bash +gcloud run worker-pools logs read temporal-worker --region us-central1 --limit 50 +``` + +You can also confirm the poller identity `@` with the +Temporal CLI: + +```bash +temporal task-queue describe --task-queue gcp-cloud-run +``` + +## 3. Start a Workflow + +Run the starter locally from the repository root, against the same Temporal +service and task queue: + +```bash +TEMPORAL_ADDRESS=your-namespace.account-id.tmprl.cloud:7233 \ +TEMPORAL_NAMESPACE=your-namespace.account-id \ +TEMPORAL_TASK_QUEUE=gcp-cloud-run \ +TEMPORAL_API_KEY="$(cat /secure/path/to/temporal-api-key)" \ + uv run python -m gcp.cloud_run.worker_id.starter +``` + +The expected output ends with: + +```text +Workflow result: Hello, Cloud Run worker pool! +``` + +## 4. Scale to zero + +Worker-pool instances are billed while running, so scale to zero when finished: + +```bash +gcloud run worker-pools update temporal-worker --instances 0 --region us-central1 +``` + +Cloud Run sends `SIGTERM`, and the worker begins a graceful Temporal Worker +shutdown before the process exits. diff --git a/gcp/cloud_run/worker_id/__init__.py b/gcp/cloud_run/worker_id/__init__.py new file mode 100644 index 00000000..17ffcc1b --- /dev/null +++ b/gcp/cloud_run/worker_id/__init__.py @@ -0,0 +1 @@ +"""Google Cloud Run worker-pool worker-identity sample.""" diff --git a/gcp/cloud_run/worker_id/activities.py b/gcp/cloud_run/worker_id/activities.py new file mode 100644 index 00000000..13540ceb --- /dev/null +++ b/gcp/cloud_run/worker_id/activities.py @@ -0,0 +1,11 @@ +"""Activity used by the Cloud Run worker sample.""" + +from __future__ import annotations + +from temporalio import activity + + +@activity.defn +async def compose_greeting(name: str) -> str: + activity.logger.info("Composing greeting for %s", name) + return f"Hello, {name}!" diff --git a/gcp/cloud_run/worker_id/pyproject.toml b/gcp/cloud_run/worker_id/pyproject.toml new file mode 100644 index 00000000..3a588e45 --- /dev/null +++ b/gcp/cloud_run/worker_id/pyproject.toml @@ -0,0 +1,32 @@ +[project] +name = "temporalio-samples-gcp-cloud-run-worker-id" +version = "0.1a1" +description = "Temporal worker identity on a Google Cloud Run worker pool" +authors = [{ name = "Temporal Technologies Inc", email = "sdk@temporal.io" }] +requires-python = ">=3.10" +readme = "README.md" +license = "MIT" +dependencies = ["temporalio>=1.31.0,<2"] + +[dependency-groups] +dev = [ + "ruff>=0.5.0,<0.6", + "mypy>=1.4.1,<2", +] + +[tool.uv] +package = false + +# TEMPORARY: the Google Cloud Run plugin +# (temporalio.contrib.gcp.cloud_run.worker_id.WorkerIDPlugin) is not +# released yet, so this pins `temporalio` to the local SDK checkout on the +# `cloud-run-worker-id` branch so the sample can be run and type-checked +# locally. Remove this [tool.uv.sources] override and rely on the released +# `temporalio` above once an SDK release that includes the helper is on PyPI. +# +# Note: this local path is NOT available inside a Docker build context (the +# build context is this directory only), so the container build must use a +# released or git-referenced `temporalio` instead -- see the Dockerfile and +# README. +[tool.uv.sources] +temporalio = { path = "../../../../sdk-python-2", editable = true } diff --git a/gcp/cloud_run/worker_id/settings.py b/gcp/cloud_run/worker_id/settings.py new file mode 100644 index 00000000..a884736e --- /dev/null +++ b/gcp/cloud_run/worker_id/settings.py @@ -0,0 +1,42 @@ +"""Temporal connection settings shared by the worker and the starter. + +Values are read from the environment so the same code runs against a local +plaintext dev server or Temporal Cloud. Set ``TEMPORAL_API_KEY`` to connect to +Temporal Cloud (which enables TLS); leave it unset for a plaintext connection. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass + + +@dataclass(frozen=True) +class Settings: + address: str + namespace: str + task_queue: str + api_key: str | None + + @property + def tls(self) -> bool: + # Temporal Cloud requires TLS; a plaintext self-hosted server does not. + return self.api_key is not None + + +def load_settings() -> Settings: + """Build connection settings from TEMPORAL_* environment variables.""" + task_queue = os.environ.get("TEMPORAL_TASK_QUEUE") + if not task_queue: + raise RuntimeError("TEMPORAL_TASK_QUEUE must be set to a non-empty value") + + # Secret managers frequently preserve a trailing newline; strip it. + api_key = os.environ.get("TEMPORAL_API_KEY") + api_key = api_key.strip() if api_key else None + + return Settings( + address=os.environ.get("TEMPORAL_ADDRESS") or "localhost:7233", + namespace=os.environ.get("TEMPORAL_NAMESPACE") or "default", + task_queue=task_queue, + api_key=api_key or None, + ) diff --git a/gcp/cloud_run/worker_id/starter.py b/gcp/cloud_run/worker_id/starter.py new file mode 100644 index 00000000..5a4b4373 --- /dev/null +++ b/gcp/cloud_run/worker_id/starter.py @@ -0,0 +1,36 @@ +"""Start a GreetingWorkflow on the Cloud Run worker's task queue. + +Run this locally against the same Temporal service the worker connects to, using +the same TEMPORAL_* environment variables. +""" + +from __future__ import annotations + +import asyncio + +from temporalio.client import Client + +from gcp.cloud_run.worker_id.settings import load_settings +from gcp.cloud_run.worker_id.workflows import GreetingWorkflow + + +async def main() -> None: + settings = load_settings() + client = await Client.connect( + settings.address, + namespace=settings.namespace, + api_key=settings.api_key, + tls=settings.tls, + ) + + result = await client.execute_workflow( + GreetingWorkflow.run, + "Cloud Run worker pool", + id="gcp-cloud-run-worker-id-sample", + task_queue=settings.task_queue, + ) + print(f"Workflow result: {result}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/gcp/cloud_run/worker_id/worker.py b/gcp/cloud_run/worker_id/worker.py new file mode 100644 index 00000000..1de4763a --- /dev/null +++ b/gcp/cloud_run/worker_id/worker.py @@ -0,0 +1,81 @@ +"""Run a long-lived Temporal worker on a Google Cloud Run worker pool. + +The worker registers ``temporalio.contrib.gcp.cloud_run.worker_id.WorkerIDPlugin`` +on the client. The plugin reads Cloud Run instance metadata and sets the client +identity to ``@``, which then propagates to the worker +automatically. The worker runs until Cloud Run sends SIGTERM (for example, on +scale-down). +""" + +from __future__ import annotations + +import asyncio +import signal + +from temporalio.client import Client +from temporalio.contrib.gcp.cloud_run.worker_id import ( + WorkerIDPlugin, + get_google_cloud_run_metadata, +) +from temporalio.worker import Worker + +from gcp.cloud_run.worker_id.activities import compose_greeting +from gcp.cloud_run.worker_id.settings import load_settings +from gcp.cloud_run.worker_id.workflows import GreetingWorkflow + + +async def main() -> None: + settings = load_settings() + + # WorkerIDPlugin reads CLOUD_RUN_WORKER_POOL/CLOUD_RUN_REVISION (worker pools) + # or K_SERVICE/K_REVISION (services), fetches this instance's unique id from + # the Cloud Run metadata server at connect time, and raises if not running on + # Cloud Run. It sets the client identity to @ so each + # running container is identifiable. Client plugins propagate to workers + # automatically, so there is nothing to wire up on the Worker. + # @@@SNIPSTART python-cloud-run-worker-id + client = await Client.connect( + settings.address, + namespace=settings.namespace, + plugins=[WorkerIDPlugin()], + api_key=settings.api_key, + tls=settings.tls, + ) + # @@@SNIPEND + + worker = Worker( + client, + task_queue=settings.task_queue, + workflows=[GreetingWorkflow], + activities=[compose_greeting], + ) + + loop = asyncio.get_running_loop() + shutdown_requested = False + + def request_shutdown() -> None: + nonlocal shutdown_requested + if shutdown_requested: + return + shutdown_requested = True + print("Worker shutdown requested", flush=True) + loop.create_task(worker.shutdown()) + + for signum in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(signum, request_shutdown) + + # The plugin already applied this identity to the client; read the Cloud Run + # instance metadata to log the same worker_identity value at startup. + metadata = get_google_cloud_run_metadata() + print( + "Worker starting " + f"identity={metadata.worker_identity} " + f"task_queue={settings.task_queue}", + flush=True, + ) + await worker.run() + print("Worker stopped", flush=True) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/gcp/cloud_run/worker_id/workflows.py b/gcp/cloud_run/worker_id/workflows.py new file mode 100644 index 00000000..dfa75a62 --- /dev/null +++ b/gcp/cloud_run/worker_id/workflows.py @@ -0,0 +1,22 @@ +"""Workflow used by the Cloud Run worker sample.""" + +from __future__ import annotations + +from datetime import timedelta + +from temporalio import workflow + +with workflow.unsafe.imports_passed_through(): + from gcp.cloud_run.worker_id.activities import compose_greeting + + +@workflow.defn +class GreetingWorkflow: + @workflow.run + async def run(self, name: str) -> str: + workflow.logger.info("GreetingWorkflow started for %s", name) + return await workflow.execute_activity( + compose_greeting, + name, + start_to_close_timeout=timedelta(seconds=10), + ) diff --git a/tests/gcp/__init__.py b/tests/gcp/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/gcp/cloud_run/__init__.py b/tests/gcp/cloud_run/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/gcp/cloud_run/worker_id/__init__.py b/tests/gcp/cloud_run/worker_id/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/gcp/cloud_run/worker_id/workflow_test.py b/tests/gcp/cloud_run/worker_id/workflow_test.py new file mode 100644 index 00000000..b360b7d2 --- /dev/null +++ b/tests/gcp/cloud_run/worker_id/workflow_test.py @@ -0,0 +1,25 @@ +import uuid + +from temporalio.client import Client +from temporalio.worker import Worker + +from gcp.cloud_run.worker_id.activities import compose_greeting +from gcp.cloud_run.worker_id.workflows import GreetingWorkflow + + +async def test_greeting_workflow(client: Client): + task_queue_name = str(uuid.uuid4()) + + async with Worker( + client, + task_queue=task_queue_name, + workflows=[GreetingWorkflow], + activities=[compose_greeting], + ): + result = await client.execute_workflow( + GreetingWorkflow.run, + "Temporal", + id=str(uuid.uuid4()), + task_queue=task_queue_name, + ) + assert result == "Hello, Temporal!"