[ETCM-49] Replace Stats-D metrics with Prometheus

Author: mirkoAlic

README.md

@@ -43,6 +43,17 @@ in the root of the project.
 
 This updates all submodules and creates a distribution zip in `~/target/universal/`.
 
+### Monitoring
+
+#### Locally build & run monitoring client
+
+```
+# Build monitoring client docker image
+projectRoot $ docker build -f ./docker/monitoring-client.Dockerfile -t mantis-monitoring-client ./docker/
+# Run monitoring client in http://localhost:9090
+projectRoot $ docker run --network=host mantis-monitoring-client
+```
+
 ### Feedback
 
 Feedback gratefully received through the Ethereum Classic Forum (http://forum.ethereumclassic.org/)

build.sbt

@@ -10,12 +10,13 @@ val commonSettings = Seq(
 // Resolver for rocksDb
 resolvers += "rocksDb" at "https://dl.bintray.com/ethereum/maven/"
 
-val dep = {
+// TODO: Move to Dependencies.scala
+val dep: Seq[ModuleID] = {
   val akkaVersion = "2.6.9"
   val akkaHttpVersion = "10.2.0"
   val circeVersion = "0.9.3"
   val rocksDb = "5.9.2"
-  
+
   Seq(
     "com.typesafe.akka" %% "akka-actor" % akkaVersion,
     "com.typesafe.akka" %% "akka-slf4j" % akkaVersion,
@@ -52,8 +53,6 @@ val dep = {
     "net.java.dev.jna" % "jna" % "4.5.1",
     "com.github.scopt" %% "scopt" % "3.7.0",
 
-    // Metrics (https://github.com/DataDog/java-dogstatsd-client)
-    "com.datadoghq" % "java-dogstatsd-client" % "2.5",
     "org.xerial.snappy" % "snappy-java" % "1.1.7.2",
     "org.web3j" % "core" % "3.4.0" % "test"
   )
@@ -75,9 +74,13 @@ val root = project.in(file("."))
     .configs(Integration, Benchmark, Evm, Ets, Snappy, Rpc)
     .settings(commonSettings: _*)
     .settings(
-      libraryDependencies ++= dep,
-      executableScriptName := name.value
+      libraryDependencies ++= Seq(
+        dep,
+        Dependencies.micrometer,
+        Dependencies.prometheus
+      ).flatten
     )
+    .settings(executableScriptName := name.value)
     .settings(inConfig(Integration)(Defaults.testSettings) : _*)
     .settings(inConfig(Benchmark)(Defaults.testSettings) : _*)
     .settings(inConfig(Evm)(Defaults.testSettings) : _*)

docker/monitoring-client.Dockerfile

@@ -0,0 +1,2 @@
+FROM prom/prometheus
+ADD ./prometheus.yml /etc/prometheus/

docker/prometheus.yml

@@ -0,0 +1,29 @@
+# Please, don't use any default port allocations.
+# https://github.com/prometheus/prometheus/wiki/Default-port-allocations
+global:
+  scrape_interval: 1m
+  scrape_timeout: 10s
+  evaluation_interval: 1m
+scrape_configs:
+- job_name: prometheus
+  honor_timestamps: true
+  scrape_interval: 5s
+  scrape_timeout: 5s
+  metrics_path: /metrics
+  scheme: http
+  static_configs:
+  - targets:
+    - localhost:9090
+    labels:
+      alias: prometheus
+- job_name: node
+  honor_timestamps: true
+  scrape_interval: 10s
+  scrape_timeout: 10s
+  metrics_path: /metrics
+  scheme: http
+  static_configs:
+  - targets:
+    - localhost:13798
+    labels:
+      alias: mantis-node

project/Dependencies.scala

@@ -0,0 +1,27 @@
+import sbt._
+
+object Dependencies {
+
+  val prometheus: Seq[ModuleID] = {
+    val provider = "io.prometheus"
+    val version = "0.8.0"
+    Seq(
+      provider % "simpleclient" % version,
+      provider % "simpleclient_logback" % version,
+      provider % "simpleclient_hotspot" % version,
+      provider % "simpleclient_httpserver" % version
+    )
+  }
+
+  val micrometer: Seq[ModuleID] = {
+    val provider = "io.micrometer"
+    val version = "1.0.4"
+    Seq(
+      // Required to compile metrics library https://github.com/micrometer-metrics/micrometer/issues/1133#issuecomment-452434205
+      "com.google.code.findbugs" % "jsr305" % "3.0.2" % Optional,
+      provider % "micrometer-core" % version,
+      provider % "micrometer-registry-jmx" % version,
+      provider % "micrometer-registry-prometheus" % version
+    )
+  }
+}

src/main/resources/application.conf

@@ -504,35 +504,12 @@ mantis {
 
   metrics {
     # Set to `true` iff your deployment supports metrics collection.
-    # We push metrics to a StatsD-compatible agent and we use Datadog for collecting them in one place.
+    # We expose metrics using a Prometheus server
     # We default to `false` here because we do not expect all deployments to support metrics collection.
     enabled = false
 
-    # The StatsD-compatible agent host.
-    host = "localhost"
-
-    # The StatsD-compatible agent port.
-    port = 8125
-
-    # All metrics sent will be tagged with the environment.
-    # Sample values are: `public`, `private`, `production`, `dev`, depending on the use-case.
-    # If metrics are `enabled`, this is mandatory and must be set explicitly to a non-null value.
-    #
-    environment = ""
-
-    # All metrics sent will be tagged with the deployment.
-    # Sample values are: `kevm-testnet`, `iele-testnet`, `raft-kevm`, depending on the use-case.
-    # If metrics are `enabled`, this is mandatory and must be set explicitly to a non-null value.
-    #
-    deployment = ""
-
-    # Size of the metrics requests queue.
-    # If the queue contains that many outstanding requests to the metrics agent, then
-    # subsequent requests are blocked until the queue has room again.
-    queue-size = 1024
-
-    # Iff true, any errors during metrics client operations will be logged.
-    log-errors = true
+    # The port for setting up a Prometheus server over localhost.
+    port = 13798
   }
 }
 

src/main/resources/logback.xml

@@ -25,9 +25,12 @@
         </encoder>
     </appender>
 
+    <appender name="METRICS" class="io.prometheus.client.logback.InstrumentedAppender" />
+
     <root level="DEBUG">
         <appender-ref ref="STDOUT" />
         <appender-ref ref="FILE" />
+        <appender-ref ref="METRICS" />
     </root>
 
     <logger name="io.iohk.ethereum.network.rlpx.RLPxConnectionHandler" level="INFO" />

src/main/scala/io/iohk/ethereum/blockchain/sync/BlockMetrics.scala

@@ -0,0 +1,42 @@
+package io.iohk.ethereum.blockchain.sync
+
+import akka.util.ByteString
+import com.google.common.util.concurrent.AtomicDouble
+import io.iohk.ethereum.domain.Block
+import io.iohk.ethereum.metrics.MetricsContainer
+
+case object BlockMetrics extends MetricsContainer {
+
+  private[this] final val BlockNumberGauge =
+    metrics.registry.gauge("sync.block.number.gauge", new AtomicDouble(0d))
+  private[this] final val BlockGasLimitGauge =
+    metrics.registry.gauge("sync.block.gasLimit.gauge", new AtomicDouble(0d))
+  private[this] final val BlockGasUsedGauge =
+    metrics.registry.gauge("sync.block.gasUsed.gauge", new AtomicDouble(0d))
+  private[this] final val BlockDifficultyGauge =
+    metrics.registry.gauge("sync.block.difficulty.gauge", new AtomicDouble(0d))
+  private[this] final val BlockTransactionsGauge =
+    metrics.registry.gauge("sync.block.transactions.gauge", new AtomicDouble(0d))
+  private[this] final val BlockUnclesGauge =
+    metrics.registry.gauge("sync.block.uncles.gauge", new AtomicDouble(0d))
+  private[this] final val TimeBetweenParentGauge =
+    metrics.registry.gauge("sync.block.timeBetweenParent.seconds.gauge", new AtomicDouble(0d))
+
+  def measure(block: Block, getBlockByHashFn: ByteString => Option[Block]): Unit = {
+    BlockNumberGauge.set(block.number.toDouble)
+    BlockGasLimitGauge.set(block.header.gasLimit.toDouble)
+    BlockGasUsedGauge.set(block.header.gasUsed.toDouble)
+    BlockDifficultyGauge.set(block.header.difficulty.toDouble)
+    BlockTransactionsGauge.set(block.body.numberOfTxs)
+    BlockUnclesGauge.set(block.body.numberOfUncles)
+
+    getBlockByHashFn(block.header.parentHash) match {
+      case Some(parentBlock) =>
+        val timeBetweenBlocksInSeconds: Long =
+          block.header.unixTimestamp - parentBlock.header.unixTimestamp
+        TimeBetweenParentGauge.set(timeBetweenBlocksInSeconds)
+      case None => ()
+    }
+  }
+
+}

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockImporter.scala

@@ -6,6 +6,7 @@ import cats.data.NonEmptyList
 import cats.instances.future._
 import cats.instances.list._
 import cats.syntax.apply._
+import io.iohk.ethereum.blockchain.sync.BlockMetrics
 import io.iohk.ethereum.blockchain.sync.regular.BlockBroadcasterActor.BroadcastBlocks
 import io.iohk.ethereum.crypto.kec256
 import io.iohk.ethereum.domain.{Block, Blockchain, SignedTransaction}
@@ -125,7 +126,10 @@ class BlockImporter(
         }
 
         errorOpt match {
-          case None => Running
+          case None => {
+            importedBlocks.map(block => BlockMetrics.measure(block, blockchain.getBlockByHash))
+            Running
+          }
           case Some(err) =>
             log.error("Block import error {}", err)
             val notImportedBlocks = blocks.drop(importedBlocks.size)

src/main/scala/io/iohk/ethereum/ledger/BlockImport.scala

@@ -5,7 +5,7 @@ import io.iohk.ethereum.consensus.validators.BlockHeaderError.HeaderParentNotFou
 import io.iohk.ethereum.domain._
 import io.iohk.ethereum.ledger.BlockExecutionError.{ UnKnownExecutionError, ValidationBeforeExecError }
 import io.iohk.ethereum.ledger.BlockQueue.Leaf
-import io.iohk.ethereum.metrics.{ Metrics, MetricsClient }
+// import io.iohk.ethereum.metrics.{ Metrics, MetricsClient }
 import io.iohk.ethereum.utils.{ BlockchainConfig, Logger }
 import org.bouncycastle.util.encoders.Hex
 
@@ -67,10 +67,11 @@ class BlockImport(
           result.toString
         })
 
-        if (importedBlocks.nonEmpty) {
-          val maxNumber = importedBlocks.map(_.block.header.number).max.toLong
-          MetricsClient.get().gauge(Metrics.LedgerImportBlockNumber, maxNumber)
-        }
+        // FIXME [ETCM-49]
+        // if (importedBlocks.nonEmpty) {
+        //   val maxNumber = importedBlocks.map(_.block.header.number).max.toLong
+        //   MetricsClient.get().gauge(Metrics.LedgerImportBlockNumber, maxNumber)
+        // }
 
         result
 

src/main/scala/io/iohk/ethereum/metrics/AppJmxConfig.scala

@@ -0,0 +1,11 @@
+package io.iohk.ethereum.metrics
+
+import io.micrometer.jmx.JmxConfig
+
+class AppJmxConfig extends JmxConfig {
+  def get(key: String): String = null
+
+  override def prefix(): String = Metrics.MetricsPrefix
+
+  override def domain(): String = prefix()
+}

src/main/scala/io/iohk/ethereum/metrics/DeltaSpikeGauge.scala

@@ -0,0 +1,33 @@
+package io.iohk.ethereum.metrics
+
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
+
+/**
+  * A gauge that starts at `0` and can be triggered to go to `1`.
+  * Next time it is sampled, it goes back to `0`.
+  * This is normally used for either one-off signals (e.g. when an application starts)
+  * or slowly re-appearing signals. Specifically, the sampling rate must be greater
+  * than the rate the signal is triggered.
+  */
+class DeltaSpikeGauge(name: String, metrics: Metrics) {
+  private[this] final val isTriggeredRef = new AtomicBoolean(false)
+  private[this] final val valueRef = new AtomicInteger(0)
+
+  private[this] def getValue(): Double = {
+    if (isTriggeredRef.compareAndSet(true, false)) {
+      valueRef.getAndSet(0)
+    } else {
+      valueRef.get()
+    }
+  }
+
+  private[this] final val gauge = metrics.gauge(name, () ⇒ getValue)
+
+  def trigger(): Unit = {
+    if (isTriggeredRef.compareAndSet(false, true)) {
+      valueRef.set(1)
+      // Let one of the exporting metric registries pick up the `1`.
+      // As soon as that happens, `getValue` will make sure that we go back to `0`.
+    }
+  }
+}

src/main/scala/io/iohk/ethereum/metrics/MeterRegistryBuilder.scala

@@ -0,0 +1,62 @@
+package io.iohk.ethereum.metrics
+
+import io.iohk.ethereum.utils.Logger
+import io.iohk.ethereum.utils.LoggingUtils.getClassName
+import io.micrometer.core.instrument.composite.CompositeMeterRegistry
+import io.micrometer.core.instrument.config.MeterFilter
+import io.micrometer.core.instrument._
+import io.micrometer.prometheus.{PrometheusMeterRegistry, PrometheusConfig}
+import io.prometheus.client.CollectorRegistry
+import io.micrometer.jmx.JmxMeterRegistry
+
+object MeterRegistryBuilder extends Logger {
+
+  private[this] final val StdMetricsClock = Clock.SYSTEM
+
+  private[this] def onMeterAdded(m: Meter): Unit =
+    log.debug(s"New ${getClassName(m)} metric: " + m.getId.getName)
+
+  /**
+    * Build our meter registry consist in:
+    * 1. Create each Meter registry
+    * 2. Config the resultant composition
+    */
+  def build(metricsPrefix: String): MeterRegistry = {
+
+    val jmxMeterRegistry = new JmxMeterRegistry(new AppJmxConfig, StdMetricsClock)
+
+    log.info(s"Build JMX Meter Registry: ${jmxMeterRegistry}")
+
+    val prometheusMeterRegistry =
+      new PrometheusMeterRegistry(
+        PrometheusConfig.DEFAULT,
+        CollectorRegistry.defaultRegistry,
+        StdMetricsClock
+      );
+
+    log.info(s"Build Prometheus Meter Registry: ${prometheusMeterRegistry}")
+
+    val registry = new CompositeMeterRegistry(
+      StdMetricsClock,
+      java.util.Arrays.asList(jmxMeterRegistry, prometheusMeterRegistry)
+    )
+
+    // Ensure that all metrics have the `Prefix`.
+    // We are of course mainly interested in those that we do not control,
+    // e.g. those coming from `JvmMemoryMetrics`.
+    val metricsPrefixDot = metricsPrefix + "."
+
+    registry
+      .config()
+      .meterFilter(new MeterFilter {
+        override def map(id: Meter.Id): Meter.Id = {
+          val name = id.getName
+          if (name.startsWith(metricsPrefixDot)) id
+          else id.withName(metricsPrefixDot + name)
+        }
+      })
+      .onMeterAdded(onMeterAdded)
+
+    registry
+  }
+}