# Wormhole ## Wormhole introduction Wormhole is a data synchronization tool with which users can implement full and incremental data synchronization via the Web control panel. ## Wormhole features Wormhole has the following features: - A SQL-based real-time pipeline framework - Support data migration between homogeneous and disparate data sources, for example, MySQL -> TiDB, MongoDB -> TiDB - Support full and incremental data synchronization - Provide nice and accessible Web management interface - High availability and auto failover - A monitor alert mechanism - Support sharding aggregation - Deploy Kubernetes using a single line of command ## Wormhole architecture As shown in the following diagram, Wormhole has three components: - Portal Synchronize the upstream data, parse the data, and write the data into Kafka - Stargate Obtain the data from Kafka, construct SQL statements, and write these statements into TiDB - Ether Manage the task scheduler of Portal and Stargate, provide the management interface, and use Kafka for persistent storage of intermediate data ![Wormhole Architecture](../media/wormhole-architecture.jpg) ## Wormhole synchronization process ![Wormhole Synchronization Process](../media/wormhole-synchronization-process.jpg) ## Wormhole usage limits ### MySQL version limit Currently Wormhole only supports MySQL v5.6 and later, MariaDB v5.5 and later and Alibaba Cloud RDS for MySQL. ### Statement limits In incremental synchronization, currently Wormhole only supports synchronization of 9 kinds of DDL statements: - create database - drop database - create table - drop table - alter table - truncate table - rename table - create index - drop index The following statements will be skipped by default: - savepoint - flush - optimize table - analyze table - repair table - create trigger - drop trigger - create procedure - alter procedure - drop procedure - create view - alter view - drop view - create function - alter function - drop function - grant - revoke - create user - alter user - rename user - drop user > **Note:** > - Incremental synchronization does not support changing the primary key. > - Other limits exit due to the TiDB limit. See [Data Definition Statements of TiDB](https://pingcap.com/docs/sql/ddl/). ### Feature limits Wormhole has the following feature limits: - For tables without the primary key, data may be processed repeatedly. - For tables without keys of the number type, too much memory space may be occupied and synchronization may fail if the full data size is large. - For the file system which is not case sensitive, Wormhole does not support uppercase characters in the synchronization name of databases and tables. ## Pre-check Wormhole provides the pre-check feature. It checks whether the configuration of MySQL, TiDB and Kafka meets the requirements and sends the check result to the user. Before synchronizing, you should make sure the following items meet the requirements. ### Check data source (MySQL) - Check the source database connection Check whether Wormhole connects to the source database successfully, including checking the network, account and password. - Check the source database version Check the MySQL version. Currently Wormhole only supports MySQL v5.5, v5.6 and v5.7 and MariaDB v5.5 and later. - Check the source database privilege Check whether the account provided by the source database has the privileges including Select, Reload (optional)，Replication Slave and Replication Client. For example, the `repl` account is granted with the privileges: ``` mysql> grant select,reload,replication slave,replication client on *.* to 'repl'@'%' identified by '123456'; Query OK, 0 rows affected, 1 warning (0.03 sec) ``` Check the privileges which the`repl` account already has: ``` mysql> show grants for 'repl'; +----------------------------------------------------------------------------------------------------------------+ | Grants for repl@% | +----------------------------------------------------------------------------------------------------------------+ | GRANT SELECT, RELOAD, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'repl'@'%' | +-----------------------------------------------------------------------------------------------------------------+ 1 row in set (0.01 sec) ``` - Check `server-id` of the source database Run the following command to check `server-id`: ``` mysql> show global variables like 'server_id'; +---------------+-------+ | Variable_name | Value | +---------------+-------+ | server_id | 1 | +---------------+-------+ 1 row in set (0.01 sec) ``` If the result is null or 0, set `server-id` to an integer which is not smaller than 1. You can run the following command to set `server-id`: ``` mysql> set global server_id=2; Query OK, 0 rows affected (0.00 sec) ``` - Check binlog status of the source database Check whether the binlog is enabled in the source database You can run the following command to check whether the binlog is enabled: ``` mysql> show global variables like 'log_bin'; +--------------------+---------+ | Variable_name | Value | +--------------------+---------+ | log_bin | ON | +--------------------+---------+ 1 row in set (0.00 sec) ``` If the result is `log_bin = OFF`, you need to enable the binlog and restart the MySQL service. For the enabling operation, see [Setting the Replication Master Configuration](https://dev.mysql.com/doc/refman/5.7/en/replication-howto-masterbaseconfig.html). - Check the source database binlog format Check whether the source database binlog format is `ROW`. You can run the following command to check the binlog format: ``` mysql> show global variables like 'binlog_format'; +--------------------+----------+ | Variable_name | Value | +--------------------+----------+ | binlog_format | ROW | +--------------------+----------+ 1 row in set (0.00 sec) ``` If the result shows the binlog format is not `ROW`, set it to `ROW`. You can run the following command to set it to `ROW`: ``` mysql> set global binlog_format=ROW; Query OK, 0 rows affected (0.01 sec) ``` After you modify the binlog format, it is recommended to kill the old connection and reconnect or restart the MySQL service. - Check the source database `binlog_row_image` When MySQL is v5.6.2 or later and MariaDB is v10.1.6 or later, Wormhole checks whether `binlog_row_image` is `FULL`. See the following references: - https://dev.mysql.com/doc/refman/5.6/en/replication-options-binary-log.html#sysvar_binlog_row_image - https://mariadb.com/kb/en/library/replication-and-binary-log-server-system-variables/#binlog_row_image You can use the following command to check `binlog_row_image`: ``` mysql> show global variables like 'binlog_row_image'; +--------------------------+---------+ | Variable_name | Value | +--------------------------+---------+ | binlog_row_image | FULL | +--------------------------+----------+ 1 row in set (0.01 sec) ``` If the check result is not `FULL`, run the following command to set it to `FULL`. ``` mysql> set global binlog_row_image = FULL; Query OK, 0 rows affected (0.01 sec) ``` - Check the unique key Check whether the table schema has the primary key or unique key. If no primary key or unique key exists in the table schema without keys of the number type, the system shows you the problem which may occur. You can deal with or just skip the problem. ### Check the target database (TiDB) - Check the target database connection Check whether Wormhole connects to the target database successfully, including checking the network, account and password. - Check the target database privilege The TiDB synchronization account should be granted with the Read and Write privileges. ### Check external dependencies - Check the Kafka connection Check whether Wormhole connects to Kafka successfully. - Check the ZooKeeper connection Check whether Wormhole connects to ZooKeeper successfully. ## Deploy ### Deploy using Docker Compose You can deploy Wormhole using Docker Compose only in a test environment. 1. Install Docker You can install Docker on Linux/Mac OS/Windows. For the installation procedure, see [official Docker documentation](https://www.docker.com/community-edition). You just need a `docker-compose.yml` file to fast install Kafka, ZooKeeper and Wormhole components. ``` version: '2' services: zookeeper: image: wurstmeister/zookeeper ports: - "2181:2181" kafka: image: wurstmeister/kafka ports: - "9092:9092" environment: # YOU MUST CHANGE THE IP TO YOURS. KAFKA_ADVERTISED_HOST_NAME: 192.168.60.54 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 # CHANGE IT TO YOURS. KAFKA_NUM_PARTITIONS: 10 KAFKA_MESSAGE_MAX_BYTES: 10485760 KAFKA_REPLICA_FETCH_MAX_BYTES: 10485760 volumes: - /var/run/docker.sock:/var/run/docker.sock ether: image: uhub.ucloud.cn/pingcap/wormhole ports: - "12379:12379" - "12380:12380" entrypoint: /ether command: - --name=ether - --client-urls=http://0.0.0.0:12379 - --peer-urls=http://0.0.0.0:12380 - --advertise-client-urls=http://ether:12379 - --advertise-peer-urls=http://ether:12380 - -L=info portal: image: uhub.ucloud.cn/pingcap/wormhole ports: - "10086:10086" entrypoint: /portal command: - --ether-urls=ether:12379 - --zookeeper-urls=zookeeper:2181 - -L=info depends_on: - "zookeeper" - "kafka" - "ether" stargate: image: uhub.ucloud.cn/pingcap/wormhole ports: - "10087:10087" entrypoint: /stargate command: - --ether-urls=ether:12379 - --zookeeper-urls=zookeeper:2181 - -L=info depends_on: - "zookeeper" - "kafka" - "ether" ``` > **Note:** You need to change the IP address of `KAFKA_ADVERTISED_HOST_NAME` to the IP address of the machine deployed with `docker-compose`. Do not set it to `127.0.0.1` or `localhost`. 2. Start your Docker Compose. ``` docker-compose up ``` > **Note:** If you want to run Docker Compose in the background, use `docker-compose up -d` to start. 3. Access http://your_ip:12379/ether/web to start your operation. Change `your_ip` to your IP address. 4. Destroy Docker Compose. ``` docker-compose down ``` For more details about Docker Compose, see [official Docker documentation](https://www.docker.com/community-edition). ### Deploy Wormhole on multiple machines using Docker This section introduces how to deploy a Wormhole cluster on multiple machines using Docker. 1. Install Docker You can install Docker on Linux/Mac OS/Windows. For the installation procedure, see [official Docker documentation](https://www.docker.com/community-edition). 2. Pull the Docker image of Wormhole Currently the three components of Wormhole are packed in one image. You can get the latest image from `docker pull uhub.ucloud.cn/pingcap/wormhole:latest`. Assume that all the following operations are performed on one host `192.168.1.101`, Kafka is `192.168.1.101:9092`, and ZooKeeper is `192.168.1.101:2181`. Ether is started before Portal and Stargate. 1. Start Ether ``` docker run -d --name ether \ -p 12379:12379 \ -p 12380:12380 \ -v /data:/data \ uhub.ucloud.cn/pingcap/wormhole /ether \ --data-dir=/data/ether \ --peer-urls http://0.0.0.0:12380 \ -client-urls http://0.0.0.0:12379 \ --advertise-client-urls="http://192.168.1.101:12379" ``` 2. Start Portal ``` docker run -d --name portal \ -p 10086:10086 \ uhub.ucloud.cn/pingcap/wormhole /portal \ -ether-urls 192.168.1.101:12379 \ -zookeeper-urls 192.168.1.101:2181 \ --status-port 10086 \ -L info ``` 3. Start Stargate ``` docker run -d --name stargate \ -p 10087:10087 \ uhub.ucloud.cn/pingcap/wormhole /stargate \ -ether-urls 192.168.1.101:12379 \ -zookeeper-urls 192.168.1.101:2181 \ --status-port 10087 \ -L info ``` ### Deploy using Binary #### Configure the environment 1. Set the environment - Kafka: It is recommended to use `kafka_2.12-0.10.2.1` or later (Wormhole supports v0.9 and later, not v0.8) - ZooKeeper: It is recommended to use `zookeeper-3.4.8` or later - When the number of partitions is large, the number of file descriptors reaches the usage limit (too many open files). You can use `ulimit -n` to check the number of file descriptors and use `ulimit -n ` to set the size of file descriptors. On Linux, you can use `cat /proc/sys/fs/file-max` to check the size limit for file descriptors. 2. Install JDK Install JDK based on the system environment. It is recommended to install JDK 8. See [JDK installation documentation](https://docs.oracle.com/javase/8/docs/technotes/guides/install/toc.html). 3. Install the ZooKeeper cluster See [ZooKeeper deployment documentation](http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#ch_deployment). 4. Install the Kafka cluster See [Kafka deployment documentation](https://kafka.apache.org/documentation/#quickstart). 5. Install the TiDB cluster See [Ansible Deployment](https://pingcap.com/docs/op-guide/ansible-deployment/). 6. Obtain the installation package of Wormhole You can only obtain the Wormhole installation package from the Wormhole developer at PingCAP 7. Start Wormhole 1. Start MySQL and TiDB. 2. Start the following applications or components in this order: ZooKeeper --> Kafka --> Ether --> Portal --> Stargate. #### Single node deployment Assume that Kafka is in the starting process and its access address is `192.168.1.101:9092` and ZooKeeper has been started and its access address is `192.168.1.101:2181`. 1. Start Ether ``` ./bin/ether -L info -peer-urls http://192.168.1.101:12380 -client-urls http://192.168.1.101:12379 ``` 2. Start Portal ``` ./bin/portal -L info -ether-urls http://192.168.1.101:12379 -zookeeper-urls 192.168.1.101:2181 -status-port 10086 ``` 3. Start Stargate ``` ./bin/stargate -L info -ether-urls http://192.168.1.101:12379 -zookeeper-urls 192.168.1.101:2181 -status-port 10087 ``` #### Multi-node deployment 1. Configure the machines (the minimum cluster topology) based on the following information: | name | host-ip | services | | ---- | ------- | -------- | | node1 | 192.168.1.101 | ZooKeeper, Kafka, Ether, Portal | | node2 | 192.168.1.102 | ZooKeeper, Kafka, Ether, Stargate | | node3 | 192.168.1.103 | ZooKeeper, Kafka, Ether | 2. Start Ether ``` ./bin/ether --name=ether1 \ --data-dir=/data/ether1 \ --client-urls="http://192.168.1.101:12379" \ --peer-urls="http://192.168.1.101:12380"\ --initial-cluster="ether1=http://192.168.1.101:12380,ether2=http://192.168.1.102:12380,ether3=http://192.168.1.103:12380" \ -L "info" ./bin/ether --name=ether2 \ --data-dir=/data/ether2 \ --client-urls="http://192.168.1.102:12379" \ --peer-urls="http://192.168.1.102:12380" \ --initial-cluster="ether1=http://192.168.1.101:12380,ether2=http://192.168.1.102:12380,ether3=http://192.168.1.103:12380" \ -L "info" ./bin/ether --name=ether3 \ --data-dir=/data/ether3 \ --client-urls="http://192.168.1.103:12379" \ --peer-urls="http://192.168.1.103:12380" \ --initial-cluster="ether1=http://192.168.1.101:12380,ether2=http://192.168.1.102:12380,ether3=http://192.168.1.103:12380" \ -L "info" ``` 3. Start Portal ``` ./bin/portal \ --ether-urls="http://192.168.1.101:12379,http://192.168.1.102:12379,http://192.168.1.103:12379" \ --zookeeper-urls="192.168.1.101:2181,192.168.1.102:2181,192.168.1.103:2181" \ --status-port 10010 ``` 4. Start Stargate ``` ./bin/stargate \ --ether-urls="http://192.168.1.101:12379,http://192.168.1.102:12379,http://192.168.1.103:12379" \ --zookeeper-urls="192.168.1.101:2181,192.168.1.102:2181,192.168.1.103:2181" \ --status-port 10089 ``` #### Add a node in the cluster 1. Add Ether ``` ./bin/ether \ --name=ether2 \ --data-dir=/data/ether1 \ --client-urls="http://192.168.1.102:12379" \ --peer-urls="http://192.168.1.102:12380" \ --join=http://192.168.1.101:12379 \ -L=info ``` 2. Add Stargate and Portal Add Stargate: ``` ./bin/stargate -L info -ether-urls http://192.168.1.101:12379 -zookeeper-urls 192.168.1.101:2181 -status-port 10087 ``` Add Portal: ``` ./bin/portal -L info -ether-urls http://192.168.1.101:12379 -zookeeper-urls 192.168.1.101:2181 -status-port 10086 ``` #### Note - If multiple error messages of port usage occur, these messages can be differentiated by `status-port`. - If you use nohup to start the cluster in the production environment, you should execute the starting command in a script file. Otherwise, the nohup starting process may receive abnormal exit message due to the exit of Shell. #### Argument description - Ether Usage of ether: -L string log level: debug, info, warn, error, fatal (default "info") -V print version information and exit -advertise-client-urls string advertise url for client traffic (default '${client-urls}') -advertise-peer-urls string advertise url for peer traffic (default '${peer-urls}') -client-urls string url for client traffic (default "http://127.0.0.1:2379") -config string Config file -data-dir string path to the data directory (default 'default.${name}') -initial-cluster string initial cluster configuration for bootstrapping, e,g. ether=http://127.0.0.1:2380 -join string join to an existing cluster (usage: cluster's '${advertise-client-urls}' -log-file string log file path -log-rotate string log file rotate type, hour/day -name string human-readable name for this pd member (default "ether") -peer-urls string url for peer traffic (default "http://127.0.0.1:2380") -version print version information and exit - Portal Usage of portal: -L string log level: debug, info, warn, error, fatal (default "info") -V print version information and exit -chunk-size unit rows per chunk (default 100) -config string Config file -contract-lease uint contract lease for renew job (default 300) -ether-urls string ether client urls, separated by comma (default "http://127.0.0.1:2379") -heartbeat-lease uint heartbeat lease for contact ether-client (default 15) -log-file string log file path -log-rotate string log file rotate type, hour/day (default "day") -mysql-serverid uint mysql-serverid for mysql slave, must be unique. (default 1001) -routine-cnt int routine cnt for dump (default 4) -status-port uint port for program status, like pprof (default 10086) -version print version information and exit -workerid-file string portal worker id file (default ".portal_workerid") -zookeeper-urls string REQUIRED: zookeeper server urls, separated by comma (default "127.0.0.1:2181") - Stargate Usage of stargate: -L string log level: debug, info, warn, error, fatal (default "info") -V print version information and exit -batch int batch numbers of sql to execute (default 10) -config string Config file -contract-lease uint contract lease for renew job (default 300) -ether-urls string ether client urls, separated by comma (default "http://127.0.0.1:2379") -heartbeat-lease uint heartbeat lease for contact ether-client (default 15) -log-file string log file path -log-rotate string log file rotate type, hour/day (default "day") -status-port uint port for program status, like pprof (default 10087) -version print version information and exit -workerid-file string stargate worker id file (default ".stargate_workerid") -zookeeper-urls string REQUIRED: zookeeper server urls, separated by comma (default "127.0.0.1:2181") - Kafka (v0.10 as the example) broker.id=1 port=9092 host.name=node1 massage.max.bytes=6525000 replica.fetch.max.bytes=1048576 \# the maximum size of the data obtained by replicas each time replica.fetch.wait.max.ms =500 \# the longest waiting time for communication between replicas and the leader. If the communication fails, it will retry num.network.threads=3 \# the maximum number of threads for broker processing message. The number of configuration threads = CPU cores + 1 num.io.threads=8 \# the number of threads for broker processing disk I/O. The number of configuration threads = CPU cores * 2. The maximum number of configuration threads should be <= CPU cores * 3 socket.send.buffer.bytes=102400 \# the buff area for socket sending message. SO_SNDBUFF socket.receive.buffer.bytes=102400 \# the buff area for socket receiving message. SO_RCVBUFF socket.request.max.bytes=104857600 \# the maximum number of socket requests. To prevent server OOM, `message.max.bytes` should be smaller than \# socket.request.max.bytes. It is covered by the specified arguments in creating the topic log.dirs=/data/kafka-logs \# kafka num.partitions=10 \# it is better to set the number smaller than 100 * b * r. "b" stands for the broker node number. "r" stands for the replicator auto.create.topics.enable =true \# whether it is allowed to create a topic automatically. If it is set to `false`, you need to create a topic using the command. **In this system，you should set it to `true`.** log.retention.hours=72 \# when massive messages are written into the Kafka server, many data files are generated and they will occupy large disk space \# if the data files are not removed in time, the disk space may be not enough. These files are retained for 7 days by default in Kafka log.segment.bytes=1073741824 \# setting the segment file to 1 GB speeds up reclaiming the disk space and the loading process of restarting Kafka \# if the file size is small, all the data files in the log.dir directory will be scanned in single-thread mode when Kafka is starting and there are many files log.cleanup.policy = delete \# log cleaning policies: performing the Delete and Compact operations on the obsolete data; \# when the number of the log files reaches the limit, the value of this argument will be covered by the specified argument in creating the topic log.retention.check.interval.ms=300000 \# check the file interval to confirm it meets the requirement of `log.cleanup.policy` log.cleaner.enable=false \# whether to enable log compression zookeeper.connect=node1:2181,node2:2181,node3:2181 \# the address of the ZooKeeper cluster. If there are multiple addresses, use a comma between addresses zookeeper.connection.timeout.ms=6000 \# the timeout of the ZooKeeper connection zookeeper.session.timeout.ms=6000 \# the longest timeout of the ZooKeeper connection, namely the interval between heartbeats. No heartbeat means the ZooKeeper fails \# do not set it to a too large value zookeeper.sync.time.ms =2000 delete.topic.enable=true \# whether to delete the topic. Set it to `true` in Wormhole - ZooKeeper (v3.4.10 as the example) \# The number of milliseconds of each tick tickTime=2000 \# The number of ticks that the initial \# synchronization phase can take initLimit=10 \# The number of ticks that can pass between \# sending a request and getting an acknowledgement syncLimit=5 \# the directory where the snapshot is stored. dataDir=/home/node1/kafka/zookeeper/data \# the port to which the clients will connect ## Synchronize data using Wormhole 1. Enter IP + port + /ether/web/ (for example, http://192.168.0.0:2379/ether/web/) to get into the Wormhole homepage: ![Wormhole Homepage](../media/wormhole-homepage.png) 2. Create the data source. 1. Click `CREATE` in the homepage to enter the following page. Currently, Wormhole supports synchronizing single data source and the sharding data. 2. Click `CREATE` to add the data source information. The fields marked in red are required fields. `Type` indicates the database type for synchronization (it supports MySQL currently); `Name` indicates the data source name: ![Create Data Source](../media/create-data-source.png) You can choose `Single Data Source` to synchronize the single data source; you can choose `Multiple Data Sources` to synchronize the sharding data; you can fill in the connection configuration of multiple databases. 3. After committing in the data source creation page, data source pre-check starts. The check result is as below: ![Pre-check Result](../media/precheck-result.png) If the pre-check fails, you should modify the data source configuration to pass the pre-check. 4. Choose tables for synchronization. Click the blue arrow in the upper right corner to move on. - If you chose `Single Data Source` in the last step, you can see the result shown in the following picture. And then select the databases and tables to synchronize. ![Select Tables](../media/select-tables.png) - If you chose sharding in the last step, you need to specify the prefix of the source database and table and specify the pattern of the target schema and table, as shown in the following picture: ![Configure Data Source](../media/configure-data-source.png) The database and table list will automatically filter the databases and tables which meet the requirements. You can also select the tables you need to synchronize. > **Note:** The sharding table schemas should be the same, including the column name, column type and index. Otherwise, the synchronization fails. 5. Create the target database (TiDB). Specify the target database information. - `Pipeline Name` is a specific name of a pipeline. - `Type` is the target source type (only supports `TiDB` currently). - `Name` is the target name. ![Create Target](../media/create-target.png) 6. Pipeline pre-check After creating a complete pipeline for data synchronization task, this pipeline will be pre-checked. The check result is as shown in the following picture: ![Pipeline Pre-check Result](../media/pipeline-precheck-result.png) 7. Synchronize If the pipeline pre-check is successful, you can start the synchronization process. You can click the eye icon in the upper right corner in the interface to show the detailed synchronization information. ## Wormhole FAQ - Q: If the Kafka cluster already exists online, can I use it when deploying Wormhole? A: Yes. There is no need for standalone deployment. But if the existing cluster is busy, the synchronization speed of Wormhole may be affected. - Q: The online machine resources are limited. Can I deploy Ether, Portal and Stargate on the current machine? A: Yes.