Merge pull request #1017 from SimunKaracic/add-reset-to-range-sampler

Add resetDistribution method to RangeSampler

Author: SimunKaracic

build.sbt

@@ -426,7 +426,6 @@ def akkaHttpVersion(scalaVersion: String) = scalaVersion match {
   case _      => "10.2.3"
 }
 
-
 lazy val `kamon-akka-http` = (project in file("instrumentation/kamon-akka-http"))
   .enablePlugins(JavaAgent)
   .disablePlugins(AssemblyPlugin)

core/kamon-core-tests/src/test/scala/kamon/metric/RangeSamplerSpec.scala

@@ -17,11 +17,12 @@
 package kamon.metric
 
 import java.time.Duration
-
 import kamon.Kamon
 import kamon.testkit.InstrumentInspection
 import org.scalatest.{Matchers, WordSpec}
 
+import scala.concurrent.duration.DurationInt
+
 class RangeSamplerSpec extends WordSpec with Matchers with InstrumentInspection.Syntax {
 
   "a RangeSampler" should {
@@ -84,6 +85,7 @@ class RangeSamplerSpec extends WordSpec with Matchers with InstrumentInspection.
         MeasurementUnit.none,
         Duration.ofMillis(1)
       ).withoutTags()
+
       rangeSampler.increment()
       rangeSampler.increment(3)
       rangeSampler.increment()
@@ -94,5 +96,25 @@ class RangeSamplerSpec extends WordSpec with Matchers with InstrumentInspection.
       snapshot.min should be(0)
       snapshot.max should be(5)
     }
+
+    "reset values to 0 after calling resetDistribution" in {
+      val rangeSampler = Kamon.rangeSampler(
+        "auto-update2",
+        MeasurementUnit.none,
+        Duration.ofMillis(1)
+      ).withoutTags()
+
+      rangeSampler.increment(5)
+      rangeSampler.resetDistribution()
+
+      Thread.sleep(50)
+      rangeSampler.resetDistribution()
+
+      val snapshot = rangeSampler.distribution()
+
+      snapshot.min should be(0)
+      snapshot.max should be(0)
+      snapshot.sum should be(0)
+    }
   }
-}
+}

core/kamon-core/src/main/scala/kamon/metric/RangeSampler.scala

@@ -65,6 +65,11 @@ trait RangeSampler extends Instrument[RangeSampler, Metric.Settings.ForDistribut
     */
   def sample(): RangeSampler
 
+  /**
+    * Resets the minimum, maximum and current value indicators.
+    */
+  def resetDistribution(): RangeSampler
+
 }
 
 
@@ -84,7 +89,7 @@ object RangeSampler {
 
     private val _min = new AtomicLongMaxUpdater(new AtomicLong(0L))
     private val _max = new AtomicLongMaxUpdater(new AtomicLong(0L))
-    private val _sum = new AtomicLong()
+    private val _current = new AtomicLong()
 
     override def increment(): RangeSampler =
       increment(1)
@@ -93,13 +98,20 @@ object RangeSampler {
       decrement(1)
 
     override def increment(times: Long): RangeSampler = {
-      val currentValue = _sum.addAndGet(times)
+      val currentValue = _current.addAndGet(times)
       _max.update(currentValue)
       this
     }
 
+    override def resetDistribution(): RangeSampler = {
+      _min.maxThenReset(0)
+      _max.maxThenReset(0)
+      _current.set(0)
+      this
+    }
+
     override def decrement(times: Long): RangeSampler = {
-      val currentValue = _sum.addAndGet(-times)
+      val currentValue = _current.addAndGet(-times)
       _min.update(-currentValue)
       this
     }
@@ -112,7 +124,7 @@ object RangeSampler {
     override def sample(): RangeSampler = {
       try {
         val currentValue = {
-          val value = _sum.get()
+          val value = _current.get()
           if (value <= 0) 0 else value
         }
 
@@ -153,7 +165,11 @@ object RangeSampler {
       def update(newMax: Long):Unit = {
         @tailrec def compare(): Long = {
           val currentMax = value.get()
-          if(newMax > currentMax) if (!value.compareAndSet(currentMax, newMax)) compare() else newMax
+          if(newMax > currentMax) {
+            if (!value.compareAndSet(currentMax, newMax)) {
+              compare()
+            } else newMax
+          }
           else currentMax
         }
         compare()