gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-12 23:32:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Drop messages on links that aren't keeping up. Fixes openziti/fabric#484

Browse Source
This commit is contained in:
Paul Lorenz
2022-09-06 23:57:12 -04:00
parent bf98f120f6
commit 52bbbb5a83
6 changed files with 93 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
events/metrics_mappers.go
+4 -1
View File
@@ -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)
router/xlink_transport/dialer.go
+13 -8
View File
@@ -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),
},
}
router/xlink_transport/factory.go
+11
View File
@@ -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
}
router/xlink_transport/listener.go
+10 -8
View File
@@ -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)
router/xlink_transport/xlink.go
+27 -13
View File
@@ -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 {
router/xlink_transport/xlink_split.go
+28 -14
View File
@@ -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 {