diff --git a/router/forwarder/forwarder.go b/router/forwarder/forwarder.go index ec2c787bf..b3cc88dd8 100644 --- a/router/forwarder/forwarder.go +++ b/router/forwarder/forwarder.go @@ -43,6 +43,7 @@ type Forwarder struct { type Destination interface { SendPayload(payload *xgress.Payload) error SendAcknowledgement(acknowledgement *xgress.Acknowledgement) error + SendControl(control *xgress.Control) error } type XgressDestination interface { @@ -188,6 +189,39 @@ func (forwarder *Forwarder) ForwardAcknowledgement(srcAddr xgress.Address, ackno } } +func (forwarder *Forwarder) ForwardControl(srcAddr xgress.Address, control *xgress.Control) error { + log := pfxlog.ContextLogger(string(srcAddr)) + + circuitId := control.CircuitId + if forwardTable, found := forwarder.circuits.getForwardTable(circuitId); found { + if dstAddr, found := forwardTable.getForwardAddress(srcAddr); found { + if dst, found := forwarder.destinations.getDestination(dstAddr); found { + if control.IsTypeTraceRoute() { + hops := control.DecrementAndGetHop() + if hops == 0 { + resp := control.CreateTraceResponse("forwarder", forwarder.metricsRegistry.SourceId()) + return forwarder.ForwardControl(dstAddr, resp) + } + } + if err := dst.SendControl(control); err != nil { + return err + } + log.Debugf("=> %s", string(dstAddr)) + return nil + + } else { + return errors.Errorf("cannot forward control, no destination for circuit=%v src=%v dst=%v", circuitId, srcAddr, dstAddr) + } + + } else { + return errors.Errorf("cannot forward control, no destination address for circuit=%v src=%v", circuitId, srcAddr) + } + + } else { + return errors.Errorf("cannot forward control, no forward table for circuit=%v src=%v", circuitId, srcAddr) + } +} + func (forwarder *Forwarder) ReportForwardingFault(circuitId string) { if forwarder.faulter != nil { forwarder.faulter.report(circuitId) diff --git a/router/handler_link/accepter.go b/router/handler_link/accepter.go index e8951382f..6ba3a01ff 100644 --- a/router/handler_link/accepter.go +++ b/router/handler_link/accepter.go @@ -27,9 +27,10 @@ func (self *channelAccepter) AcceptChannel(xlink xlink.Xlink, ch channel2.Channe ch.SetUserData(xlink.Id().Token) ch.AddCloseHandler(newCloseHandler(xlink, self.ctrl, self.forwarder, closeNotify)) ch.AddErrorHandler(newErrorHandler(xlink, self.ctrl)) - ch.AddReceiveHandler(newPayloadHandler(xlink, self.ctrl, self.forwarder)) - ch.AddReceiveHandler(newQueuingAckHandler(xlink, self.ctrl, self.forwarder, closeNotify)) + ch.AddReceiveHandler(newPayloadHandler(xlink, self.forwarder)) + ch.AddReceiveHandler(newQueuingAckHandler(xlink, self.forwarder, closeNotify)) ch.AddReceiveHandler(&channel2.LatencyHandler{}) + ch.AddReceiveHandler(newControlHandler(xlink, ch, self.forwarder, closeNotify)) ch.AddPeekHandler(metrics2.NewChannelPeekHandler(xlink.Id().Token, self.forwarder.MetricsRegistry())) ch.AddPeekHandler(trace.NewChannelPeekHandler(xlink.Id().Token, ch, self.forwarder.TraceController(), trace.NewChannelSink(self.ctrl.Channel()))) diff --git a/router/handler_link/ack.go b/router/handler_link/ack.go index 18c62434c..4837b7a00 100644 --- a/router/handler_link/ack.go +++ b/router/handler_link/ack.go @@ -26,14 +26,12 @@ import ( type ackHandler struct { link xlink.Xlink - ctrl xgress.CtrlChannel forwarder *forwarder.Forwarder } -func newAckHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder) *ackHandler { +func newAckHandler(link xlink.Xlink, forwarder *forwarder.Forwarder) *ackHandler { return &ackHandler{ link: link, - ctrl: ctrl, forwarder: forwarder, } } diff --git a/router/handler_link/control.go b/router/handler_link/control.go new file mode 100644 index 000000000..e40b556d7 --- /dev/null +++ b/router/handler_link/control.go @@ -0,0 +1,75 @@ +/* + 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_link + +import ( + "github.com/michaelquigley/pfxlog" + "github.com/openziti/fabric/router/forwarder" + "github.com/openziti/fabric/router/xgress" + "github.com/openziti/fabric/router/xlink" + "github.com/openziti/foundation/channel2" +) + +type controlHandler struct { + link xlink.Xlink + forwarder *forwarder.Forwarder + ctrls chan *xgress.Control +} + +func newControlHandler(link xlink.Xlink, ch channel2.Channel, forwarder *forwarder.Forwarder, closeNotify <-chan struct{}) *controlHandler { + result := &controlHandler{ + link: link, + forwarder: forwarder, + ctrls: make(chan *xgress.Control, 4), + } + go result.run(ch.Label(), closeNotify) + return result +} + +func (self *controlHandler) ContentType() int32 { + return xgress.ContentTypeControlType +} + +func (self *controlHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { + log := pfxlog.ContextLogger(ch.Label()) + + control, err := xgress.UnmarshallControl(msg) + if err == nil { + if err := self.forwarder.ForwardControl(xgress.Address(self.link.Id().Token), control); err != nil { + log.WithError(err).Debug("unable to forward") + } + } else { + log.Errorf("unexpected error (%v)", err) + } +} + +func (self *controlHandler) run(label string, closeNotify <-chan struct{}) { + log := pfxlog.ContextLogger(label) + log.Info("starting") + defer log.Info("exiting") + + for { + select { + case control := <-self.ctrls: + if err := self.forwarder.ForwardControl(xgress.Address(self.link.Id().Token), control); err != nil { + log.WithError(err).Debug("unable to forward") + } + case <-closeNotify: + return + } + } +} diff --git a/router/handler_link/payload.go b/router/handler_link/payload.go index 4ba5402ff..b54d46986 100644 --- a/router/handler_link/payload.go +++ b/router/handler_link/payload.go @@ -26,14 +26,12 @@ import ( type payloadHandler struct { link xlink.Xlink - ctrl xgress.CtrlChannel forwarder *forwarder.Forwarder } -func newPayloadHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder) *payloadHandler { +func newPayloadHandler(link xlink.Xlink, forwarder *forwarder.Forwarder) *payloadHandler { return &payloadHandler{ link: link, - ctrl: ctrl, forwarder: forwarder, } } diff --git a/router/handler_link/queue_ack.go b/router/handler_link/queue_ack.go index b7d720ad1..f096b2a38 100644 --- a/router/handler_link/queue_ack.go +++ b/router/handler_link/queue_ack.go @@ -28,7 +28,6 @@ import ( type queuingAckHandler struct { link xlink.Xlink - ctrl xgress.CtrlChannel forwarder *forwarder.Forwarder acks *deque.Deque ackIngest chan *xgress.Acknowledgement @@ -37,10 +36,9 @@ type queuingAckHandler struct { closeNotify <-chan struct{} } -func newQueuingAckHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder, closeNotify <-chan struct{}) *queuingAckHandler { +func newQueuingAckHandler(link xlink.Xlink, forwarder *forwarder.Forwarder, closeNotify <-chan struct{}) *queuingAckHandler { result := &queuingAckHandler{ link: link, - ctrl: ctrl, forwarder: forwarder, acks: deque.New(), ackIngest: make(chan *xgress.Acknowledgement, 16), diff --git a/router/handler_xgress/receive.go b/router/handler_xgress/receive.go index c69adb4d9..76c8bf5c1 100644 --- a/router/handler_xgress/receive.go +++ b/router/handler_xgress/receive.go @@ -32,6 +32,12 @@ func NewReceiveHandler(forwarder *forwarder.Forwarder) *receiveHandler { func (xrh *receiveHandler) HandleXgressReceive(payload *xgress.Payload, x *xgress.Xgress) { if err := xrh.forwarder.ForwardPayload(x.Address(), payload); err != nil { - pfxlog.ContextLogger(x.Label()).WithFields(payload.GetLoggerFields()).Errorf("unable to forward (%s)", err) + pfxlog.ContextLogger(x.Label()).WithFields(payload.GetLoggerFields()).WithError(err).Error("unable to forward payload") + } +} + +func (xrh *receiveHandler) HandleControlReceive(control *xgress.Control, x *xgress.Xgress) { + if err := xrh.forwarder.ForwardControl(x.Address(), control); err != nil { + pfxlog.ContextLogger(x.Label()).WithFields(control.GetLoggerFields()).WithError(err).Error("unable to forward control") } } diff --git a/router/xgress/messages.go b/router/xgress/messages.go index ef990d15b..72681604a 100644 --- a/router/xgress/messages.go +++ b/router/xgress/messages.go @@ -22,6 +22,7 @@ import ( "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/util/info" "github.com/openziti/foundation/util/uuidz" + "github.com/pkg/errors" "github.com/sirupsen/logrus" "math" ) @@ -38,11 +39,13 @@ const ( ContentTypePayloadType = 1100 ContentTypeAcknowledgementType = 1101 + ContentTypeControlType = 1102 ) var ContentTypeValue = map[string]int32{ "PayloadType": ContentTypePayloadType, "AcknowledgementType": ContentTypeAcknowledgementType, + "ControlType": ContentTypeControlType, } type Originator int32 @@ -280,3 +283,87 @@ func (payload *Payload) GetLoggerFields() logrus.Fields { "uuid": uuidz.ToString(payload.Headers[HeaderKeyUUID]), } } + +type ControlType byte + +const ( + ControlTypeTraceRoute ControlType = 1 + ControlTypeTraceRouteResponse = 2 +) + +const ( + ControlHopCount = 1 + ControlHopType = 2 + ControlHopId = 3 + ControlTimestamp = 4 + ControlCustom1 = 5 + ControlCustom2 = 6 +) + +type Control struct { + Type ControlType + CircuitId string + Headers channel2.Headers +} + +func (self *Control) Marshall() *channel2.Message { + msg := channel2.NewMessage(ContentTypeControlType, append([]byte{byte(self.Type)}, self.CircuitId...)) + msg.Headers = self.Headers + return msg +} + +func UnmarshallControl(msg *channel2.Message) (*Control, error) { + if len(msg.Body) < 2 { + return nil, errors.New("control message body too short") + } + return &Control{ + Type: ControlType(msg.Body[0]), + CircuitId: string(msg.Body[1:]), + Headers: msg.Headers, + }, nil +} + +func (self *Control) IsTypeTraceRoute() bool { + return self.Type == ControlTypeTraceRoute +} + +func (self *Control) IsTypeTraceRouteResponse() bool { + return self.Type == ControlTypeTraceRouteResponse +} + +func (self *Control) DecrementAndGetHop() uint32 { + hop, _ := self.Headers.GetUint32Header(ControlHopCount) + if hop == 0 { + return 0 + } + hop-- + self.Headers.PutUint32Header(ControlHopCount, hop) + return hop +} + +func (self *Control) CreateTraceResponse(hopType, hopId string) *Control { + resp := &Control{ + Type: ControlTypeTraceRouteResponse, + CircuitId: self.CircuitId, + Headers: self.Headers, + } + resp.Headers.PutStringHeader(ControlHopType, hopType) + resp.Headers.PutStringHeader(ControlHopId, hopId) + return resp +} + +func (self *Control) GetLoggerFields() logrus.Fields { + return logrus.Fields{ + "circuitId": self.CircuitId, + "type": self.Type, + "uuid": uuidz.ToString(self.Headers[HeaderKeyUUID]), + } +} + +func RespondToTraceRequest(headers channel2.Headers, hopType, hopId string, response ControlReceiver) { + resp := &Control{Headers: headers} + resp.DecrementAndGetHop() + resp.Headers.PutStringHeader(ControlHopType, hopType) + resp.Headers.PutStringHeader(ControlHopId, hopId) + response.HandleControlReceive(ControlTypeTraceRouteResponse, headers) +} diff --git a/router/xgress/ordering_test.go b/router/xgress/ordering_test.go index 0c9194e2e..6c7f77fcf 100644 --- a/router/xgress/ordering_test.go +++ b/router/xgress/ordering_test.go @@ -2,6 +2,7 @@ package xgress import ( "encoding/binary" + "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/metrics" "github.com/openziti/foundation/metrics/metrics_pb" @@ -40,6 +41,10 @@ func (conn *testConn) WritePayload(bytes []byte, _ map[uint8][]byte) (int, error return len(bytes), nil } +func (conn *testConn) HandleControlMsg(ControlType, channel2.Headers, ControlReceiver) error { + return nil +} + type noopForwarder struct{} func (n noopForwarder) ForwardPayload(srcAddr Address, payload *Payload) error { @@ -52,8 +57,9 @@ func (n noopForwarder) ForwardAcknowledgement(srcAddr Address, acknowledgement * type noopReceiveHandler struct{} -func (n noopReceiveHandler) HandleXgressReceive(payload *Payload, x *Xgress) { -} +func (n noopReceiveHandler) HandleXgressReceive(payload *Payload, x *Xgress) {} + +func (n noopReceiveHandler) HandleControlReceive(*Control, *Xgress) {} func Test_Ordering(t *testing.T) { closeNotify := make(chan struct{}) diff --git a/router/xgress/xgress.go b/router/xgress/xgress.go index 49f1bcd54..a86b49e59 100644 --- a/router/xgress/xgress.go +++ b/router/xgress/xgress.go @@ -21,6 +21,7 @@ import ( "github.com/michaelquigley/pfxlog" "github.com/openziti/fabric/controller/xt" "github.com/openziti/fabric/logcontext" + "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/util/concurrenz" "github.com/openziti/foundation/util/info" @@ -67,6 +68,10 @@ type BindHandler interface { HandleXgressBind(x *Xgress) } +type ControlReceiver interface { + HandleControlReceive(controlType ControlType, headers channel2.Headers) +} + // ReceiveHandler is invoked by an xgress whenever data is received from the connected peer. Generally a ReceiveHandler // is implemented to connect the xgress to a data plane data transmission system. // @@ -74,6 +79,7 @@ type ReceiveHandler interface { // HandleXgressReceive is invoked when data is received from the connected xgress peer. // HandleXgressReceive(payload *Payload, x *Xgress) + HandleControlReceive(control *Control, x *Xgress) } // CloseHandler is invoked by an xgress when the connected peer terminates the communication. @@ -103,6 +109,7 @@ type Connection interface { LogContext() string ReadPayload() ([]byte, map[uint8][]byte, error) WritePayload([]byte, map[uint8][]byte) (int, error) + HandleControlMsg(controlType ControlType, headers channel2.Headers, responder ControlReceiver) error } type Xgress struct { @@ -327,6 +334,19 @@ func (self *Xgress) SendAcknowledgement(acknowledgement *Acknowledgement) error return nil } +func (self *Xgress) SendControl(control *Control) error { + return self.peer.HandleControlMsg(control.Type, control.Headers, self) +} + +func (self *Xgress) HandleControlReceive(controlType ControlType, headers channel2.Headers) { + control := &Control{ + Type: controlType, + CircuitId: self.circuitId, + Headers: headers, + } + self.receiveHandler.HandleControlReceive(control, self) +} + func (self *Xgress) payloadIngester(payload *Payload) { if payload.IsCircuitStartFlagSet() && self.firstCircuitStartReceived() { pfxlog.ContextLogger(self.Label()).WithFields(payload.GetLoggerFields()).Debug("received circuit start, starting xgress receiver") diff --git a/router/xgress_proxy/connection.go b/router/xgress_proxy/connection.go index 4a909383a..78d96e472 100644 --- a/router/xgress_proxy/connection.go +++ b/router/xgress_proxy/connection.go @@ -16,7 +16,12 @@ package xgress_proxy -import "github.com/openziti/foundation/transport" +import ( + "github.com/openziti/fabric/router/xgress" + "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/transport" + "github.com/pkg/errors" +) type proxyXgressConnection struct { transport.Connection @@ -39,3 +44,11 @@ func (c *proxyXgressConnection) Write(p []byte) (n int, err error) { func (c *proxyXgressConnection) WritePayload(p []byte, headers map[uint8][]byte) (n int, err error) { return c.Writer().Write(p) } + +func (self *proxyXgressConnection) HandleControlMsg(controlType xgress.ControlType, headers channel2.Headers, responder xgress.ControlReceiver) error { + if controlType == xgress.ControlTypeTraceRoute { + xgress.RespondToTraceRequest(headers, "xgress", "proxy", responder) + return nil + } + return errors.Errorf("unhandled control type: %v", controlType) +} diff --git a/router/xgress_transport/connection.go b/router/xgress_transport/connection.go index ffbea4a0e..46997d549 100644 --- a/router/xgress_transport/connection.go +++ b/router/xgress_transport/connection.go @@ -17,7 +17,10 @@ package xgress_transport import ( + "github.com/openziti/fabric/router/xgress" + "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/transport" + "github.com/pkg/errors" ) type transportXgressConn struct { @@ -44,3 +47,11 @@ func (c *transportXgressConn) ReadPayload() ([]byte, map[uint8][]byte, error) { func (c *transportXgressConn) WritePayload(p []byte, _ map[uint8][]byte) (n int, err error) { return c.Writer().Write(p) } + +func (self *transportXgressConn) HandleControlMsg(controlType xgress.ControlType, headers channel2.Headers, responder xgress.ControlReceiver) error { + if controlType == xgress.ControlTypeTraceRoute { + xgress.RespondToTraceRequest(headers, "xgress", "transport", responder) + return nil + } + return errors.Errorf("unhandled control type: %v", controlType) +} diff --git a/router/xgress_transport_udp/connection.go b/router/xgress_transport_udp/connection.go index 6aca79283..efee33bec 100644 --- a/router/xgress_transport_udp/connection.go +++ b/router/xgress_transport_udp/connection.go @@ -18,7 +18,9 @@ package xgress_transport_udp import ( "github.com/openziti/fabric/router/xgress" + "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/util/info" + "github.com/pkg/errors" "net" ) @@ -39,6 +41,14 @@ func (p *packetConn) WritePayload(data []byte, headers map[uint8][]byte) (n int, return p.Write(data) } +func (self *packetConn) HandleControlMsg(controlType xgress.ControlType, headers channel2.Headers, responder xgress.ControlReceiver) error { + if controlType == xgress.ControlTypeTraceRoute { + xgress.RespondToTraceRequest(headers, "xgress", "transport_udp", responder) + return nil + } + return errors.Errorf("unhandled control type: %v", controlType) +} + func newPacketConn(conn net.Conn) xgress.Connection { return &packetConn{conn} } diff --git a/router/xgress_udp/listener.go b/router/xgress_udp/listener.go index 22ceefdb2..4d7a48f26 100644 --- a/router/xgress_udp/listener.go +++ b/router/xgress_udp/listener.go @@ -17,6 +17,7 @@ package xgress_udp import ( + "github.com/openziti/fabric/router/xgress" "io" "net" ) @@ -31,15 +32,12 @@ type Listener interface { } type Session interface { + xgress.Connection State() SessionState SetState(state SessionState) Address() net.Addr - ReadPayload() ([]byte, map[uint8][]byte, error) Write(data []byte) (n int, err error) - WritePayload(data []byte, headers map[uint8][]byte) (n int, err error) QueueRead(data []byte) - Close() error - LogContext() string TimeoutNanos() int64 MarkActivity() SessionId() string diff --git a/router/xgress_udp/session.go b/router/xgress_udp/session.go index cfbfaa55a..b7c6ffb27 100644 --- a/router/xgress_udp/session.go +++ b/router/xgress_udp/session.go @@ -17,6 +17,9 @@ package xgress_udp import ( + "github.com/openziti/fabric/router/xgress" + "github.com/openziti/foundation/channel2" + "github.com/pkg/errors" "io" "net" "time" @@ -63,6 +66,14 @@ func (s *PacketSession) WritePayload(p []byte, _ map[uint8][]byte) (n int, err e return s.Write(p) } +func (s *PacketSession) HandleControlMsg(controlType xgress.ControlType, headers channel2.Headers, responder xgress.ControlReceiver) error { + if controlType == xgress.ControlTypeTraceRoute { + xgress.RespondToTraceRequest(headers, "xgress", "udp", responder) + return nil + } + return errors.Errorf("unhandled control type: %v", controlType) +} + func (s *PacketSession) QueueRead(data []byte) { s.readC <- data } diff --git a/router/xlink/xlink.go b/router/xlink/xlink.go index 061552b5e..aef907422 100644 --- a/router/xlink/xlink.go +++ b/router/xlink/xlink.go @@ -45,6 +45,7 @@ type Xlink interface { Id() *identity.TokenId SendPayload(payload *xgress.Payload) error SendAcknowledgement(acknowledgement *xgress.Acknowledgement) error + SendControl(control *xgress.Control) error Close() error DestinationId() string } @@ -52,4 +53,5 @@ type Xlink interface { type Forwarder interface { ForwardPayload(srcAddr xgress.Address, payload *xgress.Payload) error ForwardAcknowledgement(srcAddr xgress.Address, acknowledgement *xgress.Acknowledgement) error + ForwardControl(srcAddr xgress.Address, control *xgress.Control) error } diff --git a/router/xlink_transport/xlink.go b/router/xlink_transport/xlink.go index 0f3232eba..d8ec56df8 100644 --- a/router/xlink_transport/xlink.go +++ b/router/xlink_transport/xlink.go @@ -34,6 +34,10 @@ func (self *impl) SendAcknowledgement(acknowledgement *xgress.Acknowledgement) e return self.ch.Send(acknowledgement.Marshall()) } +func (self *impl) SendControl(control *xgress.Control) error { + return self.ch.Send(control.Marshall()) +} + func (self *impl) Close() error { return self.ch.Close() } diff --git a/router/xlink_transport/xlink_split.go b/router/xlink_transport/xlink_split.go index bb6ae17cb..98eb10496 100644 --- a/router/xlink_transport/xlink_split.go +++ b/router/xlink_transport/xlink_split.go @@ -31,6 +31,10 @@ func (self *splitImpl) SendPayload(payload *xgress.Payload) error { return self.payloadCh.Send(payload.Marshall()) } +func (self *splitImpl) SendControl(control *xgress.Control) error { + return self.payloadCh.Send(control.Marshall()) +} + func (self *splitImpl) SendAcknowledgement(acknowledgement *xgress.Acknowledgement) error { return self.ackCh.Send(acknowledgement.Marshall()) }