diff --git a/common/docker/client.go b/common/docker/client.go index 5bd04f3f..cc845c28 100644 --- a/common/docker/client.go +++ b/common/docker/client.go @@ -14,6 +14,7 @@ import ( type ContainerObserver interface { ContainerStarted(ident string) ContainerDied(ident string) + ContainerDestroyed(ident string) } type Client struct { @@ -82,11 +83,11 @@ func (c *Client) AddObserver(ob ContainerObserver) error { for event := range events { switch event.Status { case "start": - id := event.ID - ob.ContainerStarted(id) + ob.ContainerStarted(event.ID) case "die": - id := event.ID - ob.ContainerDied(id) + ob.ContainerDied(event.ID) + case "destroy": + ob.ContainerDestroyed(event.ID) } } }() diff --git a/ipam/allocate.go b/ipam/allocate.go index d6dd2f10..d891f7fc 100644 --- a/ipam/allocate.go +++ b/ipam/allocate.go @@ -26,6 +26,8 @@ func (g *allocate) Try(alloc *Allocator) bool { } if addr, found := alloc.lookupOwned(g.ident, g.r); found { + // If we had heard that this container died, resurrect it + delete(alloc.dead, g.ident) // delete is no-op if key not in map g.resultChan <- allocateResult{addr, nil} return true } diff --git a/ipam/allocator.go b/ipam/allocator.go index 0d1fceeb..ae0565c3 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -21,8 +21,9 @@ const ( msgRingUpdate msgSpaceRequestDenied - paxosInterval = time.Second * 5 - MinSubnetSize = 4 // first and last addresses are excluded, so 2 would be too small + tickInterval = time.Second * 5 + MinSubnetSize = 4 // first and last addresses are excluded, so 2 would be too small + containerDiedTimeout = time.Second * 30 ) // operation represents something which Allocator wants to do, but @@ -51,9 +52,11 @@ type Allocator struct { nicknames map[mesh.PeerName]string // so we can map nicknames for rmpeer pendingAllocates []operation // held until we get some free space pendingClaims []operation // held until we know who owns the space + dead map[string]time.Time // containers we heard were dead, and when gossip mesh.Gossip // our link to the outside world for sending messages paxos *paxos.Node - paxosTicker *time.Ticker + paxosActive bool + ticker *time.Ticker shuttingDown bool // to avoid doing any requests while trying to shut down isKnownPeer func(mesh.PeerName) bool now func() time.Time @@ -69,6 +72,7 @@ func NewAllocator(ourName mesh.PeerName, ourUID mesh.PeerUID, ourNickname string paxos: paxos.NewNode(ourName, ourUID, quorum), nicknames: map[mesh.PeerName]string{ourName: ourNickname}, isKnownPeer: isKnownPeer, + dead: make(map[string]time.Time), now: time.Now, } } @@ -77,12 +81,14 @@ func NewAllocator(ourName mesh.PeerName, ourUID mesh.PeerUID, ourNickname string func (alloc *Allocator) Start() { actionChan := make(chan func(), mesh.ChannelSize) alloc.actionChan = actionChan + alloc.ticker = time.NewTicker(tickInterval) go alloc.actorLoop(actionChan) } // Stop makes the actor routine exit, for test purposes ONLY because any // calls after this is processed will hang. Async. func (alloc *Allocator) Stop() { + alloc.ticker.Stop() alloc.actionChan <- nil } @@ -220,10 +226,39 @@ func (alloc *Allocator) Claim(ident string, addr address.Address, noErrorOnUnkno return <-resultChan } -// ContainerDied is provided to satisfy the updater interface; does a 'Delete' underneath. Sync. +// ContainerDied called from the updater interface. Async. func (alloc *Allocator) ContainerDied(ident string) { - if err := alloc.Delete(ident); err == nil { - alloc.debugln("Container", ident, "died; released addresses") + alloc.actionChan <- func() { + if _, found := alloc.lookupOwned(ident, alloc.universe); found { + alloc.debugln("Container", ident, "died; noting to remove later") + alloc.dead[ident] = alloc.now() + } + // Also remove any pending ops + alloc.cancelOpsFor(&alloc.pendingAllocates, ident) + alloc.cancelOpsFor(&alloc.pendingClaims, ident) + } +} + +// ContainerDestroyed called from the updater interface. Async. +func (alloc *Allocator) ContainerDestroyed(ident string) { + alloc.actionChan <- func() { + if _, found := alloc.lookupOwned(ident, alloc.universe); found { + alloc.debugln("Container", ident, "destroyed; removing addresses") + alloc.delete(ident) + delete(alloc.dead, ident) + } + } +} + +func (alloc *Allocator) removeDeadContainers() { + cutoff := alloc.now().Add(-containerDiedTimeout) + for ident, timeOfDeath := range alloc.dead { + if timeOfDeath.Before(cutoff) { + if err := alloc.delete(ident); err == nil { + alloc.debugln("Removed addresses for container", ident) + } + delete(alloc.dead, ident) + } } } @@ -233,25 +268,24 @@ func (alloc *Allocator) ContainerStarted(ident string) {} func (alloc *Allocator) Delete(ident string) error { errChan := make(chan error) alloc.actionChan <- func() { - addrs, found := alloc.owned[ident] - for _, addr := range addrs { - alloc.space.Free(addr) - } - delete(alloc.owned, ident) - - // Also remove any pending ops - found = alloc.cancelOpsFor(&alloc.pendingAllocates, ident) || found - found = alloc.cancelOpsFor(&alloc.pendingClaims, ident) || found - - if !found { - errChan <- fmt.Errorf("Delete: no addresses for %s", ident) - return - } - errChan <- nil + errChan <- alloc.delete(ident) } return <-errChan } +func (alloc *Allocator) delete(ident string) error { + addrs, found := alloc.owned[ident] + for _, addr := range addrs { + alloc.space.Free(addr) + } + delete(alloc.owned, ident) + + if !found { + return fmt.Errorf("Delete: no addresses for %s", ident) + } + return nil +} + // Free (Sync) - release single IP address for container func (alloc *Allocator) Free(ident string, addrToFree address.Address) error { errChan := make(chan error) @@ -501,19 +535,18 @@ func (alloc *Allocator) SetInterfaces(gossip mesh.Gossip) { func (alloc *Allocator) actorLoop(actionChan <-chan func()) { for { - var tickChan <-chan time.Time - if alloc.paxosTicker != nil { - tickChan = alloc.paxosTicker.C - } - select { case action := <-actionChan: if action == nil { return } action() - case <-tickChan: - alloc.propose() + case <-alloc.ticker.C: + if alloc.paxosActive { + alloc.propose() + } + alloc.removeDeadContainers() + alloc.tryPendingOps() } alloc.assertInvariants() @@ -525,18 +558,16 @@ func (alloc *Allocator) actorLoop(actionChan <-chan func()) { // Ensure we are making progress towards an established ring func (alloc *Allocator) establishRing() { - if !alloc.ring.Empty() || alloc.paxosTicker != nil { + if !alloc.ring.Empty() || alloc.paxosActive { return } + alloc.paxosActive = true alloc.propose() if ok, cons := alloc.paxos.Consensus(); ok { // If the quorum was 1, then proposing immediately // leads to consensus alloc.createRing(cons.Value) - } else { - // re-try until we get consensus - alloc.paxosTicker = time.NewTicker(paxosInterval) } } @@ -549,13 +580,9 @@ func (alloc *Allocator) createRing(peers []mesh.PeerName) { func (alloc *Allocator) ringUpdated() { // When we have a ring, we don't need paxos any more - if alloc.paxos != nil { + if alloc.paxosActive { + alloc.paxosActive = false alloc.paxos = nil - - if alloc.paxosTicker != nil { - alloc.paxosTicker.Stop() - alloc.paxosTicker = nil - } } alloc.space.UpdateRanges(alloc.ring.OwnedRanges()) diff --git a/ipam/allocator_test.go b/ipam/allocator_test.go index 766e5041..c9eef59c 100644 --- a/ipam/allocator_test.go +++ b/ipam/allocator_test.go @@ -69,8 +69,18 @@ func TestAllocFree(t *testing.T) { require.Equal(t, testAddr2, addr4.String(), "address") alloc.ContainerDied(container2) + + // Resurrect + addr1c, err := alloc.Allocate(container2, cidr1.HostRange(), returnFalse) + require.NoError(t, err) + require.Equal(t, addr1b, addr1c, "address") + alloc.ContainerDied(container3) - require.Equal(t, address.Offset(spaceSize), alloc.NumFreeAddresses(subnet)) + alloc.Encode() // sync up + // Move the clock forward and clear out the dead container + alloc.actionChan <- func() { alloc.now = func() time.Time { return time.Now().Add(containerDiedTimeout * 2) } } + alloc.actionChan <- func() { alloc.removeDeadContainers() } + require.Equal(t, address.Offset(spaceSize-1), alloc.NumFreeAddresses(subnet)) } func TestBootstrap(t *testing.T) { diff --git a/ipam/claim.go b/ipam/claim.go index 20ec6519..0173af6a 100644 --- a/ipam/claim.go +++ b/ipam/claim.go @@ -40,6 +40,9 @@ func (c *claim) Try(alloc *Allocator) bool { alloc.establishRing() + // If we had heard that this container died, resurrect it + delete(alloc.dead, c.ident) // (delete is no-op if key not in map) + switch owner := alloc.ring.Owner(c.addr); owner { case alloc.ourName: // success @@ -78,6 +81,7 @@ func (c *claim) Try(alloc *Allocator) bool { } case c.ident: // same identifier is claiming same address; that's OK + alloc.debugln("Re-Claimed", c.addr, "for", c.ident) c.sendResult(nil) default: // Addr already owned by container on this machine diff --git a/ipam/status.go b/ipam/status.go index e7a252ba..3c9aeb12 100644 --- a/ipam/status.go +++ b/ipam/status.go @@ -36,7 +36,7 @@ func NewStatus(allocator *Allocator, defaultSubnet address.CIDR) *Status { } var paxosStatus *paxos.Status - if allocator.paxosTicker != nil { + if allocator.paxosActive { paxosStatus = paxos.NewStatus(allocator.paxos) } diff --git a/nameserver/nameserver.go b/nameserver/nameserver.go index 201bda3b..ef303728 100644 --- a/nameserver/nameserver.go +++ b/nameserver/nameserver.go @@ -121,7 +121,8 @@ func (n *Nameserver) ReverseLookup(ip address.Address) (string, error) { return match.Hostname, nil } -func (n *Nameserver) ContainerStarted(ident string) {} +func (n *Nameserver) ContainerStarted(ident string) {} +func (n *Nameserver) ContainerDestroyed(ident string) {} func (n *Nameserver) ContainerDied(ident string) { n.Lock() diff --git a/plugin/net/watcher.go b/plugin/net/watcher.go index 05241d4f..935d4ed8 100644 --- a/plugin/net/watcher.go +++ b/plugin/net/watcher.go @@ -55,3 +55,5 @@ func (w *watcher) ContainerDied(id string) { // don't need to do this as WeaveDNS removes names on container died anyway // (note by the time we get this event we can't see the EndpointID) } + +func (w *watcher) ContainerDestroyed(id string) {} diff --git a/proxy/proxy.go b/proxy/proxy.go index c316797e..1045fd51 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -458,8 +458,8 @@ func (proxy *Proxy) waitForStartByIdent(ident string) error { return nil } -func (proxy *Proxy) ContainerDied(ident string) { -} +func (proxy *Proxy) ContainerDied(ident string) {} +func (proxy *Proxy) ContainerDestroyed(ident string) {} // Check if this container needs to be attached, and return nil on success or not needed. func (proxy *Proxy) attach(containerID string, orDie bool) error { diff --git a/site/features.md b/site/features.md index aaa1a43c..b7a91923 100644 --- a/site/features.md +++ b/site/features.md @@ -157,6 +157,33 @@ connect to. The individual IP addresses given to containers must, of course, be unique - if you pick an address that the automatic allocator has already assigned you will receive a warning. +If you restart a container, it will retain the same IP addresses on +the weave network: + + host1$ docker run --name a1 -tdi ubuntu + f76b09a9fcfee04551dbb8d951d9a83e7e7d55126b02fd9f44f9f8a5f07d7c96 + host1$ weave ps a1 + a1 1e:dc:2a:db:ef:ff 10.32.0.3/12 + host1$ docker restart a1 + host1$ weave ps a1 + a1 16:c0:6f:5d:c5:73 10.32.0.3/12 + +It works the same if you stop and immediately restart: + + host1$ docker stop a1 + host1$ docker start a1 + +However, additional addresses added with the `weave attach` command +will not be retained. + +There is also a `weave restart` command, which does re-attach all +current IP addresses: + + host1$ weave restart b1 + +(note that the IP addresses are held for a limited time - currently +30 seconds) + ### Naming and discovery Named containers are automatically registered in diff --git a/test/100_cross_hosts_2_test.sh b/test/100_cross_hosts_2_test.sh index 4ea28f74..0eab1027 100755 --- a/test/100_cross_hosts_2_test.sh +++ b/test/100_cross_hosts_2_test.sh @@ -27,4 +27,12 @@ start_container $HOST2 net:$SUBNET_2 --name=c6 C6=$(container_ip $HOST2 c6) assert_raises "exec_on $HOST1 c5 $PING $C6" +# check that restart retains the same IP, and reclaims it in IPAM +weave_on $HOST2 restart c6 +assert_raises "exec_on $HOST1 c5 $PING $C6" +sleep 31 # past the IPAM timeout +start_container $HOST2 net:$SUBNET_2 --name=c7 +C7=$(container_ip $HOST2 c7) +assert_raises "[ $C6 != $C7 ]" + end_suite diff --git a/test/640_proxy_restart_reattaches_test.sh b/test/640_proxy_restart_reattaches_test.sh index 034d4704..f02889c5 100755 --- a/test/640_proxy_restart_reattaches_test.sh +++ b/test/640_proxy_restart_reattaches_test.sh @@ -2,8 +2,6 @@ . ./config.sh -C1=10.2.0.78 -C2=10.2.0.34 NAME=seetwo.weave.local check_attached() { @@ -23,11 +21,22 @@ wait_for_proxy() { exit 1 } +N=50 +# Create and remove a lot of containers in a small subnet; the failure +# mode is that this takes a long time as it has to wait for the old +# ones to time out, so we run this function inside 'timeout' +run_many() { + for i in $(seq $N); do + proxy docker_on $HOST1 run -e WEAVE_CIDR=net:10.32.4.0/28 --rm -t $SMALL_IMAGE /bin/true + done +} + start_suite "Proxy restart reattaches networking to containers" WEAVE_DOCKER_ARGS=--restart=always WEAVEPROXY_DOCKER_ARGS=--restart=always weave_on $HOST1 launch -proxy_start_container $HOST1 -e WEAVE_CIDR=$C2/24 -di --name=c2 --restart=always -h $NAME -proxy_start_container_with_dns $HOST1 -e WEAVE_CIDR=$C1/24 -di --name=c1 --restart=always +proxy_start_container $HOST1 -di --name=c2 --restart=always -h $NAME +proxy_start_container_with_dns $HOST1 -di --name=c1 --restart=always +C2=$(container_ip $HOST1 c2) proxy docker_on $HOST1 restart -t=1 c2 check_attached @@ -42,11 +51,15 @@ run_on $HOST1 sudo kill -KILL $(docker_on $HOST1 inspect --format='{{.State.Pid} sleep 1 check_attached -# Disabled because Docker 1.8 often fails to restart weave or weaveproxy # Restart docker itself, using different commands for systemd- and upstart-managed. -#run_on $HOST1 sh -c "command -v systemctl >/dev/null && sudo systemctl restart docker || sudo service docker restart" -#wait_for_proxy $HOST1 -#check_attached +run_on $HOST1 sh -c "command -v systemctl >/dev/null && sudo systemctl restart docker || sudo service docker restart" +wait_for_proxy $HOST1 +sleep 5 # allow for re-tries of attach +# Re-fetch the IP since it is not retained on docker restart +C2=$(container_ip $HOST1 c2) || (echo "container c2 has no IP address" 2>&1; exit 1) +check_attached + +assert_raises "timeout 90 cat <( run_many )" # Restarting proxy shouldn't kill unattachable containers proxy_start_container $HOST1 -di --name=c3 --restart=always # Use ipam, so it won't be attachable w/o weave diff --git a/weave b/weave index 3f5f3d87..8128bd44 100755 --- a/weave +++ b/weave @@ -63,6 +63,7 @@ weave run [--without-dns] [--no-rewrite-hosts] [--no-multicast-route] start [ ...] attach [ ...] detach [ ...] + restart weave expose [ ...] [-h ] hide [ ...] @@ -1045,6 +1046,10 @@ echo_ips() { done } +echo_cidrs() { + echo $3 +} + peer_args() { res='' sep='' @@ -2026,6 +2031,19 @@ EOF done show_addrs $ALL_CIDRS ;; + restart) + [ $# -ge 1 ] || usage + create_bridge + ALL_CIDRS=$(with_container_addresses echo_cidrs $1) + RES=$(docker restart "$1") + CONTAINER=$(container_id $1) + for CIDR in $ALL_CIDRS ; do + call_weave PUT /ip/$CONTAINER/${CIDR%/*} + done + with_container_netns_or_die $1 attach $ALL_CIDRS + when_weave_running with_container_fqdn $1 put_dns_fqdn $ALL_CIDRS + echo $RES + ;; dns-add) collect_dns_add_remove_args "$@" FN=put_dns_fqdn