From fd2b01334b0003d0757025ea22ffb4503128f3be Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Sun, 30 May 2021 04:12:32 -0400 Subject: [PATCH] more firehose event types --- src/common_queries/firehose_publisher.rs | 38 +++++++++++-------- .../endpoints/events/list_events.rs | 4 +- .../w2l/enqueue_infer_w2l_with_uploads.rs | 8 ++++ .../w2l/enqueue_upload_w2l_template.rs | 3 +- src/job_queries/w2l_inference_job_queries.rs | 4 +- src/w2l_inference_job.rs | 22 ++++++++++- 6 files changed, 56 insertions(+), 23 deletions(-) diff --git a/src/common_queries/firehose_publisher.rs b/src/common_queries/firehose_publisher.rs index e266218275..235f2462d6 100644 --- a/src/common_queries/firehose_publisher.rs +++ b/src/common_queries/firehose_publisher.rs @@ -59,44 +59,52 @@ pub struct FirehosePublisher { impl FirehosePublisher { pub async fn publish_user_sign_up(&self, user_token: &str) -> AnyhowResult<()> { - let token = random_prefix_crockford_token("EV", 32)?; - let event_type = FirehoseEvent::UserSignUp; - let _record_id = self.insert( - event_type, + FirehoseEvent::UserSignUp, Some(user_token), Some(user_token) ).await?; - Ok(()) } - pub async fn publish_w2l_template_upload_enqueue(&self, user_token: &str, job_token: &str) -> AnyhowResult<()> { - let token = random_prefix_crockford_token("EV", 32)?; - let event_type = FirehoseEvent::W2lTemplateUploadStarted; - + pub async fn enqueue_w2l_template_upload(&self, user_token: &str, job_token: &str) -> AnyhowResult<()> { let _record_id = self.insert( - event_type, + FirehoseEvent::W2lTemplateUploadStarted, Some(user_token), Some(job_token) ).await?; + Ok(()) + } + pub async fn enqueue_w2l_inference(&self, maybe_user_token: Option<&str>, job_token: &str) -> AnyhowResult<()> { + let _record_id = self.insert( + FirehoseEvent::W2lInferenceStarted, + maybe_user_token, + Some(job_token) + ).await?; Ok(()) } pub async fn publish_w2l_template_upload_finished(&self, user_token: &str, template_token: &str) -> AnyhowResult<()> { - let token = random_prefix_crockford_token("EV", 32)?; - let event_type = FirehoseEvent::W2lTemplateUploadCompleted; - let _record_id = self.insert( - event_type, + FirehoseEvent::W2lTemplateUploadCompleted, Some(user_token), Some(template_token) ).await?; - Ok(()) } + pub async fn w2l_inference_finished(&self, maybe_user_token: Option<&str>, job_token: &str) -> AnyhowResult<()> { + let _record_id = self.insert( + FirehoseEvent::W2lInferenceCompleted, + maybe_user_token, + Some(job_token) + ).await?; + Ok(()) + } + + // ======================================================================= + async fn insert( &self, event_type: FirehoseEvent, diff --git a/src/http_server/endpoints/events/list_events.rs b/src/http_server/endpoints/events/list_events.rs index a1a48fb112..8055feb660 100644 --- a/src/http_server/endpoints/events/list_events.rs +++ b/src/http_server/endpoints/events/list_events.rs @@ -5,7 +5,9 @@ use actix_web::dev::HttpResponseBuilder; use actix_web::error::ResponseError; use actix_web::http::StatusCode; use actix_web::{Responder, web, HttpResponse, error, HttpRequest, HttpMessage}; +use chrono::{DateTime, Utc}; use crate::AnyhowResult; +use crate::common_queries::list_w2l_templates::{list_w2l_templates, W2lTemplateRecordForList}; use crate::common_queries::sessions::create_session_for_user; use crate::http_server::endpoints::users::create_account::CreateAccountError::{BadInput, ServerError, UsernameTaken, EmailTaken}; use crate::http_server::endpoints::users::login::LoginSuccessResponse; @@ -23,8 +25,6 @@ use sqlx::error::DatabaseError; use sqlx::error::Error::Database; use sqlx::mysql::MySqlDatabaseError; use std::sync::Arc; -use chrono::{DateTime, Utc}; -use crate::common_queries::list_w2l_templates::{list_w2l_templates, W2lTemplateRecordForList}; #[derive(Serialize)] pub struct EventRecord { diff --git a/src/http_server/endpoints/w2l/enqueue_infer_w2l_with_uploads.rs b/src/http_server/endpoints/w2l/enqueue_infer_w2l_with_uploads.rs index e86a323ed7..e196ae6035 100644 --- a/src/http_server/endpoints/w2l/enqueue_infer_w2l_with_uploads.rs +++ b/src/http_server/endpoints/w2l/enqueue_infer_w2l_with_uploads.rs @@ -341,6 +341,14 @@ SET // .map_err(|err| HttpResponse::InternalServerError() // .body(format!("Redis Err: {:?}", err)))?; + server_state.firehose_publisher.enqueue_w2l_inference(maybe_user_token.as_deref(), &job_token) + .await + .map_err(|e| { + warn!("error publishing event: {:?}", e); + InferW2lWithUploadError::ServerError + })?; + + let response = InferW2lWithUploadSuccessResponse { success: true, job_token: job_token.to_string(), diff --git a/src/http_server/endpoints/w2l/enqueue_upload_w2l_template.rs b/src/http_server/endpoints/w2l/enqueue_upload_w2l_template.rs index e9b505d266..609badcf96 100644 --- a/src/http_server/endpoints/w2l/enqueue_upload_w2l_template.rs +++ b/src/http_server/endpoints/w2l/enqueue_upload_w2l_template.rs @@ -200,14 +200,13 @@ SET info!("new w2l template upload job id: {}", record_id); - server_state.firehose_publisher.publish_w2l_template_upload_enqueue(&user_session.user_token, &job_token) + server_state.firehose_publisher.enqueue_w2l_template_upload(&user_session.user_token, &job_token) .await .map_err(|e| { warn!("error publishing event: {:?}", e); UploadW2lTemplateError::ServerError })?; - let response = UploadW2lTemplateSuccessResponse { success: true, job_token: job_token.to_string(), diff --git a/src/job_queries/w2l_inference_job_queries.rs b/src/job_queries/w2l_inference_job_queries.rs index 20760dfc7e..e10a573686 100644 --- a/src/job_queries/w2l_inference_job_queries.rs +++ b/src/job_queries/w2l_inference_job_queries.rs @@ -163,7 +163,7 @@ pub async fn insert_w2l_result>( frame_width: u32, frame_height: u32, duration_millis: u64 -) -> AnyhowResult +) -> AnyhowResult<(u64, String)> { let inference_result_token = random_prefix_crockford_token("W2L_RES:", 32)?; @@ -216,7 +216,7 @@ SET } }; - Ok(record_id) + Ok((record_id, inference_result_token.clone())) } pub struct W2lTemplateRecord2 { diff --git a/src/w2l_inference_job.rs b/src/w2l_inference_job.rs index c43d2debf7..2bc7e220e1 100644 --- a/src/w2l_inference_job.rs +++ b/src/w2l_inference_job.rs @@ -8,6 +8,7 @@ #[macro_use] extern crate serde_derive; pub mod buckets; +pub mod common_queries; pub mod job_queries; pub mod script_execution; pub mod util; @@ -18,6 +19,7 @@ use crate::buckets::bucket_client::BucketClient; use crate::buckets::bucket_path_unifier::BucketPathUnifier; use crate::buckets::bucket_paths::hash_to_bucket_path; use crate::buckets::file_hashing::get_file_hash; +use crate::common_queries::firehose_publisher::FirehosePublisher; use crate::job_queries::w2l_inference_job_queries::W2lInferenceJobRecord; use crate::job_queries::w2l_inference_job_queries::get_w2l_template_by_token; use crate::job_queries::w2l_inference_job_queries::insert_w2l_result; @@ -43,9 +45,9 @@ use std::fs::{File, metadata}; use std::io::{BufReader, Read}; use std::path::{PathBuf, Path}; use std::process::Command; +use std::thread; use std::time::Duration; use tempdir::TempDir; -use std::thread; // Buckets (shared config) const ENV_ACCESS_KEY : &'static str = "ACCESS_KEY"; @@ -77,6 +79,8 @@ struct Inferencer { pub private_bucket_client: BucketClient, pub public_bucket_client: BucketClient, + pub firehose_publisher: FirehosePublisher, + pub bucket_path_unifier: BucketPathUnifier, pub semi_persistent_cache: SemiPersistentCacheDir, @@ -190,6 +194,10 @@ async fn main() -> AnyhowResult<()> { let w2l_end_bump_filename = easyenv::get_env_string_or_default( "W2L_END_BUMP_FILENAME", "vocodes-short-end-bump.mp4"); + let firehose_publisher = FirehosePublisher { + mysql_pool: mysql_pool.clone(), // NB: MySqlPool is clone/send/sync safe + }; + let inferencer = Inferencer { download_temp_directory: temp_directory, mysql_pool, @@ -203,6 +211,7 @@ async fn main() -> AnyhowResult<()> { w2l_inference: w2l_inference_command, bucket_path_unifier: BucketPathUnifier::default_paths(), semi_persistent_cache, + firehose_publisher, w2l_model_filename, w2l_end_bump_filename, }; @@ -510,7 +519,7 @@ async fn process_job(inferencer: &Inferencer, job: &W2lInferenceJobRecord) -> An // ==================== SAVE RECORDS ==================== // info!("Saving w2l inference record..."); - let id = insert_w2l_result( + let (id, inference_result_token) = insert_w2l_result( &inferencer.mysql_pool, job, &result_object_path, @@ -524,6 +533,15 @@ async fn process_job(inferencer: &Inferencer, job: &W2lInferenceJobRecord) -> An info!("Marking job complete..."); mark_w2l_inference_job_done(&inferencer.mysql_pool, job, true).await?; + inferencer.firehose_publisher.w2l_inference_finished( + job.maybe_creator_user_token.as_deref(), + &inference_result_token) + .await + .map_err(|e| { + warn!("error publishing event: {:?}", e); + anyhow!("error publishing event") + })?; + info!("Job {} complete success! Downloaded, ran inference, and uploaded. Saved model record: {}", job.id, id);