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

Don't return response to listen until terminator is established

Browse Source
This commit is contained in:
Paul Lorenz
2024-02-05 18:23:07 -05:00
parent 3aace85a97
commit e438f89c1d
3 changed files with 81 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
router/xgress_edge/fabric.go
+23 -18
View File
@@ -85,6 +85,7 @@ type edgeTerminator struct {
retryDelay time.Duration
establishActive atomic.Bool
createTime time.Time
establishCallback func(ok bool, msg string)
}
func (self *edgeTerminator) calculateRetry(queueFailed bool) {
@@ -115,7 +116,7 @@ func (self *edgeTerminator) inspect(fixInvalidTerminators bool) (*edge.InspectRe
}
result, err := edge.UnmarshalInspectResult(resp)
if result != nil && result.Type != edge.ConnTypeBind && fixInvalidTerminators {
self.close(true, "terminator invalid")
self.close(true, true, "terminator invalid")
}
return result, err
}
@@ -129,13 +130,13 @@ func (self *edgeTerminator) nextDialConnId() uint32 {
return nextId
}
func (self *edgeTerminator) close(notify bool, reason string) {
func (self *edgeTerminator) close(notifySdk bool, notifyCtrl bool, reason string) {
logger := pfxlog.Logger().
WithField("terminatorId", self.terminatorId.Load()).
WithField("token", self.token).
WithField("reason", reason)
if notify && !self.IsClosed() {
if notifySdk && !self.IsClosed() {
// Notify edge client of close
logger.Debug("sending closed to SDK client")
closeMsg := edge.NewStateClosedMsg(self.Id(), reason)
@@ -152,6 +153,24 @@ func (self *edgeTerminator) close(notify bool, reason string) {
self.state.Store(TerminatorStateDeleting)
self.edgeClientConn.listener.factory.hostedServices.Delete(terminatorId)
if notifyCtrl {
logger.Info("removing terminator on controller")
ctrlCh := self.edgeClientConn.listener.factory.ctrls.AnyCtrlChannel()
if ctrlCh == nil {
logger.Error("no controller available, unable to remove terminator")
} else if err := self.edgeClientConn.removeTerminator(ctrlCh, self.token, terminatorId); err != nil {
logger.WithError(err).Error("failed to remove terminator")
} else {
logger.Info("Successfully removed terminator")
}
}
} else {
logger.Warn("edge terminator closing, but no terminator id set, so can't remove on controller")
}
}
} else {
if notifyCtrl {
if terminatorId := self.terminatorId.Load(); terminatorId != "" {
logger.Info("removing terminator on controller")
ctrlCh := self.edgeClientConn.listener.factory.ctrls.AnyCtrlChannel()
if ctrlCh == nil {
@@ -159,26 +178,12 @@ func (self *edgeTerminator) close(notify bool, reason string) {
} else if err := self.edgeClientConn.removeTerminator(ctrlCh, self.token, terminatorId); err != nil {
logger.WithError(err).Error("failed to remove terminator")
} else {
logger.Info("Successfully removed terminator")
logger.Info("successfully removed terminator")
}
} else {
logger.Warn("edge terminator closing, but no terminator id set, so can't remove on controller")
}
}
} else {
if terminatorId := self.terminatorId.Load(); terminatorId != "" {
logger.Info("removing terminator on controller")
ctrlCh := self.edgeClientConn.listener.factory.ctrls.AnyCtrlChannel()
if ctrlCh == nil {
logger.Error("no controller available, unable to remove terminator")
} else if err := self.edgeClientConn.removeTerminator(ctrlCh, self.token, terminatorId); err != nil {
logger.WithError(err).Error("failed to remove terminator")
} else {
logger.Info("successfully removed terminator")
}
} else {
logger.Warn("edge terminator closing, but no terminator id set, so can't remove on controller")
}
logger.Debug("removing terminator on router")
self.edgeClientConn.listener.factory.hostedServices.Delete(self.token)
router/xgress_edge/hosted.go
+9 -11
View File
@@ -182,7 +182,7 @@ func (self *hostedServiceRegistry) cleanupServices(proxy *edgeClientConn) {
self.services.Range(func(key, value interface{}) bool {
terminator := value.(*edgeTerminator)
if terminator.edgeClientConn == proxy {
terminator.close(false, "") // don't notify, channel is already closed, we can't send messages
terminator.close(false, true, "channel closed") // don't notify, channel is already closed, we can't send messages
self.services.Delete(key)
}
return true
@@ -193,7 +193,7 @@ func (self *hostedServiceRegistry) cleanupDuplicates(newest *edgeTerminator) {
self.services.Range(func(key, value interface{}) bool {
terminator := value.(*edgeTerminator)
if terminator != newest && newest.token == terminator.token && newest.instance == terminator.instance {
terminator.close(false, "duplicate terminator") // don't notify, channel is already closed, we can't send messages
terminator.close(false, true, "duplicate terminator") // don't notify, channel is already closed, we can't send messages
self.services.Delete(key)
pfxlog.Logger().WithField("routerId", terminator.edgeClientConn.listener.id.Token).
WithField("token", terminator.token).
@@ -211,7 +211,7 @@ func (self *hostedServiceRegistry) unbindSession(connId uint32, sessionToken str
self.services.Range(func(key, value interface{}) bool {
terminator := value.(*edgeTerminator)
if terminator.MsgChannel.Id() == connId && terminator.token == sessionToken && terminator.edgeClientConn == proxy {
terminator.close(false, "unbind successful") // don't notify, sdk asked us to unbind
terminator.close(false, true, "unbind successful") // don't notify, sdk asked us to unbind
self.services.Delete(key)
pfxlog.Logger().WithField("routerId", terminator.edgeClientConn.listener.id.Token).
WithField("token", sessionToken).
@@ -342,7 +342,10 @@ func (self *hostedServiceRegistry) HandleCreateTerminatorResponse(msg *channel.M
}
if response.Result != edge_ctrl_pb.CreateTerminatorResult_Success {
terminator.close(true, response.Msg)
if terminator.establishCallback != nil {
terminator.establishCallback(false, response.Msg)
}
terminator.close(true, false, response.Msg)
return
}
@@ -364,13 +367,8 @@ func (self *hostedServiceRegistry) HandleCreateTerminatorResponse(msg *channel.M
}
}
if isValid && terminator.notifyEstablished {
notifyMsg := channel.NewMessage(edge.ContentTypeBindSuccess, nil)
notifyMsg.PutUint32Header(edge.ConnIdHeader, terminator.MsgChannel.Id())
if err := notifyMsg.WithTimeout(time.Second * 30).Send(terminator.MsgChannel.Channel); err != nil {
log.WithError(err).Error("failed to send bind success")
}
if isValid && terminator.establishCallback != nil {
terminator.establishCallback(true, "terminator established")
}
}
router/xgress_edge/listener.go
+49 -33
View File
@@ -265,7 +265,7 @@ func (self *edgeClientConn) processBindV1(req *channel.Message, ch channel.Chann
// need to remove session remove listener on close
messageSink.onClose = self.listener.factory.stateManager.AddEdgeSessionRemovedListener(token, func(token string) {
messageSink.close(true, "session ended")
messageSink.close(true, true, "session ended")
})
self.listener.factory.hostedServices.Put(token, messageSink)
@@ -290,7 +290,7 @@ func (self *edgeClientConn) processBindV1(req *channel.Message, ch channel.Chann
responseMsg, err := protobufs.MarshalTyped(request).WithTimeout(timeout).SendForReply(ctrlCh)
if err = xgress_common.CheckForFailureResult(responseMsg, err, edge_ctrl_pb.ContentType_CreateTerminatorResponseType); err != nil {
log.WithError(err).Warn("error creating terminator")
messageSink.close(false, "") // don't notify here, as we're notifying next line with a response
messageSink.close(false, false, "") // don't notify here, as we're notifying next line with a response
self.sendStateClosedReply(err.Error(), req)
return
}
@@ -301,7 +301,7 @@ func (self *edgeClientConn) processBindV1(req *channel.Message, ch channel.Chann
if messageSink.MsgChannel.IsClosed() {
log.Warn("edge channel closed while setting up terminator. cleaning up terminator now")
messageSink.close(false, "edge channel closed")
messageSink.close(false, true, "edge channel closed")
return
}
@@ -391,43 +391,59 @@ func (self *edgeClientConn) processBindV2(req *channel.Message, ch channel.Chann
notifyEstablished, _ := req.GetBoolHeader(edge.SupportsBindSuccessHeader)
terminator := &edgeTerminator{
MsgChannel: *edge.NewEdgeMsgChannel(self.ch, connId),
edgeClientConn: self,
token: token,
cost: cost,
precedence: precedence,
instance: terminatorInstance,
instanceSecret: terminatorInstanceSecret,
hostData: hostData,
assignIds: assignIds,
v2: true,
postValidate: postValidate,
notifyEstablished: notifyEstablished,
createTime: time.Now(),
MsgChannel: *edge.NewEdgeMsgChannel(self.ch, connId),
edgeClientConn: self,
token: token,
cost: cost,
precedence: precedence,
instance: terminatorInstance,
instanceSecret: terminatorInstanceSecret,
hostData: hostData,
assignIds: assignIds,
v2: true,
postValidate: postValidate,
createTime: time.Now(),
}
terminator.establishCallback = func(ok bool, msg string) {
if ok {
self.sendStateConnectedReply(req, nil)
if notifyEstablished {
notifyMsg := channel.NewMessage(edge.ContentTypeBindSuccess, nil)
notifyMsg.PutUint32Header(edge.ConnIdHeader, terminator.MsgChannel.Id())
if err := notifyMsg.WithTimeout(time.Second * 30).Send(terminator.MsgChannel.Channel); err != nil {
log.WithError(err).Error("failed to send bind success")
}
}
} else {
self.sendStateClosedReply(msg, req)
}
}
terminator.terminatorId.Store(terminatorId)
terminator.state.Store(TerminatorStatePendingEstablishment)
// need to remove session remove listener on close
terminator.onClose = self.listener.factory.stateManager.AddEdgeSessionRemovedListener(token, func(token string) {
terminator.close(true, "session ended")
})
// If the session was recently removed, the call to AddEdgeSessionRemovedListener will have asynchronously closed
// the terminator
if self.listener.factory.stateManager.WasSessionRecentlyRemoved(token) {
log.Info("invalid session, not establishing terminator")
} else {
log.Info("establishing terminator")
self.sendStateConnectedReply(req, nil)
self.listener.factory.hostedServices.EstablishTerminator(terminator)
if listenerId == "" {
// only removed dupes with a scan if we don't have an sdk provided key
self.listener.factory.hostedServices.cleanupDuplicates(terminator)
}
terminator.establishCallback(false, "invalid session")
return
}
// need to remove session remove listener on close
terminator.onClose = self.listener.factory.stateManager.AddEdgeSessionRemovedListener(token, func(token string) {
terminator.close(true, true, "session ended")
})
log.Info("establishing terminator")
self.listener.factory.hostedServices.EstablishTerminator(terminator)
if listenerId == "" {
// only removed dupes with a scan if we don't have an sdk provided key
self.listener.factory.hostedServices.cleanupDuplicates(terminator)
}
}
func (self *edgeClientConn) processUnbind(req *channel.Message, _ channel.Channel) {