diff --git a/README.md b/README.md index 284e4c32..cd0a6d19 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,7 @@ It contains the following modules: * [SpringBoot Basic](/springboot-basic): Minimal sample showing SpringBoot autoconfig integration without any extra external dependencies. * [Spring AI](/springai): demonstrates the Temporal Spring AI integration — durable AI agents with chat models, tools, MCP servers, vector stores, and embeddings. * [Lambda Worker](/lambda-worker): demonstrates running a Temporal Java Worker inside AWS Lambda. +* [Cloud Run Worker (OpenTelemetry)](/gcp/cloud-run/opentelemetry): demonstrates running a Temporal Java Worker in a Google Cloud Run worker pool, exporting SDK metrics and traces through an OpenTelemetry collector sidecar. ## Learn more about Temporal and Java SDK diff --git a/gcp/cloud-run/opentelemetry/Dockerfile b/gcp/cloud-run/opentelemetry/Dockerfile new file mode 100644 index 00000000..29e6eec2 --- /dev/null +++ b/gcp/cloud-run/opentelemetry/Dockerfile @@ -0,0 +1,17 @@ +FROM eclipse-temurin:17-jdk-jammy AS build + +WORKDIR /workspace +COPY . . + +# temporal-gcp-cloud-run-opentelemetry is unreleased; the composite build resolves it from a local SDK checkout (see README). +RUN ./gradlew --no-daemon :gcp:cloud-run:opentelemetry: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/opentelemetry/build/install/cloud-run-worker/ /app/ + +USER 10001 +ENTRYPOINT ["/app/bin/cloud-run-worker"] diff --git a/gcp/cloud-run/opentelemetry/README.md b/gcp/cloud-run/opentelemetry/README.md new file mode 100644 index 00000000..384cc120 --- /dev/null +++ b/gcp/cloud-run/opentelemetry/README.md @@ -0,0 +1,64 @@ +# Temporal Cloud Run OpenTelemetry worker + +A Temporal Worker running in a Google Cloud Run **worker pool** that exports +Temporal SDK metrics and traces through a +[Google-Built OpenTelemetry Collector](https://cloud.google.com/stackdriver/docs/instrumentation/opentelemetry-collector-cloud-run) +sidecar. Worker pools keep CPU allocated, unlike request-driven Cloud Run services. + +`CloudRunOpenTelemetryPlugin` from `io.temporal:temporal-gcp-cloud-run-opentelemetry` configures the +SDK metrics scope, tracing interceptors, OTLP exporters (default `http://localhost:4317`), and +shutdown flushing. It derives `service.name` from `CLOUD_RUN_WORKER_POOL` and reports metrics every +60 seconds, matching the OpenTelemetry SDK default across the Temporal SDKs. + +## Unreleased SDK dependency + +`temporal-gcp-cloud-run-opentelemetry` is not yet released. `settings.gradle` resolves it (and the +other `io.temporal:*` modules) from a local SDK checkout via a Gradle composite build, defaulting to +`../sdk-java` and overridable with `-PtemporalSdkPath`. CI has no checkout, so its build stays red +until the module ships; then drop the composite block and bump `javaSDKVersion`. + +```bash +./gradlew -PtemporalSdkPath=/path/to/sdk-java :gcp:cloud-run:opentelemetry:build +``` + +## Files + +- `.../cloudrun/opentelemetry/CloudRunWorker.java` — plugin, client, Temporal Worker, bounded `SIGTERM` shutdown. +- `collector-config.yaml` — collector for cumulative Prometheus metrics and batched traces. +- `worker-pool.yaml` — worker and collector containers sharing localhost, config from Secret Manager. +- `Dockerfile` — packages the Gradle application as the worker container. + +## Deploy + +Run from the repository root, with a Temporal Cloud namespace and API key. + +1. Store the API key and collector config in Secret Manager: + + ```bash + printf '%s' "$TEMPORAL_API_KEY" | \ + gcloud secrets create temporal-api-key --data-file=- --project="$PROJECT_ID" + gcloud secrets create temporal-collector-config \ + --data-file=gcp/cloud-run/opentelemetry/collector-config.yaml --project="$PROJECT_ID" + ``` + +2. Build and push the image (tag `REGION-docker.pkg.dev/PROJECT_ID/temporal-samples/cloud-run-worker:latest`) + with `docker build -f gcp/cloud-run/opentelemetry/Dockerfile .`. + +3. Replace the placeholders in `worker-pool.yaml` (project, region, namespace, address, secret + versions, image), then deploy: + + ```bash + gcloud run worker-pools replace gcp/cloud-run/opentelemetry/worker-pool.yaml --project="$PROJECT_ID" + ``` + +The worker-pool service account needs `roles/monitoring.metricWriter`, `roles/telemetry.tracesWriter`, +and `roles/secretmanager.secretAccessor`. Start a workflow on task queue `cloud-run-worker` to +generate telemetry: + +```bash +temporal workflow start --type GreetingWorkflow --task-queue cloud-run-worker \ + --workflow-id cloud-run-greeting --input '"Google Cloud"' +``` + +The collector does **not** batch cumulative metrics: a shutdown flush batched with a recent periodic +export would collide on the same Prometheus series and be rejected as `Duplicate TimeSeries`. diff --git a/gcp/cloud-run/opentelemetry/build.gradle b/gcp/cloud-run/opentelemetry/build.gradle new file mode 100644 index 00000000..b93fb457 --- /dev/null +++ b/gcp/cloud-run/opentelemetry/build.gradle @@ -0,0 +1,24 @@ +apply plugin: 'application' + +dependencies { + implementation "io.temporal:temporal-sdk:$javaSDKVersion" + implementation "io.temporal:temporal-envconfig:$javaSDKVersion" + implementation "io.temporal:temporal-gcp-cloud-run-opentelemetry:$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.opentelemetry.CloudRunWorker' + // Stable launcher name the Dockerfile relies on, independent of the Gradle project path. + applicationName = 'cloud-run-worker' +} diff --git a/gcp/cloud-run/opentelemetry/collector-config.yaml b/gcp/cloud-run/opentelemetry/collector-config.yaml new file mode 100644 index 00000000..36504f85 --- /dev/null +++ b/gcp/cloud-run/opentelemetry/collector-config.yaml @@ -0,0 +1,87 @@ +# @@@SNIPSTART java-cloud-run-otel-collector-config +receivers: + otlp: + protocols: + grpc: + endpoint: localhost:4317 + +processors: + # Batch traces for throughput. Do not add this processor to the cumulative metrics pipeline: + # a shutdown flush can otherwise be batched with a recent periodic export of the same series. + batch/traces: + send_batch_max_size: 200 + send_batch_size: 200 + timeout: 5s + memory_limiter: + # This is the collector's memory polling cadence, not the SDK metric export interval. + check_interval: 1s + limit_percentage: 65 + spike_limit_percentage: 20 + resourcedetection: + detectors: [gcp] + timeout: 10s + # Avoid collisions with labels that Google Managed Service for Prometheus adds. + transform/collision: + metric_statements: + - context: datapoint + statements: + - set(attributes["exported_location"], attributes["location"]) + - delete_key(attributes, "location") + - set(attributes["exported_cluster"], attributes["cluster"]) + - delete_key(attributes, "cluster") + - set(attributes["exported_namespace"], attributes["namespace"]) + - delete_key(attributes, "namespace") + - set(attributes["exported_job"], attributes["job"]) + - delete_key(attributes, "job") + - set(attributes["exported_instance"], attributes["instance"]) + - delete_key(attributes, "instance") + - set(attributes["exported_project_id"], attributes["project_id"]) + - delete_key(attributes, "project_id") + # The Telemetry API expects the Google Cloud project in gcp.project_id. + transform/set_project_id: + error_mode: ignore + trace_statements: + - set(resource.attributes["gcp.project_id"], resource.attributes["gcp.project.id"]) where resource.attributes["gcp.project.id"] != nil + - set(resource.attributes["gcp.project_id"], resource.attributes["cloud.account.id"]) where resource.attributes["gcp.project_id"] == nil and resource.attributes["cloud.account.id"] != nil + +exporters: + googlemanagedprometheus: + # Google Cloud's supported OTLP path for traces is the Telemetry API. + otlp: + endpoint: telemetry.googleapis.com:443 + compression: none + balancer_name: pick_first + auth: + authenticator: googleclientauth + +extensions: + # Cloud Run container dependencies require a startup probe. This endpoint is also used for the + # collector liveness probe in worker-pool.yaml. + health_check: + endpoint: 0.0.0.0:13133 + googleclientauth: + +service: + extensions: + - health_check + - googleclientauth + pipelines: + metrics/otlp: + receivers: [otlp] + processors: [memory_limiter, resourcedetection, transform/collision] + exporters: [googlemanagedprometheus] + traces: + receivers: [otlp] + processors: [memory_limiter, resourcedetection, transform/set_project_id, batch/traces] + exporters: [otlp] + # Feed collector self-metrics back through the metrics pipeline. + telemetry: + metrics: + readers: + - periodic: + exporter: + otlp: + protocol: grpc + endpoint: http://localhost:4317 + insecure: true +# @@@SNIPEND diff --git a/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/CloudRunWorker.java b/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/CloudRunWorker.java new file mode 100644 index 00000000..40c43833 --- /dev/null +++ b/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/CloudRunWorker.java @@ -0,0 +1,72 @@ +package io.temporal.samples.gcp.cloudrun.opentelemetry; + +import io.temporal.client.WorkflowClient; +import io.temporal.envconfig.ClientConfigProfile; +import io.temporal.gcp.cloudrun.opentelemetry.CloudRunOpenTelemetryPlugin; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; +import java.io.IOException; +import java.time.Duration; +import java.util.concurrent.TimeUnit; + +/** A Temporal Worker for a Cloud Run worker pool. */ +public final class CloudRunWorker { + public static final String DEFAULT_TASK_QUEUE = "cloud-run-worker"; + + private CloudRunWorker() {} + + public static void main(String[] args) throws IOException { + // @@@SNIPSTART java-cloud-run-otel-worker + ClientConfigProfile profile = ClientConfigProfile.load(); + CloudRunOpenTelemetryPlugin telemetryPlugin = CloudRunOpenTelemetryPlugin.newBuilder().build(); + + WorkflowServiceStubsOptions serviceOptions = + WorkflowServiceStubsOptions.newBuilder(profile.toWorkflowServiceStubsOptions()) + .setPlugins(telemetryPlugin) + .build(); + WorkflowServiceStubs service = WorkflowServiceStubs.newServiceStubs(serviceOptions); + // @@@SNIPEND + WorkflowClient client = WorkflowClient.newInstance(service, profile.toWorkflowClientOptions()); + WorkerFactory factory = WorkerFactory.newInstance(client); + + String taskQueue = taskQueue(); + Worker worker = factory.newWorker(taskQueue); + worker.registerWorkflowImplementationTypes(GreetingWorkflowImpl.class); + worker.registerActivitiesImplementations(new GreetingActivitiesImpl()); + + Runtime.getRuntime() + .addShutdownHook( + new Thread( + () -> shutdown(factory, service, telemetryPlugin), "temporal-worker-shutdown")); + + factory.start(); + System.out.printf( + "Temporal worker started: taskQueue=%s, otelEndpoint=%s, serviceName=%s%n", + taskQueue, telemetryPlugin.getEndpoint(), telemetryPlugin.getServiceName()); + + // Keep the process alive until Cloud Run sends SIGTERM. + factory.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS); + } + + private static String taskQueue() { + String configured = System.getenv("TEMPORAL_TASK_QUEUE"); + return configured == null || configured.trim().isEmpty() ? DEFAULT_TASK_QUEUE : configured; + } + + private static void shutdown( + WorkerFactory factory, + WorkflowServiceStubs service, + CloudRunOpenTelemetryPlugin telemetryPlugin) { + // Flush after worker shutdown to capture finishing tasks; Cloud Run allows 10s. + factory.shutdown(); + factory.awaitTermination(6, TimeUnit.SECONDS); + if (!factory.isTerminated()) { + factory.shutdownNow(); + factory.awaitTermination(1, TimeUnit.SECONDS); + } + telemetryPlugin.newFlushHook().run(Duration.ofSeconds(2)); + service.shutdown(); + } +} diff --git a/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingActivities.java b/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingActivities.java new file mode 100644 index 00000000..7d567f5a --- /dev/null +++ b/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingActivities.java @@ -0,0 +1,8 @@ +package io.temporal.samples.gcp.cloudrun.opentelemetry; + +import io.temporal.activity.ActivityInterface; + +@ActivityInterface +public interface GreetingActivities { + String composeGreeting(String name); +} diff --git a/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingActivitiesImpl.java b/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingActivitiesImpl.java new file mode 100644 index 00000000..0e89a12c --- /dev/null +++ b/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingActivitiesImpl.java @@ -0,0 +1,8 @@ +package io.temporal.samples.gcp.cloudrun.opentelemetry; + +public final class GreetingActivitiesImpl implements GreetingActivities { + @Override + public String composeGreeting(String name) { + return "Hello " + name + "!"; + } +} diff --git a/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingWorkflow.java b/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingWorkflow.java new file mode 100644 index 00000000..9b6a0c40 --- /dev/null +++ b/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingWorkflow.java @@ -0,0 +1,10 @@ +package io.temporal.samples.gcp.cloudrun.opentelemetry; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +@WorkflowInterface +public interface GreetingWorkflow { + @WorkflowMethod + String getGreeting(String name); +} diff --git a/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingWorkflowImpl.java b/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingWorkflowImpl.java new file mode 100644 index 00000000..f6a64995 --- /dev/null +++ b/gcp/cloud-run/opentelemetry/src/main/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingWorkflowImpl.java @@ -0,0 +1,17 @@ +package io.temporal.samples.gcp.cloudrun.opentelemetry; + +import io.temporal.activity.ActivityOptions; +import io.temporal.workflow.Workflow; +import java.time.Duration; + +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/opentelemetry/src/test/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingWorkflowTest.java b/gcp/cloud-run/opentelemetry/src/test/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingWorkflowTest.java new file mode 100644 index 00000000..8fb5d84d --- /dev/null +++ b/gcp/cloud-run/opentelemetry/src/test/java/io/temporal/samples/gcp/cloudrun/opentelemetry/GreetingWorkflowTest.java @@ -0,0 +1,32 @@ +package io.temporal.samples.gcp.cloudrun.opentelemetry; + +import static org.junit.Assert.assertEquals; + +import io.temporal.client.WorkflowOptions; +import io.temporal.testing.TestWorkflowRule; +import org.junit.Rule; +import org.junit.Test; + +public class GreetingWorkflowTest { + @Rule + public TestWorkflowRule testWorkflowRule = + TestWorkflowRule.newBuilder() + .setWorkflowTypes(GreetingWorkflowImpl.class) + .setDoNotStart(true) + .build(); + + @Test + public void completesGreeting() { + testWorkflowRule.getWorker().registerActivitiesImplementations(new GreetingActivitiesImpl()); + testWorkflowRule.getTestEnvironment().start(); + + GreetingWorkflow workflow = + testWorkflowRule + .getWorkflowClient() + .newWorkflowStub( + GreetingWorkflow.class, + WorkflowOptions.newBuilder().setTaskQueue(testWorkflowRule.getTaskQueue()).build()); + + assertEquals("Hello Google Cloud!", workflow.getGreeting("Google Cloud")); + } +} diff --git a/gcp/cloud-run/opentelemetry/worker-pool.yaml b/gcp/cloud-run/opentelemetry/worker-pool.yaml new file mode 100644 index 00000000..5c5e0042 --- /dev/null +++ b/gcp/cloud-run/opentelemetry/worker-pool.yaml @@ -0,0 +1,68 @@ +apiVersion: run.googleapis.com/v1 +kind: WorkerPool +metadata: + name: temporal-cloud-run-worker + labels: + cloud.googleapis.com/location: REGION + annotations: + run.googleapis.com/scalingMode: manual + run.googleapis.com/manualInstanceCount: "1" +spec: + template: + metadata: + annotations: + # Cloud Run starts the worker only after the collector startup probe succeeds. + run.googleapis.com/container-dependencies: '{"worker":["collector"]}' + run.googleapis.com/execution-environment: gen2 + spec: + containerConcurrency: 0 + serviceAccountName: temporal-cloud-run-worker@PROJECT_ID.iam.gserviceaccount.com + containers: + - name: worker + image: REGION-docker.pkg.dev/PROJECT_ID/temporal-samples/cloud-run-worker:latest + env: + - name: TEMPORAL_ADDRESS + value: NAMESPACE_ID.tmprl.cloud:7233 + - name: TEMPORAL_NAMESPACE + value: NAMESPACE_ID.ACCOUNT_ID + - name: TEMPORAL_API_KEY + valueFrom: + secretKeyRef: + key: "1" + name: temporal-api-key + - name: TEMPORAL_TASK_QUEUE + value: cloud-run-worker + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: http://localhost:4317 + resources: + limits: + cpu: "1" + memory: 512Mi + - name: collector + image: us-docker.pkg.dev/cloud-ops-agents-artifacts/google-cloud-opentelemetry-collector/otelcol-google:0.156.0 + args: + - --config=env:OTELCOL_CONFIG + env: + - name: OTELCOL_CONFIG + valueFrom: + secretKeyRef: + key: "1" + name: temporal-collector-config + startupProbe: + httpGet: + path: / + port: 13133 + timeoutSeconds: 5 + periodSeconds: 10 + failureThreshold: 12 + livenessProbe: + httpGet: + path: / + port: 13133 + timeoutSeconds: 5 + periodSeconds: 30 + failureThreshold: 3 + resources: + limits: + cpu: "1" + memory: 512Mi diff --git a/settings.gradle b/settings.gradle index b19f2fd7..e4dfc9a1 100644 --- a/settings.gradle +++ b/settings.gradle @@ -9,3 +9,10 @@ include 'springboot' include 'springboot-basic' include 'lambda-worker:starter' include 'lambda-worker:worker' +include 'gcp:cloud-run:opentelemetry' + +// The gcp:cloud-run samples use unreleased temporal-gcp-cloud-run-* modules; resolve them from a local SDK checkout (default ../sdk-java, override with -PtemporalSdkPath) via composite build. +def temporalSdkPath = gradle.startParameter.projectProperties['temporalSdkPath'] ?: '../sdk-java' +if (file(temporalSdkPath).isDirectory()) { + includeBuild temporalSdkPath +}