From 91d786713b0bc00be23011aaa3d0934373e3f49e Mon Sep 17 00:00:00 2001 From: seanbollin Date: Fri, 21 Aug 2026 12:22:49 -0700 Subject: [PATCH 1/6] Add Google Cloud Run worker identity/deployment helper Adds an experimental Google Cloud Run helper, mirroring the existing AWS Lambda module's worker-ID behavior. Because Cloud Run runs a long-lived container (unlike Lambda's per-invocation model), this is a metadata helper rather than a worker wrapper: it reads the Cloud Run instance metadata -- the instance id from the metadata server, plus the worker pool/service name and revision from CLOUD_RUN_WORKER_POOL / CLOUD_RUN_REVISION (worker pools) or K_SERVICE / K_REVISION (services) -- and derives a worker identity and a WorkerDeploymentVersion to apply to a normal long-lived worker. Covers both Cloud Run worker pools and services. Co-Authored-By: Claude Opus 4.8 --- contrib/temporal-gcp-cloud-run/README.md | 71 +++++ contrib/temporal-gcp-cloud-run/build.gradle | 7 + .../gcp/cloudrun/GoogleCloudRunMetadata.java | 249 ++++++++++++++++++ settings.gradle | 2 + 4 files changed, 329 insertions(+) create mode 100644 contrib/temporal-gcp-cloud-run/README.md create mode 100644 contrib/temporal-gcp-cloud-run/build.gradle create mode 100644 contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java diff --git a/contrib/temporal-gcp-cloud-run/README.md b/contrib/temporal-gcp-cloud-run/README.md new file mode 100644 index 0000000000..a0e3da1763 --- /dev/null +++ b/contrib/temporal-gcp-cloud-run/README.md @@ -0,0 +1,71 @@ +# Temporal Google Cloud Run support + +This module derives a Temporal worker identity and a `WorkerDeploymentVersion` from Google Cloud Run instance metadata, for both Cloud Run **worker pools** and Cloud Run **services**. + +Cloud Run runs a long-lived container, so there is no per-request handler to wrap. This module is a small metadata helper rather than a worker wrapper: fetch the metadata once during startup and apply it to your client and worker option builders. + +> Experimental: Google Cloud Run support is experimental and may change without notice. + +## Quick start + +Add `temporal-gcp-cloud-run` next to your Temporal SDK dependency, then fetch the metadata and apply it while the worker starts up: + +```java +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.gcp.cloudrun.GoogleCloudRunMetadata; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; +import io.temporal.worker.WorkerOptions; + +public final class Main { + public static void main(String[] args) { + // Read Cloud Run instance metadata once during startup. + GoogleCloudRunMetadata metadata = GoogleCloudRunMetadata.fetch(); + + WorkflowServiceStubs service = + WorkflowServiceStubs.newServiceStubs( + WorkflowServiceStubsOptions.newBuilder() + .setTarget("my-namespace.tmprl.cloud:7233") + .build()); + + // applyTo(...) sets the derived worker identity on the client options. + WorkflowClient client = + WorkflowClient.newInstance( + service, metadata.applyTo(WorkflowClientOptions.newBuilder()).build()); + + WorkerFactory factory = WorkerFactory.newInstance(client); + + // applyTo(...) sets the deployment version and enables worker versioning on the worker options. + WorkerOptions workerOptions = metadata.applyTo(WorkerOptions.newBuilder()).build(); + + Worker worker = factory.newWorker("orders", workerOptions); + worker.registerWorkflowImplementationTypes(OrderWorkflowImpl.class); + worker.registerActivitiesImplementations(new OrderActivitiesImpl()); + + factory.start(); + } +} +``` + +Both `applyTo(...)` methods return the builder they were given, so they compose with the rest of your builder configuration. + +## How it works + +`GoogleCloudRunMetadata.fetch()` resolves three values: + +- **name** (the Temporal deployment name): the first non-empty of `CLOUD_RUN_WORKER_POOL` (set on Cloud Run worker pools) then `K_SERVICE` (set on Cloud Run services). +- **revision**: the first non-empty of `CLOUD_RUN_REVISION` (worker pools) then `K_REVISION` (services). +- **instanceId**: read from the Cloud Run metadata server with a single HTTP `GET` to `http://metadata.google.internal/computeMetadata/v1/instance/id` with the required `Metadata-Flavor: Google` header. The metadata server is available on both worker pools and services. + +Worker pools receive `CLOUD_RUN_WORKER_POOL` and `CLOUD_RUN_REVISION` and no `K_*` variables, while services receive `K_SERVICE` and `K_REVISION`, so resolving each value from the worker-pool variable first and the service variable second supports both. + +`workerIdentity()` returns `@`, falling back to `@` and then to the bare `` when those values are blank. `workerDeploymentVersion()` maps the name to the deployment name and the revision to the build id, so each Cloud Run revision becomes a distinct `WorkerDeploymentVersion`. + +The two `applyTo(...)` overloads mirror the SDK's "apply defaults to your options" idiom: `applyTo(WorkflowClientOptions.Builder)` sets the worker identity on the client side, and `applyTo(WorkerOptions.Builder)` sets the deployment version (with versioning enabled) on the worker side. Each returns the builder for chaining. If you prefer to read the values yourself, call `workerIdentity()` and `workerDeploymentVersion()` directly. + +Because the metadata server is only reachable from a Cloud Run instance, `fetch()` throws `IllegalStateException` when it cannot be reached, and `workerDeploymentVersion()` (and therefore `applyTo(WorkerOptions.Builder)`) throws `IllegalStateException` when the name or revision is not set. Use `GoogleCloudRunMetadata.fetch(String metadataUrl, Duration timeout)` to override the metadata URL or the request timeout. + +This module depends only on the Temporal SDK at compile time and uses the JDK's `HttpURLConnection` for the metadata request, so it adds no additional runtime dependencies. diff --git a/contrib/temporal-gcp-cloud-run/build.gradle b/contrib/temporal-gcp-cloud-run/build.gradle new file mode 100644 index 0000000000..d08b6289f1 --- /dev/null +++ b/contrib/temporal-gcp-cloud-run/build.gradle @@ -0,0 +1,7 @@ +description = '''Temporal Google Cloud Run support''' + +dependencies { + // This module shouldn't carry temporal-sdk with it, especially for situations when users may + // be using a shaded artifact. + compileOnly project(':temporal-sdk') +} diff --git a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java new file mode 100644 index 0000000000..40c69adecf --- /dev/null +++ b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java @@ -0,0 +1,249 @@ +package io.temporal.gcp.cloudrun; + +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.Experimental; +import io.temporal.common.WorkerDeploymentVersion; +import io.temporal.worker.WorkerDeploymentOptions; +import io.temporal.worker.WorkerOptions; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Objects; + +/** + * Reads Google Cloud Run instance metadata and derives a Temporal worker identity and a {@link + * WorkerDeploymentVersion} from it. + * + *

Cloud Run runs a long-lived container rather than a per-request handler, so this class is a + * metadata helper rather than a worker wrapper. Fetch the metadata once while a worker starts up, + * then apply it to your client and worker option builders with {@link + * #applyTo(WorkflowClientOptions.Builder)} and {@link #applyTo(WorkerOptions.Builder)}. + * + *

The deployment name and revision are resolved from environment variables Cloud Run injects + * into every instance. Cloud Run worker pools set {@code CLOUD_RUN_WORKER_POOL} and {@code + * CLOUD_RUN_REVISION}; Cloud Run services set {@code K_SERVICE} and {@code K_REVISION}. The + * name is the first non-empty of {@code CLOUD_RUN_WORKER_POOL} then {@code K_SERVICE}, and the + * revision is the first non-empty of {@code CLOUD_RUN_REVISION} then {@code K_REVISION}. The unique + * instance id is only available from the Cloud Run metadata server, so {@link #fetch()} performs a + * single HTTP request against it. + * + *

Experimental: Google Cloud Run support is experimental and may change without notice. + */ +@Experimental +public final class GoogleCloudRunMetadata { + /** Name of the environment variable Cloud Run worker pools set to the worker pool name. */ + public static final String CLOUD_RUN_WORKER_POOL = "CLOUD_RUN_WORKER_POOL"; + + /** Name of the environment variable Cloud Run worker pools set to the revision name. */ + public static final String CLOUD_RUN_REVISION = "CLOUD_RUN_REVISION"; + + /** Name of the environment variable Cloud Run services set to the deployed service name. */ + public static final String K_SERVICE = "K_SERVICE"; + + /** Name of the environment variable Cloud Run services set to the deployed revision name. */ + public static final String K_REVISION = "K_REVISION"; + + /** Default Cloud Run metadata server URL that returns the unique instance id. */ + public static final String DEFAULT_METADATA_URL = + "http://metadata.google.internal/computeMetadata/v1/instance/id"; + + /** Default connect and read timeout used when contacting the metadata server. */ + public static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(2); + + private static final String METADATA_FLAVOR_HEADER = "Metadata-Flavor"; + private static final String METADATA_FLAVOR_VALUE = "Google"; + + private final String instanceId; + private final String name; + private final String revision; + + private GoogleCloudRunMetadata(String instanceId, String name, String revision) { + this.instanceId = instanceId; + this.name = name; + this.revision = revision; + } + + /** + * Fetches Cloud Run instance metadata using the {@linkplain #DEFAULT_METADATA_URL default + * metadata URL} and the {@linkplain #DEFAULT_TIMEOUT default timeout}. + * + * @return metadata describing the current Cloud Run instance. + * @throws IllegalStateException if the metadata server cannot be reached, which usually means the + * process is not running on Google Cloud Run. + */ + public static GoogleCloudRunMetadata fetch() { + return fetch(DEFAULT_METADATA_URL, DEFAULT_TIMEOUT); + } + + /** + * Fetches Cloud Run instance metadata from the supplied metadata server URL. + * + *

The deployment name is read from {@code CLOUD_RUN_WORKER_POOL} then {@code K_SERVICE}, and + * the revision from {@code CLOUD_RUN_REVISION} then {@code K_REVISION}. The unique instance id is + * read from {@code metadataUrl} with the required {@code Metadata-Flavor: Google} request header. + * + * @param metadataUrl URL of the Cloud Run metadata endpoint that returns the instance id. + * @param timeout connect and read timeout applied to the metadata request. + * @return metadata describing the current Cloud Run instance. + * @throws IllegalStateException if the metadata server cannot be reached, which usually means the + * process is not running on Google Cloud Run. + */ + public static GoogleCloudRunMetadata fetch(String metadataUrl, Duration timeout) { + Objects.requireNonNull(metadataUrl, "metadataUrl"); + Objects.requireNonNull(timeout, "timeout"); + + String name = firstNonBlank(System.getenv(CLOUD_RUN_WORKER_POOL), System.getenv(K_SERVICE)); + String revision = firstNonBlank(System.getenv(CLOUD_RUN_REVISION), System.getenv(K_REVISION)); + + HttpURLConnection connection = null; + try { + connection = (HttpURLConnection) new URL(metadataUrl).openConnection(); + connection.setRequestMethod("GET"); + connection.setRequestProperty(METADATA_FLAVOR_HEADER, METADATA_FLAVOR_VALUE); + int timeoutMillis = timeoutMillis(timeout); + connection.setConnectTimeout(timeoutMillis); + connection.setReadTimeout(timeoutMillis); + + String instanceId = readBody(connection).trim(); + return new GoogleCloudRunMetadata(instanceId, name, revision); + } catch (IOException e) { + throw new IllegalStateException( + "Unable to read the Cloud Run instance id from the metadata server at " + + metadataUrl + + "; this process may not be running on Google Cloud Run", + e); + } finally { + if (connection != null) { + connection.disconnect(); + } + } + } + + /** + * @return the unique Cloud Run instance id read from the metadata server. + */ + public String getInstanceId() { + return instanceId; + } + + /** + * @return the Cloud Run deployment name, resolved from {@code CLOUD_RUN_WORKER_POOL} then {@code + * K_SERVICE}, or {@code null} when neither was set. + */ + public String getName() { + return name; + } + + /** + * @return the Cloud Run revision name, resolved from {@code CLOUD_RUN_REVISION} then {@code + * K_REVISION}, or {@code null} when neither was set. + */ + public String getRevision() { + return revision; + } + + /** + * Builds a Temporal worker identity for this Cloud Run instance. + * + *

The identity is {@code instanceId@revision}. When the revision is blank the name is used + * instead, and when both are blank the bare instance id is returned. + * + * @return a worker identity string suitable for {@code WorkflowClientOptions} and {@code + * WorkerOptions}. + */ + public String workerIdentity() { + if (!isBlank(revision)) { + return instanceId + "@" + revision; + } + if (!isBlank(name)) { + return instanceId + "@" + name; + } + return instanceId; + } + + /** + * Builds a {@link WorkerDeploymentVersion} from the Cloud Run name and revision. + * + *

The name becomes the deployment name and the revision becomes the build id, so each Cloud + * Run revision maps to a distinct worker deployment version. + * + * @return a worker deployment version derived from the resolved name and revision. + * @throws IllegalStateException if the name or revision is blank, which usually means the process + * is not running on a Cloud Run worker pool or service. + */ + public WorkerDeploymentVersion workerDeploymentVersion() { + if (isBlank(name) || isBlank(revision)) { + throw new IllegalStateException( + "A Cloud Run name and revision are required to build a WorkerDeploymentVersion; " + + "this process may not be running on a Cloud Run worker pool or service"); + } + return new WorkerDeploymentVersion(name, revision); + } + + /** + * Applies the derived {@linkplain #workerIdentity() worker identity} to a workflow client options + * builder. + * + * @param builder the workflow client options builder to configure. + * @return the same builder, for chaining. + */ + public WorkflowClientOptions.Builder applyTo(WorkflowClientOptions.Builder builder) { + Objects.requireNonNull(builder, "builder"); + builder.setIdentity(workerIdentity()); + return builder; + } + + /** + * Applies the derived {@linkplain #workerDeploymentVersion() worker deployment version} to a + * worker options builder, enabling worker versioning. + * + * @param builder the worker options builder to configure. + * @return the same builder, for chaining. + * @throws IllegalStateException if the name or revision is blank, which usually means the process + * is not running on a Cloud Run worker pool or service. + */ + public WorkerOptions.Builder applyTo(WorkerOptions.Builder builder) { + Objects.requireNonNull(builder, "builder"); + builder.setDeploymentOptions( + WorkerDeploymentOptions.newBuilder() + .setUseVersioning(true) + .setVersion(workerDeploymentVersion()) + .build()); + return builder; + } + + private static String readBody(HttpURLConnection connection) throws IOException { + try (InputStream in = connection.getInputStream()) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + byte[] chunk = new byte[512]; + int read; + while ((read = in.read(chunk)) != -1) { + out.write(chunk, 0, read); + } + return new String(out.toByteArray(), StandardCharsets.UTF_8); + } + } + + private static int timeoutMillis(Duration timeout) { + long millis = timeout.toMillis(); + if (millis < 0) { + throw new IllegalArgumentException("timeout must not be negative"); + } + return (int) Math.min(millis, Integer.MAX_VALUE); + } + + private static String firstNonBlank(String first, String second) { + if (!isBlank(first)) { + return first; + } + return isBlank(second) ? null : second; + } + + private static boolean isBlank(String value) { + return value == null || value.trim().isEmpty(); + } +} diff --git a/settings.gradle b/settings.gradle index 3699ff1508..6cbf879490 100644 --- a/settings.gradle +++ b/settings.gradle @@ -15,6 +15,8 @@ include 'temporal-workflowstreams' project(':temporal-workflowstreams').projectDir = file('contrib/temporal-workflowstreams') include 'temporal-aws-lambda' project(':temporal-aws-lambda').projectDir = file('contrib/temporal-aws-lambda') +include 'temporal-gcp-cloud-run' +project(':temporal-gcp-cloud-run').projectDir = file('contrib/temporal-gcp-cloud-run') include 'temporal-spring-boot-autoconfigure' include 'temporal-spring-boot-starter' include 'temporal-remote-data-encoder' From 923be731bc0d799751ee17a6e209a49b006f0515 Mon Sep 17 00:00:00 2001 From: seanbollin Date: Tue, 25 Aug 2026 13:17:23 -0700 Subject: [PATCH 2/6] Set worker versioning behavior to PINNED in the Cloud Run worker apply The worker-side apply helper enabled versioning and set the deployment version but left the default versioning behavior unset, so a versioned worker with a plain (un-annotated) workflow failed to register. Default it to PINNED; a per-workflow versioning behavior still takes precedence. Co-Authored-By: Claude Opus 4.8 --- contrib/temporal-gcp-cloud-run/README.md | 2 +- .../java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java | 4 +++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/contrib/temporal-gcp-cloud-run/README.md b/contrib/temporal-gcp-cloud-run/README.md index a0e3da1763..c078d4fc42 100644 --- a/contrib/temporal-gcp-cloud-run/README.md +++ b/contrib/temporal-gcp-cloud-run/README.md @@ -64,7 +64,7 @@ Worker pools receive `CLOUD_RUN_WORKER_POOL` and `CLOUD_RUN_REVISION` and no `K_ `workerIdentity()` returns `@`, falling back to `@` and then to the bare `` when those values are blank. `workerDeploymentVersion()` maps the name to the deployment name and the revision to the build id, so each Cloud Run revision becomes a distinct `WorkerDeploymentVersion`. -The two `applyTo(...)` overloads mirror the SDK's "apply defaults to your options" idiom: `applyTo(WorkflowClientOptions.Builder)` sets the worker identity on the client side, and `applyTo(WorkerOptions.Builder)` sets the deployment version (with versioning enabled) on the worker side. Each returns the builder for chaining. If you prefer to read the values yourself, call `workerIdentity()` and `workerDeploymentVersion()` directly. +The two `applyTo(...)` overloads mirror the SDK's "apply defaults to your options" idiom: `applyTo(WorkflowClientOptions.Builder)` sets the worker identity on the client side, and `applyTo(WorkerOptions.Builder)` sets the deployment version (with versioning enabled, pinning workflows to this version by default via `VersioningBehavior.PINNED`; a per-workflow behavior takes precedence) on the worker side. Each returns the builder for chaining. If you prefer to read the values yourself, call `workerIdentity()` and `workerDeploymentVersion()` directly. Because the metadata server is only reachable from a Cloud Run instance, `fetch()` throws `IllegalStateException` when it cannot be reached, and `workerDeploymentVersion()` (and therefore `applyTo(WorkerOptions.Builder)`) throws `IllegalStateException` when the name or revision is not set. Use `GoogleCloudRunMetadata.fetch(String metadataUrl, Duration timeout)` to override the metadata URL or the request timeout. diff --git a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java index 40c69adecf..0edd847afe 100644 --- a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java +++ b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java @@ -2,6 +2,7 @@ import io.temporal.client.WorkflowClientOptions; import io.temporal.common.Experimental; +import io.temporal.common.VersioningBehavior; import io.temporal.common.WorkerDeploymentVersion; import io.temporal.worker.WorkerDeploymentOptions; import io.temporal.worker.WorkerOptions; @@ -199,7 +200,7 @@ public WorkflowClientOptions.Builder applyTo(WorkflowClientOptions.Builder build /** * Applies the derived {@linkplain #workerDeploymentVersion() worker deployment version} to a - * worker options builder, enabling worker versioning. + * worker options builder, enabling worker versioning with a PINNED default behavior. * * @param builder the worker options builder to configure. * @return the same builder, for chaining. @@ -212,6 +213,7 @@ public WorkerOptions.Builder applyTo(WorkerOptions.Builder builder) { WorkerDeploymentOptions.newBuilder() .setUseVersioning(true) .setVersion(workerDeploymentVersion()) + .setDefaultVersioningBehavior(VersioningBehavior.PINNED) .build()); return builder; } From 31a6d7efcbd300082eb248874a46ba8b67f8fbd4 Mon Sep 17 00:00:00 2001 From: seanbollin Date: Tue, 25 Aug 2026 13:42:58 -0700 Subject: [PATCH 3/6] Add unit tests for GoogleCloudRunMetadata Cover the Cloud Run metadata helper: environment-variable precedence (CLOUD_RUN_WORKER_POOL over K_SERVICE, CLOUD_RUN_REVISION over K_REVISION), worker identity fallbacks, WorkerDeploymentVersion mapping and its empty name/revision error, the metadata HTTP request (Metadata-Flavor: Google header, body trimming, non-200 and unreachable errors), and the applyTo(...) methods (client identity, and PINNED worker deployment versioning). The metadata request is served by an in-process com.sun.net.httpserver HttpServer and the environment lookup is injected through a new package-private fetch(String, Duration, Function) test seam, so the tests touch neither the network nor the real process environment. The seam is not part of the public API and does not change public behavior. Add the matching testImplementation dependencies (temporal-sdk, junit) to the module, mirroring the temporal-aws-lambda module. Co-Authored-By: Claude Opus 4.8 --- contrib/temporal-gcp-cloud-run/build.gradle | 5 + .../gcp/cloudrun/GoogleCloudRunMetadata.java | 22 +- .../cloudrun/GoogleCloudRunMetadataTest.java | 281 ++++++++++++++++++ 3 files changed, 306 insertions(+), 2 deletions(-) create mode 100644 contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadataTest.java diff --git a/contrib/temporal-gcp-cloud-run/build.gradle b/contrib/temporal-gcp-cloud-run/build.gradle index d08b6289f1..831e558db5 100644 --- a/contrib/temporal-gcp-cloud-run/build.gradle +++ b/contrib/temporal-gcp-cloud-run/build.gradle @@ -4,4 +4,9 @@ dependencies { // This module shouldn't carry temporal-sdk with it, especially for situations when users may // be using a shaded artifact. compileOnly project(':temporal-sdk') + + testImplementation project(':temporal-sdk') + testImplementation "junit:junit:${junitVersion}" + + testRuntimeOnly group: 'ch.qos.logback', name: 'logback-classic', version: "${logbackVersion}" } diff --git a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java index 0edd847afe..c0cd07091f 100644 --- a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java +++ b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java @@ -14,6 +14,7 @@ import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.Objects; +import java.util.function.Function; /** * Reads Google Cloud Run instance metadata and derives a Temporal worker identity and a {@link @@ -94,11 +95,28 @@ public static GoogleCloudRunMetadata fetch() { * process is not running on Google Cloud Run. */ public static GoogleCloudRunMetadata fetch(String metadataUrl, Duration timeout) { + return fetch(metadataUrl, timeout, System::getenv); + } + + /** + * Package-private test seam that injects the environment-variable lookup used to resolve the + * deployment name and revision. This lets unit tests exercise the environment-variable precedence + * and the metadata HTTP request deterministically, without depending on the real process + * environment. It is not part of the public API and must not be relied on outside of tests; use + * {@link #fetch(String, Duration)} instead. + * + * @param metadataUrl URL of the Cloud Run metadata endpoint that returns the instance id. + * @param timeout connect and read timeout applied to the metadata request. + * @param getenv environment-variable lookup, normally {@code System::getenv}. + */ + static GoogleCloudRunMetadata fetch( + String metadataUrl, Duration timeout, Function getenv) { Objects.requireNonNull(metadataUrl, "metadataUrl"); Objects.requireNonNull(timeout, "timeout"); + Objects.requireNonNull(getenv, "getenv"); - String name = firstNonBlank(System.getenv(CLOUD_RUN_WORKER_POOL), System.getenv(K_SERVICE)); - String revision = firstNonBlank(System.getenv(CLOUD_RUN_REVISION), System.getenv(K_REVISION)); + String name = firstNonBlank(getenv.apply(CLOUD_RUN_WORKER_POOL), getenv.apply(K_SERVICE)); + String revision = firstNonBlank(getenv.apply(CLOUD_RUN_REVISION), getenv.apply(K_REVISION)); HttpURLConnection connection = null; try { diff --git a/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadataTest.java b/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadataTest.java new file mode 100644 index 0000000000..fa0f3af2f6 --- /dev/null +++ b/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadataTest.java @@ -0,0 +1,281 @@ +package io.temporal.gcp.cloudrun; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import com.sun.net.httpserver.HttpServer; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.VersioningBehavior; +import io.temporal.common.WorkerDeploymentVersion; +import io.temporal.worker.WorkerDeploymentOptions; +import io.temporal.worker.WorkerOptions; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +/** + * Unit tests for {@link GoogleCloudRunMetadata}. + * + *

The metadata request is served by an in-process {@link HttpServer} and the environment lookup + * is injected through the package-private {@link GoogleCloudRunMetadata#fetch(String, Duration, + * java.util.function.Function)} test seam, so these tests touch neither the network nor the real + * process environment. + */ +public class GoogleCloudRunMetadataTest { + private static final Duration TIMEOUT = Duration.ofSeconds(2); + + private HttpServer server; + private final AtomicReference responseBody = new AtomicReference<>(""); + private final AtomicInteger responseStatus = new AtomicInteger(200); + private final AtomicReference capturedMetadataFlavor = new AtomicReference<>(); + private final AtomicReference capturedMethod = new AtomicReference<>(); + + @Before + public void startServer() throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext( + "/computeMetadata/v1/instance/id", + exchange -> { + capturedMetadataFlavor.set(exchange.getRequestHeaders().getFirst("Metadata-Flavor")); + capturedMethod.set(exchange.getRequestMethod()); + byte[] body = responseBody.get().getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(responseStatus.get(), body.length == 0 ? -1 : body.length); + try (OutputStream out = exchange.getResponseBody()) { + out.write(body); + } + }); + server.start(); + } + + @After + public void stopServer() { + server.stop(0); + } + + // --- Environment-variable precedence --- + + @Test + public void cloudRunWorkerPoolWinsOverKService() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + env.put(GoogleCloudRunMetadata.K_SERVICE, "service"); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "worker-pool-revision"); + env.put(GoogleCloudRunMetadata.K_REVISION, "service-revision"); + + GoogleCloudRunMetadata metadata = fetch(env); + + assertEquals("worker-pool", metadata.getName()); + assertEquals("worker-pool-revision", metadata.getRevision()); + } + + @Test + public void kServiceUsedWhenWorkerPoolAbsent() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.K_SERVICE, "service"); + env.put(GoogleCloudRunMetadata.K_REVISION, "service-revision"); + + GoogleCloudRunMetadata metadata = fetch(env); + + assertEquals("service", metadata.getName()); + assertEquals("service-revision", metadata.getRevision()); + } + + @Test + public void blankWorkerPoolVariablesFallThroughToKService() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, " "); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, ""); + env.put(GoogleCloudRunMetadata.K_SERVICE, "service"); + env.put(GoogleCloudRunMetadata.K_REVISION, "service-revision"); + + GoogleCloudRunMetadata metadata = fetch(env); + + assertEquals("service", metadata.getName()); + assertEquals("service-revision", metadata.getRevision()); + } + + @Test + public void nameAndRevisionAreNullWhenNoEnvSet() { + responseBody.set("instance-1"); + + GoogleCloudRunMetadata metadata = fetch(new HashMap<>()); + + assertNull(metadata.getName()); + assertNull(metadata.getRevision()); + } + + // --- Worker identity --- + + @Test + public void workerIdentityCombinesInstanceIdAndRevision() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); + + assertEquals("instance-1@revision-1", fetch(env).workerIdentity()); + } + + @Test + public void workerIdentityFallsBackToNameWhenRevisionBlank() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + + assertEquals("instance-1@worker-pool", fetch(env).workerIdentity()); + } + + @Test + public void workerIdentityFallsBackToInstanceIdWhenNameAndRevisionBlank() { + responseBody.set("instance-1"); + + assertEquals("instance-1", fetch(new HashMap<>()).workerIdentity()); + } + + // --- Worker deployment version --- + + @Test + public void workerDeploymentVersionMapsNameToDeploymentAndRevisionToBuildId() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); + + WorkerDeploymentVersion version = fetch(env).workerDeploymentVersion(); + + assertEquals("worker-pool", version.getDeploymentName()); + assertEquals("revision-1", version.getBuildId()); + } + + @Test + public void workerDeploymentVersionRequiresName() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); + + IllegalStateException e = + assertThrows(IllegalStateException.class, () -> fetch(env).workerDeploymentVersion()); + assertTrue(e.getMessage().contains("name and revision")); + } + + @Test + public void workerDeploymentVersionRequiresRevision() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + + assertThrows(IllegalStateException.class, () -> fetch(env).workerDeploymentVersion()); + } + + // --- Metadata HTTP request --- + + @Test + public void fetchSendsMetadataFlavorHeaderAndTrimsBody() { + responseBody.set(" instance-42\n"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); + + GoogleCloudRunMetadata metadata = fetch(env); + + assertEquals("instance-42", metadata.getInstanceId()); + assertEquals("Google", capturedMetadataFlavor.get()); + assertEquals("GET", capturedMethod.get()); + } + + @Test + public void fetchThrowsOnNonSuccessStatus() { + responseStatus.set(500); + responseBody.set("boom"); + + IllegalStateException e = + assertThrows(IllegalStateException.class, () -> fetch(new HashMap<>())); + assertTrue(e.getMessage().contains("metadata server")); + } + + @Test + public void fetchThrowsWhenServerUnreachable() { + String unreachableUrl = + "http://127.0.0.1:" + reserveUnusedPort() + "/computeMetadata/v1/instance/id"; + Map env = new HashMap<>(); + + assertThrows( + IllegalStateException.class, + () -> GoogleCloudRunMetadata.fetch(unreachableUrl, TIMEOUT, env::get)); + } + + // --- Apply methods --- + + @Test + public void applyToWorkflowClientOptionsSetsDerivedIdentity() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); + + // NOTE: The shared cross-SDK design and the Go and .NET helpers set the client identity only + // when it is unset, so a user-provided identity is preserved. The Java + // applyTo(WorkflowClientOptions.Builder) currently sets it unconditionally. This test asserts + // only the "identity unset" case, which holds under either behavior; the user-provided-identity + // case is intentionally not asserted here while that divergence is resolved. + WorkflowClientOptions options = fetch(env).applyTo(WorkflowClientOptions.newBuilder()).build(); + + assertEquals("instance-1@revision-1", options.getIdentity()); + } + + @Test + public void applyToWorkerOptionsEnablesPinnedVersioning() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); + + WorkerOptions options = fetch(env).applyTo(WorkerOptions.newBuilder()).build(); + + WorkerDeploymentOptions deploymentOptions = options.getDeploymentOptions(); + assertTrue(deploymentOptions.isUsingVersioning()); + assertEquals( + new WorkerDeploymentVersion("worker-pool", "revision-1"), deploymentOptions.getVersion()); + assertEquals(VersioningBehavior.PINNED, deploymentOptions.getDefaultVersioningBehavior()); + } + + @Test + public void applyToWorkerOptionsThrowsWhenDeploymentVersionCannotBeBuilt() { + responseBody.set("instance-1"); + + assertThrows( + IllegalStateException.class, + () -> fetch(new HashMap<>()).applyTo(WorkerOptions.newBuilder())); + } + + private GoogleCloudRunMetadata fetch(Map env) { + return GoogleCloudRunMetadata.fetch(metadataUrl(), TIMEOUT, env::get); + } + + private String metadataUrl() { + return "http://127.0.0.1:" + server.getAddress().getPort() + "/computeMetadata/v1/instance/id"; + } + + private static int reserveUnusedPort() { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } catch (IOException e) { + throw new RuntimeException(e); + } + } +} From b6552db6bb7a962703988a4d87f49b579639274e Mon Sep 17 00:00:00 2001 From: seanbollin Date: Tue, 25 Aug 2026 15:01:38 -0700 Subject: [PATCH 4/6] Avoid deprecated URL(String) constructor in GoogleCloudRunMetadata new URL(String) is deprecated since Java 20 and fails the SDK's -Werror build on newer JDKs (the Java 23 "Edge" CI job). Use URI.create(...).toURL() instead, the recommended non-deprecated replacement (MalformedURLException is still an IOException and stays caught). Co-Authored-By: Claude Opus 4.8 --- .../java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java index c0cd07091f..9d34bfc0e0 100644 --- a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java +++ b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java @@ -10,7 +10,7 @@ import java.io.IOException; import java.io.InputStream; import java.net.HttpURLConnection; -import java.net.URL; +import java.net.URI; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.Objects; @@ -120,7 +120,7 @@ static GoogleCloudRunMetadata fetch( HttpURLConnection connection = null; try { - connection = (HttpURLConnection) new URL(metadataUrl).openConnection(); + connection = (HttpURLConnection) URI.create(metadataUrl).toURL().openConnection(); connection.setRequestMethod("GET"); connection.setRequestProperty(METADATA_FLAVOR_HEADER, METADATA_FLAVOR_VALUE); int timeoutMillis = timeoutMillis(timeout); From b2c562497d4adbcedc2504b3638b1ad3e8d505d6 Mon Sep 17 00:00:00 2001 From: seanbollin Date: Mon, 31 Aug 2026 13:19:00 -0700 Subject: [PATCH 5/6] Re-architect Cloud Run worker-ID helper into a CloudRunPlugin Add CloudRunPlugin (extends io.temporal.common.SimplePlugin), mirroring the module's CloudRunOpenTelemetryPlugin. Registering it on the workflow client fetches Cloud Run instance metadata once at client-configure time, caches it, sets the client identity from the derived worker identity when one is not already set, and sets each worker's deployment version with worker versioning enabled and a PINNED default behavior. It fails fast with an IllegalStateException when run off Cloud Run. GoogleCloudRunMetadata keeps fetch() and its accessors but drops the two applyTo(...) overloads, whose logic now lives in the plugin hooks. Adds CloudRunPluginTest (identity set only when unset, PINNED worker deployment, off-platform fail-fast, fetch-once caching, injected metadata) and updates the README to lead with the plugin. A package-private Supplier constructor is the test seam, reusing the existing fetch(url, timeout, getenv) seam. Co-Authored-By: Claude Opus 4.8 --- contrib/temporal-gcp-cloud-run/README.md | 55 +++-- .../temporal/gcp/cloudrun/CloudRunPlugin.java | 157 ++++++++++++++ .../gcp/cloudrun/GoogleCloudRunMetadata.java | 45 +--- .../gcp/cloudrun/CloudRunPluginTest.java | 192 ++++++++++++++++++ .../cloudrun/GoogleCloudRunMetadataTest.java | 48 ----- 5 files changed, 390 insertions(+), 107 deletions(-) create mode 100644 contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/CloudRunPlugin.java create mode 100644 contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/CloudRunPluginTest.java diff --git a/contrib/temporal-gcp-cloud-run/README.md b/contrib/temporal-gcp-cloud-run/README.md index c078d4fc42..f9258618e2 100644 --- a/contrib/temporal-gcp-cloud-run/README.md +++ b/contrib/temporal-gcp-cloud-run/README.md @@ -1,47 +1,48 @@ # Temporal Google Cloud Run support -This module derives a Temporal worker identity and a `WorkerDeploymentVersion` from Google Cloud Run instance metadata, for both Cloud Run **worker pools** and Cloud Run **services**. +This module configures a Temporal worker for Google Cloud Run from instance metadata, for both Cloud Run **worker pools** and Cloud Run **services**. It derives the worker's Temporal identity and its `WorkerDeploymentVersion` from Cloud Run instance metadata, so every Cloud Run revision registers as a distinct, `PINNED` Worker Deployment Version. -Cloud Run runs a long-lived container, so there is no per-request handler to wrap. This module is a small metadata helper rather than a worker wrapper: fetch the metadata once during startup and apply it to your client and worker option builders. +The primary API is `CloudRunPlugin`. Register it once on your workflow client and it propagates to every worker created from that client, setting the client identity and the worker deployment version automatically. This mirrors the `CloudRunOpenTelemetryPlugin` in this same module. > Experimental: Google Cloud Run support is experimental and may change without notice. ## Quick start -Add `temporal-gcp-cloud-run` next to your Temporal SDK dependency, then fetch the metadata and apply it while the worker starts up: +Add `temporal-gcp-cloud-run` next to your Temporal SDK dependency, then register the plugin on the workflow client options: ```java import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowClientOptions; -import io.temporal.gcp.cloudrun.GoogleCloudRunMetadata; +import io.temporal.gcp.cloudrun.CloudRunPlugin; import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.serviceclient.WorkflowServiceStubsOptions; import io.temporal.worker.Worker; import io.temporal.worker.WorkerFactory; -import io.temporal.worker.WorkerOptions; public final class Main { public static void main(String[] args) { - // Read Cloud Run instance metadata once during startup. - GoogleCloudRunMetadata metadata = GoogleCloudRunMetadata.fetch(); - WorkflowServiceStubs service = WorkflowServiceStubs.newServiceStubs( WorkflowServiceStubsOptions.newBuilder() .setTarget("my-namespace.tmprl.cloud:7233") .build()); - // applyTo(...) sets the derived worker identity on the client options. + // Registering the plugin on the client: + // - reads Cloud Run instance metadata once while the client is configured, and + // - sets the client identity to the derived worker identity (unless you set one yourself). WorkflowClient client = WorkflowClient.newInstance( - service, metadata.applyTo(WorkflowClientOptions.newBuilder()).build()); + service, + WorkflowClientOptions.newBuilder() + .setNamespace("my-namespace") + .setPlugins(new CloudRunPlugin()) + .build()); WorkerFactory factory = WorkerFactory.newInstance(client); - // applyTo(...) sets the deployment version and enables worker versioning on the worker options. - WorkerOptions workerOptions = metadata.applyTo(WorkerOptions.newBuilder()).build(); - - Worker worker = factory.newWorker("orders", workerOptions); + // The plugin propagates from the client to workers and sets each worker's deployment version + // (with worker versioning enabled and a PINNED default behavior). No per-worker wiring needed. + Worker worker = factory.newWorker("orders"); worker.registerWorkflowImplementationTypes(OrderWorkflowImpl.class); worker.registerActivitiesImplementations(new OrderActivitiesImpl()); @@ -50,11 +51,11 @@ public final class Main { } ``` -Both `applyTo(...)` methods return the builder they were given, so they compose with the rest of your builder configuration. +You can also register the plugin on `WorkflowServiceStubsOptions.Builder.setPlugins(...)`; from there it propagates to the client and workers as well. ## How it works -`GoogleCloudRunMetadata.fetch()` resolves three values: +`CloudRunPlugin` reads Cloud Run instance metadata through `GoogleCloudRunMetadata`, which resolves three values: - **name** (the Temporal deployment name): the first non-empty of `CLOUD_RUN_WORKER_POOL` (set on Cloud Run worker pools) then `K_SERVICE` (set on Cloud Run services). - **revision**: the first non-empty of `CLOUD_RUN_REVISION` (worker pools) then `K_REVISION` (services). @@ -62,10 +63,26 @@ Both `applyTo(...)` methods return the builder they were given, so they compose Worker pools receive `CLOUD_RUN_WORKER_POOL` and `CLOUD_RUN_REVISION` and no `K_*` variables, while services receive `K_SERVICE` and `K_REVISION`, so resolving each value from the worker-pool variable first and the service variable second supports both. -`workerIdentity()` returns `@`, falling back to `@` and then to the bare `` when those values are blank. `workerDeploymentVersion()` maps the name to the deployment name and the revision to the build id, so each Cloud Run revision becomes a distinct `WorkerDeploymentVersion`. +The plugin then applies the metadata through the SDK's plugin hooks: + +- **Client** (`configureWorkflowClient`): sets the client identity to `@` (falling back to `@` and then the bare ``), but only when you have not already set an identity, so a user-provided identity always wins. The metadata is fetched here, once, and cached. +- **Worker** (`configureWorker`): sets the worker deployment version — the name becomes the deployment name and the revision becomes the build id — with worker versioning enabled and `VersioningBehavior.PINNED` as the default, so in-flight workflows stay on the Cloud Run revision that started them (a per-workflow `@WorkflowVersioningBehavior` takes precedence). + +Because the metadata server is only reachable from a Cloud Run instance, the plugin **fails fast**: the fetch in `configureWorkflowClient` throws `IllegalStateException` when the metadata server cannot be reached (which usually means the process is not running on Google Cloud Run), and `configureWorker` throws `IllegalStateException` when the name or revision is not set (which usually means the process is not running on a Cloud Run worker pool or service). The plugin does not silently no-op off-platform. -The two `applyTo(...)` overloads mirror the SDK's "apply defaults to your options" idiom: `applyTo(WorkflowClientOptions.Builder)` sets the worker identity on the client side, and `applyTo(WorkerOptions.Builder)` sets the deployment version (with versioning enabled, pinning workflows to this version by default via `VersioningBehavior.PINNED`; a per-workflow behavior takes precedence) on the worker side. Each returns the builder for chaining. If you prefer to read the values yourself, call `workerIdentity()` and `workerDeploymentVersion()` directly. +## Reading the metadata directly + +If you prefer to read the values yourself, or to fetch the metadata once and pass it in, use `GoogleCloudRunMetadata` directly: + +```java +GoogleCloudRunMetadata metadata = GoogleCloudRunMetadata.fetch(); +String identity = metadata.workerIdentity(); +WorkerDeploymentVersion version = metadata.workerDeploymentVersion(); + +// Or hand the already-fetched metadata to the plugin to skip its own fetch: +CloudRunPlugin plugin = new CloudRunPlugin(metadata); +``` -Because the metadata server is only reachable from a Cloud Run instance, `fetch()` throws `IllegalStateException` when it cannot be reached, and `workerDeploymentVersion()` (and therefore `applyTo(WorkerOptions.Builder)`) throws `IllegalStateException` when the name or revision is not set. Use `GoogleCloudRunMetadata.fetch(String metadataUrl, Duration timeout)` to override the metadata URL or the request timeout. +`GoogleCloudRunMetadata.fetch(String metadataUrl, Duration timeout)` overrides the metadata URL or the request timeout. This module depends only on the Temporal SDK at compile time and uses the JDK's `HttpURLConnection` for the metadata request, so it adds no additional runtime dependencies. diff --git a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/CloudRunPlugin.java b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/CloudRunPlugin.java new file mode 100644 index 0000000000..ed47dcaf2c --- /dev/null +++ b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/CloudRunPlugin.java @@ -0,0 +1,157 @@ +package io.temporal.gcp.cloudrun; + +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.Experimental; +import io.temporal.common.SimplePlugin; +import io.temporal.common.VersioningBehavior; +import io.temporal.worker.WorkerDeploymentOptions; +import io.temporal.worker.WorkerOptions; +import java.util.Objects; +import java.util.function.Supplier; + +/** + * Plugin that configures a Temporal worker for Google Cloud Run from instance metadata, for both + * Cloud Run worker pools and Cloud Run services. + * + *

Register the plugin once on the workflow client and it propagates to every worker created from + * that client. It reads {@link GoogleCloudRunMetadata Cloud Run instance metadata} once while the + * client is configured, caches it, and then: + * + *

    + *
  • sets the workflow client identity to the {@linkplain + * GoogleCloudRunMetadata#workerIdentity() derived worker identity}, but only when the caller + * has not already set an identity (a user-provided identity always wins); + *
  • sets each worker's {@link WorkerDeploymentOptions} to the {@linkplain + * GoogleCloudRunMetadata#workerDeploymentVersion() derived deployment version} with worker + * versioning enabled and a {@link VersioningBehavior#PINNED PINNED} default behavior, so + * in-flight workflows stay on the Cloud Run revision that started them. + *
+ * + *

The metadata is fetched lazily at client-configure time rather than in the constructor, + * because the fetch performs a network request to the Cloud Run metadata server that belongs at + * connect time. The metadata server is only reachable from a Cloud Run instance, so the fetch fails + * fast with an {@link IllegalStateException} when this process is not running on Cloud Run. + * + *

Register the plugin with {@link WorkflowClientOptions.Builder#setPlugins}: + * + * 

{@code
+ * WorkflowClient client =
+ *     WorkflowClient.newInstance(
+ *         service,
+ *         WorkflowClientOptions.newBuilder()
+ *             .setNamespace(namespace)
+ *             .setPlugins(new CloudRunPlugin())
+ *             .build());
+ *
+ * WorkerFactory factory = WorkerFactory.newInstance(client);
+ * Worker worker = factory.newWorker("my-task-queue");
+ * }
+ * + *

Advanced / testing: {@link #CloudRunPlugin(GoogleCloudRunMetadata)} accepts an + * already-resolved {@link GoogleCloudRunMetadata} instance, which skips the lazy fetch entirely. + * This is useful when the application fetches the metadata itself (for example to log it) or when a + * test injects fixed metadata. + * + *

Experimental: Google Cloud Run support is experimental and may change without notice. + */ +@Experimental +public final class CloudRunPlugin extends SimplePlugin { + /** Unique plugin name, used for logging and duplicate detection. */ + public static final String NAME = "io.temporal.gcp.cloudrun"; + + private final Supplier metadataSupplier; + private volatile GoogleCloudRunMetadata metadata; + + /** + * Creates a plugin that fetches Cloud Run instance metadata from the {@linkplain + * GoogleCloudRunMetadata#DEFAULT_METADATA_URL default metadata server} while the workflow client + * is configured. + */ + public CloudRunPlugin() { + this(GoogleCloudRunMetadata::fetch); + } + + /** + * Creates a plugin that uses an already-resolved {@link GoogleCloudRunMetadata} instance instead + * of fetching it. No request is made to the Cloud Run metadata server. + * + * @param metadata previously fetched Cloud Run instance metadata. + */ + public CloudRunPlugin(GoogleCloudRunMetadata metadata) { + this(pinnedSupplier(metadata)); + } + + /** + * Package-private test seam that supplies the {@link GoogleCloudRunMetadata} lazily. It lets unit + * tests point the fetch at an in-process metadata server and injected environment through the + * {@link GoogleCloudRunMetadata#fetch(String, java.time.Duration, java.util.function.Function)} + * seam, and to exercise the off-platform fail-fast path. It is not part of the public API; use + * {@link #CloudRunPlugin()} or {@link #CloudRunPlugin(GoogleCloudRunMetadata)} instead. + * + * @param metadataSupplier supplier invoked once, at client-configure time, to resolve the + * metadata. + */ + CloudRunPlugin(Supplier metadataSupplier) { + super(NAME); + this.metadataSupplier = Objects.requireNonNull(metadataSupplier, "metadataSupplier"); + } + + /** + * Fetches (once) and caches the Cloud Run instance metadata, then sets the derived worker + * identity on the client options when the caller has not already set an identity. + * + * @param builder the workflow client options builder to configure. + * @throws IllegalStateException if the Cloud Run metadata server cannot be reached, which usually + * means this process is not running on Google Cloud Run. + */ + @Override + public void configureWorkflowClient(WorkflowClientOptions.Builder builder) { + GoogleCloudRunMetadata resolved = metadata(); + if (isBlank(builder.build().getIdentity())) { + builder.setIdentity(resolved.workerIdentity()); + } + } + + /** + * Sets the worker's {@link WorkerDeploymentOptions} from the cached Cloud Run metadata, enabling + * worker versioning with a {@link VersioningBehavior#PINNED PINNED} default behavior. + * + * @param taskQueue the task queue name for the worker being created. + * @param builder the worker options builder to configure. + * @throws IllegalStateException if the Cloud Run name or revision is not set, which usually means + * this process is not running on a Cloud Run worker pool or service. + */ + @Override + public void configureWorker(String taskQueue, WorkerOptions.Builder builder) { + GoogleCloudRunMetadata resolved = metadata(); + builder.setDeploymentOptions( + WorkerDeploymentOptions.newBuilder() + .setUseVersioning(true) + .setVersion(resolved.workerDeploymentVersion()) + .setDefaultVersioningBehavior(VersioningBehavior.PINNED) + .build()); + } + + private GoogleCloudRunMetadata metadata() { + GoogleCloudRunMetadata local = metadata; + if (local == null) { + synchronized (this) { + local = metadata; + if (local == null) { + local = Objects.requireNonNull(metadataSupplier.get(), "Cloud Run metadata"); + metadata = local; + } + } + } + return local; + } + + private static Supplier pinnedSupplier(GoogleCloudRunMetadata metadata) { + Objects.requireNonNull(metadata, "metadata"); + return () -> metadata; + } + + private static boolean isBlank(String value) { + return value == null || value.trim().isEmpty(); + } +} diff --git a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java index 9d34bfc0e0..83b1079aaa 100644 --- a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java +++ b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java @@ -1,11 +1,7 @@ package io.temporal.gcp.cloudrun; -import io.temporal.client.WorkflowClientOptions; import io.temporal.common.Experimental; -import io.temporal.common.VersioningBehavior; import io.temporal.common.WorkerDeploymentVersion; -import io.temporal.worker.WorkerDeploymentOptions; -import io.temporal.worker.WorkerOptions; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; @@ -21,9 +17,11 @@ * WorkerDeploymentVersion} from it. * *

Cloud Run runs a long-lived container rather than a per-request handler, so this class is a - * metadata helper rather than a worker wrapper. Fetch the metadata once while a worker starts up, - * then apply it to your client and worker option builders with {@link - * #applyTo(WorkflowClientOptions.Builder)} and {@link #applyTo(WorkerOptions.Builder)}. + * metadata helper rather than a worker wrapper. Most applications register {@link CloudRunPlugin} + * on their workflow client instead of using this class directly; the plugin fetches this metadata + * and applies the derived identity and deployment version to the client and workers. Use this class + * directly to read the {@linkplain #workerIdentity() worker identity} or {@linkplain + * #workerDeploymentVersion() worker deployment version} yourself. * *

The deployment name and revision are resolved from environment variables Cloud Run injects * into every instance. Cloud Run worker pools set {@code CLOUD_RUN_WORKER_POOL} and {@code @@ -203,39 +201,6 @@ public WorkerDeploymentVersion workerDeploymentVersion() { return new WorkerDeploymentVersion(name, revision); } - /** - * Applies the derived {@linkplain #workerIdentity() worker identity} to a workflow client options - * builder. - * - * @param builder the workflow client options builder to configure. - * @return the same builder, for chaining. - */ - public WorkflowClientOptions.Builder applyTo(WorkflowClientOptions.Builder builder) { - Objects.requireNonNull(builder, "builder"); - builder.setIdentity(workerIdentity()); - return builder; - } - - /** - * Applies the derived {@linkplain #workerDeploymentVersion() worker deployment version} to a - * worker options builder, enabling worker versioning with a PINNED default behavior. - * - * @param builder the worker options builder to configure. - * @return the same builder, for chaining. - * @throws IllegalStateException if the name or revision is blank, which usually means the process - * is not running on a Cloud Run worker pool or service. - */ - public WorkerOptions.Builder applyTo(WorkerOptions.Builder builder) { - Objects.requireNonNull(builder, "builder"); - builder.setDeploymentOptions( - WorkerDeploymentOptions.newBuilder() - .setUseVersioning(true) - .setVersion(workerDeploymentVersion()) - .setDefaultVersioningBehavior(VersioningBehavior.PINNED) - .build()); - return builder; - } - private static String readBody(HttpURLConnection connection) throws IOException { try (InputStream in = connection.getInputStream()) { ByteArrayOutputStream out = new ByteArrayOutputStream(); diff --git a/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/CloudRunPluginTest.java b/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/CloudRunPluginTest.java new file mode 100644 index 0000000000..54bb0ed55b --- /dev/null +++ b/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/CloudRunPluginTest.java @@ -0,0 +1,192 @@ +package io.temporal.gcp.cloudrun; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import com.sun.net.httpserver.HttpServer; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.VersioningBehavior; +import io.temporal.common.WorkerDeploymentVersion; +import io.temporal.worker.WorkerDeploymentOptions; +import io.temporal.worker.WorkerOptions; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +/** + * Unit tests for {@link CloudRunPlugin}. + * + *

The metadata request is served by an in-process {@link HttpServer} and the environment lookup + * is injected through the {@link GoogleCloudRunMetadata#fetch(String, Duration, + * java.util.function.Function)} test seam, so these tests touch neither the network nor the real + * process environment. The plugin's package-private {@link CloudRunPlugin#CloudRunPlugin(Supplier)} + * seam lets each test point the plugin at that in-process server (or at an unreachable address, to + * exercise the off-platform fail-fast path). + */ +public class CloudRunPluginTest { + private static final Duration TIMEOUT = Duration.ofSeconds(2); + + private HttpServer server; + private final AtomicReference responseBody = new AtomicReference<>(""); + + @Before + public void startServer() throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext( + "/computeMetadata/v1/instance/id", + exchange -> { + byte[] body = responseBody.get().getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, body.length == 0 ? -1 : body.length); + try (OutputStream out = exchange.getResponseBody()) { + out.write(body); + } + }); + server.start(); + } + + @After + public void stopServer() { + server.stop(0); + } + + @Test + public void configureWorkflowClientSetsDerivedIdentityWhenUnset() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); + + WorkflowClientOptions.Builder builder = WorkflowClientOptions.newBuilder(); + pluginFor(env).configureWorkflowClient(builder); + + assertEquals("instance-1@revision-1", builder.build().getIdentity()); + } + + @Test + public void configureWorkflowClientPreservesUserProvidedIdentity() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); + + WorkflowClientOptions.Builder builder = + WorkflowClientOptions.newBuilder().setIdentity("user-set"); + pluginFor(env).configureWorkflowClient(builder); + + assertEquals("user-set", builder.build().getIdentity()); + } + + @Test + public void configureWorkerEnablesPinnedVersioning() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); + + WorkerOptions.Builder builder = WorkerOptions.newBuilder(); + pluginFor(env).configureWorker("orders", builder); + + WorkerDeploymentOptions deploymentOptions = builder.build().getDeploymentOptions(); + assertTrue(deploymentOptions.isUsingVersioning()); + assertEquals( + new WorkerDeploymentVersion("worker-pool", "revision-1"), deploymentOptions.getVersion()); + assertEquals(VersioningBehavior.PINNED, deploymentOptions.getDefaultVersioningBehavior()); + } + + @Test + public void configureWorkflowClientFailsFastOffCloudRun() { + String unreachableUrl = + "http://127.0.0.1:" + reserveUnusedPort() + "/computeMetadata/v1/instance/id"; + CloudRunPlugin plugin = + new CloudRunPlugin( + () -> GoogleCloudRunMetadata.fetch(unreachableUrl, TIMEOUT, name -> null)); + + IllegalStateException e = + assertThrows( + IllegalStateException.class, + () -> plugin.configureWorkflowClient(WorkflowClientOptions.newBuilder())); + assertTrue(e.getMessage().contains("metadata server")); + } + + @Test + public void configureWorkerFailsFastWhenNotWorkerPoolOrService() { + responseBody.set("instance-1"); + + // Metadata server is reachable (instance id is present) but no name/revision env is set, so the + // deployment version cannot be built. This is the "on some other platform" case. + CloudRunPlugin plugin = new CloudRunPlugin(metadata(new HashMap<>())); + + assertThrows( + IllegalStateException.class, + () -> plugin.configureWorker("orders", WorkerOptions.newBuilder())); + } + + @Test + public void metadataIsFetchedOnceAndSharedByBothHooks() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); + + GoogleCloudRunMetadata resolved = metadata(env); + AtomicInteger supplierCalls = new AtomicInteger(); + Supplier countingSupplier = + () -> { + supplierCalls.incrementAndGet(); + return resolved; + }; + CloudRunPlugin plugin = new CloudRunPlugin(countingSupplier); + + plugin.configureWorkflowClient(WorkflowClientOptions.newBuilder()); + plugin.configureWorker("orders", WorkerOptions.newBuilder()); + + assertEquals(1, supplierCalls.get()); + } + + @Test + public void injectedMetadataIsUsedWithoutFetching() { + responseBody.set("instance-1"); + Map env = new HashMap<>(); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); + env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); + + CloudRunPlugin plugin = new CloudRunPlugin(metadata(env)); + + WorkflowClientOptions.Builder builder = WorkflowClientOptions.newBuilder(); + plugin.configureWorkflowClient(builder); + + assertEquals("instance-1@revision-1", builder.build().getIdentity()); + } + + private CloudRunPlugin pluginFor(Map env) { + return new CloudRunPlugin(() -> metadata(env)); + } + + private GoogleCloudRunMetadata metadata(Map env) { + return GoogleCloudRunMetadata.fetch(metadataUrl(), TIMEOUT, env::get); + } + + private String metadataUrl() { + return "http://127.0.0.1:" + server.getAddress().getPort() + "/computeMetadata/v1/instance/id"; + } + + private static int reserveUnusedPort() { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } catch (IOException e) { + throw new RuntimeException(e); + } + } +} diff --git a/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadataTest.java b/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadataTest.java index fa0f3af2f6..6ae907c6ed 100644 --- a/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadataTest.java +++ b/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadataTest.java @@ -6,11 +6,7 @@ import static org.junit.Assert.assertTrue; import com.sun.net.httpserver.HttpServer; -import io.temporal.client.WorkflowClientOptions; -import io.temporal.common.VersioningBehavior; import io.temporal.common.WorkerDeploymentVersion; -import io.temporal.worker.WorkerDeploymentOptions; -import io.temporal.worker.WorkerOptions; import java.io.IOException; import java.io.OutputStream; import java.net.InetSocketAddress; @@ -219,50 +215,6 @@ public void fetchThrowsWhenServerUnreachable() { () -> GoogleCloudRunMetadata.fetch(unreachableUrl, TIMEOUT, env::get)); } - // --- Apply methods --- - - @Test - public void applyToWorkflowClientOptionsSetsDerivedIdentity() { - responseBody.set("instance-1"); - Map env = new HashMap<>(); - env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); - env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); - - // NOTE: The shared cross-SDK design and the Go and .NET helpers set the client identity only - // when it is unset, so a user-provided identity is preserved. The Java - // applyTo(WorkflowClientOptions.Builder) currently sets it unconditionally. This test asserts - // only the "identity unset" case, which holds under either behavior; the user-provided-identity - // case is intentionally not asserted here while that divergence is resolved. - WorkflowClientOptions options = fetch(env).applyTo(WorkflowClientOptions.newBuilder()).build(); - - assertEquals("instance-1@revision-1", options.getIdentity()); - } - - @Test - public void applyToWorkerOptionsEnablesPinnedVersioning() { - responseBody.set("instance-1"); - Map env = new HashMap<>(); - env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); - env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); - - WorkerOptions options = fetch(env).applyTo(WorkerOptions.newBuilder()).build(); - - WorkerDeploymentOptions deploymentOptions = options.getDeploymentOptions(); - assertTrue(deploymentOptions.isUsingVersioning()); - assertEquals( - new WorkerDeploymentVersion("worker-pool", "revision-1"), deploymentOptions.getVersion()); - assertEquals(VersioningBehavior.PINNED, deploymentOptions.getDefaultVersioningBehavior()); - } - - @Test - public void applyToWorkerOptionsThrowsWhenDeploymentVersionCannotBeBuilt() { - responseBody.set("instance-1"); - - assertThrows( - IllegalStateException.class, - () -> fetch(new HashMap<>()).applyTo(WorkerOptions.newBuilder())); - } - private GoogleCloudRunMetadata fetch(Map env) { return GoogleCloudRunMetadata.fetch(metadataUrl(), TIMEOUT, env::get); } From cf0f4df473b68e8a5dfce2031bedcb3a3d305aee Mon Sep 17 00:00:00 2001 From: seanbollin Date: Mon, 31 Aug 2026 15:41:07 -0700 Subject: [PATCH 6/6] Rename CloudRunPlugin to WorkerIdPlugin Cloud Run can host multiple Temporal plugins (a worker-ID plugin and an OpenTelemetry plugin) in the same module, so the worker-ID plugin must not claim the generic CloudRunPlugin name. Rename the class and file to WorkerIdPlugin, change the NAME id to io.temporal.gcp.cloudrun.workerid so it does not collide under duplicate detection, and update the test (WorkerIdPluginTest), the README, and the GoogleCloudRunMetadata doc link. The io.temporal.gcp.cloudrun package and GoogleCloudRunMetadata are unchanged. Co-Authored-By: Claude Opus 4.8 --- contrib/temporal-gcp-cloud-run/README.md | 10 +++++----- .../gcp/cloudrun/GoogleCloudRunMetadata.java | 2 +- ...loudRunPlugin.java => WorkerIdPlugin.java} | 16 +++++++-------- ...luginTest.java => WorkerIdPluginTest.java} | 20 +++++++++---------- 4 files changed, 24 insertions(+), 24 deletions(-) rename contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/{CloudRunPlugin.java => WorkerIdPlugin.java} (93%) rename contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/{CloudRunPluginTest.java => WorkerIdPluginTest.java} (92%) diff --git a/contrib/temporal-gcp-cloud-run/README.md b/contrib/temporal-gcp-cloud-run/README.md index f9258618e2..c3e1620117 100644 --- a/contrib/temporal-gcp-cloud-run/README.md +++ b/contrib/temporal-gcp-cloud-run/README.md @@ -2,7 +2,7 @@ This module configures a Temporal worker for Google Cloud Run from instance metadata, for both Cloud Run **worker pools** and Cloud Run **services**. It derives the worker's Temporal identity and its `WorkerDeploymentVersion` from Cloud Run instance metadata, so every Cloud Run revision registers as a distinct, `PINNED` Worker Deployment Version. -The primary API is `CloudRunPlugin`. Register it once on your workflow client and it propagates to every worker created from that client, setting the client identity and the worker deployment version automatically. This mirrors the `CloudRunOpenTelemetryPlugin` in this same module. +The primary API is `WorkerIdPlugin`. Register it once on your workflow client and it propagates to every worker created from that client, setting the client identity and the worker deployment version automatically. This mirrors the `CloudRunOpenTelemetryPlugin` in this same module. > Experimental: Google Cloud Run support is experimental and may change without notice. @@ -13,7 +13,7 @@ Add `temporal-gcp-cloud-run` next to your Temporal SDK dependency, then register ```java import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowClientOptions; -import io.temporal.gcp.cloudrun.CloudRunPlugin; +import io.temporal.gcp.cloudrun.WorkerIdPlugin; import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.serviceclient.WorkflowServiceStubsOptions; import io.temporal.worker.Worker; @@ -35,7 +35,7 @@ public final class Main { service, WorkflowClientOptions.newBuilder() .setNamespace("my-namespace") - .setPlugins(new CloudRunPlugin()) + .setPlugins(new WorkerIdPlugin()) .build()); WorkerFactory factory = WorkerFactory.newInstance(client); @@ -55,7 +55,7 @@ You can also register the plugin on `WorkflowServiceStubsOptions.Builder.setPlug ## How it works -`CloudRunPlugin` reads Cloud Run instance metadata through `GoogleCloudRunMetadata`, which resolves three values: +`WorkerIdPlugin` reads Cloud Run instance metadata through `GoogleCloudRunMetadata`, which resolves three values: - **name** (the Temporal deployment name): the first non-empty of `CLOUD_RUN_WORKER_POOL` (set on Cloud Run worker pools) then `K_SERVICE` (set on Cloud Run services). - **revision**: the first non-empty of `CLOUD_RUN_REVISION` (worker pools) then `K_REVISION` (services). @@ -80,7 +80,7 @@ String identity = metadata.workerIdentity(); WorkerDeploymentVersion version = metadata.workerDeploymentVersion(); // Or hand the already-fetched metadata to the plugin to skip its own fetch: -CloudRunPlugin plugin = new CloudRunPlugin(metadata); +WorkerIdPlugin plugin = new WorkerIdPlugin(metadata); ``` `GoogleCloudRunMetadata.fetch(String metadataUrl, Duration timeout)` overrides the metadata URL or the request timeout. diff --git a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java index 83b1079aaa..a6afb93795 100644 --- a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java +++ b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/GoogleCloudRunMetadata.java @@ -17,7 +17,7 @@ * WorkerDeploymentVersion} from it. * *

Cloud Run runs a long-lived container rather than a per-request handler, so this class is a - * metadata helper rather than a worker wrapper. Most applications register {@link CloudRunPlugin} + * metadata helper rather than a worker wrapper. Most applications register {@link WorkerIdPlugin} * on their workflow client instead of using this class directly; the plugin fetches this metadata * and applies the derived identity and deployment version to the client and workers. Use this class * directly to read the {@linkplain #workerIdentity() worker identity} or {@linkplain diff --git a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/CloudRunPlugin.java b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/WorkerIdPlugin.java similarity index 93% rename from contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/CloudRunPlugin.java rename to contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/WorkerIdPlugin.java index ed47dcaf2c..e6e4e1115c 100644 --- a/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/CloudRunPlugin.java +++ b/contrib/temporal-gcp-cloud-run/src/main/java/io/temporal/gcp/cloudrun/WorkerIdPlugin.java @@ -40,14 +40,14 @@ * service, * WorkflowClientOptions.newBuilder() * .setNamespace(namespace) - * .setPlugins(new CloudRunPlugin()) + * .setPlugins(new WorkerIdPlugin()) * .build()); * * WorkerFactory factory = WorkerFactory.newInstance(client); * Worker worker = factory.newWorker("my-task-queue"); * } * - *

Advanced / testing: {@link #CloudRunPlugin(GoogleCloudRunMetadata)} accepts an + *

Advanced / testing: {@link #WorkerIdPlugin(GoogleCloudRunMetadata)} accepts an * already-resolved {@link GoogleCloudRunMetadata} instance, which skips the lazy fetch entirely. * This is useful when the application fetches the metadata itself (for example to log it) or when a * test injects fixed metadata. @@ -55,9 +55,9 @@ *

Experimental: Google Cloud Run support is experimental and may change without notice. */ @Experimental -public final class CloudRunPlugin extends SimplePlugin { +public final class WorkerIdPlugin extends SimplePlugin { /** Unique plugin name, used for logging and duplicate detection. */ - public static final String NAME = "io.temporal.gcp.cloudrun"; + public static final String NAME = "io.temporal.gcp.cloudrun.workerid"; private final Supplier metadataSupplier; private volatile GoogleCloudRunMetadata metadata; @@ -67,7 +67,7 @@ public final class CloudRunPlugin extends SimplePlugin { * GoogleCloudRunMetadata#DEFAULT_METADATA_URL default metadata server} while the workflow client * is configured. */ - public CloudRunPlugin() { + public WorkerIdPlugin() { this(GoogleCloudRunMetadata::fetch); } @@ -77,7 +77,7 @@ public CloudRunPlugin() { * * @param metadata previously fetched Cloud Run instance metadata. */ - public CloudRunPlugin(GoogleCloudRunMetadata metadata) { + public WorkerIdPlugin(GoogleCloudRunMetadata metadata) { this(pinnedSupplier(metadata)); } @@ -86,12 +86,12 @@ public CloudRunPlugin(GoogleCloudRunMetadata metadata) { * tests point the fetch at an in-process metadata server and injected environment through the * {@link GoogleCloudRunMetadata#fetch(String, java.time.Duration, java.util.function.Function)} * seam, and to exercise the off-platform fail-fast path. It is not part of the public API; use - * {@link #CloudRunPlugin()} or {@link #CloudRunPlugin(GoogleCloudRunMetadata)} instead. + * {@link #WorkerIdPlugin()} or {@link #WorkerIdPlugin(GoogleCloudRunMetadata)} instead. * * @param metadataSupplier supplier invoked once, at client-configure time, to resolve the * metadata. */ - CloudRunPlugin(Supplier metadataSupplier) { + WorkerIdPlugin(Supplier metadataSupplier) { super(NAME); this.metadataSupplier = Objects.requireNonNull(metadataSupplier, "metadataSupplier"); } diff --git a/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/CloudRunPluginTest.java b/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/WorkerIdPluginTest.java similarity index 92% rename from contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/CloudRunPluginTest.java rename to contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/WorkerIdPluginTest.java index 54bb0ed55b..ab6e148e09 100644 --- a/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/CloudRunPluginTest.java +++ b/contrib/temporal-gcp-cloud-run/src/test/java/io/temporal/gcp/cloudrun/WorkerIdPluginTest.java @@ -26,16 +26,16 @@ import org.junit.Test; /** - * Unit tests for {@link CloudRunPlugin}. + * Unit tests for {@link WorkerIdPlugin}. * *

The metadata request is served by an in-process {@link HttpServer} and the environment lookup * is injected through the {@link GoogleCloudRunMetadata#fetch(String, Duration, * java.util.function.Function)} test seam, so these tests touch neither the network nor the real - * process environment. The plugin's package-private {@link CloudRunPlugin#CloudRunPlugin(Supplier)} + * process environment. The plugin's package-private {@link WorkerIdPlugin#WorkerIdPlugin(Supplier)} * seam lets each test point the plugin at that in-process server (or at an unreachable address, to * exercise the off-platform fail-fast path). */ -public class CloudRunPluginTest { +public class WorkerIdPluginTest { private static final Duration TIMEOUT = Duration.ofSeconds(2); private HttpServer server; @@ -109,8 +109,8 @@ public void configureWorkerEnablesPinnedVersioning() { public void configureWorkflowClientFailsFastOffCloudRun() { String unreachableUrl = "http://127.0.0.1:" + reserveUnusedPort() + "/computeMetadata/v1/instance/id"; - CloudRunPlugin plugin = - new CloudRunPlugin( + WorkerIdPlugin plugin = + new WorkerIdPlugin( () -> GoogleCloudRunMetadata.fetch(unreachableUrl, TIMEOUT, name -> null)); IllegalStateException e = @@ -126,7 +126,7 @@ public void configureWorkerFailsFastWhenNotWorkerPoolOrService() { // Metadata server is reachable (instance id is present) but no name/revision env is set, so the // deployment version cannot be built. This is the "on some other platform" case. - CloudRunPlugin plugin = new CloudRunPlugin(metadata(new HashMap<>())); + WorkerIdPlugin plugin = new WorkerIdPlugin(metadata(new HashMap<>())); assertThrows( IllegalStateException.class, @@ -147,7 +147,7 @@ public void metadataIsFetchedOnceAndSharedByBothHooks() { supplierCalls.incrementAndGet(); return resolved; }; - CloudRunPlugin plugin = new CloudRunPlugin(countingSupplier); + WorkerIdPlugin plugin = new WorkerIdPlugin(countingSupplier); plugin.configureWorkflowClient(WorkflowClientOptions.newBuilder()); plugin.configureWorker("orders", WorkerOptions.newBuilder()); @@ -162,7 +162,7 @@ public void injectedMetadataIsUsedWithoutFetching() { env.put(GoogleCloudRunMetadata.CLOUD_RUN_WORKER_POOL, "worker-pool"); env.put(GoogleCloudRunMetadata.CLOUD_RUN_REVISION, "revision-1"); - CloudRunPlugin plugin = new CloudRunPlugin(metadata(env)); + WorkerIdPlugin plugin = new WorkerIdPlugin(metadata(env)); WorkflowClientOptions.Builder builder = WorkflowClientOptions.newBuilder(); plugin.configureWorkflowClient(builder); @@ -170,8 +170,8 @@ public void injectedMetadataIsUsedWithoutFetching() { assertEquals("instance-1@revision-1", builder.build().getIdentity()); } - private CloudRunPlugin pluginFor(Map env) { - return new CloudRunPlugin(() -> metadata(env)); + private WorkerIdPlugin pluginFor(Map env) { + return new WorkerIdPlugin(() -> metadata(env)); } private GoogleCloudRunMetadata metadata(Map env) {