Implement prometheus metrics

Author: DavidVentura

Cargo.lock

@@ -29,8 +29,10 @@ getrandom = "0.2"
 hkd32 = { version = "0.7", default-features = false, features = ["mnemonic"] }
 hkdf = "0.12"
 k256 = { version = "0.13", features = ["ecdsa", "sha256"] }
-ledger = { version = "0.2", optional = true }
+lazy_static = { version = "1.5.0"  }
+ledger = { version = "0.2" }
 once_cell = "1.5"
+prometheus = { version = "0.13.4", default-features = false, features = [] }
 prost = "0.12"
 prost-derive = "0.12"
 rand_core = { version = "0.6", features = ["std"] }
@@ -48,6 +50,7 @@ tendermint-config = "0.35"
 tendermint-p2p = "0.35"
 tendermint-proto = "0.35"
 thiserror = "1"
+tiny_http = { version = "0.12.0" }
 url = { version = "2.2.2", features = ["serde"], optional = true }
 uuid = { version = "1", features = ["serde"], optional = true }
 wait-timeout = "0.2"

Cargo.toml

@@ -10,6 +10,7 @@ use crate::{
     config::ValidatorConfig,
     error::{Error, ErrorKind},
     prelude::*,
+    prometheus::initialize_consensus_metrics,
     session::Session,
 };
 use std::{panic, process::exit, thread, time::Duration};
@@ -88,7 +89,7 @@ fn main_loop(config: ValidatorConfig) -> Result<(), Error> {
 /// Ensure chain with given ID is properly registered
 pub fn register_chain(chain_id: &chain::Id) {
     let registry = chain::REGISTRY.get();
-
+    initialize_consensus_metrics(chain_id);
     debug!("registering chain: {}", chain_id);
     registry.get_chain(chain_id).unwrap_or_else(|| {
         status_err!(

src/client.rs

@@ -43,6 +43,7 @@ impl ConfigBuilder {
         self.add_chain_config();
         self.add_provider_config();
         self.add_validator_config();
+        self.add_metrics_config();
 
         self.contents
     }
@@ -148,6 +149,12 @@ impl ConfigBuilder {
         self.add_template_with_chain_id(include_str!("templates/keyring/fortanixdsm.toml"));
     }
 
+    /// Add `[metrics]` configurations
+    fn add_metrics_config(&mut self) {
+        self.add_section_comment("Metrics exporter configuration");
+        self.add_template(include_str!("templates/metrics.toml"));
+    }
+
     /// Append a template to the config file, substituting `$KMS_HOME`
     fn add_template(&mut self, template: &str) {
         self.add_str(&format_template(

src/commands/init/config_builder.rs

@@ -0,0 +1,2 @@
+[metrics]
+bind_address = "localhost:3333"

src/commands/init/templates/metrics.toml

@@ -1,6 +1,6 @@
 //! Start the KMS
 
-use crate::{chain, client::Client, prelude::*};
+use crate::{chain, client::Client, prelude::*, prometheus};
 use abscissa_core::Command;
 use clap::Parser;
 use std::{path::PathBuf, process};
@@ -40,6 +40,16 @@ impl StartCommand {
             process::exit(1);
         });
 
+        if let Some(config) = &APP.config().metrics {
+            let address = config.bind_address.clone();
+            info!("Starting up prometheus server on {}", address);
+            let thread_name = abscissa_core::thread::Name::new("prometheus-thread").unwrap();
+            APP.state()
+                .threads_mut()
+                .spawn(thread_name, move || prometheus::serve(&address))
+                .expect("Unable to start prometheus exporter thread");
+        }
+
         // Spawn the validator client threads
         config
             .validator

src/commands/start.rs

@@ -1,12 +1,13 @@
 //! Configuration file structures (with serde-derived parser)
 
 pub mod chain;
+pub mod metrics;
 pub mod provider;
 pub mod validator;
 
 pub use self::validator::*;
 
-use self::{chain::ChainConfig, provider::ProviderConfig};
+use self::{chain::ChainConfig, metrics::MetricsConfig, provider::ProviderConfig};
 use serde::Deserialize;
 
 /// Environment variable containing path to config file
@@ -29,4 +30,7 @@ pub struct KmsConfig {
     /// Addresses of validator nodes
     #[serde(default)]
     pub validator: Vec<ValidatorConfig>,
+
+    /// Configuration for metrics exporter
+    pub metrics: Option<MetricsConfig>,
 }

src/config.rs

@@ -0,0 +1,11 @@
+//! Metrics configuration
+
+use serde::Deserialize;
+
+/// Metrics configuration
+#[derive(Deserialize, Debug, Clone, Default)]
+#[serde(deny_unknown_fields)]
+pub struct MetricsConfig {
+    /// Address on which to bind metrics exporter
+    pub bind_address: String,
+}

src/config/metrics.rs

@@ -25,6 +25,7 @@ pub mod key_utils;
 pub mod keyring;
 pub mod prelude;
 pub mod privval;
+pub mod prometheus;
 pub mod rpc;
 pub mod session;
 

src/lib.rs

@@ -0,0 +1,130 @@
+//! Prometheus metrics collection and serving functionality
+use crate::rpc;
+
+use lazy_static::lazy_static;
+use prometheus::{
+    default_registry, register_histogram_vec, register_int_counter_vec, Encoder, HistogramVec,
+    IntCounterVec, TextEncoder,
+};
+use std::time::Duration;
+use tendermint::chain;
+
+lazy_static! {
+    static ref METRIC_CONSENSUS_UPDATES: IntCounterVec = register_int_counter_vec!(
+        "consensus_updates_total",
+        "Number of consensus updates by status.",
+        &["chain", "status"]
+    )
+    .unwrap();
+    static ref METRIC_REQUEST_DURATION: HistogramVec = register_histogram_vec!(
+        "request_duration_seconds",
+        "Duration of request",
+        &["chain", "message_type"],
+    )
+    .unwrap();
+}
+
+/// Status outcomes for consensus updates
+pub enum ConsensusUpdateStatus {
+    /// Update completed successfully
+    Success,
+    /// Generic error occurred during update
+    Error,
+    /// Double signing attempt detected
+    DoubleSign,
+}
+
+/// Type of RPC request received from validator node
+pub enum RpcRequestType {
+    /// Sign proposal or vote request
+    Sign,
+    /// Ping/heartbeat request
+    Ping,
+    /// Request for public key
+    Pubkey,
+}
+
+impl From<&rpc::Request> for RpcRequestType {
+    fn from(value: &rpc::Request) -> Self {
+        match value {
+            rpc::Request::SignProposal(_) | rpc::Request::SignVote(_) => RpcRequestType::Sign,
+            rpc::Request::PingRequest => RpcRequestType::Ping,
+            rpc::Request::ShowPublicKey => RpcRequestType::Pubkey,
+        }
+    }
+}
+impl ToString for ConsensusUpdateStatus {
+    fn to_string(&self) -> String {
+        match self {
+            ConsensusUpdateStatus::Success => "success".into(),
+            ConsensusUpdateStatus::Error => "error".into(),
+            ConsensusUpdateStatus::DoubleSign => "double_sign".into(),
+        }
+    }
+}
+
+/// Initialize metrics counters for a given chain ID
+///
+/// Creates and resets counters for all possible status outcomes
+pub fn initialize_consensus_metrics(chain_id: &chain::Id) {
+    METRIC_CONSENSUS_UPDATES
+        .with_label_values(&[
+            chain_id.as_str(),
+            &ConsensusUpdateStatus::Success.to_string(),
+        ])
+        .reset();
+    METRIC_CONSENSUS_UPDATES
+        .with_label_values(&[chain_id.as_str(), &ConsensusUpdateStatus::Error.to_string()])
+        .reset();
+    METRIC_CONSENSUS_UPDATES
+        .with_label_values(&[
+            chain_id.as_str(),
+            &ConsensusUpdateStatus::DoubleSign.to_string(),
+        ])
+        .reset();
+}
+
+/// Increment counter for a consensus update with given status
+pub fn increment_consensus_counter(chain_id: &chain::Id, status: ConsensusUpdateStatus) {
+    METRIC_CONSENSUS_UPDATES
+        .with_label_values(&[chain_id.as_str(), &status.to_string()])
+        .inc();
+}
+
+impl ToString for RpcRequestType {
+    fn to_string(&self) -> String {
+        match self {
+            RpcRequestType::Sign => "sign",
+            RpcRequestType::Ping => "ping",
+            RpcRequestType::Pubkey => "pubkey",
+        }
+        .into()
+    }
+}
+
+/// Record duration of an RPC request
+pub fn record_request_duration(
+    chain_id: &chain::Id,
+    request_type: &RpcRequestType,
+    duration: Duration,
+) {
+    METRIC_REQUEST_DURATION
+        .with_label_values(&[chain_id.as_str(), &request_type.to_string()])
+        .observe(duration.as_secs_f64());
+}
+
+/// Start HTTP server to expose Prometheus metrics
+pub fn serve(address: &str) {
+    let encoder = TextEncoder::new();
+    let registry = default_registry();
+    let server = tiny_http::Server::http(address).expect("Unable to bind to address");
+    for request in server.incoming_requests() {
+        let mut response = Vec::<u8>::new();
+        let metric_families = registry.gather();
+        // TODO
+        encoder.encode(&metric_families, &mut response).unwrap();
+        request
+            .respond(tiny_http::Response::from_data(response))
+            .unwrap();
+    }
+}