[libbeat] Fix add_labels flattening of arrays values (elastic#29211)

* Fix add_labels flattening of arrays values

The processor was not working as described in the docs. For example:

```yaml
processors:
  - add_labels:
      labels:
        number: 1
        with.dots: test
        nested:
          with.dots: nested
        array:
          - do
          - re
          - with.field: mi
```

```json
{
  "labels": {
    "array": [
      "do",
      "re",
      {
        "with": {
          "field": "mi"
        }
      }
    ],
    "nested.with.dots": "nested",
    "number": 1,
    "with.dots": "test"
  }
```

```json
{
  "labels": {
    "number": 1,
    "with.dots": "test",
    "nested.with.dots": "nested",
    "array.0": "do",
    "array.1": "re",
    "array.2.with.field": "mi"
  }
}
```

* Expose ucfg FlattenedKeys() in common.Config

CHANGELOG.next.asciidoc

@@ -142,6 +142,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix the wrong beat name on monitoring and state endpoint {issue}27755[27755]
 - Skip configuration checks in autodiscover for configurations that are already running {pull}29048[29048]
 - Fix `decode_json_processor` to always respect `add_error_key` {pull}29107[29107]
+- Fix `add_labels` flattening of array values. {pull}29211[29211]
 
 *Auditbeat*
 
@@ -188,7 +189,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Revert usageDetails api version to 2019-01-01. {pull}28995[28995]
 - Fix in `aws-s3` input regarding provider discovery through endpoint {pull}28963[28963]
 - Fix `threatintel.misp` filters configuration. {issue}27970[27970]
-- Fix opening files on Windows in filestream so open files can be deleted. {issue}29113[29113] {pull}29180[29180] 
+- Fix opening files on Windows in filestream so open files can be deleted. {issue}29113[29113] {pull}29180[29180]
 
 *Heartbeat*
 

libbeat/common/config.go

@@ -259,6 +259,11 @@ func (c *Config) IsArray() bool {
 	return c.access().IsArray()
 }
 
+// FlattenedKeys return a sorted flattened views of the set keys in the configuration.
+func (c *Config) FlattenedKeys() []string {
+	return c.access().FlattenedKeys(configOpts...)
+}
+
 func (c *Config) PrintDebugf(msg string, params ...interface{}) {
 	selector := selectorConfigWithPassword
 	filtered := false

libbeat/processors/actions/add_labels.go

@@ -41,10 +41,15 @@ func createAddLabels(c *common.Config) (processors.Processor, error) {
 	}{}
 	err := c.Unpack(&config)
 	if err != nil {
-		return nil, fmt.Errorf("fail to unpack the add_fields configuration: %s", err)
+		return nil, fmt.Errorf("fail to unpack the add_fields configuration: %w", err)
 	}
 
-	return makeFieldsProcessor(LabelsKey, config.Labels.Flatten(), true), nil
+	flatLabels, err := flattenLabels(config.Labels)
+	if err != nil {
+		return nil, fmt.Errorf("failed to flatten labels: %w", err)
+	}
+
+	return makeFieldsProcessor(LabelsKey, flatLabels, true), nil
 }
 
 // NewAddLabels creates a new processor adding the given object to events. Set
@@ -53,8 +58,32 @@ func createAddLabels(c *common.Config) (processors.Processor, error) {
 // If labels contains nested objects, NewAddLabels will flatten keys into labels by
 // by joining names with a dot ('.') .
 // The labels will be inserted into the 'labels' field.
-func NewAddLabels(labels common.MapStr, shared bool) processors.Processor {
+func NewAddLabels(labels common.MapStr, shared bool) (processors.Processor, error) {
+	flatLabels, err := flattenLabels(labels)
+	if err != nil {
+		return nil, fmt.Errorf("failed to flatten labels: %w", err)
+	}
+
 	return NewAddFields(common.MapStr{
-		LabelsKey: labels.Flatten(),
-	}, shared, true)
+		LabelsKey: flatLabels,
+	}, shared, true), nil
+}
+
+func flattenLabels(labels common.MapStr) (common.MapStr, error) {
+	labelConfig, err := common.NewConfigFrom(labels)
+	if err != nil {
+		return nil, err
+	}
+
+	flatKeys := labelConfig.FlattenedKeys()
+	flatMap := make(common.MapStr, len(flatKeys))
+	for _, k := range flatKeys {
+		v, err := labelConfig.String(k, -1)
+		if err != nil {
+			return nil, err
+		}
+		flatMap[k] = v
+	}
+
+	return flatMap, nil
 }

libbeat/processors/actions/add_labels_test.go

@@ -59,5 +59,16 @@ func TestAddLabels(t *testing.T) {
 				`{add_labels.labels: {l2: b, lc: b}}`,
 			),
 		},
+		"add array": {
+			event: common.MapStr{},
+			want: common.MapStr{
+				"labels": common.MapStr{
+					"array.0":       "foo",
+					"array.1":       "bar",
+					"array.2.hello": "world",
+				},
+			},
+			cfg: single(`{add_labels: {labels: {array: ["foo", "bar", {"hello": "world"}]}}}`),
+		},
 	})
 }