From cfee559d08f964dda4aec0260504c69e3a27177f Mon Sep 17 00:00:00 2001 From: Abel Lucas Date: Mon, 20 Jan 2025 17:59:50 +0100 Subject: [PATCH] backend: finalize v2 migration (v2 phase 4 - final) --- ...44181b2ab8440d0523b46935256d06d734ebc.json | 34 +++++ backend/windmill-api/src/db.rs | 138 ++++++++++++++++-- backend/windmill-common/src/worker.rs | 9 +- backend/windmill-queue/src/jobs.rs | 39 ++++- 4 files changed, 202 insertions(+), 18 deletions(-) create mode 100644 backend/.sqlx/query-d2720448c5c7e38659ce4577d2d44181b2ab8440d0523b46935256d06d734ebc.json diff --git a/backend/.sqlx/query-d2720448c5c7e38659ce4577d2d44181b2ab8440d0523b46935256d06d734ebc.json b/backend/.sqlx/query-d2720448c5c7e38659ce4577d2d44181b2ab8440d0523b46935256d06d734ebc.json new file mode 100644 index 0000000000..0b9f4b8653 --- /dev/null +++ b/backend/.sqlx/query-d2720448c5c7e38659ce4577d2d44181b2ab8440d0523b46935256d06d734ebc.json @@ -0,0 +1,34 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO v2_job_completed AS cj\n ( workspace_id\n , id\n , started_at\n , duration_ms\n , result\n , canceled_by\n , canceled_reason\n , flow_status\n , memory_peak\n , status\n , labels\n )\n VALUES ($1, $2, $3, COALESCE($12::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($3, now()))))*1000), $5, $7, $8, $9,$11, CASE WHEN $6::BOOL THEN 'canceled'::job_status\n WHEN $10::BOOL THEN 'skipped'::job_status\n WHEN $4::BOOL THEN 'success'::job_status\n ELSE 'failure'::job_status END, $13)\n ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $5 RETURNING duration_ms AS \"duration_ms!\"", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "duration_ms!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Varchar", + "Uuid", + "Timestamptz", + "Bool", + "Jsonb", + "Bool", + "Varchar", + "Text", + "Jsonb", + "Bool", + "Int4", + "Int8", + "VarcharArray" + ] + }, + "nullable": [ + false + ] + }, + "hash": "d2720448c5c7e38659ce4577d2d44181b2ab8440d0523b46935256d06d734ebc" +} diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 203003aab6..5344c2ee8d 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -6,14 +6,16 @@ * LICENSE-AGPL for a copy of the license. */ -use futures::FutureExt; -use sqlx::Executor; +use std::sync::atomic::Ordering; +use std::time::Duration; +use futures::FutureExt; use sqlx::{ migrate::{Migrate, MigrateError}, pool::PoolConnection, - PgConnection, Pool, Postgres, + Executor, PgConnection, Pool, Postgres, }; + use windmill_audit::audit_ee::{AuditAuthor, AuditAuthorable}; use windmill_common::utils::generate_lock_id; use windmill_common::{ @@ -199,6 +201,24 @@ pub async fn migrate(db: &DB) -> Result<(), Error> { } }); + let db2 = db.clone(); + let _ = tokio::task::spawn(async move { + use windmill_common::worker::MIN_VERSION_IS_LATEST; + loop { + if !MIN_VERSION_IS_LATEST.load(Ordering::Relaxed) { + tokio::time::sleep(Duration::from_secs(30)).await; + continue; + } + if let Err(err) = v2_finalize(&db2).await { + tracing::error!("{err:#}: Could not apply v2 finalize migration, retry in 30s.."); + tokio::time::sleep(Duration::from_secs(30)).await; + continue; + } + tracing::info!("v2 finalization step successfully applied."); + break; + } + }); + Ok(()) } @@ -250,7 +270,7 @@ async fn fix_flow_versioning_migration( } macro_rules! run_windmill_migration { - ($migration_job_name:expr, $db:expr, $code:block) => { + ($migration_job_name:expr, $db:expr, |$tx:ident| $code:block) => { { let migration_job_name = $migration_job_name; let db: &Pool = $db; @@ -264,11 +284,11 @@ macro_rules! run_windmill_migration { .unwrap_or(false); if !has_done_migration { tracing::info!("Applying {migration_job_name} migration"); - let mut tx = db.begin().await?; + let mut $tx = db.begin().await?; let mut r = false; while !r { r = sqlx::query_scalar!("SELECT pg_try_advisory_lock(4242)") - .fetch_one(&mut *tx) + .fetch_one(&mut *$tx) .await .map_err(|e| { tracing::error!("Error acquiring {migration_job_name} lock: {e:#}"); @@ -298,7 +318,7 @@ macro_rules! run_windmill_migration { "INSERT INTO windmill_migrations (name) VALUES ($1) ON CONFLICT DO NOTHING", migration_job_name ) - .execute(&mut *tx) + .execute(&mut *$tx) .await?; tracing::info!("Finished applying {migration_job_name} migration"); } else { @@ -306,9 +326,9 @@ macro_rules! run_windmill_migration { } let _ = sqlx::query("SELECT pg_advisory_unlock(4242)") - .execute(&mut *tx) + .execute(&mut *$tx) .await?; - tx.commit().await?; + $tx.commit().await?; tracing::info!("released lock for {migration_job_name}"); } else { tracing::debug!("migration {migration_job_name} already done"); @@ -318,6 +338,96 @@ macro_rules! run_windmill_migration { }; } +async fn v2_finalize(db: &DB) -> Result<(), Error> { + run_windmill_migration!("v2_finalize_disable_sync", db, |tx| { + tx.execute( + r#" + DROP FUNCTION v2_job_completed_before_insert() CASCADE; + DROP FUNCTION v2_job_flow_runtime_before_insert() CASCADE; + DROP FUNCTION v2_job_flow_runtime_before_update() CASCADE; + DROP FUNCTION v2_job_queue_after_insert() CASCADE; + DROP FUNCTION v2_job_queue_before_insert() CASCADE; + DROP FUNCTION v2_job_queue_before_update() CASCADE; + DROP FUNCTION v2_job_runtime_before_insert() CASCADE; + DROP FUNCTION v2_job_runtime_before_update() CASCADE; + + DROP VIEW completed_job, completed_job_view, job, queue, queue_view CASCADE; + "#, + ) + .await?; + }); + run_windmill_migration!("v2_finalize_job_queue", db, |tx| { + tx.execute( + r#" + ALTER TABLE v2_job_queue + DROP COLUMN __parent_job CASCADE, + DROP COLUMN __created_by CASCADE, + DROP COLUMN __script_hash CASCADE, + DROP COLUMN __script_path CASCADE, + DROP COLUMN __args CASCADE, + DROP COLUMN __logs CASCADE, + DROP COLUMN __raw_code CASCADE, + DROP COLUMN __canceled CASCADE, + DROP COLUMN __last_ping CASCADE, + DROP COLUMN __job_kind CASCADE, + DROP COLUMN __env_id CASCADE, + DROP COLUMN __schedule_path CASCADE, + DROP COLUMN __permissioned_as CASCADE, + DROP COLUMN __flow_status CASCADE, + DROP COLUMN __raw_flow CASCADE, + DROP COLUMN __is_flow_step CASCADE, + DROP COLUMN __language CASCADE, + DROP COLUMN __same_worker CASCADE, + DROP COLUMN __raw_lock CASCADE, + DROP COLUMN __pre_run_error CASCADE, + DROP COLUMN __email CASCADE, + DROP COLUMN __visible_to_owner CASCADE, + DROP COLUMN __mem_peak CASCADE, + DROP COLUMN __root_job CASCADE, + DROP COLUMN __leaf_jobs CASCADE, + DROP COLUMN __concurrent_limit CASCADE, + DROP COLUMN __concurrency_time_window_s CASCADE, + DROP COLUMN __timeout CASCADE, + DROP COLUMN __flow_step_id CASCADE, + DROP COLUMN __cache_ttl CASCADE; + "#, + ) + .await?; + }); + run_windmill_migration!("v2_finalize_job_completed", db, |tx| { + tx.execute( + r#" + ALTER TABLE v2_job_completed + DROP COLUMN __parent_job CASCADE, + DROP COLUMN __created_by CASCADE, + DROP COLUMN __created_at CASCADE, + DROP COLUMN __success CASCADE, + DROP COLUMN __script_hash CASCADE, + DROP COLUMN __script_path CASCADE, + DROP COLUMN __args CASCADE, + DROP COLUMN __logs CASCADE, + DROP COLUMN __raw_code CASCADE, + DROP COLUMN __canceled CASCADE, + DROP COLUMN __job_kind CASCADE, + DROP COLUMN __env_id CASCADE, + DROP COLUMN __schedule_path CASCADE, + DROP COLUMN __permissioned_as CASCADE, + DROP COLUMN __raw_flow CASCADE, + DROP COLUMN __is_flow_step CASCADE, + DROP COLUMN __language CASCADE, + DROP COLUMN __is_skipped CASCADE, + DROP COLUMN __raw_lock CASCADE, + DROP COLUMN __email CASCADE, + DROP COLUMN __visible_to_owner CASCADE, + DROP COLUMN __tag CASCADE, + DROP COLUMN __priority CASCADE; + "#, + ) + .await?; + }); + Ok(()) +} + async fn fix_job_completed_index(db: &DB) -> Result<(), Error> { // let has_done_migration = sqlx::query_scalar!( // "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_job_completed_index')" @@ -360,7 +470,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> { // tx.commit().await?; // } - run_windmill_migration!("fix_job_completed_index_2", &db, { + run_windmill_migration!("fix_job_completed_index_2", &db, |tx| { // sqlx::query( // "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new_2 ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, created_at DESC)" // ).execute(db).await?; @@ -380,7 +490,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> { .await?; }); - run_windmill_migration!("fix_job_completed_index_3", &db, { + run_windmill_migration!("fix_job_completed_index_3", &db, |tx| { sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_job_on_schedule_path") .execute(db) .await?; @@ -398,7 +508,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> { .await?; }); - run_windmill_migration!("fix_job_index_1", &db, { + run_windmill_migration!("fix_job_index_1", &db, |tx| { let migration_job_name = "fix_job_completed_index_4"; let mut i = 1; tracing::info!("step {i} of {migration_job_name} migration"); @@ -479,7 +589,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> { .await?; }); - run_windmill_migration!("fix_labeled_jobs_index", &db, { + run_windmill_migration!("fix_labeled_jobs_index", &db, |tx| { tracing::info!("Special migration to add index concurrently on job labels 2"); sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS labeled_jobs_on_jobs") .execute(db) @@ -489,7 +599,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> { ).execute(db).await?; }); - run_windmill_migration!("v2_fix_labeled_jobs_index", &db, { + run_windmill_migration!("v2_fix_labeled_jobs_index", &db, |tx| { tracing::info!("Special migration to add index concurrently on job labels v2"); sqlx::query!( "CREATE INDEX CONCURRENTLY v2_labeled_jobs_on_jobs ON v2_job_completed diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 67c05ef53d..4c40fcdafe 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -12,7 +12,10 @@ use std::{ io::Write, path::{Component, Path, PathBuf}, str::FromStr, - sync::{atomic::AtomicBool, Arc}, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, }; use tokio::sync::RwLock; use windmill_macros::annotations; @@ -103,6 +106,8 @@ lazy_static::lazy_static! { pub static ref DISABLE_FLOW_SCRIPT: bool = std::env::var("DISABLE_FLOW_SCRIPT").ok().is_some_and(|x| x == "1" || x == "true"); } +pub static MIN_VERSION_IS_LATEST: AtomicBool = AtomicBool::new(false); + fn format_pull_query(peek: String) -> String { format!( "WITH peek AS ( @@ -658,6 +663,8 @@ pub async fn update_min_version<'c, E: sqlx::Executor<'c, Database = sqlx::Postg tracing::info!("Minimal worker version: {min_version}"); } + MIN_VERSION_IS_LATEST.store(min_version == cur_version, Ordering::Relaxed); + *MIN_VERSION_IS_AT_LEAST_1_427.write().await = min_version >= Version::new(1, 427, 0); *MIN_VERSION_IS_AT_LEAST_1_432.write().await = min_version >= Version::new(1, 432, 0); *MIN_VERSION_IS_AT_LEAST_1_440.write().await = min_version >= Version::new(1, 440, 0); diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 86edb86014..10e82e8f47 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -396,23 +396,43 @@ pub struct WrappedError { pub trait ValidableJson { fn is_valid_json(&self) -> bool; + fn wm_labels(&self) -> Option>; +} + +#[derive(serde::Deserialize)] +struct ResultLabels { + wm_labels: Vec, } impl ValidableJson for WrappedError { fn is_valid_json(&self) -> bool { true } + + fn wm_labels(&self) -> Option> { + None + } } impl ValidableJson for Box { fn is_valid_json(&self) -> bool { !self.get().is_empty() } + + fn wm_labels(&self) -> Option> { + serde_json::from_str::(self.get()) + .ok() + .map(|r| r.wm_labels) + } } -impl ValidableJson for Arc> { +impl ValidableJson for Arc { fn is_valid_json(&self) -> bool { - !self.get().is_empty() + T::is_valid_json(&self) + } + + fn wm_labels(&self) -> Option> { + T::wm_labels(&self) } } @@ -420,12 +440,22 @@ impl ValidableJson for serde_json::Value { fn is_valid_json(&self) -> bool { true } + + fn wm_labels(&self) -> Option> { + serde_json::from_value::(self.clone()) + .ok() + .map(|r| r.wm_labels) + } } impl ValidableJson for Json { fn is_valid_json(&self) -> bool { self.0.is_valid_json() } + + fn wm_labels(&self) -> Option> { + self.0.wm_labels() + } } pub async fn register_metric( @@ -547,6 +577,7 @@ pub async fn add_completed_job( serde_json::to_string(&result).unwrap_or_else(|_| "".to_string()) ); + let labels = result.wm_labels(); let mem_peak = mem_peak.max(queued_job.mem_peak.unwrap_or(0)); // add_time!(bench, "add_completed_job query START"); @@ -562,12 +593,13 @@ pub async fn add_completed_job( , flow_status , memory_peak , status + , labels ) VALUES ($1, $2, $3, COALESCE($12::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($3, now()))))*1000), $5, $7, $8, $9,\ $11, CASE WHEN $6::BOOL THEN 'canceled'::job_status WHEN $10::BOOL THEN 'skipped'::job_status WHEN $4::BOOL THEN 'success'::job_status - ELSE 'failure'::job_status END) + ELSE 'failure'::job_status END, $13) ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $5 RETURNING duration_ms AS \"duration_ms!\"", /* $1 */ queued_job.workspace_id, /* $2 */ queued_job.id, @@ -581,6 +613,7 @@ pub async fn add_completed_job( /* $10 */ skipped, /* $11 */ if mem_peak > 0 { Some(mem_peak) } else { None }, /* $12 */ duration, + /* $13 */ labels.as_ref().map(Vec::as_slice), ) .fetch_one(&mut *tx) .await