job pause file

_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,

_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

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;

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<PathBuf>,
+
   // TODO: Rename
   /// Creates temp directories for scratch files
   pub scoped_temp_dir_creator_for_downloads: ScopedTempDirCreator,

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,