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 | |
parent | afda5a6ac02a5cd314638e40250b49f66cf3b419 (diff) | |
download | Kamon-a9f568f562e1c4a358a3f63b3dcce2b38b5e14d6.tar.gz Kamon-a9f568f562e1c4a358a3f63b3dcce2b38b5e14d6.tar.bz2 Kamon-a9f568f562e1c4a358a3f63b3dcce2b38b5e14d6.zip |
wip
Diffstat (limited to 'src/main')
-rw-r--r-- | src/main/resources/META-INF/aop.xml | 5 | ||||
-rw-r--r-- | src/main/scala/kamon/Kamon.scala | 1 | ||||
-rw-r--r-- | src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala | 43 | ||||
-rw-r--r-- | src/main/scala/kamon/metric/Metrics.scala | 6 |
4 files changed, 52 insertions, 3 deletions
diff --git a/src/main/resources/META-INF/aop.xml b/src/main/resources/META-INF/aop.xml index b0a1d40d..38e5bb73 100644 --- a/src/main/resources/META-INF/aop.xml +++ b/src/main/resources/META-INF/aop.xml @@ -1,8 +1,8 @@ <!DOCTYPE aspectj PUBLIC "-//AspectJ//DTD//EN" "http://www.eclipse.org/aspectj/dtd/aspectj.dtd"> <aspectj> - <weaver options="-verbose"> - <!--<dump within="*" beforeandafter="true"/>--> + <weaver options="-verbose -showWeaveInfo"> + <dump within="*" beforeandafter="true"/> </weaver> <aspects> @@ -18,6 +18,7 @@ <!-- <aspect name="kamon.instrumentation.ExecutorServiceFactoryProviderInstrumentation"/> <aspect name="kamon.instrumentation.NamedExecutorServiceFactoryDelegateInstrumentation"/>--> <aspect name="kamon.instrumentation.ActorSystemInstrumentation"/> + <aspect name ="kamon.instrumentation.ForkJoinPoolInstrumentation"/> diff --git a/src/main/scala/kamon/Kamon.scala b/src/main/scala/kamon/Kamon.scala index 8fb3c24a..9946a1fd 100644 --- a/src/main/scala/kamon/Kamon.scala +++ b/src/main/scala/kamon/Kamon.scala @@ -36,6 +36,7 @@ object Kamon { object Metric { val actorSystems = new ConcurrentHashMap[String, ActorSystemMetrics] asScala + def actorSystemNames: List[String] = actorSystems.keys.toList def registerActorSystem(name: String) = actorSystems.getOrElseUpdate(name, ActorSystemMetrics(name)) def actorSystem(name: String): Option[ActorSystemMetrics] = actorSystems.get(name) 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!!!") + } + + + +} + diff --git a/src/main/scala/kamon/metric/Metrics.scala b/src/main/scala/kamon/metric/Metrics.scala index b0dc8ec5..46809d8f 100644 --- a/src/main/scala/kamon/metric/Metrics.scala +++ b/src/main/scala/kamon/metric/Metrics.scala @@ -88,7 +88,11 @@ case class ActorSystemMetrics(actorSystemName: String) { private[this] def createDispatcherCollector: DispatcherMetricCollector = DispatcherMetricCollector(CodahaleHistogram(), CodahaleHistogram(), CodahaleHistogram()) - def registerDispatcher(dispatcherName: String): Option[DispatcherMetricCollector] = Some(createDispatcherCollector) + def registerDispatcher(dispatcherName: String): Option[DispatcherMetricCollector] = { + val stats = createDispatcherCollector + dispatchers.put(dispatcherName, stats) + Some(stats) + } } |