Compare commits
19 Commits
release/2.
...
fix_finali
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ed29168723 | ||
|
|
c2e9df309a | ||
|
|
82c47ee80d | ||
|
|
6a4ef55e90 | ||
|
|
98727e0fa0 | ||
|
|
9a20ac494d | ||
|
|
e302be98a2 | ||
|
|
61b9bb0f59 | ||
|
|
7ec648b4ce | ||
|
|
f098a9d5e3 | ||
|
|
bae3e8a6d3 | ||
|
|
f3574012c5 | ||
|
|
33c400fcc1 | ||
|
|
5ac938a6c9 | ||
|
|
3e3224f0a2 | ||
|
|
bfc756c092 | ||
|
|
5f2e3f01d0 | ||
|
|
2c774ff09b | ||
|
|
20b47845f0 |
1
.github/ISSUE_TEMPLATE/bug_report.md
vendored
1
.github/ISSUE_TEMPLATE/bug_report.md
vendored
@@ -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**
|
||||||
|
|||||||
60
.github/workflows/release_build_test.yaml
vendored
60
.github/workflows/release_build_test.yaml
vendored
@@ -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
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
@@ -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
|
||||||
#
|
#
|
||||||
|
|||||||
@@ -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
|
||||||
#
|
#
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
@@ -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>);
|
||||||
|
|||||||
@@ -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));
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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;
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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 &;
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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 {
|
||||||
|
|||||||
@@ -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;
|
||||||
|
|||||||
@@ -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;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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; }
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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__":
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|
||||||
|
|||||||
@@ -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():
|
||||||
|
|||||||
@@ -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
|
|
||||||
|
|||||||
@@ -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__":
|
||||||
|
|||||||
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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();
|
||||||
|
|||||||
@@ -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};
|
||||||
|
|||||||
Reference in New Issue
Block a user