PS-10244 feature: Add configuration options for specifying GTID-based / position-based replication mode

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

'replication' section of the main configuration file extended with one more required
string 'mode' configuration parameter, which can be either 'position' or 'gtid'. At this
moment only 'position' is supported.

'replication_config' class moved from the 'easymysql' namespace to 'binsrv'.

Added 'binbsrv::storage_metadata' class that is intended to keep persistent info for
binlog storages. Currently it has only one attribute 'mode' (either 'position' or 'gtid')
that indicates in which replication mode this storage was created. This metadata is
stored in the storage folder (along with binlog index and binlog data files) under the
'metadata.json' name. In addition, for forward compatibility, this metadata file also
includes the 'version' parameter that currently is alwas set to / expected to be  '1'.

'binsrv::storage' constructor extended with additional logic to read / write and
validate the content of the metadata file.

Similarly to 'util/nv_tuple_from_json.hpp' file that reads from JSON to 'util::nv_tuple',
added 'util/nv_tuple_to_json.hpp' that writes data from 'util::nv_tuple' to JSON.

Main application now prints to the log file replication mode from the configuration
file along with the one extracted from the storage metadata.

Re-factored the way how classes report their string-convertibility: now each class
interested in specializing 'util::is_string_convertable' template does this in its own header. In order to minimize dependencies, 'util::is_string_convertable' definition
extracted into separate header 'util/nv_tuple_json_support.hpp'.

We no longer use 'boost::json' as a header only library (by including
'boost/json/src.hpp'). Instead, we now use 'libboost_json.a' static library.

Updated 'set_up_binsrv_environment.inc' MTR include file - we now explicitly set
'<replication.mode>' to 'position' in the generated configuration file.

'main_config.json' sample configuration file updated with new '<replication.mode>'
parameter.

Updated 'README.md' with the '<replication.mode>' parameter description.

Author: percona-ysorokin

CMakeLists.txt

@@ -72,7 +72,7 @@ if(WITH_ASAN)
   target_link_options(binlog_server_compiler_flags INTERFACE "-fsanitize=address")
 endif()
 
-find_package(Boost 1.88.0 EXACT REQUIRED COMPONENTS url)
+find_package(Boost 1.88.0 EXACT REQUIRED COMPONENTS url json)
 
 find_package(MySQL REQUIRED)
 
@@ -252,6 +252,12 @@ set(source_files
   src/binsrv/operation_mode_type_fwd.hpp
   src/binsrv/operation_mode_type.hpp
 
+  src/binsrv/replication_config_fwd.hpp
+  src/binsrv/replication_config.hpp
+
+  src/binsrv/replication_mode_type_fwd.hpp
+  src/binsrv/replication_mode_type.hpp
+
   src/binsrv/size_unit_fwd.hpp
   src/binsrv/size_unit.hpp
   src/binsrv/size_unit.cpp
@@ -277,6 +283,10 @@ set(source_files
   src/binsrv/storage_config_fwd.hpp
   src/binsrv/storage_config.hpp
 
+  src/binsrv/storage_metadata_fwd.hpp
+  src/binsrv/storage_metadata.hpp
+  src/binsrv/storage_metadata.cpp
+
   src/binsrv/time_unit_fwd.hpp
   src/binsrv/time_unit.hpp
   src/binsrv/time_unit.cpp
@@ -316,7 +326,9 @@ set(source_files
 
   src/util/nv_tuple_fwd.hpp
   src/util/nv_tuple.hpp
+  src/util/nv_tuple_json_support.hpp
   src/util/nv_tuple_from_json.hpp
+  src/util/nv_tuple_to_json.hpp
 
   src/util/redirectable.hpp
 
@@ -343,9 +355,6 @@ set(source_files
   src/easymysql/library.hpp
   src/easymysql/library.cpp
 
-  src/easymysql/replication_config_fwd.hpp
-  src/easymysql/replication_config.hpp
-
   src/easymysql/ssl_config_fwd.hpp
   src/easymysql/ssl_config.hpp
 
@@ -360,7 +369,7 @@ add_executable(binlog_server ${source_files})
 target_link_libraries(binlog_server
   PRIVATE
     binlog_server_compiler_flags
-    Boost::headers Boost::url
+    Boost::headers Boost::json Boost::url
     ZLIB::ZLIB
     MySQL::client
     aws-cpp-sdk-s3-crt

README.md

@@ -217,7 +217,8 @@ The Percona Binary Log Server configuration file has the following format.
   "replication": {
     "server_id": 42,
     "idle_time": 10,
-    "verify_checksum": true
+    "verify_checksum": true,
+    "mode": "position"
   },
   "storage": {
     "backend": "s3",
@@ -271,6 +272,7 @@ Currently we use the following mapping:
 - `<replication.server_id>` - specifies the server ID that the utility will be using when connecting to a remote MySQL server (similar to [--connection-server-id](https://dev.mysql.com/doc/refman/8.0/en/mysqlbinlog.html#option_mysqlbinlog_connection-server-id) `mysqlbinlog` command line option).
 - `<replication.idle_time>` - the number of seconds the utility will spend in disconnected mode between reconnection attempts.
 - `<replication.verify_checksum>` - a boolean value which specifies whether the utility should verify event checksums.
+- `<replication.mode>` - the replication mode, can be either `position` for position-based replication or `gtid` for GTID-based replication.
 
 #### \<storage\> section
 - `<storage.backend>` - the type of the storage where the received binary logs should be stored:

main_config.json

@@ -29,7 +29,8 @@
   "replication": {
     "server_id": 42,
     "idle_time": 10,
-    "verify_checksum": true
+    "verify_checksum": true,
+    "mode": "position"
   },
   "storage": {
     "backend": "file",

mtr/binlog_streaming/include/set_up_binsrv_environment.inc

@@ -81,7 +81,8 @@ eval SET @binsrv_config_json = JSON_OBJECT(
   'replication', JSON_OBJECT(
      'server_id', @@server_id + 1,
      'idle_time', $binsrv_idle_time,
-     'verify_checksum', $binsrv_verify_checksum
+     'verify_checksum', $binsrv_verify_checksum,
+     'mode', 'position'
   ),
   'storage', JSON_OBJECT(
      'backend', '$storage_backend',

src/app.cpp

@@ -44,6 +44,7 @@
 #include "binsrv/logger_factory.hpp"
 #include "binsrv/main_config.hpp"
 #include "binsrv/operation_mode_type.hpp"
+#include "binsrv/replication_mode_type.hpp"
 #include "binsrv/size_unit.hpp"
 #include "binsrv/storage.hpp"
 // needed for storage_backend_type's operator <<
@@ -194,14 +195,16 @@ void log_connection_config_info(
 
 void log_replication_config_info(
     binsrv::basic_logger &logger,
-    const easymysql::replication_config &replication_config) {
+    const binsrv::replication_config &replication_config) {
 
   log_config_param<"server_id">(logger, replication_config,
                                 "mysql replication server id");
   log_config_param<"idle_time">(logger, replication_config,
                                 "mysql replication idle time (seconds)");
   log_config_param<"verify_checksum">(
       logger, replication_config, "mysql replication checksum verification");
+  log_config_param<"mode">(logger, replication_config,
+                           "mysql replication mode");
 }
 
 void log_storage_config_info(binsrv::basic_logger &logger,
@@ -225,6 +228,12 @@ void log_storage_info(binsrv::basic_logger &logger,
   msg += storage.get_backend_description();
   logger.log(binsrv::log_severity::info, msg);
 
+  msg.clear();
+  msg = "binlog storage initialized in ";
+  msg += boost::lexical_cast<std::string>(storage.get_replication_mode());
+  msg += " mode";
+  logger.log(binsrv::log_severity::info, msg);
+
   msg.clear();
   if (storage.has_current_binlog_name()) {
     msg = "binlog storage initialized at \"";
@@ -375,7 +384,8 @@ 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::storage &storage) {
+    std::uint32_t server_id, bool verify_checksum,
+    binsrv::replication_mode_type replication_mode, binsrv::storage &storage) {
   easymysql::connection connection{};
   try {
     connection = mysql_lib.create_connection(connection_config);
@@ -406,9 +416,10 @@ void receive_binlog_events(
           : easymysql::connection_replication_mode_type::blocking};
 
   try {
-    connection.switch_to_replication(server_id, current_binlog_name,
-                                     current_binlog_position, verify_checksum,
-                                     blocking_mode);
+    connection.switch_to_replication(
+        server_id, current_binlog_name, current_binlog_position,
+        verify_checksum,
+        replication_mode == binsrv::replication_mode_type::gtid, blocking_mode);
   } catch (const easymysql::core_error &) {
     if (operation_mode == binsrv::operation_mode_type::fetch) {
       throw;
@@ -419,7 +430,8 @@ void receive_binlog_events(
 
   logger.log(binsrv::log_severity::info,
              std::string{"switched to replication (checksum "} +
-                 (verify_checksum ? "enabled" : "disabled") + ")");
+                 (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);
@@ -431,7 +443,7 @@ void receive_binlog_events(
   util::const_byte_span portion;
 
   binsrv::event::reader_context context{connection.get_server_version(),
-                                        verify_checksum};
+                                        verify_checksum, replication_mode};
 
   // if binlog is still open, there is no sense to close it and re-open
   // instead, we will just instruct this loop to process the
@@ -605,9 +617,6 @@ int main(int argc, char *argv[]) {
     const auto &storage_config = config->root().get<"storage">();
     log_storage_config_info(*logger, storage_config);
 
-    binsrv::storage storage{storage_config};
-    log_storage_info(*logger, storage);
-
     const auto &connection_config = config->root().get<"connection">();
     log_connection_config_info(*logger, connection_config);
 
@@ -617,6 +626,10 @@ int main(int argc, char *argv[]) {
     const auto server_id{replication_config.get<"server_id">()};
     const auto idle_time_seconds{replication_config.get<"idle_time">()};
     const auto verify_checksum{replication_config.get<"verify_checksum">()};
+    const auto replication_mode{replication_config.get<"mode">()};
+
+    binsrv::storage storage{storage_config, replication_mode};
+    log_storage_info(*logger, storage);
 
     const easymysql::library mysql_lib;
     logger->log(binsrv::log_severity::info, "initialized mysql client library");
@@ -625,7 +638,7 @@ int main(int argc, char *argv[]) {
 
     receive_binlog_events(operation_mode, termination_flag, *logger, mysql_lib,
                           connection_config, server_id, verify_checksum,
-                          storage);
+                          replication_mode, storage);
 
     if (operation_mode == binsrv::operation_mode_type::pull) {
       std::size_t iteration_number{1U};
@@ -646,7 +659,7 @@ int main(int argc, char *argv[]) {
 
         receive_binlog_events(operation_mode, termination_flag, *logger,
                               mysql_lib, connection_config, server_id,
-                              verify_checksum, storage);
+                              verify_checksum, replication_mode, storage);
         ++iteration_number;
       }
     }

src/binsrv/event/reader_context.cpp

@@ -22,6 +22,8 @@
 #include <stdexcept>
 #include <utility>
 
+#include "binsrv/replication_mode_type_fwd.hpp"
+
 #include "binsrv/event/code_type.hpp"
 #include "binsrv/event/common_header_flag_type.hpp"
 #include "binsrv/event/event.hpp"
@@ -33,9 +35,10 @@
 namespace binsrv::event {
 
 reader_context::reader_context(std::uint32_t encoded_server_version,
-                               bool verify_checksum)
+                               bool verify_checksum,
+                               replication_mode_type replication_mode)
     : encoded_server_version_{encoded_server_version},
-      verify_checksum_{verify_checksum},
+      verify_checksum_{verify_checksum}, replication_mode_{replication_mode},
       post_header_lengths_{
           get_hardcoded_post_header_lengths(encoded_server_version_)} {}
 

src/binsrv/event/reader_context.hpp

@@ -20,6 +20,8 @@
 
 #include <cstdint>
 
+#include "binsrv/replication_mode_type_fwd.hpp"
+
 #include "binsrv/event/common_header_fwd.hpp"
 #include "binsrv/event/event_fwd.hpp"
 #include "binsrv/event/protocol_traits.hpp"
@@ -30,7 +32,8 @@ class [[nodiscard]] reader_context {
   friend class event;
 
 public:
-  reader_context(std::uint32_t encoded_server_version, bool verify_checksum);
+  reader_context(std::uint32_t encoded_server_version, bool verify_checksum,
+                 replication_mode_type replication_mode);
 
   [[nodiscard]] std::uint32_t
   get_current_encoded_server_version() const noexcept {
@@ -57,6 +60,7 @@ class [[nodiscard]] reader_context {
   state_type state_{state_type::initial};
   std::uint32_t encoded_server_version_;
   bool verify_checksum_;
+  replication_mode_type replication_mode_;
   post_header_length_container post_header_lengths_{};
   std::uint32_t position_{0U};
 

src/binsrv/log_severity_fwd.hpp

@@ -20,6 +20,8 @@
 #include <cstdint>
 #include <iosfwd>
 
+#include "util/nv_tuple_json_support.hpp"
+
 namespace binsrv {
 
 enum class log_severity : std::uint8_t;
@@ -36,4 +38,7 @@ operator>>(std::basic_istream<Char, Traits> &input, log_severity &level);
 
 } // namespace binsrv
 
+template <>
+struct util::is_string_convertable<binsrv::log_severity> : std::true_type {};
+
 #endif // BINSRV_LOG_SEVERITY_FWD_HPP

src/binsrv/main_config.cpp

@@ -22,41 +22,21 @@
 #include <stdexcept>
 #include <string>
 #include <string_view>
-#include <type_traits>
 
 #include <boost/json/parse.hpp>
-#include <boost/json/src.hpp> // IWYU pragma: keep
 
 // Needed for log_severity's operator <<
 #include "binsrv/log_severity.hpp" // IWYU pragma: keep
-#include "binsrv/size_unit_fwd.hpp"
+// Needed for replication_mode_type's operator <<
+#include "binsrv/replication_mode_type.hpp" // IWYU pragma: keep
 // Needed for storage_backend_type's operator <<
 #include "binsrv/storage_backend_type.hpp" // IWYU pragma: keep
-#include "binsrv/time_unit_fwd.hpp"
 
 // Needed for ssl_mode_type's operator <<
 #include "easymysql/ssl_mode_type.hpp" // IWYU pragma: keep
 
 #include "util/exception_location_helpers.hpp"
 #include "util/nv_tuple_from_json.hpp"
-#include "util/nv_tuple_fwd.hpp"
-
-// As this file is the only place where we call util::nv_tuple_from_json(),
-// which in turns calls tag_invoke() from the boost::json, we give the
-// serialization/unserialization hints for this library here (instead of
-// polluting each individual class definition).
-template <>
-struct util::is_string_convertable<binsrv::size_unit> : std::true_type {};
-template <>
-struct util::is_string_convertable<binsrv::time_unit> : std::true_type {};
-template <>
-struct util::is_string_convertable<binsrv::log_severity> : std::true_type {};
-template <>
-struct util::is_string_convertable<binsrv::storage_backend_type>
-    : std::true_type {};
-template <>
-struct util::is_string_convertable<easymysql::ssl_mode_type> : std::true_type {
-};
 
 namespace binsrv {
 

src/binsrv/main_config.hpp

@@ -19,10 +19,10 @@
 #include "binsrv/main_config_fwd.hpp" // IWYU pragma: export
 
 #include "binsrv/logger_config.hpp"  // IWYU pragma: export
+#include "binsrv/replication_config.hpp" // IWYU pragma: export
 #include "binsrv/storage_config.hpp" // IWYU pragma: export
 
-#include "easymysql/connection_config.hpp"  // IWYU pragma: export
-#include "easymysql/replication_config.hpp" // IWYU pragma: export
+#include "easymysql/connection_config.hpp" // IWYU pragma: export
 
 #include "util/nv_tuple.hpp"
 
@@ -34,7 +34,7 @@ class [[nodiscard]] main_config {
       // clang-format off
       util::nv<"logger"     , logger_config>,
       util::nv<"connection" , easymysql::connection_config>,
-      util::nv<"replication", easymysql::replication_config>,
+      util::nv<"replication", binsrv::replication_config>,
       util::nv<"storage"    , storage_config>
       // clang-format on
       >;