WIP: distributor.OTLPHandler: Convert directly to Mimir format

go.mod

@@ -136,7 +136,7 @@ require (
 	github.com/bits-and-blooms/bitset v1.8.0 // indirect
 	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
 	github.com/cespare/xxhash v1.1.0 // indirect
-	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0
 	github.com/coreos/go-semver v0.3.0 // indirect
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
@@ -235,7 +235,7 @@ require (
 	go.mongodb.org/mongo-driver v1.13.1 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.0.1 // indirect
-	go.opentelemetry.io/collector/semconv v0.93.0 // indirect
+	go.opentelemetry.io/collector/semconv v0.93.0
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.47.0 // indirect
 	go.opentelemetry.io/otel/metric v1.22.0 // indirect
 	go.uber.org/zap v1.21.0 // indirect
@@ -256,7 +256,7 @@ require (
 )
 
 // Using a fork of Prometheus with Mimir-specific changes.
-replace github.com/prometheus/prometheus => github.com/grafana/mimir-prometheus v0.0.0-20240213153929-d32b69f03433
+replace github.com/prometheus/prometheus => github.com/grafana/mimir-prometheus v0.0.0-20240222102121-edcd29643919
 
 // Replace memberlist with our fork which includes some fixes that haven't been
 // merged upstream yet:

go.sum

@@ -505,8 +505,8 @@ github.com/grafana/gomemcache v0.0.0-20231023152154-6947259a0586 h1:/of8Z8taCPft
 github.com/grafana/gomemcache v0.0.0-20231023152154-6947259a0586/go.mod h1:PGk3RjYHpxMM8HFPhKKo+vve3DdlPUELZLSDEFehPuU=
 github.com/grafana/memberlist v0.3.1-0.20220714140823-09ffed8adbbe h1:yIXAAbLswn7VNWBIvM71O2QsgfgW9fRXZNR0DXe6pDU=
 github.com/grafana/memberlist v0.3.1-0.20220714140823-09ffed8adbbe/go.mod h1:MS2lj3INKhZjWNqd3N0m3J+Jxf3DAOnAH9VT3Sh9MUE=
-github.com/grafana/mimir-prometheus v0.0.0-20240213153929-d32b69f03433 h1:mW1Cet0WSipL2fBpfWnUyIlS7cZqvg2/5j6IP6lbogA=
-github.com/grafana/mimir-prometheus v0.0.0-20240213153929-d32b69f03433/go.mod h1:IMaHPfxuCuOjlEoyUE0AG7FGZLEwCg6WHXHLNQwPrJQ=
+github.com/grafana/mimir-prometheus v0.0.0-20240222102121-edcd29643919 h1:UpQVzSLCyhGoohqdL7xyHbVPHYhGC91hzliz7OZGY6Y=
+github.com/grafana/mimir-prometheus v0.0.0-20240222102121-edcd29643919/go.mod h1:IMaHPfxuCuOjlEoyUE0AG7FGZLEwCg6WHXHLNQwPrJQ=
 github.com/grafana/opentracing-contrib-go-stdlib v0.0.0-20230509071955-f410e79da956 h1:em1oddjXL8c1tL0iFdtVtPloq2hRPen2MJQKoAWpxu0=
 github.com/grafana/opentracing-contrib-go-stdlib v0.0.0-20230509071955-f410e79da956/go.mod h1:qtI1ogk+2JhVPIXVc6q+NHziSmy2W5GbdQZFUHADCBU=
 github.com/grafana/pyroscope-go/godeltaprof v0.1.6 h1:nEdZ8louGAplSvIJi1HVp7kWvFvdiiYg3COLlTwJiFo=

pkg/distributor/otel.go

@@ -21,12 +21,12 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/prompb"
 	prometheustranslator "github.com/prometheus/prometheus/storage/remote/otlptranslator/prometheus"
-	"github.com/prometheus/prometheus/storage/remote/otlptranslator/prometheusremotewrite"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
 	"go.uber.org/multierr"
 
+	"github.com/grafana/mimir/pkg/distributor/otlp"
 	"github.com/grafana/mimir/pkg/mimirpb"
 	"github.com/grafana/mimir/pkg/util"
 	"github.com/grafana/mimir/pkg/util/spanlogger"
@@ -262,7 +262,7 @@ func otelMetricsToMetadata(addSuffixes bool, md pmetric.Metrics) []*mimirpb.Metr
 }
 
 func otelMetricsToTimeseries(tenantID string, addSuffixes bool, discardedDueToOtelParseError *prometheus.CounterVec, logger log.Logger, md pmetric.Metrics) ([]mimirpb.PreallocTimeseries, error) {
-	tsMap, errs := prometheusremotewrite.FromMetrics(md, prometheusremotewrite.Settings{
+	ts, errs := otlp.FromMetrics(md, otlp.Settings{
 		AddMetricSuffixes: addSuffixes,
 	})
 	if errs != nil {
@@ -274,104 +274,14 @@ func otelMetricsToTimeseries(tenantID string, addSuffixes bool, discardedDueToOt
 			parseErrs = parseErrs[:maxErrMsgLen]
 		}
 
-		if len(tsMap) == 0 {
+		if len(ts) == 0 {
 			return nil, errors.New(parseErrs)
 		}
 
 		level.Warn(logger).Log("msg", "OTLP parse error", "err", parseErrs)
 	}
 
-	mimirTs := mimirpb.PreallocTimeseriesSliceFromPool()
-	for _, promTs := range tsMap {
-		mimirTs = append(mimirTs, promToMimirTimeseries(promTs))
-	}
-
-	return mimirTs, nil
-}
-
-func promToMimirTimeseries(promTs *prompb.TimeSeries) mimirpb.PreallocTimeseries {
-	labels := make([]mimirpb.LabelAdapter, 0, len(promTs.Labels))
-	for _, label := range promTs.Labels {
-		labels = append(labels, mimirpb.LabelAdapter{
-			Name:  label.Name,
-			Value: label.Value,
-		})
-	}
-
-	samples := make([]mimirpb.Sample, 0, len(promTs.Samples))
-	for _, sample := range promTs.Samples {
-		samples = append(samples, mimirpb.Sample{
-			TimestampMs: sample.Timestamp,
-			Value:       sample.Value,
-		})
-	}
-
-	histograms := make([]mimirpb.Histogram, 0, len(promTs.Histograms))
-	for idx := range promTs.Histograms {
-		histograms = append(histograms, promToMimirHistogram(&promTs.Histograms[idx]))
-	}
-
-	exemplars := make([]mimirpb.Exemplar, 0, len(promTs.Exemplars))
-	for _, exemplar := range promTs.Exemplars {
-		labels := make([]mimirpb.LabelAdapter, 0, len(exemplar.Labels))
-		for _, label := range exemplar.Labels {
-			labels = append(labels, mimirpb.LabelAdapter{
-				Name:  label.Name,
-				Value: label.Value,
-			})
-		}
-
-		exemplars = append(exemplars, mimirpb.Exemplar{
-			Labels:      labels,
-			Value:       exemplar.Value,
-			TimestampMs: exemplar.Timestamp,
-		})
-	}
-
-	ts := mimirpb.TimeseriesFromPool()
-	ts.Labels = labels
-	ts.Samples = samples
-	ts.Histograms = histograms
-	ts.Exemplars = exemplars
-
-	return mimirpb.PreallocTimeseries{TimeSeries: ts}
-}
-
-func promToMimirHistogram(h *prompb.Histogram) mimirpb.Histogram {
-	pSpans := make([]mimirpb.BucketSpan, 0, len(h.PositiveSpans))
-	for _, span := range h.PositiveSpans {
-		pSpans = append(
-			pSpans, mimirpb.BucketSpan{
-				Offset: span.Offset,
-				Length: span.Length,
-			},
-		)
-	}
-	nSpans := make([]mimirpb.BucketSpan, 0, len(h.NegativeSpans))
-	for _, span := range h.NegativeSpans {
-		nSpans = append(
-			nSpans, mimirpb.BucketSpan{
-				Offset: span.Offset,
-				Length: span.Length,
-			},
-		)
-	}
-
-	return mimirpb.Histogram{
-		Count:          &mimirpb.Histogram_CountInt{CountInt: h.GetCountInt()},
-		Sum:            h.Sum,
-		Schema:         h.Schema,
-		ZeroThreshold:  h.ZeroThreshold,
-		ZeroCount:      &mimirpb.Histogram_ZeroCountInt{ZeroCountInt: h.GetZeroCountInt()},
-		NegativeSpans:  nSpans,
-		NegativeDeltas: h.NegativeDeltas,
-		NegativeCounts: h.NegativeCounts,
-		PositiveSpans:  pSpans,
-		PositiveDeltas: h.PositiveDeltas,
-		PositiveCounts: h.PositiveCounts,
-		Timestamp:      h.Timestamp,
-		ResetHint:      mimirpb.Histogram_ResetHint(h.ResetHint),
-	}
+	return ts, nil
 }
 
 // TimeseriesToOTLPRequest is used in tests.