diff --git a/_sql/analytics/generic_inference/generic_inference_jobs.sql b/_sql/analytics/generic_inference/generic_inference_jobs.sql index ea3bc20d18..06748d9f22 100644 --- a/_sql/analytics/generic_inference/generic_inference_jobs.sql +++ b/_sql/analytics/generic_inference/generic_inference_jobs.sql @@ -29,7 +29,7 @@ from ( jobs.attempt_count as attempts, trim(replace(replace(jobs.internal_debugging_failure_reason, "\n", ""), "\r", "")) as i_failure_reason, - u.username, + -- u.username, jobs.assigned_worker, jobs.assigned_cluster, jobs.maybe_input_source_token, diff --git a/_sql/analytics/users/paying_user_retention.sql b/_sql/analytics/users/paying_user_retention.sql index 8b238cc022..a7425f435b 100644 --- a/_sql/analytics/users/paying_user_retention.sql +++ b/_sql/analytics/users/paying_user_retention.sql @@ -30,7 +30,7 @@ FROM ( SELECT DISTINCT user_token FROM user_subscriptions ) - AND j.created_at > NOW() - INTERVAL 1 DAY + AND j.created_at > NOW() - INTERVAL 31 DAY UNION SELECT u.username as username, @@ -42,7 +42,7 @@ FROM ( SELECT DISTINCT user_token FROM user_subscriptions ) - AND j.created_at > NOW() - INTERVAL 1 DAY + AND j.created_at > NOW() - INTERVAL 31 DAY UNION SELECT u.username as username, @@ -54,7 +54,7 @@ FROM ( SELECT DISTINCT user_token FROM user_subscriptions ) - AND j.created_at > NOW() - INTERVAL 1 DAY + AND j.created_at > NOW() - INTERVAL 31 DAY ) as t GROUP BY username ORDER BY total_use_count desc diff --git a/crates/service/job/inference_job/src/job/job_loop/main_loop.rs b/crates/service/job/inference_job/src/job/job_loop/main_loop.rs index d7e5e965b6..2e6ac5c8c3 100644 --- a/crates/service/job/inference_job/src/job/job_loop/main_loop.rs +++ b/crates/service/job/inference_job/src/job/job_loop/main_loop.rs @@ -9,11 +9,15 @@ use mysql_queries::queries::generic_inference::job::list_available_generic_infer use mysql_queries::queries::generic_inference::job::mark_generic_inference_job_completely_failed::mark_generic_inference_job_completely_failed; use mysql_queries::queries::generic_inference::job::mark_generic_inference_job_failure::mark_generic_inference_job_failure; use std::time::Duration; +use filesys::file_exists::file_exists; // Job runner timeouts (guards MySQL) const START_TIMEOUT_MILLIS : u64 = 500; const INCREASE_TIMEOUT_MILLIS : u64 = 1000; +/// Pause file millis +const PAUSE_FILE_EXISTS_WAIT_MILLIS : u64 = 1000 * 30; + pub async fn main_loop(job_dependencies: JobDependencies) { let mut noop_logger = NoOpLogger::new(job_dependencies.no_op_logger_millis as i64); @@ -22,6 +26,13 @@ pub async fn main_loop(job_dependencies: JobDependencies) { let mut sort_by_priority_count = 0; loop { + if let Some(pause_file) = job_dependencies.fs.maybe_pause_file.as_deref() { + while file_exists(pause_file) { + warn!("Pause file exists. Pausing until deleted: {:?}", pause_file); + std::thread::sleep(Duration::from_millis(PAUSE_FILE_EXISTS_WAIT_MILLIS)); + } + } + // Don't completely starve low-priority jobs if sort_by_priority_count >= job_dependencies.low_priority_starvation_prevention_every_nth { sort_by_priority_count = 0; diff --git a/crates/service/job/inference_job/src/job_dependencies.rs b/crates/service/job/inference_job/src/job_dependencies.rs index ff4f70ecba..9437d11d18 100644 --- a/crates/service/job/inference_job/src/job_dependencies.rs +++ b/crates/service/job/inference_job/src/job_dependencies.rs @@ -98,6 +98,12 @@ pub struct FileSystemDetails { /// (In prod, typically python inference outputs / emptyDir mount) pub temp_directory_work: PathBuf, + /// If the pause file is set and exists on the filesystem, + /// the job will pause until the file stops existing. + /// Good for live debugging of production k8s clusters without + /// redeploying. + pub maybe_pause_file: Option, + // TODO: Rename /// Creates temp directories for scratch files pub scoped_temp_dir_creator_for_downloads: ScopedTempDirCreator, diff --git a/crates/service/job/inference_job/src/main.rs b/crates/service/job/inference_job/src/main.rs index 546bf7be96..772e534477 100644 --- a/crates/service/job/inference_job/src/main.rs +++ b/crates/service/job/inference_job/src/main.rs @@ -227,6 +227,7 @@ async fn main() -> AnyhowResult<()> { fs: FileSystemDetails { temp_directory_downloads: temp_directory_downloads.clone(), temp_directory_work: temp_directory_work.clone(), + maybe_pause_file: easyenv::get_env_pathbuf_optional("PAUSE_FILE"), scoped_temp_dir_creator_for_downloads: ScopedTempDirCreator::for_directory(&temp_directory_downloads), scoped_temp_dir_creator_for_work: ScopedTempDirCreator::for_directory(&temp_directory_work), semi_persistent_cache,