From d16395b8b98cd3a83391b9e4fd0f32128af5e502 Mon Sep 17 00:00:00 2001 From: Paul Bellamy Date: Thu, 21 May 2015 14:50:30 +0100 Subject: [PATCH 1/3] weave status should include 'Direct Peers' set with 'weave connect' --- router/connection_maker.go | 48 ++++++++++++++++++--------------- router/router.go | 5 ++-- test/150_connect_forget_test.sh | 20 +++++++++++--- 3 files changed, 47 insertions(+), 26 deletions(-) diff --git a/router/connection_maker.go b/router/connection_maker.go index 64db9e6b..aaa6a17a 100644 --- a/router/connection_maker.go +++ b/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() { @@ -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,7 +100,7 @@ func (cm *ConnectionMaker) Refresh() { cm.actionChan <- func() bool { return true } } -func (cm *ConnectionMaker) String() string { +func (cm *ConnectionMaker) Status() string { // 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 @@ -110,6 +110,12 @@ func (cm *ConnectionMaker) String() string { resultChan := make(chan string, 0) cm.actionChan <- func() bool { var buf bytes.Buffer + fmt.Fprintln(&buf, "Direct Peers:") + for peer := range cm.directPeers { + fmt.Fprintln(&buf, peer) + } + + fmt.Fprintln(&buf, "Reconnects:") for address, target := range cm.targets { fmt.Fprintf(&buf, "->[%s]", address) if target.lastError != nil { @@ -144,8 +150,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 +171,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 +185,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration { } } address := completeAddr.String() - cmdLineTarget[address] = void + directTarget[address] = void if attempt { addTarget(address) } @@ -189,7 +195,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 +248,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 +262,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 diff --git a/router/router.go b/router/router.go index ce479394..db731591 100644 --- a/router/router.go +++ b/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() } diff --git a/test/150_connect_forget_test.sh b/test/150_connect_forget_test.sh index 63777282..384488c4 100755 --- a/test/150_connect_forget_test.sh +++ b/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 From 637215fcff41c91cbc540308b7f6f227e04ffb94 Mon Sep 17 00:00:00 2001 From: Paul Bellamy Date: Fri, 22 May 2015 15:33:51 +0100 Subject: [PATCH 2/3] add status of connection-maker to status-json --- router/connection_maker.go | 78 ++++++++++++++++++++++++-------------- router/json.go | 4 +- 2 files changed, 51 insertions(+), 31 deletions(-) diff --git a/router/connection_maker.go b/router/connection_maker.go index aaa6a17a..adb67258 100644 --- a/router/connection_maker.go +++ b/router/connection_maker.go @@ -25,10 +25,10 @@ type ConnectionMaker struct { // Information about an address where we may find a peer type Target struct { - attempting bool // are we currently attempting to connect there? - lastError error // reason for disconnection last time - tryAfter time.Time // next time to try this address - tryInterval time.Duration // backoff time on next failure + Attempting bool `json:"Attempting,omitempty"` // are we currently attempting to connect there? + LastError error `json:"LastError,omitempty"` // reason for disconnection last time + TryAfter time.Time `json:"TryAfter,omitempty"` // next time to try this address + TryInterval time.Duration `json:"TryInterval,omitempty"` // backoff time on next failure } type ConnectionMakerAction func() bool @@ -65,10 +65,10 @@ 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() + target.TryAfter, target.TryInterval = tryImmediately() } } return true @@ -88,9 +88,9 @@ func (cm *ConnectionMaker) ForgetConnections(peers []string) { func (cm *ConnectionMaker) ConnectionTerminated(address string, err error) { cm.actionChan <- func() bool { if target, found := cm.targets[address]; found { - target.attempting = false - target.lastError = err - target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) + target.Attempting = false + target.LastError = err + target.TryAfter, target.TryInterval = tryAfter(target.TryInterval) } return true } @@ -100,39 +100,59 @@ func (cm *ConnectionMaker) Refresh() { cm.actionChan <- func() bool { return true } } -func (cm *ConnectionMaker) Status() 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 - fmt.Fprintln(&buf, "Direct Peers:") + status := ConnectionMakerStatus{ + DirectPeers: []string{}, + Reconnects: make(map[string]Target), + } for peer := range cm.directPeers { - fmt.Fprintln(&buf, peer) + status.DirectPeers = append(status.DirectPeers, peer) } - fmt.Fprintln(&buf, "Reconnects:") 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.Reconnects[address] = *target } - resultChan <- buf.String() + 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()) } @@ -167,7 +187,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration { return } target := &Target{} - target.tryAfter, target.tryInterval = tryImmediately() + target.TryAfter, target.TryInterval = tryImmediately() cm.targets[address] = target } @@ -252,16 +272,16 @@ func (cm *ConnectionMaker) connectToTargets(validTarget map[string]struct{}, dir now := time.Now() // make sure we catch items just added after := MaxDuration for address, target := range cm.targets { - if target.attempting { + if target.Attempting { continue } if _, valid := validTarget[address]; !valid { delete(cm.targets, address) continue } - switch duration := target.tryAfter.Sub(now); { + switch duration := target.TryAfter.Sub(now); { case duration <= 0: - target.attempting = true + target.Attempting = true _, isCmdLineTarget := directTarget[address] go cm.attemptConnection(address, isCmdLineTarget) case duration < after: diff --git a/router/json.go b/router/json.go index af48a012..1af7a50d 100644 --- a/router/json.go +++ b/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) { From 998130fa9eaf26188a247317e106c597abd685bb Mon Sep 17 00:00:00 2001 From: Paul Bellamy Date: Wed, 27 May 2015 16:47:54 +0100 Subject: [PATCH 3/3] Reformatting connection maker target JSON output So `LastError` is forced to a string, not an object. --- router/connection_maker.go | 34 +++++++++++++++++----------------- router/json.go | 17 +++++++++++++++++ 2 files changed, 34 insertions(+), 17 deletions(-) diff --git a/router/connection_maker.go b/router/connection_maker.go index adb67258..2208e841 100644 --- a/router/connection_maker.go +++ b/router/connection_maker.go @@ -25,10 +25,10 @@ type ConnectionMaker struct { // Information about an address where we may find a peer type Target struct { - Attempting bool `json:"Attempting,omitempty"` // are we currently attempting to connect there? - LastError error `json:"LastError,omitempty"` // reason for disconnection last time - TryAfter time.Time `json:"TryAfter,omitempty"` // next time to try this address - TryInterval time.Duration `json:"TryInterval,omitempty"` // backoff time on next failure + attempting bool // are we currently attempting to connect there? + lastError error // reason for disconnection last time + tryAfter time.Time // next time to try this address + tryInterval time.Duration // backoff time on next failure } type ConnectionMakerAction func() bool @@ -68,7 +68,7 @@ func (cm *ConnectionMaker) InitiateConnections(peers []string) []error { cm.directPeers[peer] = addr // curtail any existing reconnect interval if target, found := cm.targets[addr.String()]; found { - target.TryAfter, target.TryInterval = tryImmediately() + target.tryAfter, target.tryInterval = tryImmediately() } } return true @@ -88,9 +88,9 @@ func (cm *ConnectionMaker) ForgetConnections(peers []string) { func (cm *ConnectionMaker) ConnectionTerminated(address string, err error) { cm.actionChan <- func() bool { if target, found := cm.targets[address]; found { - target.Attempting = false - target.LastError = err - target.TryAfter, target.TryInterval = tryAfter(target.TryInterval) + target.attempting = false + target.lastError = err + target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) } return true } @@ -141,13 +141,13 @@ func (status ConnectionMakerStatus) String() string { 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.lastError != nil { + fmt.Fprintf(&buf, " (%s)", target.lastError) } - if target.Attempting { - fmt.Fprintf(&buf, " trying since %v\n", target.TryAfter) + if target.attempting { + fmt.Fprintf(&buf, " trying since %v\n", target.tryAfter) } else { - fmt.Fprintf(&buf, " next try at %v\n", target.TryAfter) + fmt.Fprintf(&buf, " next try at %v\n", target.tryAfter) } } return buf.String() @@ -187,7 +187,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration { return } target := &Target{} - target.TryAfter, target.TryInterval = tryImmediately() + target.tryAfter, target.tryInterval = tryImmediately() cm.targets[address] = target } @@ -272,16 +272,16 @@ func (cm *ConnectionMaker) connectToTargets(validTarget map[string]struct{}, dir now := time.Now() // make sure we catch items just added after := MaxDuration for address, target := range cm.targets { - if target.Attempting { + if target.attempting { continue } if _, valid := validTarget[address]; !valid { delete(cm.targets, address) continue } - switch duration := target.TryAfter.Sub(now); { + switch duration := target.tryAfter.Sub(now); { case duration <= 0: - target.Attempting = true + target.attempting = true _, isCmdLineTarget := directTarget[address] go cm.attemptConnection(address, isCmdLineTarget) case duration < after: diff --git a/router/json.go b/router/json.go index 1af7a50d..786e0d36 100644 --- a/router/json.go +++ b/router/json.go @@ -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) +}