instrumentation: add handleSeriesPush span

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/grafana/mimir-proxies v1.1.3-0.20240614200600-38818b085762
 	github.com/influxdata/influxdb-client-go/v2 v2.14.0
 	github.com/influxdata/influxdb/v2 v2.7.10
+	github.com/opentracing/opentracing-go v1.2.1-0.20220228012449-10b1cf09e00b
 	github.com/ory/dockertest/v3 v3.11.0
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.20.4
@@ -134,7 +135,6 @@ require (
 	github.com/opencontainers/runc v1.1.14 // indirect
 	github.com/opentracing-contrib/go-grpc v0.0.0-20210225150812-73cb765af46e // indirect
 	github.com/opentracing-contrib/go-stdlib v1.0.0 // indirect
-	github.com/opentracing/opentracing-go v1.2.1-0.20220228012449-10b1cf09e00b // indirect
 	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect

pkg/influx/api.go

@@ -12,6 +12,8 @@ import (
 	"github.com/grafana/mimir-proxies/pkg/route"
 	"github.com/grafana/mimir-proxies/pkg/server/middleware"
 	"github.com/grafana/mimir/pkg/mimirpb"
+	"github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go/ext"
 )
 
 type API struct {
@@ -40,29 +42,24 @@ func NewAPI(conf ProxyConfig, client remotewrite.Client, recorder Recorder) (*AP
 
 // HandlerForInfluxLine is a http.Handler which accepts Influx Line protocol and converts it to WriteRequests.
 func (a *API) handleSeriesPush(w http.ResponseWriter, r *http.Request) {
-	logger := a.logger
-	if traceID, ok := middleware.ExtractSampledTraceID(r.Context()); ok {
-		logger = log.With(logger, "traceID", traceID)
-	}
-	if orgID, err := user.ExtractOrgID(r.Context()); err == nil {
-		logger = log.With(logger, "orgID", orgID)
-	}
-	if userID, err := user.ExtractUserID(r.Context()); err == nil {
-		logger = log.With(logger, "userID", userID)
-	}
-	logger = log.With(logger, "path", r.URL.EscapedPath())
+	span, ctx := opentracing.StartSpanFromContext(r.Context(), "handleSeriesPush")
+	defer span.Finish()
 
+	logger := withRequestInfo(a.logger, r)
 	beforeConversion := time.Now()
 
-	ts, err, bytesRead := parseInfluxLineReader(r.Context(), r, a.maxRequestSizeBytes)
+	ts, err, bytesRead := parseInfluxLineReader(ctx, r, a.maxRequestSizeBytes)
+	span.LogKV("bytesRead", bytesRead)
 	logger = log.With(logger, "bytesRead", bytesRead)
 	if err != nil {
+		ext.LogError(span, err)
 		a.handleError(w, r, err, logger)
 		return
 	}
 
 	nosMetrics := len(ts)
 	logger = log.With(logger, "nosMetrics", nosMetrics)
+	span.LogKV("nosMetrics", nosMetrics)
 
 	a.recorder.measureMetricsParsed(nosMetrics)
 	a.recorder.measureConversionDuration(time.Since(beforeConversion))
@@ -78,12 +75,28 @@ func (a *API) handleSeriesPush(w http.ResponseWriter, r *http.Request) {
 		Timeseries: pts,
 	}
 
-	if err := a.client.Write(r.Context(), rwReq); err != nil {
+	if err := a.client.Write(ctx, rwReq); err != nil {
+		ext.LogError(span, err)
 		a.handleError(w, r, err, logger)
 		return
 	}
 	a.recorder.measureMetricsWritten(len(rwReq.Timeseries))
+	span.LogKV("nosMetricsWritten", len(rwReq.Timeseries))
 	statusCode := http.StatusNoContent
 	_ = level.Info(logger).Log("response_code", statusCode)
 	w.WriteHeader(statusCode) // Needed for Telegraf, otherwise it tries to marshal JSON and considers the write a failure.
 }
+
+func withRequestInfo(logger log.Logger, r *http.Request) log.Logger {
+	ctx := r.Context()
+	if traceID, ok := middleware.ExtractSampledTraceID(ctx); ok {
+		logger = log.With(logger, "traceID", traceID)
+	}
+	if orgID, err := user.ExtractOrgID(ctx); err == nil {
+		logger = log.With(logger, "orgID", orgID)
+	}
+	if userID, err := user.ExtractUserID(ctx); err == nil {
+		logger = log.With(logger, "userID", userID)
+	}
+	return log.With(logger, "path", r.URL.EscapedPath())
+}