Add endpoints to deactivate and restore config

 - call deactivate on instance deletion

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,58 @@ 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)
+
+	fmt.Printf("deactivateConfigs org ID %s\n", userID)
+
+	if _, err := a.db.GetConfig(userID); err != nil {
+		if err == sql.ErrNoRows {
+			http.Error(w, "No configuration", http.StatusNotFound)
+			return
+		}
+		level.Error(logger).Log("msg", "error getting config", "err", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	if err := a.db.DeactivateConfig(userID); err != nil {
+		level.Error(logger).Log("msg", "error deactivating configs", "err", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusAccepted)
+}
+
+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)
+	fmt.Printf("restore org ID %s\n", userID)
+
+	if _, err := a.db.GetConfig(userID); err != nil {
+		if err == sql.ErrNoRows {
+			http.Error(w, "No configuration", http.StatusNotFound)
+			return
+		}
+		level.Error(logger).Log("msg", "error getting config", "err", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	if err := a.db.RestoreConfig(userID); err != nil {
+		level.Error(logger).Log("msg", "error restoring config", "err", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusCreated)
+}

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
 }
 
 // GetVersionedRulesConfig specializes the view to just the rules config.

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,29 @@ func (d *DB) GetConfigs(since configs.ID) (map[string]configs.View, error) {
 	return cfgs, nil
 }
 
+// SetDeletedAtConfig sets deletedAt for last configuration for a user.
+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,6 +20,8 @@ const (
 	// schema so this isn't needed.
 	entityType = "org"
 	subsystem  = "cortex"
+	// timeout waiting for database connection to be established
+	dbTimeout = 5 * time.Minute
 )
 
 var (
@@ -42,8 +45,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 +80,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),
@@ -225,6 +252,57 @@ func (d DB) GetRulesConfigs(since configs.ID) (map[string]configs.VersionedRules
 	})
 }
 
+// GetLastConfig gets a last configuration (active or deleted).
+func (d DB) GetLastConfig(userID string) (configs.View, error) {
+	var cfgView configs.View
+	var cfgBytes []byte
+	err := d.Select("id", "config").
+		From("configs").
+		Where(squirrel.Eq{"owner_id": userID}).
+		OrderBy("id DESC").
+		Limit(1).
+		QueryRow().Scan(&cfgView.ID, &cfgBytes)
+	if err != nil {
+		return cfgView, err
+	}
+	err = json.Unmarshal(cfgBytes, &cfgView.Config)
+	log.Infof("get last config bytes: %s", cfgBytes)
+	return cfgView, err
+}
+
+// SetDeletedAtConfig sets a deletedAt for configuration.
+func (d DB) SetDeletedAtConfig(userID string, deletedAt time.Time, cfg configs.Config) error {
+	cfgBytes, err := json.Marshal(cfg)
+	if err != nil {
+		return err
+	}
+
+	log.Infof("set deletedA to %s for %s", deletedAt, cfgBytes)
+	_, 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.GetLastConfig(userID)
+	if err != nil {
+		return err
+	}
+	return d.SetDeletedAtConfig(userID, time.Now(), cfg.Config)
+}
+
+// RestoreConfig restores deactivated configuration.
+func (d DB) RestoreConfig(userID string) error {
+	cfg, err := d.GetLastConfig(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()