Instrument producer_plugin for Prometheus

plugins/chain_interface/include/eosio/chain/plugin_interface.hpp

@@ -12,7 +12,13 @@ namespace eosio { namespace chain { namespace plugin_interface {
    using namespace eosio::chain;
    using namespace appbase;
 
+   enum class metric_type {
+      gauge = 1,
+      counter = 2
+   };
+
    struct runtime_metric {
+      const metric_type type;
       const std::string family;
       const std::string label;
       std::atomic<int64_t> value;

plugins/chain_plugin/chain_plugin.cpp

@@ -154,6 +154,7 @@ class chain_plugin_impl {
    fc::microseconds                  abi_serializer_max_time_us;
    std::optional<bfs::path>          snapshot_path;
 
+   chain_plugin_metrics              chain_metrics;
 
    // retained references to channels for easy publication
    channels::pre_accepted_block::channel_type&     pre_accepted_block_channel;
@@ -2797,6 +2798,10 @@ fc::variant chain_plugin::get_log_trx(const transaction& trx) const {
     }
     return pretty_output;
 }
+
+const chain_plugin_metrics& chain_plugin::metrics() const {
+   return my->chain_metrics;
+}
 } // namespace eosio
 
 FC_REFLECT( eosio::chain_apis::detail::ram_market_exchange_state_t, (ignore1)(ignore2)(ignore3)(core_symbol)(ignore4) )

plugins/chain_plugin/include/eosio/chain_plugin/chain_plugin.hpp

@@ -832,6 +832,10 @@ class read_write {
 
 } // namespace chain_apis
 
+struct chain_plugin_metrics {
+   //runtime_metric
+};
+
 class chain_plugin : public plugin<chain_plugin> {
 public:
    APPBASE_PLUGIN_REQUIRES()
@@ -878,6 +882,8 @@ class chain_plugin : public plugin<chain_plugin> {
    // return variant of trx for logging, trace is modified to minimize log output
    fc::variant get_log_trx(const transaction& trx) const;
 
+   const chain_plugin_metrics& metrics() const;
+
 private:
    static void log_guard_exception(const chain::guard_exception& e);
 

plugins/net_plugin/include/eosio/net_plugin/net_plugin.hpp

@@ -16,9 +16,22 @@ namespace eosio {
    using namespace plugin_interface;
 
    struct net_plugin_metrics {
-      runtime_metric num_peers{"net_plugin", "num_peers", 0};
-      runtime_metric num_clients{"net_plugin", "num_clients", 0};
-      runtime_metric dropped_trxs{"net_plugin", "dropped_trxs", 0};
+      runtime_metric num_peers{metric_type::gauge, "num_peers", "num_peers", 0};
+      runtime_metric num_clients{metric_type::gauge, "num_clients", "num_clients", 0};
+      runtime_metric dropped_trxs{metric_type::counter, "dropped_trxs", "dropped_trxs", 0};
+
+      std::vector<std::reference_wrapper<runtime_metric>> metrics;
+
+      void enable(bool enabled)  {_enabled = enabled; }
+      bool enabled() {return _enabled;}
+
+      net_plugin_metrics() {
+         metrics.emplace_back(std::ref(num_peers));
+         metrics.emplace_back(std::ref(num_clients));
+         metrics.emplace_back(std::ref(dropped_trxs));
+      }
+   private:
+      bool _enabled = false;
    };
 
    class net_plugin : public appbase::plugin<net_plugin>
@@ -40,7 +53,7 @@ namespace eosio {
         std::optional<connection_status>  status( const string& endpoint )const;
         vector<connection_status>         connections()const;
 
-        std::shared_ptr<net_plugin_metrics> metrics();
+        net_plugin_metrics& metrics();
 
       private:
         std::shared_ptr<class net_plugin_impl> my;

plugins/net_plugin/net_plugin.cpp

@@ -287,7 +287,7 @@ namespace eosio {
 
       uint16_t                                       thread_pool_size = 2;
       std::optional<eosio::chain::named_thread_pool> thread_pool;
-      std::shared_ptr<net_plugin_metrics> metrics;
+      net_plugin_metrics metrics;
 
    private:
       mutable std::mutex            chain_info_mtx; // protects chain_*
@@ -2684,7 +2684,7 @@ namespace eosio {
       shared_ptr<packed_transaction> ptr = std::make_shared<packed_transaction>();
       fc::raw::unpack( ds, *ptr );
       if( trx_in_progress_sz > def_max_trx_in_progress_size) {
-         my_impl->metrics->dropped_trxs.value++;
+         ++my_impl->metrics.dropped_trxs.value;
          char reason[72];
          snprintf(reason, 72, "Dropping trx, too many trx in progress %lu bytes", trx_in_progress_sz);
          my_impl->producer_plug->log_failed_transaction(ptr->id(), ptr, reason);
@@ -3348,8 +3348,8 @@ namespace eosio {
       }
       g.unlock();
 
-      metrics->num_clients.value = num_clients;
-      metrics->num_peers.value = num_peers;
+      metrics.num_clients.value = num_clients;
+      metrics.num_peers.value = num_peers;
 
       if( num_clients > 0 || num_peers > 0 )
          fc_ilog( logger, "p2p client connections: ${num}/${max}, peer connections: ${pnum}/${pmax}",
@@ -3678,8 +3678,6 @@ namespace eosio {
             my->chain_plug->enable_accept_transactions();
          }
 
-         my->metrics = std::make_shared<net_plugin_metrics>();
-
       } FC_LOG_AND_RETHROW()
    }
 
@@ -3832,7 +3830,7 @@ namespace eosio {
       FC_CAPTURE_AND_RETHROW()
    }
 
-   std::shared_ptr<net_plugin_metrics> net_plugin::metrics() {
+   net_plugin_metrics& net_plugin::metrics() {
       return my->metrics;
    }
 

plugins/producer_plugin/include/eosio/producer_plugin/producer_plugin.hpp

@@ -9,6 +9,39 @@ namespace eosio {
 
 using boost::signals2::signal;
 
+using chain::plugin_interface::runtime_metric;
+using chain::plugin_interface::metric_type;
+
+struct producer_plugin_metrics {
+   runtime_metric unapplied_transactions{metric_type::gauge, "unapplied_transactions", "unapplied_transactions", 0};
+   runtime_metric blacklisted_transactions{metric_type::gauge, "blacklisted_transactions", "blacklisted_transactions", 0};
+   runtime_metric blocks_produced{metric_type::counter, "blocks_produced", "blocks_produced", 0};
+   runtime_metric trxs_produced{metric_type::counter, "trxs_produced", "trxs_produced", 0};
+   runtime_metric last_irreversible{metric_type::gauge, "last_irreversible", "last_irreversible", 0};
+   runtime_metric block_num{metric_type::gauge, "block_num", "block_num", 0};
+   runtime_metric subjective_bill_account_size{metric_type::gauge, "subjective_bill_account_size", "subjective_bill_account_size", 0};
+   runtime_metric subjective_bill_block_size{metric_type::gauge, "subjective_bill_block_size", "subjective_bill_block_size", 0};
+   runtime_metric scheduled_trxs{metric_type::gauge, "scheduled_trxs", "scheduled_trxs", 0};
+
+   std::vector<std::reference_wrapper<runtime_metric>> metrics;
+   void enable(bool enabled)  {_enabled = enabled; }
+   bool enabled() {return _enabled;}
+
+   producer_plugin_metrics() {
+      metrics.emplace_back(std::ref(unapplied_transactions));
+      metrics.emplace_back(std::ref(blacklisted_transactions));
+      metrics.emplace_back(std::ref(blocks_produced));
+      metrics.emplace_back(std::ref(trxs_produced));
+      metrics.emplace_back(std::ref(last_irreversible));
+      metrics.emplace_back(std::ref(block_num));
+      metrics.emplace_back(std::ref(subjective_bill_account_size));
+      metrics.emplace_back(std::ref(subjective_bill_block_size));
+      metrics.emplace_back(std::ref(scheduled_trxs));
+   }
+private:
+   bool _enabled = false;
+};
+
 class producer_plugin : public appbase::plugin<producer_plugin> {
 public:
    APPBASE_PLUGIN_REQUIRES((chain_plugin)(signature_provider_plugin))
@@ -143,6 +176,7 @@ class producer_plugin : public appbase::plugin<producer_plugin> {
 
 
    void log_failed_transaction(const transaction_id_type& trx_id, const chain::packed_transaction_ptr& packed_trx_ptr, const char* reason) const;
+   producer_plugin_metrics& metrics();
 
  private:
    std::shared_ptr<class producer_plugin_impl> my;

plugins/producer_plugin/include/eosio/producer_plugin/subjective_billing.hpp

@@ -101,7 +101,8 @@ class subjective_billing {
 
 public: // public for tests
    static constexpr uint32_t subjective_time_interval_ms = 5'000;
-
+   size_t get_account_cache_size() {return _account_subjective_bill_cache.size();}
+   size_t get_block_cache_size() {return _block_subjective_bill_cache.size();}
    void remove_subjective_billing( const transaction_id_type& trx_id, uint32_t time_ordinal ) {
       auto& idx = _trx_cache_index.get<by_id>();
       auto itr = idx.find( trx_id );

plugins/producer_plugin/producer_plugin.cpp

@@ -342,6 +342,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
       std::optional<scoped_connection>                          _accepted_block_header_connection;
       std::optional<scoped_connection>                          _irreversible_block_connection;
 
+      producer_plugin_metrics                                   _metrics;
       /*
        * HACK ALERT
        * Boost timers can be in a state where a handler has not yet executed but is not abortable.
@@ -410,6 +411,21 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
          }
       }
 
+      void update_block_metrics() {
+         _metrics.unapplied_transactions.value =  _unapplied_transactions.size();
+         _metrics.subjective_bill_account_size.value = _subjective_billing.get_account_cache_size();
+         _metrics.subjective_bill_block_size.value = _subjective_billing.get_block_cache_size();
+         _metrics.blacklisted_transactions.value = _blacklisted_transactions.size();
+         _metrics.unapplied_transactions.value = _unapplied_transactions.size();
+
+         auto& chain = chain_plug->chain();
+         _metrics.last_irreversible.value = chain.last_irreversible_block_num();
+         _metrics.block_num.value = chain.head_block_num();
+
+         const auto& sch_idx = chain.db().get_index<generated_transaction_multi_index,by_delay>();
+         _metrics.scheduled_trxs.value = sch_idx.size();
+      }
+
       void abort_block() {
          auto& chain = chain_plug->chain();
 
@@ -1577,6 +1593,9 @@ fc::time_point producer_plugin_impl::calculate_block_deadline( const fc::time_po
 
 producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
    chain::controller& chain = chain_plug->chain();
+   if (_metrics.enabled()) {
+      update_block_metrics();
+   }
 
    if( !chain_plug->accept_transactions() )
       return start_block_result::waiting_for_block;
@@ -2486,6 +2505,9 @@ void producer_plugin_impl::produce_block() {
 
    br.total_time += fc::time_point::now() - start;
 
+   ++_metrics.blocks_produced.value;
+   _metrics.trxs_produced.value += new_bs->block->transactions.size();
+
    ilog("Produced block ${id}... #${n} @ ${t} signed by ${p} "
         "[trxs: ${count}, lib: ${lib}, confirmed: ${confs}, net: ${net}, cpu: ${cpu}, elapsed: ${et}, time: ${tt}]",
         ("p",new_bs->header.producer)("id",new_bs->id.str().substr(8,16))
@@ -2504,4 +2526,7 @@ void producer_plugin::log_failed_transaction(const transaction_id_type& trx_id,
             ("entire_trx", packed_trx_ptr ? my->chain_plug->get_log_trx(packed_trx_ptr->get_transaction()) : fc::variant{trx_id}));
 }
 
+producer_plugin_metrics& producer_plugin::metrics() {
+   return my->_metrics;
+}
 } // namespace eosio

plugins/prometheus_plugin/include/eosio/prometheus_plugin/prometheus_plugin.hpp

@@ -4,6 +4,7 @@
 #include <eosio/http_plugin/http_plugin.hpp>
 #include <eosio/chain_plugin/chain_plugin.hpp>
 #include <eosio/net_plugin/net_plugin.hpp>
+#include <eosio/producer_plugin/producer_plugin.hpp>
 
 namespace eosio {
 
@@ -14,7 +15,7 @@ namespace eosio {
       prometheus_plugin();
       ~prometheus_plugin();
 
-      APPBASE_PLUGIN_REQUIRES((http_plugin) (chain_plugin) (net_plugin))
+      APPBASE_PLUGIN_REQUIRES((http_plugin))
 
       virtual void set_program_options(options_description&, options_description& cfg) override;
 

plugins/prometheus_plugin/prometheus_plugin.cpp

@@ -61,37 +61,73 @@ namespace eosio {
          const prometheus::TextSerializer _serializer;
          std::shared_ptr<Registry> _registry;
 
-         std::vector<std::tuple<Family<Gauge>&, Gauge&, runtime_metric&>> _runtime_metrics;
+
+         std::vector<std::tuple<Family<Gauge>&, Gauge&, const runtime_metric&>> _gauges;
+         std::vector<std::tuple<Family<Counter>&, Counter&, const runtime_metric&>> _counters;
 
          // metrics for prometheus_plugin itself
          std::unique_ptr<prometheus_plugin_metrics> _metrics;
 
          // hold onto other plugin metrics
-         std::shared_ptr<net_plugin_metrics> net_plugin_metrics_ptr;
-
          prometheus_plugin_impl() { }
 
-         void add_plugin_metric(runtime_metric& plugin_metric) {
+         void add_gauge_metric(const runtime_metric& plugin_metric) {
             auto& gauge_family = BuildGauge()
                   .Name(plugin_metric.family)
                   .Help("")
                   .Register(*_registry);
             auto& gauge = gauge_family.Add({});
-            _runtime_metrics.push_back(std::tuple<Family<Gauge>&, Gauge&, runtime_metric&>(gauge_family, gauge, plugin_metric));
-            ilog("Added metric ${f}:${l}", ("f", plugin_metric.family) ("l", plugin_metric.label));
+
+            _gauges.push_back(
+                  std::tuple<Family<Gauge>&, Gauge&, const runtime_metric&>(gauge_family, gauge, plugin_metric));
+
+            ilog("Added gauge metric ${f}:${l}", ("f", plugin_metric.family) ("l", plugin_metric.label));
          }
 
-         void add_plugin_metrics(std::shared_ptr<net_plugin_metrics> metrics) {
-            add_plugin_metric(metrics->num_clients);
-            add_plugin_metric(metrics->num_peers);
-            add_plugin_metric(metrics->dropped_trxs);
+         void add_counter_metric(const runtime_metric& plugin_metric) {
+                  auto &counter_family = BuildCounter()
+                        .Name(plugin_metric.family)
+                        .Help("")
+                        .Register(*_registry);
+                  auto &counter = counter_family.Add({});
+                  _counters.push_back(
+                        std::tuple<Family<Counter>&, Counter&, const runtime_metric&>(counter_family, counter, plugin_metric));
+
+            ilog("Added counter metric ${f}:${l}", ("f", plugin_metric.family) ("l", plugin_metric.label));
+         }
+
+         void add_plugin_metric(const runtime_metric& plugin_metric) {
+            switch(plugin_metric.type) {
+               case metric_type::gauge:
+                  add_gauge_metric(plugin_metric);
+                  break;
+               case metric_type::counter:
+                  add_counter_metric(plugin_metric);
+                  break;
+
+               default:
+                  break;
+            }
+         }
+
+         template <typename T>
+         void add_plugin_metrics(T& pm) {
+            pm.enable(true);
+            for (const auto& m : pm.metrics) {
+               add_plugin_metric(m);
+            }
          }
 
          void update_plugin_metrics() {
-            for (auto& rtm : _runtime_metrics) {
+            for (auto& rtm : _gauges) {
                auto new_val = static_cast<double>(std::get<2>(rtm).value);
                std::get<1>(rtm).Set(new_val);
             }
+
+            for (auto& rtm : _counters) {
+               auto new_val = static_cast<double>(std::get<2>(rtm).value);
+               std::get<1>(rtm).Increment(new_val-std::get<1>(rtm).Value());
+            }
          }
 
          void initialize_metrics() {
@@ -103,11 +139,17 @@ namespace eosio {
 
             net_plugin* np = app().find_plugin<net_plugin>();
             if (nullptr != np) {
-               net_plugin_metrics_ptr = np->metrics();
-               add_plugin_metrics(net_plugin_metrics_ptr);
+               add_plugin_metrics<net_plugin_metrics>(np->metrics());
             } else {
                dlog("net_plugin not found -- metrics not added");
             }
+
+            producer_plugin* pp = app().find_plugin<producer_plugin>();
+            if (nullptr != pp) {
+               add_plugin_metrics<producer_plugin_metrics>(pp->metrics());
+            } else {
+               dlog("producer_plugin not found -- metrics not added");
+            }
          }
 
          std::string scrape() {