Update balance & channels metrics via event

Author: Anyitechs

e2e-tests/tests/e2e.rs

@@ -810,30 +810,6 @@ async fn test_metrics_endpoint() {
 	// Set up channel and make a payment to trigger metric update
 	setup_funded_channel(&bitcoind, &server_a, &server_b, 100_000).await;
 
-	// We need to poll here because the metrics below are poll-based and
-	// updates every 60 seconds.
-	let timeout = Duration::from_secs(65);
-	let start = std::time::Instant::now();
-	loop {
-		let metrics = client.get_metrics().await.unwrap();
-		if metrics.contains("ldk_server_total_peers_count 1")
-			&& metrics.contains("ldk_server_total_channels_count 1")
-			&& metrics.contains("ldk_server_total_public_channels_count 1")
-			&& metrics.contains("ldk_server_total_payments_count 2")
-			&& !metrics.contains("ldk_server_total_lightning_balance_sats 0")
-			&& !metrics.contains("ldk_server_total_onchain_balance_sats 0")
-			&& !metrics.contains("ldk_server_spendable_onchain_balance_sats 0")
-			&& !metrics.contains("ldk_server_total_anchor_channels_reserve_sats 0")
-		{
-			break;
-		}
-
-		if start.elapsed() > timeout {
-			panic!("Timed out waiting for channel/peer metrics to update",);
-		}
-		tokio::time::sleep(Duration::from_secs(1)).await;
-	}
-
 	// Make a payment to trigger payment metric updates.
 	let invoice_resp = server_b
 		.client()
@@ -854,7 +830,13 @@ async fn test_metrics_endpoint() {
 	let start = std::time::Instant::now();
 	loop {
 		let metrics = client.get_metrics().await.unwrap();
-		if metrics.contains("ldk_server_total_successful_payments_count 1") {
+		if metrics.contains("ldk_server_total_successful_payments_count 1")
+			&& metrics.contains("ldk_server_total_channels_count 1")
+			&& !metrics.contains("ldk_server_total_lightning_balance_sats 0")
+			&& !metrics.contains("ldk_server_total_onchain_balance_sats 0")
+			&& !metrics.contains("ldk_server_spendable_onchain_balance_sats 0")
+			&& !metrics.contains("ldk_server_total_anchor_channels_reserve_sats 0")
+		{
 			break;
 		}
 		if start.elapsed() > timeout {

ldk-server/src/main.rs

@@ -327,7 +327,24 @@ fn main() {
 							if let Err(e) = event_node.event_handled() {
 								error!("Failed to mark event as handled: {e}");
 							}
+
+							if let Some(metrics) = &metrics {
+								metrics.update_channels_count(false);
+							}
 						},
+						Event::ChannelClosed { channel_id, counterparty_node_id, .. } => {
+							info!(
+								"CHANNEL_CLOSED: {} from counterparty {:?}",
+								channel_id, counterparty_node_id
+							);
+							if let Err(e) = event_node.event_handled() {
+								error!("Failed to mark event as handled: {e}");
+							}
+
+							if let Some(metrics) = &metrics {
+								metrics.update_channels_count(true);
+							}
+						}
 						Event::PaymentReceived { payment_id, payment_hash, amount_msat, .. } => {
 							info!(
 								"PAYMENT_RECEIVED: with id {:?}, hash {}, amount_msat {}",
@@ -342,6 +359,10 @@ fn main() {
 								&event_node,
 								Arc::clone(&event_publisher),
 								Arc::clone(&paginated_store)).await;
+
+							if let Some(metrics) = &metrics {
+								metrics.update_all_balances(&event_node);
+							}
 						},
 						Event::PaymentSuccessful {payment_id, ..} => {
 							let payment_id = payment_id.expect("PaymentId expected for ldk-server >=0.1");
@@ -356,6 +377,7 @@ fn main() {
 
 							if let Some(metrics) = &metrics {
 								metrics.update_payments_count(true);
+								metrics.update_all_balances(&event_node);
 							}
 						},
 						Event::PaymentFailed {payment_id, ..} => {

ldk-server/src/util/metrics.rs

@@ -84,6 +84,14 @@ impl Metrics {
 		}
 	}
 
+	pub fn update_channels_count(&self, is_closed: bool) {
+		if is_closed {
+			self.total_channels_count.fetch_sub(1, Ordering::Relaxed);
+		} else {
+			self.total_channels_count.fetch_add(1, Ordering::Relaxed);
+		}
+	}
+
 	pub fn initialize_payment_metrics(&self, node: &Node) {
 		let mut successful_payments_count = 0;
 		let mut failed_payments_count = 0;
@@ -99,9 +107,14 @@ impl Metrics {
 		self.total_successful_payments_count.store(successful_payments_count, Ordering::Relaxed);
 		self.total_failed_payments_count.store(failed_payments_count, Ordering::Relaxed);
 		self.total_pending_payments_count.store(pending_payments_count, Ordering::Relaxed);
+
+		let channels_count = node.list_channels().len() as i64;
+		self.total_channels_count.store(channels_count, Ordering::Relaxed);
+
+		self.update_all_balances(node);
 	}
 
-	fn update_all_balances(&self, node: &Node) {
+	pub fn update_all_balances(&self, node: &Node) {
 		let all_balances = node.list_balances();
 		self.total_onchain_balance_sats
 			.store(all_balances.total_onchain_balance_sats, Ordering::Relaxed);
@@ -129,9 +142,6 @@ impl Metrics {
 			.count() as i64;
 		self.total_pending_payments_count.store(pending_payments_count, Ordering::Relaxed);
 
-		let channels_count = all_channels.len() as i64;
-		self.total_channels_count.store(channels_count, Ordering::Relaxed);
-
 		let public_channels_count =
 			all_channels.iter().filter(|channel_details| channel_details.is_announced).count()
 				as i64;
@@ -143,7 +153,6 @@ impl Metrics {
 		self.total_private_channels_count.store(private_channels_count, Ordering::Relaxed);
 
 		self.update_peer_count(node);
-		self.update_all_balances(node);
 	}
 
 	/// Gathers all metrics and formats them into the Prometheus text-based format.