From 91cdda5be60231465be5719fd4df627edf59a5a8 Mon Sep 17 00:00:00 2001 From: Edward Amsden Date: Thu, 16 Jul 2026 16:07:37 -0500 Subject: [PATCH 1/3] Add Google Cloud OpenTelemetry plugin --- contrib/temporal-gcp/README.md | 51 +++++ contrib/temporal-gcp/build.gradle | 17 ++ .../temporal/gcp/GcpOpenTelemetryPlugin.java | 180 ++++++++++++++++++ .../gcp/GcpOpenTelemetryPluginTest.java | 99 ++++++++++ settings.gradle | 2 + temporal-bom/build.gradle | 1 + 6 files changed, 350 insertions(+) create mode 100644 contrib/temporal-gcp/README.md create mode 100644 contrib/temporal-gcp/build.gradle create mode 100644 contrib/temporal-gcp/src/main/java/io/temporal/gcp/GcpOpenTelemetryPlugin.java create mode 100644 contrib/temporal-gcp/src/test/java/io/temporal/gcp/GcpOpenTelemetryPluginTest.java diff --git a/contrib/temporal-gcp/README.md b/contrib/temporal-gcp/README.md new file mode 100644 index 000000000..dbe6ececb --- /dev/null +++ b/contrib/temporal-gcp/README.md @@ -0,0 +1,51 @@ +# Temporal Google Cloud module + +This module provides an OpenTelemetry plugin with defaults for Temporal Java SDK workers running on Google Cloud Run services and worker pools. + +> **Collector required by default:** The plugin exports metrics and traces to an OTLP collector at `http://localhost:4317`. It does not export directly to Google Cloud. Deploy the Google-Built OpenTelemetry Collector as a sidecar, configure another collector endpoint, or provide an application-owned `OpenTelemetry` instance. Without a collector at the configured endpoint, telemetry is not delivered to Google Cloud. + +This integration is for container-based Cloud Run workloads. It does not implement a Cloud Run functions invocation lifecycle. + +## Usage + +Add `temporal-gcp` next to your Temporal SDK dependency, then install the plugin on service stubs options before creating clients and workers: + +```java +GcpOpenTelemetryPlugin plugin = GcpOpenTelemetryPlugin.newBuilder().build(); + +WorkflowServiceStubs service = + WorkflowServiceStubs.newServiceStubs( + WorkflowServiceStubsOptions.newBuilder() + .setPlugins(plugin) + .build()); +WorkflowClient client = WorkflowClient.newInstance(service); +WorkerFactory factory = WorkerFactory.newInstance(client); +``` + +The plugin configures the SDK metrics scope, tracing interceptors, OTLP metric and trace exporters, and worker-factory shutdown flushing through `temporal-opentelemetry`. Do not install both `GcpOpenTelemetryPlugin` and `OpenTelemetryPlugin` on the same service stubs. + +The OTLP endpoint is resolved in this order: + +1. `Builder.setEndpoint(...)`. +2. `OTEL_EXPORTER_OTLP_ENDPOINT`. +3. `http://localhost:4317`. + +The OpenTelemetry service name is resolved in this order: + +1. `Builder.setServiceName(...)`. +2. `OTEL_SERVICE_NAME`. +3. `CLOUD_RUN_WORKER_POOL` for a Cloud Run worker pool. +4. `K_SERVICE` for a Cloud Run service. +5. `temporal-worker`. + +The collector should use its GCP resource detector to add the Google Cloud project, location, revision, and monitored-resource attributes. This module does not call the Google Cloud metadata server and adds no Google Cloud client libraries or exporters to the worker process. + +## Collector sidecar + +Google publishes the Google-Built OpenTelemetry Collector as a container image. Configure it as a second Cloud Run container, listen for OTLP gRPC on `localhost:4317`, and use its GCP exporters for metrics and traces. For the image, recommended collector configuration, IAM roles, health check, and Secret Manager mount, see [Deploy Google-Built OpenTelemetry Collector on Cloud Run](https://cloud.google.com/stackdriver/docs/instrumentation/opentelemetry-collector-cloud-run). + +Cloud Run worker pools support sidecar containers over localhost and are intended for continuous background work. The deployment should start the collector before the Temporal worker and use the collector health extension as its startup probe. + +To use an external collector instead, set `OTEL_EXPORTER_OTLP_ENDPOINT` or call `Builder.setEndpoint(...)`. + +To use an application-owned provider, call `Builder.setOpenTelemetry(...)`. In that path, no exporters are created; the plugin installs the Temporal metrics scope, tracing interceptors, and shutdown flush hook around the supplied provider. diff --git a/contrib/temporal-gcp/build.gradle b/contrib/temporal-gcp/build.gradle new file mode 100644 index 000000000..233176176 --- /dev/null +++ b/contrib/temporal-gcp/build.gradle @@ -0,0 +1,17 @@ +description = '''Temporal Java SDK Google Cloud 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-serviceclient') + compileOnly project(':temporal-sdk') + compileOnly "javax.annotation:javax.annotation-api:$annotationApiVersion" + + api project(':temporal-opentelemetry') + + testImplementation project(':temporal-sdk') + testImplementation project(':temporal-serviceclient') + testImplementation "junit:junit:${junitVersion}" + + testRuntimeOnly group: 'ch.qos.logback', name: 'logback-classic', version: "${logbackVersion}" +} diff --git a/contrib/temporal-gcp/src/main/java/io/temporal/gcp/GcpOpenTelemetryPlugin.java b/contrib/temporal-gcp/src/main/java/io/temporal/gcp/GcpOpenTelemetryPlugin.java new file mode 100644 index 000000000..8581677c6 --- /dev/null +++ b/contrib/temporal-gcp/src/main/java/io/temporal/gcp/GcpOpenTelemetryPlugin.java @@ -0,0 +1,180 @@ +package io.temporal.gcp; + +import io.opentelemetry.api.OpenTelemetry; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.Experimental; +import io.temporal.common.SimplePlugin; +import io.temporal.opentelemetry.OpenTelemetryPlugin; +import io.temporal.opentelemetry.OpenTelemetryWorker; +import io.temporal.opentelemetry.TimedShutdownHook; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; +import io.temporal.worker.WorkerFactory; +import io.temporal.worker.WorkerFactoryOptions; +import java.time.Duration; +import java.util.Map; +import java.util.Objects; +import java.util.function.Consumer; +import javax.annotation.Nonnull; + +/** OpenTelemetry plugin with defaults for Temporal workers running on Google Cloud Run. */ +@Experimental +public final class GcpOpenTelemetryPlugin extends SimplePlugin { + public static final String NAME = OpenTelemetryPlugin.NAME; + public static final String OTEL_EXPORTER_OTLP_ENDPOINT = + OpenTelemetryWorker.OTEL_EXPORTER_OTLP_ENDPOINT; + public static final String OTEL_SERVICE_NAME = OpenTelemetryWorker.OTEL_SERVICE_NAME; + public static final String CLOUD_RUN_WORKER_POOL = "CLOUD_RUN_WORKER_POOL"; + public static final String K_SERVICE = "K_SERVICE"; + public static final String DEFAULT_OTLP_ENDPOINT = OpenTelemetryWorker.DEFAULT_OTLP_ENDPOINT; + public static final String DEFAULT_SERVICE_NAME = OpenTelemetryWorker.DEFAULT_SERVICE_NAME; + + private final OpenTelemetryPlugin delegate; + + private GcpOpenTelemetryPlugin(Builder builder) { + super(NAME); + this.delegate = builder.buildDelegate(); + } + + public static Builder newBuilder() { + return new Builder(System.getenv()); + } + + public static Builder newBuilder(@Nonnull Map env) { + return new Builder(env); + } + + public String getEndpoint() { + return delegate.getEndpoint(); + } + + public String getServiceName() { + return delegate.getServiceName(); + } + + public OpenTelemetry getOpenTelemetry() { + return delegate.getOpenTelemetry(); + } + + /** + * Creates a flush hook that reports buffered Temporal metrics before force-flushing OpenTelemetry + * providers. + */ + public TimedShutdownHook newFlushHook() { + return delegate.newFlushHook(); + } + + @Override + public void configureServiceStubs(@Nonnull WorkflowServiceStubsOptions.Builder builder) { + delegate.configureServiceStubs(builder); + } + + @Override + public void configureWorkflowClient(@Nonnull WorkflowClientOptions.Builder builder) { + delegate.configureWorkflowClient(builder); + } + + @Override + public void configureWorkerFactory(@Nonnull WorkerFactoryOptions.Builder builder) { + delegate.configureWorkerFactory(builder); + } + + @Override + public void shutdownWorkerFactory( + @Nonnull WorkerFactory factory, @Nonnull Consumer next) { + delegate.shutdownWorkerFactory(factory, next); + } + + /** Builder for {@link GcpOpenTelemetryPlugin}. */ + public static final class Builder { + private final Map env; + private final OpenTelemetryPlugin.Builder delegate; + private String serviceName; + + private Builder(Map env) { + this.env = Objects.requireNonNull(env, "env"); + this.delegate = OpenTelemetryPlugin.newBuilder(env); + } + + /** + * Uses an application-owned OpenTelemetry instance instead of creating an SDK and exporters. + */ + public Builder setOpenTelemetry(@Nonnull OpenTelemetry openTelemetry) { + delegate.setOpenTelemetry(openTelemetry); + return this; + } + + /** Sets the OTLP metric and trace exporter endpoint used by the default SDK setup. */ + public Builder setEndpoint(@Nonnull String endpoint) { + delegate.setEndpoint(endpoint); + return this; + } + + /** Sets the service name used by the default SDK resource and Temporal metrics reporter. */ + public Builder setServiceName(@Nonnull String serviceName) { + this.serviceName = Objects.requireNonNull(serviceName, "serviceName"); + return this; + } + + /** Sets the interval used by the Temporal metrics scope and periodic metric reader. */ + public Builder setMetricsReportInterval(@Nonnull Duration metricsReportInterval) { + delegate.setMetricsReportInterval(metricsReportInterval); + return this; + } + + /** Sets how long the OpenTelemetry flush hook waits for provider flushing. */ + public Builder setFlushTimeout(@Nonnull Duration flushTimeout) { + delegate.setFlushTimeout(flushTimeout); + return this; + } + + /** Overrides the OpenTelemetry provider flush hook. */ + public Builder setFlushHook(@Nonnull Runnable flushHook) { + delegate.setFlushHook(flushHook); + return this; + } + + public String getEndpoint() { + return delegate.getEndpoint(); + } + + public String getServiceName() { + return serviceName == null ? resolveServiceName(env) : serviceName; + } + + public OpenTelemetry createOpenTelemetry() { + applyServiceNameDefault(); + return delegate.createOpenTelemetry(); + } + + public GcpOpenTelemetryPlugin build() { + return new GcpOpenTelemetryPlugin(this); + } + + private OpenTelemetryPlugin buildDelegate() { + applyServiceNameDefault(); + return delegate.build(); + } + + private void applyServiceNameDefault() { + delegate.setServiceName(getServiceName()); + } + } + + static String resolveServiceName(Map env) { + String value = nonEmptyEnv(env, OTEL_SERVICE_NAME); + if (value != null) { + return value; + } + value = nonEmptyEnv(env, CLOUD_RUN_WORKER_POOL); + if (value != null) { + return value; + } + value = nonEmptyEnv(env, K_SERVICE); + return value == null ? DEFAULT_SERVICE_NAME : value; + } + + private static String nonEmptyEnv(Map env, String name) { + String value = env.get(name); + return value == null || value.trim().isEmpty() ? null : value; + } +} diff --git a/contrib/temporal-gcp/src/test/java/io/temporal/gcp/GcpOpenTelemetryPluginTest.java b/contrib/temporal-gcp/src/test/java/io/temporal/gcp/GcpOpenTelemetryPluginTest.java new file mode 100644 index 000000000..f52324902 --- /dev/null +++ b/contrib/temporal-gcp/src/test/java/io/temporal/gcp/GcpOpenTelemetryPluginTest.java @@ -0,0 +1,99 @@ +package io.temporal.gcp; + +import static org.junit.Assert.*; + +import io.opentelemetry.api.OpenTelemetry; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.opentelemetry.OpenTelemetryPlugin; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; +import io.temporal.worker.WorkerFactoryOptions; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Test; + +public class GcpOpenTelemetryPluginTest { + @Test + public void defaultsToLocalCollectorAndGenericServiceName() { + GcpOpenTelemetryPlugin.Builder builder = GcpOpenTelemetryPlugin.newBuilder(new HashMap<>()); + + assertEquals("http://localhost:4317", builder.getEndpoint()); + assertEquals("temporal-worker", builder.getServiceName()); + } + + @Test + public void resolvesCloudRunServiceNameWithExpectedPrecedence() { + Map env = new HashMap<>(); + env.put(GcpOpenTelemetryPlugin.K_SERVICE, "cloud-run-service"); + assertEquals("cloud-run-service", GcpOpenTelemetryPlugin.newBuilder(env).getServiceName()); + + env.put(GcpOpenTelemetryPlugin.CLOUD_RUN_WORKER_POOL, "worker-pool"); + assertEquals("worker-pool", GcpOpenTelemetryPlugin.newBuilder(env).getServiceName()); + + env.put(GcpOpenTelemetryPlugin.OTEL_SERVICE_NAME, "otel-service"); + assertEquals("otel-service", GcpOpenTelemetryPlugin.newBuilder(env).getServiceName()); + + assertEquals( + "builder-service", + GcpOpenTelemetryPlugin.newBuilder(env).setServiceName("builder-service").getServiceName()); + } + + @Test + public void ignoresEmptyEnvironmentValues() { + Map env = new HashMap<>(); + env.put(GcpOpenTelemetryPlugin.OTEL_SERVICE_NAME, " "); + env.put(GcpOpenTelemetryPlugin.CLOUD_RUN_WORKER_POOL, ""); + env.put(GcpOpenTelemetryPlugin.K_SERVICE, "cloud-run-service"); + + assertEquals("cloud-run-service", GcpOpenTelemetryPlugin.newBuilder(env).getServiceName()); + } + + @Test + public void buildAppliesResolvedEndpointAndServiceName() { + Map env = new HashMap<>(); + env.put(GcpOpenTelemetryPlugin.OTEL_EXPORTER_OTLP_ENDPOINT, "http://collector:4317"); + env.put(GcpOpenTelemetryPlugin.CLOUD_RUN_WORKER_POOL, "worker-pool"); + + GcpOpenTelemetryPlugin plugin = + GcpOpenTelemetryPlugin.newBuilder(env).setOpenTelemetry(OpenTelemetry.noop()).build(); + + assertEquals("http://collector:4317", plugin.getEndpoint()); + assertEquals("worker-pool", plugin.getServiceName()); + } + + @Test + public void installsMetricsScopeAndTracingInterceptors() { + GcpOpenTelemetryPlugin plugin = + GcpOpenTelemetryPlugin.newBuilder(new HashMap<>()) + .setOpenTelemetry(OpenTelemetry.noop()) + .build(); + WorkflowServiceStubsOptions.Builder serviceOptions = WorkflowServiceStubsOptions.newBuilder(); + WorkflowClientOptions.Builder clientOptions = WorkflowClientOptions.newBuilder(); + WorkerFactoryOptions.Builder factoryOptions = WorkerFactoryOptions.newBuilder(); + + plugin.configureServiceStubs(serviceOptions); + plugin.configureWorkflowClient(clientOptions); + plugin.configureWorkerFactory(factoryOptions); + + assertEquals(OpenTelemetryPlugin.NAME, plugin.getName()); + assertNotNull(serviceOptions.build().getMetricsScope()); + assertEquals(1, clientOptions.build().getInterceptors().length); + assertEquals(1, factoryOptions.build().getWorkerInterceptors().length); + } + + @Test + public void workerFactoryShutdownFlushesByDefault() { + AtomicInteger flushes = new AtomicInteger(); + AtomicInteger shutdowns = new AtomicInteger(); + GcpOpenTelemetryPlugin plugin = + GcpOpenTelemetryPlugin.newBuilder(new HashMap<>()) + .setOpenTelemetry(OpenTelemetry.noop()) + .setFlushHook(flushes::incrementAndGet) + .build(); + + plugin.shutdownWorkerFactory(null, factory -> shutdowns.incrementAndGet()); + + assertEquals(1, shutdowns.get()); + assertEquals(1, flushes.get()); + } +} diff --git a/settings.gradle b/settings.gradle index 3699ff150..728ccd58f 100644 --- a/settings.gradle +++ b/settings.gradle @@ -8,6 +8,8 @@ include 'temporal-opentracing' project(':temporal-opentracing').projectDir = file('contrib/temporal-opentracing') include 'temporal-opentelemetry' project(':temporal-opentelemetry').projectDir = file('contrib/temporal-opentelemetry') +include 'temporal-gcp' +project(':temporal-gcp').projectDir = file('contrib/temporal-gcp') include 'temporal-kotlin' include 'temporal-spring-ai' project(':temporal-spring-ai').projectDir = file('contrib/temporal-spring-ai') diff --git a/temporal-bom/build.gradle b/temporal-bom/build.gradle index 031633473..56f7de517 100644 --- a/temporal-bom/build.gradle +++ b/temporal-bom/build.gradle @@ -7,6 +7,7 @@ description = '''Temporal Java BOM''' dependencies { constraints { api project(':temporal-kotlin') + api project(':temporal-gcp') api project(':temporal-opentelemetry') api project(':temporal-opentracing') api project(':temporal-aws-lambda') From 01eabca4aa9f46a6291f6c37f83965eb91a76a1a Mon Sep 17 00:00:00 2001 From: Edward Amsden Date: Thu, 16 Jul 2026 16:49:56 -0500 Subject: [PATCH 2/3] Address GCP plugin review feedback --- contrib/temporal-gcp/README.md | 31 ++++++++++++++++--- .../temporal/gcp/GcpOpenTelemetryPlugin.java | 20 ++++++++++-- .../gcp/GcpOpenTelemetryPluginTest.java | 21 ++++++++++++- 3 files changed, 65 insertions(+), 7 deletions(-) diff --git a/contrib/temporal-gcp/README.md b/contrib/temporal-gcp/README.md index dbe6ececb..a3e75b6b0 100644 --- a/contrib/temporal-gcp/README.md +++ b/contrib/temporal-gcp/README.md @@ -1,11 +1,13 @@ # Temporal Google Cloud module -This module provides an OpenTelemetry plugin with defaults for Temporal Java SDK workers running on Google Cloud Run services and worker pools. +This module provides an OpenTelemetry plugin with defaults for Temporal Java SDK workers running on Google Cloud Run. Cloud Run worker pools are the recommended deployment because Temporal workers are continuous, pull-based background workloads. > **Collector required by default:** The plugin exports metrics and traces to an OTLP collector at `http://localhost:4317`. It does not export directly to Google Cloud. Deploy the Google-Built OpenTelemetry Collector as a sidecar, configure another collector endpoint, or provide an application-owned `OpenTelemetry` instance. Without a collector at the configured endpoint, telemetry is not delivered to Google Cloud. This integration is for container-based Cloud Run workloads. It does not implement a Cloud Run functions invocation lifecycle. +A Cloud Run service can also host a Temporal worker, but it must use instance-based billing so CPU is available outside request handling, keep at least one instance active through minimum instances or manual scaling, and run an ingress container that listens on `PORT`. These are deployment requirements; the plugin cannot configure them from inside the worker process. + ## Usage Add `temporal-gcp` next to your Temporal SDK dependency, then install the plugin on service stubs options before creating clients and workers: @@ -22,7 +24,28 @@ WorkflowClient client = WorkflowClient.newInstance(service); WorkerFactory factory = WorkerFactory.newInstance(client); ``` -The plugin configures the SDK metrics scope, tracing interceptors, OTLP metric and trace exporters, and worker-factory shutdown flushing through `temporal-opentelemetry`. Do not install both `GcpOpenTelemetryPlugin` and `OpenTelemetryPlugin` on the same service stubs. +The plugin configures the SDK metrics scope, tracing interceptors, and OTLP metric and trace exporters through `temporal-opentelemetry`. Do not install both `GcpOpenTelemetryPlugin` and `OpenTelemetryPlugin` on the same service stubs. + +## Shutdown lifecycle + +`WorkerFactory.shutdown()` initiates asynchronous shutdown. The GCP plugin therefore does not flush from the worker-factory shutdown callback by default: doing so could miss telemetry emitted while activities and workflows finish. Wait for the factory to terminate, then flush with the time remaining before Cloud Run sends `SIGKILL`. For example, the following JVM shutdown hook reserves six seconds for graceful shutdown, one second for forced shutdown, and two seconds for telemetry flushing within Cloud Run's ten-second termination window: + +```java +Runtime.getRuntime() + .addShutdownHook( + new Thread( + () -> { + factory.shutdown(); + factory.awaitTermination(6, TimeUnit.SECONDS); + if (!factory.isTerminated()) { + factory.shutdownNow(); + factory.awaitTermination(1, TimeUnit.SECONDS); + } + plugin.newFlushHook().run(Duration.ofSeconds(2)); + })); +``` + +Applications with an existing lifecycle manager should perform the same sequence there instead of registering another JVM hook. `Builder.setFlushOnWorkerFactoryShutdown(true)` restores the underlying plugin's immediate, best-effort flush, but it should only be used when no work can emit telemetry after the shutdown request. The OTLP endpoint is resolved in this order: @@ -38,11 +61,11 @@ The OpenTelemetry service name is resolved in this order: 4. `K_SERVICE` for a Cloud Run service. 5. `temporal-worker`. -The collector should use its GCP resource detector to add the Google Cloud project, location, revision, and monitored-resource attributes. This module does not call the Google Cloud metadata server and adds no Google Cloud client libraries or exporters to the worker process. +The collector should use its GCP resource detector to add the Google Cloud attributes it recognizes. Do not rely on the detector to infer Cloud Run worker-pool-specific location or revision attributes; configure those explicitly with a collector resource processor if they are required. This module does not call the Google Cloud metadata server and adds no Google Cloud client libraries or exporters to the worker process. ## Collector sidecar -Google publishes the Google-Built OpenTelemetry Collector as a container image. Configure it as a second Cloud Run container, listen for OTLP gRPC on `localhost:4317`, and use its GCP exporters for metrics and traces. For the image, recommended collector configuration, IAM roles, health check, and Secret Manager mount, see [Deploy Google-Built OpenTelemetry Collector on Cloud Run](https://cloud.google.com/stackdriver/docs/instrumentation/opentelemetry-collector-cloud-run). +Google publishes the Google-Built OpenTelemetry Collector as a container image. Configure it as a second Cloud Run container, listen for OTLP gRPC on `localhost:4317`, and use its GCP exporters for metrics and traces. For the image, recommended collector configuration, IAM roles, health check, and Secret Manager mount, see [Deploy Google-Built OpenTelemetry Collector on Cloud Run](https://cloud.google.com/stackdriver/docs/instrumentation/opentelemetry-collector-cloud-run). That guide demonstrates a Cloud Run service; adapt its collector container and configuration when deploying a worker pool. Cloud Run worker pools support sidecar containers over localhost and are intended for continuous background work. The deployment should start the collector before the Temporal worker and use the collector health extension as its startup probe. diff --git a/contrib/temporal-gcp/src/main/java/io/temporal/gcp/GcpOpenTelemetryPlugin.java b/contrib/temporal-gcp/src/main/java/io/temporal/gcp/GcpOpenTelemetryPlugin.java index 8581677c6..f86c5bdfe 100644 --- a/contrib/temporal-gcp/src/main/java/io/temporal/gcp/GcpOpenTelemetryPlugin.java +++ b/contrib/temporal-gcp/src/main/java/io/temporal/gcp/GcpOpenTelemetryPlugin.java @@ -16,7 +16,10 @@ import java.util.function.Consumer; import javax.annotation.Nonnull; -/** OpenTelemetry plugin with defaults for Temporal workers running on Google Cloud Run. */ +/** + * OpenTelemetry plugin with defaults for Temporal workers running on Google Cloud Run, primarily in + * worker pools. + */ @Experimental public final class GcpOpenTelemetryPlugin extends SimplePlugin { public static final String NAME = OpenTelemetryPlugin.NAME; @@ -92,7 +95,7 @@ public static final class Builder { private Builder(Map env) { this.env = Objects.requireNonNull(env, "env"); - this.delegate = OpenTelemetryPlugin.newBuilder(env); + this.delegate = OpenTelemetryPlugin.newBuilder(env).setFlushOnWorkerFactoryShutdown(false); } /** @@ -133,6 +136,19 @@ public Builder setFlushHook(@Nonnull Runnable flushHook) { return this; } + /** + * Controls whether the plugin flushes immediately after {@link WorkerFactory#shutdown()} or + * {@link WorkerFactory#shutdownNow()} initiates shutdown. + * + *

This is disabled by default because worker-factory shutdown is asynchronous. Cloud Run + * applications should wait for worker termination and then run {@link + * GcpOpenTelemetryPlugin#newFlushHook()} with the remaining shutdown time. + */ + public Builder setFlushOnWorkerFactoryShutdown(boolean flushOnWorkerFactoryShutdown) { + delegate.setFlushOnWorkerFactoryShutdown(flushOnWorkerFactoryShutdown); + return this; + } + public String getEndpoint() { return delegate.getEndpoint(); } diff --git a/contrib/temporal-gcp/src/test/java/io/temporal/gcp/GcpOpenTelemetryPluginTest.java b/contrib/temporal-gcp/src/test/java/io/temporal/gcp/GcpOpenTelemetryPluginTest.java index f52324902..d815840ec 100644 --- a/contrib/temporal-gcp/src/test/java/io/temporal/gcp/GcpOpenTelemetryPluginTest.java +++ b/contrib/temporal-gcp/src/test/java/io/temporal/gcp/GcpOpenTelemetryPluginTest.java @@ -82,7 +82,7 @@ public void installsMetricsScopeAndTracingInterceptors() { } @Test - public void workerFactoryShutdownFlushesByDefault() { + public void workerFactoryShutdownDefersFlushByDefault() { AtomicInteger flushes = new AtomicInteger(); AtomicInteger shutdowns = new AtomicInteger(); GcpOpenTelemetryPlugin plugin = @@ -94,6 +94,25 @@ public void workerFactoryShutdownFlushesByDefault() { plugin.shutdownWorkerFactory(null, factory -> shutdowns.incrementAndGet()); assertEquals(1, shutdowns.get()); + assertEquals(0, flushes.get()); + + plugin.newFlushHook().run(); + + assertEquals(1, flushes.get()); + } + + @Test + public void workerFactoryShutdownFlushCanBeEnabled() { + AtomicInteger flushes = new AtomicInteger(); + GcpOpenTelemetryPlugin plugin = + GcpOpenTelemetryPlugin.newBuilder(new HashMap<>()) + .setOpenTelemetry(OpenTelemetry.noop()) + .setFlushHook(flushes::incrementAndGet) + .setFlushOnWorkerFactoryShutdown(true) + .build(); + + plugin.shutdownWorkerFactory(null, factory -> {}); + assertEquals(1, flushes.get()); } } From 4bc57576b7fec33839fa45a0a2980bbe045d872d Mon Sep 17 00:00:00 2001 From: Edward Amsden Date: Fri, 17 Jul 2026 16:19:48 -0500 Subject: [PATCH 3/3] Use safe GCP metrics export settings Use the coordinated 60-second metrics interval for the GCP plugin while preserving explicit overrides. Document an unbatched Managed Prometheus metrics pipeline so periodic exports and forced shutdown flushes cannot be combined into a duplicate-series request. --- contrib/temporal-gcp/README.md | 25 +++++++++++++++++++ .../temporal/gcp/GcpOpenTelemetryPlugin.java | 17 +++++++++++-- .../gcp/GcpOpenTelemetryPluginTest.java | 2 ++ .../opentelemetry/OpenTelemetryPlugin.java | 4 +++ 4 files changed, 46 insertions(+), 2 deletions(-) diff --git a/contrib/temporal-gcp/README.md b/contrib/temporal-gcp/README.md index a3e75b6b0..83ca73539 100644 --- a/contrib/temporal-gcp/README.md +++ b/contrib/temporal-gcp/README.md @@ -53,6 +53,10 @@ The OTLP endpoint is resolved in this order: 2. `OTEL_EXPORTER_OTLP_ENDPOINT`. 3. `http://localhost:4317`. +When the plugin creates the OpenTelemetry SDK, metrics are reported and exported every sixty seconds by default. This matches the coordinated GCP plugin default across Temporal SDKs and exceeds Google Cloud's five-second minimum export interval. If you use `Builder.setMetricsReportInterval(...)`, keep the interval above that minimum. The collector also needs the unbatched metrics pipeline described below to make a forced shutdown flush safe regardless of its timing relative to the last periodic export. + +With an application-owned `OpenTelemetry` instance, the setting only controls how often the Temporal metrics scope reports into that instance. Configure the instance's metric reader to export at an interval above the Google Cloud minimum as well. + The OpenTelemetry service name is resolved in this order: 1. `Builder.setServiceName(...)`. @@ -67,6 +71,27 @@ The collector should use its GCP resource detector to add the Google Cloud attri Google publishes the Google-Built OpenTelemetry Collector as a container image. Configure it as a second Cloud Run container, listen for OTLP gRPC on `localhost:4317`, and use its GCP exporters for metrics and traces. For the image, recommended collector configuration, IAM roles, health check, and Secret Manager mount, see [Deploy Google-Built OpenTelemetry Collector on Cloud Run](https://cloud.google.com/stackdriver/docs/instrumentation/opentelemetry-collector-cloud-run). That guide demonstrates a Cloud Run service; adapt its collector container and configuration when deploying a worker pool. +Do not put a batch processor in the `googlemanagedprometheus` metrics pipeline. A periodic cumulative metric export followed closely by a forced shutdown flush can otherwise put two points for the same time series in one request, which Managed Service for Prometheus rejects. Pass metrics through the memory limiter, GCP resource detection, and any collision transforms directly to `googlemanagedprometheus`. Keep a dedicated five-second batch processor on the traces pipeline: + +```yaml +processors: + batch/traces: + send_batch_max_size: 200 + send_batch_size: 200 + timeout: 5s + +service: + pipelines: + metrics: + 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] +``` + Cloud Run worker pools support sidecar containers over localhost and are intended for continuous background work. The deployment should start the collector before the Temporal worker and use the collector health extension as its startup probe. To use an external collector instead, set `OTEL_EXPORTER_OTLP_ENDPOINT` or call `Builder.setEndpoint(...)`. diff --git a/contrib/temporal-gcp/src/main/java/io/temporal/gcp/GcpOpenTelemetryPlugin.java b/contrib/temporal-gcp/src/main/java/io/temporal/gcp/GcpOpenTelemetryPlugin.java index f86c5bdfe..5e7460597 100644 --- a/contrib/temporal-gcp/src/main/java/io/temporal/gcp/GcpOpenTelemetryPlugin.java +++ b/contrib/temporal-gcp/src/main/java/io/temporal/gcp/GcpOpenTelemetryPlugin.java @@ -30,6 +30,7 @@ public final class GcpOpenTelemetryPlugin extends SimplePlugin { public static final String K_SERVICE = "K_SERVICE"; public static final String DEFAULT_OTLP_ENDPOINT = OpenTelemetryWorker.DEFAULT_OTLP_ENDPOINT; public static final String DEFAULT_SERVICE_NAME = OpenTelemetryWorker.DEFAULT_SERVICE_NAME; + public static final Duration DEFAULT_METRICS_REPORT_INTERVAL = Duration.ofSeconds(60); private final OpenTelemetryPlugin delegate; @@ -95,7 +96,10 @@ public static final class Builder { private Builder(Map env) { this.env = Objects.requireNonNull(env, "env"); - this.delegate = OpenTelemetryPlugin.newBuilder(env).setFlushOnWorkerFactoryShutdown(false); + this.delegate = + OpenTelemetryPlugin.newBuilder(env) + .setMetricsReportInterval(DEFAULT_METRICS_REPORT_INTERVAL) + .setFlushOnWorkerFactoryShutdown(false); } /** @@ -118,7 +122,12 @@ public Builder setServiceName(@Nonnull String serviceName) { return this; } - /** Sets the interval used by the Temporal metrics scope and periodic metric reader. */ + /** + * Sets the interval used by the Temporal metrics scope and, when the plugin creates the + * OpenTelemetry instance, its periodic metric reader. + * + *

An application-owned OpenTelemetry instance must configure its metric reader separately. + */ public Builder setMetricsReportInterval(@Nonnull Duration metricsReportInterval) { delegate.setMetricsReportInterval(metricsReportInterval); return this; @@ -157,6 +166,10 @@ public String getServiceName() { return serviceName == null ? resolveServiceName(env) : serviceName; } + public Duration getMetricsReportInterval() { + return delegate.getMetricsReportInterval(); + } + public OpenTelemetry createOpenTelemetry() { applyServiceNameDefault(); return delegate.createOpenTelemetry(); diff --git a/contrib/temporal-gcp/src/test/java/io/temporal/gcp/GcpOpenTelemetryPluginTest.java b/contrib/temporal-gcp/src/test/java/io/temporal/gcp/GcpOpenTelemetryPluginTest.java index d815840ec..44c84e351 100644 --- a/contrib/temporal-gcp/src/test/java/io/temporal/gcp/GcpOpenTelemetryPluginTest.java +++ b/contrib/temporal-gcp/src/test/java/io/temporal/gcp/GcpOpenTelemetryPluginTest.java @@ -7,6 +7,7 @@ import io.temporal.opentelemetry.OpenTelemetryPlugin; import io.temporal.serviceclient.WorkflowServiceStubsOptions; import io.temporal.worker.WorkerFactoryOptions; +import java.time.Duration; import java.util.HashMap; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; @@ -19,6 +20,7 @@ public void defaultsToLocalCollectorAndGenericServiceName() { assertEquals("http://localhost:4317", builder.getEndpoint()); assertEquals("temporal-worker", builder.getServiceName()); + assertEquals(Duration.ofSeconds(60), builder.getMetricsReportInterval()); } @Test diff --git a/contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryPlugin.java b/contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryPlugin.java index 2604e205f..3b1e2f160 100644 --- a/contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryPlugin.java +++ b/contrib/temporal-opentelemetry/src/main/java/io/temporal/opentelemetry/OpenTelemetryPlugin.java @@ -235,6 +235,10 @@ public String getServiceName() { return serviceName == null ? OpenTelemetryWorker.resolveServiceName(env) : serviceName; } + public Duration getMetricsReportInterval() { + return metricsReportInterval; + } + Builder setTelemetryFactory(OpenTelemetryWorker.TelemetryFactory telemetryFactory) { this.telemetryFactory = Objects.requireNonNull(telemetryFactory, "telemetryFactory"); return this;