toolbar/precog/monitor/monitor.go

   1 package monitor
   2
   3 import (
   4         // "encoding/binary"
   5         // "encoding/hex"
   6         "fmt"
   7         "io/ioutil"
   8         "os"
   9         // "os/user"
  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         dbm "github.com/vapor/database/leveldb"
  20         "github.com/vapor/event"
  21         "github.com/vapor/p2p"
  22         // conn "github.com/vapor/p2p/connection"
  23         "github.com/vapor/netsync/chainmgr"
  24         "github.com/vapor/netsync/consensusmgr"
  25         msgs "github.com/vapor/netsync/messages"
  26         "github.com/vapor/netsync/peers"
  27         "github.com/vapor/p2p/discover/dht"
  28         "github.com/vapor/p2p/discover/mdns"
  29         "github.com/vapor/p2p/signlib"
  30         "github.com/vapor/test/mock"
  31         "github.com/vapor/toolbar/precog/config"
  32 )
  33
  34 type monitor struct {
  35         cfg     *config.Config
  36         db      *gorm.DB
  37         nodeCfg *vaporCfg.Config
  38         sw      *p2p.Switch
  39         discvCh chan *dht.Node
  40         privKey chainkd.XPrv
  41         chain   *mock.Chain
  42         txPool  *mock.Mempool
  43 }
  44
  45 // TODO: set myself as SPV?
  46 func NewMonitor(cfg *config.Config, db *gorm.DB) *monitor {
  47         //TODO: for test
  48         cfg.CheckFreqSeconds = 1
  49
  50         // TODO: fix dir
  51         tmpDir, err := ioutil.TempDir(".", "vpPrecog")
  52         if err != nil {
  53                 log.Fatalf("failed to create temporary data folder: %v", err)
  54         }
  55
  56         nodeCfg := &vaporCfg.Config{
  57                 BaseConfig: vaporCfg.DefaultBaseConfig(),
  58                 P2P:        vaporCfg.DefaultP2PConfig(),
  59                 Federation: vaporCfg.DefaultFederationConfig(),
  60         }
  61         nodeCfg.DBPath = tmpDir
  62         nodeCfg.ChainID = "mainnet"
  63         discvCh := make(chan *dht.Node)
  64         privKey, err := signlib.NewPrivKey()
  65         if err != nil {
  66                 log.Fatal(err)
  67         }
  68
  69         chain, txPool, err := mockChainAndPool()
  70         if err != nil {
  71                 log.Fatal(err)
  72         }
  73
  74         return &monitor{
  75                 cfg:     cfg,
  76                 db:      db,
  77                 nodeCfg: nodeCfg,
  78                 discvCh: discvCh,
  79                 privKey: privKey.(chainkd.XPrv),
  80                 chain:   chain,
  81                 txPool:  txPool,
  82         }
  83 }
  84
  85 func (m *monitor) Run() {
  86         defer os.RemoveAll(m.nodeCfg.DBPath)
  87
  88         var seeds []string
  89         for _, node := range m.cfg.Nodes {
  90                 seeds = append(seeds, fmt.Sprintf("%s:%d", node.Host, node.Port))
  91                 if err := m.upSertNode(&node); err != nil {
  92                         log.Error(err)
  93                 }
  94         }
  95         m.nodeCfg.P2P.Seeds = strings.Join(seeds, ",")
  96         if err := m.makeSwitch(); err != nil {
  97                 log.Fatal(err)
  98         }
  99
 100         go m.discoveryRoutine()
 101         go m.collectDiscoveredNodes()
 102         go m.connectNodesRoutine()
 103         go m.checkStatusRoutine()
 104 }
 105
 106 func (m *monitor) makeSwitch() error {
 107         l, listenAddr := p2p.GetListener(m.nodeCfg.P2P)
 108         discv, err := dht.NewDiscover(m.nodeCfg, m.privKey, l.ExternalAddress().Port, m.cfg.NetworkID)
 109         if err != nil {
 110                 return err
 111         }
 112
 113         // no need for lanDiscv, but passing &mdns.LANDiscover{} will cause NilPointer
 114         lanDiscv := mdns.NewLANDiscover(mdns.NewProtocol(), int(l.ExternalAddress().Port))
 115         sw, err := p2p.NewSwitch(m.nodeCfg, discv, lanDiscv, l, m.privKey, listenAddr, m.cfg.NetworkID)
 116         if err != nil {
 117                 return err
 118         }
 119
 120         m.sw = sw
 121         return nil
 122 }
 123
 124 func (m *monitor) prepareReactors(peers *peers.PeerSet) error {
 125         dispatcher := event.NewDispatcher()
 126         // add ConsensusReactor for consensusChannel
 127         _ = consensusmgr.NewManager(m.sw, m.chain, peers, dispatcher)
 128         fastSyncDB := dbm.NewDB("fastsync", m.nodeCfg.DBBackend, m.nodeCfg.DBDir())
 129         // add ProtocolReactor to handle msgs
 130         _, err := chainmgr.NewManager(m.nodeCfg, m.sw, m.chain, m.txPool, dispatcher, peers, fastSyncDB)
 131         if err != nil {
 132                 return err
 133         }
 134
 135         for label, reactor := range m.sw.GetReactors() {
 136                 log.Debug("start reactor: (%s:%v)", label, reactor)
 137                 if _, err := reactor.Start(); err != nil {
 138                         return nil
 139                 }
 140         }
 141
 142         m.sw.GetSecurity().RegisterFilter(m.sw.GetNodeInfo())
 143         m.sw.GetSecurity().RegisterFilter(m.sw.GetPeers())
 144
 145         return m.sw.GetSecurity().Start()
 146 }
 147
 148 func (m *monitor) checkStatusRoutine() {
 149         peers := peers.NewPeerSet(m.sw)
 150         if err := m.prepareReactors(peers); err != nil {
 151                 log.Fatal(err)
 152         }
 153
 154         bestHeight := uint64(0)
 155         ticker := time.NewTicker(time.Duration(m.cfg.CheckFreqSeconds) * time.Second)
 156         for ; true; <-ticker.C {
 157                 // TODO: peerSet 会自动更新吗？
 158                 for _, peer := range m.sw.GetPeers().List() {
 159                         peer.Start()
 160                 }
 161
 162                 for _, reactor := range m.sw.GetReactors() {
 163                         log.Debug("AddPeer for reactor", reactor)
 164                         for _, peer := range m.sw.GetPeers().List() {
 165                                 log.Debug("AddPeer", peer)
 166                                 // TODO: 这个还要吗
 167                                 reactor.AddPeer(peer)
 168                         }
 169                 }
 170
 171                 for _, peerInfo := range peers.GetPeerInfos() {
 172                         if peerInfo.Height > bestHeight {
 173                                 bestHeight = peerInfo.Height
 174                         }
 175                 }
 176                 log.Info("bestHeight", bestHeight)
 177
 178                 // msg := struct{ msgs.BlockchainMessage }{&msgs.GetBlockMessage{Height: bestHeight + 1}}
 179                 // for _, peer := range m.sw.GetPeers().List() {
 180                 //      peers.SendMsg(peer.ID(), msgs.BlockchainChannel, msg)
 181                 // }
 182         }
 183 }