From cb70e33fa0f6bd3afb154cd0360cf9c1e0301f58 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Tue, 27 Jul 2021 12:53:41 -0400 Subject: [PATCH] Improve link logging. Add destination id to xlink --- controller/handler_ctrl/bind.go | 2 +- controller/handler_mgmt/bind.go | 2 +- router/handler_ctrl/bind.go | 2 +- router/handler_ctrl/dial.go | 45 ++++++++++--------- router/handler_link/accepter.go | 2 +- router/xlink/xlink.go | 1 + router/xlink_transport/dialer.go | 6 ++- router/xlink_transport/listener.go | 62 +++++++++++++++++---------- router/xlink_transport/xlink.go | 11 +++-- router/xlink_transport/xlink_split.go | 5 +++ trace/channel_peekhandler.go | 9 ++-- 11 files changed, 90 insertions(+), 57 deletions(-) diff --git a/controller/handler_ctrl/bind.go b/controller/handler_ctrl/bind.go index 0f0804647..d8ceff491 100644 --- a/controller/handler_ctrl/bind.go +++ b/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") { diff --git a/controller/handler_mgmt/bind.go b/controller/handler_mgmt/bind.go index d7ae2731f..332b02604 100644 --- a/controller/handler_mgmt/bind.go +++ b/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)) diff --git a/router/handler_ctrl/bind.go b/router/handler_ctrl/bind.go index 473e196f1..b69d481d1 100644 --- a/router/handler_ctrl/bind.go +++ b/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 { diff --git a/router/handler_ctrl/dial.go b/router/handler_ctrl/dial.go index 134c7fa1e..cbb85a6cb 100644 --- a/router/handler_ctrl/dial.go +++ b/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 } diff --git a/router/handler_link/accepter.go b/router/handler_link/accepter.go index 4ee158c58..e8951382f 100644 --- a/router/handler_link/accepter.go +++ b/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( diff --git a/router/xlink/xlink.go b/router/xlink/xlink.go index f32d0da9c..061552b5e 100644 --- a/router/xlink/xlink.go +++ b/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 { diff --git a/router/xlink_transport/dialer.go b/router/xlink_transport/dialer.go index 58ff0b14f..471b8c553 100644 --- a/router/xlink_transport/dialer.go +++ b/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 { diff --git a/router/xlink_transport/listener.go b/router/xlink_transport/listener.go index cd444c322..1e80a1351 100644 --- a/router/xlink_transport/listener.go +++ b/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") } diff --git a/router/xlink_transport/xlink.go b/router/xlink_transport/xlink.go index 864ccf852..0f3232eba 100644 --- a/router/xlink_transport/xlink.go +++ b/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 } diff --git a/router/xlink_transport/xlink_split.go b/router/xlink_transport/xlink_split.go index b1daf12df..bb6ae17cb 100644 --- a/router/xlink_transport/xlink_split.go +++ b/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 } diff --git a/trace/channel_peekhandler.go b/trace/channel_peekhandler.go index bef8e3785..407b4c5f0 100644 --- a/trace/channel_peekhandler.go +++ b/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,