2016-05-08 03:32:51 -04:00
|
|
|
package networkdb
|
|
|
|
|
|
|
|
import (
|
2016-06-07 17:28:28 -04:00
|
|
|
"bytes"
|
2016-05-08 03:32:51 -04:00
|
|
|
"crypto/rand"
|
2016-06-22 12:40:20 -04:00
|
|
|
"encoding/hex"
|
2016-05-08 03:32:51 -04:00
|
|
|
"fmt"
|
2016-12-12 03:29:54 -05:00
|
|
|
"log"
|
2016-05-08 03:32:51 -04:00
|
|
|
"math/big"
|
|
|
|
rnd "math/rand"
|
2016-09-20 21:21:19 -04:00
|
|
|
"net"
|
2016-05-08 03:32:51 -04:00
|
|
|
"strings"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/Sirupsen/logrus"
|
|
|
|
"github.com/hashicorp/memberlist"
|
|
|
|
)
|
|
|
|
|
2016-09-20 21:21:19 -04:00
|
|
|
const (
|
2016-10-26 16:37:46 -04:00
|
|
|
reapInterval = 60 * time.Second
|
|
|
|
reapPeriod = 5 * time.Second
|
|
|
|
retryInterval = 1 * time.Second
|
|
|
|
nodeReapInterval = 24 * time.Hour
|
|
|
|
nodeReapPeriod = 2 * time.Hour
|
2016-09-20 21:21:19 -04:00
|
|
|
)
|
2016-05-08 03:32:51 -04:00
|
|
|
|
|
|
|
type logWriter struct{}
|
|
|
|
|
|
|
|
func (l *logWriter) Write(p []byte) (int, error) {
|
|
|
|
str := string(p)
|
2016-12-12 03:29:54 -05:00
|
|
|
str = strings.TrimSuffix(str, "\n")
|
2016-05-08 03:32:51 -04:00
|
|
|
|
|
|
|
switch {
|
2016-12-12 03:29:54 -05:00
|
|
|
case strings.HasPrefix(str, "[WARN] "):
|
|
|
|
str = strings.TrimPrefix(str, "[WARN] ")
|
2016-05-08 03:32:51 -04:00
|
|
|
logrus.Warn(str)
|
2016-12-12 03:29:54 -05:00
|
|
|
case strings.HasPrefix(str, "[DEBUG] "):
|
|
|
|
str = strings.TrimPrefix(str, "[DEBUG] ")
|
2016-05-08 03:32:51 -04:00
|
|
|
logrus.Debug(str)
|
2016-12-12 03:29:54 -05:00
|
|
|
case strings.HasPrefix(str, "[INFO] "):
|
|
|
|
str = strings.TrimPrefix(str, "[INFO] ")
|
2016-05-08 03:32:51 -04:00
|
|
|
logrus.Info(str)
|
2016-12-12 03:29:54 -05:00
|
|
|
case strings.HasPrefix(str, "[ERR] "):
|
|
|
|
str = strings.TrimPrefix(str, "[ERR] ")
|
2016-05-08 03:32:51 -04:00
|
|
|
logrus.Warn(str)
|
|
|
|
}
|
|
|
|
|
|
|
|
return len(p), nil
|
|
|
|
}
|
|
|
|
|
2016-06-07 17:28:28 -04:00
|
|
|
// SetKey adds a new key to the key ring
|
|
|
|
func (nDB *NetworkDB) SetKey(key []byte) {
|
2016-06-22 12:40:20 -04:00
|
|
|
logrus.Debugf("Adding key %s", hex.EncodeToString(key)[0:5])
|
2016-11-30 01:53:17 -05:00
|
|
|
nDB.Lock()
|
|
|
|
defer nDB.Unlock()
|
2016-06-07 17:28:28 -04:00
|
|
|
for _, dbKey := range nDB.config.Keys {
|
|
|
|
if bytes.Equal(key, dbKey) {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
nDB.config.Keys = append(nDB.config.Keys, key)
|
|
|
|
if nDB.keyring != nil {
|
|
|
|
nDB.keyring.AddKey(key)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// SetPrimaryKey sets the given key as the primary key. This should have
|
|
|
|
// been added apriori through SetKey
|
|
|
|
func (nDB *NetworkDB) SetPrimaryKey(key []byte) {
|
2016-06-22 12:40:20 -04:00
|
|
|
logrus.Debugf("Primary Key %s", hex.EncodeToString(key)[0:5])
|
2016-11-30 01:53:17 -05:00
|
|
|
nDB.RLock()
|
|
|
|
defer nDB.RUnlock()
|
2016-06-07 17:28:28 -04:00
|
|
|
for _, dbKey := range nDB.config.Keys {
|
|
|
|
if bytes.Equal(key, dbKey) {
|
|
|
|
if nDB.keyring != nil {
|
|
|
|
nDB.keyring.UseKey(dbKey)
|
|
|
|
}
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// RemoveKey removes a key from the key ring. The key being removed
|
|
|
|
// can't be the primary key
|
|
|
|
func (nDB *NetworkDB) RemoveKey(key []byte) {
|
2016-06-22 12:40:20 -04:00
|
|
|
logrus.Debugf("Remove Key %s", hex.EncodeToString(key)[0:5])
|
2016-11-30 01:53:17 -05:00
|
|
|
nDB.Lock()
|
|
|
|
defer nDB.Unlock()
|
2016-06-07 17:28:28 -04:00
|
|
|
for i, dbKey := range nDB.config.Keys {
|
|
|
|
if bytes.Equal(key, dbKey) {
|
|
|
|
nDB.config.Keys = append(nDB.config.Keys[:i], nDB.config.Keys[i+1:]...)
|
|
|
|
if nDB.keyring != nil {
|
|
|
|
nDB.keyring.RemoveKey(dbKey)
|
|
|
|
}
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-05-08 03:32:51 -04:00
|
|
|
func (nDB *NetworkDB) clusterInit() error {
|
|
|
|
config := memberlist.DefaultLANConfig()
|
|
|
|
config.Name = nDB.config.NodeName
|
2016-09-22 21:42:55 -04:00
|
|
|
config.BindAddr = nDB.config.BindAddr
|
2016-07-21 17:27:47 -04:00
|
|
|
config.AdvertiseAddr = nDB.config.AdvertiseAddr
|
2016-05-08 03:32:51 -04:00
|
|
|
|
|
|
|
if nDB.config.BindPort != 0 {
|
|
|
|
config.BindPort = nDB.config.BindPort
|
|
|
|
}
|
|
|
|
|
|
|
|
config.ProtocolVersion = memberlist.ProtocolVersionMax
|
|
|
|
config.Delegate = &delegate{nDB: nDB}
|
|
|
|
config.Events = &eventDelegate{nDB: nDB}
|
2016-12-12 03:29:54 -05:00
|
|
|
// custom logger that does not add time or date, so they are not
|
|
|
|
// duplicated by logrus
|
|
|
|
config.Logger = log.New(&logWriter{}, "", 0)
|
2016-05-08 03:32:51 -04:00
|
|
|
|
2016-06-07 17:28:28 -04:00
|
|
|
var err error
|
|
|
|
if len(nDB.config.Keys) > 0 {
|
2016-06-22 12:40:20 -04:00
|
|
|
for i, key := range nDB.config.Keys {
|
|
|
|
logrus.Debugf("Encryption key %d: %s", i+1, hex.EncodeToString(key)[0:5])
|
|
|
|
}
|
2016-06-07 17:28:28 -04:00
|
|
|
nDB.keyring, err = memberlist.NewKeyring(nDB.config.Keys, nDB.config.Keys[0])
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
config.Keyring = nDB.keyring
|
|
|
|
}
|
|
|
|
|
2016-05-08 03:32:51 -04:00
|
|
|
nDB.networkBroadcasts = &memberlist.TransmitLimitedQueue{
|
|
|
|
NumNodes: func() int {
|
2016-10-03 14:52:17 -04:00
|
|
|
nDB.RLock()
|
|
|
|
num := len(nDB.nodes)
|
|
|
|
nDB.RUnlock()
|
|
|
|
return num
|
2016-05-08 03:32:51 -04:00
|
|
|
},
|
|
|
|
RetransmitMult: config.RetransmitMult,
|
|
|
|
}
|
|
|
|
|
2016-09-20 21:21:19 -04:00
|
|
|
nDB.nodeBroadcasts = &memberlist.TransmitLimitedQueue{
|
|
|
|
NumNodes: func() int {
|
2016-10-03 14:52:17 -04:00
|
|
|
nDB.RLock()
|
|
|
|
num := len(nDB.nodes)
|
|
|
|
nDB.RUnlock()
|
|
|
|
return num
|
2016-09-20 21:21:19 -04:00
|
|
|
},
|
|
|
|
RetransmitMult: config.RetransmitMult,
|
|
|
|
}
|
|
|
|
|
2016-05-08 03:32:51 -04:00
|
|
|
mlist, err := memberlist.Create(config)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("failed to create memberlist: %v", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
nDB.stopCh = make(chan struct{})
|
|
|
|
nDB.memberlist = mlist
|
|
|
|
|
|
|
|
for _, trigger := range []struct {
|
|
|
|
interval time.Duration
|
|
|
|
fn func()
|
|
|
|
}{
|
2016-09-20 21:21:19 -04:00
|
|
|
{reapPeriod, nDB.reapState},
|
2016-05-08 03:32:51 -04:00
|
|
|
{config.GossipInterval, nDB.gossip},
|
|
|
|
{config.PushPullInterval, nDB.bulkSyncTables},
|
2016-09-20 21:21:19 -04:00
|
|
|
{retryInterval, nDB.reconnectNode},
|
2016-10-26 16:37:46 -04:00
|
|
|
{nodeReapPeriod, nDB.reapDeadNode},
|
2016-05-08 03:32:51 -04:00
|
|
|
} {
|
|
|
|
t := time.NewTicker(trigger.interval)
|
|
|
|
go nDB.triggerFunc(trigger.interval, t.C, nDB.stopCh, trigger.fn)
|
|
|
|
nDB.tickers = append(nDB.tickers, t)
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2016-09-20 21:21:19 -04:00
|
|
|
func (nDB *NetworkDB) retryJoin(members []string, stop <-chan struct{}) {
|
|
|
|
t := time.NewTicker(retryInterval)
|
|
|
|
defer t.Stop()
|
|
|
|
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-t.C:
|
|
|
|
if _, err := nDB.memberlist.Join(members); err != nil {
|
|
|
|
logrus.Errorf("Failed to join memberlist %s on retry: %v", members, err)
|
|
|
|
continue
|
|
|
|
}
|
2016-09-30 14:57:14 -04:00
|
|
|
if err := nDB.sendNodeEvent(NodeEventTypeJoin); err != nil {
|
|
|
|
logrus.Errorf("failed to send node join on retry: %v", err)
|
|
|
|
continue
|
|
|
|
}
|
2016-09-20 21:21:19 -04:00
|
|
|
return
|
|
|
|
case <-stop:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
2016-05-08 03:32:51 -04:00
|
|
|
func (nDB *NetworkDB) clusterJoin(members []string) error {
|
|
|
|
mlist := nDB.memberlist
|
|
|
|
|
|
|
|
if _, err := mlist.Join(members); err != nil {
|
2016-09-20 21:21:19 -04:00
|
|
|
// Incase of failure, keep retrying join until it succeeds or the cluster is shutdown.
|
|
|
|
go nDB.retryJoin(members, nDB.stopCh)
|
|
|
|
|
2016-05-08 03:32:51 -04:00
|
|
|
return fmt.Errorf("could not join node to memberlist: %v", err)
|
|
|
|
}
|
|
|
|
|
2016-09-20 21:21:19 -04:00
|
|
|
if err := nDB.sendNodeEvent(NodeEventTypeJoin); err != nil {
|
|
|
|
return fmt.Errorf("failed to send node join: %v", err)
|
|
|
|
}
|
|
|
|
|
2016-05-08 03:32:51 -04:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (nDB *NetworkDB) clusterLeave() error {
|
|
|
|
mlist := nDB.memberlist
|
|
|
|
|
2016-09-20 21:21:19 -04:00
|
|
|
if err := nDB.sendNodeEvent(NodeEventTypeLeave); err != nil {
|
2016-09-23 19:04:48 -04:00
|
|
|
logrus.Errorf("failed to send node leave: %v", err)
|
2016-09-20 21:21:19 -04:00
|
|
|
}
|
|
|
|
|
2016-05-08 03:32:51 -04:00
|
|
|
if err := mlist.Leave(time.Second); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
close(nDB.stopCh)
|
|
|
|
|
|
|
|
for _, t := range nDB.tickers {
|
|
|
|
t.Stop()
|
|
|
|
}
|
|
|
|
|
|
|
|
return mlist.Shutdown()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (nDB *NetworkDB) triggerFunc(stagger time.Duration, C <-chan time.Time, stop <-chan struct{}, f func()) {
|
|
|
|
// Use a random stagger to avoid syncronizing
|
|
|
|
randStagger := time.Duration(uint64(rnd.Int63()) % uint64(stagger))
|
|
|
|
select {
|
|
|
|
case <-time.After(randStagger):
|
|
|
|
case <-stop:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-C:
|
|
|
|
f()
|
|
|
|
case <-stop:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-10-26 16:37:46 -04:00
|
|
|
func (nDB *NetworkDB) reapDeadNode() {
|
|
|
|
nDB.Lock()
|
|
|
|
defer nDB.Unlock()
|
|
|
|
for id, n := range nDB.failedNodes {
|
|
|
|
if n.reapTime > 0 {
|
2016-11-09 21:01:07 -05:00
|
|
|
n.reapTime -= nodeReapPeriod
|
2016-10-26 16:37:46 -04:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
logrus.Debugf("Removing failed node %v from gossip cluster", n.Name)
|
|
|
|
delete(nDB.failedNodes, id)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-09-20 21:21:19 -04:00
|
|
|
func (nDB *NetworkDB) reconnectNode() {
|
|
|
|
nDB.RLock()
|
|
|
|
if len(nDB.failedNodes) == 0 {
|
|
|
|
nDB.RUnlock()
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
nodes := make([]*node, 0, len(nDB.failedNodes))
|
|
|
|
for _, n := range nDB.failedNodes {
|
|
|
|
nodes = append(nodes, n)
|
|
|
|
}
|
|
|
|
nDB.RUnlock()
|
|
|
|
|
|
|
|
node := nodes[randomOffset(len(nodes))]
|
|
|
|
addr := net.UDPAddr{IP: node.Addr, Port: int(node.Port)}
|
|
|
|
|
|
|
|
if _, err := nDB.memberlist.Join([]string{addr.String()}); err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := nDB.sendNodeEvent(NodeEventTypeJoin); err != nil {
|
|
|
|
logrus.Errorf("failed to send node join during reconnect: %v", err)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2016-09-23 19:04:48 -04:00
|
|
|
// Update all the local table state to a new time to
|
|
|
|
// force update on the node we are trying to rejoin, just in
|
|
|
|
// case that node has these in deleting state still. This is
|
|
|
|
// facilitate fast convergence after recovering from a gossip
|
|
|
|
// failure.
|
|
|
|
nDB.updateLocalTableTime()
|
|
|
|
|
2016-09-20 21:21:19 -04:00
|
|
|
logrus.Debugf("Initiating bulk sync with node %s after reconnect", node.Name)
|
|
|
|
nDB.bulkSync([]string{node.Name}, true)
|
|
|
|
}
|
|
|
|
|
2016-10-20 14:26:04 -04:00
|
|
|
// For timing the entry deletion in the repaer APIs that doesn't use monotonic clock
|
|
|
|
// source (time.Now, Sub etc.) should be avoided. Hence we use reapTime in every
|
|
|
|
// entry which is set initially to reapInterval and decremented by reapPeriod every time
|
|
|
|
// the reaper runs. NOTE nDB.reapTableEntries updates the reapTime with a readlock. This
|
|
|
|
// is safe as long as no other concurrent path touches the reapTime field.
|
2016-05-08 03:32:51 -04:00
|
|
|
func (nDB *NetworkDB) reapState() {
|
|
|
|
nDB.reapNetworks()
|
|
|
|
nDB.reapTableEntries()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (nDB *NetworkDB) reapNetworks() {
|
|
|
|
nDB.Lock()
|
|
|
|
for name, nn := range nDB.networks {
|
|
|
|
for id, n := range nn {
|
2016-10-20 14:26:04 -04:00
|
|
|
if n.leaving {
|
|
|
|
if n.reapTime <= 0 {
|
|
|
|
delete(nn, id)
|
|
|
|
nDB.deleteNetworkNode(id, name)
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
n.reapTime -= reapPeriod
|
2016-05-08 03:32:51 -04:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
nDB.Unlock()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (nDB *NetworkDB) reapTableEntries() {
|
2016-08-23 01:20:48 -04:00
|
|
|
var paths []string
|
2016-05-08 03:32:51 -04:00
|
|
|
|
|
|
|
nDB.RLock()
|
|
|
|
nDB.indexes[byTable].Walk(func(path string, v interface{}) bool {
|
|
|
|
entry, ok := v.(*entry)
|
|
|
|
if !ok {
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
2016-10-20 14:26:04 -04:00
|
|
|
if !entry.deleting {
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
if entry.reapTime > 0 {
|
|
|
|
entry.reapTime -= reapPeriod
|
2016-05-08 03:32:51 -04:00
|
|
|
return false
|
|
|
|
}
|
|
|
|
paths = append(paths, path)
|
|
|
|
return false
|
|
|
|
})
|
|
|
|
nDB.RUnlock()
|
|
|
|
|
|
|
|
nDB.Lock()
|
2016-08-23 01:20:48 -04:00
|
|
|
for _, path := range paths {
|
2016-05-08 03:32:51 -04:00
|
|
|
params := strings.Split(path[1:], "/")
|
|
|
|
tname := params[0]
|
|
|
|
nid := params[1]
|
|
|
|
key := params[2]
|
|
|
|
|
|
|
|
if _, ok := nDB.indexes[byTable].Delete(fmt.Sprintf("/%s/%s/%s", tname, nid, key)); !ok {
|
|
|
|
logrus.Errorf("Could not delete entry in table %s with network id %s and key %s as it does not exist", tname, nid, key)
|
|
|
|
}
|
|
|
|
|
|
|
|
if _, ok := nDB.indexes[byNetwork].Delete(fmt.Sprintf("/%s/%s/%s", nid, tname, key)); !ok {
|
|
|
|
logrus.Errorf("Could not delete entry in network %s with table name %s and key %s as it does not exist", nid, tname, key)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
nDB.Unlock()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (nDB *NetworkDB) gossip() {
|
|
|
|
networkNodes := make(map[string][]string)
|
|
|
|
nDB.RLock()
|
2016-06-07 17:28:28 -04:00
|
|
|
thisNodeNetworks := nDB.networks[nDB.config.NodeName]
|
|
|
|
for nid := range thisNodeNetworks {
|
2016-05-08 03:32:51 -04:00
|
|
|
networkNodes[nid] = nDB.networkNodes[nid]
|
|
|
|
|
|
|
|
}
|
|
|
|
nDB.RUnlock()
|
|
|
|
|
|
|
|
for nid, nodes := range networkNodes {
|
|
|
|
mNodes := nDB.mRandomNodes(3, nodes)
|
|
|
|
bytesAvail := udpSendBuf - compoundHeaderOverhead
|
|
|
|
|
|
|
|
nDB.RLock()
|
2016-06-07 17:28:28 -04:00
|
|
|
network, ok := thisNodeNetworks[nid]
|
2016-05-08 03:32:51 -04:00
|
|
|
nDB.RUnlock()
|
2016-06-07 17:28:28 -04:00
|
|
|
if !ok || network == nil {
|
|
|
|
// It is normal for the network to be removed
|
|
|
|
// between the time we collect the network
|
|
|
|
// attachments of this node and processing
|
|
|
|
// them here.
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
broadcastQ := network.tableBroadcasts
|
2016-05-08 03:32:51 -04:00
|
|
|
|
|
|
|
if broadcastQ == nil {
|
|
|
|
logrus.Errorf("Invalid broadcastQ encountered while gossiping for network %s", nid)
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
msgs := broadcastQ.GetBroadcasts(compoundOverhead, bytesAvail)
|
|
|
|
if len(msgs) == 0 {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
// Create a compound message
|
|
|
|
compound := makeCompoundMessage(msgs)
|
|
|
|
|
|
|
|
for _, node := range mNodes {
|
|
|
|
nDB.RLock()
|
|
|
|
mnode := nDB.nodes[node]
|
|
|
|
nDB.RUnlock()
|
|
|
|
|
|
|
|
if mnode == nil {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
|
|
|
// Send the compound message
|
2016-09-20 21:21:19 -04:00
|
|
|
if err := nDB.memberlist.SendToUDP(&mnode.Node, compound); err != nil {
|
2016-05-08 03:32:51 -04:00
|
|
|
logrus.Errorf("Failed to send gossip to %s: %s", mnode.Addr, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (nDB *NetworkDB) bulkSyncTables() {
|
|
|
|
var networks []string
|
|
|
|
nDB.RLock()
|
2016-08-11 02:31:01 -04:00
|
|
|
for nid, network := range nDB.networks[nDB.config.NodeName] {
|
|
|
|
if network.leaving {
|
|
|
|
continue
|
|
|
|
}
|
2016-05-08 03:32:51 -04:00
|
|
|
networks = append(networks, nid)
|
|
|
|
}
|
|
|
|
nDB.RUnlock()
|
|
|
|
|
|
|
|
for {
|
|
|
|
if len(networks) == 0 {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
|
|
|
nid := networks[0]
|
|
|
|
networks = networks[1:]
|
|
|
|
|
2016-06-14 16:45:24 -04:00
|
|
|
nDB.RLock()
|
|
|
|
nodes := nDB.networkNodes[nid]
|
|
|
|
nDB.RUnlock()
|
|
|
|
|
|
|
|
// No peer nodes on this network. Move on.
|
|
|
|
if len(nodes) == 0 {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2016-09-20 21:21:19 -04:00
|
|
|
completed, err := nDB.bulkSync(nodes, false)
|
2016-05-08 03:32:51 -04:00
|
|
|
if err != nil {
|
|
|
|
logrus.Errorf("periodic bulk sync failure for network %s: %v", nid, err)
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
// Remove all the networks for which we have
|
|
|
|
// successfully completed bulk sync in this iteration.
|
|
|
|
updatedNetworks := make([]string, 0, len(networks))
|
|
|
|
for _, nid := range networks {
|
2016-06-16 17:32:12 -04:00
|
|
|
var found bool
|
2016-05-08 03:32:51 -04:00
|
|
|
for _, completedNid := range completed {
|
|
|
|
if nid == completedNid {
|
2016-06-16 17:32:12 -04:00
|
|
|
found = true
|
|
|
|
break
|
2016-05-08 03:32:51 -04:00
|
|
|
}
|
2016-06-16 17:32:12 -04:00
|
|
|
}
|
2016-05-08 03:32:51 -04:00
|
|
|
|
2016-06-16 17:32:12 -04:00
|
|
|
if !found {
|
2016-05-08 03:32:51 -04:00
|
|
|
updatedNetworks = append(updatedNetworks, nid)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
networks = updatedNetworks
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-09-20 21:21:19 -04:00
|
|
|
func (nDB *NetworkDB) bulkSync(nodes []string, all bool) ([]string, error) {
|
2016-05-08 03:32:51 -04:00
|
|
|
if !all {
|
|
|
|
// If not all, then just pick one.
|
|
|
|
nodes = nDB.mRandomNodes(1, nodes)
|
|
|
|
}
|
|
|
|
|
2016-07-12 14:27:32 -04:00
|
|
|
if len(nodes) == 0 {
|
|
|
|
return nil, nil
|
|
|
|
}
|
|
|
|
|
2016-05-08 03:32:51 -04:00
|
|
|
logrus.Debugf("%s: Initiating bulk sync with nodes %v", nDB.config.NodeName, nodes)
|
|
|
|
var err error
|
|
|
|
var networks []string
|
|
|
|
for _, node := range nodes {
|
|
|
|
if node == nDB.config.NodeName {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
networks = nDB.findCommonNetworks(node)
|
|
|
|
err = nDB.bulkSyncNode(networks, node, true)
|
|
|
|
if err != nil {
|
|
|
|
err = fmt.Errorf("bulk sync failed on node %s: %v", node, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
return networks, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Bulk sync all the table entries belonging to a set of networks to a
|
|
|
|
// single peer node. It can be unsolicited or can be in response to an
|
|
|
|
// unsolicited bulk sync
|
|
|
|
func (nDB *NetworkDB) bulkSyncNode(networks []string, node string, unsolicited bool) error {
|
|
|
|
var msgs [][]byte
|
|
|
|
|
2016-09-20 21:21:19 -04:00
|
|
|
var unsolMsg string
|
|
|
|
if unsolicited {
|
|
|
|
unsolMsg = "unsolicited"
|
|
|
|
}
|
|
|
|
|
|
|
|
logrus.Debugf("%s: Initiating %s bulk sync for networks %v with node %s", nDB.config.NodeName, unsolMsg, networks, node)
|
2016-05-08 03:32:51 -04:00
|
|
|
|
|
|
|
nDB.RLock()
|
|
|
|
mnode := nDB.nodes[node]
|
|
|
|
if mnode == nil {
|
|
|
|
nDB.RUnlock()
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, nid := range networks {
|
|
|
|
nDB.indexes[byNetwork].WalkPrefix(fmt.Sprintf("/%s", nid), func(path string, v interface{}) bool {
|
|
|
|
entry, ok := v.(*entry)
|
|
|
|
if !ok {
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
2016-09-20 21:21:19 -04:00
|
|
|
eType := TableEventTypeCreate
|
2016-06-18 22:16:51 -04:00
|
|
|
if entry.deleting {
|
2016-09-20 21:21:19 -04:00
|
|
|
eType = TableEventTypeDelete
|
2016-06-18 22:16:51 -04:00
|
|
|
}
|
|
|
|
|
2016-05-08 03:32:51 -04:00
|
|
|
params := strings.Split(path[1:], "/")
|
2016-06-07 17:28:28 -04:00
|
|
|
tEvent := TableEvent{
|
2016-09-20 21:21:19 -04:00
|
|
|
Type: eType,
|
2016-05-08 03:32:51 -04:00
|
|
|
LTime: entry.ltime,
|
|
|
|
NodeName: entry.node,
|
|
|
|
NetworkID: nid,
|
|
|
|
TableName: params[1],
|
|
|
|
Key: params[2],
|
|
|
|
Value: entry.value,
|
|
|
|
}
|
|
|
|
|
2016-06-07 17:28:28 -04:00
|
|
|
msg, err := encodeMessage(MessageTypeTableEvent, &tEvent)
|
2016-05-08 03:32:51 -04:00
|
|
|
if err != nil {
|
|
|
|
logrus.Errorf("Encode failure during bulk sync: %#v", tEvent)
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
|
|
|
msgs = append(msgs, msg)
|
|
|
|
return false
|
|
|
|
})
|
|
|
|
}
|
|
|
|
nDB.RUnlock()
|
|
|
|
|
|
|
|
// Create a compound message
|
|
|
|
compound := makeCompoundMessage(msgs)
|
|
|
|
|
2016-06-07 17:28:28 -04:00
|
|
|
bsm := BulkSyncMessage{
|
2016-05-08 03:32:51 -04:00
|
|
|
LTime: nDB.tableClock.Time(),
|
|
|
|
Unsolicited: unsolicited,
|
|
|
|
NodeName: nDB.config.NodeName,
|
|
|
|
Networks: networks,
|
2016-06-07 17:28:28 -04:00
|
|
|
Payload: compound,
|
2016-05-08 03:32:51 -04:00
|
|
|
}
|
|
|
|
|
2016-06-07 17:28:28 -04:00
|
|
|
buf, err := encodeMessage(MessageTypeBulkSync, &bsm)
|
2016-05-08 03:32:51 -04:00
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("failed to encode bulk sync message: %v", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
nDB.Lock()
|
|
|
|
ch := make(chan struct{})
|
|
|
|
nDB.bulkSyncAckTbl[node] = ch
|
|
|
|
nDB.Unlock()
|
|
|
|
|
2016-09-20 21:21:19 -04:00
|
|
|
err = nDB.memberlist.SendToTCP(&mnode.Node, buf)
|
2016-05-08 03:32:51 -04:00
|
|
|
if err != nil {
|
|
|
|
nDB.Lock()
|
|
|
|
delete(nDB.bulkSyncAckTbl, node)
|
|
|
|
nDB.Unlock()
|
|
|
|
|
|
|
|
return fmt.Errorf("failed to send a TCP message during bulk sync: %v", err)
|
|
|
|
}
|
|
|
|
|
2016-06-07 17:28:28 -04:00
|
|
|
// Wait on a response only if it is unsolicited.
|
|
|
|
if unsolicited {
|
|
|
|
startTime := time.Now()
|
2016-06-16 17:32:12 -04:00
|
|
|
t := time.NewTimer(30 * time.Second)
|
2016-06-07 17:28:28 -04:00
|
|
|
select {
|
2016-06-16 17:32:12 -04:00
|
|
|
case <-t.C:
|
2016-06-07 17:28:28 -04:00
|
|
|
logrus.Errorf("Bulk sync to node %s timed out", node)
|
|
|
|
case <-ch:
|
|
|
|
logrus.Debugf("%s: Bulk sync to node %s took %s", nDB.config.NodeName, node, time.Now().Sub(startTime))
|
|
|
|
}
|
2016-06-16 17:32:12 -04:00
|
|
|
t.Stop()
|
2016-05-08 03:32:51 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Returns a random offset between 0 and n
|
|
|
|
func randomOffset(n int) int {
|
|
|
|
if n == 0 {
|
|
|
|
return 0
|
|
|
|
}
|
|
|
|
|
|
|
|
val, err := rand.Int(rand.Reader, big.NewInt(int64(n)))
|
|
|
|
if err != nil {
|
|
|
|
logrus.Errorf("Failed to get a random offset: %v", err)
|
|
|
|
return 0
|
|
|
|
}
|
|
|
|
|
|
|
|
return int(val.Int64())
|
|
|
|
}
|
|
|
|
|
|
|
|
// mRandomNodes is used to select up to m random nodes. It is possible
|
|
|
|
// that less than m nodes are returned.
|
|
|
|
func (nDB *NetworkDB) mRandomNodes(m int, nodes []string) []string {
|
|
|
|
n := len(nodes)
|
|
|
|
mNodes := make([]string, 0, m)
|
|
|
|
OUTER:
|
|
|
|
// Probe up to 3*n times, with large n this is not necessary
|
|
|
|
// since k << n, but with small n we want search to be
|
|
|
|
// exhaustive
|
|
|
|
for i := 0; i < 3*n && len(mNodes) < m; i++ {
|
|
|
|
// Get random node
|
|
|
|
idx := randomOffset(n)
|
|
|
|
node := nodes[idx]
|
|
|
|
|
|
|
|
if node == nDB.config.NodeName {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
// Check if we have this node already
|
|
|
|
for j := 0; j < len(mNodes); j++ {
|
|
|
|
if node == mNodes[j] {
|
|
|
|
continue OUTER
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Append the node
|
|
|
|
mNodes = append(mNodes, node)
|
|
|
|
}
|
|
|
|
|
|
|
|
return mNodes
|
|
|
|
}
|