Repository: prometheus/alertmanager
File: dispatch/dispatch.go (Go)

Code before:
	}

	// If the group does not exist, create it. But check the limit first.
	if limit := d.limits.MaxNumberOfAggregationGroups(); limit > 0 && d.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())
		return
	}

	ag = newAggrGroup(d.ctx, groupLabels, route, d.timeout, d.marker.(types.AlertMarker), 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)

	// Check dispatcher and alert state to determine if we should start dispatching the alert.
	now := time.Now()
	switch d.state {
	case DISPATCHER_STATE_STARTING:
		if now.After(d.minDispatchTime) {
			d.logger.Debug("Setting dispatch state to: running")
			d.state = DISPATCHER_STATE_RUNNING
		} else {
			// Reset timer to the time when the dispatcher will be in running state.
			ag.resetTimer(time.Until(d.minDispatchTime))
		}
	case DISPATCHER_STATE_RUNNING:
		if alert.StartsAt.Before(now) {
			ag.logger.Debug(
				"Alert is old enough for immediate flush, resetting timer to zero",
				"alert", alert.Name(),
				"fingerprint", alert.Fingerprint(),
				"startsAt", alert.StartsAt,
			)
			ag.resetTimer(0)
		}
	default:
		d.logger.Warn("Unknow dispatcher state", "state", d.state)
	}

	go ag.run(func(ctx context.Context, alerts ...*types.Alert) bool {
		_, _, err := d.stage.Exec(ctx, d.logger, alerts...)
		if err != nil {
			logger := d.logger.With("aggrGroup", ag.GroupKey(), "num_alerts", len(alerts), "err", err)
			if errors.Is(ctx.Err(), context.Canceled) {
				// It is expected for the context to be canceled on
				// configuration reload or shutdown. In this case, the

Review comment (suggestion):
This implementation feels a little awkward to me. I have two problems with it:
1. Logically, the dispatcher transitions into the running state as soon as the `minDispatchTime` has passed. As implemented, the transition only happens when the dispatcher sees an alert after `minDispatchTime`.
2. We influence notifications indirectly by setting the aggrGroups' timers externally, rather than leaving it managed internally to the aggrGroup. This seems like something that'll become confusing over time.

I wonder if we could make this a little cleaner by decoupling the logic for the dispatcher state transition from the logic for ingesting an alert.

Perhaps we can start a timer that expires at `minDispatchTime` and make the state change then? We could wait on the timer in the main `select` inside `run`. And perhaps we can control the aggr group behavior by avoiding calling `ag.run` until `minDispatchTime`? We'd need to start any aggrGroups that exist but haven't been started, but that should be possible. 

Code after:

	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)
		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.metrics.aggrGroupLimitReached.Inc()
		d.logger.Error("Too many aggregation groups, cannot create new group for alert", "groups", d.aggrGroupsNum, "limit", limit, "alert", alert.Name())
		return
	}

	ag = newAggrGroup(d.ctx, groupLabels, route, d.timeout, d.marker.(types.AlertMarker), 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)

	if alert.StartsAt.Add(ag.opts.GroupWait).Before(now) {
		ag.logger.Debug(
			"Alert is old enough for immediate flush, resetting timer to zero",
			"alert", alert.Name(),
			"fingerprint", alert.Fingerprint(),
			"startsAt", alert.StartsAt,
		)
		ag.resetTimer(0)
	}
	// Check dispatcher and alert state to determine if we should run the AG now.
	switch d.state {
	case DispatcherStateWaitingToStart:
		d.logger.Debug("Dispatcher still waiting to start")
	case DispatcherStateRunning:
		d.runAG(ag)
	default:
		d.logger.Warn("unknown state detected", "state", "unknown")
	}
}

func (d *Dispatcher) runAG(ag *aggrGroup) {