diff --git a/libbeat/beat/beat.go b/libbeat/beat/beat.go index f58bd5571cbc..eb51c4e06713 100644 --- a/libbeat/beat/beat.go +++ b/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 { diff --git a/packetbeat/beater/packetbeat.go b/packetbeat/beater/packetbeat.go index 4d8e4605843c..495cff95da49 100644 --- a/packetbeat/beater/packetbeat.go +++ b/packetbeat/beater/packetbeat.go @@ -97,6 +97,13 @@ func New(b *beat.Beat, rawConfig *conf.C) (beat.Beater, error) { 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 diff --git a/packetbeat/beater/processor.go b/packetbeat/beater/processor.go index b1db9a4f1a51..72ea1e8fc626 100644 --- a/packetbeat/beater/processor.go +++ b/packetbeat/beater/processor.go @@ -49,23 +49,23 @@ 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 - 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, } } @@ -99,11 +99,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") } @@ -138,11 +140,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. @@ -218,7 +220,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, @@ -318,7 +320,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 { diff --git a/packetbeat/config/config.go b/packetbeat/config/config.go index d0d0cf3439e1..42570563002f 100644 --- a/packetbeat/config/config.go +++ b/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 diff --git a/packetbeat/tests/system/config/packetbeat.yml.j2 b/packetbeat/tests/system/config/packetbeat.yml.j2 index 88390d3a98db..6595c01b9a69 100644 --- a/packetbeat/tests/system/config/packetbeat.yml.j2 +++ b/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: diff --git a/packetbeat/tests/system/test_0060_flows.py b/packetbeat/tests/system/test_0060_flows.py index e1d262b439e3..446841aa69bb 100644 --- a/packetbeat/tests/system/test_0060_flows.py +++ b/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",