test

Author: yeya24

pkg/querier/tripperware/instantquery/instantquery.proto

@@ -44,5 +44,5 @@ message Sample {
 }
 
 message Matrix {
-  repeated tripperware.SampleStream sampleStreams = 1;
+  repeated tripperware.SampleStream sampleStreams = 1 [(gogoproto.nullable) = false];
 }

pkg/querier/tripperware/merge.go

@@ -0,0 +1,54 @@
+package tripperware
+
+import (
+	"github.com/cortexproject/cortex/pkg/cortexpb"
+	"github.com/cortexproject/cortex/pkg/querier/tripperware/utils"
+	"sort"
+)
+
+type SampleStreamResponse interface {
+	GetSampleStreams() []*SampleStream
+}
+
+func MatrixMerge(resps []SampleStreamResponse) []*SampleStream {
+	output := map[string]*SampleStream{}
+	for _, resp := range resps {
+		for _, stream := range resp.GetSampleStreams() {
+			metric := cortexpb.FromLabelAdaptersToLabels(stream.Labels).String()
+			existing, ok := output[metric]
+			if !ok {
+				existing = &SampleStream{
+					Labels: stream.Labels,
+				}
+			}
+			// We need to make sure we don't repeat samples. This causes some visualisations to be broken in Grafana.
+			// The prometheus API is inclusive of start and end timestamps.
+			if len(existing.Samples) > 0 && len(stream.Samples) > 0 {
+				existingEndTs := existing.Samples[len(existing.Samples)-1].TimestampMs
+				if existingEndTs == stream.Samples[0].TimestampMs {
+					// Typically this the cases where only 1 sample point overlap,
+					// so optimize with simple code.
+					stream.Samples = stream.Samples[1:]
+				} else if existingEndTs > stream.Samples[0].TimestampMs {
+					// Overlap might be big, use heavier algorithm to remove overlap.
+					stream.Samples = utils.SliceSamples(stream.Samples, existingEndTs)
+				} // else there is no overlap, yay!
+			}
+			existing.Samples = append(existing.Samples, stream.Samples...)
+			output[metric] = existing
+		}
+	}
+
+	keys := make([]string, 0, len(output))
+	for key := range output {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+
+	result := make([]SampleStream, 0, len(output))
+	for _, key := range keys {
+		result = append(result, *output[key])
+	}
+
+	return result
+}

pkg/querier/tripperware/queryrange/query_range.go

@@ -21,9 +21,7 @@ import (
 	"github.com/prometheus/prometheus/model/timestamp"
 	"github.com/weaveworks/common/httpgrpc"
 
-	"github.com/cortexproject/cortex/pkg/cortexpb"
 	"github.com/cortexproject/cortex/pkg/querier/tripperware"
-	"github.com/cortexproject/cortex/pkg/querier/tripperware/utils"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 )
@@ -150,7 +148,7 @@ func (c prometheusCodec) MergeResponse(responses ...tripperware.Response) (tripp
 		Status: StatusSuccess,
 		Data: PrometheusData{
 			ResultType: model.ValMatrix.String(),
-			Result:     matrixMerge(promResponses),
+			Result:     tripperware.MatrixMerge(promResponses),
 			Stats:      statsMerge(c.sharded, promResponses),
 		},
 	}
@@ -345,49 +343,6 @@ func statsMerge(shouldSumStats bool, resps []*PrometheusResponse) *tripperware.P
 	return tripperware.StatsMerge(output)
 }
 
-func matrixMerge(resps []*PrometheusResponse) []tripperware.SampleStream {
-	output := map[string]*tripperware.SampleStream{}
-	for _, resp := range resps {
-		for _, stream := range resp.Data.Result {
-			metric := cortexpb.FromLabelAdaptersToLabels(stream.Labels).String()
-			existing, ok := output[metric]
-			if !ok {
-				existing = &tripperware.SampleStream{
-					Labels: stream.Labels,
-				}
-			}
-			// We need to make sure we don't repeat samples. This causes some visualisations to be broken in Grafana.
-			// The prometheus API is inclusive of start and end timestamps.
-			if len(existing.Samples) > 0 && len(stream.Samples) > 0 {
-				existingEndTs := existing.Samples[len(existing.Samples)-1].TimestampMs
-				if existingEndTs == stream.Samples[0].TimestampMs {
-					// Typically this the cases where only 1 sample point overlap,
-					// so optimize with simple code.
-					stream.Samples = stream.Samples[1:]
-				} else if existingEndTs > stream.Samples[0].TimestampMs {
-					// Overlap might be big, use heavier algorithm to remove overlap.
-					stream.Samples = utils.SliceSamples(stream.Samples, existingEndTs)
-				} // else there is no overlap, yay!
-			}
-			existing.Samples = append(existing.Samples, stream.Samples...)
-			output[metric] = existing
-		}
-	}
-
-	keys := make([]string, 0, len(output))
-	for key := range output {
-		keys = append(keys, key)
-	}
-	sort.Strings(keys)
-
-	result := make([]tripperware.SampleStream, 0, len(output))
-	for _, key := range keys {
-		result = append(result, *output[key])
-	}
-
-	return result
-}
-
 func parseDurationMs(s string) (int64, error) {
 	if d, err := strconv.ParseFloat(s, 64); err == nil {
 		ts := d * float64(time.Second/time.Millisecond)