diff --git a/cloud-run-worker-id/Dockerfile b/cloud-run-worker-id/Dockerfile new file mode 100644 index 00000000..e18c6c1e --- /dev/null +++ b/cloud-run-worker-id/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, 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 :cloud-run-worker-id:installDist + +FROM eclipse-temurin:17-jre-jammy + +RUN useradd --create-home --uid 10001 temporal +WORKDIR /app +COPY --from=build --chown=temporal:temporal \ + /workspace/cloud-run-worker-id/build/install/cloud-run-worker-id/ /app/ + +USER 10001 +ENTRYPOINT ["/app/bin/cloud-run-worker-id"] diff --git a/cloud-run-worker-id/README.md b/cloud-run-worker-id/README.md new file mode 100644 index 00000000..c7360b97 --- /dev/null +++ b/cloud-run-worker-id/README.md @@ -0,0 +1,163 @@ +# Cloud Run Worker (Worker Identity + Deployment Versioning) + +This sample runs a continuously polling Temporal Java Worker in a Google Cloud Run +**worker pool**. It uses the `temporal-gcp-cloud-run` helper to derive the Worker's Temporal +identity and its Worker Deployment Version from Cloud Run instance metadata, so every Cloud Run +revision registers as a distinct, `PINNED` Worker Deployment Version. It registers a small greeting +Workflow and Activity and runs until Cloud Run stops the instance. + +Cloud Run runs a long-lived container rather than a per-request handler, so there is no function to +wrap: the Worker fetches the metadata once at startup and applies it to the client and worker option +builders. + +> 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`, 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` 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 create the Worker Deployment Version and 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. + +## Layout + +- `src/main/java/io/temporal/samples/cloudrunworkerid/CloudRunWorker.java` fetches the Cloud Run + metadata, applies the derived identity and deployment version, and runs a long-lived Worker with a + bounded shutdown on `SIGTERM`. +- `GreetingWorkflow` / `GreetingWorkflowImpl` and `GreetingActivities` / `GreetingActivitiesImpl` are + the sample Workflow and Activity. The Workflow method is annotated + `@WorkflowVersioningBehavior(PINNED)` to match the Worker's PINNED default. +- `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: + +- **deployment 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`). + +The Worker then applies the metadata: + +- `applyTo(WorkflowClientOptions.Builder)` sets the Worker identity to `@` + (falling back to `@` and then ``). +- `applyTo(WorkerOptions.Builder)` enables Worker Deployment Versioning with the deployment name as + the deployment, the revision as the build id, and a `PINNED` default versioning behavior, so + in-flight Workflows stay on the revision that started 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 :cloud-run-worker-id:test +``` + +Build the runnable application (from a local SDK checkout, per the note above): + +```bash +./gradlew -PtemporalSdkPath=/path/to/sdk-java :cloud-run-worker-id: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` 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 deployment creates a new revision, which the Worker reports as a new build id under +the same deployment name. + +## Create the Worker Deployment Version and start a Workflow + +After the Worker is polling, register and route the Worker Deployment Version, then start the sample +Workflow. Use the deployment name (the worker pool name, `cloud-run-worker-id`) and the build id +(the Cloud Run revision) that the Worker logs at startup: + +```bash +temporal worker deployment set-current-version \ + --deployment-name cloud-run-worker-id \ + --build-id \ + --yes + +temporal workflow start \ + --task-queue cloud-run-worker-id \ + --type GreetingWorkflow \ + --workflow-id cloud-run-greeting \ + --input '"Cloud Run"' +``` + +## 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 + +Reset routing before deleting the Worker Deployment Version, then delete the worker pool: + +```bash +temporal worker deployment set-current-version \ + --deployment-name cloud-run-worker-id \ + --unversioned \ + --allow-no-pollers \ + --yes + +gcloud run worker-pools delete cloud-run-worker-id --region "$REGION" +``` diff --git a/cloud-run-worker-id/build.gradle b/cloud-run-worker-id/build.gradle new file mode 100644 index 00000000..3ba56045 --- /dev/null +++ b/cloud-run-worker-id/build.gradle @@ -0,0 +1,21 @@ +apply plugin: 'application' + +dependencies { + implementation "io.temporal:temporal-sdk:$javaSDKVersion" + implementation "io.temporal:temporal-gcp-cloud-run:$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.cloudrunworkerid.CloudRunWorker' +} diff --git a/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/CloudRunWorker.java b/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/CloudRunWorker.java new file mode 100644 index 00000000..0a6d1da9 --- /dev/null +++ b/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/CloudRunWorker.java @@ -0,0 +1,92 @@ +package io.temporal.samples.cloudrunworkerid; + +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; +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. + 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 + // deployment such as Temporal Cloud. + WorkflowServiceStubs service = + WorkflowServiceStubs.newServiceStubs( + WorkflowServiceStubsOptions.newBuilder().setTarget(address).build()); + + // applyTo(WorkflowClientOptions.Builder) sets the derived worker identity on the client. + WorkflowClient client = + WorkflowClient.newInstance( + service, + metadata.applyTo(WorkflowClientOptions.newBuilder().setNamespace(namespace)).build()); + + WorkerFactory factory = WorkerFactory.newInstance(client); + + // applyTo(WorkerOptions.Builder) enables Worker Deployment Versioning with the Cloud Run name + // as the deployment name, the revision as the build id, and a PINNED default versioning + // behavior. + WorkerOptions workerOptions = metadata.applyTo(WorkerOptions.newBuilder()).build(); + Worker worker = factory.newWorker(taskQueue, workerOptions); + 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={}, deployment={}, buildId={}, taskQueue={})", + metadata.workerIdentity(), + metadata.getName(), + metadata.getRevision(), + 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/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/GreetingActivities.java b/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/GreetingActivities.java new file mode 100644 index 00000000..86e9b9b1 --- /dev/null +++ b/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/GreetingActivities.java @@ -0,0 +1,12 @@ +package io.temporal.samples.cloudrunworkerid; + +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/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/GreetingActivitiesImpl.java b/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/GreetingActivitiesImpl.java new file mode 100644 index 00000000..f2f7ddfd --- /dev/null +++ b/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/GreetingActivitiesImpl.java @@ -0,0 +1,16 @@ +package io.temporal.samples.cloudrunworkerid; + +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/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/GreetingWorkflow.java b/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/GreetingWorkflow.java new file mode 100644 index 00000000..1e68f3fa --- /dev/null +++ b/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/GreetingWorkflow.java @@ -0,0 +1,12 @@ +package io.temporal.samples.cloudrunworkerid; + +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/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/GreetingWorkflowImpl.java b/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/GreetingWorkflowImpl.java new file mode 100644 index 00000000..6ccad02c --- /dev/null +++ b/cloud-run-worker-id/src/main/java/io/temporal/samples/cloudrunworkerid/GreetingWorkflowImpl.java @@ -0,0 +1,28 @@ +package io.temporal.samples.cloudrunworkerid; + +import io.temporal.activity.ActivityOptions; +import io.temporal.common.VersioningBehavior; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowVersioningBehavior; +import java.time.Duration; + +/** + * Greeting workflow implementation. + * + *

The method is annotated {@link VersioningBehavior#PINNED}, matching the PINNED default that + * {@link io.temporal.gcp.cloudrun.GoogleCloudRunMetadata} applies to the worker, so executions stay + * on the Cloud Run revision that started them. + */ +public final class GreetingWorkflowImpl implements GreetingWorkflow { + + private final GreetingActivities activities = + Workflow.newActivityStub( + GreetingActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()); + + @Override + @WorkflowVersioningBehavior(VersioningBehavior.PINNED) + public String getGreeting(String name) { + return activities.composeGreeting(name); + } +} diff --git a/cloud-run-worker-id/src/main/resources/logback.xml b/cloud-run-worker-id/src/main/resources/logback.xml new file mode 100644 index 00000000..28eb2cba --- /dev/null +++ b/cloud-run-worker-id/src/main/resources/logback.xml @@ -0,0 +1,14 @@ + + + + %d{HH:mm:ss.SSS} %-5level [%thread] %logger{36} - %msg%n + + + + + + + + + + diff --git a/cloud-run-worker-id/src/test/java/io/temporal/samples/cloudrunworkerid/GreetingWorkflowTest.java b/cloud-run-worker-id/src/test/java/io/temporal/samples/cloudrunworkerid/GreetingWorkflowTest.java new file mode 100644 index 00000000..cfc8e39d --- /dev/null +++ b/cloud-run-worker-id/src/test/java/io/temporal/samples/cloudrunworkerid/GreetingWorkflowTest.java @@ -0,0 +1,31 @@ +package io.temporal.samples.cloudrunworkerid; + +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..d502a239 100644 --- a/settings.gradle +++ b/settings.gradle @@ -9,3 +9,19 @@ include 'springboot' include 'springboot-basic' include 'lambda-worker:starter' include 'lambda-worker:worker' +include 'cloud-run-worker-id' + +// TEMPORARY (draft): the cloud-run-worker-id sample depends on io.temporal:temporal-gcp-cloud-run +// (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 cloud-run-worker-id module is affected. Remove this block and bump +// javaSDKVersion in build.gradle once temporal-gcp-cloud-run 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 +}