diff options
author | Diego <diegolparra@gmail.com> | 2014-12-27 17:09:38 -0300 |
---|---|---|
committer | Ivan Topolnjak <ivantopo@gmail.com> | 2014-12-28 18:12:27 +0100 |
commit | 58ec8bbe6a5192dc9cba9752324e1f793da156f2 (patch) | |
tree | 03e38722c5da10ef91997835f417a57f142ad6ec /kamon-system-metrics | |
parent | a7c2289a31f841c4abf360db197c355e766fb3e0 (diff) | |
download | Kamon-58ec8bbe6a5192dc9cba9752324e1f793da156f2.tar.gz Kamon-58ec8bbe6a5192dc9cba9752324e1f793da156f2.tar.bz2 Kamon-58ec8bbe6a5192dc9cba9752324e1f793da156f2.zip |
+ system-metrics: fix #135 and includes the following new metrics:
** DiskMetrics
** NonHeapMetrics
** LoadAverageMetrics
** ThreadMetrics
** ClassLoadingMetrics
and closes #131
Diffstat (limited to 'kamon-system-metrics')
12 files changed, 825 insertions, 178 deletions
diff --git a/kamon-system-metrics/src/main/resources/reference.conf b/kamon-system-metrics/src/main/resources/reference.conf index 3de463b3..ba439024 100644 --- a/kamon-system-metrics/src/main/resources/reference.conf +++ b/kamon-system-metrics/src/main/resources/reference.conf @@ -7,15 +7,28 @@ kamon.sigar.folder = ${user.dir}"/native" kamon { + system-metrics { + + default-gauge-precision { + refresh-interval = 1 second + highest-trackable-value = 999999999 + significant-value-digits = 2 + } + + # Default dispatcher for all system-metrics module operations + dispatcher = ${kamon.default-dispatcher} + } + metrics { precision { + system { process-cpu { - cpu-percentage = { + cpu-percentage = { highest-trackable-value = 999999999 significant-value-digits = 2 } - total-process-time = { + total-process-time = { highest-trackable-value = 999999999 significant-value-digits = 2 } @@ -34,11 +47,11 @@ kamon { highest-trackable-value = 999 significant-value-digits = 2 } - idle ={ + idle = { highest-trackable-value = 999 significant-value-digits = 2 } - stolen ={ + stolen = { highest-trackable-value = 999 significant-value-digits = 2 } @@ -67,18 +80,49 @@ kamon { per-process-non-voluntary = ${kamon.metrics.precision.default-histogram-precision} global = ${kamon.metrics.precision.default-histogram-precision} } + + disk { + reads = ${kamon.metrics.precision.default-histogram-precision} + writes = ${kamon.metrics.precision.default-histogram-precision} + queue = ${kamon.metrics.precision.default-histogram-precision} + service-time = ${kamon.metrics.precision.default-histogram-precision} + } + + load-average { + one = ${kamon.metrics.precision.default-histogram-precision} + five = ${kamon.metrics.precision.default-histogram-precision} + fifteen = ${kamon.metrics.precision.default-histogram-precision} + } } jvm { heap { - used = ${kamon.metrics.precision.default-gauge-precision} - max = ${kamon.metrics.precision.default-gauge-precision} - committed = ${kamon.metrics.precision.default-gauge-precision} + used = ${kamon.system-metrics.default-gauge-precision} + max = ${kamon.system-metrics.default-gauge-precision} + committed = ${kamon.system-metrics.default-gauge-precision} + } + + non-heap { + used = ${kamon.system-metrics.default-gauge-precision} + max = ${kamon.system-metrics.default-gauge-precision} + committed = ${kamon.system-metrics.default-gauge-precision} + } + + thread { + daemon = ${kamon.system-metrics.default-gauge-precision} + count = ${kamon.system-metrics.default-gauge-precision} + peak = ${kamon.system-metrics.default-gauge-precision} + } + + classes { + total-loaded = ${kamon.system-metrics.default-gauge-precision} + total-unloaded = ${kamon.system-metrics.default-gauge-precision} + current-loaded = ${kamon.system-metrics.default-gauge-precision} } gc { - count = ${kamon.metrics.precision.default-gauge-precision} - time = ${kamon.metrics.precision.default-gauge-precision} + count = ${kamon.metrics.precision.default-histogram-precision} + time = ${kamon.metrics.precision.default-histogram-precision} } } } diff --git a/kamon-system-metrics/src/main/scala/kamon/metrics/ClassLoadingMetrics.scala b/kamon-system-metrics/src/main/scala/kamon/metrics/ClassLoadingMetrics.scala new file mode 100644 index 00000000..1e3bee27 --- /dev/null +++ b/kamon-system-metrics/src/main/scala/kamon/metrics/ClassLoadingMetrics.scala @@ -0,0 +1,85 @@ +/* + * ========================================================================================= + * 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.metrics + +import java.lang.management.ManagementFactory + +import akka.actor.ActorSystem +import com.typesafe.config.Config +import kamon.metric._ +import kamon.metric.instrument.{ Gauge, Histogram } + +case class ClassLoadingMetrics(name: String) extends MetricGroupIdentity { + val category = ClassLoadingMetrics +} + +object ClassLoadingMetrics extends MetricGroupCategory { + val name = "classes" + + case object Loaded extends MetricIdentity { val name = "total-loaded" } + case object Unloaded extends MetricIdentity { val name = "total-unloaded" } + case object Current extends MetricIdentity { val name = "current-loaded" } + + case class ClassLoadingMetricRecorder(loaded: Gauge, unloaded: Gauge, current: Gauge) + extends MetricGroupRecorder { + + def collect(context: CollectionContext): MetricGroupSnapshot = { + ClassLoadingMetricSnapshot(loaded.collect(context), unloaded.collect(context), current.collect(context)) + } + + def cleanup: Unit = {} + } + + case class ClassLoadingMetricSnapshot(loaded: Histogram.Snapshot, unloaded: Histogram.Snapshot, current: Histogram.Snapshot) + extends MetricGroupSnapshot { + + type GroupSnapshotType = ClassLoadingMetricSnapshot + + def merge(that: GroupSnapshotType, context: CollectionContext): GroupSnapshotType = { + ClassLoadingMetricSnapshot(loaded.merge(that.loaded, context), unloaded.merge(that.unloaded, context), current.merge(that.current, context)) + } + + lazy val metrics: Map[MetricIdentity, MetricSnapshot] = Map( + Loaded -> loaded, + Unloaded -> unloaded, + Current -> current) + } + + val Factory = ClassLoadingMetricGroupFactory +} + +case object ClassLoadingMetricGroupFactory extends MetricGroupFactory { + + import ClassLoadingMetrics._ + + val classes = ManagementFactory.getClassLoadingMXBean + + type GroupRecorder = ClassLoadingMetricRecorder + + def create(config: Config, system: ActorSystem): GroupRecorder = { + val settings = config.getConfig("precision.jvm.classes") + + val totalLoadedConfig = settings.getConfig("total-loaded") + val totalUnloadedConfig = settings.getConfig("total-unloaded") + val currentLoadedConfig = settings.getConfig("current-loaded") + + new ClassLoadingMetricRecorder( + Gauge.fromConfig(totalLoadedConfig, system)(() ⇒ classes.getTotalLoadedClassCount), + Gauge.fromConfig(totalUnloadedConfig, system)(() ⇒ classes.getUnloadedClassCount), + Gauge.fromConfig(currentLoadedConfig, system)(() ⇒ classes.getLoadedClassCount.toLong)) + } +} + diff --git a/kamon-system-metrics/src/main/scala/kamon/metrics/DiskMetrics.scala b/kamon-system-metrics/src/main/scala/kamon/metrics/DiskMetrics.scala new file mode 100644 index 00000000..eeb6002b --- /dev/null +++ b/kamon-system-metrics/src/main/scala/kamon/metrics/DiskMetrics.scala @@ -0,0 +1,85 @@ +/* + * ========================================================================================= + * 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.metrics + +import akka.actor.ActorSystem +import com.typesafe.config.Config +import kamon.metric._ +import kamon.metric.instrument.Histogram + +case class DiskMetrics(name: String) extends MetricGroupIdentity { + val category = DiskMetrics +} + +object DiskMetrics extends MetricGroupCategory { + val name = "disk" + + case object Reads extends MetricIdentity { val name = "reads" } + case object Writes extends MetricIdentity { val name = "writes" } + case object Queue extends MetricIdentity { val name = "queue" } + case object ServiceTime extends MetricIdentity { val name = "service-time" } + + case class DiskMetricsRecorder(reads: Histogram, writes: Histogram, queue: Histogram, serviceTime: Histogram) + extends MetricGroupRecorder { + + def collect(context: CollectionContext): MetricGroupSnapshot = { + DiskMetricsSnapshot(reads.collect(context), writes.collect(context), queue.collect(context), serviceTime.collect(context)) + } + + def cleanup: Unit = {} + } + + case class DiskMetricsSnapshot(reads: Histogram.Snapshot, writes: Histogram.Snapshot, queue: Histogram.Snapshot, serviceTime: Histogram.Snapshot) + extends MetricGroupSnapshot { + + type GroupSnapshotType = DiskMetricsSnapshot + + def merge(that: GroupSnapshotType, context: CollectionContext): GroupSnapshotType = { + DiskMetricsSnapshot(reads.merge(that.reads, context), writes.merge(that.writes, context), queue.merge(that.queue, context), serviceTime.merge(that.serviceTime, context)) + } + + lazy val metrics: Map[MetricIdentity, MetricSnapshot] = Map( + Reads -> reads, + Writes -> writes, + Queue -> queue, + ServiceTime -> serviceTime) + } + + val Factory = DiskMetricGroupFactory +} + +case object DiskMetricGroupFactory extends MetricGroupFactory { + + import DiskMetrics._ + + type GroupRecorder = DiskMetricsRecorder + + def create(config: Config, system: ActorSystem): GroupRecorder = { + val settings = config.getConfig("precision.system.disk") + + val readsDiskConfig = settings.getConfig("reads") + val writesDiskConfig = settings.getConfig("writes") + val queueDiskConfig = settings.getConfig("queue") + val serviceTimeDiskConfig = settings.getConfig("service-time") + + new DiskMetricsRecorder( + Histogram.fromConfig(readsDiskConfig), + Histogram.fromConfig(writesDiskConfig), + Histogram.fromConfig(queueDiskConfig), + Histogram.fromConfig(serviceTimeDiskConfig)) + } +} + diff --git a/kamon-system-metrics/src/main/scala/kamon/metrics/GCMetrics.scala b/kamon-system-metrics/src/main/scala/kamon/metrics/GCMetrics.scala index bc5fc724..5aa679c9 100644 --- a/kamon-system-metrics/src/main/scala/kamon/metrics/GCMetrics.scala +++ b/kamon-system-metrics/src/main/scala/kamon/metrics/GCMetrics.scala @@ -20,7 +20,7 @@ import java.lang.management.GarbageCollectorMXBean import akka.actor.ActorSystem import com.typesafe.config.Config import kamon.metric._ -import kamon.metric.instrument.{ Gauge, Histogram } +import kamon.metric.instrument.Histogram case class GCMetrics(name: String) extends MetricGroupIdentity { val category = GCMetrics @@ -32,7 +32,7 @@ object GCMetrics extends MetricGroupCategory { case object CollectionCount extends MetricIdentity { val name = "collection-count" } case object CollectionTime extends MetricIdentity { val name = "collection-time" } - case class GCMetricRecorder(count: Gauge, time: Gauge) + case class GCMetricRecorder(count: Histogram, time: Histogram) extends MetricGroupRecorder { def collect(context: CollectionContext): MetricGroupSnapshot = { @@ -71,8 +71,7 @@ case class GCMetricGroupFactory(gc: GarbageCollectorMXBean) extends MetricGroupF val timeConfig = settings.getConfig("time") new GCMetricRecorder( - Gauge.fromConfig(countConfig, system)(() ⇒ gc.getCollectionCount), - Gauge.fromConfig(timeConfig, system, Scale.Milli)(() ⇒ gc.getCollectionTime)) + Histogram.fromConfig(countConfig), + Histogram.fromConfig(timeConfig)) } -} - +}
\ No newline at end of file diff --git a/kamon-system-metrics/src/main/scala/kamon/metrics/HeapMetrics.scala b/kamon-system-metrics/src/main/scala/kamon/metrics/HeapMetrics.scala index ac033fe2..5bba5bf6 100644 --- a/kamon-system-metrics/src/main/scala/kamon/metrics/HeapMetrics.scala +++ b/kamon-system-metrics/src/main/scala/kamon/metrics/HeapMetrics.scala @@ -29,9 +29,9 @@ case class HeapMetrics(name: String) extends MetricGroupIdentity { object HeapMetrics extends MetricGroupCategory { val name = "heap" - case object Used extends MetricIdentity { val name = "used-heap" } - case object Max extends MetricIdentity { val name = "max-heap" } - case object Committed extends MetricIdentity { val name = "committed-heap" } + case object Used extends MetricIdentity { val name = "used" } + case object Max extends MetricIdentity { val name = "max" } + case object Committed extends MetricIdentity { val name = "committed" } case class HeapMetricRecorder(used: Gauge, max: Gauge, committed: Gauge) extends MetricGroupRecorder { diff --git a/kamon-system-metrics/src/main/scala/kamon/metrics/LoadAverageMetrics.scala b/kamon-system-metrics/src/main/scala/kamon/metrics/LoadAverageMetrics.scala new file mode 100644 index 00000000..cd196adf --- /dev/null +++ b/kamon-system-metrics/src/main/scala/kamon/metrics/LoadAverageMetrics.scala @@ -0,0 +1,80 @@ +/* + * ========================================================================================= + * 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.metrics + +import akka.actor.ActorSystem +import com.typesafe.config.Config +import kamon.metric._ +import kamon.metric.instrument.Histogram + +case class LoadAverageMetrics(name: String) extends MetricGroupIdentity { + val category = LoadAverageMetrics +} + +object LoadAverageMetrics extends MetricGroupCategory { + val name = "load-average" + + case object OneMinute extends MetricIdentity { val name = "last-minute" } + case object FiveMinutes extends MetricIdentity { val name = "last-five-minutes" } + case object FifteenMinutes extends MetricIdentity { val name = "last-fifteen-minutes" } + + case class LoadAverageMetricsRecorder(one: Histogram, five: Histogram, fifteen: Histogram) + extends MetricGroupRecorder { + + def collect(context: CollectionContext): MetricGroupSnapshot = { + LoadAverageMetricsSnapshot(one.collect(context), five.collect(context), fifteen.collect(context)) + } + + def cleanup: Unit = {} + } + + case class LoadAverageMetricsSnapshot(one: Histogram.Snapshot, five: Histogram.Snapshot, fifteen: Histogram.Snapshot) + extends MetricGroupSnapshot { + + type GroupSnapshotType = LoadAverageMetricsSnapshot + + def merge(that: GroupSnapshotType, context: CollectionContext): GroupSnapshotType = { + LoadAverageMetricsSnapshot(one.merge(that.one, context), five.merge(that.five, context), fifteen.merge(that.fifteen, context)) + } + + lazy val metrics: Map[MetricIdentity, MetricSnapshot] = Map( + OneMinute -> one, + FiveMinutes -> five, + FifteenMinutes -> fifteen) + } + + val Factory = LoadAverageMetricGroupFactory +} + +case object LoadAverageMetricGroupFactory extends MetricGroupFactory { + + import LoadAverageMetrics._ + + type GroupRecorder = LoadAverageMetricsRecorder + + def create(config: Config, system: ActorSystem): GroupRecorder = { + val settings = config.getConfig("precision.system.load-average") + + val oneMinuteConfig = settings.getConfig("one") + val fiveMinutesConfig = settings.getConfig("five") + val fifteenMinutesConfig = settings.getConfig("fifteen") + + new LoadAverageMetricsRecorder( + Histogram.fromConfig(oneMinuteConfig), + Histogram.fromConfig(fiveMinutesConfig), + Histogram.fromConfig(fifteenMinutesConfig)) + } +} diff --git a/kamon-system-metrics/src/main/scala/kamon/metrics/NonHeapMetrics.scala b/kamon-system-metrics/src/main/scala/kamon/metrics/NonHeapMetrics.scala new file mode 100644 index 00000000..c2b9f9af --- /dev/null +++ b/kamon-system-metrics/src/main/scala/kamon/metrics/NonHeapMetrics.scala @@ -0,0 +1,86 @@ +/* + * ========================================================================================= + * 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.metrics + +import java.lang.management.ManagementFactory + +import akka.actor.ActorSystem +import com.typesafe.config.Config +import kamon.metric._ +import kamon.metric.instrument.{ Gauge, Histogram } + +case class NonHeapMetrics(name: String) extends MetricGroupIdentity { + val category = NonHeapMetrics +} + +object NonHeapMetrics extends MetricGroupCategory { + val name = "non-heap" + + case object Used extends MetricIdentity { val name = "used" } + case object Max extends MetricIdentity { val name = "max" } + case object Committed extends MetricIdentity { val name = "committed" } + + case class NonHeapMetricRecorder(used: Gauge, max: Gauge, committed: Gauge) + extends MetricGroupRecorder { + + def collect(context: CollectionContext): MetricGroupSnapshot = { + NonHeapMetricSnapshot(used.collect(context), max.collect(context), committed.collect(context)) + } + + def cleanup: Unit = {} + } + + case class NonHeapMetricSnapshot(used: Histogram.Snapshot, max: Histogram.Snapshot, committed: Histogram.Snapshot) + extends MetricGroupSnapshot { + + type GroupSnapshotType = NonHeapMetricSnapshot + + def merge(that: GroupSnapshotType, context: CollectionContext): GroupSnapshotType = { + NonHeapMetricSnapshot(used.merge(that.used, context), max.merge(that.max, context), committed.merge(that.committed, context)) + } + + lazy val metrics: Map[MetricIdentity, MetricSnapshot] = Map( + Used -> used, + Max -> max, + Committed -> committed) + } + + val Factory = NonHeapMetricGroupFactory +} + +case object NonHeapMetricGroupFactory extends MetricGroupFactory { + + import NonHeapMetrics._ + import kamon.system.SystemMetricsExtension._ + + def nonHeap = ManagementFactory.getMemoryMXBean.getNonHeapMemoryUsage + + type GroupRecorder = NonHeapMetricRecorder + + def create(config: Config, system: ActorSystem): GroupRecorder = { + val settings = config.getConfig("precision.jvm.non-heap") + + val usedNonHeapConfig = settings.getConfig("used") + val maxNonHeapConfig = settings.getConfig("max") + val committedNonHeapConfig = settings.getConfig("committed") + + new NonHeapMetricRecorder( + Gauge.fromConfig(usedNonHeapConfig, system, Scale.Mega)(() ⇒ toMB(nonHeap.getUsed)), + Gauge.fromConfig(maxNonHeapConfig, system, Scale.Mega)(() ⇒ toMB(nonHeap.getMax)), + Gauge.fromConfig(committedNonHeapConfig, system, Scale.Mega)(() ⇒ toMB(nonHeap.getCommitted))) + } +} + diff --git a/kamon-system-metrics/src/main/scala/kamon/metrics/ThreadMetrics.scala b/kamon-system-metrics/src/main/scala/kamon/metrics/ThreadMetrics.scala new file mode 100644 index 00000000..fc039ffa --- /dev/null +++ b/kamon-system-metrics/src/main/scala/kamon/metrics/ThreadMetrics.scala @@ -0,0 +1,85 @@ +/* + * ========================================================================================= + * 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.metrics + +import java.lang.management.ManagementFactory + +import akka.actor.ActorSystem +import com.typesafe.config.Config +import kamon.metric._ +import kamon.metric.instrument.{ Gauge, Histogram } + +case class ThreadMetrics(name: String) extends MetricGroupIdentity { + val category = ThreadMetrics +} + +object ThreadMetrics extends MetricGroupCategory { + val name = "thread" + + case object Damon extends MetricIdentity { val name = "daemon-count" } + case object Count extends MetricIdentity { val name = "count" } + case object Peak extends MetricIdentity { val name = "peak-count" } + + case class ThreadMetricRecorder(daemon: Gauge, count: Gauge, peak: Gauge) + extends MetricGroupRecorder { + + def collect(context: CollectionContext): MetricGroupSnapshot = { + ThreadMetricSnapshot(daemon.collect(context), count.collect(context), peak.collect(context)) + } + + def cleanup: Unit = {} + } + + case class ThreadMetricSnapshot(daemon: Histogram.Snapshot, count: Histogram.Snapshot, peak: Histogram.Snapshot) + extends MetricGroupSnapshot { + + type GroupSnapshotType = ThreadMetricSnapshot + + def merge(that: GroupSnapshotType, context: CollectionContext): GroupSnapshotType = { + ThreadMetricSnapshot(daemon.merge(that.daemon, context), count.merge(that.count, context), peak.merge(that.peak, context)) + } + + lazy val metrics: Map[MetricIdentity, MetricSnapshot] = Map( + Damon -> daemon, + Count -> count, + Peak -> peak) + } + + val Factory = ThreadMetricGroupFactory +} + +case object ThreadMetricGroupFactory extends MetricGroupFactory { + + import ThreadMetrics._ + + def threads = ManagementFactory.getThreadMXBean + + type GroupRecorder = ThreadMetricRecorder + + def create(config: Config, system: ActorSystem): GroupRecorder = { + val settings = config.getConfig("precision.jvm.thread") + + val daemonThreadConfig = settings.getConfig("daemon") + val countThreadsConfig = settings.getConfig("count") + val peakThreadsConfig = settings.getConfig("peak") + + new ThreadMetricRecorder( + Gauge.fromConfig(daemonThreadConfig, system)(() ⇒ threads.getDaemonThreadCount.toLong), + Gauge.fromConfig(countThreadsConfig, system)(() ⇒ threads.getThreadCount.toLong), + Gauge.fromConfig(peakThreadsConfig, system)(() ⇒ threads.getPeakThreadCount.toLong)) + } +} + diff --git a/kamon-system-metrics/src/main/scala/kamon/system/GcMetricsCollector.scala b/kamon-system-metrics/src/main/scala/kamon/system/GcMetricsCollector.scala new file mode 100644 index 00000000..ae2f50cf --- /dev/null +++ b/kamon-system-metrics/src/main/scala/kamon/system/GcMetricsCollector.scala @@ -0,0 +1,77 @@ +/* + * ========================================================================================= + * 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.system + +import java.lang.management.GarbageCollectorMXBean + +import akka.actor.{ Actor, Props } +import kamon.metrics.GCMetrics.GCMetricRecorder + +import scala.concurrent.duration.FiniteDuration + +class GcMetricsCollector(collectInterval: FiniteDuration, recorder: Option[GCMetricRecorder], extractor: GcMetricExtractor) extends Actor { + import kamon.system.GcMetricsCollector._ + + val collectSchedule = context.system.scheduler.schedule(collectInterval, collectInterval, self, Collect)(SystemMetrics(context.system).dispatcher) + + def receive: Receive = { + case Collect ⇒ collectMetrics() + } + + override def postStop() = collectSchedule.cancel() + + def collectMetrics(): Unit = recorder.map(recordGc) + + private def recordGc(gcr: GCMetricRecorder) = { + val gcMeasure = extractor.extract() + + gcr.count.record(gcMeasure.collectionCount) + gcr.time.record(gcMeasure.collectionTime) + } +} + +object GcMetricsCollector { + case object Collect + + def props(collectInterval: FiniteDuration, recorder: Option[GCMetricRecorder], extractor: GcMetricExtractor): Props = Props(classOf[GcMetricsCollector], collectInterval, recorder, extractor) +} + +case class GcMeasure(collectionCount: Long, collectionTime: Long) + +case class GcMetricExtractor(gc: GarbageCollectorMXBean) { + var previousGcCount = 0L + var previousGcTime = 0L + + def extract(): GcMeasure = { + var diffCollectionCount = 0L + var diffCollectionTime = 0L + + val collectionCount = gc.getCollectionCount + val collectionTime = gc.getCollectionTime + + if (collectionCount > 0) + diffCollectionCount = collectionCount - previousGcCount + + if (collectionTime > 0) + diffCollectionTime = collectionTime - previousGcTime + + previousGcCount = collectionCount + previousGcTime = collectionTime + + GcMeasure(diffCollectionCount, diffCollectionTime) + } +}
\ No newline at end of file diff --git a/kamon-system-metrics/src/main/scala/kamon/system/SystemMetrics.scala b/kamon-system-metrics/src/main/scala/kamon/system/SystemMetrics.scala index 62ffdb33..cb3e2695 100644 --- a/kamon-system-metrics/src/main/scala/kamon/system/SystemMetrics.scala +++ b/kamon-system-metrics/src/main/scala/kamon/system/SystemMetrics.scala @@ -16,19 +16,16 @@ package kamon.system import java.lang.management.ManagementFactory - import akka.actor._ import akka.event.Logging import kamon.Kamon import kamon.metric.Metrics import kamon.metrics._ - import scala.collection.JavaConverters._ import scala.concurrent.duration._ object SystemMetrics extends ExtensionId[SystemMetricsExtension] with ExtensionIdProvider { override def lookup(): ExtensionId[_ <: Extension] = SystemMetrics - override def createExtension(system: ExtendedActorSystem): SystemMetricsExtension = new SystemMetricsExtension(system) } @@ -38,14 +35,25 @@ class SystemMetricsExtension(private val system: ExtendedActorSystem) extends Ka val log = Logging(system, classOf[SystemMetricsExtension]) log.info(s"Starting the Kamon(SystemMetrics) extension") + val config = system.settings.config.getConfig("kamon.system-metrics") + val dispatcher = system.dispatchers.lookup(config.getString("dispatcher")) + val sigarFolder = system.settings.config.getString("kamon.sigar.folder") val systemMetricsExtension = Kamon(Metrics)(system) + //System Metrics + system.actorOf(SystemMetricsCollector.props(1 second), "system-metrics-collector") + //JVM Metrics systemMetricsExtension.register(HeapMetrics(Heap), HeapMetrics.Factory) - garbageCollectors.map { gc ⇒ systemMetricsExtension.register(GCMetrics(gc.getName), GCMetrics.Factory(gc)) } + systemMetricsExtension.register(NonHeapMetrics(NonHeap), NonHeapMetrics.Factory) + systemMetricsExtension.register(ClassLoadingMetrics(Classes), ClassLoadingMetrics.Factory) + systemMetricsExtension.register(ThreadMetrics(Threads), ThreadMetrics.Factory) - //System Metrics - system.actorOf(SystemMetricsCollector.props(1 second), "system-metrics-collector") + garbageCollectors.map { gc ⇒ + val gcName = sanitize(gc.getName) + val recorder = systemMetricsExtension.register(GCMetrics(gcName), GCMetrics.Factory(gc)) + system.actorOf(GcMetricsCollector.props(1 second, recorder, GcMetricExtractor(gc)), s"$gcName-collector") + } } object SystemMetricsExtension { @@ -54,11 +62,17 @@ object SystemMetricsExtension { val Network = "network" val Memory = "memory" val Heap = "heap" + val NonHeap = "non-heap" + val Classes = "classes" + val Threads = "thread" val ContextSwitches = "context-switches" + val Disk = "disk" + val LoadAverage = "load-average" - def toKB(value: Long): Long = (value / 1024) - def toMB(value: Long): Long = (value / 1024 / 1024) + def toKB(value: Long): Long = value / 1024 + def toMB(value: Long): Long = value / 1024 / 1024 def toLong(value: Double): Long = math round (value * 100L) + def sanitize(str: String): String = str.replaceAll("""[^\w]""", "-") val garbageCollectors = ManagementFactory.getGarbageCollectorMXBeans.asScala.filter(_.isValid) } diff --git a/kamon-system-metrics/src/main/scala/kamon/system/SystemMetricsCollector.scala b/kamon-system-metrics/src/main/scala/kamon/system/SystemMetricsCollector.scala index c200091e..4391240a 100644 --- a/kamon-system-metrics/src/main/scala/kamon/system/SystemMetricsCollector.scala +++ b/kamon-system-metrics/src/main/scala/kamon/system/SystemMetricsCollector.scala @@ -22,16 +22,19 @@ import kamon.Kamon import kamon.metric.Metrics import kamon.metrics.CPUMetrics.CPUMetricRecorder import kamon.metrics.ContextSwitchesMetrics.ContextSwitchesMetricsRecorder +import kamon.metrics.DiskMetrics.DiskMetricsRecorder +import kamon.metrics.LoadAverageMetrics.LoadAverageMetricsRecorder import kamon.metrics.MemoryMetrics.MemoryMetricRecorder import kamon.metrics.NetworkMetrics.NetworkMetricRecorder import kamon.metrics.ProcessCPUMetrics.ProcessCPUMetricsRecorder import kamon.metrics._ import kamon.sigar.SigarProvisioner -import org.hyperic.sigar.{ Sigar, Mem, NetInterfaceStat, SigarProxy } +import org.hyperic.sigar._ +import scala.collection.concurrent.TrieMap import scala.concurrent.duration.FiniteDuration import scala.io.Source -import scala.util.control.NonFatal +import scala.collection.mutable class SystemMetricsCollector(collectInterval: FiniteDuration) extends Actor with ActorLogging with SystemMetricsBanner { import kamon.system.SystemMetricsCollector._ @@ -40,16 +43,19 @@ class SystemMetricsCollector(collectInterval: FiniteDuration) extends Actor with lazy val sigar = createSigarInstance def pid = sigar.getPid - val interfaces: Set[String] = sigar.getNetInterfaceList.toSet + val interfaces = sigar.getNetInterfaceList.filterNot(NetworkFilter).toSet + val fileSystems = sigar.getFileSystemList.filter(_.getType == FileSystem.TYPE_LOCAL_DISK).map(_.getDevName).toSet - val systemMetricsExtension = Kamon(Metrics)(context.system) - val collectSchedule = context.system.scheduler.schedule(collectInterval, collectInterval, self, Collect)(context.dispatcher) + val metricExtension = Kamon(Metrics)(context.system) + val collectSchedule = context.system.scheduler.schedule(collectInterval, collectInterval, self, Collect)(SystemMetrics(context.system).dispatcher) - val cpuRecorder = systemMetricsExtension.register(CPUMetrics(CPU), CPUMetrics.Factory) - val processCpuRecorder = systemMetricsExtension.register(ProcessCPUMetrics(ProcessCPU), ProcessCPUMetrics.Factory) - val memoryRecorder = systemMetricsExtension.register(MemoryMetrics(Memory), MemoryMetrics.Factory) - val networkRecorder = systemMetricsExtension.register(NetworkMetrics(Network), NetworkMetrics.Factory) - val contextSwitchesRecorder = systemMetricsExtension.register(ContextSwitchesMetrics(ContextSwitches), ContextSwitchesMetrics.Factory) + val cpuRecorder = metricExtension.register(CPUMetrics(CPU), CPUMetrics.Factory) + val processCpuRecorder = metricExtension.register(ProcessCPUMetrics(ProcessCPU), ProcessCPUMetrics.Factory) + val memoryRecorder = metricExtension.register(MemoryMetrics(Memory), MemoryMetrics.Factory) + val networkRecorder = metricExtension.register(NetworkMetrics(Network), NetworkMetrics.Factory) + val contextSwitchesRecorder = metricExtension.register(ContextSwitchesMetrics(ContextSwitches), ContextSwitchesMetrics.Factory) + val diskRecorder = metricExtension.register(DiskMetrics(Disk), DiskMetrics.Factory) + val loadAverageRecorder = metricExtension.register(LoadAverageMetrics(LoadAverage), LoadAverageMetrics.Factory) def receive: Receive = { case Collect ⇒ collectMetrics() @@ -62,6 +68,8 @@ class SystemMetricsCollector(collectInterval: FiniteDuration) extends Actor with processCpuRecorder.map(recordProcessCpu) memoryRecorder.map(recordMemory) networkRecorder.map(recordNetwork) + diskRecorder.map(recordDisk) + loadAverageRecorder.map(recordLoadAverage) if (OsUtils.isLinux) contextSwitchesRecorder.map(recordContextSwitches) @@ -100,23 +108,60 @@ class SystemMetricsCollector(collectInterval: FiniteDuration) extends Actor with } private def recordNetwork(nr: NetworkMetricRecorder) = { - nr.rxBytes.record(collect(sigar, interfaces)(net ⇒ toKB(net.getRxBytes))) - nr.txBytes.record(collect(sigar, interfaces)(net ⇒ toKB(net.getTxBytes))) - nr.rxErrors.record(collect(sigar, interfaces)(net ⇒ net.getRxErrors)) - nr.txErrors.record(collect(sigar, interfaces)(net ⇒ net.getTxErrors)) - nr.rxDropped.record(collect(sigar, interfaces)(net ⇒ net.getRxDropped)) - nr.txDropped.record(collect(sigar, interfaces)(net ⇒ net.getTxDropped)) - - def collect(sigar: SigarProxy, interfaces: Set[String])(block: NetInterfaceStat ⇒ Long): Long = { - interfaces.foldLeft(0L) { (totalBytes, interface) ⇒ + import Networks._ + nr.rxBytes.record(collect(sigar, interfaces, RxBytes, previousNetworkMetrics)(net ⇒ toKB(net.getRxBytes))) + nr.txBytes.record(collect(sigar, interfaces, TxBytes, previousNetworkMetrics)(net ⇒ toKB(net.getTxBytes))) + nr.rxErrors.record(collect(sigar, interfaces, RxErrors, previousNetworkMetrics)(net ⇒ net.getRxErrors)) + nr.txErrors.record(collect(sigar, interfaces, TxErrors, previousNetworkMetrics)(net ⇒ net.getTxErrors)) + nr.rxDropped.record(collect(sigar, interfaces, RxDropped, previousNetworkMetrics)(net ⇒ net.getRxDropped)) + nr.txDropped.record(collect(sigar, interfaces, TxDropped, previousNetworkMetrics)(net ⇒ net.getTxDropped)) + + def collect(sigar: SigarProxy, interfaces: Set[String], name: String, previousMetrics: TrieMap[String, mutable.Map[String, Long]])(thunk: NetInterfaceStat ⇒ Long): Long = { + interfaces.foldLeft(0L) { (acc, interface) ⇒ { val net = sigar.getNetInterfaceStat(interface) - totalBytes + block(net) + val previous = previousMetrics.getOrElse(interface, mutable.Map.empty[String, Long]) + val current = thunk(net) + val delta = current - previous.getOrElse(name, 0L) + previousMetrics.put(interface, previous += name -> current) + acc + delta } } } } + private def recordDisk(rd: DiskMetricsRecorder) = { + import Disks._ + + rd.reads.record(collect(sigar, fileSystems, Reads, previousDiskMetrics)(disk ⇒ disk.getReads)) + rd.writes.record(collect(sigar, fileSystems, Writes, previousDiskMetrics)(disk ⇒ disk.getWrites)) + rd.queue.record(collect(sigar, fileSystems, Queue, previousDiskMetrics)(disk ⇒ toLong(disk.getQueue))) + rd.serviceTime.record(collect(sigar, fileSystems, Service, previousDiskMetrics)(disk ⇒ toLong(disk.getServiceTime))) + } + + def collect(sigar: SigarProxy, fileSystems: Set[String], name: String, previousMetrics: TrieMap[String, mutable.Map[String, Long]])(thunk: DiskUsage ⇒ Long): Long = { + fileSystems.foldLeft(0L) { (acc, fileSystem) ⇒ + { + val disk = sigar.getDiskUsage(fileSystem) + val previous = previousMetrics.getOrElse(fileSystem, mutable.Map.empty[String, Long]) + val value = thunk(disk) + val current = if (value == Sigar.FIELD_NOTIMPL) 0L else value + val delta = current - previous.getOrElse(name, 0L) + previousMetrics.put(fileSystem, previous += name -> current) + acc + delta + } + } + } + + private def recordLoadAverage(lar: LoadAverageMetricsRecorder) = { + val loadAverage = sigar.getLoadAverage + val (one, five, fifteen) = (loadAverage(0), loadAverage(1), loadAverage(2)) + + lar.one.record(toLong(one)) + lar.five.record(toLong(five)) + lar.fifteen.record(toLong(fifteen)) + } + private def recordContextSwitches(rcs: ContextSwitchesMetricsRecorder) = { def contextSwitchesByProcess(pid: Long): (Long, Long) = { val filename = s"/proc/$pid/status" @@ -167,12 +212,11 @@ class SystemMetricsCollector(collectInterval: FiniteDuration) extends Actor with } def provisionSigarLibrary: Unit = { - val folder = context.system.settings.config.getString("kamon.sigar.folder") + val folder = SystemMetrics(context.system).sigarFolder SigarProvisioner.provision(new File(folder)) } def createSigarInstance: SigarProxy = { - // 1) Assume that library is already provisioned. try { return verifiedSigarInstance @@ -193,6 +237,25 @@ class SystemMetricsCollector(collectInterval: FiniteDuration) extends Actor with } object SystemMetricsCollector { + val NetworkFilter = Set("lo") + val previousDiskMetrics = TrieMap[String, mutable.Map[String, Long]]() + val previousNetworkMetrics = TrieMap[String, mutable.Map[String, Long]]() + + object Networks { + val RxBytes = "rxBytes" + val TxBytes = "txBytes" + val RxErrors = "rxErrors" + val TxErrors = "txErrors" + val RxDropped = "rxDropped" + val TxDropped = "txDropped" + } + + object Disks { + val Reads = "reads" + val Writes = "writes" + val Queue = "queue" + val Service = "service" + } case object Collect object OsUtils { diff --git a/kamon-system-metrics/src/test/scala/kamon/metrics/SystemMetricsSpec.scala b/kamon-system-metrics/src/test/scala/kamon/metrics/SystemMetricsSpec.scala index 5f0c4a10..c9530160 100644 --- a/kamon-system-metrics/src/test/scala/kamon/metrics/SystemMetricsSpec.scala +++ b/kamon-system-metrics/src/test/scala/kamon/metrics/SystemMetricsSpec.scala @@ -21,12 +21,17 @@ import com.typesafe.config.ConfigFactory import kamon.Kamon import kamon.metric.Subscriptions.TickMetricSnapshot import kamon.metrics.CPUMetrics.CPUMetricSnapshot +import kamon.metrics.ClassLoadingMetrics.ClassLoadingMetricSnapshot import kamon.metrics.ContextSwitchesMetrics.ContextSwitchesMetricsSnapshot +import kamon.metrics.DiskMetrics.DiskMetricsSnapshot import kamon.metrics.GCMetrics.GCMetricSnapshot import kamon.metrics.HeapMetrics.HeapMetricSnapshot +import kamon.metrics.LoadAverageMetrics.LoadAverageMetricsSnapshot import kamon.metrics.MemoryMetrics.MemoryMetricSnapshot import kamon.metrics.NetworkMetrics.NetworkMetricSnapshot +import kamon.metrics.NonHeapMetrics.NonHeapMetricSnapshot import kamon.metrics.ProcessCPUMetrics.ProcessCPUMetricsSnapshot +import kamon.metrics.ThreadMetrics.ThreadMetricSnapshot import kamon.metrics._ import kamon.system.SystemMetricsExtension import org.scalatest.{ Matchers, WordSpecLike } @@ -41,135 +46,8 @@ class SystemMetricsSpec extends TestKitBase with WordSpecLike with Matchers with |} | |kamon.metrics { - | | disable-aspectj-weaver-missing-error = true - | | tick-interval = 1 second - | - | system { - | cpu { - | user { - | highest-trackable-value = 999999999 - | significant-value-digits = 2 - | } - | system { - | highest-trackable-value = 999999999 - | significant-value-digits = 2 - | } - | wait { - | highest-trackable-value = 999999999 - | significant-value-digits = 2 - | } - | idle { - | highest-trackable-value = 999999999 - | significant-value-digits = 2 - | } - | stolen { - | highest-trackable-value = 999999999 - | significant-value-digits = 2 - | } - | } - | process-cpu { - | user { - | highest-trackable-value = 999999999 - | significant-value-digits = 2 - | } - | system { - | highest-trackable-value = 999999999 - | significant-value-digits = 2 - | } - | } - | memory { - | used { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | free { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | buffer { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | cache { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | swap-used { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | swap-free { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | } - | context-switches { - | per-process-voluntary { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | per-process-non-voluntary { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | global { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | } - | network { - | rx-bytes { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | tx-bytes { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | rx-errors { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | tx-errors { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | rx-dropped { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | tx-dropped { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | } - | heap { - | used { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | max { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | committed { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | } - | gc { - | count { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | time { - | highest-trackable-value = 3600000000000 - | significant-value-digits = 2 - | } - | } - | } |} """.stripMargin)) @@ -190,8 +68,8 @@ class SystemMetricsSpec extends TestKitBase with WordSpecLike with Matchers with val metricsListener = subscribeToMetrics() val GCMetrics = expectGCMetrics(metricsListener, 3 seconds) - GCMetrics.count.max should be > 0L - GCMetrics.time.max should be > 0L + GCMetrics.count.max should be >= 0L + GCMetrics.time.max should be >= 0L } } @@ -206,6 +84,62 @@ class SystemMetricsSpec extends TestKitBase with WordSpecLike with Matchers with } } + "the Kamon Non-Heap Metrics" should { + "record used, max, commited metrics" in new NonHeapMetricsListenerFixture { + val metricsListener = subscribeToMetrics() + + val NonHeapMetrics = expectNonHeapMetrics(metricsListener, 3 seconds) + NonHeapMetrics.used.max should be >= 0L + NonHeapMetrics.max.max should be >= 0L + NonHeapMetrics.committed.max should be >= 0L + } + } + + "the Kamon Thread Metrics" should { + "record daemon, count, peak metrics" in new ThreadMetricsListenerFixture { + val metricsListener = subscribeToMetrics() + + val ThreadMetrics = expectThreadMetrics(metricsListener, 3 seconds) + ThreadMetrics.daemon.max should be >= 0L + ThreadMetrics.count.max should be >= 0L + ThreadMetrics.peak.max should be >= 0L + } + } + + "the Kamon ClassLoading Metrics" should { + "record loaded, unloaded, current metrics" in new ClassLoadingMetricsListenerFixture { + val metricsListener = subscribeToMetrics() + + val ClassLoadingMetrics = expectClassLoadingMetrics(metricsListener, 3 seconds) + ClassLoadingMetrics.loaded.max should be >= 0L + ClassLoadingMetrics.unloaded.max should be >= 0L + ClassLoadingMetrics.current.max should be >= 0L + } + } + + "the Kamon Disk Metrics" should { + "record reads, writes, queue, service time metrics" in new DiskMetricsListenerFixture { + val metricsListener = subscribeToMetrics() + + val DiskMetrics = expectDiskMetrics(metricsListener, 3 seconds) + DiskMetrics.reads.max should be >= 0L + DiskMetrics.writes.max should be >= 0L + DiskMetrics.queue.max should be >= 0L + DiskMetrics.serviceTime.max should be >= 0L + } + } + + "the Kamon Load Average Metrics" should { + "record 1 minute, 5 minutes and 15 minutes metrics" in new LoadAverageMetricsListenerFixture { + val metricsListener = subscribeToMetrics() + + val LoadAverageMetrics = expectLoadAverageMetrics(metricsListener, 3 seconds) + LoadAverageMetrics.one.max should be >= 0L + LoadAverageMetrics.five.max should be >= 0L + LoadAverageMetrics.fifteen.max should be >= 0L + } + } + "the Kamon Memory Metrics" should { "record used, free, buffer, cache, swap used, swap free metrics" in new MemoryMetricsListenerFixture { val metricsListener = subscribeToMetrics() @@ -279,7 +213,7 @@ class SystemMetricsSpec extends TestKitBase with WordSpecLike with Matchers with listener.expectMsgType[TickMetricSnapshot] } - val gcMetricsOption = tickSnapshot.metrics.get(GCMetrics(SystemMetricsExtension.garbageCollectors(0).getName)) + val gcMetricsOption = tickSnapshot.metrics.get(GCMetrics(SystemMetricsExtension.garbageCollectors(0).getName.replaceAll("""[^\w]""", "-"))) gcMetricsOption should not be empty gcMetricsOption.get.asInstanceOf[GCMetricSnapshot] } @@ -313,6 +247,101 @@ class SystemMetricsSpec extends TestKitBase with WordSpecLike with Matchers with } } + trait NonHeapMetricsListenerFixture { + def subscribeToMetrics(): TestProbe = { + val metricsListener = TestProbe() + Kamon(Metrics).subscribe(NonHeapMetrics, "*", metricsListener.ref, permanently = true) + // Wait for one empty snapshot before proceeding to the test. + metricsListener.expectMsgType[TickMetricSnapshot] + metricsListener + } + } + + def expectNonHeapMetrics(listener: TestProbe, waitTime: FiniteDuration): NonHeapMetricSnapshot = { + val tickSnapshot = within(waitTime) { + listener.expectMsgType[TickMetricSnapshot] + } + val nonHeapMetricsOption = tickSnapshot.metrics.get(NonHeapMetrics(SystemMetricsExtension.NonHeap)) + nonHeapMetricsOption should not be empty + nonHeapMetricsOption.get.asInstanceOf[NonHeapMetricSnapshot] + } + + trait ThreadMetricsListenerFixture { + def subscribeToMetrics(): TestProbe = { + val metricsListener = TestProbe() + Kamon(Metrics).subscribe(ThreadMetrics, "*", metricsListener.ref, permanently = true) + // Wait for one empty snapshot before proceeding to the test. + metricsListener.expectMsgType[TickMetricSnapshot] + metricsListener + } + } + + def expectThreadMetrics(listener: TestProbe, waitTime: FiniteDuration): ThreadMetricSnapshot = { + val tickSnapshot = within(waitTime) { + listener.expectMsgType[TickMetricSnapshot] + } + val threadMetricsOption = tickSnapshot.metrics.get(ThreadMetrics(SystemMetricsExtension.Threads)) + threadMetricsOption should not be empty + threadMetricsOption.get.asInstanceOf[ThreadMetricSnapshot] + } + + trait ClassLoadingMetricsListenerFixture { + def subscribeToMetrics(): TestProbe = { + val metricsListener = TestProbe() + Kamon(Metrics).subscribe(ClassLoadingMetrics, "*", metricsListener.ref, permanently = true) + // Wait for one empty snapshot before proceeding to the test. + metricsListener.expectMsgType[TickMetricSnapshot] + metricsListener + } + } + + def expectClassLoadingMetrics(listener: TestProbe, waitTime: FiniteDuration): ClassLoadingMetricSnapshot = { + val tickSnapshot = within(waitTime) { + listener.expectMsgType[TickMetricSnapshot] + } + val classLoadingMetricsOption = tickSnapshot.metrics.get(ClassLoadingMetrics(SystemMetricsExtension.Classes)) + classLoadingMetricsOption should not be empty + classLoadingMetricsOption.get.asInstanceOf[ClassLoadingMetricSnapshot] + } + + trait DiskMetricsListenerFixture { + def subscribeToMetrics(): TestProbe = { + val metricsListener = TestProbe() + Kamon(Metrics).subscribe(DiskMetrics, "*", metricsListener.ref, permanently = true) + // Wait for one empty snapshot before proceeding to the test. + metricsListener.expectMsgType[TickMetricSnapshot] + metricsListener + } + } + + def expectDiskMetrics(listener: TestProbe, waitTime: FiniteDuration): DiskMetricsSnapshot = { + val tickSnapshot = within(waitTime) { + listener.expectMsgType[TickMetricSnapshot] + } + val diskMetricsOption = tickSnapshot.metrics.get(DiskMetrics(SystemMetricsExtension.Disk)) + diskMetricsOption should not be empty + diskMetricsOption.get.asInstanceOf[DiskMetricsSnapshot] + } + + trait LoadAverageMetricsListenerFixture { + def subscribeToMetrics(): TestProbe = { + val metricsListener = TestProbe() + Kamon(Metrics).subscribe(LoadAverageMetrics, "*", metricsListener.ref, permanently = true) + // Wait for one empty snapshot before proceeding to the test. + metricsListener.expectMsgType[TickMetricSnapshot] + metricsListener + } + } + + def expectLoadAverageMetrics(listener: TestProbe, waitTime: FiniteDuration): LoadAverageMetricsSnapshot = { + val tickSnapshot = within(waitTime) { + listener.expectMsgType[TickMetricSnapshot] + } + val loadAverageMetricsOption = tickSnapshot.metrics.get(LoadAverageMetrics(SystemMetricsExtension.LoadAverage)) + loadAverageMetricsOption should not be empty + loadAverageMetricsOption.get.asInstanceOf[LoadAverageMetricsSnapshot] + } + def expectMemoryMetrics(listener: TestProbe, waitTime: FiniteDuration): MemoryMetricSnapshot = { val tickSnapshot = within(waitTime) { listener.expectMsgType[TickMetricSnapshot] |