From 52bbbb5a83028ca943e8af9b4e256a13638e67a1 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Tue, 6 Sep 2022 23:57:12 -0400 Subject: [PATCH] Drop messages on links that aren't keeping up. Fixes openziti/fabric#484 --- events/metrics_mappers.go | 5 +++- router/xlink_transport/dialer.go | 21 +++++++++----- router/xlink_transport/factory.go | 11 +++++++ router/xlink_transport/listener.go | 18 +++++++----- router/xlink_transport/xlink.go | 40 ++++++++++++++++--------- router/xlink_transport/xlink_split.go | 42 ++++++++++++++++++--------- 6 files changed, 93 insertions(+), 44 deletions(-) diff --git a/events/metrics_mappers.go b/events/metrics_mappers.go index 8deffaebd..dae5da936 100644 --- a/events/metrics_mappers.go +++ b/events/metrics_mappers.go @@ -40,7 +40,10 @@ type linkMetricsMapper struct { func (self *linkMetricsMapper) mapMetrics(_ *metrics_pb.MetricsMessage, event *event.MetricsEvent) { if strings.HasPrefix(event.Metric, "link.") { var name, linkId string - if strings.HasSuffix(event.Metric, "latency") || strings.HasSuffix(event.Metric, "queue_time") { + if parts := strings.Split(event.Metric, ":"); len(parts) == 2 { + name = parts[0] + linkId = parts[1] + } else if strings.HasSuffix(event.Metric, "latency") || strings.HasSuffix(event.Metric, "queue_time") { name, linkId = ExtractId(event.Metric, "link.", 1) } else { name, linkId = ExtractId(event.Metric, "link.", 2) diff --git a/router/xlink_transport/dialer.go b/router/xlink_transport/dialer.go index 030fb540c..d5a8279d1 100644 --- a/router/xlink_transport/dialer.go +++ b/router/xlink_transport/dialer.go @@ -21,6 +21,7 @@ import ( "github.com/openziti/channel" "github.com/openziti/fabric/router/xlink" "github.com/openziti/identity" + "github.com/openziti/metrics" "github.com/openziti/transport/v2" "github.com/pkg/errors" "github.com/sirupsen/logrus" @@ -32,6 +33,7 @@ type dialer struct { bindHandlerFactory BindHandlerFactory acceptor xlink.Acceptor transportConfig transport.Configuration + metricsRegistry metrics.Registry } func (self *dialer) Dial(dial xlink.Dial) (xlink.Xlink, error) { @@ -75,10 +77,12 @@ func (self *dialer) dialSplit(linkId *identity.TokenId, address transport.Addres bindHandler := &splitDialBindHandler{ dialer: self, - link: &splitImpl{id: linkId, - routerId: dial.GetRouterId(), - routerVersion: dial.GetRouterVersion(), - linkProtocol: dial.GetLinkProtocol(), + link: &splitImpl{ + id: linkId, + routerId: dial.GetRouterId(), + routerVersion: dial.GetRouterVersion(), + linkProtocol: dial.GetLinkProtocol(), + droppedMsgMeter: self.metricsRegistry.Meter("link.dropped_msgs:" + linkId.Token), }, } @@ -117,10 +121,11 @@ func (self *dialer) dialSingle(linkId *identity.TokenId, address transport.Addre bindHandler := &dialBindHandler{ dialer: self, link: &impl{ - id: linkId, - routerId: dial.GetRouterId(), - linkProtocol: dial.GetLinkProtocol(), - routerVersion: dial.GetRouterVersion(), + id: linkId, + routerId: dial.GetRouterId(), + linkProtocol: dial.GetLinkProtocol(), + routerVersion: dial.GetRouterVersion(), + droppedMsgMeter: self.metricsRegistry.Meter("link.dropped_msgs:" + linkId.Token), }, } diff --git a/router/xlink_transport/factory.go b/router/xlink_transport/factory.go index 0f8549f6a..eb6efcc03 100644 --- a/router/xlink_transport/factory.go +++ b/router/xlink_transport/factory.go @@ -18,6 +18,7 @@ package xlink_transport import ( "fmt" + "github.com/openziti/channel" "github.com/openziti/fabric/router/xlink" "github.com/openziti/identity" "github.com/openziti/metrics" @@ -82,12 +83,22 @@ func (self *factory) CreateDialer(id *identity.TokenId, _ xlink.Forwarder, confi if err != nil { return nil, fmt.Errorf("error loading dialer configuration (%w)", err) } + + if config.options == nil { + config.options = channel.DefaultOptions() + } + + if config.options.OutQueueSize == channel.DefaultOutQueueSize { + config.options.OutQueueSize = 64 + } + return &dialer{ id: id, config: config, acceptor: self.acceptor, bindHandlerFactory: self.bindHandlerFactory, transportConfig: self.transportConfig, + metricsRegistry: self.metricsRegistry, }, nil } diff --git a/router/xlink_transport/listener.go b/router/xlink_transport/listener.go index ffa92db32..cff8460d5 100644 --- a/router/xlink_transport/listener.go +++ b/router/xlink_transport/listener.go @@ -177,10 +177,11 @@ func (self *listener) getOrCreateSplitLink(id, routerId, routerVersion string, b } else { pending = &pendingLink{ link: &splitImpl{ - id: binding.GetChannel().Id(), - routerId: routerId, - routerVersion: routerVersion, - linkProtocol: self.GetLinkProtocol(), + id: binding.GetChannel().Id(), + routerId: routerId, + routerVersion: routerVersion, + linkProtocol: self.GetLinkProtocol(), + droppedMsgMeter: self.metricsRegistry.Meter("link.dropped_msgs:" + binding.GetChannel().Id().Token), }, eventTime: time.Now(), } @@ -209,10 +210,11 @@ func (self *listener) getOrCreateSplitLink(id, routerId, routerVersion string, b func (self *listener) bindNonSplitChannel(binding channel.Binding, routerId, routerVersion string, log *logrus.Entry) error { xli := &impl{ - id: binding.GetChannel().Id(), - ch: binding.GetChannel(), - routerId: routerId, - linkProtocol: self.GetLinkProtocol(), + id: binding.GetChannel().Id(), + ch: binding.GetChannel(), + routerId: routerId, + linkProtocol: self.GetLinkProtocol(), + droppedMsgMeter: self.metricsRegistry.Meter("link.dropped_msgs:" + binding.GetChannel().Id().Token), } bindHandler := self.bindHandlerFactory.NewBindHandler(xli, true, true) diff --git a/router/xlink_transport/xlink.go b/router/xlink_transport/xlink.go index 48cf62dce..179664cbc 100644 --- a/router/xlink_transport/xlink.go +++ b/router/xlink_transport/xlink.go @@ -23,32 +23,46 @@ import ( "github.com/openziti/fabric/router/xgress" "github.com/openziti/foundation/v2/concurrenz" "github.com/openziti/identity" + "github.com/openziti/metrics" ) type impl struct { - id *identity.TokenId - ch channel.Channel - routerId string - routerVersion string - linkProtocol string - dialAddress string - closeNotified concurrenz.AtomicBoolean + id *identity.TokenId + ch channel.Channel + routerId string + routerVersion string + linkProtocol string + dialAddress string + closeNotified concurrenz.AtomicBoolean + droppedMsgMeter metrics.Meter } func (self *impl) Id() *identity.TokenId { return self.id } -func (self *impl) SendPayload(payload *xgress.Payload) error { - return self.ch.Send(payload.Marshall()) +func (self *impl) SendPayload(msg *xgress.Payload) error { + sent, err := self.ch.TrySend(msg.Marshall()) + if err == nil && !sent { + self.droppedMsgMeter.Mark(1) + } + return err } -func (self *impl) SendAcknowledgement(acknowledgement *xgress.Acknowledgement) error { - return self.ch.Send(acknowledgement.Marshall()) +func (self *impl) SendAcknowledgement(msg *xgress.Acknowledgement) error { + sent, err := self.ch.TrySend(msg.Marshall()) + if err == nil && !sent { + self.droppedMsgMeter.Mark(1) + } + return err } -func (self *impl) SendControl(control *xgress.Control) error { - return self.ch.Send(control.Marshall()) +func (self *impl) SendControl(msg *xgress.Control) error { + sent, err := self.ch.TrySend(msg.Marshall()) + if err == nil && !sent { + self.droppedMsgMeter.Mark(1) + } + return err } func (self *impl) Close() error { diff --git a/router/xlink_transport/xlink_split.go b/router/xlink_transport/xlink_split.go index 22119295e..1171b5bd4 100644 --- a/router/xlink_transport/xlink_split.go +++ b/router/xlink_transport/xlink_split.go @@ -23,34 +23,48 @@ import ( "github.com/openziti/fabric/router/xgress" "github.com/openziti/foundation/v2/concurrenz" "github.com/openziti/identity" + "github.com/openziti/metrics" "github.com/pkg/errors" ) type splitImpl struct { - id *identity.TokenId - payloadCh channel.Channel - ackCh channel.Channel - routerId string - routerVersion string - linkProtocol string - dialAddress string - closeNotified concurrenz.AtomicBoolean + id *identity.TokenId + payloadCh channel.Channel + ackCh channel.Channel + routerId string + routerVersion string + linkProtocol string + dialAddress string + closeNotified concurrenz.AtomicBoolean + droppedMsgMeter metrics.Meter } func (self *splitImpl) Id() *identity.TokenId { return self.id } -func (self *splitImpl) SendPayload(payload *xgress.Payload) error { - return self.payloadCh.Send(payload.Marshall()) +func (self *splitImpl) SendPayload(msg *xgress.Payload) error { + sent, err := self.payloadCh.TrySend(msg.Marshall()) + if err == nil && !sent { + self.droppedMsgMeter.Mark(1) + } + return err } -func (self *splitImpl) SendControl(control *xgress.Control) error { - return self.payloadCh.Send(control.Marshall()) +func (self *splitImpl) SendAcknowledgement(msg *xgress.Acknowledgement) error { + sent, err := self.ackCh.TrySend(msg.Marshall()) + if err == nil && !sent { + self.droppedMsgMeter.Mark(1) + } + return err } -func (self *splitImpl) SendAcknowledgement(acknowledgement *xgress.Acknowledgement) error { - return self.ackCh.Send(acknowledgement.Marshall()) +func (self *splitImpl) SendControl(msg *xgress.Control) error { + sent, err := self.payloadCh.TrySend(msg.Marshall()) + if err == nil && !sent { + self.droppedMsgMeter.Mark(1) + } + return err } func (self *splitImpl) CloseNotified() error {