diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 2c3170d5f3..16cc7a8dd6 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -0eccae6a9a9ecde09816cd4d88ca9ab305659e4c +ec458b7fd7947e1a8f0c1b87dc0a1d724fe78131 \ No newline at end of file diff --git a/backend/migrations/20260218000000_add_snapshotbuild_job_kind.down.sql b/backend/migrations/20260218000000_add_snapshotbuild_job_kind.down.sql new file mode 100644 index 0000000000..d70c9dd9aa --- /dev/null +++ b/backend/migrations/20260218000000_add_snapshotbuild_job_kind.down.sql @@ -0,0 +1 @@ +-- Postgres enum values cannot be removed; this is intentionally empty. diff --git a/backend/migrations/20260218000000_add_snapshotbuild_job_kind.up.sql b/backend/migrations/20260218000000_add_snapshotbuild_job_kind.up.sql new file mode 100644 index 0000000000..b455e7e599 --- /dev/null +++ b/backend/migrations/20260218000000_add_snapshotbuild_job_kind.up.sql @@ -0,0 +1 @@ +ALTER TYPE JOB_KIND ADD VALUE IF NOT EXISTS 'snapshotbuild'; diff --git a/backend/windmill-api/src/sandbox.rs b/backend/windmill-api/src/sandbox.rs index b75f5d921a..d03ba384e4 100644 --- a/backend/windmill-api/src/sandbox.rs +++ b/backend/windmill-api/src/sandbox.rs @@ -5,12 +5,17 @@ use axum::{ Json, Router, }; use serde::Deserialize; +use serde_json::value::RawValue as JsonRawValue; +use std::collections::HashMap; use windmill_common::{ db::DB, error::{self, JsonResult}, + users::username_to_permissioned_as, utils::require_admin, }; +use windmill_queue::{push, PushArgs, PushIsolationLevel}; use windmill_sandbox::{SandboxSnapshot, SandboxVolume}; +use windmill_types::jobs::JobPayload; use crate::db::ApiAuthed; @@ -109,9 +114,31 @@ async fn create_snapshot( .execute(&db) .await?; + let job_uuid = push_snapshot_build_job( + &db, + &w_id, + &body.name, + &body.tag, + &body.docker_image, + body.setup_script.as_deref(), + &authed, + ) + .await?; + + sqlx::query!( + "UPDATE sandbox_snapshot SET build_job_id = $4 \ + WHERE workspace_id = $1 AND name = $2 AND tag = $3", + &w_id, + &body.name, + &body.tag, + job_uuid, + ) + .execute(&db) + .await?; + Ok(format!( - "Snapshot {}:{} created with status 'pending'", - body.name, body.tag + "Snapshot {}:{} created with build job {}", + body.name, body.tag, job_uuid )) } @@ -193,6 +220,19 @@ async fn rebuild_snapshot( Path((w_id, name, tag)): Path<(String, String, String)>, ) -> error::Result { require_admin(authed.is_admin, &authed.username)?; + + let row = sqlx::query!( + "SELECT docker_image, setup_script FROM sandbox_snapshot \ + WHERE workspace_id = $1 AND name = $2 AND tag = $3", + &w_id, + &name, + &tag, + ) + .fetch_optional(&db) + .await?; + + let row = windmill_common::utils::not_found_if_none(row, "sandbox_snapshot", &name)?; + sqlx::query!( "UPDATE sandbox_snapshot SET status = 'pending', build_error = NULL, updated_at = now() \ WHERE workspace_id = $1 AND name = $2 AND tag = $3", @@ -202,7 +242,33 @@ async fn rebuild_snapshot( ) .execute(&db) .await?; - Ok(format!("Rebuild queued for snapshot {}:{}", name, tag)) + + let job_uuid = push_snapshot_build_job( + &db, + &w_id, + &name, + &tag, + &row.docker_image, + row.setup_script.as_deref(), + &authed, + ) + .await?; + + sqlx::query!( + "UPDATE sandbox_snapshot SET build_job_id = $4 \ + WHERE workspace_id = $1 AND name = $2 AND tag = $3", + &w_id, + &name, + &tag, + job_uuid, + ) + .execute(&db) + .await?; + + Ok(format!( + "Rebuild queued for snapshot {}:{} with build job {}", + name, tag, job_uuid + )) } async fn upload_snapshot( @@ -216,6 +282,73 @@ async fn upload_snapshot( windmill_sandbox::upload_snapshot_bytes(&db, &w_id, &name, &tag, body, &authed.username).await } +async fn push_snapshot_build_job( + db: &DB, + w_id: &str, + name: &str, + tag: &str, + docker_image: &str, + setup_script: Option<&str>, + authed: &ApiAuthed, +) -> error::Result { + let mut args = HashMap::new(); + args.insert( + "snapshot_name".to_string(), + JsonRawValue::from_string(serde_json::to_string(name).unwrap()).unwrap(), + ); + args.insert( + "snapshot_tag".to_string(), + JsonRawValue::from_string(serde_json::to_string(tag).unwrap()).unwrap(), + ); + args.insert( + "docker_image".to_string(), + JsonRawValue::from_string(serde_json::to_string(docker_image).unwrap()).unwrap(), + ); + args.insert( + "setup_script".to_string(), + JsonRawValue::from_string(serde_json::to_string(&setup_script).unwrap()).unwrap(), + ); + + let tx = PushIsolationLevel::IsolatedRoot(db.clone()); + let (job_uuid, tx) = push( + db, + tx, + w_id, + JobPayload::SnapshotBuild { + snapshot_name: name.to_string(), + snapshot_tag: tag.to_string(), + }, + PushArgs { args: &args, extra: None }, + &authed.username, + &authed.email, + username_to_permissioned_as(&authed.username), + None, + None, + None, + None, + None, + None, + None, + false, + false, + None, + true, + None, + None, + None, + None, + None, + false, + None, + None, + None, + ) + .await?; + tx.commit().await?; + + Ok(job_uuid) +} + // --- Volumes --- async fn list_volumes( diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index eaa75f04c1..76e1c8a40a 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -5202,6 +5202,11 @@ async fn push_inner<'c, 'd>( job_kind: JobKind::AIAgent, ..Default::default() }, + JobPayload::SnapshotBuild { snapshot_name, snapshot_tag } => JobPayloadUntagged { + runnable_path: Some(format!("sandbox/snapshot/{snapshot_name}:{snapshot_tag}")), + job_kind: JobKind::SnapshotBuild, + ..Default::default() + }, }; // Enforce concurrency limit on all dependency jobs. @@ -5303,6 +5308,7 @@ async fn push_inner<'c, 'd>( || job_kind == JobKind::FlowDependencies || job_kind == JobKind::DeploymentCallback || job_kind == JobKind::AppDependencies + || job_kind == JobKind::SnapshotBuild { // using the dependency tag for deployment callback for now. We can create a separate tag when we need "dependency".to_string() @@ -5662,6 +5668,7 @@ async fn push_inner<'c, 'd>( JobKind::UnassignedScript => "jobs.run.unassigned_script", JobKind::UnassignedFlow => "jobs.run.unassigned_flow", JobKind::UnassignedSinglestepFlow => "jobs.run.unassigned_singlestepflow", + JobKind::SnapshotBuild => "jobs.run.snapshot_build", }; let audit_author = if format!("u/{user}") != permissioned_as && user != permissioned_as { diff --git a/backend/windmill-sandbox/src/s3_oss.rs b/backend/windmill-sandbox/src/s3_oss.rs index 64adc910d6..f9f11ef5d6 100644 --- a/backend/windmill-sandbox/src/s3_oss.rs +++ b/backend/windmill-sandbox/src/s3_oss.rs @@ -223,6 +223,164 @@ pub async fn upload_volume( )) } +#[cfg(all(not(feature = "private"), feature = "parquet"))] +pub async fn build_snapshot( + w_id: &str, + name: &str, + tag: &str, + docker_image: &str, + setup_script: Option<&str>, + db: &windmill_common::DB, +) -> windmill_common::error::Result<()> { + use windmill_common::error::Error; + use tokio::process::Command; + + sqlx::query!( + "UPDATE sandbox_snapshot SET status = 'building', updated_at = now() \ + WHERE workspace_id = $1 AND name = $2 AND tag = $3", + w_id, + name, + tag, + ) + .execute(db) + .await?; + + let result = async { + let temp_dir = tempfile::tempdir() + .map_err(|e| Error::ExecutionErr(format!("Failed to create temp dir: {e}")))?; + let rootfs_dir = temp_dir.path().join("rootfs"); + tokio::fs::create_dir_all(&rootfs_dir) + .await + .map_err(|e| Error::ExecutionErr(format!("Failed to create rootfs dir: {e}")))?; + + tracing::info!("Exporting docker image {docker_image} for snapshot {name}:{tag}"); + let crane_output = Command::new("crane") + .args(["export", docker_image, "-"]) + .output() + .await + .map_err(|e| Error::ExecutionErr(format!("Failed to run crane: {e}")))?; + + if !crane_output.status.success() { + let stderr = String::from_utf8_lossy(&crane_output.stderr); + return Err(Error::ExecutionErr(format!("crane export failed: {stderr}"))); + } + + let crane_bytes = crane_output.stdout; + let rootfs_dir_clone = rootfs_dir.clone(); + tokio::task::spawn_blocking(move || -> windmill_common::error::Result<()> { + use std::io::Cursor; + use tar::Archive; + let mut archive = Archive::new(Cursor::new(crane_bytes)); + archive + .unpack(&rootfs_dir_clone) + .map_err(|e| Error::ExecutionErr(format!("Failed to unpack crane output: {e}")))?; + Ok(()) + }) + .await + .map_err(|e| Error::ExecutionErr(format!("Spawn blocking failed: {e}")))??; + + if let Some(script) = setup_script { + if !script.trim().is_empty() { + tracing::info!("Running setup script for snapshot {name}:{tag}"); + let output = Command::new("chroot") + .args([ + &rootfs_dir.to_string_lossy().to_string(), + "/bin/sh", + "-c", + script, + ]) + .output() + .await + .map_err(|e| Error::ExecutionErr(format!("Failed to run setup: {e}")))?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(Error::ExecutionErr(format!("Setup script failed: {stderr}"))); + } + } + } + + let rootfs_dir_clone = rootfs_dir.clone(); + let (bytes, content_hash) = + tokio::task::spawn_blocking(move || -> windmill_common::error::Result<(Vec, String)> { + use sha2::{Digest, Sha256}; + + let bytes = crate::tar_gz(&rootfs_dir_clone)?; + let hash = format!("{:x}", Sha256::digest(&bytes)); + Ok((bytes, hash)) + }) + .await + .map_err(|e| Error::ExecutionErr(format!("Spawn blocking failed: {e}")))??; + + let size = bytes.len(); + if size > CE_SNAPSHOT_SIZE_LIMIT { + return Err(Error::ExecutionErr(format!( + "Snapshot size ({:.1} MB) exceeds the {} MB limit. \ + Upgrade to Windmill EE for unlimited snapshot sizes.", + size as f64 / 1_048_576.0, + CE_SNAPSHOT_SIZE_LIMIT / 1_048_576 + ))); + } + + let s3_key = format!("sandbox/snapshots/{w_id}/{name}/{content_hash}.tar.gz"); + let size_i64 = size as i64; + + let os = windmill_object_store::get_workspace_object_store(db, w_id).await?; + + windmill_object_store::put_bytes_to_store(os, &s3_key, bytes.into()).await?; + + sqlx::query!( + "UPDATE sandbox_snapshot SET \ + s3_key = $4, content_hash = $5, size_bytes = $6, status = 'ready', \ + build_error = NULL, updated_at = now() \ + WHERE workspace_id = $1 AND name = $2 AND tag = $3", + w_id, + name, + tag, + &s3_key, + &content_hash, + size_i64, + ) + .execute(db) + .await?; + + tracing::info!("Snapshot {name}:{tag} built successfully ({size_i64} bytes)"); + Ok(()) + } + .await; + + if let Err(ref e) = result { + let err_str = e.to_string(); + sqlx::query!( + "UPDATE sandbox_snapshot SET status = 'failed', build_error = $4, updated_at = now() \ + WHERE workspace_id = $1 AND name = $2 AND tag = $3", + w_id, + name, + tag, + &err_str, + ) + .execute(db) + .await + .ok(); + } + + result +} + +#[cfg(all(not(feature = "private"), not(feature = "parquet")))] +pub async fn build_snapshot( + _w_id: &str, + _name: &str, + _tag: &str, + _docker_image: &str, + _setup_script: Option<&str>, + _db: &windmill_common::DB, +) -> windmill_common::error::Result<()> { + Err(windmill_common::error::Error::ExecutionErr( + "Sandbox snapshot builds require the parquet feature (S3 object store)".to_string(), + )) +} + #[cfg(all(not(feature = "private"), feature = "parquet"))] pub async fn upload_snapshot_bytes( db: &windmill_common::DB, diff --git a/backend/windmill-types/src/jobs.rs b/backend/windmill-types/src/jobs.rs index ffd3cc6ef3..ae6656d81a 100644 --- a/backend/windmill-types/src/jobs.rs +++ b/backend/windmill-types/src/jobs.rs @@ -92,6 +92,7 @@ pub enum JobKind { #[serde(rename = "unassigned_singlestepflow")] #[sqlx(rename = "unassigned_singlestepflow")] UnassignedSinglestepFlow, + SnapshotBuild, } #[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Copy, Clone)] @@ -445,6 +446,10 @@ pub enum JobPayload { AIAgent { path: String, }, + SnapshotBuild { + snapshot_name: String, + snapshot_tag: String, + }, } #[derive(Clone, Serialize, Deserialize, Debug)] @@ -491,6 +496,7 @@ impl JobPayload { JobPayload::RawScriptDependencies { .. } => JobKind::Dependencies, JobPayload::RawFlowDependencies { .. } => JobKind::FlowDependencies, JobPayload::DeploymentCallback { .. } => JobKind::DeploymentCallback, + JobPayload::SnapshotBuild { .. } => JobKind::SnapshotBuild, JobPayload::Flow { .. } | JobPayload::RestartedFlow { .. } => JobKind::Flow, } } diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs index 4d5a139b78..8300a03e1c 100644 --- a/backend/windmill-worker/src/bash_executor.rs +++ b/backend/windmill-worker/src/bash_executor.rs @@ -112,7 +112,7 @@ cleanup() {{ rm -f bp 2>/dev/null # Kill the process group of the script (negative PID value) - pkill -P $$ 2>/dev/null || kill -- -$$ 2>/dev/null || kill 0 2>/dev/null || true + pkill -P $$ 2>/dev/null || true exit }} @@ -140,7 +140,7 @@ wait $tail_pid 2>/dev/null || true # Clean up the named pipe and background processes rm -f bp -pkill -P $$ 2>/dev/null || kill -- -$$ 2>/dev/null || kill 0 2>/dev/null || true +pkill -P $$ 2>/dev/null || true # Exit with the captured status exit $exit_status diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index 0b56b45959..ce9c3790f8 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -69,6 +69,7 @@ mod rust_executor; #[cfg(test)] mod sandbox_setup; mod sanitized_sql_params; +mod snapshot_build; mod schema; pub mod sql_utils; mod universal_pkg_installer; diff --git a/backend/windmill-worker/src/snapshot_build.rs b/backend/windmill-worker/src/snapshot_build.rs new file mode 100644 index 0000000000..3214e3ebcc --- /dev/null +++ b/backend/windmill-worker/src/snapshot_build.rs @@ -0,0 +1,85 @@ +use serde_json::value::RawValue; +use sqlx::Pool; +use sqlx::Postgres; +use windmill_common::worker::Connection; +use windmill_queue::{append_logs, empty_result, MiniPulledJob}; + +pub async fn handle_snapshot_build( + job: &MiniPulledJob, + db: &Pool, + conn: &Connection, +) -> windmill_common::error::Result> { + let args = job + .args + .as_ref() + .ok_or_else(|| windmill_common::error::Error::InternalErr("Missing args".to_string()))?; + + let snapshot_name = args + .get("snapshot_name") + .and_then(|v| serde_json::from_str::(v.get()).ok()) + .ok_or_else(|| { + windmill_common::error::Error::InternalErr("Missing snapshot_name arg".to_string()) + })?; + + let snapshot_tag = args + .get("snapshot_tag") + .and_then(|v| serde_json::from_str::(v.get()).ok()) + .ok_or_else(|| { + windmill_common::error::Error::InternalErr("Missing snapshot_tag arg".to_string()) + })?; + + let docker_image = args + .get("docker_image") + .and_then(|v| serde_json::from_str::(v.get()).ok()) + .ok_or_else(|| { + windmill_common::error::Error::InternalErr("Missing docker_image arg".to_string()) + })?; + + let setup_script: Option = args + .get("setup_script") + .and_then(|v| serde_json::from_str(v.get()).ok()); + + append_logs( + &job.id, + job.workspace_id.clone(), + format!( + "Starting snapshot build {snapshot_name}:{snapshot_tag} from image {docker_image}\n" + ), + conn, + ) + .await; + + let result = windmill_sandbox::build_snapshot( + &job.workspace_id, + &snapshot_name, + &snapshot_tag, + &docker_image, + setup_script.as_deref(), + db, + ) + .await; + + match &result { + Ok(()) => { + append_logs( + &job.id, + job.workspace_id.clone(), + format!("Snapshot {snapshot_name}:{snapshot_tag} built successfully\n"), + conn, + ) + .await; + } + Err(e) => { + append_logs( + &job.id, + job.workspace_id.clone(), + format!("Snapshot build failed: {e}\n"), + conn, + ) + .await; + } + } + + result?; + Ok(empty_result()) +} diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 1a8522a3f2..735201d293 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -3413,6 +3413,17 @@ pub async fn handle_queued_job( )); } }, + JobKind::SnapshotBuild => match conn { + Connection::Sql(db) => { + Box::pin(crate::snapshot_build::handle_snapshot_build(&job, db, conn)) + .await + } + Connection::Http(_) => { + return Err(Error::internal_err( + "Could not handle snapshot build job with agent worker".to_string(), + )); + } + }, _ => { let metric_timer = Instant::now(); let preview_data = preview_data.and_then(|data| match data { @@ -3619,6 +3630,7 @@ async fn try_validate_schema( JobKind::UnassignedScript => 16, JobKind::UnassignedFlow => 17, JobKind::UnassignedSinglestepFlow => 18, + JobKind::SnapshotBuild => 19, }; let sv = match job.runnable_id { diff --git a/frontend/src/lib/components/runs/JobDetailFieldConfig.ts b/frontend/src/lib/components/runs/JobDetailFieldConfig.ts index 23a7df12cd..09f6158c43 100644 --- a/frontend/src/lib/components/runs/JobDetailFieldConfig.ts +++ b/frontend/src/lib/components/runs/JobDetailFieldConfig.ts @@ -72,6 +72,7 @@ export function getJobCategory(job: Job): JobCategory { case 'dependencies': case 'flowdependencies': case 'appdependencies': + case 'snapshotbuild': return 'dependencies' case 'flownode': diff --git a/frontend/src/lib/components/runs/JobDetailHeader.svelte b/frontend/src/lib/components/runs/JobDetailHeader.svelte index 7ca973dd00..de1eda917d 100644 --- a/frontend/src/lib/components/runs/JobDetailHeader.svelte +++ b/frontend/src/lib/components/runs/JobDetailHeader.svelte @@ -134,7 +134,9 @@ ? 'Script from hub' : job.job_kind === 'aiagent' ? 'AI Agent' - : job.job_kind || 'Unknown job type' + : job.job_kind === 'snapshotbuild' + ? 'snapshot build' + : job.job_kind || 'Unknown job type' } diff --git a/frontend/src/lib/components/runs/useJobsLoader.svelte.ts b/frontend/src/lib/components/runs/useJobsLoader.svelte.ts index f26f5c6d56..777201ce64 100644 --- a/frontend/src/lib/components/runs/useJobsLoader.svelte.ts +++ b/frontend/src/lib/components/runs/useJobsLoader.svelte.ts @@ -21,7 +21,7 @@ export function computeJobKinds(jobKindsCat: string | null): string { if (jobKindsCat == 'all') { return '' } else if (jobKindsCat == 'dependencies') { - let kinds: CompletedJob['job_kind'][] = ['dependencies', 'flowdependencies', 'appdependencies'] + let kinds: CompletedJob['job_kind'][] = ['dependencies', 'flowdependencies', 'appdependencies', 'snapshotbuild'] return kinds.join(',') } else if (jobKindsCat == 'previews') { let kinds: CompletedJob['job_kind'][] = ['preview', 'flowpreview'] diff --git a/frontend/src/lib/components/workspaceSettings/CreateWorkspace.svelte b/frontend/src/lib/components/workspaceSettings/CreateWorkspace.svelte index 82f95a03c6..2850399e0b 100644 --- a/frontend/src/lib/components/workspaceSettings/CreateWorkspace.svelte +++ b/frontend/src/lib/components/workspaceSettings/CreateWorkspace.svelte @@ -298,7 +298,7 @@ let automateUsernameCreation = $state(false) async function getAutomateUsernameCreationSetting() { automateUsernameCreation = - ((await SettingService.getGlobal({ key: 'automate_username_creation' })) as any) ?? false + ((await SettingService.getGlobal({ key: 'automate_username_creation' })) as any) ?? true if (!automateUsernameCreation) { UserService.globalWhoami().then((x) => { diff --git a/frontend/src/lib/utils.ts b/frontend/src/lib/utils.ts index 1ddebf1aec..4e7daa1aba 100644 --- a/frontend/src/lib/utils.ts +++ b/frontend/src/lib/utils.ts @@ -1824,7 +1824,8 @@ export function getJobKindIcon(jobKind: Job['job_kind']) { } else if ( jobKind === 'dependencies' || jobKind === 'appdependencies' || - jobKind === 'flowdependencies' + jobKind === 'flowdependencies' || + jobKind === 'snapshotbuild' ) { return Package } else if (