WL#7353 Recover relay log by truncating half written transactions [2/3]

Step 2: Relay log sanitization

This step implements relay log sanitization. Sanitization process runs during
applier metadata initialization, in case relay-log-recovery server option is
turned off. It is implemented in a separate class, 'Relay_log_sanitizer'.
Whenever a partially-written transaction is encountered at the end of the file,
sanitizer performs truncation of the relay log file via execution of the
existing 'truncate_update_log_file' function. After the relay log truncation
and loading information stored in applier metadata on disk, relay log sanitizer
updates the position of the receiver thread. In case no
valid source position could be obtained from the existing relay
log files, sanitizer matches the receiver thread position with the
applier source position read from the applier metadata.

Change-Id: I13203812b2d73f38fca711473097daed55721f00

Author: Karolina Szczepankiewicz

sql/CMakeLists.txt

@@ -1108,6 +1108,7 @@ SET (RPL_REPLICA_SRCS
   rpl_reporting.cc
   rpl_rli.cc
   rpl_rli_pdb.cc
+  rpl_relay_log_sanitizer.cc
   rpl_replica.cc
   rpl_replica_commit_order_manager.cc
   rpl_replica_until_options.cc

sql/binlog.h

@@ -845,6 +845,8 @@ class MYSQL_BIN_LOG : public TC_LOG {
 
  private:
   bool after_write_to_relay_log(Master_info *mi);
+
+ public:
   /**
    * Truncte log file and clear LOG_EVENT_BINLOG_IN_USE_F when update is set.
    * @param[in] log_name name of the log file to be trunacted
@@ -858,7 +860,6 @@ class MYSQL_BIN_LOG : public TC_LOG {
   bool truncate_update_log_file(const char *log_name, my_off_t valid_pos,
                                 my_off_t binlog_size, bool update);
 
- public:
   void make_log_name(char *buf, const char *log_ident);
   bool is_active(const char *log_file_name) const;
 

sql/rpl_mi.h

@@ -88,6 +88,11 @@ class Master_info : public Rpl_info {
   friend class Rpl_info_factory;
 
  public:
+  /// In case we start replication from the first binary log file and
+  /// source log name is empty, we use first_source_log_name instead of
+  /// 'master_log_name' in the error log
+  static constexpr auto first_source_log_name = "<First log>";
+
   /**
     Host name or ip address stored in the master.info.
   */

sql/rpl_relay_log_sanitizer.cc

@@ -0,0 +1,92 @@
+// Copyright (c) 2024, Oracle and/or its affiliates.
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License, version 2.0,
+// as published by the Free Software Foundation.
+//
+// This program is designed to work with certain software (including
+// but not limited to OpenSSL) that is licensed under separate terms,
+// as designated in a particular file or component or in included license
+// documentation.  The authors of MySQL hereby grant you an additional
+// permission to link the program and your derivative works with the
+// separately licensed software that they have either included with
+// the program or referenced in the documentation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License, version 2.0, for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the Free Software
+// Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA.
+
+#include "sql/rpl_relay_log_sanitizer.h"
+#include "sql/binlog.h"
+#include "sql/binlog/decompressing_event_object_istream.h"  // Decompressing_event_object_istream
+#include "sql/binlog_reader.h"  // Relay_log_file_reader
+#include "sql/rpl_mi.h"         // Master_info
+
+namespace rpl {
+
+void Relay_log_sanitizer::analyze_logs(MYSQL_BIN_LOG &log,
+                                       bool checksum_validation) {
+  Relaylog_file_reader reader(checksum_validation);
+  this->process_logs(reader, log);
+}
+
+bool Relay_log_sanitizer::sanitize_log(MYSQL_BIN_LOG &log) {
+  std::stringstream ss;
+  if (!is_fatal_error() && is_log_truncation_needed()) {
+    ss << "Truncating " << get_valid_file()
+       << " to log position: " << get_valid_pos();
+    LogErr(INFORMATION_LEVEL, ER_LOG_SANITIZATION, ss.str().c_str());
+    return log.truncate_update_log_file(
+        get_valid_file().c_str(), get_valid_pos(), m_last_file_size, false);
+  }
+  if (is_fatal_error()) {
+    ss << "Skipping log sanitization due to: " << this->m_failure_message;
+    LogErr(INFORMATION_LEVEL, ER_LOG_SANITIZATION, ss.str().c_str());
+  }
+  return false;
+}
+
+void Relay_log_sanitizer::update_source_position(Master_info *mi) {
+  if (is_fatal_error()) return;
+  std::string new_source_file{""};
+  my_off_t new_source_pos{0};
+  if (!this->m_valid_source_file.empty()) {
+    // we are sure that we were able to obtain source position
+    assert(this->m_has_valid_source_pos);
+    new_source_file = m_valid_source_file;
+    new_source_pos = m_valid_source_pos;
+
+  } else if (this->m_has_valid_source_pos) {
+    // update only postion
+    new_source_pos = m_valid_source_pos;
+    new_source_file = mi->get_master_log_name();
+  } else {
+    // no valid position could have been recovered from the source file,
+    // setting to applier source position
+    new_source_pos = std::max<ulonglong>(BIN_LOG_HEADER_SIZE,
+                                         mi->rli->get_group_master_log_pos());
+    new_source_file = mi->rli->get_group_master_log_name();
+  }
+
+  if (mi->get_master_log_name() != new_source_file ||
+      mi->get_master_log_pos() != new_source_pos) {
+    std::stringstream ss;
+    std::string new_source_file_str = new_source_file.empty()
+                                          ? Master_info::first_source_log_name
+                                          : new_source_file;
+    ss << "Changing source log coordinates from: " << mi->get_io_rpl_log_name()
+       << "; " << mi->get_master_log_pos() << " to: " << new_source_file_str
+       << "; " << new_source_pos;
+    LogErr(INFORMATION_LEVEL, ER_LOG_SANITIZATION, ss.str().c_str());
+    // set position and filename
+    mi->set_master_log_pos(new_source_pos);
+    mi->set_master_log_name(new_source_file.c_str());
+  }
+}
+
+}  // namespace rpl

sql/rpl_relay_log_sanitizer.h

@@ -0,0 +1,87 @@
+// Copyright (c) 2024, Oracle and/or its affiliates.
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License, version 2.0,
+// as published by the Free Software Foundation.
+//
+// This program is designed to work with certain software (including
+// but not limited to OpenSSL) that is licensed under separate terms,
+// as designated in a particular file or component or in included license
+// documentation.  The authors of MySQL hereby grant you an additional
+// permission to link the program and your derivative works with the
+// separately licensed software that they have either included with
+// the program or referenced in the documentation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License, version 2.0, for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the Free Software
+// Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA.
+
+#ifndef RPL_RELAY_LOG_SANITIZER_H
+#define RPL_RELAY_LOG_SANITIZER_H
+
+#include <functional>
+#include "mysql/binlog/event/binlog_event.h"
+#include "sql/binlog.h"
+#include "sql/binlog/decompressing_event_object_istream.h"  // binlog::Decompressing_event_object_istream
+#include "sql/binlog/log_sanitizer.h"
+#include "sql/binlog_ostream.h"  // binlog::tools::Iterator
+#include "sql/binlog_reader.h"   // Binlog_file_reader
+#include "sql/log_event.h"       // Log_event
+#include "sql/xa.h"              // XID
+
+namespace rpl {
+
+/// @brief Class used to recover relay log files
+/// @details Recovery of the relay log files is:
+/// - finding the last valid position outside of a transaction boundary
+///   (analyze_logs)
+/// - removing relay logs appearing after the relay log with the last valid
+///   position (analyze_logs)
+/// - truncation of the relay log file containing the last valid position
+///   to remove partially written transaction from the log (sanitize log)
+class Relay_log_sanitizer : public binlog::Log_sanitizer {
+ public:
+  /// @brief Ctor
+  Relay_log_sanitizer() { this->m_validation_started = false; }
+
+  /// @brief Dtor
+  ~Relay_log_sanitizer() override = default;
+
+  /// @brief Given specific log, performs sanitization. Reads log list obtained
+  /// from the MYSQL_BIN_LOG object and searches for last, fully written
+  /// transaction. Removes log files that are created after last finished
+  /// transaction
+  /// @param log Handle to MYSQL_BIN_LOG object, which does not need to be
+  /// open. We need specific functions from the MYSQL_BIN_LOG, e.g.
+  /// reading of the index file
+  /// @param checksum_validation True if we need to perform relay log file
+  /// checksum validation
+  void analyze_logs(MYSQL_BIN_LOG &log, bool checksum_validation);
+
+  /// @brief Sanitize opened log
+  /// @param log Handle to MYSQL_BIN_LOG object, which we will truncate
+  /// if needed
+  /// @return false on no error or when no truncation was done, true otherwise
+  bool sanitize_log(MYSQL_BIN_LOG &log);
+
+  /// @brief Updates source position if a valid source position has been
+  /// found whilst reading the relay log files
+  /// @param mi Master_info for the receiver thread.
+  void update_source_position(Master_info *mi);
+
+ protected:
+  /// @brief Function used to obtain memory key for derived classes
+  /// @returns Reference to a memory key
+  PSI_memory_key &get_memory_key() const override {
+    return key_memory_relaylog_recovery;
+  }
+};
+
+}  // namespace rpl
+
+#endif  // RPL_RELAY_LOG_SANITIZER_H

sql/rpl_replica.cc

@@ -1259,10 +1259,9 @@ static inline int fill_mts_gaps_and_recover(Master_info *mi) {
   return recovery_error;
 }
 
-int load_mi_and_rli_from_repositories(Master_info *mi, bool ignore_if_no_info,
-                                      int thread_mask,
-                                      bool skip_received_gtid_set_recovery,
-                                      bool force_load) {
+int load_mi_and_rli_from_repositories(
+    Master_info *mi, bool ignore_if_no_info, int thread_mask,
+    bool skip_received_gtid_set_and_relaylog_recovery, bool force_load) {
   DBUG_TRACE;
   assert(mi != nullptr && mi->rli != nullptr);
   int init_error = 0;
@@ -1319,7 +1318,8 @@ int load_mi_and_rli_from_repositories(Master_info *mi, bool ignore_if_no_info,
   if (!ignore_if_no_info || check_return != REPOSITORY_DOES_NOT_EXIST) {
     if ((thread_mask & SLAVE_SQL) != 0 || !(mi->rli->inited)) {
       if (!mi->rli->inited || force_load) {
-        if (mi->rli->rli_init_info(skip_received_gtid_set_recovery)) {
+        if (mi->rli->rli_init_info(
+                skip_received_gtid_set_and_relaylog_recovery)) {
           init_error = 1;
         } else {
           /*

sql/rpl_replica.h

@@ -474,15 +474,16 @@ int init_recovery(Master_info *mi);
   are not yet initialized. When true this flag forces the repositories
   to load information from table or file.
 
-  @param skip_received_gtid_set_recovery When true, skips the received GTID
-                                         set recovery.
+  @param skip_received_gtid_set_and_relaylog_recovery When true, skips the
+  received GTID set and relay log recovery.
 
   @retval 0 Success
   @retval nonzero Error
 */
 int load_mi_and_rli_from_repositories(
     Master_info *mi, bool ignore_if_no_info, int thread_mask,
-    bool skip_received_gtid_set_recovery = false, bool force_load = false);
+    bool skip_received_gtid_set_and_relaylog_recovery = false,
+    bool force_load = false);
 void end_info(Master_info *mi);
 /**
   Clear the information regarding the `Master_info` and `Relay_log_info` objects

sql/rpl_rli.cc

@@ -64,6 +64,7 @@
 #include "sql/rpl_info_handler.h"
 #include "sql/rpl_mi.h"   // Master_info
 #include "sql/rpl_msr.h"  // channel_map
+#include "sql/rpl_relay_log_sanitizer.h"
 #include "sql/rpl_replica.h"
 #include "sql/rpl_reporting.h"
 #include "sql/rpl_rli_pdb.h"  // Slave_worker
@@ -1501,14 +1502,21 @@ bool mysql_show_relaylog_events(THD *thd) {
   return res;
 }
 
-int Relay_log_info::rli_init_info(bool skip_received_gtid_set_recovery) {
+int Relay_log_info::rli_init_info(
+    bool skip_received_gtid_set_and_relaylog_recovery) {
   int error = 0;
   enum_return_check check_return = ERROR_CHECKING_REPOSITORY;
   const char *msg = nullptr;
   DBUG_TRACE;
 
   mysql_mutex_assert_owner(&data_lock);
 
+  // Prepare for relay log sanitization
+  rpl::Relay_log_sanitizer relay_log_sanitizer;
+  bool execute_relay_log_sanitization =
+      !is_relay_log_recovery && !relay_log_sanitized &&
+      !skip_received_gtid_set_and_relaylog_recovery;
+
   /*
     If Relay_log_info is issued again after a failed init_info(), for
     instance because of missing relay log files, it will generate new
@@ -1647,6 +1655,14 @@ int Relay_log_info::rli_init_info(bool skip_received_gtid_set_recovery) {
       return 1;
     }
 
+    if (execute_relay_log_sanitization) {
+      relay_log_sanitizer.analyze_logs(relay_log,
+                                       opt_replica_sql_verify_checksum);
+      // sanitize relay_log if applicable, ignore errors, they will be reported
+      // later on
+      relay_log_sanitizer.sanitize_log(relay_log);
+    }
+
     if (!gtid_retrieved_initialized) {
       /* Store the GTID of a transaction spanned in multiple relay log files */
       Gtid_monitoring_info *partial_trx = mi->get_gtid_monitoring_info();
@@ -1656,6 +1672,7 @@ int Relay_log_info::rli_init_info(bool skip_received_gtid_set_recovery) {
       gtid_set->dbug_print("set of GTIDs in relay log before initialization");
       get_tsid_lock()->unlock();
 #endif
+
       /*
         In the init_gtid_set below we pass the mi->transaction_parser.
         This will be useful to ensure that we only add a GTID to
@@ -1669,7 +1686,7 @@ int Relay_log_info::rli_init_info(bool skip_received_gtid_set_recovery) {
         init_recovery.
       */
       if (!is_relay_log_recovery && !gtid_retrieved_initialized &&
-          !skip_received_gtid_set_recovery &&
+          !skip_received_gtid_set_and_relaylog_recovery &&
           relay_log.init_gtid_sets(
               gtid_set, nullptr, opt_replica_sql_verify_checksum,
               true /*true=need lock*/, &mi->transaction_parser, partial_trx)) {
@@ -1689,14 +1706,13 @@ int Relay_log_info::rli_init_info(bool skip_received_gtid_set_recovery) {
       correctly compute the set of previous gtids.
     */
     relay_log.set_previous_gtid_set_relaylog(gtid_set);
+
     /*
       note, that if open() fails, we'll still have index file open
       but a destructor will take care of that
     */
-
     mysql_mutex_t *log_lock = relay_log.get_log_lock();
     mysql_mutex_lock(log_lock);
-
     if (relay_log.open_binlog(
             ln, nullptr,
             (max_relay_log_size ? max_relay_log_size : max_binlog_size), true,
@@ -1706,7 +1722,6 @@ int Relay_log_info::rli_init_info(bool skip_received_gtid_set_recovery) {
       LogErr(ERROR_LEVEL, ER_RPL_CANT_OPEN_LOG_IN_AM_INIT_INFO);
       return 1;
     }
-
     mysql_mutex_unlock(log_lock);
   }
 
@@ -1830,6 +1845,12 @@ int Relay_log_info::rli_init_info(bool skip_received_gtid_set_recovery) {
     goto err;
   }
 
+  // applier metadata was read from disk, update receiver positions
+  if (execute_relay_log_sanitization) {
+    relay_log_sanitizer.update_source_position(mi);
+    relay_log_sanitized = true;
+  }
+
   /*
     In case of MTS the recovery is deferred until the end of
     load_mi_and_rli_from_repositories.

sql/rpl_rli.h

@@ -764,6 +764,9 @@ class Relay_log_info : public Rpl_info {
   /* Flag that ensures the retrieved GTID set is initialized only once. */
   bool gtid_retrieved_initialized;
 
+  /// Flag that ensures the relay log is sanitized only once.
+  bool relay_log_sanitized = false;
+
   /**
     Stores information on the last processed transaction or the transaction
     that is currently being processed.
@@ -1555,13 +1558,13 @@ class Relay_log_info : public Rpl_info {
     replication log files was ON and the keyring plugin is not available
     anymore.
 
-    @param skip_received_gtid_set_recovery When true, skips the received GTID
-                                           set recovery.
+    @param skip_received_gtid_set_and_relaylog_recovery When true, skips the
+    received GTID set and relay log recovery.
 
     @retval 0 Success.
     @retval 1 Error.
   */
-  int rli_init_info(bool skip_received_gtid_set_recovery = false);
+  int rli_init_info(bool skip_received_gtid_set_and_relaylog_recovery = false);
   void end_info();
 
   /** No flush options given to relay log flush */