diff options
Diffstat (limited to 'kamon-akka/src')
19 files changed, 630 insertions, 693 deletions
diff --git a/kamon-akka/src/main/resources/reference.conf b/kamon-akka/src/main/resources/reference.conf index 4f742ee6..902a682d 100644 --- a/kamon-akka/src/main/resources/reference.conf +++ b/kamon-akka/src/main/resources/reference.conf @@ -13,39 +13,6 @@ kamon { ask-pattern-timeout-warning = off # Default dispatcher for all akka module operations - dispatcher = ${kamon.default-dispatcher} - } - - metrics.precision { - actor { - processing-time = ${kamon.metrics.precision.default-histogram-precision} - time-in-mailbox = ${kamon.metrics.precision.default-histogram-precision} - mailbox-size = ${kamon.metrics.precision.default-min-max-counter-precision} - } - - router { - routing-time = ${kamon.metrics.precision.default-histogram-precision} - processing-time = ${kamon.metrics.precision.default-histogram-precision} - time-in-mailbox = ${kamon.metrics.precision.default-histogram-precision} - } - - dispatcher { - maximum-pool-size { - highest-trackable-value = 999999999 - significant-value-digits = 2 - } - running-thread-count { - highest-trackable-value = 999999999 - significant-value-digits = 2 - } - queued-task-count { - highest-trackable-value = 999999999 - significant-value-digits = 2 - } - pool-size { - highest-trackable-value = 999999999 - significant-value-digits = 2 - } - } + dispatcher = "akka.actor.default-dispatcher" } }
\ No newline at end of file diff --git a/kamon-akka/src/main/scala/kamon/akka/ActorMetrics.scala b/kamon-akka/src/main/scala/kamon/akka/ActorMetrics.scala index b22f7fa9..c99df586 100644 --- a/kamon-akka/src/main/scala/kamon/akka/ActorMetrics.scala +++ b/kamon-akka/src/main/scala/kamon/akka/ActorMetrics.scala @@ -16,79 +16,26 @@ package kamon.akka -import akka.actor.ActorSystem -import com.typesafe.config.Config -import kamon.metric._ -import kamon.metric.instrument.{ Counter, Histogram, MinMaxCounter } +import kamon.metric.{ EntityRecorderFactory, GenericEntityRecorder } +import kamon.metric.instrument.{ Time, InstrumentFactory } -case class ActorMetrics(name: String) extends MetricGroupIdentity { - val category = ActorMetrics -} - -object ActorMetrics extends MetricGroupCategory { - val name = "actor" - - case object ProcessingTime extends MetricIdentity { val name = "processing-time" } - case object MailboxSize extends MetricIdentity { val name = "mailbox-size" } - case object TimeInMailbox extends MetricIdentity { val name = "time-in-mailbox" } - case object Errors extends MetricIdentity { val name = "errors" } - - case class ActorMetricsRecorder(processingTime: Histogram, timeInMailbox: Histogram, mailboxSize: MinMaxCounter, - errors: Counter) extends MetricGroupRecorder { - - def collect(context: CollectionContext): ActorMetricSnapshot = - ActorMetricSnapshot( - processingTime.collect(context), - timeInMailbox.collect(context), - mailboxSize.collect(context), - errors.collect(context)) - - def cleanup: Unit = { - processingTime.cleanup - mailboxSize.cleanup - timeInMailbox.cleanup - errors.cleanup - } - } - - case class ActorMetricSnapshot(processingTime: Histogram.Snapshot, timeInMailbox: Histogram.Snapshot, - mailboxSize: Histogram.Snapshot, errors: Counter.Snapshot) extends MetricGroupSnapshot { - - type GroupSnapshotType = ActorMetricSnapshot - - def merge(that: ActorMetricSnapshot, context: CollectionContext): ActorMetricSnapshot = - ActorMetricSnapshot( - processingTime.merge(that.processingTime, context), - timeInMailbox.merge(that.timeInMailbox, context), - mailboxSize.merge(that.mailboxSize, context), - errors.merge(that.errors, context)) - - lazy val metrics: Map[MetricIdentity, MetricSnapshot] = Map( - (ProcessingTime -> processingTime), - (MailboxSize -> mailboxSize), - (TimeInMailbox -> timeInMailbox), - (Errors -> errors)) - } - - val Factory = ActorMetricGroupFactory +/** + * Entity recorder for Akka Actors. The metrics being tracked are: + * + * - time-in-mailbox: Time spent from the instant when a message is enqueued in a actor's mailbox to the instant when + * that message is dequeued for processing. + * - processing-time: Time taken for the actor to process the receive function. + * - mailbox-size: Size of the actor's mailbox. + * - errors: Number or errors seen by the actor's supervision mechanism. + */ +class ActorMetrics(instrumentFactory: InstrumentFactory) extends GenericEntityRecorder(instrumentFactory) { + val timeInMailbox = histogram("time-in-mailbox", Time.Nanoseconds) + val processingTime = histogram("processing-time", Time.Nanoseconds) + val mailboxSize = minMaxCounter("mailbox-size") + val errors = counter("errors") } -case object ActorMetricGroupFactory extends MetricGroupFactory { - import kamon.akka.ActorMetrics._ - - type GroupRecorder = ActorMetricsRecorder - - def create(config: Config, system: ActorSystem): ActorMetricsRecorder = { - val settings = config.getConfig("precision.actor") - - val processingTimeConfig = settings.getConfig("processing-time") - val timeInMailboxConfig = settings.getConfig("time-in-mailbox") - val mailboxSizeConfig = settings.getConfig("mailbox-size") - - new ActorMetricsRecorder( - Histogram.fromConfig(processingTimeConfig), - Histogram.fromConfig(timeInMailboxConfig), - MinMaxCounter.fromConfig(mailboxSizeConfig, system), - Counter()) - } -} +object ActorMetrics extends EntityRecorderFactory[ActorMetrics] { + def category: String = "akka-actor" + def createRecorder(instrumentFactory: InstrumentFactory): ActorMetrics = new ActorMetrics(instrumentFactory) +}
\ No newline at end of file diff --git a/kamon-akka/src/main/scala/kamon/akka/AkkaExtension.scala b/kamon-akka/src/main/scala/kamon/akka/AkkaExtension.scala index bc013b63..cbca7db6 100644 --- a/kamon-akka/src/main/scala/kamon/akka/AkkaExtension.scala +++ b/kamon-akka/src/main/scala/kamon/akka/AkkaExtension.scala @@ -16,8 +16,8 @@ package kamon.akka -import akka.actor -import akka.actor._ +import _root_.akka.actor +import _root_.akka.actor._ import kamon._ class AkkaExtension(system: ExtendedActorSystem) extends Kamon.Extension { @@ -29,4 +29,6 @@ class AkkaExtension(system: ExtendedActorSystem) extends Kamon.Extension { object Akka extends ExtensionId[AkkaExtension] with ExtensionIdProvider { def lookup(): ExtensionId[_ <: actor.Extension] = Akka def createExtension(system: ExtendedActorSystem): AkkaExtension = new AkkaExtension(system) -}
\ No newline at end of file + +} + diff --git a/kamon-akka/src/main/scala/kamon/akka/DispatcherMetrics.scala b/kamon-akka/src/main/scala/kamon/akka/DispatcherMetrics.scala index 64e16f96..acf92e70 100644 --- a/kamon-akka/src/main/scala/kamon/akka/DispatcherMetrics.scala +++ b/kamon-akka/src/main/scala/kamon/akka/DispatcherMetrics.scala @@ -16,79 +16,71 @@ package kamon.akka -import akka.actor.ActorSystem -import com.typesafe.config.Config +import java.util.concurrent.ThreadPoolExecutor + +import _root_.akka.dispatch.ForkJoinExecutorConfigurator.AkkaForkJoinPool import kamon.metric._ -import kamon.metric.instrument.Histogram +import kamon.metric.instrument.{ DifferentialValueCollector, InstrumentFactory } -case class DispatcherMetrics(name: String) extends MetricGroupIdentity { - val category = DispatcherMetrics -} +class ForkJoinPoolDispatcherMetrics(fjp: AkkaForkJoinPool, instrumentFactory: InstrumentFactory) extends GenericEntityRecorder(instrumentFactory) { + val paralellism = minMaxCounter("parallelism") + paralellism.increment(fjp.getParallelism) // Steady value. -object DispatcherMetrics extends MetricGroupCategory { - val name = "dispatcher" + val poolSize = gauge("pool-size", () ⇒ { + fjp.getPoolSize.toLong + }) - case object MaximumPoolSize extends MetricIdentity { val name = "maximum-pool-size" } - case object RunningThreadCount extends MetricIdentity { val name = "running-thread-count" } - case object QueueTaskCount extends MetricIdentity { val name = "queued-task-count" } - case object PoolSize extends MetricIdentity { val name = "pool-size" } + val activeThreads = gauge("active-threads", () ⇒ { + fjp.getActiveThreadCount.toLong + }) - case class DispatcherMetricRecorder(maximumPoolSize: Histogram, runningThreadCount: Histogram, - queueTaskCount: Histogram, poolSize: Histogram) - extends MetricGroupRecorder { + val runningThreads = gauge("running-threads", () ⇒ { + fjp.getRunningThreadCount.toLong + }) - def collect(context: CollectionContext): MetricGroupSnapshot = - DispatcherMetricSnapshot( - maximumPoolSize.collect(context), - runningThreadCount.collect(context), - queueTaskCount.collect(context), - poolSize.collect(context)) + val queuedTaskCount = gauge("queued-task-count", () ⇒ { + fjp.getQueuedTaskCount + }) +} - def cleanup: Unit = {} +object ForkJoinPoolDispatcherMetrics { + def factory(fjp: AkkaForkJoinPool) = new EntityRecorderFactory[ForkJoinPoolDispatcherMetrics] { + def category: String = AkkaDispatcherMetrics.Category + def createRecorder(instrumentFactory: InstrumentFactory) = new ForkJoinPoolDispatcherMetrics(fjp, instrumentFactory) } +} - case class DispatcherMetricSnapshot(maximumPoolSize: Histogram.Snapshot, runningThreadCount: Histogram.Snapshot, - queueTaskCount: Histogram.Snapshot, poolSize: Histogram.Snapshot) extends MetricGroupSnapshot { +class ThreadPoolExecutorDispatcherMetrics(tpe: ThreadPoolExecutor, instrumentFactory: InstrumentFactory) extends GenericEntityRecorder(instrumentFactory) { + val corePoolSize = gauge("core-pool-size", () ⇒ { + tpe.getCorePoolSize.toLong + }) - type GroupSnapshotType = DispatcherMetricSnapshot + val maxPoolSize = gauge("max-pool-size", () ⇒ { + tpe.getMaximumPoolSize.toLong + }) - def merge(that: DispatcherMetricSnapshot, context: CollectionContext): DispatcherMetricSnapshot = - DispatcherMetricSnapshot( - maximumPoolSize.merge(that.maximumPoolSize, context), - runningThreadCount.merge(that.runningThreadCount, context), - queueTaskCount.merge(that.queueTaskCount, context), - poolSize.merge(that.poolSize, context)) + val poolSize = gauge("pool-size", () ⇒ { + tpe.getPoolSize.toLong + }) - lazy val metrics: Map[MetricIdentity, MetricSnapshot] = Map( - (MaximumPoolSize -> maximumPoolSize), - (RunningThreadCount -> runningThreadCount), - (QueueTaskCount -> queueTaskCount), - (PoolSize -> poolSize)) - } + val activeThreads = gauge("active-threads", () ⇒ { + tpe.getActiveCount.toLong + }) - val Factory = DispatcherMetricGroupFactory + val processedTasks = gauge("processed-tasks", DifferentialValueCollector(() ⇒ { + tpe.getTaskCount + })) } -case object DispatcherMetricGroupFactory extends MetricGroupFactory { +object ThreadPoolExecutorDispatcherMetrics { - import kamon.akka.DispatcherMetrics._ - - type GroupRecorder = DispatcherMetricRecorder - - def create(config: Config, system: ActorSystem): DispatcherMetricRecorder = { - val settings = config.getConfig("precision.dispatcher") - - val maximumPoolSizeConfig = settings.getConfig("maximum-pool-size") - val runningThreadCountConfig = settings.getConfig("running-thread-count") - val queueTaskCountConfig = settings.getConfig("queued-task-count") - val poolSizeConfig = settings.getConfig("pool-size") - - new DispatcherMetricRecorder( - Histogram.fromConfig(maximumPoolSizeConfig), - Histogram.fromConfig(runningThreadCountConfig), - Histogram.fromConfig(queueTaskCountConfig), - Histogram.fromConfig(poolSizeConfig)) + def factory(tpe: ThreadPoolExecutor) = new EntityRecorderFactory[ThreadPoolExecutorDispatcherMetrics] { + def category: String = AkkaDispatcherMetrics.Category + def createRecorder(instrumentFactory: InstrumentFactory) = new ThreadPoolExecutorDispatcherMetrics(tpe, instrumentFactory) } +} +object AkkaDispatcherMetrics { + val Category = "akka-dispatcher" } diff --git a/kamon-akka/src/main/scala/kamon/akka/RouterMetrics.scala b/kamon-akka/src/main/scala/kamon/akka/RouterMetrics.scala index 2eedf764..5c5bb05a 100644 --- a/kamon-akka/src/main/scala/kamon/akka/RouterMetrics.scala +++ b/kamon-akka/src/main/scala/kamon/akka/RouterMetrics.scala @@ -15,75 +15,26 @@ */ package kamon.akka -import akka.actor.ActorSystem -import com.typesafe.config.Config import kamon.metric._ -import kamon.metric.instrument.{ Counter, Histogram } +import kamon.metric.instrument.{ Time, InstrumentFactory } -case class RouterMetrics(name: String) extends MetricGroupIdentity { - val category = RouterMetrics -} - -object RouterMetrics extends MetricGroupCategory { - val name = "router" - - case object RoutingTime extends MetricIdentity { val name = "routing-time" } - case object ProcessingTime extends MetricIdentity { val name = "processing-time" } - case object TimeInMailbox extends MetricIdentity { val name = "time-in-mailbox" } - case object Errors extends MetricIdentity { val name = "errors" } - - case class RouterMetricsRecorder(routingTime: Histogram, processingTime: Histogram, timeInMailbox: Histogram, errors: Counter) extends MetricGroupRecorder { - - def collect(context: CollectionContext): RouterMetricSnapshot = - RouterMetricSnapshot(routingTime.collect(context), processingTime.collect(context), timeInMailbox.collect(context), errors.collect(context)) - - def cleanup: Unit = { - routingTime.cleanup - processingTime.cleanup - timeInMailbox.cleanup - errors.cleanup - } - } - - case class RouterMetricSnapshot(routingTime: Histogram.Snapshot, processingTime: Histogram.Snapshot, timeInMailbox: Histogram.Snapshot, errors: Counter.Snapshot) extends MetricGroupSnapshot { - - type GroupSnapshotType = RouterMetricSnapshot - - def merge(that: RouterMetricSnapshot, context: CollectionContext): RouterMetricSnapshot = - RouterMetricSnapshot( - routingTime.merge(that.routingTime, context), - processingTime.merge(that.processingTime, context), - timeInMailbox.merge(that.timeInMailbox, context), - errors.merge(that.errors, context)) - - lazy val metrics: Map[MetricIdentity, MetricSnapshot] = Map( - RoutingTime -> routingTime, - ProcessingTime -> processingTime, - TimeInMailbox -> timeInMailbox, - Errors -> errors) - } - - val Factory = RouterMetricGroupFactory -} - -case object RouterMetricGroupFactory extends MetricGroupFactory { - - import kamon.akka.RouterMetrics._ - - type GroupRecorder = RouterMetricsRecorder - - def create(config: Config, system: ActorSystem): RouterMetricsRecorder = { - val settings = config.getConfig("precision.router") - - val routingTimeConfig = settings.getConfig("routing-time") - val processingTimeConfig = settings.getConfig("processing-time") - val timeInMailboxConfig = settings.getConfig("time-in-mailbox") - - new RouterMetricsRecorder( - Histogram.fromConfig(routingTimeConfig), - Histogram.fromConfig(processingTimeConfig), - Histogram.fromConfig(timeInMailboxConfig), - Counter()) - } +/** + * Entity recorder for Akka Routers. The metrics being tracked are: + * + * - routing-time: Time taken for the router to process the routing logic. + * - time-in-mailbox: Time spent from the instant when a message is enqueued in a actor's mailbox to the instant when + * that message is dequeued for processing. + * - processing-time: Time taken for the actor to process the receive function. + * - errors: Number or errors seen by the actor's supervision mechanism. + */ +class RouterMetrics(instrumentFactory: InstrumentFactory) extends GenericEntityRecorder(instrumentFactory) { + val routingTime = histogram("routing-time", Time.Nanoseconds) + val timeInMailbox = histogram("time-in-mailbox", Time.Nanoseconds) + val processingTime = histogram("processing-time", Time.Nanoseconds) + val errors = counter("errors") } +object RouterMetrics extends EntityRecorderFactory[RouterMetrics] { + def category: String = "akka-router" + def createRecorder(instrumentFactory: InstrumentFactory): RouterMetrics = new RouterMetrics(instrumentFactory) +}
\ No newline at end of file diff --git a/kamon-akka/src/main/scala/kamon/instrumentation/akka/ActorCellInstrumentation.scala b/kamon-akka/src/main/scala/kamon/akka/instrumentation/ActorCellInstrumentation.scala index 78d88583..c961737d 100644 --- a/kamon-akka/src/main/scala/kamon/instrumentation/akka/ActorCellInstrumentation.scala +++ b/kamon-akka/src/main/scala/kamon/akka/instrumentation/ActorCellInstrumentation.scala @@ -19,11 +19,8 @@ package akka.kamon.instrumentation import akka.actor._ import akka.dispatch.{ Envelope, MessageDispatcher } import akka.routing.RoutedActorCell -import kamon.Kamon import kamon.akka.{ RouterMetrics, ActorMetrics } -import ActorMetrics.ActorMetricsRecorder -import RouterMetrics.RouterMetricsRecorder -import kamon.metric.Metrics +import kamon.metric.{ Metrics, Entity } import kamon.trace._ import org.aspectj.lang.ProceedingJoinPoint import org.aspectj.lang.annotation._ @@ -36,12 +33,13 @@ class ActorCellInstrumentation { @After("actorCellCreation(cell, system, ref, props, dispatcher, parent)") def afterCreation(cell: ActorCell, system: ActorSystem, ref: ActorRef, props: Props, dispatcher: MessageDispatcher, parent: ActorRef): Unit = { - val metricsExtension = Kamon(Metrics)(system) - val metricIdentity = ActorMetrics(ref.path.elements.mkString("/")) - val cellMetrics = cell.asInstanceOf[ActorCellMetrics] + Metrics.get(system).register(ActorMetrics, ref.path.elements.mkString("/")).map { registration ⇒ + val cellMetrics = cell.asInstanceOf[ActorCellMetrics] + + cellMetrics.entity = registration.entity + cellMetrics.recorder = Some(registration.recorder) + } - cellMetrics.actorMetricIdentity = metricIdentity - cellMetrics.actorMetricsRecorder = metricsExtension.register(metricIdentity, ActorMetrics.Factory) } @Pointcut("execution(* akka.actor.ActorCell.invoke(*)) && this(cell) && args(envelope)") @@ -54,11 +52,11 @@ class ActorCellInstrumentation { val contextAndTimestamp = envelope.asInstanceOf[TimestampedTraceContextAware] try { - TraceRecorder.withInlineTraceContextReplacement(contextAndTimestamp.traceContext) { + TraceContext.withContext(contextAndTimestamp.traceContext) { pjp.proceed() } } finally { - cellMetrics.actorMetricsRecorder.map { am ⇒ + cellMetrics.recorder.map { am ⇒ val processingTime = System.nanoTime() - timestampBeforeProcessing val timeInMailbox = timestampBeforeProcessing - contextAndTimestamp.captureNanoTime @@ -81,7 +79,7 @@ class ActorCellInstrumentation { @After("sendMessageInActorCell(cell, envelope)") def afterSendMessageInActorCell(cell: ActorCell, envelope: Envelope): Unit = { val cellMetrics = cell.asInstanceOf[ActorCellMetrics] - cellMetrics.actorMetricsRecorder.map(_.mailboxSize.increment()) + cellMetrics.recorder.map(_.mailboxSize.increment()) } @Pointcut("execution(* akka.actor.ActorCell.stop()) && this(cell)") @@ -90,15 +88,15 @@ class ActorCellInstrumentation { @After("actorStop(cell)") def afterStop(cell: ActorCell): Unit = { val cellMetrics = cell.asInstanceOf[ActorCellMetrics] - cellMetrics.actorMetricsRecorder.map { _ ⇒ - Kamon(Metrics)(cell.system).unregister(cellMetrics.actorMetricIdentity) + cellMetrics.recorder.map { _ ⇒ + Metrics.get(cell.system).unregister(cellMetrics.entity) } // The Stop can't be captured from the RoutedActorCell so we need to put this piece of cleanup here. if (cell.isInstanceOf[RoutedActorCell]) { val routedCellMetrics = cell.asInstanceOf[RoutedActorCellMetrics] - routedCellMetrics.routerMetricsRecorder.map { _ ⇒ - Kamon(Metrics)(cell.system).unregister(routedCellMetrics.routerMetricIdentity) + routedCellMetrics.routerRecorder.map { _ ⇒ + Metrics.get(cell.system).unregister(routedCellMetrics.routerEntity) } } } @@ -109,7 +107,7 @@ class ActorCellInstrumentation { @Before("actorInvokeFailure(cell)") def beforeInvokeFailure(cell: ActorCell): Unit = { val cellWithMetrics = cell.asInstanceOf[ActorCellMetrics] - cellWithMetrics.actorMetricsRecorder.map(_.errors.increment()) + cellWithMetrics.recorder.map(_.errors.increment()) // In case that this actor is behind a router, count the errors for the router as well. val envelope = cell.currentMessage.asInstanceOf[RouterAwareEnvelope] @@ -125,12 +123,12 @@ class RoutedActorCellInstrumentation { @After("routedActorCellCreation(cell, system, ref, props, dispatcher, routeeProps, supervisor)") def afterRoutedActorCellCreation(cell: RoutedActorCell, system: ActorSystem, ref: ActorRef, props: Props, dispatcher: MessageDispatcher, routeeProps: Props, supervisor: ActorRef): Unit = { - val metricsExtension = Kamon(Metrics)(system) - val metricIdentity = RouterMetrics(ref.path.elements.mkString("/")) - val cellMetrics = cell.asInstanceOf[RoutedActorCellMetrics] + Metrics.get(system).register(RouterMetrics, ref.path.elements.mkString("/")).map { registration ⇒ + val cellMetrics = cell.asInstanceOf[RoutedActorCellMetrics] - cellMetrics.routerMetricIdentity = metricIdentity - cellMetrics.routerMetricsRecorder = metricsExtension.register(metricIdentity, RouterMetrics.Factory) + cellMetrics.routerEntity = registration.entity + cellMetrics.routerRecorder = Some(registration.recorder) + } } @Pointcut("execution(* akka.routing.RoutedActorCell.sendMessage(*)) && this(cell) && args(envelope)") @@ -143,15 +141,15 @@ class RoutedActorCellInstrumentation { val contextAndTimestamp = envelope.asInstanceOf[TimestampedTraceContextAware] try { - TraceRecorder.withInlineTraceContextReplacement(contextAndTimestamp.traceContext) { + TraceContext.withContext(contextAndTimestamp.traceContext) { // The router metrics recorder will only be picked up if the message is sent from a tracked router. - RouterAwareEnvelope.dynamicRouterMetricsRecorder.withValue(cellMetrics.routerMetricsRecorder) { + RouterAwareEnvelope.dynamicRouterMetricsRecorder.withValue(cellMetrics.routerRecorder) { pjp.proceed() } } } finally { - cellMetrics.routerMetricsRecorder map { routerRecorder ⇒ + cellMetrics.routerRecorder map { routerRecorder ⇒ routerRecorder.routingTime.record(System.nanoTime() - timestampBeforeProcessing) } } @@ -159,25 +157,25 @@ class RoutedActorCellInstrumentation { } trait ActorCellMetrics { - var actorMetricIdentity: ActorMetrics = _ - var actorMetricsRecorder: Option[ActorMetricsRecorder] = _ + var entity: Entity = _ + var recorder: Option[ActorMetrics] = None } trait RoutedActorCellMetrics { - var routerMetricIdentity: RouterMetrics = _ - var routerMetricsRecorder: Option[RouterMetricsRecorder] = _ + var routerEntity: Entity = _ + var routerRecorder: Option[RouterMetrics] = None } trait RouterAwareEnvelope { - def routerMetricsRecorder: Option[RouterMetricsRecorder] + def routerMetricsRecorder: Option[RouterMetrics] } object RouterAwareEnvelope { import scala.util.DynamicVariable - private[kamon] val dynamicRouterMetricsRecorder = new DynamicVariable[Option[RouterMetricsRecorder]](None) + private[kamon] val dynamicRouterMetricsRecorder = new DynamicVariable[Option[RouterMetrics]](None) def default: RouterAwareEnvelope = new RouterAwareEnvelope { - val routerMetricsRecorder: Option[RouterMetricsRecorder] = dynamicRouterMetricsRecorder.value + val routerMetricsRecorder: Option[RouterMetrics] = dynamicRouterMetricsRecorder.value } } diff --git a/kamon-akka/src/main/scala/kamon/instrumentation/akka/ActorLoggingInstrumentation.scala b/kamon-akka/src/main/scala/kamon/akka/instrumentation/ActorLoggingInstrumentation.scala index e0e5d316..dd998c6b 100644 --- a/kamon-akka/src/main/scala/kamon/instrumentation/akka/ActorLoggingInstrumentation.scala +++ b/kamon-akka/src/main/scala/kamon/akka/instrumentation/ActorLoggingInstrumentation.scala @@ -17,7 +17,7 @@ package akka.kamon.instrumentation import kamon.trace.logging.MdcKeysSupport -import kamon.trace.{ TraceContextAware, TraceRecorder } +import kamon.trace.{ TraceContext, TraceContextAware } import org.aspectj.lang.ProceedingJoinPoint import org.aspectj.lang.annotation._ @@ -41,7 +41,7 @@ class ActorLoggingInstrumentation extends MdcKeysSupport { @Around("withMdcInvocation(logSource, logEvent, logStatement)") def aroundWithMdcInvocation(pjp: ProceedingJoinPoint, logSource: String, logEvent: TraceContextAware, logStatement: () ⇒ _): Unit = { - TraceRecorder.withInlineTraceContextReplacement(logEvent.traceContext) { + TraceContext.withContext(logEvent.traceContext) { withMdc { pjp.proceed() } diff --git a/kamon-akka/src/main/scala/kamon/instrumentation/akka/ActorSystemMessageInstrumentation.scala b/kamon-akka/src/main/scala/kamon/akka/instrumentation/ActorSystemMessageInstrumentation.scala index 48016876..0cb4ef13 100644 --- a/kamon-akka/src/main/scala/kamon/instrumentation/akka/ActorSystemMessageInstrumentation.scala +++ b/kamon-akka/src/main/scala/kamon/akka/instrumentation/ActorSystemMessageInstrumentation.scala @@ -17,7 +17,7 @@ package akka.kamon.instrumentation import akka.dispatch.sysmsg.EarliestFirstSystemMessageList -import kamon.trace.{ TraceContextAware, TraceRecorder } +import kamon.trace.{ TraceContext, TraceContextAware } import org.aspectj.lang.ProceedingJoinPoint import org.aspectj.lang.annotation._ @@ -31,7 +31,7 @@ class ActorSystemMessageInstrumentation { def aroundSystemMessageInvoke(pjp: ProceedingJoinPoint, messages: EarliestFirstSystemMessageList): Any = { if (messages.nonEmpty) { val ctx = messages.head.asInstanceOf[TraceContextAware].traceContext - TraceRecorder.withInlineTraceContextReplacement(ctx)(pjp.proceed()) + TraceContext.withContext(ctx)(pjp.proceed()) } else pjp.proceed() } @@ -73,7 +73,7 @@ class TraceContextIntoRepointableActorRefMixin { @Around("repointableActorRefCreation(repointableActorRef)") def afterRepointableActorRefCreation(pjp: ProceedingJoinPoint, repointableActorRef: TraceContextAware): Any = { - TraceRecorder.withInlineTraceContextReplacement(repointableActorRef.traceContext) { + TraceContext.withContext(repointableActorRef.traceContext) { pjp.proceed() } } diff --git a/kamon-akka/src/main/scala/kamon/instrumentation/akka/AskPatternInstrumentation.scala b/kamon-akka/src/main/scala/kamon/akka/instrumentation/AskPatternInstrumentation.scala index ebddbfc8..28bfcae9 100644 --- a/kamon-akka/src/main/scala/kamon/instrumentation/akka/AskPatternInstrumentation.scala +++ b/kamon-akka/src/main/scala/kamon/akka/instrumentation/AskPatternInstrumentation.scala @@ -16,40 +16,49 @@ package akka.kamon.instrumentation +import akka.util.Timeout import kamon.Kamon import kamon.akka.Akka -import kamon.trace.{ TraceRecorder, TraceContext, EmptyTraceContext, TraceContextAware } -import akka.actor.{ ActorSystem, ActorRef } +import kamon.trace.{ TraceContext, EmptyTraceContext, TraceContextAware } +import akka.actor.{ InternalActorRef, ActorSystem, ActorRef } import akka.event.Logging.Warning -import akka.pattern.AskTimeoutException +import akka.pattern.{ PromiseActorRef, AskTimeoutException } import org.aspectj.lang.ProceedingJoinPoint import org.aspectj.lang.annotation._ import org.aspectj.lang.reflect.SourceLocation import scala.concurrent.Future import scala.compat.Platform.EOL +import scala.concurrent.duration.FiniteDuration @Aspect class AskPatternInstrumentation { import AskPatternInstrumentation._ - @Pointcut("call(* akka.pattern.AskableActorRef$.$qmark$extension(..)) && args(actor, *, *)") - def askableActorRefAsk(actor: ActorRef): Unit = {} + @Pointcut("call(* akka.pattern.AskableActorRef$.$qmark$extension(..)) && args(actor, *, timeout)") + def askableActorRefAsk(actor: ActorRef, timeout: Timeout): Unit = {} - @Around("askableActorRefAsk(actor)") - def hookAskTimeoutWarning(pjp: ProceedingJoinPoint, actor: ActorRef): AnyRef = - TraceRecorder.withTraceContextAndSystem { (ctx, system) ⇒ - val akkaExtension = Kamon(Akka)(system) - val future = pjp.proceed().asInstanceOf[Future[AnyRef]] + @Around("askableActorRefAsk(actor, timeout)") + def hookAskTimeoutWarning(pjp: ProceedingJoinPoint, actor: ActorRef, timeout: Timeout): AnyRef = + TraceContext.map { ctx ⇒ + actor match { + // the AskPattern will only work for InternalActorRef's with these conditions. + case ref: InternalActorRef if !ref.isTerminated && timeout.duration.length > 0 ⇒ + val akkaExtension = ctx.lookupExtension(Akka) + val future = pjp.proceed().asInstanceOf[Future[AnyRef]] + val system = ref.provider.guardian.underlying.system - val handler = akkaExtension.askPatternTimeoutWarning match { - case "off" ⇒ None - case "lightweight" ⇒ Some(errorHandler(callInfo = Some(CallInfo(s"${actor.path.name} ?", pjp.getSourceLocation)))(system)) - case "heavyweight" ⇒ Some(errorHandler(stack = Some(new StackTraceCaptureException))(system)) - } + val handler = akkaExtension.askPatternTimeoutWarning match { + case "off" ⇒ None + case "lightweight" ⇒ Some(errorHandler(callInfo = Some(CallInfo(s"${actor.path.name} ?", pjp.getSourceLocation)))(system)) + case "heavyweight" ⇒ Some(errorHandler(stack = Some(new StackTraceCaptureException))(system)) + } + + handler.map(future.onFailure(_)(akkaExtension.dispatcher)) + future - handler.map(future.onFailure(_)(akkaExtension.dispatcher)) - future + case _ ⇒ pjp.proceed() // + } } getOrElse (pjp.proceed()) diff --git a/kamon-akka/src/main/scala/kamon/akka/instrumentation/DispatcherInstrumentation.scala b/kamon-akka/src/main/scala/kamon/akka/instrumentation/DispatcherInstrumentation.scala new file mode 100644 index 00000000..f4bc31c4 --- /dev/null +++ b/kamon-akka/src/main/scala/kamon/akka/instrumentation/DispatcherInstrumentation.scala @@ -0,0 +1,168 @@ +/* + * ========================================================================================= + * Copyright © 2013-2014 the kamon project <http://kamon.io/> + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file + * except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the + * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, + * either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + * ========================================================================================= + */ + +package akka.kamon.instrumentation + +import java.util.concurrent.{ ExecutorService, ThreadPoolExecutor } + +import akka.actor.{ ActorSystem, ActorSystemImpl } +import akka.dispatch.ForkJoinExecutorConfigurator.AkkaForkJoinPool +import akka.dispatch._ +import akka.kamon.instrumentation.LookupDataAware.LookupData +import kamon.akka.{ AkkaDispatcherMetrics, ThreadPoolExecutorDispatcherMetrics, ForkJoinPoolDispatcherMetrics } +import kamon.metric.{ Metrics, Entity } +import org.aspectj.lang.ProceedingJoinPoint +import org.aspectj.lang.annotation._ + +@Aspect +class DispatcherInstrumentation { + + @Pointcut("execution(* akka.actor.ActorSystemImpl.start(..)) && this(system)") + def actorSystemInitialization(system: ActorSystemImpl): Unit = {} + + @Before("actorSystemInitialization(system)") + def afterActorSystemInitialization(system: ActorSystemImpl): Unit = { + system.dispatchers.asInstanceOf[ActorSystemAware].actorSystem = system + + // The default dispatcher for the actor system is looked up in the ActorSystemImpl's initialization code and we + // can't get the Metrics extension there since the ActorSystem is not yet fully constructed. To workaround that + // we are manually selecting and registering the default dispatcher with the Metrics extension. All other dispatchers + // will by registered by the instrumentation bellow. + + // Yes, reflection sucks, but this piece of code is only executed once on ActorSystem's startup. + val defaultDispatcher = system.dispatcher + val executorServiceDelegateField = defaultDispatcher.getClass.getDeclaredField("executorServiceDelegate") + executorServiceDelegateField.setAccessible(true) + + val lazyExecutorServiceDelegate = executorServiceDelegateField.get(defaultDispatcher) + val executorField = lazyExecutorServiceDelegate.getClass.getMethod("executor") + executorField.setAccessible(true) + + val defaultDispatcherExecutor = executorField.invoke(lazyExecutorServiceDelegate).asInstanceOf[ExecutorService] + registerDispatcher(Dispatchers.DefaultDispatcherId, defaultDispatcherExecutor, system) + } + + private def registerDispatcher(dispatcherName: String, executorService: ExecutorService, system: ActorSystem): Unit = + executorService match { + case fjp: AkkaForkJoinPool ⇒ + Metrics.get(system).register(ForkJoinPoolDispatcherMetrics.factory(fjp), dispatcherName) + + case tpe: ThreadPoolExecutor ⇒ + Metrics.get(system).register(ThreadPoolExecutorDispatcherMetrics.factory(tpe), dispatcherName) + + case others ⇒ // Currently not interested in other kinds of dispatchers. + } + + @Pointcut("execution(* akka.dispatch.Dispatchers.lookup(..)) && this(dispatchers) && args(dispatcherName)") + def dispatchersLookup(dispatchers: ActorSystemAware, dispatcherName: String) = {} + + @Around("dispatchersLookup(dispatchers, dispatcherName)") + def aroundDispatchersLookup(pjp: ProceedingJoinPoint, dispatchers: ActorSystemAware, dispatcherName: String): Any = + LookupDataAware.withLookupData(LookupData(dispatcherName, dispatchers.actorSystem)) { + pjp.proceed() + } + + @Pointcut("initialization(akka.dispatch.ExecutorServiceFactory.new(..)) && target(factory)") + def executorServiceFactoryInitialization(factory: LookupDataAware): Unit = {} + + @After("executorServiceFactoryInitialization(factory)") + def afterExecutorServiceFactoryInitialization(factory: LookupDataAware): Unit = + factory.lookupData = LookupDataAware.currentLookupData + + @Pointcut("execution(* akka.dispatch.ExecutorServiceFactory+.createExecutorService()) && this(factory) && !cflow(execution(* akka.dispatch.Dispatcher.shutdown()))") + def createExecutorService(factory: LookupDataAware): Unit = {} + + @AfterReturning(pointcut = "createExecutorService(factory)", returning = "executorService") + def afterCreateExecutorService(factory: LookupDataAware, executorService: ExecutorService): Unit = { + val lookupData = factory.lookupData + + // lookupData.actorSystem will be null only during the first lookup of the default dispatcher during the + // ActorSystemImpl's initialization. + if (lookupData.actorSystem != null) + registerDispatcher(lookupData.dispatcherName, executorService, lookupData.actorSystem) + } + + @Pointcut("initialization(akka.dispatch.Dispatcher.LazyExecutorServiceDelegate.new(..)) && this(lazyExecutor)") + def lazyExecutorInitialization(lazyExecutor: LookupDataAware): Unit = {} + + @After("lazyExecutorInitialization(lazyExecutor)") + def afterLazyExecutorInitialization(lazyExecutor: LookupDataAware): Unit = + lazyExecutor.lookupData = LookupDataAware.currentLookupData + + @Pointcut("execution(* akka.dispatch.Dispatcher.LazyExecutorServiceDelegate.copy()) && this(lazyExecutor)") + def lazyExecutorCopy(lazyExecutor: LookupDataAware): Unit = {} + + @Around("lazyExecutorCopy(lazyExecutor)") + def aroundLazyExecutorCopy(pjp: ProceedingJoinPoint, lazyExecutor: LookupDataAware): Any = + LookupDataAware.withLookupData(lazyExecutor.lookupData) { + pjp.proceed() + } + + @Pointcut("execution(* akka.dispatch.Dispatcher.LazyExecutorServiceDelegate.shutdown()) && this(lazyExecutor)") + def lazyExecutorShutdown(lazyExecutor: LookupDataAware): Unit = {} + + @After("lazyExecutorShutdown(lazyExecutor)") + def afterLazyExecutorShutdown(lazyExecutor: LookupDataAware): Unit = { + import lazyExecutor.lookupData + + if (lookupData.actorSystem != null) + Metrics.get(lookupData.actorSystem).unregister(Entity(lookupData.dispatcherName, AkkaDispatcherMetrics.Category)) + } + +} + +@Aspect +class DispatcherMetricCollectionInfoIntoDispatcherMixin { + + @DeclareMixin("akka.dispatch.Dispatchers") + def mixinActorSystemAwareToDispatchers: ActorSystemAware = ActorSystemAware() + + @DeclareMixin("akka.dispatch.Dispatcher.LazyExecutorServiceDelegate") + def mixinLookupDataAwareToExecutors: LookupDataAware = LookupDataAware() + + @DeclareMixin("akka.dispatch.ExecutorServiceFactory+") + def mixinActorSystemAwareToDispatcher: LookupDataAware = LookupDataAware() +} + +trait ActorSystemAware { + @volatile var actorSystem: ActorSystem = _ +} + +object ActorSystemAware { + def apply(): ActorSystemAware = new ActorSystemAware {} +} + +trait LookupDataAware { + @volatile var lookupData: LookupData = _ +} + +object LookupDataAware { + case class LookupData(dispatcherName: String, actorSystem: ActorSystem) + + private val _currentDispatcherLookupData = new ThreadLocal[LookupData] + + def apply() = new LookupDataAware {} + + def currentLookupData: LookupData = _currentDispatcherLookupData.get() + + def withLookupData[T](lookupData: LookupData)(thunk: ⇒ T): T = { + _currentDispatcherLookupData.set(lookupData) + val result = thunk + _currentDispatcherLookupData.remove() + + result + } +}
\ No newline at end of file diff --git a/kamon-akka/src/main/scala/kamon/instrumentation/akka/DispatcherInstrumentation.scala b/kamon-akka/src/main/scala/kamon/instrumentation/akka/DispatcherInstrumentation.scala deleted file mode 100644 index 8280edca..00000000 --- a/kamon-akka/src/main/scala/kamon/instrumentation/akka/DispatcherInstrumentation.scala +++ /dev/null @@ -1,164 +0,0 @@ -/* - * ========================================================================================= - * Copyright © 2013-2014 the kamon project <http://kamon.io/> - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file - * except in compliance with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the - * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, - * either express or implied. See the License for the specific language governing permissions - * and limitations under the License. - * ========================================================================================= - */ - -package akka.kamon.instrumentation - -import java.lang.reflect.Method -import java.util.concurrent.ThreadPoolExecutor - -import akka.actor.{ ActorSystemImpl, Cancellable } -import akka.dispatch.{ Dispatcher, Dispatchers, ExecutorServiceDelegate, MessageDispatcher } -import akka.kamon.instrumentation.DispatcherMetricsCollector.DispatcherMetricsMeasurement -import kamon.Kamon -import kamon.akka.DispatcherMetrics -import DispatcherMetrics.DispatcherMetricRecorder -import kamon.metric.Metrics -import org.aspectj.lang.annotation._ - -import scala.concurrent.forkjoin.ForkJoinPool - -@Aspect -class DispatcherInstrumentation { - - @Pointcut("execution(akka.dispatch.Dispatchers.new(..)) && this(dispatchers) && cflow(execution(akka.actor.ActorSystemImpl.new(..)) && this(system))") - def onActorSystemStartup(dispatchers: Dispatchers, system: ActorSystemImpl) = {} - - @Before("onActorSystemStartup(dispatchers, system)") - def beforeActorSystemStartup(dispatchers: Dispatchers, system: ActorSystemImpl): Unit = { - val currentDispatchers = dispatchers.asInstanceOf[DispatchersWithActorSystem] - currentDispatchers.actorSystem = system - } - - @Pointcut("execution(* akka.dispatch.Dispatchers.lookup(..)) && this(dispatchers)") - def onDispatchersLookup(dispatchers: Dispatchers) = {} - - @AfterReturning(pointcut = "onDispatchersLookup(dispatchers)", returning = "dispatcher") - def afterReturningLookup(dispatchers: Dispatchers, dispatcher: Dispatcher): Unit = { - val dispatchersWithActorSystem = dispatchers.asInstanceOf[DispatchersWithActorSystem] - val dispatcherWithMetrics = dispatcher.asInstanceOf[DispatcherMetricCollectionInfo] - - dispatcherWithMetrics.actorSystem = dispatchersWithActorSystem.actorSystem - } - - @Pointcut("call(* akka.dispatch.ExecutorServiceFactory.createExecutorService(..))") - def onCreateExecutorService(): Unit = {} - - @Pointcut("cflow((execution(* akka.dispatch.MessageDispatcher.registerForExecution(..)) || execution(* akka.dispatch.MessageDispatcher.executeTask(..))) && this(dispatcher))") - def onCflowMessageDispatcher(dispatcher: Dispatcher): Unit = {} - - @Pointcut("onCreateExecutorService() && onCflowMessageDispatcher(dispatcher)") - def onDispatcherStartup(dispatcher: Dispatcher): Unit = {} - - @After("onDispatcherStartup(dispatcher)") - def afterDispatcherStartup(dispatcher: MessageDispatcher): Unit = { - - val dispatcherWithMetrics = dispatcher.asInstanceOf[DispatcherMetricCollectionInfo] - val metricsExtension = Kamon(Metrics)(dispatcherWithMetrics.actorSystem) - val metricIdentity = DispatcherMetrics(dispatcher.id) - - dispatcherWithMetrics.metricIdentity = metricIdentity - dispatcherWithMetrics.dispatcherMetricsRecorder = metricsExtension.register(metricIdentity, DispatcherMetrics.Factory) - - if (dispatcherWithMetrics.dispatcherMetricsRecorder.isDefined) { - dispatcherWithMetrics.dispatcherCollectorCancellable = metricsExtension.scheduleGaugeRecorder { - dispatcherWithMetrics.dispatcherMetricsRecorder.map { - dm ⇒ - val DispatcherMetricsMeasurement(maximumPoolSize, runningThreadCount, queueTaskCount, poolSize) = - DispatcherMetricsCollector.collect(dispatcher) - - dm.maximumPoolSize.record(maximumPoolSize) - dm.runningThreadCount.record(runningThreadCount) - dm.queueTaskCount.record(queueTaskCount) - dm.poolSize.record(poolSize) - } - } - } - } - - @Pointcut("execution(* akka.dispatch.MessageDispatcher.shutdown(..)) && this(dispatcher)") - def onDispatcherShutdown(dispatcher: MessageDispatcher): Unit = {} - - @After("onDispatcherShutdown(dispatcher)") - def afterDispatcherShutdown(dispatcher: MessageDispatcher): Unit = { - val dispatcherWithMetrics = dispatcher.asInstanceOf[DispatcherMetricCollectionInfo] - - dispatcherWithMetrics.dispatcherMetricsRecorder.map { - dispatcher ⇒ - dispatcherWithMetrics.dispatcherCollectorCancellable.cancel() - Kamon(Metrics)(dispatcherWithMetrics.actorSystem).unregister(dispatcherWithMetrics.metricIdentity) - } - } -} - -@Aspect -class DispatcherMetricCollectionInfoIntoDispatcherMixin { - - @DeclareMixin("akka.dispatch.MessageDispatcher") - def mixinDispatcherMetricsToMessageDispatcher: DispatcherMetricCollectionInfo = new DispatcherMetricCollectionInfo {} - - @DeclareMixin("akka.dispatch.Dispatchers") - def mixinDispatchersToDispatchersWithActorSystem: DispatchersWithActorSystem = new DispatchersWithActorSystem {} -} - -trait DispatcherMetricCollectionInfo { - var metricIdentity: DispatcherMetrics = _ - var dispatcherMetricsRecorder: Option[DispatcherMetricRecorder] = _ - var dispatcherCollectorCancellable: Cancellable = _ - var actorSystem: ActorSystemImpl = _ -} - -trait DispatchersWithActorSystem { - var actorSystem: ActorSystemImpl = _ -} - -object DispatcherMetricsCollector { - - case class DispatcherMetricsMeasurement(maximumPoolSize: Long, runningThreadCount: Long, queueTaskCount: Long, poolSize: Long) - - private def collectForkJoinMetrics(pool: ForkJoinPool): DispatcherMetricsMeasurement = { - DispatcherMetricsMeasurement(pool.getParallelism, pool.getActiveThreadCount, - (pool.getQueuedTaskCount + pool.getQueuedSubmissionCount), pool.getPoolSize) - } - - private def collectExecutorMetrics(pool: ThreadPoolExecutor): DispatcherMetricsMeasurement = { - DispatcherMetricsMeasurement(pool.getMaximumPoolSize, pool.getActiveCount, pool.getQueue.size(), pool.getPoolSize) - } - - private val executorServiceMethod: Method = { - // executorService is protected - val method = classOf[Dispatcher].getDeclaredMethod("executorService") - method.setAccessible(true) - method - } - - def collect(dispatcher: MessageDispatcher): DispatcherMetricsMeasurement = { - dispatcher match { - case x: Dispatcher ⇒ { - val executor = executorServiceMethod.invoke(x) match { - case delegate: ExecutorServiceDelegate ⇒ delegate.executor - case other ⇒ other - } - - executor match { - case fjp: ForkJoinPool ⇒ collectForkJoinMetrics(fjp) - case tpe: ThreadPoolExecutor ⇒ collectExecutorMetrics(tpe) - case anything ⇒ DispatcherMetricsMeasurement(0L, 0L, 0L, 0L) - } - } - case _ ⇒ new DispatcherMetricsMeasurement(0L, 0L, 0L, 0L) - } - } -} diff --git a/kamon-akka/src/test/resources/logback.xml b/kamon-akka/src/test/resources/logback.xml deleted file mode 100644 index 10c9aa35..00000000 --- a/kamon-akka/src/test/resources/logback.xml +++ /dev/null @@ -1,14 +0,0 @@ -<configuration scan="true"> - <conversionRule conversionWord="traceToken" converterClass="kamon.trace.logging.LogbackTraceTokenConverter"/> - - <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"> - <encoder> - <pattern>%date{HH:mm:ss.SSS} %-5level [%traceToken][%thread] %logger{55} - %msg%n</pattern> - </encoder> - </appender> - - <root level="error"> - <appender-ref ref="STDOUT" /> - </root> - -</configuration> diff --git a/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorCellInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorCellInstrumentationSpec.scala index 06a232bd..8f7ae613 100644 --- a/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorCellInstrumentationSpec.scala +++ b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorCellInstrumentationSpec.scala @@ -15,35 +15,32 @@ * ========================================================== */ package kamon.instrumentation.akka -import akka.actor.{ Actor, ActorSystem, Props } +import akka.actor.{ Actor, Props } import akka.pattern.{ ask, pipe } import akka.routing._ -import akka.testkit.{ TestKitBase, ImplicitSender, TestKit } import akka.util.Timeout -import com.typesafe.config.ConfigFactory -import kamon.trace.TraceRecorder -import org.scalatest.{ BeforeAndAfterAll, WordSpecLike } +import kamon.testkit.BaseKamonSpec +import kamon.trace.TraceContext import scala.concurrent.duration._ -class ActorCellInstrumentationSpec extends TestKitBase with WordSpecLike with ImplicitSender with BeforeAndAfterAll { - implicit lazy val system: ActorSystem = ActorSystem("actor-cell-instrumentation-spec") - implicit val executionContext = system.dispatcher +class ActorCellInstrumentationSpec extends BaseKamonSpec("actor-cell-instrumentation-spec") { + implicit lazy val executionContext = system.dispatcher "the message passing instrumentation" should { "propagate the TraceContext using bang" in new EchoActorFixture { - val testTraceContext = TraceRecorder.withNewTraceContext("bang-reply") { + val testTraceContext = TraceContext.withContext(newContext("bang-reply")) { ctxEchoActor ! "test" - TraceRecorder.currentContext + TraceContext.currentContext } expectMsg(testTraceContext) } "propagate the TraceContext using tell" in new EchoActorFixture { - val testTraceContext = TraceRecorder.withNewTraceContext("tell-reply") { + val testTraceContext = TraceContext.withContext(newContext("tell-reply")) { ctxEchoActor.tell("test", testActor) - TraceRecorder.currentContext + TraceContext.currentContext } expectMsg(testTraceContext) @@ -51,37 +48,37 @@ class ActorCellInstrumentationSpec extends TestKitBase with WordSpecLike with Im "propagate the TraceContext using ask" in new EchoActorFixture { implicit val timeout = Timeout(1 seconds) - val testTraceContext = TraceRecorder.withNewTraceContext("ask-reply") { + val testTraceContext = TraceContext.withContext(newContext("ask-reply")) { // The pipe pattern use Futures internally, so FutureTracing test should cover the underpinnings of it. (ctxEchoActor ? "test") pipeTo (testActor) - TraceRecorder.currentContext + TraceContext.currentContext } expectMsg(testTraceContext) } "propagate the TraceContext to actors behind a simple router" in new EchoSimpleRouterFixture { - val testTraceContext = TraceRecorder.withNewTraceContext("router-reply") { + val testTraceContext = TraceContext.withContext(newContext("router-reply")) { router.route("test", testActor) - TraceRecorder.currentContext + TraceContext.currentContext } expectMsg(testTraceContext) } "propagate the TraceContext to actors behind a pool router" in new EchoPoolRouterFixture { - val testTraceContext = TraceRecorder.withNewTraceContext("router-reply") { + val testTraceContext = TraceContext.withContext(newContext("router-reply")) { pool ! "test" - TraceRecorder.currentContext + TraceContext.currentContext } expectMsg(testTraceContext) } "propagate the TraceContext to actors behind a group router" in new EchoGroupRouterFixture { - val testTraceContext = TraceRecorder.withNewTraceContext("router-reply") { + val testTraceContext = TraceContext.withContext(newContext("router-reply")) { group ! "test" - TraceRecorder.currentContext + TraceContext.currentContext } expectMsg(testTraceContext) @@ -119,7 +116,7 @@ class ActorCellInstrumentationSpec extends TestKitBase with WordSpecLike with Im class TraceContextEcho extends Actor { def receive = { - case msg: String ⇒ sender ! TraceRecorder.currentContext + case msg: String ⇒ sender ! TraceContext.currentContext } } diff --git a/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorLoggingInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorLoggingInstrumentationSpec.scala index 598e9327..21706af9 100644 --- a/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorLoggingInstrumentationSpec.scala +++ b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorLoggingInstrumentationSpec.scala @@ -15,28 +15,33 @@ * ========================================================== */ package kamon.instrumentation.akka -import akka.actor.{ Actor, ActorLogging, ActorSystem, Props } +import akka.actor.{ Actor, ActorLogging, Props } import akka.event.Logging.LogEvent -import akka.testkit.TestKitBase import com.typesafe.config.ConfigFactory +import kamon.testkit.BaseKamonSpec import kamon.trace.TraceLocal.AvailableToMdc import kamon.trace.logging.MdcKeysSupport -import kamon.trace.{ TraceContextAware, TraceLocal, TraceRecorder } -import org.scalatest.{ BeforeAndAfterAll, Inspectors, Matchers, WordSpecLike } +import kamon.trace.{ TraceContextAware, TraceLocal, TraceContext } +import org.scalatest.Inspectors import org.slf4j.MDC -class ActorLoggingInstrumentationSpec extends TestKitBase with WordSpecLike with Matchers with Inspectors with MdcKeysSupport with BeforeAndAfterAll { - implicit lazy val system: ActorSystem = ActorSystem("actor-logging-instrumentation-spec", - ConfigFactory.parseString("""akka.loggers = ["akka.event.slf4j.Slf4jLogger"]""")) +class ActorLoggingInstrumentationSpec extends BaseKamonSpec("actor-logging-instrumentation-spec") with Inspectors with MdcKeysSupport { + override lazy val config = + ConfigFactory.parseString( + """ + |akka { + | loggers = ["akka.event.slf4j.Slf4jLogger"] + |} + """.stripMargin) "the ActorLogging instrumentation" should { "attach the TraceContext (if available) to log events" in { val loggerActor = system.actorOf(Props[LoggerActor]) system.eventStream.subscribe(testActor, classOf[LogEvent]) - val testTraceContext = TraceRecorder.withNewTraceContext("logging") { + val testTraceContext = TraceContext.withContext(newContext("logging")) { loggerActor ! "info" - TraceRecorder.currentContext + TraceContext.currentContext } fishForMessage() { @@ -50,7 +55,7 @@ class ActorLoggingInstrumentationSpec extends TestKitBase with WordSpecLike with "allow retrieve a value from the MDC when was created a key of type AvailableToMdc" in { val testString = "Hello World" - TraceRecorder.withNewTraceContext("logging-with-mdc") { + TraceContext.withContext(newContext("logging-with-mdc")) { TraceLocal.store(AvailableToMdc("some-cool-key"))(testString) withMdc { @@ -66,6 +71,6 @@ class ActorLoggingInstrumentationSpec extends TestKitBase with WordSpecLike with class LoggerActor extends Actor with ActorLogging { def receive = { - case "info" ⇒ log.info("TraceContext(name = {}, token = {})", TraceRecorder.currentContext.name, TraceRecorder.currentContext.token) + case "info" ⇒ log.info("TraceContext(name = {}, token = {})", TraceContext.currentContext.name, TraceContext.currentContext.token) } } diff --git a/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorSystemMessageInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorSystemMessageInstrumentationSpec.scala index 0e9025af..8c1033ae 100644 --- a/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorSystemMessageInstrumentationSpec.scala +++ b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorSystemMessageInstrumentationSpec.scala @@ -2,49 +2,53 @@ package kamon.instrumentation.akka import akka.actor.SupervisorStrategy.{ Escalate, Restart, Resume, Stop } import akka.actor._ -import akka.testkit.{ TestKitBase, ImplicitSender } +import akka.testkit.ImplicitSender import com.typesafe.config.ConfigFactory -import kamon.trace.{ EmptyTraceContext, TraceRecorder } +import kamon.testkit.BaseKamonSpec +import kamon.trace.{ EmptyTraceContext, TraceContext } import org.scalatest.WordSpecLike import scala.concurrent.duration._ import scala.util.control.NonFatal -class ActorSystemMessageInstrumentationSpec extends TestKitBase with WordSpecLike with ImplicitSender { - implicit lazy val system: ActorSystem = ActorSystem("actor-system-message-instrumentation-spec", ConfigFactory.parseString( - """ - |akka.loglevel = OFF - """.stripMargin)) +class ActorSystemMessageInstrumentationSpec extends BaseKamonSpec("actor-system-message-instrumentation-spec") with WordSpecLike with ImplicitSender { + override lazy val config = + ConfigFactory.parseString( + """ + |akka { + | loglevel = OFF + |} + """.stripMargin) - implicit val executionContext = system.dispatcher + implicit lazy val executionContext = system.dispatcher "the system message passing instrumentation" should { "keep the TraceContext while processing the Create message in top level actors" in { - val testTraceContext = TraceRecorder.withNewTraceContext("creating-top-level-actor") { + val testTraceContext = TraceContext.withContext(newContext("creating-top-level-actor")) { system.actorOf(Props(new Actor { - testActor ! TraceRecorder.currentContext + testActor ! TraceContext.currentContext def receive: Actor.Receive = { case any ⇒ } })) - TraceRecorder.currentContext + TraceContext.currentContext } expectMsg(testTraceContext) } "keep the TraceContext while processing the Create message in non top level actors" in { - val testTraceContext = TraceRecorder.withNewTraceContext("creating-non-top-level-actor") { + val testTraceContext = TraceContext.withContext(newContext("creating-non-top-level-actor")) { system.actorOf(Props(new Actor { def receive: Actor.Receive = { case any ⇒ context.actorOf(Props(new Actor { - testActor ! TraceRecorder.currentContext + testActor ! TraceContext.currentContext def receive: Actor.Receive = { case any ⇒ } })) } })) ! "any" - TraceRecorder.currentContext + TraceContext.currentContext } expectMsg(testTraceContext) @@ -54,9 +58,9 @@ class ActorSystemMessageInstrumentationSpec extends TestKitBase with WordSpecLik "the actor is resumed" in { val supervisor = supervisorWithDirective(Resume) - val testTraceContext = TraceRecorder.withNewTraceContext("fail-and-resume") { + val testTraceContext = TraceContext.withContext(newContext("fail-and-resume")) { supervisor ! "fail" - TraceRecorder.currentContext + TraceContext.currentContext } expectMsg(testTraceContext) // From the parent executing the supervision strategy @@ -69,9 +73,9 @@ class ActorSystemMessageInstrumentationSpec extends TestKitBase with WordSpecLik "the actor is restarted" in { val supervisor = supervisorWithDirective(Restart, sendPreRestart = true, sendPostRestart = true) - val testTraceContext = TraceRecorder.withNewTraceContext("fail-and-restart") { + val testTraceContext = TraceContext.withContext(newContext("fail-and-restart")) { supervisor ! "fail" - TraceRecorder.currentContext + TraceContext.currentContext } expectMsg(testTraceContext) // From the parent executing the supervision strategy @@ -86,9 +90,9 @@ class ActorSystemMessageInstrumentationSpec extends TestKitBase with WordSpecLik "the actor is stopped" in { val supervisor = supervisorWithDirective(Stop, sendPostStop = true) - val testTraceContext = TraceRecorder.withNewTraceContext("fail-and-stop") { + val testTraceContext = TraceContext.withContext(newContext("fail-and-stop")) { supervisor ! "fail" - TraceRecorder.currentContext + TraceContext.currentContext } expectMsg(testTraceContext) // From the parent executing the supervision strategy @@ -99,9 +103,9 @@ class ActorSystemMessageInstrumentationSpec extends TestKitBase with WordSpecLik "the failure is escalated" in { val supervisor = supervisorWithDirective(Escalate, sendPostStop = true) - val testTraceContext = TraceRecorder.withNewTraceContext("fail-and-escalate") { + val testTraceContext = TraceContext.withContext(newContext("fail-and-escalate")) { supervisor ! "fail" - TraceRecorder.currentContext + TraceContext.currentContext } expectMsg(testTraceContext) // From the parent executing the supervision strategy @@ -119,7 +123,7 @@ class ActorSystemMessageInstrumentationSpec extends TestKitBase with WordSpecLik val child = context.actorOf(Props(new Parent)) override def supervisorStrategy: SupervisorStrategy = OneForOneStrategy() { - case NonFatal(throwable) ⇒ testActor ! TraceRecorder.currentContext; Stop + case NonFatal(throwable) ⇒ testActor ! TraceContext.currentContext; Stop } def receive = { @@ -131,7 +135,7 @@ class ActorSystemMessageInstrumentationSpec extends TestKitBase with WordSpecLik val child = context.actorOf(Props(new Child)) override def supervisorStrategy: SupervisorStrategy = OneForOneStrategy() { - case NonFatal(throwable) ⇒ testActor ! TraceRecorder.currentContext; directive + case NonFatal(throwable) ⇒ testActor ! TraceContext.currentContext; directive } def receive: Actor.Receive = { @@ -139,7 +143,7 @@ class ActorSystemMessageInstrumentationSpec extends TestKitBase with WordSpecLik } override def postStop(): Unit = { - if (sendPostStop) testActor ! TraceRecorder.currentContext + if (sendPostStop) testActor ! TraceContext.currentContext super.postStop() } } @@ -147,26 +151,26 @@ class ActorSystemMessageInstrumentationSpec extends TestKitBase with WordSpecLik class Child extends Actor { def receive = { case "fail" ⇒ throw new ArithmeticException("Division by zero.") - case "context" ⇒ sender ! TraceRecorder.currentContext + case "context" ⇒ sender ! TraceContext.currentContext } override def preRestart(reason: Throwable, message: Option[Any]): Unit = { - if (sendPreRestart) testActor ! TraceRecorder.currentContext + if (sendPreRestart) testActor ! TraceContext.currentContext super.preRestart(reason, message) } override def postRestart(reason: Throwable): Unit = { - if (sendPostRestart) testActor ! TraceRecorder.currentContext + if (sendPostRestart) testActor ! TraceContext.currentContext super.postRestart(reason) } override def postStop(): Unit = { - if (sendPostStop) testActor ! TraceRecorder.currentContext + if (sendPostStop) testActor ! TraceContext.currentContext super.postStop() } override def preStart(): Unit = { - if (sendPreStart) testActor ! TraceRecorder.currentContext + if (sendPreStart) testActor ! TraceContext.currentContext super.preStart() } } diff --git a/kamon-akka/src/test/scala/kamon/instrumentation/akka/AskPatternInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/akka/instrumentation/AskPatternInstrumentationSpec.scala index 5c9905ba..0d63a19e 100644 --- a/kamon-akka/src/test/scala/kamon/instrumentation/akka/AskPatternInstrumentationSpec.scala +++ b/kamon-akka/src/test/scala/kamon/akka/instrumentation/AskPatternInstrumentationSpec.scala @@ -21,21 +21,26 @@ import java.util.concurrent.atomic.AtomicInteger import akka.actor._ import akka.event.Logging.Warning import akka.pattern.ask -import akka.testkit.{ TestProbe, TestKitBase } +import akka.testkit.TestProbe import akka.util.Timeout import com.typesafe.config.ConfigFactory import kamon.Kamon import kamon.akka.Akka -import kamon.trace.{ TraceContext, TraceContextAware, TraceRecorder } -import org.scalatest.{ BeforeAndAfterAll, Matchers, WordSpecLike } +import kamon.testkit.BaseKamonSpec +import kamon.trace.{ TraceContext, TraceContextAware } import scala.concurrent.duration._ -class AskPatternInstrumentationSpec extends TestKitBase with WordSpecLike with Matchers with BeforeAndAfterAll { - implicit lazy val system: ActorSystem = ActorSystem("ask-pattern-tracing-spec", - ConfigFactory.parseString("""akka.loggers = ["akka.event.slf4j.Slf4jLogger"]""")) +class AskPatternInstrumentationSpec extends BaseKamonSpec("ask-pattern-tracing-spec") { + override lazy val config = + ConfigFactory.parseString( + """ + |akka { + | loglevel = OFF + |} + """.stripMargin) - implicit val ec = system.dispatcher + implicit lazy val ec = system.dispatcher implicit val askTimeout = Timeout(10 millis) // TODO: Make this work with ActorSelections @@ -46,9 +51,9 @@ class AskPatternInstrumentationSpec extends TestKitBase with WordSpecLike with M setAskPatternTimeoutWarningMode("heavyweight") expectTimeoutWarning() { - TraceRecorder.withNewTraceContext("ask-timeout-warning") { + TraceContext.withContext(newContext("ask-timeout-warning")) { noReplyActorRef ? "hello" - TraceRecorder.currentContext + TraceContext.currentContext } } } @@ -59,9 +64,9 @@ class AskPatternInstrumentationSpec extends TestKitBase with WordSpecLike with M setAskPatternTimeoutWarningMode("lightweight") expectTimeoutWarning(messageSizeLimit = Some(1)) { - TraceRecorder.withNewTraceContext("ask-timeout-warning") { + TraceContext.withContext(newContext("ask-timeout-warning")) { noReplyActorRef ? "hello" - TraceRecorder.currentContext + TraceContext.currentContext } } } @@ -72,9 +77,9 @@ class AskPatternInstrumentationSpec extends TestKitBase with WordSpecLike with M setAskPatternTimeoutWarningMode("off") expectTimeoutWarning(expectWarning = false) { - TraceRecorder.withNewTraceContext("ask-timeout-warning") { + TraceContext.withContext(newContext("ask-timeout-warning")) { noReplyActorRef ? "hello" - TraceRecorder.currentContext + TraceContext.currentContext } } } diff --git a/kamon-akka/src/test/scala/kamon/metric/ActorMetricsSpec.scala b/kamon-akka/src/test/scala/kamon/metric/ActorMetricsSpec.scala index 6d16386b..322abed2 100644 --- a/kamon-akka/src/test/scala/kamon/metric/ActorMetricsSpec.scala +++ b/kamon-akka/src/test/scala/kamon/metric/ActorMetricsSpec.scala @@ -20,44 +20,29 @@ import java.nio.LongBuffer import kamon.Kamon import kamon.akka.ActorMetrics import kamon.metric.ActorMetricsTestActor._ +import kamon.metric.instrument.CollectionContext import org.scalatest.{ BeforeAndAfterAll, WordSpecLike, Matchers } import akka.testkit.{ ImplicitSender, TestProbe, TestKitBase } import akka.actor._ import com.typesafe.config.ConfigFactory import scala.concurrent.duration._ -import ActorMetrics.{ ActorMetricsRecorder, ActorMetricSnapshot } class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers with ImplicitSender with BeforeAndAfterAll { implicit lazy val system: ActorSystem = ActorSystem("actor-metrics-spec", ConfigFactory.parseString( """ - |kamon.metrics { + |kamon.metric { | tick-interval = 1 hour | default-collection-context-buffer-size = 10 | - | filters = [ - | { - | actor { - | includes = [ "user/tracked-*", "user/measuring-*", "user/clean-after-collect", "user/stop" ] - | excludes = [ "user/tracked-explicitly-excluded"] - | } - | } - | ] - | precision.actor { - | processing-time { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | - | time-in-mailbox { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 + | filters { + | akka-actor { + | includes = [ "user/tracked-*", "user/measuring-*", "user/clean-after-collect", "user/stop" ] + | excludes = [ "user/tracked-explicitly-excluded", "user/non-tracked-actor" ] | } + | } | - | mailbox-size { - | refresh-interval = 1 hour - | highest-trackable-value = 999999999 - | significant-value-digits = 2 - | } + | instrument-settings { + | akka-actor.mailbox-size.refresh-interval = 1 hour | } |} | @@ -89,16 +74,16 @@ class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers with expectMsg(Pong) val firstSnapshot = collectMetricsOf(trackedActor).get - firstSnapshot.errors.count should be(1L) - firstSnapshot.mailboxSize.numberOfMeasurements should be > 0L - firstSnapshot.processingTime.numberOfMeasurements should be(102L) // 102 examples - firstSnapshot.timeInMailbox.numberOfMeasurements should be(102L) // 102 examples + firstSnapshot.counter("errors").get.count should be(1L) + firstSnapshot.minMaxCounter("mailbox-size").get.numberOfMeasurements should be > 0L + firstSnapshot.histogram("processing-time").get.numberOfMeasurements should be(102L) // 102 examples + firstSnapshot.histogram("time-in-mailbox").get.numberOfMeasurements should be(102L) // 102 examples val secondSnapshot = collectMetricsOf(trackedActor).get // Ensure that the recorders are clean - secondSnapshot.errors.count should be(0L) - secondSnapshot.mailboxSize.numberOfMeasurements should be(3L) // min, max and current - secondSnapshot.processingTime.numberOfMeasurements should be(0L) - secondSnapshot.timeInMailbox.numberOfMeasurements should be(0L) + secondSnapshot.counter("errors").get.count should be(0L) + secondSnapshot.minMaxCounter("mailbox-size").get.numberOfMeasurements should be(3L) // min, max and current + secondSnapshot.histogram("processing-time").get.numberOfMeasurements should be(0L) + secondSnapshot.histogram("time-in-mailbox").get.numberOfMeasurements should be(0L) } } @@ -109,9 +94,9 @@ class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers with val timings = expectMsgType[TrackedTimings] val snapshot = collectMetricsOf(trackedActor).get - snapshot.processingTime.numberOfMeasurements should be(1L) - snapshot.processingTime.recordsIterator.next().count should be(1L) - snapshot.processingTime.recordsIterator.next().level should be(timings.approximateProcessingTime +- 10.millis.toNanos) + snapshot.histogram("processing-time").get.numberOfMeasurements should be(1L) + snapshot.histogram("processing-time").get.recordsIterator.next().count should be(1L) + snapshot.histogram("processing-time").get.recordsIterator.next().level should be(timings.approximateProcessingTime +- 10.millis.toNanos) } "record the number of errors" in new ActorMetricsFixtures { @@ -122,7 +107,7 @@ class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers with expectMsg(Pong) val snapshot = collectMetricsOf(trackedActor).get - snapshot.errors.count should be(10) + snapshot.counter("errors").get.count should be(10) } "record the mailbox-size" in new ActorMetricsFixtures { @@ -138,8 +123,8 @@ class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers with expectMsg(Pong) val snapshot = collectMetricsOf(trackedActor).get - snapshot.mailboxSize.min should be(0L) - snapshot.mailboxSize.max should be(11L +- 1L) + snapshot.minMaxCounter("mailbox-size").get.min should be(0L) + snapshot.minMaxCounter("mailbox-size").get.max should be(11L +- 1L) } "record the time-in-mailbox" in new ActorMetricsFixtures { @@ -149,20 +134,22 @@ class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers with val timings = expectMsgType[TrackedTimings] val snapshot = collectMetricsOf(trackedActor).get - snapshot.timeInMailbox.numberOfMeasurements should be(1L) - snapshot.timeInMailbox.recordsIterator.next().count should be(1L) - snapshot.timeInMailbox.recordsIterator.next().level should be(timings.approximateTimeInMailbox +- 10.millis.toNanos) + snapshot.histogram("time-in-mailbox").get.numberOfMeasurements should be(1L) + snapshot.histogram("time-in-mailbox").get.recordsIterator.next().count should be(1L) + snapshot.histogram("time-in-mailbox").get.recordsIterator.next().level should be(timings.approximateTimeInMailbox +- 10.millis.toNanos) } "clean up the associated recorder when the actor is stopped" in new ActorMetricsFixtures { val trackedActor = createTestActor("stop") + val firstRecorder = actorMetricsRecorderOf(trackedActor).get + // Killing the actor should remove it's ActorMetrics and registering again bellow should create a new one. val deathWatcher = TestProbe() deathWatcher.watch(trackedActor) trackedActor ! PoisonPill deathWatcher.expectTerminated(trackedActor) - actorMetricsRecorderOf(trackedActor) shouldBe empty + actorMetricsRecorderOf(trackedActor).get shouldNot be theSameInstanceAs (firstRecorder) } } @@ -175,10 +162,10 @@ class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers with def actorRecorderName(ref: ActorRef): String = ref.path.elements.mkString("/") - def actorMetricsRecorderOf(ref: ActorRef): Option[ActorMetricsRecorder] = - Kamon(Metrics)(system).storage.get(ActorMetrics(actorRecorderName(ref))).map(_.asInstanceOf[ActorMetricsRecorder]) + def actorMetricsRecorderOf(ref: ActorRef): Option[ActorMetrics] = + Kamon(Metrics)(system).register(ActorMetrics, actorRecorderName(ref)).map(_.recorder) - def collectMetricsOf(ref: ActorRef): Option[ActorMetricSnapshot] = { + def collectMetricsOf(ref: ActorRef): Option[EntitySnapshot] = { Thread.sleep(5) // Just in case the test advances a bit faster than the actor being tested. actorMetricsRecorderOf(ref).map(_.collect(collectionContext)) } diff --git a/kamon-akka/src/test/scala/kamon/metric/DispatcherMetricsSpec.scala b/kamon-akka/src/test/scala/kamon/metric/DispatcherMetricsSpec.scala index 55af3f2e..2c530da9 100644 --- a/kamon-akka/src/test/scala/kamon/metric/DispatcherMetricsSpec.scala +++ b/kamon-akka/src/test/scala/kamon/metric/DispatcherMetricsSpec.scala @@ -15,96 +15,199 @@ package kamon.metric -import akka.actor.{ ActorRef, ActorSystem, Props } +import java.nio.LongBuffer + +import akka.actor.{ PoisonPill, Props, ActorRef, ActorSystem } +import akka.dispatch.MessageDispatcher import akka.testkit.{ TestKitBase, TestProbe } import com.typesafe.config.ConfigFactory import kamon.Kamon -import kamon.akka.DispatcherMetrics -import DispatcherMetrics.DispatcherMetricSnapshot -import kamon.metric.Subscriptions.TickMetricSnapshot -import org.scalatest.{ Matchers, WordSpecLike } +import kamon.akka.{ ForkJoinPoolDispatcherMetrics, ThreadPoolExecutorDispatcherMetrics } +import kamon.metric.ActorMetricsTestActor.{ Pong, Ping } +import kamon.metric.instrument.CollectionContext +import org.scalatest.{ BeforeAndAfterAll, Matchers, WordSpecLike } +import scala.concurrent.{ Await, Future } import scala.concurrent.duration._ -class DispatcherMetricsSpec extends TestKitBase with WordSpecLike with Matchers { +class DispatcherMetricsSpec extends TestKitBase with WordSpecLike with Matchers with BeforeAndAfterAll { implicit lazy val system: ActorSystem = ActorSystem("dispatcher-metrics-spec", ConfigFactory.parseString( """ - |kamon.metrics { - | tick-interval = 1 second + |kamon.metric { + | tick-interval = 1 hour | default-collection-context-buffer-size = 10 | - | filters = [ - | { - | dispatcher { - | includes = ["*"] - | excludes = ["dispatcher-explicitly-excluded"] - | } + | filters = { + | akka-dispatcher { + | includes = [ "*" ] + | excludes = [ "explicitly-excluded" ] | } - | ] + | } + | + | default-instrument-settings { + | gauge.refresh-interval = 1 hour + | min-max-counter.refresh-interval = 1 hour + | } + |} + | + |explicitly-excluded { + | type = "Dispatcher" + | executor = "fork-join-executor" |} | - |dispatcher-explicitly-excluded { - | type = "Dispatcher" - | executor = "fork-join-executor" + |tracked-fjp { + | type = "Dispatcher" + | executor = "fork-join-executor" + | + | fork-join-executor { + | parallelism-min = 8 + | parallelism-factor = 100.0 + | parallelism-max = 22 + | } |} | - |tracked-dispatcher { - | type = "Dispatcher" - | executor = "thread-pool-executor" + |tracked-tpe { + | type = "Dispatcher" + | executor = "thread-pool-executor" + | + | thread-pool-executor { + | core-pool-size-min = 7 + | core-pool-size-factor = 100.0 + | max-pool-size-factor = 100.0 + | max-pool-size-max = 21 + | } |} | """.stripMargin)) "the Kamon dispatcher metrics" should { "respect the configured include and exclude filters" in { - system.actorOf(Props[ActorMetricsTestActor].withDispatcher("tracked-dispatcher"), "actor-with-tracked-dispatcher") - system.actorOf(Props[ActorMetricsTestActor].withDispatcher("dispatcher-explicitly-excluded"), "actor-with-excluded-dispatcher") + val defaultDispatcher = forceInit(system.dispatchers.lookup("akka.actor.default-dispatcher")) + val fjpDispatcher = forceInit(system.dispatchers.lookup("tracked-fjp")) + val tpeDispatcher = forceInit(system.dispatchers.lookup("tracked-tpe")) + val excludedDispatcher = forceInit(system.dispatchers.lookup("explicitly-excluded")) + + findDispatcherRecorder(defaultDispatcher) shouldNot be(empty) + findDispatcherRecorder(fjpDispatcher) shouldNot be(empty) + findDispatcherRecorder(tpeDispatcher) shouldNot be(empty) + findDispatcherRecorder(excludedDispatcher) should be(empty) + } - Kamon(Metrics).subscribe(DispatcherMetrics, "*", testActor, permanently = true) - expectMsgType[TickMetricSnapshot] + "record metrics for a dispatcher with thread-pool-executor" in { + implicit val tpeDispatcher = system.dispatchers.lookup("tracked-tpe") + collectDispatcherMetrics(tpeDispatcher) - within(2 seconds) { - val tickSnapshot = expectMsgType[TickMetricSnapshot] - tickSnapshot.metrics.keys should contain(DispatcherMetrics("tracked-dispatcher")) - tickSnapshot.metrics.keys should not contain (DispatcherMetrics("dispatcher-explicitly-excluded")) - } + Await.result({ + Future.sequence { + for (_ ← 1 to 100) yield submit(tpeDispatcher) + } + }, 5 seconds) + + refreshDispatcherInstruments(tpeDispatcher) + val snapshot = collectDispatcherMetrics(tpeDispatcher) + + snapshot.gauge("active-threads") should not be empty + snapshot.gauge("pool-size").get.min should be >= 7L + snapshot.gauge("pool-size").get.max should be <= 21L + snapshot.gauge("max-pool-size").get.max should be(21) + snapshot.gauge("core-pool-size").get.max should be(21) + snapshot.gauge("processed-tasks").get.max should be(102L +- 5L) + + // The processed tasks should be reset to 0 if no more tasks are submitted. + val secondSnapshot = collectDispatcherMetrics(tpeDispatcher) + secondSnapshot.gauge("processed-tasks").get.max should be(0) } - "record maximumPoolSize, runningThreadCount, queueTaskCount, poolSize metrics" in new DelayableActorFixture { - val (delayable, metricsListener) = delayableActor("worker-actor", "tracked-dispatcher") + "record metrics for a dispatcher with fork-join-executor" in { + implicit val fjpDispatcher = system.dispatchers.lookup("tracked-fjp") + collectDispatcherMetrics(fjpDispatcher) + + Await.result({ + Future.sequence { + for (_ ← 1 to 100) yield submit(fjpDispatcher) + } + }, 5 seconds) - for (_ ← 1 to 100) { - //delayable ! Discard - } + refreshDispatcherInstruments(fjpDispatcher) + val snapshot = collectDispatcherMetrics(fjpDispatcher) + + snapshot.minMaxCounter("parallelism").get.max should be(22) + snapshot.gauge("pool-size").get.min should be >= 0L + snapshot.gauge("pool-size").get.max should be <= 22L + snapshot.gauge("active-threads").get.max should be >= 0L + snapshot.gauge("running-threads").get.max should be >= 0L + snapshot.gauge("queued-task-count").get.max should be(0) - val dispatcherMetrics = expectDispatcherMetrics("tracked-dispatcher", metricsListener, 3 seconds) - dispatcherMetrics.maximumPoolSize.max should be <= 64L //fail in travis - dispatcherMetrics.poolSize.max should be <= 22L //fail in travis - dispatcherMetrics.queueTaskCount.max should be(0L) - dispatcherMetrics.runningThreadCount.max should be(0L) } - } + "clean up the metrics recorders after a dispatcher is shut down" in { + implicit val tpeDispatcher = system.dispatchers.lookup("tracked-tpe") + implicit val fjpDispatcher = system.dispatchers.lookup("tracked-fjp") + + findDispatcherRecorder(fjpDispatcher) shouldNot be(empty) + findDispatcherRecorder(tpeDispatcher) shouldNot be(empty) - def expectDispatcherMetrics(dispatcherId: String, listener: TestProbe, waitTime: FiniteDuration): DispatcherMetricSnapshot = { - val tickSnapshot = within(waitTime) { - listener.expectMsgType[TickMetricSnapshot] + shutdownDispatcher(tpeDispatcher) + shutdownDispatcher(fjpDispatcher) + + findDispatcherRecorder(fjpDispatcher) should be(empty) + findDispatcherRecorder(tpeDispatcher) should be(empty) } - val dispatcherMetricsOption = tickSnapshot.metrics.get(DispatcherMetrics(dispatcherId)) - dispatcherMetricsOption should not be empty - dispatcherMetricsOption.get.asInstanceOf[DispatcherMetricSnapshot] + + } + + val collectionContext = new CollectionContext { + val buffer: LongBuffer = LongBuffer.allocate(10000) } - trait DelayableActorFixture { - def delayableActor(name: String, dispatcher: String): (ActorRef, TestProbe) = { - val actor = system.actorOf(Props[ActorMetricsTestActor].withDispatcher(dispatcher), name) - val metricsListener = TestProbe() + def actorRecorderName(ref: ActorRef): String = ref.path.elements.mkString("/") - Kamon(Metrics).subscribe(DispatcherMetrics, "*", metricsListener.ref, permanently = true) - // Wait for one empty snapshot before proceeding to the test. - metricsListener.expectMsgType[TickMetricSnapshot] + def findDispatcherRecorder(dispatcher: MessageDispatcher): Option[EntityRecorder] = + Kamon(Metrics)(system).find(dispatcher.id, "akka-dispatcher") - (actor, metricsListener) + def collectDispatcherMetrics(dispatcher: MessageDispatcher): EntitySnapshot = + findDispatcherRecorder(dispatcher).map(_.collect(collectionContext)).get + + def refreshDispatcherInstruments(dispatcher: MessageDispatcher): Unit = { + findDispatcherRecorder(dispatcher) match { + case Some(tpe: ThreadPoolExecutorDispatcherMetrics) ⇒ + tpe.processedTasks.refreshValue() + tpe.activeThreads.refreshValue() + tpe.maxPoolSize.refreshValue() + tpe.poolSize.refreshValue() + tpe.corePoolSize.refreshValue() + + case Some(fjp: ForkJoinPoolDispatcherMetrics) ⇒ + fjp.activeThreads.refreshValue() + fjp.poolSize.refreshValue() + fjp.queuedTaskCount.refreshValue() + fjp.paralellism.refreshValues() + fjp.runningThreads.refreshValue() + + case other ⇒ } } + + def forceInit(dispatcher: MessageDispatcher): MessageDispatcher = { + val listener = TestProbe() + Future { + listener.ref ! "init done" + }(dispatcher) + listener.expectMsg("init done") + + dispatcher + } + + def submit(dispatcher: MessageDispatcher): Future[String] = Future { + "hello" + }(dispatcher) + + def shutdownDispatcher(dispatcher: MessageDispatcher): Unit = { + val shutdownMethod = dispatcher.getClass.getDeclaredMethod("shutdown") + shutdownMethod.setAccessible(true) + shutdownMethod.invoke(dispatcher) + } + + override protected def afterAll(): Unit = system.shutdown() } + diff --git a/kamon-akka/src/test/scala/kamon/metric/RouterMetricsSpec.scala b/kamon-akka/src/test/scala/kamon/metric/RouterMetricsSpec.scala index abc195ba..5f6bbb4f 100644 --- a/kamon-akka/src/test/scala/kamon/metric/RouterMetricsSpec.scala +++ b/kamon-akka/src/test/scala/kamon/metric/RouterMetricsSpec.scala @@ -18,17 +18,13 @@ package kamon.metric import java.nio.LongBuffer import akka.actor._ -import akka.kamon.instrumentation.ActorCellMetrics import akka.routing._ import akka.testkit.{ ImplicitSender, TestKitBase, TestProbe } import com.typesafe.config.ConfigFactory import kamon.Kamon -import kamon.akka.{ RouterMetrics, ActorMetrics } -import ActorMetrics.{ ActorMetricSnapshot, ActorMetricsRecorder } -import RouterMetrics._ +import kamon.akka.RouterMetrics import kamon.metric.RouterMetricsTestActor._ -import kamon.metric.Subscriptions.TickMetricSnapshot -import kamon.metric.instrument.{ Counter, Histogram } +import kamon.metric.instrument.CollectionContext import org.scalatest.{ BeforeAndAfterAll, Matchers, WordSpecLike } import scala.concurrent.duration._ @@ -36,22 +32,14 @@ import scala.concurrent.duration._ class RouterMetricsSpec extends TestKitBase with WordSpecLike with Matchers with ImplicitSender with BeforeAndAfterAll { implicit lazy val system: ActorSystem = ActorSystem("router-metrics-spec", ConfigFactory.parseString( """ - |kamon.metrics { + |kamon.metric { | tick-interval = 1 hour | default-collection-context-buffer-size = 10 | - | filters = [ - | { - | router { - | includes = [ "user/tracked-*", "user/measuring-*", "user/stop-*" ] - | excludes = [ "user/tracked-explicitly-excluded-*"] - | } - | } - | ] - | precision { - | default-histogram-precision { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 + | filters = { + | akka-router { + | includes = [ "user/tracked-*", "user/measuring-*", "user/stop-*" ] + | excludes = [ "user/tracked-explicitly-excluded-*"] | } | } |} @@ -85,7 +73,7 @@ class RouterMetricsSpec extends TestKitBase with WordSpecLike with Matchers with listener.expectMsg(Pong) val routerSnapshot = collectMetricsOf("user/measuring-routing-time-in-pool-router").get - routerSnapshot.routingTime.numberOfMeasurements should be(1L) + routerSnapshot.histogram("routing-time").get.numberOfMeasurements should be(1L) } "record the routing-time of the receive function for group routers" in new RouterMetricsFixtures { @@ -96,7 +84,7 @@ class RouterMetricsSpec extends TestKitBase with WordSpecLike with Matchers with listener.expectMsg(Pong) val routerSnapshot = collectMetricsOf("user/measuring-routing-time-in-group-router").get - routerSnapshot.routingTime.numberOfMeasurements should be(1L) + routerSnapshot.histogram("routing-time").get.numberOfMeasurements should be(1L) } "record the processing-time of the receive function for pool routers" in new RouterMetricsFixtures { @@ -107,9 +95,9 @@ class RouterMetricsSpec extends TestKitBase with WordSpecLike with Matchers with val timings = timingsListener.expectMsgType[RouterTrackedTimings] val routerSnapshot = collectMetricsOf("user/measuring-processing-time-in-pool-router").get - routerSnapshot.processingTime.numberOfMeasurements should be(1L) - routerSnapshot.processingTime.recordsIterator.next().count should be(1L) - routerSnapshot.processingTime.recordsIterator.next().level should be(timings.approximateProcessingTime +- 10.millis.toNanos) + routerSnapshot.histogram("processing-time").get.numberOfMeasurements should be(1L) + routerSnapshot.histogram("processing-time").get.recordsIterator.next().count should be(1L) + routerSnapshot.histogram("processing-time").get.recordsIterator.next().level should be(timings.approximateProcessingTime +- 10.millis.toNanos) } "record the processing-time of the receive function for group routers" in new RouterMetricsFixtures { @@ -120,9 +108,9 @@ class RouterMetricsSpec extends TestKitBase with WordSpecLike with Matchers with val timings = timingsListener.expectMsgType[RouterTrackedTimings] val routerSnapshot = collectMetricsOf("user/measuring-processing-time-in-group-router").get - routerSnapshot.processingTime.numberOfMeasurements should be(1L) - routerSnapshot.processingTime.recordsIterator.next().count should be(1L) - routerSnapshot.processingTime.recordsIterator.next().level should be(timings.approximateProcessingTime +- 10.millis.toNanos) + routerSnapshot.histogram("processing-time").get.numberOfMeasurements should be(1L) + routerSnapshot.histogram("processing-time").get.recordsIterator.next().count should be(1L) + routerSnapshot.histogram("processing-time").get.recordsIterator.next().level should be(timings.approximateProcessingTime +- 10.millis.toNanos) } "record the number of errors for pool routers" in new RouterMetricsFixtures { @@ -137,7 +125,7 @@ class RouterMetricsSpec extends TestKitBase with WordSpecLike with Matchers with listener.expectMsg(Pong) val routerSnapshot = collectMetricsOf("user/measuring-errors-in-pool-router").get - routerSnapshot.errors.count should be(10L) + routerSnapshot.counter("errors").get.count should be(10L) } "record the number of errors for group routers" in new RouterMetricsFixtures { @@ -152,7 +140,7 @@ class RouterMetricsSpec extends TestKitBase with WordSpecLike with Matchers with listener.expectMsg(Pong) val routerSnapshot = collectMetricsOf("user/measuring-errors-in-group-router").get - routerSnapshot.errors.count should be(10L) + routerSnapshot.counter("errors").get.count should be(10L) } "record the time-in-mailbox for pool routers" in new RouterMetricsFixtures { @@ -163,9 +151,9 @@ class RouterMetricsSpec extends TestKitBase with WordSpecLike with Matchers with val timings = timingsListener.expectMsgType[RouterTrackedTimings] val routerSnapshot = collectMetricsOf("user/measuring-time-in-mailbox-in-pool-router").get - routerSnapshot.timeInMailbox.numberOfMeasurements should be(1L) - routerSnapshot.timeInMailbox.recordsIterator.next().count should be(1L) - routerSnapshot.timeInMailbox.recordsIterator.next().level should be(timings.approximateTimeInMailbox +- 10.millis.toNanos) + routerSnapshot.histogram("time-in-mailbox").get.numberOfMeasurements should be(1L) + routerSnapshot.histogram("time-in-mailbox").get.recordsIterator.next().count should be(1L) + routerSnapshot.histogram("time-in-mailbox").get.recordsIterator.next().level should be(timings.approximateTimeInMailbox +- 10.millis.toNanos) } "record the time-in-mailbox for group routers" in new RouterMetricsFixtures { @@ -176,33 +164,35 @@ class RouterMetricsSpec extends TestKitBase with WordSpecLike with Matchers with val timings = timingsListener.expectMsgType[RouterTrackedTimings] val routerSnapshot = collectMetricsOf("user/measuring-time-in-mailbox-in-group-router").get - routerSnapshot.timeInMailbox.numberOfMeasurements should be(1L) - routerSnapshot.timeInMailbox.recordsIterator.next().count should be(1L) - routerSnapshot.timeInMailbox.recordsIterator.next().level should be(timings.approximateTimeInMailbox +- 10.millis.toNanos) + routerSnapshot.histogram("time-in-mailbox").get.numberOfMeasurements should be(1L) + routerSnapshot.histogram("time-in-mailbox").get.recordsIterator.next().count should be(1L) + routerSnapshot.histogram("time-in-mailbox").get.recordsIterator.next().level should be(timings.approximateTimeInMailbox +- 10.millis.toNanos) } "clean up the associated recorder when the pool router is stopped" in new RouterMetricsFixtures { val trackedRouter = createTestPoolRouter("stop-in-pool-router") - routerMetricsRecorderOf("user/stop-in-pool-router") should not be empty + val firstRecorder = routerMetricsRecorderOf("user/stop-in-pool-router").get + // Killing the router should remove it's RouterMetrics and registering again bellow should create a new one. val deathWatcher = TestProbe() deathWatcher.watch(trackedRouter) trackedRouter ! PoisonPill deathWatcher.expectTerminated(trackedRouter) - routerMetricsRecorderOf("user/stop-in-pool-router") shouldBe empty + routerMetricsRecorderOf("user/stop-in-pool-router").get shouldNot be theSameInstanceAs (firstRecorder) } "clean up the associated recorder when the group router is stopped" in new RouterMetricsFixtures { val trackedRouter = createTestPoolRouter("stop-in-group-router") - routerMetricsRecorderOf("user/stop-in-group-router") should not be empty + val firstRecorder = routerMetricsRecorderOf("user/stop-in-group-router").get + // Killing the router should remove it's RouterMetrics and registering again bellow should create a new one. val deathWatcher = TestProbe() deathWatcher.watch(trackedRouter) trackedRouter ! PoisonPill deathWatcher.expectTerminated(trackedRouter) - routerMetricsRecorderOf("user/stop-in-group-router") shouldBe empty + routerMetricsRecorderOf("user/stop-in-group-router").get shouldNot be theSameInstanceAs (firstRecorder) } } @@ -213,10 +203,10 @@ class RouterMetricsSpec extends TestKitBase with WordSpecLike with Matchers with val buffer: LongBuffer = LongBuffer.allocate(10000) } - def routerMetricsRecorderOf(routerName: String): Option[RouterMetricsRecorder] = - Kamon(Metrics)(system).storage.get(RouterMetrics(routerName)).map(_.asInstanceOf[RouterMetricsRecorder]) + def routerMetricsRecorderOf(routerName: String): Option[RouterMetrics] = + Kamon(Metrics)(system).register(RouterMetrics, routerName).map(_.recorder) - def collectMetricsOf(routerName: String): Option[RouterMetricSnapshot] = { + def collectMetricsOf(routerName: String): Option[EntitySnapshot] = { Thread.sleep(5) // Just in case the test advances a bit faster than the actor being tested. routerMetricsRecorderOf(routerName).map(_.collect(collectionContext)) } @@ -255,16 +245,6 @@ class RouterMetricsSpec extends TestKitBase with WordSpecLike with Matchers with router } } - - trait ActorMetricsFixtures { - val collectionContext = new CollectionContext { - val buffer: LongBuffer = LongBuffer.allocate(10000) - } - - def createTestActor(name: String): ActorRef = system.actorOf(Props[ActorMetricsTestActor], name) - - def takeSnapshotOf(amr: ActorMetricsRecorder): ActorMetricSnapshot = amr.collect(collectionContext) - } } class RouterMetricsTestActor extends Actor { |