removing unused code/types from ingester (#5969)

* removing unused code/types from ingester

Signed-off-by: alanprot <[email protected]>

* lint

Signed-off-by: alanprot <[email protected]>

* fix test

Signed-off-by: alanprot <[email protected]>

* Fix test

Signed-off-by: alanprot <[email protected]>

* rebasing from master

Signed-off-by: alanprot <[email protected]>

* return serieset error on MatrixFromSeriesSet

Signed-off-by: alanprot <[email protected]>

* cleaning up streamingSelect method

Signed-off-by: alanprot <[email protected]>

* checking it.Err() on MatrixFromSeriesSet

Signed-off-by: alanprot <[email protected]>

* fix TestNewDistributorsCanPushToOldIngestersWithReplication

Signed-off-by: alanprot <[email protected]>

---------

Signed-off-by: alanprot <[email protected]>

Author: alanprot

integration/backward_compatibility_test.go

@@ -21,8 +21,14 @@ var (
 	// If you change the image tag, remember to update it in the preloading done
 	// by GitHub Actions too (see .github/workflows/test-build-deploy.yml).
 	previousVersionImages = map[string]func(map[string]string) map[string]string{
-		"quay.io/cortexproject/cortex:v1.13.1": nil,
-		"quay.io/cortexproject/cortex:v1.13.2": nil,
+		"quay.io/cortexproject/cortex:v1.13.1": func(m map[string]string) map[string]string {
+			m["-ingester.stream-chunks-when-using-blocks"] = "true"
+			return m
+		},
+		"quay.io/cortexproject/cortex:v1.13.2": func(m map[string]string) map[string]string {
+			m["-ingester.stream-chunks-when-using-blocks"] = "true"
+			return m
+		},
 		"quay.io/cortexproject/cortex:v1.14.0": nil,
 		"quay.io/cortexproject/cortex:v1.14.1": nil,
 		"quay.io/cortexproject/cortex:v1.15.0": nil,

pkg/distributor/query.go

@@ -70,7 +70,7 @@ func (d *Distributor) QueryStream(ctx context.Context, from, to model.Time, matc
 		}
 
 		if s := opentracing.SpanFromContext(ctx); s != nil {
-			s.LogKV("chunk-series", len(result.GetChunkseries()), "time-series", len(result.GetTimeseries()))
+			s.LogKV("chunk-series", len(result.GetChunkseries()))
 		}
 		return nil
 	})
@@ -253,15 +253,11 @@ func (d *Distributor) queryIngesterStream(ctx context.Context, replicationSet ri
 				return nil, validation.LimitError(chunkLimitErr.Error())
 			}
 
-			s := make([][]cortexpb.LabelAdapter, 0, len(resp.Chunkseries)+len(resp.Timeseries))
+			s := make([][]cortexpb.LabelAdapter, 0, len(resp.Chunkseries))
 			for _, series := range resp.Chunkseries {
 				s = append(s, series.Labels)
 			}
 
-			for _, series := range resp.Timeseries {
-				s = append(s, series.Labels)
-			}
-
 			if limitErr := queryLimiter.AddSeries(s...); limitErr != nil {
 				return nil, validation.LimitError(limitErr.Error())
 			}
@@ -275,7 +271,6 @@ func (d *Distributor) queryIngesterStream(ctx context.Context, replicationSet ri
 			}
 
 			result.Chunkseries = append(result.Chunkseries, resp.Chunkseries...)
-			result.Timeseries = append(result.Timeseries, resp.Timeseries...)
 		}
 		return result, nil
 	})
@@ -286,7 +281,6 @@ func (d *Distributor) queryIngesterStream(ctx context.Context, replicationSet ri
 	span, _ := opentracing.StartSpanFromContext(ctx, "Distributor.MergeIngesterStreams")
 	defer span.Finish()
 	hashToChunkseries := map[string]ingester_client.TimeSeriesChunk{}
-	hashToTimeSeries := map[string]cortexpb.TimeSeries{}
 
 	for _, result := range results {
 		response := result.(*ingester_client.QueryStreamResponse)
@@ -299,40 +293,23 @@ func (d *Distributor) queryIngesterStream(ctx context.Context, replicationSet ri
 			existing.Chunks = append(existing.Chunks, series.Chunks...)
 			hashToChunkseries[key] = existing
 		}
-
-		// Parse any time series
-		for _, series := range response.Timeseries {
-			key := ingester_client.LabelsToKeyString(cortexpb.FromLabelAdaptersToLabels(series.Labels))
-			existing := hashToTimeSeries[key]
-			existing.Labels = series.Labels
-			if existing.Samples == nil {
-				existing.Samples = series.Samples
-			} else {
-				existing.Samples = mergeSamples(existing.Samples, series.Samples)
-			}
-			hashToTimeSeries[key] = existing
-		}
 	}
 
 	resp := &ingester_client.QueryStreamResponse{
 		Chunkseries: make([]ingester_client.TimeSeriesChunk, 0, len(hashToChunkseries)),
-		Timeseries:  make([]cortexpb.TimeSeries, 0, len(hashToTimeSeries)),
 	}
 	for _, series := range hashToChunkseries {
 		resp.Chunkseries = append(resp.Chunkseries, series)
 	}
-	for _, series := range hashToTimeSeries {
-		resp.Timeseries = append(resp.Timeseries, series)
-	}
 
 	respSize := resp.Size()
 	chksSize := resp.ChunksSize()
 	chksCount := resp.ChunksCount()
-	span.SetTag("fetched_series", len(resp.Chunkseries)+len(resp.Timeseries))
+	span.SetTag("fetched_series", len(resp.Chunkseries))
 	span.SetTag("fetched_chunks", chksCount)
 	span.SetTag("fetched_data_bytes", respSize)
 	span.SetTag("fetched_chunks_bytes", chksSize)
-	reqStats.AddFetchedSeries(uint64(len(resp.Chunkseries) + len(resp.Timeseries)))
+	reqStats.AddFetchedSeries(uint64(len(resp.Chunkseries)))
 	reqStats.AddFetchedChunkBytes(uint64(chksSize))
 	reqStats.AddFetchedDataBytes(uint64(respSize))
 	reqStats.AddFetchedChunks(uint64(chksCount))

pkg/ingester/client/compat.go

@@ -5,6 +5,8 @@ import (
 
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/storage"
+	"github.com/prometheus/prometheus/tsdb/chunkenc"
 
 	"github.com/cortexproject/cortex/pkg/cortexpb"
 )
@@ -66,6 +68,32 @@ func FromExemplarQueryRequest(req *ExemplarQueryRequest) (int64, int64, [][]*lab
 	return req.StartTimestampMs, req.EndTimestampMs, result, nil
 }
 
+// MatrixFromSeriesSet unpacks a SeriesSet to a model.Matrix.
+func MatrixFromSeriesSet(set storage.SeriesSet) (model.Matrix, error) {
+	m := make(model.Matrix, 0)
+	for set.Next() {
+		s := set.At()
+		var ss model.SampleStream
+		ss.Metric = cortexpb.FromLabelAdaptersToMetric(cortexpb.FromLabelsToLabelAdapters(s.Labels()))
+		ss.Values = make([]model.SamplePair, 0)
+		it := s.Iterator(nil)
+		for it.Next() != chunkenc.ValNone {
+			t, v := it.At()
+			ss.Values = append(ss.Values, model.SamplePair{
+				Value:     model.SampleValue(v),
+				Timestamp: model.Time(t),
+			})
+		}
+		if it.Err() != nil {
+			return nil, it.Err()
+		}
+
+		m = append(m, &ss)
+	}
+
+	return m, set.Err()
+}
+
 // ToQueryResponse builds a QueryResponse proto.
 func ToQueryResponse(matrix model.Matrix) *QueryResponse {
 	resp := &QueryResponse{}
@@ -118,6 +146,34 @@ func ToMetricsForLabelMatchersRequest(from, to model.Time, matchers []*labels.Ma
 	}, nil
 }
 
+// SeriesSetToQueryResponse builds a QueryResponse proto
+func SeriesSetToQueryResponse(s storage.SeriesSet) (*QueryResponse, error) {
+	result := &QueryResponse{}
+
+	var it chunkenc.Iterator
+	for s.Next() {
+		series := s.At()
+		samples := []cortexpb.Sample{}
+		it = series.Iterator(it)
+		for it.Next() != chunkenc.ValNone {
+			t, v := it.At()
+			samples = append(samples, cortexpb.Sample{
+				TimestampMs: t,
+				Value:       v,
+			})
+		}
+		if err := it.Err(); err != nil {
+			return nil, err
+		}
+		result.Timeseries = append(result.Timeseries, cortexpb.TimeSeries{
+			Labels:  cortexpb.FromLabelsToLabelAdapters(series.Labels()),
+			Samples: samples,
+		})
+	}
+
+	return result, s.Err()
+}
+
 // FromMetricsForLabelMatchersRequest unpacks a MetricsForLabelMatchersRequest proto
 func FromMetricsForLabelMatchersRequest(req *MetricsForLabelMatchersRequest) (model.Time, model.Time, [][]*labels.Matcher, error) {
 	matchersSet := make([][]*labels.Matcher, 0, len(req.MatchersSet))
@@ -133,15 +189,6 @@ func FromMetricsForLabelMatchersRequest(req *MetricsForLabelMatchersRequest) (mo
 	return from, to, matchersSet, nil
 }
 
-// FromMetricsForLabelMatchersResponse unpacks a MetricsForLabelMatchersResponse proto
-func FromMetricsForLabelMatchersResponse(resp *MetricsForLabelMatchersResponse) []model.Metric {
-	metrics := []model.Metric{}
-	for _, m := range resp.Metric {
-		metrics = append(metrics, cortexpb.FromLabelAdaptersToMetric(m.Labels))
-	}
-	return metrics
-}
-
 // ToLabelValuesRequest builds a LabelValuesRequest proto
 func ToLabelValuesRequest(labelName model.LabelName, from, to model.Time, matchers []*labels.Matcher) (*LabelValuesRequest, error) {
 	ms, err := toLabelMatchers(matchers)

pkg/ingester/client/custom.go

@@ -35,9 +35,6 @@ func (m *QueryStreamResponse) ChunksSize() int {
 }
 
 func (m *QueryStreamResponse) SamplesCount() (count int) {
-	for _, ts := range m.Timeseries {
-		count += len(ts.Samples)
-	}
 	for _, cs := range m.Chunkseries {
 		for _, c := range cs.Chunks {
 			if c.Encoding == int32(encoding.PrometheusXorChunk) {