[ETCM-573] Add metrics on block imports

Author: Leonor Boga

docker/mantis/grafana/provisioning/dashboards/mantis-dashboard.json

@@ -1,8 +1,12 @@
 package io.iohk.ethereum.sync
 
+import com.typesafe.config.ConfigValueFactory
 import io.iohk.ethereum.FreeSpecBase
+import io.iohk.ethereum.metrics.{Metrics, MetricsConfig}
 import io.iohk.ethereum.sync.util.RegularSyncItSpecUtils.FakePeer
 import io.iohk.ethereum.sync.util.SyncCommonItSpec._
+import io.iohk.ethereum.utils.Config
+import io.prometheus.client.CollectorRegistry
 import monix.execution.Scheduler
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.matchers.should.Matchers
@@ -12,27 +16,54 @@ import scala.concurrent.duration._
 class RegularSyncItSpec extends FreeSpecBase with Matchers with BeforeAndAfterAll {
   implicit val testScheduler = Scheduler.fixedPool("test", 16)
 
+  override def beforeAll(): Unit = {
+    Metrics.configure(
+      MetricsConfig(Config.config.withValue("metrics.enabled", ConfigValueFactory.fromAnyRef(true)))
+    )
+  }
+
   override def afterAll(): Unit = {
     testScheduler.shutdown()
     testScheduler.awaitTermination(120.second)
   }
 
+  val getMinedBlockValue = CollectorRegistry.defaultRegistry.getSampleValue(
+    _: String,
+    Array("class"),
+    Array("MinedBlockPropagation")
+  )
+
+  val getNewBlockValue = CollectorRegistry.defaultRegistry.getSampleValue(
+    _: String,
+    Array("class"),
+    Array("NewBlockPropagation")
+  )
+
+  val getDefaultBlockValue = CollectorRegistry.defaultRegistry.getSampleValue(
+    _: String,
+    Array("class"),
+    Array("DefaultBlockPropagation")
+  )
+
   "peer 2 should sync to the top of peer1 blockchain" - {
     "given a previously imported blockchain" in customTestCaseResourceM(FakePeer.start2FakePeersRes()) {
       case (peer1, peer2) =>
-        val blockNumer: Int = 2000
+        val blockNumber: Int = 2000
         for {
-          _ <- peer1.importBlocksUntil(blockNumer)(IdentityUpdate)
+          _ <- peer1.importBlocksUntil(blockNumber)(IdentityUpdate)
           _ <- peer2.startRegularSync()
           _ <- peer2.connectToPeers(Set(peer1.node))
-          _ <- peer2.waitForRegularSyncLoadLastBlock(blockNumer)
+          _ <- peer2.waitForRegularSyncLoadLastBlock(blockNumber)
         } yield {
           assert(peer1.bl.getBestBlock().hash == peer2.bl.getBestBlock().hash)
         }
     }
 
     "given a previously mined blockchain" in customTestCaseResourceM(FakePeer.start2FakePeersRes()) {
       case (peer1, peer2) =>
+        val metricBefore = getMinedBlockValue("app_regularsync_blocks_propagation_timer_seconds_count")
+        Option(metricBefore) shouldBe None
+
         val blockHeadersPerRequest = peer2.syncConfig.blockHeadersPerRequest
         for {
           _ <- peer1.startRegularSync()
@@ -42,6 +73,7 @@ class RegularSyncItSpec extends FreeSpecBase with Matchers with BeforeAndAfterAl
           _ <- peer2.connectToPeers(Set(peer1.node))
           _ <- peer2.waitForRegularSyncLoadLastBlock(blockHeadersPerRequest + 1)
         } yield {
+
           assert(peer1.bl.getBestBlock().hash == peer2.bl.getBestBlock().hash)
         }
     }
@@ -96,4 +128,27 @@ class RegularSyncItSpec extends FreeSpecBase with Matchers with BeforeAndAfterAl
     }
   }
 
+  "A metric about mining a new block should be available" in customTestCaseResourceM(
+    FakePeer.start2FakePeersRes()
+  ) { case (peer1, peer2) =>
+    val minedMetricBefore = getMinedBlockValue("app_regularsync_blocks_propagation_timer_seconds_count")
+    val defaultMetricBefore = getDefaultBlockValue("app_regularsync_blocks_propagation_timer_seconds_count")
+
+    for {
+      _ <- peer1.startRegularSync()
+      _ <- peer1.mineNewBlocks(10.milliseconds, 1)(IdentityUpdate)
+      _ <- peer1.waitForRegularSyncLoadLastBlock(1)
+      _ <- peer2.startRegularSync()
+      _ <- peer2.connectToPeers(Set(peer1.node))
+      _ <- peer2.waitForRegularSyncLoadLastBlock(1)
+    } yield {
+
+      val minedMetricAfter = getMinedBlockValue("app_regularsync_blocks_propagation_timer_seconds_count").doubleValue()
+      val defaultMetricAfter =
+        getDefaultBlockValue("app_regularsync_blocks_propagation_timer_seconds_count").doubleValue()
+
+      minedMetricAfter shouldBe minedMetricBefore + 1.0d
+      defaultMetricAfter shouldBe defaultMetricBefore + 1.0d
+    }
+  }
 }

src/it/scala/io/iohk/ethereum/sync/RegularSyncItSpec.scala

@@ -105,8 +105,8 @@ object RegularSyncItSpecUtils {
       }
     }
 
-    def waitForRegularSyncLoadLastBlock(blockNumer: BigInt): Task[Boolean] = {
-      retryUntilWithDelay(Task(bl.getBestBlockNumber() == blockNumer), 1.second, 90) { isDone => isDone }
+    def waitForRegularSyncLoadLastBlock(blockNumber: BigInt): Task[Boolean] = {
+      retryUntilWithDelay(Task(bl.getBestBlockNumber() == blockNumber), 1.second, 90) { isDone => isDone }
     }
 
     def mineNewBlock(
@@ -116,9 +116,9 @@ object RegularSyncItSpecUtils {
       val currentWeight = bl
         .getChainWeightByHash(block.hash)
         .getOrElse(throw new RuntimeException(s"ChainWeight by hash: ${block.hash} doesn't exist"))
-      val currentWolrd = getMptForBlock(block)
+      val currentWorld = getMptForBlock(block)
       val (newBlock, _, _) =
-        createChildBlock(block, currentWeight, currentWolrd, plusDifficulty)(updateWorldForBlock)
+        createChildBlock(block, currentWeight, currentWorld, plusDifficulty)(updateWorldForBlock)
       regularSync ! SyncProtocol.MinedBlock(newBlock)
     }
 

src/it/scala/io/iohk/ethereum/sync/util/RegularSyncItSpecUtils.scala

@@ -174,7 +174,7 @@ class FastSync(
 
       case ResponseReceived(peer, BlockHeaders(blockHeaders), timeTaken) =>
         log.info("*** Received {} block headers in {} ms ***", blockHeaders.size, timeTaken)
-        SyncMetrics.setBlockHeadersDownloadTime(timeTaken)
+        FastSyncMetrics.setBlockHeadersDownloadTime(timeTaken)
 
         requestedHeaders.get(peer).foreach { requestedNum =>
           removeRequestHandler(sender())
@@ -189,7 +189,7 @@ class FastSync(
 
       case ResponseReceived(peer, BlockBodies(blockBodies), timeTaken) =>
         log.info("Received {} block bodies in {} ms", blockBodies.size, timeTaken)
-        SyncMetrics.setBlockBodiesDownloadTime(timeTaken)
+        FastSyncMetrics.setBlockBodiesDownloadTime(timeTaken)
 
         val requestedBodies = requestedBlockBodies.getOrElse(sender(), Nil)
         requestedBlockBodies -= sender()
@@ -198,7 +198,7 @@ class FastSync(
 
       case ResponseReceived(peer, Receipts(receipts), timeTaken) =>
         log.info("Received {} receipts in {} ms", receipts.size, timeTaken)
-        SyncMetrics.setBlockReceiptsDownloadTime(timeTaken)
+        FastSyncMetrics.setBlockReceiptsDownloadTime(timeTaken)
 
         val requestedHashes = requestedReceipts.getOrElse(sender(), Nil)
         requestedReceipts -= sender()
@@ -654,7 +654,7 @@ class FastSync(
     }
 
     def processSyncing(): Unit = {
-      SyncMetrics.measure(syncState)
+      FastSyncMetrics.measure(syncState)
       if (fullySynced) {
         finish()
       } else {
@@ -674,7 +674,7 @@ class FastSync(
 
     def finish(): Unit = {
       val totalTime = totalMinutesTaken()
-      SyncMetrics.setFastSyncTotalTimeGauge(totalTime.toDouble)
+      FastSyncMetrics.setFastSyncTotalTimeGauge(totalTime.toDouble)
       log.info("Total time taken for FastSync was {} minutes", totalTime)
       log.info("Block synchronization in fast mode finished, switching to regular mode")
 

src/main/scala/io/iohk/ethereum/blockchain/sync/fast/FastSync.scala

@@ -1,13 +1,13 @@
 package io.iohk.ethereum.blockchain.sync.fast
 
-import java.util.concurrent.atomic.AtomicLong
 import com.google.common.util.concurrent.AtomicDouble
 import io.iohk.ethereum.blockchain.sync.fast.FastSync.SyncState
 import io.iohk.ethereum.metrics.MetricsContainer
 
+import java.util.concurrent.atomic.AtomicLong
 import scala.concurrent.duration.MILLISECONDS
 
-object SyncMetrics extends MetricsContainer {
+object FastSyncMetrics extends MetricsContainer {
 
   private final val PivotBlockNumberGauge =
     metrics.registry.gauge("fastsync.block.pivotBlock.number.gauge", new AtomicDouble(0d))

src/main/scala/io/iohk/ethereum/blockchain/sync/fast/SyncMetrics.scala renamed to src/main/scala/io/iohk/ethereum/blockchain/sync/fast/FastSyncMetrics.scala

@@ -21,6 +21,7 @@ import io.iohk.ethereum.utils.ByteStringUtils
 import io.iohk.ethereum.utils.Config.SyncConfig
 import monix.eval.Task
 import monix.execution.Scheduler
+
 import scala.concurrent.duration._
 
 class SyncStateSchedulerActor(
@@ -64,7 +65,7 @@ class SyncStateSchedulerActor(
   def handleRequestResults: Receive = {
     case ResponseReceived(peer, nodeData: NodeData, timeTaken) =>
       log.info("Received {} state nodes in {} ms", nodeData.values.size, timeTaken)
-      SyncMetrics.setMptStateDownloadTime(timeTaken)
+      FastSyncMetrics.setMptStateDownloadTime(timeTaken)
 
       context unwatch (sender())
       self ! RequestData(nodeData, peer)