add optional span scoping to the span metrics

4 files changed, 47 insertions, 25 deletions

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 =