add optional span scoping to the span metrics

7 files changed, 126 insertions, 50 deletions

diff --git a/kamon-core-tests/src/test/scala/kamon/context/ContextCodecSpec.scala b/kamon-core-tests/src/test/scala/kamon/context/ContextCodecSpec.scala
index 11be85a7..242c3345 100644
--- a/kamon-core-tests/src/test/scala/kamon/context/ContextCodecSpec.scala
+++ b/kamon-core-tests/src/test/scala/kamon/context/ContextCodecSpec.scala
@@ -14,5 +14,5 @@ class ContextCodecSpec extends WordSpec with Matchers {
     }
   }
 
-  val ContextCodec = new Codec(Kamon.identityProvider, Kamon.config())
+  val ContextCodec = new Codec(Kamon.config())
 }
diff --git a/kamon-core-tests/src/test/scala/kamon/trace/SpanMetrics.scala b/kamon-core-tests/src/test/scala/kamon/trace/SpanMetrics.scala
index 9ecffb24..ae0c7795 100644
--- a/kamon-core-tests/src/test/scala/kamon/trace/SpanMetrics.scala
+++ b/kamon-core-tests/src/test/scala/kamon/trace/SpanMetrics.scala
@@ -3,16 +3,15 @@ package kamon.trace
 import kamon.Kamon
 import kamon.Kamon.buildSpan
 import kamon.metric._
+import kamon.testkit.{MetricInspection, Reconfigure}
 import org.scalatest.{Matchers, WordSpecLike}
 
-class SpanMetrics extends WordSpecLike with Matchers {
-  import SpanMetricsTestHelper._
+class SpanMetrics extends WordSpecLike with Matchers with MetricInspection with Reconfigure {
 
-  val errorTag = "error" -> "true"
-  val histogramMetric: HistogramMetric = Kamon.histogram("span.elapsed-time")
+  sampleAlways()
 
   "Span Metrics" should {
-    "be recorded for successeful execution" in {
+    "be recorded for successful execution" in {
       val operation = "span-success"
       val operationTag = "operation" -> operation
 
@@ -20,11 +19,11 @@ class SpanMetrics extends WordSpecLike with Matchers {
         .start()
         .finish()
 
-      val histogram = histogramMetric.refine(operationTag)
-      histogram.distribution().count === 1
+      val histogram = Span.Metrics.ProcessingTime.refine(Map(operationTag, noErrorTag))
+      histogram.distribution().count shouldBe 1
 
-      val errorHistogram = histogramMetric.refine(Map(operationTag, errorTag)).distribution()
-      errorHistogram.count === 0
+      val errorHistogram = Span.Metrics.ProcessingTime.refine(Map(operationTag, errorTag))
+      errorHistogram.distribution().count shouldBe 0
 
     }
 
@@ -37,28 +36,75 @@ class SpanMetrics extends WordSpecLike with Matchers {
         .addSpanTag("error", true)
         .finish()
 
-      val histogram = histogramMetric.refine(operationTag)
-      histogram.distribution().count === 0
+      val histogram = Span.Metrics.ProcessingTime.refine(Map(operationTag, noErrorTag))
+      histogram.distribution().count shouldBe 0
+
+      val errorHistogram = Span.Metrics.ProcessingTime.refine(Map(operationTag, errorTag))
+      errorHistogram.distribution().count shouldBe 1
+    }
 
-      val errorHistogram = histogramMetric.refine(operationTag, errorTag).distribution()
-      errorHistogram.count === 1
+    "add a parentOperation tag to the metrics if span metrics scoping is enabled" in {
+      val parent = buildSpan("parent").start()
+      val parentOperationTag = "parentOperation" -> "parent"
+
+      val operation = "span-with-parent"
+      val operationTag = "operation" -> operation
+
+      buildSpan(operation)
+        .asChildOf(parent)
+        .start()
+        .addSpanTag("error", false)
+        .finish()
 
+      buildSpan(operation)
+        .asChildOf(parent)
+        .start()
+        .addSpanTag("error", true)
+        .finish()
+
+      val histogram = Span.Metrics.ProcessingTime.refine(Map(operationTag, noErrorTag, parentOperationTag))
+      histogram.distribution().count shouldBe 1
+
+      val errorHistogram = Span.Metrics.ProcessingTime.refine(Map(operationTag, errorTag, parentOperationTag))
+      errorHistogram.distribution().count shouldBe 1
     }
-  }
 
-}
+    "not add any parentOperation tag to the metrics if span metrics scoping is disabled" in withoutSpanScopingEnabled {
+      val parent = buildSpan("parent").start()
+      val parentOperationTag = "parentOperation" -> "parent"
 
-object SpanMetricsTestHelper {
+      val operation = "span-with-parent"
+      val operationTag = "operation" -> operation
+
+      buildSpan(operation)
+        .asChildOf(parent)
+        .start()
+        .addSpanTag("error", false)
+        .finish()
 
-  implicit class HistogramMetricSyntax(histogram: Histogram) {
-    def distribution(resetState: Boolean = true): Distribution =
-      histogram match {
-        case hm: HistogramMetric    => hm.refine(Map.empty[String, String]).distribution(resetState)
-        case h: AtomicHdrHistogram  => h.snapshot(resetState).distribution
-        case h: HdrHistogram        => h.snapshot(resetState).distribution
-      }
+      buildSpan(operation)
+        .asChildOf(parent)
+        .start()
+        .addSpanTag("error", true)
+        .finish()
+
+      val histogram = Span.Metrics.ProcessingTime.refine(Map(operationTag, noErrorTag, parentOperationTag))
+      histogram.distribution().count shouldBe 0
+
+      val errorHistogram = Span.Metrics.ProcessingTime.refine(Map(operationTag, errorTag, parentOperationTag))
+      errorHistogram.distribution().count shouldBe 0
+    }
   }
-}
 
+  val errorTag = "error" -> "true"
+  val noErrorTag = "error" -> "false"
+
+  private def withoutSpanScopingEnabled[T](f: => T): T = {
+    disableSpanMetricScoping()
+    val evaluated = f
+    enableSpanMetricScoping()
+    evaluated
+  }
+}
 
 
diff --git a/kamon-core/src/main/resources/reference.conf b/kamon-core/src/main/resources/reference.conf
index ad180f1c..5b885fd0 100644
--- a/kamon-core/src/main/resources/reference.conf
+++ b/kamon-core/src/main/resources/reference.conf
@@ -117,6 +117,13 @@ kamon {
     # Any external implementation can be configured here, as long as it can be instantiated with a parameterless constructor.
     identity-provider = "kamon.trace.IdentityProvider$Default"
 
+    span-metrics {
+
+      # When this option is enabled the metrics collected for Spans will automatically add a tag named "parentOperation"
+      # with the name of the operation on the parent Span, if any.
+      scope-spans-to-parent = yes
+    }
+
     # The SpanContextCodecs are used to encode/decode the SpanContext data into simple TextMaps, HTTP Headers or Binary
     # carriers. The decision about which one to use is based on the kamon.trace.SpanContextCodec.Format instance passed
     # to inject/extract calls.
diff --git a/kamon-core/src/main/scala/kamon/context/Codec.scala b/kamon-core/src/main/scala/kamon/context/Codec.scala
index 50b7e93d..10580c22 100644
--- a/kamon-core/src/main/scala/kamon/context/Codec.scala
+++ b/kamon-core/src/main/scala/kamon/context/Codec.scala
@@ -8,7 +8,7 @@ import org.slf4j.LoggerFactory
 
 import scala.collection.mutable
 
-class Codec(identityProvider: IdentityProvider, initialConfig: Config) {
+class Codec(initialConfig: Config) {
   private val log = LoggerFactory.getLogger(classOf[Codec])
 
   @volatile private var httpHeaders: Codec.ForContext[TextMap] = new Codec.HttpHeaders(Map.empty)
diff --git a/kamon-core/src/main/scala/kamon/trace/Span.scala b/kamon-core/src/main/scala/kamon/trace/Span.scala
index 0ec71b32..ae67f7f6 100644
--- a/kamon-core/src/main/scala/kamon/trace/Span.scala
+++ b/kamon-core/src/main/scala/kamon/trace/Span.scala
@@ -89,8 +89,8 @@ object Span {
     * @param startTimestampMicros
     * @param spanSink
     */
-  final class Local(spanContext: SpanContext, initialOperationName: String, initialSpanTags: Map[String, Span.TagValue],
-    initialMetricTags: Map[String, String], startTimestampMicros: Long, spanSink: SpanSink) extends Span {
+  final class Local(spanContext: SpanContext, parent: Option[Span], initialOperationName: String, initialSpanTags: Map[String, Span.TagValue],
+    initialMetricTags: Map[String, String], startTimestampMicros: Long, spanSink: SpanSink, scopeSpanMetrics: Boolean) extends Span {
 
     private var collectMetrics: Boolean = true
     private var open: Boolean = true
@@ -167,26 +167,28 @@ object Span {
 
     private def recordSpanMetrics(endTimestampMicros: Long): Unit = {
       val elapsedTime = endTimestampMicros - startTimestampMicros
-      val metricTags = Map("operation" -> operationName) ++ customMetricTags
+      val isError = spanTags.get("error").map {
+        case boolean: TagValue.Boolean  => boolean.text
+        case _                          => TagValue.False.text
+      } getOrElse(TagValue.False.text)
 
-      val isError = spanTags.get("error").exists {
-        errorTag => errorTag != null && errorTag.equals(Span.TagValue.True)
-      }
+      if(scopeSpanMetrics)
+        parent.foreach(parentSpan => customMetricTags = customMetricTags + ("parentOperation" -> parentSpan.asInstanceOf[Local].operationName))
 
-      val refinedMetricTags = if(isError)
-        metricTags + ("error" -> "true")
-      else
-        metricTags
+      val metricTags = Map(
+        "operation" -> operationName,
+        "error" -> isError
+      ) ++ customMetricTags
 
-      val latencyHistogram = Span.Metrics.SpanProcessingTimeMetric.refine(refinedMetricTags)
-      latencyHistogram.record(elapsedTime)
+      Span.Metrics.ProcessingTime.refine(metricTags).record(elapsedTime)
     }
   }
 
   object Local {
-    def apply(spanContext: SpanContext, initialOperationName: String, initialSpanTags: Map[String, Span.TagValue],
-        initialMetricTags: Map[String, String], startTimestampMicros: Long, reporterRegistry: ReporterRegistryImpl): Local =
-      new Local(spanContext, initialOperationName, initialSpanTags, initialMetricTags, startTimestampMicros, reporterRegistry)
+    def apply(spanContext: SpanContext, parent: Option[Span], initialOperationName: String, initialSpanTags: Map[String, Span.TagValue],
+        initialMetricTags: Map[String, String], startTimestampMicros: Long, reporterRegistry: ReporterRegistryImpl,
+        scopeSpanMetrics: Boolean): Local =
+      new Local(spanContext, parent, initialOperationName, initialSpanTags, initialMetricTags, startTimestampMicros, reporterRegistry, scopeSpanMetrics)
   }
 
 
@@ -210,16 +212,25 @@ object Span {
 
   sealed trait TagValue
   object TagValue {
-    sealed trait Boolean extends TagValue
-    case object True extends Boolean
-    case object False extends Boolean
+
+    sealed trait Boolean extends TagValue {
+      def text: java.lang.String
+    }
+
+    case object True extends Boolean {
+      override def text: java.lang.String = "true"
+    }
+
+    case object False extends Boolean {
+      override def text: java.lang.String = "false"
+    }
 
     case class String(string: java.lang.String) extends TagValue
     case class Number(number: Long) extends TagValue
   }
 
   object Metrics {
-    val SpanProcessingTimeMetric = Kamon.histogram("span.processing-time", MeasurementUnit.time.microseconds)
+    val ProcessingTime = Kamon.histogram("span.processing-time", MeasurementUnit.time.microseconds)
     val SpanErrorCount = Kamon.counter("span.error-count")
   }
 
diff --git a/kamon-core/src/main/scala/kamon/trace/Tracer.scala b/kamon-core/src/main/scala/kamon/trace/Tracer.scala
index 7d8830ca..37706926 100644
--- a/kamon-core/src/main/scala/kamon/trace/Tracer.scala
+++ b/kamon-core/src/main/scala/kamon/trace/Tracer.scala
@@ -39,7 +39,8 @@ object Tracer {
 
     private[Tracer] val tracerMetrics = new TracerMetrics(metrics)
     @volatile private[Tracer] var joinRemoteParentsWithSameSpanID: Boolean = true
-    @volatile private[Tracer] var configuredSampler: Sampler = Sampler.Never
+    @volatile private[Tracer] var scopeSpanMetrics: Boolean = true
+    @volatile private[Tracer] var _sampler: Sampler = Sampler.Never
     @volatile private[Tracer] var _identityProvider: IdentityProvider = IdentityProvider.Default()
 
     reconfigure(initialConfig)
@@ -51,7 +52,7 @@ object Tracer {
       this._identityProvider
 
     def sampler: Sampler =
-      configuredSampler
+      _sampler
 
     private[kamon] def reconfigure(config: Config): Unit = synchronized {
       Try {
@@ -66,13 +67,14 @@ object Tracer {
         }
 
         val newJoinRemoteParentsWithSameSpanID = traceConfig.getBoolean("join-remote-parents-with-same-span-id")
-
+        val newScopeSpanMetrics = traceConfig.getBoolean("span-metrics.scope-spans-to-parent")
         val newIdentityProvider = dynamic.createInstanceFor[IdentityProvider](
           traceConfig.getString("identity-provider"), immutable.Seq.empty[(Class[_], AnyRef)]
         ).get
 
-        configuredSampler = newSampler
+        _sampler = newSampler
         joinRemoteParentsWithSameSpanID = newJoinRemoteParentsWithSameSpanID
+        scopeSpanMetrics = newScopeSpanMetrics
         _identityProvider = newIdentityProvider
 
       }.failed.foreach {
@@ -136,6 +138,8 @@ object Tracer {
         .orElse(if(useActiveSpanAsParent) Some(Kamon.currentContext().get(Span.ContextKey)) else None)
         .filter(span => span != Span.Empty)
 
+      val nonRemoteParent = parentSpan.filter(s => s.isLocal() && s.nonEmpty())
+
       val samplingDecision: SamplingDecision = parentSpan
         .map(_.context.samplingDecision)
         .filter(_ != SamplingDecision.Unknown)
@@ -147,7 +151,7 @@ object Tracer {
       }
 
       tracer.tracerMetrics.createdSpans.increment()
-      Span.Local(spanContext, operationName, initialSpanTags, initialMetricTags, startTimestampMicros, reporterRegistry)
+      Span.Local(spanContext, nonRemoteParent, operationName, initialSpanTags, initialMetricTags, startTimestampMicros, reporterRegistry, tracer.scopeSpanMetrics)
     }
 
     private def joinParentContext(parent: Span, samplingDecision: SamplingDecision): SpanContext =
diff --git a/kamon-testkit/src/main/scala/kamon/testkit/Reconfigure.scala b/kamon-testkit/src/main/scala/kamon/testkit/Reconfigure.scala
index 4b3b2cdb..6a5c4709 100644
--- a/kamon-testkit/src/main/scala/kamon/testkit/Reconfigure.scala
+++ b/kamon-testkit/src/main/scala/kamon/testkit/Reconfigure.scala
@@ -17,6 +17,14 @@ trait Reconfigure {
     applyConfig("kamon.trace.sampler = never")
   }
 
+  def enableSpanMetricScoping(): Unit = {
+    applyConfig("kamon.trace.span-metrics.scope-spans-to-parent = yes")
+  }
+
+  def disableSpanMetricScoping(): Unit = {
+    applyConfig("kamon.trace.span-metrics.scope-spans-to-parent = no")
+  }
+
   private def applyConfig(configString: String): Unit = {
     Kamon.reconfigure(ConfigFactory.parseString(configString).withFallback(Kamon.config()))
   }