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

latest ref

Browse Source
This commit is contained in:
Ruben Fiszel
2026-02-18 16:35:12 +00:00
parent 39512a3953
commit c0abb961e8
16 changed files with 418 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
0eccae6a9a9ecde09816cd4d88ca9ab305659e4c
ec458b7fd7947e1a8f0c1b87dc0a1d724fe78131
backend/migrations/20260218000000_add_snapshotbuild_job_kind.down.sql
+1
View File
@@ -0,0 +1 @@
-- Postgres enum values cannot be removed; this is intentionally empty.
backend/migrations/20260218000000_add_snapshotbuild_job_kind.up.sql
+1
View File
@@ -0,0 +1 @@
ALTER TYPE JOB_KIND ADD VALUE IF NOT EXISTS 'snapshotbuild';
backend/windmill-api/src/sandbox.rs
+136 -3
View File
@@ -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<String> {
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<uuid::Uuid> {
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(
backend/windmill-queue/src/jobs.rs
+7
View File
@@ -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 {
backend/windmill-sandbox/src/s3_oss.rs
+158
View File
@@ -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<u8>, 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,
backend/windmill-types/src/jobs.rs
+6
View File
@@ -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,
}
}
backend/windmill-worker/src/bash_executor.rs
+2 -2
View File
@@ -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
backend/windmill-worker/src/lib.rs
+1
View File
@@ -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;
backend/windmill-worker/src/snapshot_build.rs
+85
View File
@@ -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<Postgres>,
conn: &Connection,
) -> windmill_common::error::Result<Box<RawValue>> {
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::<String>(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::<String>(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::<String>(v.get()).ok())
.ok_or_else(|| {
windmill_common::error::Error::InternalErr("Missing docker_image arg".to_string())
})?;
let setup_script: Option<String> = 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())
}
backend/windmill-worker/src/worker.rs
+12
View File
@@ -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 {
frontend/src/lib/components/runs/JobDetailFieldConfig.ts
+1
View File
@@ -72,6 +72,7 @@ export function getJobCategory(job: Job): JobCategory {
case 'dependencies':
case 'flowdependencies':
case 'appdependencies':
case 'snapshotbuild':
return 'dependencies'
case 'flownode':
frontend/src/lib/components/runs/JobDetailHeader.svelte
+3 -1
View File
@@ -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'
}
</script>
frontend/src/lib/components/runs/useJobsLoader.svelte.ts
+1 -1
View File
@@ -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']
frontend/src/lib/components/workspaceSettings/CreateWorkspace.svelte
+1 -1
View File
@@ -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) => {
frontend/src/lib/utils.ts
+2 -1
View File
@@ -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 (