Add cortex_ruler_rule_groups_in_store metric

Signed-off-by: Emmanuel Lodovice <lodovice@amazon.com>

Author: emanlodovice

CHANGELOG.md

@@ -36,6 +36,7 @@
 * [ENHANCEMENT] Ruler: Improve GetRules response time by refactoring mutexes and introducing a temporary rules cache in `ruler/manager.go`. #5805
 * [ENHANCEMENT] Querier: Add context error check when merging slices from ingesters for GetLabel operations. #5837
 * [ENHANCEMENT] Ring: Add experimental `-ingester.tokens-generator-strategy=minimize-spread` flag to enable the new minimize spread token generator strategy. #5855
+* [ENHANCEMENT] Ruler: Add new ruler metric `cortex_ruler_rule_groups_in_store`. #5869
 * [BUGFIX] Distributor: Do not use label with empty values for sharding #5717
 * [BUGFIX] Query Frontend: queries with negative offset should check whether it is cacheable or not. #5719
 * [BUGFIX] Redis Cache: pass `cache_size` config correctly. #5734

integration/ruler_test.go

@@ -382,6 +382,9 @@ func TestRulerSharding(t *testing.T) {
 	// between the two rulers.
 	require.NoError(t, ruler1.WaitSumMetrics(e2e.Less(numRulesGroups), "cortex_prometheus_rule_group_rules"))
 	require.NoError(t, ruler2.WaitSumMetrics(e2e.Less(numRulesGroups), "cortex_prometheus_rule_group_rules"))
+	// Even with rules sharded, we expect rulers to have the same cortex_ruler_rule_groups_in_store metric values
+	require.NoError(t, ruler1.WaitSumMetrics(e2e.Equals(numRulesGroups), "cortex_ruler_rule_groups_in_store"))
+	require.NoError(t, ruler2.WaitSumMetrics(e2e.Equals(numRulesGroups), "cortex_ruler_rule_groups_in_store"))
 
 	// Fetch the rules and ensure they match the configured ones.
 	actualGroups, err := c.GetPrometheusRules(e2ecortex.DefaultFilter)

pkg/ruler/manager_metrics.go

@@ -271,3 +271,34 @@ func (m *RuleEvalMetrics) deletePerUserMetrics(userID string) {
 		m.RulerQuerySeconds.DeleteLabelValues(userID)
 	}
 }
+
+type RuleGroupMetrics struct {
+	RuleGroupsInStore *prometheus.GaugeVec
+	tenants           map[string]struct{}
+}
+
+func NewRuleGroupMetrics(reg prometheus.Registerer) *RuleGroupMetrics {
+	m := &RuleGroupMetrics{
+		RuleGroupsInStore: promauto.With(reg).NewGaugeVec(prometheus.GaugeOpts{
+			Name: "cortex_ruler_rule_groups_in_store",
+			Help: "The number of rule groups a tenant has in store.",
+		}, []string{"user"}),
+	}
+	return m
+}
+
+// UpdateRuleGroupsInStore updates the cortex_ruler_rule_groups_in_store metric with the provided number of rule
+// groups per tenant and removing the metrics for tenants that are not present anymore
+func (r *RuleGroupMetrics) UpdateRuleGroupsInStore(ruleGroupsCount map[string]int) {
+	tenants := make(map[string]struct{}, len(ruleGroupsCount))
+	for userID, count := range ruleGroupsCount {
+		tenants[userID] = struct{}{}
+		r.RuleGroupsInStore.WithLabelValues(userID).Set(float64(count))
+	}
+	for userID := range r.tenants {
+		if _, ok := tenants[userID]; !ok {
+			r.RuleGroupsInStore.DeleteLabelValues(userID)
+		}
+	}
+	r.tenants = tenants
+}

pkg/ruler/manager_metrics_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	dto "github.com/prometheus/client_model/go"
+	io_prometheus_client "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -595,3 +596,49 @@ func TestRuleEvalMetricsDeletePerUserMetrics(t *testing.T) {
 		require.Contains(t, mfm[name].String(), "value:\"fake2\"")
 	}
 }
+
+func TestRuleGroupMetrics(t *testing.T) {
+	reg := prometheus.NewPedanticRegistry()
+	m := NewRuleGroupMetrics(reg)
+	m.UpdateRuleGroupsInStore(map[string]int{
+		"fake1": 10,
+		"fake2": 20,
+	})
+	gm, err := reg.Gather()
+	require.NoError(t, err)
+	mfm, err := util.NewMetricFamilyMap(gm)
+	require.NoError(t, err)
+	require.Equal(t, 2, len(mfm["cortex_ruler_rule_groups_in_store"].Metric))
+	requireMetricEqual(t, mfm["cortex_ruler_rule_groups_in_store"].Metric[0], map[string]string{
+		"user": "fake1",
+	}, float64(10))
+	requireMetricEqual(t, mfm["cortex_ruler_rule_groups_in_store"].Metric[1], map[string]string{
+		"user": "fake2",
+	}, float64(20))
+	m.UpdateRuleGroupsInStore(map[string]int{
+		"fake2": 30,
+	})
+	gm, err = reg.Gather()
+	require.NoError(t, err)
+	mfm, err = util.NewMetricFamilyMap(gm)
+	require.NoError(t, err)
+	require.Equal(t, 1, len(mfm["cortex_ruler_rule_groups_in_store"].Metric))
+	requireMetricEqual(t, mfm["cortex_ruler_rule_groups_in_store"].Metric[0], map[string]string{
+		"user": "fake2",
+	}, float64(30))
+	m.UpdateRuleGroupsInStore(make(map[string]int))
+	gm, err = reg.Gather()
+	require.NoError(t, err)
+	mfm, err = util.NewMetricFamilyMap(gm)
+	require.NoError(t, err)
+	require.Nil(t, mfm["cortex_ruler_rule_groups_in_store"])
+}
+
+func requireMetricEqual(t *testing.T, m *io_prometheus_client.Metric, labels map[string]string, value float64) {
+	l := m.GetLabel()
+	require.Equal(t, len(labels), len(l))
+	for _, pair := range l {
+		require.Equal(t, labels[*pair.Name], *pair.Value)
+	}
+	require.Equal(t, value, *m.Gauge.Value)
+}

pkg/ruler/ruler.go

@@ -270,6 +270,7 @@ type Ruler struct {
 	rulerSync                  *prometheus.CounterVec
 	ruleGroupStoreLoadDuration prometheus.Gauge
 	ruleGroupSyncDuration      prometheus.Gauge
+	ruleGroupMetrics           *RuleGroupMetrics
 
 	allowedTenants *util.AllowedTenants
 
@@ -312,6 +313,8 @@ func newRuler(cfg Config, manager MultiTenantManager, reg prometheus.Registerer,
 			Name: "cortex_ruler_rule_group_sync_duration_seconds",
 			Help: "The duration in seconds required to sync and load rule groups from storage.",
 		}),
+
+		ruleGroupMetrics: NewRuleGroupMetrics(reg),
 	}
 
 	if len(cfg.EnabledTenants) > 0 {
@@ -606,7 +609,9 @@ func (r *Ruler) listRulesNoSharding(ctx context.Context) (map[string]rulespb.Rul
 	if err != nil {
 		return nil, err
 	}
+	ruleGroupCounts := make(map[string]int, len(allRuleGroups))
 	for userID, groups := range allRuleGroups {
+		ruleGroupCounts[userID] = len(groups)
 		disabledRuleGroupsForUser := r.limits.DisabledRuleGroups(userID)
 		if len(disabledRuleGroupsForUser) == 0 {
 			continue
@@ -621,6 +626,7 @@ func (r *Ruler) listRulesNoSharding(ctx context.Context) (map[string]rulespb.Rul
 		}
 		allRuleGroups[userID] = filteredGroupsForUser
 	}
+	r.ruleGroupMetrics.UpdateRuleGroupsInStore(ruleGroupCounts)
 	return allRuleGroups, nil
 }
 
@@ -630,13 +636,16 @@ func (r *Ruler) listRulesShardingDefault(ctx context.Context) (map[string]rulesp
 		return nil, err
 	}
 
+	ruleGroupCounts := make(map[string]int, len(configs))
 	filteredConfigs := make(map[string]rulespb.RuleGroupList)
 	for userID, groups := range configs {
+		ruleGroupCounts[userID] = len(groups)
 		filtered := filterRuleGroups(userID, groups, r.limits.DisabledRuleGroups(userID), r.ring, r.lifecycler.GetInstanceAddr(), r.logger, r.ringCheckErrors)
 		if len(filtered) > 0 {
 			filteredConfigs[userID] = filtered
 		}
 	}
+	r.ruleGroupMetrics.UpdateRuleGroupsInStore(ruleGroupCounts)
 	return filteredConfigs, nil
 }
 
@@ -664,6 +673,7 @@ func (r *Ruler) listRulesShuffleSharding(ctx context.Context) (map[string]rulesp
 	}
 
 	if len(userRings) == 0 {
+		r.ruleGroupMetrics.UpdateRuleGroupsInStore(make(map[string]int))
 		return nil, nil
 	}
 
@@ -675,6 +685,8 @@ func (r *Ruler) listRulesShuffleSharding(ctx context.Context) (map[string]rulesp
 
 	mu := sync.Mutex{}
 	result := map[string]rulespb.RuleGroupList{}
+	gLock := sync.Mutex{}
+	ruleGroupCounts := make(map[string]int, len(userRings))
 
 	concurrency := loadRulesConcurrency
 	if len(userRings) < concurrency {
@@ -690,6 +702,10 @@ func (r *Ruler) listRulesShuffleSharding(ctx context.Context) (map[string]rulesp
 					return errors.Wrapf(err, "failed to fetch rule groups for user %s", userID)
 				}
 
+				gLock.Lock()
+				ruleGroupCounts[userID] = len(groups)
+				gLock.Unlock()
+
 				filtered := filterRuleGroups(userID, groups, r.limits.DisabledRuleGroups(userID), userRings[userID], r.lifecycler.GetInstanceAddr(), r.logger, r.ringCheckErrors)
 				if len(filtered) == 0 {
 					continue
@@ -704,6 +720,7 @@ func (r *Ruler) listRulesShuffleSharding(ctx context.Context) (map[string]rulesp
 	}
 
 	err = g.Wait()
+	r.ruleGroupMetrics.UpdateRuleGroupsInStore(ruleGroupCounts)
 	return result, err
 }