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/v2/api.go

@@ -50,6 +50,7 @@ import (
 	"github.com/prometheus/alertmanager/silence"
 	"github.com/prometheus/alertmanager/silence/silencepb"
 	"github.com/prometheus/alertmanager/types"
+	"go.opentelemetry.io/otel/trace"
 )
 
 // API represents an Alertmanager API v2.
@@ -313,7 +314,15 @@ 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)
+	// Extract trace ID and span ID from the request context
+	traceID := ""
+	spanID := ""
+	if spanCtx := trace.SpanContextFromContext(params.HTTPRequest.Context()); spanCtx.IsValid() {
+		traceID = spanCtx.TraceID().String()
+		spanID = spanCtx.SpanID().String()
+	}
+
+	alerts := OpenAPIAlertsToAlerts(params.Alerts, traceID, spanID)
 	now := time.Now()
 
 	api.mtx.RLock()

api/v2/compat.go

@@ -170,7 +170,7 @@ 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(apiAlerts open_api_models.PostableAlerts, traceID, spanID string) []*types.Alert {
 	alerts := []*types.Alert{}
 	for _, apiAlert := range apiAlerts {
 		alert := types.Alert{
@@ -181,6 +181,8 @@ func OpenAPIAlertsToAlerts(apiAlerts open_api_models.PostableAlerts) []*types.Al
 				EndsAt:       time.Time(apiAlert.EndsAt),
 				GeneratorURL: string(apiAlert.GeneratorURL),
 			},
+			TraceID: traceID,
+			SpanID:  spanID,
 		}
 		alerts = append(alerts, &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"
 )
@@ -401,6 +402,8 @@ func run() int {
 		return d + waitFunc()
 	}
 
+	tracingManager := tracing.NewManager(logger.With("component", "tracing"))
+
 	var (
 		inhibitor *inhibit.Inhibitor
 		tmpl      *template.Template
@@ -523,6 +526,13 @@ func run() int {
 		go disp.Run()
 		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
 	})
 
@@ -551,7 +561,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() {
@@ -582,6 +595,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
+}

dispatch/dispatch.go

@@ -25,13 +25,19 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/common/model"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/prometheus/alertmanager/notify"
 	"github.com/prometheus/alertmanager/provider"
 	"github.com/prometheus/alertmanager/store"
 	"github.com/prometheus/alertmanager/types"
 )
 
+var tracer = otel.Tracer("github.com/prometheus/alertmanager/dispatch")
+
 // DispatcherMetrics represents metrics associated to a dispatcher.
 type DispatcherMetrics struct {
 	aggrGroups            prometheus.Gauge
@@ -162,20 +168,12 @@ func (d *Dispatcher) run(it provider.AlertIterator) {
 				return
 			}
 
-			d.logger.Debug("Received alert", "alert", alert)
-
 			// Log errors but keep trying.
 			if err := it.Err(); err != nil {
 				d.logger.Error("Error on alert update", "err", err)
 				continue
 			}
-
-			now := time.Now()
-			for _, r := range d.route.Match(alert.Labels) {
-				d.processAlert(alert, r)
-			}
-			d.metrics.processingDuration.Observe(time.Since(now).Seconds())
-
+			d.dispatch(alert)
 		case <-maintenance.C:
 			d.doMaintenance()
 		case <-d.ctx.Done():
@@ -184,6 +182,41 @@ func (d *Dispatcher) run(it provider.AlertIterator) {
 	}
 }
 
+func (d *Dispatcher) dispatch(alert *types.Alert) {
+	d.logger.Debug("Received alert", "alert", alert)
+
+	attrs := []attribute.KeyValue{
+		attribute.String("alert.name", alert.Name()),
+		attribute.String("alert.fingerprint", alert.Fingerprint().String()),
+	}
+
+	// Build span start options
+	spanOpts := []trace.SpanStartOption{
+		trace.WithAttributes(attrs...),
+		// we'll use producer here since the alert is not processed
+		// synchronously
+		trace.WithSpanKind(trace.SpanKindProducer),
+	}
+
+	// Check if the alert has a valid trace context and link to it
+	alertSpanCtx := alert.TraceSpanContext()
+	if alertSpanCtx.IsValid() {
+		spanOpts = append(spanOpts, trace.WithLinks(trace.Link{
+			SpanContext: alertSpanCtx,
+		}))
+	}
+
+	traceCtx, span := tracer.Start(d.ctx, "dispatch.Dispatcher.dispatch", spanOpts...)
+	defer span.End()
+
+	now := time.Now()
+	for _, r := range d.route.Match(alert.Labels) {
+		d.processAlert(trace.LinkFromContext(traceCtx), alert, r)
+		span.SetAttributes(attribute.String("receiver", r.RouteOpts.Receiver))
+	}
+	d.metrics.processingDuration.Observe(time.Since(now).Seconds())
+}
+
 func (d *Dispatcher) doMaintenance() {
 	d.mtx.Lock()
 	defer d.mtx.Unlock()
@@ -310,7 +343,7 @@ type notifyFunc func(context.Context, ...*types.Alert) bool
 
 // processAlert determines in which aggregation group the alert falls
 // and inserts it.
-func (d *Dispatcher) processAlert(alert *types.Alert, route *Route) {
+func (d *Dispatcher) processAlert(dispatchLink trace.Link, alert *types.Alert, route *Route) {
 	groupLabels := getGroupLabels(alert, route)
 
 	fp := groupLabels.Fingerprint()
@@ -348,15 +381,26 @@ func (d *Dispatcher) processAlert(alert *types.Alert, route *Route) {
 	ag.insert(alert)
 
 	go ag.run(func(ctx context.Context, alerts ...*types.Alert) bool {
+		ctx, span := tracer.Start(ctx, "dispatch.Dispatch.notify",
+			trace.WithAttributes(attribute.Int("alerts.count", len(alerts))),
+			trace.WithLinks(dispatchLink),
+			trace.WithSpanKind(trace.SpanKindConsumer),
+		)
+		defer span.End()
+
 		_, _, 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
 				// message should only be logged at the debug level.
+				span.RecordError(fmt.Errorf("notify for alerts failed: %w", err))
+				span.SetStatus(codes.Error, err.Error())
 				logger.Debug("Notify for alerts failed")
 			} else {
+				span.RecordError(fmt.Errorf("notify for alerts failed: %w", err))
+				span.SetStatus(codes.Error, err.Error())
 				logger.Error("Notify for alerts failed")
 			}
 		}

examples/webhook/echo.go

@@ -22,7 +22,12 @@ import (
 )
 
 func main() {
-	log.Fatal(http.ListenAndServe(":5001", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	log.Fatal(http.ListenAndServe("127.0.0.1:5001", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		log.Printf("headers:")
+		for k, v := range r.Header {
+			log.Printf("  %s: %s", k, v)
+		}
+
 		b, err := io.ReadAll(r.Body)
 		if err != nil {
 			panic(err)