Skip to content

update inhibitor index to store all matched alerts #5542

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Spaceman1701 wants to merge 1 commit into
base: main
Choose a base branch
from
+140 −84
Draft

update inhibitor index to store all matched alerts #5542

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 33 additions & 11 deletions inhibit/index.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,41 +19,63 @@ import (
"github.com/prometheus/common/model"
)

// index contains map of fingerprints to fingerprints.
// index contains map of fingerprints to sets of fingerprints.
// The keys are fingerprints of the equal labels of source alerts.
// The values are fingerprints of the source alerts.
// The values are the fingerprints of all source alerts sharing those equal labels.
// For more info see comments on inhibitor and InhibitRule.
type index struct {
mtx sync.RWMutex
items map[model.Fingerprint]model.Fingerprint
items map[model.Fingerprint]model.FingerprintSet
}

func newIndex() *index {
return &index{
items: make(map[model.Fingerprint]model.Fingerprint),
items: make(map[model.Fingerprint]model.FingerprintSet),
}
}

func (c *index) Get(key model.Fingerprint) (model.Fingerprint, bool) {
// Get returns a copy of the fingerprints indexed under key.
func (c *index) Get(key model.Fingerprint) ([]model.Fingerprint, bool) {
c.mtx.RLock()
defer c.mtx.RUnlock()

fp, ok := c.items[key]
return fp, ok
set, ok := c.items[key]
if !ok {
return nil, false
}
fps := make([]model.Fingerprint, 0, len(set))
for fp := range set {
fps = append(fps, fp)
}
return fps, true
}

func (c *index) Set(key, value model.Fingerprint) {
func (c *index) Add(key, value model.Fingerprint) {
c.mtx.Lock()
defer c.mtx.Unlock()

c.items[key] = value
set, ok := c.items[key]
if !ok {
set = model.FingerprintSet{}
c.items[key] = set
}
set[value] = struct{}{}
}

func (c *index) Delete(key model.Fingerprint) {
// Delete removes value from the set of fingerprints indexed under key,
// removing the key entirely once the set is empty.
func (c *index) Delete(key, value model.Fingerprint) {
c.mtx.Lock()
defer c.mtx.Unlock()

delete(c.items, key)
set, ok := c.items[key]
if !ok {
return
}
delete(set, value)
if len(set) == 0 {
delete(c.items, key)
}
}

func (c *index) Len() int {
Expand Down
91 changes: 23 additions & 68 deletions inhibit/inhibit.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,14 +27,14 @@ import (
"go.opentelemetry.io/otel/propagation"
"go.opentelemetry.io/otel/trace"

"github.com/prometheus/alertmanager/alert"
amcommoncfg "github.com/prometheus/alertmanager/config/common"
"github.com/prometheus/alertmanager/eventrecorder"
"github.com/prometheus/alertmanager/marker"
"github.com/prometheus/alertmanager/pkg/labels"
"github.com/prometheus/alertmanager/provider"
"github.com/prometheus/alertmanager/store"
"github.com/prometheus/alertmanager/tracing"
"github.com/prometheus/alertmanager/types"
)

var tracer = tracing.NewTracer("github.com/prometheus/alertmanager/inhibit")
Expand Down Expand Up @@ -108,7 +108,7 @@ func (ih *Inhibitor) run(ctx context.Context) {
}
}

func (ih *Inhibitor) processAlert(ctx context.Context, a *types.Alert) {
func (ih *Inhibitor) processAlert(ctx context.Context, a *alert.Alert) {
_, span := tracer.Start(ctx, "inhibit.Inhibitor.processAlert",
trace.WithAttributes(
attribute.String("alerting.alert.name", a.Name()),
Expand All @@ -131,7 +131,7 @@ func (ih *Inhibitor) processAlert(ctx context.Context, a *types.Alert) {
continue
}
span.SetAttributes(attr)
r.updateIndex(a)
r.sindex.Add(r.fingerprintEquals(a.Labels), a.Fingerprint())
}
}
}
Expand Down Expand Up @@ -258,10 +258,9 @@ type InhibitRule struct {
// Cache of alerts matching source labels.
scache *store.Alerts

// Index of fingerprints of source alert equal labels to fingerprint of source alert.
// Index of fingerprints of source alert equal labels to fingerprints of source alerts.
// The index helps speed up source alert lookups from scache significantely in scenarios with 100s of source alerts cached.
// The index items might overwrite eachother if multiple source alerts have exact equal labels.
// Overwrites only happen if the new source alert has bigger EndsAt value.
// Every source alert with the same equal labels is indexed under the same key.
sindex *index
}

Expand Down Expand Up @@ -342,64 +341,9 @@ func (r *InhibitRule) fingerprintEquals(lset model.LabelSet) model.Fingerprint {
return equalSet.Fingerprint()
}

// updateIndex updates the source alert index if necessary.
func (r *InhibitRule) updateIndex(alert *types.Alert) {
fp := alert.Fingerprint()
// Calculate source labelset subset which is in equals.
eq := r.fingerprintEquals(alert.Labels)

// Check if the equal labelset is already in the index.
indexed, ok := r.sindex.Get(eq)
if !ok {
// If not, add it.
r.sindex.Set(eq, fp)
return
}
// If the indexed fingerprint is the same as the new fingerprint, do nothing.
if indexed == fp {
return
}

// New alert and existing index are not the same, compare them.
existing, err := r.scache.Get(indexed)
if err != nil {
// failed to get the existing alert, overwrite the index.
r.sindex.Set(eq, fp)
return
}

// If the new alert resolves after the existing alert, replace the index.
if existing.ResolvedAt(alert.EndsAt) {
r.sindex.Set(eq, fp)
return
}
// If the existing alert resolves after the new alert, do nothing.
}

// findEqualSourceAlert returns the source alert that matches the equal labels of the given label set.
func (r *InhibitRule) findEqualSourceAlert(lset model.LabelSet, now time.Time) (*types.Alert, bool) {
equalsFP := r.fingerprintEquals(lset)
sourceFP, ok := r.sindex.Get(equalsFP)
if ok {
alert, err := r.scache.Get(sourceFP)
if err != nil {
return nil, false
}

if alert.ResolvedAt(now) {
return nil, false
}

return alert, true
}

return nil, false
}

func (r *InhibitRule) gcCallback(alerts []*types.Alert) {
func (r *InhibitRule) gcCallback(alerts []*alert.Alert) {
for _, a := range alerts {
fp := r.fingerprintEquals(a.Labels)
r.sindex.Delete(fp)
r.sindex.Delete(r.fingerprintEquals(a.Labels), a.Fingerprint())
}
}

Expand All @@ -408,12 +352,23 @@ func (r *InhibitRule) gcCallback(alerts []*types.Alert) {
// is returned. If excludeTwoSidedMatch is true, alerts that match both the
// source and the target side of the rule are disregarded.
func (r *InhibitRule) hasEqual(lset model.LabelSet, excludeTwoSidedMatch bool, now time.Time) (model.Fingerprint, bool) {
equal, found := r.findEqualSourceAlert(lset, now)
if found {
if excludeTwoSidedMatch && r.TargetMatchers.Matches(equal.Labels) {
return model.Fingerprint(0), false
sourceFPs, ok := r.sindex.Get(r.fingerprintEquals(lset))
if !ok {
return model.Fingerprint(0), false
}

for _, sourceFP := range sourceFPs {
a, err := r.scache.Get(sourceFP)
if err != nil {
continue
}
if a.ResolvedAt(now) {
continue
}
if excludeTwoSidedMatch && r.TargetMatchers.Matches(a.Labels) {
continue
}
return equal.Fingerprint(), found
return sourceFP, true
}

return model.Fingerprint(0), false
Expand Down
89 changes: 84 additions & 5 deletions inhibit/inhibit_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,7 +155,7 @@ func TestInhibitRuleHasEqual(t *testing.T) {
}
for _, v := range c.initial {
r.scache.Set(v)
r.updateIndex(v)
r.sindex.Add(r.fingerprintEquals(v.Labels), v.Fingerprint())
}

if _, have := r.hasEqual(c.input, false, time.Now()); have != c.result {
Expand Down Expand Up @@ -201,12 +201,12 @@ func TestInhibitRuleMatches(t *testing.T) {
ih.rules[0].scache = store.NewAlerts()
ih.rules[0].scache.Set(sourceAlert1)
ih.rules[0].sindex = newIndex()
ih.rules[0].updateIndex(sourceAlert1)
ih.rules[0].sindex.Add(ih.rules[0].fingerprintEquals(sourceAlert1.Labels), sourceAlert1.Fingerprint())

ih.rules[1].scache = store.NewAlerts()
ih.rules[1].scache.Set(sourceAlert2)
ih.rules[1].sindex = newIndex()
ih.rules[1].updateIndex(sourceAlert2)
ih.rules[1].sindex.Add(ih.rules[1].fingerprintEquals(sourceAlert2.Labels), sourceAlert2.Fingerprint())

cases := []struct {
target model.LabelSet
Expand Down Expand Up @@ -299,12 +299,12 @@ func TestInhibitRuleMatchers(t *testing.T) {
ih.rules[0].scache = store.NewAlerts()
ih.rules[0].scache.Set(sourceAlert1)
ih.rules[0].sindex = newIndex()
ih.rules[0].updateIndex(sourceAlert1)
ih.rules[0].sindex.Add(ih.rules[0].fingerprintEquals(sourceAlert1.Labels), sourceAlert1.Fingerprint())

ih.rules[1].scache = store.NewAlerts()
ih.rules[1].scache.Set(sourceAlert2)
ih.rules[1].sindex = newIndex()
ih.rules[1].updateIndex(sourceAlert2)
ih.rules[1].sindex.Add(ih.rules[1].fingerprintEquals(sourceAlert2.Labels), sourceAlert2.Fingerprint())

cases := []struct {
target model.LabelSet
Expand Down Expand Up @@ -597,6 +597,85 @@ func TestInhibitRule_fingerprintEquals(t *testing.T) {
require.NotEqual(t, fp, rule.fingerprintEquals(lset3))
}

func TestInhibitRuleIndexSurvivesGC(t *testing.T) {
now := time.Now()
r := &InhibitRule{
Equal: map[model.LabelName]struct{}{"cluster": {}},
scache: store.NewAlerts(),
sindex: newIndex(),
}
r.scache.SetGCCallback(r.gcCallback)

active := &alert.Alert{Alert: model.Alert{
Labels: model.LabelSet{"alertname": "S1", "cluster": "c1"},
StartsAt: now.Add(-time.Hour),
EndsAt: now.Add(2 * time.Hour),
}}
resolved := &alert.Alert{Alert: model.Alert{
Labels: model.LabelSet{"alertname": "S2", "cluster": "c1"},
StartsAt: now.Add(-time.Hour),
EndsAt: now.Add(-time.Minute),
}}
for _, a := range []*alert.Alert{active, resolved} {
require.NoError(t, r.scache.Set(a))
r.sindex.Add(r.fingerprintEquals(a.Labels), a.Fingerprint())
}

target := model.LabelSet{"alertname": "T", "cluster": "c1"}
fp, ok := r.hasEqual(target, false, now)
require.True(t, ok)
require.Equal(t, active.Fingerprint(), fp)

deleted := r.scache.GC()
require.Len(t, deleted, 1)
require.Equal(t, resolved.Fingerprint(), deleted[0].Fingerprint())

fp, ok = r.hasEqual(target, false, now)
require.True(t, ok, "active source alert must still inhibit after GC of a sibling")
require.Equal(t, active.Fingerprint(), fp)
require.Equal(t, 1, r.sindex.Len())

active.EndsAt = now.Add(-time.Second)
require.NoError(t, r.scache.Set(active))
deleted = r.scache.GC()
require.Len(t, deleted, 1)
_, ok = r.hasEqual(target, false, now)
require.False(t, ok)
require.Equal(t, 0, r.sindex.Len(), "empty index keys must be removed")
}

func TestInhibitRuleTwoSidedDoesNotShadow(t *testing.T) {
now := time.Now()
targetMatcher, err := labels.NewMatcher(labels.MatchEqual, "severity", "warning")
require.NoError(t, err)
r := &InhibitRule{
Equal: map[model.LabelName]struct{}{"cluster": {}},
TargetMatchers: labels.Matchers{targetMatcher},
scache: store.NewAlerts(),
sindex: newIndex(),
}

sourceOnly := &alert.Alert{Alert: model.Alert{
Labels: model.LabelSet{"alertname": "S1", "cluster": "c1", "severity": "critical"},
StartsAt: now.Add(-time.Hour),
EndsAt: now.Add(time.Hour),
}}
twoSided := &alert.Alert{Alert: model.Alert{
Labels: model.LabelSet{"alertname": "S2", "cluster": "c1", "severity": "warning"},
StartsAt: now.Add(-time.Hour),
EndsAt: now.Add(2 * time.Hour),
}}
for _, a := range []*alert.Alert{sourceOnly, twoSided} {
require.NoError(t, r.scache.Set(a))
r.sindex.Add(r.fingerprintEquals(a.Labels), a.Fingerprint())
}

target := model.LabelSet{"alertname": "T", "cluster": "c1", "severity": "warning"}
fp, ok := r.hasEqual(target, true, now)
require.True(t, ok)
require.Equal(t, sourceOnly.Fingerprint(), fp)
}

func BenchmarkFingerprintEquals(b *testing.B) {
// Test fingerprintEquals with varying number of equal labels
for _, numLabels := range []int{1, 3, 5, 10} {
Expand Down
Loading