inhibitor: merge cache and index so there's only one lock

inhibit/cache.go

@@ -0,0 +1,124 @@
+// Copyright The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package inhibit
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/prometheus/common/model"
+
+	"github.com/prometheus/alertmanager/alert"
+)
+
+// cache contains the runtime state of the inhibit rule.
+type cache struct {
+	equal map[model.LabelName]struct{}
+
+	mtx sync.RWMutex
+	// alerts is the map of alerts that match the source matchers of the inhibit rule.
+	alerts map[model.Fingerprint]*alert.Alert
+	// index is a map of equal label fingerprint to the set of source alert fingerprints stored
+	// in the cache.
+	index map[model.Fingerprint]model.FingerprintSet
+}
+
+func newCache(equal map[model.LabelName]struct{}) *cache {
+	return &cache{
+		equal:  equal,
+		alerts: make(map[model.Fingerprint]*alert.Alert),
+		index:  make(map[model.Fingerprint]model.FingerprintSet),
+	}
+}
+
+// fingerprintEquals returns the fingerprint of the equal labels of the given label set.
+func (c *cache) fingerprintEquals(lset model.LabelSet) model.Fingerprint {
+	equalSet := make(model.LabelSet, len(c.equal))
+	for n := range c.equal {
+		equalSet[n] = lset[n]
+	}
+	return equalSet.Fingerprint()
+}
+
+// set adds or replaces the given source alert.
+func (c *cache) set(a *alert.Alert) {
+	fp := a.Fingerprint()
+	eq := c.fingerprintEquals(a.Labels)
+
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	c.alerts[fp] = a
+	set, ok := c.index[eq]
+	if !ok {
+		set = model.FingerprintSet{}
+		c.index[eq] = set
+	}
+	set[fp] = struct{}{}
+}
+
+// find returns the fingerprint of a cached source alert that shares the equal
+// labels of lset, is active at now, and satisfies match.
+func (c *cache) find(lset model.LabelSet, now time.Time, match func(*alert.Alert) bool) (model.Fingerprint, bool) {
+	eq := c.fingerprintEquals(lset)
+
+	c.mtx.RLock()
+	defer c.mtx.RUnlock()
+
+	for fp := range c.index[eq] {
+		a := c.alerts[fp]
+		if a.ResolvedAt(now) {
+			continue
+		}
+		if !match(a) {
+			continue
+		}
+		return fp, true
+	}
+
+	return model.Fingerprint(0), false
+}
+
+func (c *cache) run(ctx context.Context, interval time.Duration) {
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			c.gc()
+		}
+	}
+}
+
+func (c *cache) gc() {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	for fp, a := range c.alerts {
+		if !a.Resolved() {
+			continue
+		}
+		delete(c.alerts, fp)
+
+		eq := c.fingerprintEquals(a.Labels)
+		set := c.index[eq]
+		delete(set, fp)
+		if len(set) == 0 {
+			delete(c.index, eq)
+		}
+	}
+}

inhibit/inhibit.go

@@ -23,7 +23,6 @@ import (
 	"github.com/prometheus/common/model"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/propagation"
 	"go.opentelemetry.io/otel/trace"
 
@@ -33,7 +32,6 @@ import (
 	"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"
 )
 
@@ -123,15 +121,8 @@ func (ih *Inhibitor) processAlert(ctx context.Context, a *alert.Alert) {
 		if r.SourceMatchers.Matches(a.Labels) {
 			attr := attribute.String("alerting.inhibit_rule.name", r.Name)
 			span.AddEvent("alert matched rule source", trace.WithAttributes(attr))
-			if err := r.scache.Set(a); err != nil {
-				message := "error on set alert"
-				ih.logger.Error(message, "err", err)
-				span.SetStatus(codes.Error, message)
-				span.RecordError(err)
-				continue
-			}
 			span.SetAttributes(attr)
-			r.sindex.Add(r.fingerprintEquals(a.Labels), a.Fingerprint())
+			r.cache.set(a)
 		}
 	}
 }
@@ -153,7 +144,7 @@ func (ih *Inhibitor) Run() {
 	runCtx, runCancel := context.WithCancel(ctx)
 
 	for _, rule := range ih.rules {
-		go rule.scache.Run(runCtx, 15*time.Minute)
+		go rule.cache.run(runCtx, 15*time.Minute)
 	}
 
 	g.Add(func() error {
@@ -256,12 +247,7 @@ type InhibitRule struct {
 	Equal map[model.LabelName]struct{}
 
 	// Cache of alerts matching source labels.
-	scache *store.Alerts
-
-	// 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.
-	// Every source alert with the same equal labels is indexed under the same key.
-	sindex *index
+	cache *cache
 }
 
 // NewInhibitRule returns a new InhibitRule based on a configuration definition.
@@ -318,32 +304,12 @@ func NewInhibitRule(cr amcommoncfg.InhibitRule) *InhibitRule {
 		equal[model.LabelName(ln)] = struct{}{}
 	}
 
-	rule := &InhibitRule{
+	return &InhibitRule{
 		Name:           cr.Name,
 		SourceMatchers: sourcem,
 		TargetMatchers: targetm,
 		Equal:          equal,
-		scache:         store.NewAlerts(),
-		sindex:         newIndex(),
-	}
-
-	rule.scache.SetGCCallback(rule.gcCallback)
-
-	return rule
-}
-
-// fingerprintEquals returns the fingerprint of the equal labels of the given label set.
-func (r *InhibitRule) fingerprintEquals(lset model.LabelSet) model.Fingerprint {
-	equalSet := make(model.LabelSet, len(r.Equal))
-	for n := range r.Equal {
-		equalSet[n] = lset[n]
-	}
-	return equalSet.Fingerprint()
-}
-
-func (r *InhibitRule) gcCallback(alerts []*alert.Alert) {
-	for _, a := range alerts {
-		r.sindex.Delete(r.fingerprintEquals(a.Labels), a.Fingerprint())
+		cache:          newCache(equal),
 	}
 }
 
@@ -352,24 +318,7 @@ func (r *InhibitRule) gcCallback(alerts []*alert.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) {
-	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 sourceFP, true
-	}
-
-	return model.Fingerprint(0), false
+	return r.cache.find(lset, now, func(a *alert.Alert) bool {
+		return !excludeTwoSidedMatch || !r.TargetMatchers.Matches(a.Labels)
+	})
 }