Cloud Native Computing Foundation KubeCon + CloudNativeCon Europe 2022 [End User Sessions], 1 Jun 2022

Previous Meeting Next Meeting

⏯

youtube image

►

From YouTube: Distributing PromQL for Fast and Efficient Kubernetes Fleet Monitor... Moad Zardab & Filip Petkovski

Description

Don’t miss out! Join us at our upcoming hybrid event: KubeCon + CloudNativeCon North America 2022 from October 24-28 in Detroit (and online!). Learn more at https://kubecon.io. The conference features presentations from developers and end users of Kubernetes, Prometheus, Envoy, and all of the other CNCF-hosted projects.

Distributing PromQL for Fast and Efficient Kubernetes Fleet Monitoring - Moad Zardab, Red Hat & Filip Petkovski, Shopify

Both Thanos and Cortex have enabled the cloud native ecosystem to scale Prometheus storage with the use of blocks of data persisted across many clusters into single object storage. Whilst this unlocks cheap long term retention of metrics, it presents a significant challenge of being able to efficiently read and process large volumes of data. This talk outlines the Thanos community's efforts to improve read path performance through query pushdown and query sharding and how it compares with existing Cortex approaches. Thanos deployment's are composed of stores; components that expose a consistent Prometheus compliant read API for retrieving timeseries, and queriers; components that combine raw timeseries and evaluate PromQL expressions against them. Query pushdown gives the opportunity to pre-evaluate these expressions closer to the data, while query sharding breaks down a query into distinct, disassociated datesets that can be computed concurrently thanks to Kubernetes.

A

All right um thanks everyone for coming um today, the so the title of this talk is distribute impromptual expressions, um we'll focus in this session on some of the work that we've done uh with mod and with the thomas community on um implementing distributed query execution in the thanos project.

A

My name is philip. I am a production engineer at shopify and with me today is moat. Who is a senior software engineer at red hat?

A

The agenda that we have today is the following: one, we'll briefly cover why prom ql and the primcal engine is kind of hard to scale at the moment, specifically in prometheus, and then we'll cover two different approaches that we are working on in thanos, namely, query push down what we call query, push down and query sharding to speed up, query execution and then we'll take a look at sharding in practice and at the end, we'll cover kind of we'll give a brief outlook of where we hope thanos is going to develop in the future.

A

So if we take a look at the current state of prometheus, I mean we've heard so many times here at kubecon that it's become very well adopted in the cloud native monitoring space. Basically, anyone that's maybe used. The kubernetes cluster has seen a primitive system lying around somewhere, it's very effective for real-time monitoring and what's very good about prometheus. Is this promptql query, query language that allows us to write all kinds of expressive: queries against what we call time series data, and so the ideal use case for prometheus is really the single cluster monitoring.

A

Whenever we want to go beyond a single cluster, there are different challenges that are non-trivial to overcome, some of them being just the fact that scraping across boundary network boundaries can be. It comes with its own challenges. We have to rely on disks for retention. This can be hard to move between acs regions, clusters and so on and overall, the story about how to scale prometheus horizontally is not there yet and that's a conscious design decision. So there's nothing wrong with that um and as a result, there are all these different projects like cortex thanos.

A

Recently, mimir that come in and uh try to solve some of the scalability issues that prometheus has and if we look at thanos, we can essentially store data for as long as we want in s3 buckets.

A

In addition to that, we get things like write, replication, multi-tenancy. All of this kind of cloud native features that we take for granted today and now that we have all of the data all of this kind of long retention of data massive data, we can also execute prom ql against that that data set.

A

However, the prom ql engine that is in um primitives thanos in all of these projects, is still kind of the same engine that's used inside prometheus and that engine is single threaded single process and, as a result, has to kind of fit all of the time. Series for a query in memory has to execute the memory. uh The query in a single thread and the longer the query range and the higher the cardinality the harder this problem of executing the query becomes, and so we can visualize where this bottleneck happens.

A

Before we actually try to solve the problem, we can visualize where the problem happens. On a very simple model. Let's say we have a kubernetes cluster that has a prometheus instance inside that scraping just a set of targets, and when we outgrow that instance, we can deploy another instance that scrapes a different set of targets, and when we want to get the global view of the entire metric space, we can deploy something like a thanos querier.

A

That's going to pull data from these prometheus instances and execute broadcasters, and if we kind of outgrow this cluster, we can deploy, for example, another cluster. Let's say our production environment is now bigger, and now when we want to get a global view again of everything that's going on, we can layer a new thanos querier on top now, this tunnel square is going to kind of communicate to the intermediary ones and it can execute global queries across the entire set of metrics.

A

Now the problem is whenever we want to execute a global query, uh this specific tunnel square, that's at the root of the tree. It has to reach out to all of the underlying prometheus instances, has to fetch time series data, um that's relevant for the query in memory in a single process and execute the the query in that process.

A

What's also interesting to note here is that these intermediary queries in the middle they're not actually doing any they're, not very active in the query execution path. So what they do is uh mostly they act. They provide some basic service discovery mechanisms, so they allow the root career to discover this instances.

A

These leaves prometheus leaves at the bottom and they also kind of traffic data to the top, but other than that they're not actually executing from ql, even though they have promptly engines inside them, and so that's kind of the um the current state, the the default mode that we have in thanos, uh and this is why the the kind of the queries query execution can get slow with more data, and now that we know this, we can look at what we we've been working on.

A

The kind of the entire thousands community to to alleviate some of the problems, and so the first thing that we will look into is an approach that is referred to as query push down. So with query push down, we can avo. We can speed up queries by simply avoiding streaming a large amount of series in memory.

A

If we look at, if you want to look at how that works, let's say we have again a simple model, a query with two instances and let's say that we want to execute the following query: we want to calculate the maximum or a particular metric um yeah, I'm the maximum or particular metric, and so what we can do now is. We can take advantage of the computational power, the computational engine, that's built into these prometheus instances.

A

We can um execute the query locally first on the local data set, uh so these primiti are going to execute this specific prim field, query locally and they're going to propagate just the intermediary result upwards and then the querier is going to recalculate the query with this much smaller data set and by that we kind of distribute the query execution through the entire system.

A

So what you need to know kind of did was the feel they are with queer push down. The important thing is that it's already implemented in thanos and you can start using it today.

A

Obviously, the benefits are that it speeds squares up. It lowers from ql execution latency by processing data at rest. So, where it's at uh and it doesn't ship these megabytes or even gigabytes of data over the network, and if you want to play with it, you can enable it as a feature flag and start using it today.

A

The current limitations that are there, which we hope over time, are going to get become less and less, is that we can only apply it to a smaller set of promptql functions. So those are the mean max top k. Bottom can group they are kind of. We can refer to them as the item potent functions which we can apply over and over again at various levels in the query execution path.

A

The flip side of that is that the more common expressions such as sum rates uh calculating uh quantiles, for example, there's no good way to kind of defer execution to the leaves. Yet so we there's still more work to be done. There then there's there are kind of two other things that we have to keep in mind.

A

If we're going to execute queries, then we need components to have a query execution engine and currently in thanos it's only the tunnel sidecar that can execute queries, um so the other store components are simply going to ignore this instruction, so you can still enable push down.

A

Even if you have a mixed fleet of let's say, start side, cars receivers, store gateways.

B

A

Just that it's you can do that in a safe manner. So to say it's just that it's only the prometheus instances can take advantage of this. And finally, if you decide to do this, to enable this feature uh during the transition period, just keep in mind that query execution is not free and if these promethea are going to start executing queries, make sure you monitor for increased cpu usage increase memory usage so that you make sure they are adequately provisioned to to handle this additional load.

C

Cool, so, as philip mentioned, what we found with query pushdown was that it was super efficient for a very specific use case. What we wanted to do next was to explore a technique that would allow us to distribute queries more generally across promql.

C

We also wanted to make sure that our evaluation of queries was limited to query components. The nuance that philip mentioned around only being able to push down queries if there's a promql engine present was a critical issue.

C

We also wanted to extend some of the the existing query sharing in thanos so that we could distribute work even further. So those were our requirements going into query sharding. The tldr for career shouting is that query. Sharding is just splitting a query uh into distinct, parallelizable sub queries.

C

So before we talk about career shouting, it's probably worth looking at what a time series actually is, um so you can see for this metric http, request, duration seconds. We have this series axis and this time axis, each metric is composed of unique labeled dimensions. So in this case we have part and status with various values and then each unique time series has observations over time. These are what you would commonly refer to as samples each sample has a value associated with it and a timestamp.

C

So when we're talking about sharding, I mentioned that thanos already has a shouting implementation. There's this notion of horizontal sharding or time splitting, where the query can be split across time and then run in parallel.

C

This is because each time slice will only have the time series once and a query that spanning four days can be split into four one day: queries as an example.

C

The caveats with time splitting currently is that it only works for range queries. So if you have rules or alerts defined or if you're trying to use it for instant queries, that's not as trivially parallelizable as range queries are, and even for short intervals um on the graph that we were looking at before a time, slice can still have arbitrary cardinality. This is what the hub this is. The this is commonly known as the high cardinality problem in prometheus, which is the more time series you have the more expensive it is to execute a query.

C

So this is what time based or horizontal sharding looks like say you want to sum, buy a pod for a specific metric. You want to do a sum rate over a specific metric. um The way that time slicing works is if your original range is in this example t0 to t3. You would split that up into three distinct ranges and you would execute them in in parallel or in series, but that means that you've limited the number of samples that you might get for each time series.

C

But you haven't limited the number of time series you might get for each request.

C

So how do we manage this vertical cardinality? This uh scary problem that a lot of people in the prometheus ecosystem talk about.

C

um We wanted to work on extending and supplementing the existing sharding that we already have in thanos with this notion of vertical sharding, so the same way that horizontal sharding works across time vertical sharding works across the unique label dimension of each metric.

C

Each shot is disassociated because we're guaranteeing that each unique time series will always end up on the same logical shard. So you can process these queries in parallel, without worrying about duplication or correctness of your query, and we wanted critically for thanos to do this without too much planning ahead of time.

C

The reason why is because we don't know where time series are ahead of time with a query and we wanted to avoid additional fetch request or just checking where time series are before executing a particular query and in our implementation leaves are responsible for deterministically assigning time series to shards.

C

We also didn't want to make any changes to the thanos read path. So a commonly asked question is: if you have this issue of duplicate time series, why not just duplicate them when you write, um that's something that other projects do so, for example, cortex they dedupe uh on on right. We didn't want to change the right path. We wanted this improvement to be to be almost transparent to the user, where they don't actually need to change how they've set up thanos.

C

So the first uh expression that we wanted to implement for this vertical query. Sharding is this very common, grouped sum rate expression so assume you have the same expression where you're summing bipod and you want to see the rate of increase of a counter metric in this case http request duration seconds.

C

Instead of sharing across the time axis, we actually want to shout across the unique time series themselves. So what we're saying for each assign chart index is that the same time series will always appear there and then the query. I can process the query as it would normally with an uncharted query with a crucial final step, which is that we then concatenate the result uh of everything that's been executed.

C

Now with that being said, um you can actually use both in conjunction. So if you do a time range or a horizontal shouting pass first and then split each horizontal shard vertically uh across time, series you've now taken one one: query that that can potentially have arbitrary time range and arbitrary, cardinality and you've split it into manageable, parallelizable chunks that can be computed in parallel.

A

All right um so, as mod mentioned, we kind of worked on this approach for a while, and we have a reference implementation in a branch in thanos and we use that reference implementation to see whether the algorithm actually is going to make any improvements.

A

In order to do that, we generated a very simple benchmark. We generated a synthetic base data set essentially we're hundred thousand series, and by that we hope to simulate a setup where we have, let's say 100 clusters with a thousand calls each and each class in each cluster and then um for the we compared query execution in um uh started versus non-started setup. So for the sharded setup we use something.

A

We call a starting factor of three. That just means one query: one big query would get split into three smaller queries and would be executed a fan out in a parallel manner and in the default mode in the non-started implementation. We would just execute queries in a round-robin manner across five different queries and to keep everything simple. We we kind of did the benchmark on a single node, essentially on a single machine, and what we wanted to see is.

A

First, are we going to improve user experience, so our query is going to get executed faster and second, are we going to destabilize or stabilize the system more by sharding queries? So um what's the operator experience going to be essentially and one way we thought we could do- that is by measuring peak and average memory usage of queries over the course of this entire benchmark?

A

So if we jump right into the results or into the numbers, this particular graph shows the p90 latency of executing queries over 10 to 15 minutes without charting. So this is what you get by default in thanos. We see that at the beginning, the query latency is about nine seconds. At the end, we end up somewhere around 19 seconds and in the middle there's.

A

Also, this kind of big spike, where we, the the p90, jumps to about 55 seconds, to execute a query, so that just speaks to the fact that um just doing a more work in a single process over time can kind of destabilize a lot of things in the system and if we compare that to the um sharded setup. So when we actually split queries in smaller shards, when we start the benchmarks of the the p90 is about five seconds.

A

So already we have about the 40 decrease in query execution and when we end the benchmark, the latency is about 5.5 5.7 seconds that so what we can see from this graph, which might not be again easy to measure, maybe in numbers, is that we get a much more predictable and much smoother query response time, but simply partitioning work across multiple nodes or multiple processes. In this case, the single knowledge is multiple processes.

A

Looking at the memory usage again, this is the default mode, no sharding. What you get out of the box right now, um if you install thanos uh the peak memory for a query, is about 1.5 gigabytes to execute a single query, whereas the average is about 1.2 gigabytes and comparing that, with the charted setup we get again, we see a decrease to about 800 megabytes, so approximately 50 and the average is also about 700 megabytes and again, what's visible here is that the memory usage is much more consistent.

A

There aren't like big um jumps and drops, and the lines are much closer together, so there's much more predictability in the system, so, in summary, vertical sharding benefits very intuitive. If we split work across multiple processes, we expect lower peak memory across those processes, lower latency that all makes sense.

A

The crucial advantage, which we hope to achieve with virtual vertical sharding over specifically striding by kind of time range, is that we can apply this to instant queries. So things like alerting rules and recording rules can also be executed in a sharded manner, and just we still have a few things to hash out. So we don't support.

A

So we don't have an implementation for the full uh range of prompt qual expressions, so we need to have some sort of aggregations in the query, expressions in order to shard on those labels, uh but I'm, I think that over time, we'll just we'll just a bit need a bit more time to kind of hash out. The details there and one more thing to keep in mind- is that, as we start to split queries up, queries will be kind of smaller, but there's going to be more of them.

A

So there's going to be more requests per second, let's say throughout the system and we'll have to make sure that we keep an eye on that and make sure that all components are kind of properly provisioned and capable of handling that increased volume.

C

Cool, so I guess we've heard uh about practically how something like vertical and horizontal query: sharding works, we've seen the performance benefits and the benchmark that we've presented.

C

What I wanted to show is philip and myself both work in kind of fleet, monitoring and observability teams at our respective companies, and we wanted to give an impression of vaguely the scale of problem that this kind of solution, or that this kind of approach solves. So, let's uh suppose you have um a central observability cluster.

C

That's running thanos. Let's suppose that you have thousands of clusters that you have to monitor either platform, monitoring or user workload. Monitoring, and each of these clusters is producing some amount of series per day, and you want 30-day retention with thanos. You can really have arbitrary retention, but, let's just say it's 30-day for the sake of example, I'm going to simplify the query path just for the sake of demonstration, um but it stands to reason that if somebody is writing metrics to a central place that they probably want to query the metrics as well.

C

So let's say uh our users want to sum uh some metric by cluster.

C

We have uh our thanos queria, but this presents an interesting challenge with the caveats we've mentioned already, which is that we need the node to be big enough to be able to hold all of the time series that a particular query is fetching before it can evaluate prom qr, so we're holding 20 million active series over 30 days. Even if you uh time slice with thanos as it currently stands, that's still a lot of metrics that you might have to retrieve in a given query.

C

So um let's say you create a 100 gigabyte, thanos query, so you give it a 100 gig of memory, and you just hope that a query doesn't exceed that amount.

C

um You probably need a couple of replicas because you might you you want to serve concurrent traffic you're not going to deal with one query at a given time and you just kind of hope that this query executes and that you have enough memory to facilitate it. That's ignoring, even if you have concurrent queries that are already going on in the process and if you run out of memory, you run out of luck and your thanos query falls over.

C

So, let's look at the same example, but let's assume that we have vertical sharding, so we've taken this arbitrary length, query we've split it by time and then we've sharded it by index same expression, we're summing by cluster id, but now, instead of having single nodes having this, this uh condition that a single node has to be able to hold all of the series in memory we can shard uh the query across a fleet of kind of fungible thanos queries.

C

Query has a stateless anyway, and this query load is going to be attenuated across our fleet of thanos queries, instead of our kind of special large processes that we have to look after, and this just means uh as a as a platform engineer, if you're maintaining the system, you can sleep better at night and not worry about this arbitrary problem and or this arbitrary complexity problem and when it comes to scaling up and scaling down thanos queries because you're shouting queries.

C

It's simply the case of having a horizontal part, auto scaler, create more thanos queries or reduce them. When you don't have query load.

C

So, as philip mentioned we've, there is an implementation for this for for grouping uh aggregation queries. There is also a proposal in upstream thanos that discusses this and we're working on expanding from qr support. So now that we've done some rates uh by some value, we want to start exploring the rest of the prom qr standard library. um This was philip and myself's first foray into contributing to thanos. So I recommend that you all have a look at these. If this interests you and contribute to thanos, the community is super friendly.

C

We had a great time talking to people about these ideas and there's plenty of cool work to to take a look at for more context. This was quite a simplified view of the problem, but if you want to learn more about the prometheus tsdb in the exposition format or thanos itself, and how to do fleet monitoring in thanos, there are these great kubecon talks by members of the computer of the community of the prometheus and thanos community that I recommend you watch yeah and thanks thanks to thanks to the community and maintainers.

A

All right, so I think we'll have some time for questions and if anyone has any questions, I think you can just walk up to that mic and shoot.

B

All right, hello, hi, thank you for the talk, so it seems like the vertical sharding might be useful, useful for prometheus itself. Also, like, I think the majority of the time is spent in getting the data from the disk right, so it seems usable for a single prometheus as well. What are your thoughts on that.

C

Yeah, that's true, I guess the the thing that prometheus tries to do- and this is I assume intentional by design- is- is effectively restrict the problem. Space prometheus is very good at doing a very specific thing and introducing complexities to promql engine itself, how it processes, queries are probably out of the scope of of prometheus itself, but there is there's been talks in various meetups this week uh about what a a future promptgoal engine would look like.

C

That's built for this kind of cloud native use case instead of the single prometheus use case, but it's not an easy problem. It's quite complex.

A

I think there are also some um kind of proposals, informative itself to add sharding at the tsdb level so that the block can be. You can retrieve um shards of the block, um we're again we're we're following that work, making sure that at least what we're doing thanos is backwards compatible or compatible with prometheus, um but changing such a stable project like a 10 year old cncf graduated project takes a bit more time than adding something kind of around it or not.

D

Hey thanks for the talk again so you're proposing sharding on the query level, but if our data comes from the store than our store, can we also shorten the store level.

A

um So that's what we do we actually um so the query requests a chart from the store and the store only retrieves, a subset of data, so sharding in in this particular implementation.

A

Sharding is going to be implemented end to end from when you execute the query to the store when the store actually retrieves data sends it back to the credit. Does that answer the question or.

D

Yeah more or less, but we have one story, instance or multiple.

A

You can still maintain your existing setup um different, uh so queries that handle different charts will still talk to the same. Even if you have a single store instance, they will talk to this. One store instance: they'll just request one shard of the data, so the existing setup, for you will not change essentially.

D

E

Hi, um did you by any chance, have a look at the s3 request weight when you do vertical sharding so does that increase or because.

A

E

Our use case it's we are limited by our s3 request rate. We basically have unlimited compute and memory.

A

uh I think that's a very good point, um so there there we haven't looked um specifically at this particular metric, but in thanos there are various way to cache data from s3, so cache both index data and chunk data, and so we're hoping that, even if there is an increase, we can reduce it by adding some intermediary caching so that we don't have to fetch data twice.

A

um So to answer your question, the tl dr, we haven't specifically looked at it, but we hope that there's or we think that there's a way to actually address this problem so that we don't increase, requests towards s3.

E

C

Cool we'll be outside after the talk. If anybody wants to grab us and ask us a question, but I'd recommend you have a look at the proposal in the upstream and engage with the community. If you have any concerns or ideas.

A

All right, thank you.