Update group by dimensions logic

When users define dimensions in the config, the current implementation
groups metrics by timestamp and single dimension.

Grouping by ts + single dimension can sometimes lead to multiple
documents with the same dimension values. This does not play well with
TSDB, because it expects all documents with the same timestamp to have
a unique combination of dimensions value.

I am updating the group by dimensions logic to use all dimensions for
grouping instead of just one.

It is working fine with the test cases I am using, but this needs more
testing and understanding.

Refs: elastic/integrations#7160

x-pack/metricbeat/module/azure/client.go

@@ -46,8 +46,10 @@ func NewClient(config Config) (*Client, error) {
 	return client, nil
 }
 
-// InitResources function will retrieve and validate the resources configured by the users and then map the information configured to client metrics.
-// the mapMetric function sent in this case will handle the mapping part as different metric and aggregation options work for different metricsets
+// InitResources function retrieves and validates the resources configured by the users and then map the information
+// configured to client metrics.
+// The mapResourceMetrics function sent in this case will handle the mapping part as different metric and aggregation options
+// work for different metricsets.
 func (client *Client) InitResources(fn mapResourceMetrics) error {
 	if len(client.Config.Resources) == 0 {
 		return fmt.Errorf("no resource options defined")
@@ -72,7 +74,7 @@ func (client *Client) InitResources(fn mapResourceMetrics) error {
 			client.Log.Error(err)
 			continue
 		}
-		//map resources to the client
+		// map resources to the client; copies the Azure resource information to the client.
 		for _, resource := range resourceList {
 			if !containsResource(*resource.ID, client.Resources) {
 				client.Resources = append(client.Resources, Resource{

x-pack/metricbeat/module/azure/data.go

@@ -55,41 +55,68 @@ func EventsMapping(metrics []Metric, client *Client, report mb.ReporterV2) error
 		}
 	}
 
-	// grouping metric values by timestamp and creating events (for each metric the REST api can retrieve multiple metric values for same aggregation  but different timeframes)
+	// grouping metric values by timestamp and creating events (for each metric the REST api can retrieve multiple
+	// metric values for same aggregation  but different timeframes)
 	for _, grouped := range groupByDimensions {
 		defaultMetric := grouped[0]
 		resource := client.GetResourceForMetaData(defaultMetric)
 		groupByTimeMetrics := make(map[time.Time][]MetricValue)
+
 		for _, metric := range grouped {
 			for _, m := range metric.Values {
 				groupByTimeMetrics[m.timestamp] = append(groupByTimeMetrics[m.timestamp], m)
 			}
 		}
+
+		exists, _ := getWildcardDimensions(defaultMetric.Dimensions)
+
 		for timestamp, groupTimeValues := range groupByTimeMetrics {
 			var event mb.Event
 			var metricList mapstr.M
 			var vm VmResource
 			// group events by dimension values
-			exists, validDimensions := returnAllDimensions(defaultMetric.Dimensions)
-			if exists {
-				for _, selectedDimension := range validDimensions {
-					groupByDimensions := make(map[string][]MetricValue)
-					for _, dimGroupValue := range groupTimeValues {
-						dimKey := fmt.Sprintf("%s,%s", selectedDimension.Name, getDimensionValue(selectedDimension.Name, dimGroupValue.dimensions))
-						groupByDimensions[dimKey] = append(groupByDimensions[dimKey], dimGroupValue)
-					}
-					for _, groupDimValues := range groupByDimensions {
-						manageAndReportEvent(client, report, event, metricList, vm, timestamp, defaultMetric, resource, groupDimValues)
-					}
-				}
-			} else {
+			//exists, validDimensions := getWildcardDimensions(defaultMetric.Dimensions)
+			//exists, _ := getWildcardDimensions(defaultMetric.Dimensions)
+
+			if !exists {
+				//
+				// There are no dimensions with wildcards, so we can group all the values in one event.
+				//
 				manageAndReportEvent(client, report, event, metricList, vm, timestamp, defaultMetric, resource, groupTimeValues)
+				continue
+			}
+
+			//
+			// There are dimensions with wildcards, so we need to group the values by the dimension values.
+			//
+			groupByDimensions := make(map[string][]MetricValue)
+			for _, dimGroupValue := range groupTimeValues {
+				//dimKey := fmt.Sprintf("%s,%s", selectedDimension.Name, getDimensionValue(selectedDimension.Name, dimGroupValue.dimensions))
+				//
+				// We need to group the values by the dimension values.
+				//
+				dimKey := buildDimensionKey(dimGroupValue.dimensions)
+				groupByDimensions[dimKey] = append(groupByDimensions[dimKey], dimGroupValue)
+			}
+
+			// Create an event for each group of dimension values.
+			for _, groupDimValues := range groupByDimensions {
+				manageAndReportEvent(client, report, event, metricList, vm, timestamp, defaultMetric, resource, groupDimValues)
 			}
 		}
 	}
+
 	return nil
 }
 
+func buildDimensionKey(dimensions []Dimension) string {
+	var dimKey string
+	for _, dim := range dimensions {
+		dimKey += fmt.Sprintf("%s,%s;", dim.Name, dim.Value)
+	}
+	return dimKey
+}
+
 // manageAndReportEvent function will handle event creation and report
 func manageAndReportEvent(client *Client, report mb.ReporterV2, event mb.Event, metricList mapstr.M, vm VmResource, timestamp time.Time, defaultMetric Metric, resource Resource, groupedValues []MetricValue) {
 	event, metricList = createEvent(timestamp, defaultMetric, resource, groupedValues)
@@ -229,8 +256,8 @@ func getDimensionValue(dimension string, dimensions []Dimension) string {
 	return ""
 }
 
-// returnAllDimensions will check if users has entered a filter for all dimension values (*)
-func returnAllDimensions(dimensions []Dimension) (bool, []Dimension) {
+// getWildcardDimensions returns all user-defined dimension names with values = * (wildcard)
+func getWildcardDimensions(dimensions []Dimension) (bool, []Dimension) {
 	if len(dimensions) == 0 {
 		return false, nil
 	}

x-pack/metricbeat/module/azure/data_test.go

@@ -24,7 +24,7 @@ func TestReturnAllDimensions(t *testing.T) {
 			Name:  "SlotID",
 		},
 	}
-	result, dims := returnAllDimensions(dimensionList)
+	result, dims := getWildcardDimensions(dimensionList)
 	assert.True(t, result)
 	assert.Equal(t, len(dims), 1)
 	assert.Equal(t, dims[0].Name, "SlotID")