From c19f0dbbd48d0a25bbd8fed395926ecb8c09209e Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Thu, 23 Apr 2020 20:20:49 -0400 Subject: [PATCH] Allow router to be shutdown for tests --- controller/controller.go | 37 +++++++++++++--------- controller/network/network.go | 17 ++++++++++ router/router.go | 41 +++++++++++++++++++++++-- router/xgress/close.go | 37 ++++++++++++++++++++++ router/xgress/xgress.go | 1 + router/xgress_proxy/listener.go | 24 +++++++++------ router/xgress_proxy_udp/listener.go | 13 +++++--- router/xgress_transport/listener.go | 20 +++++++----- router/xgress_transport_udp/listener.go | 16 +++++----- router/xgress_udp/listener.go | 6 +++- router/xlink/xlink.go | 1 + router/xlink_transport/listener.go | 8 +++++ router/xlink_transwarp/listener.go | 9 +++++- 13 files changed, 184 insertions(+), 46 deletions(-) create mode 100644 router/xgress/close.go diff --git a/controller/controller.go b/controller/controller.go index f51b8ff52..b676f2aa4 100644 --- a/controller/controller.go +++ b/controller/controller.go @@ -28,6 +28,7 @@ import ( "github.com/netfoundry/ziti-fabric/controller/xmgmt" "github.com/netfoundry/ziti-foundation/channel2" "github.com/netfoundry/ziti-foundation/profiler" + "github.com/netfoundry/ziti-foundation/util/concurrenz" ) type Controller struct { @@ -40,11 +41,15 @@ type Controller struct { ctrlListener channel2.UnderlayListener mgmtListener channel2.UnderlayListener + + shutdownC chan struct{} + isShutdown concurrenz.AtomicBoolean } func NewController(cfg *Config) (*Controller, error) { c := &Controller{ - config: cfg, + config: cfg, + shutdownC: make(chan struct{}), } if n, err := network.NewNetwork(cfg.Id, cfg.Network, cfg.Db, cfg.Metrics); err == nil { @@ -97,23 +102,27 @@ func (c *Controller) Run() error { } func (c *Controller) Shutdown() { - if c.ctrlListener != nil { - if err := c.ctrlListener.Close(); err != nil { - pfxlog.Logger().WithError(err).Error("failed to close ctrl channel listener") + if c.isShutdown.CompareAndSwap(false, true) { + close(c.shutdownC) + + if c.ctrlListener != nil { + if err := c.ctrlListener.Close(); err != nil { + pfxlog.Logger().WithError(err).Error("failed to close ctrl channel listener") + } } - } - if c.mgmtListener != nil { - if err := c.mgmtListener.Close(); err != nil { - pfxlog.Logger().WithError(err).Error("failed to close mgmt channel listener") + if c.mgmtListener != nil { + if err := c.mgmtListener.Close(); err != nil { + pfxlog.Logger().WithError(err).Error("failed to close mgmt channel listener") + } } - } - c.network.Shutdown() + c.network.Shutdown() - if c.config.Db != nil { - if err := c.config.Db.Close(); err != nil { - pfxlog.Logger().WithError(err).Error("failed to close db") + if c.config.Db != nil { + if err := c.config.Db.Close(); err != nil { + pfxlog.Logger().WithError(err).Error("failed to close db") + } } } } @@ -134,7 +143,7 @@ func (c *Controller) showOptions() error { func (c *Controller) startProfiling() { if c.config.Profile.Memory.Path != "" { - go profiler.NewMemory(c.config.Profile.Memory.Path, c.config.Profile.Memory.Interval).Run() + go profiler.NewMemoryWithShutdown(c.config.Profile.Memory.Path, c.config.Profile.Memory.Interval, c.shutdownC).Run() } } diff --git a/controller/network/network.go b/controller/network/network.go index 60f927d92..4665ec4fc 100644 --- a/controller/network/network.go +++ b/controller/network/network.go @@ -591,3 +591,20 @@ func (network *Network) GetServiceCache() Cache { type Cache interface { RemoveFromCache(id string) } + +type MultipleErrors []error + +func (e MultipleErrors) Error() string { + if len(e) == 0 { + return "no errors occurred" + } + if len(e) == 1 { + return e[0].Error() + } + buf := strings.Builder{} + buf.WriteString("multiple errors occurred") + for idx, err := range e { + buf.WriteString(fmt.Sprintf(" %v: %v", idx, err)) + } + return buf.String() +} diff --git a/router/router.go b/router/router.go index d49425065..48266cb07 100644 --- a/router/router.go +++ b/router/router.go @@ -20,6 +20,7 @@ import ( "encoding/json" "fmt" "github.com/michaelquigley/pfxlog" + "github.com/netfoundry/ziti-fabric/controller/network" "github.com/netfoundry/ziti-fabric/controller/xctrl" "github.com/netfoundry/ziti-fabric/router/forwarder" "github.com/netfoundry/ziti-fabric/router/handler_ctrl" @@ -37,6 +38,7 @@ import ( "github.com/netfoundry/ziti-foundation/metrics" "github.com/netfoundry/ziti-foundation/metrics/metrics_pb" "github.com/netfoundry/ziti-foundation/profiler" + "github.com/netfoundry/ziti-foundation/util/concurrenz" "github.com/netfoundry/ziti-foundation/util/info" "github.com/sirupsen/logrus" "math/rand" @@ -55,7 +57,10 @@ type Router struct { xlinkFactories map[string]xlink.Factory xlinkListeners []xlink.Listener xlinkDialers []xlink.Dialer + xgressListeners []xgress.Listener metricsRegistry metrics.Registry + shutdownC chan struct{} + isShutdown concurrenz.AtomicBoolean } func (self *Router) Channel() channel2.Channel { @@ -69,6 +74,7 @@ func Create(config *Config) *Router { config: config, forwarder: forwarder.NewForwarder(metricsRegistry), metricsRegistry: metricsRegistry, + shutdownC: make(chan struct{}), } } @@ -103,6 +109,36 @@ func (self *Router) Start() error { return nil } +func (self *Router) Shutdown() error { + var errors []error + if self.isShutdown.CompareAndSwap(false, true) { + if err := self.ctrl.Close(); err != nil { + errors = append(errors, err) + } + + close(self.shutdownC) + + for _, xlinkListener := range self.xlinkListeners { + if err := xlinkListener.Close(); err != nil { + errors = append(errors, err) + } + } + + for _, xgressListener := range self.xgressListeners { + if err := xgressListener.Close(); err != nil { + errors = append(errors, err) + } + } + } + if len(errors) == 0 { + return nil + } + if len(errors) == 1 { + return errors[0] + } + return network.MultipleErrors(errors) +} + func (self *Router) Run() error { if err := self.Start(); err != nil { return err @@ -122,10 +158,10 @@ func (self *Router) showOptions() { func (self *Router) startProfiling() { if self.config.Profile.Memory.Path != "" { - go profiler.NewMemory(self.config.Profile.Memory.Path, self.config.Profile.Memory.Interval).Run() + go profiler.NewMemoryWithShutdown(self.config.Profile.Memory.Path, self.config.Profile.Memory.Interval, self.shutdownC).Run() } if self.config.Profile.CPU.Path != "" { - if cpu, err := profiler.NewCPU(self.config.Profile.CPU.Path); err == nil { + if cpu, err := profiler.NewCPUWithShutdown(self.config.Profile.CPU.Path, self.shutdownC); err == nil { go cpu.Run() } else { logrus.Errorf("unexpected error launching cpu profiling (%v)", err) @@ -194,6 +230,7 @@ func (self *Router) startXgressListeners() { if err != nil { logrus.Fatalf("error creating xgress listener [%s] (%v)", binding.name, err) } + self.xgressListeners = append(self.xgressListeners, listener) if address, found := binding.options["address"]; found { err = listener.Listen(address.(string), handler_xgress.NewBindHandler( diff --git a/router/xgress/close.go b/router/xgress/close.go new file mode 100644 index 000000000..b217d9388 --- /dev/null +++ b/router/xgress/close.go @@ -0,0 +1,37 @@ +package xgress + +import ( + "io" + "sync" +) + +type CloseHelper struct { + closer io.Closer + closeLock sync.Mutex +} + +func (self *CloseHelper) Init(closer io.Closer) { + if self == nil { + return + } + + self.closeLock.Lock() + defer self.closeLock.Unlock() + self.closer = closer +} + +func (self *CloseHelper) Close() error { + if self == nil { + return nil + } + + self.closeLock.Lock() + defer self.closeLock.Unlock() + + if self.closer != nil { + result := self.closer.Close() + self.closer = nil + return result + } + return nil +} diff --git a/router/xgress/xgress.go b/router/xgress/xgress.go index b7a019cce..3983999b9 100644 --- a/router/xgress/xgress.go +++ b/router/xgress/xgress.go @@ -38,6 +38,7 @@ type Address string type Listener interface { Listen(address string, bindHandler BindHandler) error + Close() error } type Dialer interface { diff --git a/router/xgress_proxy/listener.go b/router/xgress_proxy/listener.go index 442ddb199..8f8aa147a 100644 --- a/router/xgress_proxy/listener.go +++ b/router/xgress_proxy/listener.go @@ -26,10 +26,11 @@ import ( func newListener(id *identity.TokenId, ctrl xgress.CtrlChannel, options *xgress.Options, service string) xgress.Listener { return &listener{ - id: id, - ctrl: ctrl, - options: options, - service: service, + id: id, + ctrl: ctrl, + options: options, + service: service, + closeHelper: &xgress.CloseHelper{}, } } @@ -40,7 +41,7 @@ func (listener *listener) Listen(address string, bindHandler xgress.BindHandler) } incomingPeers := make(chan transport.Connection) - go txAddress.MustListen("tcp", listener.id, incomingPeers) + go listener.closeHelper.Init(txAddress.MustListen("tcp", listener.id, incomingPeers)) go func() { for { select { @@ -69,8 +70,13 @@ func (listener *listener) handleConnect(peer transport.Connection, bindHandler x } type listener struct { - id *identity.TokenId - ctrl xgress.CtrlChannel - options *xgress.Options - service string + id *identity.TokenId + ctrl xgress.CtrlChannel + options *xgress.Options + service string + closeHelper *xgress.CloseHelper +} + +func (listener *listener) Close() error { + return listener.closeHelper.Close() } diff --git a/router/xgress_proxy_udp/listener.go b/router/xgress_proxy_udp/listener.go index 0424e4a26..09a3d0d65 100644 --- a/router/xgress_proxy_udp/listener.go +++ b/router/xgress_proxy_udp/listener.go @@ -70,7 +70,11 @@ func (l *listener) LogContext() string { } func (l *listener) relay() { - defer l.close() + defer func() { + if err := l.Close(); err != nil { + logrus.Errorf("error closing packet connection (%v)", err) + } + }() for { buf := make([]byte, udp.MaxPacketSize) @@ -139,12 +143,11 @@ func (l *listener) handleConnect(session xgress_udp.Session) { } } -func (l *listener) close() { +func (l *listener) Close() error { if l.conn != nil { - if err := l.conn.Close(); err != nil { - logrus.Errorf("error closing packet connection (%v)", err) - } + return l.conn.Close() } + return nil } func newListener(service string, ctrl xgress.CtrlChannel, options *xgress.Options) xgress.Listener { diff --git a/router/xgress_transport/listener.go b/router/xgress_transport/listener.go index 4214794f6..b21d16751 100644 --- a/router/xgress_transport/listener.go +++ b/router/xgress_transport/listener.go @@ -25,16 +25,18 @@ import ( ) type listener struct { - id *identity.TokenId - ctrl xgress.CtrlChannel - options *xgress.Options + id *identity.TokenId + ctrl xgress.CtrlChannel + options *xgress.Options + closeHelper *xgress.CloseHelper } func newListener(id *identity.TokenId, ctrl xgress.CtrlChannel, options *xgress.Options) xgress.Listener { return &listener{ - id: id, - ctrl: ctrl, - options: options, + id: id, + ctrl: ctrl, + options: options, + closeHelper: &xgress.CloseHelper{}, } } @@ -45,7 +47,7 @@ func (listener *listener) Listen(address string, bindHandler xgress.BindHandler) } incomingPeers := make(chan transport.Connection) - go txAddress.MustListen("tcp", listener.id, incomingPeers) + go listener.closeHelper.Init(txAddress.MustListen("tcp", listener.id, incomingPeers)) go func() { for { select { @@ -62,6 +64,10 @@ func (listener *listener) Listen(address string, bindHandler xgress.BindHandler) return nil } +func (listener *listener) Close() error { + return listener.closeHelper.Close() +} + func (listener *listener) handleConnect(peer transport.Connection, bindHandler xgress.BindHandler) { conn := &transportXgresscConn{peer} log := pfxlog.ContextLogger(conn.LogContext()) diff --git a/router/xgress_transport_udp/listener.go b/router/xgress_transport_udp/listener.go index 30d7cacac..a97dd30f0 100644 --- a/router/xgress_transport_udp/listener.go +++ b/router/xgress_transport_udp/listener.go @@ -71,20 +71,22 @@ func (l *listener) LogContext() string { return l.address } -func (l *listener) close() { - logger := pfxlog.ContextLogger(l.address) +func (l *listener) Close() error { if l.conn != nil { - if err := l.conn.Close(); err != nil { - logger.Errorf("failure closing packet conn. (%v)", err) - } + return l.conn.Close() } + return nil } func (l *listener) relayIncomingPackets() { - defer l.close() - logger := pfxlog.ContextLogger(l.address) + defer func() { + if err := l.Close(); err != nil { + logger.Errorf("failure closing packet conn. (%v)", err) + } + }() + for { buf := make([]byte, udp.MaxPacketSize) logger.Debugf("Trying to read next packet") diff --git a/router/xgress_udp/listener.go b/router/xgress_udp/listener.go index b1d328153..22ceefdb2 100644 --- a/router/xgress_udp/listener.go +++ b/router/xgress_udp/listener.go @@ -16,9 +16,13 @@ package xgress_udp -import "net" +import ( + "io" + "net" +) type Listener interface { + io.Closer WriteTo(data []byte, addr net.Addr) (int, error) GetSession(sessionId string) (Session, bool) DeleteSession(sessionId string) diff --git a/router/xlink/xlink.go b/router/xlink/xlink.go index 923c64e0f..57953db07 100644 --- a/router/xlink/xlink.go +++ b/router/xlink/xlink.go @@ -29,6 +29,7 @@ type Factory interface { type Listener interface { Listen() error GetAdvertisement() string + Close() error } type Accepter interface { diff --git a/router/xlink_transport/listener.go b/router/xlink_transport/listener.go index 901be7f51..0f4f2e0b5 100644 --- a/router/xlink_transport/listener.go +++ b/router/xlink_transport/listener.go @@ -17,6 +17,7 @@ package xlink_transport import ( + "errors" "fmt" "github.com/netfoundry/ziti-fabric/router/xlink" "github.com/netfoundry/ziti-foundation/channel2" @@ -39,6 +40,10 @@ func (self *listener) GetAdvertisement() string { return self.config.advertise.String() } +func (self *listener) Close() error { + return self.listener.Close() +} + func (self *listener) acceptLoop() { for { ch, err := channel2.NewChannel("link", self.listener, self.config.options) @@ -58,6 +63,9 @@ func (self *listener) acceptLoop() { logrus.Infof("accepted link [%s]", "l/"+ch.Id().Token) + } else if errors.Is(err, channel2.ListenerClosedError) { + logrus.Errorf("link underlay acceptor closed") + return } else { logrus.Errorf("error creating link underlay (%v)", err) } diff --git a/router/xlink_transwarp/listener.go b/router/xlink_transwarp/listener.go index 2c96ade95..c4377eca3 100644 --- a/router/xlink_transwarp/listener.go +++ b/router/xlink_transwarp/listener.go @@ -20,6 +20,7 @@ import ( "fmt" "github.com/netfoundry/ziti-fabric/router/xlink" "github.com/netfoundry/ziti-foundation/identity/identity" + "github.com/netfoundry/ziti-foundation/util/concurrenz" "github.com/sirupsen/logrus" "net" ) @@ -57,11 +58,16 @@ func (self *listener) HandleHello(linkId *identity.TokenId, conn *net.UDPConn, p } } +func (self *listener) Close() error { + defer self.closed.Set(true) + return self.listener.Close() +} + /* * xlink_transwarp.listener */ func (self *listener) listen() { - for { + for !self.closed.Get() { if m, peer, err := readMessage(self.listener); err == nil { if m.messageType == Hello { if err := handleHello(m, self.listener, peer, self); err != nil { @@ -89,4 +95,5 @@ type listener struct { accepter xlink.Accepter forwarder xlink.Forwarder peers map[string]*impl + closed concurrenz.AtomicBoolean }