automatically mute verbose logging for known broken hosts

.env.example

@@ -27,4 +27,6 @@ REFERER="https://localhost"
 # Ping AVG period in hours
 PING_RANGE_H="3"
 # enable/disable startup scanning
-DISABLE_STARTUP_SCAN=false
+DISABLE_STARTUP_SCAN=false
+# don't emit errors when re-checking offline hosts
+AUTO_MUTE=true

entities/src/state/scanner.rs

@@ -1,3 +1,4 @@
+// SPDX-License-Identifier: AGPL-3.0-only
 use std::{sync::Arc, time::Duration};
 pub type ScannerConfig = Arc<Config>;
 pub struct Config {
@@ -23,6 +24,8 @@ pub struct Config {
     pub referer: String,
     /// Duration to average the ping/response times over
     pub ping_range: chrono::Duration,
+    /// don't emit errors for hosts which are already listed as down
+    pub auto_mute: bool,
 }
 
 impl Config {
@@ -39,6 +42,7 @@ impl Config {
             additional_hosts: vec![String::from("https://nitter.net")],
             referer: String::from(""),
             ping_range: chrono::Duration::hours(3),
+            auto_mute: true,
         })
     }
 }

scanner/src/cache_update.rs

@@ -12,6 +12,7 @@ use sea_orm::QueryFilter;
 use sea_orm::QueryOrder;
 use sea_orm::{prelude::DateTimeUtc, DbBackend, FromQueryResult, Statement};
 
+use crate::LatestCheck;
 use crate::{Result, Scanner};
 
 #[derive(Debug, FromQueryResult)]
@@ -21,12 +22,6 @@ pub struct HostStats {
     total: u32,
 }
 
-#[derive(Debug, FromQueryResult, Default)]
-pub struct LatestCheck {
-    host: i32,
-    healthy: bool,
-}
-
 #[derive(Debug, Default)]
 struct LastPings {
     avg: Option<i32>,
@@ -72,7 +67,7 @@ impl Scanner {
             stats_120d.into_iter().map(|v| (v.host, v)).collect();
 
         let version_points = self.query_versions(time_30d).await?;
-        let latest_check = self.query_latest_check().await?;
+        let latest_check = self.query_latest_check(&self.inner.db).await?;
         let latest_check: HashMap<i32, LatestCheck> =
             latest_check.into_iter().map(|v| (v.host, v)).collect();
 
@@ -193,26 +188,6 @@ impl Scanner {
         Ok(map)
     }
 
-    async fn query_latest_check(&self) -> Result<Vec<LatestCheck>> {
-        let upte_checks = LatestCheck::find_by_statement(Statement::from_sql_and_values(
-            DbBackend::Sqlite,
-            r#"
-            WITH latest AS(
-                SELECT u.host,MAX(u.time) as time FROM update_check u
-                GROUP BY u.host
-            )
-            SELECT u.host,u.time,healthy FROM update_check u
-            JOIN host h ON h.id = u.host
-            JOIN latest l ON l.host = u.host AND l.time = u.time
-            WHERE h.enabled = true
-            "#,
-            [],
-        ))
-        .all(&self.inner.db)
-        .await?;
-        Ok(upte_checks)
-    }
-
     async fn query_versions(&self, age: DateTimeUtc) -> Result<HashMap<String, f64>> {
         let stats = Version::find_by_statement(Statement::from_sql_and_values(
             DbBackend::Sqlite,

scanner/src/instance_check.rs

@@ -29,10 +29,13 @@ impl Scanner {
 
         let mut join_set = JoinSet::new();
 
+        let last_check = self.query_latest_check(&self.inner.db).await?;
+
         for model in hosts.into_iter() {
             let scanner = self.clone();
+            let muted_host = last_check.iter().find(|v|v.host == model.id).map_or(false,|check|!check.healthy);
             join_set.spawn(async move {
-                scanner.update_check_host(model).await;
+                scanner.update_check_host(model, muted_host).await;
             });
         }
         // wait till all of them are finished, preventing DoS
@@ -46,11 +49,13 @@ impl Scanner {
     }
 
     #[instrument]
-    async fn update_check_host(&self, host: host::Model) {
+    async fn update_check_host(&self, host: host::Model, muted: bool) {
         let now = Utc::now();
         let mut url = match Url::parse(&host.url) {
             Err(e) => {
-                tracing::error!(error=?e, url=host.url,"failed to parse instance URL");
+                if !muted {
+                    tracing::error!(error=?e, url=host.url,"failed to parse instance URL");
+                }
                 self.insert_failed_update_check(host.id, now, None, None)
                     .await;
                 return;
@@ -64,11 +69,13 @@ impl Scanner {
         let took_ms = end.saturating_duration_since(start).as_millis();
         match fetch_res {
             Err(e) => {
-                tracing::info!(
-                    host = host.url,
-                    took = took_ms,
-                    "couldn't ping host: {e}, marking as dead"
-                );
+                if !muted {
+                    tracing::info!(
+                        host = host.url,
+                        took = took_ms,
+                        "couldn't ping host: {e}, marking as dead"
+                    );
+                }
                 let (http_code, resp_time) = match e {
                     FetchError::HttpResponseStatus(code, _, _) => {
                         (Some(code as _), Some(took_ms as _))
@@ -79,13 +86,17 @@ impl Scanner {
                     .await;
             }
             Ok((http_code, content)) => {
-                tracing::trace!(host = host.url, took = took_ms);
+                if !muted {
+                    tracing::trace!(host = host.url, took = took_ms);
+                }
                 // check for valid profile
                 if !self.inner.health_check_regex.is_match(&content) {
-                    tracing::info!(
-                        content = content,
-                        "host doesn't contain expected profile content"
-                    );
+                    if !muted {
+                        tracing::info!(
+                            content = content,
+                            "host doesn't contain expected profile content"
+                        );
+                    }
                     self.insert_failed_update_check(
                         host.id,
                         now,
@@ -113,36 +124,45 @@ impl Scanner {
     }
 
     /// Check if rss is available
-    pub(crate) async fn has_rss(&self, url: &mut Url) -> bool {
+    pub(crate) async fn has_rss(&self, url: &mut Url, mute: bool) -> bool {
         url.set_path(&self.inner.config.rss_path);
         match self.fetch_url(url.as_str()).await {
             Ok((code, content)) => match self.inner.rss_check_regex.is_match(&content) {
                 true => return true,
                 false => {
-                    tracing::debug!(code = code, content = content, "rss content not found");
+                    if !mute {
+                        // 404 = disabled
+                        tracing::debug!(url=url.as_str(), code = code, content = content, "rss content not found");
+                    }
                     return false;
                 }
             },
             Err(e) => {
-                tracing::debug!(error=?e,"fetching rss feed failed");
+                if !mute && e.http_status_code() != Some(404) {
+                    tracing::debug!(error=?e,url=url.as_str(),"fetching rss feed failed");
+                }
                 return false;
             }
         }
     }
 
     /// Check nitter version
-    pub(crate) async fn nitter_version(&self, url: &mut Url) -> Option<String> {
+    pub(crate) async fn nitter_version(&self, url: &mut Url, mute: bool) -> Option<String> {
         url.set_path(&self.inner.config.about_path);
         match self.fetch_url(url.as_str()).await {
             Ok((code, content)) => match self.inner.about_parser.parse_about_version(&content) {
                 Ok(v) => Some(v),
                 Err(e) => {
-                    tracing::debug!(url=url.as_str(),code,content,error=?e,"failed parsing version from about page");
+                    if !mute {
+                        tracing::debug!(url=url.as_str(),code,content,error=?e,"failed parsing version from about page");
+                    }
                     None
                 }
             },
             Err(e) => {
-                tracing::debug!(url=url.as_str(),error=?e,"failed fetching about page");
+                if !mute {
+                    tracing::debug!(url=url.as_str(),error=?e,"failed fetching about page");
+                }
                 None
             }
         }

scanner/src/lib.rs

@@ -19,7 +19,7 @@ use reqwest::{
 };
 use sea_orm::{
     sea_query::OnConflict, ActiveModelTrait, ActiveValue, ColumnTrait, DatabaseConnection,
-    EntityTrait, QueryFilter, TransactionTrait,
+    EntityTrait, QueryFilter, TransactionTrait, FromQueryResult, Statement, DbBackend, ConnectionTrait,
 };
 use thiserror::Error;
 use tokio::{task::JoinSet, time::sleep};
@@ -74,6 +74,20 @@ pub enum FetchError {
     Captcha,
 }
 
+impl FetchError {
+    /// Returns the http status code, if applicable
+    fn http_status_code(&self) -> Option<u16> {
+        match self {
+            FetchError::Reqwest(e) => {
+                e.status().map(|v|v.as_u16())
+            },
+            FetchError::HttpResponseStatus(code, _, _) => Some(*code),
+            FetchError::KnownHttpResponseStatus(code, _) => Some(*code),
+            FetchError::Captcha | FetchError::RetrievingBody(_, _) => None,
+        }
+    }
+}
+
 pub fn run_scanner(db: DatabaseConnection, config: ScannerConfig, cache: Cache, disable_startup_scan: bool) -> Result<()> {
     let scanner = Scanner::new(db, config, cache);
 
@@ -102,6 +116,13 @@ struct InnerScanner {
     rss_check_regex: Regex,
 }
 
+#[derive(Debug, FromQueryResult, Default)]
+pub struct LatestCheck {
+    pub host: i32,
+    pub healthy: bool,
+    pub domain: String,
+}
+
 impl Scanner {
     fn new(db: DatabaseConnection, config: ScannerConfig, cache: Cache) -> Self {
         let mut headers = HeaderMap::with_capacity(HEADERS.len());
@@ -229,19 +250,29 @@ impl Scanner {
         }
         // now update/insert the existing ones
         let found_instances: usize = parsed_instances.len();
+        // find last update checks to detect spam
+        let last_status = self.query_latest_check(&transaction).await?;
         let mut join_set = JoinSet::new();
         for (_, instance) in parsed_instances {
             // TODO: parallelize this!
             let scanner_c = self.clone();
+            // detect already offline host and prevent log spam
+            let muted_host = match self.inner.config.auto_mute {
+                false => false,
+                true => last_status.iter().find(|v|v.domain == instance.domain).map_or(false,|check|!check.healthy),
+            };
+            // tracing::trace!(muted_host,instance=?instance,last_status=?last_status);
             join_set.spawn(async move {
                 let (rss, version) = match Url::parse(&instance.url) {
                     Err(_) => {
-                        tracing::info!(url = instance.url, "Instance URL invalid");
+                        if !muted_host {
+                            tracing::info!(url = instance.url, "Instance URL invalid");
+                        }
                         (false, None)
                     }
                     Ok(mut url) => {
-                        let rss = scanner_c.has_rss(&mut url).await;
-                        let version = scanner_c.nitter_version(&mut url).await;
+                        let rss = scanner_c.has_rss(&mut url, muted_host).await;
+                        let version = scanner_c.nitter_version(&mut url, muted_host).await;
                         (rss, version)
                     }
                 };
@@ -334,6 +365,26 @@ impl Scanner {
 
         Ok((code, body))
     }
+
+    pub async fn query_latest_check<T: ConnectionTrait>(&self, connection: &T) -> Result<Vec<LatestCheck>> {
+        let update_checks = LatestCheck::find_by_statement(Statement::from_sql_and_values(
+            DbBackend::Sqlite,
+            r#"
+            WITH latest AS(
+                SELECT u.host,MAX(u.time) as time FROM update_check u
+                GROUP BY u.host
+            )
+            SELECT u.host,healthy,h.domain FROM update_check u
+            JOIN host h ON h.id = u.host
+            JOIN latest l ON l.host = u.host AND l.time = u.time
+            WHERE h.enabled = true
+            "#,
+            [],
+        ))
+        .all(connection)
+        .await?;
+        Ok(update_checks)
+    }
 }
 
 #[cfg(test)]

src/main.rs

@@ -104,6 +104,7 @@ fn read_scanner_cfg() -> miette::Result<ScannerConfig> {
         .map(|v| v.trim().to_owned())
         .collect();
     let referer = require_env_str("REFERER")?;
+    let auto_mute = require_env_str("AUTO_MUTE")? == "true";
 
     Ok(Arc::new(entities::state::scanner::Config {
         list_fetch_interval: Duration::from_secs(instance_list_interval),
@@ -117,6 +118,7 @@ fn read_scanner_cfg() -> miette::Result<ScannerConfig> {
         additional_hosts,
         referer,
         ping_range: chrono::Duration::hours(ping_range as _),
+        auto_mute,
     }))
 }