mirror of
https://github.com/storytold/artcraft.git
synced 2026-11-08 11:01:35 +00:00
Log failure metadata for TTS inference jobs
This commit is contained in:
1 parent
8ee5eca076
commit
fb99af8a23
6 files changed
+63
-18
No files matched your search
+4
-1
@@ -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)
|
||||
|
||||
+11
-2
@@ -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)
|
||||
|
||||
+11
-1
@@ -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)
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
@@ -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))?;
|
||||
|
||||
|
||||
@@ -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)?;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user