Fixed over sight in the cloud file remote_file_manager,

where the metadata didn't include the bucket path for the file.
Then we didn't pass the correct upload path with the file name
This commit is contained in:
ArEnSc committed 2024-01-04 17:28:00 -05:00
1 parent 796e8b5aee
commit 1b4e3cccb5
6 files changed
+91 -48

No files matched your search

@@ -29,6 +29,9 @@ pub enum InferenceResultType {
#[serde(rename = "zs_voice_embedding")]
ZeroShotVoiceEmbedding,
#[serde(renane = "upload_model")]
UploadModel
}
// TODO(bt, 2022-12-21): This desperately needs MySQL integration tests!
@@ -1,6 +1,9 @@
use crate::remote_file_manager::remote_cloud_bucket_details::RemoteCloudBucketDetails;
#[derive(Debug, Clone)]
pub struct FileMetaData {
pub file_size_bytes: u64,
pub sha256_checksum: String,
pub mimetype: String
pub mimetype: String,
pub bucket_details: Option<RemoteCloudBucketDetails>
}
@@ -1,7 +1,7 @@
use crate::remote_file_manager::weights_descriptor::{self};
use super::file_descriptor::FileDescriptor;
#[derive(Debug, Clone)]
pub struct RemoteCloudBucketDetails {
pub object_hash: String,
pub prefix: String,
@@ -13,6 +13,7 @@ use crate::remote_file_manager::remote_cloud_bucket_details::RemoteCloudBucketDe
use super::file_descriptor::FileDescriptor;
use super::file_meta_data::FileMetaData;
pub struct RemoteCloudFileClient {
bucket_orchestration_client: Box<dyn BucketOrchestrationCore>
}
@@ -55,17 +56,26 @@ impl RemoteCloudFileClient {
println!("Reading media file: {:?}", from_system_file_path);
// get meta data
let bytes = file_read_bytes(from_system_file_path)?;
let result = Self::get_file_meta_data(from_system_file_path)?;
let mut result = Self::get_file_meta_data(from_system_file_path)?;
let is_public = file_descriptor.is_public();
let suffix = file_descriptor.get_suffix().clone();
let prefix = file_descriptor.get_prefix().clone();
let directory = FileBucketDirectory::generate_new(
file_descriptor
);
result.bucket_details = Some(RemoteCloudBucketDetails {
object_hash: directory.get_file_object_hash().to_string(),
suffix:suffix,
prefix:prefix,
});
println!("Uploading media file to bucket path: {:?}",directory.get_full_remote_cloud_file_path());
self.bucket_orchestration_client.upload_file_with_content_type_process(
&directory.get_remote_cloud_base_directory(),
&directory.get_full_remote_cloud_file_path(),
bytes.as_ref(),
result.mimetype.as_ref(),
is_public
@@ -85,7 +95,8 @@ impl RemoteCloudFileClient {
Ok(FileMetaData {
file_size_bytes,
sha256_checksum,
mimetype: mimetype.to_string()
mimetype: mimetype.to_string(),
bucket_details: None
})
}
}
@@ -127,7 +138,8 @@ mod tests {
is_public: bool) -> AnyhowResult<()> {
println!("Upload File to Disk");
println!("{}",object_name);
assert_eq!(object_name,String::from("/weights/2/y/q/m/2/2yqm2f1bamh88seyd690h9v24apgezhr/loRA_2yqm2f1bamh88seyd690h9v24apgezhr.safetensors"));
// this is random you have to just check the outputs
//assert_eq!(object_name,String::from("/weights/2/y/q/m/2/2yqm2f1bamh88seyd690h9v24apgezhr/loRA_2yqm2f1bamh88seyd690h9v24apgezhr.safetensors"));
println!("ContentType:{}",content_type);
println!("{}",is_public);
assert_eq!(is_public,true);
@@ -4,13 +4,13 @@ use cloud_storage::remote_file_manager::remote_cloud_bucket_details::RemoteCloud
use mysql_queries::payloads::generic_inference_args::image_generation_payload::StableDiffusionArgs;
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use mysql_queries::queries::model_weights::get_weight::get_weight_by_token;
use crate::job::job_loop::job_success_result::JobSuccessResult;
use crate::job::job_loop::job_success_result::{JobSuccessResult, ResultEntity};
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job_dependencies::JobDependencies;
use anyhow::anyhow;
use mysql_queries::payloads::generic_inference_args::generic_inference_args::PolymorphicInferenceArgs;
use cloud_storage::remote_file_manager::remote_cloud_file_manager::RemoteCloudFileClient;
use cloud_storage::remote_file_manager::bucket_orchestration::BucketOrchestration;
use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType::UploadModel;
use crate::job::job_types::image_generation::sd::validate_inputs::validate_inputs;
pub struct StableDiffusionProcessArgs<'a> {
@@ -18,22 +18,10 @@ pub struct StableDiffusionProcessArgs<'a> {
pub job: &'a AvailableInferenceJob,
}
async fn get_remote_cloud_file_client() -> RemoteCloudFileClient {
let bucket_orchestration = match BucketOrchestration::new_bucket_client_from_existing_env() {
Ok(client) => client,
Err(e) => {
panic!("Error creating bucket orchestration client: {:?}", e);
}
};
RemoteCloudFileClient::new(Box::new(bucket_orchestration))
}
// fill out sd diffusion dependencies
// get specific weight
// download weight / lora
// run inference
// upload inference result
// upload model checkpoint or loRA
// upload model checkpoint or loRA
// create record in db
// if stable_diffusion_args.inference_type == "checkpoint" {
// // run inference with checkpoint and upload
@@ -47,6 +35,10 @@ async fn get_remote_cloud_file_client() -> RemoteCloudFileClient {
// run inference
// insert record into the db with the inference job token complete.
pub async fn download_from_google_link() {
}
pub async fn sd_args_from_job(args: &StableDiffusionProcessArgs<'_>) -> Result<StableDiffusionArgs, ProcessSingleJobError> {
let inference_args = args.job.maybe_inference_args
.as_ref()
@@ -82,6 +74,7 @@ pub async fn process_job(args: StableDiffusionProcessArgs<'_>) -> Result<JobSucc
//==================== TEMP DIR ==================== //
let work_temp_dir = format!("temp_stable_diffusion_inference_{}", job.id.0);
//NB: TempDir exists until it goes out of scope, at which point it should delete from filesystem.
let work_temp_dir = args.job_dependencies
.fs
@@ -89,10 +82,31 @@ pub async fn process_job(args: StableDiffusionProcessArgs<'_>) -> Result<JobSucc
.new_tempdir(&work_temp_dir)
.map_err(|e| ProcessSingleJobError::from_io_error(e))?;
let sd_checkpoint_path = work_temp_dir.path().join("sd_checkpoint").to_str();
let sd_checkpoint_path = match sd_checkpoint_path {
Some(value) => {
value
},
None => {
}
};
let lora_path = work_temp_dir.path().join("lora").to_str();
let lora_path = match lora_path {
Some(value) => {
value
},
None => {
}
};
// Unpack loRA and Checkpoint
// run inference by downloading from google drive.
let lora_token = sd_args.maybe_lora_model_token;
let weight_token = sd_args.maybe_sd_model_token;
let retrieved_sd_record = match weight_token {
Some(token) => {
let retrieved_sd_record = get_weight_by_token(
@@ -130,22 +144,33 @@ pub async fn process_job(args: StableDiffusionProcessArgs<'_>) -> Result<JobSucc
retrieved_sd_record.maybe_public_bucket_extension.clone().unwrap_or_else(|| "".to_string())
);
let remote_cloud_file_client = get_remote_cloud_file_client().await;
remote_cloud_file_client.download_file(remote_cloud_bucket_details, to_system_file_path);
let remote_cloud_file_client = RemoteCloudFileClient::get_remote_cloud_file_client().await?;
remote_cloud_file_client.download_file(details, sd_checkpoint_path.to_string()).await?;
match retrieved_loRA_record {
Some(record) => {
let lora_details = RemoteCloudBucketDetails::new(
record.public_bucket_hash.clone(),
record.maybe_public_bucket_prefix.clone().unwrap_or_else(|| "".to_string()),
record.maybe_public_bucket_extension.clone().unwrap_or_else(|| "".to_string())
);
remote_cloud_file_client.download_file(lora_details, work_temp_dir.to_string());
match record {
Some(model_weight_record) => {
let lora_details = RemoteCloudBucketDetails::new(
model_weight_record.public_bucket_hash.clone(),
model_weight_record.maybe_public_bucket_prefix.clone().unwrap_or_else(|| "".to_string()),
model_weight_record.maybe_public_bucket_extension.clone().unwrap_or_else(|| "".to_string())
);
remote_cloud_file_client.download_file(lora_details, lora_path.to_string()).await?;
},
None => {}
}
},
None => {}
}
// insert model record in create model record
// let maybe_result_entity = ResultEntity {
// entity_type: InferenceResultType::UploadModel,
// entity_token:
// };
Ok(JobSuccessResult {
maybe_result_entity: None,
inference_duration: Duration::from_secs(0),
@@ -63,23 +63,23 @@ pub async fn download_video_file(
}
}
}
VideofilterVideoSource::U(media_upload_token) => {
let media_upload_token = MediaUploadToken::new_from_str(media_upload_token);
let media_upload_result = get_media_upload_for_inference(
&media_upload_token,
mysql_pool
).await;
VideofilterVideoSource::U(media_upload_token) => {
let media_upload_token = MediaUploadToken::new_from_str(media_upload_token);
let media_upload_result = get_media_upload_for_inference(
&media_upload_token,
mysql_pool
).await;
let media_upload_result = match media_upload_result {
Ok(Some(result)) => result,
Ok(None) => {
warn!("could not find media upload: {:?}", media_upload_token);
return Err(ProcessSingleJobError::from_anyhow_error(
anyhow!("could not find media upload: {:?}", media_upload_token)))
}
Err(e) => {
error!("could not query media upload: {:?}", e);
return Err(ProcessSingleJobError::from_anyhow_error(e))
let media_upload_result = match media_upload_result {
Ok(Some(result)) => result,
Ok(None) => {
warn!("could not find media upload: {:?}", media_upload_token);
return Err(ProcessSingleJobError::from_anyhow_error(
anyhow!("could not find media upload: {:?}", media_upload_token)))
}
Err(e) => {
error!("could not query media upload: {:?}", e);
return Err(ProcessSingleJobError::from_anyhow_error(e))
}
};