Updated vendor for exported BuildArgs

vendor/github.com/docker/libnetwork/networkdb/broadcast.go

@@ -0,0 +1,174 @@
+package networkdb
+
+import (
+	"errors"
+	"time"
+
+	"github.com/hashicorp/memberlist"
+	"github.com/hashicorp/serf/serf"
+)
+
+const broadcastTimeout = 5 * time.Second
+
+type networkEventMessage struct {
+	id   string
+	node string
+	msg  []byte
+}
+
+func (m *networkEventMessage) Invalidates(other memberlist.Broadcast) bool {
+	otherm := other.(*networkEventMessage)
+	return m.id == otherm.id && m.node == otherm.node
+}
+
+func (m *networkEventMessage) Message() []byte {
+	return m.msg
+}
+
+func (m *networkEventMessage) Finished() {
+}
+
+func (nDB *NetworkDB) sendNetworkEvent(nid string, event NetworkEvent_Type, ltime serf.LamportTime) error {
+	nEvent := NetworkEvent{
+		Type:      event,
+		LTime:     ltime,
+		NodeName:  nDB.config.NodeID,
+		NetworkID: nid,
+	}
+
+	raw, err := encodeMessage(MessageTypeNetworkEvent, &nEvent)
+	if err != nil {
+		return err
+	}
+
+	nDB.networkBroadcasts.QueueBroadcast(&networkEventMessage{
+		msg:  raw,
+		id:   nid,
+		node: nDB.config.NodeID,
+	})
+	return nil
+}
+
+type nodeEventMessage struct {
+	msg    []byte
+	notify chan<- struct{}
+}
+
+func (m *nodeEventMessage) Invalidates(other memberlist.Broadcast) bool {
+	return false
+}
+
+func (m *nodeEventMessage) Message() []byte {
+	return m.msg
+}
+
+func (m *nodeEventMessage) Finished() {
+	if m.notify != nil {
+		close(m.notify)
+	}
+}
+
+func (nDB *NetworkDB) sendNodeEvent(event NodeEvent_Type) error {
+	nEvent := NodeEvent{
+		Type:     event,
+		LTime:    nDB.networkClock.Increment(),
+		NodeName: nDB.config.NodeID,
+	}
+
+	raw, err := encodeMessage(MessageTypeNodeEvent, &nEvent)
+	if err != nil {
+		return err
+	}
+
+	notifyCh := make(chan struct{})
+	nDB.nodeBroadcasts.QueueBroadcast(&nodeEventMessage{
+		msg:    raw,
+		notify: notifyCh,
+	})
+
+	nDB.RLock()
+	noPeers := len(nDB.nodes) <= 1
+	nDB.RUnlock()
+
+	// Message enqueued, do not wait for a send if no peer is present
+	if noPeers {
+		return nil
+	}
+
+	// Wait for the broadcast
+	select {
+	case <-notifyCh:
+	case <-time.After(broadcastTimeout):
+		return errors.New("timed out broadcasting node event")
+	}
+
+	return nil
+}
+
+type tableEventMessage struct {
+	id    string
+	tname string
+	key   string
+	msg   []byte
+	node  string
+}
+
+func (m *tableEventMessage) Invalidates(other memberlist.Broadcast) bool {
+	otherm := other.(*tableEventMessage)
+	return m.tname == otherm.tname && m.id == otherm.id && m.key == otherm.key
+}
+
+func (m *tableEventMessage) Message() []byte {
+	return m.msg
+}
+
+func (m *tableEventMessage) Finished() {
+}
+
+func (nDB *NetworkDB) sendTableEvent(event TableEvent_Type, nid string, tname string, key string, entry *entry) error {
+	tEvent := TableEvent{
+		Type:      event,
+		LTime:     entry.ltime,
+		NodeName:  nDB.config.NodeID,
+		NetworkID: nid,
+		TableName: tname,
+		Key:       key,
+		Value:     entry.value,
+		// The duration in second is a float that below would be truncated
+		ResidualReapTime: int32(entry.reapTime.Seconds()),
+	}
+
+	raw, err := encodeMessage(MessageTypeTableEvent, &tEvent)
+	if err != nil {
+		return err
+	}
+
+	var broadcastQ *memberlist.TransmitLimitedQueue
+	nDB.RLock()
+	thisNodeNetworks, ok := nDB.networks[nDB.config.NodeID]
+	if ok {
+		// The network may have been removed
+		network, networkOk := thisNodeNetworks[nid]
+		if !networkOk {
+			nDB.RUnlock()
+			return nil
+		}
+
+		broadcastQ = network.tableBroadcasts
+	}
+	nDB.RUnlock()
+
+	// The network may have been removed
+	if broadcastQ == nil {
+		return nil
+	}
+
+	broadcastQ.QueueBroadcast(&tableEventMessage{
+		msg:   raw,
+		id:    nid,
+		tname: tname,
+		key:   key,
+		node:  nDB.config.NodeID,
+	})
+	return nil
+}

vendor/github.com/docker/libnetwork/networkdb/cluster.go

@@ -0,0 +1,731 @@
+package networkdb
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"fmt"
+	"log"
+	"math/big"
+	rnd "math/rand"
+	"net"
+	"strings"
+	"time"
+
+	"github.com/hashicorp/memberlist"
+	"github.com/sirupsen/logrus"
+)
+
+const (
+	reapPeriod            = 5 * time.Second
+	rejoinClusterDuration = 10 * time.Second
+	rejoinInterval        = 60 * time.Second
+	retryInterval         = 1 * time.Second
+	nodeReapInterval      = 24 * time.Hour
+	nodeReapPeriod        = 2 * time.Hour
+)
+
+type logWriter struct{}
+
+func (l *logWriter) Write(p []byte) (int, error) {
+	str := string(p)
+	str = strings.TrimSuffix(str, "\n")
+
+	switch {
+	case strings.HasPrefix(str, "[WARN] "):
+		str = strings.TrimPrefix(str, "[WARN] ")
+		logrus.Warn(str)
+	case strings.HasPrefix(str, "[DEBUG] "):
+		str = strings.TrimPrefix(str, "[DEBUG] ")
+		logrus.Debug(str)
+	case strings.HasPrefix(str, "[INFO] "):
+		str = strings.TrimPrefix(str, "[INFO] ")
+		logrus.Info(str)
+	case strings.HasPrefix(str, "[ERR] "):
+		str = strings.TrimPrefix(str, "[ERR] ")
+		logrus.Warn(str)
+	}
+
+	return len(p), nil
+}
+
+// SetKey adds a new key to the key ring
+func (nDB *NetworkDB) SetKey(key []byte) {
+	logrus.Debugf("Adding key %s", hex.EncodeToString(key)[0:5])
+	nDB.Lock()
+	defer nDB.Unlock()
+	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) {
+	logrus.Debugf("Primary Key %s", hex.EncodeToString(key)[0:5])
+	nDB.RLock()
+	defer nDB.RUnlock()
+	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) {
+	logrus.Debugf("Remove Key %s", hex.EncodeToString(key)[0:5])
+	nDB.Lock()
+	defer nDB.Unlock()
+	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
+		}
+	}
+}
+
+func (nDB *NetworkDB) clusterInit() error {
+	nDB.lastStatsTimestamp = time.Now()
+	nDB.lastHealthTimestamp = nDB.lastStatsTimestamp
+
+	config := memberlist.DefaultLANConfig()
+	config.Name = nDB.config.NodeID
+	config.BindAddr = nDB.config.BindAddr
+	config.AdvertiseAddr = nDB.config.AdvertiseAddr
+	config.UDPBufferSize = nDB.config.PacketBufferSize
+
+	if nDB.config.BindPort != 0 {
+		config.BindPort = nDB.config.BindPort
+	}
+
+	config.ProtocolVersion = memberlist.ProtocolVersion2Compatible
+	config.Delegate = &delegate{nDB: nDB}
+	config.Events = &eventDelegate{nDB: nDB}
+	// custom logger that does not add time or date, so they are not
+	// duplicated by logrus
+	config.Logger = log.New(&logWriter{}, "", 0)
+
+	var err error
+	if len(nDB.config.Keys) > 0 {
+		for i, key := range nDB.config.Keys {
+			logrus.Debugf("Encryption key %d: %s", i+1, hex.EncodeToString(key)[0:5])
+		}
+		nDB.keyring, err = memberlist.NewKeyring(nDB.config.Keys, nDB.config.Keys[0])
+		if err != nil {
+			return err
+		}
+		config.Keyring = nDB.keyring
+	}
+
+	nDB.networkBroadcasts = &memberlist.TransmitLimitedQueue{
+		NumNodes: func() int {
+			nDB.RLock()
+			num := len(nDB.nodes)
+			nDB.RUnlock()
+			return num
+		},
+		RetransmitMult: config.RetransmitMult,
+	}
+
+	nDB.nodeBroadcasts = &memberlist.TransmitLimitedQueue{
+		NumNodes: func() int {
+			nDB.RLock()
+			num := len(nDB.nodes)
+			nDB.RUnlock()
+			return num
+		},
+		RetransmitMult: config.RetransmitMult,
+	}
+
+	mlist, err := memberlist.Create(config)
+	if err != nil {
+		return fmt.Errorf("failed to create memberlist: %v", err)
+	}
+
+	nDB.ctx, nDB.cancelCtx = context.WithCancel(context.Background())
+	nDB.memberlist = mlist
+
+	for _, trigger := range []struct {
+		interval time.Duration
+		fn       func()
+	}{
+		{reapPeriod, nDB.reapState},
+		{config.GossipInterval, nDB.gossip},
+		{config.PushPullInterval, nDB.bulkSyncTables},
+		{retryInterval, nDB.reconnectNode},
+		{nodeReapPeriod, nDB.reapDeadNode},
+		{rejoinInterval, nDB.rejoinClusterBootStrap},
+	} {
+		t := time.NewTicker(trigger.interval)
+		go nDB.triggerFunc(trigger.interval, t.C, trigger.fn)
+		nDB.tickers = append(nDB.tickers, t)
+	}
+
+	return nil
+}
+
+func (nDB *NetworkDB) retryJoin(ctx context.Context, members []string) {
+	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
+			}
+			if err := nDB.sendNodeEvent(NodeEventTypeJoin); err != nil {
+				logrus.Errorf("failed to send node join on retry: %v", err)
+				continue
+			}
+			return
+		case <-ctx.Done():
+			return
+		}
+	}
+
+}
+
+func (nDB *NetworkDB) clusterJoin(members []string) error {
+	mlist := nDB.memberlist
+
+	if _, err := mlist.Join(members); err != nil {
+		// In case of failure, we no longer need to explicitly call retryJoin.
+		// rejoinClusterBootStrap, which runs every minute, will retryJoin for 10sec
+		return fmt.Errorf("could not join node to memberlist: %v", err)
+	}
+
+	if err := nDB.sendNodeEvent(NodeEventTypeJoin); err != nil {
+		return fmt.Errorf("failed to send node join: %v", err)
+	}
+
+	return nil
+}
+
+func (nDB *NetworkDB) clusterLeave() error {
+	mlist := nDB.memberlist
+
+	if err := nDB.sendNodeEvent(NodeEventTypeLeave); err != nil {
+		logrus.Errorf("failed to send node leave: %v", err)
+	}
+
+	if err := mlist.Leave(time.Second); err != nil {
+		return err
+	}
+
+	// cancel the context
+	nDB.cancelCtx()
+
+	for _, t := range nDB.tickers {
+		t.Stop()
+	}
+
+	return mlist.Shutdown()
+}
+
+func (nDB *NetworkDB) triggerFunc(stagger time.Duration, C <-chan time.Time, f func()) {
+	// Use a random stagger to avoid syncronizing
+	randStagger := time.Duration(uint64(rnd.Int63()) % uint64(stagger))
+	select {
+	case <-time.After(randStagger):
+	case <-nDB.ctx.Done():
+		return
+	}
+	for {
+		select {
+		case <-C:
+			f()
+		case <-nDB.ctx.Done():
+			return
+		}
+	}
+}
+
+func (nDB *NetworkDB) reapDeadNode() {
+	nDB.Lock()
+	defer nDB.Unlock()
+	for _, nodeMap := range []map[string]*node{
+		nDB.failedNodes,
+		nDB.leftNodes,
+	} {
+		for id, n := range nodeMap {
+			if n.reapTime > nodeReapPeriod {
+				n.reapTime -= nodeReapPeriod
+				continue
+			}
+			logrus.Debugf("Garbage collect node %v", n.Name)
+			delete(nodeMap, id)
+		}
+	}
+}
+
+// rejoinClusterBootStrap is called periodically to check if all bootStrap nodes are active in the cluster,
+// if not, call the cluster join to merge 2 separate clusters that are formed when all managers
+// stopped/started at the same time
+func (nDB *NetworkDB) rejoinClusterBootStrap() {
+	nDB.RLock()
+	if len(nDB.bootStrapIP) == 0 {
+		nDB.RUnlock()
+		return
+	}
+
+	bootStrapIPs := make([]string, 0, len(nDB.bootStrapIP))
+	for _, bootIP := range nDB.bootStrapIP {
+		for _, node := range nDB.nodes {
+			if node.Addr.Equal(bootIP) {
+				// One of the bootstrap nodes is part of the cluster, return
+				nDB.RUnlock()
+				return
+			}
+		}
+		bootStrapIPs = append(bootStrapIPs, bootIP.String())
+	}
+	nDB.RUnlock()
+	// None of the bootStrap nodes are in the cluster, call memberlist join
+	logrus.Debugf("rejoinClusterBootStrap, calling cluster join with bootStrap %v", bootStrapIPs)
+	ctx, cancel := context.WithTimeout(nDB.ctx, rejoinClusterDuration)
+	defer cancel()
+	nDB.retryJoin(ctx, bootStrapIPs)
+}
+
+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 {
+		return
+	}
+
+	logrus.Debugf("Initiating bulk sync with node %s after reconnect", node.Name)
+	nDB.bulkSync([]string{node.Name}, true)
+}
+
+// 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.
+func (nDB *NetworkDB) reapState() {
+	// The reapTableEntries leverage the presence of the network so garbage collect entries first
+	nDB.reapTableEntries()
+	nDB.reapNetworks()
+}
+
+func (nDB *NetworkDB) reapNetworks() {
+	nDB.Lock()
+	for _, nn := range nDB.networks {
+		for id, n := range nn {
+			if n.leaving {
+				if n.reapTime <= 0 {
+					delete(nn, id)
+					continue
+				}
+				n.reapTime -= reapPeriod
+			}
+		}
+	}
+	nDB.Unlock()
+}
+
+func (nDB *NetworkDB) reapTableEntries() {
+	var nodeNetworks []string
+	// This is best effort, if the list of network changes will be picked up in the next cycle
+	nDB.RLock()
+	for nid := range nDB.networks[nDB.config.NodeID] {
+		nodeNetworks = append(nodeNetworks, nid)
+	}
+	nDB.RUnlock()
+
+	cycleStart := time.Now()
+	// In order to avoid blocking the database for a long time, apply the garbage collection logic by network
+	// The lock is taken at the beginning of the cycle and the deletion is inline
+	for _, nid := range nodeNetworks {
+		nDB.Lock()
+		nDB.indexes[byNetwork].WalkPrefix(fmt.Sprintf("/%s", nid), func(path string, v interface{}) bool {
+			// timeCompensation compensate in case the lock took some time to be released
+			timeCompensation := time.Since(cycleStart)
+			entry, ok := v.(*entry)
+			if !ok || !entry.deleting {
+				return false
+			}
+
+			// In this check we are adding an extra 1 second to guarantee that when the number is truncated to int32 to fit the packet
+			// for the tableEvent the number is always strictly > 1 and never 0
+			if entry.reapTime > reapPeriod+timeCompensation+time.Second {
+				entry.reapTime -= reapPeriod + timeCompensation
+				return false
+			}
+
+			params := strings.Split(path[1:], "/")
+			nid := params[0]
+			tname := params[1]
+			key := params[2]
+
+			okTable, okNetwork := nDB.deleteEntry(nid, tname, key)
+			if !okTable {
+				logrus.Errorf("Table tree delete failed, entry with key:%s does not exists in the table:%s network:%s", key, tname, nid)
+			}
+			if !okNetwork {
+				logrus.Errorf("Network tree delete failed, entry with key:%s does not exists in the network:%s table:%s", key, nid, tname)
+			}
+
+			return false
+		})
+		nDB.Unlock()
+	}
+}
+
+func (nDB *NetworkDB) gossip() {
+	networkNodes := make(map[string][]string)
+	nDB.RLock()
+	thisNodeNetworks := nDB.networks[nDB.config.NodeID]
+	for nid := range thisNodeNetworks {
+		networkNodes[nid] = nDB.networkNodes[nid]
+	}
+	printStats := time.Since(nDB.lastStatsTimestamp) >= nDB.config.StatsPrintPeriod
+	printHealth := time.Since(nDB.lastHealthTimestamp) >= nDB.config.HealthPrintPeriod
+	nDB.RUnlock()
+
+	if printHealth {
+		healthScore := nDB.memberlist.GetHealthScore()
+		if healthScore != 0 {
+			logrus.Warnf("NetworkDB stats %v(%v) - healthscore:%d (connectivity issues)", nDB.config.Hostname, nDB.config.NodeID, healthScore)
+		}
+		nDB.lastHealthTimestamp = time.Now()
+	}
+
+	for nid, nodes := range networkNodes {
+		mNodes := nDB.mRandomNodes(3, nodes)
+		bytesAvail := nDB.config.PacketBufferSize - compoundHeaderOverhead
+
+		nDB.RLock()
+		network, ok := thisNodeNetworks[nid]
+		nDB.RUnlock()
+		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
+
+		if broadcastQ == nil {
+			logrus.Errorf("Invalid broadcastQ encountered while gossiping for network %s", nid)
+			continue
+		}
+
+		msgs := broadcastQ.GetBroadcasts(compoundOverhead, bytesAvail)
+		// Collect stats and print the queue info, note this code is here also to have a view of the queues empty
+		network.qMessagesSent += len(msgs)
+		if printStats {
+			logrus.Infof("NetworkDB stats %v(%v) - netID:%s leaving:%t netPeers:%d entries:%d Queue qLen:%d netMsg/s:%d",
+				nDB.config.Hostname, nDB.config.NodeID,
+				nid, network.leaving, broadcastQ.NumNodes(), network.entriesNumber, broadcastQ.NumQueued(),
+				network.qMessagesSent/int((nDB.config.StatsPrintPeriod/time.Second)))
+			network.qMessagesSent = 0
+		}
+
+		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
+			if err := nDB.memberlist.SendBestEffort(&mnode.Node, compound); err != nil {
+				logrus.Errorf("Failed to send gossip to %s: %s", mnode.Addr, err)
+			}
+		}
+	}
+	// Reset the stats
+	if printStats {
+		nDB.lastStatsTimestamp = time.Now()
+	}
+}
+
+func (nDB *NetworkDB) bulkSyncTables() {
+	var networks []string
+	nDB.RLock()
+	for nid, network := range nDB.networks[nDB.config.NodeID] {
+		if network.leaving {
+			continue
+		}
+		networks = append(networks, nid)
+	}
+	nDB.RUnlock()
+
+	for {
+		if len(networks) == 0 {
+			break
+		}
+
+		nid := networks[0]
+		networks = networks[1:]
+
+		nDB.RLock()
+		nodes := nDB.networkNodes[nid]
+		nDB.RUnlock()
+
+		// No peer nodes on this network. Move on.
+		if len(nodes) == 0 {
+			continue
+		}
+
+		completed, err := nDB.bulkSync(nodes, false)
+		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 {
+			var found bool
+			for _, completedNid := range completed {
+				if nid == completedNid {
+					found = true
+					break
+				}
+			}
+
+			if !found {
+				updatedNetworks = append(updatedNetworks, nid)
+			}
+		}
+
+		networks = updatedNetworks
+	}
+}
+
+func (nDB *NetworkDB) bulkSync(nodes []string, all bool) ([]string, error) {
+	if !all {
+		// Get 2 random nodes. 2nd node will be tried if the bulk sync to
+		// 1st node fails.
+		nodes = nDB.mRandomNodes(2, nodes)
+	}
+
+	if len(nodes) == 0 {
+		return nil, nil
+	}
+
+	var err error
+	var networks []string
+	for _, node := range nodes {
+		if node == nDB.config.NodeID {
+			continue
+		}
+		logrus.Debugf("%v(%v): Initiating bulk sync with node %v", nDB.config.Hostname, nDB.config.NodeID, node)
+		networks = nDB.findCommonNetworks(node)
+		err = nDB.bulkSyncNode(networks, node, true)
+		// if its periodic bulksync stop after the first successful sync
+		if !all && err == nil {
+			break
+		}
+		if err != nil {
+			err = fmt.Errorf("bulk sync to node %s failed: %v", node, err)
+			logrus.Warn(err.Error())
+		}
+	}
+
+	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
+
+	var unsolMsg string
+	if unsolicited {
+		unsolMsg = "unsolicited"
+	}
+
+	logrus.Debugf("%v(%v): Initiating %s bulk sync for networks %v with node %s",
+		nDB.config.Hostname, nDB.config.NodeID, unsolMsg, networks, node)
+
+	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
+			}
+
+			eType := TableEventTypeCreate
+			if entry.deleting {
+				eType = TableEventTypeDelete
+			}
+
+			params := strings.Split(path[1:], "/")
+			tEvent := TableEvent{
+				Type:      eType,
+				LTime:     entry.ltime,
+				NodeName:  entry.node,
+				NetworkID: nid,
+				TableName: params[1],
+				Key:       params[2],
+				Value:     entry.value,
+				// The duration in second is a float that below would be truncated
+				ResidualReapTime: int32(entry.reapTime.Seconds()),
+			}
+
+			msg, err := encodeMessage(MessageTypeTableEvent, &tEvent)
+			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)
+
+	bsm := BulkSyncMessage{
+		LTime:       nDB.tableClock.Time(),
+		Unsolicited: unsolicited,
+		NodeName:    nDB.config.NodeID,
+		Networks:    networks,
+		Payload:     compound,
+	}
+
+	buf, err := encodeMessage(MessageTypeBulkSync, &bsm)
+	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()
+
+	err = nDB.memberlist.SendReliable(&mnode.Node, buf)
+	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)
+	}
+
+	// Wait on a response only if it is unsolicited.
+	if unsolicited {
+		startTime := time.Now()
+		t := time.NewTimer(30 * time.Second)
+		select {
+		case <-t.C:
+			logrus.Errorf("Bulk sync to node %s timed out", node)
+		case <-ch:
+			logrus.Debugf("%v(%v): Bulk sync to node %s took %s", nDB.config.Hostname, nDB.config.NodeID, node, time.Since(startTime))
+		}
+		t.Stop()
+	}
+
+	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.NodeID {
+			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
+}

vendor/github.com/docker/libnetwork/networkdb/delegate.go

@@ -0,0 +1,476 @@
+package networkdb
+
+import (
+	"net"
+	"time"
+
+	"github.com/gogo/protobuf/proto"
+	"github.com/sirupsen/logrus"
+)
+
+type delegate struct {
+	nDB *NetworkDB
+}
+
+func (d *delegate) NodeMeta(limit int) []byte {
+	return []byte{}
+}
+
+func (nDB *NetworkDB) handleNodeEvent(nEvent *NodeEvent) bool {
+	// Update our local clock if the received messages has newer
+	// time.
+	nDB.networkClock.Witness(nEvent.LTime)
+
+	nDB.Lock()
+	defer nDB.Unlock()
+
+	// check if the node exists
+	n, _, _ := nDB.findNode(nEvent.NodeName)
+	if n == nil {
+		return false
+	}
+
+	// check if the event is fresh
+	if n.ltime >= nEvent.LTime {
+		return false
+	}
+
+	// If we are here means that the event is fresher and the node is known. Update the laport time
+	n.ltime = nEvent.LTime
+
+	// If the node is not known from memberlist we cannot process save any state of it else if it actually
+	// dies we won't receive any notification and we will remain stuck with it
+	if _, ok := nDB.nodes[nEvent.NodeName]; !ok {
+		logrus.Error("node: %s is unknown to memberlist", nEvent.NodeName)
+		return false
+	}
+
+	switch nEvent.Type {
+	case NodeEventTypeJoin:
+		moved, err := nDB.changeNodeState(n.Name, nodeActiveState)
+		if err != nil {
+			logrus.WithError(err).Error("unable to find the node to move")
+			return false
+		}
+		if moved {
+			logrus.Infof("%v(%v): Node join event for %s/%s", nDB.config.Hostname, nDB.config.NodeID, n.Name, n.Addr)
+		}
+		return moved
+	case NodeEventTypeLeave:
+		moved, err := nDB.changeNodeState(n.Name, nodeLeftState)
+		if err != nil {
+			logrus.WithError(err).Error("unable to find the node to move")
+			return false
+		}
+		if moved {
+			logrus.Infof("%v(%v): Node leave event for %s/%s", nDB.config.Hostname, nDB.config.NodeID, n.Name, n.Addr)
+		}
+		return moved
+	}
+
+	return false
+}
+
+func (nDB *NetworkDB) handleNetworkEvent(nEvent *NetworkEvent) bool {
+	// Update our local clock if the received messages has newer
+	// time.
+	nDB.networkClock.Witness(nEvent.LTime)
+
+	nDB.Lock()
+	defer nDB.Unlock()
+
+	if nEvent.NodeName == nDB.config.NodeID {
+		return false
+	}
+
+	nodeNetworks, ok := nDB.networks[nEvent.NodeName]
+	if !ok {
+		// We haven't heard about this node at all.  Ignore the leave
+		if nEvent.Type == NetworkEventTypeLeave {
+			return false
+		}
+
+		nodeNetworks = make(map[string]*network)
+		nDB.networks[nEvent.NodeName] = nodeNetworks
+	}
+
+	if n, ok := nodeNetworks[nEvent.NetworkID]; ok {
+		// We have the latest state. Ignore the event
+		// since it is stale.
+		if n.ltime >= nEvent.LTime {
+			return false
+		}
+
+		n.ltime = nEvent.LTime
+		n.leaving = nEvent.Type == NetworkEventTypeLeave
+		if n.leaving {
+			n.reapTime = nDB.config.reapNetworkInterval
+
+			// The remote node is leaving the network, but not the gossip cluster.
+			// Mark all its entries in deleted state, this will guarantee that
+			// if some node bulk sync with us, the deleted state of
+			// these entries will be propagated.
+			nDB.deleteNodeNetworkEntries(nEvent.NetworkID, nEvent.NodeName)
+		}
+
+		if nEvent.Type == NetworkEventTypeLeave {
+			nDB.deleteNetworkNode(nEvent.NetworkID, nEvent.NodeName)
+		} else {
+			nDB.addNetworkNode(nEvent.NetworkID, nEvent.NodeName)
+		}
+
+		return true
+	}
+
+	if nEvent.Type == NetworkEventTypeLeave {
+		return false
+	}
+
+	// If the node is not known from memberlist we cannot process save any state of it else if it actually
+	// dies we won't receive any notification and we will remain stuck with it
+	if _, ok := nDB.nodes[nEvent.NodeName]; !ok {
+		return false
+	}
+
+	// This remote network join is being seen the first time.
+	nodeNetworks[nEvent.NetworkID] = &network{
+		id:    nEvent.NetworkID,
+		ltime: nEvent.LTime,
+	}
+
+	nDB.addNetworkNode(nEvent.NetworkID, nEvent.NodeName)
+	return true
+}
+
+func (nDB *NetworkDB) handleTableEvent(tEvent *TableEvent) bool {
+	// Update our local clock if the received messages has newer time.
+	nDB.tableClock.Witness(tEvent.LTime)
+
+	// Ignore the table events for networks that are in the process of going away
+	nDB.RLock()
+	networks := nDB.networks[nDB.config.NodeID]
+	network, ok := networks[tEvent.NetworkID]
+	// Check if the owner of the event is still part of the network
+	nodes := nDB.networkNodes[tEvent.NetworkID]
+	var nodePresent bool
+	for _, node := range nodes {
+		if node == tEvent.NodeName {
+			nodePresent = true
+			break
+		}
+	}
+	nDB.RUnlock()
+
+	if !ok || network.leaving || !nodePresent {
+		// I'm out of the network OR the event owner is not anymore part of the network so do not propagate
+		return false
+	}
+
+	nDB.Lock()
+	e, err := nDB.getEntry(tEvent.TableName, tEvent.NetworkID, tEvent.Key)
+	if err == nil {
+		// We have the latest state. Ignore the event
+		// since it is stale.
+		if e.ltime >= tEvent.LTime {
+			nDB.Unlock()
+			return false
+		}
+	}
+
+	e = &entry{
+		ltime:    tEvent.LTime,
+		node:     tEvent.NodeName,
+		value:    tEvent.Value,
+		deleting: tEvent.Type == TableEventTypeDelete,
+		reapTime: time.Duration(tEvent.ResidualReapTime) * time.Second,
+	}
+
+	// All the entries marked for deletion should have a reapTime set greater than 0
+	// This case can happen if the cluster is running different versions of the engine where the old version does not have the
+	// field. If that is not the case, this can be a BUG
+	if e.deleting && e.reapTime == 0 {
+		logrus.Warnf("%v(%v) handleTableEvent object %+v has a 0 reapTime, is the cluster running the same docker engine version?",
+			nDB.config.Hostname, nDB.config.NodeID, tEvent)
+		e.reapTime = nDB.config.reapEntryInterval
+	}
+	nDB.createOrUpdateEntry(tEvent.NetworkID, tEvent.TableName, tEvent.Key, e)
+	nDB.Unlock()
+
+	if err != nil && tEvent.Type == TableEventTypeDelete {
+		// If it is a delete event and we did not have a state for it, don't propagate to the application
+		// If the residual reapTime is lower or equal to 1/6 of the total reapTime don't bother broadcasting it around
+		// most likely the cluster is already aware of it, if not who will sync with this node will catch the state too.
+		// This also avoids that deletion of entries close to their garbage collection ends up circuling around forever
+		return e.reapTime > nDB.config.reapEntryInterval/6
+	}
+
+	var op opType
+	switch tEvent.Type {
+	case TableEventTypeCreate:
+		op = opCreate
+	case TableEventTypeUpdate:
+		op = opUpdate
+	case TableEventTypeDelete:
+		op = opDelete
+	}
+
+	nDB.broadcaster.Write(makeEvent(op, tEvent.TableName, tEvent.NetworkID, tEvent.Key, tEvent.Value))
+	return true
+}
+
+func (nDB *NetworkDB) handleCompound(buf []byte, isBulkSync bool) {
+	// Decode the parts
+	parts, err := decodeCompoundMessage(buf)
+	if err != nil {
+		logrus.Errorf("Failed to decode compound request: %v", err)
+		return
+	}
+
+	// Handle each message
+	for _, part := range parts {
+		nDB.handleMessage(part, isBulkSync)
+	}
+}
+
+func (nDB *NetworkDB) handleTableMessage(buf []byte, isBulkSync bool) {
+	var tEvent TableEvent
+	if err := proto.Unmarshal(buf, &tEvent); err != nil {
+		logrus.Errorf("Error decoding table event message: %v", err)
+		return
+	}
+
+	// Ignore messages that this node generated.
+	if tEvent.NodeName == nDB.config.NodeID {
+		return
+	}
+
+	if rebroadcast := nDB.handleTableEvent(&tEvent); rebroadcast {
+		var err error
+		buf, err = encodeRawMessage(MessageTypeTableEvent, buf)
+		if err != nil {
+			logrus.Errorf("Error marshalling gossip message for network event rebroadcast: %v", err)
+			return
+		}
+
+		nDB.RLock()
+		n, ok := nDB.networks[nDB.config.NodeID][tEvent.NetworkID]
+		nDB.RUnlock()
+
+		// if the network is not there anymore, OR we are leaving the network OR the broadcast queue is not present
+		if !ok || n.leaving || n.tableBroadcasts == nil {
+			return
+		}
+
+		n.tableBroadcasts.QueueBroadcast(&tableEventMessage{
+			msg:   buf,
+			id:    tEvent.NetworkID,
+			tname: tEvent.TableName,
+			key:   tEvent.Key,
+			node:  tEvent.NodeName,
+		})
+	}
+}
+
+func (nDB *NetworkDB) handleNodeMessage(buf []byte) {
+	var nEvent NodeEvent
+	if err := proto.Unmarshal(buf, &nEvent); err != nil {
+		logrus.Errorf("Error decoding node event message: %v", err)
+		return
+	}
+
+	if rebroadcast := nDB.handleNodeEvent(&nEvent); rebroadcast {
+		var err error
+		buf, err = encodeRawMessage(MessageTypeNodeEvent, buf)
+		if err != nil {
+			logrus.Errorf("Error marshalling gossip message for node event rebroadcast: %v", err)
+			return
+		}
+
+		nDB.nodeBroadcasts.QueueBroadcast(&nodeEventMessage{
+			msg: buf,
+		})
+	}
+}
+
+func (nDB *NetworkDB) handleNetworkMessage(buf []byte) {
+	var nEvent NetworkEvent
+	if err := proto.Unmarshal(buf, &nEvent); err != nil {
+		logrus.Errorf("Error decoding network event message: %v", err)
+		return
+	}
+
+	if rebroadcast := nDB.handleNetworkEvent(&nEvent); rebroadcast {
+		var err error
+		buf, err = encodeRawMessage(MessageTypeNetworkEvent, buf)
+		if err != nil {
+			logrus.Errorf("Error marshalling gossip message for network event rebroadcast: %v", err)
+			return
+		}
+
+		nDB.networkBroadcasts.QueueBroadcast(&networkEventMessage{
+			msg:  buf,
+			id:   nEvent.NetworkID,
+			node: nEvent.NodeName,
+		})
+	}
+}
+
+func (nDB *NetworkDB) handleBulkSync(buf []byte) {
+	var bsm BulkSyncMessage
+	if err := proto.Unmarshal(buf, &bsm); err != nil {
+		logrus.Errorf("Error decoding bulk sync message: %v", err)
+		return
+	}
+
+	if bsm.LTime > 0 {
+		nDB.tableClock.Witness(bsm.LTime)
+	}
+
+	nDB.handleMessage(bsm.Payload, true)
+
+	// Don't respond to a bulk sync which was not unsolicited
+	if !bsm.Unsolicited {
+		nDB.Lock()
+		ch, ok := nDB.bulkSyncAckTbl[bsm.NodeName]
+		if ok {
+			close(ch)
+			delete(nDB.bulkSyncAckTbl, bsm.NodeName)
+		}
+		nDB.Unlock()
+
+		return
+	}
+
+	var nodeAddr net.IP
+	nDB.RLock()
+	if node, ok := nDB.nodes[bsm.NodeName]; ok {
+		nodeAddr = node.Addr
+	}
+	nDB.RUnlock()
+
+	if err := nDB.bulkSyncNode(bsm.Networks, bsm.NodeName, false); err != nil {
+		logrus.Errorf("Error in responding to bulk sync from node %s: %v", nodeAddr, err)
+	}
+}
+
+func (nDB *NetworkDB) handleMessage(buf []byte, isBulkSync bool) {
+	mType, data, err := decodeMessage(buf)
+	if err != nil {
+		logrus.Errorf("Error decoding gossip message to get message type: %v", err)
+		return
+	}
+
+	switch mType {
+	case MessageTypeNodeEvent:
+		nDB.handleNodeMessage(data)
+	case MessageTypeNetworkEvent:
+		nDB.handleNetworkMessage(data)
+	case MessageTypeTableEvent:
+		nDB.handleTableMessage(data, isBulkSync)
+	case MessageTypeBulkSync:
+		nDB.handleBulkSync(data)
+	case MessageTypeCompound:
+		nDB.handleCompound(data, isBulkSync)
+	default:
+		logrus.Errorf("%v(%v): unknown message type %d", nDB.config.Hostname, nDB.config.NodeID, mType)
+	}
+}
+
+func (d *delegate) NotifyMsg(buf []byte) {
+	if len(buf) == 0 {
+		return
+	}
+
+	d.nDB.handleMessage(buf, false)
+}
+
+func (d *delegate) GetBroadcasts(overhead, limit int) [][]byte {
+	msgs := d.nDB.networkBroadcasts.GetBroadcasts(overhead, limit)
+	msgs = append(msgs, d.nDB.nodeBroadcasts.GetBroadcasts(overhead, limit)...)
+	return msgs
+}
+
+func (d *delegate) LocalState(join bool) []byte {
+	if join {
+		// Update all the local node/network state to a new time to
+		// force update on the node we are trying to rejoin, just in
+		// case that node has these in leaving state still. This is
+		// facilitate fast convergence after recovering from a gossip
+		// failure.
+		d.nDB.updateLocalNetworkTime()
+	}
+
+	d.nDB.RLock()
+	defer d.nDB.RUnlock()
+
+	pp := NetworkPushPull{
+		LTime:    d.nDB.networkClock.Time(),
+		NodeName: d.nDB.config.NodeID,
+	}
+
+	for name, nn := range d.nDB.networks {
+		for _, n := range nn {
+			pp.Networks = append(pp.Networks, &NetworkEntry{
+				LTime:     n.ltime,
+				NetworkID: n.id,
+				NodeName:  name,
+				Leaving:   n.leaving,
+			})
+		}
+	}
+
+	buf, err := encodeMessage(MessageTypePushPull, &pp)
+	if err != nil {
+		logrus.Errorf("Failed to encode local network state: %v", err)
+		return nil
+	}
+
+	return buf
+}
+
+func (d *delegate) MergeRemoteState(buf []byte, isJoin bool) {
+	if len(buf) == 0 {
+		logrus.Error("zero byte remote network state received")
+		return
+	}
+
+	var gMsg GossipMessage
+	err := proto.Unmarshal(buf, &gMsg)
+	if err != nil {
+		logrus.Errorf("Error unmarshalling push pull message: %v", err)
+		return
+	}
+
+	if gMsg.Type != MessageTypePushPull {
+		logrus.Errorf("Invalid message type %v received from remote", buf[0])
+	}
+
+	pp := NetworkPushPull{}
+	if err := proto.Unmarshal(gMsg.Data, &pp); err != nil {
+		logrus.Errorf("Failed to decode remote network state: %v", err)
+		return
+	}
+
+	nodeEvent := &NodeEvent{
+		LTime:    pp.LTime,
+		NodeName: pp.NodeName,
+		Type:     NodeEventTypeJoin,
+	}
+	d.nDB.handleNodeEvent(nodeEvent)
+
+	for _, n := range pp.Networks {
+		nEvent := &NetworkEvent{
+			LTime:     n.LTime,
+			NodeName:  n.NodeName,
+			NetworkID: n.NetworkID,
+			Type:      NetworkEventTypeJoin,
+		}
+
+		if n.Leaving {
+			nEvent.Type = NetworkEventTypeLeave
+		}
+
+		d.nDB.handleNetworkEvent(nEvent)
+	}
+
+}

vendor/github.com/docker/libnetwork/networkdb/event_delegate.go

@@ -0,0 +1,72 @@
+package networkdb
+
+import (
+	"encoding/json"
+	"net"
+
+	"github.com/hashicorp/memberlist"
+	"github.com/sirupsen/logrus"
+)
+
+type eventDelegate struct {
+	nDB *NetworkDB
+}
+
+func (e *eventDelegate) broadcastNodeEvent(addr net.IP, op opType) {
+	value, err := json.Marshal(&NodeAddr{addr})
+	if err == nil {
+		e.nDB.broadcaster.Write(makeEvent(op, NodeTable, "", "", value))
+	} else {
+		logrus.Errorf("Error marshalling node broadcast event %s", addr.String())
+	}
+}
+
+func (e *eventDelegate) NotifyJoin(mn *memberlist.Node) {
+	logrus.Infof("Node %s/%s, joined gossip cluster", mn.Name, mn.Addr)
+	e.broadcastNodeEvent(mn.Addr, opCreate)
+	e.nDB.Lock()
+	defer e.nDB.Unlock()
+
+	// In case the node is rejoining after a failure or leave,
+	// just add the node back to active
+	if moved, _ := e.nDB.changeNodeState(mn.Name, nodeActiveState); moved {
+		return
+	}
+
+	// Every node has a unique ID
+	// Check on the base of the IP address if the new node that joined is actually a new incarnation of a previous
+	// failed or shutdown one
+	e.nDB.purgeReincarnation(mn)
+
+	e.nDB.nodes[mn.Name] = &node{Node: *mn}
+	logrus.Infof("Node %s/%s, added to nodes list", mn.Name, mn.Addr)
+}
+
+func (e *eventDelegate) NotifyLeave(mn *memberlist.Node) {
+	logrus.Infof("Node %s/%s, left gossip cluster", mn.Name, mn.Addr)
+	e.broadcastNodeEvent(mn.Addr, opDelete)
+
+	e.nDB.Lock()
+	defer e.nDB.Unlock()
+
+	n, currState, _ := e.nDB.findNode(mn.Name)
+	if n == nil {
+		logrus.Errorf("Node %s/%s not found in the node lists", mn.Name, mn.Addr)
+		return
+	}
+	// if the node was active means that did not send the leave cluster message, so it's probable that
+	// failed. Else would be already in the left list so nothing else has to be done
+	if currState == nodeActiveState {
+		moved, err := e.nDB.changeNodeState(mn.Name, nodeFailedState)
+		if err != nil {
+			logrus.WithError(err).Errorf("impossible condition, node %s/%s not present in the list", mn.Name, mn.Addr)
+			return
+		}
+		if moved {
+			logrus.Infof("Node %s/%s, added to failed nodes list", mn.Name, mn.Addr)
+		}
+	}
+}
+
+func (e *eventDelegate) NotifyUpdate(n *memberlist.Node) {
+}

vendor/github.com/docker/libnetwork/networkdb/message.go

@@ -0,0 +1,98 @@
+package networkdb
+
+import "github.com/gogo/protobuf/proto"
+
+const (
+	// Compound message header overhead 1 byte(message type) + 4
+	// bytes (num messages)
+	compoundHeaderOverhead = 5
+
+	// Overhead for each embedded message in a compound message 4
+	// bytes (len of embedded message)
+	compoundOverhead = 4
+)
+
+func encodeRawMessage(t MessageType, raw []byte) ([]byte, error) {
+	gMsg := GossipMessage{
+		Type: t,
+		Data: raw,
+	}
+
+	buf, err := proto.Marshal(&gMsg)
+	if err != nil {
+		return nil, err
+	}
+
+	return buf, nil
+}
+
+func encodeMessage(t MessageType, msg interface{}) ([]byte, error) {
+	buf, err := proto.Marshal(msg.(proto.Message))
+	if err != nil {
+		return nil, err
+	}
+
+	buf, err = encodeRawMessage(t, buf)
+	if err != nil {
+		return nil, err
+	}
+
+	return buf, nil
+}
+
+func decodeMessage(buf []byte) (MessageType, []byte, error) {
+	var gMsg GossipMessage
+
+	err := proto.Unmarshal(buf, &gMsg)
+	if err != nil {
+		return MessageTypeInvalid, nil, err
+	}
+
+	return gMsg.Type, gMsg.Data, nil
+}
+
+// makeCompoundMessage takes a list of messages and generates
+// a single compound message containing all of them
+func makeCompoundMessage(msgs [][]byte) []byte {
+	cMsg := CompoundMessage{}
+
+	cMsg.Messages = make([]*CompoundMessage_SimpleMessage, 0, len(msgs))
+	for _, m := range msgs {
+		cMsg.Messages = append(cMsg.Messages, &CompoundMessage_SimpleMessage{
+			Payload: m,
+		})
+	}
+
+	buf, err := proto.Marshal(&cMsg)
+	if err != nil {
+		return nil
+	}
+
+	gMsg := GossipMessage{
+		Type: MessageTypeCompound,
+		Data: buf,
+	}
+
+	buf, err = proto.Marshal(&gMsg)
+	if err != nil {
+		return nil
+	}
+
+	return buf
+}
+
+// decodeCompoundMessage splits a compound message and returns
+// the slices of individual messages. Returns any potential error.
+func decodeCompoundMessage(buf []byte) ([][]byte, error) {
+	var cMsg CompoundMessage
+	if err := proto.Unmarshal(buf, &cMsg); err != nil {
+		return nil, err
+	}
+
+	parts := make([][]byte, 0, len(cMsg.Messages))
+	for _, m := range cMsg.Messages {
+		parts = append(parts, m.Payload)
+	}
+
+	return parts, nil
+}

vendor/github.com/docker/libnetwork/networkdb/networkdb.go

@@ -0,0 +1,762 @@
+package networkdb
+
+//go:generate protoc -I.:../vendor/github.com/gogo/protobuf --gogo_out=import_path=github.com/docker/libnetwork/networkdb,Mgogoproto/gogo.proto=github.com/gogo/protobuf/gogoproto:. networkdb.proto
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/armon/go-radix"
+	"github.com/docker/docker/pkg/stringid"
+	"github.com/docker/go-events"
+	"github.com/docker/libnetwork/types"
+	"github.com/hashicorp/memberlist"
+	"github.com/hashicorp/serf/serf"
+	"github.com/sirupsen/logrus"
+)
+
+const (
+	byTable int = 1 + iota
+	byNetwork
+)
+
+// NetworkDB instance drives the networkdb cluster and acts the broker
+// for cluster-scoped and network-scoped gossip and watches.
+type NetworkDB struct {
+	// The clocks MUST be the first things
+	// in this struct due to Golang issue #599.
+
+	// Global lamport clock for node network attach events.
+	networkClock serf.LamportClock
+
+	// Global lamport clock for table events.
+	tableClock serf.LamportClock
+
+	sync.RWMutex
+
+	// NetworkDB configuration.
+	config *Config
+
+	// All the tree index (byTable, byNetwork) that we maintain
+	// the db.
+	indexes map[int]*radix.Tree
+
+	// Memberlist we use to drive the cluster.
+	memberlist *memberlist.Memberlist
+
+	// List of all peer nodes in the cluster not-limited to any
+	// network.
+	nodes map[string]*node
+
+	// List of all peer nodes which have failed
+	failedNodes map[string]*node
+
+	// List of all peer nodes which have left
+	leftNodes map[string]*node
+
+	// A multi-dimensional map of network/node attachmemts. The
+	// first key is a node name and the second key is a network ID
+	// for the network that node is participating in.
+	networks map[string]map[string]*network
+
+	// A map of nodes which are participating in a given
+	// network. The key is a network ID.
+	networkNodes map[string][]string
+
+	// A table of ack channels for every node from which we are
+	// waiting for an ack.
+	bulkSyncAckTbl map[string]chan struct{}
+
+	// Broadcast queue for network event gossip.
+	networkBroadcasts *memberlist.TransmitLimitedQueue
+
+	// Broadcast queue for node event gossip.
+	nodeBroadcasts *memberlist.TransmitLimitedQueue
+
+	// A central context to stop all go routines running on
+	// behalf of the NetworkDB instance.
+	ctx       context.Context
+	cancelCtx context.CancelFunc
+
+	// A central broadcaster for all local watchers watching table
+	// events.
+	broadcaster *events.Broadcaster
+
+	// List of all tickers which needed to be stopped when
+	// cleaning up.
+	tickers []*time.Ticker
+
+	// Reference to the memberlist's keyring to add & remove keys
+	keyring *memberlist.Keyring
+
+	// bootStrapIP is the list of IPs that can be used to bootstrap
+	// the gossip.
+	bootStrapIP []net.IP
+
+	// lastStatsTimestamp is the last timestamp when the stats got printed
+	lastStatsTimestamp time.Time
+
+	// lastHealthTimestamp is the last timestamp when the health score got printed
+	lastHealthTimestamp time.Time
+}
+
+// PeerInfo represents the peer (gossip cluster) nodes of a network
+type PeerInfo struct {
+	Name string
+	IP   string
+}
+
+// PeerClusterInfo represents the peer (gossip cluster) nodes
+type PeerClusterInfo struct {
+	PeerInfo
+}
+
+type node struct {
+	memberlist.Node
+	ltime serf.LamportTime
+	// Number of hours left before the reaper removes the node
+	reapTime time.Duration
+}
+
+// network describes the node/network attachment.
+type network struct {
+	// Network ID
+	id string
+
+	// Lamport time for the latest state of the entry.
+	ltime serf.LamportTime
+
+	// Node leave is in progress.
+	leaving bool
+
+	// Number of seconds still left before a deleted network entry gets
+	// removed from networkDB
+	reapTime time.Duration
+
+	// The broadcast queue for table event gossip. This is only
+	// initialized for this node's network attachment entries.
+	tableBroadcasts *memberlist.TransmitLimitedQueue
+
+	// Number of gossip messages sent related to this network during the last stats collection period
+	qMessagesSent int
+
+	// Number of entries on the network. This value is the sum of all the entries of all the tables of a specific network.
+	// Its use is for statistics purposes. It keep tracks of database size and is printed per network every StatsPrintPeriod
+	// interval
+	entriesNumber int
+}
+
+// Config represents the configuration of the networdb instance and
+// can be passed by the caller.
+type Config struct {
+	// NodeID is the node unique identifier of the node when is part of the cluster
+	NodeID string
+
+	// Hostname is the node hostname.
+	Hostname string
+
+	// BindAddr is the IP on which networkdb listens. It can be
+	// 0.0.0.0 to listen on all addresses on the host.
+	BindAddr string
+
+	// AdvertiseAddr is the node's IP address that we advertise for
+	// cluster communication.
+	AdvertiseAddr string
+
+	// BindPort is the local node's port to which we bind to for
+	// cluster communication.
+	BindPort int
+
+	// Keys to be added to the Keyring of the memberlist. Key at index
+	// 0 is the primary key
+	Keys [][]byte
+
+	// PacketBufferSize is the maximum number of bytes that memberlist will
+	// put in a packet (this will be for UDP packets by default with a NetTransport).
+	// A safe value for this is typically 1400 bytes (which is the default). However,
+	// depending on your network's MTU (Maximum Transmission Unit) you may
+	// be able to increase this to get more content into each gossip packet.
+	PacketBufferSize int
+
+	// reapEntryInterval duration of a deleted entry before being garbage collected
+	reapEntryInterval time.Duration
+
+	// reapNetworkInterval duration of a delted network before being garbage collected
+	// NOTE this MUST always be higher than reapEntryInterval
+	reapNetworkInterval time.Duration
+
+	// StatsPrintPeriod the period to use to print queue stats
+	// Default is 5min
+	StatsPrintPeriod time.Duration
+
+	// HealthPrintPeriod the period to use to print the health score
+	// Default is 1min
+	HealthPrintPeriod time.Duration
+}
+
+// entry defines a table entry
+type entry struct {
+	// node from which this entry was learned.
+	node string
+
+	// Lamport time for the most recent update to the entry
+	ltime serf.LamportTime
+
+	// Opaque value store in the entry
+	value []byte
+
+	// Deleting the entry is in progress. All entries linger in
+	// the cluster for certain amount of time after deletion.
+	deleting bool
+
+	// Number of seconds still left before a deleted table entry gets
+	// removed from networkDB
+	reapTime time.Duration
+}
+
+// DefaultConfig returns a NetworkDB config with default values
+func DefaultConfig() *Config {
+	hostname, _ := os.Hostname()
+	return &Config{
+		NodeID:            stringid.TruncateID(stringid.GenerateRandomID()),
+		Hostname:          hostname,
+		BindAddr:          "0.0.0.0",
+		PacketBufferSize:  1400,
+		StatsPrintPeriod:  5 * time.Minute,
+		HealthPrintPeriod: 1 * time.Minute,
+		reapEntryInterval: 30 * time.Minute,
+	}
+}
+
+// New creates a new instance of NetworkDB using the Config passed by
+// the caller.
+func New(c *Config) (*NetworkDB, error) {
+	// The garbage collection logic for entries leverage the presence of the network.
+	// For this reason the expiration time of the network is put slightly higher than the entry expiration so that
+	// there is at least 5 extra cycle to make sure that all the entries are properly deleted before deleting the network.
+	c.reapNetworkInterval = c.reapEntryInterval + 5*reapPeriod
+
+	nDB := &NetworkDB{
+		config:         c,
+		indexes:        make(map[int]*radix.Tree),
+		networks:       make(map[string]map[string]*network),
+		nodes:          make(map[string]*node),
+		failedNodes:    make(map[string]*node),
+		leftNodes:      make(map[string]*node),
+		networkNodes:   make(map[string][]string),
+		bulkSyncAckTbl: make(map[string]chan struct{}),
+		broadcaster:    events.NewBroadcaster(),
+	}
+
+	nDB.indexes[byTable] = radix.New()
+	nDB.indexes[byNetwork] = radix.New()
+
+	logrus.Infof("New memberlist node - Node:%v will use memberlist nodeID:%v with config:%+v", c.Hostname, c.NodeID, c)
+	if err := nDB.clusterInit(); err != nil {
+		return nil, err
+	}
+
+	return nDB, nil
+}
+
+// Join joins this NetworkDB instance with a list of peer NetworkDB
+// instances passed by the caller in the form of addr:port
+func (nDB *NetworkDB) Join(members []string) error {
+	nDB.Lock()
+	nDB.bootStrapIP = make([]net.IP, 0, len(members))
+	for _, m := range members {
+		nDB.bootStrapIP = append(nDB.bootStrapIP, net.ParseIP(m))
+	}
+	nDB.Unlock()
+	return nDB.clusterJoin(members)
+}
+
+// Close destroys this NetworkDB instance by leave the cluster,
+// stopping timers, canceling goroutines etc.
+func (nDB *NetworkDB) Close() {
+	if err := nDB.clusterLeave(); err != nil {
+		logrus.Errorf("%v(%v) Could not close DB: %v", nDB.config.Hostname, nDB.config.NodeID, err)
+	}
+
+	//Avoid (*Broadcaster).run goroutine leak
+	nDB.broadcaster.Close()
+}
+
+// ClusterPeers returns all the gossip cluster peers.
+func (nDB *NetworkDB) ClusterPeers() []PeerInfo {
+	nDB.RLock()
+	defer nDB.RUnlock()
+	peers := make([]PeerInfo, 0, len(nDB.nodes))
+	for _, node := range nDB.nodes {
+		peers = append(peers, PeerInfo{
+			Name: node.Name,
+			IP:   node.Node.Addr.String(),
+		})
+	}
+	return peers
+}
+
+// Peers returns the gossip peers for a given network.
+func (nDB *NetworkDB) Peers(nid string) []PeerInfo {
+	nDB.RLock()
+	defer nDB.RUnlock()
+	peers := make([]PeerInfo, 0, len(nDB.networkNodes[nid]))
+	for _, nodeName := range nDB.networkNodes[nid] {
+		if node, ok := nDB.nodes[nodeName]; ok {
+			peers = append(peers, PeerInfo{
+				Name: node.Name,
+				IP:   node.Addr.String(),
+			})
+		} else {
+			// Added for testing purposes, this condition should never happen else mean that the network list
+			// is out of sync with the node list
+			peers = append(peers, PeerInfo{Name: nodeName, IP: "unknown"})
+		}
+	}
+	return peers
+}
+
+// GetEntry retrieves the value of a table entry in a given (network,
+// table, key) tuple
+func (nDB *NetworkDB) GetEntry(tname, nid, key string) ([]byte, error) {
+	nDB.RLock()
+	defer nDB.RUnlock()
+	entry, err := nDB.getEntry(tname, nid, key)
+	if err != nil {
+		return nil, err
+	}
+	if entry != nil && entry.deleting {
+		return nil, types.NotFoundErrorf("entry in table %s network id %s and key %s deleted and pending garbage collection", tname, nid, key)
+	}
+
+	return entry.value, nil
+}
+
+func (nDB *NetworkDB) getEntry(tname, nid, key string) (*entry, error) {
+	e, ok := nDB.indexes[byTable].Get(fmt.Sprintf("/%s/%s/%s", tname, nid, key))
+	if !ok {
+		return nil, types.NotFoundErrorf("could not get entry in table %s with network id %s and key %s", tname, nid, key)
+	}
+
+	return e.(*entry), nil
+}
+
+// CreateEntry creates a table entry in NetworkDB for given (network,
+// table, key) tuple and if the NetworkDB is part of the cluster
+// propagates this event to the cluster. It is an error to create an
+// entry for the same tuple for which there is already an existing
+// entry unless the current entry is deleting state.
+func (nDB *NetworkDB) CreateEntry(tname, nid, key string, value []byte) error {
+	nDB.Lock()
+	oldEntry, err := nDB.getEntry(tname, nid, key)
+	if err == nil || (oldEntry != nil && !oldEntry.deleting) {
+		nDB.Unlock()
+		return fmt.Errorf("cannot create entry in table %s with network id %s and key %s, already exists", tname, nid, key)
+	}
+
+	entry := &entry{
+		ltime: nDB.tableClock.Increment(),
+		node:  nDB.config.NodeID,
+		value: value,
+	}
+
+	nDB.createOrUpdateEntry(nid, tname, key, entry)
+	nDB.Unlock()
+
+	if err := nDB.sendTableEvent(TableEventTypeCreate, nid, tname, key, entry); err != nil {
+		return fmt.Errorf("cannot send create event for table %s, %v", tname, err)
+	}
+
+	return nil
+}
+
+// UpdateEntry updates a table entry in NetworkDB for given (network,
+// table, key) tuple and if the NetworkDB is part of the cluster
+// propagates this event to the cluster. It is an error to update a
+// non-existent entry.
+func (nDB *NetworkDB) UpdateEntry(tname, nid, key string, value []byte) error {
+	nDB.Lock()
+	if _, err := nDB.getEntry(tname, nid, key); err != nil {
+		nDB.Unlock()
+		return fmt.Errorf("cannot update entry as the entry in table %s with network id %s and key %s does not exist", tname, nid, key)
+	}
+
+	entry := &entry{
+		ltime: nDB.tableClock.Increment(),
+		node:  nDB.config.NodeID,
+		value: value,
+	}
+
+	nDB.createOrUpdateEntry(nid, tname, key, entry)
+	nDB.Unlock()
+
+	if err := nDB.sendTableEvent(TableEventTypeUpdate, nid, tname, key, entry); err != nil {
+		return fmt.Errorf("cannot send table update event: %v", err)
+	}
+
+	return nil
+}
+
+// TableElem elem
+type TableElem struct {
+	Value []byte
+	owner string
+}
+
+// GetTableByNetwork walks the networkdb by the give table and network id and
+// returns a map of keys and values
+func (nDB *NetworkDB) GetTableByNetwork(tname, nid string) map[string]*TableElem {
+	entries := make(map[string]*TableElem)
+	nDB.indexes[byTable].WalkPrefix(fmt.Sprintf("/%s/%s", tname, nid), func(k string, v interface{}) bool {
+		entry := v.(*entry)
+		if entry.deleting {
+			return false
+		}
+		key := k[strings.LastIndex(k, "/")+1:]
+		entries[key] = &TableElem{Value: entry.value, owner: entry.node}
+		return false
+	})
+	return entries
+}
+
+// DeleteEntry deletes a table entry in NetworkDB for given (network,
+// table, key) tuple and if the NetworkDB is part of the cluster
+// propagates this event to the cluster.
+func (nDB *NetworkDB) DeleteEntry(tname, nid, key string) error {
+	nDB.Lock()
+	oldEntry, err := nDB.getEntry(tname, nid, key)
+	if err != nil || oldEntry == nil || oldEntry.deleting {
+		nDB.Unlock()
+		return fmt.Errorf("cannot delete entry %s with network id %s and key %s "+
+			"does not exist or is already being deleted", tname, nid, key)
+	}
+
+	entry := &entry{
+		ltime:    nDB.tableClock.Increment(),
+		node:     nDB.config.NodeID,
+		value:    oldEntry.value,
+		deleting: true,
+		reapTime: nDB.config.reapEntryInterval,
+	}
+
+	nDB.createOrUpdateEntry(nid, tname, key, entry)
+	nDB.Unlock()
+
+	if err := nDB.sendTableEvent(TableEventTypeDelete, nid, tname, key, entry); err != nil {
+		return fmt.Errorf("cannot send table delete event: %v", err)
+	}
+
+	return nil
+}
+
+func (nDB *NetworkDB) deleteNodeFromNetworks(deletedNode string) {
+	for nid, nodes := range nDB.networkNodes {
+		updatedNodes := make([]string, 0, len(nodes))
+		for _, node := range nodes {
+			if node == deletedNode {
+				continue
+			}
+
+			updatedNodes = append(updatedNodes, node)
+		}
+
+		nDB.networkNodes[nid] = updatedNodes
+	}
+
+	delete(nDB.networks, deletedNode)
+}
+
+// deleteNodeNetworkEntries is called in 2 conditions with 2 different outcomes:
+// 1) when a notification is coming of a node leaving the network
+//		- Walk all the network entries and mark the leaving node's entries for deletion
+//			These will be garbage collected when the reap timer will expire
+// 2) when the local node is leaving the network
+//		- Walk all the network entries:
+//			A) if the entry is owned by the local node
+//		  then we will mark it for deletion. This will ensure that if a node did not
+//		  yet received the notification that the local node is leaving, will be aware
+//		  of the entries to be deleted.
+//			B) if the entry is owned by a remote node, then we can safely delete it. This
+//			ensures that if we join back this network as we receive the CREATE event for
+//		  entries owned by remote nodes, we will accept them and we notify the application
+func (nDB *NetworkDB) deleteNodeNetworkEntries(nid, node string) {
+	// Indicates if the delete is triggered for the local node
+	isNodeLocal := node == nDB.config.NodeID
+
+	nDB.indexes[byNetwork].WalkPrefix(fmt.Sprintf("/%s", nid),
+		func(path string, v interface{}) bool {
+			oldEntry := v.(*entry)
+			params := strings.Split(path[1:], "/")
+			nid := params[0]
+			tname := params[1]
+			key := params[2]
+
+			// If the entry is owned by a remote node and this node is not leaving the network
+			if oldEntry.node != node && !isNodeLocal {
+				// Don't do anything because the event is triggered for a node that does not own this entry
+				return false
+			}
+
+			// If this entry is already marked for deletion and this node is not leaving the network
+			if oldEntry.deleting && !isNodeLocal {
+				// Don't do anything this entry will be already garbage collected using the old reapTime
+				return false
+			}
+
+			entry := &entry{
+				ltime:    oldEntry.ltime,
+				node:     oldEntry.node,
+				value:    oldEntry.value,
+				deleting: true,
+				reapTime: nDB.config.reapEntryInterval,
+			}
+
+			// we arrived at this point in 2 cases:
+			// 1) this entry is owned by the node that is leaving the network
+			// 2) the local node is leaving the network
+			if oldEntry.node == node {
+				if isNodeLocal {
+					// TODO fcrisciani: this can be removed if there is no way to leave the network
+					// without doing a delete of all the objects
+					entry.ltime++
+				}
+
+				if !oldEntry.deleting {
+					nDB.createOrUpdateEntry(nid, tname, key, entry)
+				}
+			} else {
+				// the local node is leaving the network, all the entries of remote nodes can be safely removed
+				nDB.deleteEntry(nid, tname, key)
+			}
+
+			// Notify to the upper layer only entries not already marked for deletion
+			if !oldEntry.deleting {
+				nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, entry.value))
+			}
+			return false
+		})
+}
+
+func (nDB *NetworkDB) deleteNodeTableEntries(node string) {
+	nDB.indexes[byTable].Walk(func(path string, v interface{}) bool {
+		oldEntry := v.(*entry)
+		if oldEntry.node != node {
+			return false
+		}
+
+		params := strings.Split(path[1:], "/")
+		tname := params[0]
+		nid := params[1]
+		key := params[2]
+
+		nDB.deleteEntry(nid, tname, key)
+
+		if !oldEntry.deleting {
+			nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, oldEntry.value))
+		}
+		return false
+	})
+}
+
+// WalkTable walks a single table in NetworkDB and invokes the passed
+// function for each entry in the table passing the network, key,
+// value. The walk stops if the passed function returns a true.
+func (nDB *NetworkDB) WalkTable(tname string, fn func(string, string, []byte, bool) bool) error {
+	nDB.RLock()
+	values := make(map[string]interface{})
+	nDB.indexes[byTable].WalkPrefix(fmt.Sprintf("/%s", tname), func(path string, v interface{}) bool {
+		values[path] = v
+		return false
+	})
+	nDB.RUnlock()
+
+	for k, v := range values {
+		params := strings.Split(k[1:], "/")
+		nid := params[1]
+		key := params[2]
+		if fn(nid, key, v.(*entry).value, v.(*entry).deleting) {
+			return nil
+		}
+	}
+
+	return nil
+}
+
+// JoinNetwork joins this node to a given network and propagates this
+// event across the cluster. This triggers this node joining the
+// sub-cluster of this network and participates in the network-scoped
+// gossip and bulk sync for this network.
+func (nDB *NetworkDB) JoinNetwork(nid string) error {
+	ltime := nDB.networkClock.Increment()
+
+	nDB.Lock()
+	nodeNetworks, ok := nDB.networks[nDB.config.NodeID]
+	if !ok {
+		nodeNetworks = make(map[string]*network)
+		nDB.networks[nDB.config.NodeID] = nodeNetworks
+	}
+	n, ok := nodeNetworks[nid]
+	var entries int
+	if ok {
+		entries = n.entriesNumber
+	}
+	nodeNetworks[nid] = &network{id: nid, ltime: ltime, entriesNumber: entries}
+	nodeNetworks[nid].tableBroadcasts = &memberlist.TransmitLimitedQueue{
+		NumNodes: func() int {
+			//TODO fcrisciani this can be optimized maybe avoiding the lock?
+			// this call is done each GetBroadcasts call to evaluate the number of
+			// replicas for the message
+			nDB.RLock()
+			defer nDB.RUnlock()
+			return len(nDB.networkNodes[nid])
+		},
+		RetransmitMult: 4,
+	}
+	nDB.addNetworkNode(nid, nDB.config.NodeID)
+	networkNodes := nDB.networkNodes[nid]
+	nDB.Unlock()
+
+	if err := nDB.sendNetworkEvent(nid, NetworkEventTypeJoin, ltime); err != nil {
+		return fmt.Errorf("failed to send leave network event for %s: %v", nid, err)
+	}
+
+	logrus.Debugf("%v(%v): joined network %s", nDB.config.Hostname, nDB.config.NodeID, nid)
+	if _, err := nDB.bulkSync(networkNodes, true); err != nil {
+		logrus.Errorf("Error bulk syncing while joining network %s: %v", nid, err)
+	}
+
+	return nil
+}
+
+// LeaveNetwork leaves this node from a given network and propagates
+// this event across the cluster. This triggers this node leaving the
+// sub-cluster of this network and as a result will no longer
+// participate in the network-scoped gossip and bulk sync for this
+// network. Also remove all the table entries for this network from
+// networkdb
+func (nDB *NetworkDB) LeaveNetwork(nid string) error {
+	ltime := nDB.networkClock.Increment()
+	if err := nDB.sendNetworkEvent(nid, NetworkEventTypeLeave, ltime); err != nil {
+		return fmt.Errorf("failed to send leave network event for %s: %v", nid, err)
+	}
+
+	nDB.Lock()
+	defer nDB.Unlock()
+
+	// Remove myself from the list of the nodes participating to the network
+	nDB.deleteNetworkNode(nid, nDB.config.NodeID)
+
+	// Update all the local entries marking them for deletion and delete all the remote entries
+	nDB.deleteNodeNetworkEntries(nid, nDB.config.NodeID)
+
+	nodeNetworks, ok := nDB.networks[nDB.config.NodeID]
+	if !ok {
+		return fmt.Errorf("could not find self node for network %s while trying to leave", nid)
+	}
+
+	n, ok := nodeNetworks[nid]
+	if !ok {
+		return fmt.Errorf("could not find network %s while trying to leave", nid)
+	}
+
+	logrus.Debugf("%v(%v): leaving network %s", nDB.config.Hostname, nDB.config.NodeID, nid)
+	n.ltime = ltime
+	n.reapTime = nDB.config.reapNetworkInterval
+	n.leaving = true
+	return nil
+}
+
+// addNetworkNode adds the node to the list of nodes which participate
+// in the passed network only if it is not already present. Caller
+// should hold the NetworkDB lock while calling this
+func (nDB *NetworkDB) addNetworkNode(nid string, nodeName string) {
+	nodes := nDB.networkNodes[nid]
+	for _, node := range nodes {
+		if node == nodeName {
+			return
+		}
+	}
+
+	nDB.networkNodes[nid] = append(nDB.networkNodes[nid], nodeName)
+}
+
+// Deletes the node from the list of nodes which participate in the
+// passed network. Caller should hold the NetworkDB lock while calling
+// this
+func (nDB *NetworkDB) deleteNetworkNode(nid string, nodeName string) {
+	nodes, ok := nDB.networkNodes[nid]
+	if !ok || len(nodes) == 0 {
+		return
+	}
+	newNodes := make([]string, 0, len(nodes)-1)
+	for _, name := range nodes {
+		if name == nodeName {
+			continue
+		}
+		newNodes = append(newNodes, name)
+	}
+	nDB.networkNodes[nid] = newNodes
+}
+
+// findCommonnetworks find the networks that both this node and the
+// passed node have joined.
+func (nDB *NetworkDB) findCommonNetworks(nodeName string) []string {
+	nDB.RLock()
+	defer nDB.RUnlock()
+
+	var networks []string
+	for nid := range nDB.networks[nDB.config.NodeID] {
+		if n, ok := nDB.networks[nodeName][nid]; ok {
+			if !n.leaving {
+				networks = append(networks, nid)
+			}
+		}
+	}
+
+	return networks
+}
+
+func (nDB *NetworkDB) updateLocalNetworkTime() {
+	nDB.Lock()
+	defer nDB.Unlock()
+
+	ltime := nDB.networkClock.Increment()
+	for _, n := range nDB.networks[nDB.config.NodeID] {
+		n.ltime = ltime
+	}
+}
+
+// createOrUpdateEntry this function handles the creation or update of entries into the local
+// tree store. It is also used to keep in sync the entries number of the network (all tables are aggregated)
+func (nDB *NetworkDB) createOrUpdateEntry(nid, tname, key string, entry interface{}) (bool, bool) {
+	_, okTable := nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)
+	_, okNetwork := nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)
+	if !okNetwork {
+		// Add only if it is an insert not an update
+		n, ok := nDB.networks[nDB.config.NodeID][nid]
+		if ok {
+			n.entriesNumber++
+		}
+	}
+	return okTable, okNetwork
+}
+
+// deleteEntry this function handles the deletion of entries into the local tree store.
+// It is also used to keep in sync the entries number of the network (all tables are aggregated)
+func (nDB *NetworkDB) deleteEntry(nid, tname, key string) (bool, bool) {
+	_, okTable := nDB.indexes[byTable].Delete(fmt.Sprintf("/%s/%s/%s", tname, nid, key))
+	_, okNetwork := nDB.indexes[byNetwork].Delete(fmt.Sprintf("/%s/%s/%s", nid, tname, key))
+	if okNetwork {
+		// Remove only if the delete is successful
+		n, ok := nDB.networks[nDB.config.NodeID][nid]
+		if ok {
+			n.entriesNumber--
+		}
+	}
+	return okTable, okNetwork
+}

vendor/github.com/docker/libnetwork/networkdb/networkdbdiagnostic.go

@@ -0,0 +1,413 @@
+package networkdb
+
+import (
+	"encoding/base64"
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/docker/libnetwork/common"
+	"github.com/docker/libnetwork/diagnostic"
+	"github.com/sirupsen/logrus"
+)
+
+const (
+	missingParameter = "missing parameter"
+	dbNotAvailable   = "database not available"
+)
+
+// NetDbPaths2Func TODO
+var NetDbPaths2Func = map[string]diagnostic.HTTPHandlerFunc{
+	"/join":         dbJoin,
+	"/networkpeers": dbPeers,
+	"/clusterpeers": dbClusterPeers,
+	"/joinnetwork":  dbJoinNetwork,
+	"/leavenetwork": dbLeaveNetwork,
+	"/createentry":  dbCreateEntry,
+	"/updateentry":  dbUpdateEntry,
+	"/deleteentry":  dbDeleteEntry,
+	"/getentry":     dbGetEntry,
+	"/gettable":     dbGetTable,
+}
+
+func dbJoin(ctx interface{}, w http.ResponseWriter, r *http.Request) {
+	r.ParseForm()
+	diagnostic.DebugHTTPForm(r)
+	_, json := diagnostic.ParseHTTPFormOptions(r)
+
+	// audit logs
+	log := logrus.WithFields(logrus.Fields{"component": "diagnostic", "remoteIP": r.RemoteAddr, "method": common.CallerName(0), "url": r.URL.String()})
+	log.Info("join cluster")
+
+	if len(r.Form["members"]) < 1 {
+		rsp := diagnostic.WrongCommand(missingParameter, fmt.Sprintf("%s?members=ip1,ip2,...", r.URL.Path))
+		log.Error("join cluster failed, wrong input")
+		diagnostic.HTTPReply(w, rsp, json)
+		return
+	}
+
+	nDB, ok := ctx.(*NetworkDB)
+	if ok {
+		err := nDB.Join(strings.Split(r.Form["members"][0], ","))
+		if err != nil {
+			rsp := diagnostic.FailCommand(fmt.Errorf("%s error in the DB join %s", r.URL.Path, err))
+			log.WithError(err).Error("join cluster failed")
+			diagnostic.HTTPReply(w, rsp, json)
+			return
+		}
+
+		log.Info("join cluster done")
+		diagnostic.HTTPReply(w, diagnostic.CommandSucceed(nil), json)
+		return
+	}
+	diagnostic.HTTPReply(w, diagnostic.FailCommand(fmt.Errorf("%s", dbNotAvailable)), json)
+}
+
+func dbPeers(ctx interface{}, w http.ResponseWriter, r *http.Request) {
+	r.ParseForm()
+	diagnostic.DebugHTTPForm(r)
+	_, json := diagnostic.ParseHTTPFormOptions(r)
+
+	// audit logs
+	log := logrus.WithFields(logrus.Fields{"component": "diagnostic", "remoteIP": r.RemoteAddr, "method": common.CallerName(0), "url": r.URL.String()})
+	log.Info("network peers")
+
+	if len(r.Form["nid"]) < 1 {
+		rsp := diagnostic.WrongCommand(missingParameter, fmt.Sprintf("%s?nid=test", r.URL.Path))
+		log.Error("network peers failed, wrong input")
+		diagnostic.HTTPReply(w, rsp, json)
+		return
+	}
+
+	nDB, ok := ctx.(*NetworkDB)
+	if ok {
+		peers := nDB.Peers(r.Form["nid"][0])
+		rsp := &diagnostic.TableObj{Length: len(peers)}
+		for i, peerInfo := range peers {
+			if peerInfo.IP == "unknown" {
+				rsp.Elements = append(rsp.Elements, &diagnostic.PeerEntryObj{Index: i, Name: "orphan-" + peerInfo.Name, IP: peerInfo.IP})
+			} else {
+				rsp.Elements = append(rsp.Elements, &diagnostic.PeerEntryObj{Index: i, Name: peerInfo.Name, IP: peerInfo.IP})
+			}
+		}
+		log.WithField("response", fmt.Sprintf("%+v", rsp)).Info("network peers done")
+		diagnostic.HTTPReply(w, diagnostic.CommandSucceed(rsp), json)
+		return
+	}
+	diagnostic.HTTPReply(w, diagnostic.FailCommand(fmt.Errorf("%s", dbNotAvailable)), json)
+}
+
+func dbClusterPeers(ctx interface{}, w http.ResponseWriter, r *http.Request) {
+	r.ParseForm()
+	diagnostic.DebugHTTPForm(r)
+	_, json := diagnostic.ParseHTTPFormOptions(r)
+
+	// audit logs
+	log := logrus.WithFields(logrus.Fields{"component": "diagnostic", "remoteIP": r.RemoteAddr, "method": common.CallerName(0), "url": r.URL.String()})
+	log.Info("cluster peers")
+
+	nDB, ok := ctx.(*NetworkDB)
+	if ok {
+		peers := nDB.ClusterPeers()
+		rsp := &diagnostic.TableObj{Length: len(peers)}
+		for i, peerInfo := range peers {
+			rsp.Elements = append(rsp.Elements, &diagnostic.PeerEntryObj{Index: i, Name: peerInfo.Name, IP: peerInfo.IP})
+		}
+		log.WithField("response", fmt.Sprintf("%+v", rsp)).Info("cluster peers done")
+		diagnostic.HTTPReply(w, diagnostic.CommandSucceed(rsp), json)
+		return
+	}
+	diagnostic.HTTPReply(w, diagnostic.FailCommand(fmt.Errorf("%s", dbNotAvailable)), json)
+}
+
+func dbCreateEntry(ctx interface{}, w http.ResponseWriter, r *http.Request) {
+	r.ParseForm()
+	diagnostic.DebugHTTPForm(r)
+	unsafe, json := diagnostic.ParseHTTPFormOptions(r)
+
+	// audit logs
+	log := logrus.WithFields(logrus.Fields{"component": "diagnostic", "remoteIP": r.RemoteAddr, "method": common.CallerName(0), "url": r.URL.String()})
+	log.Info("create entry")
+
+	if len(r.Form["tname"]) < 1 ||
+		len(r.Form["nid"]) < 1 ||
+		len(r.Form["key"]) < 1 ||
+		len(r.Form["value"]) < 1 {
+		rsp := diagnostic.WrongCommand(missingParameter, fmt.Sprintf("%s?tname=table_name&nid=network_id&key=k&value=v", r.URL.Path))
+		log.Error("create entry failed, wrong input")
+		diagnostic.HTTPReply(w, rsp, json)
+		return
+	}
+
+	tname := r.Form["tname"][0]
+	nid := r.Form["nid"][0]
+	key := r.Form["key"][0]
+	value := r.Form["value"][0]
+	decodedValue := []byte(value)
+	if !unsafe {
+		var err error
+		decodedValue, err = base64.StdEncoding.DecodeString(value)
+		if err != nil {
+			log.WithError(err).Error("create entry failed")
+			diagnostic.HTTPReply(w, diagnostic.FailCommand(err), json)
+			return
+		}
+	}
+
+	nDB, ok := ctx.(*NetworkDB)
+	if ok {
+		if err := nDB.CreateEntry(tname, nid, key, decodedValue); err != nil {
+			rsp := diagnostic.FailCommand(err)
+			diagnostic.HTTPReply(w, rsp, json)
+			log.WithError(err).Error("create entry failed")
+			return
+		}
+		log.Info("create entry done")
+		diagnostic.HTTPReply(w, diagnostic.CommandSucceed(nil), json)
+		return
+	}
+	diagnostic.HTTPReply(w, diagnostic.FailCommand(fmt.Errorf("%s", dbNotAvailable)), json)
+}
+
+func dbUpdateEntry(ctx interface{}, w http.ResponseWriter, r *http.Request) {
+	r.ParseForm()
+	diagnostic.DebugHTTPForm(r)
+	unsafe, json := diagnostic.ParseHTTPFormOptions(r)
+
+	// audit logs
+	log := logrus.WithFields(logrus.Fields{"component": "diagnostic", "remoteIP": r.RemoteAddr, "method": common.CallerName(0), "url": r.URL.String()})
+	log.Info("update entry")
+
+	if len(r.Form["tname"]) < 1 ||
+		len(r.Form["nid"]) < 1 ||
+		len(r.Form["key"]) < 1 ||
+		len(r.Form["value"]) < 1 {
+		rsp := diagnostic.WrongCommand(missingParameter, fmt.Sprintf("%s?tname=table_name&nid=network_id&key=k&value=v", r.URL.Path))
+		log.Error("update entry failed, wrong input")
+		diagnostic.HTTPReply(w, rsp, json)
+		return
+	}
+
+	tname := r.Form["tname"][0]
+	nid := r.Form["nid"][0]
+	key := r.Form["key"][0]
+	value := r.Form["value"][0]
+	decodedValue := []byte(value)
+	if !unsafe {
+		var err error
+		decodedValue, err = base64.StdEncoding.DecodeString(value)
+		if err != nil {
+			log.WithError(err).Error("update entry failed")
+			diagnostic.HTTPReply(w, diagnostic.FailCommand(err), json)
+			return
+		}
+	}
+
+	nDB, ok := ctx.(*NetworkDB)
+	if ok {
+		if err := nDB.UpdateEntry(tname, nid, key, decodedValue); err != nil {
+			log.WithError(err).Error("update entry failed")
+			diagnostic.HTTPReply(w, diagnostic.FailCommand(err), json)
+			return
+		}
+		log.Info("update entry done")
+		diagnostic.HTTPReply(w, diagnostic.CommandSucceed(nil), json)
+		return
+	}
+	diagnostic.HTTPReply(w, diagnostic.FailCommand(fmt.Errorf("%s", dbNotAvailable)), json)
+}
+
+func dbDeleteEntry(ctx interface{}, w http.ResponseWriter, r *http.Request) {
+	r.ParseForm()
+	diagnostic.DebugHTTPForm(r)
+	_, json := diagnostic.ParseHTTPFormOptions(r)
+
+	// audit logs
+	log := logrus.WithFields(logrus.Fields{"component": "diagnostic", "remoteIP": r.RemoteAddr, "method": common.CallerName(0), "url": r.URL.String()})
+	log.Info("delete entry")
+
+	if len(r.Form["tname"]) < 1 ||
+		len(r.Form["nid"]) < 1 ||
+		len(r.Form["key"]) < 1 {
+		rsp := diagnostic.WrongCommand(missingParameter, fmt.Sprintf("%s?tname=table_name&nid=network_id&key=k", r.URL.Path))
+		log.Error("delete entry failed, wrong input")
+		diagnostic.HTTPReply(w, rsp, json)
+		return
+	}
+
+	tname := r.Form["tname"][0]
+	nid := r.Form["nid"][0]
+	key := r.Form["key"][0]
+
+	nDB, ok := ctx.(*NetworkDB)
+	if ok {
+		err := nDB.DeleteEntry(tname, nid, key)
+		if err != nil {
+			log.WithError(err).Error("delete entry failed")
+			diagnostic.HTTPReply(w, diagnostic.FailCommand(err), json)
+			return
+		}
+		log.Info("delete entry done")
+		diagnostic.HTTPReply(w, diagnostic.CommandSucceed(nil), json)
+		return
+	}
+	diagnostic.HTTPReply(w, diagnostic.FailCommand(fmt.Errorf("%s", dbNotAvailable)), json)
+}
+
+func dbGetEntry(ctx interface{}, w http.ResponseWriter, r *http.Request) {
+	r.ParseForm()
+	diagnostic.DebugHTTPForm(r)
+	unsafe, json := diagnostic.ParseHTTPFormOptions(r)
+
+	// audit logs
+	log := logrus.WithFields(logrus.Fields{"component": "diagnostic", "remoteIP": r.RemoteAddr, "method": common.CallerName(0), "url": r.URL.String()})
+	log.Info("get entry")
+
+	if len(r.Form["tname"]) < 1 ||
+		len(r.Form["nid"]) < 1 ||
+		len(r.Form["key"]) < 1 {
+		rsp := diagnostic.WrongCommand(missingParameter, fmt.Sprintf("%s?tname=table_name&nid=network_id&key=k", r.URL.Path))
+		log.Error("get entry failed, wrong input")
+		diagnostic.HTTPReply(w, rsp, json)
+		return
+	}
+
+	tname := r.Form["tname"][0]
+	nid := r.Form["nid"][0]
+	key := r.Form["key"][0]
+
+	nDB, ok := ctx.(*NetworkDB)
+	if ok {
+		value, err := nDB.GetEntry(tname, nid, key)
+		if err != nil {
+			log.WithError(err).Error("get entry failed")
+			diagnostic.HTTPReply(w, diagnostic.FailCommand(err), json)
+			return
+		}
+
+		var encodedValue string
+		if unsafe {
+			encodedValue = string(value)
+		} else {
+			encodedValue = base64.StdEncoding.EncodeToString(value)
+		}
+
+		rsp := &diagnostic.TableEntryObj{Key: key, Value: encodedValue}
+		log.WithField("response", fmt.Sprintf("%+v", rsp)).Info("get entry done")
+		diagnostic.HTTPReply(w, diagnostic.CommandSucceed(rsp), json)
+		return
+	}
+	diagnostic.HTTPReply(w, diagnostic.FailCommand(fmt.Errorf("%s", dbNotAvailable)), json)
+}
+
+func dbJoinNetwork(ctx interface{}, w http.ResponseWriter, r *http.Request) {
+	r.ParseForm()
+	diagnostic.DebugHTTPForm(r)
+	_, json := diagnostic.ParseHTTPFormOptions(r)
+
+	// audit logs
+	log := logrus.WithFields(logrus.Fields{"component": "diagnostic", "remoteIP": r.RemoteAddr, "method": common.CallerName(0), "url": r.URL.String()})
+	log.Info("join network")
+
+	if len(r.Form["nid"]) < 1 {
+		rsp := diagnostic.WrongCommand(missingParameter, fmt.Sprintf("%s?nid=network_id", r.URL.Path))
+		log.Error("join network failed, wrong input")
+		diagnostic.HTTPReply(w, rsp, json)
+		return
+	}
+
+	nid := r.Form["nid"][0]
+
+	nDB, ok := ctx.(*NetworkDB)
+	if ok {
+		if err := nDB.JoinNetwork(nid); err != nil {
+			log.WithError(err).Error("join network failed")
+			diagnostic.HTTPReply(w, diagnostic.FailCommand(err), json)
+			return
+		}
+		log.Info("join network done")
+		diagnostic.HTTPReply(w, diagnostic.CommandSucceed(nil), json)
+		return
+	}
+	diagnostic.HTTPReply(w, diagnostic.FailCommand(fmt.Errorf("%s", dbNotAvailable)), json)
+}
+
+func dbLeaveNetwork(ctx interface{}, w http.ResponseWriter, r *http.Request) {
+	r.ParseForm()
+	diagnostic.DebugHTTPForm(r)
+	_, json := diagnostic.ParseHTTPFormOptions(r)
+
+	// audit logs
+	log := logrus.WithFields(logrus.Fields{"component": "diagnostic", "remoteIP": r.RemoteAddr, "method": common.CallerName(0), "url": r.URL.String()})
+	log.Info("leave network")
+
+	if len(r.Form["nid"]) < 1 {
+		rsp := diagnostic.WrongCommand(missingParameter, fmt.Sprintf("%s?nid=network_id", r.URL.Path))
+		log.Error("leave network failed, wrong input")
+		diagnostic.HTTPReply(w, rsp, json)
+		return
+	}
+
+	nid := r.Form["nid"][0]
+
+	nDB, ok := ctx.(*NetworkDB)
+	if ok {
+		if err := nDB.LeaveNetwork(nid); err != nil {
+			log.WithError(err).Error("leave network failed")
+			diagnostic.HTTPReply(w, diagnostic.FailCommand(err), json)
+			return
+		}
+		log.Info("leave network done")
+		diagnostic.HTTPReply(w, diagnostic.CommandSucceed(nil), json)
+		return
+	}
+	diagnostic.HTTPReply(w, diagnostic.FailCommand(fmt.Errorf("%s", dbNotAvailable)), json)
+}
+
+func dbGetTable(ctx interface{}, w http.ResponseWriter, r *http.Request) {
+	r.ParseForm()
+	diagnostic.DebugHTTPForm(r)
+	unsafe, json := diagnostic.ParseHTTPFormOptions(r)
+
+	// audit logs
+	log := logrus.WithFields(logrus.Fields{"component": "diagnostic", "remoteIP": r.RemoteAddr, "method": common.CallerName(0), "url": r.URL.String()})
+	log.Info("get table")
+
+	if len(r.Form["tname"]) < 1 ||
+		len(r.Form["nid"]) < 1 {
+		rsp := diagnostic.WrongCommand(missingParameter, fmt.Sprintf("%s?tname=table_name&nid=network_id", r.URL.Path))
+		log.Error("get table failed, wrong input")
+		diagnostic.HTTPReply(w, rsp, json)
+		return
+	}
+
+	tname := r.Form["tname"][0]
+	nid := r.Form["nid"][0]
+
+	nDB, ok := ctx.(*NetworkDB)
+	if ok {
+		table := nDB.GetTableByNetwork(tname, nid)
+		rsp := &diagnostic.TableObj{Length: len(table)}
+		var i = 0
+		for k, v := range table {
+			var encodedValue string
+			if unsafe {
+				encodedValue = string(v.Value)
+			} else {
+				encodedValue = base64.StdEncoding.EncodeToString(v.Value)
+			}
+			rsp.Elements = append(rsp.Elements,
+				&diagnostic.TableEntryObj{
+					Index: i,
+					Key:   k,
+					Value: encodedValue,
+					Owner: v.owner,
+				})
+			i++
+		}
+		log.WithField("response", fmt.Sprintf("%+v", rsp)).Info("get table done")
+		diagnostic.HTTPReply(w, diagnostic.CommandSucceed(rsp), json)
+		return
+	}
+	diagnostic.HTTPReply(w, diagnostic.FailCommand(fmt.Errorf("%s", dbNotAvailable)), json)
+}

vendor/github.com/docker/libnetwork/networkdb/nodemgmt.go

@@ -0,0 +1,120 @@
+package networkdb
+
+import (
+	"fmt"
+
+	"github.com/hashicorp/memberlist"
+	"github.com/sirupsen/logrus"
+)
+
+type nodeState int
+
+const (
+	nodeNotFound    nodeState = -1
+	nodeActiveState nodeState = 0
+	nodeLeftState   nodeState = 1
+	nodeFailedState nodeState = 2
+)
+
+var nodeStateName = map[nodeState]string{
+	-1: "NodeNotFound",
+	0:  "NodeActive",
+	1:  "NodeLeft",
+	2:  "NodeFailed",
+}
+
+// findNode search the node into the 3 node lists and returns the node pointer and the list
+// where it got found
+func (nDB *NetworkDB) findNode(nodeName string) (*node, nodeState, map[string]*node) {
+	for i, nodes := range []map[string]*node{
+		nDB.nodes,
+		nDB.leftNodes,
+		nDB.failedNodes,
+	} {
+		if n, ok := nodes[nodeName]; ok {
+			return n, nodeState(i), nodes
+		}
+	}
+	return nil, nodeNotFound, nil
+}
+
+// changeNodeState changes the state of the node specified, returns true if the node was moved,
+// false if there was no need to change the node state. Error will be returned if the node does not
+// exists
+func (nDB *NetworkDB) changeNodeState(nodeName string, newState nodeState) (bool, error) {
+	n, currState, m := nDB.findNode(nodeName)
+	if n == nil {
+		return false, fmt.Errorf("node %s not found", nodeName)
+	}
+
+	switch newState {
+	case nodeActiveState:
+		if currState == nodeActiveState {
+			return false, nil
+		}
+
+		delete(m, nodeName)
+		// reset the node reap time
+		n.reapTime = 0
+		nDB.nodes[nodeName] = n
+	case nodeLeftState:
+		if currState == nodeLeftState {
+			return false, nil
+		}
+
+		delete(m, nodeName)
+		nDB.leftNodes[nodeName] = n
+	case nodeFailedState:
+		if currState == nodeFailedState {
+			return false, nil
+		}
+
+		delete(m, nodeName)
+		nDB.failedNodes[nodeName] = n
+	}
+
+	logrus.Infof("Node %s change state %s --> %s", nodeName, nodeStateName[currState], nodeStateName[newState])
+
+	if newState == nodeLeftState || newState == nodeFailedState {
+		// set the node reap time, if not already set
+		// It is possible that a node passes from failed to left and the reaptime was already set so keep that value
+		if n.reapTime == 0 {
+			n.reapTime = nodeReapInterval
+		}
+		// The node leave or fails, delete all the entries created by it.
+		// If the node was temporary down, deleting the entries will guarantee that the CREATE events will be accepted
+		// If the node instead left because was going down, then it makes sense to just delete all its state
+		nDB.deleteNodeFromNetworks(n.Name)
+		nDB.deleteNodeTableEntries(n.Name)
+	}
+
+	return true, nil
+}
+
+func (nDB *NetworkDB) purgeReincarnation(mn *memberlist.Node) bool {
+	for name, node := range nDB.nodes {
+		if node.Addr.Equal(mn.Addr) && node.Port == mn.Port && mn.Name != name {
+			logrus.Infof("Node %s/%s, is the new incarnation of the active node %s/%s", mn.Name, mn.Addr, name, node.Addr)
+			nDB.changeNodeState(name, nodeLeftState)
+			return true
+		}
+	}
+
+	for name, node := range nDB.failedNodes {
+		if node.Addr.Equal(mn.Addr) && node.Port == mn.Port && mn.Name != name {
+			logrus.Infof("Node %s/%s, is the new incarnation of the failed node %s/%s", mn.Name, mn.Addr, name, node.Addr)
+			nDB.changeNodeState(name, nodeLeftState)
+			return true
+		}
+	}
+
+	for name, node := range nDB.leftNodes {
+		if node.Addr.Equal(mn.Addr) && node.Port == mn.Port && mn.Name != name {
+			logrus.Infof("Node %s/%s, is the new incarnation of the shutdown node %s/%s", mn.Name, mn.Addr, name, node.Addr)
+			nDB.changeNodeState(name, nodeLeftState)
+			return true
+		}
+	}
+
+	return false
+}

vendor/github.com/docker/libnetwork/networkdb/watch.go

@@ -0,0 +1,110 @@
+package networkdb
+
+import (
+	"net"
+
+	"github.com/docker/go-events"
+)
+
+type opType uint8
+
+const (
+	opCreate opType = 1 + iota
+	opUpdate
+	opDelete
+)
+
+type event struct {
+	Table     string
+	NetworkID string
+	Key       string
+	Value     []byte
+}
+
+// NodeTable represents table event for node join and leave
+const NodeTable = "NodeTable"
+
+// NodeAddr represents the value carried for node event in NodeTable
+type NodeAddr struct {
+	Addr net.IP
+}
+
+// CreateEvent generates a table entry create event to the watchers
+type CreateEvent event
+
+// UpdateEvent generates a table entry update event to the watchers
+type UpdateEvent event
+
+// DeleteEvent generates a table entry delete event to the watchers
+type DeleteEvent event
+
+// Watch creates a watcher with filters for a particular table or
+// network or key or any combination of the tuple. If any of the
+// filter is an empty string it acts as a wildcard for that
+// field. Watch returns a channel of events, where the events will be
+// sent.
+func (nDB *NetworkDB) Watch(tname, nid, key string) (*events.Channel, func()) {
+	var matcher events.Matcher
+
+	if tname != "" || nid != "" || key != "" {
+		matcher = events.MatcherFunc(func(ev events.Event) bool {
+			var evt event
+			switch ev := ev.(type) {
+			case CreateEvent:
+				evt = event(ev)
+			case UpdateEvent:
+				evt = event(ev)
+			case DeleteEvent:
+				evt = event(ev)
+			}
+
+			if tname != "" && evt.Table != tname {
+				return false
+			}
+
+			if nid != "" && evt.NetworkID != nid {
+				return false
+			}
+
+			if key != "" && evt.Key != key {
+				return false
+			}
+
+			return true
+		})
+	}
+
+	ch := events.NewChannel(0)
+	sink := events.Sink(events.NewQueue(ch))
+
+	if matcher != nil {
+		sink = events.NewFilter(sink, matcher)
+	}
+
+	nDB.broadcaster.Add(sink)
+	return ch, func() {
+		nDB.broadcaster.Remove(sink)
+		ch.Close()
+		sink.Close()
+	}
+}
+
+func makeEvent(op opType, tname, nid, key string, value []byte) events.Event {
+	ev := event{
+		Table:     tname,
+		NetworkID: nid,
+		Key:       key,
+		Value:     value,
+	}
+
+	switch op {
+	case opCreate:
+		return CreateEvent(ev)
+	case opUpdate:
+		return UpdateEvent(ev)
+	case opDelete:
+		return DeleteEvent(ev)
+	}
+
+	return nil
+}