toolbar/precog/monitor/monitor.go

   1 package monitor
   2
   3 import (
   4         // "encoding/binary"
   5         "io/ioutil"
   6         "os"
   7         "time"
   8
   9         "github.com/jinzhu/gorm"
  10         log "github.com/sirupsen/logrus"
  11         // dbm "github.com/vapor/database/leveldb"
  12
  13         vaporCfg "github.com/vapor/config"
  14         "github.com/vapor/p2p"
  15         // conn "github.com/vapor/p2p/connection"
  16         // "github.com/vapor/consensus"
  17         // "github.com/vapor/crypto/sha3pool"
  18         "github.com/vapor/p2p/discover/dht"
  19         "github.com/vapor/p2p/discover/mdns"
  20         "github.com/vapor/p2p/signlib"
  21         "github.com/vapor/toolbar/precog/config"
  22         "github.com/vapor/toolbar/precog/database/orm"
  23 )
  24
  25 const vaporNetID = 10817814959495988245
  26
  27 type monitor struct {
  28         cfg     *config.Config
  29         db      *gorm.DB
  30         nodeCfg *vaporCfg.Config
  31 }
  32
  33 func NewMonitor(cfg *config.Config, db *gorm.DB) *monitor {
  34         dirPath, err := ioutil.TempDir(".", "")
  35         if err != nil {
  36                 log.Fatal(err)
  37         }
  38
  39         nodeCfg := &vaporCfg.Config{
  40                 BaseConfig: vaporCfg.DefaultBaseConfig(),
  41                 P2P:        vaporCfg.DefaultP2PConfig(),
  42                 Federation: vaporCfg.DefaultFederationConfig(),
  43         }
  44         nodeCfg.DBPath = dirPath
  45
  46         return &monitor{
  47                 cfg:     cfg,
  48                 db:      db,
  49                 nodeCfg: nodeCfg,
  50         }
  51 }
  52
  53 func (m *monitor) Run() {
  54         defer os.RemoveAll(m.nodeCfg.DBPath)
  55
  56         m.updateBootstrapNodes()
  57         go m.discovery()
  58         ticker := time.NewTicker(time.Duration(m.cfg.CheckFreqSeconds) * time.Second)
  59         for ; true; <-ticker.C {
  60                 // TODO: lock?
  61                 m.monitorRountine()
  62         }
  63 }
  64
  65 // create or update: https://github.com/jinzhu/gorm/issues/1307
  66 func (m *monitor) updateBootstrapNodes() {
  67         for _, node := range m.cfg.Nodes {
  68                 ormNode := &orm.Node{
  69                         PublicKey: node.PublicKey.String(),
  70                         Alias:     node.Alias,
  71                         Host:      node.Host,
  72                         Port:      node.Port,
  73                 }
  74
  75                 if err := m.db.Where(&orm.Node{PublicKey: ormNode.PublicKey}).
  76                         Assign(&orm.Node{
  77                                 Alias: node.Alias,
  78                                 Host:  node.Host,
  79                                 Port:  node.Port,
  80                         }).FirstOrCreate(ormNode).Error; err != nil {
  81                         log.Error(err)
  82                         continue
  83                 }
  84         }
  85 }
  86
  87 // TODO:
  88 // implement logic first, and then refactor
  89 // /home/gavin/work/go/src/github.com/vapor/
  90 // p2p/test_util.go
  91 // p2p/switch_test.go
  92 func (m *monitor) discovery() {
  93         sw, err := m.makeSwitch()
  94         if err != nil {
  95                 log.Fatal(err)
  96         }
  97
  98         sw.Start()
  99 }
 100
 101 func (m *monitor) makeSwitch() (*p2p.Switch, error) {
 102         swPrivKey, err := signlib.NewPrivKey()
 103         if err != nil {
 104                 return nil, err
 105         }
 106
 107         l, listenAddr := p2p.GetListener(m.nodeCfg.P2P)
 108         discv, err := dht.NewDiscover(m.nodeCfg, swPrivKey, l.ExternalAddress().Port, vaporNetID)
 109         if err != nil {
 110                 return nil, err
 111         }
 112
 113         lanDiscv := mdns.NewLANDiscover(mdns.NewProtocol(), int(l.ExternalAddress().Port))
 114         return p2p.NewSwitch(m.nodeCfg, discv, lanDiscv, l, swPrivKey, listenAddr, vaporNetID)
 115 }
 116
 117 func (m *monitor) monitorRountine() error {
 118         // TODO: dail nodes, get lantency & best_height
 119         // TODO: decide check_height("best best_height" - "confirmations")
 120         // TODO: get blockhash by check_height, get latency
 121         // TODO: update lantency, active_time and status
 122         return nil
 123 }