Akka/Scala example source code file (MetricsKit.scala)

This example Akka source code file (MetricsKit.scala) is included in my "Source Code Warehouse" project. The intent of this project is to help you more easily find Akka and Scala source code examples by using tags.

All credit for the original source code belongs to akka.io; I'm just trying to make examples easier to find. (For my Scala work, see my Scala examples and tutorials.)

Akka tags/keywords

akka, concurrent, m, metric, metricfilter, metricregistry, metrics, metricskit, regexmetricfilter, some, string, t, test, testing, testkit, verbose

The MetricsKit.scala Akka example source code

/**
 * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
 */
package akka.testkit.metrics

import com.codahale.metrics._

import java.net.InetSocketAddress
import java.util.concurrent.TimeUnit
import scala.concurrent.duration._
import com.typesafe.config.Config
import java.util
import scala.util.matching.Regex
import scala.collection.mutable
import akka.testkit.metrics.reporter.{ GraphiteClient, AkkaGraphiteReporter, AkkaConsoleReporter }
import org.scalatest.Notifying
import scala.reflect.ClassTag

/**
 * Allows to easily measure performance / memory / file descriptor use in tests.
 *
 * WARNING: This trait should not be seen as utility for micro-benchmarking,
 * please refer to <a href="http://openjdk.java.net/projects/code-tools/jmh/">JMH</a> if that's what you're writing.
 * This trait instead aims to give an high level overview as well as data for trend-analysis of long running tests.
 *
 * Reporting defaults to [[ConsoleReporter]].
 * In order to send registry to Graphite run sbt with the following property: `-Dakka.registry.reporting.0=graphite`.
 */
private[akka] trait MetricsKit extends MetricsKitOps {
  this: Notifying ⇒

  import MetricsKit._
  import collection.JavaConverters._

  private var reporters: List[ScheduledReporter] = Nil

  /**
   * A configuration containing [[MetricsKitSettings]] under the key `akka.test.registry` must be provided.
   * This can be the ActorSystems config.
   *
   * The reason this is not handled by an Extension is thatwe do not want to enforce having to start an ActorSystem,
   * since code measured using this Kit may not need one (e.g. measuring plain Queue implementations).
   */
  def metricsConfig: Config

  private[metrics] val registry = new MetricRegistry() with AkkaMetricRegistry

  initMetricReporters()

  def initMetricReporters() {
    val settings = new MetricsKitSettings(metricsConfig)

    def configureConsoleReporter() {
      if (settings.Reporters.contains("console")) {
        val akkaConsoleReporter = new AkkaConsoleReporter(registry, settings.ConsoleReporter.Verbose)

        if (settings.ConsoleReporter.ScheduledReportInterval > Duration.Zero)
          akkaConsoleReporter.start(settings.ConsoleReporter.ScheduledReportInterval.toMillis, TimeUnit.MILLISECONDS)

        reporters ::= akkaConsoleReporter
      }
    }

    def configureGraphiteReporter() {
      if (settings.Reporters.contains("graphite")) {
        note(s"MetricsKit: Graphite reporter enabled, sending metrics to: ${settings.GraphiteReporter.Host}:${settings.GraphiteReporter.Port}")
        val address = new InetSocketAddress(settings.GraphiteReporter.Host, settings.GraphiteReporter.Port)
        val graphite = new GraphiteClient(address)
        val akkaGraphiteReporter = new AkkaGraphiteReporter(registry, settings.GraphiteReporter.Prefix, graphite, settings.GraphiteReporter.Verbose)

        if (settings.GraphiteReporter.ScheduledReportInterval > Duration.Zero) {
          akkaGraphiteReporter.start(settings.GraphiteReporter.ScheduledReportInterval.toMillis, TimeUnit.MILLISECONDS)
        }

        reporters ::= akkaGraphiteReporter
      }
    }

    configureConsoleReporter()
    configureGraphiteReporter()
  }

  /**
   * Schedule metric reports execution iterval. Should not be used multiple times
   */
  def scheduleMetricReports(every: FiniteDuration) {
    reporters foreach { _.start(every.toMillis, TimeUnit.MILLISECONDS) }
  }

  def registeredMetrics = registry.getMetrics.asScala

  /**
   * Causes immediate flush of metrics, using all registered reporters.
   * Afterwards all metrics are removed from the registry.
   *
   * HINT: this operation can be costy, run outside of your tested code, or rely on scheduled reporting.
   */
  def reportAndClearMetrics() {
    reportMetrics()
    clearMetrics()
  }

  /**
   * Causes immediate flush of metrics, using all registered reporters.
   *
   * HINT: this operation can be costy, run outside of your tested code, or rely on scheduled reporting.
   */
  def reportMetrics() {
    reporters foreach { _.report() }
  }

  /**
   * Causes immediate flush of only memory related metrics, using all registered reporters.
   *
   * HINT: this operation can be costy, run outside of your tested code, or rely on scheduled reporting.
   */
  def reportMemoryMetrics() {
    val gauges = registry.getGauges(MemMetricsFilter)

    reporters foreach { _.report(gauges, empty, empty, empty, empty) }
  }

  /**
   * Causes immediate flush of only memory related metrics, using all registered reporters.
   *
   * HINT: this operation can be costy, run outside of your tested code, or rely on scheduled reporting.
   */
  def reportGcMetrics() {
    val gauges = registry.getGauges(GcMetricsFilter)

    reporters foreach { _.report(gauges, empty, empty, empty, empty) }
  }

  /**
   * Causes immediate flush of only file descriptor metrics, using all registered reporters.
   *
   * HINT: this operation can be costy, run outside of your tested code, or rely on scheduled reporting.
   */
  def reportFileDescriptorMetrics() {
    val gauges = registry.getGauges(FileDescriptorMetricsFilter)

    reporters foreach { _.report(gauges, empty, empty, empty, empty) }
  }

  /**
   * Removes registered registry from registry.
   * You should call this method then you're done measuring something - usually at the end of your test case,
   * otherwise the registry from different tests would influence each others results (avg, min, max, ...).
   *
   * Please note that, if you have registered a `timer("thing")` previously, you will need to call `timer("thing")` again,
   * in order to register a new timer.
   */
  def clearMetrics(matching: MetricFilter = MetricFilter.ALL) {
    registry.removeMatching(matching)
  }

  /**
   * MUST be called after all tests have finished.
   */
  def shutdownMetrics() {
    reporters foreach { _.stop() }
  }

  private[metrics] def getOrRegister[M <: Metric](key: String, metric: ⇒ M)(implicit tag: ClassTag[M]): M = {
    import collection.JavaConverters._
    registry.getMetrics.asScala.find(_._1 == key).map(_._2) match {
      case Some(existing: M) ⇒ existing
      case Some(existing)    ⇒ throw new IllegalArgumentException("Key: [%s] is already for different kind of metric! Was [%s], expected [%s]".format(key, metric.getClass.getSimpleName, tag.runtimeClass.getSimpleName))
      case _                 ⇒ registry.register(key, metric)
    }
  }

  private val emptySortedMap = new util.TreeMap[String, Nothing]()
  private def empty[T] = emptySortedMap.asInstanceOf[util.TreeMap[String, T]]
}

private[akka] object MetricsKit {

  class RegexMetricFilter(regex: Regex) extends MetricFilter {
    override def matches(name: String, metric: Metric) = regex.pattern.matcher(name).matches()
  }

  val MemMetricsFilter = new RegexMetricFilter(""".*\.mem\..*""".r)

  val FileDescriptorMetricsFilter = new RegexMetricFilter(""".*\.file-descriptors\..*""".r)

  val KnownOpsInTimespanCounterFilter = new MetricFilter {
    override def matches(name: String, metric: Metric) = classOf[KnownOpsInTimespanTimer].isInstance(metric)
  }

  val GcMetricsFilter = new MetricFilter {
    val keyPattern = """.*\.gc\..*""".r.pattern

    override def matches(name: String, metric: Metric) = keyPattern.matcher(name).matches()
  }
}

/** Provides access to custom Akka [[Metric]]s, with named methods. */
trait AkkaMetricRegistry {
  this: MetricRegistry ⇒

  def getKnownOpsInTimespanCounters = filterFor(classOf[KnownOpsInTimespanTimer])
  def getHdrHistograms = filterFor(classOf[HdrHistogram])
  def getAveragingGauges = filterFor(classOf[AveragingGauge])

  import collection.JavaConverters._
  private def filterFor[T](clazz: Class[T]): mutable.Iterable[(String, T)] =
    for {
      (key, metric) ← getMetrics.asScala
      if clazz.isInstance(metric)
    } yield key -> metric.asInstanceOf[T]
}

private[akka] class MetricsKitSettings(config: Config) {

  import akka.util.Helpers._

  val Reporters = config.getStringList("akka.test.metrics.reporters")

  object GraphiteReporter {
    val Prefix = config.getString("akka.test.metrics.reporter.graphite.prefix")
    lazy val Host = config.getString("akka.test.metrics.reporter.graphite.host").requiring(v ⇒ !v.trim.isEmpty, "akka.test.metrics.reporter.graphite.host was used but was empty!")
    val Port = config.getInt("akka.test.metrics.reporter.graphite.port")
    val Verbose = config.getBoolean("akka.test.metrics.reporter.graphite.verbose")

    val ScheduledReportInterval = config.getMillisDuration("akka.test.metrics.reporter.graphite.scheduled-report-interval")
  }

  object ConsoleReporter {
    val ScheduledReportInterval = config.getMillisDuration("akka.test.metrics.reporter.console.scheduled-report-interval")
    val Verbose = config.getBoolean("akka.test.metrics.reporter.console.verbose")
  }

}