[filebeat][netflow]: fix template sharing (#42079)

Pass the share_templates configuration option into the NetflowV9Protocol
struct. The parameter was not being set, and therefore was always false so
it was not possible to use this option.

Added a test case to prevent future regressions.

Closes #42080

(cherry picked from commit 323c69e)

# Conflicts:
#	x-pack/filebeat/input/netflow/decoder/v9/v9.go

CHANGELOG.next.asciidoc

@@ -46,6 +46,72 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 *Filebeat*
 
 - Fix handling of un-parsed JSON in O365 module. {issue}37800[37800] {pull}38709[38709]
+- [Gcs Input] - Added missing locks for safe concurrency {pull}34914[34914]
+- Fix the ignore_inactive option being ignored in Filebeat's filestream input {pull}34770[34770]
+- Fix TestMultiEventForEOFRetryHandlerInput unit test of CometD input {pull}34903[34903]
+- Add input instance id to request trace filename for httpjson and cel inputs {pull}35024[35024]
+- Fixes "Can only start an input when all related states are finished" error when running under Elastic-Agent {pull}35250[35250] {issue}33653[33653]
+- [system] sync system/auth dataset with system integration 1.29.0. {pull}35581[35581]
+- [GCS Input] - Fixed an issue where bucket_timeout was being applied to the entire bucket poll interval and not individual bucket object read operations. Fixed a map write concurrency issue arising from data races when using a high number of workers. Fixed the flaky tests that were present in the GCS test suit. {pull}35605[35605]
+- Fixed concurrency and flakey tests issue in azure blob storage input. {issue}35983[35983] {pull}36124[36124]
+- Fix panic when sqs input metrics getter is invoked {pull}36101[36101] {issue}36077[36077]
+- Fix handling of Juniper SRX structured data when there is no leading junos element. {issue}36270[36270] {pull}36308[36308]
+- Fix Filebeat Cisco module with missing escape character {issue}36325[36325] {pull}36326[36326]
+- Added a fix for Crowdstrike pipeline handling process arrays {pull}36496[36496]
+- [threatintel] MISP pagination fixes {pull}37898[37898]
+- Fix file handle leak when handling errors in filestream {pull}37973[37973]
+- Fix a race condition that could crash Filebeat with a "negative WaitGroup counter" error {pull}38094[38094]
+- Fix "failed processing S3 event for object key" error on aws-s3 input when key contains the "+" character {issue}38012[38012] {pull}38125[38125]
+- Fix filebeat gcs input panic {pull}38407[38407]
+- Fix filestream's registry GC: registry entries are now removed from the in-memory and disk store when they're older than the set TTL {issue}36761[36761] {pull}38488[38488]
+- Fix filestream's registry GC: registry entries are now removed from the in-memory and disk store when they're older than the set TTL {issue}36761[36761] {pull}38488[38488]
+- [threatintel] MISP splitting fix for empty responses {issue}38739[38739] {pull}38917[38917]
+- Prevent GCP Pub/Sub input blockage by increasing default value of `max_outstanding_messages` {issue}35029[35029] {pull}38985[38985]
+- Updated Websocket input title to align with existing inputs {pull}39006[39006]
+- Restore netflow input on Windows {pull}39024[39024]
+- Upgrade azure-event-hubs-go and azure-storage-blob-go dependencies. {pull}38861[38861]
+- Fix request trace filename handling in http_endpoint input. {pull}39410[39410]
+- Upgrade github.com/hashicorp/go-retryablehttp to mitigate CVE-2024-6104 {pull}40036[40036]
+- Fix for Google Workspace duplicate events issue by adding canonical sorting over fingerprint keys array to maintain key order. {pull}40055[40055] {issue}39859[39859]
+- Fix handling of deeply nested numeric values in HTTP Endpoint CEL programs. {pull}40115[40115]
+- Prevent panic in CEL and salesforce inputs when github.com/hashicorp/go-retryablehttp exceeds maximum retries. {pull}40144[40144]
+- Fix bug in CEL input rate limit logic. {issue}40106[40106] {pull}40270[40270]
+- Relax requirements in Okta entity analytics provider user and device profile data shape. {pull}40359[40359]
+- Fix bug in Okta entity analytics rate limit logic. {issue}40106[40106] {pull}40267[40267]
+- Fix crashes in the journald input. {pull}40061[40061]
+- Fix order of configuration for EntraID entity analytics provider. {pull}40487[40487]
+- Ensure Entra ID request bodies are not truncated and trace logs are rotated before 100MB. {pull}40494[40494]
+- The Elasticsearch output now correctly logs the event fields to the event log file {issue}40509[40509] {pull}40512[40512]
+- Fix the "No such input type exist: 'azure-eventhub'" error on the Windows platform {issue}40608[40608] {pull}40609[40609]
+- awss3 input: Fix handling of SQS notifications that don't contain a region. {pull}40628[40628]
+- Fix credential handling when workload identity is being used in GCS input. {issue}39977[39977] {pull}40663[40663]
+- Fix publication of group data from the Okta entity analytics provider. {pull}40681[40681]
+- Ensure netflow custom field configuration is applied. {issue}40735[40735] {pull}40730[40730]
+- Fix replace processor handling of zero string replacement validation. {pull}40751[40751]
+- Fix long filepaths in diagnostics exceeding max path limits on Windows. {pull}40909[40909]
+- Add backup and delete for AWS S3 polling mode feature back. {pull}41071[41071]
+- Fix a bug in Salesforce input to only handle responses with 200 status code {pull}41015[41015]
+- Fixed failed job handling and removed false-positive error logs in the GCS input. {pull}41142[41142]
+- Bump github.com/elastic/go-sfdc dependency used by x-pack/filebeat/input/salesforce. {pull}41192[41192]
+- Log bad handshake details when websocket connection fails {pull}41300[41300]
+- Improve modification time handling for entities and entity deletion logic in the Active Directory entityanalytics input. {pull}41179[41179]
+- Journald input now can read events from all boots {issue}41083[41083] {pull}41244[41244]
+- Fix double encoding of client_secret in the Entity Analytics input's Azure Active Directory provider {pull}41393[41393]
+- Fix aws region in aws-s3 input s3 polling mode.  {pull}41572[41572]
+- Fix errors in SQS host resolution in the `aws-s3` input when using custom (non-AWS) endpoints. {pull}41504[41504]
+- Fix double encoding of client_secret in the Entity Analytics input's Azure Active Directory provider {pull}41393[41393]
+- The azure-eventhub input now correctly reports its status to the Elastic Agent on fatal errors {pull}41469[41469]
+- Add support for Access Points in the `aws-s3` input. {pull}41495[41495]
+- Fix the "No such input type exist: 'salesforce'" error on the Windows/AIX platform. {pull}41664[41664]
+- Fix missing key in streaming input logging. {pull}41600[41600]
+- Improve S3 object size metric calculation to support situations where Content-Length is not available. {pull}41755[41755]
+- Fix handling of http_endpoint request exceeding memory limits. {issue}41764[41764] {pull}41765[41765]
+- Rate limiting fixes in the Okta provider of the Entity Analytics input. {issue}40106[40106] {pull}41583[41583]
+- Redact authorization headers in HTTPJSON debug logs. {pull}41920[41920]
+- Further rate limiting fix in the Okta provider of the Entity Analytics input. {issue}40106[40106] {pull}41977[41977]
+- Fix streaming input handling of invalid or empty websocket messages. {pull}42036[42036]
+- Fix awss3 document ID construction when using the CSV decoder. {pull}42019[42019]
+- Fix Netflow Template Sharing configuration handling. {pull}42080[42080]
 
 *Heartbeat*
 

x-pack/filebeat/input/netflow/decoder/config/config.go

@@ -102,6 +102,11 @@ func (c *Config) SequenceResetEnabled() bool {
 	return c.detectReset
 }
 
+// ShareTemplatesEnabled returns if template sharing is enabled.
+func (c *Config) ShareTemplatesEnabled() bool {
+	return c.sharedTemplates
+}
+
 // Fields returns the configured fields.
 func (c *Config) Fields() fields.FieldDict {
 	if c.fields == nil {

x-pack/filebeat/input/netflow/decoder/v9/v9.go

@@ -43,12 +43,25 @@ func New(config config.Config) protocol.Protocol {
 }
 
 func NewProtocolWithDecoder(decoder Decoder, config config.Config, logger *log.Logger) *NetflowV9Protocol {
+<<<<<<< HEAD
 	return &NetflowV9Protocol{
 		decoder:     decoder,
 		Session:     NewSessionMap(logger),
 		logger:      logger,
 		timeout:     config.ExpirationTimeout(),
 		detectReset: config.SequenceResetEnabled(),
+=======
+	ctx, cancel := context.WithCancel(context.Background())
+	pd := &NetflowV9Protocol{
+		ctx:            ctx,
+		cancel:         cancel,
+		decoder:        decoder,
+		logger:         logger,
+		Session:        NewSessionMap(logger, config.ActiveSessionsMetric()),
+		timeout:        config.ExpirationTimeout(),
+		detectReset:    config.SequenceResetEnabled(),
+		shareTemplates: config.ShareTemplatesEnabled(),
+>>>>>>> 323c69eb7 ([filebeat][netflow]: fix template sharing (#42079))
 	}
 }
 

x-pack/filebeat/input/netflow/decoder/v9/v9_test.go

@@ -249,3 +249,53 @@ func TestCustomFields(t *testing.T) {
 	assert.Contains(t, flows[0].Fields, "customField")
 	assert.Equal(t, flows[0].Fields["customField"], "Hello :)")
 }
+
+func TestSharedTemplates(t *testing.T) {
+	templateAddr := test.MakeAddress(t, "127.0.0.1:12345")
+	flowsAddr := test.MakeAddress(t, "127.0.0.2:21234")
+	templatePacket := []uint16{
+		// Header
+		// Version, Count, Uptime, Ts, SeqNo, Source
+		9, 1, 11, 11, 22, 22, 33, 33, 0, 1234,
+		// Set #1 (template)
+		0, 20, /*len of set*/
+		999, 3, /*len*/
+		1, 4, // Fields
+		2, 4,
+		3, 4,
+	}
+	flowsPacket := []uint16{
+		// Header
+		// Version, Count, Uptime, Ts, SeqNo, Source
+		9, 1, 11, 11, 22, 22, 33, 34, 0, 1234,
+		// Set #1 (template)
+		999, 16, /*len of set*/
+		1, 1,
+		2, 2,
+		3, 3,
+	}
+
+	t.Run("Template sharing enabled", func(t *testing.T) {
+		cfg := config.Defaults()
+		cfg.WithSharedTemplates(true)
+		proto := New(cfg)
+		flows, err := proto.OnPacket(test.MakePacket(templatePacket), templateAddr)
+		assert.NoError(t, err)
+		assert.Empty(t, flows)
+		flows, err = proto.OnPacket(test.MakePacket(flowsPacket), flowsAddr)
+		assert.NoError(t, err)
+		assert.Len(t, flows, 1)
+	})
+
+	t.Run("Template sharing disabled", func(t *testing.T) {
+		cfg := config.Defaults()
+		cfg.WithSharedTemplates(false)
+		proto := New(cfg)
+		flows, err := proto.OnPacket(test.MakePacket(templatePacket), templateAddr)
+		assert.NoError(t, err)
+		assert.Empty(t, flows)
+		flows, err = proto.OnPacket(test.MakePacket(flowsPacket), flowsAddr)
+		assert.NoError(t, err)
+		assert.Empty(t, flows)
+	})
+}