[Azure Billing] Switch to Cost Management API for forecast data (#32589)

* Switch forecast to Cost Management API

The previous endpoint from Consumption API is no longer maintained
and does not support scope.

* Add comments with findings from tests

After testing with different values of includeFreshPartialCost, the best
the trade-off with the current design of the dashboard is to set it to
false and only use consolidated/final data.

* Drop rows with an unexpected format and continue

Stopping the processing of the whole set of rows when we detect
the wrong row is too hard.

With this change, when something is wrong with a row with it we:
- log the circumstances
- stop processing the current row
- continue with the next ones

CHANGELOG.next.asciidoc

@@ -134,7 +134,9 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 
 
 *Metricbeat*
+
 - Allow filtering on AWS tags by more than 1 value per key. {pull}32775[32775]
+- Azure Billing: switch to Cost Management API for forecast data {pull}32589[32589]
 
 *Packetbeat*
 

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

@@ -55,24 +55,45 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	}, nil
 }
 
+// TimeIntervalOptions represents the options used to retrieve the billing data.
+type TimeIntervalOptions struct {
+	// Usage details start time (UTC).
+	usageStart time.Time
+	// Usage details end time (UTC).
+	usageEnd time.Time
+	// Forecast start time (UTC).
+	forecastStart time.Time
+	// Forecast end time (UTC).
+	forecastEnd time.Time
+}
+
 // Fetch methods implements the data gathering and data conversion to the right metricset
 // 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 {
-	// The time interval is yesterday (00:00:00->23:59:59) in UTC.
-	startTime, endTime := previousDayFrom(time.Now())
+	// reference time used to calculate usage and forecast time intervals.
+	referenceTime := time.Now()
+
+	usageStart, usageEnd := usageIntervalFrom(referenceTime)
+	forecastStart, forecastEnd := forecastIntervalFrom(referenceTime)
+
+	timeIntervalOptions := TimeIntervalOptions{
+		usageStart:    usageStart,
+		usageEnd:      usageEnd,
+		forecastStart: forecastStart,
+		forecastEnd:   forecastEnd,
+	}
 
 	m.log.
-		With("billing.start_time", startTime).
-		With("billing.end_time", endTime).
+		With("billing.reference_time", referenceTime).
 		Infow("Fetching billing data")
 
-	results, err := m.client.GetMetrics(startTime, endTime)
+	results, err := m.client.GetMetrics(timeIntervalOptions)
 	if err != nil {
 		return fmt.Errorf("error retrieving usage information: %w", err)
 	}
 
-	events, err := EventsMapping(m.client.Config.SubscriptionId, results, startTime, endTime)
+	events, err := EventsMapping(m.client.Config.SubscriptionId, results, timeIntervalOptions, m.log)
 	if err != nil {
 		return fmt.Errorf("error mapping events: %w", err)
 	}
@@ -87,9 +108,30 @@ func (m *MetricSet) Fetch(report mb.ReporterV2) error {
 	return nil
 }
 
-// previousDayFrom returns the start/end times (00:00:00->23:59:59 UTC) of the day before, given the `reference` time.
-func previousDayFrom(reference time.Time) (time.Time, time.Time) {
-	startTime := reference.UTC().Truncate(24 * time.Hour).Add((-24) * time.Hour)
-	endTime := startTime.Add(time.Hour * 24).Add(time.Second * (-1))
-	return startTime, endTime
+// usageIntervalFrom returns the start/end times (UTC) of the usage period given the `reference` time.
+//
+// Currently, the usage period is the start/end time (00:00:00->23:59:59 UTC) of the day before the reference time.
+//
+// For example, if the reference time is 2007-01-09 09:41:00Z, the usage period is:
+//   2007-01-08 00:00:00Z -> 2007-01-08 23:59:59Z
+//
+func usageIntervalFrom(reference time.Time) (time.Time, time.Time) {
+	beginningOfDay := reference.UTC().Truncate(24 * time.Hour).Add((-24) * time.Hour)
+	endOfDay := beginningOfDay.Add(time.Hour * 24).Add(time.Second * (-1))
+	return beginningOfDay, endOfDay
+}
+
+// forecastIntervalFrom returns the start/end times (UTC) of the forecast period, given the `reference` time.
+//
+// Currently, the forecast period is the start/end times (00:00:00->23:59:59 UTC) of the current month relative to the
+// reference time.
+//
+// For example, if the reference time is 2007-01-09 09:41:00Z, the forecast period is:
+//   2007-01-01T00:00:00Z -> 2007-01-31:59:59Z
+//
+func forecastIntervalFrom(reference time.Time) (time.Time, time.Time) {
+	referenceUTC := reference.UTC()
+	beginningOfMonth := time.Date(referenceUTC.Year(), referenceUTC.Month(), 1, 0, 0, 0, 0, time.UTC)
+	endOfMonth := beginningOfMonth.AddDate(0, 1, 0).Add(-1 * time.Second)
+	return beginningOfMonth, endOfMonth
 }

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

@@ -11,16 +11,33 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestPreviousDayFrom(t *testing.T) {
-	t.Run("returns the previous day as time interval to collect metrics", func(t *testing.T) {
+func TestUsagePeriodFrom(t *testing.T) {
+	t.Run("returns the start and end times for the usage period", func(t *testing.T) {
 		referenceTime, err := time.Parse("2006-01-02 15:04:05", "2007-01-09 09:41:00")
 		assert.NoError(t, err)
 		expectedStartTime, err := time.Parse("2006-01-02 15:04:05", "2007-01-08 00:00:00")
 		assert.NoError(t, err)
 		expectedEndTime, err := time.Parse("2006-01-02 15:04:05", "2007-01-08 23:59:59")
 		assert.NoError(t, err)
 
-		actualStartTime, actualEndTime := previousDayFrom(referenceTime)
+		actualStartTime, actualEndTime := usageIntervalFrom(referenceTime)
+
+		assert.Equal(t, expectedStartTime, actualStartTime)
+		assert.Equal(t, expectedEndTime, actualEndTime)
+	})
+}
+
+func TestForecastPeriodFrom(t *testing.T) {
+	t.Run("returns the start and end times for the forecast period", func(t *testing.T) {
+		referenceTime, err := time.Parse("2006-01-02 15:04:05", "2007-01-09 09:41:00")
+		assert.NoError(t, err)
+
+		expectedStartTime, err := time.Parse("2006-01-02 15:04:05", "2007-01-01 00:00:00")
+		assert.NoError(t, err)
+		expectedEndTime, err := time.Parse("2006-01-02 15:04:05", "2007-01-31 23:59:59")
+		assert.NoError(t, err)
+
+		actualStartTime, actualEndTime := forecastIntervalFrom(referenceTime)
 
 		assert.Equal(t, expectedStartTime, actualStartTime)
 		assert.Equal(t, expectedEndTime, actualEndTime)

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

@@ -5,10 +5,12 @@
 package billing
 
 import (
+	"context"
 	"fmt"
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/services/consumption/mgmt/2019-10-01/consumption"
+	"github.com/Azure/azure-sdk-for-go/services/costmanagement/mgmt/2019-11-01/costmanagement"
 
 	"github.com/elastic/beats/v7/x-pack/metricbeat/module/azure"
 	"github.com/elastic/elastic-agent-libs/logp"
@@ -21,10 +23,10 @@ type Client struct {
 	Log            *logp.Logger
 }
 
+// Usage contains the usage details and forecast values.
 type Usage struct {
-	UsageDetails  []consumption.BasicUsageDetail
-	ActualCosts   []consumption.Forecast
-	ForecastCosts []consumption.Forecast
+	UsageDetails []consumption.BasicUsageDetail
+	Forecasts    costmanagement.QueryResult
 }
 
 // NewClient builds a new client for the azure billing service
@@ -42,55 +44,73 @@ func NewClient(config azure.Config) (*Client, error) {
 }
 
 // GetMetrics returns the usage detail and forecast values.
-func (client *Client) GetMetrics(startTime time.Time, endTime time.Time) (Usage, error) {
+func (client *Client) GetMetrics(timeOpts TimeIntervalOptions) (Usage, error) {
 	var usage Usage
+
+	//
+	// Establish the requested scope
+	//
+
 	scope := fmt.Sprintf("subscriptions/%s", client.Config.SubscriptionId)
 	if client.Config.BillingScopeDepartment != "" {
 		scope = fmt.Sprintf("/providers/Microsoft.Billing/departments/%s", client.Config.BillingScopeDepartment)
 	} else if client.Config.BillingScopeAccountId != "" {
 		scope = fmt.Sprintf("/providers/Microsoft.Billing/billingAccounts/%s", client.Config.BillingScopeAccountId)
 	}
+
+	//
+	// Fetch the usage details
+	//
+
 	client.Log.
 		With("billing.scope", scope).
-		With("billing.start_time", startTime).
-		With("billing.end_time", endTime).
+		With("billing.usage.start_time", timeOpts.usageStart).
+		With("billing.usage.end_time", timeOpts.usageEnd).
 		Infow("Getting usage details for scope")
 
-	usageDetails, err := client.BillingService.GetUsageDetails(
+	filter := fmt.Sprintf(
+		"properties/usageStart eq '%s' and properties/usageEnd eq '%s'",
+		timeOpts.usageStart.Format(time.RFC3339Nano),
+		timeOpts.usageEnd.Format(time.RFC3339Nano),
+	)
+
+	paginator, err := client.BillingService.GetUsageDetails(
 		scope,
 		"properties/meterDetails",
-		fmt.Sprintf(
-			"properties/usageStart eq '%s' and properties/usageEnd eq '%s'",
-			startTime.Format(time.RFC3339Nano),
-			endTime.Format(time.RFC3339Nano),
-		),
-		"", // skipToken
-		nil,
+		filter,
+		"",  // skipToken, used for paging, not required on the first call.
+		nil, // result page size, defaults to ?
 		consumption.MetrictypeActualCostMetricType,
-		startTime.Format("2006-01-02"), // startDate
-		endTime.Format("2006-01-02"),   // endDate
+		timeOpts.usageStart.Format("2006-01-02"), // startDate
+		timeOpts.usageEnd.Format("2006-01-02"),   // endDate
 	)
 	if err != nil {
 		return usage, fmt.Errorf("retrieving usage details failed in client: %w", err)
 	}
 
-	usage.UsageDetails = usageDetails.Values()
+	for paginator.NotDone() {
+		usage.UsageDetails = append(usage.UsageDetails, paginator.Values()...)
+		if err := paginator.NextWithContext(context.Background()); err != nil {
+			return usage, fmt.Errorf("retrieving usage details failed in client: %w", err)
+		}
+	}
 
 	//
-	// Forecast
+	// Fetch the Forecast
 	//
 
-	actualCosts, err := client.BillingService.GetForecast(fmt.Sprintf("properties/chargeType eq '%s'", "Actual"))
-	if err != nil {
-		return usage, fmt.Errorf("retrieving forecast - actual costs failed in client: %w", err)
-	}
-	usage.ActualCosts = actualCosts
+	client.Log.
+		With("billing.scope", scope).
+		With("billing.forecast.start_time", timeOpts.forecastStart).
+		With("billing.forecast.end_time", timeOpts.forecastEnd).
+		Infow("Getting forecast for scope")
 
-	forecastCosts, err := client.BillingService.GetForecast(fmt.Sprintf("properties/chargeType eq '%s'", "Forecast"))
+	queryResult, err := client.BillingService.GetForecast(scope, timeOpts.forecastStart, timeOpts.forecastEnd)
 	if err != nil {
 		return usage, fmt.Errorf("retrieving forecast - forecast costs failed in client: %w", err)
 	}
-	usage.ForecastCosts = forecastCosts
+
+	usage.Forecasts = queryResult
 
 	return usage, nil
 }

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

@@ -13,6 +13,7 @@ import (
 	"github.com/stretchr/testify/mock"
 
 	"github.com/Azure/azure-sdk-for-go/services/consumption/mgmt/2019-10-01/consumption"
+	"github.com/Azure/azure-sdk-for-go/services/costmanagement/mgmt/2019-11-01/costmanagement"
 
 	"github.com/elastic/beats/v7/x-pack/metricbeat/module/azure"
 )
@@ -22,32 +23,42 @@ var (
 )
 
 func TestClient(t *testing.T) {
-	startTime, endTime := previousDayFrom(time.Now())
+	usageStart, usageEnd := usageIntervalFrom(time.Now())
+	forecastStart, forecastEnd := forecastIntervalFrom(time.Now())
+	opts := TimeIntervalOptions{
+		usageStart:    usageStart,
+		usageEnd:      usageEnd,
+		forecastStart: forecastStart,
+		forecastEnd:   forecastEnd,
+	}
 
 	t.Run("return error not valid query", func(t *testing.T) {
 		client := NewMockClient()
 		client.Config = config
 		m := &MockService{}
-		m.On("GetForecast", mock.Anything).Return([]consumption.Forecast{}, errors.New("invalid query"))
+		m.On("GetForecast", mock.Anything, mock.Anything, mock.Anything).Return(costmanagement.QueryResult{}, errors.New("invalid query"))
 		m.On("GetUsageDetails", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(consumption.UsageDetailsListResultPage{}, nil)
 		client.BillingService = m
-		results, err := client.GetMetrics(startTime, endTime)
+		_, err := client.GetMetrics(opts)
 		assert.Error(t, err)
-		assert.Equal(t, len(results.ActualCosts), 0)
+		//assert.NotNil(t, usage.Forecasts)
+		//assert.True(t, usage.Forecasts.Rows == nil)
+		//assert.Equal(t, len(*usage.Forecasts.Rows), 0)
 		m.AssertExpectations(t)
 	})
 	t.Run("return results", func(t *testing.T) {
 		client := NewMockClient()
 		client.Config = config
 		m := &MockService{}
-		forecasts := []consumption.Forecast{{}, {}}
-		m.On("GetForecast", mock.Anything).Return(forecasts, nil)
+		forecasts := costmanagement.QueryResult{}
+		m.On("GetForecast", mock.Anything, mock.Anything, mock.Anything).Return(forecasts, nil)
 		m.On("GetUsageDetails", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(consumption.UsageDetailsListResultPage{}, nil)
 		client.BillingService = m
-		results, err := client.GetMetrics(startTime, endTime)
+		_, err := client.GetMetrics(opts)
 		assert.NoError(t, err)
-		assert.Equal(t, len(results.ActualCosts), 2)
-		assert.Equal(t, len(results.ForecastCosts), 2)
+		//assert.NotNil(t, usage.Forecasts.Rows)
+		//assert.Equal(t, len(*usage.Forecasts.Rows), 2)
+		// assert.Equal(t, len(results.ForecastCosts), 2)
 		m.AssertExpectations(t)
 	})
 }