diff --git a/go.mod b/go.mod index 0e951b0c9..6b5305e31 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 29ec5a757..bcee5e6ec 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/router/xgress_edge/accept.go b/router/xgress_edge/accept.go index 1b8d40343..a8d2c40c0 100644 --- a/router/xgress_edge/accept.go +++ b/router/xgress_edge/accept.go @@ -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()), diff --git a/router/xgress_edge/dialer.go b/router/xgress_edge/dialer.go index c9d983ba3..346e05333 100644 --- a/router/xgress_edge/dialer.go +++ b/router/xgress_edge/dialer.go @@ -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) diff --git a/router/xgress_edge/fabric.go b/router/xgress_edge/fabric.go index 70a061fa6..b9d1082f6 100644 --- a/router/xgress_edge/fabric.go +++ b/router/xgress_edge/fabric.go @@ -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 { diff --git a/router/xgress_edge/hosted.go b/router/xgress_edge/hosted.go index fa722985a..2a0c7ad11 100644 --- a/router/xgress_edge/hosted.go +++ b/router/xgress_edge/hosted.go @@ -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) } diff --git a/router/xgress_edge/listener.go b/router/xgress_edge/listener.go index b199e8421..66269b0ee 100644 --- a/router/xgress_edge/listener.go +++ b/router/xgress_edge/listener.go @@ -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) -} diff --git a/router/xgress_edge/perf_test.go b/router/xgress_edge/perf_test.go index 8cc643b23..3302d74ef 100644 --- a/router/xgress_edge/perf_test.go +++ b/router/xgress_edge/perf_test.go @@ -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), } diff --git a/tests/%SystemDrive%/ProgramData/Microsoft/Windows/Caches/cversions.2.db b/tests/%SystemDrive%/ProgramData/Microsoft/Windows/Caches/cversions.2.db deleted file mode 100644 index 6493406c6..000000000 Binary files a/tests/%SystemDrive%/ProgramData/Microsoft/Windows/Caches/cversions.2.db and /dev/null differ diff --git a/tests/%SystemDrive%/ProgramData/Microsoft/Windows/Caches/{6AF0698E-D558-4F6E-9B3C-3716689AF493}.2.ver0x0000000000000001.db b/tests/%SystemDrive%/ProgramData/Microsoft/Windows/Caches/{6AF0698E-D558-4F6E-9B3C-3716689AF493}.2.ver0x0000000000000001.db deleted file mode 100644 index 838a2652a..000000000 Binary files a/tests/%SystemDrive%/ProgramData/Microsoft/Windows/Caches/{6AF0698E-D558-4F6E-9B3C-3716689AF493}.2.ver0x0000000000000001.db and /dev/null differ diff --git a/tests/%SystemDrive%/ProgramData/Microsoft/Windows/Caches/{DDF571F2-BE98-426D-8288-1A9A39C3FDA2}.2.ver0x0000000000000001.db b/tests/%SystemDrive%/ProgramData/Microsoft/Windows/Caches/{DDF571F2-BE98-426D-8288-1A9A39C3FDA2}.2.ver0x0000000000000001.db deleted file mode 100644 index e71ff0a30..000000000 Binary files a/tests/%SystemDrive%/ProgramData/Microsoft/Windows/Caches/{DDF571F2-BE98-426D-8288-1A9A39C3FDA2}.2.ver0x0000000000000001.db and /dev/null differ