Use the same PXB version for restoring the backup as was used for its creation

myhoard/basebackup_restore_operation.py

@@ -1,9 +1,9 @@
 # Copyright (c) 2019 Aiven, Helsinki, Finland. https://aiven.io/
 from .errors import DiskFullError
-from .util import get_xtrabackup_version, parse_version, parse_xtrabackup_info
+from .util import BinVersion, find_extra_xtrabackup_executables, get_xtrabackup_version, parse_version, parse_xtrabackup_info
 from contextlib import suppress
 from rohmu.util import increase_pipe_capacity, set_stream_nonblocking
-from typing import Final, Optional, Tuple
+from typing import Final, Optional
 
 import base64
 import fnmatch
@@ -101,22 +101,28 @@ def restore_backup(self):
                         self.proc = xbstream
                         self._process_xbstream_input_output()
 
+                backup_xtabackup_version = None
                 self.data_directory_size_start = self._get_directory_size(self.temp_dir)
                 xtrabackup_info_path = os.path.join(self.temp_dir, "xtrabackup_info")
                 if os.path.exists(xtrabackup_info_path):
                     with open(xtrabackup_info_path) as fh:
                         xtrabackup_info_text = fh.read()
                     self.backup_xtrabackup_info = parse_xtrabackup_info(xtrabackup_info_text)
+                    backup_xtabackup_raw_version = self.backup_xtrabackup_info.get("tool_version")
+                    backup_xtabackup_version = (
+                        parse_version(backup_xtabackup_raw_version) if backup_xtabackup_raw_version else None
+                    )
                     self.log.info(
                         "Backup info. Tool version: %s, Server version: %s",
-                        self.backup_xtrabackup_info.get("tool_version"),
+                        backup_xtabackup_raw_version,
                         self.backup_xtrabackup_info.get("server_version"),
                     )
 
+                xtrabackup_cmd = self.get_xtrabackup_cmd(backup_xtabackup_version)
                 # TODO: Get some execution time numbers with non-trivial data sets for --prepare
                 # and --move-back commands and add progress monitoring if necessary (and feasible)
                 command_line = [
-                    "xtrabackup",
+                    xtrabackup_cmd,
                     # defaults file must be given with --defaults-file=foo syntax, space here does not work
                     f"--defaults-file={self.mysql_config_file_name}",
                     "--no-version-check",
@@ -153,7 +159,7 @@ def restore_backup(self):
                             os.remove(binlog_name)
 
                 command_line = [
-                    "xtrabackup",
+                    xtrabackup_cmd,
                     # defaults file must be given with --defaults-file=foo syntax, space here does not work
                     f"--defaults-file={self.mysql_config_file_name}",
                     "--move-back",
@@ -172,8 +178,18 @@ def restore_backup(self):
 
         self.data_directory_size_end = self._get_directory_size(self.mysql_data_directory, cleanup=True)
 
+    @staticmethod
+    def get_xtrabackup_cmd(backup_xtrabackup_version: BinVersion | None) -> str:
+        xtrabackup_cmd = "xtrabackup"
+        if backup_xtrabackup_version:
+            for bin_info in find_extra_xtrabackup_executables():
+                if bin_info.version[:3] == backup_xtrabackup_version[:3]:
+                    xtrabackup_cmd = str(bin_info.path)
+                    break
+        return xtrabackup_cmd
+
     @property
-    def backup_xtrabackup_version(self) -> Optional[Tuple[int, ...]]:
+    def backup_xtrabackup_version(self) -> Optional[BinVersion]:
         if self.backup_xtrabackup_info is None or "tool_version" not in self.backup_xtrabackup_info:
             return None
         return parse_version(self.backup_xtrabackup_info["tool_version"])

myhoard/myhoard.py

@@ -2,7 +2,12 @@
 from myhoard import version
 from myhoard.controller import Controller
 from myhoard.statsd import StatsClient
-from myhoard.util import DEFAULT_XTRABACKUP_SETTINGS, detect_running_process_id, wait_for_port
+from myhoard.util import (
+    DEFAULT_XTRABACKUP_SETTINGS,
+    detect_running_process_id,
+    find_extra_xtrabackup_executables,
+    wait_for_port,
+)
 from myhoard.web_server import WebServer
 
 import argparse
@@ -101,6 +106,10 @@ def _load_configuration(self):
         if self.config["http_address"] not in {"127.0.0.1", "::1", "localhost"}:
             self.log.warning("Binding to non-localhost address %r is highly discouraged", self.config["http_address"])
 
+        extra_pxb_bins = find_extra_xtrabackup_executables()
+        if extra_pxb_bins:
+            self.log.info("Found extra xtrabackup binaries: %r", extra_pxb_bins)
+
         self.log.info("Configuration loaded")
 
     def _notify_systemd(self):

myhoard/restore_coordinator.py

@@ -9,6 +9,7 @@
 from .table import Table
 from .util import (
     add_gtid_ranges_to_executed_set,
+    BinVersion,
     build_gtid_ranges,
     change_master_to,
     DEFAULT_MYSQL_TIMEOUT,
@@ -150,7 +151,7 @@ class State(TypedDict):
         server_uuid: Optional[str]
         target_time_reached: bool
         write_relay_log_manually: bool
-        backup_xtrabackup_version: Tuple[int, ...] | None
+        backup_xtrabackup_version: BinVersion | None
 
     POLL_PHASES = {Phase.waiting_for_apply_to_finish}
 

myhoard/util.py

@@ -5,9 +5,10 @@
 from cryptography.hazmat.primitives.hashes import SHA1
 from logging import Logger
 from math import log10
+from pathlib import Path
 from pymysql.connections import Connection
 from pymysql.cursors import DictCursor
-from typing import Dict, Iterable, Iterator, List, Literal, Optional, Tuple, TypedDict, Union
+from typing import Dict, Iterable, Iterator, List, Literal, NamedTuple, Optional, Tuple, TypedDict, Union
 
 import collections
 import contextlib
@@ -29,7 +30,7 @@
 # The follow lines used to split a version string which might
 # start with something like:
 # xtrabackup version 8.0.30-23.3.aiven
-XTRABACKUP_VERSION_REGEX = re.compile(r"^xtrabackup version ([\d\.\-]+)")
+XTRABACKUP_VERSION_REGEX = re.compile(r"xtrabackup version ([\d\.\-]+)")
 VERSION_SPLITTING_REGEX = re.compile(r"[\.-]")
 
 DEFAULT_XTRABACKUP_SETTINGS = {
@@ -40,6 +41,12 @@
 }
 
 GtidRangeTuple = tuple[int, int, str, int, int]
+BinVersion = tuple[int, ...]
+
+
+class BinInfo(NamedTuple):
+    path: Path
+    version: BinVersion
 
 
 class GtidRangeDict(TypedDict):
@@ -653,14 +660,14 @@ def restart_unexpected_dead_sql_thread(cursor, slave_status, stats, log):
     cursor.execute("START SLAVE SQL_THREAD")
 
 
-def parse_version(version: str) -> Tuple[int, ...]:
+def parse_version(version: str) -> BinVersion:
     return tuple(int(x) for x in VERSION_SPLITTING_REGEX.split(version) if len(x) > 0)
 
 
-def get_xtrabackup_version() -> Tuple[int, ...]:
-    result = subprocess.run(["xtrabackup", "--version"], capture_output=True, encoding="utf-8", check=True)
+def get_xtrabackup_version(cmd: str | Path = "xtrabackup") -> BinVersion:
+    result = subprocess.run([str(cmd), "--version"], capture_output=True, encoding="utf-8", check=True)
     version_line = result.stderr.strip().split("\n")[-1]
-    matches = XTRABACKUP_VERSION_REGEX.match(version_line)
+    matches = XTRABACKUP_VERSION_REGEX.search(version_line)
     if matches is None:
         raise Exception(f"Cannot extract xtrabackup version number from {result.stderr!r}")
     return parse_version(matches[1])
@@ -683,3 +690,18 @@ def file_name_for_basebackup_split(base_file_name: str, split_nr: int) -> str:
         return f"{base_file_name}.{split_nr:03d}"
     else:
         return base_file_name
+
+
+def find_extra_xtrabackup_executables() -> list[BinInfo]:
+    raw_paths = os.environ.get("PXB_EXTRA_BIN_PATHS")
+    if not raw_paths:
+        return []
+    result = []
+    for extra_raw_path in raw_paths.split(os.pathsep):
+        extra_path = Path(extra_raw_path)
+        if extra_path.is_dir():
+            extra_path = Path(extra_path) / "xtrabackup"
+            if extra_path.exists() and extra_path.is_file():
+                pxb_version = get_xtrabackup_version(extra_path)
+                result.append(BinInfo(version=pxb_version, path=extra_path))
+    return result

test/test_basebackup_restore_operation.py

@@ -2,6 +2,7 @@
 from . import build_statsd_client, wait_for_port
 from myhoard.basebackup_operation import BasebackupOperation
 from myhoard.basebackup_restore_operation import BasebackupRestoreOperation
+from unittest.mock import patch
 
 import myhoard.util as myhoard_util
 import os
@@ -13,6 +14,19 @@
 pytestmark = [pytest.mark.unittest, pytest.mark.all]
 
 
+def test_get_xtrabackup_cmd():
+    cmd = BasebackupRestoreOperation.get_xtrabackup_cmd(None)
+    assert cmd == "xtrabackup"
+    xtrabackup_path = shutil.which("xtrabackup")
+    xtrabackup_dir = os.path.dirname(xtrabackup_path)
+    xtrabackup_version = myhoard_util.get_xtrabackup_version()
+    with patch.dict(os.environ, {"PXB_EXTRA_BIN_PATHS": xtrabackup_dir}):
+        cmd = BasebackupRestoreOperation.get_xtrabackup_cmd(xtrabackup_version)
+        assert cmd == xtrabackup_path
+        cmd = BasebackupRestoreOperation.get_xtrabackup_cmd((8, 0, 0))
+        assert cmd == "xtrabackup"
+
+
 def test_basic_restore(mysql_master, mysql_empty):
     with myhoard_util.mysql_cursor(**mysql_master.connect_options) as cursor:
         for db_index in range(15):
@@ -49,6 +63,13 @@ def input_stream_handler(stream):
             shutil.copyfileobj(backup_file, stream)
             stream.close()
 
+        get_xtrabackup_cmd_called_with = []
+        original_get_xtrabackup_cmd = BasebackupRestoreOperation.get_xtrabackup_cmd
+
+        def patched_get_xtrabackup_cmd(backup_xtrabackup_version):
+            get_xtrabackup_cmd_called_with.append(backup_xtrabackup_version)
+            return original_get_xtrabackup_cmd(backup_xtrabackup_version)
+
         restore_op = BasebackupRestoreOperation(
             encryption_algorithm="AES256",
             encryption_key=encryption_key,
@@ -59,7 +80,10 @@ def input_stream_handler(stream):
             stream_handler=input_stream_handler,
             temp_dir=mysql_empty.base_dir,
         )
-        restore_op.restore_backup()
+        with patch.object(restore_op, "get_xtrabackup_cmd", side_effect=patched_get_xtrabackup_cmd):
+            restore_op.restore_backup()
+        # Check that correct PXB version was extracted from the backup
+        assert get_xtrabackup_cmd_called_with == [myhoard_util.get_xtrabackup_version()]
 
         assert restore_op.number_of_files >= backup_op.number_of_files
 

test/test_util.py

@@ -12,6 +12,7 @@
 import pymysql
 import pytest
 import random
+import shutil
 import subprocess
 
 pytestmark = [pytest.mark.unittest, pytest.mark.all]
@@ -406,3 +407,16 @@ def test_parse_xtrabackup_info() -> None:
         "tool_version": "8.0.30-23",
         "server_version": "8.0.30",
     }
+
+
+def test_find_extra_xtrabackup_executables() -> None:
+    bin_infos = myhoard_util.find_extra_xtrabackup_executables()
+    assert len(bin_infos) == 0
+    xtrabackup_path = shutil.which("xtrabackup")
+    assert xtrabackup_path is not None
+    xtrabackup_dir = os.path.dirname(xtrabackup_path)
+    with patch.dict(os.environ, {"PXB_EXTRA_BIN_PATHS": xtrabackup_dir}):
+        bin_infos = myhoard_util.find_extra_xtrabackup_executables()
+        assert len(bin_infos) == 1
+        assert bin_infos[0].path.name == "xtrabackup"
+        assert bin_infos[0].version >= (8, 0, 30)