Repository: cortexproject/cortex (go)

Issue:
support list all users alertmanager configs and rule groups
**Is your feature request related to a problem? Please describe.**

Aa a administrator, I want to see all alertmangaer configs and rule groups

**Describe the solution you'd like**

Add a api or web to list all configs and rules


Pull request:
add api to list all alertmanager configs and rule groups
<!--  Thanks for sending a pull request!  Before submitting:

1. Read our CONTRIBUTING.md guide
2. Rebase your PR if it gets out of sync with master
-->

**What this PR does**:

add api to list all user alertmangaer configs

**Which issue(s) this PR fixes**:
Fixes #3530 

**Checklist**
- [x] Tests updated
- [x] Documentation added
- [x] `CHANGELOG.md` updated - the order of entries should be `[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index bc0d6df641..d497eb1cd5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -27,6 +27,8 @@
 * [ENHANCEMENT] Ruler: added the following metrics when ruler sharding is enabled: #3916
   * `cortex_ruler_clients`
   * `cortex_ruler_client_request_duration_seconds`
+* [ENHANCEMENT] Alertmanager: Add API endpoint to list all tenant alertmanager configs: `GET /multitenant_alertmanager/configs`. #3529
+* [ENHANCEMENT] Ruler: Add API endpoint to list all tenant ruler rule groups: `GET /ruler/rule_groups`. #3529
 * [ENHANCEMENT] Query-frontend/scheduler: added querier forget delay (`-query-frontend.querier-forget-delay` and `-query-scheduler.querier-forget-delay`) to mitigate the blast radius in the event queriers crash because of a repeatedly sent "query of death" when shuffle-sharding is enabled. #3901
 * [ENHANCEMENT] Query-frontend: reduced memory allocations when serializing query response. #3964
 * [ENHANCEMENT] Querier / ruler: some optimizations to PromQL query engine. #3934 #3989
@@ -43,6 +45,7 @@
 * [ENHANCEMENT] Block Storage Ingester: `/flush` now accepts two new parameters: `tenant` to specify tenant to flush and `wait=true` to make call synchronous. Multiple tenants can be specified by repeating `tenant` parameter. If no `tenant` is specified, all tenants are flushed, as before. #4073
 * [ENHANCEMENT] Alertmanager: validate configured `-alertmanager.web.external-url` and fail if ends with `/`. #4081
 * [ENHANCEMENT] Allow configuration of Cassandra's host selection policy. #4069
+* [ENHANCEMENT] Store-gateway: retry synching blocks if a per-tenant sync fails. #3975 #4088
 * [BUGFIX] Ruler-API: fix bug where `/api/v1/rules/<namespace>/<group_name>` endpoint return `400` instead of `404`. #4013
 * [BUGFIX] Distributor: reverted changes done to rate limiting in #3825. #3948
 * [BUGFIX] Ingester: Fix race condition when opening and closing tsdb concurrently. #3959
@@ -55,6 +58,7 @@
 * [BUGFIX] Ruler: Rule group limit enforcement should now allow the same number of rules in a group as the limit. #3615
 * [BUGFIX] Frontend, Query-scheduler: allow querier to notify about shutdown without providing any authentication. #4066
 * [BUGFIX] Querier: fixed race condition causing queries to fail right after querier startup with the "empty ring" error. #4068
+* [BUGFIX] Compactor: Increment `cortex_compactor_runs_failed_total` if compactor failed compact a single tenant. #4094
 
 ## Blocksconvert
 
diff --git a/MAINTAINERS b/MAINTAINERS
index 854d5ed245..eee4533eff 100644
--- a/MAINTAINERS
+++ b/MAINTAINERS
@@ -1,7 +1,6 @@
 Bryan Boreham, Weaveworks <bryan@weave.works> (@bboreham)
 Goutham Veeramachaneni, Grafana Labs <gouthamve@gmail.com> (@gouthamve)
 Jacob Lisi, Grafana Labs <jacob.t.lisi@gmail.com> (@jtlisi)
-Ken Haines, Microsoft <khaines@microsoft.com> (@khaines)
 Marco Pracucci, Grafana Labs <marco@pracucci.com> (@pracucci)
 Peter Štibraný, Grafana Labs <pstibrany@gmail.com> (@pstibrany)
 Tom Wilkie, Grafana Labs <tom.wilkie@gmail.com> (@tomwilkie)
diff --git a/docs/api/_index.md b/docs/api/_index.md
index 3c4141852b..3aaaedb889 100644
--- a/docs/api/_index.md
+++ b/docs/api/_index.md
@@ -41,6 +41,7 @@ For the sake of clarity, in this document we have grouped API endpoints by servi
 | [Get tenant ingestion stats](#get-tenant-ingestion-stats) | Querier | `GET /api/v1/user_stats` |
 | [Get tenant chunks](#get-tenant-chunks) | Querier | `GET /api/v1/chunks` |
 | [Ruler ring status](#ruler-ring-status) | Ruler | `GET /ruler/ring` |
+| [Ruler rules ](#ruler-rule-groups) | Ruler | `GET /ruler/rule_groups` |
 | [List rules](#list-rules) | Ruler | `GET <prometheus-http-prefix>/api/v1/rules` |
 | [List alerts](#list-alerts) | Ruler | `GET <prometheus-http-prefix>/api/v1/alerts` |
 | [List rule groups](#list-rule-groups) | Ruler | `GET /api/v1/rules` |
@@ -51,6 +52,7 @@ For the sake of clarity, in this document we have grouped API endpoints by servi
 | [Delete namespace](#delete-namespace) | Ruler | `DELETE /api/v1/rules/{namespace}` |
 | [Delete tenant configuration](#delete-tenant-configuration) | Ruler | `POST /ruler/delete_tenant_config` |
 | [Alertmanager status](#alertmanager-status) | Alertmanager | `GET /multitenant_alertmanager/status` |
+| [Alertmanager configs](#alertmanager-configs) | Alertmanager | `GET /multitenant_alertmanager/configs` |
 | [Alertmanager ring status](#alertmanager-ring-status) | Alertmanager | `GET /multitenant_alertmanager/ring` |
 | [Alertmanager UI](#alertmanager-ui) | Alertmanager | `GET /<alertmanager-http-prefix>` |
 | [Alertmanager Delete Tenant Configuration](#alertmanager-delete-tenant-configuration) | Alertmanager | `POST /multitenant_alertmanager/delete_tenant_config` |
@@ -435,6 +437,14 @@ GET /ruler_ring
 
 Displays a web page with the ruler hash ring status, including the state, healthy and last heartbeat time of each ruler.
 
+### Ruler rules
+
+```
+GET /ruler/rule_groups
+```
+
+List all tenant rules. This endpoint is not part of ruler-API and is always available regardless of whether ruler-API is enabled or not. It should not be exposed to end users. This endpoint returns a YAML dictionary with all the rule groups for each tenant and `200` status code on success.
+
 ### List rules
 
 ```
@@ -667,6 +677,14 @@ GET /status
 
 Displays a web page with the current status of the Alertmanager, including the Alertmanager cluster members.
 
+### Alertmanager configs
+
+```
+GET /multitenant_alertmanager/configs
+```
+
+List all Alertmanager configurations. This endpoint is not part of alertmanager-API and is always available regardless of whether alertmanager-API is enabled or not. It should not be exposed to end users. This endpoint returns a YAML dictionary with all the Alertmanager configurations and `200` status code on success.
+
 ### Alertmanager ring status
 
 ```
diff --git a/pkg/alertmanager/api.go b/pkg/alertmanager/api.go
index 74c8dbaf85..8190040a3e 100644
--- a/pkg/alertmanager/api.go
+++ b/pkg/alertmanager/api.go
@@ -1,14 +1,19 @@
 package alertmanager
 
 import (
+	"context"
 	"fmt"
 	"io/ioutil"
 	"net/http"
 	"os"
 	"path/filepath"
 
+	"github.com/pkg/errors"
+
 	"github.com/cortexproject/cortex/pkg/alertmanager/alertspb"
 	"github.com/cortexproject/cortex/pkg/tenant"
+	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/concurrency"
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
 
 	"github.com/go-kit/kit/log"
@@ -25,6 +30,9 @@ const (
 	errStoringConfiguration  = "unable to store the Alertmanager config"
 	errDeletingConfiguration = "unable to delete the Alertmanager config"
 	errNoOrgID               = "unable to determine the OrgID"
+	errListAllUser           = "unable to list the Alertmanager users"
+
+	fetchConcurrency = 16
 )
 
 // UserConfig is used to communicate a users alertmanager configs
@@ -184,3 +192,48 @@ func validateUserConfig(logger log.Logger, cfg alertspb.AlertConfigDesc) error {
 
 	return nil
 }
+
+func (am *MultitenantAlertmanager) ListAllConfigs(w http.ResponseWriter, r *http.Request) {
+	logger := util_log.WithContext(r.Context(), am.logger)
+	userIDs, err := am.store.ListAllUsers(r.Context())
+	if err != nil {
+		level.Error(logger).Log("msg", "failed to list users of alertmanager", "err", err)
+		http.Error(w, fmt.Sprintf("%s: %s", errListAllUser, err.Error()), http.StatusInternalServerError)
+		return
+	}
+
+	done := make(chan struct{})
+	iter := make(chan interface{})
+
+	go func() {
+		util.StreamWriteYAMLResponse(w, iter, logger)
+		close(done)
+	}()
+
+	err = concurrency.ForEachUser(r.Context(), userIDs, fetchConcurrency, func(ctx context.Context, userID string) error {
+		cfg, err := am.store.GetAlertConfig(ctx, userID)
+		if errors.Is(err, alertspb.ErrNotFound) {
+			return nil
+		} else if err != nil {
+			return errors.Wrapf(err, "failed to fetch alertmanager config for user %s", userID)
+		}
+		data := map[string]*UserConfig{
+			userID: {
+				TemplateFiles:      alertspb.ParseTemplates(cfg),
+				AlertmanagerConfig: cfg.RawConfig,
+			},
+		}
+
+		select {
+		case iter <- data:
+		case <-done: // stop early, if sending response has already finished
+		}
+
+		return nil
+	})
+	if err != nil {
+		level.Error(logger).Log("msg", "failed to list all alertmanager configs", "err", err)
+	}
+	close(iter)
+	<-done
+}
diff --git a/pkg/api/api.go b/pkg/api/api.go
index 59f842555e..4af840eb54 100644
--- a/pkg/api/api.go
+++ b/pkg/api/api.go
@@ -168,6 +168,7 @@ func (a *API) RegisterAlertmanager(am *alertmanager.MultitenantAlertmanager, tar
 	a.indexPage.AddLink(SectionAdminEndpoints, "/multitenant_alertmanager/ring", "Alertmanager Ring Status")
 	// Ensure this route is registered before the prefixed AM route
 	a.RegisterRoute("/multitenant_alertmanager/status", am.GetStatusHandler(), false, "GET")
+	a.RegisterRoute("/multitenant_alertmanager/configs", http.HandlerFunc(am.ListAllConfigs), false, "GET")
 	a.RegisterRoute("/multitenant_alertmanager/ring", http.HandlerFunc(am.RingHandler), false, "GET", "POST")
 	a.RegisterRoute("/multitenant_alertmanager/delete_tenant_config", http.HandlerFunc(am.DeleteUserConfig), true, "POST")
 
@@ -285,6 +286,9 @@ func (a *API) RegisterRuler(r *ruler.Ruler) {
 	// Legacy Ring Route
 	a.RegisterRoute("/ruler_ring", r, false, "GET", "POST")
 
+	// List all user rule groups
+	a.RegisterRoute("/ruler/rule_groups", http.HandlerFunc(r.ListAllRules), false, "GET")
+
 	ruler.RegisterRulerServer(a.server.GRPC, r)
 }
 
diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go
index 88f8121904..2fe37f8c8c 100644
--- a/pkg/compactor/compactor.go
+++ b/pkg/compactor/compactor.go
@@ -353,6 +353,15 @@ func (c *Compactor) starting(ctx context.Context) error {
 	// Create the users scanner.
 	c.usersScanner = cortex_tsdb.NewUsersScanner(c.bucketClient, c.ownUser, c.parentLogger)
 
+	// Create the blocks cleaner (service).
+	c.blocksCleaner = NewBlocksCleaner(BlocksCleanerConfig{
+		DeletionDelay:                      c.compactorCfg.DeletionDelay,
+		CleanupInterval:                    util.DurationWithJitter(c.compactorCfg.CleanupInterval, 0.1),
+		CleanupConcurrency:                 c.compactorCfg.CleanupConcurrency,
+		BlockDeletionMarksMigrationEnabled: c.compactorCfg.BlockDeletionMarksMigrationEnabled,
+		TenantCleanupDelay:                 c.compactorCfg.TenantCleanupDelay,
+	}, c.bucketClient, c.usersScanner, c.cfgProvider, c.parentLogger, c.registerer)
+
 	// Initialize the compactors ring if sharding is enabled.
 	if c.compactorCfg.ShardingEnabled {
 		lifecyclerCfg := c.compactorCfg.ShardingRing.ToLifecyclerConfig()
@@ -406,15 +415,6 @@ func (c *Compactor) starting(ctx context.Context) error {
 		}
 	}
 
-	// Create the blocks cleaner (service).
-	c.blocksCleaner = NewBlocksCleaner(BlocksCleanerConfig{
-		DeletionDelay:                      c.compactorCfg.DeletionDelay,
-		CleanupInterval:                    util.DurationWithJitter(c.compactorCfg.CleanupInterval, 0.1),
-		CleanupConcurrency:                 c.compactorCfg.CleanupConcurrency,
-		BlockDeletionMarksMigrationEnabled: c.compactorCfg.BlockDeletionMarksMigrationEnabled,
-		TenantCleanupDelay:                 c.compactorCfg.TenantCleanupDelay,
-	}, c.bucketClient, c.usersScanner, c.cfgProvider, c.parentLogger, c.registerer)
-
 	// Ensure an initial cleanup occurred before starting the compactor.
 	if err := services.StartAndAwaitRunning(ctx, c.blocksCleaner); err != nil {
 		c.ringSubservices.StopAsync()
@@ -455,11 +455,12 @@ func (c *Compactor) running(ctx context.Context) error {
 
 func (c *Compactor) compactUsers(ctx context.Context) {
 	succeeded := false
+	compactionErrorCount := 0
 
 	c.compactionRunsStarted.Inc()
 
 	defer func() {
-		if succeeded {
+		if succeeded && compactionErrorCount == 0 {
 			c.compactionRunsCompleted.Inc()
 			c.compactionRunsLastSuccess.SetToCurrentTime()
 		} else {
@@ -526,6 +527,7 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 
 		if err = c.compactUserWithRetries(ctx, userID); err != nil {
 			c.compactionRunFailedTenants.Inc()
+			compactionErrorCount++
 			level.Error(c.logger).Log("msg", "failed to compact user blocks", "user", userID, "err", err)
 			continue
 		}
diff --git a/pkg/ruler/ruler.go b/pkg/ruler/ruler.go
index aba2f01d8d..f1af8609e8 100644
--- a/pkg/ruler/ruler.go
+++ b/pkg/ruler/ruler.go
@@ -50,7 +50,8 @@ var (
 
 const (
 	// Number of concurrent group list and group loads operations.
-	loadRulesConcurrency = 10
+	loadRulesConcurrency  = 10
+	fetchRulesConcurrency = 16
 
 	rulerSyncReasonInitial    = "initial"
 	rulerSyncReasonPeriodic   = "periodic"
@@ -59,6 +60,9 @@ const (
 	// Limit errors
 	errMaxRuleGroupsPerUserLimitExceeded        = "per-user rule groups limit (limit: %d actual: %d) exceeded"
 	errMaxRulesPerRuleGroupPerUserLimitExceeded = "per-user rules per rule group limit (limit: %d actual: %d) exceeded"
+
+	// errors
+	errListAllUser = "unable to list the ruler users"
 )
 
 // Config is the configuration for the recording rules server.
@@ -840,3 +844,42 @@ func (r *Ruler) DeleteTenantConfiguration(w http.ResponseWriter, req *http.Reque
 	level.Info(logger).Log("msg", "deleted all tenant rule groups", "user", userID)
 	w.WriteHeader(http.StatusOK)
 }
+
+func (r *Ruler) ListAllRules(w http.ResponseWriter, req *http.Request) {
+	logger := util_log.WithContext(req.Context(), r.logger)
+
+	userIDs, err := r.store.ListAllUsers(req.Context())
+	if err != nil {
+		level.Error(logger).Log("msg", errListAllUser, "err", err)
+		http.Error(w, fmt.Sprintf("%s: %s", errListAllUser, err.Error()), http.StatusInternalServerError)
+		return
+	}
+
+	done := make(chan struct{})
+	iter := make(chan interface{})
+
+	go func() {
+		util.StreamWriteYAMLResponse(w, iter, logger)
+		close(done)
+	}()
+
+	err = concurrency.ForEachUser(req.Context(), userIDs, fetchRulesConcurrency, func(ctx context.Context, userID string) error {
+		rg, err := r.store.ListRuleGroupsForUserAndNamespace(ctx, userID, "")
+		if err != nil {
+			return errors.Wrapf(err, "failed to fetch ruler config for user %s", userID)
+		}
+		data := map[string]map[string][]rulefmt.RuleGroup{userID: rg.Formatted()}
+
+		select {
+		case iter <- data:
+		case <-done: // stop early, if sending response has already finished
+		}
+
+		return nil
+	})
+	if err != nil {
+		level.Error(logger).Log("msg", "failed to list all ruler configs", "err", err)
+	}
+	close(iter)
+	<-done
+}
diff --git a/pkg/storegateway/bucket_stores.go b/pkg/storegateway/bucket_stores.go
index c10c13360f..d5e4057933 100644
--- a/pkg/storegateway/bucket_stores.go
+++ b/pkg/storegateway/bucket_stores.go
@@ -141,7 +141,7 @@ func NewBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy ShardingStra
 func (u *BucketStores) InitialSync(ctx context.Context) error {
 	level.Info(u.logger).Log("msg", "synchronizing TSDB blocks for all users")
 
-	if err := u.syncUsersBlocks(ctx, func(ctx context.Context, s *store.BucketStore) error {
+	if err := u.syncUsersBlocksWithRetries(ctx, func(ctx context.Context, s *store.BucketStore) error {
 		return s.InitialSync(ctx)
 	}); err != nil {
 		level.Warn(u.logger).Log("msg", "failed to synchronize TSDB blocks", "err", err)
@@ -161,7 +161,7 @@ func (u *BucketStores) SyncBlocks(ctx context.Context) error {
 
 func (u *BucketStores) syncUsersBlocksWithRetries(ctx context.Context, f func(context.Context, *store.BucketStore) error) error {
 	retries := util.NewBackoff(ctx, util.BackoffConfig{
-		MinBackoff: 100 * time.Millisecond,
+		MinBackoff: 1 * time.Second,
 		MaxBackoff: 10 * time.Second,
 		MaxRetries: 3,
 	})
diff --git a/pkg/util/http.go b/pkg/util/http.go
index 5e0f176391..979ca37cc5 100644
--- a/pkg/util/http.go
+++ b/pkg/util/http.go
@@ -11,6 +11,8 @@ import (
 	"net/http"
 	"strings"
 
+	"github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
 	"github.com/gogo/protobuf/proto"
 	"github.com/golang/snappy"
 	"github.com/opentracing/opentracing-go"
@@ -98,6 +100,23 @@ func RenderHTTPResponse(w http.ResponseWriter, v interface{}, t *template.Templa
 	}
 }
 
+// StreamWriteYAMLResponse stream writes data as http response
+func StreamWriteYAMLResponse(w http.ResponseWriter, iter chan interface{}, logger log.Logger) {
+	w.Header().Set("Content-Type", "application/yaml")
+	for v := range iter {
+		data, err := yaml.Marshal(v)
+		if err != nil {
+			level.Error(logger).Log("msg", "yaml marshal failed", "err", err)
+			continue
+		}
+		_, err = w.Write(data)
+		if err != nil {
+			level.Error(logger).Log("msg", "write http response failed", "err", err)
+			return
+		}
+	}
+}
+
 // CompressionType for encoding and decoding requests and responses.
 type CompressionType int
 


Test patch:
diff --git a/integration/alertmanager_test.go b/integration/alertmanager_test.go
index a1a08aee7c..9808e64192 100644
--- a/integration/alertmanager_test.go
+++ b/integration/alertmanager_test.go
@@ -27,6 +27,7 @@ import (
 
 const simpleAlertmanagerConfig = `route:
   receiver: dummy
+  group_by: [group]
 receivers:
   - name: dummy`
 
@@ -234,10 +235,13 @@ func TestAlertmanagerClustering(t *testing.T) {
 
 func TestAlertmanagerSharding(t *testing.T) {
 	tests := map[string]struct {
-		legacyAlertStore bool
+		legacyAlertStore  bool
+		replicationFactor int
 	}{
-		"legacy alertstore": {legacyAlertStore: true},
-		"bucket alertstore": {legacyAlertStore: false},
+		"legacy alertstore, RF = 2": {legacyAlertStore: true, replicationFactor: 2},
+		"bucket alertstore, RF = 2": {legacyAlertStore: false, replicationFactor: 2},
+		"legacy alertstore, RF = 3": {legacyAlertStore: true, replicationFactor: 3},
+		"bucket alertstore, RF = 3": {legacyAlertStore: false, replicationFactor: 3},
 	}
 
 	for testName, testCfg := range tests {
@@ -278,8 +282,8 @@ func TestAlertmanagerSharding(t *testing.T) {
 				require.NoError(t, err)
 			}
 
-			// 3 instances, 30 configurations and a replication factor of 2.
-			flags = mergeFlags(flags, AlertmanagerShardingFlags(consul.NetworkHTTPEndpoint(), 2))
+			// 3 instances, 30 configurations and a replication factor of 2 or 3.
+			flags = mergeFlags(flags, AlertmanagerShardingFlags(consul.NetworkHTTPEndpoint(), testCfg.replicationFactor))
 
 			// Wait for the Alertmanagers to start.
 			alertmanager1 := e2ecortex.NewAlertmanager("alertmanager-1", flags, "")
@@ -302,8 +306,7 @@ func TestAlertmanagerSharding(t *testing.T) {
 			require.NoError(t, alertmanagers.WaitSumMetrics(e2e.Equals(90), "cortex_alertmanager_tenants_discovered"))
 			// We know that the ring has settled when every instance has some tenants and the total number of tokens have been assigned.
 			// The total number of tenants across all instances is: total alertmanager configs * replication factor.
-			// In this case: 30 * 2
-			require.NoError(t, alertmanagers.WaitSumMetricsWithOptions(e2e.Equals(60), []string{"cortex_alertmanager_config_last_reload_successful"}, e2e.SkipMissingMetrics))
+			require.NoError(t, alertmanagers.WaitSumMetricsWithOptions(e2e.Equals(float64(30*testCfg.replicationFactor)), []string{"cortex_alertmanager_config_last_reload_successful"}, e2e.SkipMissingMetrics))
 			require.NoError(t, alertmanagers.WaitSumMetrics(e2e.Equals(float64(1152)), "cortex_ring_tokens_total"))
 
 			// Now, let's make sure state is replicated across instances.
@@ -361,7 +364,7 @@ func TestAlertmanagerSharding(t *testing.T) {
 
 				// Reading silences do not currently read from all replicas. We have to wait for
 				// the silence to be replicated asynchronously, before we can reliably read them.
-				require.NoError(t, waitForSilences("active", 6))
+				require.NoError(t, waitForSilences("active", 3*testCfg.replicationFactor))
 			}
 
 			assertSilences := func(list []types.Silence, s1, s2, s3 types.SilenceState) {
@@ -439,7 +442,7 @@ func TestAlertmanagerSharding(t *testing.T) {
 				// These waits are required as deletion replication is currently
 				// asynchronous, and silence reading is not consistent. Once
 				// merging is implemented on the read path, this is not needed.
-				require.NoError(t, waitForSilences("expired", 2))
+				require.NoError(t, waitForSilences("expired", 1*testCfg.replicationFactor))
 
 				for _, c := range clients {
 					list, err := c.GetSilences(context.Background())
@@ -449,7 +452,7 @@ func TestAlertmanagerSharding(t *testing.T) {
 
 				err = c2.DeleteSilence(context.Background(), id3)
 				assert.NoError(t, err)
-				require.NoError(t, waitForSilences("expired", 4))
+				require.NoError(t, waitForSilences("expired", 2*testCfg.replicationFactor))
 
 				for _, c := range clients {
 					list, err := c.GetSilences(context.Background())
@@ -459,7 +462,7 @@ func TestAlertmanagerSharding(t *testing.T) {
 
 				err = c3.DeleteSilence(context.Background(), id1)
 				assert.NoError(t, err)
-				require.NoError(t, waitForSilences("expired", 6))
+				require.NoError(t, waitForSilences("expired", 3*testCfg.replicationFactor))
 
 				for _, c := range clients {
 					list, err := c.GetSilences(context.Background())
@@ -467,6 +470,69 @@ func TestAlertmanagerSharding(t *testing.T) {
 					assertSilences(list, types.SilenceStateExpired, types.SilenceStateExpired, types.SilenceStateExpired)
 				}
 			}
+
+			alert := func(i, g int) *model.Alert {
+				return &model.Alert{
+					Labels: model.LabelSet{
+						"name":  model.LabelValue(fmt.Sprintf("alert_%d", i)),
+						"group": model.LabelValue(fmt.Sprintf("group_%d", g)),
+					},
+					StartsAt: time.Now(),
+					EndsAt:   time.Now().Add(time.Hour),
+				}
+			}
+
+			alertNames := func(list []model.Alert) (r []string) {
+				for _, a := range list {
+					r = append(r, string(a.Labels["name"]))
+				}
+				return
+			}
+
+			// Endpoint: POST /alerts
+			{
+				err = c1.SendAlertToAlermanager(context.Background(), alert(1, 1))
+				require.NoError(t, err)
+				err = c2.SendAlertToAlermanager(context.Background(), alert(2, 1))
+				require.NoError(t, err)
+				err = c3.SendAlertToAlermanager(context.Background(), alert(3, 2))
+				require.NoError(t, err)
+
+				// API does not block for the write slowest replica, and reads do not
+				// currently merge results from multiple replicas, so we have to wait.
+				require.NoError(t, alertmanagers.WaitSumMetricsWithOptions(
+					e2e.Equals(float64(3*testCfg.replicationFactor)),
+					[]string{"cortex_alertmanager_alerts_received_total"},
+					e2e.SkipMissingMetrics))
+			}
+
+			// Endpoint: GET /alerts
+			{
+				for _, c := range clients {
+					list, err := c.GetAlerts(context.Background())
+					require.NoError(t, err)
+					assert.ElementsMatch(t, []string{"alert_1", "alert_2", "alert_3"}, alertNames(list))
+				}
+			}
+
+			// Endpoint: GET /alerts/groups
+			{
+				for _, c := range clients {
+					list, err := c.GetAlertGroups(context.Background())
+					require.NoError(t, err)
+
+					assert.Equal(t, 2, len(list))
+					groups := make(map[string][]model.Alert)
+					for _, g := range list {
+						groups[string(g.Labels["group"])] = g.Alerts
+					}
+
+					require.Contains(t, groups, "group_1")
+					assert.ElementsMatch(t, []string{"alert_1", "alert_2"}, alertNames(groups["group_1"]))
+					require.Contains(t, groups, "group_2")
+					assert.ElementsMatch(t, []string{"alert_3"}, alertNames(groups["group_2"]))
+				}
+			}
 		})
 	}
 }
diff --git a/integration/e2ecortex/client.go b/integration/e2ecortex/client.go
index a53d248612..51ca05fe9f 100644
--- a/integration/e2ecortex/client.go
+++ b/integration/e2ecortex/client.go
@@ -509,6 +509,77 @@ func (c *Client) SendAlertToAlermanager(ctx context.Context, alert *model.Alert)
 	return nil
 }
 
+func (c *Client) GetAlerts(ctx context.Context) ([]model.Alert, error) {
+	u := c.alertmanagerClient.URL("api/prom/api/v1/alerts", nil)
+
+	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
+	if err != nil {
+		return nil, fmt.Errorf("error creating request: %v", err)
+	}
+
+	resp, body, err := c.alertmanagerClient.Do(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode == http.StatusNotFound {
+		return nil, ErrNotFound
+	}
+
+	if resp.StatusCode/100 != 2 {
+		return nil, fmt.Errorf("getting alerts failed with status %d and error %v", resp.StatusCode, string(body))
+	}
+
+	type response struct {
+		Status string        `json:"status"`
+		Data   []model.Alert `json:"data"`
+	}
+
+	decoded := &response{}
+	if err := json.Unmarshal(body, decoded); err != nil {
+		return nil, err
+	}
+
+	if decoded.Status != "success" {
+		return nil, fmt.Errorf("unexpected response status '%s'", decoded.Status)
+	}
+
+	return decoded.Data, nil
+}
+
+type AlertGroup struct {
+	Labels model.LabelSet `json:"labels"`
+	Alerts []model.Alert  `json:"alerts"`
+}
+
+func (c *Client) GetAlertGroups(ctx context.Context) ([]AlertGroup, error) {
+	u := c.alertmanagerClient.URL("api/prom/api/v2/alerts/groups", nil)
+
+	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
+	if err != nil {
+		return nil, fmt.Errorf("error creating request: %v", err)
+	}
+
+	resp, body, err := c.alertmanagerClient.Do(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode == http.StatusNotFound {
+		return nil, ErrNotFound
+	}
+
+	if resp.StatusCode/100 != 2 {
+		return nil, fmt.Errorf("getting alert groups failed with status %d and error %v", resp.StatusCode, string(body))
+	}
+
+	decoded := []AlertGroup{}
+	if err := json.Unmarshal(body, &decoded); err != nil {
+		return nil, err
+	}
+	return decoded, nil
+}
+
 // CreateSilence creates a new silence and returns the unique identifier of the silence.
 func (c *Client) CreateSilence(ctx context.Context, silence types.Silence) (string, error) {
 	u := c.alertmanagerClient.URL("api/prom/api/v1/silences", nil)
diff --git a/pkg/alertmanager/api_test.go b/pkg/alertmanager/api_test.go
index d0625165d5..1e7a183c5e 100644
--- a/pkg/alertmanager/api_test.go
+++ b/pkg/alertmanager/api_test.go
@@ -10,11 +10,18 @@ import (
 	"testing"
 
 	"github.com/go-kit/kit/log"
+	"github.com/gorilla/mux"
+	"github.com/prometheus/client_golang/prometheus"
+	"gopkg.in/yaml.v2"
+
 	"github.com/thanos-io/thanos/pkg/objstore"
 
+	"github.com/cortexproject/cortex/pkg/util/flagext"
+
 	"github.com/cortexproject/cortex/pkg/alertmanager/alertspb"
 	"github.com/cortexproject/cortex/pkg/alertmanager/alertstore/bucketclient"
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
+	"github.com/cortexproject/cortex/pkg/util/services"
 
 	"github.com/stretchr/testify/require"
 	"github.com/weaveworks/common/user"
@@ -199,3 +206,86 @@ func TestMultitenantAlertmanager_DeleteUserConfig(t *testing.T) {
 		require.Equal(t, 0, len(storage.Objects()))
 	}
 }
+
+func TestAMConfigListUserConfig(t *testing.T) {
+	testCases := map[string]*UserConfig{
+		"user1": {
+			AlertmanagerConfig: `
+global:
+  resolve_timeout: 5m
+route:
+  receiver: route1
+  group_by:
+  - '...'
+  continue: false
+receivers:
+- name: route1
+  webhook_configs:
+  - send_resolved: true
+    http_config: {}
+    url: http://alertmanager/api/notifications?orgId=1&rrid=7
+    max_alerts: 0
+`,
+		},
+		"user2": {
+			AlertmanagerConfig: `
+global:
+  resolve_timeout: 5m
+route:
+  receiver: route1
+  group_by:
+  - '...'
+  continue: false
+receivers:
+- name: route1
+  webhook_configs:
+  - send_resolved: true
+    http_config: {}
+    url: http://alertmanager/api/notifications?orgId=2&rrid=7
+    max_alerts: 0
+`,
+		},
+	}
+
+	storage := objstore.NewInMemBucket()
+	alertStore := bucketclient.NewBucketAlertStore(storage, nil, log.NewNopLogger())
+
+	for u, cfg := range testCases {
+		err := alertStore.SetAlertConfig(context.Background(), alertspb.AlertConfigDesc{
+			User:      u,
+			RawConfig: cfg.AlertmanagerConfig,
+		})
+		require.NoError(t, err)
+	}
+
+	externalURL := flagext.URLValue{}
+	err := externalURL.Set("http://localhost:8080/alertmanager")
+	require.NoError(t, err)
+
+	// Create the Multitenant Alertmanager.
+	reg := prometheus.NewPedanticRegistry()
+	cfg := mockAlertmanagerConfig(t)
+	am, err := createMultitenantAlertmanager(cfg, nil, nil, alertStore, nil, log.NewNopLogger(), reg)
+	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), am))
+	defer services.StopAndAwaitTerminated(context.Background(), am) //nolint:errcheck
+
+	err = am.loadAndSyncConfigs(context.Background(), reasonPeriodic)
+	require.NoError(t, err)
+	require.Len(t, am.alertmanagers, 2)
+
+	router := mux.NewRouter()
+	router.Path("/multitenant_alertmanager/configs").Methods(http.MethodGet).HandlerFunc(am.ListAllConfigs)
+	req := httptest.NewRequest("GET", "https://localhost:8080/multitenant_alertmanager/configs", nil)
+	w := httptest.NewRecorder()
+	router.ServeHTTP(w, req)
+
+	resp := w.Result()
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+	require.Equal(t, "application/yaml", resp.Header.Get("Content-Type"))
+	body, err := ioutil.ReadAll(resp.Body)
+	require.NoError(t, err)
+	old, err := yaml.Marshal(testCases)
+	require.NoError(t, err)
+	require.YAMLEq(t, string(old), string(body))
+}
diff --git a/pkg/compactor/compactor_test.go b/pkg/compactor/compactor_test.go
index 9d3c53c57e..9c14cb0924 100644
--- a/pkg/compactor/compactor_test.go
+++ b/pkg/compactor/compactor_test.go
@@ -414,6 +414,50 @@ func TestCompactor_ShouldRetryCompactionOnFailureWhileDiscoveringUsersFromBucket
 	))
 }
 
+func TestCompactor_ShouldIncrementCompactionErrorIfFailedToCompactASingleTenant(t *testing.T) {
+	t.Parallel()
+
+	userID := "test-user"
+	bucketClient := &bucket.ClientMock{}
+	bucketClient.MockIter("", []string{userID}, nil)
+	bucketClient.MockIter(userID+"/", []string{userID + "/01DTVP434PA9VFXSW2JKB3392D"}, nil)
+	bucketClient.MockIter(userID+"/markers/", nil, nil)
+	bucketClient.MockExists(path.Join(userID, cortex_tsdb.TenantDeletionMarkPath), false, nil)
+	bucketClient.MockGet(userID+"/01DTVP434PA9VFXSW2JKB3392D/meta.json", mockBlockMetaJSON("01DTVP434PA9VFXSW2JKB3392D"), nil)
+	bucketClient.MockGet(userID+"/01DTVP434PA9VFXSW2JKB3392D/deletion-mark.json", "", nil)
+	bucketClient.MockGet(userID+"/bucket-index.json.gz", "", nil)
+	bucketClient.MockUpload(userID+"/bucket-index.json.gz", nil)
+
+	c, _, tsdbPlannerMock, _, registry := prepare(t, prepareConfig(), bucketClient)
+	tsdbPlannerMock.On("Plan", mock.Anything, mock.Anything).Return([]*metadata.Meta{}, errors.New("Failed to plan"))
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
+
+	// Wait until all retry attempts have completed.
+	cortex_testutil.Poll(t, time.Second, 1.0, func() interface{} {
+		return prom_testutil.ToFloat64(c.compactionRunsFailed)
+	})
+
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), c))
+
+	assert.NoError(t, prom_testutil.GatherAndCompare(registry, strings.NewReader(`
+		# TYPE cortex_compactor_runs_started_total counter
+		# HELP cortex_compactor_runs_started_total Total number of compaction runs started.
+		cortex_compactor_runs_started_total 1
+
+		# TYPE cortex_compactor_runs_completed_total counter
+		# HELP cortex_compactor_runs_completed_total Total number of compaction runs successfully completed.
+		cortex_compactor_runs_completed_total 0
+
+		# TYPE cortex_compactor_runs_failed_total counter
+		# HELP cortex_compactor_runs_failed_total Total number of compaction runs failed.
+		cortex_compactor_runs_failed_total 1
+	`),
+		"cortex_compactor_runs_started_total",
+		"cortex_compactor_runs_completed_total",
+		"cortex_compactor_runs_failed_total",
+	))
+}
+
 func TestCompactor_ShouldIterateOverUsersAndRunCompaction(t *testing.T) {
 	t.Parallel()
 
diff --git a/pkg/ruler/ruler_test.go b/pkg/ruler/ruler_test.go
index 1426d00539..deba332685 100644
--- a/pkg/ruler/ruler_test.go
+++ b/pkg/ruler/ruler_test.go
@@ -17,6 +17,7 @@ import (
 
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
+	"github.com/gorilla/mux"
 	"github.com/prometheus/client_golang/prometheus"
 	prom_testutil "github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/prometheus/notifier"
@@ -28,6 +29,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/weaveworks/common/user"
+	"gopkg.in/yaml.v2"
 
 	"github.com/cortexproject/cortex/pkg/chunk"
 	"github.com/cortexproject/cortex/pkg/cortexpb"
@@ -843,6 +845,37 @@ type ruleGroupKey struct {
 	user, namespace, group string
 }
 
+func TestRuler_ListAllRules(t *testing.T) {
+	cfg, cleanup := defaultRulerConfig(newMockRuleStore(mockRules))
+	defer cleanup()
+
+	r, rcleanup := newTestRuler(t, cfg)
+	defer rcleanup()
+	defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck
+
+	router := mux.NewRouter()
+	router.Path("/ruler/rule_groups").Methods(http.MethodGet).HandlerFunc(r.ListAllRules)
+
+	req := requestFor(t, http.MethodGet, "https://localhost:8080/ruler/rule_groups", nil, "")
+	w := httptest.NewRecorder()
+	router.ServeHTTP(w, req)
+
+	resp := w.Result()
+	body, _ := ioutil.ReadAll(resp.Body)
+
+	// Check status code and header
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+	require.Equal(t, "application/yaml", resp.Header.Get("Content-Type"))
+
+	gs := make(map[string]map[string][]rulefmt.RuleGroup) // user:namespace:[]rulefmt.RuleGroup
+	for userID := range mockRules {
+		gs[userID] = mockRules[userID].Formatted()
+	}
+	expectedResponse, err := yaml.Marshal(gs)
+	require.NoError(t, err)
+	require.YAMLEq(t, string(expectedResponse), string(body))
+}
+
 type senderFunc func(alerts ...*notifier.Alert)
 
 func (s senderFunc) Send(alerts ...*notifier.Alert) {
diff --git a/pkg/storegateway/bucket_stores_test.go b/pkg/storegateway/bucket_stores_test.go
index 79642dd66d..5206144287 100644
--- a/pkg/storegateway/bucket_stores_test.go
+++ b/pkg/storegateway/bucket_stores_test.go
@@ -2,7 +2,9 @@ package storegateway
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"math"
 	"os"
@@ -24,6 +26,7 @@ import (
 	"github.com/stretchr/testify/require"
 	thanos_metadata "github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/extprom"
+	"github.com/thanos-io/thanos/pkg/objstore"
 	"github.com/thanos-io/thanos/pkg/store"
 	"github.com/thanos-io/thanos/pkg/store/labelpb"
 	"github.com/thanos-io/thanos/pkg/store/storepb"
@@ -118,6 +121,68 @@ func TestBucketStores_InitialSync(t *testing.T) {
 	assert.Greater(t, testutil.ToFloat64(stores.syncLastSuccess), float64(0))
 }
 
+func TestBucketStores_InitialSyncShouldRetryOnFailure(t *testing.T) {
+	ctx := context.Background()
+	cfg, cleanup := prepareStorageConfig(t)
+	defer cleanup()
+
+	storageDir, err := ioutil.TempDir(os.TempDir(), "storage-*")
+	require.NoError(t, err)
+
+	// Generate a block for the user in the storage.
+	generateStorageBlock(t, storageDir, "user-1", "series_1", 10, 100, 15)
+
+	bucket, err := filesystem.NewBucketClient(filesystem.Config{Directory: storageDir})
+	require.NoError(t, err)
+
+	// Wrap the bucket to fail the 1st Get() request.
+	bucket = &failFirstGetBucket{Bucket: bucket}
+
+	reg := prometheus.NewPedanticRegistry()
+	stores, err := NewBucketStores(cfg, NewNoShardingStrategy(), bucket, defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), reg)
+	require.NoError(t, err)
+
+	// Initial sync should succeed even if a transient error occurs.
+	require.NoError(t, stores.InitialSync(ctx))
+
+	// Query series after the initial sync.
+	seriesSet, warnings, err := querySeries(stores, "user-1", "series_1", 20, 40)
+	require.NoError(t, err)
+	assert.Empty(t, warnings)
+	require.Len(t, seriesSet, 1)
+	assert.Equal(t, []labelpb.ZLabel{{Name: labels.MetricName, Value: "series_1"}}, seriesSet[0].Labels)
+
+	assert.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
+			# HELP cortex_blocks_meta_syncs_total Total blocks metadata synchronization attempts
+			# TYPE cortex_blocks_meta_syncs_total counter
+			cortex_blocks_meta_syncs_total 2
+
+			# HELP cortex_blocks_meta_sync_failures_total Total blocks metadata synchronization failures
+			# TYPE cortex_blocks_meta_sync_failures_total counter
+			cortex_blocks_meta_sync_failures_total 1
+
+			# HELP cortex_bucket_store_blocks_loaded Number of currently loaded blocks.
+			# TYPE cortex_bucket_store_blocks_loaded gauge
+			cortex_bucket_store_blocks_loaded 1
+
+			# HELP cortex_bucket_store_block_loads_total Total number of remote block loading attempts.
+			# TYPE cortex_bucket_store_block_loads_total counter
+			cortex_bucket_store_block_loads_total 1
+
+			# HELP cortex_bucket_store_block_load_failures_total Total number of failed remote block loading attempts.
+			# TYPE cortex_bucket_store_block_load_failures_total counter
+			cortex_bucket_store_block_load_failures_total 0
+	`),
+		"cortex_blocks_meta_syncs_total",
+		"cortex_blocks_meta_sync_failures_total",
+		"cortex_bucket_store_block_loads_total",
+		"cortex_bucket_store_block_load_failures_total",
+		"cortex_bucket_store_blocks_loaded",
+	))
+
+	assert.Greater(t, testutil.ToFloat64(stores.syncLastSuccess), float64(0))
+}
+
 func TestBucketStores_SyncBlocks(t *testing.T) {
 	const (
 		userID     = "user-1"
@@ -534,3 +599,18 @@ func (u *userShardingStrategy) FilterBlocks(ctx context.Context, userID string,
 	}
 	return nil
 }
+
+// failFirstGetBucket is an objstore.Bucket wrapper which fails the first Get() request with a mocked error.
+type failFirstGetBucket struct {
+	objstore.Bucket
+
+	firstGet atomic.Bool
+}
+
+func (f *failFirstGetBucket) Get(ctx context.Context, name string) (io.ReadCloser, error) {
+	if f.firstGet.CAS(false, true) {
+		return nil, errors.New("Get() request mocked error")
+	}
+
+	return f.Bucket.Get(ctx, name)
+}
diff --git a/pkg/util/http_test.go b/pkg/util/http_test.go
index 49c931a3e0..746a1bfa34 100644
--- a/pkg/util/http_test.go
+++ b/pkg/util/http_test.go
@@ -5,14 +5,19 @@ import (
 	"context"
 	"html/template"
 	"io/ioutil"
+	"math/rand"
 	"net/http"
 	"net/http/httptest"
+	"strconv"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/yaml.v2"
 
 	"github.com/cortexproject/cortex/pkg/cortexpb"
 	"github.com/cortexproject/cortex/pkg/util"
+	util_log "github.com/cortexproject/cortex/pkg/util/log"
 )
 
 func TestRenderHTTPResponse(t *testing.T) {
@@ -84,6 +89,53 @@ func TestWriteTextResponse(t *testing.T) {
 	assert.Equal(t, "text/plain", w.Header().Get("Content-Type"))
 }
 
+func TestStreamWriteYAMLResponse(t *testing.T) {
+	type testStruct struct {
+		Name  string `yaml:"name"`
+		Value int    `yaml:"value"`
+	}
+	tt := struct {
+		name                string
+		headers             map[string]string
+		expectedOutput      string
+		expectedContentType string
+		value               map[string]*testStruct
+	}{
+		name: "Test Stream Render YAML",
+		headers: map[string]string{
+			"Content-Type": "application/yaml",
+		},
+		expectedContentType: "application/yaml",
+		value:               make(map[string]*testStruct),
+	}
+	for i := 0; i < rand.Intn(100); i++ {
+		ts := testStruct{
+			Name:  "testName" + strconv.Itoa(i),
+			Value: i,
+		}
+		tt.value[ts.Name] = &ts
+	}
+	d, err := yaml.Marshal(tt.value)
+	require.NoError(t, err)
+	tt.expectedOutput = string(d)
+	w := httptest.NewRecorder()
+
+	done := make(chan struct{})
+	iter := make(chan interface{})
+	go func() {
+		util.StreamWriteYAMLResponse(w, iter, util_log.Logger)
+		close(done)
+	}()
+	for k, v := range tt.value {
+		iter <- map[string]*testStruct{k: v}
+	}
+	close(iter)
+	<-done
+	assert.Equal(t, tt.expectedContentType, w.Header().Get("Content-Type"))
+	assert.Equal(t, 200, w.Code)
+	assert.YAMLEq(t, tt.expectedOutput, w.Body.String())
+}
+
 func TestParseProtoReader(t *testing.T) {
 	// 47 bytes compressed and 53 uncompressed
 	req := &cortexpb.PreallocWriteRequest{