diff options
author | Ivan Topolnak <ivantopo@gmail.com> | 2013-07-10 18:13:49 -0300 |
---|---|---|
committer | Ivan Topolnak <ivantopo@gmail.com> | 2013-07-10 18:13:49 -0300 |
commit | e8dd6c83986f1ecd2d717c39bffe900b23b68854 (patch) | |
tree | 43e0feaa42225a3770922a9366e126590225719a /src | |
parent | d1e22b3f446c89413c67421f19ab5215ebdfcd43 (diff) | |
download | Kamon-e8dd6c83986f1ecd2d717c39bffe900b23b68854.tar.gz Kamon-e8dd6c83986f1ecd2d717c39bffe900b23b68854.tar.bz2 Kamon-e8dd6c83986f1ecd2d717c39bffe900b23b68854.zip |
complete disaster, wip
Diffstat (limited to 'src')
6 files changed, 234 insertions, 8 deletions
diff --git a/src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala b/src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala index f3ee4ee7..3ace3e77 100644 --- a/src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala +++ b/src/main/scala/kamon/instrumentation/ExecutorServiceMetrics.scala @@ -8,6 +8,47 @@ import kamon.metric.{MetricDirectory, ExecutorServiceMetricCollector} import akka.dispatch.{MonitorableThreadFactory, ExecutorServiceFactory} +/** + * ExecutorService monitoring base: + */ +trait ExecutorServiceCollector { + def updateActiveThreadCount(diff: Int): Unit + def updateTotalThreadCount(diff: Int): Unit + def updateQueueSize(diff: Int): Unit +} + +trait WatchedExecutorService { + def collector: ExecutorServiceCollector +} + + + + + + + + + + + + + + + + + + + + + + + + + + + + + case class NamedExecutorServiceFactoryDelegate(actorSystemName: String, dispatcherName: String, delegate: ExecutorServiceFactory) extends ExecutorServiceFactory { def createExecutorService: ExecutorService = delegate.createExecutorService } @@ -15,11 +56,13 @@ case class NamedExecutorServiceFactoryDelegate(actorSystemName: String, dispatch @Aspect class ExecutorServiceFactoryProviderInstrumentation { - @Pointcut("execution(* akka.dispatch.ExecutorServiceFactoryProvider+.createExecutorServiceFactory(..)) && args(id, threadFactory)") - def factoryMethodCall(id: String, threadFactory: ThreadFactory) = {} + @Pointcut("execution(* akka.dispatch.ExecutorServiceFactoryProvider+.createExecutorServiceFactory(..)) && args(dispatcherName, threadFactory) && if()") + def factoryMethodCall(dispatcherName: String, threadFactory: ThreadFactory): Boolean = { + true + } - @Around("factoryMethodCall(id, threadFactory)") - def enrichFactoryCreationWithNames(pjp: ProceedingJoinPoint, id: String, threadFactory: ThreadFactory): ExecutorServiceFactory = { + @Around("factoryMethodCall(dispatcherName, threadFactory)") + def enrichFactoryCreationWithNames(pjp: ProceedingJoinPoint, dispatcherName: String, threadFactory: ThreadFactory): ExecutorServiceFactory = { val delegate = pjp.proceed().asInstanceOf[ExecutorServiceFactory] // Safe Cast val actorSystemName = threadFactory match { @@ -27,7 +70,7 @@ class ExecutorServiceFactoryProviderInstrumentation { case _ => "Unknown" // Find an alternative way to find the actor system name in case we start seeing "Unknown" as the AS name. } - new NamedExecutorServiceFactoryDelegate(actorSystemName, id, delegate) + new NamedExecutorServiceFactoryDelegate(actorSystemName, dispatcherName, delegate) } } @@ -67,4 +110,39 @@ case class NamedExecutorServiceDelegate(fullName: String, delegate: ExecutorServ def invokeAny[T](tasks: util.Collection[_ <: Callable[T]]): T = delegate.invokeAny(tasks) def invokeAny[T](tasks: util.Collection[_ <: Callable[T]], timeout: Long, unit: TimeUnit): T = delegate.invokeAny(tasks, timeout, unit) def execute(command: Runnable) = delegate.execute(command) -}
\ No newline at end of file +} + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/main/scala/kamon/metric/MetricFilter.scala b/src/main/scala/kamon/metric/MetricFilter.scala new file mode 100644 index 00000000..fb117968 --- /dev/null +++ b/src/main/scala/kamon/metric/MetricFilter.scala @@ -0,0 +1,6 @@ +package kamon.metric + +object MetricFilter { + def actorSystem(system: String): Boolean = !system.startsWith("kamon") + def actor(path: String, system: String): Boolean = true +} diff --git a/src/main/scala/kamon/metric/Metrics.scala b/src/main/scala/kamon/metric/Metrics.scala index 46fb2b64..352c51a0 100644 --- a/src/main/scala/kamon/metric/Metrics.scala +++ b/src/main/scala/kamon/metric/Metrics.scala @@ -1,8 +1,10 @@ package kamon.metric -import java.util.concurrent.TimeUnit +import java.util.concurrent.{ConcurrentHashMap, ConcurrentSkipListSet, TimeUnit} import com.codahale.metrics._ import akka.actor.ActorRef +import java.util.concurrent.atomic.AtomicReference +import com.codahale.metrics trait MetricDepot { def include(name: String, metric: Metric): Unit @@ -38,6 +40,11 @@ object Metrics extends MetricDepot { } } +object Watched { + case object Actor + case object Dispatcher +} + object MetricDirectory { def nameForDispatcher(actorSystem: String, dispatcher: String) = s"/ActorSystem/${actorSystem}/Dispatcher/${dispatcher}/" @@ -53,8 +60,82 @@ object MetricDirectory { } +} + + + +case class ActorSystemMetrics(actorSystemName: String) { + val dispatchers = new ConcurrentHashMap[String, DispatcherMetrics] + + def registerDispatcher(dispatcherName: String): Option[DispatcherMetricCollector] = { + ??? + } + +} + + + +case class DispatcherMetricCollector(activeThreadCount: ValueDistributionCollector, poolSize: ValueDistributionCollector, queueSize: ValueDistributionCollector) + + +trait ValueDistributionCollector { + def update(value: Long): Unit + def snapshot: HistogramLike } +trait HistogramLike { + def median: Long + def max: Long + def min: Long +} + +case class CodaHaleValueDistributionCollector extends ValueDistributionCollector { + private[this] val histogram = new Histogram(new metrics.ExponentiallyDecayingReservoir()) + + def median: Long = ??? + + def max: Long = ??? + + def min: Long = ??? + + def snapshot: HistogramLike = histogram.getSnapshot + + def update(value: Long) = histogram.update(value) +} + + + + + + + + + +/** + * Dispatcher Metrics that we care about currently with a histogram-like nature: + * - Work Queue Size + * - Total/Active Thread Count + */ + + + +import annotation.tailrec +import java.util.concurrent.atomic.AtomicReference + +object Atomic { + def apply[T]( obj : T) = new Atomic(new AtomicReference(obj)) + implicit def toAtomic[T]( ref : AtomicReference[T]) : Atomic[T] = new Atomic(ref) +} + +class Atomic[T](val atomic : AtomicReference[T]) { + @tailrec + final def update(f: T => T) : T = { + val oldValue = atomic.get() + val newValue = f(oldValue) + if (atomic.compareAndSet(oldValue, newValue)) newValue else update(f) + } + def get() = atomic.get() +}
\ No newline at end of file diff --git a/src/main/scala/spraytest/FutureTesting.scala b/src/main/scala/spraytest/FutureTesting.scala index f592f6d7..b864d6d6 100644 --- a/src/main/scala/spraytest/FutureTesting.scala +++ b/src/main/scala/spraytest/FutureTesting.scala @@ -63,7 +63,7 @@ object TraceableFuture { implicit def toRegularFuture[T](tf: TraceableFuture[T]) = tf.future def apply[T](body: => T)(implicit transactionContext: TransactionContext, executor: ExecutionContext) = { - val wrappedBody = contextSwitchWrapper(body, TransactionContext(transactionContext.id, Nil)) + val wrappedBody = contextSwitchWrapper(body, TransactionContext(transactionContext.dispatcherName, Nil)) new TraceableFuture(Future { wrappedBody }) } diff --git a/src/main/scala/test/PingPong.scala b/src/main/scala/test/PingPong.scala new file mode 100644 index 00000000..f9d6869c --- /dev/null +++ b/src/main/scala/test/PingPong.scala @@ -0,0 +1,34 @@ +package test + +import akka.actor.{Props, Actor, ActorSystem} + +object PingPong extends App { + + val as = ActorSystem("ping-pong") + + val pinger = as.actorOf(Props[Pinger]) + val ponger = as.actorOf(Props[Ponger]) + + pinger.tell(Pong, ponger) + + + Thread.sleep(30000) + as.shutdown() + + +} + +case object Ping +case object Pong + +class Pinger extends Actor { + def receive = { + case Pong => sender ! Ping + } +} + +class Ponger extends Actor { + def receive = { + case Ping => sender ! Pong + } +} diff --git a/src/test/scala/kamon/instrumentation/DispatcherInstrumentationSpec.scala b/src/test/scala/kamon/instrumentation/DispatcherInstrumentationSpec.scala new file mode 100644 index 00000000..517a4ce0 --- /dev/null +++ b/src/test/scala/kamon/instrumentation/DispatcherInstrumentationSpec.scala @@ -0,0 +1,27 @@ +package kamon.instrumentation + +import org.scalatest.{Matchers, WordSpec} +import akka.actor.ActorSystem +import kamon.metric.MetricDirectory + +class DispatcherInstrumentationSpec extends WordSpec with Matchers{ + import MetricDirectory.dispatcherStats + + + "the dispatcher instrumentation" should { + "instrument a dispatcher that belongs to a non-filtered actor system" in { + val defaultDispatcherStats = dispatcherStats("single-dispatcher", "akka.actor.default-dispatcher") + + defaultDispatcherStats should not be None + + //KamonMetrics.watch[Actor] named "ivan" + + } + } + + + trait SingleDispatcherActorSystem { + val actorSystem = ActorSystem("single-dispatcher") + } +} + |