Validate replicated IOR recovery after daemon loss (6b9b59c0) · Commits · hpc / gekkofs

REPLICA.md

+6 −0

Changes for REPLICA.md: 6 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -37,6 +37,8 @@ Implemented today:
		- Replica-count validation against the active host count.
		- Integration coverage using real daemon `SIGKILL`/restart and surviving-copy
		read fallback.
		- Optional IOR integration coverage for a completed replicated write followed
		by one daemon kill and a verified two-rank read.

		Important limitations:

		@@ -78,6 +80,10 @@ Important limitations:
		- The integration test validates real daemon failure, fallback, restart, and
		continued access, but does not claim durable repair completion across a
		client restart.
		- Active I/O interruption is stricter than post-write recovery: an IOR write
		already in progress may fail when a daemon disappears. The supported CI
		scenario currently verifies that completed replicated data remains readable
		and checkable after one daemon is killed.

		### Re-evaluated implementation status

include/client/logging.hpp

+13 −1

Changes for include/client/logging.hpp: 13 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -436,7 +436,19 @@ struct logger {
		lineno);
		}

		fmt::format_to(std::back_inserter(buffer), std::forward<Args>(args)...);
		try {
		fmt::format_to(std::back_inserter(buffer),
		std::forward<Args>(args)...);
		} catch(const fmt::format_error&) {
		// Logging must never terminate the intercepted application. This
		// fallback is especially important on failed RPC paths, where an
		// error report may be emitted while the client is already
		// unwinding from an exception.
		fmt::format_to(std::back_inserter(buffer), "[invalid log format]");
		} catch(...) {
		fmt::format_to(std::back_inserter(buffer),
		"[log formatting failure]");
		}
		fmt::format_to(std::back_inserter(buffer), "\n");
		detail::log_buffer(log_fd_, buffer);
		}

src/client/rpc/forward_data.cpp

+30 −16

Changes for src/client/rpc/forward_data.cpp: 30 added lines, 16 removed lines.
Original line number	Diff line number	Diff line
		@@ -720,33 +720,47 @@ forward_get_chunk_stat() {
		__func__);
		}

		std::unordered_set<unsigned int> hosts;
		// cppcheck-suppress useStlAlgorithm
		auto unavailable = CTX->unavailable_hosts();
		std::vector<uint64_t> hosts;
		hosts.reserve(CTX->hosts().size());
		for(std::size_t i = 0; i < CTX->hosts().size(); ++i) {
		hosts.insert(i);
		if(unavailable.count(i) == 0) {
		hosts.push_back(i);
		}
		}

		unsigned long chunk_size = gkfs::config::rpc::chunksize;
		unsigned long chunk_total = 0;
		unsigned long chunk_free = 0;
		std::size_t successful_hosts = 0;
		auto stat_rpc = CTX->rpc_engine()->define(gkfs::rpc::tag::get_chunk_stat);

		int err = forward_data_helper<gkfs::rpc::rpc_chunk_stat_in_t,
		gkfs::rpc::rpc_chunk_stat_out_t>(
		gkfs::rpc::tag::get_chunk_stat, hosts,
		[](size_t target) {
		for(const auto host : hosts) {
		gkfs::rpc::rpc_chunk_stat_in_t in;
		in.dummy = 0;
		return in;
		},
		[&](const gkfs::rpc::rpc_chunk_stat_out_t& out) {
		try {
		const auto out =
		forward_with_timeout(stat_rpc, CTX->hosts().at(host), in)
		.template as<gkfs::rpc::rpc_chunk_stat_out_t>();
		if(out.err != 0) {
		CTX->mark_host_failure(host);
		continue;
		}
		CTX->mark_host_success(host);
		chunk_total += out.chunk_total;
		chunk_free += out.chunk_free;
		});
		++successful_hosts;
		} catch(const std::exception& ex) {
		LOG(ERROR, "{}() chunk stat RPC on host {} failed: {}", __func__,
		host, ex.what());
		CTX->mark_host_failure(host);
		}
		}

		if(err)
		return make_pair(err, ChunkStat{});
		else
		return make_pair(0, ChunkStat{chunk_size, chunk_total, chunk_free});
		if(successful_hosts == 0) {
		return make_pair(EIO, ChunkStat{});
		}
		return make_pair(0, ChunkStat{gkfs::config::rpc::chunksize, chunk_total,
		chunk_free});
		}

		} // namespace gkfs::rpc

src/client/rpc/forward_metadata.cpp

+12 −1

Changes for src/client/rpc/forward_metadata.cpp: 12 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -263,15 +263,26 @@ forward_stat(const std::string& path, string& attr, string& inline_data,
		in.path = path;
		in.include_inline = include_inline;

		try {
		auto out = gkfs::rpc::forward_call<gkfs::rpc::rpc_stat_out_t>(
		CTX->rpc_engine(), endp, gkfs::rpc::tag::stat, in, __func__, path);
		CTX->rpc_engine(), endp, gkfs::rpc::tag::stat, in, __func__,
		path);

		if(out.err == 0) {
		CTX->mark_host_success(target.host);
		attr = out.db_val;
		inline_data.assign(out.inline_data.begin(), out.inline_data.end());
		} else {
		CTX->mark_host_failure(target.host);
		}

		return out.err;
		} catch(const std::exception& ex) {
		LOG(ERROR, "{}() stat RPC for '{}' copy {} failed: {}", __func__, path,
		copy, ex.what());
		CTX->mark_host_failure(target.host);
		return EIO;
		}
		}

		int

tests/integration/data/test_replication.py

+96 −0

Changes for tests/integration/data/test_replication.py: 96 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import os
		import shutil
		import signal
		import subprocess
		import time
		@@ -385,6 +386,101 @@ def test_replication_repair_journal_ignores_bad_records(
		daemon.shutdown()


		def test_replication_ior_read_survives_daemon_kill(
		test_workspace, request
		):
		"""IOR reads verified replicated data after one daemon is killed."""
		ior = shutil.which("ior")
		mpirun = shutil.which("mpirun") or shutil.which("mpiexec")
		if ior is None or mpirun is None:
		pytest.skip("IOR and MPI launcher are required")

		interface = request.config.getoption("--interface")
		d00 = Daemon(
		interface,
		"rocksdb",
		test_workspace,
		rootdir_suffix="ior_replica_0",
		keep_hosts=True,
		).run()
		d01 = Daemon(
		interface,
		"rocksdb",
		test_workspace,
		rootdir_suffix="ior_replica_1",
		keep_hosts=True,
		).run()

		file_prefix = d00.mountdir / "ior-replication"
		base_env = os.environ.copy()
		base_env.update(
		{
		"LIBGKFS_HOSTS_FILE": str(d00._hostfile),
		"LIBGKFS_NUM_REPL": "1",
		"LD_PRELOAD": str(
		Path(d00._workspace.libdirs[0]) / "libgkfs_intercept.so"
		),
		# IOR uses system MPI/libfabric. Keep system libfabric before the
		# GekkoFS dependency prefix, which provides Mercury and GekkoFS.
		"LD_LIBRARY_PATH": "/usr/lib:/lib:" + ":".join(
		str(path) for path in d00._workspace.libdirs
		),
		"OMPI_ALLOW_RUN_AS_ROOT": "1",
		"OMPI_ALLOW_RUN_AS_ROOT_CONFIRM": "1",
		}
		)

		def run_ior(*args):
		return subprocess.run(
		[
		mpirun,
		"--allow-run-as-root",
		"--map-by",
		"slot",
		"-np",
		"2",
		ior,
		"-a",
		"POSIX",
		"-i",
		"1",
		"-o",
		str(file_prefix),
		"-b",
		"8m",
		"-t",
		"1m",
		"-x",
		*args,
		"-F",
		"-k",
		],
		env=base_env,
		stdout=subprocess.PIPE,
		stderr=subprocess.STDOUT,
		text=True,
		timeout=120,
		)

		try:
		write = run_ior("-w")
		assert write.returncode == 0, write.stdout
		assert "write" in write.stdout

		os.kill(d01._proc.pid, signal.SIGKILL)
		d01._proc.wait(timeout=10)
		assert d01._proc.poll() is not None

		read = run_ior("-r", "-W")
		assert read.returncode == 0, read.stdout
		assert "read" in read.stdout
		finally:
		if d01._proc is not None and d01._proc.poll() is None:
		d01.shutdown()
		if d00._proc is not None and d00._proc.poll() is None:
		d00.shutdown()


		def test_replication_rejects_mutation_safely(
		test_workspace, gkfs_client, gkfs_shell, request
		):