Compare commits

..

19 Commits

Author SHA1 Message Date
Andreja Tonev
ed29168723 Bugfix: Using reference in a callback 2024-02-26 19:43:07 +01:00
andrejtonev
c2e9df309a Correctly call driver v1 tests (#1630) 2024-02-26 17:28:13 +00:00
andrejtonev
82c47ee80d GetInfo simplification (#1621)
* Removed force dir in the GetInfo functions
2024-02-26 14:55:45 +00:00
andrejtonev
6a4ef55e90 Better auth user/role handling (#1699)
* Stop auth module from creating users
* Explicit about auth policy (check if no users defined OR auth module used)
* Role supports database access definition
* Authenticate() returns user or role
* AuthChecker generates QueryUserOrRole (can be empty)
* QueryUserOrRole actually authorizes
* Add auth cache invalidation
* Better database access queries (GRANT, DENY, REVOKE DATABASE)
2024-02-22 14:00:39 +00:00
Marko Budiselić
98727e0fa0 Update operating systems (#1371) 2024-02-22 11:14:48 +01:00
Aidar Samerkhanov
9a20ac494d In BFS expansion filter by path we should shrink path to restore state prior to expansion only if the path was changed. (#1745) 2024-02-22 05:34:08 +00:00
Marko Barišić
e302be98a2 Push successful RC builds to S3 (#1741)
* Add new workflow which calls release build workflows

* Make the workflow build packages only on RC tags

* Change artifact names to include OS name
2024-02-21 17:08:14 +01:00
Marko Budiselić
61b9bb0f59 Add toolchain-v5 compatibility Revert to C++20 (#587)
* Upgrade cppitertools, spdlog, fmt, rapidcheck
* Make compilation work on both v4 and v5 toolchains
2024-02-19 21:09:54 +01:00
Andi
7ec648b4ce Add --experimental-enabled=high-availability (#1720) 2024-02-19 16:28:15 +00:00
Marko Budiselić
f098a9d5e3 Patch NuRaft for clang-17 compilation (#1733) 2024-02-19 14:50:37 +01:00
Josipmrden
bae3e8a6d3 Add function for property sizes (#1557)
Add function for property sizes
2024-02-19 13:56:01 +01:00
Andi
f3574012c5 Add cpp23 support (#1726) 2024-02-19 10:36:51 +00:00
Gareth Andrew Lloyd
33c400fcc1 Fixup memory e2e tests (#1715)
- Remove the e2e that did concurrent mgp_* calls on the same transaction
  (ATM this is unsupported)
- Fix up the concurrent mgp_global_alloc test to be testing it more precisely
- Reduce the memory limit on detach delete test due to recent memory
  optimizations around deltas.
- No longer throw from hook, through jemalloc C, to our C++ on other
  side. This cause mutex unlocks to not happen.
- No longer allocate error messages while inside the hook. This caused
  recursive entry back inside jamalloc which would try to relock a
  non-recursive mutex.
2024-02-16 15:35:08 +00:00
Marko Budiselić
5ac938a6c9 Remove default assignees from issue-bug template (#1730) 2024-02-16 14:41:53 +01:00
Andi
3e3224f0a2 Forbid having multiple mains in the cluster (#1727) 2024-02-16 11:41:15 +00:00
Antonio Filipovic
bfc756c092 HA: Polish flow for replicas from coordinator (#1711) 2024-02-16 10:58:01 +01:00
Marko Barišić
5f2e3f01d0 Turn e2e tests back on for release build workflows (#1725) 2024-02-15 16:20:04 +01:00
Marko Barišić
2c774ff09b Add rules for rc workflows (#1722) 2024-02-15 15:33:14 +01:00
Andi
20b47845f0 Forbid writing to cluster-managed main on restart (#1717) 2024-02-15 14:07:04 +01:00
30 changed files with 661 additions and 1801 deletions

View File

@@ -3,7 +3,6 @@ name: Bug report
about: Create a report to help us improve about: Create a report to help us improve
title: "" title: ""
labels: bug labels: bug
assignees: gitbuda
--- ---
**Memgraph version** **Memgraph version**

View File

@@ -56,15 +56,6 @@ jobs:
- name: "Build package" - name: "Build package"
run: | run: |
./release/package/run.sh package debian-10 $BUILD_TYPE ./release/package/run.sh package debian-10 $BUILD_TYPE
- name: Upload to S3
uses: jakejarvis/s3-sync-action@v0.5.1
env:
AWS_S3_BUCKET: "deps.memgraph.io"
AWS_ACCESS_KEY_ID: ${{ secrets.S3_AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.S3_AWS_SECRET_ACCESS_KEY }}
AWS_REGION: "eu-west-1"
SOURCE_DIR: "build/output"
DEST_DIR: "memgraph-unofficial/${{ github.ref_name }}/"
- name: "Upload package" - name: "Upload package"
uses: actions/upload-artifact@v4 uses: actions/upload-artifact@v4
with: with:
@@ -84,15 +75,6 @@ jobs:
- name: "Build package" - name: "Build package"
run: | run: |
./release/package/run.sh package ubuntu-22.04 $BUILD_TYPE ./release/package/run.sh package ubuntu-22.04 $BUILD_TYPE
- name: Upload to S3
uses: jakejarvis/s3-sync-action@v0.5.1
env:
AWS_S3_BUCKET: "deps.memgraph.io"
AWS_ACCESS_KEY_ID: ${{ secrets.S3_AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.S3_AWS_SECRET_ACCESS_KEY }}
AWS_REGION: "eu-west-1"
SOURCE_DIR: "build/output"
DEST_DIR: "memgraph-unofficial/${{ github.ref_name }}/"
- name: "Upload package" - name: "Upload package"
uses: actions/upload-artifact@v4 uses: actions/upload-artifact@v4
with: with:
@@ -104,7 +86,7 @@ jobs:
needs: [Ubuntu20_04] needs: [Ubuntu20_04]
runs-on: [self-hosted, DockerMgBuild, ARM64] runs-on: [self-hosted, DockerMgBuild, ARM64]
# M1 Mac mini is sometimes slower # M1 Mac mini is sometimes slower
timeout-minutes: 150 timeout-minutes: 90
steps: steps:
- name: "Set up repository" - name: "Set up repository"
uses: actions/checkout@v4 uses: actions/checkout@v4
@@ -119,26 +101,6 @@ jobs:
name: ubuntu-22.04-aarch64 name: ubuntu-22.04-aarch64
path: build/output/ubuntu-22.04-arm/memgraph*.deb path: build/output/ubuntu-22.04-arm/memgraph*.deb
PushToS3Ubuntu20_04_ARM:
if: github.ref_type == 'tag'
needs: [PackageUbuntu20_04_ARM]
runs-on: ubuntu-latest
steps:
- name: Download package
uses: actions/download-artifact@v4
with:
name: ubuntu-22.04-aarch64
path: build/output/release
- name: Upload to S3
uses: jakejarvis/s3-sync-action@v0.5.1
env:
AWS_S3_BUCKET: "deps.memgraph.io"
AWS_ACCESS_KEY_ID: ${{ secrets.S3_AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.S3_AWS_SECRET_ACCESS_KEY }}
AWS_REGION: "eu-west-1"
SOURCE_DIR: "build/output/release"
DEST_DIR: "memgraph-unofficial/${{ github.ref_name }}/"
PackageDebian11: PackageDebian11:
if: github.ref_type == 'tag' if: github.ref_type == 'tag'
needs: [Debian10, Ubuntu20_04] needs: [Debian10, Ubuntu20_04]
@@ -152,15 +114,6 @@ jobs:
- name: "Build package" - name: "Build package"
run: | run: |
./release/package/run.sh package debian-11 $BUILD_TYPE ./release/package/run.sh package debian-11 $BUILD_TYPE
- name: Upload to S3
uses: jakejarvis/s3-sync-action@v0.5.1
env:
AWS_S3_BUCKET: "deps.memgraph.io"
AWS_ACCESS_KEY_ID: ${{ secrets.S3_AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.S3_AWS_SECRET_ACCESS_KEY }}
AWS_REGION: "eu-west-1"
SOURCE_DIR: "build/output"
DEST_DIR: "memgraph-unofficial/${{ github.ref_name }}/"
- name: "Upload package" - name: "Upload package"
uses: actions/upload-artifact@v4 uses: actions/upload-artifact@v4
with: with:
@@ -172,7 +125,7 @@ jobs:
needs: [Debian10, Ubuntu20_04] needs: [Debian10, Ubuntu20_04]
runs-on: [self-hosted, DockerMgBuild, ARM64] runs-on: [self-hosted, DockerMgBuild, ARM64]
# M1 Mac mini is sometimes slower # M1 Mac mini is sometimes slower
timeout-minutes: 150 timeout-minutes: 90
steps: steps:
- name: "Set up repository" - name: "Set up repository"
uses: actions/checkout@v4 uses: actions/checkout@v4
@@ -187,15 +140,16 @@ jobs:
name: debian-11-aarch64 name: debian-11-aarch64
path: build/output/debian-11-arm/memgraph*.deb path: build/output/debian-11-arm/memgraph*.deb
PushToS3Debian11_ARM: PushToS3:
if: github.ref_type == 'tag' if: github.ref_type == 'tag'
needs: [PackageDebian11_ARM] needs: [PackageDebian10, PackageDebian11, PackageDebian11_ARM, PackageUbuntu20_04, PackageUbuntu20_04_ARM]
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- name: Download package - name: Download artifacts
uses: actions/download-artifact@v4 uses: actions/download-artifact@v4
with: with:
name: debian-11-aarch64 # name: # if name input parameter is not provided, all artifacts are downloaded
# and put in directories named after each one.
path: build/output/release path: build/output/release
- name: Upload to S3 - name: Upload to S3
uses: jakejarvis/s3-sync-action@v0.5.1 uses: jakejarvis/s3-sync-action@v0.5.1

View File

@@ -64,7 +64,7 @@ option(MG_ENTERPRISE "Build Memgraph Enterprise Edition" ON)
# Set the current version here to override the automatic version detection. The # Set the current version here to override the automatic version detection. The
# version must be specified as `X.Y.Z`. Primarily used when building new patch # version must be specified as `X.Y.Z`. Primarily used when building new patch
# versions. # versions.
set(MEMGRAPH_OVERRIDE_VERSION "2.15.0") set(MEMGRAPH_OVERRIDE_VERSION "")
# Custom suffix that this version should have. The suffix can be any arbitrary # Custom suffix that this version should have. The suffix can be any arbitrary
# string. Primarily used when building a version for a specific customer. # string. Primarily used when building a version for a specific customer.

View File

@@ -45,7 +45,6 @@ MEMGRAPH_BUILD_DEPS=(
readline-devel # for memgraph console readline-devel # for memgraph console
python3-devel # for query modules python3-devel # for query modules
openssl-devel openssl-devel
openssl
libseccomp-devel libseccomp-devel
python3 python3-pip nmap-ncat # for tests python3 python3-pip nmap-ncat # for tests
# #

View File

@@ -43,7 +43,6 @@ MEMGRAPH_BUILD_DEPS=(
readline-devel # for memgraph console readline-devel # for memgraph console
python3-devel # for query modules python3-devel # for query modules
openssl-devel openssl-devel
openssl
libseccomp-devel libseccomp-devel
python3 python-virtualenv python3-pip nmap-ncat # for qa, macro_benchmark and stress tests python3 python-virtualenv python3-pip nmap-ncat # for qa, macro_benchmark and stress tests
# #

View File

@@ -36,7 +36,7 @@ ADDITIONAL USE GRANT: You may use the Licensed Work in accordance with the
3. using the Licensed Work to create a work or solution 3. using the Licensed Work to create a work or solution
which competes (or might reasonably be expected to which competes (or might reasonably be expected to
compete) with the Licensed Work. compete) with the Licensed Work.
CHANGE DATE: 2028-28-02 CHANGE DATE: 2028-21-01
CHANGE LICENSE: Apache License, Version 2.0 CHANGE LICENSE: Apache License, Version 2.0
For information about alternative licensing arrangements, please visit: https://memgraph.com/legal. For information about alternative licensing arrangements, please visit: https://memgraph.com/legal.

View File

@@ -3035,6 +3035,8 @@ class ReplicationQuery : public memgraph::query::Query {
enum class SyncMode { SYNC, ASYNC }; enum class SyncMode { SYNC, ASYNC };
enum class ReplicaState { READY, REPLICATING, RECOVERY, MAYBE_BEHIND, DIVERGED_FROM_MAIN };
ReplicationQuery() = default; ReplicationQuery() = default;
DEFVISITABLE(QueryVisitor<void>); DEFVISITABLE(QueryVisitor<void>);

View File

@@ -297,6 +297,16 @@ inline auto convertToReplicationMode(const ReplicationQuery::SyncMode &sync_mode
class ReplQueryHandler { class ReplQueryHandler {
public: public:
struct ReplicaInfo {
std::string name;
std::string socket_address;
ReplicationQuery::SyncMode sync_mode;
std::optional<double> timeout;
uint64_t current_timestamp_of_replica;
uint64_t current_number_of_timestamp_behind_master;
ReplicationQuery::ReplicaState state;
};
explicit ReplQueryHandler(query::ReplicationQueryHandler &replication_query_handler) explicit ReplQueryHandler(query::ReplicationQueryHandler &replication_query_handler)
: handler_{&replication_query_handler} {} : handler_{&replication_query_handler} {}
@@ -387,16 +397,58 @@ class ReplQueryHandler {
} }
} }
std::vector<ReplicasInfo> ShowReplicas() const { std::vector<ReplicaInfo> ShowReplicas(const dbms::Database &db) const {
auto info = handler_->ShowReplicas(); if (handler_->IsReplica()) {
if (info.HasError()) { // replica can't show registered replicas (it shouldn't have any)
switch (info.GetError()) { throw QueryRuntimeException("Replica can't show registered replicas (it shouldn't have any)!");
case ShowReplicaError::NOT_MAIN:
throw QueryRuntimeException("Replica can't show registered replicas (it shouldn't have any)!");
}
} }
return info.GetValue().entries_; // TODO: Combine results? Have a single place with clients???
// Also authentication checks (replica + database visibility)
const auto repl_infos = db.storage()->ReplicasInfo();
std::vector<ReplicaInfo> replicas;
replicas.reserve(repl_infos.size());
const auto from_info = [](const auto &repl_info) -> ReplicaInfo {
ReplicaInfo replica;
replica.name = repl_info.name;
replica.socket_address = repl_info.endpoint.SocketAddress();
switch (repl_info.mode) {
case replication_coordination_glue::ReplicationMode::SYNC:
replica.sync_mode = ReplicationQuery::SyncMode::SYNC;
break;
case replication_coordination_glue::ReplicationMode::ASYNC:
replica.sync_mode = ReplicationQuery::SyncMode::ASYNC;
break;
}
replica.current_timestamp_of_replica = repl_info.timestamp_info.current_timestamp_of_replica;
replica.current_number_of_timestamp_behind_master =
repl_info.timestamp_info.current_number_of_timestamp_behind_master;
switch (repl_info.state) {
case storage::replication::ReplicaState::READY:
replica.state = ReplicationQuery::ReplicaState::READY;
break;
case storage::replication::ReplicaState::REPLICATING:
replica.state = ReplicationQuery::ReplicaState::REPLICATING;
break;
case storage::replication::ReplicaState::RECOVERY:
replica.state = ReplicationQuery::ReplicaState::RECOVERY;
break;
case storage::replication::ReplicaState::MAYBE_BEHIND:
replica.state = ReplicationQuery::ReplicaState::MAYBE_BEHIND;
break;
case storage::replication::ReplicaState::DIVERGED_FROM_MAIN:
replica.state = ReplicationQuery::ReplicaState::DIVERGED_FROM_MAIN;
break;
}
return replica;
};
std::transform(repl_infos.begin(), repl_infos.end(), std::back_inserter(replicas), from_info);
return replicas;
} }
private: private:
@@ -1040,98 +1092,50 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
} }
#endif #endif
bool full_info = false; callback.header = {
#ifdef MG_ENTERPRISE "name", "socket_address", "sync_mode", "current_timestamp_of_replica", "number_of_timestamp_behind_master",
full_info = license::global_license_checker.IsEnterpriseValidFast(); "state"};
#endif
callback.header = {"name", "socket_address", "sync_mode", "system_info", "data_info"};
callback.fn = [handler = ReplQueryHandler{replication_query_handler}, replica_nfields = callback.header.size(), callback.fn = [handler = ReplQueryHandler{replication_query_handler}, replica_nfields = callback.header.size(),
full_info] { db_acc = current_db.db_acc_] {
auto const sync_mode_to_tv = [](memgraph::replication_coordination_glue::ReplicationMode sync_mode) { const auto &replicas = handler.ShowReplicas(*db_acc->get());
using namespace std::string_view_literals;
switch (sync_mode) {
using enum memgraph::replication_coordination_glue::ReplicationMode;
case SYNC:
return TypedValue{"sync"sv};
case ASYNC:
return TypedValue{"async"sv};
}
};
auto const replica_sys_state_to_tv = [](memgraph::replication::ReplicationClient::State state) {
using namespace std::string_view_literals;
switch (state) {
using enum memgraph::replication::ReplicationClient::State;
case BEHIND:
return TypedValue{"invalid"sv};
case READY:
return TypedValue{"ready"sv};
case RECOVERY:
return TypedValue{"recovery"sv};
}
};
auto const sys_info_to_tv = [&](ReplicaSystemInfoState orig) {
auto info = std::map<std::string, TypedValue>{};
info.emplace("ts", TypedValue{static_cast<int64_t>(orig.ts_)});
// TODO: behind not implemented
info.emplace("behind", TypedValue{/* static_cast<int64_t>(orig.behind_) */});
info.emplace("status", replica_sys_state_to_tv(orig.state_));
return TypedValue{std::move(info)};
};
auto const replica_state_to_tv = [](memgraph::storage::replication::ReplicaState state) {
using namespace std::string_view_literals;
switch (state) {
using enum memgraph::storage::replication::ReplicaState;
case READY:
return TypedValue{"ready"sv};
case REPLICATING:
return TypedValue{"replicating"sv};
case RECOVERY:
return TypedValue{"recovery"sv};
case MAYBE_BEHIND:
return TypedValue{"invalid"sv};
case DIVERGED_FROM_MAIN:
return TypedValue{"diverged"sv};
}
};
auto const info_to_tv = [&](ReplicaInfoState orig) {
auto info = std::map<std::string, TypedValue>{};
info.emplace("ts", TypedValue{static_cast<int64_t>(orig.ts_)});
info.emplace("behind", TypedValue{static_cast<int64_t>(orig.behind_)});
info.emplace("status", replica_state_to_tv(orig.state_));
return TypedValue{std::move(info)};
};
auto const data_info_to_tv = [&](std::map<std::string, ReplicaInfoState> orig) {
auto data_info = std::map<std::string, TypedValue>{};
for (auto &[name, info] : orig) {
data_info.emplace(name, info_to_tv(info));
}
return TypedValue{std::move(data_info)};
};
auto replicas = handler.ShowReplicas();
auto typed_replicas = std::vector<std::vector<TypedValue>>{}; auto typed_replicas = std::vector<std::vector<TypedValue>>{};
typed_replicas.reserve(replicas.size()); typed_replicas.reserve(replicas.size());
for (auto &replica : replicas) { for (const auto &replica : replicas) {
std::vector<TypedValue> typed_replica; std::vector<TypedValue> typed_replica;
typed_replica.reserve(replica_nfields); typed_replica.reserve(replica_nfields);
typed_replica.emplace_back(replica.name_); typed_replica.emplace_back(replica.name);
typed_replica.emplace_back(replica.socket_address_); typed_replica.emplace_back(replica.socket_address);
typed_replica.emplace_back(sync_mode_to_tv(replica.sync_mode_));
if (full_info) { switch (replica.sync_mode) {
typed_replica.emplace_back(sys_info_to_tv(replica.system_info_)); case ReplicationQuery::SyncMode::SYNC:
} else { typed_replica.emplace_back("sync");
// Set to NULL break;
typed_replica.emplace_back(TypedValue{}); case ReplicationQuery::SyncMode::ASYNC:
typed_replica.emplace_back("async");
break;
}
typed_replica.emplace_back(static_cast<int64_t>(replica.current_timestamp_of_replica));
typed_replica.emplace_back(static_cast<int64_t>(replica.current_number_of_timestamp_behind_master));
switch (replica.state) {
case ReplicationQuery::ReplicaState::READY:
typed_replica.emplace_back("ready");
break;
case ReplicationQuery::ReplicaState::REPLICATING:
typed_replica.emplace_back("replicating");
break;
case ReplicationQuery::ReplicaState::RECOVERY:
typed_replica.emplace_back("recovery");
break;
case ReplicationQuery::ReplicaState::MAYBE_BEHIND:
typed_replica.emplace_back("invalid");
break;
case ReplicationQuery::ReplicaState::DIVERGED_FROM_MAIN:
typed_replica.emplace_back("diverged");
break;
} }
typed_replica.emplace_back(data_info_to_tv(replica.data_info_));
typed_replicas.emplace_back(std::move(typed_replica)); typed_replicas.emplace_back(std::move(typed_replica));
} }

View File

@@ -84,6 +84,16 @@ class CoordinatorQueryHandler {
CoordinatorQueryHandler(CoordinatorQueryHandler &&) = default; CoordinatorQueryHandler(CoordinatorQueryHandler &&) = default;
CoordinatorQueryHandler &operator=(CoordinatorQueryHandler &&) = default; CoordinatorQueryHandler &operator=(CoordinatorQueryHandler &&) = default;
struct Replica {
std::string name;
std::string socket_address;
ReplicationQuery::SyncMode sync_mode;
std::optional<double> timeout;
uint64_t current_timestamp_of_replica;
uint64_t current_number_of_timestamp_behind_master;
ReplicationQuery::ReplicaState state;
};
struct MainReplicaStatus { struct MainReplicaStatus {
std::string_view name; std::string_view name;
std::string_view socket_address; std::string_view socket_address;

View File

@@ -11,8 +11,6 @@
#pragma once #pragma once
#include "replication/replication_client.hpp"
#include "replication_coordination_glue/mode.hpp"
#include "replication_coordination_glue/role.hpp" #include "replication_coordination_glue/role.hpp"
#include "utils/result.hpp" #include "utils/result.hpp"
#include "utils/uuid.hpp" #include "utils/uuid.hpp"
@@ -33,7 +31,6 @@ enum class RegisterReplicaError : uint8_t {
COULD_NOT_BE_PERSISTED, COULD_NOT_BE_PERSISTED,
ERROR_ACCEPTING_MAIN ERROR_ACCEPTING_MAIN
}; };
enum class UnregisterReplicaResult : uint8_t { enum class UnregisterReplicaResult : uint8_t {
NOT_MAIN, NOT_MAIN,
COULD_NOT_BE_PERSISTED, COULD_NOT_BE_PERSISTED,
@@ -41,47 +38,6 @@ enum class UnregisterReplicaResult : uint8_t {
SUCCESS, SUCCESS,
}; };
enum class ShowReplicaError : uint8_t {
NOT_MAIN,
};
struct ReplicaSystemInfoState {
uint64_t ts_;
uint64_t behind_;
replication::ReplicationClient::State state_;
};
struct ReplicaInfoState {
ReplicaInfoState(uint64_t ts, uint64_t behind, storage::replication::ReplicaState state)
: ts_(ts), behind_(behind), state_(state) {}
uint64_t ts_;
uint64_t behind_;
storage::replication::ReplicaState state_;
};
struct ReplicasInfo {
ReplicasInfo(std::string name, std::string socket_address, replication_coordination_glue::ReplicationMode sync_mode,
ReplicaSystemInfoState system_info, std::map<std::string, ReplicaInfoState> data_info)
: name_(std::move(name)),
socket_address_(std::move(socket_address)),
sync_mode_(sync_mode),
system_info_(std::move(system_info)),
data_info_(std::move(data_info)) {}
std::string name_;
std::string socket_address_;
memgraph::replication_coordination_glue::ReplicationMode sync_mode_;
ReplicaSystemInfoState system_info_;
std::map<std::string, ReplicaInfoState> data_info_;
};
struct ReplicasInfos {
explicit ReplicasInfos(std::vector<ReplicasInfo> entries) : entries_(std::move(entries)) {}
std::vector<ReplicasInfo> entries_;
};
/// A handler type that keep in sync current ReplicationState and the MAIN/REPLICA-ness of Storage /// A handler type that keep in sync current ReplicationState and the MAIN/REPLICA-ness of Storage
struct ReplicationQueryHandler { struct ReplicationQueryHandler {
virtual ~ReplicationQueryHandler() = default; virtual ~ReplicationQueryHandler() = default;
@@ -110,8 +66,6 @@ struct ReplicationQueryHandler {
virtual auto GetRole() const -> memgraph::replication_coordination_glue::ReplicationRole = 0; virtual auto GetRole() const -> memgraph::replication_coordination_glue::ReplicationRole = 0;
virtual bool IsMain() const = 0; virtual bool IsMain() const = 0;
virtual bool IsReplica() const = 0; virtual bool IsReplica() const = 0;
virtual auto ShowReplicas() const -> utils::BasicResult<ShowReplicaError, ReplicasInfos> = 0;
}; };
} // namespace memgraph::query } // namespace memgraph::query

View File

@@ -14,9 +14,7 @@
#include "replication/config.hpp" #include "replication/config.hpp"
#include "replication_coordination_glue/messages.hpp" #include "replication_coordination_glue/messages.hpp"
#include "rpc/client.hpp" #include "rpc/client.hpp"
#include "utils/rw_lock.hpp"
#include "utils/scheduler.hpp" #include "utils/scheduler.hpp"
#include "utils/spin_lock.hpp"
#include "utils/synchronized.hpp" #include "utils/synchronized.hpp"
#include "utils/thread_pool.hpp" #include "utils/thread_pool.hpp"
@@ -116,9 +114,8 @@ struct ReplicationClient {
enum class State { enum class State {
BEHIND, BEHIND,
READY, READY,
RECOVERY,
}; };
utils::Synchronized<State, utils::WritePrioritizedRWLock> state_{State::BEHIND}; utils::Synchronized<State> state_{State::BEHIND};
replication_coordination_glue::ReplicationMode mode_{replication_coordination_glue::ReplicationMode::SYNC}; replication_coordination_glue::ReplicationMode mode_{replication_coordination_glue::ReplicationMode::SYNC};
// This thread pool is used for background tasks so we don't // This thread pool is used for background tasks so we don't

View File

@@ -39,12 +39,10 @@ void SystemRestore(replication::ReplicationClient &client, system::System &syste
const utils::UUID &main_uuid, auth::SynchedAuth &auth) { const utils::UUID &main_uuid, auth::SynchedAuth &auth) {
// Check if system is up to date // Check if system is up to date
if (client.state_.WithLock( if (client.state_.WithLock(
[](auto &state) { return state != memgraph::replication::ReplicationClient::State::BEHIND; })) [](auto &state) { return state == memgraph::replication::ReplicationClient::State::READY; }))
return; return;
// Try to recover... // Try to recover...
client.state_.WithLock(
[](auto &state) { return state != memgraph::replication::ReplicationClient::State::RECOVERY; });
{ {
using enum memgraph::flags::Experiments; using enum memgraph::flags::Experiments;
bool full_system_replication = bool full_system_replication =
@@ -141,9 +139,6 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
bool IsMain() const override; bool IsMain() const override;
bool IsReplica() const override; bool IsReplica() const override;
auto ShowReplicas() const
-> utils::BasicResult<memgraph::query::ShowReplicaError, memgraph::query::ReplicasInfos> override;
auto GetReplState() const -> const memgraph::replication::ReplicationState &; auto GetReplState() const -> const memgraph::replication::ReplicationState &;
auto GetReplState() -> memgraph::replication::ReplicationState &; auto GetReplState() -> memgraph::replication::ReplicationState &;

View File

@@ -10,28 +10,26 @@
// licenses/APL.txt. // licenses/APL.txt.
#include "replication_handler/replication_handler.hpp" #include "replication_handler/replication_handler.hpp"
#include "dbms/constants.hpp"
#include "dbms/dbms_handler.hpp" #include "dbms/dbms_handler.hpp"
#include "replication/replication_client.hpp"
#include "replication_handler/system_replication.hpp" #include "replication_handler/system_replication.hpp"
namespace memgraph::replication { namespace memgraph::replication {
namespace { namespace {
#ifdef MG_ENTERPRISE #ifdef MG_ENTERPRISE
void RecoverReplication(replication::ReplicationState &repl_state, system::System &system, void RecoverReplication(memgraph::replication::ReplicationState &repl_state, memgraph::system::System &system,
dbms::DbmsHandler &dbms_handler, auth::SynchedAuth &auth) { memgraph::dbms::DbmsHandler &dbms_handler, memgraph::auth::SynchedAuth &auth) {
/* /*
* REPLICATION RECOVERY AND STARTUP * REPLICATION RECOVERY AND STARTUP
*/ */
// Startup replication state (if recovered at startup) // Startup replication state (if recovered at startup)
auto replica = [&dbms_handler, &auth, &system](replication::RoleReplicaData &data) { auto replica = [&dbms_handler, &auth, &system](memgraph::replication::RoleReplicaData &data) {
return replication::StartRpcServer(dbms_handler, data, auth, system); return memgraph::replication::StartRpcServer(dbms_handler, data, auth, system);
}; };
// Replication recovery and frequent check start // Replication recovery and frequent check start
auto main = [&system, &dbms_handler, &auth](replication::RoleMainData &mainData) { auto main = [&system, &dbms_handler, &auth](memgraph::replication::RoleMainData &mainData) {
for (auto &client : mainData.registered_replicas_) { for (auto &client : mainData.registered_replicas_) {
if (client.try_set_uuid && if (client.try_set_uuid &&
replication_coordination_glue::SendSwapMainUUIDRpc(client.rpc_client_, mainData.uuid_)) { replication_coordination_glue::SendSwapMainUUIDRpc(client.rpc_client_, mainData.uuid_)) {
@@ -40,7 +38,7 @@ void RecoverReplication(replication::ReplicationState &repl_state, system::Syste
SystemRestore(client, system, dbms_handler, mainData.uuid_, auth); SystemRestore(client, system, dbms_handler, mainData.uuid_, auth);
} }
// DBMS here // DBMS here
dbms_handler.ForEach([&mainData](dbms::DatabaseAccess db_acc) { dbms_handler.ForEach([&mainData](memgraph::dbms::DatabaseAccess db_acc) {
dbms::DbmsHandler::RecoverStorageReplication(std::move(db_acc), mainData); dbms::DbmsHandler::RecoverStorageReplication(std::move(db_acc), mainData);
}); });
@@ -50,7 +48,7 @@ void RecoverReplication(replication::ReplicationState &repl_state, system::Syste
// Warning // Warning
if (dbms_handler.default_config().durability.snapshot_wal_mode == if (dbms_handler.default_config().durability.snapshot_wal_mode ==
storage::Config::Durability::SnapshotWalMode::DISABLED) { memgraph::storage::Config::Durability::SnapshotWalMode::DISABLED) {
spdlog::warn( spdlog::warn(
"The instance has the MAIN replication role, but durability logs and snapshots are disabled. Please " "The instance has the MAIN replication role, but durability logs and snapshots are disabled. Please "
"consider " "consider "
@@ -61,18 +59,19 @@ void RecoverReplication(replication::ReplicationState &repl_state, system::Syste
return true; return true;
}; };
auto result = std::visit(utils::Overloaded{replica, main}, repl_state.ReplicationData()); auto result = std::visit(memgraph::utils::Overloaded{replica, main}, repl_state.ReplicationData());
MG_ASSERT(result, "Replica recovery failure!"); MG_ASSERT(result, "Replica recovery failure!");
} }
#else #else
void RecoverReplication(replication::ReplicationState &repl_state, dbms::DbmsHandler &dbms_handler) { void RecoverReplication(memgraph::replication::ReplicationState &repl_state,
memgraph::dbms::DbmsHandler &dbms_handler) {
// Startup replication state (if recovered at startup) // Startup replication state (if recovered at startup)
auto replica = [&dbms_handler](replication::RoleReplicaData &data) { auto replica = [&dbms_handler](memgraph::replication::RoleReplicaData &data) {
return replication::StartRpcServer(dbms_handler, data); return memgraph::replication::StartRpcServer(dbms_handler, data);
}; };
// Replication recovery and frequent check start // Replication recovery and frequent check start
auto main = [&dbms_handler](replication::RoleMainData &mainData) { auto main = [&dbms_handler](memgraph::replication::RoleMainData &mainData) {
dbms::DbmsHandler::RecoverStorageReplication(dbms_handler.Get(), mainData); dbms::DbmsHandler::RecoverStorageReplication(dbms_handler.Get(), mainData);
for (auto &client : mainData.registered_replicas_) { for (auto &client : mainData.registered_replicas_) {
@@ -80,12 +79,12 @@ void RecoverReplication(replication::ReplicationState &repl_state, dbms::DbmsHan
replication_coordination_glue::SendSwapMainUUIDRpc(client.rpc_client_, mainData.uuid_)) { replication_coordination_glue::SendSwapMainUUIDRpc(client.rpc_client_, mainData.uuid_)) {
client.try_set_uuid = false; client.try_set_uuid = false;
} }
replication::StartReplicaClient(client, dbms_handler, mainData.uuid_); memgraph::replication::StartReplicaClient(client, dbms_handler, mainData.uuid_);
} }
// Warning // Warning
if (dbms_handler.default_config().durability.snapshot_wal_mode == if (dbms_handler.default_config().durability.snapshot_wal_mode ==
storage::Config::Durability::SnapshotWalMode::DISABLED) { memgraph::storage::Config::Durability::SnapshotWalMode::DISABLED) {
spdlog::warn( spdlog::warn(
"The instance has the MAIN replication role, but durability logs and snapshots are disabled. Please " "The instance has the MAIN replication role, but durability logs and snapshots are disabled. Please "
"consider " "consider "
@@ -96,7 +95,7 @@ void RecoverReplication(replication::ReplicationState &repl_state, dbms::DbmsHan
return true; return true;
}; };
auto result = std::visit(utils::Overloaded{replica, main}, repl_state.ReplicationData()); auto result = std::visit(memgraph::utils::Overloaded{replica, main}, repl_state.ReplicationData());
MG_ASSERT(result, "Replica recovery failure!"); MG_ASSERT(result, "Replica recovery failure!");
} }
#endif #endif
@@ -134,19 +133,20 @@ void StartReplicaClient(replication::ReplicationClient &client, dbms::DbmsHandle
spdlog::trace("Replication client started at: {}:{}", endpoint.address, endpoint.port); spdlog::trace("Replication client started at: {}:{}", endpoint.address, endpoint.port);
client.StartFrequentCheck([&, license = license::global_license_checker.IsEnterpriseValidFast(), main_uuid]( client.StartFrequentCheck([&, license = license::global_license_checker.IsEnterpriseValidFast(), main_uuid](
bool reconnect, replication::ReplicationClient &client) mutable { bool reconnect, replication::ReplicationClient &client) mutable {
if (client.try_set_uuid && replication_coordination_glue::SendSwapMainUUIDRpc(client.rpc_client_, main_uuid)) { if (client.try_set_uuid &&
memgraph::replication_coordination_glue::SendSwapMainUUIDRpc(client.rpc_client_, main_uuid)) {
client.try_set_uuid = false; client.try_set_uuid = false;
} }
// Working connection // Working connection
// Check if system needs restoration // Check if system needs restoration
if (reconnect) { if (reconnect) {
client.state_.WithLock([](auto &state) { state = replication::ReplicationClient::State::BEHIND; }); client.state_.WithLock([](auto &state) { state = memgraph::replication::ReplicationClient::State::BEHIND; });
} }
// Check if license has changed // Check if license has changed
const auto new_license = license::global_license_checker.IsEnterpriseValidFast(); const auto new_license = license::global_license_checker.IsEnterpriseValidFast();
if (new_license != license) { if (new_license != license) {
license = new_license; license = new_license;
client.state_.WithLock([](auto &state) { state = replication::ReplicationClient::State::BEHIND; }); client.state_.WithLock([](auto &state) { state = memgraph::replication::ReplicationClient::State::BEHIND; });
} }
#ifdef MG_ENTERPRISE #ifdef MG_ENTERPRISE
SystemRestore<true>(client, system, dbms_handler, main_uuid, auth); SystemRestore<true>(client, system, dbms_handler, main_uuid, auth);
@@ -154,10 +154,10 @@ void StartReplicaClient(replication::ReplicationClient &client, dbms::DbmsHandle
// Check if any database has been left behind // Check if any database has been left behind
dbms_handler.ForEach([&name = client.name_, reconnect](dbms::DatabaseAccess db_acc) { dbms_handler.ForEach([&name = client.name_, reconnect](dbms::DatabaseAccess db_acc) {
// Specific database <-> replica client // Specific database <-> replica client
db_acc->storage()->repl_storage_state_.WithClient(name, [&](storage::ReplicationStorageClient &client) { db_acc->storage()->repl_storage_state_.WithClient(name, [&](storage::ReplicationStorageClient *client) {
if (reconnect || client.State() == storage::replication::ReplicaState::MAYBE_BEHIND) { if (reconnect || client->State() == storage::replication::ReplicaState::MAYBE_BEHIND) {
// Database <-> replica might be behind, check and recover // Database <-> replica might be behind, check and recover
client.TryCheckReplicaStateAsync(db_acc->storage(), db_acc); client->TryCheckReplicaStateAsync(db_acc->storage(), db_acc);
} }
}); });
}); });
@@ -165,8 +165,9 @@ void StartReplicaClient(replication::ReplicationClient &client, dbms::DbmsHandle
} }
#ifdef MG_ENTERPRISE #ifdef MG_ENTERPRISE
ReplicationHandler::ReplicationHandler(replication::ReplicationState &repl_state, dbms::DbmsHandler &dbms_handler, ReplicationHandler::ReplicationHandler(memgraph::replication::ReplicationState &repl_state,
system::System &system, auth::SynchedAuth &auth) memgraph::dbms::DbmsHandler &dbms_handler, memgraph::system::System &system,
memgraph::auth::SynchedAuth &auth)
: repl_state_{repl_state}, dbms_handler_{dbms_handler}, system_{system}, auth_{auth} { : repl_state_{repl_state}, dbms_handler_{dbms_handler}, system_{system}, auth_{auth} {
RecoverReplication(repl_state_, system_, dbms_handler_, auth_); RecoverReplication(repl_state_, system_, dbms_handler_, auth_);
} }
@@ -178,20 +179,20 @@ ReplicationHandler::ReplicationHandler(replication::ReplicationState &repl_state
#endif #endif
bool ReplicationHandler::SetReplicationRoleMain() { bool ReplicationHandler::SetReplicationRoleMain() {
auto const main_handler = [](replication::RoleMainData &) { auto const main_handler = [](memgraph::replication::RoleMainData &) {
// If we are already MAIN, we don't want to change anything // If we are already MAIN, we don't want to change anything
return false; return false;
}; };
auto const replica_handler = [this](replication::RoleReplicaData const &) { auto const replica_handler = [this](memgraph::replication::RoleReplicaData const &) {
return DoReplicaToMainPromotion(utils::UUID{}); return DoReplicaToMainPromotion(utils::UUID{});
}; };
// TODO: under lock // TODO: under lock
return std::visit(utils::Overloaded{main_handler, replica_handler}, repl_state_.ReplicationData()); return std::visit(memgraph::utils::Overloaded{main_handler, replica_handler}, repl_state_.ReplicationData());
} }
bool ReplicationHandler::SetReplicationRoleReplica(const replication::ReplicationServerConfig &config, bool ReplicationHandler::SetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
const std::optional<utils::UUID> &main_uuid) { const std::optional<utils::UUID> &main_uuid) {
return SetReplicationRoleReplica_<true>(config, main_uuid); return SetReplicationRoleReplica_<true>(config, main_uuid);
} }
@@ -237,16 +238,18 @@ auto ReplicationHandler::RegisterReplica(const memgraph::replication::Replicatio
return RegisterReplica_<false>(config); return RegisterReplica_<false>(config);
} }
auto ReplicationHandler::UnregisterReplica(std::string_view name) -> query::UnregisterReplicaResult { auto ReplicationHandler::UnregisterReplica(std::string_view name) -> memgraph::query::UnregisterReplicaResult {
auto const replica_handler = [](replication::RoleReplicaData const &) -> query::UnregisterReplicaResult { auto const replica_handler =
return query::UnregisterReplicaResult::NOT_MAIN; [](memgraph::replication::RoleReplicaData const &) -> memgraph::query::UnregisterReplicaResult {
return memgraph::query::UnregisterReplicaResult::NOT_MAIN;
}; };
auto const main_handler = [this, name](replication::RoleMainData &mainData) -> query::UnregisterReplicaResult { auto const main_handler =
[this, name](memgraph::replication::RoleMainData &mainData) -> memgraph::query::UnregisterReplicaResult {
if (!repl_state_.TryPersistUnregisterReplica(name)) { if (!repl_state_.TryPersistUnregisterReplica(name)) {
return query::UnregisterReplicaResult::COULD_NOT_BE_PERSISTED; return memgraph::query::UnregisterReplicaResult::COULD_NOT_BE_PERSISTED;
} }
// Remove database specific clients // Remove database specific clients
dbms_handler_.ForEach([name](dbms::DatabaseAccess db_acc) { dbms_handler_.ForEach([name](memgraph::dbms::DatabaseAccess db_acc) {
db_acc->storage()->repl_storage_state_.replication_clients_.WithLock([&name](auto &clients) { db_acc->storage()->repl_storage_state_.replication_clients_.WithLock([&name](auto &clients) {
std::erase_if(clients, [name](const auto &client) { return client->Name() == name; }); std::erase_if(clients, [name](const auto &client) { return client->Name() == name; });
}); });
@@ -254,14 +257,14 @@ auto ReplicationHandler::UnregisterReplica(std::string_view name) -> query::Unre
// Remove instance level clients // Remove instance level clients
auto const n_unregistered = auto const n_unregistered =
std::erase_if(mainData.registered_replicas_, [name](auto const &client) { return client.name_ == name; }); std::erase_if(mainData.registered_replicas_, [name](auto const &client) { return client.name_ == name; });
return n_unregistered != 0 ? query::UnregisterReplicaResult::SUCCESS return n_unregistered != 0 ? memgraph::query::UnregisterReplicaResult::SUCCESS
: query::UnregisterReplicaResult::CAN_NOT_UNREGISTER; : memgraph::query::UnregisterReplicaResult::CAN_NOT_UNREGISTER;
}; };
return std::visit(utils::Overloaded{main_handler, replica_handler}, repl_state_.ReplicationData()); return std::visit(memgraph::utils::Overloaded{main_handler, replica_handler}, repl_state_.ReplicationData());
} }
auto ReplicationHandler::GetRole() const -> replication_coordination_glue::ReplicationRole { auto ReplicationHandler::GetRole() const -> memgraph::replication_coordination_glue::ReplicationRole {
return repl_state_.GetRole(); return repl_state_.GetRole();
} }
@@ -272,57 +275,10 @@ auto ReplicationHandler::GetReplicaUUID() -> std::optional<utils::UUID> {
auto ReplicationHandler::GetReplState() const -> const memgraph::replication::ReplicationState & { return repl_state_; } auto ReplicationHandler::GetReplState() const -> const memgraph::replication::ReplicationState & { return repl_state_; }
auto ReplicationHandler::GetReplState() -> replication::ReplicationState & { return repl_state_; } auto ReplicationHandler::GetReplState() -> memgraph::replication::ReplicationState & { return repl_state_; }
bool ReplicationHandler::IsMain() const { return repl_state_.IsMain(); } bool ReplicationHandler::IsMain() const { return repl_state_.IsMain(); }
bool ReplicationHandler::IsReplica() const { return repl_state_.IsReplica(); } bool ReplicationHandler::IsReplica() const { return repl_state_.IsReplica(); }
auto ReplicationHandler::ShowReplicas() const -> utils::BasicResult<query::ShowReplicaError, query::ReplicasInfos> {
using res_t = utils::BasicResult<query::ShowReplicaError, query::ReplicasInfos>;
auto main = [this](RoleMainData const &main) -> res_t {
auto entries = std::vector<query::ReplicasInfo>{};
entries.reserve(main.registered_replicas_.size());
const bool full_info = license::global_license_checker.IsEnterpriseValidFast();
for (auto const &replica : main.registered_replicas_) {
// STEP 1: data_info
auto data_info = std::map<std::string, query::ReplicaInfoState>{};
this->dbms_handler_.ForEach([&](dbms::DatabaseAccess db_acc) {
auto *storage = db_acc->storage();
// ATM we only support IN_MEMORY_TRANSACTIONAL
if (storage->storage_mode_ != storage::StorageMode::IN_MEMORY_TRANSACTIONAL) return;
if (!full_info && storage->name() == dbms::kDefaultDB) return;
auto ok =
storage->repl_storage_state_.WithClient(replica.name_, [&](storage::ReplicationStorageClient &client) {
auto ts_info = client.GetTimestampInfo(storage);
auto state = client.State();
data_info.emplace(storage->name(),
query::ReplicaInfoState{ts_info.current_timestamp_of_replica,
ts_info.current_number_of_timestamp_behind_main, state});
});
DMG_ASSERT(ok);
});
// STEP 2: system_info
#ifdef MG_ENTERPRISE
// Already locked on system transaction via the interpreter
const auto ts = system_.LastCommittedSystemTimestamp();
// NOTE: no system behind at the moment
query::ReplicaSystemInfoState system_info{ts, 0 /* behind ts not implemented */, *replica.state_.ReadLock()};
#else
query::ReplicaSystemInfoState system_info{};
#endif
// STEP 3: add entry
entries.emplace_back(replica.name_, replica.rpc_client_.Endpoint().SocketAddress(), replica.mode_, system_info,
std::move(data_info));
}
return query::ReplicasInfos{std::move(entries)};
};
auto replica = [](RoleReplicaData const &) -> res_t { return query::ShowReplicaError::NOT_MAIN; };
return std::visit(utils::Overloaded{main, replica}, repl_state_.ReplicationData());
}
} // namespace memgraph::replication } // namespace memgraph::replication

View File

@@ -26,7 +26,7 @@ namespace memgraph::storage {
struct TimestampInfo { struct TimestampInfo {
uint64_t current_timestamp_of_replica; uint64_t current_timestamp_of_replica;
uint64_t current_number_of_timestamp_behind_main; uint64_t current_number_of_timestamp_behind_master;
}; };
struct ReplicaInfo { struct ReplicaInfo {

View File

@@ -95,7 +95,7 @@ void ReplicationStorageClient::UpdateReplicaState(Storage *storage, DatabaseAcce
TimestampInfo ReplicationStorageClient::GetTimestampInfo(Storage const *storage) { TimestampInfo ReplicationStorageClient::GetTimestampInfo(Storage const *storage) {
TimestampInfo info; TimestampInfo info;
info.current_timestamp_of_replica = 0; info.current_timestamp_of_replica = 0;
info.current_number_of_timestamp_behind_main = 0; info.current_number_of_timestamp_behind_master = 0;
try { try {
auto stream{client_.rpc_client_.Stream<replication::TimestampRpc>(main_uuid_, storage->uuid())}; auto stream{client_.rpc_client_.Stream<replication::TimestampRpc>(main_uuid_, storage->uuid())};
@@ -104,9 +104,9 @@ TimestampInfo ReplicationStorageClient::GetTimestampInfo(Storage const *storage)
auto main_time_stamp = storage->repl_storage_state_.last_commit_timestamp_.load(); auto main_time_stamp = storage->repl_storage_state_.last_commit_timestamp_.load();
info.current_timestamp_of_replica = response.current_commit_timestamp; info.current_timestamp_of_replica = response.current_commit_timestamp;
info.current_number_of_timestamp_behind_main = response.current_commit_timestamp - main_time_stamp; info.current_number_of_timestamp_behind_master = response.current_commit_timestamp - main_time_stamp;
if (!is_success || info.current_number_of_timestamp_behind_main != 0) { if (!is_success || info.current_number_of_timestamp_behind_master != 0) {
replica_state_.WithLock([](auto &val) { val = replication::ReplicaState::MAYBE_BEHIND; }); replica_state_.WithLock([](auto &val) { val = replication::ReplicaState::MAYBE_BEHIND; });
LogRpcFailure(); LogRpcFailure();
} }
@@ -215,11 +215,11 @@ bool ReplicationStorageClient::FinalizeTransactionReplication(Storage *storage,
MG_ASSERT(replica_stream_, "Missing stream for transaction deltas"); MG_ASSERT(replica_stream_, "Missing stream for transaction deltas");
try { try {
auto response = replica_stream_->Finalize(); auto response = replica_stream_->Finalize();
return replica_state_.WithLock([storage, &response, db_acc = std::move(db_acc), this](auto &state) mutable { return replica_state_.WithLock([storage, response, db_acc = std::move(db_acc), this](auto &state) mutable {
replica_stream_.reset(); replica_stream_.reset();
if (!response.success || state == replication::ReplicaState::RECOVERY) { if (!response.success || state == replication::ReplicaState::RECOVERY) {
state = replication::ReplicaState::RECOVERY; state = replication::ReplicaState::RECOVERY;
client_.thread_pool_.AddTask([storage, &response, db_acc = std::move(db_acc), this] { client_.thread_pool_.AddTask([storage, response, db_acc = std::move(db_acc), this] {
this->RecoverReplica(response.current_commit_timestamp, storage); this->RecoverReplica(response.current_commit_timestamp, storage);
}); });
return false; return false;

View File

@@ -63,7 +63,7 @@ struct ReplicationStorageState {
return replication_clients_.WithLock([replica_name, cb = std::forward<F>(callback)](auto &clients) { return replication_clients_.WithLock([replica_name, cb = std::forward<F>(callback)](auto &clients) {
for (const auto &client : clients) { for (const auto &client : clients) {
if (client->Name() == replica_name) { if (client->Name() == replica_name) {
cb(*client); cb(client.get());
return true; return true;
} }
} }

View File

@@ -34,7 +34,7 @@ struct System {
if (!system_unique.try_lock_for(try_time)) { if (!system_unique.try_lock_for(try_time)) {
return std::nullopt; return std::nullopt;
} }
return Transaction{state_, std::move(system_unique), ++timestamp_}; return Transaction{state_, std::move(system_unique), timestamp_++};
} }
// TODO: this and LastCommittedSystemTimestamp maybe not needed // TODO: this and LastCommittedSystemTimestamp maybe not needed
@@ -46,7 +46,7 @@ struct System {
private: private:
State state_; State state_;
std::timed_mutex mtx_{}; std::timed_mutex mtx_{};
std::uint64_t timestamp_{0}; std::uint64_t timestamp_{};
}; };
} // namespace memgraph::system } // namespace memgraph::system

View File

@@ -1,4 +1,4 @@
// Copyright 2024 Memgraph Ltd. // Copyright 2023 Memgraph Ltd.
// //
// Use of this software is governed by the Business Source License // 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 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -39,10 +39,9 @@ struct UUID {
UUID() { uuid_generate(uuid.data()); } UUID() { uuid_generate(uuid.data()); }
explicit operator std::string() const { explicit operator std::string() const {
// Note not using UUID_STR_LEN so we can build with older libuuid auto decoded = std::array<char, UUID_STR_LEN>{};
auto decoded = std::array<char, 37 /*UUID_STR_LEN*/>{};
uuid_unparse(uuid.data(), decoded.data()); uuid_unparse(uuid.data(), decoded.data());
return std::string{decoded.data(), 37 /*UUID_STR_LEN*/ - 1}; return std::string{decoded.data(), UUID_STR_LEN - 1};
} }
explicit operator arr_t() const { return uuid; } explicit operator arr_t() const { return uuid; }

View File

@@ -312,20 +312,8 @@ def test_unregister_replicas(kill_instance):
] ]
expected_replicas = [ expected_replicas = [
( ("instance_1", "127.0.0.1:10001", "sync", 0, 0, "ready"),
"instance_1", ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"127.0.0.1:10001",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"instance_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
] ]
mg_sleep_and_assert(expected_cluster, check_coordinator3) mg_sleep_and_assert(expected_cluster, check_coordinator3)
@@ -342,13 +330,7 @@ def test_unregister_replicas(kill_instance):
] ]
expected_replicas = [ expected_replicas = [
( ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"instance_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
] ]
mg_sleep_and_assert(expected_cluster, check_coordinator3) mg_sleep_and_assert(expected_cluster, check_coordinator3)
@@ -424,13 +406,7 @@ def test_unregister_main():
] ]
expected_replicas = [ expected_replicas = [
( ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"instance_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
] ]
main_cursor = connect(host="localhost", port=7687).cursor() main_cursor = connect(host="localhost", port=7687).cursor()

View File

@@ -17,7 +17,7 @@ import tempfile
import interactive_mg_runner import interactive_mg_runner
import pytest import pytest
from common import connect, execute_and_fetch_all, safe_execute from common import connect, execute_and_fetch_all, safe_execute
from mg_utils import mg_sleep_and_assert, mg_sleep_and_assert_collection from mg_utils import mg_sleep_and_assert
interactive_mg_runner.SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__)) interactive_mg_runner.SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
interactive_mg_runner.PROJECT_DIR = os.path.normpath( interactive_mg_runner.PROJECT_DIR = os.path.normpath(
@@ -123,23 +123,11 @@ def test_distributed_automatic_failover():
main_cursor = connect(host="localhost", port=7689).cursor() main_cursor = connect(host="localhost", port=7689).cursor()
expected_data_on_main = [ expected_data_on_main = [
( ("instance_1", "127.0.0.1:10001", "sync", 0, 0, "ready"),
"instance_1", ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"127.0.0.1:10001",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"instance_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
] ]
actual_data_on_main = sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;"))) actual_data_on_main = sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
assert actual_data_on_main == sorted(expected_data_on_main) assert actual_data_on_main == expected_data_on_main
interactive_mg_runner.kill(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3") interactive_mg_runner.kill(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3")
@@ -164,42 +152,18 @@ def test_distributed_automatic_failover():
return sorted(list(execute_and_fetch_all(new_main_cursor, "SHOW REPLICAS;"))) return sorted(list(execute_and_fetch_all(new_main_cursor, "SHOW REPLICAS;")))
expected_data_on_new_main = [ expected_data_on_new_main = [
( ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"instance_2", ("instance_3", "127.0.0.1:10003", "sync", 0, 0, "invalid"),
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"instance_3",
"127.0.0.1:10003",
"sync",
{"ts": 0, "behind": None, "status": "invalid"},
{"memgraph": {"ts": 0, "behind": 0, "status": "invalid"}},
),
] ]
mg_sleep_and_assert_collection(expected_data_on_new_main, retrieve_data_show_replicas) mg_sleep_and_assert(expected_data_on_new_main, retrieve_data_show_replicas)
interactive_mg_runner.start(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3") interactive_mg_runner.start(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3")
expected_data_on_new_main_old_alive = [ expected_data_on_new_main_old_alive = [
( ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"instance_2", ("instance_3", "127.0.0.1:10003", "sync", 0, 0, "ready"),
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"instance_3",
"127.0.0.1:10003",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
] ]
mg_sleep_and_assert_collection(expected_data_on_new_main_old_alive, retrieve_data_show_replicas) mg_sleep_and_assert(expected_data_on_new_main_old_alive, retrieve_data_show_replicas)
if __name__ == "__main__": if __name__ == "__main__":

View File

@@ -67,13 +67,7 @@ def test_replication_works_on_failover():
# 2 # 2
main_cursor = connect(host="localhost", port=7687).cursor() main_cursor = connect(host="localhost", port=7687).cursor()
expected_data_on_main = [ expected_data_on_main = [
( ("shared_replica", "127.0.0.1:10001", "sync", 0, 0, "ready"),
"shared_replica",
"127.0.0.1:10001",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
] ]
actual_data_on_main = sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;"))) actual_data_on_main = sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
assert actual_data_on_main == expected_data_on_main assert actual_data_on_main == expected_data_on_main
@@ -88,20 +82,8 @@ def test_replication_works_on_failover():
return sorted(list(execute_and_fetch_all(new_main_cursor, "SHOW REPLICAS;"))) return sorted(list(execute_and_fetch_all(new_main_cursor, "SHOW REPLICAS;")))
expected_data_on_new_main = [ expected_data_on_new_main = [
( ("replica", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"replica", ("shared_replica", "127.0.0.1:10001", "sync", 0, 0, "ready"),
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"shared_replica",
"127.0.0.1:10001",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
] ]
mg_sleep_and_assert(expected_data_on_new_main, retrieve_data_show_replicas) mg_sleep_and_assert(expected_data_on_new_main, retrieve_data_show_replicas)

View File

@@ -16,7 +16,7 @@ import tempfile
import interactive_mg_runner import interactive_mg_runner
import pytest import pytest
from common import connect, execute_and_fetch_all, safe_execute from common import connect, execute_and_fetch_all, safe_execute
from mg_utils import mg_sleep_and_assert, mg_sleep_and_assert_collection from mg_utils import mg_sleep_and_assert
interactive_mg_runner.SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__)) interactive_mg_runner.SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
interactive_mg_runner.PROJECT_DIR = os.path.normpath( interactive_mg_runner.PROJECT_DIR = os.path.normpath(
@@ -106,20 +106,8 @@ def test_replication_works_on_failover():
# 2 # 2
main_cursor = connect(host="localhost", port=7687).cursor() main_cursor = connect(host="localhost", port=7687).cursor()
expected_data_on_main = [ expected_data_on_main = [
( ("instance_1", "127.0.0.1:10001", "sync", 0, 0, "ready"),
"instance_1", ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"127.0.0.1:10001",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"instance_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
] ]
actual_data_on_main = sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;"))) actual_data_on_main = sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
assert actual_data_on_main == expected_data_on_main assert actual_data_on_main == expected_data_on_main
@@ -147,41 +135,17 @@ def test_replication_works_on_failover():
return sorted(list(execute_and_fetch_all(new_main_cursor, "SHOW REPLICAS;"))) return sorted(list(execute_and_fetch_all(new_main_cursor, "SHOW REPLICAS;")))
expected_data_on_new_main = [ expected_data_on_new_main = [
( ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"instance_2", ("instance_3", "127.0.0.1:10003", "sync", 0, 0, "invalid"),
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"instance_3",
"127.0.0.1:10003",
"sync",
{"ts": 0, "behind": None, "status": "invalid"},
{"memgraph": {"ts": 0, "behind": 0, "status": "invalid"}},
),
] ]
mg_sleep_and_assert_collection(expected_data_on_new_main, retrieve_data_show_replicas) mg_sleep_and_assert(expected_data_on_new_main, retrieve_data_show_replicas)
interactive_mg_runner.start(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3") interactive_mg_runner.start(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3")
expected_data_on_new_main = [ expected_data_on_new_main = [
( ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"instance_2", ("instance_3", "127.0.0.1:10003", "sync", 0, 0, "ready"),
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"instance_3",
"127.0.0.1:10003",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
] ]
mg_sleep_and_assert_collection(expected_data_on_new_main, retrieve_data_show_replicas) mg_sleep_and_assert(expected_data_on_new_main, retrieve_data_show_replicas)
# 5 # 5
execute_and_fetch_all(new_main_cursor, "CREATE ();") execute_and_fetch_all(new_main_cursor, "CREATE ();")
@@ -209,20 +173,8 @@ def test_replication_works_on_replica_instance_restart():
# 2 # 2
main_cursor = connect(host="localhost", port=7687).cursor() main_cursor = connect(host="localhost", port=7687).cursor()
expected_data_on_main = [ expected_data_on_main = [
( ("instance_1", "127.0.0.1:10001", "sync", 0, 0, "ready"),
"instance_1", ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"127.0.0.1:10001",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"instance_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
] ]
actual_data_on_main = sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;"))) actual_data_on_main = sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
assert actual_data_on_main == expected_data_on_main assert actual_data_on_main == expected_data_on_main
@@ -241,28 +193,16 @@ def test_replication_works_on_replica_instance_restart():
("instance_2", "", "127.0.0.1:10012", False, "unknown"), ("instance_2", "", "127.0.0.1:10012", False, "unknown"),
("instance_3", "", "127.0.0.1:10013", True, "main"), ("instance_3", "", "127.0.0.1:10013", True, "main"),
] ]
mg_sleep_and_assert_collection(expected_data_on_coord, retrieve_data_show_repl_cluster) mg_sleep_and_assert(expected_data_on_coord, retrieve_data_show_repl_cluster)
def retrieve_data_show_replicas(): def retrieve_data_show_replicas():
return sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;"))) return sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
expected_data_on_main = [ expected_data_on_main = [
( ("instance_1", "127.0.0.1:10001", "sync", 0, 0, "ready"),
"instance_1", ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "invalid"),
"127.0.0.1:10001",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"instance_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "invalid"}},
),
] ]
mg_sleep_and_assert_collection(expected_data_on_main, retrieve_data_show_replicas) mg_sleep_and_assert(expected_data_on_main, retrieve_data_show_replicas)
# 4 # 4
instance_1_cursor = connect(host="localhost", port=7688).cursor() instance_1_cursor = connect(host="localhost", port=7688).cursor()
@@ -277,22 +217,10 @@ def test_replication_works_on_replica_instance_restart():
return sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;"))) return sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
expected_data_on_main = [ expected_data_on_main = [
( ("instance_1", "127.0.0.1:10001", "sync", 2, 0, "ready"),
"instance_1", ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "invalid"),
"127.0.0.1:10001",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 2, "behind": 0, "status": "ready"}},
),
(
"instance_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "invalid"}},
),
] ]
mg_sleep_and_assert_collection(expected_data_on_main, retrieve_data_show_replicas) mg_sleep_and_assert(expected_data_on_main, retrieve_data_show_replicas)
# 5. # 5.
@@ -313,22 +241,10 @@ def test_replication_works_on_replica_instance_restart():
return sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;"))) return sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
expected_data_on_main = [ expected_data_on_main = [
( ("instance_1", "127.0.0.1:10001", "sync", 2, 0, "ready"),
"instance_1", ("instance_2", "127.0.0.1:10002", "sync", 2, 0, "ready"),
"127.0.0.1:10001",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 2, "behind": 0, "status": "ready"}},
),
(
"instance_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 2, "behind": 0, "status": "ready"}},
),
] ]
mg_sleep_and_assert_collection(expected_data_on_main, retrieve_data_show_replicas) mg_sleep_and_assert(expected_data_on_main, retrieve_data_show_replicas)
# 6. # 6.
instance_2_cursor = connect(port=7689, host="localhost").cursor() instance_2_cursor = connect(port=7689, host="localhost").cursor()
@@ -397,23 +313,11 @@ def test_simple_automatic_failover():
main_cursor = connect(host="localhost", port=7687).cursor() main_cursor = connect(host="localhost", port=7687).cursor()
expected_data_on_main = [ expected_data_on_main = [
( ("instance_1", "127.0.0.1:10001", "sync", 0, 0, "ready"),
"instance_1", ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"127.0.0.1:10001",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"instance_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
] ]
actual_data_on_main = sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;"))) actual_data_on_main = sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
assert actual_data_on_main == sorted(expected_data_on_main) assert actual_data_on_main == expected_data_on_main
interactive_mg_runner.kill(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3") interactive_mg_runner.kill(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3")
@@ -436,42 +340,18 @@ def test_simple_automatic_failover():
return sorted(list(execute_and_fetch_all(new_main_cursor, "SHOW REPLICAS;"))) return sorted(list(execute_and_fetch_all(new_main_cursor, "SHOW REPLICAS;")))
expected_data_on_new_main = [ expected_data_on_new_main = [
( ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"instance_2", ("instance_3", "127.0.0.1:10003", "sync", 0, 0, "invalid"),
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"instance_3",
"127.0.0.1:10003",
"sync",
{"ts": 0, "behind": None, "status": "invalid"},
{"memgraph": {"ts": 0, "behind": 0, "status": "invalid"}},
),
] ]
mg_sleep_and_assert_collection(expected_data_on_new_main, retrieve_data_show_replicas) mg_sleep_and_assert(expected_data_on_new_main, retrieve_data_show_replicas)
interactive_mg_runner.start(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3") interactive_mg_runner.start(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3")
expected_data_on_new_main_old_alive = [ expected_data_on_new_main_old_alive = [
( ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"instance_2", ("instance_3", "127.0.0.1:10003", "sync", 0, 0, "ready"),
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"instance_3",
"127.0.0.1:10003",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
] ]
mg_sleep_and_assert_collection(expected_data_on_new_main_old_alive, retrieve_data_show_replicas) mg_sleep_and_assert(expected_data_on_new_main_old_alive, retrieve_data_show_replicas)
def test_registering_replica_fails_name_exists(): def test_registering_replica_fails_name_exists():

View File

@@ -15,21 +15,3 @@ def mg_sleep_and_assert(expected_value, function_to_retrieve_data, max_duration=
result = function_to_retrieve_data() result = function_to_retrieve_data()
return result return result
def mg_sleep_and_assert_collection(
expected_value, function_to_retrieve_data, max_duration=20, time_between_attempt=0.2
):
result = function_to_retrieve_data()
start_time = time.time()
while len(result) != len(expected_value) or any((x not in result for x in expected_value)):
duration = time.time() - start_time
if duration > max_duration:
assert (
False
), f" mg_sleep_and_assert has tried for too long and did not get the expected result! Last result was: {result}"
time.sleep(time_between_attempt)
result = function_to_retrieve_data()
return result

View File

@@ -14,7 +14,7 @@ import time
import pytest import pytest
from common import execute_and_fetch_all from common import execute_and_fetch_all
from mg_utils import mg_sleep_and_assert_collection from mg_utils import mg_sleep_and_assert
# BUGFIX: for issue https://github.com/memgraph/memgraph/issues/1515 # BUGFIX: for issue https://github.com/memgraph/memgraph/issues/1515
@@ -28,52 +28,28 @@ def test_replication_handles_delete_when_multiple_edges_of_same_type(connection)
conn = connection(7687, "main") conn = connection(7687, "main")
conn.autocommit = True conn.autocommit = True
cursor = conn.cursor() cursor = conn.cursor()
actual_data = execute_and_fetch_all(cursor, "SHOW REPLICAS;") actual_data = set(execute_and_fetch_all(cursor, "SHOW REPLICAS;"))
expected_data = [ expected_data = {
( ("replica_1", "127.0.0.1:10001", "sync", 0, 0, "ready"),
"replica_1", ("replica_2", "127.0.0.1:10002", "async", 0, 0, "ready"),
"127.0.0.1:10001", }
"sync", assert actual_data == expected_data
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"replica_2",
"127.0.0.1:10002",
"async",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
]
assert all([x in actual_data for x in expected_data])
# 1/ # 1/
execute_and_fetch_all(cursor, "CREATE (a)-[r:X]->(b) CREATE (a)-[:X]->(b) DELETE r;") execute_and_fetch_all(cursor, "CREATE (a)-[r:X]->(b) CREATE (a)-[:X]->(b) DELETE r;")
# 2/ # 2/
expected_data = [ expected_data = {
( ("replica_1", "127.0.0.1:10001", "sync", 2, 0, "ready"),
"replica_1", ("replica_2", "127.0.0.1:10002", "async", 2, 0, "ready"),
"127.0.0.1:10001", }
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 2, "behind": 0, "status": "ready"}},
),
(
"replica_2",
"127.0.0.1:10002",
"async",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 2, "behind": 0, "status": "ready"}},
),
]
def retrieve_data(): def retrieve_data():
return execute_and_fetch_all(cursor, "SHOW REPLICAS;") return set(execute_and_fetch_all(cursor, "SHOW REPLICAS;"))
actual_data = mg_sleep_and_assert_collection(expected_data, retrieve_data) actual_data = mg_sleep_and_assert(expected_data, retrieve_data)
assert all([x in actual_data for x in expected_data]) assert actual_data == expected_data
if __name__ == "__main__": if __name__ == "__main__":

View File

@@ -10,11 +10,12 @@
# licenses/APL.txt. # licenses/APL.txt.
import sys import sys
import time
import pytest import pytest
import time
from common import execute_and_fetch_all from common import execute_and_fetch_all
from mg_utils import mg_sleep_and_assert_collection from mg_utils import mg_sleep_and_assert
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -30,42 +31,25 @@ def test_show_replication_role(port, role, connection):
def test_show_replicas(connection): def test_show_replicas(connection):
cursor = connection(7687, "main").cursor() cursor = connection(7687, "main").cursor()
actual_data = execute_and_fetch_all(cursor, "SHOW REPLICAS;") actual_data = set(execute_and_fetch_all(cursor, "SHOW REPLICAS;"))
expected_column_names = { expected_column_names = {
"name", "name",
"socket_address", "socket_address",
"sync_mode", "sync_mode",
"system_info", "current_timestamp_of_replica",
"data_info", "number_of_timestamp_behind_master",
"state",
} }
actual_column_names = {x.name for x in cursor.description} actual_column_names = {x.name for x in cursor.description}
assert actual_column_names == expected_column_names assert actual_column_names == expected_column_names
expected_data = [ expected_data = {
( ("replica_1", "127.0.0.1:10001", "sync", 0, 0, "ready"),
"replica_1", ("replica_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"127.0.0.1:10001", ("replica_3", "127.0.0.1:10003", "async", 0, 0, "ready"),
"sync", }
{"ts": 0, "behind": None, "status": "ready"}, assert actual_data == expected_data
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"replica_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"replica_3",
"127.0.0.1:10003",
"async",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
]
assert all([x in actual_data for x in expected_data])
def test_show_replicas_while_inserting_data(connection): def test_show_replicas_while_inserting_data(connection):
@@ -78,108 +62,49 @@ def test_show_replicas_while_inserting_data(connection):
# 0/ # 0/
cursor = connection(7687, "main").cursor() cursor = connection(7687, "main").cursor()
actual_data = execute_and_fetch_all(cursor, "SHOW REPLICAS;") actual_data = set(execute_and_fetch_all(cursor, "SHOW REPLICAS;"))
expected_column_names = { expected_column_names = {
"name", "name",
"socket_address", "socket_address",
"sync_mode", "sync_mode",
"system_info", "current_timestamp_of_replica",
"data_info", "number_of_timestamp_behind_master",
"state",
} }
actual_column_names = {x.name for x in cursor.description} actual_column_names = {x.name for x in cursor.description}
assert actual_column_names == expected_column_names assert actual_column_names == expected_column_names
expected_data = [ expected_data = {
( ("replica_1", "127.0.0.1:10001", "sync", 0, 0, "ready"),
"replica_1", ("replica_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
"127.0.0.1:10001", ("replica_3", "127.0.0.1:10003", "async", 0, 0, "ready"),
"sync", }
{"ts": 0, "behind": None, "status": "ready"}, assert actual_data == expected_data
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"replica_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"replica_3",
"127.0.0.1:10003",
"async",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
]
assert all([x in actual_data for x in expected_data])
# 1/ # 1/
execute_and_fetch_all(cursor, "CREATE (n1:Number {name: 'forty_two', value:42});") execute_and_fetch_all(cursor, "CREATE (n1:Number {name: 'forty_two', value:42});")
# 2/ # 2/
expected_data = [ expected_data = {
( ("replica_1", "127.0.0.1:10001", "sync", 4, 0, "ready"),
"replica_1", ("replica_2", "127.0.0.1:10002", "sync", 4, 0, "ready"),
"127.0.0.1:10001", ("replica_3", "127.0.0.1:10003", "async", 4, 0, "ready"),
"sync", }
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 4, "behind": 0, "status": "ready"}},
),
(
"replica_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 4, "behind": 0, "status": "ready"}},
),
(
"replica_3",
"127.0.0.1:10003",
"async",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 4, "behind": 0, "status": "ready"}},
),
]
def retrieve_data(): def retrieve_data():
return execute_and_fetch_all(cursor, "SHOW REPLICAS;") return set(execute_and_fetch_all(cursor, "SHOW REPLICAS;"))
actual_data = mg_sleep_and_assert_collection(expected_data, retrieve_data) actual_data = mg_sleep_and_assert(expected_data, retrieve_data)
assert all([x in actual_data for x in expected_data]) assert actual_data == expected_data
# 3/ # 3/
res = execute_and_fetch_all(cursor, "MATCH (node) return node;") res = execute_and_fetch_all(cursor, "MATCH (node) return node;")
assert len(res) == 1 assert len(res) == 1
# 4/ # 4/
expected_data = [ actual_data = set(execute_and_fetch_all(cursor, "SHOW REPLICAS;"))
( assert actual_data == expected_data
"replica_1",
"127.0.0.1:10001",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 4, "behind": 0, "status": "ready"}},
),
(
"replica_2",
"127.0.0.1:10002",
"sync",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 4, "behind": 0, "status": "ready"}},
),
(
"replica_3",
"127.0.0.1:10003",
"async",
{"ts": 0, "behind": None, "status": "ready"},
{"memgraph": {"ts": 4, "behind": 0, "status": "ready"}},
),
]
actual_data = execute_and_fetch_all(cursor, "SHOW REPLICAS;")
assert all([x in actual_data for x in expected_data])
if __name__ == "__main__": if __name__ == "__main__":

File diff suppressed because it is too large Load Diff

View File

@@ -22,7 +22,7 @@ import interactive_mg_runner
import mgclient import mgclient
import pytest import pytest
from common import execute_and_fetch_all from common import execute_and_fetch_all
from mg_utils import mg_sleep_and_assert, mg_sleep_and_assert_collection from mg_utils import mg_sleep_and_assert
interactive_mg_runner.SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__)) interactive_mg_runner.SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
interactive_mg_runner.PROJECT_DIR = os.path.normpath( interactive_mg_runner.PROJECT_DIR = os.path.normpath(
@@ -35,10 +35,6 @@ BOLT_PORTS = {"main": 7687, "replica_1": 7688, "replica_2": 7689}
REPLICATION_PORTS = {"replica_1": 10001, "replica_2": 10002} REPLICATION_PORTS = {"replica_1": 10001, "replica_2": 10002}
def set_eq(actual, expected):
return len(actual) == len(expected) and all([x in actual for x in expected])
def create_memgraph_instances_with_role_recovery(data_directory: Any) -> Dict[str, Any]: def create_memgraph_instances_with_role_recovery(data_directory: Any) -> Dict[str, Any]:
return { return {
"replica_1": { "replica_1": {
@@ -178,9 +174,10 @@ def setup_main(main_cursor):
execute_and_fetch_all(main_cursor, "CREATE (:Node{on:'B'});") execute_and_fetch_all(main_cursor, "CREATE (:Node{on:'B'});")
def show_replicas_func(cursor): def show_replicas_func(cursor, db_name):
def func(): def func():
return execute_and_fetch_all(cursor, "SHOW REPLICAS;") execute_and_fetch_all(cursor, f"USE DATABASE {db_name};")
return set(execute_and_fetch_all(cursor, "SHOW REPLICAS;"))
return func return func
@@ -274,31 +271,17 @@ def test_manual_databases_create_multitenancy_replication(connection):
execute_and_fetch_all(cursor, "CREATE ()-[:EDGE]->();") execute_and_fetch_all(cursor, "CREATE ()-[:EDGE]->();")
# 2/ # 2/
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 1, 0, "ready"),
"replica_1", ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 1, 0, "ready"),
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", }
"sync", mg_sleep_and_assert(expected_data, show_replicas_func(cursor, "A"))
{"ts": 2, "behind": None, "status": "ready"},
{ expected_data = {
"A": {"ts": 1, "behind": 0, "status": "ready"}, ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 1, 0, "ready"),
"B": {"ts": 1, "behind": 0, "status": "ready"}, ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 1, 0, "ready"),
"memgraph": {"ts": 0, "behind": 0, "status": "ready"}, }
}, mg_sleep_and_assert(expected_data, show_replicas_func(cursor, "B"))
),
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{"ts": 2, "behind": None, "status": "ready"},
{
"A": {"ts": 1, "behind": 0, "status": "ready"},
"B": {"ts": 1, "behind": 0, "status": "ready"},
"memgraph": {"ts": 0, "behind": 0, "status": "ready"},
},
),
]
mg_sleep_and_assert_collection(expected_data, show_replicas_func(cursor))
cursor_replica = connection(BOLT_PORTS["replica_1"], "replica").cursor() cursor_replica = connection(BOLT_PORTS["replica_1"], "replica").cursor()
assert get_number_of_nodes_func(cursor_replica, "A")() == 1 assert get_number_of_nodes_func(cursor_replica, "A")() == 1
@@ -540,23 +523,11 @@ def test_manual_databases_create_multitenancy_replication_main_behind(connection
execute_and_fetch_all(main_cursor, "CREATE DATABASE A;") execute_and_fetch_all(main_cursor, "CREATE DATABASE A;")
# 2/ # 2/
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 0, 0, "ready"),
"replica_1", ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 0, 0, "ready"),
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", }
"sync", mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "A"))
{"ts": 3, "behind": None, "status": "ready"},
{"A": {"ts": 0, "behind": 0, "status": "ready"}, "memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{"ts": 3, "behind": None, "status": "ready"},
{"A": {"ts": 0, "behind": 0, "status": "ready"}, "memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
]
mg_sleep_and_assert_collection(expected_data, show_replicas_func(main_cursor))
databases_on_main = show_databases_func(main_cursor)() databases_on_main = show_databases_func(main_cursor)()
@@ -596,31 +567,17 @@ def test_automatic_databases_create_multitenancy_replication(connection):
execute_and_fetch_all(main_cursor, "CREATE (:Node)-[:EDGE]->(:Node)") execute_and_fetch_all(main_cursor, "CREATE (:Node)-[:EDGE]->(:Node)")
# 3/ # 3/
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 7, 0, "ready"),
"replica_1", ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 7, 0, "ready"),
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", }
"sync", mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "A"))
{"ts": 4, "behind": None, "status": "ready"},
{ expected_data = {
"A": {"ts": 7, "behind": 0, "status": "ready"}, ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 0, 0, "ready"),
"B": {"ts": 0, "behind": 0, "status": "ready"}, ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 0, 0, "ready"),
"memgraph": {"ts": 0, "behind": 0, "status": "ready"}, }
}, mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "B"))
),
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{"ts": 4, "behind": None, "status": "ready"},
{
"A": {"ts": 7, "behind": 0, "status": "ready"},
"B": {"ts": 0, "behind": 0, "status": "ready"},
"memgraph": {"ts": 0, "behind": 0, "status": "ready"},
},
),
]
mg_sleep_and_assert_collection(expected_data, show_replicas_func(main_cursor))
cursor_replica = connection(BOLT_PORTS["replica_1"], "replica").cursor() cursor_replica = connection(BOLT_PORTS["replica_1"], "replica").cursor()
assert get_number_of_nodes_func(cursor_replica, "A")() == 7 assert get_number_of_nodes_func(cursor_replica, "A")() == 7
@@ -683,26 +640,19 @@ def test_automatic_databases_multitenancy_replication_predefined(connection):
execute_and_fetch_all(cursor, "CREATE ()-[:EDGE]->();") execute_and_fetch_all(cursor, "CREATE ()-[:EDGE]->();")
# 2/ # 2/
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 1, 0, "ready"),
"replica_1", }
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", mg_sleep_and_assert(expected_data, show_replicas_func(cursor, "A"))
"sync",
{"ts": 2, "behind": None, "status": "ready"}, expected_data = {
{ ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 1, 0, "ready"),
"A": {"ts": 1, "behind": 0, "status": "ready"}, }
"B": {"ts": 1, "behind": 0, "status": "ready"}, mg_sleep_and_assert(expected_data, show_replicas_func(cursor, "B"))
"memgraph": {"ts": 0, "behind": 0, "status": "ready"},
},
),
]
mg_sleep_and_assert_collection(expected_data, show_replicas_func(cursor))
cursor_replica = connection(BOLT_PORTS["replica_1"], "replica").cursor() cursor_replica = connection(BOLT_PORTS["replica_1"], "replica").cursor()
assert get_number_of_nodes_func(cursor_replica, "A")() == 1 assert get_number_of_nodes_func(cursor_replica, "A")() == 1
assert get_number_of_edges_func(cursor_replica, "A")() == 0 assert get_number_of_edges_func(cursor_replica, "A")() == 0
assert get_number_of_nodes_func(cursor_replica, "B")() == 2
assert get_number_of_edges_func(cursor_replica, "B")() == 1
def test_automatic_databases_create_multitenancy_replication_dirty_main(connection): def test_automatic_databases_create_multitenancy_replication_dirty_main(connection):
@@ -748,16 +698,10 @@ def test_automatic_databases_create_multitenancy_replication_dirty_main(connecti
cursor = connection(BOLT_PORTS["main"], "main").cursor() cursor = connection(BOLT_PORTS["main"], "main").cursor()
# 1/ # 1/
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 1, 0, "ready"),
"replica_1", }
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", mg_sleep_and_assert(expected_data, show_replicas_func(cursor, "A"))
"sync",
{"ts": 1, "behind": None, "status": "ready"},
{"A": {"ts": 1, "behind": 0, "status": "ready"}, "memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
]
mg_sleep_and_assert_collection(expected_data, show_replicas_func(cursor))
cursor_replica = connection(BOLT_PORTS["replica_1"], "replica").cursor() cursor_replica = connection(BOLT_PORTS["replica_1"], "replica").cursor()
execute_and_fetch_all(cursor_replica, "USE DATABASE A;") execute_and_fetch_all(cursor_replica, "USE DATABASE A;")
@@ -796,85 +740,31 @@ def test_multitenancy_replication_restart_replica_w_fc(connection, replica_name)
time.sleep(3) # In order for the frequent check to run time.sleep(3) # In order for the frequent check to run
# Check that the FC did invalidate # Check that the FC did invalidate
expected_data = { expected_data = {
"replica_1": [ "replica_1": {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 0, 0, "invalid"),
"replica_1", ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 7, 0, "ready"),
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", },
"sync", "replica_2": {
{"ts": 4, "behind": None, "status": "ready"}, ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 7, 0, "ready"),
{ ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 0, 0, "invalid"),
"A": {"ts": 0, "behind": 0, "status": "invalid"}, },
"B": {"ts": 0, "behind": 0, "status": "invalid"},
"memgraph": {"ts": 0, "behind": 0, "status": "invalid"},
},
),
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{"ts": 4, "behind": None, "status": "ready"},
{
"A": {"ts": 7, "behind": 0, "status": "ready"},
"B": {"ts": 3, "behind": 0, "status": "ready"},
"memgraph": {"ts": 0, "behind": 0, "status": "ready"},
},
),
],
"replica_2": [
(
"replica_1",
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}",
"sync",
{"ts": 4, "behind": None, "status": "ready"},
{
"A": {"ts": 7, "behind": 0, "status": "ready"},
"B": {"ts": 3, "behind": 0, "status": "ready"},
"memgraph": {"ts": 0, "behind": 0, "status": "ready"},
},
),
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{"ts": 4, "behind": None, "status": "ready"},
{
"A": {"ts": 0, "behind": 0, "status": "invalid"},
"B": {"ts": 0, "behind": 0, "status": "invalid"},
"memgraph": {"ts": 0, "behind": 0, "status": "invalid"},
},
),
],
} }
assert set_eq(expected_data[replica_name], show_replicas_func(main_cursor)()) assert expected_data[replica_name] == show_replicas_func(main_cursor, "A")()
# Restart # Restart
interactive_mg_runner.start(MEMGRAPH_INSTANCES_DESCRIPTION, replica_name) interactive_mg_runner.start(MEMGRAPH_INSTANCES_DESCRIPTION, replica_name)
# 4/ # 4/
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 7, 0, "ready"),
"replica_1", ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 7, 0, "ready"),
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", }
"sync", mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "A"))
{"ts": 4, "behind": None, "status": "ready"},
{ expected_data = {
"A": {"ts": 7, "behind": 0, "status": "ready"}, ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 3, 0, "ready"),
"B": {"ts": 3, "behind": 0, "status": "ready"}, ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 3, 0, "ready"),
"memgraph": {"ts": 0, "behind": 0, "status": "ready"}, }
}, mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "B"))
),
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{"ts": 4, "behind": None, "status": "ready"},
{
"A": {"ts": 7, "behind": 0, "status": "ready"},
"B": {"ts": 3, "behind": 0, "status": "ready"},
"memgraph": {"ts": 0, "behind": 0, "status": "ready"},
},
),
]
mg_sleep_and_assert_collection(expected_data, show_replicas_func(main_cursor))
cursor_replica = connection(BOLT_PORTS[replica_name], "replica").cursor() cursor_replica = connection(BOLT_PORTS[replica_name], "replica").cursor()
@@ -915,33 +805,19 @@ def test_multitenancy_replication_restart_replica_wo_fc(connection, replica_name
interactive_mg_runner.start(MEMGRAPH_INSTANCES_DESCRIPTION, replica_name) interactive_mg_runner.start(MEMGRAPH_INSTANCES_DESCRIPTION, replica_name)
# 4/ # 4/
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 7, 0, "ready"),
"replica_1", ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 7, 0, "ready"),
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", }
"sync", mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "A"))
{"ts": 4, "behind": None, "status": "ready"},
{
"A": {"ts": 7, "behind": 0, "status": "ready"},
"B": {"ts": 3, "behind": 0, "status": "ready"},
"memgraph": {"ts": 0, "behind": 0, "status": "ready"},
},
),
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{"ts": 4, "behind": None, "status": "ready"},
{
"A": {"ts": 7, "behind": 0, "status": "ready"},
"B": {"ts": 3, "behind": 0, "status": "ready"},
"memgraph": {"ts": 0, "behind": 0, "status": "ready"},
},
),
]
mg_sleep_and_assert_collection(expected_data, show_replicas_func(main_cursor))
cursor_replica = connection(BOLT_PORTS[replica_name], replica_name).cursor() expected_data = {
("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 3, 0, "ready"),
("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 3, 0, "ready"),
}
mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "B"))
cursor_replica = connection(BOLT_PORTS[replica_name], "replica").cursor()
assert get_number_of_nodes_func(cursor_replica, "A")() == 7 assert get_number_of_nodes_func(cursor_replica, "A")() == 7
assert get_number_of_edges_func(cursor_replica, "A")() == 3 assert get_number_of_edges_func(cursor_replica, "A")() == 3
assert get_number_of_nodes_func(cursor_replica, "B")() == 2 assert get_number_of_nodes_func(cursor_replica, "B")() == 2
@@ -1023,28 +899,17 @@ def test_multitenancy_replication_drop_replica(connection, replica_name):
) )
# 4/ # 4/
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 7, 0, "ready"),
"replica_1", ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 7, 0, "ready"),
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", }
"sync", mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "A"))
{
"A": {"ts": 7, "behind": 0, "status": "ready"}, expected_data = {
"B": {"ts": 3, "behind": 0, "status": "ready"}, ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 3, 0, "ready"),
"memgraph": {"ts": 0, "behind": 0, "status": "ready"}, ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 3, 0, "ready"),
}, }
), mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "B"))
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{
"A": {"ts": 7, "behind": 0, "status": "ready"},
"B": {"ts": 3, "behind": 0, "status": "ready"},
"memgraph": {"ts": 0, "behind": 0, "status": "ready"},
},
),
]
cursor_replica = connection(BOLT_PORTS[replica_name], "replica").cursor() cursor_replica = connection(BOLT_PORTS[replica_name], "replica").cursor()
assert get_number_of_nodes_func(cursor_replica, "A")() == 7 assert get_number_of_nodes_func(cursor_replica, "A")() == 7
@@ -1128,31 +993,17 @@ def test_automatic_databases_drop_multitenancy_replication(connection):
execute_and_fetch_all(main_cursor, "CREATE (:Node{on:'A'});") execute_and_fetch_all(main_cursor, "CREATE (:Node{on:'A'});")
# 3/ # 3/
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 1, 0, "ready"),
"replica_1", ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 1, 0, "ready"),
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", }
"sync", mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "A"))
{"ts": 4, "behind": None, "status": "ready"},
{ expected_data = {
"A": {"ts": 1, "behind": 0, "status": "ready"}, ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 0, 0, "ready"),
"B": {"ts": 0, "behind": 0, "status": "ready"}, ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 0, 0, "ready"),
"memgraph": {"ts": 0, "behind": 0, "status": "ready"}, }
}, mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "B"))
),
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{"ts": 4, "behind": None, "status": "ready"},
{
"A": {"ts": 1, "behind": 0, "status": "ready"},
"B": {"ts": 0, "behind": 0, "status": "ready"},
"memgraph": {"ts": 0, "behind": 0, "status": "ready"},
},
),
]
mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor))
# 4/ # 4/
execute_and_fetch_all(main_cursor, "USE DATABASE memgraph;") execute_and_fetch_all(main_cursor, "USE DATABASE memgraph;")
@@ -1237,23 +1088,11 @@ def test_multitenancy_drop_while_replica_using(connection):
execute_and_fetch_all(main_cursor, "CREATE (:Node{on:'A'});") execute_and_fetch_all(main_cursor, "CREATE (:Node{on:'A'});")
# 3/ # 3/
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 1, 0, "ready"),
"replica_1", ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 1, 0, "ready"),
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", }
"sync", mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "A"))
{"ts": 3, "behind": None, "status": "ready"},
{"A": {"ts": 1, "behind": 0, "status": "ready"}, "memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{"ts": 3, "behind": None, "status": "ready"},
{"A": {"ts": 1, "behind": 0, "status": "ready"}, "memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
]
mg_sleep_and_assert_collection(expected_data, show_replicas_func(main_cursor))
# 4/ # 4/
replica1_cursor = connection(BOLT_PORTS["replica_1"], "replica").cursor() replica1_cursor = connection(BOLT_PORTS["replica_1"], "replica").cursor()
@@ -1271,23 +1110,11 @@ def test_multitenancy_drop_while_replica_using(connection):
execute_and_fetch_all(main_cursor, "CREATE DATABASE B;") execute_and_fetch_all(main_cursor, "CREATE DATABASE B;")
execute_and_fetch_all(main_cursor, "USE DATABASE B;") execute_and_fetch_all(main_cursor, "USE DATABASE B;")
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 0, 0, "ready"),
"replica_1", ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 0, 0, "ready"),
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", }
"sync", mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "B"))
{"ts": 8, "behind": None, "status": "ready"},
{"B": {"ts": 0, "behind": 0, "status": "ready"}, "memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{"ts": 8, "behind": None, "status": "ready"},
{"B": {"ts": 0, "behind": 0, "status": "ready"}, "memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
]
mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor))
# 6/ # 6/
assert execute_and_fetch_all(replica1_cursor, "MATCH(n) RETURN count(*);")[0][0] == 1 assert execute_and_fetch_all(replica1_cursor, "MATCH(n) RETURN count(*);")[0][0] == 1
@@ -1336,23 +1163,11 @@ def test_multitenancy_drop_and_recreate_while_replica_using(connection):
execute_and_fetch_all(main_cursor, "CREATE (:Node{on:'A'});") execute_and_fetch_all(main_cursor, "CREATE (:Node{on:'A'});")
# 3/ # 3/
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 1, 0, "ready"),
"replica_1", ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 1, 0, "ready"),
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", }
"sync", mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "A"))
{"ts": 3, "behind": None, "status": "ready"},
{"A": {"ts": 1, "behind": 0, "status": "ready"}, "memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{"ts": 3, "behind": None, "status": "ready"},
{"A": {"ts": 1, "behind": 0, "status": "ready"}, "memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
]
mg_sleep_and_assert_collection(expected_data, show_replicas_func(main_cursor))
# 4/ # 4/
replica1_cursor = connection(BOLT_PORTS["replica_1"], "replica").cursor() replica1_cursor = connection(BOLT_PORTS["replica_1"], "replica").cursor()
@@ -1369,23 +1184,11 @@ def test_multitenancy_drop_and_recreate_while_replica_using(connection):
execute_and_fetch_all(main_cursor, "CREATE DATABASE A;") execute_and_fetch_all(main_cursor, "CREATE DATABASE A;")
execute_and_fetch_all(main_cursor, "USE DATABASE A;") execute_and_fetch_all(main_cursor, "USE DATABASE A;")
expected_data = [ expected_data = {
( ("replica_1", f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", "sync", 0, 0, "ready"),
"replica_1", ("replica_2", f"127.0.0.1:{REPLICATION_PORTS['replica_2']}", "async", 0, 0, "ready"),
f"127.0.0.1:{REPLICATION_PORTS['replica_1']}", }
"sync", mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor, "A"))
{"ts": 8, "behind": None, "status": "ready"},
{"A": {"ts": 0, "behind": 0, "status": "ready"}, "memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
(
"replica_2",
f"127.0.0.1:{REPLICATION_PORTS['replica_2']}",
"async",
{"ts": 8, "behind": None, "status": "ready"},
{"A": {"ts": 0, "behind": 0, "status": "ready"}, "memgraph": {"ts": 0, "behind": 0, "status": "ready"}},
),
]
mg_sleep_and_assert(expected_data, show_replicas_func(main_cursor))
# 6/ # 6/
assert execute_and_fetch_all(replica1_cursor, "MATCH(n) RETURN count(*);")[0][0] == 1 assert execute_and_fetch_all(replica1_cursor, "MATCH(n) RETURN count(*);")[0][0] == 1

View File

@@ -1,4 +1,4 @@
// Copyright 2023 Memgraph Ltd. // Copyright 2024 Memgraph Ltd.
// //
// Use of this software is governed by the Business Source License // 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 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -113,7 +113,6 @@ struct ConsumerTest : public ::testing::Test {
void SeedTopicWithInt(const std::string &topic_name, int value) { void SeedTopicWithInt(const std::string &topic_name, int value) {
std::array<char, sizeof(int)> int_as_char{}; std::array<char, sizeof(int)> int_as_char{};
std::memcpy(int_as_char.data(), &value, int_as_char.size()); std::memcpy(int_as_char.data(), &value, int_as_char.size());
cluster.SeedTopic(topic_name, int_as_char); cluster.SeedTopic(topic_name, int_as_char);
} }

View File

@@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd. // Copyright 2024 Memgraph Ltd.
// //
// Use of this software is governed by the Business Source License // 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 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -78,10 +78,6 @@ void KafkaClusterMock::CreateTopic(const std::string &topic_name) {
} }
} }
void KafkaClusterMock::SeedTopic(const std::string &topic_name, std::string_view message) {
SeedTopic(topic_name, std::span{message.data(), message.size()});
}
void KafkaClusterMock::SeedTopic(const std::string &topic_name, std::span<const char> message) { void KafkaClusterMock::SeedTopic(const std::string &topic_name, std::span<const char> message) {
char errstr[256] = {'\0'}; char errstr[256] = {'\0'};
std::string bootstraps_servers = Bootstraps(); std::string bootstraps_servers = Bootstraps();

View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd. // Copyright 2024 Memgraph Ltd.
// //
// Use of this software is governed by the Business Source License // 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 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -41,7 +41,6 @@ class KafkaClusterMock {
std::string Bootstraps() const; std::string Bootstraps() const;
void CreateTopic(const std::string &topic_name); void CreateTopic(const std::string &topic_name);
void SeedTopic(const std::string &topic_name, std::span<const char> message); void SeedTopic(const std::string &topic_name, std::span<const char> message);
void SeedTopic(const std::string &topic_name, std::string_view message);
private: private:
RdKafkaUniquePtr rk_{nullptr}; RdKafkaUniquePtr rk_{nullptr};