bumps cortex & fixes conflicts (#2204)

* bumps cortex & fixes conflicts

* go mod tidy

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/containerd/containerd v1.3.2 // indirect
 	github.com/containerd/fifo v0.0.0-20190226154929-a9fb20d87448 // indirect
 	github.com/coreos/go-systemd v0.0.0-20191104093116-d3cd4ed1dbcf
-	github.com/cortexproject/cortex v1.1.1-0.20200529121946-014195920b4f
+	github.com/cortexproject/cortex v1.1.1-0.20200609120740-6bd667db776a
 	github.com/davecgh/go-spew v1.1.1
 	github.com/docker/distribution v2.7.1+incompatible // indirect
 	github.com/docker/docker v0.7.3-0.20190817195342-4760db040282
@@ -43,23 +43,23 @@ require (
 	github.com/opentracing/opentracing-go v1.1.1-0.20200124165624-2876d2018785
 	github.com/pierrec/lz4 v2.5.3-0.20200429092203-e876bbd321b3+incompatible
 	github.com/pkg/errors v0.9.1
-	github.com/prometheus/client_golang v1.6.0
+	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.20200528145419-83619aa9ac56
+	github.com/prometheus/prometheus v1.8.2-0.20200609052543-1627d234da06
 	github.com/shurcooL/httpfs v0.0.0-20190707220628-8d4bc4ba7749
 	github.com/shurcooL/vfsgen v0.0.0-20181202132449-6a9ea43bcacd
 	github.com/stretchr/testify v1.5.1
 	github.com/tonistiigi/fifo v0.0.0-20190226154929-a9fb20d87448
-	github.com/uber/jaeger-client-go v2.23.0+incompatible
+	github.com/uber/jaeger-client-go v2.23.1+incompatible
 	github.com/ugorji/go v1.1.7 // indirect
 	github.com/weaveworks/common v0.0.0-20200512154658-384f10054ec5
 	go.etcd.io/bbolt v1.3.4
-	golang.org/x/net v0.0.0-20200421231249-e086a090c8fd
-	google.golang.org/grpc v1.29.0
+	golang.org/x/net v0.0.0-20200602114024-627f9648deb9
+	google.golang.org/grpc v1.29.1
 	gopkg.in/alecthomas/kingpin.v2 v2.2.6
 	gopkg.in/fsnotify.v1 v1.4.7
-	gopkg.in/yaml.v2 v2.2.8
+	gopkg.in/yaml.v2 v2.3.0
 	k8s.io/klog v1.0.0
 )
 
@@ -69,9 +69,10 @@ replace github.com/Azure/azure-sdk-for-go => github.com/Azure/azure-sdk-for-go v
 
 replace github.com/Azure/go-autorest => github.com/Azure/go-autorest v13.3.0+incompatible
 
-// Without this, it attempts to upgrade to v0.18.x which has some conflicts with upstream Prometheus.
-// Also, v0.17.5 is chosen to be consistent with Thanos and more clear than using a commit hash.
-replace k8s.io/client-go => k8s.io/client-go v0.17.5
+replace k8s.io/client-go => k8s.io/client-go v0.18.3
 
 // >v1.2.0 has some conflict with prometheus/alertmanager. Hence prevent the upgrade till it's fixed.
 replace github.com/satori/go.uuid => github.com/satori/go.uuid v1.2.0
+
+// Use fork of gocql that has gokit logs and Prometheus metrics.
+replace github.com/gocql/gocql => github.com/grafana/gocql v0.0.0-20200605141915-ba5dc39ece85

pkg/distributor/distributor.go

@@ -88,7 +88,7 @@ type Distributor struct {
 }
 
 // New a distributor creates.
-func New(cfg Config, clientCfg client.Config, ingestersRing ring.ReadRing, overrides *validation.Overrides) (*Distributor, error) {
+func New(cfg Config, clientCfg client.Config, ingestersRing ring.ReadRing, overrides *validation.Overrides, registerer prometheus.Registerer) (*Distributor, error) {
 	factory := cfg.factory
 	if factory == nil {
 		factory = func(addr string) (ring_client.PoolClient, error) {
@@ -109,7 +109,7 @@ func New(cfg Config, clientCfg client.Config, ingestersRing ring.ReadRing, overr
 
 	if overrides.IngestionRateStrategy() == validation.GlobalIngestionRateStrategy {
 		var err error
-		distributorsRing, err = ring.NewLifecycler(cfg.DistributorRing.ToLifecyclerConfig(), nil, "distributor", ring.DistributorRingKey, false)
+		distributorsRing, err = ring.NewLifecycler(cfg.DistributorRing.ToLifecyclerConfig(), nil, "distributor", ring.DistributorRingKey, false, registerer)
 		if err != nil {
 			return nil, err
 		}

pkg/distributor/distributor_test.go

@@ -226,7 +226,7 @@ func prepare(t *testing.T, limits *validation.Limits, kvStore kv.Client) *Distri
 		return ingesters[addr], nil
 	}
 
-	d, err := New(distributorConfig, clientConfig, ingestersRing, overrides)
+	d, err := New(distributorConfig, clientConfig, ingestersRing, overrides, nil)
 	require.NoError(t, err)
 	require.NoError(t, services.StartAndAwaitRunning(context.Background(), d))
 

pkg/ingester/flush_test.go

@@ -173,7 +173,7 @@ func newTestStore(t require.TestingT, cfg Config) (*testStore, *Ingester) {
 	limits, err := validation.NewOverrides(defaultLimitsTestConfig(), nil)
 	require.NoError(t, err)
 
-	ing, err := New(cfg, client.Config{}, store, limits)
+	ing, err := New(cfg, client.Config{}, store, limits, nil)
 	require.NoError(t, err)
 	require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing))
 
@@ -182,7 +182,7 @@ func newTestStore(t require.TestingT, cfg Config) (*testStore, *Ingester) {
 
 // nolint
 func defaultIngesterTestConfig(t *testing.T) Config {
-	kvClient, err := kv.NewClient(kv.Config{Store: "inmemory"}, ring.GetCodec())
+	kvClient, err := kv.NewClient(kv.Config{Store: "inmemory"}, ring.GetCodec(), nil)
 	require.NoError(t, err)
 
 	cfg := Config{}

pkg/ingester/ingester.go

@@ -126,7 +126,7 @@ type ChunkStore interface {
 }
 
 // New makes a new Ingester.
-func New(cfg Config, clientConfig client.Config, store ChunkStore, limits *validation.Overrides) (*Ingester, error) {
+func New(cfg Config, clientConfig client.Config, store ChunkStore, limits *validation.Overrides, registerer prometheus.Registerer) (*Ingester, error) {
 	if cfg.ingesterClientFactory == nil {
 		cfg.ingesterClientFactory = client.New
 	}
@@ -148,7 +148,7 @@ func New(cfg Config, clientConfig client.Config, store ChunkStore, limits *valid
 		},
 	}
 
-	i.lifecycler, err = ring.NewLifecycler(cfg.LifecyclerConfig, i, "ingester", ring.IngesterRingKey, true)
+	i.lifecycler, err = ring.NewLifecycler(cfg.LifecyclerConfig, i, "ingester", ring.IngesterRingKey, true, registerer)
 	if err != nil {
 		return nil, err
 	}

pkg/ingester/ingester_test.go

@@ -33,7 +33,7 @@ func TestIngester(t *testing.T) {
 		chunks: map[string][]chunk.Chunk{},
 	}
 
-	i, err := New(ingesterConfig, client.Config{}, store, limits)
+	i, err := New(ingesterConfig, client.Config{}, store, limits, nil)
 	require.NoError(t, err)
 	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 
@@ -201,7 +201,7 @@ func TestIngesterStreamLimitExceeded(t *testing.T) {
 		chunks: map[string][]chunk.Chunk{},
 	}
 
-	i, err := New(ingesterConfig, client.Config{}, store, overrides)
+	i, err := New(ingesterConfig, client.Config{}, store, overrides, nil)
 	require.NoError(t, err)
 	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 

pkg/ingester/transfer_test.go

@@ -122,7 +122,7 @@ type testIngesterFactory struct {
 }
 
 func newTestIngesterFactory(t *testing.T) *testIngesterFactory {
-	kvClient, err := kv.NewClient(kv.Config{Store: "inmemory"}, ring.GetCodec())
+	kvClient, err := kv.NewClient(kv.Config{Store: "inmemory"}, ring.GetCodec(), nil)
 	require.NoError(t, err)
 
 	return &testIngesterFactory{

pkg/loki/modules.go

@@ -86,7 +86,7 @@ func (t *Loki) initServer() (services.Service, error) {
 func (t *Loki) initRing() (_ services.Service, err error) {
 	t.cfg.Ingester.LifecyclerConfig.RingConfig.KVStore.Multi.ConfigProvider = multiClientRuntimeConfigChannel(t.runtimeConfig)
 	t.cfg.Ingester.LifecyclerConfig.RingConfig.KVStore.MemberlistKV = t.memberlistKV.GetMemberlistKV
-	t.ring, err = ring.New(t.cfg.Ingester.LifecyclerConfig.RingConfig, "ingester", ring.IngesterRingKey)
+	t.ring, err = ring.New(t.cfg.Ingester.LifecyclerConfig.RingConfig, "ingester", ring.IngesterRingKey, prometheus.DefaultRegisterer)
 	if err != nil {
 		return
 	}
@@ -120,7 +120,7 @@ func (t *Loki) initDistributor() (services.Service, error) {
 	t.cfg.Distributor.DistributorRing.KVStore.Multi.ConfigProvider = multiClientRuntimeConfigChannel(t.runtimeConfig)
 	t.cfg.Distributor.DistributorRing.KVStore.MemberlistKV = t.memberlistKV.GetMemberlistKV
 	var err error
-	t.distributor, err = distributor.New(t.cfg.Distributor, t.cfg.IngesterClient, t.ring, t.overrides)
+	t.distributor, err = distributor.New(t.cfg.Distributor, t.cfg.IngesterClient, t.ring, t.overrides, prometheus.DefaultRegisterer)
 	if err != nil {
 		return nil, err
 	}
@@ -188,7 +188,7 @@ func (t *Loki) initIngester() (_ services.Service, err error) {
 		t.cfg.Ingester.QueryStoreMaxLookBackPeriod = mlb
 	}
 
-	t.ingester, err = ingester.New(t.cfg.Ingester, t.cfg.IngesterClient, t.store, t.overrides)
+	t.ingester, err = ingester.New(t.cfg.Ingester, t.cfg.IngesterClient, t.store, t.overrides, prometheus.DefaultRegisterer)
 	if err != nil {
 		return
 	}
@@ -231,7 +231,7 @@ func (t *Loki) initTableManager() (services.Service, error) {
 	bucketClient, err := storage.NewBucketClient(t.cfg.StorageConfig.Config)
 	util.CheckFatal("initializing bucket client", err)
 
-	t.tableManager, err = chunk.NewTableManager(t.cfg.TableManager, t.cfg.SchemaConfig, maxChunkAgeForTableManager, tableClient, bucketClient, prometheus.DefaultRegisterer)
+	t.tableManager, err = chunk.NewTableManager(t.cfg.TableManager, t.cfg.SchemaConfig, maxChunkAgeForTableManager, tableClient, bucketClient, nil, prometheus.DefaultRegisterer)
 	if err != nil {
 		return nil, err
 	}

pkg/querier/queryrange/codec.go

@@ -10,13 +10,15 @@ import (
 	"net/http"
 	"net/url"
 	"sort"
+	strings "strings"
 	"time"
 
 	"github.com/cortexproject/cortex/pkg/ingester/client"
 	"github.com/cortexproject/cortex/pkg/querier/queryrange"
 	json "github.com/json-iterator/go"
 	"github.com/opentracing/opentracing-go"
 	otlog "github.com/opentracing/opentracing-go/log"
+	"github.com/prometheus/prometheus/pkg/timestamp"
 	"github.com/weaveworks/common/httpgrpc"
 
 	"github.com/grafana/loki/pkg/loghttp"
@@ -58,6 +60,18 @@ func (r *LokiRequest) WithShards(shards logql.Shards) *LokiRequest {
 	return &new
 }
 
+func (r *LokiRequest) LogToSpan(sp opentracing.Span) {
+	sp.LogFields(
+		otlog.String("query", r.GetQuery()),
+		otlog.String("start", timestamp.Time(r.GetStart()).String()),
+		otlog.String("end", timestamp.Time(r.GetEnd()).String()),
+		otlog.Int64("step (ms)", r.GetStep()),
+		otlog.Int64("limit", int64(r.GetLimit())),
+		otlog.String("direction", r.GetDirection().String()),
+		otlog.String("shards", strings.Join(r.GetShards(), ",")),
+	)
+}
+
 func (r *LokiSeriesRequest) GetEnd() int64 {
 	return r.EndTs.UnixNano() / (int64(time.Millisecond) / int64(time.Nanosecond))
 }
@@ -86,6 +100,14 @@ func (r *LokiSeriesRequest) GetStep() int64 {
 	return 0
 }
 
+func (r *LokiSeriesRequest) LogToSpan(sp opentracing.Span) {
+	sp.LogFields(
+		otlog.String("matchers", strings.Join(r.GetMatch(), ",")),
+		otlog.String("start", timestamp.Time(r.GetStart()).String()),
+		otlog.String("end", timestamp.Time(r.GetEnd()).String()),
+	)
+}
+
 func (codec) DecodeRequest(_ context.Context, r *http.Request) (queryrange.Request, error) {
 	if err := r.ParseForm(); err != nil {
 		return nil, httpgrpc.Errorf(http.StatusBadRequest, err.Error())

pkg/querier/queryrange/split_by_interval.go

@@ -138,7 +138,7 @@ func (h *splitByInterval) loop(ctx context.Context, ch <-chan *lokiResult) {
 	for data := range ch {
 
 		sp, ctx := opentracing.StartSpanFromContext(ctx, "interval")
-		queryrange.LogToSpan(ctx, data.req)
+		data.req.LogToSpan(sp)
 
 		resp, err := h.next.Do(ctx, data.req)
 

pkg/storage/stores/local/boltdb_shipper_table_client.go

@@ -39,6 +39,10 @@ func (b *boltDBShipperTableClient) CreateTable(ctx context.Context, desc chunk.T
 	return nil
 }
 
+func (b *boltDBShipperTableClient) Stop() {
+	b.objectClient.Stop()
+}
+
 func (b *boltDBShipperTableClient) DeleteTable(ctx context.Context, name string) error {
 	objects, dirs, err := b.objectClient.List(ctx, name)
 	if err != nil {