[ETCM-123] Add metrics related to network, rpc and mempool (#688)

Author: Michał Mrożek

src/main/scala/io/iohk/ethereum/jsonrpc/JsonRpcController.scala

@@ -15,10 +15,11 @@ import io.iohk.ethereum.jsonrpc.QAService.{GetPendingTransactionsRequest, GetPen
 import io.iohk.ethereum.jsonrpc.TestService._
 import io.iohk.ethereum.jsonrpc.server.http.JsonRpcHttpServer.JsonRpcHttpServerConfig
 import io.iohk.ethereum.jsonrpc.server.ipc.JsonRpcIpcServer.JsonRpcIpcServerConfig
-
+import java.util.concurrent.TimeUnit
 import scala.concurrent.Future
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.FiniteDuration
+import scala.util.{Failure, Success}
 
 object JsonRpcController {
 
@@ -349,13 +350,31 @@ class JsonRpcController(
   }
 
   def handleRequest(request: JsonRpcRequest): Future[JsonRpcResponse] = {
+    val startTimeNanos = System.nanoTime()
+
     val notFoundFn: PartialFunction[JsonRpcRequest, Future[JsonRpcResponse]] = { case _ =>
+      JsonRpcControllerMetrics.NotFoundMethodsCounter.increment()
       Future.successful(errorResponse(request, MethodNotFound))
     }
 
-    val handleFn =
+    val handleFn: PartialFunction[JsonRpcRequest, Future[JsonRpcResponse]] =
       enabledApis.foldLeft(notFoundFn)((fn, api) => apisHandleFns.getOrElse(api, PartialFunction.empty) orElse fn)
-    handleFn(request)
+
+    handleFn(request).andThen {
+      case Success(JsonRpcResponse(_, _, Some(JsonRpcError(_, _, _)), _)) =>
+        JsonRpcControllerMetrics.MethodsErrorCounter.increment()
+
+      case Success(JsonRpcResponse(_, _, None, _)) =>
+        JsonRpcControllerMetrics.MethodsSuccessCounter.increment()
+
+        val endTimeNanos = System.nanoTime()
+        val dtNanos = endTimeNanos - startTimeNanos
+        JsonRpcControllerMetrics.MethodsTimer.record(dtNanos, TimeUnit.NANOSECONDS)
+
+      case Failure(t) =>
+        JsonRpcControllerMetrics.MethodsExceptionCounter.increment()
+        log.error("Error serving request", t)
+    }
   }
 
   private def handle[Req, Res](

src/main/scala/io/iohk/ethereum/jsonrpc/JsonRpcControllerMetrics.scala

@@ -0,0 +1,16 @@
+package io.iohk.ethereum.jsonrpc
+
+import io.iohk.ethereum.metrics.MetricsContainer
+
+case object JsonRpcControllerMetrics extends MetricsContainer {
+
+  /**
+    * Counts attempts to call non-existing methods.
+    */
+  final val NotFoundMethodsCounter = metrics.counter("json.rpc.notfound.calls.counter")
+
+  final val MethodsTimer = metrics.timer("json.rpc.methods.timer")
+  final val MethodsSuccessCounter = metrics.counter("json.rpc.methods.success.counter")
+  final val MethodsExceptionCounter = metrics.counter("json.rpc.methods.exception.counter")
+  final val MethodsErrorCounter = metrics.counter("json.rpc.methods.error.counter")
+}

src/main/scala/io/iohk/ethereum/network/EtcPeerManagerActor.scala

@@ -20,8 +20,13 @@ import io.iohk.ethereum.network.p2p.messages.WireProtocol.Disconnect
   * In order to do so it receives events for peer creation, disconnection and new messages being sent and
   * received by each peer.
   */
-class EtcPeerManagerActor(peerManagerActor: ActorRef, peerEventBusActor: ActorRef, appStateStorage: AppStateStorage,
-                          forkResolverOpt: Option[ForkResolver]) extends Actor with ActorLogging {
+class EtcPeerManagerActor(
+    peerManagerActor: ActorRef,
+    peerEventBusActor: ActorRef,
+    appStateStorage: AppStateStorage,
+    forkResolverOpt: Option[ForkResolver]
+) extends Actor
+    with ActorLogging {
 
   private type PeersWithInfo = Map[PeerId, PeerWithInfo]
 
@@ -42,6 +47,7 @@ class EtcPeerManagerActor(peerManagerActor: ActorRef, peerEventBusActor: ActorRe
     val peerBestBlockIsItsGenesisBlock = peerInfo.bestBlockHash == peerInfo.remoteStatus.genesisHash
     peerBestBlockIsItsGenesisBlock || (!peerBestBlockIsItsGenesisBlock && peerInfo.maxBlockNumber > 0)
   }
+
   /**
     * Processes both messages for sending messages and for requesting peer information
     *
@@ -56,10 +62,11 @@ class EtcPeerManagerActor(peerManagerActor: ActorRef, peerEventBusActor: ActorRe
       })
 
     case PeerInfoRequest(peerId) =>
-      val peerInfoOpt = peersWithInfo.get(peerId).map{case PeerWithInfo(_, peerInfo) => peerInfo}
+      val peerInfoOpt = peersWithInfo.get(peerId).map { case PeerWithInfo(_, peerInfo) => peerInfo }
       sender() ! PeerInfoResponse(peerInfoOpt)
 
     case EtcPeerManagerActor.SendMessage(message, peerId) =>
+      NetworkMetrics.SentMessagesCounter.increment()
       val newPeersWithInfo = updatePeersWithInfo(peersWithInfo, peerId, message.underlyingMsg, handleSentMessage)
       peerManagerActor ! PeerManagerActor.SendMessage(message, peerId)
       context become handleMessages(newPeersWithInfo)
@@ -74,6 +81,7 @@ class EtcPeerManagerActor(peerManagerActor: ActorRef, peerEventBusActor: ActorRe
 
     case MessageFromPeer(message, peerId) if peersWithInfo.contains(peerId) =>
       val newPeersWithInfo = updatePeersWithInfo(peersWithInfo, peerId, message, handleReceivedMessage)
+      NetworkMetrics.ReceivedMessagesCounter.increment()
       context become handleMessages(newPeersWithInfo)
 
     case PeerHandshakeSuccessful(peer, peerInfo: PeerInfo) =>
@@ -82,11 +90,13 @@ class EtcPeerManagerActor(peerManagerActor: ActorRef, peerEventBusActor: ActorRe
 
       //Ask for the highest block from the peer
       peer.ref ! SendMessage(GetBlockHeaders(Right(peerInfo.remoteStatus.bestHash), 1, 0, false))
+      NetworkMetrics.registerAddHandshakedPeer(peer)
       context become handleMessages(peersWithInfo + (peer.id -> PeerWithInfo(peer, peerInfo)))
 
     case PeerDisconnected(peerId) if peersWithInfo.contains(peerId) =>
       peerEventBusActor ! Unsubscribe(PeerDisconnectedClassifier(PeerSelector.WithId(peerId)))
       peerEventBusActor ! Unsubscribe(MessageClassifier(msgCodesWithInfo, PeerSelector.WithId(peerId)))
+      NetworkMetrics.registerRemoveHandshakedPeer(peersWithInfo(peerId).peer)
       context become handleMessages(peersWithInfo - peerId)
 
   }
@@ -100,9 +110,13 @@ class EtcPeerManagerActor(peerManagerActor: ActorRef, peerEventBusActor: ActorRe
     * @param messageHandler for processing the message and obtaining the new peerInfo
     * @return new information for each peer
     */
-  private def updatePeersWithInfo(peers: PeersWithInfo, peerId: PeerId, message: Message,
-                                  messageHandler: (Message, PeerWithInfo) => PeerInfo): PeersWithInfo = {
-    if(peers.contains(peerId)){
+  private def updatePeersWithInfo(
+      peers: PeersWithInfo,
+      peerId: PeerId,
+      message: Message,
+      messageHandler: (Message, PeerWithInfo) => PeerInfo
+  ): PeersWithInfo = {
+    if (peers.contains(peerId)) {
       val peerWithInfo = peers(peerId)
       val newPeerInfo = messageHandler(message, peerWithInfo)
       peers + (peerId -> peerWithInfo.copy(peerInfo = newPeerInfo))
@@ -130,11 +144,9 @@ class EtcPeerManagerActor(peerManagerActor: ActorRef, peerEventBusActor: ActorRe
   private def handleReceivedMessage(message: Message, initialPeerWithInfo: PeerWithInfo): PeerInfo = {
     (updateTotalDifficulty(message) _
       andThen updateForkAccepted(message, initialPeerWithInfo.peer)
-      andThen updateMaxBlock(message)
-      )(initialPeerWithInfo.peerInfo)
+      andThen updateMaxBlock(message))(initialPeerWithInfo.peerInfo)
   }
 
-
   /**
     * Processes the message and updates the total difficulty of the peer
     *
@@ -217,11 +229,13 @@ object EtcPeerManagerActor {
 
   val msgCodesWithInfo: Set[Int] = Set(BlockHeaders.code, NewBlock.code, NewBlockHashes.code)
 
-  case class PeerInfo(remoteStatus: Status,
-                      totalDifficulty: BigInt,
-                      forkAccepted: Boolean,
-                      maxBlockNumber: BigInt,
-                      bestBlockHash: ByteString) extends HandshakeResult {
+  case class PeerInfo(
+      remoteStatus: Status,
+      totalDifficulty: BigInt,
+      forkAccepted: Boolean,
+      maxBlockNumber: BigInt,
+      bestBlockHash: ByteString
+  ) extends HandshakeResult {
 
     def withTotalDifficulty(totalDifficulty: BigInt): PeerInfo = copy(totalDifficulty = totalDifficulty)
 
@@ -253,8 +267,12 @@ object EtcPeerManagerActor {
 
   case class SendMessage(message: MessageSerializable, peerId: PeerId)
 
-  def props(peerManagerActor: ActorRef, peerEventBusActor: ActorRef,
-            appStateStorage: AppStateStorage, forkResolverOpt: Option[ForkResolver]): Props =
+  def props(
+      peerManagerActor: ActorRef,
+      peerEventBusActor: ActorRef,
+      appStateStorage: AppStateStorage,
+      forkResolverOpt: Option[ForkResolver]
+  ): Props =
     Props(new EtcPeerManagerActor(peerManagerActor, peerEventBusActor, appStateStorage, forkResolverOpt))
 
 }

src/main/scala/io/iohk/ethereum/network/NetworkMetrics.scala

@@ -0,0 +1,39 @@
+package io.iohk.ethereum.network
+
+import io.iohk.ethereum.metrics.MetricsContainer
+import java.util.concurrent.atomic.AtomicLong
+
+case object NetworkMetrics extends MetricsContainer {
+
+  private final val HandshakedIncomingPeersGauge =
+    metrics.registry.gauge("network.peers.incoming.handshaked.gauge", new AtomicLong(0))
+  private final val HandshakedOutgoingPeersGauge =
+    metrics.registry.gauge("network.peers.outgoing.handshaked.gauge", new AtomicLong(0))
+
+  final val ReceivedMessagesCounter = metrics.counter("network.messages.received.counter")
+
+  final val SentMessagesCounter = metrics.counter("network.messages.sent.counter")
+
+  final val DiscoveredPeersSize = metrics.registry.gauge("network.discovery.foundPeers.gauge", new AtomicLong(0))
+
+  final val BlacklistedPeersSize = metrics.registry.gauge("network.peers.blacklisted.gauge", new AtomicLong(0))
+
+  final val PendingPeersSize = metrics.registry.gauge("network.peers.pending.gauge", new AtomicLong(0))
+
+  def registerAddHandshakedPeer(peer: Peer): Unit = {
+    if (peer.incomingConnection) {
+      HandshakedIncomingPeersGauge.incrementAndGet()
+    } else {
+      HandshakedOutgoingPeersGauge.incrementAndGet()
+    }
+  }
+
+  def registerRemoveHandshakedPeer(peer: Peer): Unit = {
+    if (peer.incomingConnection) {
+      HandshakedIncomingPeersGauge.decrementAndGet()
+    } else {
+      HandshakedOutgoingPeersGauge.decrementAndGet()
+    }
+  }
+
+}