From a9fb97d720808ee51ea8a4e95c76aaa4a5ca5597 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Tue, 31 Mar 2026 03:23:58 -0400 Subject: [PATCH] video-thumbnail-job alerts (#1264) --- Cargo.lock | 2 + .../job/video_thumbnail_job/Cargo.toml | 2 + .../endpoints/health_check_handler.rs | 28 +++++++- .../http_server/http_server_shared_state.rs | 3 + .../src/http_server/run_http_server.rs | 6 +- .../src/job/alert_on_error.rs | 22 ++++++ .../video_thumbnail_job/src/job/main_loop.rs | 2 + .../job/video_thumbnail_job/src/job/mod.rs | 1 + .../src/job/process_single_media_file.rs | 50 +++++++++----- .../src/job_dependencies.rs | 4 ++ .../job/video_thumbnail_job/src/main.rs | 12 ++++ .../src/startup/build_pager.rs | 69 +++++++++++++++++++ .../video_thumbnail_job/src/startup/mod.rs | 1 + 13 files changed, 183 insertions(+), 19 deletions(-) create mode 100644 crates/service/job/video_thumbnail_job/src/job/alert_on_error.rs create mode 100644 crates/service/job/video_thumbnail_job/src/startup/build_pager.rs create mode 100644 crates/service/job/video_thumbnail_job/src/startup/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 28b8c76853..877db61ff0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13360,6 +13360,8 @@ dependencies = [ "jobs_common", "log", "mysql_queries", + "pager", + "rootly_client", "serde", "serde_derive", "serde_json", diff --git a/crates/service/job/video_thumbnail_job/Cargo.toml b/crates/service/job/video_thumbnail_job/Cargo.toml index dfeb0aa1e6..66991eb097 100644 --- a/crates/service/job/video_thumbnail_job/Cargo.toml +++ b/crates/service/job/video_thumbnail_job/Cargo.toml @@ -31,6 +31,8 @@ easyenv.workspace = true errors.workspace = true jobs_common.workspace = true mysql_queries.workspace = true +pager = { path = "../../../lib/pager" } +rootly_client.workspace = true server_environment = { path = "../../../lib/server_environment" } # External 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 index 419bcba0cd..9c5f4a8b8f 100644 --- 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 @@ -7,6 +7,8 @@ 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 pager::notification::notification_details_builder::NotificationDetailsBuilder; +use pager::notification::notification_urgency::NotificationUrgency; use crate::http_server::http_server_shared_state::HttpServerSharedState; @@ -46,7 +48,7 @@ impl std::fmt::Display for HealthCheckError { } pub async fn get_health_check_handler( - _http_request: HttpRequest, + http_request: HttpRequest, server_state: web::Data>, ) -> Result { let job_stats = server_state @@ -76,6 +78,30 @@ pub async fn get_health_check_handler( let is_healthy = job_stats.consecutive_failure_count < server_state.consecutive_failure_unhealthy_threshold; + if !is_healthy { + let notification = NotificationDetailsBuilder::from_title( + format!("Health check unhealthy on {}", server_state.hostname)) + .set_description(Some(format!( + "Health check returned unhealthy.\n\n\ + Hostname: {}\n\ + Consecutive failure count: {}\n\ + Total failure count: {}\n\ + Total success count: {}", + server_state.hostname, + job_stats.consecutive_failure_count, + job_stats.total_failure_count, + job_stats.total_success_count, + ))) + .set_urgency(Some(NotificationUrgency::High)) + .set_http_method(Some(http_request.method().to_string())) + .set_http_path(Some(http_request.path().to_string())) + .build(); + + if let Err(err) = server_state.pager.enqueue_page(notification) { + error!("Failed to enqueue health check alert: {:?}", err); + } + } + let response = HealthCheckResponse { success: true, is_healthy, 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 index 534526580f..ef3e9fe208 100644 --- 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 @@ -1,7 +1,10 @@ use jobs_common::job_stats::JobStats; +use pager::client::pager::Pager; #[derive(Clone)] pub struct HttpServerSharedState { pub job_stats: JobStats, pub consecutive_failure_unhealthy_threshold: u64, + pub pager: Pager, + pub hostname: String, } 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 index b286560ff0..4a55e7aeac 100644 --- 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 @@ -8,6 +8,7 @@ use log::info; use bootstrap::bootstrap::ContainerEnvironment; use errors::AnyhowResult; use jobs_common::job_stats::JobStats; +use pager::client::pager::Pager; use crate::http_server::endpoints::health_check_handler::get_health_check_handler; use crate::http_server::http_server_shared_state::HttpServerSharedState; @@ -18,12 +19,13 @@ const DEFAULT_NUM_WORKERS: usize = 2; pub struct CreateServerArgs { pub container_environment: ContainerEnvironment, pub job_stats: JobStats, + pub pager: Pager, } 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 hostname = args.container_environment.hostname.clone(); let server_state = HttpServerSharedState { job_stats: args.job_stats.clone(), @@ -31,6 +33,8 @@ pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult { "CONSECUTIVE_FAILURE_UNHEALTHY_THRESHOLD", 3, )?, + pager: args.pager, + hostname, }; let server_state_arc = web::Data::new(Arc::new(server_state)); diff --git a/crates/service/job/video_thumbnail_job/src/job/alert_on_error.rs b/crates/service/job/video_thumbnail_job/src/job/alert_on_error.rs new file mode 100644 index 0000000000..9dd1816b3c --- /dev/null +++ b/crates/service/job/video_thumbnail_job/src/job/alert_on_error.rs @@ -0,0 +1,22 @@ +use log::error; +use pager::client::pager::Pager; +use pager::notification::notification_details_builder::NotificationDetailsBuilder; +use pager::notification::notification_urgency::NotificationUrgency; + +/// Enqueue a pager alert for the error, then return it as `Err`. +pub fn alert_pager_and_return_err( + pager: &Pager, + title: &str, + err: anyhow::Error, +) -> anyhow::Result { + let notification = NotificationDetailsBuilder::from_error(&err) + .set_title(title.to_string()) + .set_urgency(Some(NotificationUrgency::Medium)) + .build(); + + if let Err(pager_err) = pager.enqueue_page(notification) { + error!("Failed to enqueue pager alert: {:?}", pager_err); + } + + Err(err) +} diff --git a/crates/service/job/video_thumbnail_job/src/job/main_loop.rs b/crates/service/job/video_thumbnail_job/src/job/main_loop.rs index b6b09f7b2b..be4f6382b4 100644 --- a/crates/service/job/video_thumbnail_job/src/job/main_loop.rs +++ b/crates/service/job/video_thumbnail_job/src/job/main_loop.rs @@ -7,6 +7,7 @@ use mysql_queries::queries::media_files::thumbnails::list_video_media_files_with ListVideoMediaFilesWithoutThumbnailsArgs, }; +use crate::job::alert_on_error::alert_pager_and_return_err; use crate::job_dependencies::JobDependencies; use crate::job::process_single_media_file::process_single_media_file; @@ -21,6 +22,7 @@ pub async fn main_loop(deps: JobDependencies) { } Err(err) => { error!("Error in video thumbnail batch cycle: {:?}", err); + let _ = alert_pager_and_return_err::<()>(&deps.pager, "Video thumbnail batch cycle error", err); let _ = deps.job_stats.increment_failure_count(); // Wait before retrying after a failure. diff --git a/crates/service/job/video_thumbnail_job/src/job/mod.rs b/crates/service/job/video_thumbnail_job/src/job/mod.rs index 4760a9a3b3..8018e2688c 100644 --- a/crates/service/job/video_thumbnail_job/src/job/mod.rs +++ b/crates/service/job/video_thumbnail_job/src/job/mod.rs @@ -1,2 +1,3 @@ +pub mod alert_on_error; pub mod main_loop; pub mod process_single_media_file; diff --git a/crates/service/job/video_thumbnail_job/src/job/process_single_media_file.rs b/crates/service/job/video_thumbnail_job/src/job/process_single_media_file.rs index d57f2f62db..86822b4494 100644 --- a/crates/service/job/video_thumbnail_job/src/job/process_single_media_file.rs +++ b/crates/service/job/video_thumbnail_job/src/job/process_single_media_file.rs @@ -1,6 +1,6 @@ use std::path::PathBuf; -use log::info; +use log::{error, info}; use tempdir::TempDir; use bucket_paths::legacy::typified_paths::public::media_files::bucket_file_path::MediaFileBucketPath; @@ -10,6 +10,7 @@ use ffmpeg_utils::ffmpeg::ffmpeg_video_gif_preview::{ffmpeg_video_gif_preview, F use mysql_queries::queries::media_files::thumbnails::list_video_media_files_without_thumbnails_for_job::VideoMediaFileWithoutThumbnail; use mysql_queries::queries::media_files::thumbnails::update_video_media_file_with_thumbnail::update_video_media_file_with_thumbnail; +use crate::job::alert_on_error::alert_pager_and_return_err; use crate::job_dependencies::JobDependencies; /// A downloaded video file alongside its owning temp directory. @@ -32,7 +33,13 @@ pub async fn process_single_media_file( media_file.created_at, ); - let downloaded = download_video(deps, media_file).await?; + let downloaded = match download_video(deps, media_file).await { + Ok(d) => d, + Err(err) => { + error!("Failed to download video for {}: {:?}", media_file.token.as_str(), err); + return alert_pager_and_return_err(&deps.pager, "Video download failed", err); + } + }; info!( "Downloaded video to {:?}. Generating thumbnails for {:?}.", @@ -46,53 +53,62 @@ pub async fn process_single_media_file( // Generate jpg thumbnail let jpg_path = downloaded.temp_dir.path().join("thumbnail.jpg"); - ffmpeg_video_first_frame_to_jpg_thumbnail( + if let Err(err) = ffmpeg_video_first_frame_to_jpg_thumbnail( FfmpegVideoFirstFrameToJpgThumbnailArgs { input_video_path: &downloaded.file_path, output_jpg_path: &jpg_path, }, - )?; + ) { + error!("Failed to generate JPG thumbnail for {}: {:?}", media_file.token.as_str(), err); + return alert_pager_and_return_err(&deps.pager, "JPG thumbnail generation failed", err.into()); + } info!("Generated JPG thumbnail for {}", media_file.token.as_str()); let jpg_object_path = format!("{video_object_path}{VIDEO_STATIC_JPG_THUMBNAIL_SUFFIX}"); - - deps - .public_bucket_client - .upload_filename(&jpg_object_path, &jpg_path) - .await?; + + if let Err(err) = deps.public_bucket_client.upload_filename(&jpg_object_path, &jpg_path).await { + error!("Failed to upload JPG thumbnail for {}: {:?}", media_file.token.as_str(), err); + return alert_pager_and_return_err(&deps.pager, "JPG thumbnail upload failed", err); + } info!("Uploaded JPG thumbnail to {}", jpg_object_path); // Generate gif thumbnail let gif_path = downloaded.temp_dir.path().join("thumbnail.gif"); - ffmpeg_video_gif_preview( + if let Err(err) = ffmpeg_video_gif_preview( FfmpegVideoGifPreviewArgs { input_video_path: &downloaded.file_path, output_gif_path: &gif_path, }, - )?; + ) { + error!("Failed to generate GIF preview for {}: {:?}", media_file.token.as_str(), err); + return alert_pager_and_return_err(&deps.pager, "GIF preview generation failed", err.into()); + } info!("Generated GIF preview for {}", media_file.token.as_str()); let gif_object_path = format!("{video_object_path}{VIDEO_ANIMATED_GIF_THUMBNAIL_SUFFIX}"); - deps - .public_bucket_client - .upload_filename(&gif_object_path, &gif_path) - .await?; + if let Err(err) = deps.public_bucket_client.upload_filename(&gif_object_path, &gif_path).await { + error!("Failed to upload GIF preview for {}: {:?}", media_file.token.as_str(), err); + return alert_pager_and_return_err(&deps.pager, "GIF preview upload failed", err); + } info!("Uploaded GIF preview to {}", gif_object_path); info!("Marking thumbnail job for {:?} done", media_file.token); // Mark the media file as having a thumbnail in the database. - update_video_media_file_with_thumbnail( + if let Err(err) = update_video_media_file_with_thumbnail( &media_file.token, CURRENT_VIDEO_THUMBNAIL_VERSION, &deps.mysql_pool, - ).await?; + ).await { + error!("Failed to update thumbnail version for {}: {:?}", media_file.token.as_str(), err); + return alert_pager_and_return_err(&deps.pager, "Thumbnail DB update failed", err.into()); + } info!( "Updated thumbnail version for media file {:?} (id: {}, created at: {})", diff --git a/crates/service/job/video_thumbnail_job/src/job_dependencies.rs b/crates/service/job/video_thumbnail_job/src/job_dependencies.rs index 5671612f73..f7895bf446 100644 --- a/crates/service/job/video_thumbnail_job/src/job_dependencies.rs +++ b/crates/service/job/video_thumbnail_job/src/job_dependencies.rs @@ -3,6 +3,7 @@ use std::path::PathBuf; use cloud_storage::bucket_client::BucketClient; use concurrency::relaxed_atomic_bool::RelaxedAtomicBool; use jobs_common::job_stats::JobStats; +use pager::client::pager::Pager; use server_environment::ServerEnvironment; use sqlx::MySqlPool; @@ -45,4 +46,7 @@ pub struct JobDependencies { /// Set to `true` from another thread to trigger graceful shutdown. pub application_shutdown: RelaxedAtomicBool, + + /// Pager client for sending alerts. + pub pager: Pager, } diff --git a/crates/service/job/video_thumbnail_job/src/main.rs b/crates/service/job/video_thumbnail_job/src/main.rs index 2b99e243a6..0ab0d17c55 100644 --- a/crates/service/job/video_thumbnail_job/src/main.rs +++ b/crates/service/job/video_thumbnail_job/src/main.rs @@ -28,10 +28,12 @@ use server_environment::ServerEnvironment; use crate::http_server::run_http_server::{launch_http_server, CreateServerArgs}; use crate::job::main_loop::main_loop; use crate::job_dependencies::{JobDependencies, ShardInfo}; +use crate::startup::build_pager::build_pager; pub mod http_server; pub mod job; pub mod job_dependencies; +pub mod startup; // Bucket config const ENV_ACCESS_KEY: &str = "ACCESS_KEY"; @@ -147,12 +149,21 @@ async fn main() -> AnyhowResult<()> { } }; + let (pager, pager_worker) = build_pager(server_environment, &container_environment.hostname); + + info!("Spawning pager worker thread."); + + tokio::spawn(async move { + pager_worker.run().await; + }); + 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(), + pager: pager.clone(), }; let job_dependencies = JobDependencies { @@ -168,6 +179,7 @@ async fn main() -> AnyhowResult<()> { temp_dir, shard_info, application_shutdown: application_shutdown.clone(), + pager, }; std::thread::spawn(move || { diff --git a/crates/service/job/video_thumbnail_job/src/startup/build_pager.rs b/crates/service/job/video_thumbnail_job/src/startup/build_pager.rs new file mode 100644 index 0000000000..7992d759d4 --- /dev/null +++ b/crates/service/job/video_thumbnail_job/src/startup/build_pager.rs @@ -0,0 +1,69 @@ +use log::{info, warn}; +use pager::client::pager::Pager; +use pager::client::pager_builder::PagerBuilder; +use pager::worker::pager_worker::PagerWorker; +use rootly_client::creds::rootly_api_key::RootlyApiKey; + +/// `video-thumbnail-job` rootly service ID +const ROOTLY_SERVICE_ID: &str = "9d22997c-4474-47cb-af7b-53119eed1a88"; + +const ROOTLY_URGENCY_ID_HIGH: &str = "62fde143-1258-4639-9ee6-1400ebf7308d"; +const ROOTLY_URGENCY_ID_MEDIUM: &str = "7366ba5e-f6ea-4a4e-a1b4-ae43d512e1e2"; +const ROOTLY_URGENCY_ID_LOW: &str = "4db3d3ed-f4ed-4818-82f5-7746da404bd2"; + +pub fn build_pager( + server_environment: server_environment::ServerEnvironment, + hostname: &str, +) -> (Pager, PagerWorker) { + let is_paging_enabled = easyenv::get_env_bool_or_default("ENABLE_PAGING", false); + + info!("Paging enabled: {}", is_paging_enabled); + + let environment = if server_environment.is_deployed_in_production() { + "production" + } else { + "development" + }; + + let builder = PagerBuilder::new() + .application_name("video-thumbnail-job".to_string()) + .environment(environment.to_string()) + .hostname(hostname.to_string()) + .service_id(ROOTLY_SERVICE_ID.to_string()); + + // If paging is globally disabled, use a NoOp pager regardless of API key. + if !is_paging_enabled { + warn!("ENABLE_PAGING is false. Pager will be NoOp."); + return builder.build_with_worker(); + } + + let maybe_api_key = easyenv::get_env_string_optional("ROOTLY_API_KEY"); + + match maybe_api_key { + Some(api_key) => { + info!("Rootly API key found. Configuring pager with Rootly backend."); + build_rootly_pager(builder, api_key) + } + None => { + warn!("ROOTLY_API_KEY not set. Pager will not send real pages."); + builder.build_with_worker() + } + } +} + +fn build_rootly_pager(builder: PagerBuilder, api_key: String) -> (Pager, PagerWorker) { + let mut rootly_builder = builder + .rootly(RootlyApiKey::new(api_key)) + .urgency_id_high(ROOTLY_URGENCY_ID_HIGH.to_string()) + .urgency_id_medium(ROOTLY_URGENCY_ID_MEDIUM.to_string()) + .urgency_id_low(ROOTLY_URGENCY_ID_LOW.to_string()); + + let target_type = easyenv::get_env_string_optional("ROOTLY_NOTIFICATION_TARGET_TYPE"); + let target_id = easyenv::get_env_string_optional("ROOTLY_NOTIFICATION_TARGET_ID"); + + if let (Some(t_type), Some(t_id)) = (target_type, target_id) { + rootly_builder = rootly_builder.notification_target(t_type, t_id); + } + + rootly_builder.build_with_worker() +} diff --git a/crates/service/job/video_thumbnail_job/src/startup/mod.rs b/crates/service/job/video_thumbnail_job/src/startup/mod.rs new file mode 100644 index 0000000000..d074c6c0ce --- /dev/null +++ b/crates/service/job/video_thumbnail_job/src/startup/mod.rs @@ -0,0 +1 @@ +pub mod build_pager;