[libbeat]: Add support for values in lowercase processor (#41530)

* [libbeat]: Add values to lowercase processor

---------

Co-authored-by: Denis <[email protected]>
(cherry picked from commit c7078ff)

CHANGELOG.next.asciidoc

@@ -236,6 +236,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Update to Go 1.22.7. {pull}41018[41018]
 - Replace Ubuntu 20.04 with 24.04 for Docker base images {issue}40743[40743] {pull}40942[40942]
 - Reduce memory consumption of k8s autodiscovery and the add_kubernetes_metadata processor when Deployment metadata is enabled
+- Add `lowercase` processor. {issue}22254[22254] {pull}41424[41424]
 
 *Auditbeat*
 
@@ -245,7 +246,6 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Add process.entity_id, process.group.name and process.group.id in add_process_metadata processor. Make fim module with kprobes backend to always add an appropriately configured add_process_metadata processor to enrich file events {pull}38776[38776]
 
 *Auditbeat*
-- Add `lowercase` processor. {issue}22254[22254] {pull}41424[41424]
 
 *Auditbeat*
 

libbeat/processors/actions/alterFieldProcessor.go

@@ -30,6 +30,7 @@ import (
 
 type alterFieldProcessor struct {
 	Fields         []string
+	Values         []string
 	IgnoreMissing  bool
 	FailOnError    bool
 	AlterFullField bool
@@ -45,6 +46,7 @@ func NewAlterFieldProcessor(c *conf.C, processorName string, alterFunc mapstr.Al
 		IgnoreMissing  bool     `config:"ignore_missing"`
 		FailOnError    bool     `config:"fail_on_error"`
 		AlterFullField bool     `config:"alter_full_field"`
+		Values         []string `config:"values"`
 	}{
 		IgnoreMissing:  false,
 		FailOnError:    true,
@@ -77,6 +79,7 @@ func NewAlterFieldProcessor(c *conf.C, processorName string, alterFunc mapstr.Al
 		processorName:  processorName,
 		AlterFullField: config.AlterFullField,
 		alterFunc:      alterFunc,
+		Values:         config.Values,
 	}, nil
 
 }
@@ -92,7 +95,7 @@ func (a *alterFieldProcessor) Run(event *beat.Event) (*beat.Event, error) {
 	}
 
 	for _, field := range a.Fields {
-		err := a.alter(event, field)
+		err := a.alterField(event, field)
 		if err != nil {
 			if a.IgnoreMissing && errors.Is(err, mapstr.ErrKeyNotFound) {
 				continue
@@ -105,30 +108,66 @@ func (a *alterFieldProcessor) Run(event *beat.Event) (*beat.Event, error) {
 		}
 	}
 
+	for _, valueKey := range a.Values {
+		err := a.alterValue(event, valueKey)
+		if err != nil {
+			if a.IgnoreMissing && errors.Is(err, mapstr.ErrKeyNotFound) {
+				continue
+			}
+			if a.FailOnError {
+				event = backup
+				_, _ = event.PutValue("error.message", err.Error())
+				return event, err
+			}
+		}
+	}
 	return event, nil
 }
 
-func (a *alterFieldProcessor) alter(event *beat.Event, field string) error {
+func (a *alterFieldProcessor) alterField(event *beat.Event, field string) error {
 
 	// modify all segments of the key
+	var err error
 	if a.AlterFullField {
-		err := event.Fields.AlterPath(field, mapstr.CaseInsensitiveMode, a.alterFunc)
-		if err != nil {
-			return err
-		}
+		err = event.Fields.AlterPath(field, mapstr.CaseInsensitiveMode, a.alterFunc)
 	} else {
 		// modify only the last segment
 		segmentCount := strings.Count(field, ".")
-		err := event.Fields.AlterPath(field, mapstr.CaseInsensitiveMode, func(key string) (string, error) {
+		err = event.Fields.AlterPath(field, mapstr.CaseInsensitiveMode, func(key string) (string, error) {
 			if segmentCount > 0 {
 				segmentCount--
 				return key, nil
 			}
 			return a.alterFunc(key)
 		})
+	}
+
+	return err
+}
+
+func (a *alterFieldProcessor) alterValue(event *beat.Event, valueKey string) error {
+	value, err := event.GetValue(valueKey)
+	if err != nil {
+		return fmt.Errorf("could not fetch value for key: %s, Error: %w", valueKey, err)
+	}
+
+	if v, ok := value.(string); ok {
+		err = event.Delete(valueKey)
 		if err != nil {
-			return err
+			return fmt.Errorf("could not delete key: %s,  %w", v, err)
 		}
+
+		v, err = a.alterFunc(v)
+		if err != nil {
+			return fmt.Errorf("could not alter %s successfully, %w", v, err)
+		}
+
+		_, err = event.PutValue(valueKey, v)
+		if err != nil {
+			return fmt.Errorf("could not put value: %s: %v, %w", valueKey, v, err)
+		}
+	} else {
+		return fmt.Errorf("value of key %q is not a string", valueKey)
 	}
 
 	return nil

libbeat/processors/actions/docs/lowercase.asciidoc

@@ -5,7 +5,7 @@
 <titleabbrev>lowercase</titleabbrev>
 ++++
 
-The `lowercase` processor specifies a list of fields that should be converted to lowercase. This transformation applies to keys that match the specified fields. Matching is performed case-insensitively. 
+The `lowercase` processor specifies a list of `fields` and `values` to be converted to lowercase. Keys listed in `fields` will be matched case-insensitively and converted to lowercase. For `values`, only exact, case-sensitive matches are transformed to lowercase. This way, keys and values can be selectively converted based on the specified criteria.
 
 
 ==== Examples: 
@@ -18,28 +18,32 @@ processors:
   - rename:
       fields:
         - "ab.cd"
+      values:
+        - "testKey"          
       ignore_missing: false
       fail_on_error: true
-      full_path: true
+      alter_full_field: true
 ----
 [source,json]
 ----
 // Input
 {
   "AB": {"CD":"data"},
-  "CD": {"ef":"data"} 
+  "CD": {"ef":"data"},
+  "testKey": {"TESTVALUE"}   
 }
 
 
 // output
 {
   "ab": {"cd":"data"},  // `AB.CD` -> `ab.cd`
-  "CD": {"ef":"data"}  
+  "CD": {"ef":"data"},
+  "testKey": {"testvalue"} // `TESTVALUE` -> `testvalue` is lowercased 
 }
 ----
 
 [start=2]
-2. When `full_path` is false
+2. When `alter_full_field` is false (applicable only for fields)
 
 [source,yaml]
 ----
@@ -57,14 +61,14 @@ processors:
 // Input
 {
   "AB": {"CD":"data"},
-  "CD": {"ef":"data"} 
+  "CD": {"ef":"data"}, 
 }
 
 
 // output
 {
   "AB": {"cd":"data"},  // `AB.CD` -> `AB.cd` (only `cd` is lowercased)
-  "CD": {"ef":"data"}  
+  "CD": {"ef":"data"}, 
 }
 ----
 
@@ -103,6 +107,7 @@ processors:
 The `lowercase` processor has the following configuration settings:
 
 `fields`:: The field names to lowercase. The match is case-insensitive, e.g. `a.b.c.d` would match `A.b.C.d` or `A.B.C.D`.
+`values`:: (Optional) Specifies the exact values to be converted to lowercase. Each entry should include the full path to the value. Key matching is case-sensitive. If the target value is not a string, an error is triggered (`fail_on_error: true`) or the value is skipped (`fail_on_error: false`).
 `ignore_missing`:: (Optional) Indicates whether to ignore events that lack the source field.
                     The default is `false`, which will fail processing of an event if a field is missing.
 `fail_on_error`:: (Optional) If set to `true` and an error occurs, the changes are reverted and the original event is returned.

libbeat/processors/actions/lowercase.go

@@ -32,7 +32,7 @@ func init() {
 		checks.ConfigChecked(
 			NewLowerCaseProcessor,
 			checks.RequireFields("fields"),
-			checks.AllowedFields("fields", "when", "ignore_missing", "fail_on_error", "alter_full_field"),
+			checks.AllowedFields("fields", "ignore_missing", "fail_on_error", "alter_full_field", "values"),
 		),
 	)
 }

libbeat/processors/actions/lowercase_test.go

@@ -248,6 +248,115 @@ func TestLowerCaseProcessorRun(t *testing.T) {
 	})
 }
 
+func TestLowerCaseProcessorValues(t *testing.T) {
+	tests := []struct {
+		Name          string
+		Values        []string
+		IgnoreMissing bool
+		FailOnError   bool
+		FullPath      bool
+		Input         mapstr.M
+		Output        mapstr.M
+		Error         bool
+	}{
+		{
+			Name:          "Lowercase Values",
+			Values:        []string{"a.b.c"},
+			IgnoreMissing: false,
+			FailOnError:   true,
+			FullPath:      true,
+			Input: mapstr.M{
+				"a": mapstr.M{
+					"b": mapstr.M{
+						"c": "D",
+					},
+				},
+			},
+			Output: mapstr.M{
+				"a": mapstr.M{
+					"b": mapstr.M{
+						"c": "d", // d is lowercased
+					},
+				},
+			},
+			Error: false,
+		},
+		{
+			Name:          "Fail if given path to value is not a string",
+			Values:        []string{"a.B"},
+			IgnoreMissing: false,
+			FailOnError:   true,
+			FullPath:      true,
+			Input: mapstr.M{
+				"Field3": "Value",
+				"a": mapstr.M{
+					"B": mapstr.M{
+						"C": "D",
+					},
+				},
+			},
+			Output: mapstr.M{
+				"Field3": "Value",
+				"a": mapstr.M{
+					"B": mapstr.M{
+						"C": "D",
+					},
+				},
+				"error": mapstr.M{"message": "value of key \"a.B\" is not a string"},
+			},
+
+			Error: true,
+		},
+		{
+			Name:          "Fail On Missing Key Error",
+			Values:        []string{"a.B.c"},
+			IgnoreMissing: false,
+			FailOnError:   true,
+			FullPath:      true,
+			Input: mapstr.M{
+				"Field3": "Value",
+				"a": mapstr.M{
+					"B": mapstr.M{
+						"C": "D",
+					},
+				},
+			},
+			Output: mapstr.M{
+				"Field3": "Value",
+				"a": mapstr.M{
+					"B": mapstr.M{
+						"C": "D",
+					},
+				},
+				"error": mapstr.M{"message": "could not fetch value for key: a.B.c, Error: key not found"},
+			},
+
+			Error: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.Name, func(t *testing.T) {
+			p := &alterFieldProcessor{
+				Values:         test.Values,
+				IgnoreMissing:  test.IgnoreMissing,
+				FailOnError:    test.FailOnError,
+				AlterFullField: test.FullPath,
+				alterFunc:      lowerCase,
+			}
+
+			event, err := p.Run(&beat.Event{Fields: test.Input})
+
+			if !test.Error {
+				require.NoError(t, err)
+			} else {
+				require.Error(t, err)
+			}
+
+			assert.Equal(t, test.Output, event.Fields)
+		})
+	}
+}
 func BenchmarkLowerCaseProcessorRun(b *testing.B) {
 	tests := []struct {
 		Name   string