Cut patch release v0.32.1 (#6670)

* store: fix error handling in decodePostings (#6650)

Signed-off-by: Michael Hoffmann <mhoffm@posteo.de>

* store: fix ignored error in postings (#6654)

Signed-off-by: Michael Hoffmann <mhoffm@posteo.de>

* Store: fix bufio pool handling (#6655)

Signed-off-by: Michael Hoffmann <mhoffm@posteo.de>

* Fix mutable stringset memory usage (#6669)

This commit fixes the Insert function for the mutable stringset
to only insert unique labels instead of adding every label to the set.

Signed-off-by: Filip Petkovski <filip.petkovsky@gmail.com>

* Cut patch release v0.32.1

Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com>

---------

Signed-off-by: Michael Hoffmann <mhoffm@posteo.de>
Signed-off-by: Filip Petkovski <filip.petkovsky@gmail.com>
Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com>
Co-authored-by: Michael Hoffmann <mhoffm@posteo.de>
Co-authored-by: Filip Petkovski <filip.petkovsky@gmail.com>

CHANGELOG.md

@@ -18,7 +18,22 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 
 ### Removed
 
-## [v0.32.0](https://github.com/thanos-io/thanos/tree/release-v0.32.0) - 23.08.2023
+## [v0.32.1](https://github.com/thanos-io/thanos/tree/release-0.32) - 28.08.2023
+
+### Fixed
+
+- [#6650](https://github.com/thanos-io/thanos/pull/6650) Store: fix error handling in decodePostings
+- [#6654](https://github.com/thanos-io/thanos/pull/6654) Store: fix ignored error in postings
+- [#6655](https://github.com/thanos-io/thanos/pull/6655) Store: fix bufio pool handling
+- [#6669](https://github.com/thanos-io/thanos/pull/6669) Store: Fix mutable stringset memory usage
+
+### Added
+
+### Changed
+
+### Removed
+
+## [v0.32.0](https://github.com/thanos-io/thanos/tree/release-0.32) - 23.08.2023
 
 ### Added
 

VERSION

@@ -1 +1 @@
-0.32.0
+0.32.1

pkg/store/bucket.go

@@ -2792,13 +2792,13 @@ func (r *bucketIndexReader) fetchPostings(ctx context.Context, keys []labels.Lab
 		// We assume index does not have any ptrs that has 0 length.
 		length := int64(part.End) - start
 
-		brdr := bufioReaderPool.Get().(*bufio.Reader)
-		defer bufioReaderPool.Put(brdr)
-
 		// Fetch from object storage concurrently and update stats and posting list.
 		g.Go(func() error {
 			begin := time.Now()
 
+			brdr := bufioReaderPool.Get().(*bufio.Reader)
+			defer bufioReaderPool.Put(brdr)
+
 			partReader, err := r.block.bkt.GetRange(ctx, r.block.indexFilename(), start, length)
 			if err != nil {
 				return errors.Wrap(err, "read postings range")
@@ -2864,14 +2864,13 @@ func (r *bucketIndexReader) decodeCachedPostings(b []byte) (index.Postings, []fu
 	)
 	if isDiffVarintSnappyEncodedPostings(b) || isDiffVarintSnappyStreamedEncodedPostings(b) {
 		s := time.Now()
-		clPostings, err := decodePostings(b)
+		l, err = decodePostings(b)
 		r.stats.cachedPostingsDecompressions += 1
 		r.stats.CachedPostingsDecompressionTimeSum += time.Since(s)
 		if err != nil {
 			r.stats.cachedPostingsDecompressionErrors += 1
 		} else {
-			closeFns = append(closeFns, clPostings.close)
-			l = clPostings
+			closeFns = append(closeFns, l.(closeablePostings).close)
 		}
 	} else {
 		_, l, err = r.dec.Postings(b)

pkg/store/bucket_e2e_test.go

@@ -799,6 +799,7 @@ func TestBucketStore_LabelNamesSet_e2e(t *testing.T) {
 		for _, n := range []string{"a", "b", "c"} {
 			testutil.Assert(t, filter.Has(n), "expected filter to have %s", n)
 		}
+		testutil.Equals(t, 3, filter.Count())
 	})
 }
 

pkg/store/bucket_test.go

@@ -3330,3 +3330,15 @@ func TestExpandedPostingsRace(t *testing.T) {
 		wg.Wait()
 	}
 }
+
+func TestBucketIndexReader_decodeCachedPostingsErrors(t *testing.T) {
+	bir := bucketIndexReader{stats: &queryStats{}}
+	t.Run("should return error on broken cached postings without snappy prefix", func(t *testing.T) {
+		_, _, err := bir.decodeCachedPostings([]byte("foo"))
+		testutil.NotOk(t, err)
+	})
+	t.Run("should return error on broken cached postings with snappy prefix", func(t *testing.T) {
+		_, _, err := bir.decodeCachedPostings(append([]byte(codecHeaderSnappy), []byte("foo")...))
+		testutil.NotOk(t, err)
+	})
+}

pkg/store/postings.go

@@ -83,6 +83,7 @@ func (r *postingsReaderBuilder) Next() bool {
 
 	_, err := r.r.Discard(int(from - r.lastOffset))
 	if err != nil {
+		r.e = err
 		return false
 	}
 	r.lastOffset += from - r.lastOffset

pkg/stringset/set.go

@@ -10,6 +10,10 @@ import (
 type Set interface {
 	Has(string) bool
 	HasAny([]string) bool
+	// Count returns the number of elements in the set.
+	// A value of -1 indicates infinite size and can be returned by a
+	// set representing all possible string values.
+	Count() int
 }
 
 type fixedSet struct {
@@ -38,6 +42,10 @@ func (f fixedSet) Has(s string) bool {
 	return f.cuckoo.Lookup([]byte(s))
 }
 
+func (f fixedSet) Count() int {
+	return int(f.cuckoo.Count())
+}
+
 type mutableSet struct {
 	cuckoo *cuckoo.ScalableCuckooFilter
 }
@@ -54,7 +62,7 @@ func New() MutableSet {
 }
 
 func (e mutableSet) Insert(s string) {
-	e.cuckoo.Insert([]byte(s))
+	e.cuckoo.InsertUnique([]byte(s))
 }
 
 func (e mutableSet) Has(s string) bool {
@@ -70,6 +78,10 @@ func (e mutableSet) HasAny(strings []string) bool {
 	return false
 }
 
+func (e mutableSet) Count() int {
+	return int(e.cuckoo.Count())
+}
+
 type allStringsSet struct{}
 
 func (e allStringsSet) HasAny(_ []string) bool {
@@ -83,3 +95,7 @@ func AllStrings() *allStringsSet {
 func (e allStringsSet) Has(_ string) bool {
 	return true
 }
+
+func (e allStringsSet) Count() int {
+	return -1
+}