feat(receiver/azuremonitor): Adds filtering by metric and/or aggregation (#37421)

#### Description
Adds optional setting `metrics` that allows to limit scrapping to the
specific metrics and aggregations.

#### Link to tracking issue
Fixes
#37420

#### Testing
Unit test coverage extended to verify default behaviour and to cover a
new functionality

#### Documentation
Documentation updated respectfully to reflect changes

Author: an-mmx

.chloggen/feat_azuremonitorreceiver_filter_metrics.yaml

@@ -0,0 +1,28 @@
+# 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: azuremonitorreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adds filtering by metric and/or aggregation
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37420]
+
+# (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]
+

receiver/azuremonitorreceiver/README.md

@@ -25,6 +25,7 @@ The following settings are optional:
 - `auth` (default = service_principal): Specifies the used authentication method. Supported values are `service_principal`, `workload_identity`, `managed_identity`, `default_credentials`.
 - `resource_groups` (default = none): Filter metrics for specific resource groups, not setting a value will scrape metrics for all resources in the subscription.
 - `services` (default = none): Filter metrics for specific services, not setting a value will scrape metrics for all services integrated with Azure Monitor.
+- `metrics` (default = none): Filter metrics by name and aggregations. Not setting a value will scrape all metrics and their aggregations.
 - `cache_resources` (default = 86400): List of resources will be cached for the provided amount of time in seconds.
 - `cache_resources_definitions` (default = 86400): List of metrics definitions will be cached for the provided amount of time in seconds.
 - `maximum_number_of_metrics_in_a_call` (default = 20): Maximum number of metrics to fetch in per API call, current limit in Azure is 20 (as of 03/27/2023).
@@ -50,6 +51,26 @@ Authenticating using managed identities has the following optional settings:
 
 - `client_id`
 
+### Filtering metrics
+
+The `metrics` configuration setting is designed to limit scraping to specific metrics and their particular aggregations. It accepts a nested map where the key of the top-level is the Azure Metric Namespace, the key of the nested map is an Azure Metric Name, and the map values are a list of aggregation methods (e.g., Average, Minimum, Maximum, Total, Count). Additionally, the metric map value can be an empty array or an array with one element `*` (asterisk). In this case, the scraper will fetch all supported aggregations for a metric. The letter case of the Namespaces, Metric names, and Aggregations does not affect the functionality.
+
+Scraping limited metrics and aggregations:
+
+```yaml
+receivers:
+  azuremonitor:
+    resource_groups:
+      - ${resource_groups}
+    services:
+      - Microsoft.EventHub/namespaces
+      - Microsoft.AAD/DomainServices # scraper will fetch all metrics from this namespace since there are no limits under the "metrics" option
+    metrics:
+      "microsoft.eventhub/namespaces": # scraper will fetch only the metrics listed below:
+        IncomingMessages: [total]     # metric IncomingMessages with aggregation "Total"
+        NamespaceCpuUsage: [*]        # metric NamespaceCpuUsage with all known aggregations
+```
+
 ### Example Configurations
 
 Using [Service Principal](https://learn.microsoft.com/en-us/azure/developer/go/azure-sdk-authentication?tabs=bash#service-principal-with-a-secret) for authentication:
@@ -104,6 +125,7 @@ receivers:
 ```
 
 Overriding dimensions for a particular metric:
+
 ```yaml
 receivers:
   azuremonitor:
@@ -119,7 +141,6 @@ receivers:
           "NetworkRuleHit": [Reason, Status]
 ```
 
-
 ## Metrics
 
 Details about the metrics scraped by this receiver can be found in [Supported metrics with Azure Monitor](https://learn.microsoft.com/en-us/azure/azure-monitor/essentials/metrics-supported). This receiver adds the prefix "azure_" to all scraped metrics.

receiver/azuremonitorreceiver/config.go

@@ -228,9 +228,11 @@ var (
 	}
 )
 
+type NestedListAlias = map[string]map[string][]string
+
 type DimensionsConfig struct {
-	Enabled   *bool                          `mapstructure:"enabled"`
-	Overrides map[string]map[string][]string `mapstructure:"overrides"`
+	Enabled   *bool           `mapstructure:"enabled"`
+	Overrides NestedListAlias `mapstructure:"overrides"`
 }
 
 // Config defines the configuration for the various elements of the receiver agent.
@@ -246,6 +248,7 @@ type Config struct {
 	FederatedTokenFile                string                        `mapstructure:"federated_token_file"`
 	ResourceGroups                    []string                      `mapstructure:"resource_groups"`
 	Services                          []string                      `mapstructure:"services"`
+	Metrics                           NestedListAlias               `mapstructure:"metrics"`
 	CacheResources                    float64                       `mapstructure:"cache_resources"`
 	CacheResourcesDefinitions         float64                       `mapstructure:"cache_resources_definitions"`
 	MaximumNumberOfMetricsInACall     int                           `mapstructure:"maximum_number_of_metrics_in_a_call"`

receiver/azuremonitorreceiver/dimension_test.go

@@ -65,7 +65,7 @@ func TestFilterDimensions(t *testing.T) {
 				},
 				cfg: DimensionsConfig{
 					Enabled: to.Ptr(true),
-					Overrides: map[string]map[string][]string{
+					Overrides: NestedListAlias{
 						"rt1": {
 							"m1": {
 								"foo",

receiver/azuremonitorreceiver/scraper.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"regexp"
+	"slices"
 	"strings"
 	"sync"
 	"time"
@@ -56,6 +57,7 @@ const (
 	metadataPrefix         = "metadata_"
 	tagPrefix              = "tags_"
 	truncateTimeGrain      = time.Minute
+	filterAllAggregations  = "*"
 )
 
 type azureResource struct {
@@ -67,8 +69,9 @@ type azureResource struct {
 }
 
 type metricsCompositeKey struct {
-	dimensions string // comma separated sorted dimensions
-	timeGrain  string
+	dimensions   string // comma separated sorted dimensions
+	aggregations string // comma separated sorted aggregations
+	timeGrain    string
 }
 
 type azureResourceMetrics struct {
@@ -350,12 +353,18 @@ func (s *azureScraper) getResourceMetricsDefinitions(ctx context.Context, resour
 		}
 
 		for _, v := range nextResult.Value {
-			timeGrain := *v.MetricAvailabilities[0].TimeGrain
 			metricName := *v.Name.Value
+			metricAggregations := getMetricAggregations(*v.Namespace, metricName, s.cfg.Metrics)
+			if len(metricAggregations) == 0 {
+				continue
+			}
+
+			timeGrain := *v.MetricAvailabilities[0].TimeGrain
 			dimensions := filterDimensions(v.Dimensions, s.cfg.Dimensions, *s.resources[resourceID].resourceType, metricName)
 			compositeKey := metricsCompositeKey{
-				timeGrain:  timeGrain,
-				dimensions: serializeDimensions(dimensions),
+				timeGrain:    timeGrain,
+				dimensions:   serializeDimensions(dimensions),
+				aggregations: strings.Join(metricAggregations, ","),
 			}
 			s.storeMetricsDefinition(resourceID, metricName, compositeKey)
 		}
@@ -395,6 +404,7 @@ func (s *azureScraper) getResourceMetricsValues(ctx context.Context, resourceID
 				metricsByGrain.metrics,
 				compositeKey.dimensions,
 				compositeKey.timeGrain,
+				compositeKey.aggregations,
 				start,
 				end,
 				s.cfg.MaximumNumberOfRecordsPerResource,
@@ -442,6 +452,7 @@ func getResourceMetricsValuesRequestOptions(
 	metrics []string,
 	dimensionsStr string,
 	timeGrain string,
+	aggregationsStr string,
 	start int,
 	end int,
 	top int32,
@@ -450,7 +461,7 @@ func getResourceMetricsValuesRequestOptions(
 		Metricnames: to.Ptr(strings.Join(metrics[start:end], ",")),
 		Interval:    to.Ptr(timeGrain),
 		Timespan:    to.Ptr(timeGrain),
-		Aggregation: to.Ptr(strings.Join(aggregations, ",")),
+		Aggregation: to.Ptr(aggregationsStr),
 		Top:         to.Ptr(top),
 		Filter:      buildDimensionsFilter(dimensionsStr),
 	}
@@ -491,3 +502,49 @@ func (s *azureScraper) processTimeseriesData(
 		}
 	}
 }
+
+func getMetricAggregations(metricNamespace, metricName string, filters NestedListAlias) []string {
+	// default behavior when no metric filters specified: pass all metrics with all aggregations
+	if len(filters) == 0 {
+		return aggregations
+	}
+
+	metricsFilters, ok := mapFindInsensitive(filters, metricNamespace)
+	// metric namespace not found or it's empty: pass all metrics from the namespace
+	if !ok || len(metricsFilters) == 0 {
+		return aggregations
+	}
+
+	aggregationsFilters, ok := mapFindInsensitive(metricsFilters, metricName)
+	// if target metric is absent in metrics map: filter out metric
+	if !ok {
+		return []string{}
+	}
+	// allow all aggregations if others are not specified
+	if len(aggregationsFilters) == 0 || slices.Contains(aggregationsFilters, filterAllAggregations) {
+		return aggregations
+	}
+
+	// collect known supported aggregations
+	out := []string{}
+	for _, filter := range aggregationsFilters {
+		for _, aggregation := range aggregations {
+			if strings.EqualFold(aggregation, filter) {
+				out = append(out, aggregation)
+			}
+		}
+	}
+
+	return out
+}
+
+func mapFindInsensitive[T any](m map[string]T, key string) (T, bool) {
+	for k, v := range m {
+		if strings.EqualFold(key, k) {
+			return v, true
+		}
+	}
+
+	var got T
+	return got, false
+}