Querier: dedupe chunks streamed from ingesters (grafana#463)

* Querier: dedupe chunks streamed from ingesters

Samples are replicated to all ingesters, so when we query there is a
good chance we receive exactly the same chunk more than once.
Deduping at the whole-chunk level is much more efficient than
doing it in the chunkMergeIterator later.

* Different implementation of accumulateChunks

Author: bboreham

CHANGELOG.md

@@ -113,7 +113,7 @@
   * `cortex_querier_blocks_found_total`
   * `cortex_querier_blocks_queried_total`
   * `cortex_querier_blocks_with_compactor_shard_but_incompatible_query_shard_total`
-* [ENHANCEMENT] Querier: reduce latency and peak memory consumption. #459
+* [ENHANCEMENT] Querier&Ruler: reduce cpu usage, latency and peak memory consumption. #459 #463
 * [BUGFIX] Frontend: Fixes @ modifier functions (start/end) when splitting queries by time. #206
 * [BUGFIX] Fixes a panic in the query-tee when comparing result. #207
 * [BUGFIX] Upgrade Prometheus. TSDB now waits for pending readers before truncating Head block, fixing the `chunk not found` error and preventing wrong query results. #16

pkg/distributor/query.go

@@ -239,7 +239,7 @@ func (d *Distributor) queryIngesterStream(ctx context.Context, replicationSet ri
 					key := ingester_client.LabelsToKeyString(mimirpb.FromLabelAdaptersToLabels(series.Labels))
 					existing := hashToChunkseries[key]
 					existing.Labels = series.Labels
-					existing.Chunks = append(existing.Chunks, series.Chunks...)
+					existing.Chunks = accumulateChunks(existing.Chunks, series.Chunks)
 					hashToChunkseries[key] = existing
 				}
 
@@ -383,3 +383,24 @@ func sameSamples(a, b []mimirpb.Sample) bool {
 	}
 	return true
 }
+
+// Build a slice of chunks, eliminating duplicates.
+// This is O(N^2) but most of the time N is small.
+func accumulateChunks(a, b []ingester_client.Chunk) []ingester_client.Chunk {
+	ret := a
+	for j := range b {
+		if !containsChunk(a, b[j]) {
+			ret = append(ret, b[j])
+		}
+	}
+	return ret
+}
+
+func containsChunk(a []ingester_client.Chunk, b ingester_client.Chunk) bool {
+	for i := range a {
+		if a[i].Equal(b) {
+			return true
+		}
+	}
+	return false
+}