Adopt to new remote read client interface

Signed-off-by: György Krajcsovits <[email protected]>

Author: krajorama

pkg/distributor/otel.go

@@ -429,7 +429,7 @@ func otelMetricsToTimeseries(ctx context.Context, tenantID string, addSuffixes b
 // Old, less efficient, version of otelMetricsToTimeseries.
 func otelMetricsToTimeseriesOld(ctx context.Context, tenantID string, addSuffixes bool, discardedDueToOtelParseError *prometheus.CounterVec, logger log.Logger, md pmetric.Metrics) ([]mimirpb.PreallocTimeseries, error) {
 	converter := prometheusremotewrite.NewPrometheusConverter()
-	errs := converter.FromMetrics(ctx, md, prometheusremotewrite.Settings{
+	_, errs := converter.FromMetrics(ctx, md, prometheusremotewrite.Settings{
 		AddMetricSuffixes: addSuffixes,
 	})
 	promTS := converter.TimeSeries()

pkg/mimirtool/backfill/backfill.go

@@ -18,6 +18,7 @@ import (
 	"github.com/go-kit/log"
 	"github.com/pkg/errors"
 	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/histogram"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/tsdb"
 	tsdb_errors "github.com/prometheus/prometheus/tsdb/errors"
@@ -66,7 +67,7 @@ func getFormattedBytes(bytes int64, humanReadable bool) string {
 
 type Iterator interface {
 	Next() error
-	Sample() (ts int64, v float64)
+	Sample() (ts int64, v float64, h *histogram.Histogram, fh *histogram.FloatHistogram)
 	Labels() (l labels.Labels)
 }
 
@@ -117,13 +118,18 @@ func CreateBlocks(input IteratorCreator, mint, maxt int64, maxSamplesInAppender
 					return errors.Wrap(err, "input")
 				}
 
-				ts, v := i.Sample()
+				ts, v, h, fh := i.Sample()
 				if ts < t || ts >= tsUpper {
 					continue
 				}
 
 				l := i.Labels()
-				if _, err := app.Append(0, i.Labels(), ts, v); err != nil {
+				if h != nil || fh != nil {
+					_, err = app.AppendHistogram(0, i.Labels(), ts, h, fh)
+				} else {
+					_, err = app.Append(0, i.Labels(), ts, v)
+				}
+				if err != nil {
 					return errors.Wrap(err, fmt.Sprintf("add sample for metric=%s ts=%s value=%f", l, model.Time(ts).Time().Format(time.RFC3339Nano), v))
 				}
 

pkg/mimirtool/commands/remote_read.go

@@ -21,14 +21,15 @@ import (
 	"time"
 
 	"github.com/alecthomas/kingpin/v2"
-	"github.com/pkg/errors"
 	config_util "github.com/prometheus/common/config"
 	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/histogram"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/model/value"
-	"github.com/prometheus/prometheus/prompb"
 	"github.com/prometheus/prometheus/promql/parser"
+	"github.com/prometheus/prometheus/storage"
 	"github.com/prometheus/prometheus/storage/remote"
+	"github.com/prometheus/prometheus/tsdb/chunkenc"
 	log "github.com/sirupsen/logrus"
 
 	"github.com/grafana/mimir/pkg/mimirtool/backfill"
@@ -105,66 +106,59 @@ func (s *setTenantIDTransport) RoundTrip(req *http.Request) (*http.Response, err
 }
 
 type timeSeriesIterator struct {
-	posSeries int
-	posSample int
-	ts        []*prompb.TimeSeries
-
-	// labels slice is reused across samples within a series
-	labels          labels.Labels
-	labelsSeriesPos int
+	ss storage.SeriesSet
+	it chunkenc.Iterator
+	ts int64
+	v  float64
+	h  *histogram.Histogram
+	fh *histogram.FloatHistogram
 }
 
-func newTimeSeriesIterator(ts []*prompb.TimeSeries) *timeSeriesIterator {
+func newTimeSeriesIterator(seriesSet storage.SeriesSet) *timeSeriesIterator {
 	return &timeSeriesIterator{
-		posSeries: 0,
-		posSample: -1,
-
-		// ensure we are not pointing to a valid slice position
-		labelsSeriesPos: -1,
-
-		ts: ts,
+		ss: seriesSet,
+		it: chunkenc.NewNopIterator(),
 	}
-
 }
 
 func (i *timeSeriesIterator) Next() error {
-	if i.posSeries >= len(i.ts) {
-		return io.EOF
-	}
-
-	i.posSample++
-
-	if i.posSample >= len(i.ts[i.posSeries].Samples) {
-		i.posSample = -1
-		i.posSeries++
-		return i.Next()
+	// Find non empty series iterator.
+	var vt chunkenc.ValueType
+	for vt = i.it.Next(); vt == chunkenc.ValNone; vt = i.it.Next() {
+		if !i.ss.Next() {
+			err := i.ss.Err()
+			if err != nil {
+				return err
+			}
+			return io.EOF
+		}
+		i.it = i.ss.At().Iterator(i.it)
+	}
+	switch vt {
+	case chunkenc.ValFloat:
+		i.ts, i.v = i.it.At()
+		i.h = nil
+		i.fh = nil
+	case chunkenc.ValHistogram:
+		i.ts, i.h = i.it.AtHistogram(nil)
+		i.v = i.h.Sum
+		i.fh = nil
+	case chunkenc.ValFloatHistogram:
+		i.ts, i.fh = i.it.AtFloatHistogram(nil)
+		i.v = i.fh.Sum
+		i.h = nil
+	default:
+		panic("unreachable")
 	}
-
 	return nil
 }
 
 func (i *timeSeriesIterator) Labels() (l labels.Labels) {
-	// if it's the same label as previously return it
-	if i.posSeries == i.labelsSeriesPos {
-		return i.labels
-	}
-
-	series := i.ts[i.posSeries]
-	builder := labels.NewScratchBuilder(len(series.Labels))
-	for posLabel := range series.Labels {
-		builder.Add(series.Labels[posLabel].Name, series.Labels[posLabel].Value)
-	}
-	i.labels = builder.Labels()
-	i.labelsSeriesPos = i.posSeries
-	return i.labels
+	return i.ss.At().Labels()
 }
 
-func (i *timeSeriesIterator) Sample() (ts int64, v float64) {
-	series := i.ts[i.posSeries]
-	sample := series.Samples[i.posSample]
-
-	//sample.GetValue()
-	return sample.GetTimestamp(), sample.GetValue()
+func (i *timeSeriesIterator) Sample() (ts int64, v float64, h *histogram.Histogram, fh *histogram.FloatHistogram) {
+	return i.ts, i.v, i.h, i.fh
 }
 
 // this is adapted from Go 1.15 for older versions
@@ -231,7 +225,7 @@ func (c *RemoteReadCommand) readClient() (remote.ReadClient, error) {
 }
 
 // prepare() validates the input and prepares the client to query remote read endpoints
-func (c *RemoteReadCommand) prepare() (query func(context.Context) ([]*prompb.TimeSeries, error), from, to time.Time, err error) {
+func (c *RemoteReadCommand) prepare() (query func(context.Context) (storage.SeriesSet, error), from, to time.Time, err error) {
 	from, err = time.Parse(time.RFC3339, c.from)
 	if err != nil {
 		return nil, time.Time{}, time.Time{}, fmt.Errorf("error parsing from: '%s' value: %w", c.from, err)
@@ -262,14 +256,14 @@ func (c *RemoteReadCommand) prepare() (query func(context.Context) ([]*prompb.Ti
 		return nil, time.Time{}, time.Time{}, err
 	}
 
-	return func(ctx context.Context) ([]*prompb.TimeSeries, error) {
+	return func(ctx context.Context) (storage.SeriesSet, error) {
 		log.Infof("Querying time from=%s to=%s with selector=%s", from.Format(time.RFC3339), to.Format(time.RFC3339), c.selector)
-		resp, err := readClient.Read(ctx, pbQuery)
+		resp, err := readClient.Read(ctx, pbQuery, false)
 		if err != nil {
 			return nil, err
 		}
 
-		return resp.Timeseries, nil
+		return resp, nil
 
 	}, from, to, nil
 }
@@ -285,23 +279,39 @@ func (c *RemoteReadCommand) dump(_ *kingpin.ParseContext) error {
 		return err
 	}
 
-	iterator := newTimeSeriesIterator(timeseries)
-	for {
-		err := iterator.Next()
-		if err != nil {
-			if errors.Is(err, io.EOF) {
-				break
+	var it chunkenc.Iterator
+	for timeseries.Next() {
+		s := timeseries.At()
+
+		l := s.Labels().String()
+		it := s.Iterator(it)
+		for vt := it.Next(); vt != chunkenc.ValNone; vt = it.Next() {
+			switch vt {
+			case chunkenc.ValFloat:
+				ts, v := it.At()
+				comment := ""
+				if value.IsStaleNaN(v) {
+					comment = " # StaleNaN"
+				}
+				fmt.Printf("%s %g %d%s\n", l, v, ts, comment)
+			case chunkenc.ValHistogram:
+				ts, h := it.AtHistogram(nil)
+				comment := ""
+				if value.IsStaleNaN(h.Sum) {
+					comment = " # StaleNaN"
+				}
+				fmt.Printf("%s %s %d%s\n", l, h.String(), ts, comment)
+			case chunkenc.ValFloatHistogram:
+				ts, h := it.AtFloatHistogram(nil)
+				comment := ""
+				if value.IsStaleNaN(h.Sum) {
+					comment = " # StaleNaN"
+				}
+				fmt.Printf("%s %s %d%s\n", l, h.String(), ts, comment)
+			default:
+				panic("unreachable")
 			}
-			return err
-		}
-
-		l := iterator.Labels()
-		ts, v := iterator.Sample()
-		comment := ""
-		if value.IsStaleNaN(v) {
-			comment = " # StaleNaN"
 		}
-		fmt.Printf("%s %g %d%s\n", l, v, ts, comment)
 	}
 
 	return nil
@@ -331,32 +341,44 @@ func (c *RemoteReadCommand) stats(_ *kingpin.ParseContext) error {
 		Series: make(map[string]struct{}),
 	}
 
-	iterator := newTimeSeriesIterator(timeseries)
-	for {
-		err := iterator.Next()
-		if err != nil {
-			if errors.Is(err, io.EOF) {
-				break
+	var it chunkenc.Iterator
+	for timeseries.Next() {
+		s := timeseries.At()
+		it := s.Iterator(it)
+		for vt := it.Next(); vt != chunkenc.ValNone; vt = it.Next() {
+			num.Samples++
+			num.Series[s.Labels().String()] = struct{}{}
+
+			var ts int64
+			var v float64
+			switch vt {
+			case chunkenc.ValFloat:
+				ts, v = it.At()
+			case chunkenc.ValHistogram:
+				var h *histogram.Histogram
+				ts, h = it.AtHistogram(nil)
+				v = h.Sum
+			case chunkenc.ValFloatHistogram:
+				var h *histogram.FloatHistogram
+				ts, h = it.AtFloatHistogram(nil)
+				v = h.Sum
+			default:
+				panic("unreachable")
 			}
-			return err
-		}
-		num.Samples++
-		num.Series[iterator.Labels().String()] = struct{}{}
-
-		ts, v := iterator.Sample()
 
-		if int64(num.MaxT) < ts {
-			num.MaxT = model.Time(ts)
-		}
-		if num.MinT == 0 || int64(num.MinT) > ts {
-			num.MinT = model.Time(ts)
-		}
+			if int64(num.MaxT) < ts {
+				num.MaxT = model.Time(ts)
+			}
+			if num.MinT == 0 || int64(num.MinT) > ts {
+				num.MinT = model.Time(ts)
+			}
 
-		if math.IsNaN(v) {
-			num.NaNValues++
-		}
-		if value.IsStaleNaN(v) {
-			num.StaleNaNValues++
+			if math.IsNaN(v) {
+				num.NaNValues++
+			}
+			if value.IsStaleNaN(v) {
+				num.StaleNaNValues++
+			}
 		}
 	}