mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-26 00:03:59 +00:00
backend: rework queue::push after v2
This commit is contained in:
Abel Lucas
+21
@@ -0,0 +1,21 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO v2_job_queue (\n id, workspace_id, started_at, scheduled_for, running, created_at, tag, priority\n ) SELECT unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"UuidArray",
|
||||
"Varchar",
|
||||
"Timestamptz",
|
||||
"Timestamptz",
|
||||
"Bool",
|
||||
"Timestamptz",
|
||||
"Varchar",
|
||||
"Int2"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "5d0284028d0dc16fbbf353496a2204f9df256e03a588c23b33b4edab0d9c3e1a"
|
||||
}
|
||||
+47
@@ -0,0 +1,47 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO script (\n summary, description, dedicated_worker, content, workspace_id, path, hash,\n language, tag, created_by, lock\n ) VALUES ('', '', true, $1, $2, $3, $4, $5, $6, $7, '')\n ON CONFLICT (workspace_id, hash) DO NOTHING",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Varchar",
|
||||
"Varchar",
|
||||
"Int8",
|
||||
{
|
||||
"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"
|
||||
]
|
||||
}
|
||||
}
|
||||
},
|
||||
"Varchar",
|
||||
"Varchar"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "90d1c1ed47f1a9186cdde2f720fe86cde5ffefb4578d6e091855bc7a4e94d1b0"
|
||||
}
|
||||
+112
@@ -0,0 +1,112 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO v2_job (\n id, workspace_id,\n created_at, created_by, permissioned_as, permissioned_as_email,\n kind, runnable_id, runnable_path, parent_job, script_lang,\n flow_step_id, flow_root_job,\n trigger, trigger_kind,\n tag, same_worker, visible_to_owner, concurrent_limit, concurrency_time_window_s,\n cache_ttl, timeout, priority,\n args, pre_run_error,\n raw_code, raw_lock, raw_flow\n ) SELECT\n unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15,\n $16, $17, $18, $19, $20, $21, $22, $23, unnest($24::jsonb[]), $25, $26, $27, $28\n ",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"UuidArray",
|
||||
"Varchar",
|
||||
"Timestamptz",
|
||||
"Varchar",
|
||||
"Varchar",
|
||||
"Varchar",
|
||||
{
|
||||
"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"
|
||||
]
|
||||
}
|
||||
}
|
||||
},
|
||||
"Int8",
|
||||
"Varchar",
|
||||
"Uuid",
|
||||
{
|
||||
"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"
|
||||
]
|
||||
}
|
||||
}
|
||||
},
|
||||
"Varchar",
|
||||
"Uuid",
|
||||
"Varchar",
|
||||
{
|
||||
"Custom": {
|
||||
"name": "job_trigger_kind",
|
||||
"kind": {
|
||||
"Enum": [
|
||||
"webhook",
|
||||
"http",
|
||||
"websocket",
|
||||
"kafka",
|
||||
"email",
|
||||
"nats",
|
||||
"schedule",
|
||||
"app",
|
||||
"ui",
|
||||
"postgres"
|
||||
]
|
||||
}
|
||||
}
|
||||
},
|
||||
"Varchar",
|
||||
"Bool",
|
||||
"Bool",
|
||||
"Int4",
|
||||
"Int4",
|
||||
"Int4",
|
||||
"Int4",
|
||||
"Int2",
|
||||
"JsonbArray",
|
||||
"Text",
|
||||
"Text",
|
||||
"Text",
|
||||
"Jsonb"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "f8e0103dae8d80682a6b694762b6e4490d6004ba58ae26c52a4f21f867e1df70"
|
||||
}
|
||||
Generated
+1
@@ -11423,6 +11423,7 @@ dependencies = [
|
||||
"tokio-postgres 0.7.12",
|
||||
"tokio-util",
|
||||
"tracing",
|
||||
"ulid",
|
||||
"urlencoding",
|
||||
"uuid 1.12.1",
|
||||
"windmill-audit",
|
||||
|
||||
@@ -1641,7 +1641,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
|
||||
worker_name,
|
||||
send_result_never_used,
|
||||
#[cfg(feature = "benchmark")]
|
||||
&mut windmill_common::bench::BenchmarkIter::new(),
|
||||
&mut windmill_worker::bench::BenchmarkIter::new(),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
@@ -14,6 +14,7 @@ use quick_cache::sync::Cache;
|
||||
use serde_json::value::RawValue;
|
||||
use sqlx::Pool;
|
||||
use std::collections::HashMap;
|
||||
use std::iter;
|
||||
use std::ops::{Deref, DerefMut};
|
||||
use std::str::FromStr;
|
||||
#[cfg(feature = "prometheus")]
|
||||
@@ -87,7 +88,7 @@ use windmill_common::{METRICS_DEBUG_ENABLED, METRICS_ENABLED};
|
||||
use windmill_common::{get_latest_deployed_hash_for_path, BASE_URL};
|
||||
use windmill_queue::{
|
||||
cancel_job, get_result_and_success_by_id_from_flow, job_is_complete, push, PushArgs,
|
||||
PushArgsOwned, PushIsolationLevel,
|
||||
PushArgsOwned, PushIsolationLevel, RawJob,
|
||||
};
|
||||
|
||||
#[cfg(feature = "prometheus")]
|
||||
@@ -4678,7 +4679,7 @@ async fn add_batch_jobs(
|
||||
dedicated_worker,
|
||||
custom_concurrency_key,
|
||||
concurrent_limit,
|
||||
concurrent_time_window_s,
|
||||
concurrency_time_window_s,
|
||||
timeout,
|
||||
raw_code,
|
||||
raw_lock,
|
||||
@@ -4831,36 +4832,30 @@ async fn add_batch_jobs(
|
||||
};
|
||||
|
||||
let mut tx = user_db.begin(&authed).await?;
|
||||
|
||||
let uuids = sqlx::query_scalar!(
|
||||
r#"WITH uuid_table as (
|
||||
select gen_random_uuid() as uuid from generate_series(1, $16)
|
||||
)
|
||||
INSERT INTO v2_job
|
||||
(id, workspace_id, raw_code, raw_lock, raw_flow, tag, runnable_id, runnable_path, kind,
|
||||
script_lang, created_by, permissioned_as, permissioned_as_email, concurrent_limit,
|
||||
concurrency_time_window_s, timeout, args)
|
||||
(SELECT uuid, $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15,
|
||||
('{ "uuid": "' || uuid || '" }')::jsonb FROM uuid_table)
|
||||
RETURNING id AS "id!""#,
|
||||
w_id,
|
||||
raw_code,
|
||||
raw_lock,
|
||||
raw_flow.map(sqlx::types::Json) as Option<sqlx::types::Json<FlowValue>>,
|
||||
tag,
|
||||
hash.map(|h| h.0),
|
||||
path,
|
||||
job_kind.clone() as JobKind,
|
||||
language as ScriptLang,
|
||||
authed.username,
|
||||
username_to_permissioned_as(&authed.username),
|
||||
authed.email,
|
||||
let uuids = Vec::from_iter(iter::repeat_with(|| ulid::Ulid::new().into()).take(n as usize));
|
||||
let args = uuids
|
||||
.iter()
|
||||
.map(|uuid| sqlx::types::Json(to_raw_value(&serde_json::json!({ "uuid": uuid }))))
|
||||
.collect::<Vec<_>>();
|
||||
tx = RawJob {
|
||||
created_by: &authed.username,
|
||||
permissioned_as: &username_to_permissioned_as(&authed.username),
|
||||
permissioned_as_email: &authed.email,
|
||||
kind: job_kind,
|
||||
runnable_id: hash.map(|h| h.0),
|
||||
runnable_path: path.as_deref(),
|
||||
script_lang: Some(language),
|
||||
tag: &tag,
|
||||
concurrent_limit,
|
||||
concurrent_time_window_s,
|
||||
concurrency_time_window_s,
|
||||
timeout,
|
||||
n,
|
||||
)
|
||||
.fetch_all(&mut *tx)
|
||||
raw_code: raw_code.as_deref(),
|
||||
raw_lock: raw_lock.as_deref(),
|
||||
raw_flow: raw_flow.as_ref(),
|
||||
flow_status: flow_status.as_ref(),
|
||||
..RawJob::default()
|
||||
}
|
||||
.push_many(tx, &uuids, &w_id, &args)
|
||||
.await?;
|
||||
|
||||
let uuids = sqlx::query_scalar!(
|
||||
|
||||
@@ -1,246 +0,0 @@
|
||||
use crate::{
|
||||
worker::{write_file, TMP_DIR},
|
||||
DB,
|
||||
};
|
||||
use serde::Serialize;
|
||||
use tokio::time::Instant;
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct BenchmarkInfo {
|
||||
#[serde(skip)]
|
||||
pub start: Instant,
|
||||
#[serde(skip)]
|
||||
pub iters: u64,
|
||||
timings: Vec<BenchmarkIter>,
|
||||
pub iter_durations: Vec<u64>,
|
||||
pub total_duration: Option<u64>,
|
||||
}
|
||||
|
||||
impl BenchmarkInfo {
|
||||
pub fn new() -> Self {
|
||||
BenchmarkInfo {
|
||||
iters: 0,
|
||||
timings: vec![],
|
||||
start: Instant::now(),
|
||||
iter_durations: vec![],
|
||||
total_duration: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn add_iter(&mut self, bench: BenchmarkIter, inc_iters: bool) {
|
||||
if inc_iters {
|
||||
self.iters += 1;
|
||||
}
|
||||
let elapsed_total = bench.start.elapsed().as_nanos() as u64;
|
||||
self.timings.push(bench);
|
||||
self.iter_durations.push(elapsed_total);
|
||||
}
|
||||
|
||||
pub fn write_to_file(&mut self, path: &str) -> anyhow::Result<()> {
|
||||
let total_duration = self.start.elapsed().as_millis() as u64;
|
||||
self.total_duration = Some(total_duration as u64);
|
||||
|
||||
println!(
|
||||
"Writing benchmark {path}, duration of benchmark: {total_duration}s and RPS: {}",
|
||||
self.iters as f64 / total_duration as f64
|
||||
);
|
||||
write_file(TMP_DIR, path, &serde_json::to_string(&self).unwrap()).expect("write profiling");
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct BenchmarkIter {
|
||||
#[serde(skip)]
|
||||
pub start: Instant,
|
||||
#[serde(skip)]
|
||||
last_instant: Instant,
|
||||
last_step: String,
|
||||
timings: Vec<(String, u32)>,
|
||||
}
|
||||
|
||||
impl BenchmarkIter {
|
||||
pub fn new() -> Self {
|
||||
BenchmarkIter {
|
||||
last_instant: Instant::now(),
|
||||
timings: vec![],
|
||||
start: Instant::now(),
|
||||
last_step: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn add_timing(&mut self, name: &str) {
|
||||
let elapsed = self.last_instant.elapsed().as_nanos() as u32;
|
||||
self.timings
|
||||
.push((format!("{}->{}", self.last_step, name), elapsed));
|
||||
self.last_instant = Instant::now();
|
||||
self.last_step = name.to_string();
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
|
||||
use crate::{jobs::JobKind, scripts::ScriptLang};
|
||||
|
||||
let benchmark_kind = std::env::var("BENCHMARK_KIND").unwrap_or("noop".to_string());
|
||||
|
||||
if benchmark_jobs > 0 {
|
||||
let mut tx = db.begin().await.unwrap();
|
||||
match benchmark_kind.as_str() {
|
||||
"dedicated" => {
|
||||
// you need to create the script first, check https://github.com/windmill-labs/windmill/blob/b76a92cfe454c686f005c65f534e29e039f3c706/benchmarks/lib.ts#L47
|
||||
let hash = sqlx::query_scalar!(
|
||||
"SELECT hash FROM script WHERE path = $1 AND workspace_id = $2",
|
||||
"f/benchmarks/dedicated",
|
||||
"admins"
|
||||
)
|
||||
.fetch_one(&mut *tx)
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs"));
|
||||
let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9 FROM generate_series(1, $10)) RETURNING id",
|
||||
hash,
|
||||
"f/benchmarks/dedicated",
|
||||
JobKind::Script as JobKind,
|
||||
ScriptLang::Bun as ScriptLang,
|
||||
"admins:f/benchmarks/dedicated",
|
||||
"admin",
|
||||
"u/admin",
|
||||
"admin@windmill.dev",
|
||||
"admins",
|
||||
benchmark_jobs
|
||||
)
|
||||
.fetch_all(&mut *tx)
|
||||
.await.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs (1)"));
|
||||
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "admins:f/benchmarks/dedicated")
|
||||
.execute(&mut *tx)
|
||||
.await.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs (2)"));
|
||||
sqlx::query!(
|
||||
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
|
||||
&uuids
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs (3)"));
|
||||
}
|
||||
"parallelflow" => {
|
||||
//create dedicated script
|
||||
sqlx::query!("INSERT INTO script (summary, description, dedicated_worker, content, workspace_id, path, hash, language, tag, created_by, lock) VALUES ('', '', true, $1, $2, $3, $4, $5, $6, $7, '') ON CONFLICT (workspace_id, hash) DO NOTHING",
|
||||
"export async function main() {
|
||||
console.log('hello world');
|
||||
}",
|
||||
"admins",
|
||||
"u/admin/parallelflow",
|
||||
1234567890,
|
||||
ScriptLang::Deno as ScriptLang,
|
||||
"flow",
|
||||
"admin",
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs {_e:#}"));
|
||||
let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_flow) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, 1)) RETURNING id",
|
||||
None::<i64>,
|
||||
None::<String>,
|
||||
JobKind::FlowPreview as JobKind,
|
||||
ScriptLang::Deno as ScriptLang,
|
||||
"flow",
|
||||
"admin",
|
||||
"u/admin",
|
||||
"admin@windmill.dev",
|
||||
"admins",
|
||||
serde_json::from_str::<serde_json::Value>(r#"
|
||||
{
|
||||
"modules": [
|
||||
{
|
||||
"id": "a",
|
||||
"value": {
|
||||
"type": "forloopflow",
|
||||
"modules": [
|
||||
{
|
||||
"id": "b",
|
||||
"value": {
|
||||
"path": "u/admin/parallelflow",
|
||||
"type": "script",
|
||||
"tag_override": "",
|
||||
"input_transforms": {}
|
||||
},
|
||||
"summary": "calctest"
|
||||
}
|
||||
],
|
||||
"iterator": {
|
||||
"expr": "[...new Array(300)]",
|
||||
"type": "javascript"
|
||||
},
|
||||
"parallel": true,
|
||||
"parallelism": 10,
|
||||
"skip_failures": true
|
||||
}
|
||||
}
|
||||
],
|
||||
"preprocessor_module": null
|
||||
}
|
||||
"#).unwrap(),
|
||||
)
|
||||
.fetch_all(&mut *tx)
|
||||
.await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (1)"));
|
||||
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "flow")
|
||||
.execute(&mut *tx)
|
||||
.await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (2)"));
|
||||
sqlx::query!(
|
||||
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
|
||||
&uuids
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (3)"));
|
||||
sqlx::query!("INSERT INTO v2_job_flow_runtime (id, flow_status) SELECT unnest($1::uuid[]), $2", &uuids,
|
||||
serde_json::from_str::<serde_json::Value>(r#"
|
||||
{
|
||||
"step": 0,
|
||||
"modules": [
|
||||
{
|
||||
"id": "a",
|
||||
"type": "WaitingForPriorSteps"
|
||||
}
|
||||
],
|
||||
"cleanup_module": {},
|
||||
"failure_module": {
|
||||
"id": "failure",
|
||||
"type": "WaitingForPriorSteps"
|
||||
},
|
||||
"preprocessor_module": null
|
||||
}
|
||||
|
||||
"#).unwrap()
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (4)"));
|
||||
}
|
||||
_ => {
|
||||
let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9 FROM generate_series(1, $10)) RETURNING id",
|
||||
None::<i64>,
|
||||
None::<String>,
|
||||
JobKind::Noop as JobKind,
|
||||
ScriptLang::Deno as ScriptLang,
|
||||
"deno",
|
||||
"admin",
|
||||
"u/admin",
|
||||
"admin@windmill.dev",
|
||||
"admins",
|
||||
benchmark_jobs
|
||||
)
|
||||
.fetch_all(&mut *tx)
|
||||
.await.unwrap_or_else(|_e| panic!("failed to insert noop jobs (1)"));
|
||||
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "deno")
|
||||
.execute(&mut *tx)
|
||||
.await.unwrap_or_else(|_e| panic!("failed to insert noop jobs (2)"));
|
||||
sqlx::query!(
|
||||
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
|
||||
&uuids
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to insert noop jobs (3)"));
|
||||
}
|
||||
}
|
||||
tx.commit().await.unwrap();
|
||||
}
|
||||
}
|
||||
@@ -46,6 +46,22 @@ pub enum JobKind {
|
||||
AppScript,
|
||||
}
|
||||
|
||||
#[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Copy, Clone)]
|
||||
#[sqlx(type_name = "JOB_TRIGGER_KIND", rename_all = "lowercase")]
|
||||
#[serde(rename_all(serialize = "lowercase"))]
|
||||
pub enum JobTriggerKind {
|
||||
Webhook,
|
||||
Http,
|
||||
Websocket,
|
||||
Kafka,
|
||||
Email,
|
||||
Nats,
|
||||
Schedule,
|
||||
App,
|
||||
Ui,
|
||||
Postgres,
|
||||
}
|
||||
|
||||
impl JobKind {
|
||||
pub fn is_flow(&self) -> bool {
|
||||
matches!(
|
||||
|
||||
@@ -18,8 +18,6 @@ use sqlx::{Pool, Postgres};
|
||||
|
||||
pub mod apps;
|
||||
pub mod auth;
|
||||
#[cfg(feature = "benchmark")]
|
||||
pub mod bench;
|
||||
pub mod cache;
|
||||
pub mod db;
|
||||
pub mod ee;
|
||||
|
||||
@@ -41,8 +41,8 @@ use windmill_common::{
|
||||
add_virtual_items_if_necessary, FlowModule, FlowModuleValue, FlowValue, InputTransform,
|
||||
},
|
||||
jobs::{
|
||||
get_payload_tag_from_prefixed_path, JobKind, JobPayload, QueuedJob, RawCode,
|
||||
ENTRYPOINT_OVERRIDE, PREPROCESSOR_FAKE_ENTRYPOINT,
|
||||
get_payload_tag_from_prefixed_path, JobKind, JobPayload, JobTriggerKind, QueuedJob,
|
||||
RawCode, ENTRYPOINT_OVERRIDE, PREPROCESSOR_FAKE_ENTRYPOINT,
|
||||
},
|
||||
schedule::Schedule,
|
||||
scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang},
|
||||
@@ -2740,6 +2740,184 @@ lazy_static::lazy_static! {
|
||||
pub static ref RE_ARG_TAG: Regex = Regex::new(r#"\$args\[(\w+)\]"#).unwrap();
|
||||
}
|
||||
|
||||
pub struct RawJob<'a> {
|
||||
pub created_by: &'a str,
|
||||
pub permissioned_as: &'a str,
|
||||
pub permissioned_as_email: &'a str,
|
||||
pub kind: JobKind,
|
||||
pub runnable_id: Option<i64>,
|
||||
pub runnable_path: Option<&'a str>,
|
||||
pub parent_job: Option<Uuid>,
|
||||
pub script_lang: Option<ScriptLang>,
|
||||
pub flow_step_id: Option<&'a str>,
|
||||
pub flow_root_job: Option<Uuid>,
|
||||
pub trigger: Option<&'a str>,
|
||||
pub trigger_kind: Option<JobTriggerKind>,
|
||||
pub tag: &'a str,
|
||||
pub same_worker: bool,
|
||||
pub visible_to_owner: bool,
|
||||
pub concurrent_limit: Option<i32>,
|
||||
pub concurrency_time_window_s: Option<i32>,
|
||||
pub cache_ttl: Option<i32>,
|
||||
pub timeout: Option<i32>,
|
||||
pub priority: Option<i16>,
|
||||
pub pre_run_error: Option<&'a str>,
|
||||
pub raw_code: Option<&'a str>,
|
||||
pub raw_lock: Option<&'a str>,
|
||||
pub raw_flow: Option<&'a FlowValue>,
|
||||
pub flow_status: Option<&'a FlowStatus>,
|
||||
pub scheduled_for: Option<DateTime<Utc>>,
|
||||
pub running: bool,
|
||||
}
|
||||
|
||||
impl<'a> RawJob<'a> {
|
||||
pub const fn default() -> Self {
|
||||
Self {
|
||||
created_by: "missing",
|
||||
permissioned_as: "g/all",
|
||||
permissioned_as_email: "missing@email.xyz",
|
||||
kind: JobKind::Script,
|
||||
runnable_id: None,
|
||||
runnable_path: None,
|
||||
parent_job: None,
|
||||
script_lang: None,
|
||||
flow_step_id: None,
|
||||
flow_root_job: None,
|
||||
trigger: None,
|
||||
trigger_kind: None,
|
||||
tag: "other",
|
||||
same_worker: false,
|
||||
visible_to_owner: false,
|
||||
concurrent_limit: None,
|
||||
concurrency_time_window_s: None,
|
||||
cache_ttl: None,
|
||||
timeout: None,
|
||||
priority: None,
|
||||
pre_run_error: None,
|
||||
raw_code: None,
|
||||
raw_lock: None,
|
||||
raw_flow: None,
|
||||
flow_status: None,
|
||||
scheduled_for: None,
|
||||
running: false,
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn push<'c>(
|
||||
&self,
|
||||
tx: sqlx::PgTransaction<'c>,
|
||||
id: Uuid,
|
||||
workspace_id: &str,
|
||||
args: Json<Box<RawValue>>,
|
||||
) -> error::Result<sqlx::PgTransaction<'c>> {
|
||||
self.push_many(tx, &[id], workspace_id, &[args]).await
|
||||
}
|
||||
|
||||
pub async fn push_many<'c>(
|
||||
&self,
|
||||
mut tx: sqlx::PgTransaction<'c>,
|
||||
id: &[Uuid],
|
||||
workspace_id: &str,
|
||||
args: &[Json<Box<RawValue>>],
|
||||
) -> error::Result<sqlx::PgTransaction<'c>> {
|
||||
if id.len() > args.len() {
|
||||
return Err(Error::InternalErr(
|
||||
"args must be at least as long as id".to_string(),
|
||||
));
|
||||
}
|
||||
let created_at = Utc::now();
|
||||
sqlx::query!(
|
||||
"INSERT INTO v2_job (
|
||||
id, workspace_id,
|
||||
created_at, created_by, permissioned_as, permissioned_as_email,
|
||||
kind, runnable_id, runnable_path, parent_job, script_lang,
|
||||
flow_step_id, flow_root_job,
|
||||
trigger, trigger_kind,
|
||||
tag, same_worker, visible_to_owner, concurrent_limit, concurrency_time_window_s,
|
||||
cache_ttl, timeout, priority,
|
||||
args, pre_run_error,
|
||||
raw_code, raw_lock, raw_flow
|
||||
) SELECT
|
||||
unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15,
|
||||
$16, $17, $18, $19, $20, $21, $22, $23, unnest($24::jsonb[]), $25, $26, $27, $28
|
||||
",
|
||||
id,
|
||||
workspace_id,
|
||||
created_at,
|
||||
self.created_by,
|
||||
self.permissioned_as,
|
||||
self.permissioned_as_email,
|
||||
self.kind as JobKind,
|
||||
self.runnable_id,
|
||||
self.runnable_path,
|
||||
self.parent_job,
|
||||
self.script_lang as Option<ScriptLang>,
|
||||
self.flow_step_id,
|
||||
self.flow_root_job,
|
||||
self.trigger,
|
||||
self.trigger_kind as Option<JobTriggerKind>,
|
||||
self.tag,
|
||||
self.same_worker,
|
||||
self.visible_to_owner,
|
||||
self.concurrent_limit,
|
||||
self.concurrency_time_window_s,
|
||||
self.cache_ttl,
|
||||
self.timeout,
|
||||
self.priority,
|
||||
args as &[Json<Box<RawValue>>],
|
||||
self.pre_run_error,
|
||||
self.raw_code,
|
||||
self.raw_lock,
|
||||
self.raw_flow.map(Json) as Option<Json<&FlowValue>>,
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.warn_after_seconds(1)
|
||||
.await?;
|
||||
sqlx::query!(
|
||||
"INSERT INTO v2_job_queue (
|
||||
id, workspace_id, started_at, scheduled_for, running, created_at, tag, priority
|
||||
) SELECT unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8",
|
||||
id,
|
||||
workspace_id,
|
||||
self.running.then(|| created_at),
|
||||
self.scheduled_for.unwrap_or(created_at),
|
||||
self.running,
|
||||
created_at,
|
||||
self.tag,
|
||||
self.priority,
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.warn_after_seconds(1)
|
||||
.await?;
|
||||
sqlx::query!(
|
||||
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
|
||||
id
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.warn_after_seconds(1)
|
||||
.await?;
|
||||
if let Some(flow_status) = self.flow_status {
|
||||
sqlx::query!(
|
||||
"INSERT INTO v2_job_flow_runtime (id, flow_status) SELECT unnest($1::uuid[]), $2",
|
||||
id,
|
||||
Json(flow_status) as Json<&FlowStatus>,
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.warn_after_seconds(1)
|
||||
.await?;
|
||||
}
|
||||
|
||||
#[cfg(feature = "prometheus")]
|
||||
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
|
||||
// TODO: technically the job isn't queued yet, as the transaction can be rolled back.
|
||||
// Should be solved when moving these metrics to the queue abstraction.
|
||||
QUEUE_PUSH_COUNT.inc_by(id.len() as u64);
|
||||
}
|
||||
|
||||
Ok(tx)
|
||||
}
|
||||
}
|
||||
|
||||
// #[instrument(level = "trace", skip_all)]
|
||||
pub async fn push<'c, 'd>(
|
||||
_db: &Pool<Postgres>,
|
||||
@@ -3666,89 +3844,56 @@ pub async fn push<'c, 'd>(
|
||||
None
|
||||
};
|
||||
|
||||
let raw_flow = raw_flow.map(Json);
|
||||
|
||||
sqlx::query!(
|
||||
"INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,
|
||||
created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,
|
||||
script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,
|
||||
flow_root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,
|
||||
cache_ttl, priority, trigger_kind)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,
|
||||
$19, $20, $21, $22, $23, $24, $25, $26,
|
||||
CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END)",
|
||||
job_id,
|
||||
workspace_id,
|
||||
raw_code,
|
||||
raw_lock,
|
||||
raw_flow as Option<Json<FlowValue>>,
|
||||
tag,
|
||||
let pre_run_error = pre_run_error.map(Error::to_string);
|
||||
let args = serde_json::value::to_raw_value(&args).map_err(|e| {
|
||||
Error::InternalErr(format!("Could not serialize args for job {job_id}: {e:#}"))
|
||||
})?;
|
||||
tracing::debug!(
|
||||
"Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, \
|
||||
script_path: {script_path:?}, email {email}, workspace_id {workspace_id}"
|
||||
);
|
||||
tx = RawJob {
|
||||
created_by: &user,
|
||||
permissioned_as: &permissioned_as,
|
||||
permissioned_as_email: &email,
|
||||
kind: job_kind,
|
||||
runnable_id: script_hash,
|
||||
runnable_path: script_path.as_deref(),
|
||||
parent_job,
|
||||
user,
|
||||
permissioned_as,
|
||||
script_hash,
|
||||
script_path.clone(),
|
||||
Json(args) as Json<PushArgs>,
|
||||
job_kind.clone() as JobKind,
|
||||
schedule_path,
|
||||
language as Option<ScriptLang>,
|
||||
script_lang: language,
|
||||
flow_step_id: flow_step_id.as_deref(),
|
||||
flow_root_job: root_job,
|
||||
trigger: schedule_path.as_deref(),
|
||||
trigger_kind: schedule_path
|
||||
.as_ref()
|
||||
.and_then(|_| Some(JobTriggerKind::Schedule)),
|
||||
tag: &tag,
|
||||
same_worker,
|
||||
pre_run_error.map(|e| e.to_string()),
|
||||
email,
|
||||
visible_to_owner,
|
||||
root_job,
|
||||
concurrent_limit,
|
||||
if concurrent_limit.is_some() {
|
||||
concurrency_time_window_s
|
||||
} else {
|
||||
None
|
||||
},
|
||||
custom_timeout,
|
||||
flow_step_id,
|
||||
|
||||
concurrency_time_window_s,
|
||||
cache_ttl,
|
||||
final_priority,
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.warn_after_seconds(1)
|
||||
.await?;
|
||||
|
||||
tracing::debug!("Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}");
|
||||
let uuid = sqlx::query_scalar!(
|
||||
"INSERT INTO v2_job_queue
|
||||
(workspace_id, id, running, scheduled_for, started_at, tag, priority)
|
||||
VALUES ($1, $2, $3, COALESCE($4, now()), CASE WHEN $3 THEN now() END, $5, $6) \
|
||||
RETURNING id AS \"id!\"",
|
||||
workspace_id,
|
||||
job_id,
|
||||
is_running,
|
||||
scheduled_for_o,
|
||||
tag,
|
||||
final_priority,
|
||||
)
|
||||
.fetch_one(&mut *tx)
|
||||
.warn_after_seconds(1)
|
||||
timeout: custom_timeout,
|
||||
priority: final_priority,
|
||||
pre_run_error: pre_run_error.as_deref(),
|
||||
raw_code: raw_code.as_deref(),
|
||||
raw_lock: raw_lock.as_deref(),
|
||||
raw_flow: raw_flow.as_ref(),
|
||||
flow_status: flow_status.as_ref(),
|
||||
scheduled_for: scheduled_for_o,
|
||||
running: is_running,
|
||||
}
|
||||
.push(tx, job_id, workspace_id, Json(args))
|
||||
.await
|
||||
.map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}: {e:#}")))?;
|
||||
|
||||
sqlx::query!("INSERT INTO v2_job_runtime (id) VALUES ($1)", job_id)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
if let Some(flow_status) = flow_status {
|
||||
sqlx::query!(
|
||||
"INSERT INTO v2_job_flow_runtime (id, flow_status) VALUES ($1, $2)",
|
||||
job_id,
|
||||
Json(flow_status) as Json<FlowStatus>,
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
}
|
||||
|
||||
.map_err(|e| {
|
||||
Error::InternalErr(format!(
|
||||
"Could not insert into queue {job_id} with tag {tag}, \
|
||||
schedule_path {schedule_path:?}, script_path: {script_path:?}, \
|
||||
email {email}, workspace_id {workspace_id}: {e:#}"
|
||||
))
|
||||
})?;
|
||||
tracing::debug!("Pushed {job_id}");
|
||||
// TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction.
|
||||
#[cfg(feature = "prometheus")]
|
||||
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
|
||||
QUEUE_PUSH_COUNT.inc();
|
||||
}
|
||||
|
||||
if JOB_TOKEN.is_none() {
|
||||
let job_authed = match authed {
|
||||
@@ -3858,7 +4003,7 @@ pub async fn push<'c, 'd>(
|
||||
.await?;
|
||||
}
|
||||
|
||||
Ok((uuid, tx))
|
||||
Ok((job_id, tx))
|
||||
}
|
||||
|
||||
pub fn canceled_job_to_result(job: &QueuedJob) -> serde_json::Value {
|
||||
|
||||
@@ -102,6 +102,7 @@ object_store = { workspace = true, optional = true}
|
||||
convert_case.workspace = true
|
||||
yaml-rust.workspace = true
|
||||
backon.workspace = true
|
||||
ulid.workspace = true
|
||||
|
||||
opentelemetry = { workspace = true, optional = true }
|
||||
bollard = { workspace = true, optional = true }
|
||||
|
||||
@@ -0,0 +1,226 @@
|
||||
use serde::Serialize;
|
||||
use serde_json::json;
|
||||
use tokio::time::Instant;
|
||||
use windmill_common::{
|
||||
worker::{write_file, TMP_DIR},
|
||||
DB,
|
||||
};
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct BenchmarkInfo {
|
||||
#[serde(skip)]
|
||||
pub start: Instant,
|
||||
#[serde(skip)]
|
||||
pub iters: u64,
|
||||
timings: Vec<BenchmarkIter>,
|
||||
pub iter_durations: Vec<u64>,
|
||||
pub total_duration: Option<u64>,
|
||||
}
|
||||
|
||||
impl BenchmarkInfo {
|
||||
pub fn new() -> Self {
|
||||
BenchmarkInfo {
|
||||
iters: 0,
|
||||
timings: vec![],
|
||||
start: Instant::now(),
|
||||
iter_durations: vec![],
|
||||
total_duration: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn add_iter(&mut self, bench: BenchmarkIter, inc_iters: bool) {
|
||||
if inc_iters {
|
||||
self.iters += 1;
|
||||
}
|
||||
let elapsed_total = bench.start.elapsed().as_nanos() as u64;
|
||||
self.timings.push(bench);
|
||||
self.iter_durations.push(elapsed_total);
|
||||
}
|
||||
|
||||
pub fn write_to_file(&mut self, path: &str) -> anyhow::Result<()> {
|
||||
let total_duration = self.start.elapsed().as_millis() as u64;
|
||||
self.total_duration = Some(total_duration as u64);
|
||||
|
||||
println!(
|
||||
"Writing benchmark {path}, duration of benchmark: {total_duration}s and RPS: {}",
|
||||
self.iters as f64 / total_duration as f64
|
||||
);
|
||||
write_file(TMP_DIR, path, &serde_json::to_string(&self).unwrap()).expect("write profiling");
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct BenchmarkIter {
|
||||
#[serde(skip)]
|
||||
pub start: Instant,
|
||||
#[serde(skip)]
|
||||
last_instant: Instant,
|
||||
last_step: String,
|
||||
timings: Vec<(String, u32)>,
|
||||
}
|
||||
|
||||
impl BenchmarkIter {
|
||||
pub fn new() -> Self {
|
||||
BenchmarkIter {
|
||||
last_instant: Instant::now(),
|
||||
timings: vec![],
|
||||
start: Instant::now(),
|
||||
last_step: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn add_timing(&mut self, name: &str) {
|
||||
let elapsed = self.last_instant.elapsed().as_nanos() as u32;
|
||||
self.timings
|
||||
.push((format!("{}->{}", self.last_step, name), elapsed));
|
||||
self.last_instant = Instant::now();
|
||||
self.last_step = name.to_string();
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn benchmark_init(benchmark_jobs: usize, db: &DB) {
|
||||
use std::iter;
|
||||
|
||||
use windmill_common::{jobs::JobKind, scripts::ScriptLang};
|
||||
use windmill_queue::RawJob;
|
||||
|
||||
let benchmark_kind = std::env::var("BENCHMARK_KIND").unwrap_or("noop".to_string());
|
||||
let uuids = Vec::from_iter(iter::repeat_with(|| ulid::Ulid::new().into()).take(benchmark_jobs));
|
||||
|
||||
if !uuids.is_empty() {
|
||||
match benchmark_kind.as_str() {
|
||||
"dedicated" => {
|
||||
// you need to create the script first, check https://github.com/windmill-labs/windmill/blob/b76a92cfe454c686f005c65f534e29e039f3c706/benchmarks/lib.ts#L47
|
||||
let hash = sqlx::query_scalar!(
|
||||
"SELECT hash FROM script WHERE path = $1 AND workspace_id = $2",
|
||||
"f/benchmarks/dedicated",
|
||||
"admins"
|
||||
)
|
||||
.fetch_one(db)
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs"));
|
||||
RawJob {
|
||||
runnable_id: Some(hash),
|
||||
runnable_path: Some("f/benchmarks/dedicated"),
|
||||
kind: JobKind::Script,
|
||||
script_lang: Some(ScriptLang::Bun),
|
||||
tag: "admins:f/benchmarks/dedicated",
|
||||
created_by: "admin",
|
||||
permissioned_as: "u/admin",
|
||||
permissioned_as_email: "admin@windmill.dev",
|
||||
..RawJob::default()
|
||||
}
|
||||
.push_many(db.begin().await.unwrap(), uuids.as_slice(), "admins", &[])
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs"))
|
||||
.commit()
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to commit insert of dedicated jobs"));
|
||||
}
|
||||
"parallelflow" => {
|
||||
// create dedicated script
|
||||
sqlx::query!(
|
||||
"INSERT INTO script (
|
||||
summary, description, dedicated_worker, content, workspace_id, path, hash,
|
||||
language, tag, created_by, lock
|
||||
) VALUES ('', '', true, $1, $2, $3, $4, $5, $6, $7, '')
|
||||
ON CONFLICT (workspace_id, hash) DO NOTHING",
|
||||
"export async function main() {
|
||||
console.log('hello world');
|
||||
}",
|
||||
"admins",
|
||||
"u/admin/parallelflow",
|
||||
1234567890,
|
||||
ScriptLang::Deno as ScriptLang,
|
||||
"flow",
|
||||
"admin",
|
||||
)
|
||||
.execute(db)
|
||||
.await
|
||||
.unwrap_or_else(|e| panic!("failed to insert parallelflow script {e:#}"));
|
||||
RawJob {
|
||||
kind: JobKind::FlowPreview,
|
||||
tag: "flow",
|
||||
created_by: "admin",
|
||||
permissioned_as: "u/admin",
|
||||
permissioned_as_email: "admin@windmill.dev",
|
||||
raw_flow: Some(
|
||||
&serde_json::from_value(json!({
|
||||
"modules": [
|
||||
{
|
||||
"id": "a",
|
||||
"value": {
|
||||
"type": "forloopflow",
|
||||
"modules": [
|
||||
{
|
||||
"id": "b",
|
||||
"value": {
|
||||
"path": "u/admin/parallelflow",
|
||||
"type": "script",
|
||||
"tag_override": "",
|
||||
"input_transforms": {}
|
||||
},
|
||||
"summary": "calctest"
|
||||
}
|
||||
],
|
||||
"iterator": {
|
||||
"expr": "[...new Array(300)]",
|
||||
"type": "javascript"
|
||||
},
|
||||
"parallel": true,
|
||||
"parallelism": 10,
|
||||
"skip_failures": true
|
||||
}
|
||||
}
|
||||
],
|
||||
"preprocessor_module": null
|
||||
}))
|
||||
.unwrap(),
|
||||
),
|
||||
flow_status: Some(
|
||||
&serde_json::from_value(json!({
|
||||
"step": 0,
|
||||
"modules": [
|
||||
{
|
||||
"id": "a",
|
||||
"type": "WaitingForPriorSteps"
|
||||
}
|
||||
],
|
||||
"cleanup_module": {},
|
||||
"failure_module": {
|
||||
"id": "failure",
|
||||
"type": "WaitingForPriorSteps"
|
||||
},
|
||||
"preprocessor_module": null
|
||||
}))
|
||||
.unwrap(),
|
||||
),
|
||||
..RawJob::default()
|
||||
}
|
||||
.push_many(db.begin().await.unwrap(), uuids.as_slice(), "admins", &[])
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs"))
|
||||
.commit()
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to commit insert of parallelflow jobs"));
|
||||
}
|
||||
_ => {
|
||||
RawJob {
|
||||
kind: JobKind::Noop,
|
||||
tag: "deno",
|
||||
created_by: "admin",
|
||||
permissioned_as: "u/admin",
|
||||
permissioned_as_email: "admin@windmill.dev",
|
||||
..RawJob::default()
|
||||
}
|
||||
.push_many(db.begin().await.unwrap(), uuids.as_slice(), "admins", &[])
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to insert noop jobs"))
|
||||
.commit()
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to commit insert of noop jobs"));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,8 @@ mod snowflake_executor;
|
||||
mod ansible_executor;
|
||||
mod bash_executor;
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
pub mod bench;
|
||||
mod bun_executor;
|
||||
pub mod common;
|
||||
mod config;
|
||||
|
||||
@@ -26,7 +26,7 @@ use windmill_common::{
|
||||
};
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
use windmill_common::bench::{BenchmarkInfo, BenchmarkIter};
|
||||
use crate::bench::{BenchmarkInfo, BenchmarkIter};
|
||||
|
||||
use windmill_queue::{append_logs, get_queued_job, CanceledBy, WrappedError};
|
||||
|
||||
|
||||
@@ -146,7 +146,7 @@ use crate::mssql_executor::do_mssql;
|
||||
use crate::bigquery_executor::do_bigquery;
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
use windmill_common::bench::{benchmark_init, BenchmarkInfo, BenchmarkIter};
|
||||
use crate::bench::{benchmark_init, BenchmarkInfo, BenchmarkIter};
|
||||
|
||||
use windmill_common::add_time;
|
||||
|
||||
@@ -1050,9 +1050,9 @@ pub async fn run_worker(
|
||||
let is_dedicated_worker: bool = WORKER_CONFIG.read().await.dedicated_worker.is_some();
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
let benchmark_jobs: i32 = std::env::var("BENCHMARK_JOBS")
|
||||
let benchmark_jobs: usize = std::env::var("BENCHMARK_JOBS")
|
||||
.unwrap_or("5000".to_string())
|
||||
.parse::<i32>()
|
||||
.parse::<usize>()
|
||||
.unwrap();
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
|
||||
@@ -11,6 +11,8 @@ use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
use crate::bench::BenchmarkIter;
|
||||
use crate::common::{cached_result_path, save_in_cache};
|
||||
use crate::js_eval::{eval_timeout, IdContext};
|
||||
use crate::{
|
||||
@@ -29,8 +31,6 @@ use tracing::instrument;
|
||||
use uuid::Uuid;
|
||||
use windmill_common::add_time;
|
||||
use windmill_common::auth::JobPerms;
|
||||
#[cfg(feature = "benchmark")]
|
||||
use windmill_common::bench::BenchmarkIter;
|
||||
use windmill_common::cache::{self, RawData};
|
||||
use windmill_common::db::Authed;
|
||||
use windmill_common::flow_status::{
|
||||
|
||||
Reference in New Issue
Block a user