Introduces cache to TSDB postings #9621
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,157 @@ | ||
| // SPDX-License-Identifier: AGPL-3.0-only | ||
| // Provenance-includes-location: https://github.com/thanos-io/thanos/blob/main/pkg/store/postings_codec.go | ||
| // Provenance-includes-license: Apache-2.0 | ||
| // Provenance-includes-copyright: The Thanos Authors. | ||
|
|
||
| package tsdb | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "sort" | ||
| "strings" | ||
|
|
||
| "github.com/go-kit/log" | ||
| "github.com/go-kit/log/level" | ||
| "github.com/prometheus/prometheus/model/labels" | ||
| "github.com/prometheus/prometheus/storage" | ||
|
|
||
| "github.com/grafana/loki/pkg/storage/chunk/cache" | ||
| "github.com/grafana/loki/pkg/storage/stores/tsdb/index" | ||
| ) | ||
|
|
||
| type PostingsReader interface { | ||
|
There was a problem hiding this comment. nit: I think I suggested this name in one of my previous reviews, but I don't think it really fits. It's not an interface for reading the postings, but rather for applying a function over the read postings. Maybe you can find a more suitable name. There was a problem hiding this comment. hmm I think Iterator is misleading; it doesn't support the Iterator interface/pattern (.Next(), .At(), etc) |
||
| ForPostings(ctx context.Context, matchers []*labels.Matcher, fn func(index.Postings) error) error | ||
| } | ||
|
|
||
| // NewCachedPostingsReader uses the cache defined by `index_read_cache` to store and read Postings. | ||
| // | ||
| // The cache key is stored/read as `matchers:reader_checksum`. | ||
| // | ||
| // The cache value is stored as: `[n, refs...]`, where n is how many series references this entry has, and refs is | ||
| // a sequence of series references encoded as the diff between the current series and the previous one. | ||
| // | ||
| // Example: if the postings for stream "app=kubernetes,env=production" is `[1,7,30,50]` and its reader has `checksum=12345`: | ||
| // - The cache key for the entry will be: `app=kubernetes,env=production:12345` | ||
| // - The cache value for the entry will be: [4, 1, 6, 23, 20]. | ||
| func NewCachedPostingsReader(reader IndexReader, logger log.Logger, cacheClient cache.Cache) PostingsReader { | ||
| return &cachedPostingsReader{ | ||
| reader: reader, | ||
| cacheClient: cacheClient, | ||
| log: logger, | ||
| } | ||
| } | ||
|
|
||
| type cachedPostingsReader struct { | ||
| reader IndexReader | ||
|
|
||
| cacheClient cache.Cache | ||
|
|
||
| log log.Logger | ||
| } | ||
|
|
||
| func (c *cachedPostingsReader) ForPostings(ctx context.Context, matchers []*labels.Matcher, fn func(index.Postings) error) error { | ||
| checksum := c.reader.Checksum() | ||
| key := fmt.Sprintf("%s:%d", CanonicalLabelMatchersKey(matchers), checksum) | ||
| if postings, got := c.fetchPostings(ctx, key); got { | ||
| return fn(postings) | ||
| } | ||
|
|
||
| p, err := PostingsForMatchers(c.reader, nil, matchers...) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to evaluate postings for matchers: %w", err) | ||
| } | ||
|
|
||
| expandedPosts, err := index.ExpandPostings(p) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to expand postings: %w", err) | ||
| } | ||
|
|
||
| if err := c.storePostings(ctx, expandedPosts, key); err != nil { | ||
DylanGuedes marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| level.Error(c.log).Log("msg", "failed to cache postings", "err", err, "matchers", key) | ||
| } | ||
|
|
||
| // because `index.ExpandPostings` walks the iterator, we have to reset it current index by instantiating a new ListPostings. | ||
| return fn(index.NewListPostings(expandedPosts)) | ||
| } | ||
|
|
||
| func (c *cachedPostingsReader) storePostings(ctx context.Context, expandedPostings []storage.SeriesRef, canonicalMatchers string) error { | ||
| buf, err := diffVarintSnappyEncode(index.NewListPostings(expandedPostings), len(expandedPostings)) | ||
| if err != nil { | ||
| return fmt.Errorf("couldn't encode postings: %w", err) | ||
| } | ||
|
|
||
| return c.cacheClient.Store(ctx, []string{canonicalMatchers}, [][]byte{buf}) | ||
| } | ||
|
|
||
| func (c *cachedPostingsReader) fetchPostings(ctx context.Context, key string) (index.Postings, bool) { | ||
| found, bufs, _, err := c.cacheClient.Fetch(ctx, []string{key}) | ||
|
|
||
| if err != nil { | ||
| level.Error(c.log).Log("msg", "error on fetching postings", "err", err, "matchers", key) | ||
| return nil, false | ||
| } | ||
|
|
||
| if len(found) > 0 { | ||
| // we only use a single key so we only care about index=0. | ||
| p, err := decodeToPostings(bufs[0]) | ||
| if err != nil { | ||
| level.Error(c.log).Log("msg", "failed to fetch postings", "err", err) | ||
| return nil, false | ||
| } | ||
| return p, true | ||
| } | ||
|
|
||
| return nil, false | ||
| } | ||
|
|
||
| func decodeToPostings(b []byte) (index.Postings, error) { | ||
| p, err := diffVarintSnappyDecode(b) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("couldn't decode postings: %w", err) | ||
| } | ||
|
|
||
| return p, nil | ||
| } | ||
|
|
||
| // CanonicalLabelMatchersKey creates a canonical version of LabelMatchersKey | ||
| func CanonicalLabelMatchersKey(ms []*labels.Matcher) string { | ||
| sorted := make([]labels.Matcher, len(ms)) | ||
| for i := range ms { | ||
| sorted[i] = labels.Matcher{Type: ms[i].Type, Name: ms[i].Name, Value: ms[i].Value} | ||
| } | ||
| sort.Sort(sortedLabelMatchers(sorted)) | ||
|
There was a problem hiding this comment. nit: it's a little weird we're sorting something called There was a problem hiding this comment.
|
||
|
|
||
| const ( | ||
| typeLen = 2 | ||
| sepLen = 1 | ||
| ) | ||
| var size int | ||
| for _, m := range sorted { | ||
| size += len(m.Name) + len(m.Value) + typeLen + sepLen | ||
| } | ||
| sb := strings.Builder{} | ||
| sb.Grow(size) | ||
| for _, m := range sorted { | ||
| sb.WriteString(m.Name) | ||
| sb.WriteString(m.Type.String()) | ||
| sb.WriteString(m.Value) | ||
| sb.WriteByte(',') | ||
| } | ||
| return sb.String() | ||
| } | ||
|
|
||
| type sortedLabelMatchers []labels.Matcher | ||
|
|
||
| func (c sortedLabelMatchers) Less(i, j int) bool { | ||
| if c[i].Name != c[j].Name { | ||
| return c[i].Name < c[j].Name | ||
| } | ||
| if c[i].Type != c[j].Type { | ||
| return c[i].Type < c[j].Type | ||
| } | ||
| return c[i].Value < c[j].Value | ||
| } | ||
|
|
||
| func (c sortedLabelMatchers) Len() int { return len(c) } | ||
| func (c sortedLabelMatchers) Swap(i, j int) { c[i], c[j] = c[j], c[i] } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
this test passes with this flag set to
false. is there any way to this test to fail when caching is not happening? where are the objects coming from (local filesystem?), could we change the data at rest to prove the data we're getting is from cache?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I added an assertion that checks the cache added/misses/gets metrics results. they guarantee we're covering the cache behavior using the FIFO cache.