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

fix: make cleanup_module backcompatible

Browse Source
This commit is contained in:
Ruben Fiszel
2023-12-10 13:29:51 +01:00
parent 33c24e1186
commit 715cb08fcd
4 changed files with 37 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-common/src/flow_status.rs
+3 -13
View File
@@ -29,6 +29,7 @@ pub struct FlowStatus {
pub step: i32,
pub modules: Vec<FlowStatusModule>,
pub failure_module: FlowStatusModuleWParent,
#[serde(default)]
pub cleanup_module: FlowCleanupModule,
#[serde(default)]
#[serde(skip_serializing_if = "is_retry_default")]
@@ -97,13 +98,11 @@ pub struct FlowStatusModuleWParent {
pub module_status: FlowStatusModule,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
#[derive(Serialize, Deserialize, Debug, Clone, Default)]
pub struct FlowCleanupModule {
#[serde(default)]
#[serde(skip_serializing_if = "Vec::is_empty")]
pub flow_jobs_to_clean: Vec<Uuid>,
#[serde(flatten)]
pub module_status: FlowStatusModule,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
@@ -228,16 +227,7 @@ impl FlowStatus {
.unwrap_or_else(|| "failure".to_string()),
},
},
cleanup_module: FlowCleanupModule {
flow_jobs_to_clean: vec![],
module_status: FlowStatusModule::WaitingForPriorSteps {
id: f
.failure_module
.as_ref()
.map(|x| x.id.clone())
.unwrap_or_else(|| "cleanup".to_string()),
},
},
cleanup_module: FlowCleanupModule { flow_jobs_to_clean: vec![] },
retry: RetryStatus { fail_count: 0, failed_jobs: vec![] },
restarted_from: None,
}
backend/windmill-queue/src/jobs.rs
+2 -12
View File
@@ -2449,12 +2449,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
id: "failure".to_string(),
},
},
cleanup_module: FlowCleanupModule {
module_status: FlowStatusModule::WaitingForPriorSteps {
id: "cleanup".to_string(),
},
flow_jobs_to_clean: vec![],
},
cleanup_module: FlowCleanupModule { flow_jobs_to_clean: vec![] },
// retry status is reset
retry: RetryStatus { fail_count: 0, failed_jobs: vec![] },
// TODO: for now, flows with approval conditions aren't supported for restart
@@ -2540,12 +2535,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
id: "failure".to_string(),
},
},
cleanup_module: FlowCleanupModule {
module_status: FlowStatusModule::WaitingForPriorSteps {
id: "cleanup".to_string(),
},
flow_jobs_to_clean: vec![],
},
cleanup_module: FlowCleanupModule { flow_jobs_to_clean: vec![] },
// retry status is reset
retry: RetryStatus { fail_count: 0, failed_jobs: vec![] },
// TODO: for now, flows with approval conditions aren't supported for restart
backend/windmill-worker/src/worker_flow.rs
+31 -23
View File
@@ -27,7 +27,7 @@ use tokio::sync::mpsc::Sender;
use tracing::instrument;
use uuid::Uuid;
use windmill_common::flow_status::{
ApprovalConditions, FlowCleanupModule, FlowStatusModuleWParent, Iterator, JobResult,
ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult,
};
use windmill_common::flows::add_virtual_items_if_necessary;
use windmill_common::jobs::{
@@ -168,7 +168,15 @@ pub async fn update_flow_status_after_job_completion_internal<
rsmq: Option<R>,
worker_name: &str,
) -> error::Result<Option<RecUpdateFlowStatusAfterJobCompletion>> {
let (should_continue_flow, flow_job, stop_early, skip_if_stop_early, nresult, is_failure_step) = {
let (
should_continue_flow,
flow_job,
stop_early,
skip_if_stop_early,
nresult,
is_failure_step,
cleanup_module,
) = {
// tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id} {depth}");
let old_status_json = sqlx::query_scalar!(
@@ -574,6 +582,7 @@ pub async fn update_flow_status_after_job_completion_internal<
skip_if_stop_early,
nresult,
is_failure_step,
old_status.cleanup_module,
)
};
@@ -591,8 +600,7 @@ pub async fn update_flow_status_after_job_completion_internal<
#[cfg(feature = "enterprise")]
if flow_job.parent_job.is_none() {
// run the cleanup step only when the root job is complete
let cleanup_module = retrieve_cleanup_module(flow, db).await?;
if cleanup_module.flow_jobs_to_clean.len() > 0 {
if !cleanup_module.flow_jobs_to_clean.is_empty() {
tracing::debug!(
"Cleaning up jobs arguments, result and logs as they were marked as delete_after_use {:?}",
cleanup_module.flow_jobs_to_clean
@@ -837,26 +845,26 @@ async fn has_failure_module<'c>(
.map(|v| v.unwrap_or(false))
}
async fn retrieve_cleanup_module<'c>(flow_uuid: Uuid, db: &DB) -> Result<FlowCleanupModule, Error> {
tracing::warn!("Retrieving cleanup module of flow {}", flow_uuid);
let raw_value = sqlx::query_scalar!(
"SELECT flow_status->'cleanup_module' as cleanup_module
FROM queue
WHERE id = $1",
flow_uuid,
)
.fetch_one(db)
.await
.map_err(|e| Error::InternalErr(format!("error during retrieval of cleanup module: {e}")))?;
// async fn retrieve_cleanup_module<'c>(flow_uuid: Uuid, db: &DB) -> Result<FlowCleanupModule, Error> {
// tracing::warn!("Retrieving cleanup module of flow {}", flow_uuid);
// let raw_value = sqlx::query_scalar!(
// "SELECT flow_status->'cleanup_module' as cleanup_module
// FROM queue
// WHERE id = $1",
// flow_uuid,
// )
// .fetch_one(db)
// .await
// .map_err(|e| Error::InternalErr(format!("error during retrieval of cleanup module: {e}")))?;
raw_value
.clone()
.and_then(|rv| serde_json::from_value::<FlowCleanupModule>(rv).ok())
.ok_or(Error::InternalErr(format!(
"Unable to parse flow cleanup module {:?}",
raw_value
)))
}
// raw_value
// .clone()
// .and_then(|rv| serde_json::from_value::<FlowCleanupModule>(rv).ok())
// .ok_or(Error::InternalErr(format!(
// "Unable to parse flow cleanup module {:?}",
// raw_value
// )))
// }
fn next_retry(retry: &Retry, status: &RetryStatus) -> Option<(u16, Duration)> {
(status.fail_count <= MAX_RETRY_ATTEMPTS)
python-client/wmill/pyproject.toml
+1 -1
View File
@@ -16,7 +16,7 @@ include = ["wmill/py.typed"]
[tool.poetry.dependencies]
python = "^3.7"
httpx = "^0.25"
httpx = "^0.24"
[build-system]
requires = ["poetry>=1.0.2", "poetry-dynamic-versioning"]