Merge pull request #764 from paulbellamy/status-connection-maker

Add "Direct Peers" to weave status

Closes #659. Closes #660.

router/connection_maker.go

@@ -15,12 +15,12 @@ const (
 )
 
 type ConnectionMaker struct {
-	ourself      *LocalPeer
-	peers        *Peers
-	port         int
-	targets      map[string]*Target
-	cmdLinePeers map[string]*net.TCPAddr
-	actionChan   chan<- ConnectionMakerAction
+	ourself     *LocalPeer
+	peers       *Peers
+	port        int
+	targets     map[string]*Target
+	directPeers map[string]*net.TCPAddr
+	actionChan  chan<- ConnectionMakerAction
 }
 
 // Information about an address where we may find a peer
@@ -35,11 +35,11 @@ type ConnectionMakerAction func() bool
 
 func NewConnectionMaker(ourself *LocalPeer, peers *Peers, port int) *ConnectionMaker {
 	return &ConnectionMaker{
-		ourself:      ourself,
-		peers:        peers,
-		port:         port,
-		cmdLinePeers: make(map[string]*net.TCPAddr),
-		targets:      make(map[string]*Target)}
+		ourself:     ourself,
+		peers:       peers,
+		port:        port,
+		directPeers: make(map[string]*net.TCPAddr),
+		targets:     make(map[string]*Target)}
 }
 
 func (cm *ConnectionMaker) Start() {
@@ -65,7 +65,7 @@ func (cm *ConnectionMaker) InitiateConnections(peers []string) []error {
 	}
 	cm.actionChan <- func() bool {
 		for peer, addr := range addrs {
-			cm.cmdLinePeers[peer] = addr
+			cm.directPeers[peer] = addr
 			// curtail any existing reconnect interval
 			if target, found := cm.targets[addr.String()]; found {
 				target.tryAfter, target.tryInterval = tryImmediately()
@@ -79,7 +79,7 @@ func (cm *ConnectionMaker) InitiateConnections(peers []string) []error {
 func (cm *ConnectionMaker) ForgetConnections(peers []string) {
 	cm.actionChan <- func() bool {
 		for _, peer := range peers {
-			delete(cm.cmdLinePeers, peer)
+			delete(cm.directPeers, peer)
 		}
 		return false
 	}
@@ -100,33 +100,59 @@ func (cm *ConnectionMaker) Refresh() {
 	cm.actionChan <- func() bool { return true }
 }
 
-func (cm *ConnectionMaker) String() string {
+func (cm *ConnectionMaker) Status() ConnectionMakerStatus {
 	// We need to Refresh first in order to clear out any 'attempting'
 	// connections from cm.targets that have been established since
 	// the last run of cm.checkStateAndAttemptConnections. These
 	// entries are harmless but do represent stale state that we do
 	// not want to report.
 	cm.Refresh()
-	resultChan := make(chan string, 0)
+	resultChan := make(chan ConnectionMakerStatus, 0)
 	cm.actionChan <- func() bool {
-		var buf bytes.Buffer
-		for address, target := range cm.targets {
-			fmt.Fprintf(&buf, "->[%s]", address)
-			if target.lastError != nil {
-				fmt.Fprintf(&buf, " (%s)", target.lastError)
-			}
-			if target.attempting {
-				fmt.Fprintf(&buf, " trying since %v\n", target.tryAfter)
-			} else {
-				fmt.Fprintf(&buf, " next try at %v\n", target.tryAfter)
-			}
+		status := ConnectionMakerStatus{
+			DirectPeers: []string{},
+			Reconnects:  make(map[string]Target),
 		}
-		resultChan <- buf.String()
+		for peer := range cm.directPeers {
+			status.DirectPeers = append(status.DirectPeers, peer)
+		}
+
+		for address, target := range cm.targets {
+			status.Reconnects[address] = *target
+		}
+		resultChan <- status
 		return false
 	}
 	return <-resultChan
 }
 
+type ConnectionMakerStatus struct {
+	DirectPeers []string
+	Reconnects  map[string]Target
+}
+
+func (status ConnectionMakerStatus) String() string {
+	var buf bytes.Buffer
+	fmt.Fprintln(&buf, "Direct Peers:")
+	for _, peer := range status.DirectPeers {
+		fmt.Fprintln(&buf, peer)
+	}
+
+	fmt.Fprintln(&buf, "Reconnects:")
+	for address, target := range status.Reconnects {
+		fmt.Fprintf(&buf, "->[%s]", address)
+		if target.lastError != nil {
+			fmt.Fprintf(&buf, " (%s)", target.lastError)
+		}
+		if target.attempting {
+			fmt.Fprintf(&buf, " trying since %v\n", target.tryAfter)
+		} else {
+			fmt.Fprintf(&buf, " next try at %v\n", target.tryAfter)
+		}
+	}
+	return buf.String()
+}
+
 func (cm *ConnectionMaker) queryLoop(actionChan <-chan ConnectionMakerAction) {
 	timer := time.NewTimer(MaxDuration)
 	run := func() { timer.Reset(cm.checkStateAndAttemptConnections()) }
@@ -144,8 +170,8 @@ func (cm *ConnectionMaker) queryLoop(actionChan <-chan ConnectionMakerAction) {
 
 func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration {
 	var (
-		validTarget   = make(map[string]struct{})
-		cmdLineTarget = make(map[string]struct{})
+		validTarget  = make(map[string]struct{})
+		directTarget = make(map[string]struct{})
 	)
 	// Copy the set of things we are connected to, so we can access
 	// them without locking.  Also clear out any entries in cm.targets
@@ -165,8 +191,8 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration {
 		cm.targets[address] = target
 	}
 
-	// Add command-line targets that are not connected
-	for _, addr := range cm.cmdLinePeers {
+	// Add direct targets that are not connected
+	for _, addr := range cm.directPeers {
 		completeAddr := *addr
 		attempt := true
 		if completeAddr.Port == 0 {
@@ -179,7 +205,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration {
 			}
 		}
 		address := completeAddr.String()
-		cmdLineTarget[address] = void
+		directTarget[address] = void
 		if attempt {
 			addTarget(address)
 		}
@@ -189,7 +215,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration {
 	// aren't
 	cm.addPeerTargets(ourConnectedPeers, addTarget)
 
-	return cm.connectToTargets(validTarget, cmdLineTarget)
+	return cm.connectToTargets(validTarget, directTarget)
 }
 
 func (cm *ConnectionMaker) ourConnections() (PeerNameSet, map[string]struct{}, map[string]struct{}) {
@@ -242,7 +268,7 @@ func (cm *ConnectionMaker) addPeerTargets(ourConnectedPeers PeerNameSet, addTarg
 	})
 }
 
-func (cm *ConnectionMaker) connectToTargets(validTarget map[string]struct{}, cmdLineTarget map[string]struct{}) time.Duration {
+func (cm *ConnectionMaker) connectToTargets(validTarget map[string]struct{}, directTarget map[string]struct{}) time.Duration {
 	now := time.Now() // make sure we catch items just added
 	after := MaxDuration
 	for address, target := range cm.targets {
@@ -256,7 +282,7 @@ func (cm *ConnectionMaker) connectToTargets(validTarget map[string]struct{}, cmd
 		switch duration := target.tryAfter.Sub(now); {
 		case duration <= 0:
 			target.attempting = true
-			_, isCmdLineTarget := cmdLineTarget[address]
+			_, isCmdLineTarget := directTarget[address]
 			go cm.attemptConnection(address, isCmdLineTarget)
 		case duration < after:
 			after = duration

router/json.go

@@ -16,8 +16,8 @@ func (router *Router) StatusJSON(version, encryption string) ([]byte, error) {
 		Macs       *MacCache
 		Peers      *Peers
 		Routes     *Routes
-	}{version, encryption, router.Ourself.Name.String(), router.Ourself.NickName, fmt.Sprintf("%v", router.Iface), router.Macs, router.Peers, router.Routes})
-	// leaving out ConectionMaker due to async complexities
+		ConnectionMakerStatus
+	}{version, encryption, router.Ourself.Name.String(), router.Ourself.NickName, fmt.Sprintf("%v", router.Iface), router.Macs, router.Peers, router.Routes, router.ConnectionMaker.Status()})
 }
 
 func (cache *MacCache) MarshalJSON() ([]byte, error) {
@@ -96,3 +96,20 @@ func (conn *RemoteConnection) MarshalJSON() ([]byte, error) {
 func (name PeerName) MarshalJSON() ([]byte, error) {
 	return json.Marshal(name.String())
 }
+
+func (target Target) MarshalJSON() ([]byte, error) {
+	t := struct {
+		Attempting  bool      `json:"Attempting,omitempty"`
+		TryAfter    time.Time `json:"TryAfter,omitempty"`
+		TryInterval string    `json:"TryInterval,omitempty"`
+		LastError   string    `json:"LastError,omitempty"`
+	}{
+		Attempting:  target.attempting,
+		TryAfter:    target.tryAfter,
+		TryInterval: target.tryInterval.String(),
+	}
+	if target.lastError != nil {
+		t.LastError = target.lastError.Error()
+	}
+	return json.Marshal(t)
+}

router/router.go

@@ -2,7 +2,6 @@ package router
 
 import (
 	"bytes"
-	"code.google.com/p/gopacket/layers"
 	"encoding/binary"
 	"fmt"
 	"io"
@@ -10,6 +9,8 @@ import (
 	"net"
 	"syscall"
 	"time"
+
+	"code.google.com/p/gopacket/layers"
 )
 
 const macMaxAge = 10 * time.Minute // [1]
@@ -110,7 +111,7 @@ func (router *Router) Status() string {
 	fmt.Fprintf(&buf, "MACs:\n%s", router.Macs)
 	fmt.Fprintf(&buf, "Peers:\n%s", router.Peers)
 	fmt.Fprintf(&buf, "Routes:\n%s", router.Routes)
-	fmt.Fprintf(&buf, "Reconnects:\n%s", router.ConnectionMaker)
+	fmt.Fprint(&buf, router.ConnectionMaker.Status())
 	return buf.String()
 }
 

test/150_connect_forget_test.sh

@@ -5,6 +5,20 @@
 C1=10.2.1.4
 C2=10.2.1.7
 
+direct_peers() {
+  weave_on $1 status | awk "/^Direct Peers:$/{f=1;next} /Reconnects:/{f=0} f"
+}
+
+assert_connected() {
+  assert_raises "direct_peers $HOST2 | grep '^$HOST1$'"
+  assert_raises "exec_on $HOST1 c1 $PING $C2"
+}
+
+assert_disconnected() {
+  assert "direct_peers $HOST2" ""
+  assert_raises "exec_on $HOST1 c1 sh -c '! $PING $C2'"
+}
+
 start_suite "Connecting and forgetting routers after launch"
 
 weave_on $HOST1 launch
@@ -13,14 +27,14 @@ weave_on $HOST2 launch
 start_container $HOST1 $C1/24 --name=c1
 start_container $HOST2 $C2/24 --name=c2
 
-assert_raises "exec_on $HOST1 c1 $PING $C2" 1
+assert_disconnected
 assert_raises "weave_on $HOST2 connect $HOST1 $HOST2"
-assert_raises "exec_on $HOST1 c1 $PING $C2"
+assert_connected
 
 assert_raises "weave_on $HOST2 forget $HOST1 $HOST2"
 assert_raises "weave_on $HOST1 stop"
 assert_raises "weave_on $HOST1 launch"
 
-assert_raises "exec_on $HOST1 c1 $PING $C2" 1
+assert_disconnected
 
 end_suite