feat(abci): support tracing::span inside rs-tenderdash-abci (#35)

* feat(proto): DeriveFrom for ABCI Request Value

* feat(abci): add tracing spans

* feat: add request id to tracing span

* chore: log exception responses on error! level and simplify span

* chore: minor improvements  and docs

* chore: traced exception format same as response

* chore: minor fix

* chore: cargo fmt

* fix: doctest

abci/Cargo.toml

@@ -11,7 +11,7 @@ description = """tenderdash-abci provides a simple framework with which to build
 low-level applications on top of Tenderdash."""
 
 [features]
-default = ["server", "docker-tests", "crypto", "tcp", "unix"]
+default = ["server", "docker-tests", "crypto", "tcp", "unix", "tracing-span"]
 # docker-tests includes integration tests that require docker to be available
 docker-tests = ["server"]
 server = [
@@ -23,12 +23,14 @@ server = [
 crypto = ["dep:lhash"]
 tcp = ["server"]
 unix = ["server"]
+tracing-span = ["dep:uuid"]
 
 [[example]]
 name = "echo_socket"
 required-features = ["server"]
 
 [dependencies]
+uuid = { version = "1.4.1", features = ["v4", "fast-rng"], optional = true }
 tenderdash-proto = { version = "0.13.0-dev.2", path = "../proto" }
 bytes = { version = "1.0" }
 prost = { version = "0.11" }

abci/src/application.rs

@@ -150,9 +150,11 @@ pub trait RequestDispatcher {
 // Implement `RequestDispatcher` for all `Application`s.
 impl<A: Application> RequestDispatcher for A {
     fn handle(&self, request: abci::Request) -> Option<abci::Response> {
-        tracing::trace!(?request, "received request");
+        #[cfg(feature = "tracing-span")]
+        let _span = super::tracing_span::span(request.clone().value?);
+        tracing::trace!(?request, "received ABCI request");
 
-        let response: Result<response::Value, abci::ResponseException> = match request.value? {
+        let response: response::Value = match request.value? {
             request::Value::Echo(req) => self.echo(req).map(|v| v.into()),
             request::Value::Flush(req) => self.flush(req).map(|v| v.into()),
             request::Value::Info(req) => self.info(req).map(|v| v.into()),
@@ -174,15 +176,15 @@ impl<A: Application> RequestDispatcher for A {
             request::Value::VerifyVoteExtension(req) => {
                 self.verify_vote_extension(req).map(|v| v.into())
             },
-        };
+        }
+        .unwrap_or_else(|e| e.into());
 
-        let response = match response {
-            Ok(v) => v,
-            Err(e) => response::Value::from(e),
+        if let response::Value::Exception(_) = response {
+            tracing::error!(?response, "sending ABCI exception");
+        } else {
+            tracing::trace!(?response, "sending ABCI response");
         };
 
-        tracing::trace!(?response, "sending response");
-
         Some(abci::Response {
             value: Some(response),
         })

abci/src/lib.rs

@@ -26,6 +26,9 @@ pub use tenderdash_proto as proto;
 
 #[cfg(feature = "crypto")]
 pub mod signatures;
+#[cfg(feature = "tracing-span")]
+/// Create tracing::Span for better logging
+pub mod tracing_span;
 
 /// Errors that may happen during protobuf communication
 #[derive(Debug, thiserror::Error)]

abci/src/server/generic.rs

@@ -106,15 +106,15 @@ where
         let mut codec = Codec::new(listener, cancel_token.clone(), &self.runtime);
         while !cancel_token.is_cancelled() {
             let Some(request) = codec.next() else {
-            tracing::error!("client terminated stream");
-            return Ok(())
-        };
-
-            let Some(response) = self.app.handle(request.clone())  else {
-            // `RequestDispatcher` decided to stop receiving new requests:
-            info!("ABCI Application is shutting down");
-            return Ok(());
-        };
+                tracing::error!("client terminated stream");
+                return Ok(());
+            };
+
+            let Some(response) = self.app.handle(request.clone()) else {
+                // `RequestDispatcher` decided to stop receiving new requests:
+                info!("ABCI Application is shutting down");
+                return Ok(());
+            };
 
             if let Some(crate::proto::abci::response::Value::Exception(ex)) = response.value.clone()
             {

abci/src/tracing_span.rs

@@ -0,0 +1,97 @@
+use tenderdash_proto::abci::request::Value;
+use tracing::Level;
+
+const SPAN_NAME: &str = "abci";
+const LEVEL: Level = Level::ERROR;
+
+macro_rules! block_span {
+    ($request: expr, $endpoint:expr, $request_id:expr) => {
+        tracing::span!(
+            LEVEL,
+            SPAN_NAME,
+            endpoint = $endpoint,
+            request_id = $request_id,
+            height = $request.height,
+            round = $request.round
+        )
+    };
+}
+/// Creates a new span for tracing.
+///
+/// This function creates a new `tracing::span::EnteredSpan` based on the
+/// provided request. It uses the request to determine the endpoint and includes
+/// a unique request ID in the span.
+///
+/// The level of the span is set to ERROR, so it will be included on all log
+/// levels.
+///
+/// # Arguments
+///
+/// * `request` - A value that can be converted into a `Value`. Depending on the
+///   specific variant of `Value`, additional information like height, round, or
+///   path might be included in the span.
+///
+/// # Returns
+///
+/// An entered span which represents an active or entered span state.
+///
+/// # Examples
+///
+/// ```
+/// # use tenderdash_proto::abci::{RequestInfo, request};
+/// # use tenderdash_abci::tracing_span::span;
+///
+/// let request = request::Value::Info(RequestInfo::default());
+/// let span = span(request);
+/// ```
+pub fn span<T>(request: T) -> tracing::span::EnteredSpan
+where
+    T: Into<Value>,
+{
+    let value = request.into();
+
+    let endpoint = abci_method_name(&value);
+    let request_id = uuid::Uuid::new_v4().to_string();
+
+    let span = match value {
+        Value::Info(_r) => tracing::span!(LEVEL, SPAN_NAME, endpoint, request_id),
+        Value::InitChain(_r) => {
+            tracing::span!(LEVEL, SPAN_NAME, endpoint, request_id)
+        },
+        Value::PrepareProposal(r) => block_span!(r, endpoint, request_id),
+        Value::ProcessProposal(r) => block_span!(r, endpoint, request_id),
+        Value::ExtendVote(r) => block_span!(r, endpoint, request_id),
+        Value::VerifyVoteExtension(r) => block_span!(r, endpoint, request_id),
+        Value::FinalizeBlock(r) => block_span!(r, endpoint, request_id),
+        Value::CheckTx(_r) => {
+            tracing::span!(LEVEL, SPAN_NAME, endpoint, request_id)
+        },
+        Value::Query(r) => {
+            tracing::span!(LEVEL, SPAN_NAME, endpoint, request_id, path = r.path)
+        },
+        _ => tracing::span!(LEVEL, SPAN_NAME, endpoint, request_id),
+    };
+
+    span.entered()
+}
+
+fn abci_method_name(request: &Value) -> String {
+    match request {
+        Value::ApplySnapshotChunk(_) => "ApplySnapshotChunk",
+        Value::CheckTx(_) => "CheckTx",
+        Value::Echo(_) => "Echo",
+        Value::ExtendVote(_) => "ExtendVote",
+        Value::FinalizeBlock(_) => "FinalizeBlock",
+        Value::Flush(_) => "Flush",
+        Value::Info(_) => "Info",
+        Value::InitChain(_) => "InitChain",
+        Value::ListSnapshots(_) => "ListSnapshots",
+        Value::LoadSnapshotChunk(_) => "LoadSnapshotChunk",
+        Value::OfferSnapshot(_) => "OfferSnapshot",
+        Value::PrepareProposal(_) => "PrepareProposal",
+        Value::ProcessProposal(_) => "ProcessProposal",
+        Value::Query(_) => "Query",
+        Value::VerifyVoteExtension(_) => "VerifyVoteExtension",
+    }
+    .to_string()
+}

abci/tests/kvstore.rs

@@ -227,7 +227,9 @@ impl Application for KVStoreABCI<'_> {
             .collect::<Option<BTreeSet<Operation>>>()
         else {
             error!("Cannot decode transactions");
-            return Err(abci::ResponseException {error:"cannot decode transactions".to_string()});
+            return Err(abci::ResponseException {
+                error: "cannot decode transactions".to_string(),
+            });
         };
 
         // Mark transactions that should be added to the proposed transactions
@@ -253,7 +255,9 @@ impl Application for KVStoreABCI<'_> {
 
         let Some(tx_records) = tx_records_encoded else {
             error!("cannot encode transactions");
-            return Err(ResponseException{error:"cannot encode transactions".to_string()});
+            return Err(ResponseException {
+                error: "cannot encode transactions".to_string(),
+            });
         };
 
         // Put both local and proposed transactions into staging area
@@ -286,7 +290,9 @@ impl Application for KVStoreABCI<'_> {
             .map(decode_transaction)
             .collect::<Option<BTreeSet<Operation>>>()
         else {
-            return Err(ResponseException{error:"cannot decode transactions".to_string()});
+            return Err(ResponseException {
+                error: "cannot decode transactions".to_string(),
+            });
         };
 
         let tx_results = tx_results_accept(td_proposed_transactions.len());

proto-compiler/src/constants.rs

@@ -90,6 +90,7 @@ pub static CUSTOM_TYPE_ATTRIBUTES: &[(&str, &str)] = &[
     (".tendermint.types.TxProof", SERIALIZED),
     (".tendermint.crypto.Proof", SERIALIZED),
     (".tendermint.abci.Response.value", DERIVE_FROM),
+    (".tendermint.abci.Request.value", DERIVE_FROM),
 ];
 
 /// Custom field attributes applied on top of protobuf fields in (a) struct(s)