use anyhow::{Context, Result, bail};
use base64::Engine;
use crate::config::{Config, Host};
use crate::probe::State;
use crate::transport::Transport;
use crate::wrapper::{JOBS_ROOT, JobId, state_dir};
const DEFAULT_LS_WINDOW_SECS: u64 = 24 * 60 * 60;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Row {
pub id: String,
pub host: String,
pub state: State,
pub age_secs: u64,
pub cmd: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Unreachable {
pub host: String,
pub why: String,
pub remedy: Option<String>,
}
pub fn list(
cfg: &Config,
transport: &dyn Transport,
host_filter: Option<&str>,
all: bool,
) -> Result<(Vec<Row>, Vec<Unreachable>)> {
let (rows, unreachable, _) = list_with_hidden(cfg, transport, host_filter, all)?;
Ok((rows, unreachable))
}
pub fn list_with_hidden(
cfg: &Config,
transport: &dyn Transport,
host_filter: Option<&str>,
all: bool,
) -> Result<(Vec<Row>, Vec<Unreachable>, usize)> {
let hosts: Vec<&Host> = match host_filter {
Some(name) => vec![cfg.host(Some(name))?],
None => cfg.hosts().iter().collect(),
};
let mut rows = Vec::new();
let mut unreachable = Vec::new();
let mut hidden = 0;
for host in hosts {
if !transport.master_alive(host) {
unreachable.push(Unreachable {
host: host.name.clone(),
why: "no control master".into(),
remedy: Some(crate::errors::master_command(host)),
});
continue;
}
let output = transport.run(host, &list_script(host))?;
if output.code != 0 {
bail!(
"listing jobs on {} failed: {}",
host.name,
output.stderr.trim()
);
}
hidden += parse_rows(host, &output.text(), all, &mut rows)?;
}
Ok((rows, unreachable, hidden))
}
fn list_script(host: &Host) -> String {
format!(
"root={JOBS_ROOT}; [ -d \"$root\" ] || exit 0; \
live=$(tmux -L {} list-sessions -F '#{{session_name}}' 2>/dev/null | sed 's/^coop-//'); \
{{ find \"$root\" -mindepth 1 -maxdepth 1 -type d -exec stat -c '%Y %n' {{}} + 2>/dev/null \
|| find \"$root\" -mindepth 1 -maxdepth 1 -type d -exec stat -f '%m %N' {{}} + ; }} \
| awk -v now=\"$(date +%s)\" -v live=\"$live\" '\
BEGIN {{ \
n = split(live, L, \"\\n\"); \
for (i = 1; i <= n; i++) if (L[i] != \"\") alive[L[i]] = 1; \
alphabet = \"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/\"; \
for (i = 0; i < 256; i++) ord[sprintf(\"%c\", i)] = i; \
}} \
function b64(s, out, i, n, a, b, c) {{ \
for (i = 1; i <= length(s); i += 3) {{ \
n = length(s) - i + 1; \
a = ord[substr(s, i, 1)]; \
b = n > 1 ? ord[substr(s, i + 1, 1)] : 0; \
c = n > 2 ? ord[substr(s, i + 2, 1)] : 0; \
out = out substr(alphabet, int(a / 4) + 1, 1); \
out = out substr(alphabet, (a % 4) * 16 + int(b / 16) + 1, 1); \
out = out (n > 1 ? substr(alphabet, (b % 16) * 4 + int(c / 64) + 1, 1) : \"=\"); \
out = out (n > 2 ? substr(alphabet, c % 64 + 1, 1) : \"=\"); \
}} \
return out; \
}} \
{{ \
mtime = $1; dir = substr($0, length($1) + 2); \
id = dir; sub(/.*\\//, \"\", id); \
rc = \"\"; if ((getline l < (dir \"/rc\")) > 0) rc = l; \
close(dir \"/rc\"); \
cmd = \"\"; \
while ((getline l < (dir \"/cmd\")) > 0) cmd = (cmd == \"\") ? l : cmd \"\\n\" l; \
close(dir \"/cmd\"); \
printf \"%s\\t%s\\t%s\\t%s\\t%s\\n\", \
id, now - mtime, rc, (id in alive) ? 1 : 0, b64(cmd); \
}}'",
host.tmux_socket
)
}
fn parse_rows(host: &Host, reply: &str, all: bool, rows: &mut Vec<Row>) -> Result<usize> {
let mut hidden = 0;
for line in reply.lines() {
let mut fields = line.splitn(5, '\t');
let id = fields.next().context("invalid ls reply: missing id")?;
let age_secs = fields
.next()
.context("invalid ls reply: missing age")?
.parse()
.context("invalid age in ls reply")?;
let rc_text = fields.next().context("invalid ls reply: missing rc")?;
let alive = fields.next().context("invalid ls reply: missing alive")? == "1";
let cmd = String::from_utf8_lossy(&decode_command(
fields.next().context("invalid ls reply: missing cmd")?,
)?)
.into_owned();
let rc = if rc_text.is_empty() {
None
} else {
Some(rc_text.parse().context("invalid rc in ls reply")?)
};
let state = match rc {
Some(code) => State::Done(code),
None if alive => State::Running,
None => State::Orphan,
};
let recent = age_secs < DEFAULT_LS_WINDOW_SECS;
if all || recent || !matches!(state, State::Done(_)) {
rows.push(Row {
id: id.into(),
host: host.name.clone(),
state,
age_secs,
cmd,
});
} else {
hidden += 1;
}
}
Ok(hidden)
}
pub fn kill(transport: &dyn Transport, host: &Host, id: &JobId) -> Result<i32> {
crate::errors::require_master(transport, host)?;
let dir = state_dir(id);
let script = format!(
"d={dir}; [ -f $d/rc ] || echo 137 > $d/rc; \
tmux -L {socket} kill-session -t coop-{id} 2>/dev/null; \
tmux -L {socket} kill-session -t watch-{id} 2>/dev/null; \
cat $d/rc",
socket = host.tmux_socket
);
let output = transport.run(host, &script)?;
if output.code != 0 {
bail!("kill failed: {}", output.stderr.trim());
}
output
.text()
.trim()
.parse()
.context("kill returned an invalid exit code")
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Target {
One(JobId),
AllDone,
}
pub fn remove(transport: &dyn Transport, host: &Host, target: &Target) -> Result<Vec<String>> {
crate::errors::require_master(transport, host)?;
let script = match target {
Target::One(id) => {
let dir = state_dir(id);
format!(
"tmux -L {} kill-session -t coop-{id} 2>/dev/null; \
if [ -d {dir} ]; then rm -rf {dir} && echo {id}; fi; exit 0",
host.tmux_socket
)
}
Target::AllDone => format!(
"root={JOBS_ROOT}; [ -d \"$root\" ] || exit 0; \
for d in \"$root\"/*; do \
[ -d \"$d\" ] && [ -f \"$d/rc\" ] || continue; \
rm -rf \"$d\" && echo \"${{d##*/}}\"; \
done; exit 0"
),
};
let output = transport.run(host, &script)?;
if output.code != 0 {
bail!("rm failed on {}: {}", host.name, output.stderr.trim());
}
Ok(output
.text()
.lines()
.map(str::trim)
.filter(|line| !line.is_empty())
.map(str::to_string)
.collect())
}
const ORPHAN_KEEP_MULTIPLIER: u32 = 4;
pub fn prune(host: &Host) -> String {
let orphan_days = host.keep_days.saturating_mul(ORPHAN_KEEP_MULTIPLIER);
format!(
"root={JOBS_ROOT}; [ ! -d \"$root\" ] || {{ \
live=$(tmux -L {} list-sessions -F '#{{session_name}}' 2>/dev/null); \
find \"$root\" -mindepth 1 -maxdepth 1 -type d -mtime +{} \
-exec test -f '{{}}/rc' \\; -exec rm -rf '{{}}' + ; \
find \"$root\" -mindepth 1 -maxdepth 1 -type d -mtime +{orphan_days} \
-exec test ! -f '{{}}/rc' \\; -print | while IFS= read -r d; do \
id=\"${{d##*/}}\"; \
echo \"$live\" | grep -qx \"coop-$id\" && continue; \
rm -rf \"$d\"; \
done; }}",
host.tmux_socket, host.keep_days
)
}
pub fn decode_command(input: &str) -> Result<Vec<u8>> {
base64::engine::general_purpose::STANDARD
.decode(input)
.context("invalid command encoding in ls reply")
}