NodeBalanceCounter -> NodeBalanceGauge, remove unnecessary stuff in metrics service

pull/1309/head
flicker-harmony 5 years ago
parent aa419fe23c
commit cf90273acd
  1. 43
      api/service/metrics/service.go
  2. 11
      node/node_metrics.go

@ -4,17 +4,13 @@ import (
"fmt" "fmt"
"math" "math"
"math/big" "math/big"
"net"
"net/http"
"strconv" "strconv"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
msg_pb "github.com/harmony-one/harmony/api/proto/message" msg_pb "github.com/harmony-one/harmony/api/proto/message"
"github.com/harmony-one/harmony/internal/utils" "github.com/harmony-one/harmony/internal/utils"
"github.com/harmony-one/harmony/p2p" "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"
"github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/prometheus/client_golang/prometheus/push" "github.com/prometheus/client_golang/prometheus/push"
) )
@ -39,7 +35,6 @@ type Service struct {
Port string Port string
PushgatewayIP string PushgatewayIP string
PushgatewayPort string PushgatewayPort string
GetNodeIDs func() []libp2p_peer.ID
storage *Storage storage *Storage
pusher *push.Pusher pusher *push.Pusher
messageChan chan *msg_pb.Message messageChan chan *msg_pb.Message
@ -76,7 +71,7 @@ var (
Name: "connections_number", Name: "connections_number",
Help: "Get current connections number for a node.", Help: "Get current connections number for a node.",
}) })
nodeBalanceCounter = prometheus.NewCounter(prometheus.CounterOpts{ nodeBalanceGauge = prometheus.NewGauge(prometheus.GaugeOpts{
Name: "node_balance", Name: "node_balance",
Help: "Get current node balance.", Help: "Get current node balance.",
}) })
@ -90,26 +85,14 @@ var (
}) })
) )
// 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. // New returns metrics service.
func New(selfPeer *p2p.Peer, blsPublicKey, pushgatewayIP, pushgatewayPort string, GetNodeIDs func() []libp2p_peer.ID) *Service { func New(selfPeer *p2p.Peer, blsPublicKey, pushgatewayIP, pushgatewayPort string) *Service {
return &Service{ return &Service{
BlsPublicKey: blsPublicKey, BlsPublicKey: blsPublicKey,
IP: selfPeer.IP, IP: selfPeer.IP,
Port: selfPeer.Port, Port: selfPeer.Port,
PushgatewayIP: pushgatewayIP, PushgatewayIP: pushgatewayIP,
PushgatewayPort: pushgatewayPort, PushgatewayPort: pushgatewayPort,
GetNodeIDs: GetNodeIDs,
} }
} }
@ -138,23 +121,11 @@ func GetMetricsServicePort(nodePort string) string {
func (s *Service) Run() { func (s *Service) Run() {
// Init local storage for metrics. // Init local storage for metrics.
s.storage = GetStorageInstance(s.IP, s.Port, true) s.storage = GetStorageInstance(s.IP, s.Port, true)
// Init address.
addr := net.JoinHostPort("", GetMetricsServicePort(s.Port))
registry := prometheus.NewRegistry() registry := prometheus.NewRegistry()
registry.MustRegister(blockHeightCounter, connectionsNumberGauge, nodeBalanceCounter, lastConsensusGauge, blockRewardGauge, blocksAcceptedGauge, txPoolGauge, isLeaderGauge) registry.MustRegister(blockHeightCounter, connectionsNumberGauge, nodeBalanceGauge, lastConsensusGauge, blockRewardGauge, blocksAcceptedGauge, txPoolGauge, isLeaderGauge)
s.pusher = push.New("http://"+s.PushgatewayIP+":"+s.PushgatewayPort, "node_metrics").Gatherer(registry).Grouping("instance", s.IP+":"+s.Port).Grouping("bls_key", s.BlsPublicKey) 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() 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. // FormatBalance formats big.Int balance with precision.
@ -174,7 +145,7 @@ func UpdateBlockHeight(blockHeight uint64) {
// UpdateNodeBalance updates node balance. // UpdateNodeBalance updates node balance.
func UpdateNodeBalance(balance *big.Int) { func UpdateNodeBalance(balance *big.Int) {
nodeBalanceCounter.Add(FormatBalance(balance) - FormatBalance(curBalance)) nodeBalanceGauge.Set(FormatBalance(balance))
curBalance = balance curBalance = balance
metricsPush <- NodeBalancePush metricsPush <- NodeBalancePush
} }
@ -226,8 +197,8 @@ func (s *Service) PushMetrics() {
} }
if err := s.pusher.Add(); err != nil { if err := s.pusher.Add(); err != nil {
utils.Logger().Error().Err(err).Msg("Could not push to a prometheus pushgateway.") utils.Logger().Error().Err(err).Msg("Could not push to a prometheus pushgateway.")
// Dump metrics to db if couldn't push to prometheus // No dump for now, not necessarily for metrics and consumes memory, doesn't restore from db anyway.
switch metricType { /* switch metricType {
case ConnectionsNumberPush: case ConnectionsNumberPush:
s.storage.Dump(curConnectionsNumber, ConnectionsNumberPrefix) s.storage.Dump(curConnectionsNumber, ConnectionsNumberPrefix)
case BlockHeightPush: case BlockHeightPush:
@ -243,7 +214,7 @@ func (s *Service) PushMetrics() {
s.storage.Dump(curTxPoolSize, TxPoolPrefix) s.storage.Dump(curTxPoolSize, TxPoolPrefix)
case IsLeaderPush: case IsLeaderPush:
s.storage.Dump(curIsLeader, IsLeaderPrefix) s.storage.Dump(curIsLeader, IsLeaderPrefix)
} }*/
} }
} }
return return

@ -1,7 +1,6 @@
package node package node
import ( import (
"math/big"
"time" "time"
metrics "github.com/harmony-one/harmony/api/service/metrics" metrics "github.com/harmony-one/harmony/api/service/metrics"
@ -43,14 +42,13 @@ func (node *Node) UpdateTxPoolSizeForMetrics(txPoolSize uint64) {
} }
// UpdateBalanceForMetrics uppdates node balance for metrics service. // UpdateBalanceForMetrics uppdates node balance for metrics service.
func (node *Node) UpdateBalanceForMetrics(prevBalance *big.Int) *big.Int { func (node *Node) UpdateBalanceForMetrics() {
curBalance, err := node.GetBalanceOfAddress(node.Consensus.SelfAddress) curBalance, err := node.GetBalanceOfAddress(node.Consensus.SelfAddress)
if err != nil || curBalance.Cmp(prevBalance) == 0 { if err != nil {
return prevBalance return
} }
utils.Logger().Info().Msgf("Updating metrics node balance %d", curBalance.Uint64()) utils.Logger().Info().Msgf("Updating metrics node balance %d", curBalance.Uint64())
metrics.UpdateNodeBalance(curBalance) metrics.UpdateNodeBalance(curBalance)
return curBalance
} }
// UpdateLastConsensusTimeForMetrics uppdates last consensus reached time for metrics service. // UpdateLastConsensusTimeForMetrics uppdates last consensus reached time for metrics service.
@ -80,13 +78,12 @@ func (node *Node) CollectMetrics() {
utils.Logger().Info().Msg("[Metrics Service] Update metrics") utils.Logger().Info().Msg("[Metrics Service] Update metrics")
prevNumPeers := 0 prevNumPeers := 0
prevBlockHeight := uint64(0) prevBlockHeight := uint64(0)
prevBalance := big.NewInt(0)
prevLastConsensusTime := int64(0) prevLastConsensusTime := int64(0)
for range time.Tick(100 * time.Millisecond) { for range time.Tick(100 * time.Millisecond) {
prevBlockHeight = node.UpdateBlockHeightForMetrics(prevBlockHeight) prevBlockHeight = node.UpdateBlockHeightForMetrics(prevBlockHeight)
prevNumPeers = node.UpdateConnectionsNumberForMetrics(prevNumPeers) prevNumPeers = node.UpdateConnectionsNumberForMetrics(prevNumPeers)
prevBalance = node.UpdateBalanceForMetrics(prevBalance)
prevLastConsensusTime = node.UpdateLastConsensusTimeForMetrics(prevLastConsensusTime) prevLastConsensusTime = node.UpdateLastConsensusTimeForMetrics(prevLastConsensusTime)
node.UpdateBalanceForMetrics()
node.UpdateTxPoolSizeForMetrics(node.TxPool.GetTxPoolSize()) node.UpdateTxPoolSizeForMetrics(node.TxPool.GetTxPoolSize())
node.UpdateIsLeaderForMetrics() node.UpdateIsLeaderForMetrics()
} }

Loading…
Cancel
Save