fix(dispatch): reduce locking contention

Reduce the amount of time spent holding locks in the dispatcher by using
the new data structures:

- doMaintenance() snapshots empty groups and deletes them afterwards
- Groups() snapshots routes and queries groups per route on demand
- processAlert() only locks the group it is processing an alert for

None of the above 3 methods hold any locks directly.

This results in -68% maintenance overhead or +12991% alert processing rate improvement:
```
goos: darwin
goarch: arm64
pkg: github.com/prometheus/alertmanager/dispatch
cpu: Apple M3 Pro
                                                 │ bench-dispatch-main.txt │  bench-dispatch-nested-map-locks.txt   │
                                                 │         sec/op          │    sec/op      vs base                 │
Dispatch_100k_AggregationGroups_10k_Empty-12                  1.242µ ±  1%   1.193µ ±   5%    -3.91% (p=0.037 n=10)
Dispatch_100k_AggregationGroups_20k_Empty-12                  1.329µ ±  2%   1.188µ ±   3%   -10.65% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_30k_Empty-12                  1.437µ ±  3%   1.288µ ±   7%   -10.37% (p=0.001 n=10)
Dispatch_100k_AggregationGroups_40k_Empty-12                  1.695µ ±  8%   1.236µ ±   6%   -27.06% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_50k_Empty-12                  2.185µ ± 11%   1.286µ ±  11%   -41.17% (p=0.000 n=10)
Dispatch_20k_AggregationGroups_Groups_Impact-12          189007.708µ ±  6%   1.143µ ±  18%  -100.00% (p=0.000 n=10)
Dispatch_50k_AggregationGroups_Groups_Impact-12              602.17m ± 69%   28.09m ± 605%   -95.33% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_Groups_Impact-12               1.543 ± 29%    1.272 ±  96%         ~ (p=0.247 n=10)
geomean                                                       187.7µ         24.22µ          -87.10%

                                                 │ bench-dispatch-main.txt │       bench-dispatch-nested-map-locks.txt        │
                                                 │       alerts/sec        │     alerts/sec      vs base                      │
Dispatch_100k_AggregationGroups_10k_Empty-12             1.616M ±       1%        1.767M ±   3%         +9.36% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_20k_Empty-12             1.412M ±       2%        1.715M ±   1%        +21.45% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_30k_Empty-12             1.235M ±       4%        1.637M ±   4%        +32.58% (p=0.002 n=10)
Dispatch_100k_AggregationGroups_40k_Empty-12             950.9k ±      14%       1614.0k ±   8%        +69.73% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_50k_Empty-12             693.5k ±      18%       1370.7k ±  17%        +97.65% (p=0.000 n=10)
Dispatch_20k_AggregationGroups_Groups_Impact-12           5.586 ±      14%   2027690.000 ±   5%  +36299398.75% (p=0.000 n=10)
Dispatch_50k_AggregationGroups_Groups_Impact-12           3.277 ±      40%    742918.500 ± 167%  +22667131.12% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_Groups_Impact-12          1.424 ± 4255939%    252756.500 ±  73%  +17749654.21% (p=0.000 n=10)
geomean                                                  9.134k                   1.196M            +12991.42%

                                             │ bench-dispatch-main.txt │      bench-dispatch-nested-map-locks.txt       │
                                             │ maintenance_overhead_%  │ maintenance_overhead_%  vs base                │
Dispatch_100k_AggregationGroups_10k_Empty-12              17.185 ±  7%              5.905 ±   ?  -65.64% (p=0.002 n=10)
Dispatch_100k_AggregationGroups_20k_Empty-12               36.50 ± 11%              13.66 ± 26%  -62.58% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_30k_Empty-12               55.44 ± 13%              17.45 ± 26%  -68.52% (p=0.002 n=10)
Dispatch_100k_AggregationGroups_40k_Empty-12              125.65 ± 27%              27.03 ± 34%  -78.49% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_50k_Empty-12              172.40 ± 36%              64.64 ± 60%  -62.51% (p=0.000 n=10)
geomean                                                    59.62                    18.97        -68.17%
```

Signed-off-by: Siavash Safi <[email protected]>

Author: siavashs

dispatch/dispatch.go

@@ -21,6 +21,8 @@ import (
 	"sync"
 	"time"
 
+	"go.uber.org/atomic"
+
 	"github.com/prometheus/common/model"
 
 	"github.com/prometheus/alertmanager/notify"
@@ -41,8 +43,7 @@ type Dispatcher struct {
 	timeout func(time.Duration) time.Duration
 
 	mtx                sync.RWMutex
-	aggrGroupsPerRoute map[*Route]map[model.Fingerprint]*aggrGroup
-	aggrGroupsNum      int
+	aggrGroupsPerRoute routeGroups
 
 	done   chan struct{}
 	ctx    context.Context
@@ -84,8 +85,10 @@ func (d *Dispatcher) Run() {
 	d.done = make(chan struct{})
 
 	d.mtx.Lock()
-	d.aggrGroupsPerRoute = map[*Route]map[model.Fingerprint]*aggrGroup{}
-	d.aggrGroupsNum = 0
+	d.aggrGroupsPerRoute = routeGroups{
+		groupsNum: &atomic.Int64{},
+		limits:    d.limits,
+	}
 	d.metrics.aggrGroups.Set(0)
 	d.ctx, d.cancel = context.WithCancel(context.Background())
 	d.mtx.Unlock()
@@ -134,40 +137,58 @@ func (d *Dispatcher) run(it provider.AlertIterator) {
 }
 
 func (d *Dispatcher) doMaintenance() {
-	d.mtx.Lock()
-	defer d.mtx.Unlock()
-	for _, groups := range d.aggrGroupsPerRoute {
-		for _, ag := range groups {
+	type groupToRemove struct {
+		route *Route
+		fp    model.Fingerprint
+		ag    *aggrGroup
+	}
+
+	var toRemove []groupToRemove
+
+	// First pass: collect groups to remove
+	d.aggrGroupsPerRoute.Range(func(route *Route, groups *fingerprintGroups) bool {
+		groups.Range(func(fp model.Fingerprint, ag *aggrGroup) bool {
 			if ag.empty() {
-				ag.stop()
-				d.marker.DeleteByGroupKey(ag.routeID, ag.GroupKey())
-				delete(groups, ag.fingerprint())
-				d.aggrGroupsNum--
-				d.metrics.aggrGroups.Dec()
+				toRemove = append(toRemove, groupToRemove{route, fp, ag})
 			}
+			return true
+		})
+		return true
+	})
+
+	// Second pass: remove collected groups
+	for _, item := range toRemove {
+		item.ag.stop()
+		d.marker.DeleteByGroupKey(item.ag.routeID, item.ag.GroupKey())
+		groupsMap := d.aggrGroupsPerRoute.GetRoute(item.route)
+		if groupsMap != nil {
+			groupsMap.RemoveGroup(item.fp)
+			d.metrics.aggrGroups.Dec()
 		}
 	}
 }
 
 // Groups returns a slice of AlertGroups from the dispatcher's internal state.
 func (d *Dispatcher) Groups(routeFilter func(*Route) bool, alertFilter func(*types.Alert, time.Time) bool) (AlertGroups, map[model.Fingerprint][]string) {
-	groups := AlertGroups{}
-
-	d.mtx.RLock()
-	defer d.mtx.RUnlock()
+	// Snapshot the outer map in routeGroups to
+	//  avoid holding the read lock when dispatcher has 1000s of aggregation groups.
+	routeGroups := routeGroups{}
+	d.aggrGroupsPerRoute.Range(func(route *Route, groups *fingerprintGroups) bool {
+		routeGroups.AddRoute(route)
+		return true
+	})
 
-	// Keep a list of receivers for an alert to prevent checking each alert
-	// again against all routes. The alert has already matched against this
-	// route on ingestion.
+	// TODO: move this processing out of Dispatcher, it does not belong here.
+	alertGroups := AlertGroups{}
 	receivers := map[model.Fingerprint][]string{}
-
 	now := time.Now()
-	for route, ags := range d.aggrGroupsPerRoute {
+	routeGroups.Range(func(route *Route, _ *fingerprintGroups) bool {
 		if !routeFilter(route) {
-			continue
+			return true
 		}
 
-		for _, ag := range ags {
+		// Read inner fingerprintGroups if necessary.
+		d.aggrGroupsPerRoute.GetRoute(route).Range(func(fp model.Fingerprint, ag *aggrGroup) bool {
 			receiver := route.RouteOpts.Receiver
 			alertGroup := &AlertGroup{
 				Labels:   ag.labels,
@@ -197,22 +218,24 @@ func (d *Dispatcher) Groups(routeFilter func(*Route) bool, alertFilter func(*typ
 				filteredAlerts = append(filteredAlerts, a)
 			}
 			if len(filteredAlerts) == 0 {
-				continue
+				return true
 			}
 			alertGroup.Alerts = filteredAlerts
 
-			groups = append(groups, alertGroup)
-		}
-	}
-	sort.Sort(groups)
-	for i := range groups {
-		sort.Sort(groups[i].Alerts)
+			alertGroups = append(alertGroups, alertGroup)
+			return true
+		})
+		return true
+	})
+	sort.Sort(alertGroups)
+	for i := range alertGroups {
+		sort.Sort(alertGroups[i].Alerts)
 	}
 	for i := range receivers {
 		sort.Strings(receivers[i])
 	}
 
-	return groups, receivers
+	return alertGroups, receivers
 }
 
 // Stop the dispatcher.
@@ -236,42 +259,31 @@ func (d *Dispatcher) Stop() {
 // and inserts it.
 func (d *Dispatcher) processAlert(alert *types.Alert, route *Route) {
 	groupLabels := getGroupLabels(alert, route)
-
 	fp := groupLabels.Fingerprint()
 
-	d.mtx.Lock()
-	defer d.mtx.Unlock()
-
-	routeGroups, ok := d.aggrGroupsPerRoute[route]
-	if !ok {
-		routeGroups = map[model.Fingerprint]*aggrGroup{}
-		d.aggrGroupsPerRoute[route] = routeGroups
-	}
-
-	ag, ok := routeGroups[fp]
-	if ok {
-		ag.insert(alert)
+	routeGroups := d.aggrGroupsPerRoute.AddRoute(route)
+	group := routeGroups.GetGroup(fp)
+	if group != nil {
+		group.insert(alert)
 		return
 	}
 
-	// If the group does not exist, create it. But check the limit first.
-	if limit := d.limits.MaxNumberOfAggregationGroups(); limit > 0 && d.aggrGroupsNum >= limit {
+	// If the group does not exist, create it.
+	group, count, limit := routeGroups.AddGroup(fp, newAggrGroup(d.ctx, groupLabels, route, d.timeout, d.logger))
+	if group == nil {
+		// Rate limited.
 		d.metrics.aggrGroupLimitReached.Inc()
-		d.logger.Error("Too many aggregation groups, cannot create new group for alert", "groups", d.aggrGroupsNum, "limit", limit, "alert", alert.Name())
+		d.logger.Error("Too many aggregation groups, cannot create new group for alert", "groups", count, "limit", limit)
 		return
 	}
-
-	ag = newAggrGroup(d.ctx, groupLabels, route, d.timeout, d.logger)
-	routeGroups[fp] = ag
-	d.aggrGroupsNum++
 	d.metrics.aggrGroups.Inc()
 
 	// Insert the 1st alert in the group before starting the group's run()
 	// function, to make sure that when the run() will be executed the 1st
 	// alert is already there.
-	ag.insert(alert)
+	group.insert(alert)
 
-	go ag.run(func(ctx context.Context, alerts ...*types.Alert) bool {
+	go group.run(func(ctx context.Context, alerts ...*types.Alert) bool {
 		_, _, err := d.stage.Exec(ctx, d.logger, alerts...)
 		if err != nil {
 			logger := d.logger.With("num_alerts", len(alerts), "err", err)