Merge pull request #3385 from weaveworks/retry_forwarder

Add logic to re-try chooseBest() (which will select best forwarder)

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)

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 }

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()

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