English | 中文

mysql-binlog-connector-rust

Overview

• A simple but strong lib to dump mysql binlog (binlog_format=ROW) and parse Row Based Replication Events in RUST with async IO.

Supported mysql versions

• mysql 5.6 (tested in mysql:5.6.51)
• mysql 5.7 (tested in mysql:5.7.40)
• mysql 8.0 (tested in mysql:8.0.31)

Supported replication type

• binlog-file-position-based replication
• gtid-based replication

Supported event types

• FORMAT_DESCRIPTION_EVENT
• ROTATE_EVENT
• PREVIOUS_GTIDS_LOG_EVENT
• GTID_LOG_EVENT
• QUERY_EVENT
• XID_EVENT
• XA_PREPARE_LOG_EVENT
• TRANSACTION_PAYLOAD_EVENT
• ROWS_QUERY_LOG_EVENT
• TABLE_MAP_EVENT
• WRITE_ROWS_EVENT_V1
• WRITE_ROWS_EVENT
• UPDATE_ROWS_EVENT_V1
• UPDATE_ROWS_EVENT
• DELETE_ROWS_EVENT_V1
• DELETE_ROWS_EVENT
• for more details, refer to: mysql doc

Mapping between mysql columns and rust types

• for CHAR / VARCHAR columns, since binlog contains no charset information, we just get raw bytes and store them in ColumnValue::String(Vec) objects, you may need to convert them into strings based on column metadatas for further usage.
• for UNSIGNED numeric columns, since binlog contains no unsigned flags, we just parse them as signed numerics, you may need to convert them into unsigned values based on column metadatas for further usage.
• for JSON columns, we get raw bytes and store them in ColumnValue::Json(Vec) objects, we also provide a default deserializer "JsonBinary" to parse them into strings, find example later in this doc.

Quick start

Run tests

• start a mysql, enable binlog without binlog-transaction-compression

docker run -d --name mysql57 \
--platform linux/x86_64 \
-it --restart=always \
-p 3307:3306 \
-e MYSQL_ROOT_PASSWORD="123456" \
mysql:5.7.40 \
--lower_case_table_names=1 \
--character-set-server=utf8 \
--collation-server=utf8_general_ci \
--datadir=/var/lib/mysql \
--user=mysql \
--server_id=1 \
--log_bin=/var/lib/mysql/mysql-bin.log \
--max_binlog_size=100M \
--gtid_mode=ON \
--enforce_gtid_consistency=ON \
--binlog_format=ROW

• start a mysql, enable binlog with binlog-transaction-compression

docker run -d --name mysql80 \
--platform linux/x86_64 \
-it  --restart=always \
-p 3308:3306 -e MYSQL_ROOT_PASSWORD="123456" \
 mysql:8.0.31 \
 --lower_case_table_names=1 \
 --character-set-server=utf8 \
 --collation-server=utf8_general_ci \
 --datadir=/var/lib/mysql \
 --user=mysql \
 --server_id=1 \
 --log_bin=/var/lib/mysql/mysql-bin.log \
 --max_binlog_size=100M \
 --gtid_mode=ON \
 --enforce_gtid_consistency=ON \
 --binlog_format=ROW \
 --binlog-transaction-compression \
 --binlog_rows_query_log_events=ON \
 --default_authentication_plugin=mysql_native_password \
 --default_time_zone="+08:00"

• update configs in tests/.env

db_url=mysql://root:123456@127.0.0.1:3307
server_id=200
default_db="db_test"
default_tb="tb_test"
binlog_parse_millis=100

• run tests

cargo test --package mysql-binlog-connector-rust --test integration_test

• TLS configuration
• TLS is optional and disabled by default.
• To require TLS, append ssl-mode=required to db_url. To use plain TCP, set ssl-mode=disabled or omit it.
• This crate currently supports two TLS backends:
• openssl-tls: better compatibility with older MySQL 5.7 TLS stacks
• rustls: pure Rust TLS backend
• Enable exactly one TLS feature when building:

cargo test --package mysql-binlog-connector-rust --test integration_test --features openssl-tls

or

cargo test --package mysql-binlog-connector-rust --test integration_test --features rustls

• example db_url with TLS enabled:

db_url=mysql://root:123456@127.0.0.1:3307?ssl-mode=required

• example db_url with TLS disabled:

db_url=mysql://root:123456@127.0.0.1:3307?ssl-mode=disabled

• Note: current TLS implementation skips server certificate verification. It encrypts the transport, but does not verify server identity.
• each test will:
• execute sqls to create tables and generate binlogs
• dump and parse binlogs
• wait binlog_parse_millis for all binlogs to be parsed
• you may increase binlog_parse_millis for big transactions

Examples

fn main() {
    block_on(dump_and_parse())
}

async fn dump_and_parse() {
    let env_path = env::current_dir().unwrap().join("example/src/.env");
    dotenv::from_path(env_path).unwrap();
    let url = env::var("db_url").unwrap();
    let server_id: u64 = env::var("server_id").unwrap().parse().unwrap();
    let binlog_filename = env::var("binlog_filename").unwrap();
    let binlog_position: u32 = env::var("binlog_position").unwrap().parse().unwrap();
    let gtid_set = env::var("gtid_set").unwrap();

    let start_position = if !gtid_set.is_empty() {
        StartPosition::Gtid(gtid_set)
    } else if !binlog_filename.is_empty() {
        StartPosition::BinlogPosition(binlog_filename, binlog_position)
    } else {
        StartPosition::Latest
    };

    let mut stream = BinlogClient::new(url.as_str(), server_id, start_position)
        /// Heartbeat interval in seconds
        /// Server will send a heartbeat event if no binlog events are received within this interval
        /// If heartbeat_interval_secs=0, server won't send heartbeat events
        /// default is 0 means not enabled
        .with_master_heartbeat(Duration::from_secs(5))
        /// Network operation timeout in seconds
        /// Maximum wait time for operations like connection establishment and data reading
        /// default is 60 secs
        .with_read_timeout(Duration::from_secs(60))
        /// TCP keepalive idle time and keepalive interval time
        /// default is (0 secs, 0 secs) means keepalive not enabled
        .with_keepalive(Duration::from_secs(60), Duration::from_secs(10))
        .connect()
        .await
        .unwrap();

    let mut stream = client.connect().await.unwrap();

    loop {
        let (header, data) = stream.read().await.unwrap();
        println!("header: {:?}", header);
        println!("data: {:?}", data);
        println!();
    }
}

• Run the example with TLS support:

cargo run -p mysql-binlog-connector-rust-example

• The bundled example crate currently enables openssl-tls by default. If you want to switch to rustls, update example/Cargo.toml and set:

mysql-binlog-connector-rust = { path = "../", features = ["rustls"] }

Example 1: parse binlogs with binlog-transaction-compression disabled

• execute sqls

flush logs;

SET autocommit=0;
CREATE DATABASE test_db;
USE test_db;
CREATE TABLE test_tb(id INT, value INT);
INSERT INTO test_tb VALUES(1,1),(2,2),(3,3),(4,4);
UPDATE test_tb SET value=3 WHERE id in(1,2);
DELETE FROM test_tb WHERE id in (1,2);
TRUNCATE TABLE test_tb;
DROP TABLE test_tb;
commit;

• show binlog events

```sql mysql> show binary logs; +------------------+-----------+ | Log_name | File_size | +------------------+-----------+ | mysql-bin.000050 | 1255 |

mysql> show binlog events in 'mysql-bin.000050'; +------------------+------+----------------+-----------+-------------+------------------------------------------------------------------------+ | Log_name | Pos | Event_type | Server_id | End_log_pos | Info | +------------------+------+----------------+-----------+-------------+------------------------------------------------------------------------+ | mysql-bin.000050 | 4 | Format_desc | 1 | 123 | Server ver: 5.7.40-log, Binlog ver: 4 | | mysql-bin.000050 | 123 | Previous_gtids | 1 | 194 | 50dc6874-13d3-11ee-a17a-0242ac110002:1-176027 | | mysql-bin.000050 | 194 | Gtid | 1 | 259 | SET @@SESSION.GTID_NEXT= '50dc6874-13d3-11ee-a17a-0242ac110002:176028' | | mysql-bin.000050 | 259 | Query | 1 | 378 | use test_db; CREATE TABLE test_tb(id INT, value INT) | | mysql-bin.000050 | 378 | Gtid | 1 | 443 | SET @@SESSION.GTID_NEXT= '50dc6874-13d3-11ee-a17a-0242ac110002:176029' | | mysql-bin.000050 | 443 | Query | 1 | 518 | BEGIN | | mysql-bin.000050 | 518 | Table_map | 1 | 572 | table_id: 12832 (test_db.test_tb) | | mysql-bin.000050 | 572 | Write_rows | 1 | 643 | table_id: 12832 flags: STMT_END_F | | mysql-bin.000050 | 643 | Table_map | 1 | 697 | table_id: 12832 (test_db.test_tb) | | mysql-bin.000050 | 697 | Update_rows | 1 | 769 | table_id: 12832 flags: STMT_END_F | | mysql-bin.000050 | 769 | Table_map | 1 | 823 | table_id: 12832 (test_db.test_tb) | | mysql-bin.000050 | 823 | Delete_rows | 1 | 876 | table_id: 12832 flags: STMT_END_F | | mysql-bin.0000