From aaa4cde349376c04f4052b08abb9cb88004fe154 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 19 Mar 2025 09:55:53 +0100 Subject: [PATCH] feat: filter by worker + backend perf opt (#5489) * all * all * all * refactor * sqlx * ref * impr * fix * fix * fix * ee private --- ...da4cbe80b0150827c6e90f6e4f9e512587ba1.json | 14 - ...a6c391c4d0054a618b861464ee37239f1f1e0.json | 273 ++++++++++++ ...7bb9be8106a79e0683cd34459585bbd920ce4.json | 23 - ...5dbf541eb14c5c374656ffb96da7283a2a6f1.json | 23 - ...e5f4b72fc27578d988155b74b05ec5df30b9.json} | 6 +- ...665831ee3e4ec3815e9ad90e886ffecba0f1.json} | 7 +- backend/ee-repo-ref.txt | 2 +- backend/src/monitor.rs | 4 +- backend/windmill-api/openapi.yaml | 13 + backend/windmill-api/src/capture.rs | 34 +- .../windmill-api/src/concurrency_groups.rs | 1 + backend/windmill-api/src/jobs.rs | 177 ++++++-- backend/windmill-api/src/mqtt_triggers.rs | 4 +- .../src/postgres_triggers/trigger.rs | 4 +- .../windmill-api/src/websocket_triggers.rs | 4 +- backend/windmill-common/src/worker.rs | 26 +- backend/windmill-queue/src/jobs.rs | 416 +++++++++++++----- .../windmill-worker/src/ansible_executor.rs | 5 +- backend/windmill-worker/src/bash_executor.rs | 9 +- .../windmill-worker/src/bigquery_executor.rs | 5 +- backend/windmill-worker/src/bun_executor.rs | 17 +- backend/windmill-worker/src/common.rs | 47 +- .../windmill-worker/src/csharp_executor.rs | 7 +- .../windmill-worker/src/dedicated_worker.rs | 23 +- backend/windmill-worker/src/deno_executor.rs | 15 +- backend/windmill-worker/src/go_executor.rs | 5 +- .../windmill-worker/src/graphql_executor.rs | 5 +- backend/windmill-worker/src/lib.rs | 1 + backend/windmill-worker/src/mssql_executor.rs | 5 +- backend/windmill-worker/src/mysql_executor.rs | 4 +- .../windmill-worker/src/oracledb_executor.rs | 5 +- backend/windmill-worker/src/otel_ee.rs | 3 + backend/windmill-worker/src/pg_executor.rs | 6 +- backend/windmill-worker/src/php_executor.rs | 5 +- .../windmill-worker/src/python_executor.rs | 13 +- .../windmill-worker/src/result_processor.rs | 35 +- backend/windmill-worker/src/rust_executor.rs | 4 +- .../windmill-worker/src/snowflake_executor.rs | 5 +- backend/windmill-worker/src/worker.rs | 133 +++--- backend/windmill-worker/src/worker_flow.rs | 131 +++--- .../windmill-worker/src/worker_lockfiles.rs | 53 ++- .../src/lib/components/runs/JobLoader.svelte | 3 + .../src/lib/components/runs/JobPreview.svelte | 21 +- .../src/lib/components/runs/RunsFilter.svelte | 83 +++- .../src/lib/components/runs/RunsTable.svelte | 1 + .../(root)/(logged)/run/[...run]/+page.svelte | 15 + .../(logged)/runs/[...path]/+page.svelte | 33 ++ 47 files changed, 1212 insertions(+), 516 deletions(-) delete mode 100644 backend/.sqlx/query-0aa8f50fe377a23e2ae3821fd6eda4cbe80b0150827c6e90f6e4f9e512587ba1.json create mode 100644 backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json delete mode 100644 backend/.sqlx/query-5ff7df54c7908a7de494ddae5fc7bb9be8106a79e0683cd34459585bbd920ce4.json delete mode 100644 backend/.sqlx/query-a3ccf362b4f6df400b3c7a084795dbf541eb14c5c374656ffb96da7283a2a6f1.json rename backend/.sqlx/{query-4535c8effd1bae49894d13293a37e1ee949cf9108239032cb3addbf350fb33de.json => query-c50b6a4a6739d6df087a3b37c209e5f4b72fc27578d988155b74b05ec5df30b9.json} (73%) rename backend/.sqlx/{query-2f2ef9b1ccff527c48fa01cf1b78cd0e58c8d534ac22ec0356d82a854b31d087.json => query-ec1f31fd7628ea2e30995a0de1d8665831ee3e4ec3815e9ad90e886ffecba0f1.json} (57%) create mode 100644 backend/windmill-worker/src/otel_ee.rs diff --git a/backend/.sqlx/query-0aa8f50fe377a23e2ae3821fd6eda4cbe80b0150827c6e90f6e4f9e512587ba1.json b/backend/.sqlx/query-0aa8f50fe377a23e2ae3821fd6eda4cbe80b0150827c6e90f6e4f9e512587ba1.json deleted file mode 100644 index d48b865a92..0000000000 --- a/backend/.sqlx/query-0aa8f50fe377a23e2ae3821fd6eda4cbe80b0150827c6e90f6e4f9e512587ba1.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE v2_job_queue SET started_at = NOW() WHERE id = $1", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "0aa8f50fe377a23e2ae3821fd6eda4cbe80b0150827c6e90f6e4f9e512587ba1" -} diff --git a/backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json b/backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json new file mode 100644 index 0000000000..1ae47774dd --- /dev/null +++ b/backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json @@ -0,0 +1,273 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT \n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n scheduled_for,\n runnable_path,\n kind as \"kind: JobKind\",\n runnable_id as \"runnable_id: ScriptHash\",\n canceled_reason,\n canceled_by,\n permissioned_as,\n permissioned_as_email,\n flow_status as \"flow_status: sqlx::types::Json>\",\n v2_job.tag,\n script_lang as \"script_lang: ScriptLang\",\n same_worker,\n pre_run_error,\n concurrent_limit,\n concurrency_time_window_s,\n flow_innermost_root_job,\n timeout,\n flow_step_id,\n cache_ttl,\n v2_job_queue.priority,\n preprocessed,\n script_entrypoint_override,\n trigger,\n trigger_kind as \"trigger_kind: JobTriggerKind\",\n visible_to_owner\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 2, + "name": "args: sqlx::types::Json>>", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "parent_job", + "type_info": "Uuid" + }, + { + "ordinal": 4, + "name": "created_by", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "started_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 6, + "name": "scheduled_for", + "type_info": "Timestamptz" + }, + { + "ordinal": 7, + "name": "runnable_path", + "type_info": "Varchar" + }, + { + "ordinal": 8, + "name": "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": 9, + "name": "runnable_id: ScriptHash", + "type_info": "Int8" + }, + { + "ordinal": 10, + "name": "canceled_reason", + "type_info": "Text" + }, + { + "ordinal": 11, + "name": "canceled_by", + "type_info": "Varchar" + }, + { + "ordinal": 12, + "name": "permissioned_as", + "type_info": "Varchar" + }, + { + "ordinal": 13, + "name": "permissioned_as_email", + "type_info": "Varchar" + }, + { + "ordinal": 14, + "name": "flow_status: sqlx::types::Json>", + "type_info": "Jsonb" + }, + { + "ordinal": 15, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 16, + "name": "script_lang: 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" + ] + } + } + } + }, + { + "ordinal": 17, + "name": "same_worker", + "type_info": "Bool" + }, + { + "ordinal": 18, + "name": "pre_run_error", + "type_info": "Text" + }, + { + "ordinal": 19, + "name": "concurrent_limit", + "type_info": "Int4" + }, + { + "ordinal": 20, + "name": "concurrency_time_window_s", + "type_info": "Int4" + }, + { + "ordinal": 21, + "name": "flow_innermost_root_job", + "type_info": "Uuid" + }, + { + "ordinal": 22, + "name": "timeout", + "type_info": "Int4" + }, + { + "ordinal": 23, + "name": "flow_step_id", + "type_info": "Varchar" + }, + { + "ordinal": 24, + "name": "cache_ttl", + "type_info": "Int4" + }, + { + "ordinal": 25, + "name": "priority", + "type_info": "Int2" + }, + { + "ordinal": 26, + "name": "preprocessed", + "type_info": "Bool" + }, + { + "ordinal": 27, + "name": "script_entrypoint_override", + "type_info": "Varchar" + }, + { + "ordinal": 28, + "name": "trigger", + "type_info": "Varchar" + }, + { + "ordinal": 29, + "name": "trigger_kind: JobTriggerKind", + "type_info": { + "Custom": { + "name": "job_trigger_kind", + "kind": { + "Enum": [ + "webhook", + "http", + "websocket", + "kafka", + "email", + "nats", + "schedule", + "app", + "ui", + "postgres" + ] + } + } + } + }, + { + "ordinal": 30, + "name": "visible_to_owner", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false, + false, + true, + true, + false, + true, + false, + true, + false, + true, + true, + true, + false, + false, + true, + false, + true, + false, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + false + ] + }, + "hash": "4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0" +} diff --git a/backend/.sqlx/query-5ff7df54c7908a7de494ddae5fc7bb9be8106a79e0683cd34459585bbd920ce4.json b/backend/.sqlx/query-5ff7df54c7908a7de494ddae5fc7bb9be8106a79e0683cd34459585bbd920ce4.json deleted file mode 100644 index 45bca31fd0..0000000000 --- a/backend/.sqlx/query-5ff7df54c7908a7de494ddae5fc7bb9be8106a79e0683cd34459585bbd920ce4.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT flow_version.value->>'concurrency_key'\n FROM flow \n LEFT JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 AND flow.workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "?column?", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "5ff7df54c7908a7de494ddae5fc7bb9be8106a79e0683cd34459585bbd920ce4" -} diff --git a/backend/.sqlx/query-a3ccf362b4f6df400b3c7a084795dbf541eb14c5c374656ffb96da7283a2a6f1.json b/backend/.sqlx/query-a3ccf362b4f6df400b3c7a084795dbf541eb14c5c374656ffb96da7283a2a6f1.json deleted file mode 100644 index bc5e2351f9..0000000000 --- a/backend/.sqlx/query-a3ccf362b4f6df400b3c7a084795dbf541eb14c5c374656ffb96da7283a2a6f1.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT concurrency_key FROM script WHERE hash = $1 AND workspace_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "concurrency_key", - "type_info": "Varchar" - } - ], - "parameters": { - "Left": [ - "Int8", - "Text" - ] - }, - "nullable": [ - true - ] - }, - "hash": "a3ccf362b4f6df400b3c7a084795dbf541eb14c5c374656ffb96da7283a2a6f1" -} diff --git a/backend/.sqlx/query-4535c8effd1bae49894d13293a37e1ee949cf9108239032cb3addbf350fb33de.json b/backend/.sqlx/query-c50b6a4a6739d6df087a3b37c209e5f4b72fc27578d988155b74b05ec5df30b9.json similarity index 73% rename from backend/.sqlx/query-4535c8effd1bae49894d13293a37e1ee949cf9108239032cb3addbf350fb33de.json rename to backend/.sqlx/query-c50b6a4a6739d6df087a3b37c209e5f4b72fc27578d988155b74b05ec5df30b9.json index a500b04785..eef812552b 100644 --- a/backend/.sqlx/query-4535c8effd1bae49894d13293a37e1ee949cf9108239032cb3addbf350fb33de.json +++ b/backend/.sqlx/query-c50b6a4a6739d6df087a3b37c209e5f4b72fc27578d988155b74b05ec5df30b9.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT\n job_kind AS \"job_kind!: JobKind\",\n script_hash AS \"script_hash: ScriptHash\",\n flow_status AS \"flow_status!: Json>\",\n raw_flow AS \"raw_flow: Json>\"\n FROM v2_as_queue WHERE id = $1 AND workspace_id = $2 LIMIT 1", + "query": "SELECT\n kind AS \"job_kind!: JobKind\",\n runnable_id AS \"script_hash: ScriptHash\",\n flow_status AS \"flow_status!: Json>\",\n raw_flow AS \"raw_flow: Json>\"\n FROM v2_job INNER JOIN v2_job_status ON v2_job.id = v2_job_status.id WHERE v2_job.id = $1 AND v2_job.workspace_id = $2 LIMIT 1", "describe": { "columns": [ { @@ -57,11 +57,11 @@ ] }, "nullable": [ - true, + false, true, true, true ] }, - "hash": "4535c8effd1bae49894d13293a37e1ee949cf9108239032cb3addbf350fb33de" + "hash": "c50b6a4a6739d6df087a3b37c209e5f4b72fc27578d988155b74b05ec5df30b9" } diff --git a/backend/.sqlx/query-2f2ef9b1ccff527c48fa01cf1b78cd0e58c8d534ac22ec0356d82a854b31d087.json b/backend/.sqlx/query-ec1f31fd7628ea2e30995a0de1d8665831ee3e4ec3815e9ad90e886ffecba0f1.json similarity index 57% rename from backend/.sqlx/query-2f2ef9b1ccff527c48fa01cf1b78cd0e58c8d534ac22ec0356d82a854b31d087.json rename to backend/.sqlx/query-ec1f31fd7628ea2e30995a0de1d8665831ee3e4ec3815e9ad90e886ffecba0f1.json index 77e16162bd..b5629e0425 100644 --- a/backend/.sqlx/query-2f2ef9b1ccff527c48fa01cf1b78cd0e58c8d534ac22ec0356d82a854b31d087.json +++ b/backend/.sqlx/query-ec1f31fd7628ea2e30995a0de1d8665831ee3e4ec3815e9ad90e886ffecba0f1.json @@ -1,15 +1,14 @@ { "db_name": "PostgreSQL", - "query": "UPDATE v2_job_runtime SET ping = NULL\n WHERE id = $1 AND ping = $2", + "query": "UPDATE v2_job_runtime SET ping = NULL\n WHERE id = $1", "describe": { "columns": [], "parameters": { "Left": [ - "Uuid", - "Timestamptz" + "Uuid" ] }, "nullable": [] }, - "hash": "2f2ef9b1ccff527c48fa01cf1b78cd0e58c8d534ac22ec0356d82a854b31d087" + "hash": "ec1f31fd7628ea2e30995a0de1d8665831ee3e4ec3815e9ad90e886ffecba0f1" } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 1cd709f740..41ce253b23 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -280b866cd0b179c16c894197a891f24495a26549 \ No newline at end of file +94ff10ff18c121f49180ddaebf5dee8cd94d06df \ No newline at end of file diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index cf82ed7f49..0a87924669 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -50,7 +50,7 @@ use windmill_common::{ SMTP_CONFIG, TMP_DIR, WORKER_CONFIG, WORKER_GROUP, }, KillpillSender, BASE_URL, CRITICAL_ALERT_MUTE_UI_ENABLED, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS, METRICS_DEBUG_ENABLED, METRICS_ENABLED, MONITOR_LOGS_ON_OBJECT_STORE, OTEL_LOGS_ENABLED, OTEL_METRICS_ENABLED, OTEL_TRACING_ENABLED, SERVICE_LOG_RETENTION_SECS }; -use windmill_queue::cancel_job; +use windmill_queue::{cancel_job, MiniPulledJob}; use windmill_worker::{ create_token_for_owner, handle_job_error, AuthedClient, SameWorkerPayload, SameWorkerSender, SendResult, BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, @@ -1826,7 +1826,7 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker let _ = handle_job_error( db, &client, - &job, + &MiniPulledJob::from(&job), 0, None, error::Error::ExecutionErr(error_message), diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 2b4bd94f9c..6a49031517 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -6649,6 +6649,7 @@ paths: - $ref: "#/components/parameters/OrderDesc" - $ref: "#/components/parameters/CreatedBy" - $ref: "#/components/parameters/ParentJob" + - $ref: "#/components/parameters/Worker" - $ref: "#/components/parameters/ScriptExactPath" - $ref: "#/components/parameters/ScriptStartPath" - $ref: "#/components/parameters/SchedulePath" @@ -6856,6 +6857,7 @@ paths: - $ref: "#/components/parameters/OrderDesc" - $ref: "#/components/parameters/CreatedBy" - $ref: "#/components/parameters/Label" + - $ref: "#/components/parameters/Worker" - $ref: "#/components/parameters/ParentJob" - $ref: "#/components/parameters/ScriptExactPath" - $ref: "#/components/parameters/ScriptStartPath" @@ -6910,6 +6912,7 @@ paths: - $ref: "#/components/parameters/WorkspaceId" - $ref: "#/components/parameters/CreatedBy" - $ref: "#/components/parameters/Label" + - $ref: "#/components/parameters/Worker" - $ref: "#/components/parameters/ParentJob" - $ref: "#/components/parameters/ScriptExactPath" - $ref: "#/components/parameters/ScriptStartPath" @@ -12097,6 +12100,12 @@ components: in: query schema: type: string + Worker: + name: worker + description: worker this job was ran on + in: query + schema: + type: string ParentJob: name: parent_job description: @@ -12732,6 +12741,8 @@ components: type: number preprocessed: type: boolean + worker: + type: string required: - id - running @@ -12841,6 +12852,8 @@ components: type: number preprocessed: type: boolean + worker: + type: string required: - id - created_by diff --git a/backend/windmill-api/src/capture.rs b/backend/windmill-api/src/capture.rs index 402d504a2c..06e4f4ad4d 100644 --- a/backend/windmill-api/src/capture.rs +++ b/backend/windmill-api/src/capture.rs @@ -47,13 +47,13 @@ use hyper::StatusCode; use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; use sqlx::types::Json as SqlxJson; -use std::fmt; use windmill_common::{ db::UserDB, error::{JsonResult, Result}, utils::{not_found_if_none, paginate, Pagination, StripPath}, worker::{to_raw_value, CLOUD_HOSTED}, }; +use windmill_queue::TriggerKind; use windmill_queue::{PushArgs, PushArgsOwned}; const KEEP_LAST: i64 = 20; @@ -94,38 +94,6 @@ pub fn workspaced_unauthed_service() -> Router { } } -#[derive(sqlx::Type, Serialize, Deserialize, Debug)] -#[sqlx(type_name = "TRIGGER_KIND", rename_all = "lowercase")] -#[serde(rename_all = "lowercase")] -pub enum TriggerKind { - Webhook, - Http, - Websocket, - Kafka, - Email, - Nats, - Mqtt, - Sqs, - Postgres, -} - -impl fmt::Display for TriggerKind { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let s = match self { - TriggerKind::Webhook => "webhook", - TriggerKind::Http => "http", - TriggerKind::Websocket => "websocket", - TriggerKind::Kafka => "kafka", - TriggerKind::Email => "email", - TriggerKind::Nats => "nats", - TriggerKind::Mqtt => "mqtt", - TriggerKind::Sqs => "sqs", - TriggerKind::Postgres => "postgres", - }; - write!(f, "{}", s) - } -} - #[cfg(feature = "http_trigger")] #[derive(Serialize, Deserialize)] struct HttpTriggerConfig { diff --git a/backend/windmill-api/src/concurrency_groups.rs b/backend/windmill-api/src/concurrency_groups.rs index 3f7e41617f..e59821c3e5 100644 --- a/backend/windmill-api/src/concurrency_groups.rs +++ b/backend/windmill-api/src/concurrency_groups.rs @@ -199,6 +199,7 @@ async fn get_concurrent_intervals( result: None, tag: None, has_null_parent: None, + worker: None, label: None, scheduled_for_before_now: _, is_not_schedule: _, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 5148359981..4b5a4f3daf 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -676,64 +676,167 @@ async fn get_job( } macro_rules! get_job_query { - ("v2_as_completed_job", $($opts:tt)*) => { + ("v2_job_completed", $($opts:tt)*) => { get_job_query!( - @impl "v2_as_completed_job", ($($opts)*), - "duration_ms, success, result, result_columns, deleted, is_skipped, result->'wm_labels' as labels, \ + @impl "v2_job_completed", ($($opts)*), + "v2_job_completed.duration_ms, CASE WHEN status = 'success' OR status = 'skipped' THEN true ELSE false END as success, result_columns, deleted, status = 'skipped' as is_skipped, result->'wm_labels' as labels, \ CASE WHEN result is null or pg_column_size(result) < 90000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result", + "", ) }; - ("v2_as_queue", $($opts:tt)*) => { + ("v2_job_queue", $($opts:tt)*) => { get_job_query!( - @impl "v2_as_queue", ($($opts)*), - "scheduled_for, running, last_ping, suspend, suspend_until, same_worker, pre_run_error, visible_to_owner, \ - root_job, leaf_jobs, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl,\ + @impl "v2_job_queue", ($($opts)*), + "scheduled_for, running, ping as last_ping, suspend, suspend_until, same_worker, pre_run_error, visible_to_owner, \ + flow_innermost_root_job AS root_job, flow_leaf_jobs AS leaf_jobs, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl,\ script_entrypoint_override", + "LEFT JOIN v2_job_runtime ON v2_job_runtime.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id", ) }; - (@impl $table:literal, (with_logs: $with_logs:expr, $($rest:tt)*), $additional_fields:literal, $($args:tt)*) => { + (@impl $table:literal, (with_logs: $with_logs:expr, $($rest:tt)*), $additional_fields:literal, $additional_joins:literal, $($args:tt)*) => { if $with_logs { - get_job_query!(@impl $table, ($($rest)*), $additional_fields, logs = "right(job_logs.logs, 20000)", $($args)*) + get_job_query!(@impl $table, ($($rest)*), $additional_fields, $additional_joins, logs = "right(job_logs.logs, 20000)", $($args)*) } else { - get_job_query!(@impl $table, ($($rest)*), $additional_fields, logs = "null", $($args)*) + get_job_query!(@impl $table, ($($rest)*), $additional_fields, $additional_joins, logs = "null", $($args)*) } }; - (@impl $table:literal, (with_code: $with_code:expr, $($rest:tt)*), $additional_fields:literal, $($args:tt)*) => { + (@impl $table:literal, (with_code: $with_code:expr, $($rest:tt)*), $additional_fields:literal, $additional_joins:literal, $($args:tt)*) => { if $with_code { - get_job_query!(@impl $table, ($($rest)*), $additional_fields, lock = "raw_lock", code = "raw_code", $($args)*) + get_job_query!(@impl $table, ($($rest)*), $additional_fields, $additional_joins, lock = "raw_lock", code = "raw_code", $($args)*) } else { - get_job_query!(@impl $table, ($($rest)*), $additional_fields, lock = "null", code = "null", $($args)*) + get_job_query!(@impl $table, ($($rest)*), $additional_fields, $additional_joins, lock = "null", code = "null", $($args)*) } }; - (@impl $table:literal, (with_flow: $with_flow:expr, $($rest:tt)*), $additional_fields:literal, $($args:tt)*) => { + (@impl $table:literal, (with_flow: $with_flow:expr, $($rest:tt)*), $additional_fields:literal, $additional_joins:literal, $($args:tt)*) => { if $with_flow { - get_job_query!(@impl $table, ($($rest)*), $additional_fields, flow = "raw_flow", $($args)*) + get_job_query!(@impl $table, ($($rest)*), $additional_fields, $additional_joins, flow = "raw_flow", $($args)*) } else { - get_job_query!(@impl $table, ($($rest)*), $additional_fields, flow = "null", $($args)*) + get_job_query!(@impl $table, ($($rest)*), $additional_fields, $additional_joins, flow = "null", $($args)*) } }; - (@impl $table:literal, (), $additional_fields:literal, $($args:tt)*) => { + (@impl $table:literal, (), $additional_fields:literal, $additional_joins:literal, $($args:tt)*) => { const_format::formatcp!( "SELECT \ - id, {table}.workspace_id, parent_job, created_by, {table}.created_at, started_at, script_hash, script_path, \ + {table}.id, {table}.workspace_id, parent_job, v2_job.created_by, v2_job.created_at, started_at, v2_job.runnable_id as script_hash, v2_job.runnable_path as script_path, \ CASE WHEN args is null THEN NULL WHEN pg_column_size(args) < 90000 THEN CASE WHEN jsonb_typeof(args) = 'object' THEN args ELSE jsonb_build_object('value', args) END - ELSE '{{\"reason\": \"WINDMILL_TOO_BIG\"}}'::jsonb END as args, \ - {logs} as logs, {code} as raw_code, canceled, canceled_by, canceled_reason, job_kind, \ - schedule_path, permissioned_as, flow_status, {flow} as raw_flow, is_flow_step, language, \ - {lock} as raw_lock, email, visible_to_owner, mem_peak, tag, priority, preprocessed, {additional_fields} \ - FROM {table} LEFT JOIN job_logs ON id = job_id \ - WHERE id = $1 AND {table}.workspace_id = $2 AND ($3::text[] IS NULL OR tag = ANY($3)) LIMIT 1", + ELSE '{{\"reason\": \"WINDMILL_TOO_BIG\"}}'::jsonb END as args, COALESCE(flow_status, workflow_as_code_status) AS flow_status, \ + {logs} as logs, {code} as raw_code, canceled_by is not null as canceled, canceled_by, canceled_reason, kind as job_kind, \ + CASE WHEN trigger_kind = 'schedule'::job_trigger_kind THEN trigger END AS schedule_path, permissioned_as, \ + {flow} as raw_flow, flow_step_id IS NOT NULL AS is_flow_step, script_lang as language, \ + {lock} as raw_lock, permissioned_as_email as email, visible_to_owner, memory_peak as mem_peak, v2_job.tag, v2_job.priority, preprocessed, worker,\ + {additional_fields} \ + FROM {table} + INNER JOIN v2_job ON v2_job.id = {table}.id \ + {additional_joins} \ + LEFT JOIN job_logs ON {table}.id = job_id \ + WHERE {table}.id = $1 AND {table}.workspace_id = $2 AND ($3::text[] IS NULL OR v2_job.tag = ANY($3))", table = $table, additional_fields = $additional_fields, + additional_joins = $additional_joins, $($args)* ) } } +// CREATE OR REPLACE VIEW v2_as_queue AS +// SELECT +// j.id, +// j.workspace_id, +// j.parent_job, +// j.created_by, +// j.created_at, +// q.started_at, +// q.scheduled_for, +// q.running, +// j.runnable_id AS script_hash, +// j.runnable_path AS script_path, +// j.args, +// j.raw_code, +// q.canceled_by IS NOT NULL AS canceled, +// q.canceled_by, +// q.canceled_reason, +// r.ping AS last_ping, +// j.kind AS job_kind, +// CASE WHEN j.trigger_kind = 'schedule'::job_trigger_kind THEN j.trigger END +// AS schedule_path, +// j.permissioned_as, +// COALESCE(s.flow_status, s.workflow_as_code_status) AS flow_status, +// j.raw_flow, +// j.flow_step_id IS NOT NULL AS is_flow_step, +// j.script_lang AS language, +// q.suspend, +// q.suspend_until, +// j.same_worker, +// j.raw_lock, +// j.pre_run_error, +// j.permissioned_as_email AS email, +// j.visible_to_owner, +// r.memory_peak AS mem_peak, +// j.flow_innermost_root_job AS root_job, +// s.flow_leaf_jobs AS leaf_jobs, +// j.tag, +// j.concurrent_limit, +// j.concurrency_time_window_s, +// j.timeout, +// j.flow_step_id, +// j.cache_ttl, +// j.priority, +// NULL::TEXT AS logs, +// j.script_entrypoint_override, +// j.preprocessed +// FROM v2_job_queue q +// JOIN v2_job j USING (id) +// LEFT JOIN v2_job_runtime r USING (id) +// LEFT JOIN v2_job_status s USING (id) +// ; + +// -- Add up migration script here +// CREATE OR REPLACE VIEW v2_as_completed_job AS +// SELECT +// j.id, +// j.workspace_id, +// j.parent_job, +// j.created_by, +// j.created_at, +// c.duration_ms, +// c.status = 'success' OR c.status = 'skipped' AS success, +// j.runnable_id AS script_hash, +// j.runnable_path AS script_path, +// j.args, +// c.result, +// FALSE AS deleted, +// j.raw_code, +// c.status = 'canceled' AS canceled, +// c.canceled_by, +// c.canceled_reason, +// j.kind AS job_kind, +// CASE WHEN j.trigger_kind = 'schedule'::job_trigger_kind THEN j.trigger END +// AS schedule_path, +// j.permissioned_as, +// COALESCE(c.flow_status, c.workflow_as_code_status) AS flow_status, +// j.raw_flow, +// j.flow_step_id IS NOT NULL AS is_flow_step, +// j.script_lang AS language, +// c.started_at, +// c.status = 'skipped' AS is_skipped, +// j.raw_lock, +// j.permissioned_as_email AS email, +// j.visible_to_owner, +// c.memory_peak AS mem_peak, +// j.tag, +// j.priority, +// NULL::TEXT AS logs, +// c.result_columns, +// j.script_entrypoint_override, +// j.preprocessed +// FROM v2_job_completed c +// JOIN v2_job j USING (id) +// ; + #[derive(Copy, Clone)] struct GetQuery<'a> { with_logs: bool, @@ -839,7 +942,7 @@ impl<'a> GetQuery<'a> { job_id: Uuid, workspace_id: &str, ) -> error::Result>> { - let query = get_job_query!("v2_as_queue", + let query = get_job_query!("v2_job_queue", with_logs: self.with_logs, with_code: self.with_code, with_flow: self.with_flow, @@ -871,11 +974,13 @@ impl<'a> GetQuery<'a> { job_id: Uuid, workspace_id: &str, ) -> error::Result>> { - let query = get_job_query!("v2_as_completed_job", + let query = get_job_query!("v2_job_completed", with_logs: self.with_logs, with_code: self.with_code, with_flow: self.with_flow, ); + + // tracing::info!("query: {}", query); let query = sqlx::query_as::<_, JobExtended>(query) .bind(job_id) .bind(workspace_id) @@ -1259,6 +1364,7 @@ pub struct ListQueueQuery { pub order_desc: Option, pub job_kinds: Option, pub suspended: Option, + pub worker: Option, // filter by matching a subset of the args using base64 encoded json subset pub args: Option, pub tag: Option, @@ -1283,6 +1389,7 @@ impl From for ListQueueQuery { created_after: lcq.created_after, created_or_started_before: lcq.created_or_started_before, created_or_started_after: lcq.created_or_started_after, + worker: lcq.worker, running: lcq.running, parent_job: lcq.parent_job, order_desc: lcq.order_desc, @@ -1319,6 +1426,10 @@ pub fn filter_list_queue_query( sqlb.and_where_eq("v2_job.workspace_id", "?".bind(&w_id)); } + if let Some(w) = &lq.worker { + sqlb.and_where_eq("v2_job_queue.worker", "?".bind(w)); + } + if let Some(ps) = &lq.script_path_start { sqlb.and_where_like_left("runnable_path", ps); } @@ -2444,6 +2555,9 @@ pub struct JobExtended { #[serde(skip_serializing_if = "Option::is_none")] pub raw_flow: Option>>, + #[serde(skip_serializing_if = "Option::is_none")] + pub worker: Option, + #[sqlx(skip)] #[serde(skip_serializing_if = "Option::is_none")] pub self_wait_time_ms: Option, @@ -2463,6 +2577,7 @@ impl JobExtended { raw_code: None, raw_lock: None, raw_flow: None, + worker: None, self_wait_time_ms, aggregate_wait_time_ms, } @@ -2678,6 +2793,7 @@ pub struct UnifiedJob { pub self_wait_time_ms: Option, pub aggregate_wait_time_ms: Option, pub preprocessed: Option, + pub worker: Option, } const CJ_FIELDS: &[&str] = &[ @@ -2716,6 +2832,7 @@ const CJ_FIELDS: &[&str] = &[ "self_wait_time_ms", "aggregate_wait_time_ms", "v2_job.preprocessed", + "v2_job_completed.worker", ]; const QJ_FIELDS: &[&str] = &[ @@ -2754,6 +2871,7 @@ const QJ_FIELDS: &[&str] = &[ "self_wait_time_ms", "aggregate_wait_time_ms", "v2_job.preprocessed", + "v2_job_queue.worker", ]; impl UnifiedJob { @@ -3358,7 +3476,7 @@ pub async fn run_workflow_as_code( path: job.script_path, language: job.language.unwrap_or_else(|| ScriptLang::Deno), lock: raw_lock, - custom_concurrency_key: windmill_queue::custom_concurrency_key(&db, job.id) + custom_concurrency_key: windmill_queue::custom_concurrency_key(&db, &job.id) .await .map_err(to_anyhow)?, concurrent_limit: job.concurrent_limit, @@ -5328,6 +5446,10 @@ pub fn filter_list_completed_query( sqlb.and_where(&wh); } + if let Some(worker) = &lq.worker { + sqlb.and_where_eq("v2_job_completed.worker", "?".bind(worker)); + } + if w_id != "admins" || !lq.all_workspaces.is_some_and(|x| x) { sqlb.and_where_eq("v2_job.workspace_id", "?".bind(&w_id)); } @@ -5486,6 +5608,7 @@ pub struct ListCompletedQuery { pub label: Option, pub is_not_schedule: Option, pub concurrency_key: Option, + pub worker: Option, } async fn list_completed_jobs( diff --git a/backend/windmill-api/src/mqtt_triggers.rs b/backend/windmill-api/src/mqtt_triggers.rs index 5f63e95b9e..67c4c12d5b 100644 --- a/backend/windmill-api/src/mqtt_triggers.rs +++ b/backend/windmill-api/src/mqtt_triggers.rs @@ -1,10 +1,12 @@ use crate::{ - capture::{insert_capture_payload, MqttTriggerConfig, TriggerKind}, + capture::{insert_capture_payload, MqttTriggerConfig}, db::{ApiAuthed, DB}, jobs::{run_flow_by_path_inner, run_script_by_path_inner, RunJobQuery}, resources::try_get_resource_from_db_as, users::fetch_api_authed, }; +use windmill_queue::TriggerKind; + use axum::{ async_trait, extract::{Path, Query}, diff --git a/backend/windmill-api/src/postgres_triggers/trigger.rs b/backend/windmill-api/src/postgres_triggers/trigger.rs index 81f439d0ae..2e6727704f 100644 --- a/backend/windmill-api/src/postgres_triggers/trigger.rs +++ b/backend/windmill-api/src/postgres_triggers/trigger.rs @@ -1,7 +1,7 @@ use std::{collections::HashMap, pin::Pin}; use crate::{ - capture::{insert_capture_payload, PostgresTriggerConfig, TriggerKind}, + capture::{insert_capture_payload, PostgresTriggerConfig}, db::{ApiAuthed, DB}, postgres_triggers::{ relation::RelationConverter, @@ -14,6 +14,8 @@ use crate::{ resources::try_get_resource_from_db_as, users::fetch_api_authed, }; +use windmill_queue::TriggerKind; + use bytes::{BufMut, Bytes, BytesMut}; use chrono::TimeZone; use futures::{pin_mut, SinkExt, StreamExt}; diff --git a/backend/windmill-api/src/websocket_triggers.rs b/backend/windmill-api/src/websocket_triggers.rs index 4de01f14f3..9be3829d83 100644 --- a/backend/windmill-api/src/websocket_triggers.rs +++ b/backend/windmill-api/src/websocket_triggers.rs @@ -30,8 +30,10 @@ use windmill_common::{ }; use windmill_queue::PushArgsOwned; +use windmill_queue::TriggerKind; + use crate::{ - capture::{insert_capture_payload, TriggerKind, WebsocketTriggerConfig}, + capture::{insert_capture_payload, WebsocketTriggerConfig}, db::{ApiAuthed, DB}, jobs::{ run_flow_by_path_inner, run_script_by_path_inner, run_wait_result_internal, RunJobQuery, diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 7a1f53774b..856555f29d 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -124,31 +124,29 @@ fn format_pull_query(peek: String) -> String { worker = $1 WHERE id = (SELECT id FROM peek) RETURNING - started_at, scheduled_for, running, - canceled_by, canceled_reason, canceled_by IS NOT NULL AS canceled, - suspend, suspend_until + started_at, scheduled_for, + canceled_by, canceled_reason, worker ), r AS NOT MATERIALIZED ( UPDATE v2_job_runtime SET ping = now() WHERE id = (SELECT id FROM peek) ), j AS NOT MATERIALIZED ( SELECT - id, workspace_id, parent_job, created_by, created_at, runnable_id AS script_hash, - runnable_path AS script_path, args, kind AS job_kind, - CASE WHEN trigger_kind = 'schedule' THEN trigger END AS schedule_path, - permissioned_as, permissioned_as_email AS email, script_lang AS language, - flow_innermost_root_job AS root_job, flow_step_id, flow_step_id IS NOT NULL AS is_flow_step, + id, workspace_id, parent_job, created_by, created_at, runnable_id, + runnable_path, args, kind, trigger, trigger_kind, + permissioned_as, permissioned_as_email, script_lang, + flow_innermost_root_job, flow_step_id, same_worker, pre_run_error, visible_to_owner, tag, concurrent_limit, concurrency_time_window_s, timeout, cache_ttl, priority, raw_code, raw_lock, raw_flow, script_entrypoint_override, preprocessed FROM v2_job WHERE id = (SELECT id FROM peek) - ) SELECT id, workspace_id, parent_job, created_by, created_at, started_at, scheduled_for, - running, script_hash, script_path, args, null as logs, canceled, canceled_by, - canceled_reason, null as last_ping, job_kind, schedule_path, permissioned_as, - flow_status, is_flow_step, language, suspend, suspend_until, - same_worker, pre_run_error, email, visible_to_owner, null as mem_peak, - root_job, flow_leaf_jobs as leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, + ) SELECT id, workspace_id, parent_job, created_by, started_at, scheduled_for, + runnable_id, runnable_path, args, canceled_by, + canceled_reason, kind, trigger, trigger_kind, permissioned_as, permissioned_as_email, + flow_status, script_lang, + same_worker, pre_run_error, visible_to_owner, + tag, concurrent_limit, concurrency_time_window_s, flow_innermost_root_job, timeout, flow_step_id, cache_ttl, priority, raw_code, raw_lock, raw_flow, script_entrypoint_override, preprocessed FROM q, j diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index f0eda8328a..2adaa99772 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -6,6 +6,7 @@ * LICENSE-AGPL for a copy of the license. */ +use std::fmt; use std::{borrow::Borrow, collections::HashMap, sync::Arc, vec}; use anyhow::Context; @@ -17,8 +18,10 @@ use itertools::Itertools; use prometheus::IntCounter; use regex::Regex; use reqwest::Client; +use serde::Deserialize; use serde::{ser::SerializeMap, Serialize}; use serde_json::{json, value::RawValue}; +use sqlx::PgExecutor; use sqlx::{types::Json, FromRow, Pool, Postgres, Transaction}; use tokio::{sync::RwLock, time::sleep}; use ulid::Ulid; @@ -156,7 +159,7 @@ pub async fn cancel_single_job<'c>( .await; let add_job = add_completed_job_error( &db, - &job_running, + &MiniPulledJob::from(&job_running), job_running.mem_peak.unwrap_or(0), Some(CanceledBy { username: Some(username.to_string()), reason: Some(reason) }), e, @@ -488,7 +491,7 @@ where pub async fn add_completed_job_error( db: &Pool, - queued_job: &QueuedJob, + queued_job: &MiniPulledJob, mem_peak: i32, canceled_by: Option, e: serde_json::Value, @@ -544,7 +547,7 @@ lazy_static::lazy_static! { pub async fn add_completed_job( db: &Pool, - queued_job: &QueuedJob, + queued_job: &MiniPulledJob, success: bool, skipped: bool, result: Json<&T>, @@ -581,7 +584,7 @@ pub async fn add_completed_job( serde_json::to_string(&result).unwrap_or_else(|_| "".to_string()) ); - let mem_peak = mem_peak.max(queued_job.mem_peak.unwrap_or(0)); + let mem_peak = mem_peak; // add_time!(bench, "add_completed_job query START"); let _duration = sqlx::query_scalar!( @@ -638,7 +641,7 @@ pub async fn add_completed_job( .map_err(|e| Error::InternalErr(format!("Could not update job labels: {e:#}")))?; } - if !queued_job.is_flow_step { + if !queued_job.is_flow_step() { if let Some(parent_job) = queued_job.parent_job { let _ = sqlx::query_scalar!( "UPDATE v2_job_status SET @@ -670,7 +673,7 @@ pub async fn add_completed_job( tx = delete_job(tx, &job_id).await?; // tracing::error!("3 {:?}", start.elapsed()); - if queued_job.is_flow_step { + if queued_job.is_flow_step() { if let Some(parent_job) = queued_job.parent_job { // persist the flow last progress timestamp to avoid zombie flow jobs tracing::debug!( @@ -704,12 +707,12 @@ pub async fn add_completed_job( } } } else { - if queued_job.schedule_path.is_some() && queued_job.script_path.is_some() { - let schedule_path = queued_job.schedule_path.as_ref().unwrap(); - let script_path = queued_job.script_path.as_ref().unwrap(); + if queued_job.schedule_path().is_some() && queued_job.runnable_path.is_some() { + let schedule_path = queued_job.schedule_path().unwrap(); + let script_path = queued_job.runnable_path.as_ref().unwrap(); let schedule = - get_schedule_opt(&mut *tx, &queued_job.workspace_id, schedule_path).await?; + get_schedule_opt(&mut *tx, &queued_job.workspace_id, &schedule_path).await?; if let Some(schedule) = schedule { #[cfg(feature = "enterprise")] @@ -743,7 +746,7 @@ pub async fn add_completed_job( db, queued_job, &schedule, - script_path, + &script_path, &queued_job.workspace_id, ) .await @@ -760,7 +763,7 @@ pub async fn add_completed_job( if let Err(err) = apply_schedule_handlers( db, &schedule, - script_path, + &script_path, &queued_job.workspace_id, success, result, @@ -799,19 +802,17 @@ pub async fn add_completed_job( } } if queued_job.concurrent_limit.is_some() { - let concurrency_key = match concurrency_key(db, queued_job).await { + let concurrency_key = match concurrency_key(db, &queued_job.id).await { Ok(c) => c, Err(e) => { tracing::error!( "Could not get concurrency key for job {} defaulting to default key: {e:?}", queued_job.id ); - legacy_concurrency_key(db, queued_job) - .await - .unwrap_or_else(|| queued_job.full_path_with_workspace()) + "".to_string() } }; - if *DISABLE_CONCURRENCY_LIMIT { + if *DISABLE_CONCURRENCY_LIMIT || concurrency_key.is_empty() { tracing::warn!("Concurrency limit is disabled, skipping"); } else { if let Err(e) = sqlx::query_scalar!( @@ -852,16 +853,16 @@ pub async fn add_completed_job( tracing::info!( %job_id, - root_job = ?queued_job.root_job.map(|x| x.to_string()).unwrap_or_else(|| String::new()), - path = &queued_job.script_path(), - job_kind = ?queued_job.job_kind, + root_job = ?queued_job.flow_innermost_root_job.map(|x| x.to_string()).unwrap_or_else(|| String::new()), + path = &queued_job.runnable_path(), + job_kind = ?queued_job.kind, started_at = ?queued_job.started_at.map(|x| x.to_string()).unwrap_or_else(|| String::new()), duration = ?_duration, permissioned_as = ?queued_job.permissioned_as, - email = ?queued_job.email, + email = ?queued_job.permissioned_as_email, created_by = queued_job.created_by, - is_flow_step = queued_job.is_flow_step, - language = ?queued_job.language, + is_flow_step = queued_job.is_flow_step(), + language = ?queued_job.script_lang, success, "inserted completed job: {} (success: {success})", queued_job.id @@ -916,7 +917,7 @@ pub async fn add_completed_job( "INSERT INTO usage (id, is_workspace, month_, usage) VALUES ($1, FALSE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2", - queued_job.email, + queued_job.permissioned_as_email, additional_usage as i32 ) .execute(db) @@ -927,8 +928,8 @@ pub async fn add_completed_job( #[cfg(feature = "enterprise")] if !success { - async fn has_failure_module(db: &Pool, job: &QueuedJob) -> bool { - if let Ok(flow) = cache::job::fetch_flow(db, job.job_kind, job.script_hash).await { + async fn has_failure_module(db: &Pool, job: &MiniPulledJob) -> bool { + if let Ok(flow) = cache::job::fetch_flow(db, job.kind, job.runnable_id).await { return flow.value().failure_module.is_some(); } sqlx::query_scalar!( @@ -941,7 +942,7 @@ pub async fn add_completed_job( .unwrap_or(false) } - if queued_job.email == ERROR_HANDLER_USER_EMAIL { + if queued_job.permissioned_as_email == ERROR_HANDLER_USER_EMAIL { let base_url = BASE_URL.read().await; let w_id = &queued_job.workspace_id; report_critical_error( @@ -960,7 +961,7 @@ pub async fn add_completed_job( None, ) .await; - } else if queued_job.email == SCHEDULE_ERROR_HANDLER_USER_EMAIL { + } else if queued_job.permissioned_as_email == SCHEDULE_ERROR_HANDLER_USER_EMAIL { let base_url = BASE_URL.read().await; let w_id = &queued_job.workspace_id; report_error_to_workspace_handler_or_critical_side_channel( @@ -979,8 +980,8 @@ pub async fn add_completed_job( ) .await; } else if !_skip_downstream_error_handlers - && (matches!(queued_job.job_kind, JobKind::Script) - || matches!(queued_job.job_kind, JobKind::Flow) + && (matches!(queued_job.kind, JobKind::Script) + || matches!(queued_job.kind, JobKind::Flow) && !has_failure_module(db, queued_job).await) && queued_job.parent_job.is_none() { @@ -1030,8 +1031,8 @@ pub async fn add_completed_job( } } - if !queued_job.is_flow_step && queued_job.job_kind == JobKind::Script && canceled_by.is_none() { - if let Some(hash) = queued_job.script_hash { + if !queued_job.is_flow_step() && queued_job.kind == JobKind::Script && canceled_by.is_none() { + if let Some(hash) = queued_job.runnable_id { let p = sqlx::query_scalar!( "SELECT restart_unless_cancelled FROM script WHERE hash = $1 AND workspace_id = $2", hash.0, @@ -1054,7 +1055,7 @@ pub async fn add_completed_job( { let next_run = queued_job.started_at.unwrap_or(now) + chrono::Duration::try_seconds(10).unwrap(); - tracing::warn!("Perpetual script {:?} is running too fast, only 1 job per 10s it supported. Scheduling next run for {:?}", queued_job.script_path, next_run); + tracing::warn!("Perpetual script {:?} is running too fast, only 1 job per 10s it supported. Scheduling next run for {:?}", queued_job.runnable_path, next_run); Some(next_run) } else { None @@ -1067,14 +1068,14 @@ pub async fn add_completed_job( &queued_job.workspace_id, JobPayload::ScriptHash { hash, - path: queued_job.script_path().to_string(), - custom_concurrency_key: custom_concurrency_key(db, queued_job.id).await?, + path: queued_job.runnable_path().to_string(), + custom_concurrency_key: custom_concurrency_key(db, &queued_job.id).await?, concurrent_limit: queued_job.concurrent_limit, concurrency_time_window_s: queued_job.concurrency_time_window_s, cache_ttl: queued_job.cache_ttl, dedicated_worker: None, language: queued_job - .language + .script_lang .clone() .unwrap_or_else(|| ScriptLang::Deno), priority: queued_job.priority, @@ -1086,10 +1087,10 @@ pub async fn add_completed_job( .map(|x| PushArgs::from(&x.0)) .unwrap_or_else(|| PushArgs::from(&ehm)), &queued_job.created_by, - &queued_job.email, + &queued_job.permissioned_as_email, queued_job.permissioned_as.clone(), scheduled_for, - queued_job.schedule_path.clone(), + queued_job.schedule_path(), None, None, None, @@ -1116,7 +1117,7 @@ pub async fn add_completed_job( } pub async fn send_error_to_global_handler<'a, T: Serialize + Send + Sync>( - queued_job: &QueuedJob, + queued_job: &MiniPulledJob, db: &Pool, result: Json<&T>, ) -> Result<(), Error> { @@ -1131,8 +1132,8 @@ pub async fn send_error_to_global_handler<'a, T: Serialize + Send + Sync>( push_error_handler( db, queued_job.id, - queued_job.schedule_path.clone(), - queued_job.script_path.clone(), + queued_job.schedule_path(), + queued_job.runnable_path.clone(), queued_job.is_flow(), &queued_job.workspace_id, &prefixed_global_error_handler_path, @@ -1140,7 +1141,7 @@ pub async fn send_error_to_global_handler<'a, T: Serialize + Send + Sync>( None, queued_job.started_at, None, - &queued_job.email, + &queued_job.permissioned_as_email, false, true, None, @@ -1152,7 +1153,7 @@ pub async fn send_error_to_global_handler<'a, T: Serialize + Send + Sync>( } pub async fn report_error_to_workspace_handler_or_critical_side_channel( - queued_job: &QueuedJob, + queued_job: &MiniPulledJob, db: &Pool, error_message: String, ) -> () { @@ -1171,8 +1172,8 @@ pub async fn report_error_to_workspace_handler_or_critical_side_channel( if let Err(err) = push_error_handler( db, queued_job.id, - queued_job.schedule_path.clone(), - queued_job.script_path.clone(), + queued_job.schedule_path(), + queued_job.runnable_path.clone(), queued_job.is_flow(), w_id, &error_handler, @@ -1184,7 +1185,7 @@ pub async fn report_error_to_workspace_handler_or_critical_side_channel( None, queued_job.started_at, error_handler_extra_args, - &queued_job.email, + &queued_job.permissioned_as_email, false, false, None, @@ -1204,7 +1205,7 @@ pub async fn report_error_to_workspace_handler_or_critical_side_channel( } pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync>( - queued_job: &QueuedJob, + queued_job: &MiniPulledJob, is_canceled: bool, db: &Pool, result: Json<&'a T>, @@ -1224,12 +1225,12 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> } if let Some(error_handler) = error_handler { - let ws_error_handler_muted: Option = match queued_job.job_kind { + let ws_error_handler_muted: Option = match queued_job.kind { JobKind::Script => { sqlx::query_scalar!( "SELECT ws_error_handler_muted FROM script WHERE workspace_id = $1 AND hash = $2", queued_job.workspace_id, - queued_job.script_hash.unwrap().0, + queued_job.runnable_id.map(|x| x.0), ) .fetch_optional(db) .await? @@ -1238,7 +1239,7 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> sqlx::query_scalar!( "SELECT ws_error_handler_muted FROM flow WHERE workspace_id = $1 AND path = $2", queued_job.workspace_id, - queued_job.script_path.as_ref().unwrap(), + queued_job.runnable_path.clone(), ) .fetch_optional(db) .await? @@ -1253,8 +1254,8 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> push_error_handler( db, queued_job.id, - queued_job.schedule_path.clone(), - queued_job.script_path.clone(), + queued_job.schedule_path(), + queued_job.runnable_path.clone(), queued_job.is_flow(), &queued_job.workspace_id, &error_handler, @@ -1262,7 +1263,7 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> None, queued_job.started_at, error_handler_extra_args, - &queued_job.email, + &queued_job.permissioned_as_email, false, false, None, @@ -1276,7 +1277,7 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> pub async fn handle_maybe_scheduled_job<'c>( db: &Pool, - job: &QueuedJob, + job: &MiniPulledJob, schedule: &Schedule, script_path: &str, w_id: &str, @@ -1831,17 +1832,180 @@ async fn handle_successful_schedule<'a, 'c, T: Serialize + Send + Sync>( Ok(()) } -#[derive(sqlx::FromRow)] +#[derive(sqlx::Type, Serialize, Deserialize, Debug, Clone)] +#[sqlx(type_name = "TRIGGER_KIND", rename_all = "lowercase")] +#[serde(rename_all = "lowercase")] +pub enum TriggerKind { + Webhook, + Http, + Websocket, + Kafka, + Email, + Nats, + Mqtt, + Sqs, + Postgres, +} + +#[derive(sqlx::Type, Serialize, Deserialize, Debug, Clone)] +#[sqlx(type_name = "JOB_TRIGGER_KIND", rename_all = "lowercase")] +#[serde(rename_all = "lowercase")] +pub enum JobTriggerKind { + Webhook, + Http, + Websocket, + Kafka, + Email, + Nats, + Mqtt, + Sqs, + Postgres, + Schedule, +} + +impl fmt::Display for TriggerKind { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let s = match self { + TriggerKind::Webhook => "webhook", + TriggerKind::Http => "http", + TriggerKind::Websocket => "websocket", + TriggerKind::Kafka => "kafka", + TriggerKind::Email => "email", + TriggerKind::Nats => "nats", + TriggerKind::Mqtt => "mqtt", + TriggerKind::Sqs => "sqs", + TriggerKind::Postgres => "postgres", + }; + write!(f, "{}", s) + } +} + +#[derive(sqlx::FromRow, Debug, Clone)] +pub struct MiniPulledJob { + pub workspace_id: String, + pub id: Uuid, + pub args: Option>>>, + pub parent_job: Option, + pub created_by: String, + pub scheduled_for: chrono::DateTime, + pub started_at: Option>, + pub runnable_path: Option, + pub kind: JobKind, + pub runnable_id: Option, + pub canceled_reason: Option, + pub canceled_by: Option, + pub permissioned_as: String, + pub permissioned_as_email: String, + pub flow_status: Option>>, + pub tag: String, + pub script_lang: Option, + pub same_worker: bool, + pub pre_run_error: Option, + pub concurrent_limit: Option, + pub concurrency_time_window_s: Option, + pub flow_innermost_root_job: Option, + pub timeout: Option, + pub flow_step_id: Option, + pub cache_ttl: Option, + pub priority: Option, + pub preprocessed: Option, + pub script_entrypoint_override: Option, + pub trigger: Option, + pub trigger_kind: Option, + pub visible_to_owner: bool, +} + +impl MiniPulledJob { + pub fn runnable_path(&self) -> &str { + self.runnable_path + .as_ref() + .map(String::as_str) + .unwrap_or("tmp/main") + } + + pub fn is_flow_step(&self) -> bool { + self.flow_step_id.is_some() + } + + pub fn is_canceled(&self) -> bool { + self.canceled_by.is_some() + } + + pub fn parse_flow_status(&self) -> Option { + // tracing::error!("parse_flow_status: {:?}", self.flow_status); + + self.flow_status + .as_ref() + .and_then(|v| serde_json::from_str::((**v).get()).ok()) + } + + pub fn from(job: &QueuedJob) -> MiniPulledJob { + MiniPulledJob { + workspace_id: job.workspace_id.clone(), + id: job.id, + args: job.args.clone(), + parent_job: job.parent_job.clone(), + created_by: job.created_by.clone(), + started_at: job.started_at.clone(), + scheduled_for: job.scheduled_for, + runnable_path: job.script_path.clone(), + kind: job.job_kind, + runnable_id: job.script_hash.clone(), + canceled_reason: job.canceled_reason.clone(), + canceled_by: job.canceled_by.clone(), + permissioned_as: job.permissioned_as.clone(), + permissioned_as_email: job.email.clone(), + flow_status: job.flow_status.clone(), + tag: job.tag.clone(), + script_lang: job.language.clone(), + same_worker: job.same_worker, + pre_run_error: job.pre_run_error.clone(), + concurrent_limit: job.concurrent_limit.clone(), + concurrency_time_window_s: job.concurrency_time_window_s.clone(), + flow_innermost_root_job: job.root_job.clone(), + timeout: job.timeout.clone(), + flow_step_id: job.flow_step_id.clone(), + cache_ttl: job.cache_ttl.clone(), + priority: job.priority.clone(), + preprocessed: job.preprocessed.clone(), + script_entrypoint_override: job.script_entrypoint_override.clone(), + trigger: job.schedule_path.clone(), + trigger_kind: if job.schedule_path.is_some() { + Some(JobTriggerKind::Schedule) + } else { + None + }, + visible_to_owner: job.visible_to_owner.clone(), + } + } + pub fn is_flow(&self) -> bool { + self.kind.is_flow() + } + + pub fn schedule_path(&self) -> Option { + if self + .trigger_kind + .as_ref() + .is_some_and(|t| matches!(t, JobTriggerKind::Schedule)) + { + self.trigger.clone() + } else { + None + } + } +} + +#[derive(sqlx::FromRow, Debug, Clone)] pub struct PulledJob { #[sqlx(flatten)] - pub job: QueuedJob, + pub job: MiniPulledJob, pub raw_code: Option, pub raw_lock: Option, pub raw_flow: Option>>, } impl std::ops::Deref for PulledJob { - type Target = QueuedJob; + type Target = MiniPulledJob; fn deref(&self) -> &Self::Target { &self.job } @@ -1851,6 +2015,52 @@ lazy_static::lazy_static! { static ref DISABLE_CONCURRENCY_LIMIT: bool = std::env::var("DISABLE_CONCURRENCY_LIMIT").is_ok_and(|s| s == "true"); } +pub async fn get_mini_pulled_job<'c>( + e: impl PgExecutor<'c>, + job_id: &Uuid, +) -> windmill_common::error::Result> { + let job = sqlx::query_as!( + MiniPulledJob, + "SELECT + v2_job_queue.workspace_id, + v2_job_queue.id, + v2_job.args as \"args: sqlx::types::Json>>\", + v2_job.parent_job, + v2_job.created_by, + v2_job_queue.started_at, + scheduled_for, + runnable_path, + kind as \"kind: JobKind\", + runnable_id as \"runnable_id: ScriptHash\", + canceled_reason, + canceled_by, + permissioned_as, + permissioned_as_email, + flow_status as \"flow_status: sqlx::types::Json>\", + v2_job.tag, + script_lang as \"script_lang: ScriptLang\", + same_worker, + pre_run_error, + concurrent_limit, + concurrency_time_window_s, + flow_innermost_root_job, + timeout, + flow_step_id, + cache_ttl, + v2_job_queue.priority, + preprocessed, + script_entrypoint_override, + trigger, + trigger_kind as \"trigger_kind: JobTriggerKind\", + visible_to_owner + FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1", + job_id, + ) + .fetch_optional(e) + .await?; + Ok(job) +} + pub async fn pull( db: &Pool, suspend_first: bool, @@ -1880,7 +2090,10 @@ pub async fn pull( // concurrency check. If more than X jobs for this path are already running, we re-queue and pull another job from the queue let pulled_job = job; - if pulled_job.script_path.is_none() || !has_concurent_limit || pulled_job.canceled { + if pulled_job.runnable_path.is_none() + || !has_concurent_limit + || pulled_job.canceled_by.is_some() + { #[cfg(feature = "prometheus")] if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { QUEUE_PULL_COUNT.inc(); @@ -1888,16 +2101,14 @@ pub async fn pull( return Ok((Option::Some(pulled_job), suspended)); } - let job_concurrency_key = match concurrency_key(db, &pulled_job).await { + let job_concurrency_key = match concurrency_key(db, &pulled_job.id).await { Ok(key) => key, Err(e) => { tracing::error!( "Could not get concurrency key for job {} defaulting to default key: {e:?}", pulled_job.id ); - legacy_concurrency_key(db, &pulled_job) - .await - .unwrap_or_else(|| pulled_job.full_path_with_workspace()) + "".to_string() } }; tracing::debug!("Concurrency key is '{}'", job_concurrency_key); @@ -1916,21 +2127,22 @@ pub async fn pull( ) .expect("Unable to serialize job_uuids column to proper JSON"); - let (within_limit, max_ended_at) = if *DISABLE_CONCURRENCY_LIMIT { - tracing::warn!("Concurrency limit is disabled, skipping"); - (true, None) - } else { - update_concurrency_counter( - db, - &pulled_job.id, - job_concurrency_key.clone(), - jobs_uuids_init_json_value, - pulled_job.id.hyphenated().to_string(), - job_custom_concurrency_time_window_s, - job_custom_concurrent_limit, - ) - .await? - }; + let (within_limit, max_ended_at) = + if *DISABLE_CONCURRENCY_LIMIT || job_concurrency_key.is_empty() { + tracing::warn!("Concurrency limit is disabled, skipping"); + (true, None) + } else { + update_concurrency_counter( + db, + &pulled_job.id, + job_concurrency_key.clone(), + jobs_uuids_init_json_value, + pulled_job.id.hyphenated().to_string(), + job_custom_concurrency_time_window_s, + job_custom_concurrent_limit, + ) + .await? + }; if within_limit { #[cfg(feature = "prometheus")] if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { @@ -1939,7 +2151,7 @@ pub async fn pull( return Ok((Option::Some(pulled_job), suspended)); } - let job_script_path = pulled_job.script_path.clone().unwrap_or_default(); + let job_script_path = pulled_job.runnable_path.clone().unwrap_or_default(); let min_started_at = sqlx::query!( "SELECT COALESCE((SELECT MIN(started_at) as min_started_at @@ -2116,56 +2328,18 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<'c>( pub async fn custom_concurrency_key( db: &Pool, - job_id: Uuid, + job_id: &Uuid, ) -> Result, sqlx::Error> { sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", job_id) .fetch_optional(db) // this should no longer be fetch optional .await } -async fn legacy_concurrency_key(db: &Pool, queued_job: &QueuedJob) -> Option { - let r = if queued_job.is_flow() { - sqlx::query_scalar!( - "SELECT flow_version.value->>'concurrency_key' - FROM flow - LEFT JOIN flow_version - ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] - WHERE flow.path = $1 AND flow.workspace_id = $2", - queued_job.script_path, - queued_job.workspace_id - ) - .fetch_optional(db) - .await - } else { - sqlx::query_scalar!( - "SELECT concurrency_key FROM script WHERE hash = $1 AND workspace_id = $2", - queued_job.script_hash.unwrap_or(ScriptHash(0)).0, - queued_job.workspace_id - ) - .fetch_optional(db) - .await - } - .ok() - .flatten() - .flatten(); - - let ehm = HashMap::new(); - let push_args = queued_job - .args - .as_ref() - .map(|x| PushArgs::from(&x.0)) - .unwrap_or_else(|| PushArgs::from(&ehm)); - r.map(|x| interpolate_args(x, &push_args, &queued_job.workspace_id)) -} - -async fn concurrency_key( - db: &Pool, - queued_job: &QueuedJob, -) -> windmill_common::error::Result { +async fn concurrency_key(db: &Pool, id: &Uuid) -> windmill_common::error::Result { not_found_if_none( - custom_concurrency_key(db, queued_job.id).await?, + custom_concurrency_key(db, id).await?, "ConcurrencyKey", - queued_job.id.to_string(), + id.to_string(), ) } @@ -3922,7 +4096,7 @@ pub async fn push<'c, 'd>( Ok((uuid, tx)) } -pub fn canceled_job_to_result(job: &QueuedJob) -> serde_json::Value { +pub fn canceled_job_to_result(job: &MiniPulledJob) -> serde_json::Value { let reason = job .canceled_reason .as_deref() diff --git a/backend/windmill-worker/src/ansible_executor.rs b/backend/windmill-worker/src/ansible_executor.rs index a1b03acf1e..a565ff6820 100644 --- a/backend/windmill-worker/src/ansible_executor.rs +++ b/backend/windmill-worker/src/ansible_executor.rs @@ -11,9 +11,10 @@ use tokio::process::Command; use uuid::Uuid; use windmill_common::{ error, - jobs::QueuedJob, worker::{to_raw_value, write_file, write_file_at_user_defined_location, WORKER_CONFIG}, }; +use windmill_queue::MiniPulledJob; + use windmill_parser_yaml::{AnsibleRequirements, ResourceOrVariablePath}; use windmill_queue::{append_logs, CanceledBy}; @@ -180,7 +181,7 @@ pub async fn handle_ansible_job( job_dir: &str, worker_dir: &str, worker_name: &str, - job: &QueuedJob, + job: &MiniPulledJob, mem_peak: &mut i32, canceled_by: &mut Option, db: &sqlx::Pool, diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs index 9eac965ad2..86f472705c 100644 --- a/backend/windmill-worker/src/bash_executor.rs +++ b/backend/windmill-worker/src/bash_executor.rs @@ -15,7 +15,6 @@ use tokio::process::Command; use uuid::Uuid; use windmill_common::{ error::Error, - jobs::QueuedJob, worker::{to_raw_value, write_file}, }; @@ -25,7 +24,7 @@ use windmill_common::DB; #[cfg(feature = "dind")] use windmill_common::error::to_anyhow; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::{append_logs, CanceledBy, MiniPulledJob}; lazy_static::lazy_static! { pub static ref BIN_BASH: String = std::env::var("BASH_PATH").unwrap_or_else(|_| "/bin/bash".to_string()); @@ -63,7 +62,7 @@ lazy_static::lazy_static! { pub async fn handle_bash_job( mem_peak: &mut i32, canceled_by: &mut Option, - job: &QueuedJob, + job: &MiniPulledJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, content: &str, @@ -163,7 +162,7 @@ exit $exit_status let nsjail = !*DISABLE_NSJAIL && job - .script_path + .runnable_path .as_ref() .map(|x| !x.starts_with("init_script_")) .unwrap_or(true); @@ -470,7 +469,7 @@ fn raw_to_string(x: &str) -> String { pub async fn handle_powershell_job( mem_peak: &mut i32, canceled_by: &mut Option, - job: &QueuedJob, + job: &MiniPulledJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, content: &str, diff --git a/backend/windmill-worker/src/bigquery_executor.rs b/backend/windmill-worker/src/bigquery_executor.rs index e4674ed6ee..6c3b9337fd 100644 --- a/backend/windmill-worker/src/bigquery_executor.rs +++ b/backend/windmill-worker/src/bigquery_executor.rs @@ -5,7 +5,6 @@ use futures::{FutureExt, TryFutureExt}; use reqwest::Client; use serde_json::{json, value::RawValue, Value}; use windmill_common::error::to_anyhow; -use windmill_common::jobs::QueuedJob; use windmill_common::{error::Error, worker::to_raw_value}; use windmill_parser_sql::{ parse_bigquery_sig, parse_db_resource, parse_sql_blocks, parse_sql_statement_named_params, @@ -203,8 +202,10 @@ fn do_bigquery_inner<'a>( Ok(result_f.boxed()) } +use windmill_queue::MiniPulledJob; + pub async fn do_bigquery( - job: &QueuedJob, + job: &MiniPulledJob, client: &AuthedClientBackgroundTask, query: &str, db: &sqlx::Pool, diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 29ffbc3044..8e13e0d57f 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -9,7 +9,7 @@ use serde_json::value::RawValue; use uuid::Uuid; use windmill_parser_ts::remove_pinned_imports; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::{append_logs, CanceledBy, MiniPulledJob}; #[cfg(feature = "enterprise")] use crate::common::build_envs_map; @@ -41,7 +41,6 @@ use windmill_common::variables; use windmill_common::{ error::{self, Result}, get_latest_hash_for_path, - jobs::QueuedJob, scripts::ScriptLang, worker::{exists_in_cache, save_cache, write_file, DISABLE_BUNDLING}, DB, @@ -824,7 +823,7 @@ pub async fn handle_bun_job( codebase: Option<&String>, mem_peak: &mut i32, canceled_by: &mut Option, - job: &QueuedJob, + job: &MiniPulledJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, job_dir: &str, @@ -846,7 +845,7 @@ pub async fn handle_bun_job( let (local_path, remote_path) = compute_bundle_local_and_remote_path( inner_content, requirements_o, - job.script_path(), + job.runnable_path(), Some(db.clone()), &job.workspace_id, ) @@ -871,7 +870,7 @@ pub async fn handle_bun_job( annotation.nodejs = true } let main_override = job.script_entrypoint_override.as_deref(); - let apply_preprocessor = !job.is_flow_step && job.preprocessed == Some(false); + let apply_preprocessor = !job.is_flow_step() && job.preprocessed == Some(false); if has_bundle_cache { let target; @@ -936,7 +935,7 @@ pub async fn handle_bun_job( &job.workspace_id, Some(db), &client.get_token().await, - &job.script_path(), + job.runnable_path(), job_dir, base_internal_url, worker_name, @@ -1130,7 +1129,7 @@ try {{ base_internal_url, &client.get_token().await, &job.workspace_id, - &job.script_path(), + job.runnable_path(), if annotation.nodejs { LoaderMode::NodeBundle } else if annotation.native { @@ -1148,7 +1147,7 @@ try {{ base_internal_url, &client.get_token().await, &job.workspace_id, - &job.script_path(), + job.runnable_path(), if annotation.nodejs { LoaderMode::Node } else { @@ -1500,7 +1499,7 @@ pub async fn start_worker( script_path: &str, token: &str, job_completed_tx: JobCompletedSender, - jobs_rx: Receiver>, + jobs_rx: Receiver>, killpill_rx: tokio::sync::broadcast::Receiver<()>, ) -> Result<()> { let mut logs = "".to_string(); diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 0ffdabb857..3b9eadae96 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -24,12 +24,12 @@ use windmill_common::worker::{ use windmill_common::{ cache::{Cache, RawData}, error::{self, Error}, - jobs::QueuedJob, scripts::ScriptHash, variables::ContextualVariable, }; use anyhow::{anyhow, Result}; +use windmill_queue::MiniPulledJob; use std::path::Path; use std::{collections::HashMap, sync::Arc, time::Duration}; @@ -45,7 +45,7 @@ use crate::{ }; pub async fn build_args_map<'a>( - job: &'a QueuedJob, + job: &'a MiniPulledJob, client: &AuthedClientBackgroundTask, db: &Pool, ) -> error::Result>>> { @@ -73,12 +73,12 @@ pub fn check_executor_binary_exists( } pub async fn build_args_values( - job: &QueuedJob, + job: &MiniPulledJob, client: &AuthedClientBackgroundTask, db: &Pool, ) -> error::Result> { if let Some(args) = &job.args { - transform_json_as_values(client, &job.workspace_id, &args.0, &job, db).await + transform_json_as_values(client, &job.workspace_id, &args.0, job, db).await } else { Ok(HashMap::new()) } @@ -87,7 +87,7 @@ pub async fn build_args_values( #[tracing::instrument(level = "trace", skip_all)] pub async fn create_args_and_out_file( client: &AuthedClientBackgroundTask, - job: &QueuedJob, + job: &MiniPulledJob, job_dir: &str, db: &Pool, ) -> Result<(), Error> { @@ -129,7 +129,7 @@ pub async fn transform_json<'a>( client: &AuthedClientBackgroundTask, workspace: &str, vs: &'a HashMap>, - job: &QueuedJob, + job: &MiniPulledJob, db: &Pool, ) -> error::Result>>> { let mut has_match = false; @@ -168,7 +168,7 @@ pub async fn transform_json_as_values<'a>( client: &AuthedClientBackgroundTask, workspace: &str, vs: &'a HashMap>, - job: &QueuedJob, + job: &MiniPulledJob, db: &Pool, ) -> error::Result> { let mut r: HashMap = HashMap::new(); @@ -238,7 +238,7 @@ pub async fn transform_json_value( client: &AuthedClient, workspace: &str, v: Value, - job: &QueuedJob, + job: &MiniPulledJob, db: &Pool, ) -> error::Result { match v { @@ -272,7 +272,8 @@ pub async fn transform_json_value( Value::String(y) if y.starts_with("$encrypted:") => { let encrypted = y.strip_prefix("$encrypted:").unwrap(); - let root_job_id = get_root_job_id(&job.root_job.unwrap_or_else(|| job.id), db).await?; + let root_job_id = + get_root_job_id(&job.flow_innermost_root_job.unwrap_or_else(|| job.id), db).await?; let mc = build_crypt_with_key_suffix(&db, &job.workspace_id, &root_job_id.to_string()) .await?; decrypt(&mc, encrypted.to_string()).and_then(|x| { @@ -295,16 +296,16 @@ pub async fn transform_json_value( db, &job.workspace_id, &client.token, - &job.email, + &job.permissioned_as_email, &job.created_by, &job.id.to_string(), &job.permissioned_as, - job.script_path.clone(), + job.runnable_path.clone(), job.parent_job.map(|x| x.to_string()), flow_path, - job.schedule_path.clone(), + job.schedule_path(), job.flow_step_id.clone(), - job.root_job.clone().map(|x| x.to_string()), + job.flow_innermost_root_job.clone().map(|x| x.to_string()), None, Some(job.scheduled_for.clone()), ) @@ -413,7 +414,7 @@ pub fn capitalize(s: &str) -> String { #[tracing::instrument(level = "trace", skip_all)] pub async fn get_reserved_variables( - job: &QueuedJob, + job: &MiniPulledJob, token: &str, db: &sqlx::Pool, ) -> Result, Error> { @@ -430,16 +431,16 @@ pub async fn get_reserved_variables( db, &job.workspace_id, token, - &job.email, + &job.permissioned_as_email, &job.created_by, &job.id.to_string(), &job.permissioned_as, - job.script_path.clone(), + job.runnable_path.clone(), job.parent_job.map(|x| x.to_string()), flow_path, - job.schedule_path.clone(), + job.schedule_path(), job.flow_step_id.clone(), - job.root_job.clone().map(|x| x.to_string()), + job.flow_innermost_root_job.clone().map(|x| x.to_string()), None, Some(job.scheduled_for.clone()), ) @@ -669,15 +670,15 @@ async fn hash_args( pub async fn cached_result_path( db: &DB, client: &AuthedClient, - job: &QueuedJob, + job: &MiniPulledJob, raw_data: Option<&RawData>, ) -> String { let mut hasher = sha2::Sha256::new(); - hasher.update(&[job.job_kind as u8]); - if let Some(ScriptHash(hash)) = job.script_hash { + hasher.update(&[job.kind as u8]); + if let Some(ScriptHash(hash)) = job.runnable_id { hasher.update(&hash.to_le_bytes()) } else { - job.script_path + job.runnable_path .as_ref() .inspect(|x| hasher.update(x.as_bytes())); match raw_data { @@ -890,7 +891,7 @@ pub async fn get_cached_resource_value_if_valid( pub async fn save_in_cache( db: &Pool, _client: &AuthedClient, - job: &QueuedJob, + job: &MiniPulledJob, cached_path: String, r: Arc>, ) { diff --git a/backend/windmill-worker/src/csharp_executor.rs b/backend/windmill-worker/src/csharp_executor.rs index 5694920e88..811443d3b8 100644 --- a/backend/windmill-worker/src/csharp_executor.rs +++ b/backend/windmill-worker/src/csharp_executor.rs @@ -19,7 +19,6 @@ use windmill_common::{ }; use windmill_common::error::{self, Error}; -use windmill_common::jobs::QueuedJob; #[cfg(feature = "csharp")] use windmill_queue::append_logs; @@ -426,11 +425,13 @@ fn remove_lines_from_text(contents: &str, indices_to_remove: Vec) -> Stri result.join("\n") } +use windmill_queue::MiniPulledJob; + #[cfg(not(feature = "csharp"))] pub async fn handle_csharp_job( _mem_peak: &mut i32, _canceled_by: &mut Option, - _job: &QueuedJob, + _job: &MiniPulledJob, _db: &sqlx::Pool, _client: &AuthedClientBackgroundTask, _inner_content: &str, @@ -449,7 +450,7 @@ pub async fn handle_csharp_job( pub async fn handle_csharp_job( mem_peak: &mut i32, canceled_by: &mut Option, - job: &QueuedJob, + job: &MiniPulledJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, inner_content: &str, diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index b67483dd9c..2b63a6d28a 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -19,13 +19,13 @@ use windmill_common::KillpillSender; use windmill_common::{ cache, error, flows::{FlowModule, FlowModuleValue}, - jobs::QueuedJob, scripts::{ScriptHash, ScriptLang}, variables, worker::to_raw_value, DB, }; use windmill_queue::append_logs; +use windmill_queue::MiniPulledJob; use anyhow::Context; @@ -70,7 +70,7 @@ pub async fn handle_dedicated_process( mut killpill_rx: tokio::sync::broadcast::Receiver<()>, job_completed_tx: JobCompletedSender, token: &str, - mut jobs_rx: Receiver>, + mut jobs_rx: Receiver>, worker_name: &str, db: &DB, script_path: &str, @@ -78,6 +78,8 @@ pub async fn handle_dedicated_process( ) -> std::result::Result<(), error::Error> { //do not cache local dependencies + use windmill_queue::MiniPulledJob; + use crate::{handle_child::process_status, PROXY_ENVS}; let cmd_name = format!("dedicated {command_path}"); let mut child = { @@ -134,7 +136,8 @@ pub async fn handle_dedicated_process( } }); - let mut jobs: VecDeque> = VecDeque::with_capacity(MAX_BUFFERED_DEDICATED_JOBS); + let mut jobs: VecDeque> = + VecDeque::with_capacity(MAX_BUFFERED_DEDICATED_JOBS); // let mut i = 0; // let mut j = 0; let mut alive = true; @@ -179,7 +182,7 @@ pub async fn handle_dedicated_process( } tracing::debug!("processed job: |{line}|"); if line.starts_with("wm_res[") { - let job: Arc = jobs.pop_front().expect("pop"); + let job: Arc = jobs.pop_front().expect("pop"); tracing::info!("job completed on dedicated worker {script_path}: {}", job.id); match serde_json::from_str::>(&line.replace("wm_res[success]:", "").replace("wm_res[error]:", "")) { Ok(result) => { @@ -242,7 +245,7 @@ pub async fn handle_dedicated_process( type DedicatedWorker = ( String, - Sender>, + Sender>, Option>, ); @@ -262,7 +265,7 @@ async fn spawn_dedicated_workers_for_flow( job_completed_tx: &JobCompletedSender, ) -> Vec { let mut workers = vec![]; - let mut script_path_to_worker: HashMap>> = + let mut script_path_to_worker: HashMap>> = HashMap::new(); for module in modules.iter() { let value = module.get_value(); @@ -447,7 +450,7 @@ pub async fn create_dedicated_worker_map( worker_name: &str, job_completed_tx: &JobCompletedSender, ) -> ( - HashMap>>, + HashMap>>, bool, Vec>, ) { @@ -566,6 +569,7 @@ async fn spawn_dedicated_worker( scripts::{ScriptHash, ScriptLang}, utils::rd_string, }; + use windmill_queue::MiniPulledJob; use crate::{build_envs, get_script_content_by_hash, ContentReqLangEnvs, JOB_TOKEN}; @@ -578,8 +582,9 @@ async fn spawn_dedicated_worker( #[cfg(feature = "enterprise")] { - let (dedicated_worker_tx, dedicated_worker_rx) = - tokio::sync::mpsc::channel::>(MAX_BUFFERED_DEDICATED_JOBS); + let (dedicated_worker_tx, dedicated_worker_rx) = tokio::sync::mpsc::channel::< + std::sync::Arc, + >(MAX_BUFFERED_DEDICATED_JOBS); let killpill_rx = killpill_rx.resubscribe(); let db = db.clone(); let base_internal_url = base_internal_url.to_string(); diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 40350fdeee..2bd979571c 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -3,7 +3,7 @@ use std::{collections::HashMap, process::Stdio}; use itertools::Itertools; use serde_json::value::RawValue; use uuid::Uuid; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::{append_logs, CanceledBy, MiniPulledJob}; use crate::{ common::{ @@ -15,11 +15,8 @@ use crate::{ NPM_CONFIG_REGISTRY, PATH_ENV, TZ_ENV, }; use tokio::{fs::File, io::AsyncReadExt, process::Command}; +use windmill_common::error::{self}; use windmill_common::{error::Result, worker::write_file, BASE_URL}; -use windmill_common::{ - error::{self}, - jobs::QueuedJob, -}; use windmill_parser::Typ; lazy_static::lazy_static! { @@ -180,7 +177,7 @@ pub async fn handle_deno_job( requirements_o: Option<&String>, mem_peak: &mut i32, canceled_by: &mut Option, - job: &QueuedJob, + job: &MiniPulledJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, job_dir: &str, @@ -196,7 +193,7 @@ pub async fn handle_deno_job( append_logs(&job.id, &job.workspace_id, logs1, db).await; let main_override = job.script_entrypoint_override.as_deref(); - let apply_preprocessor = !job.is_flow_step && job.preprocessed == Some(false); + let apply_preprocessor = !job.is_flow_step() && job.preprocessed == Some(false); write_file(job_dir, "main.ts", inner_content)?; @@ -310,7 +307,7 @@ try {{ let write_import_map_f = build_import_map( &job.workspace_id, - job.script_path(), + job.runnable_path(), base_internal_url, job_dir, ); @@ -502,7 +499,7 @@ pub async fn start_worker( script_path: &str, token: &str, job_completed_tx: JobCompletedSender, - jobs_rx: Receiver>, + jobs_rx: Receiver>, killpill_rx: tokio::sync::broadcast::Receiver<()>, db: &sqlx::Pool, ) -> Result<()> { diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index 1a8156a1bb..23aee36fa4 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -7,12 +7,11 @@ use tokio::{fs::File, io::AsyncReadExt, process::Command}; use uuid::Uuid; use windmill_common::{ error::{self, Error}, - jobs::QueuedJob, utils::calculate_hash, worker::{save_cache, write_file}, }; use windmill_parser_go::{parse_go_imports, REQUIRE_PARSE}; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::{append_logs, CanceledBy, MiniPulledJob}; use crate::{ common::{ @@ -36,7 +35,7 @@ pub const GO_OBJECT_STORE_PREFIX: &str = "gobin/"; pub async fn handle_go_job( mem_peak: &mut i32, canceled_by: &mut Option, - job: &QueuedJob, + job: &MiniPulledJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, inner_content: &str, diff --git a/backend/windmill-worker/src/graphql_executor.rs b/backend/windmill-worker/src/graphql_executor.rs index 929235cccf..98e5ef9a32 100644 --- a/backend/windmill-worker/src/graphql_executor.rs +++ b/backend/windmill-worker/src/graphql_executor.rs @@ -4,11 +4,10 @@ use anyhow::anyhow; use futures::{stream, TryStreamExt}; use serde_json::{json, value::RawValue}; use sqlx::types::Json; -use windmill_common::jobs::QueuedJob; use windmill_common::worker::to_raw_value; use windmill_common::{error::Error, worker::CLOUD_HOSTED}; use windmill_parser_graphql::parse_graphql_sig; -use windmill_queue::CanceledBy; +use windmill_queue::{CanceledBy, MiniPulledJob}; use serde::Deserialize; @@ -35,7 +34,7 @@ struct GraphqlError { } pub async fn do_graphql( - job: &QueuedJob, + job: &MiniPulledJob, client: &AuthedClientBackgroundTask, query: &str, db: &sqlx::Pool, diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index 9b77ca840f..4c5aa1c908 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -29,6 +29,7 @@ mod js_eval; mod mysql_executor; #[cfg(feature = "oracledb")] mod oracledb_executor; +mod otel_ee; mod pg_executor; #[cfg(feature = "php")] mod php_executor; diff --git a/backend/windmill-worker/src/mssql_executor.rs b/backend/windmill-worker/src/mssql_executor.rs index 42d9c3aa8c..a033d1eff1 100644 --- a/backend/windmill-worker/src/mssql_executor.rs +++ b/backend/windmill-worker/src/mssql_executor.rs @@ -8,10 +8,11 @@ use tiberius::{AuthMethod, Client, ColumnData, Config, FromSqlOwned, Query, Row, use tokio::net::TcpStream; use tokio_util::compat::TokioAsyncWriteCompatExt; use uuid::Uuid; +use windmill_common::error::to_anyhow; use windmill_common::error::{self, Error}; use windmill_common::worker::to_raw_value; -use windmill_common::{error::to_anyhow, jobs::QueuedJob}; use windmill_parser_sql::{parse_db_resource, parse_mssql_sig}; +use windmill_queue::MiniPulledJob; use windmill_queue::{append_logs, CanceledBy}; use crate::common::{build_args_values, OccupancyMetrics}; @@ -33,7 +34,7 @@ lazy_static::lazy_static! { } pub async fn do_mssql( - job: &QueuedJob, + job: &MiniPulledJob, client: &AuthedClientBackgroundTask, query: &str, db: &sqlx::Pool, diff --git a/backend/windmill-worker/src/mysql_executor.rs b/backend/windmill-worker/src/mysql_executor.rs index d733901f93..80c41f21c3 100644 --- a/backend/windmill-worker/src/mysql_executor.rs +++ b/backend/windmill-worker/src/mysql_executor.rs @@ -12,7 +12,6 @@ use sqlx::types::Json; use tokio::sync::Mutex; use windmill_common::{ error::{to_anyhow, Error}, - jobs::QueuedJob, worker::to_raw_value, }; use windmill_parser_sql::{ @@ -20,6 +19,7 @@ use windmill_parser_sql::{ RE_ARG_MYSQL_NAMED, }; use windmill_queue::CanceledBy; +use windmill_queue::MiniPulledJob; use crate::{ common::{build_args_map, OccupancyMetrics}, @@ -103,7 +103,7 @@ pub fn do_mysql_inner<'a>( } pub async fn do_mysql( - job: &QueuedJob, + job: &MiniPulledJob, client: &AuthedClientBackgroundTask, query: &str, db: &sqlx::Pool, diff --git a/backend/windmill-worker/src/oracledb_executor.rs b/backend/windmill-worker/src/oracledb_executor.rs index e6292181b2..3127659b02 100644 --- a/backend/windmill-worker/src/oracledb_executor.rs +++ b/backend/windmill-worker/src/oracledb_executor.rs @@ -11,9 +11,10 @@ use serde_json::{json, value::RawValue, Value}; use sqlx::types::Json; use windmill_common::{ error::{to_anyhow, Error}, - jobs::QueuedJob, worker::to_raw_value, }; +use windmill_queue::MiniPulledJob; + use windmill_parser_sql::{ parse_db_resource, parse_oracledb_sig, parse_sql_blocks, parse_sql_statement_named_params, }; @@ -292,7 +293,7 @@ fn get_statement_values( } pub async fn do_oracledb( - job: &QueuedJob, + job: &MiniPulledJob, client: &AuthedClientBackgroundTask, query: &str, db: &sqlx::Pool, diff --git a/backend/windmill-worker/src/otel_ee.rs b/backend/windmill-worker/src/otel_ee.rs new file mode 100644 index 0000000000..311ffd6355 --- /dev/null +++ b/backend/windmill-worker/src/otel_ee.rs @@ -0,0 +1,3 @@ +use windmill_queue::MiniPulledJob; + +pub fn add_root_flow_job_to_otlp(_queued_job: &MiniPulledJob, _success: bool) {} diff --git a/backend/windmill-worker/src/pg_executor.rs b/backend/windmill-worker/src/pg_executor.rs index 40d1103af0..645d7ff09f 100644 --- a/backend/windmill-worker/src/pg_executor.rs +++ b/backend/windmill-worker/src/pg_executor.rs @@ -25,14 +25,14 @@ use tokio_postgres::{ Column, }; use uuid::Uuid; +use windmill_common::error::to_anyhow; use windmill_common::error::{self, Error}; use windmill_common::worker::{to_raw_value, CLOUD_HOSTED}; -use windmill_common::{error::to_anyhow, jobs::QueuedJob}; use windmill_parser::{Arg, Typ}; use windmill_parser_sql::{ parse_db_resource, parse_pg_statement_arg_indices, parse_pgsql_sig, parse_sql_blocks, }; -use windmill_queue::CanceledBy; +use windmill_queue::{CanceledBy, MiniPulledJob}; use crate::common::{build_args_values, sizeof_val, OccupancyMetrics}; use crate::handle_child::run_future_with_polling_update_job_poller; @@ -157,7 +157,7 @@ fn do_postgresql_inner<'a>( } pub async fn do_postgresql( - job: &QueuedJob, + job: &MiniPulledJob, client: &AuthedClientBackgroundTask, query: &str, db: &sqlx::Pool, diff --git a/backend/windmill-worker/src/php_executor.rs b/backend/windmill-worker/src/php_executor.rs index 4f579a54a8..fa5eb58c16 100644 --- a/backend/windmill-worker/src/php_executor.rs +++ b/backend/windmill-worker/src/php_executor.rs @@ -7,9 +7,10 @@ use tokio::{fs::File, io::AsyncReadExt, process::Command}; use uuid::Uuid; use windmill_common::{ error::{self, to_anyhow, Result}, - jobs::QueuedJob, worker::write_file, }; +use windmill_queue::MiniPulledJob; + use windmill_parser::Typ; use windmill_queue::{append_logs, CanceledBy}; @@ -136,7 +137,7 @@ pub async fn handle_php_job( requirements_o: Option<&String>, mem_peak: &mut i32, canceled_by: &mut Option, - job: &QueuedJob, + job: &MiniPulledJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, job_dir: &str, diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 1180652d35..d39580d285 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -26,7 +26,6 @@ use windmill_common::{ self, Error::{self}, }, - jobs::QueuedJob, utils::calculate_hash, worker::{write_file, PythonAnnotations, WORKER_CONFIG}, DB, @@ -709,7 +708,7 @@ pub async fn uv_pip_compile( async fn postinstall( additional_python_paths: &mut Vec, job_dir: &str, - job: &QueuedJob, + job: &MiniPulledJob, db: &sqlx::Pool, ) -> windmill_common::error::Result<()> { // It is guranteed that additional_python_paths only contains paths within windmill/cache/ @@ -836,7 +835,7 @@ pub async fn handle_python_job( job_dir: &str, worker_dir: &str, worker_name: &str, - job: &QueuedJob, + job: &MiniPulledJob, mem_peak: &mut i32, canceled_by: &mut Option, db: &sqlx::Pool, @@ -848,7 +847,7 @@ pub async fn handle_python_job( new_args: &mut Option>>, occupancy_metrics: &mut OccupancyMetrics, ) -> windmill_common::error::Result> { - let script_path = crate::common::use_flow_root_path(job.script_path()); + let script_path = crate::common::use_flow_root_path(job.runnable_path()); let (py_version, mut additional_python_paths) = handle_python_deps( job_dir, @@ -911,7 +910,7 @@ pub async fn handle_python_job( pre_spread, ) = prepare_wrapper( job_dir, - job.is_flow_step, + job.is_flow_step(), job.preprocessed, job.script_entrypoint_override.as_deref(), inner_content, @@ -2298,6 +2297,8 @@ use crate::{common::build_envs_map, dedicated_worker::handle_dedicated_process}; #[cfg(feature = "enterprise")] use windmill_common::variables; +use windmill_queue::MiniPulledJob; + #[cfg(feature = "enterprise")] pub async fn start_worker( requirements_o: Option<&String>, @@ -2311,7 +2312,7 @@ pub async fn start_worker( script_path: &str, token: &str, job_completed_tx: JobCompletedSender, - jobs_rx: tokio::sync::mpsc::Receiver>, + jobs_rx: tokio::sync::mpsc::Receiver>, killpill_rx: tokio::sync::broadcast::Receiver<()>, ) -> error::Result<()> { let mut mem_peak: i32 = 0; diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 153d6f32fc..966991d0ea 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -19,7 +19,7 @@ use uuid::Uuid; use windmill_common::{ add_time, error::{self, Error}, - jobs::{JobKind, QueuedJob}, + jobs::JobKind, utils::WarnAfterExt, worker::{to_raw_value, WORKER_GROUP}, KillpillSender, DB, @@ -28,7 +28,7 @@ use windmill_common::{ #[cfg(feature = "benchmark")] use crate::bench::{BenchmarkInfo, BenchmarkIter}; -use windmill_queue::{append_logs, get_queued_job, CanceledBy, WrappedError}; +use windmill_queue::{append_logs, get_queued_job, CanceledBy, MiniPulledJob, WrappedError}; use serde_json::{json, value::RawValue}; @@ -42,6 +42,7 @@ use windmill_queue::{add_completed_job, add_completed_job_error}; use crate::{ bash_executor::ANSI_ESCAPE_RE, common::{read_result, save_in_cache}, + otel_ee::add_root_flow_job_to_otlp, worker_flow::update_flow_status_after_job_completion, AuthedClient, JobCompleted, JobCompletedSender, SameWorkerSender, SendResult, INIT_SCRIPT_TAG, }; @@ -81,7 +82,7 @@ pub fn start_background_processor( let is_init_script_and_failure = !jc.success && jc.job.tag.as_str() == INIT_SCRIPT_TAG; let is_dependency_job = matches!( - jc.job.job_kind, + jc.job.kind, JobKind::Dependencies | JobKind::FlowDependencies ); @@ -98,14 +99,14 @@ pub fn start_background_processor( parent_job = field::Empty, otel.name = field::Empty ); - let rj = if let Some(root_job) = jc.job.root_job { + let rj = if let Some(root_job) = jc.job.flow_innermost_root_job { root_job } else { jc.job.id }; windmill_common::otel_ee::set_span_parent(&span, &rj); - if let Some(lg) = jc.job.language.as_ref() { + if let Some(lg) = jc.job.script_lang.as_ref() { span.record("language", lg.as_str()); } if let Some(step_id) = jc.job.flow_step_id.as_ref() { @@ -120,10 +121,10 @@ pub fn start_background_processor( if let Some(parent_job) = jc.job.parent_job.as_ref() { span.record("parent_job", parent_job.to_string().as_str()); } - if let Some(script_path) = jc.job.script_path.as_ref() { + if let Some(script_path) = jc.job.runnable_path.as_ref() { span.record("script_path", script_path.as_str()); } - if let Some(root_job) = jc.job.root_job.as_ref() { + if let Some(root_job) = jc.job.flow_innermost_root_job.as_ref() { span.record("root_job", root_job.to_string().as_str()); } @@ -142,7 +143,7 @@ pub fn start_background_processor( .await; if let Some(root_job) = root_job { - windmill_common::otel_ee::add_root_flow_job_to_otlp(&root_job, success); + add_root_flow_job_to_otlp(&root_job, success); } if is_init_script_and_failure { @@ -227,7 +228,7 @@ pub fn start_background_processor( async fn send_job_completed( job_completed_tx: JobCompletedSender, - job: Arc, + job: Arc, result: Arc>, result_columns: Option>, mem_peak: i32, @@ -256,7 +257,7 @@ async fn send_job_completed( } pub async fn process_result( - job: Arc, + job: Arc, result: error::Result>>, job_dir: &str, job_completed_tx: JobCompletedSender, @@ -356,7 +357,7 @@ pub async fn handle_receive_completed_job( worker_name: &str, job_completed_tx: Sender, #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, -) -> Option> { +) -> Option> { let token = jc.token.clone(); let workspace = jc.job.workspace_id.clone(); let client = AuthedClient { @@ -423,14 +424,14 @@ pub async fn process_completed_job( worker_name: &str, job_completed_tx: Sender, #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, -) -> error::Result>> { +) -> error::Result>> { if success { // println!("bef completed job{:?}", SystemTime::now()); if let Some(cached_path) = cached_res_path { save_in_cache(db, client, &job, cached_path, result.clone()).await; } - let is_flow_step = job.is_flow_step; + let is_flow_step = job.is_flow_step(); let parent_job = job.parent_job.clone(); let job_id = job.id.clone(); let workspace_id = job.workspace_id.clone(); @@ -511,7 +512,7 @@ pub async fn process_completed_job( None, ) .await?; - if job.is_flow_step { + if job.is_flow_step() { if let Some(parent_job) = job.parent_job { tracing::error!(parent_flow = %parent_job, subflow = %job.id, "process completed job error, updating flow status"); let r = update_flow_status_after_job_completion( @@ -544,7 +545,7 @@ pub async fn process_completed_job( pub async fn handle_job_error( db: &Pool, client: &AuthedClient, - job: &QueuedJob, + job: &MiniPulledJob, mem_peak: i32, canceled_by: Option, err: Error, @@ -581,7 +582,7 @@ pub async fn handle_job_error( .await }; - let update_job_future = if job.is_flow_step || job.is_flow() { + let update_job_future = if job.is_flow_step() || job.is_flow() { let (flow, job_status_to_update) = if let Some(parent_job_id) = job.parent_job { if let Err(e) = update_job_future().await { tracing::error!( @@ -630,7 +631,7 @@ pub async fn handle_job_error( .await; let _ = add_completed_job_error( db, - &parent_job, + &MiniPulledJob::from(&parent_job), mem_peak, canceled_by.clone(), e, diff --git a/backend/windmill-worker/src/rust_executor.rs b/backend/windmill-worker/src/rust_executor.rs index 95fd822100..b3fbb08d35 100644 --- a/backend/windmill-worker/src/rust_executor.rs +++ b/backend/windmill-worker/src/rust_executor.rs @@ -7,10 +7,10 @@ use itertools::Itertools; use tokio::{fs::File, io::AsyncReadExt, process::Command}; use windmill_common::{ error::{self, Error}, - jobs::QueuedJob, utils::calculate_hash, worker::{save_cache, write_file}, }; +use windmill_queue::MiniPulledJob; use windmill_queue::{append_logs, CanceledBy}; use crate::{ @@ -275,7 +275,7 @@ pub fn compute_rust_hash(code: &str, requirements_o: Option<&String>) -> String pub async fn handle_rust_job( mem_peak: &mut i32, canceled_by: &mut Option, - job: &QueuedJob, + job: &MiniPulledJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, inner_content: &str, diff --git a/backend/windmill-worker/src/snowflake_executor.rs b/backend/windmill-worker/src/snowflake_executor.rs index 3315df6bc1..7090c114e8 100644 --- a/backend/windmill-worker/src/snowflake_executor.rs +++ b/backend/windmill-worker/src/snowflake_executor.rs @@ -10,10 +10,9 @@ use sha2::{Digest, Sha256}; use std::collections::HashMap; use windmill_common::error::to_anyhow; -use windmill_common::jobs::QueuedJob; use windmill_common::{error::Error, worker::to_raw_value}; use windmill_parser_sql::{parse_db_resource, parse_snowflake_sig, parse_sql_blocks}; -use windmill_queue::{CanceledBy, HTTP_CLIENT}; +use windmill_queue::{CanceledBy, MiniPulledJob, HTTP_CLIENT}; use serde::{Deserialize, Serialize}; @@ -240,7 +239,7 @@ fn do_snowflake_inner<'a>( } pub async fn do_snowflake( - job: &QueuedJob, + job: &MiniPulledJob, client: &AuthedClientBackgroundTask, query: &str, db: &sqlx::Pool, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 7c474efd40..3ae9a1e683 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -56,7 +56,7 @@ use windmill_common::{ cache::{self, RawData}, error::{self, to_anyhow, Error}, flows::FlowNodeId, - jobs::{JobKind, QueuedJob}, + jobs::JobKind, scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang, PREVIEW_IS_CODEBASE_HASH}, users::SUPERADMIN_SECRET_EMAIL, utils::StripPath, @@ -65,8 +65,8 @@ use windmill_common::{ }; use windmill_queue::{ - append_logs, canceled_job_to_result, empty_result, pull, push, CanceledBy, PulledJob, PushArgs, - PushIsolationLevel, HTTP_CLIENT, + append_logs, canceled_job_to_result, empty_result, pull, push, CanceledBy, MiniPulledJob, + PulledJob, PushArgs, PushIsolationLevel, HTTP_CLIENT, }; #[cfg(feature = "prometheus")] @@ -154,7 +154,7 @@ use windmill_common::add_time; pub async fn create_token_for_owner_in_bg( db: &Pool, - job: &QueuedJob, + job: &MiniPulledJob, ) -> Arc> { let rw_lock = Arc::new(RwLock::new(String::new())); // skipping test runs @@ -163,7 +163,7 @@ pub async fn create_token_for_owner_in_bg( let db = db.clone(); let w_id = job.workspace_id.clone(); let owner = job.permissioned_as.clone(); - let email = job.email.clone(); + let email = job.permissioned_as_email.clone(); let job_id = job.id.clone(); let label = if job.permissioned_as != format!("u/{}", job.created_by) @@ -707,7 +707,7 @@ async fn insert_wait_time( } fn add_outstanding_wait_time( - queued_job: &QueuedJob, + queued_job: &MiniPulledJob, db: &Pool, waiting_threshold: i64, ) -> () { @@ -724,7 +724,7 @@ fn add_outstanding_wait_time( } let job_id = queued_job.id; - let root_job_id = queued_job.root_job; + let root_job_id = queued_job.flow_innermost_root_job; let db = db.clone(); tokio::spawn(async move { @@ -1083,7 +1083,7 @@ pub async fn run_worker( #[cfg(feature = "enterprise")] let (dedicated_workers, is_flow_worker, dedicated_handles): ( - HashMap>>, + HashMap>>, bool, Vec>, ) = create_dedicated_worker_map( @@ -1099,7 +1099,7 @@ pub async fn run_worker( #[cfg(not(feature = "enterprise"))] let (dedicated_workers, is_flow_worker, dedicated_handles): ( - HashMap>>, + HashMap>>, bool, Vec>, ) = (HashMap::new(), false, vec![]); @@ -1299,11 +1299,48 @@ pub async fn run_worker( same_worker_job.job_id ); let r = sqlx::query_as::<_, PulledJob>( - " - WITH ping AS ( - UPDATE v2_job_runtime SET ping = NOW() WHERE id = $1 RETURNING id + "WITH ping AS ( + UPDATE v2_job_runtime SET ping = NOW() WHERE id = $1 + ), + started_at AS ( + UPDATE v2_job_queue SET started_at = NOW() WHERE id = $1 ) - SELECT * FROM v2_as_queue WHERE id = (SELECT id FROM ping) + SELECT + v2_job_queue.workspace_id, + v2_job_queue.id, + v2_job.args, + v2_job.parent_job, + v2_job.created_by, + v2_job_queue.started_at, + scheduled_for, + runnable_path, + kind, + runnable_id, + canceled_reason, + canceled_by, + permissioned_as, + permissioned_as_email, + flow_status, + v2_job.tag, + script_lang, + same_worker, + pre_run_error, + concurrent_limit, + concurrency_time_window_s, + flow_innermost_root_job, + timeout, + flow_step_id, + cache_ttl, + v2_job_queue.priority, + preprocessed, + script_entrypoint_override, + trigger, + trigger_kind, + visible_to_owner, + raw_code, + raw_lock, + raw_flow + FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1 ", ) .bind(same_worker_job.job_id) @@ -1315,12 +1352,7 @@ pub async fn run_worker( same_worker_job.job_id, e )) }); - let _ = sqlx::query!( - "UPDATE v2_job_queue SET started_at = NOW() WHERE id = $1", - same_worker_job.job_id - ) - .execute(db) - .await; + // tracing::error!("r: {:?}", r); if r.is_err() && !same_worker_job.recoverable { tracing::error!( worker = %worker_name, hostname = %hostname, @@ -1443,12 +1475,12 @@ pub async fn run_worker( tracing::debug!(worker = %worker_name, hostname = %hostname, "started handling of job {}", job.id); - if matches!(job.job_kind, JobKind::Script | JobKind::Preview) { + if matches!(job.kind, JobKind::Script | JobKind::Preview) { if !dedicated_workers.is_empty() { let key_o = if is_flow_worker { job.flow_step_id.as_ref().map(|x| x.to_string()) } else { - job.script_path.as_ref().map(|x| x.to_string()) + job.runnable_path.as_ref().map(|x| x.to_string()) }; if let Some(key) = key_o { if let Some(dedicated_worker_tx) = dedicated_workers.get(&key) { @@ -1467,7 +1499,7 @@ pub async fn run_worker( } } } - if matches!(job.job_kind, JobKind::Noop) { + if matches!(job.kind, JobKind::Noop) { add_time!(bench, "send job completed START"); job_completed_tx .send(JobCompleted { @@ -1533,7 +1565,7 @@ pub async fn run_worker( .await; let job_root = job - .root_job + .flow_innermost_root_job .map(|x| x.to_string()) .unwrap_or_else(|| "none".to_string()); @@ -1554,7 +1586,7 @@ pub async fn run_worker( let same_worker = job.same_worker; - let folder = if job.language == Some(ScriptLang::Go) { + let folder = if job.script_lang == Some(ScriptLang::Go) { DirBuilder::new() .recursive(true) .create(&format!("{job_dir}/go")) @@ -1605,12 +1637,12 @@ pub async fn run_worker( language = field::Empty, script_path = field::Empty, flow_step_id = field::Empty, parent_job = field::Empty, otel.name = field::Empty); - let rj = if let Some(root_job) = arc_job.root_job { + let rj = if let Some(root_job) = arc_job.flow_innermost_root_job { root_job } else { arc_job.id }; - if let Some(lg) = arc_job.language.as_ref() { + if let Some(lg) = arc_job.script_lang.as_ref() { span.record("language", lg.as_str()); } if let Some(step_id) = arc_job.flow_step_id.as_ref() { @@ -1622,10 +1654,10 @@ pub async fn run_worker( if let Some(parent_job) = arc_job.parent_job.as_ref() { span.record("parent_job", parent_job.to_string().as_str()); } - if let Some(script_path) = arc_job.script_path.as_ref() { + if let Some(script_path) = arc_job.runnable_path.as_ref() { span.record("script_path", script_path.as_str()); } - if let Some(root_job) = arc_job.root_job.as_ref() { + if let Some(root_job) = arc_job.flow_innermost_root_job.as_ref() { span.record("root_job", root_job.to_string().as_str()); } @@ -1869,7 +1901,7 @@ pub enum SendResult { #[derive(Debug, Clone)] pub struct JobCompleted { - pub job: Arc, + pub job: Arc, pub result: Arc>, pub result_columns: Option>, pub mem_peak: i32, @@ -1881,7 +1913,7 @@ pub struct JobCompleted { } async fn do_nativets( - job: &QueuedJob, + job: &MiniPulledJob, client: &AuthedClientBackgroundTask, env_code: String, code: String, @@ -1923,7 +1955,7 @@ pub struct PreviousResult<'a> { } async fn handle_queued_job( - job: Arc, + job: Arc, raw_code: Option, raw_lock: Option, raw_flow: Option>>, @@ -1942,7 +1974,7 @@ async fn handle_queued_job( ) -> windmill_common::error::Result { // Extract the active span from the context - if job.canceled { + if job.canceled_by.is_some() { return Err(Error::JsonErr(canceled_job_to_result(&job))); } if let Some(e) = &job.pre_run_error { @@ -1981,7 +2013,7 @@ async fn handle_queued_job( } } - if job.is_flow_step { + if job.is_flow_step() { let _ = update_flow_status_in_progress( db, &job.workspace_id, @@ -2022,7 +2054,7 @@ async fn handle_queued_job( // Pre-fetch preview jobs raw values if necessary. // The `raw_*` values passed to this function are the original raw values from `queue` tables, // they are kept for backward compatibility as they have been moved to the `job` table. - let preview_data = match (job.job_kind, job.script_hash) { + let preview_data = match (job.kind, job.runnable_id) { ( JobKind::Preview | JobKind::Dependencies @@ -2085,7 +2117,7 @@ async fn handle_queued_job( let flow_data = match preview_data { Some(RawData::Flow(data)) => data, // Not a preview: fetch from the cache or the database. - _ => cache::job::fetch_flow(db, job.job_kind, job.script_hash).await?, + _ => cache::job::fetch_flow(db, job.kind, job.runnable_id).await?, }; handle_flow( job, @@ -2140,7 +2172,7 @@ async fn handle_queued_job( let mut column_order: Option> = None; let mut new_args: Option>> = None; - let result = match job.job_kind { + let result = match job.kind { JobKind::Dependencies => { handle_dependency_job( &job, @@ -2327,7 +2359,7 @@ pub async fn get_script_content_by_hash( #[tracing::instrument(level = "trace", skip_all)] async fn handle_code_execution_job( - job: &QueuedJob, + job: &MiniPulledJob, preview: Option>, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, @@ -2343,7 +2375,7 @@ async fn handle_code_execution_job( killpill_rx: &mut tokio::sync::broadcast::Receiver<()>, ) -> error::Result> { let script_hash = || { - job.script_hash + job.runnable_id .ok_or_else(|| Error::internal_err("expected script hash")) }; let (arc_data, arc_metadata, data, metadata): ( @@ -2352,11 +2384,9 @@ async fn handle_code_execution_job( ScriptData, ScriptMetadata, ); - let (ScriptData { code, lock }, ScriptMetadata { language, envs, codebase }) = match job - .job_kind - { + let (ScriptData { code, lock }, ScriptMetadata { language, envs, codebase }) = match job.kind { JobKind::Preview => { - let codebase = match job.script_hash.map(|x| x.0) { + let codebase = match job.runnable_id.map(|x| x.0) { Some(PREVIEW_IS_CODEBASE_HASH) => Some(job.id.to_string()), Some(PREVIEW_IS_TAR_CODEBASE_HASH) => Some(format!("{}.tar", job.id)), _ => None, @@ -2364,12 +2394,13 @@ async fn handle_code_execution_job( arc_data = preview.ok_or_else(|| Error::internal_err("expected preview".to_string()))?; - metadata = ScriptMetadata { language: job.language, codebase, envs: None }; + metadata = ScriptMetadata { language: job.script_lang, codebase, envs: None }; (arc_data.as_ref(), &metadata) } JobKind::Script_Hub => { let ContentReqLangEnvs { content, lockfile, language, envs, codebase } = - get_hub_script_content_and_requirements(job.script_path.as_ref(), Some(db)).await?; + get_hub_script_content_and_requirements(job.runnable_path.as_ref(), Some(db)) + .await?; data = ScriptData { code: content, lock: lockfile }; metadata = ScriptMetadata { language, envs, codebase }; (&data, &metadata) @@ -2380,17 +2411,17 @@ async fn handle_code_execution_job( } JobKind::FlowScript => { arc_data = cache::flow::fetch_script(db, FlowNodeId(script_hash()?.0)).await?; - metadata = ScriptMetadata { language: job.language, envs: None, codebase: None }; + metadata = ScriptMetadata { language: job.script_lang, envs: None, codebase: None }; (arc_data.as_ref(), &metadata) } JobKind::AppScript => { arc_data = cache::app::fetch_script(db, AppScriptId(script_hash()?.0)).await?; - metadata = ScriptMetadata { language: job.language, envs: None, codebase: None }; + metadata = ScriptMetadata { language: job.script_lang, envs: None, codebase: None }; (arc_data.as_ref(), &metadata) } JobKind::DeploymentCallback => { let script_path = job - .script_path + .runnable_path .as_ref() .ok_or_else(|| Error::internal_err("expected script path".to_string()))?; if script_path.starts_with("hub/") { @@ -2419,7 +2450,7 @@ async fn handle_code_execution_job( ), }; - let language = *language; + let language = language.clone(); if language == Some(ScriptLang::Postgresql) { return do_postgresql( job, @@ -2614,7 +2645,7 @@ async fn handle_code_execution_job( } let lang_str = job - .language + .script_lang .as_ref() .map(|x| format!("{x:?}")) .unwrap_or_else(|| "NO_LANG".to_string()); @@ -2626,8 +2657,8 @@ async fn handle_code_execution_job( job.id ); - let shared_mount = if job.same_worker && job.language != Some(ScriptLang::Deno) { - let folder = if job.language == Some(ScriptLang::Go) { + let shared_mount = if job.same_worker && job.script_lang != Some(ScriptLang::Deno) { + let folder = if job.script_lang == Some(ScriptLang::Go) { "/go" } else { "" diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 50f6ee8607..6a8544d722 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -40,7 +40,7 @@ use windmill_common::flow_status::{ use windmill_common::flows::{add_virtual_items_if_necessary, Branch, FlowNodeId}; use windmill_common::jobs::{ script_hash_to_tag_and_limits, script_path_to_payload, JobKind, JobPayload, OnBehalfOf, - QueuedJob, RawCode, ENTRYPOINT_OVERRIDE, + RawCode, ENTRYPOINT_OVERRIDE, }; use windmill_common::scripts::ScriptHash; use windmill_common::users::username_to_permissioned_as; @@ -56,8 +56,9 @@ use windmill_common::{ }; use windmill_queue::schedule::get_schedule_opt; use windmill_queue::{ - add_completed_job, add_completed_job_error, append_logs, handle_maybe_scheduled_job, - CanceledBy, PushArgs, PushIsolationLevel, WrappedError, + add_completed_job, add_completed_job_error, append_logs, get_mini_pulled_job, + handle_maybe_scheduled_job, CanceledBy, MiniPulledJob, PushArgs, PushIsolationLevel, + WrappedError, }; type DB = sqlx::Pool; @@ -82,7 +83,7 @@ pub async fn update_flow_status_after_job_completion( worker_name: &str, job_completed_tx: Sender, #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, -) -> error::Result>> { +) -> error::Result>> { // this is manual tailrecursion because async_recursion blows up the stack potentially_crash_for_testing(); @@ -166,7 +167,7 @@ pub async fn update_flow_status_after_job_completion( pub enum UpdateFlowStatusAfterJobCompletion { Rec(RecUpdateFlowStatusAfterJobCompletion), - Done(Arc), + Done(Arc), NotDone, NonLastParallelBranch, } @@ -217,11 +218,11 @@ pub async fn update_flow_status_after_job_completion_internal( let (job_kind, script_hash, old_status, raw_flow) = sqlx::query!( "SELECT - job_kind AS \"job_kind!: JobKind\", - script_hash AS \"script_hash: ScriptHash\", + kind AS \"job_kind!: JobKind\", + runnable_id AS \"script_hash: ScriptHash\", flow_status AS \"flow_status!: Json>\", raw_flow AS \"raw_flow: Json>\" - FROM v2_as_queue WHERE id = $1 AND workspace_id = $2 LIMIT 1", + FROM v2_job INNER JOIN v2_job_status ON v2_job.id = v2_job_status.id WHERE v2_job.id = $1 AND v2_job.workspace_id = $2 LIMIT 1", flow, w_id ) @@ -811,6 +812,13 @@ pub async fn update_flow_status_after_job_completion_internal( old_status.step }; + // tracing::error!( + // "step_counter: {:?} {} {inc_step_counter} {flow}", + // step_counter, + // old_status.step, + // ); + // panic!("stop"); + /* is_last_step is true when the step_counter (the next step index) is an invalid index */ let is_last_step = usize::try_from(step_counter) .map(|i| !(..old_status.modules.len()).contains(&i)) @@ -962,26 +970,20 @@ pub async fn update_flow_status_after_job_completion_internal( .context("remove flow status retry")?; } - let flow_job = sqlx::query_as::<_, QueuedJob>( - "SELECT * FROM v2_as_queue WHERE id = $1 AND workspace_id = $2", - ) - .bind(flow) - .bind(w_id) - .fetch_optional(&mut *tx) - .await - .map_err(Into::::into)? - .ok_or_else(|| Error::internal_err(format!("requiring flow to be in the queue")))?; + let flow_job = get_mini_pulled_job(&mut *tx, &flow) + .await? + .ok_or_else(|| Error::internal_err(format!("requiring flow to be in the queue")))?; tx.commit().await?; let job_root = flow_job - .root_job + .flow_innermost_root_job .map(|x| x.to_string()) .unwrap_or_else(|| "none".to_string()); tracing::info!(id = %flow_job.id, root_id = %job_root, "update flow status"); let should_continue_flow = match success { _ if stop_early => false, - _ if flow_job.canceled => false, + _ if flow_job.is_canceled() => false, true => !is_last_step, false if unrecoverable => false, false if skip_branch_failure || skip_loop_failures || continue_on_error => { @@ -1039,7 +1041,7 @@ pub async fn update_flow_status_after_job_completion_internal( let done = if !should_continue_flow { { - let logs = if flow_job.canceled { + let logs = if flow_job.is_canceled() { "Flow job canceled\n".to_string() } else if stop_early { format!("Flow job stopped early because of a stop early predicate returning true\n") @@ -1078,7 +1080,7 @@ pub async fn update_flow_status_after_job_completion_internal( })?; } } - if flow_job.canceled { + if flow_job.is_canceled() { add_completed_job_error( db, &flow_job, @@ -1182,7 +1184,7 @@ pub async fn update_flow_status_after_job_completion_internal( let _ = tokio::fs::remove_dir_all(format!("{worker_dir}/{}", flow_job.id)).await; } - if flow_job.is_flow_step { + if flow_job.is_flow_step() { if let Some(parent_job) = flow_job.parent_job { tracing::info!(subflow_id = %flow_job.id, parent_id = %parent_job, "subflow is finished, updating parent flow status"); @@ -1539,7 +1541,7 @@ async fn transform_input( #[instrument(level = "trace", skip_all)] pub async fn handle_flow( - flow_job: Arc, + flow_job: Arc, flow_data: &cache::FlowData, db: &sqlx::Pool, client: &AuthedClient, @@ -1554,13 +1556,14 @@ pub async fn handle_flow( .parse_flow_status() .with_context(|| "Unable to parse flow status")?; - if !flow_job.is_flow_step + let schedule_path = flow_job.schedule_path(); + if !flow_job.is_flow_step() && status.retry.fail_count == 0 - && flow_job.schedule_path.is_some() - && flow_job.script_path.is_some() + && schedule_path.is_some() + && flow_job.runnable_path.is_some() && status.step == 0 { - let schedule_path = flow_job.schedule_path.as_ref().unwrap(); + let schedule_path = schedule_path.as_ref().unwrap(); let schedule = get_schedule_opt(db, &flow_job.workspace_id, schedule_path) .warn_after_seconds(5) @@ -1571,7 +1574,7 @@ pub async fn handle_flow( db, &flow_job, &schedule, - flow_job.script_path.as_ref().unwrap(), + flow_job.runnable_path.as_ref().unwrap(), &flow_job.workspace_id, ) .warn_after_seconds(5) @@ -1652,13 +1655,13 @@ lazy_static::lazy_static! { } struct PushNextFlowJobRec { - flow_job: Arc, + flow_job: Arc, status: FlowStatus, } // #[async_recursion] // #[instrument(level = "trace", skip_all)] async fn push_next_flow_job( - flow_job: Arc, + flow_job: Arc, mut status: FlowStatus, flow: &FlowValue, db: &sqlx::Pool, @@ -1670,7 +1673,7 @@ async fn push_next_flow_job( worker_name: &str, ) -> error::Result> { let job_root = flow_job - .root_job + .flow_innermost_root_job .map(|x| x.to_string()) .unwrap_or_else(|| "none".to_string()); tracing::info!(id = %flow_job.id, root_id = %job_root, "pushing next flow job"); @@ -1690,7 +1693,7 @@ async fn push_next_flow_job( Step::FailureStep => status.failure_module.module_status.clone(), }; - let fj: mappable_rc::Marc = flow_job.clone().into(); + let fj: mappable_rc::Marc = flow_job.clone().into(); let arc_flow_job_args: Marc>> = Marc::map(fj, |x| { if let Some(args) = &x.args { &args.0 @@ -1727,10 +1730,11 @@ async fn push_next_flow_job( } if matches!(step, Step::Step(0)) { - if !flow_job.is_flow_step && flow_job.schedule_path.is_some() { + if !flow_job.is_flow_step() && flow_job.schedule_path().is_some() { + let schedule_path = flow_job.schedule_path(); let no_flow_overlap = sqlx::query_scalar!( "SELECT no_flow_overlap FROM schedule WHERE path = $1 AND workspace_id = $2", - flow_job.schedule_path.as_ref().unwrap(), + schedule_path.as_ref().unwrap(), flow_job.workspace_id.as_str() ) .fetch_one(db) @@ -1748,10 +1752,10 @@ async fn push_next_flow_job( AND parent_job IS NULL AND j.id != $3 AND running = true", - flow_job.schedule_path.as_ref().unwrap(), + schedule_path.as_ref().unwrap(), flow_job.workspace_id.as_str(), flow_job.id, - flow_job.script_path.as_ref().unwrap() + flow_job.runnable_path() ) .fetch_all(db) .await?; @@ -1966,7 +1970,7 @@ async fn push_next_flow_job( .permissioned_as .trim_start_matches("u/") .to_string(), - email: flow_job.email.clone(), + email: flow_job.permissioned_as_email.clone(), username_override: None, }; @@ -2056,9 +2060,8 @@ async fn push_next_flow_job( sqlx::query!( "UPDATE v2_job_runtime SET ping = NULL - WHERE id = $1 AND ping = $2", + WHERE id = $1", flow_job.id, - flow_job.last_ping ) .execute(&mut *tx) .await?; @@ -2464,8 +2467,8 @@ async fn push_next_flow_job( // Also check `flow_job.same_worker` for [`JobKind::Flow`] jobs as it's no // more reflected to the flow value on push. let job_same_worker = flow_job.same_worker - && matches!(flow_job.job_kind, JobKind::Flow) - && flow_job.script_hash.is_some(); + && matches!(flow_job.kind, JobKind::Flow) + && flow_job.runnable_id.is_some(); let continue_on_same_worker = (flow.same_worker || job_same_worker) && module.suspend.is_none() && module.sleep.is_none(); @@ -2635,12 +2638,17 @@ async fn push_next_flow_job( } { None } else { - flow_job.root_job.or_else(|| Some(flow_job.id)) + flow_job + .flow_innermost_root_job + .or_else(|| Some(flow_job.id)) }; // forward root job permissions to the new job let job_perms: Option = if JOB_TOKEN.is_none() { - if let Some(root_job) = &flow_job.root_job.or_else(|| Some(flow_job.id)) { + if let Some(root_job) = &flow_job + .flow_innermost_root_job + .or_else(|| Some(flow_job.id)) + { sqlx::query_as!( JobPerms, "SELECT * FROM job_perms WHERE job_id = $1 AND workspace_id = $2", @@ -2670,7 +2678,10 @@ async fn push_next_flow_job( { (&on_behalf_of.email, on_behalf_of.permissioned_as.clone()) } else { - (&flow_job.email, flow_job.permissioned_as.to_owned()) + ( + &flow_job.permissioned_as_email, + flow_job.permissioned_as.to_owned(), + ) }; let tx2 = PushIsolationLevel::Transaction(tx); let (uuid, mut inner_tx) = push( @@ -2683,7 +2694,7 @@ async fn push_next_flow_job( email, permissioned_as, scheduled_for_o, - flow_job.schedule_path.clone(), + flow_job.schedule_path(), Some(flow_job.id), root_job, None, @@ -3119,22 +3130,22 @@ fn payload_from_modules<'a>( }) } -fn get_path(flow_job: &QueuedJob, status: &FlowStatus, module: &FlowModule) -> String { +fn get_path(flow_job: &MiniPulledJob, status: &FlowStatus, module: &FlowModule) -> String { if status .preprocessor_module .as_ref() .is_some_and(|x| x.id() == module.id) { - format!("{}/preprocessor", flow_job.script_path()) + format!("{}/preprocessor", flow_job.runnable_path()) } else { - format!("{}/{}", flow_job.script_path(), module.id) + format!("{}/{}", flow_job.runnable_path(), module.id) } } async fn compute_next_flow_transform( arc_flow_job_args: Marc>>, arc_last_job_result: Arc>, - flow_job: &QueuedJob, + flow_job: &MiniPulledJob, flow: &FlowValue, by_id: Option, db: &DB, @@ -3299,7 +3310,7 @@ async fn compute_next_flow_transform( /* forloop modules are expected set `iter: { value: Value, index: usize }` as job arguments */ FlowModuleValue::ForloopFlow { modules, modules_node, iterator, parallel, .. } => { // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg - let is_simple = !matches!(flow_job.job_kind, JobKind::FlowPreview) + let is_simple = !matches!(flow_job.kind, JobKind::FlowPreview) && !parallel && is_simple_modules(&modules, flow.failure_module.as_ref()); @@ -3370,7 +3381,7 @@ async fn compute_next_flow_transform( flow.failure_module.as_ref(), flow.same_worker, || format!("{}-{i}", status.step), - || format!("{}/forloop-{i}", flow_job.script_path()), + || format!("{}/forloop-{i}", flow_job.runnable_path()), true, ) else { return None; @@ -3460,7 +3471,7 @@ async fn compute_next_flow_transform( flow.failure_module.as_ref(), flow.same_worker, || status.step.to_string(), - || format!("{}/branchone-{}", flow_job.script_path(), branch_idx), + || format!("{}/branchone-{}", flow_job.runnable_path(), branch_idx), true, ) else { return Ok(NextFlowTransform::EmptyInnerFlows { branch_chosen: Some(branch) }); @@ -3496,7 +3507,7 @@ async fn compute_next_flow_transform( flow.failure_module.as_ref(), flow.same_worker, || format!("{}-{i}", status.step), - || format!("{}/branchall-{}", flow_job.script_path(), i), + || format!("{}/branchall-{}", flow_job.runnable_path(), i), false, ) else { return None; @@ -3563,7 +3574,7 @@ async fn compute_next_flow_transform( || { format!( "{}/branchall-{}", - flow_job.script_path(), + flow_job.runnable_path(), branch_status.branch ) }, @@ -3598,13 +3609,13 @@ async fn next_loop_iteration( ns: ForloopNextIteration, modules: Vec, modules_node: Option, - flow_job: &QueuedJob, + flow_job: &MiniPulledJob, is_simple: bool, db: &sqlx::Pool, module: &FlowModule, delete_after_use: bool, ) -> Result { - let inner_path = || format!("{}/loop-{}", flow_job.script_path(), ns.index); + let inner_path = || format!("{}/loop-{}", flow_job.runnable_path(), ns.index); if is_simple { let mut value = modules[0].get_value()?; let simple_input_transforms = match &mut value { @@ -3669,7 +3680,7 @@ pub(super) fn is_simple_modules( async fn next_forloop_status( status_module: &FlowStatusModule, by_id: Option, - flow_job: &QueuedJob, + flow_job: &MiniPulledJob, previous_id: &str, status: &FlowStatus, iterator: &InputTransform, @@ -3808,7 +3819,7 @@ async fn next_forloop_status( async fn payload_from_simple_module( value: FlowModuleValue, db: &sqlx::Pool, - flow_job: &QueuedJob, + flow_job: &MiniPulledJob, module: &FlowModule, inner_path: String, ) -> Result { @@ -3929,7 +3940,7 @@ async fn script_to_payload( script_hash: Option, script_path: String, db: &sqlx::Pool, - flow_job: &QueuedJob, + flow_job: &MiniPulledJob, module: &FlowModule, tag_override: Option, ) -> Result { @@ -4003,7 +4014,7 @@ async fn script_to_payload( } async fn get_transform_context( - flow_job: &QueuedJob, + flow_job: &MiniPulledJob, previous_id: &str, status: &FlowStatus, ) -> error::Result { diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 7bf0929448..a2679913da 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -22,7 +22,6 @@ use windmill_common::{ cache::{self, RawData}, error::{self, to_anyhow}, flows::{add_virtual_items_if_necessary, FlowValue}, - jobs::QueuedJob, scripts::ScriptLang, DB, }; @@ -30,7 +29,7 @@ use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; #[cfg(feature = "python")] use windmill_parser_py_imports::parse_relative_imports; use windmill_parser_ts::parse_expr_for_imports; -use windmill_queue::{append_logs, CanceledBy, PushIsolationLevel}; +use windmill_queue::{append_logs, CanceledBy, MiniPulledJob, PushIsolationLevel}; use crate::common::OccupancyMetrics; use crate::csharp_executor::generate_nuget_lockfile; @@ -217,7 +216,7 @@ pub fn extract_relative_imports( } #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_dependency_job( - job: &QueuedJob, + job: &MiniPulledJob, preview_data: Option<&RawData>, mem_peak: &mut i32, canceled_by: &mut Option, @@ -229,7 +228,7 @@ pub async fn handle_dependency_job( token: &str, occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result> { - let script_path = job.script_path(); + let script_path = job.runnable_path(); let raw_deps = job .args .as_ref() @@ -239,7 +238,7 @@ pub async fn handle_dependency_job( }) .unwrap_or(false); let npm_mode = if job - .language + .script_lang .as_ref() .map(|v| v == &ScriptLang::Bun) .unwrap_or(false) @@ -260,7 +259,7 @@ pub async fn handle_dependency_job( // `JobKind::Dependencies` job store either: // - A saved script `hash` in the `script_hash` column. // - Preview raw lock and code in the `queue` or `job` table. - let script_data = match job.script_hash { + let script_data = match job.runnable_id { Some(hash) => &cache::script::fetch(db, hash).await?.0, _ => match preview_data { Some(RawData::Script(data)) => data, @@ -269,7 +268,7 @@ pub async fn handle_dependency_job( }; let content = capture_dependency_job( &job.id, - job.language.as_ref().map(|v| Ok(v)).unwrap_or_else(|| { + job.script_lang.as_ref().map(|v| Ok(v)).unwrap_or_else(|| { Err(Error::internal_err( "Job Language required for dependency jobs".to_owned(), )) @@ -293,14 +292,14 @@ pub async fn handle_dependency_job( match content { Ok(content) => { - if job.script_hash.is_none() { + if job.runnable_id.is_none() { // it a one-off raw script dependency job, no need to update the db return Ok(to_raw_value_owned( json!({ "status": "Successful lock file generation", "lock": content }), )); } - let hash = job.script_hash.unwrap_or(ScriptHash(0)); + let hash = job.runnable_id.unwrap_or(ScriptHash(0)); let w_id = &job.workspace_id; sqlx::query!( "UPDATE script SET lock = $1 WHERE hash = $2 AND workspace_id = $3", @@ -318,7 +317,7 @@ pub async fn handle_dependency_job( get_deployment_msg_and_parent_path_from_args(job.args.clone()); if let Err(e) = handle_deployment_metadata( - &job.email, + &job.permissioned_as_email, &job.created_by, &db, &w_id, @@ -336,7 +335,7 @@ pub async fn handle_dependency_job( } let relative_imports = - extract_relative_imports(&script_data.code, script_path, &job.language); + extract_relative_imports(&script_data.code, script_path, &job.script_lang); if let Some(relative_imports) = relative_imports { update_script_dependency_map( &job.id, @@ -362,7 +361,7 @@ pub async fn handle_dependency_job( script_path, deployment_message, parent_path, - &job.email, + &job.permissioned_as_email, &job.created_by, &job.permissioned_as, db, @@ -391,7 +390,7 @@ pub async fn handle_dependency_job( sqlx::query!( "UPDATE script SET lock_error_logs = $1 WHERE hash = $2 AND workspace_id = $3", &format!("{logs2}\n{error}"), - &job.script_hash.unwrap_or(ScriptHash(0)).0, + &job.runnable_id.unwrap_or(ScriptHash(0)).0, &job.workspace_id ) .execute(db) @@ -546,7 +545,7 @@ async fn trigger_dependents_to_recompute_dependencies( } pub async fn handle_flow_dependency_job( - job: &QueuedJob, + job: &MiniPulledJob, preview_data: Option<&RawData>, mem_peak: &mut i32, canceled_by: &mut Option, @@ -558,7 +557,7 @@ pub async fn handle_flow_dependency_job( token: &str, occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result> { - let job_path = job.script_path.clone().ok_or_else(|| { + let job_path = job.runnable_path.clone().ok_or_else(|| { error::Error::internal_err( "Cannot resolve flow dependencies for flow without path".to_string(), ) @@ -579,7 +578,7 @@ pub async fn handle_flow_dependency_job( None } else { Some( - job.script_hash + job.runnable_id .clone() .ok_or_else(|| { Error::internal_err( @@ -606,7 +605,7 @@ pub async fn handle_flow_dependency_job( // `JobKind::FlowDependencies` job store either: // - A saved flow version `id` in the `script_hash` column. // - Preview raw flow in the `queue` or `job` table. - let mut flow = match job.script_hash { + let mut flow = match job.runnable_id { Some(ScriptHash(id)) => cache::flow::fetch_version(db, id).await?, _ => match preview_data { Some(RawData::Flow(data)) => data.clone(), @@ -664,7 +663,7 @@ pub async fn handle_flow_dependency_job( sqlx::query!( "UPDATE flow SET lock_error_logs = $1 WHERE path = $2 AND workspace_id = $3", &format!("{logs2}\n{error_message}"), - &job.script_path(), + &job.runnable_path(), &job.workspace_id ) .execute(db) @@ -677,7 +676,7 @@ pub async fn handle_flow_dependency_job( } else { sqlx::query!( "UPDATE flow SET lock_error_logs = NULL WHERE path = $1 AND workspace_id = $2", - &job.script_path(), + &job.runnable_path(), &job.workspace_id ) .execute(db) @@ -746,7 +745,7 @@ pub async fn handle_flow_dependency_job( tx.commit().await?; if let Err(e) = handle_deployment_metadata( - &job.email, + &job.permissioned_as_email, &job.created_by, &db, &job.workspace_id, @@ -799,7 +798,7 @@ struct LockModuleError { async fn lock_modules<'c>( modules: Vec, - job: &QueuedJob, + job: &MiniPulledJob, mem_peak: &mut i32, canceled_by: &mut Option, job_dir: &str, @@ -1402,7 +1401,7 @@ fn skip_creating_new_lock(language: &ScriptLang, content: &str) -> bool { #[async_recursion] async fn lock_modules_app( value: Value, - job: &QueuedJob, + job: &MiniPulledJob, mem_peak: &mut i32, canceled_by: &mut Option, job_dir: &str, @@ -1454,7 +1453,7 @@ async fn lock_modules_app( worker_dir, base_internal_url, token, - &format!("{}/app", job.script_path()), + &format!("{}/app", job.runnable_path()), false, None, occupancy_metrics, @@ -1550,7 +1549,7 @@ async fn lock_modules_app( } pub async fn handle_app_dependency_job( - job: &QueuedJob, + job: &MiniPulledJob, mem_peak: &mut i32, canceled_by: &mut Option, job_dir: &str, @@ -1561,14 +1560,14 @@ pub async fn handle_app_dependency_job( token: &str, occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result<()> { - let job_path = job.script_path.clone().ok_or_else(|| { + let job_path = job.runnable_path.clone().ok_or_else(|| { error::Error::internal_err( "Cannot resolve app dependencies for app without path".to_string(), ) })?; let id = job - .script_hash + .runnable_id .clone() .ok_or_else(|| Error::internal_err("App Dependency requires script hash".to_owned()))? .0; @@ -1632,7 +1631,7 @@ pub async fn handle_app_dependency_job( get_deployment_msg_and_parent_path_from_args(job.args.clone()); if let Err(e) = handle_deployment_metadata( - &job.email, + &job.permissioned_as_email, &job.created_by, &db, &job.workspace_id, diff --git a/frontend/src/lib/components/runs/JobLoader.svelte b/frontend/src/lib/components/runs/JobLoader.svelte index 5789cdc76d..e924151f53 100644 --- a/frontend/src/lib/components/runs/JobLoader.svelte +++ b/frontend/src/lib/components/runs/JobLoader.svelte @@ -18,6 +18,7 @@ export let jobs: Job[] | undefined export let user: string | null export let label: string | null = null + export let worker: string | null = null export let folder: string | null export let path: string | null export let success: 'success' | 'suspended' | 'waiting' | 'failure' | 'running' | undefined = @@ -61,6 +62,7 @@ (path && label && success && + worker && isSkipped != undefined && jobKinds && concurrencyKey && @@ -173,6 +175,7 @@ hasNullParent: jobKindsCat != 'all' ? true : undefined, label: label === null || label === '' ? undefined : label, tag: tag === null || tag === '' ? undefined : tag, + worker: worker === null || worker === '' ? undefined : worker, isNotSchedule: showSchedules == false ? true : undefined, suspended: success == 'waiting' ? false : success == 'suspended' ? true : undefined, scheduledForBeforeNow: diff --git a/frontend/src/lib/components/runs/JobPreview.svelte b/frontend/src/lib/components/runs/JobPreview.svelte index 2dcf08946f..e22adf235b 100644 --- a/frontend/src/lib/components/runs/JobPreview.svelte +++ b/frontend/src/lib/components/runs/JobPreview.svelte @@ -106,7 +106,7 @@ {#if concurrencyKey} - This jobs has concurrency limits enabled with the key: + This job has concurrency limits enabled with the key: + + Worker: {truncateRev(job.worker, 20)} + + {/if} @@ -151,7 +159,8 @@ { label: 'Schedule path', value: 'schedulePath' }, { label: 'Concurrency key', value: 'concurrencyKey' }, { label: 'Label', value: 'label' }, - { label: 'Tag', value: 'tag' } + { label: 'Tag', value: 'tag' }, + { label: 'Worker', value: 'worker' } ]} {item} bind:selected={filterBy} @@ -415,6 +424,40 @@ /> {/key} + {:else if filterBy === 'worker'} + {#key worker} +
+ {#if worker} + + {/if} + Worker + + + { + if (workerTimeout) { + clearTimeout(workerTimeout) + } + + workerTimeout = setTimeout(() => { + worker = displayedWorker + }, 1000) + }} + /> +
+ {/key} {/if}
@@ -549,6 +592,7 @@ + @@ -745,6 +789,41 @@
{/key} + {:else if filterBy === 'worker'} + {#key worker} + + {/key} {/if} + {/if} {/if} diff --git a/frontend/src/routes/(root)/(logged)/runs/[...path]/+page.svelte b/frontend/src/routes/(root)/(logged)/runs/[...path]/+page.svelte index 3b8ef2aa29..d58d2e8c91 100644 --- a/frontend/src/routes/(root)/(logged)/runs/[...path]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/runs/[...path]/+page.svelte @@ -47,6 +47,7 @@ // All Filters // Filter by let path: string | null = $page.params.path + let worker: string | null = $page.url.searchParams.get('worker') let user: string | null = $page.url.searchParams.get('user') let folder: string | null = $page.url.searchParams.get('folder') let label: string | null = $page.url.searchParams.get('label') @@ -99,6 +100,7 @@ label = $page.url.searchParams.get('label') concurrencyKey = $page.url.searchParams.get('concurrency_key') tag = $page.url.searchParams.get('tag') + worker = $page.url.searchParams.get('worker') // Rest of filters handled by RunsFilter success = ($page.url.searchParams.get('success') ?? undefined) as | 'running' @@ -179,6 +181,7 @@ let runsTable: RunsTable $: (user || + worker || label || folder || path || @@ -209,6 +212,12 @@ searchParams.delete('user') } + if (worker) { + searchParams.set('worker', worker) + } else { + searchParams.delete('worker') + } + if (folder) { searchParams.set('folder', folder) } else { @@ -369,6 +378,7 @@ concurrencyKey = null tag = null schedulePath = undefined + worker = null } function filterByUser(e: CustomEvent) { @@ -389,6 +399,7 @@ concurrencyKey = null tag = null schedulePath = undefined + worker = null } function filterByLabel(e: CustomEvent) { @@ -399,6 +410,7 @@ concurrencyKey = null tag = null schedulePath = undefined + worker = null } function filterByConcurrencyKey(e: CustomEvent) { @@ -409,6 +421,7 @@ concurrencyKey = e.detail tag = null schedulePath = undefined + worker = null } function filterByTag(e: CustomEvent) { @@ -419,6 +432,7 @@ concurrencyKey = null tag = e.detail schedulePath = undefined + worker = null } function filterBySchedule(e: CustomEvent) { @@ -429,6 +443,18 @@ concurrencyKey = null tag = null schedulePath = e.detail + worker = null + } + + function filterByWorker(e: CustomEvent) { + path = null + user = null + folder = null + label = null + concurrencyKey = null + tag = null + schedulePath = undefined + worker = e.detail } let calendarChangeTimeout: NodeJS.Timeout | undefined = undefined @@ -538,6 +564,7 @@ schedulePath = undefined path = null tag = null + worker = null if (success == f) { success = undefined } else { @@ -555,6 +582,7 @@ {user} {folder} {path} + {worker} {label} {success} {isSkipped} @@ -683,6 +711,7 @@ bind:label bind:concurrencyKey bind:tag + bind:worker bind:path bind:success bind:argFilter @@ -1011,6 +1040,7 @@ on:filterByConcurrencyKey={filterByConcurrencyKey} on:filterByTag={filterByTag} on:filterBySchedule={filterBySchedule} + on:filterByWorker={filterByWorker} bind:this={runsTable} /> {:else} @@ -1028,6 +1058,7 @@ {:else} @@ -1070,6 +1101,7 @@ bind:path bind:user bind:label + bind:worker bind:concurrencyKey bind:tag bind:success @@ -1390,6 +1422,7 @@ on:filterByFolder={filterByFolder} on:filterByLabel={filterByLabel} on:filterByConcurrencyKey={filterByConcurrencyKey} + on:filterByWorker={filterByWorker} on:filterByTag={filterByTag} bind:this={runsTable} />