try to handle issues in Pekko tracing when the scheduler is used (#12359)
This commit is contained in:
parent
b00354140f
commit
2cb3961af7
|
@ -23,6 +23,7 @@ public class PekkoActorInstrumentationModule extends InstrumentationModule {
|
||||||
return asList(
|
return asList(
|
||||||
new PekkoDispatcherInstrumentation(),
|
new PekkoDispatcherInstrumentation(),
|
||||||
new PekkoActorCellInstrumentation(),
|
new PekkoActorCellInstrumentation(),
|
||||||
new PekkoDefaultSystemMessageQueueInstrumentation());
|
new PekkoDefaultSystemMessageQueueInstrumentation(),
|
||||||
|
new PekkoScheduleInstrumentation());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -0,0 +1,64 @@
|
||||||
|
/*
|
||||||
|
* Copyright The OpenTelemetry Authors
|
||||||
|
* SPDX-License-Identifier: Apache-2.0
|
||||||
|
*/
|
||||||
|
|
||||||
|
package io.opentelemetry.javaagent.instrumentation.pekkoactor.v1_0;
|
||||||
|
|
||||||
|
import static net.bytebuddy.matcher.ElementMatchers.named;
|
||||||
|
import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
|
||||||
|
|
||||||
|
import io.opentelemetry.context.Context;
|
||||||
|
import io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge;
|
||||||
|
import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
|
||||||
|
import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
|
||||||
|
import net.bytebuddy.asm.Advice;
|
||||||
|
import net.bytebuddy.description.type.TypeDescription;
|
||||||
|
import net.bytebuddy.matcher.ElementMatcher;
|
||||||
|
|
||||||
|
public class PekkoScheduleInstrumentation implements TypeInstrumentation {
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public ElementMatcher<TypeDescription> typeMatcher() {
|
||||||
|
return named("org.apache.pekko.actor.LightArrayRevolverScheduler");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void transform(TypeTransformer transformer) {
|
||||||
|
transformer.applyAdviceToMethod(
|
||||||
|
named("schedule")
|
||||||
|
.and(takesArgument(0, named("scala.concurrent.duration.FiniteDuration")))
|
||||||
|
.and(takesArgument(1, named("scala.concurrent.duration.FiniteDuration")))
|
||||||
|
.and(takesArgument(2, named("java.lang.Runnable")))
|
||||||
|
.and(takesArgument(3, named("scala.concurrent.ExecutionContext"))),
|
||||||
|
PekkoScheduleInstrumentation.class.getName() + "$ScheduleAdvice");
|
||||||
|
transformer.applyAdviceToMethod(
|
||||||
|
named("scheduleOnce")
|
||||||
|
.and(takesArgument(0, named("scala.concurrent.duration.FiniteDuration")))
|
||||||
|
.and(takesArgument(1, named("java.lang.Runnable")))
|
||||||
|
.and(takesArgument(2, named("scala.concurrent.ExecutionContext"))),
|
||||||
|
PekkoScheduleInstrumentation.class.getName() + "$ScheduleOnceAdvice");
|
||||||
|
}
|
||||||
|
|
||||||
|
@SuppressWarnings("unused")
|
||||||
|
public static class ScheduleAdvice {
|
||||||
|
|
||||||
|
@Advice.OnMethodEnter(suppress = Throwable.class)
|
||||||
|
public static void enterSchedule(
|
||||||
|
@Advice.Argument(value = 2, readOnly = false) Runnable runnable) {
|
||||||
|
Context context = Java8BytecodeBridge.currentContext();
|
||||||
|
runnable = context.wrap(runnable);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@SuppressWarnings("unused")
|
||||||
|
public static class ScheduleOnceAdvice {
|
||||||
|
|
||||||
|
@Advice.OnMethodEnter(suppress = Throwable.class)
|
||||||
|
public static void enterScheduleOnce(
|
||||||
|
@Advice.Argument(value = 1, readOnly = false) Runnable runnable) {
|
||||||
|
Context context = Java8BytecodeBridge.currentContext();
|
||||||
|
runnable = context.wrap(runnable);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
|
@ -0,0 +1,50 @@
|
||||||
|
/*
|
||||||
|
* Copyright The OpenTelemetry Authors
|
||||||
|
* SPDX-License-Identifier: Apache-2.0
|
||||||
|
*/
|
||||||
|
|
||||||
|
package io.opentelemetry.javaagent.instrumentation.pekkoactor.v1_0
|
||||||
|
|
||||||
|
import io.opentelemetry.api.GlobalOpenTelemetry
|
||||||
|
import io.opentelemetry.api.trace.Span
|
||||||
|
import org.apache.pekko.actor.ActorSystem
|
||||||
|
import org.apache.pekko.pattern.after
|
||||||
|
import org.assertj.core.api.Assertions.assertThat
|
||||||
|
import org.junit.jupiter.api.Test
|
||||||
|
import scala.concurrent.{Await, Future}
|
||||||
|
import scala.concurrent.duration.DurationInt
|
||||||
|
|
||||||
|
class PekkoSchedulerTest {
|
||||||
|
|
||||||
|
@Test
|
||||||
|
def checkThatSpanWorksWithPekkoScheduledEvents(): Unit = {
|
||||||
|
val system = ActorSystem("my-system")
|
||||||
|
implicit val executionContext = system.dispatcher
|
||||||
|
val tracer = GlobalOpenTelemetry.get.getTracer("test-tracer")
|
||||||
|
val initialSpan = tracer.spanBuilder("test").startSpan()
|
||||||
|
val scope = initialSpan.makeCurrent()
|
||||||
|
try {
|
||||||
|
val futureResult = for {
|
||||||
|
result1 <- Future {
|
||||||
|
compareSpanContexts(Span.current(), initialSpan)
|
||||||
|
1
|
||||||
|
}
|
||||||
|
_ = compareSpanContexts(Span.current(), initialSpan)
|
||||||
|
result2 <- after(200.millis, system.scheduler)(Future.successful(2))
|
||||||
|
_ = compareSpanContexts(Span.current(), initialSpan)
|
||||||
|
} yield result1 + result2
|
||||||
|
assertThat(Await.result(futureResult, 5.seconds)).isEqualTo(3)
|
||||||
|
} finally {
|
||||||
|
system.terminate()
|
||||||
|
scope.close()
|
||||||
|
initialSpan.end()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private def compareSpanContexts(span1: Span, span2: Span): Unit = {
|
||||||
|
assertThat(span1.getSpanContext().getTraceId())
|
||||||
|
.isEqualTo(span2.getSpanContext().getTraceId())
|
||||||
|
assertThat(span1.getSpanContext().getSpanId())
|
||||||
|
.isEqualTo(span2.getSpanContext().getSpanId())
|
||||||
|
}
|
||||||
|
}
|
Loading…
Reference in New Issue