Merge pull request #1285 from flicker-harmony/pr_metrics_service

Metrics Service

api/service/config.go

@@ -11,13 +11,15 @@ import (
 // cyclic imports
 type NodeConfig struct {
 	// The three groupID design, please refer to https://github.com/harmony-one/harmony/blob/master/node/node.md#libp2p-integration
-	Beacon       p2p.GroupID                    // the beacon group ID
+	Beacon          p2p.GroupID                    // the beacon group ID
-	ShardGroupID p2p.GroupID                    // the group ID of the shard
+	ShardGroupID    p2p.GroupID                    // the group ID of the shard
-	Client       p2p.GroupID                    // the client group ID of the shard
+	Client          p2p.GroupID                    // the client group ID of the shard
-	IsClient     bool                           // whether this node is a client node, such as wallet/txgen
+	IsClient        bool                           // whether this node is a client node, such as wallet/txgen
-	IsBeacon     bool                           // whether this node is a beacon node or not
+	IsBeacon        bool                           // whether this node is a beacon node or not
-	ShardID      uint32                         // shardID of this node
+	ShardID         uint32                         // shardID of this node
-	Actions      map[p2p.GroupID]p2p.ActionType // actions on the groups
+	Actions         map[p2p.GroupID]p2p.ActionType // actions on the groups
+	PushgatewayIP   string                         // prometheus pushgateway ip
+	PushgatewayPort string                         // prometheus pushgateway port
 }
 
 // GroupIDShards is a map of ShardGroupID ID

api/service/manager.go

@@ -27,6 +27,7 @@ const (
 	ClientSupport
 	SupportExplorer
 	Consensus
+	Metrics
 	Randomness
 	BlockProposal
 	NetworkInfo
@@ -45,6 +46,8 @@ func (t Type) String() string {
 		return "SupportExplorer"
 	case ClientSupport:
 		return "ClientSupport"
+	case Metrics:
+		return "Metrics"
 	case Consensus:
 		return "Consensus"
 	case Randomness:

api/service/metrics/service.go

@@ -0,0 +1,240 @@
+package metrics
+
+import (
+	"fmt"
+	"math/big"
+	"net"
+	"net/http"
+	"strconv"
+
+	"github.com/ethereum/go-ethereum/rpc"
+	msg_pb "github.com/harmony-one/harmony/api/proto/message"
+	"github.com/harmony-one/harmony/internal/utils"
+	"github.com/harmony-one/harmony/p2p"
+	libp2p_peer "github.com/libp2p/go-libp2p-peer"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/prometheus/client_golang/prometheus/push"
+)
+
+// Constants for metrics service.
+const (
+	BalanceScale                 int = 18
+	BalancePrecision             int = 13
+	ConnectionsNumberPush        int = 0
+	BlockHeightPush              int = 1
+	NodeBalancePush              int = 2
+	LastConsensusPush            int = 3
+	BlockRewardPush              int = 4
+	metricsServicePortDifference     = 2000
+)
+
+// Service is the struct for metrics service.
+type Service struct {
+	BlsPublicKey    string
+	IP              string
+	Port            string
+	PushgatewayIP   string
+	PushgatewayPort string
+	GetNodeIDs      func() []libp2p_peer.ID
+	storage         *Storage
+	pusher          *push.Pusher
+	messageChan     chan *msg_pb.Message
+}
+
+// init vars for prometheus
+var (
+	curBlockHeight       = uint64(0)
+	curBlocks            = uint64(0)
+	curBalance           = big.NewInt(0)
+	curConnectionsNumber = 0
+	lastBlockReward      = big.NewInt(0)
+	lastConsensusTime    = int64(0)
+	metricsPush          = make(chan int)
+	blockHeightCounter   = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "block_height",
+		Help: "Get current block height.",
+	})
+	blocksAcceptedGauge = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "blocks_accepted",
+		Help: "Get accepted blocks.",
+	})
+	connectionsNumberGauge = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "connections_number",
+		Help: "Get current connections number for a node.",
+	})
+	nodeBalanceCounter = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "node_balance",
+		Help: "Get current node balance.",
+	})
+	lastConsensusGauge = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "last_consensus",
+		Help: "Get last consensus time.",
+	})
+	blockRewardGauge = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "block_reward",
+		Help: "Get last block reward.",
+	})
+)
+
+// ConnectionsLog struct for connections stats for prometheus
+type ConnectionsLog struct {
+	Time              int
+	ConnectionsNumber int
+}
+
+// ConnectionsStatsHTTP struct for returning all connections logs
+type ConnectionsStatsHTTP struct {
+	ConnectionsLogs []ConnectionsLog
+}
+
+// New returns metrics service.
+func New(selfPeer *p2p.Peer, blsPublicKey, pushgatewayIP, pushgatewayPort string, GetNodeIDs func() []libp2p_peer.ID) *Service {
+	return &Service{
+		BlsPublicKey:    blsPublicKey,
+		IP:              selfPeer.IP,
+		Port:            selfPeer.Port,
+		PushgatewayIP:   pushgatewayIP,
+		PushgatewayPort: pushgatewayPort,
+		GetNodeIDs:      GetNodeIDs,
+	}
+}
+
+// StartService starts metrics service.
+func (s *Service) StartService() {
+	utils.Logger().Info().Msg("Starting metrics service.")
+	s.Run()
+}
+
+// StopService shutdowns metrics service.
+func (s *Service) StopService() {
+	utils.Logger().Info().Msg("Shutting down metrics service.")
+	metricsPush <- -1
+}
+
+// GetMetricsServicePort returns the port serving metrics service dashboard. This port is metricsServicePortDifference less than the node port.
+func GetMetricsServicePort(nodePort string) string {
+	if port, err := strconv.Atoi(nodePort); err == nil {
+		return fmt.Sprintf("%d", port-metricsServicePortDifference)
+	}
+	utils.Logger().Error().Msg("Error on parsing.")
+	return ""
+}
+
+// Run is to run http serving metrics service.
+func (s *Service) Run() {
+	// Init local storage for metrics.
+	s.storage = GetStorageInstance(s.IP, s.Port, true)
+	// Init address.
+	addr := net.JoinHostPort("", GetMetricsServicePort(s.Port))
+
+	registry := prometheus.NewRegistry()
+	registry.MustRegister(blockHeightCounter, connectionsNumberGauge, nodeBalanceCounter, lastConsensusGauge, blockRewardGauge, blocksAcceptedGauge)
+
+	s.pusher = push.New("http://"+s.PushgatewayIP+":"+s.PushgatewayPort, "node_metrics").Gatherer(registry).Grouping("instance", s.IP+":"+s.Port).Grouping("bls_key", s.BlsPublicKey)
+	go s.PushMetrics()
+	// Pull metrics http server
+	utils.Logger().Info().Str("port", GetMetricsServicePort(s.Port)).Msg("Listening.")
+	go func() {
+		http.Handle("/node_metrics", promhttp.Handler())
+		if err := http.ListenAndServe(addr, nil); err != nil {
+			utils.Logger().Warn().Err(err).Msg("http.ListenAndServe()")
+		}
+	}()
+	return
+}
+
+// FormatBalance formats big.Int balance with precision.
+func FormatBalance(balance *big.Int) float64 {
+	stringBalance := balance.String()
+	if len(stringBalance) < BalanceScale {
+		return 0.0
+	}
+	if len(stringBalance) == BalanceScale {
+		stringBalance = "0." + stringBalance[len(stringBalance)-BalanceScale:len(stringBalance)-BalancePrecision]
+	} else {
+		stringBalance = stringBalance[:len(stringBalance)-BalanceScale] + "." + stringBalance[len(stringBalance)-BalanceScale:len(stringBalance)-BalancePrecision]
+	}
+	if res, err := strconv.ParseFloat(stringBalance, 64); err == nil {
+		return res
+	}
+	return 0.0
+}
+
+// UpdateBlockHeight updates block height.
+func UpdateBlockHeight(blockHeight uint64) {
+	blockHeightCounter.Add(float64(blockHeight) - float64(curBlockHeight))
+	blocksAcceptedGauge.Set(float64(blockHeight) - float64(curBlockHeight))
+	curBlockHeight = blockHeight
+	metricsPush <- BlockHeightPush
+}
+
+// UpdateNodeBalance updates node balance.
+func UpdateNodeBalance(balance *big.Int) {
+	nodeBalanceCounter.Add(FormatBalance(balance) - FormatBalance(curBalance))
+	curBalance = balance
+	metricsPush <- NodeBalancePush
+}
+
+// UpdateBlockReward updates block reward.
+func UpdateBlockReward(blockReward *big.Int) {
+	blockRewardGauge.Set(FormatBalance(blockReward))
+	lastBlockReward = blockReward
+	metricsPush <- BlockRewardPush
+}
+
+// UpdateLastConsensus updates last consensus time.
+func UpdateLastConsensus(consensusTime int64) {
+	lastConsensusGauge.Set(float64(consensusTime))
+	lastConsensusTime = consensusTime
+	metricsPush <- LastConsensusPush
+}
+
+// UpdateConnectionsNumber updates connections number.
+func UpdateConnectionsNumber(connectionsNumber int) {
+	connectionsNumberGauge.Set(float64(connectionsNumber))
+	curConnectionsNumber = connectionsNumber
+	metricsPush <- ConnectionsNumberPush
+}
+
+// PushMetrics pushes metrics updates to prometheus pushgateway.
+func (s *Service) PushMetrics() {
+	for metricType := range metricsPush {
+		if metricType == -1 {
+			break
+		}
+		if err := s.pusher.Add(); err != nil {
+			utils.Logger().Error().Err(err).Msg("Could not push to a prometheus pushgateway.")
+		}
+		/*switch metricType {
+		case ConnectionsNumberPush:
+			s.storage.Dump(curConnectionsNumber, ConnectionsNumberPrefix)
+		case BlockHeightPush:
+			fmt.Println("LOL")
+			s.storage.Dump(curBlockHeight, BlockHeightPrefix)
+			s.storage.Dump(curBlocks, BlocksPrefix)
+		case BlockRewardPush:
+			s.storage.Dump(lastBlockReward, BlockHeightPrefix)
+		case NodeBalancePush:
+			s.storage.Dump(curBalance, BalancePrefix)
+		case LastConsensusPush:
+			s.storage.Dump(lastConsensusTime, ConsensusTimePrefix)
+		}*/
+	}
+	return
+}
+
+// NotifyService notify service
+func (s *Service) NotifyService(params map[string]interface{}) {
+	return
+}
+
+// SetMessageChan sets up message channel to service.
+func (s *Service) SetMessageChan(messageChan chan *msg_pb.Message) {
+	s.messageChan = messageChan
+}
+
+// APIs for the services.
+func (s *Service) APIs() []rpc.API {
+	return nil
+}

api/service/metrics/storage.go

@@ -0,0 +1,103 @@
+package metrics
+
+import (
+	"fmt"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/harmony-one/harmony/internal/utils"
+)
+
+// Constants for storage.
+const (
+	BalancePrefix           = "bap"
+	BlockHeightPrefix       = "bhp"
+	BlocksPrefix            = "bp"
+	BlockRewardPrefix       = "brp"
+	ConnectionsNumberPrefix = "cnp"
+	ConsensusTimePrefix     = "ltp"
+	TransactionsPrefix      = "tsp"
+)
+
+// GetKey returns key by prefix and pushed time momemnt.
+func GetKey(prefix string, moment int64) string {
+	return fmt.Sprintf("%s_%d", prefix, moment)
+}
+
+// storage instance
+var storage *Storage
+var onceMetrics sync.Once
+
+// Storage storage dump the block info into leveldb.
+type Storage struct {
+	db *ethdb.LDBDatabase
+}
+
+// GetStorageInstance returns attack model by using singleton pattern.
+func GetStorageInstance(ip, port string, remove bool) *Storage {
+	onceMetrics.Do(func() {
+		storage = &Storage{}
+		storage.Init(ip, port, remove)
+	})
+	return storage
+}
+
+// Init initializes storage.
+func (storage *Storage) Init(ip, port string, remove bool) {
+	dbFileName := "/.hmy/db-metrics-" + ip + "-" + port
+	var err error
+	if remove {
+		var err = os.RemoveAll(dbFileName)
+		if err != nil {
+			utils.Logger().Error().Err(err).Msg("Failed to remove existing database files.")
+		}
+	}
+	if storage.db, err = ethdb.NewLDBDatabase(dbFileName, 0, 0); err != nil {
+		utils.Logger().Error().Err(err).Msg("Failed to create new database.")
+	}
+}
+
+// GetDB returns the LDBDatabase of the storage.
+func (storage *Storage) GetDB() *ethdb.LDBDatabase {
+	return storage.db
+}
+
+// Dump data into lvdb by value and prefix.
+func (storage *Storage) Dump(value interface{}, prefix string) error {
+	currentTime := time.Now().Unix()
+	utils.Logger().Info().Msgf("Store %s %v at time %d", prefix, value, currentTime)
+	if storage.db == nil {
+	}
+	batch := storage.db.NewBatch()
+	// Update database.
+	if err := batch.Put([]byte(GetKey(prefix, currentTime)), []byte(fmt.Sprintf("%v", value))); err != nil {
+		utils.Logger().Warn().Err(err).Msgf("Cannot batch %s.", prefix)
+		return err
+	}
+	if err := batch.Write(); err != nil {
+		utils.Logger().Warn().Err(err).Msg("Cannot write batch.")
+		return err
+	}
+	return nil
+}
+
+// Read returns data list of a particular metric by since, until, prefix, interface.
+func (storage *Storage) Read(since, until int64, prefix string, varType interface{}) []interface{} {
+	dataList := make([]interface{}, 0)
+	for i := since; i <= until; i++ {
+		data, err := storage.db.Get([]byte(GetKey(prefix, i)))
+		if err != nil {
+			continue
+		}
+		decodedData := varType
+		if rlp.DecodeBytes(data, decodedData) != nil {
+			utils.Logger().Error().Msg("Error on getting data from db.")
+			os.Exit(1)
+		}
+		dataList = append(dataList, decodedData)
+	}
+	return dataList
+}

cmd/harmony/main.go

@@ -118,6 +118,10 @@ var (
 
 	// Disable view change.
 	disableViewChange = flag.Bool("disable_view_change", false, "Do not propose view change (testing only)")
+
+	// pushgateway ip and port
+	pushgatewayIP   = flag.String("pushgateway_ip", "grafana.harmony.one", "metrics view ip")
+	pushgatewayPort = flag.String("pushgateway_port", "9091", "metrics view port")
 )
 
 func initSetup() {
@@ -243,6 +247,9 @@ func createGlobalConfig() *nodeconfig.ConfigType {
 		panic(fmt.Sprintf("invalid network type: %s", *networkType))
 	}
 
+	nodeConfig.SetPushgatewayIP(*pushgatewayIP)
+	nodeConfig.SetPushgatewayPort(*pushgatewayPort)
+
 	// P2p private key is used for secure message transfer between p2p nodes.
 	nodeConfig.P2pPriKey, _, err = utils.LoadKeyFromFile(*keyFile)
 	if err != nil {
@@ -302,6 +309,10 @@ func setupConsensusAndNode(nodeConfig *nodeconfig.ConfigType) *node.Node {
 	// TODO: refactor the creation of blockchain out of node.New()
 	currentConsensus.ChainReader = currentNode.Blockchain()
 
+	// Set up prometheus pushgateway for metrics monitoring serivce.
+	currentNode.NodeConfig.SetPushgatewayIP(nodeConfig.PushgatewayIP)
+	currentNode.NodeConfig.SetPushgatewayPort(nodeConfig.PushgatewayPort)
+
 	if *isExplorer {
 		currentNode.NodeConfig.SetRole(nodeconfig.ExplorerNode)
 		currentNode.NodeConfig.SetShardGroupID(p2p.NewGroupIDByShardID(p2p.ShardID(*shardID)))
@@ -316,7 +327,6 @@ func setupConsensusAndNode(nodeConfig *nodeconfig.ConfigType) *node.Node {
 			currentNode.NodeConfig.SetShardGroupID(p2p.NewGroupIDByShardID(p2p.ShardID(nodeConfig.ShardID)))
 			currentNode.NodeConfig.SetClientGroupID(p2p.NewClientGroupIDByShardID(p2p.ShardID(nodeConfig.ShardID)))
 		}
-
 	}
 	currentNode.NodeConfig.ConsensusPubKey = nodeConfig.ConsensusPubKey
 	currentNode.NodeConfig.ConsensusPriKey = nodeConfig.ConsensusPriKey

consensus/consensus.go

@@ -159,6 +159,9 @@ type Consensus struct {
 
 	// If true, this consensus will not propose view change.
 	disableViewChange bool
+
+	// last node block reward for metrics
+	lastBlockReward *big.Int
 }
 
 // SetCommitDelay sets the commit message delay.  If set to non-zero,
@@ -226,6 +229,11 @@ func (consensus *Consensus) RewardThreshold() int {
 	return len(consensus.PublicKeys) * 9 / 10
 }
 
+// GetBlockReward returns last node block reward
+func (consensus *Consensus) GetBlockReward() *big.Int {
+	return consensus.lastBlockReward
+}
+
 // StakeInfoFinder finds the staking account for the given consensus key.
 type StakeInfoFinder interface {
 	// FindStakeInfoByNodeKey returns a list of staking information matching
@@ -281,6 +289,7 @@ func New(host p2p.Host, ShardID uint32, leader p2p.Peer, blsPriKey *bls.SecretKe
 	consensus.commitFinishChan = make(chan uint64)
 
 	consensus.ReadySignal = make(chan struct{})
+	consensus.lastBlockReward = big.NewInt(0)
 
 	// channel for receiving newly generated VDF
 	consensus.RndChannel = make(chan [vdfAndSeedSize]byte)
@@ -294,36 +303,37 @@ func New(host p2p.Host, ShardID uint32, leader p2p.Peer, blsPriKey *bls.SecretKe
 // accumulateRewards credits the coinbase of the given block with the mining
 // reward. The total reward consists of the static block reward and rewards for
 // included uncles. The coinbase of each uncle block is also rewarded.
+// Returns node block reward or error.
 func accumulateRewards(
-	bc consensus_engine.ChainReader, state *state.DB, header *types.Header,
+	bc consensus_engine.ChainReader, state *state.DB, header *types.Header, nodeAddress common.Address,
-) error {
+) (*big.Int, error) {
 	blockNum := header.Number.Uint64()
 	if blockNum == 0 {
 		// Epoch block has no parent to reward.
-		return nil
+		return nil, nil
 	}
 	// TODO ek – retrieving by parent number (blockNum - 1) doesn't work,
 	//  while it is okay with hash.  Sounds like DB inconsistency.
 	//  Figure out why.
 	parentHeader := bc.GetHeaderByHash(header.ParentHash)
 	if parentHeader == nil {
-		return ctxerror.New("cannot find parent block header in DB",
+		return nil, ctxerror.New("cannot find parent block header in DB",
 			"parentHash", header.ParentHash)
 	}
 	if parentHeader.Number.Cmp(common.Big0) == 0 {
 		// Parent is an epoch block,
 		// which is not signed in the usual manner therefore rewards nothing.
-		return nil
+		return nil, nil
 	}
 	parentShardState, err := bc.ReadShardState(parentHeader.Epoch)
 	if err != nil {
-		return ctxerror.New("cannot read shard state",
+		return nil, ctxerror.New("cannot read shard state",
 			"epoch", parentHeader.Epoch,
 		).WithCause(err)
 	}
 	parentCommittee := parentShardState.FindCommitteeByID(parentHeader.ShardID)
 	if parentCommittee == nil {
-		return ctxerror.New("cannot find shard in the shard state",
+		return nil, ctxerror.New("cannot find shard in the shard state",
 			"parentBlockNumber", parentHeader.Number,
 			"shardID", parentHeader.ShardID,
 		)
@@ -333,24 +343,24 @@ func accumulateRewards(
 		committerKey := new(bls.PublicKey)
 		err := member.BlsPublicKey.ToLibBLSPublicKey(committerKey)
 		if err != nil {
-			return ctxerror.New("cannot convert BLS public key",
+			return nil, ctxerror.New("cannot convert BLS public key",
 				"blsPublicKey", member.BlsPublicKey).WithCause(err)
 		}
 		committerKeys = append(committerKeys, committerKey)
 	}
 	mask, err := bls_cosi.NewMask(committerKeys, nil)
 	if err != nil {
-		return ctxerror.New("cannot create group sig mask").WithCause(err)
+		return nil, ctxerror.New("cannot create group sig mask").WithCause(err)
 	}
 	if err := mask.SetMask(header.LastCommitBitmap); err != nil {
-		return ctxerror.New("cannot set group sig mask bits").WithCause(err)
+		return nil, ctxerror.New("cannot set group sig mask bits").WithCause(err)
 	}
 	totalAmount := big.NewInt(0)
 	var accounts []common.Address
 	signers := []string{}
 	for idx, member := range parentCommittee.NodeList {
 		if signed, err := mask.IndexEnabled(idx); err != nil {
-			return ctxerror.New("cannot check for committer bit",
+			return nil, ctxerror.New("cannot check for committer bit",
 				"committerIndex", idx,
 			).WithCause(err)
 		} else if signed {
@@ -359,11 +369,15 @@ func accumulateRewards(
 	}
 	numAccounts := big.NewInt(int64(len(accounts)))
 	last := new(big.Int)
+	nodeReward := big.NewInt(0)
 	for i, account := range accounts {
 		cur := new(big.Int)
 		cur.Mul(BlockReward, big.NewInt(int64(i+1))).Div(cur, numAccounts)
 		diff := new(big.Int).Sub(cur, last)
 		signers = append(signers, common2.MustAddressToBech32(account))
+		if account == nodeAddress {
+			nodeReward = diff
+		}
 		state.AddBalance(account, diff)
 		totalAmount = new(big.Int).Add(totalAmount, diff)
 		last = cur
@@ -373,7 +387,7 @@ func accumulateRewards(
 		Str("TotalAmount", totalAmount.String()).
 		Strs("Signers", signers).
 		Msg("[Block Reward] Successfully paid out block reward")
-	return nil
+	return nodeReward, nil
 }
 
 // GenesisStakeInfoFinder is a stake info finder implementation using only

consensus/consensus_service.go

@@ -287,9 +287,11 @@ func (consensus *Consensus) VerifySeal(chain consensus_engine.ChainReader, heade
 func (consensus *Consensus) Finalize(chain consensus_engine.ChainReader, header *types.Header, state *state.DB, txs []*types.Transaction, receipts []*types.Receipt) (*types.Block, error) {
 	// Accumulate any block and uncle rewards and commit the final state root
 	// Header seems complete, assemble into a block and return
-	if err := accumulateRewards(chain, state, header); err != nil {
+	blockReward, err := accumulateRewards(chain, state, header, consensus.SelfAddress)
+	if err != nil {
 		return nil, ctxerror.New("cannot pay block reward").WithCause(err)
 	}
+	consensus.lastBlockReward = blockReward
 	header.Root = state.IntermediateRoot(false)
 	return types.NewBlock(header, txs, receipts), nil
 }

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/Workiva/go-datastructures v1.0.50
 	github.com/allegro/bigcache v1.2.1 // indirect
 	github.com/aristanetworks/goarista v0.0.0-20190607111240-52c2a7864a08 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/btcsuite/btcutil v0.0.0-20190207003914-4c204d697803
 	github.com/cespare/cp v1.1.1
 	github.com/davecgh/go-spew v1.1.1
@@ -14,7 +15,6 @@ require (
 	github.com/ethereum/go-ethereum v1.8.27
 	github.com/fjl/memsize v0.0.0-20180929194037-2a09253e352a
 	github.com/garslo/gogen v0.0.0-20170307003452-d6ebae628c7c // indirect
-	github.com/go-stack/stack v1.8.0 // indirect
 	github.com/golang/mock v1.3.1
 	github.com/golang/protobuf v1.3.0
 	github.com/golangci/golangci-lint v1.17.1
@@ -43,6 +43,10 @@ require (
 	github.com/natefinch/lumberjack v2.0.0+incompatible
 	github.com/pborman/uuid v1.2.0
 	github.com/pkg/errors v0.8.1
+	github.com/prometheus/client_golang v0.9.2
+	github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90 // indirect
+	github.com/prometheus/common v0.4.1 // indirect
+	github.com/prometheus/procfs v0.0.3 // indirect
 	github.com/rjeczalik/notify v0.9.2
 	github.com/rs/cors v1.6.0 // indirect
 	github.com/rs/zerolog v1.14.3
@@ -54,6 +58,8 @@ require (
 	github.com/whyrusleeping/go-logging v0.0.0-20170515211332-0457bb6b88fc
 	golang.org/x/crypto v0.0.0-20190513172903-22d7a77e9e5f
 	golang.org/x/lint v0.0.0-20190409202823-959b441ac422
+	golang.org/x/net v0.0.0-20190613194153-d28f0bde5980 // indirect
+	golang.org/x/sys v0.0.0-20190801041406-cbf593c0f2f3 // indirect
 	golang.org/x/tools v0.0.0-20190521203540-521d6ed310dd
 	google.golang.org/appengine v1.4.0 // indirect
 	google.golang.org/grpc v1.21.1

internal/configs/node/config.go

@@ -78,6 +78,8 @@ type ConfigType struct {
 	Port     string      // Port of the node.
 	IP       string      // IP of the node.
 
+	PushgatewayIP   string // metrics pushgateway prometheus ip
+	PushgatewayPort string // metrics pushgateway prometheus port
 	StringRole      string
 	Host            p2p.Host
 	StakingPriKey   *ecdsa.PrivateKey
@@ -166,6 +168,26 @@ func (conf *ConfigType) SetRole(r Role) {
 	conf.role = r
 }
 
+// SetPushgatewayIP set the pushgateway ip
+func (conf *ConfigType) SetPushgatewayIP(ip string) {
+	conf.PushgatewayIP = ip
+}
+
+// SetPushgatewayPort set the pushgateway port
+func (conf *ConfigType) SetPushgatewayPort(port string) {
+	conf.PushgatewayPort = port
+}
+
+// GetPushgatewayIP get the pushgateway ip
+func (conf *ConfigType) GetPushgatewayIP() string {
+	return conf.PushgatewayIP
+}
+
+// GetPushgatewayPort get the pushgateway port
+func (conf *ConfigType) GetPushgatewayPort() string {
+	return conf.PushgatewayPort
+}
+
 // GetBeaconGroupID returns the groupID for beacon group
 func (conf *ConfigType) GetBeaconGroupID() p2p.GroupID {
 	return conf.beacon

internal/utils/singleton.go

@@ -31,7 +31,7 @@ var (
 
 	// ZeroLog
 	zeroLogger      *zerolog.Logger
-	zeroLoggerLevel zerolog.Level = zerolog.Disabled
+	zeroLoggerLevel = zerolog.Disabled
 )
 
 // SetLogContext used to print out loggings of node with port and ip.

node/node.go

@@ -196,6 +196,9 @@ type Node struct {
 	isFirstTime bool // the node was started with a fresh database
 	// How long in second the leader needs to wait to propose a new block.
 	BlockPeriod time.Duration
+
+	// last time consensus reached for metrics
+	lastConsensusTime int64
 }
 
 // Blockchain returns the blockchain for the node's current shard.
@@ -384,6 +387,9 @@ func New(host p2p.Host, consensusObj *consensus.Consensus, chainDBFactory shardc
 	// FIXME (leo): we use beacon client topic as the global topic for now
 	go node.ReceiveGlobalMessage()
 
+	// start the goroutine to collect metrics
+	go node.CollectMetrics()
+
 	// Setup initial state of syncing.
 	node.peerRegistrationRecord = make(map[string]*syncConfig)
 
@@ -471,10 +477,12 @@ func (node *Node) initNodeConfiguration() (service.NodeConfig, chan p2p.Peer) {
 	chanPeer := make(chan p2p.Peer)
 
 	nodeConfig := service.NodeConfig{
-		IsClient:     node.NodeConfig.IsClient(),
+		PushgatewayIP:   node.NodeConfig.GetPushgatewayIP(),
-		Beacon:       p2p.GroupIDBeacon,
+		PushgatewayPort: node.NodeConfig.GetPushgatewayPort(),
-		ShardGroupID: node.NodeConfig.GetShardGroupID(),
+		IsClient:        node.NodeConfig.IsClient(),
-		Actions:      make(map[p2p.GroupID]p2p.ActionType),
+		Beacon:          p2p.GroupIDBeacon,
+		ShardGroupID:    node.NodeConfig.GetShardGroupID(),
+		Actions:         make(map[p2p.GroupID]p2p.ActionType),
 	}
 
 	if nodeConfig.IsClient {

node/node_handler.go

@@ -409,6 +409,8 @@ func (node *Node) validateNewShardState(block *types.Block, stakeInfo *map[commo
 // 1. add the new block to blockchain
 // 2. [leader] send new block to the client
 func (node *Node) PostConsensusProcessing(newBlock *types.Block) {
+	// Update last consensus time for metrics
+	node.lastConsensusTime = time.Now().Unix()
 	if node.Consensus.PubKey.IsEqual(node.Consensus.LeaderPubKey) {
 		node.BroadcastNewBlock(newBlock)
 	} else {

node/node_metrics.go

@@ -0,0 +1,74 @@
+package node
+
+import (
+	"math/big"
+	"time"
+
+	metrics "github.com/harmony-one/harmony/api/service/metrics"
+	"github.com/harmony-one/harmony/internal/utils"
+)
+
+// UpdateBlockHeightForMetrics updates block height for metrics service.
+func (node *Node) UpdateBlockHeightForMetrics(prevBlockHeight uint64) uint64 {
+	curBlock := node.Blockchain().CurrentBlock()
+	curBlockHeight := curBlock.NumberU64()
+	if curBlockHeight == prevBlockHeight {
+		return prevBlockHeight
+	}
+	utils.Logger().Info().Msgf("Updating metrics block height %d", curBlockHeight)
+	metrics.UpdateBlockHeight(curBlockHeight)
+	blockReward := node.Consensus.GetBlockReward()
+	if blockReward != nil {
+		utils.Logger().Info().Msgf("Updating metrics block reward %d", blockReward.Uint64())
+		metrics.UpdateBlockReward(blockReward)
+	}
+	return curBlockHeight
+}
+
+// UpdateConnectionsNumberForMetrics uppdates connections number for metrics service.
+func (node *Node) UpdateConnectionsNumberForMetrics(prevNumPeers int) int {
+	curNumPeers := node.numPeers
+	if curNumPeers == prevNumPeers {
+		return prevNumPeers
+	}
+	utils.Logger().Info().Msgf("Updating metrics connections number %d", curNumPeers)
+	metrics.UpdateConnectionsNumber(curNumPeers)
+	return curNumPeers
+}
+
+// UpdateBalanceForMetrics uppdates node balance for metrics service.
+func (node *Node) UpdateBalanceForMetrics(prevBalance *big.Int) *big.Int {
+	curBalance, err := node.GetBalanceOfAddress(node.Consensus.SelfAddress)
+	if err != nil || curBalance.Cmp(prevBalance) == 0 {
+		return prevBalance
+	}
+	utils.Logger().Info().Msgf("Updating metrics node balance %d", curBalance.Uint64())
+	metrics.UpdateNodeBalance(curBalance)
+	return curBalance
+}
+
+// UpdateLastConsensusTimeForMetrics uppdates last consensus reached time for metrics service.
+func (node *Node) UpdateLastConsensusTimeForMetrics(prevLastConsensusTime int64) int64 {
+	lastConsensusTime := node.lastConsensusTime
+	if lastConsensusTime == prevLastConsensusTime {
+		return prevLastConsensusTime
+	}
+	utils.Logger().Info().Msgf("Updating metrics last consensus time reached %d", lastConsensusTime)
+	metrics.UpdateLastConsensus(lastConsensusTime)
+	return lastConsensusTime
+}
+
+// CollectMetrics collects metrics: block height, connections number, node balance, block reward, last consensus, accepted blocks.
+func (node *Node) CollectMetrics() {
+	utils.Logger().Info().Msg("[Metrics Service] Update metrics")
+	prevNumPeers := 0
+	prevBlockHeight := uint64(0)
+	prevBalance := big.NewInt(0)
+	prevLastConsensusTime := int64(0)
+	for range time.Tick(1000 * time.Millisecond) {
+		prevBlockHeight = node.UpdateBlockHeightForMetrics(prevBlockHeight)
+		prevNumPeers = node.UpdateConnectionsNumberForMetrics(prevNumPeers)
+		prevBalance = node.UpdateBalanceForMetrics(prevBalance)
+		prevLastConsensusTime = node.UpdateLastConsensusTimeForMetrics(prevLastConsensusTime)
+	}
+}

node/service_setup.go

@@ -8,6 +8,7 @@ import (
 	"github.com/harmony-one/harmony/api/service/consensus"
 	"github.com/harmony-one/harmony/api/service/discovery"
 	"github.com/harmony-one/harmony/api/service/explorer"
+	"github.com/harmony-one/harmony/api/service/metrics"
 	"github.com/harmony-one/harmony/api/service/networkinfo"
 	"github.com/harmony-one/harmony/api/service/staking"
 	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
@@ -28,6 +29,8 @@ func (node *Node) setupForValidator() {
 	node.serviceManager.RegisterService(service.BlockProposal, blockproposal.New(node.Consensus.ReadySignal, node.WaitForConsensusReadyv2))
 	// Register client support service.
 	node.serviceManager.RegisterService(service.ClientSupport, clientsupport.New(node.Blockchain().State, node.CallFaucetContract, node.getDeployedStakingContract, node.SelfPeer.IP, node.SelfPeer.Port))
+	// Register new metrics service
+	node.serviceManager.RegisterService(service.Metrics, metrics.New(&node.SelfPeer, node.NodeConfig.ConsensusPubKey.SerializeToHexStr(), node.NodeConfig.GetPushgatewayIP(), node.NodeConfig.GetPushgatewayPort(), node.Consensus.GetNodeIDs))
 
 	// Register randomness service
 	// TODO: Disable drand. Currently drand isn't functioning but we want to compeletely turn it off for full protection.
@@ -47,6 +50,8 @@ func (node *Node) setupForNewNode() {
 	node.serviceManager.RegisterService(service.PeerDiscovery, discovery.New(node.host, nodeConfig, chanPeer, node.AddBeaconPeer))
 	// Register networkinfo service. "0" is the beacon shard ID
 	node.serviceManager.RegisterService(service.NetworkInfo, networkinfo.New(node.host, node.NodeConfig.GetBeaconGroupID(), chanPeer, nil))
+	// Register new metrics service
+	node.serviceManager.RegisterService(service.Metrics, metrics.New(&node.SelfPeer, node.NodeConfig.ConsensusPubKey.SerializeToHexStr(), node.NodeConfig.GetPushgatewayIP(), node.NodeConfig.GetPushgatewayPort(), node.Consensus.GetNodeIDs))
 
 	// TODO: how to restart networkinfo and discovery service after receiving shard id info from beacon chain?
 }
@@ -58,6 +63,8 @@ func (node *Node) setupForClientNode() {
 	node.serviceManager.RegisterService(service.PeerDiscovery, discovery.New(node.host, nodeConfig, chanPeer, node.AddBeaconPeer))
 	// Register networkinfo service. "0" is the beacon shard ID
 	node.serviceManager.RegisterService(service.NetworkInfo, networkinfo.New(node.host, p2p.GroupIDBeacon, chanPeer, nil))
+	// Register new metrics service
+	node.serviceManager.RegisterService(service.Metrics, metrics.New(&node.SelfPeer, node.NodeConfig.ConsensusPubKey.SerializeToHexStr(), node.NodeConfig.GetPushgatewayIP(), node.NodeConfig.GetPushgatewayPort(), node.Consensus.GetNodeIDs))
 }
 
 func (node *Node) setupForExplorerNode() {