Commit e3fcdf1e authored by Ramon Nou's avatar Ramon Nou
Browse files

Adapt replica plans to migration jobs

parent 609d9edf

include/common/rpc/data_migrator.hpp

+4 −0
Changes for include/common/rpc/data_migrator.hpp: 4 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -36,6 +36,10 @@ struct MigrationJob { 
    host_t source_node;

    host_t target_node;

    size_t bytes_transferred{0};

    int source_copy{-1};

    int target_copy{-1};

    uint64_t old_placement_generation{0};

    uint64_t new_placement_generation{0};

};



/// Helper to find which host owns a position in a partition table

include/common/rpc/replica_migration_adapter.hpp

0 → 100644
+52 −0
Changes for include/common/rpc/replica_migration_adapter.hpp: 52 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

*/



#ifndef GKFS_RPC_REPLICA_MIGRATION_ADAPTER_HPP

#define GKFS_RPC_REPLICA_MIGRATION_ADAPTER_HPP



#include <common/rpc/data_migrator.hpp>

#include <common/rpc/replica_migrator.hpp>



#include <stdexcept>

#include <vector>



namespace gkfs::rpc {



inline std::vector<MigrationJob>

adapt_replica_migration_plan(const logical_chunk_replica_plan& plan,

                             const uint64_t expected_old_generation,

                             const uint64_t expected_new_generation) {

    const auto validation = classify_replica_migration_plan(

            plan, expected_old_generation, expected_new_generation);

    if(validation != replica_migration_plan_error::none) {

        throw std::invalid_argument("replica migration plan is not executable");

    }



    std::vector<MigrationJob> jobs;

    jobs.reserve(plan.jobs.size());

    for(const auto& replica_job : plan.jobs) {

        MigrationJob job{replica_job.path, replica_job.chunk_id,

                         replica_job.source_node, replica_job.target_node};

        job.source_copy = replica_job.source_copy;

        job.target_copy = replica_job.target_copy;

        job.old_placement_generation = replica_job.old_placement_generation;

        job.new_placement_generation = replica_job.new_placement_generation;

        jobs.push_back(std::move(job));

    }

    return jobs;

}



inline bool

is_replica_migration_job(const MigrationJob& job) {

    return job.source_copy >= 0 || job.target_copy >= 0;

}



} // namespace gkfs::rpc



#endif // GKFS_RPC_REPLICA_MIGRATION_ADAPTER_HPP
 
 No newline at end of file

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
@@ -76,6 +76,7 @@ target_sources(unit_tests 
    ${CMAKE_CURRENT_LIST_DIR}/test_distributor.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_replica_placement.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_replica_migrator.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_replica_migration_adapter.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_repair_queue.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_helpers.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_random_slicing_distributor.cpp

tests/unit/test_distributor_factory_and_migrator.cpp

+1 −0
Changes for tests/unit/test_distributor_factory_and_migrator.cpp: 1 added line, 0 removed lines.
Original line number Diff line number Diff line
@@ -26,6 +26,7 @@ 
#include <catch2/catch_all.hpp>

#include <common/rpc/distributor_factory.hpp>

#include <common/rpc/data_migrator.hpp>

#include <common/rpc/replica_migration_adapter.hpp>

#include <common/rpc/cutshift_sorted.hpp>

#include <sstream>

#include <algorithm>

tests/unit/test_replica_migration_adapter.cpp

0 → 100644
+50 −0
Changes for tests/unit/test_replica_migration_adapter.cpp: 50 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 <common/rpc/replica_migration_adapter.hpp>



#include <catch2/catch_all.hpp>



TEST_CASE("replica migration adapter preserves copy and generation metadata",

          "[replication][mutation]") {

    gkfs::rpc::logical_chunk_replica_plan plan;

    plan.old_hosts = {1, 2};

    plan.new_hosts = {3, 4};

    plan.old_placement_generation = 7;

    plan.new_placement_generation = 8;

    plan.jobs.push_back({"/file", 4, 1, 0, 2, 3, 7, 8});



    const auto jobs = gkfs::rpc::adapt_replica_migration_plan(plan, 7, 8);

    REQUIRE(jobs.size() == 1);

    REQUIRE(gkfs::rpc::is_replica_migration_job(jobs.front()));

    REQUIRE(jobs.front().source_copy == 1);

    REQUIRE(jobs.front().target_copy == 0);

    REQUIRE(jobs.front().old_placement_generation == 7);

    REQUIRE(jobs.front().new_placement_generation == 8);

}



TEST_CASE("replica migration adapter rejects stale plans",

          "[replication][mutation]") {

    gkfs::rpc::logical_chunk_replica_plan plan;

    plan.old_hosts = {1, 2};

    plan.new_hosts = {3, 4};

    plan.old_placement_generation = 7;

    plan.new_placement_generation = 8;

    plan.jobs.push_back({"/file", 4, 0, 0, 1, 3, 7, 8});



    REQUIRE_THROWS_WITH(

            gkfs::rpc::adapt_replica_migration_plan(plan, 6, 8),

            "replica migration plan is not executable");

}



TEST_CASE("primary migration jobs remain non-replica jobs",

          "[replication][mutation]") {

    const gkfs::rpc::MigrationJob job{"/file", 0, 1, 2};

    REQUIRE_FALSE(gkfs::rpc::is_replica_migration_job(job));

}
 
 No newline at end of file