Merge remote-tracking branch 'origin/main' into forward_fault

router/router.go

@@ -17,11 +17,15 @@
 package router
 
 import (
+	"bufio"
+	"encoding/binary"
 	"encoding/json"
 	"fmt"
+	"github.com/golang/protobuf/proto"
 	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/fabric/controller/network"
 	"github.com/openziti/fabric/controller/xctrl"
+	"github.com/openziti/fabric/pb/ctrl_pb"
 	"github.com/openziti/fabric/router/forwarder"
 	"github.com/openziti/fabric/router/handler_ctrl"
 	"github.com/openziti/fabric/router/handler_link"
@@ -42,6 +46,7 @@ import (
 	"github.com/openziti/foundation/util/concurrenz"
 	"github.com/openziti/foundation/util/info"
 	"github.com/sirupsen/logrus"
+	"io"
 	"math/rand"
 	"time"
 )
@@ -323,3 +328,47 @@ func (self *Router) startControlPlane() error {
 
 	return nil
 }
+
+const (
+	DumpForwarderTables byte = 1
+	UpdateRoute         byte = 2
+)
+
+func (router *Router) HandleDebug(conn io.ReadWriter) error {
+	reader := bufio.NewReader(conn)
+	op, err := reader.ReadByte()
+	if err != nil {
+		return err
+	}
+
+	switch op {
+	case DumpForwarderTables:
+		tables := router.forwarder.Debug()
+		_, err := conn.Write([]byte(tables))
+		return err
+	case UpdateRoute:
+		logrus.Error("received debug operation to update routes")
+		sizeBuf := make([]byte, 4)
+		if _, err := reader.Read(sizeBuf); err != nil {
+			return err
+		}
+		size := binary.LittleEndian.Uint32(sizeBuf)
+		messageBuf := make([]byte, size)
+
+		if _, err := reader.Read(messageBuf); err != nil {
+			return err
+		}
+
+		route := &ctrl_pb.Route{}
+		if err := proto.Unmarshal(messageBuf, route); err != nil {
+			return err
+		}
+
+		logrus.Errorf("updating with route: %+v", route)
+		logrus.Errorf("updating with route: %v", route)
+
+		router.forwarder.Route(route)
+	}
+
+	return nil
+}

router/xgress/link_send_buffer.go

@@ -107,8 +107,8 @@ func NewLinkSendBuffer(x *Xgress) *LinkSendBuffer {
 	return buffer
 }
 
-func (buffer *LinkSendBuffer) CloseWhenEmpty() {
-	buffer.closeWhenEmpty.Set(true)
+func (buffer *LinkSendBuffer) CloseWhenEmpty() bool {
+	return buffer.closeWhenEmpty.CompareAndSwap(false, true)
 }
 
 func (buffer *LinkSendBuffer) BufferPayload(payload *Payload) (func(), error) {

router/xgress/options.go

@@ -43,9 +43,10 @@ type Options struct {
 	RetxStartMs  uint32
 	RetxScale    float64
 	RetxAddMs    uint32
-	MaxCloseWait time.Duration
 
-	GetSessionTimeout time.Duration
+	MaxCloseWait        time.Duration
+	GetSessionTimeout   time.Duration
+	SessionStartTimeout time.Duration
 }
 
 func LoadOptions(data OptionsData) (*Options, error) {
@@ -107,6 +108,7 @@ func LoadOptions(data OptionsData) (*Options, error) {
 		if value, found := data["retxAddMs"]; found {
 			options.RetxAddMs = uint32(value.(int))
 		}
+
 		if value, found := data["maxCloseWaitMs"]; found {
 			options.MaxCloseWait = time.Duration(value.(int)) * time.Millisecond
 		}
@@ -118,6 +120,14 @@ func LoadOptions(data OptionsData) (*Options, error) {
 			}
 			options.GetSessionTimeout = getSessionTimeout
 		}
+
+		if value, found := data["sessionStartTimeout"]; found {
+			sessionStartTimeout, err := time.ParseDuration(value.(string))
+			if err != nil {
+				return nil, errors.Wrap(err, "invalid 'sessionStartTimeout' value")
+			}
+			options.SessionStartTimeout = sessionStartTimeout
+		}
 	}
 
 	return options, nil
@@ -144,6 +154,7 @@ func DefaultOptions() *Options {
 		RetxAddMs:              100,
 		MaxCloseWait:           30 * time.Second,
 		GetSessionTimeout:      30 * time.Second,
+		SessionStartTimeout:    3 * time.Minute,
 	}
 }
 

router/xgress/xgress.go

@@ -176,6 +176,12 @@ func (self *Xgress) markSessionEndReceived() {
 	self.endOfSessionRecvd = true
 }
 
+func (self *Xgress) IsSessionStarted() bool {
+	self.flagsMutex.Lock()
+	defer self.flagsMutex.Unlock()
+	return !self.IsTerminator() || self.rxerStarted
+}
+
 func (self *Xgress) firstSessionStartReceived() bool {
 	self.flagsMutex.Lock()
 	defer self.flagsMutex.Unlock()
@@ -188,16 +194,26 @@ func (self *Xgress) firstSessionStartReceived() bool {
 
 func (self *Xgress) Start() {
 	log := pfxlog.ContextLogger(self.Label())
-	if !self.IsTerminator() {
+	if self.IsTerminator() {
+		log.Debug("terminator: waiting for session start before starting receiver")
+		if self.Options.SessionStartTimeout > time.Second {
+			time.AfterFunc(self.Options.SessionStartTimeout, self.terminateIfNotStarted)
+		}
+	} else {
 		log.Debug("initiator: sending session start")
 		self.forwardPayload(self.GetStartSession())
 		go self.rx()
-	} else {
-		log.Debug("terminator: waiting for session start before starting receiver")
 	}
 	go self.tx()
 }
 
+func (self *Xgress) terminateIfNotStarted() {
+	if !self.IsSessionStarted() {
+		logrus.WithField("xgress", self.Label()).Warn("xgress session not started in time, closing")
+		self.Close()
+	}
+}
+
 func (self *Xgress) Label() string {
 	return fmt.Sprintf("{s/%s|@/%s}<%s>", self.sessionId, string(self.address), self.originator.String())
 }
@@ -244,7 +260,9 @@ func (self *Xgress) IsEndOfSessionSent() bool {
 }
 
 func (self *Xgress) CloseTimeout(duration time.Duration) {
-	go self.closeTimeoutHandler(duration)
+	if self.payloadBuffer.CloseWhenEmpty() { // If we clear the send buffer, close sooner
+		time.AfterFunc(duration, self.Close)
+	}
 }
 
 /*
@@ -522,12 +540,6 @@ func (self *Xgress) nextReceiveSequence() int32 {
 	return next
 }
 
-func (self *Xgress) closeTimeoutHandler(duration time.Duration) {
-	self.payloadBuffer.CloseWhenEmpty() // If we clear the send buffer, close sooner
-	time.Sleep(duration)
-	self.Close()
-}
-
 func (self *Xgress) PayloadReceived(payload *Payload) {
 	log := pfxlog.ContextLogger(self.Label()).WithFields(payload.GetLoggerFields())
 	log.Debug("payload received")