Use rate limiter with all interactions with API

The metricset uses multiple API endpoint, so we need to double-check
if the max request rate applies to all endpoints or single ones.

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

@@ -6,6 +6,7 @@ package azure
 
 import (
 	"fmt"
+	"golang.org/x/time/rate"
 	"time"
 
 	"github.com/elastic/elastic-agent-libs/mapstr"
@@ -41,6 +42,8 @@ type Config struct {
 	RefreshListInterval time.Duration    `config:"refresh_list_interval"`
 	DefaultResourceType string           `config:"default_resource_type"`
 	AddCloudMetadata    bool             `config:"add_cloud_metadata"`
+	RateLimit           rate.Limit       `config:"rate_limit"`
+	RateBurst           int              `config:"rate_burst"`
 	// specific to billing
 	BillingScopeDepartment string `config:"billing_scope_department"` // retrieve usage details from department scope
 	BillingScopeAccountId  string `config:"billing_scope_account_id"` // retrieve usage details from billing account ID scope

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

@@ -7,6 +7,7 @@ package azure
 import (
 	"context"
 	"fmt"
+	"golang.org/x/time/rate"
 	"strings"
 
 	"github.com/elastic/elastic-agent-libs/logp"
@@ -27,6 +28,7 @@ type MonitorService struct {
 	resourceClient         *armresources.Client
 	context                context.Context
 	log                    *logp.Logger
+	limiter                *rate.Limiter
 }
 
 const (
@@ -91,13 +93,28 @@ func NewService(config Config) (*MonitorService, error) {
 		return nil, fmt.Errorf("couldn't create metric namespaces client: %w", err)
 	}
 
+	// Rate limiter to avoid hitting the Azure Monitoring API rate limits.
+	// According to https://learn.microsoft.com/en-us/azure/azure-monitor/service-limits
+	// the maximum number of queries is 200 per 30 seconds.
+	rateLimit := rate.Limit(6)
+	rateBurst := 1
+
+	if config.RateLimit >= 0 {
+		rateLimit = config.RateLimit
+	}
+	if config.RateBurst > 0 {
+		rateBurst = config.RateBurst
+	}
+	limiter := rate.NewLimiter(rateLimit, rateBurst) // 6 requests per second (200/30 = 6.67, rounded down to 6 requests per second
+
 	service := &MonitorService{
 		metricDefinitionClient: metricsDefinitionClient,
 		metricsClient:          metricsClient,
 		metricNamespaceClient:  metricNamespaceClient,
 		resourceClient:         resourceClient,
 		context:                context.Background(),
 		log:                    logp.NewLogger("azure monitor service"),
+		limiter:                limiter,
 	}
 
 	return service, nil
@@ -118,6 +135,13 @@ func (service MonitorService) GetResourceDefinitions(id []string, group []string
 			})
 
 			for pager.More() {
+				// Wait for the rate limiter to allow the request to be made to
+				// the Azure Monitoring API before making the request.
+				err := service.limiter.Wait(service.context)
+				if err != nil {
+					return nil, err
+				}
+
 				nextResult, err := pager.NextPage(service.context)
 				if err != nil {
 					return nil, err
@@ -169,6 +193,13 @@ func (service MonitorService) GetResourceDefinitions(id []string, group []string
 
 // GetResourceDefinitionById will retrieve the azure resource based on the resource Id
 func (service MonitorService) GetResourceDefinitionById(id string) (armresources.GenericResource, error) {
+	// Wait for the rate limiter to allow the request to be made to
+	// the Azure Monitoring API before making the request.
+	err := service.limiter.Wait(service.context)
+	if err != nil {
+		return armresources.GenericResource{}, err
+	}
+
 	resp, err := service.resourceClient.GetByID(service.context, id, ApiVersion, nil)
 	if err != nil {
 		return armresources.GenericResource{}, err
@@ -184,6 +215,13 @@ func (service *MonitorService) GetMetricNamespaces(resourceId string) (armmonito
 	metricNamespaceCollection := armmonitor.MetricNamespaceCollection{}
 
 	for pager.More() {
+		// Wait for the rate limiter to allow the request to be made to
+		// the Azure Monitoring API before making the request.
+		err := service.limiter.Wait(service.context)
+		if err != nil {
+			return armmonitor.MetricNamespaceCollection{}, err
+		}
+
 		nextPage, err := pager.NextPage(service.context)
 		if err != nil {
 			return armmonitor.MetricNamespaceCollection{}, err
@@ -208,6 +246,13 @@ func (service *MonitorService) GetMetricDefinitions(resourceId string, namespace
 	metricDefinitionCollection := armmonitor.MetricDefinitionCollection{}
 
 	for pager.More() {
+		// Wait for the rate limiter to allow the request to be made to
+		// the Azure Monitoring API before making the request.
+		err := service.limiter.Wait(service.context)
+		if err != nil {
+			return armmonitor.MetricDefinitionCollection{}, err
+		}
+
 		nextPage, err := pager.NextPage(service.context)
 		if err != nil {
 			return armmonitor.MetricDefinitionCollection{}, err
@@ -257,14 +302,21 @@ func (service *MonitorService) GetMetricValues(resourceId string, namespace stri
 			Metricnames: &metricNames,
 			Timespan:    &timespan,
 			Top:         nil,
-			// Orderby:         &orderBy,
-			ResultType: &resultTypeData,
+			ResultType:  &resultTypeData,
 		}
 
 		if namespace != "" {
 			opts.Metricnamespace = &namespace
 		}
 
+		// Wait for the rate limiter to allow the request to be made to
+		// the Azure Monitoring API before making the request.
+		err := service.limiter.Wait(service.context)
+		if err != nil {
+			// TODO: should we check if the context is cancelled?
+			return metrics, "", err
+		}
+
 		resp, err := service.metricsClient.List(service.context, resourceId, opts)
 
 		// check for applied charges before returning any errors