Add xgress control messages and support for doing route tracing

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)

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())))
 

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,
 	}
 }

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
+		}
+	}
+}

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,
 	}
 }

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),

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")
 	}
 }

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)
+}

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{})

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")

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)
+}

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)
+}

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}
 }

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

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
 }

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
 }

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()
 }

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())
 }