gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-09-21 18:13:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Clean up xgress edge abstractions and close logic

Browse Source
This commit is contained in:
Paul Lorenz
2021-01-25 12:32:57 -05:00
parent 5e5aa87c39
commit bca8d657fc
11 changed files with 177 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+2 -2
View File
@@ -38,8 +38,8 @@ require (
github.com/miekg/dns v1.1.35
github.com/mitchellh/mapstructure v1.4.1
github.com/netfoundry/secretstream v0.1.2
github.com/openziti/fabric v0.15.12
github.com/openziti/foundation v0.15.8
github.com/openziti/fabric v0.15.13
github.com/openziti/foundation v0.15.9
github.com/openziti/sdk-golang v0.15.10
github.com/orcaman/concurrent-map v0.0.0-20190826125027-8c72a8bb44f6
github.com/pkg/errors v0.9.1
go.sum
+4 -2
View File
@@ -461,10 +461,12 @@ github.com/onsi/gomega v1.8.1/go.mod h1:Ho0h+IUsWyvy1OpqCwxlQ/21gkhVunqlU8fDGcoT
github.com/onsi/gomega v1.10.1 h1:o0+MgICZLuZ7xjH7Vx6zS/zcu93/BEp1VwkIW1mEXCE=
github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTmOf0Erfk+hxe8=
github.com/openziti/fabric v0.15.12 h1:ZVQamwVrFzlZCmnqe8g4ZOXsUPo2g+GL8AdDplWUkew=
github.com/openziti/fabric v0.15.12/go.mod h1:B8NeRzN+eIdo2Q4JUCbP0covC5SYqerrxwJsBgCU+A8=
github.com/openziti/fabric v0.15.13 h1:GgS4Sbce6ubkDUtO2oPQFEzh5S9EiAjtMs50NJbuAvo=
github.com/openziti/fabric v0.15.13/go.mod h1:B8NeRzN+eIdo2Q4JUCbP0covC5SYqerrxwJsBgCU+A8=
github.com/openziti/foundation v0.15.8 h1:BFKDBkgsuXWgCtU7TS3eBPz2gceA057gK6W6IiyUQfM=
github.com/openziti/foundation v0.15.8/go.mod h1:34ifQ24R2EUNjakzZBKY1M4EvnOWeIYJOrJxRGdRNh4=
github.com/openziti/foundation v0.15.9 h1:Do/wohULRt+3H6mlB8SZmIUl/mbxUuN7/Lf7nbz6ZRo=
github.com/openziti/foundation v0.15.9/go.mod h1:34ifQ24R2EUNjakzZBKY1M4EvnOWeIYJOrJxRGdRNh4=
github.com/openziti/sdk-golang v0.15.10 h1:iwYTmBsEOBxcQlXuwSSBQLipUyT4QuRDFYXkf6zjyYs=
github.com/openziti/sdk-golang v0.15.10/go.mod h1:ZC5Tke4NFsRH3qcZnc9X51hGjpXlq0aOCcadgxnICYI=
github.com/orcaman/concurrent-map v0.0.0-20190826125027-8c72a8bb44f6 h1:lNCW6THrCKBiJBpz8kbVGjC7MgdCGKwuvBgc7LoD6sw=
router/xgress_edge/accept.go
+1 -1
View File
@@ -41,7 +41,7 @@ func (handler edgeBindHandler) BindChannel(ch channel2.Channel) error {
log.WithField("token", ch.Id()).Debug("accepting edge connection")
fpg := cert.NewFingerprintGenerator()
proxy := &ingressProxy{
proxy := &edgeClientConn{
msgMux: edge.NewCowMapMsgMux(),
listener: handler.listener,
fingerprints: fpg.FromCerts(ch.Certificates()),
router/xgress_edge/dialer.go
+19 -5
View File
@@ -24,7 +24,7 @@ import (
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/sdk-golang/ziti/edge"
log "github.com/sirupsen/logrus"
"github.com/pkg/errors"
"strings"
"time"
)
@@ -46,7 +46,7 @@ func (dialer dialer) IsTerminatorValid(id string, destination string) bool {
token := destParts[1]
log.Debug("looking up hosted service conn")
pfxlog.Logger().Debug("looking up hosted service conn")
_, found := dialer.factory.hostedServices.Get(token)
return found
}
@@ -91,6 +91,7 @@ func (dialer *dialer) Dial(destination string, sessionId *identity.TokenId, addr
if pk, ok := sessionId.Data[edge.PublicKeyHeader]; ok {
dialRequest.Headers[edge.PublicKeyHeader] = pk
}
appData, hasAppData := sessionId.Data[edge.AppDataHeader]
if hasAppData {
dialRequest.Headers[edge.AppDataHeader] = appData
@@ -102,7 +103,10 @@ func (dialer *dialer) Dial(destination string, sessionId *identity.TokenId, addr
log.Debug("router assigned connId for dial")
dialRequest.PutUint32Header(edge.RouterProvidedConnId, connId)
conn := listenConn.newSink(connId)
conn, err := listenConn.newConnection(connId)
if err != nil {
return nil, errors.Wrapf(err, "failed to create edge xgress conn for token %v", token)
}
// On the terminator, which this is, this only starts the txer, which pulls data from the link
// Since the opposing xgress doesn't start until this call returns, nothing should be coming this way yet
@@ -139,8 +143,8 @@ func (dialer *dialer) Dial(destination string, sessionId *identity.TokenId, addr
if err != nil {
return nil, err
}
result, err := edge.UnmarshalDialResult(reply)
result, err := edge.UnmarshalDialResult(reply)
if err != nil {
return nil, err
}
@@ -149,7 +153,17 @@ func (dialer *dialer) Dial(destination string, sessionId *identity.TokenId, addr
return nil, fmt.Errorf("failed to establish connection with token %v. error: (%v)", token, result.Message)
}
conn := listenConn.newSink(result.NewConnId)
conn, err := listenConn.newConnection(result.NewConnId)
if err != nil {
startFail := edge.NewStateConnectedMsg(result.ConnId)
startFail.ReplyTo(reply)
if sendErr := listenConn.SendState(startFail); sendErr != nil {
log.Debug("failed to send state disconnected")
}
return nil, errors.Wrapf(err, "failed to create edge xgress conn for token %v", token)
}
x := xgress.NewXgress(sessionId, address, conn, xgress.Terminator, &dialer.options.Options)
bindHandler.HandleXgressBind(x)
router/xgress_edge/fabric.go
+62 -41
View File
@@ -22,6 +22,7 @@ import (
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/util/concurrenz"
"github.com/openziti/sdk-golang/ziti/edge"
log "github.com/sirupsen/logrus"
"io"
"math"
"sync/atomic"
@@ -40,57 +41,78 @@ var headersFromFabric = map[uint8]int32{
PayloadFlagsHeader: edge.FlagsHeader,
}
type localMessageSink struct {
type edgeTerminator struct {
edge.MsgChannel
seq MsgQueue
closeCB func(connId uint32)
newSinkCB func(sink *localMessageSink)
closed concurrenz.AtomicBoolean
edgeClientConn *edgeClientConn
token string
terminatorId string
service string
assignIds bool
}
type localListener struct {
localMessageSink
terminatorIdRef *concurrenz.AtomicString
service string
parent *ingressProxy
assignIds bool
}
func (listener *localListener) nextDialConnId() uint32 {
nextId := atomic.AddUint32(&listener.parent.idSeq, 1)
func (listener *edgeTerminator) nextDialConnId() uint32 {
nextId := atomic.AddUint32(&listener.edgeClientConn.idSeq, 1)
if nextId < math.MaxUint32/2 {
atomic.StoreUint32(&listener.parent.idSeq, math.MaxUint32/2)
nextId = atomic.AddUint32(&listener.parent.idSeq, 1)
atomic.StoreUint32(&listener.edgeClientConn.idSeq, math.MaxUint32/2)
nextId = atomic.AddUint32(&listener.edgeClientConn.idSeq, 1)
}
return nextId
}
func (conn *localMessageSink) newSink(connId uint32) *localMessageSink {
result := &localMessageSink{
MsgChannel: *edge.NewEdgeMsgChannel(conn.Channel, connId),
seq: NewMsgQueue(4),
closeCB: conn.closeCB,
newSinkCB: conn.newSinkCB,
func (listener *edgeTerminator) close(notify bool, reason string) {
logger := pfxlog.Logger()
if notify && !listener.IsClosed() {
// Notify edge client of close
log.Debug("sending closed to SDK client")
closeMsg := edge.NewStateClosedMsg(listener.Id(), reason)
if err := listener.SendState(closeMsg); err != nil {
log.WithError(err).Warn("unable to send close msg to edge client for hosted service")
}
}
// TODO: Evaluate best way to get new conn in map. Split dial? Any way to use regular map?
if conn.newSinkCB != nil {
conn.newSinkCB(result)
logger.Debugf("removing terminator %v for token: %v", listener.terminatorId, listener.token)
if err := xgress.RemoveTerminator(listener.edgeClientConn.listener.factory, listener.terminatorId); err != nil {
logger.Errorf("failed to remove terminator %v (%v)", listener.terminatorId, err)
}
return result
logger.Debugf("removing listener for terminator %v, token: %v", listener.terminatorId, listener.token)
listener.edgeClientConn.listener.factory.hostedServices.Delete(listener.token)
}
func (conn *localMessageSink) LogContext() string {
type edgeXgressConn struct {
edge.MsgChannel
mux edge.MsgMux
seq MsgQueue
newSinkCB func(sink *edgeXgressConn)
closed concurrenz.AtomicBoolean
}
func (listener *edgeTerminator) newConnection(connId uint32) (*edgeXgressConn, error) {
mux := listener.edgeClientConn.msgMux
result := &edgeXgressConn{
mux: mux,
MsgChannel: *edge.NewEdgeMsgChannel(listener.edgeClientConn.ch, connId),
seq: NewMsgQueue(4),
}
if err := mux.AddMsgSink(result); err != nil {
return nil, err
}
return result, nil
}
func (conn *edgeXgressConn) LogContext() string {
return conn.Channel.Label()
}
func (conn *localMessageSink) ReadPayload() ([]byte, map[uint8][]byte, error) {
func (conn *edgeXgressConn) ReadPayload() ([]byte, map[uint8][]byte, error) {
log := pfxlog.ContextLogger(conn.Channel.Label()).WithField("connId", conn.Id())
msg := conn.seq.Pop()
if msg == nil {
log.Debug("sequencer closed, return EOF")
conn.closeCB(conn.Id())
return nil, nil, io.EOF // io.EOF signals xgress to shutdown
}
@@ -114,7 +136,7 @@ func (conn *localMessageSink) ReadPayload() ([]byte, map[uint8][]byte, error) {
}
}
func (conn *localMessageSink) WritePayload(p []byte, headers map[uint8][]byte) (n int, err error) {
func (conn *edgeXgressConn) WritePayload(p []byte, headers map[uint8][]byte) (n int, err error) {
var msgUUID []byte
var edgeHdrs map[int32][]byte
@@ -148,17 +170,17 @@ func (conn *localMessageSink) WritePayload(p []byte, headers map[uint8][]byte) (
return len(p), nil
}
func (conn *localMessageSink) Close() error {
func (conn *edgeXgressConn) Close() error {
conn.close(true, "close called")
return nil
}
func (conn *localMessageSink) HandleMuxClose() error {
func (conn *edgeXgressConn) HandleMuxClose() error {
conn.close(false, "channel closed")
return nil
}
func (conn *localMessageSink) close(notify bool, reason string) {
func (conn *edgeXgressConn) close(notify bool, reason string) {
if !conn.closed.CompareAndSwap(false, true) {
// already closed
return
@@ -166,17 +188,16 @@ func (conn *localMessageSink) close(notify bool, reason string) {
log := pfxlog.ContextLogger(conn.Channel.Label()).WithField("connId", conn.Id())
log.Debugf("closing message sink, reason: %v", reason)
if notify {
if notify && !conn.IsClosed() {
// Notify edge client of close
log.Debug("sennding closed to SDK client")
closeMsg := edge.NewStateClosedMsg(conn.Id(), "")
log.Debug("sending closed to SDK client")
closeMsg := edge.NewStateClosedMsg(conn.Id(), reason)
if err := conn.SendState(closeMsg); err != nil {
log.WithError(err).Warn("unable to send close msg to edge client")
}
}
// remove ourselves from mux, etc
conn.closeCB(conn.Id())
conn.mux.RemoveMsgSink(conn)
// When nextSeq is closed, GetNext in Read() will return a nil.
// This will cause an io.EOF to be returned to the xgress read loop, which will cause that
@@ -185,13 +206,13 @@ func (conn *localMessageSink) close(notify bool, reason string) {
conn.seq.Close()
}
func (conn *localMessageSink) Accept(msg *channel2.Message) {
func (conn *edgeXgressConn) Accept(msg *channel2.Message) {
if err := conn.seq.Push(msg); err != nil {
pfxlog.Logger().WithFields(edge.GetLoggerFields(msg)).Errorf("failed to dispatch to fabric: (%v)", err)
}
}
func (conn *localMessageSink) getHeaderMap(message *channel2.Message) map[uint8][]byte {
func (conn *edgeXgressConn) getHeaderMap(message *channel2.Message) map[uint8][]byte {
headers := make(map[uint8][]byte)
msgUUID, found := message.Headers[edge.UUIDHeader]
if found {
router/xgress_edge/hosted.go
+7 -7
View File
@@ -24,16 +24,16 @@ type hostedServiceRegistry struct {
services sync.Map
}
func (registry *hostedServiceRegistry) Put(hostId string, conn *localListener) {
func (registry *hostedServiceRegistry) Put(hostId string, conn *edgeTerminator) {
registry.services.Store(hostId, conn)
}
func (registry *hostedServiceRegistry) Get(hostId string) (*localListener, bool) {
func (registry *hostedServiceRegistry) Get(hostId string) (*edgeTerminator, bool) {
val, ok := registry.services.Load(hostId)
if !ok {
return nil, false
}
ch, ok := val.(*localListener)
ch, ok := val.(*edgeTerminator)
return ch, ok
}
@@ -41,11 +41,11 @@ func (registry *hostedServiceRegistry) Delete(hostId string) {
registry.services.Delete(hostId)
}
func (registry *hostedServiceRegistry) cleanupServices(proxy *ingressProxy) (listeners []*localListener) {
func (registry *hostedServiceRegistry) cleanupServices(proxy *edgeClientConn) (listeners []*edgeTerminator) {
registry.services.Range(func(key, value interface{}) bool {
listener := value.(*localListener)
if listener.parent == proxy {
listener.close(true, "underlying channel closing")
listener := value.(*edgeTerminator)
if listener.edgeClientConn == proxy {
listener.close(false, "") // don't notify, channel is already closed, we can't send messages
registry.services.Delete(key)
listeners = append(listeners, listener)
}
router/xgress_edge/listener.go
+80 -118
View File
@@ -28,7 +28,6 @@ import (
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/transport"
"github.com/openziti/foundation/util/concurrenz"
"github.com/openziti/sdk-golang/ziti/edge"
"time"
)
@@ -78,7 +77,7 @@ func (listener *listener) Close() error {
return listener.underlayListener.Close()
}
type ingressProxy struct {
type edgeClientConn struct {
msgMux edge.MsgMux
listener *listener
fingerprints cert.Fingerprints
@@ -86,23 +85,23 @@ type ingressProxy struct {
idSeq uint32
}
func (proxy *ingressProxy) HandleClose(_ channel2.Channel) {
log := pfxlog.ContextLogger(proxy.ch.Label())
func (self *edgeClientConn) HandleClose(_ channel2.Channel) {
log := pfxlog.ContextLogger(self.ch.Label())
log.Debugf("closing")
listeners := proxy.listener.factory.hostedServices.cleanupServices(proxy)
listeners := self.listener.factory.hostedServices.cleanupServices(self)
for _, listener := range listeners {
if err := xgress.RemoveTerminator(proxy.listener.factory, listener.terminatorIdRef.Get()); err != nil {
log.Warnf("failed to remove terminator on service %v for terminator %v on channel close", listener.service, listener.terminatorIdRef.Get())
if err := xgress.RemoveTerminator(self.listener.factory, listener.terminatorId); err != nil {
log.Warnf("failed to remove terminator on service %v for terminator %v on channel close", listener.service, listener.terminatorId)
}
}
proxy.msgMux.Close()
self.msgMux.Close()
}
func (proxy *ingressProxy) ContentType() int32 {
func (self *edgeClientConn) ContentType() int32 {
return edge.ContentTypeData
}
func (proxy *ingressProxy) processConnect(req *channel2.Message, ch channel2.Channel) {
func (self *edgeClientConn) processConnect(req *channel2.Message, ch channel2.Channel) {
token := string(req.Body)
log := pfxlog.ContextLogger(ch.Label()).WithField("token", token).WithFields(edge.GetLoggerFields(req))
connId, found := req.GetUint32Header(edge.ConnIdHeader)
@@ -112,41 +111,39 @@ func (proxy *ingressProxy) processConnect(req *channel2.Message, ch channel2.Cha
}
log.Debug("validating network session")
sm := fabric.GetStateManager()
ns := sm.GetSessionWithTimeout(token, proxy.listener.options.lookupSessionTimeout)
ns := sm.GetSessionWithTimeout(token, self.listener.options.lookupSessionTimeout)
if ns == nil || ns.Type != edge_ctrl_pb.SessionType_Dial {
log.WithField("token", token).Error("session not found")
proxy.sendStateClosedReply("Invalid Session", req)
self.sendStateClosedReply("Invalid Session", req)
return
}
if _, found := proxy.fingerprints.HasAny(ns.CertFingerprints); !found {
if _, found := self.fingerprints.HasAny(ns.CertFingerprints); !found {
log.WithField("token", token).
WithField("serviceFingerprints", ns.CertFingerprints).
WithField("clientFingerprints", proxy.fingerprints.Prints()).
WithField("clientFingerprints", self.fingerprints.Prints()).
Error("matching fingerprint not found for connect")
proxy.sendStateClosedReply("Invalid Session", req)
self.sendStateClosedReply("Invalid Session", req)
return
}
log.Debug("validating connection id")
removeListener := sm.AddSessionRemovedListener(ns.Token, func(token string) {
proxy.sendStateClosed(connId, "session closed")
proxy.closeConn(connId)
})
conn := &localMessageSink{
MsgChannel: *edge.NewEdgeMsgChannel(proxy.ch, connId),
conn := &edgeXgressConn{
mux: self.msgMux,
MsgChannel: *edge.NewEdgeMsgChannel(self.ch, connId),
seq: NewMsgQueue(4),
closeCB: func(connId uint32) {
removeListener()
},
}
if err := proxy.msgMux.AddMsgSink(conn); err != nil {
sm.AddSessionRemovedListener(ns.Token, func(token string) {
conn.close(true, "session closed")
})
// We can't fix conn id, since it's provided by the client
if err := self.msgMux.AddMsgSink(conn); err != nil {
log.WithField("token", token).Error(err)
proxy.sendStateClosedReply(err.Error(), req)
self.sendStateClosedReply(err.Error(), req)
return
}
@@ -168,7 +165,7 @@ func (proxy *ingressProxy) processConnect(req *channel2.Message, ch channel2.Cha
if ns.Service.EncryptionRequired && req.Headers[edge.PublicKeyHeader] == nil {
msg := "encryption required on service, initiator did not send public header"
proxy.sendStateClosedReply(msg, req)
self.sendStateClosedReply(msg, req)
conn.close(false, msg)
return
}
@@ -177,30 +174,31 @@ func (proxy *ingressProxy) processConnect(req *channel2.Message, ch channel2.Cha
if terminatorIdentity, found := req.GetStringHeader(edge.TerminatorIdentityHeader); found {
service = terminatorIdentity + "@" + service
}
sessionInfo, err := xgress.GetSession(proxy.listener.factory, ns.Id, service, proxy.listener.options.Options.GetSessionTimeout, peerData)
sessionInfo, err := xgress.GetSession(self.listener.factory, ns.Id, service, self.listener.options.Options.GetSessionTimeout, peerData)
if err != nil {
log.WithError(err).Warn("failed to dial fabric")
proxy.sendStateClosedReply(err.Error(), req)
self.sendStateClosedReply(err.Error(), req)
conn.close(false, "failed to dial fabric")
return
}
if ns.Service.EncryptionRequired && sessionInfo.SessionId.Data[edge.PublicKeyHeader] == nil {
msg := "encryption required on service, terminator did not send public header"
proxy.sendStateClosedReply(msg, req)
self.sendStateClosedReply(msg, req)
conn.close(false, msg)
return
}
x := xgress.NewXgress(sessionInfo.SessionId, sessionInfo.Address, conn, xgress.Initiator, &proxy.listener.options.Options)
proxy.listener.bindHandler.HandleXgressBind(x)
x := xgress.NewXgress(sessionInfo.SessionId, sessionInfo.Address, conn, xgress.Initiator, &self.listener.options.Options)
self.listener.bindHandler.HandleXgressBind(x)
// send the state_connected before starting the xgress. That way we can't get a state_closed before we get state_connected
proxy.sendStateConnectedReply(req, sessionInfo.SessionId.Data)
self.sendStateConnectedReply(req, sessionInfo.SessionId.Data)
x.Start()
}
func (proxy *ingressProxy) processBind(req *channel2.Message, ch channel2.Channel) {
func (self *edgeClientConn) processBind(req *channel2.Message, ch channel2.Channel) {
token := string(req.Body)
log := pfxlog.ContextLogger(ch.Label()).WithField("sessionId", token).WithFields(edge.GetLoggerFields(req))
@@ -211,20 +209,20 @@ func (proxy *ingressProxy) processBind(req *channel2.Message, ch channel2.Channe
}
log.Debug("validating network session")
sm := fabric.GetStateManager()
ns := sm.GetSessionWithTimeout(token, proxy.listener.options.lookupSessionTimeout)
ns := sm.GetSessionWithTimeout(token, self.listener.options.lookupSessionTimeout)
if ns == nil || ns.Type != edge_ctrl_pb.SessionType_Bind {
log.WithField("token", token).Error("session not found")
proxy.sendStateClosedReply("Invalid Session", req)
self.sendStateClosedReply("Invalid Session", req)
return
}
if _, found := proxy.fingerprints.HasAny(ns.CertFingerprints); !found {
if _, found := self.fingerprints.HasAny(ns.CertFingerprints); !found {
log.WithField("token", token).
WithField("serviceFingerprints", ns.CertFingerprints).
WithField("clientFingerprints", proxy.fingerprints.Prints()).
WithField("clientFingerprints", self.fingerprints.Prints()).
Error("matching fingerprint not found for bind")
proxy.sendStateClosedReply("Invalid Session", req)
self.sendStateClosedReply("Invalid Session", req)
return
}
@@ -251,48 +249,23 @@ func (proxy *ingressProxy) processBind(req *channel2.Message, ch channel2.Channe
}
}
terminatorIdRef := &concurrenz.AtomicString{}
removeListener := sm.AddSessionRemovedListener(ns.Token, func(token string) {
terminatorId := terminatorIdRef.Get()
defer func() {
log.Debugf("removing listener for terminator %v, token: %v", terminatorId, token)
proxy.listener.factory.hostedServices.Delete(token)
}()
log.Debugf("removing terminator %v for token: %v", terminatorId, token)
if err := xgress.RemoveTerminator(proxy.listener.factory, terminatorId); err != nil {
log.Errorf("failed to remove terminator %v (%v)", terminatorId, err)
}
})
assignIds, _ := req.GetBoolHeader(edge.RouterProvidedConnId)
log.Debugf("client requested router provided connection ids: %v", assignIds)
log.Debug("establishing listener")
messageSink := &localListener{
localMessageSink: localMessageSink{
MsgChannel: *edge.NewEdgeMsgChannel(ch, connId),
seq: NewMsgQueue(4),
closeCB: func(closeConnId uint32) {
if closeConnId == connId {
removeListener()
}
proxy.sendStateClosed(closeConnId, "session closed")
proxy.closeConn(closeConnId)
},
newSinkCB: func(conn *localMessageSink) {
if err := proxy.msgMux.AddMsgSink(conn); err != nil {
log.WithError(err).Error("Failed to add sink, duplicate id")
}
},
},
terminatorIdRef: terminatorIdRef,
service: ns.Service.Id,
parent: proxy,
assignIds: assignIds,
messageSink := &edgeTerminator{
MsgChannel: *edge.NewEdgeMsgChannel(self.ch, connId),
edgeClientConn: self,
token: token,
service: ns.Service.Id,
assignIds: assignIds,
}
proxy.listener.factory.hostedServices.Put(token, messageSink)
sm.AddSessionRemovedListener(ns.Token, func(token string) {
messageSink.close(true, "session ended")
})
self.listener.factory.hostedServices.Put(token, messageSink)
terminatorIdentity, _ := req.GetStringHeader(edge.TerminatorIdentityHeader)
var terminatorIdentitySecret []byte
@@ -300,22 +273,22 @@ func (proxy *ingressProxy) processBind(req *channel2.Message, ch channel2.Channe
terminatorIdentitySecret, _ = req.Headers[edge.TerminatorIdentitySecretHeader]
}
terminatorId, err := xgress.AddTerminator(proxy.listener.factory, ns.Service.Id, "edge", "hosted:"+token, terminatorIdentity, terminatorIdentitySecret, hostData, cost, precedence)
messageSink.terminatorIdRef.Set(terminatorId)
terminatorId, err := xgress.AddTerminator(self.listener.factory, ns.Service.Id, "edge", "hosted:"+token, terminatorIdentity, terminatorIdentitySecret, hostData, cost, precedence)
messageSink.terminatorId = terminatorId
log.Debugf("registered listener for terminator %v, token: %v", terminatorId, token)
if err != nil {
messageSink.closeCB(messageSink.Id())
proxy.sendStateClosedReply(err.Error(), req)
messageSink.close(false, "") // don't notify here, as we're notifying next line with a response
self.sendStateClosedReply(err.Error(), req)
return
}
log.Debug("returning connection state CONNECTED to client")
proxy.sendStateConnectedReply(req, nil)
self.sendStateConnectedReply(req, nil)
}
func (proxy *ingressProxy) processUnbind(req *channel2.Message, ch channel2.Channel) {
func (self *edgeClientConn) processUnbind(req *channel2.Message, ch channel2.Channel) {
token := string(req.Body)
log := pfxlog.ContextLogger(ch.Label()).WithField("sessionId", token).WithFields(edge.GetLoggerFields(req))
@@ -324,39 +297,39 @@ func (proxy *ingressProxy) processUnbind(req *channel2.Message, ch channel2.Chan
if ns == nil {
log.WithField("token", token).Error("session not found")
proxy.sendStateClosedReply("Invalid Session", req)
self.sendStateClosedReply("Invalid Session", req)
return
}
if _, found := proxy.fingerprints.HasAny(ns.CertFingerprints); !found {
if _, found := self.fingerprints.HasAny(ns.CertFingerprints); !found {
log.WithField("token", token).
WithField("serviceFingerprints", ns.CertFingerprints).
WithField("clientFingerprints", proxy.fingerprints.Prints()).
WithField("clientFingerprints", self.fingerprints.Prints()).
Error("matching fingerprint not found for unbind")
proxy.sendStateClosedReply("Invalid Session", req)
self.sendStateClosedReply("Invalid Session", req)
return
}
localListener, ok := proxy.listener.factory.hostedServices.Get(token)
localListener, ok := self.listener.factory.hostedServices.Get(token)
if ok {
defer proxy.listener.factory.hostedServices.Delete(token)
defer self.listener.factory.hostedServices.Delete(token)
log.Debugf("removing terminator %v for token: %v", localListener.terminatorIdRef.Get(), token)
if err := xgress.RemoveTerminator(proxy.listener.factory, localListener.terminatorIdRef.Get()); err != nil {
proxy.sendStateClosedReply(err.Error(), req)
log.Debugf("removing terminator %v for token: %v", localListener.terminatorId, token)
if err := xgress.RemoveTerminator(self.listener.factory, localListener.terminatorId); err != nil {
self.sendStateClosedReply(err.Error(), req)
} else {
proxy.sendStateClosedReply("unbind successful", req)
self.sendStateClosedReply("unbind successful", req)
}
} else {
proxy.sendStateClosedReply("unbind successful", req)
self.sendStateClosedReply("unbind successful", req)
}
}
func (proxy *ingressProxy) processUpdateBind(req *channel2.Message, ch channel2.Channel) {
func (self *edgeClientConn) processUpdateBind(req *channel2.Message, ch channel2.Channel) {
token := string(req.Body)
log := pfxlog.ContextLogger(ch.Label()).WithField("sessionId", token).WithFields(edge.GetLoggerFields(req))
localListener, ok := proxy.listener.factory.hostedServices.Get(token)
localListener, ok := self.listener.factory.hostedServices.Get(token)
if !ok {
log.Error("failed to update bind, no listener found")
@@ -368,16 +341,16 @@ func (proxy *ingressProxy) processUpdateBind(req *channel2.Message, ch channel2.
if ns == nil {
log.WithField("token", token).Error("session not found")
proxy.sendStateClosedReply("Invalid Session", req)
self.sendStateClosedReply("Invalid Session", req)
return
}
if _, found := proxy.fingerprints.HasAny(ns.CertFingerprints); !found {
if _, found := self.fingerprints.HasAny(ns.CertFingerprints); !found {
log.WithField("token", token).
WithField("serviceFingerprints", ns.CertFingerprints).
WithField("clientFingerprints", proxy.fingerprints.Prints()).
WithField("clientFingerprints", self.fingerprints.Prints()).
Error("matching fingerprint not found for update bind")
proxy.sendStateClosedReply("Invalid Session", req)
self.sendStateClosedReply("Invalid Session", req)
return
}
@@ -398,13 +371,13 @@ func (proxy *ingressProxy) processUpdateBind(req *channel2.Message, ch channel2.
precedence = &updatedPrecedence
}
log.Debugf("updating terminator %v to precedence %v and cost %v", localListener.terminatorIdRef.Get(), precedence, cost)
if err := xgress.UpdateTerminator(proxy.listener.factory, localListener.terminatorIdRef.Get(), cost, precedence); err != nil {
log.Debugf("updating terminator %v to precedence %v and cost %v", localListener.terminatorId, precedence, cost)
if err := xgress.UpdateTerminator(self.listener.factory, localListener.terminatorId, cost, precedence); err != nil {
log.WithError(err).Error("failed to update bind")
}
}
func (proxy *ingressProxy) sendStateConnectedReply(req *channel2.Message, hostData map[uint32][]byte) {
func (self *edgeClientConn) sendStateConnectedReply(req *channel2.Message, hostData map[uint32][]byte) {
connId, _ := req.GetUint32Header(edge.ConnIdHeader)
msg := edge.NewStateConnectedMsg(connId)
@@ -417,19 +390,19 @@ func (proxy *ingressProxy) sendStateConnectedReply(req *channel2.Message, hostDa
}
msg.ReplyTo(req)
err := proxy.ch.SendPrioritizedWithTimeout(msg, channel2.High, time.Second*5)
err := self.ch.SendPrioritizedWithTimeout(msg, channel2.High, time.Second*5)
if err != nil {
pfxlog.Logger().WithFields(edge.GetLoggerFields(msg)).WithError(err).Error("failed to send state response")
return
}
}
func (proxy *ingressProxy) sendStateClosedReply(message string, req *channel2.Message) {
func (self *edgeClientConn) sendStateClosedReply(message string, req *channel2.Message) {
connId, _ := req.GetUint32Header(edge.ConnIdHeader)
msg := edge.NewStateClosedMsg(connId, message)
msg.ReplyTo(req)
syncC, err := proxy.ch.SendAndSyncWithPriority(msg, channel2.High)
syncC, err := self.ch.SendAndSyncWithPriority(msg, channel2.High)
if err != nil {
pfxlog.Logger().WithFields(edge.GetLoggerFields(msg)).WithError(err).Error("failed to send state response")
return
@@ -445,11 +418,11 @@ func (proxy *ingressProxy) sendStateClosedReply(message string, req *channel2.Me
}
}
func (proxy *ingressProxy) sendStateClosed(connId uint32, message string) {
func (self *edgeClientConn) sendStateClosed(connId uint32, message string) {
msg := edge.NewStateClosedMsg(connId, message)
pfxlog.Logger().WithFields(edge.GetLoggerFields(msg)).Debug("sending state closed message")
syncC, err := proxy.ch.SendAndSyncWithPriority(msg, channel2.High)
syncC, err := self.ch.SendAndSyncWithPriority(msg, channel2.High)
if err != nil {
pfxlog.Logger().WithFields(edge.GetLoggerFields(msg)).WithError(err).Error("failed to send state response")
return
@@ -464,14 +437,3 @@ func (proxy *ingressProxy) sendStateClosed(connId uint32, message string) {
pfxlog.Logger().WithFields(edge.GetLoggerFields(msg)).WithError(err).Error("timed out sending state response")
}
}
func (proxy *ingressProxy) closeConn(connId uint32) {
// This was done in the process loop, but all the relevant data structure are concurrent safe
// and if the the proxy closed before all the connections could be closed, this would lead to
// deadlocks
log := pfxlog.ContextLogger(proxy.ch.Label()).WithField("connId", connId)
log.Debug("closeConn()")
// we don't need to close the conn here, it will get closed when the xgress closes its peer
proxy.msgMux.RemoveMsgSinkById(connId)
}
router/xgress_edge/perf_test.go
+2 -2
View File
@@ -86,14 +86,14 @@ func writePerf(b *testing.B, mux edge.MsgMux) {
listener := &listener{}
proxy := &ingressProxy{
proxy := &edgeClientConn{
msgMux: mux,
listener: listener,
fingerprints: nil,
ch: testChannel,
}
conn := &localMessageSink{
conn := &edgeXgressConn{
MsgChannel: *edge.NewEdgeMsgChannel(proxy.ch, 1),
seq: NewMsgQueue(4),
}
tests/%SystemDrive%/ProgramData/Microsoft/Windows/Caches/cversions.2.db
BIN
View File
Binary file not shown.
tests/%SystemDrive%/ProgramData/Microsoft/Windows/Caches/{6AF0698E-D558-4F6E-9B3C-3716689AF493}.2.ver0x0000000000000001.db
BIN
View File
Binary file not shown.
tests/%SystemDrive%/ProgramData/Microsoft/Windows/Caches/{DDF571F2-BE98-426D-8288-1A9A39C3FDA2}.2.ver0x0000000000000001.db
BIN
View File
Binary file not shown.