Merge remote-tracking branch 'origin/develop' into etcm-472/missing-fields-receipts

Author: KonradStaniec

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

@@ -32,6 +32,7 @@ import io.iohk.ethereum.network.{
   KnownNodesManager,
   PeerEventBusActor,
   PeerManagerActor,
+  PeerStatisticsActor,
   ServerActor
 }
 import io.iohk.ethereum.nodebuilder.PruningConfigBuilder
@@ -149,6 +150,8 @@ abstract class CommonFakePeer(peerName: String, fakePeerCustomConfig: FakePeerCu
     override val updateNodesInterval: FiniteDuration = 20.seconds
     override val shortBlacklistDuration: FiniteDuration = 1.minute
     override val longBlacklistDuration: FiniteDuration = 3.minutes
+    override val statSlotDuration: FiniteDuration = 1.minute
+    override val statSlotCount: Int = 30
   }
 
   lazy val peerEventBus = system.actorOf(PeerEventBusActor.props, "peer-event-bus")
@@ -167,12 +170,16 @@ abstract class CommonFakePeer(peerName: String, fakePeerCustomConfig: FakePeerCu
 
   lazy val authHandshaker: AuthHandshaker = AuthHandshaker(nodeKey, secureRandom)
 
+  lazy val peerStatistics =
+    system.actorOf(PeerStatisticsActor.props(peerEventBus, slotDuration = 1.minute, slotCount = 30))
+
   lazy val peerManager: ActorRef = system.actorOf(
     PeerManagerActor.props(
       peerDiscoveryManager,
       Config.Network.peer,
       peerEventBus,
       knownNodesManager,
+      peerStatistics,
       handshaker,
       authHandshaker,
       EthereumMessageDecoder,

src/it/scala/io/iohk/ethereum/txExecTest/util/DumpChainApp.scala

@@ -18,6 +18,7 @@ import io.iohk.ethereum.ledger.{InMemoryWorldStateProxy, InMemoryWorldStateProxy
 import io.iohk.ethereum.mpt.MptNode
 import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
 import io.iohk.ethereum.network.PeerManagerActor.PeerConfiguration
+import io.iohk.ethereum.network.PeerStatisticsActor
 import io.iohk.ethereum.network.discovery.DiscoveryConfig
 import io.iohk.ethereum.network.handshaker.{EtcHandshaker, EtcHandshakerConfiguration, Handshaker}
 import io.iohk.ethereum.network.p2p.EthereumMessageDecoder
@@ -60,6 +61,8 @@ object DumpChainApp extends App with NodeKeyBuilder with SecureRandomBuilder wit
     override val updateNodesInterval: FiniteDuration = 20.seconds
     override val shortBlacklistDuration: FiniteDuration = 1.minute
     override val longBlacklistDuration: FiniteDuration = 3.minutes
+    override val statSlotDuration: FiniteDuration = 1.minute
+    override val statSlotCount: Int = 30
   }
 
   val actorSystem = ActorSystem("mantis_system")
@@ -91,17 +94,20 @@ object DumpChainApp extends App with NodeKeyBuilder with SecureRandomBuilder wit
 
   val peerMessageBus = actorSystem.actorOf(PeerEventBusActor.props)
 
+  val peerStatistics = actorSystem.actorOf(PeerStatisticsActor.props(peerMessageBus, 1.minute, 30))
+
   val peerManager = actorSystem.actorOf(
     PeerManagerActor.props(
       peerDiscoveryManager = actorSystem.deadLetters, // TODO: fixme
       peerConfiguration = peerConfig,
       peerMessageBus = peerMessageBus,
+      peerStatistics = peerStatistics,
       knownNodesManager = actorSystem.deadLetters, // TODO: fixme
       handshaker = handshaker,
       authHandshaker = authHandshaker,
       messageDecoder = EthereumMessageDecoder,
-      discoveryConfig,
-      Config.Network.protocolVersion
+      discoveryConfig = discoveryConfig,
+      bestProtocolVersion = Config.Network.protocolVersion
     ),
     "peer-manager"
   )

src/main/resources/application.conf

@@ -2,9 +2,6 @@ mantis {
   # Optionally augment the client ID sent in Hello messages.
   client-identity = null
 
-  # Version string (reported by an RPC method)
-  client-version = "mantis/v2.0"
-
   # Base directory where all the data used by the node is stored, including blockchain data and private keys
   datadir = ${user.home}"/.mantis/"${mantis.blockchains.network}
 
@@ -163,6 +160,12 @@ mantis {
       # Peer which disconnect during tcp connection becouse of other reasons will not be retried for this long duration
       # other reasons include: timeout during connection, wrong protocol, incompatible network
       long-blacklist-duration = 30.minutes
+
+      # Resolution of moving window of peer statistics.
+      # Will be multiplied by `stat-slot-count` to give the overall length of peer statistics availability.
+      stat-slot-duration = 1.minute
+      # How many slots of peer statistics to keep in the moving window, e.g. 60 * 1.minute to keep stats for the last hour with 1 minute resolution.
+      stat-slot-count = 60
     }
 
     rpc {
@@ -605,7 +608,8 @@ mantis {
 
 akka {
   loggers = ["akka.event.slf4j.Slf4jLogger"]
-  loglevel = "DEBUG"
+  # Not using ${logging.logs-level} because it might be set to TRACE, which our version of Akka doesn't have.
+  loglevel = "INFO"
   logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
   logger-startup-timeout = 30s
   log-dead-letters = off
@@ -659,4 +663,7 @@ logging {
 
   # Logs filename
   logs-file = "mantis"
+
+  # Logs level
+  logs-level = "INFO"
 }

src/main/resources/logback.xml

@@ -8,6 +8,7 @@
     <load key="logging.json-output" as="ASJSON"/>
     <load key="logging.logs-dir" as="LOGSDIR"/>
     <load key="logging.logs-file" as="LOGSFILENAME"/>
+    <load key="logging.logs-level" as="LOGSLEVEL"/>
 
     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
         <encoder>
@@ -43,7 +44,7 @@
 
     <appender name="METRICS" class="io.prometheus.client.logback.InstrumentedAppender" />
 
-    <root level="DEBUG">
+    <root level="${LOGSLEVEL}">
         <if condition='p("ASJSON").contains("true")'>
             <then>
                 <appender-ref ref="STASH" />
@@ -58,9 +59,9 @@
 
     <logger name="io.netty" level="WARN"/>
     <logger name="io.iohk.scalanet" level="INFO" />
-    <logger name="io.iohk.ethereum.network.rlpx.RLPxConnectionHandler" level="DEBUG" />
     <logger name="io.iohk.ethereum.blockchain.sync.SyncController" level="INFO" />
-    <logger name="io.iohk.ethereum.network.PeerActor" level="DEBUG" />
+    <logger name="io.iohk.ethereum.network.PeerActor" level="${LOGSLEVEL}" />
+    <logger name="io.iohk.ethereum.network.rlpx.RLPxConnectionHandler" level="${LOGSLEVEL}" />
     <logger name="io.iohk.ethereum.vm.VM" level="OFF" />
 
 </configuration>

src/main/scala/io/iohk/ethereum/domain/BlockBody.scala

@@ -7,6 +7,9 @@ case class BlockBody(transactionList: Seq[SignedTransaction], uncleNodesList: Se
   override def toString: String =
     s"BlockBody{ transactionList: $transactionList, uncleNodesList: $uncleNodesList }"
 
+  def toShortString: String =
+    s"BlockBody { transactionsList: ${transactionList.map(_.hashAsHexString)}, uncleNodesList: ${uncleNodesList.map(_.hashAsHexString)} }"
+
   lazy val numberOfTxs: Int = transactionList.size
 
   lazy val numberOfUncles: Int = uncleNodesList.size

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

@@ -17,6 +17,7 @@ import io.iohk.ethereum.network.p2p._
 import io.iohk.ethereum.network.p2p.messages.WireProtocol._
 import io.iohk.ethereum.network.rlpx.RLPxConnectionHandler.RLPxConfiguration
 import io.iohk.ethereum.network.rlpx.{AuthHandshaker, RLPxConnectionHandler}
+import io.iohk.ethereum.utils.Logger
 import org.bouncycastle.util.encoders.Hex
 
 /**
@@ -244,6 +245,23 @@ class PeerActor[R <: HandshakeResult](
     stash()
   }
 
+  // The actor logs incoming messages, which can be quite verbose even for DEBUG mode.
+  // ActorLogging doesn't support TRACE, but we can push more details if trace is enabled using the normal logging facilites.
+  object MessageLogger extends Logger {
+    val isTraceEnabled = {
+      var enabled = false
+      log.whenTraceEnabled({ enabled = true })
+      enabled
+    }
+    def logMessage(peerId: PeerId, message: Message): Unit =
+      // Sometimes potentially seeing the full block in the result is useful.
+      if (isTraceEnabled) {
+        log.trace(s"Received message: {} from $peerId", message)
+      } else {
+        log.debug(s"Received message: {} from $peerId", message.toShortString)
+      }
+  }
+
   class HandshakedPeer(remoteNodeId: ByteString, rlpxConnection: RLPxConnection, handshakeResult: R) {
 
     val peer: Peer = Peer(peerAddress, self, incomingConnection, Some(remoteNodeId))
@@ -258,7 +276,7 @@ class PeerActor[R <: HandshakeResult](
         handleTerminated(rlpxConnection, 0, Handshaked) orElse {
 
           case RLPxConnectionHandler.MessageReceived(message) =>
-            log.debug(s"Received message: {} from $peerId", message)
+            MessageLogger.logMessage(peerId, message)
             peerEventBus ! Publish(MessageFromPeer(message, peer.id))
 
           case DisconnectPeer(reason) =>

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

@@ -31,6 +31,7 @@ class PeerManagerActor(
     peerDiscoveryManager: ActorRef,
     peerConfiguration: PeerConfiguration,
     knownNodesManager: ActorRef,
+    peerStatistics: ActorRef,
     peerFactory: (ActorContext, InetSocketAddress, Boolean) => ActorRef,
     discoveryConfig: DiscoveryConfig,
     externalSchedulerOpt: Option[Scheduler] = None
@@ -52,8 +53,6 @@ class PeerManagerActor(
   import PeerManagerActor._
   import akka.pattern.{ask, pipe}
 
-  private type PeerMap = Map[PeerId, Peer]
-
   implicit class ConnectedPeersOps(connectedPeers: ConnectedPeers) {
     def outgoingConnectionDemand: Int =
       peerConfiguration.maxOutgoingPeers - connectedPeers.outgoingPeersCount
@@ -339,6 +338,7 @@ object PeerManagerActor {
       peerConfiguration: PeerConfiguration,
       peerMessageBus: ActorRef,
       knownNodesManager: ActorRef,
+      peerStatistics: ActorRef,
       handshaker: Handshaker[R],
       authHandshaker: AuthHandshaker,
       messageDecoder: MessageDecoder,
@@ -362,6 +362,7 @@ object PeerManagerActor {
         peerDiscoveryManager,
         peerConfiguration,
         knownNodesManager,
+        peerStatistics,
         peerFactory = factory,
         discoveryConfig
       )
@@ -410,6 +411,8 @@ object PeerManagerActor {
     val updateNodesInterval: FiniteDuration
     val shortBlacklistDuration: FiniteDuration
     val longBlacklistDuration: FiniteDuration
+    val statSlotDuration: FiniteDuration
+    val statSlotCount: Int
   }
 
   trait FastSyncHostConfiguration {
@@ -444,5 +447,4 @@ object PeerManagerActor {
   case class OutgoingConnectionAlreadyHandled(uri: URI) extends ConnectionError
 
   case class PeerAddress(value: String) extends BlackListId
-
 }

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

@@ -0,0 +1,31 @@
+package io.iohk.ethereum.network
+
+import cats._
+import cats.implicits._
+
+case class PeerStat(
+    responsesReceived: Int,
+    requestsReceived: Int,
+    firstSeenTimeMillis: Option[Long],
+    lastSeenTimeMillis: Option[Long]
+)
+object PeerStat {
+  val empty: PeerStat = PeerStat(0, 0, None, None)
+
+  private def mergeOpt[A, B](x: A, y: A)(f: A => Option[B])(g: (B, B) => B): Option[B] = {
+    val (mx, my) = (f(x), f(y))
+    (mx, my).mapN(g) orElse mx orElse my
+  }
+
+  implicit val monoid: Monoid[PeerStat] =
+    Monoid.instance(
+      empty,
+      (a, b) =>
+        PeerStat(
+          responsesReceived = a.responsesReceived + b.responsesReceived,
+          requestsReceived = a.requestsReceived + b.requestsReceived,
+          firstSeenTimeMillis = mergeOpt(a, b)(_.firstSeenTimeMillis)(math.min),
+          lastSeenTimeMillis = mergeOpt(a, b)(_.lastSeenTimeMillis)(math.max)
+        )
+    )
+}

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

@@ -0,0 +1,85 @@
+package io.iohk.ethereum.network
+
+import akka.actor._
+import io.iohk.ethereum.network.PeerEventBusActor._
+import io.iohk.ethereum.network.p2p.messages.Codes
+import java.time.Clock
+import scala.concurrent.duration.FiniteDuration
+
+class PeerStatisticsActor(
+    peerEventBus: ActorRef,
+    var maybeStats: Option[TimeSlotStats[PeerId, PeerStat]]
+) extends Actor {
+  import PeerStatisticsActor._
+
+  override def preStart(): Unit = {
+    // Subscribe to messages received from handshaked peers to maintain stats.
+    peerEventBus ! Subscribe(MessageSubscriptionClassifier)
+    // Removing peers is an optimisation to free space, but eventually the stats would be overwritten anyway.
+    peerEventBus ! Subscribe(SubscriptionClassifier.PeerDisconnectedClassifier(PeerSelector.AllPeers))
+  }
+
+  def receive: Receive = handlePeerEvents orElse handleStatsRequests
+
+  private def handlePeerEvents: Receive = {
+    case PeerEvent.MessageFromPeer(msg, peerId) =>
+      val now = System.currentTimeMillis()
+      val obs = PeerStat(
+        responsesReceived = if (ResponseCodes(msg.code)) 1 else 0,
+        requestsReceived = if (RequestCodes(msg.code)) 1 else 0,
+        firstSeenTimeMillis = Some(now),
+        lastSeenTimeMillis = Some(now)
+      )
+      maybeStats = maybeStats.map(_.add(peerId, obs))
+
+    case PeerEvent.PeerDisconnected(peerId) =>
+      maybeStats = maybeStats.map(_.remove(peerId))
+  }
+
+  private def handleStatsRequests: Receive = {
+    case GetStatsForAll(window) =>
+      val stats = maybeStats.map(_.getAll(Some(window))).getOrElse(Map.empty)
+      sender ! StatsForAll(stats)
+
+    case GetStatsForPeer(window, peerId) =>
+      val stats = maybeStats.map(_.get(peerId, Some(window))).getOrElse(PeerStat.empty)
+      sender ! StatsForPeer(peerId, stats)
+  }
+}
+
+object PeerStatisticsActor {
+  def props(peerEventBus: ActorRef, slotDuration: FiniteDuration, slotCount: Int): Props =
+    Props {
+      implicit val clock = Clock.systemUTC()
+      new PeerStatisticsActor(peerEventBus, TimeSlotStats[PeerId, PeerStat](slotDuration, slotCount))
+    }
+
+  case class GetStatsForAll(window: FiniteDuration)
+  case class StatsForAll(stats: Map[PeerId, PeerStat])
+  case class GetStatsForPeer(window: FiniteDuration, peerId: PeerId)
+  case class StatsForPeer(peerId: PeerId, stat: PeerStat)
+
+  val ResponseCodes = Set(
+    Codes.NewBlockCode,
+    Codes.NewBlockHashesCode,
+    Codes.SignedTransactionsCode,
+    Codes.BlockHeadersCode,
+    Codes.BlockBodiesCode,
+    Codes.BlockHashesFromNumberCode,
+    Codes.NodeDataCode,
+    Codes.ReceiptsCode
+  )
+
+  val RequestCodes = Set(
+    Codes.GetBlockHeadersCode,
+    Codes.GetBlockBodiesCode,
+    Codes.GetNodeDataCode,
+    Codes.GetReceiptsCode
+  )
+
+  val MessageSubscriptionClassifier =
+    SubscriptionClassifier.MessageClassifier(
+      messageCodes = RequestCodes union ResponseCodes,
+      peerSelector = PeerSelector.AllPeers
+    )
+}