fix

[bytom/vapor.git] / toolbar / precog / monitor / stats.go

diff --git a/toolbar/precog/monitor/stats.go b/toolbar/precog/monitor/stats.go
index 6a6d08f..13c8818 100644 (file)
--- a/toolbar/precog/monitor/stats.go
+++ b/toolbar/precog/monitor/stats.go
@@ -3,23 +3,21 @@ package monitor
 import (
        "database/sql"
        "fmt"
+       "net"
+       "strconv"
        "time"
 
        "github.com/jinzhu/gorm"
        log "github.com/sirupsen/logrus"
 
-       "github.com/vapor/crypto/ed25519/chainkd"
        "github.com/vapor/netsync/peers"
+       "github.com/vapor/p2p"
        "github.com/vapor/toolbar/precog/common"
        "github.com/vapor/toolbar/precog/config"
        "github.com/vapor/toolbar/precog/database/orm"
 )
 
 func (m *monitor) upsertNode(node *config.Node) error {
-       if node.XPub != nil {
-               node.PublicKey = fmt.Sprintf("%v", node.XPub.PublicKey().String())
-       }
-
        ormNode := &orm.Node{
                IP:   node.IP,
                Port: node.Port,
@@ -31,41 +29,48 @@ func (m *monitor) upsertNode(node *config.Node) error {
        ormNode.PublicKey = node.PublicKey
        if node.XPub != nil {
                ormNode.Xpub = node.XPub.String()
+               ormNode.PublicKey = fmt.Sprintf("%v", node.XPub.PublicKey().String())
        }
        return m.db.Save(ormNode).Error
 }
 
-// TODO: maybe return connected nodes here for checkStatus
-func (m *monitor) processDialResults() error {
+func parseRemoteAddr(remoteAddr string) (string, uint16, error) {
+       host, portStr, err := net.SplitHostPort(remoteAddr)
+       if err != nil {
+               return "", 0, err
+       }
+
+       port, err := strconv.Atoi(portStr)
+       if err != nil {
+               return "", 0, err
+       }
+
+       return host, uint16(port), nil
+}
+
+func (m *monitor) processDialResults(peerList []*p2p.Peer) error {
        var ormNodes []*orm.Node
        if err := m.db.Model(&orm.Node{}).Find(&ormNodes).Error; err != nil {
                return err
        }
 
-       publicKeyMap := make(map[string]*orm.Node, len(ormNodes))
+       addressMap := make(map[string]*orm.Node, len(ormNodes))
        for _, ormNode := range ormNodes {
-               publicKeyMap[ormNode.PublicKey] = ormNode
+               addressMap[fmt.Sprintf("%s:%d", ormNode.IP, ormNode.Port)] = ormNode
        }
 
        connMap := make(map[string]bool, len(ormNodes))
        // connected peers
-       for _, peer := range m.sw.GetPeers().List() {
-               xPub := &chainkd.XPub{}
-               if err := xPub.UnmarshalText([]byte(peer.Key)); err != nil {
-                       log.WithFields(log.Fields{"xpub": peer.Key}).Error("unmarshal xpub")
-                       continue
-               }
-
-               publicKey := xPub.PublicKey().String()
-               connMap[publicKey] = true
-               if err := m.processConnectedPeer(publicKeyMap[publicKey]); err != nil {
-                       log.WithFields(log.Fields{"peer publicKey": publicKey, "err": err}).Error("processConnectedPeer")
+       for _, peer := range peerList {
+               connMap[peer.RemoteAddr] = true
+               if err := m.processConnectedPeer(addressMap[peer.RemoteAddr]); err != nil {
+                       log.WithFields(log.Fields{"peer remoteAddr": peer.RemoteAddr, "err": err}).Error("processConnectedPeer")
                }
        }
 
        // offline peers
        for _, ormNode := range ormNodes {
-               if _, ok := connMap[ormNode.PublicKey]; ok {
+               if _, ok := connMap[fmt.Sprintf("%s:%d", ormNode.IP, ormNode.Port)]; ok {
                        continue
                }
 
@@ -110,18 +115,18 @@ func (m *monitor) processPeerInfos(peerInfos []*peers.PeerInfo) {
 }
 
 func (m *monitor) processPeerInfo(dbTx *gorm.DB, peerInfo *peers.PeerInfo) error {
-       xPub := &chainkd.XPub{}
-       if err := xPub.UnmarshalText([]byte(peerInfo.ID)); err != nil {
+       ip, port, err := parseRemoteAddr(peerInfo.RemoteAddr)
+       if err != nil {
                return err
        }
 
-       ormNode := &orm.Node{}
-       if err := dbTx.Model(&orm.Node{}).Where(&orm.Node{PublicKey: xPub.PublicKey().String()}).First(ormNode).Error; err != nil {
+       ormNode := &orm.Node{IP: ip, Port: uint16(port)}
+       if err := dbTx.Where(ormNode).First(ormNode).Error; err != nil {
                return err
        }
 
        if ormNode.Status == common.NodeOfflineStatus {
-               return fmt.Errorf("node %s status error", ormNode.PublicKey)
+               return fmt.Errorf("node %s:%d status error", ormNode.IP, ormNode.Port)
        }
 
        log.WithFields(log.Fields{"ping": peerInfo.Ping}).Debug("peerInfo")
@@ -135,23 +140,28 @@ func (m *monitor) processPeerInfo(dbTx *gorm.DB, peerInfo *peers.PeerInfo) error
        var ormNodeLivenesses []*orm.NodeLiveness
        if err := dbTx.Preload("Node").Model(&orm.NodeLiveness{}).
                Where("node_id = ? AND updated_at >= ?", ormNode.ID, yesterday).
-               Order(fmt.Sprintf("created_at %s", "DESC")).
-               Find(&ormNodeLivenesses).Error; err != nil {
+               Order(fmt.Sprintf("created_at %s", "DESC")).Find(&ormNodeLivenesses).Error; err != nil {
                return err
        }
 
        // update latest liveness
        latestLiveness := ormNodeLivenesses[0]
-       lantencyMS := ping.Nanoseconds() / 1000
-       if lantencyMS != 0 {
-               ormNode.AvgLantencyMS = sql.NullInt64{
-                       Int64: (ormNode.AvgLantencyMS.Int64*int64(latestLiveness.PongTimes) + lantencyMS) / int64(latestLiveness.PongTimes+1),
+       rttMS := ping.Nanoseconds() / 1000000
+       if rttMS > 0 && rttMS < 2000 {
+               ormNode.Status = common.NodeHealthyStatus
+       } else if rttMS > 2000 {
+               ormNode.Status = common.NodeCongestedStatus
+       }
+       if rttMS != 0 {
+               ormNode.AvgRttMS = sql.NullInt64{
+                       Int64: (ormNode.AvgRttMS.Int64*int64(latestLiveness.PongTimes) + rttMS) / int64(latestLiveness.PongTimes+1),
                        Valid: true,
                }
        }
        latestLiveness.PongTimes += 1
        if peerInfo.Height != 0 {
                latestLiveness.BestHeight = peerInfo.Height
+               ormNode.BestHeight = peerInfo.Height
        }
        if err := dbTx.Save(latestLiveness).Error; err != nil {
                return err
@@ -167,12 +177,8 @@ func (m *monitor) processPeerInfo(dbTx *gorm.DB, peerInfo *peers.PeerInfo) error
 
                total += ormNodeLiveness.UpdatedAt.Sub(ormNodeLiveness.CreatedAt)
        }
-
-       return dbTx.Model(&orm.Node{}).Where(&orm.Node{PublicKey: xPub.PublicKey().String()}).
-               UpdateColumn(&orm.Node{
-                       Alias:                    peerInfo.Moniker,
-                       Xpub:                     peerInfo.ID,
-                       BestHeight:               peerInfo.Height,
-                       LatestDailyUptimeMinutes: uint64(total.Minutes()),
-               }).First(ormNode).Error
+       ormNode.LatestDailyUptimeMinutes = uint64(total.Minutes())
+       ormNode.Alias = peerInfo.Moniker
+       ormNode.Xpub = peerInfo.ID
+       return dbTx.Save(ormNode).Error
 }