[Metrics] hyperledger-iroha#837 hyperledger-iroha#883 hyperledger-iro…

…ha#919 hyperledger-iroha#896 hyperledger-iroha#835 Integrate kOnBlock to storage_impl.cpp Closes hyperledger-iroha#837 Add total number of blocks Closes hyperledger-iroha#883 Number of domains Closes hyperledger-iroha#919 Number of signatures in the last block Closes hyperledger-iroha#896 Measure total number of transactions Closes hyperledger-iroha#835 Total nodes/peers updated when kOnBlock Should close hyperledger-iroha#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 hyperledger-iroha#966 Disable prometheus-cpp self-metrics Signed-off-by: Ivan Kuvaldin <[email protected]>
kuvaldini · Apr 28, 2021 · d440174 · d440174
1 parent 4569564
commit d440174
Show file tree

Hide file tree

Showing 26 changed files with 579 additions and 120 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
@@ -17,7 +17,7 @@ SET(CMAKE_CXX_STANDARD 17)
 
 SET(CMAKE_POSITION_INDEPENDENT_CODE TRUE)
 if (NOT MSVC)
-  SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -fdiagnostics-color=always")
+  SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -fdiagnostics-color=always -fno-lto")  ##-fno-lto to fix 'bytecode stream version'
   SET(CMAKE_CXX_FLAGS_RELEASE "${CMAKE_CXX_FLAGS_RELEASE} -O3 -Wno-error=deprecated-declarations")
   SET(CMAKE_CXX_FLAGS_DEBUG   "${CMAKE_CXX_FLAGS_DEBUG} -g -Wextra -Wno-unused-parameter -Wno-deprecated-declarations -O0")
 else()

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,33 @@ namespace iroha {
       return getPeersFromSociRowSet(result);
     }
 
+    iroha::expected::Result<size_t, std::string> PostgresWsvQuery::count(
+        std::string_view what) try {
+      int count;
+      sql_ << "SELECT count(*) FROM " << what, soci::into(count);
+      return count;
+    } catch (const std::exception &e) {
+      auto msg =
+          std::string{"Failed to count "}.append(what) + ", query: " + 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");
+    }
+
     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,8 @@ 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);
+
       // 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/storage_impl.cpp b/irohad/ametsuchi/impl/storage_impl.cpp
@@ -5,14 +5,14 @@
 
 #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>
 #include <boost/tuple/tuple.hpp>
+
 #include "ametsuchi/impl/mutable_storage_impl.hpp"
 #include "ametsuchi/impl/peer_query_wsv.hpp"
 #include "ametsuchi/impl/postgres_block_index.hpp"
@@ -36,6 +36,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 +308,11 @@ 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);
+          getSubscription()->notify(EventTypes::kOnBlock, block_ptr);
         }
         return expected::makeValue(std::move(commit_result.ledger_state));
       };
@@ -357,6 +362,9 @@ namespace iroha {
         }
 
         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 +463,9 @@ namespace iroha {
         std::shared_ptr<const shared_model::interface::Block> block) {
       if (block_store_->insert(block)) {
         notifier_.get_subscriber().on_next(block);
+        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?
+      // /**
+      //  * @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
@@ -37,6 +37,7 @@
 #include "main/impl/pg_connection_init.hpp"
 #include "main/impl/storage_init.hpp"
 #include "main/server_runner.hpp"
+#include "main/subscription.hpp"
 #include "multi_sig_transactions/gossip_propagation_strategy.hpp"
 #include "multi_sig_transactions/mst_processor_impl.hpp"
 #include "multi_sig_transactions/mst_propagation_strategy_stub.hpp"
@@ -114,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),
@@ -154,6 +156,7 @@ Irohad::~Irohad() {
   }
   consensus_gate_objects_lifetime.unsubscribe();
   consensus_gate_events_subscription.unsubscribe();
+  se_->dispose();
 }
 
 /**
@@ -1011,7 +1014,11 @@ Irohad::RunResult Irohad::run() {
     storage->on_commit().subscribe(
         ordering_init.commit_notifier.get_subscriber());
 
-    ordering_init.commit_notifier.get_subscriber().on_next(std::move(block));
+    std::shared_ptr<const shared_model::interface::Block> sh_block{
+        std::move(block)};
+    ordering_init.commit_notifier.get_subscriber().on_next(sh_block);
+    getSubscription()->notify(EventTypes::kOnBlock,  // kOnInitialBlock
+                              sh_block);
 
     ordering_init.sync_event_notifier.get_subscriber().on_next(
         synchronizer::SynchronizationEvent{

diff --git a/irohad/main/application.hpp b/irohad/main/application.hpp
@@ -21,6 +21,7 @@
 #include "main/iroha_conf_loader.hpp"
 #include "main/server_runner.hpp"
 #include "main/startup_params.hpp"
+#include "main/subscription.hpp"
 #include "multi_sig_transactions/gossip_propagation_strategy_params.hpp"
 #include "torii/tls_params.hpp"
 
@@ -215,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
@@ -28,11 +28,11 @@
 #include "main/iroha_conf_literals.hpp"
 #include "main/iroha_conf_loader.hpp"
 #include "main/raw_block_loader.hpp"
+#include "maintenance/metrics.hpp"
 #include "network/impl/channel_factory.hpp"
 #include "util/status_notifier.hpp"
 #include "util/utility_service.hpp"
 #include "validators/field_validator.hpp"
-#include "maintenance/metrics.hpp"
 
 #if defined(USE_LIBURSA)
 #include "cryptography/ed25519_ursa_impl/crypto_provider.hpp"
@@ -108,8 +108,12 @@ DEFINE_string(verbosity, kLogSettingsFromConfigFile, "Log verbosity");
 DEFINE_validator(verbosity, &validateVerbosity);
 
 /// Metrics. ToDo validator
-DEFINE_string(metrics_addr, "127.0.0.1", "Prometeus HTTP server listen address");
-DEFINE_string(metrics_port, "", "Prometeus HTTP server listens port, disabled by default");
+DEFINE_string(metrics_addr,
+              "127.0.0.1",
+              "Prometeus HTTP server listen address");
+DEFINE_string(metrics_port,
+              "",
+              "Prometeus HTTP server listens port, disabled by default");
 
 std::sig_atomic_t caught_signal = 0;
 std::promise<void> exit_requested;
@@ -260,12 +264,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);
 
@@ -375,6 +373,14 @@ 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(
@@ -431,6 +437,28 @@ int main(int argc, char *argv[]) {
       return EXIT_FAILURE;
     }
 
+    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());
+      }
+    }
+
     // init pipeline components
     auto init_result = irohad->init();
     if (auto error =

diff --git a/irohad/main/subscription.hpp b/irohad/main/subscription.hpp
@@ -17,6 +17,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
 )