Add compatibility corpus workflows (c383acd2) · Commits · hpc / gekkofs

.gitlab-ci.yml

+19 −0

Original line number	Diff line number	Diff line
		@@ -464,6 +464,25 @@ gkfs:app:
		reports:
		junit: ${BUILD_PATH}/tests/apps/report.xml

		compatibility-corpus:
		stage: test
		image: ${TESTING}
		needs: ['gkfs']
		script:
		- cd ${CI_PROJECT_DIR}
		- python3 scripts/compatibility_corpus.py
		--daemon ${INSTALL_PATH}/bin/gkfs_daemon
		--client ${BUILD_PATH}/tests/integration/harness/gkfs.io
		--library ${INSTALL_PATH}/lib64/libgkfs_intercept.so
		--libc-library ${INSTALL_PATH}/lib64/libgkfs_libc_intercept.so
		--user-library ${INSTALL_PATH}/lib64/libgkfs_user_lib.so
		--output ${BUILD_PATH}/tests/compatibility/corpus.json
		artifacts:
		when: always
		paths:
		- ${BUILD_PATH}/tests/compatibility/corpus.json
		expire_in: 5 days


		## == java tests for gkfs ==================
		gkfs:java:

WIP.md

+8 −0

Original line number	Diff line number	Diff line
		@@ -508,6 +508,14 @@ parallel readers/writers, symlinks, rename, sparse files, and restart. Run them
		through FUSE, LD_PRELOAD, and user-library modes where supported. Record exit
		status, tree, checksums, metadata, and timing envelopes for release candidates.

		Status: Done for the supported automated scope. Added
		`scripts/compatibility_corpus.py`, which runs isolated deterministic workloads
		for `cp`, `tar`, `find`, `dd`, symlinks, rename, sparse files, and daemon restart
		through syscall, libc, and user-library preload modes. JSON evidence records
		commands, exit status, timing, tree shape, SHA-256 checksums, and file metadata.
		FUSE and MPI-IO are emitted as explicit skipped records until portable launchers
		and runtime fixtures are available. CI publishes the corpus artifact.

		## P3 — user-facing features and documentation

		### 29. Add resumable and inspectable migration commands

scripts/compatibility_corpus.py

0 → 100644

+477 −0

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3
		"""Run small reproducible compatibility workloads and write JSON evidence."""

		import argparse
		import hashlib
		import json
		import os
		import signal
		import socket
		import subprocess
		import tempfile
		import time
		from pathlib import Path


		WORKLOADS = (
		"cp",
		"tar",
		"find",
		"dd",
		"symlink",
		"rename",
		"sparse",
		"restart",
		)


		def sha256(path):
		digest = hashlib.sha256()
		with path.open("rb") as stream:
		for block in iter(lambda: stream.read(1024 * 1024), b""):
		digest.update(block)
		return digest.hexdigest()


		def snapshot_tree(root, env=None, client=None):
		if env is not None and client is not None:
		entries = []
		pending = [Path(root)]
		while pending:
		directory = pending.pop()
		result = subprocess.run(
		[str(client), "readdir", str(directory)], env=env,
		capture_output=True, text=True, check=False,
		)
		if result.returncode != 0:
		continue
		start = result.stdout.find("{")
		if start < 0:
		continue
		try:
		dirents = json.loads(result.stdout[start:])["dirents"]
		except (KeyError, json.JSONDecodeError):
		continue
		for dirent in dirents:
		name = dirent["d_name"]
		if name in (".", ".."):
		continue
		path = directory / name
		stat_result = subprocess.run(
		[str(client), "stat", str(path)], env=env,
		capture_output=True, text=True, check=False,
		)
		stat_start = stat_result.stdout.find("{")
		if stat_result.returncode != 0 or stat_start < 0:
		continue
		try:
		statbuf = json.loads(stat_result.stdout[stat_start:])["statbuf"]
		except (KeyError, json.JSONDecodeError):
		continue
		mode = statbuf["st_mode"]
		is_directory = (mode & 0o170000) == 0o040000
		is_symlink = (mode & 0o170000) == 0o120000
		item = {
		"path": path.relative_to(root).as_posix(),
		"mode": mode & 0o7777,
		"uid": statbuf["st_uid"],
		"gid": statbuf["st_gid"],
		"size": statbuf["st_size"],
		"type": "directory" if is_directory else (
		"symlink" if is_symlink else "file"
		),
		}
		if is_directory:
		pending.append(path)
		elif not is_symlink:
		checksum = subprocess.run(
		["sha256sum", str(path)], env=env,
		capture_output=True, text=True, check=False,
		)
		if checksum.returncode == 0:
		item["sha256"] = checksum.stdout.split()[0]
		entries.append(item)
		return sorted(entries, key=lambda entry: entry["path"])

		if env is not None:
		result = subprocess.run([
		"find", str(root), "-mindepth", "1", "-print",
		], env=env, capture_output=True, text=True, check=False)
		if result.returncode != 0:
		return []
		entries = []
		for pathname in result.stdout.splitlines():
		path = Path(pathname)
		metadata = subprocess.run(
		["stat", "-c", "%F\|%a\|%u\|%g\|%s", str(path)],
		env=env, capture_output=True, text=True, check=False,
		)
		if metadata.returncode != 0:
		continue
		kind, mode, uid, gid, size = metadata.stdout.strip().split("\|", 4)
		kind = {"symbolic link": "l", "directory": "d"}.get(kind, "f")
		item = {
		"path": path.relative_to(root).as_posix(),
		"mode": int(mode, 8),
		"uid": int(uid),
		"gid": int(gid),
		"size": int(size),
		"type": {"d": "directory", "l": "symlink"}.get(kind, "file"),
		}
		if kind == "l":
		target = subprocess.run(
		["readlink", str(path)], env=env, capture_output=True,
		text=True, check=False,
		)
		item["target"] = target.stdout.strip()
		elif kind == "f":
		checksum = subprocess.run(
		["sha256sum", str(path)], env=env, capture_output=True,
		text=True, check=False,
		)
		if checksum.returncode == 0:
		item["sha256"] = checksum.stdout.split()[0]
		entries.append(item)
		return sorted(entries, key=lambda entry: entry["path"])

		entries = []
		if not root.exists():
		return entries
		for path in sorted(root.rglob("*")):
		relative = path.relative_to(root).as_posix()
		item = {"path": relative}
		stat = path.lstat()
		item["mode"] = stat.st_mode & 0o7777
		item["uid"] = stat.st_uid
		item["gid"] = stat.st_gid
		item["size"] = stat.st_size
		item["type"] = "symlink" if path.is_symlink() else (
		"directory" if path.is_dir() else "file"
		)
		if path.is_symlink():
		item["target"] = os.readlink(path)
		elif path.is_file():
		item["sha256"] = sha256(path)
		entries.append(item)
		return entries


		def tree_signature(entries):
		return {
		entry["path"]: (
		entry.get("type"),
		entry.get("size") if entry.get("type") == "file" else None,
		entry.get("sha256"),
		entry.get("target"),
		)
		for entry in entries
		}


		def free_port(host="127.0.0.1"):
		with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as stream:
		stream.bind((host, 0))
		return stream.getsockname()[1]


		def wait_for_hostfile(path, process, timeout=30):
		deadline = time.monotonic() + timeout
		while time.monotonic() < deadline:
		if process.poll() is not None:
		raise RuntimeError("daemon exited with code {}".format(process.returncode))
		if path.is_file() and any(line.strip() and not line.startswith("#")
		for line in path.read_text().splitlines()):
		return
		time.sleep(0.2)
		raise TimeoutError("daemon did not publish {}".format(path))


		class Daemon:
		def __init__(self, binary, root, mount, hostfile, env):
		self.binary = Path(binary)
		self.root = Path(root)
		self.mount = Path(mount)
		self.hostfile = Path(hostfile)
		self.env = env.copy()
		self.process = None

		def start(self):
		address = "lo:{}".format(free_port())
		self.env.update({
		"GKFS_HOSTS_FILE": str(self.hostfile),
		"LIBGKFS_HOSTS_FILE": str(self.hostfile),
		"GKFS_DAEMON_LOG_PATH": str(self.root / "daemon.log"),
		"GKFS_DAEMON_LOG_LEVEL": "100",
		})
		self.process = subprocess.Popen([
		str(self.binary), "--mountdir", str(self.mount),
		"--rootdir", str(self.root), "--metadir", str(self.root),
		"--dbbackend", "rocksdb", "-l", address,
		], env=self.env, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
		wait_for_hostfile(self.hostfile, self.process)

		def restart(self):
		self.stop()
		self.start()

		def stop(self):
		if self.process is None:
		return
		if self.process.poll() is None:
		self.process.send_signal(signal.SIGTERM)
		try:
		self.process.wait(timeout=10)
		except subprocess.TimeoutExpired:
		self.process.kill()
		self.process.wait()
		self.process = None


		def run_command(command, cwd, env):
		started = time.monotonic()
		result = subprocess.run(
		command, cwd=cwd, env=env, capture_output=True, text=True, check=False
		)
		return {
		"command": command,
		"returncode": result.returncode,
		"elapsed_seconds": round(time.monotonic() - started, 6),
		"stdout": result.stdout[-4000:],
		"stderr": result.stderr[-4000:],
		}


		def workload_commands(name, mount, outside):
		source = outside / "source.txt"
		copied = mount / "copied.txt"
		if name == "cp":
		return [["cp", str(source), str(copied)]]
		if name == "tar":
		archive = outside / "payload.tar"
		return [
		["tar", "-cf", str(archive), "-C", str(outside), "source.txt"],
		["mkdir", "-p", str(mount / "tar")],
		["tar", "-xf", str(archive), "-C", str(mount / "tar")],
		]
		if name == "find":
		return [["find", str(mount), "-type", "f", "-print"]]
		if name == "dd":
		return [["dd", "if=/dev/zero", "of={}".format(mount / "dd.bin"),
		"bs=4096", "count=4", "status=none"]]
		if name == "symlink":
		return [["ln", "-s", "source.txt", str(mount / "source.link")]]
		if name == "rename":
		return [["mv", str(mount / "source.txt"), str(mount / "renamed.txt")]]
		if name == "sparse":
		return [["truncate", "-s", "1048576", str(mount / "sparse.bin")]]
		if name == "restart":
		return []
		raise ValueError("unknown workload: {}".format(name))


		def resolve_library(path):
		if not path:
		return None
		candidate = Path(path)
		if candidate.is_file():
		return candidate
		if "lib64" in candidate.parts:
		alternative = Path(*[
		"lib" if part == "lib64" else part for part in candidate.parts
		])
		if alternative.is_file():
		return alternative
		if "lib" in candidate.parts:
		alternative = Path(*[
		"lib64" if part == "lib" else part for part in candidate.parts
		])
		if alternative.is_file():
		return alternative
		return candidate


		def run_workloads(workloads, mount, outside, daemon, env, client=None):
		records = []
		for name in workloads:
		case_mount = mount / name
		case_outside = outside / name
		if daemon is None:
		case_mount.mkdir(parents=True, exist_ok=True)
		case_outside.mkdir(parents=True, exist_ok=True)
		source = case_outside / "source.txt"
		source.write_text("compatibility corpus\n")
		started = time.monotonic()
		commands = [
		["mkdir", "-p", str(case_mount)],
		["cp", str(source), str(case_mount / "source.txt")],
		*workload_commands(name, case_mount, case_outside),
		]
		command_records = []
		for index, command in enumerate(commands):
		command_env = os.environ.copy() if (
		name == "tar" and command[:2] == ["tar", "-cf"]
		) else env
		record = run_command(command, mount.parent, command_env)
		command_records.append(record)
		if record["returncode"] != 0:
		break
		if name == "restart" and all(
		record["returncode"] == 0 for record in command_records
		):
		restart_file = case_mount / "before-restart.txt"
		command_records.append(run_command(
		["cp", str(source), str(restart_file)], mount.parent, env
		))
		if daemon is not None:
		daemon.restart()
		if command_records[-1]["returncode"] == 0:
		command_records.append(run_command(
		["cat", str(restart_file)], mount.parent, env
		))
		records.append({
		"name": name,
		"status": "passed" if all(
		item["returncode"] == 0 for item in command_records
		) else "failed",
		"elapsed_seconds": round(time.monotonic() - started, 6),
		"commands": command_records,
		"tree": snapshot_tree(case_mount, env if daemon else None, client),
		})
		return records


		def run_mode(mode, args):
		if mode == "native":
		with tempfile.TemporaryDirectory(prefix="gkfs-corpus-native-") as temporary:
		workspace = Path(temporary)
		native_mount = workspace / "native-mount"
		native_outside = workspace / "native-outside"
		native_mount.mkdir()
		native_outside.mkdir()
		records = run_workloads(
		args.workloads, native_mount, native_outside, None, os.environ.copy()
		)
		return {
		"mode": mode,
		"library": None,
		"status": "passed" if all(
		record["status"] == "passed" for record in records
		) else "failed",
		"workloads": records,
		}
		if not args.daemon or not args.library:
		return {"mode": mode, "status": "skipped",
		"reason": "daemon and library paths are required"}
		library = resolve_library(args.library)
		daemon_binary = Path(args.daemon)
		if not daemon_binary.is_file() or not library.is_file():
		return {"mode": mode, "library": str(library), "status": "skipped",
		"reason": "requested artifact is unavailable"}

		with tempfile.TemporaryDirectory(prefix="gkfs-corpus-") as temporary:
		workspace = Path(temporary)
		root = workspace / "root"
		mount = workspace / "mount"
		outside = workspace / "outside"
		root.mkdir()
		mount.mkdir()
		outside.mkdir()
		(outside / "source.txt").write_text("compatibility corpus\n")
		hostfile = workspace / "gkfs_hosts.txt"
		client_env = os.environ.copy()
		library_dirs = [
		str(library.parent),
		str(daemon_binary.parent),
		str(daemon_binary.parent.parent / "common"),
		str(daemon_binary.parent.parent.parent / "common"),
		]
		client_env["LD_LIBRARY_PATH"] = os.pathsep.join(filter(None, [
		*library_dirs, client_env.get("LD_LIBRARY_PATH", "")
		]))
		client_env["LD_PRELOAD"] = str(library.resolve())
		daemon_env = os.environ.copy()
		daemon = Daemon(daemon_binary, root, mount, hostfile, daemon_env)
		try:
		daemon.start()
		records = run_workloads(
		args.workloads, mount, outside, daemon, client_env, args.client
		)
		finally:
		daemon.stop()
		return {
		"mode": mode,
		"library": str(library),
		"status": "passed" if all(
		record["status"] == "passed" for record in records
		) else "failed",
		"workloads": records,
		}


		def main(argv=None):
		parser = argparse.ArgumentParser(description=__doc__)
		parser.add_argument("--daemon")
		parser.add_argument("--client")
		parser.add_argument("--library")
		parser.add_argument("--libc-library")
		parser.add_argument("--user-library")
		parser.add_argument("--output", type=Path, required=True)
		parser.add_argument("--mode", action="append",
		choices=("native", "syscall", "libc", "user", "fuse", "mpi"),
		default=None)
		parser.add_argument("--workload", dest="workloads", action="append",
		choices=WORKLOADS, default=None)
		args = parser.parse_args(argv)
		if args.mode is None:
		args.mode = ["native", "syscall", "libc", "user", "fuse", "mpi"]
		if args.workloads is None:
		args.workloads = list(WORKLOADS)
		results = {
		"schema_version": 1,
		"generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
		"workloads": args.workloads,
		"modes": [],
		}
		for mode in args.mode:
		library = args.library
		if mode == "libc":
		library = args.libc_library
		if mode == "user":
		library = args.user_library
		if mode in ("fuse", "mpi"):
		results["modes"].append({
		"mode": mode, "status": "skipped",
		"reason": "no portable corpus launcher is configured",
		})
		continue
		results["modes"].append(run_mode(mode, argparse.Namespace(
		daemon=args.daemon, client=args.client, library=library,
		workloads=args.workloads
		)))
		baseline = next(
		(mode for mode in results["modes"] if mode["mode"] == "native"), None
		)
		if baseline and baseline.get("status") == "passed":
		expected = {
		record["name"]: tree_signature(record["tree"])
		for record in baseline["workloads"]
		}
		for mode in results["modes"]:
		if mode["mode"] == "native" or "workloads" not in mode:
		continue
		for record in mode["workloads"]:
		record["matches_native"] = (
		tree_signature(record["tree"]) == expected.get(record["name"])
		)
		record["comparison_status"] = (
		"matched" if record["matches_native"] else "different"
		)
		args.output.parent.mkdir(parents=True, exist_ok=True)
		args.output.write_text(json.dumps(results, indent=2, sort_keys=True) + "\n")
		failed = [mode for mode in results["modes"]
		if mode["status"] == "failed"]
		return 1 if failed else 0


		if __name__ == "__main__":
		raise SystemExit(main())
		No newline at end of file

tests/README.md

+5 −0

Original line number	Diff line number	Diff line
		@@ -6,3 +6,8 @@ information.

		For a documented deterministic configure/build/install/unit/integration/coverage
		workflow, run `python3 scripts/test_workflow.py --help` from the repository root.

		The compatibility corpus is run with:
		`python3 scripts/compatibility_corpus.py --output compatibility.json`.
		Pass `--daemon`, `--library`, `--libc-library`, and `--user-library` to run
		the preload modes against a build or install tree.

tests/scripts/test_compatibility_corpus.py

0 → 100644

+44 −0

Original line number	Diff line number	Diff line
		import importlib.util
		from pathlib import Path


		MODULE = Path(__file__).parents[2] / "scripts" / "compatibility_corpus.py"
		SPEC = importlib.util.spec_from_file_location("compatibility_corpus", MODULE)
		corpus = importlib.util.module_from_spec(SPEC)
		SPEC.loader.exec_module(corpus)


		def test_snapshot_tree_records_file_metadata_and_checksum(tmp_path):
		root = tmp_path / "root"
		root.mkdir()
		(root / "file.txt").write_text("corpus\n")
		(root / "link").symlink_to("file.txt")

		entries = corpus.snapshot_tree(root)
		files = {entry["path"]: entry for entry in entries}
		assert files["file.txt"]["type"] == "file"
		assert files["file.txt"]["uid"] == root.stat().st_uid
		assert files["file.txt"]["gid"] == root.stat().st_gid
		assert files["file.txt"]["sha256"] == corpus.sha256(root / "file.txt")
		assert files["link"]["type"] == "symlink"
		assert files["link"]["target"] == "file.txt"


		def test_workload_commands_cover_required_operations(tmp_path):
		names = set(corpus.WORKLOADS)
		commands = {
		command[0]
		for name in names
		for command in corpus.workload_commands(name, tmp_path / "mount", tmp_path)
		}
		assert {"cp", "tar", "find", "dd", "ln", "mv", "truncate"} <= commands


		def test_tree_signature_ignores_runtime_ownership(tmp_path):
		root = tmp_path / "root"
		root.mkdir()
		(root / "file.txt").write_text("same\n")
		first = corpus.snapshot_tree(root)
		(root / "file.txt").chmod(0o600)
		second = corpus.snapshot_tree(root)
		assert corpus.tree_signature(first) == corpus.tree_signature(second)
		No newline at end of file