diff --git a/router/router.go b/router/router.go index e602ac600..1892b2170 100644 --- a/router/router.go +++ b/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 +} diff --git a/router/xgress/link_send_buffer.go b/router/xgress/link_send_buffer.go index 4bec6fd12..0e96ac69c 100644 --- a/router/xgress/link_send_buffer.go +++ b/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) { diff --git a/router/xgress/options.go b/router/xgress/options.go index 69124603c..a01d9ef16 100644 --- a/router/xgress/options.go +++ b/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, } } diff --git a/router/xgress/xgress.go b/router/xgress/xgress.go index b837192ca..f60541eb9 100644 --- a/router/xgress/xgress.go +++ b/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")