fix(azuremonitorreceiver): Azure Monitor receiver should not produce gaps in data points for PT1M time grains

Author: an-mmx

.chloggen/fix_azure-monitor-receiver-time-grain.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: bug_fix
+
+# 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: "Azure Monitor receiver doesn't honor time grain"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37337]
+
+# (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/scraper.go

@@ -57,6 +57,7 @@ const (
 	attributeResourceType  = "type"
 	metadataPrefix         = "metadata_"
 	tagPrefix              = "tags_"
+	truncateTimeGrain      = time.Minute
 )
 
 type azureResource struct {
@@ -78,6 +79,16 @@ type azureResourceMetrics struct {
 
 type void struct{}
 
+type timeNowIface interface {
+	Now() time.Time
+}
+
+type timeWrapper struct{}
+
+func (*timeWrapper) Now() time.Time {
+	return time.Now()
+}
+
 func newScraper(conf *Config, settings receiver.Settings) *azureScraper {
 	return &azureScraper{
 		cfg:                             conf,
@@ -91,6 +102,7 @@ func newScraper(conf *Config, settings receiver.Settings) *azureScraper {
 		armMonitorDefinitionsClientFunc: armmonitor.NewMetricDefinitionsClient,
 		armMonitorMetricsClientFunc:     armmonitor.NewMetricsClient,
 		mutex:                           &sync.Mutex{},
+		time:                            &timeWrapper{},
 	}
 }
 
@@ -115,6 +127,7 @@ type azureScraper struct {
 	armMonitorDefinitionsClientFunc func(string, azcore.TokenCredential, *arm.ClientOptions) (*armmonitor.MetricDefinitionsClient, error)
 	armMonitorMetricsClientFunc     func(string, azcore.TokenCredential, *arm.ClientOptions) (*armmonitor.MetricsClient, error)
 	mutex                           *sync.Mutex
+	time                            timeNowIface
 }
 
 type armClient interface {
@@ -369,12 +382,13 @@ func (s *azureScraper) storeMetricsDefinition(resourceID, name string, composite
 
 func (s *azureScraper) getResourceMetricsValues(ctx context.Context, resourceID string) {
 	res := *s.resources[resourceID]
+	updatedAt := s.time.Now().Truncate(truncateTimeGrain)
 
 	for compositeKey, metricsByGrain := range res.metricsByCompositeKey {
-		if time.Since(metricsByGrain.metricsValuesUpdated).Seconds() < float64(timeGrains[compositeKey.timeGrain]) {
+		if updatedAt.Sub(metricsByGrain.metricsValuesUpdated).Seconds() < float64(timeGrains[compositeKey.timeGrain]) {
 			continue
 		}
-		metricsByGrain.metricsValuesUpdated = time.Now()
+		metricsByGrain.metricsValuesUpdated = updatedAt
 
 		start := 0
 

receiver/azuremonitorreceiver/scraper_test.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/arm"
@@ -65,6 +66,7 @@ func armMonitorMetricsClientFuncMock(string, azcore.TokenCredential, *arm.Client
 
 func TestAzureScraperStart(t *testing.T) {
 	cfg := createDefaultConfig().(*Config)
+	timeMock := getTimeMock()
 
 	tests := []struct {
 		name     string
@@ -76,6 +78,7 @@ func TestAzureScraperStart(t *testing.T) {
 			testFunc: func(t *testing.T) {
 				s := &azureScraper{
 					cfg:                             cfg,
+					time:                            timeMock,
 					azIDCredentialsFunc:             azIDCredentialsFuncMock,
 					azIDWorkloadFunc:                azIDWorkloadFuncMock,
 					armClientFunc:                   armClientFuncMock,
@@ -104,6 +107,7 @@ func TestAzureScraperStart(t *testing.T) {
 				}
 				s := &azureScraper{
 					cfg:                             customCfg,
+					time:                            timeMock,
 					azIDCredentialsFunc:             azIDCredentialsFuncMock,
 					azIDWorkloadFunc:                azIDWorkloadFuncMock,
 					armClientFunc:                   armClientFuncMock,
@@ -132,6 +136,7 @@ func TestAzureScraperStart(t *testing.T) {
 				}
 				s := &azureScraper{
 					cfg:                             customCfg,
+					time:                            timeMock,
 					azIDCredentialsFunc:             azIDCredentialsFuncMock,
 					azIDWorkloadFunc:                azIDWorkloadFuncMock,
 					armClientFunc:                   armClientFuncMock,
@@ -160,6 +165,7 @@ func TestAzureScraperStart(t *testing.T) {
 				}
 				s := &azureScraper{
 					cfg:                             customCfg,
+					time:                            timeMock,
 					azIDCredentialsFunc:             azIDCredentialsFuncMock,
 					azManagedIdentityFunc:           azManagedIdentityFuncMock,
 					armClientFunc:                   armClientFuncMock,
@@ -188,6 +194,7 @@ func TestAzureScraperStart(t *testing.T) {
 				}
 				s := &azureScraper{
 					cfg:                             customCfg,
+					time:                            timeMock,
 					azIDCredentialsFunc:             azIDCredentialsFuncMock,
 					azDefaultCredentialsFunc:        azDefaultCredentialsFuncMock,
 					armClientFunc:                   armClientFuncMock,
@@ -319,6 +326,7 @@ func TestAzureScraperScrape(t *testing.T) {
 				clientMetricsValues:      metricsValuesClientMock,
 				mb:                       metadata.NewMetricsBuilder(metadata.DefaultMetricsBuilderConfig(), settings),
 				mutex:                    &sync.Mutex{},
+				time:                     getTimeMock(),
 			}
 			s.resources = map[string]*azureResource{}
 
@@ -342,6 +350,95 @@ func TestAzureScraperScrape(t *testing.T) {
 	}
 }
 
+func TestAzureScraperScrapeHonorTimeGrain(t *testing.T) {
+	getTestScraper := func() *azureScraper {
+		armClientMock := &armClientMock{
+			current: 0,
+			pages:   getResourcesMockData(false),
+		}
+		counters, pages := getMetricsDefinitionsMockData()
+		metricsDefinitionsClientMock := &metricsDefinitionsClientMock{
+			current: counters,
+			pages:   pages,
+		}
+		metricsValuesClientMock := &metricsValuesClientMock{
+			lists: getMetricsValuesMockData(),
+		}
+
+		return &azureScraper{
+			cfg:                      createDefaultConfig().(*Config),
+			clientResources:          armClientMock,
+			clientMetricsDefinitions: metricsDefinitionsClientMock,
+			clientMetricsValues:      metricsValuesClientMock,
+			mb: metadata.NewMetricsBuilder(
+				metadata.DefaultMetricsBuilderConfig(),
+				receivertest.NewNopSettings(),
+			),
+			mutex:     &sync.Mutex{},
+			resources: map[string]*azureResource{},
+			time:      getTimeMock(),
+		}
+	}
+
+	ctx := context.Background()
+
+	t.Run("do_not_fetch_in_same_interval", func(t *testing.T) {
+		s := getTestScraper()
+
+		metrics, err := s.scrape(ctx)
+
+		require.NoError(t, err, "should not fail")
+		require.Positive(t, metrics.MetricCount(), "should return metrics on first call")
+
+		metrics, err = s.scrape(ctx)
+
+		require.NoError(t, err, "should not fail")
+		require.Equal(t, 0, metrics.MetricCount(), "should not return metrics on second call")
+	})
+
+	t.Run("fetch_each_new_interval", func(t *testing.T) {
+		timeJitter := time.Second
+		timeInterval := 50 * time.Second
+		timeIntervals := []time.Time{
+			time.Now().Add(time.Minute),
+			time.Now().Add(time.Minute + 1*timeInterval - timeJitter),
+			time.Now().Add(time.Minute + 2*timeInterval + timeJitter),
+			time.Now().Add(time.Minute + 3*timeInterval - timeJitter),
+			time.Now().Add(time.Minute + 4*timeInterval + timeJitter),
+		}
+		s := getTestScraper()
+		mockedTime := s.time.(*timeMock)
+
+		for _, timeNowNew := range timeIntervals {
+			// implementation uses time.Sub to check if timeWrapper.Now satisfy time grain
+			// we can travel in time by adjusting result of timeWrapper.Now
+			prevTime := mockedTime.time
+			mockedTime.time = timeNowNew
+
+			metrics, err := s.scrape(ctx)
+
+			require.NoError(t, err, "should not fail")
+			if prevTime.Minute() == timeNowNew.Minute() {
+				require.Equal(t, 0, metrics.MetricCount(), "should not fetch metrics in the same minute")
+			} else {
+				require.Positive(t, metrics.MetricCount(), "should fetch metrics in a new minute")
+			}
+		}
+	})
+}
+
+type timeMock struct {
+	time time.Time
+}
+
+func (t *timeMock) Now() time.Time {
+	return t.time
+}
+
+func getTimeMock() timeNowIface {
+	return &timeMock{time: time.Now()}
+}
+
 func getResourcesMockData(tags bool) []armresources.ClientListResponse {
 	id1, id2, id3, location1, name1, type1 := "/resourceGroups/group1/resourceId1",
 		"/resourceGroups/group1/resourceId2", "/resourceGroups/group1/resourceId3", "location1", "name1", "type1"