Break background cache writes into batches of 100 (#2135)

* Break background cache writes into batches of 100

This improves parallelism and observability.
Fixes #2134

Signed-off-by: Bryan Boreham <[email protected]>

Author: bboreham

CHANGELOG.md

@@ -57,6 +57,7 @@
 * [ENHANCEMENT] Cassandra Storage: User no longer need `CREATE` privilege on `<all keyspaces>` if given keyspace exists. #2032
 * [ENHANCEMENT] Cassandra Storage: added `password_file` configuration options to enable reading Cassandra password from file. #2096
 * [ENHANCEMENT] Experimental Memberlist KV: expose `-memberlist.gossip-to-dead-nodes-time` and `-memberlist.dead-node-reclaim-time` options to control how memberlist library handles dead nodes and name reuse. #2131
+* [ENHANCEMENT] Background cache writes are batched to improve parallelism and observability. #2135
 * [BUGFIX] Alertmanager: fixed panic upon applying a new config, caused by duplicate metrics registration in the `NewPipelineBuilder` function. #211
 * [BUGFIX] Experimental TSDB: fixed `/all_user_stats` and `/api/prom/user_stats` endpoints when using the experimental TSDB blocks storage. #2042
 * [BUGFIX] Experimental TSDB: fixed ruler to correctly work with the experimental TSDB blocks storage. #2101

docs/configuration/config-file-reference.md

@@ -629,7 +629,7 @@ results_cache:
       # CLI flag: -frontend.memcache.write-back-goroutines
       [writeback_goroutines: <int> | default = 10]
 
-      # How many chunks to buffer for background write back.
+      # How many key batches to buffer for background write-back.
       # CLI flag: -frontend.memcache.write-back-buffer
       [writeback_buffer: <int> | default = 10000]
 
@@ -1501,8 +1501,8 @@ index_queries_cache_config:
     # CLI flag: -store.index-cache-read.memcache.write-back-goroutines
     [writeback_goroutines: <int> | default = 10]
 
-    # Cache config for index entry reading. How many chunks to buffer for
-    # background write back.
+    # Cache config for index entry reading. How many key batches to buffer for
+    # background write-back.
     # CLI flag: -store.index-cache-read.memcache.write-back-buffer
     [writeback_buffer: <int> | default = 10000]
 
@@ -1546,8 +1546,8 @@ chunk_cache_config:
     # CLI flag: -memcache.write-back-goroutines
     [writeback_goroutines: <int> | default = 10]
 
-    # Cache config for chunks. How many chunks to buffer for background write
-    # back.
+    # Cache config for chunks. How many key batches to buffer for background
+    # write-back.
     # CLI flag: -memcache.write-back-buffer
     [writeback_buffer: <int> | default = 10000]
 
@@ -1581,8 +1581,8 @@ write_dedupe_cache_config:
     # CLI flag: -store.index-cache-write.memcache.write-back-goroutines
     [writeback_goroutines: <int> | default = 10]
 
-    # Cache config for index entry writing. How many chunks to buffer for
-    # background write back.
+    # Cache config for index entry writing. How many key batches to buffer for
+    # background write-back.
     # CLI flag: -store.index-cache-write.memcache.write-back-buffer
     [writeback_buffer: <int> | default = 10000]
 

pkg/chunk/cache/background.go

@@ -33,7 +33,7 @@ type BackgroundConfig struct {
 // RegisterFlagsWithPrefix adds the flags required to config this to the given FlagSet
 func (cfg *BackgroundConfig) RegisterFlagsWithPrefix(prefix string, description string, f *flag.FlagSet) {
 	f.IntVar(&cfg.WriteBackGoroutines, prefix+"memcache.write-back-goroutines", 10, description+"How many goroutines to use to write back to memcache.")
-	f.IntVar(&cfg.WriteBackBuffer, prefix+"memcache.write-back-buffer", 10000, description+"How many chunks to buffer for background write back.")
+	f.IntVar(&cfg.WriteBackBuffer, prefix+"memcache.write-back-buffer", 10000, description+"How many key batches to buffer for background write-back.")
 }
 
 type backgroundCache struct {
@@ -80,21 +80,33 @@ func (c *backgroundCache) Stop() {
 	c.Cache.Stop()
 }
 
+const keysPerBatch = 100
+
 // Store writes keys for the cache in the background.
 func (c *backgroundCache) Store(ctx context.Context, keys []string, bufs [][]byte) {
-	bgWrite := backgroundWrite{
-		keys: keys,
-		bufs: bufs,
-	}
-	select {
-	case c.bgWrites <- bgWrite:
-		c.queueLength.Add(float64(len(keys)))
-	default:
-		c.droppedWriteBack.Add(float64(len(keys)))
-		sp := opentracing.SpanFromContext(ctx)
-		if sp != nil {
-			sp.LogFields(otlog.Int("dropped", len(keys)))
+	for len(keys) > 0 {
+		num := keysPerBatch
+		if num > len(keys) {
+			num = len(keys)
+		}
+
+		bgWrite := backgroundWrite{
+			keys: keys[:num],
+			bufs: bufs[:num],
+		}
+		select {
+		case c.bgWrites <- bgWrite:
+			c.queueLength.Add(float64(len(keys)))
+		default:
+			c.droppedWriteBack.Add(float64(len(keys)))
+			sp := opentracing.SpanFromContext(ctx)
+			if sp != nil {
+				sp.LogFields(otlog.Int("dropped", len(keys)))
+			}
+			return // queue is full; give up
 		}
+		keys = keys[num:]
+		bufs = bufs[num:]
 	}
 }
 

pkg/chunk/cache/cache_test.go

@@ -22,11 +22,11 @@ const userID = "1"
 func fillCache(t *testing.T, cache cache.Cache) ([]string, []chunk.Chunk) {
 	const chunkLen = 13 * 3600 // in seconds
 
-	// put 100 chunks from 0 to 99
+	// put a set of chunks, larger than background batch size, with varying timestamps and values
 	keys := []string{}
 	bufs := [][]byte{}
 	chunks := []chunk.Chunk{}
-	for i := 0; i < 100; i++ {
+	for i := 0; i < 111; i++ {
 		ts := model.TimeFromUnix(int64(i * chunkLen))
 		promChunk := prom_chunk.New()
 		nc, err := promChunk.Add(model.SamplePair{
@@ -119,7 +119,7 @@ func (a byExternalKey) Less(i, j int) bool { return a[i].ExternalKey() < a[j].Ex
 
 func testCacheMiss(t *testing.T, cache cache.Cache) {
 	for i := 0; i < 100; i++ {
-		key := strconv.Itoa(rand.Int())
+		key := strconv.Itoa(rand.Int()) // arbitrary key which should fail: no chunk key is a single integer
 		found, bufs, missing := cache.Fetch(context.Background(), []string{key})
 		require.Empty(t, found)
 		require.Empty(t, bufs)