Unify shutdown timeout (#4107)

This pull request aims to unify shutdown timeout.
For now, this is must be implemented in each beat Run method due to specific code in filebeat.

Close #3588

CHANGELOG.asciidoc

@@ -111,6 +111,7 @@ https://github.com/elastic/beats/compare/v5.4.0...v6.0.0-alpha1[View commits]
 *Packetbeat*
 
 - Remove deprecated geoip. {pull}3766[3766]
+- Replace `waitstop` command line argument by `shutdown_timeout` in configuration file. {pull}3588[3588]
 
 *Winlogbeat*
 

packetbeat/beater/packetbeat.go

@@ -40,24 +40,22 @@ type packetbeat struct {
 }
 
 type flags struct {
-	file         *string
-	loop         *int
-	oneAtAtime   *bool
-	topSpeed     *bool
-	dumpfile     *string
-	waitShutdown *int
+	file       *string
+	loop       *int
+	oneAtAtime *bool
+	topSpeed   *bool
+	dumpfile   *string
 }
 
 var cmdLineArgs flags
 
 func init() {
 	cmdLineArgs = flags{
-		file:         flag.String("I", "", "Read packet data from specified file"),
-		loop:         flag.Int("l", 1, "Loop file. 0 - loop forever"),
-		oneAtAtime:   flag.Bool("O", false, "Read packets one at a time (press Enter)"),
-		topSpeed:     flag.Bool("t", false, "Read packets as fast as possible, without sleeping"),
-		dumpfile:     flag.String("dump", "", "Write all captured packets to this libpcap file"),
-		waitShutdown: flag.Int("waitstop", 0, "Additional seconds to wait before shutting down"),
+		file:       flag.String("I", "", "Read packet data from specified file"),
+		loop:       flag.Int("l", 1, "Loop file. 0 - loop forever"),
+		oneAtAtime: flag.Bool("O", false, "Read packets one at a time (press Enter)"),
+		topSpeed:   flag.Bool("t", false, "Read packets as fast as possible, without sleeping"),
+		dumpfile:   flag.String("dump", "", "Write all captured packets to this libpcap file"),
 	}
 }
 
@@ -172,9 +170,9 @@ func (pb *packetbeat) Run(b *beat.Beat) error {
 		service.Stop()
 	}
 
-	waitShutdown := pb.cmdLineArgs.waitShutdown
-	if waitShutdown != nil && *waitShutdown > 0 {
-		time.Sleep(time.Duration(*waitShutdown) * time.Second)
+	timeout := pb.config.ShutdownTimeout
+	if timeout > 0 {
+		time.Sleep(timeout)
 	}
 
 	return nil

packetbeat/config/config.go

@@ -9,13 +9,14 @@ import (
 )
 
 type Config struct {
-	Interfaces     InterfacesConfig          `config:"interfaces"`
-	Flows          *Flows                    `config:"flows"`
-	Protocols      map[string]*common.Config `config:"protocols"`
-	ProtocolsList  []*common.Config          `config:"protocols"`
-	Procs          procs.ProcsConfig         `config:"procs"`
-	IgnoreOutgoing bool                      `config:"ignore_outgoing"`
-	RunOptions     droppriv.RunOptions
+	Interfaces      InterfacesConfig          `config:"interfaces"`
+	Flows           *Flows                    `config:"flows"`
+	Protocols       map[string]*common.Config `config:"protocols"`
+	ProtocolsList   []*common.Config          `config:"protocols"`
+	Procs           procs.ProcsConfig         `config:"procs"`
+	IgnoreOutgoing  bool                      `config:"ignore_outgoing"`
+	ShutdownTimeout time.Duration             `config:"shutdown_timeout"`
+	RunOptions      droppriv.RunOptions
 }
 
 type InterfacesConfig struct {

packetbeat/docs/command-line.asciidoc

@@ -2,7 +2,7 @@
 === Command Line Options
 
 The following command line options are available for Packetbeat. To use these options,
-you need to start Packetbeat in the foreground. 
+you need to start Packetbeat in the foreground.
 
 TIP: Run `./packetbeat -h` to see the full list of options from the command line.
 
@@ -29,9 +29,6 @@ For an infinite loop, use _0_. The `-l` option is useful only for testing Packet
 *`-t`*::
 Read the packets from the pcap file as fast as possible without sleeping. Use this option in combination with the `-I` option. The `-t` option is useful only for testing Packetbeat.
 
-*`-waitstop <n>`*::
-Wait an additional `n` seconds before exiting.
-
 ==== Other Options
 
 These command line options from libbeat are also available for Packetbeat:

packetbeat/docs/reference/configuration/packetbeat-options.asciidoc

@@ -832,6 +832,23 @@ periodically afterwards, it scans the process table for
 processes that match the values specified for this option. The match is done against the
 process' command line as read from `/proc/<pid>/cmdline`.
 
+
+[[shutdown-timeout]]
+===== shutdown_timeout
+
+How long Packetbeat waits on shutdown. By default, this option is disabled. Packetbeat 
+will wait for `shutdown_timeout` and then close. It will not track if all events were sent
+previously.
+
+Example configuration:
+
+[source,yaml]
+-------------------------------------------------------------------------------------
+packetbeat.shutdown_timeout: 5s
+-------------------------------------------------------------------------------------
+
+
+
 include::../../../../libbeat/docs/generalconfig.asciidoc[]
 
 include::../../../../libbeat/docs/processors-config.asciidoc[]

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

@@ -168,6 +168,8 @@ tags: [
   {%- endif -%}
 ]
 
+packetbeat.shutdown_timeout: {{ shutdown_timeout|default(0) }}
+
 {%- if processors %}
 
 #================================ Filters =====================================

packetbeat/tests/system/packetbeat.py

@@ -29,8 +29,7 @@ def run_packetbeat(self, pcap,
                        output="packetbeat.log",
                        extra_args=[],
                        debug_selectors=[],
-                       exit_code=0,
-                       wait_stop=0):
+                       exit_code=0):
         """
         Executes packetbeat on an input pcap file.
         Waits for the process to finish before returning to
@@ -49,7 +48,6 @@ def run_packetbeat(self, pcap,
             "-t",
             "-systemTest",
             "-test.coverprofile", os.path.join(self.working_dir, "coverage.cov"),
-            "-waitstop", str(wait_stop),
         ])
 
         if extra_args:

packetbeat/tests/system/test_0041_memcache_udp_bin_basic.py

@@ -17,7 +17,6 @@ def _run(self, pcap):
             memcache_udp_transaction_timeout=10
         )
         self.run_packetbeat(pcap=pcap,
-                            extra_args=['-waitstop', '1'],
                             debug_selectors=["memcache", "udp", "publish"])
         objs = self.read_output()
         self.assert_common(objs)

packetbeat/tests/system/test_0041_memcache_udp_text_basic.py

@@ -17,7 +17,6 @@ def _run(self, pcap):
             memcache_udp_transaction_timeout=10
         )
         self.run_packetbeat(pcap=pcap,
-                            extra_args=['-waitstop', '1'],
                             debug_selectors=["memcache", "udp", "publish"])
         objs = self.read_output()
         self.assert_common(objs)

packetbeat/tests/system/test_0060_flows.py

@@ -16,10 +16,10 @@ class Test(BaseTest):
     def test_mysql_flow(self):
         self.render_config_template(
             flows=True,
+            shutdown_timeout="1s",
         )
         self.run_packetbeat(
             pcap="mysql_long.pcap",
-            wait_stop=1,
             debug_selectors=["*"])
 
         objs = self.read_output(
@@ -46,10 +46,10 @@ def test_mysql_flow(self):
     def test_memcache_udp_flow(self):
         self.render_config_template(
             flows=True,
+            shutdown_timeout="1s",
         )
         self.run_packetbeat(
             pcap="memcache/memcache_bin_udp_counter_ops.pcap",
-            wait_stop=1,
             debug_selectors=["*"])
 
         objs = self.read_output(
@@ -74,10 +74,10 @@ def test_memcache_udp_flow(self):
     def test_icmp4_ping(self):
         self.render_config_template(
             flows=True,
+            shutdown_timeout="1s",
         )
         self.run_packetbeat(
             pcap="icmp/icmp4_ping_over_vlan.pcap",
-            wait_stop=1,
             debug_selectors=["*"])
 
         objs = self.read_output(
@@ -103,10 +103,10 @@ def test_icmp4_ping(self):
     def test_icmp6_ping(self):
         self.render_config_template(
             flows=True,
+            shutdown_timeout="1s",
         )
         self.run_packetbeat(
             pcap="icmp/icmp6_ping_over_vlan.pcap",
-            wait_stop=1,
             debug_selectors=["*"])
 
         objs = self.read_output(