mirror of
https://github.com/slackhq/nebula.git
synced 2026-05-15 20:37:36 +02:00
Plug the conntrack cache ticker leak and nebula-service log.Fatal calls (#1669)
This commit is contained in:
@@ -50,10 +50,16 @@ func main() {
|
||||
os.Exit(0)
|
||||
}
|
||||
|
||||
l := logrus.New()
|
||||
l.Out = os.Stdout
|
||||
|
||||
if *serviceFlag != "" {
|
||||
doService(configPath, configTest, Build, serviceFlag)
|
||||
if err := doService(configPath, configTest, Build, serviceFlag); err != nil {
|
||||
l.WithError(err).Error("Service command failed")
|
||||
os.Exit(1)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if *configPath == "" {
|
||||
fmt.Println("-config flag must be set")
|
||||
@@ -61,9 +67,6 @@ func main() {
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
l := logrus.New()
|
||||
l.Out = os.Stdout
|
||||
|
||||
c := config.NewC(l)
|
||||
err := c.Load(*configPath)
|
||||
if err != nil {
|
||||
|
||||
@@ -57,11 +57,11 @@ func fileExists(filename string) bool {
|
||||
return true
|
||||
}
|
||||
|
||||
func doService(configPath *string, configTest *bool, build string, serviceFlag *string) {
|
||||
func doService(configPath *string, configTest *bool, build string, serviceFlag *string) error {
|
||||
if *configPath == "" {
|
||||
ex, err := os.Executable()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
return err
|
||||
}
|
||||
*configPath = filepath.Dir(ex) + "/config.yaml"
|
||||
if !fileExists(*configPath) {
|
||||
@@ -88,13 +88,13 @@ func doService(configPath *string, configTest *bool, build string, serviceFlag *
|
||||
// - above, in `Run` we create a `logrus.Logger` which is what nebula expects to use
|
||||
s, err := service.New(prg, svcConfig)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
|
||||
errs := make(chan error, 5)
|
||||
logger, err = s.Logger(errs)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
|
||||
go func() {
|
||||
@@ -109,18 +109,16 @@ func doService(configPath *string, configTest *bool, build string, serviceFlag *
|
||||
|
||||
switch *serviceFlag {
|
||||
case "run":
|
||||
err = s.Run()
|
||||
if err != nil {
|
||||
if err := s.Run(); err != nil {
|
||||
// Route any errors to the system logger
|
||||
logger.Error(err)
|
||||
}
|
||||
default:
|
||||
err := service.Control(s, *serviceFlag)
|
||||
if err != nil {
|
||||
if err := service.Control(s, *serviceFlag); err != nil {
|
||||
log.Printf("Valid actions: %q\n", service.ControlAction)
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package firewall
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
@@ -18,7 +19,7 @@ type ConntrackCacheTicker struct {
|
||||
cache ConntrackCache
|
||||
}
|
||||
|
||||
func NewConntrackCacheTicker(d time.Duration) *ConntrackCacheTicker {
|
||||
func NewConntrackCacheTicker(ctx context.Context, d time.Duration) *ConntrackCacheTicker {
|
||||
if d == 0 {
|
||||
return nil
|
||||
}
|
||||
@@ -27,17 +28,23 @@ func NewConntrackCacheTicker(d time.Duration) *ConntrackCacheTicker {
|
||||
cache: ConntrackCache{},
|
||||
}
|
||||
|
||||
go c.tick(d)
|
||||
go c.tick(ctx, d)
|
||||
|
||||
return c
|
||||
}
|
||||
|
||||
func (c *ConntrackCacheTicker) tick(d time.Duration) {
|
||||
func (c *ConntrackCacheTicker) tick(ctx context.Context, d time.Duration) {
|
||||
t := time.NewTicker(d)
|
||||
defer t.Stop()
|
||||
for {
|
||||
time.Sleep(d)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
c.cacheTick.Add(1)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Get checks if the cache ticker has moved to the next version before returning
|
||||
// the map. If it has moved, we reset the map.
|
||||
|
||||
@@ -85,6 +85,7 @@ type Interface struct {
|
||||
|
||||
conntrackCacheTimeout time.Duration
|
||||
|
||||
ctx context.Context
|
||||
writers []udp.Conn
|
||||
readers []io.ReadWriteCloser
|
||||
wg sync.WaitGroup
|
||||
@@ -170,6 +171,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
|
||||
|
||||
cs := c.pki.getCertState()
|
||||
ifce := &Interface{
|
||||
ctx: ctx,
|
||||
pki: c.pki,
|
||||
hostMap: c.HostMap,
|
||||
outside: c.Outside,
|
||||
@@ -303,7 +305,7 @@ func (f *Interface) listenOut(i int) {
|
||||
li = f.outside
|
||||
}
|
||||
|
||||
ctCache := firewall.NewConntrackCacheTicker(f.conntrackCacheTimeout)
|
||||
ctCache := firewall.NewConntrackCacheTicker(f.ctx, f.conntrackCacheTimeout)
|
||||
lhh := f.lightHouse.NewRequestHandler()
|
||||
plaintext := make([]byte, udp.MTU)
|
||||
h := &header.H{}
|
||||
@@ -328,7 +330,7 @@ func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) {
|
||||
fwPacket := &firewall.Packet{}
|
||||
nb := make([]byte, 12, 12)
|
||||
|
||||
conntrackCache := firewall.NewConntrackCacheTicker(f.conntrackCacheTimeout)
|
||||
conntrackCache := firewall.NewConntrackCacheTicker(f.ctx, f.conntrackCacheTimeout)
|
||||
|
||||
for {
|
||||
n, err := reader.Read(packet)
|
||||
|
||||
Reference in New Issue
Block a user