Update prometheus to latest (#763)

* Update prometheus to latest

* Upgrade to Go 1.10

* Fix up for API changes

* Update prometheus tsdb

* Update promql.NewEngine and v1.NewAPI

* Create querier.Config

* Reuse ruler config for ruler querier engine

* Clarify passing no flags map to NewEngine()

* Update Gopkg.lock

Author: aaron7

Gopkg.lock

@@ -19,6 +19,10 @@
   name = "github.com/prometheus/prometheus"
   branch = "master"
 
+[[constraint]]
+  name = "github.com/prometheus/tsdb"
+  branch = "master"
+
 [[override]]
   name = "k8s.io/client-go"
   revision = "3627aeb7d4f6ade38f995d2c923e459146493c7e"

Gopkg.toml

@@ -1,4 +1,4 @@
-FROM golang:1.9.3-stretch
+FROM golang:1.10.0-stretch
 RUN apt-get update && apt-get install -y python-requests python-yaml file jq unzip protobuf-compiler libprotobuf-dev && \
 	rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*
 RUN go clean -i net && \

build-image/Dockerfile

@@ -11,7 +11,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/route"
 	"github.com/prometheus/prometheus/config"
-	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/web/api/v1"
 	"github.com/prometheus/tsdb"
 	"google.golang.org/grpc"
@@ -42,6 +41,7 @@ func main() {
 
 		chunkStoreConfig  chunk.StoreConfig
 		distributorConfig distributor.Config
+		querierConfig     querier.Config
 		ingesterConfig    ingester.Config
 		configStoreConfig ruler.ConfigStoreConfig
 		rulerConfig       ruler.Config
@@ -53,7 +53,7 @@ func main() {
 	)
 	// Ingester needs to know our gRPC listen port.
 	ingesterConfig.ListenPort = &serverConfig.GRPCListenPort
-	util.RegisterFlags(&serverConfig, &chunkStoreConfig, &distributorConfig,
+	util.RegisterFlags(&serverConfig, &chunkStoreConfig, &distributorConfig, &querierConfig,
 		&ingesterConfig, &configStoreConfig, &rulerConfig, &storageConfig, &schemaConfig, &logLevel)
 	flag.BoolVar(&unauthenticated, "unauthenticated", false, "Set to true to disable multitenancy.")
 	flag.Parse()
@@ -123,13 +123,15 @@ func main() {
 	tableManager.Start()
 	defer tableManager.Stop()
 
+	engine, queryable := querier.NewEngine(dist, chunkStore, nil, querierConfig.MaxConcurrent, querierConfig.Timeout)
+
 	if configStoreConfig.ConfigsAPIURL.String() != "" || configStoreConfig.DBConfig.URI != "" {
 		rulesAPI, err := ruler.NewRulesAPI(configStoreConfig)
 		if err != nil {
 			level.Error(util.Logger).Log("msg", "error initializing ruler config store", "err", err)
 			os.Exit(1)
 		}
-		rlr, err := ruler.NewRuler(rulerConfig, dist, chunkStore)
+		rlr, err := ruler.NewRuler(rulerConfig, engine, queryable, dist)
 		if err != nil {
 			level.Error(util.Logger).Log("msg", "error initializing ruler", "err", err)
 			os.Exit(1)
@@ -147,13 +149,13 @@ func main() {
 	sampleQueryable := querier.NewQueryable(dist, chunkStore, false)
 	metadataQueryable := querier.NewQueryable(dist, chunkStore, true)
 
-	engine := promql.NewEngine(sampleQueryable, nil)
 	api := v1.NewAPI(
 		engine,
 		metadataQueryable,
 		querier.DummyTargetRetriever{},
 		querier.DummyAlertmanagerRetriever{},
 		func() config.Config { return config.Config{} },
+		nil, // flags to be served via http; we don't need this
 		func(f http.HandlerFunc) http.HandlerFunc { return f },
 		func() *tsdb.DB { return nil }, // Only needed for admin APIs.
 		false, // Disable admin APIs.

cmd/lite/main.go

@@ -36,12 +36,13 @@ func main() {
 		}
 		ringConfig        ring.Config
 		distributorConfig distributor.Config
+		querierConfig     querier.Config
 		chunkStoreConfig  chunk.StoreConfig
 		schemaConfig      chunk.SchemaConfig
 		storageConfig     storage.Config
 		logLevel          util.LogLevel
 	)
-	util.RegisterFlags(&serverConfig, &ringConfig, &distributorConfig,
+	util.RegisterFlags(&serverConfig, &ringConfig, &distributorConfig, &querierConfig,
 		&chunkStoreConfig, &schemaConfig, &storageConfig, &logLevel)
 	flag.Parse()
 
@@ -90,13 +91,14 @@ func main() {
 	sampleQueryable := querier.NewQueryable(dist, chunkStore, false)
 	metadataQueryable := querier.NewQueryable(dist, chunkStore, true)
 
-	engine := promql.NewEngine(sampleQueryable, nil)
+	engine := promql.NewEngine(util.Logger, nil, querierConfig.MaxConcurrent, querierConfig.Timeout)
 	api := v1.NewAPI(
 		engine,
 		metadataQueryable,
 		querier.DummyTargetRetriever{},
 		querier.DummyAlertmanagerRetriever{},
 		func() config.Config { return config.Config{} },
+		nil, // flags to be served via http; we don't need this
 		func(f http.HandlerFunc) http.HandlerFunc { return f },
 		func() *tsdb.DB { return nil }, // Only needed for admin APIs.
 		false, // Disable admin APIs.

cmd/querier/main.go

@@ -14,6 +14,7 @@ import (
 	"github.com/weaveworks/cortex/pkg/chunk"
 	"github.com/weaveworks/cortex/pkg/chunk/storage"
 	"github.com/weaveworks/cortex/pkg/distributor"
+	"github.com/weaveworks/cortex/pkg/querier"
 	"github.com/weaveworks/cortex/pkg/ring"
 	"github.com/weaveworks/cortex/pkg/ruler"
 	"github.com/weaveworks/cortex/pkg/util"
@@ -75,7 +76,8 @@ func main() {
 	defer dist.Stop()
 	prometheus.MustRegister(dist)
 
-	rlr, err := ruler.NewRuler(rulerConfig, dist, chunkStore)
+	engine, queryable := querier.NewEngine(dist, chunkStore, prometheus.DefaultRegisterer, rulerConfig.NumWorkers, rulerConfig.GroupTimeout)
+	rlr, err := ruler.NewRuler(rulerConfig, engine, queryable, dist)
 	if err != nil {
 		level.Error(util.Logger).Log("msg", "error initializing ruler", "err", err)
 		os.Exit(1)

cmd/ruler/main.go

@@ -12,8 +12,14 @@ type DummyTargetRetriever struct{}
 // Targets implements TargetRetriever.
 func (r DummyTargetRetriever) Targets() []*scrape.Target { return nil }
 
+// DroppedTargets implements TargetRetriever.
+func (r DummyTargetRetriever) DroppedTargets() []*scrape.Target { return nil }
+
 // DummyAlertmanagerRetriever implements AlertmanagerRetriever.
 type DummyAlertmanagerRetriever struct{}
 
 // Alertmanagers implements AlertmanagerRetriever.
 func (r DummyAlertmanagerRetriever) Alertmanagers() []*url.URL { return nil }
+
+// DroppedAlertmanagers implements AlertmanagerRetriever.
+func (r DummyAlertmanagerRetriever) DroppedAlertmanagers() []*url.URL { return nil }

pkg/querier/dummy.go

@@ -2,28 +2,45 @@ package querier
 
 import (
 	"context"
+	"flag"
 	"net/http"
+	"time"
 
 	"github.com/go-kit/kit/log/level"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/storage"
+
 	"github.com/weaveworks/cortex/pkg/ingester/client"
 	"github.com/weaveworks/cortex/pkg/prom1/storage/metric"
 
 	"github.com/weaveworks/cortex/pkg/util"
 )
 
+// Config contains the configuration require to create a querier
+type Config struct {
+	MaxConcurrent int
+	Timeout       time.Duration
+}
+
+// RegisterFlags adds the flags required to config this to the given FlagSet
+func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
+	flag.IntVar(&cfg.MaxConcurrent, "querier.max-concurrent", 20, "The maximum number of concurrent queries.")
+	flag.DurationVar(&cfg.Timeout, "querier.timeout", 2*time.Minute, "The timeout for a query.")
+}
+
 // ChunkStore is the interface we need to get chunks
 type ChunkStore interface {
 	Get(ctx context.Context, from, through model.Time, matchers ...*labels.Matcher) (model.Matrix, error)
 }
 
 // NewEngine creates a new promql.Engine for cortex.
-func NewEngine(distributor Querier, chunkStore ChunkStore) *promql.Engine {
+func NewEngine(distributor Querier, chunkStore ChunkStore, reg prometheus.Registerer, maxConcurrent int, timeout time.Duration) (*promql.Engine, storage.Queryable) {
 	queryable := NewQueryable(distributor, chunkStore, false)
-	return promql.NewEngine(queryable, nil)
+	engine := promql.NewEngine(util.Logger, reg, maxConcurrent, timeout)
+	return engine, queryable
 }
 
 // NewQueryable creates a new Queryable for cortex.
@@ -195,7 +212,7 @@ type mergeQuerier struct {
 	metadataOnly bool
 }
 
-func (mq mergeQuerier) Select(matchers ...*labels.Matcher) (storage.SeriesSet, error) {
+func (mq mergeQuerier) Select(_ *storage.SelectParams, matchers ...*labels.Matcher) (storage.SeriesSet, error) {
 	// TODO: Update underlying selectors to return errors directly.
 	if mq.metadataOnly {
 		return mq.selectMetadata(matchers...), nil

pkg/querier/querier.go

@@ -110,7 +110,7 @@ func TestMergeQuerierSortsMetricLabels(t *testing.T) {
 	}
 	m, err := labels.NewMatcher(labels.MatchEqual, model.MetricNameLabel, "testmetric")
 	require.NoError(t, err)
-	ss, err := mq.Select(m)
+	ss, err := mq.Select(nil, m)
 	require.NoError(t, err)
 	require.NoError(t, ss.Err())
 	ss.Next()

pkg/querier/querier_test.go

@@ -26,16 +26,15 @@ import (
 	"github.com/prometheus/prometheus/notifier"
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/rules"
+	"github.com/prometheus/prometheus/storage"
 	"github.com/prometheus/prometheus/util/strutil"
 	"golang.org/x/net/context"
 	"golang.org/x/net/context/ctxhttp"
 
 	"github.com/weaveworks/common/instrument"
 	"github.com/weaveworks/common/user"
-	"github.com/weaveworks/cortex/pkg/chunk"
 	"github.com/weaveworks/cortex/pkg/configs"
 	"github.com/weaveworks/cortex/pkg/distributor"
-	"github.com/weaveworks/cortex/pkg/querier"
 	"github.com/weaveworks/cortex/pkg/util"
 )
 
@@ -123,6 +122,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 // Ruler evaluates rules.
 type Ruler struct {
 	engine        *promql.Engine
+	queryable     storage.Queryable
 	pusher        Pusher
 	alertURL      *url.URL
 	notifierCfg   *config.Config
@@ -196,13 +196,14 @@ func (rn *rulerNotifier) stop() {
 }
 
 // NewRuler creates a new ruler from a distributor and chunk store.
-func NewRuler(cfg Config, d *distributor.Distributor, c *chunk.Store) (*Ruler, error) {
+func NewRuler(cfg Config, engine *promql.Engine, queryable storage.Queryable, d *distributor.Distributor) (*Ruler, error) {
 	ncfg, err := buildNotifierConfig(&cfg)
 	if err != nil {
 		return nil, err
 	}
 	return &Ruler{
-		engine:        querier.NewEngine(d, c),
+		engine:        engine,
+		queryable:     queryable,
 		pusher:        d,
 		alertURL:      cfg.ExternalURL.URL,
 		notifierCfg:   ncfg,
@@ -283,7 +284,7 @@ func (r *Ruler) newGroup(ctx context.Context, userID string, item *workItem) (*r
 	}
 	opts := &rules.ManagerOptions{
 		Appendable:  appendable,
-		QueryFunc:   rules.EngineQueryFunc(r.engine),
+		QueryFunc:   rules.EngineQueryFunc(r.engine, r.queryable),
 		Context:     ctx,
 		ExternalURL: r.alertURL,
 		NotifyFunc:  sendAlerts(notifier, r.alertURL.String()),

pkg/ruler/ruler.go

@@ -13,6 +13,7 @@ import (
 	"github.com/prometheus/prometheus/notifier"
 	"github.com/stretchr/testify/assert"
 	"github.com/weaveworks/common/user"
+	"github.com/weaveworks/cortex/pkg/querier"
 )
 
 func newTestRuler(t *testing.T, alertmanagerURL string) *Ruler {
@@ -25,7 +26,9 @@ func newTestRuler(t *testing.T, alertmanagerURL string) *Ruler {
 
 	// TODO: Populate distributor and chunk store arguments to enable
 	// other kinds of tests.
-	ruler, err := NewRuler(cfg, nil, nil)
+
+	engine, queryable := querier.NewEngine(nil, nil, nil, 20, 2*time.Minute)
+	ruler, err := NewRuler(cfg, engine, queryable, nil)
 	if err != nil {
 		t.Fatal(err)
 	}