Make the OpenTelemetry v2 logger provider replay safe

CHANGELOG.md

@@ -36,8 +36,8 @@ to docs, or any other relevant information.
   `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 made by replaying
-  workflow code.
+  workflow spans replay safe with deterministic span and trace IDs, and drops metric recordings and log records made
+  by replaying workflow code.
 
 ### Changed
 - Release notes for all future releases are now in a single CHANGELOG.md file. `releases` directory with old release

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

@@ -2,6 +2,7 @@ description = '''Temporal Java SDK OpenTelemetry v2 Module'''
 
 ext {
     otelVersion = '1.66.0'
+    otelInstrumentationVersion = '2.32.0-alpha'
 }
 
 dependencies {
@@ -22,7 +23,7 @@ dependencies {
     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}"
-
-    testRuntimeOnly group: 'ch.qos.logback', name: 'logback-classic', version: "${logbackVersion}"
+    testImplementation group: 'ch.qos.logback', name: 'logback-classic', version: "${logbackVersion}"
 }

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

@@ -3,6 +3,8 @@
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.baggage.propagation.W3CBaggagePropagator;
+import io.opentelemetry.api.logs.Logger;
+import io.opentelemetry.api.logs.LoggerBuilder;
 import io.opentelemetry.api.logs.LoggerProvider;
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.api.metrics.MeterBuilder;
@@ -23,6 +25,7 @@
 import io.opentelemetry.sdk.trace.SdkTracerProviderBuilder;
 import io.temporal.common.Experimental;
 import io.temporal.opentelemetry.v2.internal.ReplaySafeIdGenerator;
+import io.temporal.opentelemetry.v2.internal.ReplaySafeLogger;
 import io.temporal.opentelemetry.v2.internal.ReplaySafeMeter;
 import io.temporal.opentelemetry.v2.internal.ReplaySafeTracer;
 import io.temporal.opentelemetry.v2.internal.TemporalContextStorage;
@@ -31,8 +34,8 @@
 
 /**
  * The {@link OpenTelemetry} to use for OpenTelemetry integration with Temporal. Register it with
- * {@code GlobalOpenTelemetry.set}; tracers and meters obtained from it are replay safe inside
- * workflows.
+ * {@code GlobalOpenTelemetry.set}; tracers, meters, and loggers obtained from it are replay safe
+ * inside workflows.
  *
  * <p>Known limitations:
  *
@@ -51,17 +54,15 @@
  *       Promise.allOf} and {@code Promise.anyOf}, run with the context current when the promise
  *       completes.
  *   <li>Continuations on the promise returned by cancelling an external workflow run outside a
- *       workflow thread, so spans they start get random IDs, and their spans and metric recordings
- *       are exported again on every replay.
+ *       workflow thread, so spans they start get random IDs, and their spans, metric recordings,
+ *       and log records are exported again on every replay.
  * </ul>
  */
 @Experimental
 public final class ReplaySafeOpenTelemetry implements OpenTelemetry, Closeable {
   private final ReplaySafeTracerProvider tracerProvider;
   private final ReplaySafeMeterProvider meterProvider;
-  // TODO: Make the logger provider replay safe and add logger interceptor methods for Temporal and
-  // OpenTelemetry loggers.
-  private final SdkLoggerProvider loggerProvider;
+  private final ReplaySafeLoggerProvider loggerProvider;
   private final ContextPropagators propagators;
 
   static {
@@ -77,7 +78,7 @@ private ReplaySafeOpenTelemetry(Builder builder) {
                 .setIdGenerator(new ReplaySafeIdGenerator(builder.idGenerator))
                 .build());
     this.meterProvider = new ReplaySafeMeterProvider(builder.meterProviderBuilder.build());
-    this.loggerProvider = builder.loggerProviderBuilder.build();
+    this.loggerProvider = new ReplaySafeLoggerProvider(builder.loggerProviderBuilder.build());
     this.propagators = builder.propagators;
   }
 
@@ -253,6 +254,54 @@ public Meter build() {
     }
   }
 
+  private static final class ReplaySafeLoggerProvider implements LoggerProvider, Closeable {
+    private final SdkLoggerProvider delegate;
+
+    private ReplaySafeLoggerProvider(SdkLoggerProvider delegate) {
+      this.delegate = delegate;
+    }
+
+    @Override
+    public Logger get(@Nonnull String instrumentationScopeName) {
+      return new ReplaySafeLogger(delegate.get(instrumentationScopeName));
+    }
+
+    @Override
+    public LoggerBuilder loggerBuilder(@Nonnull String instrumentationScopeName) {
+      return new ReplaySafeLoggerBuilder(delegate.loggerBuilder(instrumentationScopeName));
+    }
+
+    @Override
+    public void close() {
+      delegate.close();
+    }
+  }
+
+  private static final class ReplaySafeLoggerBuilder implements LoggerBuilder {
+    private final LoggerBuilder delegate;
+
+    private ReplaySafeLoggerBuilder(LoggerBuilder delegate) {
+      this.delegate = delegate;
+    }
+
+    @Override
+    public LoggerBuilder setSchemaUrl(@Nonnull String schemaUrl) {
+      delegate.setSchemaUrl(schemaUrl);
+      return this;
+    }
+
+    @Override
+    public LoggerBuilder setInstrumentationVersion(@Nonnull String instrumentationScopeVersion) {
+      delegate.setInstrumentationVersion(instrumentationScopeVersion);
+      return this;
+    }
+
+    @Override
+    public Logger build() {
+      return new ReplaySafeLogger(delegate.build());
+    }
+  }
+
   private static final class ReplaySafeTracerBuilder implements TracerBuilder {
     private final TracerBuilder delegate;
     private final String instrumentationScopeName;