From 14f87d874f5e5fc95084b94bd7b5920ad1186f11 Mon Sep 17 00:00:00 2001 From: seanbollin Date: Wed, 19 Aug 2026 15:01:05 -0700 Subject: [PATCH 1/2] Add Cloud Run worker OpenTelemetry sample Adds gcp_cloud_run/: a long-running Temporal worker for Google Cloud Run worker pools that uses the temporalio.contrib.gcp.cloud_run OpenTelemetryPlugin to export Core metrics and traces over OTLP/gRPC to a local OpenTelemetry Collector sidecar (worker, workflow, starter, collector-config.yaml, worker-pool.yaml, Dockerfile, README). Note: the cloud-run-worker-otel SDK extra is not yet released, so pyproject.toml temporarily uses an editable local path source to resolve temporalio against a sibling sdk-python checkout. Replace it with a released temporalio[cloud-run-worker-otel] and regenerate uv.lock before this is marked ready. Co-authored-by: Edward Amsden Co-Authored-By: Claude Opus 4.8 --- README.md | 1 + gcp_cloud_run/.dockerignore | 6 + gcp_cloud_run/Dockerfile | 52 +++++ gcp_cloud_run/README.md | 287 +++++++++++++++++++++++++++ gcp_cloud_run/__init__.py | 1 + gcp_cloud_run/collector-config.yaml | 68 +++++++ gcp_cloud_run/pyproject.toml | 17 ++ gcp_cloud_run/starter.py | 60 ++++++ gcp_cloud_run/worker-pool.yaml | 58 ++++++ gcp_cloud_run/worker.py | 157 +++++++++++++++ gcp_cloud_run/workflow.py | 24 +++ pyproject.toml | 7 + tests/gcp_cloud_run/__init__.py | 0 tests/gcp_cloud_run/workflow_test.py | 24 +++ 14 files changed, 762 insertions(+) create mode 100644 gcp_cloud_run/.dockerignore create mode 100644 gcp_cloud_run/Dockerfile create mode 100644 gcp_cloud_run/README.md create mode 100644 gcp_cloud_run/__init__.py create mode 100644 gcp_cloud_run/collector-config.yaml create mode 100644 gcp_cloud_run/pyproject.toml create mode 100644 gcp_cloud_run/starter.py create mode 100644 gcp_cloud_run/worker-pool.yaml create mode 100644 gcp_cloud_run/worker.py create mode 100644 gcp_cloud_run/workflow.py create mode 100644 tests/gcp_cloud_run/__init__.py create mode 100644 tests/gcp_cloud_run/workflow_test.py diff --git a/README.md b/README.md index 164263995..7f3942301 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,7 @@ Some examples require extra dependencies. See each sample's directory for specif * [env_config](env_config) - Load client configuration from TOML files with programmatic overrides. * [external_storage](external_storage) - Offload large payloads to S3-compatible object storage, plus a codec server for the Web UI and CLI. * [external_storage_redis](external_storage_redis) - Redis driver for external storage +* [gcp_cloud_run](gcp_cloud_run) - Run a Temporal Worker on a Google Cloud Run worker pool with OpenTelemetry traces and metrics exported to a Google-Built Collector sidecar. * [gevent_async](gevent_async) - Combine gevent and Temporal. * [google_adk_agents](google_adk_agents) - Run Google ADK agents as durable Temporal workflows (model calls, tools, multi-agent, MCP, streaming). * [google_genai](google_genai) - Run the Google Gemini SDK inside durable Temporal workflows. diff --git a/gcp_cloud_run/.dockerignore b/gcp_cloud_run/.dockerignore new file mode 100644 index 000000000..2175c4ac9 --- /dev/null +++ b/gcp_cloud_run/.dockerignore @@ -0,0 +1,6 @@ +* +!Dockerfile +!pyproject.toml +!__init__.py +!worker.py +!workflow.py diff --git a/gcp_cloud_run/Dockerfile b/gcp_cloud_run/Dockerfile new file mode 100644 index 000000000..8a106534e --- /dev/null +++ b/gcp_cloud_run/Dockerfile @@ -0,0 +1,52 @@ +# syntax=docker/dockerfile:1 + +# NOTE: This image builds against a released `temporalio[cloud-run-worker-otel]`. +# While the plugin is unreleased, `pyproject.toml` uses a temporary local path +# source for `temporalio` (`../../sdk-python`), which is OUTSIDE this build +# context and therefore not available here. Before building the container, +# replace that source with the first released version (or a pushed git rev) and +# regenerate `uv.lock` with `uv lock`, then restore the `uv.lock` copy and the +# `--frozen` flag below. + +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 +COPY --chown=app:app __init__.py worker.py workflow.py /app/gcp_cloud_run/ + +USER app +CMD ["python", "-m", "gcp_cloud_run.worker"] diff --git a/gcp_cloud_run/README.md b/gcp_cloud_run/README.md new file mode 100644 index 000000000..8ec848d79 --- /dev/null +++ b/gcp_cloud_run/README.md @@ -0,0 +1,287 @@ +# Google Cloud Run OpenTelemetry Worker + +This sample runs a long-lived Temporal Worker in a [Google Cloud Run worker +pool](https://cloud.google.com/run/docs/worker-pools) and sends traces and +Temporal Core metrics to a Google-Built OpenTelemetry Collector sidecar. + +The worker uses `temporalio.contrib.gcp.cloud_run.OpenTelemetryPlugin` to +configure: + +- OTLP gRPC export to `http://localhost:4317`. +- `service.name` from the Cloud Run-provided `CLOUD_RUN_WORKER_POOL` variable. +- A replay-safe OpenTelemetry tracer provider. +- Temporal Core metrics with a 60-second export interval. + +Those are plugin defaults. This sample opts into `add_temporal_spans=True` so +operations such as `RunWorkflow:GreetingWorkflow` are traced. The collector +detects the Google Cloud resource, authenticates through the worker pool's +service account, exports traces through the Google Cloud Telemetry API, and +exports metrics to Google Managed Service for Prometheus. + +> The Cloud Run plugin is not released yet. Both `pyproject.toml` files (this +> directory and the repository root) currently use a temporary local +> `[tool.uv.sources]` path to the SDK checkout at `../sdk-python`. Replace those +> source pins with the first released `temporalio[cloud-run-worker-otel]` +> version before merging this sample. The container build additionally requires +> a released (or pushed git) SDK and a regenerated `uv.lock`, because the local +> path is not available inside the Docker build context. + +## Files + +| File | Purpose | +|------|---------| +| `workflow.py` | Greeting Workflow and Activity | +| `worker.py` | Plugin setup, collector readiness, Worker, and graceful shutdown | +| `starter.py` | Local client that starts a unique Workflow Execution | +| `collector-config.yaml` | Collector receiver, GCP detection, processing, and exporters | +| `worker-pool.yaml` | Environment-substituted two-container WorkerPool manifest | +| `Dockerfile` | Reproducible, non-root worker image | +| `pyproject.toml` | Standalone dependencies used by the container build | + +## Prerequisites + +- A Temporal Cloud namespace and API key. +- A Google Cloud project with billing enabled. +- The Google Cloud CLI, authenticated to that project. +- Permission to manage Cloud Run worker pools, builds, Artifact Registry, + service accounts, IAM bindings, and Secret Manager secrets. +- `uv` for running the starter locally. +- `envsubst` for rendering the manifest. It is preinstalled in Cloud Shell and + is available in the `gettext` package on many systems. + +Cloud Run worker pools use manual scaling. Active instances are billed +continuously, so always complete the scale-to-zero step after testing. + +## 1. Choose project-neutral resource names + +Run the commands from the repository root. The values below are examples; use +names appropriate for your project and namespace. + +```bash +export PROJECT_ID=your-project-id +export REGION=us-central1 +export REPOSITORY=temporal-workers +export WORKER_POOL=temporal-gcp-cloud-run-worker +export SERVICE_ACCOUNT=temporal-gcp-cloud-run-worker +export SERVICE_ACCOUNT_EMAIL="${SERVICE_ACCOUNT}@${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 +export COLLECTOR_CONFIG_SECRET=temporal-otel-collector-config +export TEMPORAL_API_KEY_SECRET_VERSION=1 +export COLLECTOR_CONFIG_SECRET_VERSION=1 + +export WORKER_IMAGE="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPOSITORY}/gcp-cloud-run:sample-v1" +``` + +The API key file must not be inside the repository or Docker build context. + +## 2. Enable APIs and create the runtime identity + +```bash +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 create "$REPOSITORY" \ + --location "$REGION" \ + --project "$PROJECT_ID" \ + --repository-format docker + +gcloud iam service-accounts create "$SERVICE_ACCOUNT" \ + --display-name "Temporal Cloud Run worker" \ + --project "$PROJECT_ID" +``` + +Grant only the roles needed by the collector and Cloud Run logging: + +```bash +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 +``` + +## 3. Store both values in Secret Manager + +Create the API-key secret directly from its file. Do not place the key on a +command line or copy it into the manifest. + +```bash +gcloud secrets create "$TEMPORAL_API_KEY_SECRET" \ + --data-file "$TEMPORAL_API_KEY_FILE" \ + --project "$PROJECT_ID" \ + --replication-policy automatic + +gcloud secrets create "$COLLECTOR_CONFIG_SECRET" \ + --data-file gcp_cloud_run/collector-config.yaml \ + --project "$PROJECT_ID" \ + --replication-policy automatic +``` + +If a secret already exists, add a version instead and update the corresponding +numeric version variable: + +```bash +gcloud secrets versions add "$COLLECTOR_CONFIG_SECRET" \ + --data-file gcp_cloud_run/collector-config.yaml \ + --project "$PROJECT_ID" +``` + +Grant the runtime identity access only to these two secrets: + +```bash +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 +``` + +The worker strips surrounding whitespace from `TEMPORAL_API_KEY` because +Secret Manager preserves a trailing newline present in the source file. + +## 4. Build the worker image + +The build context is the sample directory, which keeps credentials and the +rest of the repository out of the image build. + +```bash +gcloud builds submit gcp_cloud_run \ + --project "$PROJECT_ID" \ + --region "$REGION" \ + --tag "$WORKER_IMAGE" +``` + +The Dockerfile pins the build tools and installs the released SDK. The final +image contains only Python, the virtual environment, and the worker source, and +runs as a non-root user. Until the plugin is released, update the standalone +`pyproject.toml` to a released or git-pinned SDK and regenerate `uv.lock`, as +noted at the top of this file and in the `Dockerfile`. + +## 5. Render and deploy the worker pool + +The manifest pins the Google-Built Collector image to `0.156.0`. It gives the +collector an HTTP startup probe and declares the worker's dependency on that +probe. The application also performs a bounded `localhost:4317` readiness wait +to protect deployments made through older Cloud Run tooling. + +```bash +export INSTANCE_COUNT=1 +export RENDERED_MANIFEST="/tmp/${WORKER_POOL}.yaml" + +envsubst < gcp_cloud_run/worker-pool.yaml > "$RENDERED_MANIFEST" + +gcloud run worker-pools replace "$RENDERED_MANIFEST" \ + --project "$PROJECT_ID" +``` + +The collector configuration is injected as the secret environment variable +`OTELCOL_CONFIG` and loaded with `--config=env:OTELCOL_CONFIG`. This is +intentional: secret-backed file volumes have been rejected by some Cloud Run +worker-pool rollouts even where current documentation advertises support. + +The plugin supports overriding `metric_periodicity`, but the collector does not +batch cumulative metrics. It forwards each SDK export directly, including any +runtime shutdown-time export, so two points for one Managed Prometheus time +series cannot be grouped into the same request. Traces use a dedicated +five-second batch processor. + +## 6. Start a Workflow + +The local starter accepts either `TEMPORAL_API_KEY` or a file path. Prefer the +file path so the credential does not appear in shell history. + +```bash +uv sync --group gcp-cloud-run + +TEMPORAL_API_KEY_FILE="$TEMPORAL_API_KEY_FILE" \ + uv run --group gcp-cloud-run python -m gcp_cloud_run.starter +``` + +The expected result ends with: + +```text +Hello, Temporal! +``` + +## 7. Verify telemetry + +In the Cloud Run logs, verify that the worker reports the expected endpoint and +worker-pool-derived service name, without exporter errors: + +```bash +gcloud run worker-pools logs read "$WORKER_POOL" \ + --project "$PROJECT_ID" \ + --region "$REGION" \ + --limit 100 +``` + +Then verify: + +- Trace Explorer contains `RunWorkflow:GreetingWorkflow` with + `service.name` equal to the worker-pool name. +- Metrics Explorer contains + `prometheus.googleapis.com/temporal_workflow_completed_total/counter`. +- The trace and metrics have the expected Cloud Run, region, and project + resource attributes added by the collector's GCP resource detector. + +Trace indexing can lag successful OTLP export. Check collector error logs before +treating a temporarily absent Trace Explorer result as an export failure. + +## 8. Scale to zero + +Stop continuous compute charges immediately after validation: + +```bash +gcloud run worker-pools update "$WORKER_POOL" \ + --instances 0 \ + --project "$PROJECT_ID" \ + --region "$REGION" +``` + +Cloud Run sends `SIGTERM` to both containers. The worker begins Temporal Worker +shutdown, allows up to five seconds for graceful completion, then force-flushes +Python traces with a two-second limit. A clean application shutdown logs +`traces_flushed=True`. + +Temporal Core metrics are exported periodically and do not currently expose an +explicit flush operation. This is an accepted limitation: on scale-down, metric +samples accumulated since the last periodic export (up to one +`metric_periodicity` interval, 60 seconds by default) may not be delivered +before the process exits. Only the trailing, not-yet-exported window is +affected; earlier metrics are already in Google Managed Service for Prometheus. + +## Troubleshooting + +- **No Temporal spans:** Keep `add_temporal_spans=True`. The endpoint, service + name, tracer provider, runtime, and Core metrics use GCP defaults, but named + Temporal operation spans are intentionally opt-in. +- **Collector startup ordering is rejected:** Older worker-pool rollouts or CLI + versions might not accept a startup probe plus container dependency. Remove + only the `run.googleapis.com/container-dependencies` annotation; the worker's + bounded readiness wait still prevents it from connecting before the + collector listens. +- **`gcloud` reports a missing `grpc` Python module:** Prefer Cloud Shell or a + current Google Cloud CLI installation. Some local installations require a + Python environment containing `grpcio` together with + `CLOUDSDK_PYTHON_SITEPACKAGES=1`; this is a CLI issue, not a worker setting. +- **Authentication says the JWT is missing:** Confirm the configured secret + version contains only the Temporal API key. The sample strips the common + trailing newline automatically. diff --git a/gcp_cloud_run/__init__.py b/gcp_cloud_run/__init__.py new file mode 100644 index 000000000..90949901e --- /dev/null +++ b/gcp_cloud_run/__init__.py @@ -0,0 +1 @@ +"""Google Cloud Run worker-pool OpenTelemetry sample.""" diff --git a/gcp_cloud_run/collector-config.yaml b/gcp_cloud_run/collector-config.yaml new file mode 100644 index 000000000..674c8817e --- /dev/null +++ b/gcp_cloud_run/collector-config.yaml @@ -0,0 +1,68 @@ +# @@@SNIPSTART python-cloud-run-otel-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 + 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") + 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 diff --git a/gcp_cloud_run/pyproject.toml b/gcp_cloud_run/pyproject.toml new file mode 100644 index 000000000..5eb7f8c44 --- /dev/null +++ b/gcp_cloud_run/pyproject.toml @@ -0,0 +1,17 @@ +[project] +name = "temporalio-gcp-cloud-run-sample" +version = "0.1.0" +description = "Temporal Worker on a Google Cloud Run worker pool with OpenTelemetry" +requires-python = ">=3.10" +dependencies = ["temporalio[cloud-run-worker-otel]"] + +[tool.uv] +package = false + +# TEMPORARY: the Cloud Run OpenTelemetry plugin is not released yet, so this +# builds `temporalio` from the local SDK checkout. Replace this source with the +# first released `temporalio[cloud-run-worker-otel]` version (or a pushed git +# rev) before merging this sample or building the container image. The local +# path is not available inside the Docker build context. +[tool.uv.sources] +temporalio = { path = "../../sdk-python", editable = true } diff --git a/gcp_cloud_run/starter.py b/gcp_cloud_run/starter.py new file mode 100644 index 000000000..9858a3579 --- /dev/null +++ b/gcp_cloud_run/starter.py @@ -0,0 +1,60 @@ +"""Start the sample Workflow against Temporal Cloud.""" + +from __future__ import annotations + +import asyncio +import os +from pathlib import Path +from uuid import uuid4 + +from temporalio.client import Client + +from gcp_cloud_run.workflow import GreetingWorkflow + + +def _required(name: str) -> str: + value = os.environ.get(name, "").strip() + if not value: + raise RuntimeError(f"{name} must be set to a non-empty value") + return value + + +def _read_api_key() -> str: + value = os.environ.get("TEMPORAL_API_KEY", "").strip() + if value: + return value + + path = os.environ.get("TEMPORAL_API_KEY_FILE", "").strip() + if not path: + raise RuntimeError("TEMPORAL_API_KEY or TEMPORAL_API_KEY_FILE must be set") + value = Path(path).read_text(encoding="utf-8").strip() + if not value: + raise RuntimeError("TEMPORAL_API_KEY_FILE must contain a non-empty value") + return value + + +async def main() -> None: + namespace = _required("TEMPORAL_NAMESPACE") + address = os.environ.get( + "TEMPORAL_ADDRESS", f"{namespace}.tmprl.cloud:7233" + ).strip() + task_queue = _required("TEMPORAL_TASK_QUEUE") + client = await Client.connect( + address, + namespace=namespace, + api_key=_read_api_key(), + tls=True, + ) + + workflow_id = f"gcp-cloud-run-{uuid4()}" + result = await client.execute_workflow( + GreetingWorkflow.run, + "Temporal", + id=workflow_id, + task_queue=task_queue, + ) + print(f"Workflow {workflow_id} result: {result}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/gcp_cloud_run/worker-pool.yaml b/gcp_cloud_run/worker-pool.yaml new file mode 100644 index 000000000..f0cb2c671 --- /dev/null +++ b/gcp_cloud_run/worker-pool.yaml @@ -0,0 +1,58 @@ +apiVersion: run.googleapis.com/v1 +kind: WorkerPool +metadata: + annotations: + run.googleapis.com/manualInstanceCount: "${INSTANCE_COUNT}" + run.googleapis.com/scalingMode: manual + labels: + cloud.googleapis.com/location: "${REGION}" + name: "${WORKER_POOL}" +spec: + template: + metadata: + annotations: + run.googleapis.com/container-dependencies: '{"worker":["collector"]}' + run.googleapis.com/execution-environment: gen2 + spec: + containerConcurrency: 0 + containers: + - name: worker + image: "${WORKER_IMAGE}" + env: + - name: TEMPORAL_NAMESPACE + value: "${TEMPORAL_NAMESPACE}" + - name: TEMPORAL_ADDRESS + value: "${TEMPORAL_ADDRESS}" + - name: TEMPORAL_TASK_QUEUE + value: "${TEMPORAL_TASK_QUEUE}" + - name: TEMPORAL_API_KEY + valueFrom: + secretKeyRef: + key: "${TEMPORAL_API_KEY_SECRET_VERSION}" + name: "${TEMPORAL_API_KEY_SECRET}" + resources: + limits: + cpu: "1" + memory: 512Mi + - name: collector + image: us-docker.pkg.dev/cloud-ops-agents-artifacts/google-cloud-opentelemetry-collector/otelcol-google:0.156.0 + args: + - --config=env:OTELCOL_CONFIG + env: + - name: OTELCOL_CONFIG + valueFrom: + secretKeyRef: + key: "${COLLECTOR_CONFIG_SECRET_VERSION}" + name: "${COLLECTOR_CONFIG_SECRET}" + startupProbe: + httpGet: + path: / + port: 13133 + timeoutSeconds: 1 + periodSeconds: 2 + failureThreshold: 30 + resources: + limits: + cpu: "1" + memory: 512Mi + serviceAccountName: "${SERVICE_ACCOUNT_EMAIL}" diff --git a/gcp_cloud_run/worker.py b/gcp_cloud_run/worker.py new file mode 100644 index 000000000..33142549a --- /dev/null +++ b/gcp_cloud_run/worker.py @@ -0,0 +1,157 @@ +"""Run an OpenTelemetry-instrumented Temporal Worker on Cloud Run.""" + +from __future__ import annotations + +import asyncio +import os +import signal +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import timedelta + +from temporalio.client import Client +from temporalio.contrib.gcp.cloud_run import ( + DEFAULT_METRIC_PERIODICITY, + OpenTelemetryPlugin, +) +from temporalio.worker import Worker + +from gcp_cloud_run.workflow import GreetingWorkflow, compose_greeting + +COLLECTOR_HOST = "127.0.0.1" +COLLECTOR_PORT = 4317 +COLLECTOR_STARTUP_TIMEOUT = timedelta(seconds=60) +WORKER_GRACEFUL_SHUTDOWN_TIMEOUT = timedelta(seconds=5) +TRACE_FLUSH_TIMEOUT = timedelta(seconds=2) + + +@dataclass(frozen=True) +class WorkerSettings: + """Environment-backed Temporal connection settings.""" + + address: str + namespace: str + task_queue: str + api_key: str + + @classmethod + def from_environment( + cls, environment: Mapping[str, str] | None = None + ) -> WorkerSettings: + env = os.environ if environment is None else environment + namespace = _required(env, "TEMPORAL_NAMESPACE") + return cls( + address=_optional(env, "TEMPORAL_ADDRESS") + or f"{namespace}.tmprl.cloud:7233", + namespace=namespace, + task_queue=_required(env, "TEMPORAL_TASK_QUEUE"), + # Secret Manager preserves trailing newlines from the source file. + api_key=_required(env, "TEMPORAL_API_KEY"), + ) + + +def _required(environment: Mapping[str, str], name: str) -> str: + value = _optional(environment, name) + if value is None: + raise RuntimeError(f"{name} must be set to a non-empty value") + return value + + +def _optional(environment: Mapping[str, str], name: str) -> str | None: + value = environment.get(name) + if value is None: + return None + stripped = value.strip() + return stripped or None + + +async def wait_for_collector( + *, + host: str = COLLECTOR_HOST, + port: int = COLLECTOR_PORT, + timeout: timedelta = COLLECTOR_STARTUP_TIMEOUT, +) -> None: + """Wait until the local OTLP gRPC socket accepts connections.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout.total_seconds() + last_error: OSError | None = None + + while loop.time() < deadline: + try: + _, writer = await asyncio.open_connection(host, port) + writer.close() + await writer.wait_closed() + return + except OSError as err: + last_error = err + await asyncio.sleep(0.5) + + raise RuntimeError( + f"OpenTelemetry Collector did not listen on {host}:{port} " + f"within {timeout.total_seconds():g} seconds" + ) from last_error + + +async def run_worker() -> None: + """Connect to Temporal and run until Cloud Run requests shutdown.""" + settings = WorkerSettings.from_environment() + await wait_for_collector() + + expected_metric_periodicity = timedelta(seconds=60) + if DEFAULT_METRIC_PERIODICITY != expected_metric_periodicity: + raise RuntimeError( + "Expected the coordinated 60-second GCP metric periodicity, got " + f"{DEFAULT_METRIC_PERIODICITY}" + ) + + # @@@SNIPSTART python-cloud-run-otel-worker + # Endpoint, service name, Core metrics, and tracer provider all use the GCP + # plugin defaults. The opt-in adds named Temporal operation spans. + plugin = OpenTelemetryPlugin(add_temporal_spans=True) + client = await Client.connect( + settings.address, + namespace=settings.namespace, + api_key=settings.api_key, + tls=True, + plugins=[plugin], + ) + worker = Worker( + client, + task_queue=settings.task_queue, + workflows=[GreetingWorkflow], + activities=[compose_greeting], + graceful_shutdown_timeout=WORKER_GRACEFUL_SHUTDOWN_TIMEOUT, + ) + # @@@SNIPEND + + loop = asyncio.get_running_loop() + shutdown_requested = False + + def request_shutdown() -> None: + nonlocal shutdown_requested + if shutdown_requested: + return + shutdown_requested = True + print("Worker shutdown requested", flush=True) + loop.create_task(worker.shutdown()) + + for signum in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(signum, request_shutdown) + + print( + "Worker starting " + f"task_queue={settings.task_queue} " + f"otel_endpoint={plugin.endpoint} " + f"service_name={plugin.service_name} " + f"metric_periodicity={DEFAULT_METRIC_PERIODICITY.total_seconds():g}s", + flush=True, + ) + try: + await worker.run() + finally: + traces_flushed = await asyncio.to_thread(plugin.shutdown, TRACE_FLUSH_TIMEOUT) + print(f"Worker stopped traces_flushed={traces_flushed}", flush=True) + + +if __name__ == "__main__": + asyncio.run(run_worker()) diff --git a/gcp_cloud_run/workflow.py b/gcp_cloud_run/workflow.py new file mode 100644 index 000000000..4e544199e --- /dev/null +++ b/gcp_cloud_run/workflow.py @@ -0,0 +1,24 @@ +"""Workflow and Activity used by the Google Cloud Run sample.""" + +from datetime import timedelta + +from temporalio import activity, workflow + + +@activity.defn +async def compose_greeting(name: str) -> str: + """Compose a greeting outside the Workflow sandbox.""" + return f"Hello, {name}!" + + +@workflow.defn +class GreetingWorkflow: + """Run a single greeting Activity.""" + + @workflow.run + async def run(self, name: str) -> str: + return await workflow.execute_activity( + compose_greeting, + name, + start_to_close_timeout=timedelta(seconds=10), + ) diff --git a/pyproject.toml b/pyproject.toml index 4d8422610..446b79944 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,6 +44,7 @@ external-storage = [ "types_aiobotocore_s3>=2.25.2", ] external-storage-redis = ["redis>=5.0.0,<8"] +gcp-cloud-run = ["temporalio[cloud-run-worker-otel]"] gevent = ["gevent>=25.4.2 ; python_version >= '3.8'"] google-adk = ["temporalio[google-adk] >= 1.31.0", "google-adk>=2.2.0,<3"] google-genai = [ @@ -104,6 +105,12 @@ constraint-dependencies = [ "yarl!=1.24.0", ] +# TEMPORARY: the Cloud Run OpenTelemetry plugin (temporalio[cloud-run-worker-otel]) +# is not released yet, so build `temporalio` from the local SDK checkout. Remove +# this source once a release that includes the plugin is available on PyPI. +[tool.uv.sources] +temporalio = { path = "../sdk-python", editable = true } + [tool.setuptools.packages.find] exclude = ["lambda_worker*"] diff --git a/tests/gcp_cloud_run/__init__.py b/tests/gcp_cloud_run/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/gcp_cloud_run/workflow_test.py b/tests/gcp_cloud_run/workflow_test.py new file mode 100644 index 000000000..5296b2045 --- /dev/null +++ b/tests/gcp_cloud_run/workflow_test.py @@ -0,0 +1,24 @@ +import uuid + +from temporalio.client import Client +from temporalio.worker import Worker + +from gcp_cloud_run.workflow import GreetingWorkflow, compose_greeting + + +async def test_greeting_workflow(client: Client): + task_queue_name = str(uuid.uuid4()) + + async with Worker( + client, + task_queue=task_queue_name, + workflows=[GreetingWorkflow], + activities=[compose_greeting], + ): + result = await client.execute_workflow( + GreetingWorkflow.run, + "Temporal", + id=str(uuid.uuid4()), + task_queue=task_queue_name, + ) + assert result == "Hello, Temporal!" From 156ec2cab810678f210b214138d5a98f6b8e3bf1 Mon Sep 17 00:00:00 2001 From: seanbollin Date: Fri, 21 Aug 2026 11:15:28 -0700 Subject: [PATCH 2/2] Support plaintext/self-hosted Temporal in the Cloud Run worker Make TEMPORAL_API_KEY optional and enable TLS only when an API key is present, so the sample connects to a plaintext dev server or self-hosted cluster, not just Temporal Cloud (matches the Go and Java samples). Verified by deploying to Cloud Run against a dev server. Co-Authored-By: Claude Opus 4.8 --- gcp_cloud_run/worker.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/gcp_cloud_run/worker.py b/gcp_cloud_run/worker.py index 33142549a..1e54f285f 100644 --- a/gcp_cloud_run/worker.py +++ b/gcp_cloud_run/worker.py @@ -32,7 +32,7 @@ class WorkerSettings: address: str namespace: str task_queue: str - api_key: str + api_key: str | None @classmethod def from_environment( @@ -45,8 +45,10 @@ def from_environment( or f"{namespace}.tmprl.cloud:7233", namespace=namespace, task_queue=_required(env, "TEMPORAL_TASK_QUEUE"), - # Secret Manager preserves trailing newlines from the source file. - api_key=_required(env, "TEMPORAL_API_KEY"), + # Optional: set for Temporal Cloud (enables TLS). Omit for a + # plaintext self-hosted / dev server. Secret Manager preserves + # trailing newlines from the source file. + api_key=_optional(env, "TEMPORAL_API_KEY"), ) @@ -112,7 +114,8 @@ async def run_worker() -> None: settings.address, namespace=settings.namespace, api_key=settings.api_key, - tls=True, + # TLS for Temporal Cloud (api key present); plaintext for a dev server. + tls=bool(settings.api_key), plugins=[plugin], ) worker = Worker(