diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 30d6d0faa17..18076ecb54f 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -61,6 +61,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 
 *Metricbeat*
 
+- Add tags to events based on parsed identifier. {pull}33472[33472]
 
 *Packetbeat*
 
diff --git a/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch.go b/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch.go
index 7f7ac5a2d1d..77ba174ef3c 100644
--- a/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch.go
+++ b/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch.go
@@ -585,7 +585,23 @@ func (m *MetricSet) createEvents(svcCloudwatch cloudwatch.GetMetricDataAPIClient
 				events[identifierValue] = insertRootFields(events[identifierValue], output.Values[timestampIdx], labels)
 
 				// add tags to event based on identifierValue
-				insertTags(events, identifierValue, resourceTagMap)
+				// 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[identifierValue]; !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[identifierValue] = aws.InitEvent(regionName, m.AccountName, m.AccountID, timestamp)
+						insertTags(events, identifierValue, subIdentifier, resourceTagMap)
+					}
+				}
 			}
 		}
 	}
@@ -625,28 +641,20 @@ func compareAWSDimensions(dim1 []types.Dimension, dim2 []types.Dimension) bool {
 	return reflect.DeepEqual(dim1NameToValue, dim2NameToValue)
 }
 
-func insertTags(events map[string]mb.Event, identifier string, resourceTagMap map[string][]resourcegroupstaggingapitypes.Tag) {
-	// 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(identifier, dimensionSeparator)
-	for _, v := range subIdentifiers {
-		tags := resourceTagMap[v]
-		// some metric dimension values are arn format, eg: AWS/DDOS namespace metric
-		if len(tags) == 0 && strings.HasPrefix(v, "arn:") {
-			resourceID, err := aws.FindShortIdentifierFromARN(v)
-			if err == nil {
-				tags = resourceTagMap[resourceID]
-			}
+func insertTags(events map[string]mb.Event, identifier 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[identifier].RootFields.Put("aws.tags."+common.DeDot(*tag.Key), *tag.Value)
-			}
-			continue
+	}
+	if len(tags) != 0 {
+		// By default, replace dot "." using underscore "_" for tag keys.
+		// Note: tag values are not dedotted.
+		for _, tag := range tags {
+			_, _ = events[identifier].RootFields.Put("aws.tags."+common.DeDot(*tag.Key), *tag.Value)
 		}
 	}
 }
diff --git a/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch_test.go b/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch_test.go
index 5417b8323cc..c43a38960a2 100644
--- a/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch_test.go
+++ b/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch_test.go
@@ -10,6 +10,7 @@ package cloudwatch
 import (
 	"context"
 	"errors"
+	"strings"
 	"testing"
 	"time"
 
@@ -1560,7 +1561,10 @@ func TestInsertTags(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.title, func(t *testing.T) {
-			insertTags(events, c.identifier, resourceTagMap)
+			subIdentifiers := strings.Split(c.identifier, dimensionSeparator)
+			for _, subIdentifier := range subIdentifiers {
+				insertTags(events, c.identifier, subIdentifier, resourceTagMap)
+			}
 			value, err := events[c.identifier].RootFields.GetValue(c.expectedTagKey)
 			assert.NoError(t, err)
 			assert.Equal(t, c.expectedTagValue, value)