starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-09 00:04:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: add WM_SCHEDULED_FOR to contextual variables and early stop of flows

Browse Source
This commit is contained in:
Ruben Fiszel
2024-07-18 02:10:15 +02:00
parent 1a8b3e49eb
commit be25101377
12 changed files with 60 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/resources.rs
+1
View File
@@ -567,6 +567,7 @@ pub async fn transform_json_value<'c>(
job.flow_step_id.clone(),
job.root_job.map(|x| x.to_string()),
None,
Some(job.scheduled_for.clone()),
)
.await;
backend/windmill-api/src/variables.rs
+1
View File
@@ -75,6 +75,7 @@ async fn list_contextual_variables(
Some("c".to_string()),
Some("017e0ad5-f499-73b6-5488-92a61c5196dd".to_string()),
Some("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c".to_string()),
Some(chrono::offset::Utc::now())
)
.await
.to_vec(),
backend/windmill-common/src/variables.rs
+12
View File
@@ -6,6 +6,7 @@
* LICENSE-AGPL for a copy of the license.
*/
use chrono::Utc;
use magic_crypt::{MagicCrypt256, MagicCryptError, MagicCryptTrait};
use serde::{Deserialize, Serialize};
@@ -147,6 +148,8 @@ pub async fn decrypt_value_with_mc(
})?)
}
pub const WM_SCHEDULED_FOR: &str = "WM_SCHEDULED_FOR";
pub async fn get_reserved_variables(
db: &DB,
w_id: &str,
@@ -162,6 +165,7 @@ pub async fn get_reserved_variables(
step_id: Option<String>,
root_flow_id: Option<String>,
jwt_token: Option<String>,
scheduled_for: Option<chrono::DateTime<Utc>>,
) -> Vec<ContextualVariable> {
let state_path = {
let trigger = if schedule_path.is_some() {
@@ -247,6 +251,14 @@ pub async fn get_reserved_variables(
description: "Job id of the current script".to_string(),
is_custom: false,
},
ContextualVariable {
name: WM_SCHEDULED_FOR.to_string(),
value: scheduled_for
.map(|ts| ts.to_string())
.unwrap_or_else(|| "".to_string()),
description: "date-time in UTC (e.g: 2014-11-28T12:45:59.324310806Z) of when the job was scheduled".to_string(),
is_custom: false,
},
ContextualVariable {
name: "WM_JOB_PATH".to_string(),
value: path.unwrap_or_else(|| "".to_string()),
backend/windmill-worker/src/bun_executor.rs
+1
View File
@@ -945,6 +945,7 @@ pub async fn start_worker(
None,
None,
None,
None,
)
.await;
let context_envs = build_envs_map(context.to_vec()).await;
backend/windmill-worker/src/common.rs
+2
View File
@@ -314,6 +314,7 @@ pub async fn transform_json_value(
job.flow_step_id.clone(),
job.root_job.clone().map(|x| x.to_string()),
None,
Some(job.scheduled_for.clone()),
)
.await;
@@ -411,6 +412,7 @@ pub async fn get_reserved_variables(
job.flow_step_id.clone(),
job.root_job.clone().map(|x| x.to_string()),
None,
Some(job.scheduled_for.clone()),
)
.await
.to_vec();
backend/windmill-worker/src/deno_executor.rs
+1
View File
@@ -435,6 +435,7 @@ pub async fn start_worker(
None,
None,
None,
None,
)
.await;
let context_envs = build_envs_map(context.to_vec()).await;
backend/windmill-worker/src/js_eval.rs
+17 -2
View File
@@ -141,6 +141,7 @@ pub async fn eval_timeout(
flow_input: Option<mappable_rc::Marc<HashMap<String, Box<RawValue>>>>,
authed_client: Option<&AuthedClient>,
by_id: Option<IdContext>,
ctx: Option<Vec<(String, String)>>,
) -> anyhow::Result<Box<RawValue>> {
let expr = expr.trim().to_string();
@@ -305,6 +306,7 @@ pub async fn eval_timeout(
context_keys,
by_id,
has_client,
ctx,
))?;
Ok(r) as anyhow::Result<Box<RawValue>>
@@ -367,8 +369,10 @@ async fn eval(
transform_context: Vec<String>,
by_id: Option<IdContext>,
has_client: bool,
ctx: Option<Vec<(String, String)>>,
) -> anyhow::Result<Box<RawValue>> {
tracing::debug!("evaluating: {} {:#?}", expr, by_id);
let (api_code, by_id_code) = if has_client {
let by_id_code = if let Some(by_id) = by_id {
format!(
@@ -440,11 +444,21 @@ async function resource(path) {{
} else {
format!("return {expr}")
};
let ctx_str = ctx
.map(|x| {
x.into_iter()
.map(|(k, v)| format!("let {} = \"{}\";", k, v))
.join("\n")
})
.unwrap_or_default();
let code = format!(
r#"
function get_from_env(name) {{
return JSON.parse(Deno.core.ops.op_get_context(name));
}}
{ctx_str}
{api_code}
{}
{}
@@ -868,6 +882,7 @@ mod tests {
vec!["params".to_string(), "value".to_string()],
None,
false,
None,
)
.await?;
assert_eq!(res.get(), "4");
@@ -882,7 +897,7 @@ return `my ${x}
multiline template`";
let mut runtime = JsRuntime::new(RuntimeOptions::default());
let res = eval(&mut runtime, code, env, None, false).await?;
let res = eval(&mut runtime, code, env, None, false, None).await?;
assert_eq!(res.get(), "\"my 5\\nmultiline template\"");
Ok(())
}
@@ -908,7 +923,7 @@ multiline template`";
op_state.put(TransformContext { flow_input: None, envs: env.clone() })
}
let res = eval_timeout(code.to_string(), env, None, None, None).await?;
let res = eval_timeout(code.to_string(), env, None, None, None, None).await?;
assert_eq!(res.get(), "2");
Ok(())
}
backend/windmill-worker/src/python_executor.rs
+2
View File
@@ -1049,6 +1049,7 @@ pub async fn start_worker(
None,
None,
None,
None,
)
.await
.to_vec();
@@ -1167,6 +1168,7 @@ for line in sys.stdin:
None,
None,
None,
None,
)
.await;
backend/windmill-worker/src/worker_flow.rs
+13
View File
@@ -288,6 +288,7 @@ pub async fn update_flow_status_after_job_completion_internal<
None,
Some(client),
None,
None,
)
.await?
} else {
@@ -1174,6 +1175,7 @@ async fn compute_bool_from_expr(
by_id: Option<IdContext>,
client: Option<&AuthedClient>,
resumes: Option<(Arc<Box<RawValue>>, Arc<Box<RawValue>>, Arc<Box<RawValue>>)>,
ctx: Option<Vec<(String, String)>>,
) -> error::Result<bool> {
let mut context = HashMap::with_capacity(if resumes.is_some() { 7 } else { 3 });
context.insert("result".to_string(), result.clone());
@@ -1191,6 +1193,7 @@ async fn compute_bool_from_expr(
Some(flow_args),
client,
by_id,
ctx,
)
.await?
.get()
@@ -1300,6 +1303,7 @@ async fn transform_input(
Some(flow_args.clone()),
Some(client),
Some(by_id.clone()),
None,
)
.await
.map_err(|e| {
@@ -1552,6 +1556,10 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
None,
Some(client),
None,
Some(vec![(
windmill_common::variables::WM_SCHEDULED_FOR.to_string(),
flow_job.scheduled_for.to_string(),
)]),
)
.await?;
if skip {
@@ -1667,6 +1675,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
Some(arc_flow_job_args.clone()),
None,
None,
None
)
.await
.map_err(|e| {
@@ -1855,6 +1864,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
Some(arc_flow_job_args.clone()),
None,
None,
None,
)
.await
.map_err(|e| {
@@ -2936,6 +2946,7 @@ async fn compute_next_flow_transform(
Some(idcontext.clone()),
Some(client),
Some((resumes.clone(), resume.clone(), approvers.clone())),
None,
)
.await?;
@@ -3254,6 +3265,7 @@ async fn next_forloop_status(
Some(arc_flow_job_args),
Some(client),
Some(by_id),
None,
)
.await?
}
@@ -3314,6 +3326,7 @@ async fn next_forloop_status(
Some(arc_flow_job_args),
Some(client),
Some(by_id),
None,
)
.await?
}
frontend/src/lib/components/flows/content/FlowSettings.svelte
+8 -1
View File
@@ -431,8 +431,15 @@
class="small-editor"
extraLib={`declare const flow_input = ${JSON.stringify(
schemaToObject(asSchema($flowStore.schema), $previewArgs)
)};`}
)};
declare const WM_SCHEDULED_FOR: string;`}
/>
<div class="text-xs text-tertiary mt-2">
You can use the variable `flow_input` to access the inputs of the flow. <br
/>The variable `WM_SCHEDULED_FOR` contains the time the flow was scheduled for
which you can use to stop early non fresh jobs:
<pre>new Date().getTime() - new Date(WM_SCHEDULED_FOR).getTime() {'>'} X</pre>
</div>
</div>
{:else}
<textarea disabled rows="3" class="min-h-[80px]" />
frontend/src/lib/deno_fetch.d.ts.txt
+1
View File
@@ -428,6 +428,7 @@ interface Process {
WM_USERNAME: string;
WM_BASE_URL: string;
WM_JOB_ID: string;
WM_SCHEDULED_FOR: string;
WM_JOB_PATH: string;
WM_FLOW_JOB_ID: string;
WM_ROOT_FLOW_JOB_ID: string;
frontend/src/lib/process.d.ts.txt
+1
View File
@@ -164,6 +164,7 @@ declare module 'process' {
WM_USERNAME: string
WM_BASE_URL: string
WM_JOB_ID: string
WM_SCHEDULED_FOR: string
WM_JOB_PATH: string
WM_FLOW_JOB_ID: string
WM_ROOT_FLOW_JOB_ID: string