video-thumbnail-job alerts (#1264)

Cargo.lock

@@ -13360,6 +13360,8 @@ dependencies = [
  "jobs_common",
  "log",
  "mysql_queries",
+ "pager",
+ "rootly_client",
  "serde",
  "serde_derive",
  "serde_json",

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

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<Arc<HttpServerSharedState>>,
 ) -> Result<HttpResponse, HealthCheckError> {
   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,

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,
 }

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<Server> {
   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<Server> {
       "CONSECUTIVE_FAILURE_UNHEALTHY_THRESHOLD",
       3,
     )?,
+    pager: args.pager,
+    hostname,
   };
 
   let server_state_arc = web::Data::new(Arc::new(server_state));

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<T>(
+  pager: &Pager,
+  title: &str,
+  err: anyhow::Error,
+) -> anyhow::Result<T> {
+  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)
+}

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.

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;

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: {})",

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,
 }

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 || {

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()
+}

crates/service/job/video_thumbnail_job/src/startup/mod.rs

@@ -0,0 +1 @@
+pub mod build_pager;