Metrics all in one (#943)

[Metrics] #837 #883 #919 #896 #835 Integrate kOnBlock to storage_impl.cpp Closes #837 Add total number of blocks Closes #883 Number of domains Closes #919 Number of signatures in the last block Closes #896 Measure total number of transactions Closes #835 Total nodes/peers updated when kOnBlock Should close #835 no online/connected peers parameter in this commit. Add CMake files to .clang-format-ignore Disable prometheus-cpp logs Extend WSV to countDomains(), countTransactions(), countPeers() -fno-lto to link GCC-10 and clang with civetweb which was built with GCC-9 Tests dependant on ametsuchi are linked against sync_subscription, not async Closes #966 Disable prometheus-cpp self-metrics Signed-off-by: Ivan 'kuvaldini' Kuvaldin <ivan@kuvaldini.pro>
hyperledger · Jul 14, 2021 · 15b9a0d · 15b9a0d
1 parent bfe93fa
commit 15b9a0d
Show file tree

Hide file tree

Showing 33 changed files with 595 additions and 145 deletions.
diff --git a/.clang-format-ignore b/.clang-format-ignore
@@ -0,0 +1,2 @@
+CMakeLists.txt
+*.cmake
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -39,6 +39,11 @@ if(CMAKE_BUILD_TYPE MATCHES "Release")
   endif()
 endif()
 
+## FIXME revert this after change CI to GitHub actions.
+## Temporal fix for 'bytecode stream version incompatible' between gcc-9 and gcc-10 and clang
+## when dependancies were build via vcpkg with default GCC9 could not be linked with iroha built with GCC-10
+set(CMAKE_INTERPROCEDURAL_OPTIMIZATION FALSE)
+
 if(WIN32)
     # We have to set _WIN32_WINNT for gRPC
     if(${CMAKE_SYSTEM_VERSION} EQUAL 10) # Windows 10

diff --git a/irohad/ametsuchi/CMakeLists.txt b/irohad/ametsuchi/CMakeLists.txt
@@ -187,6 +187,7 @@ target_link_libraries(ametsuchi
     SOCI::postgresql
     SOCI::core
     postgres_query_executor
+    async_subscription
     )
 
 target_compile_definitions(ametsuchi

diff --git a/irohad/ametsuchi/impl/postgres_wsv_query.cpp b/irohad/ametsuchi/impl/postgres_wsv_query.cpp
@@ -6,6 +6,7 @@
 #include "ametsuchi/impl/postgres_wsv_query.hpp"
 
 #include <soci/boost-tuple.h>
+
 #include "ametsuchi/impl/soci_std_optional.hpp"
 #include "ametsuchi/impl/soci_utils.hpp"
 #include "ametsuchi/ledger_state.hpp"
@@ -79,6 +80,35 @@ namespace iroha {
       return getPeersFromSociRowSet(result);
     }
 
+    iroha::expected::Result<size_t, std::string> PostgresWsvQuery::count(
+        std::string_view table, std::string_view column /* ="*" */) try {
+      int count;
+      sql_ << "SELECT count(" << column << ") FROM " << table,
+          soci::into(count);
+      return count;
+    } catch (const std::exception &e) {
+      auto msg = fmt::format("Failed to count {}, query: {}", table, e.what());
+      log_->error(msg);
+      return iroha::expected::makeError(msg);
+    }
+
+    iroha::expected::Result<size_t, std::string>
+    PostgresWsvQuery::countPeers() {
+      return count("peer");
+    }
+
+    iroha::expected::Result<size_t, std::string>
+    PostgresWsvQuery::countDomains() {
+      return count("domain");
+    }
+
+    iroha::expected::Result<size_t, std::string>
+    PostgresWsvQuery::countTransactions() {
+      return count("tx_positions", "DISTINCT hash");
+      // OR return count("tx_status_from_hash", "*", "WHERE status=true");
+      // //select count(*) from tx_status_by_hash where status=true
+    }
+
     boost::optional<std::shared_ptr<shared_model::interface::Peer>>
     PostgresWsvQuery::getPeerByPublicKey(
         shared_model::interface::types::PublicKeyHexStringView public_key) {

diff --git a/irohad/ametsuchi/impl/postgres_wsv_query.hpp b/irohad/ametsuchi/impl/postgres_wsv_query.hpp
@@ -6,9 +6,9 @@
 #ifndef IROHA_POSTGRES_WSV_QUERY_HPP
 #define IROHA_POSTGRES_WSV_QUERY_HPP
 
-#include "ametsuchi/wsv_query.hpp"
-
 #include <soci/soci.h>
+
+#include "ametsuchi/wsv_query.hpp"
 #include "logger/logger_fwd.hpp"
 
 namespace iroha {
@@ -28,6 +28,10 @@ namespace iroha {
           std::vector<std::shared_ptr<shared_model::interface::Peer>>>
       getPeers() override;
 
+      iroha::expected::Result<size_t, std::string> countPeers() override;
+      iroha::expected::Result<size_t, std::string> countDomains() override;
+      iroha::expected::Result<size_t, std::string> countTransactions() override;
+
       boost::optional<std::shared_ptr<shared_model::interface::Peer>>
       getPeerByPublicKey(shared_model::interface::types::PublicKeyHexStringView
                              public_key) override;
@@ -43,6 +47,9 @@ namespace iroha {
       template <typename T, typename F>
       auto execute(F &&f) -> boost::optional<soci::rowset<T>>;
 
+      iroha::expected::Result<size_t, std::string> count(
+          std::string_view, std::string_view column = "*");
+
       // TODO andrei 24.09.2018: IR-1718 Consistent soci::session fields in
       // storage classes
       std::unique_ptr<soci::session> psql_;

diff --git a/irohad/ametsuchi/impl/rocksdb_wsv_query.cpp b/irohad/ametsuchi/impl/rocksdb_wsv_query.cpp
@@ -186,4 +186,17 @@ namespace iroha::ametsuchi {
     }
   }
 
+  iroha::expected::Result<size_t, std::string> RocksDBWsvQuery::countPeers(){
+    return iroha::expected::makeError("unimplemented yet");
+  }
+
+  iroha::expected::Result<size_t, std::string> RocksDBWsvQuery::countDomains(){
+      return iroha::expected::makeError("unimplemented yet");
+  }
+
+  iroha::expected::Result<size_t, std::string> 
+      RocksDBWsvQuery::countTransactions() {
+    return iroha::expected::makeError("unimplemented yet");
+  }
+
 }  // namespace iroha::ametsuchi
diff --git a/irohad/ametsuchi/impl/rocksdb_wsv_query.hpp b/irohad/ametsuchi/impl/rocksdb_wsv_query.hpp
@@ -33,6 +33,10 @@ namespace iroha {
       iroha::expected::Result<iroha::TopBlockInfo, std::string>
       getTopBlockInfo() const override;
 
+      iroha::expected::Result<size_t, std::string> countPeers() override;
+      iroha::expected::Result<size_t, std::string> countDomains() override;
+      iroha::expected::Result<size_t, std::string> countTransactions() override;
+
      private:
       std::shared_ptr<RocksDBPort> db_port_;
       std::shared_ptr<RocksDBContext> db_context_;

diff --git a/irohad/ametsuchi/impl/storage_impl.cpp b/irohad/ametsuchi/impl/storage_impl.cpp
@@ -5,10 +5,9 @@
 
 #include "ametsuchi/impl/storage_impl.hpp"
 
-#include <utility>
-
 #include <soci/callbacks.h>
 #include <soci/postgresql/soci-postgresql.h>
+
 #include <boost/algorithm/string.hpp>
 #include <boost/format.hpp>
 #include <boost/range/algorithm/replace_if.hpp>
@@ -36,6 +35,7 @@
 #include "logger/logger.hpp"
 #include "logger/logger_manager.hpp"
 #include "main/impl/pg_connection_init.hpp"
+#include "main/subscription.hpp"
 
 namespace iroha {
   namespace ametsuchi {
@@ -307,7 +307,12 @@ namespace iroha {
           if (not maybe_block) {
             return fmt::format("Failed to fetch block {}", height);
           }
-          notifier_.get_subscriber().on_next(*std::move(maybe_block));
+
+          std::shared_ptr<const shared_model::interface::Block> block_ptr =
+              std::move(maybe_block.get());
+          notifier_.get_subscriber().on_next(block_ptr);
+          log_->info("StorageImpl::commit()  notify(EventTypes::kOnBlock)");
+          getSubscription()->notify(EventTypes::kOnBlock, block_ptr);
         }
         return expected::makeValue(std::move(commit_result.ledger_state));
       };
@@ -356,7 +361,11 @@ namespace iroha {
           throw std::runtime_error(e.value());
         }
 
+        log_->info("StorageImpl::commitPrepared()  notify(EventTypes::kOnBlock)");
         notifier_.get_subscriber().on_next(block);
+        getSubscription()->notify(
+            EventTypes::kOnBlock,
+            std::shared_ptr<const shared_model::interface::Block>(block));
 
         decltype(std::declval<PostgresWsvQuery>().getPeers()) opt_ledger_peers;
         {
@@ -455,6 +464,10 @@ namespace iroha {
         std::shared_ptr<const shared_model::interface::Block> block) {
       if (block_store_->insert(block)) {
         notifier_.get_subscriber().on_next(block);
+        log_->info("StorageImpl::storeBlock()  notify(EventTypes::kOnBlock)");
+        getSubscription()->notify(
+            EventTypes::kOnBlock,
+            std::shared_ptr<const shared_model::interface::Block>(block));
         return {};
       }
       return expected::makeError("Block insertion to storage failed");

diff --git a/irohad/ametsuchi/wsv_query.hpp b/irohad/ametsuchi/wsv_query.hpp
@@ -6,9 +6,9 @@
 #ifndef IROHA_WSV_QUERY_HPP
 #define IROHA_WSV_QUERY_HPP
 
+#include <boost/optional.hpp>
 #include <vector>
 
-#include <boost/optional.hpp>
 #include "common/result.hpp"
 #include "interfaces/common_objects/peer.hpp"
 #include "interfaces/common_objects/string_view_types.hpp"
@@ -40,6 +40,35 @@ namespace iroha {
           std::vector<std::shared_ptr<shared_model::interface::Peer>>>
       getPeers() = 0;
 
+      // ToDo?(kuvaldini,iceseer) #997
+      // /**
+      //  * @brief Fetch domains stored in ledger
+      //  * @return list of domains in insertion to ledger order
+      //  */
+      // virtual iroha::expected::Result<
+      //   std::vector<std::shared_ptr<shared_model::interface::Domain>>,
+      //   std::string>
+      // getDomains() = 0;
+
+      /**
+       * @brief Fetch number of domains in ledger
+       * @return number of domains in ledger
+       */
+      virtual iroha::expected::Result<size_t, std::string> countPeers() = 0;
+
+      /**
+       * @brief Fetch number of domains in ledger
+       * @return number of domains in ledger
+       */
+      virtual iroha::expected::Result<size_t, std::string> countDomains() = 0;
+
+      /**
+       * @brief Fetch number of valid transactions in ledger
+       * @return number of transactions in ledger
+       */
+      virtual iroha::expected::Result<size_t, std::string>
+      countTransactions() = 0;
+
       /**
        * Fetch peer with given public key from ledger
        * @return the peer if found, none otherwise

diff --git a/irohad/main/application.cpp b/irohad/main/application.cpp
@@ -115,7 +115,8 @@ Irohad::Irohad(
     const boost::optional<GossipPropagationStrategyParams>
         &opt_mst_gossip_params,
     boost::optional<IrohadConfig::InterPeerTls> inter_peer_tls_config)
-    : config_(config),
+    : se_(getSubscription()),
+      config_(config),
       listen_ip_(listen_ip),
       keypair_(keypair),
       startup_wsv_sync_policy_(startup_wsv_sync_policy),

diff --git a/irohad/main/application.hpp b/irohad/main/application.hpp
@@ -216,6 +216,7 @@ class Irohad {
   virtual RunResult initWsvRestorer();
 
   // constructor dependencies
+  std::shared_ptr<iroha::Subscription> se_;
   IrohadConfig config_;
   const std::string listen_ip_;
   boost::optional<shared_model::crypto::Keypair> keypair_;

diff --git a/irohad/main/irohad.cpp b/irohad/main/irohad.cpp
@@ -259,12 +259,6 @@ int main(int argc, char *argv[]) {
           log_manager);
     }
 
-    if (FLAGS_metrics_port.size()) {
-      maintenance_metrics_init(FLAGS_metrics_addr + ":" + FLAGS_metrics_port);
-    } else if (config.metrics_addr_port.size()) {
-      maintenance_metrics_init(config.metrics_addr_port);
-    }
-
     daemon_status_notifier->notify(
         ::iroha::utility_service::Status::kInitialization);
 
@@ -374,10 +368,18 @@ int main(int argc, char *argv[]) {
 
         // clear previous storage if any
         irohad->dropStorage();
+        // Check if iroha daemon storage was successfully re-initialized
+        if (not irohad->storage) {
+          // Abort execution if not
+          log->error("Failed to re-initialize storage");
+          daemon_status_notifier->notify(
+              ::iroha::utility_service::Status::kFailed);
+          return EXIT_FAILURE;
+        }
 
         const auto txs_num = block->transactions().size();
-        if (auto e = iroha::expected::resultToOptionalError(
-                irohad->storage->insertBlock(std::move(block)))) {
+        auto inserted = irohad->storage->insertBlock(std::move(block));
+        if (auto e = iroha::expected::resultToOptionalError(inserted)) {
           log->critical("Could not apply genesis block: {}", e.value());
           return EXIT_FAILURE;
         }
@@ -391,19 +393,17 @@ int main(int argc, char *argv[]) {
             "genesis block is provided. Please pecify new genesis block using "
             "--genesis_block parameter.");
         return EXIT_FAILURE;
-      } else {
-        if (overwrite) {
-          // no genesis, blockstore present, overwrite specified -> new block
-          // store, world state should be reset
-          irohad->resetWsv();
-          if (not FLAGS_reuse_state) {
-            log->warn(
-                "No new genesis block is specified - blockstore will not be "
-                "overwritten. If you want overwrite ledger state, please "
-                "specify new genesis block using --genesis_block parameter. "
-                "If you want to reuse existing state data (WSV), consider the "
-                "--reuse_state flag.");
-          }
+      } else if (overwrite) {
+        // no genesis, blockstore present, overwrite specified -> new block
+        // store, world state should be reset
+        irohad->resetWsv();
+        if (not FLAGS_reuse_state) {
+          log->warn(
+              "No new genesis block is specified - blockstore will not be "
+              "overwritten. If you want overwrite ledger state, please "
+              "specify new genesis block using --genesis_block parameter. "
+              "If you want to reuse existing state data (WSV), consider the "
+              "--reuse_state flag.");
         }
       }
     }
@@ -446,6 +446,29 @@ int main(int argc, char *argv[]) {
     std::signal(SIGQUIT, handler);
 #endif
 
+    // start metrics
+    std::shared_ptr<Metrics> metrics;  // Must be a pointer because 'this' is
+                                       // captured to lambdas in constructor.
+    std::string metrics_addr;
+    if (FLAGS_metrics_port.size()) {
+      metrics_addr = FLAGS_metrics_addr + ":" + FLAGS_metrics_port;
+    } else if (config.metrics_addr_port.size()) {
+      metrics_addr = config.metrics_addr_port;
+    }
+    if (metrics_addr.empty()) {
+      log->info("Skiping Metrics initialization.");
+    } else {
+      try {
+        metrics =
+            Metrics::create(metrics_addr,
+                            irohad->storage,
+                            log_manager->getChild("Metrics")->getLogger());
+        log->info("Metrics listens on {}", metrics->getListenAddress());
+      } catch (std::exception const &ex) {
+        log->warn("Failed to initialize Metrics: {}", ex.what());
+      }
+    }
+
     // runs iroha
     log->info("Running iroha");
     auto run_result = irohad->run();

diff --git a/irohad/main/subscription_fwd.hpp b/irohad/main/subscription_fwd.hpp
@@ -13,6 +13,7 @@ namespace iroha {
     kYac = 0,
     kRequestProposal,
     kVoteProcess,
+    kMetrics,
     //---------------
     kTotalCount
   };

diff --git a/irohad/maintenance/CMakeLists.txt b/irohad/maintenance/CMakeLists.txt
@@ -6,4 +6,5 @@
 add_library(maintenance metrics.cpp)
 target_link_libraries(maintenance
     prometheus-cpp::core prometheus-cpp::pull
+    async_subscription
 )