Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 86 additions & 0 deletions contrib/temporal-gcp-cloud-run-worker-id/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,86 @@
# Temporal Google Cloud Run worker identity support

This module derives a Temporal worker **identity** for Google Cloud Run from instance metadata, for both Cloud Run **worker pools** and Cloud Run **services**, so each Cloud Run instance reports a stable, recognizable identity to the Temporal service.

The primary API is `WorkerIdPlugin`. Register it once on your workflow client and it sets the client identity automatically; every worker created from that client inherits it. 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);

// Workers created from this client inherit the identity the plugin set on the client. 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 Cloud Run worker pool 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 client plugin hook:

- **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. Workers created from the client inherit this identity; the plugin sets nothing else on them.

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). 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();

// 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.
12 changes: 12 additions & 0 deletions contrib/temporal-gcp-cloud-run-worker-id/build.gradle
Original file line number Diff line number Diff line change
@@ -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}"
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,212 @@
package io.temporal.gcp.cloudrun.workerid;

import io.temporal.common.Experimental;
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 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 to the client. Use this class directly to read the {@linkplain
* #workerIdentity() worker identity} yourself.
*
* <p>The 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 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 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 worker pool or service 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;
}

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();
}
}
Loading
Loading