[Cherry-pick elastic#13389 to 7.4]Aws cloudwatch msk (elastic#13452)

leweafan · Sep 2, 2019 · e16d993 · e16d993
1 parent 0b46303
commit e16d993
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 9 deletions.
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -184,6 +184,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix redis key metricset dashboard references to index pattern. {pull}13303[13303]
 - Check if fields in DBInstance is nil in rds metricset. {pull}13294[13294] {issue}13037[13037]
 - Fix silent failures in kafka and prometheus module. {pull}13353[13353] {issue}13252[13252]
+- Fix issue with aws cloudwatch module where dimensions and/or namespaces that contain space are not being parsed correctly {pull}13389[13389]
 - Fix module-level fields in Kubernetes metricsets. {pull}13433[13433]
 - Fix panic in Redis Key metricset when collecting information from a removed key. {pull}13426[13426]
 

diff --git a/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch.go b/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch.go
@@ -30,6 +30,7 @@ var (
 	identifierNameIdx  = 3
 	identifierValueIdx = 4
 	defaultStatistics  = []string{"Average", "Maximum", "Minimum", "Sum", "SampleCount"}
+	labelSeperator     = "|"
 )
 
 // init registers the MetricSet with the central registry as soon as the program
@@ -303,7 +304,7 @@ func createMetricDataQueries(listMetricsTotal []metricsWithStatistics, period ti
 
 func constructLabel(metric cloudwatch.Metric, statistic string) string {
 	// label = metricName + namespace + statistic + dimKeys + dimValues
-	label := *metric.MetricName + " " + *metric.Namespace + " " + statistic
+	label := *metric.MetricName + labelSeperator + *metric.Namespace + labelSeperator + statistic
 	dimNames := ""
 	dimValues := ""
 	for i, dim := range metric.Dimensions {
@@ -316,8 +317,8 @@ func constructLabel(metric cloudwatch.Metric, statistic string) string {
 	}
 
 	if dimNames != "" && dimValues != "" {
-		label += " " + dimNames
-		label += " " + dimValues
+		label += labelSeperator + dimNames
+		label += labelSeperator + dimValues
 	}
 	return label
 }
@@ -437,7 +438,7 @@ func (m *MetricSet) createEvents(svcCloudwatch cloudwatchiface.ClientAPI, svcRes
 
 			exists, timestampIdx := aws.CheckTimestampInArray(timestamp, output.Timestamps)
 			if exists {
-				labels := strings.Split(*output.Label, " ")
+				labels := strings.Split(*output.Label, labelSeperator)
 				if len(labels) == 5 {
 					identifierValue := labels[identifierValueIdx]
 					events[identifierValue] = insertRootFields(events[identifierValue], output.Values[timestampIdx], labels)

diff --git a/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch_test.go b/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch_test.go
@@ -158,6 +158,13 @@ var (
 		namespace: "AWS/Lambda",
 		statistic: defaultStatistics,
 	}
+
+	namespaceMSK = "AWS/Kafka"
+	metricName6  = "MemoryUsed"
+	listMetric8  = cloudwatch.Metric{
+		MetricName: &metricName6,
+		Namespace:  &namespaceMSK,
+	}
 )
 
 func TestGetIdentifiers(t *testing.T) {
@@ -175,27 +182,32 @@ func TestConstructLabel(t *testing.T) {
 		{
 			listMetric1,
 			"Average",
-			"CPUUtilization AWS/EC2 Average InstanceId i-1",
+			"CPUUtilization|AWS/EC2|Average|InstanceId|i-1",
 		},
 		{
 			listMetric2,
 			"Maximum",
-			"StatusCheckFailed AWS/EC2 Maximum InstanceId i-1",
+			"StatusCheckFailed|AWS/EC2|Maximum|InstanceId|i-1",
 		},
 		{
 			listMetric3,
 			"Minimum",
-			"StatusCheckFailed_System AWS/EC2 Minimum InstanceId i-2",
+			"StatusCheckFailed_System|AWS/EC2|Minimum|InstanceId|i-2",
 		},
 		{
 			listMetric4,
 			"Sum",
-			"StatusCheckFailed_Instance AWS/EC2 Sum InstanceId i-2",
+			"StatusCheckFailed_Instance|AWS/EC2|Sum|InstanceId|i-2",
 		},
 		{
 			listMetric5,
 			"SampleCount",
-			"CPUUtilization AWS/EC2 SampleCount",
+			"CPUUtilization|AWS/EC2|SampleCount",
+		},
+		{
+			listMetric8,
+			"SampleCount",
+			"MemoryUsed|AWS/Kafka|SampleCount",
 		},
 	}
 
@@ -358,6 +370,24 @@ func TestReadCloudwatchConfig(t *testing.T) {
 				},
 			},
 		},
+		{
+			"test AWS/Kafka MemoryUsed",
+			[]Config{
+				{
+					Namespace:  "AWS/Kafka",
+					MetricName: []string{"MemoryUsed"},
+				},
+			},
+			listMetricWithDetail{},
+			[]namespaceWithDetail{
+				{
+					namespace:          "AWS/Kafka",
+					resourceTypeFilter: "",
+					metricNames:        []string{"MemoryUsed"},
+					statistic:          defaultStatistics,
+				},
+			},
+		},
 	}
 	for _, c := range cases {
 		t.Run(c.title, func(t *testing.T) {