sidecar: Added support for streaming, chunked remote read.

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