Introduce New Observability - Remaining events

fendermint/abci/examples/kvstore.rs

@@ -75,7 +75,8 @@ impl Application for KVStore {
         &self,
         request: request::PrepareProposal,
     ) -> Result<response::PrepareProposal> {
-        let mut txs = take_until_max_size(request.txs, request.max_tx_bytes.try_into().unwrap());
+        let (txs, _) = take_until_max_size(request.txs, request.max_tx_bytes.try_into().unwrap());
+        let mut txs = txs.clone();
 
         // Enfore transaciton limit so that we don't have a problem with buffering.
         txs.truncate(MAX_TXNS);

fendermint/abci/src/application.rs

@@ -64,7 +64,7 @@ pub trait Application {
         &self,
         request: request::PrepareProposal,
     ) -> AbciResult<response::PrepareProposal> {
-        let txs = take_until_max_size(request.txs, request.max_tx_bytes.try_into().unwrap());
+        let (txs, _) = take_until_max_size(request.txs, request.max_tx_bytes.try_into().unwrap());
 
         Ok(response::PrepareProposal { txs })
     }

fendermint/abci/src/util.rs

@@ -4,7 +4,7 @@
 /// Take the first transactions until the first one that would exceed the maximum limit.
 ///
 /// The function does not skip or reorder transaction even if a later one would stay within the limit.
-pub fn take_until_max_size<T: AsRef<[u8]>>(txs: Vec<T>, max_tx_bytes: usize) -> Vec<T> {
+pub fn take_until_max_size<T: AsRef<[u8]>>(txs: Vec<T>, max_tx_bytes: usize) -> (Vec<T>, usize) {
     let mut size: usize = 0;
     let mut out = Vec::new();
     for tx in txs {
@@ -15,5 +15,5 @@ pub fn take_until_max_size<T: AsRef<[u8]>>(txs: Vec<T>, max_tx_bytes: usize) ->
         size += bz.len();
         out.push(tx);
     }
-    out
+    (out, size)
 }

fendermint/app/src/app.rs

@@ -13,7 +13,6 @@ use fendermint_abci::{AbciResult, Application};
 use fendermint_storage::{
     Codec, Encode, KVCollection, KVRead, KVReadable, KVStore, KVWritable, KVWrite,
 };
-use fendermint_tracing::emit;
 use fendermint_vm_core::Timestamp;
 use fendermint_vm_interpreter::bytes::{
     BytesMessageApplyRes, BytesMessageCheckRes, BytesMessageQuery, BytesMessageQueryRes,
@@ -37,13 +36,17 @@ use fvm_shared::chainid::ChainID;
 use fvm_shared::clock::ChainEpoch;
 use fvm_shared::econ::TokenAmount;
 use fvm_shared::version::NetworkVersion;
+use ipc_observability::emit;
 use num_traits::Zero;
 use serde::{Deserialize, Serialize};
 use tendermint::abci::request::CheckTxKind;
 use tendermint::abci::{request, response};
 use tracing::instrument;
 
-use crate::events::{NewBlock, ProposalProcessed};
+use crate::observe::{
+    BlockCommitted, BlockProposalAccepted, BlockProposalReceived, BlockProposalRejected,
+    BlockProposalSent, HexEncodableBlockHash, MpoolReceived, MpoolReceivedInvalidMessage,
+};
 use crate::AppExitCode;
 use crate::BlockHeight;
 use crate::{tmconv::*, VERSION};
@@ -620,11 +623,42 @@ where
         *guard = Some(state);
 
         let response = match result {
-            Err(e) => invalid_check_tx(AppError::InvalidEncoding, e.description),
+            Err(e) => {
+                emit(MpoolReceivedInvalidMessage {
+                    reason: "InvalidEncoding",
+                    description: e.description.as_ref(),
+                });
+                invalid_check_tx(AppError::InvalidEncoding, e.description)
+            }
             Ok(result) => match result {
-                Err(IllegalMessage) => invalid_check_tx(AppError::IllegalMessage, "".to_owned()),
-                Ok(Err(InvalidSignature(d))) => invalid_check_tx(AppError::InvalidSignature, d),
-                Ok(Ok(ret)) => to_check_tx(ret),
+                Err(IllegalMessage) => {
+                    emit(MpoolReceivedInvalidMessage {
+                        reason: "IllegalMessage",
+                        description: "",
+                    });
+
+                    invalid_check_tx(AppError::IllegalMessage, "".to_owned())
+                }
+                Ok(Err(InvalidSignature(d))) => {
+                    emit(MpoolReceivedInvalidMessage {
+                        reason: "InvalidSignature",
+                        description: d.as_ref(),
+                    });
+                    invalid_check_tx(AppError::InvalidSignature, d)
+                }
+                Ok(Ok(ret)) => {
+                    emit(MpoolReceived {
+                        from: ret.message.from.to_string().as_str(),
+                        to: ret.message.to.to_string().as_str(),
+                        value: ret.message.value.to_string().as_str(),
+                        param_len: ret.message.params.len(),
+                        gas_limit: ret.message.gas_limit,
+                        fee_cap: ret.message.gas_fee_cap.to_string().as_str(),
+                        premium: ret.message.gas_premium.to_string().as_str(),
+                        accept: ret.exit_code.is_success(),
+                    });
+                    to_check_tx(ret)
+                }
             },
         };
 
@@ -650,7 +684,13 @@ where
             .context("failed to prepare proposal")?;
 
         let txs = txs.into_iter().map(bytes::Bytes::from).collect();
-        let txs = take_until_max_size(txs, request.max_tx_bytes.try_into().unwrap());
+        let (txs, size) = take_until_max_size(txs, request.max_tx_bytes.try_into().unwrap());
+
+        emit(BlockProposalSent {
+            height: request.height.value(),
+            tx_count: txs.len(),
+            size,
+        });
 
         Ok(response::PrepareProposal { txs })
     }
@@ -666,25 +706,40 @@ where
             "process proposal"
         );
         let txs: Vec<_> = request.txs.into_iter().map(|tx| tx.to_vec()).collect();
+        let size_txs = txs.iter().map(|tx| tx.len()).sum::<usize>();
         let num_txs = txs.len();
 
-        let accept = self
+        let (accept, reason) = self
             .interpreter
             .process(self.chain_env.clone(), txs)
             .await
             .context("failed to process proposal")?;
 
-        emit!(ProposalProcessed {
-            is_accepted: accept,
-            block_height: request.height.value(),
-            block_hash: request.hash.to_string().as_str(),
-            num_txs,
-            proposer: request.proposer_address.to_string().as_str()
+        emit(BlockProposalReceived {
+            height: request.height.value(),
+            hash: HexEncodableBlockHash(request.hash.into()),
+            size: size_txs,
+            tx_count: num_txs,
+            validator: request.proposer_address.to_string().as_str(),
         });
 
         if accept {
+            emit(BlockProposalAccepted {
+                height: request.height.value(),
+                hash: HexEncodableBlockHash(request.hash.into()),
+                size: size_txs,
+                tx_count: num_txs,
+                validator: request.proposer_address.to_string().as_str(),
+            });
             Ok(response::ProcessProposal::Accept)
         } else {
+            emit(BlockProposalRejected {
+                height: request.height.value(),
+                size: size_txs,
+                tx_count: num_txs,
+                validator: request.proposer_address.to_string().as_str(),
+                reason: reason.unwrap_or_default().as_str(),
+            });
             Ok(response::ProcessProposal::Reject)
         }
     }
@@ -867,12 +922,15 @@ where
         // Commit app state to the datastore.
         self.set_committed_state(state)?;
 
-        emit!(NewBlock { block_height });
-
         // Reset check state.
         let mut guard = self.check_state.lock().await;
         *guard = None;
 
+        emit(BlockCommitted {
+            height: block_height,
+            app_hash: HexEncodableBlockHash(app_hash.clone().into()),
+        });
+
         Ok(response::Commit {
             data: app_hash.into(),
             retain_height: retain_height.try_into().expect("height is valid"),

fendermint/app/src/lib.rs

@@ -1,9 +1,9 @@
 // Copyright 2022-2024 Protocol Labs
 // SPDX-License-Identifier: Apache-2.0, MIT
 mod app;
-pub mod events;
 pub mod ipc;
 pub mod metrics;
+pub mod observe;
 mod store;
 mod tmconv;
 

fendermint/app/src/metrics/mod.rs

@@ -2,8 +2,6 @@
 // SPDX-License-Identifier: Apache-2.0, MIT
 
 mod prometheus;
-mod tracing;
 
 pub use prometheus::app::register_metrics as register_app_metrics;
 pub use prometheus::eth::register_metrics as register_eth_metrics;
-pub use tracing::layer;