Add Google Cloud Run worker identity/deployment helper

contrib/temporal-gcp-cloud-run-worker-id/README.md

@@ -0,0 +1,88 @@
+# Temporal Google Cloud Run worker identity support
+
+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 `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 the companion `temporal-gcp-cloud-run` module.
+
+> Experimental: Google Cloud Run support is experimental and may change without notice.
+
+## Quick start
+
+Add `temporal-gcp-cloud-run-worker-id` 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.workerid.WorkerIdPlugin;
+import io.temporal.serviceclient.WorkflowServiceStubs;
+import io.temporal.serviceclient.WorkflowServiceStubsOptions;
+import io.temporal.worker.Worker;
+import io.temporal.worker.WorkerFactory;
+
+public final class Main {
+  public static void main(String[] args) {
+    WorkflowServiceStubs service =
+        WorkflowServiceStubs.newServiceStubs(
+            WorkflowServiceStubsOptions.newBuilder()
+                .setTarget("my-namespace.tmprl.cloud:7233")
+                .build());
+
+    // 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,
+            WorkflowClientOptions.newBuilder()
+                .setNamespace("my-namespace")
+                .setPlugins(new WorkerIdPlugin())
+                .build());
+
+    WorkerFactory factory = WorkerFactory.newInstance(client);
+
+    // 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());
+
+    factory.start();
+  }
+}
+```
+
+You can also register the plugin on `WorkflowServiceStubsOptions.Builder.setPlugins(...)`; from there it propagates to the client and workers as well.
+
+## How it works
+
+`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).
+- **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.
+
+The plugin then applies the metadata through the SDK's plugin hooks:
+
+- **Client** (`configureWorkflowClient`): sets the client identity to `<instanceId>@<revision>` (falling back to `<instanceId>@<name>` and then the bare `<instanceId>`), 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.
+
+## 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:
+WorkerIdPlugin plugin = new WorkerIdPlugin(metadata);
+```
+
+`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.

contrib/temporal-gcp-cloud-run-worker-id/build.gradle

@@ -0,0 +1,12 @@
+description = '''Temporal Java SDK Google Cloud Run Worker Identity Support Module'''
+
+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}"
+}

contrib/temporal-gcp-cloud-run-worker-id/src/main/java/io/temporal/gcp/cloudrun/workerid/GoogleCloudRunMetadata.java

@@ -0,0 +1,234 @@
+package io.temporal.gcp.cloudrun.workerid;
+
+import io.temporal.common.Experimental;
+import io.temporal.common.WorkerDeploymentVersion;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.URI;
+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
+ * WorkerDeploymentVersion} from it.
+ *
+ * <p>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 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
+ * #workerDeploymentVersion() worker deployment version} yourself.
+ *
+ * <p>The deployment name and revision are resolved from environment variables Cloud Run injects
+ * into every instance. Cloud Run <b>worker pools</b> set {@code CLOUD_RUN_WORKER_POOL} and {@code
+ * CLOUD_RUN_REVISION}; Cloud Run <b>services</b> 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.
+ *
+ * <p><b>Experimental:</b> 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.
+   *
+   * <p>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) {
+    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<String, String> getenv) {
+    Objects.requireNonNull(metadataUrl, "metadataUrl");
+    Objects.requireNonNull(timeout, "timeout");
+    Objects.requireNonNull(getenv, "getenv");
+
+    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 {
+      connection = (HttpURLConnection) URI.create(metadataUrl).toURL().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.
+   *
+   * <p>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.
+   *
+   * <p>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);
+  }
+
+  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();
+  }
+}