diff --git a/router/xgress_edge/fabric.go b/router/xgress_edge/fabric.go index 24b5bd1fa..e1e35a981 100644 --- a/router/xgress_edge/fabric.go +++ b/router/xgress_edge/fabric.go @@ -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) diff --git a/router/xgress_edge/hosted.go b/router/xgress_edge/hosted.go index 4580703e0..ab19dbe2f 100644 --- a/router/xgress_edge/hosted.go +++ b/router/xgress_edge/hosted.go @@ -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") } } diff --git a/router/xgress_edge/listener.go b/router/xgress_edge/listener.go index 4d18789d3..3fec41ef3 100644 --- a/router/xgress_edge/listener.go +++ b/router/xgress_edge/listener.go @@ -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) {