From faa36137068b13285fb7712a717dcb6bb872d91f Mon Sep 17 00:00:00 2001 From: 4cuck <40539855+4cuck@users.noreply.github.com> Date: Sun, 30 Jul 2023 03:24:18 +0200 Subject: [PATCH] optimize database datetime as ints, strict and no rowid --- Cargo.lock | 1 + .../src/{update_check.rs => health_check.rs} | 4 +- entities/src/host.rs | 77 ++++++++++++- entities/src/mod.rs | 2 +- entities/src/prelude.rs | 2 +- migration/Cargo.toml | 2 +- migration/src/lib.rs | 8 +- .../src/m20220101_000001_create_table.rs | 1 + .../src/m20230729_010231_datetime_rowid.rs | 104 ++++++++++++++++++ .../src/m20230729_230909_datetime_int_host.rs | 102 +++++++++++++++++ scanner/src/cache_update.rs | 18 ++- scanner/src/instance_check.rs | 22 ++-- scanner/src/lib.rs | 12 +- src/main.rs | 2 +- 14 files changed, 321 insertions(+), 36 deletions(-) rename entities/src/{update_check.rs => health_check.rs} (92%) create mode 100644 migration/src/m20230729_010231_datetime_rowid.rs create mode 100644 migration/src/m20230729_230909_datetime_int_host.rs diff --git a/Cargo.lock b/Cargo.lock index 47e7fff..de45dae 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1794,6 +1794,7 @@ version = "0.1.0" dependencies = [ "async-std", "sea-orm-migration", + "tracing", ] [[package]] diff --git a/entities/src/update_check.rs b/entities/src/health_check.rs similarity index 92% rename from entities/src/update_check.rs rename to entities/src/health_check.rs index 4a3afe7..6a63f3c 100644 --- a/entities/src/update_check.rs +++ b/entities/src/health_check.rs @@ -4,10 +4,10 @@ use sea_orm::entity::prelude::*; use serde::Serialize; #[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize)] -#[sea_orm(table_name = "update_check")] +#[sea_orm(table_name = "health_check")] pub struct Model { #[sea_orm(primary_key, auto_increment = false)] - pub time: DateTimeUtc, + pub time: i64, #[sea_orm(primary_key, auto_increment = false)] pub host: i32, pub resp_time: Option, diff --git a/entities/src/host.rs b/entities/src/host.rs index 0b06e19..10ba640 100644 --- a/entities/src/host.rs +++ b/entities/src/host.rs @@ -1,9 +1,18 @@ //! `SeaORM` Entity. Generated by sea-orm-codegen 0.11.3 -use sea_orm::entity::prelude::*; +use sea_orm::{entity::prelude::*, sea_query::{Expr, SimpleExpr}}; use serde::Serialize; -#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize)] +#[derive(Copy, Clone, Default, Debug, DeriveEntity)] +pub struct Entity; + +impl EntityName for Entity { + fn table_name(&self) -> &str { + "host" + } +} + +#[derive(Clone, Debug, PartialEq, DeriveModel, DeriveActiveModel, Eq, Serialize)] #[sea_orm(table_name = "host")] pub struct Model { #[sea_orm(primary_key)] @@ -14,16 +23,72 @@ pub struct Model { pub rss: bool, pub version: Option, /// Last time the url and enabled were updated, *not* the rss - pub updated: DateTimeUtc, + pub updated: i64, } -#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +#[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)] +pub enum Column { + Id, + Domain, + Url, + Version, + Enabled, + Rss, + Updated, +} + +#[derive(Copy, Clone, Debug, EnumIter, DerivePrimaryKey)] +pub enum PrimaryKey { + Id, +} + +impl PrimaryKeyTrait for PrimaryKey { + type ValueType = i32; + fn auto_increment() -> bool { + true + } +} + +#[derive(Copy, Clone, Debug, EnumIter)] pub enum Relation { - #[sea_orm(has_many = "super::update_check::Entity")] UpdateCheck, } -impl Related for Entity { +impl ColumnTrait for Column { + type EntityName = Entity; + fn def(&self) -> ColumnDef { + match self { + Self::Id => ColumnType::Integer.def(), + Self::Domain => ColumnType::String(None).def(), + Self::Url => ColumnType::String(None).def(), + Self::Version => ColumnType::String(None).def().null(), + Self::Enabled => ColumnType::Integer.def(), + Self::Rss => ColumnType::Integer.def(), + Self::Updated => ColumnType::Integer.def(), + } + } + + // fn select_as(&self, expr: Expr) -> SimpleExpr { + // Column::CaseInsensitiveText => expr.cast_as(Alias::new("text")), + // _ => self.select_enum_as(expr), + // } + + // /// Cast value of a column into the correct type for database storage. + // fn save_as(&self, val: Expr) -> SimpleExpr { + // Column::CaseInsensitiveText => val.cast_as(Alias::new("citext")), + // _ => self.save_enum_as(val), + // } +} + +impl RelationTrait for Relation { + fn def(&self) -> RelationDef { + match self { + Self::UpdateCheck => Entity::has_many(super::health_check::Entity).into(), + } + } +} + +impl Related for Entity { fn to() -> RelationDef { Relation::UpdateCheck.def() } diff --git a/entities/src/mod.rs b/entities/src/mod.rs index b7ef3c7..2630023 100644 --- a/entities/src/mod.rs +++ b/entities/src/mod.rs @@ -3,7 +3,7 @@ pub mod prelude; pub mod host; -pub mod update_check; +pub mod health_check; // has to be re-added pub mod state; diff --git a/entities/src/prelude.rs b/entities/src/prelude.rs index 776ed20..5e2a4c7 100644 --- a/entities/src/prelude.rs +++ b/entities/src/prelude.rs @@ -1,4 +1,4 @@ //! `SeaORM` Entity. Generated by sea-orm-codegen 0.11.3 pub use super::host::Entity as Host; -pub use super::update_check::Entity as UpdateCheck; +pub use super::health_check::Entity as UpdateCheck; diff --git a/migration/Cargo.toml b/migration/Cargo.toml index 85babbf..02596f0 100644 --- a/migration/Cargo.toml +++ b/migration/Cargo.toml @@ -10,7 +10,7 @@ path = "src/lib.rs" [dependencies] async-std = { version = "^1", features = ["attributes", "tokio1"] } - +tracing = { workspace = true } [dependencies.sea-orm-migration] workspace = true features = [ diff --git a/migration/src/lib.rs b/migration/src/lib.rs index 2c605af..cf7f6ea 100644 --- a/migration/src/lib.rs +++ b/migration/src/lib.rs @@ -1,12 +1,18 @@ pub use sea_orm_migration::prelude::*; mod m20220101_000001_create_table; +mod m20230729_010231_datetime_rowid; +mod m20230729_230909_datetime_int_host; pub struct Migrator; #[async_trait::async_trait] impl MigratorTrait for Migrator { fn migrations() -> Vec> { - vec![Box::new(m20220101_000001_create_table::Migration)] + vec![ + Box::new(m20220101_000001_create_table::Migration), + Box::new(m20230729_010231_datetime_rowid::Migration), + Box::new(m20230729_230909_datetime_int_host::Migration), + ] } } diff --git a/migration/src/m20220101_000001_create_table.rs b/migration/src/m20220101_000001_create_table.rs index 8d36865..002db2a 100644 --- a/migration/src/m20220101_000001_create_table.rs +++ b/migration/src/m20220101_000001_create_table.rs @@ -1,3 +1,4 @@ +// SPDX-License-Identifier: AGPL-3.0-only use sea_orm_migration::prelude::*; #[derive(DeriveMigrationName)] diff --git a/migration/src/m20230729_010231_datetime_rowid.rs b/migration/src/m20230729_010231_datetime_rowid.rs new file mode 100644 index 0000000..196fcd8 --- /dev/null +++ b/migration/src/m20230729_010231_datetime_rowid.rs @@ -0,0 +1,104 @@ +use std::borrow::Cow; + +use sea_orm_migration::{ + prelude::*, + sea_orm::{ + prelude::DateTimeUtc, DbBackend, DeriveEntityModel, FromQueryResult, RuntimeErr, Statement, + }, +}; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[derive(Debug, FromQueryResult)] +pub struct UpdateCheck { + time: DateTimeUtc, + host: i32, + resp_time: Option, + healthy: bool, + response_code: Option, +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let cmd = r#"CREATE TABLE "health_check_new" ( + "time" integer NOT NULL, + "host" integer NOT NULL, + "resp_time" integer, + "healthy" integer NOT NULL, + "response_code" integer, + CONSTRAINT "pk_health_check_new" 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?; + db.execute_unprepared(cmd).await?; + tracing::info!("fetching all healtcheck entries.."); + let data = UpdateCheck::find_by_statement(Statement::from_sql_and_values( + DbBackend::Sqlite, + r#"SELECT * FROM update_check"#, + [], + )) + .all(db) + .await?; + + tracing::info!("migrating {} entries to temp table..", data.len()); + let mut duplicates = 0; + for entry in data.into_iter() { + if let Err(err) = db + .execute(Statement::from_sql_and_values( + DbBackend::Sqlite, + r#"INSERT INTO health_check_new ( + time, + host, + resp_time, + healthy, + response_code) VALUES ($1,$2,$3,$4,$5)"#, + [ + entry.time.timestamp().into(), + entry.host.into(), + entry.resp_time.into(), + entry.healthy.into(), + entry.response_code.into(), + ], + )) + .await + { + if let DbErr::Exec(RuntimeErr::SqlxError(e)) = &err { + if let Some(err) = e.as_database_error() { + if err.code() == Some(Cow::Borrowed("1555")) { + duplicates += 1; + continue; + } + } + } + tracing::info!(entry=?entry); + return Err(err); + } + } + tracing::info!("dropped {} duplicate timestamps..", duplicates); + tracing::info!("dropping old table.."); + db.execute_unprepared("DROP TABLE update_check").await?; + + db.execute_unprepared(&cmd.replace("health_check_new", "health_check")) + .await?; + tracing::info!("inserting back into final table.."); + db.execute_unprepared( + r#"INSERT INTO health_check + SELECT * + FROM health_check_new WHERE true"#, + ) + .await?; + tracing::info!("dropping temp table.."); + db.execute_unprepared("DROP TABLE health_check_new").await?; + tracing::info!("cleaning up db.."); + 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/migration/src/m20230729_230909_datetime_int_host.rs b/migration/src/m20230729_230909_datetime_int_host.rs new file mode 100644 index 0000000..0b4a143 --- /dev/null +++ b/migration/src/m20230729_230909_datetime_int_host.rs @@ -0,0 +1,102 @@ +use sea_orm_migration::{ + prelude::*, + sea_orm::{prelude::DateTimeUtc, DbBackend, FromQueryResult, Statement}, +}; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[derive(Debug, FromQueryResult)] +pub struct Host { + pub id: i32, + pub domain: String, + pub url: String, + pub enabled: bool, + pub rss: bool, + pub version: Option, + pub updated: DateTimeUtc, +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let cmd = r#"CREATE TABLE "host_new" ( + "id" integer NOT NULL PRIMARY KEY AUTOINCREMENT, + "domain" text NOT NULL UNIQUE, + "url" text NOT NULL, + "version" text, + "enabled" integer NOT NULL, + "rss" integer NOT NULL, + "updated" integer NOT NULL + ) STRICT;"#; + let db = manager.get_connection(); + db.execute_unprepared("PRAGMA foreign_keys = 0").await?; + db.execute_unprepared("BEGIN EXCLUSIVE").await?; + db.execute_unprepared(cmd).await?; + tracing::info!("fetching all host entries.."); + let data = Host::find_by_statement(Statement::from_sql_and_values( + DbBackend::Sqlite, + r#"SELECT * FROM host"#, + [], + )) + .all(db) + .await?; + + tracing::info!("migrating {} entries to temp table..", data.len()); + for entry in data.into_iter() { + db.execute(Statement::from_sql_and_values( + DbBackend::Sqlite, + r#"INSERT INTO host_new ( + id, + domain, + url, + version, + enabled, + rss, + updated ) VALUES ($1,$2,$3,$4,$5,$6,$7)"#, + [ + entry.id.into(), + entry.domain.into(), + entry.url.into(), + entry.version.into(), + entry.enabled.into(), + entry.rss.into(), + entry.updated.timestamp().into(), + ], + )) + .await?; + } + tracing::info!("dropping old table.."); + db.execute_unprepared("DROP TABLE host").await?; + + db.execute_unprepared(&cmd.replace("host_new", "host")) + .await?; + tracing::info!("inserting back into final table.."); + db.execute_unprepared( + r#"INSERT INTO host + SELECT * + FROM host_new WHERE true"#, + ) + .await?; + tracing::info!("dropping temp table.."); + db.execute_unprepared("DROP TABLE host_new").await?; + tracing::info!("cleaning up db.."); + db.execute_unprepared("COMMIT TRANSACTION").await?; + db.execute_unprepared("PRAGMA foreign_keys = 1").await?; + db.execute_unprepared("VACUUM").await?; + Ok(()) + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + panic!("Can't migrate down"); + } +} + +/// Learn more at https://docs.rs/sea-query#iden +#[derive(Iden)] +enum Post { + Table, + Id, + Title, + Text, +} diff --git a/scanner/src/cache_update.rs b/scanner/src/cache_update.rs index e9ce26a..b0312df 100644 --- a/scanner/src/cache_update.rs +++ b/scanner/src/cache_update.rs @@ -139,12 +139,12 @@ impl Scanner { let last_pings = PingEntry::find_by_statement(Statement::from_sql_and_values( DbBackend::Sqlite, r#" - SELECT u.host,(CASE u.healthy WHEN true THEN u.resp_time ELSE null END) as ping FROM update_check u + SELECT u.host,(CASE u.healthy WHEN true THEN u.resp_time ELSE null END) as ping FROM health_check u JOIN host h ON h.id = u.host WHERE h.enabled = true AND u.time >= $1 ORDER BY u.host,u.time ASC "#, - [age.into()], + [age.timestamp().into()], )) .all(&self.inner.db) .await?; @@ -157,6 +157,12 @@ impl Scanner { let mut current_entry = LastPings::default(); let mut last_host = first.host; let mut non_null_entries = first.ping.as_ref().map_or(0, |_|1); + if let Some(ping) = first.ping.as_ref() { + current_entry.avg = Some(*ping); + non_null_entries += 1; + current_entry.min = Some(*ping); + current_entry.max = Some(*ping); + } current_entry.pings.push(first.ping); for ping in iter { if last_host != ping.host { @@ -192,11 +198,11 @@ impl Scanner { let stats = Version::find_by_statement(Statement::from_sql_and_values( DbBackend::Sqlite, r#"SELECT version FROM host h - JOIN update_check u ON u.host = h.id + JOIN health_check u ON u.host = h.id WHERE h.enabled = true AND u.time >= $1 AND version IS NOT NULL GROUP BY version ORDER BY version ASC"#, - [age.into()], + [age.timestamp().into()], )) .all(&self.inner.db) .await?; @@ -219,11 +225,11 @@ impl Scanner { ) -> Result> { let stats: Vec = HostStats::find_by_statement(Statement::from_sql_and_values( DbBackend::Sqlite, - r#"SELECT u.host, COUNT(CASE WHEN healthy = true THEN 1 END) as good,COUNT(*) as total FROM update_check u + r#"SELECT u.host, COUNT(CASE WHEN healthy = true THEN 1 END) as good,COUNT(*) as total FROM health_check u JOIN host h ON h.id = u.host WHERE h.enabled = true AND u.time BETWEEN $1 AND $2 GROUP BY u.host "#, - [from.into(), to.into()], + [from.timestamp().into(), to.timestamp().into()], )) .all(&self.inner.db) .await?; diff --git a/scanner/src/instance_check.rs b/scanner/src/instance_check.rs index 1bd8c0a..72c7627 100644 --- a/scanner/src/instance_check.rs +++ b/scanner/src/instance_check.rs @@ -3,7 +3,7 @@ use std::time::Instant; use chrono::Utc; -use entities::update_check; +use entities::health_check; use entities::{host, prelude::*}; use reqwest::Url; use sea_orm::prelude::DateTimeUtc; @@ -35,7 +35,7 @@ impl Scanner { 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, muted_host).await; + scanner.health_check_host(model, muted_host).await; }); } // wait till all of them are finished, preventing DoS @@ -49,14 +49,14 @@ impl Scanner { } #[instrument] - async fn update_check_host(&self, host: host::Model, muted: bool) { + async fn health_check_host(&self, host: host::Model, muted: bool) { let now = Utc::now(); let mut url = match Url::parse(&host.url) { Err(e) => { if !muted { tracing::error!(error=?e, url=host.url,"failed to parse instance URL"); } - self.insert_failed_update_check(host.id, now, None, None) + self.insert_failed_health_check(host.id, now, None, None) .await; return; } @@ -82,7 +82,7 @@ impl Scanner { } _ => (None, None), }; - self.insert_failed_update_check(host.id, now, resp_time, http_code) + self.insert_failed_health_check(host.id, now, resp_time, http_code) .await; } Ok((http_code, content)) => { @@ -97,7 +97,7 @@ impl Scanner { "host doesn't contain expected profile content" ); } - self.insert_failed_update_check( + self.insert_failed_health_check( host.id, now, Some(took_ms as _), @@ -107,8 +107,8 @@ impl Scanner { } // create successfull uptime entry - if let Err(e) = (update_check::ActiveModel { - time: ActiveValue::Set(now), + if let Err(e) = (health_check::ActiveModel { + time: ActiveValue::Set(now.timestamp()), host: ActiveValue::Set(host.id), resp_time: ActiveValue::Set(Some(took_ms as _)), response_code: ActiveValue::Set(Some(http_code as _)), @@ -168,15 +168,15 @@ impl Scanner { } } - async fn insert_failed_update_check( + async fn insert_failed_health_check( &self, host: i32, time: DateTimeUtc, resp_time: Option, http_code: Option, ) { - if let Err(e) = (update_check::ActiveModel { - time: ActiveValue::Set(time), + if let Err(e) = (health_check::ActiveModel { + time: ActiveValue::Set(time.timestamp()), host: ActiveValue::Set(host), resp_time: ActiveValue::Set(resp_time), healthy: ActiveValue::Set(false), diff --git a/scanner/src/lib.rs b/scanner/src/lib.rs index 55d6f1a..a531380 100644 --- a/scanner/src/lib.rs +++ b/scanner/src/lib.rs @@ -240,7 +240,7 @@ impl Scanner { host::ActiveModel { id: ActiveValue::Set(host.id), enabled: ActiveValue::Set(false), - updated: ActiveValue::Set(time), + updated: ActiveValue::Set(time.timestamp()), ..Default::default() } .update(&transaction) @@ -283,7 +283,7 @@ impl Scanner { enabled: ActiveValue::Set(true), version: ActiveValue::Set(version), rss: ActiveValue::Set(rss), - updated: ActiveValue::Set(time), + updated: ActiveValue::Set(time.timestamp()), } }); } @@ -367,14 +367,14 @@ impl Scanner { } pub async fn query_latest_check(&self, connection: &T) -> Result> { - let update_checks = LatestCheck::find_by_statement(Statement::from_sql_and_values( + let health_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 + SELECT u.host,MAX(u.time) as time FROM health_check u GROUP BY u.host ) - SELECT u.host,healthy,h.domain FROM update_check u + SELECT u.host,healthy,h.domain FROM health_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 @@ -383,7 +383,7 @@ impl Scanner { )) .all(connection) .await?; - Ok(update_checks) + Ok(health_checks) } } diff --git a/src/main.rs b/src/main.rs index e3e36d8..151a4bf 100644 --- a/src/main.rs +++ b/src/main.rs @@ -29,7 +29,7 @@ async fn _main() -> miette::Result<()> { .with(tracing_subscriber::EnvFilter::new( var("RUST_LOG").unwrap_or_else(|_| { format!( - "warn,tower_http=debug,scanner=trace,server=debug,{}=debug", + "warn,tower_http=debug,migration=debug,scanner=trace,server=debug,{}=debug", env!("CARGO_PKG_NAME") ) .into()