Merge pull request #601 from alajko/copy_backup_node

Archival node setup

cmd/harmony/main.go

@@ -10,17 +10,16 @@ import (
 	"strconv"
 	"time"
 
-	"github.com/harmony-one/bls/ffi/go/bls"
-
-	"github.com/harmony-one/harmony/internal/utils/contract"
-
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/log"
+
+	"github.com/harmony-one/bls/ffi/go/bls"
 	"github.com/harmony-one/harmony/consensus"
 	"github.com/harmony-one/harmony/drand"
 	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
 	"github.com/harmony-one/harmony/internal/profiler"
 	"github.com/harmony-one/harmony/internal/utils"
+	"github.com/harmony-one/harmony/internal/utils/contract"
 	"github.com/harmony-one/harmony/node"
 	"github.com/harmony-one/harmony/p2p"
 	"github.com/harmony-one/harmony/p2p/p2pimpl"
@@ -85,6 +84,8 @@ var (
 	keyFile = flag.String("key", "./.hmykey", "the private key file of the harmony node")
 	// isBeacon indicates this node is a beacon chain node
 	isBeacon = flag.Bool("is_beacon", false, "true means this node is a beacon chain node")
+	// isArchival indicates this node is an archival node that will save and archive current blockchain
+	isArchival = flag.Bool("is_archival", false, "true means this node is a archival node")
 	//isNewNode indicates this node is a new node
 	isNewNode    = flag.Bool("is_newnode", false, "true means this node is a new node")
 	accountIndex = flag.Int("account_index", 0, "the index of the staking account to use")
@@ -171,6 +172,8 @@ func createGlobalConfig() *nodeconfig.ConfigType {
 	if *isLeader {
 		nodeConfig.StringRole = "leader"
 		nodeConfig.Leader = nodeConfig.SelfPeer
+	} else if *isArchival {
+		nodeConfig.StringRole = "archival"
 	} else {
 		nodeConfig.StringRole = "validator"
 	}
@@ -188,6 +191,13 @@ func createGlobalConfig() *nodeconfig.ConfigType {
 	return nodeConfig
 }
 
+func setupArchivalNode(nodeConfig *nodeconfig.ConfigType) (*node.Node, *nodeconfig.ConfigType) {
+	currentNode := node.New(nodeConfig.Host, &consensus.Consensus{ShardID: uint32(0)}, nil)
+	currentNode.NodeConfig.SetRole(nodeconfig.ArchivalNode)
+	currentNode.AddBeaconChainDatabase(nodeConfig.BeaconDB)
+	return currentNode, nodeConfig
+}
+
 func setUpConsensusAndNode(nodeConfig *nodeconfig.ConfigType) (*consensus.Consensus, *node.Node) {
 	// Consensus object.
 	// TODO: consensus object shouldn't start here
@@ -251,11 +261,14 @@ func main() {
 	flag.Parse()
 
 	initSetup()
+	var currentNode *node.Node
+	var consensus *consensus.Consensus
 	nodeConfig := createGlobalConfig()
-
-	// Init logging.
+	if *isArchival {
+		currentNode, nodeConfig = setupArchivalNode(nodeConfig)
 		loggingInit(*logFolder, nodeConfig.StringRole, *ip, *port, *onlyLogTps)
-
+		go currentNode.DoBeaconSyncing()
+	} else {
 		// Start Profiler for leader if profile argument is on
 		if nodeConfig.StringRole == "leader" && (*profile || *metricsReportURL != "") {
 			prof := profiler.GetProfiler()
@@ -264,15 +277,15 @@ func main() {
 				prof.Start()
 			}
 		}
-
-	consensus, currentNode := setUpConsensusAndNode(nodeConfig)
-
+		consensus, currentNode = setUpConsensusAndNode(nodeConfig)
 		if consensus.IsLeader {
 			go currentNode.SendPongMessage()
 		}
-
-	log.Info("New Harmony Node ====", "Role", currentNode.NodeConfig.Role(), "multiaddress", fmt.Sprintf("/ip4/%s/tcp/%s/p2p/%s", *ip, *port, nodeConfig.Host.GetID().Pretty()))
+		// Init logging.
+		loggingInit(*logFolder, nodeConfig.StringRole, *ip, *port, *onlyLogTps)
 		go currentNode.SupportSyncing()
+	}
+	log.Info("New Harmony Node ====", "Role", currentNode.NodeConfig.Role(), "multiaddress", fmt.Sprintf("/ip4/%s/tcp/%s/p2p/%s", *ip, *port, nodeConfig.Host.GetID().Pretty()))
 	currentNode.ServiceManagerSetup()
 	currentNode.RunServices()
 	currentNode.StartServer()

internal/configs/node/config.go

@@ -27,7 +27,7 @@ const (
 	NewNode
 	ClientNode
 	WalletNode
-	BackupNode
+	ArchivalNode
 )
 
 func (role Role) String() string {
@@ -48,8 +48,8 @@ func (role Role) String() string {
 		return "ClientNode"
 	case WalletNode:
 		return "WalletNode"
-	case BackupNode:
-		return "BackupNode"
+	case ArchivalNode:
+		return "ArchivalNode"
 	}
 	return "Unknown"
 }
@@ -69,6 +69,7 @@ type ConfigType struct {
 	isClient   bool        // whether this node is a client node, such as wallet/txgen
 	isBeacon   bool        // whether this node is a beacon node or not
 	isLeader   bool        // whether this node is a leader or not
+	isArchival bool        // whether this node is a archival node. archival node backups all blockchain information.
 	shardID    uint32      // shardID of this node
 	role       Role        // Role of the node
 
@@ -142,6 +143,11 @@ func (conf *ConfigType) SetIsLeader(b bool) {
 	conf.isLeader = b
 }
 
+// SetIsArchival set the isArchival configuration
+func (conf *ConfigType) SetIsArchival(b bool) {
+	conf.isArchival = b
+}
+
 // SetShardID set the shardID
 func (conf *ConfigType) SetShardID(s uint32) {
 	conf.shardID = s
@@ -182,6 +188,11 @@ func (conf *ConfigType) IsLeader() bool {
 	return conf.isLeader
 }
 
+// IsArchival returns the isArchival configuration
+func (conf *ConfigType) IsArchival() bool {
+	return conf.isArchival
+}
+
 // ShardID returns the shardID
 func (conf *ConfigType) ShardID() uint32 {
 	return conf.shardID

node/node.go

@@ -247,7 +247,6 @@ func New(host p2p.Host, consensusObj *consensus.Consensus, db ethdb.Database) *N
 		node.BlockChannel = make(chan *types.Block)
 		node.ConfirmedBlockChannel = make(chan *types.Block)
 		node.BeaconBlockChannel = make(chan *types.Block)
-
 		node.TxPool = core.NewTxPool(core.DefaultTxPoolConfig, params.TestChainConfig, chain)
 		node.Worker = worker.New(params.TestChainConfig, chain, node.Consensus, pki.GetAddressFromPublicKey(node.SelfPeer.ConsensusPubKey), node.Consensus.ShardID)
 
@@ -431,12 +430,12 @@ func (node *Node) initNodeConfiguration(isBeacon bool, isClient bool) (service.N
 	nodeConfig.Actions[p2p.GroupIDBeaconClient] = p2p.ActionStart
 
 	var err error
-	if !isBeacon {
-		node.groupReceiver, err = node.host.GroupReceiver(p2p.GroupIDBeaconClient)
-	} else {
+	if isBeacon {
 		node.groupReceiver, err = node.host.GroupReceiver(p2p.GroupIDBeacon)
 		node.clientReceiver, err = node.host.GroupReceiver(p2p.GroupIDBeaconClient)
 		node.NodeConfig.SetClientGroupID(p2p.GroupIDBeaconClient)
+	} else {
+		node.groupReceiver, err = node.host.GroupReceiver(p2p.GroupIDBeaconClient)
 	}
 
 	if err != nil {

node/node_handler.go

@@ -142,12 +142,14 @@ func (node *Node) messageHandler(content []byte, sender string) {
 				} else {
 					// for non-beaconchain node, subscribe to beacon block broadcast
 					role := node.NodeConfig.Role()
-					if proto_node.BlockMessageType(msgPayload[0]) == proto_node.Sync && (role == nodeconfig.ShardValidator || role == nodeconfig.ShardLeader || role == nodeconfig.NewNode) {
+					if proto_node.BlockMessageType(msgPayload[0]) == proto_node.Sync && (role == nodeconfig.ShardValidator || role == nodeconfig.ShardLeader || role == nodeconfig.NewNode || role == nodeconfig.ArchivalNode) {
+						utils.GetLogInstance().Info("Block being handled by block channel", "self peer", node.SelfPeer)
 						for _, block := range blocks {
 							node.BeaconBlockChannel <- block
 						}
 					}
 					if node.Client != nil && node.Client.UpdateBlocks != nil && blocks != nil {
+						utils.GetLogInstance().Info("Block being handled by client by", "self peer", node.SelfPeer)
 						node.Client.UpdateBlocks(blocks)
 					}
 				}
@@ -340,7 +342,7 @@ func (node *Node) AddNewBlock(newBlock *types.Block) {
 	if err != nil {
 		utils.GetLogInstance().Debug("Error adding new block to blockchain", "blockNum", blockNum, "Error", err)
 	} else {
-		utils.GetLogInstance().Info("adding new block to blockchain", "blockNum", blockNum)
+		utils.GetLogInstance().Info("adding new block to blockchain", "blockNum", blockNum, "by node", node.SelfPeer)
 	}
 }
 

node/node_syncing.go

@@ -41,7 +41,6 @@ func (node *Node) getNeighborPeers(neighbor *sync.Map) []p2p.Peer {
 		res = append(res, v.(p2p.Peer))
 		return true
 	})
-
 	removeID := -1
 	for i := range res {
 		if res[i].Port == node.SelfPeer.Port {
@@ -52,12 +51,12 @@ func (node *Node) getNeighborPeers(neighbor *sync.Map) []p2p.Peer {
 	if removeID != -1 {
 		res = append(res[:removeID], res[removeID+1:]...)
 	}
-	utils.GetLogInstance().Debug("GetSyncingPeers: ", "res", res, "self", node.SelfPeer)
 	return res
 }
 
 // GetBeaconSyncingPeers returns a list of peers for beaconchain syncing
 func (node *Node) GetBeaconSyncingPeers() []p2p.Peer {
+
 	return node.getNeighborPeers(&node.BeaconNeighbors)
 }
 
@@ -79,6 +78,7 @@ func (node *Node) DoBeaconSyncing() {
 			startHash := node.beaconChain.CurrentBlock().Hash()
 			node.beaconSync.AddLastMileBlock(beaconBlock)
 			node.beaconSync.StartStateSync(startHash[:], node.beaconChain, node.BeaconWorker, true)
+			utils.GetLogInstance().Debug("[SYNC] STARTING BEACON SYNC")
 		}
 	}
 }

node/service_setup.go

@@ -104,8 +104,8 @@ func (node *Node) setupForClientNode() {
 	node.serviceManager.RegisterService(service.NetworkInfo, networkinfo.New(node.host, p2p.GroupIDBeacon, chanPeer, nil))
 }
 
-func (node *Node) setupForBackupNode(isBeacon bool) {
-	nodeConfig, chanPeer := node.initNodeConfiguration(isBeacon, false)
+func (node *Node) setupForArchivalNode() {
+	nodeConfig, chanPeer := node.initNodeConfiguration(false, false)
 	// Register peer discovery service.
 	node.serviceManager.RegisterService(service.PeerDiscovery, discovery.New(node.host, nodeConfig, chanPeer, node.AddBeaconPeer))
 	// Register networkinfo service. "0" is the beacon shard ID
@@ -130,8 +130,8 @@ func (node *Node) ServiceManagerSetup() {
 		node.setupForNewNode()
 	case nodeconfig.ClientNode:
 		node.setupForClientNode()
-	case nodeconfig.BackupNode:
-		node.setupForBackupNode(true)
+	case nodeconfig.ArchivalNode:
+		node.setupForArchivalNode()
 	}
 	node.serviceManager.SetupServiceMessageChan(node.serviceMessageChan)
 }

test/configs/oneshard1.txt

@@ -9,4 +9,6 @@
 127.0.0.1 9008 validator 0
 127.0.0.1 9009 validator 0
 127.0.0.1 9010 validator 0
+127.0.0.1 9011 archival 0
 127.0.0.1 19999 client 0
+

test/deploy.sh

@@ -148,6 +148,10 @@ while IFS='' read -r line || [[ -n "$line" ]]; do
      echo "launching validator ..."
      $DRYRUN $ROOT/bin/harmony -ip $ip -port $port -log_folder $log_folder $DB -min_peers $MIN $HMY_OPT2 $HMY_OPT3 -key /tmp/$ip-$port.key 2>&1 | tee -a $LOG_FILE &
   fi
+  if [ "$mode" == "archival" ]; then
+      echo "launching archival node ... wait"
+      $DRYRUN $ROOT/bin/harmony -ip $ip -port $port -log_folder $log_folder $DB $HMY_OPT2 -key /tmp/$ip-$port.key -is_archival  2>&1 | tee -a $LOG_FILE &
+  fi
   sleep 0.5
   if [[ "$mode" == "newnode" && "$SYNC" == "true" ]]; then
      (( NUM_NN += 10 ))