diff options
Diffstat (limited to 'kamon-core/src/main/scala/kamon')
8 files changed, 189 insertions, 316 deletions
diff --git a/kamon-core/src/main/scala/kamon/instrumentation/ActorLoggingTracing.scala b/kamon-core/src/main/scala/kamon/instrumentation/ActorLoggingTracing.scala index abd3514e..954f351a 100644 --- a/kamon-core/src/main/scala/kamon/instrumentation/ActorLoggingTracing.scala +++ b/kamon-core/src/main/scala/kamon/instrumentation/ActorLoggingTracing.scala @@ -23,14 +23,14 @@ import kamon.trace.{ TraceContextAware, TraceRecorder } class ActorLoggingTracing { @DeclareMixin("akka.event.Logging.LogEvent+") - def mixin: TraceContextAware = new TraceContextAware {} + def mixinTraceContextAwareToLogEvent: TraceContextAware = TraceContextAware.default @Pointcut("execution(* akka.event.slf4j.Slf4jLogger.withMdc(..)) && args(logSource, logEvent, logStatement)") def withMdcInvocation(logSource: String, logEvent: TraceContextAware, logStatement: () ⇒ _): Unit = {} @Around("withMdcInvocation(logSource, logEvent, logStatement)") def aroundWithMdcInvocation(pjp: ProceedingJoinPoint, logSource: String, logEvent: TraceContextAware, logStatement: () ⇒ _): Unit = { - TraceRecorder.withContext(logEvent.traceContext) { + TraceRecorder.withTraceContext(logEvent.traceContext) { pjp.proceed() } } diff --git a/kamon-core/src/main/scala/kamon/instrumentation/FutureTracing.scala b/kamon-core/src/main/scala/kamon/instrumentation/FutureTracing.scala index b8725dd7..634c94a1 100644 --- a/kamon-core/src/main/scala/kamon/instrumentation/FutureTracing.scala +++ b/kamon-core/src/main/scala/kamon/instrumentation/FutureTracing.scala @@ -23,7 +23,7 @@ import kamon.trace.{ TraceContextAware, TraceRecorder } class FutureTracing { @DeclareMixin("scala.concurrent.impl.CallbackRunnable || scala.concurrent.impl.Future.PromiseCompletingRunnable") - def mixin: TraceContextAware = new TraceContextAware {} + def mixinTraceContextAwareToFutureRelatedRunnable: TraceContextAware = TraceContextAware.default @Pointcut("execution((scala.concurrent.impl.CallbackRunnable || scala.concurrent.impl.Future.PromiseCompletingRunnable).new(..)) && this(runnable)") def futureRelatedRunnableCreation(runnable: TraceContextAware): Unit = {} @@ -39,7 +39,7 @@ class FutureTracing { @Around("futureRelatedRunnableExecution(runnable)") def aroundExecution(pjp: ProceedingJoinPoint, runnable: TraceContextAware): Any = { - TraceRecorder.withContext(runnable.traceContext) { + TraceRecorder.withTraceContext(runnable.traceContext) { pjp.proceed() } } diff --git a/kamon-core/src/main/scala/kamon/metrics/TraceMetrics.scala b/kamon-core/src/main/scala/kamon/metrics/TraceMetrics.scala index 25ebce00..57a79653 100644 --- a/kamon-core/src/main/scala/kamon/metrics/TraceMetrics.scala +++ b/kamon-core/src/main/scala/kamon/metrics/TraceMetrics.scala @@ -57,9 +57,7 @@ object TraceMetrics extends MetricGroupIdentity.Category with MetricGroupFactory val elapsedTimeHdrConfig = Configuration.fromConfig(settings.getConfig("elapsed-time")) val segmentHdrConfig = Configuration.fromConfig(settings.getConfig("segment")) - new TraceMetricRecorder( - HighDynamicRangeRecorder(elapsedTimeHdrConfig), - () ⇒ HighDynamicRangeRecorder(segmentHdrConfig)) + new TraceMetricRecorder(HighDynamicRangeRecorder(elapsedTimeHdrConfig), () ⇒ HighDynamicRangeRecorder(segmentHdrConfig)) } } diff --git a/kamon-core/src/main/scala/kamon/trace/Segments.scala b/kamon-core/src/main/scala/kamon/trace/Segments.scala deleted file mode 100644 index e6d9745b..00000000 --- a/kamon-core/src/main/scala/kamon/trace/Segments.scala +++ /dev/null @@ -1,39 +0,0 @@ -/* =================================================== - * Copyright © 2013 the kamon project <http://kamon.io/> - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * ========================================================== */ - -package kamon.trace - - -import kamon.trace.TraceOld.SegmentCompletionHandle - -object Segments { - - trait Category - case object HttpClientRequest extends Category - - case class Start(category: Category, description: String = "", - attributes: Map[String, String] = Map(), timestamp: Long = System.nanoTime()) - - case class End(attributes: Map[String, String] = Map(), timestamp: Long = System.nanoTime()) - - case class Segment(start: Start, end: End) - - trait SegmentCompletionHandleAware { - var completionHandle: Option[SegmentCompletionHandle] - } - - trait ContextAndSegmentCompletionAware extends TraceContextAware with SegmentCompletionHandleAware -} diff --git a/kamon-core/src/main/scala/kamon/trace/Trace.scala b/kamon-core/src/main/scala/kamon/trace/Trace.scala deleted file mode 100644 index bdfd6aa3..00000000 --- a/kamon-core/src/main/scala/kamon/trace/Trace.scala +++ /dev/null @@ -1,123 +0,0 @@ -/* =================================================== - * Copyright © 2013 the kamon project <http://kamon.io/> - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * ========================================================== */ -package kamon.trace - -import kamon.Kamon -import akka.actor._ -import scala.Some -import kamon.trace.TraceOld.Register -import scala.concurrent.duration._ -import java.util.concurrent.atomic.AtomicLong -import scala.util.Try -import java.net.InetAddress - -object TraceOld extends ExtensionId[TraceExtension] with ExtensionIdProvider { - def lookup(): ExtensionId[_ <: Extension] = TraceOld - def createExtension(system: ExtendedActorSystem): TraceExtension = new TraceExtension(system) - - /*** Protocol */ - case object Register - - /** User API */ - //private[trace] val traceContext = new DynamicVariable[Option[TraceContext]](None) - private[trace] val traceContext = new ThreadLocal[Option[TraceContextOld]] { - override def initialValue(): Option[TraceContextOld] = None - } - private[trace] val tranid = new AtomicLong() - - def context() = traceContext.get - private def set(ctx: Option[TraceContextOld]) = traceContext.set(ctx) - - def clear: Unit = traceContext.remove() - def start(name: String, token: Option[String])(implicit system: ActorSystem): TraceContextOld = { - val ctx = newTraceContext(name, token.getOrElse(TraceToken.generate())) - ctx.start(name) - set(Some(ctx)) - - ctx - } - - def withContext[T](ctx: Option[TraceContextOld])(thunk: ⇒ T): T = { - val oldval = context - set(ctx) - - try thunk - finally set(oldval) - } - - def transformContext(f: TraceContextOld ⇒ TraceContextOld): Unit = { - context.map(f).foreach(ctx ⇒ set(Some(ctx))) - } - - def finish(): Option[TraceContextOld] = { - val ctx = context() - ctx.map(_.finish) - clear - ctx - } - - // TODO: FIX - def newTraceContext(name: String, token: String)(implicit system: ActorSystem): TraceContextOld = TraceContextOld(Kamon(TraceOld).api, tranid.getAndIncrement, name, token) - - def startSegment(category: Segments.Category, description: String = "", attributes: Map[String, String] = Map()): SegmentCompletionHandle = { - val start = Segments.Start(category, description, attributes) - SegmentCompletionHandle(start) - } - - def startSegment(start: Segments.Start): SegmentCompletionHandle = SegmentCompletionHandle(start) - - case class SegmentCompletionHandle(start: Segments.Start) { - def complete(): Unit = { - val end = Segments.End() - //println(s"Completing the Segment: $start - $end") - } - def complete(end: Segments.End): Unit = { - //println(s"Completing the Segment: $start - $end") - } - } -} - -object TraceToken { - val tokenCounter = new AtomicLong - val hostnamePrefix = Try(InetAddress.getLocalHost.getHostName).getOrElse("unknown-localhost") - - def generate(): String = "%s-%s".format(hostnamePrefix, tokenCounter.incrementAndGet()) -} - -class TraceExtension(system: ExtendedActorSystem) extends Kamon.Extension { - val api: ActorRef = system.actorOf(Props[TraceManager], "kamon-trace") -} - -class TraceManager extends Actor with ActorLogging { - var listeners: Seq[ActorRef] = Seq.empty - - def receive = { - case Register ⇒ - listeners = sender +: listeners - log.info("Registered [{}] as listener for Kamon traces", sender) - - case segment: UowSegment ⇒ - val tracerName = segment.id.toString - context.child(tracerName).getOrElse(newTracer(tracerName)) ! segment - - case trace: UowTrace ⇒ - listeners foreach (_ ! trace) - } - - def newTracer(name: String): ActorRef = { - context.actorOf(UowTraceAggregator.props(self, 30 seconds), name) - } -} diff --git a/kamon-core/src/main/scala/kamon/trace/TraceContext.scala b/kamon-core/src/main/scala/kamon/trace/TraceContext.scala index 95a3a8b2..d3759a26 100644 --- a/kamon-core/src/main/scala/kamon/trace/TraceContext.scala +++ b/kamon-core/src/main/scala/kamon/trace/TraceContext.scala @@ -1,36 +1,129 @@ -/* =================================================== +/* + * ========================================================================================= * Copyright © 2013 the kamon project <http://kamon.io/> * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file + * except in compliance with the License. You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * ========================================================== */ + * Unless required by applicable law or agreed to in writing, software distributed under the + * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, + * either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + * ========================================================================================= + */ + package kamon.trace -import java.util.UUID -import akka.actor._ -import java.util.concurrent.atomic.AtomicLong -import scala.concurrent.duration._ +import akka.actor.ActorSystem import kamon.Kamon -import kamon.trace.UowTracing.{ Finish, Start } +import kamon.metrics.{ MetricGroupRecorder, MetricIdentity, TraceMetrics, Metrics } +import java.util.concurrent.ConcurrentLinkedQueue +import kamon.trace.TraceContextAware.DefaultTraceContextAware +import kamon.trace.TraceContext.SegmentIdentity + +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]) +} + +object TraceContext { + type SegmentIdentity = MetricIdentity +} + +trait SegmentCompletionHandle { + def finish(metadata: Map[String, String]) +} + +case class SegmentData(identity: MetricIdentity, duration: Long, metadata: Map[String, String]) -// TODO: Decide if we need or not an ID, generating it takes time and it doesn't seem necessary. -case class TraceContextOld(private val collector: ActorRef, id: Long, name: String, token: String, userContext: Option[Any] = None) { +sealed trait TracingLevelOfDetail +case object OnlyMetrics extends TracingLevelOfDetail +case object SimpleTrace extends TracingLevelOfDetail +case object FullTrace extends TracingLevelOfDetail - def start(name: String) = { - collector ! Start(id, name) +trait TraceContextAware { + def captureMark: Long + def traceContext: Option[TraceContext] +} + +object TraceContextAware { + def default: TraceContextAware = new DefaultTraceContextAware + + class DefaultTraceContextAware extends TraceContextAware { + val captureMark = System.nanoTime() + val traceContext = TraceRecorder.currentContext } +} - def finish: Unit = { - collector ! Finish(id) +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(@volatile private var _name: String, val token: String, metadata: Map[String, String], + val system: ActorSystem) extends TraceContext { + @volatile private var _isOpen = true + val levelOfDetail = OnlyMetrics + val startMark = System.nanoTime() + val finishedSegments = new ConcurrentLinkedQueue[SegmentData]() + val metricsExtension = Kamon(Metrics)(system) + + def name: String = _name + + def rename(newName: String): Unit = _name = newName + + def isOpen(): Boolean = _isOpen + + def finish(metadata: Map[String, String]): Unit = { + _isOpen = false + val finishMark = System.nanoTime() + val metricRecorder = metricsExtension.register(name, TraceMetrics) + + metricRecorder.map { traceMetrics ⇒ + traceMetrics.elapsedTime.record(finishMark - startMark) + drainFinishedSegments(traceMetrics) + } + } + + private def drainFinishedSegments(metricRecorder: MetricGroupRecorder): Unit = { + while (!finishedSegments.isEmpty) { + val segmentData = finishedSegments.poll() + metricRecorder.record(segmentData.identity, segmentData.duration) + } } + private def finishSegment(identity: MetricIdentity, duration: Long, metadata: Map[String, String]): Unit = { + finishedSegments.add(SegmentData(identity, duration, metadata)) + + if (!_isOpen) { + metricsExtension.register(name, TraceMetrics).map { traceMetrics ⇒ + drainFinishedSegments(traceMetrics) + } + } + } + + def startSegment(identity: SegmentIdentity, metadata: Map[String, String]): SegmentCompletionHandle = + new SimpleMetricCollectionCompletionHandle(identity, metadata) + + class SimpleMetricCollectionCompletionHandle(identity: MetricIdentity, startMetadata: Map[String, String]) extends SegmentCompletionHandle { + val segmentStartMark = System.nanoTime() + + def finish(metadata: Map[String, String] = Map.empty): Unit = { + val segmentFinishMark = System.nanoTime() + finishSegment(identity, (segmentFinishMark - segmentStartMark), startMetadata ++ metadata) + } + } } + diff --git a/kamon-core/src/main/scala/kamon/trace/TraceCtx.scala b/kamon-core/src/main/scala/kamon/trace/TraceCtx.scala deleted file mode 100644 index 1e552563..00000000 --- a/kamon-core/src/main/scala/kamon/trace/TraceCtx.scala +++ /dev/null @@ -1,125 +0,0 @@ -/* - * ========================================================================================= - * Copyright © 2013 the kamon project <http://kamon.io/> - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file - * except in compliance with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the - * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, - * either express or implied. See the License for the specific language governing permissions - * and limitations under the License. - * ========================================================================================= - */ - -package kamon.trace - -import akka.actor.{ExtendedActorSystem, ActorSystem} -import akka.dispatch.AbstractNodeQueue -import kamon.Kamon -import kamon.metrics.{TraceMetrics, Metrics} -import java.util.concurrent.atomic.AtomicLong - -sealed trait TracingLevelOfDetail -case object OnlyMetrics extends TracingLevelOfDetail -case object SimpleTrace extends TracingLevelOfDetail -case object FullTrace extends TracingLevelOfDetail - -trait TraceContext { - def token: String - def name: String - def rename(newName: String): Unit - def levelOfDetail: TracingLevelOfDetail - def startSegment - def finish(metadata: Map[String, String]) - -} - -trait TraceContextAware { - def captureMark: Long - def traceContext: Option[TraceContext] -} - -object TraceContextAware { - def default: TraceContextAware = new TraceContextAware { - val captureMark = System.nanoTime() - val traceContext = TraceRecorder.currentContext - } -} - -object TraceContext { - -} - -class SimpleMetricCollectionContext(val token: String, @volatile private var _name: String, val system: ActorSystem, - metadata: Map[String, String]) extends TraceContext { - val levelOfDetail = OnlyMetrics - - @volatile private var _isOpen = true - - val startMark = System.nanoTime() - - def name: String = _name - - def rename(newName: String): Unit = _name = newName - - def isOpen(): Boolean = _isOpen - - def finish(metadata: Map[String, String]): Unit = { - val finishMark = System.nanoTime() - - // Store all metrics! - val metricsExtension = Kamon(Metrics)(system) - val metricRecorder = metricsExtension.register(name, TraceMetrics) - - metricRecorder.map { traceMetrics => - traceMetrics.elapsedTime.record(finishMark - startMark) - } - - } - - override def startSegment: Unit = ??? - - -} - -private[kamon] class SegmentRecordingQueue extends AbstractNodeQueue[String] - - - - -class TraceRecorder(system: ExtendedActorSystem) { - -} - -object TraceRecorder { - private val tokenCounter = new AtomicLong - private val traceContextStorage = new ThreadLocal[Option[TraceContext]] { - override def initialValue(): Option[TraceContext] = None - } - - private def newTraceContext(name: String, token: Option[String], metadata: Map[String, String], system: ActorSystem): TraceContext = ??? - - def setContext(context: Option[TraceContext]): Unit = traceContextStorage.set(context) - - def clearContext: Unit = traceContextStorage.set(None) - - def currentContext: Option[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 withContext[T](context: Option[TraceContext])(thunk: => T): T = { - val oldContext = currentContext - setContext(context) - - try thunk finally setContext(oldContext) - } - - def finish(metadata: Map[String, String] = Map.empty): Unit = currentContext.map(_.finish(metadata)) - -} diff --git a/kamon-core/src/main/scala/kamon/trace/TraceRecorder.scala b/kamon-core/src/main/scala/kamon/trace/TraceRecorder.scala new file mode 100644 index 00000000..3e3bb19f --- /dev/null +++ b/kamon-core/src/main/scala/kamon/trace/TraceRecorder.scala @@ -0,0 +1,69 @@ +/* + * ========================================================================================= + * Copyright © 2013 the kamon project <http://kamon.io/> + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file + * except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the + * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, + * either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + * ========================================================================================= + */ + +package kamon.trace + +import java.util.concurrent.atomic.AtomicLong +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 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, system) + } + + def setContext(context: Option[TraceContext]): Unit = traceContextStorage.set(context) + + def clearContext: Unit = traceContextStorage.set(None) + + def currentContext: Option[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 startSegment(identity: SegmentIdentity, metadata: Map[String, String]): Option[SegmentCompletionHandle] = + currentContext.map(_.startSegment(identity, metadata)) + + 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 withTraceContext[T](context: Option[TraceContext])(thunk: ⇒ T): T = { + val oldContext = currentContext + setContext(context) + + try thunk finally setContext(oldContext) + } + + def finish(metadata: Map[String, String] = Map.empty): Unit = currentContext.map(_.finish(metadata)) + +} |