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

do a test run on first iteration

Browse Source
This commit is contained in:
Ruben Fiszel
2023-04-13 21:54:34 +02:00
parent 58d5238e99
commit c2b66fef4f
3 changed files with 109 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/inputs.rs
+1 -1
View File
@@ -114,7 +114,7 @@ async fn get_input_history(
let sql = &format!(
"select * from (select distinct on (args) * from completed_job \
where {} = $1 and job_kind = $2 and workspace_id = $3 \
order by args, started_at desc limit $4 offset $5) t ORDER BY started_at desc",
order by args, started_at desc) t ORDER BY started_at desc limit $4 offset $5",
r.runnable_type.column_name()
);
backend/windmill-common/src/jobs.rs
+40
View File
@@ -101,6 +101,46 @@ impl QueuedJob {
}
}
impl Default for QueuedJob {
fn default() -> Self {
Self {
workspace_id: "".to_string(),
id: Uuid::default(),
parent_job: None,
created_by: "".to_string(),
created_at: chrono::Utc::now(),
started_at: None,
scheduled_for: chrono::Utc::now(),
running: false,
script_hash: None,
script_path: None,
args: None,
logs: None,
raw_code: None,
raw_lock: None,
canceled: false,
canceled_by: None,
canceled_reason: None,
last_ping: None,
job_kind: JobKind::Identity,
schedule_path: None,
permissioned_as: "".to_string(),
flow_status: None,
raw_flow: None,
is_flow_step: false,
language: None,
same_worker: false,
pre_run_error: None,
email: "".to_string(),
visible_to_owner: false,
suspend: None,
mem_peak: None,
root_job: None,
leaf_jobs: None,
}
}
}
#[derive(Debug, Clone)]
pub enum JobPayload {
ScriptHub { path: String },
backend/windmill-worker/src/worker.rs
+68 -54
View File
@@ -64,22 +64,25 @@ use crate::{
pub async fn create_token_for_owner_in_bg(db: &Pool<Postgres>, job: &QueuedJob) -> Arc<RwLock<String>> {
let rw_lock = Arc::new(RwLock::new(String::new()));
let mut locked = rw_lock.clone().write_owned().await;
let db = db.clone();
let job = job.clone();
tokio::spawn(async move {
// skipping test runs
if job.workspace_id != "" {
let mut locked = rw_lock.clone().write_owned().await;
let db = db.clone();
let job = job.clone();
let token = create_token_for_owner(
&db.clone(),
&job.workspace_id,
&job.permissioned_as,
"ephemeral-script",
*SESSION_TOKEN_EXPIRY,
&job.email,
)
.await.expect("could not create job token");
*locked = token;
});
tokio::spawn(async move {
let job = job.clone();
let token = create_token_for_owner(
&db.clone(),
&job.workspace_id,
&job.permissioned_as,
"ephemeral-script",
*SESSION_TOKEN_EXPIRY,
&job.email,
)
.await.expect("could not create job token");
*locked = token;
});
};
return rw_lock;
}
@@ -422,9 +425,11 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
let (same_worker_tx, mut same_worker_rx) = mpsc::channel::<Uuid>(5);
tracing::info!(worker = %worker_name, "listening for jobs");
let mut first_run = true;
loop {
if *METRICS_ENABLED {
worker_busy.set(0);
@@ -464,44 +469,49 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
last_sync = Instant::now();
}
let (do_break, next_job) = async {
tokio::select! {
biased;
_ = rx.recv() => {
if let Some(copy_cache_from_bucket_handle) = copy_cache_from_bucket_handle.as_ref() {
copy_cache_from_bucket_handle.abort();
}
println!("received killpill for worker {}", i_worker);
(true, Ok(None))
},
_ = _copy_bucket_rx.recv() => {
if let Err(e) = move_tmp_cache_to_cache().await {
tracing::error!(worker = %worker_name, "failed to sync tmp cache to cache: {}", e);
}
copy_cache_from_bucket_handle = None;
initialized_cache = true;
(false, Ok(None))
},
Some(job_id) = same_worker_rx.recv() => {
(false, sqlx::query_as::<_, QueuedJob>("SELECT * FROM queue WHERE id = $1")
.bind(job_id)
.fetch_optional(db)
.await
.map_err(|_| Error::InternalErr("Impossible to fetch same_worker job".to_string())))
},
(job, timer) = {
let timer = if *METRICS_ENABLED { Some(worker_pull_duration.start_timer()) } else { None };
pull(&db, WHITELIST_WORKSPACES.clone(), BLACKLIST_WORKSPACES.clone(), rsmq.clone()).map(|x| (x, timer))
} => {
timer.map(|timer| {
let duration_pull_s = timer.stop_and_record();
worker_pull_duration_counter.inc_by(duration_pull_s);
});
(false, job)
},
}
}.instrument(trace_span!("worker_get_next_job")).await;
let (do_break, next_job) = if first_run {
(false, Ok(Some(QueuedJob::default())))
} else {
async {
tokio::select! {
biased;
_ = rx.recv() => {
if let Some(copy_cache_from_bucket_handle) = copy_cache_from_bucket_handle.as_ref() {
copy_cache_from_bucket_handle.abort();
}
println!("received killpill for worker {}", i_worker);
(true, Ok(None))
},
_ = _copy_bucket_rx.recv() => {
if let Err(e) = move_tmp_cache_to_cache().await {
tracing::error!(worker = %worker_name, "failed to sync tmp cache to cache: {}", e);
}
copy_cache_from_bucket_handle = None;
initialized_cache = true;
(false, Ok(None))
},
Some(job_id) = same_worker_rx.recv() => {
(false, sqlx::query_as::<_, QueuedJob>("SELECT * FROM queue WHERE id = $1")
.bind(job_id)
.fetch_optional(db)
.await
.map_err(|_| Error::InternalErr("Impossible to fetch same_worker job".to_string())))
},
(job, timer) = {
let timer = if *METRICS_ENABLED { Some(worker_pull_duration.start_timer()) } else { None };
pull(&db, WHITELIST_WORKSPACES.clone(), BLACKLIST_WORKSPACES.clone(), rsmq.clone()).map(|x| (x, timer))
} => {
timer.map(|timer| {
let duration_pull_s = timer.stop_and_record();
worker_pull_duration_counter.inc_by(duration_pull_s);
});
(false, job)
},
}
}.await
};
first_run = false;
if do_break {
return true;
}
@@ -569,7 +579,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
.expect("could not create shared dir");
}
let authed_client = AuthedClientBackgroundTask { base_internal_url: base_internal_url.to_string(), token: token, workspace: job.workspace_id.to_string(), client: OnceCell::new() };
let authed_client = AuthedClientBackgroundTask { base_internal_url: base_internal_url.to_string(), token, workspace: job.workspace_id.to_string(), client: OnceCell::new() };
let is_flow = job.job_kind == JobKind::Flow || job.job_kind == JobKind::FlowPreview || job.job_kind == JobKind::FlowDependencies;
if let Some(err) = handle_queued_job(
@@ -834,6 +844,10 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
}
};
//it's a test job, no need to update the db
if job.workspace_id == "" {
return Ok(());
}
let client = &client.get_authed().await;
match result {
Ok(r) => {