diff options
Diffstat (limited to 'kamon-core/src/main/scala/kamon/trace')
4 files changed, 76 insertions, 50 deletions
diff --git a/kamon-core/src/main/scala/kamon/trace/MetricsOnlyContext.scala b/kamon-core/src/main/scala/kamon/trace/MetricsOnlyContext.scala index 34b68f38..973eab5a 100644 --- a/kamon-core/src/main/scala/kamon/trace/MetricsOnlyContext.scala +++ b/kamon-core/src/main/scala/kamon/trace/MetricsOnlyContext.scala @@ -1,6 +1,6 @@ /* * ========================================================================================= - * Copyright © 2013-2014 the kamon project <http://kamon.io/> + * Copyright © 2013-2016 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 @@ -21,17 +21,18 @@ import java.util.concurrent.ConcurrentLinkedQueue import akka.event.LoggingAdapter import kamon.Kamon import kamon.metric.{ SegmentMetrics, TraceMetrics } +import kamon.trace.States.Status import kamon.util.{ NanoInterval, RelativeNanoTimestamp } import scala.annotation.tailrec import scala.collection.concurrent.TrieMap -private[kamon] class MetricsOnlyContext(traceName: String, val token: String, traceTags: Map[String, String], izOpen: Boolean, val levelOfDetail: LevelOfDetail, +private[kamon] class MetricsOnlyContext(traceName: String, val token: String, traceTags: Map[String, String], currentStatus: Status, val levelOfDetail: LevelOfDetail, val startTimestamp: RelativeNanoTimestamp, log: LoggingAdapter) extends TraceContext { @volatile private var _name = traceName - @volatile private var _isOpen = izOpen + @volatile private var _status = currentStatus @volatile protected var _elapsedTime = NanoInterval.default private val _finishedSegments = new ConcurrentLinkedQueue[SegmentLatencyData]() @@ -39,30 +40,38 @@ private[kamon] class MetricsOnlyContext(traceName: String, val token: String, tr private val _tags = TrieMap.empty[String, String] ++= traceTags def rename(newName: String): Unit = - if (isOpen) + if (States.Open == status) _name = newName else log.warning("Can't rename trace from [{}] to [{}] because the trace is already closed.", name, newName) def name: String = _name def isEmpty: Boolean = false - def isOpen: Boolean = _isOpen - + def status: Status = _status def addMetadata(key: String, value: String): Unit = {} - def addTag(key: String, value: String): Unit = _tags.put(key, value) def removeTag(key: String, value: String): Boolean = _tags.remove(key, value) - def finish(): Unit = { - _isOpen = false + private def finish(withError: Boolean): Unit = { + _status = if (withError) States.FinishedWithError else States.FinishedSuccessfully val traceElapsedTime = NanoInterval.since(startTimestamp) _elapsedTime = traceElapsedTime - if (Kamon.metrics.shouldTrack(name, TraceMetrics.category)) - Kamon.metrics.entity(TraceMetrics, name, _tags.toMap).elapsedTime.record(traceElapsedTime.nanos) + if (Kamon.metrics.shouldTrack(name, TraceMetrics.category)) { + val traceEntity = Kamon.metrics.entity(TraceMetrics, name) + traceEntity.elapsedTime.record(traceElapsedTime.nanos) + if (withError) traceEntity.errors.increment() + } drainFinishedSegments() } + def finish(): Unit = finish(withError = false) + + def finishWithError(cause: Throwable): Unit = { + //we should do something with the Throwable in a near future + finish(withError = true) + } + def startSegment(segmentName: String, category: String, library: String): Segment = startSegment(segmentName, category, library, Map.empty[String, String]) @@ -77,14 +86,17 @@ private[kamon] class MetricsOnlyContext(traceName: String, val token: String, tr "category" -> segment.category, "library" -> segment.library) - if (Kamon.metrics.shouldTrack(segment.name, SegmentMetrics.category)) - Kamon.metrics.entity(SegmentMetrics, segment.name, defaultTags ++ segment.tags).elapsedTime.record(segment.duration.nanos) + if (Kamon.metrics.shouldTrack(segment.name, SegmentMetrics.category)) { + val segmentEntity = Kamon.metrics.entity(SegmentMetrics, segment.name, defaultTags ++ segment.tags) + segmentEntity.elapsedTime.record(segment.duration.nanos) + if (segment.isFinishedWithError) segmentEntity.errors.increment() + } drainFinishedSegments() } } - protected def finishSegment(segmentName: String, category: String, library: String, duration: NanoInterval, segmentTags: Map[String, String]): Unit = { - _finishedSegments.add(SegmentLatencyData(segmentName, category, library, duration, segmentTags)) + protected def finishSegment(segmentName: String, category: String, library: String, duration: NanoInterval, segmentTags: Map[String, String], isFinishedWithError: Boolean): Unit = { + _finishedSegments.add(SegmentLatencyData(segmentName, category, library, duration, segmentTags, isFinishedWithError)) if (isClosed) { drainFinishedSegments() @@ -104,36 +116,42 @@ private[kamon] class MetricsOnlyContext(traceName: String, val token: String, tr @volatile private var _segmentName = segmentName @volatile private var _elapsedTime = NanoInterval.default - @volatile private var _isOpen = true + @volatile private var _status: Status = States.Open def name: String = _segmentName def isEmpty: Boolean = false - def isOpen: Boolean = _isOpen - + def status: Status = _status def addMetadata(key: String, value: String): Unit = {} - def addTag(key: String, value: String): Unit = _tags.put(key, value) def removeTag(key: String, value: String): Boolean = _tags.remove(key, value) def rename(newName: String): Unit = - if (isOpen) + if (States.Open == status) _segmentName = newName else log.warning("Can't rename segment from [{}] to [{}] because the segment is already closed.", name, newName) - def finish: Unit = { - _isOpen = false + private def finish(withError: Boolean): Unit = { + _status = if (withError) States.FinishedWithError else States.FinishedSuccessfully val segmentElapsedTime = NanoInterval.since(_startTimestamp) _elapsedTime = segmentElapsedTime - finishSegment(name, category, library, segmentElapsedTime, _tags.toMap) + finishSegment(name, category, library, segmentElapsedTime, _tags.toMap, withError) + } + + def finishWithError(cause: Throwable): Unit = { + //we should do something with the Throwable in a near future + finish(withError = true) } + def finish(): Unit = finish(withError = false) + // Handle with care and make sure that the segment is closed before calling this method, otherwise // NanoInterval.default will be returned. def elapsedTime: NanoInterval = _elapsedTime def startTimestamp: RelativeNanoTimestamp = _startTimestamp + } } -case class SegmentLatencyData(name: String, category: String, library: String, duration: NanoInterval, tags: Map[String, String]) +case class SegmentLatencyData(name: String, category: String, library: String, duration: NanoInterval, tags: Map[String, String], isFinishedWithError: Boolean)
\ No newline at end of file diff --git a/kamon-core/src/main/scala/kamon/trace/TraceContext.scala b/kamon-core/src/main/scala/kamon/trace/TraceContext.scala index eee2ec14..6bf82ac1 100644 --- a/kamon-core/src/main/scala/kamon/trace/TraceContext.scala +++ b/kamon-core/src/main/scala/kamon/trace/TraceContext.scala @@ -19,6 +19,7 @@ package kamon.trace import java.io.ObjectStreamException import java.util +import kamon.trace.States.{ Closed, Status } import kamon.trace.TraceContextAware.DefaultTraceContextAware import kamon.util.{ Function, RelativeNanoTimestamp, SameThreadExecutionContext, Supplier } @@ -29,19 +30,16 @@ trait TraceContext { def token: String def isEmpty: Boolean def nonEmpty: Boolean = !isEmpty - def isOpen: Boolean - def isClosed: Boolean = !isOpen - + def isClosed: Boolean = !(States.Open == status) + def status: Status def finish(): Unit + def finishWithError(cause: Throwable): Unit def rename(newName: String): Unit - def startSegment(segmentName: String, category: String, library: String): Segment def startSegment(segmentName: String, category: String, library: String, tags: Map[String, String]): Segment def addMetadata(key: String, value: String) - def addTag(key: String, value: String): Unit def removeTag(key: String, value: String): Boolean - def startTimestamp: RelativeNanoTimestamp def collect[T](f: TraceContext ⇒ T): Option[T] = @@ -90,13 +88,12 @@ trait Segment { def library: String def isEmpty: Boolean def nonEmpty: Boolean = !isEmpty - def isOpen: Boolean - def isClosed: Boolean = !isOpen - + def isClosed: Boolean = !(States.Open == status) + def status: Status def finish(): Unit + def finishWithError(cause: Throwable): Unit def rename(newName: String): Unit def addMetadata(key: String, value: String): Unit - def addTag(key: String, value: String): Unit def removeTag(key: String, value: String): Boolean } @@ -105,9 +102,9 @@ case object EmptyTraceContext extends TraceContext { def name: String = "empty-trace" def token: String = "" def isEmpty: Boolean = true - def isOpen: Boolean = false - + def status: Status = Closed def finish(): Unit = {} + def finishWithError(cause: Throwable): Unit = {} def rename(name: String): Unit = {} def startSegment(segmentName: String, category: String, library: String): Segment = EmptySegment def startSegment(segmentName: String, category: String, library: String, tags: Map[String, String]): Segment = EmptySegment @@ -123,9 +120,9 @@ case object EmptyTraceContext extends TraceContext { val category: String = "empty-category" val library: String = "empty-library" def isEmpty: Boolean = true - def isOpen: Boolean = false - - def finish: Unit = {} + def status: Status = Closed + def finish(): Unit = {} + def finishWithError(cause: Throwable): Unit = {} def rename(newName: String): Unit = {} def addMetadata(key: String, value: String): Unit = {} def addTag(key: String, value: String): Unit = {} @@ -151,6 +148,14 @@ object LevelOfDetail { case object FullTrace extends LevelOfDetail } +object States { + sealed trait Status + case object Open extends Status + case object Closed extends Status + case object FinishedWithError extends Status + case object FinishedSuccessfully extends Status +} + trait TraceContextAware extends Serializable { def traceContext: TraceContext } diff --git a/kamon-core/src/main/scala/kamon/trace/TracerModule.scala b/kamon-core/src/main/scala/kamon/trace/TracerModule.scala index 60187729..2fc166b6 100644 --- a/kamon-core/src/main/scala/kamon/trace/TracerModule.scala +++ b/kamon-core/src/main/scala/kamon/trace/TracerModule.scala @@ -23,14 +23,16 @@ import akka.event.{ Logging, LoggingAdapter } import com.typesafe.config.Config import kamon.Kamon import kamon.metric.MetricsModule +import kamon.trace.States.Status import kamon.util._ + import scala.collection.JavaConverters._ trait TracerModule { def newContext(name: String): TraceContext def newContext(name: String, token: Option[String]): TraceContext def newContext(name: String, token: Option[String], tags: Map[String, String]): TraceContext - def newContext(name: String, token: Option[String], tags: Map[String, String], timestamp: RelativeNanoTimestamp, isOpen: Boolean, isLocal: Boolean): TraceContext + def newContext(name: String, token: Option[String], tags: Map[String, String], timestamp: RelativeNanoTimestamp, state: Status, isLocal: Boolean): TraceContext def subscribe(subscriber: ActorRef): Unit def unsubscribe(subscriber: ActorRef): Unit @@ -134,14 +136,14 @@ private[kamon] class TracerModuleImpl(metricsExtension: MetricsModule, config: C def newContext(name: String, token: Option[String], tags: Map[String, String]): TraceContext = createTraceContext(name, token, tags) - def newContext(name: String, token: Option[String], tags: Map[String, String], timestamp: RelativeNanoTimestamp, isOpen: Boolean, isLocal: Boolean): TraceContext = - createTraceContext(name, token, tags, timestamp, isOpen, isLocal) + def newContext(name: String, token: Option[String], tags: Map[String, String], timestamp: RelativeNanoTimestamp, status: Status, isLocal: Boolean): TraceContext = + createTraceContext(name, token, tags, timestamp, status, isLocal) private def createTraceContext(traceName: String, token: Option[String], tags: Map[String, String] = Map.empty, startTimestamp: RelativeNanoTimestamp = RelativeNanoTimestamp.now, - isOpen: Boolean = true, isLocal: Boolean = true): TraceContext = { + status: Status = States.Open, isLocal: Boolean = true): TraceContext = { def newMetricsOnlyContext(token: String): TraceContext = - new MetricsOnlyContext(traceName, token, tags, isOpen, _settings.levelOfDetail, startTimestamp, _logger) + new MetricsOnlyContext(traceName, token, tags, status, _settings.levelOfDetail, startTimestamp, _logger) val traceToken = token.getOrElse(newToken) @@ -151,7 +153,7 @@ private[kamon] class TracerModuleImpl(metricsExtension: MetricsModule, config: C case _ if !isLocal || !_settings.sampler.shouldTrace ⇒ newMetricsOnlyContext(traceToken) case _ ⇒ - new TracingContext(traceName, traceToken, tags, izOpen = true, _settings.levelOfDetail, isLocal, startTimestamp, _logger, dispatchTracingContext) + new TracingContext(traceName, traceToken, tags, currentStatus = States.Open, _settings.levelOfDetail, isLocal, startTimestamp, _logger, dispatchTracingContext) } } diff --git a/kamon-core/src/main/scala/kamon/trace/TracingContext.scala b/kamon-core/src/main/scala/kamon/trace/TracingContext.scala index 085b4b09..496d7317 100644 --- a/kamon-core/src/main/scala/kamon/trace/TracingContext.scala +++ b/kamon-core/src/main/scala/kamon/trace/TracingContext.scala @@ -20,13 +20,14 @@ import java.util.concurrent.ConcurrentLinkedQueue import java.util.concurrent.atomic.AtomicInteger import akka.event.LoggingAdapter +import kamon.trace.States.Status import kamon.util.{ NanoInterval, NanoTimestamp, RelativeNanoTimestamp } import scala.collection.concurrent.TrieMap -private[trace] class TracingContext(traceName: String, token: String, tags: Map[String, String], izOpen: Boolean, levelOfDetail: LevelOfDetail, +private[trace] class TracingContext(traceName: String, token: String, tags: Map[String, String], currentStatus: Status, levelOfDetail: LevelOfDetail, isLocal: Boolean, startTimeztamp: RelativeNanoTimestamp, log: LoggingAdapter, traceInfoSink: TracingContext ⇒ Unit) - extends MetricsOnlyContext(traceName, token, tags, izOpen, levelOfDetail, startTimeztamp, log) { + extends MetricsOnlyContext(traceName, token, tags, currentStatus, levelOfDetail, startTimeztamp, log) { private val _openSegments = new AtomicInteger(0) private val _startTimestamp = NanoTimestamp.now @@ -51,12 +52,12 @@ private[trace] class TracingContext(traceName: String, token: String, tags: Map[ traceInfoSink(this) } - override def finishSegment(segmentName: String, category: String, library: String, duration: NanoInterval, tags: Map[String, String]): Unit = { + override def finishSegment(segmentName: String, category: String, library: String, duration: NanoInterval, tags: Map[String, String], isFinishedWithError: Boolean = false): Unit = { _openSegments.decrementAndGet() - super.finishSegment(segmentName, category, library, duration, tags) + super.finishSegment(segmentName, category, library, duration, tags, isFinishedWithError) } - def shouldIncubate: Boolean = isOpen || _openSegments.get() > 0 + def shouldIncubate: Boolean = (States.Open == status) || _openSegments.get() > 0 // Handle with care, should only be used after a trace is finished. def generateTraceInfo: TraceInfo = { |