diff --git a/router/fastdp.go b/router/fastdp.go index 3de8eee2..4ecefee9 100644 --- a/router/fastdp.go +++ b/router/fastdp.go @@ -582,9 +582,11 @@ type fastDatapathForwarder struct { ackedHeartbeat bool stopChan chan struct{} stopped bool - - establishedChan chan struct{} - errorChan chan error + healthy bool + established 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,19 +703,40 @@ 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: if fwd.confirmed { + log.Debug(fwd.logPrefix(), "sending Heartbeat to peer") fwd.sendHeartbeat() } fwd.heartbeatTimer.Reset(fwd.heartbeatInterval) case <-fwd.heartbeatTimeout.C: - err = fmt.Errorf("timed out waiting for vxlan heartbeat") + 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 + + // 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 + } case <-fwd.stopChan: return @@ -744,7 +769,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. @@ -801,13 +825,23 @@ 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 // heartbeatTimeout 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 + } } } @@ -833,8 +867,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/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..7fcd564d 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) { @@ -248,6 +255,9 @@ loop: e.established = true establishedChan = nil + case healthy := <-fwd.HealthChannel(): + e.healthy = &healthy + case err := <-fwd.ErrorChannel(): e.err = err @@ -282,6 +292,8 @@ loop: fwd.established(e.index) case e.err != nil: fwd.error(e.index, e.err) + case e.healthy != nil: + fwd.healthCheck(e.index, *e.healthy) } } } @@ -318,6 +330,23 @@ 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 { + 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() + } +} + func (fwd *overlaySwitchForwarder) stopFrom(index int) { for index < len(fwd.forwarders) { subFwd := &fwd.forwarders[index] @@ -337,7 +366,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 +444,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