Add Google Cloud Run worker identity plugin

CHANGELOG.md

@@ -20,6 +20,12 @@ to include examples, links to docs, or any other relevant information.
 
 ### Added
 
+- Added the experimental `temporalio.contrib.gcp.cloud_run.worker_id` module for long-lived Temporal
+  workers on Google Cloud Run worker pools and services. Register `WorkerIDPlugin` on your client
+  (`plugins=[WorkerIDPlugin()]`); it propagates to workers automatically, setting the client identity
+  from the Cloud Run instance (unless one is already configured). The underlying
+  `GoogleCloudRunMetadata` helper (via `get_google_cloud_run_metadata`) exposes the same value for
+  advanced use.
 - Added GCP Cloud Run serverless-worker OpenTelemetry plugin in `temporalio.contrib.opentelemetry`.
 - Added new options to ActivityHandle.describe() to retrieve associated payloads, such as activity input and outcome.
 - New properties and methods in ActivityExecution and ActivityExecutionDescription.

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

@@ -0,0 +1,90 @@
+# 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 sets the client identity
+for a normal, long-lived client and worker. 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`).
+
+Client plugins automatically propagate to workers created from that client, so the worker inherits
+this identity and 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 worker pool name from
+`CLOUD_RUN_WORKER_POOL` (falling back to the service name `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. From that metadata the plugin sets:
+
+- **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.
+
+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 exposes the same value the plugin applies, for use
+# without the plugin if needed.
+```

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

@@ -0,0 +1,53 @@
+"""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 sets the client identity from the Cloud Run
+instance for a normal, long-lived client and worker.
+
+For advanced or non-plugin use, :py:func:`get_google_cloud_run_metadata` returns the underlying
+:py:class:`GoogleCloudRunMetadata`, whose ``worker_identity`` property exposes the same value 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,109 @@
+"""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 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
+
+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: The Cloud Run 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 worker pool or
+        service name is used instead (``<instance_id>@<name>``); 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
+
+
+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 worker pool name from ``CLOUD_RUN_WORKER_POOL`` (Cloud Run worker pools), falling
+    back to the service name (``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,
+    )

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

@@ -0,0 +1,103 @@
+"""Plugin setting a Temporal client's identity from Google Cloud Run instance metadata."""
+
+from __future__ import annotations
+
+import os
+import socket
+from collections.abc import Awaitable, Callable
+
+import temporalio.plugin
+from temporalio.contrib.gcp.cloud_run.worker_id._metadata import (
+    CLOUD_RUN_METADATA_URL,
+    GoogleCloudRunMetadata,
+    get_google_cloud_run_metadata,
+)
+from temporalio.service import ConnectConfig, ServiceClient
+
+
+class WorkerIDPlugin(temporalio.plugin.SimplePlugin):
+    """Set a Temporal client's identity from Google Cloud Run instance metadata.
+
+    Install this plugin once when connecting the client; the identity it sets
+    automatically propagates to workers created from that client. It sets the
+    client **identity** to a value derived from the Cloud Run instance, but only
+    when the caller did not already provide one. Both Cloud Run worker pools and
+    services are supported.
+
+    The Cloud Run instance metadata is fetched once, lazily, when the client
+    connects. 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__("WorkerIDPlugin")
+        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 _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
+    ``<pid>@<hostname>`` 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()}"