Alerting: Optimize cache metrics updates (#114134)
This commit is contained in:
Moustafa Baiou
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user