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

Cloning repos and lockfile on the commit

Browse Source
This commit is contained in:
wendrul
2025-04-16 18:26:35 +02:00
parent b1acaf9f4c
commit 213ea0102b
3 changed files with 225 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/parsers/windmill-parser-yaml/src/lib.rs
+23
View File
@@ -231,6 +231,29 @@ pub struct AnsibleRequirements {
pub git_repos: Vec<GitRepo>,
}
impl Default for AnsibleRequirements {
fn default() -> Self {
Self {
python_reqs: vec![],
collections: None,
file_resources: vec![],
inventories: vec![],
vars: vec![],
resources: vec![],
options: AnsiblePlaybookOptions {
verbosity: None,
forks: None,
timeout: None,
flush_cache: None,
force_handlers: None,
},
vault_password_file: None,
vault_id: vec![],
git_repos: vec![],
}
}
}
fn parse_inventories(inventory_yaml: &Yaml) -> anyhow::Result<Vec<AnsibleInventory>> {
if let Yaml::Array(arr) = inventory_yaml {
let mut ret = vec![];
backend/windmill-worker/src/ansible_executor.rs
+148 -15
View File
@@ -6,6 +6,7 @@ use std::{collections::HashMap, path::PathBuf, process::Stdio};
use anyhow::anyhow;
use itertools::Itertools;
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use tokio::process::Command;
use uuid::Uuid;
@@ -53,7 +54,7 @@ async fn clone_repo(
canceled_by: &mut Option<CanceledBy>,
w_id: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<()> {
) -> error::Result<String> {
let target_path = is_allowed_file_location(job_dir, &repo.target_path)?;
let mut clone_cmd = Command::new(GIT_PATH.as_str());
@@ -63,7 +64,7 @@ async fn clone_repo(
.envs(PROXY_ENVS.clone())
.env("PATH", PATH_ENV.as_str())
.env("TZ", TZ_ENV.as_str())
.arg("clone")
.args(["clone", "--quiet"])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
if let Some(branch) = &repo.branch {
@@ -101,7 +102,7 @@ async fn clone_repo(
.env("TZ", TZ_ENV.as_str())
.arg("-C")
.arg(&target_path)
.args(["checkout", commit])
.args(["checkout", "--quiet", commit])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
@@ -124,7 +125,29 @@ async fn clone_repo(
.await?;
}
Ok(())
let mut rev_parse_cmd = Command::new(GIT_PATH.as_str());
let commit_hash_output = rev_parse_cmd
.current_dir(job_dir)
.env_clear()
.envs(PROXY_ENVS.clone())
.env("PATH", PATH_ENV.as_str())
.env("TZ", TZ_ENV.as_str())
.arg("-C")
.arg(&target_path)
.args(["rev-parse", "HEAD"])
.stderr(Stdio::piped())
.output()
.await?;
if !commit_hash_output.status.success() {
let stderr = String::from_utf8(commit_hash_output.stderr)?;
return Err(anyhow!("Error getting git repo commit hash: {stderr}").into());
}
let commit_hash = String::from_utf8(commit_hash_output.stdout)?.trim().to_string();
Ok(commit_hash)
}
pub fn create_empty_dir(path: &PathBuf) -> std::io::Result<()> {
@@ -462,6 +485,85 @@ async fn install_galaxy_collections(
Ok(())
}
#[derive(Serialize, Deserialize)]
pub struct AnsibleDependencyLocks {
pub python_lockfile: String,
pub git_repos: HashMap<String, String>, // URL to full commit hash
}
pub async fn get_git_repo_full_head_commit_hash(repo: &GitRepo) -> anyhow::Result<String> {
let mut git_cmd = Command::new(GIT_PATH.as_str());
git_cmd.args(["ls-remote", &repo.url, "HEAD"]);
let output = git_cmd.stderr(Stdio::piped()).output().await?;
if !output.status.success() {
let stderr = String::from_utf8(output.stderr)?;
return Err(anyhow!("Error getting git repo commit hash: {stderr}"));
}
let stdout = String::from_utf8(output.stdout)?;
let lines: Vec<&str> = stdout.lines().collect();
if lines.len() != 1 {
return Err(anyhow!("Unexpected output format for git ls-remote",));
}
Ok(lines
.first()
.ok_or(anyhow!(
"The HEAD commit hash was not found for repo `{}`",
&repo.url
))?
.split_whitespace()
.next()
.map(|s| s.to_string())
.ok_or(anyhow!("Unexpected output format for git ls-remote"))?)
}
pub async fn get_git_repos_lock(
repos: &Vec<GitRepo>,
job_dir: &str,
job_id: &Uuid,
worker_name: &str,
conn: &Connection,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
w_id: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> anyhow::Result<HashMap<String, String>> {
let mut ret = HashMap::new();
for repo in repos {
if repo.commit.is_some() {
ret.insert(
repo.url.to_string(),
clone_repo(
repo,
job_dir,
job_id,
worker_name,
conn,
mem_peak,
canceled_by,
w_id,
occupancy_metrics,
)
.await?,
);
} else {
ret.insert(
repo.url.to_string(),
get_git_repo_full_head_commit_hash(repo).await?,
);
}
}
Ok(ret)
}
pub async fn handle_ansible_job(
requirements_o: Option<&String>,
job_dir: &str,
@@ -485,13 +587,17 @@ pub async fn handle_ansible_job(
"ansible",
)?;
let req_lockfiles: Option<AnsibleDependencyLocks> = requirements_o
.map(|s| serde_json::from_str(s))
.transpose()?;
let (logs, reqs, playbook) = windmill_parser_yaml::parse_ansible_reqs(inner_content)?;
append_logs(&job.id, &job.workspace_id, logs, conn).await;
write_file(job_dir, "main.yml", &playbook)?;
let additional_python_paths = handle_ansible_python_deps(
job_dir,
requirements_o,
req_lockfiles.as_ref().map(|r| &r.python_lockfile),
reqs.as_ref(),
&job.workspace_id,
&job.id,
@@ -566,19 +672,46 @@ pub async fn handle_ansible_job(
}
for repo in &r.git_repos {
clone_repo(
repo,
job_dir,
append_logs(
&job.id,
worker_name,
conn,
mem_peak,
canceled_by,
&job.workspace_id,
occupancy_metrics,
format!("\nCloning {}...", &repo.url),
conn,
)
.await
.map_err(|e| anyhow!("Failed to clone git repo `{}`: {e}", repo.url))?;
.await;
if let Some(full_commit_hash) = req_lockfiles
.as_ref()
.and_then(|r| r.git_repos.get(&repo.url))
{
clone_repo_without_history(
repo,
full_commit_hash,
job_dir,
&job.id,
worker_name,
conn,
mem_peak,
canceled_by,
&job.workspace_id,
occupancy_metrics,
)
.await
.map_err(|e| anyhow!("Failed to clone git repo `{}`: {e}", repo.url))?;
} else {
clone_repo(
repo,
job_dir,
&job.id,
worker_name,
conn,
mem_peak,
canceled_by,
&job.workspace_id,
occupancy_metrics,
)
.await
.map_err(|e| anyhow!("Failed to clone git repo `{}`: {e}", repo.url))?;
}
append_logs(
&job.id,
backend/windmill-worker/src/worker_lockfiles.rs
+54 -6
View File
@@ -3,6 +3,8 @@ use std::collections::HashMap;
use std::fs::{create_dir_all, remove_dir_all};
use std::path::{Component, Path, PathBuf};
#[cfg(feature = "python")]
use crate::ansible_executor::{get_git_repos_lock, AnsibleDependencyLocks};
use async_recursion::async_recursion;
use serde_json::value::RawValue;
use serde_json::{json, Value};
@@ -18,6 +20,8 @@ use windmill_common::scripts::ScriptHash;
#[cfg(feature = "python")]
use windmill_common::worker::PythonAnnotations;
use windmill_common::worker::{to_raw_value, to_raw_value_owned, write_file, Connection};
#[cfg(feature = "python")]
use windmill_parser_yaml::AnsibleRequirements;
use windmill_common::{
apps::AppScriptId,
@@ -1804,6 +1808,53 @@ async fn python_dep(
req
}
#[cfg(feature = "python")]
async fn ansible_dep(
reqs: AnsibleRequirements,
job_id: &Uuid,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
db: &sqlx::Pool<sqlx::Postgres>,
worker_name: &str,
w_id: &str,
worker_dir: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> std::result::Result<String, Error> {
let python_lockfile = python_dep(
reqs.python_reqs.join("\n").to_string(),
job_id,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
w_id,
worker_dir,
&mut Some(occupancy_metrics),
None,
PythonAnnotations::default(),
)
.await?;
let git_repos = get_git_repos_lock(
&reqs.git_repos,
job_dir,
job_id,
worker_name,
&Connection::Sql(db.clone()),
mem_peak,
canceled_by,
w_id,
occupancy_metrics,
)
.await?;
let ansible_lockfile = AnsibleDependencyLocks { python_lockfile, git_repos };
serde_json::to_string(&ansible_lockfile).map_err(|e| e.into())
}
async fn capture_dependency_job(
job_id: &Uuid,
job_language: &ScriptLang,
@@ -1888,10 +1939,9 @@ async fn capture_dependency_job(
));
}
let (_logs, reqs, _) = windmill_parser_yaml::parse_ansible_reqs(job_raw_code)?;
let reqs = reqs.map(|r| r.python_reqs.join("\n")).unwrap_or_default();
python_dep(
reqs,
ansible_dep(
reqs.unwrap_or_default(),
job_id,
mem_peak,
canceled_by,
@@ -1900,9 +1950,7 @@ async fn capture_dependency_job(
worker_name,
w_id,
worker_dir,
&mut Some(occupancy_metrics),
None,
PythonAnnotations::default(),
occupancy_metrics,
)
.await
}