PBS-42 feature: Implement basic support for COM_BINLOG_DUMP packet handling (part 4)

mtr/binlog_streaming/include/kill_binlog_dump_connection.inc

@@ -0,0 +1,17 @@
+# ==== Purpose ====
+#
+# Kill the server-side Binlog Dump connection left by a terminated Binlog
+# Server Utility process, if one is still present.
+#
+# ==== Usage ====
+#
+# --source ../include/kill_binlog_dump_connection.inc
+#
+
+--let $binlog_dump_connection_id = `SELECT ID FROM performance_schema.processlist WHERE COMMAND = 'Binlog Dump'`
+if ($binlog_dump_connection_id != '')
+{
+  --disable_query_log
+  eval KILL CONNECTION $binlog_dump_connection_id;
+  --enable_query_log
+}

mtr/binlog_streaming/include/start_proc_in_background.inc

@@ -0,0 +1,41 @@
+# ==== Purpose ====
+#
+# Start a process in the background and return its PID.
+#
+# ==== Usage ====
+#
+# --let $proc_command_line = command with arguments and redirects
+# --source ../include/start_proc_in_background.inc
+#
+# Parameters:
+#   $proc_command_line
+#     Complete command line to execute before the background operator.
+#
+# Output:
+#   $proc_pid
+#     PID of the started process.
+#
+if (!$proc_command_line)
+{
+  --die ERROR IN TEST: 'proc_command_line' parameter not specified
+}
+
+# We route the spawn through Perl's system() rather than --exec because
+# mysqltest strips the shell's `$!` before it can be expanded, leaving the
+# PID file empty. Passing the command as an env var and executing it from
+# Perl leaves the shell metacharacters intact.
+
+--let $proc_pid_file = $MYSQL_TMP_DIR/start_proc_in_background.pid
+--let $proc_spawn_command_line = $proc_command_line & echo \$! > $proc_pid_file
+--let EXPORTED_PROC_COMMAND_LINE = $proc_spawn_command_line
+--perl
+  use strict;
+  use warnings;
+  my $cmd = $ENV{'EXPORTED_PROC_COMMAND_LINE'};
+  system("$cmd");
+EOF
+
+--let $read_from_file = $proc_pid_file
+--source include/read_file_to_var.inc
+--let $proc_pid = $result
+--remove_file $proc_pid_file

mtr/binlog_streaming/include/terminate_proc.inc

@@ -0,0 +1,53 @@
+# ==== Purpose ====
+#
+# Send SIGTERM to a process and wait for it to terminate gracefully.
+#
+# ==== Usage ====
+#
+# --let $proc_pid = process PID
+# [--let $proc_termination_timeout = timeout in seconds]
+# --source ../include/terminate_proc.inc
+#
+# Parameters:
+#   $proc_pid
+#     PID of the process to terminate.
+#
+#   $proc_termination_timeout
+#     Maximum number of seconds to wait. Defaults to 30.
+#
+if (!$proc_pid)
+{
+  --die ERROR IN TEST: 'proc_pid' parameter not specified
+}
+if (!$proc_termination_timeout)
+{
+  --let $proc_termination_timeout = 60
+}
+
+--exec kill -s TERM $proc_pid
+
+--let $proc_termination_iteration = 0
+--let $proc_running = 1
+while ($proc_termination_iteration < $proc_termination_timeout)
+{
+  if ($proc_running)
+  {
+    --error 0, 1
+    --exec kill -0 $proc_pid 2>/dev/null
+    --let $proc_kill_status = $__error
+    if ($proc_kill_status != 0)
+    {
+      --let $proc_running = 0
+      --let $proc_termination_iteration = $proc_termination_timeout
+    }
+    if ($proc_kill_status == 0)
+    {
+      --sleep 1
+      --inc $proc_termination_iteration
+    }
+  }
+}
+if ($proc_running)
+{
+  --die Process did not terminate within the configured timeout
+}

mtr/binlog_streaming/include/wait_for_listening_port.inc

@@ -0,0 +1,41 @@
+# ==== Purpose ====
+#
+# Waits until a specified network port is ready to accept connections.
+#
+# ==== Usage ====
+#
+# --let $listening_port = 3306
+# --source ../include/wait_for_listening_port.inc
+#
+# Parameters:
+#   $listening_port
+#     Network port to wait for.
+#
+
+# We probe with bash's '/dev/tcp' because it uses a plain connect(2), so each
+# attempt is cheap and measures exactly the "listening on the port" state we
+# care about.
+
+--let $max_wait = 300
+--let $iteration = 0
+--let $port_open = 0
+while ($iteration < $max_wait)
+{
+  --error 0, 1
+  --exec bash -c "echo > /dev/tcp/127.0.0.1/$listening_port" 2>/dev/null
+  --let $port_status = $__error
+  if ($port_status == 0)
+  {
+    --let $port_open = 1
+    --let $iteration = $max_wait
+  }
+  if ($port_status != 0)
+  {
+    --sleep 1
+    --inc $iteration
+  }
+}
+if (!$port_open)
+{
+  --die The Binlog Server replication source listener did not become reachable within 300 seconds
+}

mtr/binlog_streaming/r/auth_method_switch.result

@@ -16,12 +16,8 @@
 *** ASAN.
 include/read_file_to_var.inc
 
-*** Waiting for the Binlog Server listener to come up on
-*** 127.0.0.1:<BINSRV_LISTENER_PORT>. We probe with bash's /dev/tcp
-*** instead of the mysql client because bash is not
-*** ASAN-instrumented and /dev/tcp uses a plain connect(2), so each
-*** attempt is cheap and measures exactly the "listening on the
-*** port" state we care about.
+*** Waiting for the replication source listener port ready to accept
+*** connections.
 
 *** Control: client picks the same plugin the server advertises
 *** (caching_sha2_password). No AuthMethodSwitch is expected on the
@@ -37,10 +33,8 @@ include/read_file_to_var.inc
 *** is what actually proves the switch happened - this line only
 *** proves the session survived it.
 
-*** Sending SIGTERM to the Binlog Server Utility and waiting for the
-*** process to terminate (poll kill -0 until the pid is gone).
-*** Graceful exit flushes std::cout, so any log line binsrv wrote
-*** during its lifetime is now safely on disk.
+*** Sending SIGTERM signal to the Binlog Server Utility and waiting for
+*** the process to terminate
 
 *** Confirming the switch actually happened by grepping the Binlog
 *** Server stdout log (now complete and flushed) for the marker
@@ -53,4 +47,3 @@ include/wait_for_pattern_in_file.inc [sent server auth method switch]
 *** Removing the Binlog Server utility log file.
 
 *** Removing the Binlog Server utility configuration file.
-KILL CONNECTION <CONNECTION_ID>;

mtr/binlog_streaming/r/pull_mode.result

@@ -60,4 +60,3 @@ FLUSH BINARY LOGS;
 *** Removing the Binlog Server utility log file.
 
 *** Removing the Binlog Server utility configuration file.
-KILL CONNECTION <CONNECTION_ID>;

mtr/binlog_streaming/r/rs_roundtrip.result

@@ -0,0 +1,59 @@
+*** Resetting replication at the very beginning of the test.
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+
+*** Determining the first binary log name.
+
+*** Creating source_db.tbl and inserting the first batch of rows.
+CREATE DATABASE source_db;
+CREATE TABLE source_db.tbl(
+id INT UNSIGNED NOT NULL,
+value VARCHAR(32) NOT NULL,
+PRIMARY KEY(id)
+) ENGINE=InnoDB;
+INSERT INTO source_db.tbl VALUES (1, 'one'), (2, 'two'), (3, 'three');
+
+*** Flushing the source binary log.
+FLUSH BINARY LOGS;
+
+*** Inserting the second batch of rows.
+INSERT INTO source_db.tbl VALUES (4, 'four'), (5, 'five'), (6, 'six');
+
+*** Fetching the source binary logs in position mode.
+
+*** Starting Binlog Server Utility in background in pull mode.
+include/read_file_to_var.inc
+
+*** Waiting for the replication source listener port ready to accept
+*** connections.
+
+*** Dumping the binary logs from the Binlog Server replication source.
+
+*** Sending SIGTERM signal to the Binlog Server Utility and waiting for
+*** the process to terminate
+
+*** Moving source_db.tbl to backup_db.tbl.
+CREATE DATABASE backup_db;
+RENAME TABLE source_db.tbl TO backup_db.tbl;
+DROP DATABASE source_db;
+
+*** Restoring source_db from the replication source dump.
+
+*** Comparing the restored and backup tables.
+include/diff_tables.inc [source_db.tbl, backup_db.tbl]
+
+*** Dropping the round-trip databases.
+DROP DATABASE source_db;
+DROP DATABASE backup_db;
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.

mtr/binlog_streaming/t/auth_method_switch.test

@@ -29,24 +29,8 @@ eval $stmt_reset_binary_logs_and_gtids;
 --echo *** SIGTERM flushes std::cout unconditionally, so we do not need to
 --echo *** rely on any per-line buffering trick that could misbehave under
 --echo *** ASAN.
---let $binsrv_pid_file = $MYSQL_TMP_DIR/auth_switch_binsrv.pid
---let $binsrv_spawn_cmd_line = $BINSRV pull $binsrv_config_file_path > /dev/null 2>&1 & echo \$! > $binsrv_pid_file
-
-# We route the spawn through Perl's system() rather than --exec because
-# mysqltest strips the shell's `$!` before it can be expanded, leaving the
-# PID file empty. Passing the command as an env var and executing it from
-# Perl leaves the shell metacharacters intact.
---let EXPORTED_BINSRV_SPAWN_CMD_LINE = $binsrv_spawn_cmd_line
---perl
-  use strict;
-  use warnings;
-  my $cmd = $ENV{'EXPORTED_BINSRV_SPAWN_CMD_LINE'};
-  system("$cmd");
-EOF
-
---let $read_from_file = $binsrv_pid_file
---source include/read_file_to_var.inc
---let $binsrv_pid = $result
+--let $proc_command_line = $BINSRV pull $binsrv_config_file_path > /dev/null
+--source ../include/start_proc_in_background.inc
 
 # SELECT VERSION() is one of the queries the Binlog Server's command loop
 # recognises (see network_service.cpp). We do not care what row comes
@@ -56,38 +40,10 @@ EOF
 --let $probe_query = SELECT VERSION()
 
 --echo
---echo *** Waiting for the Binlog Server listener to come up on
---echo *** 127.0.0.1:<BINSRV_LISTENER_PORT>. We probe with bash's /dev/tcp
---echo *** instead of the mysql client because bash is not
---echo *** ASAN-instrumented and /dev/tcp uses a plain connect(2), so each
---echo *** attempt is cheap and measures exactly the "listening on the
---echo *** port" state we care about.
---let $max_wait = 300
---let $iteration = 0
---let $port_open = 0
-while ($iteration < $max_wait)
-{
-  if (!$port_open)
-  {
-    --error 0, 1
-    --exec bash -c "echo > /dev/tcp/127.0.0.1/$binsrv_replication_source_port" 2>/dev/null
-    --let $port_status = $__error
-    if ($port_status == 0)
-    {
-      --let $port_open = 1
-      --let $iteration = $max_wait
-    }
-    if ($port_status != 0)
-    {
-      --sleep 1
-      --inc $iteration
-    }
-  }
-}
-if (!$port_open)
-{
-  --die The Binlog Server listener did not become reachable on its port within 300 seconds
-}
+--echo *** Waiting for the replication source listener port ready to accept
+--echo *** connections.
+--let $listening_port = $binsrv_replication_source_port
+--source ../include/wait_for_listening_port.inc
 
 --echo
 --echo *** Control: client picks the same plugin the server advertises
@@ -108,39 +64,9 @@ if (!$port_open)
 --exec $MYSQL --protocol=TCP --host=127.0.0.1 --port=$binsrv_replication_source_port --user=rpl --password=password --default-auth=mysql_native_password --skip-column-names -e "$probe_query" >/dev/null 2>&1
 
 --echo
---echo *** Sending SIGTERM to the Binlog Server Utility and waiting for the
---echo *** process to terminate (poll kill -0 until the pid is gone).
---echo *** Graceful exit flushes std::cout, so any log line binsrv wrote
---echo *** during its lifetime is now safely on disk.
---replace_result $binsrv_pid <BINSRV_PID>
---exec kill -s TERM $binsrv_pid
-
---let $max_exit_wait = 30
---let $iteration = 0
---let $binsrv_running = 1
-while ($iteration < $max_exit_wait)
-{
-  if ($binsrv_running)
-  {
-    --error 0, 1
-    --exec kill -0 $binsrv_pid 2>/dev/null
-    --let $kill_status = $__error
-    if ($kill_status != 0)
-    {
-      --let $binsrv_running = 0
-      --let $iteration = $max_exit_wait
-    }
-    if ($kill_status == 0)
-    {
-      --sleep 1
-      --inc $iteration
-    }
-  }
-}
-if ($binsrv_running)
-{
-  --die Binlog Server Utility did not terminate within 30 seconds
-}
+--echo *** Sending SIGTERM signal to the Binlog Server Utility and waiting for
+--echo *** the process to terminate
+--source ../include/terminate_proc.inc
 
 --echo
 --echo *** Confirming the switch actually happened by grepping the Binlog
@@ -152,17 +78,13 @@ if ($binsrv_running)
 --let $wait_timeout = 60
 --source include/wait_for_pattern_in_file.inc
 
---remove_file $binsrv_pid_file
-
 # cleaning up
 --source ../include/tear_down_binsrv_environment.inc
 
 # As the Binlog Server Utility interrupts the connection upon timeout, here we
 # need to close it on the MySQL server side as well in order to make sure that
 # MTR 'check-test' before and after the test produces the same output.
---let $binlog_dump_connection_id = `SELECT ID FROM performance_schema.processlist WHERE COMMAND = 'Binlog Dump'`
---replace_result $binlog_dump_connection_id <CONNECTION_ID>
-eval KILL CONNECTION $binlog_dump_connection_id;
+--source ../include/kill_binlog_dump_connection.inc
 
 # Also, we use 'count_sessions' include files to make sure that 'Binlog Dump'
 # connection is indeed closed.