From fb99af8a23a4822d3c8231c94480abdc89465548 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Wed, 2 Nov 2022 22:30:38 -0400 Subject: [PATCH] Log failure metadata for TTS inference jobs --- .../mark_tts_inference_job_done.rs | 5 +++- .../mark_tts_inference_job_failure.rs | 13 +++++++-- ...mark_tts_inference_job_permanently_dead.rs | 12 +++++++- .../src/job_steps/job_args.rs | 13 +++++++++ .../src/job_steps/process_single_job.rs | 10 ++----- crates/service/tts_inference_job/src/main.rs | 28 ++++++++++++++----- 6 files changed, 63 insertions(+), 18 deletions(-) diff --git a/crates/lib/database_queries/src/queries/tts/tts_inference_jobs/mark_tts_inference_job_done.rs b/crates/lib/database_queries/src/queries/tts/tts_inference_jobs/mark_tts_inference_job_done.rs index abbd33a670..1816933470 100644 --- a/crates/lib/database_queries/src/queries/tts/tts_inference_jobs/mark_tts_inference_job_done.rs +++ b/crates/lib/database_queries/src/queries/tts/tts_inference_jobs/mark_tts_inference_job_done.rs @@ -8,7 +8,8 @@ pub async fn mark_tts_inference_job_done( pool: &MySqlPool, job_id: TtsInferenceJobId, success: bool, - maybe_result_token: Option<&str> + maybe_result_token: Option<&str>, + last_assigned_worker: &str, ) -> AnyhowResult<()> { let status = if success { "complete_success" } else { "complete_failure" }; @@ -18,12 +19,14 @@ UPDATE tts_inference_jobs SET status = ?, on_success_result_token = ?, + last_assigned_worker = ?, failure_reason = NULL, retry_at = NULL WHERE id = ? "#, status, maybe_result_token, + last_assigned_worker, job_id.0 ) .execute(pool) diff --git a/crates/lib/database_queries/src/queries/tts/tts_inference_jobs/mark_tts_inference_job_failure.rs b/crates/lib/database_queries/src/queries/tts/tts_inference_jobs/mark_tts_inference_job_failure.rs index a68476b592..ea0c6fec6f 100644 --- a/crates/lib/database_queries/src/queries/tts/tts_inference_jobs/mark_tts_inference_job_failure.rs +++ b/crates/lib/database_queries/src/queries/tts/tts_inference_jobs/mark_tts_inference_job_failure.rs @@ -11,7 +11,9 @@ pub async fn mark_tts_inference_job_failure( pool: &MySqlPool, job: &AvailableTtsInferenceJob, failure_reason: &str, - max_attempts: i32 + internal_debugging_failure_reason: &str, + max_attempts: i32, + last_assigned_worker: &str, ) -> AnyhowResult<()> { // statuses: "attempt_failed", "complete_failure", "dead" @@ -22,17 +24,24 @@ pub async fn mark_tts_inference_job_failure( next_status = "dead"; } + let mut internal_debugging_failure_reason = internal_debugging_failure_reason.to_string(); + internal_debugging_failure_reason.truncate(255); // Field is VARCHAR(255) + let query_result = sqlx::query!( r#" UPDATE tts_inference_jobs SET status = ?, failure_reason = ?, + internal_debugging_failure_reason = ?, + last_assigned_worker = ?, retry_at = NOW() + interval 1 minute WHERE id = ? "#, next_status, - failure_reason.to_string(), + failure_reason, + internal_debugging_failure_reason, + last_assigned_worker, job.id.0, ) .execute(pool) diff --git a/crates/lib/database_queries/src/queries/tts/tts_inference_jobs/mark_tts_inference_job_permanently_dead.rs b/crates/lib/database_queries/src/queries/tts/tts_inference_jobs/mark_tts_inference_job_permanently_dead.rs index 0b7c65e649..19e2de037f 100644 --- a/crates/lib/database_queries/src/queries/tts/tts_inference_jobs/mark_tts_inference_job_permanently_dead.rs +++ b/crates/lib/database_queries/src/queries/tts/tts_inference_jobs/mark_tts_inference_job_permanently_dead.rs @@ -9,17 +9,27 @@ pub async fn mark_tts_inference_job_permanently_dead( pool: &MySqlPool, job_id: TtsInferenceJobId, failure_reason: &str, + internal_debugging_failure_reason: &str, + last_assigned_worker: &str, ) -> AnyhowResult<()> { + + let mut internal_debugging_failure_reason = internal_debugging_failure_reason.to_string(); + internal_debugging_failure_reason.truncate(255); // Field is VARCHAR(255) + let query_result = sqlx::query!( r#" UPDATE tts_inference_jobs SET status = "dead", failure_reason = ?, + internal_debugging_failure_reason = ?, + last_assigned_worker = ?, retry_at = NULL WHERE id = ? "#, - failure_reason.to_string(), + failure_reason, + internal_debugging_failure_reason, + last_assigned_worker, job_id.0, ) .execute(pool) diff --git a/crates/service/tts_inference_job/src/job_steps/job_args.rs b/crates/service/tts_inference_job/src/job_steps/job_args.rs index 38f8276748..be4e37ea6b 100644 --- a/crates/service/tts_inference_job/src/job_steps/job_args.rs +++ b/crates/service/tts_inference_job/src/job_steps/job_args.rs @@ -109,4 +109,17 @@ pub struct JobCaches { pub struct JobHttpClients { pub tts_inference_sidecar_client: TtsInferenceSidecarClient, pub tts_sidecar_health_check_client: TtsSidecarHealthCheckClient, +} + +impl JobArgs { + + /// Get the best name for the worker. + pub fn get_worker_name(&self) -> String { + // Default to showing the k8s node (machine) name, if possible, as this benefits + // debugging on-prem workloads. + self.worker_details.k8s_node_name.as_deref() + .or(self.worker_details.k8s_pod_name.as_deref()) + .map(|name| name.to_string()) + .unwrap_or_else(|| self.worker_details.worker_hostname.clone()) + } } \ No newline at end of file diff --git a/crates/service/tts_inference_job/src/job_steps/process_single_job.rs b/crates/service/tts_inference_job/src/job_steps/process_single_job.rs index f2673e0b46..bfd23022b1 100644 --- a/crates/service/tts_inference_job/src/job_steps/process_single_job.rs +++ b/crates/service/tts_inference_job/src/job_steps/process_single_job.rs @@ -361,12 +361,7 @@ pub async fn process_single_job( let text_hash = hash_string_sha2(&cleaned_inference_text) .map_err(|e| ProcessSingleJobError::Other(e))?; - // Default to showing the k8s node (machine) name, if possible, as this benefits - // debugging on-prem workloads. - let worker_name = job_args.worker_details.k8s_node_name.as_deref() - .or(job_args.worker_details.k8s_pod_name.as_deref()) - .map(|name| name.to_string()) - .unwrap_or_else(|| job_args.worker_details.worker_hostname.clone()); + let worker_name = job_args.get_worker_name(); info!("Saving tts inference record..."); @@ -390,7 +385,8 @@ pub async fn process_single_job( &job_args.mysql_pool, job.id, true, - Some(&inference_result_token)) + Some(&inference_result_token), + &worker_name) .await .map_err(|e| ProcessSingleJobError::Other(e))?; diff --git a/crates/service/tts_inference_job/src/main.rs b/crates/service/tts_inference_job/src/main.rs index da9ec0d87a..a171ffa07d 100644 --- a/crates/service/tts_inference_job/src/main.rs +++ b/crates/service/tts_inference_job/src/main.rs @@ -480,6 +480,8 @@ async fn process_jobs( ModelStateError::DatabaseError { .. } => ("unknown database error", false), }; + let internal_debugging_failure_reason = format!("model error: {:?}", e); + let mut job_progress_reporter = inferencer .job_progress_reporter .new_tts_inference(&job.inference_job_token)?; @@ -492,14 +494,18 @@ async fn process_jobs( let _r = mark_tts_inference_job_permanently_dead( &inferencer.mysql_pool, job.id, - failure_reason + failure_reason, + &internal_debugging_failure_reason, + &inferencer.get_worker_name(), ).await; } else { let _r = mark_tts_inference_job_failure( &inferencer.mysql_pool, &job, failure_reason, - inferencer.job_max_attempts + &internal_debugging_failure_reason, + inferencer.job_max_attempts, + &inferencer.get_worker_name(), ).await; } @@ -522,12 +528,17 @@ async fn process_jobs( match maybe_strategy { Err(err) => { warn!("Unable to process job: {:?}", err); - let failure_reason = ""; + + let failure_reason = "cache error"; + let internal_debugging_failure_reason = format!("cache error: {:?}", err); + let _r = mark_tts_inference_job_failure( &inferencer.mysql_pool, &job, failure_reason, - inferencer.job_max_attempts + &internal_debugging_failure_reason, + inferencer.job_max_attempts, + &inferencer.get_worker_name(), ).await; continue; }, @@ -564,12 +575,16 @@ async fn process_jobs( maybe_sidecar_health_issue = true; - let failure_reason = ""; + let failure_reason = "failure processing job"; + let internal_debugging_failure_reason = format!("job error: {:?}", e); + let _r = mark_tts_inference_job_failure( &inferencer.mysql_pool, &job, failure_reason, - inferencer.job_max_attempts + &internal_debugging_failure_reason, + inferencer.job_max_attempts, + &inferencer.get_worker_name(), ).await; match e { @@ -581,7 +596,6 @@ async fn process_jobs( delete_tts_synthesizers_from_cache(&inferencer.semi_persistent_cache)?; } } - } } }