From 3c67c3c5f1c5ac39dd7f6ed9c3c0260e7a967c11 Mon Sep 17 00:00:00 2001 From: Murali Reddy Date: Wed, 29 Aug 2018 10:38:55 +0530 Subject: [PATCH 1/4] 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 From 62758ca337b033453bba611d1ceefaa0128f6f2c Mon Sep 17 00:00:00 2001 From: Murali Reddy Date: Thu, 30 Aug 2018 14:47:02 +0530 Subject: [PATCH 2/4] log the heartbeat send/recieve/miss --- router/fastdp.go | 13 ++++++++++++- router/overlay_switch.go | 2 ++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/router/fastdp.go b/router/fastdp.go index 84f5098b..439c6c31 100644 --- a/router/fastdp.go +++ b/router/fastdp.go @@ -713,12 +713,18 @@ func (fwd *fastDatapathForwarder) doHeartbeats() { select { case <-fwd.heartbeatTimer.C: if fwd.confirmed { + log.Debug(fwd.logPrefix(), "sending Heartbeat to peer") fwd.sendHeartbeat() } fwd.heartbeatTimer.Reset(fwd.heartbeatInterval) case <-fwd.heartbeatTimeout.C: + log.Debug(fwd.logPrefix(), "missed Heartbeat from peer, marking fastdp forwarder as un-healthy") + + // treat missed heartbeat as transient error. Indicate to overlay forwarder + // that Forwarder is un-healthy so it can pick next best forwarder fwd.healthChan <- false + if fwd.healthy { fwd.healthy = false } else { @@ -756,7 +762,6 @@ func (fwd *fastDatapathForwarder) handleError(err error) { func (fwd *fastDatapathForwarder) sendHeartbeat() { fwd.lock.RLock() - log.Debug(fwd.logPrefix(), "sendHeartbeat") // the heartbeat payload consists of the 64-bit connection uid // followed by the 16-bit packet size. @@ -813,7 +818,10 @@ func (fwd *fastDatapathForwarder) handleVxlanSpecialPacket(frame []byte, sender if !fwd.ackedHeartbeat { fwd.ackedHeartbeat = true + log.Debug(fwd.logPrefix(), "Ack Heartbeat from peer") fwd.handleError(fwd.sendControlMsg(FastDatapathHeartbeatAck, nil)) + } else { + log.Debug(fwd.logPrefix(), "Got Heartbeat Ack from peer") } // we can receive a heartbeat before Confirm() has set up @@ -821,6 +829,9 @@ func (fwd *fastDatapathForwarder) handleVxlanSpecialPacket(frame []byte, sender if fwd.heartbeatTimeout != nil { fwd.heartbeatTimeout.Reset(HeartbeatTimeout) if !fwd.healthy { + // If fastdp was marked as unhealthy earlier due to missed heartbeat then indicate to + // overlay forwarder that Forwarder is healthy now so it can pick next best forwarder + log.Debug(fwd.logPrefix(), "got Heartbeat from peer, marking fastdp forwarder as healthy") fwd.healthy = true fwd.healthChan <- true } diff --git a/router/overlay_switch.go b/router/overlay_switch.go index 04d7288b..b7268846 100644 --- a/router/overlay_switch.go +++ b/router/overlay_switch.go @@ -336,11 +336,13 @@ func (fwd *overlaySwitchForwarder) healthCheck(index int, healthy bool) { defer fwd.lock.Unlock() if healthy == true && fwd.forwarders[index].onHold { + log.Debug(fwd.logPrefix(), "Adding "+fwd.forwarders[index].overlayName+" to the list of forwarders") fwd.forwarders[index].onHold = false fwd.chooseBest() } if healthy == false && !fwd.forwarders[index].onHold { + log.Debug(fwd.logPrefix(), "Removing "+fwd.forwarders[index].overlayName+" from the list of forwarders") fwd.forwarders[index].onHold = true fwd.chooseBest() } From b3cb38eda4b5bed60b4828c1e3fb7f9590b3ba40 Mon Sep 17 00:00:00 2001 From: Murali Reddy Date: Mon, 24 Sep 2018 17:53:28 +0530 Subject: [PATCH 3/4] addressing review comments --- router/fastdp.go | 23 ++++++++++++++++++----- router/overlay_switch.go | 11 +++++------ 2 files changed, 23 insertions(+), 11 deletions(-) diff --git a/router/fastdp.go b/router/fastdp.go index 439c6c31..b9032e89 100644 --- a/router/fastdp.go +++ b/router/fastdp.go @@ -583,10 +583,10 @@ type fastDatapathForwarder struct { stopChan chan struct{} stopped bool healthy bool - - establishedChan chan struct{} - errorChan chan error - healthChan chan bool + established bool + establishedChan chan struct{} + errorChan chan error + healthChan chan bool } func (fastdp fastDatapathOverlay) PrepareConnection(params mesh.OverlayConnectionParams) (mesh.OverlayConnection, error) { @@ -725,6 +725,15 @@ func (fwd *fastDatapathForwarder) doHeartbeats() { // that Forwarder is un-healthy so it can pick next best forwarder fwd.healthChan <- false + // switch from fast-heartbeat to slow-heartbeat to avoid aggressive heartbeats + // when there is no estbalished session with peer yet + if fwd.heartbeatInterval != SlowHeartbeat { + fwd.heartbeatInterval = SlowHeartbeat + if fwd.heartbeatTimer != nil { + fwd.heartbeatTimer.Reset(fwd.heartbeatInterval) + } + } + if fwd.healthy { fwd.healthy = false } else { @@ -860,8 +869,12 @@ func (fwd *fastDatapathForwarder) Attrs() map[string]interface{} { func (fwd *fastDatapathForwarder) handleHeartbeatAck() { log.Debug(fwd.logPrefix(), "handleHeartbeatAck") - if fwd.heartbeatInterval != SlowHeartbeat { + if !fwd.established { close(fwd.establishedChan) + fwd.established = true + } + + if fwd.heartbeatInterval != SlowHeartbeat { fwd.heartbeatInterval = SlowHeartbeat if fwd.heartbeatTimer != nil { fwd.heartbeatTimer.Reset(fwd.heartbeatInterval) diff --git a/router/overlay_switch.go b/router/overlay_switch.go index b7268846..7fcd564d 100644 --- a/router/overlay_switch.go +++ b/router/overlay_switch.go @@ -172,7 +172,7 @@ type subForwarderEvent struct { err error // event to indicate if forwarder is in healthy state to be considered for best forwarded - healthy bool + healthy *bool } func (osw *OverlaySwitch) PrepareConnection(params mesh.OverlayConnectionParams) (mesh.OverlayConnection, error) { @@ -246,7 +246,6 @@ 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} @@ -256,8 +255,8 @@ loop: e.established = true establishedChan = nil - case healthy := <-healthChan: - e.healthy = healthy + case healthy := <-fwd.HealthChannel(): + e.healthy = &healthy case err := <-fwd.ErrorChannel(): e.err = err @@ -293,8 +292,8 @@ loop: fwd.established(e.index) case e.err != nil: fwd.error(e.index, e.err) - default: - fwd.healthCheck(e.index, e.healthy) + case e.healthy != nil: + fwd.healthCheck(e.index, *e.healthy) } } } From 6b2631e37851d8e11b012a44067e7ad8e122b2d6 Mon Sep 17 00:00:00 2001 From: Murali Reddy Date: Fri, 5 Oct 2018 09:47:23 +0530 Subject: [PATCH 4/4] addressing review comment. keep forwarder running forever even on heart beat timeouts --- router/fastdp.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/router/fastdp.go b/router/fastdp.go index b9032e89..4ecefee9 100644 --- a/router/fastdp.go +++ b/router/fastdp.go @@ -736,8 +736,6 @@ func (fwd *fastDatapathForwarder) doHeartbeats() { if fwd.healthy { fwd.healthy = false - } else { - err = fmt.Errorf("timed out waiting for vxlan heartbeat") } case <-fwd.stopChan: