Fixes after review

Author: povilasv

cmd/thanos/query.go

@@ -308,7 +308,7 @@ func runQuery(
 			},
 			dialOpts,
 		)
-		proxy            = store.NewProxyStore(logger, reg, stores.Get, component.Query, selectorLset, storeResponseTimeout)
+		proxy            = store.NewProxyStore(logger, stores.Get, component.Query, selectorLset, storeResponseTimeout)
 		queryableCreator = query.NewQueryableCreator(logger, proxy, replicaLabel)
 		engine           = promql.NewEngine(
 			promql.EngineOpts{

pkg/store/proxy.go

@@ -15,7 +15,6 @@ import (
 	"github.com/improbable-eng/thanos/pkg/store/storepb"
 	"github.com/improbable-eng/thanos/pkg/strutil"
 	"github.com/pkg/errors"
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/tsdb/labels"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc/codes"
@@ -45,50 +44,28 @@ type ProxyStore struct {
 	component      component.StoreAPI
 	selectorLabels labels.Labels
 
-	responseTimeout        time.Duration
-	streamResponseDuration *prometheus.HistogramVec
-	streamDuration         *prometheus.HistogramVec
+	responseTimeout time.Duration
 }
 
 // NewProxyStore returns a new ProxyStore that uses the given clients that implements storeAPI to fan-in all series to the client.
 // Note that there is no deduplication support. Deduplication should be done on the highest level (just before PromQL)
 func NewProxyStore(
 	logger log.Logger,
-	reg prometheus.Registerer,
 	stores func() []Client,
 	component component.StoreAPI,
 	selectorLabels labels.Labels,
 	responseTimeout time.Duration,
 ) *ProxyStore {
-	streamResponseDuration := prometheus.NewHistogramVec(prometheus.HistogramOpts{
-		Name: "thanos_query_stream_read_duration_seconds",
-		Help: "Time it takes to perform a single read from GRPC stream.",
-		Buckets: []float64{
-			0.005, 0.01, 0.02, 0.03, 0.04, 0.05, 0.1, 0.25, 0.6, 1, 2, 3.5, 5,
-		}}, []string{"store"})
-	streamDuration := prometheus.NewHistogramVec(prometheus.HistogramOpts{
-		Name: "thanos_query_stream_duration_seconds",
-		Help: "Time it takes to consume GRPC stream.",
-		Buckets: []float64{
-			0.01, 0.02, 0.03, 0.04, 0.05, 0.1, 0.25, 0.6, 1, 2, 3.5, 5, 7.5, 10, 15, 20,
-		}}, []string{"store"})
-	reg.MustRegister(
-		streamResponseDuration,
-		streamDuration,
-	)
-
 	if logger == nil {
 		logger = log.NewNopLogger()
 	}
 
 	s := &ProxyStore{
-		logger:                 logger,
-		stores:                 stores,
-		component:              component,
-		selectorLabels:         selectorLabels,
-		responseTimeout:        responseTimeout,
-		streamResponseDuration: streamResponseDuration,
-		streamDuration:         streamDuration,
+		logger:          logger,
+		stores:          stores,
+		component:       component,
+		selectorLabels:  selectorLabels,
+		responseTimeout: responseTimeout,
 	}
 	return s
 }
@@ -198,7 +175,7 @@ func (s *ProxyStore) Series(r *storepb.SeriesRequest, srv storepb.Store_SeriesSe
 			}
 
 			// Schedule streamSeriesSet that translates gRPC streamed response into seriesSet (if series) or respCh if warnings.
-			seriesSet = append(seriesSet, startStreamSeriesSet(seriesCtx, s.logger, closeSeries, wg, sc, respSender, st.String(), !r.PartialResponseDisabled, s.responseTimeout, s.streamResponseDuration.WithLabelValues(st.Addr()), s.streamDuration.WithLabelValues(st.Addr())))
+			seriesSet = append(seriesSet, startStreamSeriesSet(seriesCtx, s.logger, closeSeries, wg, sc, respSender, st.String(), !r.PartialResponseDisabled, s.responseTimeout))
 		}
 
 		level.Debug(s.logger).Log("msg", strings.Join(storeDebugMsgs, ";"))
@@ -254,10 +231,8 @@ type streamSeriesSet struct {
 	name            string
 	partialResponse bool
 
-	responseTimeout        time.Duration
-	closeSeries            context.CancelFunc
-	streamResponseDuration prometheus.Observer
-	streamDuration         prometheus.Observer
+	responseTimeout time.Duration
+	closeSeries     context.CancelFunc
 }
 
 func startStreamSeriesSet(
@@ -270,35 +245,26 @@ func startStreamSeriesSet(
 	name string,
 	partialResponse bool,
 	responseTimeout time.Duration,
-	streamResponseDuration prometheus.Observer,
-	streamDuration prometheus.Observer,
 ) *streamSeriesSet {
 	s := &streamSeriesSet{
-		ctx:                    ctx,
-		logger:                 logger,
-		closeSeries:            closeSeries,
-		stream:                 stream,
-		warnCh:                 warnCh,
-		recvCh:                 make(chan *storepb.Series, 10),
-		name:                   name,
-		partialResponse:        partialResponse,
-		responseTimeout:        responseTimeout,
-		streamResponseDuration: streamResponseDuration,
-		streamDuration:         streamDuration,
+		ctx:             ctx,
+		logger:          logger,
+		closeSeries:     closeSeries,
+		stream:          stream,
+		warnCh:          warnCh,
+		recvCh:          make(chan *storepb.Series, 10),
+		name:            name,
+		partialResponse: partialResponse,
+		responseTimeout: responseTimeout,
 	}
 
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		defer close(s.recvCh)
 
-		timer := prometheus.NewTimer(s.streamDuration)
-		defer timer.ObserveDuration()
-
 		for {
-			timer := prometheus.NewTimer(s.streamResponseDuration)
 			r, err := s.stream.Recv()
-			timer.ObserveDuration()
 
 			if err == io.EOF {
 				return

pkg/store/proxy_test.go

@@ -14,7 +14,6 @@ import (
 	"github.com/improbable-eng/thanos/pkg/store/storepb"
 	"github.com/improbable-eng/thanos/pkg/testutil"
 	"github.com/pkg/errors"
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/tsdb/chunkenc"
 	tlabels "github.com/prometheus/tsdb/labels"
@@ -53,7 +52,7 @@ func TestProxyStore_Info(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	q := NewProxyStore(nil, prometheus.NewRegistry(),
+	q := NewProxyStore(nil,
 		func() []Client { return nil },
 		component.Query,
 		nil, 0*time.Second,
@@ -407,7 +406,7 @@ func TestProxyStore_Series(t *testing.T) {
 	} {
 
 		if ok := t.Run(tc.title, func(t *testing.T) {
-			q := NewProxyStore(nil, prometheus.NewRegistry(),
+			q := NewProxyStore(nil,
 				func() []Client { return tc.storeAPIs },
 				component.Query,
 				tc.selectorLabels,
@@ -529,7 +528,7 @@ func TestProxyStore_SeriesSlowStores(t *testing.T) {
 		},
 	} {
 		if ok := t.Run(tc.title, func(t *testing.T) {
-			q := NewProxyStore(nil, prometheus.NewRegistry(),
+			q := NewProxyStore(nil,
 				func() []Client { return tc.storeAPIs },
 				component.Query,
 				tc.selectorLabels,
@@ -571,7 +570,7 @@ func TestProxyStore_Series_RequestParamsProxied(t *testing.T) {
 			maxTime:     300,
 		},
 	}
-	q := NewProxyStore(nil, prometheus.NewRegistry(),
+	q := NewProxyStore(nil,
 		func() []Client { return cls },
 		component.Query,
 		nil,
@@ -630,7 +629,7 @@ func TestProxyStore_Series_RegressionFillResponseChannel(t *testing.T) {
 
 	}
 
-	q := NewProxyStore(nil, prometheus.NewRegistry(),
+	q := NewProxyStore(nil,
 		func() []Client { return cls },
 		component.Query,
 		tlabels.FromStrings("fed", "a"),
@@ -668,7 +667,7 @@ func TestProxyStore_LabelValues(t *testing.T) {
 			},
 		}},
 	}
-	q := NewProxyStore(nil, prometheus.NewRegistry(),
+	q := NewProxyStore(nil,
 		func() []Client { return cls },
 		component.Query,
 		nil,