diff --git a/router/router.go b/router/router.go index 0b7854ef8..8437a6b68 100644 --- a/router/router.go +++ b/router/router.go @@ -32,6 +32,7 @@ import ( "github.com/netfoundry/ziti-fabric/router/xgress_proxy_udp" "github.com/netfoundry/ziti-fabric/router/xgress_transport" "github.com/netfoundry/ziti-fabric/router/xgress_transport_udp" + "github.com/netfoundry/ziti-fabric/router/xlink" "github.com/netfoundry/ziti-foundation/channel2" "github.com/netfoundry/ziti-foundation/profiler" "github.com/netfoundry/ziti-foundation/util/info" @@ -49,6 +50,7 @@ type Router struct { forwarder *forwarder.Forwarder xctrls []xctrl.Xctrl xctrlDone chan struct{} + xlinks map[string]xlink.Xlink metricsRegistry metrics.Registry } diff --git a/router/xlink/xlink.go b/router/xlink/xlink.go index 2e2532595..3096adab4 100644 --- a/router/xlink/xlink.go +++ b/router/xlink/xlink.go @@ -24,6 +24,6 @@ type Factory interface { type Xlink interface { Listen() error - Dial(address string) error + Dial(address, linkId string) error GetAdvertisement() string } \ No newline at end of file diff --git a/router/xlink_transport/factory.go b/router/xlink_transport/factory.go index 53bc99d66..b87cf1efa 100644 --- a/router/xlink_transport/factory.go +++ b/router/xlink_transport/factory.go @@ -18,6 +18,7 @@ package xlink_transport import ( "fmt" + "github.com/netfoundry/ziti-fabric/metrics" forwarder2 "github.com/netfoundry/ziti-fabric/router/forwarder" "github.com/netfoundry/ziti-fabric/router/xgress" "github.com/netfoundry/ziti-fabric/router/xlink" @@ -25,12 +26,13 @@ import ( "github.com/netfoundry/ziti-foundation/identity/identity" ) -func NewFactory(ctrl xgress.CtrlChannel, forwarder *forwarder2.Forwarder, forwarderOptions *forwarder2.Options, options *channel2.Options) xlink.Factory { +func NewFactory(ctrl xgress.CtrlChannel, f *forwarder2.Forwarder, fo *forwarder2.Options, mr metrics.Registry, lo *channel2.Options) xlink.Factory { return &factory{ ctrl: ctrl, - forwarder: forwarder, - forwarderOptions: forwarderOptions, - options: options, + forwarder: f, + forwarderOptions: fo, + metricsRegistry: mr, + linkOptions: lo, } } @@ -45,7 +47,7 @@ func (self *factory) Create(id *identity.TokenId, configData map[interface{}]int ctrl: self.ctrl, forwarder: self.forwarder, forwarderOptions: self.forwarderOptions, - options: self.options, + linkOptions: self.linkOptions, }, nil } @@ -53,5 +55,6 @@ type factory struct { ctrl xgress.CtrlChannel forwarder *forwarder2.Forwarder forwarderOptions *forwarder2.Options - options *channel2.Options + metricsRegistry metrics.Registry + linkOptions *channel2.Options } diff --git a/router/xlink_transport/xlink.go b/router/xlink_transport/xlink.go index bcb8ff88b..8d43421fa 100644 --- a/router/xlink_transport/xlink.go +++ b/router/xlink_transport/xlink.go @@ -18,11 +18,16 @@ package xlink_transport import ( "fmt" + "github.com/golang/protobuf/proto" + "github.com/netfoundry/ziti-fabric/metrics" + "github.com/netfoundry/ziti-fabric/pb/ctrl_pb" "github.com/netfoundry/ziti-fabric/router/forwarder" "github.com/netfoundry/ziti-fabric/router/handler_link" "github.com/netfoundry/ziti-fabric/router/xgress" "github.com/netfoundry/ziti-foundation/channel2" "github.com/netfoundry/ziti-foundation/identity/identity" + "github.com/netfoundry/ziti-foundation/transport" + log "github.com/sirupsen/logrus" ) func (self *impl) Listen() error { @@ -30,11 +35,63 @@ func (self *impl) Listen() error { if err := self.listener.Listen(); err != nil { return fmt.Errorf("error listening (%w)", err) } - go handler_link.NewAccepter(self.id, self.ctrl, self.forwarder, self.listener, self.options, self.forwarderOptions) + go handler_link.NewAccepter(self.id, self.ctrl, self.forwarder, self.listener, self.linkOptions, self.forwarderOptions) return nil } -func (_ *impl) Dial(address string) error { +func (self *impl) Dial(addressString, linkId string) error { + address, err := transport.ParseAddress(addressString) + if err == nil { + linkId := self.id.ShallowCloneWithNewToken(linkId) + name := "l/" + linkId.Token + + dialer := channel2.NewClassicDialer(linkId, address, nil) + ch, err := channel2.NewChannel(name, dialer, self.linkOptions) + if err == nil { + link := forwarder.NewLink(linkId, ch) + if err := link.Channel.Bind(handler_link.NewBindHandler(self.id, link, self.ctrl, self.forwarder)); err == nil { + self.forwarder.RegisterLink(link) + + go metrics.ProbeLatency( + link.Channel, + self.metricsRegistry.Histogram("link."+linkId.Token+".latency"), + self.forwarderOptions.LatencyProbeInterval, + ) + + linkMsg := &ctrl_pb.Link{Id: link.Id.Token} + body, err := proto.Marshal(linkMsg) + if err == nil { + msg := channel2.NewMessage(int32(ctrl_pb.ContentType_LinkType), body) + if err := self.ctrl.Channel().Send(msg); err == nil { + log.Infof("link [l/%s] up", link.Id.Token) + } else { + log.Errorf("unexpected error sending link (%s)", err) + } + + } else { + log.Errorf("unexpected error (%s)", err) + } + } else { + log.Errorf("error binding link (%s)", err) + } + } else { + log.Errorf("link dialing failed [%s]", address.String()) + + fault := &ctrl_pb.Fault{Subject: ctrl_pb.FaultSubject_LinkFault, Id: linkId.Token} + body, err := proto.Marshal(fault) + if err == nil { + msg := channel2.NewMessage(int32(ctrl_pb.ContentType_FaultType), body) + if err := self.ctrl.Channel().Send(msg); err != nil { + log.Errorf("error sending fault (%s)", err) + } + } else { + log.Errorf("unexpected error (%s)", err) + } + } + } else { + log.Errorf("link address parsing failed (%s)", err) + } + return nil } @@ -49,5 +106,6 @@ type impl struct { ctrl xgress.CtrlChannel forwarder *forwarder.Forwarder forwarderOptions *forwarder.Options - options *channel2.Options + metricsRegistry metrics.Registry + linkOptions *channel2.Options }