[netflow] Ensure custom fields configuration is applied (#40730) (#40739

)

- Fix custom fields not being applied when provided through configuration
- Add regression test

(cherry picked from commit f9b41f2)

Co-authored-by: Taylor Swanson <90622908+taylor-swanson@users.noreply.github.com>

CHANGELOG.next.asciidoc

@@ -97,6 +97,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - 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]
 - 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]
 
 *Heartbeat*
 

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

@@ -66,6 +66,7 @@ func (im *netflowInputManager) Create(cfg *conf.C) (v2.Input, error) {
 
 	input := &netflowInput{
 		cfg:              inputCfg,
+		customFields:     customFields,
 		internalNetworks: inputCfg.InternalNetworks,
 		logger:           im.log,
 		queueSize:        inputCfg.PacketQueueSize,

x-pack/filebeat/input/netflow/input_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 
 	"github.com/elastic/beats/v7/libbeat/tests/resources"
+	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/fields"
 
 	conf "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
@@ -29,3 +30,24 @@ func TestNewInputDone(t *testing.T) {
 	_, err = Plugin(logp.NewLogger("netflow_test")).Manager.Create(config)
 	require.NoError(t, err)
 }
+
+func TestConfig_CustomDefinitions(t *testing.T) {
+	goroutines := resources.NewGoroutinesChecker()
+	defer goroutines.Check(t)
+
+	wantDefinitions, err := LoadFieldDefinitionsFromFile("testdata/fields/netflow9_cisco_asa_custom.yaml")
+	require.NoError(t, err)
+	want := []fields.FieldDict{wantDefinitions}
+
+	config, err := conf.NewConfigFrom(mapstr.M{
+		"custom_definitions": []string{"testdata/fields/netflow9_cisco_asa_custom.yaml"},
+	})
+	require.NoError(t, err)
+
+	v2input, err := Plugin(logp.NewLogger("netflow_test")).Manager.Create(config)
+	require.NoError(t, err)
+
+	input := v2input.(*netflowInput)
+
+	require.EqualValues(t, input.customFields, want)
+}