From bbb6a15da7fae5cdb04fd9dd9ed780d8dbff5e4b Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sun, 9 Apr 2023 15:39:25 +0200 Subject: [PATCH] only register prometheus metrics if they are enabled --- backend/src/main.rs | 12 +---- backend/windmill-api/src/webhook_util.rs | 5 +- backend/windmill-common/src/lib.rs | 11 +++++ backend/windmill-queue/src/jobs.rs | 11 +++-- backend/windmill-worker/src/jobs.rs | 8 +++- backend/windmill-worker/src/worker.rs | 61 ++++++++++++++---------- 6 files changed, 66 insertions(+), 42 deletions(-) diff --git a/backend/src/main.rs b/backend/src/main.rs index d1890ebf3c..412cc7dbbc 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -10,7 +10,7 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use git_version::git_version; use sqlx::{Pool, Postgres}; -use windmill_common::utils::rd_string; +use windmill_common::{utils::rd_string, METRICS_ADDR}; const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version"); const DEFAULT_NUM_WORKERS: usize = 3; @@ -30,15 +30,7 @@ async fn main() -> anyhow::Result<()> { .and_then(|x| x.parse::().ok()) .unwrap_or(DEFAULT_NUM_WORKERS as i32); - let metrics_addr: Option = std::env::var("METRICS_ADDR") - .ok() - .map(|s| { - s.parse::() - .map(|b| b.then(|| SocketAddr::from(([0, 0, 0, 0], 8001)))) - .or_else(|_| s.parse::().map(Some)) - }) - .transpose()? - .flatten(); + let metrics_addr: Option = *METRICS_ADDR; let server_bind_address: IpAddr = std::env::var("SERVER_BIND_ADDR") .ok() diff --git a/backend/windmill-api/src/webhook_util.rs b/backend/windmill-api/src/webhook_util.rs index 72f72c3dae..e4c76b17c7 100644 --- a/backend/windmill-api/src/webhook_util.rs +++ b/backend/windmill-api/src/webhook_util.rs @@ -2,6 +2,7 @@ use std::time::Duration; use serde::Serialize; use tokio::{select, sync::mpsc, time::interval}; +use windmill_common::METRICS_ENABLED; use crate::db::DB; @@ -89,9 +90,9 @@ impl WebhookShared { }; let webook_opt = url_guard.value(); if let Some(url) = webook_opt { - let timer = WEBHOOK_REQUEST_COUNT.start_timer(); + let timer = if *METRICS_ENABLED { Some(WEBHOOK_REQUEST_COUNT.start_timer()) } else { None }; let _ = client.post(url).json(&message).send().await; - timer.stop_and_record(); + timer.map(|x| x.stop_and_record()); drop(url_guard); } }, diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 4bd08f407c..a0411efd1b 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -30,6 +30,17 @@ pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50; pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 3; lazy_static::lazy_static! { + pub static ref METRICS_ADDR: Option = std::env::var("METRICS_ADDR") + .ok() + .map(|s| { + s.parse::() + .map(|b| b.then(|| SocketAddr::from(([0, 0, 0, 0], 8001)))) + .or_else(|_| s.parse::().map(Some)) + }) + .transpose().ok() + .flatten() + .flatten(); + pub static ref METRICS_ENABLED: bool = METRICS_ADDR.is_some(); pub static ref BASE_URL: String = std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost".to_string()); pub static ref IS_READY: Arc = Arc::new(std::sync::atomic::AtomicBool::new(false)); } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index fc232554d6..f69f3409af 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -22,6 +22,7 @@ use windmill_common::{ flows::{FlowModule, FlowModuleValue, FlowValue}, scripts::{get_full_hub_script_by_path, HubScript, ScriptHash, ScriptLang}, utils::StripPath, + METRICS_ENABLED, }; lazy_static::lazy_static! { @@ -146,7 +147,7 @@ pub async fn pull( .fetch_optional(db) .await?; - if job.is_some() { + if job.is_some() && *METRICS_ENABLED { QUEUE_PULL_COUNT.inc(); } @@ -211,7 +212,9 @@ pub async fn delete_job( w_id: &str, job_id: Uuid, ) -> windmill_common::error::Result<()> { - QUEUE_DELETE_COUNT.inc(); + if *METRICS_ENABLED { + QUEUE_DELETE_COUNT.inc(); + } let job_removed = sqlx::query_scalar!( "DELETE FROM queue WHERE workspace_id = $1 AND id = $2 RETURNING 1", w_id, @@ -542,7 +545,9 @@ pub async fn push<'c>( .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(); + if *METRICS_ENABLED { + QUEUE_PUSH_COUNT.inc(); + } { let uuid_string = job_id.to_string(); diff --git a/backend/windmill-worker/src/jobs.rs b/backend/windmill-worker/src/jobs.rs index 8c467969c2..24b444c619 100644 --- a/backend/windmill-worker/src/jobs.rs +++ b/backend/windmill-worker/src/jobs.rs @@ -9,7 +9,9 @@ use sqlx::{Pool, Postgres, Transaction}; use tracing::instrument; use uuid::Uuid; -use windmill_common::{error::Error, flow_status::FlowStatusModule, schedule::Schedule}; +use windmill_common::{ + error::Error, flow_status::FlowStatusModule, schedule::Schedule, METRICS_ENABLED, +}; use windmill_queue::{delete_job, schedule::get_schedule_opt, JobKind, QueuedJob, CLOUD_HOSTED}; #[instrument(level = "trace", skip_all)] @@ -20,7 +22,9 @@ pub async fn add_completed_job_error( e: serde_json::Value, metrics: Option, ) -> Result { - metrics.map(|m| m.worker_execution_failed.inc()); + if *METRICS_ENABLED { + metrics.map(|m| m.worker_execution_failed.inc()); + } let result = serde_json::json!({ "error": e }); let _ = add_completed_job(db, &queued_job, false, false, result.clone(), logs).await?; Ok(result) diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 16c2bf5991..2c1bfda8db 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -27,7 +27,7 @@ use windmill_common::{ flows::{FlowModuleValue, FlowValue}, scripts::{ScriptHash, ScriptLang}, utils::{rd_string, calculate_hash}, - variables, BASE_URL, users::SUPERADMIN_SECRET_EMAIL, IS_READY, + variables, BASE_URL, users::SUPERADMIN_SECRET_EMAIL, IS_READY, METRICS_ENABLED, }; use windmill_queue::{canceled_job_to_result, get_queued_job, pull, JobKind, QueuedJob, CLOUD_HOSTED}; @@ -610,8 +610,9 @@ pub async fn run_worker( let mut jobs_executed = 0; - - WORKER_STARTED.inc(); + if *METRICS_ENABLED { + WORKER_STARTED.inc(); + } let (_copy_bucket_tx, mut _copy_bucket_rx) = mpsc::channel::<()>(2); @@ -644,13 +645,14 @@ pub async fn run_worker( loop { - worker_busy.set(0); - - uptime_metric.inc_by( - (((Instant::now() - start_time).as_millis() as f64)/1000.0 - uptime_metric.get()) - .try_into() - .unwrap(), - ); + if *METRICS_ENABLED { + worker_busy.set(0); + uptime_metric.inc_by( + (((Instant::now() - start_time).as_millis() as f64)/1000.0 - uptime_metric.get()) + .try_into() + .unwrap(), + ); + } let do_break = async { @@ -707,10 +709,12 @@ pub async fn run_worker( .map_err(|_| Error::InternalErr("Impossible to fetch same_worker job".to_string()))) }, (job, timer) = { - let timer = worker_pull_duration.start_timer(); + let timer = if *METRICS_ENABLED { Some(worker_pull_duration.start_timer()) } else { None }; pull(&db, WHITELIST_WORKSPACES.clone(), BLACKLIST_WORKSPACES.clone()).map(|x| (x, timer)) } => { - let duration_pull_s = timer.stop_and_record(); - worker_pull_duration_counter.inc_by(duration_pull_s); + timer.map(|timer| { + let duration_pull_s = timer.stop_and_record(); + worker_pull_duration_counter.inc_by(duration_pull_s); + }); (false, job) }, } @@ -718,8 +722,9 @@ pub async fn run_worker( if do_break { return true; } - - worker_busy.set(1); + if *METRICS_ENABLED { + worker_busy.set(1); + } match next_job { Ok(Some(job)) => { let label_values = [ @@ -732,9 +737,11 @@ pub async fn run_worker( .start_timer(); jobs_executed += 1; - worker_execution_count - .with_label_values(label_values.as_slice()) - .inc(); + if *METRICS_ENABLED { + worker_execution_count + .with_label_values(label_values.as_slice()) + .inc(); + } let metrics = Metrics { worker_execution_failed: worker_execution_failed @@ -825,12 +832,12 @@ pub async fn run_worker( } } Ok(None) => { - - let _timer = worker_sleep_duration - .start_timer(); + let _timer = if *METRICS_ENABLED { Some(worker_sleep_duration.start_timer()) } else { None }; tokio::time::sleep(Duration::from_millis(*SLEEP_QUEUE)).await; - let duration = _timer.stop_and_record(); - worker_sleep_duration_counter.inc_by(duration); + _timer.map(|timer| { + let duration = timer.stop_and_record(); + worker_sleep_duration_counter.inc_by(duration); + }); } Err(err) => { tracing::error!(worker = %worker_name, "run_worker: pulling jobs: {}", err); @@ -2915,7 +2922,9 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str) { .ok() .unwrap_or_else(|| vec![]); - QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _); + if *METRICS_ENABLED { + QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _); + } for r in restarted { tracing::info!( "restarted zombie job {} {} {}", @@ -2940,7 +2949,9 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str) { .ok() .unwrap_or_else(|| vec![]); - QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _); + if *METRICS_ENABLED { + QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _); + } for job in timeouts { tracing::info!( "timedout zombie job {} {}",