Add a counter for active subscriptions to the payments poller metrics

Author: maximopalopoli

aggregation_mode/payments_poller/src/db.rs

@@ -43,4 +43,21 @@ impl Db {
         .await
         .map(|_| ())
     }
+
+    pub async fn count_total_active_subscriptions(
+        &self,
+        epoch: BigDecimal,
+    ) -> Result<i64, sqlx::Error> {
+        let (count,) = sqlx::query_as::<_, (i64,)>(
+            "
+            SELECT COUNT(*)
+            FROM payment_events
+            WHERE started_at < $1 AND $1 < valid_until",
+        )
+        .bind(epoch)
+        .fetch_one(&self.pool)
+        .await?;
+
+        Ok(count)
+    }
 }

aggregation_mode/payments_poller/src/metrics.rs

@@ -6,6 +6,7 @@ use std::sync::Arc;
 pub struct PaymentsPollerMetrics {
     pub registry: Registry,
     pub last_processed_block: Gauge,
+    pub active_subscriptions: Gauge,
 }
 
 impl PaymentsPollerMetrics {
@@ -17,14 +18,21 @@ impl PaymentsPollerMetrics {
             "Last processed block by poller"
         ))?;
 
+        let active_subscriptions = Gauge::with_opts(opts!(
+            "active_subscriptions",
+            "Active payment subscriptions by poller"
+        ))?;
+
         registry.register(Box::new(last_processed_block.clone()))?;
+        registry.register(Box::new(active_subscriptions.clone()))?;
 
         // Arc is used because metrics are a shared resource accessed by both the background and metrics HTTP
         // server and the application code, across multiple Actix worker threads. The server outlives start(),
         // so the data must be static and safely shared between threads.
         let metrics = Arc::new(Self {
             registry,
             last_processed_block,
+            active_subscriptions,
         });
 
         let server_metrics = metrics.clone();
@@ -63,4 +71,8 @@ impl PaymentsPollerMetrics {
     pub fn register_last_processed_block(&self, value: u64) {
         self.last_processed_block.set(value as f64);
     }
+
+    pub fn register_active_subscriptions(&self, value: i64) {
+        self.active_subscriptions.set(value as f64);
+    }
 }

aggregation_mode/payments_poller/src/payments.rs

@@ -1,4 +1,8 @@
-use std::{str::FromStr, sync::Arc};
+use std::{
+    str::FromStr,
+    sync::Arc,
+    time::{SystemTime, UNIX_EPOCH},
+};
 
 use crate::{
     config::Config,
@@ -131,7 +135,28 @@ impl PaymentsPoller {
                 continue;
             };
 
+            let now_epoch = match SystemTime::now().duration_since(UNIX_EPOCH) {
+                Ok(duration) => duration.as_secs(),
+                Err(_) => {
+                    continue;
+                }
+            };
+
+            // Note: This implies a call to the database, and may be optimized to reduce the amount of calls
+            let Ok(active_subscriptions_amount) = self
+                .db
+                .count_total_active_subscriptions(
+                    BigDecimal::from_str(&now_epoch.to_string()).unwrap(),
+                )
+                .await
+            else {
+                tracing::error!("Failed to get the active subscriptions amount");
+                continue;
+            };
+
             self.metrics.register_last_processed_block(current_block);
+            self.metrics
+                .register_active_subscriptions(active_subscriptions_amount);
 
             tokio::time::sleep(std::time::Duration::from_secs(
                 seconds_to_wait_between_polls,