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

This example Akka source code file (AkkaGraphiteReporter.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, akkagraphitereporter, concurrent, double, histogram, int, list, long, metrics, string, t, test, testing, testkit, timer, unit

The AkkaGraphiteReporter.scala Akka example source code

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

import java.text.DateFormat
import java.util
import java.util.concurrent.TimeUnit
import com.codahale.metrics._
import java.util.{ Locale, Date }
import akka.testkit.metrics._
import scala.concurrent.duration._

/**
 * Used to report [[Metric]] types that the original [[com.codahale.metrics.graphite.GraphiteReporter]] is unaware of (cannot re-use directly because of private constructor).
 */
class AkkaGraphiteReporter(
  registry: AkkaMetricRegistry,
  prefix: String,
  graphite: GraphiteClient,
  verbose: Boolean = false)
  extends ScheduledReporter(registry.asInstanceOf[MetricRegistry], "akka-graphite-reporter", MetricFilter.ALL, TimeUnit.SECONDS, TimeUnit.NANOSECONDS) {

  // todo get rid of ScheduledReporter (would mean removing codahale metrics)?

  private final val ConsoleWidth = 80

  val locale = Locale.getDefault
  val dateFormat = DateFormat.getDateTimeInstance(DateFormat.SHORT, DateFormat.MEDIUM, locale)
  val clock = Clock.defaultClock()

  override def report(gauges: util.SortedMap[String, Gauge[_]], counters: util.SortedMap[String, Counter], histograms: util.SortedMap[String, Histogram], meters: util.SortedMap[String, Meter], timers: util.SortedMap[String, Timer]) {
    val dateTime = dateFormat.format(new Date(clock.getTime))

    // akka-custom metrics
    val knownOpsInTimespanCounters = registry.getKnownOpsInTimespanCounters
    val hdrHistograms = registry.getHdrHistograms
    val averagingGauges = registry.getAveragingGauges

    val metricsCount = List(gauges, counters, histograms, meters, timers).map(_.size).sum + List(knownOpsInTimespanCounters, hdrHistograms).map(_.size).sum
    sendWithBanner("== AkkaGraphiteReporter @ " + dateTime + " == (" + metricsCount + " metrics)", '=')

    try {
      // graphite takes timestamps in seconds
      val now = System.currentTimeMillis.millis.toSeconds

      // default Metrics types
      import collection.JavaConverters._
      sendMetrics(now, gauges.asScala, sendGauge)
      sendMetrics(now, counters.asScala, sendCounter)
      sendMetrics(now, histograms.asScala, sendHistogram)
      sendMetrics(now, meters.asScala, sendMetered)
      sendMetrics(now, timers.asScala, sendTimer)

      sendMetrics(now, knownOpsInTimespanCounters, sendKnownOpsInTimespanCounter)
      sendMetrics(now, hdrHistograms, sendHdrHistogram)
      sendMetrics(now, averagingGauges, sendAveragingGauge)

    } catch {
      case ex: Exception ⇒ throw new RuntimeException("Unable to send metrics to Graphite!", ex)
    }
  }

  def sendMetrics[T <: Metric](now: Long, metrics: Iterable[(String, T)], send: (Long, String, T) ⇒ Unit) {
    for ((key, metric) ← metrics) {
      if (verbose)
        println("  " + key)
      send(now, key, metric)
    }
  }

  private def sendHistogram(now: Long, key: String, histogram: Histogram) {
    val snapshot = histogram.getSnapshot
    send(key + ".count", histogram.getCount, now)
    send(key + ".max", snapshot.getMax, now)
    send(key + ".mean", snapshot.getMean, now)
    send(key + ".min", snapshot.getMin, now)
    send(key + ".stddev", snapshot.getStdDev, now)
    send(key + ".p50", snapshot.getMedian, now)
    send(key + ".p75", snapshot.get75thPercentile, now)
    send(key + ".p95", snapshot.get95thPercentile, now)
    send(key + ".p98", snapshot.get98thPercentile, now)
    send(key + ".p99", snapshot.get99thPercentile, now)
    send(key + ".p999", snapshot.get999thPercentile, now)
  }

  private def sendTimer(now: Long, key: String, timer: Timer) {
    val snapshot = timer.getSnapshot
    send(key + ".max", convertDuration(snapshot.getMax), now)
    send(key + ".mean", convertDuration(snapshot.getMean), now)
    send(key + ".min", convertDuration(snapshot.getMin), now)
    send(key + ".stddev", convertDuration(snapshot.getStdDev), now)
    send(key + ".p50", convertDuration(snapshot.getMedian), now)
    send(key + ".p75", convertDuration(snapshot.get75thPercentile), now)
    send(key + ".p95", convertDuration(snapshot.get95thPercentile), now)
    send(key + ".p98", convertDuration(snapshot.get98thPercentile), now)
    send(key + ".p99", convertDuration(snapshot.get99thPercentile), now)
    send(key + ".p999", convertDuration(snapshot.get999thPercentile), now)
    sendMetered(now, key, timer)
  }

  private def sendMetered(now: Long, key: String, meter: Metered) {
    send(key + ".count", meter.getCount, now)
    send(key + ".m1_rate", convertRate(meter.getOneMinuteRate), now)
    send(key + ".m5_rate", convertRate(meter.getFiveMinuteRate), now)
    send(key + ".m15_rate", convertRate(meter.getFifteenMinuteRate), now)
    send(key + ".mean_rate", convertRate(meter.getMeanRate), now)
  }

  private def sendGauge(now: Long, key: String, gauge: Gauge[_]) {
    sendNumericOrIgnore(key + ".gauge", gauge.getValue, now)
  }

  private def sendCounter(now: Long, key: String, counter: Counter) {
    sendNumericOrIgnore(key + ".count", counter.getCount, now)
  }

  private def sendKnownOpsInTimespanCounter(now: Long, key: String, counter: KnownOpsInTimespanTimer) {
    send(key + ".ops", counter.getCount, now)
    send(key + ".time", counter.elapsedTime, now)
    send(key + ".opsPerSec", counter.opsPerSecond, now)
    send(key + ".avg", counter.avgDuration, now)
  }

  private def sendHdrHistogram(now: Long, key: String, hist: HdrHistogram) {
    val snapshot = hist.getData
    send(key + ".min", snapshot.getMinValue, now)
    send(key + ".max", snapshot.getMaxValue, now)
    send(key + ".mean", snapshot.getMean, now)
    send(key + ".stddev", snapshot.getStdDeviation, now)
    send(key + ".p75", snapshot.getValueAtPercentile(75.0), now)
    send(key + ".p95", snapshot.getValueAtPercentile(95.0), now)
    send(key + ".p98", snapshot.getValueAtPercentile(98.0), now)
    send(key + ".p99", snapshot.getValueAtPercentile(99.0), now)
    send(key + ".p999", snapshot.getValueAtPercentile(99.9), now)
  }

  private def sendAveragingGauge(now: Long, key: String, gauge: AveragingGauge) {
    sendNumericOrIgnore(key + ".avg-gauge", gauge.getValue, now)
  }

  override def stop(): Unit = try {
    super.stop()
    graphite.close()
  } catch {
    case ex: Exception ⇒ System.err.println("Was unable to close Graphite connection: " + ex.getMessage)
  }

  private def sendNumericOrIgnore(key: String, value: Any, now: Long) {
    // seriously nothing better than this? (without Any => String => Num)
    value match {
      case v: Int    ⇒ send(key, v, now)
      case v: Long   ⇒ send(key, v, now)
      case v: Byte   ⇒ send(key, v, now)
      case v: Short  ⇒ send(key, v, now)
      case v: Float  ⇒ send(key, v, now)
      case v: Double ⇒ send(key, v, now)
      case _         ⇒ // ignore non-numeric metric...
    }
  }

  private def send(key: String, value: Double, now: Long) {
    if (value >= 0)
      graphite.send(s"$prefix.$key", "%2.2f".format(value), now)
  }

  private def send(key: String, value: Long, now: Long) {
    if (value >= 0)
      graphite.send(s"$prefix.$key", value.toString, now)
  }

  private def sendWithBanner(s: String, c: Char) {
    print(s)
    print(' ')
    var i: Int = 0
    while (i < (ConsoleWidth - s.length - 1)) {
      print(c)
      i += 1
    }
    println()
  }

}