From 02bb0c8290700d52e7cac490a0abc66b011c6734 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Tue, 17 Mar 2026 20:19:10 -0400 Subject: [PATCH] job basis --- Cargo.lock | 30 ++++ Cargo.toml | 1 + .../job/video_thumbnail_job/Cargo.toml | 56 ++++++++ .../video-thumbnail-job.development.env | 4 + .../endpoints/health_check_handler.rs | 101 +++++++++++++ .../src/http_server/endpoints/mod.rs | 1 + .../http_server/http_server_shared_state.rs | 7 + .../src/http_server/mod.rs | 3 + .../src/http_server/run_http_server.rs | 67 +++++++++ .../src/job_dependencies.rs | 22 +++ .../job/video_thumbnail_job/src/main.rs | 136 ++++++++++++++++++ .../job/video_thumbnail_job/src/main_loop.rs | 27 ++++ .../src/process_job/mod.rs | 0 13 files changed, 455 insertions(+) create mode 100644 crates/service/job/video_thumbnail_job/Cargo.toml create mode 100644 crates/service/job/video_thumbnail_job/config/video-thumbnail-job.development.env create mode 100644 crates/service/job/video_thumbnail_job/src/http_server/endpoints/health_check_handler.rs create mode 100644 crates/service/job/video_thumbnail_job/src/http_server/endpoints/mod.rs create mode 100644 crates/service/job/video_thumbnail_job/src/http_server/http_server_shared_state.rs create mode 100644 crates/service/job/video_thumbnail_job/src/http_server/mod.rs create mode 100644 crates/service/job/video_thumbnail_job/src/http_server/run_http_server.rs create mode 100644 crates/service/job/video_thumbnail_job/src/job_dependencies.rs create mode 100644 crates/service/job/video_thumbnail_job/src/main.rs create mode 100644 crates/service/job/video_thumbnail_job/src/main_loop.rs create mode 100644 crates/service/job/video_thumbnail_job/src/process_job/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 2ae89ced2b..964077dc38 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13686,6 +13686,36 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "video-thumbnail-job" +version = "0.0.1" +dependencies = [ + "actix-web", + "actix_helpers", + "anyhow", + "bootstrap", + "bucket_paths", + "cloud_storage", + "concurrency", + "config", + "easyenv", + "enums", + "errors", + "hashing", + "jobs_common", + "log", + "mysql_queries", + "reqwest 0.12.23", + "serde", + "serde_derive", + "serde_json", + "server_environment", + "sqlx", + "tokens", + "tokio", + "workspace-build-acceleration", +] + [[package]] name = "videos" version = "0.0.1" diff --git a/Cargo.toml b/Cargo.toml index 6829050baa..cc2eba54ae 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -122,6 +122,7 @@ members = [ "crates/service/job/es-update-job", "crates/service/job/inference_job", "crates/service/job/seedance2_pro_job", + "crates/service/job/video_thumbnail_job", "crates/service/job/worldlabs_job", "crates/service/job/tts_download_job", "crates/service/job/tts_inference_job", diff --git a/crates/service/job/video_thumbnail_job/Cargo.toml b/crates/service/job/video_thumbnail_job/Cargo.toml new file mode 100644 index 0000000000..ce8adaeb89 --- /dev/null +++ b/crates/service/job/video_thumbnail_job/Cargo.toml @@ -0,0 +1,56 @@ +[package] +name = "video-thumbnail-job" +edition = "2021" +version = "0.0.1" +authors = [ + "Brandon Thomas ", + "Brandon Thomas ", +] +publish = false + +[[bin]] +name = "video-thumbnail-job" +path = "src/main.rs" + +[package.metadata.cargo-udeps.ignore] +normal = ["workspace-build-acceleration"] + +[dependencies] +# Workspace hack (faster compile times) +workspace-build-acceleration = { workspace = true } + +# Internal +actix_helpers.workspace = true +bootstrap = { path = "../../../lib/bootstrap" } +bucket_paths = { path = "../../../lib/bucket_paths" } +cloud_storage.workspace = true +concurrency = { path = "../../../lib/concurrency" } +config = { path = "../../../lib/deprecated/config" } +easyenv = { path = "../../../lib/easyenv" } +enums.workspace = true +errors.workspace = true +hashing = { path = "../../../lib/files/hashing" } +jobs_common = { path = "../../../lib/jobs_common" } +mysql_queries = { path = "../../../schema/database/mysql_queries" } +server_environment = { path = "../../../lib/server_environment" } +tokens.workspace = true + +# External +anyhow.workspace = true +log.workspace = true + +sqlx = { version = "0.7.4", features = [ "mysql", "runtime-tokio-rustls", "chrono" ] } + +# Runtime +tokio = { version = "1.40.0", features = ["macros", "sync"] } + +# HTTP Client +reqwest.workspace = true + +# HTTP Server +actix-web.workspace = true + +# Serialization +serde.workspace = true +serde_derive.workspace = true +serde_json.workspace = true diff --git a/crates/service/job/video_thumbnail_job/config/video-thumbnail-job.development.env b/crates/service/job/video_thumbnail_job/config/video-thumbnail-job.development.env new file mode 100644 index 0000000000..91b973c216 --- /dev/null +++ b/crates/service/job/video_thumbnail_job/config/video-thumbnail-job.development.env @@ -0,0 +1,4 @@ +# default .env configurations (these can be overridden with k8s) + +# Development +SERVER_ENVIRONMENT='development' diff --git a/crates/service/job/video_thumbnail_job/src/http_server/endpoints/health_check_handler.rs b/crates/service/job/video_thumbnail_job/src/http_server/endpoints/health_check_handler.rs new file mode 100644 index 0000000000..419bcba0cd --- /dev/null +++ b/crates/service/job/video_thumbnail_job/src/http_server/endpoints/health_check_handler.rs @@ -0,0 +1,101 @@ +use std::sync::Arc; + +use actix_web::error::ResponseError; +use actix_web::http::header::ContentType; +use actix_web::http::StatusCode; +use actix_web::{web, HttpRequest, HttpResponse}; +use log::error; + +use actix_helpers::response_serializers::error_to_json_http_response::error_to_json_http_response; + +use crate::http_server::http_server_shared_state::HttpServerSharedState; + +#[derive(Serialize)] +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 total_failure_ratio: f32, + pub total_success_ratio: f32, +} + +#[derive(Debug, Serialize)] +pub enum HealthCheckError { + ServerError, +} + +impl ResponseError for HealthCheckError { + fn status_code(&self) -> StatusCode { + match *self { + HealthCheckError::ServerError => StatusCode::INTERNAL_SERVER_ERROR, + } + } + + fn error_response(&self) -> HttpResponse { + error_to_json_http_response(self) + } +} + +impl std::fmt::Display for HealthCheckError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{:?}", self) + } +} + +pub async fn get_health_check_handler( + _http_request: HttpRequest, + server_state: web::Data>, +) -> Result { + let job_stats = server_state + .job_stats + .get_status() + .map_err(|e| { + error!("Error serving health check status: {:?}", e); + HealthCheckError::ServerError + })?; + + 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 is_healthy = + job_stats.consecutive_failure_count < server_state.consecutive_failure_unhealthy_threshold; + + let response = HealthCheckResponse { + success: true, + is_healthy, + consecutive_failure_count: job_stats.consecutive_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).map_err(|_e| HealthCheckError::ServerError)?; + + if is_healthy { + Ok(HttpResponse::Ok() + .content_type(ContentType::json()) + .body(body)) + } else { + Ok(HttpResponse::InternalServerError() + .content_type(ContentType::json()) + .body(body)) + } +} diff --git a/crates/service/job/video_thumbnail_job/src/http_server/endpoints/mod.rs b/crates/service/job/video_thumbnail_job/src/http_server/endpoints/mod.rs new file mode 100644 index 0000000000..0f408270bf --- /dev/null +++ b/crates/service/job/video_thumbnail_job/src/http_server/endpoints/mod.rs @@ -0,0 +1 @@ +pub mod health_check_handler; diff --git a/crates/service/job/video_thumbnail_job/src/http_server/http_server_shared_state.rs b/crates/service/job/video_thumbnail_job/src/http_server/http_server_shared_state.rs new file mode 100644 index 0000000000..534526580f --- /dev/null +++ b/crates/service/job/video_thumbnail_job/src/http_server/http_server_shared_state.rs @@ -0,0 +1,7 @@ +use jobs_common::job_stats::JobStats; + +#[derive(Clone)] +pub struct HttpServerSharedState { + pub job_stats: JobStats, + pub consecutive_failure_unhealthy_threshold: u64, +} diff --git a/crates/service/job/video_thumbnail_job/src/http_server/mod.rs b/crates/service/job/video_thumbnail_job/src/http_server/mod.rs new file mode 100644 index 0000000000..5d888824cc --- /dev/null +++ b/crates/service/job/video_thumbnail_job/src/http_server/mod.rs @@ -0,0 +1,3 @@ +pub mod endpoints; +pub mod http_server_shared_state; +pub mod run_http_server; diff --git a/crates/service/job/video_thumbnail_job/src/http_server/run_http_server.rs b/crates/service/job/video_thumbnail_job/src/http_server/run_http_server.rs new file mode 100644 index 0000000000..b286560ff0 --- /dev/null +++ b/crates/service/job/video_thumbnail_job/src/http_server/run_http_server.rs @@ -0,0 +1,67 @@ +use std::sync::Arc; + +use actix_web::dev::Server; +use actix_web::middleware::Logger; +use actix_web::{web, App, HttpResponse, HttpServer}; +use log::info; + +use bootstrap::bootstrap::ContainerEnvironment; +use errors::AnyhowResult; +use jobs_common::job_stats::JobStats; + +use crate::http_server::endpoints::health_check_handler::get_health_check_handler; +use crate::http_server::http_server_shared_state::HttpServerSharedState; + +const DEFAULT_BIND_ADDRESS: &str = "0.0.0.0:12345"; +const DEFAULT_NUM_WORKERS: usize = 2; + +pub struct CreateServerArgs { + pub container_environment: ContainerEnvironment, + pub job_stats: JobStats, +} + +pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult { + 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 = args.container_environment.hostname.clone(); + + let server_state = HttpServerSharedState { + job_stats: args.job_stats.clone(), + consecutive_failure_unhealthy_threshold: easyenv::get_env_num( + "CONSECUTIVE_FAILURE_UNHEALTHY_THRESHOLD", + 3, + )?, + }; + + let server_state_arc = web::Data::new(Arc::new(server_state)); + + info!("Starting HTTP service (for k8s health checking)."); + + let log_format = "[%{HOSTNAME}e] IP=[%{X-Forwarded-For}i] \"%r\" %s %b \"%{Referer}i\" \"%{User-Agent}i\" %T"; + + let handle = HttpServer::new(move || { + App::new() + .app_data(server_state_arc.clone()) + .wrap(Logger::new(&log_format).exclude("/_status")) + .service( + web::resource("/") + .route(web::get().to(|| HttpResponse::Ok())) + .route(web::head().to(|| HttpResponse::Ok())), + ) + .service( + web::resource("/_status") + .route(web::get().to(get_health_check_handler)) + .route(web::head().to(|| HttpResponse::Ok())), + ) + }) + .bind(bind_address)? + .workers(num_workers) + .run(); + + Ok(handle) +} + +pub async fn launch_http_server(args: CreateServerArgs) -> AnyhowResult<()> { + run_http_server(args)?.await?; + Ok(()) +} diff --git a/crates/service/job/video_thumbnail_job/src/job_dependencies.rs b/crates/service/job/video_thumbnail_job/src/job_dependencies.rs new file mode 100644 index 0000000000..568b274dbf --- /dev/null +++ b/crates/service/job/video_thumbnail_job/src/job_dependencies.rs @@ -0,0 +1,22 @@ +use cloud_storage::bucket_client::BucketClient; +use concurrency::relaxed_atomic_bool::RelaxedAtomicBool; +use jobs_common::job_stats::JobStats; +use server_environment::ServerEnvironment; +use sqlx::MySqlPool; + +pub struct JobDependencies { + pub mysql_pool: MySqlPool, + + /// Public GCS/S3 bucket for storing generated thumbnails. + pub public_bucket_client: BucketClient, + + pub server_environment: ServerEnvironment, + + pub job_stats: JobStats, + + /// How long to sleep between poll iterations (milliseconds). + pub poll_interval_millis: u64, + + /// Set to `true` from another thread to trigger graceful shutdown. + pub application_shutdown: RelaxedAtomicBool, +} diff --git a/crates/service/job/video_thumbnail_job/src/main.rs b/crates/service/job/video_thumbnail_job/src/main.rs new file mode 100644 index 0000000000..b88808b485 --- /dev/null +++ b/crates/service/job/video_thumbnail_job/src/main.rs @@ -0,0 +1,136 @@ +// Never allow these +#![forbid(private_bounds)] +#![forbid(private_interfaces)] +#![forbid(unused_must_use)] + +// Always allow +#![allow(dead_code)] +#![allow(non_snake_case)] + +#[macro_use] extern crate serde_derive; + +use std::time::Duration; + +use anyhow::anyhow; +use log::{info, warn}; +use sqlx::mysql::MySqlPoolOptions; + +use bootstrap::bootstrap::{bootstrap, BootstrapArgs}; +use cloud_storage::bucket_client::BucketClient; +use concurrency::relaxed_atomic_bool::RelaxedAtomicBool; +use config::shared_constants::DEFAULT_MYSQL_CONNECTION_STRING; +use config::shared_constants::DEFAULT_RUST_LOG; +use errors::AnyhowResult; +use jobs_common::job_stats::JobStats; +use server_environment::ServerEnvironment; + +use crate::http_server::run_http_server::{launch_http_server, CreateServerArgs}; +use crate::main_loop::main_loop; +use crate::job_dependencies::JobDependencies; + +pub mod http_server; +pub mod job_dependencies; +pub mod main_loop; +pub mod process_job; + +// Bucket config +const ENV_ACCESS_KEY: &str = "ACCESS_KEY"; +const ENV_SECRET_KEY: &str = "SECRET_KEY"; +const ENV_REGION_NAME: &str = "REGION_NAME"; +const ENV_PUBLIC_BUCKET_NAME: &str = "PUBLIC_BUCKET_NAME"; +const ENV_S3_ENDPOINT: &str = "S3_COMPATIBLE_ENDPOINT_URL"; + +#[actix_web::main] +async fn main() -> AnyhowResult<()> { + + let container_environment = bootstrap(BootstrapArgs { + app_name: "video-thumbnail-job", + default_logging_override: Some(DEFAULT_RUST_LOG), + config_search_directories: &[".", "./config", "crates/service/job/video_thumbnail_job/config"], + ignore_legacy_dot_env_file: true, + })?; + + info!("Hostname: {}", &container_environment.hostname); + + let _k8s_node_name = easyenv::get_env_string_optional("K8S_NODE_NAME"); + let _k8s_pod_name = easyenv::get_env_string_optional("K8S_POD_NAME"); + + let db_connection_string = easyenv::get_env_string_or_default( + "MYSQL_URL", + DEFAULT_MYSQL_CONNECTION_STRING, + ); + + info!("Connecting to database..."); + + let mysql_pool = MySqlPoolOptions::new() + .max_connections(2) + .connect(&db_connection_string) + .await?; + + info!("Connected to MySQL."); + + let server_environment = ServerEnvironment::from_str( + &easyenv::get_env_string_required("SERVER_ENVIRONMENT")?, + ) + .ok_or(anyhow!("invalid server environment"))?; + + // Bucket setup + let access_key = easyenv::get_env_string_required(ENV_ACCESS_KEY)?; + let secret_key = easyenv::get_env_string_required(ENV_SECRET_KEY)?; + let region_name = easyenv::get_env_string_required(ENV_REGION_NAME)?; + let public_bucket_name = easyenv::get_env_string_required(ENV_PUBLIC_BUCKET_NAME)?; + let s3_compatible_endpoint_url = easyenv::get_env_string_required(ENV_S3_ENDPOINT)?; + + let bucket_timeout = easyenv::get_env_duration_seconds_or_default( + "BUCKET_TIMEOUT_SECONDS", + Duration::from_secs(60 * 5), + ); + + let public_bucket_client = BucketClient::create( + &access_key, + &secret_key, + ®ion_name, + &public_bucket_name, + &s3_compatible_endpoint_url, + None, + Some(bucket_timeout), + )?; + + // How often to poll for work (default: 30 seconds) + let poll_interval_millis: u64 = easyenv::get_env_num( + "VIDEO_THUMBNAIL_POLL_INTERVAL_MILLIS", + 30_000, + )?; + + let application_shutdown = RelaxedAtomicBool::new(false); + let job_stats = JobStats::new(); + + let create_server_args = CreateServerArgs { + container_environment: container_environment.clone(), + job_stats: job_stats.clone(), + }; + + let job_dependencies = JobDependencies { + mysql_pool, + public_bucket_client, + server_environment, + job_stats, + poll_interval_millis, + application_shutdown: application_shutdown.clone(), + }; + + std::thread::spawn(move || { + let actix_runtime = actix_web::rt::System::new(); + let http_server_handle = launch_http_server(create_server_args); + + actix_runtime.block_on(http_server_handle) + .expect("HTTP server should not exit."); + + warn!("Server thread is shut down."); + application_shutdown.set(true); + }); + + main_loop(job_dependencies).await; + + Ok(()) +} diff --git a/crates/service/job/video_thumbnail_job/src/main_loop.rs b/crates/service/job/video_thumbnail_job/src/main_loop.rs new file mode 100644 index 0000000000..5001f676ef --- /dev/null +++ b/crates/service/job/video_thumbnail_job/src/main_loop.rs @@ -0,0 +1,27 @@ +use std::time::Duration; + +use log::{error, info, warn}; + +use crate::job_dependencies::JobDependencies; + +pub async fn main_loop(job_dependencies: JobDependencies) { + while !job_dependencies.application_shutdown.get() { + let result = run_iteration(&job_dependencies).await; + + if let Err(err) = result { + error!("Error in video thumbnail iteration: {:?}", err); + let _ = job_dependencies.job_stats.increment_failure_count(); + } + + tokio::time::sleep(Duration::from_millis(job_dependencies.poll_interval_millis)).await; + } + + warn!("Video thumbnail job main loop is shut down."); +} + +async fn run_iteration(_deps: &JobDependencies) -> anyhow::Result<()> { + // TODO: Query for video media files without thumbnails, generate thumbnails, + // upload to bucket, and update media file records. + info!("Video thumbnail job iteration (not yet implemented)."); + Ok(()) +} diff --git a/crates/service/job/video_thumbnail_job/src/process_job/mod.rs b/crates/service/job/video_thumbnail_job/src/process_job/mod.rs new file mode 100644 index 0000000000..e69de29bb2