Merge pull request #305 from storytold/feb-19-mithoo

more labels on inference job metrics, thumbnails for face animator

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

@@ -1,7 +1,7 @@
 use std::time::{Duration, Instant};
 
-use log::{error, info, warn};
 use opentelemetry::KeyValue as OtelAttribute;
+use log::{error, info, warn};
 
 use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
 use errors::AnyhowResult;
@@ -132,9 +132,21 @@ async fn process_job_batch(job_dependencies: &JobDependencies, jobs: Vec<Availab
 
     let start_time = Instant::now();
     let result = process_single_job(job_dependencies, &job).await;
+    let job_duration = Instant::now().duration_since(start_time);
+
+    let model_type_str = match job.maybe_model_type {
+      Some(model_type) => model_type.to_string(),
+      None => "unknown".to_string(),
+    };
+
+    let mut job_duration_instrumentation_attributes = vec![
+      OtelAttribute::new("job_user_is_premium", job.is_from_premium_user),
+      OtelAttribute::new("job_model", model_type_str),
+      OtelAttribute::new("job_inference_category", job.inference_category.to_str()),
+    ];
 
 
-    match result {
+    let managed_result =  match result {
       Ok(success_case) => {
         info!("Job loop iteration ({i} of {job_count} batch) \"success\": {:?}", success_case);
 
@@ -150,32 +162,24 @@ async fn process_job_batch(job_dependencies: &JobDependencies, jobs: Vec<Availab
           job_dependencies.job_instruments.job_success_count.add(1, &[]);
           warn!("Success stats: {:?}", stats);
         }
+
+        Ok(())
       },
       Err(err) => {
         error!(
           r#"Failure to process job ({i} of {job_count} batch): {:?} -
             {:?}
           "#,job.inference_job_token, err);
-        let _r = handle_error(&job_dependencies, &job, err).await?;
+        // we try to handle the error and report details
+        handle_error(&job_dependencies, &job, err).await
       }
-    }
-    let job_duration = Instant::now().duration_since(start_time);;
-
-    let model_type_str = match job.maybe_model_type {
-      Some(model_type) => model_type.to_string(),
-      None => "unknown".to_string(),
     };
 
-    job_dependencies.job_instruments.job_duration.record(job_duration.as_millis() as u64,
-      &[
-        OtelAttribute::new("job_user_is_premium", job.is_from_premium_user),
-        OtelAttribute::new("job_model", model_type_str),
-        OtelAttribute::new("job_status", job.status.to_str()),
-        OtelAttribute::new("job_inference_category", job.inference_category.to_str()),
-      ]
-    );
-  }
+    job_duration_instrumentation_attributes.push(OtelAttribute::new("job_status", managed_result.is_err().then(|| "possibly_failed").unwrap_or("possibly_succeeded")));
 
+    job_dependencies.job_instruments.job_duration.record(job_duration.as_millis() as u64,
+        &job_duration_instrumentation_attributes);
+  }
   Ok(())
 }
 

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

@@ -23,6 +23,7 @@ use crate::job::job_types::vc::process_single_vc_job::process_single_vc_job;
 use crate::job::job_types::videofilter::process_single_vf_job::process_single_vf_job;
 use crate::job::job_types::workflow::process_single_wf_job::process_single_wf_job;
 use crate::job_dependencies::JobDependencies;
+use opentelemetry::{global as otel, KeyValue as OtelAttribute, metrics::Unit};
 
 pub async fn process_single_job(
   job_dependencies: &JobDependencies,
@@ -238,7 +239,17 @@ async fn do_process_single_job(
 
   info!("Saved model record: {} - {}", job.id.0, &job.inference_job_token);
 
-  job_dependencies.job_instruments.inference_command_execution_duration.record(inference_duration.as_millis() as u64, &[]);
+  let model_type_str = job.maybe_model_type
+      .as_ref()
+      .map(|model_type| model_type.to_str())
+      .unwrap_or("unknown");
+
+  job_dependencies.job_instruments.inference_command_execution_duration.record(inference_duration.as_millis() as u64, &[
+    OtelAttribute::new("job_user_is_premium", job.is_from_premium_user),
+    OtelAttribute::new("job_model", model_type_str),
+    OtelAttribute::new("job_status", "complete_success"),
+    OtelAttribute::new("job_inference_category", job.inference_category.to_str()),
+  ]);
 
   // TODO(bt, 2023-01-11): Need to publish that the job finished.
   //  Publish the *correct type* of event.

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

@@ -2,18 +2,20 @@ use std::fs::read_to_string;
 use std::time::Instant;
 
 use anyhow::anyhow;
-use log::{error, info, warn};
+use log::{debug, error, info, warn};
 
 use buckets::public::media_files::bucket_file_path::MediaFileBucketPath;
 use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType;
 use filesys::check_file_exists::check_file_exists;
 use filesys::file_size::file_size;
+use filesys::path_to_string::path_to_string;
 use filesys::safe_delete_temp_directory::safe_delete_temp_directory;
 use filesys::safe_delete_temp_file::safe_delete_temp_file;
 use hashing::sha256::sha256_hash_file::sha256_hash_file;
 use mimetypes::mimetype_for_file::get_mimetype_for_file;
 use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
 use mysql_queries::queries::media_files::create::insert_media_file_from_face_animation::{insert_media_file_from_face_animation, InsertArgs};
+use thumbnail_generator::task_client::thumbnail_task::ThumbnailTaskBuilder;
 use tokens::tokens::users::UserToken;
 
 use crate::job::job_loop::job_success_result::{JobSuccessResult, ResultEntity};
@@ -317,6 +319,39 @@ pub async fn process_job(args: SadTalkerProcessJobArgs<'_>) -> Result<JobSuccess
       .await
       .map_err(|e| ProcessSingleJobError::Other(e))?;
 
+  let thumbnail_types = match mimetype.as_str() {
+    "video/mp4" => vec!["image/gif", "image/jpeg"],
+    _ => vec![],
+  };
+
+  for output_type in thumbnail_types {
+    match get_output_file_extension_from_mimetype(output_type) {
+      Ok(output_extension) => {
+        let thumbnail_task_result = ThumbnailTaskBuilder::new()
+            .with_bucket(&*args.job_dependencies.buckets.public_bucket_client.bucket_name())
+            .with_path(&*path_to_string(result_bucket_object_pathbuf.clone()))
+            .with_source_mimetype(mimetype.as_str())
+            .with_output_mimetype(output_type)
+            .with_output_suffix("thumb")
+            .with_output_extension(output_extension)
+            .with_event_id(&job.id.0.to_string())
+            .send()
+            .await;
+
+        match (thumbnail_task_result) {
+          Ok(thumbnail_task) => {
+            debug!("Thumbnail task created: {:?}", thumbnail_task);
+          },
+          Err(e) => {
+            error!("Failed to create thumbnail task: {:?}", e);
+          }
+        }
+      },
+      Err(e) => {
+          error!("Failed to get output file extension from mimetype: {:?}", e);
+      }
+    }
+  }
   // ==================== DELETE TEMP FILES ==================== //
 
   safe_delete_temp_file(&output_video_fs_path);
@@ -376,3 +411,12 @@ pub async fn process_job(args: SadTalkerProcessJobArgs<'_>) -> Result<JobSuccess
     inference_duration,
   })
 }
+
+fn get_output_file_extension_from_mimetype(mimetype: &str) -> anyhow::Result<&'static str> {
+  let ext = match mimetype {
+    "image/jpeg" => "jpg",
+    "image/gif" => "gif",
+    _ => return Err(anyhow!("Mimetype not supported: {}", mimetype)),
+  };
+  Ok(ext)
+}

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

@@ -35,12 +35,15 @@ pub fn init_otel_metrics_pipeline(
         .with_aggregation_selector(DefaultAggregationSelector::new())
         .with_temporality_selector(DefaultTemporalitySelector::new())
         .build()?;
+
     opentelemetry::global::set_meter_provider(provider);
 
-    if let err= opentelemetry::global::set_error_handler(|error| {
+    match opentelemetry::global::set_error_handler(|error| {
         warn!("OpenTelemetry error: {}", error);
-    }){
-        warn!("Failed to set OpenTelemetry error handler: {:?}", err);
+    }) {
+        (err) => {
+            warn!("Failed to set OpenTelemetry error handler: {:?}", err);
+        }
     };
 
     Ok(())