mirror of
https://github.com/openziti/ziti.git
synced 2026-10-12 23:32:02 +00:00
Merge pull request #1587 from openziti/fix-early-ack-stall
Fix stall cause by ack arriving before payload is received by link send buffer. Fixes #1586
This commit is contained in:
Paul Lorenz
@@ -1,3 +1,20 @@
|
||||
# Release 0.31.4
|
||||
|
||||
## What's New
|
||||
|
||||
* Bug fix for a data flow stall which is especially likely to happen on circuits with single router paths
|
||||
|
||||
## Thanks
|
||||
|
||||
* @marvkis - for providing high quality debug data which made tracking down a couple of flow control stall issues much easier
|
||||
|
||||
## Component Updates and Bug Fixes
|
||||
|
||||
* github.com/openziti/metrics: [v1.2.40 -> v1.2.41](https://github.com/openziti/metrics/compare/v1.2.40...v1.2.41)
|
||||
* github.com/openziti/storage: [v0.2.26 -> v0.2.27](https://github.com/openziti/storage/compare/v0.2.26...v0.2.27)
|
||||
* github.com/openziti/ziti: [v0.31.3 -> v0.31.4](https://github.com/openziti/ziti/compare/v0.31.3...v0.31.4)
|
||||
* [Issue #1586](https://github.com/openziti/ziti/issues/1586) - If ack is received before payload is processed by link send buffer, a stall can result
|
||||
|
||||
# Release 0.31.3
|
||||
|
||||
## What's New
|
||||
|
||||
@@ -28,11 +28,11 @@ require (
|
||||
github.com/golang-jwt/jwt/v5 v5.0.0
|
||||
github.com/google/go-cmp v0.6.0
|
||||
github.com/google/gopacket v1.1.19
|
||||
github.com/google/uuid v1.4.0
|
||||
github.com/google/uuid v1.5.0
|
||||
github.com/gorilla/handlers v1.5.2
|
||||
github.com/gorilla/mux v1.8.1
|
||||
github.com/gorilla/websocket v1.5.1
|
||||
github.com/hashicorp/go-hclog v1.6.1
|
||||
github.com/hashicorp/go-hclog v1.6.2
|
||||
github.com/hashicorp/golang-lru/v2 v2.0.7
|
||||
github.com/hashicorp/raft v1.6.0
|
||||
github.com/hashicorp/raft-boltdb v0.0.0-20220329195025-15018e9b97e0
|
||||
@@ -52,11 +52,11 @@ require (
|
||||
github.com/openziti/foundation/v2 v2.0.35
|
||||
github.com/openziti/identity v1.0.68
|
||||
github.com/openziti/jwks v1.0.3
|
||||
github.com/openziti/metrics v1.2.40
|
||||
github.com/openziti/metrics v1.2.41
|
||||
github.com/openziti/runzmd v1.0.36
|
||||
github.com/openziti/sdk-golang v0.21.2
|
||||
github.com/openziti/secretstream v0.1.14
|
||||
github.com/openziti/storage v0.2.26
|
||||
github.com/openziti/storage v0.2.27
|
||||
github.com/openziti/transport/v2 v2.0.119
|
||||
github.com/openziti/x509-claims v1.0.3
|
||||
github.com/openziti/xweb/v2 v2.1.0
|
||||
|
||||
@@ -337,8 +337,8 @@ github.com/google/pprof v0.0.0-20210226084205-cbba55b83ad5/go.mod h1:kpwsk12EmLe
|
||||
github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
|
||||
github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4=
|
||||
github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
|
||||
github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/googleapis/gax-go v2.0.0+incompatible/go.mod h1:SFVmujtThgffbyetf+mdk2eWhX2bMyUtNHzFKcPA9HY=
|
||||
github.com/googleapis/gax-go/v2 v2.0.3/go.mod h1:LLvjysVCY1JZeum8Z6l8qUty8fiNwE08qbEPm1M08qg=
|
||||
github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
|
||||
@@ -367,8 +367,8 @@ github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brv
|
||||
github.com/hashicorp/go-cleanhttp v0.5.0/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
|
||||
github.com/hashicorp/go-cleanhttp v0.5.1/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
|
||||
github.com/hashicorp/go-hclog v0.9.1/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
|
||||
github.com/hashicorp/go-hclog v1.6.1 h1:pa92nu9bPoAqI7p+uPDCIWGAibUdlCi6TYWJEQQkLf8=
|
||||
github.com/hashicorp/go-hclog v1.6.1/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M=
|
||||
github.com/hashicorp/go-hclog v1.6.2 h1:NOtoftovWkDheyUM/8JW3QMiXyxJK3uHRK7wV04nD2I=
|
||||
github.com/hashicorp/go-hclog v1.6.2/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M=
|
||||
github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
|
||||
github.com/hashicorp/go-immutable-radix v1.3.1 h1:DKHmCUm2hRBK510BaiZlwvpD40f8bJFeZnpfm2KLowc=
|
||||
github.com/hashicorp/go-immutable-radix v1.3.1/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
|
||||
@@ -592,16 +592,16 @@ github.com/openziti/identity v1.0.68 h1:SaFr7BeFQBoWQDiT28vUb8D9w7v6lIAK6/9RkwmV
|
||||
github.com/openziti/identity v1.0.68/go.mod h1:HbOu3TQ032v8xE6xZWjO51azF4fUxRLjO/l/oGqJwUI=
|
||||
github.com/openziti/jwks v1.0.3 h1:hf8wkb+Cg4nH/HM0KROFd7u+C3DkRVcFZJ7tDV+4icc=
|
||||
github.com/openziti/jwks v1.0.3/go.mod h1:t4xxq8vlXGsPn29kiQVnZBBDDnEoOFqtJoHibkJunQQ=
|
||||
github.com/openziti/metrics v1.2.40 h1:gySRgR8prCPqaEjmUtX0eXFs7NkI9uPAzp+z6A8+JqA=
|
||||
github.com/openziti/metrics v1.2.40/go.mod h1:HXdVryf3xpZfnY4VcaOjMxiBv+qw0wJlEJNLbooB9hY=
|
||||
github.com/openziti/metrics v1.2.41 h1:JShcFb6qJPA2cMiWQLtcSXiJjsrhEWpH+aVcjT/Mcbs=
|
||||
github.com/openziti/metrics v1.2.41/go.mod h1:L9h0NrliMA3+p7+ascKgvx28qoKHymN9l+CMA+Q+sZc=
|
||||
github.com/openziti/runzmd v1.0.36 h1:HOqTZFzTTFu52qmCAQfFvKDmCSl8ZqP1PQQ0UnJIA4E=
|
||||
github.com/openziti/runzmd v1.0.36/go.mod h1:jYqummjskmFh63htJFF2SrUuvxNQifqd5REUhYVaY/A=
|
||||
github.com/openziti/sdk-golang v0.21.2 h1:P66cslOAmQX37VFan+df+MoD2PqaFjHWDNMpuhhXHSo=
|
||||
github.com/openziti/sdk-golang v0.21.2/go.mod h1:mepEUD39DsBm/v1WVLedYRoYCFdet5mmJ5Sxqm/zkFI=
|
||||
github.com/openziti/secretstream v0.1.14 h1:Ta+nB5Prcct+L5LIKUA1nE56QhWS6lMPQYTlpxUltU0=
|
||||
github.com/openziti/secretstream v0.1.14/go.mod h1:/hhuLfu+GIv0+cnapfsu/VOnXEvmTt3GKtCu+lQ0RIw=
|
||||
github.com/openziti/storage v0.2.26 h1:15EbOC6A//dsdLSs/RYJP6Qn3Rj6Od4btXEWGezatxc=
|
||||
github.com/openziti/storage v0.2.26/go.mod h1:pDCkPIN7h9L+FyJP+hcfOwk+GofzrrNoYsAThOjNza0=
|
||||
github.com/openziti/storage v0.2.27 h1:WdFD0KxXZxSoWOaojdi5r1LO0BTvn4x/7wwdwhRPssc=
|
||||
github.com/openziti/storage v0.2.27/go.mod h1:p/04So5E3NT2jsGew0AS4hYTfP5srti3VytOjlVcx0M=
|
||||
github.com/openziti/transport/v2 v2.0.119 h1:KOgHU+9EZUVPvv8ncifqHmNEcFUHbJHigo3jyPvWnOc=
|
||||
github.com/openziti/transport/v2 v2.0.119/go.mod h1:H2IIBP6ed9isE/eJHGXtAZL0d73ApYOpLG9sSvutNNI=
|
||||
github.com/openziti/x509-claims v1.0.3 h1:HNdQ8Nf1agB3lBs1gahcO6zfkeS4S5xoQ2/PkY4HRX0=
|
||||
|
||||
@@ -35,6 +35,11 @@ type Faulter struct {
|
||||
closeNotify chan struct{}
|
||||
}
|
||||
|
||||
type FaultReceiver interface {
|
||||
Report(circuitId string, ctrlId string)
|
||||
NotifyInvalidLink(linkId string)
|
||||
}
|
||||
|
||||
func NewFaulter(ctrls env.NetworkControllers, interval time.Duration, closeNotify chan struct{}) *Faulter {
|
||||
f := &Faulter{
|
||||
ctrls: ctrls,
|
||||
@@ -50,13 +55,13 @@ func NewFaulter(ctrls env.NetworkControllers, interval time.Duration, closeNotif
|
||||
return f
|
||||
}
|
||||
|
||||
func (self *Faulter) report(circuitId string, ctrlId string) {
|
||||
func (self *Faulter) Report(circuitId string, ctrlId string) {
|
||||
if self.interval > 0 {
|
||||
self.circuitIds.Set(circuitId, ctrlId)
|
||||
}
|
||||
}
|
||||
|
||||
func (self *Faulter) notifyInvalidLink(linkId string) {
|
||||
func (self *Faulter) NotifyInvalidLink(linkId string) {
|
||||
log := pfxlog.Logger()
|
||||
self.ctrls.ForEach(func(ctrlId string, ch channel.Channel) {
|
||||
fault := &ctrl_pb.Fault{Subject: ctrl_pb.FaultSubject_LinkFault, Id: linkId}
|
||||
|
||||
@@ -24,6 +24,7 @@ import (
|
||||
"github.com/openziti/ziti/common/inspect"
|
||||
"github.com/openziti/ziti/common/pb/ctrl_pb"
|
||||
"github.com/openziti/ziti/common/trace"
|
||||
"github.com/openziti/ziti/router/env"
|
||||
"github.com/openziti/ziti/router/xgress"
|
||||
"github.com/openziti/ziti/router/xlink"
|
||||
"github.com/pkg/errors"
|
||||
@@ -34,8 +35,7 @@ import (
|
||||
type Forwarder struct {
|
||||
circuits *circuitTable
|
||||
destinations *destinationTable
|
||||
faulter *Faulter
|
||||
scanner *Scanner
|
||||
faulter FaultReceiver
|
||||
metricsRegistry metrics.UsageRegistry
|
||||
traceController trace.Controller
|
||||
Options *Options
|
||||
@@ -58,21 +58,30 @@ type XgressDestination interface {
|
||||
GetTimeOfLastRxFromLink() int64
|
||||
}
|
||||
|
||||
func NewForwarder(metricsRegistry metrics.UsageRegistry, faulter *Faulter, scanner *Scanner, options *Options, closeNotify <-chan struct{}) *Forwarder {
|
||||
func NewForwarder(metricsRegistry metrics.UsageRegistry, faulter FaultReceiver, options *Options, closeNotify <-chan struct{}) *Forwarder {
|
||||
f := &Forwarder{
|
||||
circuits: newCircuitTable(),
|
||||
destinations: newDestinationTable(),
|
||||
faulter: faulter,
|
||||
scanner: scanner,
|
||||
metricsRegistry: metricsRegistry,
|
||||
traceController: trace.NewController(closeNotify),
|
||||
Options: options,
|
||||
CloseNotify: closeNotify,
|
||||
}
|
||||
f.scanner.setCircuitTable(f.circuits)
|
||||
return f
|
||||
}
|
||||
|
||||
func (forwarder *Forwarder) StartScanner(ctrls env.NetworkControllers) {
|
||||
scanner := newScanner(ctrls, forwarder.Options, forwarder.CloseNotify)
|
||||
scanner.setCircuitTable(forwarder.circuits)
|
||||
|
||||
if scanner.interval > 0 {
|
||||
go scanner.run()
|
||||
} else {
|
||||
logrus.Warnf("scanner disabled")
|
||||
}
|
||||
}
|
||||
|
||||
func (forwarder *Forwarder) MetricsRegistry() metrics.UsageRegistry {
|
||||
return forwarder.metricsRegistry
|
||||
}
|
||||
@@ -129,7 +138,7 @@ func (forwarder *Forwarder) Route(ctrlId string, route *ctrl_pb.Route) error {
|
||||
for _, forward := range route.Forwards {
|
||||
if !forwarder.HasDestination(xgress.Address(forward.DstAddress)) {
|
||||
if forward.DstType == ctrl_pb.DestType_Link {
|
||||
forwarder.faulter.notifyInvalidLink(forward.DstAddress)
|
||||
forwarder.faulter.NotifyInvalidLink(forward.DstAddress)
|
||||
return errors.Errorf("invalid link destination %v", forward.DstAddress)
|
||||
}
|
||||
if forward.DstType == ctrl_pb.DestType_End {
|
||||
@@ -258,7 +267,7 @@ func (forwarder *Forwarder) ForwardControl(srcAddr xgress.Address, control *xgre
|
||||
|
||||
func (forwarder *Forwarder) ReportForwardingFault(circuitId string, ctrlId string) {
|
||||
if forwarder.faulter != nil {
|
||||
forwarder.faulter.report(circuitId, ctrlId)
|
||||
forwarder.faulter.Report(circuitId, ctrlId)
|
||||
} else {
|
||||
logrus.Error("nil faulter, cannot accept forwarding fault report")
|
||||
}
|
||||
|
||||
@@ -34,18 +34,13 @@ type Scanner struct {
|
||||
closeNotify <-chan struct{}
|
||||
}
|
||||
|
||||
func NewScanner(ctrls env.NetworkControllers, options *Options, closeNotify <-chan struct{}) *Scanner {
|
||||
func newScanner(ctrls env.NetworkControllers, options *Options, closeNotify <-chan struct{}) *Scanner {
|
||||
s := &Scanner{
|
||||
ctrls: ctrls,
|
||||
interval: options.IdleTxInterval,
|
||||
timeout: options.IdleCircuitTimeout,
|
||||
closeNotify: closeNotify,
|
||||
}
|
||||
if s.interval > 0 {
|
||||
go s.run()
|
||||
} else {
|
||||
logrus.Warnf("scanner disabled")
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
|
||||
+2
-3
@@ -72,7 +72,6 @@ type Router struct {
|
||||
ctrls env.NetworkControllers
|
||||
ctrlBindhandler channel.BindHandler
|
||||
faulter *forwarder.Faulter
|
||||
scanner *forwarder.Scanner
|
||||
forwarder *forwarder.Forwarder
|
||||
xrctrls []env.Xrctrl
|
||||
xlinkFactories map[string]xlink.Factory
|
||||
@@ -196,8 +195,8 @@ func Create(config *Config, versionProvider versions.VersionProvider) *Router {
|
||||
router.ctrls = env.NewNetworkControllers(config.Ctrl.DefaultRequestTimeout, router.connectToController, &config.Ctrl.Heartbeats)
|
||||
router.xlinkRegistry = link.NewLinkRegistry(router)
|
||||
router.faulter = forwarder.NewFaulter(router.ctrls, config.Forwarder.FaultTxInterval, closeNotify)
|
||||
router.scanner = forwarder.NewScanner(router.ctrls, config.Forwarder, closeNotify)
|
||||
router.forwarder = forwarder.NewForwarder(metricsRegistry, router.faulter, router.scanner, config.Forwarder, closeNotify)
|
||||
router.forwarder = forwarder.NewForwarder(metricsRegistry, router.faulter, config.Forwarder, closeNotify)
|
||||
router.forwarder.StartScanner(router.ctrls)
|
||||
|
||||
xgress.InitPayloadIngester(closeNotify)
|
||||
xgress.InitAcker(router.forwarder, metricsRegistry, closeNotify)
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
//go:build perf
|
||||
|
||||
/*
|
||||
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 test
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/openziti/channel/v2"
|
||||
"github.com/openziti/metrics"
|
||||
"github.com/openziti/metrics/metrics_pb"
|
||||
"github.com/openziti/ziti/common/pb/ctrl_pb"
|
||||
"github.com/openziti/ziti/router/forwarder"
|
||||
"github.com/openziti/ziti/router/handler_xgress"
|
||||
"github.com/openziti/ziti/router/xgress"
|
||||
"github.com/stretchr/testify/require"
|
||||
"io"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type testSrcConn struct {
|
||||
sendCount atomic.Uint32
|
||||
}
|
||||
|
||||
func (self *testSrcConn) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (self *testSrcConn) LogContext() string {
|
||||
return "test"
|
||||
}
|
||||
|
||||
func (self *testSrcConn) ReadPayload() ([]byte, map[uint8][]byte, error) {
|
||||
if self.sendCount.Load() > 1000 {
|
||||
return nil, nil, io.EOF
|
||||
}
|
||||
|
||||
count := self.sendCount.Add(1)
|
||||
now := time.Now().UnixMilli()
|
||||
|
||||
buf := make([]byte, 10240)
|
||||
binary.BigEndian.PutUint32(buf, count)
|
||||
binary.BigEndian.PutUint64(buf[4:], uint64(now))
|
||||
|
||||
return buf, nil, nil
|
||||
}
|
||||
|
||||
func (self *testSrcConn) WritePayload(bytes []byte, m map[uint8][]byte) (int, error) {
|
||||
return len(bytes), nil
|
||||
}
|
||||
|
||||
func (self *testSrcConn) HandleControlMsg(xgress.ControlType, channel.Headers, xgress.ControlReceiver) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
type testDstConn struct {
|
||||
closeNotify chan struct{}
|
||||
recvCount atomic.Uint32
|
||||
done atomic.Bool
|
||||
notifyDone chan struct{}
|
||||
}
|
||||
|
||||
func (self *testDstConn) waitForDone(timeout time.Duration) error {
|
||||
select {
|
||||
case <-self.notifyDone:
|
||||
return nil
|
||||
case <-time.After(timeout):
|
||||
return errors.New("timed out")
|
||||
}
|
||||
}
|
||||
|
||||
func (self *testDstConn) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (self *testDstConn) LogContext() string {
|
||||
return "test"
|
||||
}
|
||||
|
||||
func (self *testDstConn) ReadPayload() ([]byte, map[uint8][]byte, error) {
|
||||
<-self.closeNotify
|
||||
return nil, nil, io.EOF
|
||||
}
|
||||
|
||||
func (self *testDstConn) WritePayload(bytes []byte, m map[uint8][]byte) (int, error) {
|
||||
count := binary.BigEndian.Uint32(bytes)
|
||||
start := binary.BigEndian.Uint64(bytes[4:])
|
||||
val := self.recvCount.Add(1)
|
||||
fmt.Printf("%v/%v: %v\n", val, count, time.Now().UnixMilli()-int64(start))
|
||||
if val > 1000 {
|
||||
if self.done.CompareAndSwap(false, true) {
|
||||
close(self.notifyDone)
|
||||
}
|
||||
}
|
||||
return len(bytes), nil
|
||||
}
|
||||
|
||||
func (self *testDstConn) HandleControlMsg(xgress.ControlType, channel.Headers, xgress.ControlReceiver) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
type testFaultReceiver struct{}
|
||||
|
||||
func (t testFaultReceiver) Report(circuitId string, ctrlId string) {}
|
||||
|
||||
func (t testFaultReceiver) NotifyInvalidLink(linkId string) {}
|
||||
|
||||
type testXgCloseHandler struct{}
|
||||
|
||||
func (t testXgCloseHandler) HandleXgressClose(x *xgress.Xgress) {
|
||||
}
|
||||
|
||||
type eventSink struct{}
|
||||
|
||||
func (e eventSink) AcceptMetrics(message *metrics_pb.MetricsMessage) {
|
||||
}
|
||||
|
||||
func Test_SingleRouterPerf(t *testing.T) {
|
||||
closeNotify := make(chan struct{})
|
||||
defer close(closeNotify)
|
||||
|
||||
options := xgress.DefaultOptions()
|
||||
srcConn := &testSrcConn{}
|
||||
dstConn := &testDstConn{
|
||||
closeNotify: closeNotify,
|
||||
notifyDone: make(chan struct{}),
|
||||
}
|
||||
|
||||
registry := metrics.NewUsageRegistry("router", nil, closeNotify)
|
||||
registry.StartReporting(&eventSink{}, time.Minute, 10)
|
||||
fwd := forwarder.NewForwarder(registry, testFaultReceiver{}, forwarder.DefaultOptions(), closeNotify)
|
||||
|
||||
xgress.InitPayloadIngester(closeNotify)
|
||||
xgress.InitAcker(fwd, registry, closeNotify)
|
||||
xgress.InitMetrics(registry)
|
||||
xgress.InitRetransmitter(fwd, fwd, registry, closeNotify)
|
||||
|
||||
bindHandler := handler_xgress.NewBindHandler(handler_xgress.NewReceiveHandler(fwd), testXgCloseHandler{}, fwd)
|
||||
|
||||
srcXg := xgress.NewXgress("test", "ctrl", "src", srcConn, xgress.Initiator, options, nil)
|
||||
bindHandler.HandleXgressBind(srcXg)
|
||||
|
||||
dstXg := xgress.NewXgress("test", "ctrl", "dst", dstConn, xgress.Terminator, options, nil)
|
||||
bindHandler.HandleXgressBind(dstXg)
|
||||
|
||||
req := require.New(t)
|
||||
err := fwd.Route("ctrl", &ctrl_pb.Route{
|
||||
CircuitId: "test",
|
||||
Attempt: 0,
|
||||
Forwards: []*ctrl_pb.Route_Forward{
|
||||
{
|
||||
SrcAddress: "src",
|
||||
DstAddress: "dst",
|
||||
DstType: ctrl_pb.DestType_End,
|
||||
},
|
||||
{
|
||||
SrcAddress: "dst",
|
||||
DstAddress: "src",
|
||||
DstType: ctrl_pb.DestType_Start,
|
||||
},
|
||||
},
|
||||
})
|
||||
req.NoError(err)
|
||||
dstXg.Start()
|
||||
srcXg.Start()
|
||||
|
||||
err = dstConn.waitForDone(2 * time.Second)
|
||||
req.NoError(err)
|
||||
}
|
||||
@@ -101,10 +101,12 @@ func (self *txPayload) isRetransmittable() bool {
|
||||
func NewLinkSendBuffer(x *Xgress) *LinkSendBuffer {
|
||||
logrus.Debugf("txPortalStartSize = %d", x.Options.TxPortalStartSize)
|
||||
|
||||
// newlyBuffered should be size 0, otherwise payloads can be sent and acks received before the payload is
|
||||
// processed by the LinkSendBuffer
|
||||
buffer := &LinkSendBuffer{
|
||||
x: x,
|
||||
buffer: make(map[int32]*txPayload),
|
||||
newlyBuffered: make(chan *txPayload, x.Options.TxQueueSize),
|
||||
newlyBuffered: make(chan *txPayload),
|
||||
newlyReceivedAcks: make(chan *Acknowledgement, 2),
|
||||
closeNotify: make(chan struct{}),
|
||||
windowsSize: x.Options.TxPortalStartSize,
|
||||
@@ -139,7 +141,12 @@ func (buffer *LinkSendBuffer) ReceiveAcknowledgement(ack *Acknowledgement) {
|
||||
case buffer.newlyReceivedAcks <- ack:
|
||||
log.Debug("ack processed")
|
||||
case <-buffer.closeNotify:
|
||||
log.Error("payload buffer closed")
|
||||
// if end of circuit was received, we've cleanly shutdown and can ignore any trailing acks
|
||||
if buffer.x.IsEndOfCircuitReceived() {
|
||||
log.Debug("payload buffer closed")
|
||||
} else {
|
||||
log.Error("payload buffer closed")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -4,14 +4,14 @@ import (
|
||||
"crypto/x509"
|
||||
"github.com/michaelquigley/pfxlog"
|
||||
"github.com/openziti/channel/v2"
|
||||
"github.com/openziti/metrics"
|
||||
"github.com/openziti/sdk-golang/ziti/edge"
|
||||
"github.com/openziti/ziti/common/inspect"
|
||||
"github.com/openziti/ziti/common/pb/ctrl_pb"
|
||||
"github.com/openziti/ziti/router/forwarder"
|
||||
"github.com/openziti/ziti/router/handler_xgress"
|
||||
metrics2 "github.com/openziti/ziti/router/metrics"
|
||||
"github.com/openziti/ziti/router/xgress"
|
||||
"github.com/openziti/metrics"
|
||||
"github.com/openziti/sdk-golang/ziti/edge"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"testing"
|
||||
@@ -139,7 +139,7 @@ func writePerf(b *testing.B, mux edge.MsgMux) {
|
||||
xgress.InitMetrics(metricsRegistry)
|
||||
|
||||
fwdOptions := forwarder.DefaultOptions()
|
||||
fwd := forwarder.NewForwarder(metricsRegistry, nil, nil, fwdOptions, nil)
|
||||
fwd := forwarder.NewForwarder(metricsRegistry, nil, fwdOptions, nil)
|
||||
|
||||
link := newMirrorLink(fwd)
|
||||
|
||||
@@ -216,7 +216,7 @@ func Benchmark_BaselinePerf(b *testing.B) {
|
||||
xgress.InitMetrics(metricsRegistry)
|
||||
|
||||
fwdOptions := forwarder.DefaultOptions()
|
||||
fwd := forwarder.NewForwarder(metricsRegistry, nil, nil, fwdOptions, nil)
|
||||
fwd := forwarder.NewForwarder(metricsRegistry, nil, fwdOptions, nil)
|
||||
|
||||
link := newMirrorLink(fwd)
|
||||
|
||||
|
||||
+4
-4
@@ -8,7 +8,7 @@ require (
|
||||
github.com/Jeffail/gabs v1.4.0
|
||||
github.com/Jeffail/gabs/v2 v2.7.0
|
||||
github.com/google/go-cmp v0.6.0
|
||||
github.com/google/uuid v1.4.0
|
||||
github.com/google/uuid v1.5.0
|
||||
github.com/michaelquigley/pfxlog v0.6.10
|
||||
github.com/openziti/agent v1.0.16
|
||||
github.com/openziti/channel/v2 v2.0.111
|
||||
@@ -16,7 +16,7 @@ require (
|
||||
github.com/openziti/foundation/v2 v2.0.35
|
||||
github.com/openziti/identity v1.0.68
|
||||
github.com/openziti/sdk-golang v0.21.2
|
||||
github.com/openziti/storage v0.2.26
|
||||
github.com/openziti/storage v0.2.27
|
||||
github.com/openziti/transport/v2 v2.0.119
|
||||
github.com/openziti/ziti v0.28.3
|
||||
github.com/pkg/errors v0.9.1
|
||||
@@ -84,7 +84,7 @@ require (
|
||||
github.com/gorilla/schema v1.2.0 // indirect
|
||||
github.com/gorilla/securecookie v1.1.1 // indirect
|
||||
github.com/gorilla/websocket v1.5.1 // indirect
|
||||
github.com/hashicorp/go-hclog v1.6.1 // indirect
|
||||
github.com/hashicorp/go-hclog v1.6.2 // indirect
|
||||
github.com/hashicorp/go-immutable-radix v1.3.1 // indirect
|
||||
github.com/hashicorp/go-msgpack v0.5.5 // indirect
|
||||
github.com/hashicorp/go-msgpack/v2 v2.1.1 // indirect
|
||||
@@ -131,7 +131,7 @@ require (
|
||||
github.com/openziti/dilithium v0.3.3 // indirect
|
||||
github.com/openziti/edge-api v0.26.6 // indirect
|
||||
github.com/openziti/jwks v1.0.3 // indirect
|
||||
github.com/openziti/metrics v1.2.40 // indirect
|
||||
github.com/openziti/metrics v1.2.41 // indirect
|
||||
github.com/openziti/runzmd v1.0.36 // indirect
|
||||
github.com/openziti/secretstream v0.1.14 // indirect
|
||||
github.com/openziti/x509-claims v1.0.3 // indirect
|
||||
|
||||
+8
-8
@@ -336,8 +336,8 @@ github.com/google/pprof v0.0.0-20210226084205-cbba55b83ad5/go.mod h1:kpwsk12EmLe
|
||||
github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
|
||||
github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4=
|
||||
github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
|
||||
github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/googleapis/gax-go v2.0.0+incompatible/go.mod h1:SFVmujtThgffbyetf+mdk2eWhX2bMyUtNHzFKcPA9HY=
|
||||
github.com/googleapis/gax-go/v2 v2.0.3/go.mod h1:LLvjysVCY1JZeum8Z6l8qUty8fiNwE08qbEPm1M08qg=
|
||||
github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
|
||||
@@ -366,8 +366,8 @@ github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brv
|
||||
github.com/hashicorp/go-cleanhttp v0.5.0/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
|
||||
github.com/hashicorp/go-cleanhttp v0.5.1/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
|
||||
github.com/hashicorp/go-hclog v0.9.1/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
|
||||
github.com/hashicorp/go-hclog v1.6.1 h1:pa92nu9bPoAqI7p+uPDCIWGAibUdlCi6TYWJEQQkLf8=
|
||||
github.com/hashicorp/go-hclog v1.6.1/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M=
|
||||
github.com/hashicorp/go-hclog v1.6.2 h1:NOtoftovWkDheyUM/8JW3QMiXyxJK3uHRK7wV04nD2I=
|
||||
github.com/hashicorp/go-hclog v1.6.2/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M=
|
||||
github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
|
||||
github.com/hashicorp/go-immutable-radix v1.3.1 h1:DKHmCUm2hRBK510BaiZlwvpD40f8bJFeZnpfm2KLowc=
|
||||
github.com/hashicorp/go-immutable-radix v1.3.1/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
|
||||
@@ -606,16 +606,16 @@ github.com/openziti/identity v1.0.68 h1:SaFr7BeFQBoWQDiT28vUb8D9w7v6lIAK6/9RkwmV
|
||||
github.com/openziti/identity v1.0.68/go.mod h1:HbOu3TQ032v8xE6xZWjO51azF4fUxRLjO/l/oGqJwUI=
|
||||
github.com/openziti/jwks v1.0.3 h1:hf8wkb+Cg4nH/HM0KROFd7u+C3DkRVcFZJ7tDV+4icc=
|
||||
github.com/openziti/jwks v1.0.3/go.mod h1:t4xxq8vlXGsPn29kiQVnZBBDDnEoOFqtJoHibkJunQQ=
|
||||
github.com/openziti/metrics v1.2.40 h1:gySRgR8prCPqaEjmUtX0eXFs7NkI9uPAzp+z6A8+JqA=
|
||||
github.com/openziti/metrics v1.2.40/go.mod h1:HXdVryf3xpZfnY4VcaOjMxiBv+qw0wJlEJNLbooB9hY=
|
||||
github.com/openziti/metrics v1.2.41 h1:JShcFb6qJPA2cMiWQLtcSXiJjsrhEWpH+aVcjT/Mcbs=
|
||||
github.com/openziti/metrics v1.2.41/go.mod h1:L9h0NrliMA3+p7+ascKgvx28qoKHymN9l+CMA+Q+sZc=
|
||||
github.com/openziti/runzmd v1.0.36 h1:HOqTZFzTTFu52qmCAQfFvKDmCSl8ZqP1PQQ0UnJIA4E=
|
||||
github.com/openziti/runzmd v1.0.36/go.mod h1:jYqummjskmFh63htJFF2SrUuvxNQifqd5REUhYVaY/A=
|
||||
github.com/openziti/sdk-golang v0.21.2 h1:P66cslOAmQX37VFan+df+MoD2PqaFjHWDNMpuhhXHSo=
|
||||
github.com/openziti/sdk-golang v0.21.2/go.mod h1:mepEUD39DsBm/v1WVLedYRoYCFdet5mmJ5Sxqm/zkFI=
|
||||
github.com/openziti/secretstream v0.1.14 h1:Ta+nB5Prcct+L5LIKUA1nE56QhWS6lMPQYTlpxUltU0=
|
||||
github.com/openziti/secretstream v0.1.14/go.mod h1:/hhuLfu+GIv0+cnapfsu/VOnXEvmTt3GKtCu+lQ0RIw=
|
||||
github.com/openziti/storage v0.2.26 h1:15EbOC6A//dsdLSs/RYJP6Qn3Rj6Od4btXEWGezatxc=
|
||||
github.com/openziti/storage v0.2.26/go.mod h1:pDCkPIN7h9L+FyJP+hcfOwk+GofzrrNoYsAThOjNza0=
|
||||
github.com/openziti/storage v0.2.27 h1:WdFD0KxXZxSoWOaojdi5r1LO0BTvn4x/7wwdwhRPssc=
|
||||
github.com/openziti/storage v0.2.27/go.mod h1:p/04So5E3NT2jsGew0AS4hYTfP5srti3VytOjlVcx0M=
|
||||
github.com/openziti/transport/v2 v2.0.119 h1:KOgHU+9EZUVPvv8ncifqHmNEcFUHbJHigo3jyPvWnOc=
|
||||
github.com/openziti/transport/v2 v2.0.119/go.mod h1:H2IIBP6ed9isE/eJHGXtAZL0d73ApYOpLG9sSvutNNI=
|
||||
github.com/openziti/x509-claims v1.0.3 h1:HNdQ8Nf1agB3lBs1gahcO6zfkeS4S5xoQ2/PkY4HRX0=
|
||||
|
||||
@@ -38,7 +38,7 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
const ZitiEdgeTunnelVersion = "v0.22.12"
|
||||
const ZitiEdgeTunnelVersion = "v0.22.17"
|
||||
|
||||
//go:embed configs
|
||||
var configResource embed.FS
|
||||
|
||||
@@ -85,7 +85,7 @@ func loadInfoFile(file string) string {
|
||||
if os.IsNotExist(err) {
|
||||
return ""
|
||||
}
|
||||
logrus.WithError(err).Info("unable to read cert info file '%s'", file)
|
||||
logrus.WithError(err).Infof("unable to read cert info file '%s'", file)
|
||||
}
|
||||
return string(data)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user