Add restart-safe mutation checkpoint recovery (609d9edf) · Commits · hpc / gekkofs

include/daemon/malleability/migration_checkpoint.hpp

0 → 100644

+98 −0

Changes for include/daemon/malleability/migration_checkpoint.hpp: 98 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		/*
		Copyright 2018-2025, Barcelona Supercomputing Center (BSC), Spain
		Copyright 2015-2025, Johannes Gutenberg Universitaet Mainz, Germany

		This file is part of GekkoFS.

		SPDX-License-Identifier: LGPL-3.0-or-later
		*/

		#ifndef GKFS_DAEMON_MIGRATION_CHECKPOINT_HPP
		#define GKFS_DAEMON_MIGRATION_CHECKPOINT_HPP

		#include <daemon/classes/fs_data.hpp>

		#include <cstdint>
		#include <sstream>
		#include <string>
		#include <unordered_map>

		namespace gkfs::malleable {

		enum class migration_checkpoint_state {
		malformed,
		complete,
		interrupted,
		failed,
		};

		struct migration_checkpoint_record {
		migration_checkpoint_state state{migration_checkpoint_state::malformed};
		gkfs::daemon::mutation_phase phase{gkfs::daemon::mutation_phase::idle};
		uint64_t old_placement_generation{};
		uint64_t new_placement_generation{};
		uint64_t placement_generation{};
		uint64_t jobs_total{};
		uint64_t jobs_completed{};
		uint64_t jobs_succeeded{};
		uint64_t jobs_failed{};
		uint64_t bytes_transferred{};
		};

		inline migration_checkpoint_record
		parse_migration_checkpoint(const std::string& content) {
		std::unordered_map<std::string, std::string> fields;
		std::istringstream input(content);
		std::string line;
		while(std::getline(input, line)) {
		const auto separator = line.find('=');
		if(separator == std::string::npos) {
		continue;
		}
		fields[line.substr(0, separator)] = line.substr(separator + 1);
		}

		migration_checkpoint_record record;
		const auto state = fields.find("state");
		if(state == fields.end()) {
		return record;
		}

		try {
		const auto number = [&](const char* key) {
		const auto it = fields.find(key);
		return it == fields.end() ? uint64_t{0} : std::stoull(it->second);
		};
		record.old_placement_generation = number("placement_old_generation");
		record.new_placement_generation = number("placement_new_generation");
		record.placement_generation = number("placement_generation");
		record.jobs_total = number("jobs_total");
		record.jobs_completed = number("jobs_completed");
		record.jobs_succeeded = number("jobs_succeeded");
		record.jobs_failed = number("jobs_failed");
		record.bytes_transferred = number("bytes_transferred");
		const auto phase = number("phase");
		if(phase > static_cast<uint64_t>(
		gkfs::daemon::mutation_phase::ready_to_finalize)) {
		return {};
		}
		record.phase = static_cast<gkfs::daemon::mutation_phase>(phase);

		if(state->second == "complete") {
		record.state = migration_checkpoint_state::complete;
		} else if(state->second == "failed") {
		record.state = migration_checkpoint_state::failed;
		} else if(state->second == "running" \|\| state->second == "planned") {
		record.state = migration_checkpoint_state::interrupted;
		} else {
		return {};
		}
		return record;
		} catch(const std::exception&) {
		return {};
		}
		}

		} // namespace gkfs::malleable

		#endif // GKFS_DAEMON_MIGRATION_CHECKPOINT_HPP
		No newline at end of file

src/daemon/malleability/malleable_manager.cpp

+12 −41

Changes for src/daemon/malleability/malleable_manager.cpp: 12 added lines, 41 removed lines.
Original line number	Diff line number	Diff line
		@@ -37,6 +37,7 @@
		*/

		#include <daemon/malleability/malleable_manager.hpp>
		#include <daemon/malleability/migration_checkpoint.hpp>
		#include <daemon/malleability/rpc/forward_redistribution.hpp>
		#include <daemon/backend/metadata/db.hpp>
		#include <daemon/backend/data/chunk_storage.hpp>
		@@ -208,55 +209,25 @@ MalleableManager::load_migration_checkpoint() {
		return;
		}

		std::unordered_map<std::string, std::string> fields;
		std::string line;
		while(std::getline(input, line)) {
		const auto separator = line.find('=');
		if(separator != std::string::npos) {
		fields.emplace(line.substr(0, separator),
		line.substr(separator + 1));
		}
		}

		const auto state_it = fields.find("state");
		if(state_it == fields.end()) {
		std::ostringstream contents;
		contents << input.rdbuf();
		const auto record =
		gkfs::malleable::parse_migration_checkpoint(contents.str());
		if(record.state == gkfs::malleable::migration_checkpoint_state::malformed) {
		return;
		}

		try {
		const auto old_generation =
		std::stoull(fields.count("placement_old_generation")
		? fields.at("placement_old_generation")
		: "0");
		const auto new_generation =
		std::stoull(fields.count("placement_new_generation")
		? fields.at("placement_new_generation")
		: "0");
		const auto committed_generation =
		std::stoull(fields.count("placement_generation")
		? fields.at("placement_generation")
		: "0");
		const auto phase = static_cast<gkfs::daemon::mutation_phase>(
		std::stoul(fields.count("phase") ? fields.at("phase") : "0"));

		if(state_it->second == "complete") {
		GKFS_DATA->restore_placement_generation(committed_generation);
		return;
		}

		if(state_it->second == "running" \|\| state_it->second == "failed" \|\|
		state_it->second == "complete") {
		GKFS_DATA->restore_mutation_recovery_state(
		old_generation, new_generation,
		phase == gkfs::daemon::mutation_phase::failed
		? phase
		: gkfs::daemon::mutation_phase::failed);
		GKFS_DATA->redist_failed(true);
		if(record.state ==
		gkfs::malleable::migration_checkpoint_state::complete) {
		GKFS_DATA->restore_placement_generation(
		record.placement_generation);
		return;
		}

		GKFS_DATA->restore_mutation_recovery_state(
		old_generation, new_generation,
		record.old_placement_generation,
		record.new_placement_generation,
		gkfs::daemon::mutation_phase::failed);
		GKFS_DATA->redist_failed(true);
		} catch(const std::exception& e) {

tests/unit/CMakeLists.txt

+1 −0

Changes for tests/unit/CMakeLists.txt: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -86,6 +86,7 @@ target_sources(unit_tests
		${CMAKE_CURRENT_LIST_DIR}/test_unique_fd.cpp
		${CMAKE_CURRENT_LIST_DIR}/test_repair_journal_lock.cpp
		${CMAKE_CURRENT_LIST_DIR}/test_fs_data_faults.cpp
		${CMAKE_CURRENT_LIST_DIR}/test_migration_checkpoint.cpp
		${CMAKE_SOURCE_DIR}/src/common/hostfile_management.cpp
		${CMAKE_SOURCE_DIR}/src/daemon/classes/fs_data.cpp)

tests/unit/test_migration_checkpoint.cpp

0 → 100644

+64 −0

Changes for tests/unit/test_migration_checkpoint.cpp: 64 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		/*
		Copyright 2018-2025, Barcelona Supercomputing Center (BSC), Spain
		Copyright 2015-2025, Johannes Gutenberg Universitaet Mainz, Germany

		This file is part of GekkoFS.

		SPDX-License-Identifier: GPL-3.0-or-later
		*/

		#include <daemon/malleability/migration_checkpoint.hpp>

		#include <catch2/catch_all.hpp>

		TEST_CASE("migration checkpoint parses complete placement state",
		"[resilience][mutation][checkpoint]") {
		const auto record = gkfs::malleable::parse_migration_checkpoint(
		"format=2\nstate=complete\nphase=0\n"
		"placement_old_generation=4\nplacement_new_generation=5\n"
		"placement_generation=5\njobs_total=7\njobs_completed=7\n"
		"jobs_succeeded=7\njobs_failed=0\nbytes_transferred=99\n");

		REQUIRE(record.state == gkfs::malleable::migration_checkpoint_state::complete);
		REQUIRE(record.placement_generation == 5);
		REQUIRE(record.old_placement_generation == 4);
		REQUIRE(record.jobs_total == 7);
		REQUIRE(record.bytes_transferred == 99);
		}

		TEST_CASE("migration checkpoint marks running work as interrupted",
		"[resilience][mutation][checkpoint]") {
		const auto record = gkfs::malleable::parse_migration_checkpoint(
		"format=2\nstate=running\nphase=2\n"
		"placement_old_generation=4\nplacement_new_generation=5\n"
		"placement_generation=4\njobs_total=7\njobs_completed=3\n");

		REQUIRE(record.state ==
		gkfs::malleable::migration_checkpoint_state::interrupted);
		REQUIRE(record.phase == gkfs::daemon::mutation_phase::running);
		REQUIRE(record.jobs_completed == 3);
		}

		TEST_CASE("migration checkpoint parses failed work",
		"[resilience][mutation][checkpoint]") {
		const auto record = gkfs::malleable::parse_migration_checkpoint(
		"state=failed\nphase=3\nplacement_old_generation=1\n"
		"placement_new_generation=2\nerror=migration failed\n");

		REQUIRE(record.state == gkfs::malleable::migration_checkpoint_state::failed);
		REQUIRE(record.phase == gkfs::daemon::mutation_phase::failed);
		REQUIRE(record.new_placement_generation == 2);
		}

		TEST_CASE("malformed migration checkpoints are ignored",
		"[resilience][mutation][checkpoint]") {
		const auto empty =
		gkfs::malleable::parse_migration_checkpoint("not-a-record\n");
		REQUIRE(empty.state ==
		gkfs::malleable::migration_checkpoint_state::malformed);

		const auto invalid = gkfs::malleable::parse_migration_checkpoint(
		"state=running\nphase=invalid\nplacement_old_generation=x\n");
		REQUIRE(invalid.state ==
		gkfs::malleable::migration_checkpoint_state::malformed);
		}
		No newline at end of file