mirror of
https://github.com/slackhq/nebula.git
synced 2026-08-15 03:07:01 +02:00
Make Control safe to stop and wait on from any lifecycle state (#1794)
This commit is contained in:
@@ -53,7 +53,12 @@ func main() {
|
|||||||
l := logging.NewLogger(os.Stdout)
|
l := logging.NewLogger(os.Stdout)
|
||||||
|
|
||||||
if *serviceFlag != "" {
|
if *serviceFlag != "" {
|
||||||
if err := doService(configPath, configTest, Build, serviceFlag); err != nil {
|
if *configTest {
|
||||||
|
fmt.Println("-test is not supported with -service, run the config test without -service")
|
||||||
|
os.Exit(1)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := doService(configPath, Build, serviceFlag); err != nil {
|
||||||
l.Error("Service command failed", "error", err)
|
l.Error("Service command failed", "error", err)
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
@@ -93,15 +98,14 @@ func main() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if !*configTest {
|
if !*configTest {
|
||||||
wait, err := ctrl.Start()
|
if err := ctrl.Start(); err != nil {
|
||||||
if err != nil {
|
|
||||||
util.LogWithContextIfNeeded("Error while running", err, l)
|
util.LogWithContextIfNeeded("Error while running", err, l)
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
|
|
||||||
go ctrl.ShutdownBlock()
|
go ctrl.ShutdownBlock()
|
||||||
|
|
||||||
if err := wait(); err != nil {
|
if err := ctrl.Wait(); err != nil {
|
||||||
l.Error("Nebula stopped due to fatal error", "error", err)
|
l.Error("Nebula stopped due to fatal error", "error", err)
|
||||||
os.Exit(2)
|
os.Exit(2)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package main
|
|||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
|
"os"
|
||||||
|
|
||||||
"github.com/kardianos/service"
|
"github.com/kardianos/service"
|
||||||
"github.com/slackhq/nebula"
|
"github.com/slackhq/nebula"
|
||||||
@@ -14,7 +15,6 @@ var logger service.Logger
|
|||||||
|
|
||||||
type program struct {
|
type program struct {
|
||||||
configPath *string
|
configPath *string
|
||||||
configTest *bool
|
|
||||||
build string
|
build string
|
||||||
control *nebula.Control
|
control *nebula.Control
|
||||||
}
|
}
|
||||||
@@ -40,22 +40,41 @@ func (p *program) Start(s service.Service) error {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
p.control, err = nebula.Main(c, *p.configTest, Build, l, nil)
|
p.control, err = nebula.Main(c, false, Build, l, nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
p.control.Start()
|
if err := p.control.Start(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Nebula can stop itself on a fatal packet reader error, make sure to log it if it happens.
|
||||||
|
go func() {
|
||||||
|
if err := p.control.Wait(); err != nil {
|
||||||
|
logger.Error(fmt.Sprintf("Nebula stopped due to fatal error: %v", err))
|
||||||
|
os.Exit(2)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *program) Stop(s service.Service) error {
|
func (p *program) Stop(s service.Service) error {
|
||||||
logger.Info("Nebula service stopping.")
|
logger.Info("Nebula service stopping.")
|
||||||
|
if p.control == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
p.control.Stop()
|
p.control.Stop()
|
||||||
|
|
||||||
|
// block until nebula has fully drained before reporting stopped.
|
||||||
|
// error logging is handled by Start.
|
||||||
|
_ = p.control.Wait()
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func doService(configPath *string, configTest *bool, build string, serviceFlag *string) error {
|
func doService(configPath *string, build string, serviceFlag *string) error {
|
||||||
if *configPath == "" {
|
if *configPath == "" {
|
||||||
p, err := config.DefaultPath()
|
p, err := config.DefaultPath()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -73,7 +92,6 @@ func doService(configPath *string, configTest *bool, build string, serviceFlag *
|
|||||||
|
|
||||||
prg := &program{
|
prg := &program{
|
||||||
configPath: configPath,
|
configPath: configPath,
|
||||||
configTest: configTest,
|
|
||||||
build: build,
|
build: build,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -105,8 +123,9 @@ func doService(configPath *string, configTest *bool, build string, serviceFlag *
|
|||||||
switch *serviceFlag {
|
switch *serviceFlag {
|
||||||
case "run":
|
case "run":
|
||||||
if err := s.Run(); err != nil {
|
if err := s.Run(); err != nil {
|
||||||
// Route any errors to the system logger
|
// Route any errors to the system logger and report the failure
|
||||||
logger.Error(err)
|
logger.Error(err)
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
default:
|
default:
|
||||||
if err := service.Control(s, *serviceFlag); err != nil {
|
if err := service.Control(s, *serviceFlag); err != nil {
|
||||||
|
|||||||
+2
-3
@@ -84,8 +84,7 @@ func main() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if !*configTest {
|
if !*configTest {
|
||||||
wait, err := ctrl.Start()
|
if err := ctrl.Start(); err != nil {
|
||||||
if err != nil {
|
|
||||||
util.LogWithContextIfNeeded("Error while running", err, l)
|
util.LogWithContextIfNeeded("Error while running", err, l)
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
@@ -93,7 +92,7 @@ func main() {
|
|||||||
go ctrl.ShutdownBlock()
|
go ctrl.ShutdownBlock()
|
||||||
notifyReady(l)
|
notifyReady(l)
|
||||||
|
|
||||||
if err := wait(); err != nil {
|
if err := ctrl.Wait(); err != nil {
|
||||||
l.Error("Nebula stopped due to fatal error", "error", err)
|
l.Error("Nebula stopped due to fatal error", "error", err)
|
||||||
os.Exit(2)
|
os.Exit(2)
|
||||||
}
|
}
|
||||||
|
|||||||
+49
-23
@@ -69,29 +69,29 @@ type ControlHostInfo struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Start actually runs nebula, this is a nonblocking call.
|
// Start actually runs nebula, this is a nonblocking call.
|
||||||
// The returned function blocks until nebula has fully stopped and returns the
|
// Use Wait to block until nebula has fully stopped and to learn whether a fatal reader error caused the shutdown.
|
||||||
// first fatal reader error (if any). A nil error means nebula shut down
|
func (c *Control) Start() error {
|
||||||
// gracefully; a non-nil error means a reader hit an unexpected failure that
|
|
||||||
// triggered the shutdown.
|
|
||||||
func (c *Control) Start() (func() error, error) {
|
|
||||||
c.stateLock.Lock()
|
c.stateLock.Lock()
|
||||||
defer c.stateLock.Unlock()
|
defer c.stateLock.Unlock()
|
||||||
switch c.state {
|
switch c.state {
|
||||||
case StateReady:
|
case StateReady:
|
||||||
//yay!
|
//yay!
|
||||||
case StateStopped, StateStopping:
|
case StateStopped, StateStopping:
|
||||||
return nil, ErrAlreadyStopped
|
return ErrAlreadyStopped
|
||||||
case StateStarted:
|
case StateStarted:
|
||||||
return nil, ErrAlreadyStarted
|
return ErrAlreadyStarted
|
||||||
default:
|
default:
|
||||||
return nil, ErrUnknownState
|
return ErrUnknownState
|
||||||
}
|
}
|
||||||
|
|
||||||
// Activate the interface
|
// Activate the interface
|
||||||
err := c.f.activate()
|
err := c.f.activate()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
// Cancel before Close so a caller returning from Wait always observes a dead Context
|
||||||
|
c.cancel()
|
||||||
|
_ = c.f.Close()
|
||||||
c.state = StateStopped
|
c.state = StateStopped
|
||||||
return nil, err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
// Call all the delayed funcs that waited patiently for the interface to be created.
|
// Call all the delayed funcs that waited patiently for the interface to be created.
|
||||||
@@ -114,13 +114,9 @@ func (c *Control) Start() (func() error, error) {
|
|||||||
c.f.triggerShutdown = c.Stop
|
c.f.triggerShutdown = c.Stop
|
||||||
|
|
||||||
// Start reading packets.
|
// Start reading packets.
|
||||||
out, err := c.f.run()
|
c.f.run()
|
||||||
if err != nil {
|
|
||||||
c.state = StateStopped
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
c.state = StateStarted
|
c.state = StateStarted
|
||||||
return out, nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Control) State() RunState {
|
func (c *Control) State() RunState {
|
||||||
@@ -133,10 +129,26 @@ func (c *Control) Context() context.Context {
|
|||||||
return c.ctx
|
return c.ctx
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stop is a non-blocking call that signals nebula to close all tunnels and shut down
|
// Stop tears nebula down, closing all tunnels and releasing everything it holds.
|
||||||
|
// Use Wait to block until the shutdown has completed.
|
||||||
|
// A Control that has been stopped cannot be started again, Start will return ErrAlreadyStopped.
|
||||||
func (c *Control) Stop() {
|
func (c *Control) Stop() {
|
||||||
c.stateLock.Lock()
|
c.stateLock.Lock()
|
||||||
if c.state != StateStarted {
|
switch c.state {
|
||||||
|
case StateStarted:
|
||||||
|
// Fall through to the full teardown below
|
||||||
|
|
||||||
|
case StateReady:
|
||||||
|
// Never started
|
||||||
|
c.cancel()
|
||||||
|
c.state = StateStopped
|
||||||
|
if err := c.f.Close(); err != nil {
|
||||||
|
c.l.Error("Close interface failed", "error", err)
|
||||||
|
}
|
||||||
|
c.stateLock.Unlock()
|
||||||
|
return
|
||||||
|
|
||||||
|
default:
|
||||||
c.stateLock.Unlock()
|
c.stateLock.Unlock()
|
||||||
// We are stopping or stopped already
|
// We are stopping or stopped already
|
||||||
return
|
return
|
||||||
@@ -145,19 +157,26 @@ func (c *Control) Stop() {
|
|||||||
c.state = StateStopping
|
c.state = StateStopping
|
||||||
c.stateLock.Unlock()
|
c.stateLock.Unlock()
|
||||||
|
|
||||||
// Stop the handshakeManager (and other services), to prevent new tunnels from
|
// Closing tunnels can be slow with a large hostmap, don't hold the lock for it
|
||||||
// being created while we're shutting them all down.
|
|
||||||
c.cancel()
|
c.cancel()
|
||||||
|
|
||||||
c.CloseAllTunnels(false)
|
c.CloseAllTunnels(false)
|
||||||
|
|
||||||
|
c.stateLock.Lock()
|
||||||
|
c.state = StateStopped
|
||||||
if err := c.f.Close(); err != nil {
|
if err := c.f.Close(); err != nil {
|
||||||
c.l.Error("Close interface failed", "error", err)
|
c.l.Error("Close interface failed", "error", err)
|
||||||
}
|
}
|
||||||
c.stateLock.Lock()
|
|
||||||
c.state = StateStopped
|
|
||||||
c.stateLock.Unlock()
|
c.stateLock.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Wait blocks until nebula has fully stopped, either via Stop or an internal fatal error,
|
||||||
|
// and returns the first fatal packet reader error if there was one.
|
||||||
|
// It is safe to call from multiple goroutines and at any point in the lifecycle,
|
||||||
|
// but a Wait on a Control that is never started and never stopped will block forever.
|
||||||
|
func (c *Control) Wait() error {
|
||||||
|
return c.f.wait()
|
||||||
|
}
|
||||||
|
|
||||||
// ShutdownBlock will listen for and block on term and interrupt signals, calling Control.Stop() once signalled
|
// ShutdownBlock will listen for and block on term and interrupt signals, calling Control.Stop() once signalled
|
||||||
func (c *Control) ShutdownBlock() {
|
func (c *Control) ShutdownBlock() {
|
||||||
sigChan := make(chan os.Signal, 1)
|
sigChan := make(chan os.Signal, 1)
|
||||||
@@ -170,8 +189,15 @@ func (c *Control) ShutdownBlock() {
|
|||||||
c.Stop()
|
c.Stop()
|
||||||
}
|
}
|
||||||
|
|
||||||
// RebindUDPServer asks the UDP listener to rebind it's listener. Mainly used on mobile clients when interfaces change
|
// RebindUDPServer asks the UDP listener to rebind it's listener. Mainly used on mobile clients when interfaces change.
|
||||||
func (c *Control) RebindUDPServer() {
|
func (c *Control) RebindUDPServer() {
|
||||||
|
c.stateLock.Lock()
|
||||||
|
defer c.stateLock.Unlock()
|
||||||
|
|
||||||
|
if c.state != StateStarted {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
_ = c.f.outside.Rebind()
|
_ = c.f.outside.Rebind()
|
||||||
|
|
||||||
// Trigger a lighthouse update, useful for mobile clients that should have an update interval of 0
|
// Trigger a lighthouse update, useful for mobile clients that should have an update interval of 0
|
||||||
|
|||||||
@@ -0,0 +1,292 @@
|
|||||||
|
package nebula
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"io"
|
||||||
|
"net/netip"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/gaissmai/bart"
|
||||||
|
"github.com/slackhq/nebula/config"
|
||||||
|
"github.com/slackhq/nebula/routing"
|
||||||
|
"github.com/slackhq/nebula/test"
|
||||||
|
"github.com/slackhq/nebula/udp"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
)
|
||||||
|
|
||||||
|
type fakeDevice struct {
|
||||||
|
closeOnce sync.Once
|
||||||
|
closedCh chan struct{}
|
||||||
|
closed bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func newFakeDevice() *fakeDevice {
|
||||||
|
return &fakeDevice{closedCh: make(chan struct{})}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Read blocks until Close like a real tun with no traffic, then reports EOF
|
||||||
|
// the same way a closed device does
|
||||||
|
func (d *fakeDevice) Read(p []byte) (int, error) {
|
||||||
|
<-d.closedCh
|
||||||
|
return 0, io.EOF
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d *fakeDevice) Write(p []byte) (int, error) { return len(p), nil }
|
||||||
|
|
||||||
|
func (d *fakeDevice) Close() error {
|
||||||
|
d.closeOnce.Do(func() {
|
||||||
|
d.closed = true
|
||||||
|
close(d.closedCh)
|
||||||
|
})
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d *fakeDevice) Activate() error { return nil }
|
||||||
|
func (d *fakeDevice) Networks() []netip.Prefix { return nil }
|
||||||
|
func (d *fakeDevice) Name() string { return "fake" }
|
||||||
|
func (d *fakeDevice) RoutesFor(netip.Addr) routing.Gateways { return nil }
|
||||||
|
func (d *fakeDevice) SupportsMultiqueue() bool { return false }
|
||||||
|
func (d *fakeDevice) NewMultiQueueReader() (io.ReadWriteCloser, error) {
|
||||||
|
return nil, errors.New("unsupported")
|
||||||
|
}
|
||||||
|
|
||||||
|
// newReadyControl hand-builds the minimum Control that Main would have
|
||||||
|
// produced right before Start, including the construction token NewInterface
|
||||||
|
// takes so waiters block until Close releases the resources
|
||||||
|
func newReadyControl(t *testing.T) (*Control, *fakeDevice, *fakeConn) {
|
||||||
|
l := test.NewLogger()
|
||||||
|
dev := newFakeDevice()
|
||||||
|
conn := &fakeConn{}
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
|
||||||
|
myVpnNet := netip.MustParsePrefix("10.128.0.1/16")
|
||||||
|
nt := new(bart.Lite)
|
||||||
|
nt.Insert(myVpnNet)
|
||||||
|
cs := &CertState{
|
||||||
|
myVpnNetworks: []netip.Prefix{myVpnNet},
|
||||||
|
myVpnNetworksTable: nt,
|
||||||
|
}
|
||||||
|
lh, err := NewLightHouseFromConfig(ctx, l, config.NewC(l), cs, nil, nil)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
f := &Interface{
|
||||||
|
ctx: ctx,
|
||||||
|
inside: dev,
|
||||||
|
outside: conn,
|
||||||
|
writers: []udp.Conn{conn},
|
||||||
|
readers: make([]io.ReadWriteCloser, 1),
|
||||||
|
routines: 1,
|
||||||
|
hostMap: newHostMap(l),
|
||||||
|
lightHouse: lh,
|
||||||
|
l: l,
|
||||||
|
}
|
||||||
|
f.wg.Add(1)
|
||||||
|
|
||||||
|
return &Control{
|
||||||
|
state: StateReady,
|
||||||
|
f: f,
|
||||||
|
l: l,
|
||||||
|
ctx: ctx,
|
||||||
|
cancel: cancel,
|
||||||
|
}, dev, conn
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestControl_StopBeforeStart(t *testing.T) {
|
||||||
|
c, dev, conn := newReadyControl(t)
|
||||||
|
|
||||||
|
// A Stop on a never started control must release everything Main acquired
|
||||||
|
c.Stop()
|
||||||
|
assert.Equal(t, StateStopped, c.State())
|
||||||
|
assert.True(t, dev.closed, "the tun device should have been closed")
|
||||||
|
assert.True(t, conn.closed, "the udp socket should have been closed")
|
||||||
|
require.ErrorIs(t, c.ctx.Err(), context.Canceled, "the service context should have been cancelled")
|
||||||
|
|
||||||
|
// Wait must return promptly now that the resources are released
|
||||||
|
require.NoError(t, c.Wait())
|
||||||
|
|
||||||
|
// A stopped control can never be started
|
||||||
|
require.ErrorIs(t, c.Start(), ErrAlreadyStopped)
|
||||||
|
|
||||||
|
// A second Stop is a harmless no-op
|
||||||
|
c.Stop()
|
||||||
|
assert.Equal(t, StateStopped, c.State())
|
||||||
|
require.NoError(t, c.Wait())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestControl_WaitBlocksUntilStop(t *testing.T) {
|
||||||
|
c, _, _ := newReadyControl(t)
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- c.Wait() }()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-done:
|
||||||
|
t.Fatal("Wait returned before Stop")
|
||||||
|
case <-time.After(50 * time.Millisecond):
|
||||||
|
}
|
||||||
|
|
||||||
|
c.Stop()
|
||||||
|
select {
|
||||||
|
case err := <-done:
|
||||||
|
require.NoError(t, err)
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("Wait did not return after Stop")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type fakeConn struct {
|
||||||
|
closed bool
|
||||||
|
rebinds int
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *fakeConn) Rebind() error { c.rebinds++; return nil }
|
||||||
|
func (c *fakeConn) LocalAddr() (netip.AddrPort, error) { return netip.AddrPort{}, nil }
|
||||||
|
func (c *fakeConn) ListenOut(_ udp.EncReader) error { return nil }
|
||||||
|
func (c *fakeConn) WriteTo(_ []byte, _ netip.AddrPort) error { return nil }
|
||||||
|
func (c *fakeConn) ReloadConfig(_ *config.C) {}
|
||||||
|
func (c *fakeConn) SupportsMultipleReaders() bool { return true }
|
||||||
|
func (c *fakeConn) Close() error { c.closed = true; return nil }
|
||||||
|
|
||||||
|
type multiqueueDevice struct {
|
||||||
|
*fakeDevice
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d *multiqueueDevice) SupportsMultiqueue() bool { return true }
|
||||||
|
|
||||||
|
func TestControl_StartMultiqueueFailureReleases(t *testing.T) {
|
||||||
|
dev := &multiqueueDevice{fakeDevice: newFakeDevice()}
|
||||||
|
conn := &fakeConn{}
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
f := &Interface{
|
||||||
|
ctx: ctx,
|
||||||
|
inside: dev,
|
||||||
|
outside: conn,
|
||||||
|
writers: []udp.Conn{conn},
|
||||||
|
readers: make([]io.ReadWriteCloser, 2),
|
||||||
|
routines: 2,
|
||||||
|
l: test.NewLogger(),
|
||||||
|
}
|
||||||
|
f.wg.Add(1)
|
||||||
|
|
||||||
|
c := &Control{
|
||||||
|
state: StateReady,
|
||||||
|
f: f,
|
||||||
|
l: test.NewLogger(),
|
||||||
|
ctx: ctx,
|
||||||
|
cancel: cancel,
|
||||||
|
}
|
||||||
|
|
||||||
|
// The second reader fails to open, everything must be released
|
||||||
|
require.Error(t, c.Start())
|
||||||
|
assert.Equal(t, StateStopped, c.State())
|
||||||
|
assert.True(t, dev.closed, "the tun device should have been closed")
|
||||||
|
assert.True(t, conn.closed, "the udp socket should have been closed")
|
||||||
|
require.ErrorIs(t, c.ctx.Err(), context.Canceled)
|
||||||
|
|
||||||
|
// And Wait must not hang on the construction token
|
||||||
|
require.NoError(t, c.Wait())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestInterface_CloseIsIdempotent(t *testing.T) {
|
||||||
|
dev := newFakeDevice()
|
||||||
|
f := &Interface{
|
||||||
|
inside: dev,
|
||||||
|
l: test.NewLogger(),
|
||||||
|
}
|
||||||
|
f.wg.Add(1)
|
||||||
|
|
||||||
|
require.NoError(t, f.Close())
|
||||||
|
assert.True(t, dev.closed)
|
||||||
|
|
||||||
|
// A second Close must not double release the wg token or the device
|
||||||
|
require.NoError(t, f.Close())
|
||||||
|
require.NoError(t, f.wait())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestControl_FatalErrorReportsThroughWait(t *testing.T) {
|
||||||
|
c, dev, conn := newReadyControl(t)
|
||||||
|
|
||||||
|
// Mirror what Start wires up, without needing real packet readers
|
||||||
|
c.f.triggerShutdown = c.Stop
|
||||||
|
c.state = StateStarted
|
||||||
|
|
||||||
|
boom := errors.New("boom")
|
||||||
|
c.f.onFatal(boom)
|
||||||
|
|
||||||
|
require.ErrorIs(t, c.Wait(), boom)
|
||||||
|
assert.Equal(t, StateStopped, c.State())
|
||||||
|
assert.True(t, dev.closed)
|
||||||
|
assert.True(t, conn.closed)
|
||||||
|
|
||||||
|
// A second fatal error must not fire the shutdown again or replace the first
|
||||||
|
c.f.onFatal(errors.New("later"))
|
||||||
|
require.ErrorIs(t, c.Wait(), boom)
|
||||||
|
|
||||||
|
// Wait stays factual, a Stop after the death does not mask the error
|
||||||
|
c.Stop()
|
||||||
|
require.ErrorIs(t, c.Wait(), boom)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestControl_ConcurrentStopAndStart(t *testing.T) {
|
||||||
|
c, _, _ := newReadyControl(t)
|
||||||
|
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
for i := 0; i < 2; i++ {
|
||||||
|
wg.Go(func() { c.Stop() })
|
||||||
|
}
|
||||||
|
wg.Go(func() { _ = c.Start() })
|
||||||
|
wg.Go(func() {
|
||||||
|
_ = c.Wait()
|
||||||
|
// A returned Wait must always observe the final state, no matter how
|
||||||
|
// the race resolved
|
||||||
|
assert.Equal(t, StateStopped, c.State())
|
||||||
|
})
|
||||||
|
wg.Wait()
|
||||||
|
|
||||||
|
// However the race resolves, the control must end fully stopped with no
|
||||||
|
// panic and Wait must observe the final state
|
||||||
|
require.NoError(t, c.Wait())
|
||||||
|
assert.Equal(t, StateStopped, c.State())
|
||||||
|
require.ErrorIs(t, c.Start(), ErrAlreadyStopped)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestControl_StartStopLifecycle(t *testing.T) {
|
||||||
|
c, dev, conn := newReadyControl(t)
|
||||||
|
|
||||||
|
require.NoError(t, c.Start())
|
||||||
|
assert.Equal(t, StateStarted, c.State())
|
||||||
|
require.ErrorIs(t, c.Start(), ErrAlreadyStarted)
|
||||||
|
|
||||||
|
// Stop must unpark the reader blocked in the device and release everything
|
||||||
|
c.Stop()
|
||||||
|
assert.Equal(t, StateStopped, c.State())
|
||||||
|
assert.True(t, dev.closed, "the tun device should have been closed")
|
||||||
|
assert.True(t, conn.closed, "the udp socket should have been closed")
|
||||||
|
require.ErrorIs(t, c.ctx.Err(), context.Canceled)
|
||||||
|
|
||||||
|
// The reader drained off a closed device, that is not a fatal error
|
||||||
|
require.NoError(t, c.Wait())
|
||||||
|
require.ErrorIs(t, c.Start(), ErrAlreadyStopped)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestControl_RebindIsGatedByState(t *testing.T) {
|
||||||
|
c, _, conn := newReadyControl(t)
|
||||||
|
|
||||||
|
// A rebind before Start reaches nothing, the interface is not up
|
||||||
|
c.RebindUDPServer()
|
||||||
|
assert.Equal(t, 0, conn.rebinds, "rebind before start must be a no-op")
|
||||||
|
|
||||||
|
require.NoError(t, c.Start())
|
||||||
|
c.RebindUDPServer()
|
||||||
|
assert.Equal(t, 1, conn.rebinds, "rebind while started must reach the conn")
|
||||||
|
|
||||||
|
// A rebind racing a completed stop must not touch the closed conn
|
||||||
|
c.Stop()
|
||||||
|
require.NoError(t, c.Wait())
|
||||||
|
c.RebindUDPServer()
|
||||||
|
assert.Equal(t, 1, conn.rebinds, "rebind after stop must be a no-op")
|
||||||
|
}
|
||||||
+29
-14
@@ -215,6 +215,9 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
|
|||||||
|
|
||||||
ifce.connectionManager.intf = ifce
|
ifce.connectionManager.intf = ifce
|
||||||
|
|
||||||
|
// Held until Close so waiting on the interface blocks until the resources are actually released
|
||||||
|
ifce.wg.Add(1)
|
||||||
|
|
||||||
return ifce, nil
|
return ifce, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -258,17 +261,16 @@ func (f *Interface) activate() error {
|
|||||||
f.readers[i] = reader
|
f.readers[i] = reader
|
||||||
}
|
}
|
||||||
|
|
||||||
f.wg.Add(1) // for us to wait on Close() to return
|
// On error the caller owns the cleanup, Control.Start cancels the service context
|
||||||
|
// before releasing our resources so a waiter never observes a live context
|
||||||
if err = f.inside.Activate(); err != nil {
|
if err = f.inside.Activate(); err != nil {
|
||||||
f.wg.Done()
|
|
||||||
f.inside.Close()
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (f *Interface) run() (func() error, error) {
|
func (f *Interface) run() {
|
||||||
// Launch n queues to read packets from udp
|
// Launch n queues to read packets from udp
|
||||||
for i := 0; i < f.routines; i++ {
|
for i := 0; i < f.routines; i++ {
|
||||||
f.wg.Go(func() {
|
f.wg.Go(func() {
|
||||||
@@ -283,13 +285,14 @@ func (f *Interface) run() (func() error, error) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
return func() error {
|
}
|
||||||
f.wg.Wait()
|
|
||||||
if e := f.fatalErr.Load(); e != nil {
|
func (f *Interface) wait() error {
|
||||||
return *e
|
f.wg.Wait()
|
||||||
}
|
if e := f.fatalErr.Load(); e != nil {
|
||||||
return nil
|
return *e
|
||||||
}, nil
|
}
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// onFatal stores the first fatal reader error, and calls triggerShutdown if it was the first one
|
// onFatal stores the first fatal reader error, and calls triggerShutdown if it was the first one
|
||||||
@@ -322,7 +325,10 @@ func (f *Interface) listenOut(i int) {
|
|||||||
f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, plaintext[:0], payload, h, fwPacket, lhh, nb, i, ctCache.Get())
|
f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, plaintext[:0], payload, h, fwPacket, lhh, nb, i, ctCache.Get())
|
||||||
})
|
})
|
||||||
|
|
||||||
if err != nil && !f.closed.Load() {
|
// An error after teardown began is shutdown noise, the closed flag covers resources
|
||||||
|
// Close releases itself and the cancelled ctx covers ones torn down by their owners
|
||||||
|
// reacting to it, like the user device pipes
|
||||||
|
if err != nil && !f.closed.Load() && f.ctx.Err() == nil {
|
||||||
f.l.Error("Error while reading inbound packet, closing", "error", err)
|
f.l.Error("Error while reading inbound packet, closing", "error", err)
|
||||||
f.onFatal(err)
|
f.onFatal(err)
|
||||||
}
|
}
|
||||||
@@ -341,7 +347,8 @@ func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) {
|
|||||||
for {
|
for {
|
||||||
n, err := reader.Read(packet)
|
n, err := reader.Read(packet)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if !f.closed.Load() {
|
// Same shutdown noise handling as listenOut
|
||||||
|
if !f.closed.Load() && f.ctx.Err() == nil {
|
||||||
f.l.Error("Error while reading outbound packet, closing", "error", err, "reader", i)
|
f.l.Error("Error while reading outbound packet, closing", "error", err, "reader", i)
|
||||||
f.onFatal(err)
|
f.onFatal(err)
|
||||||
}
|
}
|
||||||
@@ -542,9 +549,15 @@ func (f *Interface) GetCertState() *CertState {
|
|||||||
return f.pki.getCertState()
|
return f.pki.getCertState()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Close releases the interface's resources: the udp sockets and the tun device.
|
||||||
|
// It is idempotent and safe to call at any point in the lifecycle, including on an interface that never activated,
|
||||||
|
// calls after the first return nil without doing anything.
|
||||||
func (f *Interface) Close() error {
|
func (f *Interface) Close() error {
|
||||||
|
if !f.closed.CompareAndSwap(false, true) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
var errs []error
|
var errs []error
|
||||||
f.closed.Store(true)
|
|
||||||
|
|
||||||
// Release the udp readers
|
// Release the udp readers
|
||||||
for i, u := range f.writers {
|
for i, u := range f.writers {
|
||||||
@@ -560,6 +573,8 @@ func (f *Interface) Close() error {
|
|||||||
if closeErr != nil {
|
if closeErr != nil {
|
||||||
errs = append(errs, closeErr)
|
errs = append(errs, closeErr)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Release the construction token so waiters know the resources are gone
|
||||||
f.wg.Done()
|
f.wg.Done()
|
||||||
return errors.Join(errs...)
|
return errors.Join(errs...)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -130,6 +130,17 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
|
|||||||
udpConns := make([]udp.Conn, routines)
|
udpConns := make([]udp.Conn, routines)
|
||||||
port := c.GetInt("listen.port", 0)
|
port := c.GetInt("listen.port", 0)
|
||||||
|
|
||||||
|
// Callers get no handle to these until the Control is returned, release them on any error.
|
||||||
|
defer func() {
|
||||||
|
if reterr != nil {
|
||||||
|
for _, u := range udpConns {
|
||||||
|
if u != nil {
|
||||||
|
_ = u.Close()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
if !configTest {
|
if !configTest {
|
||||||
rawListenHost := c.GetString("listen.host", "0.0.0.0")
|
rawListenHost := c.GetString("listen.host", "0.0.0.0")
|
||||||
var listenHost netip.Addr
|
var listenHost netip.Addr
|
||||||
|
|||||||
@@ -40,6 +40,7 @@ func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip
|
|||||||
|
|
||||||
err := t.reload(c, true)
|
err := t.reload(c, true)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
_ = file.Close()
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ import (
|
|||||||
"github.com/slackhq/nebula/config"
|
"github.com/slackhq/nebula/config"
|
||||||
"github.com/slackhq/nebula/routing"
|
"github.com/slackhq/nebula/routing"
|
||||||
"github.com/slackhq/nebula/util"
|
"github.com/slackhq/nebula/util"
|
||||||
|
"golang.org/x/sys/unix"
|
||||||
)
|
)
|
||||||
|
|
||||||
type tun struct {
|
type tun struct {
|
||||||
@@ -33,6 +34,12 @@ func newTun(_ *config.C, _ *slog.Logger, _ []netip.Prefix, _ bool) (*tun, error)
|
|||||||
}
|
}
|
||||||
|
|
||||||
func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) {
|
func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) {
|
||||||
|
if err := unix.SetNonblock(deviceFd, true); err != nil {
|
||||||
|
// We own the fd from the moment it is handed to us, same as the reload error path below
|
||||||
|
_ = unix.Close(deviceFd)
|
||||||
|
return nil, fmt.Errorf("failed to set the tun fd to non-blocking mode: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
file := os.NewFile(uintptr(deviceFd), "/dev/tun")
|
file := os.NewFile(uintptr(deviceFd), "/dev/tun")
|
||||||
t := &tun{
|
t := &tun{
|
||||||
vpnNetworks: vpnNetworks,
|
vpnNetworks: vpnNetworks,
|
||||||
@@ -42,6 +49,7 @@ func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip
|
|||||||
|
|
||||||
err := t.reload(c, true)
|
err := t.reload(c, true)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
_ = file.Close()
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+16
-8
@@ -43,12 +43,25 @@ type Service struct {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func New(control *nebula.Control) (*Service, error) {
|
func New(control *nebula.Control) (_ *Service, reterr error) {
|
||||||
wait, err := control.Start()
|
// Check this before Start so a failure doesn't leave a running nebula
|
||||||
|
device, ok := control.Device().(*overlay.UserDevice)
|
||||||
|
if !ok {
|
||||||
|
return nil, errors.New("must be using user device")
|
||||||
|
}
|
||||||
|
|
||||||
|
err := control.Start()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Anything that fails after a successful Start must tear nebula back down
|
||||||
|
defer func() {
|
||||||
|
if reterr != nil {
|
||||||
|
control.Stop()
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
ctx := control.Context()
|
ctx := control.Context()
|
||||||
eg, ctx := errgroup.WithContext(ctx)
|
eg, ctx := errgroup.WithContext(ctx)
|
||||||
s := Service{
|
s := Service{
|
||||||
@@ -57,11 +70,6 @@ func New(control *nebula.Control) (*Service, error) {
|
|||||||
}
|
}
|
||||||
s.mu.listeners = map[uint16]*tcpListener{}
|
s.mu.listeners = map[uint16]*tcpListener{}
|
||||||
|
|
||||||
device, ok := control.Device().(*overlay.UserDevice)
|
|
||||||
if !ok {
|
|
||||||
return nil, errors.New("must be using user device")
|
|
||||||
}
|
|
||||||
|
|
||||||
s.ipstack = stack.New(stack.Options{
|
s.ipstack = stack.New(stack.Options{
|
||||||
NetworkProtocols: []stack.NetworkProtocolFactory{ipv4.NewProtocol, ipv6.NewProtocol},
|
NetworkProtocols: []stack.NetworkProtocolFactory{ipv4.NewProtocol, ipv6.NewProtocol},
|
||||||
TransportProtocols: []stack.TransportProtocolFactory{tcp.NewProtocol, udp.NewProtocol, icmp.NewProtocol4, icmp.NewProtocol6},
|
TransportProtocols: []stack.TransportProtocolFactory{tcp.NewProtocol, udp.NewProtocol, icmp.NewProtocol4, icmp.NewProtocol6},
|
||||||
@@ -147,7 +155,7 @@ func New(control *nebula.Control) (*Service, error) {
|
|||||||
// Add the nebula wait function to the group so a fatal reader error
|
// Add the nebula wait function to the group so a fatal reader error
|
||||||
// propagates out through errgroup.Wait().
|
// propagates out through errgroup.Wait().
|
||||||
eg.Go(func() error {
|
eg.Go(func() error {
|
||||||
return wait()
|
return control.Wait()
|
||||||
})
|
})
|
||||||
|
|
||||||
return &s, nil
|
return &s, nil
|
||||||
|
|||||||
Reference in New Issue
Block a user