Metrics all in one

.clang-format-ignore

@@ -0,0 +1,2 @@
+CMakeLists.txt
+*.cmake

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

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

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) {

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_;

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

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_;

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");

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

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),

irohad/main/application.hpp

@@ -230,6 +230,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_;

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();

irohad/main/subscription_fwd.hpp

@@ -13,6 +13,7 @@ namespace iroha {
     kYac = 0,
     kRequestProposal,
     kVoteProcess,
+    kMetrics,
     //---------------
     kTotalCount
   };

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
 )