[Azure] [Metrics] Update group by dimensions logic

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{
@@ -170,21 +172,21 @@ func (client *Client) CreateMetric(resourceId string, subResourceId string, name
 
 // MapMetricByPrimaryAggregation will map the primary aggregation of the metric definition to the client metric
 func (client *Client) MapMetricByPrimaryAggregation(metrics []armmonitor.MetricDefinition, resourceId string, subResourceId string, namespace string, dim []Dimension, timegrain string) []Metric {
-	var clientMetrics []Metric
-
 	metricGroups := make(map[string][]armmonitor.MetricDefinition)
-
 	for _, met := range metrics {
 		metricGroups[string(*met.PrimaryAggregationType)] = append(metricGroups[string(*met.PrimaryAggregationType)], met)
 	}
 
+	clientMetrics := make([]Metric, 0, len(metricGroups))
 	for key, metricGroup := range metricGroups {
 		var metricNames []string
 		for _, metricName := range metricGroup {
 			metricNames = append(metricNames, *metricName.Name.Value)
 		}
+
 		clientMetrics = append(clientMetrics, client.CreateMetric(resourceId, subResourceId, namespace, metricNames, key, dim, timegrain))
 	}
+
 	return clientMetrics
 }
 

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

@@ -55,46 +55,72 @@ 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
+			//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 {
-				manageAndReportEvent(client, report, event, metricList, vm, timestamp, defaultMetric, resource, groupTimeValues)
+			//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, 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, 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)
+func manageAndReportEvent(client *Client, report mb.ReporterV2, timestamp time.Time, defaultMetric Metric, resource Resource, groupedValues []MetricValue) {
+	event, metricList := createEvent(timestamp, defaultMetric, resource, groupedValues)
 	if client.Config.AddCloudMetadata {
-		vm = client.GetVMForMetaData(&resource, groupedValues)
+		vm := client.GetVMForMetaData(&resource, groupedValues)
 		addCloudVMMetadata(&event, vm, resource.Subscription)
 	}
 	if client.Config.DefaultResourceType == "" {
@@ -197,7 +223,7 @@ func createEvent(timestamp time.Time, metric Metric, resource Resource, metricVa
 
 	metricList := mapstr.M{}
 	for _, value := range metricValues {
-		metricNameString := fmt.Sprintf("%s", managePropertyName(value.name))
+		metricNameString := managePropertyName(value.name)
 		if value.min != nil {
 			metricList.Put(fmt.Sprintf("%s.%s", metricNameString, "min"), *value.min)
 		}
@@ -222,15 +248,15 @@ func createEvent(timestamp time.Time, metric Metric, resource Resource, metricVa
 // getDimensionValue will return dimension value for the key provided
 func getDimensionValue(dimension string, dimensions []Dimension) string {
 	for _, dim := range dimensions {
-		if strings.ToLower(dim.Name) == strings.ToLower(dimension) {
+		if strings.EqualFold(dim.Name, dimension) {
 			return dim.Value
 		}
 	}
 	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")