mirror of
https://github.com/openziti/ziti.git
synced 2026-10-13 15:51:21 +00:00
Merge pull request openziti/fabric#118 from openziti/fix-xgress-setup-race-condition
Fix xgress start race condition. Fixes GH-117
This commit is contained in:
Paul Lorenz
@@ -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)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -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))
|
||||
|
||||
+60
-64
@@ -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,
|
||||
}
|
||||
|
||||
@@ -19,7 +19,6 @@ enum ContentType {
|
||||
RemoveTerminatorRequestType = 1012;
|
||||
InspectRequestType = 1013;
|
||||
InspectResponseType = 1014;
|
||||
StartXgressType = 1015;
|
||||
SessionFailedType = 1016;
|
||||
ValidateTerminatorsRequestType = 1017;
|
||||
UpdateTerminatorRequestType = 10018;
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)))
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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}
|
||||
}
|
||||
|
||||
|
||||
+60
-36
@@ -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()
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user