Gather .health instance stats

Reworks instance checks to use the same timestamp and be transactional. Also prevents long DB lock times during list updates.

scanner/Cargo.toml

@@ -19,6 +19,8 @@ regex = { workspace = true }
 git2 = "0.17.2"
 # testing
 serde = { workspace = true, features = ["derive"] }
+# health parsing
+serde_json = "1"
 
 [dev-dependencies]
 tracing-test = { workspace = true }

scanner/src/instance_check.rs

@@ -40,6 +40,7 @@ impl Scanner {
                 .map_or(false, |check| !check.healthy);
             join_set.spawn(async move {
                 scanner.health_check_host(model, muted_host).await;
+                
             });
         }
         // wait till all of them are finished, preventing DoS

scanner/src/lib.rs

@@ -35,6 +35,7 @@ mod instance_parser;
 mod list_update;
 mod profile_parser;
 mod version_check;
+mod update_stats;
 
 const CAPTCHA_TEXT: &'static str = "Enable JavaScript and cookies to continue";
 const CAPTCHA_CODE: u16 = 403;
@@ -62,8 +63,18 @@ pub enum ScannerError {
     FetchError(#[from] FetchError),
     #[error("Failed fetching git: {0}")]
     GitFetch(#[from] git2::Error),
-    #[error("Couldn't find git branch")]
-    GitBranch,
+    /// Configured git branch can't be found
+    #[error("Couldn't find configured git branch")]
+    GitBranchNotFound,
+    /// Failed to parse .health endpoint
+    #[error("Failed to parse .health endpoint. {0} Body: {1}")]
+    StatsParsing(serde_json::Error, String),
+    /// Generic error for parsing an instance URL to [reqwest::Url]
+    #[error("Can't parse instance URL")]
+    InstanceUrlParse,
+    /// Failed to join a tokio thread in a JoinSet
+    #[error("Failed to join worker task {0}")]
+    JoinError(#[from] tokio::task::JoinError)
 }
 
 #[derive(Error, Debug)]
@@ -154,6 +165,7 @@ struct InnerScanner {
     about_parser: AboutParser,
     profile_parser: ProfileParser,
     last_list_fetch: Mutex<DateTime<Utc>>,
+    last_stats_fetch: Mutex<DateTime<Utc>>,
     last_uptime_check: Mutex<DateTime<Utc>>,
     rss_check_regex: Regex,
     client_ipv4: Client,
@@ -197,11 +209,16 @@ impl Scanner {
             .build()
             .into_diagnostic()?;
 
-        let last_uptime_check = Self::query_last_fetch(&db)
+        let last_uptime_check = Self::query_last_entry_for_table(&db,"health_check")
             .await
             .into_diagnostic()
             .wrap_err("Fetching last uptime check failed!")?;
-        tracing::info!(?last_uptime_check);
+
+        let last_stats_check = Self::query_last_entry_for_table(&db,"instance_stats")
+            .await
+            .into_diagnostic()
+            .wrap_err("Fetching last stats check failed!")?;
+        tracing::info!(?last_uptime_check, ?last_stats_check);// TODO: wrong timestamp
         let scanner = Self {
             inner: Arc::new(InnerScanner {
                 db,
@@ -215,6 +232,7 @@ impl Scanner {
                 profile_parser: ProfileParser::new(),
                 last_list_fetch: Mutex::new(last_uptime_check),
                 last_uptime_check: Mutex::new(last_uptime_check),
+                last_stats_fetch: Mutex::new(last_stats_check),
                 rss_check_regex: builder_regex_rss
                     .build()
                     .into_diagnostic()
@@ -229,8 +247,8 @@ impl Scanner {
         Ok(scanner)
     }
 
-    /// Retrieves the last uptime fetch that happened
-    pub async fn query_last_fetch(db: &DatabaseConnection) -> Result<DateTime<Utc>> {
+    /// Retrieves the last stats fetching that happened
+    pub async fn query_last_entry_for_table(db: &DatabaseConnection, table: &str) -> Result<DateTime<Utc>> {
         #[derive(Debug, FromQueryResult, Default)]
         pub(crate) struct TimeMax {
             pub max: Option<i64>,
@@ -239,7 +257,7 @@ impl Scanner {
         // will crash on an empty DB
         let time_max = TimeMax::find_by_statement(Statement::from_sql_and_values(
             DbBackend::Sqlite,
-            r#"SELECT MAX(time) FROM health_check"#,
+            format!("SELECT MAX(time) as max FROM {table}"),
             [],
         ))
         .one(db)
@@ -282,23 +300,28 @@ impl Scanner {
             }
             if self.is_instance_check_outdated() {
                 if let Err(e) = self.check_uptime().await {
-                    tracing::error!(error=?e,"Failed checking instance");
+                    tracing::error!(error=?e,"Failed checking instances for health");
+                }
+            }
+            if self.is_instance_stats_outdated() {
+                if let Err(e) = self.check_health().await {
+                    tracing::error!(error=?e,"Failed checking instances for stats");
                 }
             }
             if let Err(e) = self.update_cache().await {
-                tracing::error!(error=?e,"Failed updating cache!");
+                tracing::error!(error=?e,"Failed to update cache!");
             }
             self.sleep_till_deadline().await;
         }
     }
 
     async fn sleep_till_deadline(&self) {
-        let delay_instance_check =
-            self.last_uptime_check() + self.inner.config.instance_check_interval;
-
+        let delay_instance_check = self.last_uptime_check() + self.inner.config.instance_check_interval;
         let delay_list_update = self.last_list_fetch() + self.inner.config.list_fetch_interval;
-        tracing::debug!(?delay_list_update, ?delay_instance_check);
-        let next_deadline = delay_instance_check.min(delay_list_update);
+        let delay_stats_update = self.last_stats_fetch() + self.inner.config.instance_stats_interval;
+        tracing::debug!(?delay_list_update, ?delay_instance_check, ?delay_stats_update);
+
+        let next_deadline = delay_instance_check.min(delay_list_update).min(delay_stats_update);
         let now = Utc::now();
         let sleep_time = next_deadline.signed_duration_since(now);
         if sleep_time <= Duration::zero() {
@@ -318,6 +341,10 @@ impl Scanner {
         *self.inner.last_list_fetch.lock().unwrap()
     }
 
+    fn last_stats_fetch(&self) -> DateTime<Utc> {
+        *self.inner.last_stats_fetch.lock().unwrap()
+    }
+
     fn is_instance_check_outdated(&self) -> bool {
         let val = self.last_uptime_check();
         Utc::now().signed_duration_since(val).to_std().unwrap()
@@ -330,6 +357,12 @@ impl Scanner {
             >= self.inner.config.list_fetch_interval
     }
 
+    fn is_instance_stats_outdated(&self) -> bool {
+        let val = self.last_stats_fetch();
+        Utc::now().signed_duration_since(val).to_std().unwrap()
+            >= self.inner.config.instance_stats_interval
+    }
+
     async fn fetch_instance_list(&self) -> Result<String> {
         let (_, body) = self.fetch_url(&self.inner.config.instance_list_url).await?;
         Ok(body)

scanner/src/list_update.rs

@@ -59,6 +59,10 @@ impl Scanner {
         let found_instances: usize = parsed_instances.len();
         // find last update checks to detect spam
         let last_status = self.query_latest_check(&transaction).await?;
+        // prevent blocking the DB forever during host updates
+        // worst case we disable hosts that should be disabled and add no new ones
+        transaction.commit().await?;
+
         let mut join_set = JoinSet::new();
         for (_, instance) in parsed_instances {
             // TODO: parallelize this!
@@ -112,25 +116,30 @@ impl Scanner {
                 }
             });
         }
+
+        let mut update_models = Vec::with_capacity(join_set.len());
         while let Some(update_model) = join_set.join_next().await.map(|v| v.unwrap()) {
-            Host::insert(update_model)
-                .on_conflict(
-                    OnConflict::column(host::Column::Domain)
-                        .update_columns([
-                            host::Column::Enabled,
-                            host::Column::Updated,
-                            host::Column::Url,
-                            host::Column::Rss,
-                            host::Column::Version,
-                            host::Column::VersionUrl,
-                            host::Column::Country,
-                            host::Column::Connectivity,
-                        ])
-                        .to_owned(),
-                )
-                .exec(&transaction)
-                .await?;
+            update_models.push(update_model);
         }
+        // insert all at once
+        let transaction = self.inner.db.begin().await?;
+        Host::insert_many(update_models)
+        .on_conflict(
+            OnConflict::column(host::Column::Domain)
+                .update_columns([
+                    host::Column::Enabled,
+                    host::Column::Updated,
+                    host::Column::Url,
+                    host::Column::Rss,
+                    host::Column::Version,
+                    host::Column::VersionUrl,
+                    host::Column::Country,
+                    host::Column::Connectivity,
+                ])
+                .to_owned(),
+        )
+        .exec(&transaction)
+        .await?;
 
         transaction.commit().await?;
         let end = Instant::now();

scanner/src/update_stats.rs

@@ -0,0 +1,126 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//! for .health endpoint statistics gathering
+
+use std::time::Instant;
+
+use chrono::Utc;
+use entities::{host, instance_stats};
+use entities::prelude::Host;
+use reqwest::Url;
+use sea_orm::prelude::DateTimeUtc;
+use sea_orm::{QueryFilter, ColumnTrait, EntityTrait, ActiveValue};
+use serde::Deserialize;
+use tokio::task::JoinSet;
+
+use crate::{Result, Scanner, ScannerError};
+
+/// Instance stats reported by .health
+#[derive(Debug, Deserialize)]
+struct InstanceStats {
+    accounts: InstanceStatsAccs,
+    requests: RequestStats,
+}
+
+#[derive(Debug, Deserialize)]
+struct InstanceStatsAccs {
+    total: i32,
+    limited: i32,
+    oldest: DateTimeUtc,
+    newest: DateTimeUtc,
+    average: DateTimeUtc,
+}
+
+#[derive(Debug, Deserialize)]
+struct RequestStats {
+    total: i64,
+    apis: APIStats,
+}
+
+/// Instance api stats reported by .health
+#[derive(Debug, Deserialize)]
+#[allow(non_snake_case)]
+struct APIStats {
+    pub photoRail: i32,
+    pub userScreenName: i32,
+    pub search: i32,
+    pub listTweets: i32,
+    pub userMedia: i32,
+    pub tweetDetail: i32,
+    pub list: i32,
+    pub userTweets: i32,
+    pub userTweetsAndReplies: i32,
+}
+
+impl Scanner {
+    pub(crate) async fn check_health(&self) -> Result<()> {
+        let hosts = Host::find()
+            .filter(host::Column::Enabled.eq(true))
+            .all(&self.inner.db)
+            .await?;
+        let start = Instant::now();
+
+        let mut join_set = JoinSet::new();
+        let time = Utc::now();
+        for model in hosts.into_iter() {
+            let scanner = self.clone();
+            join_set.spawn(async move {
+                let res = scanner.fetch_instance_stats(time, &model).await;
+                if let Err(e) = &res {
+                    tracing::debug!(host=model.id, error=?e,"Failed to fetch instance stats");
+                }
+                res.ok()
+            });
+        }
+
+        let mut stat_data = Vec::with_capacity(join_set.len());
+        while let Some(join_res) = join_set.join_next().await {
+            if let Some(data) = join_res? {
+                stat_data.push(data);
+            }
+        }
+        tracing::debug!(db_stats_entries=stat_data.len());
+        if !stat_data.is_empty() {
+            instance_stats::Entity::insert_many(stat_data)
+                .exec(&self.inner.db).await?;
+        }
+
+        let end = Instant::now();
+        let duration = end - start;
+        {
+            *self.inner.last_stats_fetch.lock().unwrap() = Utc::now();
+        }
+        tracing::debug!(duration=?duration,"stats check finished");
+        Ok(())
+    }
+
+    async fn fetch_instance_stats(&self, time: DateTimeUtc, host: &host::Model) -> Result<instance_stats::ActiveModel> {
+        let mut url = Url::parse(&host.url)
+            .map_err(|e|ScannerError::InstanceUrlParse)?;
+        url.set_path(".health");
+        let (_code, body) = self.fetch_url(url.as_str()).await?;
+
+        let stats_data: InstanceStats = 
+        serde_json::from_str(&body)
+            .map_err(|e|ScannerError::StatsParsing(e,body))?;
+
+        let stats_model = instance_stats::ActiveModel {
+            time: ActiveValue::Set(time.timestamp()),
+            host: ActiveValue::Set(host.id),
+            limited_accs: ActiveValue::Set(stats_data.accounts.limited),
+            total_accs: ActiveValue::Set(stats_data.accounts.total),
+            total_requests: ActiveValue::Set(stats_data.requests.total),
+            req_photo_rail: ActiveValue::Set(stats_data.requests.apis.photoRail),
+            req_user_screen_name: ActiveValue::Set(stats_data.requests.apis.userScreenName),
+            req_search: ActiveValue::Set(stats_data.requests.apis.search),
+            req_list_tweets: ActiveValue::Set(stats_data.requests.apis.listTweets),
+            req_user_media: ActiveValue::Set(stats_data.requests.apis.userMedia),
+            req_tweet_detail: ActiveValue::Set(stats_data.requests.apis.tweetDetail),
+            req_list: ActiveValue::Set(stats_data.requests.apis.list),
+            req_user_tweets: ActiveValue::Set(stats_data.requests.apis.userTweets),
+            req_user_tweets_and_replies: ActiveValue::Set(stats_data.requests.apis.userTweetsAndReplies),
+        };
+
+        Ok(stats_model)
+    }
+}

scanner/src/version_check.rs

@@ -56,7 +56,7 @@ pub(crate) fn fetch_git_state(config: ScannerConfig) -> Result<CurrentVersion> {
             version: commit,
             config,
         })
-        .ok_or(ScannerError::GitBranch)?)
+        .ok_or(ScannerError::GitBranchNotFound)?)
 }
 
 #[cfg(test)]