diff --git a/gcp/cloud-run/workerid/Dockerfile b/gcp/cloud-run/workerid/Dockerfile new file mode 100644 index 00000000..64b82d34 --- /dev/null +++ b/gcp/cloud-run/workerid/Dockerfile @@ -0,0 +1,22 @@ +FROM eclipse-temurin:17-jdk-jammy AS build + +WORKDIR /workspace +COPY . . + +# TEMPORARY (draft): this sample depends on io.temporal:temporal-gcp-cloud-run-worker-id, which is +# not yet released to Maven Central. Until it ships, the Gradle build resolves it from a local +# Temporal Java SDK checkout through a composite build (see README.md and settings.gradle). For an +# image build the local SDK checkout must be available in the build context (or the module published +# to Maven Local); once the module is released, bump javaSDKVersion in the samples root build.gradle +# and this builds unchanged from Maven Central. +RUN ./gradlew --no-daemon :gcp:cloud-run:workerid:installDist + +FROM eclipse-temurin:17-jre-jammy + +RUN useradd --create-home --uid 10001 temporal +WORKDIR /app +COPY --from=build --chown=temporal:temporal \ + /workspace/gcp/cloud-run/workerid/build/install/cloud-run-worker-id/ /app/ + +USER 10001 +ENTRYPOINT ["/app/bin/cloud-run-worker-id"] diff --git a/gcp/cloud-run/workerid/README.md b/gcp/cloud-run/workerid/README.md new file mode 100644 index 00000000..369227d5 --- /dev/null +++ b/gcp/cloud-run/workerid/README.md @@ -0,0 +1,150 @@ +# Temporal Cloud Run worker-identity worker + +This sample runs a continuously polling Temporal Java Worker in a Google Cloud Run **worker pool**. +It registers the `WorkerIdPlugin` from the `temporal-gcp-cloud-run-worker-id` module on the Temporal +client so the Worker's Temporal identity is derived from Cloud Run instance metadata as +`{instanceId}@{revision}`. It registers a small greeting Workflow and Activity and runs until Cloud +Run stops the instance. Identity only: the plugin sets the worker identity and nothing else. + +Cloud Run runs a long-lived container rather than a per-request handler, so there is no function to +wrap: registering the plugin on the client fetches the metadata once at startup and applies the +derived identity to the client and the Workers created from it. + +> Experimental: Google Cloud Run support is experimental and may change without notice. + +## Unreleased SDK dependency + +This sample depends on `io.temporal:temporal-gcp-cloud-run-worker-id`, which is **not yet released** +to Maven Central. Until it ships, the samples build wires the module from a local Temporal Java SDK +checkout through a Gradle composite build (`includeBuild`), configured in the samples root +`settings.gradle`. + +- It defaults to a sibling `../sdk-java-2` checkout on the `cloud-run-worker-id` branch. +- Override the location with `-PtemporalSdkPath=/path/to/sdk-java`. +- When that checkout is absent, the composite build is skipped and only this module is affected; the + other samples still build. + +Once `temporal-gcp-cloud-run-worker-id` is released, remove the composite-build block from +`settings.gradle` and bump `javaSDKVersion` in the samples root `build.gradle` to the released +version; the standard Maven Central build then works without the local checkout. This sample's pull +request stays a draft until then. + +## Prerequisites + +- Java 17+ +- The Temporal CLI (to start Workflows) +- The Google Cloud CLI (`gcloud`) with a project that has Cloud Run enabled +- A Temporal Service reachable from Cloud Run. A plaintext connection is used by default; configure + TLS or an API key in `CloudRunWorker.java` for a secured Service such as Temporal Cloud. + +## Files + +- `src/main/java/io/temporal/samples/gcp/cloudrun/workerid/CloudRunWorker.java` fetches the Cloud + Run metadata, registers `WorkerIdPlugin` on the client to apply the derived identity, and runs a + long-lived Worker with a bounded shutdown on `SIGTERM`. +- `GreetingWorkflow` / `GreetingWorkflowImpl` and `GreetingActivities` / `GreetingActivitiesImpl` are + the sample Workflow and Activity. +- `Dockerfile` packages the Gradle application as the Worker container. + +## How it works + +Cloud Run **worker pools** set `CLOUD_RUN_WORKER_POOL` and `CLOUD_RUN_REVISION` on every instance +(Cloud Run **services** set `K_SERVICE` and `K_REVISION`). `GoogleCloudRunMetadata.fetch()` resolves: + +- **name**: the first non-empty of `CLOUD_RUN_WORKER_POOL` then `K_SERVICE`. +- **revision**: the first non-empty of `CLOUD_RUN_REVISION` then `K_REVISION`. +- **instance id**: a single HTTP `GET` to the Cloud Run metadata server + (`http://metadata.google.internal/computeMetadata/v1/instance/id`, header `Metadata-Flavor: + Google`). + +`WorkerIdPlugin`, registered on the client with `WorkflowClientOptions.Builder.setPlugins(...)`, then +sets the Worker identity to `{instanceId}@{revision}` (falling back to `{instanceId}@{name}` and then +`{instanceId}`) unless an identity is already set. Workers created from the client inherit that +identity; the plugin sets nothing else on them. + +The Worker reads its connection settings from the environment: + +```bash +TEMPORAL_ADDRESS # host:port of the Temporal frontend (default 127.0.0.1:7233) +TEMPORAL_NAMESPACE # Temporal Namespace (default "default") +TEMPORAL_TASK_QUEUE # Task Queue to poll (default "cloud-run-worker-id") +``` + +`CLOUD_RUN_WORKER_POOL` and `CLOUD_RUN_REVISION` are injected by Cloud Run and do not need to be set +manually. + +## Build and test locally + +The unit test uses `TestWorkflowRule` and needs neither Cloud Run nor a running Temporal Service: + +```bash +./gradlew :gcp:cloud-run:workerid:test +``` + +Build the runnable application (from a local SDK checkout, per the note above): + +```bash +./gradlew -PtemporalSdkPath=/path/to/sdk-java :gcp:cloud-run:workerid:installDist +``` + +## Deploy to a Cloud Run worker pool + +Worker pools keep CPU allocated so the Temporal Worker can poll continuously; they are not +request-driven Cloud Run services. Set your connection values and deploy from the sample directory: + +```bash +export REGION=us-central1 +export TEMPORAL_ADDRESS=..tmprl.cloud:7233 +export TEMPORAL_NAMESPACE=. +export TEMPORAL_TASK_QUEUE=cloud-run-worker-id + +gcloud run worker-pools deploy cloud-run-worker-id \ + --source . \ + --region "$REGION" \ + --set-env-vars "TEMPORAL_ADDRESS=$TEMPORAL_ADDRESS,TEMPORAL_NAMESPACE=$TEMPORAL_NAMESPACE,TEMPORAL_TASK_QUEUE=$TEMPORAL_TASK_QUEUE" +``` + +`--source .` builds the container from the included `Dockerfile`. Because the image build resolves +the unreleased `temporal-gcp-cloud-run-worker-id` module, a remote source build succeeds only once +that module is released (or published to your Maven Local and made available to the build). Until +then, build the image locally against your SDK checkout and deploy it with `--image` instead: + +```bash +gcloud run worker-pools deploy cloud-run-worker-id \ + --image "$REGION-docker.pkg.dev/$PROJECT_ID//cloud-run-worker-id:latest" \ + --region "$REGION" \ + --set-env-vars "TEMPORAL_ADDRESS=$TEMPORAL_ADDRESS,TEMPORAL_NAMESPACE=$TEMPORAL_NAMESPACE,TEMPORAL_TASK_QUEUE=$TEMPORAL_TASK_QUEUE" +``` + +Each Cloud Run revision starts a fresh instance whose Worker reports a distinct identity, which the +Worker logs at startup. + +## Start a Workflow + +After the Worker is polling, start the sample Workflow on the same Task Queue: + +```bash +temporal workflow start \ + --task-queue cloud-run-worker-id \ + --type GreetingWorkflow \ + --workflow-id cloud-run-greeting \ + --input '"Cloud Run"' +``` + +The Worker's identity appears on its Task Queue pollers (for example in `temporal task-queue +describe`) and on the events it records. + +## Shutdown + +Cloud Run sends `SIGTERM` and allows a short grace period before `SIGKILL`. The shutdown hook stops +polling, waits up to six seconds for in-flight tasks to drain, escalates to a forced shutdown if +needed, and then closes the service connection. Long-running Activities should still heartbeat and +handle cancellation so they can stop within the platform's shutdown window. + +## Clean up + +Delete the worker pool when you are done: + +```bash +gcloud run worker-pools delete cloud-run-worker-id --region "$REGION" +``` diff --git a/gcp/cloud-run/workerid/build.gradle b/gcp/cloud-run/workerid/build.gradle new file mode 100644 index 00000000..8e6eb290 --- /dev/null +++ b/gcp/cloud-run/workerid/build.gradle @@ -0,0 +1,24 @@ +apply plugin: 'application' + +dependencies { + implementation "io.temporal:temporal-sdk:$javaSDKVersion" + implementation "io.temporal:temporal-gcp-cloud-run-worker-id:$javaSDKVersion" + runtimeOnly group: 'ch.qos.logback', name: 'logback-classic', version: '1.5.6' + + testImplementation "io.temporal:temporal-testing:$javaSDKVersion" + testImplementation "junit:junit:4.13.2" + testImplementation(platform("org.junit:junit-bom:5.10.3")) + testRuntimeOnly "org.junit.vintage:junit-vintage-engine" + + dependencies { + errorproneJavac('com.google.errorprone:javac:9+181-r4173-1') + errorprone('com.google.errorprone:error_prone_core:2.28.0') + } +} + +application { + mainClass = 'io.temporal.samples.gcp.cloudrun.workerid.CloudRunWorker' + // Keep a stable launcher/installDist name independent of the nested Gradle + // project name (:gcp:cloud-run:workerid), which the Dockerfile relies on. + applicationName = 'cloud-run-worker-id' +} diff --git a/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/CloudRunWorker.java b/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/CloudRunWorker.java new file mode 100644 index 00000000..30383b5a --- /dev/null +++ b/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/CloudRunWorker.java @@ -0,0 +1,93 @@ +package io.temporal.samples.gcp.cloudrun.workerid; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.gcp.cloudrun.workerid.GoogleCloudRunMetadata; +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; +import java.util.concurrent.TimeUnit; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** A continuously polling Temporal Worker for a Google Cloud Run worker pool. */ +public final class CloudRunWorker { + private static final Logger logger = LoggerFactory.getLogger(CloudRunWorker.class); + + static final String ADDRESS_ENV = "TEMPORAL_ADDRESS"; + static final String NAMESPACE_ENV = "TEMPORAL_NAMESPACE"; + static final String TASK_QUEUE_ENV = "TEMPORAL_TASK_QUEUE"; + + static final String DEFAULT_ADDRESS = "127.0.0.1:7233"; + static final String DEFAULT_NAMESPACE = "default"; + static final String DEFAULT_TASK_QUEUE = "cloud-run-worker-id"; + + private CloudRunWorker() {} + + public static void main(String[] args) { + // Read Cloud Run instance metadata once during startup. This performs a single HTTP request to + // the Cloud Run metadata server and throws IllegalStateException when it is unreachable, which + // usually means the process is not running on Google Cloud Run. + // @@@SNIPSTART java-cloud-run-worker-id + GoogleCloudRunMetadata metadata = GoogleCloudRunMetadata.fetch(); + + String address = envOrDefault(ADDRESS_ENV, DEFAULT_ADDRESS); + String namespace = envOrDefault(NAMESPACE_ENV, DEFAULT_NAMESPACE); + String taskQueue = envOrDefault(TASK_QUEUE_ENV, DEFAULT_TASK_QUEUE); + + // Plaintext connection to the Temporal Service. Configure TLS or an API key here for a secured + // Service such as Temporal Cloud. + WorkflowServiceStubs service = + WorkflowServiceStubs.newServiceStubs( + WorkflowServiceStubsOptions.newBuilder().setTarget(address).build()); + + // Register WorkerIdPlugin on the client. It sets the derived worker identity + // ({instanceId}@{revision}) on the client, and workers created from the client inherit it. + // Passing the already-fetched metadata avoids a second call to the Cloud Run metadata server. + WorkflowClient client = + WorkflowClient.newInstance( + service, + WorkflowClientOptions.newBuilder() + .setNamespace(namespace) + .setPlugins(new WorkerIdPlugin(metadata)) + .build()); + // @@@SNIPEND + + WorkerFactory factory = WorkerFactory.newInstance(client); + + Worker worker = factory.newWorker(taskQueue); + worker.registerWorkflowImplementationTypes(GreetingWorkflowImpl.class); + worker.registerActivitiesImplementations(new GreetingActivitiesImpl()); + + Runtime.getRuntime() + .addShutdownHook(new Thread(() -> shutdown(factory, service), "temporal-worker-shutdown")); + + factory.start(); + logger.info( + "Temporal worker started (identity={}, taskQueue={})", + metadata.workerIdentity(), + taskQueue); + + // Cloud Run worker pools are continuous workloads, so keep the process alive until SIGTERM. + factory.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS); + } + + private static void shutdown(WorkerFactory factory, WorkflowServiceStubs service) { + // Cloud Run sends SIGTERM and allows a short grace period before SIGKILL. Stop polling, drain + // in-flight tasks, then close the service connection. + factory.shutdown(); + factory.awaitTermination(6, TimeUnit.SECONDS); + if (!factory.isTerminated()) { + factory.shutdownNow(); + factory.awaitTermination(1, TimeUnit.SECONDS); + } + service.shutdown(); + } + + private static String envOrDefault(String name, String defaultValue) { + String value = System.getenv(name); + return value == null || value.trim().isEmpty() ? defaultValue : value; + } +} diff --git a/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingActivities.java b/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingActivities.java new file mode 100644 index 00000000..f04446b5 --- /dev/null +++ b/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingActivities.java @@ -0,0 +1,12 @@ +package io.temporal.samples.gcp.cloudrun.workerid; + +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; + +/** Activity interface used by {@link GreetingWorkflow}. */ +@ActivityInterface +public interface GreetingActivities { + + @ActivityMethod + String composeGreeting(String name); +} diff --git a/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingActivitiesImpl.java b/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingActivitiesImpl.java new file mode 100644 index 00000000..ae83f37d --- /dev/null +++ b/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingActivitiesImpl.java @@ -0,0 +1,16 @@ +package io.temporal.samples.gcp.cloudrun.workerid; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Activity implementation that returns a simple greeting. */ +public final class GreetingActivitiesImpl implements GreetingActivities { + + private static final Logger logger = LoggerFactory.getLogger(GreetingActivitiesImpl.class); + + @Override + public String composeGreeting(String name) { + logger.info("Composing greeting for {}", name); + return "Hello, " + name + "!"; + } +} diff --git a/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingWorkflow.java b/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingWorkflow.java new file mode 100644 index 00000000..4fc2a7de --- /dev/null +++ b/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingWorkflow.java @@ -0,0 +1,12 @@ +package io.temporal.samples.gcp.cloudrun.workerid; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +/** A small greeting workflow run by the Cloud Run worker. */ +@WorkflowInterface +public interface GreetingWorkflow { + + @WorkflowMethod + String getGreeting(String name); +} diff --git a/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingWorkflowImpl.java b/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingWorkflowImpl.java new file mode 100644 index 00000000..f63908c1 --- /dev/null +++ b/gcp/cloud-run/workerid/src/main/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingWorkflowImpl.java @@ -0,0 +1,19 @@ +package io.temporal.samples.gcp.cloudrun.workerid; + +import io.temporal.activity.ActivityOptions; +import io.temporal.workflow.Workflow; +import java.time.Duration; + +/** Greeting workflow implementation. */ +public final class GreetingWorkflowImpl implements GreetingWorkflow { + + private final GreetingActivities activities = + Workflow.newActivityStub( + GreetingActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()); + + @Override + public String getGreeting(String name) { + return activities.composeGreeting(name); + } +} diff --git a/gcp/cloud-run/workerid/src/main/resources/logback.xml b/gcp/cloud-run/workerid/src/main/resources/logback.xml new file mode 100644 index 00000000..28eb2cba --- /dev/null +++ b/gcp/cloud-run/workerid/src/main/resources/logback.xml @@ -0,0 +1,14 @@ + + + + %d{HH:mm:ss.SSS} %-5level [%thread] %logger{36} - %msg%n + + + + + + + + + + diff --git a/gcp/cloud-run/workerid/src/test/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingWorkflowTest.java b/gcp/cloud-run/workerid/src/test/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingWorkflowTest.java new file mode 100644 index 00000000..a4f6d457 --- /dev/null +++ b/gcp/cloud-run/workerid/src/test/java/io/temporal/samples/gcp/cloudrun/workerid/GreetingWorkflowTest.java @@ -0,0 +1,31 @@ +package io.temporal.samples.gcp.cloudrun.workerid; + +import static org.junit.Assert.assertEquals; + +import io.temporal.client.WorkflowOptions; +import io.temporal.testing.TestWorkflowRule; +import org.junit.Rule; +import org.junit.Test; + +/** Unit test for the sample Workflow and Activity. */ +public class GreetingWorkflowTest { + + @Rule + public TestWorkflowRule testWorkflowRule = + TestWorkflowRule.newBuilder() + .setWorkflowTypes(GreetingWorkflowImpl.class) + .setActivityImplementations(new GreetingActivitiesImpl()) + .build(); + + @Test + public void returnsGreeting() { + GreetingWorkflow workflow = + testWorkflowRule + .getWorkflowClient() + .newWorkflowStub( + GreetingWorkflow.class, + WorkflowOptions.newBuilder().setTaskQueue(testWorkflowRule.getTaskQueue()).build()); + + assertEquals("Hello, Cloud Run!", workflow.getGreeting("Cloud Run")); + } +} diff --git a/settings.gradle b/settings.gradle index b19f2fd7..1a9e9b2e 100644 --- a/settings.gradle +++ b/settings.gradle @@ -9,3 +9,20 @@ include 'springboot' include 'springboot-basic' include 'lambda-worker:starter' include 'lambda-worker:worker' +include 'gcp:cloud-run:workerid' + +// TEMPORARY (draft): the gcp:cloud-run:workerid sample depends on +// io.temporal:temporal-gcp-cloud-run-worker-id (and the worker-identity APIs it builds on), which +// are not yet released to Maven Central. Until they ship, wire the sample against a local Temporal +// Java SDK checkout with a Gradle composite build so it can compile and run. Defaults to a sibling +// ../sdk-java-2 checkout on the cloud-run-worker-id branch; override the location with +// -PtemporalSdkPath=/path/to/sdk-java. When the checkout is absent (for example on CI building the +// other samples) the composite build is skipped and only the gcp:cloud-run:workerid module is +// affected. Remove this block and bump javaSDKVersion in build.gradle once +// temporal-gcp-cloud-run-worker-id is released. +def temporalSdkPath = gradle.startParameter.projectProperties['temporalSdkPath'] ?: '../sdk-java-2' +def temporalSdkFile = new File(temporalSdkPath) +def temporalSdkDir = temporalSdkFile.isAbsolute() ? temporalSdkFile : new File(settingsDir, temporalSdkPath) +if (temporalSdkDir.isDirectory()) { + includeBuild temporalSdkPath +}