feat(node): new observability architecture + events

Cargo.toml

@@ -13,6 +13,7 @@ members = [
   "ipc/provider",
   "ipc/api",
   "ipc/types",
+  "ipc/observability",
 
   # ipld
   "ipld/resolver",
@@ -157,7 +158,11 @@ tokio-util = { version = "0.7.8", features = ["compat"] }
 tokio-tungstenite = { version = "0.18.0", features = ["native-tls"] }
 toml = "0.8"
 tracing = "0.1"
-tracing-subscriber = { version = "0.3", features = ["env-filter", "json", "registry"] }
+tracing-subscriber = { version = "0.3", features = [
+  "env-filter",
+  "json",
+  "registry",
+] }
 tracing-appender = "0.2.3"
 url = { version = "2.4.1", features = ["serde"] }
 zeroize = "1.6"
@@ -168,6 +173,7 @@ ipc-provider = { path = "ipc/provider" }
 ipc-wallet = { path = "ipc/wallet", features = ["with-ethers"] }
 ipc_ipld_resolver = { path = "ipld/resolver" }
 ipc-types = { path = "ipc/types" }
+ipc-observability = { path = "ipc/observability" }
 ipc_actors_abis = { path = "contracts/binding" }
 
 # Vendored for cross-compilation, see https://github.com/cross-rs/cross/wiki/Recipes#openssl

fendermint/app/Cargo.toml

@@ -64,6 +64,7 @@ fendermint_vm_resolver = { path = "../vm/resolver" }
 fendermint_vm_snapshot = { path = "../vm/snapshot" }
 fendermint_vm_topdown = { path = "../vm/topdown" }
 
+
 fvm = { workspace = true }
 fvm_ipld_blockstore = { workspace = true }
 fvm_ipld_car = { workspace = true }
@@ -72,6 +73,7 @@ fvm_shared = { workspace = true }
 ipc-api = { workspace = true }
 ipc-provider = { workspace = true }
 ipc_ipld_resolver = { workspace = true }
+ipc-observability = { workspace = true }
 
 [dev-dependencies]
 tempfile = { workspace = true }

fendermint/app/src/cmd/run.rs

@@ -4,13 +4,11 @@
 use anyhow::{anyhow, bail, Context};
 use async_stm::atomically_or_err;
 use fendermint_abci::ApplicationService;
-use fendermint_app::events::{ParentFinalityVoteAdded, ParentFinalityVoteIgnored};
 use fendermint_app::ipc::{AppParentFinalityQuery, AppVote};
 use fendermint_app::{App, AppConfig, AppStore, BitswapBlockstore};
 use fendermint_app_settings::AccountKind;
 use fendermint_crypto::SecretKey;
 use fendermint_rocksdb::{blockstore::NamespaceBlockstore, namespaces, RocksDb, RocksDbConfig};
-use fendermint_tracing::emit;
 use fendermint_vm_actor_interface::eam::EthAddress;
 use fendermint_vm_interpreter::chain::ChainEnv;
 use fendermint_vm_interpreter::fvm::upgrades::UpgradeScheduler;
@@ -22,9 +20,10 @@ use fendermint_vm_interpreter::{
 };
 use fendermint_vm_resolver::ipld::IpldResolver;
 use fendermint_vm_snapshot::{SnapshotManager, SnapshotParams};
-use fendermint_vm_topdown::proxy::IPCProviderProxy;
+use fendermint_vm_topdown::observe::register_metrics as register_topdown_metrics;
+use fendermint_vm_topdown::proxy::{IPCProviderProxy, IPCProviderProxyWithLatency};
 use fendermint_vm_topdown::sync::launch_polling_syncer;
-use fendermint_vm_topdown::voting::{publish_vote_loop, Error as VoteError, VoteTally};
+use fendermint_vm_topdown::voting::{publish_vote_loop, VoteTally};
 use fendermint_vm_topdown::{CachedFinalityProvider, IPCParentFinality, Toggle};
 use fvm_shared::address::{current_network, Address, Network};
 use ipc_ipld_resolver::{Event as ResolverEvent, VoteRecord};
@@ -71,6 +70,8 @@ async fn run(settings: Settings) -> anyhow::Result<()> {
     let metrics_registry = if settings.metrics.enabled {
         let registry = prometheus::Registry::new();
 
+        register_topdown_metrics(&registry).context("failed to register topdown metrics")?;
+
         fendermint_app::metrics::register_app_metrics(&registry)
             .context("failed to register metrics")?;
 
@@ -249,11 +250,16 @@ async fn run(settings: Settings) -> anyhow::Result<()> {
             config = config.with_max_cache_blocks(v);
         }
 
-        let ipc_provider = Arc::new(make_ipc_provider_proxy(&settings)?);
-        let finality_provider =
-            CachedFinalityProvider::uninitialized(config.clone(), ipc_provider.clone()).await?;
+        let ipc_provider = make_ipc_provider_proxy(&settings)?;
+        let ipc_provider_with_latency = Arc::new(IPCProviderProxyWithLatency::new(ipc_provider));
+
+        let finality_provider = CachedFinalityProvider::uninitialized(
+            config.clone(),
+            ipc_provider_with_latency.clone(),
+        )
+        .await?;
         let p = Arc::new(Toggle::enabled(finality_provider));
-        (p, Some((ipc_provider, config)))
+        (p, Some((ipc_provider_with_latency, config)))
     } else {
         info!("topdown finality disabled");
         (Arc::new(Toggle::disabled()), None)
@@ -533,56 +539,14 @@ async fn dispatch_vote(
                 tracing::debug!("ignoring vote; topdown disabled");
                 return;
             }
-            let res = atomically_or_err(|| {
+            let _res = atomically_or_err(|| {
                 parent_finality_votes.add_vote(
                     vote.public_key.clone(),
                     f.height,
                     f.block_hash.clone(),
                 )
             })
             .await;
-
-            let added = match res {
-                Ok(added) => {
-                    added
-                }
-                Err(e @ VoteError::Equivocation(_, _, _, _)) => {
-                    tracing::warn!(error = e.to_string(), "failed to handle vote");
-                    false
-                }
-                Err(e @ (
-                      VoteError::Uninitialized // early vote, we're not ready yet
-                    | VoteError::UnpoweredValidator(_) // maybe arrived too early or too late, or spam
-                    | VoteError::UnexpectedBlock(_, _) // won't happen here
-                )) => {
-                    tracing::debug!(error = e.to_string(), "failed to handle vote");
-                    false
-                }
-            };
-
-            let block_height = f.height;
-            let block_hash = &hex::encode(&f.block_hash);
-            let validator = &format!("{:?}", vote.public_key);
-
-            if added {
-                emit!(
-                    DEBUG,
-                    ParentFinalityVoteAdded {
-                        block_height,
-                        block_hash,
-                        validator,
-                    }
-                )
-            } else {
-                emit!(
-                    DEBUG,
-                    ParentFinalityVoteIgnored {
-                        block_height,
-                        block_hash,
-                        validator,
-                    }
-                )
-            }
         }
     }
 }

fendermint/app/src/events.rs

@@ -5,7 +5,7 @@ use crate::BlockHeight;
 
 /// Re-export other events, just to provide the visibility of where they are.
 pub use fendermint_vm_event::{
-    NewBottomUpCheckpoint, NewParentView, ParentFinalityCommitted, ParentFinalityMissingQuorum,
+    NewBottomUpCheckpoint, ParentFinalityCommitted, ParentFinalityMissingQuorum,
 };
 
 /// Hex encoded block hash.
@@ -24,20 +24,3 @@ pub struct ProposalProcessed<'a> {
 pub struct NewBlock {
     pub block_height: BlockHeight,
 }
-
-#[derive(Debug, Default)]
-pub struct ParentFinalityVoteAdded<'a> {
-    pub block_height: BlockHeight,
-    pub block_hash: BlockHashHex<'a>,
-    pub validator: &'a str,
-}
-
-#[derive(Debug, Default)]
-pub struct ParentFinalityVoteIgnored<'a> {
-    pub block_height: BlockHeight,
-    pub block_hash: BlockHashHex<'a>,
-    pub validator: &'a str,
-}
-
-// TODO: Add new events for:
-// * snapshots

fendermint/app/src/main.rs

@@ -3,73 +3,30 @@
 
 pub use fendermint_app_options as options;
 pub use fendermint_app_settings as settings;
-use tracing_appender::{
-    non_blocking::WorkerGuard,
-    rolling::{RollingFileAppender, Rotation},
+use ipc_observability::traces::{
+    register_tracing_subscriber, ConsoleLayerOpts, FileLayerOpts, WorkerGuard,
 };
-use tracing_subscriber::fmt::format::FmtSpan;
-use tracing_subscriber::{fmt, layer::SubscriberExt, Layer};
 
 mod cmd;
 
-fn init_tracing(opts: &options::Options) -> Option<WorkerGuard> {
-    let console_filter = opts.log_console_filter().expect("invalid filter");
-    let file_filter = opts.log_file_filter().expect("invalid filter");
-
-    // log all traces to stderr (reserving stdout for any actual output such as from the CLI commands)
-    let console_layer = fmt::layer()
-        .with_writer(std::io::stderr)
-        .with_target(false)
-        .with_file(true)
-        .with_line_number(true)
-        .with_filter(console_filter);
-
-    // add a file layer if log_dir is set
-    let (file_layer, file_guard) = match &opts.log_dir {
-        Some(log_dir) => {
-            let filename = match &opts.log_file_prefix {
-                Some(prefix) => format!("{}-{}", prefix, "fendermint"),
-                None => "fendermint".to_string(),
-            };
-
-            let appender = RollingFileAppender::builder()
-                .filename_prefix(filename)
-                .filename_suffix("log")
-                .rotation(Rotation::DAILY)
-                .max_log_files(7)
-                .build(log_dir)
-                .expect("failed to initialize rolling file appender");
-
-            let (non_blocking, file_guard) = tracing_appender::non_blocking(appender);
-
-            let file_layer = fmt::layer()
-                .json()
-                .with_writer(non_blocking)
-                .with_span_events(FmtSpan::CLOSE)
-                .with_target(false)
-                .with_file(true)
-                .with_line_number(true)
-                .with_filter(file_filter);
-
-            (Some(file_layer), Some(file_guard))
-        }
-        None => (None, None),
-    };
+fn init_tracing(_opts: &options::Options) -> Option<WorkerGuard> {
+    // let console_filter = opts.log_console_filter().expect("invalid filter");
+    // let file_filter = opts.log_file_filter().expect("invalid filter");
 
-    let metrics_layer = if opts.metrics_enabled() {
-        Some(fendermint_app::metrics::layer())
-    } else {
-        None
+    // TODO Karel - map the config to journal options
+    let console_opts = ConsoleLayerOpts {
+        enabled: true,
+        ..Default::default()
     };
 
-    let registry = tracing_subscriber::registry()
-        .with(console_layer)
-        .with(file_layer)
-        .with(metrics_layer);
-
-    tracing::subscriber::set_global_default(registry).expect("Unable to set a global collector");
+    // TODO Karel - map the config to journal options
+    let journal_opts = FileLayerOpts {
+        enabled: false,
+        directory: Some("/var/logs/fendermint"),
+        ..FileLayerOpts::default()
+    };
 
-    file_guard
+    register_tracing_subscriber(console_opts, journal_opts)
 }
 
 /// Install a panic handler that prints stuff to the logs, otherwise it only shows up in the console.

fendermint/app/src/metrics/tracing.rs

@@ -55,17 +55,6 @@ macro_rules! set_gauge {
     };
 }
 
-/// Set a gauge to the maximum of its value and a field in an event.
-macro_rules! max_gauge {
-    ($event:ident, $event_ty:ident :: $field:ident, $gauge:expr) => {
-        check_field!($event_ty::$field);
-        let mut fld = visitors::FindU64::new(stringify!($field));
-        $event.record(&mut fld);
-        let curr = $gauge.get();
-        $gauge.set(std::cmp::max(fld.value as i64, curr));
-    };
-}
-
 /// Increment a counter by the value of a field in the event.
 macro_rules! inc_counter {
     ($event:ident, $event_ty:ident :: $field:ident, $counter:expr) => {
@@ -119,26 +108,9 @@ macro_rules! event_match {
 impl<S: Subscriber> Layer<S> for MetricsLayer<S> {
     fn on_event(&self, event: &Event<'_>, _ctx: layer::Context<'_, S>) {
         event_match!(event {
-            NewParentView {
-                block_height              => set_gauge   ! &am::TOPDOWN_VIEW_BLOCK_HEIGHT,
-                num_msgs                  => inc_counter ! &am::TOPDOWN_VIEW_NUM_MSGS,
-                num_validator_changes     => inc_counter ! &am::TOPDOWN_VIEW_NUM_VAL_CHNGS,
-            },
             ParentFinalityCommitted {
                 block_height              => set_gauge   ! &am::TOPDOWN_FINALIZED_BLOCK_HEIGHT,
             },
-            ParentFinalityVoteAdded {
-                // This one can move up and down randomly as votes come in, but statistically should
-                // be less likely to be affected by Byzantine validators casting nonsense votes.
-                block_height              => set_gauge    ! &am::TOPDOWN_FINALITY_VOTE_BLOCK_HEIGHT,
-                // This one should only move up, showing the highest vote in the tally.
-                // It should be easy to produce this on Grafana as well from the one above.
-                block_height              => max_gauge    ! &am::TOPDOWN_FINALITY_VOTE_MAX_BLOCK_HEIGHT,
-                validator                 => inc1_counter ! &am::TOPDOWN_FINALITY_VOTE_ADDED,
-            },
-            ParentFinalityVoteIgnored {
-                validator                 => inc1_counter ! &am::TOPDOWN_FINALITY_VOTE_IGNORED,
-            },
             ParentFinalityMissingQuorum {
                 block_hash                => inc1_counter ! &am::TOPDOWN_FINALITY_MISSING_QUORUM,
             },

fendermint/vm/event/src/lib.rs

@@ -5,15 +5,6 @@ pub type BlockHeight = u64;
 /// Hex encoded block hash.
 pub type BlockHashHex<'a> = &'a str;
 
-#[derive(Debug, Default)]
-pub struct NewParentView<'a> {
-    pub is_null: bool,
-    pub block_height: BlockHeight,
-    pub block_hash: Option<BlockHashHex<'a>>, // hex encoded, unless null block
-    pub num_msgs: usize,
-    pub num_validator_changes: usize,
-}
-
 #[derive(Debug, Default)]
 pub struct ParentFinalityCommitted<'a> {
     pub block_height: BlockHeight,