diff --git a/components/engine/vendor.conf b/components/engine/vendor.conf index 8c32cadc8f..d46fe89c07 100644 --- a/components/engine/vendor.conf +++ b/components/engine/vendor.conf @@ -27,7 +27,7 @@ github.com/imdario/mergo 0.2.1 golang.org/x/sync de49d9dcd27d4f764488181bea099dfe6179bcf0 #get libnetwork packages -github.com/docker/libnetwork 20bf4e23da1bb3f9d3dc70e4cf531b84c45947bd +github.com/docker/libnetwork f0b9939b255aa2d2be3ec03d478a0c5e030ee07e github.com/docker/go-events 18b43f1bc85d9cdd42c05a6cd2d444c7a200a894 github.com/armon/go-radix e39d623f12e8e41c7b5529e9a9dd67a1e2261f80 github.com/armon/go-metrics eb0af217e5e9747e41dd5303755356b62d28e3ec @@ -145,4 +145,4 @@ github.com/opencontainers/selinux v1.0.0-rc1 # git --git-dir ./go/.git --work-tree ./go checkout revert-prefix-ignore # cp -a go/src/archive/tar ./vendor/archive/tar # rm -rf ./go -# vndr \ No newline at end of file +# vndr diff --git a/components/engine/vendor/github.com/docker/libnetwork/common/caller.go b/components/engine/vendor/github.com/docker/libnetwork/common/caller.go new file mode 100644 index 0000000000..0dec3bc0bc --- /dev/null +++ b/components/engine/vendor/github.com/docker/libnetwork/common/caller.go @@ -0,0 +1,29 @@ +package common + +import ( + "runtime" + "strings" +) + +func callerInfo(i int) string { + ptr, _, _, ok := runtime.Caller(i) + fName := "unknown" + if ok { + f := runtime.FuncForPC(ptr) + if f != nil { + // f.Name() is like: github.com/docker/libnetwork/common.MethodName + tmp := strings.Split(f.Name(), ".") + if len(tmp) > 0 { + fName = tmp[len(tmp)-1] + } + } + } + + return fName +} + +// CallerName returns the name of the function at the specified level +// level == 0 means current method name +func CallerName(level int) string { + return callerInfo(2 + level) +} diff --git a/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/joinleave.go b/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/joinleave.go index cdbb428281..31c311f4fc 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/joinleave.go +++ b/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/joinleave.go @@ -120,8 +120,7 @@ func (d *driver) Join(nid, eid string, sboxKey string, jinfo driverapi.JoinInfo, } } - d.peerDbAdd(nid, eid, ep.addr.IP, ep.addr.Mask, ep.mac, - net.ParseIP(d.advertiseAddress), true) + d.peerAdd(nid, eid, ep.addr.IP, ep.addr.Mask, ep.mac, net.ParseIP(d.advertiseAddress), true, false, false, true) if err := d.checkEncryption(nid, nil, n.vxlanID(s), true, true); err != nil { logrus.Warn(err) @@ -205,7 +204,7 @@ func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key stri return } - d.peerAdd(nid, eid, addr.IP, addr.Mask, mac, vtep, true, false, false) + d.peerAdd(nid, eid, addr.IP, addr.Mask, mac, vtep, true, false, false, false) } // Leave method is invoked when a Sandbox detaches from an endpoint. diff --git a/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/ov_network.go b/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/ov_network.go index 6be88d9179..cd7e94848c 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/ov_network.go +++ b/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/ov_network.go @@ -767,9 +767,7 @@ func (n *network) watchMiss(nlSock *nl.NetlinkSocket) { continue } - if err := n.driver.peerAdd(n.id, "dummy", ip, IPmask, mac, vtep, true, l2Miss, l3Miss); err != nil { - logrus.Errorf("could not add neighbor entry for missed peer %q: %v", ip, err) - } + n.driver.peerAdd(n.id, "dummy", ip, IPmask, mac, vtep, true, l2Miss, l3Miss, false) } } } diff --git a/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/ov_serf.go b/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/ov_serf.go index 9002bce6b7..20954ef237 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/ov_serf.go +++ b/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/ov_serf.go @@ -120,15 +120,10 @@ func (d *driver) processEvent(u serf.UserEvent) { switch action { case "join": - if err := d.peerAdd(nid, eid, net.ParseIP(ipStr), net.IPMask(net.ParseIP(maskStr).To4()), mac, - net.ParseIP(vtepStr), true, false, false); err != nil { - logrus.Errorf("Peer add failed in the driver: %v\n", err) - } + d.peerAdd(nid, eid, net.ParseIP(ipStr), net.IPMask(net.ParseIP(maskStr).To4()), mac, net.ParseIP(vtepStr), + true, false, false, false) case "leave": - if err := d.peerDelete(nid, eid, net.ParseIP(ipStr), net.IPMask(net.ParseIP(maskStr).To4()), mac, - net.ParseIP(vtepStr), true); err != nil { - logrus.Errorf("Peer delete failed in the driver: %v\n", err) - } + d.peerDelete(nid, eid, net.ParseIP(ipStr), net.IPMask(net.ParseIP(maskStr).To4()), mac, net.ParseIP(vtepStr), true) } } diff --git a/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/overlay.go b/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/overlay.go index 8d19b2e1d4..11eda6781b 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/overlay.go +++ b/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/overlay.go @@ -3,6 +3,7 @@ package overlay //go:generate protoc -I.:../../Godeps/_workspace/src/github.com/gogo/protobuf --gogo_out=import_path=github.com/docker/libnetwork/drivers/overlay,Mgogoproto/gogo.proto=github.com/gogo/protobuf/gogoproto:. overlay.proto import ( + "context" "fmt" "net" "sync" @@ -50,6 +51,8 @@ type driver struct { joinOnce sync.Once localJoinOnce sync.Once keys []*key + peerOpCh chan *peerOperation + peerOpCancel context.CancelFunc sync.Mutex } @@ -64,10 +67,16 @@ func Init(dc driverapi.DriverCallback, config map[string]interface{}) error { peerDb: peerNetworkMap{ mp: map[string]*peerMap{}, }, - secMap: &encrMap{nodes: map[string][]*spi{}}, - config: config, + secMap: &encrMap{nodes: map[string][]*spi{}}, + config: config, + peerOpCh: make(chan *peerOperation), } + // Launch the go routine for processing peer operations + ctx, cancel := context.WithCancel(context.Background()) + d.peerOpCancel = cancel + go d.peerOpRoutine(ctx, d.peerOpCh) + if data, ok := config[netlabel.GlobalKVClient]; ok { var err error dsc, ok := data.(discoverapi.DatastoreConfigData) @@ -161,7 +170,7 @@ func (d *driver) restoreEndpoints() error { } n.incEndpointCount() - d.peerDbAdd(ep.nid, ep.id, ep.addr.IP, ep.addr.Mask, ep.mac, net.ParseIP(d.advertiseAddress), true) + d.peerAdd(ep.nid, ep.id, ep.addr.IP, ep.addr.Mask, ep.mac, net.ParseIP(d.advertiseAddress), true, false, false, true) } return nil } @@ -170,6 +179,11 @@ func (d *driver) restoreEndpoints() error { func Fini(drv driverapi.Driver) { d := drv.(*driver) + // Notify the peer go routine to return + if d.peerOpCancel != nil { + d.peerOpCancel() + } + if d.exitCh != nil { waitCh := make(chan struct{}) diff --git a/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/peerdb.go b/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/peerdb.go index 6551e7bcad..47e30f1b27 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/peerdb.go +++ b/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/peerdb.go @@ -1,12 +1,14 @@ package overlay import ( + "context" "fmt" "net" "sync" "syscall" "github.com/Sirupsen/logrus" + "github.com/docker/libnetwork/common" ) const ovPeerTable = "overlay_peer_table" @@ -59,8 +61,6 @@ func (pKey *peerKey) Scan(state fmt.ScanState, verb rune) error { return nil } -var peerDbWg sync.WaitGroup - func (d *driver) peerDbWalk(f func(string, *peerKey, *peerEntry) bool) error { d.peerDb.Lock() nids := []string{} @@ -141,8 +141,6 @@ func (d *driver) peerDbSearch(nid string, peerIP net.IP) (net.HardwareAddr, net. func (d *driver) peerDbAdd(nid, eid string, peerIP net.IP, peerIPMask net.IPMask, peerMac net.HardwareAddr, vtep net.IP, isLocal bool) { - peerDbWg.Wait() - d.peerDb.Lock() pMap, ok := d.peerDb.mp[nid] if !ok { @@ -173,7 +171,6 @@ func (d *driver) peerDbAdd(nid, eid string, peerIP net.IP, peerIPMask net.IPMask func (d *driver) peerDbDelete(nid, eid string, peerIP net.IP, peerIPMask net.IPMask, peerMac net.HardwareAddr, vtep net.IP) peerEntry { - peerDbWg.Wait() d.peerDb.Lock() pMap, ok := d.peerDb.mp[nid] @@ -215,9 +212,6 @@ func (d *driver) peerDbUpdateSandbox(nid string) { } d.peerDb.Unlock() - peerDbWg.Add(1) - - var peerOps []func() pMap.Lock() for pKeyStr, pEntry := range pMap.mp { var pKey peerKey @@ -233,28 +227,67 @@ func (d *driver) peerDbUpdateSandbox(nid string) { // pointing to the same memory location for every iteration. Make // a copy of pEntry before capturing it in the following closure. entry := pEntry - op := func() { - if err := d.peerAdd(nid, entry.eid, pKey.peerIP, entry.peerIPMask, - pKey.peerMac, entry.vtep, - false, false, false); err != nil { - fmt.Printf("peerdbupdate in sandbox failed for ip %s and mac %s: %v", - pKey.peerIP, pKey.peerMac, err) - } - } - peerOps = append(peerOps, op) + d.peerAdd(nid, entry.eid, pKey.peerIP, entry.peerIPMask, pKey.peerMac, entry.vtep, false, false, false, false) } pMap.Unlock() +} - for _, op := range peerOps { - op() +type peerOperation struct { + isAdd bool + networkID string + endpointID string + peerIP net.IP + peerIPMask net.IPMask + peerMac net.HardwareAddr + vtepIP net.IP + updateDB bool + l2Miss bool + l3Miss bool + localPeer bool + callerName string +} + +func (d *driver) peerOpRoutine(ctx context.Context, ch chan *peerOperation) { + var err error + for { + select { + case <-ctx.Done(): + return + case op := <-ch: + if op.isAdd { + err = d.peerAddOp(op.networkID, op.endpointID, op.peerIP, op.peerIPMask, op.peerMac, op.vtepIP, op.updateDB, op.l2Miss, op.l3Miss, op.localPeer) + } else { + err = d.peerDeleteOp(op.networkID, op.endpointID, op.peerIP, op.peerIPMask, op.peerMac, op.vtepIP, op.localPeer) + } + if err != nil { + logrus.Warnf("Peer operation failed:%s op:%v", err, op) + } + } } - - peerDbWg.Done() } func (d *driver) peerAdd(nid, eid string, peerIP net.IP, peerIPMask net.IPMask, - peerMac net.HardwareAddr, vtep net.IP, updateDb, l2Miss, l3Miss bool) error { + peerMac net.HardwareAddr, vtep net.IP, updateDb, l2Miss, l3Miss, localPeer bool) { + callerName := common.CallerName(1) + d.peerOpCh <- &peerOperation{ + isAdd: true, + networkID: nid, + endpointID: eid, + peerIP: peerIP, + peerIPMask: peerIPMask, + peerMac: peerMac, + vtepIP: vtep, + updateDB: updateDb, + l2Miss: l2Miss, + l3Miss: l3Miss, + localPeer: localPeer, + callerName: callerName, + } +} + +func (d *driver) peerAddOp(nid, eid string, peerIP net.IP, peerIPMask net.IPMask, + peerMac net.HardwareAddr, vtep net.IP, updateDb, l2Miss, l3Miss, updateOnlyDB bool) error { if err := validateID(nid, eid); err != nil { return err @@ -262,6 +295,9 @@ func (d *driver) peerAdd(nid, eid string, peerIP net.IP, peerIPMask net.IPMask, if updateDb { d.peerDbAdd(nid, eid, peerIP, peerIPMask, peerMac, vtep, false) + if updateOnlyDB { + return nil + } } n := d.network(nid) @@ -311,6 +347,22 @@ func (d *driver) peerAdd(nid, eid string, peerIP net.IP, peerIPMask net.IPMask, } func (d *driver) peerDelete(nid, eid string, peerIP net.IP, peerIPMask net.IPMask, + peerMac net.HardwareAddr, vtep net.IP, updateDb bool) { + callerName := common.CallerName(1) + d.peerOpCh <- &peerOperation{ + isAdd: false, + networkID: nid, + endpointID: eid, + peerIP: peerIP, + peerIPMask: peerIPMask, + peerMac: peerMac, + vtepIP: vtep, + updateDB: updateDb, + callerName: callerName, + } +} + +func (d *driver) peerDeleteOp(nid, eid string, peerIP net.IP, peerIPMask net.IPMask, peerMac net.HardwareAddr, vtep net.IP, updateDb bool) error { if err := validateID(nid, eid); err != nil { diff --git a/components/engine/vendor/github.com/docker/libnetwork/iptables/iptables.go b/components/engine/vendor/github.com/docker/libnetwork/iptables/iptables.go index 20edb9b5d6..caa202b366 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/iptables/iptables.go +++ b/components/engine/vendor/github.com/docker/libnetwork/iptables/iptables.go @@ -151,11 +151,11 @@ func ProgramChain(c *ChainInfo, bridgeName string, hairpinMode, enable bool) err "-j", c.Name} if !Exists(Nat, "PREROUTING", preroute...) && enable { if err := c.Prerouting(Append, preroute...); err != nil { - return fmt.Errorf("Failed to inject docker in PREROUTING chain: %s", err) + return fmt.Errorf("Failed to inject %s in PREROUTING chain: %s", c.Name, err) } } else if Exists(Nat, "PREROUTING", preroute...) && !enable { if err := c.Prerouting(Delete, preroute...); err != nil { - return fmt.Errorf("Failed to remove docker in PREROUTING chain: %s", err) + return fmt.Errorf("Failed to remove %s in PREROUTING chain: %s", c.Name, err) } } output := []string{ @@ -167,11 +167,11 @@ func ProgramChain(c *ChainInfo, bridgeName string, hairpinMode, enable bool) err } if !Exists(Nat, "OUTPUT", output...) && enable { if err := c.Output(Append, output...); err != nil { - return fmt.Errorf("Failed to inject docker in OUTPUT chain: %s", err) + return fmt.Errorf("Failed to inject %s in OUTPUT chain: %s", c.Name, err) } } else if Exists(Nat, "OUTPUT", output...) && !enable { if err := c.Output(Delete, output...); err != nil { - return fmt.Errorf("Failed to inject docker in OUTPUT chain: %s", err) + return fmt.Errorf("Failed to inject %s in OUTPUT chain: %s", c.Name, err) } } case Filter: diff --git a/components/engine/vendor/github.com/docker/libnetwork/networkdb/broadcast.go b/components/engine/vendor/github.com/docker/libnetwork/networkdb/broadcast.go index 3fe9f6271a..52e96ec639 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/networkdb/broadcast.go +++ b/components/engine/vendor/github.com/docker/libnetwork/networkdb/broadcast.go @@ -114,7 +114,8 @@ type tableEventMessage struct { } func (m *tableEventMessage) Invalidates(other memberlist.Broadcast) bool { - return false + otherm := other.(*tableEventMessage) + return m.tname == otherm.tname && m.id == otherm.id && m.key == otherm.key } func (m *tableEventMessage) Message() []byte { diff --git a/components/engine/vendor/github.com/docker/libnetwork/networkdb/cluster.go b/components/engine/vendor/github.com/docker/libnetwork/networkdb/cluster.go index 9156d0da68..e8c00808fc 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/networkdb/cluster.go +++ b/components/engine/vendor/github.com/docker/libnetwork/networkdb/cluster.go @@ -287,13 +287,6 @@ func (nDB *NetworkDB) reconnectNode() { return } - // Update all the local table state to a new time to - // force update on the node we are trying to rejoin, just in - // case that node has these in deleting state still. This is - // facilitate fast convergence after recovering from a gossip - // failure. - nDB.updateLocalTableTime() - logrus.Debugf("Initiating bulk sync with node %s after reconnect", node.Name) nDB.bulkSync([]string{node.Name}, true) } @@ -310,12 +303,11 @@ func (nDB *NetworkDB) reapState() { func (nDB *NetworkDB) reapNetworks() { nDB.Lock() - for name, nn := range nDB.networks { + for _, nn := range nDB.networks { for id, n := range nn { if n.leaving { if n.reapTime <= 0 { delete(nn, id) - nDB.deleteNetworkNode(id, name) continue } n.reapTime -= reapPeriod diff --git a/components/engine/vendor/github.com/docker/libnetwork/networkdb/delegate.go b/components/engine/vendor/github.com/docker/libnetwork/networkdb/delegate.go index 6df358382f..21deacde9e 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/networkdb/delegate.go +++ b/components/engine/vendor/github.com/docker/libnetwork/networkdb/delegate.go @@ -127,25 +127,12 @@ func (nDB *NetworkDB) handleNodeEvent(nEvent *NodeEvent) bool { } func (nDB *NetworkDB) handleNetworkEvent(nEvent *NetworkEvent) bool { - var flushEntries bool // Update our local clock if the received messages has newer // time. nDB.networkClock.Witness(nEvent.LTime) nDB.Lock() - defer func() { - nDB.Unlock() - // When a node leaves a network on the last task removal cleanup the - // local entries for this network & node combination. When the tasks - // on a network are removed we could have missed the gossip updates. - // Not doing this cleanup can leave stale entries because bulksyncs - // from the node will no longer include this network state. - // - // deleteNodeNetworkEntries takes nDB lock. - if flushEntries { - nDB.deleteNodeNetworkEntries(nEvent.NetworkID, nEvent.NodeName) - } - }() + defer nDB.Unlock() if nEvent.NodeName == nDB.config.NodeName { return false @@ -173,10 +160,20 @@ func (nDB *NetworkDB) handleNetworkEvent(nEvent *NetworkEvent) bool { n.leaving = nEvent.Type == NetworkEventTypeLeave if n.leaving { n.reapTime = reapInterval - flushEntries = true + + // 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) } - nDB.addNetworkNode(nEvent.NetworkID, nEvent.NodeName) return true } @@ -203,17 +200,22 @@ func (nDB *NetworkDB) handleTableEvent(tEvent *TableEvent) bool { nDB.RLock() networks := nDB.networks[nDB.config.NodeName] network, ok := networks[tEvent.NetworkID] - nDB.RUnlock() - if !ok || network.leaving { - return true + // 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 + } } - - e, err := nDB.getEntry(tEvent.TableName, tEvent.NetworkID, tEvent.Key) - if err != nil && tEvent.Type == TableEventTypeDelete { - // If it is a delete event and we don't have the entry here nothing to do. + 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 } + 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. @@ -238,6 +240,11 @@ func (nDB *NetworkDB) handleTableEvent(tEvent *TableEvent) bool { nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", tEvent.NetworkID, tEvent.TableName, tEvent.Key), e) nDB.Unlock() + if err != nil && tEvent.Type == TableEventTypeDelete { + // If it is a delete event and we didn't have the entry here don't repropagate + return true + } + var op opType switch tEvent.Type { case TableEventTypeCreate: @@ -278,8 +285,7 @@ func (nDB *NetworkDB) handleTableMessage(buf []byte, isBulkSync bool) { return } - // Do not rebroadcast a bulk sync - if rebroadcast := nDB.handleTableEvent(&tEvent); rebroadcast && !isBulkSync { + if rebroadcast := nDB.handleTableEvent(&tEvent); rebroadcast { var err error buf, err = encodeRawMessage(MessageTypeTableEvent, buf) if err != nil { diff --git a/components/engine/vendor/github.com/docker/libnetwork/networkdb/event_delegate.go b/components/engine/vendor/github.com/docker/libnetwork/networkdb/event_delegate.go index 23e16832e7..0f32194cf3 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/networkdb/event_delegate.go +++ b/components/engine/vendor/github.com/docker/libnetwork/networkdb/event_delegate.go @@ -45,9 +45,12 @@ func (e *eventDelegate) NotifyLeave(mn *memberlist.Node) { var failed bool logrus.Infof("Node %s/%s, left gossip cluster", mn.Name, mn.Addr) e.broadcastNodeEvent(mn.Addr, opDelete) - e.nDB.deleteNodeTableEntries(mn.Name) - e.nDB.deleteNetworkEntriesForNode(mn.Name) + // The node left or failed, 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 e.nDB.Lock() + e.nDB.deleteNetworkEntriesForNode(mn.Name) + e.nDB.deleteNodeTableEntries(mn.Name) if n, ok := e.nDB.nodes[mn.Name]; ok { delete(e.nDB.nodes, mn.Name) @@ -61,7 +64,6 @@ func (e *eventDelegate) NotifyLeave(mn *memberlist.Node) { if failed { logrus.Infof("Node %s/%s, added to failed nodes list", mn.Name, mn.Addr) } - } func (e *eventDelegate) NotifyUpdate(n *memberlist.Node) { diff --git a/components/engine/vendor/github.com/docker/libnetwork/networkdb/networkdb.go b/components/engine/vendor/github.com/docker/libnetwork/networkdb/networkdb.go index b93a90d019..69be8e99a8 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/networkdb/networkdb.go +++ b/components/engine/vendor/github.com/docker/libnetwork/networkdb/networkdb.go @@ -361,7 +361,6 @@ func (nDB *NetworkDB) DeleteEntry(tname, nid, key string) error { } func (nDB *NetworkDB) deleteNetworkEntriesForNode(deletedNode string) { - nDB.Lock() for nid, nodes := range nDB.networkNodes { updatedNodes := make([]string, 0, len(nodes)) for _, node := range nodes { @@ -376,11 +375,25 @@ func (nDB *NetworkDB) deleteNetworkEntriesForNode(deletedNode string) { } delete(nDB.networks, deletedNode) - nDB.Unlock() } +// 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) { - nDB.Lock() + // Indicates if the delete is triggered for the local node + isNodeLocal := node == nDB.config.NodeName + nDB.indexes[byNetwork].WalkPrefix(fmt.Sprintf("/%s", nid), func(path string, v interface{}) bool { oldEntry := v.(*entry) @@ -389,7 +402,15 @@ func (nDB *NetworkDB) deleteNodeNetworkEntries(nid, node string) { tname := params[1] key := params[2] - if oldEntry.node != node { + // 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 } @@ -401,17 +422,29 @@ func (nDB *NetworkDB) deleteNodeNetworkEntries(nid, node string) { 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) + // 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++ + } + 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) + } else { + // the local node is leaving the network, all the entries of remote nodes can be safely removed + nDB.indexes[byTable].Delete(fmt.Sprintf("/%s/%s/%s", tname, nid, key)) + nDB.indexes[byNetwork].Delete(fmt.Sprintf("/%s/%s/%s", nid, tname, key)) + } nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, entry.value)) return false }) - 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 { @@ -423,21 +456,12 @@ func (nDB *NetworkDB) deleteNodeTableEntries(node string) { nid := params[1] key := params[2] - entry := &entry{ - ltime: oldEntry.ltime, - node: node, - value: oldEntry.value, - deleting: true, - reapTime: reapInterval, - } + nDB.indexes[byTable].Delete(fmt.Sprintf("/%s/%s/%s", tname, nid, key)) + nDB.indexes[byNetwork].Delete(fmt.Sprintf("/%s/%s/%s", nid, tname, key)) - 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)) + nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, oldEntry.value)) return false }) - nDB.Unlock() } // WalkTable walks a single table in NetworkDB and invokes the passed @@ -487,7 +511,7 @@ func (nDB *NetworkDB) JoinNetwork(nid string) error { }, RetransmitMult: 4, } - nDB.networkNodes[nid] = append(nDB.networkNodes[nid], nDB.config.NodeName) + nDB.addNetworkNode(nid, nDB.config.NodeName) networkNodes := nDB.networkNodes[nid] nDB.Unlock() @@ -517,35 +541,12 @@ func (nDB *NetworkDB) LeaveNetwork(nid string) error { 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 - } + // Remove myself from the list of the nodes participating to the network + nDB.deleteNetworkNode(nid, nDB.config.NodeName) - 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) - } - } + // Update all the local entries marking them for deletion and delete all the remote entries + nDB.deleteNodeNetworkEntries(nid, nDB.config.NodeName) nodeNetworks, ok := nDB.networks[nDB.config.NodeName] if !ok { @@ -558,6 +559,7 @@ func (nDB *NetworkDB) LeaveNetwork(nid string) error { } n.ltime = ltime + n.reapTime = reapInterval n.leaving = true return nil } @@ -580,7 +582,10 @@ func (nDB *NetworkDB) addNetworkNode(nid string, nodeName string) { // passed network. Caller should hold the NetworkDB lock while calling // this func (nDB *NetworkDB) deleteNetworkNode(nid string, nodeName string) { - nodes := nDB.networkNodes[nid] + 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 { @@ -618,27 +623,3 @@ func (nDB *NetworkDB) updateLocalNetworkTime() { 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 - }) -} diff --git a/components/engine/vendor/github.com/docker/libnetwork/service_linux.go b/components/engine/vendor/github.com/docker/libnetwork/service_linux.go index 7284e9a00e..ec002d6a1a 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/service_linux.go +++ b/components/engine/vendor/github.com/docker/libnetwork/service_linux.go @@ -372,6 +372,7 @@ func programIngress(gwIP net.IP, ingressPorts []*PortConfig, isDelete bool) erro if err := iptables.RawCombinedOutput("-I", "FORWARD", "-j", ingressChain); err != nil { return fmt.Errorf("failed to add jump rule to %s in filter table forward chain: %v", ingressChain, err) } + arrangeUserFilterRule() } oifName, err := findOIFName(gwIP) @@ -438,7 +439,9 @@ func programIngress(gwIP net.IP, ingressPorts []*PortConfig, isDelete bool) erro return nil } -// In the filter table FORWARD chain first rule should be to jump to INGRESS-CHAIN +// In the filter table FORWARD chain the first rule should be to jump to +// DOCKER-USER so the user is able to filter packet first. +// The second rule should be jump to INGRESS-CHAIN. // This chain has the rules to allow access to the published ports for swarm tasks // from local bridge networks and docker_gwbridge (ie:taks on other swarm netwroks) func arrangeIngressFilterRule() {