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

If conn is closed after terminator create, remove terminator.

Browse Source 
Also clean up some logging and make sure terminatorId is handled
in a threadsafe manner. Fixes #1027
This commit is contained in:
Paul Lorenz
2022-06-01 17:13:54 -04:00
parent 32bf51f193
commit bfdb275fd0
3 changed files with 27 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
router/xgress_edge/dialer.go
+1 -1
View File
@@ -50,7 +50,7 @@ func (dialer dialer) IsTerminatorValid(id string, destination string) bool {
pfxlog.Logger().Debug("looking up hosted service conn")
terminator, found := dialer.factory.hostedServices.Get(token)
return found && terminator.terminatorId == id
return found && terminator.terminatorId.Load() == id
}
func newDialer(factory *Factory, options *Options) xgress.Dialer {
router/xgress_edge/fabric.go
+3 -3
View File
@@ -42,7 +42,7 @@ type edgeTerminator struct {
edge.MsgChannel
edgeClientConn *edgeClientConn
token string
terminatorId string
terminatorId concurrenz.AtomicValue[string]
assignIds bool
onClose func()
}
@@ -58,7 +58,7 @@ func (self *edgeTerminator) nextDialConnId() uint32 {
func (self *edgeTerminator) close(notify bool, reason string) {
logger := pfxlog.Logger().
WithField("terminatorId", self.terminatorId).
WithField("terminatorId", self.terminatorId.Load()).
WithField("token", self.token)
if notify && !self.IsClosed() {
@@ -70,7 +70,7 @@ func (self *edgeTerminator) close(notify bool, reason string) {
}
}
if self.terminatorId != "" {
if self.terminatorId.Load() != "" {
logger.Info("removing terminator on controller")
if err := self.edgeClientConn.removeTerminator(self); err != nil {
logger.WithError(err).Error("failed to remove terminator")
router/xgress_edge/listener.go
+23 -14
View File
@@ -103,10 +103,11 @@ func (self *edgeClientConn) HandleClose(_ channel.Channel) {
log.Debugf("closing")
terminators := self.listener.factory.hostedServices.cleanupServices(self)
for _, terminator := range terminators {
tLog := log.WithField("terminatorId", terminator.terminatorId.Load()).WithField("token", terminator.token)
if err := self.removeTerminator(terminator); err != nil {
log.Warnf("failed to remove terminator %v for session with token %v on channel close", terminator.terminatorId, terminator.token)
tLog.Warn("failed to remove terminator for session on channel close")
} else {
log.WithField("terminatorId", terminator.terminatorId).WithField("token", terminator.token).Info("Successfully removed terminator on channel close")
tLog.Info("Successfully removed terminator on channel close")
}
}
self.msgMux.Close()
@@ -186,7 +187,8 @@ func (self *edgeClientConn) processBind(req *channel.Message, ch channel.Channel
log := pfxlog.ContextLogger(ch.Label()).
WithField("sessionId", token).
WithFields(edge.GetLoggerFields(req))
WithFields(edge.GetLoggerFields(req)).
WithField("routerId", self.listener.id.Token)
connId, found := req.GetUint32Header(edge.ConnIdHeader)
if !found {
@@ -221,6 +223,7 @@ func (self *edgeClientConn) processBind(req *channel.Message, ch channel.Channel
log.Debugf("client requested router provided connection ids: %v", assignIds)
log.Debug("establishing listener")
messageSink := &edgeTerminator{
MsgChannel: *edge.NewEdgeMsgChannel(self.ch, connId),
edgeClientConn: self,
@@ -260,14 +263,21 @@ func (self *edgeClientConn) processBind(req *channel.Message, ch channel.Channel
return
}
messageSink.terminatorId = string(responseMsg.Body)
terminatorId := string(responseMsg.Body)
messageSink.terminatorId.Store(terminatorId)
log = log.WithField("terminatorId", terminatorIdentity)
log.Debugf("registered listener for terminator %v, token: %v", messageSink.terminatorId, token)
if messageSink.MsgChannel.IsClosed() {
log.Warn("edge channel closed while setting up terminator. cleaning up terminator now")
messageSink.close(false, "edge channel closed")
return
}
log.Debug("registered listener for terminator")
log.Debug("returning connection state CONNECTED to client")
self.sendStateConnectedReply(req, nil)
log.WithField("routerId", self.listener.id.Token).
WithField("terminatorId", messageSink.terminatorId).Info("created terminator")
log.Info("created terminator")
}
func (self *edgeClientConn) processUnbind(req *channel.Message, ch channel.Channel) {
@@ -279,7 +289,7 @@ func (self *edgeClientConn) processUnbind(req *channel.Message, ch channel.Chann
terminator, ok := self.listener.factory.hostedServices.Get(token)
if ok {
log = log.WithField("routerId", self.listener.id.Token).
WithField("terminatorId", terminator.terminatorId)
WithField("terminatorId", terminator.terminatorId.Load())
defer self.listener.factory.hostedServices.Delete(token)
@@ -300,7 +310,7 @@ func (self *edgeClientConn) removeTerminator(terminator *edgeTerminator) error {
request := &edge_ctrl_pb.RemoveTerminatorRequest{
SessionToken: terminator.token,
Fingerprints: self.fingerprints.Prints(),
TerminatorId: terminator.terminatorId,
TerminatorId: terminator.terminatorId.Load(),
}
timeout := self.listener.factory.DefaultRequestTimeout()
@@ -322,7 +332,7 @@ func (self *edgeClientConn) processUpdateBind(req *channel.Message, ch channel.C
request := &edge_ctrl_pb.UpdateTerminatorRequest{
SessionToken: token,
Fingerprints: self.fingerprints.Prints(),
TerminatorId: terminator.terminatorId,
TerminatorId: terminator.terminatorId.Load(),
}
if costVal, hasCost := req.GetUint16Header(edge.CostHeader); hasCost {
@@ -341,7 +351,7 @@ func (self *edgeClientConn) processUpdateBind(req *channel.Message, ch channel.C
}
}
log = log.WithField("terminator", terminator.terminatorId).
log = log.WithField("terminator", terminator.terminatorId.Load()).
WithField("precedence", request.Precedence).
WithField("cost", request.Cost).
WithField("updatingPrecedence", request.UpdatePrecedence).
@@ -374,12 +384,11 @@ func (self *edgeClientConn) processHealthEvent(req *channel.Message, ch channel.
request := &edge_ctrl_pb.HealthEventRequest{
SessionToken: token,
Fingerprints: self.fingerprints.Prints(),
TerminatorId: terminator.terminatorId,
TerminatorId: terminator.terminatorId.Load(),
CheckPassed: checkPassed,
}
log = log.WithField("terminator", terminator.terminatorId).
WithField("checkPassed", checkPassed)
log = log.WithField("terminator", terminator.terminatorId.Load()).WithField("checkPassed", checkPassed)
log.Debug("sending health event")
if err := protobufs.MarshalTyped(request).Send(self.listener.factory.Channel()); err != nil {