diff --git a/README.md b/README.md index 987c0566..19a5e64d 100644 --- a/README.md +++ b/README.md @@ -832,7 +832,7 @@ optional arguments: - Additionally, all specified test suites are downloaded, regardless of codec ### Local Mirror -When running fluster on multiple machines or in a CI environment, downloading test vectors from the internet for each run can be slow. Fluster supports a **local mirror** to serve resources from a server on your LAN instead. +When running fluster on multiple machines or in a CI environment, downloading test vectors from the internet for each run can be slow. Fluster supports a **local mirror** to serve resources from a server on your LAN instead. The mirror can be a plain HTTP-served directory or a radosgw (Ceph RGW) bucket (see [Setting up a mirror](#setting-up-a-mirror)). #### How it works @@ -865,14 +865,21 @@ The `--mirror` option works with all other download options: #### Setting up a mirror -Use the `scripts/mirror_sync.py` script to populate a directory with all test vector resources: +Use the `scripts/mirror_sync.py` script to populate a mirror with all test vector resources. It scans all test suite JSON files and fetches every source URL, laying files out in a tree that mirrors the original URL structure (`/`). Already-present files are skipped on subsequent runs, so the script is safe to re-run incrementally. + +The script supports two mirror backends: + +- a **local directory** served by any HTTP server (default), or +- a **radosgw (Ceph RGW) bucket** that files are uploaded to over HTTP. + +##### Option A: local directory + +Populate a directory with all resources: ```bash python3 scripts/mirror_sync.py -o /path/to/mirror -j 8 ``` -This will scan all test suite JSON files and download every source URL into a directory tree that mirrors the original URL structure. Already-downloaded files are skipped on subsequent runs. - Then serve the directory with any HTTP server: ```bash @@ -888,17 +895,51 @@ Use the same root path as the `--mirror` argument: ./fluster.py download --mirror http://mirror.local:8080/ ``` +##### Option B: radosgw bucket + +Instead of writing to a local directory, upload every resource to a radosgw +bucket by passing the radosgw endpoint with `--rgw-host` and the target bucket +with `--bucket`: + +```bash +python3 scripts/mirror_sync.py --rgw-host --bucket -j 8 +``` + +Each file is downloaded from its original source and then uploaded to the bucket +with an HTTP `PUT` (equivalent to +`curl -X PUT --data-binary @file http:////`), +preserving the same `/` key layout. Objects that already exist in +the bucket are skipped (checked via an HTTP `HEAD` request). + +Because the bucket keeps the same layout as a local mirror, the bucket URL can +be used directly as the `--mirror` base: + +```bash +./fluster.py download --mirror http://// +``` + +> **Note:** uploads use plain, unauthenticated HTTP `PUT`, so the bucket must +> allow anonymous writes. If your radosgw requires AWS SigV4 credentials, this +> script does not sign requests. + #### mirror_sync.py options ```bash python3 scripts/mirror_sync.py --help -usage: mirror_sync.py [-h] [-o OUTPUT] [-t TEST_SUITES_DIR] [-j JOBS] [-r RETRIES] +usage: mirror_sync.py [-h] [-o OUTPUT] [--rgw-host RGW_HOST] [--bucket BUCKET] + [-t TEST_SUITES_DIR] [-j JOBS] [-r RETRIES] options: -h, --help show this help message and exit -o OUTPUT, --output OUTPUT - output directory for the mirror tree (default: ./mirror) + output directory for the local mirror tree (default: + ./mirror). Ignored when --rgw-host is set. + --rgw-host RGW_HOST IP or host of the radosgw endpoint (e.g. RGW_HOST or + RGW_HOST:PORT). When set, resources are uploaded to a + bucket instead of a local directory. + --bucket BUCKET name of the radosgw bucket to fill (e.g. BUCKET). + Required when --rgw-host is set. -t TEST_SUITES_DIR, --test-suites-dir TEST_SUITES_DIR directory containing test suite JSON files -j JOBS, --jobs JOBS number of parallel downloads (default: 4) diff --git a/scripts/mirror_sync.py b/scripts/mirror_sync.py index 3c3c8ce3..427a279b 100755 --- a/scripts/mirror_sync.py +++ b/scripts/mirror_sync.py @@ -21,14 +21,18 @@ import json import os import sys +import tempfile +import urllib.error import urllib.parse +import urllib.request from functools import partial +from typing import Callable, List sys.path.append(os.path.join(os.path.dirname(__file__), "..")) from fluster import utils -def collect_source_urls(test_suites_dir: str) -> list[str]: +def collect_source_urls(test_suites_dir: str) -> List[str]: urls = [] for root, _, files in os.walk(test_suites_dir): for filename in files: @@ -53,14 +57,21 @@ def collect_source_urls(test_suites_dir: str) -> list[str]: def url_to_mirror_path(url: str) -> str: + """Returns the mirror key for a URL as a forward-slash path (``/``). + + This matches the layout produced by ``fluster``'s ``rewrite_url`` and is used + both as the radosgw object key and, after converting separators, as the local + mirror file path. It always uses '/' so bucket keys are consistent across + platforms (Windows included).""" parsed = urllib.parse.urlparse(url) - return os.path.join(parsed.netloc, parsed.path.lstrip("/")) + return f"{parsed.netloc}/{parsed.path.lstrip('/')}" def _sync_one(url: str, output_dir: str, retries: int) -> None: mirror_path = url_to_mirror_path(url) - dest_dir = os.path.join(output_dir, os.path.dirname(mirror_path)) - dest_file = os.path.join(output_dir, mirror_path) + local_path = mirror_path.replace("/", os.sep) + dest_dir = os.path.join(output_dir, os.path.dirname(local_path)) + dest_file = os.path.join(output_dir, local_path) if os.path.exists(dest_file): print(f" SKIP (exists): {mirror_path}") @@ -70,28 +81,101 @@ def _sync_one(url: str, output_dir: str, retries: int) -> None: utils.download(url, dest_dir, max_retries=retries) -def sync_urls(urls: list[str], output_dir: str, jobs: int, retries: int) -> None: +def _bucket_object_url(rgw_host: str, bucket: str, key: str) -> str: + """Builds the radosgw object URL for a given key, e.g. + http:////.""" + base = rgw_host if "://" in rgw_host else f"http://{rgw_host}" + # Keep ':' and '/' unquoted so the object key layout matches exactly what + # fluster's rewrite_url (used by `fluster download --mirror`) requests. + quoted_key = urllib.parse.quote(key.lstrip("/"), safe="/:") + return f"{base.rstrip('/')}/{urllib.parse.quote(bucket, safe='')}/{quoted_key}" + + +def _object_exists(object_url: str, timeout: int = 60) -> bool: + """Returns True if the object already exists in the bucket (HTTP HEAD 2xx).""" + req = urllib.request.Request(object_url, method="HEAD") + try: + with urllib.request.urlopen(req, timeout=timeout) as response: + return bool(200 <= response.status < 300) + except urllib.error.HTTPError: + return False + except urllib.error.URLError: + return False + + +def _put_object(object_url: str, src_path: str, timeout: int = 300) -> None: + """Uploads a local file to the bucket via HTTP PUT, equivalent to + 'curl -X PUT --data-binary @src_path object_url'.""" + size = os.path.getsize(src_path) + with open(src_path, "rb") as src: + req = urllib.request.Request(object_url, data=src, method="PUT") + req.add_header("Content-Type", "application/octet-stream") + req.add_header("Content-Length", str(size)) + with urllib.request.urlopen(req, timeout=timeout) as response: + if not 200 <= response.status < 300: + raise RuntimeError(f"Unexpected status {response.status} uploading to {object_url}") + + +def _upload_one(url: str, rgw_host: str, bucket: str, retries: int) -> None: + key = url_to_mirror_path(url) + object_url = _bucket_object_url(rgw_host, bucket, key) + + if _object_exists(object_url): + print(f" SKIP (exists): {key}") + return + + last_exc = None + for attempt in range(retries + 1): + try: + with tempfile.TemporaryDirectory() as tmpdir: + print(f" DOWNLOAD: {key}") + utils.download(url, tmpdir, max_retries=retries) + src_file = os.path.join(tmpdir, key.rsplit("/", 1)[-1]) + print(f" UPLOAD: {object_url}") + _put_object(object_url, src_file) + return + except (OSError, RuntimeError, urllib.error.URLError) as e: + last_exc = e + if attempt < retries: + print(f" WARNING: attempt {attempt + 1} failed for {key}: {e}") + print(f" ERROR: failed to upload {key}: {last_exc}") + + +def sync_urls(urls: List[str], worker: Callable[[str], None], jobs: int) -> None: from multiprocessing import Pool if jobs <= 1: for url in urls: - _sync_one(url, output_dir, retries) + worker(url) else: with Pool(jobs) as pool: - pool.map(partial(_sync_one, output_dir=output_dir, retries=retries), urls) + pool.map(worker, urls) def main() -> None: parser = argparse.ArgumentParser( - description="Populate a local mirror directory with all fluster test suite resources. " - "The resulting directory can be served by any HTTP server (nginx, Apache, etc.) " - "and used with: fluster download --mirror http://HOST/ROOT" + description="Populate a mirror with all fluster test suite resources. " + "By default it fills a local directory tree that can be served by any HTTP " + "server (nginx, Apache, etc.). With --rgw-host/--bucket it instead uploads " + "the resources to a radosgw (Ceph RGW) bucket. Either mirror can be used with: " + "fluster download --mirror http://HOST/ROOT" ) parser.add_argument( "-o", "--output", default="mirror", - help="output directory for the mirror tree (default: ./mirror)", + help="output directory for the local mirror tree (default: ./mirror). Ignored when --rgw-host is set.", + ) + parser.add_argument( + "--rgw-host", + default=None, + help="IP or host of the radosgw endpoint (e.g. RGW_HOST or RGW_HOST:PORT). " + "When set, resources are uploaded to a bucket instead of a local directory.", + ) + parser.add_argument( + "--bucket", + default=None, + help="name of the radosgw bucket to fill (e.g. BUCKET). Required when --rgw-host is set.", ) parser.add_argument( "-t", @@ -115,8 +199,10 @@ def main() -> None: ) args = parser.parse_args() + if args.rgw_host and not args.bucket: + sys.exit("--bucket is required when --rgw-host is set") + test_suites_dir = os.path.abspath(args.test_suites_dir) - output_dir = os.path.abspath(args.output) if not os.path.isdir(test_suites_dir): sys.exit(f"Test suites directory not found: {test_suites_dir}") @@ -126,14 +212,23 @@ def main() -> None: sys.exit(f"No source URLs found in {test_suites_dir}") print(f"Found {len(urls)} unique source URLs in {test_suites_dir}") - print(f"Mirror output directory: {output_dir}\n") - os.makedirs(output_dir, exist_ok=True) - sync_urls(urls, output_dir, args.jobs, args.retries) - - print(f"\nDone. Serve {output_dir} with an HTTP server, e.g.:") - print(f" cd {output_dir} && python3 -m http.server 8080") - print("Then use: fluster download --mirror http://:8080/") + if args.rgw_host: + print(f"Uploading to radosgw bucket: {args.bucket} on {args.rgw_host}\n") + worker = partial(_upload_one, rgw_host=args.rgw_host, bucket=args.bucket, retries=args.retries) + sync_urls(urls, worker, args.jobs) + mirror_url = _bucket_object_url(args.rgw_host, args.bucket, "").rstrip("/") + "/" + print("\nDone. Use the bucket as a mirror with:") + print(f" fluster download --mirror {mirror_url}") + else: + output_dir = os.path.abspath(args.output) + print(f"Mirror output directory: {output_dir}\n") + os.makedirs(output_dir, exist_ok=True) + worker = partial(_sync_one, output_dir=output_dir, retries=args.retries) + sync_urls(urls, worker, args.jobs) + print(f"\nDone. Serve {output_dir} with an HTTP server, e.g.:") + print(f" cd {output_dir} && python3 -m http.server 8080") + print("Then use: fluster download --mirror http://:8080/") if __name__ == "__main__": diff --git a/tests/test_mirror.py b/tests/test_mirror.py index 6a29a268..b4eae64e 100644 --- a/tests/test_mirror.py +++ b/tests/test_mirror.py @@ -21,13 +21,20 @@ import functools import http.server +import importlib.util import os import tempfile import threading import unittest +import urllib.parse from fluster import utils +_MIRROR_SYNC_PATH = os.path.join(os.path.dirname(__file__), "..", "scripts", "mirror_sync.py") +_spec = importlib.util.spec_from_file_location("mirror_sync", _MIRROR_SYNC_PATH) +mirror_sync = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(mirror_sync) + class TestRewriteUrl(unittest.TestCase): def test_basic_https(self) -> None: @@ -166,5 +173,123 @@ def test_download_without_mirror(self) -> None: server.shutdown() +def _key_to_fs_path(root: str, key: str) -> str: + """Maps an object key (URL path) to a filesystem path under ``root``. + + Object keys can contain characters that are illegal in Windows paths (most + notably ':' from ``host:port``) and may be percent-encoded. Decode them and + replace ':' so the fake bucket can store them on any platform.""" + key = urllib.parse.unquote(key).lstrip("/") + parts = [part.replace(":", "_") for part in key.split("/")] + return os.path.join(root, *parts) + + +class _BucketHandler(http.server.BaseHTTPRequestHandler): + """Minimal radosgw-like handler storing objects under a root dir.""" + + root: str = "" + + def log_message(self, fmt, *args): + pass + + def _object_path(self) -> str: + return _key_to_fs_path(self.root, self.path) + + def do_HEAD(self): # noqa: N802 + if os.path.isfile(self._object_path()): + self.send_response(200) + else: + self.send_response(404) + self.end_headers() + + def do_PUT(self): # noqa: N802 + length = int(self.headers.get("Content-Length", 0)) + data = self.rfile.read(length) + path = self._object_path() + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "wb") as f: + f.write(data) + self.send_response(200) + self.end_headers() + + +class TestBucketUpload(unittest.TestCase): + def _serve_bucket(self, root: str) -> tuple: + handler = type("_Handler", (_BucketHandler,), {"root": root}) + server = http.server.HTTPServer(("127.0.0.1", 0), handler) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server, port + + def test_upload_puts_object_with_mirror_key(self) -> None: + test_content = b"bucket upload content" + with tempfile.TemporaryDirectory() as tmpdir: + # Source HTTP server. + source_root = os.path.join(tmpdir, "source") + src_sub = os.path.join(source_root, "data") + os.makedirs(src_sub, exist_ok=True) + with open(os.path.join(src_sub, "vector.bin"), "wb") as f: + f.write(test_content) + src_handler = functools.partial(_SilentHandler, directory=source_root) + src_server = http.server.HTTPServer(("127.0.0.1", 0), src_handler) + src_port = src_server.server_address[1] + threading.Thread(target=src_server.serve_forever, daemon=True).start() + + # Bucket server. + bucket_root = os.path.join(tmpdir, "bucket") + os.makedirs(bucket_root, exist_ok=True) + bucket_server, bucket_port = self._serve_bucket(bucket_root) + + try: + url = f"http://127.0.0.1:{src_port}/data/vector.bin" + mirror_sync._upload_one( # noqa: SLF001 + url, + rgw_host=f"127.0.0.1:{bucket_port}", + bucket="test-bucket", + retries=1, + ) + expected = _key_to_fs_path(bucket_root, f"test-bucket/127.0.0.1:{src_port}/data/vector.bin") + self.assertTrue(os.path.exists(expected), f"missing {expected}") + with open(expected, "rb") as f: + self.assertEqual(f.read(), test_content) + finally: + src_server.shutdown() + bucket_server.shutdown() + + def test_upload_skips_existing_object(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + bucket_root = os.path.join(tmpdir, "bucket") + existing = _key_to_fs_path(bucket_root, "test-bucket/127.0.0.1:9/data/vector.bin") + os.makedirs(os.path.dirname(existing), exist_ok=True) + with open(existing, "wb") as f: + f.write(b"already here") + + bucket_server, bucket_port = self._serve_bucket(bucket_root) + try: + # Source points at an unreachable port; if not skipped, it would fail. + url = "http://127.0.0.1:9/data/vector.bin" + mirror_sync._upload_one( # noqa: SLF001 + url, + rgw_host=f"127.0.0.1:{bucket_port}", + bucket="test-bucket", + retries=0, + ) + with open(existing, "rb") as f: + self.assertEqual(f.read(), b"already here") + finally: + bucket_server.shutdown() + + +class TestBucketObjectUrl(unittest.TestCase): + def test_host_without_scheme(self) -> None: + result = mirror_sync._bucket_object_url("rgw.example.com", "mybucket", "example.com/a/b.bin") # noqa: SLF001 + self.assertEqual(result, "http://rgw.example.com/mybucket/example.com/a/b.bin") + + def test_host_with_scheme(self) -> None: + result = mirror_sync._bucket_object_url("http://rgw.example.com/", "bucket", "/host/file.bin") # noqa: SLF001 + self.assertEqual(result, "http://rgw.example.com/bucket/host/file.bin") + + if __name__ == "__main__": unittest.main(verbosity=2)