Until recently, about 50 TB of real-time data in Odnoklassniki was stored in SQL Server. Providing fast, reliable, and fault-tolerant data center access for such a volume using SQL DBMS is practically impossible. Typically, one of the NoSQL storage solutions is used in such cases, but not everything can be transitioned to NoSQL; some entities require ACID transaction guarantees.
This led us to use a NewSQL storage solution, which is a DBMS that provides the fault tolerance, scalability, and performance of NoSQL systems while maintaining the ACID guarantees familiar to traditional systems. There are few operational industrial systems of this new class, so we developed such a system ourselves and launched it into production.
How it works and what we achieved — read on.
Today, the monthly audience of Odnoklassniki exceeds 70 million unique visitors. We the largest social networks in the world and among the top twenty websites where users spend the most time. The infrastructure of 'OK' handles extremely high loads: over one million HTTP requests per second at the front. The server park, consisting of more than 8000 units, is located close to each other — in four Moscow data centers, which allows for a network latency of less than 1 ms between them.
We have been using Cassandra since 2010, starting from version 0.6. Today, several dozen clusters are in operation. The fastest cluster processes over 4 million operations per second, while the largest stores 260 TB.
However, all of these are regular NoSQL clusters used for storing data. We wanted to replace the primary consistent storage, Microsoft SQL Server, which had been used since the founding of 'Odnoklassniki'. The storage consisted of more than 300 SQL Server Standard Edition machines that held 50 TB of data — business entities. This data is modified within ACID transactions and requires .
For distributing data across SQL Server nodes, we employed both vertical and horizontal (sharding). Historically, we used a simple data sharding scheme: each entity was assigned a token — a function of the entity's ID. Entities with the same token were placed on the same SQL server. The master-detail relationship was implemented in such a way that the tokens of both the main and child records always matched and were located on one server. In a social network, almost all records are generated on behalf of a user — meaning all the user's data within a single functional subsystem is stored on one server. Thus, business transactions almost always involved tables from a single SQL server, ensuring data consistency through local ACID transactions, without the need for distributed ACID transactions.
Thanks to sharding and to speed up SQL performance:
- We do not use Foreign key constraints since, during sharding, the entity ID may reside on a different server.
- We don't use stored procedures and triggers due to the extra CPU load they place on the DBMS.
- We avoid JOINs for the reasons mentioned above and due to numerous random reads from disk.
- To reduce deadlocks, we use the Read Uncommitted isolation level outside of transactions.
- We only perform short transactions (averaging under 100 ms).
- We don't use multi-row UPDATE and DELETE due to the high number of deadlocks — we update only one record at a time.
- Queries are always executed using indexes — a query with a full table scan plan indicates database overload and potential failure.
These steps have maximized performance from SQL servers. However, problems have been increasing. Let's review them.
SQL Problems
- Since we utilized custom sharding, adding new shards was done manually by administrators. During this time, the scalable data replicas did not handle any requests.
- As the number of records in the table increases, the speed of insertion and modification decreases. Adding indexes to an existing table can significantly reduce performance, and the creation and recreation of indexes involves downtime.
- Having a limited number of Windows servers for SQL Server in production complicates infrastructure management.
But the main issue is—
Fault tolerance
Classic SQL servers have poor fault tolerance. For instance, if you have only one database server and it fails once every three years, the site will be down for 20 minutes, which is acceptable. If you have 64 servers, then the site will be down once every three weeks. If there are 200 servers, the downtime occurs every week. This is a problem.
What can be done to improve the fault tolerance of SQL servers? Wikipedia suggests building a : where in case of failure of any component there is a redundant one.
This requires a fleet of expensive equipment: extensive redundancy, fiber optics, shared storage, and also the inclusion of backups that are unreliable, with about 10% of activations resulting in a failure of the backup node following the main node.
However, the main drawback of such a highly available cluster is zero availability in the event of a failure in the data center where it is located. Odnoklassniki has four data centers, and we need to ensure operation during a complete failure in one of them.
To address this, we could apply replication built into SQL Server. This solution is significantly more expensive due to software costs and suffers from well-known replication issues — unpredictable transaction delays during synchronous replication and delays in applying replications (and, consequently, lost modifications) during asynchronous replication. The implied renders this option completely impractical for us.
All these issues required radical solutions, and we began a detailed analysis. Here, we need to understand the primary role of SQL Server — transactions.
Simple Transaction
Let's consider a basic transaction from the perspective of an application SQL programmer: adding a photo to an album. Albums and photos are stored in separate tables. An album has a counter for public photos. Therefore, this transaction can be broken down into the following steps:
- Lock the album by key.
- Create a record in the photos table.
- If the photo has a public status, increment the public photo counter in the album, update the record, and commit the transaction.
Or in pseudocode:
TX.start("Albums", id);
Album album = albums.lock(id);
Photo photo = photos.create(…);
if (photo.status == PUBLIC ) {
album.incPublicPhotosCount();
}
album.update();
TX.commit();We see that the most common scenario for a business transaction is to read data from the database into the application server's memory, make some changes, and save the new values back to the database. Typically, in such a transaction, we update several entities across multiple tables.
During a transaction, concurrent modifications to the same data from another system may occur. For instance, the anti-spam system might determine that a user is suspicious, leading to all of their photos being marked as non-public and requiring moderation. This means changing photo.status to a different value and adjusting the corresponding counters. It is evident that if this operation takes place without guarantees of atomicity and isolation of competing modifications, like in , the outcome will be incorrect—either the photo counter will display an inaccurate value or not all photos will be sent for moderation.
A significant amount of code that manipulates various business entities within a single transaction has been written during the existence of Odnoklassniki. Based on migration experiences to NoSQL with We understand that the biggest challenges (and the most time-consuming aspects) arise from the need to develop code aimed at maintaining data consistency. Therefore, a primary requirement for the new storage system was to ensure support for true ACID transactions in the application logic.
Other equally important requirements included:
- In the event of a data center failure, both reading from and writing to the new storage must remain available.
- Maintaining the current development speed. This means that when working with the new storage, the amount of code should be roughly the same, without the need to write additional code for the storage or develop conflict resolution algorithms, maintain secondary indexes, etc.
- The performance of the new storage must be sufficiently high for both data reading and transaction processing, effectively ruling out academically rigorous, universal, but slow solutions such as .
- Automatic scaling on the fly.
- Utilization of standard, inexpensive servers, without the need to purchase exotic hardware.
- The ability to develop storage capabilities in-house by the company's developers. In other words, the preference was given to proprietary or open-source solutions, preferably in Java.
Solutions, Options
Analyzing possible solutions, we narrowed it down to two architectural choices:
The first option is to take any SQL server and implement the required fault tolerance, scaling mechanism, fault-tolerant cluster, conflict resolution, and distributed, reliable, and fast ACID transactions. We assessed this option as quite non-trivial and labor-intensive.
The second option is to use a ready-made NoSQL storage solution with built-in scaling, a fault-tolerant cluster, and conflict resolution, while implementing transactions and SQL ourselves. At first glance, even the task of implementing SQL, not to mention ACID transactions, seems like a multi-year endeavor. But then we realized that the set of SQL capabilities we use in practice is as far from ANSI SQL as is from ANSI SQL. Upon closer inspection of CQL, we understood that it is quite similar to our needs.
Cassandra and CQL
So, what makes Cassandra interesting and what capabilities does it offer?
First, you can create tables that support various data types, and you can perform SELECT or UPDATE operations by primary key.
CREATE TABLE photos (id bigint PRIMARY KEY, owner bigint,…);
SELECT * FROM photos WHERE id=?;
UPDATE photos SET … WHERE id=?;To ensure data consistency across replicas, Cassandra employs . In the simplest terms, this means that when placing three replicas of the same row on different nodes of the cluster, the write is considered successful if the majority of nodes (i.e., two out of three) confirm the success of the write operation. Row data is considered consistent if, during reading, a majority of nodes were queried and confirmed their status. Thus, with three replicas, full and instantaneous data consistency is guaranteed even if one node fails. This approach has allowed us to implement an even more reliable scheme: always sending requests to all three replicas while waiting for responses from the two fastest. The delayed response from the third replica is discarded in this case. The node with the delayed response may have significant issues—such as lag, garbage collection in the JVM, direct memory reclaim in the Linux kernel, hardware failure, or network disconnection. However, this does not affect client operations or data.
The approach where we query three nodes and receive a response from two is called : requests to excess replicas are sent even before the node 'drops out'.
One of the advantages of Cassandra is Batchlog — a mechanism that ensures either the complete application or the complete rejection of the batch of changes you make. This allows us to achieve A in ACID — atomicity out of the box.
The closest thing to transactions in Cassandra is the so-called ““. However, they are far from “real” ACID transactions: in reality, this allows for a on data from just one record, using consensus through the heavyweight Paxos protocol. Hence, the speed of such transactions is limited.
What we missed in Cassandra
Thus, we aimed to implement true ACID transactions in Cassandra. With which we could easily enable two other convenient features typical of classic DBMS: consistent fast indexing, allowing us to retrieve data not just by the primary key, and a traditional generator for monotonous auto-incrementing IDs.
C*One
This led to the creation of a new DBMS C*One, consisting of three types of server nodes:
- Storage — (almost) standard Cassandra servers responsible for storing data on local disks. As the load and data volume increase, they can be easily scaled up to dozens or even hundreds.
- Transaction coordinators — ensure the execution of transactions.
- Clients — application servers that carry out business operations and initiate transactions. There can be thousands of such clients.
Servers of all types are part of a common cluster, using the internal Cassandra messaging protocol to communicate with each other and exchange cluster information. Through Heartbeat, servers learn about mutual failures, maintain a consistent data schema — tables, their structure and replication; partitioning schema, cluster topology, etc.
Clients
Instead of standard drivers, a Fat Client mode is employed. This node does not store data but can act as a coordinator for executing requests; the Client itself fulfills the coordinator role for its requests: it queries storage replicas and resolves conflicts. This method is not only more reliable and faster than the standard driver, which requires communication with a remote coordinator, but also allows for better management of request transmission. Outside of an open transaction on the client, requests are directed to the storage. If the client opens a transaction, all requests within that transaction are directed to the transaction coordinator.
Transaction Coordinator C*One
The coordinator is what we developed from scratch for C*One. It is responsible for managing transactions, locks, and the order of transaction applications.
For each transaction processed, the coordinator generates a timestamp: each subsequent one is greater than that of the previous transaction. Since in Cassandra, the conflict resolution system is based on timestamps (of two conflicting records, the one with the later timestamp is considered valid), a conflict will always be resolved in favor of the subsequent transaction. This way, we have implemented — a cost-effective way to resolve conflicts in a distributed system.
Locks
To ensure isolation, we decided to use the simplest method—pessimistic locking based on the primary key of the record. In other words, in a transaction, the record must first be locked, then read, modified, and saved. Only after a successful commit can the record be unlocked so that competing transactions can utilize it.
Implementing such a lock is straightforward in a non-distributed environment. In a distributed system, there are two main approaches: either to implement distributed locking at the cluster level or to coordinate transactions so that those involving a single record are always managed by the same coordinator.
Since in our case the data is already distributed across groups of local transactions in SQL, it was decided to assign coordinators to these groups: one coordinator handles all transactions with tokens from 0 to 9, the second coordinator manages tokens from 10 to 19, and so on. As a result, each instance of the coordinator becomes a master for its group of transactions.
Therefore, locks can be implemented as a simple HashMap in the coordinator's memory.
Coordinator failures
Since a single coordinator exclusively serves a group of transactions, it is crucial to quickly identify its failure to ensure that the retry of the transaction completes within the timeout. To achieve this quickly and reliably, we applied a fully-connected quorum heartbeat protocol:
Each data center hosts at least two coordinator nodes. Periodically, each coordinator sends a heartbeat message to the other coordinators, informing them of its status and which coordinator's heartbeat messages it received last.
By receiving similar information from the others in their heartbeat messages, each coordinator determines which cluster nodes are functioning and which are not, based on the quorum principle: if node X receives confirmation of normal message receipt from the majority of nodes in the cluster regarding node Y, it means Y is operational. Conversely, as soon as the majority reports a loss of messages from node Y, it indicates Y has failed. Interestingly, if the quorum tells node X that it has not received messages from it for a while, node X will consider itself failed.
Heartbeat messages are sent at a high frequency, about 20 times per second, with a 50 ms interval. In Java, it's challenging to ensure an application response within 50 ms due to comparable pause durations caused by the garbage collector. We achieved this response time using the G1 garbage collector, which allows us to specify a pause duration target for GC. However, occasionally — though rarely — the pauses exceed 50 ms, which can lead to false failure detection. To prevent this, the coordinator does not report a remote node failure upon missing the first heartbeat message, only when several consecutive ones are lost. This allowed us to achieve coordinator node failure detection within 200 ms.
It's not enough to quickly identify which node has stopped functioning. Action needs to be taken.
Redundancy
The classic scheme suggests initiating the election of a new master in case of a failure using one of the algorithms. However, such algorithms have well-known issues with convergence time and the duration of the election process itself. We managed to avoid these additional delays using the coordinator substitution scheme in a fully connected network:
Let’s say we want to execute a transaction in group 50. We will first define the substitution scheme, specifying which nodes will carry out transactions for group 50 in the event of the primary coordinator's failure. Our goal is to maintain system functionality during a data center failure. We will determine that the first backup will be a node from another data center, and the second backup will be a node from a third data center. This scheme is established once and does not change unless the cluster's topology changes, meaning new nodes join (which happens very rarely). The order of choosing a new active master in case of the old one’s failure will always be as follows: the first backup becomes the active master, and if it also ceases to function — the second backup.
This scheme is more reliable than a universal algorithm because activating a new master only requires confirming the failure of the old one.
But how will clients know which master is currently active? It's impossible to send information to thousands of clients in just 50 ms. A scenario may arise where a client sends a request to open a transaction, still unaware that the current master is no longer functioning, causing the request to hang on timeout. To prevent this, clients speculatively send their transaction opening requests to the group master and both of its backups, but only the active master will respond to that request. All subsequent communication regarding the transaction will be conducted solely with the active master.
Backup masters place received requests for transactions not belonging to them in a queue of pending transactions, where they are stored for a certain period. If the active master fails, a new master processes requests to open transactions from its queue and responds to the client. If the client has already opened a transaction with the old master, the second response is ignored (and, obviously, that transaction will not complete and will need to be repeated by the client).
How the transaction works
Suppose a client sends a request to the coordinator to open a transaction for a specific entity with a particular primary key. The coordinator blocks this entity and places it in the lock table in memory. If necessary, the coordinator reads this entity from storage and saves the retrieved data in the transaction state in its memory.
When a client wants to modify data in the transaction, they send a modification request for the entity to the coordinator, who places the new data in the transaction state table in memory. At this point, the entry is complete — no data is written to storage.
When a client requests their own modified data within an active transaction, the coordinator acts as follows:
- if the ID is already present in the transaction, the data is retrieved from memory;
- if the ID is not in memory, the missing data is read from the node storages, combined with what is already in memory, and the result is returned to the client.
Thus, the client can read their own changes, while other clients do not see these changes because they are only stored in the coordinator's memory; they are not yet present in the Cassandra nodes.
When a client sends a commit, the state held in the service's memory is saved by the coordinator in a logged batch, which is then sent to the Cassandra stores. The stores ensure that this batch is applied atomically (in full) and return a response to the coordinator, which releases locks and confirms the transaction's success to the client.
To roll back, the coordinator only needs to free the memory occupied by the transaction's state.
As a result of the above enhancements, we have implemented ACID principles:
- Atomicity. This guarantees that no transaction will be partially recorded in the system; either all of its sub-operations will be executed, or none at all. We uphold this principle through the logged batch in Cassandra.
- Consistency. Each successful transaction, by definition, records only permissible outcomes. If, after opening a transaction and executing part of the operations, it is discovered that the result is not permissible, a rollback is performed.
- Isolation. When executing a transaction, concurrent transactions should not affect its outcome. Competing transactions are isolated using pessimistic locks at the coordinator. For reads outside of transactions, the Read Committed isolation principle applies.
- Durability. Regardless of issues at lower levels — system power loss, hardware failure — changes made by a successfully completed transaction must remain saved after the system resumes operation.
Index-based Reads
Let's take a simple table:
CREATE TABLE photos (
id bigint primary key,
owner bigint,
modified timestamp,
…)It has an ID (primary key), an owner, and a modification date. We need to create a very simple query — to select data by owner with a modification date of "in the last day."
SELECT *
WHERE owner=?
AND modified>?For such a query to execute quickly, you need to build an index on the columns (owner, modified) in a traditional SQL database. We can accomplish this quite easily, as we now have ACID guarantees!
Indexes in C*One
There is a source table with photos, where the record ID is the primary key.
For the C*One index, a new table is created that is a copy of the original. The key matches the index expression, which also includes the primary key of the record from the original table.
Now the query for "owner in the last 24 hours" can be rewritten as a select from another table:
SELECT * FROM i1_test
WHERE owner=?
AND modified>?The consistency of the original photos table and the index i1 is automatically maintained by the coordinator. Based solely on the data schema, when changes are received, the coordinator generates and records changes not only for the main table but also for the copies. No additional actions with the index table are performed, logs are not read, and locks are not used. This means that adding indexes consumes almost no resources and has little impact on the speed of applying modifications.
With ACID, we successfully implemented SQL-like indexes. They are consistent, scalable, fast, can be composite, and are integrated into the CQL query language. No changes are required in the application code to support the indexes. It's as straightforward as SQL. Most importantly, the indexes do not affect the execution speed of modifications to the original transaction table.
What we achieved
We developed C*One three years ago and launched it into production.
So, what did we accomplish? Let's evaluate this using the example of the photo processing and storage subsystem, one of the most critical types of data in a social network. It's not just about the photos themselves but also about the various metadata. Currently, there are about 20 billion such records in ‘Odnoklassniki,’ the system handles 80,000 read requests per second and up to 8,000 ACID transactions per second related to data modifications.
When we used SQL with a replication factor of 1 (but in RAID 10), the metadata for the images was stored on a highly available cluster of 32 machines with Microsoft SQL Server (plus 11 backups). Additionally, 10 servers were allocated for storing backups, bringing the total to 50 expensive machines. The system operated under nominal load, without any surplus.
After migrating to the new system, we achieved a replication factor of 3 — one copy in each data center. The system consists of 63 Cassandra storage nodes and 6 coordinator machines, totaling 69 servers. However, these machines are significantly cheaper, with a total cost amounting to about 30% of the SQL system's cost. The load remains at approximately 30%.
With the implementation of C*One, latency has decreased: in SQL, the write operation took approximately 4.5 ms. In C*One, it's around 1.6 ms. The average transaction duration is under 40 ms, with commits completing in 2 ms and read/write durations averaging 2 ms. The 99th percentile is only 3-3.1 ms, and the number of timeouts has decreased by a factor of 100 — all due to the extensive application of speculation.
Currently, most SQL Server nodes have been decommissioned, and new products are only being developed using C*One. We have adapted C*One for operation in our cloud. , which has accelerated the deployment of new clusters, simplified configuration, and automated maintenance. Without the source code, this would have been significantly more complex and cumbersome.
We are now working on migrating our other storage systems to the cloud — but that's a completely different story.
Source: habr.com
