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

feat: make flow lock deployment error visible in UI

Browse Source
This commit is contained in:
Ruben Fiszel
2025-02-22 15:02:16 +01:00
parent e89b2feeac
commit 3d6015e44a
18 changed files with 256 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7.json
Generated
+15 -15
View File
@@ -130,28 +130,28 @@
},
{
"ordinal": 25,
"name": "ai_models",
"type_info": "VarcharArray"
},
{
"ordinal": 26,
"name": "code_completion_model",
"type_info": "Varchar"
},
{
"ordinal": 27,
"name": "teams_command_script",
"type_info": "Text"
},
{
"ordinal": 28,
"ordinal": 26,
"name": "teams_team_id",
"type_info": "Text"
},
{
"ordinal": 29,
"ordinal": 27,
"name": "teams_team_name",
"type_info": "Text"
},
{
"ordinal": 28,
"name": "ai_models",
"type_info": "VarcharArray"
},
{
"ordinal": 29,
"name": "code_completion_model",
"type_info": "Varchar"
}
],
"parameters": {
@@ -185,10 +185,10 @@
true,
true,
true,
true,
true,
true,
false,
true,
true,
true,
true
]
},
backend/.sqlx/query-ba8bde5018fdf7b12f85cd3a6557c4accfc78bf160c1277f35d9d8ddcd056963.json → backend/.sqlx/query-2b9607ed838c8c62eb0f2856420389f7be648f52edbb875ff52c96219ed3ba84.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE flow SET path = $1, summary = $2, description = $3,dependency_job = NULL, draft_only = NULL, tag = $4, dedicated_worker = $5, visible_to_runner_only = $6, on_behalf_of_email = $7, value = $8, schema = $9::text::json, edited_by = $10, edited_at = now()\n WHERE path = $11 AND workspace_id = $12",
"query": "UPDATE flow SET path = $1, summary = $2, description = $3,dependency_job = NULL, lock_error_logs = '', draft_only = NULL, tag = $4, dedicated_worker = $5, visible_to_runner_only = $6, on_behalf_of_email = $7, value = $8, schema = $9::text::json, edited_by = $10, edited_at = now()\n WHERE path = $11 AND workspace_id = $12",
"describe": {
"columns": [],
"parameters": {
@@ -21,5 +21,5 @@
},
"nullable": []
},
"hash": "ba8bde5018fdf7b12f85cd3a6557c4accfc78bf160c1277f35d9d8ddcd056963"
"hash": "2b9607ed838c8c62eb0f2856420389f7be648f52edbb875ff52c96219ed3ba84"
}
backend/.sqlx/query-55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2.json
Generated
+15 -15
View File
@@ -130,28 +130,28 @@
},
{
"ordinal": 25,
"name": "ai_models",
"type_info": "VarcharArray"
},
{
"ordinal": 26,
"name": "code_completion_model",
"type_info": "Varchar"
},
{
"ordinal": 27,
"name": "teams_command_script",
"type_info": "Text"
},
{
"ordinal": 28,
"ordinal": 26,
"name": "teams_team_id",
"type_info": "Text"
},
{
"ordinal": 29,
"ordinal": 27,
"name": "teams_team_name",
"type_info": "Text"
},
{
"ordinal": 28,
"name": "ai_models",
"type_info": "VarcharArray"
},
{
"ordinal": 29,
"name": "code_completion_model",
"type_info": "Varchar"
}
],
"parameters": {
@@ -185,10 +185,10 @@
true,
true,
true,
true,
true,
true,
false,
true,
true,
true,
true
]
},
backend/.sqlx/query-97bf27f210572499b42ce04f19f116cc87ed06c49dcca04360250ddfd89d7ab3.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT lock_error_logs FROM flow WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "lock_error_logs",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
true
]
},
"hash": "97bf27f210572499b42ce04f19f116cc87ed06c49dcca04360250ddfd89d7ab3"
}
backend/.sqlx/query-ba285edd1c1b1e400e85168ff4f05cf5281fd341096d433c7c0e5712e7726fb0.json
Generated
+16
View File
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE flow SET lock_error_logs = $1 WHERE path = $2 AND workspace_id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "ba285edd1c1b1e400e85168ff4f05cf5281fd341096d433c7c0e5712e7726fb0"
}
backend/.sqlx/query-d1876c46c0b1aba168efaebd3a056e999c400998eb699d862d718e7ab4c1f427.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE flow SET lock_error_logs = NULL WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "d1876c46c0b1aba168efaebd3a056e999c400998eb699d862d718e7ab4c1f427"
}
backend/.sqlx/query-4205d237c123d8d1c9ff2d61118027a80ccc8cd75e9703cb1d014b45f57c2be6.json → backend/.sqlx/query-e4f1ee1568ce3c186b569421c6c8a3039f73d04fc53c67c70e67371f06416ef3.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO flow (workspace_id, path, summary, description, dependency_job, draft_only, tag, dedicated_worker, visible_to_runner_only, on_behalf_of_email, value, schema, edited_by, edited_at) \n VALUES ($1, $2, $3, $4, NULL, $5, $6, $7, $8, $9, $10, $11::text::json, $12, now())",
"query": "INSERT INTO flow (workspace_id, path, summary, description, dependency_job, lock_error_logs, draft_only, tag, dedicated_worker, visible_to_runner_only, on_behalf_of_email, value, schema, edited_by, edited_at) \n VALUES ($1, $2, $3, $4, NULL, '', $5, $6, $7, $8, $9, $10, $11::text::json, $12, now())",
"describe": {
"columns": [],
"parameters": {
@@ -21,5 +21,5 @@
},
"nullable": []
},
"hash": "4205d237c123d8d1c9ff2d61118027a80ccc8cd75e9703cb1d014b45f57c2be6"
"hash": "e4f1ee1568ce3c186b569421c6c8a3039f73d04fc53c67c70e67371f06416ef3"
}
backend/.sqlx/query-fec6d5674dc6b5a6a0ece419c40508835affcb7679a48f2a443777e829bd1e74.json
Generated
+5 -5
View File
@@ -41,11 +41,11 @@
]
},
"nullable": [
true,
true,
true,
true,
true,
false,
false,
false,
false,
false,
true
]
},
backend/migrations/20250205131520_add_flow_lock_errors.down.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add down migration script here
ALTER TABLE flow DROP COLUMN lock_error_logs;
backend/migrations/20250205131520_add_flow_lock_errors.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TABLE flow ADD COLUMN lock_error_logs TEXT;
backend/tests/worker.rs
+1 -1
View File
@@ -2900,7 +2900,7 @@ async fn test_flow_lock_all(db: Pool<Postgres>) {
.await
.unwrap()
.into_inner()
.subtype_0
.open_flow
.value
.modules;
modules.into_iter()
backend/windmill-api/openapi.yaml
+25
View File
@@ -5366,6 +5366,27 @@ paths:
schema:
$ref: "#/components/schemas/Flow"
/w/{workspace}/flows/deployment_status/p/{path}:
get:
summary: get flow deployment status
operationId: getFlowDeploymentStatus
tags:
- flow
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/ScriptPath"
responses:
"200":
description: flow status
content:
application/json:
schema:
type: object
properties:
lock_error_logs:
type: string
/w/{workspace}/flows/get_triggers_count/{path}:
get:
summary: get triggers count of flow
@@ -14336,6 +14357,10 @@ components:
allOf:
- $ref: "../../openflow.openapi.yaml#/components/schemas/OpenFlow"
- $ref: "#/components/schemas/FlowMetadata"
- type: object
properties:
lock_error_logs:
type: string
ExtraPerms:
type: object
backend/windmill-api/src/ai.rs
+8 -4
View File
@@ -422,7 +422,7 @@ impl TryFrom<&str> for AIProvider {
#[derive(Deserialize, Debug)]
pub struct AIResource {
pub path: String,
pub path: Option<String>,
pub provider: AIProvider,
}
@@ -485,22 +485,26 @@ async fn proxy(
let ai_resource = serde_json::from_value::<AIResource>(ai_resource.unwrap())
.map_err(|e| Error::BadRequest(e.to_string()))?;
let path = ai_resource.path.unwrap_or("".to_string());
if path.is_empty() {
return Err(Error::BadRequest("Resource path is empty".to_string()));
}
let resource = sqlx::query_scalar!(
"SELECT value
FROM resource
WHERE path = $1 AND workspace_id = $2",
&ai_resource.path,
&path,
&w_id
)
.fetch_optional(&db)
.await?
.ok_or_else(|| {
Error::NotFound(format!(
"Could not find the {:?} resource at path {}, update the resource path in the workspace settings", ai_resource.provider, ai_resource.path
"Could not find the {:?} resource at path {}, update the resource path in the workspace settings", ai_resource.provider, path
))
})?;
(resource, ai_resource.path, ai_resource.provider)
(resource, path, ai_resource.provider)
};
if resource.is_none() {
backend/windmill-api/src/flows.rs
+30 -4
View File
@@ -59,6 +59,7 @@ pub fn workspaced_service() -> Router {
.route("/get_triggers_count/*path", get(get_triggers_count))
.route("/list_tokens/*path", get(list_tokens))
.route("/get/*path", get(get_flow_by_path))
.route("/deployment_status/p/*path", get(get_deployment_status))
.route("/get/draft/*path", get(get_flow_by_path_w_draft))
.route("/exists/*path", get(exists_flow_by_path))
.route("/list_paths", get(list_paths))
@@ -356,8 +357,8 @@ async fn create_flow(
sqlx::query!(
"INSERT INTO flow (workspace_id, path, summary, description, \
dependency_job, draft_only, tag, dedicated_worker, visible_to_runner_only, on_behalf_of_email, value, schema, edited_by, edited_at)
VALUES ($1, $2, $3, $4, NULL, $5, $6, $7, $8, $9, $10, $11::text::json, $12, now())",
dependency_job, lock_error_logs, draft_only, tag, dedicated_worker, visible_to_runner_only, on_behalf_of_email, value, schema, edited_by, edited_at)
VALUES ($1, $2, $3, $4, NULL, '', $5, $6, $7, $8, $9, $10, $11::text::json, $12, now())",
w_id,
nf.path,
nf.summary,
@@ -683,7 +684,7 @@ async fn update_flow(
sqlx::query!(
"UPDATE flow SET path = $1, summary = $2, description = $3,\
dependency_job = NULL, draft_only = NULL, tag = $4, dedicated_worker = $5, visible_to_runner_only = $6, on_behalf_of_email = $7, \
dependency_job = NULL, lock_error_logs = '', draft_only = NULL, tag = $4, dedicated_worker = $5, visible_to_runner_only = $6, on_behalf_of_email = $7, \
value = $8, schema = $9::text::json, edited_by = $10, edited_at = now()
WHERE path = $11 AND workspace_id = $12",
if is_new_path { flow_path } else { &nf.path }, // if new path, do not rename directly (to avoid flow_version foreign key constraint)
@@ -950,6 +951,31 @@ async fn list_tokens(
list_tokens_internal(&db, &w_id, &path, true).await
}
#[derive(FromRow, Serialize)]
struct DeploymentStatus {
lock_error_logs: Option<String>,
}
async fn get_deployment_status(
Extension(db): Extension<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> JsonResult<DeploymentStatus> {
let path = path.to_path();
let mut tx = db.begin().await?;
let status_o: Option<DeploymentStatus> = sqlx::query_as!(
DeploymentStatus,
"SELECT lock_error_logs FROM flow WHERE path = $1 AND workspace_id = $2",
path,
w_id,
)
.fetch_optional(&mut *tx)
.await?;
let status = not_found_if_none(status_o, "DeploymentStatus", path)?;
tx.commit().await?;
Ok(Json(status))
}
async fn get_flow_by_path(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
@@ -961,7 +987,7 @@ async fn get_flow_by_path(
let flow_o = if query.with_starred_info.unwrap_or(false) {
sqlx::query_as::<_, FlowWithStarred>(
"SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow.on_behalf_of_email, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by, favorite.path IS NOT NULL as starred
"SELECT flow.workspace_id, flow.path, flow.lock_error_logs, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow.on_behalf_of_email, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by, favorite.path IS NOT NULL as starred
FROM flow
LEFT JOIN favorite
ON favorite.favorite_kind = 'flow'
backend/windmill-api/src/workspaces.rs
+3 -2
View File
@@ -732,7 +732,8 @@ async fn edit_copilot_config(
.await?;
if let Some(cached) = AI_KEY_CACHE.get(&w_id) {
if cached.path != parsed_ai_resource.path {
if parsed_ai_resource.path.is_none() || parsed_ai_resource.path.unwrap() != cached.path
{
AI_KEY_CACHE.remove(&w_id);
}
}
@@ -1362,7 +1363,7 @@ struct UsedTriggers {
pub kafka_used: bool,
pub nats_used: bool,
pub postgres_used: bool,
pub sqs_used: bool
pub sqs_used: bool,
}
async fn get_used_triggers(
backend/windmill-common/src/flows.rs
+2
View File
@@ -60,6 +60,8 @@ pub struct FlowWithStarred {
pub flow: Flow,
#[serde(skip_serializing_if = "Option::is_none")]
pub starred: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub lock_error_logs: Option<String>,
}
fn is_none_or_false(b: &Option<bool>) -> bool {
backend/windmill-worker/src/worker_lockfiles.rs
+52 -29
View File
@@ -621,7 +621,8 @@ pub async fn handle_flow_dependency_job(
tx = clear_dependency_parent_path(&parent_path, &job_path, &job.workspace_id, "flow", tx)
.await?;
let modified_ids;
(flow.modules, tx, modified_ids) = lock_modules(
let errors;
(flow.modules, tx, modified_ids, errors) = lock_modules(
flow.modules,
job,
mem_peak,
@@ -638,6 +639,43 @@ pub async fn handle_flow_dependency_job(
occupancy_metrics,
)
.await?;
if !errors.is_empty() {
let error_message = errors
.iter()
.map(|e| format!("{}: {}", e.id, e.error))
.collect::<Vec<String>>()
.join("\n");
let logs2 = sqlx::query_scalar!(
"SELECT logs FROM job_logs WHERE job_id = $1 AND workspace_id = $2",
&job.id,
&job.workspace_id
)
.fetch_optional(db)
.await?
.flatten()
.unwrap_or_else(|| "no logs".to_string());
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.workspace_id
)
.execute(db)
.await?;
return Err(Error::ExecutionErr(format!(
"Error locking flow modules:\n{}\n\nlogs:\n{}",
error_message,
remove_ansi_codes(&logs2)
)));
} else {
sqlx::query!(
"UPDATE flow SET lock_error_logs = NULL WHERE path = $1 AND workspace_id = $2",
&job.script_path(),
&job.workspace_id
)
.execute(db)
.await?;
}
let new_flow_value = Json(serde_json::value::to_raw_value(&flow).map_err(to_anyhow)?);
// Re-check cancellation to ensure we don't accidentally override a flow.
@@ -772,6 +810,7 @@ async fn lock_modules<'c>(
Vec<FlowModule>,
sqlx::Transaction<'c, sqlx::Postgres>,
Vec<String>,
Vec<LockModuleError>,
)> {
let mut new_flow_modules = Vec::new();
let mut modified_ids = Vec::new();
@@ -802,7 +841,7 @@ async fn lock_modules<'c>(
parallelism,
} => {
let nmodules;
(nmodules, tx, nmodified_ids) = Box::pin(lock_modules(
(nmodules, tx, modified_ids, errors) = Box::pin(lock_modules(
modules,
job,
mem_peak,
@@ -835,7 +874,8 @@ async fn lock_modules<'c>(
for mut b in branches {
let nmodules;
let inner_modified_ids;
(nmodules, tx, inner_modified_ids) = Box::pin(lock_modules(
let inner_errors;
(nmodules, tx, inner_modified_ids, inner_errors) = Box::pin(lock_modules(
b.modules,
job,
mem_peak,
@@ -853,6 +893,7 @@ async fn lock_modules<'c>(
))
.await?;
nmodified_ids.extend(inner_modified_ids);
errors.extend(inner_errors);
b.modules = nmodules;
nbranches.push(b)
}
@@ -860,7 +901,7 @@ async fn lock_modules<'c>(
}
FlowModuleValue::WhileloopFlow { modules, modules_node, skip_failures } => {
let nmodules;
(nmodules, tx, nmodified_ids) = Box::pin(lock_modules(
(nmodules, tx, nmodified_ids, errors) = Box::pin(lock_modules(
modules,
job,
mem_peak,
@@ -890,8 +931,8 @@ async fn lock_modules<'c>(
for mut b in branches {
let nmodules;
let inner_modified_ids;
(nmodules, tx, inner_modified_ids) = Box::pin(lock_modules(
let inner_errors;
(nmodules, tx, inner_modified_ids, inner_errors) = Box::pin(lock_modules(
b.modules,
job,
mem_peak,
@@ -909,11 +950,13 @@ async fn lock_modules<'c>(
))
.await?;
nmodified_ids.extend(inner_modified_ids);
errors.extend(inner_errors);
b.modules = nmodules;
nbranches.push(b)
}
let ndefault;
(ndefault, tx, nmodified_ids) = Box::pin(lock_modules(
let ninner_errors;
(ndefault, tx, nmodified_ids, ninner_errors) = Box::pin(lock_modules(
default,
job,
mem_peak,
@@ -930,6 +973,7 @@ async fn lock_modules<'c>(
occupancy_metrics,
))
.await?;
errors.extend(ninner_errors);
e.value = FlowModuleValue::BranchOne {
branches: nbranches,
default: ndefault,
@@ -1048,29 +1092,8 @@ async fn lock_modules<'c>(
new_flow_modules.push(e);
continue;
}
if !errors.is_empty() {
let error_message = errors
.iter()
.map(|e| format!("{}: {}", e.id, e.error))
.collect::<Vec<String>>()
.join("\n");
let logs2 = sqlx::query_scalar!(
"SELECT logs FROM job_logs WHERE job_id = $1 AND workspace_id = $2",
&job.id,
&job.workspace_id
)
.fetch_optional(db)
.await?
.flatten()
.unwrap_or_else(|| "no logs".to_string());
return Err(Error::ExecutionErr(format!(
"Error locking flow modules:\n{}\n\nlogs:\n{}",
error_message,
remove_ansi_codes(&logs2)
)));
}
Ok((new_flow_modules, tx, modified_ids))
Ok((new_flow_modules, tx, modified_ids, errors))
}
async fn insert_flow_node<'c>(
frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte
+38 -4
View File
@@ -67,18 +67,22 @@
import PostgresTriggersPanel from '$lib/components/triggers/postgres/PostgresTriggersPanel.svelte'
import Toggle from '$lib/components/Toggle.svelte'
import SqsTriggerPanel from '$lib/components/triggers/sqs/SqsTriggerPanel.svelte'
import { onDestroy } from 'svelte'
import LogViewer from '$lib/components/LogViewer.svelte'
let flow: Flow | undefined
let can_write = false
$: path = $page.params.path
let shareModal: ShareModal
let deploymentInProgress = false
let scheduledForStr: string | undefined = undefined
let invisible_to_owner: boolean | undefined = undefined
let overrideTag: string | undefined = undefined
let inputSelected: 'saved' | 'history' | undefined = undefined
let jsonView = false
let deploymentInProgress = false
let intervalId: NodeJS.Timeout | undefined = undefined
const triggersCount = writable<TriggersCount | undefined>(undefined)
@@ -129,12 +133,31 @@
if (!flow.path.startsWith(`u/${$userStore?.username}`) && flow.path.split('/').length > 2) {
invisible_to_owner = flow.visible_to_runner_only
}
intervalId && clearInterval(intervalId)
deploymentInProgress = flow.lock_error_logs == ''
if (deploymentInProgress) {
intervalId = setInterval(syncer, 500)
}
can_write = canWrite(flow.path, flow.extra_perms!, $userStore)
}
let isValid = true
let loading = false
async function syncer(): Promise<void> {
if (flow) {
const status = await FlowService.getFlowDeploymentStatus({
workspace: $workspaceStore!,
path: flow.path
})
if (status.lock_error_logs == undefined || status.lock_error_logs != '') {
deploymentInProgress = false
flow.lock_error_logs = status.lock_error_logs
clearInterval(intervalId)
}
}
}
async function runFlow(
scheduledForStr: string | undefined,
args: Record<string, any>,
@@ -328,6 +351,10 @@
return menuItems
}
onDestroy(() => {
intervalId && clearInterval(intervalId)
})
function onKeyDown(event: KeyboardEvent) {
switch (event.key) {
case 'Enter':
@@ -442,10 +469,17 @@
</div>
{#if deploymentInProgress}
<Badge color="yellow">
<HeaderBadge color="yellow">
<Loader2 size={12} class="inline animate-spin mr-1" />
Deployment in progress
</Badge>
</HeaderBadge>
{/if}
{#if flow.lock_error_logs && flow.lock_error_logs != ''}
<div class="bg-red-100 dark:bg-red-700 border-l-4 border-red-500 p-4" role="alert">
<p class="font-bold">Error deploying this flow</p>
<p> This flow has not been deployed successfully because of the following errors: </p>
<LogViewer content={flow.lock_error_logs} isLoading={false} tag={undefined} />
</div>
{/if}
<div class="flex flex-col align-left">
@@ -587,7 +621,7 @@
<PostgresTriggersPanel path={flow.path ?? ''} isFlow />
</div>
</svelte:fragment>
<svelte:fragment slot="nats">
<div class="p-2">
<NatsTriggersPanel path={flow.path ?? ''} isFlow />