diff options
author | Ivan Topolnjak <ivantopo@gmail.com> | 2014-10-31 03:14:32 +0100 |
---|---|---|
committer | Ivan Topolnjak <ivantopo@gmail.com> | 2014-10-31 03:14:32 +0100 |
commit | 8ac7d1e06be61f53eb90abb4d42b67b476d35317 (patch) | |
tree | b565973c8b6aa569dcf09d5d06ac48ff10178c7b /kamon-core/src/main | |
parent | 30940181424be69e0fd64e945fe5a64b4523457b (diff) | |
parent | e93f4ee1e845ed07b9c50df5de530196cef5774e (diff) | |
download | Kamon-8ac7d1e06be61f53eb90abb4d42b67b476d35317.tar.gz Kamon-8ac7d1e06be61f53eb90abb4d42b67b476d35317.tar.bz2 Kamon-8ac7d1e06be61f53eb90abb4d42b67b476d35317.zip |
Merge branch 'wip/improve-trace-context-and-segments-api', closes #65
Diffstat (limited to 'kamon-core/src/main')
6 files changed, 173 insertions, 134 deletions
diff --git a/kamon-core/src/main/scala/kamon/instrumentation/akka/ActorCellInstrumentation.scala b/kamon-core/src/main/scala/kamon/instrumentation/akka/ActorCellInstrumentation.scala index 9b541a32..90928ba0 100644 --- a/kamon-core/src/main/scala/kamon/instrumentation/akka/ActorCellInstrumentation.scala +++ b/kamon-core/src/main/scala/kamon/instrumentation/akka/ActorCellInstrumentation.scala @@ -59,7 +59,7 @@ class ActorCellInstrumentation { def aroundBehaviourInvoke(pjp: ProceedingJoinPoint, cell: ActorCell, envelope: Envelope): Any = { val cellWithMetrics = cell.asInstanceOf[ActorCellMetrics] val timestampBeforeProcessing = System.nanoTime() - val contextAndTimestamp = envelope.asInstanceOf[TraceContextAware] + val contextAndTimestamp = envelope.asInstanceOf[TimestampedTraceContextAware] try { TraceRecorder.withInlineTraceContextReplacement(contextAndTimestamp.traceContext) { @@ -154,13 +154,13 @@ class ActorCellMetricsIntoActorCellMixin { class TraceContextIntoEnvelopeMixin { @DeclareMixin("akka.dispatch.Envelope") - def mixinTraceContextAwareToEnvelope: TraceContextAware = TraceContextAware.default + def mixinTraceContextAwareToEnvelope: TimestampedTraceContextAware = TimestampedTraceContextAware.default @Pointcut("execution(akka.dispatch.Envelope.new(..)) && this(ctx)") - def envelopeCreation(ctx: TraceContextAware): Unit = {} + def envelopeCreation(ctx: TimestampedTraceContextAware): Unit = {} @After("envelopeCreation(ctx)") - def afterEnvelopeCreation(ctx: TraceContextAware): Unit = { + def afterEnvelopeCreation(ctx: TimestampedTraceContextAware): Unit = { // Necessary to force the initialization of ContextAware at the moment of creation. ctx.traceContext } diff --git a/kamon-core/src/main/scala/kamon/metric/TraceMetrics.scala b/kamon-core/src/main/scala/kamon/metric/TraceMetrics.scala index 10dbcc01..7246ccb5 100644 --- a/kamon-core/src/main/scala/kamon/metric/TraceMetrics.scala +++ b/kamon-core/src/main/scala/kamon/metric/TraceMetrics.scala @@ -30,12 +30,11 @@ object TraceMetrics extends MetricGroupCategory { val name = "trace" case object ElapsedTime extends MetricIdentity { val name = "elapsed-time" } - case class HttpClientRequest(name: String) extends MetricIdentity case class TraceMetricRecorder(elapsedTime: Histogram, private val segmentRecorderFactory: () ⇒ Histogram) extends MetricGroupRecorder { - private val segments = TrieMap[MetricIdentity, Histogram]() + val segments = TrieMap[MetricIdentity, Histogram]() def segmentRecorder(segmentIdentity: MetricIdentity): Histogram = segments.getOrElseUpdate(segmentIdentity, segmentRecorderFactory.apply()) diff --git a/kamon-core/src/main/scala/kamon/trace/TraceContext.scala b/kamon-core/src/main/scala/kamon/trace/TraceContext.scala index 64ee70be..c4c28a68 100644 --- a/kamon-core/src/main/scala/kamon/trace/TraceContext.scala +++ b/kamon-core/src/main/scala/kamon/trace/TraceContext.scala @@ -23,113 +23,73 @@ import kamon.Kamon import kamon.metric._ import java.util.concurrent.ConcurrentLinkedQueue import kamon.trace.TraceContextAware.DefaultTraceContextAware -import kamon.trace.TraceContext.SegmentIdentity import kamon.metric.TraceMetrics.TraceMetricRecorder import scala.annotation.tailrec -trait TraceContext { +sealed trait TraceContext { def name: String def token: String - def system: ActorSystem def rename(name: String): Unit - def levelOfDetail: TracingLevelOfDetail - def startSegment(identity: SegmentIdentity, metadata: Map[String, String]): SegmentCompletionHandle - def finish(metadata: Map[String, String]) + def finish(): Unit def origin: TraceContextOrigin - def startMilliTime: Long def isOpen: Boolean - - private[kamon] val traceLocalStorage: TraceLocalStorage = new TraceLocalStorage -} - -object TraceContext { - type SegmentIdentity = MetricIdentity -} - -trait SegmentCompletionHandle { - def finish(metadata: Map[String, String] = Map.empty) -} - -case class SegmentData(identity: MetricIdentity, duration: Long, metadata: Map[String, String]) - -sealed trait TracingLevelOfDetail -case object OnlyMetrics extends TracingLevelOfDetail -case object SimpleTrace extends TracingLevelOfDetail -case object FullTrace extends TracingLevelOfDetail - -sealed trait TraceContextOrigin -object TraceContextOrigin { - case object Local extends TraceContextOrigin - case object Remote extends TraceContextOrigin + def isEmpty: Boolean + def nonEmpty: Boolean = !isEmpty + def startSegment(segmentName: String, label: String): Segment + def nanoTimestamp: Long } -trait TraceContextAware extends Serializable { - def captureNanoTime: Long - def traceContext: Option[TraceContext] +sealed trait Segment { + def name: String + def rename(newName: String): Unit + def label: String + def finish(): Unit + def isEmpty: Boolean } -object TraceContextAware { - def default: TraceContextAware = new DefaultTraceContextAware - - class DefaultTraceContextAware extends TraceContextAware { - @transient val captureNanoTime = System.nanoTime() - @transient val traceContext = TraceRecorder.currentContext - - // - // Beware of this hack, it might bite us in the future! - // - // When using remoting/cluster all messages carry the TraceContext in the envelope in which they - // are sent but that doesn't apply to System Messages. We are certain that the TraceContext is - // available (if any) when the system messages are read and this will make sure that it is correctly - // captured and propagated. - @throws[ObjectStreamException] - private def readResolve: AnyRef = { - new DefaultTraceContextAware - } +case object EmptyTraceContext extends TraceContext { + def name: String = "empty-trace" + def token: String = "" + def rename(name: String): Unit = {} + def finish(): Unit = {} + def origin: TraceContextOrigin = TraceContextOrigin.Local + def isOpen: Boolean = false + def isEmpty: Boolean = true + def startSegment(segmentName: String, label: String): Segment = EmptySegment + def nanoTimestamp: Long = 0L + + case object EmptySegment extends Segment { + val name: String = "empty-segment" + val label: String = "empty-label" + def isEmpty: Boolean = true + def rename(newName: String): Unit = {} + def finish: Unit = {} } } -trait SegmentCompletionHandleAware extends TraceContextAware { - @volatile var segmentCompletionHandle: Option[SegmentCompletionHandle] = None -} - -object SegmentCompletionHandleAware { - def default: SegmentCompletionHandleAware = new DefaultSegmentCompletionHandleAware - - class DefaultSegmentCompletionHandleAware extends DefaultTraceContextAware with SegmentCompletionHandleAware {} -} - -class SimpleMetricCollectionContext(traceName: String, val token: String, metadata: Map[String, String], - val origin: TraceContextOrigin, val system: ActorSystem, val startMilliTime: Long = System.currentTimeMillis, - izOpen: Boolean = true) extends TraceContext { +class DefaultTraceContext(traceName: String, val token: String, izOpen: Boolean, val levelOfDetail: LevelOfDetail, + val origin: TraceContextOrigin, nanoTimeztamp: Long, val system: ActorSystem) extends TraceContext { + val isEmpty: Boolean = false @volatile private var _name = traceName @volatile private var _isOpen = izOpen - val levelOfDetail = OnlyMetrics - val startNanoTime = System.nanoTime() - val finishedSegments = new ConcurrentLinkedQueue[SegmentData]() - val metricsExtension = Kamon(Metrics)(system) + private val _nanoTimestamp = nanoTimeztamp + private val finishedSegments = new ConcurrentLinkedQueue[SegmentData]() + private val metricsExtension = Kamon(Metrics)(system) + private[kamon] val traceLocalStorage: TraceLocalStorage = new TraceLocalStorage def name: String = _name + def rename(newName: String): Unit = + if (isOpen) _name = newName // TODO: log a warning about renaming a closed trace. - def rename(newName: String): Unit = _name = newName - - def isOpen(): Boolean = _isOpen + def isOpen: Boolean = _isOpen + def nanoTimestamp: Long = _nanoTimestamp - def finish(metadata: Map[String, String]): Unit = { + def finish(): Unit = { _isOpen = false - - val elapsedNanoTime = - if (origin == TraceContextOrigin.Local) - // Everything is local, nanoTime is still the best resolution we can use. - System.nanoTime() - startNanoTime - else - // For a remote TraceContext we can only rely on the startMilliTime and we need to scale it to nanoseconds - // to be consistent with unit used for all latency measurements. - (System.currentTimeMillis() - startMilliTime) * 1000000L - + val elapsedNanoTime = System.nanoTime() - _nanoTimestamp val metricRecorder = metricsExtension.register(TraceMetrics(name), TraceMetrics.Factory) metricRecorder.map { traceMetrics ⇒ @@ -138,6 +98,8 @@ class SimpleMetricCollectionContext(traceName: String, val token: String, metada } } + def startSegment(segmentName: String, segmentLabel: String): Segment = new DefaultSegment(segmentName, segmentLabel) + @tailrec private def drainFinishedSegments(metricRecorder: TraceMetricRecorder): Unit = { val segment = finishedSegments.poll() if (segment != null) { @@ -146,8 +108,8 @@ class SimpleMetricCollectionContext(traceName: String, val token: String, metada } } - private def finishSegment(identity: MetricIdentity, duration: Long, metadata: Map[String, String]): Unit = { - finishedSegments.add(SegmentData(identity, duration, metadata)) + private def finishSegment(segmentName: String, label: String, duration: Long): Unit = { + finishedSegments.add(SegmentData(SegmentMetricIdentity(segmentName, label), duration)) if (!_isOpen) { metricsExtension.register(TraceMetrics(name), TraceMetrics.Factory).map { traceMetrics ⇒ @@ -156,16 +118,81 @@ class SimpleMetricCollectionContext(traceName: String, val token: String, metada } } - def startSegment(identity: SegmentIdentity, metadata: Map[String, String]): SegmentCompletionHandle = - new SimpleMetricCollectionCompletionHandle(identity, metadata) + class DefaultSegment(segmentName: String, val label: String) extends Segment { + private val _segmentStartNanoTime = System.nanoTime() + @volatile private var _segmentName = segmentName + @volatile private var _isOpen = true - class SimpleMetricCollectionCompletionHandle(identity: MetricIdentity, startMetadata: Map[String, String]) extends SegmentCompletionHandle { - val segmentStartNanoTime = System.nanoTime() + def name: String = _segmentName + def rename(newName: String): Unit = _segmentName = newName + def isEmpty: Boolean = false - def finish(metadata: Map[String, String] = Map.empty): Unit = { + def finish: Unit = { val segmentFinishNanoTime = System.nanoTime() - finishSegment(identity, (segmentFinishNanoTime - segmentStartNanoTime), startMetadata ++ metadata) + finishSegment(name, label, (segmentFinishNanoTime - _segmentStartNanoTime)) + } + } +} + +case class SegmentMetricIdentity(name: String, label: String) extends MetricIdentity +case class SegmentData(identity: SegmentMetricIdentity, duration: Long) + +object SegmentMetricIdentityLabel { + val HttpClient = "http-client" +} + +sealed trait LevelOfDetail +object LevelOfDetail { + case object OnlyMetrics extends LevelOfDetail + case object SimpleTrace extends LevelOfDetail + case object FullTrace extends LevelOfDetail +} + +sealed trait TraceContextOrigin +object TraceContextOrigin { + case object Local extends TraceContextOrigin + case object Remote extends TraceContextOrigin +} + +trait TraceContextAware extends Serializable { + def traceContext: TraceContext +} + +object TraceContextAware { + def default: TraceContextAware = new DefaultTraceContextAware + + class DefaultTraceContextAware extends TraceContextAware { + @transient val traceContext = TraceRecorder.currentContext + + // + // Beware of this hack, it might bite us in the future! + // + // When using remoting/cluster all messages carry the TraceContext in the envelope in which they + // are sent but that doesn't apply to System Messages. We are certain that the TraceContext is + // available (if any) when the system messages are read and this will make sure that it is correctly + // captured and propagated. + @throws[ObjectStreamException] + private def readResolve: AnyRef = { + new DefaultTraceContextAware } } } +trait TimestampedTraceContextAware extends TraceContextAware { + def captureNanoTime: Long +} + +object TimestampedTraceContextAware { + def default: TimestampedTraceContextAware = new DefaultTraceContextAware with TimestampedTraceContextAware { + @transient val captureNanoTime = System.nanoTime() + } +} + +trait SegmentAware { + @volatile var segment: Segment = EmptyTraceContext.EmptySegment +} + +object SegmentAware { + def default: SegmentAware = new DefaultSegmentAware + class DefaultSegmentAware extends DefaultTraceContextAware with SegmentAware {} +}
\ No newline at end of file diff --git a/kamon-core/src/main/scala/kamon/trace/TraceLocal.scala b/kamon-core/src/main/scala/kamon/trace/TraceLocal.scala index 3ff074b6..0766af74 100644 --- a/kamon-core/src/main/scala/kamon/trace/TraceLocal.scala +++ b/kamon-core/src/main/scala/kamon/trace/TraceLocal.scala @@ -24,18 +24,20 @@ object TraceLocal { type ValueType } - def store(key: TraceLocalKey)(value: key.ValueType): Unit = - TraceRecorder.currentContext.map(_.traceLocalStorage.store(key)(value)) - - def retrieve(key: TraceLocalKey): Option[key.ValueType] = - TraceRecorder.currentContext.flatMap(_.traceLocalStorage.retrieve(key)) + def store(key: TraceLocalKey)(value: key.ValueType): Unit = TraceRecorder.currentContext match { + case ctx: DefaultTraceContext ⇒ ctx.traceLocalStorage.store(key)(value) + case EmptyTraceContext ⇒ // Can't store in the empty context. + } + def retrieve(key: TraceLocalKey): Option[key.ValueType] = TraceRecorder.currentContext match { + case ctx: DefaultTraceContext ⇒ ctx.traceLocalStorage.retrieve(key) + case EmptyTraceContext ⇒ None // Can't retrieve anything from the empty context. + } } class TraceLocalStorage { val underlyingStorage = TrieMap[TraceLocal.TraceLocalKey, Any]() def store(key: TraceLocalKey)(value: key.ValueType): Unit = underlyingStorage.put(key, value) - def retrieve(key: TraceLocalKey): Option[key.ValueType] = underlyingStorage.get(key).map(_.asInstanceOf[key.ValueType]) } diff --git a/kamon-core/src/main/scala/kamon/trace/TraceRecorder.scala b/kamon-core/src/main/scala/kamon/trace/TraceRecorder.scala index 778edc42..8da187cb 100644 --- a/kamon-core/src/main/scala/kamon/trace/TraceRecorder.scala +++ b/kamon-core/src/main/scala/kamon/trace/TraceRecorder.scala @@ -23,65 +23,70 @@ import kamon.macros.InlineTraceContextMacro import scala.util.Try import java.net.InetAddress import akka.actor.ActorSystem -import kamon.trace.TraceContext.SegmentIdentity object TraceRecorder { - private val traceContextStorage = new ThreadLocal[Option[TraceContext]] { - override def initialValue(): Option[TraceContext] = None + private val traceContextStorage = new ThreadLocal[TraceContext] { + override def initialValue(): TraceContext = EmptyTraceContext } private val tokenCounter = new AtomicLong private val hostnamePrefix = Try(InetAddress.getLocalHost.getHostName).getOrElse("unknown-localhost") - def newToken = "%s-%s".format(hostnamePrefix, tokenCounter.incrementAndGet()) - - private def newTraceContext(name: String, token: Option[String], metadata: Map[String, String], - system: ActorSystem): TraceContext = { - - // In the future this should select between implementations. - val finalToken = token.getOrElse(newToken) - new SimpleMetricCollectionContext(name, finalToken, metadata, TraceContextOrigin.Local, system) + def newToken: String = hostnamePrefix + "-" + String.valueOf(tokenCounter.incrementAndGet()) + + private def newTraceContext(name: String, token: Option[String], system: ActorSystem): TraceContext = { + new DefaultTraceContext( + name, + token.getOrElse(newToken), + izOpen = true, + LevelOfDetail.OnlyMetrics, + TraceContextOrigin.Local, + nanoTimeztamp = System.nanoTime, + system) } def joinRemoteTraceContext(traceName: String, traceToken: String, startMilliTime: Long, isOpen: Boolean, system: ActorSystem): TraceContext = { - new SimpleMetricCollectionContext( + val equivalentNanotime = System.nanoTime() - ((System.currentTimeMillis() - startMilliTime) * 1000000) + new DefaultTraceContext( traceName, traceToken, - Map.empty, + isOpen, + LevelOfDetail.OnlyMetrics, TraceContextOrigin.Remote, - system, - startMilliTime, - isOpen) + equivalentNanotime, + system) } - def setContext(context: Option[TraceContext]): Unit = traceContextStorage.set(context) + def setContext(context: TraceContext): Unit = traceContextStorage.set(context) - def clearContext: Unit = traceContextStorage.set(None) + def clearContext: Unit = traceContextStorage.set(EmptyTraceContext) - def currentContext: Option[TraceContext] = traceContextStorage.get() + def currentContext: TraceContext = traceContextStorage.get() - def start(name: String, token: Option[String] = None, metadata: Map[String, String] = Map.empty)(implicit system: ActorSystem) = { - val ctx = newTraceContext(name, token, metadata, system) - traceContextStorage.set(Some(ctx)) + def start(name: String, token: Option[String] = None)(implicit system: ActorSystem) = { + val ctx = newTraceContext(name, token, system) + traceContextStorage.set(ctx) } - def startSegment(identity: SegmentIdentity, metadata: Map[String, String] = Map.empty): Option[SegmentCompletionHandle] = - currentContext.map(_.startSegment(identity, metadata)) - - def rename(name: String): Unit = currentContext.map(_.rename(name)) + def rename(name: String): Unit = currentContext.rename(name) - def withNewTraceContext[T](name: String, token: Option[String] = None, metadata: Map[String, String] = Map.empty)(thunk: ⇒ T)(implicit system: ActorSystem): T = - withTraceContext(Some(newTraceContext(name, token, metadata, system)))(thunk) + def withNewTraceContext[T](name: String, token: Option[String] = None)(thunk: ⇒ T)(implicit system: ActorSystem): T = + withTraceContext(newTraceContext(name, token, system))(thunk) - def withTraceContext[T](context: Option[TraceContext])(thunk: ⇒ T): T = { + def withTraceContext[T](context: TraceContext)(thunk: ⇒ T): T = { val oldContext = currentContext setContext(context) try thunk finally setContext(oldContext) } - def withInlineTraceContextReplacement[T](traceCtx: Option[TraceContext])(thunk: ⇒ T): T = macro InlineTraceContextMacro.withInlineTraceContextImpl[T, Option[TraceContext]] + def withTraceContextAndSystem[T](thunk: (TraceContext, ActorSystem) ⇒ T): Option[T] = currentContext match { + case ctx: DefaultTraceContext ⇒ Some(thunk(ctx, ctx.system)) + case EmptyTraceContext ⇒ None + } + + def withInlineTraceContextReplacement[T](traceCtx: TraceContext)(thunk: ⇒ T): T = macro InlineTraceContextMacro.withInlineTraceContextImpl[T, TraceContext] - def finish(metadata: Map[String, String] = Map.empty): Unit = currentContext.map(_.finish(metadata)) + def finish(): Unit = currentContext.finish() } diff --git a/kamon-core/src/main/scala/kamon/trace/logging/LogbackTraceTokenConverter.scala b/kamon-core/src/main/scala/kamon/trace/logging/LogbackTraceTokenConverter.scala index 4b7dbb28..f052f009 100644 --- a/kamon-core/src/main/scala/kamon/trace/logging/LogbackTraceTokenConverter.scala +++ b/kamon-core/src/main/scala/kamon/trace/logging/LogbackTraceTokenConverter.scala @@ -20,5 +20,11 @@ import ch.qos.logback.classic.spi.ILoggingEvent import kamon.trace.TraceRecorder class LogbackTraceTokenConverter extends ClassicConverter { - def convert(event: ILoggingEvent): String = TraceRecorder.currentContext.map(_.token).getOrElse("undefined") + def convert(event: ILoggingEvent): String = { + val ctx = TraceRecorder.currentContext + if (ctx.isEmpty) + "undefined" + else + ctx.token + } } |