From 477de64f0d2890e8cdd813b28e56b39be54453aa Mon Sep 17 00:00:00 2001 From: seanbollin Date: Fri, 21 Aug 2026 12:22:49 -0700 Subject: [PATCH 1/6] Add Google Cloud Run worker identity/deployment helper Adds an experimental Google Cloud Run helper, mirroring the existing AWS Lambda module's worker-ID behavior. Because Cloud Run runs a long-lived container (unlike Lambda's per-invocation model), this is a metadata helper rather than a worker wrapper: it reads the Cloud Run instance metadata -- the instance id from the metadata server, plus the worker pool/service name and revision from CLOUD_RUN_WORKER_POOL / CLOUD_RUN_REVISION (worker pools) or K_SERVICE / K_REVISION (services) -- and derives a worker identity and a WorkerDeploymentVersion to apply to a normal long-lived worker. Covers both Cloud Run worker pools and services. Co-Authored-By: Claude Opus 4.8 --- temporalio/contrib/gcp/__init__.py | 1 + temporalio/contrib/gcp/cloud_run/README.md | 73 +++++++++ temporalio/contrib/gcp/cloud_run/__init__.py | 47 ++++++ temporalio/contrib/gcp/cloud_run/_metadata.py | 150 ++++++++++++++++++ 4 files changed, 271 insertions(+) create mode 100644 temporalio/contrib/gcp/__init__.py create mode 100644 temporalio/contrib/gcp/cloud_run/README.md create mode 100644 temporalio/contrib/gcp/cloud_run/__init__.py create mode 100644 temporalio/contrib/gcp/cloud_run/_metadata.py diff --git a/temporalio/contrib/gcp/__init__.py b/temporalio/contrib/gcp/__init__.py new file mode 100644 index 000000000..f87ca7c64 --- /dev/null +++ b/temporalio/contrib/gcp/__init__.py @@ -0,0 +1 @@ +"""Google Cloud integrations for Temporal SDK.""" diff --git a/temporalio/contrib/gcp/cloud_run/README.md b/temporalio/contrib/gcp/cloud_run/README.md new file mode 100644 index 000000000..ad9df9bba --- /dev/null +++ b/temporalio/contrib/gcp/cloud_run/README.md @@ -0,0 +1,73 @@ +# cloud_run + +> ⚠️ **This package is currently at an experimental release stage.** ⚠️ + +A metadata helper for running [Temporal](https://temporal.io) workers on Google Cloud Run. +Cloud Run runs a long-lived container -- there is no per-invocation handler to wrap -- so this is +**not** a worker wrapper. Instead, `get_google_cloud_run_metadata` reads Cloud Run instance metadata +and hands you a worker identity string and a `WorkerDeploymentConfig` to drop into your normal, +long-lived worker. Both Cloud Run **worker pools** and **services** are supported. + +## Quick start + +```python +import asyncio + +from temporalio.client import Client +from temporalio.contrib.gcp.cloud_run import get_google_cloud_run_metadata +from temporalio.worker import Worker + +from my_workflows import MyWorkflow +from my_activities import my_activity + + +async def main() -> None: + metadata = get_google_cloud_run_metadata() + + client = await Client.connect( + "localhost:7233", + identity=metadata.worker_identity, + ) + + worker = Worker( + client, + task_queue="my-task-queue", + workflows=[MyWorkflow], + activities=[my_activity], + deployment_config=metadata.worker_deployment_config, + ) + await worker.run() + + +if __name__ == "__main__": + asyncio.run(main()) +``` + +## How it works + +Cloud Run exposes workload metadata through environment variables and a metadata server: + +- **Worker pools** get `CLOUD_RUN_WORKER_POOL` and `CLOUD_RUN_REVISION` (and no `K_*` variables). +- **Services** get `K_SERVICE`, `K_REVISION`, and `K_CONFIGURATION` (and no `CLOUD_RUN_*` variables). + +The unique instance id is not available as an environment variable on either; it is only exposed by +the +[Cloud Run metadata server](https://cloud.google.com/run/docs/container-contract#metadata-server) +at `http://metadata.google.internal/computeMetadata/v1/instance/id`, which requires the +`Metadata-Flavor: Google` request header. + +`get_google_cloud_run_metadata` resolves the deployment name from `CLOUD_RUN_WORKER_POOL` (falling +back to `K_SERVICE`) and the revision from `CLOUD_RUN_REVISION` (falling back to `K_REVISION`), then +performs a single synchronous HTTP GET to the metadata server for the instance id. It returns a +`GoogleCloudRunMetadata` with these conveniences: + +- `worker_identity` -- `@`, uniquely identifying this worker instance in + Temporal tooling. It falls back to `@`, then to just ``, when the + revision or name is unavailable. +- `worker_deployment_version` -- a `WorkerDeploymentVersion` whose `deployment_name` is the Cloud + Run workload name and whose `build_id` is the Cloud Run revision, for use with Worker Versioning. +- `worker_deployment_config` -- a `WorkerDeploymentConfig` wrapping that version with + `use_worker_versioning=True`, ready to pass to `Worker(..., deployment_config=...)`. + +Because the metadata server is only reachable from within Cloud Run, calling this helper elsewhere +raises a clear error. It uses only the Python standard library and adds no new dependencies. diff --git a/temporalio/contrib/gcp/cloud_run/__init__.py b/temporalio/contrib/gcp/cloud_run/__init__.py new file mode 100644 index 000000000..341b19097 --- /dev/null +++ b/temporalio/contrib/gcp/cloud_run/__init__.py @@ -0,0 +1,47 @@ +"""Metadata helpers for running Temporal workers on Google Cloud Run. + +Cloud Run runs a long-lived container rather than a per-invocation handler, so this module is a +small metadata helper -- **not** a worker wrapper. :py:func:`get_google_cloud_run_metadata` reads +Cloud Run instance metadata (from a worker pool or a service) and hands you a worker identity string +and a :py:class:`temporalio.worker.WorkerDeploymentConfig` to drop into a normal, long-lived worker. + +.. warning:: + Google Cloud Run support is experimental. + +Quick start:: + + import asyncio + + from temporalio.client import Client + from temporalio.contrib.gcp.cloud_run import get_google_cloud_run_metadata + from temporalio.worker import Worker + + async def main() -> None: + metadata = get_google_cloud_run_metadata() + + client = await Client.connect( + "localhost:7233", + identity=metadata.worker_identity, + ) + + worker = Worker( + client, + task_queue="my-task-queue", + workflows=[MyWorkflow], + activities=[my_activity], + deployment_config=metadata.worker_deployment_config, + ) + await worker.run() + + asyncio.run(main()) +""" + +from temporalio.contrib.gcp.cloud_run._metadata import ( + GoogleCloudRunMetadata, + get_google_cloud_run_metadata, +) + +__all__ = [ + "GoogleCloudRunMetadata", + "get_google_cloud_run_metadata", +] diff --git a/temporalio/contrib/gcp/cloud_run/_metadata.py b/temporalio/contrib/gcp/cloud_run/_metadata.py new file mode 100644 index 000000000..5eef68ea9 --- /dev/null +++ b/temporalio/contrib/gcp/cloud_run/_metadata.py @@ -0,0 +1,150 @@ +"""Read Google Cloud Run instance metadata for Temporal worker configuration. + +Cloud Run runs a long-lived container rather than a per-invocation handler, so this module is a +small metadata helper -- not a worker wrapper. It derives a worker identity and a +:py:class:`temporalio.common.WorkerDeploymentVersion` from Cloud Run instance metadata for use with +a normal, long-lived worker. Both Cloud Run worker pools and services are supported. + +.. warning:: + Google Cloud Run support is experimental. +""" + +from __future__ import annotations + +import os +import urllib.request +from collections.abc import Callable +from dataclasses import dataclass +from typing import TYPE_CHECKING + +import temporalio.common + +if TYPE_CHECKING: + import temporalio.worker + + +@dataclass(frozen=True) +class GoogleCloudRunMetadata: + """Identifying metadata for the current Google Cloud Run instance. + + Both Cloud Run worker pools and services are supported. Worker pools expose + ``CLOUD_RUN_WORKER_POOL`` and ``CLOUD_RUN_REVISION``; services expose ``K_SERVICE`` and + ``K_REVISION``. + + Attributes: + instance_id: Unique id of this Cloud Run container instance, read from the Cloud Run + metadata server. + name: Deployment name of this Cloud Run workload -- the worker pool name + (``CLOUD_RUN_WORKER_POOL``) or, for a service, the service name (``K_SERVICE``). May be + empty when the process is not running on Cloud Run. + revision: Cloud Run revision name (``CLOUD_RUN_REVISION`` for worker pools or ``K_REVISION`` + for services). May be empty when the process is not running on Cloud Run. + """ + + instance_id: str + name: str + revision: str + + @property + def worker_identity(self) -> str: + """Worker identity string uniquely identifying this Cloud Run instance. + + The format is ``@``. When the revision is empty the deployment name + is used instead (``@``), and when both are empty the instance id is + returned on its own. + """ + if self.revision: + return f"{self.instance_id}@{self.revision}" + if self.name: + return f"{self.instance_id}@{self.name}" + return self.instance_id + + @property + def worker_deployment_version(self) -> temporalio.common.WorkerDeploymentVersion: + """Worker Versioning deployment version derived from this instance's metadata. + + The deployment name is the Cloud Run workload name and the build id is the Cloud Run + revision. + + Raises: + ValueError: If either the name or the revision is empty, which usually means the process + is not running on a Cloud Run worker pool or service. + """ + if not self.name or not self.revision: + raise ValueError( + "Cannot build a WorkerDeploymentVersion without both a Cloud Run deployment name " + "(CLOUD_RUN_WORKER_POOL or K_SERVICE) and revision (CLOUD_RUN_REVISION or " + "K_REVISION); this process may not be running on a Cloud Run worker pool or " + "service." + ) + return temporalio.common.WorkerDeploymentVersion( + deployment_name=self.name, + build_id=self.revision, + ) + + @property + def worker_deployment_config(self) -> temporalio.worker.WorkerDeploymentConfig: + """Worker deployment config with Worker Versioning enabled for this instance. + + Pass this straight to :py:class:`temporalio.worker.Worker` as its ``deployment_config``. + + Raises: + ValueError: If either the name or the revision is empty, which usually means the process + is not running on a Cloud Run worker pool or service. + """ + from temporalio.worker import WorkerDeploymentConfig + + return WorkerDeploymentConfig( + version=self.worker_deployment_version, + use_worker_versioning=True, + ) + + +def get_google_cloud_run_metadata( + *, + timeout: float = 2.0, + metadata_url: str = "http://metadata.google.internal/computeMetadata/v1/instance/id", + getenv: Callable[[str], str] = os.environ.get, # type: ignore[assignment] +) -> GoogleCloudRunMetadata: + """Read metadata identifying the current Google Cloud Run instance. + + Resolves the deployment name from ``CLOUD_RUN_WORKER_POOL`` (Cloud Run worker pools), falling + back to ``K_SERVICE`` (Cloud Run services), and the revision from ``CLOUD_RUN_REVISION`` falling + back to ``K_REVISION``. The unique instance id is fetched from the Cloud Run metadata server + with a single synchronous HTTP GET. Intended to be called once at worker startup. + + Args: + timeout: Timeout, in seconds, for the request to the metadata server. + metadata_url: URL of the Cloud Run metadata server endpoint that returns the instance id. + getenv: Callable used to look up environment variables. Defaults to ``os.environ.get`` and + exists primarily for testing. + + Returns: + A :py:class:`GoogleCloudRunMetadata` describing the current instance. + + Raises: + RuntimeError: If the metadata server cannot be reached, which usually means the process is + not running on a Cloud Run worker pool or service. + """ + name = getenv("CLOUD_RUN_WORKER_POOL") or getenv("K_SERVICE") or "" + revision = getenv("CLOUD_RUN_REVISION") or getenv("K_REVISION") or "" + + request = urllib.request.Request( + metadata_url, + headers={"Metadata-Flavor": "Google"}, + method="GET", + ) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + instance_id = response.read().decode("utf-8").strip() + except OSError as err: + raise RuntimeError( + f"Failed to reach the Cloud Run metadata server at {metadata_url!r}; " + "this process may not be running on a Cloud Run worker pool or service." + ) from err + + return GoogleCloudRunMetadata( + instance_id=instance_id, + name=name, + revision=revision, + ) From 388d3efdecf4626bb1bf12368dc6c94b9fb1052e Mon Sep 17 00:00:00 2001 From: seanbollin Date: Tue, 25 Aug 2026 13:17:21 -0700 Subject: [PATCH 2/6] Set worker versioning behavior to PINNED in the Cloud Run worker apply The worker-side apply helper enabled versioning and set the deployment version but left the default versioning behavior unset, so a versioned worker with a plain (un-annotated) workflow failed to register. Default it to PINNED; a per-workflow versioning behavior still takes precedence. Co-Authored-By: Claude Opus 4.8 --- temporalio/contrib/gcp/cloud_run/README.md | 3 ++- temporalio/contrib/gcp/cloud_run/_metadata.py | 1 + 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/temporalio/contrib/gcp/cloud_run/README.md b/temporalio/contrib/gcp/cloud_run/README.md index ad9df9bba..041f5121e 100644 --- a/temporalio/contrib/gcp/cloud_run/README.md +++ b/temporalio/contrib/gcp/cloud_run/README.md @@ -67,7 +67,8 @@ performs a single synchronous HTTP GET to the metadata server for the instance i - `worker_deployment_version` -- a `WorkerDeploymentVersion` whose `deployment_name` is the Cloud Run workload name and whose `build_id` is the Cloud Run revision, for use with Worker Versioning. - `worker_deployment_config` -- a `WorkerDeploymentConfig` wrapping that version with - `use_worker_versioning=True`, ready to pass to `Worker(..., deployment_config=...)`. + `use_worker_versioning=True` and `default_versioning_behavior=VersioningBehavior.PINNED` (a + per-workflow behavior takes precedence), ready to pass to `Worker(..., deployment_config=...)`. Because the metadata server is only reachable from within Cloud Run, calling this helper elsewhere raises a clear error. It uses only the Python standard library and adds no new dependencies. diff --git a/temporalio/contrib/gcp/cloud_run/_metadata.py b/temporalio/contrib/gcp/cloud_run/_metadata.py index 5eef68ea9..784c8ead8 100644 --- a/temporalio/contrib/gcp/cloud_run/_metadata.py +++ b/temporalio/contrib/gcp/cloud_run/_metadata.py @@ -97,6 +97,7 @@ def worker_deployment_config(self) -> temporalio.worker.WorkerDeploymentConfig: return WorkerDeploymentConfig( version=self.worker_deployment_version, use_worker_versioning=True, + default_versioning_behavior=temporalio.common.VersioningBehavior.PINNED, ) From f5848b7ccadae8a20be250e2fa2f14f97098cc74 Mon Sep 17 00:00:00 2001 From: seanbollin Date: Tue, 25 Aug 2026 13:42:11 -0700 Subject: [PATCH 3/6] Add unit tests for the Google Cloud Run metadata helper Cover the temporalio.contrib.gcp.cloud_run helper: - Environment precedence for the deployment name (CLOUD_RUN_WORKER_POOL over K_SERVICE) and revision (CLOUD_RUN_REVISION over K_REVISION). - Worker identity formatting and its revision -> name -> instance-id fallbacks. - WorkerDeploymentVersion derivation and its ValueError when name/revision empty. - WorkerDeploymentConfig enabling worker versioning with PINNED default behavior. - The metadata HTTP fetch via a local in-process server: asserts the Metadata-Flavor: Google header is sent, the body is trimmed, and a clear RuntimeError is raised on non-200 and unreachable responses. Uses the helper's dependency-injection seams (getenv and metadata_url) so no real environment or network access is required. Co-Authored-By: Claude Opus 4.8 --- tests/contrib/gcp/__init__.py | 0 tests/contrib/gcp/cloud_run/__init__.py | 0 tests/contrib/gcp/cloud_run/test_metadata.py | 272 +++++++++++++++++++ 3 files changed, 272 insertions(+) create mode 100644 tests/contrib/gcp/__init__.py create mode 100644 tests/contrib/gcp/cloud_run/__init__.py create mode 100644 tests/contrib/gcp/cloud_run/test_metadata.py diff --git a/tests/contrib/gcp/__init__.py b/tests/contrib/gcp/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/contrib/gcp/cloud_run/__init__.py b/tests/contrib/gcp/cloud_run/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/contrib/gcp/cloud_run/test_metadata.py b/tests/contrib/gcp/cloud_run/test_metadata.py new file mode 100644 index 000000000..11c1ba763 --- /dev/null +++ b/tests/contrib/gcp/cloud_run/test_metadata.py @@ -0,0 +1,272 @@ +"""Tests for temporalio.contrib.gcp.cloud_run.""" + +from __future__ import annotations + +import socket +import threading +from collections.abc import Iterator +from email.message import Message +from http.server import BaseHTTPRequestHandler, HTTPServer +from typing import Any + +import pytest + +from temporalio.common import VersioningBehavior, WorkerDeploymentVersion +from temporalio.contrib.gcp.cloud_run import ( + GoogleCloudRunMetadata, + get_google_cloud_run_metadata, +) + + +def _metadata( + *, + instance_id: str = "instance-1", + name: str = "", + revision: str = "", +) -> GoogleCloudRunMetadata: + return GoogleCloudRunMetadata( + instance_id=instance_id, + name=name, + revision=revision, + ) + + +# ---- Local metadata-server fixture ---- + + +class _MetadataServer(HTTPServer): + """In-process stand-in for the Cloud Run metadata server. + + Records the headers and path of the last request and serves a configurable + status and body so tests can assert on both the request and the response. + """ + + url: str = "" + response_status: int = 200 + response_body: str = "instance-1" + received_path: str | None = None + received_headers: Message[str, str] | None = None + + +class _Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 (http.server naming) + server: _MetadataServer = self.server # type: ignore[assignment] + server.received_path = self.path + server.received_headers = self.headers + body = server.response_body.encode("utf-8") + self.send_response(server.response_status) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 + # Silence the default stderr request logging. The parameter is named + # ``format`` to match BaseHTTPRequestHandler.log_message. + pass + + +@pytest.fixture +def metadata_server() -> Iterator[_MetadataServer]: + server = _MetadataServer(("127.0.0.1", 0), _Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + server.url = f"http://127.0.0.1:{port}/computeMetadata/v1/instance/id" + try: + yield server + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def _closed_port() -> int: + """Return a port number that nothing is listening on.""" + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +# ---- Environment precedence ---- + + +class TestEnvPrecedence: + def test_worker_pool_wins_over_service( + self, metadata_server: _MetadataServer + ) -> None: + env = {"CLOUD_RUN_WORKER_POOL": "my-pool", "K_SERVICE": "my-service"} + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv=env.get, # type: ignore[arg-type] + ) + assert metadata.name == "my-pool" + + def test_service_used_when_pool_absent( + self, metadata_server: _MetadataServer + ) -> None: + env = {"K_SERVICE": "my-service"} + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv=env.get, # type: ignore[arg-type] + ) + assert metadata.name == "my-service" + + def test_name_empty_when_neither_set( + self, metadata_server: _MetadataServer + ) -> None: + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv={}.get, # type: ignore[arg-type] + ) + assert metadata.name == "" + + def test_cloud_run_revision_wins_over_k_revision( + self, metadata_server: _MetadataServer + ) -> None: + env = {"CLOUD_RUN_REVISION": "rev-cr", "K_REVISION": "rev-k"} + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv=env.get, # type: ignore[arg-type] + ) + assert metadata.revision == "rev-cr" + + def test_k_revision_used_when_cloud_run_revision_absent( + self, metadata_server: _MetadataServer + ) -> None: + env = {"K_REVISION": "rev-k"} + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv=env.get, # type: ignore[arg-type] + ) + assert metadata.revision == "rev-k" + + def test_revision_empty_when_neither_set( + self, metadata_server: _MetadataServer + ) -> None: + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv={}.get, # type: ignore[arg-type] + ) + assert metadata.revision == "" + + +# ---- Worker identity ---- + + +class TestWorkerIdentity: + def test_identity_uses_revision(self) -> None: + metadata = _metadata(instance_id="abc", name="my-pool", revision="rev-1") + assert metadata.worker_identity == "abc@rev-1" + + def test_identity_falls_back_to_name(self) -> None: + metadata = _metadata(instance_id="abc", name="my-pool", revision="") + assert metadata.worker_identity == "abc@my-pool" + + def test_identity_falls_back_to_instance_id(self) -> None: + metadata = _metadata(instance_id="abc", name="", revision="") + assert metadata.worker_identity == "abc" + + +# ---- Worker deployment version ---- + + +class TestWorkerDeploymentVersion: + def test_version_from_name_and_revision(self) -> None: + metadata = _metadata(instance_id="abc", name="my-pool", revision="rev-1") + assert metadata.worker_deployment_version == WorkerDeploymentVersion( + deployment_name="my-pool", + build_id="rev-1", + ) + + def test_version_errors_when_name_empty(self) -> None: + metadata = _metadata(instance_id="abc", name="", revision="rev-1") + with pytest.raises(ValueError, match="deployment name"): + _ = metadata.worker_deployment_version + + def test_version_errors_when_revision_empty(self) -> None: + metadata = _metadata(instance_id="abc", name="my-pool", revision="") + with pytest.raises(ValueError, match="revision"): + _ = metadata.worker_deployment_version + + +# ---- Worker deployment config ---- + + +class TestWorkerDeploymentConfig: + def test_config_enables_pinned_versioning(self) -> None: + metadata = _metadata(instance_id="abc", name="my-pool", revision="rev-1") + config = metadata.worker_deployment_config + assert config.use_worker_versioning is True + assert config.default_versioning_behavior == VersioningBehavior.PINNED + assert config.version == WorkerDeploymentVersion( + deployment_name="my-pool", + build_id="rev-1", + ) + + def test_config_errors_when_name_empty(self) -> None: + metadata = _metadata(instance_id="abc", name="", revision="rev-1") + with pytest.raises(ValueError, match="deployment name"): + _ = metadata.worker_deployment_config + + def test_config_errors_when_revision_empty(self) -> None: + metadata = _metadata(instance_id="abc", name="my-pool", revision="") + with pytest.raises(ValueError, match="revision"): + _ = metadata.worker_deployment_config + + +# ---- HTTP fetch ---- + + +class TestHttpFetch: + def test_sends_metadata_flavor_header_and_trims_body( + self, metadata_server: _MetadataServer + ) -> None: + metadata_server.response_body = " instance-xyz\n" + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv={}.get, # type: ignore[arg-type] + ) + assert metadata.instance_id == "instance-xyz" + assert metadata_server.received_headers is not None + assert metadata_server.received_headers.get("Metadata-Flavor") == "Google" + assert metadata_server.received_path == "/computeMetadata/v1/instance/id" + + def test_errors_on_non_200(self, metadata_server: _MetadataServer) -> None: + metadata_server.response_status = 500 + metadata_server.response_body = "boom" + with pytest.raises(RuntimeError, match="metadata server"): + get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv={}.get, # type: ignore[arg-type] + ) + + def test_errors_when_unreachable(self) -> None: + url = f"http://127.0.0.1:{_closed_port()}/computeMetadata/v1/instance/id" + with pytest.raises(RuntimeError, match="metadata server"): + get_google_cloud_run_metadata( + metadata_url=url, + timeout=1.0, + getenv={}.get, # type: ignore[arg-type] + ) + + def test_end_to_end_from_env_and_server( + self, metadata_server: _MetadataServer + ) -> None: + metadata_server.response_body = "instance-42" + env = {"CLOUD_RUN_WORKER_POOL": "my-pool", "CLOUD_RUN_REVISION": "rev-7"} + metadata = get_google_cloud_run_metadata( + metadata_url=metadata_server.url, + getenv=env.get, # type: ignore[arg-type] + ) + assert metadata == GoogleCloudRunMetadata( + instance_id="instance-42", + name="my-pool", + revision="rev-7", + ) + assert metadata.worker_identity == "instance-42@rev-7" + assert metadata.worker_deployment_version == WorkerDeploymentVersion( + deployment_name="my-pool", + build_id="rev-7", + ) From 4dbc31fc03d60c8e30986e9a2aa32d3900af72c2 Mon Sep 17 00:00:00 2001 From: seanbollin Date: Mon, 31 Aug 2026 13:13:31 -0700 Subject: [PATCH 4/6] Add CloudRunPlugin for Google Cloud Run worker defaults Re-architect the Cloud Run worker-ID helper into a plugin mirroring the SDK's OpenTelemetry Cloud Run plugin. CloudRunPlugin subclasses temporalio.plugin.SimplePlugin and is registered once on the client via Client.connect(plugins=[...]); it propagates to workers automatically. The plugin fetches Cloud Run instance metadata lazily at client connect and caches it, then sets the client identity (only when the caller did not provide one) and configures the worker with a PINNED WorkerDeploymentConfig derived from the Cloud Run revision. Connecting off Cloud Run fails fast with a clear error. The GoogleCloudRunMetadata dataclass and its worker_identity / worker_deployment_version / worker_deployment_config properties are kept for advanced and non-plugin use. Co-Authored-By: Claude Opus 4.8 --- temporalio/contrib/gcp/cloud_run/README.md | 68 +++++--- temporalio/contrib/gcp/cloud_run/__init__.py | 27 ++-- temporalio/contrib/gcp/cloud_run/_metadata.py | 7 +- temporalio/contrib/gcp/cloud_run/_plugin.py | 118 ++++++++++++++ tests/contrib/gcp/cloud_run/test_plugin.py | 148 ++++++++++++++++++ 5 files changed, 335 insertions(+), 33 deletions(-) create mode 100644 temporalio/contrib/gcp/cloud_run/_plugin.py create mode 100644 tests/contrib/gcp/cloud_run/test_plugin.py diff --git a/temporalio/contrib/gcp/cloud_run/README.md b/temporalio/contrib/gcp/cloud_run/README.md index 041f5121e..1edc9f67c 100644 --- a/temporalio/contrib/gcp/cloud_run/README.md +++ b/temporalio/contrib/gcp/cloud_run/README.md @@ -2,11 +2,21 @@ > ⚠️ **This package is currently at an experimental release stage.** ⚠️ -A metadata helper for running [Temporal](https://temporal.io) workers on Google Cloud Run. -Cloud Run runs a long-lived container -- there is no per-invocation handler to wrap -- so this is -**not** a worker wrapper. Instead, `get_google_cloud_run_metadata` reads Cloud Run instance metadata -and hands you a worker identity string and a `WorkerDeploymentConfig` to drop into your normal, -long-lived worker. Both Cloud Run **worker pools** and **services** are supported. +A plugin for running [Temporal](https://temporal.io) workers on Google Cloud Run. Cloud Run runs a +long-lived container -- there is no per-invocation handler to wrap -- so this is **not** a worker +wrapper. Instead, `CloudRunPlugin` reads Cloud Run instance metadata and configures a normal, +long-lived client and worker for you. Both Cloud Run **worker pools** and **services** are supported. + +Register the plugin once when connecting the client and it: + +- sets the client **identity** to a value derived from the Cloud Run instance (unless you already + passed an `identity`), and +- configures the worker with a `WorkerDeploymentConfig` that enables Worker Versioning with a + `PINNED` default behavior, so each Cloud Run revision is a distinct, pinned worker deployment + version. + +Client plugins automatically propagate to workers created from that client, so there is nothing to +wire up on the worker. ## Quick start @@ -14,7 +24,7 @@ long-lived worker. Both Cloud Run **worker pools** and **services** are supporte import asyncio from temporalio.client import Client -from temporalio.contrib.gcp.cloud_run import get_google_cloud_run_metadata +from temporalio.contrib.gcp.cloud_run import CloudRunPlugin from temporalio.worker import Worker from my_workflows import MyWorkflow @@ -22,11 +32,10 @@ from my_activities import my_activity async def main() -> None: - metadata = get_google_cloud_run_metadata() - + # Install the plugin on the client; it propagates to workers automatically. client = await Client.connect( "localhost:7233", - identity=metadata.worker_identity, + plugins=[CloudRunPlugin()], ) worker = Worker( @@ -34,7 +43,6 @@ async def main() -> None: task_queue="my-task-queue", workflows=[MyWorkflow], activities=[my_activity], - deployment_config=metadata.worker_deployment_config, ) await worker.run() @@ -56,19 +64,35 @@ the at `http://metadata.google.internal/computeMetadata/v1/instance/id`, which requires the `Metadata-Flavor: Google` request header. -`get_google_cloud_run_metadata` resolves the deployment name from `CLOUD_RUN_WORKER_POOL` (falling -back to `K_SERVICE`) and the revision from `CLOUD_RUN_REVISION` (falling back to `K_REVISION`), then -performs a single synchronous HTTP GET to the metadata server for the instance id. It returns a -`GoogleCloudRunMetadata` with these conveniences: +When the client connects, `CloudRunPlugin` resolves the deployment name from `CLOUD_RUN_WORKER_POOL` +(falling back to `K_SERVICE`) and the revision from `CLOUD_RUN_REVISION` (falling back to +`K_REVISION`), then performs a single synchronous HTTP GET to the metadata server for the instance +id. The result is **cached on the plugin**, so the worker hook reuses it without another network +call. From that metadata the plugin applies: -- `worker_identity` -- `@`, uniquely identifying this worker instance in +- **Client identity** -- `@`, uniquely identifying this worker instance in Temporal tooling. It falls back to `@`, then to just ``, when the - revision or name is unavailable. -- `worker_deployment_version` -- a `WorkerDeploymentVersion` whose `deployment_name` is the Cloud - Run workload name and whose `build_id` is the Cloud Run revision, for use with Worker Versioning. -- `worker_deployment_config` -- a `WorkerDeploymentConfig` wrapping that version with + revision or name is unavailable. An `identity` you pass to `Client.connect` always wins. +- **Worker deployment config** -- a `WorkerDeploymentConfig` whose version has `deployment_name` set + to the Cloud Run workload name and `build_id` set to the Cloud Run revision, with `use_worker_versioning=True` and `default_versioning_behavior=VersioningBehavior.PINNED` (a - per-workflow behavior takes precedence), ready to pass to `Worker(..., deployment_config=...)`. + per-workflow behavior takes precedence). + +Because the metadata server is only reachable from within Cloud Run, connecting elsewhere **fails +fast** with a clear error rather than silently doing nothing. The plugin uses only the Python +standard library and adds no new dependencies. + +## Advanced / non-plugin use -Because the metadata server is only reachable from within Cloud Run, calling this helper elsewhere -raises a clear error. It uses only the Python standard library and adds no new dependencies. +For advanced scenarios or unit tests you can bypass the metadata server by passing a pre-built +metadata object, or steer the fetch with `getenv` / `metadata_url` / `timeout`: + +```python +from temporalio.contrib.gcp.cloud_run import CloudRunPlugin, get_google_cloud_run_metadata + +metadata = get_google_cloud_run_metadata() +plugin = CloudRunPlugin(metadata=metadata) + +# metadata.worker_identity and metadata.worker_deployment_config expose the same +# values the plugin applies, for use without the plugin if needed. +``` diff --git a/temporalio/contrib/gcp/cloud_run/__init__.py b/temporalio/contrib/gcp/cloud_run/__init__.py index 341b19097..f86f79608 100644 --- a/temporalio/contrib/gcp/cloud_run/__init__.py +++ b/temporalio/contrib/gcp/cloud_run/__init__.py @@ -1,9 +1,14 @@ -"""Metadata helpers for running Temporal workers on Google Cloud Run. +"""Run Temporal workers on Google Cloud Run. -Cloud Run runs a long-lived container rather than a per-invocation handler, so this module is a -small metadata helper -- **not** a worker wrapper. :py:func:`get_google_cloud_run_metadata` reads -Cloud Run instance metadata (from a worker pool or a service) and hands you a worker identity string -and a :py:class:`temporalio.worker.WorkerDeploymentConfig` to drop into a normal, long-lived worker. +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. :py:class:`CloudRunPlugin` reads Cloud Run +instance metadata (from a worker pool or a service) and configures a normal, long-lived client and +worker: it sets the client identity from the Cloud Run instance and enables Worker Versioning with a +``PINNED`` deployment version derived from the Cloud Run revision. + +For advanced or non-plugin use, :py:func:`get_google_cloud_run_metadata` returns the underlying +:py:class:`GoogleCloudRunMetadata`, whose ``worker_identity`` and ``worker_deployment_config`` +properties expose the same values the plugin applies. .. warning:: Google Cloud Run support is experimental. @@ -13,15 +18,14 @@ import asyncio from temporalio.client import Client - from temporalio.contrib.gcp.cloud_run import get_google_cloud_run_metadata + from temporalio.contrib.gcp.cloud_run import CloudRunPlugin from temporalio.worker import Worker async def main() -> None: - metadata = get_google_cloud_run_metadata() - + # Install the plugin on the client; it propagates to workers automatically. client = await Client.connect( "localhost:7233", - identity=metadata.worker_identity, + plugins=[CloudRunPlugin()], ) worker = Worker( @@ -29,7 +33,6 @@ async def main() -> None: task_queue="my-task-queue", workflows=[MyWorkflow], activities=[my_activity], - deployment_config=metadata.worker_deployment_config, ) await worker.run() @@ -37,11 +40,15 @@ async def main() -> None: """ from temporalio.contrib.gcp.cloud_run._metadata import ( + CLOUD_RUN_METADATA_URL, GoogleCloudRunMetadata, get_google_cloud_run_metadata, ) +from temporalio.contrib.gcp.cloud_run._plugin import CloudRunPlugin __all__ = [ + "CLOUD_RUN_METADATA_URL", + "CloudRunPlugin", "GoogleCloudRunMetadata", "get_google_cloud_run_metadata", ] diff --git a/temporalio/contrib/gcp/cloud_run/_metadata.py b/temporalio/contrib/gcp/cloud_run/_metadata.py index 784c8ead8..8af19f38a 100644 --- a/temporalio/contrib/gcp/cloud_run/_metadata.py +++ b/temporalio/contrib/gcp/cloud_run/_metadata.py @@ -22,6 +22,11 @@ if TYPE_CHECKING: import temporalio.worker +CLOUD_RUN_METADATA_URL = ( + "http://metadata.google.internal/computeMetadata/v1/instance/id" +) +"""Default Cloud Run metadata server endpoint returning the unique instance id.""" + @dataclass(frozen=True) class GoogleCloudRunMetadata: @@ -104,7 +109,7 @@ def worker_deployment_config(self) -> temporalio.worker.WorkerDeploymentConfig: def get_google_cloud_run_metadata( *, timeout: float = 2.0, - metadata_url: str = "http://metadata.google.internal/computeMetadata/v1/instance/id", + metadata_url: str = CLOUD_RUN_METADATA_URL, getenv: Callable[[str], str] = os.environ.get, # type: ignore[assignment] ) -> GoogleCloudRunMetadata: """Read metadata identifying the current Google Cloud Run instance. diff --git a/temporalio/contrib/gcp/cloud_run/_plugin.py b/temporalio/contrib/gcp/cloud_run/_plugin.py new file mode 100644 index 000000000..35e2955f7 --- /dev/null +++ b/temporalio/contrib/gcp/cloud_run/_plugin.py @@ -0,0 +1,118 @@ +"""Plugin applying Google Cloud Run worker defaults to a Temporal client and worker.""" + +from __future__ import annotations + +import os +import socket +from collections.abc import Awaitable, Callable + +import temporalio.plugin +from temporalio.contrib.gcp.cloud_run._metadata import ( + CLOUD_RUN_METADATA_URL, + GoogleCloudRunMetadata, + get_google_cloud_run_metadata, +) +from temporalio.service import ConnectConfig, ServiceClient +from temporalio.worker import WorkerConfig + + +class CloudRunPlugin(temporalio.plugin.SimplePlugin): + """Configure a Temporal client and worker from Google Cloud Run instance metadata. + + Install this plugin once when connecting the client; it automatically + propagates to workers created from that client. It sets the client + **identity** to a value derived from the Cloud Run instance (unless the caller + already provided one) and configures the worker with a + :py:class:`temporalio.worker.WorkerDeploymentConfig` that enables Worker + Versioning with a ``PINNED`` default behavior, so each Cloud Run revision is a + distinct, pinned worker deployment version. Both Cloud Run worker pools and + services are supported. + + The Cloud Run instance metadata is fetched once, lazily, when the client + connects and then cached on the plugin. If the metadata cannot be read -- which + usually means the process is not running on a Cloud Run worker pool or service + -- connecting fails fast with a clear error rather than silently doing nothing. + + Unit tests and advanced callers can bypass the metadata server by passing a + pre-built ``metadata`` object, or steer the fetch with ``getenv`` / + ``metadata_url`` / ``timeout``. + + .. warning:: + Google Cloud Run support is experimental and may change in future versions. + """ + + def __init__( + self, + *, + metadata: GoogleCloudRunMetadata | None = None, + timeout: float = 2.0, + metadata_url: str = CLOUD_RUN_METADATA_URL, + getenv: Callable[[str], str | None] = os.environ.get, + ) -> None: + """Create a Cloud Run plugin. + + Args: + metadata: Pre-fetched Cloud Run instance metadata. When supplied, the + plugin uses it directly and never contacts the metadata server. + Primarily for testing and advanced use. + timeout: Timeout, in seconds, for the request to the metadata server. + Ignored when ``metadata`` is supplied. + metadata_url: URL of the Cloud Run metadata server endpoint that + returns the instance id. Ignored when ``metadata`` is supplied. + getenv: Callable used to look up environment variables. Defaults to + ``os.environ.get`` and exists primarily for testing. Ignored when + ``metadata`` is supplied. + """ + super().__init__("CloudRunPlugin") + self._metadata = metadata + self._timeout = timeout + self._metadata_url = metadata_url + self._getenv = getenv + + async def connect_service_client( + self, + config: ConnectConfig, + next: Callable[[ConnectConfig], Awaitable[ServiceClient]], + ) -> ServiceClient: + """Fetch Cloud Run metadata and set the client identity before connecting. + + The identity is only set when the caller did not provide one, so an + explicit ``identity`` passed to :py:meth:`temporalio.client.Client.connect` + always wins. + """ + metadata = self._resolve_metadata() + if not config.identity or config.identity == _default_identity(): + config.identity = metadata.worker_identity + return await super().connect_service_client(config, next) + + def configure_worker(self, config: WorkerConfig) -> WorkerConfig: + """Set the worker deployment config from the cached Cloud Run metadata. + + The deployment config enables Worker Versioning with a ``PINNED`` default + behavior, deriving the deployment name and build id from the Cloud Run + workload name and revision. + """ + config = super().configure_worker(config) + config["deployment_config"] = self._resolve_metadata().worker_deployment_config + return config + + def _resolve_metadata(self) -> GoogleCloudRunMetadata: + """Return the cached Cloud Run metadata, fetching it once on first use.""" + if self._metadata is None: + self._metadata = get_google_cloud_run_metadata( + timeout=self._timeout, + metadata_url=self._metadata_url, + getenv=self._getenv, # type: ignore[arg-type] + ) + return self._metadata + + +def _default_identity() -> str: + """Recreate the identity ``ConnectConfig`` auto-generates when none is given. + + :py:class:`temporalio.service.ConnectConfig` fills an unset identity with + ``@`` in ``__post_init__``, so by the time this plugin runs the + identity is never literally empty. Matching that value lets the plugin tell an + auto-generated identity (safe to replace) from one the caller chose (kept). + """ + return f"{os.getpid()}@{socket.gethostname()}" diff --git a/tests/contrib/gcp/cloud_run/test_plugin.py b/tests/contrib/gcp/cloud_run/test_plugin.py new file mode 100644 index 000000000..b87eaf1c9 --- /dev/null +++ b/tests/contrib/gcp/cloud_run/test_plugin.py @@ -0,0 +1,148 @@ +"""Tests for the Google Cloud Run plugin.""" + +from __future__ import annotations + +import os +import socket +from typing import cast +from unittest.mock import Mock + +import pytest + +from temporalio.common import VersioningBehavior, WorkerDeploymentVersion +from temporalio.contrib.gcp.cloud_run import ( + CloudRunPlugin, + GoogleCloudRunMetadata, +) +from temporalio.service import ConnectConfig, ServiceClient +from temporalio.worker import WorkerConfig + + +def _metadata( + *, + instance_id: str = "instance-1", + name: str = "my-pool", + revision: str = "rev-1", +) -> GoogleCloudRunMetadata: + return GoogleCloudRunMetadata( + instance_id=instance_id, + name=name, + revision=revision, + ) + + +def _closed_port() -> int: + """Return a port number that nothing is listening on.""" + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +def _service_client() -> ServiceClient: + return cast(ServiceClient, Mock(spec=ServiceClient)) + + +# ---- Client identity ---- + + +class TestClientIdentity: + @pytest.mark.asyncio + async def test_sets_identity_when_unset(self) -> None: + plugin = CloudRunPlugin(metadata=_metadata(instance_id="abc", revision="rev-1")) + # ConnectConfig auto-fills identity with @ when none is given. + config = ConnectConfig(target_host="localhost:7233") + assert config.identity == f"{os.getpid()}@{socket.gethostname()}" + service_client = _service_client() + + async def connect(input: ConnectConfig) -> ServiceClient: + assert input.identity == "abc@rev-1" + return service_client + + assert await plugin.connect_service_client(config, connect) is service_client + assert config.identity == "abc@rev-1" + + @pytest.mark.asyncio + async def test_preserves_caller_identity(self) -> None: + plugin = CloudRunPlugin(metadata=_metadata(instance_id="abc", revision="rev-1")) + config = ConnectConfig(target_host="localhost:7233", identity="my-identity") + service_client = _service_client() + + async def connect(input: ConnectConfig) -> ServiceClient: + assert input.identity == "my-identity" + return service_client + + assert await plugin.connect_service_client(config, connect) is service_client + assert config.identity == "my-identity" + + +# ---- Worker deployment config ---- + + +class TestConfigureWorker: + def test_sets_pinned_deployment_config(self) -> None: + plugin = CloudRunPlugin( + metadata=_metadata(instance_id="abc", name="my-pool", revision="rev-1") + ) + config = plugin.configure_worker(cast(WorkerConfig, {})) + deployment_config = config.get("deployment_config") + assert deployment_config is not None + assert deployment_config.use_worker_versioning is True + assert ( + deployment_config.default_versioning_behavior == VersioningBehavior.PINNED + ) + assert deployment_config.version == WorkerDeploymentVersion( + deployment_name="my-pool", + build_id="rev-1", + ) + + +# ---- Metadata fetching / caching ---- + + +class TestMetadataFetch: + def test_construction_does_not_fetch(self) -> None: + # A bad metadata URL must not raise at construction -- the fetch is lazy. + CloudRunPlugin( + metadata_url=f"http://127.0.0.1:{_closed_port()}/instance/id", + getenv={}.get, # type: ignore[arg-type] + ) + + @pytest.mark.asyncio + async def test_connect_fails_fast_off_platform(self) -> None: + plugin = CloudRunPlugin( + timeout=1.0, + metadata_url=f"http://127.0.0.1:{_closed_port()}/instance/id", + getenv={}.get, # type: ignore[arg-type] + ) + config = ConnectConfig(target_host="localhost:7233") + + async def connect(input: ConnectConfig) -> ServiceClient: + raise AssertionError("should not connect when metadata is unavailable") + + with pytest.raises(RuntimeError, match="metadata server"): + await plugin.connect_service_client(config, connect) + + @pytest.mark.asyncio + async def test_metadata_fetched_once_and_reused_by_worker( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + fetch = Mock(return_value=_metadata(instance_id="abc", revision="rev-1")) + monkeypatch.setattr( + "temporalio.contrib.gcp.cloud_run._plugin.get_google_cloud_run_metadata", + fetch, + ) + plugin = CloudRunPlugin() + config = ConnectConfig(target_host="localhost:7233") + + async def connect(input: ConnectConfig) -> ServiceClient: + return _service_client() + + await plugin.connect_service_client(config, connect) + worker_config = plugin.configure_worker(cast(WorkerConfig, {})) + + # Fetched exactly once at connect; the worker hook reuses the cached value. + fetch.assert_called_once() + assert config.identity == "abc@rev-1" + assert worker_config.get("deployment_config") is not None From acf7f11d4709ee5964baa8f5060173b9c1c8745a Mon Sep 17 00:00:00 2001 From: seanbollin Date: Mon, 31 Aug 2026 15:39:29 -0700 Subject: [PATCH 5/6] Rename CloudRunPlugin to WorkerIDPlugin and fix test type errors Cloud Run can host multiple plugins (a worker-ID plugin and an OpenTelemetry plugin both live in the same cloud_run area), so the worker-ID plugin needs a specific name rather than the generic CloudRunPlugin. - Rename class CloudRunPlugin -> WorkerIDPlugin and move _plugin.py -> _worker_id_plugin.py (cloud_run package and GoogleCloudRunMetadata unchanged). - Update the package __init__ export/__all__, quick-start, and README. - Rename test_plugin.py -> test_worker_id_plugin.py and fix the basedpyright reportInvalidCast errors by constructing WorkerConfig() instead of cast(WorkerConfig, {}); silence reportUnusedParameter on the unused connect() callbacks. Co-Authored-By: Claude Opus 4.8 --- temporalio/contrib/gcp/cloud_run/README.md | 12 ++++----- temporalio/contrib/gcp/cloud_run/__init__.py | 10 +++---- .../{_plugin.py => _worker_id_plugin.py} | 4 +-- ...est_plugin.py => test_worker_id_plugin.py} | 26 +++++++++---------- 4 files changed, 26 insertions(+), 26 deletions(-) rename temporalio/contrib/gcp/cloud_run/{_plugin.py => _worker_id_plugin.py} (98%) rename tests/contrib/gcp/cloud_run/{test_plugin.py => test_worker_id_plugin.py} (86%) diff --git a/temporalio/contrib/gcp/cloud_run/README.md b/temporalio/contrib/gcp/cloud_run/README.md index 1edc9f67c..b305a4924 100644 --- a/temporalio/contrib/gcp/cloud_run/README.md +++ b/temporalio/contrib/gcp/cloud_run/README.md @@ -4,7 +4,7 @@ A plugin for running [Temporal](https://temporal.io) workers on Google Cloud Run. Cloud Run runs a long-lived container -- there is no per-invocation handler to wrap -- so this is **not** a worker -wrapper. Instead, `CloudRunPlugin` reads Cloud Run instance metadata and configures a normal, +wrapper. Instead, `WorkerIDPlugin` reads Cloud Run instance metadata and configures a normal, long-lived client and worker for you. Both Cloud Run **worker pools** and **services** are supported. Register the plugin once when connecting the client and it: @@ -24,7 +24,7 @@ wire up on the worker. import asyncio from temporalio.client import Client -from temporalio.contrib.gcp.cloud_run import CloudRunPlugin +from temporalio.contrib.gcp.cloud_run import WorkerIDPlugin from temporalio.worker import Worker from my_workflows import MyWorkflow @@ -35,7 +35,7 @@ async def main() -> None: # Install the plugin on the client; it propagates to workers automatically. client = await Client.connect( "localhost:7233", - plugins=[CloudRunPlugin()], + plugins=[WorkerIDPlugin()], ) worker = Worker( @@ -64,7 +64,7 @@ the at `http://metadata.google.internal/computeMetadata/v1/instance/id`, which requires the `Metadata-Flavor: Google` request header. -When the client connects, `CloudRunPlugin` resolves the deployment name from `CLOUD_RUN_WORKER_POOL` +When the client connects, `WorkerIDPlugin` resolves the deployment name from `CLOUD_RUN_WORKER_POOL` (falling back to `K_SERVICE`) and the revision from `CLOUD_RUN_REVISION` (falling back to `K_REVISION`), then performs a single synchronous HTTP GET to the metadata server for the instance id. The result is **cached on the plugin**, so the worker hook reuses it without another network @@ -88,10 +88,10 @@ For advanced scenarios or unit tests you can bypass the metadata server by passi metadata object, or steer the fetch with `getenv` / `metadata_url` / `timeout`: ```python -from temporalio.contrib.gcp.cloud_run import CloudRunPlugin, get_google_cloud_run_metadata +from temporalio.contrib.gcp.cloud_run import WorkerIDPlugin, get_google_cloud_run_metadata metadata = get_google_cloud_run_metadata() -plugin = CloudRunPlugin(metadata=metadata) +plugin = WorkerIDPlugin(metadata=metadata) # metadata.worker_identity and metadata.worker_deployment_config expose the same # values the plugin applies, for use without the plugin if needed. diff --git a/temporalio/contrib/gcp/cloud_run/__init__.py b/temporalio/contrib/gcp/cloud_run/__init__.py index f86f79608..cfecb1f8a 100644 --- a/temporalio/contrib/gcp/cloud_run/__init__.py +++ b/temporalio/contrib/gcp/cloud_run/__init__.py @@ -1,7 +1,7 @@ """Run Temporal workers on Google Cloud Run. 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. :py:class:`CloudRunPlugin` reads Cloud Run +metadata-driven plugin -- **not** a worker wrapper. :py:class:`WorkerIDPlugin` reads Cloud Run instance metadata (from a worker pool or a service) and configures a normal, long-lived client and worker: it sets the client identity from the Cloud Run instance and enables Worker Versioning with a ``PINNED`` deployment version derived from the Cloud Run revision. @@ -18,14 +18,14 @@ import asyncio from temporalio.client import Client - from temporalio.contrib.gcp.cloud_run import CloudRunPlugin + from temporalio.contrib.gcp.cloud_run import WorkerIDPlugin from temporalio.worker import Worker async def main() -> None: # Install the plugin on the client; it propagates to workers automatically. client = await Client.connect( "localhost:7233", - plugins=[CloudRunPlugin()], + plugins=[WorkerIDPlugin()], ) worker = Worker( @@ -44,11 +44,11 @@ async def main() -> None: GoogleCloudRunMetadata, get_google_cloud_run_metadata, ) -from temporalio.contrib.gcp.cloud_run._plugin import CloudRunPlugin +from temporalio.contrib.gcp.cloud_run._worker_id_plugin import WorkerIDPlugin __all__ = [ "CLOUD_RUN_METADATA_URL", - "CloudRunPlugin", "GoogleCloudRunMetadata", + "WorkerIDPlugin", "get_google_cloud_run_metadata", ] diff --git a/temporalio/contrib/gcp/cloud_run/_plugin.py b/temporalio/contrib/gcp/cloud_run/_worker_id_plugin.py similarity index 98% rename from temporalio/contrib/gcp/cloud_run/_plugin.py rename to temporalio/contrib/gcp/cloud_run/_worker_id_plugin.py index 35e2955f7..adc979610 100644 --- a/temporalio/contrib/gcp/cloud_run/_plugin.py +++ b/temporalio/contrib/gcp/cloud_run/_worker_id_plugin.py @@ -16,7 +16,7 @@ from temporalio.worker import WorkerConfig -class CloudRunPlugin(temporalio.plugin.SimplePlugin): +class WorkerIDPlugin(temporalio.plugin.SimplePlugin): """Configure a Temporal client and worker from Google Cloud Run instance metadata. Install this plugin once when connecting the client; it automatically @@ -63,7 +63,7 @@ def __init__( ``os.environ.get`` and exists primarily for testing. Ignored when ``metadata`` is supplied. """ - super().__init__("CloudRunPlugin") + super().__init__("WorkerIDPlugin") self._metadata = metadata self._timeout = timeout self._metadata_url = metadata_url diff --git a/tests/contrib/gcp/cloud_run/test_plugin.py b/tests/contrib/gcp/cloud_run/test_worker_id_plugin.py similarity index 86% rename from tests/contrib/gcp/cloud_run/test_plugin.py rename to tests/contrib/gcp/cloud_run/test_worker_id_plugin.py index b87eaf1c9..9243fe56b 100644 --- a/tests/contrib/gcp/cloud_run/test_plugin.py +++ b/tests/contrib/gcp/cloud_run/test_worker_id_plugin.py @@ -1,4 +1,4 @@ -"""Tests for the Google Cloud Run plugin.""" +"""Tests for the Google Cloud Run worker-ID plugin.""" from __future__ import annotations @@ -11,8 +11,8 @@ from temporalio.common import VersioningBehavior, WorkerDeploymentVersion from temporalio.contrib.gcp.cloud_run import ( - CloudRunPlugin, GoogleCloudRunMetadata, + WorkerIDPlugin, ) from temporalio.service import ConnectConfig, ServiceClient from temporalio.worker import WorkerConfig @@ -50,7 +50,7 @@ def _service_client() -> ServiceClient: class TestClientIdentity: @pytest.mark.asyncio async def test_sets_identity_when_unset(self) -> None: - plugin = CloudRunPlugin(metadata=_metadata(instance_id="abc", revision="rev-1")) + plugin = WorkerIDPlugin(metadata=_metadata(instance_id="abc", revision="rev-1")) # ConnectConfig auto-fills identity with @ when none is given. config = ConnectConfig(target_host="localhost:7233") assert config.identity == f"{os.getpid()}@{socket.gethostname()}" @@ -65,7 +65,7 @@ async def connect(input: ConnectConfig) -> ServiceClient: @pytest.mark.asyncio async def test_preserves_caller_identity(self) -> None: - plugin = CloudRunPlugin(metadata=_metadata(instance_id="abc", revision="rev-1")) + plugin = WorkerIDPlugin(metadata=_metadata(instance_id="abc", revision="rev-1")) config = ConnectConfig(target_host="localhost:7233", identity="my-identity") service_client = _service_client() @@ -82,10 +82,10 @@ async def connect(input: ConnectConfig) -> ServiceClient: class TestConfigureWorker: def test_sets_pinned_deployment_config(self) -> None: - plugin = CloudRunPlugin( + plugin = WorkerIDPlugin( metadata=_metadata(instance_id="abc", name="my-pool", revision="rev-1") ) - config = plugin.configure_worker(cast(WorkerConfig, {})) + config = plugin.configure_worker(WorkerConfig()) deployment_config = config.get("deployment_config") assert deployment_config is not None assert deployment_config.use_worker_versioning is True @@ -104,21 +104,21 @@ def test_sets_pinned_deployment_config(self) -> None: class TestMetadataFetch: def test_construction_does_not_fetch(self) -> None: # A bad metadata URL must not raise at construction -- the fetch is lazy. - CloudRunPlugin( + WorkerIDPlugin( metadata_url=f"http://127.0.0.1:{_closed_port()}/instance/id", getenv={}.get, # type: ignore[arg-type] ) @pytest.mark.asyncio async def test_connect_fails_fast_off_platform(self) -> None: - plugin = CloudRunPlugin( + plugin = WorkerIDPlugin( timeout=1.0, metadata_url=f"http://127.0.0.1:{_closed_port()}/instance/id", getenv={}.get, # type: ignore[arg-type] ) config = ConnectConfig(target_host="localhost:7233") - async def connect(input: ConnectConfig) -> ServiceClient: + async def connect(_input: ConnectConfig) -> ServiceClient: raise AssertionError("should not connect when metadata is unavailable") with pytest.raises(RuntimeError, match="metadata server"): @@ -130,17 +130,17 @@ async def test_metadata_fetched_once_and_reused_by_worker( ) -> None: fetch = Mock(return_value=_metadata(instance_id="abc", revision="rev-1")) monkeypatch.setattr( - "temporalio.contrib.gcp.cloud_run._plugin.get_google_cloud_run_metadata", + "temporalio.contrib.gcp.cloud_run._worker_id_plugin.get_google_cloud_run_metadata", fetch, ) - plugin = CloudRunPlugin() + plugin = WorkerIDPlugin() config = ConnectConfig(target_host="localhost:7233") - async def connect(input: ConnectConfig) -> ServiceClient: + async def connect(_input: ConnectConfig) -> ServiceClient: return _service_client() await plugin.connect_service_client(config, connect) - worker_config = plugin.configure_worker(cast(WorkerConfig, {})) + worker_config = plugin.configure_worker(WorkerConfig()) # Fetched exactly once at connect; the worker hook reuses the cached value. fetch.assert_called_once() From 7b57c502d5bbb1c28b2883ac27ad6d452a7e98f5 Mon Sep 17 00:00:00 2001 From: seanbollin Date: Mon, 31 Aug 2026 16:09:15 -0700 Subject: [PATCH 6/6] Move Cloud Run worker-ID plugin under cloud_run/worker_id/ Relocate the Google Cloud Run worker-ID plugin from directly inside temporalio/contrib/gcp/cloud_run/ into a new worker_id/ sub-package so it owns its own __init__.py and README.md. This avoids a hard collision with the separate OTel Cloud Run plugin, which also owns cloud_run/README.md and cloud_run/__init__.py; after the move the two plugins share only the minimal cloud_run/ and gcp/ namespace-marker __init__.py files. The public names are unchanged; only the import path gains .worker_id: from temporalio.contrib.gcp.cloud_run.worker_id import WorkerIDPlugin cloud_run/__init__.py is reduced to a minimal namespace-marker docstring with no worker-ID exports. Mirrors the Go (contrib/gcp/cloudrun/workerid) and .NET (CloudRun.WorkerId) layouts. Co-Authored-By: Claude Opus 4.8 --- temporalio/contrib/gcp/cloud_run/__init__.py | 55 +------------------ .../gcp/cloud_run/{ => worker_id}/README.md | 6 +- .../gcp/cloud_run/worker_id/__init__.py | 54 ++++++++++++++++++ .../cloud_run/{ => worker_id}/_metadata.py | 0 .../{ => worker_id}/_worker_id_plugin.py | 2 +- .../gcp/cloud_run/worker_id/__init__.py | 0 .../{ => worker_id}/test_metadata.py | 4 +- .../{ => worker_id}/test_worker_id_plugin.py | 4 +- 8 files changed, 63 insertions(+), 62 deletions(-) rename temporalio/contrib/gcp/cloud_run/{ => worker_id}/README.md (95%) create mode 100644 temporalio/contrib/gcp/cloud_run/worker_id/__init__.py rename temporalio/contrib/gcp/cloud_run/{ => worker_id}/_metadata.py (100%) rename temporalio/contrib/gcp/cloud_run/{ => worker_id}/_worker_id_plugin.py (98%) create mode 100644 tests/contrib/gcp/cloud_run/worker_id/__init__.py rename tests/contrib/gcp/cloud_run/{ => worker_id}/test_metadata.py (98%) rename tests/contrib/gcp/cloud_run/{ => worker_id}/test_worker_id_plugin.py (96%) diff --git a/temporalio/contrib/gcp/cloud_run/__init__.py b/temporalio/contrib/gcp/cloud_run/__init__.py index cfecb1f8a..b0f134efb 100644 --- a/temporalio/contrib/gcp/cloud_run/__init__.py +++ b/temporalio/contrib/gcp/cloud_run/__init__.py @@ -1,54 +1 @@ -"""Run Temporal workers on Google Cloud Run. - -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. :py:class:`WorkerIDPlugin` reads Cloud Run -instance metadata (from a worker pool or a service) and configures a normal, long-lived client and -worker: it sets the client identity from the Cloud Run instance and enables Worker Versioning with a -``PINNED`` deployment version derived from the Cloud Run revision. - -For advanced or non-plugin use, :py:func:`get_google_cloud_run_metadata` returns the underlying -:py:class:`GoogleCloudRunMetadata`, whose ``worker_identity`` and ``worker_deployment_config`` -properties expose the same values the plugin applies. - -.. warning:: - Google Cloud Run support is experimental. - -Quick start:: - - import asyncio - - from temporalio.client import Client - from temporalio.contrib.gcp.cloud_run import WorkerIDPlugin - from temporalio.worker import Worker - - async def main() -> None: - # Install the plugin on the client; it propagates to workers automatically. - client = await Client.connect( - "localhost:7233", - plugins=[WorkerIDPlugin()], - ) - - worker = Worker( - client, - task_queue="my-task-queue", - workflows=[MyWorkflow], - activities=[my_activity], - ) - await worker.run() - - asyncio.run(main()) -""" - -from temporalio.contrib.gcp.cloud_run._metadata import ( - CLOUD_RUN_METADATA_URL, - GoogleCloudRunMetadata, - get_google_cloud_run_metadata, -) -from temporalio.contrib.gcp.cloud_run._worker_id_plugin import WorkerIDPlugin - -__all__ = [ - "CLOUD_RUN_METADATA_URL", - "GoogleCloudRunMetadata", - "WorkerIDPlugin", - "get_google_cloud_run_metadata", -] +"""Google Cloud Run integrations for Temporal (see the worker_id sub-package).""" diff --git a/temporalio/contrib/gcp/cloud_run/README.md b/temporalio/contrib/gcp/cloud_run/worker_id/README.md similarity index 95% rename from temporalio/contrib/gcp/cloud_run/README.md rename to temporalio/contrib/gcp/cloud_run/worker_id/README.md index b305a4924..13f5184c7 100644 --- a/temporalio/contrib/gcp/cloud_run/README.md +++ b/temporalio/contrib/gcp/cloud_run/worker_id/README.md @@ -1,4 +1,4 @@ -# cloud_run +# worker_id > ⚠️ **This package is currently at an experimental release stage.** ⚠️ @@ -24,7 +24,7 @@ wire up on the worker. import asyncio from temporalio.client import Client -from temporalio.contrib.gcp.cloud_run import WorkerIDPlugin +from temporalio.contrib.gcp.cloud_run.worker_id import WorkerIDPlugin from temporalio.worker import Worker from my_workflows import MyWorkflow @@ -88,7 +88,7 @@ For advanced scenarios or unit tests you can bypass the metadata server by passi metadata object, or steer the fetch with `getenv` / `metadata_url` / `timeout`: ```python -from temporalio.contrib.gcp.cloud_run import WorkerIDPlugin, get_google_cloud_run_metadata +from temporalio.contrib.gcp.cloud_run.worker_id import WorkerIDPlugin, get_google_cloud_run_metadata metadata = get_google_cloud_run_metadata() plugin = WorkerIDPlugin(metadata=metadata) diff --git a/temporalio/contrib/gcp/cloud_run/worker_id/__init__.py b/temporalio/contrib/gcp/cloud_run/worker_id/__init__.py new file mode 100644 index 000000000..fcb1a0013 --- /dev/null +++ b/temporalio/contrib/gcp/cloud_run/worker_id/__init__.py @@ -0,0 +1,54 @@ +"""Run Temporal workers on Google Cloud Run. + +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. :py:class:`WorkerIDPlugin` reads Cloud Run +instance metadata (from a worker pool or a service) and configures a normal, long-lived client and +worker: it sets the client identity from the Cloud Run instance and enables Worker Versioning with a +``PINNED`` deployment version derived from the Cloud Run revision. + +For advanced or non-plugin use, :py:func:`get_google_cloud_run_metadata` returns the underlying +:py:class:`GoogleCloudRunMetadata`, whose ``worker_identity`` and ``worker_deployment_config`` +properties expose the same values the plugin applies. + +.. warning:: + Google Cloud Run support is experimental. + +Quick start:: + + import asyncio + + from temporalio.client import Client + from temporalio.contrib.gcp.cloud_run.worker_id import WorkerIDPlugin + from temporalio.worker import Worker + + async def main() -> None: + # Install the plugin on the client; it propagates to workers automatically. + client = await Client.connect( + "localhost:7233", + plugins=[WorkerIDPlugin()], + ) + + worker = Worker( + client, + task_queue="my-task-queue", + workflows=[MyWorkflow], + activities=[my_activity], + ) + await worker.run() + + asyncio.run(main()) +""" + +from temporalio.contrib.gcp.cloud_run.worker_id._metadata import ( + CLOUD_RUN_METADATA_URL, + GoogleCloudRunMetadata, + get_google_cloud_run_metadata, +) +from temporalio.contrib.gcp.cloud_run.worker_id._worker_id_plugin import WorkerIDPlugin + +__all__ = [ + "CLOUD_RUN_METADATA_URL", + "GoogleCloudRunMetadata", + "WorkerIDPlugin", + "get_google_cloud_run_metadata", +] diff --git a/temporalio/contrib/gcp/cloud_run/_metadata.py b/temporalio/contrib/gcp/cloud_run/worker_id/_metadata.py similarity index 100% rename from temporalio/contrib/gcp/cloud_run/_metadata.py rename to temporalio/contrib/gcp/cloud_run/worker_id/_metadata.py diff --git a/temporalio/contrib/gcp/cloud_run/_worker_id_plugin.py b/temporalio/contrib/gcp/cloud_run/worker_id/_worker_id_plugin.py similarity index 98% rename from temporalio/contrib/gcp/cloud_run/_worker_id_plugin.py rename to temporalio/contrib/gcp/cloud_run/worker_id/_worker_id_plugin.py index adc979610..70b0507dd 100644 --- a/temporalio/contrib/gcp/cloud_run/_worker_id_plugin.py +++ b/temporalio/contrib/gcp/cloud_run/worker_id/_worker_id_plugin.py @@ -7,7 +7,7 @@ from collections.abc import Awaitable, Callable import temporalio.plugin -from temporalio.contrib.gcp.cloud_run._metadata import ( +from temporalio.contrib.gcp.cloud_run.worker_id._metadata import ( CLOUD_RUN_METADATA_URL, GoogleCloudRunMetadata, get_google_cloud_run_metadata, diff --git a/tests/contrib/gcp/cloud_run/worker_id/__init__.py b/tests/contrib/gcp/cloud_run/worker_id/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/contrib/gcp/cloud_run/test_metadata.py b/tests/contrib/gcp/cloud_run/worker_id/test_metadata.py similarity index 98% rename from tests/contrib/gcp/cloud_run/test_metadata.py rename to tests/contrib/gcp/cloud_run/worker_id/test_metadata.py index 11c1ba763..8179915a4 100644 --- a/tests/contrib/gcp/cloud_run/test_metadata.py +++ b/tests/contrib/gcp/cloud_run/worker_id/test_metadata.py @@ -1,4 +1,4 @@ -"""Tests for temporalio.contrib.gcp.cloud_run.""" +"""Tests for temporalio.contrib.gcp.cloud_run.worker_id.""" from __future__ import annotations @@ -12,7 +12,7 @@ import pytest from temporalio.common import VersioningBehavior, WorkerDeploymentVersion -from temporalio.contrib.gcp.cloud_run import ( +from temporalio.contrib.gcp.cloud_run.worker_id import ( GoogleCloudRunMetadata, get_google_cloud_run_metadata, ) diff --git a/tests/contrib/gcp/cloud_run/test_worker_id_plugin.py b/tests/contrib/gcp/cloud_run/worker_id/test_worker_id_plugin.py similarity index 96% rename from tests/contrib/gcp/cloud_run/test_worker_id_plugin.py rename to tests/contrib/gcp/cloud_run/worker_id/test_worker_id_plugin.py index 9243fe56b..4f3d46bfb 100644 --- a/tests/contrib/gcp/cloud_run/test_worker_id_plugin.py +++ b/tests/contrib/gcp/cloud_run/worker_id/test_worker_id_plugin.py @@ -10,7 +10,7 @@ import pytest from temporalio.common import VersioningBehavior, WorkerDeploymentVersion -from temporalio.contrib.gcp.cloud_run import ( +from temporalio.contrib.gcp.cloud_run.worker_id import ( GoogleCloudRunMetadata, WorkerIDPlugin, ) @@ -130,7 +130,7 @@ async def test_metadata_fetched_once_and_reused_by_worker( ) -> None: fetch = Mock(return_value=_metadata(instance_id="abc", revision="rev-1")) monkeypatch.setattr( - "temporalio.contrib.gcp.cloud_run._worker_id_plugin.get_google_cloud_run_metadata", + "temporalio.contrib.gcp.cloud_run.worker_id._worker_id_plugin.get_google_cloud_run_metadata", fetch, ) plugin = WorkerIDPlugin()