The UMP (Unified MySQL Platform) system is a low-cost, high-performance MySQL cloud data solution developed by the core system database team at Taobao, with key modules implemented in Erlang. The system includes components such as controller servers, proxy servers, agent servers, API/Web servers, log analysis servers, and statistics servers, and relies on open-source components like Mnesia, LVS, RabbitMQ, and ZooKeeper.
In our previous article, “Exploring the Architecture of a Low-Cost, High-Performance MySQL Cloud Database,” we introduced the system structure and the functions of each component of UMP. In this article, we will further explore the application of RabbitMQ and ZooKeeper in the system, the implementation of the proxy server, and how the entire system achieves disaster recovery, read/write splitting, sharding and table partitioning, as well as introduce techniques for resource management, isolation, and scheduling, and the methods used to safeguard user data security.
RabbitMQ
RabbitMQ is an industrial-grade message queue product developed in Erlang. Communication between nodes in the cluster (excluding the transmission of large data streams like SQL queries and logs, which still go directly over TCP) all passes through RabbitMQ, serving as a messaging middleware to guarantee the reliability of message delivery.
When the cluster initializes, a queue is created in RabbitMQ for each node in the cluster, acting as the node’s “mailbox.” When sending messages between nodes, regardless of whether the recipient is online, the sender simply writes the message to the recipient’s “mailbox.” Subsequently, the RabbitMQ client running on the recipient node receives the message and invokes the corresponding processing routine. After the message is processed, the client replies with an ACK packet to RabbitMQ, removing the message from the “mailbox.” RPC can be implemented based on RabbitMQ: in addition to replying with an ACK packet to RabbitMQ to delete the Request message, the client also writes a Reply message to the sender’s “mailbox.” RabbitMQ supports transactions, ensuring that the deletion of the Request message and the writing of the Reply message are completed in a single atomic operation.
Figure 1: Nodes Implementing RPC via RabbitMQ
If the receiver crashes while processing a message, the message remains stored in RabbitMQ. After a restart, the message will be pushed again for the receiver to continue processing.
RabbitMQ can guarantee that a message is sent out and processed by the receiver, but unfortunately, it cannot guarantee that a message is sent/processed only once. The main reason is that RabbitMQ does not support XA. First, the sender cannot complete writing the message to MQ and writing a local log within the same transaction. If the sender crashes after writing the message to MQ but before writing the local log, it cannot determine whether the message was sent upon restart and can only attempt a resend. Similarly, the message receiver cannot place processing the message and deleting it from MQ within the same transaction. If the message receiver crashes after processing the message but before deleting it from MQ, it will continue to receive and process this message again after restarting.
Therefore, the message receiver must ensure idempotency when processing messages, meaning that processing the same message multiple times has no side effects. For example, when the controller sends a backup command to an agent, it can include the timestamp of the last backup. The agent checks if this timestamp matches before executing the backup operation, ensuring that multiple sends of the same backup command do not create multiple backups.
Using RabbitMQ’s routing capabilities (Exchange) also enables message broadcasting. For instance, an Exchange called proxy is created in the system, with the type configured as ’fanout’. When a new proxy server registers, the node’s “mailbox” binds to this Exchange. This way, when the controller server needs to send a notification to all proxy servers, such as executing a master-slave switch operation, the message sent to the Exchange is written to all proxy servers’ “mailboxes.”
RabbitMQ also implements a mirrored queue algorithm to provide HA. When creating a queue, you can pass in &ldquoThe “x-ha-policy” parameter configures the queue as a mirrored queue, which is stored across multiple RabbitMQ nodes in a primary-multi-slave architecture. The “x-ha-policy-params” parameter can specify the list of master and slave nodes. All operations sent to the mirrored queue, such as message publishing and deletion, are first executed on the master node and then synchronized to all slave nodes via an atomic broadcast algorithm called GM (Guaranteed Multicast). The GM algorithm uses a two-phase commit to ensure that messages sent from the master to all slave nodes either all succeed or all fail. By using a ring-based message sending order—where the master sends a message to one slave, that slave forwards it to the next, and the message eventually returns to the master—it ensures that the load difference between master and slave nodes remains minimal.
ZooKeeper
ZooKeeper provides distributed locks, naming services, and more in distributed clusters. It likens the distributed cluster to a zoo, while it plays the role of the zookeeper. ZooKeeper was originally developed by Yahoo! and used within the Hadoop software stack to fulfill the role of Google Chubby. In our project, we use ZooKeeper independently to achieve three functions:
1. Acting as a global configuration server. Configuration files were originally stored locally, and changing configurations required modifications on all nodes, which was not only repetitive but also error-prone. After moving them to ZooKeeper, all nodes watch for changes to the configuration file. Once the file is modified, all nodes reload it and trigger corresponding actions.
2. Providing distributed locks. Multiple controller servers are deployed in the cluster to achieve HA through hot standby, but these controller servers cannot perform the same operation simultaneously. For example, if a MySQL instance goes down and all controller servers track and handle it, initiating a master-slave switchover process, the proxy servers and agent servers would receive multiple switchover commands, throwing the cluster into chaos. Therefore, for simplicity, we stipulate that at any given time, only one leader can be elected from the multiple controller servers in the entire cluster, and this leader is responsible for initiating various system tasks. The leader election function is implemented using ZooKeeper’s distributed lock capability.
3. Monitoring all MySQL instances. We developed a ZooKeeper client plugin for the MySQL server. Upon startup, it connects to the ZooKeeper server and creates an ephemeral node. If the MySQL process dies, this ephemeral node is deleted after a 5-second timeout, which is then detected by a background monitoring daemon. If the dead MySQL process was the master database, it triggers a master-slave switchover process; if it was a slave database, the read weight for that slave is set to 0.
Disaster Recovery
When a MySQL server fails, the system executes a failure recovery process that is transparent to the user. Users are unaware of master database crashes and recovery events, as the proxy server hides these events from them, providing an always-available database connection.
For each user, the system maintains two MySQL instances: a master and a slave. The replication relationship between the master and slave is configured as a Dual Master structure, meaning each MySQL instance sets the other as its own Master, reads data updates from the other, and replicates them locally. This way, writing data to either MySQL instance will be updated on the other. The problem with the Dual Master structure is that if both MySQL instances modify the same row of data simultaneously, conflicts can occur, resulting in inconsistent data versions between the two instances. Therefore, to ensure data consistency, it is also necessary to maintain a "single write" policy, i.e., writing only to the master database. This is guaranteed by the proxy server.
When the master database crashes, the ephemeral node maintained by the MySQL plugin on ZooKeeper is deleted due to session timeout. Upon detecting this event, the controller server initiates a master-slave switchover operation, marks the master as unavailable in the routing table, and notifies all proxy servers via RabbitMQ to perform the switchover.
When the crashed master comes back online, the strategy becomes slightly more complex. At this point, the slave’s data is newer than the master’s, and the master needs some time to perform updates. When the master’s version approaches that of the slave, the controller server sends a write-stop command to the slave. After waiting for the master and slave states to be completely consistent, it initiates a master-slave switchover operation, restores the master to active status in the routing table, and notifies the proxy server to switch write operations back to the master. Once all steps are complete, the slave is modified back to a writable state. From the process described above, it can be seen that configuring the master-slave replication relationship as a Dual Master structure simplifies the execution of masterthe switching steps.
In the process described above, bringing the crashed primary database back online will cause a brief period of write unavailability noticeable to users. To mitigate this, the proxy server can mask the issue by catching errors and using delayed retry methods.
Read/Write Splitting
We have also implemented read/write splitting that is transparent to users. When this feature is enabled, the proxy server parses incoming SQL statements, directing write operations to the primary database and distributing read operations across both the primary and replica databases in a load-balanced manner. To prevent scenarios where a user writes data to the primary and then reads from a replica before the data is synchronized, resulting in missing or stale data, we add a timer after every write operation. For 300 milliseconds following a write, all read requests from that user are forcibly routed to the primary database. With multi-threaded replication technology, 300 milliseconds is generally sufficient to guarantee data synchronization from the primary to the replica, and this value is configurable.
The proxy server also needs to parse MySQL connection-related attributes, such as the default database set by the user via connection parameters or the “use database” statement, as well as session variables set via the SET statement. These parameters are configured on the connections to both the primary and replica databases and recorded in an in-memory table. When a new connection to a backend database is established or a disconnected one is re-established, these environment parameters are reapplied, ensuring the user perceives no difference.
Database and Table Sharding
We have also implemented database and table sharding (horizontal partitioning) that is transparent to users. When creating a user account, you must specify the multi-instance type and set the number of instances, which will create multiple groups of MySQL instances. When users create tables, they need to specify the sharding rules. These rules must define the partition key, how the partition key maps to sharded tables, and how those sharded tables map to multiple instances. These rules can be passed in by adding SQL comments before the CREATE TABLE statement.
First, the proxy server performs syntactic analysis on the incoming SQL statement to extract the information needed for rewriting and distributing the SQL. This includes the name of the table being operated on, the value corresponding to the partition key in each record of an INSERT statement (this value is mandatory), conditions in the WHERE clause of a query, fields in ORDER BY and GROUP BY statements, and the result limit in LIMIT statements. Currently, supported SQL is limited to the basic forms of the four DML statements: INSERT, SELECT, UPDATE, and DELETE. Table joins and nested SELECT queries are not yet supported, and ORDER BY and GROUP BY are limited to a single field. These areas require further development effort to implement and refine.
The next step is to rewrite the SQL statement into sub-statements for execution on each sharded table. This mainly involves table name replacement and WHERE condition rewriting. The sub-statements are then sent concurrently to the corresponding sharded tables for execution.
Finally, the execution results returned from each sub-table are received and merged. To prevent excessively large query result sets from overwhelming the proxy server’s memory, or to reduce communication overhead when the user only needs a portion of the results, we have optimized the process of receiving and merging query results. By setting a buffer size, we can limit the number of result rows returned by a MySQL instance at a time. Once partial results are returned from all sharded tables, a merge sort begins, and the sorted results are returned to the user. When the results from a particular shard are exhausted, the socket is read again to fill the buffer and fetch the next batch of results. The entire process is quite similar to how a search engine distributes queries to retrieval servers and then merges the results.
Figure 2: Implementation Layers of the Proxy Server
To enhance performance, SQL parsing, rewriting, and merging result sets returned from multiple MySQL servers are all implemented in C++. These components are invoked by the state machine written in Erlang through the NIF interface.
Resource Management
We referenced resource management methods from cloud computing systems like VMware DRS and implemented a resource pool mechanism to manage computing resources such as CPU, memory, and disk on database servers. Administrators first divide multiple resource pools based on factors like the server model and data center location across the entire cluster. After the agent process on a server starts, it registers with the controller node, and administrators then add each server to the appropriate resource pool through the web management interface.
The unit of instance allocation is the resource pool. Administrators can specify the resource pools for the primary and standby databases based on factors such as which data centers the application is deployed in and the required computing resources. The instance management service then selects a server with a lighter load from the resource pool to create the instance. In the future, we will also develop scheduling management within the resource pool. If the load of one server in a resource pool is consistently and significantly higher than others, the scheduling process will migrate its MySQL instances to lower-load machines.
In addition to dividing servers into resource pools, we also use Cgroups to further refine resources within each server for easier management and isolation. For example, on a server with 16 cores and 48GB of memory, we divide its resources into 16 process groups, which is equivalent to allocating one CPU core and 2GB of memory to each process group. This allows 8 MySQL processes with a 256MB memory specification to be placed in one process group, while a MySQL process requiring 4GB of memory can be achieved by merging two process groups. Cgroups can limit the upper bound of resources used by each process group and ensure isolation between them. One thing to note is that this resource management approach can cause fragmentation. For instance, if a 256MB MySQL process is allocated to each of the 16 process groups, only 4GB of total memory is used, leaving 44GB free. However, at this point, it would be impossible to allocate a MySQL process requiring 4GB of memory. This problem can be solved using the Buddy System.
Resource Scheduling
The system currently supports three user tiers:
The first type is users with relatively small data volumes and traffic, such as blog sites, small applications, and applications under development. Multiple small users can share a single MySQL instance, with one database per user. A single machine can support hundreds to thousands of small users, but an excessive number of files can negatively impact system performance.
The second type is medium-scale users, where each user exclusively occupies one MySQL instance. The memory footprint per instance ranges from 256MB to 32GB. Users’ memory and disk space can also be adjusted. When the current machine cannot meet a user’s resource requirements, the instance can be migrated to a server with available resources or higher specifications.
Figure 3 Resource Scheduling through Instance Migration
The third type is users who require sharding, where a user can own multiple independent MySQL instances. These instances can coexist with other instances on the same physical machine, or each instance can exclusively occupy a physical machine as the business data volume grows.
A user’s specification can be designated at creation time, or upgraded/downgraded via the migration tool. We use the YuGong system developed by the group’s middleware team, a tool that combines full data replication with bin log analysis for incremental replication, enabling dynamic expansion, contraction, and migration without downtime. Currently, user specification upgrades and downgrades need to be triggered manually on the console. In the future, we hope to enable automated scheduling based on statistical analysis of a user’s database usage over a past period.
Resource Isolation
Resource isolation becomes especially critical when multiple users share a single MySQL instance, or multiple MySQL instances share the same physical machine. For example, if a user executes an SQL statement with a heavy IO load—such as a full table scan on a large table without an index on the relevant field—it can severely impact other users’ experience. Currently, we use a combination of methods for resource isolation: limiting MySQL process resources with Cgroup on the database server, and restricting QPS on the proxy server side.
The first method involves creating process groups and using Cgroup’s cpuset, memcg, and blkio subsystems to respectively limit the maximum CPU usage, memory, and IOPS available to a user’s MySQL process. This method suits scenarios where multiple MySQL instances share a single physical machine.
The second method uses an agent deployed on the database side to analyze the MySQL process’s slow query log, collecting and summarizing the cost of recently executed SQL statements by the user, and periodically feeding this information back to the controller server. The controller server compares the data against the user’s quota. If it significantly exceeds the limit, it notifies the proxy side to restrict the user’s QPS by adding latency, thereby reducing the system resources consumed by that user. This method is better suited for scenarios where multiple users share a single MySQL instance, since Cgroup cannot be used for inter-process restrictions in that context.
Data Security
Both users and corporate security departments are highly concerned about data security. We have implemented multiple methods to ensure the safety of user data:
- SSL connection support: the proxy server implements the full MySQL client/server protocol and can establish encrypted connections with clients.
- IP whitelisting to set the list of addresses permitted to access the database. Users can configure the whitelist to only include their application server addresses, enhancing account security.
- The proxy server records all user database operations to a log analysis server, allowing the security department to periodically export log files for scanning and vulnerability checks.
- The proxy server can intercept various types of SQL statements according to the security department’s requirements, such as full-table `select *` statements or queries where the result row count exceeds a set limit.
In a later phase, we also plan to retain the MySQL instance’s bin log and slow query log. This way, if a user accidentally deletes data without a backup, they can recover it using bin log tools. A slow query log analysis tool will run periodically in the background to analyze aspects such as index usage and the number of IO operations during SQL execution, guiding users to improve their SQL statements.
Conclusion
In engineering practice, we adhere to the principle of not reinventing the wheel, making full use of open-source, mature technologies and tools. For instance, we implemented a high-performance proxy server on top of Erlang’s network programming framework, built a message middleware based on RabbitMQ, used ZooKeeper to manage server heartbeats, and fully leveraged the group’s mature solutions for data backup, migration, scaling up/down, and other bin log tools. This principle allows us to focus our limited resources on reducing costs and improving user experience.