#3513: Mark muted alerts

cmd/alertmanager/main.go

@@ -438,6 +438,7 @@ func run() int {
 			inhibitor,
 			silencer,
 			intervener,
+			marker,
 			notificationLog,
 			pipelinePeer,
 		)

dispatch/dispatch.go

@@ -78,6 +78,7 @@ type Dispatcher struct {
 	route   *Route
 	alerts  provider.Alerts
 	stage   notify.Stage
+	marker  types.GroupMarker
 	metrics *DispatcherMetrics
 	limits  Limits
 
@@ -107,7 +108,7 @@ func NewDispatcher(
 	ap provider.Alerts,
 	r *Route,
 	s notify.Stage,
-	mk types.AlertMarker,
+	mk types.GroupMarker,
 	to func(time.Duration) time.Duration,
 	lim Limits,
 	l log.Logger,
@@ -121,6 +122,7 @@ func NewDispatcher(
 		alerts:  ap,
 		stage:   s,
 		route:   r,
+		marker:  mk,
 		timeout: to,
 		logger:  log.With(l, "component", "dispatcher"),
 		metrics: m,
@@ -145,8 +147,8 @@ func (d *Dispatcher) Run() {
 }
 
 func (d *Dispatcher) run(it provider.AlertIterator) {
-	cleanup := time.NewTicker(30 * time.Second)
-	defer cleanup.Stop()
+	maintenance := time.NewTicker(30 * time.Second)
+	defer maintenance.Stop()
 
 	defer it.Close()
 
@@ -175,28 +177,30 @@ func (d *Dispatcher) run(it provider.AlertIterator) {
 			}
 			d.metrics.processingDuration.Observe(time.Since(now).Seconds())
 
-		case <-cleanup.C:
-			d.mtx.Lock()
-
-			for _, groups := range d.aggrGroupsPerRoute {
-				for _, ag := range groups {
-					if ag.empty() {
-						ag.stop()
-						delete(groups, ag.fingerprint())
-						d.aggrGroupsNum--
-						d.metrics.aggrGroups.Dec()
-					}
-				}
-			}
-
-			d.mtx.Unlock()
-
+		case <-maintenance.C:
+			d.doMaintenance()
 		case <-d.ctx.Done():
 			return
 		}
 	}
 }
 
+func (d *Dispatcher) doMaintenance() {
+	d.mtx.Lock()
+	for _, groups := range d.aggrGroupsPerRoute {
+		for _, ag := range groups {
+			if ag.empty() {
+				ag.stop()
+				d.marker.DeleteByGroupKey(ag.routeID, ag.GroupKey())
+				delete(groups, ag.fingerprint())
+				d.aggrGroupsNum--
+				d.metrics.aggrGroups.Dec()
+			}
+		}
+	}
+	d.mtx.Unlock()
+}
+
 // AlertGroup represents how alerts exist within an aggrGroup.
 type AlertGroup struct {
 	Alerts   types.AlertSlice
@@ -374,6 +378,7 @@ type aggrGroup struct {
 	labels   model.LabelSet
 	opts     *RouteOpts
 	logger   log.Logger
+	routeID  string
 	routeKey string
 
 	alerts  *store.Alerts
@@ -394,6 +399,7 @@ func newAggrGroup(ctx context.Context, labels model.LabelSet, r *Route, to func(
 	}
 	ag := &aggrGroup{
 		labels:   labels,
+		routeID:  r.ID(),
 		routeKey: r.Key(),
 		opts:     &r.RouteOpts,
 		timeout:  to,
@@ -447,6 +453,7 @@ func (ag *aggrGroup) run(nf notifyFunc) {
 			ctx = notify.WithRepeatInterval(ctx, ag.opts.RepeatInterval)
 			ctx = notify.WithMuteTimeIntervals(ctx, ag.opts.MuteTimeIntervals)
 			ctx = notify.WithActiveTimeIntervals(ctx, ag.opts.ActiveTimeIntervals)
+			ctx = notify.WithRouteID(ctx, ag.routeID)
 
 			// Wait the configured interval before calling flush again.
 			ag.mtx.Lock()

dispatch/dispatch_test.go

@@ -691,3 +691,48 @@ type limits struct {
 func (l limits) MaxNumberOfAggregationGroups() int {
 	return l.groups
 }
+
+func TestDispatcher_DoMaintenance(t *testing.T) {
+	r := prometheus.NewRegistry()
+	marker := types.NewMarker(r)
+
+	alerts, err := mem.NewAlerts(context.Background(), marker, time.Minute, nil, log.NewNopLogger(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	route := &Route{
+		RouteOpts: RouteOpts{
+			GroupBy:       map[model.LabelName]struct{}{"alertname": {}},
+			GroupWait:     0,
+			GroupInterval: 5 * time.Minute, // Should never hit in this test.
+		},
+	}
+	timeout := func(d time.Duration) time.Duration { return d }
+	recorder := &recordStage{alerts: make(map[string]map[model.Fingerprint]*types.Alert)}
+
+	ctx := context.Background()
+	dispatcher := NewDispatcher(alerts, route, recorder, marker, timeout, nil, log.NewNopLogger(), NewDispatcherMetrics(false, r))
+	aggrGroups := make(map[*Route]map[model.Fingerprint]*aggrGroup)
+	aggrGroups[route] = make(map[model.Fingerprint]*aggrGroup)
+
+	// Insert an aggregation group with no alerts.
+	labels := model.LabelSet{"alertname": "1"}
+	aggrGroup1 := newAggrGroup(ctx, labels, route, timeout, log.NewNopLogger())
+	aggrGroups[route][aggrGroup1.fingerprint()] = aggrGroup1
+	dispatcher.aggrGroupsPerRoute = aggrGroups
+	// Must run otherwise doMaintenance blocks on aggrGroup1.stop().
+	go aggrGroup1.run(func(context.Context, ...*types.Alert) bool { return true })
+
+	// Insert a marker for the aggregation group's group key.
+	marker.SetMuted(route.ID(), aggrGroup1.GroupKey(), []string{"weekends"})
+	mutedBy, isMuted := marker.Muted(route.ID(), aggrGroup1.GroupKey())
+	require.True(t, isMuted)
+	require.Equal(t, []string{"weekends"}, mutedBy)
+
+	// Run the maintenance and the marker should be removed.
+	dispatcher.doMaintenance()
+	mutedBy, isMuted = marker.Muted(route.ID(), aggrGroup1.GroupKey())
+	require.False(t, isMuted)
+	require.Empty(t, mutedBy)
+}

notify/notify.go

@@ -119,6 +119,7 @@ const (
 	keyNow
 	keyMuteTimeIntervals
 	keyActiveTimeIntervals
+	keyRouteID
 )
 
 // WithReceiverName populates a context with a receiver name.
@@ -165,6 +166,10 @@ func WithActiveTimeIntervals(ctx context.Context, at []string) context.Context {
 	return context.WithValue(ctx, keyActiveTimeIntervals, at)
 }
 
+func WithRouteID(ctx context.Context, routeID string) context.Context {
+	return context.WithValue(ctx, keyRouteID, routeID)
+}
+
 // RepeatInterval extracts a repeat interval from the context. Iff none exists, the
 // second argument is false.
 func RepeatInterval(ctx context.Context) (time.Duration, bool) {
@@ -228,6 +233,13 @@ func ActiveTimeIntervalNames(ctx context.Context) ([]string, bool) {
 	return v, ok
 }
 
+// RouteID extracts a RouteID from the context. Iff none exists, the
+// // second argument is false.
+func RouteID(ctx context.Context) (string, bool) {
+	v, ok := ctx.Value(keyRouteID).(string)
+	return v, ok
+}
+
 // A Stage processes alerts under the constraints of the given context.
 type Stage interface {
 	Exec(ctx context.Context, l log.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error)
@@ -384,15 +396,16 @@ func (pb *PipelineBuilder) New(
 	inhibitor *inhibit.Inhibitor,
 	silencer *silence.Silencer,
 	intervener *timeinterval.Intervener,
+	marker types.GroupMarker,
 	notificationLog NotificationLog,
 	peer Peer,
 ) RoutingStage {
 	rs := make(RoutingStage, len(receivers))
 
 	ms := NewGossipSettleStage(peer)
 	is := NewMuteStage(inhibitor, pb.metrics)
-	tas := NewTimeActiveStage(intervener, pb.metrics)
-	tms := NewTimeMuteStage(intervener, pb.metrics)
+	tas := NewTimeActiveStage(intervener, marker, pb.metrics)
+	tms := NewTimeMuteStage(intervener, marker, pb.metrics)
 	ss := NewMuteStage(silencer, pb.metrics)
 
 	for name := range receivers {
@@ -923,18 +936,29 @@ func (n SetNotifiesStage) Exec(ctx context.Context, l log.Logger, alerts ...*typ
 
 type timeStage struct {
 	muter   types.TimeMuter
+	marker  types.GroupMarker
 	metrics *Metrics
 }
 
 type TimeMuteStage timeStage
 
-func NewTimeMuteStage(m types.TimeMuter, metrics *Metrics) *TimeMuteStage {
-	return &TimeMuteStage{m, metrics}
+func NewTimeMuteStage(muter types.TimeMuter, marker types.GroupMarker, metrics *Metrics) *TimeMuteStage {
+	return &TimeMuteStage{muter, marker, metrics}
 }
 
 // Exec implements the stage interface for TimeMuteStage.
 // TimeMuteStage is responsible for muting alerts whose route is not in an active time.
 func (tms TimeMuteStage) Exec(ctx context.Context, l log.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error) {
+	routeID, ok := RouteID(ctx)
+	if !ok {
+		return ctx, nil, errors.New("route ID missing")
+	}
+
+	gkey, ok := GroupKey(ctx)
+	if !ok {
+		return ctx, nil, errors.New("group key missing")
+	}
+
 	muteTimeIntervalNames, ok := MuteTimeIntervalNames(ctx)
 	if !ok {
 		return ctx, alerts, nil
@@ -949,29 +973,42 @@ func (tms TimeMuteStage) Exec(ctx context.Context, l log.Logger, alerts ...*type
 		return ctx, alerts, nil
 	}
 
-	muted, _, err := tms.muter.Mutes(muteTimeIntervalNames, now)
+	muted, mutedBy, err := tms.muter.Mutes(muteTimeIntervalNames, now)
 	if err != nil {
 		return ctx, alerts, err
 	}
+	// If muted is false then mutedBy is nil and the muted marker is removed.
+	tms.marker.SetMuted(routeID, gkey, mutedBy)
 
 	// If the current time is inside a mute time, all alerts are removed from the pipeline.
 	if muted {
 		tms.metrics.numNotificationSuppressedTotal.WithLabelValues(SuppressedReasonMuteTimeInterval).Add(float64(len(alerts)))
 		level.Debug(l).Log("msg", "Notifications not sent, route is within mute time", "alerts", len(alerts))
 		return ctx, nil, nil
 	}
+
 	return ctx, alerts, nil
 }
 
 type TimeActiveStage timeStage
 
-func NewTimeActiveStage(m types.TimeMuter, metrics *Metrics) *TimeActiveStage {
-	return &TimeActiveStage{m, metrics}
+func NewTimeActiveStage(muter types.TimeMuter, marker types.GroupMarker, metrics *Metrics) *TimeActiveStage {
+	return &TimeActiveStage{muter, marker, metrics}
 }
 
 // Exec implements the stage interface for TimeActiveStage.
 // TimeActiveStage is responsible for muting alerts whose route is not in an active time.
 func (tas TimeActiveStage) Exec(ctx context.Context, l log.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error) {
+	routeID, ok := RouteID(ctx)
+	if !ok {
+		return ctx, nil, errors.New("route ID missing")
+	}
+
+	gkey, ok := GroupKey(ctx)
+	if !ok {
+		return ctx, nil, errors.New("group key missing")
+	}
+
 	activeTimeIntervalNames, ok := ActiveTimeIntervalNames(ctx)
 	if !ok {
 		return ctx, alerts, nil
@@ -987,13 +1024,22 @@ func (tas TimeActiveStage) Exec(ctx context.Context, l log.Logger, alerts ...*ty
 		return ctx, alerts, errors.New("missing now timestamp")
 	}
 
-	muted, _, err := tas.muter.Mutes(activeTimeIntervalNames, now)
+	active, _, err := tas.muter.Mutes(activeTimeIntervalNames, now)
 	if err != nil {
 		return ctx, alerts, err
 	}
 
+	var mutedBy []string
+	if !active {
+		// If the group is muted, then it must be muted by all active time intervals.
+		// Otherwise, the group must be in at least one active time interval for it
+		// to be active.
+		mutedBy = activeTimeIntervalNames
+	}
+	tas.marker.SetMuted(routeID, gkey, mutedBy)
+
 	// If the current time is not inside an active time, all alerts are removed from the pipeline
-	if !muted {
+	if !active {
 		tas.metrics.numNotificationSuppressedTotal.WithLabelValues(SuppressedReasonActiveTimeInterval).Add(float64(len(alerts)))
 		level.Debug(l).Log("msg", "Notifications not sent, route is not within active time", "alerts", len(alerts))
 		return ctx, nil, nil