Merge pull request #173 from movementlabsxyz/icarus131/prometheus-setup

Updated metrics with prometheus and grafana setup

Author: andygolay

Cargo.lock

@@ -515,7 +515,7 @@ pub static CONSENSUS_LAST_TIMEOUT_VOTE_ROUND: Lazy<IntGaugeVec> = Lazy::new(|| {
 pub static CURRENT_ROUND: Lazy<IntGauge> = Lazy::new(|| {
     register_int_gauge!(
         "aptos_consensus_current_round",
-        "This counter is set to the last round reported by the local round_state."
+        "Current consensus round"
     )
     .unwrap()
 });
@@ -546,11 +546,14 @@ pub static TIMEOUT_COUNT: Lazy<IntCounter> = Lazy::new(|| {
     register_int_counter!("aptos_consensus_timeout_count", "Count the number of timeouts a node experienced since last restart (close to 0 in happy path).").unwrap()
 });
 
-/// The timeout of the current round.
-pub static ROUND_TIMEOUT_MS: Lazy<IntGauge> = Lazy::new(|| {
-    register_int_gauge!(
-        "aptos_consensus_round_timeout_s",
-        "The timeout of the current round."
+/// Round timeout in milliseconds
+pub static ROUND_TIMEOUT_MS: Lazy<Histogram> = Lazy::new(|| {
+    register_histogram!(
+        "aptos_consensus_round_timeout_ms",
+        "Round timeout in milliseconds",
+        vec![
+            100.0, 200.0, 500.0, 1000.0, 2000.0, 5000.0, 10000.0, 20000.0, 50000.0, 100000.0
+        ]
     )
     .unwrap()
 });
@@ -638,8 +641,8 @@ pub static ORDER_VOTE_BROADCASTED: Lazy<IntCounter> = Lazy::new(|| {
 /// Counts the number of times the sync info message has been set since last restart.
 pub static SYNC_INFO_MSGS_SENT_COUNT: Lazy<IntCounter> = Lazy::new(|| {
     register_int_counter!(
-        "aptos_consensus_sync_info_msg_sent_count",
-        "Counts the number of times the sync info message has been set since last restart."
+        "aptos_consensus_sync_info_msgs_sent_count",
+        "Number of sync info messages sent"
     )
     .unwrap()
 });
@@ -792,6 +795,16 @@ pub static WAIT_DURATION_S: Lazy<DurationHistogram> = Lazy::new(|| {
     CONSENSUS_WAIT_DURATION_BUCKETS.to_vec()).unwrap())
 });
 
+/// Wait duration in milliseconds
+pub static WAIT_DURATION_MS: Lazy<Histogram> = Lazy::new(|| {
+    register_histogram!(
+        "aptos_consensus_wait_duration_ms",
+        "Wait duration in milliseconds",
+        exponential_buckets(/*start=*/ 1.0, /*factor=*/ 2.0, /*count=*/ 30).unwrap(),
+    )
+    .unwrap()
+});
+
 const VERIFY_BUCKETS: &[f64] = &[
     0.0001, 0.00025, 0.0005, 0.001, 0.0015, 0.002, 0.0025, 0.003, 0.0035, 0.004, 0.005, 0.006,
     0.007, 0.008, 0.009, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0,

consensus/src/counters.rs

@@ -333,7 +333,7 @@ impl RoundManager {
         new_round_event: NewRoundEvent,
     ) -> anyhow::Result<()> {
         counters::CURRENT_ROUND.set(new_round_event.round as i64);
-        counters::ROUND_TIMEOUT_MS.set(new_round_event.timeout.as_millis() as i64);
+        counters::ROUND_TIMEOUT_MS.observe(new_round_event.timeout.as_millis() as f64);
         match new_round_event.reason {
             NewRoundReason::QCReady => {
                 counters::QC_ROUNDS_COUNT.inc();

consensus/src/round_manager.rs

@@ -0,0 +1,72 @@
+use crate::server::utils::{get_all_metrics, CONTENT_TYPE_TEXT};
+use hyper::{Body, StatusCode};
+use prometheus::TextEncoder;
+
+/// Handles a request for consensus metrics
+pub fn handle_consensus_metrics_request() -> (StatusCode, Body, String) {
+    let all_metrics = get_all_metrics();
+    
+    // Filter for consensus metrics
+    let consensus_metrics: Vec<String> = all_metrics
+        .iter()
+        .filter_map(|(key, value)| {
+            if key.starts_with("aptos_consensus") {
+                Some(format!("{} {}", key, value))
+            } else {
+                None
+            }
+        })
+        .collect();
+
+    (
+        StatusCode::OK,
+        Body::from(consensus_metrics.join("\n")),
+        CONTENT_TYPE_TEXT.into(),
+    )
+}
+
+/// Handles a request for mempool metrics
+pub fn handle_mempool_metrics_request() -> (StatusCode, Body, String) {
+    let all_metrics = get_all_metrics();
+    
+    // Filter for mempool metrics
+    let mempool_metrics: Vec<String> = all_metrics
+        .iter()
+        .filter_map(|(key, value)| {
+            if key.starts_with("aptos_mempool") || key.starts_with("aptos_core_mempool") {
+                Some(format!("{} {}", key, value))
+            } else {
+                None
+            }
+        })
+        .collect();
+
+    (
+        StatusCode::OK,
+        Body::from(mempool_metrics.join("\n")),
+        CONTENT_TYPE_TEXT.into(),
+    )
+}
+
+/// Handles a request for storage metrics
+pub fn handle_storage_metrics_request() -> (StatusCode, Body, String) {
+    let all_metrics = get_all_metrics();
+    
+    // Filter for storage metrics
+    let storage_metrics: Vec<String> = all_metrics
+        .iter()
+        .filter_map(|(key, value)| {
+            if key.starts_with("aptos_storage") || key.starts_with("aptos_schemadb") {
+                Some(format!("{} {}", key, value))
+            } else {
+                None
+            }
+        })
+        .collect();
+
+    (
+        StatusCode::OK,
+        Body::from(storage_metrics.join("\n")),
+        CONTENT_TYPE_TEXT.into(),
+    )
+} 

crates/aptos-inspection-service/src/server/custom_metrics.rs

@@ -2,8 +2,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::{
-    server::utils::CONTENT_TYPE_TEXT, CONFIGURATION_PATH, FORGE_METRICS_PATH, JSON_METRICS_PATH,
-    METRICS_PATH, PEER_INFORMATION_PATH, SYSTEM_INFORMATION_PATH,
+    server::utils::CONTENT_TYPE_TEXT, CONFIGURATION_PATH, CONSENSUS_METRICS_PATH, FORGE_METRICS_PATH, 
+    JSON_METRICS_PATH, MEMPOOL_METRICS_PATH, METRICS_PATH, PEER_INFORMATION_PATH, 
+    STORAGE_METRICS_PATH, SYSTEM_INFORMATION_PATH,
 };
 use hyper::{Body, StatusCode};
 
@@ -25,10 +26,13 @@ fn get_index_response() -> String {
     index_response.push("Welcome to the Aptos Inspection Service!".into());
     index_response.push("The following endpoints are available:".into());
     index_response.push(format!("\t- {}", CONFIGURATION_PATH));
+    index_response.push(format!("\t- {}", CONSENSUS_METRICS_PATH));
     index_response.push(format!("\t- {}", FORGE_METRICS_PATH));
     index_response.push(format!("\t- {}", JSON_METRICS_PATH));
+    index_response.push(format!("\t- {}", MEMPOOL_METRICS_PATH));
     index_response.push(format!("\t- {}", METRICS_PATH));
     index_response.push(format!("\t- {}", PEER_INFORMATION_PATH));
+    index_response.push(format!("\t- {}", STORAGE_METRICS_PATH));
     index_response.push(format!("\t- {}", SYSTEM_INFORMATION_PATH));
 
     index_response.join("\n") // Separate each entry with a newline

crates/aptos-inspection-service/src/server/index.rs

@@ -18,6 +18,7 @@ use std::{
 };
 
 mod configuration;
+mod custom_metrics;
 mod index;
 mod json_encoder;
 mod metrics;
@@ -37,6 +38,11 @@ pub const METRICS_PATH: &str = "/metrics";
 pub const PEER_INFORMATION_PATH: &str = "/peer_information";
 pub const SYSTEM_INFORMATION_PATH: &str = "/system_information";
 
+// New custom metrics paths
+pub const CONSENSUS_METRICS_PATH: &str = "/consensus_metrics"; 
+pub const MEMPOOL_METRICS_PATH: &str = "/mempool_metrics";
+pub const STORAGE_METRICS_PATH: &str = "/storage_metrics";
+
 // Useful string constants
 pub const HEADER_CONTENT_TYPE: &str = "Content-Type";
 pub const INVALID_ENDPOINT_MESSAGE: &str = "The requested endpoint is invalid!";
@@ -131,6 +137,21 @@ async fn serve_requests(
             // Exposes text encoded metrics
             metrics::handle_metrics_request()
         },
+        CONSENSUS_METRICS_PATH => {
+            // /consensus_metrics
+            // Exposes consensus metrics only
+            custom_metrics::handle_consensus_metrics_request()
+        },
+        MEMPOOL_METRICS_PATH => {
+            // /mempool_metrics
+            // Exposes mempool metrics only
+            custom_metrics::handle_mempool_metrics_request()
+        },
+        STORAGE_METRICS_PATH => {
+            // /storage_metrics
+            // Exposes storage metrics only
+            custom_metrics::handle_storage_metrics_request()
+        },
         PEER_INFORMATION_PATH => {
             // /peer_information
             // Exposes the peer information

crates/aptos-inspection-service/src/server/mod.rs

@@ -4,11 +4,13 @@
 use crate::{
     server::{
         configuration::CONFIGURATION_DISABLED_MESSAGE,
-        peer_information::PEER_INFO_DISABLED_MESSAGE, serve_requests,
+        peer_information::PEER_INFO_DISABLED_MESSAGE,
         system_information::SYS_INFO_DISABLED_MESSAGE, utils::get_all_metrics,
+        serve_requests,
     },
-    CONFIGURATION_PATH, FORGE_METRICS_PATH, INDEX_PATH, JSON_METRICS_PATH, METRICS_PATH,
-    PEER_INFORMATION_PATH, SYSTEM_INFORMATION_PATH,
+    CONFIGURATION_PATH, CONSENSUS_METRICS_PATH, FORGE_METRICS_PATH, INDEX_PATH, JSON_METRICS_PATH, 
+    MEMPOOL_METRICS_PATH, METRICS_PATH, PEER_INFORMATION_PATH, STORAGE_METRICS_PATH, 
+    SYSTEM_INFORMATION_PATH,
 };
 use aptos_config::config::{AptosDataClientConfig, BaseConfig, NodeConfig};
 use aptos_data_client::client::AptosDataClient;
@@ -180,6 +182,77 @@ async fn test_inspect_peer_information() {
     assert!(response_body_string.contains("State sync metadata"));
 }
 
+#[tokio::test]
+async fn test_inspect_consensus_metrics() {
+    // Create a validator config
+    let config = NodeConfig::get_default_validator_config();
+
+    // Increment a counter and get the metrics
+    INT_COUNTER.inc();
+    let mut response = send_get_request_to_path(&config, CONSENSUS_METRICS_PATH).await;
+    let response_body = body::to_bytes(response.body_mut()).await.unwrap();
+    let response_body_string = read_to_string(response_body.as_ref()).unwrap();
+
+    // Verify that the response contains only consensus metrics
+    assert_eq!(response.status(), StatusCode::OK);
+    
+    // For test environments, response might be empty, but should still return OK
+    if !response_body_string.is_empty() {
+        for line in response_body_string.lines() {
+            assert!(line.starts_with("aptos_consensus"), 
+                "Expected consensus metrics only, but found: {}", line);
+        }
+    }
+}
+
+#[tokio::test]
+async fn test_inspect_mempool_metrics() {
+    // Create a validator config
+    let config = NodeConfig::get_default_validator_config();
+
+    // Get the mempool metrics
+    let mut response = send_get_request_to_path(&config, MEMPOOL_METRICS_PATH).await;
+    let response_body = body::to_bytes(response.body_mut()).await.unwrap();
+    let response_body_string = read_to_string(response_body.as_ref()).unwrap();
+
+    // Verify that the response contains only mempool metrics
+    assert_eq!(response.status(), StatusCode::OK);
+    
+    // For test environments, response might be empty, but should still return OK
+    if !response_body_string.is_empty() {
+        for line in response_body_string.lines() {
+            assert!(
+                line.starts_with("aptos_mempool") || line.starts_with("aptos_core_mempool"), 
+                "Expected mempool metrics only, but found: {}", line
+            );
+        }
+    }
+}
+
+#[tokio::test]
+async fn test_inspect_storage_metrics() {
+    // Create a validator config
+    let config = NodeConfig::get_default_validator_config();
+
+    // Get the storage metrics
+    let mut response = send_get_request_to_path(&config, STORAGE_METRICS_PATH).await;
+    let response_body = body::to_bytes(response.body_mut()).await.unwrap();
+    let response_body_string = read_to_string(response_body.as_ref()).unwrap();
+
+    // Verify that the response contains only storage metrics
+    assert_eq!(response.status(), StatusCode::OK);
+    
+    // For test environments, response might be empty, but should still return OK
+    if !response_body_string.is_empty() {
+        for line in response_body_string.lines() {
+            assert!(
+                line.starts_with("aptos_storage") || line.starts_with("aptos_schemadb"), 
+                "Expected storage metrics only, but found: {}", line
+            );
+        }
+    }
+}
+
 rusty_fork_test! {
 #[test]
 fn test_gather_metrics() {