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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
/*
* =========================================================================================
* 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.metrics
import org.HdrHistogram.{ AbstractHistogram, AtomicHistogram }
import kamon.util.GlobPathFilter
import scala.collection.concurrent.TrieMap
import scala.collection.JavaConversions.iterableAsScalaIterable
import akka.actor._
import kamon.metrics.ActorMetricsDispatcher.{ ActorMetricsSnapshot, FlushMetrics }
import kamon.Kamon
import scala.concurrent.duration._
import java.util.concurrent.TimeUnit
import kamon.metrics.ActorMetricsDispatcher.Subscribe
trait ActorMetricsOps {
self: ActorMetricsExtension ⇒
val config = system.settings.config.getConfig("kamon.metrics.actors")
val actorMetrics = TrieMap[String, HdrActorMetricsRecorder]()
val trackedActors: Vector[GlobPathFilter] = config.getStringList("tracked").map(glob ⇒ new GlobPathFilter(glob)).toVector
val excludedActors: Vector[GlobPathFilter] = config.getStringList("excluded").map(glob ⇒ new GlobPathFilter(glob)).toVector
val actorMetricsFactory: () ⇒ HdrActorMetricsRecorder = {
val settings = config.getConfig("hdr-settings")
val processingTimeHdrConfig = HdrConfiguration.fromConfig(settings.getConfig("processing-time"))
val timeInMailboxHdrConfig = HdrConfiguration.fromConfig(settings.getConfig("time-in-mailbox"))
val mailboxSizeHdrConfig = HdrConfiguration.fromConfig(settings.getConfig("mailbox-size"))
() ⇒ new HdrActorMetricsRecorder(processingTimeHdrConfig, timeInMailboxHdrConfig, mailboxSizeHdrConfig)
}
import scala.concurrent.duration._
system.scheduler.schedule(0.seconds, 10.seconds)(
actorMetrics.collect {
case (name, recorder: HdrActorMetricsRecorder) ⇒
println(s"Actor: $name")
recorder.processingTimeHistogram.copy.getHistogramData.outputPercentileDistribution(System.out, 1000000D)
})(system.dispatcher)
def shouldTrackActor(path: String): Boolean =
trackedActors.exists(glob ⇒ glob.accept(path)) && !excludedActors.exists(glob ⇒ glob.accept(path))
def registerActor(path: String): HdrActorMetricsRecorder = actorMetrics.getOrElseUpdate(path, actorMetricsFactory())
def unregisterActor(path: String): Unit = actorMetrics.remove(path)
}
class HdrActorMetricsRecorder(processingTimeHdrConfig: HdrConfiguration, timeInMailboxHdrConfig: HdrConfiguration,
mailboxSizeHdrConfig: HdrConfiguration) {
val processingTimeHistogram = new AtomicHistogram(processingTimeHdrConfig.highestTrackableValue, processingTimeHdrConfig.significantValueDigits)
val timeInMailboxHistogram = new AtomicHistogram(timeInMailboxHdrConfig.highestTrackableValue, timeInMailboxHdrConfig.significantValueDigits)
val mailboxSizeHistogram = new AtomicHistogram(mailboxSizeHdrConfig.highestTrackableValue, mailboxSizeHdrConfig.significantValueDigits)
def recordTimeInMailbox(waitTime: Long): Unit = timeInMailboxHistogram.recordValue(waitTime)
def recordProcessingTime(processingTime: Long): Unit = processingTimeHistogram.recordValue(processingTime)
def snapshot(): HdrActorMetricsSnapshot = {
HdrActorMetricsSnapshot(processingTimeHistogram.copy(), timeInMailboxHistogram.copy(), mailboxSizeHistogram.copy())
}
def reset(): Unit = {
processingTimeHistogram.reset()
timeInMailboxHistogram.reset()
mailboxSizeHistogram.reset()
}
}
case class HdrActorMetricsSnapshot(processingTimeHistogram: AbstractHistogram, timeInMailboxHistogram: AbstractHistogram,
mailboxSizeHistogram: AbstractHistogram)
class ActorMetricsDispatcher extends Actor {
val tickInterval = Duration(context.system.settings.config.getNanoseconds("kamon.metrics.tick-interval"), TimeUnit.NANOSECONDS)
val flushMetricsSchedule = context.system.scheduler.schedule(tickInterval, tickInterval, self, FlushMetrics)(context.dispatcher)
var subscribedForever: Map[GlobPathFilter, List[ActorRef]] = Map.empty
var subscribedForOne: Map[GlobPathFilter, List[ActorRef]] = Map.empty
var lastTick = System.currentTimeMillis()
def receive = {
case Subscribe(path, true) ⇒ subscribeForever(path, sender)
case Subscribe(path, false) ⇒ subscribeOneOff(path, sender)
case FlushMetrics ⇒ flushMetrics()
}
def subscribeForever(path: String, receiver: ActorRef): Unit = subscribedForever = subscribe(receiver, path, subscribedForever)
def subscribeOneOff(path: String, receiver: ActorRef): Unit = subscribedForOne = subscribe(receiver, path, subscribedForOne)
def subscribe(receiver: ActorRef, path: String, target: Map[GlobPathFilter, List[ActorRef]]): Map[GlobPathFilter, List[ActorRef]] = {
val pathFilter = new GlobPathFilter(path)
val oldReceivers = target.get(pathFilter).getOrElse(Nil)
target.updated(pathFilter, receiver :: oldReceivers)
}
def flushMetrics(): Unit = {
val currentTick = System.currentTimeMillis()
val snapshots = Kamon(ActorMetrics)(context.system).actorMetrics.map {
case (path, metrics) ⇒
val snapshot = metrics.snapshot()
metrics.reset()
(path, snapshot)
}.toMap
dispatchMetricsTo(subscribedForOne, snapshots, currentTick)
dispatchMetricsTo(subscribedForever, snapshots, currentTick)
subscribedForOne = Map.empty
lastTick = currentTick
}
def dispatchMetricsTo(subscribers: Map[GlobPathFilter, List[ActorRef]], snapshots: Map[String, HdrActorMetricsSnapshot],
currentTick: Long): Unit = {
for ((subscribedPath, receivers) ← subscribers) {
val metrics = snapshots.filterKeys(snapshotPath ⇒ subscribedPath.accept(snapshotPath))
val actorMetrics = ActorMetricsSnapshot(lastTick, currentTick, metrics)
receivers.foreach(ref ⇒ ref ! actorMetrics)
}
}
}
object ActorMetricsDispatcher {
case class Subscribe(path: String, forever: Boolean = false)
case class UnSubscribe(path: String)
case class ActorMetricsSnapshot(fromMillis: Long, toMillis: Long, metrics: Map[String, HdrActorMetricsSnapshot])
case object FlushMetrics
}
|
