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