[AWS] Add tags to events based on parsed identifier (#33472)

(cherry picked from commit 50c6ae5)

# Conflicts:
#	x-pack/metricbeat/module/aws/cloudwatch/cloudwatch.go
#	x-pack/metricbeat/module/aws/cloudwatch/cloudwatch_test.go

CHANGELOG.next.asciidoc

@@ -41,6 +41,15 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 *Metricbeat*
 
+- Fix GCP storage field naming {pull}32806[32806]
+- in module/windows/perfmon, changed collection method of the second counter value required to create a displayable value {pull}32305[32305]
+- Fix and improve AWS metric period calculation to avoid zero-length intervals {pull}32724[32724]
+- Add missing cluster metadata to k8s module metricsets {pull}32979[32979] {pull}33032[33032]
+- Change max query size for GetMetricData API to 500 and add RecentlyActive for ListMetrics API call {pull}33105[33105]
+- Add GCP CloudSQL region filter {pull}32943[32943]
+- Fix logstash cgroup mappings {pull}33131[33131]
+- Remove unused `elasticsearch.node_stats.indices.bulk.avg_time.bytes` mapping {pull}33263[33263]
+- Add tags to events based on parsed identifier. {pull}33472[33472]
 
 *Packetbeat*
 

x-pack/metricbeat/module/aws/cloudwatch/cloudwatch.go

@@ -569,6 +569,7 @@ func (m *MetricSet) createEvents(svcCloudwatch cloudwatchiface.ClientAPI, svcRes
 				}
 
 				identifierValue := labels[identifierValueIdx]
+<<<<<<< HEAD
 				if _, ok := events[identifierValue]; !ok {
 					// when tagsFilter is not empty but no entry in
 					// resourceTagMap for this identifier, do not initialize
@@ -582,6 +583,29 @@ func (m *MetricSet) createEvents(svcCloudwatch cloudwatchiface.ClientAPI, svcRes
 
 				// add tags to event based on identifierValue
 				insertTags(events, identifierValue, resourceTagMap)
+=======
+				uniqueIdentifierValue := *output.Label + fmt.Sprint("-", valI)
+
+				// add tags to event based on identifierValue
+				// Check if identifier includes dimensionSeparator (comma in this case),
+				// split the identifier and check for each sub-identifier.
+				// For example, identifier might be [storageType, s3BucketName].
+				// And tags are only store under s3BucketName in resourceTagMap.
+				subIdentifiers := strings.Split(identifierValue, dimensionSeparator)
+				for _, subIdentifier := range subIdentifiers {
+					if _, ok := events[uniqueIdentifierValue]; !ok {
+						// when tagsFilter is not empty but no entry in
+						// resourceTagMap for this identifier, do not initialize
+						// an event for this identifier.
+						if len(tagsFilter) != 0 && resourceTagMap[subIdentifier] == nil {
+							continue
+						}
+						events[uniqueIdentifierValue] = aws.InitEvent(regionName, m.AccountName, m.AccountID, output.Timestamps[valI])
+					}
+					events[uniqueIdentifierValue] = insertRootFields(events[uniqueIdentifierValue], metricDataResultValue, labels)
+					insertTags(events, uniqueIdentifierValue, subIdentifier, resourceTagMap)
+				}
+>>>>>>> 50c6ae5f3a ([AWS] Add tags to events based on parsed identifier (#33472))
 			}
 		}
 	}
@@ -630,6 +654,7 @@ func compareAWSDimensions(dim1 []cloudwatch.Dimension, dim2 []cloudwatch.Dimensi
 	return reflect.DeepEqual(dim1NameToValue, dim2NameToValue)
 }
 
+<<<<<<< HEAD
 func insertTags(events map[string]mb.Event, identifier string, resourceTagMap map[string][]resourcegroupstaggingapi.Tag) {
 	// Check if identifier includes dimensionSeparator (comma in this case),
 	// split the identifier and check for each sub-identifier.
@@ -652,6 +677,22 @@ func insertTags(events map[string]mb.Event, identifier string, resourceTagMap ma
 				_, _ = events[identifier].RootFields.Put("aws.tags."+common.DeDot(*tag.Key), *tag.Value)
 			}
 			continue
+=======
+func insertTags(events map[string]mb.Event, uniqueIdentifierValue string, subIdentifier string, resourceTagMap map[string][]resourcegroupstaggingapitypes.Tag) {
+	tags := resourceTagMap[subIdentifier]
+	// some metric dimension values are arn format, eg: AWS/DDOS namespace metric
+	if len(tags) == 0 && strings.HasPrefix(subIdentifier, "arn:") {
+		resourceID, err := aws.FindShortIdentifierFromARN(subIdentifier)
+		if err == nil {
+			tags = resourceTagMap[resourceID]
+		}
+	}
+	if len(tags) != 0 {
+		// By default, replace dot "." using underscore "_" for tag keys.
+		// Note: tag values are not dedotted.
+		for _, tag := range tags {
+			_, _ = events[uniqueIdentifierValue].RootFields.Put("aws.tags."+common.DeDot(*tag.Key), *tag.Value)
+>>>>>>> 50c6ae5f3a ([AWS] Add tags to events based on parsed identifier (#33472))
 		}
 	}
 }

x-pack/metricbeat/module/aws/cloudwatch/cloudwatch_test.go

@@ -9,7 +9,11 @@ package cloudwatch
 
 import (
 	"errors"
+<<<<<<< HEAD
 	"net/http"
+=======
+	"strings"
+>>>>>>> 50c6ae5f3a ([AWS] Add tags to events based on parsed identifier (#33472))
 	"testing"
 	"time"
 
@@ -1636,7 +1640,14 @@ func TestInsertTags(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.title, func(t *testing.T) {
+<<<<<<< HEAD
 			insertTags(events, c.identifier, resourceTagMap)
+=======
+			subIdentifiers := strings.Split(c.identifier, dimensionSeparator)
+			for _, subIdentifier := range subIdentifiers {
+				insertTags(events, c.identifier, subIdentifier, resourceTagMap)
+			}
+>>>>>>> 50c6ae5f3a ([AWS] Add tags to events based on parsed identifier (#33472))
 			value, err := events[c.identifier].RootFields.GetValue(c.expectedTagKey)
 			assert.NoError(t, err)
 			assert.Equal(t, c.expectedTagValue, value)