Replace dropwizard-metrics with micrometer 1.15.4

pom.xml

@@ -47,7 +47,7 @@
         <jackson.version>3.1.5</jackson.version>
         <junit.jupiter.version>6.1.2</junit.jupiter.version>
         <logback.version>1.5.38</logback.version>
-        <metrics.version>4.2.38</metrics.version>
+        <micrometer.version>1.15.4</micrometer.version>
         <netty.version>4.2.16.Final</netty.version>
         <roaring.bitmap.version>1.6.14</roaring.bitmap.version>
         <scala.version>3.3.8</scala.version> <!-- Scala LTS -->

toolbox/pom.xml

@@ -33,15 +33,15 @@
         </dependency>
 
         <dependency>
-            <groupId>io.dropwizard.metrics</groupId>
-            <artifactId>metrics-core</artifactId>
-            <version>${metrics.version}</version>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-core</artifactId>
+            <version>${micrometer.version}</version>
         </dependency>
 
         <dependency>
-            <groupId>io.dropwizard.metrics</groupId>
-            <artifactId>metrics-jmx</artifactId>
-            <version>${metrics.version}</version>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-registry-jmx</artifactId>
+            <version>${micrometer.version}</version>
         </dependency>
 
         <dependency>

toolbox/src/main/scala/org/finos/toolbox/jmx/MetricsProvider.scala

@@ -1,28 +1,29 @@
 package org.finos.toolbox.jmx
 
-import com.codahale.metrics._
-import com.codahale.metrics.jmx._
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry
+import io.micrometer.core.instrument.{Clock, Counter, DistributionSummary, MeterRegistry}
+import io.micrometer.jmx.{JmxConfig, JmxMeterRegistry}
 
 
 trait MetricsProvider{
-  def meter(name: String): Meter
+  def meter(name: String): Counter
   def counter(name: String): Counter
-  def histogram(name: String): Histogram
+  def histogram(name: String): DistributionSummary
 }
 
 class MetricsProviderImpl() extends MetricsProvider{
-  final val metrics = new MetricRegistry()
 
-  @volatile var reporter: JmxReporter = null
+  final val metrics: MeterRegistry =
+    if (JmxInfra.isJmxEnabled) new JmxMeterRegistry(JmxConfig.DEFAULT, Clock.SYSTEM)
+    else new SimpleMeterRegistry()
 
-  override def histogram(name: String): Histogram = metrics.histogram(name)
-  override def meter(name: String): Meter = metrics.meter(name)
-  override def counter(name: String): Counter = metrics.counter(name)
+  override def histogram(name: String): DistributionSummary =
+    DistributionSummary.builder(name)
+      .publishPercentiles(0.5, 0.75, 0.99, 0.999)
+      .register(metrics)
 
-  if(JmxInfra.isJmxEnabled){
-    reporter = JmxReporter.forRegistry(metrics).build()
-    reporter.start()
-  }
+  override def meter(name: String): Counter = metrics.counter(name)
 
-}
+  override def counter(name: String): Counter = metrics.counter(name)
 
+}

toolbox/src/test/scala/org/finos/toolbox/metrics/MetricsProviderTest.scala

@@ -14,16 +14,17 @@ class MetricsProviderTest extends AnyFeatureSpec with Matchers {
 
       val histogram = metrics.histogram("foo.bar")
 
-      histogram.update(1L)
+      histogram.record(1.0)
 
-      histogram.update(2L)
+      histogram.record(2.0)
 
-      histogram.update(3L)
+      histogram.record(3.0)
 
-      val snapshot = histogram.getSnapshot
+      val snapshot = histogram.takeSnapshot()
 
-      snapshot.getMax should equal(3L)
-      snapshot.getMin should equal(1L)
+      snapshot.max() should equal(3.0)
+      snapshot.count() should equal(3L)
+      snapshot.mean() should equal(2.0)
 
     }
 

vuu/src/main/scala/org/finos/vuu/core/module/metrics/MetricsGroupByProvider.scala

@@ -1,7 +1,8 @@
 package org.finos.vuu.core.module.metrics
 
-import com.codahale.metrics.Histogram
 import com.typesafe.scalalogging.StrictLogging
+import io.micrometer.core.instrument.DistributionSummary
+import org.finos.vuu.core.module.metrics.MicrometerMetrics.percentileValue
 import org.finos.vuu.core.table.{DataTable, RowWithData}
 import org.finos.vuu.provider.Provider
 import org.finos.vuu.viewport.ViewPortContainer
@@ -32,7 +33,7 @@ class MetricsGroupByProvider(table: DataTable, viewPortContainer: ViewPortContai
 
   override val lifecycleId: String = "metricsGroupByProvider"
 
-  def buildColumnsForHistogram(prefix: String, hist: Histogram): Map[String, Any] = {
+  def buildColumnsForHistogram(prefix: String, hist: DistributionSummary): Map[String, Any] = {
     /*
     val settree_mean = "settree_mean"
     val settree_samples = "settree_samples"
@@ -45,11 +46,11 @@ class MetricsGroupByProvider(table: DataTable, viewPortContainer: ViewPortContai
       Map()
     } else {
 
-      val snapshot = hist.getSnapshot
+      val snapshot = hist.takeSnapshot()
 
-      Map(prefix + "_mean" -> snapshot.getMean, prefix + "_samples" -> snapshot.getValues.length,
-        prefix + "_50_perc" -> snapshot.getMedian, prefix + "_75_perc" -> snapshot.get75thPercentile(),
-        prefix + "_99_perc" -> snapshot.get99thPercentile(), prefix + "_99_9_perc" -> snapshot.get999thPercentile()
+      Map(prefix + "_mean" -> snapshot.mean(), prefix + "_samples" -> snapshot.count(),
+        prefix + "_50_perc" -> percentileValue(snapshot, 0.5), prefix + "_75_perc" -> percentileValue(snapshot, 0.75),
+        prefix + "_99_perc" -> percentileValue(snapshot, 0.99), prefix + "_99_9_perc" -> percentileValue(snapshot, 0.999)
       )
     }
   }

vuu/src/main/scala/org/finos/vuu/core/module/metrics/MetricsTableProvider.scala

@@ -9,6 +9,8 @@ import org.finos.vuu.core.table.{DataTable, RowWithData, TableContainer}
 import org.finos.vuu.provider.Provider
 import org.finos.vuu.viewport.ViewPortTable
 
+import java.util.concurrent.ConcurrentHashMap
+
 class MetricsTableProvider(table: DataTable, tableContainer: TableContainer)(implicit clock: Clock, lifecycleContainer: LifecycleContainer,
                                                                              metrics: MetricsProvider) extends Provider with StrictLogging {
 
@@ -38,16 +40,19 @@ class MetricsTableProvider(table: DataTable, tableContainer: TableContainer)(imp
     }
   }
 
+  private val updateRates = new ConcurrentHashMap[String, CounterRatePerSecond]()
+
   private def getMetricsData(vpTable: ViewPortTable): Map[String, Any] = {
     val counter = metrics.counter(vpTable.table + ".processUpdates.Counter")
     val size = tableContainer.getTable(vpTable.table).size()
-    val meter = metrics.meter(vpTable.table + ".processUpdates.Meter")
+    val updateRate = updateRates.computeIfAbsent(vpTable.table,
+      t => new CounterRatePerSecond(metrics.meter(t + ".processUpdates.Meter")))
 
     Map(
       "table" -> (vpTable.module + "-" + vpTable.table),
-      "updateCount" -> counter.getCount,
+      "updateCount" -> counter.count().toLong,
       "size" -> size,
-      "updatesPerSecond" -> meter.getOneMinuteRate
+      "updatesPerSecond" -> updateRate.perSecond()
     )
   }
 }

vuu/src/main/scala/org/finos/vuu/core/module/metrics/MetricsViewPortParallelismProvider.scala

@@ -18,15 +18,15 @@ class MetricsViewPortParallelismProvider(table: DataTable, viewPortContainer: Vi
 
   lifecycleContainer(this).dependsOn(runner)
 
-  private def runOnce(): Unit = {
-    val flatWorkMeter = viewPortContainer.totalFlatWorkHistogram
-    val treeWorkMeter = viewPortContainer.totalTreeWorkHistogram
+  private val treeWorkRate = new CounterRatePerSecond(viewPortContainer.totalTreeWorkHistogram)
+  private val flatWorkRate = new CounterRatePerSecond(viewPortContainer.totalFlatWorkHistogram)
 
-    val treeFiveMinRate = treeWorkMeter.getOneMinuteRate // this is events/sec over x minutes
-    val flatFiveMinRate = flatWorkMeter.getOneMinuteRate
+  private def runOnce(): Unit = {
+    val treeRate = treeWorkRate.perSecond() // this is ms of work done per second
+    val flatRate = flatWorkRate.perSecond()
 
-    val tree = Map("type" -> "tree", ViewPortParallelism.work_ms_in_1m -> treeFiveMinRate, ViewPortParallelism.work_par_ratio -> (treeFiveMinRate / 1000))
-    val flat = Map("type" -> "flat", ViewPortParallelism.work_ms_in_1m -> flatFiveMinRate, ViewPortParallelism.work_par_ratio -> (flatFiveMinRate / 1000))
+    val tree = Map("type" -> "tree", ViewPortParallelism.work_ms_in_1m -> treeRate, ViewPortParallelism.work_par_ratio -> (treeRate / 1000))
+    val flat = Map("type" -> "flat", ViewPortParallelism.work_ms_in_1m -> flatRate, ViewPortParallelism.work_par_ratio -> (flatRate / 1000))
 
     table.processUpdate("tree", RowWithData("tree", tree))
     table.processUpdate("flat", RowWithData("flat", flat))

vuu/src/main/scala/org/finos/vuu/core/module/metrics/MetricsViewPortProvider.scala

@@ -5,6 +5,7 @@ import org.finos.toolbox.jmx.MetricsProvider
 import org.finos.toolbox.lifecycle.LifecycleContainer
 import org.finos.toolbox.thread.LifeCycleRunner
 import org.finos.toolbox.time.Clock
+import org.finos.vuu.core.module.metrics.MicrometerMetrics.percentileValue
 import org.finos.vuu.core.table.datatype.{Scale, ScaledDecimal}
 import org.finos.vuu.core.table.{DataTable, RowWithData}
 import org.finos.vuu.provider.Provider
@@ -44,20 +45,20 @@ class MetricsViewPortProvider(table: DataTable, viewPortContainer: ViewPortConta
       val toDelete = viewportIds.filterNot(kv => mapOfHistograms.contains(kv._1)).toMap
 
       mapOfHistograms.foreach({ case (key, histogram) => {
-        val snapshot = histogram.getSnapshot
+        val snapshot = histogram.takeSnapshot()
         val vp = viewPortContainer.getViewPorts.find(f => f.id == key).orNull
         if (vp != null) {
           val upMap = Map(
             "id" -> key,
             "table" -> vp.table.name,
-            "mean" -> ScaledDecimal(snapshot.getMean, Scale.Two),
-            "max" -> snapshot.getMax,
+            "mean" -> ScaledDecimal(snapshot.mean(), Scale.Two),
+            "max" -> snapshot.max(),
             "structureHash" -> vp.getStructuralHashCode(),
             "updateCount" -> vp.getTableUpdateCount(),
             "keyBuildCount" -> vp.keyBuildCount,
-            "75Perc" -> ScaledDecimal(snapshot.get75thPercentile(), Scale.Four),
-            "99Perc" -> ScaledDecimal(snapshot.get99thPercentile(), Scale.Six),
-            "99_9Perc" -> ScaledDecimal(snapshot.get999thPercentile(), Scale.Eight)
+            "75Perc" -> ScaledDecimal(percentileValue(snapshot, 0.75), Scale.Four),
+            "99Perc" -> ScaledDecimal(percentileValue(snapshot, 0.99), Scale.Six),
+            "99_9Perc" -> ScaledDecimal(percentileValue(snapshot, 0.999), Scale.Eight)
           )
           table.processUpdate(key, RowWithData(key, upMap))
         } else {

vuu/src/main/scala/org/finos/vuu/core/module/metrics/MicrometerMetrics.scala

@@ -0,0 +1,41 @@
+package org.finos.vuu.core.module.metrics
+
+import io.micrometer.core.instrument.Counter
+import io.micrometer.core.instrument.distribution.HistogramSnapshot
+import org.finos.toolbox.time.Clock
+
+/**
+ * Computes a per-second rate from a monotonically increasing counter by sampling
+ * it on each call. Dropwizard meters exposed a one-minute exponentially weighted
+ * rate; Micrometer counters leave rate computation to the consumer, so the
+ * providers that used to read getOneMinuteRate sample the counter on their run
+ * cycle instead.
+ */
+class CounterRatePerSecond(counter: Counter)(implicit clock: Clock) {
+
+  private var lastTimeMillis: Long = clock.now()
+  private var lastCount: Double = counter.count()
+
+  def perSecond(): Double = {
+    val nowMillis = clock.now()
+    val count = counter.count()
+    val elapsedMillis = nowMillis - lastTimeMillis
+    val rate = if (elapsedMillis <= 0) 0.0 else (count - lastCount) * 1000.0 / elapsedMillis
+    lastTimeMillis = nowMillis
+    lastCount = count
+    rate
+  }
+}
+
+object MicrometerMetrics {
+
+  /**
+   * Returns the value recorded for a percentile the summary was configured to
+   * publish (see MetricsProviderImpl), or 0.0 when nothing has been recorded.
+   */
+  def percentileValue(snapshot: HistogramSnapshot, percentile: Double): Double =
+    snapshot.percentileValues()
+      .find(pv => math.abs(pv.percentile() - percentile) < 1e-6)
+      .map(_.value())
+      .getOrElse(0.0)
+}

vuu/src/main/scala/org/finos/vuu/core/table/AutoSubscribeTable.scala

@@ -16,10 +16,10 @@ class AutoSubscribeTable(tableDef: TableDef, joinProvider: JoinTableProvider)(im
 
   def tryAndSubscribe(key: String): Unit = {
 
-    totalSubscribe.inc()
+    totalSubscribe.increment()
 
     if (!subscriptionKeys.contains(key)) {
-      onTrySubscribe.inc()
+      onTrySubscribe.increment()
 
       subscriptionKeys.add(key)
       getProvider.subscribe(key)

vuu/src/main/scala/org/finos/vuu/core/table/InMemDataTable.scala

@@ -386,15 +386,15 @@ class InMemDataTable(val tableDef: TableDef, val joinProvider: JoinTableProvider
   }
 
   def sendToJoinSink(rowKey: String, rowData: RowData): Unit = {
-    eventIntoJoiner.inc()
+    eventIntoJoiner.increment()
     if (joinProvider.hasJoins(this.tableDef.name)) {
       val event = toEvent(rowKey, rowData)
       joinProvider.sendEvent(this.tableDef.name, event)
     }
   }
 
   def sendDeleteToJoinSink(rowKey: String, rowData: RowData): Unit = {
-    eventIntoJoiner.inc()
+    eventIntoJoiner.increment()
     if (joinProvider.hasJoins(this.tableDef.name)) {
       val event = toDeleteEvent(rowKey, rowData)
       joinProvider.sendEvent(this.tableDef.name, event)
@@ -403,9 +403,9 @@ class InMemDataTable(val tableDef: TableDef, val joinProvider: JoinTableProvider
 
   def processUpdate(rowKey: String, rowData: RowData): Unit = {
 
-    onUpdateMeter.mark()
+    onUpdateMeter.increment()
 
-    onUpdateCounter.inc()
+    onUpdateCounter.increment()
 
     val updatedRowData = update(rowKey, rowData)
 
@@ -418,9 +418,9 @@ class InMemDataTable(val tableDef: TableDef, val joinProvider: JoinTableProvider
 
   def processDelete(rowKey: String): Unit = {
 
-    onDeleteMeter.mark()
+    onDeleteMeter.increment()
 
-    onUpdateCounter.inc()
+    onUpdateCounter.increment()
 
     val rowData = delete(rowKey)
 

vuu/src/main/scala/org/finos/vuu/core/table/JoinTable.scala

@@ -79,7 +79,7 @@ class JoinTable(val tableDef: JoinTableDef,
 
   override def processUpdate(rowKey: String, rowUpdate: RowData): Unit = {
 
-    onUpdateMeter.mark()
+    onUpdateMeter.increment()
 
     logger.trace(s"$name processing row update: $rowKey $rowUpdate")