Merge pull request #294 from jozic/scale

Scale

9 files changed, 175 insertions, 12 deletions

diff --git a/kamon-core/src/main/scala/kamon/metric/EntityRecorder.scala b/kamon-core/src/main/scala/kamon/metric/EntityRecorder.scala
index 15e20d1a..71b40ea6 100644
--- a/kamon-core/src/main/scala/kamon/metric/EntityRecorder.scala
+++ b/kamon-core/src/main/scala/kamon/metric/EntityRecorder.scala
@@ -89,7 +89,7 @@ case class GaugeRecorder(key: MetricKey, instrument: Gauge) extends SingleInstru
 
 /**
  *  Base class with plenty of utility methods to facilitate the creation of [[EntityRecorder]] implementations.
- *  It is not required to use this base class for defining custom a custom [[EntityRecorder]], but it is certainly
+ *  It is not required to use this base class for defining a custom [[EntityRecorder]], but it is certainly
  *  the most convenient way to do it and the preferred approach throughout the Kamon codebase.
  */
 abstract class GenericEntityRecorder(instrumentFactory: InstrumentFactory) extends EntityRecorder {
diff --git a/kamon-core/src/main/scala/kamon/metric/MetricScaleDecorator.scala b/kamon-core/src/main/scala/kamon/metric/MetricScaleDecorator.scala
new file mode 100644
index 00000000..e096429d
--- /dev/null
+++ b/kamon-core/src/main/scala/kamon/metric/MetricScaleDecorator.scala
@@ -0,0 +1,55 @@
+/*
+ * =========================================================================================
+ * Copyright © 2013-2015 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 kamon.metric
+
+import akka.actor.{ Actor, ActorRef, Props }
+import kamon.metric.SubscriptionsDispatcher.TickMetricSnapshot
+import kamon.metric.instrument._
+
+/**
+ * Can be used as a decorator to scale TickMetricSnapshot messages to given `timeUnits` and/or `memoryUnits`
+ * before forwarding to original receiver
+ * @param timeUnits Optional time units to scale time metrics to
+ * @param memoryUnits Optional memory units to scale memory metrics to
+ * @param receiver Receiver of scaled metrics snapshot, usually a backend sender
+ */
+class MetricScaleDecorator(timeUnits: Option[Time], memoryUnits: Option[Memory], receiver: ActorRef) extends Actor {
+  require(timeUnits.isDefined || memoryUnits.isDefined,
+    "Use MetricScaleDecorator only when any of units is defined")
+
+  override def receive: Receive = {
+    case tick: TickMetricSnapshot ⇒
+      val scaled = tick.copy(metrics = tick.metrics.mapValues { entitySnapshot ⇒
+        new DefaultEntitySnapshot(entitySnapshot.metrics.map {
+          case (metricKey, metricSnapshot) ⇒
+            val scaledSnapshot = (metricKey.unitOfMeasurement, timeUnits, memoryUnits) match {
+              case (time: Time, Some(to), _)     ⇒ metricSnapshot.scale(time, to)
+              case (memory: Memory, _, Some(to)) ⇒ metricSnapshot.scale(memory, to)
+              case _                             ⇒ metricSnapshot
+            }
+            metricKey -> scaledSnapshot
+        })
+      })
+      receiver forward scaled
+  }
+}
+
+object MetricScaleDecorator {
+  def props(timeUnits: Option[Time], memoryUnits: Option[Memory], receiver: ActorRef): Props =
+    Props(new MetricScaleDecorator(timeUnits, memoryUnits, receiver))
+}
+
diff --git a/kamon-core/src/main/scala/kamon/metric/instrument/Counter.scala b/kamon-core/src/main/scala/kamon/metric/instrument/Counter.scala
index c1b69cbe..349a12bd 100644
--- a/kamon-core/src/main/scala/kamon/metric/instrument/Counter.scala
+++ b/kamon-core/src/main/scala/kamon/metric/instrument/Counter.scala
@@ -33,6 +33,7 @@ object Counter {
   trait Snapshot extends InstrumentSnapshot {
     def count: Long
     def merge(that: InstrumentSnapshot, context: CollectionContext): Counter.Snapshot
+    def scale(from: UnitOfMeasurement, to: UnitOfMeasurement): Counter.Snapshot
   }
 }
 
@@ -57,4 +58,8 @@ case class CounterSnapshot(count: Long) extends Counter.Snapshot {
     case CounterSnapshot(thatCount) ⇒ CounterSnapshot(count + thatCount)
     case other                      ⇒ sys.error(s"Cannot merge a CounterSnapshot with the incompatible [${other.getClass.getName}] type.")
   }
+
+  override def scale(from: UnitOfMeasurement, to: UnitOfMeasurement): Counter.Snapshot =
+    CounterSnapshot(from.tryScale(to)(count).toLong)
+
 }
\ No newline at end of file
diff --git a/kamon-core/src/main/scala/kamon/metric/instrument/Histogram.scala b/kamon-core/src/main/scala/kamon/metric/instrument/Histogram.scala
index d2fb88a2..dc9a4bbf 100644
--- a/kamon-core/src/main/scala/kamon/metric/instrument/Histogram.scala
+++ b/kamon-core/src/main/scala/kamon/metric/instrument/Histogram.scala
@@ -85,6 +85,39 @@ object Histogram {
     def recordsIterator: Iterator[Record]
     def merge(that: InstrumentSnapshot, context: CollectionContext): InstrumentSnapshot
     def merge(that: Histogram.Snapshot, context: CollectionContext): Histogram.Snapshot
+
+    override def scale(from: UnitOfMeasurement, to: UnitOfMeasurement): Histogram.Snapshot =
+      new ScaledSnapshot(from, to, this)
+  }
+
+  class ScaledSnapshot(from: UnitOfMeasurement, to: UnitOfMeasurement, snapshot: Snapshot) extends Snapshot {
+    private def doScale(v: Long) = from.tryScale(to)(v).toLong
+    override def numberOfMeasurements: Long = snapshot.numberOfMeasurements
+
+    override def max: Long = doScale(snapshot.max)
+
+    override def merge(that: InstrumentSnapshot, context: CollectionContext): InstrumentSnapshot = snapshot.merge(that, context)
+
+    override def merge(that: Snapshot, context: CollectionContext): Snapshot = snapshot.merge(that, context)
+
+    override def percentile(percentile: Double): Long = doScale(snapshot.percentile(percentile))
+
+    override def min: Long = doScale(snapshot.min)
+
+    override def sum: Long = doScale(snapshot.sum)
+
+    override def recordsIterator: Iterator[Record] = {
+      snapshot.recordsIterator.map(record ⇒ new Record {
+        override def count: Long = record.count
+
+        override def level: Long = doScale(record.level)
+
+        override private[kamon] def rawCompactRecord: Long = record.rawCompactRecord
+      })
+    }
+
+    override def scale(from: UnitOfMeasurement, to: UnitOfMeasurement): Histogram.Snapshot =
+      if (this.from == from && this.to == to) this else super.scale(from, to)
   }
 
   object Snapshot {
@@ -97,6 +130,7 @@ object Histogram {
       override def merge(that: InstrumentSnapshot, context: CollectionContext): InstrumentSnapshot = that
       override def merge(that: Histogram.Snapshot, context: CollectionContext): Histogram.Snapshot = that
       override def numberOfMeasurements: Long = 0L
+      override def scale(from: UnitOfMeasurement, to: UnitOfMeasurement): Histogram.Snapshot = this
     }
   }
 }
diff --git a/kamon-core/src/main/scala/kamon/metric/instrument/Instrument.scala b/kamon-core/src/main/scala/kamon/metric/instrument/Instrument.scala
index 089dbeec..2c4b4319 100644
--- a/kamon-core/src/main/scala/kamon/metric/instrument/Instrument.scala
+++ b/kamon-core/src/main/scala/kamon/metric/instrument/Instrument.scala
@@ -18,10 +18,6 @@ package kamon.metric.instrument
 
 import java.nio.LongBuffer
 
-import akka.actor.{ Scheduler, Cancellable }
-import akka.dispatch.MessageDispatcher
-import scala.concurrent.duration.FiniteDuration
-
 private[kamon] trait Instrument {
   type SnapshotType <: InstrumentSnapshot
 
@@ -31,6 +27,8 @@ private[kamon] trait Instrument {
 
 trait InstrumentSnapshot {
   def merge(that: InstrumentSnapshot, context: CollectionContext): InstrumentSnapshot
+
+  def scale(from: UnitOfMeasurement, to: UnitOfMeasurement): InstrumentSnapshot
 }
 
 trait CollectionContext {
diff --git a/kamon-core/src/main/scala/kamon/metric/instrument/InstrumentSettings.scala b/kamon-core/src/main/scala/kamon/metric/instrument/InstrumentSettings.scala
index 4423964a..f7516262 100644
--- a/kamon-core/src/main/scala/kamon/metric/instrument/InstrumentSettings.scala
+++ b/kamon-core/src/main/scala/kamon/metric/instrument/InstrumentSettings.scala
@@ -1,7 +1,5 @@
 package kamon.metric.instrument
 
-import java.util.concurrent.TimeUnit
-
 import com.typesafe.config.Config
 import kamon.metric.instrument.Histogram.DynamicRange
 
diff --git a/kamon-core/src/main/scala/kamon/metric/instrument/UnitOfMeasurement.scala b/kamon-core/src/main/scala/kamon/metric/instrument/UnitOfMeasurement.scala
index c5a1b81a..5952b906 100644
--- a/kamon-core/src/main/scala/kamon/metric/instrument/UnitOfMeasurement.scala
+++ b/kamon-core/src/main/scala/kamon/metric/instrument/UnitOfMeasurement.scala
@@ -22,14 +22,27 @@ package kamon.metric.instrument
  *  recorders and might be used to scale certain kinds of measurements in metric backends.
  */
 trait UnitOfMeasurement {
+  type U <: UnitOfMeasurement
+
   def name: String
   def label: String
+  def scale(toUnit: U)(value: Double): Double = value
+
+  def tryScale(toUnit: UnitOfMeasurement)(value: Double): Double =
+    if (canScale(toUnit)) scale(toUnit.asInstanceOf[U])(value)
+    else throw new IllegalArgumentException(s"Can't scale different types of units `$name` and `${toUnit.name}`")
+
+  protected def canScale(toUnit: UnitOfMeasurement): Boolean
+
 }
 
 object UnitOfMeasurement {
   case object Unknown extends UnitOfMeasurement {
+    override type U = Unknown.type
     val name = "unknown"
     val label = "unknown"
+
+    override protected def canScale(toUnit: UnitOfMeasurement): Boolean = UnitOfMeasurement.isUnknown(toUnit)
   }
 
   def isUnknown(uom: UnitOfMeasurement): Boolean =
@@ -47,10 +60,13 @@ object UnitOfMeasurement {
  *  UnitOfMeasurement representing time.
  */
 case class Time(factor: Double, label: String) extends UnitOfMeasurement {
+  override type U = Time
   val name = "time"
 
-  def scale(toUnit: Time)(value: Double): Double =
+  override def scale(toUnit: Time)(value: Double): Double =
     (value * factor) / toUnit.factor
+
+  override protected def canScale(toUnit: UnitOfMeasurement): Boolean = UnitOfMeasurement.isTime(toUnit)
 }
 
 object Time {
@@ -58,22 +74,36 @@ object Time {
   val Microseconds = Time(1E-6, "µs")
   val Milliseconds = Time(1E-3, "ms")
   val Seconds = Time(1, "s")
+
+  val units = List(Nanoseconds, Microseconds, Milliseconds, Seconds)
+
+  def apply(time: String): Time = units.find(_.label.toLowerCase == time.toLowerCase) getOrElse {
+    throw new IllegalArgumentException(s"Can't recognize time unit '$time'")
+  }
 }
 
 /**
  *  UnitOfMeasurement representing computer memory space.
  */
 case class Memory(factor: Double, label: String) extends UnitOfMeasurement {
+  override type U = Memory
   val name = "bytes"
 
-  def scale(toUnit: Memory)(value: Double): Double =
+  override def scale(toUnit: Memory)(value: Double): Double =
     (value * factor) / toUnit.factor
+
+  override protected def canScale(toUnit: UnitOfMeasurement): Boolean = UnitOfMeasurement.isMemory(toUnit)
 }
 
 object Memory {
   val Bytes = Memory(1, "b")
   val KiloBytes = Memory(1024, "Kb")
-  val MegaBytes = Memory(1024E2, "Mb")
-  val GigaBytes = Memory(1024E3, "Gb")
-}
+  val MegaBytes = Memory(1024 * 1024, "Mb")
+  val GigaBytes = Memory(1024 * 1024 * 1024, "Gb")
 
+  val units = List(Bytes, KiloBytes, MegaBytes, GigaBytes)
+
+  def apply(memory: String): Memory = units.find(_.label.toLowerCase == memory.toLowerCase) getOrElse {
+    throw new IllegalArgumentException(s"Can't recognize memory unit '$memory'")
+  }
+}
diff --git a/kamon-core/src/main/scala/kamon/util/ConfigTools.scala b/kamon-core/src/main/scala/kamon/util/ConfigTools.scala
index bcec22c3..d0665764 100644
--- a/kamon-core/src/main/scala/kamon/util/ConfigTools.scala
+++ b/kamon-core/src/main/scala/kamon/util/ConfigTools.scala
@@ -22,6 +22,8 @@ import com.typesafe.config.Config
 
 import scala.concurrent.duration.FiniteDuration
 
+import kamon.metric.instrument.{ Memory, Time }
+
 object ConfigTools {
   implicit class Syntax(val config: Config) extends AnyVal {
     // We are using the deprecated .getNanoseconds option to keep Kamon source code compatible with
@@ -37,6 +39,10 @@ object ConfigTools {
         case entry ⇒ entry.getKey.takeWhile(_ != '.')
       } toSet
     }
+
+    def time(path: String): Time = Time(config.getString(path))
+
+    def memory(path: String): Memory = Memory(config.getString(path))
   }
 
 }
diff --git a/kamon-core/src/main/scala/kamon/util/NeedToScale.scala b/kamon-core/src/main/scala/kamon/util/NeedToScale.scala
new file mode 100644
index 00000000..19e1ae06
--- /dev/null
+++ b/kamon-core/src/main/scala/kamon/util/NeedToScale.scala
@@ -0,0 +1,37 @@
+/*
+ * =========================================================================================
+ * Copyright © 2013-2015 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 kamon.util
+
+import com.typesafe.config.Config
+import kamon.metric.instrument.{ Memory, Time }
+import kamon.util.ConfigTools._
+
+object NeedToScale {
+  val TimeUnits = "time-units"
+  val MemoryUnits = "memory-units"
+
+  def unapply(config: Config): Option[(Option[Time], Option[Memory])] = {
+    val scaleTimeTo: Option[Time] =
+      if (config.hasPath(TimeUnits)) Some(config.time(TimeUnits)) else None
+
+    val scaleMemoryTo: Option[Memory] =
+      if (config.hasPath(MemoryUnits)) Some(config.memory(MemoryUnits)) else None
+    if (scaleTimeTo.isDefined || scaleMemoryTo.isDefined) Some(scaleTimeTo -> scaleMemoryTo)
+    else None
+  }
+}
+