azuremonitorreceiver: discover and collect custom metric namespaces (#49035)

## Description

The `MetricDefinitions` API only returns custom metric namespaces (e.g.
`azure.vm.linux.guestmetrics` published by AMA/MetricsExtension) when
the `metricnamespace` query parameter is explicitly set. Without it,
only standard platform metrics are returned, so any metrics configured
under `receiver::metrics` that belong to a custom namespace were
silently dropped — the receiver never discovered or collected them.

### Root cause

`loadMetricsDefinitions` called
`MetricDefinitionsClient.NewListPager(resourceID, nil)` (nil = no
namespace filter). Azure Monitor returns definitions only for the
resource's default namespace (e.g. `Microsoft.Compute/virtualMachines`).
Custom namespaces like `azure.vm.linux.guestmetrics` require an explicit
`metricnamespace` query parameter.

Additionally, the `Metrics` (values) API also requires the
`metricnamespace` parameter when querying a non-default namespace, so
`newResourceMetricsValuesRequestOptions` needed the same fix. The batch
scraper's `QueryResources` call had the same issue: it was passing
`resourceType` as the namespace parameter instead of the actual metric
namespace from the composite key.

### Changes

- **`scraper.go`**: Add `namespace` field to `metricsCompositeKey` so
metrics from different namespaces are batched in separate `Metrics` API
calls (Azure Monitor accepts only one namespace per request). Refactor
`loadMetricsDefinitions` into a `collectMetricDefinitions` helper that
accepts optional `MetricDefinitionsClientListOptions`. After the default
(no-filter) call, make one additional call per namespace configured in
`receiver.metrics` that was not already discovered, using the explicit
`metricnamespace` parameter. Pass `compositeKey.namespace` to
`newResourceMetricsValuesRequestOptions`.
- **`scraper_batch.go`**: Apply the same two-pass fix. Extract the pager
loop into a `collectMetricDefinitionsByType` helper with
`discoveredNamespaces` tracking. After the default call, make one
additional namespace-filtered call per custom namespace in `cfg.Metrics`
not already returned. Pass `compositeKey.namespace` instead of
`resourceType` to `QueryResources` so the correct namespace is used in
the batch values API call.
- **`scraper_test.go`**: Add
`TestAzureScraperScrapeCustomNamespaceMetrics` covering the end-to-end
flow: a VM with no standard metrics but a custom namespace metric
(`azure.vm.linux.guestmetrics/disk/free_percent`) is fully discovered
and collected.
- **`scraper_batch_test.go`**: Update mock data to use the actual
namespace values from metric definitions (instead of the resource type
string). Add `TestAzureScraperBatchScrapeCustomNamespaceMetrics`
covering the same end-to-end flow for the batch scraper path.
- **`mocks_test.go`**: Fix the mock to return an empty success page
(instead of a zero `PagerResponder`) when no matching key exists. The
Azure SDK fake transport uses the URL path as the tracker key (ignoring
query params), so without this fix the empty `PagerResponder` from the
default call was reused for subsequent namespace-filtered calls,
preventing the mock from being invoked again.
- **`testdata/expected_metrics/metrics_custom_namespace.yaml`**: Golden
file for the new tests.
- **`.chloggen/40989-azuremonitor-custom-namespace-discovery.yaml`**:
Changelog entry for the bug fix.
- **`README.md`**: Add a note explaining custom namespace API behavior,
an example config for `azure.vm.linux.guestmetrics`, and a cardinality
note in the API calls section.

### Testing

```
go test ./receiver/azuremonitorreceiver/... -v -run TestAzureScraperScrape
```

All existing tests continue to pass. The new tests
`TestAzureScraperScrapeCustomNamespaceMetrics` and
`TestAzureScraperBatchScrapeCustomNamespaceMetrics` cover the custom
namespace discovery path for both scraper implementations.

### Related issues

This fix enables collection of metrics published by the Azure Monitor
Agent (AMA) / MetricsExtension to custom namespaces, which is necessary
for guest OS metrics (disk, memory, CPU from within the VM) that are not
available through standard Azure platform metrics.

Fixes:
#40989

Author: Zapulin

.chloggen/40989-azuremonitor-custom-namespace-discovery.yaml

@@ -0,0 +1,13 @@
+change_type: bug_fix
+component: receiver/azure_monitor
+note: Fix discovery and collection of custom metric namespace definitions (e.g. `azure.vm.linux.guestmetrics` published by Azure Monitor Agent / MetricsExtension)
+issues: [40989]
+subtext: >
+  The MetricDefinitions API only returns custom namespace metrics when the
+  `metricnamespace` query parameter is explicitly set. Previously, metrics
+  configured under `receiver::metrics` for a custom namespace were silently
+  dropped because the API call used no filter and only returned the
+  resource's default namespace. The receiver now makes an additional
+  namespace-filtered call for each custom namespace in the `metrics` config
+  that was not returned by the default call.
+change_logs: [user]

receiver/azuremonitorreceiver/README.md

@@ -71,6 +71,11 @@ It accepts a nested map where
     In this case, the scraper will fetch **all supported aggregations** for that metric, which is also the case if no
     `metrics` configuration is provided.
 > - **Case Insensitive**: The letter case of the Namespaces, Metric names, and Aggregations does not affect the functionality.
+> - **Custom metric namespaces**: Metrics published by the [Azure Monitor Agent (AMA)](https://learn.microsoft.com/en-us/azure/azure-monitor/agents/azure-monitor-agent-overview)
+    or MetricsExtension (e.g. guest OS metrics under `azure.vm.linux.guestmetrics`) belong to a custom namespace that
+    the MetricDefinitions API only returns when queried with an explicit `metricnamespace` filter. To collect these
+    metrics, add the custom namespace as a top-level key under `metrics`; the receiver will issue the necessary
+    namespace-filtered discovery call automatically.
 
 > [!WARNING]  
 > If you started providing a `metrics` configuration for a namespace, you have to specify all the metrics and their 
@@ -93,6 +98,18 @@ receivers:
         ActiveConnections: []         # metric ActiveConnections with all known aggregations (same effect than [*])
 ```
 
+Scraping guest OS metrics from a custom namespace (e.g. published by Azure Monitor Agent):
+
+```yaml
+receivers:
+  azure_monitor:
+    services:
+      - Microsoft.Compute/virtualMachines
+    metrics:
+      "azure.vm.linux.guestmetrics":
+        "filesystem % free space": [Average]  # metric names depend on agent config; check MetricDefinitions API for your namespace
+```
+
 ### Use Batch API (experimental)
 
 There's two API to collect metrics in Azure Monitor:
@@ -248,6 +265,8 @@ conditions: always
 cardinality:
   - if use_batch_api is false, once per res id and *page of metrics def
   - if use_batch_api is true, once per res type and *page of metrics def
+  - one additional call per custom namespace configured under `metrics` that
+    was not returned by the default call (e.g. azure.vm.linux.guestmetrics)
 ```
 
 ### [Metrics - List](https://learn.microsoft.com/en-us/rest/api/monitor/metrics/list?view=rest-monitor-2023-10-01&tabs=HTTP)

receiver/azuremonitorreceiver/mocks_test.go

@@ -59,11 +59,28 @@ func newMockResourcesListPager(resourcesPages []armresources.ClientListResponse)
 
 // newMockMetricsDefinitionListPager is a helper function to create a list pager for metrics definitions.
 // Don't use it, it's designed to be called in the newMockClientOptionsResolver ctor only.
+//
+// The map key is either a plain resourceURI (for the default no-filter call) or
+// "resourceURI::namespace" (for namespace-filtered calls). When a namespace-filtered
+// call is made and no namespace-specific key exists, an empty response is returned
+// rather than falling back to the plain-URI key, which prevents duplicate definitions.
 func newMockMetricsDefinitionListPager(metricDefinitionsPagesByResourceURI map[string][]armmonitor.MetricDefinitionsClientListResponse) func(resourceURI string, options *armmonitor.MetricDefinitionsClientListOptions) (resp azfake.PagerResponder[armmonitor.MetricDefinitionsClientListResponse]) {
-	return func(resourceURI string, _ *armmonitor.MetricDefinitionsClientListOptions) (resp azfake.PagerResponder[armmonitor.MetricDefinitionsClientListResponse]) {
+	return func(resourceURI string, options *armmonitor.MetricDefinitionsClientListOptions) (resp azfake.PagerResponder[armmonitor.MetricDefinitionsClientListResponse]) {
 		resourceURI = fmt.Sprintf("/%s", resourceURI) // Hack the fake API as it's not taking starting slash from called request
-		for _, page := range metricDefinitionsPagesByResourceURI[resourceURI] {
-			resp.AddPage(http.StatusOK, page, nil)
+		key := resourceURI
+		if options != nil && options.Metricnamespace != nil {
+			key = resourceURI + "::" + *options.Metricnamespace
+		}
+		pages, found := metricDefinitionsPagesByResourceURI[key]
+		if found {
+			for _, page := range pages {
+				resp.AddPage(http.StatusOK, page, nil)
+			}
+		} else {
+			// Return an empty success page so the fake transport removes this tracker entry,
+			// allowing subsequent calls with different query params (e.g. metricnamespace) to
+			// create a fresh tracker entry and invoke the mock handler again.
+			resp.AddPage(http.StatusOK, armmonitor.MetricDefinitionsClientListResponse{}, nil)
 		}
 		return resp
 	}

receiver/azuremonitorreceiver/scraper.go

@@ -78,6 +78,7 @@ type azureResource struct {
 }
 
 type metricsCompositeKey struct {
+	namespace    string
 	dimensions   string // comma separated sorted dimensions
 	aggregations string // comma separated sorted aggregations
 	timeGrain    string
@@ -483,8 +484,46 @@ func (s *azureScraper) loadMetricsDefinitions(ctx context.Context, subscriptionI
 		return
 	}
 
-	pager := clientMetricsDefinitions.NewListPager(resourceID, nil)
+	// discoveredNamespaces tracks namespaces returned by the default (no-filter) call
+	// so we can skip redundant namespace-filtered calls for those namespaces.
+	discoveredNamespaces := map[string]struct{}{}
 
+	s.collectMetricDefinitions(ctx, subscriptionID, resourceID, clientMetricsDefinitions, nil, discoveredNamespaces)
+
+	// The Azure Monitor MetricDefinitions API only returns custom metric namespace
+	// definitions (e.g. "azure.vm.linux.guestmetrics" published by AMA/MetricsExtension)
+	// when the metricnamespace query parameter is set explicitly. Make additional calls
+	// for each namespace configured in the metrics filter that was not already returned
+	// by the default call above.
+	for configNamespace := range s.cfg.Metrics {
+		if _, found := discoveredNamespaces[strings.ToLower(configNamespace)]; found {
+			continue
+		}
+		opts := &armmonitor.MetricDefinitionsClientListOptions{
+			Metricnamespace: to.Ptr(configNamespace),
+		}
+		s.collectMetricDefinitions(ctx, subscriptionID, resourceID, clientMetricsDefinitions, opts, nil)
+	}
+
+	s.resources[subscriptionID][resourceID].metricsDefinitionsUpdated = time.Now()
+	s.settings.Logger.Info("Loaded the list of Azure Metrics Definitions",
+		zap.Int("metrics_definitions_count", len(s.resources[subscriptionID][resourceID].metricsByCompositeKey)),
+		zap.String("resource_id", resourceID),
+		zap.String("subscription_id", subscriptionID))
+}
+
+// collectMetricDefinitions pages through a MetricDefinitions pager and registers each
+// metric definition into the resource's metricsByCompositeKey map.
+// discoveredNamespaces, when non-nil, is populated with the lowercased namespaces seen in
+// the response so callers can skip redundant follow-up calls.
+func (s *azureScraper) collectMetricDefinitions(
+	ctx context.Context,
+	subscriptionID, resourceID string,
+	client *armmonitor.MetricDefinitionsClient,
+	opts *armmonitor.MetricDefinitionsClientListOptions,
+	discoveredNamespaces map[string]struct{},
+) {
+	pager := client.NewListPager(resourceID, opts)
 	page := 0
 	for pager.More() {
 		nextResult, err := pager.NextPage(ctx)
@@ -505,27 +544,28 @@ func (s *azureScraper) loadMetricsDefinitions(ctx context.Context, subscriptionI
 
 		for _, v := range nextResult.Value {
 			metricName := *v.Name.Value
-			metricAggregations := getMetricAggregations(*v.Namespace, metricName, s.cfg.Metrics, convertAggregationsToStr(v.SupportedAggregationTypes))
+			metricNamespace := *v.Namespace
+
+			if discoveredNamespaces != nil {
+				discoveredNamespaces[strings.ToLower(metricNamespace)] = struct{}{}
+			}
+
+			metricAggregations := getMetricAggregations(metricNamespace, metricName, s.cfg.Metrics, convertAggregationsToStr(v.SupportedAggregationTypes))
 			if len(metricAggregations) == 0 {
 				continue
 			}
 
 			timeGrain := *v.MetricAvailabilities[0].TimeGrain
 			dimensions := filterDimensions(v.Dimensions, s.cfg.Dimensions, *s.resources[subscriptionID][resourceID].resourceType, metricName)
 			compositeKey := metricsCompositeKey{
+				namespace:    metricNamespace,
 				timeGrain:    timeGrain,
 				dimensions:   serializeDimensions(dimensions),
 				aggregations: strings.Join(metricAggregations, ","),
 			}
 			s.loadMetricsDefinition(subscriptionID, resourceID, metricName, compositeKey)
 		}
 	}
-
-	s.resources[subscriptionID][resourceID].metricsDefinitionsUpdated = time.Now()
-	s.settings.Logger.Info("Loaded the list of Azure Metrics Definitions",
-		zap.Int("metrics_definitions_count", len(s.resources[subscriptionID][resourceID].metricsByCompositeKey)),
-		zap.String("resource_id", resourceID),
-		zap.String("subscription_id", subscriptionID))
 }
 
 func (s *azureScraper) loadMetricsDefinition(subscriptionID, resourceID, metricName string, compositeKey metricsCompositeKey) {
@@ -577,6 +617,7 @@ func (s *azureScraper) loadMetricsValues(ctx context.Context, subscriptionID, re
 				compositeKey.dimensions,
 				compositeKey.timeGrain,
 				compositeKey.aggregations,
+				compositeKey.namespace,
 				start,
 				end,
 				s.cfg.MaximumNumberOfRecordsPerResource,
@@ -638,18 +679,23 @@ func newResourceMetricsValuesRequestOptions(
 	dimensionsStr string,
 	timeGrain string,
 	aggregationsStr string,
+	namespace string,
 	start int,
 	end int,
 	top int32,
 ) armmonitor.MetricsClientListOptions {
-	return armmonitor.MetricsClientListOptions{
+	opts := armmonitor.MetricsClientListOptions{
 		Metricnames: to.Ptr(strings.Join(metrics[start:end], ",")),
 		Interval:    to.Ptr(timeGrain),
 		Timespan:    to.Ptr(timeGrain),
 		Aggregation: to.Ptr(aggregationsStr),
 		Top:         to.Ptr(top),
 		Filter:      buildDimensionsFilter(dimensionsStr),
 	}
+	if namespace != "" {
+		opts.Metricnamespace = to.Ptr(namespace)
+	}
+	return opts
 }
 
 func (s *azureScraper) processTimeseriesData(

receiver/azuremonitorreceiver/scraper_batch.go

@@ -442,7 +442,44 @@ func (s *azureBatchScraper) loadResourceMetricsDefinitionsByType(ctx context.Con
 		return
 	}
 
-	pager := clientMetricsDefinitions.NewListPager(resourceIDs[0], nil)
+	discoveredNamespaces := map[string]struct{}{}
+
+	s.collectMetricDefinitionsByType(ctx, subscriptionID, resourceType, resourceIDs[0], clientMetricsDefinitions, nil, discoveredNamespaces)
+
+	// The Azure Monitor MetricDefinitions API only returns custom metric namespace
+	// definitions (e.g. "azure.vm.linux.guestmetrics" published by AMA/MetricsExtension)
+	// when the metricnamespace query parameter is set explicitly. Make additional calls
+	// for each namespace configured in the metrics filter that was not already returned
+	// by the default call above.
+	for configNamespace := range s.cfg.Metrics {
+		if _, found := discoveredNamespaces[strings.ToLower(configNamespace)]; found {
+			continue
+		}
+		opts := &armmonitor.MetricDefinitionsClientListOptions{
+			Metricnamespace: to.Ptr(configNamespace),
+		}
+		s.collectMetricDefinitionsByType(ctx, subscriptionID, resourceType, resourceIDs[0], clientMetricsDefinitions, opts, nil)
+	}
+
+	s.resourceTypes[subscriptionID][resourceType].metricsDefinitionsUpdated = time.Now()
+	s.settings.Logger.Info("Loaded the list of Azure Metrics Definitions",
+		zap.Int("metrics_definitions_count", len(s.resourceTypes[subscriptionID][resourceType].metricsByCompositeKey)),
+		zap.String("resource_type", resourceType),
+		zap.String("subscription_id", subscriptionID))
+}
+
+// collectMetricDefinitionsByType pages through a MetricDefinitions pager and registers each
+// metric definition into the resourceType's metricsByCompositeKey map.
+// discoveredNamespaces, when non-nil, is populated with the lowercased namespaces seen in this call.
+// TODO: Partially duplicate of collectMetricDefinitions in scraper.go
+func (s *azureBatchScraper) collectMetricDefinitionsByType(
+	ctx context.Context,
+	subscriptionID, resourceType, resourceID string,
+	clientMetricsDefinitions *armmonitor.MetricDefinitionsClient,
+	opts *armmonitor.MetricDefinitionsClientListOptions,
+	discoveredNamespaces map[string]struct{},
+) {
+	pager := clientMetricsDefinitions.NewListPager(resourceID, opts)
 	page := 0
 	for pager.More() {
 		nextResult, err := pager.NextPage(ctx)
@@ -462,26 +499,28 @@ func (s *azureBatchScraper) loadResourceMetricsDefinitionsByType(ctx context.Con
 
 		for _, v := range nextResult.Value {
 			metricName := *v.Name.Value
-			metricAggregations := getMetricAggregations(*v.Namespace, metricName, s.cfg.Metrics, convertAggregationsToStr(v.SupportedAggregationTypes))
+			metricNamespace := *v.Namespace
+
+			if discoveredNamespaces != nil {
+				discoveredNamespaces[strings.ToLower(metricNamespace)] = struct{}{}
+			}
+
+			metricAggregations := getMetricAggregations(metricNamespace, metricName, s.cfg.Metrics, convertAggregationsToStr(v.SupportedAggregationTypes))
 			if len(metricAggregations) == 0 {
 				continue
 			}
 
 			timeGrain := *v.MetricAvailabilities[0].TimeGrain
 			dimensions := filterDimensions(v.Dimensions, s.cfg.Dimensions, resourceType, metricName)
 			compositeKey := metricsCompositeKey{
+				namespace:    metricNamespace,
 				timeGrain:    timeGrain,
 				dimensions:   serializeDimensions(dimensions),
 				aggregations: strings.Join(metricAggregations, ","),
 			}
 			s.loadMetricsDefinitionByType(subscriptionID, resourceType, metricName, compositeKey)
 		}
 	}
-	s.resourceTypes[subscriptionID][resourceType].metricsDefinitionsUpdated = time.Now()
-	s.settings.Logger.Info("Loaded the list of Azure Metrics Definitions",
-		zap.Int("metrics_definitions_count", len(s.resourceTypes[subscriptionID][resourceType].metricsByCompositeKey)),
-		zap.String("resource_type", resourceType),
-		zap.String("subscription_id", subscriptionID))
 }
 
 // TODO: duplicate
@@ -580,7 +619,7 @@ func (s *azureBatchScraper) loadBatchMetricsValues(ctx context.Context, subscrip
 					response, err := clientMetrics.QueryResources(
 						ctx,
 						subscriptionID,
-						resourceType,
+						compositeKey.namespace,
 						metricsByGrain.metrics[start:end],
 						azmetrics.ResourceIDList{ResourceIDs: resType.resourceIDs[startResources:endResources]},
 						&opts,