Merge pull request #2722 from cloudnoize/elerer/pruning_metrics

V4 - Optimizing recovery, removing redundant measurements and adding metrics

Author: cloudnoize

bftengine/include/bftengine/IRequestHandler.hpp

@@ -22,7 +22,6 @@
 #include <deque>
 #include "OpenTracing.hpp"
 #include "TimeService.hpp"
-#include "ISystemResourceEntity.hpp"
 #include "PersistentStorageImp.hpp"
 
 namespace concord::reconfiguration {
@@ -55,9 +54,7 @@ class IRequestsHandler {
   };
 
   static std::shared_ptr<IRequestsHandler> createRequestsHandler(
-      std::shared_ptr<IRequestsHandler> userReqHandler,
-      const std::shared_ptr<concord::cron::CronTableRegistry> &,
-      concord::performance::ISystemResourceEntity &);
+      std::shared_ptr<IRequestsHandler> userReqHandler, const std::shared_ptr<concord::cron::CronTableRegistry> &);
   typedef std::deque<ExecutionRequest> ExecutionRequestsQueue;
 
   virtual void execute(ExecutionRequestsQueue &requests,

bftengine/src/bftengine/BFTEngine.cpp

@@ -287,9 +287,8 @@ IReplica::IReplicaPtr IReplica::createNewRoReplica(const ReplicaConfig &replicaC
 
 std::shared_ptr<IRequestsHandler> IRequestsHandler::createRequestsHandler(
     std::shared_ptr<IRequestsHandler> userReqHandler,
-    const std::shared_ptr<concord::cron::CronTableRegistry> &cronTableRegistry,
-    concord::performance::ISystemResourceEntity &resourceEntity) {
-  auto reqHandler = new bftEngine::RequestHandler(resourceEntity);
+    const std::shared_ptr<concord::cron::CronTableRegistry> &cronTableRegistry) {
+  auto reqHandler = new bftEngine::RequestHandler();
   reqHandler->setUserRequestHandler(userReqHandler);
   reqHandler->setCronTableRegistry(cronTableRegistry);
   return std::shared_ptr<IRequestsHandler>(reqHandler);

bftengine/src/bftengine/RequestHandler.cpp

@@ -183,14 +183,7 @@ void RequestHandler::execute(IRequestsHandler::ExecutionRequestsQueue& requests,
     return;
   }
   if (userRequestsHandler_) {
-    // Do not measure pre-exec and read requests.
-    auto isPost =
-        (requests.size() > 0 && !(requests.back().flags & (bftEngine::PRE_PROCESS_FLAG | bftEngine::READ_ONLY_FLAG)));
-    ISystemResourceEntity::scopedDurMeasurment m(
-        resourceEntity_, ISystemResourceEntity::type::post_execution_utilization, isPost);
     userRequestsHandler_->execute(requests, timestamp, batchCid, parent_span);
-    // the size of the queue resembles how many requests have passed consensus.
-    resourceEntity_.addMeasurement({ISystemResourceEntity::type::transactions_accumulated, requests.size(), 0, 0});
   }
   return;
 }

bftengine/src/bftengine/RequestHandler.h

@@ -25,9 +25,7 @@ namespace bftEngine {
 class RequestHandler : public IRequestsHandler {
  public:
   RequestHandler(
-      concord::performance::ISystemResourceEntity &resourceEntity,
-      std::shared_ptr<concordMetrics::Aggregator> aggregator_ = std::make_shared<concordMetrics::Aggregator>())
-      : resourceEntity_(resourceEntity) {
+      std::shared_ptr<concordMetrics::Aggregator> aggregator_ = std::make_shared<concordMetrics::Aggregator>()) {
     using namespace concord::reconfiguration;
     reconfig_handler_.push_back(std::make_shared<ReconfigurationHandler>());
     for (const auto &rh : reconfig_handler_) {
@@ -74,7 +72,6 @@ class RequestHandler : public IRequestsHandler {
   std::shared_ptr<IRequestsHandler> userRequestsHandler_;
   concord::reconfiguration::Dispatcher reconfig_dispatcher_;
   std::shared_ptr<concord::cron::CronTableRegistry> cron_table_registry_;
-  concord::performance::ISystemResourceEntity &resourceEntity_;
 };
 
 }  // namespace bftEngine

kvbc/include/kvbc_adapter/replica_adapter.hpp

@@ -27,7 +27,6 @@
 #include "db_interfaces.h"
 #include "bcstatetransfer/SimpleBCStateTransfer.hpp"
 #include "state_snapshot_interface.hpp"
-#include "ISystemResourceEntity.hpp"
 #include "replica_adapter_auxilliary_types.hpp"
 #include "categorization/kv_blockchain.h"
 #include "v4blockchain/v4_blockchain.h"
@@ -51,7 +50,7 @@ class ReplicaBlockchain : public IBlocksDeleter,
   /////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
   // IBlocksDeleter implementation
   void deleteGenesisBlock() override final { return deleter_->deleteGenesisBlock(); }
-  BlockId deleteBlocksUntil(BlockId until) override final { return deleter_->deleteBlocksUntil(until); }
+  BlockId deleteBlocksUntil(BlockId until) override final;
   void deleteLastReachableBlock() override final { return deleter_->deleteLastReachableBlock(); }
 
   // Helper method, not part of the interface
@@ -277,6 +276,7 @@ class ReplicaBlockchain : public IBlocksDeleter,
   mutable concordMetrics::GaugeHandle add_block_duration;
   mutable concordMetrics::GaugeHandle multiget_latest_duration;
   mutable concordMetrics::GaugeHandle multiget_version_duration;
+  mutable concordMetrics::GaugeHandle delete_blocks_until_duration;
   mutable concordMetrics::CounterHandle get_counter;
   mutable concordMetrics::CounterHandle multiget_lat_version_counter;
 };

kvbc/include/kvbc_adapter/v4blockchain/blocks_deleter_adapter.hpp

@@ -18,7 +18,6 @@
 #include "db_interfaces.h"
 #include "v4blockchain/v4_blockchain.h"
 #include "kvbc_adapter/replica_adapter_auxilliary_types.hpp"
-#include "resources-manager/ISystemResourceEntity.hpp"
 #include "performance_handler.h"
 #include "diagnostics.h"
 
@@ -27,8 +26,7 @@ namespace concord::kvbc::adapter::v4blockchain {
 class BlocksDeleterAdapter : public IBlocksDeleter {
  public:
   virtual ~BlocksDeleterAdapter() { kvbc_ = nullptr; }
-  explicit BlocksDeleterAdapter(std::shared_ptr<concord::kvbc::v4blockchain::KeyValueBlockchain> &kvbc,
-                                const std::optional<aux::AdapterAuxTypes> &aux_types = std::nullopt);
+  explicit BlocksDeleterAdapter(std::shared_ptr<concord::kvbc::v4blockchain::KeyValueBlockchain> &kvbc);
 
   /////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
   // IBlocksDeleter implementation
@@ -39,7 +37,6 @@ class BlocksDeleterAdapter : public IBlocksDeleter {
 
  private:
   concord::kvbc::v4blockchain::KeyValueBlockchain *kvbc_{nullptr};
-  std::shared_ptr<concord::performance::ISystemResourceEntity> replica_resources_;
 
   struct Recorders {
     static constexpr uint64_t MAX_VALUE_MICROSECONDS = 2ULL * 1000ULL * 1000ULL;  // 2 seconds

kvbc/include/v4blockchain/detail/latest_keys.h

@@ -148,12 +148,15 @@ class LatestKeys {
                 bool* /*value_changed*/) const override;
   };
 
+  void setDeletedKeysMetric(concordMetrics::CounterHandle* m) { deleted_keys_ = m; }
+
  private:
   // This filter is used to delete stale on update keys if their version is smaller than the genesis block
   // It's being called by RocksDB on compaction
 
   std::shared_ptr<concord::storage::rocksdb::NativeClient> native_client_;
   v4blockchain::detail::Categories category_mapping_;
+  concordMetrics::CounterHandle* deleted_keys_{nullptr};
 };
 
 }  // namespace concord::kvbc::v4blockchain::detail

kvbc/include/v4blockchain/v4_blockchain.h

@@ -158,7 +158,9 @@ class KeyValueBlockchain {
   // path - the path of the blockchain db.
   struct BlockchainRecovery {
     BlockchainRecovery(const std::string &path, const std::optional<kvbc::BlockId> &block_id_at_chkpnt);
-    static std::shared_ptr<storage::rocksdb::NativeClient> getRecoveryDB(const std::string &path, bool is_chkpnt);
+    static std::shared_ptr<storage::rocksdb::NativeClient> getRecoveryDB(const std::string &path,
+                                                                         bool is_chkpnt,
+                                                                         bool read_only);
     static void removeRecoveryDB(const std::string &path, bool is_chkpnt);
     static std::string getPath(const std::string &path, bool is_chkpnt);
   };
@@ -267,12 +269,12 @@ class KeyValueBlockchain {
   std::map<kvbc::BlockId, const ::rocksdb::Snapshot *> chkpnt_snap_shots_;
   // const ::rocksdb::Snapshot *chkpoint_snap_shot_{nullptr};
   util::ThreadPool thread_pool_{1};
-  std::optional<kvbc::BlockId> chkpnt_block_id_;
 
   // Metrics
   std::shared_ptr<concordMetrics::Aggregator> aggregator_;
   concordMetrics::Component v4_metrics_comp_;
   concordMetrics::GaugeHandle blocks_deleted_;
+  concordMetrics::CounterHandle deleted_keys_;
 
  public:
   void setAggregator(std::shared_ptr<concordMetrics::Aggregator> aggregator) {

kvbc/src/Replica.cpp

@@ -62,8 +62,7 @@ Status Replica::initInternals() {
   if (replicaConfig_.isReadOnly) {
     LOG_INFO(logger,
              "ReadOnly Replica Status:" << KVLOG(getLastBlockNum(), getLastBlockId(), getLastReachableBlockNum()));
-    auto requestHandler =
-        bftEngine::IRequestsHandler::createRequestsHandler(m_cmdHandler, cronTableRegistry_, replicaResources_);
+    auto requestHandler = bftEngine::IRequestsHandler::createRequestsHandler(m_cmdHandler, cronTableRegistry_);
     requestHandler->setReconfigurationHandler(std::make_shared<pruning::ReadOnlyReplicaPruningHandler>(*this));
     m_replicaPtr = bftEngine::IReplica::createNewRoReplica(
         replicaConfig_, requestHandler, m_stateTransfer, m_ptrComm, m_metadataStorage);
@@ -124,10 +123,9 @@ class KvbcRequestHandler : public bftEngine::RequestHandler {
       const std::shared_ptr<IRequestsHandler> &user_req_handler,
       const std::shared_ptr<concord::cron::CronTableRegistry> &cron_table_registry,
       adapter::ReplicaBlockchain &blockchain,
-      std::shared_ptr<concordMetrics::Aggregator> aggregator_,
-      ISystemResourceEntity &resourceEntity) {
+      std::shared_ptr<concordMetrics::Aggregator> aggregator_) {
     return std::shared_ptr<KvbcRequestHandler>{
-        new KvbcRequestHandler{user_req_handler, cron_table_registry, blockchain, aggregator_, resourceEntity}};
+        new KvbcRequestHandler{user_req_handler, cron_table_registry, blockchain, aggregator_}};
   }
 
  public:
@@ -150,9 +148,8 @@ class KvbcRequestHandler : public bftEngine::RequestHandler {
   KvbcRequestHandler(const std::shared_ptr<IRequestsHandler> &user_req_handler,
                      const std::shared_ptr<concord::cron::CronTableRegistry> &cron_table_registry,
                      adapter::ReplicaBlockchain &blockchain,
-                     std::shared_ptr<concordMetrics::Aggregator> aggregator_,
-                     ISystemResourceEntity &resourceEntity)
-      : bftEngine::RequestHandler(resourceEntity, aggregator_), blockchain_{blockchain} {
+                     std::shared_ptr<concordMetrics::Aggregator> aggregator_)
+      : bftEngine::RequestHandler(aggregator_), blockchain_{blockchain} {
     setUserRequestHandler(user_req_handler);
     setCronTableRegistry(cron_table_registry);
   }
@@ -304,8 +301,7 @@ void Replica::saveReconfigurationCmdToResPages(const std::string &key) {
 
 void Replica::createReplicaAndSyncState() {
   ConcordAssertNE(m_kvBlockchain, nullptr);
-  auto requestHandler =
-      KvbcRequestHandler::create(m_cmdHandler, cronTableRegistry_, *m_kvBlockchain, aggregator_, replicaResources_);
+  auto requestHandler = KvbcRequestHandler::create(m_cmdHandler, cronTableRegistry_, *m_kvBlockchain, aggregator_);
   registerReconfigurationHandlers(requestHandler);
   m_replicaPtr = bftEngine::IReplica::createNewReplica(
       replicaConfig_,

kvbc/src/kvbc_adapter/replica_adapter.cpp

@@ -54,6 +54,7 @@ ReplicaBlockchain::ReplicaBlockchain(
       add_block_duration{metrics_comp_.RegisterGauge("AddBlockDurationMicro", 0)},
       multiget_latest_duration{metrics_comp_.RegisterGauge("MultigetLatestDurationMicro", 0)},
       multiget_version_duration{metrics_comp_.RegisterGauge("MultigetLatestVersionDurationMicro", 0)},
+      delete_blocks_until_duration{metrics_comp_.RegisterGauge("AccumulatedDeleteBlocksUntilDurationMicro", 0)},
       get_counter{metrics_comp_.RegisterCounter("GetCounter")},
       multiget_lat_version_counter{metrics_comp_.RegisterCounter("MultiGetLatestVersionCounter")} {
   metrics_comp_.Register();
@@ -85,7 +86,7 @@ ReplicaBlockchain::ReplicaBlockchain(
     if (aux_types.has_value()) {
       v4_kvbc_->setAggregator(aux_types->aggregator_);
     }
-    up_deleter_ = std::make_unique<concord::kvbc::adapter::v4blockchain::BlocksDeleterAdapter>(v4_kvbc_, aux_types);
+    up_deleter_ = std::make_unique<concord::kvbc::adapter::v4blockchain::BlocksDeleterAdapter>(v4_kvbc_);
     up_reader_ = std::make_unique<concord::kvbc::adapter::v4blockchain::BlocksReaderAdapter>(v4_kvbc_);
     up_adder_ = std::make_unique<concord::kvbc::adapter::v4blockchain::BlocksAdderAdapter>(v4_kvbc_);
     up_app_state_ = std::make_unique<concord::kvbc::adapter::v4blockchain::AppStateAdapter>(v4_kvbc_);
@@ -107,4 +108,14 @@ ReplicaBlockchain::ReplicaBlockchain(
   ConcordAssertNE(db_chkpt_, nullptr);
 }
 
+BlockId ReplicaBlockchain::deleteBlocksUntil(BlockId until) {
+  auto prev_dur = delete_blocks_until_duration.Get().Get();
+  auto start = std::chrono::steady_clock::now();
+  auto id = deleter_->deleteBlocksUntil(until);
+  auto dur = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - start).count();
+  delete_blocks_until_duration.Get().Set(prev_dur + dur);
+  metrics_comp_.UpdateAggregator();
+  return id;
+}
+
 }  // namespace concord::kvbc::adapter