packetbeat/sniffer: fix shutdown logic

CHANGELOG.next.asciidoc

@@ -97,6 +97,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 *Packetbeat*
 
 - Fix panic on memcache transaction with no request or response. {issue}33852[33852] {pull}33853[33853]
+- Fix termination logic. {pull}33979[33979]
 
 *Winlogbeat*
 

packetbeat/sniffer/sniffer.go

@@ -44,13 +44,13 @@ import (
 // to a Worker.
 type Sniffer struct {
 	sniffers []sniffer
+	cancel   func()
 }
 
 type sniffer struct {
 	config config.InterfaceConfig
 
-	state atomic.Int32  // store snifferState
-	done  chan struct{} // done is required to wire state into a select.
+	state atomic.Int32 // store snifferState
 
 	// device is the first active device after calling New.
 	// It is not updated by default route polling.
@@ -184,16 +184,17 @@ func validateAfPacketConfig(cfg *config.InterfaceConfig) error {
 // Run opens the sniffing device and processes packets being read from that device.
 // Worker instances are instantiated as needed.
 func (s *Sniffer) Run() error {
-	g, ctx := errgroup.WithContext(context.Background())
-	for _, c := range s.sniffers {
-		c := c
+	ctx, cancel := context.WithCancel(context.Background())
+	s.cancel = cancel
+	g, ctx := errgroup.WithContext(ctx)
+	for i := range s.sniffers {
+		c := &s.sniffers[i]
 		g.Go(func() error {
 			var (
 				defaultRoute chan string
 				refresh      chan struct{}
 			)
 			if c.followDefault {
-				c.done = make(chan struct{})
 				defaultRoute = make(chan string)
 				refresh = make(chan struct{}, 1)
 				go c.pollDefaultRoute(ctx, defaultRoute, refresh)
@@ -208,8 +209,8 @@ func (s *Sniffer) Run() error {
 }
 
 // pollDefaultRoute repeatedly polls the default route's device at intervals
-// specified in config.PollDefaultRoute. The poller is terminated by closing
-// done and the device chan can be read for changes in the default route.
+// specified in config.PollDefaultRoute. The poller is terminated by cancelling
+// the context and the device chan can be read for changes in the default route.
 // Changes in default route will put the Sniffer into the inactive state to
 // trigger a new sniffer connection. Termination of the sniffer is not under
 // the control of the poller.
@@ -232,11 +233,6 @@ func (s *sniffer) pollDefaultRoute(ctx context.Context, device chan<- string, re
 				logp.Debug("sniffer", "requested new default route")
 				current = s.poll(current, device)
 			case <-ctx.Done():
-				logp.Info("polling cancelled")
-				close(device)
-				tick.Stop()
-				return
-			case <-s.done:
 				logp.Info("closing default route poller")
 				close(device)
 				tick.Stop()
@@ -459,11 +455,14 @@ func (s *sniffer) open(device string) (snifferHandle, error) {
 // Stop marks a sniffer as stopped. The Run method will return once the stop
 // signal has been given.
 func (s *Sniffer) Stop() {
+	logp.Debug("sniffer", "sending stop to all sniffers")
 	for _, c := range s.sniffers {
+		logp.Debug("sniffer", "sending closing to %s", c.config.Device)
 		c.state.Store(snifferClosing)
-		if c.done != nil {
-			close(c.done)
-		}
+	}
+	if s.cancel != nil {
+		logp.Debug("sniffer", "cancelling sniffers")
+		s.cancel()
 	}
 }