Add OpenTelemetry v2 plugin

CHANGELOG.md

@@ -56,6 +56,10 @@ to docs, or any other relevant information.
 - Added experimental `ActivityClientPlugin` and `NexusClientPlugin`, which plugins on the service stubs use to configure
   `ActivityClient` and `NexusClient`. `SimplePlugin` implements both, and its builder gained
   `addScheduleClientInterceptors`, `addActivityClientInterceptors`, and `addNexusClientInterceptors`.
+- Added the experimental `temporal-opentelemetry-v2` module. `OpenTelemetryPlugin` propagates OpenTelemetry context
+  through Temporal headers and can optionally create spans for Temporal operations. `ReplaySafeOpenTelemetry` makes
+  workflow spans replay safe with deterministic span and trace IDs, and drops metric recordings and log records made
+  by replaying workflow code.
 
 ### Changed
 - Moved the Spring AI integration to [temporalio/ai-integrations](https://github.com/temporalio/ai-integrations/tree/main/java/spring-ai).

contrib/temporal-opentelemetry-v2/README.md

@@ -0,0 +1,164 @@
+# OpenTelemetry v2 integration for the Temporal Java SDK
+
+Module `io.temporal:temporal-opentelemetry-v2` provides replay-safe
+OpenTelemetry tracing, metrics, and logs for Temporal.
+
+> [!WARNING]
+> This module is experimental, and its API may change.
+
+## Setup
+
+Use the same version as the rest of your Temporal Java SDK dependencies:
+
+```groovy
+implementation 'io.temporal:temporal-opentelemetry-v2:<temporal-java-sdk-version>'
+// Add the exporters you use, such as this OTLP exporter.
+implementation 'io.opentelemetry:opentelemetry-exporter-otlp'
+```
+
+Create a replay-safe OpenTelemetry instance, register it as the global, and
+attach the plugin to your service stubs:
+
+```java
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.exporter.otlp.logs.OtlpGrpcLogRecordExporter;
+import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter;
+import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
+import io.opentelemetry.sdk.logs.SdkLoggerProvider;
+import io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor;
+import io.opentelemetry.sdk.metrics.SdkMeterProvider;
+import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
+import io.temporal.client.WorkflowClient;
+import io.temporal.opentelemetry.v2.OpenTelemetryPlugin;
+import io.temporal.opentelemetry.v2.ReplaySafeOpenTelemetry;
+import io.temporal.serviceclient.WorkflowServiceStubs;
+import io.temporal.serviceclient.WorkflowServiceStubsOptions;
+import io.temporal.worker.WorkerFactory;
+
+ReplaySafeOpenTelemetry openTelemetry =
+    ReplaySafeOpenTelemetry.newBuilder()
+        .setTracerProviderBuilder(
+            SdkTracerProvider.builder()
+                .addSpanProcessor(
+                    BatchSpanProcessor.builder(OtlpGrpcSpanExporter.builder().build()).build()))
+        .setMeterProviderBuilder(
+            SdkMeterProvider.builder()
+                .registerMetricReader(
+                    PeriodicMetricReader.builder(OtlpGrpcMetricExporter.builder().build())
+                        .build()))
+        .setLoggerProviderBuilder(
+            SdkLoggerProvider.builder()
+                .addLogRecordProcessor(
+                    BatchLogRecordProcessor.builder(OtlpGrpcLogRecordExporter.builder().build())
+                        .build()))
+        .build();
+GlobalOpenTelemetry.set(openTelemetry);
+
+WorkflowServiceStubs service =
+    WorkflowServiceStubs.newServiceStubs(
+        WorkflowServiceStubsOptions.newBuilder()
+            .setPlugins(OpenTelemetryPlugin.newBuilder().build())
+            .build());
+
+WorkflowClient client = WorkflowClient.newInstance(service);
+WorkerFactory factory = WorkerFactory.newInstance(client);
+```
+
+Plugins configured on `WorkflowServiceStubsOptions` propagate to clients and
+workers created from those stubs. It can also be configured directly on
+`WorkflowClientOptions`.
+
+Build `ReplaySafeOpenTelemetry` before anything else uses the OpenTelemetry
+`Context`, and register it as the global before building the plugin. Otherwise,
+`OpenTelemetryPlugin.Builder.build()` throws `IllegalStateException`.
+
+Call `ReplaySafeOpenTelemetry.close()` after your workers and clients have
+stopped to export buffered telemetry and shut down its providers.
+
+## Tracing
+
+The plugin propagates application trace context through Temporal headers.
+Application spans remain connected across clients, workflows, activities, and
+Nexus operations.
+
+Set `OpenTelemetryPlugin.Builder.setAddTemporalSpans(true)` to emit spans for
+operations such as `StartWorkflow`, `RunWorkflow`, `RunActivity`, and
+`ContinueAsNew`.
+
+Create spans in workflow, client, and activity code with the standard OpenTelemetry API:
+
+```java
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Scope;
+
+Span span =
+    GlobalOpenTelemetry.getTracer("my-workflows")
+        .spanBuilder("my-span")
+        .startSpan();
+try (Scope ignored = span.makeCurrent()) {
+  activity.doWork();
+} finally {
+  span.end();
+}
+```
+
+## Metrics
+
+Create metrics in workflow, client, and activity code with the standard
+OpenTelemetry API. Workflow code supports only synchronous instruments:
+
+```java
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.metrics.LongCounter;
+
+LongCounter counter =
+    GlobalOpenTelemetry.getMeter("my-workflows")
+        .counterBuilder("workflow.items.processed")
+        .build();
+counter.add(1);
+```
+
+## Logs
+
+Create logs in workflow, client, and activity code with SLF4J, and send them to
+OpenTelemetry by installing a logging appender, such as the
+[OpenTelemetry Logback appender](https://github.com/open-telemetry/opentelemetry-java-instrumentation/tree/main/instrumentation/logback/logback-appender-1.0/library),
+with the replay-safe instance. The appender drops records from replaying
+workflow code only when it is installed with `ReplaySafeOpenTelemetry`, not a
+plain `OpenTelemetrySdk`:
+
+```java
+import io.opentelemetry.exporter.otlp.logs.OtlpGrpcLogRecordExporter;
+import io.opentelemetry.instrumentation.logback.appender.v1_0.OpenTelemetryAppender;
+import io.opentelemetry.sdk.logs.SdkLoggerProvider;
+import io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor;
+import io.temporal.opentelemetry.v2.ReplaySafeOpenTelemetry;
+
+ReplaySafeOpenTelemetry openTelemetry =
+    ReplaySafeOpenTelemetry.newBuilder()
+        .setLoggerProviderBuilder(
+            SdkLoggerProvider.builder()
+                .addLogRecordProcessor(
+                    BatchLogRecordProcessor.builder(OtlpGrpcLogRecordExporter.builder().build())
+                        .build()))
+        .build();
+OpenTelemetryAppender.install(openTelemetry);
+```
+
+Then log with SLF4J:
+
+```java
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+Logger logger = LoggerFactory.getLogger(MyWorkflowImpl.class);
+logger.info("workflow step completed");
+```
+
+Configure the appender as its setup guide describes. Replay is checked on the
+thread that emits the record, so behind an asynchronous appender, such as
+Logback's `AsyncAppender`, log from workflow code with `Workflow.getLogger()`,
+which skips logging during replay.

contrib/temporal-opentelemetry-v2/build.gradle

@@ -0,0 +1,29 @@
+description = '''Temporal Java SDK OpenTelemetry v2 Module'''
+
+ext {
+    otelVersion = '1.66.0'
+    otelInstrumentationVersion = '2.32.0-alpha'
+}
+
+dependencies {
+    api platform("io.opentelemetry:opentelemetry-bom:$otelVersion")
+
+    compileOnly project(':temporal-serviceclient')
+    compileOnly project(':temporal-sdk')
+    compileOnly "javax.annotation:javax.annotation-api:$annotationApiVersion"
+
+    implementation "com.google.guava:guava:$guavaVersion"
+
+    api "io.opentelemetry:opentelemetry-api"
+    api "io.opentelemetry:opentelemetry-sdk-trace"
+    api "io.opentelemetry:opentelemetry-sdk-metrics"
+    api "io.opentelemetry:opentelemetry-sdk-logs"
+
+    testImplementation project(':temporal-sdk')
+    testImplementation project(':temporal-serviceclient')
+    testImplementation project(':temporal-testing')
+    testImplementation "io.opentelemetry:opentelemetry-sdk-testing"
+    testImplementation "io.opentelemetry.instrumentation:opentelemetry-logback-appender-1.0:$otelInstrumentationVersion"
+    testImplementation "junit:junit:${junitVersion}"
+    testImplementation group: 'ch.qos.logback', name: 'logback-classic', version: "${logbackVersion}"
+}

contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryActivityClientInterceptor.java

@@ -0,0 +1,20 @@
+package io.temporal.opentelemetry.v2;
+
+import io.temporal.common.interceptors.ActivityClientCallsInterceptor;
+import io.temporal.common.interceptors.ActivityClientInterceptorBase;
+import io.temporal.opentelemetry.v2.internal.InterceptorTracer;
+import io.temporal.opentelemetry.v2.internal.OpenTelemetryActivityClientCallsInterceptor;
+
+public class OpenTelemetryActivityClientInterceptor extends ActivityClientInterceptorBase {
+  private final InterceptorTracer tracer;
+
+  public OpenTelemetryActivityClientInterceptor(InterceptorTracer tracer) {
+    this.tracer = tracer;
+  }
+
+  @Override
+  public ActivityClientCallsInterceptor activityClientCallsInterceptor(
+      ActivityClientCallsInterceptor next) {
+    return new OpenTelemetryActivityClientCallsInterceptor(tracer, next);
+  }
+}

contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryClientInterceptor.java

@@ -0,0 +1,20 @@
+package io.temporal.opentelemetry.v2;
+
+import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
+import io.temporal.common.interceptors.WorkflowClientInterceptorBase;
+import io.temporal.opentelemetry.v2.internal.InterceptorTracer;
+import io.temporal.opentelemetry.v2.internal.OpenTelemetryWorkflowClientCallsInterceptor;
+
+public class OpenTelemetryClientInterceptor extends WorkflowClientInterceptorBase {
+  private final InterceptorTracer tracer;
+
+  public OpenTelemetryClientInterceptor(InterceptorTracer tracer) {
+    this.tracer = tracer;
+  }
+
+  @Override
+  public WorkflowClientCallsInterceptor workflowClientCallsInterceptor(
+      WorkflowClientCallsInterceptor next) {
+    return new OpenTelemetryWorkflowClientCallsInterceptor(tracer, next);
+  }
+}

contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryNexusClientInterceptor.java

@@ -0,0 +1,19 @@
+package io.temporal.opentelemetry.v2;
+
+import io.temporal.common.interceptors.NexusClientCallsInterceptor;
+import io.temporal.common.interceptors.NexusClientInterceptorBase;
+import io.temporal.opentelemetry.v2.internal.InterceptorTracer;
+import io.temporal.opentelemetry.v2.internal.OpenTelemetryNexusClientCallsInterceptor;
+
+public class OpenTelemetryNexusClientInterceptor extends NexusClientInterceptorBase {
+  private final InterceptorTracer tracer;
+
+  public OpenTelemetryNexusClientInterceptor(InterceptorTracer tracer) {
+    this.tracer = tracer;
+  }
+
+  @Override
+  public NexusClientCallsInterceptor nexusClientCallsInterceptor(NexusClientCallsInterceptor next) {
+    return new OpenTelemetryNexusClientCallsInterceptor(tracer, next);
+  }
+}

contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryPlugin.java

@@ -0,0 +1,83 @@
+package io.temporal.opentelemetry.v2;
+
+import io.opentelemetry.context.ContextStorage;
+import io.temporal.common.Experimental;
+import io.temporal.common.SimplePlugin;
+import io.temporal.opentelemetry.v2.internal.InterceptorTracer;
+import io.temporal.opentelemetry.v2.internal.OpenTelemetryContextPropagator;
+import io.temporal.opentelemetry.v2.internal.TemporalContextStorage;
+
+/**
+ * OpenTelemetry v2 plugin for Temporal clients and workers.
+ *
+ * <p>Set it on {@code WorkflowServiceStubsOptions} or {@code WorkflowClientOptions}. The SDK
+ * propagates plugins down that chain.
+ *
+ * <p>Register a {@link ReplaySafeOpenTelemetry} with {@code GlobalOpenTelemetry.set} before calling
+ * {@link Builder#build()}.
+ */
+@Experimental
+public final class OpenTelemetryPlugin extends SimplePlugin {
+  public static final String NAME = "io.temporal.opentelemetry.v2";
+
+  private OpenTelemetryPlugin(InterceptorTracer tracer, String headerKey) {
+    super(
+        SimplePlugin.newBuilder(NAME)
+            .addClientInterceptors(new OpenTelemetryClientInterceptor(tracer))
+            .addScheduleClientInterceptors(new OpenTelemetryScheduleClientInterceptor(tracer))
+            .addActivityClientInterceptors(new OpenTelemetryActivityClientInterceptor(tracer))
+            .addNexusClientInterceptors(new OpenTelemetryNexusClientInterceptor(tracer))
+            .addWorkerInterceptors(new OpenTelemetryWorkerInterceptor(tracer))
+            .addContextPropagators(new OpenTelemetryContextPropagator(headerKey)));
+  }
+
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  /** Every option is optional; an unset one keeps its default. */
+  public static final class Builder {
+    private String headerKey = "_tracer-data";
+    private boolean addTemporalSpans;
+
+    private Builder() {}
+
+    /**
+     * The Temporal header key to serialize the span to. Defaults to {@code _tracer-data}, which
+     * Temporal uses; overriding it breaks trace continuity with workers using the standard key.
+     */
+    public Builder setHeaderKey(String headerKey) {
+      this.headerKey = headerKey;
+      return this;
+    }
+
+    /**
+     * Whether to create spans for Temporal operations such as StartWorkflow, RunWorkflow, and
+     * RunActivity. Defaults to false: trace context still propagates through Temporal headers, so
+     * spans created by application code remain connected.
+     */
+    public Builder setAddTemporalSpans(boolean addTemporalSpans) {
+      this.addTemporalSpans = addTemporalSpans;
+      return this;
+    }
+
+    public OpenTelemetryPlugin build() {
+      if (!ReplaySafeOpenTelemetry.isRegisteredGlobally()) {
+        throw new IllegalStateException(
+            "the global OpenTelemetry must be a ReplaySafeOpenTelemetry; build one with "
+                + "ReplaySafeOpenTelemetry.newBuilder() and register it with "
+                + "GlobalOpenTelemetry.set before building this plugin");
+      }
+      // Initializing the storage applies every registered wrapper, and a wrapper registered after
+      // ours ends up outermost, so check that ours was applied rather than what is outermost.
+      ContextStorage.get();
+      if (!TemporalContextStorage.isInstalled()) {
+        throw new IllegalStateException(
+            "the OpenTelemetry ContextStorage initialized before ReplaySafeOpenTelemetry could "
+                + "wrap it; build ReplaySafeOpenTelemetry before anything uses the OpenTelemetry "
+                + "Context");
+      }
+      return new OpenTelemetryPlugin(new InterceptorTracer(headerKey, addTemporalSpans), headerKey);
+    }
+  }
+}

contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryScheduleClientInterceptor.java

@@ -0,0 +1,20 @@
+package io.temporal.opentelemetry.v2;
+
+import io.temporal.common.interceptors.ScheduleClientCallsInterceptor;
+import io.temporal.common.interceptors.ScheduleClientInterceptorBase;
+import io.temporal.opentelemetry.v2.internal.InterceptorTracer;
+import io.temporal.opentelemetry.v2.internal.OpenTelemetryScheduleClientCallsInterceptor;
+
+public class OpenTelemetryScheduleClientInterceptor extends ScheduleClientInterceptorBase {
+  private final InterceptorTracer tracer;
+
+  public OpenTelemetryScheduleClientInterceptor(InterceptorTracer tracer) {
+    this.tracer = tracer;
+  }
+
+  @Override
+  public ScheduleClientCallsInterceptor scheduleClientCallsInterceptor(
+      ScheduleClientCallsInterceptor next) {
+    return new OpenTelemetryScheduleClientCallsInterceptor(tracer, next);
+  }
+}