Fixes early notifications after config reload (#2492)

dispatch/dispatch.go

@@ -339,6 +339,10 @@ func (d *Dispatcher) processAlert(alert *types.Alert, route *Route) {
 	d.aggrGroupsNum++
 	d.metrics.aggrGroups.Inc()
 
+	ag.syncFlushInterval(func(ctx context.Context) (time.Time, error) {
+		return d.stage.LastExecTime(ctx, d.logger)
+	})
+
 	// 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.
@@ -429,6 +433,53 @@ func (ag *aggrGroup) String() string {
 	return ag.GroupKey()
 }
 
+func (ag *aggrGroup) syncFlushInterval(n func(ctx context.Context) (time.Time, error)) {
+	ag.mtx.Lock()
+	defer ag.mtx.Unlock()
+
+	ctx, cancel := context.WithCancel(ag.ctx)
+	defer cancel()
+	ctx = ag.populateContext(ctx)
+
+	lastTime, err := n(ctx)
+	if err != nil {
+		level.Error(ag.logger).Log("msg", "error on determining last exec time", "err", err)
+		return
+	}
+
+	if lastTime.IsZero() {
+		return
+	}
+
+	ag.hasFlushed = true
+	sleepDur := calcDuration(time.Now(), lastTime, ag.opts.GroupInterval)
+	ag.next.Reset(sleepDur)
+}
+
+func calcDuration(now, lastTime time.Time, groupInterval time.Duration) time.Duration {
+	durationPassed := now.Sub(lastTime)
+
+	rest := int64(durationPassed) % int64(groupInterval)
+	if time.Duration(rest).Milliseconds() <= 1 {
+		return time.Duration(rest)
+	}
+
+	return time.Duration(groupInterval.Nanoseconds() - rest)
+}
+
+func (ag *aggrGroup) populateContext(ctx context.Context) context.Context {
+	// Populate context with information needed along the pipeline.
+	ctx = notify.WithGroupInterval(ctx, ag.opts.GroupInterval)
+	ctx = notify.WithGroupKey(ctx, ag.GroupKey())
+	ctx = notify.WithGroupLabels(ctx, ag.labels)
+	ctx = notify.WithReceiverName(ctx, ag.opts.Receiver)
+	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)
+	return ctx
+}
+
 func (ag *aggrGroup) run(nf notifyFunc) {
 	defer close(ag.done)
 	defer ag.next.Stop()
@@ -445,15 +496,7 @@ func (ag *aggrGroup) run(nf notifyFunc) {
 			// Calculating the current time directly is prone to flaky behavior,
 			// which usually only becomes apparent in tests.
 			ctx = notify.WithNow(ctx, now)
-
-			// Populate context with information needed along the pipeline.
-			ctx = notify.WithGroupKey(ctx, ag.GroupKey())
-			ctx = notify.WithGroupLabels(ctx, ag.labels)
-			ctx = notify.WithReceiverName(ctx, ag.opts.Receiver)
-			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)
+			ctx = ag.populateContext(ctx)
 
 			// Wait the configured interval before calling flush again.
 			ag.mtx.Lock()

dispatch/dispatch_test.go

@@ -609,6 +609,10 @@ func (r *recordStage) Alerts() []*types.Alert {
 	return alerts
 }
 
+func (r *recordStage) LastExecTime(ctx context.Context, l log.Logger) (time.Time, error) {
+	return time.Time{}, nil
+}
+
 func (r *recordStage) Exec(ctx context.Context, l log.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error) {
 	r.mtx.Lock()
 	defer r.mtx.Unlock()
@@ -760,3 +764,48 @@ func TestDispatcher_DoMaintenance(t *testing.T) {
 	require.False(t, isMuted)
 	require.Empty(t, mutedBy)
 }
+
+func TestCalcDuration(t *testing.T) {
+	baseTime := time.Date(2024, 5, 26, 14, 33, 12, 100, time.UTC)
+
+	tests := map[string]struct {
+		lastTime      time.Time
+		now           time.Time
+		groupInterval time.Duration
+
+		expected time.Duration
+	}{
+		"next_tick": {
+			lastTime:      baseTime,
+			now:           baseTime.Add(time.Second * 10),
+			groupInterval: time.Second * 27,
+			expected:      time.Second * 17,
+		},
+		"exact_tick": {
+			lastTime:      baseTime,
+			now:           baseTime.Add(time.Second * 27),
+			groupInterval: time.Second * 27,
+			expected:      time.Second * 0,
+		},
+		"missed_one_exact_tick": {
+			lastTime:      baseTime,
+			now:           baseTime.Add(time.Second * 27 * 2),
+			groupInterval: time.Second * 27,
+			expected:      time.Second * 0,
+		},
+		"missed_few_ticks": {
+			lastTime:      baseTime,
+			now:           baseTime.Add(time.Second * 27 * 3).Add(time.Second * 10),
+			groupInterval: time.Second * 27,
+			expected:      time.Second * 17,
+		},
+	}
+
+	for name, tc := range tests {
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			d := calcDuration(tc.now, tc.lastTime, tc.groupInterval)
+			require.Equal(t, tc.expected, d)
+		})
+	}
+}

nflog/nflog.go

@@ -377,7 +377,7 @@ func stateKey(k string, r *pb.Receiver) string {
 	return fmt.Sprintf("%s:%s", k, receiverKey(r))
 }
 
-func (l *Log) Log(r *pb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, expiry time.Duration) error {
+func (l *Log) Log(r *pb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, expiry time.Duration, dispatchTime time.Time) error {
 	// Write all st with the same timestamp.
 	now := l.now()
 	key := stateKey(gkey, r)
@@ -405,6 +405,7 @@ func (l *Log) Log(r *pb.Receiver, gkey string, firingAlerts, resolvedAlerts []ui
 			Timestamp:      now,
 			FiringAlerts:   firingAlerts,
 			ResolvedAlerts: resolvedAlerts,
+			DispatchTime:   dispatchTime,
 		},
 		ExpiresAt: expiresAt,
 	}

nflog/nflog_test.go

@@ -355,7 +355,7 @@ func TestQuery(t *testing.T) {
 	firingAlerts := []uint64{1, 2, 3}
 	resolvedAlerts := []uint64{4, 5}
 
-	err = nl.Log(recv, "key", firingAlerts, resolvedAlerts, 0)
+	err = nl.Log(recv, "key", firingAlerts, resolvedAlerts, 0, time.Time{})
 	require.NoError(t, err, "logging notification failed")
 
 	entries, err := nl.Query(QGroupKey("key"), QReceiver(recv))