Refactor Ruler/Alertmanager API & Decouple configdb

Makefile

@@ -52,6 +52,8 @@ pkg/ring/ring.pb.go: pkg/ring/ring.proto
 pkg/querier/frontend/frontend.pb.go: pkg/querier/frontend/frontend.proto
 pkg/chunk/storage/caching_index_client.pb.go: pkg/chunk/storage/caching_index_client.proto
 pkg/distributor/ha_tracker.pb.go: pkg/distributor/ha_tracker.proto
+pkg/storage/rules/rules.pb.go: pkg/storage/rules/rules.proto
+pkg/util/usertracker/usertracker.pb.go: pkg/util/usertracker/usertracker.proto
 all: $(UPTODATE_FILES)
 test: protos
 mod-check: protos

pkg/alertmanager/api.go

@@ -0,0 +1,116 @@
+package alertmanager
+
+import (
+	"io/ioutil"
+	"net/http"
+
+	"github.com/cortexproject/cortex/pkg/storage/alerts"
+	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/go-kit/kit/log/level"
+	"github.com/gorilla/mux"
+	"github.com/weaveworks/common/user"
+	"gopkg.in/yaml.v2"
+)
+
+// RegisterRoutes registers the configs API HTTP routes with the provided Router.
+func (am *MultitenantAlertmanager) RegisterRoutes(r *mux.Router) {
+	// if no store is set return without regisering routes
+	if am.store == nil {
+		return
+	}
+	for _, route := range []struct {
+		name, method, path string
+		handler            http.HandlerFunc
+	}{
+		{"get_config", "GET", "/api/prom/alertmanager", am.getUserConfig},
+		{"set_config", "POST", "/api/prom/alertmanager", am.setUserConfig},
+		{"delete_config", "DELETE", "/api/prom/alertmanager", am.deleteUserConfig},
+	} {
+		r.Handle(route.path, route.handler).Methods(route.method).Name(route.name)
+	}
+}
+
+func (am *MultitenantAlertmanager) getUserConfig(w http.ResponseWriter, r *http.Request) {
+	userID, _, err := user.ExtractOrgIDFromHTTPRequest(r)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusUnauthorized)
+		return
+	}
+
+	logger := util.WithContext(r.Context(), util.Logger)
+	if err != nil {
+		level.Error(logger).Log("err", err.Error())
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	if userID == "" {
+		level.Error(logger).Log("err", err.Error())
+		http.Error(w, err.Error(), http.StatusUnauthorized)
+		return
+	}
+
+	cfg, err := am.store.GetAlertConfig(r.Context(), userID)
+
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	d, err := yaml.Marshal(&cfg)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/yaml")
+	if _, err := w.Write(d); err != nil {
+		level.Error(logger).Log("msg", "error marshalling yaml alertmanager config", "err", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+}
+
+func (am *MultitenantAlertmanager) setUserConfig(w http.ResponseWriter, r *http.Request) {
+	userID, _, err := user.ExtractOrgIDFromHTTPRequest(r)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusUnauthorized)
+		return
+	}
+
+	logger := util.WithContext(r.Context(), util.Logger)
+
+	if userID == "" {
+		level.Error(logger).Log("err", err.Error())
+		http.Error(w, err.Error(), http.StatusUnauthorized)
+		return
+	}
+
+	payload, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		level.Error(logger).Log("err", err.Error())
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	cfg := alerts.AlertConfig{}
+	err = yaml.Unmarshal(payload, &cfg)
+	if err != nil {
+		level.Error(logger).Log("err", err.Error())
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	err = am.store.SetAlertConfig(r.Context(), userID, cfg)
+	if err != nil {
+		level.Error(logger).Log("err", err.Error())
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	w.WriteHeader(http.StatusOK)
+}
+
+func (am *MultitenantAlertmanager) deleteUserConfig(w http.ResponseWriter, r *http.Request) {
+
+}

pkg/alertmanager/kv_poller.go

@@ -0,0 +1,101 @@
+package alertmanager
+
+import (
+	"context"
+
+	"github.com/cortexproject/cortex/pkg/storage/alerts"
+	"github.com/cortexproject/cortex/pkg/util/usertracker"
+)
+
+// trackedAlertPoller checks for updated user configs and
+// Retrieves the updated configuration from the backend
+type trackedAlertPoller struct {
+	tracker *usertracker.Tracker
+	store   alerts.AlertStore
+
+	initialized bool
+}
+
+func newTrackedAlertPoller(tracker *usertracker.Tracker, store alerts.AlertStore) (*trackedAlertPoller, error) {
+	return &trackedAlertPoller{
+		tracker: tracker,
+		store:   store,
+
+		initialized: false,
+	}, nil
+}
+
+func (p *trackedAlertPoller) trackedAlertStore() *trackedAlertStore {
+	return &trackedAlertStore{
+		tracker: p.tracker,
+		store:   p.store,
+	}
+}
+
+// PollAlerts returns the alerts changed since the last poll
+// All alert configurations are returned on the first poll
+func (p *trackedAlertPoller) PollAlerts(ctx context.Context) (map[string]alerts.AlertConfig, error) {
+	updatedConfigs := map[string]alerts.AlertConfig{}
+
+	// First poll will return all rule groups
+	if !p.initialized {
+		p.initialized = true
+		return p.store.ListAlertConfigs(ctx)
+	}
+
+	// Get the changed users from the user update tracker
+	users := p.tracker.GetUpdatedUsers(ctx)
+
+	// Retrieve user configuration from the rule store
+	// TODO: Add Retry logic for failed requests
+	// TODO: store users that were failed to be updated and reattempt to retrieve on the next poll
+	for _, u := range users {
+		cfg, err := p.store.GetAlertConfig(ctx, u)
+		if err != nil {
+			return nil, err
+		}
+
+		updatedConfigs[u] = cfg
+	}
+
+	return updatedConfigs, nil
+}
+
+func (p *trackedAlertPoller) Stop() {
+	p.tracker.Stop()
+}
+
+type trackedAlertStore struct {
+	tracker *usertracker.Tracker
+	store   alerts.AlertStore
+}
+
+// ListAlertConfigs passes through to the embedded alert store
+func (w *trackedAlertStore) ListAlertConfigs(ctx context.Context) (map[string]alerts.AlertConfig, error) {
+	return w.store.ListAlertConfigs(ctx)
+}
+
+// GetAlertConfig passes through to the embedded alert store
+func (w *trackedAlertStore) GetAlertConfig(ctx context.Context, id string) (alerts.AlertConfig, error) {
+	return w.store.GetAlertConfig(ctx, id)
+}
+
+// SetAlertConfig passes through to the embedded alert store, and tracks a user change
+func (w *trackedAlertStore) SetAlertConfig(ctx context.Context, id string, cfg alerts.AlertConfig) error {
+	err := w.store.SetAlertConfig(ctx, id, cfg)
+	if err != nil {
+		return err
+	}
+
+	return w.tracker.UpdateUser(ctx, id)
+}
+
+// DeleteAlertConfig passes through to the embedded alert store, and tracks a user change
+func (w *trackedAlertStore) DeleteAlertConfig(ctx context.Context, id string) error {
+	err := w.store.DeleteAlertConfig(ctx, id)
+	if err != nil {
+		return err
+	}
+
+	return w.tracker.UpdateUser(ctx, id)
+}

pkg/alertmanager/multitenant.go

@@ -22,8 +22,7 @@ import (
 	"github.com/weaveworks/common/user"
 	"github.com/weaveworks/mesh"
 
-	"github.com/cortexproject/cortex/pkg/configs"
-	configs_client "github.com/cortexproject/cortex/pkg/configs/client"
+	"github.com/cortexproject/cortex/pkg/storage/alerts"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
 )
@@ -185,10 +184,13 @@ type MultitenantAlertmanagerConfig struct {
 	FallbackConfigFile string
 	AutoWebhookRoot    string
 	AutoSlackRoot      string
+
+	AlertStore AlertStoreConfig
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet.
 func (cfg *MultitenantAlertmanagerConfig) RegisterFlags(f *flag.FlagSet) {
+	cfg.AlertStore.RegisterFlags(f)
 	flag.StringVar(&cfg.DataDir, "alertmanager.storage.path", "data/", "Base path for data storage.")
 	flag.DurationVar(&cfg.Retention, "alertmanager.storage.retention", 5*24*time.Hour, "How long to keep data for.")
 
@@ -214,22 +216,20 @@ func (cfg *MultitenantAlertmanagerConfig) RegisterFlags(f *flag.FlagSet) {
 type MultitenantAlertmanager struct {
 	cfg *MultitenantAlertmanagerConfig
 
-	configsAPI configs_client.Client
+	store  alerts.AlertStore
+	poller alerts.AlertPoller
 
 	// The fallback config is stored as a string and parsed every time it's needed
 	// because we mutate the parsed results and don't want those changes to take
 	// effect here.
 	fallbackConfig string
 
 	// All the organization configurations that we have. Only used for instrumentation.
-	cfgs map[string]configs.Config
+	cfgs map[string]alerts.AlertConfig
 
 	alertmanagersMtx sync.Mutex
 	alertmanagers    map[string]*Alertmanager
 
-	latestConfig configs.ID
-	latestMutex  sync.RWMutex
-
 	meshRouter   *gossipFactory
 	srvDiscovery *srvDiscovery
 
@@ -238,17 +238,12 @@ type MultitenantAlertmanager struct {
 }
 
 // NewMultitenantAlertmanager creates a new MultitenantAlertmanager.
-func NewMultitenantAlertmanager(cfg *MultitenantAlertmanagerConfig, cfgCfg configs_client.Config) (*MultitenantAlertmanager, error) {
+func NewMultitenantAlertmanager(cfg *MultitenantAlertmanagerConfig) (*MultitenantAlertmanager, error) {
 	err := os.MkdirAll(cfg.DataDir, 0777)
 	if err != nil {
 		return nil, fmt.Errorf("unable to create Alertmanager data directory %q: %s", cfg.DataDir, err)
 	}
 
-	configsAPI, err := configs_client.New(cfgCfg)
-	if err != nil {
-		return nil, err
-	}
-
 	mrouter := initMesh(cfg.MeshListenAddr, cfg.MeshHWAddr, cfg.MeshNickname, cfg.MeshPassword)
 	mrouter.Start()
 
@@ -265,11 +260,17 @@ func NewMultitenantAlertmanager(cfg *MultitenantAlertmanagerConfig, cfgCfg confi
 	}
 
 	gf := newGossipFactory(mrouter)
+
+	poller, store, err := NewAlertStore(cfg.AlertStore)
+	if err != nil {
+		return nil, fmt.Errorf("unable to create Alertmanager storage, %s", err)
+	}
+
 	am := &MultitenantAlertmanager{
 		cfg:            cfg,
-		configsAPI:     configsAPI,
+		poller:         poller,
+		store:          store,
 		fallbackConfig: string(fallbackConfig),
-		cfgs:           map[string]configs.Config{},
 		alertmanagers:  map[string]*Alertmanager{},
 		meshRouter:     &gf,
 		srvDiscovery:   newSRVDiscovery(cfg.MeshPeerService, cfg.MeshPeerHost, cfg.MeshPeerRefreshInterval),
@@ -314,6 +315,7 @@ func (am *MultitenantAlertmanager) Run() {
 
 // Stop stops the MultitenantAlertmanager.
 func (am *MultitenantAlertmanager) Stop() {
+	am.poller.Stop()
 	am.srvDiscovery.Stop()
 	close(am.stop)
 	<-am.done
@@ -326,7 +328,7 @@ func (am *MultitenantAlertmanager) Stop() {
 
 // Load the full set of configurations from the server, retrying with backoff
 // until we can get them.
-func (am *MultitenantAlertmanager) loadAllConfigs() map[string]configs.View {
+func (am *MultitenantAlertmanager) loadAllConfigs() map[string]alerts.AlertConfig {
 	backoff := util.NewBackoff(context.Background(), backoffConfig)
 	for {
 		cfgs, err := am.poll()
@@ -349,25 +351,21 @@ func (am *MultitenantAlertmanager) updateConfigs(now time.Time) error {
 }
 
 // poll the configuration server. Not re-entrant.
-func (am *MultitenantAlertmanager) poll() (map[string]configs.View, error) {
-	configID := am.latestConfig
-	cfgs, err := am.configsAPI.GetAlerts(context.Background(), configID)
+func (am *MultitenantAlertmanager) poll() (map[string]alerts.AlertConfig, error) {
+	cfgs, err := am.poller.PollAlerts(context.Background())
 	if err != nil {
 		level.Warn(util.Logger).Log("msg", "MultitenantAlertmanager: configs server poll failed", "err", err)
 		return nil, err
 	}
-	am.latestMutex.Lock()
-	am.latestConfig = cfgs.GetLatestConfigID()
-	am.latestMutex.Unlock()
-	return cfgs.Configs, nil
+	return cfgs, nil
 }
 
-func (am *MultitenantAlertmanager) addNewConfigs(cfgs map[string]configs.View) {
+func (am *MultitenantAlertmanager) addNewConfigs(cfgs map[string]alerts.AlertConfig) {
 	// 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 {
 
-		err := am.setConfig(userID, config.Config)
+		err := am.setConfig(userID, config)
 		if err != nil {
 			level.Warn(util.Logger).Log("msg", "MultitenantAlertmanager: error applying config", "err", err)
 			continue
@@ -426,7 +424,7 @@ func (am *MultitenantAlertmanager) createTemplatesFile(userID, fn, content strin
 
 // setConfig applies the given configuration to the alertmanager for `userID`,
 // creating an alertmanager if it doesn't already exist.
-func (am *MultitenantAlertmanager) setConfig(userID string, config configs.Config) error {
+func (am *MultitenantAlertmanager) setConfig(userID string, config alerts.AlertConfig) error {
 	_, hasExisting := am.alertmanagers[userID]
 	var amConfig *amconfig.Config
 	var err error
@@ -489,7 +487,7 @@ func (am *MultitenantAlertmanager) setConfig(userID string, config configs.Confi
 }
 
 // alertmanagerConfigFromConfig returns the Alertmanager config from the Cortex configuration.
-func alertmanagerConfigFromConfig(c configs.Config) (*amconfig.Config, error) {
+func alertmanagerConfigFromConfig(c alerts.AlertConfig) (*amconfig.Config, error) {
 	cfg, err := amconfig.Load(c.AlertmanagerConfig)
 	if err != nil {
 		return nil, fmt.Errorf("error parsing Alertmanager config: %s", err)