diff --git a/router/internal/fabric/manager.go b/router/internal/fabric/manager.go index 9dd798db4..828b64b8b 100644 --- a/router/internal/fabric/manager.go +++ b/router/internal/fabric/manager.go @@ -53,7 +53,7 @@ type StateManager interface { //ApiSessions GetApiSession(token string) *edge_ctrl_pb.ApiSession - GetApiSessionByFingerprint(fingerprint string) chan *edge_ctrl_pb.ApiSession + GetApiSessionWithTimeout(token string, timeout time.Duration) *edge_ctrl_pb.ApiSession AddApiSession(ns *edge_ctrl_pb.ApiSession) UpdateApiSession(ns *edge_ctrl_pb.ApiSession) RemoveApiSession(token string) @@ -212,6 +212,27 @@ func (sm *StateManagerImpl) RemoveSession(token string) { } } +func (sm *StateManagerImpl) GetApiSessionWithTimeout(token string, timeout time.Duration) *edge_ctrl_pb.ApiSession { + deadline := time.Now().Add(timeout) + session := sm.GetApiSession(token) + + if session == nil { + //convert this to return a channel instead of sleeping + waitTime := time.Millisecond + for time.Now().Before(deadline) { + session = sm.GetApiSession(token) + if session != nil { + return session + } + time.Sleep(waitTime) + if waitTime < time.Second { + waitTime = waitTime * 2 + } + } + } + return session +} + func (sm *StateManagerImpl) GetApiSession(token string) *edge_ctrl_pb.ApiSession { if val, ok := sm.apiSessionsByToken.Load(token); ok { if session, ok := val.(*edge_ctrl_pb.ApiSession); ok { @@ -221,31 +242,6 @@ func (sm *StateManagerImpl) GetApiSession(token string) *edge_ctrl_pb.ApiSession return nil } -func (sm *StateManagerImpl) GetApiSessionByFingerprint(fingerprint string) chan *edge_ctrl_pb.ApiSession { - ch := make(chan *edge_ctrl_pb.ApiSession) - go func() { - found := false - sm.apiSessionsByToken.Range(func(_, value interface{}) bool { - if session, ok := value.(*edge_ctrl_pb.ApiSession); ok { - for _, curPrint := range session.CertFingerprints { - if fingerprint == curPrint { - ch <- session - found = true - return false - } - } - } - return true - }) - - if !found { - ch <- nil - } - }() - - return ch -} - func (sm *StateManagerImpl) GetSession(token string) *edge_ctrl_pb.Session { if obj, ok := sm.sessionsByToken.Load(token); ok { if ns, ok := obj.(*edge_ctrl_pb.Session); ok { diff --git a/router/xgress_edge/accept.go b/router/xgress_edge/accept.go index 5ab6f6c42..1b8d40343 100644 --- a/router/xgress_edge/accept.go +++ b/router/xgress_edge/accept.go @@ -87,7 +87,7 @@ func (handler edgeBindHandler) BindChannel(ch channel2.Channel) error { func NewAccepter(listener *listener, uListener channel2.UnderlayListener, options *channel2.Options) *Accepter { edgeBindHandler := &edgeBindHandler{listener: listener} - sessionHandler := newSessionConnectHandler(fabric.GetStateManager()) + sessionHandler := newSessionConnectHandler(fabric.GetStateManager(), listener.options) optionsWithBind := options if optionsWithBind == nil { @@ -109,7 +109,7 @@ func (accepter *Accepter) Run() { defer log.Warn("exiting") for { - if _, err := channel2.NewChannel("edge", accepter.uListener, accepter.options); err != nil { + if err := channel2.AcceptNextChannel("edge", accepter.uListener, accepter.options, nil); err != nil { log.Errorf("error accepting (%v)", err) if errors.Is(err, channel2.ListenerClosedError) { return diff --git a/router/xgress_edge/connections.go b/router/xgress_edge/connections.go index 46d9c6861..8e0487c89 100644 --- a/router/xgress_edge/connections.go +++ b/router/xgress_edge/connections.go @@ -28,10 +28,11 @@ import ( type sessionConnectionHandler struct { stateManager fabric.StateManager + options *Options } -func newSessionConnectHandler(stateManager fabric.StateManager) *sessionConnectionHandler { - return &sessionConnectionHandler{stateManager: stateManager} +func newSessionConnectHandler(stateManager fabric.StateManager, options *Options) *sessionConnectionHandler { + return &sessionConnectionHandler{stateManager: stateManager, options: options} } func (handler *sessionConnectionHandler) BindChannel(ch channel2.Channel) error { @@ -49,12 +50,12 @@ func (handler *sessionConnectionHandler) BindChannel(ch channel2.Channel) error fpg := cert.NewFingerprintGenerator() fingerprints := fpg.FromCerts(certificates) - apiSession := handler.stateManager.GetApiSession(token) + apiSession := handler.stateManager.GetApiSessionWithTimeout(token, handler.options.lookupApiSessionTimeout) if apiSession == nil { _ = ch.Close() - subjects := []string{} + var subjects []string for _, cert := range certificates { subjects = append(subjects, cert.Subject.String()) diff --git a/router/xgress_edge/factory.go b/router/xgress_edge/factory.go index 434756c1a..f17483045 100644 --- a/router/xgress_edge/factory.go +++ b/router/xgress_edge/factory.go @@ -18,6 +18,7 @@ package xgress_edge import ( "fmt" + "github.com/michaelquigley/pfxlog" "github.com/openziti/edge/router/handler_edge_ctrl" "github.com/openziti/edge/router/internal/apiproxy" "github.com/openziti/edge/router/internal/fabric" @@ -28,6 +29,8 @@ import ( "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/storage/boltz" "github.com/pkg/errors" + "strings" + "time" ) type Factory struct { @@ -114,6 +117,8 @@ func (factory *Factory) CreateListener(optionsData xgress.OptionsData) (xgress.L return nil, err } + pfxlog.Logger().Debugf("xgress edge listener options: %v", options.ToLoggableString()) + versionInfo := factory.versionProvider.AsVersionInfo() versionHeader, err := factory.versionProvider.EncoderDecoder().Encode(versionInfo) @@ -144,7 +149,42 @@ func (factory *Factory) CreateDialer(optionsData xgress.OptionsData) (xgress.Dia type Options struct { xgress.Options - channelOptions *channel2.Options + channelOptions *channel2.Options + lookupApiSessionTimeout time.Duration + lookupSessionTimeout time.Duration +} + +func (options *Options) ToLoggableString() string { + buf := strings.Builder{} + buf.WriteString(fmt.Sprintf("mtu=%v\n", options.Mtu)) + buf.WriteString(fmt.Sprintf("randomDrops=%v\n", options.RandomDrops)) + buf.WriteString(fmt.Sprintf("drop1InN=%v\n", options.Drop1InN)) + buf.WriteString(fmt.Sprintf("txQueueSize=%v\n", options.TxQueueSize)) + buf.WriteString(fmt.Sprintf("txPortalStartSize=%v\n", options.TxPortalStartSize)) + buf.WriteString(fmt.Sprintf("txPortalMaxSize=%v\n", options.TxPortalMaxSize)) + buf.WriteString(fmt.Sprintf("txPortalMinSize=%v\n", options.TxPortalMinSize)) + buf.WriteString(fmt.Sprintf("txPortalIncreaseThresh=%v\n", options.TxPortalIncreaseThresh)) + buf.WriteString(fmt.Sprintf("txPortalIncreaseScale=%v\n", options.TxPortalIncreaseScale)) + buf.WriteString(fmt.Sprintf("txPortalRetxThresh=%v\n", options.TxPortalRetxThresh)) + buf.WriteString(fmt.Sprintf("txPortalRetxScale=%v\n", options.TxPortalRetxScale)) + buf.WriteString(fmt.Sprintf("txPortalDupAckThresh=%v\n", options.TxPortalDupAckThresh)) + buf.WriteString(fmt.Sprintf("txPortalDupAckScale=%v\n", options.TxPortalDupAckScale)) + buf.WriteString(fmt.Sprintf("rxBufferSize=%v\n", options.RxBufferSize)) + buf.WriteString(fmt.Sprintf("retxStartMs=%v\n", options.RetxStartMs)) + buf.WriteString(fmt.Sprintf("retxScale=%v\n", options.RetxScale)) + buf.WriteString(fmt.Sprintf("retxAddMs=%v\n", options.RetxAddMs)) + buf.WriteString(fmt.Sprintf("maxCloseWait=%v\n", options.MaxCloseWait)) + buf.WriteString(fmt.Sprintf("getSessionTimeout=%v\n", options.GetSessionTimeout)) + + buf.WriteString(fmt.Sprintf("lookupApiSessionTimeout=%v\n", options.lookupApiSessionTimeout)) + buf.WriteString(fmt.Sprintf("lookupSessionTimeout=%v\n", options.lookupSessionTimeout)) + + buf.WriteString(fmt.Sprintf("channel.outQueueSize=%v\n", options.channelOptions.OutQueueSize)) + buf.WriteString(fmt.Sprintf("channel.connectTimeoutMs=%v\n", options.channelOptions.ConnectTimeoutMs)) + buf.WriteString(fmt.Sprintf("channel.maxOutstandingConnects=%v\n", options.channelOptions.MaxOutstandingConnects)) + buf.WriteString(fmt.Sprintf("channel.maxQueuedConnects=%v\n", options.channelOptions.MaxQueuedConnects)) + + return buf.String() } func (options *Options) load(data xgress.OptionsData) error { @@ -153,6 +193,8 @@ func (options *Options) load(data xgress.OptionsData) error { return errors.Wrap(err, "error loading options") } options.Options = *o + options.lookupSessionTimeout = 5 * time.Second + options.lookupApiSessionTimeout = 5 * time.Second if value, found := data["options"]; found { data = value.(map[interface{}]interface{}) @@ -161,6 +203,22 @@ func (options *Options) load(data xgress.OptionsData) error { if err := options.channelOptions.Validate(); err != nil { return fmt.Errorf("error loading options for [edge/options]: %v", err) } + + if value, found := data["lookupSessionTimeout"]; found { + timeout, err := time.ParseDuration(value.(string)) + if err != nil { + return errors.Wrap(err, "invalid 'lookupSessionTimeout' value") + } + options.lookupSessionTimeout = timeout + } + + if value, found := data["lookupApiSessionTimeout"]; found { + timeout, err := time.ParseDuration(value.(string)) + if err != nil { + return errors.Wrap(err, "invalid 'lookupApiSessionTimeout' value") + } + options.lookupApiSessionTimeout = timeout + } } else { options.channelOptions = channel2.DefaultOptions() } diff --git a/router/xgress_edge/listener.go b/router/xgress_edge/listener.go index 071a5b289..b199e8421 100644 --- a/router/xgress_edge/listener.go +++ b/router/xgress_edge/listener.go @@ -112,7 +112,7 @@ func (proxy *ingressProxy) processConnect(req *channel2.Message, ch channel2.Cha } log.Debug("validating network session") sm := fabric.GetStateManager() - ns := sm.GetSessionWithTimeout(token, time.Second*5) + ns := sm.GetSessionWithTimeout(token, proxy.listener.options.lookupSessionTimeout) if ns == nil || ns.Type != edge_ctrl_pb.SessionType_Dial { log.WithField("token", token).Error("session not found") @@ -211,7 +211,7 @@ func (proxy *ingressProxy) processBind(req *channel2.Message, ch channel2.Channe } log.Debug("validating network session") sm := fabric.GetStateManager() - ns := sm.GetSessionWithTimeout(token, time.Second*5) + ns := sm.GetSessionWithTimeout(token, proxy.listener.options.lookupSessionTimeout) if ns == nil || ns.Type != edge_ctrl_pb.SessionType_Bind { log.WithField("token", token).Error("session not found")