diff --git a/src/dbms/replication_handler.cpp b/src/dbms/replication_handler.cpp index 8834d4760..f97c21bd4 100644 --- a/src/dbms/replication_handler.cpp +++ b/src/dbms/replication_handler.cpp @@ -75,6 +75,7 @@ bool ReplicationHandler::SetReplicationRoleMain() { return true; }; + // COORDINATOR cannot become main auto const coordinator_handler = [](replication::RoleCoordinatorData const &) { return false; }; // TODO: under lock @@ -83,13 +84,14 @@ bool ReplicationHandler::SetReplicationRoleMain() { } bool ReplicationHandler::SetReplicationRoleCoordinator() { - auto const main_handler = [](RoleMainData const &) { return false; }; + // Upgrading REPLICA to COORDINATOR is not supported auto const replica_handler = [](RoleReplicaData const &) { return false; }; - - auto const coordinator_handler = [](replication::RoleCoordinatorData const &) { - // TODO - // set RPCs - return true; + // If we are already COORDINATOR, we don't want to change anything + auto const coordinator_handler = [](replication::RoleCoordinatorData const &) { return false; }; + // Upgrade MAIN to COORDINATOR + // TODO: (andi) Probably more complex steps will be necessary here. + auto const main_handler = [this](RoleMainData const &) { + return dbms_handler_.ReplicationState().SetReplicationRoleCoordinator(); }; // TODO: under lock @@ -99,7 +101,7 @@ bool ReplicationHandler::SetReplicationRoleCoordinator() { bool ReplicationHandler::SetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config) { // We don't want to restart the server if we're already a REPLICA - if (dbms_handler_.ReplicationState().IsReplica()) { + if (dbms_handler_.ReplicationState().IsReplica() || dbms_handler_.ReplicationState().IsCoordinator()) { return false; } @@ -131,6 +133,7 @@ bool ReplicationHandler::SetReplicationRoleReplica(const memgraph::replication:: } return true; }, + /// TODO: (andi) Assert that this code cannot execute [](replication::RoleCoordinatorData const &) { return false; }}, dbms_handler_.ReplicationState().ReplicationData()); // TODO Handle error (restore to main?) diff --git a/src/query/interpreter.cpp b/src/query/interpreter.cpp index f014a0421..5ab637af5 100644 --- a/src/query/interpreter.cpp +++ b/src/query/interpreter.cpp @@ -286,7 +286,7 @@ class ReplQueryHandler final : public query::ReplicationQueryHandler { } } else if (replication_role == ReplicationQuery::ReplicationRole::COORDINATOR) { if (!handler_.SetReplicationRoleCoordinator()) { - throw QueryRuntimeException("Couldn't set role to main!"); + throw QueryRuntimeException("Couldn't set role to coordinator!"); } } else { if (!port || *port < 0 || *port > std::numeric_limits::max()) { diff --git a/src/replication/include/replication/state.hpp b/src/replication/include/replication/state.hpp index 396c9ea6f..c6c68b2c0 100644 --- a/src/replication/include/replication/state.hpp +++ b/src/replication/include/replication/state.hpp @@ -54,9 +54,18 @@ struct RoleReplicaData { }; struct RoleCoordinatorData { + RoleCoordinatorData() = default; + ~RoleCoordinatorData() = default; + + RoleCoordinatorData(RoleCoordinatorData const &) = delete; + RoleCoordinatorData &operator=(RoleCoordinatorData const &) = delete; + RoleCoordinatorData(RoleCoordinatorData &&) = default; + RoleCoordinatorData &operator=(RoleCoordinatorData &&) = default; + std::list registered_replicas_{}; std::unique_ptr main; - // TODO does it need epoch + + // TODO: (andi) Does it need epoch or some other way or tracking what is going on? }; // Global (instance) level object @@ -94,6 +103,9 @@ struct ReplicationState { bool ShouldPersist() const { return nullptr != durability_; } bool TryPersistRoleMain(std::string new_epoch); bool TryPersistRoleReplica(const ReplicationServerConfig &config); + /// TODO: (andi) If we will need epoch or something to track, we will need to pass it here as argument + bool TryPersistRoleCoordinator(); + bool TryPersistUnregisterReplica(std::string_view name); bool TryPersistRegisteredReplica(const ReplicationClientConfig &config); @@ -103,8 +115,8 @@ struct ReplicationState { utils::BasicResult RegisterReplica(const ReplicationClientConfig &config); bool SetReplicationRoleMain(); - bool SetReplicationRoleReplica(const ReplicationServerConfig &config); + bool SetReplicationRoleCoordinator(); private: bool HandleVersionMigration(durability::ReplicationRoleEntry &data) const; diff --git a/src/replication/include/replication/status.hpp b/src/replication/include/replication/status.hpp index 943db423a..353fa513b 100644 --- a/src/replication/include/replication/status.hpp +++ b/src/replication/include/replication/status.hpp @@ -46,11 +46,17 @@ struct ReplicaRole { friend bool operator==(ReplicaRole const &, ReplicaRole const &) = default; }; +// fragment of key: "__replication_role" +// TODO: (andi) Check if you will need to add some epoch or something for managing coordinator role +struct CoordinatorRole { + friend bool operator==(CoordinatorRole const &, CoordinatorRole const &) = default; +}; + // from key: "__replication_role" struct ReplicationRoleEntry { DurabilityVersion version = DurabilityVersion::V2; // if not latest then migration required for kReplicationReplicaPrefix - std::variant role; + std::variant role; friend bool operator==(ReplicationRoleEntry const &, ReplicationRoleEntry const &) = default; }; diff --git a/src/replication/state.cpp b/src/replication/state.cpp index 418c221c1..9a119c1a0 100644 --- a/src/replication/state.cpp +++ b/src/replication/state.cpp @@ -91,6 +91,20 @@ bool ReplicationState::TryPersistRoleMain(std::string new_epoch) { return false; } +bool ReplicationState::TryPersistRoleCoordinator() { + if (!ShouldPersist()) return true; + + /// TODO: (andi) Epoch/identifier possible change + auto data = durability::ReplicationRoleEntry{.role = durability::CoordinatorRole{}}; + + if (durability_->Put(durability::kReplicationRoleName, nlohmann::json(data).dump())) { + role_persisted = RolePersisted::YES; + return true; + } + spdlog::error("Error when saving COORDINATOR replication role in settings."); + return false; +} + bool ReplicationState::TryPersistUnregisterReplica(std::string_view name) { if (!ShouldPersist()) return true; @@ -151,6 +165,9 @@ auto ReplicationState::FetchReplicationData() -> FetchReplicationResult_t { [&](durability::ReplicaRole &&r) -> FetchReplicationResult_t { return {RoleReplicaData{r.config, std::make_unique(r.config)}}; }, + /// TODO: (andi) This must change for sure because this is the step in which we should create + /// ReplicationClient for MAIN and for REPLICAs + [&](durability::CoordinatorRole &&) -> FetchReplicationResult_t { return {RoleCoordinatorData{}}; }, }, std::move(data.role)); } catch (...) { @@ -233,6 +250,16 @@ bool ReplicationState::SetReplicationRoleReplica(const ReplicationServerConfig & return true; } +bool ReplicationState::SetReplicationRoleCoordinator() { + /// TODO: (andi) Think if need epoch and how is this going to be tracked. + if (!TryPersistRoleCoordinator()) { + return false; + } + + replication_data_ = RoleCoordinatorData{}; + return true; +} + utils::BasicResult ReplicationState::RegisterReplica( const ReplicationClientConfig &config) { auto const replica_handler = [](RoleReplicaData const &) { return RegisterReplicaError::NOT_MAIN; }; diff --git a/src/replication/status.cpp b/src/replication/status.cpp index f4e821313..28b4a8719 100644 --- a/src/replication/status.cpp +++ b/src/replication/status.cpp @@ -28,6 +28,8 @@ constexpr auto *kReplicationRole = "replication_role"; constexpr auto *kEpoch = "epoch"; constexpr auto *kVersion = "durability_version"; +/// TODO: (andi) This will have to change for MAIN because now also MAIN will have its own server. Do we then have to +/// introduce new durability version? void to_json(nlohmann::json &j, const ReplicationRoleEntry &p) { auto processMAIN = [&](MainRole const &main) { j = nlohmann::json{{kVersion, p.version}, {kReplicationRole, ReplicationRole::MAIN}, {kEpoch, main.epoch.id()}}; @@ -41,7 +43,13 @@ void to_json(nlohmann::json &j, const ReplicationRoleEntry &p) { // TODO: SSL }; }; - std::visit(utils::Overloaded{processMAIN, processREPLICA}, p.role); + + /// TODO: (andi) Extend this if you need to introduce some kind of timestamp to coordinator. + auto processCOORDINATOR = [&](CoordinatorRole const &) { + j = nlohmann::json{{kVersion, p.version}, {kReplicationRole, ReplicationRole::COORDINATOR}}; + }; + + std::visit(utils::Overloaded{processMAIN, processREPLICA, processCOORDINATOR}, p.role); } void from_json(const nlohmann::json &j, ReplicationRoleEntry &p) { diff --git a/tests/e2e/replication/CMakeLists.txt b/tests/e2e/replication/CMakeLists.txt index fb8c8c5fc..be1432da5 100644 --- a/tests/e2e/replication/CMakeLists.txt +++ b/tests/e2e/replication/CMakeLists.txt @@ -14,7 +14,9 @@ copy_e2e_python_files(replication_show conftest.py) copy_e2e_python_files(replication_show show.py) copy_e2e_python_files(replication_show show_while_creating_invalid_state.py) copy_e2e_python_files(replication_show edge_delete.py) -copy_e2e_python_files(replication_show replication_general.py) +copy_e2e_python_files(replication_show set_coordinator_role.py) +copy_e2e_python_files(replication_show coordinator_test.py) +copy_e2e_python_files(replication_show replica_test.py) copy_e2e_python_files(replication_show manual_failover.py) copy_e2e_python_files_from_parent_folder(replication_show ".." memgraph.py) copy_e2e_python_files_from_parent_folder(replication_show ".." interactive_mg_runner.py) diff --git a/tests/e2e/replication/coordinator_test.py b/tests/e2e/replication/coordinator_test.py new file mode 100644 index 000000000..3df987ce9 --- /dev/null +++ b/tests/e2e/replication/coordinator_test.py @@ -0,0 +1,38 @@ +# Copyright 2023 Memgraph Ltd. +# +# Use of this software is governed by the Business Source License +# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source +# License, and you may not use this file except in compliance with the Business Source License. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0, included in the file +# licenses/APL.txt. + +import sys + +import pytest +from common import connect_default, execute_and_fetch_all + + +def test_registering_only_replicas_on_coordinator(): + # cursor = connect_default().cursor() + # execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO COORDINATOR;") + # execute_and_fetch_all(cursor, "REGISTER REPLICA replica_1 SYNC TO '127.0.0.1:10001';") + # execute_and_fetch_all(cursor, "REGISTER REPLICA replica_2 SYNC TO '127.0.0.1:10002';") + # execute_and_fetch_all(cursor, "SHOW REPLICATION CLUSTER;") + # TODO: (andi) Test that the replicas are registered + pass + + +def test_registering_only_main_on_coordinator(): + pass + # cursor = connect_default().cursor() + # execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO COORDINATOR;") + # execute_and_fetch_all(cursor, "REGISTER MAIN TO '127.0.0.1:10005';") + # execute_and_fetch_all(cursor, "SHOW REPLICATION CLUSTER;") + # TODO: (andi) Test that the main is registered + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-rA"])) diff --git a/tests/e2e/replication/main_test.py b/tests/e2e/replication/main_test.py new file mode 100644 index 000000000..9adf7babb --- /dev/null +++ b/tests/e2e/replication/main_test.py @@ -0,0 +1,28 @@ +# Copyright 2023 Memgraph Ltd. +# +# Use of this software is governed by the Business Source License +# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source +# License, and you may not use this file except in compliance with the Business Source License. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0, included in the file +# licenses/APL.txt. + +import sys + +import pytest +from common import connect_default, execute_and_fetch_all + + +def test_setting_port_on_main(): + pass + # cursor = connect_default().cursor() + # execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO MAIN WITH PORT 10011;") + # res = execute_and_fetch_all(cursor, "SHOW REPLICATION ROLE;") + # assert cursor.description[0].name == "replication role" + # assert res[0][0] == "main" + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-rA"])) diff --git a/tests/e2e/replication/replica_test.py b/tests/e2e/replication/replica_test.py new file mode 100644 index 000000000..decb69aeb --- /dev/null +++ b/tests/e2e/replication/replica_test.py @@ -0,0 +1,28 @@ +# Copyright 2023 Memgraph Ltd. +# +# Use of this software is governed by the Business Source License +# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source +# License, and you may not use this file except in compliance with the Business Source License. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0, included in the file +# licenses/APL.txt. + +import sys + +import pytest +from common import execute_and_fetch_all + + +def test_replica_cannot_register_replica(connection): + cursor = connection(7688, "role").cursor() + try: + execute_and_fetch_all(cursor, "REGISTER REPLICA replica_1 SYNC TO '127.0.0.1:10001'") + assert False + except Exception as e: + assert str(e) == "Replica can't register another replica!" + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-rA"])) diff --git a/tests/e2e/replication/replication_general.py b/tests/e2e/replication/set_coordinator_role.py similarity index 53% rename from tests/e2e/replication/replication_general.py rename to tests/e2e/replication/set_coordinator_role.py index 9f5b47164..668d43127 100644 --- a/tests/e2e/replication/replication_general.py +++ b/tests/e2e/replication/set_coordinator_role.py @@ -33,44 +33,24 @@ def test_coordinator_role_no_port(): assert res[0][0] == "coordinator" -def test_setting_port_on_main(): - cursor = connect_default().cursor() - execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO MAIN WITH PORT 10011;") - res = execute_and_fetch_all(cursor, "SHOW REPLICATION ROLE;") - assert cursor.description[0].name == "replication role" - assert res[0][0] == "main" - - -def test_registering_replicas_no_main_on_coordinator(): - cursor = connect_default().cursor() - execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO COORDINATOR;") - execute_and_fetch_all(cursor, "REGISTER REPLICA replica_1 SYNC TO '127.0.0.1:10001';") - execute_and_fetch_all(cursor, "REGISTER REPLICA replica_2 SYNC TO '127.0.0.1:10002';") +def test_registering_only_replicas_on_coordinator(): + # cursor = connect_default().cursor() + # execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO COORDINATOR;") + # execute_and_fetch_all(cursor, "REGISTER REPLICA replica_1 SYNC TO '127.0.0.1:10001';") + # execute_and_fetch_all(cursor, "REGISTER REPLICA replica_2 SYNC TO '127.0.0.1:10002';") # execute_and_fetch_all(cursor, "SHOW REPLICATION CLUSTER;") # TODO: (andi) Test that the replicas are registered + pass -def test_registering_main_no_replicas_on_coordinator(): - cursor = connect_default().cursor() - execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO COORDINATOR;") - execute_and_fetch_all(cursor, "REGISTER MAIN TO '127.0.0.1:10005';") +def test_registering_only_main_on_coordinator(): + pass + # cursor = connect_default().cursor() + # execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO COORDINATOR;") + # execute_and_fetch_all(cursor, "REGISTER MAIN TO '127.0.0.1:10005';") # execute_and_fetch_all(cursor, "SHOW REPLICATION CLUSTER;") # TODO: (andi) Test that the main is registered -# TODO: Test that SHOW REPLICATION CLUSTER can be only called on coordinator -# TODO: Test that REGISTER MAIN passes on coordinator - - -def test_replica_cannot_register_replica(): - cursor = connect_default().cursor() - execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO REPLICA WITH PORT 10011;") - try: - execute_and_fetch_all(cursor, "REGISTER REPLICA replica_1 SYNC TO '127.0.0.1:10001'") - assert False - except Exception as e: - assert str(e) == "Replica can't register another replica!" - - if __name__ == "__main__": sys.exit(pytest.main([__file__, "-rA"])) diff --git a/tests/e2e/replication/workloads.yaml b/tests/e2e/replication/workloads.yaml index 7e08b940a..97c100fd9 100644 --- a/tests/e2e/replication/workloads.yaml +++ b/tests/e2e/replication/workloads.yaml @@ -52,19 +52,38 @@ template_cluster: &template_cluster ] <<: *template_validation_queries -in_memory_cluster: &in_memory_cluster +set_coordinator_role_cluster: &set_coordinator_role_cluster cluster: main: args: ["--bolt-port", "7687", "--log-level=TRACE"] - log_file: "replication_general_e2e.log" + log_file: "set_coordinator_role_e2e.log" setup_queries: [] validation_queries: [] +coordinator_role_cluster: &coordinator_role_cluster + cluster: + main: + args: ["--bolt-port", "7687", "--log-level=TRACE"] + log_file: "coordinator_role_e2e.log" + setup_queries: ["SET REPLICATION ROLE TO COORDINATOR;"] + validation_queries: [] + + workloads: - - name: "Replication general" + - name: "Set coordinator role" binary: "tests/e2e/pytest_runner.sh" - args: ["replication/replication_general.py"] - <<: *in_memory_cluster + args: ["replication/set_coordinator_role.py"] + <<: *set_coordinator_role_cluster + + - name: "Coordinator test" + binary: "tests/e2e/pytest_runner.sh" + args: ["replication/coordinator_test.py"] + <<: *coordinator_role_cluster + + - name: "Replica test" + binary: "tests/e2e/pytest_runner.sh" + args: ["replication/replica_test.py"] + <<: *template_cluster - name: "Constraints" binary: "tests/e2e/replication/memgraph__e2e__replication__constraints" @@ -81,40 +100,14 @@ workloads: args: [] <<: *template_cluster - - name: "Manual failover cluster setup" + - name: "Show while creating invalid state" binary: "tests/e2e/pytest_runner.sh" - args: ["replication/manual_failover.py"] - cluster: - replica_1: - args: ["--bolt-port", "7688", "--log-level=trace"] - log_file: "replication-e2e-replica1.log" - setup_queries: ["SET REPLICATION ROLE TO REPLICA WITH PORT 10001;"] - validation_queries: [] - replica_2: - args: ["--bolt-port", "7689", "--log-level=trace"] - log_file: "replication-e2e-replica2.log" - setup_queries: ["SET REPLICATION ROLE TO REPLICA WITH PORT 10002;"] - validation_queries: [] - main: - args: ["--bolt-port", "7687", "--log-level=trace"] - log_file: "replication-e2e-main.log" - setup_queries: [ - "SET REPLICATION ROLE TO MAIN WITH PORT 10003;", - "REGISTER REPLICA REPLICA_1 SYNC TO '127.0.0.1:10001'", - "REGISTER REPLICA REPLICA_2 SYNC TO '127.0.0.1:10002'", - ] - validation_queries: [] - coordinator: - args: ["--bolt-port", "7690", "--log-level=trace"] - log_file: "replication-e2e-coordinator.log" - setup_queries: [ - "SET REPLICATION ROLE TO COORDINATOR;", - "REGISTER MAIN TO '127.0.0.1:10003'", - "REGISTER REPLICA REPLICA_1 SYNC TO '127.0.0.1:10001'", - "REGISTER REPLICA REPLICA_2 SYNC TO '127.0.0.1:10002'", - ] - validation_queries: [] + args: ["replication/show_while_creating_invalid_state.py"] + - name: "Delete edge replication" + binary: "tests/e2e/pytest_runner.sh" + args: ["replication/edge_delete.py"] + <<: *template_simple_cluster - name: "show" binary: "tests/e2e/pytest_runner.sh" @@ -144,12 +137,3 @@ workloads: "register replica replica_3 async to '127.0.0.1:10003'" ] validation_queries: [] - - - name: "Show while creating invalid state" - binary: "tests/e2e/pytest_runner.sh" - args: ["replication/show_while_creating_invalid_state.py"] - - - name: "Delete edge replication" - binary: "tests/e2e/pytest_runner.sh" - args: ["replication/edge_delete.py"] - <<: *template_simple_cluster