Merge pull request #328 from andrewhsu/ln
[17.09] bump vndr of libnetwork to 690b4c0
This commit is contained in:
@@ -28,7 +28,7 @@ github.com/imdario/mergo 0.2.1
|
||||
golang.org/x/sync de49d9dcd27d4f764488181bea099dfe6179bcf0
|
||||
|
||||
#get libnetwork packages
|
||||
github.com/docker/libnetwork 5b28c0ec98236c489e39ae6a9e1aeb802e071681
|
||||
github.com/docker/libnetwork 690b4c05d4a30f627f8e128d6e5fcd99431cc892
|
||||
github.com/docker/go-events 9461782956ad83b30282bf90e31fa6a70c255ba9
|
||||
github.com/armon/go-radix e39d623f12e8e41c7b5529e9a9dd67a1e2261f80
|
||||
github.com/armon/go-metrics eb0af217e5e9747e41dd5303755356b62d28e3ec
|
||||
|
||||
+1
@@ -341,6 +341,7 @@ func (c *controller) clusterAgentInit() {
|
||||
// should still be present when cleaning up
|
||||
// service bindings
|
||||
c.agentClose()
|
||||
c.cleanupServiceDiscovery("")
|
||||
c.cleanupServiceBindings("")
|
||||
|
||||
c.agentStopComplete()
|
||||
|
||||
+1
-5
@@ -763,11 +763,7 @@ func (d *driver) createNetwork(config *networkConfiguration) error {
|
||||
|
||||
// Apply the prepared list of steps, and abort at the first error.
|
||||
bridgeSetup.queueStep(setupDeviceUp)
|
||||
if err = bridgeSetup.apply(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
return bridgeSetup.apply()
|
||||
}
|
||||
|
||||
func (d *driver) DeleteNetwork(nid string) error {
|
||||
|
||||
Generated
Vendored
+2
-9
@@ -169,11 +169,7 @@ func setupIPTablesInternal(bridgeIface string, addr net.Addr, icc, ipmasq, hairp
|
||||
}
|
||||
|
||||
// Set Accept on all non-intercontainer outgoing packets.
|
||||
if err := programChainRule(outRule, "ACCEPT NON_ICC OUTGOING", enable); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
return programChainRule(outRule, "ACCEPT NON_ICC OUTGOING", enable)
|
||||
}
|
||||
|
||||
func programChainRule(rule iptRule, ruleDescr string, insert bool) error {
|
||||
@@ -304,10 +300,7 @@ func setupInternalNetworkRules(bridgeIface string, addr net.Addr, icc, insert bo
|
||||
return err
|
||||
}
|
||||
// Set Inter Container Communication.
|
||||
if err := setIcc(bridgeIface, icc, insert); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
return setIcc(bridgeIface, icc, insert)
|
||||
}
|
||||
|
||||
func clearEndpointConnections(nlh *netlink.Handle, ep *bridgeEndpoint) {
|
||||
|
||||
Generated
Vendored
+1
-1
@@ -201,5 +201,5 @@ func delDummyLink(linkName string) error {
|
||||
|
||||
// getDummyName returns the name of a dummy parent with truncated net ID and driver prefix
|
||||
func getDummyName(netID string) string {
|
||||
return fmt.Sprintf("%s%s", dummyPrefix, netID)
|
||||
return dummyPrefix + netID
|
||||
}
|
||||
|
||||
Generated
Vendored
+1
-1
@@ -205,5 +205,5 @@ func delDummyLink(linkName string) error {
|
||||
|
||||
// getDummyName returns the name of a dummy parent with truncated net ID and driver prefix
|
||||
func getDummyName(netID string) string {
|
||||
return fmt.Sprintf("%s%s", dummyPrefix, netID)
|
||||
return dummyPrefix + netID
|
||||
}
|
||||
|
||||
Generated
Vendored
-1
@@ -21,7 +21,6 @@ import (
|
||||
|
||||
const (
|
||||
r = 0xD0C4E3
|
||||
timeout = 30
|
||||
pktExpansion = 26 // SPI(4) + SeqN(4) + IV(8) + PadLength(1) + NextHeader(1) + ICV(8)
|
||||
)
|
||||
|
||||
|
||||
Generated
Vendored
+8
-10
@@ -68,7 +68,7 @@ func (d *driver) Join(nid, eid string, sboxKey string, jinfo driverapi.JoinInfo,
|
||||
|
||||
ep.ifName = containerIfName
|
||||
|
||||
if err := d.writeEndpointToStore(ep); err != nil {
|
||||
if err = d.writeEndpointToStore(ep); err != nil {
|
||||
return fmt.Errorf("failed to update overlay endpoint %s to local data store: %v", ep.id[0:7], err)
|
||||
}
|
||||
|
||||
@@ -86,7 +86,7 @@ func (d *driver) Join(nid, eid string, sboxKey string, jinfo driverapi.JoinInfo,
|
||||
return err
|
||||
}
|
||||
|
||||
if err := sbox.AddInterface(overlayIfName, "veth",
|
||||
if err = sbox.AddInterface(overlayIfName, "veth",
|
||||
sbox.InterfaceOptions().Master(s.brName)); err != nil {
|
||||
return fmt.Errorf("could not add veth pair inside the network sandbox: %v", err)
|
||||
}
|
||||
@@ -100,7 +100,7 @@ func (d *driver) Join(nid, eid string, sboxKey string, jinfo driverapi.JoinInfo,
|
||||
return err
|
||||
}
|
||||
|
||||
if err := nlh.LinkSetHardwareAddr(veth, ep.mac); err != nil {
|
||||
if err = nlh.LinkSetHardwareAddr(veth, ep.mac); err != nil {
|
||||
return fmt.Errorf("could not set mac address (%v) to the container interface: %v", ep.mac, err)
|
||||
}
|
||||
|
||||
@@ -108,7 +108,7 @@ func (d *driver) Join(nid, eid string, sboxKey string, jinfo driverapi.JoinInfo,
|
||||
if sub == s {
|
||||
continue
|
||||
}
|
||||
if err := jinfo.AddStaticRoute(sub.subnetIP, types.NEXTHOP, s.gwIP.IP); err != nil {
|
||||
if err = jinfo.AddStaticRoute(sub.subnetIP, types.NEXTHOP, s.gwIP.IP); err != nil {
|
||||
logrus.Errorf("Adding subnet %s static route in network %q failed\n", s.subnetIP, n.id)
|
||||
}
|
||||
}
|
||||
@@ -122,7 +122,7 @@ func (d *driver) Join(nid, eid string, sboxKey string, jinfo driverapi.JoinInfo,
|
||||
|
||||
d.peerAdd(nid, eid, ep.addr.IP, ep.addr.Mask, ep.mac, net.ParseIP(d.advertiseAddress), false, false, true)
|
||||
|
||||
if err := d.checkEncryption(nid, nil, n.vxlanID(s), true, true); err != nil {
|
||||
if err = d.checkEncryption(nid, nil, n.vxlanID(s), true, true); err != nil {
|
||||
logrus.Warn(err)
|
||||
}
|
||||
|
||||
@@ -200,7 +200,7 @@ func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key stri
|
||||
}
|
||||
|
||||
if etype == driverapi.Delete {
|
||||
d.peerDelete(nid, eid, addr.IP, addr.Mask, mac, vtep)
|
||||
d.peerDelete(nid, eid, addr.IP, addr.Mask, mac, vtep, false)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -232,11 +232,9 @@ func (d *driver) Leave(nid, eid string) error {
|
||||
}
|
||||
}
|
||||
|
||||
n.leaveSandbox()
|
||||
d.peerDelete(nid, eid, ep.addr.IP, ep.addr.Mask, ep.mac, net.ParseIP(d.advertiseAddress), true)
|
||||
|
||||
if err := d.checkEncryption(nid, nil, 0, true, false); err != nil {
|
||||
logrus.Warn(err)
|
||||
}
|
||||
n.leaveSandbox()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
Generated
Vendored
+2
-9
@@ -144,11 +144,7 @@ func (d *driver) deleteEndpointFromStore(e *endpoint) error {
|
||||
return fmt.Errorf("overlay local store not initialized, ep not deleted")
|
||||
}
|
||||
|
||||
if err := d.localStore.DeleteObjectAtomic(e); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
return d.localStore.DeleteObjectAtomic(e)
|
||||
}
|
||||
|
||||
func (d *driver) writeEndpointToStore(e *endpoint) error {
|
||||
@@ -156,10 +152,7 @@ func (d *driver) writeEndpointToStore(e *endpoint) error {
|
||||
return fmt.Errorf("overlay local store not initialized, ep not added")
|
||||
}
|
||||
|
||||
if err := d.localStore.PutObjectAtomic(e); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
return d.localStore.PutObjectAtomic(e)
|
||||
}
|
||||
|
||||
func (ep *endpoint) DataScope() string {
|
||||
|
||||
Generated
Vendored
+19
-51
@@ -251,8 +251,9 @@ func (d *driver) DeleteNetwork(nid string) error {
|
||||
if err := d.deleteEndpointFromStore(ep); err != nil {
|
||||
logrus.Warnf("Failed to delete overlay endpoint %s from local store: %v", ep.id[0:7], err)
|
||||
}
|
||||
|
||||
}
|
||||
// flush the peerDB entries
|
||||
d.peerFlush(nid)
|
||||
d.deleteNetwork(nid)
|
||||
|
||||
vnis, err := n.releaseVxlanID()
|
||||
@@ -494,7 +495,7 @@ func (n *network) restoreSubnetSandbox(s *subnet, brName, vxlanName string) erro
|
||||
brIfaceOption := make([]osl.IfaceOption, 2)
|
||||
brIfaceOption = append(brIfaceOption, sbox.InterfaceOptions().Address(s.gwIP))
|
||||
brIfaceOption = append(brIfaceOption, sbox.InterfaceOptions().Bridge(true))
|
||||
Ifaces[fmt.Sprintf("%s+%s", brName, "br")] = brIfaceOption
|
||||
Ifaces[brName+"+br"] = brIfaceOption
|
||||
|
||||
err := sbox.Restore(Ifaces, nil, nil, nil)
|
||||
if err != nil {
|
||||
@@ -504,12 +505,8 @@ func (n *network) restoreSubnetSandbox(s *subnet, brName, vxlanName string) erro
|
||||
Ifaces = make(map[string][]osl.IfaceOption)
|
||||
vxlanIfaceOption := make([]osl.IfaceOption, 1)
|
||||
vxlanIfaceOption = append(vxlanIfaceOption, sbox.InterfaceOptions().Master(brName))
|
||||
Ifaces[fmt.Sprintf("%s+%s", vxlanName, "vxlan")] = vxlanIfaceOption
|
||||
err = sbox.Restore(Ifaces, nil, nil, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
Ifaces[vxlanName+"+vxlan"] = vxlanIfaceOption
|
||||
return sbox.Restore(Ifaces, nil, nil, nil)
|
||||
}
|
||||
|
||||
func (n *network) setupSubnetSandbox(s *subnet, brName, vxlanName string) error {
|
||||
@@ -760,58 +757,38 @@ func (n *network) watchMiss(nlSock *nl.NetlinkSocket) {
|
||||
continue
|
||||
}
|
||||
|
||||
logrus.Debugf("miss notification: dest IP %v, dest MAC %v", ip, mac)
|
||||
|
||||
if neigh.State&(netlink.NUD_STALE|netlink.NUD_INCOMPLETE) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
if n.driver.isSerfAlive() {
|
||||
logrus.Debugf("miss notification: dest IP %v, dest MAC %v", ip, mac)
|
||||
mac, IPmask, vtep, err := n.driver.resolvePeer(n.id, ip)
|
||||
if err != nil {
|
||||
logrus.Errorf("could not resolve peer %q: %v", ip, err)
|
||||
continue
|
||||
}
|
||||
n.driver.peerAdd(n.id, "dummy", ip, IPmask, mac, vtep, l2Miss, l3Miss, false)
|
||||
} 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 {
|
||||
} else if l3Miss && time.Since(t) > time.Second {
|
||||
// All the local peers will trigger a miss notification but this one is expected and the local container will reply
|
||||
// autonomously to the ARP request
|
||||
// In case the gc_thresh3 values is low kernel might reject new entries during peerAdd. This will trigger the following
|
||||
// extra logs that will inform of the possible issue.
|
||||
// Entries created would not be deleted see documentation http://man7.org/linux/man-pages/man7/arp.7.html:
|
||||
// Entries which are marked as permanent are never deleted by the garbage-collector.
|
||||
// The time limit here is to guarantee that the dbSearch is not
|
||||
// done too frequently causing a stall of the peerDB operations.
|
||||
pKey, pEntry, err := n.driver.peerDbSearch(n.id, ip)
|
||||
if err == nil && !pEntry.isLocal {
|
||||
t = time.Now()
|
||||
n.programNeighbor(ip)
|
||||
logrus.Warnf("miss notification for peer:%+v l3Miss:%t l2Miss:%t, if the problem persist check the gc_thresh on the host pKey:%+v pEntry:%+v err:%v",
|
||||
neigh, l3Miss, l2Miss, *pKey, *pEntry, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
@@ -1090,15 +1067,6 @@ 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 {
|
||||
|
||||
+4
-4
@@ -122,7 +122,7 @@ func (d *driver) processEvent(u serf.UserEvent) {
|
||||
case "join":
|
||||
d.peerAdd(nid, eid, net.ParseIP(ipStr), net.IPMask(net.ParseIP(maskStr).To4()), mac, net.ParseIP(vtepStr), false, false, false)
|
||||
case "leave":
|
||||
d.peerDelete(nid, eid, net.ParseIP(ipStr), net.IPMask(net.ParseIP(maskStr).To4()), mac, net.ParseIP(vtepStr))
|
||||
d.peerDelete(nid, eid, net.ParseIP(ipStr), net.IPMask(net.ParseIP(maskStr).To4()), mac, net.ParseIP(vtepStr), false)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -135,13 +135,13 @@ func (d *driver) processQuery(q *serf.Query) {
|
||||
fmt.Printf("Failed to scan query payload string: %v\n", err)
|
||||
}
|
||||
|
||||
peerMac, peerIPMask, vtep, err := d.peerDbSearch(nid, net.ParseIP(ipStr))
|
||||
pKey, pEntry, err := d.peerDbSearch(nid, net.ParseIP(ipStr))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
logrus.Debugf("Sending peer query resp mac %s, mask %s, vtep %s", peerMac, net.IP(peerIPMask), vtep)
|
||||
q.Respond([]byte(fmt.Sprintf("%s %s %s", peerMac.String(), net.IP(peerIPMask).String(), vtep.String())))
|
||||
logrus.Debugf("Sending peer query resp mac %v, mask %s, vtep %s", pKey.peerMac, net.IP(pEntry.peerIPMask).String(), pEntry.vtep)
|
||||
q.Respond([]byte(fmt.Sprintf("%s %s %s", pKey.peerMac.String(), net.IP(pEntry.peerIPMask).String(), pEntry.vtep.String())))
|
||||
}
|
||||
|
||||
func (d *driver) resolvePeer(nid string, peerIP net.IP) (net.HardwareAddr, net.IPMask, net.IP, error) {
|
||||
|
||||
+3
-3
@@ -162,7 +162,7 @@ func (d *driver) restoreEndpoints() error {
|
||||
Ifaces := make(map[string][]osl.IfaceOption)
|
||||
vethIfaceOption := make([]osl.IfaceOption, 1)
|
||||
vethIfaceOption = append(vethIfaceOption, n.sbox.InterfaceOptions().Master(s.brName))
|
||||
Ifaces[fmt.Sprintf("%s+%s", "veth", "veth")] = vethIfaceOption
|
||||
Ifaces["veth+veth"] = vethIfaceOption
|
||||
|
||||
err := n.sbox.Restore(Ifaces, nil, nil, nil)
|
||||
if err != nil {
|
||||
@@ -262,7 +262,7 @@ func (d *driver) nodeJoin(advertiseAddress, bindAddress string, self bool) {
|
||||
d.Unlock()
|
||||
|
||||
// If containers are already running on this network update the
|
||||
// advertiseaddress in the peerDB
|
||||
// advertise address in the peerDB
|
||||
d.localJoinOnce.Do(func() {
|
||||
d.peerDBUpdateSelf()
|
||||
})
|
||||
@@ -270,7 +270,7 @@ func (d *driver) nodeJoin(advertiseAddress, bindAddress string, self bool) {
|
||||
// If there is no cluster store there is no need to start serf.
|
||||
if d.store != nil {
|
||||
if err := validateSelf(advertiseAddress); err != nil {
|
||||
logrus.Warnf("%s", err.Error())
|
||||
logrus.Warn(err.Error())
|
||||
}
|
||||
err := d.serfInit()
|
||||
if err != nil {
|
||||
|
||||
Generated
Vendored
+1
-1
@@ -24,4 +24,4 @@ message PeerRecord {
|
||||
// which this container is running and can be reached by
|
||||
// building a tunnel to that host IP.
|
||||
string tunnel_endpoint_ip = 3 [(gogoproto.customname) = "TunnelEndpointIP"];
|
||||
}
|
||||
}
|
||||
|
||||
+163
-83
@@ -8,6 +8,7 @@ import (
|
||||
"syscall"
|
||||
|
||||
"github.com/docker/libnetwork/common"
|
||||
"github.com/docker/libnetwork/osl"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
@@ -22,16 +23,48 @@ type peerEntry struct {
|
||||
eid string
|
||||
vtep net.IP
|
||||
peerIPMask net.IPMask
|
||||
inSandbox bool
|
||||
isLocal bool
|
||||
}
|
||||
|
||||
func (p *peerEntry) MarshalDB() peerEntryDB {
|
||||
ones, bits := p.peerIPMask.Size()
|
||||
return peerEntryDB{
|
||||
eid: p.eid,
|
||||
vtep: p.vtep.String(),
|
||||
peerIPMaskOnes: ones,
|
||||
peerIPMaskBits: bits,
|
||||
isLocal: p.isLocal,
|
||||
}
|
||||
}
|
||||
|
||||
// This the structure saved into the set (SetMatrix), due to the implementation of it
|
||||
// the value inserted in the set has to be Hashable so the []byte had to be converted into
|
||||
// strings
|
||||
type peerEntryDB struct {
|
||||
eid string
|
||||
vtep string
|
||||
peerIPMaskOnes int
|
||||
peerIPMaskBits int
|
||||
isLocal bool
|
||||
}
|
||||
|
||||
func (p *peerEntryDB) UnMarshalDB() peerEntry {
|
||||
return peerEntry{
|
||||
eid: p.eid,
|
||||
vtep: net.ParseIP(p.vtep),
|
||||
peerIPMask: net.CIDRMask(p.peerIPMaskOnes, p.peerIPMaskBits),
|
||||
isLocal: p.isLocal,
|
||||
}
|
||||
}
|
||||
|
||||
type peerMap struct {
|
||||
mp map[string]peerEntry
|
||||
// set of peerEntry, note they have to be objects and not pointers to maintain the proper equality checks
|
||||
mp common.SetMatrix
|
||||
sync.Mutex
|
||||
}
|
||||
|
||||
type peerNetworkMap struct {
|
||||
// map with key peerKey
|
||||
mp map[string]*peerMap
|
||||
sync.Mutex
|
||||
}
|
||||
@@ -54,11 +87,7 @@ func (pKey *peerKey) Scan(state fmt.ScanState, verb rune) error {
|
||||
}
|
||||
|
||||
pKey.peerMac, err = net.ParseMAC(string(macB))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
return err
|
||||
}
|
||||
|
||||
func (d *driver) peerDbWalk(f func(string, *peerKey, *peerEntry) bool) error {
|
||||
@@ -87,10 +116,13 @@ func (d *driver) peerDbNetworkWalk(nid string, f func(*peerKey, *peerEntry) bool
|
||||
}
|
||||
|
||||
mp := map[string]peerEntry{}
|
||||
|
||||
pMap.Lock()
|
||||
for pKeyStr, pEntry := range pMap.mp {
|
||||
mp[pKeyStr] = pEntry
|
||||
for _, pKeyStr := range pMap.mp.Keys() {
|
||||
entryDBList, ok := pMap.mp.Get(pKeyStr)
|
||||
if ok {
|
||||
peerEntryDB := entryDBList[0].(peerEntryDB)
|
||||
mp[pKeyStr] = peerEntryDB.UnMarshalDB()
|
||||
}
|
||||
}
|
||||
pMap.Unlock()
|
||||
|
||||
@@ -107,45 +139,38 @@ func (d *driver) peerDbNetworkWalk(nid string, f func(*peerKey, *peerEntry) bool
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *driver) peerDbSearch(nid string, peerIP net.IP) (net.HardwareAddr, net.IPMask, net.IP, error) {
|
||||
var (
|
||||
peerMac net.HardwareAddr
|
||||
vtep net.IP
|
||||
peerIPMask net.IPMask
|
||||
found bool
|
||||
)
|
||||
|
||||
func (d *driver) peerDbSearch(nid string, peerIP net.IP) (*peerKey, *peerEntry, error) {
|
||||
var pKeyMatched *peerKey
|
||||
var pEntryMatched *peerEntry
|
||||
err := d.peerDbNetworkWalk(nid, func(pKey *peerKey, pEntry *peerEntry) bool {
|
||||
if pKey.peerIP.Equal(peerIP) {
|
||||
peerMac = pKey.peerMac
|
||||
peerIPMask = pEntry.peerIPMask
|
||||
vtep = pEntry.vtep
|
||||
found = true
|
||||
return found
|
||||
pKeyMatched = pKey
|
||||
pEntryMatched = pEntry
|
||||
return true
|
||||
}
|
||||
|
||||
return found
|
||||
return false
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return nil, nil, nil, fmt.Errorf("peerdb search for peer ip %q failed: %v", peerIP, err)
|
||||
return nil, nil, fmt.Errorf("peerdb search for peer ip %q failed: %v", peerIP, err)
|
||||
}
|
||||
|
||||
if !found {
|
||||
return nil, nil, nil, fmt.Errorf("peer ip %q not found in peerdb", peerIP)
|
||||
if pKeyMatched == nil || pEntryMatched == nil {
|
||||
return nil, nil, fmt.Errorf("peer ip %q not found in peerdb", peerIP)
|
||||
}
|
||||
|
||||
return peerMac, peerIPMask, vtep, nil
|
||||
return pKeyMatched, pEntryMatched, nil
|
||||
}
|
||||
|
||||
func (d *driver) peerDbAdd(nid, eid string, peerIP net.IP, peerIPMask net.IPMask,
|
||||
peerMac net.HardwareAddr, vtep net.IP, isLocal bool) {
|
||||
peerMac net.HardwareAddr, vtep net.IP, isLocal bool) (bool, int) {
|
||||
|
||||
d.peerDb.Lock()
|
||||
pMap, ok := d.peerDb.mp[nid]
|
||||
if !ok {
|
||||
d.peerDb.mp[nid] = &peerMap{
|
||||
mp: make(map[string]peerEntry),
|
||||
mp: common.NewSetMatrix(),
|
||||
}
|
||||
|
||||
pMap = d.peerDb.mp[nid]
|
||||
@@ -165,18 +190,24 @@ func (d *driver) peerDbAdd(nid, eid string, peerIP net.IP, peerIPMask net.IPMask
|
||||
}
|
||||
|
||||
pMap.Lock()
|
||||
pMap.mp[pKey.String()] = pEntry
|
||||
pMap.Unlock()
|
||||
defer pMap.Unlock()
|
||||
b, i := pMap.mp.Insert(pKey.String(), pEntry.MarshalDB())
|
||||
if i != 1 {
|
||||
// Transient case, there is more than one endpoint that is using the same IP,MAC pair
|
||||
s, _ := pMap.mp.String(pKey.String())
|
||||
logrus.Warnf("peerDbAdd transient condition - Key:%s cardinality:%d db state:%s", pKey.String(), i, s)
|
||||
}
|
||||
return b, i
|
||||
}
|
||||
|
||||
func (d *driver) peerDbDelete(nid, eid string, peerIP net.IP, peerIPMask net.IPMask,
|
||||
peerMac net.HardwareAddr, vtep net.IP) peerEntry {
|
||||
peerMac net.HardwareAddr, vtep net.IP, isLocal bool) (bool, int) {
|
||||
|
||||
d.peerDb.Lock()
|
||||
pMap, ok := d.peerDb.mp[nid]
|
||||
if !ok {
|
||||
d.peerDb.Unlock()
|
||||
return peerEntry{}
|
||||
return false, 0
|
||||
}
|
||||
d.peerDb.Unlock()
|
||||
|
||||
@@ -185,22 +216,22 @@ func (d *driver) peerDbDelete(nid, eid string, peerIP net.IP, peerIPMask net.IPM
|
||||
peerMac: peerMac,
|
||||
}
|
||||
|
||||
pMap.Lock()
|
||||
|
||||
pEntry, ok := pMap.mp[pKey.String()]
|
||||
if ok {
|
||||
// Mismatched endpoint ID(possibly outdated). Do not
|
||||
// delete peerdb
|
||||
if pEntry.eid != eid {
|
||||
pMap.Unlock()
|
||||
return pEntry
|
||||
}
|
||||
pEntry := peerEntry{
|
||||
eid: eid,
|
||||
vtep: vtep,
|
||||
peerIPMask: peerIPMask,
|
||||
isLocal: isLocal,
|
||||
}
|
||||
|
||||
delete(pMap.mp, pKey.String())
|
||||
pMap.Unlock()
|
||||
|
||||
return pEntry
|
||||
pMap.Lock()
|
||||
defer pMap.Unlock()
|
||||
b, i := pMap.mp.Remove(pKey.String(), pEntry.MarshalDB())
|
||||
if i != 0 {
|
||||
// Transient case, there is more than one endpoint that is using the same IP,MAC pair
|
||||
s, _ := pMap.mp.String(pKey.String())
|
||||
logrus.Warnf("peerDbDelete transient condition - Key:%s cardinality:%d db state:%s", pKey.String(), i, s)
|
||||
}
|
||||
return b, i
|
||||
}
|
||||
|
||||
// The overlay uses a lazy initialization approach, this means that when a network is created
|
||||
@@ -224,6 +255,7 @@ const (
|
||||
peerOperationINIT peerOperationType = iota
|
||||
peerOperationADD
|
||||
peerOperationDELETE
|
||||
peerOperationFLUSH
|
||||
)
|
||||
|
||||
type peerOperation struct {
|
||||
@@ -253,7 +285,9 @@ func (d *driver) peerOpRoutine(ctx context.Context, ch chan *peerOperation) {
|
||||
case peerOperationADD:
|
||||
err = d.peerAddOp(op.networkID, op.endpointID, op.peerIP, op.peerIPMask, op.peerMac, op.vtepIP, op.l2Miss, op.l3Miss, true, op.localPeer)
|
||||
case peerOperationDELETE:
|
||||
err = d.peerDeleteOp(op.networkID, op.endpointID, op.peerIP, op.peerIPMask, op.peerMac, op.vtepIP)
|
||||
err = d.peerDeleteOp(op.networkID, op.endpointID, op.peerIP, op.peerIPMask, op.peerMac, op.vtepIP, op.localPeer)
|
||||
case peerOperationFLUSH:
|
||||
err = d.peerFlushOp(op.networkID)
|
||||
}
|
||||
if err != nil {
|
||||
logrus.Warnf("Peer operation failed:%s op:%v", err, op)
|
||||
@@ -286,7 +320,6 @@ func (d *driver) peerInitOp(nid string) error {
|
||||
|
||||
func (d *driver) peerAdd(nid, eid string, peerIP net.IP, peerIPMask net.IPMask,
|
||||
peerMac net.HardwareAddr, vtep net.IP, l2Miss, l3Miss, localPeer bool) {
|
||||
callerName := common.CallerName(1)
|
||||
d.peerOpCh <- &peerOperation{
|
||||
opType: peerOperationADD,
|
||||
networkID: nid,
|
||||
@@ -298,24 +331,32 @@ func (d *driver) peerAdd(nid, eid string, peerIP net.IP, peerIPMask net.IPMask,
|
||||
l2Miss: l2Miss,
|
||||
l3Miss: l3Miss,
|
||||
localPeer: localPeer,
|
||||
callerName: callerName,
|
||||
callerName: common.CallerName(1),
|
||||
}
|
||||
}
|
||||
|
||||
func (d *driver) peerAddOp(nid, eid string, peerIP net.IP, peerIPMask net.IPMask,
|
||||
peerMac net.HardwareAddr, vtep net.IP, l2Miss, l3Miss, updateDB, updateOnlyDB bool) error {
|
||||
peerMac net.HardwareAddr, vtep net.IP, l2Miss, l3Miss, updateDB, localPeer bool) error {
|
||||
|
||||
if err := validateID(nid, eid); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var dbEntries int
|
||||
var inserted bool
|
||||
if updateDB {
|
||||
d.peerDbAdd(nid, eid, peerIP, peerIPMask, peerMac, vtep, false)
|
||||
if updateOnlyDB {
|
||||
return nil
|
||||
inserted, dbEntries = d.peerDbAdd(nid, eid, peerIP, peerIPMask, peerMac, vtep, localPeer)
|
||||
if !inserted {
|
||||
logrus.Warnf("Entry already present in db: nid:%s eid:%s peerIP:%v peerMac:%v isLocal:%t vtep:%v",
|
||||
nid, eid, peerIP, peerMac, localPeer, vtep)
|
||||
}
|
||||
}
|
||||
|
||||
// Local peers do not need any further configuration
|
||||
if localPeer {
|
||||
return nil
|
||||
}
|
||||
|
||||
n := d.network(nid)
|
||||
if n == nil {
|
||||
return nil
|
||||
@@ -353,21 +394,26 @@ func (d *driver) peerAddOp(nid, eid string, peerIP net.IP, peerIPMask net.IPMask
|
||||
|
||||
// Add neighbor entry for the peer IP
|
||||
if err := sbox.AddNeighbor(peerIP, peerMac, l3Miss, sbox.NeighborOptions().LinkName(s.vxlanName)); err != nil {
|
||||
return fmt.Errorf("could not add neighbor entry into the sandbox: %v", err)
|
||||
if _, ok := err.(osl.NeighborSearchError); ok && dbEntries > 1 {
|
||||
// We are in the transient case so only the first configuration is programmed into the kernel
|
||||
// Upon deletion if the active configuration is deleted the next one from the database will be restored
|
||||
// Note we are skipping also the next configuration
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("could not add neighbor entry for nid:%s eid:%s into the sandbox:%v", nid, eid, err)
|
||||
}
|
||||
|
||||
// Add fdb entry to the bridge for the peer mac
|
||||
if err := sbox.AddNeighbor(vtep, peerMac, l2Miss, sbox.NeighborOptions().LinkName(s.vxlanName),
|
||||
sbox.NeighborOptions().Family(syscall.AF_BRIDGE)); err != nil {
|
||||
return fmt.Errorf("could not add fdb entry into the sandbox: %v", err)
|
||||
return fmt.Errorf("could not add fdb entry for nid:%s eid:%s into the sandbox:%v", nid, eid, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *driver) peerDelete(nid, eid string, peerIP net.IP, peerIPMask net.IPMask,
|
||||
peerMac net.HardwareAddr, vtep net.IP) {
|
||||
callerName := common.CallerName(1)
|
||||
peerMac net.HardwareAddr, vtep net.IP, localPeer bool) {
|
||||
d.peerOpCh <- &peerOperation{
|
||||
opType: peerOperationDELETE,
|
||||
networkID: nid,
|
||||
@@ -376,18 +422,23 @@ func (d *driver) peerDelete(nid, eid string, peerIP net.IP, peerIPMask net.IPMas
|
||||
peerIPMask: peerIPMask,
|
||||
peerMac: peerMac,
|
||||
vtepIP: vtep,
|
||||
callerName: callerName,
|
||||
callerName: common.CallerName(1),
|
||||
localPeer: localPeer,
|
||||
}
|
||||
}
|
||||
|
||||
func (d *driver) peerDeleteOp(nid, eid string, peerIP net.IP, peerIPMask net.IPMask,
|
||||
peerMac net.HardwareAddr, vtep net.IP) error {
|
||||
peerMac net.HardwareAddr, vtep net.IP, localPeer bool) error {
|
||||
|
||||
if err := validateID(nid, eid); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
pEntry := d.peerDbDelete(nid, eid, peerIP, peerIPMask, peerMac, vtep)
|
||||
deleted, dbEntries := d.peerDbDelete(nid, eid, peerIP, peerIPMask, peerMac, vtep, localPeer)
|
||||
if !deleted {
|
||||
logrus.Warnf("Entry was not in db: nid:%s eid:%s peerIP:%v peerMac:%v isLocal:%t vtep:%v",
|
||||
nid, eid, peerIP, peerMac, localPeer, vtep)
|
||||
}
|
||||
|
||||
n := d.network(nid)
|
||||
if n == nil {
|
||||
@@ -399,30 +450,59 @@ func (d *driver) peerDeleteOp(nid, eid string, peerIP net.IP, peerIPMask net.IPM
|
||||
return nil
|
||||
}
|
||||
|
||||
// Delete fdb entry to the bridge for the peer mac only if the
|
||||
// entry existed in local peerdb. If it is a stale delete
|
||||
// request, still call DeleteNeighbor but only to cleanup any
|
||||
// leftover sandbox neighbor cache and not actually delete the
|
||||
// kernel state.
|
||||
if (eid == pEntry.eid && vtep.Equal(pEntry.vtep)) ||
|
||||
(eid != pEntry.eid && !vtep.Equal(pEntry.vtep)) {
|
||||
if err := sbox.DeleteNeighbor(vtep, peerMac,
|
||||
eid == pEntry.eid && vtep.Equal(pEntry.vtep)); err != nil {
|
||||
return fmt.Errorf("could not delete fdb entry into the sandbox: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Delete neighbor entry for the peer IP
|
||||
if eid == pEntry.eid {
|
||||
if err := sbox.DeleteNeighbor(peerIP, peerMac, true); err != nil {
|
||||
return fmt.Errorf("could not delete neighbor entry into the sandbox: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := d.checkEncryption(nid, vtep, 0, false, false); err != nil {
|
||||
if err := d.checkEncryption(nid, vtep, 0, localPeer, false); err != nil {
|
||||
logrus.Warn(err)
|
||||
}
|
||||
|
||||
// Local peers do not have any local configuration to delete
|
||||
if !localPeer {
|
||||
// Remove fdb entry to the bridge for the peer mac
|
||||
if err := sbox.DeleteNeighbor(vtep, peerMac, true); err != nil {
|
||||
if _, ok := err.(osl.NeighborSearchError); ok && dbEntries > 0 {
|
||||
// We fall in here if there is a transient state and if the neighbor that is being deleted
|
||||
// was never been configured into the kernel (we allow only 1 configuration at the time per <ip,mac> mapping)
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("could not delete fdb entry for nid:%s eid:%s into the sandbox:%v", nid, eid, err)
|
||||
}
|
||||
|
||||
// Delete neighbor entry for the peer IP
|
||||
if err := sbox.DeleteNeighbor(peerIP, peerMac, true); err != nil {
|
||||
return fmt.Errorf("could not delete neighbor entry for nid:%s eid:%s into the sandbox:%v", nid, eid, err)
|
||||
}
|
||||
}
|
||||
|
||||
if dbEntries == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// If there is still an entry into the database and the deletion went through without errors means that there is now no
|
||||
// configuration active in the kernel.
|
||||
// Restore one configuration for the <ip,mac> directly from the database, note that is guaranteed that there is one
|
||||
peerKey, peerEntry, err := d.peerDbSearch(nid, peerIP)
|
||||
if err != nil {
|
||||
logrus.Errorf("peerDeleteOp unable to restore a configuration for nid:%s ip:%v mac:%v err:%s", nid, peerIP, peerMac, err)
|
||||
return err
|
||||
}
|
||||
return d.peerAddOp(nid, peerEntry.eid, peerIP, peerEntry.peerIPMask, peerKey.peerMac, peerEntry.vtep, false, false, false, peerEntry.isLocal)
|
||||
}
|
||||
|
||||
func (d *driver) peerFlush(nid string) {
|
||||
d.peerOpCh <- &peerOperation{
|
||||
opType: peerOperationFLUSH,
|
||||
networkID: nid,
|
||||
callerName: common.CallerName(1),
|
||||
}
|
||||
}
|
||||
|
||||
func (d *driver) peerFlushOp(nid string) error {
|
||||
d.peerDb.Lock()
|
||||
defer d.peerDb.Unlock()
|
||||
_, ok := d.peerDb.mp[nid]
|
||||
if !ok {
|
||||
return fmt.Errorf("Unable to find the peerDB for nid:%s", nid)
|
||||
}
|
||||
delete(d.peerDb.mp, nid)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
Generated
Vendored
+2
-9
@@ -134,11 +134,7 @@ func (d *driver) deleteEndpointFromStore(e *endpoint) error {
|
||||
return fmt.Errorf("overlay local store not initialized, ep not deleted")
|
||||
}
|
||||
|
||||
if err := d.localStore.DeleteObjectAtomic(e); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
return d.localStore.DeleteObjectAtomic(e)
|
||||
}
|
||||
|
||||
func (d *driver) writeEndpointToStore(e *endpoint) error {
|
||||
@@ -146,10 +142,7 @@ func (d *driver) writeEndpointToStore(e *endpoint) error {
|
||||
return fmt.Errorf("overlay local store not initialized, ep not added")
|
||||
}
|
||||
|
||||
if err := d.localStore.PutObjectAtomic(e); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
return d.localStore.PutObjectAtomic(e)
|
||||
}
|
||||
|
||||
func (ep *endpoint) DataScope() string {
|
||||
|
||||
Generated
Vendored
+2
-2
@@ -331,7 +331,7 @@ func (n *network) restoreSubnetSandbox(s *subnet, brName, vxlanName string) erro
|
||||
brIfaceOption := make([]osl.IfaceOption, 2)
|
||||
brIfaceOption = append(brIfaceOption, sbox.InterfaceOptions().Address(s.gwIP))
|
||||
brIfaceOption = append(brIfaceOption, sbox.InterfaceOptions().Bridge(true))
|
||||
Ifaces[fmt.Sprintf("%s+%s", brName, "br")] = brIfaceOption
|
||||
Ifaces[brName+"+br"] = brIfaceOption
|
||||
|
||||
err := sbox.Restore(Ifaces, nil, nil, nil)
|
||||
if err != nil {
|
||||
@@ -341,7 +341,7 @@ func (n *network) restoreSubnetSandbox(s *subnet, brName, vxlanName string) erro
|
||||
Ifaces = make(map[string][]osl.IfaceOption)
|
||||
vxlanIfaceOption := make([]osl.IfaceOption, 1)
|
||||
vxlanIfaceOption = append(vxlanIfaceOption, sbox.InterfaceOptions().Master(brName))
|
||||
Ifaces[fmt.Sprintf("%s+%s", vxlanName, "vxlan")] = vxlanIfaceOption
|
||||
Ifaces[vxlanName+"+vxlan"] = vxlanIfaceOption
|
||||
err = sbox.Restore(Ifaces, nil, nil, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
Generated
Vendored
+2
-2
@@ -141,7 +141,7 @@ func (d *driver) restoreEndpoints() error {
|
||||
Ifaces := make(map[string][]osl.IfaceOption)
|
||||
vethIfaceOption := make([]osl.IfaceOption, 1)
|
||||
vethIfaceOption = append(vethIfaceOption, n.sbox.InterfaceOptions().Master(s.brName))
|
||||
Ifaces[fmt.Sprintf("%s+%s", "veth", "veth")] = vethIfaceOption
|
||||
Ifaces["veth+veth"] = vethIfaceOption
|
||||
|
||||
err := n.sbox.Restore(Ifaces, nil, nil, nil)
|
||||
if err != nil {
|
||||
@@ -234,7 +234,7 @@ func (d *driver) nodeJoin(advertiseAddress, bindAddress string, self bool) {
|
||||
// If there is no cluster store there is no need to start serf.
|
||||
if d.store != nil {
|
||||
if err := validateSelf(advertiseAddress); err != nil {
|
||||
logrus.Warnf("%s", err.Error())
|
||||
logrus.Warn(err.Error())
|
||||
}
|
||||
err := d.serfInit()
|
||||
if err != nil {
|
||||
|
||||
+1
-5
@@ -199,11 +199,7 @@ func (ep *endpoint) Info() EndpointInfo {
|
||||
return ep
|
||||
}
|
||||
|
||||
if epi := sb.getEndpoint(ep.ID()); epi != nil {
|
||||
return epi
|
||||
}
|
||||
|
||||
return nil
|
||||
return sb.getEndpoint(ep.ID())
|
||||
}
|
||||
|
||||
func (ep *endpoint) Iface() InterfaceInfo {
|
||||
|
||||
+2
-2
@@ -579,7 +579,7 @@ func (a *Allocator) DumpDatabase() string {
|
||||
s = fmt.Sprintf("\n\n%s Config", as)
|
||||
aSpace.Lock()
|
||||
for k, config := range aSpace.subnets {
|
||||
s = fmt.Sprintf("%s%s", s, fmt.Sprintf("\n%v: %v", k, config))
|
||||
s += fmt.Sprintf("\n%v: %v", k, config)
|
||||
if config.Range == nil {
|
||||
a.retrieveBitmask(k, config.Pool)
|
||||
}
|
||||
@@ -589,7 +589,7 @@ func (a *Allocator) DumpDatabase() string {
|
||||
|
||||
s = fmt.Sprintf("%s\n\nBitmasks", s)
|
||||
for k, bm := range a.addresses {
|
||||
s = fmt.Sprintf("%s%s", s, fmt.Sprintf("\n%s: %s", k, bm))
|
||||
s += fmt.Sprintf("\n%s: %s", k, bm)
|
||||
}
|
||||
|
||||
return s
|
||||
|
||||
+3
-10
@@ -276,11 +276,7 @@ func (c *ChainInfo) Forward(action Action, ip net.IP, port int, proto, destAddr
|
||||
"--dport", strconv.Itoa(destPort),
|
||||
"-j", "MASQUERADE",
|
||||
}
|
||||
if err := ProgramRule(Nat, "POSTROUTING", action, args); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
return ProgramRule(Nat, "POSTROUTING", action, args)
|
||||
}
|
||||
|
||||
// Link adds reciprocal ACCEPT rule for two supplied IP addresses.
|
||||
@@ -301,10 +297,7 @@ func (c *ChainInfo) Link(action Action, ip1, ip2 net.IP, port int, proto string,
|
||||
// reverse
|
||||
args[7], args[9] = args[9], args[7]
|
||||
args[10] = "--sport"
|
||||
if err := ProgramRule(Filter, c.Name, action, args); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
return ProgramRule(Filter, c.Name, action, args)
|
||||
}
|
||||
|
||||
// ProgramRule adds the rule specified by args only if the
|
||||
@@ -463,7 +456,7 @@ func RawCombinedOutputNative(args ...string) error {
|
||||
|
||||
// ExistChain checks if a chain exists
|
||||
func ExistChain(chain string, table Table) bool {
|
||||
if _, err := Raw("-t", string(table), "-L", chain); err == nil {
|
||||
if _, err := Raw("-t", string(table), "-nL", chain); err == nil {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
|
||||
+4
@@ -995,6 +995,10 @@ func (n *network) delete(force bool) error {
|
||||
logrus.Errorf("Failed leaving network %s from the agent cluster: %v", n.Name(), err)
|
||||
}
|
||||
|
||||
// Cleanup the service discovery for this network
|
||||
c.cleanupServiceDiscovery(n.ID())
|
||||
|
||||
// Cleanup the load balancer
|
||||
c.cleanupServiceBindings(n.ID())
|
||||
|
||||
removeFromStore:
|
||||
|
||||
+2
@@ -134,6 +134,8 @@ func (nDB *NetworkDB) sendTableEvent(event TableEvent_Type, nid string, tname st
|
||||
TableName: tname,
|
||||
Key: key,
|
||||
Value: entry.value,
|
||||
// The duration in second is a float that below would be truncated
|
||||
ResidualReapTime: int32(entry.reapTime.Seconds()),
|
||||
}
|
||||
|
||||
raw, err := encodeMessage(MessageTypeTableEvent, &tEvent)
|
||||
|
||||
+55
-39
@@ -17,11 +17,15 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
reapInterval = 30 * time.Minute
|
||||
reapPeriod = 5 * time.Second
|
||||
retryInterval = 1 * time.Second
|
||||
nodeReapInterval = 24 * time.Hour
|
||||
nodeReapPeriod = 2 * time.Hour
|
||||
// The garbage collection logic for entries leverage the presence of the network.
|
||||
// For this reason the expiration time of the network is put slightly higher than the entry expiration so that
|
||||
// there is at least 5 extra cycle to make sure that all the entries are properly deleted before deleting the network.
|
||||
reapEntryInterval = 30 * time.Minute
|
||||
reapNetworkInterval = reapEntryInterval + 5*reapPeriod
|
||||
reapPeriod = 5 * time.Second
|
||||
retryInterval = 1 * time.Second
|
||||
nodeReapInterval = 24 * time.Hour
|
||||
nodeReapPeriod = 2 * time.Hour
|
||||
)
|
||||
|
||||
type logWriter struct{}
|
||||
@@ -300,8 +304,9 @@ func (nDB *NetworkDB) reconnectNode() {
|
||||
// the reaper runs. NOTE nDB.reapTableEntries updates the reapTime with a readlock. This
|
||||
// is safe as long as no other concurrent path touches the reapTime field.
|
||||
func (nDB *NetworkDB) reapState() {
|
||||
nDB.reapNetworks()
|
||||
// The reapTableEntries leverage the presence of the network so garbage collect entries first
|
||||
nDB.reapTableEntries()
|
||||
nDB.reapNetworks()
|
||||
}
|
||||
|
||||
func (nDB *NetworkDB) reapNetworks() {
|
||||
@@ -321,43 +326,51 @@ func (nDB *NetworkDB) reapNetworks() {
|
||||
}
|
||||
|
||||
func (nDB *NetworkDB) reapTableEntries() {
|
||||
var paths []string
|
||||
|
||||
var nodeNetworks []string
|
||||
// This is best effort, if the list of network changes will be picked up in the next cycle
|
||||
nDB.RLock()
|
||||
nDB.indexes[byTable].Walk(func(path string, v interface{}) bool {
|
||||
entry, ok := v.(*entry)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
|
||||
if !entry.deleting {
|
||||
return false
|
||||
}
|
||||
if entry.reapTime > 0 {
|
||||
entry.reapTime -= reapPeriod
|
||||
return false
|
||||
}
|
||||
paths = append(paths, path)
|
||||
return false
|
||||
})
|
||||
for nid := range nDB.networks[nDB.config.NodeName] {
|
||||
nodeNetworks = append(nodeNetworks, nid)
|
||||
}
|
||||
nDB.RUnlock()
|
||||
|
||||
nDB.Lock()
|
||||
for _, path := range paths {
|
||||
params := strings.Split(path[1:], "/")
|
||||
tname := params[0]
|
||||
nid := params[1]
|
||||
key := params[2]
|
||||
cycleStart := time.Now()
|
||||
// In order to avoid blocking the database for a long time, apply the garbage collection logic by network
|
||||
// The lock is taken at the beginning of the cycle and the deletion is inline
|
||||
for _, nid := range nodeNetworks {
|
||||
nDB.Lock()
|
||||
nDB.indexes[byNetwork].WalkPrefix(fmt.Sprintf("/%s", nid), func(path string, v interface{}) bool {
|
||||
// timeCompensation compensate in case the lock took some time to be released
|
||||
timeCompensation := time.Since(cycleStart)
|
||||
entry, ok := v.(*entry)
|
||||
if !ok || !entry.deleting {
|
||||
return false
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
// In this check we are adding an extra 1 second to guarantee that when the number is truncated to int32 to fit the packet
|
||||
// for the tableEvent the number is always strictly > 1 and never 0
|
||||
if entry.reapTime > reapPeriod+timeCompensation+time.Second {
|
||||
entry.reapTime -= reapPeriod + timeCompensation
|
||||
return false
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
params := strings.Split(path[1:], "/")
|
||||
nid := params[0]
|
||||
tname := params[1]
|
||||
key := params[2]
|
||||
|
||||
okTable, okNetwork := nDB.deleteEntry(nid, tname, key)
|
||||
if !okTable {
|
||||
logrus.Errorf("Table tree delete failed, entry with key:%s does not exists in the table:%s network:%s", key, tname, nid)
|
||||
}
|
||||
if !okNetwork {
|
||||
logrus.Errorf("Network tree delete failed, entry with key:%s does not exists in the network:%s table:%s", key, nid, tname)
|
||||
}
|
||||
|
||||
return false
|
||||
})
|
||||
nDB.Unlock()
|
||||
}
|
||||
nDB.Unlock()
|
||||
}
|
||||
|
||||
func (nDB *NetworkDB) gossip() {
|
||||
@@ -406,8 +419,9 @@ func (nDB *NetworkDB) gossip() {
|
||||
// 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)))
|
||||
logrus.Infof("NetworkDB stats - netID:%s leaving:%t netPeers:%d entries:%d Queue qLen:%d netMsg/s:%d",
|
||||
nid, network.leaving, broadcastQ.NumNodes(), network.entriesNumber, broadcastQ.NumQueued(),
|
||||
network.qMessagesSent/int((nDB.config.StatsPrintPeriod/time.Second)))
|
||||
network.qMessagesSent = 0
|
||||
}
|
||||
|
||||
@@ -572,6 +586,8 @@ func (nDB *NetworkDB) bulkSyncNode(networks []string, node string, unsolicited b
|
||||
TableName: params[1],
|
||||
Key: params[2],
|
||||
Value: entry.value,
|
||||
// The duration in second is a float that below would be truncated
|
||||
ResidualReapTime: int32(entry.reapTime.Seconds()),
|
||||
}
|
||||
|
||||
msg, err := encodeMessage(MessageTypeTableEvent, &tEvent)
|
||||
|
||||
+20
-19
@@ -1,9 +1,9 @@
|
||||
package networkdb
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gogo/protobuf/proto"
|
||||
"github.com/sirupsen/logrus"
|
||||
@@ -165,7 +165,7 @@ func (nDB *NetworkDB) handleNetworkEvent(nEvent *NetworkEvent) bool {
|
||||
n.ltime = nEvent.LTime
|
||||
n.leaving = nEvent.Type == NetworkEventTypeLeave
|
||||
if n.leaving {
|
||||
n.reapTime = reapInterval
|
||||
n.reapTime = reapNetworkInterval
|
||||
|
||||
// The remote node is leaving the network, but not the gossip cluster.
|
||||
// Mark all its entries in deleted state, this will guarantee that
|
||||
@@ -198,8 +198,7 @@ func (nDB *NetworkDB) handleNetworkEvent(nEvent *NetworkEvent) bool {
|
||||
}
|
||||
|
||||
func (nDB *NetworkDB) handleTableEvent(tEvent *TableEvent) bool {
|
||||
// Update our local clock if the received messages has newer
|
||||
// time.
|
||||
// Update our local clock if the received messages has newer time.
|
||||
nDB.tableClock.Witness(tEvent.LTime)
|
||||
|
||||
// Ignore the table events for networks that are in the process of going away
|
||||
@@ -235,20 +234,27 @@ func (nDB *NetworkDB) handleTableEvent(tEvent *TableEvent) bool {
|
||||
node: tEvent.NodeName,
|
||||
value: tEvent.Value,
|
||||
deleting: tEvent.Type == TableEventTypeDelete,
|
||||
reapTime: time.Duration(tEvent.ResidualReapTime) * time.Second,
|
||||
}
|
||||
|
||||
if e.deleting {
|
||||
e.reapTime = reapInterval
|
||||
// All the entries marked for deletion should have a reapTime set greater than 0
|
||||
// This case can happen if the cluster is running different versions of the engine where the old version does not have the
|
||||
// field. If that is not the case, this can be a BUG
|
||||
if e.deleting && e.reapTime == 0 {
|
||||
logrus.Warnf("handleTableEvent object %+v has a 0 reapTime, is the cluster running the same docker engine version?", tEvent)
|
||||
e.reapTime = reapEntryInterval
|
||||
}
|
||||
|
||||
nDB.Lock()
|
||||
nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tEvent.TableName, tEvent.NetworkID, tEvent.Key), e)
|
||||
nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", tEvent.NetworkID, tEvent.TableName, tEvent.Key), e)
|
||||
nDB.createOrUpdateEntry(tEvent.NetworkID, tEvent.TableName, tEvent.Key, e)
|
||||
nDB.Unlock()
|
||||
|
||||
if err != nil && tEvent.Type == TableEventTypeDelete {
|
||||
// If it is a delete event and we didn't have the entry here don't repropagate
|
||||
return true
|
||||
// If it is a delete event and we did not have a state for it, don't propagate to the application
|
||||
// If the residual reapTime is lower or equal to 1/6 of the total reapTime don't bother broadcasting it around
|
||||
// most likely the cluster is already aware of it, if not who will sync with this node will catch the state too.
|
||||
// This also avoids that deletion of entries close to their garbage collection ends up circuling around forever
|
||||
return e.reapTime > reapEntryInterval/6
|
||||
}
|
||||
|
||||
var op opType
|
||||
@@ -303,22 +309,17 @@ func (nDB *NetworkDB) handleTableMessage(buf []byte, isBulkSync bool) {
|
||||
n, ok := nDB.networks[nDB.config.NodeName][tEvent.NetworkID]
|
||||
nDB.RUnlock()
|
||||
|
||||
if !ok {
|
||||
// if the network is not there anymore, OR we are leaving the network OR the broadcast queue is not present
|
||||
if !ok || n.leaving || n.tableBroadcasts == nil {
|
||||
return
|
||||
}
|
||||
|
||||
broadcastQ := n.tableBroadcasts
|
||||
|
||||
if broadcastQ == nil {
|
||||
return
|
||||
}
|
||||
|
||||
broadcastQ.QueueBroadcast(&tableEventMessage{
|
||||
n.tableBroadcasts.QueueBroadcast(&tableEventMessage{
|
||||
msg: buf,
|
||||
id: tEvent.NetworkID,
|
||||
tname: tEvent.TableName,
|
||||
key: tEvent.Key,
|
||||
node: nDB.config.NodeName,
|
||||
node: tEvent.NodeName,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
+60
-18
@@ -141,6 +141,11 @@ type network struct {
|
||||
|
||||
// Number of gossip messages sent related to this network during the last stats collection period
|
||||
qMessagesSent int
|
||||
|
||||
// Number of entries on the network. This value is the sum of all the entries of all the tables of a specific network.
|
||||
// Its use is for statistics purposes. It keep tracks of database size and is printed per network every StatsPrintPeriod
|
||||
// interval
|
||||
entriesNumber int
|
||||
}
|
||||
|
||||
// Config represents the configuration of the networdb instance and
|
||||
@@ -338,8 +343,7 @@ func (nDB *NetworkDB) CreateEntry(tname, nid, key string, value []byte) error {
|
||||
}
|
||||
|
||||
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.createOrUpdateEntry(nid, tname, key, entry)
|
||||
nDB.Unlock()
|
||||
|
||||
return nil
|
||||
@@ -365,8 +369,7 @@ func (nDB *NetworkDB) UpdateEntry(tname, nid, key string, value []byte) error {
|
||||
}
|
||||
|
||||
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.createOrUpdateEntry(nid, tname, key, entry)
|
||||
nDB.Unlock()
|
||||
|
||||
return nil
|
||||
@@ -402,7 +405,7 @@ func (nDB *NetworkDB) DeleteEntry(tname, nid, key string) error {
|
||||
node: nDB.config.NodeName,
|
||||
value: value,
|
||||
deleting: true,
|
||||
reapTime: reapInterval,
|
||||
reapTime: reapEntryInterval,
|
||||
}
|
||||
|
||||
if err := nDB.sendTableEvent(TableEventTypeDelete, nid, tname, key, entry); err != nil {
|
||||
@@ -410,8 +413,7 @@ func (nDB *NetworkDB) DeleteEntry(tname, nid, key string) error {
|
||||
}
|
||||
|
||||
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.createOrUpdateEntry(nid, tname, key, entry)
|
||||
nDB.Unlock()
|
||||
|
||||
return nil
|
||||
@@ -473,10 +475,10 @@ func (nDB *NetworkDB) deleteNodeNetworkEntries(nid, node string) {
|
||||
|
||||
entry := &entry{
|
||||
ltime: oldEntry.ltime,
|
||||
node: node,
|
||||
node: oldEntry.node,
|
||||
value: oldEntry.value,
|
||||
deleting: true,
|
||||
reapTime: reapInterval,
|
||||
reapTime: reapEntryInterval,
|
||||
}
|
||||
|
||||
// we arrived at this point in 2 cases:
|
||||
@@ -488,15 +490,19 @@ func (nDB *NetworkDB) deleteNodeNetworkEntries(nid, node string) {
|
||||
// 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)
|
||||
|
||||
if !oldEntry.deleting {
|
||||
nDB.createOrUpdateEntry(nid, tname, key, entry)
|
||||
}
|
||||
} else {
|
||||
// the local node is leaving the network, all the entries of remote nodes can be safely removed
|
||||
nDB.indexes[byTable].Delete(fmt.Sprintf("/%s/%s/%s", tname, nid, key))
|
||||
nDB.indexes[byNetwork].Delete(fmt.Sprintf("/%s/%s/%s", nid, tname, key))
|
||||
nDB.deleteEntry(nid, tname, key)
|
||||
}
|
||||
|
||||
nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, entry.value))
|
||||
// Notify to the upper layer only entries not already marked for deletion
|
||||
if !oldEntry.deleting {
|
||||
nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, entry.value))
|
||||
}
|
||||
return false
|
||||
})
|
||||
}
|
||||
@@ -513,8 +519,7 @@ func (nDB *NetworkDB) deleteNodeTableEntries(node string) {
|
||||
nid := params[1]
|
||||
key := params[2]
|
||||
|
||||
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.deleteEntry(nid, tname, key)
|
||||
|
||||
nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, oldEntry.value))
|
||||
return false
|
||||
@@ -558,7 +563,12 @@ func (nDB *NetworkDB) JoinNetwork(nid string) error {
|
||||
nodeNetworks = make(map[string]*network)
|
||||
nDB.networks[nDB.config.NodeName] = nodeNetworks
|
||||
}
|
||||
nodeNetworks[nid] = &network{id: nid, ltime: ltime}
|
||||
n, ok := nodeNetworks[nid]
|
||||
var entries int
|
||||
if ok {
|
||||
entries = n.entriesNumber
|
||||
}
|
||||
nodeNetworks[nid] = &network{id: nid, ltime: ltime, entriesNumber: entries}
|
||||
nodeNetworks[nid].tableBroadcasts = &memberlist.TransmitLimitedQueue{
|
||||
NumNodes: func() int {
|
||||
nDB.RLock()
|
||||
@@ -567,6 +577,7 @@ func (nDB *NetworkDB) JoinNetwork(nid string) error {
|
||||
},
|
||||
RetransmitMult: 4,
|
||||
}
|
||||
|
||||
nDB.addNetworkNode(nid, nDB.config.NodeName)
|
||||
networkNodes := nDB.networkNodes[nid]
|
||||
nDB.Unlock()
|
||||
@@ -614,8 +625,9 @@ func (nDB *NetworkDB) LeaveNetwork(nid string) error {
|
||||
return fmt.Errorf("could not find network %s while trying to leave", nid)
|
||||
}
|
||||
|
||||
logrus.Debugf("%s: leaving network %s", nDB.config.NodeName, nid)
|
||||
n.ltime = ltime
|
||||
n.reapTime = reapInterval
|
||||
n.reapTime = reapNetworkInterval
|
||||
n.leaving = true
|
||||
return nil
|
||||
}
|
||||
@@ -679,3 +691,33 @@ func (nDB *NetworkDB) updateLocalNetworkTime() {
|
||||
n.ltime = ltime
|
||||
}
|
||||
}
|
||||
|
||||
// createOrUpdateEntry this function handles the creation or update of entries into the local
|
||||
// tree store. It is also used to keep in sync the entries number of the network (all tables are aggregated)
|
||||
func (nDB *NetworkDB) createOrUpdateEntry(nid, tname, key string, entry interface{}) (bool, bool) {
|
||||
_, okTable := nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)
|
||||
_, okNetwork := nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)
|
||||
if !okNetwork {
|
||||
// Add only if it is an insert not an update
|
||||
n, ok := nDB.networks[nDB.config.NodeName][nid]
|
||||
if ok {
|
||||
n.entriesNumber++
|
||||
}
|
||||
}
|
||||
return okTable, okNetwork
|
||||
}
|
||||
|
||||
// deleteEntry this function handles the deletion of entries into the local tree store.
|
||||
// It is also used to keep in sync the entries number of the network (all tables are aggregated)
|
||||
func (nDB *NetworkDB) deleteEntry(nid, tname, key string) (bool, bool) {
|
||||
_, okTable := nDB.indexes[byTable].Delete(fmt.Sprintf("/%s/%s/%s", tname, nid, key))
|
||||
_, okNetwork := nDB.indexes[byNetwork].Delete(fmt.Sprintf("/%s/%s/%s", nid, tname, key))
|
||||
if okNetwork {
|
||||
// Remove only if the delete is successful
|
||||
n, ok := nDB.networks[nDB.config.NodeName][nid]
|
||||
if ok {
|
||||
n.entriesNumber--
|
||||
}
|
||||
}
|
||||
return okTable, okNetwork
|
||||
}
|
||||
|
||||
+495
-316
File diff suppressed because it is too large
Load Diff
+6
-2
@@ -109,7 +109,7 @@ message NetworkEntry {
|
||||
// network event was recorded.
|
||||
uint64 l_time = 2 [(gogoproto.customtype) = "github.com/hashicorp/serf/serf.LamportTime", (gogoproto.nullable) = false];
|
||||
// Source node name where this network attachment happened.
|
||||
string node_name = 3;
|
||||
string node_name = 3 [(gogoproto.customname) = "NodeName"];
|
||||
// Indicates if a leave from this network is in progress.
|
||||
bool leaving = 4;
|
||||
}
|
||||
@@ -119,6 +119,8 @@ message NetworkPushPull {
|
||||
// Lamport time when this push pull was initiated.
|
||||
uint64 l_time = 1 [(gogoproto.customtype) = "github.com/hashicorp/serf/serf.LamportTime", (gogoproto.nullable) = false];
|
||||
repeated NetworkEntry networks = 2;
|
||||
// Name of the node sending this push pull payload.
|
||||
string node_name = 3 [(gogoproto.customname) = "NodeName"];
|
||||
}
|
||||
|
||||
// TableEvent message payload definition.
|
||||
@@ -152,6 +154,8 @@ message TableEvent {
|
||||
string key = 6;
|
||||
// Entry value.
|
||||
bytes value = 7;
|
||||
// Residual reap time for the entry before getting deleted in seconds
|
||||
int32 residual_reap_time = 8 [(gogoproto.customname) = "ResidualReapTime"];;
|
||||
}
|
||||
|
||||
// BulkSync message payload definition.
|
||||
@@ -180,4 +184,4 @@ message CompoundMessage {
|
||||
|
||||
// A list of simple messages.
|
||||
repeated SimpleMessage messages = 1;
|
||||
}
|
||||
}
|
||||
|
||||
+26
-11
@@ -9,6 +9,17 @@ import (
|
||||
"github.com/vishvananda/netlink"
|
||||
)
|
||||
|
||||
// NeighborSearchError indicates that the neighbor is already present
|
||||
type NeighborSearchError struct {
|
||||
ip net.IP
|
||||
mac net.HardwareAddr
|
||||
present bool
|
||||
}
|
||||
|
||||
func (n NeighborSearchError) Error() string {
|
||||
return fmt.Sprintf("Search neighbor failed for IP %v, mac %v, present in db:%t", n.ip, n.mac, n.present)
|
||||
}
|
||||
|
||||
// NeighOption is a function option type to set interface options
|
||||
type NeighOption func(nh *neigh)
|
||||
|
||||
@@ -41,7 +52,7 @@ func (n *networkNamespace) DeleteNeighbor(dstIP net.IP, dstMac net.HardwareAddr,
|
||||
|
||||
nh := n.findNeighbor(dstIP, dstMac)
|
||||
if nh == nil {
|
||||
return fmt.Errorf("could not find the neighbor entry to delete")
|
||||
return NeighborSearchError{dstIP, dstMac, false}
|
||||
}
|
||||
|
||||
if osDelete {
|
||||
@@ -103,26 +114,27 @@ func (n *networkNamespace) DeleteNeighbor(dstIP net.IP, dstMac net.HardwareAddr,
|
||||
}
|
||||
}
|
||||
n.Unlock()
|
||||
logrus.Debugf("Neighbor entry deleted for IP %v, mac %v", dstIP, dstMac)
|
||||
logrus.Debugf("Neighbor entry deleted for IP %v, mac %v osDelete:%t", dstIP, dstMac, osDelete)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (n *networkNamespace) AddNeighbor(dstIP net.IP, dstMac net.HardwareAddr, force bool, options ...NeighOption) error {
|
||||
var (
|
||||
iface netlink.Link
|
||||
err error
|
||||
iface netlink.Link
|
||||
err error
|
||||
neighborAlreadyPresent bool
|
||||
)
|
||||
|
||||
// If the namespace already has the neighbor entry but the AddNeighbor is called
|
||||
// because of a miss notification (force flag) program the kernel anyway.
|
||||
nh := n.findNeighbor(dstIP, dstMac)
|
||||
if nh != nil {
|
||||
neighborAlreadyPresent = true
|
||||
logrus.Warnf("Neighbor entry already present for IP %v, mac %v neighbor:%+v forceUpdate:%t", dstIP, dstMac, nh, force)
|
||||
if !force {
|
||||
logrus.Warnf("Neighbor entry already present for IP %v, mac %v", dstIP, dstMac)
|
||||
return nil
|
||||
return NeighborSearchError{dstIP, dstMac, true}
|
||||
}
|
||||
logrus.Warnf("Force kernel update, Neighbor entry already present for IP %v, mac %v", dstIP, dstMac)
|
||||
}
|
||||
|
||||
nh = &neigh{
|
||||
@@ -146,8 +158,7 @@ func (n *networkNamespace) AddNeighbor(dstIP net.IP, dstMac net.HardwareAddr, fo
|
||||
if nh.linkDst != "" {
|
||||
iface, err = nlh.LinkByName(nh.linkDst)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not find interface with destination name %s: %v",
|
||||
nh.linkDst, err)
|
||||
return fmt.Errorf("could not find interface with destination name %s: %v", nh.linkDst, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -167,13 +178,17 @@ func (n *networkNamespace) AddNeighbor(dstIP net.IP, dstMac net.HardwareAddr, fo
|
||||
}
|
||||
|
||||
if err := nlh.NeighSet(nlnh); err != nil {
|
||||
return fmt.Errorf("could not add neighbor entry: %v", err)
|
||||
return fmt.Errorf("could not add neighbor entry:%+v error:%v", nlnh, err)
|
||||
}
|
||||
|
||||
if neighborAlreadyPresent {
|
||||
return nil
|
||||
}
|
||||
|
||||
n.Lock()
|
||||
n.neighbors = append(n.neighbors, nh)
|
||||
n.Unlock()
|
||||
logrus.Debugf("Neighbor entry added for IP %v, mac %v", dstIP, dstMac)
|
||||
logrus.Debugf("Neighbor entry added for IP:%v, mac:%v on ifc:%s", dstIP, dstMac, nh.linkName)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
+1
-5
@@ -67,11 +67,7 @@ func (sb *sandbox) setupResolutionFiles() error {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := sb.setupDNS(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
return sb.setupDNS()
|
||||
}
|
||||
|
||||
func (sb *sandbox) buildHostsFile() error {
|
||||
|
||||
+13
-9
@@ -161,6 +161,19 @@ func (c *controller) getLBIndex(sid, nid string, ingressPorts []*PortConfig) int
|
||||
return int(lb.fwMark)
|
||||
}
|
||||
|
||||
// cleanupServiceDiscovery when the network is being deleted, erase all the associated service discovery records
|
||||
func (c *controller) cleanupServiceDiscovery(cleanupNID string) {
|
||||
c.Lock()
|
||||
defer c.Unlock()
|
||||
if cleanupNID == "" {
|
||||
logrus.Debugf("cleanupServiceDiscovery for all networks")
|
||||
c.svcRecords = make(map[string]svcInfo)
|
||||
return
|
||||
}
|
||||
logrus.Debugf("cleanupServiceDiscovery for network:%s", cleanupNID)
|
||||
delete(c.svcRecords, cleanupNID)
|
||||
}
|
||||
|
||||
func (c *controller) cleanupServiceBindings(cleanupNID string) {
|
||||
var cleanupFuncs []func()
|
||||
|
||||
@@ -184,15 +197,6 @@ func (c *controller) cleanupServiceBindings(cleanupNID string) {
|
||||
continue
|
||||
}
|
||||
|
||||
// The network is being deleted, erase all the associated service discovery records
|
||||
// TODO(fcrisciani) separate the Load Balancer from the Service discovery, this operation
|
||||
// can be done safely here, but the rmServiceBinding is still keeping consistency in the
|
||||
// data structures that are tracking the endpoint to IP mapping.
|
||||
c.Lock()
|
||||
logrus.Debugf("cleanupServiceBindings erasing the svcRecords for %s", nid)
|
||||
delete(c.svcRecords, nid)
|
||||
c.Unlock()
|
||||
|
||||
for eid, ip := range lb.backEnds {
|
||||
epID := eid
|
||||
epIP := ip
|
||||
|
||||
+2
-2
@@ -129,11 +129,11 @@ func (p *PortBinding) GetCopy() PortBinding {
|
||||
func (p *PortBinding) String() string {
|
||||
ret := fmt.Sprintf("%s/", p.Proto)
|
||||
if p.IP != nil {
|
||||
ret = fmt.Sprintf("%s%s", ret, p.IP.String())
|
||||
ret += p.IP.String()
|
||||
}
|
||||
ret = fmt.Sprintf("%s:%d/", ret, p.Port)
|
||||
if p.HostIP != nil {
|
||||
ret = fmt.Sprintf("%s%s", ret, p.HostIP.String())
|
||||
ret += p.HostIP.String()
|
||||
}
|
||||
ret = fmt.Sprintf("%s:%d", ret, p.HostPort)
|
||||
return ret
|
||||
|
||||
Reference in New Issue
Block a user