gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-09-11 13:29:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support for peerData in parallel routing. (#168)

Browse Source
This commit is contained in:
Michael Quigley
2021-02-08 11:21:42 -05:00
parent d36f10b185
commit cd3e31df78
3 changed files with 32 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/handler_ctrl/route_result.go
+9 -2
View File
@@ -19,6 +19,7 @@ package handler_ctrl
import (
"github.com/golang/protobuf/proto"
"github.com/openziti/fabric/controller/network"
"github.com/openziti/fabric/controller/xt"
"github.com/openziti/fabric/ctrl_msg"
"github.com/openziti/fabric/pb/ctrl_pb"
"github.com/openziti/foundation/channel2"
@@ -41,10 +42,16 @@ func (self *routeResultHandler) ContentType() int32 {
return ctrl_msg.RouteResultType
}
func (self *routeResultHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (self *routeResultHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) {
_, success := msg.Headers[ctrl_msg.RouteResultSuccessHeader]
sessionId := string(msg.Body)
routing := self.network.RouteResult(self.r, sessionId, success)
peerData := xt.PeerData{}
for k, v := range msg.Headers {
if k > 0 && k != ctrl_msg.RouteResultSuccessHeader && k != ctrl_msg.RouteResultErrorHeader {
peerData[uint32(k)] = v
}
}
routing := self.network.RouteResult(self.r, sessionId, success, peerData)
if !routing {
go self.notRoutingSession(sessionId)
}
controller/network/network.go
+5 -2
View File
@@ -178,8 +178,8 @@ func (network *Network) GetAllSessions() []*session {
return network.sessionController.all()
}
func (network *Network) RouteResult(r *Router, sessionId string, success bool) bool {
return network.routeSenderController.forwardRouteResult(r, sessionId, success)
func (network *Network) RouteResult(r *Router, sessionId string, success bool, peerData xt.PeerData) bool {
return network.routeSenderController.forwardRouteResult(r, sessionId, success, peerData)
}
func (network *Network) GetEventDispatcher() event.Dispatcher {
@@ -334,6 +334,9 @@ func (network *Network) CreateSession(srcR *Router, clientId *identity.TokenId,
}
rms[len(rms)-1].Egress.PeerData = clientId.Data
// 5: Routing
// 5: Route Egress
peerData, err := sendRoute(circuit.Path[len(circuit.Path)-1], rms[len(rms)-1], network.options.TerminationTimeout)
if err != nil {
controller/network/routesender.go
+18 -9
View File
@@ -18,6 +18,7 @@ package network
import (
"github.com/golang/protobuf/proto"
"github.com/openziti/fabric/controller/xt"
"github.com/openziti/fabric/pb/ctrl_pb"
"github.com/openziti/foundation/channel2"
cmap "github.com/orcaman/concurrent-map"
@@ -34,11 +35,11 @@ func newRouteSenderController() *routeSenderController {
return &routeSenderController{}
}
func (self *routeSenderController) forwardRouteResult(r *Router, sessionId string, success bool) bool {
func (self *routeSenderController) forwardRouteResult(r *Router, sessionId string, success bool, peerData xt.PeerData) bool {
v, found := self.senders.Get(sessionId)
if found {
routeSender := v.(*routeSender)
routeSender.in <- &routeStatus{r: r, sessionId: sessionId, success: success}
routeSender.in <- &routeStatus{r: r, sessionId: sessionId, success: success, peerData: peerData}
return true
}
return false
@@ -64,23 +65,25 @@ func newRouteSender(sessionId string, timeout time.Duration, maxTries int) *rout
}
}
func (self *routeSender) route(circuit *Circuit, routeMsgs []*ctrl_pb.Route) error {
func (self *routeSender) route(circuit *Circuit, routeMsgs []*ctrl_pb.Route, strategy xt.Strategy, terminator xt.Terminator) (xt.PeerData, error) {
defer func() {
// remove sender
}()
// send route messages
tr := circuit.Path[len(circuit.Path)-1]
for i := 0; i < len(circuit.Path); i++ {
r := circuit.Path[i]
go self.sendRoute(r, routeMsgs[i])
self.attendance[r.Id] = false
// count termination attempts
if r == circuit.Path[len(circuit.Path)-1] {
if r == tr {
self.tries++
}
}
var peerData xt.PeerData
deadline := time.Now().Add(self.timeout)
timeout := time.Until(deadline)
attendance:
@@ -89,29 +92,33 @@ attendance:
case status := <-self.in:
if status.success {
self.attendance[status.r.Id] = true
if status.r == tr {
peerData = status.peerData
}
timeout = time.Until(deadline)
} else {
if status.r == circuit.Path[len(circuit.Path)-1] {
if status.r == tr {
if self.tries < self.maxTries {
strategy.NotifyEvent(xt.NewDialFailedEvent(terminator))
self.tries++
logrus.Warnf("retrying terminator, attempt [%d] of [%d]", self.tries, self.maxTries)
go self.sendRoute(status.r, routeMsgs[len(circuit.Path)-1])
} else {
self.tearDownTheSuccesful()
return errors.Errorf("error creating route [s/%s] on [r/%s], maximum retry attempts exceeded", self.sessionId, status.r.Id)
return nil, errors.Errorf("error creating route [s/%s] on [r/%s], maximum retry attempts exceeded", self.sessionId, status.r.Id)
}
} else {
self.tearDownTheSuccesful()
return errors.Errorf("error creating route for [s/%s] on [r/%s]", self.sessionId, status.r.Id)
return nil, errors.Errorf("error creating route for [s/%s] on [r/%s]", self.sessionId, status.r.Id)
}
}
case <-time.After(timeout):
self.tearDownTheSuccesful()
return errors.Errorf("timeout creating routes for [s/%s]", self.sessionId)
return nil, errors.Errorf("timeout creating routes for [s/%s]", self.sessionId)
}
allPresent := true
for _, v := range self.attendance {
@@ -123,7 +130,8 @@ attendance:
break attendance
}
}
return nil
strategy.NotifyEvent(xt.NewDialSucceeded(terminator))
return peerData, nil
}
func (self *routeSender) sendRoute(r *Router, routeMsg *ctrl_pb.Route) {
@@ -144,4 +152,5 @@ type routeStatus struct {
r *Router
sessionId string
success bool
peerData xt.PeerData
}