Merge branch 'master' into release-0.2

Conflicts:
	version.sbt

6 files changed, 250 insertions, 206 deletions

diff --git a/kamon-core/src/main/scala/kamon/instrumentation/akka/ActorCellInstrumentation.scala b/kamon-core/src/main/scala/kamon/instrumentation/akka/ActorCellInstrumentation.scala
index 5fce4555..446bc487 100644
--- a/kamon-core/src/main/scala/kamon/instrumentation/akka/ActorCellInstrumentation.scala
+++ b/kamon-core/src/main/scala/kamon/instrumentation/akka/ActorCellInstrumentation.scala
@@ -82,7 +82,6 @@ class ActorCellInstrumentation {
     val cellWithMetrics = cell.asInstanceOf[ActorCellMetrics]
 
     cellWithMetrics.actorMetricsRecorder.map { p ⇒
-      cellWithMetrics.mailboxSizeCollectorCancellable.cancel()
       Kamon(Metrics)(cell.system).unregister(cellWithMetrics.metricIdentity)
     }
   }
@@ -103,7 +102,6 @@ class ActorCellInstrumentation {
 trait ActorCellMetrics {
   var metricIdentity: ActorMetrics = _
   var actorMetricsRecorder: Option[ActorMetricsRecorder] = _
-  var mailboxSizeCollectorCancellable: Cancellable = _
 }
 
 @Aspect
diff --git a/kamon-core/src/main/scala/kamon/metric/MetricsExtension.scala b/kamon-core/src/main/scala/kamon/metric/MetricsExtension.scala
index 7febb0ec..05bab175 100644
--- a/kamon-core/src/main/scala/kamon/metric/MetricsExtension.scala
+++ b/kamon-core/src/main/scala/kamon/metric/MetricsExtension.scala
@@ -52,7 +52,7 @@ class MetricsExtension(system: ExtendedActorSystem) extends Kamon.Extension {
   }
 
   def unregister(identity: MetricGroupIdentity): Unit = {
-    storage.remove(identity)
+    storage.remove(identity).map(_.cleanup)
   }
 
   def subscribe[C <: MetricGroupCategory](category: C, selection: String, subscriber: ActorRef, permanently: Boolean = false): Unit =
diff --git a/kamon-core/src/main/scala/kamon/metric/UserMetrics.scala b/kamon-core/src/main/scala/kamon/metric/UserMetrics.scala
index f3803d37..b511b4bc 100644
--- a/kamon-core/src/main/scala/kamon/metric/UserMetrics.scala
+++ b/kamon-core/src/main/scala/kamon/metric/UserMetrics.scala
@@ -1,163 +1,188 @@
 package kamon.metric
 
 import akka.actor
-import akka.actor.{ ActorSystem, ExtendedActorSystem, ExtensionIdProvider, ExtensionId }
-import com.typesafe.config.Config
+import akka.actor.{ ExtendedActorSystem, ExtensionIdProvider, ExtensionId }
 import kamon.Kamon
 import kamon.metric.instrument.{ Gauge, MinMaxCounter, Counter, Histogram }
 
-import scala.collection.concurrent.TrieMap
 import scala.concurrent.duration.FiniteDuration
 
 class UserMetricsExtension(system: ExtendedActorSystem) extends Kamon.Extension {
-  lazy val userMetricsRecorder = Kamon(Metrics)(system).register(UserMetrics, UserMetrics.Factory).get
+  import UserMetrics._
 
-  def registerHistogram(name: String, precision: Histogram.Precision, highestTrackableValue: Long): Histogram =
-    userMetricsRecorder.buildHistogram(name, precision, highestTrackableValue)
+  lazy val metricsExtension = Kamon(Metrics)(system)
+  val precisionConfig = system.settings.config.getConfig("kamon.metrics.precision")
 
-  def registerHistogram(name: String): Histogram =
-    userMetricsRecorder.buildHistogram(name)
+  val defaultHistogramPrecisionConfig = precisionConfig.getConfig("default-histogram-precision")
+  val defaultMinMaxCounterPrecisionConfig = precisionConfig.getConfig("default-min-max-counter-precision")
+  val defaultGaugePrecisionConfig = precisionConfig.getConfig("default-gauge-precision")
 
-  def registerCounter(name: String): Counter =
-    userMetricsRecorder.buildCounter(name)
+  def registerHistogram(name: String, precision: Histogram.Precision, highestTrackableValue: Long): Histogram = {
+    metricsExtension.storage.getOrElseUpdate(UserHistogram(name), {
+      UserHistogramRecorder(Histogram(highestTrackableValue, precision, Scale.Unit))
+    }).asInstanceOf[UserHistogramRecorder].histogram
+  }
+
+  def registerHistogram(name: String): Histogram = {
+    metricsExtension.storage.getOrElseUpdate(UserHistogram(name), {
+      UserHistogramRecorder(Histogram.fromConfig(defaultHistogramPrecisionConfig))
+    }).asInstanceOf[UserHistogramRecorder].histogram
+  }
+
+  def registerCounter(name: String): Counter = {
+    metricsExtension.storage.getOrElseUpdate(UserCounter(name), {
+      UserCounterRecorder(Counter())
+    }).asInstanceOf[UserCounterRecorder].counter
+  }
 
   def registerMinMaxCounter(name: String, precision: Histogram.Precision, highestTrackableValue: Long,
     refreshInterval: FiniteDuration): MinMaxCounter = {
-    userMetricsRecorder.buildMinMaxCounter(name, precision, highestTrackableValue, refreshInterval)
+    metricsExtension.storage.getOrElseUpdate(UserMinMaxCounter(name), {
+      UserMinMaxCounterRecorder(MinMaxCounter(highestTrackableValue, precision, Scale.Unit, refreshInterval, system))
+    }).asInstanceOf[UserMinMaxCounterRecorder].minMaxCounter
   }
 
-  def registerMinMaxCounter(name: String): MinMaxCounter =
-    userMetricsRecorder.buildMinMaxCounter(name)
+  def registerMinMaxCounter(name: String): MinMaxCounter = {
+    metricsExtension.storage.getOrElseUpdate(UserMinMaxCounter(name), {
+      UserMinMaxCounterRecorder(MinMaxCounter.fromConfig(defaultMinMaxCounterPrecisionConfig, system))
+    }).asInstanceOf[UserMinMaxCounterRecorder].minMaxCounter
+  }
 
-  def registerGauge(name: String)(currentValueCollector: Gauge.CurrentValueCollector): Gauge =
-    userMetricsRecorder.buildGauge(name)(currentValueCollector)
+  def registerGauge(name: String)(currentValueCollector: Gauge.CurrentValueCollector): Gauge = {
+    metricsExtension.storage.getOrElseUpdate(UserGauge(name), {
+      UserGaugeRecorder(Gauge.fromConfig(defaultGaugePrecisionConfig, system)(currentValueCollector))
+    }).asInstanceOf[UserGaugeRecorder].gauge
+  }
 
   def registerGauge(name: String, precision: Histogram.Precision, highestTrackableValue: Long,
-    refreshInterval: FiniteDuration)(currentValueCollector: Gauge.CurrentValueCollector): Gauge =
-    userMetricsRecorder.buildGauge(name, precision, highestTrackableValue, refreshInterval, currentValueCollector)
+    refreshInterval: FiniteDuration)(currentValueCollector: Gauge.CurrentValueCollector): Gauge = {
+    metricsExtension.storage.getOrElseUpdate(UserGauge(name), {
+      UserGaugeRecorder(Gauge(precision, highestTrackableValue, Scale.Unit, refreshInterval, system)(currentValueCollector))
+    }).asInstanceOf[UserGaugeRecorder].gauge
+  }
 
   def removeHistogram(name: String): Unit =
-    userMetricsRecorder.removeHistogram(name)
+    metricsExtension.unregister(UserHistogram(name))
 
   def removeCounter(name: String): Unit =
-    userMetricsRecorder.removeCounter(name)
+    metricsExtension.unregister(UserCounter(name))
 
   def removeMinMaxCounter(name: String): Unit =
-    userMetricsRecorder.removeMinMaxCounter(name)
+    metricsExtension.unregister(UserMinMaxCounter(name))
 
   def removeGauge(name: String): Unit =
-    userMetricsRecorder.removeGauge(name)
+    metricsExtension.unregister(UserGauge(name))
 }
 
-object UserMetrics extends ExtensionId[UserMetricsExtension] with ExtensionIdProvider with MetricGroupIdentity {
+object UserMetrics extends ExtensionId[UserMetricsExtension] with ExtensionIdProvider {
   def lookup(): ExtensionId[_ <: actor.Extension] = Metrics
+
   def createExtension(system: ExtendedActorSystem): UserMetricsExtension = new UserMetricsExtension(system)
 
-  val name: String = "user-metrics-recorder"
-  val category = new MetricGroupCategory {
-    val name: String = "user-metrics"
+  sealed trait UserMetricGroup
+  //
+  // Histograms
+  //
+
+  case class UserHistogram(name: String) extends MetricGroupIdentity with UserMetricGroup {
+    val category = UserHistograms
   }
 
-  val Factory = new MetricGroupFactory {
-    type GroupRecorder = UserMetricsRecorder
-    def create(config: Config, system: ActorSystem): UserMetricsRecorder = new UserMetricsRecorder(system)
+  case class UserHistogramRecorder(histogram: Histogram) extends MetricGroupRecorder {
+    def collect(context: CollectionContext): MetricGroupSnapshot =
+      UserHistogramSnapshot(histogram.collect(context))
+
+    def cleanup: Unit = histogram.cleanup
   }
 
-  class UserMetricsRecorder(system: ActorSystem) extends MetricGroupRecorder {
-    val precisionConfig = system.settings.config.getConfig("kamon.metrics.precision")
-    val defaultHistogramPrecisionConfig = precisionConfig.getConfig("default-histogram-precision")
-    val defaultMinMaxCounterPrecisionConfig = precisionConfig.getConfig("default-min-max-counter-precision")
-    val defaultGaugePrecisionConfig = precisionConfig.getConfig("default-gauge-precision")
+  case class UserHistogramSnapshot(histogramSnapshot: Histogram.Snapshot) extends MetricGroupSnapshot {
+    type GroupSnapshotType = UserHistogramSnapshot
 
-    val histograms = TrieMap[String, Histogram]()
-    val counters = TrieMap[String, Counter]()
-    val minMaxCounters = TrieMap[String, MinMaxCounter]()
-    val gauges = TrieMap[String, Gauge]()
+    def merge(that: UserHistogramSnapshot, context: CollectionContext): UserHistogramSnapshot =
+      UserHistogramSnapshot(that.histogramSnapshot.merge(histogramSnapshot, context))
 
-    def buildHistogram(name: String, precision: Histogram.Precision, highestTrackableValue: Long): Histogram =
-      histograms.getOrElseUpdate(name, Histogram(highestTrackableValue, precision, Scale.Unit))
+    def metrics: Map[MetricIdentity, MetricSnapshot] = Map((RecordedValues, histogramSnapshot))
+  }
 
-    def buildHistogram(name: String): Histogram =
-      histograms.getOrElseUpdate(name, Histogram.fromConfig(defaultHistogramPrecisionConfig))
+  //
+  // Counters
+  //
 
-    def buildCounter(name: String): Counter =
-      counters.getOrElseUpdate(name, Counter())
+  case class UserCounter(name: String) extends MetricGroupIdentity with UserMetricGroup {
+    val category = UserCounters
+  }
 
-    def buildMinMaxCounter(name: String, precision: Histogram.Precision, highestTrackableValue: Long,
-      refreshInterval: FiniteDuration): MinMaxCounter = {
-      minMaxCounters.getOrElseUpdate(name, MinMaxCounter(highestTrackableValue, precision, Scale.Unit, refreshInterval, system))
-    }
+  case class UserCounterRecorder(counter: Counter) extends MetricGroupRecorder {
+    def collect(context: CollectionContext): MetricGroupSnapshot =
+      UserCounterSnapshot(counter.collect(context))
 
-    def buildMinMaxCounter(name: String): MinMaxCounter =
-      minMaxCounters.getOrElseUpdate(name, MinMaxCounter.fromConfig(defaultMinMaxCounterPrecisionConfig, system))
+    def cleanup: Unit = counter.cleanup
+  }
 
-    def buildGauge(name: String, precision: Histogram.Precision, highestTrackableValue: Long,
-      refreshInterval: FiniteDuration, currentValueCollector: Gauge.CurrentValueCollector): Gauge =
-      gauges.getOrElseUpdate(name, Gauge(precision, highestTrackableValue, Scale.Unit, refreshInterval, system)(currentValueCollector))
+  case class UserCounterSnapshot(counterSnapshot: Counter.Snapshot) extends MetricGroupSnapshot {
+    type GroupSnapshotType = UserCounterSnapshot
 
-    def buildGauge(name: String)(currentValueCollector: Gauge.CurrentValueCollector): Gauge =
-      gauges.getOrElseUpdate(name, Gauge.fromConfig(defaultGaugePrecisionConfig, system)(currentValueCollector))
+    def merge(that: UserCounterSnapshot, context: CollectionContext): UserCounterSnapshot =
+      UserCounterSnapshot(that.counterSnapshot.merge(counterSnapshot, context))
 
-    def removeHistogram(name: String): Unit =
-      histograms.remove(name)
+    def metrics: Map[MetricIdentity, MetricSnapshot] = Map((Count, counterSnapshot))
+  }
 
-    def removeCounter(name: String): Unit =
-      counters.remove(name)
+  //
+  // MinMaxCounters
+  //
 
-    def removeMinMaxCounter(name: String): Unit =
-      minMaxCounters.remove(name).map(_.cleanup)
+  case class UserMinMaxCounter(name: String) extends MetricGroupIdentity with UserMetricGroup {
+    val category = UserMinMaxCounters
+  }
 
-    def removeGauge(name: String): Unit =
-      gauges.remove(name).map(_.cleanup)
+  case class UserMinMaxCounterRecorder(minMaxCounter: MinMaxCounter) extends MetricGroupRecorder {
+    def collect(context: CollectionContext): MetricGroupSnapshot =
+      UserMinMaxCounterSnapshot(minMaxCounter.collect(context))
 
-    def collect(context: CollectionContext): UserMetricsSnapshot = {
-      val histogramSnapshots = histograms.map {
-        case (name, histogram) ⇒
-          (UserHistogram(name), histogram.collect(context))
-      } toMap
+    def cleanup: Unit = minMaxCounter.cleanup
+  }
 
-      val counterSnapshots = counters.map {
-        case (name, counter) ⇒
-          (UserCounter(name), counter.collect(context))
-      } toMap
+  case class UserMinMaxCounterSnapshot(minMaxCounterSnapshot: Histogram.Snapshot) extends MetricGroupSnapshot {
+    type GroupSnapshotType = UserMinMaxCounterSnapshot
 
-      val minMaxCounterSnapshots = minMaxCounters.map {
-        case (name, minMaxCounter) ⇒
-          (UserMinMaxCounter(name), minMaxCounter.collect(context))
-      } toMap
+    def merge(that: UserMinMaxCounterSnapshot, context: CollectionContext): UserMinMaxCounterSnapshot =
+      UserMinMaxCounterSnapshot(that.minMaxCounterSnapshot.merge(minMaxCounterSnapshot, context))
 
-      val gaugeSnapshots = gauges.map {
-        case (name, gauge) ⇒
-          (UserGauge(name), gauge.collect(context))
-      } toMap
+    def metrics: Map[MetricIdentity, MetricSnapshot] = Map((RecordedValues, minMaxCounterSnapshot))
+  }
 
-      UserMetricsSnapshot(histogramSnapshots, counterSnapshots, minMaxCounterSnapshots, gaugeSnapshots)
-    }
+  //
+  // Gauges
+  //
 
-    def cleanup: Unit = {}
+  case class UserGauge(name: String) extends MetricGroupIdentity with UserMetricGroup {
+    val category = UserGauges
   }
 
-  case class UserHistogram(name: String) extends MetricIdentity
-  case class UserCounter(name: String) extends MetricIdentity
-  case class UserMinMaxCounter(name: String) extends MetricIdentity
-  case class UserGauge(name: String) extends MetricIdentity
+  case class UserGaugeRecorder(gauge: Gauge) extends MetricGroupRecorder {
+    def collect(context: CollectionContext): MetricGroupSnapshot =
+      UserGaugeSnapshot(gauge.collect(context))
 
-  case class UserMetricsSnapshot(histograms: Map[UserHistogram, Histogram.Snapshot],
-    counters: Map[UserCounter, Counter.Snapshot],
-    minMaxCounters: Map[UserMinMaxCounter, Histogram.Snapshot],
-    gauges: Map[UserGauge, Histogram.Snapshot])
-      extends MetricGroupSnapshot {
+    def cleanup: Unit = gauge.cleanup
+  }
 
-    type GroupSnapshotType = UserMetricsSnapshot
+  case class UserGaugeSnapshot(gaugeSnapshot: Histogram.Snapshot) extends MetricGroupSnapshot {
+    type GroupSnapshotType = UserGaugeSnapshot
 
-    def merge(that: UserMetricsSnapshot, context: CollectionContext): UserMetricsSnapshot =
-      UserMetricsSnapshot(
-        combineMaps(histograms, that.histograms)((l, r) ⇒ l.merge(r, context)),
-        combineMaps(counters, that.counters)((l, r) ⇒ l.merge(r, context)),
-        combineMaps(minMaxCounters, that.minMaxCounters)((l, r) ⇒ l.merge(r, context)),
-        combineMaps(gauges, that.gauges)((l, r) ⇒ l.merge(r, context)))
+    def merge(that: UserGaugeSnapshot, context: CollectionContext): UserGaugeSnapshot =
+      UserGaugeSnapshot(that.gaugeSnapshot.merge(gaugeSnapshot, context))
 
-    def metrics: Map[MetricIdentity, MetricSnapshot] = histograms ++ counters ++ minMaxCounters ++ gauges
+    def metrics: Map[MetricIdentity, MetricSnapshot] = Map((RecordedValues, gaugeSnapshot))
   }
 
+  case object UserHistograms extends MetricGroupCategory { val name: String = "histogram" }
+  case object UserCounters extends MetricGroupCategory { val name: String = "counter" }
+  case object UserMinMaxCounters extends MetricGroupCategory { val name: String = "min-max-counter" }
+  case object UserGauges extends MetricGroupCategory { val name: String = "gauge" }
+
+  case object RecordedValues extends MetricIdentity { val name: String = "values" }
+  case object Count extends MetricIdentity { val name: String = "count" }
+
 }
+
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 7d8022f7..67db5d93 100644
--- a/kamon-core/src/main/scala/kamon/metric/instrument/Histogram.scala
+++ b/kamon-core/src/main/scala/kamon/metric/instrument/Histogram.scala
@@ -145,7 +145,7 @@ class HdrHistogram(lowestTrackableValue: Long, highestTrackableValue: Long, sign
 
   private def reestablishTotalCount(diff: Long): Unit = {
     def tryUpdateTotalCount: Boolean = {
-      val previousTotalCount = getTotalCount
+      val previousTotalCount = totalCountUpdater.get(this)
       val newTotalCount = previousTotalCount - diff
 
       totalCountUpdater.compareAndSet(this, previousTotalCount, newTotalCount)
diff --git a/kamon-core/src/test/scala/kamon/metric/ActorMetricsSpec.scala b/kamon-core/src/test/scala/kamon/metric/ActorMetricsSpec.scala
index 0029fd7c..21e0bbba 100644
--- a/kamon-core/src/test/scala/kamon/metric/ActorMetricsSpec.scala
+++ b/kamon-core/src/test/scala/kamon/metric/ActorMetricsSpec.scala
@@ -18,6 +18,7 @@ package kamon.metric
 import java.nio.LongBuffer
 
 import akka.instrumentation.ActorCellMetrics
+import kamon.Kamon
 import kamon.metric.ActorMetricsTestActor._
 import org.scalatest.{ WordSpecLike, Matchers }
 import akka.testkit.{ ImplicitSender, TestProbe, TestKitBase }
@@ -37,7 +38,7 @@ class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
       |  filters = [
       |    {
       |      actor {
-      |        includes = [ "user/tracked-*", "user/measuring-*", "user/clean-after-collect" ]
+      |        includes = [ "user/tracked-*", "user/measuring-*", "user/clean-after-collect", "user/stop" ]
       |        excludes = [ "user/tracked-explicitly-excluded"]
       |      }
       |    }
@@ -49,7 +50,7 @@ class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
       |    }
       |
       |    default-min-max-counter-precision {
-      |      refresh-interval = 1 second
+      |      refresh-interval = 1 hour
       |      highest-trackable-value = 999999999
       |      significant-value-digits = 2
       |    }
@@ -87,9 +88,9 @@ class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
 
       val secondSnapshot = takeSnapshotOf(trackedActorMetrics) // Ensure that the recorders are clean
       secondSnapshot.errors.count should be(0L)
-      secondSnapshot.mailboxSize.numberOfMeasurements should be <= 3L
-      secondSnapshot.processingTime.numberOfMeasurements should be(0L) // 102 examples + Initialize message
-      secondSnapshot.timeInMailbox.numberOfMeasurements should be(0L) // 102 examples + Initialize message
+      secondSnapshot.mailboxSize.numberOfMeasurements should be(3L) // min, max and current
+      secondSnapshot.processingTime.numberOfMeasurements should be(0L)
+      secondSnapshot.timeInMailbox.numberOfMeasurements should be(0L)
     }
 
     "record the processing-time of the receive function" in new ActorMetricsFixtures {
@@ -151,6 +152,19 @@ class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
       snapshot.timeInMailbox.recordsIterator.next().count should be(1L)
       snapshot.timeInMailbox.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")
+      actorMetricsRecorderOf(trackedActor).get // force the actor to be initialized
+      Kamon(Metrics).storage.get(ActorMetrics("user/stop")) should not be empty
+
+      val deathWatcher = TestProbe()
+      deathWatcher.watch(trackedActor)
+      trackedActor ! PoisonPill
+      deathWatcher.expectTerminated(trackedActor)
+
+      Kamon(Metrics).storage.get(ActorMetrics("user/stop")) shouldBe empty
+    }
   }
 
   trait ActorMetricsFixtures {
diff --git a/kamon-core/src/test/scala/kamon/metric/UserMetricsSpec.scala b/kamon-core/src/test/scala/kamon/metric/UserMetricsSpec.scala
index 66ee49ea..e072d3ef 100644
--- a/kamon-core/src/test/scala/kamon/metric/UserMetricsSpec.scala
+++ b/kamon-core/src/test/scala/kamon/metric/UserMetricsSpec.scala
@@ -1,11 +1,12 @@
 package kamon.metric
 
-import akka.actor.ActorSystem
+import akka.actor.{ Props, ActorSystem }
 import akka.testkit.{ ImplicitSender, TestKitBase }
 import com.typesafe.config.ConfigFactory
 import kamon.Kamon
-import kamon.metric.UserMetrics.{ UserGauge, UserMinMaxCounter, UserCounter, UserHistogram }
-import kamon.metric.instrument.Histogram
+import kamon.metric.Subscriptions.TickMetricSnapshot
+import kamon.metric.UserMetrics._
+import kamon.metric.instrument.{ Histogram, Counter, MinMaxCounter, Gauge }
 import kamon.metric.instrument.Histogram.MutableRecord
 import org.scalatest.{ Matchers, WordSpecLike }
 import scala.concurrent.duration._
@@ -113,32 +114,34 @@ class UserMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
       }
     }
 
-    "allow unregistering metrics from the extension" in {
-      val userMetricsRecorder = Kamon(Metrics).register(UserMetrics, UserMetrics.Factory).get
-      val counter = Kamon(UserMetrics).registerCounter("counter-for-remove")
-      val histogram = Kamon(UserMetrics).registerHistogram("histogram-for-remove")
-      val minMaxCounter = Kamon(UserMetrics).registerMinMaxCounter("min-max-counter-for-remove")
-      val gauge = Kamon(UserMetrics).registerGauge("gauge-for-remove") { () ⇒ 2L }
+    "allow un-registering user metrics" in {
+      val metricsExtension = Kamon(Metrics)
+      Kamon(UserMetrics).registerCounter("counter-for-remove")
+      Kamon(UserMetrics).registerHistogram("histogram-for-remove")
+      Kamon(UserMetrics).registerMinMaxCounter("min-max-counter-for-remove")
+      Kamon(UserMetrics).registerGauge("gauge-for-remove") { () ⇒ 2L }
 
-      userMetricsRecorder.counters.keys should contain("counter-for-remove")
-      userMetricsRecorder.histograms.keys should contain("histogram-for-remove")
-      userMetricsRecorder.minMaxCounters.keys should contain("min-max-counter-for-remove")
-      userMetricsRecorder.gauges.keys should contain("gauge-for-remove")
+      metricsExtension.storage.keys should contain(UserCounter("counter-for-remove"))
+      metricsExtension.storage.keys should contain(UserHistogram("histogram-for-remove"))
+      metricsExtension.storage.keys should contain(UserMinMaxCounter("min-max-counter-for-remove"))
+      metricsExtension.storage.keys should contain(UserGauge("gauge-for-remove"))
 
       Kamon(UserMetrics).removeCounter("counter-for-remove")
       Kamon(UserMetrics).removeHistogram("histogram-for-remove")
       Kamon(UserMetrics).removeMinMaxCounter("min-max-counter-for-remove")
       Kamon(UserMetrics).removeGauge("gauge-for-remove")
 
-      userMetricsRecorder.counters.keys should not contain ("counter-for-remove")
-      userMetricsRecorder.histograms.keys should not contain ("histogram-for-remove")
-      userMetricsRecorder.minMaxCounters.keys should not contain ("min-max-counter-for-remove")
-      userMetricsRecorder.gauges.keys should not contain ("gauge-for-remove")
+      metricsExtension.storage.keys should not contain (UserCounter("counter-for-remove"))
+      metricsExtension.storage.keys should not contain (UserHistogram("histogram-for-remove"))
+      metricsExtension.storage.keys should not contain (UserMinMaxCounter("min-max-counter-for-remove"))
+      metricsExtension.storage.keys should not contain (UserGauge("gauge-for-remove"))
     }
 
-    "generate a snapshot containing all the registered user metrics and reset all instruments" in {
-      val context = Kamon(Metrics).buildDefaultCollectionContext
-      val userMetricsRecorder = Kamon(Metrics).register(UserMetrics, UserMetrics.Factory).get
+    "include all the registered metrics in the a tick snapshot and reset all recorders" in {
+      Kamon(Metrics).subscribe(UserHistograms, "*", testActor, permanently = true)
+      Kamon(Metrics).subscribe(UserCounters, "*", testActor, permanently = true)
+      Kamon(Metrics).subscribe(UserMinMaxCounters, "*", testActor, permanently = true)
+      Kamon(Metrics).subscribe(UserGauges, "*", testActor, permanently = true)
 
       val histogramWithSettings = Kamon(UserMetrics).registerHistogram("histogram-with-settings", Histogram.Precision.Normal, 10000L)
       val histogramWithDefaultConfiguration = Kamon(UserMetrics).registerHistogram("histogram-with-default-configuration")
@@ -159,94 +162,93 @@ class UserMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
 
       gauge.record(15)
 
-      val firstSnapshot = userMetricsRecorder.collect(context)
+      Kamon(Metrics).subscriptions ! Subscriptions.FlushMetrics
+      val firstSnapshot = expectMsgType[TickMetricSnapshot].metrics
 
-      firstSnapshot.histograms.size should be(2)
-      firstSnapshot.histograms.keys should contain allOf (
+      firstSnapshot.keys should contain allOf (
         UserHistogram("histogram-with-settings"),
         UserHistogram("histogram-with-default-configuration"))
 
-      firstSnapshot.histograms(UserHistogram("histogram-with-settings")).min shouldBe (10)
-      firstSnapshot.histograms(UserHistogram("histogram-with-settings")).max shouldBe (20)
-      firstSnapshot.histograms(UserHistogram("histogram-with-settings")).numberOfMeasurements should be(101)
-      firstSnapshot.histograms(UserHistogram("histogram-with-settings")).recordsIterator.toStream should contain allOf (
+      firstSnapshot(UserHistogram("histogram-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (10)
+      firstSnapshot(UserHistogram("histogram-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (20)
+      firstSnapshot(UserHistogram("histogram-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(101)
+      firstSnapshot(UserHistogram("histogram-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream should contain allOf (
         MutableRecord(10, 1),
         MutableRecord(20, 100))
 
-      firstSnapshot.histograms(UserHistogram("histogram-with-default-configuration")).min shouldBe (40)
-      firstSnapshot.histograms(UserHistogram("histogram-with-default-configuration")).max shouldBe (40)
-      firstSnapshot.histograms(UserHistogram("histogram-with-default-configuration")).numberOfMeasurements should be(1)
-      firstSnapshot.histograms(UserHistogram("histogram-with-default-configuration")).recordsIterator.toStream should contain only (
+      firstSnapshot(UserHistogram("histogram-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (40)
+      firstSnapshot(UserHistogram("histogram-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (40)
+      firstSnapshot(UserHistogram("histogram-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(1)
+      firstSnapshot(UserHistogram("histogram-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream should contain only (
         MutableRecord(40, 1))
 
-      firstSnapshot.counters(UserCounter("counter")).count should be(17)
+      firstSnapshot(UserCounter("counter")).metrics(Count).asInstanceOf[Counter.Snapshot].count should be(17)
 
-      firstSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-settings")).min shouldBe (0)
-      firstSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-settings")).max shouldBe (43)
-      firstSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-settings")).numberOfMeasurements should be(3)
-      firstSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-settings")).recordsIterator.toStream should contain allOf (
+      firstSnapshot(UserMinMaxCounter("min-max-counter-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (0)
+      firstSnapshot(UserMinMaxCounter("min-max-counter-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (43)
+      firstSnapshot(UserMinMaxCounter("min-max-counter-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(3)
+      firstSnapshot(UserMinMaxCounter("min-max-counter-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream should contain allOf (
         MutableRecord(0, 1), // min
         MutableRecord(42, 1), // current
         MutableRecord(43, 1)) // max
 
-      firstSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-default-configuration")).min shouldBe (0)
-      firstSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-default-configuration")).max shouldBe (0)
-      firstSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-default-configuration")).numberOfMeasurements should be(3)
-      firstSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-default-configuration")).recordsIterator.toStream should contain only (
+      firstSnapshot(UserMinMaxCounter("min-max-counter-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (0)
+      firstSnapshot(UserMinMaxCounter("min-max-counter-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (0)
+      firstSnapshot(UserMinMaxCounter("min-max-counter-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(3)
+      firstSnapshot(UserMinMaxCounter("min-max-counter-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream should contain only (
         MutableRecord(0, 3)) // min, max and current
 
-      firstSnapshot.gauges(UserGauge("gauge-with-default-configuration")).min shouldBe (15)
-      firstSnapshot.gauges(UserGauge("gauge-with-default-configuration")).max shouldBe (15)
-      firstSnapshot.gauges(UserGauge("gauge-with-default-configuration")).numberOfMeasurements should be(1)
-      firstSnapshot.gauges(UserGauge("gauge-with-default-configuration")).recordsIterator.toStream should contain only (
+      firstSnapshot(UserGauge("gauge-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (15)
+      firstSnapshot(UserGauge("gauge-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (15)
+      firstSnapshot(UserGauge("gauge-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(1)
+      firstSnapshot(UserGauge("gauge-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream should contain only (
         MutableRecord(15, 1)) // only the manually recorded value
 
-      val secondSnapshot = userMetricsRecorder.collect(context)
+      Kamon(Metrics).subscriptions ! Subscriptions.FlushMetrics
+      val secondSnapshot = expectMsgType[TickMetricSnapshot].metrics
 
-      secondSnapshot.histograms.size should be(2)
-      secondSnapshot.histograms.keys should contain allOf (
+      secondSnapshot.keys should contain allOf (
         UserHistogram("histogram-with-settings"),
         UserHistogram("histogram-with-default-configuration"))
 
-      secondSnapshot.histograms(UserHistogram("histogram-with-settings")).min shouldBe (0)
-      secondSnapshot.histograms(UserHistogram("histogram-with-settings")).max shouldBe (0)
-      secondSnapshot.histograms(UserHistogram("histogram-with-settings")).numberOfMeasurements should be(0)
-      secondSnapshot.histograms(UserHistogram("histogram-with-settings")).recordsIterator.toStream shouldBe empty
-
-      secondSnapshot.histograms(UserHistogram("histogram-with-default-configuration")).min shouldBe (0)
-      secondSnapshot.histograms(UserHistogram("histogram-with-default-configuration")).max shouldBe (0)
-      secondSnapshot.histograms(UserHistogram("histogram-with-default-configuration")).numberOfMeasurements should be(0)
-      secondSnapshot.histograms(UserHistogram("histogram-with-default-configuration")).recordsIterator.toStream shouldBe empty
-
-      secondSnapshot.counters(UserCounter("counter")).count should be(0)
-
-      secondSnapshot.minMaxCounters.size should be(2)
-      secondSnapshot.minMaxCounters.keys should contain allOf (
-        UserMinMaxCounter("min-max-counter-with-settings"),
-        UserMinMaxCounter("min-max-counter-with-default-configuration"))
-
-      secondSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-settings")).min shouldBe (42)
-      secondSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-settings")).max shouldBe (42)
-      secondSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-settings")).numberOfMeasurements should be(3)
-      secondSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-settings")).recordsIterator.toStream should contain only (
-        MutableRecord(42, 3)) // min, max and current
-
-      secondSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-default-configuration")).min shouldBe (0)
-      secondSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-default-configuration")).max shouldBe (0)
-      secondSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-default-configuration")).numberOfMeasurements should be(3)
-      secondSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-with-default-configuration")).recordsIterator.toStream should contain only (
+      secondSnapshot(UserHistogram("histogram-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (0)
+      secondSnapshot(UserHistogram("histogram-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (0)
+      secondSnapshot(UserHistogram("histogram-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(0)
+      secondSnapshot(UserHistogram("histogram-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream shouldBe empty
+
+      secondSnapshot(UserHistogram("histogram-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (0)
+      secondSnapshot(UserHistogram("histogram-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (0)
+      secondSnapshot(UserHistogram("histogram-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(0)
+      secondSnapshot(UserHistogram("histogram-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream shouldBe empty
+
+      secondSnapshot(UserCounter("counter")).metrics(Count).asInstanceOf[Counter.Snapshot].count should be(0)
+
+      secondSnapshot(UserMinMaxCounter("min-max-counter-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (42)
+      secondSnapshot(UserMinMaxCounter("min-max-counter-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (42)
+      secondSnapshot(UserMinMaxCounter("min-max-counter-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(3)
+      secondSnapshot(UserMinMaxCounter("min-max-counter-with-settings")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream should contain only (
+        MutableRecord(42, 3)) // max
+
+      secondSnapshot(UserMinMaxCounter("min-max-counter-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (0)
+      secondSnapshot(UserMinMaxCounter("min-max-counter-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (0)
+      secondSnapshot(UserMinMaxCounter("min-max-counter-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(3)
+      secondSnapshot(UserMinMaxCounter("min-max-counter-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream should contain only (
         MutableRecord(0, 3)) // min, max and current
 
-      secondSnapshot.gauges(UserGauge("gauge-with-default-configuration")).min shouldBe (0)
-      secondSnapshot.gauges(UserGauge("gauge-with-default-configuration")).max shouldBe (0)
-      secondSnapshot.gauges(UserGauge("gauge-with-default-configuration")).numberOfMeasurements should be(0)
-      secondSnapshot.gauges(UserGauge("gauge-with-default-configuration")).recordsIterator shouldBe empty
+      secondSnapshot(UserGauge("gauge-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (0)
+      secondSnapshot(UserGauge("gauge-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (0)
+      secondSnapshot(UserGauge("gauge-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(0)
+      secondSnapshot(UserGauge("gauge-with-default-configuration")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream shouldBe empty
 
+      Kamon(Metrics).unsubscribe(testActor)
     }
 
     "generate a snapshot that can be merged with another" in {
-      val context = Kamon(Metrics).buildDefaultCollectionContext
-      val userMetricsRecorder = Kamon(Metrics).register(UserMetrics, UserMetrics.Factory).get
+      val buffer = system.actorOf(TickMetricSnapshotBuffer.props(1 hours, testActor))
+      Kamon(Metrics).subscribe(UserHistograms, "*", buffer, permanently = true)
+      Kamon(Metrics).subscribe(UserCounters, "*", buffer, permanently = true)
+      Kamon(Metrics).subscribe(UserMinMaxCounters, "*", buffer, permanently = true)
+      Kamon(Metrics).subscribe(UserGauges, "*", buffer, permanently = true)
 
       val histogram = Kamon(UserMetrics).registerHistogram("histogram-for-merge")
       val counter = Kamon(UserMetrics).registerCounter("counter-for-merge")
@@ -259,7 +261,8 @@ class UserMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
       minMaxCounter.decrement(10)
       gauge.record(50)
 
-      val firstSnapshot = userMetricsRecorder.collect(context)
+      Kamon(Metrics).subscriptions ! Subscriptions.FlushMetrics
+      Thread.sleep(2000) // Make sure that the snapshots are taken before proceeding
 
       val extraCounter = Kamon(UserMetrics).registerCounter("extra-counter")
       histogram.record(200)
@@ -268,37 +271,41 @@ class UserMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
       minMaxCounter.decrement(50)
       gauge.record(70)
 
-      val secondSnapshot = userMetricsRecorder.collect(context)
-      val mergedSnapshot = firstSnapshot.merge(secondSnapshot, context)
+      Kamon(Metrics).subscriptions ! Subscriptions.FlushMetrics
+      Thread.sleep(2000) // Make sure that the metrics are buffered.
+      buffer ! TickMetricSnapshotBuffer.FlushBuffer
+      val snapshot = expectMsgType[TickMetricSnapshot].metrics
 
-      mergedSnapshot.histograms.keys should contain(UserHistogram("histogram-for-merge"))
+      snapshot.keys should contain(UserHistogram("histogram-for-merge"))
 
-      mergedSnapshot.histograms(UserHistogram("histogram-for-merge")).min shouldBe (100)
-      mergedSnapshot.histograms(UserHistogram("histogram-for-merge")).max shouldBe (200)
-      mergedSnapshot.histograms(UserHistogram("histogram-for-merge")).numberOfMeasurements should be(2)
-      mergedSnapshot.histograms(UserHistogram("histogram-for-merge")).recordsIterator.toStream should contain allOf (
+      snapshot(UserHistogram("histogram-for-merge")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (100)
+      snapshot(UserHistogram("histogram-for-merge")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (200)
+      snapshot(UserHistogram("histogram-for-merge")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(2)
+      snapshot(UserHistogram("histogram-for-merge")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream should contain allOf (
         MutableRecord(100, 1),
         MutableRecord(200, 1))
 
-      mergedSnapshot.counters(UserCounter("counter-for-merge")).count should be(10)
-      mergedSnapshot.counters(UserCounter("extra-counter")).count should be(20)
+      snapshot(UserCounter("counter-for-merge")).metrics(Count).asInstanceOf[Counter.Snapshot].count should be(10)
+      snapshot(UserCounter("extra-counter")).metrics(Count).asInstanceOf[Counter.Snapshot].count should be(20)
 
-      mergedSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-for-merge")).min shouldBe (0)
-      mergedSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-for-merge")).max shouldBe (80)
-      mergedSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-for-merge")).numberOfMeasurements should be(6)
-      mergedSnapshot.minMaxCounters(UserMinMaxCounter("min-max-counter-for-merge")).recordsIterator.toStream should contain allOf (
+      snapshot(UserMinMaxCounter("min-max-counter-for-merge")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (0)
+      snapshot(UserMinMaxCounter("min-max-counter-for-merge")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (80)
+      snapshot(UserMinMaxCounter("min-max-counter-for-merge")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(6)
+      snapshot(UserMinMaxCounter("min-max-counter-for-merge")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream should contain allOf (
         MutableRecord(0, 1), // min in first snapshot
         MutableRecord(30, 2), // min and current in second snapshot
         MutableRecord(40, 1), // current in first snapshot
         MutableRecord(50, 1), // max in first snapshot
         MutableRecord(80, 1)) // max in second snapshot
 
-      mergedSnapshot.gauges(UserGauge("gauge-for-merge")).min shouldBe (50)
-      mergedSnapshot.gauges(UserGauge("gauge-for-merge")).max shouldBe (70)
-      mergedSnapshot.gauges(UserGauge("gauge-for-merge")).numberOfMeasurements should be(2)
-      mergedSnapshot.gauges(UserGauge("gauge-for-merge")).recordsIterator.toStream should contain allOf (
+      snapshot(UserGauge("gauge-for-merge")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].min shouldBe (50)
+      snapshot(UserGauge("gauge-for-merge")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].max shouldBe (70)
+      snapshot(UserGauge("gauge-for-merge")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].numberOfMeasurements should be(2)
+      snapshot(UserGauge("gauge-for-merge")).metrics(RecordedValues).asInstanceOf[Histogram.Snapshot].recordsIterator.toStream should contain allOf (
         MutableRecord(50, 1),
         MutableRecord(70, 1))
+
+      Kamon(Metrics).unsubscribe(testActor)
     }
   }
 }