UMP (Unified MySQL Platform) is a low-cost, high-performance MySQL cloud data solution developed by the core systems 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 depends on open-source components like Mnesia, LVS, RabbitMQ, and ZooKeeper.
In the article “Exploring the Architecture of Low-Cost, High-Performance MySQL Cloud Databases,” we introduced the UMP system structure and the functions of its components. In this article, we will further explore the applications of RabbitMQ and ZooKeeper in the system, the implementation of the proxy server, how the entire system achieves disaster recovery, read/write splitting, sharding, and other features, and introduce resource management, isolation, and scheduling technologies, as well as practices for ensuring 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 such as SQL queries and logs, which still go directly over TCP) is conducted through RabbitMQ, serving as the messaging middleware to ensure the reliability of message delivery.
During cluster initialization, a queue is created in RabbitMQ for each node in the cluster, acting as the node’s “mailbox.” When a node sends a message, regardless of whether the recipient is online, it 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, deleting the message from the “mailbox.” RPC can be implemented based on RabbitMQ; besides 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 deleting the Request message and writing the Reply message are completed in a single atomic operation.
Figure 1 RPC between nodes implemented 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 and processed by the receiver, but unfortunately, it cannot guarantee that a message is sent/processed only once. The primary reason is that RabbitMQ does not support XA. First, the sender cannot complete writing a 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 after restarting and can only attempt a resend. Similarly, the message receiver cannot complete processing the message and deleting it from MQ within the same transaction. If the receiver crashes after processing the message but before deleting it from MQ, it will still receive and process this message again after restarting.
Therefore, the message receiver must ensure idempotent processing, meaning that processing the same message multiple times will have no side effects. For example, when the controller sends a backup command to the agent, it can include the timestamp of the last backup. The agent checks if the timestamp matches before executing the backup operation, ensuring that multiple sends of the same backup command do not create multiple backups.
By using RabbitMQ’s routing feature (Exchange), message broadcasting can also be implemented. For instance, an Exchange called proxy is created in the system, with its type configured as ’fanout’. When a new proxy server registers, the node’s “mailbox” binds to this Exchange. Thus, when the controller server needs to send a notification to all proxy servers, such as executing a master-slave switchover operation, the message sent to the Exchange is written to the “mailboxes” of all proxy servers.
RabbitMQ also implements a mirrored queue algorithm to provide HA. When creating a queue, you can set it as a mirrored queue by passing the “x-ha-policy” parameter. Mirrored queues are stored on multiple RabbitMQ nodes and configured in a one-master, multiple-slave structure. The “x-ha-policy-params” parameter can specify the lists of master and slave nodes. All operations sent to the mirrored queue, such as message sending and deletion, are first executed on the master node and then synchronized to all slave nodes through 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 slaves either all execute successfully or all fail. Through 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 ultimately returns to the master, it ensures that the load difference between the master and slave nodes is minimal.
ZooKeeper
ZooKeeper provides distributed locks, naming services, and more in distributed clusters. It likens a distributed cluster to a zoo and itself to the zookeeper. Originally developed by Yahoo! to play the role of Google Chubby in the Hadoop software stack, we use ZooKeeper standalone in our project to implement three functions:
1. As a global configuration server. Configuration files were originally stored locally, requiring modifications on all nodes for any change, which was repetitive and error-prone. After placing them on ZooKeeper, all nodes monitor changes to the configuration files. Once a 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 controllers cannot perform the same operation simultaneously. For example, if a MySQL instance goes down and all controller servers track the event and initiate a master-slave switchover, the proxy and agent servers would receive multiple switchover commands, causing chaos in the cluster. Therefore, for simplicity, we stipulate that at any given time, only one leader is elected among the multiple controller servers in the cluster, and this leader is responsible for initiating various system tasks. Leader election is implemented using ZooKeeper’s distributed lock functionality.
3. Monitoring all MySQL instances. We developed a ZooKeeper client plugin for MySQL servers. Upon startup, it connects to the ZooKeeper server and creates an ephemeral node. If the MySQL process dies, after a 5-second timeout, this ephemeral node is deleted and detected by the monitoring daemon running in the background. If the dead MySQL process was the master, a master-slave switchover is triggered; if it was a slave, the slave’s read weight is set to 0.
Disaster Recovery
When a MySQL server fails, the system performs a failure recovery process that is transparent to the user. Users are unaware of master database downtime and online events; the proxy server hides these events from users, 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 both MySQL instances set each other as their master, read data updates from the other, and replicate them locally. This way, writing data to either MySQL instance will update the other. A problem with the Dual Master structure is the potential for conflicts if both MySQL instances modify the same row of data simultaneously, resulting in inconsistent data versions in the two instances. Therefore, to ensure data consistency, "single write" must be maintained, meaning data is only written to the master, which 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. After detecting this event, the controller server initiates a master-slave switchover, marks the master as unavailable in the routing table, and notifies all proxy servers to perform the switchover via RabbitMQ.
When the crashed master comes back online, the strategy is slightly more complex. At this point, the data on the slave is newer than that on the master, and the master needs time to catch up on updates. When the master’s version is close to the slave’s, the controller server sends a stop-write command to the slave, waits for the master and slave states to be fully consistent, then initiates a master-slave switchover, restores the master to active status in the routing table, and notifies proxy servers to switch write operations back to the master. Once complete, the slave is changed back to a writable state. As seen from this process, configuring the master-slave replication as a Dual Master structure simplifies the steps for executing a master-slave switchover.
During the above process, bringing the crashed master back online may cause a brief period where the user cannot write. Further, the proxy server can mask this problem by catching errors and implementing delayed retries.
Read/Write Splitting
We have also implemented user-transparent read/write splitting. When this feature is enabled, the proxy server parses the incoming SQL statements from the user, sending write operations to the master and distributing read operations to both the master and slave with load balancing. To avoid a situation where a user writes data to the master and then reads from the slave before the data is synchronized, thereby getting no data or an old version, a timer is added after every write operation. For 300 milliseconds after a user writes, any read data is forcibly distributed to the master. With master-slave multi-threaded replication technology, 300 milliseconds typically ensures data synchronization from master to slave, and this value is also configurable.
The proxy server also needs to parse MySQL connection-related properties, such as the default database set by the user through connection parameters or the “use database” statement, and session variables set via set statements. These parameters are set on the connections to both the master and slave and recorded in an in-memory table. When establishing a new connection to the backend database or reconnecting after a disconnection, these environment parameters are reset to avoid any user-perceptible differences.
Sharding
We have also implemented user-transparent sharding (horizontal partitioning). When creating a user account, you must specify the type as multi-instance and set the number of instances, which will create multiple groups of MySQL instances. When creating a table, the user needs to specify the sharding rules, which must define the partition key, how the partition key maps to the shard, and how the shard maps to multiple instances. These rules can be passed in by adding SQL comments before the table creation statement.
First, the proxy server parses the incoming SQL statement to extract the necessary information for rewriting and distributing the SQL, such as the table name, the value corresponding to the partition key in each record of an insert statement (which must be included), the conditions in the where clause of a query, the fields in order by and group by statements, and the limit on the number of results in a limit statement. Currently, the 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 single fields. These areas still require further development and improvement.
The next step is to rewrite the SQL statement into sub-statements for execution on each shard, primarily involving table name replacement and where clause rewriting, then sending the sub-statements concurrently to the corresponding shards for execution.
Finally, the results returned from each shard are received and merged. To prevent the query result set from being too large and 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, the number of rows returned by each MySQL instance at a time can be limited. Once partial results from all shards are returned, a merge sort begins, and the sorted results are returned to the user. When the results from a certain shard are exhausted, the socket is read again to fill the buffer for the next batch. The entire process is similar to distributing queries to search servers and then merging the results in a search engine.
Figure 2 Implementation Layers of the Proxy Server
To improve performance, SQL parsing, rewriting, and merging result sets from multiple MySQL servers are implemented in C++ and invoked by the Erlang state machine via the NIF interface.
Resource Management
We referenced resource management methods from cloud computing systems like VMware DRS to implement a resource pool mechanism for managing computing resources such as CPU, memory, and disk on database servers. Administrators first divide all servers in the cluster into multiple resource pools based on factors like server model and data center location. After the agent process on a server starts and registers with the controller node, the administrator adds each server to the appropriate resource pool through the web management interface.
The unit for allocating instances is the resource pool. Administrators can specify the resource pools for the master and slave based on factors like the data centers where applications are deployed and the required computing resources. The instance management service then selects servers with lighter loads from the resource pool to create instances. In the future, we will develop scheduling management within resource pools. If a server within a resource pool has a consistently significantly higher load than others over a period, the scheduling process will migrate its MySQL instances to lower-load machines.
In addition to dividing servers into resource pools, we also utilize Cgroup within each server to further refine resources for easier management and isolation. For example, a 16-core, 48G server will have its resources divided into 16 process groups, each allocated one CPU core and 2G of memory. Thus, one process group can accommodate eight MySQL processes with a 256M memory specification, while a MySQL process requiring 4G memory can be implemented by combining two process groups. Cgroup can limit the upper resource usage for each process group and ensure isolation between groups. One point to note is that this resource management approach can lead to fragmentation. For instance, if a 256M MySQL process is allocated to each of the 16 process groups, the total memory used is only 4G, with 44G idle. However, it is impossible to allocate a new 4G MySQL process in this state. This problem can be solved by the Buddy System.
Resource Scheduling
The system currently supports three specifications of users:
The first type is users with relatively small data volumes and traffic, such as blog sites, small applications, and those in development. Multiple small users can share a single MySQL instance, each with their own database. A single machine can support hundreds to thousands of small users, but an excessive number of files can adversely affect system performance.
The second type is medium-scale users, each occupying a dedicated MySQL instance with memory ranging from 256M to 32G per instance. The user’s memory and disk space are also adjustable. If the current machine cannot meet the user’s resource requirements, the instance can be migrated to a server with spare resources or a higher configuration.
Figure 3 Resource Scheduling Through Instance Migration
The third type requires sharding, where users can possess multiple independent MySQL instances. These instances can coexist with other instances on a single physical machine, or, as the business data volume grows, each instance can occupy a dedicated physical machine.
User specifications can be designated at creation time or upgraded/downgraded via migration tools. We use the YuGong system developed by the group’s middleware team, a tool that combines full replication with binlog analysis for incremental replication, enabling dynamic expansion, contraction, and migration without downtime. Currently, upgrading and downgrading user specifications must be triggered on the console. In the future, we hope to automate scheduling based on statistical information of the user’s database usage over a past period.
Resource Isolation
Resource isolation becomes especially critical when multiple users share a single MySQL instance, or when multiple MySQL instances share the same physical machine. For instance, a user executing an IO-heavy SQL statement — such as a full table scan on a large table due to missing indexes — can severely degrade the experience for other users. Currently, we combine Cgroup-based MySQL process resource limits on the database server with QPS throttling on the proxy server to achieve resource isolation.
The first method creates process groups and uses the cpuset, memcg, and blkio subsystems of Cgroup to cap the maximum CPU usage, memory, and IOPS available to a user’s MySQL processes. This approach is suitable when multiple MySQL instances share a single physical machine.
The second method deploys an agent on the database server to analyze the MySQL slow query log, collecting and aggregating the cost of recently executed SQL statements for each user. This information is periodically reported to a controller server, which compares the data against the user’s quota. If the usage significantly exceeds the quota, the controller instructs the proxy to throttle the user’s QPS by injecting additional latency, thereby reducing the system resources consumed by that user. This method is more suitable when multiple users share the same MySQL instance, since Cgroup cannot enforce per-process restrictions in that scenario.
Data Security
Both users and enterprise security teams are deeply concerned about data security. We have implemented multiple measures to safeguard user data:
- SSL connections are supported. The proxy server implements the full MySQL client/server protocol and can establish encrypted connections with clients.
- An IP whitelist specifies which addresses are allowed to access the database. Users can configure the whitelist to include only their application server addresses, enhancing account security.
- The proxy server logs every database operation to a log analysis server. The security team can periodically export these logs to scan for security vulnerabilities.
- The proxy server can intercept various types of SQL statements as required by the security team, such as full-table
SELECT *queries or statements whose result sets exceed a specified row limit.
In future phases, we will also retain the MySQL instance’s bin log and slow query log. If a user accidentally deletes data and has no backup, they can recover it using bin log tools. Additionally, a slow query log analysis tool will run periodically in the background to examine index usage, the number of IO operations, and other metrics during SQL execution, providing guidance to users on how to optimize their SQL statements.
Closing Remarks
In our engineering practice, we adhere to the principle of not reinventing the wheel, making full use of open-source, mature technologies and tools. For example, we built a high-performance proxy server on top of the Erlang network programming framework, used RabbitMQ for message queuing, and leveraged ZooKeeper for server heartbeat management. We also took full advantage of our organization’s battle-tested solutions for data backup, migration, scaling up/down, and other bin log utilities. This principle allows us to focus our limited resources on reducing costs and improving the user experience.