diff options
Diffstat (limited to 'src/main/scala/akka/instrumentation/ActorRefTellInstrumentation.scala')
-rw-r--r-- | src/main/scala/akka/instrumentation/ActorRefTellInstrumentation.scala | 47 |
1 files changed, 38 insertions, 9 deletions
diff --git a/src/main/scala/akka/instrumentation/ActorRefTellInstrumentation.scala b/src/main/scala/akka/instrumentation/ActorRefTellInstrumentation.scala index f631b79a..218c09cc 100644 --- a/src/main/scala/akka/instrumentation/ActorRefTellInstrumentation.scala +++ b/src/main/scala/akka/instrumentation/ActorRefTellInstrumentation.scala @@ -1,12 +1,14 @@ package akka.instrumentation -import org.aspectj.lang.annotation.{Around, Pointcut, Aspect} +import org.aspectj.lang.annotation.{Before, Around, Pointcut, Aspect} import org.aspectj.lang.ProceedingJoinPoint -import akka.actor.{ActorRef} +import akka.actor.{Props, ActorSystem, ActorRef} import kamon.{Kamon, TraceContext} import akka.dispatch.Envelope +import com.codahale.metrics.{ExponentiallyDecayingReservoir, Histogram} +import kamon.metric.{MetricDirectory, Metrics} -case class TraceableMessage(traceContext: TraceContext, message: Any) +case class TraceableEnvelope(traceContext: TraceContext, message: Any, timeStamp: Long = System.nanoTime()) @Aspect @@ -22,8 +24,12 @@ class ActorRefTellInstrumentation { Kamon.context() match { case Some(ctx) => { - val traceableMessage = TraceableMessage(ctx, message) - proceed(getArgs.updated(0, traceableMessage)) + val traceableMessage = TraceableEnvelope(ctx, message) + + // update the args with the new message + val args = getArgs + args.update(0, traceableMessage) + proceed(args) } case None => proceed } @@ -31,19 +37,42 @@ class ActorRefTellInstrumentation { } -@Aspect +@Aspect("perthis(actorCellCreation(..))") class ActorCellInvokeInstrumentation { + val latencyHistogram: Histogram = new Histogram(new ExponentiallyDecayingReservoir) + val messagesPer + @volatile var shouldTrack = false + + @Pointcut("execution(akka.actor.ActorCell.new(..)) && args(system, ref, props, parent)") + def actorCellCreation(system: ActorSystem, ref: ActorRef, props: Props, parent: ActorRef): Unit = {} + + @Before("actorCellCreation(system, ref, props, parent)") + def registerMetricsInRegistry(system: ActorSystem, ref: ActorRef, props: Props, parent: ActorRef): Unit = { + val actorName = MetricDirectory.nameForActor(ref) + val histogramName = MetricDirectory.nameForMailbox(system.name, actorName) + + // TODO: Find a better way to filter the thins we don't want to measure. + if(system.name != "kamon" && actorName.startsWith("/user")) { + Metrics.registry.register(histogramName + "/cell", latencyHistogram) + shouldTrack = true + } + } + + + @Pointcut("execution(* akka.actor.ActorCell.invoke(*)) && args(envelope)") def invokingActorBehaviourAtActorCell(envelope: Envelope) = {} @Around("invokingActorBehaviourAtActorCell(envelope)") - def around(pjp: ProceedingJoinPoint, envelope: Envelope) = { + def around(pjp: ProceedingJoinPoint, envelope: Envelope): Unit = { import pjp._ envelope match { - case Envelope(TraceableMessage(ctx, msg), sender) => { + case Envelope(TraceableEnvelope(ctx, msg, timeStamp), sender) => { + latencyHistogram.update(System.nanoTime() - timeStamp) + Kamon.set(ctx) val originalEnvelope = envelope.copy(message = msg) @@ -54,4 +83,4 @@ class ActorCellInvokeInstrumentation { case _ => proceed } } -}
\ No newline at end of file +} |