feat: add prometheus monitoring (#356)

* feat: add prometheus monitoring

* chore: progress

* fix: create object

* chore: port config

* chore: set initial metrics on start

* fix: monitor predicates

* test: add

Author: rafaelcr

Cargo.lock

@@ -37,6 +37,7 @@ use ordhook::scan::bitcoin::scan_bitcoin_chainstate_via_rpc_using_predicate;
 use ordhook::service::observers::initialize_observers_db;
 use ordhook::service::{start_observer_forwarding, Service};
 use ordhook::utils::bitcoind::bitcoind_get_block_height;
+use ordhook::utils::monitoring::PrometheusMonitoring;
 use ordhook::{hex, initialize_databases, try_error, try_info, try_warn};
 use reqwest::Client as HttpClient;
 use std::collections::HashSet;
@@ -883,8 +884,12 @@ async fn handle_command(opts: Opts, ctx: &Context) -> Result<(), String> {
                     _ => None,
                 };
                 let blocks = cmd.get_blocks();
-                let inscription_indexing_processor =
-                    start_inscription_indexing_processor(&config, ctx, block_post_processor);
+                let inscription_indexing_processor = start_inscription_indexing_processor(
+                    &config,
+                    ctx,
+                    block_post_processor,
+                    &PrometheusMonitoring::new(),
+                );
 
                 download_and_pipeline_blocks(
                     &config,

components/ordhook-cli/src/cli/mod.rs

@@ -1,13 +1,12 @@
-use ordhook::chainhook_sdk::indexer::IndexerConfig;
 use ordhook::chainhook_sdk::observer::DEFAULT_INGESTION_PORT;
 use ordhook::chainhook_sdk::types::{
     BitcoinBlockSignaling, BitcoinNetwork, StacksNetwork, StacksNodeConfig,
 };
 use ordhook::config::{
-    Config, LogConfig, MetaProtocolsConfig, PredicatesApi, PredicatesApiConfig, ResourcesConfig,
-    SnapshotConfig, SnapshotConfigDownloadUrls, StorageConfig, DEFAULT_BITCOIND_RPC_THREADS,
-    DEFAULT_BITCOIND_RPC_TIMEOUT, DEFAULT_BRC20_LRU_CACHE_SIZE, DEFAULT_CONTROL_PORT,
-    DEFAULT_MEMORY_AVAILABLE, DEFAULT_ULIMIT,
+    Config, IndexerConfig, LogConfig, MetaProtocolsConfig, PredicatesApi, PredicatesApiConfig,
+    ResourcesConfig, SnapshotConfig, SnapshotConfigDownloadUrls, StorageConfig,
+    DEFAULT_BITCOIND_RPC_THREADS, DEFAULT_BITCOIND_RPC_TIMEOUT, DEFAULT_BRC20_LRU_CACHE_SIZE,
+    DEFAULT_CONTROL_PORT, DEFAULT_MEMORY_AVAILABLE, DEFAULT_ULIMIT,
 };
 use std::fs::File;
 use std::io::{BufReader, Read};
@@ -43,7 +42,7 @@ impl ConfigFile {
     }
 
     pub fn from_config_file(config_file: ConfigFile) -> Result<Config, String> {
-        let (stacks_network, bitcoin_network) = match config_file.network.mode.as_str() {
+        let (_, bitcoin_network) = match config_file.network.mode.as_str() {
             "devnet" => (StacksNetwork::Devnet, BitcoinNetwork::Regtest),
             "testnet" => (StacksNetwork::Testnet, BitcoinNetwork::Testnet),
             "mainnet" => (StacksNetwork::Mainnet, BitcoinNetwork::Mainnet),
@@ -115,14 +114,11 @@ impl ConfigFile {
                 bitcoin_block_signaling: match config_file.network.bitcoind_zmq_url {
                     Some(ref zmq_url) => BitcoinBlockSignaling::ZeroMQ(zmq_url.clone()),
                     None => BitcoinBlockSignaling::Stacks(StacksNodeConfig::default_localhost(
-                        config_file
-                            .network
-                            .stacks_events_ingestion_port
-                            .unwrap_or(DEFAULT_INGESTION_PORT),
+                        DEFAULT_INGESTION_PORT,
                     )),
                 },
-                stacks_network,
                 bitcoin_network,
+                prometheus_monitoring_port: config_file.network.prometheus_monitoring_port,
             },
             logs: LogConfig {
                 ordinals_internals: config_file
@@ -220,6 +216,5 @@ pub struct NetworkConfigFile {
     pub bitcoind_rpc_username: String,
     pub bitcoind_rpc_password: String,
     pub bitcoind_zmq_url: Option<String>,
-    pub stacks_node_rpc_url: Option<String>,
-    pub stacks_events_ingestion_port: Option<u16>,
+    pub prometheus_monitoring_port: Option<u16>,
 }

components/ordhook-cli/src/config/file.rs

@@ -46,6 +46,7 @@ hyper = { version = "=0.14.27" }
 lazy_static = { version = "1.4.0" }
 ciborium = "0.2.1"
 regex = "1.10.3"
+prometheus = "0.13.3"
 
 [dev-dependencies]
 test-case = "3.1.0"

components/ordhook-core/Cargo.toml

@@ -1,5 +1,4 @@
 use crate::core::OrdhookConfig;
-pub use chainhook_sdk::indexer::IndexerConfig;
 use chainhook_sdk::observer::EventObserverConfig;
 use chainhook_sdk::types::{
     BitcoinBlockSignaling, BitcoinNetwork, StacksNetwork, StacksNodeConfig,
@@ -81,6 +80,16 @@ pub struct UrlConfig {
     pub file_url: String,
 }
 
+#[derive(Debug, Clone)]
+pub struct IndexerConfig {
+    pub bitcoin_network: BitcoinNetwork,
+    pub bitcoind_rpc_url: String,
+    pub bitcoind_rpc_username: String,
+    pub bitcoind_rpc_password: String,
+    pub bitcoin_block_signaling: BitcoinBlockSignaling,
+    pub prometheus_monitoring_port: Option<u16>,
+}
+
 #[derive(Deserialize, Debug, Clone)]
 pub struct ResourcesConfig {
     pub ulimit: usize,
@@ -136,7 +145,7 @@ impl Config {
             display_logs: false,
             cache_path: self.storage.working_dir.clone(),
             bitcoin_network: self.network.bitcoin_network.clone(),
-            stacks_network: self.network.stacks_network.clone(),
+            stacks_network: StacksNetwork::Devnet,
             prometheus_monitoring_port: None,
             data_handler_tx: None,
         }
@@ -192,8 +201,8 @@ impl Config {
                 bitcoin_block_signaling: BitcoinBlockSignaling::Stacks(
                     StacksNodeConfig::default_localhost(DEFAULT_INGESTION_PORT),
                 ),
-                stacks_network: StacksNetwork::Devnet,
                 bitcoin_network: BitcoinNetwork::Regtest,
+                prometheus_monitoring_port: None,
             },
             logs: LogConfig {
                 ordinals_internals: true,
@@ -227,8 +236,8 @@ impl Config {
                 bitcoin_block_signaling: BitcoinBlockSignaling::Stacks(
                     StacksNodeConfig::default_localhost(DEFAULT_INGESTION_PORT),
                 ),
-                stacks_network: StacksNetwork::Testnet,
                 bitcoin_network: BitcoinNetwork::Testnet,
+                prometheus_monitoring_port: Some(9153),
             },
             logs: LogConfig {
                 ordinals_internals: true,
@@ -265,8 +274,8 @@ impl Config {
                 bitcoin_block_signaling: BitcoinBlockSignaling::Stacks(
                     StacksNodeConfig::default_localhost(DEFAULT_INGESTION_PORT),
                 ),
-                stacks_network: StacksNetwork::Mainnet,
                 bitcoin_network: BitcoinNetwork::Mainnet,
+                prometheus_monitoring_port: Some(9153),
             },
             logs: LogConfig {
                 ordinals_internals: true,

components/ordhook-core/src/config/mod.rs

@@ -39,11 +39,12 @@ use crate::{
         OrdhookConfig,
     },
     db::{
-        get_any_entry_in_ordinal_activities, open_ordhook_db_conn_rocks_db_loop,
-        open_readonly_ordhook_db_conn,
+        get_any_entry_in_ordinal_activities, get_latest_indexed_inscription_number,
+        open_ordhook_db_conn_rocks_db_loop, open_readonly_ordhook_db_conn,
     },
     service::write_brc20_block_operations,
     try_error, try_info,
+    utils::monitoring::PrometheusMonitoring,
 };
 
 use crate::db::{TransactionBytesCursor, TraversalResult};
@@ -61,12 +62,14 @@ pub fn start_inscription_indexing_processor(
     config: &Config,
     ctx: &Context,
     post_processor: Option<Sender<BitcoinBlockData>>,
+    prometheus: &PrometheusMonitoring,
 ) -> PostProcessorController {
     let (commands_tx, commands_rx) = crossbeam_channel::bounded::<PostProcessorCommand>(2);
     let (events_tx, events_rx) = crossbeam_channel::unbounded::<PostProcessorEvent>();
 
     let config = config.clone();
     let ctx = ctx.clone();
+    let prometheus = prometheus.clone();
     let handle: JoinHandle<()> = hiro_system_kit::thread_named("Inscription indexing runloop")
         .spawn(move || {
             let cache_l2 = Arc::new(new_traversals_lazy_cache(2048));
@@ -143,6 +146,7 @@ pub fn start_inscription_indexing_processor(
                     &mut brc20_db_conn_rw,
                     &ordhook_config,
                     &post_processor,
+                    &prometheus,
                     &ctx,
                 );
 
@@ -181,6 +185,7 @@ pub fn process_blocks(
     brc20_db_conn_rw: &mut Option<Connection>,
     ordhook_config: &OrdhookConfig,
     post_processor: &Option<Sender<BitcoinBlockData>>,
+    prometheus: &PrometheusMonitoring,
     ctx: &Context,
 ) -> Vec<BitcoinBlockData> {
     let mut cache_l1 = BTreeMap::new();
@@ -216,6 +221,7 @@ pub fn process_blocks(
             &inscriptions_db_tx,
             brc20_db_tx.as_ref(),
             brc20_cache.as_mut(),
+            prometheus,
             ordhook_config,
             ctx,
         );
@@ -284,6 +290,7 @@ pub fn process_block(
     inscriptions_db_tx: &Transaction,
     brc20_db_tx: Option<&Transaction>,
     brc20_cache: Option<&mut Brc20MemoryCache>,
+    prometheus: &PrometheusMonitoring,
     ordhook_config: &OrdhookConfig,
     ctx: &Context,
 ) -> Result<(), String> {
@@ -307,7 +314,7 @@ pub fn process_block(
         Context::empty()
     };
 
-    // Handle inscriptions
+    // Inscriptions
     if any_processable_transactions {
         let _ = augment_block_with_ordinals_inscriptions_data_and_write_to_db_tx(
             block,
@@ -317,10 +324,9 @@ pub fn process_block(
             &inner_ctx,
         );
     }
-
-    // Handle transfers
+    // Transfers
     let _ = augment_block_with_ordinals_transfer_data(block, inscriptions_db_tx, true, &inner_ctx);
-
+    // BRC-20
     match (brc20_db_tx, brc20_cache) {
         (Some(brc20_db_tx), Some(brc20_cache)) => write_brc20_block_operations(
             block,
@@ -332,5 +338,11 @@ pub fn process_block(
         _ => {}
     }
 
+    // Monitoring
+    prometheus.metrics_block_indexed(block.block_identifier.index);
+    prometheus.metrics_inscription_indexed(
+        get_latest_indexed_inscription_number(inscriptions_db_tx, &inner_ctx).unwrap_or(0),
+    );
+
     Ok(())
 }

components/ordhook-core/src/core/pipeline/processors/inscription_indexing.rs

@@ -835,6 +835,12 @@ pub fn find_latest_transfers_block_height(db_conn: &Connection, ctx: &Context) -
     entry
 }
 
+pub fn get_latest_indexed_inscription_number(db_conn: &Connection, ctx: &Context) -> Option<u64> {
+    let args: &[&dyn ToSql] = &[];
+    let query = "SELECT MAX(jubilee_inscription_number) FROM inscriptions";
+    perform_query_one(query, args, db_conn, ctx, |row| row.get(0).unwrap())
+}
+
 #[derive(Debug, Clone)]
 pub struct TransferData {
     pub inscription_offset_intra_output: u64,

components/ordhook-core/src/db/mod.rs

@@ -22,12 +22,10 @@ use rocket::{
 use rocket::{response::status::Custom, State};
 
 use crate::{
-    config::PredicatesApi,
-    service::observers::{
+    config::PredicatesApi, service::observers::{
         insert_entry_in_observers, open_readwrite_observers_db_conn, remove_entry_from_observers,
         update_observer_progress, update_observer_streaming_enabled,
-    },
-    try_error, try_info,
+    }, try_error, try_info, utils::monitoring::PrometheusMonitoring
 };
 
 use super::observers::{
@@ -39,6 +37,7 @@ pub async fn start_observers_http_server(
     observer_commands_tx: &std::sync::mpsc::Sender<ObserverCommand>,
     observer_event_rx: crossbeam_channel::Receiver<ObserverEvent>,
     bitcoin_scan_op_tx: crossbeam_channel::Sender<BitcoinChainhookSpecification>,
+    prometheus: &PrometheusMonitoring,
     ctx: &Context,
 ) -> Result<Shutdown, String> {
     // Build and start HTTP server.
@@ -51,6 +50,7 @@ pub async fn start_observers_http_server(
     // Spawn predicate observer event tread.
     let moved_config = config.clone();
     let moved_ctx = ctx.clone();
+    let moved_prometheus = prometheus.clone();
     let _ = hiro_system_kit::thread_named("observers_api-events").spawn(move || loop {
         let event = match observer_event_rx.recv() {
             Ok(cmd) => cmd,
@@ -72,8 +72,10 @@ pub async fn start_observers_http_server(
                     };
                 let report = ObserverReport::default();
                 insert_entry_in_observers(&spec, &report, &observers_db_conn, &moved_ctx);
+                moved_prometheus.metrics_register_predicate();
                 match spec {
                     ChainhookSpecification::Bitcoin(predicate_spec) => {
+                        // TODO: This action blocks this thread until the scan operation is complete. We should not do this.
                         let _ = bitcoin_scan_op_tx.send(predicate_spec);
                     }
                     _ => {}
@@ -109,6 +111,7 @@ pub async fn start_observers_http_server(
                         }
                     };
                 remove_entry_from_observers(&uuid, &observers_db_conn, &moved_ctx);
+                moved_prometheus.metrics_deregister_predicate();
             }
             ObserverEvent::BitcoinPredicateTriggered(data) => {
                 if let Some(ref tip) = data.apply.last() {
@@ -426,7 +429,7 @@ mod test {
 
     use crate::{
         config::{Config, PredicatesApi, PredicatesApiConfig},
-        service::observers::{delete_observers_db, initialize_observers_db},
+        service::observers::{delete_observers_db, initialize_observers_db}, utils::monitoring::PrometheusMonitoring,
     };
 
     use super::start_observers_http_server;
@@ -448,6 +451,7 @@ mod test {
             &observer_command_tx,
             observer_event_rx,
             bitcoin_scan_op_tx,
+            &PrometheusMonitoring::new(),
             &ctx,
         )
         .await