Merge remote-tracking branch 'origin/develop' into ETCM-446-connection-limit-ranges

Author: aakoshh

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}
 
@@ -623,7 +620,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
@@ -677,4 +675,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

@@ -323,7 +323,7 @@ class PeerManagerActor(
   /** Disconnect some incoming connections so we can free up slots. */
   private def pruneIncomingPeers(
       connectedPeers: ConnectedPeers,
-      stats: Map[PeerId, PeerStatisticsActor.Stat]
+      stats: Map[PeerId, PeerStat]
   ): ConnectedPeers = {
     val pruneCount = PeerManagerActor.numberOfIncomingConnectionsToPrune(connectedPeers, peerConfiguration)
     val now = System.currentTimeMillis
@@ -535,7 +535,7 @@ object PeerManagerActor {
   /** Assign a priority to peers that we can use to order connections,
     * with lower priorities being the ones to prune first.
     */
-  def prunePriority(stats: Map[PeerId, PeerStatisticsActor.Stat], currentTimeMillis: Long)(peerId: PeerId): Double = {
+  def prunePriority(stats: Map[PeerId, PeerStat], currentTimeMillis: Long)(peerId: PeerId): Double = {
     stats
       .get(peerId)
       .flatMap { stat =>

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

@@ -1,31 +1,30 @@
 package io.iohk.ethereum.network
 
 import akka.actor._
-import cats._
-import cats.implicits._
 import io.iohk.ethereum.network.PeerEventBusActor._
 import io.iohk.ethereum.network.p2p.messages.Codes
 import java.time.Clock
 import scala.concurrent.duration.FiniteDuration
-import cats.kernel.Monoid
 
 class PeerStatisticsActor(
     peerEventBus: ActorRef,
-    var maybeStats: Option[TimeSlotStats[PeerId, PeerStatisticsActor.Stat]]
+    var maybeStats: Option[TimeSlotStats[PeerId, PeerStat]]
 ) extends Actor {
   import PeerStatisticsActor._
 
-  // 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))
+  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 = Stat(
+      val obs = PeerStat(
         responsesReceived = if (ResponseCodes(msg.code)) 1 else 0,
         requestsReceived = if (RequestCodes(msg.code)) 1 else 0,
         firstSeenTimeMillis = Some(now),
@@ -43,49 +42,22 @@ class PeerStatisticsActor(
       sender ! StatsForAll(stats)
 
     case GetStatsForPeer(window, peerId) =>
-      val stats = maybeStats.map(_.get(peerId, Some(window))).getOrElse(Stat.empty)
+      val stats = maybeStats.map(_.get(peerId, Some(window))).getOrElse(PeerStat.empty)
       sender ! StatsForPeer(peerId, stats)
   }
 }
 
 object PeerStatisticsActor {
-  case class Stat(
-      responsesReceived: Int,
-      requestsReceived: Int,
-      firstSeenTimeMillis: Option[Long],
-      lastSeenTimeMillis: Option[Long]
-  )
-  object Stat {
-    val empty: Stat = Stat(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[Stat] =
-      Monoid.instance(
-        empty,
-        (a, b) =>
-          Stat(
-            responsesReceived = a.responsesReceived + b.responsesReceived,
-            requestsReceived = a.requestsReceived + b.requestsReceived,
-            firstSeenTimeMillis = mergeOpt(a, b)(_.firstSeenTimeMillis)(math.min),
-            lastSeenTimeMillis = mergeOpt(a, b)(_.lastSeenTimeMillis)(math.max)
-          )
-      )
-  }
-
   def props(peerEventBus: ActorRef, slotDuration: FiniteDuration, slotCount: Int): Props =
     Props {
       implicit val clock = Clock.systemUTC()
-      new PeerStatisticsActor(peerEventBus, TimeSlotStats[PeerId, Stat](slotDuration, slotCount))
+      new PeerStatisticsActor(peerEventBus, TimeSlotStats[PeerId, PeerStat](slotDuration, slotCount))
     }
 
   case class GetStatsForAll(window: FiniteDuration)
-  case class StatsForAll(stats: Map[PeerId, Stat])
+  case class StatsForAll(stats: Map[PeerId, PeerStat])
   case class GetStatsForPeer(window: FiniteDuration, peerId: PeerId)
-  case class StatsForPeer(peerId: PeerId, stat: Stat)
+  case class StatsForPeer(peerId: PeerId, stat: PeerStat)
 
   val ResponseCodes = Set(
     Codes.NewBlockCode,

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

@@ -4,6 +4,7 @@ import cats._
 import cats.implicits._
 import java.time.Clock
 import scala.concurrent.duration.{Duration, FiniteDuration}
+import scala.annotation.tailrec
 
 /** Track statistics over time a fixed size timewindow. */
 class TimeSlotStats[K, V: Monoid] private (
@@ -59,6 +60,7 @@ class TimeSlotStats[K, V: Monoid] private (
   private def fold[A](init: A, window: Duration)(f: (A, Map[K, V]) => A) = {
     val (start, end) = slotRange(currentTimeMillis, window)
 
+    @tailrec
     def loop(idx: Int, acc: List[Map[K, V]]): List[Map[K, V]] = {
       val entry = buffer(idx)
       if (entry.slotId < start || end < entry.slotId)

src/main/scala/io/iohk/ethereum/network/p2p/Message.scala

@@ -11,6 +11,7 @@ object Message {
 
 trait Message {
   def code: Int
+  def toShortString: String
 }
 
 trait MessageSerializable extends Message {
@@ -21,7 +22,6 @@ trait MessageSerializable extends Message {
   def toBytes: Array[Byte]
 
   def underlyingMsg: Message
-
 }
 
 trait MessageDecoder { self =>

src/main/scala/io/iohk/ethereum/network/p2p/MessageSerializableImplicit.scala

@@ -13,4 +13,5 @@ abstract class MessageSerializableImplicit[T <: Message](val msg: T) extends Mes
 
   override def hashCode(): Int = msg.hashCode()
 
+  override def toShortString: String = msg.toShortString
 }

src/main/scala/io/iohk/ethereum/network/p2p/messages/CommonMessages.scala

@@ -67,6 +67,7 @@ object CommonMessages {
         s"genesisHash: ${Hex.toHexString(genesisHash.toArray[Byte])}," +
         s"}"
 
+    override def toShortString: String = toString
     override def code: Int = Codes.StatusCode
   }
 
@@ -121,6 +122,13 @@ object CommonMessages {
         s"totalDifficulty: $totalDifficulty" +
         s"}"
 
+    override def toShortString: String =
+      s"NewBlock { " +
+        s"code: $code, " +
+        s"block.header: ${block.header}, " +
+        s"totalDifficulty: $totalDifficulty" +
+        s"}"
+
     override def code: Int = Codes.NewBlockCode
   }
 
@@ -193,5 +201,7 @@ object CommonMessages {
 
   case class SignedTransactions(txs: Seq[SignedTransaction]) extends Message {
     override def code: Int = Codes.SignedTransactionsCode
+    override def toShortString: String =
+      s"SignedTransactions { txs: ${txs.map(_.hashAsHexString)} }"
   }
 }

src/main/scala/io/iohk/ethereum/network/p2p/messages/PV61.scala

@@ -5,6 +5,7 @@ import io.iohk.ethereum.network.p2p.{Message, MessageSerializableImplicit}
 import io.iohk.ethereum.rlp.RLPImplicitConversions._
 import io.iohk.ethereum.rlp.RLPImplicits._
 import io.iohk.ethereum.rlp._
+import org.bouncycastle.util.encoders.Hex
 
 object PV61 {
 
@@ -29,6 +30,8 @@ object PV61 {
 
   case class NewBlockHashes(hashes: Seq[ByteString]) extends Message {
     override def code: Int = Codes.NewBlockHashesCode
+    override def toShortString: String =
+      s"NewBlockHashes { hashes: ${hashes.map(h => Hex.toHexString(h.toArray[Byte]))} } "
   }
 
   object BlockHashesFromNumber {
@@ -51,6 +54,9 @@ object PV61 {
 
   case class BlockHashesFromNumber(number: BigInt, maxBlocks: BigInt) extends Message {
     override def code: Int = Codes.BlockHashesFromNumberCode
+    override def toString: String =
+      s"BlockHashesFromNumber { number: $number, maxBlocks: $maxBlocks }"
+    override def toShortString: String = toString
   }
 
 }

src/main/scala/io/iohk/ethereum/network/p2p/messages/PV62.scala

@@ -60,6 +60,11 @@ object PV62 {
 
   case class NewBlockHashes(hashes: Seq[BlockHash]) extends Message {
     override def code: Int = Codes.NewBlockHashesCode
+    override def toString: String =
+      s"NewBlockHashes { " +
+        s"hashes: ${hashes} " +
+        s"}"
+    override def toShortString: String = toString
   }
 
   object GetBlockHeaders {
@@ -102,6 +107,8 @@ object PV62 {
         s"skip: $skip " +
         s"reverse: $reverse " +
         s"}"
+
+    override def toShortString: String = toString
   }
 
   object BlockBodies {
@@ -123,6 +130,8 @@ object PV62 {
 
   case class BlockBodies(bodies: Seq[BlockBody]) extends Message {
     val code: Int = Codes.BlockBodiesCode
+    override def toShortString: String =
+      s"BlockBodies { bodies: ${bodies.map(_.toShortString)} }"
   }
 
   object BlockHeaders {
@@ -148,6 +157,8 @@ object PV62 {
 
   case class BlockHeaders(headers: Seq[BlockHeader]) extends Message {
     override def code: Int = Codes.BlockHeadersCode
+    override def toShortString: String =
+      s"BlockHeaders { headers: ${headers.map(_.hashAsHexString)} }"
   }
 
   object GetBlockBodies {
@@ -176,5 +187,6 @@ object PV62 {
       s"GetBlockBodies { " +
         s"hashes: ${hashes.map(h => Hex.toHexString(h.toArray[Byte]))} " +
         s"}"
+    override def toShortString: String = toString
   }
 }