feat: aggregate metrics

migrations/20240325124203_aggregates.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS indexer_aggregates;

migrations/20240325124203_aggregates.up.sql

@@ -0,0 +1,9 @@
+CREATE TABLE IF NOT EXISTS indexer_aggregates
+(
+ id SERIAL PRIMARY KEY,
+ timestamp BIGINT NOT NULL,
+ graph_account VARCHAR(255) NOT NULL,
+ message_count BIGINT NOT NULL,
+ subgraphs_count BIGINT NOT NULL,
+ UNIQUE(graph_account, timestamp)
+);

src/db/resolver.rs

@@ -1,6 +1,6 @@
 use async_graphql::{OutputType, SimpleObject};
 use chrono::Utc;
-use serde::{de::DeserializeOwned, Serialize};
+use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use sqlx::{postgres::PgQueryResult, types::Json, FromRow, PgPool, Row as SqliteRow};
 use std::ops::Deref;
 use tracing::trace;
@@ -21,11 +21,11 @@ pub struct MessageID {
 }
 
 #[allow(dead_code)]
-#[derive(FromRow, SimpleObject, Serialize, Debug, Clone)]
+#[derive(FromRow, SimpleObject, Serialize, Deserialize, Debug, Clone)]
 pub struct IndexerStats {
- graph_account: String,
- message_count: i64,
- subgraphs_count: i64,
+ pub graph_account: String,
+ pub message_count: i64,
+ pub subgraphs_count: i64,
 }
 
 // Define graphql type for the Row in Messages
@@ -315,7 +315,7 @@ pub async fn get_indexer_stats(
  SELECT 
  message->>'graph_account' as graph_account, 
  COUNT(*) as message_count, 
- COUNT(DISTINCT message->>'identifier') as subgraphs_count -- Updated field name
+ COUNT(DISTINCT message->>'identifier') as subgraphs_count
  FROM messages 
  WHERE (CAST(message->>'nonce' AS BIGINT)) > $1";
 
@@ -352,6 +352,83 @@ pub async fn get_indexer_stats(
  Ok(stats)
 }
 
+pub async fn insert_aggregate(
+ pool: &PgPool,
+ timestamp: i64,
+ graph_account: String,
+ message_count: i64,
+ subgraphs_count: i64,
+) -> anyhow::Result<()> {
+ let _ = sqlx::query!(
+ "INSERT INTO indexer_aggregates (timestamp, graph_account, message_count, subgraphs_count) VALUES ($1, $2, $3, $4)",
+ timestamp,
+ graph_account,
+ message_count,
+ subgraphs_count
+ )
+ .execute(pool)
+ .await?;
+
+ Ok(())
+}
+
+pub async fn fetch_aggregates(
+ pool: &PgPool,
+ since_timestamp: i64,
+) -> Result<Vec<IndexerStats>, anyhow::Error> {
+ let aggregates = sqlx::query_as!(
+ IndexerStats,
+ "SELECT graph_account, message_count, subgraphs_count FROM indexer_aggregates WHERE timestamp > $1",
+ since_timestamp
+ )
+ .fetch_all(pool)
+ .await
+ .map_err(anyhow::Error::new)?;
+
+ let results: Vec<IndexerStats> = aggregates
+ .clone()
+ .into_iter()
+ .map(|agg| IndexerStats {
+ graph_account: agg.graph_account,
+ message_count: agg.message_count,
+ subgraphs_count: agg.subgraphs_count,
+ })
+ .collect();
+
+ Ok(results)
+}
+
+pub async fn count_distinct_subgraphs(
+ pool: &PgPool,
+ from_timestamp: i64,
+) -> Result<i64, anyhow::Error> {
+ let result = sqlx::query!(
+ "
+ SELECT COUNT(DISTINCT message->>'identifier') AS count
+ FROM messages
+ WHERE (CAST(message->>'nonce' AS BIGINT)) > $1
+ ",
+ from_timestamp
+ )
+ .fetch_one(pool)
+ .await
+ .map_err(anyhow::Error::new)?;
+
+ Ok(result.count.unwrap_or(0) as i64)
+}
+
+pub async fn prune_old_aggregates(pool: &PgPool) -> Result<i64, anyhow::Error> {
+ let since_timestamp = (Utc::now() - chrono::Duration::try_days(90).unwrap()).timestamp();
+ let result = sqlx::query!(
+ "DELETE FROM indexer_aggregates WHERE timestamp < $1",
+ since_timestamp
+ )
+ .execute(pool)
+ .await?;
+
+ Ok(result.rows_affected() as i64)
+}
+
 #[cfg(test)]
 mod tests {
  use crate::message_types::PublicPoiMessage;

src/operator/mod.rs

@@ -1,4 +1,5 @@
 use anyhow::anyhow;
+use chrono::Utc;
 use graphcast_sdk::WakuMessage;
 use sqlx::postgres::PgPoolOptions;
 use sqlx::{Pool, Postgres};
@@ -9,11 +10,14 @@ use std::thread::{self, JoinHandle};
 use std::time::Duration;
 use tokio::runtime::Runtime;
 use tokio::time::{interval, sleep, timeout};
-use tracing::{debug, info, trace, warn};
+use tracing::{debug, error, info, trace, warn};
 
 use graphcast_sdk::graphcast_agent::{message_typing::GraphcastMessage, GraphcastAgent};
 
-use crate::db::resolver::{count_messages, prune_old_messages, retain_max_storage};
+use crate::db::resolver::{
+ count_messages, get_indexer_stats, insert_aggregate, prune_old_aggregates, prune_old_messages,
+ retain_max_storage,
+};
 use crate::metrics::{CONNECTED_PEERS, GOSSIP_PEERS, PRUNED_MESSAGES, RECEIVED_MESSAGES};
 use crate::{
  config::Config,
@@ -110,6 +114,7 @@ impl RadioOperator {
 
  let mut network_update_interval = interval(Duration::from_secs(600));
  let mut summary_interval = interval(Duration::from_secs(180));
+ let mut daily_aggregate_interval = interval(Duration::from_secs(86400)); // 24 hours
 
  let iteration_timeout = Duration::from_secs(180);
  let update_timeout = Duration::from_secs(5);
@@ -214,6 +219,34 @@ impl RadioOperator {
  }
  }
  },
+ _ = daily_aggregate_interval.tick() => {
+ if skip_iteration.load(Ordering::SeqCst) {
+ skip_iteration.store(false, Ordering::SeqCst);
+ continue;
+ }
+
+ let pool = &self.db;
+
+ match prune_old_aggregates(pool).await {
+ Ok(deleted_count) => trace!("Pruned {} old aggregate entries.", deleted_count),
+ Err(e) => error!("Failed to prune old aggregates: {:?}", e),
+ }
+
+ let from_timestamp = (Utc::now() - Duration::from_secs(86400)).timestamp();
+
+ match get_indexer_stats(pool, None, from_timestamp).await {
+ Ok(stats) => {
+ for stat in stats {
+ match insert_aggregate(pool, Utc::now().timestamp(), stat.graph_account, stat.message_count, stat.subgraphs_count).await {
+ Ok(_) => trace!("Successfully inserted daily aggregate."),
+ Err(e) => error!("Failed to insert daily aggregate: {:?}", e),
+ }
+ }
+ },
+ Err(e) => error!("Failed to fetch Indexer stats: {:?}", e),
+ }
+ },
+
  else => break,
  }
 

src/server/model/mod.rs

@@ -3,14 +3,16 @@ use async_graphql::{Context, EmptySubscription, Object, OutputType, Schema, Simp
 use chrono::Utc;
 use serde::{de::DeserializeOwned, Serialize};
 use sqlx::{Pool, Postgres};
-use std::{sync::Arc, time::Duration};
+use std::{collections::HashMap, sync::Arc, time::Duration};
 use thiserror::Error;
+use tracing::error;
 
 use crate::{
  config::Config,
  db::resolver::{
- delete_message_all, delete_message_by_id, get_indexer_stats, list_active_indexers,
- list_messages, list_rows, message_by_id, IndexerStats,
+ count_distinct_subgraphs, delete_message_all, delete_message_by_id, fetch_aggregates,
+ get_indexer_stats, list_active_indexers, list_messages, list_rows, message_by_id,
+ IndexerStats,
  },
  operator::radio_types::RadioPayloadMessage,
 };
@@ -35,6 +37,13 @@ impl RadioContext {
  }
 }
 
+#[derive(Serialize, SimpleObject)]
+pub struct Summary {
+ total_message_count: HashMap<String, i64>,
+ average_subgraphs_count: HashMap<String, i64>,
+ total_subgraphs_covered: i64,
+}
+
 // Unified query object for resolvers
 #[derive(Default)]
 pub struct QueryRoot;
@@ -127,6 +136,62 @@ impl QueryRoot {
  message_by_id(pool, id).await?.get_message();
  Ok(msg)
  }
+
+ async fn query_aggregate_stats(
+ &self,
+ ctx: &Context<'_>,
+ days: i32,
+ ) -> Result<Summary, HttpServiceError> {
+ let pool = ctx.data_unchecked::<Pool<Postgres>>();
+
+ let since_timestamp =
+ (Utc::now() - chrono::Duration::try_days(days.into()).unwrap()).timestamp();
+ let aggregates = fetch_aggregates(pool, since_timestamp)
+ .await
+ .map_err(HttpServiceError::Others)?;
+
+ let mut total_message_count: HashMap<String, i64> = HashMap::new();
+ let mut total_subgraphs_count: HashMap<String, i64> = HashMap::new();
+
+ let mut subgraphs_counts = HashMap::new();
+
+ for stat in aggregates {
+ *total_message_count
+ .entry(stat.graph_account.clone())
+ .or_default() += stat.message_count;
+ *total_subgraphs_count
+ .entry(stat.graph_account.clone())
+ .or_default() += stat.subgraphs_count;
+ subgraphs_counts
+ .entry(stat.graph_account.clone())
+ .or_insert_with(Vec::new)
+ .push(stat.subgraphs_count);
+ }
+
+ let average_subgraphs_count: HashMap<String, i64> = total_subgraphs_count
+ .iter()
+ .map(|(key, &total_count)| {
+ let count = subgraphs_counts.get(key).map_or(1, |counts| counts.len());
+ (
+ key.clone(),
+ if count > 0 {
+ (total_count as f64 / count as f64).ceil() as i64
+ } else {
+ 0
+ },
+ )
+ })
+ .collect();
+
+ let total_subgraphs_covered = count_distinct_subgraphs(pool, since_timestamp)
+ .await
+ .map_err(HttpServiceError::Others)?;
+ Ok(Summary {
+ total_message_count,
+ average_subgraphs_count,
+ total_subgraphs_covered,
+ })
+ }
 }
 
 // Unified query object for resolvers