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

https://perconadev.atlassian.net/browse/PS-10245

Refactored 'easymysql::connection::switch_to_replication()' method. We now have 3 different methods:
- one 'switch_to_gtid_replication()' for GTID mode
- two 'switch_to_position_replication()' for position-based mode (one for starting from the very beginning and another one when binlog name / position are known).

In the main application we now print more info in 'log_replication_info()' depending on
the mode (GTID or position-mased).

Author: percona-ysorokin

src/app.cpp

@@ -51,10 +51,12 @@
 #include "binsrv/storage_backend_type.hpp" // IWYU pragma: keep
 #include "binsrv/time_unit.hpp"
 
+#include "binsrv/gtids/common_types.hpp"
+#include "binsrv/gtids/gtid_set.hpp"
+
 #include "binsrv/event/code_type.hpp"
 #include "binsrv/event/common_header_flag_type.hpp"
 #include "binsrv/event/event.hpp"
-#include "binsrv/event/protocol_traits_fwd.hpp"
 #include "binsrv/event/reader_context.hpp"
 
 #include "easymysql/connection.hpp"
@@ -276,19 +278,42 @@ void log_connection_info(binsrv::basic_logger &logger,
 
 void log_replication_info(
     binsrv::basic_logger &logger, std::uint32_t server_id,
-    std::string_view current_binlog_name, std::uint64_t current_binlog_position,
+    const binsrv::storage &storage, bool verify_checksum,
     easymysql::connection_replication_mode_type blocking_mode) {
-  std::string msg{};
+  const auto replication_mode{storage.get_replication_mode()};
+
+  std::string msg{"switched to replication (checksum "};
+  msg += (verify_checksum ? "enabled" : "disabled");
+  msg += ", ";
+  msg += boost::lexical_cast<std::string>(replication_mode);
+  msg += +" mode)";
+  logger.log(binsrv::log_severity::info, msg);
+
   msg = "replication info (server id ";
   msg += std::to_string(server_id);
   msg += ", ";
-  msg += (current_binlog_name.empty() ? "<empty>" : current_binlog_name);
-  msg += ":";
-  msg += std::to_string(current_binlog_position);
-  msg += ", ";
   msg += (blocking_mode == easymysql::connection_replication_mode_type::blocking
               ? "blocking"
               : "non-blocking");
+  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 {
+    const auto &gtids{storage.get_gtids()};
+    if (gtids.is_empty()) {
+      msg += "an empty";
+    } else {
+      msg += "the ";
+      msg += boost::lexical_cast<std::string>(gtids);
+    }
+    msg += " GTID set";
+  }
   msg += ")";
   logger.log(binsrv::log_severity::info, msg);
 }
@@ -384,8 +409,7 @@ void receive_binlog_events(
     const volatile std::atomic_flag &termination_flag,
     binsrv::basic_logger &logger, const easymysql::library &mysql_lib,
     const easymysql::connection_config &connection_config,
-    std::uint32_t server_id, bool verify_checksum,
-    binsrv::replication_mode_type replication_mode, binsrv::storage &storage) {
+    std::uint32_t server_id, bool verify_checksum, binsrv::storage &storage) {
   easymysql::connection connection{};
   try {
     connection = mysql_lib.create_connection(connection_config);
@@ -403,23 +427,34 @@ void receive_binlog_events(
 
   log_connection_info(logger, connection);
 
-  const auto current_binlog_name{storage.get_current_binlog_name()};
-  // if storage binlog name is detected to be empty (empty data directory), we
-  // start streaming from the position 'magic_binlog_offset' (4)
-  const auto current_binlog_position{current_binlog_name.empty()
-                                         ? binsrv::event::magic_binlog_offset
-                                         : storage.get_current_position()};
-
+  const auto replication_mode{storage.get_replication_mode()};
   const auto blocking_mode{
       operation_mode == binsrv::operation_mode_type::fetch
           ? easymysql::connection_replication_mode_type::non_blocking
           : easymysql::connection_replication_mode_type::blocking};
 
   try {
-    connection.switch_to_replication(
-        server_id, current_binlog_name, current_binlog_position,
-        verify_checksum,
-        replication_mode == binsrv::replication_mode_type::gtid, blocking_mode);
+    if (replication_mode == binsrv::replication_mode_type::position) {
+      if (storage.has_current_binlog_name()) {
+        connection.switch_to_position_replication(
+            server_id, storage.get_current_binlog_name(),
+            storage.get_current_position(), verify_checksum, blocking_mode);
+      } else {
+        connection.switch_to_position_replication(server_id, verify_checksum,
+                                                  blocking_mode);
+      }
+    } else {
+      const binsrv::gtids::gtid_set empty_gtids{};
+      const auto encoded_size{empty_gtids.calculate_encoded_size()};
+
+      binsrv::gtids::gtid_set_storage buffer(encoded_size);
+      util::byte_span destination{buffer};
+      empty_gtids.encode_to(destination);
+
+      connection.switch_to_gtid_replication(server_id,
+                                            util::const_byte_span{buffer},
+                                            verify_checksum, blocking_mode);
+    }
   } catch (const easymysql::core_error &) {
     if (operation_mode == binsrv::operation_mode_type::fetch) {
       throw;
@@ -428,13 +463,8 @@ void receive_binlog_events(
     return;
   }
 
-  logger.log(binsrv::log_severity::info,
-             std::string{"switched to replication (checksum "} +
-                 (verify_checksum ? "enabled" : "disabled") + ", " +
-                 boost::lexical_cast<std::string>(replication_mode) + "mode)");
-
-  log_replication_info(logger, server_id, current_binlog_name,
-                       current_binlog_position, blocking_mode);
+  log_replication_info(logger, server_id, storage, verify_checksum,
+                       blocking_mode);
 
   // Network streams are requested with COM_BINLOG_DUMP and
   // each Binlog Event response is prepended with 00 OK-byte.
@@ -638,7 +668,7 @@ int main(int argc, char *argv[]) {
 
     receive_binlog_events(operation_mode, termination_flag, *logger, mysql_lib,
                           connection_config, server_id, verify_checksum,
-                          replication_mode, storage);
+                          storage);
 
     if (operation_mode == binsrv::operation_mode_type::pull) {
       std::size_t iteration_number{1U};
@@ -659,7 +689,7 @@ int main(int argc, char *argv[]) {
 
         receive_binlog_events(operation_mode, termination_flag, *logger,
                               mysql_lib, connection_config, server_id,
-                              verify_checksum, replication_mode, storage);
+                              verify_checksum, storage);
         ++iteration_number;
       }
     }

src/binsrv/storage.hpp

@@ -27,6 +27,8 @@
 #include "binsrv/replication_mode_type_fwd.hpp"
 #include "binsrv/storage_config_fwd.hpp"
 
+#include "binsrv/gtids/gtid_set.hpp"
+
 #include "util/byte_span_fwd.hpp"
 
 namespace binsrv {
@@ -65,6 +67,10 @@ class [[nodiscard]] storage {
     return position_;
   }
 
+  [[nodiscard]] const gtids::gtid_set &get_gtids() const noexcept {
+    return gtids_;
+  }
+
   [[nodiscard]] static bool
   check_binlog_name(std::string_view binlog_name) noexcept;
 
@@ -81,6 +87,8 @@ class [[nodiscard]] storage {
   binlog_name_container binlog_names_;
   std::uint64_t position_{0ULL};
 
+  gtids::gtid_set gtids_{};
+
   std::uint64_t checkpoint_size_bytes_{0ULL};
   std::uint64_t last_checkpoint_position_{0ULL};
 

src/easymysql/connection.cpp

@@ -17,6 +17,7 @@
 
 #include <cassert>
 #include <concepts>
+#include <cstddef>
 #include <cstdint>
 #include <memory>
 #include <optional>
@@ -113,6 +114,10 @@ void connection::mysql_deleter::operator()(void *ptr) const noexcept {
 
 class connection::rpl_impl {
 public:
+  // https://github.com/mysql/mysql-server/blob/mysql-8.0.43/sql/log_event.h#L216
+  // https://github.com/mysql/mysql-server/blob/mysql-8.4.6/sql/log_event.h#L214
+  static constexpr std::uint64_t default_binlog_position{4U};
+
   rpl_impl(connection &conn, std::uint32_t server_id,
            std::string_view file_name, std::uint64_t position,
            connection_replication_mode_type blocking_mode)
@@ -121,14 +126,39 @@ class connection::rpl_impl {
              .file_name = std::data(file_name),
              .start_position = position,
              .server_id = server_id,
-             .flags = get_rpl_flags(blocking_mode),
+             .flags = get_rpl_flags(false, blocking_mode),
              .gtid_set_encoded_size = 0U,
              .fix_gtid_set = nullptr,
              .gtid_set_arg = nullptr,
              .size = 0U,
              .buffer = nullptr} {
     if (mysql_binlog_open(conn_, &rpl_) != 0) {
-      raise_core_error_from_connection("cannot open binary log", conn);
+      raise_core_error_from_connection(
+          "cannot open binary log in position mode", conn);
+    }
+  }
+  rpl_impl(connection &conn, std::uint32_t server_id,
+           util::const_byte_span encoded_gtid_set,
+           connection_replication_mode_type blocking_mode)
+      : conn_{mysql_deimpl::get(conn.mysql_impl_)},
+        rpl_{.file_name_length = 0U,
+             .file_name = nullptr,
+             .start_position = default_binlog_position,
+             .server_id = server_id,
+             .flags = get_rpl_flags(true, blocking_mode),
+             .gtid_set_encoded_size = std::size(encoded_gtid_set),
+             .fix_gtid_set = nullptr,
+             // it is OK to use const_cast here as MySQL client uses this
+             // pointer only for reading (checked on MySQL 8.4.6)
+             // NOLINTBEGIN(cppcoreguidelines-pro-type-const-cast)
+             .gtid_set_arg =
+                 const_cast<std::byte *>(std::data(encoded_gtid_set)),
+             // NOLINTEND(cppcoreguidelines-pro-type-const-cast)
+             .size = 0U,
+             .buffer = nullptr} {
+    if (mysql_binlog_open(conn_, &rpl_) != 0) {
+      raise_core_error_from_connection("cannot open binary log in GTID mode",
+                                       conn);
     }
   }
 
@@ -170,8 +200,9 @@ class connection::rpl_impl {
   // locally
   static constexpr unsigned int private_binlog_dump_non_block{1U};
   [[nodiscard]] static constexpr unsigned int
-  get_rpl_flags(connection_replication_mode_type blocking_mode) noexcept {
-    return MYSQL_RPL_SKIP_HEARTBEAT |
+  get_rpl_flags(bool gtid_mode,
+                connection_replication_mode_type blocking_mode) noexcept {
+    return MYSQL_RPL_SKIP_HEARTBEAT | (gtid_mode ? MYSQL_RPL_GTID : 0U) |
            (blocking_mode == connection_replication_mode_type::non_blocking
                 ? private_binlog_dump_non_block
                 : 0U);
@@ -180,6 +211,17 @@ class connection::rpl_impl {
 
 connection::connection() noexcept = default;
 
+void connection::set_binlog_checksum(bool verify_checksum) {
+  // WL#2540: Replication event checksums
+  // https://dev.mysql.com/worklog/task/?id=2540
+  const std::string checksum_algorithm_label{verify_checksum ? "CRC32"
+                                                             : "NONE"};
+  const auto set_binlog_checksum_query{
+      "SET @source_binlog_checksum = '" + checksum_algorithm_label +
+      "', @master_binlog_checksum = '" + checksum_algorithm_label + "'"};
+  execute_generic_query_noresult(set_binlog_checksum_query);
+}
+
 void connection::process_ssl_config(const ssl_config &config) {
   auto *casted_impl = mysql_deimpl::get(mysql_impl_);
 
@@ -402,30 +444,39 @@ bool connection::ping() {
   return mysql_ping(casted_impl) == 0;
 }
 
-void connection::switch_to_replication(
+void connection::switch_to_position_replication(
     std::uint32_t server_id, std::string_view file_name, std::uint64_t position,
-    bool verify_checksum, bool gtid_mode,
-    connection_replication_mode_type blocking_mode) {
+    bool verify_checksum, connection_replication_mode_type blocking_mode) {
   assert(!is_empty());
-  if (gtid_mode) {
+  if (is_in_replication_mode()) {
     util::exception_location().raise<std::logic_error>(
-        "switching to GTID replication is not yet implemented");
+        "connection has already been swithed to replication");
   }
+
+  set_binlog_checksum(verify_checksum);
+  rpl_impl_ = std::make_unique<rpl_impl>(*this, server_id, file_name, position,
+                                         blocking_mode);
+}
+
+void connection::switch_to_position_replication(
+    std::uint32_t server_id, bool verify_checksum,
+    connection_replication_mode_type blocking_mode) {
+  switch_to_position_replication(server_id, {},
+                                 rpl_impl::default_binlog_position,
+                                 verify_checksum, blocking_mode);
+}
+
+void connection::switch_to_gtid_replication(
+    std::uint32_t server_id, util::const_byte_span encoded_gtid_set,
+    bool verify_checksum, connection_replication_mode_type blocking_mode) {
+  assert(!is_empty());
   if (is_in_replication_mode()) {
     util::exception_location().raise<std::logic_error>(
         "connection has already been swithed to replication");
   }
 
-  // WL#2540: Replication event checksums
-  // https://dev.mysql.com/worklog/task/?id=2540
-  const std::string checksum_algorithm_label{verify_checksum ? "CRC32"
-                                                             : "NONE"};
-  const auto set_binlog_checksum_query{
-      "SET @source_binlog_checksum = '" + checksum_algorithm_label +
-      "', @master_binlog_checksum = '" + checksum_algorithm_label + "'"};
-  execute_generic_query_noresult(set_binlog_checksum_query);
-
-  rpl_impl_ = std::make_unique<rpl_impl>(*this, server_id, file_name, position,
+  set_binlog_checksum(verify_checksum);
+  rpl_impl_ = std::make_unique<rpl_impl>(*this, server_id, encoded_gtid_set,
                                          blocking_mode);
 }
 

src/easymysql/connection.hpp

@@ -64,10 +64,18 @@ class [[nodiscard]] connection {
     return static_cast<bool>(rpl_impl_);
   }
 
-  void switch_to_replication(std::uint32_t server_id,
-                             std::string_view file_name, std::uint64_t position,
-                             bool verify_checksum, bool gtid_mode,
-                             connection_replication_mode_type blocking_mode);
+  void switch_to_position_replication(
+      std::uint32_t server_id, std::string_view file_name,
+      std::uint64_t position, bool verify_checksum,
+      connection_replication_mode_type blocking_mode);
+  // a simplified version for starting from the very beginning
+  void switch_to_position_replication(
+      std::uint32_t server_id, bool verify_checksum,
+      connection_replication_mode_type blocking_mode);
+
+  void switch_to_gtid_replication(
+      std::uint32_t server_id, util::const_byte_span encoded_gtid_set,
+      bool verify_checksum, connection_replication_mode_type blocking_mode);
 
   // returns false on 'connection closed' / 'timeout'
   // returns true and sets 'portion' to en empty span on EOF (last event read)
@@ -76,6 +84,8 @@ class [[nodiscard]] connection {
   [[nodiscard]] bool fetch_binlog_event(util::const_byte_span &portion);
 
 private:
+  void set_binlog_checksum(bool verify_checksum);
+
   void process_ssl_config(const ssl_config &config);
   void process_tls_config(const tls_config &config);
   void process_connection_config(const connection_config &config);