Add W3C Trace Context propagation (#949)

* Add W3C Trace Context propagation

* Update documentation
* Check identifier-scheme for W3C propagation

Author: duxet

core/kamon-core-tests/src/test/scala/kamon/context/HttpPropagationSpec.scala

@@ -113,7 +113,7 @@ class HttpPropagationSpec extends WordSpec with Matchers with OptionValues {
         |entries.outgoing.string = "kamon.context.HttpPropagationSpec$StringEntryCodec"
         |
       """.stripMargin
-    ).withFallback(ConfigFactory.load().getConfig("kamon.propagation")))
+    ).withFallback(ConfigFactory.load().getConfig("kamon.propagation")), identifierScheme = "single")
 
 
   def headerReaderFromMap(map: Map[String, String]): HttpPropagation.HeaderReader = new HttpPropagation.HeaderReader {

core/kamon-core-tests/src/test/scala/kamon/trace/W3CTraceContextSpanPropagationSpec.scala

@@ -0,0 +1,75 @@
+package kamon.trace
+
+import kamon.context.{Context, HttpPropagation}
+import kamon.trace.Trace.SamplingDecision
+import org.scalatest.{Matchers, OptionValues, WordSpecLike}
+
+import scala.collection.mutable
+
+class W3CTraceContextSpanPropagationSpec extends WordSpecLike with Matchers with OptionValues {
+  val traceContextPropagation = SpanPropagation.W3CTraceContext()
+
+  "The TraceContext Span propagation for HTTP" should {
+    "write the Span data into headers" in {
+      val headersMap = mutable.Map.empty[String, String]
+      traceContextPropagation.write(testContext(), headerWriterFromMap(headersMap))
+
+      headersMap.get("traceparent").value shouldBe "00-00000000000000000000000001020304-0000000004030201-01"
+      headersMap.get("tracestate").value shouldBe ""
+    }
+
+    "not inject anything if there is no Span in the Context" in {
+      val headersMap = mutable.Map.empty[String, String]
+      traceContextPropagation.write(Context.Empty, headerWriterFromMap(headersMap))
+      headersMap.values shouldBe empty
+    }
+
+    "extract a RemoteSpan from incoming headers when all fields are set" in {
+      val headersMap = Map(
+        "traceparent" -> "00-00000000000000000000000001020304-0000000004030201-01",
+        "tracestate" -> "2222"
+      )
+
+      val spanContext = traceContextPropagation.read(headerReaderFromMap(headersMap), Context.Empty).get(Span.Key)
+      spanContext.id.string shouldBe "00000000000000000000000001020304"
+      spanContext.parentId.string shouldBe "0000000004030201"
+      spanContext.trace.id.string shouldBe "00000000000000000000000001020304"
+      spanContext.trace.samplingDecision shouldBe SamplingDecision.Sample
+    }
+  }
+
+  def headerWriterFromMap(map: mutable.Map[String, String]): HttpPropagation.HeaderWriter = new HttpPropagation.HeaderWriter {
+    override def write(header: String, value: String): Unit = map.put(header, value)
+  }
+
+  def headerReaderFromMap(map: Map[String, String]): HttpPropagation.HeaderReader = new HttpPropagation.HeaderReader {
+    override def read(header: String): Option[String] = {
+      if(map.contains("fail"))
+        sys.error("failing on purpose")
+
+      map.get(header)
+    }
+
+    override def readAll(): Map[String, String] = map
+  }
+
+  def testContext(): Context =
+    Context.of(Span.Key, Span.Remote(
+      id = Identifier("4321", Array[Byte](4, 3, 2, 1)),
+      parentId = Identifier("2222", Array[Byte](2, 2, 2, 2)),
+      trace = Trace(
+        id = Identifier("1234", Array[Byte](1, 2, 3, 4)),
+        samplingDecision = SamplingDecision.Sample
+      )
+    ))
+
+  def testContextWithoutParent(): Context =
+    Context.of(Span.Key, Span.Remote(
+      id = Identifier("4321", Array[Byte](4, 3, 2, 1)),
+      parentId = Identifier.Empty,
+      trace = Trace(
+        id = Identifier("1234", Array[Byte](1, 2, 3, 4)),
+        samplingDecision = SamplingDecision.Sample
+      )
+    ))
+}

core/kamon-core/src/main/resources/reference.conf

@@ -315,6 +315,7 @@ kamon {
             #   - b3 for the B3 format (see https://github.com/openzipkin/b3-propagation)
             #   - b3-single for the Zipkin B3 single header format (see https://github.com/openzipkin/b3-propagation#single-header)
             #   - uber for the Uber/Jaeger format (see https://www.jaegertracing.io/docs/1.16/client-libraries/#propagation-format)
+            #   - w3c for the W3C Trace Context format (see https://www.w3.org/TR/trace-context-1/) - requires identifier-scheme = double
             #
             span = "b3"
           }
@@ -328,6 +329,7 @@ kamon {
             #   - b3 for the B3 format (see https://github.com/openzipkin/b3-propagation)
             #   - b3-single for the Zipkin B3 single header format (see https://github.com/openzipkin/b3-propagation#single-header)
             #   - uber for the Uber/Jaeger format (see https://www.jaegertracing.io/docs/1.16/client-libraries/#propagation-format)
+            #   - w3c for the W3C Trace Context format (see https://www.w3.org/TR/trace-context-1/) - requires identifier-scheme = double
             #
             span = "b3"
           }

core/kamon-core/src/main/scala/kamon/ContextPropagation.scala

@@ -89,9 +89,10 @@ object ContextPropagation {
       val propagationConfig = config.getConfig("kamon.propagation")
       val httpChannelsConfig = propagationConfig.getConfig("http").configurations
       val binaryChannelsConfig = propagationConfig.getConfig("binary").configurations
+      val identifierScheme = config.getString("kamon.trace.identifier-scheme")
 
       val httpChannels = httpChannelsConfig.map {
-        case (channelName, channelConfig) => (channelName -> HttpPropagation.from(channelConfig))
+        case (channelName, channelConfig) => (channelName -> HttpPropagation.from(channelConfig, identifierScheme))
       }
 
       val binaryChannels = binaryChannelsConfig.map {

core/kamon-core/src/main/scala/kamon/context/HttpPropagation.scala

@@ -64,8 +64,8 @@ object HttpPropagation {
   /**
     * Create a new HTTP propagation instance from the provided configuration.
     */
-  def from(config: Config): Propagation[HttpPropagation.HeaderReader, HttpPropagation.HeaderWriter] = {
-    new HttpPropagation.Default(Settings.from(config))
+  def from(propagationConfig: Config, identifierScheme: String): Propagation[HttpPropagation.HeaderReader, HttpPropagation.HeaderWriter] = {
+    new HttpPropagation.Default(Settings.from(propagationConfig, identifierScheme))
   }
 
   /**
@@ -226,16 +226,22 @@ object HttpPropagation {
     private val readerWriterShortcuts: Map[String, ReaderWriter] = Map(
       "span/b3" -> SpanPropagation.B3(),
       "span/uber" -> SpanPropagation.Uber(),
-      "span/b3-single" -> SpanPropagation.B3Single()
+      "span/b3-single" -> SpanPropagation.B3Single(),
+      "span/w3c" -> SpanPropagation.W3CTraceContext()
     )
 
-    def from(config: Config): Settings = {
+    def from(config: Config, identifierScheme: String): Settings = {
       def buildInstances[ExpectedType : ClassTag](mappings: Map[String, String]): Map[String, ExpectedType] = {
         val instanceMap = Map.newBuilder[String, ExpectedType]
 
         mappings.foreach {
           case (contextKey, componentClass) =>
             val shortcut = s"$contextKey/$componentClass".toLowerCase()
+
+            if (componentClass == "span/w3c" && identifierScheme != "double") {
+              log.warn("W3C TraceContext propagation should be used only with identifier-scheme = double")
+            }
+
             readerWriterShortcuts.get(shortcut).fold({
               try {
                 instanceMap += (contextKey -> ClassLoading.createInstance[ExpectedType](componentClass, Nil))

core/kamon-core/src/main/scala/kamon/trace/SpanPropagation.scala

@@ -40,6 +40,76 @@ object SpanPropagation {
 
   import Util._
 
+  /**
+    * Reads and Writes a Span instance using the W3C Trace Context propagation format.
+    * The specification can be found here: https://www.w3.org/TR/trace-context-1/
+    */
+  class W3CTraceContext extends Propagation.EntryReader[HeaderReader] with Propagation.EntryWriter[HeaderWriter] {
+    import W3CTraceContext._
+
+    override def read(medium: HeaderReader, context: Context): Context = {
+      val contextWithParent = for {
+        traceParent <- medium.read(Headers.TraceParent)
+        span <- decodeTraceParent(traceParent)
+      } yield {
+        val traceState = medium.read(Headers.TraceState).getOrElse("")
+        context.withEntry(Span.Key, span).withEntry(TraceStateKey, traceState)
+      }
+
+      contextWithParent.getOrElse(context)
+    }
+
+    override def write(context: Context, medium: HeaderWriter): Unit = {
+      val span = context.get(Span.Key)
+
+      if (span != Span.Empty) {
+        medium.write(Headers.TraceParent, encodeTraceParent(span))
+        medium.write(Headers.TraceState, context.get(TraceStateKey))
+      }
+    }
+  }
+
+object W3CTraceContext {
+  val Version: String = "00"
+  val TraceStateKey: Context.Key[String] = Context.key("tracestate", "")
+
+  object Headers {
+    val TraceParent = "traceparent"
+    val TraceState = "tracestate"
+  }
+
+  def apply(): W3CTraceContext =
+    new W3CTraceContext()
+
+  def decodeTraceParent(traceParent: String): Option[Span] = {
+    val identityProvider = Identifier.Scheme.Double
+
+    def unpackSamplingDecision(decision: String): SamplingDecision =
+      if ("01" == decision) SamplingDecision.Sample else SamplingDecision.Unknown
+
+    val traceParentComponents = traceParent.split("-")
+
+    if (traceParentComponents.length != 4) None else {
+      val traceID = identityProvider.traceIdFactory.from(traceParentComponents(1))
+      val spanID = identityProvider.spanIdFactory.from(traceParentComponents(2))
+      val samplingDecision = unpackSamplingDecision(traceParentComponents(3))
+
+      Some(Span.Remote(traceID, spanID, Trace(traceID, samplingDecision)))
+    }
+  }
+
+  def encodeTraceParent(parent: Span): String = {
+    def idToHex(identifier: Identifier, length: Int): String = {
+      val leftPad = (string: String) => "0" * (length - string.length) + string
+      leftPad(identifier.bytes.map("%02X" format _).mkString)
+    }
+
+    val samplingDecision = if (parent.trace.samplingDecision == SamplingDecision.Sample) "01" else "00"
+
+    s"$Version-${idToHex(parent.trace.id, 32)}-${idToHex(parent.id, 16)}-${samplingDecision}"
+  }
+}
+
   /**
     * Reads and Writes a Span instance using the B3 propagation format. The specification and semantics of the B3
     * Propagation format can be found here: https://github.com/openzipkin/b3-propagation