diff options
author | Ivan Topolnak <itopolnak@despegar.com> | 2014-01-27 19:09:40 -0300 |
---|---|---|
committer | Ivan Topolnak <itopolnak@despegar.com> | 2014-01-27 19:09:40 -0300 |
commit | 966ba65ee12bdc60b231d421ab9d31b7c050b630 (patch) | |
tree | 6024c787845d098224a0485527ec015279920afb /kamon-core/src/main | |
parent | 633b4058207d084724743c6b6367ac60f95ff987 (diff) | |
download | Kamon-966ba65ee12bdc60b231d421ab9d31b7c050b630.tar.gz Kamon-966ba65ee12bdc60b231d421ab9d31b7c050b630.tar.bz2 Kamon-966ba65ee12bdc60b231d421ab9d31b7c050b630.zip |
first implementation of a universal metrics repository for any kind of metrics we might be interested in
Diffstat (limited to 'kamon-core/src/main')
-rw-r--r-- | kamon-core/src/main/resources/reference.conf | 19 | ||||
-rw-r--r-- | kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala | 22 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/metrics/ActorMetrics.scala | 75 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/metrics/ActorMetricsOps.scala | 20 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/metrics/HighDynamicRangeRecorder.scala (renamed from kamon-core/src/main/scala/kamon/metrics/AtomicSnapshotableHistogram.scala) | 33 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/metrics/MetricsExtension.scala | 124 | ||||
-rw-r--r-- | kamon-core/src/main/scala/kamon/metrics/package.scala | 31 |
7 files changed, 237 insertions, 87 deletions
diff --git a/kamon-core/src/main/resources/reference.conf b/kamon-core/src/main/resources/reference.conf index 11e7cbb4..f05c11f6 100644 --- a/kamon-core/src/main/resources/reference.conf +++ b/kamon-core/src/main/resources/reference.conf @@ -2,11 +2,24 @@ kamon { metrics { tick-interval = 1 second - actors { - tracked = [] + filters = [ + { + actor { + includes = [] + excludes = [ "system/*", "user/IO-*" ] + } + }, + { + dispatcher { + includes = [ "default-dispatcher" ] + excludes = [] + } + } + ] + - excluded = [ "system/*", "user/IO-*" ] + actors { hdr-settings { processing-time { highest-trackable-value = 3600000000000 diff --git a/kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala b/kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala index 6cede344..68d606ba 100644 --- a/kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala +++ b/kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala @@ -20,26 +20,24 @@ import org.aspectj.lang.ProceedingJoinPoint import akka.actor.{ Cell, Props, ActorSystem, ActorRef } import akka.dispatch.{ Envelope, MessageDispatcher } import kamon.trace.{ TraceContext, ContextAware, Trace } -import kamon.metrics.{ HdrActorMetricsRecorder, ActorMetrics } +import kamon.metrics.{ ActorMetrics, HdrActorMetricsRecorder, Metrics } import kamon.Kamon +import kamon.metrics.ActorMetrics.ActorMetricRecorder @Aspect("perthis(actorCellCreation(*, *, *, *, *))") class BehaviourInvokeTracing { - var path: Option[String] = None - var actorMetrics: Option[HdrActorMetricsRecorder] = None + var path: String = _ + var actorMetrics: Option[ActorMetricRecorder] = None @Pointcut("execution(akka.actor.ActorCell.new(..)) && args(system, ref, props, dispatcher, parent)") def actorCellCreation(system: ActorSystem, ref: ActorRef, props: Props, dispatcher: MessageDispatcher, parent: ActorRef): Unit = {} @After("actorCellCreation(system, ref, props, dispatcher, parent)") def afterCreation(system: ActorSystem, ref: ActorRef, props: Props, dispatcher: MessageDispatcher, parent: ActorRef): Unit = { - val metricsExtension = Kamon(ActorMetrics)(system) - val simplePathString = ref.path.elements.mkString("/") + val metricsExtension = Kamon(Metrics)(system) - if (metricsExtension.shouldTrackActor(simplePathString)) { - path = Some(ref.path.toString) - actorMetrics = Some(metricsExtension.registerActor(simplePathString)) - } + path = ref.path.elements.mkString("/") + actorMetrics = metricsExtension.register(path, ActorMetrics) } @Pointcut("(execution(* akka.actor.ActorCell.invoke(*)) || execution(* akka.routing.RoutedActorCell.sendMessage(*))) && args(envelope)") @@ -55,8 +53,8 @@ class BehaviourInvokeTracing { } actorMetrics.map { am ⇒ - am.recordProcessingTime(System.nanoTime() - timestampBeforeProcessing) - am.recordTimeInMailbox(timestampBeforeProcessing - contextAndTimestamp.timestamp) + am.processingTime.record(System.nanoTime() - timestampBeforeProcessing) + am.timeInMailbox.record(timestampBeforeProcessing - contextAndTimestamp.timestamp) } } @@ -65,7 +63,7 @@ class BehaviourInvokeTracing { @After("actorStop(cell)") def afterStop(cell: Cell): Unit = { - path.map(p ⇒ Kamon(ActorMetrics)(cell.system).unregisterActor(p)) + actorMetrics.map(p ⇒ Kamon(Metrics)(cell.system).unregister(path, ActorMetrics)) } } diff --git a/kamon-core/src/main/scala/kamon/metrics/ActorMetrics.scala b/kamon-core/src/main/scala/kamon/metrics/ActorMetrics.scala index 72e473e8..96d2cd48 100644 --- a/kamon-core/src/main/scala/kamon/metrics/ActorMetrics.scala +++ b/kamon-core/src/main/scala/kamon/metrics/ActorMetrics.scala @@ -1,31 +1,66 @@ -/* =================================================== +/* + * ========================================================================================= * 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 + * 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 + * 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. - * ========================================================== */ + * 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 akka.actor.{ Props, ExtendedActorSystem, ExtensionIdProvider, ExtensionId } -import akka.actor -import kamon.Kamon +import com.typesafe.config.Config +import org.HdrHistogram.HighDynamicRangeRecorder -object ActorMetrics extends ExtensionId[ActorMetricsExtension] with ExtensionIdProvider { - def lookup(): ExtensionId[_ <: actor.Extension] = ActorMetrics - def createExtension(system: ExtendedActorSystem): ActorMetricsExtension = new ActorMetricsExtension(system) +object ActorMetrics extends MetricGroupIdentity.Category with MetricGroupFactory { + val name: String = "actor" + type Group = ActorMetricRecorder -} + case object ProcessingTime extends MetricIdentity { val name = "ProcessingTime" } + case object MailboxSize extends MetricIdentity { val name = "MailboxSize" } + case object TimeInMailbox extends MetricIdentity { val name = "TimeInMailbox" } + + case class ActorMetricRecorder(processingTime: MetricRecorder, mailboxSize: MetricRecorder, timeInMailbox: MetricRecorder) + extends MetricGroupRecorder { + + def record(identity: MetricIdentity, value: Long): Unit = identity match { + case ProcessingTime ⇒ processingTime.record(value) + case MailboxSize ⇒ mailboxSize.record(value) + case TimeInMailbox ⇒ timeInMailbox.record(value) + } + + def collect: MetricGroupSnapshot = { + ActorMetricSnapshot(processingTime.collect(), mailboxSize.collect(), timeInMailbox.collect()) + } + } + + case class ActorMetricSnapshot(processingTime: MetricSnapshot, mailboxSize: MetricSnapshot, timeInMailbox: MetricSnapshot) + extends MetricGroupSnapshot { + + def metrics: Map[MetricIdentity, MetricSnapshot] = Map( + (ProcessingTime -> processingTime), + (MailboxSize -> mailboxSize), + (TimeInMailbox -> timeInMailbox)) + } + + def create(config: Config): ActorMetricRecorder = { + import HighDynamicRangeRecorder.Configuration + + val settings = config.getConfig("kamon.metrics.actors.hdr-settings") + val processingTimeHdrConfig = Configuration.fromConfig(settings.getConfig("processing-time")) + val mailboxSizeHdrConfig = Configuration.fromConfig(settings.getConfig("mailbox-size")) + val timeInMailboxHdrConfig = Configuration.fromConfig(settings.getConfig("time-in-mailbox")) -class ActorMetricsExtension(val system: ExtendedActorSystem) extends Kamon.Extension with ActorMetricsOps { - lazy val metricsDispatcher = system.actorOf(Props[ActorMetricsDispatcher], "kamon-actor-metrics") + ActorMetricRecorder( + HighDynamicRangeRecorder(processingTimeHdrConfig), + HighDynamicRangeRecorder(mailboxSizeHdrConfig), + HighDynamicRangeRecorder(timeInMailboxHdrConfig)) + } } diff --git a/kamon-core/src/main/scala/kamon/metrics/ActorMetricsOps.scala b/kamon-core/src/main/scala/kamon/metrics/ActorMetricsOps.scala index 4a6c98f4..0e3af5fd 100644 --- a/kamon-core/src/main/scala/kamon/metrics/ActorMetricsOps.scala +++ b/kamon-core/src/main/scala/kamon/metrics/ActorMetricsOps.scala @@ -16,7 +16,7 @@ package kamon.metrics -import org.HdrHistogram.{ AbstractHistogram, AtomicHistogram } +import org.HdrHistogram.{ HighDynamicRangeRecorder, AbstractHistogram, AtomicHistogram } import kamon.util.GlobPathFilter import scala.collection.concurrent.TrieMap import scala.collection.JavaConversions.iterableAsScalaIterable @@ -28,7 +28,7 @@ import java.util.concurrent.TimeUnit import kamon.metrics.ActorMetricsDispatcher.Subscribe trait ActorMetricsOps { - self: ActorMetricsExtension ⇒ + self: MetricsExtension ⇒ val config = system.settings.config.getConfig("kamon.metrics.actors") val actorMetrics = TrieMap[String, HdrActorMetricsRecorder]() @@ -38,9 +38,9 @@ trait ActorMetricsOps { 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")) + val processingTimeHdrConfig = HighDynamicRangeRecorder.Configuration.fromConfig(settings.getConfig("processing-time")) + val timeInMailboxHdrConfig = HighDynamicRangeRecorder.Configuration.fromConfig(settings.getConfig("time-in-mailbox")) + val mailboxSizeHdrConfig = HighDynamicRangeRecorder.Configuration.fromConfig(settings.getConfig("mailbox-size")) () ⇒ new HdrActorMetricsRecorder(processingTimeHdrConfig, timeInMailboxHdrConfig, mailboxSizeHdrConfig) } @@ -53,8 +53,8 @@ trait ActorMetricsOps { def unregisterActor(path: String): Unit = actorMetrics.remove(path) } -class HdrActorMetricsRecorder(processingTimeHdrConfig: HdrConfiguration, timeInMailboxHdrConfig: HdrConfiguration, - mailboxSizeHdrConfig: HdrConfiguration) { +class HdrActorMetricsRecorder(processingTimeHdrConfig: HighDynamicRangeRecorder.Configuration, timeInMailboxHdrConfig: HighDynamicRangeRecorder.Configuration, + mailboxSizeHdrConfig: HighDynamicRangeRecorder.Configuration) { val processingTimeHistogram = new AtomicHistogram(processingTimeHdrConfig.highestTrackableValue, processingTimeHdrConfig.significantValueDigits) val timeInMailboxHistogram = new AtomicHistogram(timeInMailboxHdrConfig.highestTrackableValue, timeInMailboxHdrConfig.significantValueDigits) @@ -103,8 +103,8 @@ class ActorMetricsDispatcher extends Actor { } def flushMetrics(): Unit = { - val currentTick = System.currentTimeMillis() - val snapshots = Kamon(ActorMetrics)(context.system).actorMetrics.map { + /* val currentTick = System.currentTimeMillis() + val snapshots = Kamon(Metrics)(context.system).actorMetrics.map { case (path, metrics) ⇒ val snapshot = metrics.snapshot() metrics.reset() @@ -116,7 +116,7 @@ class ActorMetricsDispatcher extends Actor { dispatchMetricsTo(subscribedForever, snapshots, currentTick) subscribedForOne = Map.empty - lastTick = currentTick + lastTick = currentTick*/ } def dispatchMetricsTo(subscribers: Map[GlobPathFilter, List[ActorRef]], snapshots: Map[String, HdrActorMetricsSnapshot], diff --git a/kamon-core/src/main/scala/kamon/metrics/AtomicSnapshotableHistogram.scala b/kamon-core/src/main/scala/kamon/metrics/HighDynamicRangeRecorder.scala index c9e47792..e31d0e11 100644 --- a/kamon-core/src/main/scala/kamon/metrics/AtomicSnapshotableHistogram.scala +++ b/kamon-core/src/main/scala/kamon/metrics/HighDynamicRangeRecorder.scala @@ -18,20 +18,24 @@ package org.HdrHistogram import java.util.concurrent.atomic.AtomicLongFieldUpdater import scala.annotation.tailrec -import org.HdrHistogram.AtomicSnapshotableHistogram.{ Value, Snapshot } +import kamon.metrics.{ DefaultMetricSnapshot, MetricSnapshot, MetricRecorder } +import com.typesafe.config.Config +import org.HdrHistogram.HighDynamicRangeRecorder.Configuration /** * This implementation aims to be used for real time data collection where data snapshots are taken often over time. * The snapshotAndReset() operation extracts all the recorded values from the histogram and resets the counts, but still * leave it in a consistent state even in the case of concurrent modification while the snapshot is being taken. */ -class AtomicSnapshotableHistogram(highestTrackableValue: Long, numberOfSignificantValueDigits: Int) - extends AtomicHistogram(1L, highestTrackableValue, numberOfSignificantValueDigits) { +class HighDynamicRangeRecorder(configuration: Configuration) + extends AtomicHistogram(1L, configuration.highestTrackableValue, configuration.significantValueDigits) with MetricRecorder { - import AtomicSnapshotableHistogram.totalCountUpdater + import HighDynamicRangeRecorder.totalCountUpdater - def snapshotAndReset(): Snapshot = { - val entries = Vector.newBuilder[Value] + def record(value: Long): Unit = recordValue(value) + + def collect(): MetricSnapshot = { + val entries = Vector.newBuilder[MetricSnapshot.Measurement] val countsLength = counts.length() @tailrec def iterate(index: Int, previousValue: Long, nrOfRecordings: Long, bucketLimit: Long, increment: Long): Long = { @@ -40,7 +44,7 @@ class AtomicSnapshotableHistogram(highestTrackableValue: Long, numberOfSignifica val countAtValue = counts.getAndSet(index, 0) if (countAtValue > 0) - entries += Value(currentValue, countAtValue) + entries += MetricSnapshot.Measurement(currentValue, countAtValue) if (currentValue == bucketLimit) iterate(index + 1, currentValue, nrOfRecordings + countAtValue, (bucketLimit << 1) + 1, increment << 1) @@ -62,14 +66,21 @@ class AtomicSnapshotableHistogram(highestTrackableValue: Long, numberOfSignifica while (!tryUpdateTotalCount) {} - Snapshot(nrOfRecordings, entries.result()) + DefaultMetricSnapshot(nrOfRecordings, entries.result()) } } -object AtomicSnapshotableHistogram { +object HighDynamicRangeRecorder { val totalCountUpdater = AtomicLongFieldUpdater.newUpdater(classOf[AtomicHistogram], "totalCount") - case class Snapshot(nrOfRecordings: Long, values: Vector[Value]) - case class Value(magnitude: Long, count: Long) + def apply(configuration: Configuration): HighDynamicRangeRecorder = new HighDynamicRangeRecorder(configuration) + + case class Configuration(highestTrackableValue: Long, significantValueDigits: Int) + + case object Configuration { + def fromConfig(config: Config): Configuration = { + Configuration(config.getLong("highest-trackable-value"), config.getInt("significant-value-digits")) + } + } } diff --git a/kamon-core/src/main/scala/kamon/metrics/MetricsExtension.scala b/kamon-core/src/main/scala/kamon/metrics/MetricsExtension.scala new file mode 100644 index 00000000..4bac3519 --- /dev/null +++ b/kamon-core/src/main/scala/kamon/metrics/MetricsExtension.scala @@ -0,0 +1,124 @@ +/* + * ========================================================================================= + * 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 scala.collection.concurrent.TrieMap +import akka.actor.{ ExtensionIdProvider, ExtensionId, ExtendedActorSystem } +import com.typesafe.config.Config +import kamon.util.GlobPathFilter +import kamon.Kamon +import akka.actor +import kamon.metrics.Metrics.MetricGroupFilter + +case class MetricGroupIdentity(name: String, category: MetricGroupIdentity.Category) + +trait MetricIdentity { + def name: String +} + +trait MetricGroupRecorder { + def record(identity: MetricIdentity, value: Long) + def collect: MetricGroupSnapshot +} + +trait MetricGroupSnapshot { + def metrics: Map[MetricIdentity, MetricSnapshot] +} + +trait MetricRecorder { + def record(value: Long) + def collect(): MetricSnapshot +} + +trait MetricSnapshot { + def numberOfMeasurements: Long + def measurementLevels: Vector[MetricSnapshot.Measurement] +} + +object MetricSnapshot { + case class Measurement(value: Long, count: Long) +} + +case class DefaultMetricSnapshot(numberOfMeasurements: Long, measurementLevels: Vector[MetricSnapshot.Measurement]) extends MetricSnapshot + +object MetricGroupIdentity { + trait Category { + def name: String + } +} + +trait MetricGroupFactory { + type Group <: MetricGroupRecorder + def create(config: Config): Group +} + +class MetricsExtension(val system: ExtendedActorSystem) extends Kamon.Extension { + val config = system.settings.config + val storage = TrieMap[MetricGroupIdentity, MetricGroupRecorder]() + val filters = loadFilters(config) + + def register(name: String, category: MetricGroupIdentity.Category with MetricGroupFactory): Option[category.Group] = { + if (shouldTrack(name, category)) + Some(storage.getOrElseUpdate(MetricGroupIdentity(name, category), category.create(config)).asInstanceOf[category.Group]) + else + None + } + + def unregister(name: String, category: MetricGroupIdentity.Category with MetricGroupFactory): Unit = { + storage.remove(MetricGroupIdentity(name, category)) + } + + def collect: Map[MetricGroupIdentity, MetricGroupSnapshot] = { + (for ((identity, recorder) ← storage) yield (identity, recorder.collect)).toMap + } + + private def shouldTrack(name: String, category: MetricGroupIdentity.Category): Boolean = { + filters.get(category.name).map(filter ⇒ filter.accept(name)).getOrElse(false) + } + + def loadFilters(config: Config): Map[String, MetricGroupFilter] = { + import scala.collection.JavaConverters._ + + val filters = config.getObjectList("kamon.metrics.filters").asScala + + val allFilters = + for ( + filter ← filters; + entry ← filter.entrySet().asScala + ) yield { + val key = entry.getKey + val keyBasedConfig = entry.getValue.atKey(key) + + val includes = keyBasedConfig.getStringList(s"$key.includes").asScala.map(inc ⇒ new GlobPathFilter(inc)).toList + val excludes = keyBasedConfig.getStringList(s"$key.excludes").asScala.map(exc ⇒ new GlobPathFilter(exc)).toList + + (key, MetricGroupFilter(includes, excludes)) + } + + allFilters.toMap + } + +} + +object Metrics extends ExtensionId[MetricsExtension] with ExtensionIdProvider { + def lookup(): ExtensionId[_ <: actor.Extension] = Metrics + def createExtension(system: ExtendedActorSystem): MetricsExtension = new MetricsExtension(system) + + case class MetricGroupFilter(includes: List[GlobPathFilter], excludes: List[GlobPathFilter]) { + def accept(name: String): Boolean = includes.exists(_.accept(name)) && !excludes.exists(_.accept(name)) + } +} diff --git a/kamon-core/src/main/scala/kamon/metrics/package.scala b/kamon-core/src/main/scala/kamon/metrics/package.scala deleted file mode 100644 index d6359ead..00000000 --- a/kamon-core/src/main/scala/kamon/metrics/package.scala +++ /dev/null @@ -1,31 +0,0 @@ -/* - * ========================================================================================= - * 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 - -import scala.concurrent.duration._ -import com.typesafe.config.Config - -package object metrics { - val OneHour = 1.hour.toNanos - - case class HdrConfiguration(highestTrackableValue: Long, significantValueDigits: Int) - case object HdrConfiguration { - def fromConfig(config: Config): HdrConfiguration = { - HdrConfiguration(config.getLong("highest-trackable-value"), config.getInt("significant-value-digits")) - } - } -} |