Merge pull request openziti/fabric#254 from openziti/reduce-link-growth

Improve link logging. Don't make multiple links to same router at sam…

controller/handler_ctrl/bind.go

@@ -51,7 +51,7 @@ func (self *bindHandler) BindChannel(ch channel2.Channel) error {
 	ch.AddReceiveHandler(newTraceHandler(traceDispatchWrapper))
 	ch.AddReceiveHandler(newInspectHandler(self.network))
 	ch.AddReceiveHandler(newPingHandler())
-	ch.AddPeekHandler(trace.NewChannelPeekHandler(self.network.GetAppId(), ch, self.network.GetTraceController(), traceDispatchWrapper))
+	ch.AddPeekHandler(trace.NewChannelPeekHandler(self.network.GetAppId().Token, ch, self.network.GetTraceController(), traceDispatchWrapper))
 	ch.AddPeekHandler(metrics2.NewCtrlChannelPeekHandler(self.router.Id, self.network.GetMetricsRegistry()))
 
 	if self.router.VersionInfo.HasMinimumVersion("0.18.7") {

controller/handler_mgmt/bind.go

@@ -69,7 +69,7 @@ func (bindHandler *BindHandler) BindChannel(ch channel2.Channel) error {
 	ch.AddReceiveHandler(newTogglePipeTracesHandler(network))
 
 	traceDispatchWrapper := trace.NewDispatchWrapper(network.GetEventDispatcher().Dispatch)
-	ch.AddPeekHandler(trace.NewChannelPeekHandler(network.GetAppId(), ch, network.GetTraceController(), traceDispatchWrapper))
+	ch.AddPeekHandler(trace.NewChannelPeekHandler(network.GetAppId().Token, ch, network.GetTraceController(), traceDispatchWrapper))
 
 	ch.AddReceiveHandler(newSnapshotDbHandler(network))
 

router/handler_ctrl/bind.go

@@ -62,7 +62,7 @@ func (self *bindHandler) BindChannel(ch channel2.Channel) error {
 	ch.AddReceiveHandler(newUnrouteHandler(self.forwarder))
 	ch.AddReceiveHandler(newTraceHandler(self.id, self.forwarder.TraceController()))
 	ch.AddReceiveHandler(newInspectHandler(self.id))
-	ch.AddPeekHandler(trace.NewChannelPeekHandler(self.id, ch, self.forwarder.TraceController(), trace.NewChannelSink(ch)))
+	ch.AddPeekHandler(trace.NewChannelPeekHandler(self.id.Token, ch, self.forwarder.TraceController(), trace.NewChannelSink(ch)))
 	metrics.AddLatencyProbeResponder(ch)
 
 	for _, x := range self.xctrls {

router/handler_ctrl/dial.go

@@ -17,14 +17,15 @@
 package handler_ctrl
 
 import (
-	"fmt"
 	"github.com/golang/protobuf/proto"
+	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/fabric/pb/ctrl_pb"
 	"github.com/openziti/fabric/router/forwarder"
 	"github.com/openziti/fabric/router/xgress"
 	"github.com/openziti/fabric/router/xlink"
 	"github.com/openziti/foundation/channel2"
 	"github.com/openziti/foundation/identity/identity"
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
 
@@ -64,56 +65,58 @@ func (self *dialHandler) handle(msg *channel2.Message, _ channel2.Channel) {
 	logrus.Info("received link connect request")
 	dial := &ctrl_pb.Dial{}
 	if err := proto.Unmarshal(msg.Body, dial); err == nil {
+		log := pfxlog.ChannelLogger("link", "linkDialer").
+			WithFields(logrus.Fields{"linkId": dial.LinkId, "routerId": dial.RouterId, "address": dial.Address})
+
 		linkId := self.id.ShallowCloneWithNewToken(dial.LinkId)
 		if len(self.dialers) == 1 {
-			logrus.Infof("dialing link to [r/%s]", dial.RouterId)
+			log.Info("dialing link")
 			if err := self.dialers[0].Dial(dial.Address, linkId, self.id.Token); err == nil {
-				if err := self.sendLinkMessage(linkId); err != nil {
-					logrus.Errorf("error sending link message [l/%s] (%v)", linkId.Token, err)
+				if err := self.sendLinkMessage(dial.LinkId); err != nil {
+					log.WithError(err).Error("error sending link message ")
 				}
-				logrus.Infof("link [l/%s] established", linkId.Token)
-
+				log.Info("link established")
 			} else {
-				logrus.Errorf("link dialing failed [%s] (%v)", dial.Address, err)
-				if err := self.sendLinkFault(linkId); err != nil {
-					logrus.Errorf("error sending fault [l/%s] (%v)", linkId.Token, err)
+				log.WithError(err).Error("link dialing failed")
+				if err := self.sendLinkFault(dial.LinkId); err != nil {
+					log.WithError(err).Error("error sending fault")
 				}
 			}
 		} else {
-			logrus.Errorf("invalid Xlink dialers configuration")
-			if err := self.sendLinkFault(linkId); err != nil {
-				logrus.Errorf("error sending link fault [l/%s] (%v)", linkId.Token, err)
+			log.Errorf("invalid Xlink dialers configuration")
+			if err := self.sendLinkFault(dial.LinkId); err != nil {
+				log.WithError(err).Error("error sending link fault")
 			}
 		}
 	} else {
-		logrus.Errorf("error unmarshaling dial message (%v)", err)
+		logrus.WithError(err).Error("error unmarshaling dial message")
 	}
 }
 
-func (self *dialHandler) sendLinkMessage(linkId *identity.TokenId) error {
-	linkMsg := &ctrl_pb.Link{Id: linkId.Token}
+func (self *dialHandler) sendLinkMessage(linkId string) error {
+	linkMsg := &ctrl_pb.Link{Id: linkId}
 	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 {
-			return fmt.Errorf("error sending link message (%w)", err)
+			return errors.Wrap(err, "error sending link message")
 		}
 	} else {
-		return fmt.Errorf("error marshaling link message (%w)", err)
+		return errors.Wrap(err, "error marshaling link message")
 	}
 	return nil
 }
 
-func (self *dialHandler) sendLinkFault(linkId *identity.TokenId) error {
-	fault := &ctrl_pb.Fault{Subject: ctrl_pb.FaultSubject_LinkFault, Id: linkId.Token}
+func (self *dialHandler) sendLinkFault(linkId string) error {
+	fault := &ctrl_pb.Fault{Subject: ctrl_pb.FaultSubject_LinkFault, Id: linkId}
 	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 {
-			return fmt.Errorf("error sending fault (%w)", err)
+			return errors.Wrap(err, "error sending fault")
 		}
 	} else {
-		return fmt.Errorf("error marshaling fault (%w)", err)
+		return errors.Wrapf(err, "error marshaling fault")
 	}
 	return nil
 }

router/handler_link/accepter.go

@@ -31,7 +31,7 @@ func (self *channelAccepter) AcceptChannel(xlink xlink.Xlink, ch channel2.Channe
 	ch.AddReceiveHandler(newQueuingAckHandler(xlink, self.ctrl, self.forwarder, closeNotify))
 	ch.AddReceiveHandler(&channel2.LatencyHandler{})
 	ch.AddPeekHandler(metrics2.NewChannelPeekHandler(xlink.Id().Token, self.forwarder.MetricsRegistry()))
-	ch.AddPeekHandler(trace.NewChannelPeekHandler(xlink.Id(), ch, self.forwarder.TraceController(), trace.NewChannelSink(self.ctrl.Channel())))
+	ch.AddPeekHandler(trace.NewChannelPeekHandler(xlink.Id().Token, ch, self.forwarder.TraceController(), trace.NewChannelSink(self.ctrl.Channel())))
 
 	if trackLatency {
 		go metrics.ProbeLatency(

router/xlink/xlink.go

@@ -46,6 +46,7 @@ type Xlink interface {
 	SendPayload(payload *xgress.Payload) error
 	SendAcknowledgement(acknowledgement *xgress.Acknowledgement) error
 	Close() error
+	DestinationId() string
 }
 
 type Forwarder interface {

router/xlink_transport/dialer.go

@@ -110,7 +110,11 @@ func (self *dialer) dialSingle(linkId *identity.TokenId, address transport.Addre
 		return nil, errors.Wrapf(err, "dialing link [l/%s] for payload", linkId.Token)
 	}
 
-	xli := &impl{id: linkId, ch: payloadCh}
+	xli := &impl{
+		id:       linkId,
+		ch:       payloadCh,
+		routerId: routerId,
+	}
 
 	if self.chAccepter != nil {
 		if err := self.chAccepter.AcceptChannel(xli, payloadCh, true); err != nil {

router/xlink_transport/listener.go

@@ -59,11 +59,16 @@ func (self *listener) acceptLoop() {
 			continue
 		}
 
+		log := pfxlog.ChannelLogger("link", "linkListener").WithField("linkId", ch.Id().Token)
+
 		headers := ch.Underlay().Headers()
 		channelType := byte(0)
+		routerId := ""
 		if headers != nil {
 			if v, ok := headers[LinkHeaderRouterId]; ok {
-				logrus.Infof("accepting link from [r/%s]", string(v))
+				routerId = string(v)
+				log = log.WithField("routerId", routerId)
+				log.Info("accepting link")
 			}
 			if val, ok := headers[LinkHeaderType]; ok {
 				channelType = val[0]
@@ -72,37 +77,43 @@ func (self *listener) acceptLoop() {
 		if channelType != 0 {
 			id, ok := headers[LinkHeaderConnId]
 			if !ok {
-				logrus.Errorf("split conn %v received but missing connection id. closing", ch.Id())
+				log.Error("split conn received but missing connection id. closing")
 				_ = ch.Close()
 				continue
 			}
 
-			logrus.Infof("split conn of type %v for link [l/%s]", channelType, ch.Id().Token)
+			log.Infof("accepted %v part of split conn", channelType)
 
 			event := &newChannelEvent{
 				ch:          ch,
 				channelType: channelType,
 				id:          string(id),
 				eventTime:   time.Now(),
+				routerId:    routerId,
 			}
 			self.eventC <- event
 			continue
 		}
 
-		xlink := &impl{id: ch.Id(), ch: ch}
-		logrus.Infof("accepting link id [l/%s]", xlink.Id().Token)
+		xli := &impl{
+			id:       ch.Id(),
+			ch:       ch,
+			routerId: routerId,
+		}
+
+		log.Info("accepting link")
 
 		if self.chAccepter != nil {
-			if err := self.chAccepter.AcceptChannel(xlink, ch, true); err != nil {
-				logrus.Errorf("error accepting incoming channel (%v)", err)
+			if err := self.chAccepter.AcceptChannel(xli, ch, true); err != nil {
+				log.WithError(err).Error("error accepting incoming channel")
 			}
 		}
 
-		if err := self.accepter.Accept(xlink); err != nil {
-			logrus.Errorf("error accepting incoming Xlink (%v)", err)
+		if err := self.accepter.Accept(xli); err != nil {
+			log.WithError(err).Error("error accepting incoming Xlink")
 		}
 
-		logrus.Infof("accepted link [%s]", "l/"+ch.Id().Token)
+		log.Info("accepted link")
 	}
 }
 
@@ -144,9 +155,14 @@ type newChannelEvent struct {
 	channelType byte
 	id          string
 	eventTime   time.Time
+	routerId    string
 }
 
 func (event *newChannelEvent) handle(l *listener) {
+	log := pfxlog.ChannelLogger("link", "linkListener").
+		WithField("linkId", event.ch.Id().Token).
+		WithField("routerId", event.routerId)
+
 	partner, ok := l.pendingChannels[event.id]
 	if !ok {
 		l.pendingChannels[event.id] = event
@@ -171,34 +187,34 @@ func (event *newChannelEvent) handle(l *listener) {
 	}
 
 	if payloadCh == nil || ackCh == nil {
-		pfxlog.Logger().Errorf("got two link channels, but types aren't correct. %v %v", partner.channelType, event.channelType)
+		log.Errorf("got two link channels, but types aren't correct. %v %v", partner.channelType, event.channelType)
 		return
 	}
 
-	xlink := &splitImpl{
+	xli := &splitImpl{
 		id:        event.ch.Id(),
 		payloadCh: payloadCh,
 		ackCh:     ackCh,
+		routerId:  event.routerId,
 	}
 
-	logrus.Infof("accepting split link with id [l/%s]", xlink.Id().Token)
+	log.Info("accepting split link")
 
 	if l.chAccepter != nil {
-		if err := l.chAccepter.AcceptChannel(xlink, xlink.payloadCh, true); err != nil {
-			logrus.Errorf("error accepting incoming channel (%v)", err)
-			_ = xlink.Close()
+		if err := l.chAccepter.AcceptChannel(xli, xli.payloadCh, true); err != nil {
+			log.WithError(err).Error("error accepting incoming channel")
+			_ = xli.Close()
 		}
 
-		if err := l.chAccepter.AcceptChannel(xlink, xlink.ackCh, true); err != nil {
-			logrus.Errorf("error accepting incoming channel (%v)", err)
-			_ = xlink.Close()
+		if err := l.chAccepter.AcceptChannel(xli, xli.ackCh, true); err != nil {
+			log.WithError(err).Error("error accepting incoming channel")
+			_ = xli.Close()
 		}
 	}
 
-	if err := l.accepter.Accept(xlink); err != nil {
-		logrus.Errorf("error accepting incoming Xlink (%v)", err)
+	if err := l.accepter.Accept(xli); err != nil {
+		log.WithError(err).Error("error accepting incoming Xlink")
 	}
 
-	logrus.Infof("accepted link [%s]", "l/"+xlink.Id().Token)
-
+	log.Info("accepted link")
 }

router/xlink_transport/xlink.go

@@ -38,7 +38,12 @@ func (self *impl) Close() error {
 	return self.ch.Close()
 }
 
-type impl struct {
-	id *identity.TokenId
-	ch channel2.Channel
+func (self *impl) DestinationId() string {
+	return self.routerId
+}
+
+type impl struct {
+	id       *identity.TokenId
+	ch       channel2.Channel
+	routerId string
 }

router/xlink_transport/xlink_split.go

@@ -47,8 +47,13 @@ func (self *splitImpl) Close() error {
 	return errors.Errorf("multiple failures while closing transport link (%v) (%v)", err, err2)
 }
 
+func (self *splitImpl) DestinationId() string {
+	return self.routerId
+}
+
 type splitImpl struct {
 	id        *identity.TokenId
 	payloadCh channel2.Channel
 	ackCh     channel2.Channel
+	routerId  string
 }

trace/channel_peekhandler.go

@@ -24,7 +24,6 @@ import (
 	"github.com/openziti/fabric/pb/mgmt_pb"
 	"github.com/openziti/fabric/router/xgress"
 	"github.com/openziti/foundation/channel2"
-	"github.com/openziti/foundation/identity/identity"
 	trace_pb "github.com/openziti/foundation/trace/pb"
 	"sync/atomic"
 	"time"
@@ -33,7 +32,7 @@ import (
 var decoders = []channel2.TraceMessageDecoder{&channel2.Decoder{}, &ctrl_pb.Decoder{}, &xgress.Decoder{}, &mgmt_pb.Decoder{}}
 
 type ChannelPeekHandler struct {
-	appId      *identity.TokenId
+	appId      string
 	ch         channel2.Channel
 	enabled    int32
 	controller Controller
@@ -60,10 +59,10 @@ func (handler *ChannelPeekHandler) ToggleTracing(sourceType SourceType, matcher
 	}
 	resultChan <- &ToggleApplyResultImpl{matched,
 		fmt.Sprintf("Link %v.%v matched? %v. Old trace state: %v, New trace state: %v",
-			handler.appId.Token, name, matched, prevState, nextState)}
+			handler.appId, name, matched, prevState, nextState)}
 }
 
-func NewChannelPeekHandler(appId *identity.TokenId, ch channel2.Channel, controller Controller, eventSink EventHandler) *ChannelPeekHandler {
+func NewChannelPeekHandler(appId string, ch channel2.Channel, controller Controller, eventSink EventHandler) *ChannelPeekHandler {
 	handler := &ChannelPeekHandler{
 		appId:      appId,
 		ch:         ch,
@@ -122,7 +121,7 @@ func (handler *ChannelPeekHandler) trace(msg *channel2.Message, ch channel2.Chan
 
 	traceMsg := &trace_pb.ChannelMessage{
 		Timestamp:   time.Now().UnixNano(),
-		Identity:    handler.appId.Token,
+		Identity:    handler.appId,
 		Channel:     ch.LogicalName(),
 		IsRx:        rx,
 		ContentType: msg.ContentType,