Add a span for OpenTelemetry to Prometheus push request conversion in distributors

CHANGELOG.md

@@ -55,6 +55,7 @@
 * [ENHANCEMENT] Distributor: add support for ingesting exponential histograms that are over the native histogram scale limit of 8 in OpenTelemetry format by downscaling them. #5532
 * [ENHANCEMENT] General: buffered logging: #5506
   * `-log.buffered`: Enable buffered logging
+* [ENHANCEMENT] Distributor: add more detailed information to traces generated while processing OTLP write requests. #5539
 * [BUGFIX] Ingester: Handle when previous ring state is leaving and the number of tokens has changed. #5204
 * [BUGFIX] Querier: fix issue where queries that use the `timestamp()` function fail with `execution: attempted to read series at index 0 from stream, but the stream has already been exhausted` if streaming chunks from ingesters to queriers is enabled. #5370
 * [BUGFIX] memberlist: bring back `memberlist_client_kv_store_count` metric that used to exist in Cortex, but got lost during dskit updates before Mimir 2.0. #5377

pkg/util/push/otel.go

@@ -27,6 +27,7 @@ import (
 	"github.com/grafana/mimir/pkg/mimirpb"
 	"github.com/grafana/mimir/pkg/util"
 	"github.com/grafana/mimir/pkg/util/log"
+	"github.com/grafana/mimir/pkg/util/spanlogger"
 	"github.com/grafana/mimir/pkg/util/validation"
 )
 
@@ -76,7 +77,8 @@ func OTLPHandler(
 
 		reader := r.Body
 		// Handle compression.
-		switch r.Header.Get("Content-Encoding") {
+		contentEncoding := r.Header.Get("Content-Encoding")
+		switch contentEncoding {
 		case "gzip":
 			gr, err := gzip.NewReader(reader)
 			if err != nil {
@@ -88,7 +90,7 @@ func OTLPHandler(
 			// No compression.
 
 		default:
-			return nil, httpgrpc.Errorf(http.StatusUnsupportedMediaType, "unsupported compression: %s. Only \"gzip\" or no compression supported", r.Header.Get("Content-Encoding"))
+			return nil, httpgrpc.Errorf(http.StatusUnsupportedMediaType, "unsupported compression: %s. Only \"gzip\" or no compression supported", contentEncoding)
 		}
 
 		// Protect against a large input.
@@ -109,16 +111,44 @@ func OTLPHandler(
 			return body, err
 		}
 
+		log, ctx := spanlogger.NewWithLogger(ctx, logger, "Distributor.OTLPHandler.decodeAndConvert")
+		defer log.Span.Finish()
+
+		log.SetTag("content_type", contentType)
+		log.SetTag("content_encoding", contentEncoding)
+		log.SetTag("content_length", r.ContentLength)
+
 		otlpReq, err := decoderFunc(body)
 		if err != nil {
 			return body, err
 		}
 
+		level.Debug(log).Log("msg", "decoding complete, starting conversion")
+
 		metrics, err := otelMetricsToTimeseries(ctx, discardedDueToOtelParseError, logger, otlpReq.Metrics())
 		if err != nil {
 			return body, err
 		}
 
+		metricCount := len(metrics)
+		sampleCount := 0
+		histogramCount := 0
+		exemplarCount := 0
+
+		for _, m := range metrics {
+			sampleCount += len(m.Samples)
+			histogramCount += len(m.Histograms)
+			exemplarCount += len(m.Exemplars)
+		}
+
+		level.Debug(log).Log(
+			"msg", "OTLP to Prometheus conversion complete",
+			"metric_count", metricCount,
+			"sample_count", sampleCount,
+			"histogram_count", histogramCount,
+			"exemplar_count", exemplarCount,
+		)
+
 		req.Timeseries = metrics
 		return body, nil
 	})