From 3c67c3c5f1c5ac39dd7f6ed9c3c0260e7a967c11 Mon Sep 17 00:00:00 2001 From: Murali Reddy Date: Wed, 29 Aug 2018 10:38:55 +0530 Subject: [PATCH] changes differentiate beteween fatal errors (for e.g. ipsec init) from transient errors (heartbeast misses). In case of transient errors OverlayForwarder is marked to be unhealty. When overlay forwarder is unhealty its hept on hold so chooseBest() will skip selecting the forwrder Once OverlayForwarder is considered healthy run chooseBest() again so the forwarder can be selected if its the best forwarder Fixes #1737 --- router/fastdp.go | 20 ++++++++++++++++++-- router/network_overlay.go | 2 ++ router/overlay_switch.go | 34 +++++++++++++++++++++++++++++++++- router/sleeve.go | 5 +++++ 4 files changed, 58 insertions(+), 3 deletions(-) diff --git a/router/fastdp.go b/router/fastdp.go index 3de8eee2..84f5098b 100644 --- a/router/fastdp.go +++ b/router/fastdp.go @@ -582,9 +582,11 @@ type fastDatapathForwarder struct { ackedHeartbeat bool stopChan chan struct{} stopped bool + healthy bool establishedChan chan struct{} errorChan chan error + healthChan chan bool } func (fastdp fastDatapathOverlay) PrepareConnection(params mesh.OverlayConnectionParams) (mesh.OverlayConnection, error) { @@ -622,6 +624,7 @@ func (fastdp fastDatapathOverlay) PrepareConnection(params mesh.OverlayConnectio connUID: params.ConnUID, vxlanVportID: vxlanVportID, sessionKey: params.SessionKey, + healthy: true, remoteAddr: remoteAddr, heartbeatInterval: FastHeartbeat, @@ -629,6 +632,7 @@ func (fastdp fastDatapathOverlay) PrepareConnection(params mesh.OverlayConnectio establishedChan: make(chan struct{}), errorChan: make(chan error, 1), + healthChan: make(chan bool), } return fwd, nil @@ -699,9 +703,12 @@ func (fwd *fastDatapathForwarder) ErrorChannel() <-chan error { return fwd.errorChan } +func (fwd *fastDatapathForwarder) HealthChannel() <-chan bool { + return fwd.healthChan +} + func (fwd *fastDatapathForwarder) doHeartbeats() { var err error - for err == nil { select { case <-fwd.heartbeatTimer.C: @@ -711,7 +718,12 @@ func (fwd *fastDatapathForwarder) doHeartbeats() { fwd.heartbeatTimer.Reset(fwd.heartbeatInterval) case <-fwd.heartbeatTimeout.C: - err = fmt.Errorf("timed out waiting for vxlan heartbeat") + fwd.healthChan <- false + if fwd.healthy { + fwd.healthy = false + } else { + err = fmt.Errorf("timed out waiting for vxlan heartbeat") + } case <-fwd.stopChan: return @@ -808,6 +820,10 @@ func (fwd *fastDatapathForwarder) handleVxlanSpecialPacket(frame []byte, sender // heartbeatTimeout if fwd.heartbeatTimeout != nil { fwd.heartbeatTimeout.Reset(HeartbeatTimeout) + if !fwd.healthy { + fwd.healthy = true + fwd.healthChan <- true + } } } diff --git a/router/network_overlay.go b/router/network_overlay.go index 7ed00e42..8ec2f426 100644 --- a/router/network_overlay.go +++ b/router/network_overlay.go @@ -31,6 +31,8 @@ type OverlayForwarder interface { // Confirm(). The return value nil means the key could not be // handled by this forwarder. Forward(ForwardPacketKey) FlowOp + + HealthChannel() <-chan bool } type NullNetworkOverlay struct{ mesh.NullOverlay } diff --git a/router/overlay_switch.go b/router/overlay_switch.go index 20f7dcbb..04d7288b 100644 --- a/router/overlay_switch.go +++ b/router/overlay_switch.go @@ -142,6 +142,7 @@ type overlaySwitchForwarder struct { alreadyEstablished bool establishedChan chan struct{} errorChan chan error + healthChan chan bool } // A subsidiary forwarder @@ -154,6 +155,9 @@ type subForwarder struct { // closed to tell the forwarder monitor goroutine to stop stopChan chan<- struct{} + + // set to true to skip this forwarder in chooseBest() + onHold bool } // An event from a subsidiary forwarder @@ -166,6 +170,9 @@ type subForwarderEvent struct { // is this an error event? err error + + // event to indicate if forwarder is in healthy state to be considered for best forwarded + healthy bool } func (osw *OverlaySwitch) PrepareConnection(params mesh.OverlayConnectionParams) (mesh.OverlayConnection, error) { @@ -239,6 +246,7 @@ func (osw *OverlaySwitch) PrepareConnection(params mesh.OverlayConnectionParams) func monitorForwarder(index int, eventsChan chan<- subForwarderEvent, stopChan <-chan struct{}, fwd OverlayForwarder) { establishedChan := fwd.EstablishedChannel() + healthChan := fwd.HealthChannel() loop: for { e := subForwarderEvent{index: index} @@ -248,6 +256,9 @@ loop: e.established = true establishedChan = nil + case healthy := <-healthChan: + e.healthy = healthy + case err := <-fwd.ErrorChannel(): e.err = err @@ -282,6 +293,8 @@ loop: fwd.established(e.index) case e.err != nil: fwd.error(e.index, e.err) + default: + fwd.healthCheck(e.index, e.healthy) } } } @@ -318,6 +331,21 @@ func (fwd *overlaySwitchForwarder) error(index int, err error) { fwd.chooseBest() } +func (fwd *overlaySwitchForwarder) healthCheck(index int, healthy bool) { + fwd.lock.Lock() + defer fwd.lock.Unlock() + + if healthy == true && fwd.forwarders[index].onHold { + fwd.forwarders[index].onHold = false + fwd.chooseBest() + } + + if healthy == false && !fwd.forwarders[index].onHold { + fwd.forwarders[index].onHold = true + fwd.chooseBest() + } +} + func (fwd *overlaySwitchForwarder) stopFrom(index int) { for index < len(fwd.forwarders) { subFwd := &fwd.forwarders[index] @@ -337,7 +365,7 @@ func (fwd *overlaySwitchForwarder) chooseBest() { for i := range fwd.forwarders { subFwd := &fwd.forwarders[i] - if subFwd.fwd == nil { + if subFwd.fwd == nil || subFwd.onHold { continue } @@ -415,6 +443,10 @@ func (fwd *overlaySwitchForwarder) ErrorChannel() <-chan error { return fwd.errorChan } +func (fwd *overlaySwitchForwarder) HealthChannel() <-chan bool { + return fwd.healthChan +} + func (fwd *overlaySwitchForwarder) Stop() { fwd.lock.Lock() defer fwd.lock.Unlock() diff --git a/router/sleeve.go b/router/sleeve.go index 2c6e507d..4cda180c 100644 --- a/router/sleeve.go +++ b/router/sleeve.go @@ -326,6 +326,7 @@ type sleeveForwarder struct { // listener channels establishedChan chan struct{} errorChan chan error + healthChan chan bool // Explicitly locked state lock sync.RWMutex @@ -445,6 +446,10 @@ func (fwd *sleeveForwarder) ErrorChannel() <-chan error { return fwd.errorChan } +func (fwd *sleeveForwarder) HealthChannel() <-chan bool { + return fwd.healthChan +} + type curriedForward struct { NonDiscardingFlowOp fwd *sleeveForwarder