Jorge Bay Gondra

We recently introduced changes in the client drivers that affects how incoming requests are distributed across an Apache Cassandra cluster. The new default load balancing algorithm that is being rolled out in the drivers helps reduce long tail latency and provides better overall response times.

<h2>How data partitioning and replication works in Cassandra</h2>

Cassandra places the data on a node according to the value of the partition key. For each partition key, multiple replicas are used to ensure reliability and fault tolerance. A replication strategy determines the nodes where replicas are placed. The total number of replicas across the cluster is referred to as the replication factor. For example, a replication factor of 3 means that there are three replicas of each row in a given data center or cluster.

 
For a given query request, the driver selects a replica that acts as coordinator for the query and is responsible for meeting the <a href="https://docs.datastax.com/en/cassandra-oss/3.x/cassandra/dml/dmlAboutDataConsistency.html">consistency level</a> guarantees.
<img alt="1-Ibp" data-caption="&lt;em&gt;Write-path example: The client driver selects node A, that is a replica for a given key and acts as coordinator for the request, sending the data to the other replicas (E and C)&lt;/em&gt;" data-entity-type="file" data-entity-uuid="bc8f388f-7baf-4ac1-af57-6932c2b09126" src="https://www.datastax.com/sites/default/files/inline-images/1-lbp-client-coordinator-replica.jpg" />
<h2>&nbsp;</h2>

<h2>How client routing works</h2>

The DataStax Drivers control the distribution of incoming queries across the cluster. Using cluster metadata, the driver knows which nodes are replicas for a given key, sending the query request directly to a replica to avoid extra network hops.

When the replication factor is greater than 1, the driver balances the requests across the different replicas. Historically, the drivers used to select the coordinator of the query from the replica set in a random manner.

<h2>Random selection</h2>

Given that microservice applications usually consist of multiple service instances, we have to consider the client driver as a distributed load balancer (n clients routing traffic to m nodes). Random selection makes a good distributed load balancer algorithm because it's stateless, requires no communication between client instances and generates a uniform load.
<img alt="2Ibp" data-caption="&lt;em&gt;Multiple clients distribute requests to the same nodes&lt;/em&gt;" data-entity-type="file" data-entity-uuid="0e04e940-e01e-4fc9-8719-efe2794944c6" src="https://www.datastax.com/sites/default/files/inline-images/2-lbp-multiple-clients%20%281%29.jpg" />
&nbsp;

With random selection, the request load is distributed uniformly but not in a fair manner: it doesn't account for size/complexity of the requests and other tasks that might be occurring in the background of the server node.

<h2>Power of two random choices</h2>

Deterministically selecting a single candidate based on a signal can be dangerous in a distributed system: multiple clients with the same logic may identify the same server node as the best candidate and redirect all traffic to it, causing load spikes and risking a cascading failure. For this reason, we still need an algorithm with some degree of randomness.

We first encountered the <a href="https://www.eecs.harvard.edu/~michaelm/postscripts/mythesis.pdf">Power of Two Random Choices</a> algorithm via <a href="https://www.infoq.com/presentations/load-balancing/">this talk</a> by <a href="https://twitter.com/tbmcmullen">Tyler McMullen</a>. Since then, it has been implemented by load balancing software like <a href="https://netflixtechblog.com/netflix-edge-load-balancing-695308b5548c">Netflix's Zuul</a>, <a href="https://www.nginx.com/blog/nginx-power-of-two-choices-load-balancing-algorithm/">Nginx</a> and <a href="https://www.haproxy.com/blog/power-of-two-load-balancing/">Haproxy</a> and in numerous service mesh deployments, like at <a href="https://blog.twitter.com/engineering/en_us/topics/infrastructure/2019/daperture-load-balancer.html">Twitter</a>. The logic is simple: instead of trying to select the absolute best candidate using a signal, with “power of two choices” you pick two candidates at random and choose the better option of the two, avoiding the worse choice.

There are several approaches we can use to identify a node as a best candidate. In the past, we exposed a way to use latency information (LatencyAwarePolicy) as a signal. Since we are routing database requests without considering the complexity of the queries, using aggregated latency information of a node is not a good indicator of current health or state of the server instance. For example, a query might take longer because it involves consolidating lots of data but the server could still have plenty of resources available for handling other queries. Additionally, the response time of a request is determined after the response is received, providing an out-of-date view of the server state.

We decided to choose the best candidates using the number of in-flight requests to each node from the client driver instance. In a nutshell, we select two replicas at random and we pick the replica with fewer in-flight requests as coordinator.

<h2>Stale queue detection</h2>

Thanks to our knowledge of how the nodes behave, we found more room for improvement. When a node is undergoing <a href="https://docs.datastax.com/en/cassandra-oss/3.x/cassandra/dml/dmlHowDataMaintain.html#dmlHowDataMaintain__dml-compaction">compaction</a> or GC, it might take longer to process items from the request queue. Even if the request queue for a node is stale, under a scenario of increased load the client driver might continue to route traffic to it if we base the algorithm solely on the shortest queue.

As an example, consider the following scenario.
<img alt="3-queue" data-caption="&lt;em&gt;The system with in-flight requests represented as queues.&lt;/em&gt;" data-entity-type="file" data-entity-uuid="3ae1cb8c-e11b-49ef-ad90-9da3a864d01c" src="https://www.datastax.com/sites/default/files/inline-images/3-queue-stale.jpg" />
&nbsp;

There's a steady flow of requests and a client driver instance within an application is balancing requests between the replicas. At some point in time, the node C becomes busy (doesn't process requests for some time). The other healthy replicas will continue processing requests like the diagram shows.

At the same time the incoming load to the system increases, to which point that the healthy replicas have a larger number of in-flight requests.
<img alt="4 queue" data-caption="&lt;em&gt;The number of in-flight requests on the healthy nodes increased as a result of increased load.&lt;/em&gt;" data-entity-type="file" data-entity-uuid="0d4f6d15-9179-4d12-90ed-df1aaad563eb" src="https://www.datastax.com/sites/default/files/inline-images/4-queue-avoidance.jpg" />
&nbsp;

In that case, we would still be routing requests to a node we know is not processing them at a healthy pace.

To overcome the scenario explained above, we introduced a second indicator: whether a node has requests in-flight (10+) and hasn't sent back any response in the last few hundred milliseconds. This second indicator is not factored in as a signal of the power of two choices but as initial probing of the health of the replicas.

When there is a majority of healthy replicas, unhealthy ones are excluded as initial coordinators (will still be considered for retries and speculative executions). Why probing the health of replicas as a first step? Because if there are a large number of nodes being considered unhealthy, it could mean that this indicator doesn't apply the workload or/ type of queries or expected latencies. That way, we make sure we don't make things worse by making bad decisions based on a signal that in a particular case doesn't apply.

<h2>Summary</h2>

We tested the new algorithm against a series of scenarios with multiple clients at the same time, simulating nodes under stress using CPU burns (stress-ng) and artificial network latency (tc) and the result is that the new algorithm outperformed all other approaches.

As a sample, the following diagram shows latency by percentile distribution, comparing the new algorithm to random selection with some of the nodes experiencing CPU burns.
<img alt="5 latency" data-caption="&lt;em&gt;The new algorithm (blue) with a lower tail latency&lt;/em&gt;" data-entity-type="file" data-entity-uuid="00b7d732-b426-44bc-b93d-3efe7882a3ab" src="https://www.datastax.com/sites/default/files/inline-images/5-latency-distribution.jpg" />
&nbsp;

The new load balancing algorithm, based on power of two choices with additional logic for busy node avoidance, is becoming the new default on the DataStax drivers. With a simple logic, it avoids the undesired herd behavior by bypassing the worst candidate and still distributing traffic with a degree of randomness, resulting in significant lower tail latencies. 
We are still rolling it out across all client drivers, with Java and Node.js drivers already using this algorithm as default since version 4.4 and the rest of the drivers following shortly after. If you have questions or doubts, feel free to reach us on the mailing lists for <a href="https://groups.google.com/a/lists.datastax.com/forum/#!forum/cpp-driver-user">C++</a>, <a href="https://groups.google.com/a/lists.datastax.com/forum/#!forum/csharp-driver-user">C#</a>, <a href="https://groups.google.com/a/lists.datastax.com/forum/#!forum/java-driver-user">Java</a>, <a href="https://groups.google.com/a/lists.datastax.com/forum/#!forum/nodejs-driver-user">Node.js</a> and <a href="https://groups.google.com/a/lists.datastax.com/forum/#!forum/python-driver-user">Python</a> drivers.

Improved Client Request Routing for Apache Cassandra™

Jorge Bay Gondra

Share

Share

How data partitioning and replication works in Cassandra

How client routing works

Random selection

Power of two random choices

Stale queue detection

Summary

More Technology

How to Build a Crystal Image Search App with Vector Search

Knowledge Graphs for RAG without a GraphDB

How Winweb Built its AI Assistant with DataStax Astra DB and LangChain

Vercel + Astra DB: Get Data into Your GenAI Apps Fast

One-stop Data API for Production GenAI