#3345

Build # #3345

Build Type

push

Committed by

Commit Message

Merge pull request #1674 from sanimej/inspect

Run Details

154 of 154 new or added lines in 14 files covered. (100.0%)

11544 of 26494 relevant lines covered (43.57%)

33493.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.86

/networkdb/networkdb.go

package networkdb

//go:generate protoc -I.:../Godeps/_workspace/src/github.com/gogo/protobuf  --gogo_out=import_path=github.com/docker/libnetwork/networkdb,Mgogoproto/gogo.proto=github.com/gogo/protobuf/gogoproto:. networkdb.proto

import (
        "fmt"
        "strings"
        "sync"
        "time"

        "github.com/Sirupsen/logrus"
        "github.com/armon/go-radix"
        "github.com/docker/go-events"
        "github.com/docker/libnetwork/types"
        "github.com/hashicorp/memberlist"
        "github.com/hashicorp/serf/serf"
)

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 stop channel to stop all go routines running on
        // behalf of the NetworkDB instance.
        stopCh chan struct{}

        // 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
}

// PeerInfo represents the peer (gossip cluster) nodes of a network
type PeerInfo struct {
        Name string
        IP   string
}

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
}

// Config represents the configuration of the networdb instance and
// can be passed by the caller.
type Config struct {
        // NodeName is the cluster wide unique name for this node.
        NodeName 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
}

// 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
}

// New creates a new instance of NetworkDB using the Config passed by
// the caller.
func New(c *Config) (*NetworkDB, error) {
        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()

        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 {
        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("Could not close DB %s: %v", nDB.config.NodeName, err)
        }
}

// 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(),
                        })
                }
        }
        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) {
        entry, err := nDB.getEntry(tname, nid, key)
        if err != nil {
                return nil, err
        }

        return entry.value, nil
}

func (nDB *NetworkDB) getEntry(tname, nid, key string) (*entry, error) {
        nDB.RLock()
        defer nDB.RUnlock()

        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 {
        oldEntry, err := nDB.getEntry(tname, nid, key)
        if err != nil {
                if _, ok := err.(types.NotFoundError); !ok {
                        return fmt.Errorf("cannot create entry in table %s with network id %s and key %s: %v", tname, nid, key, err)
                }
        }
        if oldEntry != nil && !oldEntry.deleting {
                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.NodeName,
                value: value,
        }

        if err := nDB.sendTableEvent(TableEventTypeCreate, nid, tname, key, entry); err != nil {
                return fmt.Errorf("cannot send create event for table %s, %v", tname, err)
        }

        nDB.Lock()
        nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)
        nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)
        nDB.Unlock()

        nDB.broadcaster.Write(makeEvent(opCreate, tname, nid, key, value))
        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 {
        if _, err := nDB.GetEntry(tname, nid, key); err != nil {
                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.NodeName,
                value: value,
        }

        if err := nDB.sendTableEvent(TableEventTypeUpdate, nid, tname, key, entry); err != nil {
                return fmt.Errorf("cannot send table update event: %v", err)
        }

        nDB.Lock()
        nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)
        nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)
        nDB.Unlock()

        nDB.broadcaster.Write(makeEvent(opUpdate, tname, nid, key, value))
        return nil
}

// 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]interface{} {
        entries := make(map[string]interface{})
        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] = entry.value
                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 {
        value, err := nDB.GetEntry(tname, nid, key)
        if err != nil {
                return fmt.Errorf("cannot delete 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.NodeName,
                value:    value,
                deleting: true,
                reapTime: reapInterval,
        }

        if err := nDB.sendTableEvent(TableEventTypeDelete, nid, tname, key, entry); err != nil {
                return fmt.Errorf("cannot send table delete event: %v", err)
        }

        nDB.Lock()
        nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)
        nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)
        nDB.Unlock()

        nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, value))
        return nil
}

func (nDB *NetworkDB) deleteNetworkEntriesForNode(deletedNode string) {
        nDB.Lock()
        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)
        nDB.Unlock()
}

func (nDB *NetworkDB) deleteNodeTableEntries(node string) {
        nDB.Lock()
        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]

                entry := &entry{
                        ltime:    oldEntry.ltime,
                        node:     node,
                        value:    oldEntry.value,
                        deleting: true,
                        reapTime: reapInterval,
                }

                nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)
                nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)

                nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, entry.value))
                return false
        })
        nDB.Unlock()
}

// 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) 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) {
                        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.NodeName]
        if !ok {
                nodeNetworks = make(map[string]*network)
                nDB.networks[nDB.config.NodeName] = nodeNetworks
        }
        nodeNetworks[nid] = &network{id: nid, ltime: ltime}
        nodeNetworks[nid].tableBroadcasts = &memberlist.TransmitLimitedQueue{
                NumNodes: func() int {
                        nDB.RLock()
                        num := len(nDB.networkNodes[nid])
                        nDB.RUnlock()
                        return num
                },
                RetransmitMult: 4,
        }
        nDB.networkNodes[nid] = append(nDB.networkNodes[nid], nDB.config.NodeName)
        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("%s: joined network %s", nDB.config.NodeName, 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()
        var (
                paths   []string
                entries []*entry
        )

        nwWalker := func(path string, v interface{}) bool {
                entry, ok := v.(*entry)
                if !ok {
                        return false
                }
                paths = append(paths, path)
                entries = append(entries, entry)
                return false
        }

        nDB.indexes[byNetwork].WalkPrefix(fmt.Sprintf("/%s", nid), nwWalker)
        for _, path := range paths {
                params := strings.Split(path[1:], "/")
                tname := 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)
                }
        }

        nodeNetworks, ok := nDB.networks[nDB.config.NodeName]
        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)
        }

        n.ltime = ltime
        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 := nDB.networkNodes[nid]
        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.NodeName] {
                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.NodeName] {
                n.ltime = ltime
        }
}

func (nDB *NetworkDB) updateLocalTableTime() {
        nDB.Lock()
        defer nDB.Unlock()

        ltime := nDB.tableClock.Increment()
        nDB.indexes[byTable].Walk(func(path string, v interface{}) bool {
                entry := v.(*entry)
                if entry.node != nDB.config.NodeName {
                        return false
                }

                params := strings.Split(path[1:], "/")
                tname := params[0]
                nid := params[1]
                key := params[2]
                entry.ltime = ltime

                nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)
                nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)

                return false
        })
}

1	package networkdb
2
3	//go:generate protoc -I.:../Godeps/_workspace/src/github.com/gogo/protobuf --gogo_out=import_path=github.com/docker/libnetwork/networkdb,Mgogoproto/gogo.proto=github.com/gogo/protobuf/gogoproto:. networkdb.proto
4
5	import (
6	"fmt"
7	"strings"
8	"sync"
9	"time"
10
11	"github.com/Sirupsen/logrus"
12	"github.com/armon/go-radix"
13	"github.com/docker/go-events"
14	"github.com/docker/libnetwork/types"
15	"github.com/hashicorp/memberlist"
16	"github.com/hashicorp/serf/serf"
17	)
18
19	const (
20	byTable int = 1 + iota
21	byNetwork
22	)
23
24	// NetworkDB instance drives the networkdb cluster and acts the broker
25	// for cluster-scoped and network-scoped gossip and watches.
26	type NetworkDB struct {
27	// The clocks MUST be the first things
28	// in this struct due to Golang issue #599.
29
30	// Global lamport clock for node network attach events.
31	networkClock serf.LamportClock
32
33	// Global lamport clock for table events.
34	tableClock serf.LamportClock
35
36	sync.RWMutex
37
38	// NetworkDB configuration.
39	config *Config
40
41	// All the tree index (byTable, byNetwork) that we maintain
42	// the db.
43	indexes map[int]*radix.Tree
44
45	// Memberlist we use to drive the cluster.
46	memberlist *memberlist.Memberlist
47
48	// List of all peer nodes in the cluster not-limited to any
49	// network.
50	nodes map[string]*node
51
52	// List of all peer nodes which have failed
53	failedNodes map[string]*node
54
55	// List of all peer nodes which have left
56	leftNodes map[string]*node
57
58	// A multi-dimensional map of network/node attachmemts. The
59	// first key is a node name and the second key is a network ID
60	// for the network that node is participating in.
61	networks map[string]map[string]*network
62
63	// A map of nodes which are participating in a given
64	// network. The key is a network ID.
65	networkNodes map[string][]string
66
67	// A table of ack channels for every node from which we are
68	// waiting for an ack.
69	bulkSyncAckTbl map[string]chan struct{}
70
71	// Broadcast queue for network event gossip.
72	networkBroadcasts *memberlist.TransmitLimitedQueue
73
74	// Broadcast queue for node event gossip.
75	nodeBroadcasts *memberlist.TransmitLimitedQueue
76
77	// A central stop channel to stop all go routines running on
78	// behalf of the NetworkDB instance.
79	stopCh chan struct{}
80
81	// A central broadcaster for all local watchers watching table
82	// events.
83	broadcaster *events.Broadcaster
84
85	// List of all tickers which needed to be stopped when
86	// cleaning up.
87	tickers []*time.Ticker
88
89	// Reference to the memberlist's keyring to add & remove keys
90	keyring *memberlist.Keyring
91	}
92
93	// PeerInfo represents the peer (gossip cluster) nodes of a network
94	type PeerInfo struct {
95	Name string
96	IP string
97	}
98
99	type node struct {
100	memberlist.Node
101	ltime serf.LamportTime
102	// Number of hours left before the reaper removes the node
103	reapTime time.Duration
104	}
105
106	// network describes the node/network attachment.
107	type network struct {
108	// Network ID
109	id string
110
111	// Lamport time for the latest state of the entry.
112	ltime serf.LamportTime
113
114	// Node leave is in progress.
115	leaving bool
116
117	// Number of seconds still left before a deleted network entry gets
118	// removed from networkDB
119	reapTime time.Duration
120
121	// The broadcast queue for table event gossip. This is only
122	// initialized for this node's network attachment entries.
123	tableBroadcasts *memberlist.TransmitLimitedQueue
124	}
125
126	// Config represents the configuration of the networdb instance and
127	// can be passed by the caller.
128	type Config struct {
129	// NodeName is the cluster wide unique name for this node.
130	NodeName string
131
132	// BindAddr is the IP on which networkdb listens. It can be
133	// 0.0.0.0 to listen on all addresses on the host.
134	BindAddr string
135
136	// AdvertiseAddr is the node's IP address that we advertise for
137	// cluster communication.
138	AdvertiseAddr string
139
140	// BindPort is the local node's port to which we bind to for
141	// cluster communication.
142	BindPort int
143
144	// Keys to be added to the Keyring of the memberlist. Key at index
145	// 0 is the primary key
146	Keys [][]byte
147	}
148
149	// entry defines a table entry
150	type entry struct {
151	// node from which this entry was learned.
152	node string
153
154	// Lamport time for the most recent update to the entry
155	ltime serf.LamportTime
156
157	// Opaque value store in the entry
158	value []byte
159
160	// Deleting the entry is in progress. All entries linger in
161	// the cluster for certain amount of time after deletion.
162	deleting bool
163
164	// Number of seconds still left before a deleted table entry gets
165	// removed from networkDB
166	reapTime time.Duration
167	}
168
169	// New creates a new instance of NetworkDB using the Config passed by
170	// the caller.
171	func New(c Config) (NetworkDB, error) {	22✔
172	nDB := &NetworkDB{	22✔
173	config: c,	22✔
174	indexes: make(map[int]*radix.Tree),	22✔
175	networks: make(map[string]map[string]*network),	22✔
176	nodes: make(map[string]*node),	22✔
177	failedNodes: make(map[string]*node),	22✔
178	leftNodes: make(map[string]*node),	22✔
179	networkNodes: make(map[string][]string),	22✔
180	bulkSyncAckTbl: make(map[string]chan struct{}),	22✔
181	broadcaster: events.NewBroadcaster(),	22✔
182	}	22✔
183		22✔
184	nDB.indexes[byTable] = radix.New()	22✔
185	nDB.indexes[byNetwork] = radix.New()	22✔
186		22✔
187	if err := nDB.clusterInit(); err != nil {	22✔
188	return nil, err	×
189	}	×
190
191	return nDB, nil	22✔
192	}
193
194	// Join joins this NetworkDB instance with a list of peer NetworkDB
195	// instances passed by the caller in the form of addr:port
196	func (nDB *NetworkDB) Join(members []string) error {	13✔
197	return nDB.clusterJoin(members)	13✔
198	}	13✔
199
200	// Close destroys this NetworkDB instance by leave the cluster,
201	// stopping timers, canceling goroutines etc.
202	func (nDB *NetworkDB) Close() {	22✔
203	if err := nDB.clusterLeave(); err != nil {	22✔
204	logrus.Errorf("Could not close DB %s: %v", nDB.config.NodeName, err)	×
205	}	×
206	}
207
208	// Peers returns the gossip peers for a given network.
209	func (nDB *NetworkDB) Peers(nid string) []PeerInfo {	×
210	nDB.RLock()	×
211	defer nDB.RUnlock()	×
212	peers := make([]PeerInfo, 0, len(nDB.networkNodes[nid]))	×
213	for _, nodeName := range nDB.networkNodes[nid] {	×
214	if node, ok := nDB.nodes[nodeName]; ok {	×
215	peers = append(peers, PeerInfo{	×
216	Name: node.Name,	×
217	IP: node.Addr.String(),	×
218	})	×
219	}	×
220	}
221	return peers	×
222	}
223
224	// GetEntry retrieves the value of a table entry in a given (network,
225	// table, key) tuple
226	func (nDB *NetworkDB) GetEntry(tname, nid, key string) ([]byte, error) {	26✔
227	entry, err := nDB.getEntry(tname, nid, key)	26✔
228	if err != nil {	26✔
229	return nil, err	×
230	}	×
231
232	return entry.value, nil	26✔
233	}
234
235	func (nDB NetworkDB) getEntry(tname, nid, key string) (entry, error) {	3,408✔
236	nDB.RLock()	3,408✔
237	defer nDB.RUnlock()	3,408✔
238		3,408✔
239	e, ok := nDB.indexes[byTable].Get(fmt.Sprintf("/%s/%s/%s", tname, nid, key))	3,408✔
240	if !ok {	5,488✔
241	return nil, types.NotFoundErrorf("could not get entry in table %s with network id %s and key %s", tname, nid, key)	2,080✔
242	}	2,080✔
243
244	return e.(*entry), nil	1,328✔
245	}
246
247	// CreateEntry creates a table entry in NetworkDB for given (network,
248	// table, key) tuple and if the NetworkDB is part of the cluster
249	// propagates this event to the cluster. It is an error to create an
250	// entry for the same tuple for which there is already an existing
251	// entry unless the current entry is deleting state.
252	func (nDB *NetworkDB) CreateEntry(tname, nid, key string, value []byte) error {	1,024✔
253	oldEntry, err := nDB.getEntry(tname, nid, key)	1,024✔
254	if err != nil {	2,048✔
255	if _, ok := err.(types.NotFoundError); !ok {	1,024✔
256	return fmt.Errorf("cannot create entry in table %s with network id %s and key %s: %v", tname, nid, key, err)	×
257	}	×
258	}
259	if oldEntry != nil && !oldEntry.deleting {	1,024✔
260	return fmt.Errorf("cannot create entry in table %s with network id %s and key %s, already exists", tname, nid, key)	×
261	}	×
262
263	entry := &entry{	1,024✔
264	ltime: nDB.tableClock.Increment(),	1,024✔
265	node: nDB.config.NodeName,	1,024✔
266	value: value,	1,024✔
267	}	1,024✔
268		1,024✔
269	if err := nDB.sendTableEvent(TableEventTypeCreate, nid, tname, key, entry); err != nil {	1,024✔
270	return fmt.Errorf("cannot send create event for table %s, %v", tname, err)	×
271	}	×
272
273	nDB.Lock()	1,024✔
274	nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)	1,024✔
275	nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)	1,024✔
276	nDB.Unlock()	1,024✔
277		1,024✔
278	nDB.broadcaster.Write(makeEvent(opCreate, tname, nid, key, value))	1,024✔
279	return nil	1,024✔
280	}
281
282	// UpdateEntry updates a table entry in NetworkDB for given (network,
283	// table, key) tuple and if the NetworkDB is part of the cluster
284	// propagates this event to the cluster. It is an error to update a
285	// non-existent entry.
286	func (nDB *NetworkDB) UpdateEntry(tname, nid, key string, value []byte) error {	3✔
287	if _, err := nDB.GetEntry(tname, nid, key); err != nil {	3✔
288	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)	×
289	}	×
290
291	entry := &entry{	3✔
292	ltime: nDB.tableClock.Increment(),	3✔
293	node: nDB.config.NodeName,	3✔
294	value: value,	3✔
295	}	3✔
296		3✔
297	if err := nDB.sendTableEvent(TableEventTypeUpdate, nid, tname, key, entry); err != nil {	3✔
298	return fmt.Errorf("cannot send table update event: %v", err)	×
299	}	×
300
301	nDB.Lock()	3✔
302	nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)	3✔
303	nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)	3✔
304	nDB.Unlock()	3✔
305		3✔
306	nDB.broadcaster.Write(makeEvent(opUpdate, tname, nid, key, value))	3✔
307	return nil	3✔
308	}
309
310	// GetTableByNetwork walks the networkdb by the give table and network id and
311	// returns a map of keys and values
312	func (nDB *NetworkDB) GetTableByNetwork(tname, nid string) map[string]interface{} {	×
313	entries := make(map[string]interface{})	×
314	nDB.indexes[byTable].WalkPrefix(fmt.Sprintf("/%s/%s", tname, nid), func(k string, v interface{}) bool {	×
315	entry := v.(*entry)	×
316	if entry.deleting {	×
317	return false	×
318	}	×
319	key := k[strings.LastIndex(k, "/")+1:]	×
320	entries[key] = entry.value	×
321	return false	×
322	})
323	return entries	×
324	}
325
326	// DeleteEntry deletes a table entry in NetworkDB for given (network,
327	// table, key) tuple and if the NetworkDB is part of the cluster
328	// propagates this event to the cluster.
329	func (nDB *NetworkDB) DeleteEntry(tname, nid, key string) error {	23✔
330	value, err := nDB.GetEntry(tname, nid, key)	23✔
331	if err != nil {	23✔
332	return fmt.Errorf("cannot delete entry as the entry in table %s with network id %s and key %s does not exist", tname, nid, key)	×
333	}	×
334
335	entry := &entry{	23✔
336	ltime: nDB.tableClock.Increment(),	23✔
337	node: nDB.config.NodeName,	23✔
338	value: value,	23✔
339	deleting: true,	23✔
340	reapTime: reapInterval,	23✔
341	}	23✔
342		23✔
343	if err := nDB.sendTableEvent(TableEventTypeDelete, nid, tname, key, entry); err != nil {	23✔
344	return fmt.Errorf("cannot send table delete event: %v", err)	×
345	}	×
346
347	nDB.Lock()	23✔
348	nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)	23✔
349	nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)	23✔
350	nDB.Unlock()	23✔
351		23✔
352	nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, value))	23✔
353	return nil	23✔
354	}
355
356	func (nDB *NetworkDB) deleteNetworkEntriesForNode(deletedNode string) {	42✔
357	nDB.Lock()	42✔
358	for nid, nodes := range nDB.networkNodes {	138✔
359	updatedNodes := make([]string, 0, len(nodes))	96✔
360	for _, node := range nodes {	221✔
361	if node == deletedNode {	179✔
362	continue	54✔
363	}
364
365	updatedNodes = append(updatedNodes, node)	71✔
366	}
367
368	nDB.networkNodes[nid] = updatedNodes	96✔
369	}
370
371	delete(nDB.networks, deletedNode)	42✔
372	nDB.Unlock()	42✔
373	}
374
375	func (nDB *NetworkDB) deleteNodeTableEntries(node string) {	42✔
376	nDB.Lock()	42✔
377	nDB.indexes[byTable].Walk(func(path string, v interface{}) bool {	3,126✔
378	oldEntry := v.(*entry)	3,084✔
379	if oldEntry.node != node {	4,127✔
380	return false	1,043✔
381	}	1,043✔
382
383	params := strings.Split(path[1:], "/")	2,041✔
384	tname := params[0]	2,041✔
385	nid := params[1]	2,041✔
386	key := params[2]	2,041✔
387		2,041✔
388	entry := &entry{	2,041✔
389	ltime: oldEntry.ltime,	2,041✔
390	node: node,	2,041✔
391	value: oldEntry.value,	2,041✔
392	deleting: true,	2,041✔
393	reapTime: reapInterval,	2,041✔
394	}	2,041✔
395		2,041✔
396	nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)	2,041✔
397	nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)	2,041✔
398		2,041✔
399	nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, entry.value))	2,041✔
400	return false	2,041✔
401	})
402	nDB.Unlock()	42✔
403	}
404
405	// WalkTable walks a single table in NetworkDB and invokes the passed
406	// function for each entry in the table passing the network, key,
407	// value. The walk stops if the passed function returns a true.
408	func (nDB *NetworkDB) WalkTable(tname string, fn func(string, string, []byte) bool) error {	×
409	nDB.RLock()	×
410	values := make(map[string]interface{})	×
411	nDB.indexes[byTable].WalkPrefix(fmt.Sprintf("/%s", tname), func(path string, v interface{}) bool {	×
412	values[path] = v	×
413	return false	×
414	})	×
415	nDB.RUnlock()	×
416		×
417	for k, v := range values {	×
418	params := strings.Split(k[1:], "/")	×
419	nid := params[1]	×
420	key := params[2]	×
421	if fn(nid, key, v.(*entry).value) {	×
422	return nil	×
423	}	×
424	}
425
426	return nil	×
427	}
428
429	// JoinNetwork joins this node to a given network and propagates this
430	// event across the cluster. This triggers this node joining the
431	// sub-cluster of this network and participates in the network-scoped
432	// gossip and bulk sync for this network.
433	func (nDB *NetworkDB) JoinNetwork(nid string) error {	36✔
434	ltime := nDB.networkClock.Increment()	36✔
435		36✔
436	nDB.Lock()	36✔
437	nodeNetworks, ok := nDB.networks[nDB.config.NodeName]	36✔
438	if !ok {	54✔
439	nodeNetworks = make(map[string]*network)	18✔
440	nDB.networks[nDB.config.NodeName] = nodeNetworks	18✔
441	}	18✔
442	nodeNetworks[nid] = &network{id: nid, ltime: ltime}	36✔
443	nodeNetworks[nid].tableBroadcasts = &memberlist.TransmitLimitedQueue{	36✔
444	NumNodes: func() int {	117✔
445	nDB.RLock()	81✔
446	num := len(nDB.networkNodes[nid])	81✔
447	nDB.RUnlock()	81✔
448	return num	81✔
449	},	81✔
450	RetransmitMult: 4,
451	}
452	nDB.networkNodes[nid] = append(nDB.networkNodes[nid], nDB.config.NodeName)	36✔
453	networkNodes := nDB.networkNodes[nid]	36✔
454	nDB.Unlock()	36✔
455		36✔
456	if err := nDB.sendNetworkEvent(nid, NetworkEventTypeJoin, ltime); err != nil {	36✔
457	return fmt.Errorf("failed to send leave network event for %s: %v", nid, err)	×
458	}	×
459
460	logrus.Debugf("%s: joined network %s", nDB.config.NodeName, nid)	36✔
461	if _, err := nDB.bulkSync(networkNodes, true); err != nil {	36✔
462	logrus.Errorf("Error bulk syncing while joining network %s: %v", nid, err)	×
463	}	×
464
465	return nil	36✔
466	}
467
468	// LeaveNetwork leaves this node from a given network and propagates
469	// this event across the cluster. This triggers this node leaving the
470	// sub-cluster of this network and as a result will no longer
471	// participate in the network-scoped gossip and bulk sync for this
472	// network. Also remove all the table entries for this network from
473	// networkdb
474	func (nDB *NetworkDB) LeaveNetwork(nid string) error {	21✔
475	ltime := nDB.networkClock.Increment()	21✔
476	if err := nDB.sendNetworkEvent(nid, NetworkEventTypeLeave, ltime); err != nil {	21✔
477	return fmt.Errorf("failed to send leave network event for %s: %v", nid, err)	×
478	}	×
479
480	nDB.Lock()	21✔
481	defer nDB.Unlock()	21✔
482	var (	21✔
483	paths []string	21✔
484	entries []*entry	21✔
485	)	21✔
486		21✔
487	nwWalker := func(path string, v interface{}) bool {	21✔
488	entry, ok := v.(*entry)	×
489	if !ok {	×
490	return false	×
491	}	×
492	paths = append(paths, path)	×
493	entries = append(entries, entry)	×
494	return false	×
495	}
496
497	nDB.indexes[byNetwork].WalkPrefix(fmt.Sprintf("/%s", nid), nwWalker)	21✔
498	for _, path := range paths {	21✔
499	params := strings.Split(path[1:], "/")	×
500	tname := params[1]	×
501	key := params[2]	×
502		×
503	if _, ok := nDB.indexes[byTable].Delete(fmt.Sprintf("/%s/%s/%s", tname, nid, key)); !ok {	×
504	logrus.Errorf("Could not delete entry in table %s with network id %s and key %s as it does not exist", tname, nid, key)	×
505	}	×
506
507	if _, ok := nDB.indexes[byNetwork].Delete(fmt.Sprintf("/%s/%s/%s", nid, tname, key)); !ok {	×
508	logrus.Errorf("Could not delete entry in network %s with table name %s and key %s as it does not exist", nid, tname, key)	×
509	}	×
510	}
511
512	nodeNetworks, ok := nDB.networks[nDB.config.NodeName]	21✔
513	if !ok {	21✔
514	return fmt.Errorf("could not find self node for network %s while trying to leave", nid)	×
515	}	×
516
517	n, ok := nodeNetworks[nid]	21✔
518	if !ok {	21✔
519	return fmt.Errorf("could not find network %s while trying to leave", nid)	×
520	}	×
521
522	n.ltime = ltime	21✔
523	n.leaving = true	21✔
524	return nil	21✔
525	}
526
527	// addNetworkNode adds the node to the list of nodes which participate
528	// in the passed network only if it is not already present. Caller
529	// should hold the NetworkDB lock while calling this
530	func (nDB *NetworkDB) addNetworkNode(nid string, nodeName string) {	74✔
531	nodes := nDB.networkNodes[nid]	74✔
532	for _, node := range nodes {	153✔
533	if node == nodeName {	100✔
534	return	21✔
535	}	21✔
536	}
537
538	nDB.networkNodes[nid] = append(nDB.networkNodes[nid], nodeName)	53✔
539	}
540
541	// Deletes the node from the list of nodes which participate in the
542	// passed network. Caller should hold the NetworkDB lock while calling
543	// this
544	func (nDB *NetworkDB) deleteNetworkNode(nid string, nodeName string) {	10✔
545	nodes := nDB.networkNodes[nid]	10✔
546	newNodes := make([]string, 0, len(nodes)-1)	10✔
547	for _, name := range nodes {	20✔
548	if name == nodeName {	20✔
549	continue	10✔
550	}
551	newNodes = append(newNodes, name)	×
552	}
553	nDB.networkNodes[nid] = newNodes	10✔
554	}
555
556	// findCommonnetworks find the networks that both this node and the
557	// passed node have joined.
558	func (nDB *NetworkDB) findCommonNetworks(nodeName string) []string {	3✔
559	nDB.RLock()	3✔
560	defer nDB.RUnlock()	3✔
561		3✔
562	var networks []string	3✔
563	for nid := range nDB.networks[nDB.config.NodeName] {	6✔
564	if n, ok := nDB.networks[nodeName][nid]; ok {	6✔
565	if !n.leaving {	6✔
566	networks = append(networks, nid)	3✔
567	}	3✔
568	}
569	}
570
571	return networks	3✔
572	}
573
574	func (nDB *NetworkDB) updateLocalNetworkTime() {	52✔
575	nDB.Lock()	52✔
576	defer nDB.Unlock()	52✔
577		52✔
578	ltime := nDB.networkClock.Increment()	52✔
579	for _, n := range nDB.networks[nDB.config.NodeName] {	52✔
580	n.ltime = ltime	×
581	}	×
582	}
583
584	func (nDB *NetworkDB) updateLocalTableTime() {	×
585	nDB.Lock()	×
586	defer nDB.Unlock()	×
587		×
588	ltime := nDB.tableClock.Increment()	×
589	nDB.indexes[byTable].Walk(func(path string, v interface{}) bool {	×
590	entry := v.(*entry)	×
591	if entry.node != nDB.config.NodeName {	×
592	return false	×
593	}	×
594
595	params := strings.Split(path[1:], "/")	×
596	tname := params[0]	×
597	nid := params[1]	×
598	key := params[2]	×
599	entry.ltime = ltime	×
600		×
601	nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)	×
602	nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)	×
603		×
604	return false	×
605	})
606	}

docker / libnetwork / #3345

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous