package kamon.trace
import java.util.concurrent.ConcurrentLinkedQueue
import akka.actor.ActorSystem
import akka.event.LoggingAdapter
import kamon.{ RelativeNanoTimestamp, NanoInterval }
import kamon.metric.TraceMetrics.TraceMetricRecorder
import kamon.metric.{ MetricsExtension, TraceMetrics }
import scala.annotation.tailrec
private[trace] class MetricsOnlyContext(traceName: String, val token: String, izOpen: Boolean, val levelOfDetail: LevelOfDetail, val origin: TraceContextOrigin,
val startRelativeTimestamp: RelativeNanoTimestamp, log: LoggingAdapter, metricsExtension: MetricsExtension, val system: ActorSystem)
extends TraceContext {
@volatile private var _name = traceName
@volatile private var _isOpen = izOpen
@volatile protected var _elapsedTime = NanoInterval.default
private val _finishedSegments = new ConcurrentLinkedQueue[SegmentLatencyData]()
private val _traceLocalStorage = new TraceLocalStorage
def rename(newName: String): Unit =
if (isOpen)
_name = newName
else if (log.isWarningEnabled)
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 addMetadata(key: String, value: String): Unit = {}
def finish(): Unit = {
_isOpen = false
val traceElapsedTime = NanoInterval.since(startRelativeTimestamp)
_elapsedTime = traceElapsedTime
val metricRecorder = metricsExtension.register(TraceMetrics(name), TraceMetrics.Factory)
metricRecorder.map { traceMetrics ⇒
traceMetrics.elapsedTime.record(traceElapsedTime.nanos)
drainFinishedSegments(traceMetrics)
}
}
def startSegment(segmentName: String, category: String, library: String): Segment =
new MetricsOnlySegment(segmentName, category, library)
@tailrec private def drainFinishedSegments(metricRecorder: TraceMetricRecorder): Unit = {
val segment = _finishedSegments.poll()
if (segment != null) {
metricRecorder.segmentRecorder(segment.identity).record(segment.duration.nanos)
drainFinishedSegments(metricRecorder)
}
}
protected def finishSegment(segmentName: String, category: String, library: String, duration: NanoInterval): Unit = {
_finishedSegments.add(SegmentLatencyData(SegmentMetricIdentity(segmentName, category, library), duration))
if (isClosed) {
metricsExtension.register(TraceMetrics(name), TraceMetrics.Factory).map { traceMetrics ⇒
drainFinishedSegments(traceMetrics)
}
}
}
// Should only be used by the TraceLocal utilities.
def traceLocalStorage: TraceLocalStorage = _traceLocalStorage
// Handle with care and make sure that the trace is closed before calling this method, otherwise NanoInterval.default
// will be returned.
def elapsedTime: NanoInterval = _elapsedTime
class MetricsOnlySegment(segmentName: String, val category: String, val library: String) extends Segment {
private val _startTimestamp = RelativeNanoTimestamp.now
@volatile private var _segmentName = segmentName
@volatile private var _elapsedTime = NanoInterval.default
@volatile private var _isOpen = true
def name: String = _segmentName
def isEmpty: Boolean = false
def addMetadata(key: String, value: String): Unit = {}
def isOpen: Boolean = _isOpen
def rename(newName: String): Unit =
if (isOpen)
_segmentName = newName
else if (log.isWarningEnabled)
log.warning("Can't rename segment from [{}] to [{}] because the segment is already closed.", name, newName)
def finish: Unit = {
_isOpen = false
val segmentElapsedTime = NanoInterval.since(_startTimestamp)
_elapsedTime = segmentElapsedTime
finishSegment(name, category, library, segmentElapsedTime)
}
// 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
}
}