1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
|
/*
* =========================================================================================
* 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
*
* 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.ConcurrentLinkedQueue
import java.util.concurrent.atomic.AtomicInteger
import akka.event.LoggingAdapter
import kamon.util.{NanoInterval, NanoTimestamp, RelativeNanoTimestamp}
import scala.collection.concurrent.TrieMap
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, currentStatus, levelOfDetail, startTimeztamp, log) {
private val _openSegments = new AtomicInteger(0)
private val _startTimestamp = NanoTimestamp.now
private val _allSegments = new ConcurrentLinkedQueue[TracingSegment]()
private val _metadata = TrieMap.empty[String, String]
override def addMetadata(key: String, value: String): Unit = _metadata.put(key, value)
override def startSegment(segmentName: String, category: String, library: String): Segment = {
startSegment(segmentName, category, library, Map.empty[String, String])
}
override def startSegment(segmentName: String, category: String, library: String, tags: Map[String, String]): Segment = {
_openSegments.incrementAndGet()
val newSegment = new TracingSegment(segmentName, category, library, tags)
_allSegments.add(newSegment)
newSegment
}
override def finish(): Unit = {
super.finish()
traceInfoSink(this)
}
override def finishWithError(cause: Throwable): Unit = {
super.finishWithError(cause)
traceInfoSink(this)
}
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, isFinishedWithError)
}
def shouldIncubate: Boolean = (Status.Open == status) || _openSegments.get() > 0
// Handle with care, should only be used after a trace is finished.
def generateTraceInfo: TraceInfo = {
require(isClosed, "Can't generated a TraceInfo if the Trace has not closed yet.")
val currentSegments = _allSegments.iterator()
var segmentsInfo = List.newBuilder[SegmentInfo]
while (currentSegments.hasNext) {
val segment = currentSegments.next()
if (segment.isClosed)
segmentsInfo += segment.createSegmentInfo(_startTimestamp, startTimestamp)
else
log.warning("Segment [{}] will be left out of TraceInfo because it was still open.", segment.name)
}
TraceInfo(name, token, _startTimestamp, elapsedTime, _metadata.toMap, tags, segmentsInfo.result(), status)
}
class TracingSegment(
segmentName: String,
category: String,
library: String,
tags: Map[String, String]
) extends MetricsOnlySegment(segmentName, category, library, tags) {
private val metadata = TrieMap.empty[String, String]
override def addMetadata(key: String, value: String): Unit = metadata.put(key, value)
// Handle with care, should only be used after the segment has finished.
def createSegmentInfo(traceStartTimestamp: NanoTimestamp, traceRelativeTimestamp: RelativeNanoTimestamp): SegmentInfo = {
require(isClosed, "Can't generated a SegmentInfo if the Segment has not closed yet.")
// We don't have a epoch-based timestamp for the segments because calling System.currentTimeMillis() is both
// expensive and inaccurate, but we can do that once for the trace and calculate all the segments relative to it.
val segmentStartTimestamp = new NanoTimestamp((this.startTimestamp.nanos - traceRelativeTimestamp.nanos) + traceStartTimestamp.nanos)
SegmentInfo(this.name, category, library, segmentStartTimestamp, this.elapsedTime, metadata.toMap, tags, status)
}
}
}
|
