Merge pull request #629 from weaveworks/cleanup

Add endpoints to deactivate and restore config

Author: lelenanam

pkg/configs/api/api.go

@@ -61,6 +61,8 @@ func (a *API) RegisterRoutes(r *mux.Router) {
 		{"get_alertmanager_config", "GET", "/api/prom/configs/alertmanager", a.getConfig},
 		{"set_alertmanager_config", "POST", "/api/prom/configs/alertmanager", a.setConfig},
 		{"validate_alertmanager_config", "POST", "/api/prom/configs/alertmanager/validate", a.validateAlertmanagerConfig},
+		{"deactivate_config", "DELETE", "/api/prom/configs/deactivate", a.deactivateConfig},
+		{"restore_config", "POST", "/api/prom/configs/restore", a.restoreConfig},
 		// Internal APIs.
 		{"private_get_rules", "GET", "/private/api/prom/configs/rules", a.getConfigs},
 		{"private_get_alertmanager_config", "GET", "/private/api/prom/configs/alertmanager", a.getConfigs},
@@ -217,3 +219,47 @@ func (a *API) getConfigs(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 }
+
+func (a *API) deactivateConfig(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 := a.db.DeactivateConfig(userID); err != nil {
+		if err == sql.ErrNoRows {
+			level.Info(logger).Log("msg", "deactivate config - no configuration", "userID", userID)
+			http.Error(w, "No configuration", http.StatusNotFound)
+			return
+		}
+		level.Error(logger).Log("msg", "error deactivating config", "err", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	level.Info(logger).Log("msg", "config deactivated", "userID", userID)
+	w.WriteHeader(http.StatusOK)
+}
+
+func (a *API) restoreConfig(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 := a.db.RestoreConfig(userID); err != nil {
+		if err == sql.ErrNoRows {
+			level.Info(logger).Log("msg", "restore config - no configuration", "userID", userID)
+			http.Error(w, "No configuration", http.StatusNotFound)
+			return
+		}
+		level.Error(logger).Log("msg", "error restoring config", "err", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	level.Info(logger).Log("msg", "config restored", "userID", userID)
+	w.WriteHeader(http.StatusOK)
+}

pkg/configs/configs.go

@@ -2,6 +2,7 @@ package configs
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/rules"
@@ -26,8 +27,9 @@ type Config struct {
 // _version_ of a configuration a unique ID and guarantees that later versions
 // have greater IDs.
 type View struct {
-	ID     ID     `json:"id"`
-	Config Config `json:"config"`
+	ID        ID        `json:"id"`
+	Config    Config    `json:"config"`
+	DeletedAt time.Time `json:"deleted_at"`
 }
 
 // GetVersionedRulesConfig specializes the view to just the rules config.
@@ -36,8 +38,9 @@ func (v View) GetVersionedRulesConfig() *VersionedRulesConfig {
 		return nil
 	}
 	return &VersionedRulesConfig{
-		ID:     v.ID,
-		Config: v.Config.RulesFiles,
+		ID:        v.ID,
+		Config:    v.Config.RulesFiles,
+		DeletedAt: v.DeletedAt,
 	}
 }
 
@@ -93,6 +96,7 @@ func (c RulesConfig) Parse() ([]rules.Rule, error) {
 // VersionedRulesConfig is a RulesConfig together with a version.
 // `data Versioned a = Versioned { id :: ID , config :: a }`
 type VersionedRulesConfig struct {
-	ID     ID          `json:"id"`
-	Config RulesConfig `json:"config"`
+	ID        ID          `json:"id"`
+	Config    RulesConfig `json:"config"`
+	DeletedAt time.Time   `json:"deleted_at"`
 }

pkg/configs/db/db.go

@@ -48,6 +48,9 @@ type DB interface {
 	GetAllConfigs() (map[string]configs.View, error)
 	GetConfigs(since configs.ID) (map[string]configs.View, error)
 
+	DeactivateConfig(userID string) error
+	RestoreConfig(userID string) error
+
 	Close() error
 }
 

pkg/configs/db/memory/memory.go

@@ -2,6 +2,7 @@ package memory
 
 import (
 	"database/sql"
+	"time"
 
 	"github.com/weaveworks/cortex/pkg/configs"
 )
@@ -52,6 +53,31 @@ func (d *DB) GetConfigs(since configs.ID) (map[string]configs.View, error) {
 	return cfgs, nil
 }
 
+// SetDeletedAtConfig sets a deletedAt for configuration
+// by adding a single new row with deleted_at set
+// the same as SetConfig is actually insert
+func (d *DB) SetDeletedAtConfig(userID string, deletedAt time.Time) error {
+	cv, err := d.GetConfig(userID)
+	if err != nil {
+		return err
+	}
+	cv.DeletedAt = deletedAt
+	cv.ID = configs.ID(d.id)
+	d.cfgs[userID] = cv
+	d.id++
+	return nil
+}
+
+// DeactivateConfig deactivates configuration for a user by creating new configuration with DeletedAt set to now
+func (d *DB) DeactivateConfig(userID string) error {
+	return d.SetDeletedAtConfig(userID, time.Now())
+}
+
+// RestoreConfig restores deactivated configuration for a user by creating new configuration with empty DeletedAt
+func (d *DB) RestoreConfig(userID string) error {
+	return d.SetDeletedAtConfig(userID, time.Time{})
+}
+
 // Close finishes using the db. Noop.
 func (d *DB) Close() error {
 	return nil

pkg/configs/db/postgres/postgres.go

@@ -3,13 +3,14 @@ package postgres
 import (
 	"database/sql"
 	"encoding/json"
-	"errors"
+	"time"
 
 	"github.com/Masterminds/squirrel"
 	"github.com/go-kit/kit/log/level"
 	_ "github.com/lib/pq"                         // Import the postgres sql driver
 	_ "github.com/mattes/migrate/driver/postgres" // Import the postgres migrations driver
 	"github.com/mattes/migrate/migrate"
+	"github.com/pkg/errors"
 	"github.com/weaveworks/cortex/pkg/configs"
 	"github.com/weaveworks/cortex/pkg/util"
 )
@@ -19,11 +20,12 @@ const (
 	// schema so this isn't needed.
 	entityType = "org"
 	subsystem  = "cortex"
+	// timeout waiting for database connection to be established
+	dbTimeout = 5 * time.Minute
 )
 
 var (
-	activeConfig = squirrel.Eq{
-		"deleted_at": nil,
+	allConfigs = squirrel.Eq{
 		"owner_type": entityType,
 		"subsystem":  subsystem,
 	}
@@ -42,8 +44,32 @@ type dbProxy interface {
 	Prepare(query string) (*sql.Stmt, error)
 }
 
+// dbWait waits for database connection to be established
+func dbWait(db *sql.DB) error {
+	deadline := time.Now().Add(dbTimeout)
+	var err error
+	for tries := 0; time.Now().Before(deadline); tries++ {
+		err = db.Ping()
+		if err == nil {
+			return nil
+		}
+		level.Warn(util.Logger).Log("msg", "db connection not established, retrying...", "error", err)
+		time.Sleep(time.Second << uint(tries))
+	}
+	return errors.Wrapf(err, "db connection not established after %s", dbTimeout)
+}
+
 // New creates a new postgres DB
 func New(uri, migrationsDir string) (DB, error) {
+	db, err := sql.Open("postgres", uri)
+	if err != nil {
+		return DB{}, errors.Wrap(err, "cannot open postgres db")
+	}
+
+	if err := dbWait(db); err != nil {
+		return DB{}, errors.Wrap(err, "cannot establish db connection")
+	}
+
 	if migrationsDir != "" {
 		level.Info(util.Logger).Log("msg", "running database migrations...")
 		if errs, ok := migrate.UpSync(uri, migrationsDir); !ok {
@@ -53,7 +79,7 @@ func New(uri, migrationsDir string) (DB, error) {
 			return DB{}, errors.New("database migrations failed")
 		}
 	}
-	db, err := sql.Open("postgres", uri)
+
 	return DB{
 		dbProxy:              db,
 		StatementBuilderType: statementBuilder(db),
@@ -97,7 +123,7 @@ func (d DB) GetConfig(userID string) (configs.View, error) {
 	var cfgBytes []byte
 	err := d.Select("id", "config").
 		From("configs").
-		Where(squirrel.And{activeConfig, squirrel.Eq{"owner_id": userID}}).
+		Where(squirrel.And{allConfigs, squirrel.Eq{"owner_id": userID}}).
 		OrderBy("id DESC").
 		Limit(1).
 		QueryRow().Scan(&cfgView.ID, &cfgBytes)
@@ -123,13 +149,13 @@ func (d DB) SetConfig(userID string, cfg configs.Config) error {
 
 // GetAllConfigs gets all of the configs.
 func (d DB) GetAllConfigs() (map[string]configs.View, error) {
-	return d.findConfigs(activeConfig)
+	return d.findConfigs(allConfigs)
 }
 
 // GetConfigs gets all of the configs that have changed recently.
 func (d DB) GetConfigs(since configs.ID) (map[string]configs.View, error) {
 	return d.findConfigs(squirrel.And{
-		activeConfig,
+		allConfigs,
 		squirrel.Gt{"id": since},
 	})
 }
@@ -214,17 +240,50 @@ func (d DB) findRulesConfigs(filter squirrel.Sqlizer) (map[string]configs.Versio
 
 // GetAllRulesConfigs gets all alertmanager configs for all users.
 func (d DB) GetAllRulesConfigs() (map[string]configs.VersionedRulesConfig, error) {
-	return d.findRulesConfigs(activeConfig)
+	return d.findRulesConfigs(allConfigs)
 }
 
 // GetRulesConfigs gets all the alertmanager configs that have changed since a given config.
 func (d DB) GetRulesConfigs(since configs.ID) (map[string]configs.VersionedRulesConfig, error) {
 	return d.findRulesConfigs(squirrel.And{
-		activeConfig,
+		allConfigs,
 		squirrel.Gt{"id": since},
 	})
 }
 
+// SetDeletedAtConfig sets a deletedAt for configuration
+// by adding a single new row with deleted_at set
+// the same as SetConfig is actually insert
+func (d DB) SetDeletedAtConfig(userID string, deletedAt time.Time, cfg configs.Config) error {
+	cfgBytes, err := json.Marshal(cfg)
+	if err != nil {
+		return err
+	}
+	_, err = d.Insert("configs").
+		Columns("owner_id", "owner_type", "subsystem", "deleted_at", "config").
+		Values(userID, entityType, subsystem, deletedAt, cfgBytes).
+		Exec()
+	return err
+}
+
+// DeactivateConfig deactivates a configuration.
+func (d DB) DeactivateConfig(userID string) error {
+	cfg, err := d.GetConfig(userID)
+	if err != nil {
+		return err
+	}
+	return d.SetDeletedAtConfig(userID, time.Now(), cfg.Config)
+}
+
+// RestoreConfig restores configuration.
+func (d DB) RestoreConfig(userID string) error {
+	cfg, err := d.GetConfig(userID)
+	if err != nil {
+		return err
+	}
+	return d.SetDeletedAtConfig(userID, time.Time{}, cfg.Config)
+}
+
 // Transaction runs the given function in a postgres transaction. If fn returns
 // an error the txn will be rolled back.
 func (d DB) Transaction(f func(DB) error) error {

pkg/configs/db/timed.go

@@ -69,6 +69,18 @@ func (t timed) GetConfigs(since configs.ID) (cfgs map[string]configs.View, err e
 	return
 }
 
+func (t timed) DeactivateConfig(userID string) (err error) {
+	return t.timeRequest("DeactivateConfig", func(_ context.Context) error {
+		return t.d.DeactivateConfig(userID)
+	})
+}
+
+func (t timed) RestoreConfig(userID string) (err error) {
+	return t.timeRequest("RestoreConfig", func(_ context.Context) error {
+		return t.d.RestoreConfig(userID)
+	})
+}
+
 func (t timed) Close() error {
 	return t.timeRequest("Close", func(_ context.Context) error {
 		return t.d.Close()

pkg/configs/db/traced.go

@@ -37,6 +37,16 @@ func (t traced) GetConfigs(since configs.ID) (cfgs map[string]configs.View, err
 	return t.d.GetConfigs(since)
 }
 
+func (t traced) DeactivateConfig(userID string) (err error) {
+	defer func() { t.trace("DeactivateConfig", userID, err) }()
+	return t.d.DeactivateConfig(userID)
+}
+
+func (t traced) RestoreConfig(userID string) (err error) {
+	defer func() { t.trace("RestoreConfig", userID, err) }()
+	return t.d.RestoreConfig(userID)
+}
+
 func (t traced) Close() (err error) {
 	defer func() { t.trace("Close", err) }()
 	return t.d.Close()

pkg/ruler/scheduler.go

@@ -192,10 +192,14 @@ func (s *scheduler) addNewConfigs(now time.Time, cfgs map[string]configs.Version
 			level.Warn(util.Logger).Log("msg", "scheduler: invalid Cortex configuration", "user_id", userID, "err", err)
 			continue
 		}
-
 		level.Info(util.Logger).Log("msg", "scheduler: updating rules for user", "user_id", userID, "num_rules", len(rules))
 		s.Lock()
-		s.cfgs[userID] = rules
+		// if deleted remove from map
+		if !config.DeletedAt.IsZero() {
+			delete(s.cfgs, userID)
+		} else {
+			s.cfgs[userID] = rules
+		}
 		s.Unlock()
 		s.addWorkItem(workItem{userID, rules, now})
 	}