diff --git a/.env.example b/.env.example index 5e10d65..2ee14b7 100644 --- a/.env.example +++ b/.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 \ No newline at end of file +DISABLE_STARTUP_SCAN=false +# don't emit errors when re-checking offline hosts +AUTO_MUTE=true \ No newline at end of file diff --git a/entities/src/state/scanner.rs b/entities/src/state/scanner.rs index e7f8bf5..40e2863 100644 --- a/entities/src/state/scanner.rs +++ b/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; 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, }) } } diff --git a/scanner/src/cache_update.rs b/scanner/src/cache_update.rs index aeaddb3..e9ce26a 100644 --- a/scanner/src/cache_update.rs +++ b/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, @@ -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 = 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> { - 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> { let stats = Version::find_by_statement(Statement::from_sql_and_values( DbBackend::Sqlite, diff --git a/scanner/src/instance_check.rs b/scanner/src/instance_check.rs index 8944045..1bd8c0a 100644 --- a/scanner/src/instance_check.rs +++ b/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 { + pub(crate) async fn nitter_version(&self, url: &mut Url, mute: bool) -> Option { 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 } } diff --git a/scanner/src/lib.rs b/scanner/src/lib.rs index 6b74c7f..55d6f1a 100644 --- a/scanner/src/lib.rs +++ b/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 { + 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(&self, connection: &T) -> Result> { + 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)] diff --git a/src/main.rs b/src/main.rs index 50f710a..e3e36d8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -104,6 +104,7 @@ fn read_scanner_cfg() -> miette::Result { .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 { additional_hosts, referer, ping_range: chrono::Duration::hours(ping_range as _), + auto_mute, })) }