Azure Monitor: adjust grouping logic and avoid duplicating documents to make the metricset TSDB-friendly

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

@@ -9,6 +9,7 @@ import (
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
 
+// addHostMetadata enriches the event with host metadata.
 func addHostMetadata(event *mb.Event, metricList mapstr.M) {
 	hostFieldTable := map[string]string{
 		"percentage_cpu.avg":      "host.cpu.usage",
@@ -30,24 +31,28 @@ func addHostMetadata(event *mb.Event, metricList mapstr.M) {
 			if metricName == "percentage_cpu.avg" {
 				value = value / 100
 			}
-			event.RootFields.Put(hostName, value)
+			_, _ = event.RootFields.Put(hostName, value)
 		}
 	}
 }
 
+// addCloudVMMetadata enriches the event with cloud VM metadata.
 func addCloudVMMetadata(event *mb.Event, vm VmResource, subscriptionId string) {
 	if vm.Name != "" {
-		event.RootFields.Put("cloud.instance.name", vm.Name)
-		event.RootFields.Put("host.name", vm.Name)
+		_, _ = event.RootFields.Put("cloud.instance.name", vm.Name)
+		_, _ = event.RootFields.Put("host.name", vm.Name)
 	}
+
 	if vm.Id != "" {
-		event.RootFields.Put("cloud.instance.id", vm.Id)
-		event.RootFields.Put("host.id", vm.Id)
+		_, _ = event.RootFields.Put("cloud.instance.id", vm.Id)
+		_, _ = event.RootFields.Put("host.id", vm.Id)
 	}
+
 	if vm.Size != "" {
-		event.RootFields.Put("cloud.machine.type", vm.Size)
+		_, _ = event.RootFields.Put("cloud.machine.type", vm.Size)
 	}
+
 	if subscriptionId != "" {
-		event.RootFields.Put("cloud.account.id", subscriptionId)
+		_, _ = event.RootFields.Put("cloud.account.id", subscriptionId)
 	}
 }

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

@@ -87,24 +87,43 @@ func NewMetricSet(base mb.BaseMetricSet) (*MetricSet, error) {
 // It publishes the event which is then forwarded to the output. In case
 // of an error set the Error field of mb.Event or simply call report.Error().
 func (m *MetricSet) Fetch(report mb.ReporterV2) error {
+	// Initialize cloud resources and monitor metrics
+	// information.
+	//
+	// The client collects and stores:
+	// - existing cloud resource definitions (e.g. VMs, DBs, etc.)
+	// - metric definitions for the resources (e.g. CPU, memory, etc.)
+	//
+	// The metricset periodically refreshes the information
+	// after `RefreshListInterval` (default 600s for
+	// most metricsets).
 	err := m.Client.InitResources(m.MapMetrics)
 	if err != nil {
 		return err
 	}
+
 	if len(m.Client.ResourceConfigurations.Metrics) == 0 {
-		// error message is previously logged in the InitResources, no error event should be created
+		// error message is previously logged in the InitResources,
+		// no error event should be created
 		return nil
 	}
-	// retrieve metrics
-	groupedMetrics := groupMetricsByResource(m.Client.ResourceConfigurations.Metrics)
 
-	for _, metrics := range groupedMetrics {
-		results := m.Client.GetMetricValues(metrics, report)
-		err := EventsMapping(results, m.Client, report)
-		if err != nil {
-			return fmt.Errorf("error running EventsMapping: %w", err)
+	// Group metric definitions by cloud resource ID.
+	//
+	// We group the metric definitions by resource ID to fetch
+	// metric values for each cloud resource in one API call.
+	metricsByResourceId := groupMetricsDefinitionsByResourceId(m.Client.ResourceConfigurations.Metrics)
+
+	for _, metricsDefinition := range metricsByResourceId {
+		// Fetch metric values for each resource.
+		metricValues := m.Client.GetMetricValues(metricsDefinition, report)
+
+		// Turns metric values into events and sends them to Elasticsearch.
+		if err := mapToEvents(metricValues, m.Client, report); err != nil {
+			return fmt.Errorf("error mapping metrics to events: %w", err)
 		}
 	}
+
 	return nil
 }