From c6c94cfe56913cb29d1b36cf9eb94b6ec4d4f0ba Mon Sep 17 00:00:00 2001 From: Kasisnu <5044120+kasisnu@users.noreply.github.com> Date: Tue, 20 Feb 2024 18:31:09 +0530 Subject: [PATCH 1/2] slightly better instrumentation w/labels --- .../src/job/job_loop/main_loop.rs | 40 ++++++++++--------- .../src/job/job_loop/process_single_job.rs | 13 +++++- .../inference_job/src/util/instrumentation.rs | 9 +++-- 3 files changed, 40 insertions(+), 22 deletions(-) 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 d734a7ae21..3812f276cc 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 @@ -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 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 { 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(()) } diff --git a/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs b/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs index 029b4b564b..373cdf92e0 100644 --- a/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs +++ b/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. diff --git a/crates/service/job/inference_job/src/util/instrumentation.rs b/crates/service/job/inference_job/src/util/instrumentation.rs index 5cbee01ad3..8dcf92abe2 100644 --- a/crates/service/job/inference_job/src/util/instrumentation.rs +++ b/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(()) From 9c674b1579e1db5e3c3de90d232db20007194e49 Mon Sep 17 00:00:00 2001 From: Kasisnu <5044120+kasisnu@users.noreply.github.com> Date: Tue, 20 Feb 2024 18:31:33 +0530 Subject: [PATCH 2/2] send thumbnail task requests from face animator --- .../lipsync/sad_talker/process_job.rs | 46 ++++++++++++++++++- 1 file changed, 45 insertions(+), 1 deletion(-) diff --git a/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs b/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs index 6af98d3035..459a82be19 100644 --- a/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs +++ b/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 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 anyhow::Result<&'static str> { + let ext = match mimetype { + "image/jpeg" => "jpg", + "image/gif" => "gif", + _ => return Err(anyhow!("Mimetype not supported: {}", mimetype)), + }; + Ok(ext) +}