[processor/metricstransformprocessor] Support count aggregation type (

open-telemetry#32935)

**Description:** Implements
open-telemetry#24978.

.chloggen/count-aggregation-type.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: metricstransformprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Adds the 'count' aggregation type to the Metrics Transform Processor."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [24978]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

processor/metricstransformprocessor/README.md

@@ -86,7 +86,7 @@ processors:
         # new_name specifies the updated name of the metric; if action is insert or combine, new_name is required
         new_name: <new_metric_name_inserted>
         # aggregation_type defines how combined data points will be aggregated; if action is combine, aggregation_type is required
-        aggregation_type: {sum, mean, min, max}
+        aggregation_type: {sum, mean, min, max, count}
         # submatch_case specifies the case that should be used when adding label values based on regexp submatches when performing a combine action; leave blank to use the submatch value as is
         submatch_case: {lower, upper}
         # operations contain a list of operations that will be performed on the resulting metric(s)
@@ -106,7 +106,7 @@ processors:
             # label_set contains a list of labels that will remain after aggregation; if action is aggregate_labels, label_set is required
             label_set: [labels...]
             # aggregation_type defines how data points will be aggregated; if action is aggregate_labels or aggregate_label_values, aggregation_type is required
-            aggregation_type: {sum, mean, min, max}
+            aggregation_type: {sum, mean, min, max, count}
             # experimental_scale specifies the scalar to apply to values
             experimental_scale: <scalar>
             # value_actions contain a list of operations that will be performed on the selected label

processor/metricstransformprocessor/config.go

@@ -231,9 +231,12 @@ const (
 
 	// max indicates taking the max of the aggregated data.
 	max aggregationType = "max"
+
+	// count indicates taking the count of the aggregated data.
+	count aggregationType = "count"
 )
 
-var aggregationTypes = []aggregationType{sum, mean, min, max}
+var aggregationTypes = []aggregationType{sum, mean, min, max, count}
 
 func (at aggregationType) isValid() bool {
 	for _, aggregationType := range aggregationTypes {

processor/metricstransformprocessor/metrics_transform_processor_testcases_test.go

@@ -298,6 +298,35 @@ var (
 					addIntDatapoint(1, 2, 3, "label1-value1").build(),
 			},
 		},
+		{
+			name: "metric_label_aggregation_count_int_update",
+			transforms: []internalTransform{
+				{
+					MetricIncludeFilter: internalFilterStrict{include: "metric1"},
+					Action:              Update,
+					Operations: []internalOperation{
+						{
+							configOperation: Operation{
+								Action:          aggregateLabels,
+								AggregationType: count,
+								LabelSet:        []string{"label1"},
+							},
+							labelSetMap: map[string]bool{"label1": true},
+						},
+					},
+				},
+			},
+			in: []pmetric.Metric{
+				metricBuilder(pmetric.MetricTypeGauge, "metric1", "label1", "label2").
+					addIntDatapoint(1, 2, 1, "label1-value1", "label2-value1").
+					addIntDatapoint(1, 2, 4, "label1-value1", "label2-value2").
+					addIntDatapoint(1, 2, 2, "label1-value1", "label2-value2").build(),
+			},
+			out: []pmetric.Metric{
+				metricBuilder(pmetric.MetricTypeGauge, "metric1", "label1").
+					addIntDatapoint(1, 2, 3, "label1-value1").build(),
+			},
+		},
 		{
 			name: "metric_label_aggregation_min_int_update",
 			transforms: []internalTransform{
@@ -411,6 +440,34 @@ var (
 					addDoubleDatapoint(1, 2, 3, "label1-value1").build(),
 			},
 		},
+		{
+			name: "metric_label_aggregation_count_double_update",
+			transforms: []internalTransform{
+				{
+					MetricIncludeFilter: internalFilterStrict{include: "metric1"},
+					Action:              Update,
+					Operations: []internalOperation{
+						{
+							configOperation: Operation{
+								Action:          aggregateLabels,
+								AggregationType: count,
+								LabelSet:        []string{"label1"},
+							},
+							labelSetMap: map[string]bool{"label1": true},
+						},
+					},
+				},
+			},
+			in: []pmetric.Metric{
+				metricBuilder(pmetric.MetricTypeGauge, "metric1", "label1", "label2").
+					addDoubleDatapoint(1, 2, 3, "label1-value1", "label2-value1").
+					addDoubleDatapoint(1, 2, 1, "label1-value1", "label2-value2").build(),
+			},
+			out: []pmetric.Metric{
+				metricBuilder(pmetric.MetricTypeGauge, "metric1", "label1").
+					addDoubleDatapoint(1, 2, 2, "label1-value1").build(),
+			},
+		},
 		{
 			name: "metric_label_aggregation_min_double_update",
 			transforms: []internalTransform{

processor/metricstransformprocessor/operation_aggregate_labels.go

@@ -167,6 +167,8 @@ func mergeNumberDataPoints(dpsMap map[string]pmetric.NumberDataPointSlice, agg a
 					dp.SetDoubleValue(math.Max(dp.DoubleValue(), doubleVal(dps.At(i))))
 				case min:
 					dp.SetDoubleValue(math.Min(dp.DoubleValue(), doubleVal(dps.At(i))))
+				case count:
+					dp.SetDoubleValue(float64(dps.Len()))
 				}
 				if dps.At(i).StartTimestamp() < dp.StartTimestamp() {
 					dp.SetStartTimestamp(dps.At(i).StartTimestamp())
@@ -188,6 +190,8 @@ func mergeNumberDataPoints(dpsMap map[string]pmetric.NumberDataPointSlice, agg a
 					if dp.IntValue() > intVal(dps.At(i)) {
 						dp.SetIntValue(intVal(dps.At(i)))
 					}
+				case count:
+					dp.SetIntValue(int64(dps.Len()))
 				}
 				if dps.At(i).StartTimestamp() < dp.StartTimestamp() {
 					dp.SetStartTimestamp(dps.At(i).StartTimestamp())