feat: Skip writeback for chunks fetched by queriers older than a duration

clients/pkg/logentry/stages/json_test.go

@@ -56,6 +56,11 @@ func TestPipeline_JSON(t *testing.T) {
 		entry           string
 		expectedExtract map[string]interface{}
 	}{
+		"successfully run a pipeline with 1 logfmt stage with log not using json formatted string": {
+			testJSONYamlSingleStageWithoutSource,
+			"2012-11-01T22:08:41+00:00 [WARN] app:loki duration:125 - this log line is not in logfmt",
+			map[string]interface{}{},
+		},
 		"successfully run a pipeline with 1 json stage without source": {
 			testJSONYamlSingleStageWithoutSource,
 			testJSONLogLine,

clients/pkg/logentry/stages/logfmt_test.go

@@ -44,6 +44,11 @@ func TestPipeline_Logfmt(t *testing.T) {
 		entry           string
 		expectedExtract map[string]interface{}
 	}{
+		"successfully run a pipeline with 1 logfmt stage with log not using logfmt formatted string": {
+			testLogfmtYamlSingleStageWithoutSource,
+			"2012-11-01T22:08:41+00:00 [WARN] app:loki duration:125 - this log line is not in logfmt",
+			map[string]interface{}{},
+		},
 		"successfully run a pipeline with 1 logfmt stage without source": {
 			testLogfmtYamlSingleStageWithoutSource,
 			testLogfmtLogLine,

docs/sources/shared/configuration.md

@@ -1719,6 +1719,11 @@ The `chunk_store_config` block configures how chunks will be cached and how long
 # The CLI flags prefix for this block configuration is: store.index-cache-write
 [write_dedupe_cache_config: <cache_config>]
 
+# Chunks fetched from queriers before this duration will not be written to the
+# cache. A value of 0 will write all chunks to the cache
+# CLI flag: -store.skip-query-writeback-older-than
+[skip_query_writeback_cache_older_than: <duration> | default = 0s]
+
 # Chunks will be handed off to the L2 cache after this duration. 0 to disable L2
 # cache.
 # CLI flag: -store.chunks-cache-l2.handoff

pkg/distributor/distributor.go

@@ -848,7 +848,7 @@ func (d *Distributor) shardStream(stream logproto.Stream, pushSize int, tenantID
 
 	d.streamShardCount.Inc()
 	if shardStreamsCfg.LoggingEnabled {
-		level.Info(logger).Log("msg", "sharding request", "shard_count", shardCount)
+		level.Info(logger).Log("msg", "sharding request", "shard_count", shardCount, "push_size", pushSize)
 	}
 
 	return d.divideEntriesBetweenShards(tenantID, shardCount, shardStreamsCfg, stream)

pkg/storage/chunk/cache/cache_test.go

@@ -132,7 +132,7 @@ func testChunkFetcher(t *testing.T, c cache.Cache, chunks []chunk.Chunk) {
 		},
 	}
 
-	fetcher, err := fetcher.New(c, nil, false, s, nil, 0)
+	fetcher, err := fetcher.New(c, nil, false, s, nil, 0, 0)
 	require.NoError(t, err)
 	defer fetcher.Stop()
 

pkg/storage/chunk/fetcher/fetcher.go

@@ -49,7 +49,8 @@ type Fetcher struct {
 	cachel2    cache.Cache
 	cacheStubs bool
 
-	l2CacheHandoff time.Duration
+	l2CacheHandoff                   time.Duration
+	skipQueryWritebackCacheOlderThan time.Duration
 
 	wait           sync.WaitGroup
 	decodeRequests chan decodeRequest
@@ -69,15 +70,16 @@ type decodeResponse struct {
 }
 
 // New makes a new ChunkFetcher.
-func New(cache cache.Cache, cachel2 cache.Cache, cacheStubs bool, schema config.SchemaConfig, storage client.Client, l2CacheHandoff time.Duration) (*Fetcher, error) {
+func New(cache cache.Cache, cachel2 cache.Cache, cacheStubs bool, schema config.SchemaConfig, storage client.Client, l2CacheHandoff time.Duration, skipQueryWritebackOlderThan time.Duration) (*Fetcher, error) {
 	c := &Fetcher{
-		schema:         schema,
-		storage:        storage,
-		cache:          cache,
-		cachel2:        cachel2,
-		l2CacheHandoff: l2CacheHandoff,
-		cacheStubs:     cacheStubs,
-		decodeRequests: make(chan decodeRequest),
+		schema:                           schema,
+		storage:                          storage,
+		cache:                            cache,
+		cachel2:                          cachel2,
+		l2CacheHandoff:                   l2CacheHandoff,
+		skipQueryWritebackCacheOlderThan: skipQueryWritebackOlderThan,
+		cacheStubs:                       cacheStubs,
+		decodeRequests:                   make(chan decodeRequest),
 	}
 
 	c.wait.Add(chunkDecodeParallelism)
@@ -211,7 +213,6 @@ func (c *Fetcher) FetchChunks(ctx context.Context, chunks []chunk.Chunk) ([]chun
 	st.AddCacheBytesSent(stats.ChunkCache, bytes)
 
 	// Always cache any chunks we did get
-
 	if cacheErr := c.WriteBackCache(ctx, fromStorage); cacheErr != nil {
 		level.Warn(log).Log("msg", "could not store chunks in chunk cache", "err", cacheErr)
 	}
@@ -230,6 +231,10 @@ func (c *Fetcher) WriteBackCache(ctx context.Context, chunks []chunk.Chunk) erro
 	keysL2 := make([]string, 0, len(chunks))
 	bufsL2 := make([][]byte, 0, len(chunks))
 	for i := range chunks {
+		if c.skipQueryWritebackCacheOlderThan > 0 && chunks[i].From.Time().Before(time.Now().UTC().Add(-c.skipQueryWritebackCacheOlderThan)) {
+			continue
+		}
+
 		var encoded []byte
 		var err error
 		if !c.cacheStubs {