From 39ac1a0037aaa86282084f7e9462d681926ccd0d Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 9 Feb 2017 21:01:52 +0000 Subject: [PATCH] Collect existing weave addresses at start-up, to avoid a race when doing it from the script Note we need to run weaver in the host pid namespace, so it can search for existing container IPs. Also now we are firing off sub-processes it's better to not be PID 1 so Linux doesn't expect us to reap zombies. --- ipam/allocator.go | 15 ++++++++++++ ipam/allocator_test.go | 13 ++++++++-- ipam/claim.go | 2 +- ipam/testutils_test.go | 11 ++++++--- prog/weaver/main.go | 7 ++++-- prog/weaver/reclaim.go | 55 ++++++++++++++++++++++++++++++++++++++++++ weave | 1 + 7 files changed, 96 insertions(+), 8 deletions(-) create mode 100644 prog/weaver/reclaim.go diff --git a/ipam/allocator.go b/ipam/allocator.go index fa7ddb26..fa8cfd30 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -76,6 +76,11 @@ type Allocator struct { now func() time.Time } +type PreClaim struct { + Ident string + Cidr address.CIDR +} + type Config struct { OurName mesh.PeerName OurUID mesh.PeerUID @@ -83,6 +88,7 @@ type Config struct { Seed []mesh.PeerName Universe address.CIDR IsObserver bool + PreClaims []PreClaim Quorum func() uint Db db.DB IsKnownPeer func(name mesh.PeerName) bool @@ -123,6 +129,12 @@ func NewAllocator(config Config) *Allocator { dead: make(map[string]time.Time), now: time.Now, } + + alloc.pendingClaims = make([]operation, len(config.PreClaims)) + for i, c := range config.PreClaims { + alloc.pendingClaims[i] = &claim{ident: c.Ident, cidr: c.Cidr} + } + return alloc } @@ -156,6 +168,9 @@ func (alloc *Allocator) Start() { default: alloc.infof("Initialising as observer - awaiting IPAM data from another peer") } + if loadedPersistedData { // do any pre-claims right away + alloc.tryOps(&alloc.pendingClaims) + } actionChan := make(chan func(), mesh.ChannelSize) stopChan := make(chan struct{}) alloc.actionChan = actionChan diff --git a/ipam/allocator_test.go b/ipam/allocator_test.go index ce7c3759..64addc3f 100644 --- a/ipam/allocator_test.go +++ b/ipam/allocator_test.go @@ -168,11 +168,13 @@ func TestAllocatorClaim(t *testing.T) { container1 = "abcdef" container3 = "b01df00d" universe = "10.0.3.0/24" - testAddr1 = "10.0.3.2/24" + testAddr1 = "10.0.3.5/24" testAddr2 = "10.0.4.2/24" + testPre = "10.0.3.1/24" ) - allocs, router, subnet := makeNetworkOfAllocators(2, universe) + preAddr, _ := address.ParseCIDR(testPre) + allocs, router, subnet := makeNetworkOfAllocators(2, universe, []PreClaim{{container1, preAddr}}) defer stopNetworkOfAllocators(allocs, router) alloc := allocs[1] addr1, _ := address.ParseCIDR(testAddr1) @@ -184,6 +186,8 @@ func TestAllocatorClaim(t *testing.T) { alloc.Prime() // Do an allocate on the other peer, which we will try to claim later addrx, err := allocs[0].Allocate(container1, subnet, true, returnFalse) + // Should not get the address we pre-claimed + require.NotEqual(t, addrx, preAddr) router.Flush() // Now try the claim again @@ -201,6 +205,11 @@ func TestAllocatorClaim(t *testing.T) { // claiming the address allocated on the other peer should fail err = alloc.SimplyClaim(container1, address.MakeCIDR(subnet, addrx)) require.Error(t, err, "claiming address allocated on other peer should fail") + // claiming the pre-claimed address should fail on both peers + err = alloc.SimplyClaim(container3, preAddr) + require.Error(t, err, "claiming address allocated on other peer should fail") + err = allocs[0].SimplyClaim(container3, preAddr) + require.Error(t, err, "claiming address allocated on other peer should fail") // Check an address outside of our universe addr2, _ := address.ParseCIDR(testAddr2) err = alloc.SimplyClaim(container1, addr2) diff --git a/ipam/claim.go b/ipam/claim.go index e027f723..dfd3ab76 100644 --- a/ipam/claim.go +++ b/ipam/claim.go @@ -35,7 +35,7 @@ func (c *claim) sendResult(result error) { // Try returns true for success (or failure), false if we need to try again later func (c *claim) Try(alloc *Allocator) bool { - if c.hasBeenCancelled() { + if c.hasBeenCancelled != nil && c.hasBeenCancelled() { c.Cancel() return true } diff --git a/ipam/testutils_test.go b/ipam/testutils_test.go index 13fa8171..d6b2f07f 100644 --- a/ipam/testutils_test.go +++ b/ipam/testutils_test.go @@ -128,7 +128,7 @@ type mockDB struct{} func (d *mockDB) Load(_ string, _ interface{}) (bool, error) { return false, nil } func (d *mockDB) Save(_ string, _ interface{}) error { return nil } -func makeAllocator(name string, cidrStr string, quorum uint) (*Allocator, address.CIDR) { +func makeAllocator(name string, cidrStr string, quorum uint, preClaims ...PreClaim) (*Allocator, address.CIDR) { peername, err := mesh.PeerNameFromString(name) if err != nil { panic(err) @@ -145,6 +145,7 @@ func makeAllocator(name string, cidrStr string, quorum uint) (*Allocator, addres OurNickname: "nick-" + name, Universe: cidr, IsObserver: quorum == 0, + PreClaims: preClaims, Quorum: func() uint { return quorum }, Db: new(mockDB), IsKnownPeer: func(mesh.PeerName) bool { return true }, @@ -213,15 +214,19 @@ func AssertNothingSentErr(t *testing.T, ch <-chan error) { } } -func makeNetworkOfAllocators(size int, cidr string) ([]*Allocator, *gossip.TestRouter, address.CIDR) { +func makeNetworkOfAllocators(size int, cidr string, preClaims ...[]PreClaim) ([]*Allocator, *gossip.TestRouter, address.CIDR) { gossipRouter := gossip.NewTestRouter(0.0) allocs := make([]*Allocator, size) var subnet address.CIDR for i := 0; i < size; i++ { var alloc *Allocator + preClaim := []PreClaim{} + if i < len(preClaims) { + preClaim = preClaims[i] + } alloc, subnet = makeAllocator(fmt.Sprintf("%02d:00:00:02:00:00", i), - cidr, uint(size/2+1)) + cidr, uint(size/2+1), preClaim...) alloc.SetInterfaces(gossipRouter.Connect(alloc.ourName, alloc)) alloc.Start() allocs[i] = alloc diff --git a/prog/weaver/main.go b/prog/weaver/main.go index 92609b79..b4f7fcc6 100644 --- a/prog/weaver/main.go +++ b/prog/weaver/main.go @@ -306,7 +306,9 @@ func main() { } trackerName = "awsvpc" } - allocator, defaultSubnet = createAllocator(router, ipamConfig, db, t, isKnownPeer) + preClaims, err := findExistingAddresses(weavenet.WeaveBridgeName) + checkFatal(err) + allocator, defaultSubnet = createAllocator(router, ipamConfig, preClaims, db, t, isKnownPeer) observeContainers(allocator) if dockerCli != nil { ids, err := dockerCli.AllContainerIDs() @@ -445,7 +447,7 @@ func createOverlay(datapathName string, ifaceName string, isAWSVPC bool, host st return overlay, bridge } -func createAllocator(router *weave.NetworkRouter, config ipamConfig, db db.DB, track tracker.LocalRangeTracker, isKnownPeer func(mesh.PeerName) bool) (*ipam.Allocator, address.CIDR) { +func createAllocator(router *weave.NetworkRouter, config ipamConfig, preClaims []ipam.PreClaim, db db.DB, track tracker.LocalRangeTracker, isKnownPeer func(mesh.PeerName) bool) (*ipam.Allocator, address.CIDR) { ipRange, err := ipam.ParseCIDRSubnet(config.IPRangeCIDR) checkFatal(err) defaultSubnet := ipRange @@ -464,6 +466,7 @@ func createAllocator(router *weave.NetworkRouter, config ipamConfig, db db.DB, t Seed: config.SeedPeerNames, Universe: ipRange, IsObserver: config.Observer, + PreClaims: preClaims, Quorum: func() uint { return determineQuorum(config.PeerCount, router) }, Db: db, IsKnownPeer: isKnownPeer, diff --git a/prog/weaver/reclaim.go b/prog/weaver/reclaim.go new file mode 100644 index 00000000..8678fe8d --- /dev/null +++ b/prog/weaver/reclaim.go @@ -0,0 +1,55 @@ +package main + +import ( + "net" + + "github.com/weaveworks/weave/api" + "github.com/weaveworks/weave/common" + "github.com/weaveworks/weave/ipam" + weavenet "github.com/weaveworks/weave/net" + "github.com/weaveworks/weave/net/address" +) + +func a(cidr *net.IPNet) address.CIDR { + prefixLength, _ := cidr.Mask.Size() + return address.CIDR{Addr: address.FromIP4(cidr.IP), PrefixLen: prefixLength} +} + +// Get all the existing Weave IPs at startup, so we can stop IPAM +// giving out any as duplicates +func findExistingAddresses(bridgeName string) (addrs []ipam.PreClaim, err error) { + // First get the address for the bridge + bridgeNetDev, err := weavenet.GetBridgeNetDev(bridgeName) + if err != nil { + return nil, err + } + for _, cidr := range bridgeNetDev.CIDRs { + addrs = append(addrs, ipam.PreClaim{Ident: "weave:expose", Cidr: a(cidr)}) + } + + // Then find all veths connected to the bridge + peerIDs, err := weavenet.ConnectedToBridgeVethPeerIds(bridgeName) + if err != nil { + return nil, err + } + + pids, err := common.AllPids("/proc") + if err != nil { + return nil, err + } + + // Now iterate over all processes to see if they have a network namespace with an attached interface + for _, pid := range pids { + netDevs, err := weavenet.GetNetDevsByVethPeerIds(pid, peerIDs) + if err != nil { + return nil, err + } + for _, netDev := range netDevs { + for _, cidr := range netDev.CIDRs { + // We don't know the container ID, so use special magic string + addrs = append(addrs, ipam.PreClaim{Ident: api.NoContainerID, Cidr: a(cidr)}) + } + } + } + return addrs, nil +} diff --git a/weave b/weave index d4c79e12..576b058b 100755 --- a/weave +++ b/weave @@ -1691,6 +1691,7 @@ launch_router() { ROUTER_CONTAINER=$(docker run -d --name=$CONTAINER_NAME \ $(docker_run_options) \ $RESTART_POLICY \ + --pid=host \ --volumes-from $DB_CONTAINER_NAME \ -v $RESOLV_CONF_DIR:/var/run/weave/etc \ -e WEAVE_PASSWORD \