gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-12 15:28:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add timeout to api session lookup. Make api-session/session lookup timeouts configurable

Browse Source
This commit is contained in:
Paul Lorenz
2021-01-25 15:43:56 -05:00
parent 6b2276acd9
commit cdba8994ff
5 changed files with 90 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
router/internal/fabric/manager.go
+22 -26
View File
@@ -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 {
router/xgress_edge/accept.go
+2 -2
View File
@@ -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
router/xgress_edge/connections.go
+5 -4
View File
@@ -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())
router/xgress_edge/factory.go
+59 -1
View File
@@ -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()
}
router/xgress_edge/listener.go
+2 -2
View File
@@ -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")