Merge pull request #1191 from weaveworks/1047-same-ip-on-restart

Don't delete record of IP allocation immediately after container death

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)
 			}
 		}
 	}()

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
 	}

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())

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) {

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

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)
 	}
 

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()

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) {}

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 {

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)
+
 ### <a name="naming-and-discovery"></a>Naming and discovery
 
 Named containers are automatically registered in

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

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

weave

@@ -63,6 +63,7 @@ weave run           [--without-dns] [--no-rewrite-hosts] [--no-multicast-route]
       start         [<addr> ...] <container_id>
       attach        [<addr> ...] <container_id>
       detach        [<addr> ...] <container_id>
+      restart       <container_id>
 
 weave expose        [<addr> ...] [-h <fqdn>]
       hide          [<addr> ...]
@@ -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