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
|
/*
* =========================================================================================
* Copyright © 2013-2014 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.{ActorLogging, Props, Actor, ActorRef}
import kamon.trace.Incubator.{CheckForCompletedTraces, IncubatingTrace}
import kamon.util.{NanoInterval, RelativeNanoTimestamp}
import scala.annotation.tailrec
import scala.collection.immutable.Queue
import kamon.util.ConfigTools.Syntax
class Incubator(subscriptions: ActorRef) extends Actor with ActorLogging {
import context.dispatcher
val config = context.system.settings.config.getConfig("kamon.trace.incubator")
val minIncubationTime = new NanoInterval(config.getFiniteDuration("min-incubation-time").toNanos)
val maxIncubationTime = new NanoInterval(config.getFiniteDuration("max-incubation-time").toNanos)
val checkInterval = config.getFiniteDuration("check-interval")
val checkSchedule = context.system.scheduler.schedule(checkInterval, checkInterval, self, CheckForCompletedTraces)
var waitingForMinimumIncubation = Queue.empty[IncubatingTrace]
var waitingForIncubationFinish = List.empty[IncubatingTrace]
def receive = {
case tc: TracingContext ⇒ incubate(tc)
case CheckForCompletedTraces ⇒
checkWaitingForMinimumIncubation()
checkWaitingForIncubationFinish()
}
def incubate(tc: TracingContext): Unit =
waitingForMinimumIncubation = waitingForMinimumIncubation.enqueue(IncubatingTrace(tc, RelativeNanoTimestamp.now))
@tailrec private def checkWaitingForMinimumIncubation(): Unit = {
if (waitingForMinimumIncubation.nonEmpty) {
val it = waitingForMinimumIncubation.head
if (NanoInterval.since(it.incubationStart) >= minIncubationTime) {
waitingForMinimumIncubation = waitingForMinimumIncubation.tail
if (it.tc.shouldIncubate)
waitingForIncubationFinish = it :: waitingForIncubationFinish
else
dispatchTraceInfo(it.tc)
checkWaitingForMinimumIncubation()
}
}
}
private def checkWaitingForIncubationFinish(): Unit = {
waitingForIncubationFinish = waitingForIncubationFinish.filter {
case IncubatingTrace(context, incubationStart) ⇒
if (!context.shouldIncubate) {
dispatchTraceInfo(context)
false
} else {
if (NanoInterval.since(incubationStart) >= maxIncubationTime) {
log.warning("Trace [{}] with token [{}] has reached the maximum incubation time, will be reported as is.", context.name, context.token)
dispatchTraceInfo(context);
false
} else true
}
}
}
def dispatchTraceInfo(tc: TracingContext): Unit = subscriptions ! tc.generateTraceInfo
override def postStop(): Unit = {
super.postStop()
checkSchedule.cancel()
}
}
object Incubator {
def props(subscriptions: ActorRef): Props = Props(new Incubator(subscriptions))
case object CheckForCompletedTraces
case class IncubatingTrace(tc: TracingContext, incubationStart: RelativeNanoTimestamp)
}
|
