From 55d3f38b8b2db0a76bda3c267c3107c7ae3208d3 Mon Sep 17 00:00:00 2001 From: Kai Jellinghaus Date: Wed, 19 Oct 2022 10:15:42 +0200 Subject: [PATCH] More Metrics (#735) * Expose Worker Tokio Metrics * Add metrics tracking the job queue This can be used to (approximate) the queue length This can be used to (approximate) the running jobs * Remove testing code * Rename metrics to exclude TOTAL_ * Sleep in worker metrics loop * Add jobs_executed * Add zombie job metrics * Add uptime metric * Fix metric naming * Remove poll stats * unify worker execution metrics * Rename jobs_* metrics * Rename metrics to match exposed name * Remove leftover import * Rename variables to match further * Fix Merge error Co-authored-by: Ruben Fiszel --- backend/Cargo.lock | 13 ++++ backend/Cargo.toml | 2 + backend/src/jobs.rs | 29 +++++++- backend/src/lib.rs | 8 ++- backend/src/worker.rs | 164 ++++++++++++++++++++++++++---------------- 5 files changed, 152 insertions(+), 64 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 8212741e15..a52c55ef3e 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -3546,6 +3546,17 @@ dependencies = [ "syn", ] +[[package]] +name = "tokio-metrics" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bcb585a0069b53171684e22d5255984ec30d1c7304fd0a4a9a603ffd8c765cdd" +dependencies = [ + "futures-util", + "pin-project-lite", + "tokio", +] + [[package]] name = "tokio-native-tls" version = "0.3.0" @@ -4255,6 +4266,7 @@ dependencies = [ "hyper", "itertools 0.10.5", "json-pointer", + "lazy_static", "lettre", "magic-crypt", "mime_guess", @@ -4280,6 +4292,7 @@ dependencies = [ "thiserror", "time 0.3.15", "tokio", + "tokio-metrics", "tokio-tar", "tokio-util", "tower", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 160579534c..bb6f757a4e 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -66,3 +66,5 @@ sqlx = { version = "^0", features = ["offline", "macros", "migrate", "uuid", "js dotenv = "^0" ulid = { version = "^1", features = ["uuid"] } futures = "^0" +tokio-metrics = "0.1.0" +lazy_static = "1.4.0" diff --git a/backend/src/jobs.rs b/backend/src/jobs.rs index 9d63c8b8d1..847732a7c9 100644 --- a/backend/src/jobs.rs +++ b/backend/src/jobs.rs @@ -1315,6 +1315,25 @@ pub enum JobPayload { RawFlow { value: FlowValue, path: Option }, } +lazy_static::lazy_static! { + // TODO: these aren't synced, they should be moved into the queue abstraction once/if that happens. + static ref QUEUE_PUSH_COUNT: prometheus::IntCounter = prometheus::register_int_counter!( + "queue_push_count", + "Total number of jobs pushed to the queue." + ) + .unwrap(); + static ref QUEUE_DELETE_COUNT: prometheus::IntCounter = prometheus::register_int_counter!( + "queue_delete_count", + "Total number of jobs deleted from the queue." + ) + .unwrap(); + static ref QUEUE_PULL_COUNT: prometheus::IntCounter = prometheus::register_int_counter!( + "queue_pull_count", + "Total number of jobs pulled from the queue." + ) + .unwrap(); +} + #[instrument(level = "trace", skip_all)] pub async fn push<'c>( mut tx: Transaction<'c, Postgres>, @@ -1533,6 +1552,8 @@ pub async fn push<'c>( .fetch_one(&mut tx) .await .map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id}: {e}")))?; + // TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction. + QUEUE_PUSH_COUNT.inc(); { let uuid_string = job_id.to_string(); @@ -1575,7 +1596,7 @@ pub async fn add_completed_job_error( e: E, metrics: Option, ) -> Result<(Uuid, Map), Error> { - metrics.map(|m| m.jobs_failed.inc()); + metrics.map(|m| m.worker_execution_failed.inc()); let mut output_map = serde_json::Map::new(); output_map.insert( "error".to_string(), @@ -1734,11 +1755,17 @@ pub async fn pull(db: &DB) -> Result, crate::Error> { ) .fetch_optional(db) .await?; + + if job.is_some() { + QUEUE_PULL_COUNT.inc(); + } + Ok(job) } #[instrument(level = "trace", skip_all)] pub async fn delete_job(db: &DB, w_id: &str, job_id: Uuid) -> Result<(), crate::Error> { + QUEUE_DELETE_COUNT.inc(); let job_removed = sqlx::query_scalar!( "DELETE FROM queue WHERE workspace_id = $1 AND id = $2 RETURNING 1", w_id, diff --git a/backend/src/lib.rs b/backend/src/lib.rs index 12678c32e2..91a0962875 100644 --- a/backend/src/lib.rs +++ b/backend/src/lib.rs @@ -219,13 +219,14 @@ pub async fn run_workers( rx: tokio::sync::broadcast::Receiver<()>, ) -> anyhow::Result<()> { let instance_name = rd_string(5); + let monitor = tokio_metrics::TaskMonitor::new(); let ip = external_ip::get_ip().await.unwrap_or_else(|e| { tracing::warn!(error = e.to_string(), "failed to get external IP"); "unretrievable IP".to_string() }); - let mut handles = Vec::new(); + let mut handles = Vec::with_capacity(num_workers as usize); for i in 1..(num_workers + 1) { let db1 = db.clone(); @@ -234,7 +235,7 @@ pub async fn run_workers( let ip = ip.clone(); let rx = rx.resubscribe(); let worker_config = worker_config.clone(); - handles.push(tokio::spawn(async move { + handles.push(tokio::spawn(monitor.instrument(async move { tracing::info!(addr = %addr.to_string(), worker = %worker_name, "starting worker"); worker::run_worker( &db1, @@ -249,8 +250,9 @@ pub async fn run_workers( rx, ) .await - })); + }))); } + futures::future::try_join_all(handles).await?; Ok(()) } diff --git a/backend/src/worker.rs b/backend/src/worker.rs index cb4b45ecba..e337790690 100644 --- a/backend/src/worker.rs +++ b/backend/src/worker.rs @@ -86,7 +86,7 @@ const GO_REQ_SPLITTER: &str = "//go.sum"; #[derive(Clone)] pub struct Metrics { - pub jobs_failed: prometheus::IntCounter, + pub worker_execution_failed: prometheus::IntCounter, } #[derive(Clone)] @@ -98,6 +98,28 @@ pub struct WorkerConfig { pub keep_job_dir: bool, } +lazy_static::lazy_static! { + static ref WORKER_STARTED: prometheus::IntGauge = prometheus::register_int_gauge!( + "worker_started", + "Total number of workers started." + ) + .unwrap(); + static ref QUEUE_ZOMBIE_RESTART_COUNT: prometheus::IntCounter = prometheus::register_int_counter!( + "queue_zombie_restart_count", + "Total number of jobs restarted due to ping timeout." + ) + .unwrap(); + static ref QUEUE_ZOMBIE_DELETE_COUNT: prometheus::IntCounter = prometheus::register_int_counter!( + "queue_zombie_delete_count", + "Total number of jobs deleted due to their ping timing out in an unrecoverable state." + ) + .unwrap(); + static ref WORKER_UPTIME_OPTS: prometheus::Opts = prometheus::opts!( + "worker_uptime", + "Total number of milliseconds since the worker has started" + ); +} + pub async fn run_worker( db: &DB, timeout: i32, @@ -110,6 +132,8 @@ pub async fn run_worker( worker_config: WorkerConfig, mut rx: tokio::sync::broadcast::Receiver<()>, ) { + let start_time = Instant::now(); + let worker_dir = format!("{TMP_DIR}/{worker_name}"); tracing::debug!(worker_dir = %worker_dir, worker_name = %worker_name, "Creating worker dir"); @@ -138,23 +162,19 @@ pub async fn run_worker( insert_initial_ping(worker_instance, &worker_name, ip, db).await; - prometheus::register_int_gauge!(prometheus::Opts::new( - "start_time_seconds", - "Start time of worker as seconds since unix epoch", - ) - .const_label("name", &worker_name)) - .expect("register prometheus metric") - .set( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .ok() - .map(|duration| duration.as_secs() as i64) - .unwrap_or(0), + let uptime_metric = prometheus::register_int_counter!(WORKER_UPTIME_OPTS + .clone() + .const_label("name", &worker_name)) + .unwrap(); + uptime_metric.inc_by( + ((Instant::now() - start_time).as_millis() - uptime_metric.get() as u128) + .try_into() + .unwrap(), ); - let job_duration_seconds = prometheus::register_histogram_vec!( + let worker_execution_duration = prometheus::register_histogram_vec!( prometheus::HistogramOpts::new( - "job_duration_seconds", + "worker_execution_duration", "Duration between receiving a job and completing it", ) .const_label("name", &worker_name), @@ -162,8 +182,15 @@ pub async fn run_worker( ) .expect("register prometheus metric"); - let jobs_failed = prometheus::register_int_counter_vec!( - prometheus::Opts::new("jobs_failed", "Number of failed jobs",) + let worker_execution_failed = prometheus::register_int_counter_vec!( + prometheus::Opts::new("worker_execution_failed", "Number of failed jobs",) + .const_label("name", &worker_name), + &["workspace_id", "language"], + ) + .expect("register prometheus metric"); + + let worker_execution_count = prometheus::register_int_counter_vec!( + prometheus::Opts::new("worker_execution_count", "Number of executed jobs",) .const_label("name", &worker_name), &["workspace_id", "language"], ) @@ -198,10 +225,16 @@ pub async fn run_worker( pip_extra_index_url, pip_trusted_host, }; + WORKER_STARTED.inc(); let (same_worker_tx, mut same_worker_rx) = mpsc::channel::(5); loop { + uptime_metric.inc_by( + ((Instant::now() - start_time).as_millis() - uptime_metric.get() as u128) + .try_into() + .unwrap(), + ); if last_ping.elapsed().as_secs() > NUM_SECS_ENV_CHECK { sqlx::query!( "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1 WHERE worker = $2", @@ -238,14 +271,19 @@ pub async fn run_worker( job.language.as_ref().map(|l| l.as_str()).unwrap_or(""), ]; - let _timer = job_duration_seconds + let _timer = worker_execution_duration .with_label_values(label_values.as_slice()) .start_timer(); jobs_executed += 1; + worker_execution_count + .with_label_values(label_values.as_slice()) + .inc(); - let metrics = - Metrics { jobs_failed: jobs_failed.with_label_values(label_values.as_slice()) }; + let metrics = Metrics { + worker_execution_failed: worker_execution_failed + .with_label_values(label_values.as_slice()), + }; tracing::info!(worker = %worker_name, id = %job.id, "fetched job {}", job.id); @@ -1885,7 +1923,20 @@ pub async fn handle_zombie_jobs_periodically( mut rx: tokio::sync::broadcast::Receiver<()>, ) { loop { - let restarted = sqlx::query!( + handle_zombie_jobs(db, timeout).await; + + tokio::select! { + _ = tokio::time::sleep(Duration::from_secs(60)) => (), + _ = rx.recv() => { + println!("received killpill for monitor job"); + break; + } + } + } +} + +async fn handle_zombie_jobs(db: &DB, timeout: i32) { + let restarted = sqlx::query!( "UPDATE queue SET running = false WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND same_worker = false RETURNING id, workspace_id, last_ping", (timeout * 5).to_string(), JobKind::Flow: JobKind, @@ -1895,16 +1946,17 @@ pub async fn handle_zombie_jobs_periodically( .ok() .unwrap_or_else(|| vec![]); - for r in restarted { - tracing::info!( - "restarted zombie job {} {} {}", - r.id, - r.workspace_id, - r.last_ping - ); - } + QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _); + for r in restarted { + tracing::info!( + "restarted zombie job {} {} {}", + r.id, + r.workspace_id, + r.last_ping + ); + } - let timeouts = sqlx::query_as::<_, QueuedJob>( + let timeouts = sqlx::query_as::<_, QueuedJob>( "SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND same_worker = true", ) .bind((timeout * 5).to_string()) @@ -1914,38 +1966,30 @@ pub async fn handle_zombie_jobs_periodically( .ok() .unwrap_or_else(|| vec![]); - for job in timeouts { - tracing::info!( - "timedouts zombie same_worker job {} {}", - job.id, - job.workspace_id, - ); + QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _); + for job in timeouts { + tracing::info!( + "timedouts zombie same_worker job {} {}", + job.id, + job.workspace_id, + ); - // since the job is unrecoverable, the same worker queue should never be sent anything - let (same_worker_tx_never_used, _same_worker_rx_never_used) = mpsc::channel::(1); + // since the job is unrecoverable, the same worker queue should never be sent anything + let (same_worker_tx_never_used, _same_worker_rx_never_used) = mpsc::channel::(1); - let _ = handle_job_error( - db, - job, - error::Error::ExecutionErr("Same worker job timed out".to_string()), - None, - true, - same_worker_tx_never_used, - "", - true, - &std::env::var("BASE_INTERNAL_URL") - .unwrap_or_else(|_| "http://localhost:8000".to_string()), - ) - .await; - } - - tokio::select! { - _ = tokio::time::sleep(Duration::from_secs(60)) => (), - _ = rx.recv() => { - println!("received killpill for monitor job"); - break; - } - } + let _ = handle_job_error( + db, + job, + error::Error::ExecutionErr("Same worker job timed out".to_string()), + None, + true, + same_worker_tx_never_used, + "", + true, + &std::env::var("BASE_INTERNAL_URL") + .unwrap_or_else(|_| "http://localhost:8000".to_string()), + ) + .await; } }