chore: Use github.com/coder/quartz instead of time (#13542)

go.mod

@@ -1,8 +1,8 @@
 module github.com/grafana/loki/v3
 
-go 1.21
+go 1.21.8
 
-toolchain go1.21.3
+toolchain go1.22.4
 
 require (
 	cloud.google.com/go/bigtable v1.18.1
@@ -120,6 +120,7 @@ require (
 	github.com/IBM/ibm-cos-sdk-go v1.10.0
 	github.com/axiomhq/hyperloglog v0.0.0-20240124082744-24bca3a5b39b
 	github.com/buger/jsonparser v1.1.1
+	github.com/coder/quartz v0.1.0
 	github.com/d4l3k/messagediff v1.2.1
 	github.com/dolthub/swiss v0.2.1
 	github.com/efficientgo/core v1.0.0-rc.2

go.sum

@@ -452,6 +452,8 @@ github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa/go.mod h1:x/1Gn8zydmfq
 github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMeY4+DwBQ=
 github.com/cockroachdb/datadriven v0.0.0-20190809214429-80d97fb3cbaa/go.mod h1:zn76sxSg3SzpJ0PPJaLDCu+Bu0Lg3sKTORVIj19EIF8=
 github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd/go.mod h1:sE/e/2PUdi/liOCUjSTXgM1o87ZssimdTWN964YiIeI=
+github.com/coder/quartz v0.1.0 h1:cLL+0g5l7xTf6ordRnUMMiZtRE8Sq5LxpghS63vEXrQ=
+github.com/coder/quartz v0.1.0/go.mod h1:vsiCc+AHViMKH2CQpGIpFgdHIEQsxwm8yCscqKmzbRA=
 github.com/containerd/containerd v1.4.1/go.mod h1:bC6axHOhabU15QhwfG7w5PipXdVtMXFTttgp+kVtyUA=
 github.com/containerd/fifo v1.0.0 h1:6PirWBr9/L7GDamKr+XM0IeUFXu5mf3M/BPpH9gaLBU=
 github.com/containerd/fifo v1.0.0/go.mod h1:ocF/ME1SX5b1AOlWi9r677YJmCPSwwWnQ9O123vzpE4=

pkg/storage/wal/manager.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/coder/quartz"
 	"github.com/prometheus/prometheus/model/labels"
 
 	"github.com/grafana/loki/v3/pkg/logproto"
@@ -115,9 +116,11 @@ type Manager struct {
 	// flushed, the segment is reset and moved to the back of the available
 	// list to accept writes again.
 	pending *list.List
+	closed  bool
+	mu      sync.Mutex
 
-	closed bool
-	mu     sync.Mutex
+	// Used in tests.
+	clock quartz.Clock
 }
 
 // item is similar to PendingItem, but it is an internal struct used in the
@@ -145,6 +148,7 @@ func NewManager(cfg Config, metrics *Metrics) (*Manager, error) {
 		metrics:   metrics.ManagerMetrics,
 		available: list.New(),
 		pending:   list.New(),
+		clock:     quartz.NewReal(),
 	}
 	m.metrics.NumPending.Set(0)
 	m.metrics.NumFlushing.Set(0)
@@ -177,12 +181,12 @@ func (m *Manager) Append(r AppendRequest) (*AppendResult, error) {
 		// This is the first append to the segment. This time will be used in
 		// know when the segment has exceeded its maximum age and should be
 		// moved to the pending list.
-		it.firstAppendedAt = time.Now()
+		it.firstAppendedAt = m.clock.Now()
 	}
 	it.w.Append(r.TenantID, r.LabelsStr, r.Labels, r.Entries)
 	// If the segment exceeded the maximum age or the maximum size, move it to
 	// the closed list to be flushed.
-	if time.Since(it.firstAppendedAt) >= m.cfg.MaxAge || it.w.InputSize() >= m.cfg.MaxSegmentSize {
+	if m.clock.Since(it.firstAppendedAt) >= m.cfg.MaxAge || it.w.InputSize() >= m.cfg.MaxSegmentSize {
 		m.pending.PushBack(it)
 		m.metrics.NumPending.Inc()
 		m.available.Remove(el)
@@ -218,7 +222,7 @@ func (m *Manager) NextPending() (*PendingItem, error) {
 			// should be moved to the pending list.
 			el := m.available.Front()
 			it := el.Value.(*item)
-			if !it.firstAppendedAt.IsZero() && time.Since(it.firstAppendedAt) >= m.cfg.MaxAge {
+			if !it.firstAppendedAt.IsZero() && m.clock.Since(it.firstAppendedAt) >= m.cfg.MaxAge {
 				m.pending.PushBack(it)
 				m.metrics.NumPending.Inc()
 				m.available.Remove(el)

pkg/storage/wal/manager_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/coder/quartz"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/prometheus/model/labels"
@@ -94,6 +95,10 @@ func TestManager_AppendMaxAge(t *testing.T) {
 	}, NewMetrics(nil))
 	require.NoError(t, err)
 
+	// Create a mock clock.
+	clock := quartz.NewMock(t)
+	m.clock = clock
+
 	// Append 1B of data.
 	lbs := labels.Labels{{Name: "a", Value: "b"}}
 	entries := []*logproto.Entry{{Timestamp: time.Now(), Line: "c"}}
@@ -112,7 +117,7 @@ func TestManager_AppendMaxAge(t *testing.T) {
 	require.Equal(t, 0, m.pending.Len())
 
 	// Wait 100ms and append some more data.
-	time.Sleep(100 * time.Millisecond)
+	clock.Advance(100 * time.Millisecond)
 	entries = []*logproto.Entry{{Timestamp: time.Now(), Line: "c"}}
 	res, err = m.Append(AppendRequest{
 		TenantID:  "1",
@@ -325,6 +330,10 @@ func TestManager_NexPendingMaxAge(t *testing.T) {
 	}, NewMetrics(nil))
 	require.NoError(t, err)
 
+	// Create a mock clock.
+	clock := quartz.NewMock(t)
+	m.clock = clock
+
 	// Append 1B of data.
 	lbs := labels.Labels{{Name: "a", Value: "b"}}
 	entries := []*logproto.Entry{{Timestamp: time.Now(), Line: "c"}}
@@ -347,7 +356,7 @@ func TestManager_NexPendingMaxAge(t *testing.T) {
 
 	// Wait 100ms. The segment that was just appended to should have reached
 	// the maximum age.
-	time.Sleep(100 * time.Millisecond)
+	clock.Advance(100 * time.Millisecond)
 	it, err = m.NextPending()
 	require.NoError(t, err)
 	require.NotNil(t, it)