proxystore: stabilize order for duplicates in response losertree

Currently the order of responses in the losertree on collisions in the
labelset is random. This can happen if we drop the replica label in an
endpoint.
In the case of sidecars the order of responses has effect on
deduplication. The primary iterator is used until we find a large enough
gap to failover to the replica iterator, where primary and replica is
determined by the order they are returned from the proxy losertree.
This can lead to slight differences if we repeat a query since different
sidecars have scraped at different times possibly.
Using the store labelset as tiebreaker is an attempt at stabilizing
this.

Signed-off-by: Michael Hoffmann <[email protected]>

Author: MichaHoffmann

pkg/store/proxy_merge.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"math"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -166,12 +167,21 @@ func (d *responseDeduplicator) At() *storepb.SeriesResponse {
 	return d.bufferedResp[d.buffRespI]
 }
 
+type seriesResponseWithStoreLabelset struct {
+	labelset string
+	response *storepb.SeriesResponse
+}
+
+type proxyResponseLoserTree struct {
+	tree *losertree.Tree[seriesResponseWithStoreLabelset, respSet]
+}
+
 // NewProxyResponseLoserTree returns heap that k-way merge series together.
-// It's agnostic to duplicates and overlaps, it forwards all duplicated series in random order.
-func NewProxyResponseLoserTree(seriesSets ...respSet) *losertree.Tree[*storepb.SeriesResponse, respSet] {
-	var maxVal *storepb.SeriesResponse = storepb.NewSeriesResponse(nil)
+// It's agnostic to duplicates and overlaps, it forwards all duplicated series ordered by the labelset of their endpoint.
+func NewProxyResponseLoserTree(seriesSets ...respSet) *proxyResponseLoserTree {
+	var maxVal seriesResponseWithStoreLabelset = seriesResponseWithStoreLabelset{}
 
-	less := func(a, b *storepb.SeriesResponse) bool {
+	less := func(a, b seriesResponseWithStoreLabelset) bool {
 		if a == maxVal && b != maxVal {
 			return false
 		}
@@ -181,29 +191,52 @@ func NewProxyResponseLoserTree(seriesSets ...respSet) *losertree.Tree[*storepb.S
 		if a == maxVal && b == maxVal {
 			return true
 		}
-		if a.GetSeries() != nil && b.GetSeries() != nil {
-			iLbls := labelpb.ZLabelsToPromLabels(a.GetSeries().Labels)
-			jLbls := labelpb.ZLabelsToPromLabels(b.GetSeries().Labels)
+		ar, br := a.response, b.response
 
-			return labels.Compare(iLbls, jLbls) < 0
-		} else if a.GetSeries() == nil && b.GetSeries() != nil {
+		if ar.GetSeries() != nil && br.GetSeries() != nil {
+			aLbls := labelpb.ZLabelsToPromLabels(ar.GetSeries().Labels)
+			bLbls := labelpb.ZLabelsToPromLabels(br.GetSeries().Labels)
+
+			if cmp := labels.Compare(aLbls, bLbls); cmp == 0 {
+				// use store labelset as tiebreaker for a stable order of responses and predictable deduplication
+				return strings.Compare(a.labelset, b.labelset) < 0
+			} else {
+				return cmp < 0
+			}
+		} else if ar.GetSeries() == nil && br.GetSeries() != nil {
 			return true
-		} else if a.GetSeries() != nil && b.GetSeries() == nil {
+		} else if ar.GetSeries() != nil && br.GetSeries() == nil {
 			return false
 		}
 
-		if a.GetWarning() != "" && b.GetWarning() != "" {
-			return len(a.GetWarning()) < len(b.GetWarning())
+		if ar.GetWarning() != "" && br.GetWarning() != "" {
+			return len(ar.GetWarning()) < len(br.GetWarning())
 		}
-
 		return false
 	}
 
-	return losertree.New[*storepb.SeriesResponse, respSet](seriesSets, maxVal, func(s respSet) *storepb.SeriesResponse {
-		return s.At()
-	}, less, func(s respSet) {
-		s.Close()
-	})
+	return &proxyResponseLoserTree{
+		tree: losertree.New(
+			seriesSets,
+			maxVal,
+			func(s respSet) seriesResponseWithStoreLabelset {
+				return seriesResponseWithStoreLabelset{response: s.At(), labelset: s.Labelset()}
+			},
+			less,
+			func(s respSet) { s.Close() },
+		)}
+}
+
+func (lt *proxyResponseLoserTree) Next() bool {
+	return lt.tree.Next()
+}
+
+func (lt *proxyResponseLoserTree) At() *storepb.SeriesResponse {
+	return lt.tree.At().response
+}
+
+func (lt *proxyResponseLoserTree) Close() {
+	lt.tree.Close()
 }
 
 func (l *lazyRespSet) StoreID() string {