job basis

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"

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

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 <bt@brand.io>",
+    "Brandon Thomas <echelon@gmail.com>",
+]
+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

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'

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

crates/service/job/video_thumbnail_job/src/http_server/endpoints/mod.rs

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

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

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;

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

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

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

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