feat(querier/query-frontend): track and log congestion control latency (

#12058)

Signed-off-by: Danny Kopping <[email protected]>

pkg/logql/metrics.go

@@ -189,6 +189,8 @@ func RecordRangeAndInstantQueryMetrics(
 		"ingester_chunk_decompressed_bytes", humanizeBytes(uint64(stats.Ingester.Store.Chunk.GetDecompressedBytes())),
 		// Total lines post filtering.
 		"ingester_post_filter_lines", stats.Ingester.Store.Chunk.GetPostFilterLines(),
+		// Time spent being blocked on congestion control.
+		"congestion_control_latency", stats.CongestionControlLatency(),
 	}...)
 
 	logValues = append(logValues, tagsToKeyValues(queryTags)...)

pkg/logqlmodel/stats/context.go

@@ -96,6 +96,11 @@ func (c *Context) Ingester() Ingester {
 	}
 }
 
+// Querier returns the store statistics accumulated so far.
+func (c *Context) Querier() Querier {
+	return c.querier
+}
+
 // Caches returns the cache statistics accumulated so far.
 func (c *Context) Caches() Caches {
 	return Caches{
@@ -279,6 +284,10 @@ func (r Result) ChunkRefsFetchTime() time.Duration {
 	return time.Duration(r.Querier.Store.ChunkRefsFetchTime + r.Ingester.Store.ChunkRefsFetchTime)
 }
 
+func (r Result) CongestionControlLatency() time.Duration {
+	return time.Duration(r.Querier.Store.CongestionControlLatency)
+}
+
 func (r Result) TotalDuplicates() int64 {
 	return r.Querier.Store.Chunk.TotalDuplicates + r.Ingester.Store.Chunk.TotalDuplicates
 }
@@ -360,6 +369,10 @@ func (c *Context) AddChunkRefsFetchTime(i time.Duration) {
 	atomic.AddInt64(&c.store.ChunkRefsFetchTime, int64(i))
 }
 
+func (c *Context) AddCongestionControlLatency(i time.Duration) {
+	atomic.AddInt64(&c.querier.Store.CongestionControlLatency, int64(i))
+}
+
 func (c *Context) AddChunksDownloaded(i int64) {
 	atomic.AddInt64(&c.store.TotalChunksDownloaded, i)
 }

pkg/logqlmodel/stats/stats.proto

@@ -136,6 +136,9 @@ message Store {
 
   // Time spent fetching chunk refs from index.
   int64 chunkRefsFetchTime = 5 [(gogoproto.jsontag) = "chunkRefsFetchTime"];
+
+  // Time spent being blocked on congestion control.
+  int64 congestionControlLatency = 6 [(gogoproto.jsontag) = "congestionControlLatency"];
 }
 
 message Chunk {

pkg/querier/queryrange/codec_test.go

@@ -1561,6 +1561,7 @@ var (
 					"totalDuplicates": 8
 				},
 				"chunksDownloadTime": 0,
+				"congestionControlLatency": 0,
 				"totalChunksRef": 0,
 				"totalChunksDownloaded": 0,
 				"chunkRefsFetchTime": 0,
@@ -1585,6 +1586,7 @@ var (
 					"totalDuplicates": 19
 				},
 				"chunksDownloadTime": 16,
+				"congestionControlLatency": 0,
 				"totalChunksRef": 17,
 				"totalChunksDownloaded": 18,
 				"chunkRefsFetchTime": 19,
@@ -2018,6 +2020,7 @@ var (
 					TotalDuplicates:   19,
 				},
 				ChunksDownloadTime:        16,
+				CongestionControlLatency:  0,
 				TotalChunksRef:            17,
 				TotalChunksDownloaded:     18,
 				ChunkRefsFetchTime:        19,

pkg/querier/queryrange/prometheus_test.go

@@ -16,6 +16,7 @@ var emptyStats = `"stats": {
 	"ingester" : {
 		"store": {
 			"chunksDownloadTime": 0,
+			"congestionControlLatency": 0,
 			"totalChunksRef": 0,
 			"totalChunksDownloaded": 0,
 			"chunkRefsFetchTime": 0,
@@ -40,6 +41,7 @@ var emptyStats = `"stats": {
 	"querier": {
 		"store": {
 			"chunksDownloadTime": 0,
+			"congestionControlLatency": 0,
 			"totalChunksRef": 0,
 			"totalChunksDownloaded": 0,
 			"chunkRefsFetchTime": 0,

pkg/storage/chunk/client/congestion/controller.go

@@ -10,6 +10,7 @@ import (
 	"github.com/go-kit/log"
 	"golang.org/x/time/rate"
 
+	"github.com/grafana/loki/pkg/logqlmodel/stats"
 	"github.com/grafana/loki/pkg/storage/chunk/client"
 )
 
@@ -92,6 +93,9 @@ func (a *AIMDController) GetObject(ctx context.Context, objectKey string) (io.Re
 
 	// TODO(dannyk): use hedging client to handle requests, do NOT hedge retries
 
+	start := time.Now()
+	statsCtx := stats.FromContext(ctx)
+
 	rc, sz, err := a.retrier.Do(
 		func(attempt int) (io.ReadCloser, int64, error) {
 			a.metrics.requests.Add(1)
@@ -111,6 +115,8 @@ func (a *AIMDController) GetObject(ctx context.Context, objectKey string) (io.Re
 				a.metrics.backoffSec.Add(delay.Seconds())
 			}
 
+			statsCtx.AddCongestionControlLatency(time.Since(start))
+
 			// It is vitally important that retries are DISABLED in the inner implementation.
 			// Some object storage clients implement retries internally, and this will interfere here.
 			return a.inner.GetObject(ctx, objectKey)

pkg/storage/chunk/client/congestion/controller_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
 
+	"github.com/grafana/loki/pkg/logqlmodel/stats"
 	"github.com/grafana/loki/pkg/storage/chunk/client"
 )
 
@@ -165,8 +166,10 @@ func TestAIMDReducedThroughput(t *testing.T) {
 	cli := newMockObjectClient(triggeredFailer{trigger: &trigger})
 	ctrl.Wrap(cli)
 
+	statsCtx, ctx := stats.NewContext(context.Background())
+
 	// run for 1 second, measure the per-second rate of requests & successful responses
-	count, success := runAndMeasureRate(ctrl, time.Second)
+	count, success := runAndMeasureRate(ctx, ctrl, time.Second)
 	require.Greater(t, count, 1.0)
 	require.Greater(t, success, 1.0)
 	// no time spent backing off because the per-second limit will not be hit
@@ -195,7 +198,7 @@ func TestAIMDReducedThroughput(t *testing.T) {
 	}(&trigger)
 
 	// now, run the requests again but there will now be a failure rate & some throttling involved
-	count, success = runAndMeasureRate(ctrl, time.Second)
+	count, success = runAndMeasureRate(ctx, ctrl, time.Second)
 	done <- true
 
 	wg.Wait()
@@ -206,9 +209,12 @@ func TestAIMDReducedThroughput(t *testing.T) {
 
 	// should have fewer successful requests than total since we are failing some
 	require.Less(t, success, count)
+
+	// should have registered some congestion latency in stats
+	require.NotZero(t, statsCtx.Querier().Store.CongestionControlLatency)
 }
 
-func runAndMeasureRate(ctrl Controller, duration time.Duration) (float64, float64) {
+func runAndMeasureRate(ctx context.Context, ctrl Controller, duration time.Duration) (float64, float64) {
 	var count, success float64
 
 	tick := time.NewTimer(duration)
@@ -218,8 +224,6 @@ func runAndMeasureRate(ctrl Controller, duration time.Duration) (float64, float6
 		case <-tick.C:
 			goto result
 		default:
-			ctx := context.Background()
-
 			count++
 			_, _, err := ctrl.GetObject(ctx, "foo")
 			if err == nil {

pkg/util/marshal/legacy/marshal_test.go

@@ -59,6 +59,7 @@ var queryTests = []struct {
 				"ingester" : {
 					"store": {
 						"chunksDownloadTime": 0,
+						"congestionControlLatency": 0,
 						"totalChunksRef": 0,
 						"totalChunksDownloaded": 0,
 						"chunkRefsFetchTime": 0,
@@ -83,6 +84,7 @@ var queryTests = []struct {
 				"querier": {
 					"store": {
 						"chunksDownloadTime": 0,
+						"congestionControlLatency": 0,
 						"totalChunksRef": 0,
 						"totalChunksDownloaded": 0,
 						"chunkRefsFetchTime": 0,

pkg/util/marshal/marshal_test.go

@@ -27,6 +27,7 @@ const emptyStats = `{
 	"ingester" : {
 		"store": {
 			"chunksDownloadTime": 0,
+			"congestionControlLatency": 0,
 			"totalChunksRef": 0,
 			"totalChunksDownloaded": 0,
 			"chunkRefsFetchTime": 0,
@@ -51,6 +52,7 @@ const emptyStats = `{
 	"querier": {
 		"store": {
 			"chunksDownloadTime": 0,
+			"congestionControlLatency": 0,
 			"totalChunksRef": 0,
 			"totalChunksDownloaded": 0,
 			"chunkRefsFetchTime": 0,