[NEW] Primary replica role at the slot level

[zvi-code]

The problem/use-case that the feature addresses

Currently, replicas provide availability, increased durability (from a domain failure perspective), and performance improvements when using Read from Replica (RFR). However, performance scaling is limited to stale reads and does not extend to regular writes\reads. Many customers do not use RFR for various reasons. Additionally, when a primary node fails, all write traffic to that node fails [potentially requiring application-level logic to handle the failure].

Description of the feature

We propose redefining role assignments from the node level to the slot level. In this model, a node can be the primary for certain slots and a replica for others. This involves adjusting the codebase so that any primary/replica designations are applied to slots rather than nodes. Essentially, the node becomes a logical container of compute, memory, and services that manages atomic data entities (slots).

With this approach, we can scale the performance of both writes and reads based on the number of nodes in a shard, eliminating the concept of a replica node. If a node fails, only the slots for which it was the primary are directly impacted, improving fault granularity\isolation.

The recent introduction of the dict-per-slot has shifted many processes to operate at the slot level, which facilitates the transition to this model. As part of this feature we will need to continue going down this path for other flows in the system, including bgsave, for example.

This change would require client support, but for clients that do not have the support we can initially implement the feature in a degenerated form where all slots in a shard have the same primary node, maintaining backward compatibility.

Additional information

An added benefit of this approach is the potential to reduce code complexity by unifying the code paths of replication and slot migration, which are currently two similar processes for maintaining data consistency between nodes.

For Cluster Mode Disabled (CMD), we can consider all data to reside in slot 0. In the long term, we might consider enabling slots (or logical grouping) for CMD, allowing customers to gain the benefits of this model without adopting Cluster Mode Enabled (CME).

[zvi-code]

Even if we support the idea directionally. there are so many things to consider, opened this issue to hear initial feedback to the direction, see if there is community interest, or maybe it was discussed before and so on...

[madolson]

I've heard a lot of feedback that folks like the way hazel cast supports "homogenous clusters". The idea being that all nodes own some set of data and replicate it to others, which sounds very similar to what you're proposing. I overall really like this, and would dramatically simplify a lot of setups.

We did also recently introduce the concept of a "shard", which makes this flexible slot ownership more difficult. We could implement something like a shard can have multiple primaries over different parts of the data, either that or allow nodes to be part of multiple shards. It would also hopefully keep the replication stream pretty straightforward.

We also need per slot replication effectively for atomic slot migration.

[zvi-code]

Interesting, didn't think of non-homogenous hardware explicitly, I was thinking of, "a mirror of the same problem" that is non-homogenous slots [in cpu\memory usage], essentially the same technical problem but with a different user/business problem[utilize non homogenous hw vs handle unbalanced slots].
IMO when going down this road we should choose the path that is incremental. Semantically I suggest to move the primary\replica role to the slot[or call it atomic-data-entity].
An example for this consideration is wether we have the concept of shard or not. So incremental change is for example, consider these 2 options: 1) No shard would mean we place copies of slot based on usage\availability and not based on any additional grouping assumption. 2) Have a shard idea and place copies of a slot within a shard with some flexibility.
So option 2 of keeping the shard concept with equal copies for each slot in the shard is the natural incremental step from current state.
I guess that this can serve well also the idea of sub-cluster, in the context of next get cluster design (cluster bus...), because we can have associate sub-cluster with set of slots and the sub-cluster can reconcile it's owned slots locations without the rest of the cluster - but this is not a well thought, by me, idea TBH

[murphyjacob4]

+1 to this idea. I think this is the right long term vision for a feature like atomic slot migration. The main challenge for this will be that the code is very coupled to the idea of a node either being just a primary or just a replica, and this assumption is made in how some things are designed (I.e. we assume that serving traffic while full syncing isn't very important since it is just a replica).

Brainstorming some incremental improvements that we can make to reach this goal:

1. The "replication link" concept will need to go from a "one-to-one" relationship with a node to a "many-to-one" relationship with a node. We would refactor existing replication logic to remove dependencies on server level global variables and instead make them operate on a per-link basis. I.e. we can have many SYNCs, PSYNCs, active replication streams ongoing at the same time and need to de-globalize the state of those
2. Async loading needs improvements. If a node is going to be a primary and a replica at the same time, we need to support a true asynchronous load of RDB (for full sync) that doesn't drastically disrupt the primary traffic. Right now, async loading is more of an afterthought where we sometimes process events (at most once every 100ms IIUC and at most every 2 MiB of loaded data) and that won't work for a primary serving production workloads. I think delegating async loading to a background thread could work - so long as we are not serving traffic in the dictionary that is being loaded. I think there will be considerations to be had regarding modules here (modules probably won't support thread safety when getting an RDB load callback from a background thread). Probably it will come with some module compatibility flag to support background thread load, and if not supported module RDB loads will have to be enqueued back to the main thread.

Regarding appcompat - I think probably the best path forward is to add a CLIENT CAPA for "replication-at-slot-level" or something like that. If it isn't supported, we can sanitize the CLUSTER SLOTS output as follows:

• If a node has any primaryship of any number of slots, it will be presented as primary of those slots
• If a node has no primaryship of any slots, it will be displayed as a replica if and only if all slots are owned by only one node
• If a node has no primaryshp of any slots, and it is replicating slots from >1 other nodes, we can present as a replica if we replicate all slots from at least one other node. I.e. we could present as replica of node A if it owns slots 1-10 and we replicate slots 1-10 and 100-110.
• If a node has no primaryhsip of any slots, and it is not replicating 100% of the slots from at least one primary, it will have to either be presented as an empty primary or completely removed from the CLUSTER SLOTS output.

We can still have features like atomic slot migration at that point, but clients wouldn't be able to observe the intermediary state of "node B owns slot N and is a replica of slot M in node A" and instead would just see an immediate transition from "node B owns slot N and node A owns slot M" to "node B owns slot N and M and node A owns nothing".

[asafpamzn]

@madolson can you please add label Client support required. It will be easier for clients maintainers to track these changes

[nandihalli]

Apart from what @murphyjacob4 called out, I think we should also consider the implications of what happens when a node (which was primary for multiple slots) goes down? Since these replica nodes for different slots can be spread across multiple nodes, it is likely that we would need multiple nodes across the cluster needing to be elected via leader election?

In the worst case scenario, let's say that in a 100 node cluster, a node owned slots 1-1000 and was the primary for those slots. If the replica for slots 1-1000 was such that they are distributed across the 100 nodes, it would mean that all 100 nodes are potential candidates to start owning these 1000 slots when the existing node owning 1-1000 goes down. Would they all compete for leader election? So, we probably still need some sort of rules around how do we want to restrict the ownership of slots across nodes? Also, leader election would now happen per slot vs per node? We will need to think about the scalability aspect of it as well?

[zvi-code]

@murphyjacob4, I missed you comment earlier...

Async loading needs improvements. If a node is going to be a primary and a replica at the same time, we need to support a true asynchronous load of RDB (for full sync) that doesn't drastically disrupt the primary traffic. Right now, async loading is more of an afterthought where we sometimes process events (at most once every 100ms IIUC and at most every 2 MiB of loaded data) and that won't work for a primary serving production workloads. I think delegating async loading to a background thread could work - so long as we are not serving traffic in the dictionary that is being loaded. I think there will be considerations to be had regarding modules here (modules probably won't support thread safety when getting an RDB load callback from a background thread). Probably it will come with some module compatibility flag to support background thread load, and if not supported module RDB loads will have to be enqueued back to the main thread.

Agree about the need to handle rdb load differently as it would break current behavior. One way I would think to handle this is by first populating the slot dict, in a thread, and add it to the kvstore only when it's fully populated with the data (you can think of it as the equivalent to flush async). This might help simplify the access control while slot is not loaded.

The "replication link" concept will need to go from a "one-to-one" relationship with a node to a "many-to-one" relationship with a node. We would refactor existing replication logic to remove dependencies on server level global variables and instead make them operate on a per-link basis. I.e. we can have many SYNCs, PSYNCs, active replication streams ongoing at the same time and need to de-globalize the state of those

I agree we need to decouple dependency. If we keep the shard boundary, it will be sufficient to have a connection between all nodes in the shard and use some form of multiplexing over the connections for combining send\respond from\to both sides. For full-sync I assume we can use dedicated new connection, as i would assume this flow to be very similar to that of slot-migration[hope we can reuse the same mechanism].

---

@nandihalli, I agree this could be an issue and should be thought of carefully! If we maintain the shard boundaries(shard:="set of nodes where all nodes hold same slots & for each of the slots in the shard, one node in the shard is the primary and rest are replicas at any pit") node failure does not change the shard of a slot, but might move between nodes in the same shard.
If we completely remove the shard boundaries, we can still maintain logical buckets of slots that move together (call them super-slot) to simplify management and reduce processing. [BTW:It's a topic for a different discussion, but I think that even in the current code, there is no value to have 16K slots in 2 shards cluster and a super-slot concept could make sense as well]

[zuiderkwast]

Yeah, there are several good points in this issue. Several databases use this "homogeneous cluster" style.

• The load on each node is more even.
• It allows smaller clusters. The smallest cluster can now be 3 nodes where each node is primary of 1/3 of the slots and replica of other slots.

The deployment might get more complicated though, and it's already quite complicated. It's not easy to get the things like auto-scaling and rolling upgrade right and I don't like that we rely on proprietary control planes. I would like to see more control-plane things being handled automatically by the cluster itself. (Atomic slot migration is a step in this direction.)

The "shard" concept can probably be redefined. I don't think it's a major problem. For example, we can say that a shard is a set of slots that live on the same primary and the same replicas.

This change would require client support, but for clients that do not have the support we can initially implement the feature in a degenerated form where all slots in a shard have the same primary node, maintaining backward compatibility.

We can make this work for CLUSTER SLOTS. Each slot range has a primary and some replicas. Most clients use CLUSTER SLOTS. It might already work in many clients.

Apart from what @murphyjacob4 called out, I think we should also consider the implications of what happens when a node (which was primary for multiple slots) goes down? Since these replica nodes for different slots can be spread across multiple nodes, it is likely that we would need multiple nodes across the cluster needing to be elected via leader election?

I think only the replica of some slot can become primaries of that slot after a failover. We could do leader election per slot-range (the new shard definition).

Long term idea: If we can require that each slot has at least two replicas, then we could let only the replicas of each slot vote and have a quorum per slot. A raft cluster per slot (or slot-range). If we use raft-based replication too, then we can ensure no dataloss on failover as in #1355. This seems very far away now but it's kind of the direction I would like to see in the long term.