fix(dispatch): improve locking contention

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

Author: siavashs

dispatch/dispatch.go

@@ -18,6 +18,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"maps"
 	"sort"
 	"sync"
 	"time"
@@ -185,19 +186,32 @@ func (d *Dispatcher) run(it provider.AlertIterator) {
 }
 
 func (d *Dispatcher) doMaintenance() {
+	d.mtx.RLock()
+	empty := make(map[*Route]map[model.Fingerprint]*aggrGroup)
+	for route, groups := range d.aggrGroupsPerRoute {
+		for fp, ag := range groups {
+			if ag.empty() {
+				if empty[route] == nil {
+					empty[route] = make(map[model.Fingerprint]*aggrGroup)
+				}
+				empty[route][fp] = ag
+			}
+		}
+	}
+	d.mtx.RUnlock()
 	d.mtx.Lock()
-	defer d.mtx.Unlock()
-	for _, groups := range d.aggrGroupsPerRoute {
-		for _, ag := range groups {
+	for route, groups := range empty {
+		for fp, ag := range groups {
 			if ag.empty() {
 				ag.stop()
 				d.marker.DeleteByGroupKey(ag.routeID, ag.GroupKey())
-				delete(groups, ag.fingerprint())
+				delete(d.aggrGroupsPerRoute[route], fp)
 				d.aggrGroupsNum--
 				d.metrics.aggrGroups.Dec()
 			}
 		}
 	}
+	d.mtx.Unlock()
 }
 
 // AlertGroup represents how alerts exist within an aggrGroup.
@@ -224,16 +238,19 @@ func (ag AlertGroups) Len() int { return len(ag) }
 func (d *Dispatcher) Groups(routeFilter func(*Route) bool, alertFilter func(*types.Alert, time.Time) bool) (AlertGroups, map[model.Fingerprint][]string) {
 	groups := AlertGroups{}
 
+	// Take a snapshot copy of the map to avoid blocking alert processing.
+	snapshot := make(map[*Route]map[model.Fingerprint]*aggrGroup)
 	d.mtx.RLock()
-	defer d.mtx.RUnlock()
+	maps.Copy(d.aggrGroupsPerRoute, snapshot)
+	d.mtx.RUnlock()
 
 	// 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.
 	receivers := map[model.Fingerprint][]string{}
 
 	now := time.Now()
-	for route, ags := range d.aggrGroupsPerRoute {
+	for route, ags := range snapshot {
 		if !routeFilter(route) {
 			continue
 		}
@@ -316,30 +333,34 @@ func (d *Dispatcher) processAlert(alert *types.Alert, route *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]
+	d.mtx.Unlock()
 	if ok {
 		ag.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 {
+	d.mtx.RLock()
+	aggrGroupsNum := d.aggrGroupsNum
+	d.mtx.RUnlock()
+	if limit := d.limits.MaxNumberOfAggregationGroups(); limit > 0 && aggrGroupsNum >= limit {
 		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", aggrGroupsNum, "limit", limit, "alert", alert.Name())
 		return
 	}
 
 	ag = newAggrGroup(d.ctx, groupLabels, route, d.timeout, d.logger)
+	d.mtx.Lock()
 	routeGroups[fp] = ag
 	d.aggrGroupsNum++
+	d.mtx.Unlock()
 	d.metrics.aggrGroups.Inc()
 
 	// Insert the 1st alert in the group before starting the group's run()