kabu_defi_health_monitor/
metrics_recorder_actor.rs

1use eyre::{Result, eyre};
2use influxdb::{Timestamp, WriteQuery};
3use kabu_core_components::Component;
4use std::sync::Arc;
5use tokio::sync::{RwLock, broadcast};
6
7use kabu_core_blockchain::{Blockchain, BlockchainState};
8use kabu_evm_db::DatabaseKabuExt;
9use kabu_types_events::MarketEvents;
10use kabu_types_market::{Market, MarketState};
11use reth_node_types::NodePrimitives;
12use reth_tasks::TaskExecutor;
13use revm::DatabaseRef;
14use std::time::Duration;
15use tikv_jemalloc_ctl::stats;
16use tokio::sync::broadcast::error::RecvError;
17use tracing::{error, info};
18
19async fn metrics_recorder_worker<DB: DatabaseKabuExt + DatabaseRef + Send + Sync + 'static>(
20    market: Arc<RwLock<Market>>,
21    market_state: Arc<RwLock<MarketState<DB>>>,
22    market_events_rx: broadcast::Sender<MarketEvents>,
23    influx_channel_tx: Option<broadcast::Sender<WriteQuery>>,
24) -> Result<()> {
25    let mut market_events_receiver = market_events_rx.subscribe();
26    loop {
27        let market_event = match market_events_receiver.recv().await {
28            Ok(event) => event,
29            Err(e) => match e {
30                RecvError::Closed => {
31                    error!("Market events channel closed");
32                    return Err(eyre!("Market events channel closed".to_string()));
33                }
34                RecvError::Lagged(lag) => {
35                    info!("Market events channel lagged: {}", lag);
36                    continue;
37                }
38            },
39        };
40
41        // Only process BlockHeaderUpdate events for metrics
42        let (block_number, timestamp) = match market_event {
43            MarketEvents::BlockHeaderUpdate { block_number, timestamp, .. } => (block_number, timestamp),
44            _ => continue, // Skip non-block events
45        };
46
47        let current_timestamp = chrono::Utc::now();
48        let block_latency = current_timestamp.timestamp() as f64 - timestamp as f64;
49
50        // check if we received twice the same block number
51
52        let allocated = stats::allocated::read().unwrap_or_default();
53
54        let market_state_guard = market_state.read().await;
55        let accounts = market_state_guard.state_db.accounts_len();
56        let contracts = market_state_guard.state_db.contracts_len();
57
58        drop(market_state_guard);
59
60        let market_guard = market.read().await;
61        let pools_disabled = market_guard.disabled_pools_count();
62        let paths = market_guard.swap_paths().len();
63        let paths_disabled = market_guard.swap_paths().disabled_len();
64        drop(market_guard);
65
66        let influx_channel_clone = influx_channel_tx.clone();
67
68        if let Some(influx_tx) = influx_channel_clone
69            && let Err(e) = tokio::time::timeout(Duration::from_secs(2), async move {
70                let write_query = WriteQuery::new(Timestamp::from(current_timestamp), "state_accounts")
71                    .add_field("value", accounts as f32)
72                    .add_field("block_number", block_number);
73                if let Err(e) = influx_tx.send(write_query) {
74                    error!("Failed to send block latency to influxdb: {:?}", e);
75                }
76
77                let write_query = WriteQuery::new(Timestamp::from(current_timestamp), "state_contracts")
78                    .add_field("value", contracts as f32)
79                    .add_field("block_number", block_number);
80                if let Err(e) = influx_tx.send(write_query) {
81                    error!("Failed to send block latency to influxdb: {:?}", e);
82                }
83
84                let write_query = WriteQuery::new(Timestamp::from(current_timestamp), "pools_disabled")
85                    .add_field("value", pools_disabled as f32)
86                    .add_field("block_number", block_number);
87                if let Err(e) = influx_tx.send(write_query) {
88                    error!("Failed to send pools_disabled to influxdb: {:?}", e);
89                }
90
91                let write_query = WriteQuery::new(Timestamp::from(current_timestamp), "paths")
92                    .add_field("value", paths as f32)
93                    .add_field("block_number", block_number);
94                if let Err(e) = influx_tx.send(write_query) {
95                    error!("Failed to send pools_disabled to influxdb: {:?}", e);
96                }
97
98                let write_query = WriteQuery::new(Timestamp::from(current_timestamp), "paths_disabled")
99                    .add_field("value", paths_disabled as f32)
100                    .add_field("block_number", block_number);
101                if let Err(e) = influx_tx.send(write_query) {
102                    error!("Failed to send pools_disabled to influxdb: {:?}", e);
103                }
104
105                let write_query = WriteQuery::new(Timestamp::from(current_timestamp), "pools_disabled")
106                    .add_field("value", pools_disabled as f32)
107                    .add_field("block_number", block_number);
108                if let Err(e) = influx_tx.send(write_query) {
109                    error!("Failed to send pools_disabled to influxdb: {:?}", e);
110                }
111
112                let write_query = WriteQuery::new(Timestamp::from(current_timestamp), "jemalloc_allocated")
113                    .add_field("value", (allocated >> 20) as f32)
114                    .add_field("block_number", block_number);
115                if let Err(e) = influx_tx.send(write_query) {
116                    error!("Failed to send jemalloc_allocator latency to influxdb: {:?}", e);
117                }
118
119                let write_query = WriteQuery::new(Timestamp::from(current_timestamp), "block_latency")
120                    .add_field("value", block_latency)
121                    .add_field("block_number", block_number);
122                if let Err(e) = influx_tx.send(write_query) {
123                    error!("Failed to send block latency to influxdb: {:?}", e);
124                }
125            })
126            .await
127        {
128            error!("Failed to send data to influxdb: {:?}", e);
129        }
130    }
131}
132
133pub struct MetricsRecorderActor<DB: Clone + Send + Sync + 'static> {
134    market: Option<Arc<RwLock<Market>>>,
135
136    market_state: Option<Arc<RwLock<MarketState<DB>>>>,
137
138    market_events_rx: Option<broadcast::Sender<MarketEvents>>,
139
140    influxdb_write_channel_tx: Option<broadcast::Sender<WriteQuery>>,
141}
142
143impl<DB> Default for MetricsRecorderActor<DB>
144where
145    DB: DatabaseRef + DatabaseKabuExt + Clone + Send + Sync + 'static,
146{
147    fn default() -> Self {
148        Self::new()
149    }
150}
151
152impl<DB> MetricsRecorderActor<DB>
153where
154    DB: DatabaseRef + DatabaseKabuExt + Clone + Send + Sync + 'static,
155{
156    pub fn new() -> Self {
157        Self { market: None, market_state: None, market_events_rx: None, influxdb_write_channel_tx: None }
158    }
159
160    pub fn on_bc<NP: NodePrimitives>(self, bc: &Blockchain<NP>, bc_state: &BlockchainState<DB, NP>) -> Self {
161        Self {
162            market: Some(bc.market()),
163            market_state: Some(bc_state.market_state()),
164            market_events_rx: Some(bc.market_events_channel()),
165            influxdb_write_channel_tx: bc.influxdb_write_channel(),
166        }
167    }
168}
169
170impl<DB> Component for MetricsRecorderActor<DB>
171where
172    DB: DatabaseRef + DatabaseKabuExt + Clone + Send + Sync + 'static,
173{
174    fn spawn(self, executor: TaskExecutor) -> Result<()> {
175        let name = self.name();
176
177        let market = self.market.ok_or_else(|| eyre!("market not set"))?;
178        let market_state = self.market_state.ok_or_else(|| eyre!("market_state not set"))?;
179        let market_events_rx = self.market_events_rx.ok_or_else(|| eyre!("market_events_rx not set"))?;
180
181        executor.spawn_critical(name, async move {
182            if let Err(e) = metrics_recorder_worker(market, market_state, market_events_rx, self.influxdb_write_channel_tx).await {
183                tracing::error!("Metrics recorder worker failed: {}", e);
184            }
185        });
186
187        Ok(())
188    }
189    fn name(&self) -> &'static str {
190        "BlockLatencyRecorderActor"
191    }
192}
kabu_defi_health_monitor/metrics_recorder_actor.rs

kabu_defi_health_monitor/
metrics_recorder_actor.rs
