diff --git a/crates/service/job/inference_job/src/http_server/endpoints/health_check_handler.rs b/crates/service/job/inference_job/src/http_server/endpoints/health_check_handler.rs index 1e642075e8..21027ebd8a 100644 --- a/crates/service/job/inference_job/src/http_server/endpoints/health_check_handler.rs +++ b/crates/service/job/inference_job/src/http_server/endpoints/health_check_handler.rs @@ -14,11 +14,14 @@ pub struct HealthCheckResponse { pub success: bool, pub is_healthy: bool, + pub consecutive_failure_count: u64, pub consecutive_success_count: u64, + + pub total_failure_count: u64, pub total_success_count: u64, - pub consecutive_failure_count: u64, - pub total_failure_count: u64, + pub total_failure_ratio: f32, + pub total_success_ratio: f32, } @@ -60,15 +63,32 @@ pub async fn get_health_check_handler( HealthCheckError::ServerError })?; + // TODO: Configurability let is_healthy = job_stats.consecutive_failure_count < 10; + let total_tries = job_stats.total_failure_count.saturating_add(job_stats.total_success_count); + + let total_success_ratio = if total_tries > 0 { + (job_stats.total_success_count as f32) / (total_tries as f32) + } else { + 0.0 + }; + + let total_failure_ratio = if total_tries > 0 { + 1.0 - total_success_ratio + } else { + 0.0 + }; + let response = HealthCheckResponse { success: true, is_healthy, consecutive_failure_count: job_stats.consecutive_failure_count, - total_failure_count: job_stats.total_failure_count, consecutive_success_count: job_stats.consecutive_success_count, + total_failure_count: job_stats.total_failure_count, total_success_count: job_stats.total_success_count, + total_failure_ratio, + total_success_ratio, }; let body = serde_json::to_string(&response) diff --git a/crates/service/job/inference_job/src/http_server/run_http_server.rs b/crates/service/job/inference_job/src/http_server/run_http_server.rs index 55752d37f1..e25d4b8440 100644 --- a/crates/service/job/inference_job/src/http_server/run_http_server.rs +++ b/crates/service/job/inference_job/src/http_server/run_http_server.rs @@ -1,3 +1,4 @@ +use actix_web::dev::Server; use actix_web::middleware::Logger; use actix_web::{App, HttpResponse, HttpServer, middleware, web}; use crate::http_server::endpoints::health_check_handler::get_health_check_handler; @@ -6,19 +7,26 @@ use crate::job_dependencies::JobDependencies; use errors::AnyhowResult; use log::info; use std::sync::Arc; +use bootstrap::bootstrap::ContainerEnvironment; +use jobs_common::job_stats::JobStats; const DEFAULT_BIND_ADDRESS : &'static str = "0.0.0.0:12345"; const DEFAULT_NUM_WORKERS : usize = 4; -pub async fn run_http_server(job_dependencies: &JobDependencies) -> AnyhowResult<()> +pub struct CreateServerArgs { + pub container_environment: ContainerEnvironment, + pub job_stats: JobStats, +} + +pub async fn run_http_server(args: CreateServerArgs) -> AnyhowResult { // HTTP server args let bind_address = easyenv::get_env_string_or_default("HTTP_BIND_ADDRESS", DEFAULT_BIND_ADDRESS); let num_workers = easyenv::get_env_num("HTTP_NUM_WORKERS", DEFAULT_NUM_WORKERS)?; - let hostname = job_dependencies.container.hostname.clone(); + let hostname = args.container_environment.hostname.clone(); let server_state = HttpServerSharedState { - job_stats: job_dependencies.job_stats.clone(), + job_stats: args.job_stats.clone(), }; let server_state_arc = web::Data::new(Arc::new(server_state)); @@ -28,24 +36,28 @@ pub async fn run_http_server(job_dependencies: &JobDependencies) -> AnyhowResult // NB: We shouldn't be logging much as the /_status endpoint is all we aim to expose. let log_format = "[%{HOSTNAME}e] IP=[%{X-Forwarded-For}i] \"%r\" %s %b \"%{Referer}i\" \"%{User-Agent}i\" %T"; - HttpServer::new(move || { + let handle = HttpServer::new(move || { // NB: app_data being clone()'d below should all be safe (dependencies included) App::new() .app_data(server_state_arc.clone()) .wrap(Logger::new(&log_format) .exclude("/_status") ) - .wrap(middleware::Compress::default()) + //.wrap(middleware::Compress::default()) + .service( + web::resource("/") + .route(web::get().to(|| HttpResponse::Ok())) + .route(web::head().to(|| HttpResponse::Ok())) + ) .service( web::resource("/_status") - .route(web::post().to(get_health_check_handler)) + .route(web::get().to(get_health_check_handler)) .route(web::head().to(|| HttpResponse::Ok())) ) }) .bind(bind_address)? .workers(num_workers) - .run() - .await?; + .run(); - Ok(()) + Ok(handle) } diff --git a/crates/service/job/inference_job/src/main.rs b/crates/service/job/inference_job/src/main.rs index ff4d774479..4de9dbb3b7 100644 --- a/crates/service/job/inference_job/src/main.rs +++ b/crates/service/job/inference_job/src/main.rs @@ -56,6 +56,7 @@ use std::path::PathBuf; use std::time::Duration; use subprocess_common::docker_options::{DockerEnvVar, DockerFilesystemMount, DockerGpu, DockerOptions}; use crate::http_server::run_http_server::run_http_server; +use crate::http_server::run_http_server::CreateServerArgs; // Buckets (shared config) const ENV_ACCESS_KEY : &'static str = "ACCESS_KEY"; @@ -69,7 +70,8 @@ const ENV_PUBLIC_BUCKET_NAME : &'static str = "PUBLIC_BUCKET_NAME"; // HTTP sidecar const ENV_TTS_INFERENCE_SIDECAR_HOSTNAME: &'static str = "TTS_INFERENCE_SIDECAR_HOSTNAME"; -#[tokio::main] +//#[tokio::main] +#[actix_web::main] async fn main() -> AnyhowResult<()> { let container_environment = bootstrap(BootstrapArgs { @@ -225,6 +227,13 @@ async fn main() -> AnyhowResult<()> { } }; + let job_stats = JobStats::new(); + + let create_server_args = CreateServerArgs { + container_environment: container_environment.clone(), + job_stats: job_stats.clone(), + }; + let job_dependencies = JobDependencies { fs: FileSystemDetails { temp_directory_downloads: temp_directory_downloads.clone(), @@ -240,7 +249,7 @@ async fn main() -> AnyhowResult<()> { job_progress_reporter, public_bucket_client, private_bucket_client, - job_stats: JobStats::new(), + job_stats, newrelic_client, newrelic_disabled, worker_details: JobWorkerDetails { @@ -298,7 +307,16 @@ async fn main() -> AnyhowResult<()> { }, }; - //run_http_server(&job_dependencies).await?; + std::thread::spawn(move || { + let http_server_handle = run_http_server(create_server_args); + + let runtime = actix_web::rt::System::new(); + + runtime.block_on(http_server_handle) + }); + + //tokio::spawn(http_server_handle); + main_loop(job_dependencies).await;