sidecar: Added support for streaming, chunked remote read. (#1268)

Fixes: #488

Signed-off-by: Bartek Plotka <bwplotka@gmail.com>

CHANGELOG.md

@@ -14,6 +14,8 @@ We use *breaking* word for marking changes that are not backward compatible (rel
 ### Added
 
 - [#1378](https://github.com/thanos-io/thanos/pull/1378) Thanos Receive now exposes `thanos_receive_config_hash`, `thanos_receive_config_last_reload_successful` and `thanos_receive_config_last_reload_success_timestamp_seconds` metrics to track latest configuration change
+- [#1268](https://github.com/thanos-io/thanos/pull/1268) Added support for newest Prometheus streaming remote read added [here](https://github.com/prometheus/prometheus/pull/5703). This massively improves memory required by single 
+  request for both Prometheus and sidecar. Single requests now should take constant amount of memory on sidecar, so resource consumption prediction is now straightforward.   
 - [#1358](https://github.com/thanos-io/thanos/pull/1358) Added `part_size` configuration option for HTTP multipart requests minimum part size for S3 storage type
 - [#1363](https://github.com/thanos-io/thanos/pull/1363) Thanos Receive now exposes `thanos_receive_hashring_nodes` and `thanos_receive_hashring_tenants` metrics to monitor status of hash-rings
 

pkg/query/iter.go

@@ -15,17 +15,52 @@ import (
 // promSeriesSet implements the SeriesSet interface of the Prometheus storage
 // package on top of our storepb SeriesSet.
 type promSeriesSet struct {
-	set        storepb.SeriesSet
+	set       storepb.SeriesSet
+	initiated bool
+	done      bool
+
 	mint, maxt int64
 	aggr       resAggr
+
+	currLset   []storepb.Label
+	currChunks []storepb.AggrChunk
 }
 
-func (s promSeriesSet) Next() bool { return s.set.Next() }
-func (s promSeriesSet) Err() error { return s.set.Err() }
+func (s *promSeriesSet) Next() bool {
+	if !s.initiated {
+		s.initiated = true
+		s.done = s.set.Next()
+	}
 
-func (s promSeriesSet) At() storage.Series {
-	lset, chunks := s.set.At()
-	return newChunkSeries(lset, chunks, s.mint, s.maxt, s.aggr)
+	if !s.done {
+		return false
+	}
+
+	// storage.Series are more strict then SeriesSet: It requires storage.Series to iterate over full series.
+	s.currLset, s.currChunks = s.set.At()
+	for {
+		s.done = s.set.Next()
+		if !s.done {
+			break
+		}
+		nextLset, nextChunks := s.set.At()
+		if storepb.CompareLabels(s.currLset, nextLset) != 0 {
+			break
+		}
+		s.currChunks = append(s.currChunks, nextChunks...)
+	}
+	return true
+}
+
+func (s *promSeriesSet) At() storage.Series {
+	if !s.initiated || s.set.Err() != nil {
+		return nil
+	}
+	return newChunkSeries(s.currLset, s.currChunks, s.mint, s.maxt, s.aggr)
+}
+
+func (s *promSeriesSet) Err() error {
+	return s.set.Err()
 }
 
 func translateMatcher(m *labels.Matcher) (storepb.LabelMatcher, error) {

pkg/query/querier.go

@@ -183,7 +183,7 @@ func (q *querier) Select(params *storage.SelectParams, ms ...*labels.Matcher) (s
 
 	if !q.isDedupEnabled() {
 		// Return data without any deduplication.
-		return promSeriesSet{
+		return &promSeriesSet{
 			mint: q.mint,
 			maxt: q.maxt,
 			set:  newStoreSeriesSet(resp.seriesSet),
@@ -195,7 +195,7 @@ func (q *querier) Select(params *storage.SelectParams, ms ...*labels.Matcher) (s
 	// to make true streaming possible.
 	sortDedupLabels(resp.seriesSet, q.replicaLabel)
 
-	set := promSeriesSet{
+	set := &promSeriesSet{
 		mint: q.mint,
 		maxt: q.maxt,
 		set:  newStoreSeriesSet(resp.seriesSet),

pkg/query/querier_test.go

@@ -163,8 +163,12 @@ func TestQuerier_Series(t *testing.T) {
 
 	testProxy := &storeServer{
 		resps: []*storepb.SeriesResponse{
+			// Expected sorted  series per seriesSet input. However we Series API allows for single series being chunks across multiple frames.
+			// This should be handled here.
 			storeSeriesResponse(t, labels.FromStrings("a", "a"), []sample{{0, 0}, {2, 1}, {3, 2}}),
 			storepb.NewWarnSeriesResponse(errors.New("partial error")),
+			storeSeriesResponse(t, labels.FromStrings("a", "a"), []sample{{5, 5}, {6, 6}, {7, 7}}),
+			storeSeriesResponse(t, labels.FromStrings("a", "a"), []sample{{5, 5}, {6, 66}}), // Overlap samples for some reason.
 			storeSeriesResponse(t, labels.FromStrings("a", "b"), []sample{{2, 2}, {3, 3}, {4, 4}}, []sample{{1, 1}, {2, 2}, {3, 3}}),
 			storeSeriesResponse(t, labels.FromStrings("a", "c"), []sample{{100, 1}, {300, 3}, {400, 4}}),
 		},
@@ -184,7 +188,7 @@ func TestQuerier_Series(t *testing.T) {
 	}{
 		{
 			lset:    labels.FromStrings("a", "a"),
-			samples: []sample{{2, 1}, {3, 2}},
+			samples: []sample{{2, 1}, {3, 2}, {5, 5}, {6, 6}, {7, 7}},
 		},
 		{
 			lset:    labels.FromStrings("a", "b"),
@@ -349,7 +353,7 @@ func TestDedupSeriesSet(t *testing.T) {
 			},
 		})
 	}
-	set := promSeriesSet{
+	set := &promSeriesSet{
 		mint: 1,
 		maxt: math.MaxInt64,
 		set:  newStoreSeriesSet(series),
@@ -521,6 +525,7 @@ func (s *storeServer) Series(r *storepb.SeriesRequest, srv storepb.Store_SeriesS
 	return nil
 }
 
+// storeSeriesResponse creates test storepb.SeriesResponse that includes series with single chunk that stores all the given samples.
 func storeSeriesResponse(t testing.TB, lset labels.Labels, smplChunks ...[]sample) *storepb.SeriesResponse {
 	var s storepb.Series