PS-10245 feature: Implement receiving binlog events in GTID mode (part 3) (#83)

percona-ysorokin · web-flow · commit f4a05587bbce · 2025-12-22T20:12:38.000+01:00
https://perconadev.atlassian.net/browse/PS-10245 Implemented more sophisticated model of 'binstv::event::reader_context' state machine. Instead of (ROTATE(artificial) FORMAT_DESCRIPTION <ANY>* (ROTATE|STOP)?)+ we now also track GTID-related events and transaction boundaries ( ROTATE(artificial) FORMAT_DESCRIPTION PREVIOUS_GTIDS_LOG? ((ANONYMOUS_GTID_LOG | GTID_LOG | GTID_TAGGED_LOG) <ANY>*)* (ROTATE | STOP)? )+ Implemented additional method 'is_at_transaction_boundary()' that can be used to check if we reached the end of transaction after processing the event. Main application extended with simplified transaction event logging at 'info' level: - we now write simple messages with event code types and event flags if any (e.g. 'artificial') - we also write corresponding GTIDs at the end of transactions. Added new 'binlog_streaming.kill_and_restart' MTR test case which checks for different endings of binlog files after rotation: - normal 'ROTATE' event (after 'FLUSH BINARY LOGS') - normal 'STOP' event (after server shutdown) - termination without 'ROTATE' / 'STOP' event (in case server crashed / was killed and restarted) - currently open binlog file.
diff --git a/mtr/binlog_streaming/r/kill_and_restart.result b/mtr/binlog_streaming/r/kill_and_restart.result
@@ -0,0 +1,59 @@
+*** Resetting replication at the very beginning of the test.
+
+*** Determining the first fresh binary log name.
+
+*** Creating a simple table and filling it with some data.
+CREATE TABLE t1(id INT UNSIGNED NOT NULL AUTO_INCREMENT, PRIMARY KEY(id)) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(DEFAULT);
+
+*** Restarting the server gracefully
+# restart
+
+*** Determining the second binary log name after restart
+
+*** Filling the table with some more data.
+INSERT INTO t1 VALUES(DEFAULT);
+
+*** Killing and restarting the server
+# Kill and restart
+
+*** Determining the third binary log name after killing and restarting
+
+*** Filling the table with some more data.
+INSERT INTO t1 VALUES(DEFAULT);
+
+*** Flushing the third binary log and switching to the fourth one.
+FLUSH BINARY LOGS;
+
+*** Determining the fourth binary log name after binlog rotation
+
+*** Filling the table with some more data.
+INSERT INTO t1 VALUES(DEFAULT);
+
+*** Checking if the first binlog file ends with the STOP event.
+
+*** Checking if the second binlog file ends with neither stop, nor ROTATE event (XID in this case).
+
+*** Checking if the third binlog file ends with the ROTATE event.
+
+*** Checking if the fourth (still open) binlog file ends with the commit of a transaction (XID).
+
+*** 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.
+
+*** Executing the Binlog Server utility to download all binlog data
+*** from the server to the <BINSRV_STORAGE_PATH> directory
+
+*** Dropping the table.
+DROP TABLE t1;
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
diff --git a/mtr/binlog_streaming/t/kill_and_restart.test b/mtr/binlog_streaming/t/kill_and_restart.test
@@ -0,0 +1,145 @@
+--source ../include/have_binsrv.inc
+
+--source ../include/v80_v84_compatibility_defines.inc
+
+# in case of --repeat=N, we need to start from a fresh binary log to make
+# this test deterministic
+--echo *** Resetting replication at the very beginning of the test.
+--disable_query_log
+eval $stmt_reset_binary_logs_and_gtids;
+--enable_query_log
+
+--echo
+--echo *** Determining the first fresh binary log name.
+--let $first_binlog = query_get_value($stmt_show_binary_log_status, File, 1)
+
+--echo
+--echo *** Creating a simple table and filling it with some data.
+CREATE TABLE t1(id INT UNSIGNED NOT NULL AUTO_INCREMENT, PRIMARY KEY(id)) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(DEFAULT);
+
+--echo
+--echo *** Restarting the server gracefully
+--source include/restart_mysqld.inc
+
+--echo
+--echo *** Determining the second binary log name after restart
+--let $second_binlog = query_get_value($stmt_show_binary_log_status, File, 1)
+
+--echo
+--echo *** Filling the table with some more data.
+INSERT INTO t1 VALUES(DEFAULT);
+
+--echo
+--echo *** Killing and restarting the server
+--source include/kill_and_restart_mysqld.inc
+
+--echo
+--echo *** Determining the third binary log name after killing and restarting
+--let $third_binlog = query_get_value($stmt_show_binary_log_status, File, 1)
+
+--echo
+--echo *** Filling the table with some more data.
+INSERT INTO t1 VALUES(DEFAULT);
+
+--echo
+--echo *** Flushing the third binary log and switching to the fourth one.
+FLUSH BINARY LOGS;
+
+--echo
+--echo *** Determining the fourth binary log name after binlog rotation
+--let $fourth_binlog = query_get_value($stmt_show_binary_log_status, File, 1)
+
+--echo
+--echo *** Filling the table with some more data.
+INSERT INTO t1 VALUES(DEFAULT);
+
+--echo
+--echo *** Checking if the first binlog file ends with the STOP event.
+# SHOW BINLOG EVENTS IN 'binlog.000001';
+# Log_name        Pos     Event_type      Server_id       End_log_pos     Info
+# binlog.000001   4       Format_desc     1       126     Server ver: 8.0.43-34, Binlog ver: 4
+# binlog.000001   126     Previous_gtids  1       157
+# binlog.000001   157     Anonymous_Gtid  1       236     SET @@SESSION.GTID_NEXT= 'ANONYMOUS'
+# binlog.000001   236     Query   1       411     use `test`; CREATE TABLE t1(id INT UNSIGNED NOT NULL AUTO_INCREMENT, PRIMARY KEY(id)) ENGINE=InnoDB /* xid=6 */
+# binlog.000001   411     Anonymous_Gtid  1       490     SET @@SESSION.GTID_NEXT= 'ANONYMOUS'
+# binlog.000001   490     Query   1       565     BEGIN
+# binlog.000001   565     Table_map       1       613     table_id: 125 (test.t1)
+# binlog.000001   613     Write_rows      1       653     table_id: 125 flags: STMT_END_F
+# binlog.000001   653     Xid     1       684     COMMIT /* xid=7 */
+# binlog.000001   684     Stop    1       707
+--let $extracted_event_type = query_get_value(SHOW BINLOG EVENTS IN '$first_binlog', Event_type, 10)
+--assert($extracted_event_type == "Stop")
+--let $extracted_event_type = query_get_value(SHOW BINLOG EVENTS IN '$first_binlog', Event_type, 11)
+--assert($extracted_event_type == "No such row")
+
+--echo
+--echo *** Checking if the second binlog file ends with neither stop, nor ROTATE event (XID in this case).
+# SHOW BINLOG EVENTS IN 'binlog.000002';
+# Log_name        Pos     Event_type      Server_id       End_log_pos     Info
+# binlog.000002   4       Format_desc     1       126     Server ver: 8.0.43-34, Binlog ver: 4
+# binlog.000002   126     Previous_gtids  1       157
+# binlog.000002   157     Anonymous_Gtid  1       236     SET @@SESSION.GTID_NEXT= 'ANONYMOUS'
+# binlog.000002   236     Query   1       311     BEGIN
+# binlog.000002   311     Table_map       1       359     table_id: 89 (test.t1)
+# binlog.000002   359     Write_rows      1       399     table_id: 89 flags: STMT_END_F
+# binlog.000002   399     Xid     1       430     COMMIT /* xid=9 */
+--let $extracted_event_type = query_get_value(SHOW BINLOG EVENTS IN '$second_binlog', Event_type, 7)
+--assert($extracted_event_type == "Xid")
+--let $extracted_event_type = query_get_value(SHOW BINLOG EVENTS IN '$second_binlog', Event_type, 8)
+--assert($extracted_event_type == "No such row")
+
+--echo
+--echo *** Checking if the third binlog file ends with the ROTATE event.
+# SHOW BINLOG EVENTS IN 'binlog.000003';
+# Log_name        Pos     Event_type      Server_id       End_log_pos     Info
+# binlog.000003   4       Format_desc     1       126     Server ver: 8.0.43-34, Binlog ver: 4
+# binlog.000003   126     Previous_gtids  1       157
+# binlog.000003   157     Anonymous_Gtid  1       236     SET @@SESSION.GTID_NEXT= 'ANONYMOUS'
+# binlog.000003   236     Query   1       311     BEGIN
+# binlog.000003   311     Table_map       1       359     table_id: 89 (test.t1)
+# binlog.000003   359     Write_rows      1       399     table_id: 89 flags: STMT_END_F
+# binlog.000003   399     Xid     1       430     COMMIT /* xid=9 */
+# binlog.000003   430     Rotate  1       474     binlog.000004;pos=4
+--let $extracted_event_type = query_get_value(SHOW BINLOG EVENTS IN '$third_binlog', Event_type, 8)
+--assert($extracted_event_type == "Rotate")
+--let $extracted_event_type = query_get_value(SHOW BINLOG EVENTS IN '$third_binlog', Event_type, 9)
+--assert($extracted_event_type == "No such row")
+
+--echo
+--echo *** Checking if the fourth (still open) binlog file ends with the commit of a transaction (XID).
+# SHOW BINLOG EVENTS IN 'binlog.000004';
+# Log_name        Pos     Event_type      Server_id       End_log_pos     Info
+# binlog.000004   4       Format_desc     1       126     Server ver: 8.0.43-34, Binlog ver: 4
+# binlog.000004   126     Previous_gtids  1       157
+# binlog.000004   157     Anonymous_Gtid  1       236     SET @@SESSION.GTID_NEXT= 'ANONYMOUS'
+# binlog.000004   236     Query   1       311     BEGIN
+# binlog.000004   311     Table_map       1       359     table_id: 89 (test.t1)
+# binlog.000004   359     Write_rows      1       399     table_id: 89 flags: STMT_END_F
+# binlog.000004   399     Xid     1       430     COMMIT /* xid=12 */
+--let $extracted_event_type = query_get_value(SHOW BINLOG EVENTS IN '$fourth_binlog', Event_type, 7)
+--assert($extracted_event_type == "Xid")
+--let $extracted_event_type = query_get_value(SHOW BINLOG EVENTS IN '$fourth_binlog', Event_type, 8)
+--assert($extracted_event_type == "No such row")
+
+# identifying backend storage type ('file' or 's3')
+--source ../include/identify_storage_backend.inc
+
+# creating data directory, configuration file, etc.
+--let $binsrv_connect_timeout = 20
+--let $binsrv_read_timeout = 60
+--let $binsrv_idle_time = 10
+--let $binsrv_verify_checksum = TRUE
+--source ../include/set_up_binsrv_environment.inc
+
+--echo
+--echo *** Executing the Binlog Server utility to download all binlog data
+--echo *** from the server to the <BINSRV_STORAGE_PATH> directory
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+
+--echo
+--echo *** Dropping the table.
+DROP TABLE t1;
+
+# cleaning up
+--source ../include/tear_down_binsrv_environment.inc
diff --git a/src/app.cpp b/src/app.cpp
@@ -298,11 +298,11 @@ void log_replication_info(
   msg += ", starting from ";
   if (replication_mode == binsrv::replication_mode_type::position) {
     if (storage.has_current_binlog_name()) {
-      msg += "the very beginning";
-    } else {
       msg += storage.get_current_binlog_name();
       msg += ":";
       msg += std::to_string(storage.get_current_position());
+    } else {
+      msg += "the very beginning";
     }
   } else {
     const auto &gtids{storage.get_gtids()};
@@ -320,6 +320,9 @@ void log_replication_info(
 
 void log_span_dump(binsrv::basic_logger &logger,
                    util::const_byte_span portion) {
+  logger.log(binsrv::log_severity::debug,
+             "fetched " + std::to_string(std::size(portion)) +
+                 "-byte(s) event from binlog");
   static constexpr std::size_t bytes_per_dump_line{16U};
   std::size_t offset{0U};
   while (offset < std::size(portion)) {
@@ -489,19 +492,28 @@ void receive_binlog_events(
           "unexpected event prefix");
     }
     portion = portion.subspan(1U);
-    logger.log(binsrv::log_severity::info,
-               "fetched " + std::to_string(std::size(portion)) +
-                   "-byte(s) event from binlog");
     log_span_dump(logger, portion);
 
     // TODO: just for redirection to another byte stream we need to parse
     //       the ROTATE and FORMAT_DESCRIPTION events only, every other one
     //       can be just considered as a data portion (unless we want to do
     //       basic integrity checks like event sizes / position and CRC)
     const binsrv::event::event current_event{context, portion};
+    const auto &current_header{current_event.get_common_header()};
+    auto readable_flags{current_header.get_readable_flags()};
+    logger.log(
+        binsrv::log_severity::info,
+        "event: " + std::string{current_header.get_readable_type_code()} +
+            (readable_flags.empty() ? "" : "(" + readable_flags + ")"));
     logger.log(binsrv::log_severity::debug,
                "Parsed event:\n" +
                    boost::lexical_cast<std::string>(current_event));
+    if (context.is_at_transaction_boundary()) {
+      logger.log(
+          binsrv::log_severity::info,
+          "encountered the end of transaction " +
+              boost::lexical_cast<std::string>(context.get_transaction_gtid()));
+    }
 
     process_binlog_event(current_event, portion, storage, skip_open_binlog);
   }
diff --git a/src/binsrv/event/gtid_log_post_header.cpp b/src/binsrv/event/gtid_log_post_header.cpp
@@ -27,6 +27,8 @@
 
 #include "binsrv/event/gtid_log_flag_type.hpp"
 
+#include "binsrv/gtids/common_types.hpp"
+#include "binsrv/gtids/gtid.hpp"
 #include "binsrv/gtids/uuid.hpp"
 
 #include "util/byte_span.hpp"
@@ -132,6 +134,13 @@ gtid_log_post_header::get_flags() const noexcept {
   return boost::lexical_cast<std::string>(get_uuid());
 }
 
+[[nodiscard]] gtids::gtid gtid_log_post_header::get_gtid() const {
+  if (get_gno() < gtids::min_gno) {
+    return {};
+  }
+  return {get_uuid(), get_gno()};
+}
+
 std::ostream &operator<<(std::ostream &output,
                          const gtid_log_post_header &obj) {
   return output << "flags: " << obj.get_readable_flags()
diff --git a/src/binsrv/event/gtid_log_post_header.hpp b/src/binsrv/event/gtid_log_post_header.hpp
@@ -24,7 +24,8 @@
 #include "binsrv/event/gtid_log_flag_type_fwd.hpp"
 
 #include "binsrv/gtids/common_types.hpp"
-#include "binsrv/gtids/uuid.hpp"
+#include "binsrv/gtids/gtid_fwd.hpp"
+#include "binsrv/gtids/uuid_fwd.hpp"
 
 #include "util/byte_span_fwd.hpp"
 
@@ -50,7 +51,12 @@ class [[nodiscard]] gtid_log_post_header {
   [[nodiscard]] gtids::uuid get_uuid() const noexcept;
   [[nodiscard]] std::string get_readable_uuid() const;
 
+  [[nodiscard]] gtids::gtid get_gtid() const;
+
   [[nodiscard]] std::int64_t get_gno_raw() const noexcept { return gno_; }
+  [[nodiscard]] gtids::gno_t get_gno() const noexcept {
+    return static_cast<gtids::gno_t>(get_gno_raw());
+  }
 
   [[nodiscard]] std::uint8_t get_logical_ts_code_raw() const noexcept {
     return logical_ts_code_;
diff --git a/src/binsrv/event/gtid_tagged_log_body_impl.cpp b/src/binsrv/event/gtid_tagged_log_body_impl.cpp
@@ -38,6 +38,8 @@
 #include "binsrv/event/code_type.hpp"
 #include "binsrv/event/gtid_log_flag_type.hpp"
 
+#include "binsrv/gtids/gtid.hpp"
+#include "binsrv/gtids/tag.hpp"
 #include "binsrv/gtids/uuid.hpp"
 
 #include "util/bounded_string_storage.hpp"
@@ -160,11 +162,22 @@ generic_body_impl<code_type::gtid_tagged_log>::get_readable_uuid() const {
   return boost::lexical_cast<std::string>(get_uuid());
 }
 
+[[nodiscard]] gtids::tag
+generic_body_impl<code_type::gtid_tagged_log>::get_tag() const {
+  return gtids::tag{get_tag_raw()};
+}
+
 [[nodiscard]] std::string_view
-generic_body_impl<code_type::gtid_tagged_log>::get_tag() const noexcept {
+generic_body_impl<code_type::gtid_tagged_log>::get_readable_tag()
+    const noexcept {
   return util::to_string_view(get_tag_raw());
 }
 
+[[nodiscard]] gtids::gtid
+generic_body_impl<code_type::gtid_tagged_log>::get_gtid() const {
+  return {get_uuid(), get_tag(), get_gno()};
+}
+
 [[nodiscard]] std::string generic_body_impl<code_type::gtid_tagged_log>::
     get_readable_immediate_commit_timestamp() const {
   // threre is still no way to get string representationof the
@@ -204,7 +217,8 @@ operator<<(std::ostream &output,
            const generic_body_impl<code_type::gtid_tagged_log> &obj) {
   output << "flags: " << obj.get_readable_flags()
          << ", uuid: " << obj.get_readable_uuid()
-         << ", gno: " << obj.get_gno_raw() << ", tag: " << obj.get_tag()
+         << ", gno: " << obj.get_gno_raw()
+         << ", tag: " << obj.get_readable_tag()
          << ", last_committed: " << obj.get_last_committed_raw()
          << ", sequence_number: " << obj.get_sequence_number_raw();
 
diff --git a/src/binsrv/event/gtid_tagged_log_body_impl.hpp b/src/binsrv/event/gtid_tagged_log_body_impl.hpp
@@ -18,13 +18,16 @@
 
 #include "binsrv/event/gtid_tagged_log_body_impl_fwd.hpp" // IWYU pragma: export
 
+#include <chrono>
 #include <cstddef>
 #include <cstdint>
 
 #include "binsrv/event/gtid_log_flag_type_fwd.hpp"
 
 #include "binsrv/gtids/common_types.hpp"
-#include "binsrv/gtids/uuid.hpp"
+#include "binsrv/gtids/gtid_fwd.hpp"
+#include "binsrv/gtids/tag_fwd.hpp"
+#include "binsrv/gtids/uuid_fwd.hpp"
 
 #include "util/bounded_string_storage.hpp"
 #include "util/byte_span_fwd.hpp"
@@ -49,11 +52,17 @@ template <> class [[nodiscard]] generic_body_impl<code_type::gtid_tagged_log> {
   [[nodiscard]] std::string get_readable_uuid() const;
 
   [[nodiscard]] std::int64_t get_gno_raw() const noexcept { return gno_; }
+  [[nodiscard]] gtids::gno_t get_gno() const noexcept {
+    return static_cast<gtids::gno_t>(get_gno_raw());
+  }
 
   [[nodiscard]] const gtids::tag_storage &get_tag_raw() const noexcept {
     return tag_;
   }
-  [[nodiscard]] std::string_view get_tag() const noexcept;
+  [[nodiscard]] gtids::tag get_tag() const;
+  [[nodiscard]] std::string_view get_readable_tag() const noexcept;
+
+  [[nodiscard]] gtids::gtid get_gtid() const;
 
   [[nodiscard]] std::int64_t get_last_committed_raw() const noexcept {
     return last_committed_;
diff --git a/src/binsrv/event/reader_context.cpp b/src/binsrv/event/reader_context.cpp
diff --git a/src/binsrv/event/reader_context.hpp b/src/binsrv/event/reader_context.hpp
