Make Cortex work with updated Prometheus packages

Author: juliusv

cmd/lite/main.go

@@ -13,6 +13,7 @@ import (
 	"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"
 
 	"github.com/weaveworks/common/middleware"
@@ -155,6 +156,8 @@ func main() {
 		querier.DummyAlertmanagerRetriever{},
 		func() config.Config { return config.Config{} },
 		func(f http.HandlerFunc) http.HandlerFunc { return f },
+		func() *tsdb.DB { return nil }, // Only needed for admin APIs.
+		false, // Disable admin APIs.
 	)
 	promRouter := route.New().WithPrefix("/api/prom/api/v1")
 	api.Register(promRouter)

cmd/querier/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/prometheus/prometheus/config"
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/web/api/v1"
+	"github.com/prometheus/tsdb"
 
 	"github.com/weaveworks/common/middleware"
 	"github.com/weaveworks/common/server"
@@ -98,6 +99,8 @@ func main() {
 		querier.DummyAlertmanagerRetriever{},
 		func() config.Config { return config.Config{} },
 		func(f http.HandlerFunc) http.HandlerFunc { return f },
+		func() *tsdb.DB { return nil }, // Only needed for admin APIs.
+		false, // Disable admin APIs.
 	)
 	promRouter := route.New().WithPrefix("/api/prom/api/v1")
 	api.Register(promRouter)

pkg/querier/querier.go

@@ -195,27 +195,28 @@ type mergeQuerier struct {
 	metadataOnly bool
 }
 
-func (mq mergeQuerier) Select(matchers ...*labels.Matcher) storage.SeriesSet {
+func (mq mergeQuerier) Select(matchers ...*labels.Matcher) (storage.SeriesSet, error) {
+	// TODO: Update underlying selectors to return errors directly.
 	if mq.metadataOnly {
-		return mq.selectMetadata(matchers...)
+		return mq.selectMetadata(matchers...), nil
 	}
-	return mq.selectSamples(matchers...)
+	return mq.selectSamples(matchers...), nil
 }
 
 func (mq mergeQuerier) selectMetadata(matchers ...*labels.Matcher) storage.SeriesSet {
 	// NB that we don't do this in parallel, as in practice we only have two queriers,
 	// one of which is the chunk store, which doesn't implement this yet.
-	var set storage.SeriesSet
+	seriesSets := make([]storage.SeriesSet, 0, len(mq.queriers))
 	for _, q := range mq.queriers {
 		ms, err := q.MetricsForLabelMatchers(mq.ctx, model.Time(mq.mint), model.Time(mq.maxt), matchers...)
 		if err != nil {
 			return errSeriesSet{err: err}
 		}
 		ss := metricsToSeriesSet(ms)
-		set = storage.DeduplicateSeriesSet(set, ss)
+		seriesSets = append(seriesSets, ss)
 	}
 
-	return set
+	return storage.NewMergeSeriesSet(seriesSets)
 }
 
 func (mq mergeQuerier) selectSamples(matchers ...*labels.Matcher) storage.SeriesSet {

pkg/ruler/ruler.go

@@ -1,6 +1,7 @@
 package ruler
 
 import (
+	native_ctx "context"
 	"flag"
 	"fmt"
 	"net/http"
@@ -12,11 +13,17 @@ import (
 	gklog "github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
 	"github.com/prometheus/client_golang/prometheus"
+	config_util "github.com/prometheus/common/config"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/config"
+	"github.com/prometheus/prometheus/discovery"
+	sd_config "github.com/prometheus/prometheus/discovery/config"
+	"github.com/prometheus/prometheus/discovery/dns"
+	"github.com/prometheus/prometheus/discovery/targetgroup"
 	"github.com/prometheus/prometheus/notifier"
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/rules"
+	"github.com/prometheus/prometheus/util/strutil"
 	"golang.org/x/net/context"
 	"golang.org/x/net/context/ctxhttp"
 
@@ -97,7 +104,59 @@ type Ruler struct {
 
 	// Per-user notifiers with separate queues.
 	notifiersMtx sync.Mutex
-	notifiers    map[string]*notifier.Notifier
+	notifiers    map[string]*rulerNotifier
+}
+
+type rulerNotifier struct {
+	notifier  *notifier.Notifier
+	sdCtx     context.Context
+	sdCancel  context.CancelFunc
+	sdManager *discovery.Manager
+	wg        sync.WaitGroup
+	logger    gklog.Logger
+}
+
+func newRulerNotifier(o *notifier.Options, l gklog.Logger) *rulerNotifier {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &rulerNotifier{
+		notifier:  notifier.New(o, l),
+		sdCtx:     ctx,
+		sdCancel:  cancel,
+		sdManager: discovery.NewManager(l),
+		logger:    l,
+	}
+}
+
+func (rn *rulerNotifier) run() {
+	rn.wg.Add(2)
+	go func() {
+		if err := rn.sdManager.Run(rn.sdCtx); err != nil {
+			level.Error(rn.logger).Log("msg", "error starting notifier discovery manager", "err", err)
+		}
+	}()
+	go func() {
+		rn.notifier.Run(rn.sdManager.SyncCh())
+	}()
+}
+
+func (rn *rulerNotifier) applyConfig(cfg *config.Config) error {
+	if err := rn.notifier.ApplyConfig(cfg); err != nil {
+		return err
+	}
+
+	amConfigs := cfg.AlertingConfig.AlertmanagerConfigs
+	if len(amConfigs) != 1 {
+		return fmt.Errorf("ruler alerting config should have exactly one AlertmanagerConfig")
+	}
+	return rn.sdManager.ApplyConfig(
+		map[string]sd_config.ServiceDiscoveryConfig{"ruler": amConfigs[0].ServiceDiscoveryConfig},
+	)
+}
+
+func (rn *rulerNotifier) stop() {
+	rn.sdCancel()
+	rn.notifier.Stop()
+	rn.wg.Wait()
 }
 
 // NewRuler creates a new ruler from a distributor and chunk store.
@@ -112,7 +171,7 @@ func NewRuler(cfg Config, d *distributor.Distributor, c *chunk.Store) (*Ruler, e
 		alertURL:      cfg.ExternalURL.URL,
 		notifierCfg:   ncfg,
 		queueCapacity: cfg.NotificationQueueCapacity,
-		notifiers:     map[string]*notifier.Notifier{},
+		notifiers:     map[string]*rulerNotifier{},
 	}, nil
 }
 
@@ -124,23 +183,23 @@ func buildNotifierConfig(rulerConfig *Config) (*config.Config, error) {
 	}
 
 	u := rulerConfig.AlertmanagerURL
-	var sdConfig config.ServiceDiscoveryConfig
+	var sdConfig sd_config.ServiceDiscoveryConfig
 	if rulerConfig.AlertmanagerDiscovery {
 		if !strings.Contains(u.Host, "_tcp.") {
 			return nil, fmt.Errorf("When alertmanager-discovery is on, host name must be of the form _portname._tcp.service.fqdn (is %q)", u.Host)
 		}
-		dnsSDConfig := config.DNSSDConfig{
+		dnsSDConfig := dns.SDConfig{
 			Names:           []string{u.Host},
 			RefreshInterval: model.Duration(rulerConfig.AlertmanagerRefreshInterval),
 			Type:            "SRV",
 			Port:            0, // Ignored, because of SRV.
 		}
-		sdConfig = config.ServiceDiscoveryConfig{
-			DNSSDConfigs: []*config.DNSSDConfig{&dnsSDConfig},
+		sdConfig = sd_config.ServiceDiscoveryConfig{
+			DNSSDConfigs: []*dns.SDConfig{&dnsSDConfig},
 		}
 	} else {
-		sdConfig = config.ServiceDiscoveryConfig{
-			StaticConfigs: []*config.TargetGroup{
+		sdConfig = sd_config.ServiceDiscoveryConfig{
+			StaticConfigs: []*targetgroup.Group{
 				{
 					Targets: []model.LabelSet{
 						{
@@ -165,14 +224,14 @@ func buildNotifierConfig(rulerConfig *Config) (*config.Config, error) {
 	}
 
 	if u.User != nil {
-		amConfig.HTTPClientConfig = config.HTTPClientConfig{
-			BasicAuth: &config.BasicAuth{
+		amConfig.HTTPClientConfig = config_util.HTTPClientConfig{
+			BasicAuth: &config_util.BasicAuth{
 				Username: u.User.Username(),
 			},
 		}
 
 		if password, isSet := u.User.Password(); isSet {
-			amConfig.HTTPClientConfig.BasicAuth.Password = config.Secret(password)
+			amConfig.HTTPClientConfig.BasicAuth.Password = config_util.Secret(password)
 		}
 	}
 
@@ -191,26 +250,59 @@ func (r *Ruler) newGroup(ctx context.Context, rs []rules.Rule) (*rules.Group, er
 	}
 	opts := &rules.ManagerOptions{
 		Appendable:  appendable,
-		QueryEngine: r.engine,
+		QueryFunc:   rules.EngineQueryFunc(r.engine),
 		Context:     ctx,
 		ExternalURL: r.alertURL,
-		Notifier:    notifier,
+		NotifyFunc:  sendAlerts(notifier, r.alertURL.String()),
 		Logger:      gklog.NewNopLogger(),
+		Registerer:  prometheus.DefaultRegisterer,
 	}
 	delay := 0 * time.Second // Unused, so 0 value is fine.
 	return rules.NewGroup("default", "none", delay, rs, opts), nil
 }
 
+// sendAlerts implements a the rules.NotifyFunc for a Notifier.
+// It filters any non-firing alerts from the input.
+//
+// Copied from Prometheus's main.go.
+func sendAlerts(n *notifier.Notifier, externalURL string) rules.NotifyFunc {
+	return func(ctx native_ctx.Context, expr string, alerts ...*rules.Alert) error {
+		var res []*notifier.Alert
+
+		for _, alert := range alerts {
+			// Only send actually firing alerts.
+			if alert.State == rules.StatePending {
+				continue
+			}
+			a := &notifier.Alert{
+				StartsAt:     alert.FiredAt,
+				Labels:       alert.Labels,
+				Annotations:  alert.Annotations,
+				GeneratorURL: externalURL + strutil.TableLinkForExpression(expr),
+			}
+			if !alert.ResolvedAt.IsZero() {
+				a.EndsAt = alert.ResolvedAt
+			}
+			res = append(res, a)
+		}
+
+		if len(alerts) > 0 {
+			n.Send(res...)
+		}
+		return nil
+	}
+}
+
 func (r *Ruler) getOrCreateNotifier(userID string) (*notifier.Notifier, error) {
 	r.notifiersMtx.Lock()
 	defer r.notifiersMtx.Unlock()
 
 	n, ok := r.notifiers[userID]
 	if ok {
-		return n, nil
+		return n.notifier, nil
 	}
 
-	n = notifier.New(&notifier.Options{
+	n = newRulerNotifier(&notifier.Options{
 		QueueCapacity: r.queueCapacity,
 		Do: func(ctx context.Context, client *http.Client, req *http.Request) (*http.Response, error) {
 			// Note: The passed-in context comes from the Prometheus rule group code
@@ -222,17 +314,18 @@ func (r *Ruler) getOrCreateNotifier(userID string) (*notifier.Notifier, error) {
 			}
 			return ctxhttp.Do(ctx, client, req)
 		},
-	}, gklog.NewNopLogger())
+	}, util.Logger)
+
+	go n.run()
 
 	// This should never fail, unless there's a programming mistake.
-	if err := n.ApplyConfig(r.notifierCfg); err != nil {
+	if err := n.applyConfig(r.notifierCfg); err != nil {
 		return nil, err
 	}
-	go n.Run()
 
 	// TODO: Remove notifiers for stale users. Right now this is a slow leak.
 	r.notifiers[userID] = n
-	return n, nil
+	return n.notifier, nil
 }
 
 // Evaluate a list of rules in the given context.
@@ -245,7 +338,7 @@ func (r *Ruler) Evaluate(ctx context.Context, rs []rules.Rule) {
 		level.Error(logger).Log("msg", "failed to create rule group", "err", err)
 		return
 	}
-	g.Eval(start)
+	g.Eval(ctx, start)
 
 	// The prometheus routines we're calling have their own instrumentation
 	// but, a) it's rule-based, not group-based, b) it's a summary, not a
@@ -260,7 +353,7 @@ func (r *Ruler) Stop() {
 	defer r.notifiersMtx.Unlock()
 
 	for _, n := range r.notifiers {
-		n.Stop()
+		n.stop()
 	}
 }