toolbar/precog/monitor/monitor.go

   1 package monitor
   2
   3 import (
   4         "io/ioutil"
   5         "os"
   6         "time"
   7
   8         "github.com/jinzhu/gorm"
   9         log "github.com/sirupsen/logrus"
  10         // dbm "github.com/vapor/database/leveldb"
  11
  12         vaporCfg "github.com/vapor/config"
  13         "github.com/vapor/p2p"
  14         // conn "github.com/vapor/p2p/connection"
  15         // "github.com/vapor/p2p/signlib"
  16         // "github.com/vapor/consensus"
  17         "github.com/vapor/toolbar/precog/config"
  18         "github.com/vapor/toolbar/precog/database/orm"
  19 )
  20
  21 type monitor struct {
  22         cfg     *config.Config
  23         db      *gorm.DB
  24         nodeCfg *vaporCfg.Config
  25 }
  26
  27 func NewMonitor(cfg *config.Config, db *gorm.DB) *monitor {
  28         dirPath, err := ioutil.TempDir(".", "")
  29         if err != nil {
  30                 log.Fatal(err)
  31         }
  32
  33         nodeCfg := &vaporCfg.Config{
  34                 BaseConfig: vaporCfg.DefaultBaseConfig(),
  35                 P2P:        vaporCfg.DefaultP2PConfig(),
  36                 Federation: vaporCfg.DefaultFederationConfig(),
  37         }
  38         nodeCfg.DBPath = dirPath
  39
  40         return &monitor{
  41                 cfg:     cfg,
  42                 db:      db,
  43                 nodeCfg: nodeCfg,
  44         }
  45 }
  46
  47 func (m *monitor) Run() {
  48         defer os.RemoveAll(m.nodeCfg.DBPath)
  49
  50         m.updateBootstrapNodes()
  51         go m.discovery()
  52         ticker := time.NewTicker(time.Duration(m.cfg.CheckFreqSeconds) * time.Second)
  53         for ; true; <-ticker.C {
  54                 // TODO: lock?
  55                 m.monitorRountine()
  56         }
  57 }
  58
  59 // create or update: https://github.com/jinzhu/gorm/issues/1307
  60 func (m *monitor) updateBootstrapNodes() {
  61         for _, node := range m.cfg.Nodes {
  62                 ormNode := &orm.Node{
  63                         PublicKey: node.PublicKey.String(),
  64                         Alias:     node.Alias,
  65                         Host:      node.Host,
  66                         Port:      node.Port,
  67                 }
  68
  69                 if err := m.db.Where(&orm.Node{PublicKey: ormNode.PublicKey}).
  70                         Assign(&orm.Node{
  71                                 Alias: node.Alias,
  72                                 Host:  node.Host,
  73                                 Port:  node.Port,
  74                         }).FirstOrCreate(ormNode).Error; err != nil {
  75                         log.Error(err)
  76                         continue
  77                 }
  78         }
  79 }
  80
  81 // TODO:
  82 // implement logic first, and then refactor
  83 // /home/gavin/work/go/src/github.com/vapor/
  84 // p2p/test_util.go
  85 // p2p/switch_test.go
  86 func (m *monitor) discovery() {
  87         sw, err := m.makeSwitch()
  88         if err != nil {
  89                 log.Fatal(err)
  90         }
  91
  92         sw.Start()
  93         defer sw.Stop()
  94 }
  95
  96 func (m *monitor) calcNetID() (*p2p.Switch, error) {
  97         var data []byte
  98         var h [32]byte
  99         data = append(data, m.nodeCfg.GenesisBlock().Hash().Bytes()...)
 100         magic := make([]byte, 8)
 101         magicNumber := uint64(0x054c5638)
 102         binary.BigEndian.PutUint64(magic, magicNumber)
 103         data = append(data, magic[:]...)
 104         sha3pool.Sum256(h[:], data)
 105         return binary.BigEndian.Uint64(h[:8])
 106 }
 107
 108 func (m *monitor) makeSwitch() (*p2p.Switch, error) {
 109         // TODO: 包一下？  common cfg 之类的？
 110
 111         var err error
 112         var l Listener
 113         var listenAddr string
 114         var discv *dht.Network
 115         var lanDiscv *mdns.LANDiscover
 116
 117         // swPrivKey, err := signlib.NewPrivKey()
 118         // if err != nil {
 119         //      log.Fatal(err)
 120         // }
 121
 122         // TODO: whatz that for
 123         // testDB := dbm.NewDB("testdb", "leveldb", dirPath)
 124         // TODO: clean up
 125         // log.Println("Federation.Xpubs", mCfg.Federation.Xpubs)
 126         sw, err := p2p.NewSwitch(mCfg)
 127         if err != nil {
 128                 return nil, err
 129         }
 130
 131         return sw, nil
 132 }
 133
 134 func (m *monitor) monitorRountine() error {
 135         // TODO: dail nodes, get lantency & best_height
 136         // TODO: decide check_height("best best_height" - "confirmations")
 137         // TODO: get blockhash by check_height, get latency
 138         // TODO: update lantency, active_time and status
 139         return nil
 140 }