Merge branch 'master' into 29_unknown_peer

README.md

@@ -294,6 +294,52 @@ then we can connect to it from our application container on $HOST2 with
 
 The same command will work from any application container.
 
+### Service binding
+
+Importing a service provides a degree of indirection that allows late
+and dynamic binding, similar to what can be achieved with a proxy. In
+our example, application containers are unaware that the service they
+are accessing at `10.0.1.101:3322` is in fact residing on
+`$HOST3:2211`. We can point application containers at another service
+location by changing the above NAT rule, without having to alter the
+applications themselves.
+
+### Service routing
+
+The [service export](#service-export) and
+[service import](#service-import) features can be combined to
+establish connectivity between applications and services residing on
+disjoint networks, even when those networks are separated by firewalls
+and might have overlapping IP ranges. Each network imports its
+services into weave, and in turn exports from weave services required
+by its applications. There are no application containers in this
+scenario (though of course there could be); weave is acting purely as
+an address translation and routing facility, using the weave
+application network as an intermediary.
+
+In our example above, the netcat service on $HOST3 is imported into
+weave via $HOST1. We can export it on $HOST2 by first exposing the
+application network with
+
+    host2# weave expose 10.0.1.102/24
+
+and then adding a NAT rule which routes traffic from the $HOST2
+network (i.e. anything which can connect to $HOST2) to the service
+endpoint in the weave network
+
+    host2# iptables -t nat -A PREROUTING -p tcp -i eth0 --dport 4433 \
+           -j DNAT --to-destination 10.0.1.101:3322
+
+Now any host on the same network as $HOST2 can access the service with
+
+    echo 'Hello, world.' | nc $HOST2 4433
+
+Furthermore, as explained in [service-binding](#service-binding), we
+can dynamically alter the service locations without having to touch
+the applications that access them, e.g. we could move the example
+netcat service to $HOST4:2211 while retaining its 10.0.1.101:3322
+endpoint in the weave network.
+
 ### Multi-cloud networking
 
 Weave can network containers hosted in different cloud providers /
@@ -349,6 +395,16 @@ quickly enough.
 
 ## Troubleshooting
 
+Check what version of weave you are running:
+
+    weave version
+
+and compare against
+https://registry.hub.docker.com/u/zettio/weave/tags/manage/.  If you
+do not have the latest released version you can download it with:
+
+    docker pull zettio/weave
+
 Check the weave container logs with
 
     docker logs weave

connection.go

@@ -51,7 +51,7 @@ func (conn *RemoteConnection) String() string {
 
 // Async. Does not return anything. If the connection is successful,
 // it will end up in the local peer's connections map.
-func NewLocalConnection(connRemote *RemoteConnection, expectedName PeerName, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) {
+func NewLocalConnection(connRemote *RemoteConnection, acceptNewPeer bool, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) {
 	if connRemote.local != router.Ourself {
 		log.Fatal("Attempt to create local connection from a peer which is not ourself")
 	}
@@ -65,7 +65,7 @@ func NewLocalConnection(connRemote *RemoteConnection, expectedName PeerName, tcp
 		remoteUDPAddr:    udpAddr,
 		effectivePMTU:    DefaultPMTU,
 		queryChan:        queryChan}
-	go connLocal.queryLoop(queryChan, expectedName)
+	go connLocal.queryLoop(queryChan, acceptNewPeer)
 }
 
 func (conn *LocalConnection) Established() bool {
@@ -161,8 +161,8 @@ func (conn *LocalConnection) SendTCP(msg []byte) {
 
 // ACTOR server
 
-func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction, expectedName PeerName) {
-	err := conn.handshake(expectedName)
+func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction, acceptNewPeer bool) {
+	err := conn.handshake(acceptNewPeer)
 	if err != nil {
 		log.Printf("->[%s] encountered error during handshake: %v\n", conn.remoteTCPAddr, err)
 		conn.handleShutdown()
@@ -278,12 +278,10 @@ func (conn *LocalConnection) handleShutdown() {
 	// try to send any more
 	conn.stopForwarders()
 
-	if conn.remote != nil {
-		conn.Router.ConnectionMaker.EnsureConnection(conn.remote.Name, conn.remoteTCPAddr)
-	}
+	conn.Router.ConnectionMaker.ConnectionTerminated(conn.remoteTCPAddr)
 }
 
-func (conn *LocalConnection) handshake(expectedName PeerName) error {
+func (conn *LocalConnection) handshake(acceptNewPeer bool) error {
 	// We do not need to worry about locking in here as at this point,
 	// the connection is not reachable by any go-routine other than
 	// ourself. Only when we add this connection to the conn.local
@@ -344,8 +342,10 @@ func (conn *LocalConnection) handshake(expectedName PeerName) error {
 	if err != nil {
 		return err
 	}
-	if expectedName != UnknownPeerName && expectedName != name {
-		return fmt.Errorf("Unexpected remote name: expected %s, received %s", expectedName, name)
+	if !acceptNewPeer {
+		if _, found := conn.Router.Peers.Fetch(name); !found {
+			return fmt.Errorf("Found unknown remote name: %s at %s", name, conn.remoteTCPAddr)
+		}
 	}
 
 	uidStr, err := checkHandshakeStringField("UID", "", handshakeRecv)

connection_maker.go

@@ -6,33 +6,41 @@ import (
 	"log"
 	"math/rand"
 	"net"
-	"strconv"
 	"time"
 )
 
 const (
 	InitialInterval = 5 * time.Second
 	MaxInterval     = 10 * time.Minute
-	MaxAttemptCount = 100
-	CMEnsure        = iota
-	CMStatus        = iota
+)
+
+const (
+	CMInitiate   = iota
+	CMStatus     = iota
+	CMTerminated = iota
 )
 
 func StartConnectionMaker(router *Router) *ConnectionMaker {
 	queryChan := make(chan *ConnectionMakerInteraction, ChannelSize)
 	state := &ConnectionMaker{
-		router:            router,
-		queryChan:         queryChan,
-		failedConnections: make(map[PeerName]*FailedConnection)}
+		router:         router,
+		queryChan:      queryChan,
+		cmdLineAddress: make(map[string]bool),
+		targets:        make(map[string]*Target)}
 	go state.queryLoop(queryChan)
 	return state
 }
 
-func (cm *ConnectionMaker) EnsureConnection(name PeerName, foundAt string) {
+func (cm *ConnectionMaker) InitiateConnection(address string) {
 	cm.queryChan <- &ConnectionMakerInteraction{
-		Interaction: Interaction{code: CMEnsure},
-		name:        name,
-		foundAt:     foundAt}
+		Interaction: Interaction{code: CMInitiate},
+		address:     address}
+}
+
+func (cm *ConnectionMaker) ConnectionTerminated(address string) {
+	cm.queryChan <- &ConnectionMakerInteraction{
+		Interaction: Interaction{code: CMTerminated},
+		address:     address}
 }
 
 func (cm *ConnectionMaker) String() string {
@@ -46,7 +54,8 @@ func (cm *ConnectionMaker) String() string {
 func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteraction) {
 	var tick <-chan time.Time
 	maybeTick := func() {
-		if tick == nil && len(cm.failedConnections) > 0 {
+		// would be nice to optimise this to stop ticking when there is nothing worth trying
+		if tick == nil {
 			tick = time.After(5 * time.Second)
 		}
 	}
@@ -57,82 +66,124 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio
 				return
 			}
 			switch {
-			case query.code == CMEnsure:
-				if query.name != cm.router.Ourself.Name {
-					cm.addToFailedConnection(query.name, query.foundAt)
-					maybeTick()
-				}
+			case query.code == CMInitiate:
+				cm.cmdLineAddress[NormalisePeerAddr(query.address)] = true
+				cm.checkStateAndAttemptConnections(time.Now())
+				maybeTick()
 			case query.code == CMStatus:
 				query.resultChan <- cm.status()
+			case query.code == CMTerminated:
+				if target, found := cm.targets[query.address]; found {
+					target.attempting = false
+					target.tryAfter, target.tryInterval = tryAfter(target.tryInterval)
+					maybeTick()
+				}
 			default:
 				log.Fatal("Unexpected connection maker query:", query)
 			}
 		case now := <-tick:
-			for name, failedConn := range cm.failedConnections {
-				if now.After(failedConn.tryAfter) {
-					if _, found := cm.router.Ourself.ConnectionTo(name); found {
-						delete(cm.failedConnections, name)
-						continue
-					} else if failedConn.attemptCount == MaxAttemptCount {
-						delete(cm.failedConnections, name)
-						continue
-					}
-					after, interval := tryAfter(failedConn.tryInterval)
-					failedConn.tryInterval = interval
-					failedConn.tryAfter = after
-					failedConn.attemptCount += 1
-					for target := range failedConn.foundAt {
-						go cm.attemptConnection(target, name)
-					}
-				}
-			}
+			cm.checkStateAndAttemptConnections(now)
 			tick = nil
 			maybeTick()
 		}
 	}
 }
 
-func (cm *ConnectionMaker) addToFailedConnection(name PeerName, foundAt string) {
-	failed := cm.failedConnections[name]
-	if failed == nil {
-		after, interval := tryAfter(InitialInterval)
-		failed = &FailedConnection{
-			foundAt:     make(map[string]bool),
-			tryInterval: interval,
-			tryAfter:    after}
-	}
-	foundAtHost, foundAtPortStr, err := net.SplitHostPort(foundAt)
-	if err == nil {
-		// ensure port-less version is there
-		failed.foundAt[foundAtHost] = true
-		if foundAtPort, err := strconv.Atoi(foundAtPortStr); err == nil && foundAtPort != Port {
-			failed.foundAt[foundAt] = true
+func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) {
+	ourself := cm.router.Ourself
+	validTarget := make(map[string]bool)
+
+	// copy the set of things we are connected to, so we can access them without locking
+	our_connected_peers := make(map[PeerName]bool)
+	our_connected_targets := make(map[string]bool)
+	ourself.ForEachConnection(func(peer PeerName, conn Connection) {
+		our_connected_peers[peer] = true
+		our_connected_targets[conn.RemoteTCPAddr()] = true
+	})
+
+	addTarget := func(address string) {
+		if !our_connected_targets[address] {
+			validTarget[address] = true
+			cm.addTarget(address)
 		}
-	} else {
-		// can't split it, assume it must not have port on it
-		failed.foundAt[foundAt] = true
 	}
-	cm.failedConnections[name] = failed
+
+	// Add command-line targets that are not connected
+	for address, _ := range cm.cmdLineAddress {
+		addTarget(address)
+	}
+
+	// Add targets for peers that someone else is connected to, but we
+	// aren't
+	cm.router.Peers.ForEach(func(name PeerName, peer *Peer) {
+		peer.ForEachConnection(func(otherPeer PeerName, conn Connection) {
+			if otherPeer == ourself.Name || our_connected_peers[otherPeer] {
+				return
+			}
+			address := conn.RemoteTCPAddr()
+			// try both portnumber of connection and standard port
+			addTarget(address)
+			if host, _, err := net.SplitHostPort(address); err == nil {
+				addTarget(NormalisePeerAddr(host))
+			}
+		})
+	})
+
+	now = time.Now() // make sure we catch items just added
+	for address, target := range cm.targets {
+		if our_connected_targets[address] {
+			delete(cm.targets, address)
+			continue
+		}
+		if target.attempting {
+			continue
+		}
+		if !validTarget[address] {
+			delete(cm.targets, address)
+			continue
+		}
+		if now.After(target.tryAfter) {
+			target.attempting = true
+			go cm.attemptConnection(address, cm.cmdLineAddress[address])
+		}
+	}
+}
+
+func (cm *ConnectionMaker) addTarget(address string) {
+	if _, found := cm.targets[address]; !found {
+		target := &Target{}
+		target.tryAfter, target.tryInterval = tryImmediately()
+		cm.targets[address] = target
+	}
 }
 
 func (cm *ConnectionMaker) status() string {
 	var buf bytes.Buffer
-	for name, failedConn := range cm.failedConnections {
-		foundAt := make([]string, 0, len(failedConn.foundAt))
-		for target := range failedConn.foundAt {
-			foundAt = append(foundAt, target)
+	for address, target := range cm.targets {
+		var fmtStr string
+		if target.attempting {
+			fmtStr = "%s (trying since %v)\n"
+		} else {
+			fmtStr = "%s (next try at %v)\n"
 		}
-		buf.WriteString(fmt.Sprintf("%s (%v attempts, next at %v): %v\n", name, failedConn.attemptCount, failedConn.tryAfter, foundAt))
+		buf.WriteString(fmt.Sprintf(fmtStr, address, target.tryAfter))
 	}
 	return buf.String()
 }
 
-func (cm *ConnectionMaker) attemptConnection(foundAt string, targetName PeerName) {
-	if err := cm.router.Ourself.CreateConnection(foundAt, targetName); err != nil {
+func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) {
+	log.Println("Attempting connection to", address)
+	if err := cm.router.Ourself.CreateConnection(address, acceptNewPeer); err != nil {
 		log.Println(err)
+		cm.ConnectionTerminated(address)
 	}
 }
 
+func tryImmediately() (time.Time, time.Duration) {
+	interval := time.Duration(rand.Int63n(int64(InitialInterval)))
+	return time.Now(), interval
+}
+
 func tryAfter(interval time.Duration) (time.Time, time.Duration) {
 	interval += time.Duration(rand.Int63n(int64(interval)))
 	if interval > MaxInterval {

crypto.go

@@ -139,7 +139,7 @@ func (ne *NonEncryptor) AppendFrame(frame *ForwardedFrame) {
 	bufTail = bufTail[2:]
 	copy(bufTail, frame.frame)
 	ne.bufTail = bufTail[frameLen:]
-	ne.buffered = ne.buffered + srcLen + dstLen + 2 + frameLen
+	ne.buffered += srcLen + dstLen + 2 + frameLen
 }
 
 func (ne *NonEncryptor) TotalLen() int {

peer.go

@@ -109,9 +109,6 @@ func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerN
 	defer peer.Unlock()
 	peer.version = version
 	peer.connections = connections
-	for remoteName, conn := range connections {
-		peer.Router.ConnectionMaker.EnsureConnection(remoteName, conn.RemoteTCPAddr())
-	}
 }
 
 func (peer *Peer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error {
@@ -190,18 +187,12 @@ func (peer *Peer) StartLocalPeer() {
 	go peer.queryLoop(queryChan)
 }
 
-func (peer *Peer) CreateConnection(peerAddr string, expectedName PeerName) error {
+func (peer *Peer) CreateConnection(peerAddr string, acceptNewPeer bool) error {
 	if err := peer.checkConnectionLimit(); err != nil {
 		return err
 	}
 	// We're dialing the remote so that means connections will come from random ports
-	_, _, err := net.SplitHostPort(peerAddr)
-	var addrStr string
-	if err == nil {
-		addrStr = peerAddr
-	} else {
-		addrStr = fmt.Sprintf("%s:%d", peerAddr, Port)
-	}
+	addrStr := NormalisePeerAddr(peerAddr)
 	tcpAddr, tcpErr := net.ResolveTCPAddr("tcp4", addrStr)
 	udpAddr, udpErr := net.ResolveUDPAddr("udp4", addrStr)
 	if tcpErr != nil || udpErr != nil {
@@ -216,7 +207,7 @@ func (peer *Peer) CreateConnection(peerAddr string, expectedName PeerName) error
 		return err
 	}
 	connRemote := NewRemoteConnection(peer, nil, tcpConn.RemoteAddr().String())
-	NewLocalConnection(connRemote, expectedName, tcpConn, udpAddr, peer.Router)
+	NewLocalConnection(connRemote, acceptNewPeer, tcpConn, udpAddr, peer.Router)
 	return nil
 }
 

peer_cache.go

@@ -105,8 +105,8 @@ func (cache *PeerCache) String() string {
 	var buf bytes.Buffer
 	cache.ForEach(func(name PeerName, peer *Peer) {
 		buf.WriteString(fmt.Sprint(peer, "\n"))
-		peer.ForEachConnection(func(remoteName PeerName, _ Connection) {
-			buf.WriteString(fmt.Sprintf("   -> %v\n", remoteName))
+		peer.ForEachConnection(func(remoteName PeerName, conn Connection) {
+			buf.WriteString(fmt.Sprintf("   -> %v [%v]\n", remoteName, conn.RemoteTCPAddr()))
 		})
 	})
 	return buf.String()

router.go

@@ -154,7 +154,7 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) {
 	// on Port and we wait for them to send us something on UDP to
 	// start.
 	connRemote := NewRemoteConnection(router.Ourself, nil, tcpConn.RemoteAddr().String())
-	NewLocalConnection(connRemote, UnknownPeerName, tcpConn, nil, router)
+	NewLocalConnection(connRemote, true, tcpConn, nil, router)
 }
 
 func (router *Router) listenUDP(localPort int, po PacketSink) *net.UDPConn {

types.go

@@ -156,22 +156,23 @@ type Topology struct {
 
 type ConnectionMakerInteraction struct {
 	Interaction
-	foundAt string
-	name    PeerName
+	address string
 }
 
 type ConnectionMaker struct {
-	router            *Router
-	queryChan         chan<- *ConnectionMakerInteraction
-	activeConnections map[string]string
-	failedConnections map[PeerName]*FailedConnection
+	router         *Router
+	queryChan      chan<- *ConnectionMakerInteraction
+	targets        map[string]*Target
+	cmdLineAddress map[string]bool
 }
 
-type FailedConnection struct {
-	foundAt      map[string]bool
-	tryAfter     time.Time
-	tryInterval  time.Duration
-	attemptCount int
+type ConnectionState int
+
+// Information about an address where we may find a peer
+type Target struct {
+	attempting   bool          // are we currently attempting to connect there?
+	tryAfter     time.Time     // next time to try this address
+	tryInterval  time.Duration // backoff time on next failure
 }
 
 // UDPSender interface and implementations

utils.go

@@ -97,3 +97,14 @@ func (lop ListOfPeers) Swap(i, j int) {
 func (lop ListOfPeers) Less(i, j int) bool {
 	return lop[i].Name < lop[i].Name
 }
+
+// given an address like '1.2.3.4:567', return the address if it has a port,
+// otherwise return the address with weave's standard port number
+func NormalisePeerAddr(peerAddr string) string {
+	_, _, err := net.SplitHostPort(peerAddr)
+	if err == nil {
+		return peerAddr
+	} else {
+		return fmt.Sprintf("%s:%d", peerAddr, Port)
+	}
+}

weaver/Makefile

@@ -13,6 +13,10 @@ docker-image: /tmp/weave.tar
 	sudo docker build -t zettio/weave .
 	sudo docker save zettio/weave > /tmp/weave.tar
 
+publish: docker-image
+	sudo docker tag zettio/weave zettio/weave:git-`git rev-parse --short=12 HEAD`
+	sudo docker push zettio/weave
+
 clean:
 	-sudo docker rmi zettio/weave
 	rm -f weaver /tmp/weave.tar

weaver/main.go

@@ -124,12 +124,7 @@ func main() {
 	router := weave.NewRouter(iface, ourName, []byte(password), connLimit, bufSz*1024*1024, logFrame)
 	router.Start()
 	for _, peer := range peers {
-		targetPeer := peer
-		go func() {
-			if err := router.Ourself.CreateConnection(targetPeer, weave.UnknownPeerName); err != nil {
-				log.Println(err)
-			}
-		}()
+		router.ConnectionMaker.InitiateConnection(peer)
 	}
 	go handleHttp(router)
 	handleSignals(router)

weaver/weave

@@ -8,6 +8,7 @@ usage() {
     echo "weave expose <ipaddr>/<subnet>"
     echo "weave hide   <ipaddr>/<subnet>"
     echo "weave status"
+    echo "weave version"
     echo "weave stop"
     echo "weave reset"
     exit 1
@@ -20,7 +21,7 @@ command_exists () {
 }
 
 [ `id -u` = 0 ] || {
-    echo "weave must be run as 'root'"
+    echo "weave must be run as 'root'" >&2
     exit 1
 }
 
@@ -103,11 +104,11 @@ stop_masquerading() {
 configure_container_networking() {
     DOCKERPID=$(docker inspect --format='{{ .State.Pid }}' $1)
     [ "$DOCKERPID" = 0 ] && {
-        echo "Docker inspect returned invalid PID 0"
+        echo "Container $1 not running." >&2
         exit 1
     }
     [ "$DOCKERPID" = "<no value>" ] && {
-        echo "Container $1 unknown to Docker."
+        echo "Container $1 unknown to Docker." >&2
         exit 1
     }
     NSPID=$DOCKERPID
@@ -185,6 +186,16 @@ case "$COMMAND" in
             exit 1
         fi
         ;;
+    version)
+        # Try the running container first; if no luck try the image
+        if ! DOCKERIMAGE=$(docker inspect --format='{{ .Image }}' weave 2>/dev/null) ; then
+            if ! DOCKERIMAGE=$(docker inspect --format='{{ .Id }}' zettio/weave 2>/dev/null) ; then
+                echo "Unable to find zettio/weave image." >&2
+                exit 1
+            fi
+        fi
+        docker images --no-trunc | grep $DOCKERIMAGE | grep -v latest | tr -s ' ' | cut -d ' ' -f 2
+        ;;
     run)
         [ $# -gt 0 ] || usage
         IPADDR=$1
@@ -230,7 +241,7 @@ case "$COMMAND" in
         destroy_bridge
         ;;
     *)
-        echo "Unknown weave command '$COMMAND'"
+        echo "Unknown weave command '$COMMAND'" >&2
         usage
         ;;
 esac