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..041f5121e --- /dev/null +++ b/temporalio/contrib/gcp/cloud_run/README.md @@ -0,0 +1,74 @@ +# 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` 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/__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..784c8ead8 --- /dev/null +++ b/temporalio/contrib/gcp/cloud_run/_metadata.py @@ -0,0 +1,151 @@ +"""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, + default_versioning_behavior=temporalio.common.VersioningBehavior.PINNED, + ) + + +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, + ) 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", + )