Merge pull request openziti/fabric#430 from openziti/prevent-strange-loops

Prevent strange loops

controller/network/circuit.go

@@ -21,6 +21,7 @@ import (
 	"github.com/openziti/fabric/controller/xt"
 	"github.com/openziti/foundation/v2/concurrenz"
 	"github.com/orcaman/concurrent-map/v2"
+	"time"
 )
 
 type Circuit struct {
@@ -31,6 +32,7 @@ type Circuit struct {
 	Path       *Path
 	Rerouting  concurrenz.AtomicBoolean
 	PeerData   xt.PeerData
+	CreatedAt  time.Time
 }
 
 func (self *Circuit) cost() int64 {
@@ -68,29 +70,29 @@ func newCircuitController() *circuitController {
 	}
 }
 
-func (c *circuitController) nextCircuitId() (string, error) {
-	return c.idGenerator.NextAlphaNumericPrefixedId()
+func (self *circuitController) nextCircuitId() (string, error) {
+	return self.idGenerator.NextAlphaNumericPrefixedId()
 }
 
-func (c *circuitController) add(sn *Circuit) {
-	c.circuits.Set(sn.Id, sn)
+func (self *circuitController) add(circuit *Circuit) {
+	self.circuits.Set(circuit.Id, circuit)
 }
 
-func (c *circuitController) get(id string) (*Circuit, bool) {
-	if circuit, found := c.circuits.Get(id); found {
+func (self *circuitController) get(id string) (*Circuit, bool) {
+	if circuit, found := self.circuits.Get(id); found {
 		return circuit, true
 	}
 	return nil, false
 }
 
-func (c *circuitController) all() []*Circuit {
+func (self *circuitController) all() []*Circuit {
 	circuits := make([]*Circuit, 0)
-	for tuple := range c.circuits.IterBuffered() {
+	for tuple := range self.circuits.IterBuffered() {
 		circuits = append(circuits, tuple.Val)
 	}
 	return circuits
 }
 
-func (c *circuitController) remove(sn *Circuit) {
-	c.circuits.Remove(sn.Id)
+func (self *circuitController) remove(circuit *Circuit) {
+	self.circuits.Remove(circuit.Id)
 }

controller/network/network.go

@@ -524,6 +524,7 @@ func (network *Network) CreateCircuit(srcR *Router, clientId *identity.TokenId,
 			Path:       path,
 			Terminator: terminator,
 			PeerData:   peerData,
+			CreatedAt:  time.Now(),
 		}
 		network.circuitController.add(circuit)
 		creationTimespan := time.Since(startTime)
@@ -870,6 +871,24 @@ func (network *Network) rerouteLink(l *Link, deadline time.Time) error {
 	return nil
 }
 
+func (network *Network) rerouteCircuitWithTries(circuit *Circuit, retries int) {
+	log := pfxlog.Logger().WithField("circuitId", circuit.Id)
+
+	for i := 0; i < retries; i++ {
+		deadline := time.Now().Add(DefaultNetworkOptionsRouteTimeout)
+		err := network.rerouteCircuit(circuit, deadline)
+		if err == nil {
+			return
+		}
+
+		log.WithError(err).WithField("attempt", i).Error("error re-routing circuit")
+	}
+
+	if err := network.RemoveCircuit(circuit.Id, true); err != nil {
+		log.WithError(err).Error("failure while removing circuit after failed re-route attempt")
+	}
+}
+
 func (network *Network) rerouteCircuit(circuit *Circuit, deadline time.Time) error {
 	log := pfxlog.Logger().WithField("circuitId", circuit.Id)
 	if circuit.Rerouting.CompareAndSwap(false, true) {
@@ -901,28 +920,30 @@ func (network *Network) rerouteCircuit(circuit *Circuit, deadline time.Time) err
 	}
 }
 
-func (network *Network) smartReroute(s *Circuit, cq *Path, deadline time.Time) error {
-	if s.Rerouting.CompareAndSwap(false, true) {
-		defer s.Rerouting.Set(false)
+func (network *Network) smartReroute(circuit *Circuit, cq *Path, deadline time.Time) bool {
+	retry := false
+	log := pfxlog.Logger().WithField("circuitId", circuit.Id)
+	if circuit.Rerouting.CompareAndSwap(false, true) {
+		defer circuit.Rerouting.Set(false)
 
-		s.Path = cq
+		circuit.Path = cq
 
-		rms := cq.CreateRouteMessages(SmartRerouteAttempt, s.Id, s.Terminator, deadline)
+		rms := cq.CreateRouteMessages(SmartRerouteAttempt, circuit.Id, circuit.Terminator, deadline)
 
 		for i := 0; i < len(cq.Nodes); i++ {
 			if _, err := sendRoute(cq.Nodes[i], rms[i], network.options.RouteTimeout); err != nil {
-				logrus.Errorf("error sending route to [r/%s] (%s)", cq.Nodes[i].Id, err)
+				retry = true
+				log.WithField("routerId", cq.Nodes[i].Id).WithError(err).Error("error sending smart route update to router")
+				break
 			}
 		}
 
-		logrus.Debugf("rerouted circuit [s/%s]", s.Id)
-		network.CircuitEvent(CircuitUpdated, s, nil, nil)
-		return nil
-
-	} else {
-		logrus.Infof("not rerouting [s/%s], already in progress", s.Id)
-		return nil
+		if !retry {
+			logrus.Debug("rerouted circuit")
+			network.CircuitEvent(CircuitUpdated, circuit, nil, nil)
+		}
 	}
+	return retry
 }
 
 func (network *Network) AcceptMetrics(metrics *metrics_pb.MetricsMessage) {

controller/network/smart.go

@@ -85,8 +85,8 @@ func (network *Network) smart() {
 	 * Reroute.
 	 */
 	for _, circuit := range candidates {
-		if err := network.smartReroute(circuit, newPaths[circuit], time.Now().Add(DefaultNetworkOptionsRouteTimeout)); err != nil {
-			log.Errorf("error rerouting [s/%s] (%s)", circuit.Id, err)
+		if retry := network.smartReroute(circuit, newPaths[circuit], time.Now().Add(DefaultNetworkOptionsRouteTimeout)); retry {
+			go network.rerouteCircuitWithTries(circuit, DefaultNetworkOptionsCreateCircuitRetries)
 		}
 	}
 	/* */