override time.Now() for predictable report interval calculations

Author: purnesh42H

xds/internal/clients/lrsclient/load_store.go

@@ -25,6 +25,9 @@ import (
 	"time"
 )
 
+// clockNow is used to get the current time. It can be overridden in tests.
+var clockNow = time.Now
+
 // A LoadStore aggregates loads for multiple clusters and services that are
 // intended to be reported via LRS.
 //
@@ -84,7 +87,7 @@ func (ls *LoadStore) ReporterForCluster(clusterName, serviceName string) *PerClu
 	p := &PerClusterReporter{
 		cluster:          clusterName,
 		service:          serviceName,
-		lastLoadReportAt: time.Now(),
+		lastLoadReportAt: clockNow(),
 	}
 	c[serviceName] = p
 	return p
@@ -245,8 +248,8 @@ func (p *PerClusterReporter) stats() *loadData {
 	})
 
 	p.mu.Lock()
-	sd.reportInterval = time.Since(p.lastLoadReportAt)
-	p.lastLoadReportAt = time.Now()
+	sd.reportInterval = clockNow().Sub(p.lastLoadReportAt)
+	p.lastLoadReportAt = clockNow()
 	p.mu.Unlock()
 
 	if sd.totalDrops == 0 && len(sd.drops) == 0 && len(sd.localityStats) == 0 {

xds/internal/clients/lrsclient/load_store_test.go

@@ -22,6 +22,7 @@ import (
 	"sort"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
@@ -471,3 +472,49 @@ func TestStoreStatsEmptyDataNotReported(t *testing.T) {
 		t.Error(err)
 	}
 }
+
+// TestStoreReportInterval tests that the report interval is correctly
+// calculated between consecutive calls to stats().
+func TestStoreReportInterval(t *testing.T) {
+	originalClockNow := clockNow
+	t.Cleanup(func() { clockNow = originalClockNow })
+
+	// Initial time for reporter creation
+	currentTime := time.Now()
+	clockNow = func() time.Time {
+		return currentTime
+	}
+
+	store := newLoadStore()
+	reporter := store.ReporterForCluster("test-cluster", "test-service")
+	// To ensure stats() returns non-nil data, report a dummy drop.
+	reporter.CallDropped("dummy-category")
+
+	// First call to stats() calculates the report interval from reporter
+	// creation time.
+	currentTime = currentTime.Add(5 * time.Second)
+	stats1 := reporter.stats()
+
+	if stats1 == nil {
+		t.Fatalf("stats1 is nil after reporting a drop")
+	}
+	wantInterval := 5 * time.Second
+	if stats1.reportInterval != wantInterval {
+		t.Errorf("First call stats() = %v, want %v", stats1.reportInterval, wantInterval)
+	}
+
+	// Second call to stats() calculates the report interval from last stats()
+	// call time.
+	currentTime = currentTime.Add(10 * time.Second)
+	// Report another dummy drop to ensure stats2 is not nil.
+	reporter.CallDropped("dummy-category-2")
+	stats2 := reporter.stats()
+
+	if stats2 == nil {
+		t.Fatalf("stats2 is nil after reporting a drop")
+	}
+	wantInterval = 10 * time.Second
+	if stats2.reportInterval != wantInterval {
+		t.Errorf("Second call stats() = %v, want %v", stats2.reportInterval, wantInterval)
+	}
+}

xds/internal/clients/lrsclient/loadreport_test.go

@@ -159,7 +159,7 @@ func (s) TestReportLoad_ConnectionCreation(t *testing.T) {
 		t.Fatal("Timeout when waiting for LRS stream to be created")
 	}
 
-	// Call the load reporting API to report load to the second management
+	// Call the load reporting API to report load to the first management
 	// server, and ensure that a connection to the server is created.
 	serverIdentifier2 := clients.ServerIdentifier{ServerURI: mgmtServer2.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
 	loadStore2, err := client.ReportLoad(serverIdentifier2)
@@ -199,11 +199,10 @@ func (s) TestReportLoad_ConnectionCreation(t *testing.T) {
 	}
 
 	// Send a response from the server with a small deadline.
-	serverReportInterval := 50 * time.Millisecond
 	lrsServer.LRSResponseChan <- &fakeserver.Response{
 		Resp: &v3lrspb.LoadStatsResponse{
 			SendAllClusters:       true,
-			LoadReportingInterval: &durationpb.Duration{Nanos: int32(serverReportInterval.Nanoseconds())}, // 50ms
+			LoadReportingInterval: &durationpb.Duration{Nanos: 50000000}, // 50ms
 		},
 	}
 
@@ -217,19 +216,6 @@ func (s) TestReportLoad_ConnectionCreation(t *testing.T) {
 		t.Fatalf("Received load for %d clusters, want 1", l)
 	}
 
-	// Verify that LoadReportInterval for the first load report is positive but
-	// not excessively large.
-	//
-	// Max expected: serverReportInterval + tolerance (e.g., 500ms).
-	firstLoadReportInterval := gotLoad[0].GetLoadReportInterval().AsDuration()
-	if firstLoadReportInterval <= 0 {
-		t.Fatalf("First LoadReportInterval = %v, want > 0", firstLoadReportInterval)
-	}
-	tolerance := 500 * time.Millisecond
-	if firstLoadReportInterval > serverReportInterval+tolerance {
-		t.Errorf("First LoadReportInterval is unexpectedly large: %v", firstLoadReportInterval)
-	}
-
 	// This field is set by the client to indicate the actual time elapsed since
 	// the last report was sent. We cannot deterministically compare this, and
 	// we cannot use the cmpopts.IgnoreFields() option on proto structs, since
@@ -338,11 +324,10 @@ func (s) TestReportLoad_StreamCreation(t *testing.T) {
 	}
 
 	// Send a response from the server with a small deadline.
-	serverReportInterval := 50 * time.Millisecond
 	lrsServer.LRSResponseChan <- &fakeserver.Response{
 		Resp: &v3lrspb.LoadStatsResponse{
 			SendAllClusters:       true,
-			LoadReportingInterval: &durationpb.Duration{Nanos: int32(serverReportInterval.Nanoseconds())}, // 50ms
+			LoadReportingInterval: &durationpb.Duration{Nanos: 50000000}, // 50ms
 		},
 	}
 
@@ -356,19 +341,6 @@ func (s) TestReportLoad_StreamCreation(t *testing.T) {
 		t.Fatalf("Received load for %d clusters, want 1", l)
 	}
 
-	// Verify that LoadReportInterval for the first load report is positive but
-	// not excessively large.
-	//
-	// Max expected: serverReportInterval + tolerance (e.g., 500ms).
-	firstLoadReportInterval := gotLoad[0].GetLoadReportInterval().AsDuration()
-	if firstLoadReportInterval <= 0 {
-		t.Fatalf("First LoadReportInterval for cluster1 = %v, want > 0", firstLoadReportInterval)
-	}
-	tolerance := 500 * time.Millisecond
-	if firstLoadReportInterval > serverReportInterval+tolerance {
-		t.Errorf("First LoadReportInterval for cluster1 is unexpectedly large: %v", firstLoadReportInterval)
-	}
-
 	// This field is set by the client to indicate the actual time elapsed since
 	// the last report was sent. We cannot deterministically compare this, and
 	// we cannot use the cmpopts.IgnoreFields() option on proto structs, since
@@ -434,17 +406,6 @@ func (s) TestReportLoad_StreamCreation(t *testing.T) {
 		if l := len(gotLoad); l != 1 {
 			continue
 		}
-		// Verify that LoadReportInterval for the subsequent load reports is
-		// positive but not excessively large.
-		//
-		// Max expected: serverReportInterval + tolerance (e.g., 500ms).
-		loadReportInterval := gotLoad[0].GetLoadReportInterval().AsDuration()
-		if loadReportInterval <= 0 {
-			t.Fatalf("LoadReportInterval = %v, want > 0", firstLoadReportInterval)
-		}
-		if loadReportInterval > serverReportInterval+tolerance {
-			t.Errorf("LoadReportInterval is unexpectedly large: %v", loadReportInterval)
-		}
 		gotLoad[0].LoadReportInterval = nil
 		wantLoad := &v3endpointpb.ClusterStats{
 			ClusterName:          "cluster2",
@@ -551,11 +512,10 @@ func (s) TestReportLoad_StopWithContext(t *testing.T) {
 	}
 
 	// Send a response from the server with a small deadline.
-	serverReportInterval := 50 * time.Millisecond
 	lrsServer.LRSResponseChan <- &fakeserver.Response{
 		Resp: &v3lrspb.LoadStatsResponse{
 			SendAllClusters:       true,
-			LoadReportingInterval: &durationpb.Duration{Nanos: int32(serverReportInterval.Nanoseconds())}, // 50ms
+			LoadReportingInterval: &durationpb.Duration{Nanos: 50000000}, // 50ms
 		},
 	}
 
@@ -569,19 +529,6 @@ func (s) TestReportLoad_StopWithContext(t *testing.T) {
 		t.Fatalf("Received load for %d clusters, want 1", l)
 	}
 
-	// Verify that LoadReportInterval for the first load report is positive but
-	// not excessively large.
-	//
-	// Max expected: serverReportInterval + tolerance (e.g., 500ms).
-	firstLoadReportInterval := gotLoad[0].GetLoadReportInterval().AsDuration()
-	if firstLoadReportInterval <= 0 {
-		t.Fatalf("First LoadReportInterval = %v, want > 0", firstLoadReportInterval)
-	}
-	tolerance := 500 * time.Millisecond
-	if firstLoadReportInterval > serverReportInterval+tolerance {
-		t.Errorf("First LoadReportInterval is unexpectedly large: %v", firstLoadReportInterval)
-	}
-
 	// This field is set by the client to indicate the actual time elapsed since
 	// the last report was sent. We cannot deterministically compare this, and
 	// we cannot use the cmpopts.IgnoreFields() option on proto structs, since
@@ -644,17 +591,6 @@ func (s) TestReportLoad_StopWithContext(t *testing.T) {
 		if l := len(gotLoad); l != 1 {
 			continue
 		}
-		// Verify that LoadReportInterval for the subsequent load reports is
-		// positive but not excessively large.
-		//
-		// Max expected: serverReportInterval + tolerance (e.g., 500ms).
-		loadReportInterval := gotLoad[0].GetLoadReportInterval().AsDuration()
-		if loadReportInterval <= 0 {
-			t.Fatalf("LoadReportInterval = %v, want > 0", firstLoadReportInterval)
-		}
-		if loadReportInterval > serverReportInterval+tolerance {
-			t.Errorf("LoadReportInterval is unexpectedly large: %v", loadReportInterval)
-		}
 		gotLoad[0].LoadReportInterval = nil
 		wantLoad := &v3endpointpb.ClusterStats{
 			ClusterName:          "cluster2",

xds/internal/xdsclient/load/store.go

@@ -25,6 +25,9 @@ import (
 
 const negativeOneUInt64 = ^uint64(0)
 
+// clockNow is used to get the current time. It can be overridden in tests.
+var clockNow = time.Now
+
 // Store keeps the loads for multiple clusters and services to be reported via
 // LRS. It contains loads to reported to one LRS server. Create multiple stores
 // for multiple servers.
@@ -117,7 +120,7 @@ func (s *Store) PerCluster(clusterName, serviceName string) PerClusterReporter {
 	p := &perClusterStore{
 		cluster:          clusterName,
 		service:          serviceName,
-		lastLoadReportAt: time.Now(),
+		lastLoadReportAt: clockNow(),
 	}
 	c[serviceName] = p
 	return p
@@ -330,8 +333,8 @@ func (ls *perClusterStore) stats() *Data {
 	})
 
 	ls.mu.Lock()
-	sd.ReportInterval = time.Since(ls.lastLoadReportAt)
-	ls.lastLoadReportAt = time.Now()
+	sd.ReportInterval = clockNow().Sub(ls.lastLoadReportAt)
+	ls.lastLoadReportAt = clockNow()
 	ls.mu.Unlock()
 
 	if sd.TotalDrops == 0 && len(sd.Drops) == 0 && len(sd.LocalityStats) == 0 {

xds/internal/xdsclient/load/store_test.go

@@ -22,6 +22,7 @@ import (
 	"sort"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
@@ -466,3 +467,49 @@ func TestStoreStatsEmptyDataNotReported(t *testing.T) {
 		t.Errorf("store.stats() returned unexpected diff (-want +got):\n%s", diff)
 	}
 }
+
+// TestStoreReportInterval tests that the report interval is correctly
+// calculated between consecutive calls to Stats().
+func TestStoreReportInterval(t *testing.T) {
+	originalClockNow := clockNow
+	t.Cleanup(func() { clockNow = originalClockNow })
+
+	// Initial time for reporter creation
+	currentTime := time.Now()
+	clockNow = func() time.Time {
+		return currentTime
+	}
+
+	store := NewStore()
+	reporter := store.PerCluster("test-cluster", "test-service")
+	// To ensure Stats() returns non-nil data, report a dummy drop.
+	reporter.CallDropped("dummy-category")
+
+	// First call to Stats() calculates the report interval from reporter
+	// creation time.
+	currentTime = currentTime.Add(5 * time.Second)
+	stats1 := store.Stats(nil)
+
+	if stats1 == nil {
+		t.Fatalf("stats1 is nil after reporting a drop")
+	}
+	wantInterval := 5 * time.Second
+	if stats1[0].ReportInterval != wantInterval {
+		t.Errorf("First call stats() = %v, want %v", stats1[0].ReportInterval, wantInterval)
+	}
+
+	// Second call to Stats() calculates the report interval from last Stats()
+	// call time.
+	currentTime = currentTime.Add(10 * time.Second)
+	// Report another dummy drop to ensure stats2 is not nil.
+	reporter.CallDropped("dummy-category-2")
+	stats2 := store.Stats(nil)
+
+	if stats2 == nil {
+		t.Fatalf("stats2 is nil after reporting a drop")
+	}
+	wantInterval = 10 * time.Second
+	if stats2[0].ReportInterval != wantInterval {
+		t.Errorf("Second call stats() = %v, want %v", stats2[0].ReportInterval, wantInterval)
+	}
+}