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

feat: Ansible improvements (vault, roles and git repos) (#5655)

Browse Source 
* Ansible vault + roles

* Clone arbitrary repos

* Fix cloning logic after merge

* Make function for cloning without history any commit

* Cloning repos and lockfile on the commit

* Improve error messages

* Create lockfile for roles and collections

* Simplify ansible ssh identity interface

* Ansible vault password: pass just a variable instead of 2 step approach

* Lock lockfiles for roles and collections

* fix typo

* Change git ssh identity section name

* Rename variable

* Update init script for ansible

* Suppress error when no roles

---------

Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
wendrul
2025-04-24 16:04:28 +02:00
committed by GitHub
parent c4148d7756
commit fdd1642ce1
7 changed files with 1043 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/parsers/windmill-parser-yaml/src/lib.rs
+198 -18
View File
@@ -1,3 +1,5 @@
use std::collections::HashMap;
use anyhow::anyhow;
use serde_json::json;
use windmill_parser::{Arg, MainArgSignature, ObjectProperty, Typ};
@@ -208,15 +210,52 @@ pub struct AnsibleInventory {
resource_type: Option<String>,
pub pinned_resource: Option<String>,
}
#[derive(Debug, Clone)]
pub struct GitRepo {
pub url: String,
pub commit: Option<String>,
pub branch: Option<String>,
pub target_path: String,
}
#[derive(Debug, Clone)]
pub struct AnsibleRequirements {
pub python_reqs: Vec<String>,
pub collections: Option<String>,
pub roles_and_collections: Option<String>,
pub file_resources: Vec<FileResource>,
pub inventories: Vec<AnsibleInventory>,
pub vars: Vec<(String, String)>,
pub resources: Vec<(String, String)>,
pub options: AnsiblePlaybookOptions,
pub vault_password: Option<String>,
pub vault_id: Vec<String>,
pub git_repos: Vec<GitRepo>,
pub git_ssh_identity: Vec<String>,
}
impl Default for AnsibleRequirements {
fn default() -> Self {
Self {
python_reqs: vec![],
roles_and_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: None,
vault_id: vec![],
git_repos: vec![],
git_ssh_identity: vec![],
}
}
}
fn parse_inventories(inventory_yaml: &Yaml) -> anyhow::Result<Vec<AnsibleInventory>> {
@@ -275,22 +314,7 @@ pub fn parse_ansible_reqs(
return Ok((logs, None, inner_content.to_string()));
}
let opts = AnsiblePlaybookOptions {
verbosity: None,
forks: None,
timeout: None,
flush_cache: None,
force_handlers: None,
};
let mut ret = AnsibleRequirements {
python_reqs: vec![],
collections: None,
file_resources: vec![],
inventories: vec![],
vars: vec![],
resources: vec![],
options: opts,
};
let mut ret = AnsibleRequirements::default();
if let Yaml::Hash(doc) = &docs[0] {
for (key, value) in doc {
@@ -303,7 +327,7 @@ pub fn parse_ansible_reqs(
let mut out_str = String::new();
let mut emitter = YamlEmitter::new(&mut out_str);
emitter.dump(galaxy_requirements)?;
ret.collections = Some(out_str);
ret.roles_and_collections = Some(out_str);
}
if let Some(Yaml::Array(py_reqs)) =
deps.get(&Yaml::String("python".to_string()))
@@ -345,11 +369,60 @@ pub fn parse_ansible_reqs(
Yaml::String(key) if key == "inventory" => {
ret.inventories = parse_inventories(value)?;
}
Yaml::String(key) if key == "vault_password" => {
let Yaml::String(filename) = value else {
return Err(anyhow!(
"Vault Password File expects a String containing the file name"
));
};
ret.vault_password = Some(filename.to_string());
}
Yaml::String(key) if key == "vault_id" => {
let Yaml::Array(filenames) = value else {
return Err(anyhow!("Vault ID field expects an array of strings in the format: `label@filename`"));
};
for f in filenames {
let Yaml::String(filename) = f else {
return Err(anyhow!("The elements of the vault_id field should be strings in the format: `label@filename`"));
};
ret.vault_id.push(filename.to_string());
}
}
Yaml::String(key) if key == "options" => {
if let Yaml::Array(opts) = &value {
ret.options = parse_ansible_options(opts);
}
}
Yaml::String(key) if key == "git_repos" => {
let Yaml::Array(repos) = &value else {
return Err(anyhow!("git_repos field expects an array of repos"));
};
for r in repos {
ret.git_repos.push(
parse_git_repo(r)
.map_err(|e| anyhow!("Failed to parse git repo: {e}"))?,
);
}
}
Yaml::String(key) if key == "git_ssh_identity" => {
let Yaml::Array(indentities) = &value else {
return Err(anyhow!(
"git_ssh_identity expects an array of windmill variables (or secrets) containing ssh IDs"
));
};
for r in indentities {
let Yaml::String(file_name) = r else {
return Err(anyhow!(
"Git ssh identity file must be a string path to a Windmill variable/secret"
));
};
ret.git_ssh_identity.push(file_name.clone());
}
}
Yaml::String(key) => logs.push_str(&format!("\nUnknown field `{}`. Ignoring", key)),
_ => (),
}
@@ -364,6 +437,38 @@ pub fn parse_ansible_reqs(
Ok((logs, Some(ret), out_str))
}
fn parse_git_repo(r: &Yaml) -> anyhow::Result<GitRepo> {
let Yaml::Hash(repo) = r else {
return Err(anyhow!("Should be a Map"));
};
let url = repo
.get(&Yaml::String("url".to_string()))
.and_then(|v| v.as_str())
.map(|s| s.to_string())
.ok_or(anyhow!("Expected `url` field"))?;
let target_path = repo
.get(&Yaml::String("target".to_string()))
.and_then(|v| v.as_str())
.map(|s| s.to_string())
.ok_or(anyhow!(
"Expected `target` field (target directory for cloning the repo)"
))?;
let branch = repo
.get(&Yaml::String("branch".to_string()))
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let commit = repo
.get(&Yaml::String("commit".to_string()))
.and_then(|v| v.as_str())
.map(|s| s.to_string());
Ok(GitRepo { url, commit, branch, target_path })
}
fn parse_ansible_options(opts: &Vec<Yaml>) -> AnsiblePlaybookOptions {
let mut ret = AnsiblePlaybookOptions {
verbosity: None,
@@ -529,3 +634,78 @@ fn yaml_to_json(yaml: &Yaml) -> serde_json::Value {
_ => serde_json::Value::Null,
}
}
fn update_versions(
section: &str,
yaml: &mut Yaml,
versions: &HashMap<String, String>,
) -> anyhow::Result<String> {
let mut logs = String::new();
let Yaml::Hash(ref mut m) = yaml else {
return Err(anyhow!("{section} dependency should be a map"));
};
if let Some(Yaml::Array(elements)) = m.get_mut(&Yaml::String(section.to_string())) {
for el in elements {
let Yaml::Hash(ref mut h) = el else {
return Err(anyhow!("{section} dependency element should be a map"));
};
if let Some(name) = h
.get(&Yaml::String("name".to_string()))
.and_then(|n| n.as_str())
{
if let Some(version) = versions.get(name) {
h.insert(
Yaml::String("version".to_string()),
Yaml::String(version.to_string()),
);
} else {
logs.push_str(&format!("WARNING: {section} dependency `{name}` has no locked version, using the latest or system installed version.\n"));
}
} else {
return Err(anyhow!(
"{section} dependency element: missing or invalid `name` field"
));
}
}
}
Ok(logs)
}
pub fn add_versions_to_requirements_yaml(
input: &str,
role_versions: &HashMap<String, String>,
collection_versions: &HashMap<String, String>,
) -> anyhow::Result<(String,String)> {
let mut docs =
YamlLoader::load_from_str(input).map_err(|e| anyhow!("YAML parse error: {}", e))?;
let doc = &mut docs[0];
let mut logs = String::new();
logs.push_str(
&update_versions("roles", doc, role_versions)
.map_err(|e| anyhow!("Error updating role versions: {e}"))?,
);
logs.push_str(
&update_versions("collections", doc, collection_versions)
.map_err(|e| anyhow!("Error updating collection versions: {e}"))?,
);
if !logs.is_empty() {
logs.push_str("WARNING: You might want to try adding manual versions for these, otherwise there could be breaking changes on deployed scripts\n");
}
let mut out_str = String::new();
{
let mut emitter = YamlEmitter::new(&mut out_str);
emitter
.dump(doc)
.map_err(|e| anyhow!("YAML emit error: {}", e))?;
}
Ok((out_str, logs))
}
backend/windmill-common/src/worker.rs
+13 -3
View File
@@ -381,11 +381,10 @@ fn normalize_path(path: &Path) -> PathBuf {
}
ret
}
pub fn write_file_at_user_defined_location(
pub fn is_allowed_file_location(
job_dir: &str,
user_defined_path: &str,
content: &str,
mode: Option<u32>,
) -> error::Result<PathBuf> {
let job_dir = Path::new(job_dir);
let user_path = PathBuf::from(user_defined_path);
@@ -405,6 +404,17 @@ pub fn write_file_at_user_defined_location(
.into());
}
Ok(normalized_full_path)
}
pub fn write_file_at_user_defined_location(
job_dir: &str,
user_defined_path: &str,
content: &str,
mode: Option<u32>,
) -> error::Result<PathBuf> {
let normalized_full_path = is_allowed_file_location(job_dir, user_defined_path)?;
let full_path = normalized_full_path.as_path();
if let Some(parent_dir) = full_path.parent() {
std::fs::create_dir_all(parent_dir)?;
backend/windmill-worker/src/ansible_executor.rs
+690 -26
View File
@@ -5,19 +5,22 @@ use std::{collections::HashMap, os::unix::fs::PermissionsExt, path::PathBuf, pro
use std::{collections::HashMap, path::PathBuf, process::Stdio};
use anyhow::anyhow;
use futures::future::try_join_all;
use itertools::Itertools;
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use tokio::process::Command;
use uuid::Uuid;
use windmill_common::{
error,
worker::{
to_raw_value, write_file, write_file_at_user_defined_location, Connection, WORKER_CONFIG,
is_allowed_file_location, to_raw_value, write_file, write_file_at_user_defined_location,
Connection, WORKER_CONFIG,
},
};
use windmill_queue::MiniPulledJob;
use windmill_parser_yaml::{AnsibleRequirements, ResourceOrVariablePath};
use windmill_parser_yaml::{AnsibleRequirements, GitRepo, ResourceOrVariablePath};
use windmill_queue::{append_logs, CanceledBy};
use crate::{
@@ -28,8 +31,8 @@ use crate::{
},
handle_child::handle_child,
python_executor::{create_dependencies_dir, handle_python_reqs, uv_pip_compile, PyVersion},
AuthedClient, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NSJAIL_PATH, PATH_ENV, PROXY_ENVS,
PY_INSTALL_DIR, TZ_ENV,
AuthedClient, DISABLE_NSJAIL, DISABLE_NUSER, GIT_PATH, HOME_ENV, NSJAIL_PATH, PATH_ENV,
PROXY_ENVS, PY_INSTALL_DIR, TZ_ENV,
};
lazy_static::lazy_static! {
@@ -41,7 +44,294 @@ lazy_static::lazy_static! {
}
const NSJAIL_CONFIG_RUN_ANSIBLE_CONTENT: &str = include_str!("../nsjail/run.ansible.config.proto");
const WINDMILL_ANSIBLE_PASSWORD_FILENAME: &str = ".windmill.ansible_vault_password_file";
async fn clone_repo(
repo: &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,
git_ssh_cmd: &str,
) -> 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());
clone_cmd
.current_dir(job_dir)
.env_clear()
.envs(PROXY_ENVS.clone())
.env("PATH", PATH_ENV.as_str())
.env("TZ", TZ_ENV.as_str())
.env("GIT_SSH_COMMAND", git_ssh_cmd)
.args(["clone", "--quiet"])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
if let Some(branch) = &repo.branch {
clone_cmd.args(["--branch", branch]);
}
clone_cmd.arg(&repo.url);
clone_cmd.arg(&target_path);
let clone_cmd_child = start_child_process(clone_cmd, GIT_PATH.as_str()).await?;
handle_child(
job_id,
conn,
mem_peak,
canceled_by,
clone_cmd_child,
false,
worker_name,
w_id,
"git clone",
None,
false,
&mut Some(occupancy_metrics),
None,
)
.await?;
// Checkout specific commit if provided
if let Some(commit) = &repo.commit {
let mut checkout_cmd = Command::new(GIT_PATH.as_str());
checkout_cmd
.current_dir(job_dir)
.env_clear()
.envs(PROXY_ENVS.clone())
.env("PATH", PATH_ENV.as_str())
.env("TZ", TZ_ENV.as_str())
.env("GIT_SSH_COMMAND", git_ssh_cmd)
.arg("-C")
.arg(&target_path)
.args(["checkout", "--quiet", commit])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let checkout_cmd_child = start_child_process(checkout_cmd, GIT_PATH.as_str()).await?;
handle_child(
job_id,
conn,
mem_peak,
canceled_by,
checkout_cmd_child,
false,
worker_name,
w_id,
"git checkout",
None,
false,
&mut Some(occupancy_metrics),
None,
)
.await?;
}
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())
.env("GIT_SSH_COMMAND", git_ssh_cmd)
.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<()> {
if path.exists() {
if path.is_dir() {
let mut entries = std::fs::read_dir(&path)?;
if entries.next().is_some() {
return Err(std::io::Error::new(
std::io::ErrorKind::AlreadyExists,
format!(
"Directory '{}' already exists and is not empty",
path.display()
),
));
}
Ok(())
} else {
Err(std::io::Error::new(
std::io::ErrorKind::AlreadyExists,
format!("Path '{}' exists and is not a directory", path.display()),
))
}
} else {
std::fs::create_dir_all(path)
}
}
async fn clone_repo_without_history(
repo: &GitRepo,
full_commit: &str,
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,
git_ssh_cmd: &str,
) -> error::Result<()> {
let target_path = is_allowed_file_location(job_dir, &repo.target_path)?;
create_empty_dir(&target_path)?;
let mut init_cmd = Command::new(GIT_PATH.as_str());
init_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(["init", "--quiet"])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
if let Some(branch) = &repo.branch {
init_cmd.args(["--initial-branch", branch]);
}
let init_cmd_child = start_child_process(init_cmd, GIT_PATH.as_str()).await?;
handle_child(
job_id,
conn,
mem_peak,
canceled_by,
init_cmd_child,
false,
worker_name,
w_id,
"git init",
None,
false,
&mut Some(occupancy_metrics),
None,
)
.await?;
let mut add_remote_cmd = Command::new(GIT_PATH.as_str());
add_remote_cmd
.current_dir(job_dir)
.env_clear()
.envs(PROXY_ENVS.clone())
.env("PATH", PATH_ENV.as_str())
.env("TZ", TZ_ENV.as_str())
.env("GIT_SSH_COMMAND", git_ssh_cmd)
.arg("-C")
.arg(&target_path)
.args(vec!["remote", "add", "origin", &repo.url])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let add_remote_cmd_child = start_child_process(add_remote_cmd, GIT_PATH.as_str()).await?;
handle_child(
job_id,
conn,
mem_peak,
canceled_by,
add_remote_cmd_child,
false,
worker_name,
w_id,
"git add remote",
None,
false,
&mut Some(occupancy_metrics),
None,
)
.await?;
let mut fetch_cmd = Command::new(GIT_PATH.as_str());
fetch_cmd
.current_dir(job_dir)
.env_clear()
.envs(PROXY_ENVS.clone())
.env("PATH", PATH_ENV.as_str())
.env("TZ", TZ_ENV.as_str())
.env("GIT_SSH_COMMAND", git_ssh_cmd)
.arg("-C")
.arg(&target_path)
.args(vec!["fetch", "--depth=1", "--quiet", "origin", full_commit])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let fetch_cmd_child = start_child_process(fetch_cmd, GIT_PATH.as_str()).await?;
handle_child(
job_id,
conn,
mem_peak,
canceled_by,
fetch_cmd_child,
false,
worker_name,
w_id,
"git fetch",
None,
false,
&mut Some(occupancy_metrics),
None,
)
.await?;
let mut checkout_cmd = Command::new(GIT_PATH.as_str());
checkout_cmd
.current_dir(job_dir)
.env_clear()
.envs(PROXY_ENVS.clone())
.env("PATH", PATH_ENV.as_str())
.env("TZ", TZ_ENV.as_str())
.env("GIT_SSH_COMMAND", git_ssh_cmd)
.arg("-C")
.arg(&target_path)
.args(["checkout", "--quiet", "FETCH_HEAD"])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let checkout_cmd_child = start_child_process(checkout_cmd, GIT_PATH.as_str()).await?;
handle_child(
job_id,
conn,
mem_peak,
canceled_by,
checkout_cmd_child,
false,
worker_name,
w_id,
"git checkout",
None,
false,
&mut Some(occupancy_metrics),
None,
)
.await?;
Ok(())
}
async fn handle_ansible_python_deps(
job_dir: &str,
requirements_o: Option<&String>,
@@ -118,7 +408,7 @@ async fn handle_ansible_python_deps(
Ok(additional_python_paths)
}
async fn install_galaxy_collections(
pub async fn install_galaxy_collections(
collections_yml: &str,
job_dir: &str,
job_id: &Uuid,
@@ -128,6 +418,7 @@ async fn install_galaxy_collections(
canceled_by: &mut Option<CanceledBy>,
conn: &Connection,
occupancy_metrics: &mut OccupancyMetrics,
git_ssh_cmd: &str,
) -> anyhow::Result<()> {
write_file(job_dir, "requirements.yml", collections_yml)?;
@@ -138,15 +429,52 @@ async fn install_galaxy_collections(
conn,
)
.await;
let mut galaxy_command = Command::new(ANSIBLE_GALAXY_PATH.as_str());
galaxy_command
let mut galaxy_roles_cmd = Command::new(ANSIBLE_GALAXY_PATH.as_str());
galaxy_roles_cmd
.current_dir(job_dir)
.env_clear()
.envs(PROXY_ENVS.clone())
.env("PATH", PATH_ENV.as_str())
.env("TZ", TZ_ENV.as_str())
// .env("BASE_INTERNAL_URL", base_internal_url)
// .env("HOME", HOME_ENV.as_str())
.env("GIT_SSH_COMMAND", git_ssh_cmd)
.args(vec![
"role",
"install",
"-r",
"requirements.yml",
"-p",
"./roles",
])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let child = start_child_process(galaxy_roles_cmd, ANSIBLE_GALAXY_PATH.as_str()).await?;
handle_child(
job_id,
conn,
mem_peak,
canceled_by,
child,
!*DISABLE_NSJAIL,
worker_name,
w_id,
"ansible-galaxy role install",
None,
false,
&mut Some(occupancy_metrics),
None,
)
.await?;
let mut galaxy_collections_cmd = Command::new(ANSIBLE_GALAXY_PATH.as_str());
galaxy_collections_cmd
.current_dir(job_dir)
.env_clear()
.envs(PROXY_ENVS.clone())
.env("PATH", PATH_ENV.as_str())
.env("TZ", TZ_ENV.as_str())
.env("GIT_SSH_COMMAND", git_ssh_cmd)
.args(vec![
"collection",
"install",
@@ -158,7 +486,7 @@ async fn install_galaxy_collections(
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let child = start_child_process(galaxy_command, ANSIBLE_GALAXY_PATH.as_str()).await?;
let child = start_child_process(galaxy_collections_cmd, ANSIBLE_GALAXY_PATH.as_str()).await?;
handle_child(
job_id,
conn,
@@ -168,7 +496,7 @@ async fn install_galaxy_collections(
!*DISABLE_NSJAIL,
worker_name,
w_id,
"ansible galaxy install",
"ansible-galaxy collection install",
None,
false,
&mut Some(occupancy_metrics),
@@ -179,6 +507,258 @@ 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 collections_and_roles: String,
pub collections_and_roles_logs: String,
// pub collection_versions: HashMap<String, String>, //
// pub role_versions: HashMap<String, String>,
}
pub async fn get_collection_locks(
job_dir: &str,
) -> anyhow::Result<(HashMap<String, String>, String)> {
let mut ansible_cmd = Command::new(ANSIBLE_GALAXY_PATH.as_str());
ansible_cmd
.current_dir(job_dir)
.args(["collection", "list", "--format", "json", "-p", "./"]);
let output = ansible_cmd.output().await?;
let mut ret = HashMap::new();
let mut logs = String::new();
if !output.status.success() {
let stderr = String::from_utf8(output.stderr)?;
return Err(anyhow!(
"Error getting ansible collection versions: {stderr}"
));
}
let stdout = String::from_utf8(output.stdout)?;
let val: serde_json::Value = serde_json::from_str(&stdout)?;
let Some(own_collections) = val.get(format!("{}/ansible_collections", job_dir)) else {
return Ok((ret, logs));
};
let collections = own_collections.as_object().ok_or(anyhow!(
"Expected an object (map) for the `ansible-galaxy collection list` command output and got {}",
own_collections
))?;
for (c_name, c) in collections.iter() {
if let Some(v) = c.get("version").and_then(|v| v.as_str()) {
// TODO: Check if version is not something like `(undefined)`
ret.insert(c_name.clone(), v.to_string());
} else {
logs.push_str(&format!("Failed to get version for collection `{}`. Expected an object with a string in the `version` field but received {}\n", c_name, c));
}
}
Ok((ret, logs))
}
pub async fn get_role_locks(job_dir: &str) -> anyhow::Result<(HashMap<String, String>, String)> {
let mut ansible_cmd = Command::new(ANSIBLE_GALAXY_PATH.as_str());
ansible_cmd
.current_dir(job_dir)
.args(["role", "list", "-p", "./roles"]);
let output = ansible_cmd.output().await?;
let mut ret = HashMap::new();
let mut logs = String::new();
if !output.status.success() {
let stderr = String::from_utf8(output.stderr)?;
logs.push_str(&format!("Error getting ansible role versions: {stderr}"));
return Ok((ret, logs));
}
let stdout = String::from_utf8(output.stdout)?;
let mut lines = stdout.lines();
while let Some(line) = lines.next() {
if line == format!("# {}/roles", job_dir) {
break;
}
}
for line in lines {
let line = line.strip_prefix("-").unwrap_or(line);
let mut cols = line.split(",");
if let Some(name) = cols.next().map(|n| n.trim()) {
if let Some(version) = cols.next().map(|v| v.trim()) {
// TODO: Check if version is not something like `(undefined)`
ret.insert(name.to_string(), version.to_string());
} else {
logs.push_str(&format!("Failed to get version for role `{}`.", name));
}
}
}
Ok((ret, logs))
}
pub async fn get_git_repo_full_head_commit_hash(
repo: &GitRepo,
git_ssh_cmd: &str,
) -> anyhow::Result<String> {
let mut git_cmd = Command::new(GIT_PATH.as_str());
git_cmd
.env("GIT_SSH_COMMAND", git_ssh_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,
git_ssh_cmd: &str,
) -> 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,
git_ssh_cmd,
)
.await?,
);
} else {
ret.insert(
repo.url.to_string(),
get_git_repo_full_head_commit_hash(repo, git_ssh_cmd).await?,
);
}
}
Ok(ret)
}
pub fn create_ansible_cfg(
reqs: Option<&AnsibleRequirements>,
job_dir: &str,
vault_password_file_exists: bool,
) -> error::Result<()> {
let mut passwords_cfg = String::new();
if vault_password_file_exists {
passwords_cfg.push_str(&format!(
"vault_password_file = {WINDMILL_ANSIBLE_PASSWORD_FILENAME}\n"
));
}
if let Some(vault_ids) = reqs.as_ref().map(|r| &r.vault_id) {
if !vault_ids.is_empty() {
let password_files = vault_ids.join(",");
passwords_cfg.push_str(&format!("vault_identity_list = {password_files}\n"));
}
}
let ansible_cfg_content = format!(
r#"
[defaults]
collections_path = ./
roles_path = ./roles
home={job_dir}/.ansible
local_tmp={job_dir}/.ansible/tmp
remote_tmp={job_dir}/.ansible/tmp
{passwords_cfg}
"#
);
write_file(job_dir, "ansible.cfg", &ansible_cfg_content)?;
Ok(())
}
pub async fn get_git_ssh_cmd(
reqs: &AnsibleRequirements,
job_dir: &str,
client: &AuthedClient,
) -> error::Result<String> {
let ssh_id_files = try_join_all(reqs.git_ssh_identity.iter().enumerate().map(
async |(i, var_path)| -> error::Result<String> {
let id_file_name = format!(".ssh_id_priv_{}", i);
let loc = is_allowed_file_location(job_dir, &id_file_name)?;
let mut content = client.get_variable_value(var_path).await.map_err(|e| {
error::Error::NotFound(format!(
"Variable {var_path} not found for git ssh identity: {e:#}"
))
})?;
content.push_str("\n");
let file = write_file(job_dir, &id_file_name, &content)?;
#[cfg(unix)]
{
let perm = std::os::unix::fs::PermissionsExt::from_mode(0o600);
file.set_permissions(perm)?;
}
Ok(format!(
" -i '{}'",
loc.to_string_lossy().replace('\'', r"'\''")
))
},
))
.await?;
let git_ssh_cmd = format!("ssh -o StrictHostKeyChecking=no{}", ssh_id_files.join(""));
Ok(git_ssh_cmd)
}
pub async fn handle_ansible_job(
requirements_o: Option<&String>,
job_dir: &str,
@@ -202,13 +782,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,
@@ -221,6 +805,11 @@ pub async fn handle_ansible_job(
)
.await?;
let git_ssh_cmd = &match &reqs {
Some(r) => get_git_ssh_cmd(r, job_dir, client).await?,
None => "ssh".to_string(),
};
let interpolated_args;
if let Some(args) = &job.args {
let mut args = args.0.clone();
@@ -268,7 +857,7 @@ pub async fn handle_ansible_job(
.unwrap_or_else(|| vec![]);
let mut nsjail_extra_mounts = vec![];
if let Some(r) = reqs {
if let Some(r) = reqs.as_ref() {
if let Some(db) = conn.as_sql() {
nsjail_extra_mounts = create_file_resources(
&job.id,
@@ -282,9 +871,82 @@ pub async fn handle_ansible_job(
.await?;
}
if let Some(collections) = r.collections {
for repo in &r.git_repos {
append_logs(
&job.id,
&job.workspace_id,
format!("\nCloning {}...\n", &repo.url),
conn,
)
.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,
git_ssh_cmd,
)
.await
.map_err(|e| anyhow!("Failed to clone git repo `{}`: {e}", repo.url))?;
} else {
if req_lockfiles.is_some() {
append_logs(
&job.id,
&job.workspace_id,
format!("Warning: `{}` is using latest commit because the lockfile didn't store a commit hash for this repo. Updates to the repo could break the deployed playbook.\n", &repo.url),
conn,
)
.await;
}
clone_repo(
repo,
job_dir,
&job.id,
worker_name,
conn,
mem_peak,
canceled_by,
&job.workspace_id,
occupancy_metrics,
git_ssh_cmd,
)
.await
.map_err(|e| anyhow!("Failed to clone git repo `{}`: {e}", repo.url))?;
}
append_logs(
&job.id,
&job.workspace_id,
format!("Cloned {} into {}\n", &repo.url, &repo.target_path),
conn,
)
.await;
}
if let Some(collections) = r.roles_and_collections.as_ref() {
let empty = String::new();
let (lockfile, logs) =
req_lockfiles
.as_ref()
.map(|r| (&r.collections_and_roles, &r.collections_and_roles_logs))
.unwrap_or((collections, &empty));
if !logs.is_empty() {
append_logs(&job.id, &job.workspace_id, logs, conn).await;
}
install_galaxy_collections(
collections.as_str(),
lockfile,
job_dir,
&job.id,
worker_name,
@@ -293,10 +955,12 @@ pub async fn handle_ansible_job(
canceled_by,
conn,
occupancy_metrics,
git_ssh_cmd,
)
.await?;
}
}
append_logs(
&job.id,
&job.workspace_id,
@@ -304,17 +968,17 @@ pub async fn handle_ansible_job(
conn,
)
.await;
let ansible_cfg_content = format!(
r#"
[defaults]
collections_path = ./
roles_path = ./roles
home={job_dir}/.ansible
local_tmp={job_dir}/.ansible/tmp
remote_tmp={job_dir}/.ansible/tmp
"#
);
write_file(job_dir, "ansible.cfg", &ansible_cfg_content)?;
let vault_password_file_exists = match reqs.as_ref().and_then(|x| x.vault_password.as_ref()) {
Some(var_path) => {
let password = client.get_variable_value(&var_path).await?;
write_file(job_dir, WINDMILL_ANSIBLE_PASSWORD_FILENAME, &password)?;
true
}
None => false,
};
create_ansible_cfg(reqs.as_ref(), job_dir, vault_password_file_exists)?;
let mut reserved_variables =
get_reserved_variables(job, &client.token, conn, parent_runnable_path).await?;
backend/windmill-worker/src/worker.rs
+1
View File
@@ -306,6 +306,7 @@ lazy_static::lazy_static! {
pub static ref NSJAIL_PATH: String = std::env::var("NSJAIL_PATH").unwrap_or_else(|_| "nsjail".to_string());
pub static ref PATH_ENV: String = std::env::var("PATH").unwrap_or_else(|_| String::new());
pub static ref HOME_ENV: String = std::env::var("HOME").unwrap_or_else(|_| "/tmp".to_string());
pub static ref GIT_PATH: String = std::env::var("GIT_PATH").unwrap_or_else(|_| "/usr/bin/git".to_string());
pub static ref NODE_PATH: Option<String> = std::env::var("NODE_PATH").ok();
backend/windmill-worker/src/worker_lockfiles.rs
+126 -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,
@@ -1917,6 +1921,123 @@ 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,
token: &str,
base_internal_url: &str,
) -> std::result::Result<String, Error> {
use windmill_parser_yaml::add_versions_to_requirements_yaml;
use crate::{
ansible_executor::{
create_ansible_cfg, get_collection_locks, get_git_ssh_cmd, get_role_locks,
install_galaxy_collections,
},
AuthedClient,
};
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 conn = &Connection::Sql(db.clone());
let authed_client = AuthedClient {
base_internal_url: base_internal_url.to_string(),
token: token.to_string(),
workspace: w_id.to_string(),
force_client: None,
};
let git_ssh_cmd = get_git_ssh_cmd(&reqs, job_dir, &authed_client).await?;
let git_repos = get_git_repos_lock(
&reqs.git_repos,
job_dir,
job_id,
worker_name,
conn,
mem_peak,
canceled_by,
w_id,
occupancy_metrics,
&git_ssh_cmd,
)
.await?;
let ansible_lockfile;
create_ansible_cfg(Some(&reqs), job_dir, false)?;
if let Some(collections) = reqs.roles_and_collections.as_ref() {
install_galaxy_collections(
collections,
job_dir,
job_id,
worker_name,
w_id,
mem_peak,
canceled_by,
conn,
occupancy_metrics,
&git_ssh_cmd,
)
.await?;
let (collection_versions, logs1) = get_collection_locks(job_dir).await?;
let (role_versions, logs2) = if collections.contains("roles:") {
get_role_locks(job_dir).await?
} else {
(HashMap::new(), String::new())
};
let (reqs_yaml, logs3) = add_versions_to_requirements_yaml(&collections, &role_versions, &collection_versions)?;
let logs = format!("\n{logs1}\n{logs2}\n{logs3}\n");
append_logs(job_id, w_id, &logs, conn).await;
ansible_lockfile = AnsibleDependencyLocks {
python_lockfile,
git_repos,
collections_and_roles: reqs_yaml,
collections_and_roles_logs: logs,
};
} else {
ansible_lockfile = AnsibleDependencyLocks {
python_lockfile,
git_repos,
collections_and_roles: String::new(),
collections_and_roles_logs: String::new(),
};
}
serde_json::to_string(&ansible_lockfile).map_err(|e| e.into())
}
async fn capture_dependency_job(
job_id: &Uuid,
job_language: &ScriptLang,
@@ -2001,10 +2122,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,
@@ -2013,9 +2133,9 @@ async fn capture_dependency_job(
worker_name,
w_id,
worker_dir,
&mut Some(occupancy_metrics),
None,
PythonAnnotations::default(),
occupancy_metrics,
token,
base_internal_url,
)
.await
}
flake.nix
+5 -1
View File
@@ -94,6 +94,7 @@
oracle-instantclient
# LSP/Local dev
svelte-language-server
ansible
taplo
]);
packages = [
@@ -158,6 +159,7 @@
];
inherit PKG_CONFIG_PATH RUSTY_V8_ARCHIVE;
GIT_PATH = "${pkgs.git}/bin/git";
NODE_ENV = "development";
NODE_OPTIONS = "--max-old-space-size=16384";
DATABASE_URL = "postgres://postgres:changeme@127.0.0.1:5432/";
@@ -172,12 +174,14 @@
JAVA_PATH = "${pkgs.jdk21}/bin/java";
JAVAC_PATH = "${pkgs.jdk21}/bin/javac";
COURSIER_PATH = "${coursier}/coursier";
# for related places search: ADD_NEW_LANG
# for related places search: ADD_NEW_LANG
FLOCK_PATH = "${pkgs.flock}/bin/flock";
CARGO_PATH = "${rust}/bin/cargo";
DOTNET_PATH = "${pkgs.dotnet-sdk_9}/bin/dotnet";
DOTNET_ROOT = "${pkgs.dotnet-sdk_9}/share/dotnet";
ORACLE_LIB_DIR = "${pkgs.oracle-instantclient.lib}/lib";
ANSIBLE_PLAYBOOK_PATH = "${pkgs.ansible}/bin/ansible-playbook";
ANSIBLE_GALAXY_PATH = "${pkgs.ansible}/bin/ansible-galaxy";
RUST_LOG = "debug";
};
packages.default = self.packages.${system}.windmill;
frontend/src/lib/script_helpers.ts
+10
View File
@@ -931,6 +931,12 @@ inventory:
- resource_type: ansible_inventory
# You can pin an inventory to this script by hardcoding the resource path:
# resource: u/user/your_resource
# - name: hcloud.yml
# resource_type: dynamic_inventory
options:
- verbosity: vvv
# File resources will be written in the relative \`target\` location before
# running the playbook
@@ -946,11 +952,15 @@ extra_vars:
world_qualifier:
type: string
# If using Ansible Vault:
# vault_password: u/user/ansible_vault_password
dependencies:
galaxy:
collections:
- name: community.general
- name: community.vmware
roles:
python:
- jmespath
---