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
+}