4 Commits

Author SHA1 Message Date
de31eeef7a Format Rust code using rustfmt 2025-06-27 11:15:08 +00:00
aea2a927c2 Bump 2025-05-05 17:31:12 +03:00
e1c2363113 Added host ordering 2025-05-05 17:29:39 +03:00
d8cdfbbf59 Update CI 2025-04-08 18:46:13 +01:00
4 changed files with 166 additions and 104 deletions

View File

@ -104,7 +104,7 @@ jobs:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
with: with:
allowUpdates: true allowUpdates: true
body: "Релиз rexec версии ${{ env.VERSION }}. Статические сборки для Linux (x86_64), Windows (x86_64) и MacOS (arm64)." body: "Release rexec ${{ env.VERSION }}. Static build for Linux (x86_64), Windows (x86_64) и MacOS (arm64)."
upload: upload:
name: Upload Release Assets name: Upload Release Assets

2
Cargo.lock generated
View File

@ -1111,7 +1111,7 @@ checksum = "436b050e76ed2903236f032a59761c1eb99e1b0aead2c257922771dab1fc8c78"
[[package]] [[package]]
name = "rexec" name = "rexec"
version = "1.1.0" version = "1.2.1"
dependencies = [ dependencies = [
"brace-expand", "brace-expand",
"clap 4.3.4", "clap 4.3.4",

View File

@ -1,6 +1,6 @@
[package] [package]
name = "rexec" name = "rexec"
version = "1.2.0" version = "1.2.1"
readme = "https://github.com/house-of-vanity/rexec#readme" readme = "https://github.com/house-of-vanity/rexec#readme"
edition = "2021" edition = "2021"
description = "Parallel SSH executor" description = "Parallel SSH executor"

View File

@ -13,7 +13,7 @@ use colored::*;
use dns_lookup::lookup_host; use dns_lookup::lookup_host;
use env_logger::Env; use env_logger::Env;
use itertools::Itertools; use itertools::Itertools;
use log::{error, info}; use log::{error, info, warn};
use massh::{MasshClient, MasshConfig, MasshHostConfig, SshAuth}; use massh::{MasshClient, MasshConfig, MasshHostConfig, SshAuth};
use question::{Answer, Question}; use question::{Answer, Question};
use rayon::prelude::*; use rayon::prelude::*;
@ -185,7 +185,7 @@ fn main() {
all_hosts all_hosts
}; };
// Dedup hosts from known_hosts file // Dedup hosts from known_hosts file but preserve original order
let matched_hosts: Vec<_> = hosts.into_iter().unique().collect(); let matched_hosts: Vec<_> = hosts.into_iter().unique().collect();
// Build MasshHostConfig hostnames list // Build MasshHostConfig hostnames list
@ -199,36 +199,66 @@ fn main() {
}); });
} }
info!("Matched hosts:"); // Store hosts with their indices to preserve order
let resolved_ips = Arc::new(Mutex::new(Vec::<(String, IpAddr)>::new())); let mut host_with_indices: Vec<(Host, usize)> = Vec::new();
for (idx, host) in matched_hosts.iter().enumerate() {
host_with_indices.push((host.clone(), idx));
}
matched_hosts.par_iter().for_each(|host| { info!("Matched hosts:");
match lookup_host(&host.name) {
// Do DNS resolution in parallel but store results for ordered display later
let resolved_ips_with_indices = Arc::new(Mutex::new(Vec::<(String, IpAddr, usize)>::new()));
host_with_indices
.par_iter()
.for_each(|(host, idx)| match lookup_host(&host.name) {
Ok(ips) if !ips.is_empty() => { Ok(ips) if !ips.is_empty() => {
let ip = ips[0]; let ip = ips[0];
let mut results = resolved_ips_with_indices.lock().unwrap();
info!("{} [{}]", &host.name, ip); results.push((host.name.clone(), ip, *idx));
let mut results = resolved_ips.lock().unwrap();
results.push((host.name.clone(), ip));
} }
Ok(_) => { Ok(_) => {
error!("DNS resolved, but IP not found: {}", &host.name.red()); let mut results = resolved_ips_with_indices.lock().unwrap();
results.push((
host.name.clone(),
IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 0)),
*idx,
));
} }
Err(_) => { Err(_) => {
error!("DNS resolve failed: {}", &host.name.red()); let mut results = resolved_ips_with_indices.lock().unwrap();
} results.push((
host.name.clone(),
IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 0)),
*idx,
));
} }
}); });
let mut hosts_and_ips: HashMap<IpAddr, String> = HashMap::new(); // Sort by original index to ensure hosts are displayed in order
let mut resolved_hosts = resolved_ips_with_indices.lock().unwrap().clone();
resolved_hosts.sort_by_key(|(_, _, idx)| *idx);
// Now print the hosts in the correct order
for (hostname, ip, _) in &resolved_hosts {
if ip.is_unspecified() {
error!("DNS resolve failed: {}", hostname.red());
} else {
info!("{} [{}]", hostname, ip);
}
}
// Create massh_hosts in the correct order
let mut hosts_and_ips: HashMap<IpAddr, (String, usize)> = HashMap::new();
let mut massh_hosts: Vec<MasshHostConfig> = Vec::new(); let mut massh_hosts: Vec<MasshHostConfig> = Vec::new();
if let Ok(results) = resolved_ips.lock() { for (hostname, ip, idx) in resolved_hosts {
for (hostname, ip) in results.iter() { // Skip hosts that couldn't be resolved
hosts_and_ips.insert(*ip, hostname.clone()); if !ip.is_unspecified() {
hosts_and_ips.insert(ip, (hostname.clone(), idx));
massh_hosts.push(MasshHostConfig { massh_hosts.push(MasshHostConfig {
addr: *ip, addr: ip,
auth: None, auth: None,
port: None, port: None,
user: None, user: None,
@ -236,23 +266,15 @@ fn main() {
} }
} }
// Build MasshConfig using massh_hosts vector // Process hosts in batches to maintain order
let config = MasshConfig { let batch_size = args.parallel as usize;
default_auth: SshAuth::Agent,
default_port: 22,
default_user: args.username,
threads: args.parallel as u64,
timeout: 0,
hosts: massh_hosts,
};
let massh = MasshClient::from(&config);
// Ask for confirmation // Ask for confirmation
if config.hosts.len() != 0 if !massh_hosts.is_empty()
&& (args.noconfirm == true && (args.noconfirm
|| match Question::new(&*format!( || match Question::new(&*format!(
"Continue on following {} servers?", "Continue on following {} servers?",
&config.hosts.len() &massh_hosts.len()
)) ))
.confirm() .confirm()
{ {
@ -261,11 +283,41 @@ fn main() {
_ => unreachable!(), _ => unreachable!(),
}) })
{ {
info!("Run command on {} servers.", &config.hosts.len()); info!("Run command on {} servers.", &massh_hosts.len());
// Run a command on all the configured hosts. let mut processed = 0;
// Receive the result of the command for each host and print its output.
let rx = massh.execute(args.command); while processed < massh_hosts.len() {
let end = std::cmp::min(processed + batch_size, massh_hosts.len());
// Create a new config and vector for this batch
let mut batch_hosts = Vec::new();
for host in &massh_hosts[processed..end] {
batch_hosts.push(MasshHostConfig {
addr: host.addr,
auth: None,
port: None,
user: None,
});
}
// Create a new MasshClient for this batch
let batch_config = MasshConfig {
default_auth: SshAuth::Agent,
default_port: 22,
default_user: args.username.clone(),
threads: batch_hosts.len() as u64,
timeout: 0,
hosts: batch_hosts,
};
let batch_massh = MasshClient::from(&batch_config);
// Run commands on this batch
let rx = batch_massh.execute(args.command.clone());
// Collect all results from this batch before moving to the next
let mut batch_results = Vec::new();
while let Ok((host, result)) = rx.recv() { while let Ok((host, result)) = rx.recv() {
let ip: String = host.split('@').collect::<Vec<_>>()[1] let ip: String = host.split('@').collect::<Vec<_>>()[1]
@ -273,16 +325,21 @@ fn main() {
.collect::<Vec<_>>()[0] .collect::<Vec<_>>()[0]
.to_string(); .to_string();
let ip = ip.parse::<IpAddr>().unwrap(); let ip = ip.parse::<IpAddr>().unwrap();
println!(
"\n{}", if let Some((hostname, idx)) = hosts_and_ips.get(&ip) {
hosts_and_ips batch_results.push((hostname.clone(), ip, result, *idx));
.get(&ip) } else {
.unwrap_or(&"Couldn't parse IP".to_string()) error!("Unexpected IP address in result: {}", ip);
.to_string() }
.yellow() }
.bold()
.to_string() // Sort the batch results by index to ensure they're displayed in order
); batch_results.sort_by_key(|(_, _, _, idx)| *idx);
// Display the results
for (hostname, _ip, result, _) in batch_results {
println!("\n{}", hostname.yellow().bold().to_string());
let output = match result { let output = match result {
Ok(output) => output, Ok(output) => output,
Err(e) => { Err(e) => {
@ -290,15 +347,17 @@ fn main() {
continue; continue;
} }
}; };
let code_string = if output.exit_status == 0 { let code_string = if output.exit_status == 0 {
format!("{}", output.exit_status.to_string().green()) format!("{}", output.exit_status.to_string().green())
} else { } else {
format!("{}", output.exit_status.to_string().red()) format!("{}", output.exit_status.to_string().red())
}; };
println!( println!(
"{}", "{}",
format!( format!(
"Exit code [{}] | std out/err [{}/{}] bytes", "Exit code [{}] / stdout {} bytes / stderr {} bytes",
code_string, code_string,
output.stdout.len(), output.stdout.len(),
output.stderr.len() output.stderr.len()
@ -336,6 +395,9 @@ fn main() {
} }
} }
} }
processed = end;
}
} else { } else {
warn!("Stopped"); warn!("Stopped");
} }