add api to list all alertmanager configs and rule groups

.vscode/launch.json

@@ -0,0 +1,18 @@
+{
+    // 使用 IntelliSense 了解相关属性。 
+    // 悬停以查看现有属性的描述。
+    // 欲了解更多信息，请访问: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "alertmanager",
+            "type": "go",
+            "request": "launch",
+            "mode": "debug",
+            "program": "${workspaceFolder}/cmd/cortex",
+            "args": [
+                "--config.file=${workspaceFolder}/docs/configuration/single-process-config-alertmanager.yaml"
+            ]
+        }
+    ]
+}

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`. #3259
+* [ENHANCEMENT] Ruler: Add API endpoint to list all tenant ruler rule groups: `GET /ruler/rule_groups`. #3259
 * [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

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` |
@@ -431,6 +433,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
 
 ```
@@ -663,6 +673,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
 
 ```

docs/configuration/single-process-config-alertmanager.yaml

@@ -0,0 +1,28 @@
+
+# Configuration for running Cortex in single-process mode.
+# This configuration should not be used in production.
+# It is only for getting started and development.
+
+# Disable the requirement that every request to Cortex has a
+# X-Scope-OrgID header. `fake` will be substituted in instead.
+auth_enabled: true
+
+target: alertmanager
+
+server:
+  http_listen_port: 9009
+  log_level: debug
+
+  # Configure the server to allow messages up to 100MB.
+  grpc_server_max_recv_msg_size: 104857600
+  grpc_server_max_send_msg_size: 104857600
+  grpc_server_max_concurrent_streams: 1000
+
+
+alertmanager:
+  external_url: http://127.0.0.1/
+  storage:
+    type: local
+    local:
+      path: /data/cortex/data/amconfig/
+

pkg/alertmanager/api.go

@@ -1,14 +1,20 @@
 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/chunk"
 	"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 +31,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 +193,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, chunk.ErrStorageObjectNotFound) {
+			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
+}

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,84 @@ 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, _ := ioutil.ReadAll(resp.Body)
+	old, _ := yaml.Marshal(testCases)
+	require.YAMLEq(t, string(old), string(body))
+}

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)
 }
 

pkg/ruler/ruler.go

@@ -24,6 +24,7 @@ import (
 	"github.com/weaveworks/common/user"
 	"golang.org/x/sync/errgroup"
 
+	"github.com/cortexproject/cortex/pkg/chunk"
 	"github.com/cortexproject/cortex/pkg/cortexpb"
 	"github.com/cortexproject/cortex/pkg/ring"
 	ring_client "github.com/cortexproject/cortex/pkg/ring/client"
@@ -50,7 +51,8 @@ var (
 
 const (
 	// Number of concurrent group list and group loads operations.
-	loadRulesConcurrency = 10
+	loadRulesConcurrency  = 10
+	fetchRulesConcurrency = 16
 
 	rulerSyncReasonInitial    = "initial"
 	rulerSyncReasonPeriodic   = "periodic"
@@ -59,6 +61,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 +845,44 @@ 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 errors.Is(err, chunk.ErrStorageObjectNotFound) {
+			return nil
+		} else 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
+}