diff options
author | Ivan Topolnjak <ivantopo@gmail.com> | 2017-08-14 17:30:16 +0200 |
---|---|---|
committer | Ivan Topolnjak <ivantopo@gmail.com> | 2017-08-14 17:30:16 +0200 |
commit | 3a8c0fa25f12230b27e943d1fffe07f814c650fe (patch) | |
tree | 75a12128af7387f40e3eba040812e1bd87b9a455 /kamon-core/src/main/scala | |
parent | a6113cf33ba1b98cc73d35176ccf8a2f76b77875 (diff) | |
download | Kamon-3a8c0fa25f12230b27e943d1fffe07f814c650fe.tar.gz Kamon-3a8c0fa25f12230b27e943d1fffe07f814c650fe.tar.bz2 Kamon-3a8c0fa25f12230b27e943d1fffe07f814c650fe.zip |
implement Span propagation on top of Kamon.Context
Diffstat (limited to 'kamon-core/src/main/scala')
-rw-r--r-- | kamon-core/src/main/scala/kamon/Kamon.scala | 31 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/context/Codec.scala | 132 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/context/Context.scala | 127 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/context/Mixin.scala (renamed from kamon-core/src/main/scala/kamon/util/Mixin.scala) | 21 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/context/Storage.scala | 39 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/trace/ActiveSpanStorage.scala | 74 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/trace/Span.scala | 78 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/trace/SpanContext.scala | 15 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/trace/SpanContextCodec.scala | 105 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/trace/Tracer.scala | 95 |
10 files changed, 280 insertions, 437 deletions
diff --git a/kamon-core/src/main/scala/kamon/Kamon.scala b/kamon-core/src/main/scala/kamon/Kamon.scala index 7c3beb84..5c33b3b5 100644 --- a/kamon-core/src/main/scala/kamon/Kamon.scala +++ b/kamon-core/src/main/scala/kamon/Kamon.scala @@ -24,7 +24,7 @@ import scala.concurrent.Future import java.time.Duration import java.util.concurrent.{Executors, ScheduledExecutorService, ScheduledThreadPoolExecutor} -import kamon.trace.SpanContextCodec.Format +import kamon.context.{Context, Storage} import org.slf4j.LoggerFactory import scala.util.Try @@ -41,6 +41,7 @@ object Kamon extends MetricLookup with ReporterRegistry with Tracer { private val _metrics = new MetricRegistry(_config, _scheduler) private val _reporters = new ReporterRegistryImpl(_metrics, _config) private val _tracer = Tracer.Default(Kamon, _reporters, _config) + private val _contextStorage = Storage.ThreadLocal() private var _onReconfigureHooks = Seq.empty[OnReconfigureHook] def environment: Environment = @@ -93,30 +94,16 @@ object Kamon extends MetricLookup with ReporterRegistry with Tracer { override def buildSpan(operationName: String): Tracer.SpanBuilder = _tracer.buildSpan(operationName) - override def extract[C](format: Format[C], carrier: C): Option[SpanContext] = - _tracer.extract(format, carrier) + def currentContext(): Context = + _contextStorage.current() - override def inject[C](spanContext: SpanContext, format: Format[C], carrier: C): C = - _tracer.inject(spanContext, format, carrier) - - override def inject[C](spanContext: SpanContext, format: Format[C]): C = - _tracer.inject(spanContext, format) - - override def activeSpan(): Span = - _tracer.activeSpan() - - override def activate(span: Span): Scope = - _tracer.activate(span) - - - /** - * Makes the provided Span active before code is evaluated and deactivates it afterwards. - */ - def withActiveSpan[T](span: Span)(code: => T): T = { - val scope = activate(span) + def storeContext(context: Context): Storage.Scope = + _contextStorage.store(context) + def withContext[T](context: Context)(f: => T): T = { + val scope = _contextStorage.store(context) try { - code + f } finally { scope.close() } diff --git a/kamon-core/src/main/scala/kamon/context/Codec.scala b/kamon-core/src/main/scala/kamon/context/Codec.scala new file mode 100644 index 00000000..957c3e26 --- /dev/null +++ b/kamon-core/src/main/scala/kamon/context/Codec.scala @@ -0,0 +1,132 @@ +package kamon +package context + +import com.typesafe.config.Config +import kamon.util.DynamicAccess +import org.slf4j.LoggerFactory +import scala.collection.mutable + +class Codec(initialConfig: Config) { + private val log = LoggerFactory.getLogger(classOf[Codec]) + + @volatile private var httpHeaders: Codec.ForContext[TextMap] = new Codec.HttpHeaders(Map.empty) + //val Binary: Codec.ForContext[ByteBuffer] = _ + reconfigure(initialConfig) + + + def HttpHeaders: Codec.ForContext[TextMap] = + httpHeaders + + def reconfigure(config: Config): Unit = { + httpHeaders = new Codec.HttpHeaders(readEntryCodecs("kamon.context.encoding.http-headers", config)) + + + // Kamon.contextCodec.httpHeaderExport(current) + // Kamon.exportContext(HTTP, context) + // Kamon.importContext(HTTP, textMap) + // Kamon.currentContext() + // Kamon.storeContext(context) + + } + + private def readEntryCodecs[T](rootKey: String, config: Config): Map[String, Codec.ForEntry[T]] = { + val rootConfig = config.getConfig(rootKey) + val dynamic = new DynamicAccess(getClass.getClassLoader) + val entries = Map.newBuilder[String, Codec.ForEntry[T]] + + rootConfig.topLevelKeys.foreach(key => { + try { + val fqcn = rootConfig.getString(key) + entries += ((key, dynamic.createInstanceFor[Codec.ForEntry[T]](fqcn, Nil).get)) + } catch { + case e: Throwable => + log.error(s"Failed to initialize codec for key [$key]", e) + } + }) + + entries.result() + } +} + +object Codec { + + trait ForContext[T] { + def encode(context: Context): T + def decode(carrier: T): Context + } + + trait ForEntry[T] { + def encode(context: Context): T + def decode(carrier: T, context: Context): Context + } + + final class HttpHeaders(entryCodecs: Map[String, Codec.ForEntry[TextMap]]) extends Codec.ForContext[TextMap] { + private val log = LoggerFactory.getLogger(classOf[HttpHeaders]) + + override def encode(context: Context): TextMap = { + val encoded = TextMap.Default() + + context.entries.foreach { + case (key, _) if key.broadcast => + entryCodecs.get(key.name) match { + case Some(codec) => + try { + codec.encode(context).values.foreach(pair => encoded.put(pair._1, pair._2)) + } catch { + case e: Throwable => log.error(s"Failed to encode key [${key.name}]", e) + } + + case None => + log.error("Context key [{}] should be encoded in HttpHeaders but no codec was found for it.", key.name) + } + } + + encoded + } + + override def decode(carrier: TextMap): Context = { + var context: Context = Context.Empty + + try { + context = entryCodecs.foldLeft(context)((ctx, codecEntry) => { + val (_, codec) = codecEntry + codec.decode(carrier, ctx) + }) + + } catch { + case e: Throwable => + log.error("Failed to decode context from HttpHeaders", e) + } + + context + } + } + +} + + +trait TextMap { + def get(key: String): Option[String] + + def put(key: String, value: String): Unit + + def values: Iterator[(String, String)] +} + +object TextMap { + + class Default extends TextMap { + private val storage = mutable.Map.empty[String, String] + + override def get(key: String): Option[String] = storage.get(key) + + override def put(key: String, value: String): Unit = storage.put(key, value) + + override def values: Iterator[(String, String)] = storage.toIterator + } + + object Default { + def apply(): Default = new Default() + } + +}
\ No newline at end of file diff --git a/kamon-core/src/main/scala/kamon/context/Context.scala b/kamon-core/src/main/scala/kamon/context/Context.scala index f7c78388..f8a4662f 100644 --- a/kamon-core/src/main/scala/kamon/context/Context.scala +++ b/kamon-core/src/main/scala/kamon/context/Context.scala @@ -1,18 +1,27 @@ package kamon.context -class Context private (private val keys: Map[Key[_], Any]) { +class Context private (private[context] val entries: Map[Key[_], Any]) { def get[T](key: Key[T]): T = - keys.get(key).getOrElse(key.emptyValue).asInstanceOf[T] + entries.get(key).getOrElse(key.emptyValue).asInstanceOf[T] def withKey[T](key: Key[T], value: T): Context = - new Context(keys.updated(key, value)) + new Context(entries.updated(key, value)) } object Context { val Empty = new Context(Map.empty) - def apply(): Context = Empty - def create(): Context = Empty + def apply(): Context = + Empty + + def create(): Context = + Empty + + def apply[T](key: Key[T], value: T): Context = + new Context(Map(key -> value)) + + def create[T](key: Key[T], value: T): Context = + apply(key, value) } @@ -38,110 +47,4 @@ object Key { override def equals(that: Any): Boolean = that.isInstanceOf[Default[_]] && that.asInstanceOf[Default[_]].name == this.name } -} - -trait Storage { - def current(): Context - def store(context: Context): Scope - - trait Scope { - def context: Context - def close(): Unit - } -} - -object Storage { - - class ThreadLocal extends Storage { - private val tls = new java.lang.ThreadLocal[Context]() { - override def initialValue(): Context = Context.Empty - } - - override def current(): Context = - tls.get() - - override def store(context: Context): Scope = { - val newContext = context - val previousContext = tls.get() - tls.set(newContext) - - new Scope { - override def context: Context = newContext - override def close(): Unit = tls.set(previousContext) - } - } - } -} - -trait KeyCodec[T] { - def encode(context: Context): T - def decode(carrier: T, context: Context): Context -} - -/* -object Example { - // this is defined somewhere statically, only once. - val User = Key.local[Option[User]]("user", None) - val Client = Key.local[Option[User]]("client", null) - val Span = Key.broadcast[Span]("span", EmptySpan) - val storage = Kamon.contextStorage // or something similar. - - storage.get(Span) // returns a Span instance or EmptySpan. - storage.get(User) // Returns Option[User] or None if not set. - storage.get(Client) // Returns Option[Client] or null if not set. - - // Context Propagation works the very same way as before. - - val scope = storage.store(context) - // do something here - scope.close() - - // Configuration for codecs would be handled sort of like this: - - // kamon.context.propagation { - // http-header-codecs { - // "span" = kamon.trace.propagation.B3 - // } - // - // binary-codecs { - // "span" = kamon.trace.propagation.Binary - // } - // } - - - - - -}*/ - - - -/* - - - - - -class Context(private val keys: Map[Key[_], Any]) { - - - - - -} - -object Context { - - -} - -sealed trait Key[T] { - def name: String -} - -object Key { - - def local[T](name: String): Key[T] = Local(name) - - case class Local[T](name: String) extends Key[T] -}*/ +}
\ No newline at end of file diff --git a/kamon-core/src/main/scala/kamon/util/Mixin.scala b/kamon-core/src/main/scala/kamon/context/Mixin.scala index 2fd7be24..52c97e84 100644 --- a/kamon-core/src/main/scala/kamon/util/Mixin.scala +++ b/kamon-core/src/main/scala/kamon/context/Mixin.scala @@ -13,34 +13,31 @@ * ========================================================================================= */ -package kamon -package util - -import kamon.trace.Span +package kamon.context /** * Utility trait that marks objects carrying a reference to a Span. * */ -trait HasSpan { - def span: Span +trait HasContext { + def context: Context } -object HasSpan { - private case class Default(span: Span) extends HasSpan +object HasContext { + private case class Default(context: Context) extends HasContext /** * Construct a HasSpan instance that references the provided Span. * */ - def from(span: Span): HasSpan = - Default(span) + def from(context: Context): HasContext = + Default(context) /** * Construct a HasSpan instance that references the currently ActiveSpan in Kamon's tracer. * */ - def fromActiveSpan(): HasSpan = - Default(Kamon.activeSpan()) +// def fromActiveSpan(): HasContext = +// Default(Kamon.activeSpan()) } diff --git a/kamon-core/src/main/scala/kamon/context/Storage.scala b/kamon-core/src/main/scala/kamon/context/Storage.scala new file mode 100644 index 00000000..6b92ff85 --- /dev/null +++ b/kamon-core/src/main/scala/kamon/context/Storage.scala @@ -0,0 +1,39 @@ +package kamon.context + +trait Storage { + def current(): Context + def store(context: Context): Storage.Scope +} + +object Storage { + + trait Scope { + def context: Context + def close(): Unit + } + + + class ThreadLocal extends Storage { + private val tls = new java.lang.ThreadLocal[Context]() { + override def initialValue(): Context = Context.Empty + } + + override def current(): Context = + tls.get() + + override def store(context: Context): Scope = { + val newContext = context + val previousContext = tls.get() + tls.set(newContext) + + new Scope { + override def context: Context = newContext + override def close(): Unit = tls.set(previousContext) + } + } + } + + object ThreadLocal { + def apply(): ThreadLocal = new ThreadLocal() + } +}
\ No newline at end of file diff --git a/kamon-core/src/main/scala/kamon/trace/ActiveSpanStorage.scala b/kamon-core/src/main/scala/kamon/trace/ActiveSpanStorage.scala deleted file mode 100644 index 85e94ef2..00000000 --- a/kamon-core/src/main/scala/kamon/trace/ActiveSpanStorage.scala +++ /dev/null @@ -1,74 +0,0 @@ -package kamon.trace - -/** - * A means of storing and retrieving the currently active Span. The application code execution is always considered to - * contribute to the completion of the operation represented by the currently active Span. - * - * The activation of a Span is of temporary nature and users of this API must ensure that all Scopes created via calls - * to `activate(span)` are timely closed; failing to do so might lead to unexpected behavior. Typically, the same block - * of code designating a Span as currently active will close the created Scope after finishing execution. - * - */ -trait ActiveSpanStorage { - - /** - * @return the currently active Span. - */ - def activeSpan(): Span - - /** - * Sets - * @param span the Span to be set as currently active. - * @return a [[Scope]] that will finish the designation of the given Span as active once it's closed. - */ - def activate(span: Span): Scope - -} - -/** - * Encapsulates the state (if any) required to handle the removal of a Span from it's currently active designation. - * - * Typically a Scope will enclose the previously active Span and return the previously active Span when closed, - * although no assumptions are made. - * - */ -trait Scope extends AutoCloseable { - - /** - * Removes the currently active Span from the ActiveSpanStorage. - * - */ - def close(): Unit -} - -object ActiveSpanStorage { - - /** - * A ActiveSpanStorage that uses a [[java.lang.ThreadLocal]] as the underlying storage. - * - */ - final class ThreadLocal extends ActiveSpanStorage { - private val emptySpan = Span.Empty(this) - private val storage: java.lang.ThreadLocal[Span] = new java.lang.ThreadLocal[Span] { - override def initialValue(): Span = emptySpan - } - - override def activeSpan(): Span = - storage.get() - - override def activate(span: Span): Scope = { - val previouslyActiveSpan = storage.get() - storage.set(span) - - new Scope { - override def close(): Unit = { - storage.set(previouslyActiveSpan) - } - } - } - } - - object ThreadLocal { - def apply(): ThreadLocal = new ThreadLocal() - } -}
\ No newline at end of file diff --git a/kamon-core/src/main/scala/kamon/trace/Span.scala b/kamon-core/src/main/scala/kamon/trace/Span.scala index 84cc5625..161042d5 100644 --- a/kamon-core/src/main/scala/kamon/trace/Span.scala +++ b/kamon-core/src/main/scala/kamon/trace/Span.scala @@ -17,15 +17,18 @@ package kamon package trace import kamon.ReporterRegistry.SpanSink +import kamon.context.Key import kamon.trace.SpanContext.SamplingDecision - import kamon.util.{Clock, MeasurementUnit} -/** - * Minimum set of capabilities that should be provided by a Span, all additional sugar is provided by extensions - * in the Span trait bellow. - */ -trait BaseSpan { + +trait Span { + + def isEmpty(): Boolean + def isLocal(): Boolean + + def nonEmpty(): Boolean = !isEmpty() + def isRemote(): Boolean = !isLocal() def context(): SpanContext @@ -39,21 +42,11 @@ trait BaseSpan { def addMetricTag(key: String, value: String): Span - def addBaggage(key: String, value: String): Span - - def getBaggage(key: String): Option[String] - def setOperationName(name: String): Span def disableMetricsCollection(): Span def finish(finishTimestampMicros: Long): Unit -} - -/** - * - */ -trait Span extends BaseSpan { def finish(): Unit = finish(Clock.microTimestamp()) @@ -71,25 +64,22 @@ trait Span extends BaseSpan { object Span { - final class Empty(activeSpanSource: ActiveSpanStorage) extends Span { - override val context: SpanContext = SpanContext.EmptySpanContext + val ContextKey = Key.broadcast[Span]("span", Span.Empty) + object Empty extends Span { + override val context: SpanContext = SpanContext.EmptySpanContext + override def isEmpty(): Boolean = true + override def isLocal(): Boolean = true override def annotate(annotation: Annotation): Span = this override def addSpanTag(key: String, value: String): Span = this override def addSpanTag(key: String, value: Long): Span = this override def addSpanTag(key: String, value: Boolean): Span = this override def addMetricTag(key: String, value: String): Span = this - override def addBaggage(key: String, value: String): Span = this - override def getBaggage(key: String): Option[String] = None override def setOperationName(name: String): Span = this override def disableMetricsCollection(): Span = this override def finish(finishTimestampMicros: Long): Unit = {} } - object Empty { - def apply(activeSpanSource: ActiveSpanStorage): Empty = new Empty(activeSpanSource) - } - /** * * @param spanContext @@ -98,8 +88,8 @@ object Span { * @param startTimestampMicros * @param spanSink */ - final class Real(spanContext: SpanContext, initialOperationName: String, initialSpanTags: Map[String, Span.TagValue], - initialMetricTags: Map[String, String], startTimestampMicros: Long, spanSink: SpanSink, activeSpanSource: ActiveSpanStorage) extends Span { + final class Local(spanContext: SpanContext, initialOperationName: String, initialSpanTags: Map[String, Span.TagValue], + initialMetricTags: Map[String, String], startTimestampMicros: Long, spanSink: SpanSink) extends Span { private var collectMetrics: Boolean = true private var open: Boolean = true @@ -110,6 +100,9 @@ object Span { private var customMetricTags = initialMetricTags private var annotations = List.empty[Span.Annotation] + override def isEmpty(): Boolean = false + override def isLocal(): Boolean = true + def annotate(annotation: Annotation): Span = synchronized { if(sampled && open) annotations = annotation :: annotations @@ -142,14 +135,6 @@ object Span { this } - override def addBaggage(key: String, value: String): Span = { - spanContext.baggage.add(key, value) - this - } - - override def getBaggage(key: String): Option[String] = - spanContext.baggage.get(key) - override def disableMetricsCollection(): Span = synchronized { collectMetrics = false this @@ -194,10 +179,29 @@ object Span { } } - object Real { + object Local { def apply(spanContext: SpanContext, initialOperationName: String, initialSpanTags: Map[String, Span.TagValue], - initialMetricTags: Map[String, String], startTimestampMicros: Long, reporterRegistry: ReporterRegistryImpl, tracer: Tracer): Real = - new Real(spanContext, initialOperationName, initialSpanTags, initialMetricTags, startTimestampMicros, reporterRegistry, tracer) + initialMetricTags: Map[String, String], startTimestampMicros: Long, reporterRegistry: ReporterRegistryImpl): Local = + new Local(spanContext, initialOperationName, initialSpanTags, initialMetricTags, startTimestampMicros, reporterRegistry) + } + + + final class Remote(val context: SpanContext) extends Span { + override def isEmpty(): Boolean = false + override def isLocal(): Boolean = false + override def annotate(annotation: Annotation): Span = this + override def addSpanTag(key: String, value: String): Span = this + override def addSpanTag(key: String, value: Long): Span = this + override def addSpanTag(key: String, value: Boolean): Span = this + override def addMetricTag(key: String, value: String): Span = this + override def setOperationName(name: String): Span = this + override def disableMetricsCollection(): Span = this + override def finish(finishTimestampMicros: Long): Unit = {} + } + + object Remote { + def apply(spanContext: SpanContext): Remote = + new Remote(spanContext) } sealed trait TagValue diff --git a/kamon-core/src/main/scala/kamon/trace/SpanContext.scala b/kamon-core/src/main/scala/kamon/trace/SpanContext.scala index ae92f46d..e8b239ba 100644 --- a/kamon-core/src/main/scala/kamon/trace/SpanContext.scala +++ b/kamon-core/src/main/scala/kamon/trace/SpanContext.scala @@ -16,7 +16,7 @@ package kamon.trace import kamon.trace.IdentityProvider.Identifier -import kamon.trace.SpanContext.{Baggage, SamplingDecision, Source} +import kamon.trace.SpanContext.SamplingDecision /** * @@ -24,9 +24,8 @@ import kamon.trace.SpanContext.{Baggage, SamplingDecision, Source} * @param spanID * @param parentID * @param samplingDecision - * @param baggage */ -case class SpanContext(traceID: Identifier, spanID: Identifier, parentID: Identifier, samplingDecision: SamplingDecision, baggage: Baggage, source: Source) { +case class SpanContext(traceID: Identifier, spanID: Identifier, parentID: Identifier, samplingDecision: SamplingDecision) { def createChild(childSpanID: Identifier, samplingDecision: SamplingDecision): SpanContext = this.copy(parentID = this.spanID, spanID = childSpanID) @@ -34,19 +33,11 @@ case class SpanContext(traceID: Identifier, spanID: Identifier, parentID: Identi object SpanContext { - sealed trait Source - object Source { - case object Local extends Source - case object Remote extends Source - } - val EmptySpanContext = SpanContext( traceID = IdentityProvider.NoIdentifier, spanID = IdentityProvider.NoIdentifier, parentID = IdentityProvider.NoIdentifier, - samplingDecision = SamplingDecision.DoNotSample, - baggage = Baggage.EmptyBaggage, - source = Source.Local + samplingDecision = SamplingDecision.DoNotSample ) diff --git a/kamon-core/src/main/scala/kamon/trace/SpanContextCodec.scala b/kamon-core/src/main/scala/kamon/trace/SpanContextCodec.scala index 43b5e8e4..1db55694 100644 --- a/kamon-core/src/main/scala/kamon/trace/SpanContextCodec.scala +++ b/kamon-core/src/main/scala/kamon/trace/SpanContextCodec.scala @@ -15,53 +15,36 @@ package kamon.trace -import java.lang.StringBuilder import java.net.{URLDecoder, URLEncoder} -import java.nio.ByteBuffer -import kamon.trace.SpanContext.{Baggage, SamplingDecision, Source} -import scala.collection.mutable -trait SpanContextCodec[T] { - def inject(spanContext: SpanContext, carrier: T): T - def inject(spanContext: SpanContext): T - def extract(carrier: T): Option[SpanContext] -} +import kamon.context.{Codec, Context, TextMap} +import kamon.trace.SpanContext.SamplingDecision -object SpanContextCodec { - sealed trait Format[C] - object Format { - case object TextMap extends Format[TextMap] - case object HttpHeaders extends Format[TextMap] - case object Binary extends Format[ByteBuffer] - } +object SpanContextCodec { - class ExtendedB3(identityProvider: IdentityProvider) extends SpanContextCodec[TextMap] { + class ExtendedB3(identityProvider: IdentityProvider) extends Codec.ForEntry[TextMap] { import ExtendedB3.Headers - override def inject(spanContext: SpanContext, carrier: TextMap): TextMap = { - if(spanContext != SpanContext.EmptySpanContext) { + override def encode(context: Context): TextMap = { + val span = context.get(Span.ContextKey) + val carrier = TextMap.Default() + + if(span.nonEmpty()) { + val spanContext = span.context carrier.put(Headers.TraceIdentifier, urlEncode(spanContext.traceID.string)) carrier.put(Headers.SpanIdentifier, urlEncode(spanContext.spanID.string)) carrier.put(Headers.ParentSpanIdentifier, urlEncode(spanContext.parentID.string)) - carrier.put(Headers.Baggage, encodeBaggage(spanContext.baggage)) encodeSamplingDecision(spanContext.samplingDecision).foreach { samplingDecision => carrier.put(Headers.Sampled, samplingDecision) } - - spanContext.baggage.get(Headers.Flags).foreach { flags => - carrier.put(Headers.Flags, flags) - } } carrier } - override def inject(spanContext: SpanContext): TextMap = - inject(spanContext, TextMap.Default()) - - override def extract(carrier: TextMap): Option[SpanContext] = { + override def decode(carrier: TextMap, context: Context): Context = { val traceID = carrier.get(Headers.TraceIdentifier) .map(id => identityProvider.traceIdentifierGenerator().from(urlDecode(id))) .getOrElse(IdentityProvider.NoIdentifier) @@ -75,56 +58,17 @@ object SpanContextCodec { .map(id => identityProvider.spanIdentifierGenerator().from(urlDecode(id))) .getOrElse(IdentityProvider.NoIdentifier) - val baggage = decodeBaggage(carrier.get(Headers.Baggage)) val flags = carrier.get(Headers.Flags) - flags.foreach { flags => - baggage.add(Headers.Flags, flags) - } - val samplingDecision = flags.orElse(carrier.get(Headers.Sampled)) match { case Some(sampled) if sampled == "1" => SamplingDecision.Sample case Some(sampled) if sampled == "0" => SamplingDecision.DoNotSample case _ => SamplingDecision.Unknown } - Some(SpanContext(traceID, spanID, parentID, samplingDecision, baggage, Source.Remote)) - - } else None - } - - private def encodeBaggage(baggage: Baggage): String = { - if(baggage.getAll().nonEmpty) { - val encodedBaggage = new StringBuilder() - baggage.getAll().foreach { - case (key, value) => - if(key != Headers.Flags) { - if (encodedBaggage.length() > 0) - encodedBaggage.append(';') - - encodedBaggage - .append(urlEncode(key)) - .append('=') - .append(urlEncode(value)) - } - } - - encodedBaggage.toString() - } else "" - } + context.withKey(Span.ContextKey, Span.Remote(SpanContext(traceID, spanID, parentID, samplingDecision))) - private def decodeBaggage(encodedBaggage: Option[String]): Baggage = { - val baggage = Baggage() - encodedBaggage.foreach { baggageString => - baggageString.split(";").foreach { group => - val pair = group.split("=") - if(pair.length >= 2 && pair(0).nonEmpty) { - baggage.add(urlDecode(pair(0)), urlDecode(pair(1))) - } - } - } - - baggage + } else context } private def encodeSamplingDecision(samplingDecision: SamplingDecision): Option[String] = samplingDecision match { @@ -135,7 +79,6 @@ object SpanContextCodec { private def urlEncode(s: String): String = URLEncoder.encode(s, "UTF-8") private def urlDecode(s: String): String = URLDecoder.decode(s, "UTF-8") - } object ExtendedB3 { @@ -149,26 +92,6 @@ object SpanContextCodec { val SpanIdentifier = "X-B3-SpanId" val Sampled = "X-B3-Sampled" val Flags = "X-B3-Flags" - val Baggage = "X-B3-Extra-Baggage" } } -} - -trait TextMap { - def get(key: String): Option[String] - def put(key: String, value: String): Unit - def values: Iterator[(String, String)] -} - -object TextMap { - class Default extends TextMap { - private val storage = mutable.Map.empty[String, String] - override def get(key: String): Option[String] = storage.get(key) - override def put(key: String, value: String): Unit = storage.put(key, value) - override def values: Iterator[(String, String)] = storage.toIterator - } - - object Default { - def apply(): Default = new Default() - } -} +}
\ No newline at end of file diff --git a/kamon-core/src/main/scala/kamon/trace/Tracer.scala b/kamon-core/src/main/scala/kamon/trace/Tracer.scala index bfdd561d..65307b95 100644 --- a/kamon-core/src/main/scala/kamon/trace/Tracer.scala +++ b/kamon-core/src/main/scala/kamon/trace/Tracer.scala @@ -15,13 +15,11 @@ package kamon.trace -import java.nio.ByteBuffer - import com.typesafe.config.Config -import kamon.ReporterRegistryImpl +import kamon.{Kamon, ReporterRegistryImpl} import kamon.metric.MetricLookup import kamon.trace.Span.TagValue -import kamon.trace.SpanContext.{SamplingDecision, Source} +import kamon.trace.SpanContext.SamplingDecision import kamon.trace.Tracer.SpanBuilder import kamon.util.{Clock, DynamicAccess} import org.slf4j.LoggerFactory @@ -29,56 +27,25 @@ import org.slf4j.LoggerFactory import scala.collection.immutable import scala.util.Try -trait Tracer extends ActiveSpanStorage { +trait Tracer { def buildSpan(operationName: String): SpanBuilder - - def extract[C](format: SpanContextCodec.Format[C], carrier: C): Option[SpanContext] - def inject[C](spanContext: SpanContext, format: SpanContextCodec.Format[C], carrier: C): C - def inject[C](spanContext: SpanContext, format: SpanContextCodec.Format[C]): C } object Tracer { final class Default(metrics: MetricLookup, reporterRegistry: ReporterRegistryImpl, initialConfig: Config) extends Tracer { private val logger = LoggerFactory.getLogger(classOf[Tracer]) - private val activeSpanSource = ActiveSpanStorage.ThreadLocal() 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 identityProvider: IdentityProvider = IdentityProvider.Default() - @volatile private[Tracer] var textMapSpanContextCodec: SpanContextCodec[TextMap] = SpanContextCodec.ExtendedB3(identityProvider) - @volatile private[Tracer] var httpHeaderSpanContextCodec: SpanContextCodec[TextMap] = SpanContextCodec.ExtendedB3(identityProvider) reconfigure(initialConfig) override def buildSpan(operationName: String): SpanBuilder = new SpanBuilder(operationName, this, reporterRegistry) - override def extract[C](format: SpanContextCodec.Format[C], carrier: C): Option[SpanContext] = format match { - case SpanContextCodec.Format.HttpHeaders => httpHeaderSpanContextCodec.extract(carrier.asInstanceOf[TextMap]) - case SpanContextCodec.Format.TextMap => textMapSpanContextCodec.extract(carrier.asInstanceOf[TextMap]) - case SpanContextCodec.Format.Binary => None - } - - override def inject[C](spanContext: SpanContext, format: SpanContextCodec.Format[C], carrier: C): C = format match { - case SpanContextCodec.Format.HttpHeaders => httpHeaderSpanContextCodec.inject(spanContext, carrier.asInstanceOf[TextMap]) - case SpanContextCodec.Format.TextMap => textMapSpanContextCodec.inject(spanContext, carrier.asInstanceOf[TextMap]) - case SpanContextCodec.Format.Binary => carrier - } - - override def inject[C](spanContext: SpanContext, format: SpanContextCodec.Format[C]): C = format match { - case SpanContextCodec.Format.HttpHeaders => httpHeaderSpanContextCodec.inject(spanContext) - case SpanContextCodec.Format.TextMap => textMapSpanContextCodec.inject(spanContext) - case SpanContextCodec.Format.Binary => ByteBuffer.allocate(0) // TODO: Implement binary encoding. - } - - override def activeSpan(): Span = - activeSpanSource.activeSpan() - - override def activate(span: Span): Scope = - activeSpanSource.activate(span) - def sampler: Sampler = configuredSampler @@ -100,25 +67,9 @@ object Tracer { traceConfig.getString("identity-provider"), immutable.Seq.empty[(Class[_], AnyRef)] ).get - val spanContextCodecs = traceConfig.getConfig("span-context-codec") - val newTextMapSpanContextCodec = dynamic.createInstanceFor[SpanContextCodec[TextMap]]( - spanContextCodecs.getString("text-map"), immutable.Seq((classOf[IdentityProvider], newIdentityProvider)) - ).get - - val newHttpHeadersSpanContextCodec = dynamic.createInstanceFor[SpanContextCodec[TextMap]]( - spanContextCodecs.getString("http-headers"), immutable.Seq((classOf[IdentityProvider], newIdentityProvider)) - ).get - -// val newBinarySpanContextCodec = dynamic.createInstanceFor[SpanContextCodec[TextMap]]( -// spanContextCodecs.getString("binary"), immutable.Seq((classOf[IdentityProvider], newIdentityProvider)) -// ).get // TODO: Make it happen! - - configuredSampler = newSampler joinRemoteParentsWithSameSpanID = newJoinRemoteParentsWithSameSpanID identityProvider = newIdentityProvider - textMapSpanContextCodec = newTextMapSpanContextCodec - httpHeaderSpanContextCodec = newHttpHeadersSpanContextCodec }.failed.foreach { ex => logger.error("Unable to reconfigure Kamon Tracer", ex) @@ -132,25 +83,17 @@ object Tracer { } final class SpanBuilder(operationName: String, tracer: Tracer.Default, reporterRegistry: ReporterRegistryImpl) { - private var parentContext: SpanContext = _ + private var parentSpan: Span = _ private var startTimestamp = 0L private var initialSpanTags = Map.empty[String, Span.TagValue] private var initialMetricTags = Map.empty[String, String] private var useActiveSpanAsParent = true - def asChildOf(parentContext: SpanContext): SpanBuilder = { - this.parentContext = parentContext + def asChildOf(parent: Span): SpanBuilder = { + if(parent != Span.Empty) this.parentSpan = parent this } - def asChildOf(parentContext: Option[SpanContext]): SpanBuilder = { - parentContext.foreach(asChildOf) - this - } - - def asChildOf(parentSpan: Span): SpanBuilder = - asChildOf(parentSpan.context()) - def withMetricTag(key: String, value: String): SpanBuilder = { this.initialMetricTags = this.initialMetricTags + (key -> value) this @@ -185,38 +128,36 @@ object Tracer { def start(): Span = { val startTimestampMicros = if(startTimestamp != 0L) startTimestamp else Clock.microTimestamp() - val parentSpanContext: Option[SpanContext] = Option(parentContext) - .orElse(if(useActiveSpanAsParent) Some(tracer.activeSpan().context()) else None) - .filter(spanContext => spanContext != SpanContext.EmptySpanContext) + val parentSpan: Option[Span] = Option(this.parentSpan) + .orElse(if(useActiveSpanAsParent) Some(Kamon.currentContext().get(Span.ContextKey)) else None) + .filter(span => span != Span.Empty) - val samplingDecision: SamplingDecision = parentSpanContext - .map(_.samplingDecision) + val samplingDecision: SamplingDecision = parentSpan + .map(_.context.samplingDecision) .filter(_ != SamplingDecision.Unknown) .getOrElse(tracer.sampler.decide(operationName, initialSpanTags)) - val spanContext = parentSpanContext match { + val spanContext = parentSpan match { case Some(parent) => joinParentContext(parent, samplingDecision) case None => newSpanContext(samplingDecision) } tracer.tracerMetrics.createdSpans.increment() - Span.Real(spanContext, operationName, initialSpanTags, initialMetricTags, startTimestampMicros, reporterRegistry, tracer) + Span.Local(spanContext, operationName, initialSpanTags, initialMetricTags, startTimestampMicros, reporterRegistry) } - private def joinParentContext(parent: SpanContext, samplingDecision: SamplingDecision): SpanContext = - if(parent.source == Source.Remote && tracer.joinRemoteParentsWithSameSpanID) - parent.copy(samplingDecision = samplingDecision) + private def joinParentContext(parent: Span, samplingDecision: SamplingDecision): SpanContext = + if(parent.isRemote() && tracer.joinRemoteParentsWithSameSpanID) + parent.context().copy(samplingDecision = samplingDecision) else - parent.createChild(tracer.identityProvider.spanIdentifierGenerator().generate(), samplingDecision) + parent.context().createChild(tracer.identityProvider.spanIdentifierGenerator().generate(), samplingDecision) private def newSpanContext(samplingDecision: SamplingDecision): SpanContext = SpanContext( traceID = tracer.identityProvider.traceIdentifierGenerator().generate(), spanID = tracer.identityProvider.spanIdentifierGenerator().generate(), parentID = IdentityProvider.NoIdentifier, - samplingDecision = samplingDecision, - baggage = SpanContext.Baggage(), - source = Source.Local + samplingDecision = samplingDecision ) } |