diff --git a/backend/windmill-worker/src/java_executor.rs b/backend/windmill-worker/src/java_executor.rs index 35541c3ee3..bb0518f22d 100644 --- a/backend/windmill-worker/src/java_executor.rs +++ b/backend/windmill-worker/src/java_executor.rs @@ -1,4 +1,8 @@ -use std::{collections::HashMap, path::PathBuf, process::Stdio}; +use std::{ + collections::HashMap, + path::{Path, PathBuf}, + process::Stdio, +}; use crate::global_cache::save_cache; use anyhow::{anyhow, bail}; @@ -23,12 +27,13 @@ use windmill_queue::{append_logs, CanceledBy, MiniPulledJob}; use crate::{ common::{ build_command_with_isolation, create_args_and_out_file, get_reserved_variables, - read_result, resolve_nsjail_timeout, resolve_nsjail_tmp_mount_block, start_child_process, - OccupancyMetrics, + read_result, resolve_job_timeout, resolve_nsjail_timeout, resolve_nsjail_tmp_mount_block, + start_child_process, OccupancyMetrics, }, handle_child, is_sandboxing_enabled, read_ee_registry_bool_with_workspace_override, read_ee_registry_with_workspace_override, universal_pkg_installer::{par_install_language_dependencies_all_at_once, RequiredDependency}, + worker_utils::JobPingHeartbeat, COURSIER_CACHE_DIR, DISABLE_NUSER, JAVA_CACHE_DIR, JAVA_HOME_DIR, JAVA_REPOSITORY_DIR, MAVEN_REPOS, NO_DEFAULT_MAVEN, NSJAIL_PATH, PATH_ENV, PROXY_ENVS, }; @@ -294,7 +299,21 @@ pub async fn resolve<'a>( std::env::var("TMP").unwrap_or_else(|_| String::from("/tmp")), ); } - let output = cmd.output().await?; + // The lockfile has to be read from a clean stdout, so this cannot go through handle_child's + // polling loop. That leaves resolution with neither a ping nor a time bound of its own: it + // needs the heartbeat not to be reaped as a zombie mid-resolution, and the timeout so a + // wedged registry connection cannot park the job in `running` forever. + cmd.kill_on_drop(true); + let (timeout, ..) = resolve_job_timeout(conn, w_id, *job_id, None).await; + let _heartbeat = JobPingHeartbeat::start(conn, *job_id, "java lockfile resolution"); + let output = tokio::time::timeout(timeout, cmd.output()) + .await + .map_err(|_| { + Error::ExecutionErr(format!( + "resolving the java lockfile timed out after {}s", + timeout.as_secs() + )) + })??; // Check if the command was successful if output.status.success() { String::from_utf8(output.stdout).expect("Failed to convert output to String") @@ -371,8 +390,8 @@ async fn install<'a>( ); let job_dir = job_dir.to_owned(); let fetch_dir = format!("{}/tmp-fetch-{}", *JAVA_CACHE_DIR, Uuid::new_v4()); - let fetch_dir2 = fetch_dir.clone(); - par_install_language_dependencies_all_at_once( + let (cmd_fetch_dir, postinstall_fetch_dir) = (fetch_dir.clone(), fetch_dir.clone()); + let installed = par_install_language_dependencies_all_at_once( deps, "java", "java", @@ -431,7 +450,7 @@ async fn install<'a>( "--parallel", &format!("{}", *JAVA_CONCURRENT_DOWNLOADS), "--cache", - &fetch_dir, + &cmd_fetch_dir, ]) .args(&repos) .arg("--intransitive") @@ -450,35 +469,11 @@ async fn install<'a>( Ok(cmd) }, - async move |_| { - move_to_repository(&fetch_dir2, 0).await?; - remove_dir_all(&fetch_dir2).await?; - #[async_recursion] - async fn move_to_repository(path: &str, depth: u8) -> anyhow::Result<()> { - if depth == 3 { - copy_dir_recursively( - &PathBuf::from(path), - &PathBuf::from(&*JAVA_REPOSITORY_DIR), - )?; - - return Ok(()); - } - let mut entries = tokio::fs::read_dir(path).await?; - loop { - let Some(entry) = entries.next_entry().await? else { - break Ok(()); - }; - - let path = entry - .path() - .to_str() - .ok_or(anyhow!("Internal Error: Cannot convert Path to Str"))? - .to_owned(); - - move_to_repository(&path, depth + 1).await?; - } - } - Ok(()) + async move |dependencies| { + Ok( + move_to_repository(&postinstall_fetch_dir, &*JAVA_REPOSITORY_DIR, &dependencies) + .await?, + ) }, &job.id, &job.workspace_id, @@ -486,10 +481,122 @@ async fn install<'a>( is_sandboxing_enabled(), conn, ) - .await?; + .await; + + // coursier failing against the registry bails before the postinstall step ever runs, so the + // cache it was writing into has to be reclaimed on every path out + match remove_dir_all(&fetch_dir).await { + Err(e) if e.kind() != std::io::ErrorKind::NotFound => { + tracing::warn!("could not remove java fetch dir {fetch_dir}: {e}") + } + _ => {} + } + installed?; Ok(classpath) } +/// Copies every fetched artifact out of coursier's cache into the `// +/// ` location the classpath is built from. Coursier's cache is laid out as +/// `////...`, so the depth at which the maven +/// layout starts varies with the registry url and only the coordinate can be matched on. +async fn move_to_repository( + fetch_dir: &str, + repository_dir: &str, + deps: &[RequiredDependency<()>], +) -> anyhow::Result<()> { + struct Wanted { + coordinate: Vec, + destination: String, + display_name: String, + found: bool, + } + + #[async_recursion] + async fn find_and_copy( + dir: &Path, + // components walked past below the fetch dir, compared against coordinates one component + // at a time so that windows' native separator cannot defeat the match + below: &mut Vec, + wanted: &mut Vec, + ) -> anyhow::Result<()> { + if wanted.iter().all(|w| w.found) { + return Ok(()); + } + let (mut subdirs, mut holds_artifact) = (vec![], false); + let mut entries = tokio::fs::read_dir(dir).await?; + while let Some(entry) = entries.next_entry().await? { + let name = entry.file_name().to_string_lossy().into_owned(); + if entry.file_type().await?.is_dir() { + subdirs.push((name, entry)); + } else { + // coursier's own bookkeeping (`..checked`, `..error`, checksums) is + // dot-prefixed; only a downloaded artifact is not + holds_artifact |= !name.starts_with('.'); + } + } + // a repository coursier probed and did not get the artifact from keeps a directory at the + // coordinate holding nothing but that bookkeeping, and default repositories are probed + // before the configured ones, so claiming the first match would cache an empty directory + // as the installed artifact + if holds_artifact { + if let Some(w) = wanted + .iter_mut() + .find(|w| !w.found && below.ends_with(&w.coordinate)) + { + copy_dir_recursively(dir, &PathBuf::from(&w.destination))?; + w.found = true; + return Ok(()); + } + } + for (name, entry) in subdirs { + below.push(name); + find_and_copy(&entry.path(), below, wanted).await?; + below.pop(); + } + Ok(()) + } + + let mut wanted = deps + .iter() + .map(|RequiredDependency { path, display_name, .. }| { + let suffix = path + .strip_prefix(repository_dir) + .filter(|suffix| suffix.starts_with('/')) + .ok_or_else(|| anyhow!("Internal Error: {path} is not under {repository_dir}"))?; + Ok(Wanted { + coordinate: suffix + .split('/') + .filter(|component| !component.is_empty()) + .map(str::to_owned) + .collect(), + destination: path.clone(), + display_name: display_name.clone(), + found: false, + }) + }) + .collect::>>()?; + // longest coordinate first: a group id ending in another one's coordinates (com.org.foo:bar + // over org.foo:bar) would otherwise be free to claim the shorter one's directory + wanted.sort_by_key(|w| std::cmp::Reverse(w.coordinate.len())); + + find_and_copy(&PathBuf::from(fetch_dir), &mut vec![], &mut wanted).await?; + + let missing = wanted + .iter() + .filter(|w| !w.found) + .map(|w| w.display_name.as_str()) + .sorted() + .collect_vec(); + if !missing.is_empty() { + bail!( + "the configured maven repositories did not serve: {}. \ + Coursier reported success but no artifact for them was found in its cache.", + missing.join(", ") + ); + } + Ok(()) +} + async fn compile<'a>( JobHandlerInput { occupancy_metrics, @@ -1012,3 +1119,161 @@ class Wmill { } } "#; + +#[cfg(test)] +mod tests { + use super::*; + + fn dep( + repository_dir: &str, + group_path: &str, + artifact: &str, + version: &str, + ) -> RequiredDependency<()> { + RequiredDependency { + path: format!("{repository_dir}/{group_path}/{artifact}/{version}"), + _s3_handle: format!("{}:{artifact}:{version}", group_path.replace("/", ".")), + display_name: format!("{artifact}:{version}"), + custom_payload: (), + } + } + + async fn touch(path: &str) { + let path = PathBuf::from(path); + create_dir_all(path.parent().unwrap()).await.unwrap(); + File::create(path).await.unwrap(); + } + + #[tokio::test] + async fn artifacts_are_found_whatever_the_registry_url_path_is() { + let tmp = tempfile::tempdir().unwrap(); + let fetch_dir = tmp.path().join("fetch").to_str().unwrap().to_owned(); + let repository_dir = tmp.path().join("repository").to_str().unwrap().to_owned(); + + // maven central has a single url segment before the maven layout, a nexus repository two, + // and a root-hosted mirror none at all + touch(&format!("{fetch_dir}/https/repo1.maven.org/maven2/commons-cli/commons-cli/1.4/commons-cli-1.4.jar")).await; + touch(&format!("{fetch_dir}/https/nexus.local/repository/maven-public/com/google/code/gson/gson/2.8.9/gson-2.8.9.jar")).await; + touch(&format!("{fetch_dir}/https/maven.local/org/apache/commons/commons-lang3/3.8.1/commons-lang3-3.8.1.jar")).await; + + let deps = vec![ + dep(&repository_dir, "commons-cli", "commons-cli", "1.4"), + dep(&repository_dir, "com/google/code/gson", "gson", "2.8.9"), + dep( + &repository_dir, + "org/apache/commons", + "commons-lang3", + "3.8.1", + ), + ]; + + move_to_repository(&fetch_dir, &repository_dir, &deps) + .await + .unwrap(); + + for (dep, jar) in deps.iter().zip([ + "commons-cli-1.4.jar", + "gson-2.8.9.jar", + "commons-lang3-3.8.1.jar", + ]) { + assert!( + metadata(format!("{}/{jar}", dep.path)).await.is_ok(), + "{} was not copied to {}", + dep.display_name, + dep.path + ); + } + } + + #[tokio::test] + async fn the_empty_directory_a_404_leaves_behind_does_not_claim_the_coordinate() { + let tmp = tempfile::tempdir().unwrap(); + let fetch_dir = tmp.path().join("fetch").to_str().unwrap().to_owned(); + let repository_dir = tmp.path().join("repository").to_str().unwrap().to_owned(); + + // a repository that 404s is left with an empty directory at the coordinate. Each artifact + // is served by one of the two repositories and left empty under the other, so whichever + // one the walk reaches first, an empty directory precedes an artifact. + let central = format!("{fetch_dir}/https/repo1.maven.org/maven2"); + let nexus = format!("{fetch_dir}/https/nexus.local/repository/maven-public"); + touch(&format!("{central}/com/corp/public/1.0/public-1.0.jar")).await; + touch(&format!( + "{nexus}/com/corp/public/1.0/.public-1.0.jar.error" + )) + .await; + touch(&format!("{nexus}/com/corp/internal/1.0/internal-1.0.jar")).await; + touch(&format!( + "{central}/com/corp/internal/1.0/.internal-1.0.jar.error" + )) + .await; + + let deps = vec![ + dep(&repository_dir, "com/corp", "public", "1.0"), + dep(&repository_dir, "com/corp", "internal", "1.0"), + ]; + + move_to_repository(&fetch_dir, &repository_dir, &deps) + .await + .unwrap(); + + for (dep, jar) in deps.iter().zip(["public-1.0.jar", "internal-1.0.jar"]) { + assert!( + metadata(format!("{}/{jar}", dep.path)).await.is_ok(), + "{} was not copied to {}", + dep.display_name, + dep.path + ); + } + } + + #[tokio::test] + async fn a_group_id_ending_in_another_coordinate_does_not_claim_it() { + let tmp = tempfile::tempdir().unwrap(); + let fetch_dir = tmp.path().join("fetch").to_str().unwrap().to_owned(); + let repository_dir = tmp.path().join("repository").to_str().unwrap().to_owned(); + let root = format!("{fetch_dir}/https/nexus.local/repository/maven-public"); + + touch(&format!("{root}/org/foo/bar/1/bar-1.jar")).await; + touch(&format!("{root}/com/org/foo/bar/1/bar-1.jar")).await; + + let deps = vec![ + dep(&repository_dir, "org/foo", "bar", "1"), + dep(&repository_dir, "com/org/foo", "bar", "1"), + ]; + + move_to_repository(&fetch_dir, &repository_dir, &deps) + .await + .unwrap(); + + for dep in &deps { + assert!( + metadata(format!("{}/bar-1.jar", dep.path)).await.is_ok(), + "{} was not copied to {}", + dep.display_name, + dep.path + ); + } + } + + #[tokio::test] + async fn artifact_missing_from_the_cache_is_named_in_the_error() { + let tmp = tempfile::tempdir().unwrap(); + let fetch_dir = tmp.path().join("fetch").to_str().unwrap().to_owned(); + let repository_dir = tmp.path().join("repository").to_str().unwrap().to_owned(); + + touch(&format!("{fetch_dir}/https/nexus.local/repository/maven-public/commons-cli/commons-cli/1.4/commons-cli-1.4.jar")).await; + + let deps = vec![ + dep(&repository_dir, "commons-cli", "commons-cli", "1.4"), + dep(&repository_dir, "com/google/code/gson", "gson", "2.8.9"), + ]; + + let err = move_to_repository(&fetch_dir, &repository_dir, &deps) + .await + .unwrap_err() + .to_string(); + + assert!(err.contains("gson:2.8.9"), "unexpected error: {err}"); + assert!(!err.contains("commons-cli:1.4"), "unexpected error: {err}"); + } +} diff --git a/backend/windmill-worker/src/universal_pkg_installer.rs b/backend/windmill-worker/src/universal_pkg_installer.rs index 47b3d95358..f36a5e6fc9 100644 --- a/backend/windmill-worker/src/universal_pkg_installer.rs +++ b/backend/windmill-worker/src/universal_pkg_installer.rs @@ -291,6 +291,11 @@ pub async fn par_install_language_dependencies_all_at_once< let child = start_child_process(cmd, &installer_executable_name, false).await?; let mut buf = "".to_owned(); let pipe_stdout = if stdout_on_err { Some(&mut buf) } else { None }; + // handle_child only pings the job id it is handed, and that one is nil while output is + // buffered, so nothing would refresh this job's ping for the length of the batch install. + let _heartbeat = stdout_on_err.then(|| { + crate::worker_utils::JobPingHeartbeat::start(conn, *job_id, "dependency installation") + }); if let Err(e) = crate::handle_child::handle_child( &(if pipe_stdout.is_some() {