From 2dcbbd9e01249a251382dc073744add861601ffb Mon Sep 17 00:00:00 2001 From: Flavio Crisciani Date: Thu, 13 Jul 2017 16:54:24 -0700 Subject: [PATCH 1/2] Libnetwork vendoring Signed-off-by: Flavio Crisciani Upstream-commit: 68d8198ed07acde9146a3e8e2021bd9d57250d81 Component: engine --- components/engine/vendor.conf | 4 +- .../github.com/docker/libnetwork/agent.go | 31 +++++---- .../docker/libnetwork/config/config.go | 34 ++++++---- .../libnetwork/drivers/overlay/ov_network.go | 65 +++++++++++++++---- .../libnetwork/drivers/windows/labels.go | 3 + .../libnetwork/drivers/windows/windows.go | 14 ++++ .../docker/libnetwork/drivers_windows.go | 1 + .../docker/libnetwork/endpoint_info.go | 4 +- .../docker/libnetwork/iptables/iptables.go | 8 +-- .../github.com/docker/libnetwork/network.go | 23 ++----- .../docker/libnetwork/network_windows.go | 2 +- .../docker/libnetwork/networkdb/cluster.go | 40 +++++++++--- .../docker/libnetwork/networkdb/delegate.go | 12 +++- .../docker/libnetwork/networkdb/message.go | 4 -- .../docker/libnetwork/networkdb/networkdb.go | 54 +++++++++++++-- .../github.com/docker/libnetwork/resolver.go | 2 +- .../github.com/docker/libnetwork/sandbox.go | 16 ++--- .../libnetwork/sandbox_externalkey_unix.go | 1 - .../docker/libnetwork/sandbox_store.go | 4 +- .../docker/libnetwork/service_linux.go | 5 +- 20 files changed, 225 insertions(+), 102 deletions(-) diff --git a/components/engine/vendor.conf b/components/engine/vendor.conf index 46934bae28..98fbc9f1f9 100644 --- a/components/engine/vendor.conf +++ b/components/engine/vendor.conf @@ -26,7 +26,7 @@ github.com/imdario/mergo 0.2.1 golang.org/x/sync de49d9dcd27d4f764488181bea099dfe6179bcf0 #get libnetwork packages -github.com/docker/libnetwork 6426d1e66f33c0b0c8bb135b7ee547447f54d043 +github.com/docker/libnetwork e23c06b2917c82f6eed18c368f515060af78a09f github.com/docker/go-events 18b43f1bc85d9cdd42c05a6cd2d444c7a200a894 github.com/armon/go-radix e39d623f12e8e41c7b5529e9a9dd67a1e2261f80 github.com/armon/go-metrics eb0af217e5e9747e41dd5303755356b62d28e3ec @@ -143,4 +143,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/agent.go b/components/engine/vendor/github.com/docker/libnetwork/agent.go index a45a569500..4877df1c34 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/agent.go +++ b/components/engine/vendor/github.com/docker/libnetwork/agent.go @@ -214,8 +214,8 @@ func (c *controller) agentSetup(clusterProvider cluster.Provider) error { listen := clusterProvider.GetListenAddress() listenAddr, _, _ := net.SplitHostPort(listen) - logrus.Infof("Initializing Libnetwork Agent Listen-Addr=%s Local-addr=%s Adv-addr=%s Data-addr=%s Remote-addr-list=%v", - listenAddr, bindAddr, advAddr, dataAddr, remoteAddrList) + logrus.Infof("Initializing Libnetwork Agent Listen-Addr=%s Local-addr=%s Adv-addr=%s Data-addr=%s Remote-addr-list=%v MTU=%d", + listenAddr, bindAddr, advAddr, dataAddr, remoteAddrList, c.Config().Daemon.NetworkControlPlaneMTU) if advAddr != "" && agent == nil { if err := c.agentInit(listenAddr, bindAddr, advAddr, dataAddr); err != nil { logrus.Errorf("error in agentInit: %v", err) @@ -286,12 +286,19 @@ func (c *controller) agentInit(listenAddr, bindAddrOrInterface, advertiseAddr, d nodeName := hostname + "-" + stringid.TruncateID(stringid.GenerateRandomID()) logrus.Info("Gossip cluster hostname ", nodeName) - nDB, err := networkdb.New(&networkdb.Config{ - BindAddr: listenAddr, - AdvertiseAddr: advertiseAddr, - NodeName: nodeName, - Keys: keys, - }) + netDBConf := networkdb.DefaultConfig() + netDBConf.NodeName = nodeName + netDBConf.BindAddr = listenAddr + netDBConf.AdvertiseAddr = advertiseAddr + netDBConf.Keys = keys + if c.Config().Daemon.NetworkControlPlaneMTU != 0 { + // Consider the MTU remove the IP hdr (IPv4 or IPv6) and the TCP/UDP hdr. + // To be on the safe side let's cut 100 bytes + netDBConf.PacketBufferSize = (c.Config().Daemon.NetworkControlPlaneMTU - 100) + logrus.Debugf("Control plane MTU: %d will initialize NetworkDB with: %d", + c.Config().Daemon.NetworkControlPlaneMTU, netDBConf.PacketBufferSize) + } + nDB, err := networkdb.New(netDBConf) if err != nil { return err @@ -383,15 +390,11 @@ func (c *controller) agentClose() { agent.Lock() for _, cancelFuncs := range agent.driverCancelFuncs { - for _, cancel := range cancelFuncs { - cancelList = append(cancelList, cancel) - } + cancelList = append(cancelList, cancelFuncs...) } // Add also the cancel functions for the network db - for _, cancel := range agent.coreCancelFuncs { - cancelList = append(cancelList, cancel) - } + cancelList = append(cancelList, agent.coreCancelFuncs...) agent.Unlock() for _, cancel := range cancelList { diff --git a/components/engine/vendor/github.com/docker/libnetwork/config/config.go b/components/engine/vendor/github.com/docker/libnetwork/config/config.go index 3acb4320c4..a2e43e3a53 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/config/config.go +++ b/components/engine/vendor/github.com/docker/libnetwork/config/config.go @@ -26,14 +26,15 @@ type Config struct { // DaemonCfg represents libnetwork core configuration type DaemonCfg struct { - Debug bool - Experimental bool - DataDir string - DefaultNetwork string - DefaultDriver string - Labels []string - DriverCfg map[string]interface{} - ClusterProvider cluster.Provider + Debug bool + Experimental bool + DataDir string + DefaultNetwork string + DefaultDriver string + Labels []string + DriverCfg map[string]interface{} + ClusterProvider cluster.Provider + NetworkControlPlaneMTU int } // ClusterCfg represents cluster configuration @@ -221,6 +222,18 @@ func OptionExperimental(exp bool) Option { } } +// OptionNetworkControlPlaneMTU function returns an option setter for control plane MTU +func OptionNetworkControlPlaneMTU(exp int) Option { + return func(c *Config) { + logrus.Debugf("Network Control Plane MTU: %d", exp) + if exp < 1500 { + // if exp == 0 the value won't be used + logrus.Warnf("Received a MTU of %d, this value is very low, the network control plane can misbehave", exp) + } + c.Daemon.NetworkControlPlaneMTU = exp + } +} + // ProcessOptions processes options and stores it in config func (c *Config) ProcessOptions(options ...Option) { for _, opt := range options { @@ -232,10 +245,7 @@ func (c *Config) ProcessOptions(options ...Option) { // IsValidName validates configuration objects supported by libnetwork func IsValidName(name string) bool { - if strings.TrimSpace(name) == "" { - return false - } - return true + return strings.TrimSpace(name) != "" } // OptionLocalKVProvider function returns an option setter for kvstore provider 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 3a4ea41bfc..01f6287bed 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 @@ -12,6 +12,7 @@ import ( "strings" "sync" "syscall" + "time" "github.com/Sirupsen/logrus" "github.com/docker/docker/pkg/reexec" @@ -705,6 +706,7 @@ func (n *network) initSandbox(restore bool) error { } func (n *network) watchMiss(nlSock *nl.NetlinkSocket) { + t := time.Now() for { msgs, err := nlSock.Receive() if err != nil { @@ -757,23 +759,55 @@ func (n *network) watchMiss(nlSock *nl.NetlinkSocket) { continue } - if !n.driver.isSerfAlive() { - continue - } + if n.driver.isSerfAlive() { + mac, IPmask, vtep, err := n.driver.resolvePeer(n.id, ip) + if err != nil { + logrus.Errorf("could not resolve peer %q: %v", ip, err) + continue + } - mac, IPmask, vtep, err := n.driver.resolvePeer(n.id, ip) - if err != nil { - logrus.Errorf("could not resolve peer %q: %v", ip, err) - 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) + 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) + } + } else { + // If the gc_thresh values are lower kernel might knock off the neighor entries. + // When we get a L3 miss check if its a valid peer and reprogram the neighbor + // entry again. Rate limit it to once attempt every 500ms, just in case a faulty + // container sends a flood of packets to invalid peers + if !l3Miss { + continue + } + if time.Since(t) > 500*time.Millisecond { + t = time.Now() + n.programNeighbor(ip) + } } } } } +func (n *network) programNeighbor(ip net.IP) { + peerMac, _, _, err := n.driver.peerDbSearch(n.id, ip) + if err != nil { + logrus.Errorf("Reprogramming on L3 miss failed for %s, no peer entry", ip) + return + } + s := n.getSubnetforIPAddr(ip) + if s == nil { + logrus.Errorf("Reprogramming on L3 miss failed for %s, not a valid subnet", ip) + return + } + sbox := n.sandbox() + if sbox == nil { + logrus.Errorf("Reprogramming on L3 miss failed for %s, overlay sandbox missing", ip) + return + } + if err := sbox.AddNeighbor(ip, peerMac, true, sbox.NeighborOptions().LinkName(s.vxlanName)); err != nil { + logrus.Errorf("Reprogramming on L3 miss failed for %s: %v", ip, err) + return + } +} + func (d *driver) addNetwork(n *network) { d.Lock() d.networks[n.id] = n @@ -1052,6 +1086,15 @@ func (n *network) contains(ip net.IP) bool { return false } +func (n *network) getSubnetforIPAddr(ip net.IP) *subnet { + for _, s := range n.subnets { + if s.subnetIP.Contains(ip) { + return s + } + } + return nil +} + // getSubnetforIP returns the subnet to which the given IP belongs func (n *network) getSubnetforIP(ip *net.IPNet) *subnet { for _, s := range n.subnets { diff --git a/components/engine/vendor/github.com/docker/libnetwork/drivers/windows/labels.go b/components/engine/vendor/github.com/docker/libnetwork/drivers/windows/labels.go index b32c6ffb61..6cb077cb4f 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/drivers/windows/labels.go +++ b/components/engine/vendor/github.com/docker/libnetwork/drivers/windows/labels.go @@ -28,6 +28,9 @@ const ( // DNSServers of the network DNSServers = "com.docker.network.windowsshim.dnsservers" + // MacPool of the network + MacPool = "com.docker.network.windowsshim.macpool" + // SourceMac of the network SourceMac = "com.docker.network.windowsshim.sourcemac" diff --git a/components/engine/vendor/github.com/docker/libnetwork/drivers/windows/windows.go b/components/engine/vendor/github.com/docker/libnetwork/drivers/windows/windows.go index eb1522a74a..19b2e685b4 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/drivers/windows/windows.go +++ b/components/engine/vendor/github.com/docker/libnetwork/drivers/windows/windows.go @@ -38,6 +38,7 @@ type networkConfiguration struct { VLAN uint VSID uint DNSServers string + MacPools []hcsshim.MacPool DNSSuffix string SourceMac string NetworkAdapterName string @@ -168,6 +169,18 @@ func (d *driver) parseNetworkOptions(id string, genericOptions map[string]string config.DNSSuffix = value case DNSServers: config.DNSServers = value + case MacPool: + config.MacPools = make([]hcsshim.MacPool, 0) + s := strings.Split(value, ",") + if len(s)%2 != 0 { + return nil, types.BadRequestErrorf("Invalid mac pool. You must specify both a start range and an end range") + } + for i := 0; i < len(s)-1; i += 2 { + config.MacPools = append(config.MacPools, hcsshim.MacPool{ + StartMacAddress: s[i], + EndMacAddress: s[i+1], + }) + } case VLAN: vlan, err := strconv.ParseUint(value, 10, 32) if err != nil { @@ -274,6 +287,7 @@ func (d *driver) CreateNetwork(id string, option map[string]interface{}, nInfo d Subnets: subnets, DNSServerList: config.DNSServers, DNSSuffix: config.DNSSuffix, + MacPools: config.MacPools, SourceMac: config.SourceMac, NetworkAdapterName: config.NetworkAdapterName, } diff --git a/components/engine/vendor/github.com/docker/libnetwork/drivers_windows.go b/components/engine/vendor/github.com/docker/libnetwork/drivers_windows.go index 384d855cb8..a037c16efb 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/drivers_windows.go +++ b/components/engine/vendor/github.com/docker/libnetwork/drivers_windows.go @@ -16,5 +16,6 @@ func getInitializers(experimental bool) []initializer { {windows.GetInit("l2bridge"), "l2bridge"}, {windows.GetInit("l2tunnel"), "l2tunnel"}, {windows.GetInit("nat"), "nat"}, + {windows.GetInit("ics"), "ics"}, } } diff --git a/components/engine/vendor/github.com/docker/libnetwork/endpoint_info.go b/components/engine/vendor/github.com/docker/libnetwork/endpoint_info.go index 202c27b308..17e93b9765 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/endpoint_info.go +++ b/components/engine/vendor/github.com/docker/libnetwork/endpoint_info.go @@ -154,9 +154,7 @@ func (epi *endpointInterface) CopyTo(dstEpi *endpointInterface) error { dstEpi.v6PoolID = epi.v6PoolID if len(epi.llAddrs) != 0 { dstEpi.llAddrs = make([]*net.IPNet, 0, len(epi.llAddrs)) - for _, ll := range epi.llAddrs { - dstEpi.llAddrs = append(dstEpi.llAddrs, ll) - } + dstEpi.llAddrs = append(dstEpi.llAddrs, epi.llAddrs...) } for _, route := range epi.routes { 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/network.go b/components/engine/vendor/github.com/docker/libnetwork/network.go index 9f99064e11..72deeea660 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/network.go +++ b/components/engine/vendor/github.com/docker/libnetwork/network.go @@ -434,15 +434,11 @@ func (n *network) applyConfigurationTo(to *network) error { } if len(n.ipamV4Config) > 0 { to.ipamV4Config = make([]*IpamConf, 0, len(n.ipamV4Config)) - for _, v4conf := range n.ipamV4Config { - to.ipamV4Config = append(to.ipamV4Config, v4conf) - } + to.ipamV4Config = append(to.ipamV4Config, n.ipamV4Config...) } if len(n.ipamV6Config) > 0 { to.ipamV6Config = make([]*IpamConf, 0, len(n.ipamV6Config)) - for _, v6conf := range n.ipamV6Config { - to.ipamV6Config = append(to.ipamV6Config, v6conf) - } + to.ipamV6Config = append(to.ipamV6Config, n.ipamV6Config...) } if len(n.generic) > 0 { to.generic = options.Generic{} @@ -873,8 +869,7 @@ func (n *network) resolveDriver(name string, load bool) (driverapi.Driver, *driv d, cap := c.drvRegistry.Driver(name) if d == nil { if load { - var err error - err = c.loadDriver(name) + err := c.loadDriver(name) if err != nil { return nil, nil, err } @@ -1451,11 +1446,7 @@ func (n *network) ipamAllocate() error { } err = n.ipamAllocateVersion(6, ipam) - if err != nil { - return err - } - - return nil + return err } func (n *network) requestPoolHelper(ipam ipamapi.Ipam, addressSpace, preferredPool, subPool string, options map[string]string, v6 bool) (string, *net.IPNet, map[string]string, error) { @@ -1654,9 +1645,7 @@ func (n *network) getIPInfo(ipVer int) []*IpamInfo { } l := make([]*IpamInfo, 0, len(info)) n.Lock() - for _, d := range info { - l = append(l, d) - } + l = append(l, info...) n.Unlock() return l } @@ -1870,7 +1859,7 @@ func (n *network) ResolveName(req string, ipType int) ([]net.IP, bool) { // the docker network domain. If the network is not v6 enabled // set ipv6Miss to filter the DNS query from going to external // resolvers. - if ok && n.enableIPv6 == false { + if ok && !n.enableIPv6 { ipv6Miss = true } ipSet, ok = sr.svcIPv6Map.Get(req) diff --git a/components/engine/vendor/github.com/docker/libnetwork/network_windows.go b/components/engine/vendor/github.com/docker/libnetwork/network_windows.go index 07a1c1d424..ddcd3345f2 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/network_windows.go +++ b/components/engine/vendor/github.com/docker/libnetwork/network_windows.go @@ -29,7 +29,7 @@ func executeInCompartment(compartmentID uint32, x func()) { func (n *network) startResolver() { n.resolverOnce.Do(func() { - logrus.Debugf("Launching DNS server for network", n.Name()) + logrus.Debugf("Launching DNS server for network %q", n.Name()) options := n.Info().DriverOptions() hnsid := options[windows.HNSID] 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..e011a5336e 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/networkdb/cluster.go +++ b/components/engine/vendor/github.com/docker/libnetwork/networkdb/cluster.go @@ -98,10 +98,14 @@ func (nDB *NetworkDB) RemoveKey(key []byte) { } func (nDB *NetworkDB) clusterInit() error { + nDB.lastStatsTimestamp = time.Now() + nDB.lastHealthTimestamp = nDB.lastStatsTimestamp + config := memberlist.DefaultLANConfig() config.Name = nDB.config.NodeName 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 @@ -199,9 +203,8 @@ func (nDB *NetworkDB) clusterJoin(members []string) error { mlist := nDB.memberlist if _, err := mlist.Join(members); err != nil { - // Incase of failure, keep retrying join until it succeeds or the cluster is shutdown. + // In case of failure, keep retrying join until it succeeds or the cluster is shutdown. go nDB.retryJoin(members, nDB.stopCh) - return fmt.Errorf("could not join node to memberlist: %v", err) } @@ -310,12 +313,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 @@ -373,11 +375,21 @@ func (nDB *NetworkDB) gossip() { 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 - healthscore:%d (connectivity issues)", healthScore) + } + nDB.lastHealthTimestamp = time.Now() + } + for nid, nodes := range networkNodes { mNodes := nDB.mRandomNodes(3, nodes) - bytesAvail := udpSendBuf - compoundHeaderOverhead + bytesAvail := nDB.config.PacketBufferSize - compoundHeaderOverhead nDB.RLock() network, ok := thisNodeNetworks[nid] @@ -398,6 +410,14 @@ func (nDB *NetworkDB) gossip() { } 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 - Queue net:%s qLen:%d netPeers:%d netMsg/s:%d", + nid, broadcastQ.NumQueued(), broadcastQ.NumNodes(), network.qMessagesSent/int((nDB.config.StatsPrintPeriod/time.Second))) + network.qMessagesSent = 0 + } + if len(msgs) == 0 { continue } @@ -415,11 +435,15 @@ func (nDB *NetworkDB) gossip() { } // Send the compound message - if err := nDB.memberlist.SendToUDP(&mnode.Node, compound); err != nil { + 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() { @@ -590,7 +614,7 @@ func (nDB *NetworkDB) bulkSyncNode(networks []string, node string, unsolicited b nDB.bulkSyncAckTbl[node] = ch nDB.Unlock() - err = nDB.memberlist.SendToTCP(&mnode.Node, buf) + err = nDB.memberlist.SendReliable(&mnode.Node, buf) if err != nil { nDB.Lock() delete(nDB.bulkSyncAckTbl, node) @@ -607,7 +631,7 @@ func (nDB *NetworkDB) bulkSyncNode(networks []string, node string, unsolicited b case <-t.C: logrus.Errorf("Bulk sync to node %s timed out", node) case <-ch: - logrus.Debugf("%s: Bulk sync to node %s took %s", nDB.config.NodeName, node, time.Now().Sub(startTime)) + logrus.Debugf("%s: Bulk sync to node %s took %s", nDB.config.NodeName, node, time.Since(startTime)) } t.Stop() } 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..b3ef000d7e 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/networkdb/delegate.go +++ b/components/engine/vendor/github.com/docker/libnetwork/networkdb/delegate.go @@ -111,9 +111,12 @@ func (nDB *NetworkDB) handleNodeEvent(nEvent *NodeEvent) bool { switch nEvent.Type { case NodeEventTypeJoin: nDB.Lock() + _, found := nDB.nodes[n.Name] nDB.nodes[n.Name] = n nDB.Unlock() - logrus.Infof("Node join event for %s/%s", n.Name, n.Addr) + if !found { + logrus.Infof("Node join event for %s/%s", n.Name, n.Addr) + } return true case NodeEventTypeLeave: nDB.Lock() @@ -176,7 +179,12 @@ func (nDB *NetworkDB) handleNetworkEvent(nEvent *NetworkEvent) bool { flushEntries = true } - nDB.addNetworkNode(nEvent.NetworkID, nEvent.NodeName) + if nEvent.Type == NetworkEventTypeLeave { + nDB.deleteNetworkNode(nEvent.NetworkID, nEvent.NodeName) + } else { + nDB.addNetworkNode(nEvent.NetworkID, nEvent.NodeName) + } + return true } diff --git a/components/engine/vendor/github.com/docker/libnetwork/networkdb/message.go b/components/engine/vendor/github.com/docker/libnetwork/networkdb/message.go index a861752bd4..81a6d832a6 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/networkdb/message.go +++ b/components/engine/vendor/github.com/docker/libnetwork/networkdb/message.go @@ -3,10 +3,6 @@ package networkdb import "github.com/gogo/protobuf/proto" const ( - // Max udp message size chosen to avoid network packet - // fragmentation. - udpSendBuf = 1400 - // Compound message header overhead 1 byte(message type) + 4 // bytes (num messages) compoundHeaderOverhead = 5 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..e4cc9df12c 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/networkdb/networkdb.go +++ b/components/engine/vendor/github.com/docker/libnetwork/networkdb/networkdb.go @@ -1,10 +1,11 @@ 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 +//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 ( "fmt" "net" + "os" "strings" "sync" "time" @@ -93,6 +94,12 @@ type NetworkDB struct { // 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 @@ -126,6 +133,9 @@ type network struct { // 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 } // Config represents the configuration of the networdb instance and @@ -149,6 +159,21 @@ type Config struct { // 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 + + // 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 @@ -171,6 +196,18 @@ type entry struct { reapTime time.Duration } +// DefaultConfig returns a NetworkDB config with default values +func DefaultConfig() *Config { + hostname, _ := os.Hostname() + return &Config{ + NodeName: hostname, + BindAddr: "0.0.0.0", + PacketBufferSize: 1400, + StatsPrintPeriod: 5 * time.Minute, + HealthPrintPeriod: 1 * time.Minute, + } +} + // New creates a new instance of NetworkDB using the Config passed by // the caller. func New(c *Config) (*NetworkDB, error) { @@ -200,6 +237,7 @@ func New(c *Config) (*NetworkDB, error) { // 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)) } @@ -481,13 +519,12 @@ func (nDB *NetworkDB) JoinNetwork(nid string) error { nodeNetworks[nid].tableBroadcasts = &memberlist.TransmitLimitedQueue{ NumNodes: func() int { nDB.RLock() - num := len(nDB.networkNodes[nid]) - nDB.RUnlock() - return num + defer nDB.RUnlock() + return len(nDB.networkNodes[nid]) }, RetransmitMult: 4, } - nDB.networkNodes[nid] = append(nDB.networkNodes[nid], nDB.config.NodeName) + nDB.addNetworkNode(nid, nDB.config.NodeName) networkNodes := nDB.networkNodes[nid] nDB.Unlock() @@ -522,6 +559,8 @@ func (nDB *NetworkDB) LeaveNetwork(nid string) error { entries []*entry ) + nDB.deleteNetworkNode(nid, nDB.config.NodeName) + nwWalker := func(path string, v interface{}) bool { entry, ok := v.(*entry) if !ok { @@ -580,7 +619,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 { diff --git a/components/engine/vendor/github.com/docker/libnetwork/resolver.go b/components/engine/vendor/github.com/docker/libnetwork/resolver.go index cc7692621f..ff472d0019 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/resolver.go +++ b/components/engine/vendor/github.com/docker/libnetwork/resolver.go @@ -446,7 +446,7 @@ func (r *resolver) ServeDNS(w dns.ResponseWriter, query *dns.Msg) { defer co.Close() // limits the number of outstanding concurrent queries. - if r.forwardQueryStart() == false { + if !r.forwardQueryStart() { old := r.tStamp r.tStamp = time.Now() if r.tStamp.Sub(old) > logInterval { diff --git a/components/engine/vendor/github.com/docker/libnetwork/sandbox.go b/components/engine/vendor/github.com/docker/libnetwork/sandbox.go index 9454c5c286..46674067a1 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/sandbox.go +++ b/components/engine/vendor/github.com/docker/libnetwork/sandbox.go @@ -621,7 +621,7 @@ func (sb *sandbox) resolveName(req string, networkName string, epList []*endpoin func (sb *sandbox) SetKey(basePath string) error { start := time.Now() defer func() { - logrus.Debugf("sandbox set key processing took %s for container %s", time.Now().Sub(start), sb.ContainerID()) + logrus.Debugf("sandbox set key processing took %s for container %s", time.Since(start), sb.ContainerID()) }() if basePath == "" { @@ -773,9 +773,7 @@ func (sb *sandbox) restoreOslSandbox() error { } Ifaces[fmt.Sprintf("%s+%s", i.srcName, i.dstPrefix)] = ifaceOptions if joinInfo != nil { - for _, r := range joinInfo.StaticRoutes { - routes = append(routes, r) - } + routes = append(routes, joinInfo.StaticRoutes...) } if ep.needResolver() { sb.startResolver(true) @@ -789,11 +787,7 @@ func (sb *sandbox) restoreOslSandbox() error { // restore osl sandbox err := sb.osSbox.Restore(Ifaces, routes, gwep.joinInfo.gw, gwep.joinInfo.gw6) - if err != nil { - return err - } - - return nil + return err } func (sb *sandbox) populateNetworkResources(ep *endpoint) error { @@ -958,9 +952,7 @@ func (sb *sandbox) joinLeaveStart() { joinLeaveDone := sb.joinLeaveDone sb.Unlock() - select { - case <-joinLeaveDone: - } + <-joinLeaveDone sb.Lock() } diff --git a/components/engine/vendor/github.com/docker/libnetwork/sandbox_externalkey_unix.go b/components/engine/vendor/github.com/docker/libnetwork/sandbox_externalkey_unix.go index 4a7ac06b59..c33398f953 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/sandbox_externalkey_unix.go +++ b/components/engine/vendor/github.com/docker/libnetwork/sandbox_externalkey_unix.go @@ -52,7 +52,6 @@ func processSetKeyReexec() { controllerID := os.Args[2] err = SetExternalKey(controllerID, containerID, fmt.Sprintf("/proc/%d/ns/net", state.Pid)) - return } // SetExternalKey provides a convenient way to set an External key to a sandbox diff --git a/components/engine/vendor/github.com/docker/libnetwork/sandbox_store.go b/components/engine/vendor/github.com/docker/libnetwork/sandbox_store.go index 38b2bd7e8b..b92a544dc9 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/sandbox_store.go +++ b/components/engine/vendor/github.com/docker/libnetwork/sandbox_store.go @@ -115,9 +115,7 @@ func (sbs *sbState) CopyTo(o datastore.KVObject) error { dstSbs.dbExists = sbs.dbExists dstSbs.EpPriority = sbs.EpPriority - for _, eps := range sbs.Eps { - dstSbs.Eps = append(dstSbs.Eps, eps) - } + dstSbs.Eps = append(dstSbs.Eps, sbs.Eps...) if len(sbs.ExtDNS2) > 0 { for _, dns := range sbs.ExtDNS2 { 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 1cf7ee91aa..784c1784ec 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() { From c202b3f97f8d6435cfd420556302aaf252b8cd49 Mon Sep 17 00:00:00 2001 From: Flavio Crisciani Date: Fri, 28 Jul 2017 13:18:49 -0700 Subject: [PATCH 2/2] Allow to set the control plane MTU Add daemon config to allow the user to specify the MTU of the control plane network. The first user of this new parameter is actually libnetwork that can seed the gossip with the proper MTU value allowing to pack multiple messages per UDP packet sent. If the value is not specified or is lower than 1500 the logic will set it to the default. Signed-off-by: Flavio Crisciani Upstream-commit: f9f25ca5e44c89d7c1ebdfa9865076eb2cde9bb2 Component: engine --- components/engine/cmd/dockerd/config.go | 1 + components/engine/daemon/config/config.go | 2 ++ components/engine/daemon/daemon.go | 2 ++ 3 files changed, 5 insertions(+) diff --git a/components/engine/cmd/dockerd/config.go b/components/engine/cmd/dockerd/config.go index 11084ec8d6..fa5565ee46 100644 --- a/components/engine/cmd/dockerd/config.go +++ b/components/engine/cmd/dockerd/config.go @@ -63,6 +63,7 @@ func installCommonConfigFlags(conf *config.Config, flags *pflag.FlagSet) { flags.StringVar(&conf.MetricsAddress, "metrics-addr", "", "Set default address and port to serve the metrics api on") flags.StringVar(&conf.NodeGenericResources, "node-generic-resources", "", "user defined resources (e.g. fpga=2;gpu={UUID1,UUID2,UUID3})") + flags.IntVar(&conf.NetworkControlPlaneMTU, "network-control-plane-mtu", config.DefaultNetworkMtu, "Network Control plane MTU") // "--deprecated-key-path" is to allow configuration of the key used // for the daemon ID and the deprecated image signing. It was never diff --git a/components/engine/daemon/config/config.go b/components/engine/daemon/config/config.go index 86a1356c57..5adb7e69e6 100644 --- a/components/engine/daemon/config/config.go +++ b/components/engine/daemon/config/config.go @@ -171,6 +171,8 @@ type CommonConfig struct { // Exposed node Generic Resources NodeGenericResources string `json:"node-generic-resources,omitempty"` + // NetworkControlPlaneMTU allows to specify the control plane MTU, this will allow to optimize the network use in some components + NetworkControlPlaneMTU int `json:"network-control-plane-mtu,omitempty"` } // IsValueSet returns true if a configuration value diff --git a/components/engine/daemon/daemon.go b/components/engine/daemon/daemon.go index e62ef08b69..4d4b98c8de 100644 --- a/components/engine/daemon/daemon.go +++ b/components/engine/daemon/daemon.go @@ -1149,6 +1149,8 @@ func (daemon *Daemon) networkOptions(dconfig *config.Config, pg plugingetter.Plu options = append(options, nwconfig.OptionPluginGetter(pg)) } + options = append(options, nwconfig.OptionNetworkControlPlaneMTU(dconfig.NetworkControlPlaneMTU)) + return options, nil }