From a94256621a83f5707f16b477da1f61750bb11272 Mon Sep 17 00:00:00 2001 From: Guido Trotter Date: Thu, 14 May 2026 14:40:06 -0400 Subject: [PATCH 1/2] Fix dispatcher goroutines leak on swap If an alertgroup is marked as destroyed, and we receive a new alert that would go inside it, we now swap it with a new one. When we do that we can't rely on the maintenance job to stop the old group anymore, since it won't be present anymore in the map. So upon a successful "swap" we need to cancel the previous alert group that was in the map, to avoid its goroutine continuing. In addition we also self-cancel inside the alertgroup itself, if we find it is destroyed, so that the goroutine stops running when the first of "doMaintenance", run on destroyed, or swapped out, happens. In addition we fix a more unlikely deletion from the marker in doMaintenance. Signed-off-by: Guido Trotter --- dispatch/dispatch.go | 15 +++++++++- dispatch/dispatch_test.go | 60 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 1 deletion(-) diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index 5990f13570..e1d1f2b13d 100644 --- a/dispatch/dispatch.go +++ b/dispatch/dispatch.go @@ -317,9 +317,14 @@ func (d *Dispatcher) doMaintenance() { ag := el.(*aggrGroup) if ag.destroyed() { ag.stop() - d.marker.DeleteByGroupKey(ag.routeID, ag.GroupKey()) deleted := d.routeGroupsSlice[i].groups.CompareAndDelete(ag.fingerprint(), ag) if deleted { + // Deletion from the marker should only happen if we really deleted the group. + // While it's possible that a new group with the same fingerprint is added between + // CompareAndDelete and DeleteByGroupKey, it should not have flushed in between, + // because of flush waits. A full prevention would require DeleteByGroupKey to also + // take the group as an argument and only delete if the group itself matches. + d.marker.DeleteByGroupKey(ag.routeID, ag.GroupKey()) d.routeGroupsSlice[i].groupsLen.Add(-1) d.aggrGroupsNum.Add(-1) d.metrics.aggrGroups.Set(float64(d.aggrGroupsNum.Load())) @@ -516,6 +521,9 @@ func (d *Dispatcher) groupAlert(ctx context.Context, alert *types.Alert, route * // Try to store the new group in the map. If another goroutine has already created the same group, use the existing one. swapped := d.routeGroupsSlice[route.Idx].groups.CompareAndSwap(fp, el, ag) if swapped { + // Since we swapped the new group in, we need to cancel the old one, + // as doMaintenance will not be able to find it in the map anymore. + el.(*aggrGroup).cancel() // We swapped the new group in, we can break and start it. break } @@ -735,6 +743,11 @@ func (ag *aggrGroup) run(nf notifyFunc) { cancel() + // If destroyed, exit: this particular alert group won't be used anymore. + if ag.destroyed() { + return + } + case <-ag.ctx.Done(): return } diff --git a/dispatch/dispatch_test.go b/dispatch/dispatch_test.go index 6ecfa5991a..a46e2f203a 100644 --- a/dispatch/dispatch_test.go +++ b/dispatch/dispatch_test.go @@ -899,6 +899,66 @@ func TestGroupAlert_RecoversWhenCASFails(t *testing.T) { require.Positive(t, testutil.ToFloat64(metrics.aggrGroupCreationRetries), "contended CAS path was not exercised in %d rounds — scheduler is unusually serial", rounds) } +// TestGroupAlert_DisplacedAggrGroupGoroutineExits is a regression test for a +// goroutine leak: when groupAlert CAS-replaces a destroyed aggrGroup in the +// map, the displaced group's run goroutine must be torn down. Otherwise it +// stays parked in its select forever (doMaintenance can no longer find it +// because it's been removed from the map), accumulating one stuck goroutine +// per replacement for the lifetime of the process. +func TestGroupAlert_DisplacedAggrGroupGoroutineExits(t *testing.T) { + logger := promslog.NewNopLogger() + reg := prometheus.NewRegistry() + marker := types.NewMarker(reg) + alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, 0, nil, logger, eventrecorder.NopRecorder(), reg, nil) + require.NoError(t, err) + defer alerts.Close() + + route := &Route{ + RouteOpts: RouteOpts{ + Receiver: "test", + GroupBy: map[model.LabelName]struct{}{"alertname": {}}, + GroupWait: time.Hour, // never auto-flush during the test + GroupInterval: time.Hour, + RepeatInterval: time.Hour, + }, + Idx: 0, + } + timeout := func(d time.Duration) time.Duration { return d } + recorder := &recordStage{alerts: make(map[string]map[model.Fingerprint]*types.Alert)} + dispatcher := NewDispatcher(alerts, route, recorder, marker, timeout, testMaintenanceInterval, nil, logger, eventrecorder.NopRecorder(), NewDispatcherMetrics(false, reg)) + dispatcher.routeGroupsSlice = []routeAggrGroups{{route: route}} + // WaitingToStart so groupAlert won't auto-start the new ag — keeps the + // test focused on the displaced group. + dispatcher.state.Store(DispatcherStateWaitingToStart) + + groupLabels := model.LabelSet{"alertname": "displaced"} + displaced := newAggrGroup(context.Background(), groupLabels, route, timeout, marker, eventrecorder.NopRecorder(), logger) + // Mark destroyed so groupAlert can't insert into it and is forced down + // the CAS-replace path. + require.NoError(t, displaced.alerts.DeleteIfNotModified(types.AlertSlice{}, true)) + require.True(t, displaced.destroyed()) + dispatcher.routeGroupsSlice[0].groups.Store(displaced.fingerprint(), displaced) + + // Start the run goroutine on the displaced group — this is the orphan + // candidate. Without the fix it would never exit. + go displaced.run(func(context.Context, ...*types.Alert) bool { return true }) + + // Trigger the CAS replacement. + dispatcher.groupAlert(context.Background(), newAlert(groupLabels), route) + + // The displaced group should have been swapped out for a fresh one. + el, ok := dispatcher.routeGroupsSlice[0].groups.Load(displaced.fingerprint()) + require.True(t, ok) + require.NotSame(t, displaced, el.(*aggrGroup), "destroyed group must have been replaced") + + // And its run goroutine must have exited. + select { + case <-displaced.done: + case <-time.After(2 * time.Second): + t.Fatal("displaced aggrGroup.run goroutine did not exit after CAS replacement") + } +} + func TestDispatcher_DeleteResolvedAlertsFromMarker(t *testing.T) { t.Run("successful flush deletes markers for resolved alerts", func(t *testing.T) { ctx := context.Background() From 7190028f90400f064c7748c7c52941303b33c248 Mon Sep 17 00:00:00 2001 From: Guido Trotter Date: Mon, 18 May 2026 06:00:34 -0400 Subject: [PATCH 2/2] Add a TODO about the DeleteByGroupKey call Signed-off-by: Guido Trotter --- dispatch/dispatch.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index e1d1f2b13d..013ac02329 100644 --- a/dispatch/dispatch.go +++ b/dispatch/dispatch.go @@ -319,11 +319,11 @@ func (d *Dispatcher) doMaintenance() { ag.stop() deleted := d.routeGroupsSlice[i].groups.CompareAndDelete(ag.fingerprint(), ag) if deleted { + // TODO(ultrotter, siavash): // Deletion from the marker should only happen if we really deleted the group. - // While it's possible that a new group with the same fingerprint is added between - // CompareAndDelete and DeleteByGroupKey, it should not have flushed in between, - // because of flush waits. A full prevention would require DeleteByGroupKey to also - // take the group as an argument and only delete if the group itself matches. + // Fully fixing the case where a new group with the same fingerprint is created between + // CompareAndDelete and DeleteByGroupKey would require changes to the marker interface, + // so we leave it as a fix for after landing the pending marker changes. d.marker.DeleteByGroupKey(ag.routeID, ag.GroupKey()) d.routeGroupsSlice[i].groupsLen.Add(-1) d.aggrGroupsNum.Add(-1)