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 a5375921a0..f993d8f71c 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,8 +1,8 @@ use std::time::Duration; use log::{error, info, warn}; -use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory; +use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory; use errors::AnyhowResult; use filesys::file_exists::file_exists; use jobs_common::noop_logger::NoOpLogger; @@ -23,22 +23,6 @@ const INCREASE_TIMEOUT_MILLIS : u64 = 1000; /// Pause file millis const PAUSE_FILE_EXISTS_WAIT_MILLIS : u64 = 1000 * 30; -#[derive(Eq,PartialEq)] -enum JobFailureClass { - // Jobs that can be retried - TransientFailure, - // Jobs that cannot be retried and must be marked dead - PermanentFailure, -} - -#[derive(Eq,PartialEq)] -enum ContainerHealth { - // No impact to container health - Ignore, - // Increment the container health failure counter - IncrementContainerFailCount, -} - pub async fn main_loop(job_dependencies: JobDependencies) { let mut noop_logger = NoOpLogger::new(job_dependencies.no_op_logger_millis as i64); @@ -140,113 +124,134 @@ async fn process_job_batch(job_dependencies: &JobDependencies, jobs: Vec { - warn!("Failure to process job: {:?}", e); - - let ( - job_failure_class, - container_health_report, - internal_failure_reason, - maybe_public_failure_reason, // TODO(bt,2023-10-11): Remove this column in favor of "frontend_failure_category". - maybe_frontend_failure_category - ) = match e { - // Permanent failures - ProcessSingleJobError::KeepAliveElapsed => - ( - JobFailureClass::PermanentFailure, - ContainerHealth::Ignore, - "keepalive elapsed".to_string(), - Some("keepalive elapsed"), - None, - ), - ProcessSingleJobError::InvalidJob(ref err) => - ( - JobFailureClass::PermanentFailure, - ContainerHealth::Ignore, - format!("InvalidJob: {:?}", err), - Some("invalid job"), - None, - ), - ProcessSingleJobError::NotYetImplemented => - ( - JobFailureClass::PermanentFailure, - ContainerHealth::Ignore, - "not yet implemented".to_string(), - Some("not yet implemented"), - None, - ), - ProcessSingleJobError::FaceDetectionFailure => - ( - JobFailureClass::PermanentFailure, - ContainerHealth::Ignore, - "face not detected".to_string(), - Some("face not detected"), - Some(FrontendFailureCategory::FaceNotDetected), - ), - - // Non-permanent failures - ProcessSingleJobError::FilesystemFull => - ( - JobFailureClass::TransientFailure, - ContainerHealth::IncrementContainerFailCount, - "worker filesystem full".to_string(), - Some("worker filesystem full"), - Some(FrontendFailureCategory::RetryableWorkerError), - ), - ProcessSingleJobError::Other(ref err) => - ( - JobFailureClass::TransientFailure, - ContainerHealth::IncrementContainerFailCount, - format!("OtherErr: {:?}", err), - None, - Some(FrontendFailureCategory::RetryableWorkerError), - ), - }; - - if container_health_report == ContainerHealth::IncrementContainerFailCount { - // NB: We only increment the fail count for events that may indicate the job server is stuck. - let stats = job_dependencies.job_stats.increment_failure_count().ok(); - warn!("Failure stats: {:?}", stats); - } - - match job_failure_class { - JobFailureClass::PermanentFailure => { - let _r = mark_generic_inference_job_completely_failed( - &job_dependencies.mysql_pool, - &job, - maybe_public_failure_reason, - Some(&internal_failure_reason), - maybe_frontend_failure_category, - ).await; - } - JobFailureClass::TransientFailure => { - let _r = mark_generic_inference_job_failure( - &job_dependencies.mysql_pool, - &job, - maybe_public_failure_reason, - &internal_failure_reason, - maybe_frontend_failure_category, - job_dependencies.job_max_attempts - ).await; - } - } - - match e { - // Post failure handling - ProcessSingleJobError::FilesystemFull => { - warn!("Clearing full filesystem..."); - clear_full_filesystem(&job_dependencies.fs.semi_persistent_cache)?; - } - // No-op - ProcessSingleJobError::Other(_) => {} - ProcessSingleJobError::InvalidJob(_) => {} - ProcessSingleJobError::KeepAliveElapsed => {} - ProcessSingleJobError::NotYetImplemented => {} - ProcessSingleJobError::FaceDetectionFailure => {} - } + Err(err) => { + warn!("Failure to process job: {:?}", err); + let _r = handle_error(&job_dependencies, &job, err).await?; } } } Ok(()) } + +#[derive(Eq,PartialEq)] +enum JobFailureClass { + // Jobs that can be retried + TransientFailure, + // Jobs that cannot be retried and must be marked dead + PermanentFailure, +} + +#[derive(Eq,PartialEq)] +enum ContainerHealth { + // No impact to container health + Ignore, + // Increment the container health failure counter + IncrementContainerFailCount, +} + +async fn handle_error(job_dependencies: &&JobDependencies, job: &AvailableInferenceJob, error: ProcessSingleJobError) -> AnyhowResult<()> { + let ( + job_failure_class, + container_health_report, + internal_failure_reason, + maybe_public_failure_reason, // TODO(bt,2023-10-11): Remove this column in favor of "frontend_failure_category". + maybe_frontend_failure_category + ) = match error { + // Permanent failures + ProcessSingleJobError::KeepAliveElapsed => + ( + JobFailureClass::PermanentFailure, + ContainerHealth::Ignore, + "keepalive elapsed".to_string(), + Some("keepalive elapsed"), + None, + ), + ProcessSingleJobError::InvalidJob(ref err) => + ( + JobFailureClass::PermanentFailure, + ContainerHealth::Ignore, + format!("InvalidJob: {:?}", err), + Some("invalid job"), + None, + ), + ProcessSingleJobError::NotYetImplemented => + ( + JobFailureClass::PermanentFailure, + ContainerHealth::Ignore, + "not yet implemented".to_string(), + Some("not yet implemented"), + None, + ), + ProcessSingleJobError::FaceDetectionFailure => + ( + JobFailureClass::PermanentFailure, + ContainerHealth::Ignore, + "face not detected".to_string(), + Some("face not detected"), + Some(FrontendFailureCategory::FaceNotDetected), + ), + + // Non-permanent failures + ProcessSingleJobError::FilesystemFull => + ( + JobFailureClass::TransientFailure, + ContainerHealth::IncrementContainerFailCount, + "worker filesystem full".to_string(), + Some("worker filesystem full"), + Some(FrontendFailureCategory::RetryableWorkerError), + ), + ProcessSingleJobError::Other(ref err) => + ( + JobFailureClass::TransientFailure, + ContainerHealth::IncrementContainerFailCount, + format!("OtherErr: {:?}", err), + None, + Some(FrontendFailureCategory::RetryableWorkerError), + ), + }; + + if container_health_report == ContainerHealth::IncrementContainerFailCount { + // NB: We only increment the fail count for events that may indicate the job server is stuck. + let stats = job_dependencies.job_stats.increment_failure_count().ok(); + warn!("Failure stats: {:?}", stats); + } + + match job_failure_class { + JobFailureClass::PermanentFailure => { + let _r = mark_generic_inference_job_completely_failed( + &job_dependencies.mysql_pool, + &job, + maybe_public_failure_reason, + Some(&internal_failure_reason), + maybe_frontend_failure_category, + ).await; + } + JobFailureClass::TransientFailure => { + let _r = mark_generic_inference_job_failure( + &job_dependencies.mysql_pool, + &job, + maybe_public_failure_reason, + &internal_failure_reason, + maybe_frontend_failure_category, + job_dependencies.job_max_attempts + ).await; + } + } + + match error { + // Post failure handling + ProcessSingleJobError::FilesystemFull => { + warn!("Clearing full filesystem..."); + clear_full_filesystem(&job_dependencies.fs.semi_persistent_cache)?; + } + // No-op + ProcessSingleJobError::Other(_) => {} + ProcessSingleJobError::InvalidJob(_) => {} + ProcessSingleJobError::KeepAliveElapsed => {} + ProcessSingleJobError::NotYetImplemented => {} + ProcessSingleJobError::FaceDetectionFailure => {} + } + + Ok(()) +}