toolbar/precog/monitor/monitor.go

   1 package monitor
   2
   3 import (
   4         // "encoding/binary"
   5         // "encoding/hex"
   6         // "io/ioutil"
   7         "fmt"
   8         "net"
   9         "os"
  10         "strings"
  11         "time"
  12
  13         "github.com/jinzhu/gorm"
  14         log "github.com/sirupsen/logrus"
  15         // dbm "github.com/vapor/database/leveldb"
  16
  17         vaporCfg "github.com/vapor/config"
  18         "github.com/vapor/crypto/ed25519/chainkd"
  19         "github.com/vapor/p2p"
  20         // conn "github.com/vapor/p2p/connection"
  21         // "github.com/vapor/consensus"
  22         // "github.com/vapor/crypto/sha3pool"
  23         "github.com/vapor/p2p/discover/dht"
  24         "github.com/vapor/p2p/discover/mdns"
  25         "github.com/vapor/p2p/signlib"
  26         "github.com/vapor/toolbar/precog/config"
  27         "github.com/vapor/toolbar/precog/database/orm"
  28 )
  29
  30 var (
  31         nodesToDiscv = 150
  32         discvFreqSec = 1
  33 )
  34
  35 type monitor struct {
  36         cfg     *config.Config
  37         db      *gorm.DB
  38         nodeCfg *vaporCfg.Config
  39         sw      *p2p.Switch
  40         discvCh chan *dht.Node
  41         privKey chainkd.XPrv
  42 }
  43
  44 func NewMonitor(cfg *config.Config, db *gorm.DB) *monitor {
  45         //TODO: for test
  46         cfg.CheckFreqSeconds = 1
  47
  48         nodeCfg := &vaporCfg.Config{
  49                 BaseConfig: vaporCfg.DefaultBaseConfig(),
  50                 P2P:        vaporCfg.DefaultP2PConfig(),
  51                 Federation: vaporCfg.DefaultFederationConfig(),
  52         }
  53         nodeCfg.DBPath = "vapor_precog_data"
  54         nodeCfg.ChainID = "mainnet"
  55         discvCh := make(chan *dht.Node)
  56         privKey, err := signlib.NewPrivKey()
  57         if err != nil {
  58                 log.Fatal(err)
  59         }
  60
  61         return &monitor{
  62                 cfg:     cfg,
  63                 db:      db,
  64                 nodeCfg: nodeCfg,
  65                 discvCh: discvCh,
  66                 privKey: privKey.(chainkd.XPrv),
  67         }
  68 }
  69
  70 func (m *monitor) Run() {
  71         defer os.RemoveAll(m.nodeCfg.DBPath)
  72
  73         var seeds []string
  74         for _, node := range m.cfg.Nodes {
  75                 seeds = append(seeds, fmt.Sprintf("%s:%d", node.Host, node.Port))
  76                 if err := m.upSertNode(&node); err != nil {
  77                         log.Error(err)
  78                 }
  79         }
  80         m.nodeCfg.P2P.Seeds = strings.Join(seeds, ",")
  81         if err := m.makeSwitch(); err != nil {
  82                 log.Fatal(err)
  83         }
  84
  85         go m.discoveryRoutine()
  86         go m.collectDiscoveredNodes()
  87         go m.connectNodesRoutine()
  88 }
  89
  90 // create or update: https://github.com/jinzhu/gorm/issues/1307
  91 func (m *monitor) upSertNode(node *config.Node) error {
  92         if node.XPub != nil {
  93                 node.PublicKey = fmt.Sprintf("%v", node.XPub.PublicKey().String())
  94         }
  95
  96         ormNode := &orm.Node{PublicKey: node.PublicKey}
  97         if err := m.db.Where(&orm.Node{PublicKey: node.PublicKey}).First(ormNode).Error; err != nil && err != gorm.ErrRecordNotFound {
  98                 return err
  99         }
 100
 101         if node.Alias != "" {
 102                 ormNode.Alias = node.Alias
 103         }
 104         if node.XPub != nil {
 105                 ormNode.Xpub = node.XPub.String()
 106         }
 107         ormNode.Host = node.Host
 108         ormNode.Port = node.Port
 109         return m.db.Where(&orm.Node{PublicKey: ormNode.PublicKey}).
 110                 Assign(&orm.Node{
 111                         Xpub:  ormNode.Xpub,
 112                         Alias: ormNode.Alias,
 113                         Host:  ormNode.Host,
 114                         Port:  ormNode.Port,
 115                 }).FirstOrCreate(ormNode).Error
 116 }
 117
 118 func (m *monitor) makeSwitch() error {
 119         l, listenAddr := p2p.GetListener(m.nodeCfg.P2P)
 120         discv, err := dht.NewDiscover(m.nodeCfg, m.privKey, l.ExternalAddress().Port, m.cfg.NetworkID)
 121         if err != nil {
 122                 return err
 123         }
 124
 125         // no need for lanDiscv, but passing &mdns.LANDiscover{} will cause NilPointer
 126         lanDiscv := mdns.NewLANDiscover(mdns.NewProtocol(), int(l.ExternalAddress().Port))
 127         sw, err := p2p.NewSwitch(m.nodeCfg, discv, lanDiscv, l, m.privKey, listenAddr, m.cfg.NetworkID)
 128         if err != nil {
 129                 return err
 130         }
 131
 132         m.sw = sw
 133         return nil
 134 }
 135
 136 func (m *monitor) discoveryRoutine() {
 137         ticker := time.NewTicker(time.Duration(discvFreqSec) * time.Second)
 138         for range ticker.C {
 139                 nodes := make([]*dht.Node, nodesToDiscv)
 140                 n := m.sw.GetDiscv().ReadRandomNodes(nodes)
 141                 for i := 0; i < n; i++ {
 142                         m.discvCh <- nodes[i]
 143                 }
 144         }
 145 }
 146
 147 func (m *monitor) collectDiscoveredNodes() {
 148         // nodeMap maps a node's public key to the node itself
 149         nodeMap := make(map[string]*dht.Node)
 150         for node := range m.discvCh {
 151                 if n, ok := nodeMap[node.ID.String()]; ok && n.String() == node.String() {
 152                         continue
 153                 }
 154                 log.Info("discover new node: ", node)
 155
 156                 if err := m.upSertNode(&config.Node{
 157                         PublicKey: node.ID.String(),
 158                         Host:      node.IP.String(),
 159                         Port:      node.TCP,
 160                 }); err != nil {
 161                         log.Error(err)
 162                 }
 163
 164                 nodeMap[node.ID.String()] = node
 165         }
 166 }
 167
 168 func (m *monitor) connectNodesRoutine() {
 169         ticker := time.NewTicker(time.Duration(m.cfg.CheckFreqSeconds) * time.Second)
 170         for ; true; <-ticker.C {
 171                 // TODO: lock?
 172                 if err := m.dialNodes(); err != nil {
 173                         log.Error(err)
 174                 }
 175         }
 176 }
 177
 178 func (m *monitor) dialNodes() error {
 179         var nodes []*orm.Node
 180         if err := m.db.Model(&orm.Node{}).Find(&nodes).Error; err != nil {
 181                 return err
 182         }
 183
 184         addresses := make([]*p2p.NetAddress, 0)
 185         for i := 0; i < len(nodes); i++ {
 186                 ips, err := net.LookupIP(nodes[i].Host)
 187                 if err != nil {
 188                         log.Error(err)
 189                         continue
 190                 }
 191                 if len(ips) == 0 {
 192                         log.Error("fail to look up ip")
 193                         continue
 194                 }
 195
 196                 address := p2p.NewNetAddressIPPort(ips[0], nodes[i].Port)
 197                 addresses = append(addresses, address)
 198         }
 199
 200         m.sw.DialPeers(addresses)
 201         return nil
 202 }
 203
 204 // TODO:
 205 // implement logic first, and then refactor
 206 // /home/gavin/work/go/src/github.com/vapor/
 207 // p2p/test_util.go
 208 // p2p/switch_test.go
 209 // syncManager
 210 // notificationMgr
 211
 212 // TODO: dial nodes, get lantency & best_height
 213 // TODO: decide check_height("best best_height" - "confirmations")
 214 // TODO: get blockhash by check_height, get latency
 215 // TODO: update lantency, active_time and status