feat(aggregation-mode): add datasources and improve grafana dashboard (#2227)

Author: maximopalopoli

aggregation_mode/docker-compose.yaml

@@ -1,6 +1,11 @@
 volumes:
   postgres_data:
 
+networks:
+  aligned-network:
+    name: aligned-network
+    driver: bridge
+
 name: aggregation-mode
 services:
   postgres:
@@ -14,10 +19,15 @@ services:
       - postgres_data:/var/lib/postgresql/data
     ports:
       - 5435:5432
+    networks:
+      - aligned-network
+
   adminer:
     image: adminer
     restart: always
     depends_on:
       - postgres
     ports:
       - 8090:8080
+    networks:
+      - aligned-network

aggregation_mode/payments_poller/src/db.rs

@@ -80,4 +80,23 @@ impl Db {
             })
             .await
     }
+
+    pub async fn count_total_active_subscriptions(
+        &self,
+        epoch: BigDecimal,
+    ) -> Result<i64, sqlx::Error> {
+        self.orchestrator
+            .query(async |pool| {
+                sqlx::query_scalar::<_, i64>(
+                    "
+            SELECT COUNT(*)
+            FROM payment_events
+            WHERE started_at < $1 AND $1 < valid_until",
+                )
+                .bind(&epoch)
+                .fetch_one(&pool)
+                .await
+            })
+            .await
+    }
 }

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,