From f9ff9b8c9ffd6ec520418d970bfd73ce5308e661 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Tue, 12 Dec 2023 11:57:42 -0500 Subject: [PATCH] Kill newrelic. This is painful to keep upgrading Keeping the vendor directory because we may introduce later. We've already done some upkeep. --- Cargo.lock | 189 +++--------------- Cargo.toml | 3 +- .../service/job/email_sender_job/Cargo.toml | 3 - crates/service/job/inference_job/Cargo.toml | 3 - .../job/inference_job/src/job_dependencies.rs | 5 - crates/service/job/inference_job/src/main.rs | 9 - .../service/job/tts_inference_job/Cargo.toml | 3 - .../src/job_steps/job_args.rs | 5 - .../src/job_steps/process_single_job.rs | 47 +---- .../service/job/tts_inference_job/src/main.rs | 88 +++----- 10 files changed, 64 insertions(+), 291 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cfd5d64312..25191d742f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -103,7 +103,7 @@ dependencies = [ "flate2", "futures-core", "h2", - "http 0.2.11", + "http", "httparse", "httpdate", "itoa 1.0.9", @@ -176,7 +176,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d66ff4d247d2b160861fa2866457e85706833527840e4133f8f49aa423a38799" dependencies = [ "bytestring", - "http 0.2.11", + "http", "regex", "serde", "tracing", @@ -614,8 +614,8 @@ dependencies = [ "hex", "hmac 0.12.1", "http-types", - "hyper 0.14.27", - "hyper-tls 0.5.0", + "hyper", + "hyper-tls", "serde", "serde_json", "serde_path_to_error", @@ -676,7 +676,7 @@ version = "0.22.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1fcf00bc6d5abb29b5f97e3c61a90b6d3caa12f3faf897d4a3e3607c050a35a7" dependencies = [ - "http 0.2.11", + "http", "log", "native-tls", "serde", @@ -1924,7 +1924,7 @@ dependencies = [ "google_drive_common", "hashing", "hostname", - "hyper 0.14.27", + "hyper", "jobs_common", "log", "mimetypes", @@ -2114,7 +2114,7 @@ dependencies = [ "filesys", "hashing", "hostname", - "hyper 0.14.27", + "hyper", "images", "jobs_common", "lfu", @@ -2599,7 +2599,7 @@ dependencies = [ "futures-core", "futures-sink", "futures-util", - "http 0.2.11", + "http", "indexmap", "slab", "tokio", @@ -2759,17 +2759,6 @@ dependencies = [ "itoa 1.0.9", ] -[[package]] -name = "http" -version = "1.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b32afd38673a8016f7c9ae69e5af41a58f81b1d31689040f2f1959594ce194ea" -dependencies = [ - "bytes 1.5.0", - "fnv", - "itoa 1.0.9", -] - [[package]] name = "http-body" version = "0.4.5" @@ -2777,30 +2766,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d5f38f16d184e36f2408a55281cd658ecbd3ca05cce6d6510a176eca393e26d1" dependencies = [ "bytes 1.5.0", - "http 0.2.11", - "pin-project-lite", -] - -[[package]] -name = "http-body" -version = "1.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1cac85db508abc24a2e48553ba12a996e87244a0395ce011e62b37158745d643" -dependencies = [ - "bytes 1.5.0", - "http 1.0.0", -] - -[[package]] -name = "http-body-util" -version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "41cb79eb393015dadd30fc252023adb0b2400a0caee0fa2a077e6e21a551e840" -dependencies = [ - "bytes 1.5.0", - "futures-util", - "http 1.0.0", - "http-body 1.0.0", + "http", "pin-project-lite", ] @@ -2820,7 +2786,7 @@ dependencies = [ "async-channel", "base64 0.13.1", "futures-lite", - "http 0.2.11", + "http", "infer 0.2.3", "pin-project-lite", "rand 0.7.3", @@ -2884,8 +2850,8 @@ dependencies = [ "futures-core", "futures-util", "h2", - "http 0.2.11", - "http-body 0.4.5", + "http", + "http-body", "httparse", "httpdate", "itoa 1.0.9", @@ -2897,22 +2863,6 @@ dependencies = [ "want", ] -[[package]] -name = "hyper" -version = "1.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "403f9214f3e703236b221f1a9cd88ec8b4adfa5296de01ab96216361f4692f56" -dependencies = [ - "bytes 1.5.0", - "futures-channel", - "futures-util", - "http 1.0.0", - "http-body 1.0.0", - "pin-project-lite", - "tokio", - "want", -] - [[package]] name = "hyper-tls" version = "0.5.0" @@ -2920,48 +2870,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905" dependencies = [ "bytes 1.5.0", - "hyper 0.14.27", + "hyper", "native-tls", "tokio", "tokio-native-tls", ] -[[package]] -name = "hyper-tls" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "70206fc6890eaca9fde8a0bf71caa2ddfc9fe045ac9e5c70df101a7dbde866e0" -dependencies = [ - "bytes 1.5.0", - "http-body-util", - "hyper 1.0.1", - "hyper-util", - "native-tls", - "tokio", - "tokio-native-tls", - "tower-service", -] - -[[package]] -name = "hyper-util" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ca339002caeb0d159cc6e023dff48e199f081e42fa039895c7c6f38b37f2e9d" -dependencies = [ - "bytes 1.5.0", - "futures-channel", - "futures-util", - "http 1.0.0", - "http-body 1.0.0", - "hyper 1.0.1", - "pin-project-lite", - "socket2 0.5.5", - "tokio", - "tower", - "tower-service", - "tracing", -] - [[package]] name = "iana-time-zone" version = "0.1.58" @@ -3095,7 +3009,7 @@ dependencies = [ "filesys", "hashing", "hostname", - "hyper 0.14.27", + "hyper", "images", "jobs_common", "lfu", @@ -3105,7 +3019,6 @@ dependencies = [ "mimetypes", "my-workspace-hack", "mysql_queries", - "newrelic-telemetry", "once_cell", "r2d2_redis", "redis_common", @@ -3772,7 +3685,7 @@ dependencies = [ "hashbrown 0.14.2", "heck", "hmac 0.12.1", - "hyper 0.14.27", + "hyper", "indexmap", "lazy_static", "libc", @@ -3903,25 +3816,6 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e4a24736216ec316047a1fc4252e27dabb04218aa4a3f37c6e7ddbf1f9782b54" -[[package]] -name = "newrelic-telemetry" -version = "0.1.0" -dependencies = [ - "anyhow", - "env_logger", - "flate2", - "futures 0.3.29", - "hyper 0.14.27", - "hyper-tls 0.6.0", - "log", - "my-workspace-hack", - "rand 0.8.5", - "serde", - "serde_json", - "tokio", - "uuid 0.8.2", -] - [[package]] name = "nix" version = "0.26.4" @@ -4884,10 +4778,10 @@ dependencies = [ "futures-core", "futures-util", "h2", - "http 0.2.11", - "http-body 0.4.5", - "hyper 0.14.27", - "hyper-tls 0.5.0", + "http", + "http-body", + "hyper", + "hyper-tls", "ipnet", "js-sys", "log", @@ -5045,7 +4939,7 @@ dependencies = [ "futures 0.3.29", "hex", "hmac 0.12.1", - "http 0.2.11", + "http", "log", "maybe-async", "md5", @@ -6009,7 +5903,7 @@ dependencies = [ "hmac 0.10.1", "hostname", "http_server_common", - "hyper 0.14.27", + "hyper", "infer 0.5.0", "jwt", "language-tags", @@ -6783,28 +6677,6 @@ dependencies = [ "winnow", ] -[[package]] -name = "tower" -version = "0.4.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c" -dependencies = [ - "futures-core", - "futures-util", - "pin-project", - "pin-project-lite", - "tokio", - "tower-layer", - "tower-service", - "tracing", -] - -[[package]] -name = "tower-layer" -version = "0.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c20c8dbed6283a09604c3e69b4b7eeb54e298b8a600d4d5ecb5ad39de609f1d0" - [[package]] name = "tower-service" version = "0.3.2" @@ -6865,7 +6737,7 @@ dependencies = [ "google_drive_common", "hashing", "hostname", - "hyper 0.14.27", + "hyper", "jobs_common", "log", "my-workspace-hack", @@ -6896,14 +6768,13 @@ dependencies = [ "filesys", "hashing", "hostname", - "hyper 0.14.27", + "hyper", "jobs_common", "lfu", "log", "memory_caching", "my-workspace-hack", "mysql_queries", - "newrelic-telemetry", "r2d2_redis", "serde", "serde_derive", @@ -6941,7 +6812,7 @@ dependencies = [ "base64 0.13.1", "byteorder", "bytes 1.5.0", - "http 0.2.11", + "http", "httparse", "log", "rand 0.8.5", @@ -6971,7 +6842,7 @@ dependencies = [ "futures 0.3.29", "futures-util", "hostname", - "hyper 0.14.27", + "hyper", "lazy_static", "log", "magic-crypt", @@ -7013,7 +6884,7 @@ dependencies = [ "chrono", "displaydoc", "futures 0.3.29", - "http 0.2.11", + "http", "once_cell", "reqwest", "serde", @@ -7053,7 +6924,7 @@ dependencies = [ "async-trait", "base64 0.13.1", "displaydoc", - "http 0.2.11", + "http", "once_cell", "rand 0.8.5", "reqwest", @@ -7357,7 +7228,7 @@ dependencies = [ "google_drive_common", "hashing", "hostname", - "hyper 0.14.27", + "hyper", "jobs_common", "log", "my-workspace-hack", @@ -7386,7 +7257,7 @@ dependencies = [ "filesys", "google_drive_common", "hostname", - "hyper 0.14.27", + "hyper", "jobs_common", "log", "my-workspace-hack", @@ -7606,7 +7477,7 @@ dependencies = [ "futures-util", "hostname", "http_server_common", - "hyper 0.14.27", + "hyper", "lazy_static", "limitation", "log", diff --git a/Cargo.toml b/Cargo.toml index 7649065a0d..1ea5d8fefc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -87,7 +87,8 @@ members = [ # Vendored code "crates/vendor/limitation", - "crates/vendor/newrelic-telemetry-sdk-rust", + # NB(bt,2023-12-12): Newrelic keeps breaking with upgrades. Kill for now. + #"crates/vendor/newrelic-telemetry-sdk-rust", # cargo-hakari workspace hack (compilation speedup hack) "crates/my-workspace-hack", diff --git a/crates/service/job/email_sender_job/Cargo.toml b/crates/service/job/email_sender_job/Cargo.toml index f72b7f7fba..733388d420 100644 --- a/crates/service/job/email_sender_job/Cargo.toml +++ b/crates/service/job/email_sender_job/Cargo.toml @@ -44,9 +44,6 @@ tts_common = { path = "../../../lib/tts_common" } # Workspace hack (faster compile times) my-workspace-hack = { version = "0.1", path = "../../../my-workspace-hack" } -# Vendored code -# newrelic-telemetry = { path = "../../../vendor/newrelic-telemetry-sdk-rust" } - # External anyhow = "1.0.75" chrono = { version = "0.4.22", features = ["serde"] } diff --git a/crates/service/job/inference_job/Cargo.toml b/crates/service/job/inference_job/Cargo.toml index 9f8205ca14..199d70f522 100644 --- a/crates/service/job/inference_job/Cargo.toml +++ b/crates/service/job/inference_job/Cargo.toml @@ -41,9 +41,6 @@ tts_common = { path = "../../../lib/tts_common" } # Workspace hack (faster compile times) my-workspace-hack = { version = "0.1", path = "../../../my-workspace-hack" } -# Vendored code -newrelic-telemetry = { path = "../../../vendor/newrelic-telemetry-sdk-rust" } - # External anyhow = "1.0.75" chrono = { version = "0.4.22", features = ["serde"] } diff --git a/crates/service/job/inference_job/src/job_dependencies.rs b/crates/service/job/inference_job/src/job_dependencies.rs index 504d2fb08d..79d8d703c1 100644 --- a/crates/service/job/inference_job/src/job_dependencies.rs +++ b/crates/service/job/inference_job/src/job_dependencies.rs @@ -22,7 +22,6 @@ use mysql_queries::common_inputs::container_environment_arg::ContainerEnvironmen use mysql_queries::mediators::firehose_publisher::FirehosePublisher; use mysql_queries::queries::tts::tts_models::get_tts_model_for_inference_improved::TtsModelForInferenceRecord; use mysql_queries::queries::voice_conversion::inference::get_voice_conversion_model_for_inference::VoiceConversionModelForInference; -use newrelic_telemetry::Client as NewRelicClient; use crate::job_specific_dependencies::JobSpecificDependencies; use crate::util::scoped_execution::ScopedExecution; @@ -135,10 +134,6 @@ pub struct ClientDependencies { pub job_progress_reporter: Box, pub firehose_publisher: FirehosePublisher, - - pub newrelic_client: NewRelicClient, - - pub newrelic_disabled: bool, } pub struct FileSystemDetails { diff --git a/crates/service/job/inference_job/src/main.rs b/crates/service/job/inference_job/src/main.rs index 7779535799..186e87c9f4 100644 --- a/crates/service/job/inference_job/src/main.rs +++ b/crates/service/job/inference_job/src/main.rs @@ -44,7 +44,6 @@ use memory_caching::multi_item_ttl_cache::MultiItemTtlCache; use memory_caching::ttl_key_counter::TtlKeyCounter; use mysql_queries::common_inputs::container_environment_arg::ContainerEnvironmentArg; use mysql_queries::mediators::firehose_publisher::FirehosePublisher; -use newrelic_telemetry::ClientBuilder; use crate::http_server::run_http_server::CreateServerArgs; use crate::http_server::run_http_server::launch_http_server; @@ -149,12 +148,6 @@ async fn main() -> AnyhowResult<()> { mysql_pool: mysql_pool.clone(), // NB: MySqlPool is clone/send/sync safe }; - let license_key = easyenv::get_env_string_required("NEWRELIC_API_KEY")?; - - let newrelic_disabled = easyenv::get_env_bool_or_default("IS_NEWRELIC_DISABLED", false); - - let newrelic_client = ClientBuilder::new(&license_key).build().unwrap(); - let maybe_minimum_priority = easyenv::get_env_string_optional("MAYBE_MINIMUM_PRIORITY") .map(|priority_string| { priority_string.parse::() @@ -237,8 +230,6 @@ async fn main() -> AnyhowResult<()> { clients: ClientDependencies { job_progress_reporter, firehose_publisher, - newrelic_client, - newrelic_disabled, }, job: JobSystemDependencies { system: JobSystemControls { diff --git a/crates/service/job/tts_inference_job/Cargo.toml b/crates/service/job/tts_inference_job/Cargo.toml index cbfdd24de1..2d1f6fa732 100644 --- a/crates/service/job/tts_inference_job/Cargo.toml +++ b/crates/service/job/tts_inference_job/Cargo.toml @@ -30,9 +30,6 @@ tts_common = { path = "../../../lib/tts_common" } # Workspace hack (faster compile times) my-workspace-hack = { version = "0.1", path = "../../../my-workspace-hack" } -# Vendored code -newrelic-telemetry = { path = "../../../vendor/newrelic-telemetry-sdk-rust" } - # External anyhow = "1.0.75" chrono = { version = "0.4.22", features = ["serde"] } diff --git a/crates/service/job/tts_inference_job/src/job_steps/job_args.rs b/crates/service/job/tts_inference_job/src/job_steps/job_args.rs index 433c04771f..3b4296a0d3 100644 --- a/crates/service/job/tts_inference_job/src/job_steps/job_args.rs +++ b/crates/service/job/tts_inference_job/src/job_steps/job_args.rs @@ -10,7 +10,6 @@ use jobs_common::semi_persistent_cache_dir::SemiPersistentCacheDir; use memory_caching::multi_item_ttl_cache::MultiItemTtlCache; use mysql_queries::mediators::firehose_publisher::FirehosePublisher; use mysql_queries::queries::tts::tts_models::get_tts_model_for_inference::TtsModelForInferenceRecord; -use newrelic_telemetry::Client as NewRelicClient; use crate::caching::cache_miss_strategizer_multi::SyncMultiCacheMissStrategizer; use crate::caching::virtual_lfu_cache::SyncVirtualLfuCache; @@ -38,10 +37,6 @@ pub struct JobArgs { pub job_stats: JobStats, - pub newrelic_client: NewRelicClient, - - pub newrelic_disabled: bool, - pub worker_details: JobWorkerDetails, // Keep tabs of which models to hold in the sidecar memory with this virtual LRU cache diff --git a/crates/service/job/tts_inference_job/src/job_steps/process_single_job.rs b/crates/service/job/tts_inference_job/src/job_steps/process_single_job.rs index bb9cd522a8..1767a674e2 100644 --- a/crates/service/job/tts_inference_job/src/job_steps/process_single_job.rs +++ b/crates/service/job/tts_inference_job/src/job_steps/process_single_job.rs @@ -6,14 +6,13 @@ use std::fs::File; use std::io::Read; use std::path::PathBuf; -use std::time::{Instant, SystemTime, UNIX_EPOCH}; +use std::time::{SystemTime, UNIX_EPOCH}; use anyhow::anyhow; use log::{error, info, warn}; use tempdir::TempDir; use container_common::anyhow_result::AnyhowResult; -use container_common::token::random_uuid::generate_random_uuid; use filesys::check_file_exists::check_file_exists; use filesys::safe_delete_temp_directory::safe_delete_temp_directory; use filesys::safe_delete_temp_file::safe_delete_temp_file; @@ -24,7 +23,6 @@ use mysql_queries::queries::tts::tts_inference_jobs::mark_tts_inference_job_done use mysql_queries::queries::tts::tts_inference_jobs::mark_tts_inference_job_pending_and_grab_lock::mark_tts_inference_job_pending_and_grab_lock; use mysql_queries::queries::tts::tts_models::get_tts_model_for_inference::{get_tts_model_for_inference, TtsModelForInferenceRecord}; use mysql_queries::queries::tts::tts_results::insert_tts_result::{insert_tts_result, JobType}; -use newrelic_telemetry::Span; use tts_common::clean_symbols::clean_symbols; use tts_common::text_pipelines::guess_pipeline::guess_text_pipeline_heuristic; use tts_common::text_pipelines::text_pipeline_type::TextPipelineType; @@ -56,34 +54,13 @@ pub async fn process_single_job( job_args: &JobArgs, job: &AvailableTtsInferenceJob, cached_model_record: &TtsModelForInferenceRecord, -) -> Result<(Span, Span), ProcessSingleJobError> { +) -> Result<(), ProcessSingleJobError> { // NB: Hack to allow cache bypass for debug requests // This will let us iterate faster with model changes. let mut maybe_queried_model = None; let mut model_record = cached_model_record; - let start = Instant::now(); - - let span_id = generate_random_uuid(); - let trace_id = generate_random_uuid(); - let maybe_user_token = job.maybe_creator_user_token.as_deref().unwrap_or(""); - - let mut job_iteration_span = Span::new(&span_id, &trace_id, get_timestamp_millis()) - .name("single job execution") - .attribute("user_token", maybe_user_token) - .service_name("tts-inference-job"); - - let span_id = generate_random_uuid(); - let trace_id = generate_random_uuid(); - - let created_at_timestamp = (job.created_at.timestamp() as u64) * 1000; - - let mut since_creation_span = Span::new(&span_id, &trace_id, created_at_timestamp) - .name("job since creation") - .attribute("user_token", maybe_user_token) - .service_name("tts-inference-job"); - let mut job_progress_reporter = job_args .job_progress_reporter .new_tts_inference(&job.inference_job_token) @@ -100,15 +77,7 @@ pub async fn process_single_job( if !lock_acquired { warn!("Could not acquire job lock for: {:?}", &job.id); - let duration = start.elapsed(); - - since_creation_span.set_attribute("status", "failure"); - since_creation_span.set_duration(duration); - - job_iteration_span.set_attribute("status", "failure"); - job_iteration_span.set_duration(duration); - - return Ok((since_creation_span, job_iteration_span)); + return Ok(()); } info!("Lock acquired for job: {}", job.inference_job_token); @@ -448,15 +417,7 @@ pub async fn process_single_job( info!("Job {:?} complete success! Downloaded, ran inference, and uploaded. Saved model record: {}, Result Token: {}", job.id, id, &inference_result_token); - let duration = start.elapsed(); - - since_creation_span.set_attribute("status", "success"); - since_creation_span.set_duration(duration); - - job_iteration_span.set_attribute("status", "success"); - job_iteration_span.set_duration(duration); - - Ok((since_creation_span, job_iteration_span)) + Ok(()) } fn get_timestamp_millis() -> u64 { diff --git a/crates/service/job/tts_inference_job/src/main.rs b/crates/service/job/tts_inference_job/src/main.rs index 9738caa13c..db471482ac 100644 --- a/crates/service/job/tts_inference_job/src/main.rs +++ b/crates/service/job/tts_inference_job/src/main.rs @@ -45,8 +45,6 @@ use mysql_queries::queries::tts::tts_inference_jobs::list_available_tts_inferenc use mysql_queries::queries::tts::tts_inference_jobs::mark_tts_inference_job_failure::mark_tts_inference_job_failure; use mysql_queries::queries::tts::tts_inference_jobs::mark_tts_inference_job_permanently_dead::mark_tts_inference_job_permanently_dead; use mysql_queries::queries::tts::tts_models::get_tts_model_for_inference::{get_tts_model_for_inference, TtsModelForInferenceError, TtsModelForInferenceRecord}; -use newrelic_telemetry::ClientBuilder; -use newrelic_telemetry::Span; use crate::caching::cache_miss_strategizer::CacheMissStrategizer; use crate::caching::cache_miss_strategizer::CacheMissStrategy; @@ -253,12 +251,6 @@ async fn main() -> AnyhowResult<()> { easyenv::get_env_num("TTS_MODEL_RECORD_CACHE_MILLIS", 300_000)?, // Five minutes ); - let license_key = easyenv::get_env_string_required("NEWRELIC_API_KEY")?; - - let newrelic_disabled = easyenv::get_env_bool_or_default("IS_NEWRELIC_DISABLED", false); - - let newrelic_client = ClientBuilder::new(&license_key).build().unwrap(); - let maybe_minimum_priority = easyenv::get_env_string_optional("MAYBE_MINIMUM_PRIORITY") .map(|priority_string| { priority_string.parse::() @@ -301,8 +293,6 @@ async fn main() -> AnyhowResult<()> { tts_sidecar_health_check_client, }, job_stats: JobStats::new(), - newrelic_client, - newrelic_disabled, worker_details: JobWorkerDetails { is_on_prem, worker_hostname: server_hostname.clone(), @@ -344,8 +334,6 @@ async fn main_loop(job_args: JobArgs) { let mut noop_logger = NoOpLogger::new(job_args.no_op_logger_millis as i64); - let mut span_batch = Vec::new(); - let mut sort_by_priority = true; let mut sort_by_priority_count = 0; @@ -411,23 +399,11 @@ async fn main_loop(job_args: JobArgs) { needs_health_check_at_start = false; } - let mut spans = match batch_result { - Ok(spans) => spans, - Err(e) => { - warn!("Error running job batch: {:?}", e); - std::thread::sleep(Duration::from_millis(error_timeout_millis)); - error_timeout_millis += INCREASE_TIMEOUT_MILLIS; - continue; - } - }; - - if !job_args.newrelic_disabled { - span_batch.append(&mut spans); - - if span_batch.len() > 50 { - let spans_to_send = span_batch.split_off(0).into(); - job_args.newrelic_client.send_spans(spans_to_send).await; - } + if let Err(e) = batch_result { + warn!("Error running job batch: {:?}", e); + std::thread::sleep(Duration::from_millis(error_timeout_millis)); + error_timeout_millis += INCREASE_TIMEOUT_MILLIS; + continue; } error_timeout_millis = START_TIMEOUT_MILLIS; // reset @@ -441,7 +417,7 @@ async fn process_jobs( inferencer: &JobArgs, jobs: Vec, needs_health_check_at_start: bool, -) -> AnyhowResult> { +) -> AnyhowResult<()> { if needs_health_check_at_start { maybe_block_on_sidecar_health_check(&inferencer.http_clients.tts_sidecar_health_check_client).await; @@ -449,8 +425,6 @@ async fn process_jobs( let mut maybe_sidecar_health_issue = false; - let mut batch_spans = Vec::new(); - for job in jobs.into_iter() { let model_state_result = ModelState::query_model_and_check_filesystem( &job, @@ -555,44 +529,38 @@ async fn process_jobs( } let result = process_single_job(inferencer, &job, &model_state.model_record).await; - match result { - Ok((span1, span2)) => { - batch_spans.push(span1); - batch_spans.push(span2); - }, - Err(e) => { - warn!("Failure to process job: {:?}", e); + if let Err(e) = result { + warn!("Failure to process job: {:?}", e); - record_failure_and_maybe_slow_down(&inferencer.job_stats); + record_failure_and_maybe_slow_down(&inferencer.job_stats); - maybe_sidecar_health_issue = true; + maybe_sidecar_health_issue = true; - let failure_reason = "failure processing job"; - let internal_debugging_failure_reason = format!("job error: {:?}", e); + let failure_reason = "failure processing job"; + let internal_debugging_failure_reason = format!("job error: {:?}", e); - let _r = mark_tts_inference_job_failure( - &inferencer.mysql_pool, - &job, - failure_reason, - &internal_debugging_failure_reason, - inferencer.job_max_attempts, - &inferencer.get_worker_name(), - ).await; + let _r = mark_tts_inference_job_failure( + &inferencer.mysql_pool, + &job, + failure_reason, + &internal_debugging_failure_reason, + inferencer.job_max_attempts, + &inferencer.get_worker_name(), + ).await; - match e { - ProcessSingleJobError::Other(_) => {} // No-op - ProcessSingleJobError::FilesystemFull => { - // TODO: Refactor - we should stop processing all of these jobs as we'll lose out - // on this entire batch by attempting to clear the filesystem. This should be handled - // in the calling code. - delete_tts_synthesizers_from_cache(&inferencer.semi_persistent_cache)?; - } + match e { + ProcessSingleJobError::Other(_) => {} // No-op + ProcessSingleJobError::FilesystemFull => { + // TODO: Refactor - we should stop processing all of these jobs as we'll lose out + // on this entire batch by attempting to clear the filesystem. This should be handled + // in the calling code. + delete_tts_synthesizers_from_cache(&inferencer.semi_persistent_cache)?; } } } } - Ok(batch_spans) + Ok(()) } fn record_failure_and_maybe_slow_down(job_stats: &JobStats) {