diff --git a/hashicorp_kmip_setup.py b/hashicorp_kmip_setup.py index a1c4e31..1d3d7dc 100755 --- a/hashicorp_kmip_setup.py +++ b/hashicorp_kmip_setup.py @@ -58,15 +58,15 @@ def setup_directories(config_dir: Path, data_dir: Path, log_dir: Path, certs_dir def check_license(license_path: str | None) -> str: info("Checking for license file...") - # First check environment variable (like bash: export VAULT_LICENSE=...) - env_license = os.environ.get("VAULT_LICENSE") + # First check environment variable (like bash: export HASHICORP_VAULT_LICENSE=...) + env_license = os.environ.get("HASHICORP_VAULT_LICENSE") if env_license: - info("Using license from environment variable VAULT_LICENSE") + info("Using license from environment variable HASHICORP_VAULT_LICENSE") return env_license.strip() # Fall back to --license file path if not license_path: - error("No license provided. Either set VAULT_LICENSE env var or use --license=/path/to/vault.hclic") + error("No license provided. Either set HASHICORP_VAULT_LICENSE env var or use --license=/path/to/vault.hclic") sys.exit(1) p = Path(license_path) @@ -286,7 +286,7 @@ def parse_args() -> argparse.Namespace: epilog=""" LICENSE: Provide via environment variable (recommended): - export VAULT_LICENSE=$(cat /path/to/vault.hclic) + export HASHICORP_VAULT_LICENSE=$(cat /path/to/vault.hclic) python3 hashicorp-kmip-setup.py Or via file path: @@ -296,7 +296,7 @@ def parse_args() -> argparse.Namespace: python3 hashicorp-kmip-setup.py --license=/path/to/vault.hclic python3 hashicorp-kmip-setup.py --license=/path/to/vault.hclic --cert-dir=/custom/path/certs python3 hashicorp-kmip-setup.py --license=/path/to/vault.hclic --verbose - export VAULT_LICENSE=$(cat vault.hclic) && python3 hashicorp-kmip-setup.py --verbose + export HASHICORP_VAULT_LICENSE=$(cat vault.hclic) && python3 hashicorp-kmip-setup.py --verbose """, ) parser.add_argument( @@ -314,7 +314,7 @@ def parse_args() -> argparse.Namespace: "--license", required=False, metavar="FILE", - help="Path to vault.hclic. Can also be set via: export VAULT_LICENSE=", + help="Path to vault.hclic. Can also be set via: export HASHICORP_VAULT_LICENSE=", ) return parser.parse_args() diff --git a/test_scripts/pxb/README.md b/test_scripts/pxb/README.md index 3e755eb..196725b 100644 --- a/test_scripts/pxb/README.md +++ b/test_scripts/pxb/README.md @@ -41,6 +41,10 @@ - [Run a specific test suite](#run-a-specific-test-suite-3) - [Run all tests](#run-all-tests-3) - [Test reference — upgrade\_backup\_tests.py](#test-reference--upgrade_backup_testspy) +- [xbstream\_fifo\_tests.py — FIFO-streamed cloud backup tests](#xbstream_fifo_testspy--fifo-streamed-cloud-backup-tests) + - [Additional environment variables](#additional-environment-variables-3) + - [How to run tests](#how-to-run-tests-4) + - [Test reference — xbstream\_fifo\_tests.py](#test-reference--xbstream_fifo_testspy) --- @@ -105,7 +109,7 @@ If any of these are unset, KMS tests are skipped. PS 8.0+ only; skipped on MS. **For KMIP tests** (`test_kmip_component_backup`, `test_crash_backup_encrypted_kmip`): -- Vault types come from `KMIP_CONFIGS` in `test_helper.py` (currently `pykmip`, `fortanix`). +- Vault types come from `KMIP_CONFIGS` in `test_helper.py` (currently `pykmip`, `fortanix`, `hashicorp`). - KMIP tests require PS 8.0+ (skipped on 5.7); `keyring_kmip` tests are skipped on MS. - For **Fortanix** vault variants, export: @@ -116,6 +120,14 @@ export FORTANIX_PASSWORD= If Fortanix vars are not set, Fortanix-only variants are skipped. +- For **HashiCorp Vault** variants, export a real Vault Enterprise license: + +```bash +export HASHICORP_VAULT_LICENSE=$(cat /path/to/vault.hclic) +``` + +If unset, `hashicorp`-only variants are skipped. + ### How to run tests Use **pytest** with the test file. Useful options: @@ -290,8 +302,8 @@ pytest inc_backup_load_tests.py --collect-only -q | `test_page_tracking_backup`| Non-param | Page tracking; skipped on 5.7 | | `test_crash_backup` | Param | `[innodb-no_pt]`, `[innodb-pt]`, `[rocksdb-no_pt]`, `[rocksdb-pt]` | | `test_crash_backup_encrypted_keyring_file` | Param `[no_pt]`, `[pt]` | Encrypted crash flow using keyring_file component | -| `test_crash_backup_encrypted_kmip` | Param | One id per `vault_type` x page tracking, e.g. `[pykmip-no_pt]`, `[pykmip-pt]`; Fortanix variants require `FORTANIX_EMAIL`, `FORTANIX_PASSWORD` | -| `test_kmip_component_backup`| Param | One id per vault in `KMIP_CONFIGS` (e.g. `[pykmip]`, `[fortanix]`); Fortanix variants require `FORTANIX_EMAIL`, `FORTANIX_PASSWORD` | +| `test_crash_backup_encrypted_kmip` | Param | One id per `vault_type` x page tracking, e.g. `[pykmip-no_pt]`, `[pykmip-pt]`; Fortanix variants require `FORTANIX_EMAIL`, `FORTANIX_PASSWORD`; hashicorp variants require `HASHICORP_VAULT_LICENSE` | +| `test_kmip_component_backup`| Param | One id per vault in `KMIP_CONFIGS` (e.g. `[pykmip]`, `[fortanix]`, `[hashicorp]`); Fortanix variants require `FORTANIX_EMAIL`, `FORTANIX_PASSWORD`; hashicorp variants require `HASHICORP_VAULT_LICENSE` | | `test_kms_component_backup` | Param `[no_pt]`, `[pt]` | keyring_kms component; requires `KMS_KEYID`, `KMS_SECRET_KEY`, `KMS_AUTH_KEY`, `KMS_REGION`; skipped on 5.7 and MS | --- @@ -830,3 +842,60 @@ python upgrade_backup_tests.py All | `test_upgrade_full_backup` | 1. full backup with previous PXB → prepare/restore with current PXB | Plain `--log-bin=binlog` server options; sysbench load runs in background | | `test_upgrade_inc_backup` | 1. full + inc with previous PXB → prepare/restore with current PXB; 2. full with previous PXB + inc with current PXB → prepare/restore with current PXB | Two scenarios run sequentially against the same primary; sysbench load on each | | `test_upgrade_backup_encrypt` | 1. full prev → prepare/restore current; 2. full + inc prev → prepare/restore current; 3. full prev + inc current → prepare/restore current | Auto-detects server type (PS / MS / 5.7) and applies matching keyring + encryption options; uses `--keyring_file_data=/keyring` and per-binary `--xtrabackup-plugin-dir` for both PXB binaries | + +--- + +## xbstream_fifo_tests.py — FIFO-streamed cloud backup tests + +Tests in `xbstream_fifo_tests.py` run full/incremental/compressed/partition-table/encrypted backups streamed through `xtrabackup`/`xbcloud`/`xbstream` named pipes (`--fifo-streams`/`--fifo-dir`) to a local [SeaweedFS](https://github.com/seaweedfs/seaweedfs) S3-gateway container, instead of a single-process shell pipe (which is what `take_backup(backup_type="cloud")` in the other test files uses). This exercises `xbcloud`'s multi-stream parallel upload/download path against an S3-compatible backend, without needing real cloud credentials. + +**Additional requirement:** Docker must be installed and running — `xbstream_fifo_tests.py` starts a `chrislusf/seaweedfs` container (`seaweedfs_helper.py`) once per test session (port `9000` on the host by default) and stops it when the session ends. + +### Additional environment variables + +```bash +export LOAD_TOOL=pstress +export LOAD_TOOL_DIR=$HOME/lab/pstress/src +export FIFO_STREAM=30 # number of named pipes; default 30 +export FIFO_DIR=/tmp/xbstream_fifo # FIFO pipe directory; default /tmp/xbstream_fifo +``` + +`test_fifo_partition_tables`, `test_fifo_keyring_file_backup`, and `test_fifo_kmip_backup` require `LOAD_TOOL=pstress` (they're skipped otherwise, since they rely on pstress's partition-table/encrypted-table DDL). + +**For KMIP tests** (`test_fifo_kmip_backup`): same requirements as `inc_backup_load_tests.py`'s `test_kmip_component_backup` — vault types come from `KMIP_CONFIGS`, skipped on 5.7/MS, Fortanix variants need `FORTANIX_EMAIL`/`FORTANIX_PASSWORD`, and hashicorp variants need `HASHICORP_VAULT_LICENSE` (a real Vault Enterprise license). + +### How to run tests + +```bash +# Single test +pytest xbstream_fifo_tests.py -v -s -k test_fifo_full_backup_and_restore + +# Multiple tests +pytest xbstream_fifo_tests.py -v -s -k "test_fifo_full_backup_and_restore or test_fifo_compressed_backup" + +# One KMIP vault variant +pytest xbstream_fifo_tests.py -v -s -k "test_fifo_kmip_backup[pykmip]" + +# Whole file +pytest xbstream_fifo_tests.py -v -s + +# Via the script's built-in suites +python xbstream_fifo_tests.py Fifo_Backup_tests +python xbstream_fifo_tests.py Fifo_Partition_tests +python xbstream_fifo_tests.py Fifo_Encryption_tests +python xbstream_fifo_tests.py Fifo_Kmip_tests +``` + +Redirecting `-s` output to a log file for `tail -f`? Python fully buffers stdout when it isn't a TTY, so a genuinely-progressing test can look stuck. Use `python3 -u -m pytest ...` (or `export PYTHONUNBUFFERED=1`) to keep the log flushing in real time. + +### Test reference — xbstream_fifo_tests.py + +| Test | Type | Notes | +|------|------|-------| +| `test_fifo_full_backup_and_restore` | Non-param | Full backup and restore | +| `test_fifo_incremental_backup` | Non-param | Full + 3 incrementals (5s apart), each restored/prepared separately (mirrors the original bash script's fixed-count design, not the load-until-done loop `take_backup()` uses elsewhere) | +| `test_fifo_compressed_backup` | Non-param | Full backup with `--compress=zstd --compress-zstd-level=19` | +| `test_fifo_partition_tables` | Non-param | Incremental backup of pstress-generated partitioned tables; requires `LOAD_TOOL=pstress` | +| `test_fifo_keyring_file_backup` | Non-param | keyring_file component encrypted incremental backup; requires `LOAD_TOOL=pstress`. As of PXB 8.4.0-7/PS 8.4.10-10 this can occasionally crash `xtrabackup` with an InnoDB assertion (`fil0fil.cc:...:page_id.space() != TRX_SYS_SPACE`) while parsing the redo log of an encrypted incremental — a product-level PXB/InnoDB bug, not a bug in this test | +| `test_fifo_kmip_backup` | Param | One id per vault in `KMIP_CONFIGS` (e.g. `[pykmip]`, `[fortanix]`, `[hashicorp]`); requires `LOAD_TOOL=pstress`; skipped on 5.7/MS; Fortanix variants require `FORTANIX_EMAIL`, `FORTANIX_PASSWORD`; hashicorp variants require `HASHICORP_VAULT_LICENSE` | +| `test_fifo_encrypted_backup` | Non-param | Full backup encrypted with xbcrypt (`--encrypt=AES256`, not keyring-based) | diff --git a/test_scripts/pxb/kmip_helper.py b/test_scripts/pxb/kmip_helper.py index 14d482d..2e4d3e5 100644 --- a/test_scripts/pxb/kmip_helper.py +++ b/test_scripts/pxb/kmip_helper.py @@ -38,7 +38,7 @@ class KMIPHelper: DEFAULT_KMIP_CONFIGS = { "pykmip": "addr=127.0.0.1,image=satyapercona/kmip:latest,port=5696,name=kmip_pykmip", # "fortanix": "addr=216.180.120.88,port=5696,name=kmip_fortanix,setup_script=fortanix_kmip_setup.py", - # "hashicorp": "addr=127.0.0.1,port=5696,name=kmip_hashicorp,setup_script=hashicorp_kmip_setup.py", + "hashicorp": "addr=127.0.0.1,port=5696,name=kmip_hashicorp,setup_script=hashicorp_kmip_setup.py", # "ciphertrust": "addr=127.0.0.1,port=5696,name=kmip_ciphertrust,setup_script=setup_kmip_api.py", } @@ -402,7 +402,8 @@ def setup_hashicorp(self) -> bool: script_content = response.read().decode("utf-8") if not script_content: - print("Downloaded script is empty") + self.last_error = "Downloaded HashiCorp setup script is empty" + print(self.last_error) return False if os.path.exists(cert_dir): @@ -411,21 +412,29 @@ def setup_hashicorp(self) -> bool: os.makedirs(cert_dir, exist_ok=True) - # Execute the script + # Execute the script (it's a real Python script, run via + # `python3 -` to read it from stdin, matching how the retired + # bash kmip_helper.sh invoked it: `python3 - --cert-dir=...`) result = subprocess.run( - ["bash", "-s", "--", f"--cert-dir={cert_dir}"], + ["python3", "-", f"--cert-dir={cert_dir}"], input=script_content, text=True, capture_output=True, check=False, ) if result.returncode != 0: - print(f"Failed to execute script {setup_script}, (exit code: {result.returncode})") + detail = (result.stderr or result.stdout or "").strip() + self.last_error = ( + f"HashiCorp setup script exited with code {result.returncode}" + + (f": {detail}" if detail else "") + ) + print(f"Failed to execute script {setup_script}: {self.last_error}") return False self.generate_kmip_config(kmip_type, addr, port, cert_dir) except Exception as e: - print(f"Failed to setup HashiCorp: {e}") + self.last_error = f"Failed to setup HashiCorp: {e}" + print(self.last_error) return False self.kmip_config["cert_dir"] = cert_dir diff --git a/test_scripts/pxb/seaweedfs_helper.py b/test_scripts/pxb/seaweedfs_helper.py new file mode 100644 index 0000000..264b43b --- /dev/null +++ b/test_scripts/pxb/seaweedfs_helper.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python3 +""" +SeaweedFS Helper Library +Usage: from seaweedfs_helper import SeaweedFSHelper + +Manages a local SeaweedFS S3-gateway container used as the S3-compatible +backend for xbstream_fifo_tests.py's cloud/FIFO backup tests. Ported from +xbstream_fifo_test.sh's start_seaweedfs()/cleanup_exit() shell functions. +Required: Docker must be installed and running. +""" + +import json +import os +import subprocess +import time +import urllib.request +import urllib.error +from typing import Optional + +CONTAINER_NAME = "seaweedfs" +IMAGE = "chrislusf/seaweedfs:latest" +HOST_PORT = 9000 +CONTAINER_S3_PORT = 8333 +S3_ACCESS_KEY = "admin" +S3_SECRET_KEY = "password" + + +class SeaweedFSHelper: + """Helper class for starting/stopping the SeaweedFS S3-gateway container.""" + + def __init__(self, data_dir: Optional[str] = None, host_port: int = HOST_PORT): + self.data_dir = data_dir or os.path.join(os.path.expanduser("~"), "seaweedfs", "data") + self.config_dir = os.path.dirname(self.data_dir) + self.host_port = host_port + self.last_error: str = "" + + def _container_status(self) -> Optional[str]: + """Return 'running', 'stopped', or None if the container doesn't exist.""" + result = subprocess.run( + ["docker", "ps", "--filter", f"name={CONTAINER_NAME}", "--filter", "status=running", + "--format", "{{.Names}}"], + capture_output=True, text=True, check=False, + ) + if CONTAINER_NAME in result.stdout.split(): + return "running" + result = subprocess.run( + ["docker", "ps", "-a", "--filter", f"name={CONTAINER_NAME}", "--format", "{{.Names}}"], + capture_output=True, text=True, check=False, + ) + if CONTAINER_NAME in result.stdout.split(): + return "stopped" + return None + + def _write_s3_config(self) -> str: + """Write the S3 identity config (admin/password) SeaweedFS's S3 gateway needs.""" + os.makedirs(self.config_dir, exist_ok=True) + os.makedirs(self.data_dir, exist_ok=True) + config = { + "identities": [ + { + "name": "admin", + "credentials": [{"accessKey": S3_ACCESS_KEY, "secretKey": S3_SECRET_KEY}], + "actions": ["Admin", "Read", "Write"], + } + ] + } + config_path = os.path.join(self.config_dir, "s3.json") + with open(config_path, "w", encoding="utf-8") as f: + json.dump(config, f, indent=2) + return config_path + + def _wait_ready(self, timeout: int = 60) -> bool: + """Poll the S3 gateway port. SeaweedFS has no MinIO-style + /minio/health/ready endpoint, so any HTTP response (even an error + status) is enough to know xbcloud can reach it.""" + url = f"http://localhost:{self.host_port}/" + for _ in range(timeout): + try: + urllib.request.urlopen(url, timeout=1) + return True + except urllib.error.HTTPError: + # Any HTTP response (e.g. 403/404) means the gateway is up. + return True + except (urllib.error.URLError, OSError): + time.sleep(1) + return False + + def start(self) -> bool: + """Start the SeaweedFS container if it isn't already running.""" + status = self._container_status() + if status == "running": + print("SeaweedFS is already running.") + elif status == "stopped": + print("Found stopped SeaweedFS container. Starting it...") + result = subprocess.run( + ["docker", "start", CONTAINER_NAME], + capture_output=True, text=True, check=False, + ) + if result.returncode != 0: + self.last_error = result.stderr or result.stdout + print(f"ERR: Failed to start SeaweedFS container: {self.last_error}") + return False + else: + if os.path.isdir(self.data_dir): + for entry in os.listdir(self.data_dir): + path = os.path.join(self.data_dir, entry) + subprocess.run(["rm", "-rf", path], check=False) + config_path = self._write_s3_config() + print("No SeaweedFS container found. Creating and starting one...") + result = subprocess.run( + [ + "docker", "run", "-d", + "-p", f"{self.host_port}:{CONTAINER_S3_PORT}", + "--name", CONTAINER_NAME, + "-v", f"{self.data_dir}:/data", + "-v", f"{config_path}:/etc/seaweedfs/s3.json", + IMAGE, + "server", "-s3", f"-s3.port={CONTAINER_S3_PORT}", + "-s3.config=/etc/seaweedfs/s3.json", "-dir=/data", + ], + capture_output=True, text=True, check=False, + ) + if result.returncode != 0: + self.last_error = result.stderr + print(f"ERR: Failed to start SeaweedFS container: {result.stderr}") + return False + + print("Waiting for SeaweedFS to become ready", end="", flush=True) + if self._wait_ready(): + print("\nSeaweedFS is ready!\n") + return True + + print("\nSeaweedFS failed to become ready in time.") + logs = subprocess.run(["docker", "logs", CONTAINER_NAME], capture_output=True, text=True, check=False) + self.last_error = logs.stdout + logs.stderr + print(self.last_error) + return False + + def stop(self) -> None: + """Stop the SeaweedFS container if it's running.""" + if self._container_status() == "running": + print("Stopping SeaweedFS container...") + subprocess.run(["docker", "stop", CONTAINER_NAME], capture_output=True, check=False) diff --git a/test_scripts/pxb/test_helper.py b/test_scripts/pxb/test_helper.py index d39a8fc..19db041 100644 --- a/test_scripts/pxb/test_helper.py +++ b/test_scripts/pxb/test_helper.py @@ -53,7 +53,7 @@ KMIP_CONFIGS = { "pykmip": "addr=127.0.0.1,image=satyapercona/kmip:latest,port=5696,name=kmip_pykmip", "fortanix": "addr=216.180.120.88,port=5696,name=kmip_fortanix,setup_script=fortanix_kmip_setup.py", - # "hashicorp": "addr=127.0.0.1,port=5696,name=kmip_hashicorp,setup_script=hashicorp_kmip_setup.py", + "hashicorp": "addr=127.0.0.1,port=5696,name=kmip_hashicorp,setup_script=hashicorp_kmip_setup.py", # "ciphertrust": "addr=127.0.0.1,port=5696,name=kmip_ciphertrust,setup_script=setup_kmip_api.py", } @@ -83,6 +83,12 @@ ENCRYPT_KEY = os.environ.get("ENCRYPT_KEY", "mHU3Zs5sRcSB7zBAJP1BInPP5lgShKly") RANDOM_TYPE = os.environ.get("RANDOM_TYPE", "uniform") +# FIFO streaming configuration (used by xbstream_fifo_tests.py to stream +# backups through xtrabackup/xbcloud/xbstream named pipes, rather than the +# single-process shell pipe that take_backup()'s "cloud" backup_type uses). +FIFO_STREAM = int(os.environ.get("FIFO_STREAM", "30")) +FIFO_DIR = os.environ.get("FIFO_DIR", "/tmp/xbstream_fifo") + class MySQLServer: """A single mysqld instance (primary or replica). @@ -1563,40 +1569,14 @@ def take_backup( time.sleep(2) - # --- Stop server and move data directory --- - print("Stopping mysql server and moving data directory") - self.primary.stop(timeout=MYSQL_SHUTDOWN_TIMEOUT) - - data_orig = os.path.join(self.backup_dir, f"data_orig_{datetime.now().strftime('%Y%m%d_%H%M%S')}") - if os.path.exists(data_orig): - shutil.rmtree(data_orig) - shutil.move(self.datadir, data_orig) - - # --- Restore backup --- - print("=>Restoring full backup") - cmd = self._xtrabackup_cmd_prefix() + [ - "--no-defaults", "--copy-back", - f"--target-dir={full_target}", - f"--datadir={self.datadir}", - ] + self.restore_params.split() - - log_file = os.path.join(self.logdir, f"res_backup_{log_date}_log") - result = self.run_command(cmd, check=False, log_file=log_file) - if result.returncode != 0: - pytest.fail(f"ERR: Restore of full backup failed. Please check the log at: {log_file}") - else: - print(f"..Restore of full backup was successful.\n Logs available at: {log_file}") - - # Copy .pem files from original datadir for SSL - for pem_file in glob.glob(os.path.join(data_orig, "*.pem")): - shutil.copy2(pem_file, self.datadir) - - # Copy keyring file if it exists as transition-key - keyring_in_backup = os.path.join(full_target, "keyring") - if os.path.isfile(keyring_in_backup): - shutil.copy2(keyring_in_backup, self.logdir) - - self.start_server() + # --- Stop server, copy-back, restart (shared with FIFO restore path) --- + data_orig = self.restore_datadir_from( + full_target, + data_orig_dir=os.path.join(self.backup_dir, f"data_orig_{datetime.now().strftime('%Y%m%d_%H%M%S')}"), + copy_ssl_pem=True, + copy_keyring_to_logdir=True, + log_file=os.path.join(self.logdir, f"res_backup_{log_date}_log"), + ) # --- Verify --- if use_detailed_verify: @@ -1617,8 +1597,8 @@ def take_backup( print(f"Xtrabackup binlog position: {xb_binlog_file}, {xb_binlog_pos}") print(f"Applying binlog to restored data starting from {xb_binlog_file}, {xb_binlog_pos}") - binlog_path = os.path.join(data_orig, xb_binlog_file) - if os.path.exists(binlog_path): + binlog_path = os.path.join(data_orig, xb_binlog_file) if data_orig else "" + if binlog_path and os.path.exists(binlog_path): mysqlbinlog = subprocess.Popen( [os.path.join(self.mysqldir, "bin/mysqlbinlog"), binlog_path, f"--start-position={xb_binlog_pos}"], stdout=subprocess.PIPE, @@ -1995,7 +1975,7 @@ def build_cloud_params(self, verbose: bool = True) -> str: return " ".join(params) def xbcloud_put(self, cloud_params: str, name: str, stream_file: str): - """Upload backup stream to cloud.""" + """Upload a local stream file to cloud (`cat file | xbcloud put`).""" cmd = f"cat {stream_file} | {os.path.join(self.xtrabackup_dir, 'xbcloud')} put {cloud_params} {name}" log_file = os.path.join(self.logdir, f"xbcloud_put_{name}.log") result = subprocess.run(cmd, shell=True, capture_output=True, text=True, check=False) @@ -2004,22 +1984,330 @@ def xbcloud_put(self, cloud_params: str, name: str, stream_file: str): if result.returncode != 0: pytest.fail(f"xbcloud put failed for {name}. Log: {log_file}") - def xbcloud_get(self, cloud_params: str, name: str, target_dir: str): - """Download backup from cloud.""" + def take_backup_fifo( + self, + cloud_params: str, + name: str, + extra_lsndir: str, + incremental_basedir: Optional[str] = None, + keyring_backup_opts: str = "", + fifo_streams: Optional[int] = None, + fifo_dir: Optional[str] = None, + log_file: Optional[str] = None, + ) -> None: + """Take one backup and upload via FIFO-streamed xbcloud put. + + Unlike :meth:`xbcloud_put` (file-based `cat | put`) and + ``take_backup(backup_type="cloud")`` (single shell pipe), this runs + xtrabackup and xbcloud as two processes that rendezvous through + ``--fifo-streams`` named pipes under ``--fifo-dir``. ``--extra-lsndir`` + captures LSN metadata on local disk (the backup payload itself never + lands locally), which is required for the next incremental's + ``--incremental-basedir``. ``keyring_backup_opts`` should carry + ``--xtrabackup-plugin-dir`` for encrypted backups. + """ + if fifo_streams is None: + fifo_streams = FIFO_STREAM + if fifo_dir is None: + fifo_dir = FIFO_DIR + os.makedirs(extra_lsndir, exist_ok=True) + os.makedirs(fifo_dir, exist_ok=True) + log_file = log_file or os.path.join(self.logdir, f"fifo_backup_{name}.log") + + cmd = self._xtrabackup_cmd_prefix() + [ + "--backup", f"--user={self.backup_user}", "--password=", + f"-S{self.socket_path}", f"--datadir={self.datadir}", + f"--extra-lsndir={extra_lsndir}", + ] + if incremental_basedir: + cmd.append(f"--incremental-basedir={incremental_basedir}") + if keyring_backup_opts: + cmd += keyring_backup_opts.split() + cmd += self.backup_params.split() + cmd += [f"--fifo-streams={fifo_streams}", f"--fifo-dir={fifo_dir}"] + backup_proc = self.run_command(cmd, background=True, log_file=log_file) + + put_cmd = ( + f"{os.path.join(self.xtrabackup_dir, 'xbcloud')} put {cloud_params} " + f"--fifo-streams={fifo_streams} --fifo-dir={fifo_dir} {name} >>{log_file} 2>&1" + ) + put_result = subprocess.run(put_cmd, shell=True, check=False) + backup_returncode = backup_proc.wait() + if put_result.returncode != 0 or backup_returncode != 0: + pytest.fail(f"ERR: FIFO backup/upload failed for '{name}'. Please check the log at: {log_file}") + + def xbcloud_get( + self, + cloud_params: str, + name: str, + target_dir: str, + *, + use_fifo: bool = False, + fifo_streams: Optional[int] = None, + fifo_dir: Optional[str] = None, + log_file: Optional[str] = None, + ) -> None: + """Download backup from cloud into ``target_dir``. + + Default transport is a single shell pipe + (``xbcloud get | xbstream -xvC``). With ``use_fifo=True``, xbstream and + xbcloud run as two processes that rendezvous through ``--fifo-streams`` + named pipes under ``--fifo-dir`` (parallel multi-stream download). + """ os.makedirs(target_dir, exist_ok=True) - cmd = f"{os.path.join(self.xtrabackup_dir, 'xbcloud')} get {cloud_params} {name} | {os.path.join(self.xtrabackup_dir, 'xbstream')} -xvC {target_dir}" - log_file = os.path.join(self.logdir, f"xbcloud_get_{name}.log") - result = subprocess.run(cmd, shell=True, capture_output=True, text=True, check=False) - with open(log_file, "w") as f: - f.write(result.stdout + "\n" + result.stderr) - if result.returncode != 0: - pytest.fail(f"xbcloud get failed for {name}. Log: {log_file}") + log_file = log_file or os.path.join(self.logdir, f"xbcloud_get_{name}.log") + + if not use_fifo: + cmd = ( + f"{os.path.join(self.xtrabackup_dir, 'xbcloud')} get {cloud_params} {name} | " + f"{os.path.join(self.xtrabackup_dir, 'xbstream')} -xvC {target_dir}" + ) + result = subprocess.run(cmd, shell=True, capture_output=True, text=True, check=False) + with open(log_file, "w") as f: + f.write(result.stdout + "\n" + result.stderr) + if result.returncode != 0: + pytest.fail(f"xbcloud get failed for {name}. Log: {log_file}") + return + + if fifo_streams is None: + fifo_streams = FIFO_STREAM + if fifo_dir is None: + fifo_dir = FIFO_DIR + os.makedirs(fifo_dir, exist_ok=True) + + xbstream_cmd = [ + os.path.join(self.xtrabackup_dir, "xbstream"), "-x", "-C", target_dir, + f"--fifo-streams={fifo_streams}", f"--fifo-dir={fifo_dir}", "--parallel=64", + ] + restore_proc = self.run_command(xbstream_cmd, background=True, log_file=log_file) + + get_cmd = ( + f"{os.path.join(self.xtrabackup_dir, 'xbcloud')} get {cloud_params} " + f"--fifo-streams={fifo_streams} --fifo-dir={fifo_dir} {name} >>{log_file} 2>&1" + ) + get_result = subprocess.run(get_cmd, shell=True, check=False) + restore_returncode = restore_proc.wait() + if get_result.returncode != 0 or restore_returncode != 0: + pytest.fail(f"ERR: FIFO restore/download failed for '{name}'. Please check the log at: {log_file}") def xbcloud_delete(self, cloud_params: str, name: str): """Delete backup from cloud.""" cmd = f"{os.path.join(self.xtrabackup_dir, 'xbcloud')} delete {cloud_params} {name}" subprocess.run(cmd, shell=True, capture_output=True, check=False) + def cleanup_fifo_state(self, cloud_params: str, names: List[str], fifo_dir: Optional[str] = None) -> None: + """Delete named S3 objects and clear stale FIFO pipes before a scenario. + + Mirrors xbstream_fifo_test.sh's cleanup(): each FIFO scenario starts + from a clean bucket (no leftover objects from a previous scenario/run + under the same name) and an empty FIFO_DIR (stale pipes left behind + by a killed/crashed process would make the next xtrabackup/xbstream + hang trying to reopen them). + """ + for name in names: + self.xbcloud_delete(cloud_params, name) + if fifo_dir is None: + fifo_dir = FIFO_DIR + if os.path.isdir(fifo_dir): + for entry in os.listdir(fifo_dir): + try: + os.remove(os.path.join(fifo_dir, entry)) + except OSError: + pass + + def take_full_backup_fifo( + self, + cloud_params: str, + keyring_backup_opts: str = "", + fifo_streams: Optional[int] = None, + fifo_dir: Optional[str] = None, + ) -> str: + """Full backup -> FIFO put -> FIFO get -> decrypt/decompress -> prepare. + + Returns ``backup_dir/full``, ready for :meth:`restore_datadir_from`. + Ported from xbstream_fifo_test.sh's full_backup_and_restore(). + """ + if os.path.exists(self.backup_dir): + shutil.rmtree(self.backup_dir) + os.makedirs(self.backup_dir) + + extra_lsndir = os.path.join(self.logdir, "lsn", "full") + if os.path.exists(extra_lsndir): + shutil.rmtree(extra_lsndir) + + print("=>Taking full FIFO backup") + self.take_backup_fifo( + cloud_params, "full_backup", extra_lsndir, + keyring_backup_opts=keyring_backup_opts, + fifo_streams=fifo_streams, fifo_dir=fifo_dir, + ) + print("..Backup successful") + + full_target = os.path.join(self.backup_dir, "full") + print("=>Restoring FIFO backup") + self.xbcloud_get( + cloud_params, "full_backup", full_target, + use_fifo=True, fifo_streams=fifo_streams, fifo_dir=fifo_dir, + ) + print("..Restore successful") + + self._decrypt_decompress(full_target, self.backup_params) + + print("=>Preparing backup") + cmd = self._xtrabackup_cmd_prefix() + [ + "--no-defaults", "--prepare", f"--target-dir={full_target}", + ] + self._prepare_args_for_pxb_version(self.prepare_params) + log_file = os.path.join(self.logdir, "fifo_prepare_full.log") + result = self.run_command(cmd, check=False, log_file=log_file) + if result.returncode != 0: + pytest.fail(f"ERR: xtrabackup --prepare failed. Please check the log at: {log_file}") + print("..Prepare successful") + + self.xbcloud_delete(cloud_params, "full_backup") + return full_target + + def take_incremental_backup_fifo( + self, + cloud_params: str, + keyring_backup_opts: str = "", + num_incrementals: int = 3, + fifo_streams: Optional[int] = None, + fifo_dir: Optional[str] = None, + ) -> str: + """Full + N incremental FIFO backups, restore each, chained prepare. + + Incrementals are 5s apart; each is restored into its own + ``backup_dir/`` (never merged locally) then applied with a + chained ``--apply-log-only`` prepare. Returns ``backup_dir/full``, + ready for :meth:`restore_datadir_from`. Ported from + xbstream_fifo_test.sh's incremental_backup_and_restore(). + """ + if os.path.exists(self.backup_dir): + shutil.rmtree(self.backup_dir) + os.makedirs(self.backup_dir) + + lsn_root = os.path.join(self.logdir, "lsn") + if os.path.exists(lsn_root): + shutil.rmtree(lsn_root) + + names = ["full"] + [f"inc{i}" for i in range(1, num_incrementals + 1)] + lsn_dirs = {name: os.path.join(lsn_root, name) for name in names} + + print("=>Taking Full Backup") + self.take_backup_fifo( + cloud_params, "full", lsn_dirs["full"], + keyring_backup_opts=keyring_backup_opts, + fifo_streams=fifo_streams, fifo_dir=fifo_dir, + ) + print("..Full Backup successful") + + prev = "full" + for i in range(1, num_incrementals + 1): + time.sleep(5) + name = f"inc{i}" + print(f"=>Taking Incremental Backup {i}") + self.take_backup_fifo( + cloud_params, name, lsn_dirs[name], + incremental_basedir=lsn_dirs[prev], + keyring_backup_opts=keyring_backup_opts, + fifo_streams=fifo_streams, fifo_dir=fifo_dir, + ) + print("..Successful") + prev = name + + print("=>Restoring Backup...") + targets = {} + for name in names: + target = os.path.join(self.backup_dir, name) + self.xbcloud_get( + cloud_params, name, target, + use_fifo=True, fifo_streams=fifo_streams, fifo_dir=fifo_dir, + ) + self._decrypt_decompress(target, self.backup_params) + targets[name] = target + + full_target = targets["full"] + print("=>Preparing Full Backup") + cmd = self._xtrabackup_cmd_prefix() + [ + "--no-defaults", "--prepare", "--apply-log-only", f"--target-dir={full_target}", + ] + self._prepare_args_for_pxb_version(self.prepare_params) + log_file = os.path.join(self.logdir, "fifo_prepare_full.log") + result = self.run_command(cmd, check=False, log_file=log_file) + if result.returncode != 0: + pytest.fail(f"ERR: Prepare of full backup failed. Please check the log at: {log_file}") + print("..Prepare successful") + + for i in range(1, num_incrementals + 1): + name = f"inc{i}" + is_last = i == num_incrementals + print(f"=>Preparing Incremental Backup {i}") + cmd = self._xtrabackup_cmd_prefix() + [ + "--no-defaults", "--prepare", + ] + ([] if is_last else ["--apply-log-only"]) + [ + f"--target-dir={full_target}", f"--incremental-dir={targets[name]}", + ] + self._prepare_args_for_pxb_version(self.prepare_params) + log_file = os.path.join(self.logdir, f"fifo_prepare_{name}.log") + result = self.run_command(cmd, check=False, log_file=log_file) + if result.returncode != 0: + pytest.fail(f"ERR: Prepare of incremental backup {i} failed. Please check the log at: {log_file}") + print("..Successful") + + for name in names: + self.xbcloud_delete(cloud_params, name) + + return full_target + + def restore_datadir_from( + self, + source_target_dir: str, + *, + data_orig_dir: Optional[str] = None, + copy_ssl_pem: bool = False, + copy_keyring_to_logdir: bool = False, + log_file: Optional[str] = None, + ) -> str: + """Stop the server, move the current datadir aside, ``--copy-back``, restart. + + Shared by FIFO scenarios and :meth:`take_backup`'s restore path. + Returns the path of the moved-aside original datadir (empty string if + there was no datadir to move). + """ + print("Stopping mysql server and moving data directory") + self.primary.stop(timeout=MYSQL_SHUTDOWN_TIMEOUT) + + data_orig = "" + if os.path.exists(self.datadir): + data_orig = data_orig_dir or os.path.join( + os.path.dirname(self.datadir), f"data_orig_{datetime.now().strftime('%Y%m%d_%H%M%S')}" + ) + if os.path.exists(data_orig): + shutil.rmtree(data_orig) + shutil.move(self.datadir, data_orig) + + os.makedirs(self.datadir, exist_ok=True) + print("=>Restoring full backup") + log_file = log_file or os.path.join(self.logdir, "copy_back.log") + cmd = self._xtrabackup_cmd_prefix() + [ + "--no-defaults", "--copy-back", + f"--target-dir={source_target_dir}", f"--datadir={self.datadir}", + ] + self.restore_params.split() + result = self.run_command(cmd, check=False, log_file=log_file) + if result.returncode != 0: + pytest.fail(f"ERR: xtrabackup --copy-back failed. Please check the log at: {log_file}") + print(f"..Restored datadir at: {self.datadir}") + + if copy_ssl_pem and data_orig: + for pem_file in glob.glob(os.path.join(data_orig, "*.pem")): + shutil.copy2(pem_file, self.datadir) + + if copy_keyring_to_logdir: + keyring_in_backup = os.path.join(source_target_dir, "keyring") + if os.path.isfile(keyring_in_backup): + shutil.copy2(keyring_in_backup, self.logdir) + + self.start_server() + return data_orig + def run_ddl_in_background(self, ddl_func, *args, **kwargs) -> threading.Thread: """Launch a DDL operation in a background thread. Returns the thread handle.""" thread = threading.Thread(target=ddl_func, args=args, kwargs=kwargs, daemon=True, name=f"ddl_{ddl_func.__name__}") @@ -2767,6 +3055,9 @@ def run_kmip_component_backup(self, vault_type: str) -> None: ): pytest.skip("Fortanix KMIP requires FORTANIX_EMAIL and FORTANIX_PASSWORD environment variables") + if vault_type == "hashicorp" and not os.environ.get("HASHICORP_VAULT_LICENSE", "").strip(): + pytest.skip("HashiCorp Vault KMIP requires a HASHICORP_VAULT_LICENSE environment variable (Vault Enterprise license)") + if not self.kmip_helper: self.kmip_helper = KMIPHelper(KMIP_CONFIGS, cert_base_dir=TEST_BASE_DIR) if not self.kmip_helper.start_kmip_server(vault_type): @@ -3043,6 +3334,10 @@ def run_crash_tests_pstress_encrypted( pytest.skip( "Fortanix KMIP requires FORTANIX_EMAIL and FORTANIX_PASSWORD" ) + if vault_type == "hashicorp" and not os.environ.get("HASHICORP_VAULT_LICENSE", "").strip(): + pytest.skip( + "HashiCorp Vault KMIP requires a HASHICORP_VAULT_LICENSE environment variable" + ) print(f"Testing keyring_kmip with vault {vault_type} (encrypted crash)...") if not self.kmip_helper: self.kmip_helper = KMIPHelper(KMIP_CONFIGS, cert_base_dir=TEST_BASE_DIR) diff --git a/test_scripts/pxb/xbstream_fifo_tests.py b/test_scripts/pxb/xbstream_fifo_tests.py new file mode 100644 index 0000000..b058305 --- /dev/null +++ b/test_scripts/pxb/xbstream_fifo_tests.py @@ -0,0 +1,330 @@ +#!/usr/bin/env python3 +""" +PXB FIFO-streamed cloud backup tests. + +Pytest port of xbstream_fifo_test.sh: full/incremental/compressed/ +partition-table/encrypted backups streamed through xtrabackup/xbcloud/ +xbstream named pipes (--fifo-streams/--fifo-dir) to a local SeaweedFS +S3-gateway container, instead of a single-process shell pipe. + +Assumption: PS and PXB are already installed as tarballs, and Docker is +installed and running (for the SeaweedFS container). +""" + +import os +import pytest + +from test_helper import BackupTestHelper, TEST_BASE_DIR, KMIP_CONFIGS, CORE_FILE_OPT +from seaweedfs_helper import SeaweedFSHelper + +try: + from kmip_helper import KMIPHelper +except ImportError: + KMIPHelper = None + +# Local S3 credentials/bucket for the SeaweedFS container (matches +# xbstream_fifo_test.sh's hardcoded admin/password/my-bucket). +SEAWEEDFS_BUCKET = "my-bucket" +SEAWEEDFS_ACCESS_KEY = "admin" +SEAWEEDFS_SECRET_KEY = "password" +SEAWEEDFS_REGION = "us-east-1" + +VAULT_TYPES = list(KMIP_CONFIGS.keys()) + + +# Pytest fixtures +@pytest.fixture(scope="session") +def seaweedfs(): + """Start the local SeaweedFS S3-gateway container once for the whole + test session (mirrors xbstream_fifo_test.sh, which starts the container + once at the top of the script and leaves it running across scenarios), + and stop it when the session ends.""" + helper = SeaweedFSHelper() + if not helper.start(): + pytest.fail(f"Failed to start SeaweedFS: {helper.last_error}") + yield helper + if os.environ.get("DISABLE_CLEANUP") != "1": + helper.stop() + + +@pytest.fixture(scope="function") +def test_helper(request): + """Create a test helper instance for each test.""" + test_name = request.node.name if hasattr(request, "node") else None + helper = BackupTestHelper(test_name=test_name) + helper.server_version, helper.server_version_normalized = helper.get_mysql_version() + yield helper + if os.environ.get("DISABLE_CLEANUP") != "1": + helper.cleanup() + + +@pytest.fixture(scope="function", autouse=True) +def setup_logdir(test_helper): + """Ensure log directory exists.""" + if not os.path.exists(test_helper.logdir): + os.makedirs(test_helper.logdir) + + +@pytest.fixture(scope="function") +def cloud_params(seaweedfs, test_helper): + """Point test_helper at the local SeaweedFS container and return the + assembled xbcloud S3 option string.""" + test_helper.s3_bucket = SEAWEEDFS_BUCKET + test_helper.s3_access_key = SEAWEEDFS_ACCESS_KEY + test_helper.s3_secret_key = SEAWEEDFS_SECRET_KEY + test_helper.s3_region = SEAWEEDFS_REGION + test_helper.s3_endpoint = f"http://localhost:{seaweedfs.host_port}" + return test_helper.build_cloud_params() + + +# Module-level helpers (mirrors the _pstress_tool_options()/_run_load() style +# used by innodb_myrocks_backup_tests.py) +def _default_tool_options(test_helper) -> str: + opts = ( + f"--tables {test_helper.num_tables} --records {test_helper.table_size} " + f"--threads {test_helper.threads} --seconds {test_helper.seconds} " + "--no-encryption --undo-tbs-sql 0" + ) + if test_helper.server_type == "MS": + opts += " --no-column-compression --no-temp-tables" + return opts + + +def _pstress_tool_options(seconds: int = 120, only_partition_tables: bool = False, no_encryption: bool = True) -> str: + opts = f"--tables 150 --records 1000 --seconds {seconds} --threads 10" + if no_encryption: + opts += " --no-encryption" + if only_partition_tables: + opts += " --only-partition-tables" + return opts + + +# Test functions +def test_fifo_full_backup_and_restore(test_helper, cloud_params): + """Full backup and restore streamed via FIFO to SeaweedFS.""" + test_helper.backup_params = f"{CORE_FILE_OPT}" + test_helper.prepare_params = f"{CORE_FILE_OPT}" + test_helper.restore_params = "" + + test_helper.initialize_db() + test_helper.run_load(_default_tool_options(test_helper)) + + test_helper.cleanup_fifo_state(cloud_params, ["full_backup"]) + full_target = test_helper.take_full_backup_fifo(cloud_params) + test_helper.restore_datadir_from(full_target) + test_helper.check_tables() + + +def test_fifo_incremental_backup(test_helper, cloud_params): + """Full + 3 incremental backups (5s apart) streamed via FIFO to SeaweedFS.""" + test_helper.backup_params = f"{CORE_FILE_OPT}" + test_helper.prepare_params = f"{CORE_FILE_OPT}" + test_helper.restore_params = "" + + test_helper.initialize_db() + test_helper.run_load(_default_tool_options(test_helper)) + + test_helper.cleanup_fifo_state(cloud_params, ["full", "inc1", "inc2", "inc3"]) + full_target = test_helper.take_incremental_backup_fifo(cloud_params) + test_helper.restore_datadir_from(full_target) + test_helper.check_tables() + + +def test_fifo_compressed_backup(test_helper, cloud_params): + """Full backup with zstd compression streamed via FIFO to SeaweedFS.""" + test_helper.backup_params = f"--compress=zstd --compress-zstd-level=19 --compress-threads=10 {CORE_FILE_OPT}" + test_helper.prepare_params = f"{CORE_FILE_OPT}" + test_helper.restore_params = "" + + test_helper.initialize_db() + test_helper.run_load(_default_tool_options(test_helper)) + + test_helper.cleanup_fifo_state(cloud_params, ["full_backup"]) + full_target = test_helper.take_full_backup_fifo(cloud_params) + test_helper.restore_datadir_from(full_target) + test_helper.check_tables() + + +def test_fifo_partition_tables(test_helper, cloud_params): + """Incremental backup of pstress-generated partitioned tables, streamed via FIFO to SeaweedFS.""" + if test_helper.load_tool != "pstress": + pytest.skip("Partition-table load requires LOAD_TOOL=pstress") + + test_helper.backup_params = f"{CORE_FILE_OPT}" + test_helper.prepare_params = f"{CORE_FILE_OPT}" + test_helper.restore_params = "" + + test_helper.initialize_db() + test_helper.run_load(_pstress_tool_options(seconds=120, only_partition_tables=True)) + + test_helper.cleanup_fifo_state(cloud_params, ["full", "inc1", "inc2", "inc3"]) + full_target = test_helper.take_incremental_backup_fifo(cloud_params) + test_helper.restore_datadir_from(full_target) + test_helper.check_tables() + + +def test_fifo_keyring_file_backup(test_helper, cloud_params): + """keyring_file encrypted incremental backup streamed via FIFO to SeaweedFS. + + Known issue: as of PXB 8.4.0-7 / PS 8.4.10-10, taking an incremental + backup of a keyring_file-encrypted tablespace can crash xtrabackup with + an InnoDB assertion (fil0fil.cc:10943:page_id.space() != TRX_SYS_SPACE) + while parsing the redo log. That's a product-level PXB/InnoDB bug (not + a bug in the FIFO/SeaweedFS plumbing here) -- file/track it against PXB + separately if hit; this test still documents/exercises the scenario. + """ + if test_helper.load_tool != "pstress": + pytest.skip("keyring_file encrypted-table load requires LOAD_TOOL=pstress") + + test_helper.create_keyring_manifest("component_keyring_file") + config_file = test_helper.create_keyring_config( + "keyring_file", keyring_path=os.path.join(test_helper.logdir, "keyring") + ) + + keyring_backup_opts = f"--xtrabackup-plugin-dir={test_helper.xtrabackup_dir}/../lib/plugin" + test_helper.backup_params = f"{CORE_FILE_OPT}" + test_helper.prepare_params = f"{CORE_FILE_OPT} {keyring_backup_opts} --component-keyring-config={config_file}" + test_helper.restore_params = test_helper.prepare_params + + test_helper.initialize_db() + # No --no-encryption: pstress creates some ENCRYPTION='Y' tables, matching + # xbstream_fifo_test.sh's keyring_file scenario. + test_helper.run_load(_pstress_tool_options(seconds=120, no_encryption=False)) + + test_helper.cleanup_fifo_state(cloud_params, ["full", "inc1", "inc2", "inc3"]) + full_target = test_helper.take_incremental_backup_fifo( + cloud_params, keyring_backup_opts=keyring_backup_opts + ) + test_helper.restore_datadir_from(full_target) + test_helper.check_tables() + + +@pytest.mark.parametrize("vault_type", VAULT_TYPES) +def test_fifo_kmip_backup(test_helper, cloud_params, vault_type): + """keyring_kmip encrypted incremental backup streamed via FIFO to SeaweedFS.""" + if test_helper.load_tool != "pstress": + pytest.skip("keyring_kmip encrypted-table load requires LOAD_TOOL=pstress") + if test_helper.server_version_normalized < 80000: + pytest.skip("KMIP component is not supported in 5.7") + if test_helper.server_type == "MS": + pytest.skip("MS does not support keyring kmip for encryption") + if not KMIPHelper: + pytest.skip("KMIP helper not available (kmip_helper module)") + if vault_type not in KMIP_CONFIGS: + pytest.skip(f"Unknown vault_type '{vault_type}'. Available: {list(KMIP_CONFIGS.keys())}") + if vault_type == "fortanix" and ( + not os.environ.get("FORTANIX_EMAIL", "").strip() or not os.environ.get("FORTANIX_PASSWORD", "").strip() + ): + pytest.skip("Fortanix KMIP requires FORTANIX_EMAIL and FORTANIX_PASSWORD environment variables") + if vault_type == "hashicorp" and not os.environ.get("HASHICORP_VAULT_LICENSE", "").strip(): + pytest.skip("HashiCorp Vault KMIP requires a HASHICORP_VAULT_LICENSE environment variable (Vault Enterprise license)") + + if not test_helper.kmip_helper: + test_helper.kmip_helper = KMIPHelper(KMIP_CONFIGS, cert_base_dir=TEST_BASE_DIR) + if not test_helper.kmip_helper.start_kmip_server(vault_type): + detail = getattr(test_helper.kmip_helper, "last_error", None) or "unknown" + pytest.fail(f"Failed to start KMIP server for vault_type={vault_type}. {detail}") + + test_helper.create_keyring_manifest("component_keyring_kmip") + config_file = test_helper.create_keyring_config( + "keyring_kmip", cert_dir=test_helper.kmip_helper.kmip_config["cert_dir"] + ) + + keyring_backup_opts = f"--xtrabackup-plugin-dir={test_helper.xtrabackup_dir}/../lib/plugin" + test_helper.backup_params = f"{CORE_FILE_OPT}" + test_helper.prepare_params = f"{CORE_FILE_OPT} {keyring_backup_opts} --component-keyring-config={config_file}" + test_helper.restore_params = test_helper.prepare_params + + test_helper.initialize_db() + test_helper.run_load(_pstress_tool_options(seconds=120, no_encryption=False)) + + test_helper.cleanup_fifo_state(cloud_params, ["full", "inc1", "inc2", "inc3"]) + full_target = test_helper.take_incremental_backup_fifo( + cloud_params, keyring_backup_opts=keyring_backup_opts + ) + test_helper.restore_datadir_from(full_target) + test_helper.check_tables() + + +def test_fifo_encrypted_backup(test_helper, cloud_params): + """Full backup encrypted with xbcrypt (--encrypt/--encrypt-key, not + keyring-based), streamed via FIFO to SeaweedFS.""" + test_helper.backup_params = f"--encrypt=AES256 --encrypt-key={test_helper.encrypt_key} {CORE_FILE_OPT}" + test_helper.prepare_params = f"{CORE_FILE_OPT}" + test_helper.restore_params = "" + + test_helper.initialize_db() + test_helper.run_load(_pstress_tool_options(seconds=60)) + + test_helper.cleanup_fifo_state(cloud_params, ["full_backup"]) + full_target = test_helper.take_full_backup_fifo(cloud_params) + test_helper.restore_datadir_from(full_target) + test_helper.check_tables() + + +if __name__ == "__main__": + # Allow running as a script for easier debugging + import argparse + import sys + + parser = argparse.ArgumentParser(description="PXB FIFO-streamed cloud backup tests") + parser.add_argument( + "test_suites", + nargs="*", + choices=["Fifo_Backup_tests", "Fifo_Partition_tests", "Fifo_Encryption_tests", "Fifo_Kmip_tests"], + help="Test suites to run", + ) + parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output") + + args = parser.parse_args() + + if not args.test_suites: + print("This script tests FIFO-streamed backups (xtrabackup/xbcloud/xbstream named pipes) against a local SeaweedFS S3 gateway") + print("Assumption: PS and PXB are already installed as tarballs, and Docker is installed and running") + print("Usage: ") + print("1. Set environment variables (or use defaults):") + print(" export TEST_BASE_DIR=$HOME/inc_backup_load_tests") + print(" export XTRABACKUP_DIR=$HOME/percona-xtrabackup-8.4.0-7-Linux-x86_64.glibc2.36-minimal/bin") + print(" export MYSQLDIR=$HOME/Percona-Server-8.4.10-10-Linux.x86_64.glibc2.35-minimal") + print(" export QASCRIPTS=$HOME/server-qa") + print(" export LOAD_TOOL=pstress") + print(" export LOAD_TOOL_DIR=$HOME/pstress/src") + print(" export FIFO_STREAM=30") + print(" export FIFO_DIR=/tmp/xbstream_fifo") + print(" (If not set, defaults will be used from the script)") + print("2. Run the script as: pytest xbstream_fifo_tests.py -k -s -v") + print(" Or: python xbstream_fifo_tests.py ") + print(" Test Suites: ") + print(" Fifo_Backup_tests") + print(" Fifo_Partition_tests") + print(" Fifo_Encryption_tests") + print(" Fifo_Kmip_tests") + print(" ") + print("3. Logs are available at:", TEST_BASE_DIR, "(test-specific directories)") + sys.exit(1) + + pytest_args = [__file__, "-v"] + if args.verbose: + pytest_args.append("-s") + + test_mapping = { + "Fifo_Backup_tests": [ + "test_fifo_full_backup_and_restore", + "test_fifo_incremental_backup", + "test_fifo_compressed_backup", + ], + "Fifo_Partition_tests": ["test_fifo_partition_tables"], + "Fifo_Encryption_tests": ["test_fifo_keyring_file_backup", "test_fifo_encrypted_backup"], + "Fifo_Kmip_tests": ["test_fifo_kmip_backup"], + } + + selected_tests = [] + for suite in args.test_suites: + if suite in test_mapping: + selected_tests.extend(test_mapping[suite]) + + if selected_tests: + k_expr = " or ".join(selected_tests) + pytest_args.extend(["-k", k_expr]) + + pytest.main(pytest_args)