Skip to content

Commit 0e1ee7c

Browse files
MicaiahReidMicaiahReid
authored
feat: add metrics to /ping response of event observer server (#297)
Fixes #285
1 parent a2d3b14 commit 0e1ee7c

File tree

5 files changed

+332
-8
lines changed

5 files changed

+332
-8
lines changed

‎components/chainhook-cli/src/service/mod.rs

+1-1
Original file line numberDiff line numberDiff line change
@@ -216,7 +216,7 @@ impl Service {
216216
match event {
217217
ObserverEvent::PredicateRegistered(spec) => {
218218
// If start block specified, use it.
219-
// I no start block specified, depending on the nature the hook, we'd like to retrieve:
219+
// If no start block specified, depending on the nature the hook, we'd like to retrieve:
220220
// - contract-id
221221
if let PredicatesApi::On(ref config) = self.config.http_api {
222222
let mut predicates_db_conn = match open_readwrite_predicates_db_conn(config)

‎components/chainhook-sdk/Cargo.toml

+1-1
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ edition = "2021"
88
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
99

1010
[dependencies]
11-
serde = "1"
11+
serde = {version = "1", features = ["rc"]}
1212
serde_json = { version = "1", features = ["arbitrary_precision"] }
1313
serde-hex = "0.1.0"
1414
serde_derive = "1"

‎components/chainhook-sdk/src/observer/http.rs

+6-3
Original file line numberDiff line numberDiff line change
@@ -8,16 +8,19 @@ use std::sync::mpsc::Sender;
88
use std::sync::{Arc, Mutex, RwLock};
99

1010
use super::{
11-
BitcoinConfig, BitcoinRPCRequest, MempoolAdmissionData, ObserverCommand,
11+
BitcoinConfig, BitcoinRPCRequest, MempoolAdmissionData, ObserverCommand, ObserverMetrics,
1212
StacksChainMempoolEvent,
1313
};
1414

1515
#[rocket::get("/ping", format = "application/json")]
16-
pub fn handle_ping(ctx: &State<Context>) -> Json<JsonValue> {
16+
pub fn handle_ping(
17+
ctx: &State<Context>,
18+
metrics_rw_lock: &State<Arc<RwLock<ObserverMetrics>>>,
19+
) -> Json<JsonValue> {
1720
ctx.try_log(|logger| slog::info!(logger, "GET /ping"));
1821
Json(json!({
1922
"status": 200,
20-
"result": "Ok",
23+
"result": metrics_rw_lock.inner(),
2124
}))
2225
}
2326

‎components/chainhook-sdk/src/observer/mod.rs

+195-3
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ use std::str;
4646
use std::str::FromStr;
4747
use std::sync::mpsc::{Receiver, Sender};
4848
use std::sync::{Arc, Mutex, RwLock};
49-
use std::time::Duration;
49+
use std::time::{Duration, SystemTime, UNIX_EPOCH};
5050
#[cfg(feature = "zeromq")]
5151
use zeromq::{Socket, SocketRecv};
5252

@@ -359,6 +359,35 @@ impl ChainhookStore {
359359
}
360360
}
361361

362+
#[derive(Debug, Default, Serialize, Clone)]
363+
pub struct ReorgMetrics {
364+
timestamp: i64,
365+
applied_blocks: usize,
366+
rolled_back_blocks: usize,
367+
}
368+
369+
#[derive(Debug, Default, Serialize, Clone)]
370+
pub struct ChainMetrics {
371+
pub tip_height: u64,
372+
pub last_reorg: Option<ReorgMetrics>,
373+
pub last_block_ingestion_at: u128,
374+
pub registered_predicates: usize,
375+
pub deregistered_predicates: usize,
376+
}
377+
378+
impl ChainMetrics {
379+
pub fn deregister_prediate(&mut self) {
380+
self.registered_predicates -= 1;
381+
self.deregistered_predicates += 1;
382+
}
383+
}
384+
385+
#[derive(Debug, Default, Serialize, Clone)]
386+
pub struct ObserverMetrics {
387+
pub bitcoin: ChainMetrics,
388+
pub stacks: ChainMetrics,
389+
}
390+
362391
pub async fn start_event_observer(
363392
mut config: EventObserverConfig,
364393
observer_commands_tx: Sender<ObserverCommand>,
@@ -409,6 +438,18 @@ pub async fn start_event_observer(
409438

410439
let background_job_tx_mutex = Arc::new(Mutex::new(observer_commands_tx.clone()));
411440

441+
let observer_metrics = ObserverMetrics {
442+
bitcoin: ChainMetrics {
443+
registered_predicates: chainhook_store.predicates.bitcoin_chainhooks.len(),
444+
..Default::default()
445+
},
446+
stacks: ChainMetrics {
447+
registered_predicates: chainhook_store.predicates.stacks_chainhooks.len(),
448+
..Default::default()
449+
},
450+
};
451+
let observer_metrics_rw_lock = Arc::new(RwLock::new(observer_metrics));
452+
412453
let limits = Limits::default().limit("json", 20.megabytes());
413454
let mut shutdown_config = config::Shutdown::default();
414455
shutdown_config.ctrlc = false;
@@ -451,6 +492,7 @@ pub async fn start_event_observer(
451492
.manage(background_job_tx_mutex)
452493
.manage(bitcoin_config)
453494
.manage(ctx_cloned)
495+
.manage(observer_metrics_rw_lock.clone())
454496
.mount("/", routes)
455497
.ignite()
456498
.await?;
@@ -470,6 +512,7 @@ pub async fn start_event_observer(
470512
observer_commands_rx,
471513
observer_events_tx,
472514
ingestion_shutdown,
515+
observer_metrics_rw_lock.clone(),
473516
ctx,
474517
)
475518
.await
@@ -653,6 +696,7 @@ pub async fn start_observer_commands_handler(
653696
observer_commands_rx: Receiver<ObserverCommand>,
654697
observer_events_tx: Option<crossbeam_channel::Sender<ObserverEvent>>,
655698
ingestion_shutdown: Option<Shutdown>,
699+
observer_metrics: Arc<RwLock<ObserverMetrics>>,
656700
ctx: Context,
657701
) -> Result<(), Box<dyn Error>> {
658702
let mut chainhooks_occurrences_tracker: HashMap<String, u64> = HashMap::new();
@@ -728,6 +772,21 @@ pub async fn start_observer_commands_handler(
728772
}
729773
};
730774
};
775+
match observer_metrics.write() {
776+
Ok(mut metrics) => {
777+
if new_block.block_identifier.index > metrics.bitcoin.tip_height {
778+
metrics.bitcoin.tip_height = new_block.block_identifier.index;
779+
}
780+
metrics.bitcoin.last_block_ingestion_at = SystemTime::now()
781+
.duration_since(UNIX_EPOCH)
782+
.expect("Could not get current time in ms")
783+
.as_millis()
784+
.into();
785+
}
786+
Err(e) => ctx.try_log(|logger| {
787+
slog::warn!(logger, "unable to acquire observer_metrics_rw_lock:{}", e)
788+
}),
789+
};
731790
bitcoin_block_store.insert(new_block.block_identifier.clone(), new_block);
732791
}
733792
ObserverCommand::CacheBitcoinBlock(block) => {
@@ -974,6 +1033,29 @@ pub async fn start_observer_commands_handler(
9741033
}
9751034
}
9761035

1036+
match blocks_to_apply
1037+
.iter()
1038+
.max_by_key(|b| b.block_identifier.index)
1039+
{
1040+
Some(highest_tip_block) => match observer_metrics.write() {
1041+
Ok(mut metrics) => {
1042+
metrics.bitcoin.last_reorg = Some(ReorgMetrics {
1043+
timestamp: highest_tip_block.timestamp.into(),
1044+
applied_blocks: blocks_to_apply.len(),
1045+
rolled_back_blocks: blocks_to_rollback.len(),
1046+
});
1047+
}
1048+
Err(e) => ctx.try_log(|logger| {
1049+
slog::warn!(
1050+
logger,
1051+
"unable to acquire observer_metrics_rw_lock:{}",
1052+
e
1053+
)
1054+
}),
1055+
},
1056+
None => {}
1057+
}
1058+
9771059
BitcoinChainEvent::ChainUpdatedWithReorg(BitcoinChainUpdatedWithReorgData {
9781060
blocks_to_apply,
9791061
blocks_to_rollback,
@@ -1108,6 +1190,17 @@ pub async fn start_observer_commands_handler(
11081190
ChainhookSpecification::Bitcoin(chainhook),
11091191
));
11101192
}
1193+
1194+
match observer_metrics.write() {
1195+
Ok(mut metrics) => metrics.bitcoin.deregister_prediate(),
1196+
Err(e) => ctx.try_log(|logger| {
1197+
slog::warn!(
1198+
logger,
1199+
"unable to acquire observer_metrics_rw_lock:{}",
1200+
e
1201+
)
1202+
}),
1203+
}
11111204
}
11121205
}
11131206

@@ -1157,6 +1250,66 @@ pub async fn start_observer_commands_handler(
11571250
stacks_chainhooks.len()
11581251
)
11591252
});
1253+
// track stacks chain metrics
1254+
match &chain_event {
1255+
StacksChainEvent::ChainUpdatedWithBlocks(update) => {
1256+
match update
1257+
.new_blocks
1258+
.iter()
1259+
.max_by_key(|b| b.block.block_identifier.index)
1260+
{
1261+
Some(highest_tip_update) => match observer_metrics.write() {
1262+
Ok(mut metrics) => {
1263+
if highest_tip_update.block.block_identifier.index
1264+
> metrics.stacks.tip_height
1265+
{
1266+
metrics.stacks.tip_height =
1267+
highest_tip_update.block.block_identifier.index;
1268+
}
1269+
metrics.stacks.last_block_ingestion_at = SystemTime::now()
1270+
.duration_since(UNIX_EPOCH)
1271+
.expect("Could not get current time in ms")
1272+
.as_millis()
1273+
.into();
1274+
}
1275+
Err(e) => ctx.try_log(|logger| {
1276+
slog::warn!(
1277+
logger,
1278+
"unable to acquire observer_metrics_rw_lock:{}",
1279+
e
1280+
)
1281+
}),
1282+
},
1283+
None => {}
1284+
}
1285+
}
1286+
StacksChainEvent::ChainUpdatedWithReorg(update) => {
1287+
match update
1288+
.blocks_to_apply
1289+
.iter()
1290+
.max_by_key(|b| b.block.block_identifier.index)
1291+
{
1292+
Some(highest_tip_update) => match observer_metrics.write() {
1293+
Ok(mut metrics) => {
1294+
metrics.stacks.last_reorg = Some(ReorgMetrics {
1295+
timestamp: highest_tip_update.block.timestamp.into(),
1296+
applied_blocks: update.blocks_to_apply.len(),
1297+
rolled_back_blocks: update.blocks_to_rollback.len(),
1298+
});
1299+
}
1300+
Err(e) => ctx.try_log(|logger| {
1301+
slog::warn!(
1302+
logger,
1303+
"unable to acquire observer_metrics_rw_lock:{}",
1304+
e
1305+
)
1306+
}),
1307+
},
1308+
None => {}
1309+
}
1310+
}
1311+
_ => {}
1312+
}
11601313

11611314
// process hooks
11621315
let (predicates_triggered, predicates_evaluated) =
@@ -1241,6 +1394,17 @@ pub async fn start_observer_commands_handler(
12411394
ChainhookSpecification::Stacks(chainhook),
12421395
));
12431396
}
1397+
1398+
match observer_metrics.write() {
1399+
Ok(mut metrics) => metrics.stacks.deregister_prediate(),
1400+
Err(e) => ctx.try_log(|logger| {
1401+
slog::warn!(
1402+
logger,
1403+
"unable to acquire observer_metrics_rw_lock:{}",
1404+
e
1405+
)
1406+
}),
1407+
}
12441408
}
12451409
}
12461410

@@ -1286,7 +1450,7 @@ pub async fn start_observer_commands_handler(
12861450
.predicates
12871451
.register_full_specification(networks, spec)
12881452
{
1289-
Ok(uuid) => uuid,
1453+
Ok(spec) => spec,
12901454
Err(e) => {
12911455
ctx.try_log(|logger| {
12921456
slog::error!(
@@ -1300,11 +1464,25 @@ pub async fn start_observer_commands_handler(
13001464
};
13011465
ctx.try_log(|logger| slog::info!(logger, "Registering chainhook {}", spec.uuid(),));
13021466
if let Some(ref tx) = observer_events_tx {
1303-
let _ = tx.send(ObserverEvent::PredicateRegistered(spec));
1467+
let _ = tx.send(ObserverEvent::PredicateRegistered(spec.clone()));
13041468
} else {
13051469
ctx.try_log(|logger| slog::info!(logger, "Enabling Predicate {}", spec.uuid()));
13061470
chainhook_store.predicates.enable_specification(&mut spec);
13071471
}
1472+
1473+
match observer_metrics.write() {
1474+
Ok(mut metrics) => match spec {
1475+
ChainhookSpecification::Bitcoin(_) => {
1476+
metrics.bitcoin.registered_predicates += 1
1477+
}
1478+
ChainhookSpecification::Stacks(_) => {
1479+
metrics.stacks.registered_predicates += 1
1480+
}
1481+
},
1482+
Err(e) => ctx.try_log(|logger| {
1483+
slog::warn!(logger, "unable to acquire observer_metrics_rw_lock:{}", e)
1484+
}),
1485+
};
13081486
}
13091487
ObserverCommand::EnablePredicate(mut spec) => {
13101488
ctx.try_log(|logger| slog::info!(logger, "Enabling Predicate {}", spec.uuid()));
@@ -1323,6 +1501,13 @@ pub async fn start_observer_commands_handler(
13231501
ChainhookSpecification::Stacks(hook),
13241502
));
13251503
}
1504+
1505+
match observer_metrics.write() {
1506+
Ok(mut metrics) => metrics.stacks.deregister_prediate(),
1507+
Err(e) => ctx.try_log(|logger| {
1508+
slog::warn!(logger, "unable to acquire observer_metrics_rw_lock:{}", e)
1509+
}),
1510+
}
13261511
}
13271512
ObserverCommand::DeregisterBitcoinPredicate(hook_uuid) => {
13281513
ctx.try_log(|logger| {
@@ -1335,6 +1520,13 @@ pub async fn start_observer_commands_handler(
13351520
let _ = tx.send(ObserverEvent::PredicateDeregistered(
13361521
ChainhookSpecification::Bitcoin(hook),
13371522
));
1523+
1524+
match observer_metrics.write() {
1525+
Ok(mut metrics) => metrics.bitcoin.deregister_prediate(),
1526+
Err(e) => ctx.try_log(|logger| {
1527+
slog::warn!(logger, "unable to acquire observer_metrics_rw_lock:{}", e)
1528+
}),
1529+
}
13381530
}
13391531
}
13401532
}

0 commit comments

Comments
 (0)