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
This commit is contained in:
1 parent
fc759e2df9
commit
3c67c3c5f1
4 files changed
+57
-2
No files matched your search
+17
-1
@@ -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:
|
||||
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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user