Add Google Cloud Run worker identity/deployment helper

temporalio/contrib/gcp/__init__.py

@@ -0,0 +1 @@
+"""Google Cloud integrations for Temporal SDK."""

temporalio/contrib/gcp/cloud_run/__init__.py

@@ -0,0 +1 @@
+"""Google Cloud Run integrations for Temporal (see the worker_id sub-package)."""

temporalio/contrib/gcp/cloud_run/worker_id/README.md

@@ -0,0 +1,98 @@
+# worker_id
+
+> ⚠️ **This package is currently at an experimental release stage.** ⚠️
+
+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, `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:
+
+- 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
+
+```python
+import asyncio
+
+from temporalio.client import Client
+from temporalio.contrib.gcp.cloud_run.worker_id import WorkerIDPlugin
+from temporalio.worker import Worker
+
+from my_workflows import MyWorkflow
+from my_activities import my_activity
+
+
+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()
+
+
+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.
+
+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
+call. From that metadata the plugin applies:
+
+- **Client identity** -- `<instance_id>@<revision>`, uniquely identifying this worker instance in
+  Temporal tooling. It falls back to `<instance_id>@<name>`, then to just `<instance_id>`, when the
+  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).
+
+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
+
+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.worker_id import WorkerIDPlugin, get_google_cloud_run_metadata
+
+metadata = get_google_cloud_run_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.
+```

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",
+]

temporalio/contrib/gcp/cloud_run/worker_id/_metadata.py

@@ -0,0 +1,156 @@
+"""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
+
+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:
+    """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 ``<instance_id>@<revision>``. When the revision is empty the deployment name
+        is used instead (``<instance_id>@<name>``), 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 = CLOUD_RUN_METADATA_URL,
+    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,
+    )