Rename circuit to path

This commit is contained in:
Paul Lorenz
2021-07-26 13:25:42 -04:00
parent 29753846a4
commit bcd1c374e8
18 changed files with 902 additions and 947 deletions
+1 -1
View File
@@ -52,7 +52,7 @@ func (h *sessionRequestHandler) HandleReceive(msg *channel2.Message, ch channel2
go func() {
id := &identity.TokenId{Token: request.IngressId, Data: request.PeerData}
if session, err := h.network.CreateSession(h.r, id, request.ServiceId); err == nil {
responseMsg := ctrl_msg.NewSessionSuccessMsg(session.Id.Token, session.Circuit.IngressId)
responseMsg := ctrl_msg.NewSessionSuccessMsg(session.Id.Token, session.Path.IngressId)
responseMsg.ReplyTo(msg)
//static terminator peer data
+8 -8
View File
@@ -53,7 +53,7 @@ func (h *listSessionsHandler) HandleReceive(msg *channel2.Message, ch channel2.C
Id: s.Id.Token,
ClientId: s.ClientId.Token,
ServiceId: s.Service.Id,
Circuit: NewCircuit(s.Circuit),
Path: NewPath(s.Path),
}
response.Sessions = append(response.Sessions, responseSession)
}
@@ -70,13 +70,13 @@ func (h *listSessionsHandler) HandleReceive(msg *channel2.Message, ch channel2.C
}
}
func NewCircuit(circuit *network.Circuit) *mgmt_pb.Circuit {
mgmtCircuit := &mgmt_pb.Circuit{}
for _, r := range circuit.Path {
mgmtCircuit.Path = append(mgmtCircuit.Path, r.Id)
func NewPath(path *network.Path) *mgmt_pb.Path {
mgmtPath := &mgmt_pb.Path{}
for _, r := range path.Nodes {
mgmtPath.Nodes = append(mgmtPath.Nodes, r.Id)
}
for _, l := range circuit.Links {
mgmtCircuit.Links = append(mgmtCircuit.Links, l.Id.Token)
for _, l := range path.Links {
mgmtPath.Links = append(mgmtPath.Links, l.Id.Token)
}
return mgmtCircuit
return mgmtPath
}
+5 -5
View File
@@ -62,13 +62,13 @@ type SessionsStreamHandler struct {
ch channel2.Channel
}
func (handler *SessionsStreamHandler) SessionCreated(sessionId *identity.TokenId, clientId *identity.TokenId, serviceId string, circuit *network.Circuit) {
func (handler *SessionsStreamHandler) SessionCreated(sessionId *identity.TokenId, clientId *identity.TokenId, serviceId string, path *network.Path) {
event := &mgmt_pb.StreamSessionsEvent{
EventType: mgmt_pb.StreamSessionEventType_SessionCreated,
SessionId: sessionId.Token,
ClientId: clientId.Token,
ServiceId: serviceId,
Circuit: NewCircuit(circuit),
Path: NewPath(path),
}
handler.sendEvent(event)
}
@@ -82,11 +82,11 @@ func (handler *SessionsStreamHandler) SessionDeleted(sessionId *identity.TokenId
handler.sendEvent(event)
}
func (handler *SessionsStreamHandler) CircuitUpdated(sessionId *identity.TokenId, circuit *network.Circuit) {
func (handler *SessionsStreamHandler) PathUpdated(sessionId *identity.TokenId, path *network.Path) {
event := &mgmt_pb.StreamSessionsEvent{
EventType: mgmt_pb.StreamSessionEventType_CircuitUpdated,
EventType: mgmt_pb.StreamSessionEventType_PathUpdated,
SessionId: sessionId.Token,
Circuit: NewCircuit(circuit),
Path: NewPath(path),
}
handler.sendEvent(event)
}
-155
View File
@@ -1,155 +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 network
import (
"fmt"
"github.com/openziti/fabric/pb/ctrl_pb"
"github.com/openziti/foundation/identity/identity"
)
type Circuit struct {
Path []*Router
Links []*Link
Binding string
IngressId string
EgressId string
}
func (circuit *Circuit) String() string {
if len(circuit.Path) < 1 {
return "{}"
}
if len(circuit.Links) != len(circuit.Path)-1 {
return "{malformed}"
}
out := fmt.Sprintf("[r/%s]", circuit.Path[0].Id)
for i := 0; i < len(circuit.Links); i++ {
out += fmt.Sprintf("->[l/%s]", circuit.Links[i].Id.Token)
out += fmt.Sprintf("->[r/%s]", circuit.Path[i+1].Id)
}
return out
}
func (circuit *Circuit) EqualPath(other *Circuit) bool {
if len(circuit.Path) != len(other.Path) {
return false
}
if len(circuit.Links) != len(other.Links) {
return false
}
for i := 0; i < len(circuit.Path); i++ {
if circuit.Path[i] != other.Path[i] {
return false
}
}
for i := 0; i < len(circuit.Links); i++ {
if circuit.Links[i] != other.Links[i] {
return false
}
}
return true
}
func (circuit *Circuit) EgressRouter() *Router {
if len(circuit.Path) > 0 {
return circuit.Path[len(circuit.Path)-1]
}
return nil
}
func (circuit *Circuit) CreateRouteMessages(attempt uint32, sessionId *identity.TokenId, egressAddress string) ([]*ctrl_pb.Route, error) {
var routeMessages []*ctrl_pb.Route
if len(circuit.Links) == 0 {
// single router path
routeMessage := &ctrl_pb.Route{SessionId: sessionId.Token, Attempt: attempt}
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: circuit.IngressId,
DstAddress: circuit.EgressId,
})
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: circuit.EgressId,
DstAddress: circuit.IngressId,
})
routeMessage.Egress = &ctrl_pb.Route_Egress{
Binding: circuit.Binding,
Address: circuit.EgressId,
Destination: egressAddress,
}
routeMessages = append(routeMessages, routeMessage)
}
for i, link := range circuit.Links {
if i == 0 {
// ingress
routeMessage := &ctrl_pb.Route{SessionId: sessionId.Token, Attempt: attempt}
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: circuit.IngressId,
DstAddress: link.Id.Token,
})
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: link.Id.Token,
DstAddress: circuit.IngressId,
})
routeMessages = append(routeMessages, routeMessage)
}
if i >= 0 && i < len(circuit.Links)-1 {
// transit
nextLink := circuit.Links[i+1]
routeMessage := &ctrl_pb.Route{SessionId: sessionId.Token, Attempt: attempt}
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: link.Id.Token,
DstAddress: nextLink.Id.Token,
})
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: nextLink.Id.Token,
DstAddress: link.Id.Token,
})
routeMessages = append(routeMessages, routeMessage)
}
if i == len(circuit.Links)-1 {
// egress
routeMessage := &ctrl_pb.Route{SessionId: sessionId.Token, Attempt: attempt}
routeMessage.Egress = &ctrl_pb.Route_Egress{
Binding: circuit.Binding,
Address: circuit.EgressId,
Destination: egressAddress,
}
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: circuit.EgressId,
DstAddress: link.Id.Token,
})
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: link.Id.Token,
DstAddress: circuit.EgressId,
})
routeMessages = append(routeMessages, routeMessage)
}
}
return routeMessages, nil
}
func (circuit *Circuit) usesLink(l *Link) bool {
if circuit.Links != nil {
for _, o := range circuit.Links {
if o == l {
return true
}
}
}
return false
}
-231
View File
@@ -1,231 +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 network
import (
"github.com/openziti/fabric/controller/db"
"github.com/openziti/fabric/controller/models"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/common"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/transport"
"github.com/openziti/foundation/transport/tcp"
"github.com/stretchr/testify/assert"
"runtime"
"testing"
"time"
)
func TestSimpleCircuit2(t *testing.T) {
ctx := db.NewTestContext(t)
defer ctx.Cleanup()
closeNotify := make(chan struct{})
defer close(closeNotify)
nodeId := &identity.TokenId{Token: "test"}
network, err := NewNetwork(nodeId, nil, ctx.GetDb(), nil, NewVersionProviderTest(), closeNotify)
assert.Nil(t, err)
addr := "tcp:0.0.0.0:0"
transportAddr, err := tcp.AddressParser{}.Parse(addr)
assert.Nil(t, err)
r0 := newRouterForTest("r0", "", transportAddr, nil)
network.Routers.markConnected(r0)
r1 := newRouterForTest("r1", "", transportAddr, nil)
network.Routers.markConnected(r1)
l0 := newLink(&identity.TokenId{Token: "l0"})
l0.Src = r0
l0.Dst = r1
l0.addState(newLinkState(Connected))
network.linkController.add(l0)
circuit, err := network.CreateCircuit(r0, r1)
assert.NotNil(t, circuit)
assert.Nil(t, err)
assert.Equal(t, 2, len(circuit.Path))
assert.Equal(t, r0, circuit.Path[0])
assert.Equal(t, r1, circuit.Path[1])
assert.Equal(t, 1, len(circuit.Links))
assert.Equal(t, l0, circuit.Links[0])
assert.Equal(t, r1, circuit.EgressRouter())
s0 := &identity.TokenId{Token: "s0"}
routeMessages, err := circuit.CreateRouteMessages(0, s0, addr)
assert.NotNil(t, routeMessages)
assert.Nil(t, err)
assert.Equal(t, 2, len(routeMessages))
// ingress route message
rm0 := routeMessages[0]
assert.Equal(t, s0.Token, rm0.SessionId)
assert.Nil(t, rm0.Egress)
assert.Equal(t, 2, len(rm0.Forwards))
assert.Equal(t, circuit.IngressId, rm0.Forwards[0].SrcAddress)
assert.Equal(t, l0.Id.Token, rm0.Forwards[0].DstAddress)
assert.Equal(t, l0.Id.Token, rm0.Forwards[1].SrcAddress)
assert.Equal(t, circuit.IngressId, rm0.Forwards[1].DstAddress)
// egress route message
rm1 := routeMessages[1]
assert.Equal(t, s0.Token, rm1.SessionId)
assert.NotNil(t, rm1.Egress)
assert.Equal(t, circuit.EgressId, rm1.Egress.Address)
assert.Equal(t, addr, rm1.Egress.Destination)
assert.Equal(t, circuit.EgressId, rm1.Forwards[0].SrcAddress)
assert.Equal(t, l0.Id.Token, rm1.Forwards[0].DstAddress)
assert.Equal(t, l0.Id.Token, rm1.Forwards[1].SrcAddress)
assert.Equal(t, circuit.EgressId, rm1.Forwards[1].DstAddress)
}
func TestTransitCircuit2(t *testing.T) {
ctx := db.NewTestContext(t)
defer ctx.Cleanup()
closeNotify := make(chan struct{})
defer close(closeNotify)
nodeId := &identity.TokenId{Token: "test"}
network, err := NewNetwork(nodeId, nil, ctx.GetDb(), nil, NewVersionProviderTest(), closeNotify)
assert.Nil(t, err)
addr := "tcp:0.0.0.0:0"
transportAddr, err := tcp.AddressParser{}.Parse(addr)
assert.Nil(t, err)
r0 := newRouterForTest("r0", "", transportAddr, nil)
network.Routers.markConnected(r0)
r1 := newRouterForTest("r1", "", transportAddr, nil)
network.Routers.markConnected(r1)
r2 := newRouterForTest("r2", "", transportAddr, nil)
network.Routers.markConnected(r2)
l0 := newLink(&identity.TokenId{Token: "l0"})
l0.Src = r0
l0.Dst = r1
l0.addState(newLinkState(Connected))
network.linkController.add(l0)
l1 := newLink(&identity.TokenId{Token: "l1"})
l1.Src = r1
l1.Dst = r2
l1.addState(newLinkState(Connected))
network.linkController.add(l1)
circuit, err := network.CreateCircuit(r0, r2)
assert.NotNil(t, circuit)
assert.Nil(t, err)
assert.Equal(t, 3, len(circuit.Path))
assert.Equal(t, r0, circuit.Path[0])
assert.Equal(t, r1, circuit.Path[1])
assert.Equal(t, r2, circuit.Path[2])
assert.Equal(t, 2, len(circuit.Links))
assert.Equal(t, l0, circuit.Links[0])
assert.Equal(t, l1, circuit.Links[1])
assert.Equal(t, r2, circuit.EgressRouter())
s0 := &identity.TokenId{Token: "s0"}
routeMessages, err := circuit.CreateRouteMessages(0, s0, addr)
assert.NotNil(t, routeMessages)
assert.Nil(t, err)
assert.Equal(t, 3, len(routeMessages))
// ingress route message
rm0 := routeMessages[0]
assert.Equal(t, s0.Token, rm0.SessionId)
assert.Nil(t, rm0.Egress)
assert.Equal(t, 2, len(rm0.Forwards))
assert.Equal(t, circuit.IngressId, rm0.Forwards[0].SrcAddress)
assert.Equal(t, l0.Id.Token, rm0.Forwards[0].DstAddress)
assert.Equal(t, l0.Id.Token, rm0.Forwards[1].SrcAddress)
assert.Equal(t, circuit.IngressId, rm0.Forwards[1].DstAddress)
// transit route message
rm1 := routeMessages[1]
assert.Equal(t, s0.Token, rm1.SessionId)
assert.Nil(t, rm1.Egress)
assert.Equal(t, 2, len(rm1.Forwards))
assert.Equal(t, l0.Id.Token, rm1.Forwards[0].SrcAddress)
assert.Equal(t, l1.Id.Token, rm1.Forwards[0].DstAddress)
assert.Equal(t, l1.Id.Token, rm1.Forwards[1].SrcAddress)
assert.Equal(t, l0.Id.Token, rm1.Forwards[1].DstAddress)
// egress route message
rm2 := routeMessages[2]
assert.Equal(t, s0.Token, rm2.SessionId)
assert.NotNil(t, rm2.Egress)
assert.Equal(t, circuit.EgressId, rm2.Egress.Address)
assert.Equal(t, transportAddr.String(), rm2.Egress.Destination)
assert.Equal(t, circuit.EgressId, rm2.Forwards[0].SrcAddress)
assert.Equal(t, l1.Id.Token, rm2.Forwards[0].DstAddress)
assert.Equal(t, l1.Id.Token, rm2.Forwards[1].SrcAddress)
assert.Equal(t, circuit.EgressId, rm2.Forwards[1].DstAddress)
}
func newRouterForTest(id string, fingerprint string, advLstnr transport.Address, ctrl channel2.Channel) *Router {
r := &Router{
BaseEntity: models.BaseEntity{Id: id},
Fingerprint: &fingerprint,
Control: ctrl,
}
if advLstnr != nil {
r.AdvertisedListener = advLstnr.String()
}
return r
}
type VersionProviderTest struct {
}
func (v VersionProviderTest) Branch() string {
return "local"
}
func (v VersionProviderTest) EncoderDecoder() common.VersionEncDec {
return &common.StdVersionEncDec
}
func (v VersionProviderTest) Version() string {
return "v0.0.0"
}
func (v VersionProviderTest) BuildDate() string {
return time.Now().String()
}
func (v VersionProviderTest) Revision() string {
return ""
}
func (v VersionProviderTest) AsVersionInfo() *common.VersionInfo {
return &common.VersionInfo{
Version: v.Version(),
Revision: v.Revision(),
BuildDate: v.BuildDate(),
OS: runtime.GOOS,
Arch: runtime.GOARCH,
}
}
func NewVersionProviderTest() common.VersionProvider {
return &VersionProviderTest{}
}
+56 -65
View File
@@ -348,29 +348,21 @@ func (network *Network) CreateSession(srcR *Router, clientId *identity.TokenId,
}
// 3: select terminator
strategy, terminator, path, err := network.selectPath(srcR, svc, targetIdentity)
strategy, terminator, pathNodes, err := network.selectPath(srcR, svc, targetIdentity)
if err != nil {
network.ServiceDialOtherError(serviceId)
return nil, err
}
// 4: Create Circuit
circuit, err := network.CreateCircuitWithPath(path)
// 4: Create Path
path, err := network.CreatePathWithNodes(pathNodes)
if err != nil {
network.ServiceDialOtherError(serviceId)
return nil, err
}
circuit.Binding = "transport"
if terminator.GetBinding() != "" {
circuit.Binding = terminator.GetBinding()
} else if strings.HasPrefix(terminator.GetBinding(), "hosted") {
circuit.Binding = "edge"
} else if strings.HasPrefix(terminator.GetAddress(), "udp") {
circuit.Binding = "udp"
}
// 4a: Create Route Messages
rms, err := circuit.CreateRouteMessages(attempt, sessionId, terminator.GetAddress())
rms, err := path.CreateRouteMessages(attempt, sessionId, terminator)
if err != nil {
network.ServiceDialOtherError(serviceId)
return nil, err
@@ -379,7 +371,7 @@ func (network *Network) CreateSession(srcR *Router, clientId *identity.TokenId,
// 5: Routing
logrus.Debugf("route attempt [#%d] for [s/%s]", attempt+1, sessionId.Token)
peerData, cleanups, err := rs.route(attempt, circuit, rms, strategy, terminator)
peerData, cleanups, err := rs.route(attempt, path, rms, strategy, terminator)
for k, v := range cleanups {
allCleanups[k] = v
}
@@ -392,7 +384,7 @@ func (network *Network) CreateSession(srcR *Router, clientId *identity.TokenId,
} else {
// revert successful routes
logrus.Warnf("session creation failed after [%d] attempts, sending cleanup unroutes for [s/%s]", network.options.CreateSessionRetries, sessionId.Token)
for cleanupRId, _ := range allCleanups {
for cleanupRId := range allCleanups {
if r, err := network.GetRouter(cleanupRId); err == nil {
if err := sendUnroute(r, sessionId, true); err == nil {
logrus.Debugf("sent cleanup unroute for [s/%s] to [r/%s]", sessionId.Token, r.Id)
@@ -410,11 +402,11 @@ func (network *Network) CreateSession(srcR *Router, clientId *identity.TokenId,
// 5.a: Unroute Abandoned Routers (from Previous Attempts)
usedRouters := make(map[string]struct{})
for _, r := range circuit.Path {
for _, r := range path.Nodes {
usedRouters[r.Id] = struct{}{}
}
cleanupCount := 0
for cleanupRId, _ := range allCleanups {
for cleanupRId := range allCleanups {
if _, found := usedRouters[cleanupRId]; !found {
cleanupCount++
if r, err := network.GetRouter(cleanupRId); err == nil {
@@ -435,14 +427,14 @@ func (network *Network) CreateSession(srcR *Router, clientId *identity.TokenId,
Id: sessionId,
ClientId: clientId,
Service: svc,
Circuit: circuit,
Path: path,
Terminator: terminator,
PeerData: peerData,
}
network.sessionController.add(ss)
network.SessionCreated(ss.Id, ss.ClientId, ss.Service.Id, ss.Circuit)
network.SessionCreated(ss.Id, ss.ClientId, ss.Service.Id, ss.Path)
logrus.Debugf("created session [s/%s] ==> %s", sessionId.Token, ss.Circuit)
logrus.Debugf("created session [s/%s] ==> %s", sessionId.Token, ss.Path)
return ss, nil
}
}
@@ -456,9 +448,9 @@ func parseIdentityAndService(service string) (string, string) {
if atIndex < 0 {
return "", service
}
identity := service[0:atIndex]
identityId := service[0:atIndex]
serviceId := service[atIndex+1:]
return identity, serviceId
return identityId, serviceId
}
func (network *Network) selectPath(srcR *Router, svc *Service, identity string) (xt.Strategy, xt.Terminator, []*Router, error) {
@@ -557,7 +549,7 @@ func (network *Network) RemoveSession(sessionId *identity.TokenId, now bool) err
log := pfxlog.Logger()
if ss, found := network.sessionController.get(sessionId); found {
for _, r := range ss.Circuit.Path {
for _, r := range ss.Path.Nodes {
err := sendUnroute(r, ss.Id, now)
if err != nil {
log.Errorf("error sending unroute to [r/%s] (%s)", r.Id, err)
@@ -579,7 +571,7 @@ func (network *Network) RemoveSession(sessionId *identity.TokenId, now bool) err
return InvalidSessionError{sessionId: sessionId.Token}
}
func (network *Network) CreateCircuit(srcR, dstR *Router) (*Circuit, error) {
func (network *Network) CreatePath(srcR, dstR *Router) (*Path, error) {
ingressId, err := network.sequence.NextHash()
if err != nil {
return nil, err
@@ -590,19 +582,19 @@ func (network *Network) CreateCircuit(srcR, dstR *Router) (*Circuit, error) {
return nil, err
}
circuit := &Circuit{
path := &Path{
Links: make([]*Link, 0),
IngressId: ingressId,
EgressId: egressId,
Path: make([]*Router, 0),
Nodes: make([]*Router, 0),
}
circuit.Path = append(circuit.Path, srcR)
circuit.Path = append(circuit.Path, dstR)
path.Nodes = append(path.Nodes, srcR)
path.Nodes = append(path.Nodes, dstR)
return network.UpdateCircuit(circuit)
return network.UpdatePath(path)
}
func (network *Network) CreateCircuitWithPath(path []*Router) (*Circuit, error) {
func (network *Network) CreatePathWithNodes(nodes []*Router) (*Path, error) {
ingressId, err := network.sequence.NextHash()
if err != nil {
return nil, err
@@ -613,44 +605,43 @@ func (network *Network) CreateCircuitWithPath(path []*Router) (*Circuit, error)
return nil, err
}
circuit := &Circuit{
Path: path,
path := &Path{
Nodes: nodes,
IngressId: ingressId,
EgressId: egressId,
}
if err := network.setLinks(circuit); err != nil {
if err := network.setLinks(path); err != nil {
return nil, err
}
return circuit, nil
return path, nil
}
func (network *Network) UpdateCircuit(circuit *Circuit) (*Circuit, error) {
srcR := circuit.Path[0]
dstR := circuit.Path[len(circuit.Path)-1]
path, _, err := network.shortestPath(srcR, dstR)
func (network *Network) UpdatePath(path *Path) (*Path, error) {
srcR := path.Nodes[0]
dstR := path.Nodes[len(path.Nodes)-1]
nodes, _, err := network.shortestPath(srcR, dstR)
if err != nil {
return nil, err
}
circuit2 := &Circuit{
Path: path,
Binding: circuit.Binding,
IngressId: circuit.IngressId,
EgressId: circuit.EgressId,
path2 := &Path{
Nodes: nodes,
IngressId: path.IngressId,
EgressId: path.EgressId,
}
if err := network.setLinks(circuit2); err != nil {
if err := network.setLinks(path2); err != nil {
return nil, err
}
return circuit2, nil
return path2, nil
}
func (network *Network) setLinks(circuit *Circuit) error {
if len(circuit.Path) > 1 {
for i := 0; i < len(circuit.Path)-1; i++ {
if link, found := network.linkController.leastExpensiveLink(circuit.Path[i], circuit.Path[i+1]); found {
circuit.Links = append(circuit.Links, link)
func (network *Network) setLinks(path *Path) error {
if len(path.Nodes) > 1 {
for i := 0; i < len(path.Nodes)-1; i++ {
if link, found := network.linkController.leastExpensiveLink(path.Nodes[i], path.Nodes[i+1]); found {
path.Links = append(path.Links, link)
} else {
return errors.Errorf("no link from r/%v to r/%v", circuit.Path[i].Id, circuit.Path[i+1].Id)
return errors.Errorf("no link from r/%v to r/%v", path.Nodes[i].Id, path.Nodes[i+1].Id)
}
}
}
@@ -720,7 +711,7 @@ func (network *Network) rerouteLink(l *Link) error {
sessions := network.sessionController.all()
for _, s := range sessions {
if s.Circuit.usesLink(l) {
if s.Path.usesLink(l) {
logrus.Infof("session [s/%s] uses link [l/%s]", s.Id.Token, l.Id.Token)
if err := network.rerouteSession(s); err != nil {
logrus.Errorf("error rerouting session [s/%s], removing", s.Id.Token)
@@ -740,24 +731,24 @@ func (network *Network) rerouteSession(s *Session) error {
logrus.Warnf("rerouting [s/%s]", s.Id.Token)
if cq, err := network.UpdateCircuit(s.Circuit); err == nil {
s.Circuit = cq
if cq, err := network.UpdatePath(s.Path); err == nil {
s.Path = cq
rms, err := cq.CreateRouteMessages(SmartRerouteAttempt, s.Id, s.Terminator.GetAddress())
rms, err := cq.CreateRouteMessages(SmartRerouteAttempt, s.Id, s.Terminator)
if err != nil {
logrus.Errorf("error creating route messages (%s)", err)
return err
}
for i := 0; i < len(cq.Path); i++ {
if _, err := sendRoute(cq.Path[i], rms[i], network.options.RouteTimeout); err != nil {
logrus.Errorf("error sending route to [r/%s] (%s)", cq.Path[i].Id, err)
for i := 0; i < len(cq.Nodes); i++ {
if _, err := sendRoute(cq.Nodes[i], rms[i], network.options.RouteTimeout); err != nil {
logrus.Errorf("error sending route to [r/%s] (%s)", cq.Nodes[i].Id, err)
}
}
logrus.Infof("rerouted session [s/%s]", s.Id.Token)
network.CircuitUpdated(s.Id, s.Circuit)
network.PathUpdated(s.Id, s.Path)
return nil
} else {
@@ -769,27 +760,27 @@ func (network *Network) rerouteSession(s *Session) error {
}
}
func (network *Network) smartReroute(s *Session, cq *Circuit) error {
func (network *Network) smartReroute(s *Session, cq *Path) error {
if s.Rerouting.CompareAndSwap(false, true) {
defer s.Rerouting.Set(false)
s.Circuit = cq
s.Path = cq
rms, err := cq.CreateRouteMessages(SmartRerouteAttempt, s.Id, s.Terminator.GetAddress())
rms, err := cq.CreateRouteMessages(SmartRerouteAttempt, s.Id, s.Terminator)
if err != nil {
logrus.Errorf("error creating route messages (%s)", err)
return err
}
for i := 0; i < len(cq.Path); i++ {
if _, err := sendRoute(cq.Path[i], rms[i], network.options.RouteTimeout); err != nil {
logrus.Errorf("error sending route to [r/%s] (%s)", cq.Path[i].Id, err)
for i := 0; i < len(cq.Nodes); i++ {
if _, err := sendRoute(cq.Nodes[i], rms[i], network.options.RouteTimeout); err != nil {
logrus.Errorf("error sending route to [r/%s] (%s)", cq.Nodes[i].Id, err)
}
}
logrus.Debugf("rerouted session [s/%s]", s.Id.Token)
network.CircuitUpdated(s.Id, s.Circuit)
network.PathUpdated(s.Id, s.Path)
return nil
+135 -1
View File
@@ -17,11 +17,145 @@
package network
import (
"errors"
"fmt"
"github.com/openziti/fabric/controller/xt"
"github.com/openziti/fabric/pb/ctrl_pb"
"github.com/openziti/foundation/identity/identity"
"github.com/pkg/errors"
"math"
)
type Path struct {
Nodes []*Router
Links []*Link
IngressId string
EgressId string
}
func (self *Path) String() string {
if len(self.Nodes) < 1 {
return "{}"
}
if len(self.Links) != len(self.Nodes)-1 {
return "{malformed}"
}
out := fmt.Sprintf("[r/%s]", self.Nodes[0].Id)
for i := 0; i < len(self.Links); i++ {
out += fmt.Sprintf("->[l/%s]", self.Links[i].Id.Token)
out += fmt.Sprintf("->[r/%s]", self.Nodes[i+1].Id)
}
return out
}
func (self *Path) EqualPath(other *Path) bool {
if len(self.Nodes) != len(other.Nodes) {
return false
}
if len(self.Links) != len(other.Links) {
return false
}
for i := 0; i < len(self.Nodes); i++ {
if self.Nodes[i] != other.Nodes[i] {
return false
}
}
for i := 0; i < len(self.Links); i++ {
if self.Links[i] != other.Links[i] {
return false
}
}
return true
}
func (self *Path) EgressRouter() *Router {
if len(self.Nodes) > 0 {
return self.Nodes[len(self.Nodes)-1]
}
return nil
}
func (self *Path) CreateRouteMessages(attempt uint32, sessionId *identity.TokenId, terminator xt.Terminator) ([]*ctrl_pb.Route, error) {
var routeMessages []*ctrl_pb.Route
if len(self.Links) == 0 {
// single router path
routeMessage := &ctrl_pb.Route{SessionId: sessionId.Token, Attempt: attempt}
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: self.IngressId,
DstAddress: self.EgressId,
})
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: self.EgressId,
DstAddress: self.IngressId,
})
routeMessage.Egress = &ctrl_pb.Route_Egress{
Binding: terminator.GetBinding(),
Address: self.EgressId,
Destination: terminator.GetAddress(),
}
routeMessages = append(routeMessages, routeMessage)
}
for i, link := range self.Links {
if i == 0 {
// ingress
routeMessage := &ctrl_pb.Route{SessionId: sessionId.Token, Attempt: attempt}
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: self.IngressId,
DstAddress: link.Id.Token,
})
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: link.Id.Token,
DstAddress: self.IngressId,
})
routeMessages = append(routeMessages, routeMessage)
}
if i >= 0 && i < len(self.Links)-1 {
// transit
nextLink := self.Links[i+1]
routeMessage := &ctrl_pb.Route{SessionId: sessionId.Token, Attempt: attempt}
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: link.Id.Token,
DstAddress: nextLink.Id.Token,
})
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: nextLink.Id.Token,
DstAddress: link.Id.Token,
})
routeMessages = append(routeMessages, routeMessage)
}
if i == len(self.Links)-1 {
// egress
routeMessage := &ctrl_pb.Route{SessionId: sessionId.Token, Attempt: attempt}
routeMessage.Egress = &ctrl_pb.Route_Egress{
Binding: terminator.GetBinding(),
Address: self.EgressId,
Destination: terminator.GetAddress(),
}
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: self.EgressId,
DstAddress: link.Id.Token,
})
routeMessage.Forwards = append(routeMessage.Forwards, &ctrl_pb.Route_Forward{
SrcAddress: link.Id.Token,
DstAddress: self.EgressId,
})
routeMessages = append(routeMessages, routeMessage)
}
}
return routeMessages, nil
}
func (self *Path) usesLink(l *Link) bool {
if self.Links != nil {
for _, o := range self.Links {
if o == l {
return true
}
}
}
return false
}
func (network *Network) shortestPath(srcR *Router, dstR *Router) ([]*Router, int64, error) {
if srcR == nil || dstR == nil {
return nil, 0, errors.New("not routable (!srcR||!dstR)")
+208
View File
@@ -18,12 +18,220 @@ package network
import (
"github.com/openziti/fabric/controller/db"
"github.com/openziti/fabric/controller/models"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/common"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/transport"
"github.com/openziti/foundation/transport/tcp"
"github.com/stretchr/testify/assert"
"runtime"
"testing"
"time"
)
func TestSimplePath2(t *testing.T) {
ctx := db.NewTestContext(t)
defer ctx.Cleanup()
closeNotify := make(chan struct{})
defer close(closeNotify)
nodeId := &identity.TokenId{Token: "test"}
network, err := NewNetwork(nodeId, nil, ctx.GetDb(), nil, NewVersionProviderTest(), closeNotify)
assert.Nil(t, err)
addr := "tcp:0.0.0.0:0"
transportAddr, err := tcp.AddressParser{}.Parse(addr)
assert.Nil(t, err)
r0 := newRouterForTest("r0", "", transportAddr, nil)
network.Routers.markConnected(r0)
r1 := newRouterForTest("r1", "", transportAddr, nil)
network.Routers.markConnected(r1)
l0 := newLink(&identity.TokenId{Token: "l0"})
l0.Src = r0
l0.Dst = r1
l0.addState(newLinkState(Connected))
network.linkController.add(l0)
path, err := network.CreatePath(r0, r1)
assert.NotNil(t, path)
assert.Nil(t, err)
assert.Equal(t, 2, len(path.Nodes))
assert.Equal(t, r0, path.Nodes[0])
assert.Equal(t, r1, path.Nodes[1])
assert.Equal(t, 1, len(path.Links))
assert.Equal(t, l0, path.Links[0])
assert.Equal(t, r1, path.EgressRouter())
s0 := &identity.TokenId{Token: "s0"}
terminator := &Terminator{Address: addr, Binding: "transport"}
routeMessages, err := path.CreateRouteMessages(0, s0, terminator)
assert.NotNil(t, routeMessages)
assert.Nil(t, err)
assert.Equal(t, 2, len(routeMessages))
// ingress route message
rm0 := routeMessages[0]
assert.Equal(t, s0.Token, rm0.SessionId)
assert.Nil(t, rm0.Egress)
assert.Equal(t, 2, len(rm0.Forwards))
assert.Equal(t, path.IngressId, rm0.Forwards[0].SrcAddress)
assert.Equal(t, l0.Id.Token, rm0.Forwards[0].DstAddress)
assert.Equal(t, l0.Id.Token, rm0.Forwards[1].SrcAddress)
assert.Equal(t, path.IngressId, rm0.Forwards[1].DstAddress)
// egress route message
rm1 := routeMessages[1]
assert.Equal(t, s0.Token, rm1.SessionId)
assert.NotNil(t, rm1.Egress)
assert.Equal(t, path.EgressId, rm1.Egress.Address)
assert.Equal(t, addr, rm1.Egress.Destination)
assert.Equal(t, path.EgressId, rm1.Forwards[0].SrcAddress)
assert.Equal(t, l0.Id.Token, rm1.Forwards[0].DstAddress)
assert.Equal(t, l0.Id.Token, rm1.Forwards[1].SrcAddress)
assert.Equal(t, path.EgressId, rm1.Forwards[1].DstAddress)
}
func TestTransitPath2(t *testing.T) {
ctx := db.NewTestContext(t)
defer ctx.Cleanup()
closeNotify := make(chan struct{})
defer close(closeNotify)
nodeId := &identity.TokenId{Token: "test"}
network, err := NewNetwork(nodeId, nil, ctx.GetDb(), nil, NewVersionProviderTest(), closeNotify)
assert.Nil(t, err)
addr := "tcp:0.0.0.0:0"
transportAddr, err := tcp.AddressParser{}.Parse(addr)
assert.Nil(t, err)
r0 := newRouterForTest("r0", "", transportAddr, nil)
network.Routers.markConnected(r0)
r1 := newRouterForTest("r1", "", transportAddr, nil)
network.Routers.markConnected(r1)
r2 := newRouterForTest("r2", "", transportAddr, nil)
network.Routers.markConnected(r2)
l0 := newLink(&identity.TokenId{Token: "l0"})
l0.Src = r0
l0.Dst = r1
l0.addState(newLinkState(Connected))
network.linkController.add(l0)
l1 := newLink(&identity.TokenId{Token: "l1"})
l1.Src = r1
l1.Dst = r2
l1.addState(newLinkState(Connected))
network.linkController.add(l1)
path, err := network.CreatePath(r0, r2)
assert.NotNil(t, path)
assert.Nil(t, err)
assert.Equal(t, 3, len(path.Nodes))
assert.Equal(t, r0, path.Nodes[0])
assert.Equal(t, r1, path.Nodes[1])
assert.Equal(t, r2, path.Nodes[2])
assert.Equal(t, 2, len(path.Links))
assert.Equal(t, l0, path.Links[0])
assert.Equal(t, l1, path.Links[1])
assert.Equal(t, r2, path.EgressRouter())
s0 := &identity.TokenId{Token: "s0"}
terminator := &Terminator{Address: addr, Binding: "transport"}
routeMessages, err := path.CreateRouteMessages(0, s0, terminator)
assert.NotNil(t, routeMessages)
assert.Nil(t, err)
assert.Equal(t, 3, len(routeMessages))
// ingress route message
rm0 := routeMessages[0]
assert.Equal(t, s0.Token, rm0.SessionId)
assert.Nil(t, rm0.Egress)
assert.Equal(t, 2, len(rm0.Forwards))
assert.Equal(t, path.IngressId, rm0.Forwards[0].SrcAddress)
assert.Equal(t, l0.Id.Token, rm0.Forwards[0].DstAddress)
assert.Equal(t, l0.Id.Token, rm0.Forwards[1].SrcAddress)
assert.Equal(t, path.IngressId, rm0.Forwards[1].DstAddress)
// transit route message
rm1 := routeMessages[1]
assert.Equal(t, s0.Token, rm1.SessionId)
assert.Nil(t, rm1.Egress)
assert.Equal(t, 2, len(rm1.Forwards))
assert.Equal(t, l0.Id.Token, rm1.Forwards[0].SrcAddress)
assert.Equal(t, l1.Id.Token, rm1.Forwards[0].DstAddress)
assert.Equal(t, l1.Id.Token, rm1.Forwards[1].SrcAddress)
assert.Equal(t, l0.Id.Token, rm1.Forwards[1].DstAddress)
// egress route message
rm2 := routeMessages[2]
assert.Equal(t, s0.Token, rm2.SessionId)
assert.NotNil(t, rm2.Egress)
assert.Equal(t, path.EgressId, rm2.Egress.Address)
assert.Equal(t, transportAddr.String(), rm2.Egress.Destination)
assert.Equal(t, path.EgressId, rm2.Forwards[0].SrcAddress)
assert.Equal(t, l1.Id.Token, rm2.Forwards[0].DstAddress)
assert.Equal(t, l1.Id.Token, rm2.Forwards[1].SrcAddress)
assert.Equal(t, path.EgressId, rm2.Forwards[1].DstAddress)
}
func newRouterForTest(id string, fingerprint string, advLstnr transport.Address, ctrl channel2.Channel) *Router {
r := &Router{
BaseEntity: models.BaseEntity{Id: id},
Fingerprint: &fingerprint,
Control: ctrl,
}
if advLstnr != nil {
r.AdvertisedListener = advLstnr.String()
}
return r
}
type VersionProviderTest struct {
}
func (v VersionProviderTest) Branch() string {
return "local"
}
func (v VersionProviderTest) EncoderDecoder() common.VersionEncDec {
return &common.StdVersionEncDec
}
func (v VersionProviderTest) Version() string {
return "v0.0.0"
}
func (v VersionProviderTest) BuildDate() string {
return time.Now().String()
}
func (v VersionProviderTest) Revision() string {
return ""
}
func (v VersionProviderTest) AsVersionInfo() *common.VersionInfo {
return &common.VersionInfo{
Version: v.Version(),
Revision: v.Revision(),
BuildDate: v.BuildDate(),
OS: runtime.GOOS,
Arch: runtime.GOARCH,
}
}
func NewVersionProviderTest() common.VersionProvider {
return &VersionProviderTest{}
}
func TestShortestPath(t *testing.T) {
ctx := db.NewTestContext(t)
defer ctx.Cleanup()
+9 -9
View File
@@ -57,7 +57,7 @@ func (self *routeSenderController) removeRouteSender(rs *routeSender) {
type routeSender struct {
sessionId string
circuit *Circuit
path *Path
routeMsgs []*ctrl_pb.Route
timeout time.Duration
in chan *routeStatus
@@ -75,11 +75,11 @@ func newRouteSender(sessionId string, timeout time.Duration, serviceCounters Ser
}
}
func (self *routeSender) route(attempt uint32, circuit *Circuit, routeMsgs []*ctrl_pb.Route, strategy xt.Strategy, terminator xt.Terminator) (peerData xt.PeerData, cleanups map[string]struct{}, err error) {
func (self *routeSender) route(attempt uint32, path *Path, routeMsgs []*ctrl_pb.Route, strategy xt.Strategy, terminator xt.Terminator) (peerData xt.PeerData, cleanups map[string]struct{}, err error) {
// send route messages
tr := circuit.Path[len(circuit.Path)-1]
for i := 0; i < len(circuit.Path); i++ {
r := circuit.Path[i]
tr := path.Nodes[len(path.Nodes)-1]
for i := 0; i < len(path.Nodes); i++ {
r := path.Nodes[i]
go self.sendRoute(r, routeMsgs[i])
self.attendance[r.Id] = false
}
@@ -112,7 +112,7 @@ attendance:
strategy.NotifyEvent(xt.NewDialFailedEvent(terminator))
self.serviceCounters.ServiceDialFail(terminator.GetServiceId())
}
cleanups = self.cleanups(circuit)
cleanups = self.cleanups(path)
return nil, cleanups, errors.Errorf("error creating route for [s/%s] on [r/%s] (%v)", self.sessionId, status.r.Id, status.rerr)
} else {
@@ -121,7 +121,7 @@ attendance:
}
case <-time.After(timeout):
cleanups = self.cleanups(circuit)
cleanups = self.cleanups(path)
self.serviceCounters.ServiceDialTimeout(terminator.GetServiceId())
return nil, cleanups, &routeTimeoutError{sessionId: self.sessionId}
}
@@ -153,9 +153,9 @@ func (self *routeSender) sendRoute(r *Router, routeMsg *ctrl_pb.Route) {
logrus.Debugf("sent route message for [s/%s] to [r/%s]", routeMsg.SessionId, r.Id)
}
func (self *routeSender) cleanups(circuit *Circuit) map[string]struct{} {
func (self *routeSender) cleanups(path *Path) map[string]struct{} {
cleanups := make(map[string]struct{})
for _, r := range circuit.Path {
for _, r := range path.Nodes {
success, found := self.attendance[r.Id]
if found && success {
cleanups[r.Id] = struct{}{}
+2 -2
View File
@@ -28,14 +28,14 @@ type Session struct {
ClientId *identity.TokenId
Service *Service
Terminator xt.Terminator
Circuit *Circuit
Path *Path
Rerouting concurrenz.AtomicBoolean
PeerData xt.PeerData
}
func (s *Session) latency() int64 {
var latency int64
for _, l := range s.Circuit.Links {
for _, l := range s.Path.Links {
latency += l.GetSrcLatency()
latency += l.GetDstLatency()
}
+13 -13
View File
@@ -28,17 +28,17 @@ func getSessionEventHandlers() []SessionEventHandler {
}
type SessionEventHandler interface {
SessionCreated(sessionId *identity.TokenId, clientId *identity.TokenId, serviceId string, circuit *Circuit)
SessionCreated(sessionId *identity.TokenId, clientId *identity.TokenId, serviceId string, path *Path)
SessionDeleted(sessionId *identity.TokenId, clientId *identity.TokenId)
CircuitUpdated(sessionId *identity.TokenId, circuit *Circuit)
PathUpdated(sessionId *identity.TokenId, path *Path)
}
func (network *Network) SessionCreated(sessionId *identity.TokenId, clientId *identity.TokenId, serviceId string, circuit *Circuit) {
func (network *Network) SessionCreated(sessionId *identity.TokenId, clientId *identity.TokenId, serviceId string, path *Path) {
event := &sessionCreatedEvent{
sessionId: sessionId,
clientId: clientId,
serviceId: serviceId,
circuit: circuit,
path: path,
}
network.eventDispatcher.Dispatch(event)
}
@@ -51,10 +51,10 @@ func (network *Network) SessionDeleted(sessionId *identity.TokenId, clientId *id
network.eventDispatcher.Dispatch(event)
}
func (network *Network) CircuitUpdated(sessionId *identity.TokenId, circuit *Circuit) {
event := &sessionCircuitEvent{
func (network *Network) PathUpdated(sessionId *identity.TokenId, path *Path) {
event := &sessionPathEvent{
sessionId: sessionId,
circuit: circuit,
path: path,
}
network.eventDispatcher.Dispatch(event)
}
@@ -63,13 +63,13 @@ type sessionCreatedEvent struct {
sessionId *identity.TokenId
clientId *identity.TokenId
serviceId string
circuit *Circuit
path *Path
}
func (event *sessionCreatedEvent) Handle() {
handlers := getSessionEventHandlers()
for _, handler := range handlers {
go handler.SessionCreated(event.sessionId, event.clientId, event.serviceId, event.circuit)
go handler.SessionCreated(event.sessionId, event.clientId, event.serviceId, event.path)
}
}
@@ -85,14 +85,14 @@ func (event *sessionDeletedEvent) Handle() {
}
}
type sessionCircuitEvent struct {
type sessionPathEvent struct {
sessionId *identity.TokenId
circuit *Circuit
path *Path
}
func (event *sessionCircuitEvent) Handle() {
func (event *sessionPathEvent) Handle() {
handlers := getSessionEventHandlers()
for _, handler := range handlers {
go handler.CircuitUpdated(event.sessionId, event.circuit)
go handler.PathUpdated(event.sessionId, event.path)
}
}
+6 -6
View File
@@ -53,7 +53,7 @@ func (network *Network) smart() {
/*
* Develop candidates for rerouting.
*/
newCircuits := make(map[*Session]*Circuit)
newPaths := make(map[*Session]*Path)
var candidates []*Session
count := 0
ceiling := int(float32(len(sessions)) * network.options.Smart.RerouteFraction)
@@ -66,13 +66,13 @@ func (network *Network) smart() {
log.Tracef("smart reroute ceiling [%d]", ceiling)
for _, sId := range orderedSessions {
if session, found := network.GetSession(sId); found {
if updatedCircuit, err := network.UpdateCircuit(session.Circuit); err == nil {
if !updatedCircuit.EqualPath(session.Circuit) {
if updatedPath, err := network.UpdatePath(session.Path); err == nil {
if !updatedPath.EqualPath(session.Path) {
if count < ceiling {
count++
candidates = append(candidates, session)
newCircuits[session] = updatedCircuit
log.Debugf("rerouting [s/%s] [l:%d] %s ==> %s", session.Id.Token, sessionLatencies[session.Id], session.Circuit.String(), updatedCircuit.String())
newPaths[session] = updatedPath
log.Debugf("rerouting [s/%s] [l:%d] %s ==> %s", session.Id.Token, sessionLatencies[session.Id], session.Path.String(), updatedPath.String())
}
}
}
@@ -84,7 +84,7 @@ func (network *Network) smart() {
* Reroute.
*/
for _, session := range candidates {
if err := network.smartReroute(session, newCircuits[session]); err != nil {
if err := network.smartReroute(session, newPaths[session]); err != nil {
log.Errorf("error rerouting [s/%s] (%s)", session.Id.Token, err)
}
}