diff options
Diffstat (limited to 'src/main/scala/kamon/instrumentation')
-rw-r--r-- | src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala | 43 |
1 files changed, 43 insertions, 0 deletions
diff --git a/src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala b/src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala index 6c79806d..1f3564d3 100644 --- a/src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala +++ b/src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala @@ -8,6 +8,8 @@ import kamon.metric.{DispatcherMetricCollector, Histogram, MetricDirectory, Exec import akka.dispatch.{MonitorableThreadFactory, ExecutorServiceFactory} import com.typesafe.config.Config import kamon.Kamon +import scala.concurrent.forkjoin.ForkJoinPool +import akka.dispatch.ForkJoinExecutorConfigurator.AkkaForkJoinPool @Aspect @@ -23,6 +25,47 @@ class ActorSystemInstrumentation { } } +@Aspect("perthis(forkJoinPoolInstantiation(int, scala.concurrent.forkjoin.ForkJoinPool.ForkJoinWorkerThreadFactory, java.lang.Thread.UncaughtExceptionHandler))") +class ForkJoinPoolInstrumentation { + var activeThreadsHistogram: Histogram = _ + + @Pointcut("execution(akka.dispatch.ForkJoinExecutorConfigurator.AkkaForkJoinPool.new(..)) && args(parallelism, threadFactory, exceptionHandler)") + def forkJoinPoolInstantiation(parallelism: Int, threadFactory: ForkJoinPool.ForkJoinWorkerThreadFactory, exceptionHandler: Thread.UncaughtExceptionHandler) = {} + + @After("forkJoinPoolInstantiation(parallelism, threadFactory, exceptionHandler)") + def initializeMetrics(parallelism: Int, threadFactory: ForkJoinPool.ForkJoinWorkerThreadFactory, exceptionHandler: Thread.UncaughtExceptionHandler): Unit = { + val (actorSystemName, dispatcherName) = threadFactory match { + case mtf: MonitorableThreadFactory => splitName(mtf.name, Kamon.Metric.actorSystemNames) + case _ => ("Unknown", "Unknown") + } + + val metrics = Kamon.Metric.actorSystem(actorSystemName).get.registerDispatcher(dispatcherName) + for(m <- metrics) { + activeThreadsHistogram = m.activeThreadCount + println(s"Registered $dispatcherName for actor system $actorSystemName") + } + } + + def splitName(threadFactoryName: String, knownActorSystems: List[String]): (String, String) = { + knownActorSystems.find(threadFactoryName.startsWith(_)).map(asName => (asName, threadFactoryName.substring(asName.length+1))).getOrElse(("Unkown", "Unkown")) + } + + + + + @Pointcut("execution(* scala.concurrent.forkjoin.ForkJoinPool.scan(..)) && this(fjp)") + def forkJoinScan(fjp: AkkaForkJoinPool): Unit = {} + + @After("forkJoinScan(fjp)") + def updateMetrics(fjp: AkkaForkJoinPool): Unit = { + activeThreadsHistogram.update(fjp.getActiveThreadCount) + println("UPDATED THE COUNT TWOOOO!!!") + } + + + +} + |