toolbar/precog/monitor/monitor.go

   1 package monitor
   2
   3 import (
   4         "fmt"
   5         "os"
   6         "os/user"
   7         "strings"
   8
   9         "github.com/jinzhu/gorm"
  10         log "github.com/sirupsen/logrus"
  11
  12         vaporCfg "github.com/vapor/config"
  13         "github.com/vapor/crypto/ed25519/chainkd"
  14         dbm "github.com/vapor/database/leveldb"
  15         "github.com/vapor/event"
  16         "github.com/vapor/netsync/chainmgr"
  17         "github.com/vapor/netsync/consensusmgr"
  18         "github.com/vapor/netsync/peers"
  19         "github.com/vapor/p2p"
  20         "github.com/vapor/p2p/discover/dht"
  21         "github.com/vapor/p2p/discover/mdns"
  22         "github.com/vapor/p2p/signlib"
  23         "github.com/vapor/test/mock"
  24         "github.com/vapor/toolbar/precog/config"
  25 )
  26
  27 // TODO:
  28 // 4. 碰到一个玄学问题，究竟是以ip为单位，还是pubkey为单位。 如果同一个pubkey出现在2个不同的ip，会不会让数据混乱？
  29 // 7. m这个为什么需要锁呀？一个是节点发现，一个是生命探测，中间交互都是数据库把？
  30
  31 type monitor struct {
  32         cfg            *config.Config
  33         db             *gorm.DB
  34         nodeCfg        *vaporCfg.Config
  35         sw             *p2p.Switch
  36         privKey        chainkd.XPrv
  37         chain          *mock.Chain
  38         txPool         *mock.Mempool
  39         bestHeightSeen uint64
  40         peers          *peers.PeerSet
  41 }
  42
  43 func NewMonitor(cfg *config.Config, db *gorm.DB) *monitor {
  44         dbPath, err := makePath()
  45         if err != nil {
  46                 log.WithFields(log.Fields{"err": err}).Fatal("makePath")
  47         }
  48
  49         nodeCfg := &vaporCfg.Config{
  50                 BaseConfig: vaporCfg.DefaultBaseConfig(),
  51                 P2P:        vaporCfg.DefaultP2PConfig(),
  52                 Federation: vaporCfg.DefaultFederationConfig(),
  53         }
  54         nodeCfg.DBPath = dbPath
  55         nodeCfg.ChainID = "mainnet"
  56         privKey, err := signlib.NewPrivKey()
  57         if err != nil {
  58                 log.WithFields(log.Fields{"err": err}).Fatal("NewPrivKey")
  59         }
  60
  61         chain, txPool, err := mockChainAndPool()
  62         if err != nil {
  63                 log.WithFields(log.Fields{"err": err}).Fatal("mockChainAndPool")
  64         }
  65
  66         return &monitor{
  67                 cfg:            cfg,
  68                 db:             db,
  69                 nodeCfg:        nodeCfg,
  70                 privKey:        privKey.(chainkd.XPrv),
  71                 chain:          chain,
  72                 txPool:         txPool,
  73                 bestHeightSeen: uint64(0),
  74         }
  75 }
  76
  77 func makePath() (string, error) {
  78         usr, err := user.Current()
  79         if err != nil {
  80                 return "", err
  81         }
  82
  83         dataPath := usr.HomeDir + "/.vapor/precog"
  84         if err := os.MkdirAll(dataPath, os.ModePerm); err != nil {
  85                 return "", err
  86         }
  87
  88         return dataPath, nil
  89 }
  90
  91 func (m *monitor) Run() {
  92         if err := m.makeSwitch(); err != nil {
  93                 log.WithFields(log.Fields{"err": err}).Fatal("makeSwitch")
  94         }
  95
  96         go m.discoveryRoutine()
  97         go m.connectionRoutine()
  98 }
  99
 100 func (m *monitor) makeSwitch() error {
 101         var seeds []string
 102         for _, node := range m.cfg.Nodes {
 103                 seeds = append(seeds, fmt.Sprintf("%s:%d", node.IP, node.Port))
 104         }
 105         m.nodeCfg.P2P.Seeds = strings.Join(seeds, ",")
 106
 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         m.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.peers = peers.NewPeerSet(m.sw)
 121         return m.prepareReactors()
 122 }
 123
 124 func (m *monitor) prepareReactors() error {
 125         dispatcher := event.NewDispatcher()
 126         // add ConsensusReactor for consensusChannel
 127         _ = consensusmgr.NewManager(m.sw, m.chain, m.peers, dispatcher)
 128         fastSyncDB := dbm.NewDB("fastsync", m.nodeCfg.DBBackend, m.nodeCfg.DBDir())
 129         // add ProtocolReactor to handle msgs
 130         if _, err := chainmgr.NewManager(m.nodeCfg, m.sw, m.chain, m.txPool, dispatcher, m.peers, fastSyncDB); err != nil {
 131                 return err
 132         }
 133
 134         for label, reactor := range m.sw.GetReactors() {
 135                 log.WithFields(log.Fields{"label": label, "reactor": reactor}).Debug("start reactor")
 136                 if _, err := reactor.Start(); err != nil {
 137                         return err
 138                 }
 139         }
 140
 141         m.sw.GetSecurity().RegisterFilter(m.sw.GetNodeInfo())
 142         m.sw.GetSecurity().RegisterFilter(m.sw.GetPeers())
 143         return m.sw.GetSecurity().Start()
 144 }