This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new fde63d427e docs: document the internal APIs that the OpenTelemetry
agent instruments (#3473)
fde63d427e is described below
commit fde63d427e92385556e4a767b78f48de86a4a9c1
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Aug 28 07:29:25 2026 +0100
docs: document the internal APIs that the OpenTelemetry agent instruments
(#3473)
Motivation:
The OpenTelemetry Java agent propagates trace context through pekko actors,
streams and remoting by attaching bytecode advice to specific internal
methods.
Its muzzle checks verify the classes the advice code calls, not the method
matchers, so renaming, inlining or restructuring a matched method silently
stops
context propagation with no error and no log.
Modification:
Document the matched methods where they are declared, naming the agent,
saying
what it uses the method for and linking to the instrumentation source. Mark
them
`@noinline` so the compiler cannot inline them out of the bytecode. List the
RemoteInstrument identifiers known to be taken by other projects so that
the next
implementer can pick a free one. No API changes.
Result:
The load-bearing internal APIs are recorded next to the code, and new
InstrumentationPointsSpec suites fail loudly if a matched name, arity or
parameter type disappears from the bytecode.
Tests:
- sbt "actor-tests/testOnly
org.apache.pekko.dispatch.InstrumentationPointsSpec" - 6 passed
- sbt "stream-tests/testOnly
org.apache.pekko.stream.impl.fusing.InstrumentationPointsSpec" - 1 passed
- sbt "remote/testOnly org.apache.pekko.remote.InstrumentationPointsSpec" -
10 passed
- sbt actor/mimaReportBinaryIssues stream/mimaReportBinaryIssues
remote/mimaReportBinaryIssues - clean
- sbt scalafmtCheckAll scalafmtSbtCheck - pass
- sbt headerCreateAll - headers added for the three new files
References:
Fixes #3472, Refs apache/pekko-http#1240,
Refs
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
---
.../pekko/dispatch/InstrumentationPointsSpec.scala | 85 +++++++++++++++++
.../scala/org/apache/pekko/actor/ActorCell.scala | 20 ++++
.../pekko/actor/LightArrayRevolverScheduler.scala | 18 ++++
.../org/apache/pekko/dispatch/Dispatcher.scala | 9 ++
.../scala/org/apache/pekko/dispatch/Mailbox.scala | 10 ++
.../scala/org/apache/pekko/remote/Endpoint.scala | 20 ++++
.../scala/org/apache/pekko/remote/Remoting.scala | 10 ++
.../pekko/remote/artery/InboundEnvelope.scala | 18 ++++
.../pekko/remote/artery/MessageDispatcher.scala | 9 ++
.../pekko/remote/artery/OutboundEnvelope.scala | 27 ++++++
.../pekko/remote/artery/RemoteInstrument.scala | 40 ++++++++
.../pekko/remote/transport/PekkoPduCodec.scala | 18 ++++
.../pekko/remote/InstrumentationPointsSpec.scala | 101 +++++++++++++++++++++
.../impl/fusing/InstrumentationPointsSpec.scala | 44 +++++++++
.../stream/impl/fusing/GraphInterpreter.scala | 11 +++
15 files changed, 440 insertions(+)
diff --git
a/actor-tests/src/test/scala/org/apache/pekko/dispatch/InstrumentationPointsSpec.scala
b/actor-tests/src/test/scala/org/apache/pekko/dispatch/InstrumentationPointsSpec.scala
new file mode 100644
index 0000000000..d86b1d00bf
--- /dev/null
+++
b/actor-tests/src/test/scala/org/apache/pekko/dispatch/InstrumentationPointsSpec.scala
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.dispatch
+
+import java.lang.reflect.Method
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+/**
+ * Guards the pekko-actor internals that the OpenTelemetry Java agent attaches
bytecode advice to.
+ *
+ * The agent matches these by name and signature, and its muzzle checks do not
verify the matchers. When a
+ * match stops applying the instrumentation is silently disabled, so this spec
fails loudly instead. Keep it
+ * in sync with https://github.com/apache/pekko/issues/3472
+ */
+class InstrumentationPointsSpec extends AnyWordSpec with Matchers {
+
+ private def declaredMethods(className: String): Array[Method] =
+ Class.forName(className, false, getClass.getClassLoader).getDeclaredMethods
+
+ private def paramTypeNames(m: Method): Seq[String] =
m.getParameterTypes.toIndexedSeq.map(_.getName)
+
+ "The pekko-actor methods instrumented by the OpenTelemetry Java agent"
should {
+
+ "include Dispatcher.dispatch(ActorCell, Envelope)" in {
+ declaredMethods("org.apache.pekko.dispatch.Dispatcher").filter(_.getName
== "dispatch").exists(
+ paramTypeNames(_) == Seq(
+ "org.apache.pekko.actor.ActorCell",
+ "org.apache.pekko.dispatch.Envelope")) shouldBe true
+ }
+
+ "include ActorCell.invoke(Envelope)" in {
+ declaredMethods("org.apache.pekko.actor.ActorCell").filter(_.getName ==
"invoke").exists(
+ paramTypeNames(_) == Seq("org.apache.pekko.dispatch.Envelope"))
shouldBe true
+ }
+
+ "include ActorCell.systemInvoke(SystemMessage)" in {
+ declaredMethods("org.apache.pekko.actor.ActorCell").filter(_.getName ==
"systemInvoke").exists(
+ paramTypeNames(_) ==
Seq("org.apache.pekko.dispatch.sysmsg.SystemMessage")) shouldBe true
+ }
+
+ "include DefaultSystemMessageQueue.systemEnqueue(ActorRef, SystemMessage)"
in {
+
declaredMethods("org.apache.pekko.dispatch.DefaultSystemMessageQueue").filter(
+ _.getName == "systemEnqueue").exists(
+ paramTypeNames(_) == Seq(
+ "org.apache.pekko.actor.ActorRef",
+ "org.apache.pekko.dispatch.sysmsg.SystemMessage")) shouldBe true
+ }
+
+ "include LightArrayRevolverScheduler.schedule taking a Runnable" in {
+
declaredMethods("org.apache.pekko.actor.LightArrayRevolverScheduler").filter(
+ _.getName == "schedule").exists(
+ paramTypeNames(_) == Seq(
+ "scala.concurrent.duration.FiniteDuration",
+ "scala.concurrent.duration.FiniteDuration",
+ "java.lang.Runnable",
+ "scala.concurrent.ExecutionContext")) shouldBe true
+ }
+
+ "include LightArrayRevolverScheduler.scheduleOnce taking a Runnable" in {
+
declaredMethods("org.apache.pekko.actor.LightArrayRevolverScheduler").filter(
+ _.getName == "scheduleOnce").exists(
+ paramTypeNames(_) == Seq(
+ "scala.concurrent.duration.FiniteDuration",
+ "java.lang.Runnable",
+ "scala.concurrent.ExecutionContext")) shouldBe true
+ }
+ }
+}
diff --git a/actor/src/main/scala/org/apache/pekko/actor/ActorCell.scala
b/actor/src/main/scala/org/apache/pekko/actor/ActorCell.scala
index 7d36e05889..36af280cee 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/ActorCell.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/ActorCell.scala
@@ -476,7 +476,17 @@ private[pekko] class ActorCell(
/*
* MESSAGE PROCESSING
*/
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent makes the
+ * context that `DefaultSystemMessageQueue.systemEnqueue` attached to the
system
+ * message current again while the message is handled.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/6f9ca5672ce84edbbe36ce0e14386c31d68f479f/instrumentation/pekko/pekko-actor-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pekkoactor/v1_0/PekkoActorCellInstrumentation.java
// Memory consistency is handled by the Mailbox (reading mailbox status then
processing messages, then writing mailbox status
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
final def systemInvoke(message: SystemMessage): Unit = {
/*
* When recreate/suspend/resume are received while restarting (i.e. between
@@ -544,7 +554,17 @@ private[pekko] class ActorCell(
invokeAll(new EarliestFirstSystemMessageList(message), calculateState)
}
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent makes the
+ * context that `Dispatcher.dispatch` attached to the `Envelope` current
again while
+ * the message is handled.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/6f9ca5672ce84edbbe36ce0e14386c31d68f479f/instrumentation/pekko/pekko-actor-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pekkoactor/v1_0/PekkoActorCellInstrumentation.java
// Memory consistency is handled by the Mailbox (reading mailbox status then
processing messages, then writing mailbox status
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
final def invoke(messageHandle: Envelope): Unit = {
val msg = messageHandle.message
val timeoutBeforeReceive = cancelReceiveTimeoutIfNeeded(msg)
diff --git
a/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
b/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
index 2fe787cb1e..f7e68fcac8 100644
---
a/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
+++
b/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
@@ -132,6 +132,15 @@ class LightArrayRevolverScheduler(config: Config, log:
LoggingAdapter, threadFac
super.scheduleWithFixedDelay(initialDelay, delay)(runnable)
}
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent replaces the
+ * `runnable` argument with a wrapper that carries the scheduling context
into every run.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/6f9ca5672ce84edbbe36ce0e14386c31d68f479f/instrumentation/pekko/pekko-actor-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pekkoactor/v1_0/PekkoScheduleInstrumentation.java
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
override protected def schedule(initialDelay: FiniteDuration, delay:
FiniteDuration, runnable: Runnable)(
implicit executor: ExecutionContext): Cancellable = {
checkPeriod(delay)
@@ -161,6 +170,15 @@ class LightArrayRevolverScheduler(config: Config, log:
LoggingAdapter, threadFac
}
}
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent replaces the
+ * `runnable` argument with a wrapper that carries the scheduling context
into the run.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/6f9ca5672ce84edbbe36ce0e14386c31d68f479f/instrumentation/pekko/pekko-actor-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pekkoactor/v1_0/PekkoScheduleInstrumentation.java
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
override def scheduleOnce(delay: FiniteDuration, runnable: Runnable)(
implicit executor: ExecutionContext): Cancellable =
try schedule(executor, runnable, roundUp(delay))
diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/Dispatcher.scala
b/actor/src/main/scala/org/apache/pekko/dispatch/Dispatcher.scala
index 951cd75d21..d53ece09c3 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/Dispatcher.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/Dispatcher.scala
@@ -67,7 +67,16 @@ class Dispatcher(
/**
* INTERNAL API
+ *
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent attaches the
+ * context that is current at send time to the [[Envelope]] here, and
+ * `ActorCell.invoke` makes it current again while the message is handled.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
*/
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/6f9ca5672ce84edbbe36ce0e14386c31d68f479f/instrumentation/pekko/pekko-actor-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pekkoactor/v1_0/PekkoDispatcherInstrumentation.java
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
protected[pekko] def dispatch(receiver: ActorCell, invocation: Envelope):
Unit = {
val mbox = receiver.mailbox
mbox.enqueue(receiver.self, invocation)
diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
b/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
index 03ebf96a50..fe1aee48af 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
@@ -497,7 +497,17 @@ private[pekko] trait SystemMessageQueue {
*/
private[pekko] trait DefaultSystemMessageQueue { self: Mailbox =>
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent attaches the
+ * context that is current at enqueue time to the system message here, and
+ * `ActorCell.systemInvoke` makes it current again while the message is
handled.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/6f9ca5672ce84edbbe36ce0e14386c31d68f479f/instrumentation/pekko/pekko-actor-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pekkoactor/v1_0/PekkoDefaultSystemMessageQueueInstrumentation.java
@tailrec
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
final def systemEnqueue(receiver: ActorRef, message: SystemMessage): Unit = {
assert(message.unlinked)
if (Mailbox.debug) println("" + receiver + " having enqueued " + message)
diff --git a/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
b/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
index 866607d52c..ed3af16840 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
@@ -73,6 +73,16 @@ private[remote] class DefaultMessageDispatcher(
private val remoteDaemon = provider.remoteDaemon
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent makes the
+ * received context current while the message is delivered.
`EndpointReader.dispatchMessage` reads like
+ * the natural hook but is inlined away by the compiler, so this is the
method that is matched.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
override def dispatch(
recipient: InternalActorRef,
recipientAddress: Address,
@@ -943,6 +953,16 @@ private[remote] class EndpointWriter(
trySendPureAck()
}
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent restores the
+ * context captured by `EndpointManager.Send` while the message is
serialized, which can happen much
+ * later than the send when the endpoint buffers.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
def writeSend(s: Send): Boolean =
try {
handle match {
diff --git a/remote/src/main/scala/org/apache/pekko/remote/Remoting.scala
b/remote/src/main/scala/org/apache/pekko/remote/Remoting.scala
index d314c1bc6a..9626bb8b7e 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/Remoting.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/Remoting.scala
@@ -325,6 +325,16 @@ private[remote] object EndpointManager {
final case class Listen(addressesPromise:
Promise[Seq[(PekkoProtocolTransport, Address)]]) extends RemotingCommand
case object StartupFinished extends RemotingCommand
case object ShutdownAndFlush extends RemotingCommand
+
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this class so avoid
changing it. The agent captures the
+ * sender's context in the constructor and carries it over in the
four-argument `copy`, which is the one
+ * used to add a sequence number.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
@InternalStableApi
final case class Send(
message: Any,
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/InboundEnvelope.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/InboundEnvelope.scala
index 906f355100..b10da25f28 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/InboundEnvelope.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/InboundEnvelope.scala
@@ -121,6 +121,15 @@ private[remote] final class ReusableInboundEnvelope
extends InboundEnvelope {
this
}
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. Envelopes are pooled,
+ * so the agent clears the previous context here to stop it leaking into
the next use.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
def clear(): Unit = {
_recipient = OptionVal.None
_message = null
@@ -130,6 +139,15 @@ private[remote] final class ReusableInboundEnvelope
extends InboundEnvelope {
_lane = 0
}
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent clears any
+ * context left over from the pooled envelope's previous use, and the
arity is part of the match.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
def init(
recipient: OptionVal[InternalActorRef],
sender: OptionVal[ActorRef],
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/MessageDispatcher.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/MessageDispatcher.scala
index ba9ae0241c..f8cac6df0f 100644
---
a/remote/src/main/scala/org/apache/pekko/remote/artery/MessageDispatcher.scala
+++
b/remote/src/main/scala/org/apache/pekko/remote/artery/MessageDispatcher.scala
@@ -36,6 +36,15 @@ private[remote] class MessageDispatcher(system:
ExtendedActorSystem, provider: R
private val log = Logging.withMarker(system, getClass.getName)
private val debugLogEnabled: Boolean = log.isDebugEnabled
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent makes the
+ * received context current while the message is delivered to the
recipient.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
def dispatch(inboundEnvelope: InboundEnvelope): Unit = {
import Logging.messageClassName
import provider.remoteSettings.Artery._
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/OutboundEnvelope.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/OutboundEnvelope.scala
index 6c3de542b2..61310f1865 100644
---
a/remote/src/main/scala/org/apache/pekko/remote/artery/OutboundEnvelope.scala
+++
b/remote/src/main/scala/org/apache/pekko/remote/artery/OutboundEnvelope.scala
@@ -71,15 +71,42 @@ private[remote] final class ReusableOutboundEnvelope
extends OutboundEnvelope {
this
}
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent carries the
+ * captured context over to the copy.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
def copy(): OutboundEnvelope =
(new ReusableOutboundEnvelope).init(_recipient, _message, _sender)
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. Envelopes are pooled,
+ * so the agent clears the previous context here to stop it leaking into
the next use.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
def clear(): Unit = {
_recipient = OptionVal.None
_message = null
_sender = OptionVal.None
}
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent captures the
+ * sender's context here, and the arity is part of the match.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
def init(recipient: OptionVal[RemoteActorRef], message: AnyRef, sender:
OptionVal[ActorRef]): OutboundEnvelope = {
_recipient = recipient
_message = message
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/RemoteInstrument.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/RemoteInstrument.scala
index 8315921c5a..9ff107afb3 100644
---
a/remote/src/main/scala/org/apache/pekko/remote/artery/RemoteInstrument.scala
+++
b/remote/src/main/scala/org/apache/pekko/remote/artery/RemoteInstrument.scala
@@ -52,6 +52,16 @@ abstract class RemoteInstrument {
* MUST be >=1 and <32.
*
* Values between 1 and 7 are reserved for Pekko internal use.
+ *
+ * Identifiers known to be taken, listed so that a new implementation can
pick a free one without having
+ * to grep other projects for collisions:
+ *
+ * - 0: Lightbend Telemetry (Cinnamon)
+ * - 1: Pekko's own `LoggingRemoteInstrument`
+ * - 8: Kamon
+ * - 9: Opentelemetry Java Instrumentation
+ *
+ * See https://github.com/apache/pekko/issues/3472
*/
def identifier: Byte
@@ -190,6 +200,16 @@ private[remote] final class RemoteInstruments(
// does any of the instruments want serialization timing?
private val serializationTimingEnabled =
instruments.exists(_.serializationTimingEnabled)
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. A `RemoteInstrument`
+ * is only handed the message, so the agent uses this method to associate
the envelope being serialized
+ * with the instrument that writes the context metadata.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
def serialize(outboundEnvelope: OptionVal[OutboundEnvelope], buffer:
ByteBuffer): Unit = {
if (instruments.nonEmpty && outboundEnvelope.isDefined) {
val startPos = buffer.position()
@@ -251,6 +271,16 @@ private[remote] final class RemoteInstruments(
}
}
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. It is the inbound
+ * counterpart of `serialize` and associates the envelope with the
instrument that reads the context
+ * metadata.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
def deserialize(inboundEnvelope: InboundEnvelope): Unit = {
if (inboundEnvelope.flag(EnvelopeBuffer.MetadataPresentFlag)) {
inboundEnvelope.envelopeBuffer.byteBuffer.position(EnvelopeBuffer.MetadataContainerAndLiteralSectionOffset)
@@ -408,7 +438,17 @@ private[remote] object RemoteInstruments {
def getKey(kl: Int): Byte = (kl >>> 26).toByte
def getLength(kl: Int): Int = kl & lengthMask
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent appends its
+ * own `RemoteInstrument` to the returned vector, which is what lets it
propagate context without users
+ * having to add it to `pekko.remote.artery.advanced.instruments`.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
@InternalStableApi
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
def create(system: ExtendedActorSystem, @nowarn("msg=never used") log:
LoggingAdapter): Vector[RemoteInstrument] = {
val c = system.settings.config
val path = "pekko.remote.artery.advanced.instruments"
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/transport/PekkoPduCodec.scala
b/remote/src/main/scala/org/apache/pekko/remote/transport/PekkoPduCodec.scala
index ecc4f0b5b7..3595cf8bfe 100644
---
a/remote/src/main/scala/org/apache/pekko/remote/transport/PekkoPduCodec.scala
+++
b/remote/src/main/scala/org/apache/pekko/remote/transport/PekkoPduCodec.scala
@@ -141,6 +141,15 @@ private[remote] object PekkoPduProtobufCodec extends
PekkoPduCodec {
ackBuilder
}
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent writes the
+ * propagated context into the PDU here.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
override def constructMessage(
localAddress: Address,
recipient: ActorRef,
@@ -211,6 +220,15 @@ private[remote] object PekkoPduProtobufCodec extends
PekkoPduCodec {
}
}
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent reads the
+ * propagated context out of the PDU here.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/pull/19823
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
override def decodeMessage(
raw: ByteString,
provider: RemoteActorRefProvider,
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/InstrumentationPointsSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/InstrumentationPointsSpec.scala
new file mode 100644
index 0000000000..e74b8cb7c6
--- /dev/null
+++
b/remote/src/test/scala/org/apache/pekko/remote/InstrumentationPointsSpec.scala
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.remote
+
+import java.lang.reflect.Method
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+/**
+ * Guards the pekko-remote internals that the OpenTelemetry Java agent
attaches bytecode advice to.
+ *
+ * The agent matches these by name and signature, and its muzzle checks do not
verify the matchers. When a
+ * match stops applying the instrumentation is silently disabled, so this spec
fails loudly instead. Keep it
+ * in sync with https://github.com/apache/pekko/issues/3472
+ */
+class InstrumentationPointsSpec extends AnyWordSpec with Matchers {
+
+ private def declaredMethods(className: String): Array[Method] =
+ Class.forName(className, false, getClass.getClassLoader).getDeclaredMethods
+
+ private def paramTypeNames(m: Method): Seq[String] =
m.getParameterTypes.toIndexedSeq.map(_.getName)
+
+ "The artery members instrumented by the OpenTelemetry Java agent" should {
+
+ "include RemoteInstruments.create returning a Vector" in {
+
declaredMethods("org.apache.pekko.remote.artery.RemoteInstruments$").filter(_.getName
== "create").exists(
+ _.getReturnType.getName == "scala.collection.immutable.Vector")
shouldBe true
+ }
+
+ "include RemoteInstruments.serialize taking a ByteBuffer as second
argument" in {
+
declaredMethods("org.apache.pekko.remote.artery.RemoteInstruments").filter(_.getName
== "serialize").exists(
+ _.getParameterTypes.lift(1).exists(_.getName ==
"java.nio.ByteBuffer")) shouldBe true
+ }
+
+ "include RemoteInstruments.deserialize taking an InboundEnvelope" in {
+
declaredMethods("org.apache.pekko.remote.artery.RemoteInstruments").filter(_.getName
== "deserialize").exists(
+ paramTypeNames(_) ==
Seq("org.apache.pekko.remote.artery.InboundEnvelope")) shouldBe true
+ }
+
+ "include the three argument ReusableOutboundEnvelope.init, and no arg copy
and clear" in {
+ val methods =
declaredMethods("org.apache.pekko.remote.artery.ReusableOutboundEnvelope")
+ methods.filter(_.getName == "init").exists(_.getParameterCount == 3)
shouldBe true
+ methods.filter(_.getName == "copy").exists(_.getParameterCount == 0)
shouldBe true
+ methods.filter(_.getName == "clear").exists(_.getParameterCount == 0)
shouldBe true
+ }
+
+ "include the nine argument ReusableInboundEnvelope.init, and no arg clear"
in {
+ val methods =
declaredMethods("org.apache.pekko.remote.artery.ReusableInboundEnvelope")
+ methods.filter(_.getName == "init").exists(_.getParameterCount == 9)
shouldBe true
+ methods.filter(_.getName == "clear").exists(_.getParameterCount == 0)
shouldBe true
+ }
+
+ "include artery MessageDispatcher.dispatch taking an InboundEnvelope" in {
+
declaredMethods("org.apache.pekko.remote.artery.MessageDispatcher").filter(_.getName
== "dispatch").exists(
+ paramTypeNames(_) ==
Seq("org.apache.pekko.remote.artery.InboundEnvelope")) shouldBe true
+ }
+ }
+
+ "The classic remoting members instrumented by the OpenTelemetry Java agent"
should {
+
+ "include the four argument EndpointManager.Send constructor and copy" in {
+ val send = Class.forName("org.apache.pekko.remote.EndpointManager$Send",
false, getClass.getClassLoader)
+ send.getDeclaredConstructors.exists(_.getParameterCount == 4) shouldBe
true
+ send.getDeclaredMethods.filter(_.getName ==
"copy").exists(_.getParameterCount == 4) shouldBe true
+ }
+
+ "include EndpointWriter.writeSend taking an EndpointManager.Send" in {
+
declaredMethods("org.apache.pekko.remote.EndpointWriter").filter(_.getName ==
"writeSend").exists(
+ paramTypeNames(_) ==
Seq("org.apache.pekko.remote.EndpointManager$Send")) shouldBe true
+ }
+
+ "include DefaultMessageDispatcher.dispatch taking a SerializedMessage as
third argument" in {
+
declaredMethods("org.apache.pekko.remote.DefaultMessageDispatcher").filter(_.getName
== "dispatch").exists(
+ _.getParameterTypes.lift(2).exists(
+ _.getName ==
"org.apache.pekko.remote.WireFormats$SerializedMessage")) shouldBe true
+ }
+
+ "include PekkoPduProtobufCodec.constructMessage and decodeMessage" in {
+ val codec =
declaredMethods("org.apache.pekko.remote.transport.PekkoPduProtobufCodec$")
+ codec.exists(_.getName == "constructMessage") shouldBe true
+ codec.filter(_.getName == "decodeMessage").exists(
+ _.getParameterTypes.headOption.exists(_.getName ==
"org.apache.pekko.util.ByteString")) shouldBe true
+ }
+ }
+}
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/InstrumentationPointsSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/InstrumentationPointsSpec.scala
new file mode 100644
index 0000000000..149446a795
--- /dev/null
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/InstrumentationPointsSpec.scala
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.impl.fusing
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+/**
+ * Guards the pekko-stream internal that the OpenTelemetry Java agent attaches
bytecode advice to.
+ *
+ * `GraphInterpreter.processPush` is a private method matched by name only,
and it is where the agent makes
+ * the context current before a stream stage hands an element to user code. A
rename or an inlining disables
+ * that silently, so this spec fails loudly instead. Keep it in sync with
+ * https://github.com/apache/pekko/issues/3472
+ */
+class InstrumentationPointsSpec extends AnyWordSpec with Matchers {
+
+ "The pekko-stream methods instrumented by the OpenTelemetry Java agent"
should {
+
+ "include GraphInterpreter.processPush(Connection)" in {
+ Class
+ .forName("org.apache.pekko.stream.impl.fusing.GraphInterpreter",
false, getClass.getClassLoader)
+ .getDeclaredMethods
+ .filter(_.getName == "processPush")
+ .exists(_.getParameterTypes.toIndexedSeq.map(_.getName) == Seq(
+ "org.apache.pekko.stream.impl.fusing.GraphInterpreter$Connection"))
shouldBe true
+ }
+ }
+}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala
index e91c25c508..914699d69f 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala
@@ -694,7 +694,18 @@ import pekko.stream.stage._
}
}
+ /**
+ * <p>
+ * Opentelemetry Java Instrumentation relies on this method so avoid
changing it. The agent makes the
+ * context of the in-flight element current here, which is how server
context reaches user code running
+ * inside a stream stage. It is matched by name only, and the method is
private, so a rename or an
+ * inlining silently disables context propagation.
+ * See https://github.com/apache/pekko/issues/3472
+ * </p>
+ */
+ // see
https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/6f9ca5672ce84edbbe36ce0e14386c31d68f479f/instrumentation/pekko/pekko-http-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/pekkohttp/v1_0/server/GraphInterpreterInstrumentation.java
@InternalStableApi
+ @noinline // Not inlined so that the agent can match the method in the
bytecode
private def processPush(connection: Connection): Unit = {
if (Debug)
println(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
