starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-21 08:02:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: filter by worker + backend perf opt (#5489)

Browse Source 
* all

* all

* all

* refactor

* sqlx

* ref

* impr

* fix

* fix

* fix

* ee private
This commit is contained in:
Ruben Fiszel
2025-03-19 09:55:53 +01:00
committed by GitHub
parent e7868f097a
commit aaa4cde349
47 changed files with 1212 additions and 516 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0aa8f50fe377a23e2ae3821fd6eda4cbe80b0150827c6e90f6e4f9e512587ba1.json
Generated
-14
View File
@@ -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"
}
backend/.sqlx/query-4aaab98ebdaa90f1edf49ac96fba6c391c4d0054a618b861464ee37239f1f1e0.json
Generated
+273
View File
@@ -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<HashMap<String, Box<RawValue>>>\",\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<Box<RawValue>>\",\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<HashMap<String, Box<RawValue>>>",
"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<Box<RawValue>>",
"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"
}
backend/.sqlx/query-5ff7df54c7908a7de494ddae5fc7bb9be8106a79e0683cd34459585bbd920ce4.json
Generated
-23
View File
@@ -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"
}
backend/.sqlx/query-a3ccf362b4f6df400b3c7a084795dbf541eb14c5c374656ffb96da7283a2a6f1.json
Generated
-23
View File
@@ -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"
}
backend/.sqlx/query-4535c8effd1bae49894d13293a37e1ee949cf9108239032cb3addbf350fb33de.json → backend/.sqlx/query-c50b6a4a6739d6df087a3b37c209e5f4b72fc27578d988155b74b05ec5df30b9.json
Generated
+3 -3
View File
@@ -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<Box<RawValue>>\",\n raw_flow AS \"raw_flow: Json<Box<RawValue>>\"\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<Box<RawValue>>\",\n raw_flow AS \"raw_flow: Json<Box<RawValue>>\"\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"
}
backend/.sqlx/query-2f2ef9b1ccff527c48fa01cf1b78cd0e58c8d534ac22ec0356d82a854b31d087.json → backend/.sqlx/query-ec1f31fd7628ea2e30995a0de1d8665831ee3e4ec3815e9ad90e886ffecba0f1.json
Generated
+3 -4
View File
@@ -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"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
280b866cd0b179c16c894197a891f24495a26549
94ff10ff18c121f49180ddaebf5dee8cd94d06df
backend/src/monitor.rs
+2 -2
View File
@@ -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<Postgres>, base_internal_url: &str, worker
let _ = handle_job_error(
db,
&client,
&job,
&MiniPulledJob::from(&job),
0,
None,
error::Error::ExecutionErr(error_message),
backend/windmill-api/openapi.yaml
+13
View File
@@ -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
backend/windmill-api/src/capture.rs
+1 -33
View File
@@ -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 {
backend/windmill-api/src/concurrency_groups.rs
+1
View File
@@ -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: _,
backend/windmill-api/src/jobs.rs
+150 -27
View File
@@ -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<Option<JobExtended<QueuedJob>>> {
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<Option<JobExtended<CompletedJob>>> {
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<CompletedJob>>(query)
.bind(job_id)
.bind(workspace_id)
@@ -1259,6 +1364,7 @@ pub struct ListQueueQuery {
pub order_desc: Option<bool>,
pub job_kinds: Option<String>,
pub suspended: Option<bool>,
pub worker: Option<String>,
// filter by matching a subset of the args using base64 encoded json subset
pub args: Option<String>,
pub tag: Option<String>,
@@ -1283,6 +1389,7 @@ impl From<ListCompletedQuery> 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<T> {
#[serde(skip_serializing_if = "Option::is_none")]
pub raw_flow: Option<sqlx::types::Json<Box<RawValue>>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub worker: Option<String>,
#[sqlx(skip)]
#[serde(skip_serializing_if = "Option::is_none")]
pub self_wait_time_ms: Option<i64>,
@@ -2463,6 +2577,7 @@ impl<T> JobExtended<T> {
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<i64>,
pub aggregate_wait_time_ms: Option<i64>,
pub preprocessed: Option<bool>,
pub worker: Option<String>,
}
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<String>,
pub is_not_schedule: Option<bool>,
pub concurrency_key: Option<String>,
pub worker: Option<String>,
}
async fn list_completed_jobs(
backend/windmill-api/src/mqtt_triggers.rs
+3 -1
View File
@@ -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},
backend/windmill-api/src/postgres_triggers/trigger.rs
+3 -1
View File
@@ -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};
backend/windmill-api/src/websocket_triggers.rs
+3 -1
View File
@@ -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,
backend/windmill-common/src/worker.rs
+12 -14
View File
@@ -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
backend/windmill-queue/src/jobs.rs
+295 -121
View File
@@ -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<Postgres>,
queued_job: &QueuedJob,
queued_job: &MiniPulledJob,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
e: serde_json::Value,
@@ -544,7 +547,7 @@ lazy_static::lazy_static! {
pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
db: &Pool<Postgres>,
queued_job: &QueuedJob,
queued_job: &MiniPulledJob,
success: bool,
skipped: bool,
result: Json<&T>,
@@ -581,7 +584,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
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<T: Serialize + Send + Sync + ValidableJson>(
.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<T: Serialize + Send + Sync + ValidableJson>(
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<T: Serialize + Send + Sync + ValidableJson>(
}
}
} 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<T: Serialize + Send + Sync + ValidableJson>(
db,
queued_job,
&schedule,
script_path,
&script_path,
&queued_job.workspace_id,
)
.await
@@ -760,7 +763,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
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<T: Serialize + Send + Sync + ValidableJson>(
}
}
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<T: Serialize + Send + Sync + ValidableJson>(
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<T: Serialize + Send + Sync + ValidableJson>(
"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<T: Serialize + Send + Sync + ValidableJson>(
#[cfg(feature = "enterprise")]
if !success {
async fn has_failure_module(db: &Pool<Postgres>, 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<Postgres>, 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<T: Serialize + Send + Sync + ValidableJson>(
.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<T: Serialize + Send + Sync + ValidableJson>(
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<T: Serialize + Send + Sync + ValidableJson>(
)
.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<T: Serialize + Send + Sync + ValidableJson>(
}
}
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<T: Serialize + Send + Sync + ValidableJson>(
{
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<T: Serialize + Send + Sync + ValidableJson>(
&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<T: Serialize + Send + Sync + ValidableJson>(
.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<T: Serialize + Send + Sync + ValidableJson>(
}
pub async fn send_error_to_global_handler<'a, T: Serialize + Send + Sync>(
queued_job: &QueuedJob,
queued_job: &MiniPulledJob,
db: &Pool<Postgres>,
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<Postgres>,
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<Postgres>,
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<bool> = match queued_job.job_kind {
let ws_error_handler_muted: Option<bool> = 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<Postgres>,
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<Json<HashMap<String, Box<RawValue>>>>,
pub parent_job: Option<Uuid>,
pub created_by: String,
pub scheduled_for: chrono::DateTime<chrono::Utc>,
pub started_at: Option<chrono::DateTime<chrono::Utc>>,
pub runnable_path: Option<String>,
pub kind: JobKind,
pub runnable_id: Option<ScriptHash>,
pub canceled_reason: Option<String>,
pub canceled_by: Option<String>,
pub permissioned_as: String,
pub permissioned_as_email: String,
pub flow_status: Option<Json<Box<RawValue>>>,
pub tag: String,
pub script_lang: Option<ScriptLang>,
pub same_worker: bool,
pub pre_run_error: Option<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
pub flow_innermost_root_job: Option<Uuid>,
pub timeout: Option<i32>,
pub flow_step_id: Option<String>,
pub cache_ttl: Option<i32>,
pub priority: Option<i16>,
pub preprocessed: Option<bool>,
pub script_entrypoint_override: Option<String>,
pub trigger: Option<String>,
pub trigger_kind: Option<JobTriggerKind>,
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<FlowStatus> {
// tracing::error!("parse_flow_status: {:?}", self.flow_status);
self.flow_status
.as_ref()
.and_then(|v| serde_json::from_str::<FlowStatus>((**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<String> {
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<String>,
pub raw_lock: Option<String>,
pub raw_flow: Option<Json<Box<RawValue>>>,
}
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<Option<MiniPulledJob>> {
let job = sqlx::query_as!(
MiniPulledJob,
"SELECT
v2_job_queue.workspace_id,
v2_job_queue.id,
v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",
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<Box<RawValue>>\",
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<Postgres>,
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<Postgres>,
job_id: Uuid,
job_id: &Uuid,
) -> Result<Option<String>, 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<Postgres>, queued_job: &QueuedJob) -> Option<String> {
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<Postgres>,
queued_job: &QueuedJob,
) -> windmill_common::error::Result<String> {
async fn concurrency_key(db: &Pool<Postgres>, id: &Uuid) -> windmill_common::error::Result<String> {
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()
backend/windmill-worker/src/ansible_executor.rs
+3 -2
View File
@@ -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<CanceledBy>,
db: &sqlx::Pool<sqlx::Postgres>,
backend/windmill-worker/src/bash_executor.rs
+4 -5
View File
@@ -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<CanceledBy>,
job: &QueuedJob,
job: &MiniPulledJob,
db: &sqlx::Pool<sqlx::Postgres>,
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<CanceledBy>,
job: &QueuedJob,
job: &MiniPulledJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &AuthedClientBackgroundTask,
content: &str,
backend/windmill-worker/src/bigquery_executor.rs
+3 -2
View File
@@ -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<sqlx::Postgres>,
backend/windmill-worker/src/bun_executor.rs
+8 -9
View File
@@ -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<CanceledBy>,
job: &QueuedJob,
job: &MiniPulledJob,
db: &sqlx::Pool<sqlx::Postgres>,
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<std::sync::Arc<QueuedJob>>,
jobs_rx: Receiver<std::sync::Arc<MiniPulledJob>>,
killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> Result<()> {
let mut logs = "".to_string();
backend/windmill-worker/src/common.rs
+24 -23
View File
@@ -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<Postgres>,
) -> error::Result<Option<HashMap<String, Box<RawValue>>>> {
@@ -73,12 +73,12 @@ pub fn check_executor_binary_exists(
}
pub async fn build_args_values(
job: &QueuedJob,
job: &MiniPulledJob,
client: &AuthedClientBackgroundTask,
db: &Pool<Postgres>,
) -> error::Result<HashMap<String, serde_json::Value>> {
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<Postgres>,
) -> Result<(), Error> {
@@ -129,7 +129,7 @@ pub async fn transform_json<'a>(
client: &AuthedClientBackgroundTask,
workspace: &str,
vs: &'a HashMap<String, Box<RawValue>>,
job: &QueuedJob,
job: &MiniPulledJob,
db: &Pool<Postgres>,
) -> error::Result<Option<HashMap<String, Box<RawValue>>>> {
let mut has_match = false;
@@ -168,7 +168,7 @@ pub async fn transform_json_as_values<'a>(
client: &AuthedClientBackgroundTask,
workspace: &str,
vs: &'a HashMap<String, Box<RawValue>>,
job: &QueuedJob,
job: &MiniPulledJob,
db: &Pool<Postgres>,
) -> error::Result<HashMap<String, serde_json::Value>> {
let mut r: HashMap<String, serde_json::Value> = HashMap::new();
@@ -238,7 +238,7 @@ pub async fn transform_json_value(
client: &AuthedClient,
workspace: &str,
v: Value,
job: &QueuedJob,
job: &MiniPulledJob,
db: &Pool<Postgres>,
) -> error::Result<Value> {
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<sqlx::Postgres>,
) -> Result<HashMap<String, String>, 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<Postgres>,
_client: &AuthedClient,
job: &QueuedJob,
job: &MiniPulledJob,
cached_path: String,
r: Arc<Box<RawValue>>,
) {
backend/windmill-worker/src/csharp_executor.rs
+4 -3
View File
@@ -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<usize>) -> 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<CanceledBy>,
_job: &QueuedJob,
_job: &MiniPulledJob,
_db: &sqlx::Pool<sqlx::Postgres>,
_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<CanceledBy>,
job: &QueuedJob,
job: &MiniPulledJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &AuthedClientBackgroundTask,
inner_content: &str,
backend/windmill-worker/src/dedicated_worker.rs
+14 -9
View File
@@ -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<std::sync::Arc<QueuedJob>>,
mut jobs_rx: Receiver<std::sync::Arc<MiniPulledJob>>,
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<Arc<QueuedJob>> = VecDeque::with_capacity(MAX_BUFFERED_DEDICATED_JOBS);
let mut jobs: VecDeque<Arc<MiniPulledJob>> =
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<QueuedJob> = jobs.pop_front().expect("pop");
let job: Arc<MiniPulledJob> = jobs.pop_front().expect("pop");
tracing::info!("job completed on dedicated worker {script_path}: {}", job.id);
match serde_json::from_str::<Box<serde_json::value::RawValue>>(&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<std::sync::Arc<QueuedJob>>,
Sender<std::sync::Arc<MiniPulledJob>>,
Option<JoinHandle<()>>,
);
@@ -262,7 +265,7 @@ async fn spawn_dedicated_workers_for_flow(
job_completed_tx: &JobCompletedSender,
) -> Vec<DedicatedWorker> {
let mut workers = vec![];
let mut script_path_to_worker: HashMap<String, Sender<std::sync::Arc<QueuedJob>>> =
let mut script_path_to_worker: HashMap<String, Sender<std::sync::Arc<MiniPulledJob>>> =
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<String, Sender<std::sync::Arc<QueuedJob>>>,
HashMap<String, Sender<std::sync::Arc<MiniPulledJob>>>,
bool,
Vec<JoinHandle<()>>,
) {
@@ -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::<std::sync::Arc<QueuedJob>>(MAX_BUFFERED_DEDICATED_JOBS);
let (dedicated_worker_tx, dedicated_worker_rx) = tokio::sync::mpsc::channel::<
std::sync::Arc<MiniPulledJob>,
>(MAX_BUFFERED_DEDICATED_JOBS);
let killpill_rx = killpill_rx.resubscribe();
let db = db.clone();
let base_internal_url = base_internal_url.to_string();
backend/windmill-worker/src/deno_executor.rs
+6 -9
View File
@@ -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<CanceledBy>,
job: &QueuedJob,
job: &MiniPulledJob,
db: &sqlx::Pool<sqlx::Postgres>,
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<std::sync::Arc<QueuedJob>>,
jobs_rx: Receiver<std::sync::Arc<MiniPulledJob>>,
killpill_rx: tokio::sync::broadcast::Receiver<()>,
db: &sqlx::Pool<sqlx::Postgres>,
) -> Result<()> {
backend/windmill-worker/src/go_executor.rs
+2 -3
View File
@@ -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<CanceledBy>,
job: &QueuedJob,
job: &MiniPulledJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &AuthedClientBackgroundTask,
inner_content: &str,
backend/windmill-worker/src/graphql_executor.rs
+2 -3
View File
@@ -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<sqlx::Postgres>,
backend/windmill-worker/src/lib.rs
+1
View File
@@ -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;
backend/windmill-worker/src/mssql_executor.rs
+3 -2
View File
@@ -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<sqlx::Postgres>,
backend/windmill-worker/src/mysql_executor.rs
+2 -2
View File
@@ -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<sqlx::Postgres>,
backend/windmill-worker/src/oracledb_executor.rs
+3 -2
View File
@@ -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<sqlx::Postgres>,
backend/windmill-worker/src/otel_ee.rs
+3
View File
@@ -0,0 +1,3 @@
use windmill_queue::MiniPulledJob;
pub fn add_root_flow_job_to_otlp(_queued_job: &MiniPulledJob, _success: bool) {}
backend/windmill-worker/src/pg_executor.rs
+3 -3
View File
@@ -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<sqlx::Postgres>,
backend/windmill-worker/src/php_executor.rs
+3 -2
View File
@@ -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<CanceledBy>,
job: &QueuedJob,
job: &MiniPulledJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &AuthedClientBackgroundTask,
job_dir: &str,
backend/windmill-worker/src/python_executor.rs
+7 -6
View File
@@ -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<String>,
job_dir: &str,
job: &QueuedJob,
job: &MiniPulledJob,
db: &sqlx::Pool<sqlx::Postgres>,
) -> 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<CanceledBy>,
db: &sqlx::Pool<sqlx::Postgres>,
@@ -848,7 +847,7 @@ pub async fn handle_python_job(
new_args: &mut Option<HashMap<String, Box<RawValue>>>,
occupancy_metrics: &mut OccupancyMetrics,
) -> windmill_common::error::Result<Box<RawValue>> {
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<std::sync::Arc<QueuedJob>>,
jobs_rx: tokio::sync::mpsc::Receiver<std::sync::Arc<MiniPulledJob>>,
killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> error::Result<()> {
let mut mem_peak: i32 = 0;
backend/windmill-worker/src/result_processor.rs
+18 -17
View File
@@ -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<QueuedJob>,
job: Arc<MiniPulledJob>,
result: Arc<Box<RawValue>>,
result_columns: Option<Vec<String>>,
mem_peak: i32,
@@ -256,7 +257,7 @@ async fn send_job_completed(
}
pub async fn process_result(
job: Arc<QueuedJob>,
job: Arc<MiniPulledJob>,
result: error::Result<Arc<Box<RawValue>>>,
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<SendResult>,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> Option<Arc<QueuedJob>> {
) -> Option<Arc<MiniPulledJob>> {
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<SendResult>,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> error::Result<Option<Arc<QueuedJob>>> {
) -> error::Result<Option<Arc<MiniPulledJob>>> {
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<Postgres>,
client: &AuthedClient,
job: &QueuedJob,
job: &MiniPulledJob,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
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,
backend/windmill-worker/src/rust_executor.rs
+2 -2
View File
@@ -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<CanceledBy>,
job: &QueuedJob,
job: &MiniPulledJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &AuthedClientBackgroundTask,
inner_content: &str,
backend/windmill-worker/src/snowflake_executor.rs
+2 -3
View File
@@ -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<sqlx::Postgres>,
backend/windmill-worker/src/worker.rs
+82 -51
View File
@@ -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<Postgres>,
job: &QueuedJob,
job: &MiniPulledJob,
) -> Arc<RwLock<String>> {
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<Postgres>,
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<String, Sender<Arc<QueuedJob>>>,
HashMap<String, Sender<Arc<MiniPulledJob>>>,
bool,
Vec<JoinHandle<()>>,
) = 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<String, Sender<Arc<QueuedJob>>>,
HashMap<String, Sender<Arc<MiniPulledJob>>>,
bool,
Vec<JoinHandle<()>>,
) = (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<QueuedJob>,
pub job: Arc<MiniPulledJob>,
pub result: Arc<Box<RawValue>>,
pub result_columns: Option<Vec<String>>,
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<QueuedJob>,
job: Arc<MiniPulledJob>,
raw_code: Option<String>,
raw_lock: Option<String>,
raw_flow: Option<Json<Box<RawValue>>>,
@@ -1942,7 +1974,7 @@ async fn handle_queued_job(
) -> windmill_common::error::Result<bool> {
// 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<Vec<String>> = None;
let mut new_args: Option<HashMap<String, Box<RawValue>>> = 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<Arc<ScriptData>>,
db: &sqlx::Pool<sqlx::Postgres>,
client: &AuthedClientBackgroundTask,
@@ -2343,7 +2375,7 @@ async fn handle_code_execution_job(
killpill_rx: &mut tokio::sync::broadcast::Receiver<()>,
) -> error::Result<Box<RawValue>> {
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 {
""
backend/windmill-worker/src/worker_flow.rs
+71 -60
View File
@@ -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<sqlx::Postgres>;
@@ -82,7 +83,7 @@ pub async fn update_flow_status_after_job_completion(
worker_name: &str,
job_completed_tx: Sender<SendResult>,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> error::Result<Option<Arc<QueuedJob>>> {
) -> error::Result<Option<Arc<MiniPulledJob>>> {
// 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<QueuedJob>),
Done(Arc<MiniPulledJob>),
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<Box<RawValue>>\",
raw_flow AS \"raw_flow: Json<Box<RawValue>>\"
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::<Error>::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<QueuedJob>,
flow_job: Arc<MiniPulledJob>,
flow_data: &cache::FlowData,
db: &sqlx::Pool<sqlx::Postgres>,
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<QueuedJob>,
flow_job: Arc<MiniPulledJob>,
status: FlowStatus,
}
// #[async_recursion]
// #[instrument(level = "trace", skip_all)]
async fn push_next_flow_job(
flow_job: Arc<QueuedJob>,
flow_job: Arc<MiniPulledJob>,
mut status: FlowStatus,
flow: &FlowValue,
db: &sqlx::Pool<sqlx::Postgres>,
@@ -1670,7 +1673,7 @@ async fn push_next_flow_job(
worker_name: &str,
) -> error::Result<Option<PushNextFlowJobRec>> {
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<QueuedJob> = flow_job.clone().into();
let fj: mappable_rc::Marc<MiniPulledJob> = flow_job.clone().into();
let arc_flow_job_args: Marc<HashMap<String, Box<RawValue>>> = 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<Authed> = 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<HashMap<String, Box<RawValue>>>,
arc_last_job_result: Arc<Box<RawValue>>,
flow_job: &QueuedJob,
flow_job: &MiniPulledJob,
flow: &FlowValue,
by_id: Option<IdContext>,
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<FlowModule>,
modules_node: Option<FlowNodeId>,
flow_job: &QueuedJob,
flow_job: &MiniPulledJob,
is_simple: bool,
db: &sqlx::Pool<sqlx::Postgres>,
module: &FlowModule,
delete_after_use: bool,
) -> Result<NextFlowTransform, Error> {
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<IdContext>,
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<sqlx::Postgres>,
flow_job: &QueuedJob,
flow_job: &MiniPulledJob,
module: &FlowModule,
inner_path: String,
) -> Result<JobPayloadWithTag, Error> {
@@ -3929,7 +3940,7 @@ async fn script_to_payload(
script_hash: Option<windmill_common::scripts::ScriptHash>,
script_path: String,
db: &sqlx::Pool<sqlx::Postgres>,
flow_job: &QueuedJob,
flow_job: &MiniPulledJob,
module: &FlowModule,
tag_override: Option<String>,
) -> Result<JobPayloadWithTag, Error> {
@@ -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<IdContext> {
backend/windmill-worker/src/worker_lockfiles.rs
+26 -27
View File
@@ -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<CanceledBy>,
@@ -229,7 +228,7 @@ pub async fn handle_dependency_job(
token: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<Box<RawValue>> {
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<CanceledBy>,
@@ -558,7 +557,7 @@ pub async fn handle_flow_dependency_job(
token: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<Box<serde_json::value::RawValue>> {
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<FlowModule>,
job: &QueuedJob,
job: &MiniPulledJob,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
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<CanceledBy>,
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<CanceledBy>,
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,
frontend/src/lib/components/runs/JobLoader.svelte
+3
View File
@@ -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:
frontend/src/lib/components/runs/JobPreview.svelte
+20 -1
View File
@@ -106,7 +106,7 @@
{#if concurrencyKey}
<Popover notClickable>
<svelte:fragment slot="text">
This jobs has concurrency limits enabled with the key:
This job has concurrency limits enabled with the key:
<Button
class="inline-text"
size="xs2"
@@ -122,6 +122,25 @@
<Badge large>Concurrency: {truncateRev(concurrencyKey, 20)}</Badge>
</Popover>
{/if}
{#if job?.worker}
<Popover notClickable>
<svelte:fragment slot="text">
This job was run on worker:
<Button
class="inline-text"
size="xs2"
color="light"
on:click={() => {
dispatch('filterByWorker', job?.worker)
}}
>
{job.worker}
<ListFilter class="inline-block" size={10} />
</Button>
</svelte:fragment>
<Badge large>Worker: {truncateRev(job.worker, 20)}</Badge>
</Popover>
{/if}
</div>
<a
href="{base}/run/{job?.id}?workspace={job?.workspace_id}"
frontend/src/lib/components/runs/RunsFilter.svelte
+81 -2
View File
@@ -26,6 +26,7 @@
export let path: string | null = null
export let label: string | null = null
export let concurrencyKey: string | null = null
export let worker: string | null = null
export let tag: string | null = null
export let success:
| 'running'
@@ -56,6 +57,7 @@
$: displayedConcurrencyKey = concurrencyKey
$: displayedTag = tag
$: displayedSchedule = schedulePath
$: displayedWorker = worker
let copyArgFilter = argFilter
let copyResultFilter = resultFilter
@@ -66,6 +68,7 @@
| 'folder'
| 'label'
| 'concurrencyKey'
| 'worker'
| 'tag'
| 'schedulePath' = 'path'
@@ -73,7 +76,8 @@
let autoSet = false
$: (path || user || folder || label || concurrencyKey || tag || schedulePath) && autosetFilter()
$: (path || user || folder || label || worker || concurrencyKey || tag || schedulePath) &&
autosetFilter()
function autosetFilter() {
if (path !== null && path !== '' && filterBy !== 'path') {
@@ -97,12 +101,16 @@
} else if (schedulePath !== undefined && schedulePath !== '' && filterBy !== 'schedulePath') {
autoSet = true
filterBy = 'schedulePath'
} else if (worker !== null && worker !== '' && filterBy !== 'worker') {
autoSet = true
filterBy = 'worker'
}
}
let labelTimeout: NodeJS.Timeout | undefined = undefined
let concurrencyKeyTimeout: NodeJS.Timeout | undefined = undefined
let tagTimeout: NodeJS.Timeout | undefined = undefined
let workerTimeout: NodeJS.Timeout | undefined = undefined
let allWorkspacesValue = allWorkspaces ? 'all' : 'admins'
</script>
@@ -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 @@
/>
</div>
{/key}
{:else if filterBy === 'worker'}
{#key worker}
<div class="relative">
{#if worker}
<button
class="absolute top-2 right-2 z-50"
on:click={() => {
worker = null
dispatch('reset')
}}
>
<X size={14} />
</button>
{/if}
<span class="text-xs absolute -top-4"> Worker </span>
<!-- svelte-ignore a11y-autofocus -->
<input
autofocus
type="text"
class="!h-[32px] py-1 !text-xs !w-64"
bind:value={displayedWorker}
on:keydown={(e) => {
if (workerTimeout) {
clearTimeout(workerTimeout)
}
workerTimeout = setTimeout(() => {
worker = displayedWorker
}, 1000)
}}
/>
</div>
{/key}
{/if}
</div>
<div class="relative">
@@ -549,6 +592,7 @@
<ToggleButton value="concurrencyKey" label="Concurrency" {item} />
<ToggleButton value="tag" label="Tag" {item} />
<ToggleButton value="label" label="Label" {item} />
<ToggleButton value="worker" label="Worker" {item} />
</ToggleButtonGroup>
</Label>
@@ -745,6 +789,41 @@
</div>
</Label>
{/key}
{:else if filterBy === 'worker'}
{#key worker}
<Label label="worker">
<div class="relative w-full">
{#if concurrencyKey}
<button
class="absolute top-2 right-2 z-50"
on:click={() => {
worker = null
// dispatch('reset')
}}
>
<X size={14} />
</button>
{/if}
<!-- svelte-ignore a11y-autofocus -->
<input
autofocus
type="text"
class="!h-[32px] py-1 !text-xs !w-80"
bind:value={displayedWorker}
on:keydown={(e) => {
if (workerTimeout) {
clearTimeout(workerTimeout)
}
workerTimeout = setTimeout(() => {
worker = displayedWorker
}, 1000)
}}
/>
</div>
</Label>
{/key}
{/if}
<Label label="Kind">
frontend/src/lib/components/runs/RunsTable.svelte
+1
View File
@@ -301,6 +301,7 @@
on:filterByFolder
on:filterByConcurrencyKey
on:filterBySchedule
on:filterByWorker
{containerWidth}
/>
</div>
frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte
+15
View File
@@ -799,6 +799,21 @@
</Tooltip>
</div>
{/if}
{#if job?.worker}
<div>
<Tooltip notClickable>
<svelte:fragment slot="text">
Executed on worker
<a href={`${base}/runs/?job_kinds=all&worker=${job?.worker}`}>
{job.worker}
</a>
</svelte:fragment>
<a href={`${base}/runs/?job_kinds=all&worker=${job?.worker}`}>
<Badge>Worker: {truncateRev(job?.worker, 20)}</Badge></a
>
</Tooltip>
</div>
{/if}
</div>
{/if}
</div>
frontend/src/routes/(root)/(logged)/runs/[...path]/+page.svelte
+33
View File
@@ -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<string>) {
@@ -389,6 +399,7 @@
concurrencyKey = null
tag = null
schedulePath = undefined
worker = null
}
function filterByLabel(e: CustomEvent<string>) {
@@ -399,6 +410,7 @@
concurrencyKey = null
tag = null
schedulePath = undefined
worker = null
}
function filterByConcurrencyKey(e: CustomEvent<string>) {
@@ -409,6 +421,7 @@
concurrencyKey = e.detail
tag = null
schedulePath = undefined
worker = null
}
function filterByTag(e: CustomEvent<string>) {
@@ -419,6 +432,7 @@
concurrencyKey = null
tag = e.detail
schedulePath = undefined
worker = null
}
function filterBySchedule(e: CustomEvent<string>) {
@@ -429,6 +443,18 @@
concurrencyKey = null
tag = null
schedulePath = e.detail
worker = null
}
function filterByWorker(e: CustomEvent<string>) {
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}
<JobPreview
on:filterByConcurrencyKey={filterByConcurrencyKey}
on:filterByWorker={filterByWorker}
id={selectedIds[0]}
workspace={selectedWorkspace}
/>
@@ -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}
/>