toolbar/precog/monitor/monitor.go

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