From 4a62f823474edc69398bf7525fadd48e7acae28c Mon Sep 17 00:00:00 2001
From: Andrew Kroh <andrew.kroh@elastic.co>
Date: Tue, 30 Nov 2021 19:06:52 -0500
Subject: [PATCH] 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"
  }
}
```
---
 CHANGELOG.next.asciidoc                       |  3 +-
 libbeat/processors/actions/add_labels.go      | 42 ++++++++++++++++---
 libbeat/processors/actions/add_labels_test.go | 11 +++++
 3 files changed, 50 insertions(+), 6 deletions(-)

diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index b8bb5c5d542..8b41d1aa75d 100644
--- a/CHANGELOG.next.asciidoc
+++ b/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*
 
diff --git a/libbeat/processors/actions/add_labels.go b/libbeat/processors/actions/add_labels.go
index 76696659265..a57eb1b51d5 100644
--- a/libbeat/processors/actions/add_labels.go
+++ b/libbeat/processors/actions/add_labels.go
@@ -20,6 +20,8 @@ package actions
 import (
 	"fmt"
 
+	"github.com/elastic/go-ucfg"
+
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/processors"
 	"github.com/elastic/beats/v7/libbeat/processors/checks"
@@ -41,10 +43,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)
+	}
+
+	flatLabels, err := flattenLabels(config.Labels)
+	if err != nil {
+		return nil, fmt.Errorf("failed to flatten labels: %w", err)
 	}
 
-	return makeFieldsProcessor(LabelsKey, config.Labels.Flatten(), true), nil
+	return makeFieldsProcessor(LabelsKey, flatLabels, true), nil
 }
 
 // NewAddLabels creates a new processor adding the given object to events. Set
@@ -53,8 +60,33 @@ 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 := (*ucfg.Config)(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
 }
diff --git a/libbeat/processors/actions/add_labels_test.go b/libbeat/processors/actions/add_labels_test.go
index 24ddcbab58f..ca9dd22a528 100644
--- a/libbeat/processors/actions/add_labels_test.go
+++ b/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"}]}}}`),
+		},
 	})
 }