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

libbeat/beat/beat.go

@@ -18,6 +18,8 @@
 package beat
 
 import (
+	"time"
+
 	"github.com/elastic/beats/v7/libbeat/api"
 	"github.com/elastic/beats/v7/libbeat/beatmonitoring"
 	"github.com/elastic/beats/v7/libbeat/common/reload"
@@ -88,6 +90,12 @@ type Beat struct {
 
 	API      *api.Server      // API server. This is nil unless the http endpoint is enabled.
 	Registry *reload.Registry // input, & output registry for configuration manager, should be instantiated in NewBeat
+
+	// ShutdownTimeout, when set by a Creator, tells the framework how long the
+	// Beater may take to drain in-flight events during shutdown. On this branch
+	// the field is stored for beater use / forward compatibility; full libbeat
+	// watchdog drain support requires the #51136 backport.
+	ShutdownTimeout time.Duration
 }
 
 func (beat *Beat) userAgentMode() useragent.AgentManagementMode {

packetbeat/beater/packetbeat.go

@@ -97,6 +97,13 @@
 		configurator = initialConfig().FromStatic
 	}
 
+	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
@@ -156,7 +163,7 @@
 			"input_metrics.json", "application/json", func() []byte {
 				data, err := inputmon.MetricSnapshotJSON(b.Monitoring.InputsRegistry())
 				if err != nil {
 					logp.L().Warnw("Failed to collect input metric snapshot for Agent diagnostics.", "error", err)
 					return []byte(err.Error())
 				}
 				return data
@@ -173,7 +180,7 @@
 				return err
 			}
 		} else {
 			logp.L().Warn(pipelinesWarning)
 		}
 
 		return pb.runStatic(b, pb.factory)
@@ -216,7 +223,7 @@
 
 	// Start the manager after all the hooks are registered and terminates when
 	// the function return.
 	if err := b.Manager.Start(); err != nil {
 		return err
 	}
 

packetbeat/beater/processor.go

@@ -49,23 +49,23 @@
 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
-	status          status.StatusReporter
+	wg             sync.WaitGroup
+	publisher      *publish.TransactionPublisher
+	flows          *flows.Flows
+	sniffer        *sniffer.Sniffer
+	err            chan error
+	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,
 	}
 }
 
@@ -78,7 +78,7 @@
 		p.flows.Start()
 	}
 	p.wg.Add(1)
 	go func() {
 		defer p.wg.Done()
 
 		p.UpdateStatus(status.Running, "running packetbeat processor")
@@ -99,11 +99,13 @@
 		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")
 }
@@ -138,11 +140,11 @@
 		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.
@@ -218,7 +220,7 @@
 		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,
@@ -318,7 +320,7 @@
 		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

@@ -42,6 +42,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",