pkg/receive: add tracing to forwarded requests (#1613)

This commit instruments the Thanos receive component's HTTP server with
the specified tracer. Specifically, it adds spans for each request
forwarded between receive nodes.

Signed-off-by: Lucas Servén Marín <lserven@gmail.com>

cmd/thanos/receive.go

@@ -160,6 +160,7 @@ func runReceive(
 		TenantHeader:      tenantHeader,
 		ReplicaHeader:     replicaHeader,
 		ReplicationFactor: replicationFactor,
+		Tracer:            tracer,
 	})
 
 	statusProber := prober.NewProber(comp, logger, prometheus.WrapRegistererWithPrefix("thanos_", reg))

pkg/receive/handler.go

@@ -25,6 +25,7 @@ import (
 	terrors "github.com/prometheus/prometheus/tsdb/errors"
 	extpromhttp "github.com/thanos-io/thanos/pkg/extprom/http"
 	"github.com/thanos-io/thanos/pkg/runutil"
+	"github.com/thanos-io/thanos/pkg/tracing"
 )
 
 // Options for the web Handler.
@@ -36,10 +37,12 @@ type Options struct {
 	TenantHeader      string
 	ReplicaHeader     string
 	ReplicationFactor uint64
+	Tracer            opentracing.Tracer
 }
 
 // Handler serves a Prometheus remote write receiving HTTP endpoint.
 type Handler struct {
+	client   *http.Client
 	logger   log.Logger
 	writer   *Writer
 	router   *route.Router
@@ -58,7 +61,13 @@ func NewHandler(logger log.Logger, o *Options) *Handler {
 		logger = log.NewNopLogger()
 	}
 
+	client := &http.Client{}
+	if o.Tracer != nil {
+		client.Transport = tracing.HTTPTripperware(logger, http.DefaultTransport)
+	}
+
 	h := &Handler{
+		client:  client,
 		logger:  logger,
 		writer:  o.Writer,
 		router:  route.New(),
@@ -79,6 +88,9 @@ func NewHandler(logger log.Logger, o *Options) *Handler {
 
 	readyf := h.testReady
 	instrf := func(name string, next func(w http.ResponseWriter, r *http.Request)) http.HandlerFunc {
+		if o.Tracer != nil {
+			next = tracing.HTTPMiddleware(o.Tracer, name, logger, http.HandlerFunc(next))
+		}
 		return ins.NewHandler(name, http.HandlerFunc(next))
 	}
 
@@ -342,10 +354,14 @@ func (h *Handler) parallelizeRequests(ctx context.Context, tenant string, replic
 				h.forwardRequestsTotal.WithLabelValues("success").Inc()
 			}()
 
+			// Create a span to track the request made to another receive node.
+			span, ctx := tracing.StartSpan(ctx, "thanos_receive_forward")
+			defer span.Finish()
+
 			// Actually make the request against the endpoint
 			// we determined should handle these time series.
 			var res *http.Response
-			res, err = http.DefaultClient.Do(req.WithContext(ctx))
+			res, err = h.client.Do(req.WithContext(ctx))
 			if err != nil {
 				level.Error(h.logger).Log("msg", "forwarding request", "err", err, "endpoint", endpoint)
 				ec <- err

pkg/tracing/http.go

@@ -12,8 +12,8 @@ import (
 	"github.com/opentracing/opentracing-go/ext"
 )
 
-// HTTPMiddleware returns HTTP handler that injects given tracer and starts new server span. If any client span is fetched
-// wire we include that as our parent.
+// HTTPMiddleware returns an HTTP handler that injects the given tracer and starts a new server span.
+// If any client span is fetched from the wire, we include that as our parent.
 func HTTPMiddleware(tracer opentracing.Tracer, name string, logger log.Logger, next http.Handler) http.HandlerFunc {
 	operationName := fmt.Sprintf("/%s HTTP[server]", name)