Rethinking StoreAPI

We will be talking about Thanos’s StoreAPI, its shortcomings and how to solve them. The biggest two are:

It is still inherently associated with the slicelabels build in Prometheus. It means that labels.Labels (labels is a Prometheus package used in all sorts of internal interfaces) is a slice instead of something more compact/efficient like stringlabels. If you specify a collection a labels as one “big” string, it puts less pressure on the garbage collector. In the past (and still now), labels are stored with the slicelabels tag as []struct { Name, Value string } both on the RPC level and the storage level. This allowed us to convert from the RPC level to the storage level with no conversion but, in fact, it takes less CPU resources to copy memory from one format to another instead of putting a lot more pressure on the garbage collector.
To deduplicate streams of labels coming from multiple servers, we kind of need to compare a lot of labels with labels.Compare. Using a loser’s tree helps but labels.Compare is still inherently a O(series * labels per series) task because in the worst case you need to compare each Name and Value with all others.

So, in conclusion, we need very efficient way of transfering labels and then merging streams of labels. Ideally we need to implement this in such a way so that we wouldn’t have to return to this topic ever again (: or at least for many years!

For the first deficiency, we can come up with a much more efficient transfer format for labels. I think we can exploit the following characteristic: in most of Prometheus client libraries, each metric is a descendant of a “metric family”. Up front, when defining a new metric, one needs to define the fixed label names that will be added to each metric. So, when a user queries something, it is very, very likely that it will just be a stream with the same set of label names just with varying label values.

Hence, we can encode the set of label names as a “schema”. Thus, a batch just becomes a collection of definition of schemas, a set of symbols, and then a list of symbol references that encode the label values, and finally - the list of chunks.

Right now merging streams is also very slow because we first decompress all batches and re-compress them again. We must avoid doing that as much as possible. For that, we shall store the first & last labels (or hashes in case of projections) so that it would be possible to avoid decompressing batches altogether.

However, sometimes, some series might not exist in one replica that exist in another. This would immediately make two otherwise identical batches “unmergeable” (they are mergeable but you’d need to decode & re-encode). This is what exactly we want to avoid. So, to avoid this problem, I suggest lifting the requirement for each batch to be of the same size as it is right now. We could cut the batch every, for instance, hash(lset) % 64. This, on average, should give us a batch that is 64 in size. To avoid batches that are too small, we can bound the minimum batch size 32. Do not cap the top batch size because otherwise we would risk interfering with the natural order. There’s a tradeoff here: bigger batch sizes would make the payload smaller but less resilient against churn. Smaller batch sizes mean more encoding & decoding, bigger payload but more resilient against churn. By using the hash in calculating the batch size, even if there is a gap in one batch coming from multiple replicas, after one iteration the batches are back on the same order and we can avoid decompression & compression.

So, when merging we can see three multiple cases:

Where external labels after replica labels removal actually identify a completely different set of series. So, we can just take a look at first & last labels (hashes) to merge batches.
If there is some overlap then we use binary search to quickly find where it would make sense to “cut” one batch and add the series from the other batch “on top”.
Where, after replica labels removal, all labels in a group of batches are the same. In this case, we can just check the chunks to see if they are identical or not. If they are not identical then we can chain the chunks (samples).

Let’s talk about the streamability of merging results.

The Prometheus Exec() interface mandates that the whole response must be available at once. In other words, it’s not possible to stream it to the JSON encoder that gets called at the end of /api/v1/query. So, labels must always be “materialized”. However, maybe we could avoid materializing all of the []bytes immediately, on the start of query execution? Seems like that could only be possible if we separated chunks from labels in the gRPC messages. Initially, the Thanos PromQL engine calls Select() and gathers all labels. Only then when stepping through, it calls Iterator() in batches.

The “columnar” RPC format enables this fetching in two steps. We can first send the labels and then chunks. If the series are keyed by some index then the chunks are automatically keyed by them too. Only complication is that the current Prometheus Select() interface gives you a lazy iterator over a set of series and both chunks (samples) + labels are returned at once. The comment above the interface says that the chunks/labels must be alive even when Next() is called again so memory is not reused immediately.

In fact, Thanos already stores outgoing responses because due to the addition of external labels, the stream might become unsorted. That’s why, unfortunately, we cannot just send out everything. It’s another problem but it also enables us to send labels & chunks in two steps. However, will we really win anything if everything is stored in memory either way before sending it off? At that point, we just need to transfer that data as fast as possible. So, we need some way to “plan” - fetch matching series IDs. Then, get labels and finally - chunks.

We can kind of say that this reimplements Apache Arrow. And it is true in a way but at the moment I think we really do not want to reach for yet another RPC framework. It’ really a lot of work to implement a yet another framework and I believe that to apply Thanos-specific optimizations, we will still end up implementing something similar, even in Apache Arrow or somewhere. Also, Cap N' Proto integration was started on Thanos Receive but apart from that no one really stepped up to introduce it in more places. We really do not have much resources so any solution implemented must ideally not put more work on us. This is in part the purpose of this blog post - to write out my ideas in public, get some feedback, and think through this really well.

External Labels issues

We talked a bit previously about what problems external labels are causing but I think we can rework them entirely.

This is a thing that has also been bugging me for years and I think it is also finally time to address it. If you didn’t know, external labels in Prometheus is a functionality in which Prometheus dynamically adds those external labels whenever Prometheus exposes time series through some way e.g. through the /api/v1/read interface or just query execution. It is not a bad idea but the bad part about it is that it conflates multiple features under one umbrella:

If the user wants to forcibly add some labels to each series then they should use relabeling rules and always override some label with some value
If the use case is to route requests based on the matchers and external labels, some labels from series could just be “promoted” to be “routing” labels. In other words, periodically fetch values of a given label through InfoAPI and then they can be used for routing requests just like external labels are used now. There’s a bonus: such dynamic routing labels are really dynamic and not set in stone like with external labels.
If the use case is to group a stream of series with a common identifier, we could add a parameter like “–workload.id” that in each case should be unique throughout all metrics producers.
If the use case is how to vertically compact streams of series, typically the only way to do this is through a Receive which is tenant based. Vertical compaction using the penalty algorithm also exists but it is pretty much unusable because one needs to decode each Series to apply the algorithm. On the scale of millions of series, it is just not feasible. So, I suggest dropping vertical compaction using penalty algorithm. Vertical compaction should only happen on streams that have the same tenant.

Now, it is great that such a simple functionality like external labels implements a lot of these things but most importantly right now it forbids implementing streaming of results. In reality, whenever we are fetching data from a node, we really should not have to do anything on top to it. The current labels sorting rules look like this (taken from Prometheus proper):

// Compare compares the two label sets.
// The result will be 0 if a==b, <0 if a < b, and >0 if a > b.
func Compare(a, b Labels) int {
	l := min(len(b), len(a))

	for i := range l {
		if a[i].Name != b[i].Name {
			if a[i].Name < b[i].Name {
				return -1
			}
			return 1
		}
		if a[i].Value != b[i].Value {
			if a[i].Value < b[i].Value {
				return -1
			}
			return 1
		}
	}
	// If all labels so far were in common, the set with fewer labels comes first.
	return len(a) - len(b)
}

But, external labels right now force us to buffer everything. Imagine that external labels are set to {a=1} and we have this stream of series:

{a=2, b=2}
{b=1, c=1}

After applying external labels, the stream becomes:

{a=1, b=2}
{a=1, b=1, c=1}

Adding (or applying any kind of modification to a stream really) external labels to it makes the stream unsorted and, hence, we have to buffer everything

resort before sending it off. The current PromQL engine does not depend on the exact order of series because it just buffers everything BUT it is important for deduplication. Before an iterator is given for a certain series, the dedup algorithm must know that at some later point in time another copy of that series will NOT come.

By distinguish “routing” labels separately we will be able to store the metrics as is and not buffer at all because we won’t have to worry about the orderedness changing at runtime. Mind you also that replica labels in Thanos can be chosen dynamically so it’s not like we can apply them on write. But with this new model we won’t even have replica labels anymore - if the labels are the same then they are the same series & they are deduped. If not - they are separate series.

Individual (most used) columns (label values) in columnar formats should still remain sorted because that’s what allows to quickly find matching values when any matcher comes in using binary search. That’s a much more important characteristic because with object storage reads are most certainly not “free”.

External Labels issues#

External Labels issues