tursom/memgraph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Updated telemetry client-side (#1337)

Browse Source
This commit is contained in:
andrejtonev 2023-10-16 14:16:00 +02:00 committed by GitHub
parent 7b0bafa21e
commit 22d8ef75e0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
72 changed files with 1472 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/CMakeLists.txt
View File

@ -41,7 +41,7 @@ set(mg_single_node_v2_sources
add_executable(memgraph ${mg_single_node_v2_sources})
target_include_directories(memgraph PUBLIC ${CMAKE_SOURCE_DIR}/include)
target_link_libraries(memgraph stdc++fs Threads::Threads
mg-telemetry mg-communication mg-memory mg-utils mg-license mg-settings mg-glue mg-flags)
mg-telemetry mg-communication mg-communication-metrics mg-memory mg-utils mg-license mg-settings mg-glue mg-flags)
# NOTE: `include/mg_procedure.syms` describes a pattern match for symbols which
# should be dynamically exported, so that `dlopen` can correctly link the

3
src/auth/exceptions.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -21,5 +21,6 @@ namespace memgraph::auth {
class AuthException : public utils::BasicException {
public:
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(AuthException)
};
} // namespace memgraph::auth

5
src/communication/CMakeLists.txt
View File

@ -16,8 +16,11 @@ set(communication_src_files
find_package(Boost REQUIRED)
add_library(mg-communication-metrics STATIC metrics.cpp)
target_link_libraries(mg-communication-metrics json)
add_library(mg-communication STATIC ${communication_src_files})
target_link_libraries(mg-communication Boost::headers Threads::Threads mg-utils mg-io mg-auth fmt::fmt gflags)
target_link_libraries(mg-communication Boost::headers Threads::Threads mg-utils mg-io mg-auth fmt::fmt gflags mg-communication-metrics mg-events)
find_package(OpenSSL REQUIRED)
target_link_libraries(mg-communication ${OPENSSL_LIBRARIES})

3
src/communication/bolt/client.cpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -264,4 +264,5 @@ bool Client::ReadMessageData(Marker marker, Value &ret) {
}
return false;
}
} // namespace memgraph::communication::bolt

8
src/communication/bolt/client.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -38,6 +38,7 @@ class FailureResponseException : public utils::BasicException {
: utils::BasicException{message}, code_{code} {}
const std::string &code() const { return code_; }
SPECIALIZE_GET_EXCEPTION_NAME(FailureResponseException)
private:
std::string code_;
@ -49,6 +50,7 @@ class FailureResponseException : public utils::BasicException {
class ClientQueryException : public FailureResponseException {
public:
using FailureResponseException::FailureResponseException;
SPECIALIZE_GET_EXCEPTION_NAME(ClientQueryException)
};
/// This exception is thrown whenever a fatal error occurs during query
@ -57,6 +59,7 @@ class ClientQueryException : public FailureResponseException {
class ClientFatalException : public utils::BasicException {
public:
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(ClientFatalException)
};
// Internal exception used whenever a communication error occurs. You should
@ -64,6 +67,7 @@ class ClientFatalException : public utils::BasicException {
class ServerCommunicationException : public ClientFatalException {
public:
ServerCommunicationException() : ClientFatalException("Couldn't communicate with the server!") {}
SPECIALIZE_GET_EXCEPTION_NAME(ServerCommunicationException)
};
// Internal exception used whenever a malformed data error occurs. You should
@ -71,6 +75,7 @@ class ServerCommunicationException : public ClientFatalException {
class ServerMalformedDataException : public ClientFatalException {
public:
ServerMalformedDataException() : ClientFatalException("The server sent malformed data!") {}
SPECIALIZE_GET_EXCEPTION_NAME(ServerMalformedDataException)
};
/// Structure that is used to return results from an executed query.
@ -155,4 +160,5 @@ class Client final {
ChunkedEncoderBuffer<communication::ClientOutputStream> encoder_buffer_{output_stream_};
ClientEncoder encoder_{encoder_buffer_};
};
} // namespace memgraph::communication::bolt

55
src/communication/bolt/metrics.hpp Normal file
View File

@ -0,0 +1,55 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include "communication/bolt/v1/value.hpp"
#include "communication/metrics.hpp"
namespace memgraph::communication::bolt {
template <typename TSession>
inline void RegisterNewSession(TSession &session, Value &metadata) {
auto &data = metadata.ValueMap();
session.metrics_ = bolt_metrics.Add(data.contains("user_agent") ? data["user_agent"].ValueString() : "unknown",
fmt::format("{}.{}", session.version_.major, session.version_.minor),
session.client_supported_bolt_versions_);
++session.metrics_.value()->sessions;
auto conn_type = (!data.contains("scheme") || data["scheme"].ValueString() == "none")
? BoltMetrics::ConnectionType::kAnonymous
: BoltMetrics::ConnectionType::kBasic;
++session.metrics_.value()->connection_types[(int)conn_type];
}
template <typename TSession>
inline void TouchNewSession(TSession &session, Value &metadata) {
auto &data = metadata.ValueMap();
session.metrics_ = bolt_metrics.Add(data.contains("user_agent") ? data["user_agent"].ValueString() : "unknown");
}
template <typename TSession>
inline void UpdateNewSession(TSession &session, Value &metadata) {
auto &data = metadata.ValueMap();
session.metrics_.value()->bolt_v = fmt::format("{}.{}", session.version_.major, session.version_.minor);
session.metrics_.value()->supported_bolt_v = session.client_supported_bolt_versions_;
++session.metrics_.value()->sessions;
auto conn_type = (!data.contains("scheme") || data["scheme"].ValueString() == "none")
? BoltMetrics::ConnectionType::kAnonymous
: BoltMetrics::ConnectionType::kBasic;
++session.metrics_.value()->connection_types[(int)conn_type];
}
template <typename TSession>
inline void IncrementQueryMetrics(TSession &session) {
++session.metrics_.value()->queries;
}
} // namespace memgraph::communication::bolt

4
src/communication/bolt/v1/exceptions.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -30,6 +30,7 @@ namespace memgraph::communication::bolt {
class ClientError : public utils::BasicException {
public:
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(ClientError)
};
/**
@ -67,6 +68,7 @@ class VerboseError : public utils::BasicException {
code_(fmt::format("Memgraph.{}.{}.{}", ClassificationToString(classification), category, title)) {}
const std::string &code() const noexcept { return code_; }
SPECIALIZE_GET_EXCEPTION_NAME(VerboseError)
private:
std::string ClassificationToString(Classification classification) {

4
src/communication/bolt/v1/session.hpp
View File

@ -27,6 +27,7 @@
#include "communication/bolt/v1/states/handshake.hpp"
#include "communication/bolt/v1/states/init.hpp"
#include "communication/bolt/v1/value.hpp"
#include "communication/metrics.hpp"
#include "dbms/constants.hpp"
#include "dbms/global.hpp"
#include "utils/exceptions.hpp"
@ -43,6 +44,7 @@ namespace memgraph::communication::bolt {
class SessionException : public utils::BasicException {
public:
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(SessionException)
};
/**
@ -207,6 +209,8 @@ class Session {
};
Version version_;
std::vector<std::string> client_supported_bolt_versions_;
std::optional<BoltMetrics::Metrics> metrics_;
virtual std::string GetCurrentDB() const = 0;
std::string UUID() const { return session_uuid_; }

7
src/communication/bolt/v1/states/handlers.hpp
View File

@ -18,6 +18,7 @@
#include <string_view>
#include <vector>
#include "communication/bolt/metrics.hpp"
#include "communication/bolt/v1/codes.hpp"
#include "communication/bolt/v1/constants.hpp"
#include "communication/bolt/v1/exceptions.hpp"
@ -210,6 +211,9 @@ State HandleRunV1(TSession &session, const State state, const Marker marker) {
spdlog::debug("[Run - {}] '{}'", session.GetCurrentDB(), query.ValueString());
// Increment number of queries in the metrics
IncrementQueryMetrics(session);
try {
// Interpret can throw.
const auto [header, qid] = session.Interpret(query.ValueString(), params.ValueMap(), {});
@ -274,6 +278,9 @@ State HandleRunV4(TSession &session, const State state, const Marker marker) {
spdlog::debug("[Run - {}] '{}'", session.GetCurrentDB(), query.ValueString());
// Increment number of queries in the metrics
IncrementQueryMetrics(session);
try {
// Interpret can throw.
const auto [header, qid] = session.Interpret(query.ValueString(), params.ValueMap(), extra.ValueMap());

29
src/communication/bolt/v1/states/handshake.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -24,6 +24,31 @@
namespace memgraph::communication::bolt {
inline std::vector<std::string> StringifySupportedVersions(uint8_t *data) {
std::vector<std::string> res;
uint8_t range_i = 1;
uint8_t minor_i = 2;
uint8_t major_i = 3;
uint8_t chunk_size = 4;
uint8_t n_chunks = 4;
auto stringify_version = [](uint8_t major, uint8_t minor) { return fmt::format("{}.{}", major, minor); };
for (uint8_t i = 0; i < n_chunks; ++i) {
const uint32_t full_version = *((uint32_t *)data);
if (full_version == 0) break;
if (data[1] != 0) { // Supports a range of versions
uint8_t range = data[range_i];
auto max_minor = data[minor_i];
for (uint8_t r = 0; r <= range; ++r) {
res.push_back(stringify_version(data[major_i], max_minor - r));
}
} else {
res.push_back(stringify_version(data[major_i], data[minor_i]));
}
data += chunk_size;
}
return res;
}
inline bool CopyProtocolInformationIfSupported(uint16_t version, uint8_t *protocol) {
const auto *supported_version = std::find(std::begin(kSupportedVersions), std::end(kSupportedVersions), version);
if (supported_version != std::end(kSupportedVersions)) {
@ -71,6 +96,8 @@ State StateHandshakeRun(TSession &session) {
auto dataPosition = session.input_stream_.data() + sizeof(kPreamble);
uint8_t protocol[4] = {0x00};
session.client_supported_bolt_versions_ = std::move(StringifySupportedVersions(dataPosition));
for (int i = 0; i < 4 && !protocol[3]; ++i) {
// If there is an offset defined (e.g. 0x00 0x03 0x03 0x04) the second byte
// That would enable the client to pick between 4.0 and 4.3 versions

16
src/communication/bolt/v1/states/init.hpp
View File

@ -11,6 +11,7 @@
#pragma once
#include <fmt/core.h>
#include <fmt/format.h>
#include <optional>
@ -18,6 +19,7 @@
#include "communication/bolt/v1/state.hpp"
#include "communication/bolt/v1/value.hpp"
#include "communication/exceptions.hpp"
#include "communication/metrics.hpp"
#include "spdlog/spdlog.h"
#include "utils/likely.hpp"
#include "utils/logging.hpp"
@ -201,6 +203,9 @@ State StateInitRunV1(TSession &session, const Marker marker, const Signature sig
return result.value();
}
// Register session to metrics
RegisterNewSession(session, *maybeMetadata);
return SendSuccessMessage(session);
}
@ -227,6 +232,9 @@ State StateInitRunV4(TSession &session, Marker marker, Signature signature) {
return result.value();
}
// Register session to metrics
RegisterNewSession(session, *maybeMetadata);
return SendSuccessMessage(session);
}
@ -247,6 +255,10 @@ State StateInitRunV5(TSession &session, Marker marker, Signature signature) {
if (SendSuccessMessage(session) == State::Close) {
return State::Close;
}
// Register session to metrics
TouchNewSession(session, *maybeMetadata);
// Stay in Init
return State::Init;
}
@ -274,6 +286,10 @@ State StateInitRunV5(TSession &session, Marker marker, Signature signature) {
if (SendSuccessMessage(session) == State::Close) {
return State::Close;
}
// Register session to metrics
UpdateNewSession(session, *maybeMetadata);
return State::Idle;
}

1
src/communication/bolt/v1/value.hpp
View File

@ -276,6 +276,7 @@ class ValueException : public utils::BasicException {
public:
using utils::BasicException::BasicException;
ValueException() : BasicException("Incompatible template param and type!") {}
SPECIALIZE_GET_EXCEPTION_NAME(ValueException)
};
/**

3
src/communication/exceptions.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -21,5 +21,6 @@ namespace memgraph::communication {
*/
class SessionClosedException : public utils::BasicException {
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(SessionClosedException)
};
} // namespace memgraph::communication

42
src/communication/metrics.cpp Normal file
View File

@ -0,0 +1,42 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include "metrics.hpp"
namespace {
constexpr auto kName = "name";
constexpr auto kSupportedBoltVersions = "supported_bolt_versions";
constexpr auto kBoltVersion = "bolt_version";
constexpr auto kConnectionTypes = "connection_types";
constexpr auto kSessions = "sessions";
constexpr auto kQueries = "queries";
} // namespace
namespace memgraph::communication {
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
BoltMetrics bolt_metrics;
nlohmann::json BoltMetrics::Info::ToJson() const {
nlohmann::json res;
res[kName] = name;
res[kSupportedBoltVersions] = nlohmann::json::array();
for (const auto &sbv : supported_bolt_v) {
res[kSupportedBoltVersions].push_back(sbv);
}
res[kBoltVersion] = bolt_v;
res[kConnectionTypes] = {{ConnectionTypeStr((ConnectionType)0), connection_types[0].load()},
{ConnectionTypeStr((ConnectionType)1), connection_types[1].load()}};
res[kSessions] = sessions.load();
res[kQueries] = queries.load();
return res;
}
} // namespace memgraph::communication

116
src/communication/metrics.hpp Normal file
View File

@ -0,0 +1,116 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include <atomic>
#include <map>
#include <mutex>
#include <shared_mutex>
#include <string>
#include <vector>
#include <json/json.hpp>
namespace memgraph::communication {
class BoltMetrics {
public:
enum class ConnectionType { kAnonymous = 0, kBasic, Count };
static constexpr std::array<std::string_view, (int)ConnectionType::Count> ct_to_str = {"anonymous", "basic"};
static std::string ConnectionTypeStr(ConnectionType type) { return std::string(ct_to_str[(int)type]); }
class Metrics;
struct Info {
explicit Info(std::string name) : name(std::move(name)) {}
Info(std::string name, std::string bolt_v, std::vector<std::string> supported_bolt_v)
: name(std::move(name)), bolt_v(std::move(bolt_v)), supported_bolt_v(std::move(supported_bolt_v)) {}
const std::string name; //!< Driver name
std::string bolt_v; //!< Bolt version used
std::vector<std::string> supported_bolt_v; //!< Bolt versions supported by the driver
std::atomic_int connection_types[(int)ConnectionType::Count]; //!< Authentication used when connecting
std::atomic_int sessions; //!< Number of sessions using the same driver
std::atomic_int queries; //!< Queries executed by the driver
nlohmann::json ToJson() const;
private:
friend class Metrics;
std::atomic_int concurrent_users;
};
class Metrics {
friend class BoltMetrics;
explicit Metrics(Info &info) : info_(&info) { ++info_->concurrent_users; }
public:
~Metrics() {
if (info_) --info_->concurrent_users;
}
Metrics(const Metrics &other) : info_(other.info_) { ++info_->concurrent_users; }
Metrics(Metrics &&other) noexcept : info_(other.info_) { other.info_ = nullptr; }
Metrics &operator=(const Metrics &other) {
if (this != &other) {
if (info_) --info_->concurrent_users;
info_ = other.info_;
if (info_) ++info_->concurrent_users;
}
return *this;
}
Metrics &operator=(Metrics &&other) noexcept {
if (this != &other) {
if (info_) --info_->concurrent_users;
info_ = other.info_;
other.info_ = nullptr; // Invalidate the source object
}
return *this;
}
Info *operator->() { return info_; }
Info *info_;
};
Metrics Add(std::string name) {
std::unique_lock<std::mutex> l(mtx);
auto key = name;
auto [it, _] = info.emplace(std::piecewise_construct, std::forward_as_tuple(std::move(key)),
std::forward_as_tuple(std::move(name)));
return Metrics(it->second);
}
Metrics Add(std::string name, std::string bolt_v, std::vector<std::string> supported_bolt_v) {
std::unique_lock<std::mutex> l(mtx);
auto key = name;
auto [it, _] = info.emplace(std::piecewise_construct, std::forward_as_tuple(std::move(key)),
std::forward_as_tuple(std::move(name), std::move(bolt_v), std::move(supported_bolt_v)));
return Metrics(it->second);
}
nlohmann::json ToJson() {
std::unique_lock<std::mutex> l(mtx);
auto res = nlohmann::json::array();
for (const auto &[_, client_info] : info) {
res.push_back(client_info.ToJson());
}
return res;
}
private:
mutable std::mutex mtx;
std::map<std::string, Info> info;
};
extern BoltMetrics bolt_metrics;
} // namespace memgraph::communication

1
src/csv/include/csv/parsing.hpp
View File

@ -35,6 +35,7 @@ namespace memgraph::csv {
class CsvReadException : public utils::BasicException {
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(CsvReadException)
};
class FileCsvSource {

19
src/dbms/database.hpp
View File

@ -27,6 +27,14 @@
namespace memgraph::dbms {
struct DatabaseInfo {
storage::StorageInfo storage_info;
uint64_t triggers;
uint64_t streams;
};
static inline nlohmann::json ToJson(const DatabaseInfo &info) { return ToJson(info.storage_info); }
/**
* @brief Class containing everything associated with a single Database
*
@ -89,9 +97,16 @@ class Database {
/**
* @brief Get the storage info
*
* @return storage::StorageInfo
* @param force_directory Use the configured directory, do not try to decipher the multi-db version
* @return DatabaseInfo
*/
storage::StorageInfo GetInfo() const { return storage_->GetInfo(); }
DatabaseInfo GetInfo(bool force_directory = false) const {
DatabaseInfo info;
info.storage_info = storage_->GetInfo(force_directory);
info.triggers = trigger_store_.GetTriggerInfo().size();
info.streams = streams_.GetStreamInfo().size();
return info;
}
/**
* @brief Switch storage to OnDisk

89
src/dbms/dbms_handler.hpp
View File

@ -25,6 +25,7 @@
#include "auth/auth.hpp"
#include "constants.hpp"
#include "dbms/database.hpp"
#include "dbms/database_handler.hpp"
#include "global.hpp"
#include "query/config.hpp"
@ -32,6 +33,7 @@
#include "spdlog/spdlog.h"
#include "storage/v2/durability/durability.hpp"
#include "storage/v2/durability/paths.hpp"
#include "storage/v2/isolation_level.hpp"
#include "utils/exceptions.hpp"
#include "utils/file.hpp"
#include "utils/logging.hpp"
@ -42,6 +44,43 @@
namespace memgraph::dbms {
struct Statistics {
uint64_t num_vertex; //!< Sum of vertexes in every database
uint64_t num_edges; //!< Sum of edges in every database
uint64_t triggers; //!< Sum of triggers in every database
uint64_t streams; //!< Sum of streams in every database
uint64_t users; //!< Number of defined users
uint64_t num_databases; //!< Number of isolated databases
uint64_t indices; //!< Sum of indices in every database
uint64_t constraints; //!< Sum of constraints in every database
uint64_t storage_modes[3]; //!< Number of databases in each storage mode [IN_MEM_TX, IN_MEM_ANA, ON_DISK_TX]
uint64_t isolation_levels[3]; //!< Number of databases in each isolation level [SNAPSHOT, READ_COMM, READ_UNC]
uint64_t snapshot_enabled; //!< Number of databases with snapshots enabled
uint64_t wal_enabled; //!< Number of databases with WAL enabled
};
static inline nlohmann::json ToJson(const Statistics &stats) {
nlohmann::json res;
res["edges"] = stats.num_edges;
res["vertices"] = stats.num_vertex;
res["triggers"] = stats.triggers;
res["streams"] = stats.streams;
res["users"] = stats.users;
res["databases"] = stats.num_databases;
res["indices"] = stats.indices;
res["constraints"] = stats.constraints;
res["storage_modes"] = {{storage::StorageModeToString((storage::StorageMode)0), stats.storage_modes[0]},
{storage::StorageModeToString((storage::StorageMode)1), stats.storage_modes[1]},
{storage::StorageModeToString((storage::StorageMode)2), stats.storage_modes[2]}};
res["isolation_levels"] = {{storage::IsolationLevelToString((storage::IsolationLevel)0), stats.isolation_levels[0]},
{storage::IsolationLevelToString((storage::IsolationLevel)1), stats.isolation_levels[1]},
{storage::IsolationLevelToString((storage::IsolationLevel)2), stats.isolation_levels[2]}};
res["durability"] = {{"snapshot_enabled", stats.snapshot_enabled}, {"WAL_enabled", stats.wal_enabled}};
return res;
}
#ifdef MG_ENTERPRISE
using DeleteResult = utils::BasicResult<DeleteError>;
@ -54,12 +93,6 @@ class DbmsHandler {
using LockT = utils::RWLock;
using NewResultT = utils::BasicResult<NewError, DatabaseAccess>;
struct Statistics {
uint64_t num_vertex; //!< Sum of vertexes in every database
uint64_t num_edges; //!< Sum of edges in every database
uint64_t num_databases; //! number of isolated databases
};
/**
* @brief Initialize the handler.
*
@ -180,25 +213,51 @@ class DbmsHandler {
}
/**
* @brief Return the number of vertex across all databases.
* @brief Return the statistics all databases.
*
* @return uint64_t
* @return Statistics
*/
Statistics Info() {
Statistics Stats() {
Statistics stats{};
// TODO: Handle overflow?
uint64_t nv = 0;
uint64_t ne = 0;
std::shared_lock<LockT> rd(lock_);
const uint64_t ndb = std::distance(db_handler_.cbegin(), db_handler_.cend());
for (auto &[_, db_gk] : db_handler_) {
auto db_acc_opt = db_gk.access();
if (!db_acc_opt) continue;
auto &db_acc = *db_acc_opt;
const auto &info = db_acc->GetInfo();
nv += info.vertex_count;
ne += info.edge_count;
const auto &storage_info = info.storage_info;
stats.num_vertex += storage_info.vertex_count;
stats.num_edges += storage_info.edge_count;
stats.triggers += info.triggers;
stats.streams += info.streams;
++stats.num_databases;
stats.indices += storage_info.label_indices + storage_info.label_property_indices;
stats.constraints += storage_info.existence_constraints + storage_info.unique_constraints;
++stats.storage_modes[(int)storage_info.storage_mode];
++stats.isolation_levels[(int)storage_info.isolation_level];
stats.snapshot_enabled += storage_info.durability_snapshot_enabled;
stats.wal_enabled += storage_info.durability_wal_enabled;
}
return {nv, ne, ndb};
return stats;
}
/**
* @brief Return a vector with all database info.
*
* @return std::vector<DatabaseInfo>
*/
std::vector<DatabaseInfo> Info() {
std::vector<DatabaseInfo> res;
res.reserve(std::distance(db_handler_.cbegin(), db_handler_.cend()));
std::shared_lock<LockT> rd(lock_);
for (auto &[_, db_gk] : db_handler_) {
auto db_acc_opt = db_gk.access();
if (!db_acc_opt) continue;
auto &db_acc = *db_acc_opt;
res.push_back(db_acc->GetInfo());
}
return res;
}
/**

2
src/dbms/global.hpp
View File

@ -48,6 +48,7 @@ enum class SetForResult : uint8_t {
class UnknownSessionException : public utils::BasicException {
public:
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(UnknownSessionException)
};
/**
@ -58,6 +59,7 @@ class UnknownSessionException : public utils::BasicException {
class UnknownDatabaseException : public utils::BasicException {
public:
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(UnknownDatabaseException)
};
} // namespace memgraph::dbms

9
src/glue/SessionHL.cpp
View File

@ -22,6 +22,7 @@
#include "license/license.hpp"
#include "query/discard_value_stream.hpp"
#include "query/interpreter_context.hpp"
#include "utils/event_map.hpp"
#include "utils/spin_lock.hpp"
namespace memgraph::metrics {
@ -155,6 +156,8 @@ std::map<std::string, memgraph::communication::bolt::Value> SessionHL::Discard(s
memgraph::query::DiscardValueResultStream stream;
return DecodeSummary(interpreter_.Pull(&stream, n, qid));
} catch (const memgraph::query::QueryException &e) {
// Count the number of specific exceptions thrown
metrics::IncrementCounter(GetExceptionName(e));
// Wrap QueryException into ClientError, because we want to allow the
// client to fix their query.
throw memgraph::communication::bolt::ClientError(e.what());
@ -169,6 +172,8 @@ std::map<std::string, memgraph::communication::bolt::Value> SessionHL::Pull(Sess
TypedValueResultStream<TEncoder> stream(encoder, db->storage());
return DecodeSummary(interpreter_.Pull(&stream, n, qid));
} catch (const memgraph::query::QueryException &e) {
// Count the number of specific exceptions thrown
metrics::IncrementCounter(GetExceptionName(e));
// Wrap QueryException into ClientError, because we want to allow the
// client to fix their query.
throw memgraph::communication::bolt::ClientError(e.what());
@ -212,10 +217,14 @@ std::pair<std::vector<std::string>, std::optional<int>> SessionHL::Interpret(
return {std::move(result.headers), result.qid};
} catch (const memgraph::query::QueryException &e) {
// Count the number of specific exceptions thrown
metrics::IncrementCounter(GetExceptionName(e));
// Wrap QueryException into ClientError, because we want to allow the
// client to fix their query.
throw memgraph::communication::bolt::ClientError(e.what());
} catch (const memgraph::query::ReplicationException &e) {
// Count the number of specific exceptions thrown
metrics::IncrementCounter(GetExceptionName(e));
throw memgraph::communication::bolt::ClientError(e.what());
}
}

2
src/http_handlers/CMakeLists.txt
View File

@ -1,4 +1,4 @@
set(mg_http_handlers_sources)
add_library(mg-http-handlers STATIC ${mg_http_handlers_sources})
target_link_libraries(mg-http-handlers mg-query mg-storage-v2)
target_link_libraries(mg-http-handlers mg-query mg-storage-v2 mg-events)

4
src/http_handlers/metrics.hpp
View File

@ -57,10 +57,10 @@ class MetricsService {
}
private:
const storage::Storage *db_;
storage::Storage *const db_;
MetricsResponse GetMetrics() {
auto info = db_->GetInfo();
auto info = db_->GetBaseInfo();
return MetricsResponse{.vertex_count = info.vertex_count,
.edge_count = info.edge_count,

12
src/integrations/kafka/exceptions.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -18,12 +18,14 @@
namespace memgraph::integrations::kafka {
class KafkaStreamException : public utils::BasicException {
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(KafkaStreamException)
};
class ConsumerFailedToInitializeException : public KafkaStreamException {
public:
ConsumerFailedToInitializeException(const std::string_view consumer_name, const std::string_view error)
: KafkaStreamException("Failed to initialize Kafka consumer {} : {}", consumer_name, error) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerFailedToInitializeException)
};
class SettingCustomConfigFailed : public ConsumerFailedToInitializeException {
@ -33,47 +35,55 @@ class SettingCustomConfigFailed : public ConsumerFailedToInitializeException {
: ConsumerFailedToInitializeException(
consumer_name,
fmt::format(R"(failed to set custom config ("{}": "{}"), because of error {})", key, value, error)) {}
SPECIALIZE_GET_EXCEPTION_NAME(SettingCustomConfigFailed)
};
class ConsumerRunningException : public KafkaStreamException {
public:
explicit ConsumerRunningException(const std::string_view consumer_name)
: KafkaStreamException("Kafka consumer {} is already running", consumer_name) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerRunningException)
};
class ConsumerStoppedException : public KafkaStreamException {
public:
explicit ConsumerStoppedException(const std::string_view consumer_name)
: KafkaStreamException("Kafka consumer {} is already stopped", consumer_name) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerStoppedException)
};
class ConsumerCheckFailedException : public KafkaStreamException {
public:
explicit ConsumerCheckFailedException(const std::string_view consumer_name, const std::string_view error)
: KafkaStreamException("Kafka consumer {} check failed: {}", consumer_name, error) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerCheckFailedException)
};
class ConsumerStartFailedException : public KafkaStreamException {
public:
explicit ConsumerStartFailedException(const std::string_view consumer_name, const std::string_view error)
: KafkaStreamException("Starting Kafka consumer {} failed: {}", consumer_name, error) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerStartFailedException)
};
class TopicNotFoundException : public KafkaStreamException {
public:
TopicNotFoundException(const std::string_view consumer_name, const std::string_view topic_name)
: KafkaStreamException("Kafka consumer {} cannot find topic {}", consumer_name, topic_name) {}
SPECIALIZE_GET_EXCEPTION_NAME(TopicNotFoundException)
};
class ConsumerCommitFailedException : public KafkaStreamException {
public:
ConsumerCommitFailedException(const std::string_view consumer_name, const std::string_view error)
: KafkaStreamException("Committing offset of consumer {} failed: {}", consumer_name, error) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerCommitFailedException)
};
class ConsumerReadMessagesFailedException : public KafkaStreamException {
public:
ConsumerReadMessagesFailedException(const std::string_view consumer_name, const std::string_view error)
: KafkaStreamException("Error happened in consumer {} while fetching messages: {}", consumer_name, error) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerReadMessagesFailedException)
};
} // namespace memgraph::integrations::kafka

11
src/integrations/pulsar/exceptions.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -18,53 +18,62 @@
namespace memgraph::integrations::pulsar {
class PulsarStreamException : public utils::BasicException {
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(PulsarStreamException)
};
class ConsumerFailedToInitializeException : public PulsarStreamException {
public:
ConsumerFailedToInitializeException(const std::string &consumer_name, const std::string &error)
: PulsarStreamException("Failed to initialize Pulsar consumer {} : {}", consumer_name, error) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerFailedToInitializeException)
};
class ConsumerRunningException : public PulsarStreamException {
public:
explicit ConsumerRunningException(const std::string &consumer_name)
: PulsarStreamException("Pulsar consumer {} is already running", consumer_name) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerRunningException)
};
class ConsumerStoppedException : public PulsarStreamException {
public:
explicit ConsumerStoppedException(const std::string &consumer_name)
: PulsarStreamException("Pulsar consumer {} is already stopped", consumer_name) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerStoppedException)
};
class ConsumerCheckFailedException : public PulsarStreamException {
public:
explicit ConsumerCheckFailedException(const std::string &consumer_name, const std::string &error)
: PulsarStreamException("Pulsar consumer {} check failed: {}", consumer_name, error) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerCheckFailedException)
};
class ConsumerStartFailedException : public PulsarStreamException {
public:
explicit ConsumerStartFailedException(const std::string &consumer_name, const std::string &error)
: PulsarStreamException("Starting Pulsar consumer {} failed: {}", consumer_name, error) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerStartFailedException)
};
class TopicNotFoundException : public PulsarStreamException {
public:
TopicNotFoundException(const std::string &consumer_name, const std::string &topic_name)
: PulsarStreamException("Pulsar consumer {} cannot find topic {}", consumer_name, topic_name) {}
SPECIALIZE_GET_EXCEPTION_NAME(TopicNotFoundException)
};
class ConsumerReadMessagesFailedException : public PulsarStreamException {
public:
ConsumerReadMessagesFailedException(const std::string_view consumer_name, const std::string_view error)
: PulsarStreamException("Error happened in consumer {} while fetching messages: {}", consumer_name, error) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerReadMessagesFailedException)
};
class ConsumerAcknowledgeMessagesFailedException : public PulsarStreamException {
public:
explicit ConsumerAcknowledgeMessagesFailedException(const std::string_view consumer_name)
: PulsarStreamException("Acknowledging a message of consumer {} has failed!", consumer_name) {}
SPECIALIZE_GET_EXCEPTION_NAME(ConsumerAcknowledgeMessagesFailedException)
};
} // namespace memgraph::integrations::pulsar

1
src/kvstore/kvstore.hpp
View File

@ -25,6 +25,7 @@ namespace memgraph::kvstore {
class KVStoreError : public utils::BasicException {
public:
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(KVStoreError)
};
/**

34
src/memgraph.cpp
View File

@ -10,6 +10,7 @@
// licenses/APL.txt.
#include "audit/log.hpp"
#include "communication/metrics.hpp"
#include "communication/websocket/auth.hpp"
#include "communication/websocket/server.hpp"
#include "dbms/constants.hpp"
@ -31,6 +32,7 @@
#include "requests/requests.hpp"
#include "telemetry/telemetry.hpp"
#include "utils/signals.hpp"
#include "utils/skip_list.hpp"
#include "utils/sysinfo/memory.hpp"
#include "utils/system_info.hpp"
#include "utils/terminate_handler.hpp"
@ -428,31 +430,19 @@ int main(int argc, char **argv) {
std::optional<memgraph::telemetry::Telemetry> telemetry;
if (FLAGS_telemetry_enabled) {
telemetry.emplace(telemetry_server, data_directory / "telemetry", memgraph::glue::run_id_, machine_id,
std::chrono::minutes(10));
service_name == "BoltS", FLAGS_data_directory, std::chrono::minutes(10));
#ifdef MG_ENTERPRISE
telemetry->AddCollector("storage", [&new_handler]() -> nlohmann::json {
const auto &info = new_handler.Info();
return {{"vertices", info.num_vertex}, {"edges", info.num_edges}, {"databases", info.num_databases}};
});
telemetry->AddStorageCollector(new_handler, auth_);
telemetry->AddDatabaseCollector(new_handler);
#else
telemetry->AddCollector("storage", [gk = &db_gatekeeper]() -> nlohmann::json {
auto db_acc = gk->access();
MG_ASSERT(db_acc, "Failed to get access to the default database");
auto info = db_acc->get()->GetInfo();
return {{"vertices", info.vertex_count}, {"edges", info.edge_count}};
});
telemetry->AddStorageCollector(db_gatekeeper, auth_);
telemetry->AddDatabaseCollector();
#endif
telemetry->AddCollector("event_counters", []() -> nlohmann::json {
nlohmann::json ret;
for (size_t i = 0; i < memgraph::metrics::CounterEnd(); ++i) {
ret[memgraph::metrics::GetCounterName(i)] =
memgraph::metrics::global_counters[i].load(std::memory_order_relaxed);
}
return ret;
});
telemetry->AddCollector("query_module_counters", []() -> nlohmann::json {
return memgraph::query::plan::CallProcedure::GetAndResetCounters();
});
telemetry->AddClientCollector();
telemetry->AddEventsCollector();
telemetry->AddQueryModuleCollector();
telemetry->AddExceptionCollector();
telemetry->AddReplicationCollector();
}
memgraph::license::LicenseInfoSender license_info_sender(telemetry_server, memgraph::glue::run_id_, machine_id,
memory_limit,

1
src/mg_import_csv.cpp
View File

@ -162,6 +162,7 @@ struct hash<NodeId> {
class LoadException : public memgraph::utils::BasicException {
public:
using memgraph::utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(LoadException)
};
enum class CsvParserState {

3
src/query/CMakeLists.txt
View File

@ -55,7 +55,8 @@ target_link_libraries(mg-query PUBLIC dl
mg-memory
mg::csv
mg-flags
mg-dbms)
mg-dbms
mg-events)
if(NOT "${MG_PYTHON_PATH}" STREQUAL "")
set(Python3_ROOT_DIR "${MG_PYTHON_PATH}")
endif()

47
src/query/exceptions.hpp
View File

@ -14,6 +14,7 @@
#include "utils/exceptions.hpp"
#include <fmt/format.h>
#include <exception>
namespace memgraph::query {
@ -25,18 +26,21 @@ namespace memgraph::query {
*/
class QueryException : public utils::BasicException {
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(QueryException)
};
class LexingException : public QueryException {
public:
using QueryException::QueryException;
LexingException() : QueryException("") {}
SPECIALIZE_GET_EXCEPTION_NAME(LexingException)
};
class SyntaxException : public QueryException {
public:
using QueryException::QueryException;
SyntaxException() : QueryException("") {}
SPECIALIZE_GET_EXCEPTION_NAME(SyntaxException)
};
// TODO: Figure out what information to put in exception.
@ -52,16 +56,19 @@ class SemanticException : public QueryException {
public:
using QueryException::QueryException;
SemanticException() : QueryException("") {}
SPECIALIZE_GET_EXCEPTION_NAME(SemanticException)
};
class UnboundVariableError : public SemanticException {
public:
explicit UnboundVariableError(const std::string &name) : SemanticException("Unbound variable: " + name + ".") {}
SPECIALIZE_GET_EXCEPTION_NAME(UnboundVariableError)
};
class RedeclareVariableError : public SemanticException {
public:
explicit RedeclareVariableError(const std::string &name) : SemanticException("Redeclaring variable: " + name + ".") {}
SPECIALIZE_GET_EXCEPTION_NAME(RedeclareVariableError)
};
class TypeMismatchError : public SemanticException {
@ -69,23 +76,27 @@ class TypeMismatchError : public SemanticException {
TypeMismatchError(const std::string &name, const std::string &datum, const std::string &expected)
: SemanticException(fmt::format("Type mismatch: {} already defined as {}, expected {}.", name, datum, expected)) {
}
SPECIALIZE_GET_EXCEPTION_NAME(TypeMismatchError)
};
class UnprovidedParameterError : public QueryException {
public:
using QueryException::QueryException;
SPECIALIZE_GET_EXCEPTION_NAME(UnprovidedParameterError)
};
class ProfileInMulticommandTxException : public QueryException {
public:
using QueryException::QueryException;
ProfileInMulticommandTxException() : QueryException("PROFILE not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(ProfileInMulticommandTxException)
};
class IndexInMulticommandTxException : public QueryException {
public:
using QueryException::QueryException;
IndexInMulticommandTxException() : QueryException("Index manipulation not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(IndexInMulticommandTxException)
};
class ConstraintInMulticommandTxException : public QueryException {
@ -95,12 +106,14 @@ class ConstraintInMulticommandTxException : public QueryException {
: QueryException(
"Constraint manipulation not allowed in multicommand "
"transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(ConstraintInMulticommandTxException)
};
class InfoInMulticommandTxException : public QueryException {
public:
using QueryException::QueryException;
InfoInMulticommandTxException() : QueryException("Info reporting not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(InfoInMulticommandTxException)
};
/**
@ -110,6 +123,7 @@ class InfoInMulticommandTxException : public QueryException {
class QueryRuntimeException : public QueryException {
public:
using QueryException::QueryException;
SPECIALIZE_GET_EXCEPTION_NAME(QueryRuntimeException)
};
enum class AbortReason : uint8_t {
@ -132,6 +146,7 @@ class HintedAbortError : public utils::BasicException {
public:
using utils::BasicException::BasicException;
explicit HintedAbortError(AbortReason reason) : utils::BasicException(AsMsg(reason)), reason_{reason} {}
SPECIALIZE_GET_EXCEPTION_NAME(HintedAbortError)
auto Reason() const -> AbortReason { return reason_; }
@ -156,17 +171,20 @@ class HintedAbortError : public utils::BasicException {
class ExplicitTransactionUsageException : public QueryRuntimeException {
public:
using QueryRuntimeException::QueryRuntimeException;
SPECIALIZE_GET_EXCEPTION_NAME(ExplicitTransactionUsageException)
};
class DatabaseContextRequiredException : public QueryRuntimeException {
public:
using QueryRuntimeException::QueryRuntimeException;
SPECIALIZE_GET_EXCEPTION_NAME(DatabaseContextRequiredException)
};
class WriteVertexOperationInEdgeImportModeException : public QueryException {
public:
WriteVertexOperationInEdgeImportModeException()
: QueryException("Write operations on vertices are forbidden while the edge import mode is active.") {}
SPECIALIZE_GET_EXCEPTION_NAME(WriteVertexOperationInEdgeImportModeException)
};
class TransactionSerializationException : public QueryException {
@ -176,6 +194,7 @@ class TransactionSerializationException : public QueryException {
: QueryException(
"Cannot resolve conflicting transactions. You can retry this transaction when the conflicting transaction "
"is finished") {}
SPECIALIZE_GET_EXCEPTION_NAME(TransactionSerializationException)
};
class ReconstructionException : public QueryException {
@ -184,6 +203,7 @@ class ReconstructionException : public QueryException {
: QueryException(
"Record invalid after WITH clause. Most likely deleted by a "
"preceeding DELETE.") {}
SPECIALIZE_GET_EXCEPTION_NAME(ReconstructionException)
};
class RemoveAttachedVertexException : public QueryRuntimeException {
@ -192,76 +212,89 @@ class RemoveAttachedVertexException : public QueryRuntimeException {
: QueryRuntimeException(
"Failed to remove node because of it's existing "
"connections. Consider using DETACH DELETE.") {}
SPECIALIZE_GET_EXCEPTION_NAME(RemoveAttachedVertexException)
};
class UserModificationInMulticommandTxException : public QueryException {
public:
UserModificationInMulticommandTxException()
: QueryException("Authentication clause not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(UserModificationInMulticommandTxException)
};
class InvalidArgumentsException : public QueryException {
public:
InvalidArgumentsException(const std::string &argument_name, const std::string &message)
: QueryException(fmt::format("Invalid arguments sent: {} - {}", argument_name, message)) {}
SPECIALIZE_GET_EXCEPTION_NAME(InvalidArgumentsException)
};
class ReplicationModificationInMulticommandTxException : public QueryException {
public:
ReplicationModificationInMulticommandTxException()
: QueryException("Replication clause not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(ReplicationModificationInMulticommandTxException)
};
class ReplicationDisabledOnDiskStorage : public QueryException {
public:
ReplicationDisabledOnDiskStorage() : QueryException("Replication is not supported while in on-disk storage mode.") {}
SPECIALIZE_GET_EXCEPTION_NAME(ReplicationDisabledOnDiskStorage)
};
class LockPathModificationInMulticommandTxException : public QueryException {
public:
LockPathModificationInMulticommandTxException()
: QueryException("Lock path query not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(LockPathModificationInMulticommandTxException)
};
class LockPathDisabledOnDiskStorage : public QueryException {
public:
LockPathDisabledOnDiskStorage()
: QueryException("Lock path disabled on disk storage since all data is already persisted. ") {}
SPECIALIZE_GET_EXCEPTION_NAME(LockPathDisabledOnDiskStorage)
};
class FreeMemoryModificationInMulticommandTxException : public QueryException {
public:
FreeMemoryModificationInMulticommandTxException()
: QueryException("Free memory query not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(FreeMemoryModificationInMulticommandTxException)
};
class FreeMemoryDisabledOnDiskStorage : public QueryException {
public:
FreeMemoryDisabledOnDiskStorage() : QueryException("Free memory does nothing when using disk storage. ") {}
SPECIALIZE_GET_EXCEPTION_NAME(FreeMemoryDisabledOnDiskStorage)
};
class ShowConfigModificationInMulticommandTxException : public QueryException {
public:
ShowConfigModificationInMulticommandTxException()
: QueryException("Show config query not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(ShowConfigModificationInMulticommandTxException)
};
class TriggerModificationInMulticommandTxException : public QueryException {
public:
TriggerModificationInMulticommandTxException()
: QueryException("Trigger queries not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(ShowConfigModificationInMulticommandTxException)
};
class StreamQueryInMulticommandTxException : public QueryException {
public:
StreamQueryInMulticommandTxException()
: QueryException("Stream queries are not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(StreamQueryInMulticommandTxException)
};
class IsolationLevelModificationInMulticommandTxException : public QueryException {
public:
IsolationLevelModificationInMulticommandTxException()
: QueryException("Isolation level cannot be modified in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(IsolationLevelModificationInMulticommandTxException)
};
class IsolationLevelModificationInAnalyticsException : public QueryException {
@ -271,53 +304,62 @@ class IsolationLevelModificationInAnalyticsException : public QueryException {
"Isolation level cannot be modified when storage mode is set to IN_MEMORY_ANALYTICAL."
"IN_MEMORY_ANALYTICAL mode doesn't provide any isolation guarantees, "
"you can think about it as an equivalent to READ_UNCOMMITED.") {}
SPECIALIZE_GET_EXCEPTION_NAME(IsolationLevelModificationInAnalyticsException)
};
class StorageModeModificationInMulticommandTxException : public QueryException {
public:
StorageModeModificationInMulticommandTxException()
: QueryException("Storage mode cannot be modified in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(StorageModeModificationInMulticommandTxException)
};
class EdgeImportModeModificationInMulticommandTxException : public QueryException {
public:
EdgeImportModeModificationInMulticommandTxException()
: QueryException("Edge import mode cannot be modified in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(EdgeImportModeModificationInMulticommandTxException)
};
class CreateSnapshotInMulticommandTxException final : public QueryException {
public:
CreateSnapshotInMulticommandTxException()
: QueryException("Snapshot cannot be created in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(CreateSnapshotInMulticommandTxException)
};
class CreateSnapshotDisabledOnDiskStorage final : public QueryException {
public:
CreateSnapshotDisabledOnDiskStorage() : QueryException("In the on-disk storage mode data is already persistent.") {}
SPECIALIZE_GET_EXCEPTION_NAME(CreateSnapshotDisabledOnDiskStorage)
};
class EdgeImportModeQueryDisabledOnDiskStorage final : public QueryException {
public:
EdgeImportModeQueryDisabledOnDiskStorage()
: QueryException("Edge import mode is only allowed for on-disk storage mode.") {}
SPECIALIZE_GET_EXCEPTION_NAME(EdgeImportModeQueryDisabledOnDiskStorage)
};
class SettingConfigInMulticommandTxException final : public QueryException {
public:
SettingConfigInMulticommandTxException()
: QueryException("Settings cannot be changed or fetched in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(SettingConfigInMulticommandTxException)
};
class VersionInfoInMulticommandTxException : public QueryException {
public:
VersionInfoInMulticommandTxException()
: QueryException("Version info query not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(VersionInfoInMulticommandTxException)
};
class AnalyzeGraphInMulticommandTxException : public QueryException {
public:
AnalyzeGraphInMulticommandTxException()
: QueryException("Analyze graph query not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(AnalyzeGraphInMulticommandTxException)
};
class ReplicationException : public utils::BasicException {
@ -326,28 +368,33 @@ class ReplicationException : public utils::BasicException {
explicit ReplicationException(const std::string &message)
: utils::BasicException("Replication Exception: {} Check the status of the replicas using 'SHOW REPLICAS' query.",
message) {}
SPECIALIZE_GET_EXCEPTION_NAME(ReplicationException)
};
class TransactionQueueInMulticommandTxException : public QueryException {
public:
TransactionQueueInMulticommandTxException()
: QueryException("Transaction queue queries not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(TransactionQueueInMulticommandTxException)
};
class IndexPersistenceException : public QueryException {
public:
IndexPersistenceException() : QueryException("Persisting index on disk failed.") {}
SPECIALIZE_GET_EXCEPTION_NAME(IndexPersistenceException)
};
class ConstraintsPersistenceException : public QueryException {
public:
ConstraintsPersistenceException() : QueryException("Persisting constraints on disk failed.") {}
SPECIALIZE_GET_EXCEPTION_NAME(ConstraintsPersistenceException)
};
class MultiDatabaseQueryInMulticommandTxException : public QueryException {
public:
MultiDatabaseQueryInMulticommandTxException()
: QueryException("Multi-database queries are not allowed in multicommand transactions.") {}
SPECIALIZE_GET_EXCEPTION_NAME(MultiDatabaseQueryInMulticommandTxException)
};
} // namespace memgraph::query

15
src/query/interpreter.cpp
View File

@ -62,6 +62,7 @@
#include "query/procedure/module.hpp"
#include "query/stream.hpp"
#include "query/stream/common.hpp"
#include "query/stream/sources.hpp"
#include "query/stream/streams.hpp"
#include "query/trigger.hpp"
#include "query/typed_value.hpp"
@ -2679,7 +2680,7 @@ Callback SwitchMemoryDevice(storage::StorageMode current_mode, storage::StorageM
}
std::unique_lock main_guard{in.storage()->main_lock_}; // do we need this?
if (auto vertex_cnt_approx = in.storage()->GetInfo().vertex_count; vertex_cnt_approx > 0) {
if (auto vertex_cnt_approx = in.storage()->GetBaseInfo().vertex_count; vertex_cnt_approx > 0) {
throw utils::BasicException(
"You cannot switch from an in-memory storage mode to the on-disk storage mode when the database "
"contains data. Delete all entries from the database, run FREE MEMORY and then repeat this "
@ -3077,7 +3078,7 @@ PreparedQuery PrepareSystemInfoQuery(ParsedQuery parsed_query, bool in_explicit_
header = {"storage info", "value"};
handler = [storage = current_db.db_acc_->get()->storage(), interpreter_isolation_level,
next_transaction_isolation_level] {
auto info = storage->GetInfo();
auto info = storage->GetBaseInfo();
std::vector<std::vector<TypedValue>> results{
{TypedValue("name"), TypedValue(storage->id())},
{TypedValue("vertex_count"), TypedValue(static_cast<int64_t>(info.vertex_count))},
@ -3836,7 +3837,10 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
return {query_execution->prepared_query->header, query_execution->prepared_query->privileges, qid,
query_execution->prepared_query->db};
} catch (const utils::BasicException &) {
// Trigger first failed query
metrics::FirstFailedQuery();
memgraph::metrics::IncrementCounter(memgraph::metrics::FailedQuery);
memgraph::metrics::IncrementCounter(memgraph::metrics::FailedPrepare);
AbortCommand(query_execution_ptr);
throw;
}
@ -3862,10 +3866,12 @@ std::vector<TypedValue> Interpreter::GetQueries() {
}
void Interpreter::Abort() {
bool decrement = true;
auto expected = TransactionStatus::ACTIVE;
while (!transaction_status_.compare_exchange_weak(expected, TransactionStatus::STARTED_ROLLBACK)) {
if (expected == TransactionStatus::TERMINATED || expected == TransactionStatus::IDLE) {
transaction_status_.store(TransactionStatus::STARTED_ROLLBACK);
decrement = false;
break;
}
expected = TransactionStatus::ACTIVE;
@ -3881,7 +3887,10 @@ void Interpreter::Abort() {
current_timeout_timer_.reset();
current_transaction_.reset();
memgraph::metrics::DecrementCounter(memgraph::metrics::ActiveTransactions);
if (decrement) {
// Decrement only if the transaction was active when we started to Abort
memgraph::metrics::DecrementCounter(memgraph::metrics::ActiveTransactions);
}
// if (!current_db_.db_transactional_accessor_) return;
current_db_.CleanupDBTransaction(true);

10
src/query/interpreter.hpp
View File

@ -39,6 +39,7 @@
#include "storage/v2/isolation_level.hpp"
#include "storage/v2/storage.hpp"
#include "utils/event_counter.hpp"
#include "utils/event_trigger.hpp"
#include "utils/logging.hpp"
#include "utils/memory.hpp"
#include "utils/settings.hpp"
@ -51,6 +52,9 @@
namespace memgraph::metrics {
extern const Event FailedQuery;
extern const Event FailedPrepare;
extern const Event FailedPull;
extern const Event SuccessfulQuery;
} // namespace memgraph::metrics
namespace memgraph::query {
@ -439,12 +443,18 @@ std::map<std::string, TypedValue> Interpreter::Pull(TStream *result_stream, std:
query_execution.reset(nullptr);
throw;
} catch (const utils::BasicException &) {
// Trigger first failed query
metrics::FirstFailedQuery();
memgraph::metrics::IncrementCounter(memgraph::metrics::FailedQuery);
memgraph::metrics::IncrementCounter(memgraph::metrics::FailedPull);
AbortCommand(&query_execution);
throw;
}
if (maybe_summary) {
// Toggle first successfully completed query
metrics::FirstSuccessfulQuery();
memgraph::metrics::IncrementCounter(memgraph::metrics::SuccessfulQuery);
// return the execution summary
maybe_summary->insert_or_assign("has_more", false);
return std::move(*maybe_summary);

7
src/query/procedure/mg_procedure_impl.cpp
View File

@ -104,30 +104,37 @@ void MgpFreeImpl(memgraph::utils::MemoryResource &memory, void *const p) noexcep
}
struct DeletedObjectException : public memgraph::utils::BasicException {
using memgraph::utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(DeletedObjectException)
};
struct KeyAlreadyExistsException : public memgraph::utils::BasicException {
using memgraph::utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(KeyAlreadyExistsException)
};
struct InsufficientBufferException : public memgraph::utils::BasicException {
using memgraph::utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(InsufficientBufferException)
};
struct ImmutableObjectException : public memgraph::utils::BasicException {
using memgraph::utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(ImmutableObjectException)
};
struct ValueConversionException : public memgraph::utils::BasicException {
using memgraph::utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(ValueConversionException)
};
struct SerializationException : public memgraph::utils::BasicException {
using memgraph::utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(SerializationException)
};
struct AuthorizationException : public memgraph::utils::BasicException {
using memgraph::utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(AuthorizationException)
};
template <typename TFunc, typename TReturn>

1
src/query/stream/streams.hpp
View File

@ -41,6 +41,7 @@ namespace stream {
class StreamsException : public utils::BasicException {
public:
using BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(StreamsException)
};
template <typename T>

1
src/query/typed_value.hpp
View File

@ -566,6 +566,7 @@ class TypedValue {
class TypedValueException : public utils::BasicException {
public:
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(TypedValueException)
};
// binary bool operators

3
src/rpc/exceptions.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -29,6 +29,7 @@ class RpcFailedException final : public utils::BasicException {
/// Returns the endpoint associated with the error.
const io::network::Endpoint &endpoint() const { return endpoint_; }
SPECIALIZE_GET_EXCEPTION_NAME(RpcFailedException)
private:
io::network::Endpoint endpoint_;

3
src/rpc/protocol.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -38,6 +38,7 @@ class Server;
*/
class SessionException : public utils::BasicException {
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(SessionException)
};
/**

1
src/slk/serialization.hpp
View File

@ -48,6 +48,7 @@ static_assert(std::is_same_v<std::uint8_t, char> || std::is_same_v<std::uint8_t,
class SlkDecodeException : public utils::BasicException {
public:
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(SlkDecodeException)
};
// Forward declarations for all recursive `Save` and `Load` functions must be

3
src/slk/streams.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -67,6 +67,7 @@ class Builder {
class SlkReaderException : public utils::BasicException {
public:
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(SlkReaderException)
};
/// Reader used to read data from a SLK segment stream.

2
src/storage/v2/CMakeLists.txt
View File

@ -43,7 +43,7 @@ add_library(mg-storage-v2 STATIC
inmemory/replication/replication_server.cpp
inmemory/replication/replication_client.cpp
)
target_link_libraries(mg-storage-v2 mg::replication Threads::Threads mg-utils gflags absl::flat_hash_map mg-rpc mg-slk)
target_link_libraries(mg-storage-v2 mg::replication Threads::Threads mg-utils gflags absl::flat_hash_map mg-rpc mg-slk mg-events)
# Until we get LTO there is an advantage to do some unity builds
set_target_properties(mg-storage-v2

1
src/storage/v2/config.hpp
View File

@ -23,6 +23,7 @@ namespace memgraph::storage {
/// Exception used to signal configuration errors.
class StorageConfigException : public utils::BasicException {
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(StorageConfigException)
};
/// Pass this class to the \ref Storage constructor to change the behavior of

36
src/storage/v2/disk/storage.cpp
View File

@ -794,16 +794,38 @@ uint64_t DiskStorage::GetDiskSpaceUsage() const {
durability_disk_storage_size;
}
StorageInfo DiskStorage::GetInfo() const {
uint64_t edge_count = edge_count_.load(std::memory_order_acquire);
uint64_t vertex_count = vertex_count_.load(std::memory_order_acquire);
double average_degree = 0.0;
if (vertex_count) {
StorageInfo DiskStorage::GetBaseInfo(bool /* unused */) {
StorageInfo info{};
info.vertex_count = vertex_count_;
info.edge_count = edge_count_.load(std::memory_order_acquire);
if (info.vertex_count) {
// NOLINTNEXTLINE(bugprone-narrowing-conversions, cppcoreguidelines-narrowing-conversions)
average_degree = 2.0 * edge_count / static_cast<double>(vertex_count);
info.average_degree = 2.0 * static_cast<double>(info.edge_count) / info.vertex_count;
}
info.memory_usage = utils::GetMemoryUsage();
info.disk_usage = GetDiskSpaceUsage();
return info;
}
return {vertex_count, edge_count, average_degree, utils::GetMemoryUsage(), GetDiskSpaceUsage()};
StorageInfo DiskStorage::GetInfo(bool force_dir) {
StorageInfo info = GetBaseInfo(force_dir);
{
auto access = Access(std::nullopt);
const auto &lbl = access->ListAllIndices();
info.label_indices = lbl.label.size();
info.label_property_indices = lbl.label_property.size();
const auto &con = access->ListAllConstraints();
info.existence_constraints = con.existence.size();
info.unique_constraints = con.unique.size();
}
info.storage_mode = storage_mode_;
info.isolation_level = isolation_level_;
info.durability_snapshot_enabled =
config_.durability.snapshot_wal_mode != Config::Durability::SnapshotWalMode::DISABLED ||
config_.durability.snapshot_on_exit;
info.durability_wal_enabled =
config_.durability.snapshot_wal_mode == Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL;
return info;
}
void DiskStorage::SetEdgeImportMode(EdgeImportMode edge_import_status) {

3
src/storage/v2/disk/storage.hpp
View File

@ -297,7 +297,8 @@ class DiskStorage final : public Storage {
std::vector<std::pair<std::string, std::string>> SerializeVerticesForLabelPropertyIndex(LabelId label,
PropertyId property);
StorageInfo GetInfo() const override;
StorageInfo GetBaseInfo(bool force_directory) override;
StorageInfo GetInfo(bool force_directory) override;
void FreeMemory(std::unique_lock<utils::ResourceLock> /*lock*/) override {}

3
src/storage/v2/durability/exceptions.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -18,6 +18,7 @@ namespace memgraph::storage::durability {
/// Exception used to handle errors during recovery.
class RecoveryFailure : public utils::BasicException {
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(RecoveryFailure)
};
} // namespace memgraph::storage::durability

55
src/storage/v2/inmemory/storage.cpp
View File

@ -10,6 +10,7 @@
// licenses/APL.txt.
#include "storage/v2/inmemory/storage.hpp"
#include "dbms/constants.hpp"
#include "storage/v2/durability/durability.hpp"
#include "storage/v2/durability/snapshot.hpp"
#include "storage/v2/metadata_delta.hpp"
@ -1475,16 +1476,54 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::ResourceLock> main_
template void InMemoryStorage::CollectGarbage<true>(std::unique_lock<utils::ResourceLock>);
template void InMemoryStorage::CollectGarbage<false>(std::unique_lock<utils::ResourceLock>);
StorageInfo InMemoryStorage::GetInfo() const {
auto vertex_count = vertices_.size();
auto edge_count = edge_count_.load(std::memory_order_acquire);
double average_degree = 0.0;
if (vertex_count) {
StorageInfo InMemoryStorage::GetBaseInfo(bool force_directory) {
StorageInfo info{};
info.vertex_count = vertices_.size();
info.edge_count = edge_count_.load(std::memory_order_acquire);
if (info.vertex_count) {
// NOLINTNEXTLINE(bugprone-narrowing-conversions, cppcoreguidelines-narrowing-conversions)
average_degree = 2.0 * static_cast<double>(edge_count) / vertex_count;
info.average_degree = 2.0 * static_cast<double>(info.edge_count) / info.vertex_count;
}
return {vertex_count, edge_count, average_degree, utils::GetMemoryUsage(),
utils::GetDirDiskUsage(config_.durability.storage_directory)};
info.memory_usage = utils::GetMemoryUsage();
// Special case for the default database
auto update_path = [&](const std::filesystem::path &dir) {
if (!force_directory && std::filesystem::is_directory(dir) && dir.has_filename()) {
const auto end = dir.end();
auto it = end;
--it;
if (it != end) {
--it;
if (it != end && *it != "databases") {
// Default DB points to the root (for back-compatibility); update to the "database" dir
return dir / "databases" / dbms::kDefaultDB;
}
}
}
return dir;
};
info.disk_usage = utils::GetDirDiskUsage<false>(update_path(config_.durability.storage_directory));
return info;
}
StorageInfo InMemoryStorage::GetInfo(bool force_directory) {
StorageInfo info = GetBaseInfo(force_directory);
{
auto access = Access(std::nullopt);
const auto &lbl = access->ListAllIndices();
info.label_indices = lbl.label.size();
info.label_property_indices = lbl.label_property.size();
const auto &con = access->ListAllConstraints();
info.existence_constraints = con.existence.size();
info.unique_constraints = con.unique.size();
}
info.storage_mode = storage_mode_;
info.isolation_level = isolation_level_;
info.durability_snapshot_enabled =
config_.durability.snapshot_wal_mode != Config::Durability::SnapshotWalMode::DISABLED ||
config_.durability.snapshot_on_exit;
info.durability_wal_enabled =
config_.durability.snapshot_wal_mode == Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL;
return info;
}
bool InMemoryStorage::InitializeWalFile(memgraph::replication::ReplicationEpoch &epoch) {

4
src/storage/v2/inmemory/storage.hpp
View File

@ -348,7 +348,9 @@ class InMemoryStorage final : public Storage {
bool InitializeWalFile(memgraph::replication::ReplicationEpoch &epoch);
void FinalizeWalFile();
StorageInfo GetInfo() const override;
StorageInfo GetBaseInfo(bool force_directory) override;
StorageInfo GetInfo(bool force_directory) override;
/// Return true in all cases excepted if any sync replicas have not sent confirmation.
[[nodiscard]] bool AppendToWalDataManipulation(const Transaction &transaction, uint64_t final_commit_timestamp);
/// Return true in all cases excepted if any sync replicas have not sent confirmation.

1
src/storage/v2/property_value.hpp
View File

@ -27,6 +27,7 @@ namespace memgraph::storage {
class PropertyValueException : public utils::BasicException {
public:
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(PropertyValueException)
};
/// Encapsulation of a value and its type in a class that has no compile-time

47
src/storage/v2/storage.hpp
View File

@ -71,8 +71,35 @@ struct StorageInfo {
double average_degree;
uint64_t memory_usage;
uint64_t disk_usage;
uint64_t label_indices;
uint64_t label_property_indices;
uint64_t existence_constraints;
uint64_t unique_constraints;
StorageMode storage_mode;
IsolationLevel isolation_level;
bool durability_snapshot_enabled;
bool durability_wal_enabled;
};
static inline nlohmann::json ToJson(const StorageInfo &info) {
nlohmann::json res;
res["edges"] = info.edge_count;
res["vertices"] = info.vertex_count;
res["memory"] = info.memory_usage;
res["disk"] = info.disk_usage;
res["label_indices"] = info.label_indices;
res["label_prop_indices"] = info.label_property_indices;
res["existence_constraints"] = info.existence_constraints;
res["unique_constraints"] = info.unique_constraints;
res["storage_mode"] = storage::StorageModeToString(info.storage_mode);
res["isolation_level"] = storage::IsolationLevelToString(info.isolation_level);
res["durability"] = {{"snapshot_enabled", info.durability_snapshot_enabled},
{"WAL_enabled", info.durability_wal_enabled}};
return res;
}
struct EdgeInfoForDeletion {
std::unordered_set<Gid> partial_src_edge_ids{};
std::unordered_set<Gid> partial_dest_edge_ids{};
@ -289,7 +316,25 @@ class Storage {
utils::BasicResult<SetIsolationLevelError> SetIsolationLevel(IsolationLevel isolation_level);
IsolationLevel GetIsolationLevel() const noexcept;
virtual StorageInfo GetInfo() const = 0;
virtual StorageInfo GetBaseInfo(bool force_directory) = 0;
StorageInfo GetBaseInfo() {
#if MG_ENTERPRISE
const bool force_dir = false;
#else
const bool force_dir = true; //!< Use the configured directory (multi-tenancy reroutes to another dir)
#endif
return GetBaseInfo(force_dir);
}
virtual StorageInfo GetInfo(bool force_directory) = 0;
StorageInfo GetInfo() {
#if MG_ENTERPRISE
const bool force_dir = false;
#else
const bool force_dir = true; //!< Use the configured directory (multi-tenancy reroutes to another dir)
#endif
return GetInfo(force_dir);
}
virtual Transaction CreateTransaction(IsolationLevel isolation_level, StorageMode storage_mode) = 0;

4
src/telemetry/CMakeLists.txt
View File

@ -3,7 +3,9 @@ set(telemetry_src_files
telemetry.cpp)
add_library(mg-telemetry STATIC ${telemetry_src_files})
target_link_libraries(mg-telemetry mg-requests mg-kvstore mg-utils)
target_include_directories(mg-telemetry PRIVATE ${CMAKE_SOURCE_DIR}/include)
include_directories(${CMAKE_CURRENT_BINARY_DIR}/..)
target_link_libraries(mg-telemetry mg-requests mg-kvstore mg-utils mg-events mg-dbms)
option(MG_TELEMETRY_ID_OVERRIDE "Override for the telemetry ID" STRING)
if(MG_TELEMETRY_ID_OVERRIDE)

6
src/telemetry/collectors.cpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -56,7 +56,7 @@ const std::pair<const std::string, const double> GetCpuUsage(pid_t pid, pid_t ti
return {name, cpu};
}
const nlohmann::json GetResourceUsage() {
const nlohmann::json GetResourceUsage(std::filesystem::path root_directory) {
// Get PID of entire process.
pid_t pid = getpid();
@ -79,7 +79,7 @@ const nlohmann::json GetResourceUsage() {
auto cpu_total = GetCpuUsage(pid);
cpu["usage"] = cpu_total.second;
return {{"cpu", cpu}, {"memory", utils::GetMemoryUsage()}};
return {{"cpu", cpu}, {"memory", utils::GetMemoryUsage()}, {"disk", utils::GetDirDiskUsage(root_directory)}};
}
} // namespace memgraph::telemetry

4
src/telemetry/collectors.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -19,6 +19,6 @@ namespace memgraph::telemetry {
* This function returs a dictionary containing resource usage information
* (total cpu usage and current memory usage).
*/
const nlohmann::json GetResourceUsage();
const nlohmann::json GetResourceUsage(std::filesystem::path root_directory);
} // namespace memgraph::telemetry

109
src/telemetry/telemetry.cpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -17,26 +17,40 @@
#include "requests/requests.hpp"
#include "telemetry/collectors.hpp"
#include "utils/event_counter.hpp"
#include "utils/event_map.hpp"
#include "utils/event_trigger.hpp"
#include "utils/file.hpp"
#include "utils/logging.hpp"
#include "utils/system_info.hpp"
#include "utils/timestamp.hpp"
#include "utils/uuid.hpp"
#include "version.hpp"
namespace memgraph::telemetry {
constexpr auto kMaxBatchSize{100};
Telemetry::Telemetry(std::string url, std::filesystem::path storage_directory, std::string uuid, std::string machine_id,
std::chrono::duration<int64_t> refresh_interval, const uint64_t send_every_n)
bool ssl, std::filesystem::path root_directory, std::chrono::duration<int64_t> refresh_interval,
const uint64_t send_every_n)
: url_(std::move(url)),
uuid_(uuid),
machine_id_(machine_id),
ssl_(ssl),
send_every_n_(send_every_n),
storage_(std::move(storage_directory)) {
StoreData("startup", utils::GetSystemInfo());
AddCollector("resources", GetResourceUsage);
AddCollector("resources", [&, root_directory = std::move(root_directory)]() -> nlohmann::json {
return GetResourceUsage(root_directory);
});
AddCollector("uptime", [&]() -> nlohmann::json { return GetUptime(); });
AddCollector("query", [&]() -> nlohmann::json {
return {
{"first_successful_query",
metrics::global_one_shot_events[metrics::OneShotEvents::kFirstSuccessfulQueryTs].load()},
{"first_failed_query", metrics::global_one_shot_events[metrics::OneShotEvents::kFirstFailedQueryTs].load()}};
});
scheduler_.Run("Telemetry", refresh_interval, [&] { CollectData(); });
}
@ -51,9 +65,14 @@ Telemetry::~Telemetry() {
}
void Telemetry::StoreData(const nlohmann::json &event, const nlohmann::json &data) {
nlohmann::json payload = {
{"run_id", uuid_}, {"type", "telemetry"}, {"machine_id", machine_id_},
{"event", event}, {"data", data}, {"timestamp", utils::Timestamp::Now().SecWithNsecSinceTheEpoch()}};
nlohmann::json payload = {{"run_id", uuid_},
{"type", "telemetry"},
{"machine_id", machine_id_},
{"event", event},
{"data", data},
{"timestamp", utils::Timestamp::Now().SecWithNsecSinceTheEpoch()},
{"ssl", ssl_},
{"version", version_string}};
storage_.Put(fmt::format("{}:{}", uuid_, event.dump()), payload.dump());
}
@ -104,4 +123,82 @@ void Telemetry::CollectData(const std::string &event) {
const nlohmann::json Telemetry::GetUptime() { return timer_.Elapsed().count(); }
void Telemetry::AddQueryModuleCollector() {
AddCollector("query_module_counters",
[]() -> nlohmann::json { return memgraph::query::plan::CallProcedure::GetAndResetCounters(); });
}
void Telemetry::AddEventsCollector() {
AddCollector("event_counters", []() -> nlohmann::json {
nlohmann::json ret;
for (size_t i = 0; i < memgraph::metrics::CounterEnd(); ++i) {
ret[memgraph::metrics::GetCounterName(i)] = memgraph::metrics::global_counters[i].load(std::memory_order_relaxed);
}
return ret;
});
}
void Telemetry::AddClientCollector() {
AddCollector("client", []() -> nlohmann::json { return memgraph::communication::bolt_metrics.ToJson(); });
}
#ifdef MG_ENTERPRISE
void Telemetry::AddDatabaseCollector(dbms::DbmsHandler &dbms_handler) {
AddCollector("database", [&dbms_handler]() -> nlohmann::json {
const auto &infos = dbms_handler.Info();
auto dbs = nlohmann::json::array();
for (const auto &db_info : infos) {
dbs.push_back(memgraph::dbms::ToJson(db_info));
}
return dbs;
});
}
#else
#endif
#ifdef MG_ENTERPRISE
void Telemetry::AddStorageCollector(
dbms::DbmsHandler &dbms_handler,
memgraph::utils::Synchronized<memgraph::auth::Auth, memgraph::utils::WritePrioritizedRWLock> &auth) {
AddCollector("storage", [&dbms_handler, &auth]() -> nlohmann::json {
auto stats = dbms_handler.Stats();
stats.users = auth->AllUsers().size();
return ToJson(stats);
});
}
#else
void Telemetry::AddStorageCollector(
memgraph::utils::Gatekeeper<memgraph::dbms::Database> &db_gatekeeper,
memgraph::utils::Synchronized<memgraph::auth::Auth, memgraph::utils::WritePrioritizedRWLock> &auth) {
AddCollector("storage", [&db_gatekeeper, &auth]() -> nlohmann::json {
memgraph::dbms::Statistics stats;
auto db_acc_opt = db_gatekeeper.access();
MG_ASSERT(db_acc_opt, "Failed to get access to the default database");
auto &db_acc = *db_acc_opt;
const auto &info = db_acc->GetInfo();
const auto &storage_info = info.storage_info;
stats.num_vertex = storage_info.vertex_count;
stats.num_edges = storage_info.edge_count;
stats.triggers = info.triggers;
stats.streams = info.streams;
stats.num_databases = 1;
stats.indices += storage_info.label_indices + storage_info.label_property_indices;
stats.constraints += storage_info.existence_constraints + storage_info.unique_constraints;
++stats.storage_modes[(int)storage_info.storage_mode];
++stats.isolation_levels[(int)storage_info.isolation_level];
stats.snapshot_enabled = storage_info.durability_snapshot_enabled;
stats.wal_enabled = storage_info.durability_wal_enabled;
stats.users = auth->AllUsers().size();
return ToJson(stats);
});
}
#endif
void Telemetry::AddExceptionCollector() {
AddCollector("exception", []() -> nlohmann::json { return memgraph::metrics::global_counters_map.ToJson(); });
}
void Telemetry::AddReplicationCollector() {
// TODO Waiting for the replication refactor to be done before implementing the telemetry
AddCollector("replication", []() -> nlohmann::json { return {{"async", -1}, {"sync", -1}}; });
}
} // namespace memgraph::telemetry

30
src/telemetry/telemetry.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -16,6 +16,7 @@
#include <json/json.hpp>
#include "dbms/dbms_handler.hpp"
#include "kvstore/kvstore.hpp"
#include "utils/scheduler.hpp"
#include "utils/timer.hpp"
@ -35,10 +36,36 @@ namespace memgraph::telemetry {
class Telemetry final {
public:
Telemetry(std::string url, std::filesystem::path storage_directory, std::string uuid, std::string machine_id,
bool ssl, std::filesystem::path root_directory,
std::chrono::duration<int64_t> refresh_interval = std::chrono::minutes(10), uint64_t send_every_n = 10);
// Generic/user-defined collector
void AddCollector(const std::string &name, const std::function<const nlohmann::json(void)> &func);
// Specialized collectors
#ifdef MG_ENTERPRISE
void AddStorageCollector(
dbms::DbmsHandler &dbms_handler,
memgraph::utils::Synchronized<memgraph::auth::Auth, memgraph::utils::WritePrioritizedRWLock> &auth);
#else
void AddStorageCollector(
memgraph::utils::Gatekeeper<memgraph::dbms::Database> &db_gatekeeper,
memgraph::utils::Synchronized<memgraph::auth::Auth, memgraph::utils::WritePrioritizedRWLock> &auth);
#endif
#ifdef MG_ENTERPRISE
void AddDatabaseCollector(dbms::DbmsHandler &dbms_handler);
#else
void AddDatabaseCollector() {
AddCollector("database", []() -> nlohmann::json { return nlohmann::json::array(); });
}
#endif
void AddClientCollector();
void AddEventsCollector();
void AddQueryModuleCollector();
void AddExceptionCollector();
void AddReplicationCollector();
~Telemetry();
Telemetry(const Telemetry &) = delete;
@ -56,6 +83,7 @@ class Telemetry final {
const std::string url_;
const std::string uuid_;
const std::string machine_id_;
const bool ssl_;
uint64_t num_{0};
utils::Scheduler scheduler_;
utils::Timer timer_;

6
src/utils/CMakeLists.txt
View File

@ -1,9 +1,6 @@
set(utils_src_files
async_timer.cpp
base64.cpp
event_counter.cpp
event_gauge.cpp
event_histogram.cpp
file.cpp
file_locker.cpp
memory.cpp
@ -34,3 +31,6 @@ set(settings_src_files
add_library(mg-settings STATIC ${settings_src_files})
target_link_libraries(mg-settings mg-kvstore mg-slk mg-utils)
add_library(mg-events STATIC event_counter.cpp event_gauge.cpp event_histogram.cpp event_trigger.cpp event_map.cpp)
target_link_libraries(mg-events mg-utils json)

5
src/utils/event_counter.cpp
View File

@ -74,7 +74,10 @@
M(ActiveTransactions, Transaction, "Number of active transactions.") \
M(CommitedTransactions, Transaction, "Number of committed transactions.") \
M(RollbackedTransactions, Transaction, "Number of rollbacked transactions.") \
M(FailedQuery, Transaction, "Number of times executing a query failed.")
M(FailedQuery, Transaction, "Number of times executing a query failed.") \
M(FailedPrepare, Transaction, "Number of times preparing a query failed.") \
M(FailedPull, Transaction, "Number of times executing a prepared query failed.") \
M(SuccessfulQuery, Transaction, "Number of successful queries.")
namespace memgraph::metrics {
// define every Event as an index in the array of counters

70
src/utils/event_map.cpp Normal file
View File

@ -0,0 +1,70 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include "utils/event_map.hpp"
namespace {
template <typename T, typename K>
int NameToId(const T &names, const K &name) {
const auto &id = std::find(names.begin(), names.end(), name);
if (id == names.end()) return -1;
return std::distance(names.begin(), id);
}
template <typename T, typename K>
int NameToId(T &names, const K &name, uint64_t &free) {
const auto id = NameToId(names, name);
if (id != -1) return id;
// Add new name
if (free < 1) return -1; // No more space
int idx = names.size() - free;
names[idx] = name;
--free;
return idx;
}
} // namespace
namespace memgraph::metrics {
// Initialize global counters memory
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
Counter global_counters_map_array[EventMap::kMaxCounters]{};
// Initialize global counters map
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
EventMap global_counters_map(global_counters_map_array);
bool EventMap::Increment(const std::string_view event, Count amount) {
const auto id = NameToId(name_to_id_, event, num_free_counters_);
if (id != -1) {
counters_[id].fetch_add(amount, std::memory_order_relaxed);
return true;
}
return false;
}
bool EventMap::Decrement(const std::string_view event, Count amount) {
const auto id = NameToId(name_to_id_, event, num_free_counters_);
if (id != -1) {
counters_[id].fetch_sub(amount, std::memory_order_relaxed);
return true;
}
return false;
}
bool IncrementCounter(const std::string_view event, Count amount) {
return global_counters_map.Increment(event, amount);
}
bool DecrementCounter(const std::string_view event, Count amount) {
return global_counters_map.Decrement(event, amount);
}
} // namespace memgraph::metrics

61
src/utils/event_map.hpp Normal file
View File

@ -0,0 +1,61 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include <atomic>
#include <cstdlib>
#include <memory>
#include "json/json.hpp"
namespace memgraph::metrics {
using Count = uint64_t;
using Counter = std::atomic<Count>;
class EventMap {
public:
static constexpr int kMaxCounters = 48;
explicit EventMap(Counter *allocated_counters) noexcept : counters_(allocated_counters) {}
auto &operator[](std::string_view event);
const auto &operator[](std::string_view event) const;
bool Increment(std::string_view event, Count amount = 1);
bool Decrement(std::string_view event, Count amount = 1);
nlohmann::json ToJson() {
auto res = nlohmann::json::array();
for (size_t i = 0; i < kMaxCounters - num_free_counters_; ++i) {
const auto &event_name = name_to_id_[i];
res.push_back({{"name", event_name}, {"count", counters_[i].load()}});
}
return res;
}
uint64_t num_free_counters_{kMaxCounters};
std::array<std::string, kMaxCounters> name_to_id_;
private:
Counter *counters_;
const auto &operator[](int event) const;
};
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
extern EventMap global_counters_map;
bool IncrementCounter(std::string_view event, Count amount = 1);
bool DecrementCounter(std::string_view event, Count amount = 1);
} // namespace memgraph::metrics

41
src/utils/event_trigger.cpp Normal file
View File

@ -0,0 +1,41 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include "utils/event_trigger.hpp"
#include "utils/timestamp.hpp"
namespace memgraph::metrics {
// Initialize array for the global array with all values set to 0
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
std::atomic<double> global_one_shot_array[(int)OneShotEvents::kNum]{};
// Initialize one shot events
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
EventOneShot global_one_shot_events(global_one_shot_array);
void FirstSuccessfulQuery() {
constexpr double kBusyMark = -1.0;
// Mark as busy (if it fails, someone already set it)
if (global_one_shot_events.Trigger(OneShotEvents::kFirstSuccessfulQueryTs, kBusyMark)) {
global_one_shot_events.Trigger(OneShotEvents::kFirstSuccessfulQueryTs,
utils::Timestamp::Now().SecWithNsecSinceTheEpoch(), kBusyMark);
}
}
void FirstFailedQuery() {
constexpr double kBusyMark = -1.0;
// Mark as busy (if it fails, someone already set it)
if (global_one_shot_events.Trigger(OneShotEvents::kFirstFailedQueryTs, kBusyMark)) {
global_one_shot_events.Trigger(OneShotEvents::kFirstFailedQueryTs,
utils::Timestamp::Now().SecWithNsecSinceTheEpoch(), kBusyMark);
}
}
} // namespace memgraph::metrics

49
src/utils/event_trigger.hpp Normal file
View File

@ -0,0 +1,49 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#pragma once
#include <atomic>
#include <cstdlib>
#include <memory>
namespace memgraph::metrics {
enum class OneShotEvents {
kFirstSuccessfulQueryTs,
kFirstFailedQueryTs,
kNum /* leave at the end */
};
class EventOneShot {
public:
explicit EventOneShot(std::atomic<double> *allocated_values) noexcept : values_(allocated_values) {}
auto &operator[](const OneShotEvents event) { return values_[(int)event]; }
const auto &operator[](const OneShotEvents event) const { return values_[(int)event]; }
bool Trigger(OneShotEvents event, double val, double expected = 0) {
return values_[(int)event].compare_exchange_weak(expected, val);
}
private:
std::atomic<double> *values_;
};
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
extern EventOneShot global_one_shot_events;
void FirstSuccessfulQuery();
void FirstFailedQuery();
} // namespace memgraph::metrics

12
src/utils/exceptions.hpp
View File

@ -25,6 +25,9 @@
namespace memgraph::utils {
#define SPECIALIZE_GET_EXCEPTION_NAME(exep) \
std::string name() const override { return #exep; }
/**
* @brief Base class for all regular exceptions.
*
@ -69,6 +72,8 @@ class BasicException : public std::exception {
*/
const char *what() const noexcept override { return msg_.c_str(); }
virtual std::string name() const { return "BasicException"; }
protected:
/**
* @brief Error message.
@ -146,6 +151,8 @@ class NotYetImplemented final : public BasicException {
template <class... Args>
explicit NotYetImplemented(fmt::format_string<Args...> fmt, Args &&...args) noexcept
: NotYetImplemented(fmt::format(fmt, std::forward<Args>(args)...)) {}
SPECIALIZE_GET_EXCEPTION_NAME(NotYetImplemented)
};
class ParseException final : public BasicException {
@ -156,6 +163,11 @@ class ParseException final : public BasicException {
template <class... Args>
explicit ParseException(fmt::format_string<Args...> fmt, Args &&...args) noexcept
: ParseException(fmt::format(fmt, std::forward<Args>(args)...)) {}
SPECIALIZE_GET_EXCEPTION_NAME(ParseException)
};
inline std::string GetExceptionName(const std::exception &e) { return typeid(e).name(); }
inline std::string GetExceptionName(const utils::BasicException &be) { return be.name(); }
} // namespace memgraph::utils

2
src/utils/java_string_formatter.hpp
View File

@ -26,6 +26,8 @@ class JStringFormatException final : public BasicException {
template <class... Args>
explicit JStringFormatException(fmt::format_string<Args...> fmt, Args &&...args) noexcept
: JStringFormatException(fmt::format(fmt, std::forward<Args>(args)...)) {}
SPECIALIZE_GET_EXCEPTION_NAME(JStringFormatException)
};
template <typename T>

3
src/utils/memory_tracker.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -20,6 +20,7 @@ namespace memgraph::utils {
class OutOfMemoryException : public utils::BasicException {
public:
explicit OutOfMemoryException(const std::string &msg) : utils::BasicException(msg) {}
SPECIALIZE_GET_EXCEPTION_NAME(OutOfMemoryException)
};
class MemoryTracker final {

3
src/utils/stat.hpp
View File

@ -22,12 +22,13 @@ namespace memgraph::utils {
/// Returns the number of bytes a directory is using on disk. If the given path
/// isn't a directory, zero will be returned.
template <bool IgnoreSymlink = true>
inline uint64_t GetDirDiskUsage(const std::filesystem::path &path) {
if (!std::filesystem::is_directory(path)) return 0;
uint64_t size = 0;
for (auto &p : std::filesystem::directory_iterator(path)) {
if (std::filesystem::is_symlink(p)) continue;
if (IgnoreSymlink && std::filesystem::is_symlink(p)) continue;
if (std::filesystem::is_directory(p)) {
size += GetDirDiskUsage(p);
} else if (std::filesystem::is_regular_file(p)) {

3
src/utils/temporal.hpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -54,6 +54,7 @@ bool Underflows(const TType &lhs, const TType &rhs) {
namespace temporal {
struct InvalidArgumentException : public utils::BasicException {
using utils::BasicException::BasicException;
SPECIALIZE_GET_EXCEPTION_NAME(InvalidArgumentException)
};
} // namespace temporal

2
tests/integration/telemetry/CMakeLists.txt
View File

@ -3,4 +3,4 @@ set(client_target_name ${target_name}__client)
add_executable(${client_target_name} client.cpp)
set_target_properties(${client_target_name} PROPERTIES OUTPUT_NAME client)
target_link_libraries(${client_target_name} mg-requests mg-telemetry)
target_link_libraries(${client_target_name} mg-requests mg-telemetry mg-storage-v2 mg-dbms mg-query mg-glue)

45
tests/integration/telemetry/client.cpp
View File

@ -1,4 +1,4 @@
// Copyright 2022 Memgraph Ltd.
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@ -11,7 +11,11 @@
#include <gflags/gflags.h>
#include "dbms/dbms_handler.hpp"
#include "glue/auth_checker.hpp"
#include "glue/auth_handler.hpp"
#include "requests/requests.hpp"
#include "storage/v2/config.hpp"
#include "telemetry/telemetry.hpp"
#include "utils/system_info.hpp"
#include "utils/uuid.hpp"
@ -20,21 +24,56 @@ DEFINE_string(endpoint, "http://127.0.0.1:9000/", "Endpoint that should be used
DEFINE_int64(interval, 1, "Interval used for reporting telemetry in seconds.");
DEFINE_int64(duration, 10, "Duration of the test in seconds.");
DEFINE_string(storage_directory, "", "Path to the storage directory where to save telemetry data.");
DEFINE_string(root_directory, "", "Path to the database durability root dir.");
int main(int argc, char **argv) {
gflags::SetVersionString("telemetry");
gflags::ParseCommandLineFlags(&argc, &argv, true);
// Memgraph backend
std::filesystem::path data_directory{std::filesystem::temp_directory_path() / "MG_telemetry_integration_test"};
memgraph::utils::Synchronized<memgraph::auth::Auth, memgraph::utils::WritePrioritizedRWLock> auth_{data_directory /
"auth"};
memgraph::glue::AuthQueryHandler auth_handler(&auth_, "");
memgraph::glue::AuthChecker auth_checker(&auth_);
memgraph::storage::Config db_config;
memgraph::storage::UpdatePaths(db_config, data_directory);
#ifdef MG_ENTERPRISE
memgraph::dbms::DbmsHandler dbms_handler(db_config, &auth_, false, false);
memgraph::query::InterpreterContext interpreter_context_({}, &dbms_handler, &auth_handler, &auth_checker);
#else
memgraph::utils::Gatekeeper<memgraph::dbms::Database> db_gatekeeper{db_config};
memgraph::query::InterpreterContext interpreter_context_({}, nullptr, &auth_handler, &auth_checker);
#endif
memgraph::requests::Init();
memgraph::telemetry::Telemetry telemetry(FLAGS_endpoint, FLAGS_storage_directory, memgraph::utils::GenerateUUID(),
memgraph::utils::GetMachineId(), std::chrono::seconds(FLAGS_interval), 1);
memgraph::utils::GetMachineId(), false, FLAGS_root_directory,
std::chrono::seconds(FLAGS_interval), 1);
// User defined collector
uint64_t counter = 0;
telemetry.AddCollector("db", [&counter]() -> nlohmann::json {
telemetry.AddCollector("test", [&counter]() -> nlohmann::json {
++counter;
return {{"vertices", counter}, {"edges", counter}};
});
// Memgraph specific collectors
#ifdef MG_ENTERPRISE
telemetry.AddStorageCollector(dbms_handler, auth_);
telemetry.AddDatabaseCollector(dbms_handler);
#else
telemetry.AddStorageCollector(db_gatekeeper, auth_);
telemetry.AddDatabaseCollector();
#endif
telemetry.AddClientCollector();
telemetry.AddEventsCollector();
telemetry.AddQueryModuleCollector();
telemetry.AddExceptionCollector();
telemetry.AddReplicationCollector();
std::this_thread::sleep_for(std::chrono::seconds(FLAGS_duration));
return 0;

8
tests/integration/telemetry/runner.py
View File

@ -27,6 +27,7 @@ def execute_test(**kwargs):
client_binary = kwargs.pop("client")
server_binary = kwargs.pop("server")
storage_directory = kwargs.pop("storage")
root_directory = kwargs.pop("root")
start_server = kwargs.pop("start_server", True)
endpoint = kwargs.pop("endpoint", "")
@ -55,6 +56,8 @@ def execute_test(**kwargs):
duration,
"--storage-directory",
storage_directory,
"--root-directory",
root_directory,
]
if endpoint:
client_args.extend(["--endpoint", endpoint])
@ -108,6 +111,7 @@ if __name__ == "__main__":
args = parser.parse_args()
storage = tempfile.TemporaryDirectory()
durability_root = tempfile.TemporaryDirectory()
for test in TESTS:
print("\033[1;36m~~ Executing test with arguments:", json.dumps(test, sort_keys=True), "~~\033[0m")
@ -120,7 +124,9 @@ if __name__ == "__main__":
assert proc.wait() == 0
try:
success = execute_test(client=args.client, server=args.server, storage=storage.name, **test)
success = execute_test(
client=args.client, server=args.server, storage=storage.name, root=durability_root, **test
)
except Exception as e:
print("\033[1;33m", e, "\033[0m", sep="")
success = False

41
tests/integration/telemetry/server.py
View File

@ -109,9 +109,14 @@ def item_sort_key(obj):
def verify_storage(storage, args):
rid = storage[0]["run_id"]
version = storage[0]["version"]
assert version != ""
timestamp = 0
for i, item in enumerate(storage):
assert item["run_id"] == rid
assert item["version"] == version
print(item)
print("\n")
assert item["timestamp"] >= timestamp
timestamp = item["timestamp"]
@ -132,12 +137,17 @@ def verify_storage(storage, args):
assert "os" in item["data"]
assert "swap" in item["data"]
else:
assert item["data"]["db"]["vertices"] == i
assert item["data"]["db"]["edges"] == i
assert "ssl" in item
# User defined data
assert item["data"]["test"]["vertices"] == i
assert item["data"]["test"]["edges"] == i
# Global data
assert "resources" in item["data"]
assert "cpu" in item["data"]["resources"]
assert "memory" in item["data"]["resources"]
assert "disk" in item["data"]["resources"]
assert "uptime" in item["data"]
uptime = item["data"]["uptime"]
@ -149,6 +159,33 @@ def verify_storage(storage, args):
expected = uptime
assert uptime >= expected - 4 and uptime <= expected + 4
# Memgraph specific data
# TODO Missing clients and other usage based data
assert "client" in item["data"]
assert "database" in item["data"]
assert "disk" in item["data"]["database"][0]
assert "durability" in item["data"]["database"][0]
assert "WAL_enabled" in item["data"]["database"][0]["durability"]
assert "snapshot_enabled" in item["data"]["database"][0]["durability"]
assert "edges" in item["data"]["database"][0]
assert "existence_constraints" in item["data"]["database"][0]
assert "isolation_level" in item["data"]["database"][0]
assert "label_indices" in item["data"]["database"][0]
assert "label_prop_indices" in item["data"]["database"][0]
assert "memory" in item["data"]["database"][0]
assert "storage_mode" in item["data"]["database"][0]
assert "unique_constraints" in item["data"]["database"][0]
assert "vertices" in item["data"]["database"][0]
assert "event_counters" in item["data"]
assert "exception" in item["data"]
assert "query" in item["data"]
assert "first_failed_query" in item["data"]["query"]
assert "first_successful_query" in item["data"]["query"]
assert "query_module_counters" in item["data"]
assert "replication" in item["data"]
assert "async" in item["data"]["replication"]
assert "sync" in item["data"]["replication"]
if __name__ == "__main__":
parser = argparse.ArgumentParser()

10
tests/unit/CMakeLists.txt
View File

@ -228,7 +228,7 @@ add_unit_test(utils_exceptions.cpp)
target_link_libraries(${test_prefix}utils_exceptions mg-utils)
add_unit_test(utils_histogram.cpp)
target_link_libraries(${test_prefix}utils_histogram mg-utils)
target_link_libraries(${test_prefix}utils_histogram mg-utils mg-events)
add_unit_test(utils_file.cpp)
target_link_libraries(${test_prefix}utils_file mg-utils)
@ -354,6 +354,12 @@ target_link_libraries(${test_prefix}storage_v2_isolation_level mg-storage-v2)
add_unit_test(storage_v2_show_storage_info.cpp)
target_link_libraries(${test_prefix}storage_v2_show_storage_info mg-storage-v2)
add_unit_test(storage_v2_get_info.cpp)
target_link_libraries(${test_prefix}storage_v2_get_info mg-storage-v2)
add_unit_test(database_get_info.cpp)
target_link_libraries(${test_prefix}database_get_info mg-storage-v2 mg-glue mg-dbms)
add_unit_test(storage_v2_storage_mode.cpp)
target_link_libraries(${test_prefix}storage_v2_storage_mode mg-storage-v2 storage_test_utils mg-query mg-glue)
@ -411,8 +417,8 @@ if(MG_ENTERPRISE)
target_link_libraries(${test_prefix}dbms_handler mg-query mg-auth mg-glue mg-dbms)
endif()
# Test distributed
add_unit_test(distributed_lamport_clock.cpp)
target_link_libraries(${test_prefix}distributed_lamport_clock mg-distributed)
target_include_directories(${test_prefix}distributed_lamport_clock PRIVATE ${CMAKE_SOURCE_DIR}/include)

179
tests/unit/database_get_info.cpp Normal file
View File

@ -0,0 +1,179 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include <filesystem>
#include "dbms/database.hpp"
#include "disk_test_utils.hpp"
#include "query/interpret/awesome_memgraph_functions.hpp"
#include "query/interpreter_context.hpp"
#include "storage/v2/disk/storage.hpp"
#include "storage/v2/inmemory/storage.hpp"
// NOLINTNEXTLINE(google-build-using-namespace)
using namespace memgraph::storage;
constexpr auto testSuite = "database_v2_get_info";
const std::filesystem::path storage_directory{std::filesystem::temp_directory_path() / testSuite};
template <typename StorageType>
class InfoTest : public testing::Test {
protected:
void SetUp() {
auto db_acc_opt = db_gk->access();
MG_ASSERT(db_acc_opt, "Failed to access db");
auto &db_acc = *db_acc_opt;
MG_ASSERT(db_acc->GetStorageMode() == (std::is_same_v<StorageType, memgraph::storage::DiskStorage>
? memgraph::storage::StorageMode::ON_DISK_TRANSACTIONAL
: memgraph::storage::StorageMode::IN_MEMORY_TRANSACTIONAL),
"Wrong storage mode!");
db_acc_ = std::move(db_acc);
}
void TearDown() {
db_acc_.reset();
db_gk.reset();
if (std::is_same<StorageType, memgraph::storage::DiskStorage>::value) {
disk_test_utils::RemoveRocksDbDirs(testSuite);
}
std::filesystem::remove_all(storage_directory);
}
StorageMode mode{std::is_same_v<StorageType, DiskStorage> ? StorageMode::ON_DISK_TRANSACTIONAL
: StorageMode::IN_MEMORY_TRANSACTIONAL};
std::optional<memgraph::dbms::DatabaseAccess> db_acc_;
std::optional<memgraph::utils::Gatekeeper<memgraph::dbms::Database>> db_gk{
[&]() {
memgraph::storage::Config config{};
memgraph::storage::UpdatePaths(config, storage_directory);
config.durability.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL;
if constexpr (std::is_same_v<StorageType, memgraph::storage::DiskStorage>) {
config.disk = disk_test_utils::GenerateOnDiskConfig(testSuite).disk;
config.force_on_disk = true;
}
return config;
}() // iile
};
};
using StorageTypes = ::testing::Types<memgraph::storage::InMemoryStorage, memgraph::storage::DiskStorage>;
TYPED_TEST_CASE(InfoTest, StorageTypes);
// TYPED_TEST_CASE(IndexTest, InMemoryStorageType);
// NOLINTNEXTLINE(hicpp-special-member-functions)
TYPED_TEST(InfoTest, InfoCheck) {
auto &db_acc = *this->db_acc_;
auto lbl = db_acc->storage()->NameToLabel("label");
auto lbl2 = db_acc->storage()->NameToLabel("abc");
auto lbl3 = db_acc->storage()->NameToLabel("3");
auto prop = db_acc->storage()->NameToProperty("prop");
auto prop2 = db_acc->storage()->NameToProperty("another prop");
{
{
auto unique_acc = db_acc->storage()->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateExistenceConstraint(lbl, prop).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = db_acc->storage()->UniqueAccess();
ASSERT_FALSE(unique_acc->DropExistenceConstraint(lbl, prop).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
auto acc = db_acc->Access();
auto v1 = acc->CreateVertex();
auto v2 = acc->CreateVertex();
auto v3 = acc->CreateVertex();
auto v4 = acc->CreateVertex();
auto v5 = acc->CreateVertex();
ASSERT_FALSE(v2.AddLabel(lbl).HasError());
ASSERT_FALSE(v3.AddLabel(lbl).HasError());
ASSERT_FALSE(v3.SetProperty(prop, PropertyValue(42)).HasError());
ASSERT_FALSE(v4.AddLabel(lbl).HasError());
auto et = acc->NameToEdgeType("et5");
ASSERT_FALSE(acc->CreateEdge(&v1, &v2, et).HasError());
ASSERT_FALSE(acc->CreateEdge(&v4, &v3, et).HasError());
ASSERT_FALSE(acc->Commit().HasError());
}
{
auto unique_acc = db_acc->storage()->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateIndex(lbl).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = db_acc->storage()->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateIndex(lbl, prop).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = db_acc->storage()->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateIndex(lbl, prop2).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = db_acc->storage()->UniqueAccess();
ASSERT_FALSE(unique_acc->DropIndex(lbl, prop).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = db_acc->storage()->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateUniqueConstraint(lbl, {prop2}).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = db_acc->storage()->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateUniqueConstraint(lbl2, {prop}).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = db_acc->storage()->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateUniqueConstraint(lbl3, {prop}).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = db_acc->storage()->UniqueAccess();
ASSERT_EQ(unique_acc->DropUniqueConstraint(lbl, {prop2}),
memgraph::storage::UniqueConstraints::DeletionStatus::SUCCESS);
ASSERT_FALSE(unique_acc->Commit().HasError());
}
const auto &info = db_acc->GetInfo(true); // force to use configured directory
ASSERT_EQ(info.storage_info.vertex_count, 5);
ASSERT_EQ(info.storage_info.edge_count, 2);
ASSERT_EQ(info.storage_info.average_degree, 0.8);
ASSERT_GT(info.storage_info.memory_usage, 10'000'000); // 200MB < > 10MB
ASSERT_LT(info.storage_info.memory_usage, 200'000'000);
ASSERT_GT(info.storage_info.disk_usage, 100); // 1MB < > 100B
ASSERT_LT(info.storage_info.disk_usage, 1000'000);
ASSERT_EQ(info.storage_info.label_indices, 1);
ASSERT_EQ(info.storage_info.label_property_indices, 1);
ASSERT_EQ(info.storage_info.existence_constraints, 0);
ASSERT_EQ(info.storage_info.unique_constraints, 2);
ASSERT_EQ(info.storage_info.storage_mode, this->mode);
ASSERT_EQ(info.storage_info.isolation_level, IsolationLevel::SNAPSHOT_ISOLATION);
ASSERT_EQ(info.storage_info.durability_snapshot_enabled, true);
ASSERT_EQ(info.storage_info.durability_wal_enabled, true);
ASSERT_EQ(info.triggers, 0);
ASSERT_EQ(info.streams, 0);
}

2
tests/unit/storage_v2_durability_inmemory.cpp
View File

@ -663,7 +663,7 @@ class DurabilityTest : public ::testing::TestWithParam<bool> {
}
if (verify_info) {
auto info = store->GetInfo();
auto info = store->GetBaseInfo();
if (have_base_dataset) {
if (have_extended_dataset) {
ASSERT_EQ(info.vertex_count, kNumBaseVertices + kNumExtendedVertices);

155
tests/unit/storage_v2_get_info.cpp Normal file
View File

@ -0,0 +1,155 @@
// Copyright 2023 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
// License, and you may not use this file except in compliance with the Business Source License.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.
#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include <filesystem>
#include "disk_test_utils.hpp"
#include "storage/v2/disk/storage.hpp"
#include "storage/v2/inmemory/storage.hpp"
#include "storage/v2/isolation_level.hpp"
#include "storage/v2/storage.hpp"
#include "storage/v2/storage_error.hpp"
// NOLINTNEXTLINE(google-build-using-namespace)
using namespace memgraph::storage;
constexpr auto testSuite = "storage_v2_get_info";
const std::filesystem::path storage_directory{std::filesystem::temp_directory_path() / testSuite};
template <typename StorageType>
class InfoTest : public testing::Test {
protected:
void SetUp() override {
std::filesystem::remove_all(storage_directory);
memgraph::storage::UpdatePaths(config_, storage_directory);
config_.durability.snapshot_wal_mode =
memgraph::storage::Config::Durability::SnapshotWalMode::PERIODIC_SNAPSHOT_WITH_WAL;
this->storage = std::make_unique<StorageType>(config_);
}
void TearDown() override {
std::filesystem::remove_all(storage_directory);
this->storage.reset(nullptr);
}
memgraph::storage::Config config_;
std::unique_ptr<memgraph::storage::Storage> storage;
StorageMode mode{std::is_same_v<StorageType, DiskStorage> ? StorageMode::ON_DISK_TRANSACTIONAL
: StorageMode::IN_MEMORY_TRANSACTIONAL};
};
using StorageTypes = ::testing::Types<memgraph::storage::InMemoryStorage, memgraph::storage::DiskStorage>;
TYPED_TEST_CASE(InfoTest, StorageTypes);
// TYPED_TEST_CASE(IndexTest, InMemoryStorageType);
// NOLINTNEXTLINE(hicpp-special-member-functions)
TYPED_TEST(InfoTest, InfoCheck) {
auto lbl = this->storage->NameToLabel("label");
auto lbl2 = this->storage->NameToLabel("abc");
auto lbl3 = this->storage->NameToLabel("3");
auto prop = this->storage->NameToProperty("prop");
auto prop2 = this->storage->NameToProperty("another prop");
{
{
auto unique_acc = this->storage->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateExistenceConstraint(lbl, prop).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = this->storage->UniqueAccess();
ASSERT_FALSE(unique_acc->DropExistenceConstraint(lbl, prop).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
auto acc = this->storage->Access();
auto v1 = acc->CreateVertex();
auto v2 = acc->CreateVertex();
auto v3 = acc->CreateVertex();
auto v4 = acc->CreateVertex();
auto v5 = acc->CreateVertex();
ASSERT_FALSE(v2.AddLabel(lbl).HasError());
ASSERT_FALSE(v3.AddLabel(lbl).HasError());
ASSERT_FALSE(v3.SetProperty(prop, PropertyValue(42)).HasError());
ASSERT_FALSE(v4.AddLabel(lbl).HasError());
auto et = acc->NameToEdgeType("et5");
ASSERT_FALSE(acc->CreateEdge(&v1, &v2, et).HasError());
ASSERT_FALSE(acc->CreateEdge(&v4, &v3, et).HasError());
ASSERT_FALSE(acc->Commit().HasError());
}
{
auto unique_acc = this->storage->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateIndex(lbl).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = this->storage->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateIndex(lbl, prop).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = this->storage->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateIndex(lbl, prop2).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = this->storage->UniqueAccess();
ASSERT_FALSE(unique_acc->DropIndex(lbl, prop).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = this->storage->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateUniqueConstraint(lbl, {prop2}).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = this->storage->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateUniqueConstraint(lbl2, {prop}).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = this->storage->UniqueAccess();
ASSERT_FALSE(unique_acc->CreateUniqueConstraint(lbl3, {prop}).HasError());
ASSERT_FALSE(unique_acc->Commit().HasError());
}
{
auto unique_acc = this->storage->UniqueAccess();
ASSERT_EQ(unique_acc->DropUniqueConstraint(lbl, {prop2}),
memgraph::storage::UniqueConstraints::DeletionStatus::SUCCESS);
ASSERT_FALSE(unique_acc->Commit().HasError());
}
StorageInfo info = this->storage->GetInfo(true); // force to use configured directory
ASSERT_EQ(info.vertex_count, 5);
ASSERT_EQ(info.edge_count, 2);
ASSERT_EQ(info.average_degree, 0.8);
ASSERT_GT(info.memory_usage, 10'000'000); // 200MB < > 10MB
ASSERT_LT(info.memory_usage, 200'000'000);
ASSERT_GT(info.disk_usage, 100); // 1MB < > 100B
ASSERT_LT(info.disk_usage, 1000'000);
ASSERT_EQ(info.label_indices, 1);
ASSERT_EQ(info.label_property_indices, 1);
ASSERT_EQ(info.existence_constraints, 0);
ASSERT_EQ(info.unique_constraints, 2);
ASSERT_EQ(info.storage_mode, this->mode);
ASSERT_EQ(info.isolation_level, IsolationLevel::SNAPSHOT_ISOLATION);
ASSERT_EQ(info.durability_snapshot_enabled, true);
ASSERT_EQ(info.durability_wal_enabled, true);
}

4
tests/unit/storage_v2_show_storage_info.cpp
View File

@ -52,11 +52,11 @@ TEST_F(ShowStorageInfoTest, CountOnAbort) {
ASSERT_EQ(edge.EdgeType(), et);
ASSERT_EQ(edge.FromVertex(), src_vertex);
ASSERT_EQ(edge.ToVertex(), dest_vertex);
memgraph::storage::StorageInfo info_before_abort = this->storage->GetInfo();
memgraph::storage::StorageInfo info_before_abort = this->storage->GetBaseInfo();
ASSERT_EQ(info_before_abort.vertex_count, 2);
ASSERT_EQ(info_before_abort.edge_count, 1);
acc->Abort();
memgraph::storage::StorageInfo info_after_abort = this->storage->GetInfo();
memgraph::storage::StorageInfo info_after_abort = this->storage->GetBaseInfo();
ASSERT_EQ(info_after_abort.vertex_count, 0);
ASSERT_EQ(info_after_abort.edge_count, 0);
}