diff --git a/contrib/temporal-opentelemetry-v2/build.gradle b/contrib/temporal-opentelemetry-v2/build.gradle new file mode 100644 index 0000000000..3fb51d1651 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/build.gradle @@ -0,0 +1,28 @@ +description = '''Temporal Java SDK OpenTelemetry v2 Module''' + +ext { + otelVersion = '1.66.0' +} + +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 "junit:junit:${junitVersion}" + + testRuntimeOnly group: 'ch.qos.logback', name: 'logback-classic', version: "${logbackVersion}" +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryActivityClientInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryActivityClientInterceptor.java new file mode 100644 index 0000000000..3a06c5bde9 --- /dev/null +++ b/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); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryClientInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryClientInterceptor.java new file mode 100644 index 0000000000..fb7af88eab --- /dev/null +++ b/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); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryNexusClientInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryNexusClientInterceptor.java new file mode 100644 index 0000000000..46853d0bc3 --- /dev/null +++ b/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); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryPlugin.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryPlugin.java new file mode 100644 index 0000000000..8bef3a9ae3 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryPlugin.java @@ -0,0 +1,72 @@ +package io.temporal.opentelemetry.v2; + +import io.temporal.common.Experimental; +import io.temporal.common.SimplePlugin; +import io.temporal.opentelemetry.v2.internal.InterceptorTracer; +import io.temporal.opentelemetry.v2.internal.OpenTelemetryContextPropagator; + +/** + * OpenTelemetry v2 plugin for Temporal clients and workers. + * + *

Set it on {@code WorkflowServiceStubsOptions}, {@code WorkflowClientOptions}, or {@code + * WorkerFactoryOptions}. The SDK propagates plugins down that chain. + * + *

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) { + 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())); + } + + 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"); + } + return new OpenTelemetryPlugin(new InterceptorTracer(headerKey, addTemporalSpans)); + } + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryScheduleClientInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryScheduleClientInterceptor.java new file mode 100644 index 0000000000..01ab09bad2 --- /dev/null +++ b/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); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryWorkerInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryWorkerInterceptor.java new file mode 100644 index 0000000000..cf7d414f4a --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/OpenTelemetryWorkerInterceptor.java @@ -0,0 +1,35 @@ +package io.temporal.opentelemetry.v2; + +import io.nexusrpc.handler.OperationContext; +import io.temporal.common.interceptors.ActivityInboundCallsInterceptor; +import io.temporal.common.interceptors.NexusOperationInboundCallsInterceptor; +import io.temporal.common.interceptors.WorkerInterceptor; +import io.temporal.common.interceptors.WorkflowInboundCallsInterceptor; +import io.temporal.opentelemetry.v2.internal.InterceptorTracer; +import io.temporal.opentelemetry.v2.internal.OpenTelemetryActivityInboundCallsInterceptor; +import io.temporal.opentelemetry.v2.internal.OpenTelemetryNexusOperationInboundCallsInterceptor; +import io.temporal.opentelemetry.v2.internal.OpenTelemetryWorkflowInboundCallsInterceptor; + +public class OpenTelemetryWorkerInterceptor implements WorkerInterceptor { + private final InterceptorTracer tracer; + + public OpenTelemetryWorkerInterceptor(InterceptorTracer tracer) { + this.tracer = tracer; + } + + @Override + public WorkflowInboundCallsInterceptor interceptWorkflow(WorkflowInboundCallsInterceptor next) { + return new OpenTelemetryWorkflowInboundCallsInterceptor(tracer, next); + } + + @Override + public ActivityInboundCallsInterceptor interceptActivity(ActivityInboundCallsInterceptor next) { + return new OpenTelemetryActivityInboundCallsInterceptor(tracer, next); + } + + @Override + public NexusOperationInboundCallsInterceptor interceptNexusOperation( + OperationContext context, NexusOperationInboundCallsInterceptor next) { + return new OpenTelemetryNexusOperationInboundCallsInterceptor(tracer, next); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/ReplaySafeOpenTelemetry.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/ReplaySafeOpenTelemetry.java new file mode 100644 index 0000000000..d12e51c1fd --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/ReplaySafeOpenTelemetry.java @@ -0,0 +1,184 @@ +package io.temporal.opentelemetry.v2; + +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.baggage.propagation.W3CBaggagePropagator; +import io.opentelemetry.api.logs.LoggerProvider; +import io.opentelemetry.api.metrics.MeterProvider; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.api.trace.TracerBuilder; +import io.opentelemetry.api.trace.TracerProvider; +import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; +import io.opentelemetry.context.propagation.ContextPropagators; +import io.opentelemetry.context.propagation.TextMapPropagator; +import io.opentelemetry.sdk.logs.SdkLoggerProvider; +import io.opentelemetry.sdk.logs.SdkLoggerProviderBuilder; +import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.SdkTracerProviderBuilder; +import io.temporal.common.Experimental; +import io.temporal.opentelemetry.v2.internal.ReplaySafeIdGenerator; +import io.temporal.opentelemetry.v2.internal.ReplaySafeTracer; +import java.io.Closeable; +import javax.annotation.Nonnull; + +/** + * The {@link OpenTelemetry} to use for OpenTelemetry integration with Temporal. Register it with + * {@code GlobalOpenTelemetry.set}; tracers obtained from it are replay safe inside workflows. + */ +@Experimental +public final class ReplaySafeOpenTelemetry implements OpenTelemetry, Closeable { + private final ReplaySafeTracerProvider tracerProvider; + private final SdkMeterProvider meterProvider; // TODO - Make the meter provider replay safe + // TODO: Make the logger provider replay safe and add logger interceptor methods for Temporal and + // OpenTelemetry loggers. + private final SdkLoggerProvider loggerProvider; + private final ContextPropagators propagators; + + private ReplaySafeOpenTelemetry(Builder builder) { + this.tracerProvider = + new ReplaySafeTracerProvider( + builder.tracerProviderBuilder.setIdGenerator(new ReplaySafeIdGenerator()).build()); + this.meterProvider = builder.meterProviderBuilder.build(); + this.loggerProvider = builder.loggerProviderBuilder.build(); + this.propagators = builder.propagators; + } + + public static Builder newBuilder() { + return new Builder(); + } + + @Override + public TracerProvider getTracerProvider() { + return tracerProvider; + } + + @Override + public MeterProvider getMeterProvider() { + return meterProvider; + } + + @Override + public LoggerProvider getLogsBridge() { + return loggerProvider; + } + + @Override + public ContextPropagators getPropagators() { + return propagators; + } + + static boolean isRegisteredGlobally() { + return GlobalOpenTelemetry.getTracerProvider() instanceof ReplaySafeTracerProvider; + } + + /** Shuts down every provider. Call after the clients and workers using them have stopped. */ + @Override + public void close() { + tracerProvider.close(); + meterProvider.close(); + loggerProvider.close(); + } + + /** Every provider is optional; an unset one is built from the SDK's default builder. */ + public static final class Builder { + private SdkTracerProviderBuilder tracerProviderBuilder = SdkTracerProvider.builder(); + private SdkMeterProviderBuilder meterProviderBuilder = SdkMeterProvider.builder(); + private SdkLoggerProviderBuilder loggerProviderBuilder = SdkLoggerProvider.builder(); + private ContextPropagators propagators = + ContextPropagators.create( + TextMapPropagator.composite( + W3CTraceContextPropagator.getInstance(), W3CBaggagePropagator.getInstance())); + + private Builder() {} + + public Builder setTracerProviderBuilder(SdkTracerProviderBuilder tracerProviderBuilder) { + this.tracerProviderBuilder = tracerProviderBuilder; + return this; + } + + public Builder setMeterProviderBuilder(SdkMeterProviderBuilder meterProviderBuilder) { + this.meterProviderBuilder = meterProviderBuilder; + return this; + } + + public Builder setLoggerProviderBuilder(SdkLoggerProviderBuilder loggerProviderBuilder) { + this.loggerProviderBuilder = loggerProviderBuilder; + return this; + } + + /** + * The propagators returned by {@link ReplaySafeOpenTelemetry#getPropagators()}. Defaults to W3C + * trace context plus baggage, which is what {@link OpenTelemetryPlugin} serializes into + * Temporal headers when it is left to resolve its propagator from the global. + */ + public Builder setPropagators(ContextPropagators propagators) { + this.propagators = propagators; + return this; + } + + public ReplaySafeOpenTelemetry build() { + return new ReplaySafeOpenTelemetry(this); + } + } + + private static final class ReplaySafeTracerProvider implements TracerProvider, Closeable { + private final SdkTracerProvider delegate; + + private ReplaySafeTracerProvider(SdkTracerProvider delegate) { + this.delegate = delegate; + } + + @Override + public Tracer get(@Nonnull String instrumentationScopeName) { + return new ReplaySafeTracer(delegate.get(instrumentationScopeName), instrumentationScopeName); + } + + @Override + public Tracer get( + @Nonnull String instrumentationScopeName, @Nonnull String instrumentationScopeVersion) { + return new ReplaySafeTracer( + delegate.get(instrumentationScopeName, instrumentationScopeVersion), + instrumentationScopeName); + } + + @Override + public TracerBuilder tracerBuilder(@Nonnull String instrumentationScopeName) { + return new ReplaySafeTracerBuilder( + delegate.tracerBuilder(instrumentationScopeName), instrumentationScopeName); + } + + @Override + public void close() { + delegate.close(); + } + } + + private static final class ReplaySafeTracerBuilder implements TracerBuilder { + private final TracerBuilder delegate; + private final String instrumentationScopeName; + + private ReplaySafeTracerBuilder(TracerBuilder delegate, String instrumentationScopeName) { + this.delegate = delegate; + this.instrumentationScopeName = instrumentationScopeName; + } + + @Override + public TracerBuilder setSchemaUrl(@Nonnull String schemaUrl) { + delegate.setSchemaUrl(schemaUrl); + return this; + } + + @Override + public TracerBuilder setInstrumentationVersion(@Nonnull String instrumentationScopeVersion) { + delegate.setInstrumentationVersion(instrumentationScopeVersion); + return this; + } + + @Override + public Tracer build() { + return new ReplaySafeTracer(delegate.build(), instrumentationScopeName); + } + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/InterceptorTracer.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/InterceptorTracer.java new file mode 100644 index 0000000000..171a912700 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/InterceptorTracer.java @@ -0,0 +1,191 @@ +package io.temporal.opentelemetry.v2.internal; + +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import io.temporal.common.interceptors.Header; +import io.temporal.failure.ApplicationErrorCategory; +import io.temporal.failure.ApplicationFailure; +import io.temporal.internal.sync.DestroyWorkflowThreadError; +import java.util.List; +import java.util.Map; + +/** Wraps intercepted Temporal calls in a span and propagates it through headers. */ +public final class InterceptorTracer { + private static final String INSTRUMENTATION_NAME = "temporal-sdk-java"; + + private final Tracer tracer; + private final SpanCodec codec; + private final boolean addTemporalSpans; + + public InterceptorTracer(String headerKey, boolean addTemporalSpans) { + this.tracer = GlobalOpenTelemetry.getTracer(INSTRUMENTATION_NAME); + this.codec = new SpanCodec(headerKey); + this.addTemporalSpans = addTemporalSpans; + } + + /** + * An intercepted call. {@code E} is the checked exception it declares, or {@link + * RuntimeException} when it declares none. + */ + @FunctionalInterface + interface Call { + R call() throws E; + } + + /** An intercepted call with no result. */ + @FunctionalInterface + interface VoidCall { + void call() throws E; + } + + R traceInbound( + String operation, String name, Attributes attributes, Header header, Call call) + throws E { + return traceInbound(operation, name, attributes, codec.read(header), call); + } + + void traceInbound( + String operation, String name, Attributes attributes, Header header, VoidCall call) + throws E { + traceInbound(operation, name, attributes, codec.read(header), asCall(call)); + } + + R traceNexusInbound( + String operation, + String name, + Attributes attributes, + Map nexusHeaders, + Call call) + throws E { + return traceInbound(operation, name, attributes, codec.read(nexusHeaders), call); + } + + private R traceInbound( + String operation, String name, Attributes attributes, Context parent, Call call) + throws E { + try (Scope ignoredParent = parent.makeCurrent()) { + if (!addTemporalSpans) { + return call.call(); + } + + Span span = startSpan(operation, name, attributes, SpanKind.SERVER); + try (Scope ignored = span.makeCurrent()) { + return run(span, call); + } finally { + span.end(); + } + } + } + + R traceOutbound( + String operation, String name, Attributes attributes, Header header, Call call) + throws E { + return traceOutbound(operation, name, attributes, () -> codec.write(header), call); + } + + void injectOutboundHeader(Header header) { + codec.write(header); + } + + void clearOutboundHeader(Header header) { + codec.clear(header); + } + + R traceOutbound( + String operation, String name, Attributes attributes, Call call) throws E { + return traceOutbound(operation, name, attributes, () -> {}, call); + } + + void traceOutbound( + String operation, String name, Attributes attributes, Header header, VoidCall call) + throws E { + traceOutbound(operation, name, attributes, () -> codec.write(header), asCall(call)); + } + + R traceOutbound( + String operation, String name, Attributes attributes, List

headers, Call call) + throws E { + return traceOutbound(operation, name, attributes, () -> headers.forEach(codec::write), call); + } + + R traceNexusOutbound( + String operation, + String name, + Attributes attributes, + Map nexusHeaders, + Call call) + throws E { + return traceOutbound(operation, name, attributes, () -> codec.write(nexusHeaders), call); + } + + private R traceOutbound( + String operation, String name, Attributes attributes, Runnable writeHeader, Call call) + throws E { + if (!addTemporalSpans) { + writeHeader.run(); + return call.call(); + } + + Span span = startSpan(operation, name, attributes, SpanKind.CLIENT); + try (Scope ignored = span.makeCurrent()) { + writeHeader.run(); + return run(span, call); + } finally { + span.end(); + } + } + + private static Call asCall(VoidCall call) { + return () -> { + call.call(); + return null; + }; + } + + /** Records a failure on {@code span} before letting it propagate. */ + private static R run(Span span, Call call) throws E { + try { + return call.call(); + } catch (DestroyWorkflowThreadError unwind) { + throw unwind; + } catch (Throwable failure) { + span.recordException(failure); + if (!isBenign(failure)) { + span.setStatus(StatusCode.ERROR, failure.toString()); + } + throw failure; + } + } + + private Span startSpan(String operation, String name, Attributes attributes, SpanKind kind) { + try (Scope ignored = + Context.current().with(ReplaySafeIdGenerator.INTERCEPTOR_SPAN, true).makeCurrent()) { + return tracer + .spanBuilder(spanName(operation, name)) + .setSpanKind(kind) + .setAllAttributes(attributes) + .startSpan(); + } + } + + static String spanName(String operation, String name) { + if (operation.isEmpty()) { + return name; + } + if (name.isEmpty()) { + return operation; + } + return operation + ":" + name; + } + + private static boolean isBenign(Throwable failure) { + return failure instanceof ApplicationFailure + && ((ApplicationFailure) failure).getCategory() == ApplicationErrorCategory.BENIGN; + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryActivityClientCallsInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryActivityClientCallsInterceptor.java new file mode 100644 index 0000000000..fb901c2f59 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryActivityClientCallsInterceptor.java @@ -0,0 +1,28 @@ +package io.temporal.opentelemetry.v2.internal; + +import static io.temporal.opentelemetry.v2.internal.TagKeys.ACTIVITY_ID; + +import io.opentelemetry.api.common.Attributes; +import io.temporal.common.interceptors.ActivityClientCallsInterceptor; +import io.temporal.common.interceptors.ActivityClientCallsInterceptorBase; + +public class OpenTelemetryActivityClientCallsInterceptor + extends ActivityClientCallsInterceptorBase { + private final InterceptorTracer tracer; + + public OpenTelemetryActivityClientCallsInterceptor( + InterceptorTracer tracer, ActivityClientCallsInterceptor next) { + super(next); + this.tracer = tracer; + } + + @Override + public StartActivityOutput startActivity(StartActivityInput input) { + return tracer.traceOutbound( + "StartActivity", + input.getActivityType(), + Attributes.of(ACTIVITY_ID, input.getOptions().getId()), + input.getHeader(), + () -> super.startActivity(input)); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryActivityInboundCallsInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryActivityInboundCallsInterceptor.java new file mode 100644 index 0000000000..6da63b04d6 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryActivityInboundCallsInterceptor.java @@ -0,0 +1,43 @@ +package io.temporal.opentelemetry.v2.internal; + +import static io.temporal.opentelemetry.v2.internal.TagKeys.*; + +import io.opentelemetry.api.common.Attributes; +import io.temporal.activity.ActivityExecutionContext; +import io.temporal.activity.ActivityInfo; +import io.temporal.common.interceptors.ActivityInboundCallsInterceptor; +import io.temporal.common.interceptors.ActivityInboundCallsInterceptorBase; + +public class OpenTelemetryActivityInboundCallsInterceptor + extends ActivityInboundCallsInterceptorBase { + private final InterceptorTracer tracer; + // Activity code reaches its context through Activity.getExecutionContext(), but interceptors + // only see it in init. + private ActivityExecutionContext context; + + public OpenTelemetryActivityInboundCallsInterceptor( + InterceptorTracer tracer, ActivityInboundCallsInterceptor next) { + super(next); + this.tracer = tracer; + } + + @Override + public void init(ActivityExecutionContext context) { + this.context = context; + super.init(context); + } + + @Override + public ActivityOutput execute(ActivityInput input) { + ActivityInfo info = context.getInfo(); + return tracer.traceInbound( + "RunActivity", + info.getActivityType(), + Attributes.of( + WORKFLOW_ID, info.getWorkflowId(), + RUN_ID, info.getWorkflowRunId(), + ACTIVITY_ID, info.getActivityId()), + input.getHeader(), + () -> super.execute(input)); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryContextPropagator.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryContextPropagator.java new file mode 100644 index 0000000000..a01b582b54 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryContextPropagator.java @@ -0,0 +1,35 @@ +package io.temporal.opentelemetry.v2.internal; + +import io.opentelemetry.context.Context; +import io.temporal.api.common.v1.Payload; +import io.temporal.common.context.ContextPropagator; +import java.util.Collections; +import java.util.Map; + +/** Copies the current OpenTelemetry context between Temporal workflow threads. */ +public final class OpenTelemetryContextPropagator implements ContextPropagator { + @Override + public String getName() { + return "io.temporal.opentelemetry.v2.workflow-context"; + } + + @Override + public Map serializeContext(Object context) { + return Collections.emptyMap(); + } + + @Override + public Object deserializeContext(Map header) { + return Context.root(); + } + + @Override + public Object getCurrentContext() { + return Context.current(); + } + + @Override + public void setCurrentContext(Object context) { + TemporalContextStorage.setWorkflowContext((Context) context); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryNexusClientCallsInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryNexusClientCallsInterceptor.java new file mode 100644 index 0000000000..5cd729a725 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryNexusClientCallsInterceptor.java @@ -0,0 +1,31 @@ +package io.temporal.opentelemetry.v2.internal; + +import static io.temporal.opentelemetry.v2.internal.TagKeys.*; + +import io.opentelemetry.api.common.Attributes; +import io.temporal.common.interceptors.NexusClientCallsInterceptor; +import io.temporal.common.interceptors.NexusClientCallsInterceptorBase; + +public class OpenTelemetryNexusClientCallsInterceptor extends NexusClientCallsInterceptorBase { + private final InterceptorTracer tracer; + + public OpenTelemetryNexusClientCallsInterceptor( + InterceptorTracer tracer, NexusClientCallsInterceptor next) { + super(next); + this.tracer = tracer; + } + + @Override + public StartNexusOperationExecutionOutput startNexusOperationExecution( + StartNexusOperationExecutionInput input) { + return tracer.traceNexusOutbound( + "StartNexusOperation", + input.getService() + "/" + input.getOperation(), + Attributes.of( + NEXUS_ENDPOINT, input.getEndpoint(), + NEXUS_SERVICE, input.getService(), + NEXUS_OPERATION, input.getOperation()), + input.getHeaders(), + () -> super.startNexusOperationExecution(input)); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryNexusOperationInboundCallsInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryNexusOperationInboundCallsInterceptor.java new file mode 100644 index 0000000000..c06203c3cc --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryNexusOperationInboundCallsInterceptor.java @@ -0,0 +1,58 @@ +package io.temporal.opentelemetry.v2.internal; + +import static io.temporal.opentelemetry.v2.internal.TagKeys.*; + +import io.nexusrpc.OperationException; +import io.nexusrpc.handler.OperationContext; +import io.opentelemetry.api.common.Attributes; +import io.temporal.common.interceptors.NexusOperationInboundCallsInterceptor; +import io.temporal.common.interceptors.NexusOperationInboundCallsInterceptorBase; +import io.temporal.common.interceptors.NexusOperationOutboundCallsInterceptor; + +public class OpenTelemetryNexusOperationInboundCallsInterceptor + extends NexusOperationInboundCallsInterceptorBase { + + private final InterceptorTracer tracer; + + public OpenTelemetryNexusOperationInboundCallsInterceptor( + InterceptorTracer tracer, NexusOperationInboundCallsInterceptor next) { + super(next); + this.tracer = tracer; + } + + @Override + public void init(NexusOperationOutboundCallsInterceptor outboundCalls) { + super.init(new OpenTelemetryNexusOperationOutboundCallsInterceptor(outboundCalls)); + } + + @Override + public StartOperationOutput startOperation(StartOperationInput input) throws OperationException { + OperationContext context = input.getOperationContext(); + return tracer.traceNexusInbound( + "RunStartNexusOperationHandler", + spanName(context), + nexusTags(context), + context.getHeaders(), + () -> super.startOperation(input)); + } + + @Override + public CancelOperationOutput cancelOperation(CancelOperationInput input) { + OperationContext context = input.getOperationContext(); + return tracer.traceNexusInbound( + "RunCancelNexusOperationHandler", + spanName(context), + nexusTags(context), + context.getHeaders(), + () -> super.cancelOperation(input)); + } + + private static String spanName(OperationContext context) { + return context.getService() + "/" + context.getOperation(); + } + + private static Attributes nexusTags(OperationContext context) { + return Attributes.of( + NEXUS_SERVICE, context.getService(), NEXUS_OPERATION, context.getOperation()); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryNexusOperationOutboundCallsInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryNexusOperationOutboundCallsInterceptor.java new file mode 100644 index 0000000000..c8f1a3e5c8 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryNexusOperationOutboundCallsInterceptor.java @@ -0,0 +1,12 @@ +package io.temporal.opentelemetry.v2.internal; + +import io.temporal.common.interceptors.NexusOperationOutboundCallsInterceptor; +import io.temporal.common.interceptors.NexusOperationOutboundCallsInterceptorBase; + +public class OpenTelemetryNexusOperationOutboundCallsInterceptor + extends NexusOperationOutboundCallsInterceptorBase { + public OpenTelemetryNexusOperationOutboundCallsInterceptor( + NexusOperationOutboundCallsInterceptor next) { + super(next); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryScheduleClientCallsInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryScheduleClientCallsInterceptor.java new file mode 100644 index 0000000000..7cc0d9e98e --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryScheduleClientCallsInterceptor.java @@ -0,0 +1,61 @@ +package io.temporal.opentelemetry.v2.internal; + +import io.opentelemetry.api.common.Attributes; +import io.temporal.client.schedules.ScheduleActionStartWorkflow; +import io.temporal.client.schedules.ScheduleUpdate; +import io.temporal.client.schedules.ScheduleUpdateInput; +import io.temporal.common.interceptors.Header; +import io.temporal.common.interceptors.ScheduleClientCallsInterceptor; +import io.temporal.common.interceptors.ScheduleClientCallsInterceptorBase; + +public class OpenTelemetryScheduleClientCallsInterceptor + extends ScheduleClientCallsInterceptorBase { + private final InterceptorTracer tracer; + + public OpenTelemetryScheduleClientCallsInterceptor( + InterceptorTracer tracer, ScheduleClientCallsInterceptor next) { + super(next); + this.tracer = tracer; + } + + @Override + public void createSchedule(CreateScheduleInput input) { + if (!(input.getSchedule().getAction() instanceof ScheduleActionStartWorkflow)) { + super.createSchedule(input); + return; + } + + Header header = ((ScheduleActionStartWorkflow) input.getSchedule().getAction()).getHeader(); + tracer.clearOutboundHeader(header); + tracer.traceOutbound( + "CreateSchedule", + input.getId(), + Attributes.empty(), + header, + () -> super.createSchedule(input)); + } + + @Override + public void updateSchedule(UpdateScheduleInput input) { + tracer.traceOutbound( + "UpdateSchedule", + input.getDescription().getId(), + Attributes.empty(), + () -> { + super.updateSchedule( + new UpdateScheduleInput( + input.getDescription(), updateInput -> applyUpdate(input, updateInput))); + return null; + }); + } + + private ScheduleUpdate applyUpdate(UpdateScheduleInput input, ScheduleUpdateInput updateInput) { + ScheduleUpdate update = input.getUpdater().apply(updateInput); + if (update != null && update.getSchedule().getAction() instanceof ScheduleActionStartWorkflow) { + Header header = ((ScheduleActionStartWorkflow) update.getSchedule().getAction()).getHeader(); + tracer.clearOutboundHeader(header); + tracer.injectOutboundHeader(header); + } + return update; + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryWorkflowClientCallsInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryWorkflowClientCallsInterceptor.java new file mode 100644 index 0000000000..d76609a452 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryWorkflowClientCallsInterceptor.java @@ -0,0 +1,121 @@ +package io.temporal.opentelemetry.v2.internal; + +import static io.temporal.opentelemetry.v2.internal.TagKeys.*; + +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.temporal.client.WorkflowUpdateHandle; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptorBase; +import java.util.Arrays; + +public class OpenTelemetryWorkflowClientCallsInterceptor + extends WorkflowClientCallsInterceptorBase { + private final InterceptorTracer tracer; + + public OpenTelemetryWorkflowClientCallsInterceptor( + InterceptorTracer tracer, WorkflowClientCallsInterceptor next) { + super(next); + this.tracer = tracer; + } + + @Override + public WorkflowStartOutput start(WorkflowStartInput input) { + return tracer.traceOutbound( + "StartWorkflow", + input.getWorkflowType(), + Attributes.of(WORKFLOW_ID, input.getWorkflowId()), + input.getHeader(), + () -> super.start(input)); + } + + @Override + public WorkflowSignalOutput signal(WorkflowSignalInput input) { + return tracer.traceOutbound( + "SignalWorkflow", + input.getSignalName(), + workflowExecutionTags(input.getWorkflowExecution()), + input.getHeader(), + () -> super.signal(input)); + } + + @Override + public WorkflowSignalWithStartOutput signalWithStart(WorkflowSignalWithStartInput input) { + WorkflowStartInput start = input.getWorkflowStartInput(); + return tracer.traceOutbound( + "SignalWithStartWorkflow", + start.getWorkflowType(), + Attributes.of(WORKFLOW_ID, start.getWorkflowId()), + start.getHeader(), + () -> super.signalWithStart(input)); + } + + @Override + public QueryOutput query(QueryInput input) { + return tracer.traceOutbound( + "QueryWorkflow", + input.getQueryType(), + workflowExecutionTags(input.getWorkflowExecution()), + input.getHeader(), + () -> super.query(input)); + } + + @Override + public WorkflowUpdateHandle startUpdate(StartUpdateInput input) { + AttributesBuilder attributes = workflowExecutionTags(input.getWorkflowExecution()).toBuilder(); + attributes.put(UPDATE_ID, input.getUpdateId()); + return tracer.traceOutbound( + "StartWorkflowUpdate", + input.getUpdateName(), + attributes.build(), + input.getHeader(), + () -> super.startUpdate(input)); + } + + @Override + public CancelOutput cancel(CancelInput input) { + return tracer.traceOutbound( + "CancelWorkflow", + "", + workflowExecutionTags(input.getWorkflowExecution()), + () -> super.cancel(input)); + } + + @Override + public TerminateOutput terminate(TerminateInput input) { + AttributesBuilder attributes = workflowExecutionTags(input.getWorkflowExecution()).toBuilder(); + if (input.getReason() != null) { + attributes.put(TERMINATE_REASON, input.getReason()); + } + return tracer.traceOutbound( + "TerminateWorkflow", "", attributes.build(), () -> super.terminate(input)); + } + + @Override + public DescribeWorkflowOutput describe(DescribeWorkflowInput input) { + return tracer.traceOutbound( + "DescribeWorkflow", + "", + workflowExecutionTags(input.getWorkflowExecution()), + () -> super.describe(input)); + } + + @Override + public WorkflowUpdateWithStartOutput updateWithStart( + WorkflowUpdateWithStartInput input) { + WorkflowStartInput start = input.getWorkflowStartInput(); + StartUpdateInput update = input.getStartUpdateInput(); + // The start header reaches the workflow and the update header reaches the update handler. + return tracer.traceOutbound( + "UpdateWithStartWorkflow", + update.getUpdateName(), + Attributes.of(WORKFLOW_ID, start.getWorkflowId(), UPDATE_ID, update.getUpdateId()), + Arrays.asList(start.getHeader(), update.getHeader()), + () -> super.updateWithStart(input)); + } + + private static Attributes workflowExecutionTags( + io.temporal.api.common.v1.WorkflowExecution execution) { + return Attributes.of(WORKFLOW_ID, execution.getWorkflowId(), RUN_ID, execution.getRunId()); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryWorkflowInboundCallsInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryWorkflowInboundCallsInterceptor.java new file mode 100644 index 0000000000..c0d033bde7 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryWorkflowInboundCallsInterceptor.java @@ -0,0 +1,89 @@ +package io.temporal.opentelemetry.v2.internal; + +import static io.temporal.opentelemetry.v2.internal.TagKeys.*; + +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.temporal.common.interceptors.WorkflowInboundCallsInterceptor; +import io.temporal.common.interceptors.WorkflowInboundCallsInterceptorBase; +import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInfo; + +public class OpenTelemetryWorkflowInboundCallsInterceptor + extends WorkflowInboundCallsInterceptorBase { + + private final InterceptorTracer tracer; + + public OpenTelemetryWorkflowInboundCallsInterceptor( + InterceptorTracer tracer, WorkflowInboundCallsInterceptor next) { + super(next); + this.tracer = tracer; + } + + @Override + public void init(WorkflowOutboundCallsInterceptor outboundCalls) { + super.init(new OpenTelemetryWorkflowOutboundCallsInterceptor(tracer, outboundCalls)); + } + + @Override + public WorkflowOutput execute(WorkflowInput input) { + return tracer.traceInbound( + "RunWorkflow", + Workflow.getInfo().getWorkflowType(), + workflowTags(), + input.getHeader(), + () -> super.execute(input)); + } + + @Override + public void handleSignal(SignalInput input) { + tracer.traceInbound( + "HandleSignal", + input.getSignalName(), + workflowTags(), + input.getHeader(), + () -> super.handleSignal(input)); + } + + @Override + public QueryOutput handleQuery(QueryInput input) { + return tracer.traceInbound( + "HandleQuery", + input.getQueryName(), + workflowTags(), + input.getHeader(), + () -> super.handleQuery(input)); + } + + @Override + public void validateUpdate(UpdateInput input) { + tracer.traceInbound( + "ValidateUpdate", + input.getUpdateName(), + updateTags(), + input.getHeader(), + () -> super.validateUpdate(input)); + } + + @Override + public UpdateOutput executeUpdate(UpdateInput input) { + return tracer.traceInbound( + "HandleUpdate", + input.getUpdateName(), + updateTags(), + input.getHeader(), + () -> super.executeUpdate(input)); + } + + private static Attributes workflowTags() { + WorkflowInfo info = Workflow.getInfo(); + return Attributes.of(WORKFLOW_ID, info.getWorkflowId(), RUN_ID, info.getRunId()); + } + + private static Attributes updateTags() { + AttributesBuilder tags = workflowTags().toBuilder(); + Workflow.getCurrentUpdateInfo().ifPresent(update -> tags.put(UPDATE_ID, update.getUpdateId())); + return tags.build(); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryWorkflowOutboundCallsInterceptor.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryWorkflowOutboundCallsInterceptor.java new file mode 100644 index 0000000000..e6b852670e --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/OpenTelemetryWorkflowOutboundCallsInterceptor.java @@ -0,0 +1,120 @@ +package io.temporal.opentelemetry.v2.internal; + +import static io.temporal.opentelemetry.v2.internal.TagKeys.*; + +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor; +import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptorBase; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInfo; + +public class OpenTelemetryWorkflowOutboundCallsInterceptor + extends WorkflowOutboundCallsInterceptorBase { + + private final InterceptorTracer tracer; + + public OpenTelemetryWorkflowOutboundCallsInterceptor( + InterceptorTracer tracer, WorkflowOutboundCallsInterceptor next) { + super(next); + this.tracer = tracer; + } + + @Override + public ActivityOutput executeActivity(ActivityInput input) { + return tracer.traceOutbound( + "StartActivity", + input.getActivityName(), + workflowTags(), + input.getHeader(), + () -> super.executeActivity(input)); + } + + @Override + public LocalActivityOutput executeLocalActivity(LocalActivityInput input) { + return tracer.traceOutbound( + "StartActivity", + input.getActivityName(), + workflowTags(), + input.getHeader(), + () -> super.executeLocalActivity(input)); + } + + @Override + public ChildWorkflowOutput executeChildWorkflow(ChildWorkflowInput input) { + return tracer.traceOutbound( + "StartChildWorkflow", + input.getWorkflowType(), + childWorkflowTags(input), + input.getHeader(), + () -> super.executeChildWorkflow(input)); + } + + @Override + public ExecuteNexusOperationOutput executeNexusOperation( + ExecuteNexusOperationInput input) { + return tracer.traceNexusOutbound( + "StartNexusOperation", + input.getService() + "/" + input.getOperation(), + nexusTags(input), + input.getHeaders(), + () -> super.executeNexusOperation(input)); + } + + @Override + public SignalExternalOutput signalExternalWorkflow(SignalExternalInput input) { + return tracer.traceOutbound( + "SignalExternalWorkflow", + input.getSignalName(), + workflowExecutionTags(input.getExecution()), + input.getHeader(), + () -> super.signalExternalWorkflow(input)); + } + + @Override + public CancelWorkflowOutput cancelWorkflow(CancelWorkflowInput input) { + return tracer.traceOutbound( + "CancelWorkflow", + "", + workflowExecutionTags(input.getExecution()), + () -> super.cancelWorkflow(input)); + } + + @Override + public void continueAsNew(ContinueAsNewInput input) { + String workflowType = input.getWorkflowType(); + if (workflowType == null) { + workflowType = Workflow.getInfo().getWorkflowType(); + } + tracer.traceOutbound( + "ContinueAsNew", + workflowType, + workflowTags(), + input.getHeader(), + () -> super.continueAsNew(input)); + } + + private static Attributes workflowTags() { + WorkflowInfo info = Workflow.getInfo(); + return Attributes.of(WORKFLOW_ID, info.getWorkflowId(), RUN_ID, info.getRunId()); + } + + private static Attributes workflowExecutionTags(WorkflowExecution execution) { + return Attributes.of(WORKFLOW_ID, execution.getWorkflowId(), RUN_ID, execution.getRunId()); + } + + private static Attributes childWorkflowTags(ChildWorkflowInput input) { + return Attributes.of(WORKFLOW_ID, input.getWorkflowId()); + } + + private static Attributes nexusTags(ExecuteNexusOperationInput input) { + AttributesBuilder tags = + workflowTags().toBuilder() + .put(NEXUS_SERVICE, input.getService()) + .put(NEXUS_OPERATION, input.getOperation()) + .put(NEXUS_ENDPOINT, input.getEndpoint()); + + return tags.build(); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/ReplaySafeIdGenerator.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/ReplaySafeIdGenerator.java new file mode 100644 index 0000000000..803947f0f5 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/ReplaySafeIdGenerator.java @@ -0,0 +1,84 @@ +package io.temporal.opentelemetry.v2.internal; + +import io.opentelemetry.api.trace.SpanId; +import io.opentelemetry.api.trace.TraceId; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.ContextKey; +import io.opentelemetry.sdk.trace.IdGenerator; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.unsafe.WorkflowUnsafe; +import java.util.Random; +import javax.annotation.Nullable; + +/** + * Generates span and trace IDs that are replay safe. + * + *

Mirrors OpenTelemetry's RandomIdGenerator, + * replacing its platform random source with a workflow random stream. + * + *

Note: {@link Random} has 48 bits of state, so IDs can collide across workflows once an + * installation has generated on the order of 2^24 trace IDs. The JDK has no deterministic random + * source with a wider seed that can also be reseeded. A random source with more state may be + * considered in the future. + */ +public final class ReplaySafeIdGenerator implements IdGenerator { + static final ContextKey INTERCEPTOR_SPAN = ContextKey.named("temporal-interceptor-span"); + private static final String INTERCEPTOR_STREAM = "io.temporal.opentelemetry.v2/interceptor"; + private static final String APPLICATION_STREAM = "io.temporal.opentelemetry.v2/application"; + + private static final long INVALID_ID = 0; + + @Override + public String generateSpanId() { + Random stream = getStream(); + if (stream == null) { + return IdGenerator.random().generateSpanId(); + } + + long id; + do { + id = stream.nextLong(); + } while (id == INVALID_ID); + return SpanId.fromLong(id); + } + + @Override + public String generateTraceId() { + Random stream = getStream(); + if (stream == null) { + return IdGenerator.random().generateTraceId(); + } + + long idHi = stream.nextLong(); + long idLo; + do { + idLo = stream.nextLong(); + } while (idLo == INVALID_ID); + return TraceId.fromLongs(idHi, idLo); + } + + /** + * Interceptor spans and application spans draw from separate streams so their IDs never collide. + * Null means the regular PRNG can be used. + */ + @Nullable + private static Random getStream() { + if (!WorkflowUnsafe.isWorkflowThread() || !WorkflowUnsafe.isSubjectToReplay()) { + return null; + } + + Context context = Context.current(); + if (context.get(INTERCEPTOR_SPAN) != null) { + return Workflow.getRandomStream(INTERCEPTOR_STREAM); + } + + String tracerName = context.get(ReplaySafeTracer.TRACER_NAME); + if (tracerName == null) { + throw new IllegalStateException( + "Workflow span started without a replay safe tracer. Ensure tracers used in workflows " + + "are obtained from ReplaySafeOpenTelemetry"); + } + return Workflow.getRandomStream(APPLICATION_STREAM + "/" + tracerName); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/ReplaySafeSpan.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/ReplaySafeSpan.java new file mode 100644 index 0000000000..08b41b5642 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/ReplaySafeSpan.java @@ -0,0 +1,84 @@ +package io.temporal.opentelemetry.v2.internal; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.StatusCode; +import io.temporal.workflow.unsafe.WorkflowUnsafe; +import java.util.concurrent.TimeUnit; + +/** Wraps a span so that replayed code does not end it, which would export a duplicate. */ +public final class ReplaySafeSpan implements Span { + private final Span delegate; + + public ReplaySafeSpan(Span delegate) { + this.delegate = delegate; + } + + @Override + public void end() { + if (WorkflowUnsafe.isWorkflowThread() + && WorkflowUnsafe.isSubjectToReplay() + && WorkflowUnsafe.isReplaying()) { + return; + } + delegate.end(); + } + + @Override + public void end(long timestamp, TimeUnit unit) { + if (WorkflowUnsafe.isWorkflowThread() + && WorkflowUnsafe.isSubjectToReplay() + && WorkflowUnsafe.isReplaying()) { + return; + } + delegate.end(timestamp, unit); + } + + @Override + public Span setAttribute(AttributeKey key, T value) { + delegate.setAttribute(key, value); + return this; + } + + @Override + public Span addEvent(String name, Attributes attributes) { + delegate.addEvent(name, attributes); + return this; + } + + @Override + public Span addEvent(String name, Attributes attributes, long timestamp, TimeUnit unit) { + delegate.addEvent(name, attributes, timestamp, unit); + return this; + } + + @Override + public Span setStatus(StatusCode statusCode, String description) { + delegate.setStatus(statusCode, description); + return this; + } + + @Override + public Span recordException(Throwable exception, Attributes additionalAttributes) { + delegate.recordException(exception, additionalAttributes); + return this; + } + + @Override + public Span updateName(String name) { + delegate.updateName(name); + return this; + } + + @Override + public SpanContext getSpanContext() { + return delegate.getSpanContext(); + } + + @Override + public boolean isRecording() { + return delegate.isRecording(); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/ReplaySafeTracer.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/ReplaySafeTracer.java new file mode 100644 index 0000000000..daa68b8319 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/ReplaySafeTracer.java @@ -0,0 +1,130 @@ +package io.temporal.opentelemetry.v2.internal; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanBuilder; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.ContextKey; +import io.opentelemetry.context.Scope; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.unsafe.WorkflowUnsafe; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nonnull; + +/** + * Wraps a tracer so the spans it starts inside a workflow are replay safe. The tracer name is + * published on the current {@link Context} while the span starts, which is where {@link + * ReplaySafeIdGenerator} reads it. + */ +public final class ReplaySafeTracer implements Tracer { + /** The instrumentation name of the tracer starting the current span. */ + @Nonnull static final ContextKey TRACER_NAME = ContextKey.named("temporal-tracer-name"); + + private final Tracer delegate; + private final String name; + + public ReplaySafeTracer(Tracer delegate, String name) { + this.delegate = delegate; + this.name = name; + } + + @Override + public SpanBuilder spanBuilder(String spanName) { + return new NamedStreamSpanBuilder(delegate.spanBuilder(spanName), name); + } + + private static final class NamedStreamSpanBuilder implements SpanBuilder { + private final SpanBuilder delegate; + private final String tracerName; + private boolean startTimestampSet; + + NamedStreamSpanBuilder(SpanBuilder delegate, String tracerName) { + this.delegate = delegate; + this.tracerName = tracerName; + } + + @Override + public Span startSpan() { + if (WorkflowUnsafe.isWorkflowThread() + && WorkflowUnsafe.isSubjectToReplay() + && WorkflowUnsafe.isReplaying() + && !startTimestampSet) { + delegate.setStartTimestamp(Workflow.currentTimeMillis(), TimeUnit.MILLISECONDS); + } + try (Scope ignored = Context.current().with(TRACER_NAME, tracerName).makeCurrent()) { + return new ReplaySafeSpan(delegate.startSpan()); + } + } + + @Override + public SpanBuilder setParent(Context context) { + delegate.setParent(context); + return this; + } + + @Override + public SpanBuilder setNoParent() { + delegate.setNoParent(); + return this; + } + + @Override + public SpanBuilder addLink(SpanContext spanContext) { + delegate.addLink(spanContext); + return this; + } + + @Override + public SpanBuilder addLink(SpanContext spanContext, Attributes attributes) { + delegate.addLink(spanContext, attributes); + return this; + } + + @Override + public SpanBuilder setAttribute(String key, String value) { + delegate.setAttribute(key, value); + return this; + } + + @Override + public SpanBuilder setAttribute(String key, long value) { + delegate.setAttribute(key, value); + return this; + } + + @Override + public SpanBuilder setAttribute(String key, double value) { + delegate.setAttribute(key, value); + return this; + } + + @Override + public SpanBuilder setAttribute(String key, boolean value) { + delegate.setAttribute(key, value); + return this; + } + + @Override + public SpanBuilder setAttribute(AttributeKey key, T value) { + delegate.setAttribute(key, value); + return this; + } + + @Override + public SpanBuilder setSpanKind(SpanKind spanKind) { + delegate.setSpanKind(spanKind); + return this; + } + + @Override + public SpanBuilder setStartTimestamp(long startTimestamp, TimeUnit unit) { + startTimestampSet = true; + delegate.setStartTimestamp(startTimestamp, unit); + return this; + } + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/SpanCodec.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/SpanCodec.java new file mode 100644 index 0000000000..a7c38fe927 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/SpanCodec.java @@ -0,0 +1,121 @@ +package io.temporal.opentelemetry.v2.internal; + +import com.google.common.reflect.TypeToken; +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.propagation.TextMapGetter; +import io.opentelemetry.context.propagation.TextMapPropagator; +import io.opentelemetry.context.propagation.TextMapSetter; +import io.temporal.api.common.v1.Payload; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.StdConverterBackwardsCompatAdapter; +import io.temporal.common.interceptors.Header; +import java.lang.reflect.Type; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.TreeMap; +import javax.annotation.Nullable; + +/** + * Serializes the current span and baggage into Temporal and Nexus headers and reads them back. + * + *

Temporal headers carry one {@link Properties} payload under the configured key. Nexus headers + * are flat and use HTTP header semantics, so they are read case-insensitively. + */ +final class SpanCodec { + private static final TextMapSetter> MAP_SETTER = Map::put; + + private static final TextMapGetter> MAP_GETTER = + new TextMapGetter>() { + @Override + public Iterable keys(Map carrier) { + return carrier.keySet(); + } + + @Override + @Nullable + public String get(Map carrier, String key) { + return carrier.get(key); + } + }; + private static final TextMapSetter PROPERTIES_SETTER = Properties::setProperty; + private static final TextMapGetter PROPERTIES_GETTER = + new TextMapGetter() { + @Override + public Iterable keys(Properties carrier) { + return carrier.stringPropertyNames(); + } + + @Override + @Nullable + public String get(Properties carrier, String key) { + return carrier.getProperty(key); + } + }; + private static final Type HASH_MAP_STRING_STRING_TYPE = + new TypeToken>() {}.getType(); + + private final TextMapPropagator propagator; + private final String headerKey; + + SpanCodec(String headerKey) { + this.propagator = GlobalOpenTelemetry.getPropagators().getTextMapPropagator(); + this.headerKey = headerKey; + } + + /** The current context extended with the span and baggage carried by {@code header}. */ + Context read(Header header) { + Payload payload = header.getValues().get(headerKey); + if (payload == null) { + return Context.current(); + } + return extract(decode(payload), PROPERTIES_GETTER); + } + + /** The current context extended with the span and baggage carried by {@code nexusHeaders}. */ + Context read(Map nexusHeaders) { + // Nexus headers use HTTP header semantics, so the propagator must see them case-insensitively. + // See https://opentelemetry.io/docs/specs/otel/context/api-propagators/#get. + Map carrier = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + carrier.putAll(nexusHeaders); + return extract(carrier, MAP_GETTER); + } + + /** Writes the current span and baggage into {@code header}; leaves it untouched if empty. */ + void write(Header header) { + Properties carrier = new Properties(); + propagator.inject(Context.current(), carrier, PROPERTIES_SETTER); + if (!carrier.isEmpty()) { + header.getValues().put(headerKey, encode(carrier)); + } + } + + /** Removes the tracing header from {@code header}. */ + void clear(Header header) { + header.getValues().remove(headerKey); + } + + /** Writes the current span and baggage into {@code nexusHeaders}. */ + void write(Map nexusHeaders) { + propagator.inject(Context.current(), nexusHeaders, MAP_SETTER); + } + + private Context extract(C carrier, TextMapGetter getter) { + Context current = Context.current(); + return propagator.extract(current, carrier, getter); + } + + private static Payload encode(Properties carrier) { + return DefaultDataConverter.STANDARD_INSTANCE.toPayload(carrier).get(); + } + + static Properties decode(Payload payload) { + Map decoded = + StdConverterBackwardsCompatAdapter.fromPayload( + payload, HashMap.class, HASH_MAP_STRING_STRING_TYPE); + Properties properties = new Properties(); + properties.putAll(decoded); + return properties; + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/TagKeys.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/TagKeys.java new file mode 100644 index 0000000000..a1ce777011 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/TagKeys.java @@ -0,0 +1,19 @@ +package io.temporal.opentelemetry.v2.internal; + +import io.opentelemetry.api.common.AttributeKey; + +final class TagKeys { + static final AttributeKey WORKFLOW_ID = AttributeKey.stringKey("temporalWorkflowID"); + static final AttributeKey RUN_ID = AttributeKey.stringKey("temporalRunID"); + static final AttributeKey ACTIVITY_ID = AttributeKey.stringKey("temporalActivityID"); + static final AttributeKey UPDATE_ID = AttributeKey.stringKey("temporalUpdateID"); + static final AttributeKey TERMINATE_REASON = + AttributeKey.stringKey("temporalTerminateReason"); + static final AttributeKey NEXUS_SERVICE = AttributeKey.stringKey("temporalNexusService"); + static final AttributeKey NEXUS_OPERATION = + AttributeKey.stringKey("temporalNexusOperation"); + static final AttributeKey NEXUS_ENDPOINT = + AttributeKey.stringKey("temporalNexusEndpoint"); + + private TagKeys() {} +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/TemporalContextStorage.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/TemporalContextStorage.java new file mode 100644 index 0000000000..881a65f0b5 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/TemporalContextStorage.java @@ -0,0 +1,52 @@ +package io.temporal.opentelemetry.v2.internal; + +import io.opentelemetry.context.Context; +import io.opentelemetry.context.ContextStorage; +import io.opentelemetry.context.Scope; +import io.temporal.workflow.WorkflowThreadLocal; +import io.temporal.workflow.unsafe.WorkflowUnsafe; + +/** Stores the current OpenTelemetry context in Temporal workflow threads. */ +public final class TemporalContextStorage implements ContextStorage { + private static final WorkflowThreadLocal WORKFLOW_CONTEXT = new WorkflowThreadLocal<>(); + + private final ContextStorage delegate; + + TemporalContextStorage(ContextStorage delegate) { + this.delegate = delegate; + } + + static void setWorkflowContext(Context context) { + if (WorkflowUnsafe.isWorkflowThread()) { + WORKFLOW_CONTEXT.set(context); + } + } + + @Override + public Scope attach(Context context) { + if (!WorkflowUnsafe.isWorkflowThread()) { + return delegate.attach(context); + } + + Context previous = WORKFLOW_CONTEXT.get(); + if (context == previous) { + return () -> {}; + } + WORKFLOW_CONTEXT.set(context); + return () -> { + if (WORKFLOW_CONTEXT.get() == context) { + WORKFLOW_CONTEXT.set(previous); + } + }; + } + + @Override + public Context current() { + return WorkflowUnsafe.isWorkflowThread() ? WORKFLOW_CONTEXT.get() : delegate.current(); + } + + @Override + public Context root() { + return delegate.root(); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/TemporalContextStorageProvider.java b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/TemporalContextStorageProvider.java new file mode 100644 index 0000000000..0c57f929a6 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/java/io/temporal/opentelemetry/v2/internal/TemporalContextStorageProvider.java @@ -0,0 +1,12 @@ +package io.temporal.opentelemetry.v2.internal; + +import io.opentelemetry.context.ContextStorage; +import io.opentelemetry.context.ContextStorageProvider; + +/** Provides the Temporal-aware OpenTelemetry context storage before OpenTelemetry initializes. */ +public final class TemporalContextStorageProvider implements ContextStorageProvider { + @Override + public ContextStorage get() { + return new TemporalContextStorage(ContextStorage.defaultStorage()); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/main/resources/META-INF/services/io.opentelemetry.context.ContextStorageProvider b/contrib/temporal-opentelemetry-v2/src/main/resources/META-INF/services/io.opentelemetry.context.ContextStorageProvider new file mode 100644 index 0000000000..08b96b97c1 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/main/resources/META-INF/services/io.opentelemetry.context.ContextStorageProvider @@ -0,0 +1 @@ +io.temporal.opentelemetry.v2.internal.TemporalContextStorageProvider diff --git a/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/InterceptorTest.java b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/InterceptorTest.java new file mode 100644 index 0000000000..31012d297b --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/InterceptorTest.java @@ -0,0 +1,371 @@ +package io.temporal.opentelemetry.v2; + +import static io.temporal.opentelemetry.v2.TestWorkflows.TASK_TOKENS; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.enums.v1.WorkflowIdConflictPolicy; +import io.temporal.client.UpdateOptions; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowFailedException; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.client.WorkflowTargetOptions; +import io.temporal.client.WorkflowUpdateStage; +import io.temporal.opentelemetry.v2.TestWorkflows.AsyncCompletionWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.AsyncCompletionWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.AsyncLambdaWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.AsyncLambdaWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.BenignErrorWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.BenignErrorWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.ChildWorkflowWithSignal; +import io.temporal.opentelemetry.v2.TestWorkflows.ChildWorkflowWithSignalImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.ErrorWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.ErrorWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.PromiseCallbackWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.PromiseCallbackWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.SpanKindWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.SpanKindWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.TestActivitiesImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.TracerWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.TracerWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.UnservedWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.UnservedWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.UpdateTargetWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.WorkflowOutboundCall; +import io.temporal.opentelemetry.v2.TestWorkflows.WorkflowOutboundTagsWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.WorkflowOutboundTagsWorkflowImpl; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import org.junit.Rule; +import org.junit.Test; + +/** Tests interceptor behavior. */ +public class InterceptorTest extends OtelTestBase { + private static final String UNSERVED_TASK_QUEUE = "opentelemetry-v2-unserved"; + private static final String UPDATE_WITH_START_ID = "interceptor-update-with-start"; + private static final String TARGET_UPDATE_ID = "interceptor-update"; + private static final AttributeKey WORKFLOW_ID = + AttributeKey.stringKey("temporalWorkflowID"); + private static final AttributeKey RUN_ID = AttributeKey.stringKey("temporalRunID"); + private static final AttributeKey UPDATE_ID = AttributeKey.stringKey("temporalUpdateID"); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + newRuleBuilder(true) + .setWorkflowTypes( + SpanKindWorkflowImpl.class, + BenignErrorWorkflowImpl.class, + ErrorWorkflowImpl.class, + AsyncLambdaWorkflowImpl.class, + PromiseCallbackWorkflowImpl.class, + AsyncCompletionWorkflowImpl.class, + TracerWorkflowImpl.class, + ChildWorkflowWithSignalImpl.class, + UnservedWorkflowImpl.class, + UpdateTargetWorkflowImpl.class, + WorkflowOutboundTagsWorkflowImpl.class) + .setActivityImplementations(new TestActivitiesImpl()) + .build(); + + @Test + public void spanKind() { + testWorkflowRule.newWorkflowStub(SpanKindWorkflow.class).run(); + + Map kinds = new HashMap<>(); + for (SpanData span : endedSpans()) { + kinds.put(span.getName(), span.getKind()); + } + assertEquals(SpanKind.SERVER, kinds.get("RunWorkflow:SpanKindWorkflow")); + assertEquals(SpanKind.CLIENT, kinds.get("StartActivity:NopActivity")); + assertEquals(SpanKind.SERVER, kinds.get("RunActivity:NopActivity")); + } + + @Test + public void asyncLambdaPreservesApplicationContext() { + testWorkflowRule.newWorkflowStub(AsyncLambdaWorkflow.class).run(); + + assertSpanTree( + Arrays.asList( + "StartWorkflow:AsyncLambdaWorkflow", + " RunWorkflow:AsyncLambdaWorkflow", + " parent", + " child", + " StartActivity:NopActivity", + " RunActivity:NopActivity"), + endedSpans()); + } + + @Test + public void promiseCallbackPreservesApplicationContext() { + PromiseCallbackWorkflow workflow = + testWorkflowRule.newWorkflowStub(PromiseCallbackWorkflow.class); + WorkflowClient.start(workflow::run); + testWorkflowRule + .getWorkflowClient() + .newActivityCompletionClient() + .complete(takeTaskToken(), null); + WorkflowStub.fromTyped(workflow).getResult(Void.class); + + assertSpanTree( + Arrays.asList( + "StartWorkflow:PromiseCallbackWorkflow", + " RunWorkflow:PromiseCallbackWorkflow", + " application", + " StartActivity:AsyncCompletionActivity", + " RunActivity:AsyncCompletionActivity", + " callback", + " StartActivity:NopActivity", + " RunActivity:NopActivity"), + endedSpans()); + } + + @Test + public void workflowClientSignalIncludesTargetRunId() { + TracerWorkflow workflow = testWorkflowRule.newWorkflowStub(TracerWorkflow.class); + WorkflowExecution execution = WorkflowClient.start(workflow::run, false); + TracerWorkflow target = targetWorkflow(execution); + target.gate(); + WorkflowStub.fromTyped(workflow).getResult(Void.class); + + assertWorkflowExecutionTags(requireSpanNamed(endedSpans(), "SignalWorkflow:gate"), execution); + } + + @Test + public void workflowClientQueryIncludesTargetRunId() { + TracerWorkflow workflow = testWorkflowRule.newWorkflowStub(TracerWorkflow.class); + WorkflowExecution execution = WorkflowClient.start(workflow::run, false); + TracerWorkflow target = targetWorkflow(execution); + + assertEquals("ok", target.query()); + target.gate(); + WorkflowStub.fromTyped(workflow).getResult(Void.class); + + assertWorkflowExecutionTags(requireSpanNamed(endedSpans(), "QueryWorkflow:query"), execution); + } + + @Test + public void workflowClientUpdateIncludesTargetRunIdAndUpdateId() { + TracerWorkflow workflow = testWorkflowRule.newWorkflowStub(TracerWorkflow.class); + WorkflowExecution execution = WorkflowClient.start(workflow::run, false); + TracerWorkflow target = targetWorkflow(execution); + + WorkflowStub.fromTyped(target) + .startUpdate( + UpdateOptions.newBuilder(Void.class) + .setUpdateName("update") + .setUpdateId(TARGET_UPDATE_ID) + .setWaitForStage(WorkflowUpdateStage.COMPLETED) + .build()) + .getResult(); + target.gate(); + WorkflowStub.fromTyped(workflow).getResult(Void.class); + + SpanData update = requireSpanNamed(endedSpans(), "StartWorkflowUpdate:update"); + assertWorkflowExecutionTags(update, execution); + assertEquals(TARGET_UPDATE_ID, update.getAttributes().get(UPDATE_ID)); + } + + @Test + public void workflowClientStartIncludesWorkflowId() { + TracerWorkflow workflow = testWorkflowRule.newWorkflowStub(TracerWorkflow.class); + WorkflowExecution execution = WorkflowClient.start(workflow::run, true); + + SpanData start = requireSpanNamed(endedSpans(), "StartWorkflow:TracerWorkflow"); + assertEquals(execution.getWorkflowId(), start.getAttributes().get(WORKFLOW_ID)); + assertNull(start.getAttributes().get(RUN_ID)); + } + + @Test + public void workflowClientUpdateWithStartIncludesWorkflowIdAndUpdateId() { + String workflowId = "interceptor-update-with-start-target"; + WorkflowStub target = + testWorkflowRule + .getWorkflowClient() + .newUntypedWorkflowStub( + "UpdateTargetWorkflow", + WorkflowOptions.newBuilder() + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setWorkflowId(workflowId) + .setWorkflowIdConflictPolicy( + WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING) + .build()); + + target + .startUpdateWithStart( + UpdateOptions.newBuilder(Void.class) + .setUpdateName("doUpdate") + .setUpdateId(UPDATE_WITH_START_ID) + .setWaitForStage(WorkflowUpdateStage.COMPLETED) + .build(), + new Object[0], + new Object[0]) + .getResult(); + target.signal("updateSignal"); + target.getResult(Void.class); + + SpanData updateWithStart = requireSpanNamed(endedSpans(), "UpdateWithStartWorkflow:doUpdate"); + assertEquals(workflowId, updateWithStart.getAttributes().get(WORKFLOW_ID)); + assertEquals(UPDATE_WITH_START_ID, updateWithStart.getAttributes().get(UPDATE_ID)); + } + + @Test + public void workflowOutboundChildStartIncludesWorkflowId() { + runWorkflowOutboundCall(WorkflowOutboundCall.CHILD_WORKFLOW); + + SpanData start = requireSpanNamed(endedSpans(), "StartChildWorkflow:ChildWorkflowWithSignal"); + SpanData run = requireSpanNamed(endedSpans(), "RunWorkflow:ChildWorkflowWithSignal"); + assertEquals(run.getAttributes().get(WORKFLOW_ID), start.getAttributes().get(WORKFLOW_ID)); + assertNull(start.getAttributes().get(RUN_ID)); + } + + @Test + public void workflowOutboundChildSignalIncludesTargetExecution() { + runWorkflowOutboundCall(WorkflowOutboundCall.CHILD_WORKFLOW); + + SpanData run = requireSpanNamed(endedSpans(), "RunWorkflow:ChildWorkflowWithSignal"); + SpanData signal = requireSpanNamed(endedSpans(), "SignalExternalWorkflow:childSignal"); + assertWorkflowExecutionTags(signal, workflowExecution(run)); + } + + @Test + public void workflowOutboundExternalSignalIncludesTargetExecution() { + ChildWorkflowWithSignal target = + testWorkflowRule + .getWorkflowClient() + .newWorkflowStub( + ChildWorkflowWithSignal.class, + WorkflowOptions.newBuilder() + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setWorkflowId("external-signal-target-" + UUID.randomUUID()) + .build()); + WorkflowExecution execution = WorkflowClient.start(target::run); + + WorkflowOutboundTagsWorkflow workflow = + testWorkflowRule.newWorkflowStub(WorkflowOutboundTagsWorkflow.class); + workflow.run(WorkflowOutboundCall.EXTERNAL_SIGNAL, execution); + + assertWorkflowExecutionTags( + requireSpanNamed(endedSpans(), "SignalExternalWorkflow:childSignal"), execution); + } + + @Test + public void workflowOutboundCancelIncludesTargetExecution() { + UnservedWorkflow target = + testWorkflowRule + .getWorkflowClient() + .newWorkflowStub( + UnservedWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(UNSERVED_TASK_QUEUE) + .setWorkflowId("external-cancel-target-" + UUID.randomUUID()) + .build()); + WorkflowExecution execution = WorkflowClient.start(target::run); + + WorkflowOutboundTagsWorkflow workflow = + testWorkflowRule.newWorkflowStub(WorkflowOutboundTagsWorkflow.class); + workflow.run(WorkflowOutboundCall.EXTERNAL_CANCEL, execution); + + assertWorkflowExecutionTags(requireSpanNamed(endedSpans(), "CancelWorkflow"), execution); + } + + private TracerWorkflow targetWorkflow(WorkflowExecution execution) { + return testWorkflowRule + .getWorkflowClient() + .newWorkflowStub( + TracerWorkflow.class, + WorkflowTargetOptions.newBuilder().setWorkflowExecution(execution).build()); + } + + private void runWorkflowOutboundCall(WorkflowOutboundCall call) { + WorkflowOutboundTagsWorkflow workflow = + testWorkflowRule.newWorkflowStub(WorkflowOutboundTagsWorkflow.class); + workflow.run(call, WorkflowExecution.getDefaultInstance()); + } + + private static WorkflowExecution workflowExecution(SpanData span) { + return WorkflowExecution.newBuilder() + .setWorkflowId(span.getAttributes().get(WORKFLOW_ID)) + .setRunId(span.getAttributes().get(RUN_ID)) + .build(); + } + + private static void assertWorkflowExecutionTags(SpanData span, WorkflowExecution execution) { + assertEquals(execution.getWorkflowId(), span.getAttributes().get(WORKFLOW_ID)); + assertEquals(execution.getRunId(), span.getAttributes().get(RUN_ID)); + } + + @Test + public void benignErrorLeavesSpanStatusUnset() { + assertThrows( + WorkflowFailedException.class, + () -> testWorkflowRule.newWorkflowStub(BenignErrorWorkflow.class).run()); + + assertSpanTree( + Arrays.asList("StartWorkflow:BenignErrorWorkflow", " RunWorkflow:BenignErrorWorkflow"), + endedSpans()); + SpanData run = requireSpanNamed(endedSpans(), "RunWorkflow:BenignErrorWorkflow"); + assertEquals(StatusCode.UNSET, run.getStatus().getStatusCode()); + } + + @Test + public void errorSetsSpanStatusError() { + assertThrows( + WorkflowFailedException.class, + () -> testWorkflowRule.newWorkflowStub(ErrorWorkflow.class).run()); + + assertSpanTree( + Arrays.asList("StartWorkflow:ErrorWorkflow", " RunWorkflow:ErrorWorkflow"), endedSpans()); + SpanData run = requireSpanNamed(endedSpans(), "RunWorkflow:ErrorWorkflow"); + assertEquals(StatusCode.ERROR, run.getStatus().getStatusCode()); + } + + @Test + public void continueAsNewLeavesWorkflowSpanUnset() { + TracerWorkflow workflow = testWorkflowRule.newWorkflowStub(TracerWorkflow.class); + WorkflowClient.start(workflow::run, false); + workflow.gate(); + WorkflowStub.fromTyped(workflow).getResult(Void.class); + + SpanData run = requireSpanNamed(endedSpans(), "RunWorkflow:TracerWorkflow"); + assertEquals(StatusCode.UNSET, run.getStatus().getStatusCode()); + assertTrue(run.getEvents().toString(), run.getEvents().isEmpty()); + } + + @Test + public void pendingActivityLeavesActivitySpanUnset() { + AsyncCompletionWorkflow workflow = + testWorkflowRule.newWorkflowStub(AsyncCompletionWorkflow.class); + WorkflowClient.start(workflow::run); + byte[] taskToken = takeTaskToken(); + testWorkflowRule.getWorkflowClient().newActivityCompletionClient().complete(taskToken, null); + WorkflowStub.fromTyped(workflow).getResult(Void.class); + + SpanData activity = requireSpanNamed(endedSpans(), "RunActivity:AsyncCompletionActivity"); + assertEquals(StatusCode.UNSET, activity.getStatus().getStatusCode()); + assertTrue(activity.getEvents().toString(), activity.getEvents().isEmpty()); + } + + private static byte[] takeTaskToken() { + try { + byte[] taskToken = TASK_TOKENS.poll(30, TimeUnit.SECONDS); + assertNotNull("timed out waiting for activity task token", taskToken); + return taskToken; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/OpenTelemetryPluginGlobalTest.java b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/OpenTelemetryPluginGlobalTest.java new file mode 100644 index 0000000000..5f79be250c --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/OpenTelemetryPluginGlobalTest.java @@ -0,0 +1,32 @@ +package io.temporal.opentelemetry.v2; + +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import io.opentelemetry.api.GlobalOpenTelemetry; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class OpenTelemetryPluginGlobalTest { + @Before + @After + public void resetGlobalOpenTelemetry() { + GlobalOpenTelemetry.resetForTest(); + } + + @Test + public void buildRejectsAnUnregisteredGlobal() { + IllegalStateException e = + assertThrows(IllegalStateException.class, () -> OpenTelemetryPlugin.newBuilder().build()); + assertTrue(e.getMessage(), e.getMessage().contains("ReplaySafeOpenTelemetry")); + } + + @Test + public void buildAcceptsAReplaySafeGlobal() { + try (ReplaySafeOpenTelemetry openTelemetry = ReplaySafeOpenTelemetry.newBuilder().build()) { + GlobalOpenTelemetry.set(openTelemetry); + OpenTelemetryPlugin.newBuilder().build(); + } + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/OtelTestBase.java b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/OtelTestBase.java new file mode 100644 index 0000000000..31276368d3 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/OtelTestBase.java @@ -0,0 +1,151 @@ +package io.temporal.opentelemetry.v2; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.fail; + +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.worker.WorkerFactoryOptions; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; + +public abstract class OtelTestBase { + static final InMemorySpanExporter spanExporter = InMemorySpanExporter.create(); + private static ReplaySafeOpenTelemetry openTelemetry; + + @BeforeClass + public static void registerGlobalOpenTelemetry() { + openTelemetry = + ReplaySafeOpenTelemetry.newBuilder() + .setTracerProviderBuilder( + SdkTracerProvider.builder() + .addSpanProcessor(SimpleSpanProcessor.create(spanExporter))) + .build(); + GlobalOpenTelemetry.set(openTelemetry); + } + + @AfterClass + public static void resetGlobalOpenTelemetry() { + GlobalOpenTelemetry.resetForTest(); + openTelemetry.close(); + } + + @Before + public void clearSpans() { + spanExporter.reset(); + } + + /** + * A rule whose service stubs carry the plugin, so it propagates to every client and the worker, + * and whose workers keep no sticky cache so replay runs on every task. + */ + static SDKTestWorkflowRule.Builder newRuleBuilder(boolean addTemporalSpans) { + return SDKTestWorkflowRule.newBuilder() + .setWorkflowServiceStubsOptions( + WorkflowServiceStubsOptions.newBuilder() + .setPlugins( + OpenTelemetryPlugin.newBuilder().setAddTemporalSpans(addTemporalSpans).build()) + .build()) + .setWorkerFactoryOptions(WorkerFactoryOptions.newBuilder().setWorkflowCacheSize(0).build()); + } + + static List endedSpans() { + return spanExporter.getFinishedSpanItems(); + } + + static SpanData requireSpanNamed(List spans, String name) { + for (SpanData span : spans) { + if (span.getName().equals(name)) { + return span; + } + } + fail(name + " span not found in " + spanTree(spans)); + return null; + } + + static String requireSpanAttribute(SpanData span, AttributeKey key) { + String value = span.getAttributes().get(key); + assertNotNull(key.getKey() + " attribute not found on " + span.getName(), value); + return value; + } + + static void requireUniqueSpanIds(List spans) { + Map namesById = new HashMap<>(); + for (SpanData span : spans) { + String previous = namesById.put(span.getSpanId(), span.getName()); + if (previous != null) { + fail("span " + span.getName() + " shares an ID with span " + previous); + } + } + } + + static void assertSpanTree(List expected, List spans) { + assertEquals(String.join("\n", expected), String.join("\n", spanTree(spans))); + } + + /** Returns the spans as an indented tree in end order. */ + static List spanTree(List spans) { + Map> childrenByParent = new HashMap<>(); + for (int child = 0; child < spans.size(); child++) { + childrenByParent + .computeIfAbsent(closestParentIndex(spans, spans.get(child)), k -> new ArrayList<>()) + .add(child); + } + List tree = new ArrayList<>(); + appendChildren(spans, childrenByParent, -1, 0, tree); + return tree; + } + + /** + * Resets can emit the same span ID and start time more than once; the nearest end time identifies + * the matching parent. + */ + private static int closestParentIndex(List spans, SpanData child) { + String parentId = child.getParentSpanContext().getSpanId(); + int closest = -1; + long closestDistance = Long.MAX_VALUE; + for (int i = 0; i < spans.size(); i++) { + if (!spans.get(i).getSpanId().equals(parentId)) { + continue; + } + long distance = Math.abs(child.getEndEpochNanos() - spans.get(i).getEndEpochNanos()); + if (distance < closestDistance) { + closest = i; + closestDistance = distance; + } + } + return closest; + } + + private static void appendChildren( + List spans, + Map> childrenByParent, + int parent, + int depth, + List tree) { + for (int child : childrenByParent.getOrDefault(parent, new ArrayList<>())) { + tree.add(repeat(" ", depth) + spans.get(child).getName()); + appendChildren(spans, childrenByParent, child, depth + 1, tree); + } + } + + private static String repeat(String s, int times) { + StringBuilder out = new StringBuilder(); + for (int i = 0; i < times; i++) { + out.append(s); + } + return out.toString(); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/PropagationTest.java b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/PropagationTest.java new file mode 100644 index 0000000000..f9e18018f4 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/PropagationTest.java @@ -0,0 +1,507 @@ +package io.temporal.opentelemetry.v2; + +import static io.temporal.opentelemetry.v2.TestWorkflows.NEXUS_CANCEL_OPERATION_NAME; +import static io.temporal.opentelemetry.v2.TestWorkflows.NEXUS_OPERATION_NAME; +import static io.temporal.opentelemetry.v2.TestWorkflows.NEXUS_SERVICE_NAME; +import static io.temporal.opentelemetry.v2.TestWorkflows.TASK_TOKENS; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assume.assumeTrue; + +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.temporal.api.enums.v1.WorkflowIdConflictPolicy; +import io.temporal.client.StartActivityOptions; +import io.temporal.client.StartNexusOperationOptions; +import io.temporal.client.UpdateOptions; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.client.WorkflowUpdateStage; +import io.temporal.client.schedules.Schedule; +import io.temporal.client.schedules.ScheduleActionStartWorkflow; +import io.temporal.client.schedules.ScheduleClient; +import io.temporal.client.schedules.ScheduleClientOptions; +import io.temporal.client.schedules.ScheduleHandle; +import io.temporal.client.schedules.ScheduleOptions; +import io.temporal.client.schedules.ScheduleSpec; +import io.temporal.client.schedules.ScheduleUpdate; +import io.temporal.opentelemetry.v2.TestWorkflows.AsyncCompletionWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.AsyncCompletionWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.ChildWorkflowWithSignalImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.ComprehensiveNexusServiceImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.ComprehensiveWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.ComprehensiveWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.ContinueAsNewToDifferentWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.ContinueAsNewToDifferentWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.DifferentWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.ExternalWorkflowWithSignal; +import io.temporal.opentelemetry.v2.TestWorkflows.ExternalWorkflowWithSignalImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.NexusCancelHandlerWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.NexusHandlerWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.SchedulePropagationReceiver; +import io.temporal.opentelemetry.v2.TestWorkflows.SchedulePropagationReceiverImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.SchedulePropagationWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.SignalWithStartTargetImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.StandaloneWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.StandaloneWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.TestActivities; +import io.temporal.opentelemetry.v2.TestWorkflows.TestActivitiesImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.UnservedWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.UpdateTargetWorkflowImpl; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; + +/** + * Covers one scenario touching every traced operation, asserted as a span tree with Temporal spans + * on and off. + * + *

Needs a real server: client-started Nexus operations, standalone activities, and schedules are + * not implemented by the in-memory test server. CI runs it in the dev-server job. + */ +@RunWith(Parameterized.class) +public class PropagationTest extends OtelTestBase { + private static final String COMPREHENSIVE_UPDATE_ID = "comprehensive-update"; + private static final String UPDATE_WITH_START_UPDATE_ID = "comprehensive-update-with-start"; + private static final String UNSERVED_TASK_QUEUE = "opentelemetry-v2-unserved"; + private static final String TERMINATE_REASON = "otel-terminate-reason"; + + @Parameters(name = "addTemporalSpans={0}") + public static List addTemporalSpans() { + return Arrays.asList(true, false); + } + + private final boolean addTemporalSpans; + + private final String runId = UUID.randomUUID().toString(); + private final String scheduleId = "otel-schedule-" + runId; + private final String reusedScheduleSeedId = "otel-reused-schedule-seed-" + runId; + private final String reusedScheduleId = "otel-reused-schedule-" + runId; + private final String reusedScheduleWorkflowId = "otel-reused-schedule-workflow-" + runId; + private final String reusedScheduleReceiverId = "otel-reused-schedule-receiver-" + runId; + private final String externalWorkflowId = "externalWorkflowWithSignal-" + runId; + private final String comprehensiveWorkflowId = "comprehensive-outbound-" + runId; + private final String updateWithStartWorkflowId = "otel-update-with-start-" + runId; + + @Rule public SDKTestWorkflowRule testWorkflowRule; + + public PropagationTest(boolean addTemporalSpans) { + this.addTemporalSpans = addTemporalSpans; + this.testWorkflowRule = + newRuleBuilder(addTemporalSpans) + .setWorkflowTypes( + ComprehensiveWorkflowImpl.class, + ContinueAsNewToDifferentWorkflowImpl.class, + DifferentWorkflowImpl.class, + ChildWorkflowWithSignalImpl.class, + ExternalWorkflowWithSignalImpl.class, + NexusHandlerWorkflowImpl.class, + NexusCancelHandlerWorkflowImpl.class, + StandaloneWorkflowImpl.class, + SchedulePropagationWorkflowImpl.class, + SchedulePropagationReceiverImpl.class, + SignalWithStartTargetImpl.class, + UpdateTargetWorkflowImpl.class, + AsyncCompletionWorkflowImpl.class, + UnservedWorkflowImpl.class) + .setActivityImplementations(new TestActivitiesImpl()) + .setNexusServiceImplementation(new ComprehensiveNexusServiceImpl()) + .build(); + } + + @Before + public void requireRealServer() { + assumeTrue( + "Test Server doesn't support client Nexus operations, standalone activities, or schedules", + SDKTestWorkflowRule.useExternalService); + } + + @Test + public void comprehensive() { + List spans = runScenario(); + assertSpanTree(addTemporalSpans ? fullTree() : noTemporalSpansTree(), spans); + requireUniqueSpanIds(spans); + } + + /** Drives every traced operation under one client span and returns the ended spans. */ + private List runScenario() { + WorkflowClient client = testWorkflowRule.getWorkflowClient(); + String taskQueue = testWorkflowRule.getTaskQueue(); + String nexusEndpoint = testWorkflowRule.getNexusEndpoint().getSpec().getName(); + + // All client calls share this parent span. + Span clientSpan = + GlobalOpenTelemetry.getTracer("client").spanBuilder("client-span").startSpan(); + try (Scope ignored = clientSpan.makeCurrent()) { + // Start the external signal target first. + ExternalWorkflowWithSignal external = + client.newWorkflowStub( + ExternalWorkflowWithSignal.class, options(taskQueue, externalWorkflowId)); + WorkflowClient.start(external::run); + + client + .newWorkflowStub( + ContinueAsNewToDifferentWorkflow.class, + options(taskQueue, "otel-continue-as-new-" + UUID.randomUUID())) + .run(); + + ComprehensiveWorkflow comprehensive = + client.newWorkflowStub( + ComprehensiveWorkflow.class, options(taskQueue, comprehensiveWorkflowId)); + WorkflowClient.start(comprehensive::run, false, nexusEndpoint, externalWorkflowId); + WorkflowStub comprehensiveStub = WorkflowStub.fromTyped(comprehensive); + + comprehensiveStub + .startUpdate( + UpdateOptions.newBuilder(Void.class) + .setUpdateName("testUpdate") + .setUpdateId(COMPREHENSIVE_UPDATE_ID) + .setWaitForStage(WorkflowUpdateStage.COMPLETED) + .build()) + .getResult(); + assertEquals("ok", comprehensive.getStatus()); + comprehensive.proceed(); + comprehensiveStub.getResult(Void.class); + + testWorkflowRule + .getActivityClient() + .execute( + TestActivities.class, + TestActivities::standaloneActivity, + StartActivityOptions.newBuilder() + .setId("otel-standalone-activity-" + UUID.randomUUID()) + .setTaskQueue(taskQueue) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build()); + + client + .newWorkflowStub( + StandaloneWorkflow.class, + options(taskQueue, "otel-standalone-workflow-" + UUID.randomUUID())) + .run(null); + + AsyncCompletionWorkflow asyncCompletion = + client.newWorkflowStub( + AsyncCompletionWorkflow.class, + options(taskQueue, "otel-async-completion-" + UUID.randomUUID())); + WorkflowClient.start(asyncCompletion::run); + byte[] taskToken = takeTaskToken(); + client.newActivityCompletionClient().complete(taskToken, null); + WorkflowStub.fromTyped(asyncCompletion).getResult(Void.class); + + WorkflowStub cancelTarget = + client.newUntypedWorkflowStub( + "UnservedWorkflow", + options(UNSERVED_TASK_QUEUE, "otel-cancel-target-" + UUID.randomUUID())); + cancelTarget.start(); + cancelTarget.cancel(); + + WorkflowStub terminateTarget = + client.newUntypedWorkflowStub( + "UnservedWorkflow", + options(UNSERVED_TASK_QUEUE, "otel-terminate-target-" + UUID.randomUUID())); + terminateTarget.start(); + terminateTarget.terminate(TERMINATE_REASON); + + comprehensiveStub.describe(); + + ScheduleActionStartWorkflow scheduleAction = + ScheduleActionStartWorkflow.newBuilder() + .setWorkflowType("StandaloneWorkflow") + .setOptions(options(taskQueue, "otel-schedule-workflow-" + UUID.randomUUID())) + .build(); + ScheduleClient scheduleClient = + ScheduleClient.newInstance( + testWorkflowRule.getWorkflowServiceStubs(), + ScheduleClientOptions.newBuilder() + .setNamespace(client.getOptions().getNamespace()) + .build()); + ScheduleHandle schedule = + scheduleClient.createSchedule( + scheduleId, + Schedule.newBuilder() + .setAction(scheduleAction) + .setSpec(ScheduleSpec.newBuilder().build()) + .build(), + ScheduleOptions.newBuilder().build()); + try { + String updatedScheduleWorkflowId = "otel-schedule-updated-workflow-" + UUID.randomUUID(); + try (Scope rootScope = Context.root().makeCurrent()) { + schedule.update( + input -> + new ScheduleUpdate( + Schedule.newBuilder(input.getDescription().getSchedule()) + .setAction( + ScheduleActionStartWorkflow.newBuilder(scheduleAction) + .setOptions(options(taskQueue, updatedScheduleWorkflowId)) + .setArguments(externalWorkflowId) + .build()) + .build())); + } + schedule.trigger(); + String scheduledWorkflowId = WorkflowStub.fromTyped(external).getResult(String.class); + client.newUntypedWorkflowStub(scheduledWorkflowId).getResult(Void.class); + + // Reusing this action verifies a root-context create cannot retain the first create's + // trace. + SchedulePropagationReceiver reusedScheduleReceiver = + client.newWorkflowStub( + SchedulePropagationReceiver.class, options(taskQueue, reusedScheduleReceiverId)); + WorkflowClient.start(reusedScheduleReceiver::run); + ScheduleActionStartWorkflow reusedScheduleAction = + ScheduleActionStartWorkflow.newBuilder() + .setWorkflowType("SchedulePropagationWorkflow") + .setOptions(options(taskQueue, reusedScheduleWorkflowId)) + .setArguments(reusedScheduleReceiverId) + .build(); + ScheduleHandle reusedScheduleSeed = + scheduleClient.createSchedule( + reusedScheduleSeedId, + Schedule.newBuilder() + .setAction(reusedScheduleAction) + .setSpec(ScheduleSpec.newBuilder().build()) + .build(), + ScheduleOptions.newBuilder().build()); + ScheduleHandle reusedSchedule = null; + try { + try (Scope rootScope = Context.root().makeCurrent()) { + reusedSchedule = + scheduleClient.createSchedule( + reusedScheduleId, + Schedule.newBuilder() + .setAction(reusedScheduleAction) + .setSpec(ScheduleSpec.newBuilder().build()) + .build(), + ScheduleOptions.newBuilder().build()); + } + reusedSchedule.trigger(); + String reusedScheduledWorkflowId = + WorkflowStub.fromTyped(reusedScheduleReceiver).getResult(String.class); + client.newUntypedWorkflowStub(reusedScheduledWorkflowId).getResult(Void.class); + } finally { + reusedScheduleSeed.delete(); + if (reusedSchedule != null) { + reusedSchedule.delete(); + } + } + + WorkflowStub signalWithStart = + client.newUntypedWorkflowStub( + "SignalWithStartTarget", + options(taskQueue, "otel-signal-with-start-" + UUID.randomUUID())); + signalWithStart.signalWithStart("startSignal", new Object[0], new Object[0]); + signalWithStart.getResult(Void.class); + + WorkflowStub updateWithStart = + client.newUntypedWorkflowStub( + "UpdateTargetWorkflow", + options(taskQueue, updateWithStartWorkflowId).toBuilder() + .setWorkflowIdConflictPolicy( + WorkflowIdConflictPolicy.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING) + .build()); + updateWithStart + .startUpdateWithStart( + UpdateOptions.newBuilder(Void.class) + .setUpdateName("doUpdate") + .setUpdateId(UPDATE_WITH_START_UPDATE_ID) + .setWaitForStage(WorkflowUpdateStage.COMPLETED) + .build(), + new Object[0], + new Object[0]) + .getResult(); + updateWithStart.signal("updateSignal"); + updateWithStart.getResult(Void.class); + + testWorkflowRule + .getNexusClient() + .newUntypedNexusServiceClient(nexusEndpoint, NEXUS_SERVICE_NAME) + .execute( + NEXUS_OPERATION_NAME, + Void.class, + StartNexusOperationOptions.newBuilder() + .setId("otel-nexus-operation-" + UUID.randomUUID()) + .setScheduleToCloseTimeout(Duration.ofSeconds(10)) + .build(), + ""); + } finally { + schedule.delete(); + } + } finally { + clientSpan.end(); + } + return endedSpans(); + } + + private static WorkflowOptions options(String taskQueue, String workflowId) { + return WorkflowOptions.newBuilder().setTaskQueue(taskQueue).setWorkflowId(workflowId).build(); + } + + private static byte[] takeTaskToken() { + try { + byte[] taskToken = TASK_TOKENS.poll(30, TimeUnit.SECONDS); + assertNotNull("timed out waiting for activity task token", taskToken); + return taskToken; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + + private static String nexusOp(String operation) { + return NEXUS_SERVICE_NAME + "/" + operation; + } + + /** The span tree with Temporal spans enabled. */ + private List fullTree() { + return Arrays.asList( + // The scheduled workflow finishes before the client span, so its root appears first. + "UpdateSchedule:" + scheduleId, + " RunWorkflow:StandaloneWorkflow", + " standalone-workflow-span", + " SignalExternalWorkflow:scheduleStarted", + " HandleSignal:scheduleStarted", + "CreateSchedule:" + reusedScheduleId, + " RunWorkflow:SchedulePropagationWorkflow", + " schedule-propagation-workflow-span", + " SignalExternalWorkflow:scheduleStarted", + " HandleSignal:scheduleStarted", + "client-span", + " StartWorkflow:ExternalWorkflowWithSignal", + " RunWorkflow:ExternalWorkflowWithSignal", + " external-workflow-with-signal-span", + " StartWorkflow:ContinueAsNewToDifferentWorkflow", + " RunWorkflow:ContinueAsNewToDifferentWorkflow", + " ContinueAsNew:DifferentWorkflow", + " RunWorkflow:DifferentWorkflow", + " StartWorkflow:ComprehensiveWorkflow", + " RunWorkflow:ComprehensiveWorkflow", + " StartActivity:Activity", + " RunActivity:Activity", + " activity-span", + " StartActivity:LocalActivity", + " RunActivity:LocalActivity", + " local-activity-span", + " StartChildWorkflow:ChildWorkflowWithSignal", + " RunWorkflow:ChildWorkflowWithSignal", + " child-workflow-with-signal-span", + // There is no SignalChildWorkflow outbound method; child signals go through + // signalExternalWorkflow, so this span is named for that method. + " SignalExternalWorkflow:childSignal", + " HandleSignal:childSignal", + " SignalExternalWorkflow:externalSignal", + " HandleSignal:externalSignal", + " StartNexusOperation:" + nexusOp(NEXUS_OPERATION_NAME), + " RunStartNexusOperationHandler:" + nexusOp(NEXUS_OPERATION_NAME), + " StartWorkflow:NexusHandlerWorkflow", + " RunWorkflow:NexusHandlerWorkflow", + " workflow-with-nexus-handler-span", + " StartNexusOperation:" + nexusOp(NEXUS_CANCEL_OPERATION_NAME), + " RunStartNexusOperationHandler:" + nexusOp(NEXUS_CANCEL_OPERATION_NAME), + " StartWorkflow:NexusCancelHandlerWorkflow", + " RunWorkflow:NexusCancelHandlerWorkflow", + " nexus-cancel-handler-span", + " RunCancelNexusOperationHandler:" + nexusOp(NEXUS_CANCEL_OPERATION_NAME), + " CancelWorkflow", + // Continue-as-new links the outbound, continued-run, and user spans. + " ContinueAsNew:ComprehensiveWorkflow", + " RunWorkflow:ComprehensiveWorkflow", + " comprehensive-outbound-workflow-span", + " comprehensive-outbound-workflow-span", + // Update user spans follow their current inbound operation. + " StartWorkflowUpdate:testUpdate", + " ValidateUpdate:testUpdate", + " validate-update-span", + " validate-update-span-child", + " HandleUpdate:testUpdate", + " update-handler-span", + " update-handler-child-span", + // Query handler spans parent under the query that ran them. + " QueryWorkflow:getStatus", + " HandleQuery:getStatus", + " query-handler-span", + " query-handler-child-span", + " SignalWorkflow:proceed", + " HandleSignal:proceed", + // Headers link standalone StartActivity and RunActivity spans. + " StartActivity:StandaloneActivity", + " RunActivity:StandaloneActivity", + " StartWorkflow:StandaloneWorkflow", + " RunWorkflow:StandaloneWorkflow", + " StartWorkflow:AsyncCompletionWorkflow", + " RunWorkflow:AsyncCompletionWorkflow", + " StartActivity:AsyncCompletionActivity", + " RunActivity:AsyncCompletionActivity", + // Cancel, terminate, and describe have no propagation carrier but are still traced. + " StartWorkflow:UnservedWorkflow", + " CancelWorkflow", + " StartWorkflow:UnservedWorkflow", + " TerminateWorkflow", + " DescribeWorkflow", + " CreateSchedule:" + scheduleId, + " StartWorkflow:SchedulePropagationReceiver", + " RunWorkflow:SchedulePropagationReceiver", + " CreateSchedule:" + reusedScheduleSeedId, + // Signal-with-start links client, worker, signal, and user spans. + " SignalWithStartWorkflow:SignalWithStartTarget", + " HandleSignal:startSignal", + " RunWorkflow:SignalWithStartTarget", + " signal-with-start-target-span", + // Update-with-start links validation, execution, worker, and user spans. + " UpdateWithStartWorkflow:doUpdate", + " ValidateUpdate:doUpdate", + " HandleUpdate:doUpdate", + " update start", + " RunWorkflow:UpdateTargetWorkflow", + " update-target-workflow-span", + " SignalWorkflow:updateSignal", + " HandleSignal:updateSignal", + " StartNexusOperation:" + nexusOp(NEXUS_OPERATION_NAME), + " RunStartNexusOperationHandler:" + nexusOp(NEXUS_OPERATION_NAME), + " StartWorkflow:NexusHandlerWorkflow", + " RunWorkflow:NexusHandlerWorkflow", + " workflow-with-nexus-handler-span"); + } + + /** + * {@link #fullTree()} without any Temporal spans. Only user spans remain, and each reattaches to + * the nearest surviving ancestor. + */ + private static List noTemporalSpansTree() { + return Arrays.asList( + "standalone-workflow-span", + "schedule-propagation-workflow-span", + "client-span", + " validate-update-span", + " validate-update-span-child", + " update-handler-span", + " update-handler-child-span", + " query-handler-span", + " query-handler-child-span", + " activity-span", + " local-activity-span", + " child-workflow-with-signal-span", + " workflow-with-nexus-handler-span", + " nexus-cancel-handler-span", + // Continue-as-new emits the user span once per run. + " comprehensive-outbound-workflow-span", + " comprehensive-outbound-workflow-span", + " external-workflow-with-signal-span", + " signal-with-start-target-span", + " update start", + " update-target-workflow-span", + " workflow-with-nexus-handler-span"); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/ReplaySafeOpenTelemetryTest.java b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/ReplaySafeOpenTelemetryTest.java new file mode 100644 index 0000000000..84d310af53 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/ReplaySafeOpenTelemetryTest.java @@ -0,0 +1,28 @@ +package io.temporal.opentelemetry.v2; + +import static org.junit.Assert.assertTrue; + +import io.opentelemetry.context.Context; +import io.opentelemetry.context.ContextStorageProvider; +import io.temporal.opentelemetry.v2.internal.TemporalContextStorageProvider; +import java.util.ServiceLoader; +import org.junit.Test; + +public class ReplaySafeOpenTelemetryTest { + @Test + public void buildsAfterContextStorageIsInitialized() { + boolean providerRegistered = false; + for (ContextStorageProvider provider : ServiceLoader.load(ContextStorageProvider.class)) { + if (provider instanceof TemporalContextStorageProvider) { + providerRegistered = true; + break; + } + } + assertTrue(providerRegistered); + + Context.current(); + + try (ReplaySafeOpenTelemetry first = ReplaySafeOpenTelemetry.newBuilder().build(); + ReplaySafeOpenTelemetry second = ReplaySafeOpenTelemetry.newBuilder().build()) {} + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/TestWorkflows.java b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/TestWorkflows.java new file mode 100644 index 0000000000..edecc4fb90 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/TestWorkflows.java @@ -0,0 +1,879 @@ +package io.temporal.opentelemetry.v2; + +import io.nexusrpc.Operation; +import io.nexusrpc.Service; +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Scope; +import io.temporal.activity.Activity; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityOptions; +import io.temporal.activity.LocalActivityOptions; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.client.WorkflowOptions; +import io.temporal.failure.ApplicationErrorCategory; +import io.temporal.failure.ApplicationFailure; +import io.temporal.failure.CanceledFailure; +import io.temporal.failure.NexusOperationFailure; +import io.temporal.nexus.Nexus; +import io.temporal.nexus.WorkflowRunOperation; +import io.temporal.workflow.Async; +import io.temporal.workflow.CancellationScope; +import io.temporal.workflow.NexusOperationHandle; +import io.temporal.workflow.NexusOperationOptions; +import io.temporal.workflow.NexusServiceOptions; +import io.temporal.workflow.Promise; +import io.temporal.workflow.QueryMethod; +import io.temporal.workflow.SignalMethod; +import io.temporal.workflow.UpdateMethod; +import io.temporal.workflow.UpdateValidatorMethod; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.time.Duration; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +/** Workflow and activity fixtures used by the OpenTelemetry v2 tests. */ +public final class TestWorkflows { + private TestWorkflows() {} + + static final String TRACER_TEST_QUERY_NAME = "query"; + static final String TRACER_TEST_UPDATE_NAME = "update"; + static final String TRACER_TEST_SIGNAL_NAME = "gate"; + static final String NEXUS_SERVICE_NAME = "ComprehensiveNexusService"; + static final String NEXUS_OPERATION_NAME = "nexusHandlerWorkflow"; + static final String NEXUS_CANCEL_OPERATION_NAME = "nexusCancelHandlerWorkflow"; + + /** Stands in for the external system an async activity hands its task token to. */ + static final BlockingQueue TASK_TOKENS = new LinkedBlockingQueue<>(); + + static Span startSpan(String tracerName, String spanName) { + return GlobalOpenTelemetry.getTracer(tracerName).spanBuilder(spanName).startSpan(); + } + + static void spanAround(String tracerName, String spanName, Runnable body) { + Span span = startSpan(tracerName, spanName); + try (Scope ignored = span.makeCurrent()) { + body.run(); + } finally { + span.end(); + } + } + + // --------------------------------------------------------------------------------------------- + // Activities + // --------------------------------------------------------------------------------------------- + + @ActivityInterface + public interface TestActivities { + void activity(); + + void localActivity(); + + void asyncCompletionActivity(); + + void standaloneActivity(); + + void nopActivity(); + } + + public static class TestActivitiesImpl implements TestActivities { + @Override + public void activity() { + spanAround("activity", "activity-span", () -> {}); + } + + @Override + public void localActivity() { + spanAround("localActivity", "local-activity-span", () -> {}); + } + + @Override + public void asyncCompletionActivity() { + TASK_TOKENS.add(Activity.getExecutionContext().getTaskToken()); + Activity.getExecutionContext().doNotCompleteOnReturn(); + } + + @Override + public void standaloneActivity() {} + + @Override + public void nopActivity() {} + } + + // --------------------------------------------------------------------------------------------- + // Interceptor tests + // --------------------------------------------------------------------------------------------- + + @WorkflowInterface + public interface SpanKindWorkflow { + @WorkflowMethod + void run(); + } + + public static class SpanKindWorkflowImpl implements SpanKindWorkflow { + @Override + public void run() { + activities().nopActivity(); + } + } + + @WorkflowInterface + public interface AsyncLambdaWorkflow { + @WorkflowMethod + void run(); + } + + public static class AsyncLambdaWorkflowImpl implements AsyncLambdaWorkflow { + @Override + public void run() { + Span parent = startSpan("asyncLambda", "parent"); + try (Scope ignored = parent.makeCurrent()) { + Async.function( + () -> { + Span child = startSpan("asyncLambda", "child"); + try (Scope ignoredChild = child.makeCurrent()) { + activities().nopActivity(); + } finally { + child.end(); + } + return null; + }) + .get(); + } finally { + parent.end(); + } + } + } + + @WorkflowInterface + public interface PromiseCallbackWorkflow { + @WorkflowMethod + void run(); + } + + public static class PromiseCallbackWorkflowImpl implements PromiseCallbackWorkflow { + @Override + public void run() { + Span application = startSpan("promiseCallback", "application"); + try (Scope ignored = application.makeCurrent()) { + Promise continuation = + Async.procedure(activities()::asyncCompletionActivity) + .thenApply( + ignoredResult -> { + Span callback = startSpan("promiseCallback", "callback"); + try (Scope ignoredCallback = callback.makeCurrent()) { + activities().nopActivity(); + } finally { + callback.end(); + } + return null; + }); + continuation.get(); + } finally { + application.end(); + } + } + } + + @WorkflowInterface + public interface BenignErrorWorkflow { + @WorkflowMethod + void run(); + } + + public static class BenignErrorWorkflowImpl implements BenignErrorWorkflow { + @Override + public void run() { + throw ApplicationFailure.newBuilder() + .setMessage("expected error") + .setType("BenignError") + .setCategory(ApplicationErrorCategory.BENIGN) + .build(); + } + } + + @WorkflowInterface + public interface ErrorWorkflow { + @WorkflowMethod + void run(); + } + + public static class ErrorWorkflowImpl implements ErrorWorkflow { + @Override + public void run() { + throw ApplicationFailure.newFailure("unexpected error", "UnexpectedError"); + } + } + + // --------------------------------------------------------------------------------------------- + // Tracer tests + // --------------------------------------------------------------------------------------------- + + @WorkflowInterface + public interface TracerWorkflow { + @WorkflowMethod + void run(boolean end); + + @QueryMethod(name = TRACER_TEST_QUERY_NAME) + String query(); + + @UpdateValidatorMethod(updateName = TRACER_TEST_UPDATE_NAME) + void validateUpdate(); + + @UpdateMethod(name = TRACER_TEST_UPDATE_NAME) + void update(); + + @SignalMethod(name = TRACER_TEST_SIGNAL_NAME) + void gate(); + } + + public static class TracerWorkflowImpl implements TracerWorkflow { + private boolean gated; + + @Override + public void run(boolean end) { + if (!end) { + Workflow.await(() -> gated); + } + + // Both spans end before continue-as-new, so the next run's context starts from the root. + Span beginProcessing = startSpan("processorTracer", "process start"); + try (Scope ignored = beginProcessing.makeCurrent()) { + startSpan("recorderTracer", "record results").end(); + } finally { + beginProcessing.end(); + } + + if (!end) { + Workflow.continueAsNew(true); + } + } + + @Override + public String query() { + startSpan("queryTracer", "query start").end(); + return "ok"; + } + + @Override + public void update() { + startSpan("updateTracer", "update start").end(); + } + + @Override + public void validateUpdate() { + startSpan("validatorTracer", "validate start").end(); + } + + @Override + public void gate() { + gated = true; + } + } + + public enum WorkflowOutboundCall { + CHILD_WORKFLOW, + EXTERNAL_SIGNAL, + EXTERNAL_CANCEL + } + + @WorkflowInterface + public interface WorkflowOutboundTagsWorkflow { + @WorkflowMethod + void run(WorkflowOutboundCall call, WorkflowExecution targetExecution); + } + + public static class WorkflowOutboundTagsWorkflowImpl implements WorkflowOutboundTagsWorkflow { + @Override + public void run(WorkflowOutboundCall call, WorkflowExecution targetExecution) { + switch (call) { + case CHILD_WORKFLOW: + ChildWorkflowWithSignal child = + Workflow.newChildWorkflowStub(ChildWorkflowWithSignal.class); + Promise childResult = Async.procedure(child::run); + Workflow.getWorkflowExecution(child).get(); + child.childSignal(); + childResult.get(); + break; + case EXTERNAL_SIGNAL: + Workflow.newUntypedExternalWorkflowStub(targetExecution).signal("childSignal"); + break; + case EXTERNAL_CANCEL: + Workflow.newUntypedExternalWorkflowStub(targetExecution).cancel(); + break; + } + } + } + + @WorkflowInterface + public interface ChainedContinueAsNewWorkflow { + @WorkflowMethod + void run(boolean finalRun); + } + + /** Continues as new inside its user span, so the next run parents under that span. */ + public static class ChainedContinueAsNewWorkflowImpl implements ChainedContinueAsNewWorkflow { + @Override + public void run(boolean finalRun) { + spanAround( + "chainedContinueAsNewWorkflow", + "chained-span", + () -> { + if (!finalRun) { + Workflow.continueAsNew(true); + } + }); + } + } + + @WorkflowInterface + public interface ContinueAsNewToDifferentWorkflow { + @WorkflowMethod + void run(); + } + + public static class ContinueAsNewToDifferentWorkflowImpl + implements ContinueAsNewToDifferentWorkflow { + @Override + public void run() { + Workflow.continueAsNew(DifferentWorkflow.class.getSimpleName(), null); + } + } + + @WorkflowInterface + public interface DifferentWorkflow { + @WorkflowMethod + void run(); + } + + public static class DifferentWorkflowImpl implements DifferentWorkflow { + @Override + public void run() {} + } + + @WorkflowInterface + public interface TracerResetWorkflow { + @WorkflowMethod + void run(); + } + + public static class TracerResetWorkflowImpl implements TracerResetWorkflow { + @Override + public void run() { + // Ended but kept current, so later spans parent to it. + Span beginProcessing = startSpan("processorTracer", "process start"); + try (Scope ignored = beginProcessing.makeCurrent()) { + beginProcessing.end(); + Workflow.newActivityStub( + TestActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()) + .nopActivity(); + startSpan("recorderTracer", "record results").end(); + } + } + } + + @WorkflowInterface + public interface TracerResetLateSourceWorkflow { + @WorkflowMethod + void run(); + } + + /** Obtains the second tracer only after the reset point. */ + public static class TracerResetLateSourceWorkflowImpl implements TracerResetLateSourceWorkflow { + @Override + public void run() { + Span beginProcessing = startSpan("processorTracer", "process start"); + try (Scope ignored = beginProcessing.makeCurrent()) { + beginProcessing.end(); + Workflow.newActivityStub( + TestActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()) + .nopActivity(); + Tracer recorder = GlobalOpenTelemetry.getTracer("recorderTracer"); + recorder.spanBuilder("record results").startSpan().end(); + } + } + } + + @WorkflowInterface + public interface TracerResetDuringSpan { + @WorkflowMethod + void run(); + } + + /** Both spans stay open across the reset point. */ + public static class TracerResetDuringSpanImpl implements TracerResetDuringSpan { + @Override + public void run() { + Span beginProcessing = startSpan("processorTracer", "process start"); + try (Scope ignored = beginProcessing.makeCurrent()) { + Span recordingResults = startSpan("recorderTracer", "record results"); + Workflow.newActivityStub( + TestActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()) + .nopActivity(); + beginProcessing.end(); + recordingResults.end(); + } + } + } + + @WorkflowInterface + public interface TracerWorkflowTaskRetry { + @WorkflowMethod + void run(); + } + + public static class TracerWorkflowTaskRetryImpl implements TracerWorkflowTaskRetry { + @Override + public void run() { + startSpan("test", "workflow-task-retry-span").end(); + throw new RuntimeException("intentional workflow task failure"); + } + } + + @WorkflowInterface + public interface TracerSpanTimestampWorkflow { + @WorkflowMethod + void run(); + } + + public static class TracerSpanTimestampWorkflowImpl implements TracerSpanTimestampWorkflow { + @Override + public void run() { + Span span = + GlobalOpenTelemetry.getTracer("timestampTracer") + .spanBuilder("explicit timestamp") + .setStartTimestamp(123456789L, TimeUnit.MILLISECONDS) + .startSpan(); + try { + Workflow.newActivityStub( + TestActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()) + .nopActivity(); + } finally { + span.end(); + } + } + } + + // --------------------------------------------------------------------------------------------- + // Comprehensive scenario + // --------------------------------------------------------------------------------------------- + + @WorkflowInterface + public interface ExternalWorkflowWithSignal { + @WorkflowMethod + String run(); + + @SignalMethod + void externalSignal(); + + @SignalMethod + void scheduleStarted(String workflowId); + } + + public static class ExternalWorkflowWithSignalImpl implements ExternalWorkflowWithSignal { + private boolean externalSignaled; + private String scheduledWorkflowId; + + @Override + public String run() { + spanAround( + "externalWorkflowWithSignal", + "external-workflow-with-signal-span", + () -> Workflow.await(() -> externalSignaled && scheduledWorkflowId != null)); + return scheduledWorkflowId; + } + + @Override + public void externalSignal() { + externalSignaled = true; + } + + @Override + public void scheduleStarted(String workflowId) { + scheduledWorkflowId = workflowId; + } + } + + @WorkflowInterface + public interface ChildWorkflowWithSignal { + @WorkflowMethod + void run(); + + @SignalMethod + void childSignal(); + } + + public static class ChildWorkflowWithSignalImpl implements ChildWorkflowWithSignal { + private boolean signaled; + + @Override + public void run() { + spanAround( + "childWorkflowWithSignal", + "child-workflow-with-signal-span", + () -> Workflow.await(() -> signaled)); + } + + @Override + public void childSignal() { + signaled = true; + } + } + + @WorkflowInterface + public interface NexusHandlerWorkflow { + @WorkflowMethod + Void run(String input); + } + + public static class NexusHandlerWorkflowImpl implements NexusHandlerWorkflow { + @Override + public Void run(String input) { + spanAround("workflowWithNexusHandler", "workflow-with-nexus-handler-span", () -> {}); + return null; + } + } + + @WorkflowInterface + public interface NexusCancelHandlerWorkflow { + @WorkflowMethod + Void run(String input); + } + + /** Waits until the Nexus caller cancels it. */ + public static class NexusCancelHandlerWorkflowImpl implements NexusCancelHandlerWorkflow { + @Override + public Void run(String input) { + spanAround( + "nexusCancelHandlerWorkflow", + "nexus-cancel-handler-span", + () -> Workflow.await(() -> false)); + return null; + } + } + + @Service(name = NEXUS_SERVICE_NAME) + public interface ComprehensiveNexusService { + @Operation(name = NEXUS_OPERATION_NAME) + Void nexusHandlerWorkflow(String input); + + @Operation(name = NEXUS_CANCEL_OPERATION_NAME) + Void nexusCancelHandlerWorkflow(String input); + } + + @ServiceImpl(service = ComprehensiveNexusService.class) + public static class ComprehensiveNexusServiceImpl { + @OperationImpl + public OperationHandler nexusHandlerWorkflow() { + return WorkflowRunOperation.fromWorkflowMethod( + (context, details, input) -> + Nexus.getOperationContext() + .getWorkflowClient() + .newWorkflowStub( + NexusHandlerWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId("nexus-handler-" + details.getRequestId()) + .build()) + ::run); + } + + @OperationImpl + public OperationHandler nexusCancelHandlerWorkflow() { + return WorkflowRunOperation.fromWorkflowMethod( + (context, details, input) -> + Nexus.getOperationContext() + .getWorkflowClient() + .newWorkflowStub( + NexusCancelHandlerWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId("nexus-cancel-handler-" + details.getRequestId()) + .build()) + ::run); + } + } + + @WorkflowInterface + public interface ComprehensiveWorkflow { + @WorkflowMethod + void run(boolean finalRun, String nexusEndpoint, String externalWorkflowId); + + @QueryMethod(name = "getStatus") + String getStatus(); + + @UpdateMethod(name = "testUpdate") + void testUpdate(); + + @UpdateValidatorMethod(updateName = "testUpdate") + void validateTestUpdate(); + + @SignalMethod + void proceed(); + } + + public static class ComprehensiveWorkflowImpl implements ComprehensiveWorkflow { + private boolean proceed; + + @Override + public void run(boolean finalRun, String nexusEndpoint, String externalWorkflowId) { + // The returned context is discarded, so the span is not made current. The outbound calls + // below parent to the RunWorkflow span and this span is their sibling. + Span span = startSpan("comprehensiveWorkflow", "comprehensive-outbound-workflow-span"); + try { + if (finalRun) { + return; + } + + Workflow.await(() -> proceed); + + activities().activity(); + Workflow.newLocalActivityStub( + TestActivities.class, + LocalActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build()) + .localActivity(); + + ChildWorkflowWithSignal child = + Workflow.newChildWorkflowStub(ChildWorkflowWithSignal.class); + Promise childResult = Async.procedure(child::run); + Workflow.getWorkflowExecution(child).get(); + child.childSignal(); + childResult.get(); + + Workflow.newExternalWorkflowStub(ExternalWorkflowWithSignal.class, externalWorkflowId) + .externalSignal(); + + ComprehensiveNexusService nexus = + Workflow.newNexusServiceStub( + ComprehensiveNexusService.class, + NexusServiceOptions.newBuilder() + .setEndpoint(nexusEndpoint) + .setOperationOptions( + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(10)) + .build()) + .build()); + nexus.nexusHandlerWorkflow(""); + + try { + Workflow.newCancellationScope( + () -> { + NexusOperationHandle handle = + Workflow.startNexusOperation(nexus::nexusCancelHandlerWorkflow, ""); + handle.getExecution().get(); + CancellationScope.current().cancel(); + handle.getResult().get(); + }) + .run(); + } catch (NexusOperationFailure failure) { + // Cancellation is expected. + if (!(failure.getCause() instanceof CanceledFailure)) { + throw failure; + } + } + + Workflow.continueAsNew(true, nexusEndpoint, externalWorkflowId); + } finally { + span.end(); + } + } + + @Override + public String getStatus() { + spanAround( + "comprehensiveWorkflow", + "query-handler-span", + () -> startSpan("comprehensiveWorkflow", "query-handler-child-span").end()); + return "ok"; + } + + @Override + public void testUpdate() { + spanAround( + "comprehensiveWorkflow", + "update-handler-span", + () -> startSpan("comprehensiveWorkflow", "update-handler-child-span").end()); + } + + @Override + public void validateTestUpdate() { + spanAround( + "comprehensiveWorkflow", + "validate-update-span", + () -> startSpan("comprehensiveWorkflow", "validate-update-span-child").end()); + } + + @Override + public void proceed() { + proceed = true; + } + } + + @WorkflowInterface + public interface StandaloneWorkflow { + @WorkflowMethod + void run(String signalReceiverWorkflowId); + } + + public static class StandaloneWorkflowImpl implements StandaloneWorkflow { + @Override + public void run(String signalReceiverWorkflowId) { + if (signalReceiverWorkflowId != null) { + spanAround( + "standaloneWorkflow", + "standalone-workflow-span", + () -> + Workflow.newExternalWorkflowStub( + ExternalWorkflowWithSignal.class, signalReceiverWorkflowId) + .scheduleStarted(Workflow.getInfo().getWorkflowId())); + } + } + } + + @WorkflowInterface + public interface SchedulePropagationWorkflow { + @WorkflowMethod + void run(String signalReceiverWorkflowId); + } + + public static class SchedulePropagationWorkflowImpl implements SchedulePropagationWorkflow { + @Override + public void run(String signalReceiverWorkflowId) { + spanAround( + "schedulePropagationWorkflow", + "schedule-propagation-workflow-span", + () -> + Workflow.newExternalWorkflowStub( + SchedulePropagationReceiver.class, signalReceiverWorkflowId) + .scheduleStarted(Workflow.getInfo().getWorkflowId())); + } + } + + @WorkflowInterface + public interface SchedulePropagationReceiver { + @WorkflowMethod + String run(); + + @SignalMethod + void scheduleStarted(String workflowId); + } + + public static class SchedulePropagationReceiverImpl implements SchedulePropagationReceiver { + private String scheduledWorkflowId; + + @Override + public String run() { + Workflow.await(() -> scheduledWorkflowId != null); + return scheduledWorkflowId; + } + + @Override + public void scheduleStarted(String workflowId) { + scheduledWorkflowId = workflowId; + } + } + + @WorkflowInterface + public interface UnservedWorkflow { + @WorkflowMethod + void run(); + } + + public static class UnservedWorkflowImpl implements UnservedWorkflow { + @Override + public void run() {} + } + + @WorkflowInterface + public interface SignalWithStartTarget { + @WorkflowMethod + void run(); + + @SignalMethod + void startSignal(); + } + + public static class SignalWithStartTargetImpl implements SignalWithStartTarget { + private boolean signaled; + + @Override + public void run() { + spanAround( + "signalWithStartTarget", + "signal-with-start-target-span", + () -> Workflow.await(() -> signaled)); + } + + @Override + public void startSignal() { + signaled = true; + } + } + + @WorkflowInterface + public interface UpdateTargetWorkflow { + @WorkflowMethod + void run(); + + @UpdateMethod(name = "doUpdate") + void doUpdate(); + + @SignalMethod + void updateSignal(); + } + + public static class UpdateTargetWorkflowImpl implements UpdateTargetWorkflow { + private boolean signaled; + + @Override + public void run() { + spanAround( + "updateTargetWorkflow", + "update-target-workflow-span", + () -> Workflow.await(() -> signaled)); + } + + @Override + public void doUpdate() { + startSpan("updateTracer", "update start").end(); + } + + @Override + public void updateSignal() { + signaled = true; + } + } + + @WorkflowInterface + public interface AsyncCompletionWorkflow { + @WorkflowMethod + void run(); + } + + public static class AsyncCompletionWorkflowImpl implements AsyncCompletionWorkflow { + @Override + public void run() { + Workflow.newActivityStub( + TestActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(15)).build()) + .asyncCompletionActivity(); + } + } + + private static TestActivities activities() { + return Workflow.newActivityStub( + TestActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/TracerTest.java b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/TracerTest.java new file mode 100644 index 0000000000..ca406131a1 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/TracerTest.java @@ -0,0 +1,210 @@ +package io.temporal.opentelemetry.v2; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assume.assumeTrue; + +import io.opentelemetry.sdk.trace.data.SpanData; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.enums.v1.EventType; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.api.workflowservice.v1.ResetWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.ResetWorkflowExecutionResponse; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowFailedException; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.client.WorkflowTargetOptions; +import io.temporal.common.RetryOptions; +import io.temporal.failure.TimeoutFailure; +import io.temporal.opentelemetry.v2.TestWorkflows.ChainedContinueAsNewWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.ChainedContinueAsNewWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.TestActivitiesImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.TracerResetDuringSpanImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.TracerResetLateSourceWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.TracerResetWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.TracerSpanTimestampWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.TracerSpanTimestampWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.TracerWorkflow; +import io.temporal.opentelemetry.v2.TestWorkflows.TracerWorkflowImpl; +import io.temporal.opentelemetry.v2.TestWorkflows.TracerWorkflowTaskRetry; +import io.temporal.opentelemetry.v2.TestWorkflows.TracerWorkflowTaskRetryImpl; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import java.time.Duration; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import org.junit.Rule; +import org.junit.Test; + +/** + * Verifies that spans started by workflow code through the replay-safe global preserve parentage, + * with Temporal spans disabled. + */ +public class TracerTest extends OtelTestBase { + @Rule + public SDKTestWorkflowRule testWorkflowRule = + newRuleBuilder(false) + .setWorkflowTypes( + TracerWorkflowImpl.class, + ChainedContinueAsNewWorkflowImpl.class, + TracerResetWorkflowImpl.class, + TracerResetLateSourceWorkflowImpl.class, + TracerResetDuringSpanImpl.class, + TracerWorkflowTaskRetryImpl.class, + TracerSpanTimestampWorkflowImpl.class) + .setActivityImplementations(new TestActivitiesImpl()) + .build(); + + @Test + public void tracerWorkflow() { + TracerWorkflow workflow = testWorkflowRule.newWorkflowStub(TracerWorkflow.class); + WorkflowClient.start(workflow::run, false); + + workflow.update(); + workflow.query(); + workflow.gate(); + WorkflowStub.fromTyped(workflow).getResult(Void.class); + workflow.query(); + + List spans = endedSpans(); + assertSpanTree( + Arrays.asList( + "validate start", + "update start", + "query start", + "process start", + " record results", + "process start", // ContinueAsNew + " record results", + // The query after completion replays the run with no span current, so its span is a + // root like the first query's. + "query start"), + spans); + requireUniqueSpanIds(spans); + } + + @Test + public void continueAsNewUnderUserSpan() { + testWorkflowRule.newWorkflowStub(ChainedContinueAsNewWorkflow.class).run(false); + + List spans = endedSpans(); + assertSpanTree(Arrays.asList("chained-span", " chained-span"), spans); + requireUniqueSpanIds(spans); + } + + @Test + public void workflowTaskRetryReusesSpanId() { + TracerWorkflowTaskRetry workflow = + testWorkflowRule + .getWorkflowClient() + .newWorkflowStub( + TracerWorkflowTaskRetry.class, + WorkflowOptions.newBuilder() + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setWorkflowRunTimeout(Duration.ofSeconds(1)) + .setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(1).build()) + .build()); + try { + workflow.run(); + throw new AssertionError("expected the workflow run to time out"); + } catch (WorkflowFailedException e) { + assertTrue(e.getCause().toString(), e.getCause() instanceof TimeoutFailure); + } + + List spans = endedSpans(); + assertTrue("expected the span from more than one task attempt", spans.size() > 1); + Set spanIds = new HashSet<>(); + for (SpanData span : spans) { + spanIds.add(span.getSpanId()); + } + assertEquals(spanTree(spans).toString(), 1, spanIds.size()); + } + + @Test + public void explicitSpanStartTimestampSurvivesReplay() { + TracerSpanTimestampWorkflow workflow = + testWorkflowRule.newWorkflowStub(TracerSpanTimestampWorkflow.class); + workflow.run(); + + SpanData span = requireSpanNamed(endedSpans(), "explicit timestamp"); + assertEquals(TimeUnit.MILLISECONDS.toNanos(123456789L), span.getStartEpochNanos()); + } + + @Test + public void resetWithTracerCreatedBeforeResetPoint() { + List spans = runAndReset(TracerResetWorkflowImpl.class); + assertSpanTree(Arrays.asList("process start", " record results", " record results"), spans); + requireUniqueSpanIds(spans); + } + + @Test + public void resetWithTracerCreatedAfterResetPoint() { + List spans = runAndReset(TracerResetLateSourceWorkflowImpl.class); + assertSpanTree(Arrays.asList("process start", " record results", " record results"), spans); + requireUniqueSpanIds(spans); + } + + @Test + public void resetWithSpanCrossingResetPoint() { + List spans = runAndReset(TracerResetDuringSpanImpl.class); + assertSpanTree( + Arrays.asList("process start", " record results", "process start", " record results"), + spans); + // These spans reuse their IDs because they were created before the reset point. + assertEquals(spans.get(0).getSpanContext(), spans.get(2).getSpanContext()); + assertEquals(spans.get(1).getSpanContext(), spans.get(3).getSpanContext()); + // The new spans end after the old ones. + assertTrue(spans.get(2).getEndEpochNanos() > spans.get(0).getEndEpochNanos()); + assertTrue(spans.get(3).getEndEpochNanos() > spans.get(1).getEndEpochNanos()); + } + + /** + * Runs a gated workflow to completion, resets it to the task that handled the gate signal so the + * work after the signal is redone, and returns every span from both runs. + */ + private List runAndReset(Class workflowImpl) { + assumeTrue( + "Test Server doesn't support reset workflow", SDKTestWorkflowRule.useExternalService); + WorkflowClient client = testWorkflowRule.getWorkflowClient(); + WorkflowStub stub = + client.newUntypedWorkflowStub( + workflowImpl.getInterfaces()[0].getSimpleName(), + WorkflowOptions.newBuilder().setTaskQueue(testWorkflowRule.getTaskQueue()).build()); + stub.start(); + WorkflowExecution execution = stub.getExecution(); + stub.getResult(Void.class); + + ResetWorkflowExecutionResponse response = + client + .getWorkflowServiceStubs() + .blockingStub() + .resetWorkflowExecution( + ResetWorkflowExecutionRequest.newBuilder() + .setNamespace(client.getOptions().getNamespace()) + .setWorkflowExecution(execution) + .setWorkflowTaskFinishEventId(secondWorkflowTaskCompletedEventId(execution)) + .setReason("Integration test") + .setRequestId(UUID.randomUUID().toString()) + .build()); + client + .newUntypedWorkflowStub( + WorkflowTargetOptions.newBuilder() + .setWorkflowId(execution.getWorkflowId()) + .setRunId(response.getRunId()) + .build()) + .getResult(Void.class); + + return endedSpans(); + } + + private long secondWorkflowTaskCompletedEventId(WorkflowExecution execution) { + List completed = + testWorkflowRule.getHistoryEvents( + execution.getWorkflowId(), EventType.EVENT_TYPE_WORKFLOW_TASK_COMPLETED); + return completed.get(1).getEventId(); + } +} diff --git a/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/internal/CodecTest.java b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/internal/CodecTest.java new file mode 100644 index 0000000000..8e53b4c7a6 --- /dev/null +++ b/contrib/temporal-opentelemetry-v2/src/test/java/io/temporal/opentelemetry/v2/internal/CodecTest.java @@ -0,0 +1,163 @@ +package io.temporal.opentelemetry.v2.internal; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import com.google.protobuf.ByteString; +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.baggage.Baggage; +import io.opentelemetry.api.baggage.propagation.W3CBaggagePropagator; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.SpanId; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceId; +import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import io.opentelemetry.context.propagation.ContextPropagators; +import io.opentelemetry.context.propagation.TextMapPropagator; +import io.temporal.api.common.v1.Payload; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.GlobalDataConverter; +import io.temporal.common.converter.JacksonJsonPayloadConverter; +import io.temporal.common.interceptors.Header; +import java.lang.reflect.Type; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +/** Unit tests for {@link SpanCodec} write/read on Temporal headers and Nexus maps. */ +public class CodecTest { + private static final String HEADER_KEY = "_tracer-data"; + private static final SpanContext SPAN = + SpanContext.create( + TraceId.fromLongs(0, 1), + SpanId.fromLong(2), + TraceFlags.getSampled(), + TraceState.getDefault()); + private static final Baggage BAGGAGE = Baggage.builder().put("key", "value").build(); + + private SpanCodec codec; + + @Before + public void installPropagators() { + GlobalOpenTelemetry.resetForTest(); + GlobalOpenTelemetry.set( + OpenTelemetry.propagating( + ContextPropagators.create( + TextMapPropagator.composite( + W3CTraceContextPropagator.getInstance(), W3CBaggagePropagator.getInstance())))); + codec = new SpanCodec(HEADER_KEY); + } + + @After + public void resetGlobalOpenTelemetry() { + GlobalOpenTelemetry.resetForTest(); + } + + @Test + public void writesAndReadsTemporalHeaders() { + Header header = new Header(new HashMap<>()); + try (Scope ignored = current().makeCurrent()) { + codec.write(header); + } + assertPropagated(codec.read(header)); + } + + @Test + public void writesAndReadsNexusHeaders() { + Map headers = new HashMap<>(); + try (Scope ignored = current().makeCurrent()) { + codec.write(headers); + } + assertPropagated(codec.read(headers)); + } + + @Test + public void leavesTemporalHeaderAloneWhenNothingToInject() { + Header header = new Header(new HashMap<>()); + try (Scope ignored = Context.root().makeCurrent()) { + codec.write(header); + } + assertNull(header.getValues().get(HEADER_KEY)); + } + + @Test + public void missingTemporalHeaderKeepsCurrentContext() { + try (Scope ignored = current().makeCurrent()) { + assertPropagated(codec.read(new Header(new HashMap<>()))); + } + } + + @Test + public void readsNexusHeadersCaseInsensitively() { + Map written = new HashMap<>(); + try (Scope ignored = current().makeCurrent()) { + codec.write(written); + } + Map upper = new HashMap<>(); + written.forEach((key, value) -> upper.put(key.toUpperCase(), value)); + assertPropagated(codec.read(upper)); + } + + @Test + public void preservesNexusHeaderKeyCase() { + Map headers = new HashMap<>(); + try (Scope ignored = current().makeCurrent()) { + codec.write(headers); + } + assertTrue(headers.containsKey("traceparent")); + assertTrue(headers.containsKey("baggage")); + } + + @Test + public void decodesPropertiesPayload() { + Properties carrier = new Properties(); + carrier.setProperty("traceparent", "00-trace-id-span-id-01"); + Payload payload = DefaultDataConverter.STANDARD_INSTANCE.toPayload(carrier).get(); + assertEquals(carrier, SpanCodec.decode(payload)); + } + + @Test + public void decodesLegacyMapPayloadViaGlobalConverter() { + Map carrier = new HashMap<>(); + carrier.put("traceparent", "00-trace-id-span-id-01"); + Payload payload = + Payload.newBuilder() + .putMetadata("encoding", ByteString.copyFromUtf8("legacy/json")) + .build(); + + DataConverter original = GlobalDataConverter.get(); + GlobalDataConverter.register( + new DefaultDataConverter(new JacksonJsonPayloadConverter()) { + @Override + public T fromPayload(Payload ignored, Class valueClass, Type type) { + return valueClass.cast(carrier); + } + }); + try { + assertEquals(carrier, SpanCodec.decode(payload)); + } finally { + GlobalDataConverter.register(original); + } + } + + private static Context current() { + return Context.root().with(Span.wrap(SPAN)).with(BAGGAGE); + } + + private static void assertPropagated(Context context) { + SpanContext span = Span.fromContext(context).getSpanContext(); + assertEquals(SPAN.getTraceId(), span.getTraceId()); + assertEquals(SPAN.getSpanId(), span.getSpanId()); + assertEquals("value", Baggage.fromContext(context).getEntryValue("key")); + } +} diff --git a/settings.gradle b/settings.gradle index 0b6bb9bda1..04b2e1ac4e 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-opentelemetry-v2' +project(':temporal-opentelemetry-v2').projectDir = file('contrib/temporal-opentelemetry-v2') 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 01ddcc47e2..3b264080f7 100644 --- a/temporal-bom/build.gradle +++ b/temporal-bom/build.gradle @@ -8,6 +8,7 @@ dependencies { constraints { api project(':temporal-kotlin') api project(':temporal-opentelemetry') + api project(':temporal-opentelemetry-v2') api project(':temporal-opentracing') api project(':temporal-aws-lambda') api project(':temporal-gcp-cloud-run-opentelemetry')