Added tests for binlog_server: Part-1

mtr/binlog_streaming/include/diff_with_storage_object.inc

@@ -33,7 +33,7 @@ if ($storage_backend == file)
 if ($storage_backend == s3)
 {
   --let $downloaded_file_path = $MYSQL_TMP_DIR/diff_with_storage_object.downloaded
-  --exec $aws_cli s3 cp s3://$aws_s3_bucket$storage_object $downloaded_file_path > /dev/null
+  --exec $aws_cli s3 cp s3://$aws_s3_bucket/$storage_object $downloaded_file_path > /dev/null
   --diff_files $local_file $downloaded_file_path
   --remove_file $downloaded_file_path
 }

mtr/binlog_streaming/include/patch_binlog_in_use_flag.inc

@@ -0,0 +1,31 @@
+# Copy a binlog file and clear LOG_EVENT_BINLOG_IN_USE_F in the patched copy.
+#
+# Usage:
+#   --let $binlog_file_to_patch = $binlog_base_dir/$binlog_name
+#   --let PATCHED_BINLOG_FILE = $MYSQL_TMP_DIR/$binlog_name.patched
+#   --source ../include/patch_binlog_in_use_flag.inc
+
+--echo
+--echo *** Patching the server version of the binlog file to clear the
+--echo *** LOG_EVENT_BINLOG_IN_USE_F (currently in use) flag.
+--copy_file $binlog_file_to_patch $PATCHED_BINLOG_FILE
+
+--perl
+  use strict;
+  use warnings;
+  use constant MAGIC_OFFSET => 21;
+  my $binlog_file_perl = $ENV{'PATCHED_BINLOG_FILE'};
+
+  open(my $fh, '+<:raw', $binlog_file_perl) or die "Failed to open file: $!";
+
+  seek($fh, MAGIC_OFFSET, 0);
+  my $byte;
+  read($fh, $byte, 1);
+
+  $byte = ord($byte) & 0xFE;
+
+  seek($fh, MAGIC_OFFSET, 0);
+  print $fh pack('C', $byte);
+
+  close($fh);
+EOF

mtr/binlog_streaming/include/set_up_binsrv_environment.inc

@@ -38,11 +38,11 @@ if ($storage_backend == file)
 if ($storage_backend == s3)
 {
   --let $binsrv_buffer_path = $MYSQL_TMP_DIR/buffer
-  --let $binsrv_storage_path = `SELECT CONCAT('/mtr-', UUID())`
+  --let $binsrv_storage_path = `SELECT CONCAT('mtr-', UUID())`
   --let $aws_s3_bucket = $MTR_BINSRV_AWS_S3_BUCKET
   if ($MTR_BINSRV_AWS_S3_ENDPOINT != '')
   {
-    eval SET @storage_uri = CONCAT('http://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', REPLACE('$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '/', '%2F'), '@', '$MTR_BINSRV_AWS_S3_ENDPOINT', '/', '$MTR_BINSRV_AWS_S3_BUCKET', '$binsrv_storage_path');
+    eval SET @storage_uri = CONCAT('http://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', REPLACE('$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '/', '%2F'), '@', '$MTR_BINSRV_AWS_S3_ENDPOINT', '/', '$MTR_BINSRV_AWS_S3_BUCKET', '/', '$binsrv_storage_path');
   }
   if ($MTR_BINSRV_AWS_S3_ENDPOINT == '')
   {
@@ -51,7 +51,7 @@ if ($storage_backend == s3)
     {
       --let $qualified_bucket = $qualified_bucket.$MTR_BINSRV_AWS_S3_REGION
     }
-    eval SET @storage_uri = CONCAT('s3://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', REPLACE('$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '/', '%2F'), '@', '$qualified_bucket', '$binsrv_storage_path');
+    eval SET @storage_uri = CONCAT('s3://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', REPLACE('$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '/', '%2F'), '@', '$qualified_bucket', '/', '$binsrv_storage_path');
   }
 }
 

mtr/binlog_streaming/include/tear_down_binsrv_environment.inc

@@ -6,7 +6,7 @@ if ($storage_backend == file)
 }
 if ($storage_backend == s3)
 {
-  --exec $aws_cli s3 rm s3://$aws_s3_bucket$binsrv_storage_path/ --recursive > /dev/null
+  --exec $aws_cli s3 rm s3://$aws_s3_bucket/$binsrv_storage_path/ --recursive > /dev/null
   --force-rmdir $binsrv_buffer_path
 }
 

mtr/binlog_streaming/include/terminate_binsrv.inc

@@ -0,0 +1,19 @@
+# Terminate the Binlog Server Utility and wait until the process exits.
+#
+# Usage:
+#   --let $binsrv_pid = <pid>
+#   --source ../include/terminate_binsrv.inc
+
+--exec kill -s TERM $binsrv_pid
+--let EXPORTED_BINSRV_PID = $binsrv_pid
+--perl
+  use strict;
+  use warnings;
+  use Errno;
+  my $pid = $ENV{'EXPORTED_BINSRV_PID'};
+  my $not_present = (!kill(0, $pid) && $! == Errno::ESRCH);
+  while (!$not_present) {
+    sleep(1);
+    $not_present = (!kill(0, $pid) && $! == Errno::ESRCH);
+  }
+EOF

mtr/binlog_streaming/include/trigger_binsrv_pull.inc

@@ -0,0 +1,13 @@
+# Start the Binlog Server Utility pull command in background.
+#
+# Usage:
+#   --let $binsrv_spawn_cmd_line = $BINSRV pull ... & echo \$! > $binsrv_pid_file
+#   --source ../include/trigger_binsrv_pull.inc
+
+--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

mtr/binlog_streaming/include/wait_for_pattern_count_in_file.inc

@@ -0,0 +1,53 @@
+# Wait until $grep_pattern appears at least $grep_match_count times in
+# $grep_file.
+#
+# Usage:
+#   --let $grep_pattern = <fixed string to search for>
+#   --let $grep_file = FILENAME
+#   --let $grep_match_count = N
+#   [--let $wait_timeout = 120]
+#   [--let $skip_log_pattern = 1]
+#   --source ../include/wait_for_pattern_count_in_file.inc
+
+if ($grep_match_count == "")
+{
+  --die grep_match_count must be set for wait_for_pattern_count_in_file.inc
+}
+if ($wait_timeout == "")
+{
+  --let $wait_timeout = 120
+}
+
+if ($skip_log_pattern != 1)
+{
+  --echo include/wait_for_pattern_count_in_file.inc [$grep_pattern] count=$grep_match_count
+}
+
+--let EXPORTED_WFPCIF_PATTERN = $grep_pattern
+--let EXPORTED_WFPCIF_FILE = $grep_file
+--let EXPORTED_WFPCIF_COUNT = $grep_match_count
+--let EXPORTED_WFPCIF_TIMEOUT = $wait_timeout
+--perl
+  use strict;
+  use warnings;
+  my $pattern = $ENV{'EXPORTED_WFPCIF_PATTERN'}
+      or die 'grep_pattern is not set';
+  my $file = $ENV{'EXPORTED_WFPCIF_FILE'}
+      or die 'grep_file is not set';
+  my $required = $ENV{'EXPORTED_WFPCIF_COUNT'} + 0;
+  my $timeout = ($ENV{'EXPORTED_WFPCIF_TIMEOUT'} // '') + 0;
+  $timeout = 120 if $timeout <= 0;
+  for (my $attempt = 0; $attempt < $timeout; $attempt++) {
+    my $count = 0;
+    if (-f $file) {
+      open my $fh, '<', $file or die "cannot open $file: $!";
+      while (my $line = <$fh>) {
+        $count++ if index($line, $pattern) >= 0;
+      }
+      close $fh;
+    }
+    exit 0 if $count >= $required;
+    sleep 1 if $attempt + 1 < $timeout;
+  }
+  die "Timed out waiting for $required occurrence(s) of '$pattern' in $file";
+EOF

mtr/binlog_streaming/r/binsrv.result

@@ -29,9 +29,13 @@ INSERT INTO t1 VALUES(DEFAULT);
 *** Checking that the Binlog Server utility detected an empty storage
 include/assert_grep.inc [Binlog storage must be initialized on an empty directory]
 
+*** Verifying that binlog files were written to storage
+include/assert_grep.inc [First binlog should exist in storage]
+include/assert_grep.inc [Second binlog also hould exist in storage]
+
 *** Comparing server and downloaded versions of the first binlog file.
 
-*** Patching the server version of the second binlog file to clear the
+*** Patching the server version of the binlog file to clear the
 *** LOG_EVENT_BINLOG_IN_USE_F (currently in use) flag.
 
 *** Comparing server and downloaded versions of the second binlog file.

mtr/binlog_streaming/r/checkpointing_verification.result

@@ -0,0 +1,127 @@
+*** Testing checkpointing verification
+*** For S3 backend, each checkpoint triggers a full object re-upload.
+
+*** Resetting replication at the very beginning of the test.
+
+*** Creating a table and generating large binlog events
+CREATE TABLE t1(
+id SERIAL,
+val CHAR(64) CHARACTER SET ascii NOT NULL,
+PRIMARY KEY(id)
+) ENGINE=InnoDB;
+
+*** Generating data to trigger checkpointing
+
+*** Determining current open binlog file name
+
+*** Setting up checkpointing configuration
+
+*** Setting up configuration
+SET @save_binsrv_checkpoint_size = 'binsrv_checkpoint_size';
+SET @save_binsrv_checkpoint_interval = 'binsrv_checkpoint_interval';
+
+*** Test 1: Test checkpointing with existing open binary log
+*** Expectation:
+*** - Without checkpointing: storage object stays at magic payload size while binsrv runs
+*** - With size checkpointing: storage object grows beyond magic payload while binsrv runs
+*** Note: Checkpointing only works with S3 backend, not with file backend
+
+Executing binlog_server in pull mode without checkpointing
+
+*** 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.
+include/read_file_to_var.inc
+
+*** Waiting until the Binlog Server Utility has downloaded the
+*** existing binlog data.
+include/read_file_to_var.inc
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+
+Executing binlog_server in pull mode with checkpointing
+
+*** 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.
+include/read_file_to_var.inc
+
+*** Waiting until the Binlog Server Utility has downloaded the
+*** existing binlog data.
+include/read_file_to_var.inc
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+*** Test 2: Test that new updates in binary log on server will be flushed
+*** to the binlog_server storage directory due to checkpointing.
+
+
+*** Starting pull mode in background to monitor checkpointing
+
+*** 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.
+include/read_file_to_var.inc
+
+*** Waiting until the Binlog Server Utility has downloaded all
+*** existing binlog data.
+
+*** Capturing current storage object size (before workload)
+include/read_file_to_var.inc
+
+*** Generating more data to trigger checkpointing during pull
+
+*** Waiting until the Binlog Server Utility has downloaded the
+*** latest binlog changes.
+
+*** Capturing storage object size after checkpointing workload
+include/read_file_to_var.inc
+
+*** Verifying checkpointing occurred (by storage object size change)
+
+*** Patching the server version of the binlog file to clear the
+*** LOG_EVENT_BINLOG_IN_USE_F (currently in use) flag.
+
+*** Terminating pull mode
+
+*** Verifying binlog file integrity after checkpointing
+
+*** Starting pull mode again to verify it resumes correctly without any issues.
+include/read_file_to_var.inc
+
+*** Adding more data
+INSERT INTO t1(val) VALUES(SHA2(LAST_INSERT_ID(), 256));
+FLUSH BINARY LOGS;
+
+*** Waiting until the Binlog Server Utility has downloaded the
+*** flushed binlog file and started the new one.
+
+*** Terminating pull mode
+
+*** Verifying that resume worked correctly (no duplicate data)
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+DROP TABLE t1;

mtr/binlog_streaming/r/pull_mode.result

@@ -60,4 +60,61 @@ FLUSH BINARY LOGS;
 *** Removing the Binlog Server utility log file.
 
 *** Removing the Binlog Server utility configuration file.
-KILL CONNECTION <CONNECTION_ID>;
+
+*** Pull mode: zero idle_time
+
+*** 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.
+
+*** Starting Binlog Server Utility in background in pull mode with zero idle time
+include/read_file_to_var.inc
+
+*** Creating data.
+CREATE TABLE t1(id INT UNSIGNED NOT NULL AUTO_INCREMENT, PRIMARY KEY(id)) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(DEFAULT);
+*** Waiting for read timeout to trigger reconnection.
+INSERT INTO t1 VALUES(DEFAULT);
+
+*** Terminating the utility
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+
+*** Pull mode: very short timeouts and rapid reconnection
+
+*** 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.
+
+*** Starting Binlog Server Utility in background
+include/read_file_to_var.inc
+
+*** Creating data and restarting server multiple times
+INSERT INTO t1 VALUES(DEFAULT);
+FLUSH BINARY LOGS;
+# restart
+INSERT INTO t1 VALUES(DEFAULT);
+FLUSH BINARY LOGS;
+# restart
+INSERT INTO t1 VALUES(DEFAULT);
+
+*** Terminating the utility
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+DROP TABLE t1;