Scaling Valkey the Right Way: Kubernetes at XL Scale - Sarthak Aggarwal & Madelyn Olson, AWS

In this talk, Madeline Olson and Sarthak discuss scaling large clusters, specifically focusing on Valkyrie, an open source fork of Redis. They highlight the importance of scalability by referencing cases such as the Pokémon Go launch, which required robust infrastructure due to unexpected demand. The speakers elaborate on Valkyrie's architecture, including its ability to scale horizontally through sharding and its unique communication model via a cluster bus. They also share best practices for deploying large clusters, including proper resource management and the use of connection pooling. Additionally, they discuss their experiences with testing Valkyrie's limits and improvements made to enhance its performance under load.

Hello, hello. Welcome everyone. My name is Madeline Olson. We're going to be talking about Valkyrie today. I like to say that me and Sarthak here are foreigners in a strange lands. Both, you know, we're Americans over here in Europe, as well as we come from Amazon, which is mostly a VM based world. So, we're all we look over here at Kubernetes and we're like, "Wow, this technology

is amazing. I wish we could use more of it." But, we're hoping that we can take some of our learnings that we have seen sort of helping build extremely large clusters for caching from the Amazon ElastiCache world, which is the in-memory managed caching service that Amazon provides, and I'll provide some, you know, hopefully useful guidance for this. So, as I said, my name is Madeline. I am

a principal engineer at Amazon ElastiCache. I'm also one of the open source maintainers of Valkyrie. Don't worry if you don't know what that is. We'll talk about a little bit more in a second. And I'm joined with Sarthak, who is another engineer at Amazon ElastiCache, and also a contributor to the open source Valkyrie project. So, the theme of our talk today is really about, you know, scaling

large clusters. And I want to start with a story first about why scaling large clusters is important. Do people here remember the Pokémon Go craze that happened back in 2016? People out going wandering around late at night trying to catch Pokémon. And I really like this story because they highlight the fact that when Niantic they made some projections about how much usage and capacity they were going

to need. So, they kind of, you know, did some math and figured out that orange line at the bottom, that was their baseline. That's how much capacity they thought they would need. And then they scaled four five x that. They're like, "That's the worst case." And the green line is what happened. Which is, you know, it peaked at almost 50 times what they expected. But, the beauty

is they talked a lot about how they were built on top of Kubernetes and cloud native technologies and they were able to basically scale up the whole way. And so the thing I want to highlight is um we're going to talk about 2,000 node clusters for Valkey a lot and uh you shouldn't be running those every day in production. They are finicky. There's basically no vertical scalability

of those clusters. Um but you might have to and so the goal of today is sort of how do you be prepared so that if you do end up in this situation where you need to run these large clusters, uh you know the best practices, you know how to run them effectively and uh they don't explode in your face. Cool. Uh so with that I'm going to

talk a little bit about Valkey. So um how many here how many people here know what Redis is? How many people here knew what Valkey was before walking in? Oh, that actually shocks me. That's so many people. I'm so happy. Um like I did this we did a similar talk last year and like maybe a quarter of the room raised their raised their hands. So I'm very

excited about that. Um so I can be a little shorter and more abbreviated here. Uh so Valkey not a CNCF project but a Linux Foundation project. Uh so we're still backed uh we Valkey was created out of the fork of Redis open source 7.2. So all of the technologies that you use with Redis still work with Valkey. They're basically all API compatible. Uh but it is built

and maintained by the contributors that were formally associated with the Redis open source So when we see folks deploying Valkey in production, there's basically two predominant modes. There's what we call Valkey standalone. Valkey standalone is a single instance deployment. Um most people do it as a stateless set. Uh there are great helm charts. Valkey provides a helm chart that you can use. If you just need a

simple cache, a simple service, just launch that. It will work well. Um if the node dies, you know, it'll get replaced and the cache will come up empty but that's fine for most users. Uh but that's not what we're going to be talking about today. We're going to be talking about Valkey cluster. So Valkey cluster is the horizontal scalable mode of Valky. So, let's look a little

bit about how that actually looks. So, Valky as I said was a fork of Redis open source. So, uh Redis predates a lot of the cloud native technologies by a good bit. So, it has a little bit of a weird deployment model. It doesn't exactly fit into cloud native, but it you know, kind of fits. Um so, Valky is distributing work across shards. So, shards is a

horizontal slice of all of the data. And then within each shard there is a primary that serves the data and replicas which uh receive the data from the primaries and can serve read requests. So, uh when you want to basically add more capacity, add more throughput, you add more shards. Um and that allows you to scale both write capacity and read capacity and everything scales horizontally. And

so, this is all powered by what we call the cluster bus. So, a much more native cloud cloud native solution to this might be storing all of this topology information something like that. CD. Uh what Valky does instead is every shard owns their states and they constantly are communicating with other nodes in the on the shards. Uh this allows you to make it very easy basically if

you have a cluster of nodes, uh you can basically start with one primary. You can add a second node to it, make it a replica, and basically they will talk to each other, they meet each other, and then they join a single Once you add a a third node to this cluster, it will start gossiping with the other nodes. And then they'll automatically kind of join this

full mesh. And that continues on. So, when you add, you know, a fourth node, it will automatically learn about the other nodes in the cluster. So, what intuition we'll see here is this is a full mesh. Every node talks to every other node. Uh so, at some point you do start seeing scaling limitations on this cluster clustering link. So, we kind of see around when you get

to around 1,000 nodes you'll see about 1 to 2% CPU. Um and so, that that's kind of why that that's that's the vertical limit of sort of where clusters go. We're working on making this a little bit better, but for now, that's how it works. Valky has automatic detection, so if a node fails, basically all these nodes are constantly talking to each other. So, if a node

dies, it will detect that within a couple of seconds. There's a built-in system called a cluster node timeout. This is basically the detection time for how long other nodes will wait for this node to like respond to pings. So, the default is 15 seconds. That constant works pretty well for clusters up to about 2,000 nodes. You can lower this a lot, you can bring it down to

1 second, 500 milliseconds for small clusters. But, just know that won't scale very well, right? When you scale up the cluster, it will start taking more time. So, when a node's detected, a replica is able to basically promote itself to become a primary automatically. So, this will happen without anything going on. You don't need an operator to figure this out. It will happen just automatically as long

as all the nodes are connected in a full mesh inside the But, you will need something to go eventually replace the node. Nodes that rejoin the cluster will automatically make themselves replicas. There's some built-in technology inside Valky so that they will automatically go and find primaries without replicas and go attach them. So, without One thing we'll talk about a little bit later is we are working on

an operator cuz some of this stuff isn't fully managed. Some stuff like automatic scaling does need an operator intervention, but we'll talk about a little bit today. And if you are sticking around, Joe will also talk about in the next talk with Valky. Another thing that's very important to remember when kind of making sure Valky clusters scale is zones. So, basically all cloud cloud native systems, AWS,

GCP, Digital Ocean, have availability zones. You want to make sure both that you have good distribution of primaries across zones. So, you need a quorum of primaries to actually do a failover. So, in this example, we have a nice splits between all three zones of the primaries. So, if one of them fails, a failover will be able to happen. It's also actually important to make sure replicas

are in different zones from their primary, so you don't lose both all all the copies of your data in one situation. So, in this case, this would be a bad example. The replica for one of the shards is in the same AZ as the uh a primary. So, we want to proactively move that. So, Valkey doesn't have any built-in technology, but the operator that we're working on

will help kind of solve that. The next thing I want to talk about is um some people like to put something like an Envoy proxy in front of their Valkey Redis clusters so that they don't have to worry too much about topology. And this is a little bit of an antiquated concept because a lot of people are used to the olden days when the clients that connected

to Redis or Valkey were quite bad. And a lot of this is because uh the protocol that Valkey uses is a is a little bit bespoke. And so, I'll talk a little bit about a little bit and I'll talk about some of the benefits it has over basically talking through an Envoy proxy. So, the way that clients uh connect to Valkey is they connect directly to one

node in the cluster. So, typically in Kubernetes, you'll maybe have a headless service that just always has one configured node. You'll go and talk to this node and you'll get the cluster topology back. So, this is the topology state for a single shard in the cluster. So, let's look at some of this information here. Uh the first we have our slots. I haven't talked about slots yet,

but that's sort of the actual atomic unit of sharding within Valkey. So, all keys are mapped to a specific slot, and then slots are assigned to This is actually another limitation you'll eventually run into with large clusters. There are a fixed number of 16,000 slots. And so, when you start getting to around 2,000 or clusters, you start hitting the point where you uh when you actually try

to like move slots around, there's too much uh variance in heat between various slots uh to actually kind of have a uniform cluster, and you'll start seeing spikiness. So, uh as I said, shards basically have a slot range assigned to it, and then all nodes are given a random identifier. Uh this identifier is assigned randomly by the cluster. You can assign it yourself if you want it

to be like related to pods, but I haven't seen too many people use that in production. Uh the next set of information that's relevant is all of the endpoint information. So, we actually have two pieces of endpoint information. We have both the IP and the port, and then we also have host names associated with it. Um sometimes host names are attached as like the pod name that

you might see in a cluster, but we also see um a lot of people want to directly connect to IP, so they don't have to worry about DNS resolution. They don't want CoreDNS to have any latency or anything. So, we generally recommend directly connecting to the IP address when possible. Uh the host name is typically only used for TLS verification if your application has that enabled. And

that last piece of information that's very helpful is the availability zone. Most modern Valky clients are actually smart enough to be aware uh about doing basically zonal local reads, which is important for caching, um less important for you know, other high-performance things if you're using it as like a session store or a streaming system. But, if you uh can avoid doing cross-AZ charges for both latency and

cost, it's uh preferable to do a zonal read. Um just kind of summarize what I was talking about like this specific protocol gives a lot of benefits over HTTP. So, if you're using something like Envoy, which was sort of built more at that L7 layer, uh you won't be able to get a lot of these great internal benefits that we see by directly using the REST protocol

and directly talking to the engine. Um some of the best practices that we've seen at AWS is make sure you're using connection pooling. That's one of the biggest problems we see on Kubernetes clusters is you have, you know, thousands of pods, they're all opening connections to clusters, and that basically ends up taking down the individual nodes. Um, this is actually one good reason to use something like

Envoy to actually do that connection pooling at the Envoy layer. So that you're not doing thousands of thousands of connections to all the individual cache nodes. Um, also, it's very important to limit like connection retries. We see a lot of folks like when a node dies, they'll all try to redo the connection establishment. Um, the topology command I just showed was is actually quite expensive to execute.

We actually spent a bunch of time optimizing inside the engine. So we actually cache the response um, in a like to make sure that it's served quickly and reliably. Um, just to make sure that, you know, if you do end up doing a lot of requests, you're not taking down the server. Uh, Valkyrie and Redis are still notoriously the main thread is single-threaded. Um, and at the

end of the day, it doesn't matter for most cases, but it does impact workloads when you're doing a lot of connections all at once. We also um, built a a a new client called Valkyrie Glide, which implements something called where we try to limit the number of topology refreshes. So instead of every individual uh, connection having its own topology refresh mechanism, we try to pull all those

together. So a lot of people used some of the old uh, clustering clients in the past and saw a lot of issues. So if you did have those, we recommend moving to something like Glide, which is um, the client that we have been working on that is built with a lot of these more modern practices. And I think I already talked about Envoy, so we'll keep moving

to the next thing I want to talk about, which is resource management. Uh, one of the other things that we at AWS specifically want to improve was uh, observability within large clusters. So originally, like most people are aware that, you know, if you have a single pod and it has very high CPU, high net high network, you're like, this is a problematic node, but that's kind of

all the observability you could do at the time. So, one of the things we try to do is actually build more slot specific information to do those deep dives. Uh so, we built a command called uh cluster slot stats, which basically allows you to introspect in the Valkey clusters to figure out basically what's uh actually the bottleneck. So, in most caching workloads, most high performance in-memory workloads,

you basically have one of four bottlenecks, which are network bytes in, network bytes out, CPU or memory. When we actually go and look at like clusters we saw in Amazon, the vast majority were bound on memory. So, key count is sort of the best approximation of that. And so, when you have a node that's running very hot and you want to basically try to move that heat

somewhere else, do some re-sharding, um you basically want to go find the slot that's the second hottest and then move that data off. It's a little bit counterintuitive. A lot of people think it should be the hottest slot, but that's actually the most amount of work if you take the hottest slot and move it somewhere else. So, you should take you basically want to get everything that's

not the hottest slot off the node first. So, this is something that was available in Valkey 8 after the fork. Um so, this is some of the technology we've been trying to build into Valkey since the the license change happened. Uh some other useful situations we've seen with improving resource management, uh we've seen a lot of folks uh that run on Kubernetes clusters try to co-locate caches

with CP- CPU-intensive workloads. So, caching tends to be very memory-dominant. And you know, things like web services, other types of batch jobs tend to be very CPU-dominant. So, if you can use Kubernetes to sort of co-locate those together, you can get great utilization on both sides. Uh if you are doing that, it's important to make sure you're not saying too much CPU limitation on the caches because

you will then start seeing latency spikes if you have a bunch of concurrent commands. Uh I already touched on the fact that um if you do need to do migrations, try to focus on migrating the least uh the not don't move the hottest information in the cluster. Focus on moving basically everything else off of that shard. Um, we actually also built some finally built some technology. So,

in the olden days in Redis and Valky, when you had to do migrations, it was very painful. You had to basically run a CLI tool that went and like delete keys from one shard, put them in the other. And now it acts much more like replication where data can just transparently move between shards in the cluster. um, so basically what I've been talking about sort of is

some of the things we've learned um, from our customers that we've seen inside Amazon. And I'm going to hand it over to Sarthak, who's going to talk about um, some of the proactive steps we've been taking in the last year or so to also continue help making the system scale better and some of the improvements we made we made along the way. Sorry. >> Hey, hi everyone.

Um, thanks Marlin for introducing this and uh, yeah. So, we talked a lot about like scaling and how can we scale large clusters and we have seen the use cases like, you know, there are cases with ride hailing services or streaming services which sometimes require you to scale. Sometimes you don't need to, but there are situations that can occur. So, the question had to be asked, how

far can we scale Valky? And to prove this over the past year, we like we spent a lot of months trying to test a bunch of cases. One of the initial cases was that we started killing uh, one primary node in a 2,000 node cluster and we slowly ramped it up to one primary, 100 primaries, 250, 330, and up to 499 nodes. Why 499? Because in a

2,000 node cluster, even if you have one primary, one replica, you probably need for five you you need 501 nodes to maintain the quorum so that the cluster can self heal. And we wanted to check like how far can we push Valky so that the we also distributed our nodes in like two AZ, three AZ setups where we have uniformly distributed the replicas and primaries across AZs

and we just failed the AZ. And we wanted to see if we were able to recover from that. Uh then we also wanted to check what is the absolute limit we can push in terms of workload. And we started slow. We started with 25 million RPS, 100 million, and then slowly ramped it up to 1 billion. I'll talk more about this in a bit. Uh we tested

with TLS, non-TLS clusters. Uh we also tested with a lot of client libraries because that's how users interact. They use a lot of client libraries to uh talk to Valkey and we wanted to make sure that the open source libraries like ValkeyPy, ValkeyGlide, ValkeyGo, these are compatible and they're able to support these large clusters. And then again, like the one primary, one replica setup versus the one

primary, two replica setups. So, you know, just playing around with 2,000 node clusters. Uh so, yeah, before I go into what went wrong, I'll just talk about what worked. one of the first things that worked was um I think in a couple of versions before we did some optimizations with IO threads, which with which we were able to prove that a single node was able to scale

up to 1 million RPS. But now we wanted to check, can we scale that experiment with 1,000 primaries? And we kind of were. So, like to put it pretty simply, in a 1,000 in a 2,000 node cluster where there are 1,000 primaries and we are sending a bunch of write workload, we are able to reach up to 1 billion RPS. So, that was pretty cool. Uh the

second thing, we kind of expected the number of nodes that as the number of nodes increase, these nodes communicate with each other. They communicate each with each other through cluster gossip. So, there is a lot of com- there is some compute that is going uh gossiping and then there is some network bandwidth also that is getting utilized in that. And And kind of hoped that as we

scale the nodes, it will kind of scale linearly. And that's what we saw. So, that was a nice observation that as the nodes are increasing, we uh the resources kind of just scale Now, now we'll run it Now, we'll talk more about the weird issues that what actually broke when we were doing these tests. Um so, yeah. This is the first one. Uh so, this was a

very peculiar case where when we were killing a lot of primary nodes, say 415 nodes for 455 nodes, a lot of the other nodes which were still alive were trying to still reconnect with the failed nodes again and again. And they were trying to reconnect in a very small window, which was 100 ms. And due to which uh there was a lot of compute that was just

getting spent in the opening the connections and then trying to free that connection. So, even if we knew that the connection is not going to get successful, the lot of compute was just going into that. So, we basically implemented a smart connection throttling mechanism uh where we want to ensure that within within a cluster node timeout, which Merlin just mentioned like it's 15 seconds by default, we

ensure that all the nodes get a fair chance to get reconnected, but at the same time we don't want our Valkyrie nodes to get overwhelmed just by doing the reconnections. So, this was Now, the second one. So, the two spikes that you see, these are just after the moment that where we have killed 499 primary nodes. And this was kind of interesting because there is a huge

spike and then it automatically comes down. But we were curious like why is this happening? And we kind of found out that uh this is happening because when the nodes fail, they try to gossip with each other and then there is a lot of processing that is going behind the scenes to include those failure reports, to expire the failure reports after they are recalled. And we all

kind of wanted to optimize that because even for a brief moment of time, it is still a lot of compute. And we did that. So, earlier it was a list that was maintaining uh the failure reports and it was like an O and operation and O and operations can be expensive with 2,000 node. So, we replaced it with a radix tree uh where we basically have each

of the nodes that are based out of timestamp so that we were able to group some of the nodes together and yeah and there was a good improvement. Uh we are practically so from 100% the we were around 28 to 30% uh of compute just after failing. Now, yeah, this is the third problem. Uh so, this was also this is this was I think I think the

main problem that we saw while we were scaling large clusters with the shard vote problem. So, what happens is if you have a five shard system within Valkyrie, uh each of these nodes are connected with each other through a cluster bus. So, it's a proper Now, what happens if two of the shards fail? So, as in this example, shard three and shard one have uh the primary

of those shards are failed. Now, we have our failover semantics where the replica is able to take over the primary in the event of uh failure so that we are we stay available. what could happen So, when the first shard failed, uh the replica would want to ask the vote from the alive primaries that okay, now I want to be the primary. So, you know, please vote

for me. And at the same time, the replica of shard three will also ask for the votes. Now, when these two ask for votes together at the same time, it is possible that each the alive shards or the alive primaries will have a split vote where one of the shards, say shard two, will vote for just shard one and shard five and four will vote for shard

three. And this happens because within Valkyrie, we had the system where at a particular epoch, which is like a unit of time, we'll just have one vote that which the primary will be able to just vote once. Which makes sense because we don't want it to vote for multiple, so that we don't want to run into that loop where there are multiple replicas trying to ask for

ownership of that shard, and it keeps on doing that. So, this was a problem, and the vote was split, so nobody got quorum. So, how did we solve this? So, Bin Bin from who is a Valky maintainer, he basically came up with the solution ordered the number ordered the request and how they will come, the failover request. And we ordered it lexicographically. So, as you can see,

shard one was the first in line, and the replica of shard one was able to request the request for votes at T1 independently. Uh, it got all the votes, and then it was promoted to the new primary. Similarly, the shard the replica of the other shard that failed got all the votes and was able to promote to the primary. So, this way we maintained there was we

maintained that there is a small jitter between the two requests, so that there is no overlap and there is no vote split. So, this was also this this was actually really instrumental in uh helping Valky scale. Uh, and yeah, what was the outcome of that? So, once uh we did these improvements, we wanted to test how far how quickly the cluster can recover in the worst case.

So, starting from one primary failover to up to uh 499 primary fail uh failure, we we are able to see that in less than a minute we are able to recover, and the cluster is able to self-heal itself in the event of such failures. Uh, and yeah, this is the last improvement uh that we made in order to scale Valky. So, uh pub sub So, a lot

of users want to use uh Valky as a pub sub mechanism, and it's kind of convenient to use it as well, because it leverages the same cluster bus connection. So, you could have somebody who is just publishing a bunch of data, and you got to multiple subscribers listening to it, and it can use the cluster bus network. But, the important part in this was that if you're

sending a 100-byte pub/sub message, we were adding a lot of overhead to it. And this overhead was as huge as 2 KB. And this uh 2 KB was basically all the slot information that we basically were sharing. But, it is not really relevant uh pub/sub messages because they don't really care about slots. So, what we did was we introduced a lightweight message header, which basically reduced the

2K message header to 16 bytes. Just kept the information that we really need. And thus, we saved like a lot of I mean, these figures are up to scale, so you can imagine how much we saved in terms of each cluster bus message. So, yeah. Just to conclude on this talk, Valkey clusters are really scalable up to 2,000 nodes. 1 billion RPS is achievable. Uh we have

verified it a few times. Uh at the same time, it is important to follow the best It it is there would be cases where you wouldn't need this sort of a scale. But again, if you need it, you have it. Uh Valkey operator is something that Joe will talk about just after this. And it is being developed actively in GitHub and uh this is something that will

be supported by the Valkey project. So, yes, come try it out. Help us build that. And uh finally, if you're using Redis, consider switching it to Valkey because >> It's better. >> No, this is on. >> Well, Madeline was saying it's better. And I agree. but yeah, yeah, as I said, Valkey operator is actively being developed. Uh we have made some really good progress. We sync up

every week uh with each other to track track that and make some progress. So, feel free to join these communities. There are like some awesome people who are working. And yes, if Core Valkyrie is something that interests you, get involved with the Uh, we are always happy to have as many contributors as possible. Yeah, thank you so much. >> Is this back on? Yeah, so I guess

we have some interesting some Q&A if you want to. Um, we'll be around for the next like half an hour or so just chilling. But so if folks want to come up and ask some questions, we have time. There's a microphone right behind you. >> Okay. >> Just So the the question was for the 1 billion requests per second, was that using connection pooling? And I believe

the answer is yes. >> Yeah, yeah. We we use connection pooling I mean we honestly weren't really hesitant on scale as well. We just scaled as many nodes as possible and >> Yeah, it's important to note like a typical request inside Valkyrie only takes on the order of like 1 microsecond, but connection establishment takes hundreds of microseconds. So like if you're doing a connection for every request,

like you would need a lot of you're wasting a lot of CPU. >> Yeah. >> Hello. Uh, what did you use for profiling when you were doing all this? >> Uh, it was just a normal perf tool. Like uh, a perf tool that is available I think in most Linux systems as well. So it helps you capture um, like for a designated amount and then through that

basically you can create the uh, flame graphs as well. So that you know the amount of compute that is actually getting spent in like basically what function. So yeah, it's pretty cool. Uh, it helps it helps you basically know everything about the system. So yeah, that that the name of the tool is just perf. So Well, if there's no other questions, thanks a lot. And if you

want any Valkyrie stickers, I do a Oh, wait, perfect. Yes. Friends. >> Uh last one, I guess. Um I have a quick question. Um how Well, using Valkyrie cluster, uh how do you ensure that when they initially ini- initialization happens, that it correctly uh assigns uh the primaries with the replicas instead of um all the pods becoming uh well, masters they're no longer being a clusters, essentially.

>> Yeah. So, today there's not a good So, like that's really the problem the operator is going to have to solve. Um the the default way is you basically just put all the Valkyrie nodes in the cluster, and then some people have had like two different deployment sets, one which can self-configure as replicas. And as long as you basically put the nodes in the cluster and set

them as replicas, there's a thing called um the replica migration like tool uh or config, which will basically kind of configure the right amount of nodes to be replicas. Um so, basically, if there's a primary without a replica in the cluster, an an idle node will just go and presu- assume that. Um that's the poor man's solution. I think the better solution will be the to try

out the operator, which does basically support these first-class constructs as shards. Thank you. Also, Valkyrie 9 is a pretty good update. >> Oh, thanks. >> From from Yeah. >> Thank you, pal. Thank you. >> Thank you so much.