diff --git a/Cargo.lock b/Cargo.lock index 81067d44bb..9516faba69 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -640,6 +640,33 @@ dependencies = [ "syn", ] +[[package]] +name = "analytics-job" +version = "0.0.1" +dependencies = [ + "anyhow", + "chrono", + "clap 2.34.0", + "config", + "container_common", + "database_queries", + "dotenv", + "easyenv", + "enums", + "errors", + "hostname", + "jobs_common", + "log", + "rand 0.8.4", + "reusable_types", + "serde", + "serde_derive", + "serde_json", + "sqlx", + "tokens", + "tokio 1.16.1", +] + [[package]] name = "ansi_term" version = "0.12.1" diff --git a/Cargo.toml b/Cargo.toml index caa8deb96c..5512d02a5f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,6 +39,7 @@ members = [ "crates/lib/user_input_common", # Services: API servers, jobs, workers, etc. + "crates/service/job/analytics_job", "crates/service/job/download_job", "crates/service/job/inference_job", "crates/service/job/tts_download_job", diff --git a/Dockerfile b/Dockerfile index 709e9f751f..ffc07ef485 100644 --- a/Dockerfile +++ b/Dockerfile @@ -78,6 +78,12 @@ RUN SQLX_OFFLINE=true \ --release \ --bin storyteller-web +RUN SQLX_OFFLINE=true \ + LD_LIBRARY_PATH=/usr/lib:${LD_LIBRARY_PATH} \ + $HOME/.cargo/bin/cargo build \ + --release \ + --bin analytics-job \ + RUN SQLX_OFFLINE=true \ LD_LIBRARY_PATH=/usr/lib:${LD_LIBRARY_PATH} \ $HOME/.cargo/bin/cargo build \ @@ -149,6 +155,7 @@ RUN echo -n ${GIT_SHA} > GIT_SHA # Copy all the binaries. COPY --from=builder /tmp/target/release/storyteller-web / +COPY --from=builder /tmp/target/release/analytics-job / COPY --from=builder /tmp/target/release/download-job / COPY --from=builder /tmp/target/release/inference-job / COPY --from=builder /tmp/target/release/tts-download-job / diff --git a/_migrations/2023-01-16-211642_trending_model_analytics/down.sql b/_migrations/2023-01-16-211642_trending_model_analytics/down.sql new file mode 100644 index 0000000000..04d03ee442 --- /dev/null +++ b/_migrations/2023-01-16-211642_trending_model_analytics/down.sql @@ -0,0 +1,5 @@ +-- noinspection SqlDialectInspectionForFile +-- noinspection SqlNoDataSourceInspectionForFile +-- noinspection SqlResolveForFile + +DROP TABLE trending_model_analytics; diff --git a/_migrations/2023-01-16-211642_trending_model_analytics/up.sql b/_migrations/2023-01-16-211642_trending_model_analytics/up.sql new file mode 100644 index 0000000000..1b8a58726f --- /dev/null +++ b/_migrations/2023-01-16-211642_trending_model_analytics/up.sql @@ -0,0 +1,40 @@ +-- noinspection SqlDialectInspectionForFile +-- noinspection SqlNoDataSourceInspectionForFile +-- noinspection SqlResolveForFile + +-- Periodically capture analytics on trending models +CREATE TABLE trending_model_analytics ( + -- Not used for anything except replication. + id BIGINT(20) NOT NULL AUTO_INCREMENT, + + -- The type of model. TTS, VC, etc. + model_type VARCHAR(32) NOT NULL, + + -- The primary key of the TTS, VC, or other model in question. + model_token VARCHAR(32) NOT NULL, + + -- The name of the window, eg. "last_24_hours", "last_5_hours", etc. + window_name VARCHAR(32) NOT NULL, + + -- The measured numerical value of the statistic, eg "500 uses" = 500. + numeric_value INT(10) NOT NULL DEFAULT 0, + + -- ========== VECTOR CLOCK ========== + + -- Incremented with every update. + version INT NOT NULL DEFAULT 0, + + -- ========== TIMESTAMPS ========== + + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + + -- INDICES -- + PRIMARY KEY (id), + UNIQUE KEY (model_token, model_type, window_name), + KEY index_model_type_model_token (model_type, model_token), + KEY index_model_type (model_type), + KEY index_model_token (model_token), + KEY index_numeric_value (numeric_value) + +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; diff --git a/analytics/tts_models.sql b/analytics/tts_models.sql index ae77c0f035..5acbb689ce 100644 --- a/analytics/tts_models.sql +++ b/analytics/tts_models.sql @@ -7,8 +7,9 @@ -- select m.token, - m.title, + m.text_pipeline_type, m.ietf_language_tag, + m.title, u.username, r.use_count, m.user_deleted_at, @@ -37,25 +38,18 @@ select m.token, m.title, r.use_count from ( join tts_models as m on m.token = r.model_token; --- --- Top 100 models by use count, last 30 days --- -select m.token, m.title, m.ietf_language_tag, r.use_count from ( - select model_token, count(*) as use_count - from tts_results - where created_at > ( CURDATE() - INTERVAL 30 DAY ) - group by model_token - order by use_count desc - limit 500 -) as r - join tts_models as m - on m.token = r.model_token; - -- -- Top 500 models by use count, last 5 days -- Limited due to tmux scrollback. -- -select m.token, m.title, m.ietf_language_tag, u.username, r.use_count from ( +select + m.token, + m.text_pipeline_type, + m.ietf_language_tag, + m.title, + u.username, + r.use_count +from ( select model_token, count(*) as use_count from tts_results where created_at > ( CURDATE() - INTERVAL 5 DAY ) @@ -74,8 +68,9 @@ order by r.use_count desc -- select m.token, - m.title, + m.text_pipeline_type, m.ietf_language_tag, + m.title, u.username, r.use_count, m.user_deleted_at, diff --git a/crates/_types/enums/src/by_table/mod.rs b/crates/_types/enums/src/by_table/mod.rs index b04ada0161..4ee11bc1ac 100644 --- a/crates/_types/enums/src/by_table/mod.rs +++ b/crates/_types/enums/src/by_table/mod.rs @@ -1,2 +1,3 @@ pub mod media_uploads; pub mod model_categories; +pub mod trending_model_analytics; diff --git a/crates/_types/enums/src/by_table/trending_model_analytics/mod.rs b/crates/_types/enums/src/by_table/trending_model_analytics/mod.rs new file mode 100644 index 0000000000..fcb0d46e1b --- /dev/null +++ b/crates/_types/enums/src/by_table/trending_model_analytics/mod.rs @@ -0,0 +1,2 @@ +pub mod model_type; +pub mod window_name; \ No newline at end of file diff --git a/crates/_types/enums/src/by_table/trending_model_analytics/model_type.rs b/crates/_types/enums/src/by_table/trending_model_analytics/model_type.rs new file mode 100644 index 0000000000..4489cdb05f --- /dev/null +++ b/crates/_types/enums/src/by_table/trending_model_analytics/model_type.rs @@ -0,0 +1,63 @@ +use serde::Deserialize; +use serde::Serialize; + +#[cfg(test)] +use strum::EnumCount; +#[cfg(test)] +use strum::EnumIter; + +/// Used in the `trending_model_analytics` table in a `VARCHAR` field. +/// +/// DO NOT CHANGE VALUES WITHOUT A MIGRATION STRATEGY. +#[cfg_attr(test, derive(EnumIter, EnumCount))] +#[derive(Clone, Copy, Eq, PartialEq, Deserialize, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum ModelType { + /// TTS models + Tts, + + // NB: It's assumed we'll use this same system to track + // trending VC models too, so the next type would be "VC". +} + +// TODO(bt, 2023-01-17): This desperately needs MySQL integration tests! +impl_enum_display_and_debug_using_to_str!(ModelType); +impl_enum_sqlx_coders!(ModelType); + +/// NB: Legacy API for older code. +impl ModelType { + pub fn to_str(&self) -> &'static str { + match self { + Self::Tts => "tts", + } + } + + pub fn from_str(value: &str) -> Result { + match value { + "tts" => Ok(Self::Tts), + _ => Err(format!("invalid value: {:?}", value)), + } + } +} + +#[cfg(test)] +mod tests { + use crate::by_table::trending_model_analytics::model_type::ModelType; + use crate::test_helpers::assert_serialization; + + #[test] + fn test_serialization() { + assert_serialization(ModelType::Tts, "tts"); + } + + #[test] + fn test_to_str() { + assert_eq!(ModelType::Tts.to_str(), "tts"); + } + + #[test] + fn test_from_str() { + assert_eq!(ModelType::from_str("tts").unwrap(), ModelType::Tts); + assert!(ModelType::from_str("foo").is_err()); + } +} diff --git a/crates/_types/enums/src/by_table/trending_model_analytics/window_name.rs b/crates/_types/enums/src/by_table/trending_model_analytics/window_name.rs new file mode 100644 index 0000000000..5fa3b382d7 --- /dev/null +++ b/crates/_types/enums/src/by_table/trending_model_analytics/window_name.rs @@ -0,0 +1,88 @@ +use serde::Deserialize; +use serde::Serialize; + +#[cfg(test)] +use strum::EnumCount; +#[cfg(test)] +use strum::EnumIter; + +/// Used in the `window_name` table in a `VARCHAR` field. +/// +/// Contrary to most of this crate and unlike most "enum"-types +/// that are inflexible, new window names can be added/removed +/// without breaking too much. +#[cfg_attr(test, derive(EnumIter, EnumCount))] +#[derive(Clone, Copy, Eq, PartialEq, Hash, Deserialize, Serialize)] +pub enum WindowName { + /// Analytics over the last three hours + #[serde(rename = "last_3_hours")] + Last3Hours, + + /// Analytics over the last three hours + #[serde(rename = "last_3_days")] + Last3Days, + + /// Analytics over all historical records + #[serde(rename = "all_time")] + AllTime, +} + +// TODO(bt, 2023-01-17): This desperately needs MySQL integration tests! +impl_enum_display_and_debug_using_to_str!(WindowName); +impl_enum_sqlx_coders!(WindowName); + +/// NB: Legacy API for older code. +impl WindowName { + pub fn to_str(&self) -> &'static str { + match self { + Self::Last3Hours => "last_3_hours", + Self::Last3Days => "last_3_days", + Self::AllTime => "all_time", + } + } + + pub fn from_str(value: &str) -> Result { + match value { + "last_3_hours" => Ok(Self::Last3Hours), + "last_3_days" => Ok(Self::Last3Days), + "all_time" => Ok(Self::AllTime), + _ => Err(format!("invalid value: {:?}", value)), + } + } +} + +#[cfg(test)] +mod tests { + use crate::by_table::trending_model_analytics::window_name::WindowName; + use crate::test_helpers::assert_serialization; + + mod serde { + use super::*; + + #[test] + fn test_serialization() { + assert_serialization(WindowName::Last3Hours, "last_3_hours"); + assert_serialization(WindowName::Last3Days, "last_3_days"); + assert_serialization(WindowName::AllTime, "all_time"); + } + } + + mod impl_methods { + use super::*; + + #[test] + fn test_to_str() { + assert_eq!(WindowName::Last3Hours.to_str(), "last_3_hours"); + assert_eq!(WindowName::Last3Days.to_str(), "last_3_days"); + assert_eq!(WindowName::AllTime.to_str(), "all_time"); + } + + #[test] + fn test_from_str() { + assert_eq!(WindowName::from_str("last_3_hours").unwrap(), WindowName::Last3Hours); + assert_eq!(WindowName::from_str("last_3_days").unwrap(), WindowName::Last3Days); + assert_eq!(WindowName::from_str("all_time").unwrap(), WindowName::AllTime); + assert!(WindowName::from_str("foo").is_err()); + } + } +} diff --git a/crates/lib/database_queries/src/queries/mod.rs b/crates/lib/database_queries/src/queries/mod.rs index b15d0be795..dc18feaa49 100644 --- a/crates/lib/database_queries/src/queries/mod.rs +++ b/crates/lib/database_queries/src/queries/mod.rs @@ -7,6 +7,7 @@ pub mod ip_bans; pub mod media_uploads; pub mod model_categories; pub mod stats; +pub mod trending_model_analytics; pub mod tts; pub mod twitch; pub mod users; diff --git a/crates/lib/database_queries/src/queries/trending_model_analytics/list_trending_tts_models.rs b/crates/lib/database_queries/src/queries/trending_model_analytics/list_trending_tts_models.rs new file mode 100644 index 0000000000..5c25a010c6 --- /dev/null +++ b/crates/lib/database_queries/src/queries/trending_model_analytics/list_trending_tts_models.rs @@ -0,0 +1,117 @@ +use sqlx::{MySql, MySqlPool}; +use sqlx::pool::PoolConnection; +use enums::by_table::trending_model_analytics::model_type::ModelType; +use enums::by_table::trending_model_analytics::window_name::WindowName; +use errors::AnyhowResult; +use tokens::tokens::tts_models::TtsModelToken; + +pub struct TrendingModels { + pub models: Vec, +} + +pub struct TrendingModel { + pub model_token: TtsModelToken, + + // NB: "en", "es", etc. not "en-US", "es-419", etc. + pub ietf_primary_language_subtag: String, + + // Time window + pub window_name: WindowName, + + // Count of uses within the window. + pub numeric_value: u64, +} + +// TODO(bt,2023-01-17): This query is massive, so is not type checked :( + +pub async fn list_trending_tts_models_with_pool( + mysql_pool: &MySqlPool, +) -> AnyhowResult { + let mut mysql_connection = mysql_pool.acquire().await?; + list_trending_tts_models(&mut mysql_connection).await +} + +pub async fn list_trending_tts_models( + mysql_connection: &mut PoolConnection, +) -> AnyhowResult { + let query_parts = vec![ + // English + make_subquery(ModelType::Tts, WindowName::AllTime, "en"), + make_subquery(ModelType::Tts, WindowName::Last3Hours, "en"), + make_subquery(ModelType::Tts, WindowName::Last3Days, "en"), + + // Spanish + make_subquery(ModelType::Tts, WindowName::AllTime, "es"), + make_subquery(ModelType::Tts, WindowName::Last3Hours, "es"), + make_subquery(ModelType::Tts, WindowName::Last3Days, "es"), + + // Italian + make_subquery(ModelType::Tts, WindowName::AllTime, "it"), + make_subquery(ModelType::Tts, WindowName::Last3Hours, "it"), + make_subquery(ModelType::Tts, WindowName::Last3Days, "it"), + + // French + make_subquery(ModelType::Tts, WindowName::AllTime, "fr"), + make_subquery(ModelType::Tts, WindowName::Last3Hours, "fr"), + make_subquery(ModelType::Tts, WindowName::Last3Days, "fr"), + + // German + make_subquery(ModelType::Tts, WindowName::AllTime, "de"), + make_subquery(ModelType::Tts, WindowName::Last3Hours, "de"), + make_subquery(ModelType::Tts, WindowName::Last3Days, "de"), + ]; + + let query = query_parts.join(" UNION "); + let mut query = sqlx::query_as::<_, RawTrendingModel>(&query); + + let results = query.fetch_all(mysql_connection).await?; + + let results = results.into_iter() + .map(|model| TrendingModel { + model_token: TtsModelToken::new(model.model_token), + ietf_primary_language_subtag: model.ietf_primary_language_subtag, + window_name: WindowName::from_str(&model.window_name).unwrap_or(WindowName::AllTime), // TODO: Fail properly + numeric_value: model.numeric_value as u64, + }) + .collect::>(); + + Ok(TrendingModels { + models: results, + }) +} + +// Build queries meant to be UNION-d together. +pub fn make_subquery(model_type: ModelType, window_name: WindowName, ietf_primary_language_subtag: &str) -> String { + format!(r#" +( + SELECT + m.token as model_token, + m.ietf_primary_language_subtag, + t.window_name, + t.numeric_value + + FROM trending_model_analytics AS t + JOIN tts_models AS m + ON m.token = t.model_token + + WHERE + t.model_type = "{}" + AND t.window_name = "{}" + AND m.ietf_primary_language_subtag = "{}" + + ORDER BY numeric_value DESC + LIMIT 10 +) + "#, + model_type, + window_name, + ietf_primary_language_subtag) +} + +#[derive(sqlx::FromRow)] +struct RawTrendingModel { + pub model_token: String, + pub ietf_primary_language_subtag: String, + pub window_name: String, + pub numeric_value: i64, +} \ No newline at end of file diff --git a/crates/lib/database_queries/src/queries/trending_model_analytics/mod.rs b/crates/lib/database_queries/src/queries/trending_model_analytics/mod.rs new file mode 100644 index 0000000000..4f3e9ff090 --- /dev/null +++ b/crates/lib/database_queries/src/queries/trending_model_analytics/mod.rs @@ -0,0 +1,2 @@ +pub mod list_trending_tts_models; +pub mod upsert_trending_model_analytics; \ No newline at end of file diff --git a/crates/lib/database_queries/src/queries/trending_model_analytics/upsert_trending_model_analytics.rs b/crates/lib/database_queries/src/queries/trending_model_analytics/upsert_trending_model_analytics.rs new file mode 100644 index 0000000000..3775af39c0 --- /dev/null +++ b/crates/lib/database_queries/src/queries/trending_model_analytics/upsert_trending_model_analytics.rs @@ -0,0 +1,50 @@ +use container_common::anyhow_result::AnyhowResult; +use enums::by_table::trending_model_analytics::model_type::ModelType; +use enums::by_table::trending_model_analytics::window_name::WindowName; +use sqlx::MySql; +use sqlx::pool::PoolConnection; +use tokens::tokens::tts_models::TtsModelToken; + +// NB: Only TTS for now. New enum variants for other token types. +pub enum ModelToken<'a> { + Tts(&'a TtsModelToken), +} + +pub struct Args<'a> { + pub model_token: ModelToken<'a>, + pub window_name: WindowName, + pub numeric_value: u64, + pub mysql_connection: &'a mut PoolConnection, +} + +pub async fn upsert_trending_model_analytics(args: Args<'_>) -> AnyhowResult<()> { + + let (model_type, model_token) = match args.model_token { + ModelToken::Tts(ref token) => (ModelType::Tts, token.as_str()), + }; + + let query = sqlx::query!( + r#" +INSERT INTO trending_model_analytics +SET + model_type = ?, + model_token = ?, + window_name = ?, + numeric_value = ?, + version = 1 + +ON DUPLICATE KEY UPDATE + numeric_value = ?, + version = version + 1 + "#, + // Insert + model_type, + model_token, + args.window_name, + args.numeric_value, + args.numeric_value, + ); + + let _r = query.execute(args.mysql_connection).await?; + Ok(()) +} diff --git a/crates/lib/database_queries/src/queries/tts/tts_models/count_tts_model_uses.rs b/crates/lib/database_queries/src/queries/tts/tts_models/count_tts_model_uses.rs new file mode 100644 index 0000000000..eb58cbfd0c --- /dev/null +++ b/crates/lib/database_queries/src/queries/tts/tts_models/count_tts_model_uses.rs @@ -0,0 +1,45 @@ +//! NB: This query is potentially long-lived and not suitable for low-latency HTTP endpoints. + +use errors::AnyhowResult; +use log::info; +use sqlx::MySql; +use sqlx::pool::PoolConnection; +use tokens::tokens::tts_models::TtsModelToken; + +pub struct TtsModelUseCountInfo { + pub use_count: u64, +} + +pub async fn count_tts_model_uses( + tts_model_token: &TtsModelToken, + last_n_minutes: u64, + mysql_connection: &mut PoolConnection, +) -> AnyhowResult { + + info!("Querying for TTS model {} use count ({} minutes)", tts_model_token, last_n_minutes); + + let use_count = sqlx::query_as!( + RawTtsModelUseCountInfo, + r#" + SELECT + count(*) AS use_count + FROM + tts_results + WHERE + created_at > ( CURDATE() - INTERVAL ? MINUTE ) + AND model_token = ? + "#, + last_n_minutes, + tts_model_token + ) + .fetch_one(mysql_connection) + .await?; + + Ok(TtsModelUseCountInfo { + use_count: use_count.use_count as u64, + }) +} + +struct RawTtsModelUseCountInfo { + pub use_count: i64, +} diff --git a/crates/lib/database_queries/src/queries/tts/tts_models/count_tts_model_uses_total.rs b/crates/lib/database_queries/src/queries/tts/tts_models/count_tts_model_uses_total.rs new file mode 100644 index 0000000000..44b92e2611 --- /dev/null +++ b/crates/lib/database_queries/src/queries/tts/tts_models/count_tts_model_uses_total.rs @@ -0,0 +1,42 @@ +//! NB: This query is potentially long-lived and not suitable for low-latency HTTP endpoints. + +use errors::AnyhowResult; +use log::info; +use sqlx::MySql; +use sqlx::pool::PoolConnection; +use tokens::tokens::tts_models::TtsModelToken; + +pub struct TtsModelTotalUseCountInfo { + pub total_use_count: u64, +} + +pub async fn count_tts_model_uses_total( + tts_model_token: &TtsModelToken, + mysql_connection: &mut PoolConnection, +) -> AnyhowResult { + + info!("Querying for TTS model {} total use count", tts_model_token); + + let total_use_count = sqlx::query_as!( + RawTtsModelTotalUseCountInfo, + r#" + SELECT + count(*) AS total_use_count + FROM + tts_results + WHERE + model_token = ? + "#, + tts_model_token + ) + .fetch_one(mysql_connection) + .await?; + + Ok(TtsModelTotalUseCountInfo { + total_use_count: total_use_count.total_use_count as u64, + }) +} + +struct RawTtsModelTotalUseCountInfo { + pub total_use_count: i64, +} diff --git a/crates/lib/database_queries/src/queries/tts/tts_models/list_all_tts_model_tokens.rs b/crates/lib/database_queries/src/queries/tts/tts_models/list_all_tts_model_tokens.rs new file mode 100644 index 0000000000..c3d7812bf0 --- /dev/null +++ b/crates/lib/database_queries/src/queries/tts/tts_models/list_all_tts_model_tokens.rs @@ -0,0 +1,51 @@ +use log::info; +use errors::AnyhowResult; +use sqlx::MySql; +use sqlx::pool::PoolConnection; +use tokens::tokens::tts_models::TtsModelToken; + +/// List of all TTS model tokens. +/// This is typically used to run jobs that calculate analytics on models. +/// Currently does not need to be batched, but future scale may dictate batching. +#[derive(Clone)] +pub struct TtsModelTokens { + pub tokens: Vec, +} + +#[derive(Clone)] +pub struct TtsModelTokenInfo { + pub token: TtsModelToken, + //pub ietf_language_tag: String, + //pub ietf_primary_language_subtag: String, + //pub creator_set_visibility: Visibility, + //pub maybe_user_deleted_at: Option>, + //pub maybe_mod_deleted_at: Option>, +} + +pub async fn list_all_tts_model_tokens(mysql_connection: &mut PoolConnection) -> AnyhowResult { + info!("Querying for all TTS model tokens"); + + let tokens = sqlx::query_as!( + RawTtsModelTokenInfo, + r#" +SELECT + token as `token: tokens::tokens::tts_models::TtsModelToken` +FROM tts_models + "#) + .fetch_all(mysql_connection) + .await?; + + let tokens = tokens.into_iter() + .map(|token| TtsModelTokenInfo { + token: token.token, + }) + .collect::>(); + + Ok(TtsModelTokens { + tokens, + }) +} + +struct RawTtsModelTokenInfo { + pub token: TtsModelToken, +} diff --git a/crates/lib/database_queries/src/queries/tts/tts_models/mod.rs b/crates/lib/database_queries/src/queries/tts/tts_models/mod.rs index 48396c091f..334a9f9dd3 100644 --- a/crates/lib/database_queries/src/queries/tts/tts_models/mod.rs +++ b/crates/lib/database_queries/src/queries/tts/tts_models/mod.rs @@ -1,7 +1,10 @@ +pub mod count_tts_model_uses; +pub mod count_tts_model_uses_total; pub mod delete_tts_model_various_scopes; pub mod edit_tts_model_details; pub mod edit_tts_model_moderator_details; pub mod get_tts_model; pub mod get_tts_model_for_inference; pub mod insert_tts_model_from_download_job; -pub mod list_tts_models; \ No newline at end of file +pub mod list_all_tts_model_tokens; +pub mod list_tts_models; diff --git a/crates/service/job/analytics_job/Cargo.toml b/crates/service/job/analytics_job/Cargo.toml new file mode 100644 index 0000000000..7d9a9eb705 --- /dev/null +++ b/crates/service/job/analytics_job/Cargo.toml @@ -0,0 +1,50 @@ +[package] +name = "analytics-job" +publish = false +edition = "2018" +version = "0.0.1" +authors = [ + "Brandon Thomas ", + "Brandon Thomas ", +] + +[[bin]] +name = "analytics-job" +path = "src/main.rs" + +[dependencies] + +# Internal +config = { path = "../../../lib/config" } +container_common = { path = "../../../lib/container_common" } +database_queries = { path = "../../../lib/database_queries" } +enums = { path = "../../../_types/enums" } +errors = { path = "../../../lib/errors" } +jobs_common = { path = "../../../lib/jobs_common" } +reusable_types = { path = "../../../_types/reusable_types" } +tokens = { path = "../../../_types/tokens" } + +# External +anyhow = "1.0.40" +chrono = { version = "0.4.19", features = ["serde"] } +clap = "2.34.0" # TODO: This is really out of date, but has conflicts when updated +dotenv = "0.15.0" +easyenv = "0.0.6" +hostname = "0.3.1" +log = "0.4.14" +rand = "0.8.3" + +# MySql Database +# "offline" supports typesafe builds without a live database +# "rustls" doesn't appear to work with IP addresses, per: +# https://github.com/launchbadge/sqlx/issues/1095#issuecomment-798050828 +sqlx = { version = "0.6.2", features = [ "mysql", "runtime-actix-native-tls", "offline", "chrono" ] } + +# Runtime +# NB: We switched to Tokio from `async-std` for broader ecosystem support. +tokio = { version = "1.6.0", features = ["macros"] } + +# Serialization +serde = "1.0.125" +serde_derive = "1.0.125" +serde_json = "1.0.64" diff --git a/crates/service/job/analytics_job/src/job_state.rs b/crates/service/job/analytics_job/src/job_state.rs new file mode 100644 index 0000000000..3c53b14075 --- /dev/null +++ b/crates/service/job/analytics_job/src/job_state.rs @@ -0,0 +1,22 @@ +use sqlx::MySqlPool; + +pub struct JobState { + pub mysql_pool: MySqlPool, + + pub sleep_config: SleepConfigs, +} + +/// Use sleep to not overload the database. +pub struct SleepConfigs { + // How long to wait between individual "jobs". + pub between_job_wait_millis: u64, + + // How long to wait between individual job batches. + pub between_job_batch_wait_millis: u64, + + // How long to wait between individual queries. + pub between_query_wait_millis: u64, + + // How long to wait between errors. + pub between_error_wait_millis: u64, +} diff --git a/crates/service/job/analytics_job/src/jobs/calculate_model_analytics.rs b/crates/service/job/analytics_job/src/jobs/calculate_model_analytics.rs new file mode 100644 index 0000000000..666c97562b --- /dev/null +++ b/crates/service/job/analytics_job/src/jobs/calculate_model_analytics.rs @@ -0,0 +1,118 @@ +use crate::job_state::JobState; +use database_queries::queries::trending_model_analytics::upsert_trending_model_analytics::{Args, ModelToken, upsert_trending_model_analytics}; +use database_queries::queries::tts::tts_models::count_tts_model_uses::count_tts_model_uses; +use database_queries::queries::tts::tts_models::count_tts_model_uses_total::count_tts_model_uses_total; +use database_queries::queries::tts::tts_models::list_all_tts_model_tokens::{list_all_tts_model_tokens, TtsModelTokens}; +use errors::AnyhowResult; +use log::info; +use rand::seq::SliceRandom; +use rand::thread_rng; +use sqlx::MySql; +use sqlx::pool::PoolConnection; +use std::time::Duration; +use enums::by_table::trending_model_analytics::window_name::WindowName; +use tokens::tokens::tts_models::TtsModelToken; + +// TODO(bt, 2023-01-16): Make a job trait and make this an instance. +// Jobs can store state, batch progress, and not starve one another. +// It'll be easier to interleave jobs. + +pub async fn calculate_model_analytics(job_state: &JobState) -> AnyhowResult<()> { + // TODO(bt, 2023-01-16): It's conceivable the pool connection dies mid-workload and we starve + // our tokens. An ideal fix would be a self-healing pool connection, but for now we'll use + // ugly hacks. + let mut mysql_connection = job_state.mysql_pool.acquire().await?; + + let tokens = query_all_model_tokens(&mut mysql_connection).await?; + + std::thread::sleep(Duration::from_millis(100)); + + for token in tokens.tokens { + info!("Running analytics for TTS model {}", token.token); + + let _r = query_single_model_statistics( + job_state, + &mut mysql_connection, + &token.token).await; + + std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_job_wait_millis)); + } + + Ok(()) +} + +async fn query_all_model_tokens( + mysql_connection: &mut PoolConnection, +) -> AnyhowResult { + let mut tokens = list_all_tts_model_tokens(mysql_connection).await?; + + // NB: This is just a hack to not starve the work queue. We should replace this with a + // self-healing connection and batch cursors. + let mut rng = thread_rng(); + tokens.tokens.shuffle(&mut rng); + + Ok(tokens) +} + +async fn query_single_model_statistics( + job_state: &JobState, + mysql_connection: &mut PoolConnection, + model_token: &TtsModelToken, +) -> AnyhowResult<()> { + { + let three_hours_in_minutes = 60 * 3; + + let result = count_tts_model_uses( + model_token, + three_hours_in_minutes, + mysql_connection).await?; + + info!("TTS model {} uses {} times (three hours)", model_token, result.use_count); + + let _r = upsert_trending_model_analytics(Args { + model_token: ModelToken::Tts(model_token), + window_name: WindowName::Last3Hours, + numeric_value: result.use_count, + mysql_connection, + }).await?; + } + + std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_query_wait_millis)); + + { + let three_days_in_minutes = 60 * 24 * 3; + + let result = count_tts_model_uses( + model_token, + three_days_in_minutes, + mysql_connection).await?; + + info!("TTS model {} uses {} times (three days)", model_token, result.use_count); + + let _r = upsert_trending_model_analytics(Args { + model_token: ModelToken::Tts(model_token), + window_name: WindowName::Last3Days, + numeric_value: result.use_count, + mysql_connection, + }).await?; + } + + std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_query_wait_millis)); + + { + let result = count_tts_model_uses_total( + model_token, + mysql_connection).await?; + + info!("TTS model {} uses {} times (total)", model_token, result.total_use_count); + + let _r = upsert_trending_model_analytics(Args { + model_token: ModelToken::Tts(model_token), + window_name: WindowName::AllTime, + numeric_value: result.total_use_count, + mysql_connection, + }).await?; + } + + Ok(()) +} diff --git a/crates/service/job/analytics_job/src/jobs/mod.rs b/crates/service/job/analytics_job/src/jobs/mod.rs new file mode 100644 index 0000000000..3b2fdb8c68 --- /dev/null +++ b/crates/service/job/analytics_job/src/jobs/mod.rs @@ -0,0 +1 @@ +pub mod calculate_model_analytics; \ No newline at end of file diff --git a/crates/service/job/analytics_job/src/main.rs b/crates/service/job/analytics_job/src/main.rs new file mode 100644 index 0000000000..125b5b8502 --- /dev/null +++ b/crates/service/job/analytics_job/src/main.rs @@ -0,0 +1,71 @@ +// Never allow these +#![forbid(private_in_public)] +#![forbid(unused_must_use)] // NB: It's unsafe to not close/check some things + +// Okay to toggle +#![forbid(unreachable_patterns)] +#![forbid(unused_imports)] +#![forbid(unused_mut)] +#![forbid(unused_variables)] + +// Always allow +#![allow(dead_code)] +#![allow(non_snake_case)] + +// Strict +//#![forbid(warnings)] + +pub mod job_state; +pub mod jobs; +pub mod main_loop; + +use config::shared_constants::DEFAULT_MYSQL_CONNECTION_STRING; +use config::shared_constants::DEFAULT_RUST_LOG; +use crate::job_state::{JobState, SleepConfigs}; +use crate::main_loop::main_loop; +use errors::AnyhowResult; +use log::info; +use sqlx::mysql::MySqlPoolOptions; + +#[tokio::main] +async fn main() -> AnyhowResult<()> { + easyenv::init_all_with_default_logging(Some(DEFAULT_RUST_LOG)); + + let _ = dotenv::from_filename(".env-analytics-job").ok(); // NB: Specific to `analytics-job` app. + let _ = dotenv::from_filename(".env-secrets").ok(); // NB: Secrets not to live in source control. + + info!("Obtaining hostname..."); + + let server_hostname = hostname::get() + .ok() + .and_then(|h| h.into_string().ok()) + .unwrap_or("analytics-job".to_string()); + + info!("Hostname: {}", &server_hostname); + + let db_connection_string = + easyenv::get_env_string_or_default( + "MYSQL_URL", + DEFAULT_MYSQL_CONNECTION_STRING); + + info!("Connecting to database..."); + + let mysql_pool = MySqlPoolOptions::new() + .max_connections(5) + .connect(&db_connection_string) + .await?; + + let job_state = JobState { + mysql_pool, + sleep_config: SleepConfigs { + between_job_wait_millis: easyenv::get_env_num("BETWEEN_JOB_WAIT_MILLIS", 100)?, + between_job_batch_wait_millis: easyenv::get_env_num("BETWEEN_JOB_BATCH_WAIT_MILLIS", 5000)?, + between_query_wait_millis: easyenv::get_env_num("BETWEEN_QUERY_WAIT_MILLIS", 100)?, + between_error_wait_millis: easyenv::get_env_num("BETWEEN_ERROR_WAIT_MILLIS", 10_000)?, + }, + }; + + let _r = main_loop(job_state).await; + + Ok(()) +} diff --git a/crates/service/job/analytics_job/src/main_loop.rs b/crates/service/job/analytics_job/src/main_loop.rs new file mode 100644 index 0000000000..cd2d32cf66 --- /dev/null +++ b/crates/service/job/analytics_job/src/main_loop.rs @@ -0,0 +1,21 @@ +use crate::job_state::JobState; +use crate::jobs::calculate_model_analytics::calculate_model_analytics; +use errors::AnyhowResult; +use std::time::Duration; +use log::{error, info}; + +pub async fn main_loop(job_state: JobState) -> AnyhowResult<()> { + loop { + info!("Begin analytics job batch."); + + match calculate_model_analytics(&job_state).await { + Ok(_) => { + std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_job_batch_wait_millis)); + } + Err(e) => { + error!("Error: {:?}", e); + std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_error_wait_millis)); + } + } + } +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_fully_computed_assigned_tts_categories.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_fully_computed_assigned_tts_categories.rs index b299eac7b7..f70f213deb 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_fully_computed_assigned_tts_categories.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_fully_computed_assigned_tts_categories.rs @@ -26,9 +26,10 @@ use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::fmt; use std::sync::Arc; use chrono::{DateTime, Utc}; +use database_queries::queries::trending_model_analytics::list_trending_tts_models::{list_trending_tts_models, TrendingModels}; use tokens::tokens::model_categories::ModelCategoryToken; use tokens::tokens::tts_models::TtsModelToken; -use crate::model::categories::synthetic_category_list::SYNTHETIC_CATEGORY_LATEST_TTS_MODELS; +use crate::model::categories::synthetic_category_list::{SYNTHETIC_CATEGORY_LATEST_TTS_MODELS, SYNTHETIC_CATEGORY_TRENDING_TTS_MODELS}; // =============== Success Response =============== @@ -177,7 +178,12 @@ async fn query_and_construct_payload( category_cache: &SingleItemTtlCache, mysql_pool: &MySqlPool ) -> Result { - let (categories, models, model_category_map) = { + let ( + categories, + models, + model_category_map, + trending_models + ) = { let mut mysql_connection = mysql_pool.acquire() .await .map_err(|e| { @@ -206,12 +212,20 @@ async fn query_and_construct_payload( ListFullyComputedAssignedTtsCategoriesError::ServerError })?; - (categories, models, model_category_map) + let trending_models = list_trending_tts_models(&mut mysql_connection) + .await + .map_err(|e| { + error!("Error querying trending TTS models: {:?}", e); + ListFullyComputedAssignedTtsCategoriesError::ServerError + })?; + + (categories, models, model_category_map, trending_models) }; let mut recursive_category_to_model_map = recursive_category_to_model_map(&model_category_map, &categories); add_recent_models(&mut recursive_category_to_model_map, models); + add_trending_models(&mut recursive_category_to_model_map, trending_models); Ok(ModelTokensByCategoryToken { recursive: recursive_category_to_model_map, @@ -435,6 +449,19 @@ fn add_recent_models(recursive_category_to_model_map: &mut CategoryTokenToModelT recursive_category_to_model_map.insert(synthetic_token, model_tokens); } +fn add_trending_models(recursive_category_to_model_map: &mut CategoryTokenToModelTokensMap, trending_models: TrendingModels) { + // TODO: This is a grossly bad guess. + let model_tokens = trending_models.models.iter() + .take(30) + .map(|trending_model| { + trending_model.model_token.clone() + }) + .collect::>(); + + let synthetic_token = ModelCategoryToken::new_from_str(SYNTHETIC_CATEGORY_TRENDING_TTS_MODELS.category_token); + recursive_category_to_model_map.insert(synthetic_token, model_tokens); +} + #[cfg(test)] mod tests { // TODO: This really needs tests! diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/mod.rs index ed3ae9a1b4..c5950fa2a0 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/mod.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/mod.rs @@ -10,6 +10,7 @@ pub mod misc; pub mod moderation; pub mod service; pub mod stubs; +pub mod trending; pub mod tts; pub mod twitch; pub mod vocoders; diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/trending/list_trending_tts_models.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/trending/list_trending_tts_models.rs new file mode 100644 index 0000000000..2bccd5a54a --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/trending/list_trending_tts_models.rs @@ -0,0 +1,152 @@ +// NB: Incrementally getting rid of build warnings... +//#![forbid(unused_imports)] +//#![forbid(unused_mut)] +//#![forbid(unused_variables)] + +//! This endpoint recursively calculates (and caches) the list of every category a TTS model +//! belongs to. This saves an enormous amount of clientside CPU compute. +//! + +use actix_web::error::ResponseError; +use actix_web::http::StatusCode; +use actix_web::{web, HttpResponse, HttpRequest}; +use chrono::{DateTime, Utc}; +use container_common::anyhow_result::AnyhowResult; +use crate::model::cached_queries::list_cached_tts_categories_for_public_dropdown::list_cached_tts_categories_for_public_dropdown; +use crate::model::categories::synthetic_category_list::SYNTHETIC_CATEGORY_LATEST_TTS_MODELS; +use crate::server_state::ServerState; +use database_queries::queries::model_categories::list_categories_query_builder::CategoryList; +use database_queries::queries::trending_model_analytics::list_trending_tts_models::{list_trending_tts_models, list_trending_tts_models_with_pool}; +use database_queries::queries::tts::tts_category_assignments::fetch_and_build_tts_model_category_map::fetch_and_build_tts_model_category_map_with_connection; +use database_queries::queries::tts::tts_models::list_tts_models::list_tts_models_with_connection; +use enums::by_table::trending_model_analytics::window_name::WindowName; +use http_server_common::response::serialize_as_json_error::serialize_as_json_error; +use log::{info, error}; +use memory_caching::single_item_ttl_cache::SingleItemTtlCache; +use sqlx::pool::PoolConnection; +use sqlx::{MySql, MySqlPool}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; +use std::fmt; +use std::sync::Arc; +use tokens::tokens::model_categories::ModelCategoryToken; +use tokens::tokens::tts_models::TtsModelToken; + +// TODO TODO TODO: This endpoint is not done! +// TODO TODO TODO: This endpoint is not done! +// TODO TODO TODO: This endpoint is not done! +// TODO TODO TODO: This endpoint is not done! +// TODO TODO TODO: This endpoint is not done! +// TODO TODO TODO: This endpoint is not done! + +// =============== Success Response =============== + +#[derive(Serialize)] +pub struct ListTrendingTtsModelsResponse { + pub success: bool, + + pub top_trending: WindowTrends, + + pub top_trending_by_language_code: HashMap, +} + +pub type WindowTrends = HashMap>; + + +// =============== Error Response =============== + +#[derive(Debug, Serialize)] +pub enum ListTrendingTtsModelsError { + ServerError, +} + +impl ResponseError for ListTrendingTtsModelsError { + fn status_code(&self) -> StatusCode { + match *self { + ListTrendingTtsModelsError::ServerError => StatusCode::INTERNAL_SERVER_ERROR, + } + } + + fn error_response(&self) -> HttpResponse { + serialize_as_json_error(self) + } +} + +// NB: Not using derive_more::Display since Clion doesn't understand it. +impl fmt::Display for ListTrendingTtsModelsError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{:?}", self) + } +} + +// =============== Handler =============== + +pub async fn list_trending_tts_models_handler( + _http_request: HttpRequest, + server_state: web::Data>) -> Result +{ + +// TODO: Cache the outputs! + +// let maybe_category_assignments = server_state.caches.tts_model_category_assignments.copy_without_bump_if_unexpired() +// .map_err(|e| { +// error!("Error consulting cache: {:?}", e); +// ListTrendingTtsModelsError::ServerError +// })?; +// +// let category_assignments = match maybe_category_assignments { +// Some(category_assignments) => { +// info!("Serving TTS category assignments from cache"); +// category_assignments +// }, +// None => { +// let category_assignments = query_and_construct_payload( +// &server_state.caches.database_tts_category_list, +// &server_state.mysql_pool) +// .await?; +// +// server_state.caches.tts_model_category_assignments.store_copy(&category_assignments) +// .map_err(|e| { +// error!("Error storing cache: {:?}", e); +// ListTrendingTtsModelsError::ServerError +// })?; +// +// category_assignments +// }, +// }; + + let trending_models= list_trending_tts_models_with_pool(&server_state.mysql_pool).await + .map_err(|e| { + error!("Query error: {:?}", e); + ListTrendingTtsModelsError::ServerError + })?; + +// TODO: Actually generate the response body sensibly. + + let mut top_trending = HashMap::new(); + let mut top_trending_by_language_code = HashMap::new(); + + for trending_model in trending_models.models.iter() { + if !top_trending_by_language_code.contains_key(&trending_model.ietf_primary_language_subtag) { + top_trending_by_language_code.insert(trending_model.ietf_primary_language_subtag.clone(), HashMap::new()); + } + + if let Some(window_map) = top_trending_by_language_code.get_mut(&trending_model.ietf_primary_language_subtag) { + if !window_map.contains_key(&trending_model.window_name) { + window_map.insert(trending_model.window_name, Vec::new()); + } + } + } + + let response = ListTrendingTtsModelsResponse { + success: true, + top_trending, + top_trending_by_language_code, + }; + + let body = serde_json::to_string(&response) + .map_err(|_e| ListTrendingTtsModelsError::ServerError)?; + + Ok(HttpResponse::Ok() + .content_type("application/json") + .body(body)) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/trending/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/trending/mod.rs new file mode 100644 index 0000000000..cecf230e52 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/trending/mod.rs @@ -0,0 +1 @@ +pub mod list_trending_tts_models; \ No newline at end of file diff --git a/crates/service/web/storyteller_web/src/routes.rs b/crates/service/web/storyteller_web/src/routes.rs index 84f6c53a33..a8f37cec10 100644 --- a/crates/service/web/storyteller_web/src/routes.rs +++ b/crates/service/web/storyteller_web/src/routes.rs @@ -52,6 +52,7 @@ use crate::http_server::endpoints::stubs::app_model_downloads::get_app_model_dow use crate::http_server::endpoints::stubs::app_news::get_app_news_handler; use crate::http_server::endpoints::stubs::app_plans::get_app_plans_handler; use crate::http_server::endpoints::stubs::post_app_analytics::post_app_analytics_handler; +use crate::http_server::endpoints::trending::list_trending_tts_models::list_trending_tts_models_handler; use crate::http_server::endpoints::tts::delete_tts_model::delete_tts_model_handler; use crate::http_server::endpoints::tts::delete_tts_result::delete_tts_inference_result_handler; use crate::http_server::endpoints::tts::edit_tts_model::edit_tts_model_handler; @@ -128,6 +129,7 @@ pub fn add_routes (app: App) -> App app = add_flag_routes(app); /* /flag */ app = add_desktop_app_routes(app); /* /v1/vc/... */ app = add_media_upload_routes(app); /* /v1/media_upload/... */ + app = add_trending_routes(app); /* /v1/trending/... */ // From components app = add_suggested_api_v1_account_creation_and_session_routes(app); // /create_account, /session, /login, /logout @@ -905,6 +907,28 @@ fn add_media_upload_routes (app: App) -> App ) } +// ==================== TRENDING ROUTES ==================== + +fn add_trending_routes (app: App) -> App + where + B: MessageBody, + T: ServiceFactory< + ServiceRequest, + Config = (), + Response = ServiceResponse, + Error = Error, + InitError = (), + >, +{ + app.service(web::scope("/v1/trending") + .service(web::resource("/tts_models") + .route(web::get().to(list_trending_tts_models_handler)) + .route(web::head().to(|| HttpResponse::Ok())) + ) + ) +} + + // ==================== TWITCH ROUTES ==================== // TODO: Maybe move these into an "oauth-gateway" type http service.