feat: add distributed tracing support

Add tracing support using otel:
- api: extract trace and span IDs from request context and add to alerts
- types: add trace and span IDs to alerts
- dispatch: add distributed tracing support
- notify: add distributed tracing support

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

@@ -14,6 +14,7 @@
 package api
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -27,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"
@@ -203,7 +205,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)
 }
 
@@ -244,7 +246,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

@@ -14,6 +14,7 @@
 package v2
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -31,6 +32,7 @@ import (
 	prometheus_model "github.com/prometheus/common/model"
 	"github.com/prometheus/common/version"
 	"github.com/rs/cors"
+	"go.opentelemetry.io/otel"
 
 	"github.com/prometheus/alertmanager/api/metrics"
 	open_api_models "github.com/prometheus/alertmanager/api/v2/models"
@@ -52,6 +54,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
@@ -80,7 +84,7 @@ type (
 	groupsFn         func(func(*dispatch.Route) bool, func(*types.Alert, time.Time) bool) (dispatch.AlertGroups, map[prometheus_model.Fingerprint][]string)
 	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.
@@ -272,14 +276,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)
@@ -289,13 +294,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()
 
@@ -313,7 +318,12 @@ 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 := params.HTTPRequest.Context()
+	ctx, span := tracer.Start(ctx, "api.postAlertsHandler")
+	defer span.End()
+
+	alerts := OpenAPIAlertsToAlerts(ctx, params.Alerts)
+
 	now := time.Now()
 
 	api.mtx.RLock()
@@ -359,13 +369,15 @@ func (api *API) postAlertsHandler(params alert_ops.PostAlertsParams) middleware.
 		}
 		validAlerts = append(validAlerts, a)
 	}
-	if err := api.alerts.Put(validAlerts...); err != nil {
+	if err := api.alerts.Put(ctx, validAlerts...); err != nil {
 		logger.Error("Failed to create alerts", "err", err)
+		span.RecordError(err)
 		return alert_ops.NewPostAlertsInternalServerError().WithPayload(err.Error())
 	}
 
 	if validationErrs.Len() > 0 {
 		logger.Error("Failed to validate alerts", "err", validationErrs.Error())
+		span.RecordError(validationErrs)
 		return alert_ops.NewPostAlertsBadRequest().WithPayload(validationErrs.Error())
 	}
 
@@ -436,12 +448,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())

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{

cmd/alertmanager/main.go

@@ -59,6 +59,7 @@ import (
 	"github.com/prometheus/alertmanager/silence"
 	"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"
 )
@@ -404,6 +405,8 @@ func run() int {
 		return d + waitFunc()
 	}
 
+	tracingManager := tracing.NewManager(logger.With("component", "tracing"))
+
 	var (
 		inhibitor *inhibit.Inhibitor
 		tmpl      *template.Template
@@ -491,9 +494,9 @@ func run() int {
 		configuredIntegrations.Set(float64(integrationsNum))
 		configuredInhibitionRules.Set(float64(len(conf.InhibitRules)))
 
-		api.Update(conf, func(labels model.LabelSet) {
-			inhibitor.Mutes(labels)
-			silencer.Mutes(labels)
+		api.Update(conf, func(ctx context.Context, labels model.LabelSet) {
+			inhibitor.Mutes(ctx, labels)
+			silencer.Mutes(ctx, labels)
 		})
 
 		disp = dispatch.NewDispatcher(
@@ -536,6 +539,13 @@ func run() int {
 		go disp.Run(startTime.Add(*DispatchStartDelay))
 		go inhibitor.Run()
 
+		err = tracingManager.ApplyConfig(conf)
+		if err != nil {
+			return fmt.Errorf("failed to apply tracing config: %w", err)
+		}
+
+		go tracingManager.Run()
+
 		return nil
 	})
 
@@ -564,7 +574,10 @@ func run() int {
 
 	mux := api.Register(router, *routePrefix)
 
-	srv := &http.Server{Handler: mux}
+	srv := &http.Server{
+		// instrument all handlers with tracing
+		Handler: tracing.Middleware(mux),
+	}
 	srvc := make(chan struct{})
 
 	go func() {
@@ -595,6 +608,11 @@ func run() int {
 			errc <- configCoordinator.Reload()
 		case <-term:
 			logger.Info("Received SIGTERM, exiting gracefully...")
+
+			// shut down the tracing manager to flush any remaining spans.
+			// this blocks for up to 5s
+			tracingManager.Stop()
+
 			return 0
 		case <-srvc:
 			return 1

config/config.go

@@ -351,6 +351,8 @@ type Config struct {
 	MuteTimeIntervals []MuteTimeInterval `yaml:"mute_time_intervals,omitempty" json:"mute_time_intervals,omitempty"`
 	TimeIntervals     []TimeInterval     `yaml:"time_intervals,omitempty" json:"time_intervals,omitempty"`
 
+	TracingConfig TracingConfig `yaml:"tracing,omitempty" json:"tracing,omitempty"`
+
 	// original is the input from which the config was parsed.
 	original string
 }
@@ -1182,3 +1184,111 @@ func (m Matchers) MarshalJSON() ([]byte, error) {
 	}
 	return json.Marshal(result)
 }
+
+// TODO: probably move these into prometheus/common since they're copied from
+// prometheus/prometheus?
+
+type TracingClientType string
+
+const (
+	TracingClientHTTP TracingClientType = "http"
+	TracingClientGRPC TracingClientType = "grpc"
+
+	GzipCompression = "gzip"
+)
+
+// UnmarshalYAML implements the yaml.Unmarshaler interface.
+func (t *TracingClientType) UnmarshalYAML(unmarshal func(interface{}) error) error {
+	*t = TracingClientType("")
+	type plain TracingClientType
+	if err := unmarshal((*plain)(t)); err != nil {
+		return err
+	}
+
+	if *t != TracingClientHTTP && *t != TracingClientGRPC {
+		return fmt.Errorf("expected tracing client type to be to be %s or %s, but got %s",
+			TracingClientHTTP, TracingClientGRPC, *t,
+		)
+	}
+
+	return nil
+}
+
+// TracingConfig configures the tracing options.
+type TracingConfig struct {
+	ClientType       TracingClientType   `yaml:"client_type,omitempty"`
+	Endpoint         string              `yaml:"endpoint,omitempty"`
+	SamplingFraction float64             `yaml:"sampling_fraction,omitempty"`
+	Insecure         bool                `yaml:"insecure,omitempty"`
+	TLSConfig        commoncfg.TLSConfig `yaml:"tls_config,omitempty"`
+	Headers          map[string]string   `yaml:"headers,omitempty"`
+	Compression      string              `yaml:"compression,omitempty"`
+	Timeout          model.Duration      `yaml:"timeout,omitempty"`
+}
+
+// SetDirectory joins any relative file paths with dir.
+func (t *TracingConfig) SetDirectory(dir string) {
+	t.TLSConfig.SetDirectory(dir)
+}
+
+// UnmarshalYAML implements the yaml.Unmarshaler interface.
+func (t *TracingConfig) UnmarshalYAML(unmarshal func(interface{}) error) error {
+	*t = TracingConfig{
+		ClientType: TracingClientGRPC,
+	}
+	type plain TracingConfig
+	if err := unmarshal((*plain)(t)); err != nil {
+		return err
+	}
+
+	if err := validateHeadersForTracing(t.Headers); err != nil {
+		return err
+	}
+
+	if t.Endpoint == "" {
+		return errors.New("tracing endpoint must be set")
+	}
+
+	if t.Compression != "" && t.Compression != GzipCompression {
+		return fmt.Errorf("invalid compression type %s provided, valid options: %s",
+			t.Compression, GzipCompression)
+	}
+
+	return nil
+}
+
+var reservedHeaders = map[string]struct{}{
+	// NOTE: authorization is checked specially,
+	// see RemoteWriteConfig.UnmarshalYAML.
+	// "authorization":                  {},
+	"host":                              {},
+	"content-encoding":                  {},
+	"content-length":                    {},
+	"content-type":                      {},
+	"user-agent":                        {},
+	"connection":                        {},
+	"keep-alive":                        {},
+	"proxy-authenticate":                {},
+	"proxy-authorization":               {},
+	"www-authenticate":                  {},
+	"accept-encoding":                   {},
+	"x-prometheus-remote-write-version": {},
+	"x-prometheus-remote-read-version":  {},
+
+	// Added by SigV4.
+	"x-amz-date":           {},
+	"x-amz-security-token": {},
+	"x-amz-content-sha256": {},
+}
+
+func validateHeadersForTracing(headers map[string]string) error {
+	for header := range headers {
+		if strings.ToLower(header) == "authorization" {
+			return errors.New("custom authorization header configuration is not yet supported")
+		}
+		if _, ok := reservedHeaders[strings.ToLower(header)]; ok {
+			return fmt.Errorf("%s is a reserved header. It must not be changed", header)
+		}
+	}
+	return nil
+}