diff --git a/backend/.sqlx/query-1b56a720d99a689e80d12ee1efbfeb71d7cd7bb17e936746749c958062cdff9e.json b/backend/.sqlx/query-1b56a720d99a689e80d12ee1efbfeb71d7cd7bb17e936746749c958062cdff9e.json new file mode 100644 index 0000000000..f9cb3add9e --- /dev/null +++ b/backend/.sqlx/query-1b56a720d99a689e80d12ee1efbfeb71d7cd7bb17e936746749c958062cdff9e.json @@ -0,0 +1,26 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT tag AS \"tag!\", count(*) AS \"count!\" FROM v2_job_queue WHERE\n running = true\n GROUP BY tag", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "tag!", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "count!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false, + null + ] + }, + "hash": "1b56a720d99a689e80d12ee1efbfeb71d7cd7bb17e936746749c958062cdff9e" +} diff --git a/backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json b/backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json index 5783969e29..18260b8416 100644 --- a/backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json +++ b/backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json @@ -148,7 +148,8 @@ "oracledb", "nu", "java", - "duckdb" + "duckdb", + "ruby" ] } } diff --git a/backend/.sqlx/query-f22964772dc2d67aee437bbbd08b64792c00da1d713d7ca8f9904ccce7bfdae7.json b/backend/.sqlx/query-f22964772dc2d67aee437bbbd08b64792c00da1d713d7ca8f9904ccce7bfdae7.json deleted file mode 100644 index fb865ca894..0000000000 --- a/backend/.sqlx/query-f22964772dc2d67aee437bbbd08b64792c00da1d713d7ca8f9904ccce7bfdae7.json +++ /dev/null @@ -1,248 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT\n cj.id AS \"id!\",\n cj.workspace_id AS \"workspace_id!\",\n cj.parent_job,\n cj.created_by AS \"created_by!\",\n cj.duration_ms AS \"duration_ms!\",\n cj.success AS \"success!\",\n cj.script_hash AS \"script_hash!: Option\",\n cj.script_path,\n cj.args AS \"args: sqlx::types::Json>>\",\n cj.result AS \"result: sqlx::types::Json>\",\n cj.deleted AS \"deleted!\",\n cj.canceled AS \"canceled!\",\n cj.canceled_by,\n cj.canceled_reason,\n cj.job_kind AS \"job_kind!: JobKind\",\n cj.schedule_path,\n cj.permissioned_as AS \"permissioned_as!\",\n cj.is_flow_step AS \"is_flow_step!\",\n cj.language AS \"language: ScriptLang\",\n cj.is_skipped AS \"is_skipped!\",\n cj.email AS \"email!\",\n cj.visible_to_owner AS \"visible_to_owner!\",\n cj.mem_peak,\n cj.tag AS \"tag!\",\n cj.created_at AS \"created_at!\",\n cj.started_at,\n job_logs.logs,\n job_logs.log_offset AS \"log_offset?\",\n job_logs.log_file_index\n\n FROM v2_as_completed_job AS cj\n LEFT JOIN job_logs ON cj.id = job_logs.job_id\n WHERE cj.created_at < $1\n ORDER BY cj.created_at ASC LIMIT $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id!", - "type_info": "Uuid" - }, - { - "ordinal": 1, - "name": "workspace_id!", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "parent_job", - "type_info": "Uuid" - }, - { - "ordinal": 3, - "name": "created_by!", - "type_info": "Varchar" - }, - { - "ordinal": 4, - "name": "duration_ms!", - "type_info": "Int8" - }, - { - "ordinal": 5, - "name": "success!", - "type_info": "Bool" - }, - { - "ordinal": 6, - "name": "script_hash!: Option", - "type_info": "Int8" - }, - { - "ordinal": 7, - "name": "script_path", - "type_info": "Varchar" - }, - { - "ordinal": 8, - "name": "args: sqlx::types::Json>>", - "type_info": "Jsonb" - }, - { - "ordinal": 9, - "name": "result: sqlx::types::Json>", - "type_info": "Jsonb" - }, - { - "ordinal": 10, - "name": "deleted!", - "type_info": "Bool" - }, - { - "ordinal": 11, - "name": "canceled!", - "type_info": "Bool" - }, - { - "ordinal": 12, - "name": "canceled_by", - "type_info": "Varchar" - }, - { - "ordinal": 13, - "name": "canceled_reason", - "type_info": "Text" - }, - { - "ordinal": 14, - "name": "job_kind!: JobKind", - "type_info": { - "Custom": { - "name": "job_kind", - "kind": { - "Enum": [ - "script", - "preview", - "flow", - "dependencies", - "flowpreview", - "script_hub", - "identity", - "flowdependencies", - "http", - "graphql", - "postgresql", - "noop", - "appdependencies", - "deploymentcallback", - "singlescriptflow", - "flowscript", - "flownode", - "appscript" - ] - } - } - } - }, - { - "ordinal": 15, - "name": "schedule_path", - "type_info": "Varchar" - }, - { - "ordinal": 16, - "name": "permissioned_as!", - "type_info": "Varchar" - }, - { - "ordinal": 17, - "name": "is_flow_step!", - "type_info": "Bool" - }, - { - "ordinal": 18, - "name": "language: ScriptLang", - "type_info": { - "Custom": { - "name": "script_lang", - "kind": { - "Enum": [ - "python3", - "deno", - "go", - "bash", - "postgresql", - "nativets", - "bun", - "mysql", - "bigquery", - "snowflake", - "graphql", - "powershell", - "mssql", - "php", - "bunnative", - "rust", - "ansible", - "csharp", - "oracledb", - "nu", - "java", - "duckdb", - "ruby" - ] - } - } - } - }, - { - "ordinal": 19, - "name": "is_skipped!", - "type_info": "Bool" - }, - { - "ordinal": 20, - "name": "email!", - "type_info": "Varchar" - }, - { - "ordinal": 21, - "name": "visible_to_owner!", - "type_info": "Bool" - }, - { - "ordinal": 22, - "name": "mem_peak", - "type_info": "Int4" - }, - { - "ordinal": 23, - "name": "tag!", - "type_info": "Varchar" - }, - { - "ordinal": 24, - "name": "created_at!", - "type_info": "Timestamptz" - }, - { - "ordinal": 25, - "name": "started_at", - "type_info": "Timestamptz" - }, - { - "ordinal": 26, - "name": "logs", - "type_info": "Text" - }, - { - "ordinal": 27, - "name": "log_offset?", - "type_info": "Int4" - }, - { - "ordinal": 28, - "name": "log_file_index", - "type_info": "TextArray" - } - ], - "parameters": { - "Left": [ - "Timestamptz", - "Int8" - ] - }, - "nullable": [ - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - false, - true - ] - }, - "hash": "f22964772dc2d67aee437bbbd08b64792c00da1d713d7ca8f9904ccce7bfdae7" -} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 31f376d4b6..27a56d5613 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -114,6 +114,12 @@ lazy_static::lazy_static! { &["tag"] ).unwrap(); + static ref QUEUE_RUNNING_COUNT: prometheus::IntGaugeVec = prometheus::register_int_gauge_vec!( + "queue_running_count", + "Number of running jobs in the queue", + &["tag"] + ).unwrap(); + } lazy_static::lazy_static! { static ref ZOMBIE_JOB_TIMEOUT: String = std::env::var("ZOMBIE_JOB_TIMEOUT") @@ -140,6 +146,7 @@ lazy_static::lazy_static! { pub static ref WORKERS_NAMES: Arc>> = Arc::new(RwLock::new(Vec::new())); static ref QUEUE_COUNT_TAGS: Arc>> = Arc::new(RwLock::new(Vec::new())); + static ref QUEUE_RUNNING_COUNT_TAGS: Arc>> = Arc::new(RwLock::new(Vec::new())); static ref DISABLE_CONCURRENCY_LIMIT: bool = std::env::var("DISABLE_CONCURRENCY_LIMIT").is_ok_and(|s| s == "true"); static ref STALE_JOB_TRESHOLD_MINUTES: Option = std::env::var("STALE_JOB_TRESHOLD_MINUTES") @@ -1324,7 +1331,10 @@ pub async fn reload_url_list_setting( match url::Url::parse(url_str) { Ok(url) => urls.push(url), Err(e) => { - return Err(error::Error::BadRequest(format!("Invalid URL in FORCE_{}: '{}': {}", std_env_var, url_str, e))); + return Err(error::Error::BadRequest(format!( + "Invalid URL in FORCE_{}: '{}': {}", + std_env_var, url_str, e + ))); } } } @@ -1347,7 +1357,11 @@ pub async fn reload_url_list_setting( } } } - if urls.is_empty() { None } else { Some(urls) } + if urls.is_empty() { + None + } else { + Some(urls) + } } else { None }; @@ -1365,7 +1379,11 @@ pub async fn reload_url_list_setting( } } } - tracing::info!("Loaded setting {} from db config: {} URLs", setting_name, urls.len()); + tracing::info!( + "Loaded setting {} from db config: {} URLs", + setting_name, + urls.len() + ); value = if urls.is_empty() { None } else { Some(urls) }; } else { tracing::error!("Could not parse {} found: {:#?}", setting_name, &q); @@ -1671,6 +1689,30 @@ pub async fn expose_queue_metrics(db: &Pool) { let mut w = QUEUE_COUNT_TAGS.write().await; *w = tags_to_watch; } + + #[cfg(feature = "prometheus")] + if metrics_enabled { + // Handle queue running count metrics + let queue_running_counts = windmill_common::queue::get_queue_running_counts(db).await; + + for q in QUEUE_RUNNING_COUNT_TAGS.read().await.iter() { + if queue_running_counts.get(q).is_none() { + (*QUEUE_RUNNING_COUNT).with_label_values(&[q]).set(0); + } + } + + let mut running_tags_to_watch = vec![]; + for q in queue_running_counts { + let count = q.1; + let tag = q.0; + + let metric = (*QUEUE_RUNNING_COUNT).with_label_values(&[&tag]); + metric.set(count as i64); + running_tags_to_watch.push(tag.to_string()); + } + let mut w = QUEUE_RUNNING_COUNT_TAGS.write().await; + *w = running_tags_to_watch; + } } // clean queue metrics older than 14 days diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index b662a9f1d0..86c0601bc7 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -11502,6 +11502,22 @@ paths: additionalProperties: type: integer + /workers/queue_running_counts: + get: + summary: get counts of currently running jobs per tag + operationId: getCountsOfRunningJobsPerTag + tags: + - worker + responses: + "200": + description: queue running counts + content: + application/json: + schema: + type: object + additionalProperties: + type: integer + /configs/list_worker_groups: get: summary: list worker groups diff --git a/backend/windmill-api/src/workers.rs b/backend/windmill-api/src/workers.rs index 8e6b7a86fd..1165a5e2f3 100644 --- a/backend/windmill-api/src/workers.rs +++ b/backend/windmill-api/src/workers.rs @@ -37,6 +37,7 @@ pub fn global_service() -> Router { .route("/get_default_tags", get(get_default_tags)) .route("/queue_metrics", get(get_queue_metrics)) .route("/queue_counts", get(get_queue_counts)) + .route("/queue_running_counts", get(get_queue_running_counts)) } #[derive(FromRow, Serialize, Deserialize)] @@ -219,3 +220,12 @@ async fn get_queue_counts( let queue_counts = windmill_common::queue::get_queue_counts(&db).await; Ok(Json(queue_counts)) } + +async fn get_queue_running_counts( + authed: ApiAuthed, + Extension(db): Extension, +) -> JsonResult> { + require_super_admin(&db, &authed.email).await?; + let queue_running_counts = windmill_common::queue::get_queue_running_counts(&db).await; + Ok(Json(queue_running_counts)) +} diff --git a/backend/windmill-common/src/queue.rs b/backend/windmill-common/src/queue.rs index 39ec349c91..2656ce5af5 100644 --- a/backend/windmill-common/src/queue.rs +++ b/backend/windmill-common/src/queue.rs @@ -14,3 +14,16 @@ pub async fn get_queue_counts(db: &Pool) -> HashMap { .map(|v| v.into_iter().map(|x| (x.tag, x.count as u32)).collect()) .unwrap_or_else(|| HashMap::new()) } + +pub async fn get_queue_running_counts(db: &Pool) -> HashMap { + sqlx::query!( + "SELECT tag AS \"tag!\", count(*) AS \"count!\" FROM v2_job_queue WHERE + running = true + GROUP BY tag", + ) + .fetch_all(db) + .await + .ok() + .map(|v| v.into_iter().map(|x| (x.tag, x.count as u32)).collect()) + .unwrap_or_else(|| HashMap::new()) +}