feat(dispatch): add group markers

This change adds group markers to the dispatcher.
Group markers replace the global marker and are used to track
the state of alerts in an aggregation group.

This change touches many components of the alertmanager.
Group markers are passed to the notifiers and then inhibitor
and silencer using context.

The API has no breaking changes:
- /alerts uses a temporary marker to track the state of alerts
- /alerts/groups returns the group markers

The metrics are also updated to use group markers.
The `alertmanager_alerts` metric is moved to dispatcher.
The `alertmanager_marked_alerts` metric is removed.
By default it behaves the same as before, by aggregating
all alerts in the groups.
Enabling `group-key-in-metrics` flag will cause the metrics
to be grouped by `group_key`.

Signed-off-by: Siavash Safi <siavash@cloudflare.com>

Author: siavashs

api/api.go

@@ -59,9 +59,6 @@ type Options struct {
 	Alerts provider.Alerts
 	// Silences to be used by the API. Mandatory.
 	Silences *silence.Silences
-	// AlertStatusFunc is used be the API to retrieve the AlertStatus of an
-	// alert. Mandatory.
-	AlertStatusFunc func(model.Fingerprint) types.AlertStatus
 	// GroupMutedFunc is used be the API to know if an alert is muted.
 	// Mandatory.
 	GroupMutedFunc func(routeID, groupKey string) ([]string, bool)
@@ -95,9 +92,6 @@ func (o Options) validate() error {
 	if o.Silences == nil {
 		return errors.New("mandatory field Silences not set")
 	}
-	if o.AlertStatusFunc == nil {
-		return errors.New("mandatory field AlertStatusFunc not set")
-	}
 	if o.GroupMutedFunc == nil {
 		return errors.New("mandatory field GroupMutedFunc not set")
 	}
@@ -125,7 +119,6 @@ func New(opts Options) (*API, error) {
 	v2, err := apiv2.NewAPI(
 		opts.Alerts,
 		opts.GroupFunc,
-		opts.AlertStatusFunc,
 		opts.GroupMutedFunc,
 		opts.Silences,
 		opts.Peer,

api/v2/api.go

@@ -36,6 +36,7 @@ import (
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/codes"
 
+	"github.com/prometheus/alertmanager/alert"
 	"github.com/prometheus/alertmanager/api/metrics"
 	open_api_models "github.com/prometheus/alertmanager/api/v2/models"
 	"github.com/prometheus/alertmanager/api/v2/restapi"
@@ -48,12 +49,12 @@ import (
 	"github.com/prometheus/alertmanager/cluster"
 	"github.com/prometheus/alertmanager/config"
 	"github.com/prometheus/alertmanager/dispatch"
+	"github.com/prometheus/alertmanager/marker"
 	"github.com/prometheus/alertmanager/matcher/compat"
 	"github.com/prometheus/alertmanager/pkg/labels"
 	"github.com/prometheus/alertmanager/provider"
 	"github.com/prometheus/alertmanager/silence"
 	"github.com/prometheus/alertmanager/silence/silencepb"
-	"github.com/prometheus/alertmanager/types"
 )
 
 var tracer = otel.Tracer("github.com/prometheus/alertmanager/api/v2")
@@ -64,7 +65,6 @@ type API struct {
 	silences       *silence.Silences
 	alerts         provider.Alerts
 	alertGroups    groupsFn
-	getAlertStatus getAlertStatusFn
 	groupMutedFunc groupMutedFunc
 	uptime         time.Time
 
@@ -83,17 +83,15 @@ type API struct {
 }
 
 type (
-	groupsFn         func(context.Context, func(*dispatch.Route) bool, func(*types.Alert, time.Time) bool) (dispatch.AlertGroups, map[prometheus_model.Fingerprint][]string, error)
+	groupsFn         func(context.Context, func(*dispatch.Route) bool, func(*alert.Alert, time.Time) bool) (dispatch.AlertGroups, map[prometheus_model.Fingerprint][]string, error)
 	groupMutedFunc   func(routeID, groupKey string) ([]string, bool)
-	getAlertStatusFn func(prometheus_model.Fingerprint) types.AlertStatus
 	setAlertStatusFn func(ctx context.Context, labels prometheus_model.LabelSet)
 )
 
 // NewAPI returns a new Alertmanager API v2.
 func NewAPI(
 	alerts provider.Alerts,
 	gf groupsFn,
-	asf getAlertStatusFn,
 	gmf groupMutedFunc,
 	silences *silence.Silences,
 	peer cluster.ClusterPeer,
@@ -102,7 +100,6 @@ func NewAPI(
 ) (*API, error) {
 	api := API{
 		alerts:         alerts,
-		getAlertStatus: asf,
 		alertGroups:    gf,
 		groupMutedFunc: gmf,
 		peer:           peer,
@@ -281,7 +278,8 @@ func (api *API) getAlertsHandler(params alert_ops.GetAlertsParams) middleware.Re
 	alerts := api.alerts.GetPending()
 	defer alerts.Close()
 
-	alertFilter := api.alertFilter(matchers, *params.Silenced, *params.Inhibited, *params.Active)
+	tempMarker := marker.NewAlertMarker()
+	alertFilter := api.alertFilter(matchers, *params.Silenced, *params.Inhibited, *params.Active, tempMarker)
 	now := time.Now()
 
 	api.mtx.RLock()
@@ -308,7 +306,7 @@ func (api *API) getAlertsHandler(params alert_ops.GetAlertsParams) middleware.Re
 			continue
 		}
 
-		openAlert := AlertToOpenAPIAlert(alert, api.getAlertStatus(alert.Fingerprint()), receivers, nil)
+		openAlert := AlertToOpenAPIAlert(alert, tempMarker.Status(alert.Fingerprint()), receivers, nil)
 
 		res = append(res, openAlert)
 	}
@@ -365,7 +363,7 @@ func (api *API) postAlertsHandler(params alert_ops.PostAlertsParams) middleware.
 
 	// Make a best effort to insert all alerts that are valid.
 	var (
-		validAlerts    = make([]*types.Alert, 0, len(alerts))
+		validAlerts    = make([]*alert.Alert, 0, len(alerts))
 		validationErrs error
 	)
 	for _, a := range alerts {
@@ -432,7 +430,7 @@ func (api *API) getAlertGroupsHandler(params alertgroup_ops.GetAlertGroupsParams
 		}
 	}(receiverFilter)
 
-	af := api.alertFilter(matchers, *params.Silenced, *params.Inhibited, *params.Active)
+	af := api.alertFilter(matchers, *params.Silenced, *params.Inhibited, *params.Active, marker.NewAlertMarker())
 	alertGroups, allReceivers, err := api.alertGroups(ctx, rf, af)
 	if err != nil {
 		message := "Failed to get alert groups"
@@ -459,7 +457,7 @@ func (api *API) getAlertGroupsHandler(params alertgroup_ops.GetAlertGroupsParams
 		for _, alert := range alertGroup.Alerts {
 			fp := alert.Fingerprint()
 			receivers := allReceivers[fp]
-			status := api.getAlertStatus(fp)
+			status := alertGroup.AlertStatuses[fp]
 			apiAlert := AlertToOpenAPIAlert(alert, status, receivers, mutedBy)
 			ag.Alerts = append(ag.Alerts, apiAlert)
 		}
@@ -469,8 +467,8 @@ func (api *API) getAlertGroupsHandler(params alertgroup_ops.GetAlertGroupsParams
 	return alertgroup_ops.NewGetAlertGroupsOK().WithPayload(res)
 }
 
-func (api *API) alertFilter(matchers []*labels.Matcher, silenced, inhibited, active bool) func(a *types.Alert, now time.Time) bool {
-	return func(a *types.Alert, now time.Time) bool {
+func (api *API) alertFilter(matchers []*labels.Matcher, silenced, inhibited, active bool, m marker.AlertMarker) func(a *alert.Alert, now time.Time) bool {
+	return func(a *alert.Alert, now time.Time) bool {
 		ctx, span := tracer.Start(context.Background(), "alertFilter")
 		defer span.End()
 
@@ -479,12 +477,14 @@ func (api *API) alertFilter(matchers []*labels.Matcher, silenced, inhibited, act
 		}
 
 		// Set alert's current status based on its label set.
+		// The inhibitor and silencer write to m via the context.
+		ctx = marker.WithAlertMarker(ctx, m)
 		api.setAlertStatus(ctx, a.Labels)
 
 		// Get alert's current status after seeing if it is suppressed.
-		status := api.getAlertStatus(a.Fingerprint())
+		status := m.Status(a.Fingerprint())
 
-		if !active && status.State == types.AlertStateActive {
+		if !active && status.State == alert.AlertStateActive {
 			return false
 		}
 

api/v2/api_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
+	"github.com/prometheus/alertmanager/alert"
 	open_api_models "github.com/prometheus/alertmanager/api/v2/models"
 	general_ops "github.com/prometheus/alertmanager/api/v2/restapi/operations/general"
 	receiver_ops "github.com/prometheus/alertmanager/api/v2/restapi/operations/receiver"
@@ -40,7 +41,6 @@ import (
 	"github.com/prometheus/alertmanager/pkg/labels"
 	"github.com/prometheus/alertmanager/silence"
 	"github.com/prometheus/alertmanager/silence/silencepb"
-	"github.com/prometheus/alertmanager/types"
 )
 
 // If api.peers == nil, Alertmanager cluster feature is disabled. Make sure to
@@ -494,15 +494,15 @@ func TestAlertToOpenAPIAlert(t *testing.T) {
 		fp        = "0223b772b51c29e1"
 		receivers = []string{"receiver1", "receiver2"}
 
-		alert = &types.Alert{
+		a = &alert.Alert{
 			Alert: model.Alert{
 				Labels:   model.LabelSet{"severity": "critical", "alertname": "alert1"},
 				StartsAt: start,
 			},
 			UpdatedAt: updated,
 		}
 	)
-	openAPIAlert := AlertToOpenAPIAlert(alert, types.AlertStatus{State: types.AlertStateActive}, receivers, nil)
+	openAPIAlert := AlertToOpenAPIAlert(a, alert.AlertStatus{State: alert.AlertStateActive}, receivers, nil)
 	require.Equal(t, &open_api_models.GettableAlert{
 		Annotations: open_api_models.LabelSet{},
 		Alert: open_api_models.Alert{

api/v2/compat.go

@@ -22,10 +22,10 @@ import (
 	prometheus_model "github.com/prometheus/common/model"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
+	"github.com/prometheus/alertmanager/alert"
 	open_api_models "github.com/prometheus/alertmanager/api/v2/models"
 	"github.com/prometheus/alertmanager/silence"
 	"github.com/prometheus/alertmanager/silence/silencepb"
-	"github.com/prometheus/alertmanager/types"
 )
 
 // GettableSilenceFromProto converts *silencepb.Silence to open_api_models.GettableSilence.
@@ -139,7 +139,7 @@ func PostableSilenceToProto(s *open_api_models.PostableSilence) (*silencepb.Sile
 }
 
 // AlertToOpenAPIAlert converts internal alerts, alert types, and receivers to *open_api_models.GettableAlert.
-func AlertToOpenAPIAlert(alert *types.Alert, status types.AlertStatus, receivers, mutedBy []string) *open_api_models.GettableAlert {
+func AlertToOpenAPIAlert(alert *alert.Alert, status alert.AlertStatus, receivers, mutedBy []string) *open_api_models.GettableAlert {
 	startsAt := strfmt.DateTime(alert.StartsAt)
 	updatedAt := strfmt.DateTime(alert.UpdatedAt)
 	endsAt := strfmt.DateTime(alert.EndsAt)
@@ -191,14 +191,14 @@ func AlertToOpenAPIAlert(alert *types.Alert, status types.AlertStatus, receivers
 	return aa
 }
 
-// OpenAPIAlertsToAlerts converts open_api_models.PostableAlerts to []*types.Alert.
-func OpenAPIAlertsToAlerts(ctx context.Context, apiAlerts open_api_models.PostableAlerts) []*types.Alert {
+// OpenAPIAlertsToAlerts converts open_api_models.PostableAlerts to []*alert.Alert.
+func OpenAPIAlertsToAlerts(ctx context.Context, apiAlerts open_api_models.PostableAlerts) []*alert.Alert {
 	_, span := tracer.Start(ctx, "OpenAPIAlertsToAlerts")
 	defer span.End()
 
-	alerts := []*types.Alert{}
+	alerts := []*alert.Alert{}
 	for _, apiAlert := range apiAlerts {
-		alert := types.Alert{
+		alert := alert.Alert{
 			Alert: prometheus_model.Alert{
 				Labels:       APILabelSetToModelLabelSet(apiAlert.Labels),
 				Annotations:  APILabelSetToModelLabelSet(apiAlert.Annotations),

cmd/alertmanager/main.go

@@ -46,13 +46,15 @@ import (
 	webflag "github.com/prometheus/exporter-toolkit/web/kingpinflag"
 	"go.uber.org/automaxprocs/maxprocs"
 
+	"github.com/prometheus/alertmanager/alert"
 	"github.com/prometheus/alertmanager/api"
 	"github.com/prometheus/alertmanager/cluster"
 	"github.com/prometheus/alertmanager/config"
 	"github.com/prometheus/alertmanager/config/receiver"
 	"github.com/prometheus/alertmanager/dispatch"
 	"github.com/prometheus/alertmanager/featurecontrol"
 	"github.com/prometheus/alertmanager/inhibit"
+	"github.com/prometheus/alertmanager/marker"
 	"github.com/prometheus/alertmanager/matcher/compat"
 	"github.com/prometheus/alertmanager/nflog"
 	"github.com/prometheus/alertmanager/notify"
@@ -61,7 +63,6 @@ import (
 	"github.com/prometheus/alertmanager/template"
 	"github.com/prometheus/alertmanager/timeinterval"
 	"github.com/prometheus/alertmanager/tracing"
-	"github.com/prometheus/alertmanager/types"
 	"github.com/prometheus/alertmanager/ui"
 )
 
@@ -291,7 +292,7 @@ func run() int {
 		notificationLog.Maintenance(*maintenanceInterval, filepath.Join(*dataDir, "nflog"), stopc, nil)
 	})
 
-	marker := types.NewMarker(prometheus.DefaultRegisterer)
+	marker := marker.NewGroupMarker()
 
 	silenceOpts := silence.Options{
 		SnapshotFile: filepath.Join(*dataDir, "silences"),
@@ -324,7 +325,7 @@ func run() int {
 		wg.Wait()
 	}()
 
-	silencer := silence.NewSilencer(silences, marker, logger)
+	silencer := silence.NewSilencer(silences, logger)
 
 	// Peer state listeners have been registered, now we can join and get the initial state.
 	if peer != nil {
@@ -347,7 +348,6 @@ func run() int {
 
 	alerts, err := mem.NewAlerts(
 		context.Background(),
-		marker,
 		*alertGCInterval,
 		*perAlertNameLimit,
 		silencer,
@@ -366,7 +366,7 @@ func run() int {
 		disp.Load().Stop()
 	}()
 
-	groupFn := func(ctx context.Context, routeFilter func(*dispatch.Route) bool, alertFilter func(*types.Alert, time.Time) bool) (dispatch.AlertGroups, map[model.Fingerprint][]string, error) {
+	groupFn := func(ctx context.Context, routeFilter func(*dispatch.Route) bool, alertFilter func(*alert.Alert, time.Time) bool) (dispatch.AlertGroups, map[model.Fingerprint][]string, error) {
 		return disp.Load().Groups(ctx, routeFilter, alertFilter)
 	}
 
@@ -381,7 +381,6 @@ func run() int {
 	api, err := api.New(api.Options{
 		Alerts:          alerts,
 		Silences:        silences,
-		AlertStatusFunc: marker.Status,
 		GroupMutedFunc:  marker.Muted,
 		Peer:            clusterPeer,
 		Timeout:         *httpTimeout,
@@ -421,7 +420,7 @@ func run() int {
 		tmpl      *template.Template
 	)
 
-	dispMetrics := dispatch.NewDispatcherMetrics(false, prometheus.DefaultRegisterer)
+	dispMetrics := dispatch.NewDispatcherMetrics(false, prometheus.DefaultRegisterer, ff)
 	pipelineBuilder := notify.NewPipelineBuilder(prometheus.DefaultRegisterer, ff)
 	configLogger := logger.With("component", "configuration")
 	configCoordinator := config.NewCoordinator(
@@ -476,7 +475,7 @@ func run() int {
 		inhibitor.Load().Stop()
 		disp.Load().Stop()
 
-		newInhibitor := inhibit.NewInhibitor(alerts, conf.InhibitRules, marker, logger)
+		newInhibitor := inhibit.NewInhibitor(alerts, conf.InhibitRules, logger)
 		inhibitor.Store(newInhibitor)
 
 		// An interface value that holds a nil concrete value is non-nil.