Add Cloud Run worker sample (CloudRunId + OpenTelemetry plugins)

gcp/cloud_run/.dockerignore

@@ -0,0 +1,7 @@
+*
+!Dockerfile
+!pyproject.toml
+!__init__.py
+!settings.py
+!worker.py
+!workflow.py

gcp/cloud_run/Dockerfile

@@ -0,0 +1,48 @@
+# syntax=docker/dockerfile:1
+
+# The git-pinned temporalio compiles the Rust core; drop this builder for a Python base once the plugin ships on PyPI.
+FROM rust:1.91.0-slim-bookworm AS builder
+
+COPY --from=ghcr.io/astral-sh/uv:0.8.15 /uv /uvx /bin/
+
+RUN apt-get update \
+    && apt-get install --no-install-recommends --yes \
+        build-essential \
+        ca-certificates \
+        git \
+        libprotobuf-dev \
+        pkg-config \
+        protobuf-compiler \
+        python3 \
+        python3-dev \
+    && rm -rf /var/lib/apt/lists/*
+
+ENV UV_COMPILE_BYTECODE=1 \
+    UV_LINK_MODE=copy \
+    UV_PYTHON=/usr/bin/python3
+
+WORKDIR /app
+COPY pyproject.toml ./
+RUN uv sync --no-dev
+
+FROM debian:bookworm-slim
+
+ENV PATH=/app/.venv/bin:$PATH \
+    PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+RUN apt-get update \
+    && apt-get install --no-install-recommends --yes ca-certificates python3 \
+    && rm -rf /var/lib/apt/lists/* \
+    && groupadd --system app \
+    && useradd --system --gid app --create-home app
+
+WORKDIR /app
+COPY --from=builder /app/.venv /app/.venv
+# The build context is this sample directory (flat); recreate the package path.
+COPY --chown=app:app __init__.py settings.py worker.py workflow.py /app/gcp/cloud_run/
+RUN touch /app/gcp/__init__.py \
+    && chown -R app:app /app/gcp
+
+USER app
+CMD ["python", "-m", "gcp.cloud_run.worker"]

gcp/cloud_run/README.md

@@ -0,0 +1,68 @@
+# Google Cloud Run Worker
+
+Run a Temporal Worker on a [Google Cloud Run worker
+pool](https://cloud.google.com/run/docs/worker-pools) with two GCP plugins on
+one client. `temporalio.contrib.gcp.cloud_run.opentelemetry.OpenTelemetryPlugin`
+exports Temporal Core metrics and traces over OTLP/gRPC to a Google-Built
+OpenTelemetry Collector sidecar, which forwards traces to the Cloud Telemetry API
+and metrics to Google Managed Service for Prometheus; endpoint, service name
+(from `CLOUD_RUN_WORKER_POOL`), tracer provider, and 60s metric export are plugin
+defaults, and `worker.py` opts into `add_temporal_spans=True`.
+`temporalio.contrib.gcp.cloud_run.id.CloudRunIdPlugin` sets the client identity
+to `<instance_id>@<revision>` from Cloud Run instance metadata, so each container
+is individually identifiable as a poller.
+
+`CloudRunIdPlugin` is unreleased, so `pyproject.toml` pins `temporalio` to the
+SDK commit that adds it (a git source, which also builds inside the container);
+drop the `[tool.uv.sources]` override once it ships on PyPI.
+
+Prerequisites: a Temporal Cloud namespace and API key; a Google Cloud project
+with billing and an authenticated `gcloud` CLI; `envsubst` (`gettext` package).
+Worker pools bill continuously, so scale to zero after testing (see below).
+
+## Deploy
+
+Run from the repository root. Set your own values (also used by the verify step
+below), then run the deploy script:
+
+```bash
+export PROJECT_ID=your-project-id REGION=us-central1
+export REPOSITORY=temporal-workers WORKER_POOL=temporal-gcp-cloud-run
+export SERVICE_ACCOUNT_EMAIL="cloud-run-worker@${PROJECT_ID}.iam.gserviceaccount.com"
+export TEMPORAL_NAMESPACE=your-namespace.account-id
+export TEMPORAL_ADDRESS="${TEMPORAL_NAMESPACE}.tmprl.cloud:7233"
+export TEMPORAL_TASK_QUEUE=gcp-cloud-run
+export TEMPORAL_API_KEY_FILE=/secure/path/to/temporal-api-key
+export TEMPORAL_API_KEY_SECRET=temporal-api-key TEMPORAL_API_KEY_SECRET_VERSION=1
+export COLLECTOR_CONFIG_SECRET=temporal-otel-collector COLLECTOR_CONFIG_SECRET_VERSION=1
+export WORKER_IMAGE="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPOSITORY}/gcp-cloud-run:v1"
+export INSTANCE_COUNT=1
+
+./gcp/cloud_run/deploy.sh
+```
+
+`deploy.sh` enables the required APIs; creates the Artifact Registry repo, the
+runtime service account, and the API key and collector-config secrets; grants the
+telemetry and secret-access roles; then builds the image and deploys the
+two-container worker pool. The create steps are one-time and re-run safely.
+
+Worker pools bill continuously, so scale to zero when done testing:
+
+```bash
+gcloud run worker-pools update "$WORKER_POOL" --instances 0 \
+  --region "$REGION" --project "$PROJECT_ID"
+```
+
+## Run a Workflow and verify
+
+```bash
+TEMPORAL_API_KEY="$(cat "$TEMPORAL_API_KEY_FILE")" \
+  uv run python -m gcp.cloud_run.starter
+```
+
+The starter prints `Hello, Temporal!`. In Google Cloud, Trace Explorer shows
+`RunWorkflow:GreetingWorkflow` (with `service.name` = the worker-pool name) and
+Metrics Explorer shows
+`prometheus.googleapis.com/temporal_workflow_completed_total/counter`. In the
+Temporal UI, the task queue's pollers report the `<instance_id>@<revision>`
+identity set by `CloudRunIdPlugin`.

gcp/cloud_run/__init__.py

@@ -0,0 +1 @@
+"""Google Cloud Run worker-pool sample."""

gcp/cloud_run/collector-config.yaml

@@ -0,0 +1,70 @@
+# @@@SNIPSTART python-cloud-run-collector-config
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: localhost:4317
+
+processors:
+  batch/traces:
+    send_batch_max_size: 200
+    send_batch_size: 200
+    timeout: 5s
+  memory_limiter:
+    check_interval: 1s
+    limit_percentage: 65
+    spike_limit_percentage: 20
+  resource_detection:
+    detectors: [gcp]
+    timeout: 10s
+  # Rename Temporal datapoint labels that collide with the target labels Managed Service for Prometheus injects (e.g. namespace).
+  transform/collision:
+    metric_statements:
+      - context: datapoint
+        statements:
+          - set(attributes["exported_location"], attributes["location"])
+          - delete_key(attributes, "location")
+          - set(attributes["exported_cluster"], attributes["cluster"])
+          - delete_key(attributes, "cluster")
+          - set(attributes["exported_namespace"], attributes["namespace"])
+          - delete_key(attributes, "namespace")
+          - set(attributes["exported_job"], attributes["job"])
+          - delete_key(attributes, "job")
+          - set(attributes["exported_instance"], attributes["instance"])
+          - delete_key(attributes, "instance")
+          - set(attributes["exported_project_id"], attributes["project_id"])
+          - delete_key(attributes, "project_id")
+  # The Telemetry API expects the Google Cloud project in gcp.project_id.
+  transform/set_project_id:
+    error_mode: ignore
+    trace_statements:
+      - set(resource.attributes["gcp.project_id"], resource.attributes["gcp.project.id"]) where resource.attributes["gcp.project.id"] != nil
+      - set(resource.attributes["gcp.project_id"], resource.attributes["cloud.account.id"]) where resource.attributes["gcp.project_id"] == nil and resource.attributes["cloud.account.id"] != nil
+
+exporters:
+  googlemanagedprometheus:
+  otlp_grpc:
+    endpoint: telemetry.googleapis.com:443
+    compression: none
+    balancer_name: pick_first
+    auth:
+      authenticator: googleclientauth
+
+extensions:
+  googleclientauth:
+  health_check:
+    endpoint: 0.0.0.0:13133
+
+service:
+  extensions: [googleclientauth, health_check]
+  pipelines:
+    metrics:
+      receivers: [otlp]
+      processors: [memory_limiter, resource_detection, transform/collision]
+      exporters: [googlemanagedprometheus]
+    traces:
+      receivers: [otlp]
+      processors:
+        [memory_limiter, resource_detection, transform/set_project_id, batch/traces]
+      exporters: [otlp_grpc]
+# @@@SNIPEND

gcp/cloud_run/deploy.sh

@@ -0,0 +1,50 @@
+#!/usr/bin/env bash
+# Deploy the Temporal Cloud Run worker sample. Export the variables documented in
+# README.md ("Deploy") first. The create steps are one-time and skipped when the
+# resource already exists, so this is safe to re-run to redeploy.
+set -euo pipefail
+
+: "${PROJECT_ID:?set PROJECT_ID}"
+: "${REGION:?set REGION}"
+: "${REPOSITORY:?set REPOSITORY}"
+: "${SERVICE_ACCOUNT_EMAIL:?set SERVICE_ACCOUNT_EMAIL}"
+: "${TEMPORAL_API_KEY_SECRET:?set TEMPORAL_API_KEY_SECRET}"
+: "${TEMPORAL_API_KEY_FILE:?set TEMPORAL_API_KEY_FILE}"
+: "${COLLECTOR_CONFIG_SECRET:?set COLLECTOR_CONFIG_SECRET}"
+: "${WORKER_IMAGE:?set WORKER_IMAGE}"
+
+# Enable APIs, create the Artifact Registry repo and runtime service account.
+gcloud services enable artifactregistry.googleapis.com cloudbuild.googleapis.com \
+  monitoring.googleapis.com run.googleapis.com secretmanager.googleapis.com \
+  telemetry.googleapis.com --project "$PROJECT_ID"
+gcloud artifacts repositories describe "$REPOSITORY" --location "$REGION" --project "$PROJECT_ID" >/dev/null 2>&1 ||
+  gcloud artifacts repositories create "$REPOSITORY" --location "$REGION" \
+    --repository-format docker --project "$PROJECT_ID"
+gcloud iam service-accounts describe "$SERVICE_ACCOUNT_EMAIL" --project "$PROJECT_ID" >/dev/null 2>&1 ||
+  gcloud iam service-accounts create "${SERVICE_ACCOUNT_EMAIL%%@*}" --project "$PROJECT_ID"
+
+# Grant the service account the collector's telemetry roles.
+for role in roles/logging.logWriter roles/monitoring.metricWriter roles/telemetry.tracesWriter; do
+  gcloud projects add-iam-policy-binding "$PROJECT_ID" \
+    --member "serviceAccount:${SERVICE_ACCOUNT_EMAIL}" --role "$role"
+done
+
+# Store the API key and collector config as secrets the service account can read.
+gcloud secrets describe "$TEMPORAL_API_KEY_SECRET" --project "$PROJECT_ID" >/dev/null 2>&1 ||
+  gcloud secrets create "$TEMPORAL_API_KEY_SECRET" --data-file "$TEMPORAL_API_KEY_FILE" --project "$PROJECT_ID"
+gcloud secrets describe "$COLLECTOR_CONFIG_SECRET" --project "$PROJECT_ID" >/dev/null 2>&1 ||
+  gcloud secrets create "$COLLECTOR_CONFIG_SECRET" \
+    --data-file gcp/cloud_run/collector-config.yaml --project "$PROJECT_ID"
+for secret in "$TEMPORAL_API_KEY_SECRET" "$COLLECTOR_CONFIG_SECRET"; do
+  gcloud secrets add-iam-policy-binding "$secret" \
+    --member "serviceAccount:${SERVICE_ACCOUNT_EMAIL}" \
+    --role roles/secretmanager.secretAccessor --project "$PROJECT_ID"
+done
+
+# Build the image (build context is the sample dir, keeping credentials out).
+gcloud builds submit gcp/cloud_run --region "$REGION" \
+  --tag "$WORKER_IMAGE" --project "$PROJECT_ID"
+
+# Render and deploy the two-container worker pool.
+envsubst < gcp/cloud_run/worker-pool.yaml > /tmp/worker-pool.yaml
+gcloud run worker-pools replace /tmp/worker-pool.yaml --project "$PROJECT_ID"

gcp/cloud_run/pyproject.toml

@@ -0,0 +1,22 @@
+[project]
+name = "temporalio-samples-gcp-cloud-run"
+version = "0.1a1"
+description = "Temporal Worker on a Google Cloud Run worker pool with the Cloud Run Id and OpenTelemetry plugins"
+authors = [{ name = "Temporal Technologies Inc", email = "sdk@temporal.io" }]
+requires-python = ">=3.10"
+readme = "README.md"
+license = "MIT"
+dependencies = ["temporalio[cloud-run-worker-otel]>=1.33.0,<2"]
+
+[dependency-groups]
+dev = [
+    "ruff>=0.5.0,<0.6",
+    "mypy>=1.4.1,<2",
+]
+
+[tool.uv]
+package = false
+
+# The Cloud Run Id plugin is unreleased; pin temporalio to the SDK commit that adds it until it ships on PyPI.
+[tool.uv.sources]
+temporalio = { git = "https://github.com/temporalio/sdk-python", rev = "fa5cf46ea04920be138bafca7e20ca6208822c30" }

gcp/cloud_run/settings.py

@@ -0,0 +1,34 @@
+"""Temporal connection settings shared by the worker and the starter.
+
+Set ``TEMPORAL_API_KEY`` to connect to Temporal Cloud (which enables TLS); leave
+it unset for a plaintext dev server.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Settings:
+    address: str
+    namespace: str
+    task_queue: str
+    api_key: str | None
+
+    @property
+    def tls(self) -> bool:
+        return self.api_key is not None
+
+
+def load_settings() -> Settings:
+    namespace = os.environ.get("TEMPORAL_NAMESPACE") or "default"
+    api_key = os.environ.get("TEMPORAL_API_KEY")
+    return Settings(
+        address=os.environ.get("TEMPORAL_ADDRESS") or f"{namespace}.tmprl.cloud:7233",
+        namespace=namespace,
+        task_queue=os.environ.get("TEMPORAL_TASK_QUEUE") or "gcp-cloud-run",
+        # Secret managers frequently preserve a trailing newline; strip it.
+        api_key=api_key.strip() if api_key else None,
+    )

gcp/cloud_run/starter.py

@@ -0,0 +1,34 @@
+"""Start the sample Workflow against the Cloud Run worker's task queue."""
+
+from __future__ import annotations
+
+import asyncio
+from uuid import uuid4
+
+from temporalio.client import Client
+
+from gcp.cloud_run.settings import load_settings
+from gcp.cloud_run.workflow import GreetingWorkflow
+
+
+async def main() -> None:
+    settings = load_settings()
+    client = await Client.connect(
+        settings.address,
+        namespace=settings.namespace,
+        api_key=settings.api_key,
+        tls=settings.tls,
+    )
+
+    workflow_id = f"gcp-cloud-run-{uuid4()}"
+    result = await client.execute_workflow(
+        GreetingWorkflow.run,
+        "Temporal",
+        id=workflow_id,
+        task_queue=settings.task_queue,
+    )
+    print(f"Workflow {workflow_id} result: {result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())