Streaming store-gateway Series()

[dimitarvdimitrov]

This is a POC and is not meant to be merged yet.

Overview

Instead of loading all series (label sets and chunks) in memory before
responding to a Series() RPC, we can batch them and load X at a time.
This gives more predictability on the memory utilization of the
store-gateway. The tradeoff is having to do one trip to the index cache
and the bucket for each batch, which will affect overall latency of requests.

How to use

This change disables batch series loading by default and adds two flags
to control this - whether it's enabled via
-blocks-storage.bucket-store.batched-series-loading=false and how many
series go into each batch via
-blocks-storage.bucket-store.batch-series-size=65536.

Limiting

Ideally we want ot put a limit on the number of bytes that we want to
load in each batch instead of the number of series. For now limiting
the number of series should still give us some resilience against "big"
requests, while still being vulnerable to a flurry of many requests.

Testing

I've changed all tests within pkg/storegateway to use this new loading
strategy. This should give confidence that it is producing correct
results. Further work should improve testing around resource utilization
(i.e. batches are indeed freed one after the other)
and should test both batched and non-batched strategies.

This commit has TODOs, which should be addressed before merging this.

Signed-off-by: Dimitar Dimitrov dimitar.dimitrov@grafana.com

[pracucci]

Thanks for working on this! I didn't review every single line of code for correctness, cause I want to focus on the overall design first.

I'm leaving here a feedback reported on Slack too. Let's keep talking on Slack cause discussing through comments here is harder and slower.

My main feedback is that with this design the query will underperform. Let's say we query 100 blocks. We preload 1 batch from 1 block at the time. Everything is serialized. I would like to keep having parallelization but be in control of it (e.g. max memory allocated).

[dimitarvdimitrov]

i gave this a try and memory does indeed flatten. In this graph I did the change to zone-a of the backend component of a read-write deployment mode cluster at around 19:20. Below is the average working set memory in bytes per pod in each zone.

[dimitarvdimitrov]

I went back to this to see how latency looked like during that period. We expected to see increased latency. But that wasn't the case. The p99 for the Series() rpc of zone-a during that time also lost its spikes in comparison with the other two zones.

For context, the batch size was set to 10K, and my queries were touching around 550K series over 12h and were sharded 32 ways.

This was a very ad-hoc experiment, and it's probably not as good as it seems. The plan for improvements is to

• add concurrency to this approach to further reduce latency and
• control concurrency based on the memory usage - we can have many small requests in parallel or few big ones
• move to global batching (currently batching is per block per request)

I'm now trying to make the code a bit easier to undergo the 3 changes above - defining more interfaces, breaking up the index and chunk readers

[dimitarvdimitrov]

593d0a4 is my proposal for the interfaces. This commit should help you imagine where the whole PR would go with concurrency, bytes-limits, elegant cleanups and theoretically offloading to disk. It shows the different components (via interfaces) and their dependencies (via factories).

The interfaces are sometimes intentionally brief and omit things like errors handling, contexts, loggers, metrics, stats. I believe these can be added later without disruption.

[pracucci]

You're doing an amazing work! I'm super excited about it. The design is way more clear than the initial one. I haven't reviewed every single bit (PR too big, we'll need to split into smaller PRs to get proper reviews, I will help you doing it) but I looked at the overall design which looks pretty solid to me.

[replay]

The CHANGELOG has just been cut to prepare for the next Mimir release. Please rebase main and eventually move the CHANGELOG entry added / updated in this PR to the top of the CHANGELOG document. Thanks!

[dimitarvdimitrov]

This means that we effectively block until we fetch all postings for all blocks before starting to load series and merging them. I think we can do better and fetch them asynchronously.

[dimitarvdimitrov]

agreed offline to do this change after merging this PR

[pracucci]

Good job, LGTM! I reviewed all this code already, so no big surprises to me. I noticed few last things I would suggest to take a look at.

I think this work deserve a CHANGELOG entry. Remember to add all PR numbers of previously merged PRs too.

Can you also double check we haven't left any nolint:unused?

[pracucci]

Note to reviewers: removed because was not used.

[pracucci]

The main difference of moving this error wrapping logic here is that now it's applied to all errors returned, while previously it was only applied to some of them. Problem is that I don't know what's the correct one, because I don't remember what is this used for. Do you? Can you check how the error is handled in the querier (where this function is called via gRPC)?

[dimitarvdimitrov]

i think it kind of depends. i had to do this because some of the errors which were previously sruficing here

mimir/pkg/storegateway/bucket.go

Lines 1023 to 1029 in ad526f8

	if err != nil {
	code := codes.Aborted
	if s, ok := status.FromError(errors.Cause(err)); ok {
	code = s.Code()
	}
	return nil, cleanup, status.Error(code, err.Error())
	}

are now surfacing here

mimir/pkg/storegateway/bucket.go

Lines 951 to 954 in 78ed13f

	if seriesSet.Err() != nil {
	err = errors.Wrap(seriesSet.Err(), "expand series set")
	return
	}

---

so with the collective changes we are removing this case where we return Unknown

mimir/pkg/storegateway/bucket.go

Lines 910 to 913 in ad526f8

	if set.Err() != nil {
	err = status.Error(codes.Unknown, errors.Wrap(set.Err(), "expand series set").Error())
	return
	}

i couldn't find a place which looks for the Unknown code.

---

to answer your question on the querier. In the querier we just ignore the errors unless it's EOF

mimir/pkg/querier/blocks_store_queryable.go

Lines 737 to 744 in 3592c25

	resp, err := stream.Recv()
	if errors.Is(err, io.EOF) {
	break
	}
	if err != nil {
	level.Warn(spanLog).Log("msg", "failed to receive series", "remote", c.RemoteAddress(), "err", err)
	return nil
	}

[pracucci]

Thanks for the details. I also can't see a reason why this change could break any error handling, given the querier is not checking it and io.EOF comes from the client, not the server.

[pracucci]

[nit] What if, for simplicity, we always call newChunkReaders() regardless req.SkipChunks?

[pracucci]

Has this changed intentionally? Logic is the same, but I've the feeling this has changed unintentionally.

[dimitarvdimitrov]

i was changed and then changed back.. and in the meantime it got a new form. i will revert it

[pracucci]

3.2.1.... LGTM!