[9.5](backport #52005) [beatreceiver][packetbeat] Delegate shutdown_timeout logic to libbeat

packetbeat/beater/packetbeat.go

@@ -103,6 +103,13 @@ func New(b *beat.Beat, rawConfig *conf.C) (beat.Beater, error) {
 
 	logger := b.Info.Logger
 
+	timeout, err := config.GetShutDownTimeOut(rawConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	b.ShutdownTimeout = timeout
+
 	factory := newProcessorFactory(b.Info.Name, make(chan error, maxSniffers), b, configurator)
 	if err := factory.CheckConfig(rawConfig); err != nil {
 		return nil, err

packetbeat/beater/processor.go

@@ -49,24 +49,24 @@ type noopReporter struct{}
 func (noopReporter) UpdateStatus(status.Status, string) {}
 
 type processor struct {
-	wg              sync.WaitGroup
-	publisher       *publish.TransactionPublisher
-	flows           *flows.Flows
-	sniffer         *sniffer.Sniffer
-	shutdownTimeout time.Duration
-	err             chan error
-	statusMu        sync.RWMutex
-	status          status.StatusReporter
+	wg             sync.WaitGroup
+	publisher      *publish.TransactionPublisher
+	flows          *flows.Flows
+	sniffer        *sniffer.Sniffer
+	err            chan error
+	statusMu       sync.RWMutex
+	status         status.StatusReporter
+	publishTimeout time.Duration
 }
 
-func newProcessor(shutdownTimeout time.Duration, publisher *publish.TransactionPublisher, flows *flows.Flows, sniffer *sniffer.Sniffer, err chan error, status status.StatusReporter) *processor {
+func newProcessor(publishTimeout time.Duration, publisher *publish.TransactionPublisher, flows *flows.Flows, sniffer *sniffer.Sniffer, err chan error, status status.StatusReporter) *processor {
 	return &processor{
-		publisher:       publisher,
-		flows:           flows,
-		sniffer:         sniffer,
-		err:             err,
-		shutdownTimeout: shutdownTimeout,
-		status:          status,
+		publisher:      publisher,
+		flows:          flows,
+		sniffer:        sniffer,
+		err:            err,
+		status:         status,
+		publishTimeout: publishTimeout,
 	}
 }
 
@@ -79,7 +79,7 @@ func (p *processor) Start() {
 		p.flows.Start()
 	}
 	p.wg.Add(1)
-	go func() {
+	p.wg.Go(func() {
 		defer p.wg.Done()
 
 		p.UpdateStatus(status.Running, "running packetbeat processor")
@@ -90,7 +90,7 @@ func (p *processor) Start() {
 			return
 		}
 		p.err <- nil
-	}()
+	})
 }
 
 func (p *processor) Stop() {
@@ -100,11 +100,13 @@ func (p *processor) Stop() {
 		p.flows.Stop()
 	}
 	p.wg.Wait()
-	// wait for shutdownTimeout to let the publisher flush
+
+	// wait for publish timeout to let the publisher flush
 	// whatever pending events
-	if p.shutdownTimeout > 0 {
-		time.Sleep(p.shutdownTimeout)
+	if p.publishTimeout > 0 {
+		time.Sleep(p.publishTimeout)
 	}
+
 	p.publisher.Stop()
 	p.UpdateStatus(status.Stopped, "stopped packetbeat processor")
 }
@@ -152,11 +154,11 @@ func (p *processorFactory) CreateWithReporter(pipeline beat.PipelineConnector, c
 		statusReporter = noopReporter{}
 	}
 	statusReporter.UpdateStatus(status.Configuring, "starting packetbeat processor configuration")
-	duration, publisher, flows, sniffer, errChan, err := p.create(pipeline, cfg, statusReporter)
+	publishTimeout, publisher, flows, sniffer, errChan, err := p.create(pipeline, cfg, statusReporter)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create packetbeat processor: %w", err)
 	}
-	return newProcessor(duration, publisher, flows, sniffer, errChan, statusReporter), nil
+	return newProcessor(publishTimeout, publisher, flows, sniffer, errChan, statusReporter), nil
 }
 
 // Create returns a new module runner that publishes to the provided pipeline, configured from cfg.
@@ -233,7 +235,7 @@ func (p *processorFactory) create(pipeline beat.PipelineConnector, cfg *conf.C,
 		return 0, nil, nil, nil, nil, err
 	}
 
-	return config.ShutdownTimeout, publisher, flows, sniffer, p.err, nil
+	return config.PublishTimeout, publisher, flows, sniffer, p.err, nil
 }
 
 // setupFlows returns a *flows.Flows that will publish to the provided pipeline,
@@ -341,7 +343,7 @@ func configID(config *conf.C) (string, error) {
 		return tmp.ID, nil
 	}
 
-	var h map[string]interface{}
+	var h map[string]any
 	_ = config.Unpack(&h)
 	id, err := hashstructure.Hash(h, nil)
 	if err != nil {

packetbeat/config/config.go

@@ -43,6 +43,27 @@ type Config struct {
 	IgnoreOutgoing     bool               `config:"ignore_outgoing"`
 	ShutdownTimeout    time.Duration      `config:"shutdown_timeout"`
 	OverwritePipelines bool               `config:"overwrite_pipelines"` // Only used by standalone Packetbeat.
+
+	// For internal use only
+	PublishTimeout time.Duration `config:"publish_timeout"`
+}
+
+func GetShutDownTimeOut(cfg *conf.C) (time.Duration, error) {
+	timeout := struct {
+		ShutdownTimeout time.Duration `config:"shutdown_timeout"`
+	}{
+		ShutdownTimeout: 0,
+	}
+
+	if err := cfg.Unpack(&timeout); err != nil {
+		return 0, fmt.Errorf("error reading shutdown_timeout: %w", err)
+	}
+
+	if timeout.ShutdownTimeout <= 0 {
+		timeout.ShutdownTimeout = 0
+	}
+	return timeout.ShutdownTimeout, nil
+
 }
 
 // FromStatic initializes a configuration given a config.C

packetbeat/tests/system/config/packetbeat.yml.j2

@@ -197,7 +197,7 @@ tags: [
   {%- endif -%}
 ]
 
-packetbeat.shutdown_timeout: {{ shutdown_timeout|default('400ms') }}
+packetbeat.publish_timeout: {{ publish_timeout|default('400ms') }}
 
 {%- if include_mime %}
 processors:

packetbeat/tests/system/test_0060_flows.py

@@ -29,7 +29,7 @@ class Test(BaseTest):
     def test_mysql_flow(self):
         self.render_config_template(
             flows=True,
-            shutdown_timeout="1s",
+            publish_timeout="1s",
         )
         self.run_packetbeat(
             pcap="mysql_long.pcap",
@@ -64,7 +64,7 @@ def test_mysql_flow(self):
     def test_memcache_udp_flow(self):
         self.render_config_template(
             flows=True,
-            shutdown_timeout="1s",
+            publish_timeout="1s",
         )
         self.run_packetbeat(
             pcap="memcache/memcache_bin_udp_counter_ops.pcap",
@@ -92,7 +92,7 @@ def test_memcache_udp_flow(self):
     def test_icmp4_ping(self):
         self.render_config_template(
             flows=True,
-            shutdown_timeout="1s",
+            publish_timeout="1s",
         )
         self.run_packetbeat(
             pcap="icmp/icmp4_ping_over_vlan.pcap",
@@ -120,7 +120,7 @@ def test_icmp4_ping(self):
     def test_icmp6_ping(self):
         self.render_config_template(
             flows=True,
-            shutdown_timeout="1s",
+            publish_timeout="1s",
         )
         self.run_packetbeat(
             pcap="icmp/icmp6_ping_over_vlan.pcap",
@@ -150,7 +150,7 @@ def test_icmp6_ping(self):
     def test_q_in_q_flow(self):
         self.render_config_template(
             flows=True,
-            shutdown_timeout="1s",
+            publish_timeout="1s",
         )
         self.run_packetbeat(
             pcap="802.1q-q-in-q-icmp.pcap",
@@ -209,7 +209,7 @@ def test_community_id_ipv4_udp(self):
     def check_community_id(self, pcap):
         self.render_config_template(
             flows=True,
-            shutdown_timeout="1s",
+            publish_timeout="1s",
             processors=[{
                 "drop_event": {
                     "when": "not.equals.type: flow",