[libbeat] Allow per beat.Client control of event normalization

CHANGELOG-developer.next.asciidoc

@@ -147,6 +147,7 @@ The list below covers the major changes between 7.0.0-rc2 and main only.
 - Added `.python-version` file {pull}32323[32323]
 - Add support for multiple regions in GCP {pull}32964[32964]
 - Use `T.TempDir` to create temporary test directory {pull}33082[33082]
+- Add an option to disable event normalization when creating a `beat.Client`. {pull}33657[33657]
 
 ==== Deprecated
 

libbeat/beat/pipeline.go

@@ -121,6 +121,10 @@ type ProcessingConfig struct {
 	// Disables the addition of host.name if it was enabled for the publisher.
 	DisableHost bool
 
+	// EventNormalization controls whether the event normalization processor
+	// is applied to events. If nil the Beat's default behavior prevails.
+	EventNormalization *bool
+
 	// Private contains additional information to be passed to the processing
 	// pipeline builder.
 	Private interface{}

libbeat/publisher/processing/default.go

@@ -288,8 +288,12 @@ func (b *builder) Create(cfg beat.ProcessingConfig, drop bool) (beat.Processor,
 		builtin = tmp
 	}
 
-	if !b.skipNormalize {
-		// setup 1: generalize/normalize output (P)
+	// setup 1: generalize/normalize output (P)
+	if cfg.EventNormalization != nil {
+		if *cfg.EventNormalization {
+			processors.add(newGeneralizeProcessor(cfg.KeepNull))
+		}
+	} else if !b.skipNormalize {
 		processors.add(newGeneralizeProcessor(cfg.KeepNull))
 	}
 

libbeat/publisher/processing/default_test.go

@@ -280,6 +280,43 @@ func TestProcessorsConfigs(t *testing.T) {
 	}
 }
 
+// TestEventNormalizationOverride verifies that the EventNormalization option
+// in beat.ProcessingConfig overrides the "skipNormalize" setting that is
+// specified in the builder (this is the default value set by the Beat).
+func TestEventNormalizationOverride(t *testing.T) {
+	boolPtr := func(b bool) *bool { return &b }
+
+	testCases := []struct {
+		skipNormalize          bool
+		normalizeOverride      *bool
+		hasGeneralizeProcessor bool
+	}{
+		{false, nil, true},
+		{false, boolPtr(false), false},
+		{false, boolPtr(true), true},
+		{true, nil, false},
+		{true, boolPtr(false), false},
+		{true, boolPtr(true), true},
+	}
+
+	for _, tc := range testCases {
+		builder, err := newBuilder(beat.Info{}, logp.NewLogger(""), nil, mapstr.EventMetadata{}, nil, tc.skipNormalize, false)
+		require.NoError(t, err)
+
+		processor, err := builder.Create(beat.ProcessingConfig{EventNormalization: tc.normalizeOverride}, false)
+		require.NoError(t, err)
+		group := processor.(*group)
+
+		if tc.hasGeneralizeProcessor {
+			if assert.NotEmpty(t, group.list) {
+				assert.Equal(t, "generalizeEvent", group.list[0].String())
+			}
+		} else {
+			assert.Empty(t, group.list)
+		}
+	}
+}
+
 func TestNormalization(t *testing.T) {
 	cases := map[string]struct {
 		normalize bool