Define store-gateway protobuf and client

Makefile

@@ -119,7 +119,9 @@ $(EXES):
 protos: $(PROTO_GOS)
 
 %.pb.go:
-	protoc -I $(GOPATH)/src:./vendor:./$(@D) --gogoslick_out=plugins=grpc,Mgoogle/protobuf/any.proto=github.com/gogo/protobuf/types,:./$(@D) ./$(patsubst %.pb.go,%.proto,$@)
+	@# The store-gateway RPC is based on Thanos which uses relative references to other protos, so we need
+	@# to configure all such relative paths.
+	protoc -I $(GOPATH)/src:./vendor/github.com/thanos-io/thanos/pkg/store:./vendor/github.com/thanos-io/thanos/pkg/store/storepb:./vendor/github.com/gogo/protobuf:./vendor:./$(@D) --gogoslick_out=plugins=grpc,Mgoogle/protobuf/any.proto=github.com/gogo/protobuf/types,:./$(@D) ./$(patsubst %.pb.go,%.proto,$@)
 
 lint:
 	misspell -error docs

pkg/chunk/gcp/bigtable_index_client.go

@@ -51,7 +51,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.BoolVar(&cfg.TableCacheEnabled, "bigtable.table-cache.enabled", true, "If enabled, once a tables info is fetched, it is cached.")
 	f.DurationVar(&cfg.TableCacheExpiration, "bigtable.table-cache.expiration", 30*time.Minute, "Duration to cache tables before checking again.")
 
-	cfg.GRPCClientConfig.RegisterFlags("bigtable", f)
+	cfg.GRPCClientConfig.RegisterFlagsWithPrefix("bigtable", f)
 }
 
 // storageClientColumnKey implements chunk.storageClient for GCP.

pkg/ingester/client/client.go

@@ -3,17 +3,13 @@ package client
 import (
 	"flag"
 
-	otgrpc "github.com/opentracing-contrib/go-grpc"
-	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
-	"github.com/weaveworks/common/middleware"
 	"google.golang.org/grpc"
 	_ "google.golang.org/grpc/encoding/gzip" // get gzip compressor registered
 	"google.golang.org/grpc/health/grpc_health_v1"
 
 	"github.com/cortexproject/cortex/pkg/util/grpcclient"
-	cortex_middleware "github.com/cortexproject/cortex/pkg/util/middleware"
 )
 
 var ingesterClientRequestDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
@@ -36,22 +32,10 @@ type closableHealthAndIngesterClient struct {
 	conn *grpc.ClientConn
 }
 
-func instrumentation() ([]grpc.UnaryClientInterceptor, []grpc.StreamClientInterceptor) {
-	return []grpc.UnaryClientInterceptor{
-			otgrpc.OpenTracingClientInterceptor(opentracing.GlobalTracer()),
-			middleware.ClientUserHeaderInterceptor,
-			cortex_middleware.PrometheusGRPCUnaryInstrumentation(ingesterClientRequestDuration),
-		}, []grpc.StreamClientInterceptor{
-			otgrpc.OpenTracingStreamClientInterceptor(opentracing.GlobalTracer()),
-			middleware.StreamClientUserHeaderInterceptor,
-			cortex_middleware.PrometheusGRPCStreamInstrumentation(ingesterClientRequestDuration),
-		}
-}
-
 // MakeIngesterClient makes a new IngesterClient
 func MakeIngesterClient(addr string, cfg Config) (HealthAndIngesterClient, error) {
 	opts := []grpc.DialOption{grpc.WithInsecure()}
-	opts = append(opts, cfg.GRPCClientConfig.DialOption(instrumentation())...)
+	opts = append(opts, cfg.GRPCClientConfig.DialOption(grpcclient.Instrument(ingesterClientRequestDuration))...)
 	conn, err := grpc.Dial(addr, opts...)
 	if err != nil {
 		return nil, err
@@ -74,5 +58,5 @@ type Config struct {
 
 // RegisterFlags registers configuration settings used by the ingester client config.
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
-	cfg.GRPCClientConfig.RegisterFlags("ingester.client", f)
+	cfg.GRPCClientConfig.RegisterFlagsWithPrefix("ingester.client", f)
 }

pkg/querier/frontend/worker.go

@@ -44,7 +44,7 @@ func (cfg *WorkerConfig) RegisterFlags(f *flag.FlagSet) {
 	f.IntVar(&cfg.Parallelism, "querier.worker-parallelism", 10, "Number of simultaneous queries to process.")
 	f.DurationVar(&cfg.DNSLookupDuration, "querier.dns-lookup-period", 10*time.Second, "How often to query DNS.")
 
-	cfg.GRPCClientConfig.RegisterFlags("querier.frontend-client", f)
+	cfg.GRPCClientConfig.RegisterFlagsWithPrefix("querier.frontend-client", f)
 }
 
 // Worker is the counter-part to the frontend, actually processing requests.

pkg/querier/store_gateway_client.go

@@ -0,0 +1,52 @@
+package querier
+
+import (
+	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/health/grpc_health_v1"
+
+	"github.com/cortexproject/cortex/pkg/ring/client"
+	"github.com/cortexproject/cortex/pkg/storegateway/storegatewaypb"
+	"github.com/cortexproject/cortex/pkg/util/grpcclient"
+)
+
+func NewStoreGatewayClientFactory(cfg grpcclient.Config, reg prometheus.Registerer) client.PoolFactory {
+	requestDuration := promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
+		Namespace:   "cortex",
+		Name:        "storegateway_client_request_duration_seconds",
+		Help:        "Time spent executing requests on store-gateway.",
+		Buckets:     prometheus.ExponentialBuckets(0.008, 4, 7),
+		ConstLabels: prometheus.Labels{"client": "querier"},
+	}, []string{"operation", "status_code"})
+
+	return func(addr string) (client.PoolClient, error) {
+		return dialStoreGatewayClient(cfg, addr, requestDuration)
+	}
+}
+
+func dialStoreGatewayClient(cfg grpcclient.Config, addr string, requestDuration *prometheus.HistogramVec) (*storeGatewayClient, error) {
+	opts := []grpc.DialOption{grpc.WithInsecure()}
+	opts = append(opts, cfg.DialOption(grpcclient.Instrument(requestDuration))...)
+	conn, err := grpc.Dial(addr, opts...)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to dial store-gateway %s", addr)
+	}
+
+	return &storeGatewayClient{
+		StoreGatewayClient: storegatewaypb.NewStoreGatewayClient(conn),
+		HealthClient:       grpc_health_v1.NewHealthClient(conn),
+		conn:               conn,
+	}, nil
+}
+
+type storeGatewayClient struct {
+	storegatewaypb.StoreGatewayClient
+	grpc_health_v1.HealthClient
+	conn *grpc.ClientConn
+}
+
+func (c *storeGatewayClient) Close() error {
+	return c.conn.Close()
+}

pkg/querier/store_gateway_client_test.go

@@ -0,0 +1,74 @@
+package querier
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/thanos-io/thanos/pkg/store/storepb"
+	"github.com/weaveworks/common/user"
+	"google.golang.org/grpc"
+
+	"github.com/cortexproject/cortex/pkg/storegateway/storegatewaypb"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/grpcclient"
+)
+
+func TestNewStoreGatewayClientFactory(t *testing.T) {
+	// Create a GRPC server used to query the mocked service.
+	grpcServer := grpc.NewServer()
+	defer grpcServer.GracefulStop()
+
+	srv := &mockStoreGatewayServer{}
+	storegatewaypb.RegisterStoreGatewayServer(grpcServer, srv)
+
+	listener, err := net.Listen("tcp", "localhost:0")
+	require.NoError(t, err)
+
+	go func() {
+		require.NoError(t, grpcServer.Serve(listener))
+	}()
+
+	// Create a client factory and query back the mocked service
+	// with different clients.
+	cfg := grpcclient.Config{}
+	flagext.DefaultValues(&cfg)
+
+	reg := prometheus.NewPedanticRegistry()
+	factory := NewStoreGatewayClientFactory(cfg, reg)
+
+	for i := 0; i < 2; i++ {
+		client, err := factory(listener.Addr().String())
+		require.NoError(t, err)
+		defer client.Close() //nolint:errcheck
+
+		ctx := user.InjectOrgID(context.Background(), "test")
+		stream, err := client.(*storeGatewayClient).Series(ctx, &storepb.SeriesRequest{})
+		assert.NoError(t, err)
+
+		// Read the entire response from the stream.
+		for _, err = stream.Recv(); err == nil; {
+		}
+	}
+
+	// Assert on the request duration metric, but since it's a duration histogram and
+	// we can't predict the exact time it took, we need to workaround it.
+	metrics, err := reg.Gather()
+	require.NoError(t, err)
+
+	assert.Len(t, metrics, 1)
+	assert.Equal(t, "cortex_storegateway_client_request_duration_seconds", metrics[0].GetName())
+	assert.Equal(t, dto.MetricType_HISTOGRAM, metrics[0].GetType())
+	assert.Len(t, metrics[0].GetMetric(), 1)
+	assert.Equal(t, uint64(2), metrics[0].GetMetric()[0].GetHistogram().GetSampleCount())
+}
+
+type mockStoreGatewayServer struct{}
+
+func (m *mockStoreGatewayServer) Series(_ *storepb.SeriesRequest, srv storegatewaypb.StoreGateway_SeriesServer) error {
+	return nil
+}

pkg/storegateway/storegatewaypb/gateway.proto

@@ -0,0 +1,17 @@
+syntax = "proto3";
+package gatewaypb;
+
+import "github.com/thanos-io/thanos/pkg/store/storepb/rpc.proto";
+
+option go_package = "storegatewaypb";
+
+service StoreGateway {
+    // Series streams each Series for given label matchers and time range.
+    //
+    // Series should strictly stream full series after series, optionally split by time. This means that a single frame can contain
+    // partition of the single series, but once a new series is started to be streamed it means that no more data will
+    // be sent for previous one.
+    //
+    // Series are sorted.
+    rpc Series(thanos.SeriesRequest) returns (stream thanos.SeriesResponse);
+}