diff --git a/.env.example b/.env.example index 4c08072..50296bb 100644 --- a/.env.example +++ b/.env.example @@ -51,4 +51,6 @@ CLEANUP_INTERVAL_S=86400 # amount of errors to keep per host ERROR_RETENTION_PER_HOST=100 # session database URI -SESSION_DB_URI="sqlite:./sessions.db?mode=rwc" \ No newline at end of file +SESSION_DB_URI="sqlite:./sessions.db?mode=rwc" +# interval in seconds for gathering .health endpoint data +STATS_INTERVAL_S=900 \ No newline at end of file diff --git a/.gitignore b/.gitignore index 14f914c..a0498ca 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,4 @@ /.env /sqlite.db-journal /out.txt +/sessions.db diff --git a/Cargo.lock b/Cargo.lock index a77d1e2..46aca48 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2244,7 +2244,7 @@ checksum = "e4a24736216ec316047a1fc4252e27dabb04218aa4a3f37c6e7ddbf1f9782b54" [[package]] name = "nitter-status" -version = "0.4.0" +version = "0.5.0" dependencies = [ "chrono", "clap", @@ -3229,6 +3229,7 @@ dependencies = [ "sea-orm", "sea-query", "serde", + "serde_json", "thiserror", "tokio", "tracing", diff --git a/Cargo.toml b/Cargo.toml index a01846f..3bde907 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "nitter-status" -version = "0.4.0" +version = "0.5.0" edition = "2021" publish = false license = "AGPL-3.0-only" @@ -15,7 +15,7 @@ chrono = "0.4.22" serde = "1.0.144" sea-orm-migration = "0.12.0" sea-orm = "0.12.0" -sea-query = "0.30" # keep in step sea-orm +sea-query = "0.30" # keep in step with sea-orm futures = "0.3" tracing = "0.1" tracing-subscriber = "0.3" diff --git a/entities/src/health_check.rs b/entities/src/health_check.rs index 8808c0a..a095e38 100644 --- a/entities/src/health_check.rs +++ b/entities/src/health_check.rs @@ -46,6 +46,7 @@ pub struct HealthyAmount { } impl HealthyAmount { + /// Fetch health check graph data for all or selected hosts in the selected time range. pub async fn fetch(db: &DatabaseConnection, from: DateTimeUtc, to: DateTimeUtc, hosts: Option<&[i32]>) -> Result,DbErr> { let builder = db.get_database_backend(); let mut stmt: sea_query::SelectStatement = Query::select(); @@ -90,6 +91,6 @@ mod test { .checked_sub_signed(chrono::Duration::days(14)) .unwrap(); let hosts = vec![1,2,3]; - dbg!(HealthyAmount::fetch(&db, time_3h, time_now,Some(&hosts)).await.unwrap()); + HealthyAmount::fetch(&db, time_3h, time_now,Some(&hosts)).await.unwrap(); } } \ No newline at end of file diff --git a/entities/src/instance_stats.rs b/entities/src/instance_stats.rs new file mode 100644 index 0000000..eae4fda --- /dev/null +++ b/entities/src/instance_stats.rs @@ -0,0 +1,98 @@ +//! `SeaORM` Entity. Generated by sea-orm-codegen 0.11.3 + +use sea_orm::{entity::prelude::*, FromQueryResult}; +use sea_query::{Query, SimpleExpr, Order, Alias}; +use serde::Serialize; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize)] +#[sea_orm(table_name = "instance_stats")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub time: i64, + #[sea_orm(primary_key, auto_increment = false)] + pub host: i32, + pub limited_accs: i32, + pub total_accs: i32, + pub total_requests: i64, + pub req_photo_rail: i32, + pub req_user_screen_name: i32, + pub req_search: i32, + pub req_list_tweets: i32, + pub req_user_media: i32, + pub req_tweet_detail: i32, + pub req_list: i32, + pub req_user_tweets: i32, + pub req_user_tweets_and_replies: i32, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation { + #[sea_orm( + belongs_to = "super::host::Entity", + from = "Column::Host", + to = "super::host::Column::Id", + on_update = "Cascade", + on_delete = "Cascade" + )] + Host, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::Host.def() + } +} + +impl ActiveModelBehavior for ActiveModel {} + +#[derive(Debug, FromQueryResult, Serialize)] +pub struct StatsAmount { + pub time: i64, + pub limited_accs_max: i32, + pub limited_accs_avg: i32, + pub total_accs_max: i32, + pub total_accs_avg: i32, + pub total_requests_max: i64, + pub total_requests_avg: i64, + pub req_photo_rail_max: i32, + pub req_photo_rail_avg: i32, + pub req_user_screen_name_max: i32, + pub req_user_screen_name_avg: i32, + pub req_search_max: i32, + pub req_search_avg: i32, + pub req_list_tweets_max: i32, + pub req_list_tweets_avg: i32, + pub req_user_media_max: i32, + pub req_user_media_avg: i32, + pub req_tweet_detail_max: i32, + pub req_tweet_detail_avg: i32, + pub req_list_max: i32, + pub req_list_avg: i32, + pub req_user_tweets_max: i32, + pub req_user_tweets_avg: i32, + pub req_user_tweets_and_replies_max: i32, + pub req_user_tweets_and_replies_avg: i32, +} + +impl StatsAmount { + /// Fetch health check graph data for all or selected hosts in the selected time range. + pub async fn fetch(db: &DatabaseConnection, from: DateTimeUtc, to: DateTimeUtc, hosts: Option<&[i32]>) -> Result,DbErr> { + let builder = db.get_database_backend(); + let columns = ["limited_accs","total_accs","total_requests","req_photo_rail","req_user_screen_name","req_search","req_list_tweets","req_user_media","req_tweet_detail","req_list","req_user_tweets","req_user_tweets_and_replies"]; + let mut stmt: sea_query::SelectStatement = Query::select(); + let col_stmt = stmt.column(self::Column::Time); + for col in columns { + col_stmt.expr_as(SimpleExpr::Custom(format!("MAX({col})")), Alias::new(format!("{col}_max"))) + .expr_as(SimpleExpr::Custom(format!("CAST(ifnull(AVG({col}),0) as int)")), Alias::new(format!("{col}_avg"))); + } + col_stmt.group_by_col(self::Column::Time) + .from(self::Entity) + .and_where(self::Column::Time.between(from.timestamp(), to.timestamp())); + if let Some(hosts) = hosts { + stmt.and_where(self::Column::Host.is_in(hosts.iter().map(|v| *v))); + } + stmt.group_by_col(self::Column::Time) + .order_by(self::Column::Time, Order::Asc); + StatsAmount::find_by_statement(builder.build(&stmt)).all(db).await + } +} \ No newline at end of file diff --git a/entities/src/mod.rs b/entities/src/mod.rs index 0e5fa18..7ceaffe 100644 --- a/entities/src/mod.rs +++ b/entities/src/mod.rs @@ -5,6 +5,7 @@ pub mod prelude; pub mod check_errors; pub mod health_check; pub mod host; +pub mod instance_stats; // has to be re-added on entity regeneration -pub mod state; +pub mod state; \ No newline at end of file diff --git a/entities/src/prelude.rs b/entities/src/prelude.rs index 1d8c268..4004557 100644 --- a/entities/src/prelude.rs +++ b/entities/src/prelude.rs @@ -3,3 +3,4 @@ pub use super::check_errors::Entity as CheckErrors; pub use super::health_check::Entity as HealthCheck; pub use super::host::Entity as Host; +pub use super::instance_stats::Entity as InstanceStats; diff --git a/entities/src/state/scanner.rs b/entities/src/state/scanner.rs index 68466f3..9dce4e5 100644 --- a/entities/src/state/scanner.rs +++ b/entities/src/state/scanner.rs @@ -7,6 +7,8 @@ pub struct Config { pub list_fetch_interval: Duration, /// time until next instance ping check pub instance_check_interval: Duration, + /// time until next instance statistics check + pub instance_stats_interval: Duration, /// instances list URL pub instance_list_url: String, /// profile path for health check @@ -48,8 +50,9 @@ pub struct Config { impl Config { pub fn test_defaults() -> ScannerConfig { Arc::new(Config { - list_fetch_interval: Duration::from_secs(60 * 5), - instance_check_interval: Duration::from_secs(60 * 5), + instance_stats_interval: Duration::from_secs(15 * 60), + list_fetch_interval: Duration::from_secs(15 * 60), + instance_check_interval: Duration::from_secs(15 * 60), instance_list_url: String::from("https://github.com/zedeus/nitter/wiki/Instances"), profile_path: String::from("/jack"), rss_path: String::from("/jack/rss"), diff --git a/migration/src/lib.rs b/migration/src/lib.rs index 6824fd4..f14722c 100644 --- a/migration/src/lib.rs +++ b/migration/src/lib.rs @@ -7,6 +7,7 @@ mod m20230803_154714_version_url; mod m20230829_201916_country; mod m20230914_231514_connectivity; mod m20231011_231223_errors; +mod m20231112_142206_stats; pub struct Migrator; @@ -21,6 +22,7 @@ impl MigratorTrait for Migrator { Box::new(m20230829_201916_country::Migration), Box::new(m20230914_231514_connectivity::Migration), Box::new(m20231011_231223_errors::Migration), + Box::new(m20231112_142206_stats::Migration), ] } } diff --git a/migration/src/m20231112_142206_stats.rs b/migration/src/m20231112_142206_stats.rs new file mode 100644 index 0000000..97e408a --- /dev/null +++ b/migration/src/m20231112_142206_stats.rs @@ -0,0 +1,39 @@ +use sea_orm_migration::prelude::*; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let cmd = r#"CREATE TABLE "instance_stats" ( + "time" integer NOT NULL, + "host" integer NOT NULL, + "limited_accs" integer NOT NULL, + "total_accs" integer NOT NULL, + "total_requests" integer NOT NULL, + "req_photo_rail" integer NOT NULL, + "req_user_screen_name" integer NOT NULL, + "req_search" integer NOT NULL, + "req_list_tweets" integer NOT NULL, + "req_user_media" integer NOT NULL, + "req_tweet_detail" integer NOT NULL, + "req_list" integer NOT NULL, + "req_user_tweets" integer NOT NULL, + "req_user_tweets_and_replies" integer NOT NULL, + CONSTRAINT "pk_instance_stats" PRIMARY KEY ("host", "time"), + FOREIGN KEY ("host") REFERENCES "host" ("id") ON DELETE CASCADE ON UPDATE CASCADE + ) WITHOUT ROWID, STRICT;"#; + let db = manager.get_connection(); + db.execute_unprepared("BEGIN EXCLUSIVE").await?; + tracing::info!("adding instance_stats table.."); + db.execute_unprepared(cmd).await?; + db.execute_unprepared("COMMIT TRANSACTION").await?; + db.execute_unprepared("VACUUM").await?; + Ok(()) + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + panic!("Can't migrate down"); + } +} diff --git a/scanner/Cargo.toml b/scanner/Cargo.toml index 6629f5d..5112f23 100644 --- a/scanner/Cargo.toml +++ b/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 } diff --git a/scanner/src/instance_check.rs b/scanner/src/instance_check.rs index c63d34b..866ab0a 100644 --- a/scanner/src/instance_check.rs +++ b/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 diff --git a/scanner/src/lib.rs b/scanner/src/lib.rs index a725362..ed6cb17 100644 --- a/scanner/src/lib.rs +++ b/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>, + last_stats_fetch: Mutex>, last_uptime_check: Mutex>, 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> { + /// Retrieves the last stats fetching that happened + pub async fn query_last_entry_for_table(db: &DatabaseConnection, table: &str) -> Result> { #[derive(Debug, FromQueryResult, Default)] pub(crate) struct TimeMax { pub max: Option, @@ -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 { + *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 { let (_, body) = self.fetch_url(&self.inner.config.instance_list_url).await?; Ok(body) diff --git a/scanner/src/list_update.rs b/scanner/src/list_update.rs index 103a109..c387aa3 100644 --- a/scanner/src/list_update.rs +++ b/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(); diff --git a/scanner/src/update_stats.rs b/scanner/src/update_stats.rs new file mode 100644 index 0000000..4901dae --- /dev/null +++ b/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 { + 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) + } +} \ No newline at end of file diff --git a/scanner/src/version_check.rs b/scanner/src/version_check.rs index 12daa0a..82fbe5f 100644 --- a/scanner/src/version_check.rs +++ b/scanner/src/version_check.rs @@ -56,7 +56,7 @@ pub(crate) fn fetch_git_state(config: ScannerConfig) -> Result { version: commit, config, }) - .ok_or(ScannerError::GitBranch)?) + .ok_or(ScannerError::GitBranchNotFound)?) } #[cfg(test)] diff --git a/server/src/admin.rs b/server/src/admin.rs index 2cea0d0..e95fcd2 100644 --- a/server/src/admin.rs +++ b/server/src/admin.rs @@ -14,6 +14,7 @@ use constant_time_eq::constant_time_eq; use entities::check_errors; use entities::health_check; use entities::host; +use entities::instance_stats; use entities::state::AppState; use hyper::header::REFERER; use hyper::HeaderMap; @@ -319,15 +320,18 @@ pub async fn history_json( struct ReturnData { pub global: Vec, pub user: Vec, + pub stats: Vec, } let (_login, hosts) = get_all_login_hosts(&session, db, false).await?; let host_ids: Vec<_> = hosts.iter().map(|host|host.id).collect(); let data_owned = health_check::HealthyAmount::fetch(db, input.start, input.end,Some(&host_ids)).await?; let data_global = health_check::HealthyAmount::fetch(db, input.start, input.end,None).await?; + let data_stats = instance_stats::StatsAmount::fetch(db, input.start, input.end,None).await?; Ok(Json(ReturnData { global: data_global, user: data_owned, + stats: data_stats, }).into_response()) } @@ -338,15 +342,29 @@ pub async fn history_json_specific( Json(input): Json, ) -> Result { let host = get_specific_login_host(host, &session, db).await?; + #[derive(Debug, Serialize)] + struct ReturnData { + pub stats: Vec, + pub health: Vec, + } - let history: Vec = health_check::Entity::find() + let history_health: Vec = health_check::Entity::find() .filter(health_check::Column::Host.eq(host.id)) .order_by_asc(health_check::Column::Time) .filter(health_check::Column::Time.between(input.start.timestamp(), input.end.timestamp())) .all(db) .await?; + let history_stats: Vec = instance_stats::Entity::find() + .filter(instance_stats::Column::Host.eq(host.id)) + .order_by_asc(instance_stats::Column::Time) + .filter(instance_stats::Column::Time.between(input.start.timestamp(), input.end.timestamp())) + .all(db) + .await?; - Ok(Json(history).into_response()) + Ok(Json(ReturnData{ + health: history_health, + stats: history_stats + }).into_response()) } pub async fn history_view( diff --git a/server/static/admin.js b/server/static/admin.js index f993cb2..255008b 100644 --- a/server/static/admin.js +++ b/server/static/admin.js @@ -39,73 +39,209 @@ async function fetchDataAndCreateChart(startDate,endDate) { } catch (error) { console.error('Failed to fetch data:', error); } + // Create arrays for y-values for both datasets + const healthyData = data.global.map(entry => { + return {x: moment.unix(entry.time), y: entry.alive}; + }); - // Extracting data for chart - const timestamps = data.global.map(entry => entry.time * 1000); // Converting UNIX timestamps to milliseconds - const healthyData = data.global.map(entry => entry.alive); - const unhealthyData = data.global.map(entry => entry.dead); - const userUnhealthyData = data.user.filter(entry => entry.dead > 0).map(entry => ({"x": entry.time * 1000, "y": entry.dead})); - const paramData = {"global": {"healthyData": healthyData, "unhealthyData": unhealthyData}, "user": userUnhealthyData}; - console.log(paramData); - createChart(timestamps, paramData); + const unhealthyData = data.global.map(entry => { + return {x: moment.unix(entry.time), y: entry.dead}; + }); + + let statsData = []; + let statsKeys = []; + if (data.stats.length > 0) { + + statsKeys = Object.keys(data.stats[0]).filter(key => key !== 'time'); + statsData = statsKeys.map(key => ({"key": key, data: data.stats.map(entry => ({x: moment.unix(entry.time), y: entry[key]}))})); + } + + const userUnhealthyData = data.user.filter(entry => entry.dead > 0).map(entry => ({"x": entry.time * 1000, "y": entry.dead})); + const paramData = {"healthyData": healthyData, "unhealthyData": unhealthyData, "user": userUnhealthyData, "stats": statsData, "statsKeys": statsKeys}; + + createChartOverview(paramData); } -function createChart(timestamps, data) { - const ctx = document.getElementById('graph').getContext('2d'); - let high_data = timestamps.length > 1000; +function mapKey(key) { + const keyMappings = { + "limited_accs_max": "Max Lim. Accs", + "limited_accs_avg": "Avg Lim. Accs", + "total_accs_max": "Max Accs", + "total_accs_avg": "Avg Accs", + "total_requests_max": "Max Requests Req.", + "total_requests_avg": "Avg Requests Req.", + "req_photo_rail_max": "Max PhotoRail Req.", + "req_photo_rail_avg": "Avg PhotoRail Req.", + "req_user_screen_name_max": "Max UserScreenName Req.", + "req_user_screen_name_avg": "Avg UserScreenName Req.", + "req_search_max": "Max Search Req.", + "req_search_avg": "Avg Search Req.", + "req_list_tweets_max": "Max ListTweets Req.", + "req_list_tweets_avg": "Avg ListTweets Req.", + "req_user_media_max": "Max UserMedia Req.", + "req_user_media_avg": "Avg UserMedia Req.", + "req_tweet_detail_max": "Max TweetDetail Req.", + "req_tweet_detail_avg": "Avg TweetDetail Req.", + "req_list_max": "Max List Req.", + "req_list_avg": "Avg List Req.", + "req_user_tweets_max": "Max UserTweets Req.", + "req_user_tweets_avg": "Avg UserTweets Req.", + "req_user_tweets_and_replies_max": "Max UserTweetsAndReplies Req.", + "req_user_tweets_and_replies_avg": "Avg UserTweetsAndReplies Req.", + } + return keyMappings[key]; +} + +function nameToColor(name) { + const colors = ["#fafa6e" + ,"#eafb71" + ,"#dafb75" + ,"#cafb7b" + ,"#b9fa81" + ,"#a8f989" + ,"#97f890" + ,"#85f799" + ,"#71f5a1" + ,"#5cf3aa" + ,"#42f1b3" + ,"#15efbc" + ,"#00ecc4" + ,"#00e9cd" + ,"#00e6d5" + ,"#00e3dd" + ,"#00e0e4" + ,"#00dcea" + ,"#00d8f0" + ,"#00d5f5" + ,"#00d0f9" + ,"#00ccfd" + ,"#00c8ff" + ,"#00c3ff" + ,"#00beff" + ,"#00b9ff" + ,"#00b4ff"]; + var hash = hashStr(name); + var index = hash % colors.length; + return colors[index]; +} + +function hashStr(str) { + var hash = 0; + for (var i = 0; i < str.length; i++) { + var charCode = str.charCodeAt(i); + hash += charCode; + } + return hash; +} + +function createChartOverview(data) { + const ctx = document.getElementById('graph-health').getContext('2d'); + let high_data = data.stats.length > 1000; if (chart) { chart.destroy(); } + let datasets = [{ + label: 'Healthy', + data: data.healthyData, + borderColor: 'green', + backgroundColor: 'rgba(0, 255, 0, 0.2)', + fill: true + }, + { + label: 'Unhealthy', + data: data.unhealthyData, + borderColor: 'orange', + backgroundColor: 'rgba(241, 90, 34, 0.2)', + fill: true + }, + { + label: 'Own Instances Unhealthy', + data: data.user, + pointRadius: 5, + pointBackgroundColor: 'rgba(255, 0, 0, 1)', + fill: false, + datasetIndex: "x", + }]; + datasets = datasets.concat(data.stats.map(entry => ({ + yAxisID: 'yStats', + label: mapKey(entry.key), + data: entry.data, + borderColor: nameToColor(entry.key), + fill: false, + hidden: (entry.key !== 'total_requests_avg' && entry.key !== 'limited_accs_avg'), + }))); + console.log(datasets); chart = new Chart(ctx, { type: 'line', data: { - labels: timestamps, - datasets: [ - { - label: 'Healthy', - data: data.global.healthyData, - borderColor: 'green', - backgroundColor: 'rgba(0, 255, 0, 0.2)', - fill: true - }, - { - label: 'Unhealthy', - data: data.global.unhealthyData, - borderColor: 'orange', - backgroundColor: 'rgba(241, 90, 34, 0.2)', - fill: true - }, - { - label: 'Own Instances Unhealthy', - data: data.user, - pointRadius: 5, - pointBackgroundColor: 'rgba(255, 0, 0, 1)', - fill: false, - datasetIndex: "x", - } - ] + datasets: datasets, }, options: { + responsive: false, + maintainAspectRatio: false, plugins: { + colors: { + enabled: true + }, decimation: { enabled: high_data, algorithm: 'min-max', }, + tooltip: { + // callbacks: { + // label: function(context) { + // console.log(context); + // if (context.dataset) { + // if (context.dataset.yAxisID == 'yStats') { + // let dataset = context.dataset.data[context.dataIndex]; + // if (!dataset) { + // console.warn(context.dataset); + // console.warn(context.dataset.data[context.dataIndex]); + // } + // let mainValue = dataset.y; + // let detailValue = dataset.additional.total_requests_max; + // return [`${context.dataset.label}: ${mainValue}`,`Max Total Requests: ${detailValue}`]; + // } else { + // return `${context.dataset.label}: ${context.formattedValue}`; + // } + // } else { + // console.log(context); + // return null; + // } + // // let label = context.dataset.label || ''; + + // // if (label) { + // // label += ': '; + // // } + // // if (context.parsed.y !== null) { + // // label += new Intl.NumberFormat('en-US', { style: 'currency', currency: 'USD' }).format(context.parsed.y); + // // } + // return 'asdf'; + // } + // } + } }, animation: !high_data, scales: { x: { type: 'time', time: { - unit: 'day' // You might want to adjust this based on your data resolution + unit: 'day' }, autoSkip: true, }, y: { beginAtZero: true + }, + yStats: { + beginAtZero: false } - } + }, + interaction: { + mode: 'index', + intersect: false + }, } }); } \ No newline at end of file diff --git a/server/static/instance.js b/server/static/instance.js index d43cf82..9baa065 100644 --- a/server/static/instance.js +++ b/server/static/instance.js @@ -25,6 +25,24 @@ document.addEventListener("DOMContentLoaded", function() { fetchDataAndCreateChart(initialStartDate,initialEndDate); }); +function mapKey(key) { + const keyMappings = { + "limited_accs": "Limited Accs", + "total_accs": "Total Accs", + "total_requests": "Total Requests", + "req_photo_rail": "PhotoRail Requests", + "req_user_screen_name": "UserScreeName Req.", + "req_search": "Search Requests", + "req_list_tweets": "ListTeweets Req.", + "req_user_media": "UserMedia Req.", + "req_tweet_detail": "TweetDetail Req.", + "req_list": "List Requests", + "req_user_tweets": "UserTweets Req.", + "req_user_tweets_and_replies": "UserTweetsAndReplies Req.", + } + return keyMappings[key]; +} + async function fetchDataAndCreateChart(startDate,endDate) { let jsonData; try { @@ -41,15 +59,14 @@ async function fetchDataAndCreateChart(startDate,endDate) { console.error('Failed to fetch data:', error); } if (jsonData) { - createChart(jsonData); + createChartOverview(jsonData); } } -function createChart(jsonData) { - const timeLabels = jsonData.map(dataPoint => new Date(dataPoint.time * 1000)); - console.log(timeLabels); - const healthyData = jsonData.map(dataPoint => dataPoint.healthy ? (dataPoint.resp_time ? dataPoint.resp_time : 0) : null); - const unhealthyData = jsonData.map(dataPoint => dataPoint.healthy ? null : (dataPoint.resp_time ? dataPoint.resp_time : 0)); +function createChartOverview(jsonData) { + const healthyData = jsonData.health.filter(entry => entry.healthy).map(entry => ({x: moment.unix(entry.time), y: entry.resp_time})); + const unhealthyData = jsonData.health.filter(entry => !entry.healthy).map(entry => ({x: moment.unix(entry.time), y: entry.resp_time})); + const statsData = jsonData.stats.map(entry => ({x: moment.unix(entry.time), y: entry.total_requests})); const ctx = document.getElementById('graph').getContext('2d'); if (chart) { @@ -58,23 +75,33 @@ function createChart(jsonData) { chart = new Chart(ctx, { type: 'line', data: { - labels: timeLabels, datasets: [ { - label: 'Healthy', + label: 'Healthy Response Time (ms)', data: healthyData, borderColor: 'green', fill: false, }, { - label: 'Unhealthy', + label: 'Unhealthy Response Time (ms)', data: unhealthyData, borderColor: 'red', fill: false, }, + { + label: 'Total API Requests', + data: statsData, + borderColor: 'blue', + fill: false, + yAxisID: 'yStats', + }, ] }, options: { + interaction: { + mode: 'index', + intersect: false + }, scales: { x: { type: 'time', @@ -91,6 +118,12 @@ function createChart(jsonData) { display: true, labelString: 'Response Time' } + }, + yStats: { + scaleLabel: { + display: true, + labelString: 'Requests' + } } } } diff --git a/server/templates/admin.html.j2 b/server/templates/admin.html.j2 index 3a73c7b..377bdcb 100644 --- a/server/templates/admin.html.j2 +++ b/server/templates/admin.html.j2 @@ -16,7 +16,7 @@

Instances

{% if is_admin %}

root mode

{% endif %} -
+
diff --git a/src/main.rs b/src/main.rs index 6ddba2a..fbf64e3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -135,8 +135,12 @@ fn read_scanner_cfg() -> miette::Result { let error_retention_per_host: usize = require_env_str("ERROR_RETENTION_PER_HOST")? .parse() .expect("CLEANUP_INTERVAL_S must be a number"); + let instance_stats_interval: u64 = require_env_str("STATS_INTERVAL_S")? + .parse() + .expect("STATS_INTERVAL_S must be a positive number"); Ok(Arc::new(entities::state::scanner::Config { + instance_stats_interval: Duration::from_secs(instance_stats_interval), list_fetch_interval: Duration::from_secs(instance_list_interval), instance_check_interval: Duration::from_secs(instance_ping_interval), instance_list_url: nitter_instancelist,