feat: update api compatibility to make it compatible with GMD based p…

plugins/inputs/cloudwatch_metric_streams/README.md

@@ -52,7 +52,8 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   # access_key = "test-key"
 
   ## An optional flag to keep Metric Streams metrics compatible with
-  ## CloudWatch's API naming
+  ## Cloudwatch Input Plugin. When enabled, it'll store metrics in same 
+  ## format as with Cloudwatch API-based Input Plugin.
   # api_compatability = false
 
   ## Set one or more allowed client CA certificate file names to

plugins/inputs/cloudwatch_metric_streams/cloudwatch_metric_streams.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/config"
+	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/internal/choice"
 	common_tls "github.com/influxdata/telegraf/plugins/common/tls"
 	"github.com/influxdata/telegraf/plugins/inputs"
@@ -342,26 +343,56 @@ func (cms *CloudWatchMetricStreams) composeMetrics(data data) {
 
 	// Rename Statistics to match the CloudWatch API if in API Compatability mode
 	if cms.APICompatability {
-		if v, ok := fields["max"]; ok {
-			fields["maximum"] = v
+		// Adding average stat, which is present in the API plugin
+		measurement = sanitizeMeasurement(data.Namespace)
+		metricName := snakeCase(data.MetricName)
+
+		_sum := data.Value["sum"]
+		_count := data.Value["count"]
+		average := float64(0)
+		if _count > 0 {
+			average = _sum / _count
+		}
+
+		fields[metricName+"_average"] = average
+
+		max, ok := fields["max"]
+		if ok {
+			fields[metricName+"_maximum"] = max
 			delete(fields, "max")
 		}
 
-		if v, ok := fields["min"]; ok {
-			fields["minimum"] = v
+		min, ok := fields["min"]
+		if ok {
+			fields[metricName+"_minimum"] = min
 			delete(fields, "min")
 		}
 
-		if v, ok := fields["count"]; ok {
-			fields["samplecount"] = v
+		count, ok := fields["count"]
+		if ok {
+			fields[metricName+"_sample_count"] = count
 			delete(fields, "count")
 		}
+
+		sum, ok := fields["sum"]
+		if ok {
+			fields[metricName+"_sum"] = sum
+			delete(fields, "sum")
+		}
+	}
+
+	if cms.APICompatability {
+		tags["account"] = data.AccountID
+	} else {
+		tags["accountId"] = data.AccountID
 	}
 
-	tags["accountId"] = data.AccountID
 	tags["region"] = data.Region
 
 	for dimension, value := range data.Dimensions {
+		if cms.APICompatability {
+			dimension = snakeCase(dimension)
+		}
 		tags[dimension] = value
 	}
 
@@ -422,3 +453,16 @@ func init() {
 		}
 	})
 }
+
+func sanitizeMeasurement(namespace string) string {
+	namespace = strings.ReplaceAll(namespace, "/", "_")
+	namespace = snakeCase(namespace)
+	return "cloudwatch_" + namespace
+}
+
+func snakeCase(s string) string {
+	s = internal.SnakeCase(s)
+	s = strings.ReplaceAll(s, " ", "_")
+	s = strings.ReplaceAll(s, "__", "_")
+	return s
+}

plugins/inputs/cloudwatch_metric_streams/cloudwatch_metric_streams_test.go

@@ -352,9 +352,10 @@ func TestComposeAPICompatibleMetrics(t *testing.T) {
 	metricStream.composeMetrics(data)
 
 	acc.Wait(1)
-	acc.AssertContainsTaggedFields(t, "aws_ec2_cpuutilization",
-		map[string]interface{}{"maximum": 0.4366666666666666, "minimum": 0.3683333333333333, "sum": 1.9399999999999997, "samplecount": 5.0},
-		map[string]string{"AutoScalingGroupName": "test-autoscaling-group", "accountId": "546734499701", "region": "us-west-2"},
+	acc.AssertContainsTaggedFields(t, "cloudwatch_aws_ec2",
+		map[string]interface{}{"cpu_utilization_maximum": 0.4366666666666666, "cpu_utilization_minimum": 0.3683333333333333,
+			"cpu_utilization_sum": 1.9399999999999997, "cpu_utilization_sample_count": 5.0, "cpu_utilization_average": 0.38799999999999996},
+		map[string]string{"auto_scaling_group_name": "test-autoscaling-group", "account": "546734499701", "region": "us-west-2"},
 	)
 }