Use rules DB directly from ruler

Will fall back to using configs service if no DB information supplied.

Author: jml

cmd/lite/main.go

@@ -42,6 +42,7 @@ func main() {
 		chunkStoreConfig  chunk.StoreConfig
 		distributorConfig distributor.Config
 		ingesterConfig    ingester.Config
+		configStoreConfig ruler.ConfigStoreConfig
 		rulerConfig       ruler.Config
 		schemaConfig      chunk.SchemaConfig
 		storageConfig     storage.Config
@@ -52,7 +53,7 @@ func main() {
 	// Ingester needs to know our gRPC listen port.
 	ingesterConfig.ListenPort = &serverConfig.GRPCListenPort
 	util.RegisterFlags(&serverConfig, &chunkStoreConfig, &distributorConfig,
-		&ingesterConfig, &rulerConfig, &storageConfig, &schemaConfig, &logLevel)
+		&ingesterConfig, &configStoreConfig, &rulerConfig, &storageConfig, &schemaConfig, &logLevel)
 	flag.BoolVar(&unauthenticated, "unauthenticated", false, "Set to true to disable multitenancy.")
 	flag.Parse()
 	schemaConfig.MaxChunkAge = ingesterConfig.MaxChunkAge
@@ -122,15 +123,20 @@ func main() {
 	tableManager.Start()
 	defer tableManager.Stop()
 
-	if rulerConfig.ConfigsAPIURL.String() != "" {
+	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)
 		if err != nil {
 			level.Error(util.Logger).Log("msg", "error initializing ruler", "err", err)
 			os.Exit(1)
 		}
 		defer rlr.Stop()
 
-		rulerServer, err := ruler.NewServer(rulerConfig, rlr)
+		rulerServer, err := ruler.NewServer(rulerConfig, rlr, rulesAPI)
 		if err != nil {
 			level.Error(util.Logger).Log("msg", "error initializing ruler server", "err", err)
 			os.Exit(1)

cmd/ruler/main.go

@@ -32,10 +32,11 @@ func main() {
 		chunkStoreConfig  chunk.StoreConfig
 		schemaConfig      chunk.SchemaConfig
 		storageConfig     storage.Config
+		configStoreConfig ruler.ConfigStoreConfig
 		logLevel          util.LogLevel
 	)
 	util.RegisterFlags(&serverConfig, &ringConfig, &distributorConfig,
-		&rulerConfig, &chunkStoreConfig, &storageConfig, &schemaConfig, &logLevel)
+		&rulerConfig, &chunkStoreConfig, &storageConfig, &schemaConfig, &configStoreConfig, &logLevel)
 	flag.Parse()
 
 	util.InitLogger(logLevel.AllowedLevel)
@@ -75,7 +76,13 @@ func main() {
 	}
 	defer rlr.Stop()
 
-	rulerServer, err := ruler.NewServer(rulerConfig, rlr)
+	rulesAPI, err := ruler.NewRulesAPI(configStoreConfig)
+	if err != nil {
+		level.Error(util.Logger).Log("msg", "error initializing rules API", "err", err)
+		os.Exit(1)
+	}
+
+	rulerServer, err := ruler.NewServer(rulerConfig, rlr, rulesAPI)
 	if err != nil {
 		level.Error(util.Logger).Log("msg", "error initializing ruler server: %v", err)
 		os.Exit(1)

pkg/configs/client/configs.go

@@ -85,7 +85,8 @@ func AlertmanagerConfigFromConfig(c configs.Config) (*config.Config, error) {
 	return cfg, nil
 }
 
-func getConfigs(endpoint string, timeout time.Duration, since configs.ID) (*ConfigsResponse, error) {
+// GetConfigs gets configurations from the configs server.
+func GetConfigs(endpoint string, timeout time.Duration, since configs.ID) (*ConfigsResponse, error) {
 	req, err := http.NewRequest("GET", endpoint, nil)
 	if err != nil {
 		return nil, err
@@ -120,7 +121,7 @@ func (c *AlertManagerConfigsAPI) GetConfigs(since configs.ID) (*ConfigsResponse,
 		suffix = fmt.Sprintf("?since=%d", since)
 	}
 	endpoint := fmt.Sprintf("%s/private/api/prom/configs/alertmanager%s", c.URL.String(), suffix)
-	return getConfigs(endpoint, c.Timeout, since)
+	return GetConfigs(endpoint, c.Timeout, since)
 }
 
 // RulesAPI allows retrieving recording and alerting rules.
@@ -137,5 +138,5 @@ func (c *RulesAPI) GetConfigs(since configs.ID) (*ConfigsResponse, error) {
 		suffix = fmt.Sprintf("?since=%d", since)
 	}
 	endpoint := fmt.Sprintf("%s/private/api/prom/configs/rules%s", c.URL.String(), suffix)
-	return getConfigs(endpoint, c.Timeout, since)
+	return GetConfigs(endpoint, c.Timeout, since)
 }

pkg/configs/configs.go

@@ -1,5 +1,13 @@
 package configs
 
+import (
+	"fmt"
+
+	"github.com/prometheus/prometheus/promql"
+	"github.com/prometheus/prometheus/rules"
+	"github.com/weaveworks/cortex/pkg/util"
+)
+
 // An ID is the ID of a single users's Cortex configuration. When a
 // configuration changes, it gets a new ID.
 type ID int
@@ -33,6 +41,36 @@ func (v View) GetVersionedRulesConfig() VersionedRulesConfig {
 // RulesConfig are the set of rules files for a particular organization.
 type RulesConfig map[string]string
 
+// Parse rules from the Cortex configuration.
+//
+// Strongly inspired by `loadGroups` in Prometheus.
+func (c RulesConfig) Parse() ([]rules.Rule, error) {
+	result := []rules.Rule{}
+	for fn, content := range c {
+		stmts, err := promql.ParseStmts(content)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing %s: %s", fn, err)
+		}
+
+		for _, stmt := range stmts {
+			var rule rules.Rule
+
+			switch r := stmt.(type) {
+			case *promql.AlertStmt:
+				rule = rules.NewAlertingRule(r.Name, r.Expr, r.Duration, r.Labels, r.Annotations, util.Logger)
+
+			case *promql.RecordStmt:
+				rule = rules.NewRecordingRule(r.Name, r.Expr, r.Labels)
+
+			default:
+				return nil, fmt.Errorf("ruler.GetRules: unknown statement type")
+			}
+			result = append(result, rule)
+		}
+	}
+	return result, nil
+}
+
 // VersionedRulesConfig is a RulesConfig together with a version.
 // `data Versioned a = Versioned { id :: ID , config :: a }`
 type VersionedRulesConfig struct {

pkg/configs/db/db.go

@@ -69,3 +69,12 @@ func New(cfg Config) (DB, error) {
 	}
 	return traced{timed{d}}, nil
 }
+
+// NewRulesDB creates a new rules config database.
+func NewRulesDB(cfg Config) (RulesDB, error) {
+	db, err := New(cfg)
+	if err != nil {
+		return nil, err
+	}
+	return db, err
+}

pkg/ruler/configs.go

@@ -0,0 +1,105 @@
+package ruler
+
+import (
+	"flag"
+	"fmt"
+	"net/url"
+	"time"
+
+	"github.com/weaveworks/cortex/pkg/configs"
+	configs_client "github.com/weaveworks/cortex/pkg/configs/client"
+	"github.com/weaveworks/cortex/pkg/configs/db"
+	"github.com/weaveworks/cortex/pkg/util"
+)
+
+// ConfigStoreConfig says where we can find the ruler configs.
+type ConfigStoreConfig struct {
+	DBConfig db.Config
+
+	// DEPRECATED
+	ConfigsAPIURL util.URLValue
+
+	// DEPRECATED. HTTP timeout duration for requests made to the Weave Cloud
+	// configs service.
+	ClientTimeout time.Duration
+}
+
+// RegisterFlags adds the flags required to config this to the given FlagSet
+func (cfg *ConfigStoreConfig) RegisterFlags(f *flag.FlagSet) {
+	cfg.DBConfig.RegisterFlags(f)
+	f.Var(&cfg.ConfigsAPIURL, "ruler.configs.url", "DEPRECATED. URL of configs API server.")
+	f.DurationVar(&cfg.ClientTimeout, "ruler.client-timeout", 5*time.Second, "DEPRECATED. Timeout for requests to Weave Cloud configs service.")
+}
+
+// RulesAPI is what the ruler needs from a config store to process rules.
+type RulesAPI interface {
+	// GetConfigs returns all Cortex configurations from a configs API server
+	// that have been updated after the given configs.ID was last updated.
+	GetConfigs(since configs.ID) (map[string]configs.VersionedRulesConfig, error)
+}
+
+// NewRulesAPI creates a new RulesAPI.
+func NewRulesAPI(cfg ConfigStoreConfig) (RulesAPI, error) {
+	// All of this falderal is to allow for a smooth transition away from
+	// using the configs server and toward directly connecting to the database.
+	// See https://github.com/weaveworks/cortex/issues/619
+	if cfg.DBConfig.URI == "" {
+		return configsClient{
+			URL:     cfg.ConfigsAPIURL.URL,
+			Timeout: cfg.ClientTimeout,
+		}, nil
+	}
+	db, err := db.NewRulesDB(cfg.DBConfig)
+	if err != nil {
+		return nil, err
+	}
+	return dbStore{db: db}, nil
+}
+
+// configsClient allows retrieving recording and alerting rules from the configs server.
+type configsClient struct {
+	URL     *url.URL
+	Timeout time.Duration
+}
+
+// GetConfigs implements RulesAPI.
+func (c configsClient) GetConfigs(since configs.ID) (map[string]configs.VersionedRulesConfig, error) {
+	suffix := ""
+	if since != 0 {
+		suffix = fmt.Sprintf("?since=%d", since)
+	}
+	endpoint := fmt.Sprintf("%s/private/api/prom/configs/rules%s", c.URL.String(), suffix)
+	response, err := configs_client.GetConfigs(endpoint, c.Timeout, since)
+	if err != nil {
+		return nil, err
+	}
+	configs := map[string]configs.VersionedRulesConfig{}
+	for id, view := range response.Configs {
+		configs[id] = view.GetVersionedRulesConfig()
+	}
+	return configs, nil
+}
+
+type dbStore struct {
+	db db.RulesDB
+}
+
+// GetConfigs implements RulesAPI.
+func (d dbStore) GetConfigs(since configs.ID) (map[string]configs.VersionedRulesConfig, error) {
+	if since == 0 {
+		return d.db.GetAllRulesConfigs()
+	}
+	return d.db.GetRulesConfigs(since)
+}
+
+// getLatestConfigID gets the latest configs ID.
+// max [latest, max (map getID cfgs)]
+func getLatestConfigID(cfgs map[string]configs.VersionedRulesConfig, latest configs.ID) configs.ID {
+	ret := latest
+	for _, config := range cfgs {
+		if config.ID > ret {
+			ret = config.ID
+		}
+	}
+	return ret
+}

pkg/ruler/ruler.go

@@ -22,7 +22,6 @@ import (
 
 	"github.com/weaveworks/common/user"
 	"github.com/weaveworks/cortex/pkg/chunk"
-	configs "github.com/weaveworks/cortex/pkg/configs/client"
 	"github.com/weaveworks/cortex/pkg/distributor"
 	"github.com/weaveworks/cortex/pkg/querier"
 	"github.com/weaveworks/cortex/pkg/util"
@@ -54,12 +53,6 @@ func init() {
 
 // Config is the configuration for the recording rules server.
 type Config struct {
-	ConfigsAPIURL util.URLValue
-
-	// HTTP timeout duration for requests made to the Weave Cloud configs
-	// service.
-	ClientTimeout time.Duration
-
 	// This is used for template expansion in alerts; must be a valid URL
 	ExternalURL util.URLValue
 
@@ -84,10 +77,8 @@ type Config struct {
 // RegisterFlags adds the flags required to config this to the given FlagSet
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	cfg.ExternalURL.URL, _ = url.Parse("") // Must be non-nil
-	f.Var(&cfg.ConfigsAPIURL, "ruler.configs.url", "URL of configs API server.")
 	f.Var(&cfg.ExternalURL, "ruler.external.url", "URL of alerts return path.")
 	f.DurationVar(&cfg.EvaluationInterval, "ruler.evaluation-interval", 15*time.Second, "How frequently to evaluate rules")
-	f.DurationVar(&cfg.ClientTimeout, "ruler.client-timeout", 5*time.Second, "Timeout for requests to Weave Cloud configs service.")
 	f.IntVar(&cfg.NumWorkers, "ruler.num-workers", 1, "Number of rule evaluator worker routines in this process")
 	f.Var(&cfg.AlertmanagerURL, "ruler.alertmanager-url", "URL of the Alertmanager to send notifications to.")
 	f.BoolVar(&cfg.AlertmanagerDiscovery, "ruler.alertmanager-discovery", false, "Use DNS SRV records to discover alertmanager hosts.")
@@ -280,13 +271,9 @@ type Server struct {
 }
 
 // NewServer makes a new rule processing server.
-func NewServer(cfg Config, ruler *Ruler) (*Server, error) {
-	c := configs.RulesAPI{
-		URL:     cfg.ConfigsAPIURL.URL,
-		Timeout: cfg.ClientTimeout,
-	}
+func NewServer(cfg Config, ruler *Ruler, rulesAPI RulesAPI) (*Server, error) {
 	// TODO: Separate configuration for polling interval.
-	s := newScheduler(c, cfg.EvaluationInterval, cfg.EvaluationInterval)
+	s := newScheduler(rulesAPI, cfg.EvaluationInterval, cfg.EvaluationInterval)
 	if cfg.NumWorkers <= 0 {
 		return nil, fmt.Errorf("must have at least 1 worker, got %d", cfg.NumWorkers)
 	}

pkg/ruler/scheduler.go

@@ -14,7 +14,6 @@ import (
 
 	"github.com/weaveworks/common/instrument"
 	"github.com/weaveworks/cortex/pkg/configs"
-	configs_client "github.com/weaveworks/cortex/pkg/configs/client"
 	"github.com/weaveworks/cortex/pkg/util"
 )
 
@@ -79,7 +78,7 @@ func (w workItem) String() string {
 }
 
 type scheduler struct {
-	configsAPI         configs_client.RulesAPI
+	rulesAPI           RulesAPI
 	evaluationInterval time.Duration // how often we re-evaluate each rule set
 	q                  *SchedulingQueue
 
@@ -94,9 +93,9 @@ type scheduler struct {
 }
 
 // newScheduler makes a new scheduler.
-func newScheduler(configsAPI configs_client.RulesAPI, evaluationInterval, pollInterval time.Duration) scheduler {
+func newScheduler(rulesAPI RulesAPI, evaluationInterval, pollInterval time.Duration) scheduler {
 	return scheduler{
-		configsAPI:         configsAPI,
+		rulesAPI:           rulesAPI,
 		evaluationInterval: evaluationInterval,
 		pollInterval:       pollInterval,
 		q:                  NewSchedulingQueue(clockwork.NewRealClock()),
@@ -137,7 +136,7 @@ func (s *scheduler) Stop() {
 
 // Load the full set of configurations from the server, retrying with backoff
 // until we can get them.
-func (s *scheduler) loadAllConfigs() map[string]configs.View {
+func (s *scheduler) loadAllConfigs() map[string]configs.VersionedRulesConfig {
 	backoff := util.NewBackoff(context.Background(), backoffConfig)
 	for {
 		cfgs, err := s.poll()
@@ -160,31 +159,31 @@ func (s *scheduler) updateConfigs(now time.Time) error {
 }
 
 // poll the configuration server. Not re-entrant.
-func (s *scheduler) poll() (map[string]configs.View, error) {
+func (s *scheduler) poll() (map[string]configs.VersionedRulesConfig, error) {
 	s.Lock()
 	configID := s.latestConfig
 	s.Unlock()
-	var cfgs *configs_client.ConfigsResponse
+	var cfgs map[string]configs.VersionedRulesConfig
 	err := instrument.TimeRequestHistogram(context.Background(), "Configs.GetConfigs", configsRequestDuration, func(_ context.Context) error {
 		var err error
-		cfgs, err = s.configsAPI.GetConfigs(configID)
+		cfgs, err = s.rulesAPI.GetConfigs(configID)
 		return err
 	})
 	if err != nil {
 		level.Warn(util.Logger).Log("msg", "scheduler: configs server poll failed", "err", err)
 		return nil, err
 	}
 	s.Lock()
-	s.latestConfig = cfgs.GetLatestConfigID()
+	s.latestConfig = getLatestConfigID(cfgs, configID)
 	s.Unlock()
-	return cfgs.Configs, nil
+	return cfgs, nil
 }
 
-func (s *scheduler) addNewConfigs(now time.Time, cfgs map[string]configs.View) {
+func (s *scheduler) addNewConfigs(now time.Time, cfgs map[string]configs.VersionedRulesConfig) {
 	// TODO: instrument how many configs we have, both valid & invalid.
 	level.Debug(util.Logger).Log("msg", "adding configurations", "num_configs", len(cfgs))
 	for userID, config := range cfgs {
-		rules, err := configs_client.RulesFromConfig(config.Config)
+		rules, err := config.Config.Parse()
 		if err != nil {
 			// XXX: This means that if a user has a working configuration and
 			// they submit a broken one, we'll keep processing the last known