Feature/rdb metrics (#1692)

* rocksdb metrics Signed-off-by: iceseer <[email protected]> Signed-off-by: Alexander Lednev <[email protected]>
hyperledger-iroha · Dec 29, 2021 · 0fc3c5d · 0fc3c5d
1 parent 24ad890
commit 0fc3c5d
Show file tree

Hide file tree

Showing 11 changed files with 157 additions and 3 deletions.
diff --git a/irohad/ametsuchi/CMakeLists.txt b/irohad/ametsuchi/CMakeLists.txt
@@ -258,4 +258,5 @@ target_link_libraries(ametsuchi_rocksdb
     shared_model_interfaces_factories
     rocksdb_indexer
     shared_model_interfaces
+    async_subscription
     )
diff --git a/irohad/ametsuchi/impl/rocksdb_command_executor.cpp b/irohad/ametsuchi/impl/rocksdb_command_executor.cpp
@@ -34,6 +34,8 @@
 #include "interfaces/commands/set_setting_value.hpp"
 #include "interfaces/commands/subtract_asset_quantity.hpp"
 #include "interfaces/commands/transfer_asset.hpp"
+#include "main/rdb_status.hpp"
+#include "main/subscription.hpp"
 
 using namespace iroha;
 using namespace iroha::ametsuchi;
@@ -53,6 +55,23 @@ RocksDbCommandExecutor::RocksDbCommandExecutor(
       vm_caller_{vm_caller},
       db_transaction_(db_context_) {
   assert(db_context_);
+
+  getSubscription()->dispatcher()->repeat(
+      SubscriptionEngineHandlers::kMetrics,
+      std::chrono::seconds(5ull), /// repeat task execution period
+      [wdb_context_(utils::make_weak(db_context_))]() {
+        if (auto db_context = wdb_context_.lock()) {
+          RocksDbCommon common(db_context);
+          getSubscription()->notify(
+              EventTypes::kOnRdbStats,
+              RocksDbStatus{common.propGetBlockCacheCapacity(),
+                            common.propGetBlockCacheUsage(),
+                            common.propGetCurSzAllMemTables(),
+                            common.propGetNumSnapshots(),
+                            common.propGetTotalSSTFilesSize()});
+        }
+      },
+      []() { return true; });
 }
 
 RocksDbCommandExecutor::~RocksDbCommandExecutor() = default;

diff --git a/irohad/ametsuchi/impl/rocksdb_common.hpp b/irohad/ametsuchi/impl/rocksdb_common.hpp
@@ -534,6 +534,24 @@ namespace iroha::ametsuchi {
       }
     }
 
+    std::optional<uint64_t> getPropUInt64(const rocksdb::Slice &property) {
+      if (transaction_db_) {
+        uint64_t value;
+        transaction_db_->GetIntProperty(property, &value);
+        return value;
+      }
+      return std::nullopt;
+    }
+
+    std::optional<std::string> getPropStr(const rocksdb::Slice &property) {
+      if (transaction_db_) {
+        std::string value;
+        transaction_db_->GetProperty(property, &value);
+        return value;
+      }
+      return std::nullopt;
+    }
+
    private:
     std::unique_ptr<rocksdb::OptimisticTransactionDB> transaction_db_;
     std::optional<std::string> db_name_;
@@ -655,6 +673,29 @@ namespace iroha::ametsuchi {
       tx_context_->db_port->printStatus(log);
     }
 
+    auto propGetBlockCacheUsage() {
+      return tx_context_->db_port->getPropUInt64("rocksdb.block-cache-usage");
+    }
+
+    auto propGetCurSzAllMemTables() {
+      return tx_context_->db_port->getPropUInt64(
+          "rocksdb.cur-size-all-mem-tables");
+    }
+
+    auto propGetNumSnapshots() {
+      return tx_context_->db_port->getPropUInt64("rocksdb.num-snapshots");
+    }
+
+    auto propGetTotalSSTFilesSize() {
+      return tx_context_->db_port->getPropUInt64(
+          "rocksdb.total-sst-files-size");
+    }
+
+    auto propGetBlockCacheCapacity() {
+      return tx_context_->db_port->getPropUInt64(
+          "rocksdb.block-cache-capacity");
+    }
+
     /// Makes commit to DB
     auto commit() {
       rocksdb::Status status;

diff --git a/irohad/iroha_migrate/iroha_migrate.cpp b/irohad/iroha_migrate/iroha_migrate.cpp
@@ -45,6 +45,7 @@
 #include "validators/default_validator.hpp"
 #include "validators/protobuf/proto_block_validator.hpp"
 #include "validators/protobuf/proto_query_validator.hpp"
+#include "main/subscription.hpp"
 
 #define STR(y) STRH(y)
 #define STRH(x) #x
@@ -280,7 +281,13 @@ expected::Result<void, std::string> restoreWsv() {
   return {};
 }
 
+std::shared_ptr<iroha::Subscription> subscription_manager;
 int main(int argc, char *argv[]) try {
+  subscription_manager = iroha::getSubscription();
+  std::unique_ptr<int, void(*)(int*)> keeper((int*)0x01, [](auto *){
+      subscription_manager->dispose();
+  });
+
   gflags::SetVersionString("1.2");
   gflags::ParseCommandLineFlags(&argc, &argv, true);
   gflags::SetUsageMessage(

diff --git a/irohad/main/rdb_status.hpp b/irohad/main/rdb_status.hpp
@@ -0,0 +1,23 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef IROHA_RDB_STATUS_HPP
+#define IROHA_RDB_STATUS_HPP
+
+#include <optional>
+
+namespace iroha {
+
+  struct RocksDbStatus {
+    std::optional<uint64_t> block_cache_capacity;
+    std::optional<uint64_t> block_cache_usage;
+    std::optional<uint64_t> all_mem_tables_sz;
+    std::optional<uint64_t> num_snapshots;
+    std::optional<uint64_t> sst_files_size;
+  };
+
+}  // namespace iroha
+
+#endif  // IROHA_RDB_STATUS_HPP
diff --git a/irohad/main/subscription_fwd.hpp b/irohad/main/subscription_fwd.hpp
@@ -43,6 +43,9 @@ namespace iroha {
     kOnConsensusGateEvent,
     kSendBatchComplete,
 
+    // RDB
+    kOnRdbStats,
+
     // Node status
     kOnIrohaStatus,
 

diff --git a/irohad/maintenance/metrics.cpp b/irohad/maintenance/metrics.cpp
@@ -186,6 +186,58 @@ Metrics::Metrics(std::string const &listen_addr,
         number_of_pending_mst_transactions.Set(std::get<1>(mstmetr));
       });
 
+  ////////////////////////////////////////////////////////////
+
+  auto &param_block_cache_cap = BuildGauge()
+                                    .Name("rdb_block_cache_capacity")
+                                    .Help("RocksDB block cache capacity")
+                                    .Register(*registry_)
+                                    .Add({});
+
+  auto &param_block_cache_usage = BuildGauge()
+                                      .Name("rdb_block_cache_usage")
+                                      .Help("RocksDB block cache usage")
+                                      .Register(*registry_)
+                                      .Add({});
+
+  auto &param_all_mem_tables_sz = BuildGauge()
+                                      .Name("rdb_all_mem_tables_sz")
+                                      .Help("RocksDB all mem tables size")
+                                      .Register(*registry_)
+                                      .Add({});
+
+  auto &param_num_snapshots = BuildGauge()
+                                  .Name("rdb_num_snapshots")
+                                  .Help("RocksDB number of snapshots")
+                                  .Register(*registry_)
+                                  .Add({});
+
+  auto &param_sst_files_size = BuildGauge()
+                                   .Name("rdb_sst_files_size")
+                                   .Help("RocksDB SST files size")
+                                   .Register(*registry_)
+                                   .Add({});
+
+  rdb_subscriber_ =
+      SubscriberCreator<bool, iroha::RocksDbStatus>::template create<
+          EventTypes::kOnRdbStats>(
+          SubscriptionEngineHandlers::kMetrics,
+          [&](auto &, iroha::RocksDbStatus status) {
+            if (status.block_cache_capacity)
+              param_block_cache_cap.Set(*status.block_cache_capacity);
+
+            if (status.block_cache_usage)
+              param_block_cache_usage.Set(*status.block_cache_usage);
+
+            if (status.all_mem_tables_sz)
+              param_all_mem_tables_sz.Set(*status.all_mem_tables_sz);
+
+            if (status.num_snapshots)
+              param_num_snapshots.Set(*status.num_snapshots);
+
+            if (status.sst_files_size)
+              param_sst_files_size.Set(*status.sst_files_size);
+          });
   ///////////////////////////////
 
   auto calc_uptime_ms = [uptime_start_timepoint_(uptime_start_timepoint_)] {

diff --git a/irohad/maintenance/metrics.hpp b/irohad/maintenance/metrics.hpp
@@ -20,6 +20,7 @@
 #include "interfaces/common_objects/types.hpp"
 #include "interfaces/iroha_internal/block.hpp"
 #include "logger/logger_fwd.hpp"
+#include "main/rdb_status.hpp"
 #include "main/iroha_status.hpp"
 #include "main/subscription.hpp"
 #include "network/ordering_gate_common.hpp"
@@ -32,13 +33,15 @@ class Metrics : public std::enable_shared_from_this<Metrics> {
   using BlockSubscriber = iroha::BaseSubscriber<bool, BlockPtr>;
   using MstMetrics = std::tuple<size_t, size_t>;
   using MstSubscriber = iroha::BaseSubscriber<bool, MstMetrics>;
+  using RdbSubscriber = iroha::BaseSubscriber<bool, iroha::RocksDbStatus>;
 
   std::string listen_addr_port_;
   std::shared_ptr<prometheus::Exposer> exposer_;
   std::shared_ptr<prometheus::Registry> registry_;
   std::shared_ptr<iroha::ametsuchi::Storage> storage_;
   std::shared_ptr<BlockSubscriber> block_subscriber_;
   std::shared_ptr<MstSubscriber> mst_subscriber_;
+  std::shared_ptr<RdbSubscriber> rdb_subscriber_;
   logger::LoggerPtr logger_;
   std::chrono::steady_clock::time_point uptime_start_timepoint_;
   std::thread uptime_thread_;

diff --git a/irohad/subscription/sync_dispatcher_impl.hpp b/irohad/subscription/sync_dispatcher_impl.hpp
@@ -40,7 +40,7 @@ namespace iroha::subscription {
                 std::chrono::microseconds timeout,
                 typename Parent::Task &&task,
                 typename Parent::Predicate &&pred) override {
-      while (!pred || pred()) task();
+      if (!pred || pred()) task();
     }
 
     std::optional<Tid> bind(std::shared_ptr<IScheduler> scheduler) override {

diff --git a/test/integration/executor/executor_fixture_param.cpp b/test/integration/executor/executor_fixture_param.cpp
@@ -10,6 +10,9 @@
 using namespace executor_testing;
 
 ExecutorTestParam::ExecutorTestParam()
-    : vm_caller_(std::make_unique<iroha::ametsuchi::MockVmCaller>()) {}
+    : vm_caller_(std::make_unique<iroha::ametsuchi::MockVmCaller>()),
+      subscription_manager_(iroha::getSubscription()) {}
 
-ExecutorTestParam::~ExecutorTestParam() = default;
+ExecutorTestParam::~ExecutorTestParam() {
+  subscription_manager_->dispose();
+}
diff --git a/test/integration/executor/executor_fixture_param.hpp b/test/integration/executor/executor_fixture_param.hpp
@@ -10,6 +10,7 @@
 
 #include <gtest/gtest.h>
 #include "interfaces/common_objects/types.hpp"
+#include "main/subscription.hpp"
 
 namespace iroha::ametsuchi {
   class BlockIndex;
@@ -52,6 +53,7 @@ namespace executor_testing {
     virtual std::string toString() const = 0;
 
     std::unique_ptr<iroha::ametsuchi::MockVmCaller> vm_caller_;
+    std::shared_ptr<iroha::Subscription> subscription_manager_;
   };
 
 }  // namespace executor_testing