feat: add distributed tracing support

Add tracing support using otel to the the following components:
- api: extract trace and span IDs from request context
- provider: mem put
- dispatch: split logic and use better naming
- inhibit: source and target traces, mutes, etc. drop metrics
- silence: query, expire, mutes
- notify: add distributed tracing support to stages and all http requests

Note: inhibitor metrics are dropped since we have tracing now and they
are not needed. We have not released any version with these metrics so
we can drop them safely, this is not a breaking change.

This change borrows part of the implementation from #3673
Fixes #3670

Signed-off-by: Dave Henderson <dhenderson@gmail.com>
Signed-off-by: Siavash Safi <siavash@cloudflare.com>

Author: hairyhenderson

api/api.go

@@ -28,6 +28,7 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/common/promslog"
 	"github.com/prometheus/common/route"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 
 	apiv2 "github.com/prometheus/alertmanager/api/v2"
 	"github.com/prometheus/alertmanager/cluster"
@@ -201,7 +202,7 @@ func (api *API) Register(r *route.Router, routePrefix string) *http.ServeMux {
 
 // Update config and resolve timeout of each API. APIv2 also needs
 // setAlertStatus to be updated.
-func (api *API) Update(cfg *config.Config, setAlertStatus func(model.LabelSet)) {
+func (api *API) Update(cfg *config.Config, setAlertStatus func(ctx context.Context, labels model.LabelSet)) {
 	api.v2.Update(cfg, setAlertStatus)
 }
 
@@ -242,7 +243,7 @@ func (api *API) instrumentHandler(prefix string, h http.Handler) http.Handler {
 		}
 		promhttp.InstrumentHandlerDuration(
 			api.requestDuration.MustCurryWith(prometheus.Labels{"handler": path}),
-			h,
+			otelhttp.WithRouteTag(path, h),
 		).ServeHTTP(w, r)
 	})
 }

api/v2/api.go

@@ -33,6 +33,8 @@ import (
 	prometheus_model "github.com/prometheus/common/model"
 	"github.com/prometheus/common/version"
 	"github.com/rs/cors"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/codes"
 
 	"github.com/prometheus/alertmanager/api/metrics"
 	open_api_models "github.com/prometheus/alertmanager/api/v2/models"
@@ -54,6 +56,8 @@ import (
 	"github.com/prometheus/alertmanager/types"
 )
 
+var tracer = otel.Tracer("github.com/prometheus/alertmanager/api/v2")
+
 // API represents an Alertmanager API v2.
 type API struct {
 	peer           cluster.ClusterPeer
@@ -82,7 +86,7 @@ type (
 	groupsFn         func(context.Context, func(*dispatch.Route) bool, func(*types.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(prometheus_model.LabelSet)
+	setAlertStatusFn func(ctx context.Context, labels prometheus_model.LabelSet)
 )
 
 // NewAPI returns a new Alertmanager API v2.
@@ -173,6 +177,9 @@ func (api *API) getStatusHandler(params general_ops.GetStatusParams) middleware.
 	api.mtx.RLock()
 	defer api.mtx.RUnlock()
 
+	_, span := tracer.Start(params.HTTPRequest.Context(), "api.getStatusHandler")
+	defer span.End()
+
 	original := api.alertmanagerConfig.String()
 	uptime := strfmt.DateTime(api.uptime)
 
@@ -229,6 +236,9 @@ func (api *API) getReceiversHandler(params receiver_ops.GetReceiversParams) midd
 	api.mtx.RLock()
 	defer api.mtx.RUnlock()
 
+	_, span := tracer.Start(params.HTTPRequest.Context(), "api.getReceiversHandler")
+	defer span.End()
+
 	receivers := make([]*open_api_models.Receiver, 0, len(api.alertmanagerConfig.Receivers))
 	for i := range api.alertmanagerConfig.Receivers {
 		receivers = append(receivers, &open_api_models.Receiver{Name: &api.alertmanagerConfig.Receivers[i].Name})
@@ -243,11 +253,13 @@ func (api *API) getAlertsHandler(params alert_ops.GetAlertsParams) middleware.Re
 		// Initialize result slice to prevent api returning `null` when there
 		// are no alerts present
 		res = open_api_models.GettableAlerts{}
-		ctx = params.HTTPRequest.Context()
 
 		logger = api.requestLogger(params.HTTPRequest)
 	)
 
+	ctx, span := tracer.Start(params.HTTPRequest.Context(), "api.getAlertsHandler")
+	defer span.End()
+
 	matchers, err := parseFilter(params.Filter)
 	if err != nil {
 		logger.Debug("Failed to parse matchers", "err", err)
@@ -274,14 +286,15 @@ func (api *API) getAlertsHandler(params alert_ops.GetAlertsParams) middleware.Re
 
 	api.mtx.RLock()
 	for a := range alerts.Next() {
+		alert := a.Data
 		if err = alerts.Err(); err != nil {
 			break
 		}
 		if err = ctx.Err(); err != nil {
 			break
 		}
 
-		routes := api.route.Match(a.Labels)
+		routes := api.route.Match(alert.Labels)
 		receivers := make([]string, 0, len(routes))
 		for _, r := range routes {
 			receivers = append(receivers, r.RouteOpts.Receiver)
@@ -291,13 +304,13 @@ func (api *API) getAlertsHandler(params alert_ops.GetAlertsParams) middleware.Re
 			continue
 		}
 
-		if !alertFilter(a, now) {
+		if !alertFilter(alert, now) {
 			continue
 		}
 
-		alert := AlertToOpenAPIAlert(a, api.getAlertStatus(a.Fingerprint()), receivers, nil)
+		openAlert := AlertToOpenAPIAlert(alert, api.getAlertStatus(alert.Fingerprint()), receivers, nil)
 
-		res = append(res, alert)
+		res = append(res, openAlert)
 	}
 	api.mtx.RUnlock()
 
@@ -315,7 +328,11 @@ func (api *API) getAlertsHandler(params alert_ops.GetAlertsParams) middleware.Re
 func (api *API) postAlertsHandler(params alert_ops.PostAlertsParams) middleware.Responder {
 	logger := api.requestLogger(params.HTTPRequest)
 
-	alerts := OpenAPIAlertsToAlerts(params.Alerts)
+	ctx, span := tracer.Start(params.HTTPRequest.Context(), "api.postAlertsHandler")
+	defer span.End()
+
+	alerts := OpenAPIAlertsToAlerts(ctx, params.Alerts)
+
 	now := time.Now()
 
 	api.mtx.RLock()
@@ -361,13 +378,19 @@ func (api *API) postAlertsHandler(params alert_ops.PostAlertsParams) middleware.
 		}
 		validAlerts = append(validAlerts, a)
 	}
-	if err := api.alerts.Put(validAlerts...); err != nil {
-		logger.Error("Failed to create alerts", "err", err)
+	if err := api.alerts.Put(ctx, validAlerts...); err != nil {
+		message := "Failed to create alerts"
+		logger.Error(message, "err", err)
+		span.SetStatus(codes.Error, message)
+		span.RecordError(err)
 		return alert_ops.NewPostAlertsInternalServerError().WithPayload(err.Error())
 	}
 
 	if validationErrs.Len() > 0 {
-		logger.Error("Failed to validate alerts", "err", validationErrs.Error())
+		message := "Failed to validate alerts"
+		logger.Error(message, "err", validationErrs.Error())
+		span.SetStatus(codes.Error, message)
+		span.RecordError(validationErrs)
 		return alert_ops.NewPostAlertsBadRequest().WithPayload(validationErrs.Error())
 	}
 
@@ -377,6 +400,9 @@ func (api *API) postAlertsHandler(params alert_ops.PostAlertsParams) middleware.
 func (api *API) getAlertGroupsHandler(params alertgroup_ops.GetAlertGroupsParams) middleware.Responder {
 	logger := api.requestLogger(params.HTTPRequest)
 
+	ctx, span := tracer.Start(params.HTTPRequest.Context(), "api.getAlertGroupsHandler")
+	defer span.End()
+
 	matchers, err := parseFilter(params.Filter)
 	if err != nil {
 		logger.Debug("Failed to parse matchers", "err", err)
@@ -407,8 +433,12 @@ func (api *API) getAlertGroupsHandler(params alertgroup_ops.GetAlertGroupsParams
 	}(receiverFilter)
 
 	af := api.alertFilter(matchers, *params.Silenced, *params.Inhibited, *params.Active)
-	alertGroups, allReceivers, err := api.alertGroups(params.HTTPRequest.Context(), rf, af)
+	alertGroups, allReceivers, err := api.alertGroups(ctx, rf, af)
 	if err != nil {
+		message := "Failed to get alert groups"
+		logger.Error(message, "err", err)
+		span.SetStatus(codes.Error, message)
+		span.RecordError(err)
 		return alertgroup_ops.NewGetAlertGroupsInternalServerError()
 	}
 
@@ -441,12 +471,15 @@ func (api *API) getAlertGroupsHandler(params alertgroup_ops.GetAlertGroupsParams
 
 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 {
+		ctx, span := tracer.Start(context.Background(), "alertFilter")
+		defer span.End()
+
 		if !a.EndsAt.IsZero() && a.EndsAt.Before(now) {
 			return false
 		}
 
 		// Set alert's current status based on its label set.
-		api.setAlertStatus(a.Labels)
+		api.setAlertStatus(ctx, a.Labels)
 
 		// Get alert's current status after seeing if it is suppressed.
 		status := api.getAlertStatus(a.Fingerprint())
@@ -510,13 +543,16 @@ func matchFilterLabels(matchers []*labels.Matcher, sms map[string]string) bool {
 func (api *API) getSilencesHandler(params silence_ops.GetSilencesParams) middleware.Responder {
 	logger := api.requestLogger(params.HTTPRequest)
 
+	ctx, span := tracer.Start(params.HTTPRequest.Context(), "api.getSilencesHandler")
+	defer span.End()
+
 	matchers, err := parseFilter(params.Filter)
 	if err != nil {
 		logger.Debug("Failed to parse matchers", "err", err)
 		return silence_ops.NewGetSilencesBadRequest().WithPayload(err.Error())
 	}
 
-	psils, _, err := api.silences.Query()
+	psils, _, err := api.silences.Query(ctx)
 	if err != nil {
 		logger.Error("Failed to get silences", "err", err)
 		return silence_ops.NewGetSilencesInternalServerError().WithPayload(err.Error())
@@ -606,7 +642,10 @@ func CheckSilenceMatchesFilterLabels(s *silencepb.Silence, matchers []*labels.Ma
 func (api *API) getSilenceHandler(params silence_ops.GetSilenceParams) middleware.Responder {
 	logger := api.requestLogger(params.HTTPRequest)
 
-	sils, _, err := api.silences.Query(silence.QIDs(params.SilenceID.String()))
+	ctx, span := tracer.Start(params.HTTPRequest.Context(), "api.getSilenceHandler")
+	defer span.End()
+
+	sils, _, err := api.silences.Query(ctx, silence.QIDs(params.SilenceID.String()))
 	if err != nil {
 		logger.Error("Failed to get silence by id", "err", err, "id", params.SilenceID.String())
 		return silence_ops.NewGetSilenceInternalServerError().WithPayload(err.Error())
@@ -629,8 +668,11 @@ func (api *API) getSilenceHandler(params silence_ops.GetSilenceParams) middlewar
 func (api *API) deleteSilenceHandler(params silence_ops.DeleteSilenceParams) middleware.Responder {
 	logger := api.requestLogger(params.HTTPRequest)
 
+	ctx, span := tracer.Start(params.HTTPRequest.Context(), "api.deleteSilenceHandler")
+	defer span.End()
+
 	sid := params.SilenceID.String()
-	if err := api.silences.Expire(sid); err != nil {
+	if err := api.silences.Expire(ctx, sid); err != nil {
 		logger.Error("Failed to expire silence", "err", err)
 		if errors.Is(err, silence.ErrNotFound) {
 			return silence_ops.NewDeleteSilenceNotFound()
@@ -643,6 +685,9 @@ func (api *API) deleteSilenceHandler(params silence_ops.DeleteSilenceParams) mid
 func (api *API) postSilencesHandler(params silence_ops.PostSilencesParams) middleware.Responder {
 	logger := api.requestLogger(params.HTTPRequest)
 
+	ctx, span := tracer.Start(params.HTTPRequest.Context(), "api.postSilencesHandler")
+	defer span.End()
+
 	sil, err := PostableSilenceToProto(params.Silence)
 	if err != nil {
 		logger.Error("Failed to marshal silence to proto", "err", err)
@@ -663,7 +708,7 @@ func (api *API) postSilencesHandler(params silence_ops.PostSilencesParams) middl
 		return silence_ops.NewPostSilencesBadRequest().WithPayload(msg)
 	}
 
-	if err = api.silences.Set(sil); err != nil {
+	if err = api.silences.Set(ctx, sil); err != nil {
 		logger.Error("Failed to create silence", "err", err)
 		if errors.Is(err, silence.ErrNotFound) {
 			return silence_ops.NewPostSilencesNotFound().WithPayload(err.Error())

api/v2/api_test.go

@@ -53,7 +53,15 @@ func TestGetStatusHandlerWithNilPeer(t *testing.T) {
 	}
 
 	// Test ensures this method call does not panic.
-	status := api.getStatusHandler(general_ops.GetStatusParams{}).(*general_ops.GetStatusOK)
+	status := api.getStatusHandler(
+		general_ops.GetStatusParams{
+			HTTPRequest: httptest.NewRequest(
+				"GET",
+				"/api/v2/status",
+				nil,
+			),
+		},
+	).(*general_ops.GetStatusOK)
 
 	c := status.Payload.Cluster
 
@@ -160,16 +168,16 @@ func TestDeleteSilenceHandler(t *testing.T) {
 		EndsAt:    now.Add(time.Hour),
 		UpdatedAt: now,
 	}
-	require.NoError(t, silences.Set(unexpiredSil))
+	require.NoError(t, silences.Set(t.Context(), unexpiredSil))
 
 	expiredSil := &silencepb.Silence{
 		Matchers:  []*silencepb.Matcher{m},
 		StartsAt:  now.Add(-time.Hour),
 		EndsAt:    now.Add(time.Hour),
 		UpdatedAt: now,
 	}
-	require.NoError(t, silences.Set(expiredSil))
-	require.NoError(t, silences.Expire(expiredSil.Id))
+	require.NoError(t, silences.Set(t.Context(), expiredSil))
+	require.NoError(t, silences.Expire(t.Context(), expiredSil.Id))
 
 	for i, tc := range []struct {
 		sid          string
@@ -222,16 +230,16 @@ func TestPostSilencesHandler(t *testing.T) {
 		EndsAt:    now.Add(time.Hour),
 		UpdatedAt: now,
 	}
-	require.NoError(t, silences.Set(unexpiredSil))
+	require.NoError(t, silences.Set(t.Context(), unexpiredSil))
 
 	expiredSil := &silencepb.Silence{
 		Matchers:  []*silencepb.Matcher{m},
 		StartsAt:  now.Add(-time.Hour),
 		EndsAt:    now.Add(time.Hour),
 		UpdatedAt: now,
 	}
-	require.NoError(t, silences.Set(expiredSil))
-	require.NoError(t, silences.Expire(expiredSil.Id))
+	require.NoError(t, silences.Set(t.Context(), expiredSil))
+	require.NoError(t, silences.Expire(t.Context(), expiredSil.Id))
 
 	t.Run("Silences CRUD", func(t *testing.T) {
 		for i, tc := range []struct {

api/v2/compat.go

@@ -14,6 +14,7 @@
 package v2
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -170,7 +171,10 @@ func AlertToOpenAPIAlert(alert *types.Alert, status types.AlertStatus, receivers
 }
 
 // OpenAPIAlertsToAlerts converts open_api_models.PostableAlerts to []*types.Alert.
-func OpenAPIAlertsToAlerts(apiAlerts open_api_models.PostableAlerts) []*types.Alert {
+func OpenAPIAlertsToAlerts(ctx context.Context, apiAlerts open_api_models.PostableAlerts) []*types.Alert {
+	_, span := tracer.Start(ctx, "OpenAPIAlertsToAlerts")
+	defer span.End()
+
 	alerts := []*types.Alert{}
 	for _, apiAlert := range apiAlerts {
 		alert := types.Alert{