+ Kamon-core: introduce finishWithError(Throwable) for Traces and Segments

* + kamon-core: introduce finishWithError(Throwable) for Traces and Segments

5 files changed, 79 insertions, 51 deletions

diff --git a/kamon-core/src/main/scala/kamon/metric/TraceMetrics.scala b/kamon-core/src/main/scala/kamon/metric/TraceMetrics.scala
index eb4f327a..014825cd 100644
--- a/kamon-core/src/main/scala/kamon/metric/TraceMetrics.scala
+++ b/kamon-core/src/main/scala/kamon/metric/TraceMetrics.scala
@@ -1,6 +1,6 @@
 /*
  * =========================================================================================
- * Copyright © 2013 the kamon project <http://kamon.io/>
+ * Copyright © 2013-2016 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
@@ -24,6 +24,7 @@ class TraceMetrics(instrumentFactory: InstrumentFactory) extends GenericEntityRe
    *  Records blah blah
    */
   val elapsedTime = histogram("elapsed-time", unitOfMeasurement = Time.Nanoseconds)
+  val errors = counter("errors")
 }
 
 object TraceMetrics extends EntityRecorderFactory[TraceMetrics] {
@@ -40,6 +41,7 @@ class SegmentMetrics(instrumentFactory: InstrumentFactory) extends GenericEntity
    *  Records blah blah
    */
   val elapsedTime = histogram("elapsed-time", unitOfMeasurement = Time.Nanoseconds)
+  val errors = counter("errors")
 }
 
 object SegmentMetrics extends EntityRecorderFactory[SegmentMetrics] {
diff --git a/kamon-core/src/main/scala/kamon/trace/MetricsOnlyContext.scala b/kamon-core/src/main/scala/kamon/trace/MetricsOnlyContext.scala
index 34b68f38..973eab5a 100644
--- a/kamon-core/src/main/scala/kamon/trace/MetricsOnlyContext.scala
+++ b/kamon-core/src/main/scala/kamon/trace/MetricsOnlyContext.scala
@@ -1,6 +1,6 @@
 /*
  * =========================================================================================
- * Copyright © 2013-2014 the kamon project <http://kamon.io/>
+ * Copyright © 2013-2016 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
@@ -21,17 +21,18 @@ import java.util.concurrent.ConcurrentLinkedQueue
 import akka.event.LoggingAdapter
 import kamon.Kamon
 import kamon.metric.{ SegmentMetrics, TraceMetrics }
+import kamon.trace.States.Status
 import kamon.util.{ NanoInterval, RelativeNanoTimestamp }
 
 import scala.annotation.tailrec
 import scala.collection.concurrent.TrieMap
 
-private[kamon] class MetricsOnlyContext(traceName: String, val token: String, traceTags: Map[String, String], izOpen: Boolean, val levelOfDetail: LevelOfDetail,
+private[kamon] class MetricsOnlyContext(traceName: String, val token: String, traceTags: Map[String, String], currentStatus: Status, val levelOfDetail: LevelOfDetail,
   val startTimestamp: RelativeNanoTimestamp, log: LoggingAdapter)
     extends TraceContext {
 
   @volatile private var _name = traceName
-  @volatile private var _isOpen = izOpen
+  @volatile private var _status = currentStatus
   @volatile protected var _elapsedTime = NanoInterval.default
 
   private val _finishedSegments = new ConcurrentLinkedQueue[SegmentLatencyData]()
@@ -39,30 +40,38 @@ private[kamon] class MetricsOnlyContext(traceName: String, val token: String, tr
   private val _tags = TrieMap.empty[String, String] ++= traceTags
 
   def rename(newName: String): Unit =
-    if (isOpen)
+    if (States.Open == status)
       _name = newName
     else
       log.warning("Can't rename trace from [{}] to [{}] because the trace is already closed.", name, newName)
 
   def name: String = _name
   def isEmpty: Boolean = false
-  def isOpen: Boolean = _isOpen
-
+  def status: Status = _status
   def addMetadata(key: String, value: String): Unit = {}
-
   def addTag(key: String, value: String): Unit = _tags.put(key, value)
   def removeTag(key: String, value: String): Boolean = _tags.remove(key, value)
 
-  def finish(): Unit = {
-    _isOpen = false
+  private def finish(withError: Boolean): Unit = {
+    _status = if (withError) States.FinishedWithError else States.FinishedSuccessfully
     val traceElapsedTime = NanoInterval.since(startTimestamp)
     _elapsedTime = traceElapsedTime
 
-    if (Kamon.metrics.shouldTrack(name, TraceMetrics.category))
-      Kamon.metrics.entity(TraceMetrics, name, _tags.toMap).elapsedTime.record(traceElapsedTime.nanos)
+    if (Kamon.metrics.shouldTrack(name, TraceMetrics.category)) {
+      val traceEntity = Kamon.metrics.entity(TraceMetrics, name)
+      traceEntity.elapsedTime.record(traceElapsedTime.nanos)
+      if (withError) traceEntity.errors.increment()
+    }
     drainFinishedSegments()
   }
 
+  def finish(): Unit = finish(withError = false)
+
+  def finishWithError(cause: Throwable): Unit = {
+    //we should do something with the Throwable in a near future
+    finish(withError = true)
+  }
+
   def startSegment(segmentName: String, category: String, library: String): Segment =
     startSegment(segmentName, category, library, Map.empty[String, String])
 
@@ -77,14 +86,17 @@ private[kamon] class MetricsOnlyContext(traceName: String, val token: String, tr
         "category" -> segment.category,
         "library" -> segment.library)
 
-      if (Kamon.metrics.shouldTrack(segment.name, SegmentMetrics.category))
-        Kamon.metrics.entity(SegmentMetrics, segment.name, defaultTags ++ segment.tags).elapsedTime.record(segment.duration.nanos)
+      if (Kamon.metrics.shouldTrack(segment.name, SegmentMetrics.category)) {
+        val segmentEntity = Kamon.metrics.entity(SegmentMetrics, segment.name, defaultTags ++ segment.tags)
+        segmentEntity.elapsedTime.record(segment.duration.nanos)
+        if (segment.isFinishedWithError) segmentEntity.errors.increment()
+      }
       drainFinishedSegments()
     }
   }
 
-  protected def finishSegment(segmentName: String, category: String, library: String, duration: NanoInterval, segmentTags: Map[String, String]): Unit = {
-    _finishedSegments.add(SegmentLatencyData(segmentName, category, library, duration, segmentTags))
+  protected def finishSegment(segmentName: String, category: String, library: String, duration: NanoInterval, segmentTags: Map[String, String], isFinishedWithError: Boolean): Unit = {
+    _finishedSegments.add(SegmentLatencyData(segmentName, category, library, duration, segmentTags, isFinishedWithError))
 
     if (isClosed) {
       drainFinishedSegments()
@@ -104,36 +116,42 @@ private[kamon] class MetricsOnlyContext(traceName: String, val token: String, tr
 
     @volatile private var _segmentName = segmentName
     @volatile private var _elapsedTime = NanoInterval.default
-    @volatile private var _isOpen = true
+    @volatile private var _status: Status = States.Open
 
     def name: String = _segmentName
     def isEmpty: Boolean = false
-    def isOpen: Boolean = _isOpen
-
+    def status: Status = _status
     def addMetadata(key: String, value: String): Unit = {}
-
     def addTag(key: String, value: String): Unit = _tags.put(key, value)
     def removeTag(key: String, value: String): Boolean = _tags.remove(key, value)
 
     def rename(newName: String): Unit =
-      if (isOpen)
+      if (States.Open == status)
         _segmentName = newName
       else
         log.warning("Can't rename segment from [{}] to [{}] because the segment is already closed.", name, newName)
 
-    def finish: Unit = {
-      _isOpen = false
+    private def finish(withError: Boolean): Unit = {
+      _status = if (withError) States.FinishedWithError else States.FinishedSuccessfully
       val segmentElapsedTime = NanoInterval.since(_startTimestamp)
       _elapsedTime = segmentElapsedTime
 
-      finishSegment(name, category, library, segmentElapsedTime, _tags.toMap)
+      finishSegment(name, category, library, segmentElapsedTime, _tags.toMap, withError)
+    }
+
+    def finishWithError(cause: Throwable): Unit = {
+      //we should do something with the Throwable in a near future
+      finish(withError = true)
     }
 
+    def finish(): Unit = finish(withError = false)
+
     // Handle with care and make sure that the segment is closed before calling this method, otherwise
     // NanoInterval.default will be returned.
     def elapsedTime: NanoInterval = _elapsedTime
     def startTimestamp: RelativeNanoTimestamp = _startTimestamp
+
   }
 }
 
-case class SegmentLatencyData(name: String, category: String, library: String, duration: NanoInterval, tags: Map[String, String])
+case class SegmentLatencyData(name: String, category: String, library: String, duration: NanoInterval, tags: Map[String, String], isFinishedWithError: Boolean)
\ No newline at end of file
diff --git a/kamon-core/src/main/scala/kamon/trace/TraceContext.scala b/kamon-core/src/main/scala/kamon/trace/TraceContext.scala
index eee2ec14..6bf82ac1 100644
--- a/kamon-core/src/main/scala/kamon/trace/TraceContext.scala
+++ b/kamon-core/src/main/scala/kamon/trace/TraceContext.scala
@@ -19,6 +19,7 @@ package kamon.trace
 import java.io.ObjectStreamException
 import java.util
 
+import kamon.trace.States.{ Closed, Status }
 import kamon.trace.TraceContextAware.DefaultTraceContextAware
 import kamon.util.{ Function, RelativeNanoTimestamp, SameThreadExecutionContext, Supplier }
 
@@ -29,19 +30,16 @@ trait TraceContext {
   def token: String
   def isEmpty: Boolean
   def nonEmpty: Boolean = !isEmpty
-  def isOpen: Boolean
-  def isClosed: Boolean = !isOpen
-
+  def isClosed: Boolean = !(States.Open == status)
+  def status: Status
   def finish(): Unit
+  def finishWithError(cause: Throwable): Unit
   def rename(newName: String): Unit
-
   def startSegment(segmentName: String, category: String, library: String): Segment
   def startSegment(segmentName: String, category: String, library: String, tags: Map[String, String]): Segment
   def addMetadata(key: String, value: String)
-
   def addTag(key: String, value: String): Unit
   def removeTag(key: String, value: String): Boolean
-
   def startTimestamp: RelativeNanoTimestamp
 
   def collect[T](f: TraceContext ⇒ T): Option[T] =
@@ -90,13 +88,12 @@ trait Segment {
   def library: String
   def isEmpty: Boolean
   def nonEmpty: Boolean = !isEmpty
-  def isOpen: Boolean
-  def isClosed: Boolean = !isOpen
-
+  def isClosed: Boolean = !(States.Open == status)
+  def status: Status
   def finish(): Unit
+  def finishWithError(cause: Throwable): Unit
   def rename(newName: String): Unit
   def addMetadata(key: String, value: String): Unit
-
   def addTag(key: String, value: String): Unit
   def removeTag(key: String, value: String): Boolean
 }
@@ -105,9 +102,9 @@ case object EmptyTraceContext extends TraceContext {
   def name: String = "empty-trace"
   def token: String = ""
   def isEmpty: Boolean = true
-  def isOpen: Boolean = false
-
+  def status: Status = Closed
   def finish(): Unit = {}
+  def finishWithError(cause: Throwable): Unit = {}
   def rename(name: String): Unit = {}
   def startSegment(segmentName: String, category: String, library: String): Segment = EmptySegment
   def startSegment(segmentName: String, category: String, library: String, tags: Map[String, String]): Segment = EmptySegment
@@ -123,9 +120,9 @@ case object EmptyTraceContext extends TraceContext {
     val category: String = "empty-category"
     val library: String = "empty-library"
     def isEmpty: Boolean = true
-    def isOpen: Boolean = false
-
-    def finish: Unit = {}
+    def status: Status = Closed
+    def finish(): Unit = {}
+    def finishWithError(cause: Throwable): Unit = {}
     def rename(newName: String): Unit = {}
     def addMetadata(key: String, value: String): Unit = {}
     def addTag(key: String, value: String): Unit = {}
@@ -151,6 +148,14 @@ object LevelOfDetail {
   case object FullTrace extends LevelOfDetail
 }
 
+object States {
+  sealed trait Status
+  case object Open extends Status
+  case object Closed extends Status
+  case object FinishedWithError extends Status
+  case object FinishedSuccessfully extends Status
+}
+
 trait TraceContextAware extends Serializable {
   def traceContext: TraceContext
 }
diff --git a/kamon-core/src/main/scala/kamon/trace/TracerModule.scala b/kamon-core/src/main/scala/kamon/trace/TracerModule.scala
index 60187729..2fc166b6 100644
--- a/kamon-core/src/main/scala/kamon/trace/TracerModule.scala
+++ b/kamon-core/src/main/scala/kamon/trace/TracerModule.scala
@@ -23,14 +23,16 @@ import akka.event.{ Logging, LoggingAdapter }
 import com.typesafe.config.Config
 import kamon.Kamon
 import kamon.metric.MetricsModule
+import kamon.trace.States.Status
 import kamon.util._
+
 import scala.collection.JavaConverters._
 
 trait TracerModule {
   def newContext(name: String): TraceContext
   def newContext(name: String, token: Option[String]): TraceContext
   def newContext(name: String, token: Option[String], tags: Map[String, String]): TraceContext
-  def newContext(name: String, token: Option[String], tags: Map[String, String], timestamp: RelativeNanoTimestamp, isOpen: Boolean, isLocal: Boolean): TraceContext
+  def newContext(name: String, token: Option[String], tags: Map[String, String], timestamp: RelativeNanoTimestamp, state: Status, isLocal: Boolean): TraceContext
 
   def subscribe(subscriber: ActorRef): Unit
   def unsubscribe(subscriber: ActorRef): Unit
@@ -134,14 +136,14 @@ private[kamon] class TracerModuleImpl(metricsExtension: MetricsModule, config: C
   def newContext(name: String, token: Option[String], tags: Map[String, String]): TraceContext =
     createTraceContext(name, token, tags)
 
-  def newContext(name: String, token: Option[String], tags: Map[String, String], timestamp: RelativeNanoTimestamp, isOpen: Boolean, isLocal: Boolean): TraceContext =
-    createTraceContext(name, token, tags, timestamp, isOpen, isLocal)
+  def newContext(name: String, token: Option[String], tags: Map[String, String], timestamp: RelativeNanoTimestamp, status: Status, isLocal: Boolean): TraceContext =
+    createTraceContext(name, token, tags, timestamp, status, isLocal)
 
   private def createTraceContext(traceName: String, token: Option[String], tags: Map[String, String] = Map.empty, startTimestamp: RelativeNanoTimestamp = RelativeNanoTimestamp.now,
-    isOpen: Boolean = true, isLocal: Boolean = true): TraceContext = {
+    status: Status = States.Open, isLocal: Boolean = true): TraceContext = {
 
     def newMetricsOnlyContext(token: String): TraceContext =
-      new MetricsOnlyContext(traceName, token, tags, isOpen, _settings.levelOfDetail, startTimestamp, _logger)
+      new MetricsOnlyContext(traceName, token, tags, status, _settings.levelOfDetail, startTimestamp, _logger)
 
     val traceToken = token.getOrElse(newToken)
 
@@ -151,7 +153,7 @@ private[kamon] class TracerModuleImpl(metricsExtension: MetricsModule, config: C
       case _ if !isLocal || !_settings.sampler.shouldTrace ⇒
         newMetricsOnlyContext(traceToken)
       case _ ⇒
-        new TracingContext(traceName, traceToken, tags, izOpen = true, _settings.levelOfDetail, isLocal, startTimestamp, _logger, dispatchTracingContext)
+        new TracingContext(traceName, traceToken, tags, currentStatus = States.Open, _settings.levelOfDetail, isLocal, startTimestamp, _logger, dispatchTracingContext)
     }
   }
 
diff --git a/kamon-core/src/main/scala/kamon/trace/TracingContext.scala b/kamon-core/src/main/scala/kamon/trace/TracingContext.scala
index 085b4b09..496d7317 100644
--- a/kamon-core/src/main/scala/kamon/trace/TracingContext.scala
+++ b/kamon-core/src/main/scala/kamon/trace/TracingContext.scala
@@ -20,13 +20,14 @@ import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.atomic.AtomicInteger
 
 import akka.event.LoggingAdapter
+import kamon.trace.States.Status
 import kamon.util.{ NanoInterval, NanoTimestamp, RelativeNanoTimestamp }
 
 import scala.collection.concurrent.TrieMap
 
-private[trace] class TracingContext(traceName: String, token: String, tags: Map[String, String], izOpen: Boolean, levelOfDetail: LevelOfDetail,
+private[trace] class TracingContext(traceName: String, token: String, tags: Map[String, String], currentStatus: Status, levelOfDetail: LevelOfDetail,
   isLocal: Boolean, startTimeztamp: RelativeNanoTimestamp, log: LoggingAdapter, traceInfoSink: TracingContext ⇒ Unit)
-    extends MetricsOnlyContext(traceName, token, tags, izOpen, levelOfDetail, startTimeztamp, log) {
+    extends MetricsOnlyContext(traceName, token, tags, currentStatus, levelOfDetail, startTimeztamp, log) {
 
   private val _openSegments = new AtomicInteger(0)
   private val _startTimestamp = NanoTimestamp.now
@@ -51,12 +52,12 @@ private[trace] class TracingContext(traceName: String, token: String, tags: Map[
     traceInfoSink(this)
   }
 
-  override def finishSegment(segmentName: String, category: String, library: String, duration: NanoInterval, tags: Map[String, String]): Unit = {
+  override def finishSegment(segmentName: String, category: String, library: String, duration: NanoInterval, tags: Map[String, String], isFinishedWithError: Boolean = false): Unit = {
     _openSegments.decrementAndGet()
-    super.finishSegment(segmentName, category, library, duration, tags)
+    super.finishSegment(segmentName, category, library, duration, tags, isFinishedWithError)
   }
 
-  def shouldIncubate: Boolean = isOpen || _openSegments.get() > 0
+  def shouldIncubate: Boolean = (States.Open == status) || _openSegments.get() > 0
 
   // Handle with care, should only be used after a trace is finished.
   def generateTraceInfo: TraceInfo = {