test: de-flake internal/metrics timing-sensitive tests

internal/metrics/relay_metrics_collector.go

@@ -63,19 +63,27 @@ type RelayMetricsCollector struct {
 	pollingCounts      map[connectionsKeyType]pollingCounts
 	mu                 sync.Mutex
 	closer             chan struct{}
+	now                func() time.Time
 }
 
 func newRelayMetricsCollector(relayID, envName string, publisher events.EventPublisher, flushInterval time.Duration, logger *slog.Logger) *RelayMetricsCollector {
+	return newRelayMetricsCollectorWithTimeSource(relayID, envName, publisher, flushInterval, logger, time.Now)
+}
+
+// newRelayMetricsCollectorWithTimeSource allows tests to control the timestamps
+// used for interval boundaries.
+func newRelayMetricsCollectorWithTimeSource(relayID, envName string, publisher events.EventPublisher, flushInterval time.Duration, logger *slog.Logger, now func() time.Time) *RelayMetricsCollector {
 	c := &RelayMetricsCollector{
 		relayID:            relayID,
 		envName:            envName,
 		publisher:          publisher,
 		logger:             logger,
 		closer:             make(chan struct{}),
-		intervalStartTime:  time.Now(),
+		intervalStartTime:  now(),
 		pollingDataIsDirty: false,
 		currentConnections: make(map[connectionsKeyType]int64),
 		pollingCounts:      make(map[connectionsKeyType]pollingCounts),
+		now:                now,
 	}
 
 	flushTicker := time.NewTicker(flushInterval)
@@ -135,7 +143,7 @@ func (c *RelayMetricsCollector) hasMetricDataToReport() bool {
 func (c *RelayMetricsCollector) flush() {
 	c.mu.Lock()
 	startTime := c.intervalStartTime
-	stopTime := time.Now()
+	stopTime := c.now()
 	c.intervalStartTime = stopTime
 
 	if !c.hasMetricDataToReport() {

internal/metrics/relay_metrics_collector_test.go

@@ -122,15 +122,24 @@ func TestRelayMetricsCollector(t *testing.T) {
 
 	t.Run("the event start time still shifts when events are not sent", func(t *testing.T) {
 		publisher := newTestEventsPublisher()
-		withCollector(publisher, func(c *RelayMetricsCollector, relayID string) {
-			time.Sleep(time.Millisecond * 10)
-			startTime := ldtime.UnixMillisNow()
-			time.Sleep(time.Millisecond * 1)
-			c.RecordConnectionChange(platformValue, userAgentValue, "", 1)
+		clk := newFakeClock()
+		c := newRelayMetricsCollectorWithTimeSource(uuid.New(), "envName", publisher, time.Hour, slog.Default(), clk.now)
+		defer c.close()
 
-			c.flush()
-			metricsEvent := publisher.expectMetricsEvent(t, time.Second)
-			assert.True(t, metricsEvent.StartDate >= startTime)
-		})
+		// A flush with no data publishes nothing, but should still move the
+		// interval start forward.
+		clk.advance(time.Millisecond * 10)
+		c.flush()
+		publisher.expectNoMetricsEvent(t, time.Millisecond*50)
+		shiftedStart := clk.now()
+
+		clk.advance(time.Millisecond)
+		c.RecordConnectionChange(platformValue, userAgentValue, "", 1)
+		clk.advance(time.Millisecond)
+		c.flush()
+
+		metricsEvent := publisher.expectMetricsEvent(t, time.Second)
+		assert.Equal(t, ldtime.UnixMillisFromTime(shiftedStart), metricsEvent.StartDate)
+		assert.Equal(t, ldtime.UnixMillisFromTime(clk.now()), metricsEvent.EndDate)
 	})
 }

internal/metrics/test_utils_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"log/slog"
+	"sync"
 	"testing"
 	"time"
 
@@ -75,6 +76,30 @@ func testWithOTel(t *testing.T, action func(testWithOTelParams)) {
 	})
 }
 
+// fakeClock is a controllable time source for tests that measure durations.
+// Advancing it is the test's replacement for sleeping real time, which is
+// unreliable under CI scheduling jitter.
+type fakeClock struct {
+	mu sync.Mutex
+	t  time.Time
+}
+
+func newFakeClock() *fakeClock {
+	return &fakeClock{t: time.Date(2026, time.January, 1, 12, 0, 0, 0, time.UTC)}
+}
+
+func (c *fakeClock) now() time.Time {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.t
+}
+
+func (c *fakeClock) advance(d time.Duration) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.t = c.t.Add(d)
+}
+
 type testEventsPublisher struct {
 	events chan json.RawMessage
 }

internal/metrics/usage.go

@@ -48,11 +48,17 @@ type usageActivityMessage struct {
 	platformCategory string
 	instanceID       string
 	tagsHeader       string
+
+	// The time at which the activity was recorded. Stamped when the message is
+	// handed to an environmentMetricUsage, before it crosses into the
+	// processing goroutine, so that tests can substitute a controllable clock
+	// and observe deterministic durations.
+	timestamp time.Time
 }
 
 type (
-	usageActivityFlush    struct{}
-	usageActivityShutdown struct{}
+	usageActivityFlush    struct{ timestamp time.Time }
+	usageActivityShutdown struct{ timestamp time.Time }
 )
 
 // metricUsage is used to track usage information for a single
@@ -137,18 +143,24 @@ type environmentMetricUsage struct {
 	publisher     events.EventPublisher
 	flushInterval time.Duration
 	usageCh       chan interface{}
+	now           func() time.Time
 
 	// All of this data is expected to only be accessed from within a single go
 	// routine
 	usages map[usageKeyType]*metricUsage
 }
 
 func NewEnvironmentMetricUsage(relayID string, publisher events.EventPublisher, flushInterval time.Duration) *environmentMetricUsage {
+	return newEnvironmentMetricUsage(relayID, publisher, flushInterval, time.Now)
+}
+
+func newEnvironmentMetricUsage(relayID string, publisher events.EventPublisher, flushInterval time.Duration, now func() time.Time) *environmentMetricUsage {
 	e := &environmentMetricUsage{
 		relayID:       relayID,
 		publisher:     publisher,
 		flushInterval: flushInterval,
 		usageCh:       make(chan interface{}),
+		now:           now,
 
 		usages: make(map[usageKeyType]*metricUsage),
 	}
@@ -159,6 +171,7 @@ func NewEnvironmentMetricUsage(relayID string, publisher events.EventPublisher,
 }
 
 func (e *environmentMetricUsage) usageActivityMessage(usage *usageActivityMessage) {
+	usage.timestamp = e.now()
 	e.usageCh <- usage
 }
 
@@ -169,20 +182,20 @@ func (e *environmentMetricUsage) run() {
 	for {
 		select {
 		case <-ticker.C:
-			e.flushInternal()
+			e.flushInternal(e.now())
 		case usage, ok := <-e.usageCh:
 			if !ok {
 				return
 			}
-			now := time.Now()
 
 			switch u := usage.(type) {
 			case *usageActivityShutdown:
-				e.flushInternal()
+				e.flushInternal(u.timestamp)
 				return
 			case *usageActivityFlush:
-				e.flushInternal()
+				e.flushInternal(u.timestamp)
 			case *usageActivityMessage:
+				now := u.timestamp
 				key := usageKeyType{userAgent: u.userAgent, platformCategory: u.platformCategory, instanceID: u.instanceID, tagsHeader: u.tagsHeader}
 				if e.publisher == nil {
 					continue
@@ -238,15 +251,15 @@ func (e *environmentMetricUsage) run() {
 }
 
 func (e *environmentMetricUsage) flush() { //nolint:unused // used only in tests
-	e.usageCh <- &usageActivityFlush{}
+	e.usageCh <- &usageActivityFlush{timestamp: e.now()}
 }
 
 func (e *environmentMetricUsage) close() {
-	e.usageCh <- &usageActivityShutdown{}
+	e.usageCh <- &usageActivityShutdown{timestamp: e.now()}
 	close(e.usageCh)
 }
 
-func (e *environmentMetricUsage) flushInternal() {
+func (e *environmentMetricUsage) flushInternal(now time.Time) {
 	if e.publisher == nil {
 		return
 	}
@@ -255,8 +268,6 @@ func (e *environmentMetricUsage) flushInternal() {
 		return
 	}
 
-	now := time.Now()
-
 	for key, usage := range e.usages {
 		// Refer back to the metricUsage comment for an explanation on this
 		// calculation.