From c7ae352255c69bb873695b46738c04509cb6ecd9 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Mon, 10 Aug 2020 17:06:57 -0400 Subject: [PATCH] Fix xgress start race condition. Fixes GH-117 --- controller/handler_ctrl/session_request.go | 18 +-- controller/network/network.go | 17 --- pb/ctrl_pb/ctrl.pb.go | 124 ++++++++++----------- pb/ctrl_pb/ctrl.proto | 1 - pb/ctrl_pb/decoder.go | 9 -- router/forwarder/forwarder.go | 13 --- router/handler_ctrl/bind.go | 1 - router/handler_ctrl/start_xgress.go | 44 -------- router/xgress/messages.go | 9 +- router/xgress/request.go | 9 -- router/xgress/xgress.go | 96 ++++++++++------ router/xgress_transport/dialer.go | 1 + router/xgress_transport_udp/dialer.go | 1 + version | 2 +- 14 files changed, 135 insertions(+), 210 deletions(-) delete mode 100644 router/handler_ctrl/start_xgress.go diff --git a/controller/handler_ctrl/session_request.go b/controller/handler_ctrl/session_request.go index d77ceb351..117d91819 100644 --- a/controller/handler_ctrl/session_request.go +++ b/controller/handler_ctrl/session_request.go @@ -65,25 +65,17 @@ func (h *sessionRequestHandler) HandleReceive(msg *channel2.Message, ch channel2 responseMsg.Headers[int32(k)] = v } - if startXgressSession, err := h.r.Control.SendAndWaitWithTimeout(responseMsg, time.Second*10); err != nil { - log.Errorf("unable to respond (%s)", err) - h.network.RemoveSession(session.Id, true) - } else { - if startXgressSession.ContentType == int32(ctrl_pb.ContentType_StartXgressType) { - if err = h.network.StartSessionEgress(session.Id); err != nil { - log.WithError(err).Error("failure starting xgress") - h.network.RemoveSession(session.Id, true) - } - } else { - log.Errorf("unexpected reply to dial response: %v", startXgressSession.ContentType) - h.network.RemoveSession(session.Id, true) + if err := h.r.Control.SendWithTimeout(responseMsg, time.Second*10); err != nil { + log.Errorf("unable to respond with success to create session request for session %v (%s)", session.Id, err) + if err := h.network.RemoveSession(session.Id, true); err != nil { + log.Errorf("unable to remove session %v (%v)", session.Id, err) } } } else { responseMsg := ctrl_msg.NewSessionFailedMsg(err.Error()) responseMsg.ReplyTo(msg) if err := h.r.Control.Send(responseMsg); err != nil { - log.Errorf("unable to respond (%s)", err) + log.Errorf("unable to respond with failure to create session request for service %v (%s)", request.ServiceId, err) } } }() diff --git a/controller/network/network.go b/controller/network/network.go index bd06b477d..623aac230 100644 --- a/controller/network/network.go +++ b/controller/network/network.go @@ -450,15 +450,6 @@ func (network *Network) RemoveSession(sessionId *identity.TokenId, now bool) err return InvalidSessionError{sessionId: sessionId.Token} } -func (network *Network) StartSessionEgress(sessionId *identity.TokenId) error { - if ss, found := network.sessionController.get(sessionId); found { - terminatingRouter := ss.Circuit.Path[len(ss.Circuit.Path)-1] - pfxlog.Logger().Debugf("started session egress [s/%s]", ss.Id.Token) - return sendStartXgress(terminatingRouter, sessionId) - } - return fmt.Errorf("invalid session (%s)", sessionId.Token) -} - func (network *Network) CreateCircuit(srcR, dstR *Router) (*Circuit, error) { ingressId, err := network.sequence.NextHash() if err != nil { @@ -737,14 +728,6 @@ func sendUnroute(r *Router, sessionId *identity.TokenId, now bool) error { return nil } -func sendStartXgress(r *Router, sessionId *identity.TokenId) error { - msg := channel2.NewMessage(int32(ctrl_pb.ContentType_StartXgressType), []byte(sessionId.Token)) - if err := r.Control.Send(msg); err != nil { - return err - } - return nil -} - func (network *Network) showOptions() { if jsonOptions, err := json.MarshalIndent(network.options, "", " "); err == nil { pfxlog.Logger().Infof("network = %s", string(jsonOptions)) diff --git a/pb/ctrl_pb/ctrl.pb.go b/pb/ctrl_pb/ctrl.pb.go index a6d6b4851..2fb5416a9 100644 --- a/pb/ctrl_pb/ctrl.pb.go +++ b/pb/ctrl_pb/ctrl.pb.go @@ -38,7 +38,6 @@ const ( ContentType_RemoveTerminatorRequestType ContentType = 1012 ContentType_InspectRequestType ContentType = 1013 ContentType_InspectResponseType ContentType = 1014 - ContentType_StartXgressType ContentType = 1015 ContentType_SessionFailedType ContentType = 1016 ContentType_ValidateTerminatorsRequestType ContentType = 1017 ContentType_UpdateTerminatorRequestType ContentType = 10018 @@ -60,7 +59,6 @@ var ContentType_name = map[int32]string{ 1012: "RemoveTerminatorRequestType", 1013: "InspectRequestType", 1014: "InspectResponseType", - 1015: "StartXgressType", 1016: "SessionFailedType", 1017: "ValidateTerminatorsRequestType", 10018: "UpdateTerminatorRequestType", @@ -82,7 +80,6 @@ var ContentType_value = map[string]int32{ "RemoveTerminatorRequestType": 1012, "InspectRequestType": 1013, "InspectResponseType": 1014, - "StartXgressType": 1015, "SessionFailedType": 1016, "ValidateTerminatorsRequestType": 1017, "UpdateTerminatorRequestType": 10018, @@ -1006,66 +1003,65 @@ func init() { } var fileDescriptor_a0572e205f89e843 = []byte{ - // 966 bytes of a gzipped FileDescriptorProto + // 954 bytes of a gzipped FileDescriptorProto 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xb4, 0x56, 0x5b, 0x6f, 0xe3, 0x44, - 0x14, 0xc6, 0xce, 0x65, 0x92, 0x93, 0x34, 0x0d, 0xd3, 0xa5, 0x8d, 0x02, 0xac, 0x2a, 0xaf, 0x90, - 0xaa, 0x4a, 0x04, 0x54, 0x84, 0xb4, 0x2c, 0xaa, 0xa0, 0xea, 0x45, 0x14, 0x01, 0xaa, 0x26, 0xdd, - 0x15, 0xe2, 0xcd, 0xb5, 0x67, 0x2b, 0xb3, 0xa9, 0x1d, 0xc6, 0x93, 0xac, 0xfa, 0x77, 0xf8, 0x09, - 0x3c, 0xf3, 0xce, 0x0b, 0xe2, 0x17, 0xf0, 0x03, 0xb8, 0x83, 0xb8, 0xf3, 0xc6, 0xcc, 0x99, 0x89, - 0x33, 0x0e, 0x09, 0x2a, 0x42, 0xfb, 0x36, 0xe7, 0xe6, 0x73, 0xbe, 0xf3, 0x9d, 0x33, 0x63, 0x80, - 0x48, 0x8a, 0xd1, 0x60, 0x2c, 0x32, 0x99, 0x51, 0x62, 0xce, 0x17, 0xc1, 0xe7, 0x1e, 0x74, 0x86, - 0x3c, 0xcf, 0x93, 0x2c, 0x65, 0xfc, 0xa3, 0x09, 0xcf, 0x25, 0x7d, 0x0e, 0x9a, 0x49, 0x7a, 0x29, - 0x94, 0xf2, 0x34, 0xee, 0x79, 0xdb, 0xde, 0x4e, 0x93, 0xcd, 0x15, 0xda, 0x9a, 0x73, 0x31, 0x4d, - 0x22, 0xae, 0xac, 0xbe, 0xb1, 0x16, 0x0a, 0x7a, 0x00, 0x8d, 0x31, 0xe7, 0xe2, 0x28, 0x94, 0x61, - 0xaf, 0xb2, 0x5d, 0xd9, 0x69, 0xed, 0xbd, 0x30, 0xb0, 0xa9, 0x06, 0xe5, 0x34, 0x83, 0x33, 0xeb, - 0x77, 0x9c, 0x4a, 0x71, 0xcd, 0x8a, 0xb0, 0xfe, 0xeb, 0xb0, 0x56, 0x32, 0xd1, 0x2e, 0x54, 0x1e, - 0xf1, 0x6b, 0xac, 0x64, 0x8d, 0xe9, 0x23, 0xbd, 0x05, 0xb5, 0x69, 0x38, 0x9a, 0x70, 0xcc, 0xdf, - 0x66, 0x46, 0xb8, 0xe7, 0xdf, 0xf5, 0x82, 0xcf, 0x7c, 0xd8, 0x3a, 0x14, 0x3c, 0x94, 0xfc, 0x9c, - 0x8b, 0xab, 0x24, 0x0d, 0x65, 0x26, 0x1c, 0x5c, 0xff, 0x52, 0x79, 0x0f, 0xc8, 0x45, 0x92, 0xc6, - 0x0a, 0xa8, 0x2a, 0x5c, 0xdb, 0x66, 0xa2, 0xb6, 0x84, 0x71, 0xac, 0xe1, 0xf7, 0xaa, 0xc6, 0x62, - 0x45, 0xfa, 0xb6, 0x83, 0xb6, 0x86, 0x68, 0x07, 0x05, 0xda, 0x15, 0x55, 0xac, 0x82, 0x4d, 0x29, - 0x54, 0xa3, 0x2c, 0x97, 0xbd, 0x3a, 0xc2, 0xc4, 0x33, 0xdd, 0x07, 0x18, 0x0b, 0x1e, 0xf1, 0x98, - 0xa7, 0x11, 0xef, 0x11, 0x65, 0xe9, 0xec, 0x3d, 0x5f, 0x64, 0x98, 0x7f, 0xfb, 0xac, 0x70, 0x62, - 0x4e, 0xc0, 0xff, 0xeb, 0xe4, 0x3e, 0x6c, 0x31, 0x7e, 0x95, 0x4d, 0x97, 0x34, 0x32, 0x80, 0xb6, - 0x2c, 0x94, 0xc5, 0x8c, 0x94, 0x74, 0xc1, 0x19, 0xc0, 0x3c, 0x90, 0x76, 0xc0, 0x4f, 0x66, 0x7e, - 0xea, 0xe4, 0x36, 0xdb, 0x5f, 0xd9, 0xec, 0x4a, 0xa9, 0xd9, 0xc1, 0x10, 0xfa, 0x0f, 0xc2, 0x51, - 0x12, 0x97, 0xba, 0x9a, 0xcf, 0x6a, 0x7a, 0x15, 0x5a, 0xf3, 0xfc, 0xb9, 0x4a, 0xa5, 0xd9, 0xd8, - 0x58, 0xd2, 0x2b, 0xe6, 0xfa, 0x05, 0x5f, 0x7a, 0xb0, 0x75, 0x7f, 0x1c, 0x2f, 0x9d, 0x97, 0x1b, - 0xc0, 0xa4, 0xbb, 0xd0, 0x9d, 0x60, 0xf8, 0x9c, 0x02, 0x44, 0xd4, 0x60, 0xff, 0xd0, 0xd3, 0xdb, - 0x00, 0x46, 0x77, 0xa8, 0x79, 0xae, 0xa0, 0x97, 0xa3, 0x59, 0x60, 0xbb, 0xfa, 0x1f, 0xd9, 0x2e, - 0x06, 0xa8, 0x36, 0x1f, 0xa0, 0xe0, 0x65, 0xa8, 0x1e, 0x25, 0xe1, 0x68, 0x59, 0xff, 0x67, 0x5d, - 0xf6, 0xcb, 0x5d, 0xde, 0x84, 0xea, 0x3b, 0x49, 0xfa, 0x68, 0x31, 0x22, 0x78, 0x0b, 0x6a, 0x27, - 0xe1, 0x64, 0x24, 0xe9, 0x4b, 0x40, 0xf2, 0xc9, 0xc5, 0x87, 0x3c, 0x92, 0x68, 0xed, 0xec, 0x3d, - 0x53, 0x94, 0x88, 0x0e, 0x43, 0x63, 0x64, 0x33, 0x2f, 0xfb, 0x25, 0xbf, 0xf8, 0xd2, 0x27, 0x15, - 0xa8, 0xb1, 0x6c, 0x22, 0xb9, 0x59, 0x48, 0xbc, 0x13, 0xe6, 0x17, 0x4d, 0xa1, 0xa0, 0x2f, 0x42, - 0x9d, 0x5f, 0x16, 0x25, 0xb6, 0x9c, 0x3c, 0x18, 0x3d, 0x38, 0x46, 0x23, 0xb3, 0x4e, 0x74, 0x0f, - 0x1a, 0x0f, 0x33, 0xf1, 0x38, 0x14, 0x71, 0x6e, 0x6f, 0x9e, 0xcd, 0x85, 0x80, 0x13, 0x63, 0x66, - 0x85, 0x5f, 0x5f, 0xb1, 0x5f, 0x37, 0x9f, 0x71, 0x27, 0xd2, 0x5b, 0x39, 0x91, 0xe5, 0x5e, 0xd1, - 0x6d, 0x68, 0xc5, 0x6a, 0x50, 0x34, 0x2b, 0xaa, 0x64, 0x3b, 0xaf, 0xae, 0x8a, 0xbe, 0xe1, 0x5c, - 0x10, 0x55, 0x2c, 0xea, 0xce, 0x52, 0x14, 0x4f, 0xe4, 0x32, 0xec, 0x9f, 0x02, 0xb1, 0x98, 0xf5, - 0xec, 0xe5, 0x22, 0x3a, 0xb0, 0x38, 0x0c, 0x42, 0x47, 0xa3, 0xed, 0x71, 0x2e, 0x0f, 0x4a, 0x38, - 0x1d, 0x4d, 0xf0, 0x1a, 0x90, 0xfb, 0xa9, 0xb8, 0x01, 0x6b, 0xaa, 0xbe, 0x34, 0x7b, 0x6c, 0x77, - 0x40, 0x1f, 0x83, 0x7b, 0xd0, 0x39, 0x4d, 0xf3, 0xb1, 0x9e, 0x09, 0xbb, 0x58, 0x3b, 0xb0, 0x2e, - 0xcc, 0x91, 0xc7, 0x0f, 0x74, 0xb5, 0x66, 0x5f, 0x9b, 0x6c, 0x51, 0x1d, 0x7c, 0xea, 0xc1, 0x7a, - 0x11, 0x9c, 0x8f, 0xb3, 0x34, 0xe7, 0x9a, 0x8f, 0x7c, 0x12, 0x45, 0x33, 0x1c, 0x0d, 0x36, 0x13, - 0xe9, 0xa6, 0x9a, 0x18, 0x21, 0xf4, 0xfa, 0xfb, 0xf8, 0x39, 0x2b, 0xa9, 0xc5, 0xaa, 0x4f, 0x4d, - 0x9a, 0xc5, 0x27, 0x69, 0xe1, 0xdb, 0x33, 0x19, 0xb3, 0x33, 0x1b, 0xd4, 0xbf, 0x0b, 0x6d, 0x57, - 0xaf, 0x17, 0x2d, 0x0d, 0xaf, 0xb8, 0xc5, 0x8e, 0xe7, 0x32, 0x09, 0x4d, 0x4b, 0xc2, 0xee, 0x17, - 0x15, 0x68, 0x1d, 0x66, 0xa9, 0xe4, 0xa9, 0x3c, 0xbf, 0x1e, 0x73, 0xda, 0x80, 0xea, 0x07, 0x5c, - 0x64, 0xdd, 0xa7, 0xe8, 0x16, 0xd0, 0xf2, 0x73, 0xa8, 0xed, 0xdd, 0xaf, 0x88, 0x63, 0x18, 0x1a, - 0x54, 0x68, 0xf8, 0x9a, 0xd0, 0x35, 0x68, 0xe8, 0x55, 0x46, 0xf1, 0x1b, 0x14, 0xf5, 0x9e, 0xa2, - 0xf8, 0x2d, 0x51, 0x4b, 0xd6, 0xc4, 0xed, 0x43, 0xf9, 0x3b, 0x94, 0x71, 0xbe, 0x50, 0xfe, 0x9e, - 0x28, 0x5a, 0x5a, 0x96, 0x3f, 0xd4, 0xfc, 0x80, 0x9a, 0x77, 0xb9, 0x14, 0x49, 0x64, 0x32, 0xfc, - 0x48, 0xd4, 0x38, 0x3f, 0x7b, 0x9e, 0x5d, 0x5e, 0x8e, 0xf8, 0x59, 0x32, 0xe6, 0xe7, 0x22, 0x54, - 0xd9, 0xdd, 0xe2, 0x7e, 0x22, 0x74, 0x03, 0x3a, 0xa8, 0x3f, 0x9e, 0x5a, 0x44, 0xdd, 0x9f, 0x31, - 0x6c, 0xc5, 0x5b, 0x87, 0x1e, 0xbf, 0xa0, 0xc7, 0x8a, 0xa7, 0x04, 0x3d, 0x7e, 0x45, 0xd4, 0xe5, - 0x19, 0x41, 0xc3, 0x6f, 0x44, 0x91, 0xbd, 0xb1, 0xc0, 0x11, 0x5a, 0x7e, 0x27, 0xaa, 0xe3, 0xeb, - 0x43, 0x19, 0x0a, 0xf9, 0x3e, 0x2e, 0x10, 0x6a, 0xff, 0x20, 0x6a, 0x04, 0x9e, 0xb6, 0xed, 0x3b, - 0x09, 0x93, 0x11, 0x8f, 0x51, 0xff, 0x27, 0xa1, 0x77, 0xe0, 0xf6, 0xea, 0xc7, 0x03, 0x9d, 0xfe, - 0xc2, 0x3a, 0x57, 0xbc, 0x05, 0xe8, 0xf1, 0xf1, 0x7b, 0xbb, 0xfb, 0x70, 0x6b, 0xd9, 0x3d, 0x4c, - 0x5b, 0x40, 0x8e, 0xf8, 0x43, 0x4d, 0x80, 0xe2, 0xb6, 0x0d, 0x0d, 0x1d, 0x96, 0x08, 0x1e, 0x77, - 0x3d, 0x0a, 0x50, 0x37, 0xa5, 0x74, 0xfd, 0xdd, 0x37, 0xa1, 0xed, 0xde, 0x91, 0x8a, 0x03, 0x35, - 0x59, 0x58, 0xff, 0x89, 0x8d, 0x5d, 0x87, 0xd6, 0xb1, 0xa3, 0xf0, 0x14, 0xcf, 0x4d, 0xcd, 0xb3, - 0x11, 0xfd, 0x8b, 0x3a, 0xfe, 0xbe, 0xbd, 0xf2, 0x77, 0x00, 0x00, 0x00, 0xff, 0xff, 0xab, 0x89, - 0x60, 0xe3, 0xcc, 0x09, 0x00, 0x00, + 0x14, 0xc6, 0xce, 0xc5, 0xc9, 0x49, 0x9a, 0x86, 0xe9, 0xd2, 0x46, 0x01, 0x56, 0x95, 0x57, 0x48, + 0x55, 0x25, 0x02, 0x2a, 0x42, 0x5a, 0x16, 0x55, 0x50, 0xf5, 0x22, 0x8a, 0x00, 0x55, 0x6e, 0x77, + 0x1f, 0x78, 0x73, 0xed, 0xb3, 0x95, 0xd9, 0xd4, 0x0e, 0xe3, 0x49, 0x56, 0xfd, 0x3b, 0xfc, 0x04, + 0x9e, 0x78, 0xe0, 0x9d, 0x17, 0x7e, 0x02, 0x3f, 0x80, 0x3b, 0x88, 0x9b, 0xe0, 0x8d, 0x99, 0x33, + 0x13, 0x67, 0x1c, 0x62, 0xb4, 0x08, 0xed, 0xdb, 0x9c, 0xdb, 0x9c, 0xf3, 0x9d, 0xef, 0x9c, 0xb1, + 0x01, 0x22, 0xc1, 0xc7, 0xa3, 0x09, 0xcf, 0x44, 0xc6, 0x3c, 0x7d, 0xbe, 0xf4, 0xbf, 0x74, 0xa0, + 0x77, 0x8e, 0x79, 0x9e, 0x64, 0x69, 0x80, 0x1f, 0x4f, 0x31, 0x17, 0xec, 0x05, 0x68, 0x27, 0xe9, + 0x15, 0x97, 0xca, 0xd3, 0x78, 0xe0, 0x6c, 0x3b, 0x3b, 0xed, 0x60, 0xa1, 0x50, 0xd6, 0x1c, 0xf9, + 0x2c, 0x89, 0x50, 0x5a, 0x5d, 0x6d, 0x2d, 0x14, 0xec, 0x00, 0x5a, 0x13, 0x44, 0x7e, 0x14, 0x8a, + 0x70, 0x50, 0xdb, 0xae, 0xed, 0x74, 0xf6, 0x5e, 0x1a, 0x99, 0x54, 0xa3, 0x72, 0x9a, 0xd1, 0x99, + 0xf1, 0x3b, 0x4e, 0x05, 0xbf, 0x09, 0x8a, 0xb0, 0xe1, 0x9b, 0xb0, 0x56, 0x32, 0xb1, 0x3e, 0xd4, + 0x1e, 0xe1, 0x0d, 0x55, 0xb2, 0x16, 0xa8, 0x23, 0xbb, 0x05, 0x8d, 0x59, 0x38, 0x9e, 0x22, 0xe5, + 0xef, 0x06, 0x5a, 0xb8, 0xe7, 0xde, 0x75, 0xfc, 0x2f, 0x5c, 0xd8, 0x3a, 0xe4, 0x18, 0x0a, 0xbc, + 0x40, 0x7e, 0x9d, 0xa4, 0xa1, 0xc8, 0xb8, 0x85, 0xeb, 0x5f, 0x2a, 0x1f, 0x80, 0x77, 0x99, 0xa4, + 0xb1, 0x04, 0x2a, 0x0b, 0x57, 0xb6, 0xb9, 0xa8, 0x2c, 0x61, 0x1c, 0x2b, 0xf8, 0x83, 0xba, 0xb6, + 0x18, 0x91, 0xbd, 0x6b, 0xa1, 0x6d, 0x10, 0xda, 0x51, 0x81, 0xb6, 0xa2, 0x8a, 0x2a, 0xd8, 0x8c, + 0x41, 0x3d, 0xca, 0x72, 0x31, 0x68, 0x12, 0x4c, 0x3a, 0xb3, 0x7d, 0x80, 0x09, 0xc7, 0x08, 0x63, + 0x4c, 0x23, 0x1c, 0x78, 0xd2, 0xd2, 0xdb, 0x7b, 0xb1, 0xc8, 0xb0, 0xb8, 0xfb, 0xac, 0x70, 0x0a, + 0xac, 0x80, 0xff, 0xd7, 0xc9, 0x7d, 0xd8, 0x0a, 0xf0, 0x3a, 0x9b, 0xad, 0x68, 0xa4, 0x0f, 0x5d, + 0x51, 0x28, 0x8b, 0x19, 0x29, 0xe9, 0xfc, 0x33, 0x80, 0x45, 0x20, 0xeb, 0x81, 0x9b, 0xcc, 0xfd, + 0xe4, 0xc9, 0x6e, 0xb6, 0x5b, 0xd9, 0xec, 0x5a, 0xa9, 0xd9, 0xfe, 0x39, 0x0c, 0x1f, 0x84, 0xe3, + 0x24, 0x2e, 0x75, 0x35, 0x9f, 0xd7, 0xf4, 0x3a, 0x74, 0x16, 0xf9, 0x73, 0x99, 0x4a, 0xb1, 0xb1, + 0xb1, 0xa2, 0x57, 0x81, 0xed, 0xe7, 0x7f, 0xe5, 0xc0, 0xd6, 0xfd, 0x49, 0xbc, 0x72, 0x5e, 0x9e, + 0x00, 0x26, 0xdb, 0x85, 0xfe, 0x94, 0xc2, 0x17, 0x14, 0x10, 0xa2, 0x56, 0xf0, 0x0f, 0x3d, 0xbb, + 0x0d, 0xa0, 0x75, 0x87, 0x8a, 0xe7, 0x1a, 0x79, 0x59, 0x9a, 0x25, 0xb6, 0xeb, 0xff, 0x91, 0xed, + 0x62, 0x80, 0x1a, 0x8b, 0x01, 0xf2, 0x5f, 0x85, 0xfa, 0x51, 0x12, 0x8e, 0x57, 0xf5, 0x7f, 0xde, + 0x65, 0xb7, 0xdc, 0xe5, 0x4d, 0xa8, 0xbf, 0x97, 0xa4, 0x8f, 0x96, 0x23, 0xfc, 0x77, 0xa0, 0x71, + 0x12, 0x4e, 0xc7, 0x82, 0xbd, 0x02, 0x5e, 0x3e, 0xbd, 0xfc, 0x08, 0x23, 0x41, 0xd6, 0xde, 0xde, + 0x73, 0x45, 0x89, 0xe4, 0x70, 0xae, 0x8d, 0xc1, 0xdc, 0xcb, 0xdc, 0xe4, 0x16, 0x37, 0x7d, 0x5a, + 0x83, 0x46, 0x90, 0x4d, 0x05, 0xea, 0x85, 0xa4, 0x37, 0x61, 0xf1, 0xd0, 0x14, 0x0a, 0xf6, 0x32, + 0x34, 0xf1, 0xaa, 0x28, 0xb1, 0x63, 0xe5, 0xa1, 0xe8, 0xd1, 0x31, 0x19, 0x03, 0xe3, 0xc4, 0xf6, + 0xa0, 0xf5, 0x30, 0xe3, 0x8f, 0x43, 0x1e, 0xe7, 0xe6, 0xe5, 0xd9, 0x5c, 0x0a, 0x38, 0xd1, 0xe6, + 0xa0, 0xf0, 0x1b, 0x4a, 0xf6, 0x9b, 0xfa, 0x1a, 0x7b, 0x22, 0x9d, 0xca, 0x89, 0x2c, 0xf7, 0x8a, + 0x6d, 0x43, 0x27, 0x96, 0x83, 0xa2, 0x58, 0x91, 0x25, 0x9b, 0x79, 0xb5, 0x55, 0xec, 0x2d, 0xeb, + 0x81, 0xa8, 0x53, 0x51, 0x77, 0x56, 0xa2, 0x78, 0x2a, 0x8f, 0xe1, 0xf0, 0x14, 0x3c, 0x83, 0x59, + 0xcd, 0x5e, 0xce, 0xa3, 0x03, 0x83, 0x43, 0x23, 0xb4, 0x34, 0xca, 0x1e, 0xe7, 0xe2, 0xa0, 0x84, + 0xd3, 0xd2, 0xf8, 0x6f, 0x80, 0x77, 0x3f, 0xe5, 0x4f, 0xc0, 0x9a, 0xac, 0x2f, 0xcd, 0x1e, 0x9b, + 0x1d, 0x50, 0x47, 0xff, 0x1e, 0xf4, 0x4e, 0xd3, 0x7c, 0xa2, 0x66, 0xc2, 0x2c, 0xd6, 0x0e, 0xac, + 0x73, 0x7d, 0xc4, 0xf8, 0x81, 0xaa, 0x56, 0xef, 0x6b, 0x3b, 0x58, 0x56, 0xfb, 0x9f, 0x3b, 0xb0, + 0x5e, 0x04, 0xe7, 0x93, 0x2c, 0xcd, 0x51, 0xf1, 0x91, 0x4f, 0xa3, 0x68, 0x8e, 0xa3, 0x15, 0xcc, + 0x45, 0xb6, 0x29, 0x27, 0x86, 0x73, 0xb5, 0xfe, 0x2e, 0x5d, 0x67, 0x24, 0xb9, 0x58, 0xcd, 0x99, + 0x4e, 0xb3, 0xfc, 0x49, 0x5a, 0xba, 0x7b, 0x2e, 0x53, 0xf6, 0xc0, 0x04, 0x0d, 0xef, 0x42, 0xd7, + 0xd6, 0xab, 0x45, 0x4b, 0xc3, 0x6b, 0x34, 0xd8, 0xe9, 0x5c, 0x26, 0xa1, 0x6d, 0x48, 0xd8, 0xfd, + 0xac, 0x06, 0x9d, 0xc3, 0x2c, 0x15, 0x98, 0x8a, 0x8b, 0x9b, 0x09, 0xb2, 0x16, 0xd4, 0x3f, 0x44, + 0x9e, 0xf5, 0x9f, 0x61, 0x5b, 0xc0, 0xca, 0x9f, 0x43, 0x65, 0xef, 0x7f, 0xed, 0x59, 0x86, 0x73, + 0x8d, 0x8a, 0x0c, 0xdf, 0x78, 0x6c, 0x0d, 0x5a, 0x6a, 0x95, 0x49, 0xfc, 0x96, 0x44, 0xb5, 0xa7, + 0x24, 0x7e, 0xe7, 0xc9, 0x25, 0x6b, 0xd3, 0xf6, 0x91, 0xfc, 0x3d, 0xc9, 0x34, 0x5f, 0x24, 0xff, + 0xe0, 0x49, 0x5a, 0x3a, 0x86, 0x3f, 0xd2, 0xfc, 0x48, 0x9a, 0xf7, 0x51, 0xf0, 0x24, 0xd2, 0x19, + 0x7e, 0xf2, 0xe4, 0x38, 0x3f, 0x7f, 0x91, 0x5d, 0x5d, 0x8d, 0xf1, 0x2c, 0x99, 0xe0, 0x05, 0x0f, + 0x65, 0x76, 0xbb, 0xb8, 0x9f, 0x3d, 0xb6, 0x01, 0x3d, 0xd2, 0x1f, 0xcf, 0x0c, 0xa2, 0xfe, 0x2f, + 0x14, 0x56, 0xf1, 0xad, 0x23, 0x8f, 0x5f, 0xc9, 0xa3, 0xe2, 0x53, 0x42, 0x1e, 0xbf, 0x11, 0xea, + 0xf2, 0x8c, 0x90, 0xe1, 0x77, 0x4f, 0x92, 0xbd, 0xb1, 0xc4, 0x11, 0x59, 0xfe, 0xf0, 0x24, 0xd9, + 0xcf, 0x9a, 0x46, 0x9d, 0x84, 0xc9, 0x18, 0x63, 0xd2, 0xff, 0xe9, 0xb1, 0x3b, 0x70, 0xbb, 0xfa, + 0x33, 0x41, 0x4e, 0x7f, 0x51, 0x45, 0x15, 0xaf, 0x3e, 0x79, 0x7c, 0xf2, 0xc1, 0xee, 0x3e, 0xdc, + 0x5a, 0xf5, 0xe2, 0xb2, 0x0e, 0x78, 0x47, 0xf8, 0x50, 0xb5, 0x5a, 0xb2, 0xd8, 0x85, 0x96, 0x0a, + 0x4b, 0x38, 0xc6, 0x7d, 0x87, 0x01, 0x34, 0x75, 0x29, 0x7d, 0x77, 0xf7, 0x6d, 0xe8, 0xda, 0xaf, + 0xa1, 0xec, 0xb6, 0x9c, 0x21, 0x5a, 0xf5, 0x13, 0x13, 0xbb, 0x0e, 0x9d, 0x63, 0x4b, 0xe1, 0x48, + 0x46, 0xdb, 0x8a, 0x51, 0x2d, 0xba, 0x97, 0x4d, 0xfa, 0x51, 0x7b, 0xed, 0xef, 0x00, 0x00, 0x00, + 0xff, 0xff, 0x1b, 0xc9, 0xcd, 0x42, 0xb6, 0x09, 0x00, 0x00, } diff --git a/pb/ctrl_pb/ctrl.proto b/pb/ctrl_pb/ctrl.proto index f92ef2129..fdb12ae01 100644 --- a/pb/ctrl_pb/ctrl.proto +++ b/pb/ctrl_pb/ctrl.proto @@ -19,7 +19,6 @@ enum ContentType { RemoveTerminatorRequestType = 1012; InspectRequestType = 1013; InspectResponseType = 1014; - StartXgressType = 1015; SessionFailedType = 1016; ValidateTerminatorsRequestType = 1017; UpdateTerminatorRequestType = 10018; diff --git a/pb/ctrl_pb/decoder.go b/pb/ctrl_pb/decoder.go index 91bf01c2e..bc77d0b57 100644 --- a/pb/ctrl_pb/decoder.go +++ b/pb/ctrl_pb/decoder.go @@ -238,15 +238,6 @@ func (d Decoder) Decode(msg *channel2.Message) ([]byte, bool) { pfxlog.Logger().Errorf("unexpected error (%s)", err) } - case int32(ContentType_StartXgressType): - meta := channel2.NewTraceMessageDecode(DECODER, "Start Xgress") - meta["sessionId"] = string(msg.Body) - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - return nil, true - } - return data, true - case int32(ContentType_MetricsType): metricsMsg := &metrics_pb.MetricsMessage{} if err := proto.Unmarshal(msg.Body, metricsMsg); err == nil { diff --git a/router/forwarder/forwarder.go b/router/forwarder/forwarder.go index e2ba32e2e..09f116bf0 100644 --- a/router/forwarder/forwarder.go +++ b/router/forwarder/forwarder.go @@ -97,19 +97,6 @@ func (forwarder *Forwarder) UnregisterDestinations(sessionId *identity.TokenId) } } -func (forwarder *Forwarder) StartDestinations(sessionId *identity.TokenId) { - if addresses, found := forwarder.destinations.getAddressesForSession(sessionId); found { - for _, address := range addresses { - if destination, found := forwarder.destinations.getDestination(address); found { - if xgressDest, ok := destination.(XgressDestination); ok && xgressDest.IsTerminator() { - xgressDest.Start() - pfxlog.Logger().Debugf("started Xgress for session %v", sessionId.Token) - } - } - } - } -} - func (forwarder *Forwarder) HasDestination(address xgress.Address) bool { _, found := forwarder.destinations.getDestination(address) return found diff --git a/router/handler_ctrl/bind.go b/router/handler_ctrl/bind.go index a93fc09c6..051fd9bae 100644 --- a/router/handler_ctrl/bind.go +++ b/router/handler_ctrl/bind.go @@ -56,7 +56,6 @@ func (self *bindHandler) BindChannel(ch channel2.Channel) error { ch.AddReceiveHandler(newRouteHandler(self.id, self.ctrl, self.dialerCfg, self.forwarder)) ch.AddReceiveHandler(newValidateTerminatorsHandler(self.ctrl, self.dialerCfg)) ch.AddReceiveHandler(newUnrouteHandler(self.forwarder)) - ch.AddReceiveHandler(newStartXgressHandler(self.forwarder)) ch.AddReceiveHandler(newTraceHandler(self.id, self.forwarder.TraceController())) ch.AddReceiveHandler(newInspectHandler(self.id)) ch.AddPeekHandler(trace.NewChannelPeekHandler(self.id, ch, self.forwarder.TraceController(), trace.NewChannelSink(ch))) diff --git a/router/handler_ctrl/start_xgress.go b/router/handler_ctrl/start_xgress.go deleted file mode 100644 index 6ec0c2775..000000000 --- a/router/handler_ctrl/start_xgress.go +++ /dev/null @@ -1,44 +0,0 @@ -/* - Copyright NetFoundry, Inc. - - Licensed under the Apache License, Version 2.0 (the "License"); - you may not use this file except in compliance with the License. - You may obtain a copy of the License at - - https://www.apache.org/licenses/LICENSE-2.0 - - Unless required by applicable law or agreed to in writing, software - distributed under the License is distributed on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - See the License for the specific language governing permissions and - limitations under the License. -*/ - -package handler_ctrl - -import ( - "github.com/michaelquigley/pfxlog" - "github.com/openziti/fabric/pb/ctrl_pb" - "github.com/openziti/fabric/router/forwarder" - "github.com/openziti/foundation/channel2" - "github.com/openziti/foundation/identity/identity" -) - -type startXgressHandler struct { - forwarder *forwarder.Forwarder -} - -func newStartXgressHandler(forwarder *forwarder.Forwarder) *startXgressHandler { - return &startXgressHandler{forwarder: forwarder} -} - -func (h *startXgressHandler) ContentType() int32 { - return int32(ctrl_pb.ContentType_StartXgressType) -} - -func (h *startXgressHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { - token := string(msg.Body) - pfxlog.Logger().Debugf("attempting to start xgress for session %v", token) - sessionId := &identity.TokenId{Token: token} - h.forwarder.StartDestinations(sessionId) -} diff --git a/router/xgress/messages.go b/router/xgress/messages.go index 8cdf6fe9b..316514e58 100644 --- a/router/xgress/messages.go +++ b/router/xgress/messages.go @@ -59,8 +59,9 @@ func (o Originator) String() string { type PayloadFlag uint32 const ( - PayloadFlagSessionEnd PayloadFlag = 1 - PayloadFlagEgress PayloadFlag = 2 + PayloadFlagSessionEnd PayloadFlag = 1 + PayloadFlagEgress PayloadFlag = 2 + PayloadFlagSessionStart PayloadFlag = 4 ) type Header struct { @@ -233,6 +234,10 @@ func (payload *Payload) IsSessionEndFlagSet() bool { return isPayloadFlagSet(payload.Flags, PayloadFlagSessionEnd) } +func (payload *Payload) IsSessionStartFlagSet() bool { + return isPayloadFlagSet(payload.Flags, PayloadFlagSessionStart) +} + func SetOriginatorFlag(flags uint32, originator Originator) uint32 { if originator == Initiator { return ^uint32(PayloadFlagEgress) & flags diff --git a/router/xgress/request.go b/router/xgress/request.go index 7f7ac3d86..9f24e4824 100644 --- a/router/xgress/request.go +++ b/router/xgress/request.go @@ -134,12 +134,6 @@ type SessionInfo struct { ctrl CtrlChannel } -func (si *SessionInfo) SendStartEgress() error { - msg := channel2.NewMessage(int32(ctrl_pb.ContentType_StartXgressType), nil) - msg.ReplyTo(si.ResponseMsg) - return si.ctrl.Channel().SendWithTimeout(msg, time.Second*5) -} - var authError = errors.New("unexpected failure while authenticating") func GetSession(ctrl CtrlChannel, ingressId string, serviceId string, peerData map[uint32][]byte) (*SessionInfo, error) { @@ -202,9 +196,6 @@ func CreateSession(ctrl CtrlChannel, peer Connection, request *Request, bindHand bindHandler.HandleXgressBind(sessionInfo.SessionId, sessionInfo.Address, Initiator, x) x.Start() - if err = sessionInfo.SendStartEgress(); err != nil { - return &Response{Success: false, Message: err.Error()} - } return &Response{Success: true, SessionId: sessionInfo.SessionId.Token} } diff --git a/router/xgress/xgress.go b/router/xgress/xgress.go index 596d2ed55..55c58bb98 100644 --- a/router/xgress/xgress.go +++ b/router/xgress/xgress.go @@ -154,14 +154,33 @@ func (self *Xgress) AddPeekHandler(peekHandler PeekHandler) { } func (self *Xgress) Start() { + log := pfxlog.ContextLogger(self.Label()) + if !self.IsTerminator() { + log.Debug("initiator: sending session start") + self.forwardPayload(self.GetStartSession()) + go self.rx() + } else { + log.Debug("terminator: waiting for session start before starting receiver") + } go self.tx() - go self.rx() } func (self *Xgress) Label() string { return fmt.Sprintf("{s/%s|@/%s}<%s>", self.sessionId.Token, string(self.address), self.originator.String()) } +func (self *Xgress) GetStartSession() *Payload { + startSession := &Payload{ + Header: Header{ + SessionId: self.sessionId.Token, + Flags: SetOriginatorFlag(uint32(PayloadFlagSessionStart), self.originator), + }, + Sequence: self.nextReceiveSequence(), + Data: nil, + } + return startSession +} + func (self *Xgress) GetEndSession() *Payload { endSession := &Payload{ Header: Header{ @@ -249,25 +268,32 @@ func (self *Xgress) tx() { for { select { case inPayload := <-self.txQueue: - if inPayload != nil && !inPayload.IsSessionEndFlagSet() { - payloadLogger := log.WithFields(inPayload.GetLoggerFields()) - if !self.Options.RandomDrops || rand.Int31n(self.Options.Drop1InN) != 1 { - payloadLogger.Debug("adding to transmit buffer") - self.txBuffer.ReceiveUnordered(inPayload) - if self.Options.Retransmission && self.payloadBuffer != nil { - payloadLogger.Debug("acknowledging") - self.payloadBuffer.AcknowledgePayload(inPayload) - } - } else { - payloadLogger.Error("drop!") + if inPayload == nil || inPayload.IsSessionEndFlagSet() { + return + } + if inPayload.IsSessionStartFlagSet() { + log.Debug("received session start, starting xgress receiver") + go self.rx() + } + payloadLogger := log.WithFields(inPayload.GetLoggerFields()) + if !self.Options.RandomDrops || rand.Int31n(self.Options.Drop1InN) != 1 { + payloadLogger.Debug("adding to transmit buffer") + self.txBuffer.ReceiveUnordered(inPayload) + if self.Options.Retransmission && self.payloadBuffer != nil { + payloadLogger.Debug("acknowledging") + self.payloadBuffer.AcknowledgePayload(inPayload) } + } else { + payloadLogger.Error("drop!") + } - ready := self.txBuffer.ReadyForTransmit() - for _, outPayload := range ready { - outPayloadLogger := log.WithFields(outPayload.GetLoggerFields()) - for _, peekHandler := range self.peekHandlers { - peekHandler.Tx(self, outPayload) - } + ready := self.txBuffer.ReadyForTransmit() + for _, outPayload := range ready { + outPayloadLogger := log.WithFields(outPayload.GetLoggerFields()) + for _, peekHandler := range self.peekHandlers { + peekHandler.Tx(self, outPayload) + } + if !outPayload.IsSessionStartFlagSet() { n, err := self.peer.WritePayload(outPayload.Data, outPayload.Headers) if err != nil { outPayloadLogger.Warnf("write failed (%s)", err) @@ -275,13 +301,10 @@ func (self *Xgress) tx() { outPayloadLogger.Debugf("sent (#%d) [%s]", outPayload.GetSequence(), info.ByteCount(int64(n))) } } + } - if len(ready) < 1 { - payloadLogger.Debug("queued transmit") - } - - } else { - return + if len(ready) < 1 { + payloadLogger.Debug("queued transmit") } } } @@ -339,19 +362,8 @@ func (self *Xgress) rx() { remaining -= length payloads++ + self.forwardPayload(payload) payloadLogger := log.WithFields(payload.GetLoggerFields()) - - if self.Options.Retransmission && self.payloadBuffer != nil { - //payloadLogger.Debug("buffering payload") - self.payloadBuffer.BufferPayload(payload) - } - - for _, peekHandler := range self.peekHandlers { - peekHandler.Rx(self, payload) - } - - self.receiveHandler.HandleXgressReceive(payload, self) - payloadLogger.Debugf("received [%s]", info.ByteCount(int64(n))) } @@ -363,6 +375,18 @@ func (self *Xgress) rx() { } } +func (self *Xgress) forwardPayload(payload *Payload) { + if self.Options.Retransmission && self.payloadBuffer != nil { + self.payloadBuffer.BufferPayload(payload) + } + + for _, peekHandler := range self.peekHandlers { + peekHandler.Rx(self, payload) + } + + self.receiveHandler.HandleXgressReceive(payload, self) +} + func (self *Xgress) nextReceiveSequence() int32 { self.rxSequenceLock.Lock() defer self.rxSequenceLock.Unlock() diff --git a/router/xgress_transport/dialer.go b/router/xgress_transport/dialer.go index e7e222c90..04eb93d67 100644 --- a/router/xgress_transport/dialer.go +++ b/router/xgress_transport/dialer.go @@ -59,6 +59,7 @@ func (txd *dialer) Dial(destination string, sessionId *identity.TokenId, address conn := &transportXgresscConn{peer} x := xgress.NewXgress(sessionId, address, conn, xgress.Terminator, txd.options) bindHandler.HandleXgressBind(sessionId, address, xgress.Terminator, x) + x.Start() return nil, nil } diff --git a/router/xgress_transport_udp/dialer.go b/router/xgress_transport_udp/dialer.go index 24c298f45..8d724ab5a 100644 --- a/router/xgress_transport_udp/dialer.go +++ b/router/xgress_transport_udp/dialer.go @@ -43,6 +43,7 @@ func (txd *dialer) Dial(destination string, sessionId *identity.TokenId, address x := xgress.NewXgress(sessionId, address, newPacketConn(conn), xgress.Terminator, txd.options) bindHandler.HandleXgressBind(sessionId, address, xgress.Terminator, x) + x.Start() return nil, nil } diff --git a/version b/version index 51176c7c8..c43e1055f 100644 --- a/version +++ b/version @@ -1 +1 @@ -0.11 +0.12