toolbar/precog/monitor/monitor.go

   1 package monitor
   2
   3 import (
   4         // "encoding/binary"
   5         // "encoding/hex"
   6         "fmt"
   7         "io/ioutil"
   8         "net"
   9         "os"
  10         // "os/user"
  11         "strings"
  12         "time"
  13
  14         "github.com/jinzhu/gorm"
  15         log "github.com/sirupsen/logrus"
  16         // dbm "github.com/vapor/database/leveldb"
  17
  18         vaporCfg "github.com/vapor/config"
  19         "github.com/vapor/consensus"
  20         "github.com/vapor/crypto/ed25519/chainkd"
  21         dbm "github.com/vapor/database/leveldb"
  22         "github.com/vapor/event"
  23         "github.com/vapor/p2p"
  24         // conn "github.com/vapor/p2p/connection"
  25         "github.com/vapor/netsync/peers"
  26         // "github.com/vapor/consensus"
  27         // "github.com/vapor/crypto/sha3pool"
  28         "github.com/vapor/netsync/chainmgr"
  29         "github.com/vapor/netsync/consensusmgr"
  30         "github.com/vapor/p2p/discover/dht"
  31         "github.com/vapor/p2p/discover/mdns"
  32         "github.com/vapor/p2p/signlib"
  33         "github.com/vapor/test/mock"
  34         "github.com/vapor/toolbar/precog/config"
  35         "github.com/vapor/toolbar/precog/database/orm"
  36 )
  37
  38 var (
  39         nodesToDiscv = 150
  40         discvFreqSec = 60
  41 )
  42
  43 type monitor struct {
  44         cfg     *config.Config
  45         db      *gorm.DB
  46         nodeCfg *vaporCfg.Config
  47         sw      *p2p.Switch
  48         discvCh chan *dht.Node
  49         privKey chainkd.XPrv
  50         chain   *mock.Chain
  51         txPool  *mock.Mempool
  52 }
  53
  54 // TODO: set myself as SPV?
  55 func NewMonitor(cfg *config.Config, db *gorm.DB) *monitor {
  56         //TODO: for test
  57         cfg.CheckFreqSeconds = 1
  58
  59         tmpDir, err := ioutil.TempDir(".", "vpPrecog")
  60         if err != nil {
  61                 log.Fatalf("failed to create temporary data folder: %v", err)
  62         }
  63
  64         nodeCfg := &vaporCfg.Config{
  65                 BaseConfig: vaporCfg.DefaultBaseConfig(),
  66                 P2P:        vaporCfg.DefaultP2PConfig(),
  67                 Federation: vaporCfg.DefaultFederationConfig(),
  68         }
  69         nodeCfg.DBPath = tmpDir
  70         nodeCfg.ChainID = "mainnet"
  71         discvCh := make(chan *dht.Node)
  72         privKey, err := signlib.NewPrivKey()
  73         if err != nil {
  74                 log.Fatal(err)
  75         }
  76
  77         chain, txPool, err := mockChainAndPool()
  78         if err != nil {
  79                 log.Fatal(err)
  80         }
  81
  82         return &monitor{
  83                 cfg:     cfg,
  84                 db:      db,
  85                 nodeCfg: nodeCfg,
  86                 discvCh: discvCh,
  87                 privKey: privKey.(chainkd.XPrv),
  88                 chain:   chain,
  89                 txPool:  txPool,
  90         }
  91 }
  92
  93 func (m *monitor) Run() {
  94         defer os.RemoveAll(m.nodeCfg.DBPath)
  95
  96         var seeds []string
  97         for _, node := range m.cfg.Nodes {
  98                 seeds = append(seeds, fmt.Sprintf("%s:%d", node.Host, node.Port))
  99                 if err := m.upSertNode(&node); err != nil {
 100                         log.Error(err)
 101                 }
 102         }
 103         m.nodeCfg.P2P.Seeds = strings.Join(seeds, ",")
 104         if err := m.makeSwitch(); err != nil {
 105                 log.Fatal(err)
 106         }
 107
 108         go m.discoveryRoutine()
 109         go m.collectDiscoveredNodes()
 110         go m.connectNodesRoutine()
 111         go m.checkStatusRoutine()
 112 }
 113
 114 // create or update: https://github.com/jinzhu/gorm/issues/1307
 115 func (m *monitor) upSertNode(node *config.Node) error {
 116         if node.XPub != nil {
 117                 node.PublicKey = fmt.Sprintf("%v", node.XPub.PublicKey().String())
 118         }
 119
 120         ormNode := &orm.Node{PublicKey: node.PublicKey}
 121         if err := m.db.Where(&orm.Node{PublicKey: node.PublicKey}).First(ormNode).Error; err != nil && err != gorm.ErrRecordNotFound {
 122                 return err
 123         }
 124
 125         if node.Alias != "" {
 126                 ormNode.Alias = node.Alias
 127         }
 128         if node.XPub != nil {
 129                 ormNode.Xpub = node.XPub.String()
 130         }
 131         ormNode.Host = node.Host
 132         ormNode.Port = node.Port
 133         return m.db.Where(&orm.Node{PublicKey: ormNode.PublicKey}).
 134                 Assign(&orm.Node{
 135                         Xpub:  ormNode.Xpub,
 136                         Alias: ormNode.Alias,
 137                         Host:  ormNode.Host,
 138                         Port:  ormNode.Port,
 139                 }).FirstOrCreate(ormNode).Error
 140 }
 141
 142 func (m *monitor) makeSwitch() error {
 143         l, listenAddr := p2p.GetListener(m.nodeCfg.P2P)
 144         discv, err := dht.NewDiscover(m.nodeCfg, m.privKey, l.ExternalAddress().Port, m.cfg.NetworkID)
 145         if err != nil {
 146                 return err
 147         }
 148
 149         // no need for lanDiscv, but passing &mdns.LANDiscover{} will cause NilPointer
 150         lanDiscv := mdns.NewLANDiscover(mdns.NewProtocol(), int(l.ExternalAddress().Port))
 151         sw, err := p2p.NewSwitch(m.nodeCfg, discv, lanDiscv, l, m.privKey, listenAddr, m.cfg.NetworkID)
 152         if err != nil {
 153                 return err
 154         }
 155
 156         m.sw = sw
 157         return nil
 158 }
 159
 160 func (m *monitor) discoveryRoutine() {
 161         ticker := time.NewTicker(time.Duration(discvFreqSec) * time.Second)
 162         for range ticker.C {
 163                 nodes := make([]*dht.Node, nodesToDiscv)
 164                 n := m.sw.GetDiscv().ReadRandomNodes(nodes)
 165                 for i := 0; i < n; i++ {
 166                         m.discvCh <- nodes[i]
 167                 }
 168         }
 169 }
 170
 171 func (m *monitor) collectDiscoveredNodes() {
 172         // nodeMap maps a node's public key to the node itself
 173         nodeMap := make(map[string]*dht.Node)
 174         for node := range m.discvCh {
 175                 if n, ok := nodeMap[node.ID.String()]; ok && n.String() == node.String() {
 176                         continue
 177                 }
 178                 log.Info("discover new node: ", node)
 179
 180                 if err := m.upSertNode(&config.Node{
 181                         PublicKey: node.ID.String(),
 182                         Host:      node.IP.String(),
 183                         Port:      node.TCP,
 184                 }); err != nil {
 185                         log.Error(err)
 186                 }
 187
 188                 nodeMap[node.ID.String()] = node
 189         }
 190 }
 191
 192 func (m *monitor) connectNodesRoutine() {
 193         // TODO: change name?
 194         ticker := time.NewTicker(time.Duration(m.cfg.CheckFreqSeconds) * time.Second)
 195         for ; true; <-ticker.C {
 196                 if err := m.dialNodes(); err != nil {
 197                         log.Error(err)
 198                 }
 199         }
 200 }
 201
 202 func (m *monitor) dialNodes() error {
 203         var nodes []*orm.Node
 204         if err := m.db.Model(&orm.Node{}).Find(&nodes).Error; err != nil {
 205                 return err
 206         }
 207
 208         addresses := make([]*p2p.NetAddress, 0)
 209         for i := 0; i < len(nodes); i++ {
 210                 ips, err := net.LookupIP(nodes[i].Host)
 211                 if err != nil {
 212                         log.Error(err)
 213                         continue
 214                 }
 215                 if len(ips) == 0 {
 216                         log.Errorf("fail to look up ip for %s", nodes[i].Host)
 217                         continue
 218                 }
 219
 220                 address := p2p.NewNetAddressIPPort(ips[0], nodes[i].Port)
 221                 addresses = append(addresses, address)
 222         }
 223
 224         m.sw.DialPeers(addresses)
 225         return nil
 226 }
 227
 228 func (m *monitor) checkStatusRoutine() {
 229         dispatcher := event.NewDispatcher()
 230         peers := peers.NewPeerSet(m.sw)
 231         // add ConsensusReactor for consensusChannel
 232         _ = consensusmgr.NewManager(m.sw, m.chain, peers, dispatcher)
 233         fastSyncDB := dbm.NewDB("fastsync", m.nodeCfg.DBBackend, m.nodeCfg.DBDir())
 234         // add ProtocolReactor to handle msgs
 235         _, err := chainmgr.NewManager(m.nodeCfg, m.sw, m.chain, m.txPool, dispatcher, peers, fastSyncDB)
 236         if err != nil {
 237                 log.Fatal(err)
 238         }
 239
 240         // TODO: clean up?? only start reactors??
 241         m.sw.Start()
 242
 243         // for label, reactor := range m.sw.GetReactors() {
 244         //      log.Debug("start reactor: (%s:%v)", label, reactor)
 245         //      if _, err := reactor.Start(); err != nil {
 246         //              return
 247         //      }
 248         // }
 249
 250         // m.sw.GetSecurity().RegisterFilter(m.sw.GetNodeInfo())
 251         // m.sw.GetSecurity().RegisterFilter(m.sw.GetPeers())
 252         // if err := m.sw.GetSecurity().Start(); err != nil {
 253         //      return
 254         // }
 255
 256         ticker := time.NewTicker(time.Duration(m.cfg.CheckFreqSeconds) * time.Second)
 257         for ; true; <-ticker.C {
 258                 for _, v := range m.sw.GetReactors() {
 259                         for _, peer := range m.sw.GetPeers().List() {
 260                                 log.Debug("AddPeer for", v, peer)
 261                                 // TODO: if not in sw
 262                                 v.AddPeer(peer)
 263                         }
 264                 }
 265
 266                 // TODO: SFSPV?
 267                 log.Debug("best", peers.BestPeer(consensus.SFFullNode))
 268                 for _, peerInfo := range peers.GetPeerInfos() {
 269                         log.Info(peerInfo)
 270                 }
 271         }
 272 }
 273
 274 // TODO:
 275 // implement logic first, and then refactor
 276 // /home/gavin/work/go/src/github.com/vapor/
 277 // p2p/test_util.go
 278 // p2p/switch_test.go
 279
 280 // TODO: get lantency
 281 // TODO: get best_height
 282 // TODO: decide check_height("best best_height" - "confirmations")
 283 // TODO: get blockhash by check_height, get latency
 284 // TODO: update lantency, active_time and status