Merge pull request #699 from input-output-hk/update-sync-logging

[ETCM-120] Add more logs

Author: Michał Mrożek

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

@@ -43,6 +43,7 @@ class PeersClient(
         val requester = sender()
         selectPeer(peerSelector) match {
           case Some(peer) =>
+            log.debug("Selected peer {} with address {}", peer.id, peer.remoteAddress.getHostString)
             val handler =
               makeRequest(peer, message, responseMsgCode(message), toSerializable)(scheduler, responseClassTag(message))
             val newRequesters = requesters + (handler -> requester)
@@ -61,9 +62,8 @@ class PeersClient(
       peer: Peer,
       requestMsg: RequestMsg,
       responseMsgCode: Int,
-      toSerializable: RequestMsg => MessageSerializable)(
-      implicit scheduler: Scheduler,
-      classTag: ClassTag[ResponseMsg]): ActorRef =
+      toSerializable: RequestMsg => MessageSerializable
+  )(implicit scheduler: Scheduler, classTag: ClassTag[ResponseMsg]): ActorRef =
     context.actorOf(
       PeerRequestHandler.props[RequestMsg, ResponseMsg](
         peer = peer,
@@ -72,7 +72,8 @@ class PeersClient(
         peerEventBus = peerEventBus,
         requestMsg = requestMsg,
         responseMsgCode = responseMsgCode
-      )(classTag, scheduler, toSerializable))
+      )(classTag, scheduler, toSerializable)
+    )
 
   private def handleResponse[ResponseMsg <: ResponseMessage](requesters: Requesters, responseMsg: ResponseMsg): Unit = {
     val requestHandler = sender()
@@ -112,13 +113,14 @@ object PeersClient {
   case class Request[RequestMsg <: Message](
       message: RequestMsg,
       peerSelector: PeerSelector,
-      toSerializable: RequestMsg => MessageSerializable)
-      extends PeersClientMessage
+      toSerializable: RequestMsg => MessageSerializable
+  ) extends PeersClientMessage
 
   object Request {
 
-    def create[RequestMsg <: Message](message: RequestMsg, peerSelector: PeerSelector)(
-        implicit toSerializable: RequestMsg => MessageSerializable): Request[RequestMsg] =
+    def create[RequestMsg <: Message](message: RequestMsg, peerSelector: PeerSelector)(implicit
+        toSerializable: RequestMsg => MessageSerializable
+    ): Request[RequestMsg] =
       Request(message, peerSelector, toSerializable)
   }
   case object PrintStatus extends PeersClientMessage
@@ -133,9 +135,8 @@ object PeersClient {
 
   def bestPeer(peersToDownloadFrom: Map[Peer, PeerInfo]): Option[Peer] = {
     val peersToUse = peersToDownloadFrom
-      .collect {
-        case (ref, PeerInfo(_, totalDifficulty, true, _, _)) =>
-          (ref, totalDifficulty)
+      .collect { case (ref, PeerInfo(_, totalDifficulty, true, _, _)) =>
+        (ref, totalDifficulty)
       }
 
     if (peersToUse.nonEmpty) {

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

@@ -47,16 +47,14 @@ class BlockFetcher(
     peerEventBus ! Unsubscribe()
   }
 
-  private def idle(): Receive = handleCommonMessages(None) orElse {
-    case Start(importer, blockNr) =>
-      BlockFetcherState.initial(importer, blockNr) |> fetchBlocks
-      peerEventBus ! Subscribe(MessageClassifier(Set(NewBlock.code, NewBlockHashes.code), PeerSelector.AllPeers))
+  private def idle(): Receive = handleCommonMessages(None) orElse { case Start(importer, blockNr) =>
+    BlockFetcherState.initial(importer, blockNr) |> fetchBlocks
+    peerEventBus ! Subscribe(MessageClassifier(Set(NewBlock.code, NewBlockHashes.code), PeerSelector.AllPeers))
   }
 
-  def handleCommonMessages(state: Option[BlockFetcherState]): Receive = {
-    case PrintStatus =>
-      log.info("{}", state.map(_.status))
-      log.debug("{}", state.map(_.statusDetailed))
+  def handleCommonMessages(state: Option[BlockFetcherState]): Receive = { case PrintStatus =>
+    log.info("{}", state.map(_.status))
+    log.debug("{}", state.map(_.statusDetailed))
   }
 
   private def started(state: BlockFetcherState): Receive =
@@ -83,9 +81,8 @@ class BlockFetcher(
       fetchBlocks(newState)
     case InvalidateBlocksFrom(blockNr, reason, withBlacklist) =>
       val (blockProvider, newState) = state.invalidateBlocksFrom(blockNr, withBlacklist)
-
+      log.debug("Invalidate blocks from {}", blockNr)
       blockProvider.foreach(peersClient ! BlacklistPeer(_, reason))
-
       fetchBlocks(newState)
   }
 
@@ -125,7 +122,8 @@ class BlockFetcher(
           .asRight[String]
           .ensure(s"Empty response from peer $peer, blacklisting")(_.nonEmpty)
           .ensure("Fetched node state hash doesn't match requested one, blacklisting peer")(nodes =>
-            fetcher.hash == kec256(nodes.head))
+            fetcher.hash == kec256(nodes.head)
+          )
 
         validatedNode match {
           case Left(err) =>
@@ -141,6 +139,7 @@ class BlockFetcher(
 
   private def handleNewBlockMessages(state: BlockFetcherState): Receive = {
     case MessageFromPeer(NewBlockHashes(hashes), _) =>
+      log.debug("Received NewBlockHashes numbers {}", hashes.map(_.number).mkString(", "))
       val newState = state.validatedHashes(hashes) match {
         case Left(_) => state
         case Right(validHashes) => state.withPossibleNewTopAt(validHashes.lastOption.map(_.number))
@@ -155,19 +154,23 @@ class BlockFetcher(
 
       // we're on top, so we can pass block directly to importer
       if (newBlockNr == nextExpectedBlock && state.isOnTop) {
+        log.debug("Pass block directly to importer")
         val newState = state.withPeerForBlocks(peerId, Seq(newBlockNr)).withKnownTopAt(newBlockNr)
         state.importer ! OnTop
         state.importer ! ImportNewBlock(block, peerId)
         context become started(newState)
         // there are some blocks waiting for import but it seems that we reached top on fetch side so we can enqueue new block for import
       } else if (newBlockNr == nextExpectedBlock && !state.isFetching && state.waitingHeaders.isEmpty) {
+        log.debug("Enqueue new block for import")
         val newState = state.appendNewBlock(block, peerId)
         context become started(newState)
         // waiting for some bodies but we don't have this header yet - at least we can use new block header
       } else if (newBlockNr == state.nextToLastBlock && !state.isFetchingHeaders) {
+        log.debug("Waiting for bodies. Add only headers")
         state.appendHeaders(List(block.header)) |> fetchBlocks
         // we're far from top
       } else if (newBlockNr > nextExpectedBlock) {
+        log.debug("Far from top")
         val newState = state.withKnownTopAt(newBlockNr)
         fetchBlocks(newState)
       }
@@ -177,13 +180,13 @@ class BlockFetcher(
       fetchBlocks(newState)
   }
 
-  private def handlePickedBlocks(state: BlockFetcherState)(
-      pickResult: Option[(NonEmptyList[Block], BlockFetcherState)]): BlockFetcherState =
+  private def handlePickedBlocks(
+      state: BlockFetcherState
+  )(pickResult: Option[(NonEmptyList[Block], BlockFetcherState)]): BlockFetcherState =
     pickResult
-      .tap {
-        case (blocks, newState) =>
-          sender() ! PickedBlocks(blocks)
-          newState.importer ! (if (newState.isOnTop) OnTop else NotOnTop)
+      .tap { case (blocks, newState) =>
+        sender() ! PickedBlocks(blocks)
+        newState.importer ! (if (newState.isOnTop) OnTop else NotOnTop)
       }
       .fold(state)(_._2)
 
@@ -243,6 +246,7 @@ class BlockFetcher(
     makeRequest(Request.create(msg, BestPeer), RetryHeadersRequest)
       .flatMap {
         case Response(_, BlockHeaders(headers)) if headers.isEmpty =>
+          log.debug("Empty BlockHeaders response. Retry in {}", syncConfig.syncRetryInterval)
           Future.successful(RetryHeadersRequest).delayedBy(syncConfig.syncRetryInterval)
         case res => Future.successful(res)
       }

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

@@ -24,22 +24,37 @@ class RegularSync(
     context.actorOf(BlockFetcher.props(peersClient, peerEventBus, syncConfig, scheduler), "block-fetcher")
   val broadcaster: ActorRef = context.actorOf(
     BlockBroadcasterActor
-      .props(new BlockBroadcast(etcPeerManager, syncConfig), peerEventBus, etcPeerManager, syncConfig, scheduler), "block-broadcaster")
+      .props(new BlockBroadcast(etcPeerManager, syncConfig), peerEventBus, etcPeerManager, syncConfig, scheduler),
+    "block-broadcaster"
+  )
   val importer: ActorRef =
     context.actorOf(
       BlockImporter.props(fetcher, ledger, blockchain, syncConfig, ommersPool, broadcaster, pendingTransactionsManager),
-      "block-importer")
+      "block-importer"
+    )
 
   val printFetcherSchedule: Cancellable =
-    scheduler.scheduleWithFixedDelay(syncConfig.printStatusInterval, syncConfig.printStatusInterval, fetcher, BlockFetcher.PrintStatus)(context.dispatcher)
+    scheduler.scheduleWithFixedDelay(
+      syncConfig.printStatusInterval,
+      syncConfig.printStatusInterval,
+      fetcher,
+      BlockFetcher.PrintStatus
+    )(context.dispatcher)
   val printImporterSchedule: Cancellable =
-    scheduler.scheduleWithFixedDelay(syncConfig.printStatusInterval, syncConfig.printStatusInterval, importer, BlockImporter.PrintStatus)(context.dispatcher)
+    scheduler.scheduleWithFixedDelay(
+      syncConfig.printStatusInterval,
+      syncConfig.printStatusInterval,
+      importer,
+      BlockImporter.PrintStatus
+    )(context.dispatcher)
 
   override def receive: Receive = {
     case Start =>
       log.info("Starting regular sync")
       importer ! BlockImporter.Start
-    case MinedBlock(block) => importer ! BlockImporter.MinedBlock(block)
+    case MinedBlock(block) =>
+      log.info(s"Block mined [number = {}, hash = {}]", block.number, block.header.hashAsHexString)
+      importer ! BlockImporter.MinedBlock(block)
   }
 
   override def supervisorStrategy: SupervisorStrategy = AllForOneStrategy()(SupervisorStrategy.defaultDecider)
@@ -73,7 +88,9 @@ object RegularSync {
         syncConfig,
         ommersPool,
         pendingTransactionsManager,
-        scheduler))
+        scheduler
+      )
+    )
 
   sealed trait NewRegularSyncMsg
   case object Start extends NewRegularSyncMsg

src/main/scala/io/iohk/ethereum/consensus/ethash/EthashMiner.scala

@@ -13,7 +13,7 @@ import io.iohk.ethereum.jsonrpc.EthService
 import io.iohk.ethereum.jsonrpc.EthService.SubmitHashRateRequest
 import io.iohk.ethereum.nodebuilder.Node
 import io.iohk.ethereum.utils.BigIntExtensionMethods._
-import io.iohk.ethereum.utils.ByteUtils
+import io.iohk.ethereum.utils.{ByteStringUtils, ByteUtils}
 import java.io.{File, FileInputStream, FileOutputStream}
 import org.bouncycastle.util.encoders.Hex
 import scala.concurrent.ExecutionContext.Implicits.global
@@ -71,7 +71,6 @@ class EthashMiner(
             }
             res.getOrElse(generateDagAndSaveToFile(epoch, dagNumHashes, seed))
           }
-
         currentEpoch = Some(epoch)
         currentEpochDag = Some(dag)
         currentEpochDagSize = Some(dagSize)
@@ -90,13 +89,15 @@ class EthashMiner(
         ethService.submitHashRate(SubmitHashRateRequest(hashRate, ByteString("mantis-miner")))
         mineResult match {
           case MiningSuccessful(_, pow, nonce) =>
+            log.info(
+              s"Mining successful with ${ByteStringUtils.hash2string(pow.mixHash)} and nonce ${ByteStringUtils.hash2string(nonce)}"
+            )
             syncController ! RegularSync.MinedBlock(
               block.copy(header = block.header.copy(nonce = nonce, mixHash = pow.mixHash))
             )
-          case _ => // nothing
+          case _ => log.info("Mining unsuccessful")
         }
         self ! ProcessMining
-
       case Failure(ex) =>
         log.error(ex, "Unable to get block for mining")
         context.system.scheduler.scheduleOnce(10.seconds, self, ProcessMining)
@@ -111,7 +112,6 @@ class EthashMiner(
 
   private def generateDagAndSaveToFile(epoch: Long, dagNumHashes: Int, seed: ByteString): Array[Array[Int]] = {
     // scalastyle:off magic.number
-
     val file = dagFile(seed)
     if (file.exists()) file.delete()
     file.getParentFile.mkdirs()

src/main/scala/io/iohk/ethereum/faucet/FaucetApi.scala

@@ -1,7 +1,6 @@
 package io.iohk.ethereum.faucet
 
 import java.time.Clock
-
 import akka.http.scaladsl.model.{RemoteAddress, StatusCodes}
 import akka.http.scaladsl.server.Route
 import akka.http.scaladsl.server.Directives._
@@ -12,10 +11,9 @@ import com.twitter.util.LruMap
 import io.iohk.ethereum.domain.{Address, Transaction}
 import io.iohk.ethereum.keystore.KeyStore
 import io.iohk.ethereum.mallet.service.RpcClient
-import io.iohk.ethereum.utils.Logger
+import io.iohk.ethereum.utils.{ByteStringUtils, Logger}
 import io.iohk.ethereum.rlp
 import io.iohk.ethereum.network.p2p.messages.CommonMessages.SignedTransactions.SignedTransactionEnc
-import org.bouncycastle.util.encoders.Hex
 
 class FaucetApi(rpcClient: RpcClient, keyStore: KeyStore, config: FaucetConfig, clock: Clock = Clock.systemUTC())
     extends Logger {
@@ -25,7 +23,6 @@ class FaucetApi(rpcClient: RpcClient, keyStore: KeyStore, config: FaucetConfig,
   private val wallet = keyStore.unlockAccount(config.walletAddress, config.walletPassword) match {
     case Right(w) => w
     case Left(err) =>
-      log.info("accounts " + keyStore.listAccounts().right.get.mkString(", "))
       throw new RuntimeException(s"Cannot unlock wallet for use in faucet (${config.walletAddress}), because of $err")
   }
 
@@ -54,8 +51,9 @@ class FaucetApi(rpcClient: RpcClient, keyStore: KeyStore, config: FaucetConfig,
 
       res match {
         case Right(txId) =>
-          log.info(s"Sending ${config.txValue} ETH to $targetAddress in tx: $txId. Requested by $clientAddr")
-          complete(StatusCodes.OK, s"0x${Hex.toHexString(txId.toArray[Byte])}")
+          val txIdHex = s"0x${ByteStringUtils.hash2string(txId)}"
+          log.info(s"Sending ${config.txValue} ETH to $targetAddress in tx: $txIdHex. Requested by $clientAddr")
+          complete(StatusCodes.OK, txIdHex)
 
         case Left(err) =>
           log.error(s"An error occurred while using faucet: $err")