Upgrade prometheus to 2.19.1 (#2756)

* Upgraded Prometheus

Signed-off-by: Marco Pracucci <[email protected]>

* Fixed SeriesSet interface in Cortex

Signed-off-by: Marco Pracucci <[email protected]>

* Upgraded Thanos

Signed-off-by: Marco Pracucci <[email protected]>

* Partially fixed tests

Signed-off-by: Marco Pracucci <[email protected]>

* Fixed test

Signed-off-by: Marco Pracucci <[email protected]>

* Fixed remote read API

Signed-off-by: Marco Pracucci <[email protected]>

* Reuse EmptySeriesSet and ErrSeriesSet from Prometheus

Signed-off-by: Marco Pracucci <[email protected]>

* Updated Prometheus again to fix response status code

Signed-off-by: Marco Pracucci <[email protected]>

* Renamed back some variables

Signed-off-by: Marco Pracucci <[email protected]>

* Cleaned up lazyQuerier

Signed-off-by: Marco Pracucci <[email protected]>

Author: pracucci

go.mod

@@ -45,12 +45,12 @@ require (
 	github.com/prometheus/client_golang v1.6.1-0.20200604110148-03575cad4e55
 	github.com/prometheus/client_model v0.2.0
 	github.com/prometheus/common v0.10.0
-	github.com/prometheus/prometheus v1.8.2-0.20200609052543-1627d234da06
+	github.com/prometheus/prometheus v1.8.2-0.20200622142935-153f859b7499
 	github.com/rafaeljusto/redigomock v0.0.0-20190202135759-257e089e14a1
 	github.com/segmentio/fasthash v0.0.0-20180216231524-a72b379d632e
 	github.com/spf13/afero v1.2.2
 	github.com/stretchr/testify v1.5.1
-	github.com/thanos-io/thanos v0.12.3-0.20200603113103-6d3e730e154d
+	github.com/thanos-io/thanos v0.12.3-0.20200618165043-6c513e5f5c5f
 	github.com/uber/jaeger-client-go v2.23.1+incompatible
 	github.com/weaveworks/common v0.0.0-20200512154658-384f10054ec5
 	go.etcd.io/bbolt v1.3.5-0.20200615073812-232d8fc87f50

go.sum

@@ -4,6 +4,7 @@ package main
 
 import (
 	"bytes"
+	"context"
 	"io/ioutil"
 	"net/http"
 	"testing"
@@ -90,7 +91,11 @@ func TestQuerierRemoteRead(t *testing.T) {
 	require.NoError(t, err)
 	compressed := snappy.Encode(nil, data)
 
-	httpReq, err := http.NewRequest("POST", "http://"+querier.HTTPEndpoint()+"/prometheus/api/v1/read", bytes.NewReader(compressed))
+	// Call the remote read API endpoint with a timeout.
+	httpReqCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	httpReq, err := http.NewRequestWithContext(httpReqCtx, "POST", "http://"+querier.HTTPEndpoint()+"/prometheus/api/v1/read", bytes.NewReader(compressed))
 	require.NoError(t, err)
 	httpReq.Header.Set("X-Scope-OrgID", "user-1")
 	httpReq.Header.Add("Content-Encoding", "snappy")

integration/querier_remote_read_test.go

@@ -497,11 +497,7 @@ func (ng *Engine) populateSeries(ctx context.Context, q storage.Queryable, s *Ev
 				params.End = params.End - offsetMilliseconds
 			}
 
-			set, _, err = querier.Select(false, params, n.LabelMatchers...)
-			if err != nil {
-				level.Error(ng.logger).Log("msg", "error selecting series set", "err", err)
-				return err
-			}
+			set = querier.Select(false, params, n.LabelMatchers...)
 			n.series, err = expandSeriesSet(ctx, set)
 			if err != nil {
 				// TODO(fabxc): use multi-error.
@@ -520,11 +516,7 @@ func (ng *Engine) populateSeries(ctx context.Context, q storage.Queryable, s *Ev
 				params.End = params.End - offsetMilliseconds
 			}
 
-			set, _, err = querier.Select(false, params, n.LabelMatchers...)
-			if err != nil {
-				level.Error(ng.logger).Log("msg", "error selecting series set", "err", err)
-				return err
-			}
+			set = querier.Select(false, params, n.LabelMatchers...)
 			n.series, err = expandSeriesSet(ctx, set)
 			if err != nil {
 				level.Error(ng.logger).Log("msg", "error expanding series set", "err", err)

pkg/configs/legacy_promql/engine.go

@@ -129,24 +129,15 @@ type errQuerier struct {
 	err error
 }
 
-func (q *errQuerier) Select(bool, *storage.SelectHints, ...*labels.Matcher) (storage.SeriesSet, storage.Warnings, error) {
-	return errSeriesSet{err: q.err}, nil, q.err
+func (q *errQuerier) Select(bool, *storage.SelectHints, ...*labels.Matcher) storage.SeriesSet {
+	return storage.ErrSeriesSet(q.err)
 }
 func (q *errQuerier) LabelValues(name string) ([]string, storage.Warnings, error) {
 	return nil, nil, q.err
 }
 func (q *errQuerier) LabelNames() ([]string, storage.Warnings, error) { return nil, nil, q.err }
 func (q *errQuerier) Close() error                                    { return q.err }
 
-// errSeriesSet implements storage.SeriesSet which always returns error.
-type errSeriesSet struct {
-	err error
-}
-
-func (errSeriesSet) Next() bool         { return false }
-func (errSeriesSet) At() storage.Series { return nil }
-func (e errSeriesSet) Err() error       { return e.err }
-
 func TestQueryError(t *testing.T) {
 	engine := NewEngine(nil, nil, 10, 10*time.Second)
 	errStorage := ErrStorage(fmt.Errorf("storage error"))

pkg/configs/legacy_promql/engine_test.go

@@ -492,9 +492,9 @@ func (i *Ingester) v2Query(ctx context.Context, req *client.QueryRequest) (*clie
 	defer q.Close()
 
 	// It's not required to return sorted series because series are sorted by the Cortex querier.
-	ss, _, err := q.Select(false, nil, matchers...)
-	if err != nil {
-		return nil, err
+	ss := q.Select(false, nil, matchers...)
+	if ss.Err() != nil {
+		return nil, ss.Err()
 	}
 
 	numSamples := 0
@@ -615,9 +615,9 @@ func (i *Ingester) v2MetricsForLabelMatchers(ctx context.Context, req *client.Me
 	}
 
 	for _, matchers := range matchersSet {
-		seriesSet, _, err := q.Select(false, nil, matchers...)
-		if err != nil {
-			return nil, err
+		seriesSet := q.Select(false, nil, matchers...)
+		if seriesSet.Err() != nil {
+			return nil, seriesSet.Err()
 		}
 
 		for seriesSet.Next() {
@@ -723,9 +723,9 @@ func (i *Ingester) v2QueryStream(req *client.QueryRequest, stream client.Ingeste
 	defer q.Close()
 
 	// It's not required to return sorted series because series are sorted by the Cortex querier.
-	ss, _, err := q.Select(false, nil, matchers...)
-	if err != nil {
-		return err
+	ss := q.Select(false, nil, matchers...)
+	if ss.Err() != nil {
+		return ss.Err()
 	}
 
 	timeseries := make([]client.TimeSeries, 0, queryStreamBatchSize)

pkg/ingester/ingester_v2.go

@@ -86,7 +86,7 @@ type blocksQuerier struct {
 
 // Select implements storage.Querier interface.
 // The bool passed is ignored because the series is always sorted.
-func (b *blocksQuerier) Select(_ bool, sp *storage.SelectHints, matchers ...*labels.Matcher) (storage.SeriesSet, storage.Warnings, error) {
+func (b *blocksQuerier) Select(_ bool, sp *storage.SelectHints, matchers ...*labels.Matcher) storage.SeriesSet {
 	log, ctx := spanlogger.New(b.ctx, "blocksQuerier.Select")
 	defer log.Span.Finish()
 
@@ -106,12 +106,15 @@ func (b *blocksQuerier) Select(_ bool, sp *storage.SelectHints, matchers ...*lab
 		PartialResponseStrategy: storepb.PartialResponseStrategy_ABORT,
 	})
 	if err != nil {
-		return nil, nil, promql.ErrStorage{Err: err}
+		return storage.ErrSeriesSet(promql.ErrStorage{Err: err})
 	}
 
 	level.Debug(log).Log("series", len(series), "warnings", len(warnings))
 
-	return &blockQuerierSeriesSet{series: series}, warnings, nil
+	return &blockQuerierSeriesSet{
+		series:   series,
+		warnings: warnings,
+	}
 }
 
 func convertMatchersToLabelMatcher(matchers []*labels.Matcher) []storepb.LabelMatcher {
@@ -155,7 +158,8 @@ func (b *blocksQuerier) Close() error {
 
 // Implementation of storage.SeriesSet, based on individual responses from store client.
 type blockQuerierSeriesSet struct {
-	series []*storepb.Series
+	series   []*storepb.Series
+	warnings storage.Warnings
 
 	// next response to process
 	next int
@@ -195,6 +199,10 @@ func (bqss *blockQuerierSeriesSet) Err() error {
 	return nil
 }
 
+func (bqss *blockQuerierSeriesSet) Warnings() storage.Warnings {
+	return bqss.warnings
+}
+
 // newBlockQuerierSeries makes a new blockQuerierSeries. Input labels must be already sorted by name.
 func newBlockQuerierSeries(lbls []storepb.Label, chunks []storepb.AggrChunk) *blockQuerierSeries {
 	sort.Slice(chunks, func(i, j int) bool {

pkg/querier/block.go

@@ -279,15 +279,15 @@ type blocksStoreQuerier struct {
 
 // Select implements storage.Querier interface.
 // The bool passed is ignored because the series is always sorted.
-func (q *blocksStoreQuerier) Select(_ bool, sp *storage.SelectHints, matchers ...*labels.Matcher) (storage.SeriesSet, storage.Warnings, error) {
-	set, warnings, err := q.selectSorted(sp, matchers...)
+func (q *blocksStoreQuerier) Select(_ bool, sp *storage.SelectHints, matchers ...*labels.Matcher) storage.SeriesSet {
+	set := q.selectSorted(sp, matchers...)
 
 	// We need to wrap the error in order to have Prometheus returning a 5xx error.
-	if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
-		err = promql.ErrStorage{Err: err}
+	if err := set.Err(); err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
+		set = storage.ErrSeriesSet(promql.ErrStorage{Err: err})
 	}
 
-	return set, warnings, err
+	return set
 }
 
 func (q *blocksStoreQuerier) LabelValues(name string) ([]string, storage.Warnings, error) {
@@ -304,7 +304,7 @@ func (q *blocksStoreQuerier) Close() error {
 	return nil
 }
 
-func (q *blocksStoreQuerier) selectSorted(sp *storage.SelectHints, matchers ...*labels.Matcher) (storage.SeriesSet, storage.Warnings, error) {
+func (q *blocksStoreQuerier) selectSorted(sp *storage.SelectHints, matchers ...*labels.Matcher) storage.SeriesSet {
 	spanLog, spanCtx := spanlogger.New(q.ctx, "blocksStoreQuerier.selectSorted")
 	defer spanLog.Span.Finish()
 
@@ -329,20 +329,20 @@ func (q *blocksStoreQuerier) selectSorted(sp *storage.SelectHints, matchers ...*
 		if maxT < minT {
 			q.metrics.storesHit.Observe(0)
 			level.Debug(spanLog).Log("msg", "empty query time range after max time manipulation")
-			return series.NewEmptySeriesSet(), nil, nil
+			return storage.EmptySeriesSet()
 		}
 	}
 
 	// Find the list of blocks we need to query given the time range.
 	knownMetas, knownDeletionMarks, err := q.finder.GetBlocks(q.userID, minT, maxT)
 	if err != nil {
-		return nil, nil, err
+		return storage.ErrSeriesSet(err)
 	}
 
 	if len(knownMetas) == 0 {
 		q.metrics.storesHit.Observe(0)
 		level.Debug(spanLog).Log("msg", "no blocks found")
-		return series.NewEmptySeriesSet(), nil, nil
+		return storage.EmptySeriesSet()
 	}
 
 	level.Debug(spanLog).Log("msg", "found blocks to query", "expected", BlockMetas(knownMetas).String())
@@ -371,15 +371,15 @@ func (q *blocksStoreQuerier) selectSorted(sp *storage.SelectHints, matchers ...*
 				break
 			}
 
-			return nil, nil, err
+			return storage.ErrSeriesSet(err)
 		}
 		level.Debug(spanLog).Log("msg", "found store-gateway instances to query", "num instances", len(clients), "attempt", attempt)
 
 		// Fetch series from stores. If an error occur we do not retry because retries
 		// are only meant to cover missing blocks.
 		seriesSets, queriedBlocks, warnings, err := q.fetchSeriesFromStores(spanCtx, clients, minT, maxT, convertedMatchers)
 		if err != nil {
-			return nil, nil, err
+			return storage.ErrSeriesSet(err)
 		}
 		level.Debug(spanLog).Log("msg", "received series from all store-gateways", "queried blocks", strings.Join(convertULIDsToString(queriedBlocks), " "))
 
@@ -402,7 +402,9 @@ func (q *blocksStoreQuerier) selectSorted(sp *storage.SelectHints, matchers ...*
 			q.metrics.storesHit.Observe(float64(len(touchedStores)))
 			q.metrics.refetches.Observe(float64(attempt - 1))
 
-			return storage.NewMergeSeriesSet(resSeriesSets, storage.ChainedSeriesMerge), resWarnings, nil
+			return series.NewSeriesSetWithWarnings(
+				storage.NewMergeSeriesSet(resSeriesSets, storage.ChainedSeriesMerge),
+				resWarnings)
 		}
 
 		level.Debug(spanLog).Log("msg", "consistency check failed", "attempt", attempt, "missing blocks", strings.Join(convertULIDsToString(missingBlocks), " "))
@@ -415,7 +417,7 @@ func (q *blocksStoreQuerier) selectSorted(sp *storage.SelectHints, matchers ...*
 	err = fmt.Errorf("consistency check failed because some blocks were not queried: %s", strings.Join(convertULIDsToString(remainingBlocks), " "))
 	level.Warn(util.WithContext(spanCtx, spanLog)).Log("msg", "failed consistency check", "err", err)
 
-	return nil, nil, err
+	return storage.ErrSeriesSet(err)
 }
 
 func (q *blocksStoreQuerier) fetchSeriesFromStores(

pkg/querier/blocks_store_queryable.go

@@ -408,16 +408,16 @@ func TestBlocksStoreQuerier_SelectSorted(t *testing.T) {
 				labels.MustNewMatcher(labels.MatchEqual, labels.MetricName, metricName),
 			}
 
-			set, warnings, err := q.Select(true, nil, matchers...)
+			set := q.Select(true, nil, matchers...)
 			if testData.expectedErr != "" {
-				assert.EqualError(t, err, testData.expectedErr)
-				assert.Nil(t, set)
-				assert.Nil(t, warnings)
+				assert.EqualError(t, set.Err(), testData.expectedErr)
+				assert.False(t, set.Next())
+				assert.Nil(t, set.Warnings())
 				return
 			}
 
-			require.NoError(t, err)
-			assert.Len(t, warnings, 0)
+			require.NoError(t, set.Err())
+			assert.Len(t, set.Warnings(), 0)
 
 			// Read all returned series and their values.
 			var actualSeries []seriesResult
@@ -514,8 +514,8 @@ func TestBlocksStoreQuerier_SelectSortedShouldHonorQueryStoreAfter(t *testing.T)
 				End:   testData.queryMaxT,
 			}
 
-			_, _, err := q.selectSorted(sp, nil)
-			require.NoError(t, err)
+			set := q.selectSorted(sp, nil)
+			require.NoError(t, set.Err())
 
 			if testData.expectedMinT == 0 && testData.expectedMaxT == 0 {
 				assert.Len(t, finder.Calls, 0)

pkg/querier/blocks_store_queryable_test.go

@@ -39,29 +39,29 @@ type chunkStoreQuerier struct {
 
 // Select implements storage.Querier interface.
 // The bool passed is ignored because the series is always sorted.
-func (q *chunkStoreQuerier) Select(_ bool, sp *storage.SelectHints, matchers ...*labels.Matcher) (storage.SeriesSet, storage.Warnings, error) {
+func (q *chunkStoreQuerier) Select(_ bool, sp *storage.SelectHints, matchers ...*labels.Matcher) storage.SeriesSet {
 	userID, err := user.ExtractOrgID(q.ctx)
 	if err != nil {
-		return nil, nil, err
+		return storage.ErrSeriesSet(err)
 	}
 	chunks, err := q.store.Get(q.ctx, userID, model.Time(sp.Start), model.Time(sp.End), matchers...)
 	if err != nil {
 		switch err.(type) {
 		case promql.ErrStorage, promql.ErrTooManySamples, promql.ErrQueryCanceled, promql.ErrQueryTimeout:
 			// Recognized by Prometheus API, vendor/github.com/prometheus/prometheus/promql/engine.go:91.
 			// Don't translate those, just in case we use them internally.
-			return nil, nil, err
+			return storage.ErrSeriesSet(err)
 		case chunk.QueryError:
 			// This will be returned with status code 422 by Prometheus API.
 			// vendor/github.com/prometheus/prometheus/web/api/v1/api.go:1393
-			return nil, nil, err
+			return storage.ErrSeriesSet(err)
 		default:
 			// All other errors will be returned as 500.
-			return nil, nil, promql.ErrStorage{Err: err}
+			return storage.ErrSeriesSet(promql.ErrStorage{Err: err})
 		}
 	}
 
-	return partitionChunks(chunks, q.mint, q.maxt, q.chunkIteratorFunc), nil, nil
+	return partitionChunks(chunks, q.mint, q.maxt, q.chunkIteratorFunc)
 }
 
 // Series in the returned set are sorted alphabetically by labels.