diff --git a/README.md b/README.md index 7d661868..986f259f 100644 --- a/README.md +++ b/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 diff --git a/connection.go b/connection.go index 08168870..c08bd5f0 100644 --- a/connection.go +++ b/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) diff --git a/connection_maker.go b/connection_maker.go index 94d1cf3c..8c259bd5 100644 --- a/connection_maker.go +++ b/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 { diff --git a/crypto.go b/crypto.go index 14c868ae..4f47f0bb 100644 --- a/crypto.go +++ b/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 { diff --git a/peer.go b/peer.go index b01a3053..b62e8b27 100644 --- a/peer.go +++ b/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 } diff --git a/peer_cache.go b/peer_cache.go index 69e785bb..d9910c95 100644 --- a/peer_cache.go +++ b/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() diff --git a/router.go b/router.go index 39a7bac6..ec2d9de1 100644 --- a/router.go +++ b/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 { diff --git a/types.go b/types.go index 83daed99..c4807582 100644 --- a/types.go +++ b/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 diff --git a/utils.go b/utils.go index 72b8629d..6979b158 100644 --- a/utils.go +++ b/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) + } +} diff --git a/weaver/Makefile b/weaver/Makefile index e4f6cbd7..96a251a4 100644 --- a/weaver/Makefile +++ b/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 diff --git a/weaver/main.go b/weaver/main.go index e11f1c49..63bb60d1 100644 --- a/weaver/main.go +++ b/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) diff --git a/weaver/weave b/weaver/weave index 189be878..8ba2b3ac 100755 --- a/weaver/weave +++ b/weaver/weave @@ -8,6 +8,7 @@ usage() { echo "weave expose /" echo "weave hide /" 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" = "" ] && { - 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