diff options
author | Ivan Topolnak <ivantopo@gmail.com> | 2013-07-16 18:40:02 -0300 |
---|---|---|
committer | Ivan Topolnak <ivantopo@gmail.com> | 2013-07-16 18:40:02 -0300 |
commit | a9f568f562e1c4a358a3f63b3dcce2b38b5e14d6 (patch) | |
tree | c6ac11a04d8d65954a8310c359ea8f2014eabbf8 /src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala | |
parent | afda5a6ac02a5cd314638e40250b49f66cf3b419 (diff) | |
download | Kamon-a9f568f562e1c4a358a3f63b3dcce2b38b5e14d6.tar.gz Kamon-a9f568f562e1c4a358a3f63b3dcce2b38b5e14d6.tar.bz2 Kamon-a9f568f562e1c4a358a3f63b3dcce2b38b5e14d6.zip |
wip
Diffstat (limited to 'src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala')
-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!!!") + } + + + +} + |