wip: bonds and unbonds (#4)

* wip: tx size

* wip: added batch size

* wip: added bonds and unbonds

* added prometheus metrics

Author: Fraccaman

src/checks/mod.rs

@@ -1,21 +1,25 @@
 pub mod block;
 pub mod epoch;
 pub mod total_supply_native;
+pub mod tx_size;
 
 use block::BlockHeigtCheck;
 use epoch::EpochCheck;
 use total_supply_native::TotalSupplyNativeCheck;
+use tx_size::TxSizeCheck;
 
 pub enum Checks {
     BlockHeightCheck(BlockHeigtCheck),
     EpochCheck(EpochCheck),
     TotalSupplyNative(TotalSupplyNativeCheck),
+    TxSize(TxSizeCheck),
 }
 
 pub fn all_checks() -> Vec<Checks> {
     vec![
         Checks::BlockHeightCheck(BlockHeigtCheck::default()),
         Checks::EpochCheck(EpochCheck::default()),
         Checks::TotalSupplyNative(TotalSupplyNativeCheck::default()),
+        Checks::TxSize(TxSizeCheck::default()),
     ]
 }

src/checks/tx_size.rs

@@ -0,0 +1,12 @@
+#[derive(Clone, Debug, Default)]
+pub struct TxSizeCheck {}
+
+impl TxSizeCheck {
+    pub async fn run(
+        &self,
+        pre_state: &crate::state::State,
+        post_state: &crate::state::State,
+    ) -> anyhow::Result<()> {
+        Ok(())
+    }
+}

src/main.rs

@@ -80,9 +80,13 @@ async fn main() -> anyhow::Result<()> {
                     .query_total_supply(&native_token)
                     .await
                     .into_retry_error()?;
+                let (future_bonds, future_unbonds) = rpc
+                    .query_future_bonds_and_unbonds(epoch)
+                    .await
+                    .into_retry_error()?;
 
                 let mut post_state_lock = state.write().await;
-                post_state_lock.update(block, total_supply_native);
+                post_state_lock.update(block, total_supply_native, future_bonds, future_unbonds);
                 let post_state = post_state_lock.clone();
                 drop(post_state_lock);
 
@@ -97,6 +101,7 @@ async fn main() -> anyhow::Result<()> {
                         checks::Checks::TotalSupplyNative(check) => {
                             check.run(&pre_state, &post_state).await
                         }
+                        checks::Checks::TxSize(check) => check.run(&pre_state, &post_state).await,
                     };
                     if let Err(error) = check_res {
                         tracing::error!("Error: {}", error.to_string());
@@ -107,6 +112,8 @@ async fn main() -> anyhow::Result<()> {
                     }
                 }
 
+                tracing::info!("Done block {}", block_height);
+
                 Ok(())
             },
             notify,

src/rpc.rs

@@ -3,7 +3,7 @@ use std::str::FromStr;
 use anyhow::Context;
 use futures::FutureExt;
 use namada_sdk::{
-    address::Address as NamadaAddress, hash::Hash, io::Client, rpc,
+    address::Address as NamadaAddress, hash::Hash, io::Client, rpc, state::Epoch as NamadaEpoch,
     state::Key,
 };
 use tendermint_rpc::{HttpClient, Url};
@@ -106,4 +106,28 @@ impl Rpc {
         res.context("Should be able to query native token")
             .map(|amount| amount.raw_amount().as_u64())
     }
+
+    pub async fn query_future_bonds_and_unbonds(&self, epoch: Epoch) -> anyhow::Result<(u64, u64)> {
+        let pipeline_epoch = NamadaEpoch(epoch + 1);
+        let futures = self.clients.iter().map(|client| {
+            rpc::enriched_bonds_and_unbonds(client, pipeline_epoch, &None, &None).boxed()
+        });
+
+        let (res, _ready_future_index, _remaining_futures) =
+            futures::future::select_all(futures).await;
+
+        res.context("Should be able to query native token")
+            .map(|summary| {
+                (
+                    summary
+                        .bonds_total_active()
+                        .map(|amount| amount.raw_amount().as_u64())
+                        .unwrap_or(0),
+                    summary
+                        .unbonds_total_active()
+                        .map(|amount| amount.raw_amount().as_u64())
+                        .unwrap_or(0),
+                )
+            })
+    }
 }

src/shared/namada.rs

@@ -1,4 +1,6 @@
-use namada_sdk::borsh::BorshDeserialize;
+use std::fmt::Display;
+
+use namada_sdk::borsh::{BorshDeserialize, BorshSerializeExt};
 use namada_sdk::governance::{InitProposalData, VoteProposalData};
 use namada_sdk::ibc::{decode_message, IbcMessage};
 use namada_sdk::key::common::PublicKey;
@@ -49,6 +51,28 @@ pub enum InnerKind {
     Unknown(Vec<u8>),
 }
 
+impl Display for InnerKind {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            InnerKind::TransparentTransfer(_) => write!(f, "transparent_transfer"),
+            InnerKind::ShieldedTransfer(_) => write!(f, "shielded_transfer"),
+            InnerKind::IbcMsgTransfer(_) => write!(f, "ibc"),
+            InnerKind::Bond(_) => write!(f, "bond"),
+            InnerKind::Redelegation(_) => write!(f, "redelegate"),
+            InnerKind::Unbond(_) => write!(f, "unbond"),
+            InnerKind::Withdraw(_) => write!(f, "withdraw"),
+            InnerKind::ClaimRewards(_) => write!(f, "claim_rewards"),
+            InnerKind::ProposalVote(_) => write!(f, "vote_proposal"),
+            InnerKind::InitProposal(_) => write!(f, "init_proposal"),
+            InnerKind::MetadataChange(_) => write!(f, "metadata_change"),
+            InnerKind::CommissionChange(_) => write!(f, "commission_change"),
+            InnerKind::RevealPk(_) => write!(f, "reveal_public_key"),
+            InnerKind::BecomeValidator(_) => write!(f, "become_validator"),
+            InnerKind::Unknown(_) => write!(f, "unknown"),
+        }
+    }
+}
+
 impl InnerKind {
     pub fn from_code_name(name: &str, data: &[u8]) -> Self {
         match name {
@@ -160,6 +184,74 @@ impl InnerKind {
             _ => InnerKind::Unknown(data.to_vec()),
         }
     }
+
+    pub fn size(&self) -> usize {
+        match self {
+            InnerKind::TransparentTransfer(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::ShieldedTransfer(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::IbcMsgTransfer(tx) => tx
+                .clone()
+                .map(|data| match data {
+                    IbcMessage::Envelope(msg_envelope) => msg_envelope.serialize_to_vec().len(),
+                    IbcMessage::Transfer(msg_transfer) => msg_transfer.serialize_to_vec().len(),
+                    IbcMessage::NftTransfer(msg_nft_transfer) => {
+                        msg_nft_transfer.serialize_to_vec().len()
+                    }
+                })
+                .unwrap_or(0),
+            InnerKind::Bond(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::Redelegation(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::Unbond(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::Withdraw(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::ClaimRewards(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::ProposalVote(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::InitProposal(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::MetadataChange(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::CommissionChange(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::RevealPk(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::BecomeValidator(tx) => tx
+                .clone()
+                .map(|data| data.serialize_to_vec().len())
+                .unwrap_or(0),
+            InnerKind::Unknown(tx) => tx.len(),
+        }
+    }
 }
 
 #[derive(Clone, Debug)]

src/state.rs

@@ -2,8 +2,8 @@ use std::num::NonZeroUsize;
 
 use lru::LruCache;
 use prometheus_exporter::prometheus::{
-    core::{AtomicU64, GenericCounter},
-    Registry,
+    core::{AtomicU64, GenericCounter, GenericCounterVec},
+    Histogram, HistogramOpts, IntCounterVec, Opts, Registry,
 };
 
 use crate::shared::{
@@ -26,6 +26,11 @@ pub struct PrometheusMetrics {
     pub block_height_counter: GenericCounter<AtomicU64>,
     pub epoch_counter: GenericCounter<AtomicU64>,
     pub total_supply_native_token: GenericCounter<AtomicU64>,
+    pub transaction_size: Histogram,
+    pub transaction_inner_size: Histogram,
+    pub bonds_per_epoch: GenericCounterVec<AtomicU64>,
+    pub unbonds_per_epoch: GenericCounterVec<AtomicU64>,
+    pub transaction_kind: GenericCounterVec<AtomicU64>,
     registry: Registry,
 }
 
@@ -52,18 +57,65 @@ impl PrometheusMetrics {
         )
         .expect("unable to create counter total supply");
 
+        let transaction_size_opts = HistogramOpts::new(
+            "transaction_size_bytes",
+            "The sizes of transactions in bytes",
+        )
+        .buckets(vec![
+            10.0, 50.0, 100.0, 500.0, 1000.0, 5000.0, 10000.0, 50000.0,
+        ]);
+        let transaction_size = Histogram::with_opts(transaction_size_opts)
+            .expect("unable to create histogram transaction sizes");
+
+        let transaction_inner_size_opts =
+            HistogramOpts::new("transaction_inners", "The number of inner tx for a wrapper")
+                .buckets(vec![2.0, 4.0, 8.0, 16.0, 32.0, 64.0, 128.0]);
+        let transaction_inner_size = Histogram::with_opts(transaction_inner_size_opts)
+            .expect("unable to create histogram transaction sizes");
+
+        let bonds_per_epoch_opts = Opts::new("bonds_per_epoch", "Total bonds per epoch");
+        let bonds_per_epoch = IntCounterVec::new(bonds_per_epoch_opts, &["epoch"])
+            .expect("unable to create histogram transaction sizes");
+
+        let unbonds_per_epoch_opts = Opts::new("unbonds_per_epoch", "Total unbonds per epoch");
+        let unbonds_per_epoch = IntCounterVec::new(unbonds_per_epoch_opts, &["epoch"])
+            .expect("unable to create histogram transaction sizes");
+
+        let transaction_kind_opts =
+            Opts::new("transaction_kind", "Total transaction per transaction kind");
+        let transaction_kind =
+            IntCounterVec::new(transaction_kind_opts, &["kind", "epoch", "height"])
+                .expect("unable to create histogram transaction sizes");
+
         registry
             .register(Box::new(block_height_counter.clone()))
             .unwrap();
         registry.register(Box::new(epoch_counter.clone())).unwrap();
         registry
             .register(Box::new(total_supply_native_token.clone()))
             .unwrap();
+        registry
+            .register(Box::new(transaction_size.clone()))
+            .unwrap();
+        registry
+            .register(Box::new(bonds_per_epoch.clone()))
+            .unwrap();
+        registry
+            .register(Box::new(unbonds_per_epoch.clone()))
+            .unwrap();
+        registry
+            .register(Box::new(transaction_kind.clone()))
+            .unwrap();
 
         Self {
             block_height_counter,
             epoch_counter,
             total_supply_native_token,
+            transaction_size,
+            transaction_inner_size,
+            bonds_per_epoch,
+            unbonds_per_epoch,
+            transaction_kind,
             registry,
         }
     }
@@ -95,7 +147,13 @@ impl State {
             .unwrap_or(1)
     }
 
-    pub fn update(&mut self, block: Block, total_supply_native: u64) {
+    pub fn update(
+        &mut self,
+        block: Block,
+        total_supply_native: u64,
+        future_bonds: u64,
+        future_unbonds: u64,
+    ) {
         if let Some(height) = self.latest_block_height {
             self.metrics
                 .block_height_counter
@@ -110,7 +168,7 @@ impl State {
                 .block_height_counter
                 .inc_by(block.epoch - epoch);
         } else {
-            self.metrics.block_height_counter.inc_by(block.epoch);
+            self.metrics.epoch_counter.inc_by(block.epoch);
         }
         self.latest_epoch = Some(block.epoch);
 
@@ -125,6 +183,49 @@ impl State {
         }
         self.latest_total_supply_native = Some(total_supply_native);
 
+        for tx in &block.transactions {
+            self.metrics
+                .transaction_inner_size
+                .observe(tx.inners.len() as f64);
+        }
+
+        for tx in &block.transactions {
+            for inner in &tx.inners {
+                let inner_kind = inner.kind.to_string();
+
+                self.metrics
+                    .transaction_kind
+                    .with_label_values(&[
+                        &inner_kind,
+                        &block.epoch.to_string(),
+                        &block.height.to_string(),
+                    ])
+                    .inc();
+
+                self.metrics
+                    .transaction_size
+                    .observe(inner.kind.size() as f64);
+            }
+        }
+
+        self.metrics
+            .bonds_per_epoch
+            .with_label_values(&[&(block.epoch + 1).to_string()])
+            .reset();
+        self.metrics
+            .bonds_per_epoch
+            .with_label_values(&[&(block.epoch + 1).to_string()])
+            .inc_by(future_bonds);
+
+        self.metrics
+            .unbonds_per_epoch
+            .with_label_values(&[&(block.epoch + 1).to_string()])
+            .reset();
+        self.metrics
+            .unbonds_per_epoch
+            .with_label_values(&[&(block.epoch + 1).to_string()])
+            .inc_by(future_unbonds);
+
         self.blocks.put(block.height, block);
     }