From 1f9ed37b1b42fc042b80569de94a346fd6382084 Mon Sep 17 00:00:00 2001 From: Moustafa Baiou Date: Wed, 19 Nov 2025 11:39:13 -0500 Subject: [PATCH] Alerting: Optimize cache metrics updates (#114134) --- pkg/services/ngalert/state/cache.go | 70 +++++++++++---- pkg/services/ngalert/state/cache_test.go | 109 +++++++++++++++++++++++ 2 files changed, 162 insertions(+), 17 deletions(-) diff --git a/pkg/services/ngalert/state/cache.go b/pkg/services/ngalert/state/cache.go index 592fb292697..c1598d4aa3b 100644 --- a/pkg/services/ngalert/state/cache.go +++ b/pkg/services/ngalert/state/cache.go @@ -25,6 +25,13 @@ type ruleStates struct { type cache struct { states map[int64]map[string]*ruleStates // orgID > alertRuleUID > stateID > state mtxStates sync.RWMutex + metrics alertMetrics +} + +type alertMetrics struct { + lastUpdate time.Time + mtx sync.RWMutex + stateCounts map[eval.State]float64 } func newCache() *cache { @@ -33,6 +40,51 @@ func newCache() *cache { } } +func (c *cache) calcMetrics(states map[eval.State]struct{}) map[eval.State]float64 { + c.mtxStates.RLock() + defer c.mtxStates.RUnlock() + counts := make(map[eval.State]float64, len(states)) + for state := range states { + counts[state] = 0 + } + for _, orgMap := range c.states { + for _, rule := range orgMap { + for _, st := range rule.states { + for state := range states { + if st.State == state { + counts[state] += 1 + } + } + } + } + } + return counts +} + +func (c *cache) updateMetrics() { + c.metrics.mtx.Lock() + defer c.metrics.mtx.Unlock() + if time.Since(c.metrics.lastUpdate) < time.Second { + return // avoid updating too frequently + } + newMetrics := c.calcMetrics(map[eval.State]struct{}{ + eval.Normal: {}, + eval.Alerting: {}, + eval.Pending: {}, + eval.Error: {}, + eval.NoData: {}, + eval.Recovering: {}, + }) + c.metrics.lastUpdate = time.Now() + c.metrics.stateCounts = newMetrics +} + +func (c *cache) countAlertsBy(state eval.State) float64 { + c.metrics.mtx.RLock() + defer c.metrics.mtx.RUnlock() + return c.metrics.stateCounts[state] +} + // RegisterMetrics registers a set of Gauges in the form of collectors for the alerts in the cache. func (c *cache) RegisterMetrics(r prometheus.Registerer) { newAlertCountByState := func(state eval.State) prometheus.GaugeFunc { @@ -43,6 +95,7 @@ func (c *cache) RegisterMetrics(r prometheus.Registerer) { Help: "How many alerts by state are in the scheduler.", ConstLabels: prometheus.Labels{"state": strings.ToLower(state.String())}, }, func() float64 { + c.updateMetrics() return c.countAlertsBy(state) }) } @@ -55,23 +108,6 @@ func (c *cache) RegisterMetrics(r prometheus.Registerer) { r.MustRegister(newAlertCountByState(eval.Recovering)) } -func (c *cache) countAlertsBy(state eval.State) float64 { - c.mtxStates.RLock() - defer c.mtxStates.RUnlock() - var count float64 - for _, orgMap := range c.states { - for _, rule := range orgMap { - for _, st := range rule.states { - if st.State == state { - count++ - } - } - } - } - - return count -} - func expandAnnotationsAndLabels(ctx context.Context, log log.Logger, alertRule *ngModels.AlertRule, result eval.Result, extraLabels data.Labels, externalURL *url.URL) (data.Labels, data.Labels) { var reserved []string resultLabels := result.Instance diff --git a/pkg/services/ngalert/state/cache_test.go b/pkg/services/ngalert/state/cache_test.go index f83c1aa9698..985e7fc1f05 100644 --- a/pkg/services/ngalert/state/cache_test.go +++ b/pkg/services/ngalert/state/cache_test.go @@ -216,6 +216,115 @@ func TestCacheMetrics(t *testing.T) { }) } +func TestCacheMetricsDebounce(t *testing.T) { + t.Run("updates after debounce window", func(t *testing.T) { + orgID := int64(1) + + reg := prometheus.NewPedanticRegistry() + cache := newCache() + + // Add initial state: 1 alerting + cache.set(&State{ + OrgID: orgID, + AlertRuleUID: "rule1", + CacheID: data.Fingerprint(rand.Int63()), + State: eval.Alerting, + }) + + cache.RegisterMetrics(reg) + + // First gather should reflect the single alerting instance + expectedInitial := ` + # HELP grafana_alerting_alerts How many alerts by state are in the scheduler. + # TYPE grafana_alerting_alerts gauge + grafana_alerting_alerts{state="alerting"} 1 + grafana_alerting_alerts{state="error"} 0 + grafana_alerting_alerts{state="nodata"} 0 + grafana_alerting_alerts{state="normal"} 0 + grafana_alerting_alerts{state="pending"} 0 + grafana_alerting_alerts{state="recovering"} 0 + ` + + err := testutil.GatherAndCompare(reg, bytes.NewBufferString(expectedInitial), "grafana_alerting_alerts") + require.NoError(t, err) + + // Modify cache immediately: add 2 more alerting and 1 error state. + // Due to debounce (1s), next gather should still return the initial values. + cache.set(&State{OrgID: orgID, AlertRuleUID: "rule1", CacheID: data.Fingerprint(rand.Int63()), State: eval.Alerting}) + cache.set(&State{OrgID: orgID, AlertRuleUID: "rule1", CacheID: data.Fingerprint(rand.Int63()), State: eval.Alerting}) + cache.set(&State{OrgID: orgID, AlertRuleUID: "rule1", CacheID: data.Fingerprint(rand.Int63()), State: eval.Error}) + + // Immediate gather should still show the initial counts because of debounce. + err = testutil.GatherAndCompare(reg, bytes.NewBufferString(expectedInitial), "grafana_alerting_alerts") + require.NoError(t, err) + + // Bypass debounce by setting lastUpdate in the past (>1s) under lock. + cache.metrics.mtx.Lock() + cache.metrics.lastUpdate = time.Now().Add(-2 * time.Second) + cache.metrics.mtx.Unlock() + + expectedAfter := ` + # HELP grafana_alerting_alerts How many alerts by state are in the scheduler. + # TYPE grafana_alerting_alerts gauge + grafana_alerting_alerts{state="alerting"} 3 + grafana_alerting_alerts{state="error"} 1 + grafana_alerting_alerts{state="nodata"} 0 + grafana_alerting_alerts{state="normal"} 0 + grafana_alerting_alerts{state="pending"} 0 + grafana_alerting_alerts{state="recovering"} 0 + ` + + err = testutil.GatherAndCompare(reg, bytes.NewBufferString(expectedAfter), "grafana_alerting_alerts") + require.NoError(t, err) + }) + + t.Run("no update within debounce window", func(t *testing.T) { + orgID := int64(1) + + reg := prometheus.NewPedanticRegistry() + cache := newCache() + + // Seed with one alerting state + cache.set(&State{OrgID: orgID, AlertRuleUID: "rule1", CacheID: data.Fingerprint(rand.Int63()), State: eval.Alerting}) + + cache.RegisterMetrics(reg) + + // Initial gather populates metrics (1 alerting) + expectedInitial := ` + # HELP grafana_alerting_alerts How many alerts by state are in the scheduler. + # TYPE grafana_alerting_alerts gauge + grafana_alerting_alerts{state="alerting"} 1 + grafana_alerting_alerts{state="error"} 0 + grafana_alerting_alerts{state="nodata"} 0 + grafana_alerting_alerts{state="normal"} 0 + grafana_alerting_alerts{state="pending"} 0 + grafana_alerting_alerts{state="recovering"} 0 + ` + err := testutil.GatherAndCompare(reg, bytes.NewBufferString(expectedInitial), "grafana_alerting_alerts") + require.NoError(t, err) + + // Add more states that would change counts if recalculated + cache.set(&State{OrgID: orgID, AlertRuleUID: "rule1", CacheID: data.Fingerprint(rand.Int63()), State: eval.Alerting}) + cache.set(&State{OrgID: orgID, AlertRuleUID: "rule1", CacheID: data.Fingerprint(rand.Int63()), State: eval.Error}) + + // Force debounce window by setting lastUpdate to now, capture it + cache.metrics.mtx.Lock() + ts := time.Now() + cache.metrics.lastUpdate = ts + cache.metrics.mtx.Unlock() + + // Gather should NOT update metrics due to debounce; counts should remain initial + err = testutil.GatherAndCompare(reg, bytes.NewBufferString(expectedInitial), "grafana_alerting_alerts") + require.NoError(t, err) + + // Confirm lastUpdate did not change (no metrics refresh happened) + cache.metrics.mtx.RLock() + tsAfter := cache.metrics.lastUpdate + cache.metrics.mtx.RUnlock() + require.True(t, tsAfter.Equal(ts), "expected lastUpdate to remain unchanged within debounce window") + }) +} + func randomSate(ruleKey models.AlertRuleKey) State { return State{ OrgID: ruleKey.OrgID,