chore: Improve tracing visibility into distributor (#4361)

Author: simonswine

pkg/distributor/distributor.go

@@ -292,10 +292,15 @@ func (d *Distributor) GetProfileLanguage(series *distributormodel.ProfileSeries)
 }
 
 func (d *Distributor) PushBatch(ctx context.Context, req *distributormodel.PushRequest) error {
+	sp, ctx := opentracing.StartSpanFromContext(ctx, "Distributor.PushBatch")
+	defer sp.Finish()
+
 	tenantID, err := tenant.ExtractTenantIDFromContext(ctx)
 	if err != nil {
 		return connect.NewError(connect.CodeUnauthenticated, err)
 	}
+	sp.SetTag("tenant_id", tenantID)
+
 	if len(req.Series) == 0 {
 		return noNewProfilesReceivedError()
 	}
@@ -415,9 +420,13 @@ func (d *Distributor) pushSeries(ctx context.Context, req *distributormodel.Prof
 	// Normalisation is quite an expensive operation,
 	// therefore it should be done after the rate limit check.
 	if req.Language == "go" {
+		sp, _ := opentracing.StartSpanFromContext(ctx, "pprof.FixGoProfile")
 		req.Profile.Profile = pprof.FixGoProfile(req.Profile.Profile)
+		sp.Finish()
 	}
+	sp, _ := opentracing.StartSpanFromContext(ctx, "Profile.Normalize")
 	req.Profile.Normalize()
+	sp.Finish()
 
 	if len(req.Profile.Sample) == 0 {
 		// TODO(kolesnikovae):

pkg/distributor/writepath/router.go

@@ -94,6 +94,8 @@ func (m *Router) running(ctx context.Context) error {
 }
 
 func (m *Router) Send(ctx context.Context, req *distributormodel.ProfileSeries) error {
+	sp, ctx := opentracing.StartSpanFromContext(ctx, "Router.Send")
+	defer sp.Finish()
 	config := m.overrides.WritePathOverrides(req.TenantID)
 	switch config.WritePath {
 	case SegmentWriterPath:

pkg/ingester/pyroscope/ingest_handler.go

@@ -177,9 +177,17 @@ func readInputRawDataFromRequest(ctx context.Context, r *http.Request, input *in
 	}
 
 	buf := bytes.NewBuffer(make([]byte, 0, 64<<10))
-	if n, err := io.Copy(buf, r.Body); err != nil {
+	n, err := io.Copy(buf, r.Body)
+	if err != nil {
 		return fmt.Errorf("error reading request body bytes_read %d: %w", n, err)
 	}
+
+	if sp != nil {
+		sp.SetTag("content_length", n)
+		sp.LogFields(
+			otlog.String("msg", "read body from request"),
+		)
+	}
 	b := buf.Bytes()
 
 	switch {

pkg/util/delayhandler/http.go

@@ -3,6 +3,8 @@ package delayhandler
 import (
 	"net/http"
 	"time"
+
+	"github.com/opentracing/opentracing-go"
 )
 
 func wrapResponseWriter(w http.ResponseWriter, end time.Time) (http.ResponseWriter, *delayedResponseWriter) {
@@ -65,15 +67,21 @@ func NewHTTP(limits Limits) func(h http.Handler) http.Handler {
 	return func(h http.Handler) http.Handler {
 		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			start := timeNow()
+			ctx := r.Context()
 
-			delay := getDelay(r.Context(), limits)
+			delay := getDelay(ctx, limits)
 			var delayRw *delayedResponseWriter
 			if delay > 0 {
 				w, delayRw = wrapResponseWriter(w, start.Add(delay))
+
+				// only add a span when delay is active
+				var sp opentracing.Span
+				sp, ctx = opentracing.StartSpanFromContext(ctx, "delayhandler.Handler")
+				defer sp.Finish()
 			}
 
 			// now run the chain after me
-			h.ServeHTTP(w, r)
+			h.ServeHTTP(w, r.WithContext(ctx))
 
 			// if we didn't delay, return immediately
 			if delayRw == nil {
@@ -96,6 +104,11 @@ func NewHTTP(limits Limits) func(h http.Handler) http.Handler {
 				addDelayHeader(w.Header(), delayLeft)
 			}
 
+			// create a separate span to make the artificial delay clear
+			sp, _ := opentracing.StartSpanFromContext(ctx, "delayhandler.Delay")
+			sp.SetTag("delayed_by", delayLeft.String())
+			defer sp.Finish()
+
 			// wait for the delay to elapse
 			<-timeAfter(delayLeft)
 		})