! core: first implementetion of kamon counter intrument and actor errors metrics

5 files changed, 48 insertions, 9 deletions

diff --git a/kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala b/kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala
index ae9d20f6..78c170de 100644
--- a/kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala
+++ b/kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala
@@ -24,7 +24,6 @@ import kamon.metrics.{ ActorMetrics, Metrics }
 import kamon.Kamon
 import kamon.metrics.ActorMetrics.ActorMetricRecorder
 import java.util.concurrent.atomic.AtomicInteger
-import kamon.metrics.instruments.Counter
 
 @Aspect
 class BehaviourInvokeTracing {
@@ -33,7 +32,7 @@ class BehaviourInvokeTracing {
   def actorCellCreation(cell: ActorCell, system: ActorSystem, ref: ActorRef, props: Props, dispatcher: MessageDispatcher, parent: ActorRef): Unit = {}
 
   @After("actorCellCreation(cell, system, ref, props, dispatcher, parent)")
-  def afterCreation(cell: ActorCellMetrics, system: ActorSystem, ref: ActorRef, props: Props, dispatcher: MessageDispatcher, parent: ActorRef): Unit = {
+  def afterCreation(cell: ActorCell, system: ActorSystem, ref: ActorRef, props: Props, dispatcher: MessageDispatcher, parent: ActorRef): Unit = {
     val metricsExtension = Kamon(Metrics)(system)
     val metricIdentity = ActorMetrics(ref.path.elements.mkString("/"))
     val cellWithMetrics = cell.asInstanceOf[ActorCellMetrics]
@@ -92,11 +91,13 @@ class BehaviourInvokeTracing {
   }
 
   @Pointcut("execution(* akka.actor.ActorCell.handleInvokeFailure(..)) && this(cell)")
-  def actorInvokeFailure(cell: ActorCellMetrics): Unit = {}
+  def actorInvokeFailure(cell: ActorCell): Unit = {}
 
   @Before("actorInvokeFailure(cell)")
-  def beforeInvokeFailure(cell: ActorCellMetrics): Unit = {
-    cell.actorMetricsRecorder.map {
+  def beforeInvokeFailure(cell: ActorCell): Unit = {
+    val cellWithMetrics = cell.asInstanceOf[ActorCellMetrics]
+
+    cellWithMetrics.actorMetricsRecorder.map {
       am ⇒ am.errorCounter.record(1L)
     }
   }
diff --git a/kamon-core/src/main/scala/kamon/metrics/instruments/CounterRecorder.scala b/kamon-core/src/main/scala/kamon/metrics/instruments/CounterRecorder.scala
index 1ab743d2..0fd56105 100644
--- a/kamon-core/src/main/scala/kamon/metrics/instruments/CounterRecorder.scala
+++ b/kamon-core/src/main/scala/kamon/metrics/instruments/CounterRecorder.scala
@@ -23,7 +23,9 @@ import jsr166e.LongAdder
 class CounterRecorder extends MetricRecorder {
   private val counter = new LongAdder
 
-  def record(value: Long): Unit = counter.add(value)
+  def record(value: Long): Unit = {
+    counter.add(value)
+  }
 
   def collect(): MetricSnapshotLike = {
     val sum = counter.sumThenReset()
diff --git a/kamon-core/src/test/scala/kamon/metrics/ActorMetricsSpec.scala b/kamon-core/src/test/scala/kamon/metrics/ActorMetricsSpec.scala
index 61d15bc9..2144bd55 100644
--- a/kamon-core/src/test/scala/kamon/metrics/ActorMetricsSpec.scala
+++ b/kamon-core/src/test/scala/kamon/metrics/ActorMetricsSpec.scala
@@ -103,6 +103,17 @@ class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
     }
   }
 
+  "track the number of errors" in new ErrorActorFixture {
+    val (error, metricsListener) = failedActor("tracked-errors")
+
+    for (_ ← 1 to 5) {
+      error ! Error
+    }
+
+    val actorMetrics = expectActorMetrics("user/tracked-errors", metricsListener, 3 seconds)
+    actorMetrics.errorCounter.numberOfMeasurements should be(5L)
+  }
+
   def expectActorMetrics(actorPath: String, listener: TestProbe, waitTime: FiniteDuration): ActorMetricSnapshot = {
     val tickSnapshot = within(waitTime) {
       listener.expectMsgType[TickMetricSnapshot]
@@ -124,6 +135,19 @@ class ActorMetricsSpec extends TestKitBase with WordSpecLike with Matchers {
       (actor, metricsListener)
     }
   }
+
+  trait ErrorActorFixture {
+    def failedActor(name: String): (ActorRef, TestProbe) = {
+      val actor = system.actorOf(Props[FailedActor], name)
+      val metricsListener = TestProbe()
+
+      Kamon(Metrics).subscribe(ActorMetrics, "user/" + name, metricsListener.ref, permanently = true)
+      // Wait for one empty snapshot before proceeding to the test.
+      metricsListener.expectMsgType[TickMetricSnapshot]
+
+      (actor, metricsListener)
+    }
+  }
 }
 
 class DelayableActor extends Actor {
@@ -133,5 +157,12 @@ class DelayableActor extends Actor {
   }
 }
 
+class FailedActor extends Actor {
+  def receive = {
+    case Error   ⇒ 1 / 0
+    case Discard ⇒
+  }
+}
 case object Discard
 case class Delay(time: FiniteDuration)
+case class Error()
diff --git a/kamon-core/src/test/scala/kamon/metrics/MetricSnapshotSpec.scala b/kamon-core/src/test/scala/kamon/metrics/MetricSnapshotSpec.scala
index b4f33ec3..4d6ebc49 100644
--- a/kamon-core/src/test/scala/kamon/metrics/MetricSnapshotSpec.scala
+++ b/kamon-core/src/test/scala/kamon/metrics/MetricSnapshotSpec.scala
@@ -25,25 +25,28 @@ class MetricSnapshotSpec extends WordSpec with Matchers {
     "support a max operation" in new SnapshotFixtures {
       snapshotA.max should be(17)
       snapshotB.max should be(10)
+      snapshotC.max should be(1)
     }
 
     "support a min operation" in new SnapshotFixtures {
       snapshotA.min should be(1)
       snapshotB.min should be(2)
+      snapshotC.min should be(1)
     }
 
     "be able to merge with other snapshot" in new SnapshotFixtures {
-      val merged = snapshotA.merge(snapshotB)
+      val merged = snapshotA.merge(snapshotB).merge(snapshotC)
 
       merged.min should be(1)
       merged.max should be(17)
-      merged.numberOfMeasurements should be(200)
+      merged.numberOfMeasurements should be(300)
       merged.measurements.map(_.value) should contain inOrderOnly (1, 2, 4, 5, 7, 10, 17)
     }
 
     "be able to merge with empty snapshots" in new SnapshotFixtures {
       snapshotA.merge(emptySnapshot) should be(snapshotA)
       emptySnapshot.merge(snapshotA).merge(emptySnapshot) should be(snapshotA)
+      snapshotC.merge(emptySnapshot) should be(snapshotC)
     }
 
   }
@@ -63,5 +66,7 @@ class MetricSnapshotSpec extends WordSpec with Matchers {
       Measurement(4, 48),
       Measurement(5, 39),
       Measurement(10, 7)))
+
+    val snapshotC = MetricSnapshot(InstrumentTypes.Counter, 100, Scale.Unit, Vector(Measurement(1, 100)))
   }
 }
diff --git a/kamon-play/src/test/scala/kamon/play/WSInstrumentationSpec.scala b/kamon-play/src/test/scala/kamon/play/WSInstrumentationSpec.scala
index f76b20b8..f8801d8e 100644
--- a/kamon-play/src/test/scala/kamon/play/WSInstrumentationSpec.scala
+++ b/kamon-play/src/test/scala/kamon/play/WSInstrumentationSpec.scala
@@ -63,4 +63,4 @@ class WSInstrumentationSpec extends PlaySpecification {
       Thread.sleep(2000) //wait to complete the future
     }
   }
-} 
\ No newline at end of file
+}
\ No newline at end of file