Merge pull request #179 from storytold/brandon-fix-job-pathing

Brandon fix job pathing

crates/lib/deprecated/container_common/src/filesystem/check_directory_exists.rs

@@ -1,15 +1,17 @@
-use std::path::PathBuf;
+use std::path::Path;
 
 use anyhow::bail;
 
 use crate::anyhow_result::AnyhowResult;
 
-pub fn check_directory_exists(path: &PathBuf) -> AnyhowResult<()> {
-  if !path.exists() {
-    bail!("Path doesn't exist: {:?}", path);
+pub fn check_directory_exists<P: AsRef<Path>>(path: P) -> AnyhowResult<()> {
+  let check_path = path.as_ref();
+
+  if !check_path.exists() {
+    bail!("Path doesn't exist: {:?}", check_path);
   }
-  if !path.is_dir() {
-    bail!("Path isn't a directory: {:?}", path);
+  if !check_path.is_dir() {
+    bail!("Path isn't a directory: {:?}", check_path);
   }
   Ok(())
 }

crates/service/job/inference_job/src/job/job_loop/process_single_job.rs

@@ -160,18 +160,19 @@ async fn do_process_single_job(
     }
   }
 
-  // ==================== SETUP TEMP DIRS ==================== //
-
-  let temp_dir = format!("temp_{}", job.id.0);
-  let temp_dir = job_dependencies.fs.scoped_temp_dir_creator_for_downloads.new_tempdir(&temp_dir)
-      .map_err(|err| ProcessSingleJobError::Other(anyhow!("filesystem error: {:?}", err)))?;
-
-  let _p = temp_dir.path(); // TODO: Just so the build doesn't complain about unused. Remove.
+// NB(bt,2023-11-29): This looks dead, but hopefully isn't leaned on for downstream side effects.
+// I'll leave this commented out while dealing with the outage, but this can probably be removed
+// in a short while after we redeploy and verify all the jobs.
+//  // ==================== SETUP TEMP DIRS ==================== //
+//
+//  let temp_dir = format!("temp_{}", job.id.0);
+//  let temp_dir = job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads.new_tempdir(&temp_dir)
+//      .map_err(|err| ProcessSingleJobError::Other(anyhow!("filesystem error: {:?}", err)))?;
+//
+//  let _p = temp_dir.path(); // TODO: Just so the build doesn't complain about unused. Remove.
 
   // ==================== HANDLE DIFFERENT INFERENCE TYPES ==================== //
 
-  let mut maybe_result_entity : Option<ResultEntity>;
-
   let job_success_result = match job.inference_category {
     InferenceCategory::LipsyncAnimation => {
       process_single_lipsync_job(job_dependencies, job).await?

crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs

@@ -74,7 +74,7 @@ pub async fn process_job(args: SadTalkerProcessJobArgs<'_>) -> Result<JobSuccess
 
     let result = downloader.download_if_not_on_filesystem(
       &args.job_dependencies.buckets.private_bucket_client,
-      &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads,
+      &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads,
     ).await;
 
     if let Err(e) = result {

crates/service/job/inference_job/src/job/job_types/tts/tacotron2_v2_early_fakeyou/download_static_dependencies.rs

@@ -39,7 +39,7 @@ pub async fn download_static_dependencies<'a>(
       &mut job_progress_reporter,
       "downloading vocoder (1 of 3)",
       job.id.0,
-      &job_dependencies.fs.scoped_temp_dir_creator_for_downloads,
+      &job_dependencies.fs.scoped_temp_dir_creator_for_long_lived_downloads,
     ).await?;
 
     waveglow_vocoder_model_fs_path
@@ -61,7 +61,7 @@ pub async fn download_static_dependencies<'a>(
       &mut job_progress_reporter,
       "downloading vocoder (2 of 3)",
       job.id.0,
-      &job_dependencies.fs.scoped_temp_dir_creator_for_downloads,
+      &job_dependencies.fs.scoped_temp_dir_creator_for_long_lived_downloads,
     ).await?;
 
     hifigan_vocoder_model_fs_path
@@ -83,7 +83,7 @@ pub async fn download_static_dependencies<'a>(
       &mut job_progress_reporter,
       "downloading vocoder (3 of 3)",
       job.id.0,
-      &job_dependencies.fs.scoped_temp_dir_creator_for_downloads,
+      &job_dependencies.fs.scoped_temp_dir_creator_for_long_lived_downloads,
     ).await?;
 
     hifigan_superres_vocoder_model_fs_path

crates/service/job/inference_job/src/job/job_types/tts/tacotron2_v2_early_fakeyou/process_job.rs

@@ -127,7 +127,7 @@ async fn process_job_with_cleanup(
         &mut job_progress_reporter,
         "downloading user vocoder",
         job.id.0,
-        &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads,
+        &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads,
       ).await?;
 
       Some(custom_vocoder_fs_path)
@@ -148,7 +148,7 @@ async fn process_job_with_cleanup(
       &mut job_progress_reporter,
       "downloading synthesizer",
       job.id.0,
-      &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads,
+      &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads,
     ).await?;
 
     tts_synthesizer_fs_path

crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job_create_voice.rs

@@ -177,7 +177,7 @@ pub async fn process_create_voice(
   for downloader in model_dependencies.downloaders.all_downloaders() {
     let result = downloader.download_if_not_on_filesystem(
       &args.job_dependencies.buckets.private_bucket_client,
-      &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads
+      &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads
     ).await;
     if let Err(e) = result {
       error!("could not download: {:?}", e);

crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job_inference_voice.rs

@@ -82,7 +82,7 @@ pub async fn process_inference_voice(
   for downloader in model_dependencies.downloaders.all_downloaders() {
     let result = downloader.download_if_not_on_filesystem(
       &args.job_dependencies.buckets.private_bucket_client,
-      &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads
+      &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads
     ).await;
 
     if let Err(e) = result {

crates/service/job/inference_job/src/job/job_types/tts/vits/process_job.rs

@@ -78,7 +78,7 @@ pub async fn process_job(args: VitsProcessJobArgs<'_>) -> Result<JobSuccessResul
       &mut job_progress_reporter,
       "downloading synthesizer",
       job.id.0,
-      &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads,
+      &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads,
     ).await?;
 
     vits_traced_synthesizer_fs_path

crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/process_job.rs

@@ -64,7 +64,7 @@ pub async fn process_job(args: RvcV2ProcessJobArgs<'_>) -> Result<JobSuccessResu
 
   model_dependencies.pretrained_hubert_model.download_if_not_on_filesystem(
     &args.job_dependencies.buckets.private_bucket_client,
-    &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads)
+    &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads)
       .await
       .map_err(|e| {
         error!("could not download hubert: {:?}", e);
@@ -95,7 +95,7 @@ pub async fn process_job(args: RvcV2ProcessJobArgs<'_>) -> Result<JobSuccessResu
       &mut job_progress_reporter,
       "downloading rvc (v2) model",
       job.id.0,
-      &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads,
+      &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads,
     ).await?;
 
     fs_path
@@ -113,7 +113,7 @@ pub async fn process_job(args: RvcV2ProcessJobArgs<'_>) -> Result<JobSuccessResu
 
     let result = downloader.download_if_not_on_filesystem(
       &args.job_dependencies.buckets.private_bucket_client,
-      &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads,
+      &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads,
     ).await;
 
     if let Err(e) = result {
@@ -150,7 +150,7 @@ pub async fn process_job(args: RvcV2ProcessJobArgs<'_>) -> Result<JobSuccessResu
         &mut job_progress_reporter,
         "downloading rvc (v2) model index",
         job.id.0,
-        &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads,
+        &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads,
       ).await?;
 
       Some(fs_path)

crates/service/job/inference_job/src/job/job_types/vc/so_vits_svc/process_job.rs

@@ -84,7 +84,7 @@ pub async fn process_job(args: SoVitsSvcProcessJobArgs<'_>) -> Result<JobSuccess
       &mut job_progress_reporter,
       "downloading so-vits-svc model",
       job.id.0,
-      &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads,
+      &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads,
     ).await?;
 
     so_vits_svc_fs_path

crates/service/job/inference_job/src/job_dependencies.rs

@@ -138,26 +138,28 @@ pub struct ClientDependencies {
 }
 
 pub struct FileSystemDetails {
-  /// Temporary directory for storing downloads
-  /// (In prod, typically model files from GCS / NFS PVC mount)
-  pub temp_directory_downloads: PathBuf,
-
-  /// Temporary directory for storing work
-  /// (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
+  /// Directory to store long-term downloads (models)
+  /// (In prod, typically model files from GCS / NFS PVC mount)
+  pub directory_long_lived_downloads: PathBuf,
+
+  /// Temporary directory for storing downloads
   /// Creates temp directories for scratch files
-  pub scoped_temp_dir_creator_for_downloads: ScopedTempDirCreator,
+  pub scoped_temp_dir_creator_for_short_lived_downloads: ScopedTempDirCreator,
+
+  /// Temporary directory for storing downloads
+  /// (In prod, typically model files from GCS / NFS PVC mount)
+  pub scoped_temp_dir_creator_for_long_lived_downloads: ScopedTempDirCreator,
+
+  /// Temporary directory for storing work
+  /// (In prod, typically python inference outputs / emptyDir mount)
   pub scoped_temp_dir_creator_for_work: ScopedTempDirCreator,
 
-  // TODO: Rename
   /// Organizes downloaded files
   pub semi_persistent_cache: SemiPersistentCacheDir,
 }

crates/service/job/inference_job/src/main.rs

@@ -119,35 +119,10 @@ async fn main() -> AnyhowResult<()> {
   )?;
 
   // Where we download models and resources to (typically a shared NFS volume in prod).
-  let temp_directory_downloads = easyenv::get_env_pathbuf_or_default(
-    "TEMP_DIR_DOWNLOADS",
-    PathBuf::from("/tmp/downloads")
-  );
 
-  // Where we store scratch files for workloads (temporary processing).
-  let temp_directory_work = easyenv::get_env_pathbuf_or_default(
-    "TEMP_DIR_WORK",
-    PathBuf::from("/tmp/work")
-  );
-
-  if !container_environment.server_environment.is_deployed_in_production() {
-    warn!("Creating temporary directories for non-production / development only!");
-    create_dir_all_if_missing(&temp_directory_downloads)?;
-    create_dir_all_if_missing(&temp_directory_work)?;
-  }
-
-  check_directory_exists(&temp_directory_downloads)?;
-  check_directory_exists(&temp_directory_work)?;
-
-  let semi_persistent_cache =
-      SemiPersistentCacheDir::configured_root(&temp_directory_downloads);
-
-  // TODO(bt,2023.05.22): create_all_paths() for some subset of jobs.
-  info!("Creating pod semi-persistent cache dirs...");
-  semi_persistent_cache.create_custom_vocoder_model_path()?;
-  semi_persistent_cache.create_tts_pretrained_vocoder_model_path()?;
-  semi_persistent_cache.create_tts_synthesizer_model_path()?;
-  semi_persistent_cache.create_voice_conversion_model_path()?;
+  let semi_persistent_cache = SemiPersistentCacheDir::configured_root(
+    easyenv::get_env_pathbuf_or_default("SEMI_PERSISTENT_DIR_ROOT",
+                                        PathBuf::from("/tmp/persistent")));
 
   let db_connection_string =
       easyenv::get_env_string_or_default(
@@ -242,11 +217,18 @@ async fn main() -> AnyhowResult<()> {
       maybe_keepalive_redis_pool,
     },
     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),
+      directory_long_lived_downloads: easyenv::get_env_pathbuf_or_default(
+        "DIR_LONG_LIVED_DOWNLOADS", PathBuf::from("/tmp/downloads_long_lived")),
+      scoped_temp_dir_creator_for_short_lived_downloads: ScopedTempDirCreator::for_directory(
+        easyenv::get_env_pathbuf_or_default(
+          "SCOPED_TEMP_DIR_SHORT_LIVED_DOWNLOADS", PathBuf::from("/tmp/downloads_long_lived"))),
+      scoped_temp_dir_creator_for_long_lived_downloads: ScopedTempDirCreator::for_directory(
+        easyenv::get_env_pathbuf_or_default(
+          "SCOPED_TEMP_DIR_LONG_LIVED_DOWNLOADS", PathBuf::from("/tmp/downloads_long_lived"))),
+      scoped_temp_dir_creator_for_work: ScopedTempDirCreator::for_directory(
+        easyenv::get_env_pathbuf_or_default(
+          "SCOPED_TEMP_DIR_WORK", PathBuf::from("/tmp/downloads_long_lived"))),
       semi_persistent_cache,
     },
     buckets: BucketDependencies {
@@ -306,6 +288,8 @@ async fn main() -> AnyhowResult<()> {
     },
   };
 
+  set_up_directories(&job_dependencies)?;
+
   std::thread::spawn(move || {
     let actix_runtime = actix_web::rt::System::new();
     let http_server_handle = launch_http_server(create_server_args);
@@ -322,3 +306,32 @@ async fn main() -> AnyhowResult<()> {
   Ok(())
 }
 
+fn set_up_directories(job_dependencies: &JobDependencies) -> AnyhowResult<()> {
+  info!("Setting up and checking file system paths for temporary work, ephemeral and persistent downloads, etc...");
+
+  let fs = &job_dependencies.fs;
+
+  if !job_dependencies.job.info.container.server_environment.is_deployed_in_production() {
+    warn!("Creating directories for non-production / development only!");
+
+    create_dir_all_if_missing(&fs.directory_long_lived_downloads)?;
+    create_dir_all_if_missing(&fs.scoped_temp_dir_creator_for_long_lived_downloads.get_base_dir())?;
+    create_dir_all_if_missing(&fs.scoped_temp_dir_creator_for_short_lived_downloads.get_base_dir())?;
+    create_dir_all_if_missing(&fs.scoped_temp_dir_creator_for_work.get_base_dir())?;
+  }
+
+  check_directory_exists(&fs.directory_long_lived_downloads)?;
+  check_directory_exists(fs.scoped_temp_dir_creator_for_long_lived_downloads.get_base_dir())?;
+  check_directory_exists(fs.scoped_temp_dir_creator_for_short_lived_downloads.get_base_dir())?;
+  check_directory_exists(fs.scoped_temp_dir_creator_for_work.get_base_dir())?;
+
+  // TODO(bt,2023.05.22): create_all_paths() for some subset of jobs.
+  info!("Creating semi-persistent cache dirs...");
+
+  fs.semi_persistent_cache.create_custom_vocoder_model_path()?;
+  fs.semi_persistent_cache.create_tts_pretrained_vocoder_model_path()?;
+  fs.semi_persistent_cache.create_tts_synthesizer_model_path()?;
+  fs.semi_persistent_cache.create_voice_conversion_model_path()?;
+
+  Ok(())
+}

crates/service/job/inference_job/src/util/maybe_download_file_from_bucket.rs

@@ -13,7 +13,7 @@ use crate::util::scoped_temp_dir_creator::ScopedTempDirCreator;
 
 pub async fn maybe_download_file_from_bucket(
   name_or_description_of_file: &str,
-  file_path: &Path,
+  final_filesystem_file_path: &Path,
   bucket_object_path: &Path,
   bucket_client: &BucketClient,
   job_progress_reporter: &mut Box<dyn JobProgressReporter>,
@@ -22,12 +22,12 @@ pub async fn maybe_download_file_from_bucket(
   scoped_tempdir_creator: &ScopedTempDirCreator,
 ) -> Result<(), ProcessSingleJobError> {
 
-  if file_path.exists() {
+  if final_filesystem_file_path.exists() {
     // TODO(bt, 2022-07-15): Check signature of file
     return Ok(())
   }
 
-  warn!("{} does not exist at path: {:?}", name_or_description_of_file, &file_path);
+  warn!("{} does not exist at path: {:?}", name_or_description_of_file, &final_filesystem_file_path);
 
   job_progress_reporter.log_status(job_progress_update_description)
       .map_err(|e| ProcessSingleJobError::Other(e))?;
@@ -54,15 +54,15 @@ pub async fn maybe_download_file_from_bucket(
   info!("Downloaded {} from bucket!", name_or_description_of_file);
 
   info!("Renaming {} temp file from {:?} to {:?}!",
-    name_or_description_of_file, &temp_path, &file_path);
+    name_or_description_of_file, &temp_path, &final_filesystem_file_path);
 
-  std::fs::rename(&temp_path, &file_path)
+  std::fs::rename(&temp_path, &final_filesystem_file_path)
       .map_err(|e| {
         safe_delete_temp_directory(&temp_dir);
         ProcessSingleJobError::from_io_error(e)
       })?;
 
-  info!("Finished downloading {} file to {:?}", name_or_description_of_file, &file_path);
+  info!("Finished downloading {} file to {:?}", name_or_description_of_file, &final_filesystem_file_path);
 
   safe_delete_temp_directory(&temp_dir);
 

crates/service/job/inference_job/src/util/scoped_temp_dir_creator.rs

@@ -15,7 +15,11 @@ impl ScopedTempDirCreator {
     }
   }
 
+  pub fn get_base_dir(&self) -> &Path {
+    &self.base_dir
+  }
+
   pub fn new_tempdir(&self, name: &str) -> std::io::Result<TempDir> {
     TempDir::new_in(&self.base_dir, name)
   }
-}
+}