//! One run: a Job, its pod, the steps, and the result. //! //! This is the shape `knotless-run` would call from its executor seam: //! start a run (the Job), run each step in the same environment, stream //! the log, copy outputs back, and stop (delete the Job). use std::fs::{self, File}; use std::io::Write; use std::path::PathBuf; use std::process::Stdio; use std::thread; use std::time::{Duration, Instant}; use crate::cli::{self, Options}; use crate::json::{option, quote}; use crate::kubectl::Kubectl; use crate::manifest; use crate::steps::{self, Step}; /// What one step came to, as `dev.knotless.check` records it. #[derive(Clone, Debug, PartialEq, Eq)] pub struct StepResult { pub name: String, pub status: String, pub exit_code: Option, pub duration_ms: u128, pub timed_out: bool, } /// The result of one proof-of-concept run. #[derive(Clone, Debug)] pub struct Outcome { pub name: String, pub job: String, pub namespace: String, pub pod: Option, pub status: String, pub steps: Vec, pub log: PathBuf, pub outputs: Option, pub job_deleted: bool, pub dry_run: bool, pub error: Option, } impl Outcome { /// Whether the binary should exit zero. pub fn success(&self) -> bool { self.dry_run || self.status == "success" } /// The result as JSON, the shape a check's steps take. pub fn to_json(&self) -> String { let steps: Vec = self .steps .iter() .map(|step| { format!( "{{\"name\":{},\"status\":{},\"exitCode\":{},\"durationMs\":{},\"timedOut\":{}}}", quote(&step.name), quote(&step.status), step.exit_code .map(|code| code.to_string()) .unwrap_or_else(|| "null".to_string()), step.duration_ms, step.timed_out ) }) .collect(); format!( concat!( "{{", "\"name\":{},", "\"job\":{},", "\"namespace\":{},", "\"pod\":{},", "\"status\":{},", "\"steps\":[{}],", "\"log\":{},", "\"outputs\":{},", "\"jobDeleted\":{},", "\"dryRun\":{},", "\"error\":{}", "}}" ), quote(&self.name), quote(&self.job), quote(&self.namespace), option(self.pod.as_deref()), quote(&self.status), steps.join(","), quote(&self.log.display().to_string()), option(self.outputs.as_ref().map(|p| p.display().to_string()).as_deref()), self.job_deleted, self.dry_run, option(self.error.as_deref()) ) } } /// Runs one proof-of-concept workflow. pub fn run(options: &Options) -> Result { let steps = steps::load(options)?; let job = manifest::job_name(&options.name)?; let log_path = cli::log_path(options); let outputs = options.outputs.clone(); let mut outcome = Outcome { name: options.name.clone(), job: job.clone(), namespace: options.namespace.clone(), pod: None, status: "error".to_string(), steps: Vec::new(), log: log_path.clone(), outputs: outputs.clone(), job_deleted: false, dry_run: options.dry_run, error: None, }; if options.dry_run { dry_run(options, &job, &steps, &mut outcome)?; write_result(options, &outcome)?; return Ok(outcome); } let kubectl = Kubectl::new(options); let manifest = manifest::job_json(options)?; let mut log = File::create(&log_path) .map_err(|err| format!("creating {}: {err}", log_path.display()))?; writeln!( log, "==> run {} job {} image {}", options.name, job, options.image ) .map_err(|err| format!("writing {}: {err}", log_path.display()))?; let (created, error) = match execute(options, &kubectl, &manifest, &job, &steps, &mut log, &mut outcome) { Ok(created) => (created, None), Err(err) => (outcome.pod.is_some(), Some(err)), }; outcome.error = error; if created && !options.keep { match kubectl.delete(&job) { Ok(()) => outcome.job_deleted = true, Err(err) => { let _ = writeln!(log, "==> cleanup: {err}"); } } } else if options.keep { let _ = writeln!(log, "==> kept Job {job}"); } outcome.status = if outcome.error.is_some() { "error".to_string() } else if outcome.steps.iter().all(|step| step.status == "success") { "success".to_string() } else { "failure".to_string() }; let _ = writeln!(log, "==> status {}", outcome.status); write_result(options, &outcome)?; Ok(outcome) } fn execute( options: &Options, kubectl: &Kubectl, manifest: &str, job: &str, steps: &[Step], log: &mut File, outcome: &mut Outcome, ) -> Result { kubectl.apply(manifest)?; // The Job exists from here on, so cleanup must still find it even if // the pod never becomes Ready. outcome.pod = Some(String::new()); if let Err(err) = kubectl.wait_pod(job, options.timeout) { return Err(err); } let pod = kubectl.pod(job)?; outcome.pod = Some(pod.clone()); kubectl.mkdirs(&pod)?; kubectl.tar_in(&pod, "/work", &options.workdir)?; if let Some(steps_dir) = &options.steps_dir { kubectl.tar_in(&pod, "/steps", steps_dir)?; } for (index, step) in steps.iter().enumerate() { let _ = writeln!(log, "==> step {}: {}", step.name, step.script); let mut command = kubectl.exec(&pod, &step.script); command .stdin(Stdio::null()) .stdout(Stdio::from(log.try_clone().map_err(stringify)?)) .stderr(Stdio::from(log.try_clone().map_err(stringify)?)); let started = Instant::now(); let mut child = match command.spawn() { Ok(child) => child, Err(err) => { outcome.steps.push(StepResult { name: step.name.clone(), status: "error".to_string(), exit_code: None, duration_ms: started.elapsed().as_millis(), timed_out: false, }); skip_rest(steps, index + 1, outcome); return Err(format!("starting step {}: {err}", step.name)); } }; let mut exit = None; let mut timed_out = false; loop { match child.try_wait() { Ok(Some(status)) => { exit = status.code(); break; } Ok(None) => {} Err(err) => { outcome.steps.push(StepResult { name: step.name.clone(), status: "error".to_string(), exit_code: None, duration_ms: started.elapsed().as_millis(), timed_out: false, }); skip_rest(steps, index + 1, outcome); return Err(format!("waiting for step {}: {err}", step.name)); } } if started.elapsed() >= options.step_timeout { let _ = child.kill(); let _ = child.wait(); timed_out = true; break; } thread::sleep(Duration::from_millis(50)); } let duration_ms = started.elapsed().as_millis(); let status = if timed_out { "failure" } else if exit == Some(0) { "success" } else { "failure" }; let _ = writeln!( log, "==> step {} {} exit={:?} in {}ms{}", step.name, status, exit, duration_ms, if timed_out { " (timed out)" } else { "" } ); outcome.steps.push(StepResult { name: step.name.clone(), status: status.to_string(), exit_code: exit, duration_ms, timed_out, }); if timed_out { skip_rest(steps, index + 1, outcome); return Ok(true); } if exit != Some(0) { skip_rest(steps, index + 1, outcome); return Ok(true); } } if let Some(outputs) = &options.outputs { fs::create_dir_all(outputs) .map_err(|err| format!("creating {}: {err}", outputs.display()))?; kubectl.tar_out(&pod, "/outputs", outputs)?; } Ok(true) } fn skip_rest(steps: &[Step], from: usize, outcome: &mut Outcome) { for step in &steps[from..] { outcome.steps.push(StepResult { name: step.name.clone(), status: "skipped".to_string(), exit_code: None, duration_ms: 0, timed_out: false, }); } } fn dry_run( options: &Options, job: &str, steps: &[Step], outcome: &mut Outcome, ) -> Result<(), String> { let kubectl = Kubectl::new(options); println!("{}", manifest::job_json(options)?); println!("# {}", kubectl.planned(&["apply", "-f", "-"])); println!( "# {}", kubectl.planned(&[ "wait", "--for=condition=Ready", "pod", "-l", &format!("job-name={job}"), "--timeout", &format!("{}s", options.timeout.as_secs()), ]) ); println!( "# {}", kubectl.planned(&[ "get", "pods", "-l", &format!("job-name={job}"), "-o", "jsonpath={.items[0].metadata.name}", ]) ); println!( "# {}", kubectl.planned(&["exec", "", "--", "mkdir", "-p", "/work", "/outputs", "/steps"]) ); println!("# tar -cf - -C {} . | {}", options.workdir.display(), kubectl.planned(&["exec", "-i", "", "--", "tar", "-xf", "-", "-C", "/work"])); if let Some(steps_dir) = &options.steps_dir { println!("# tar -cf - -C {} . | {}", steps_dir.display(), kubectl.planned(&["exec", "-i", "", "--", "tar", "-xf", "-", "-C", "/steps"])); } for step in steps { println!("# {}", kubectl.planned(&["exec", "", "--", "sh", "-ec", &step.script])); outcome.steps.push(StepResult { name: step.name.clone(), status: "skipped".to_string(), exit_code: None, duration_ms: 0, timed_out: false, }); } if let Some(outputs) = &options.outputs { println!("# {} | tar -xf - -C {}", kubectl.planned(&["exec", "", "--", "tar", "-cf", "-", "-C", "/outputs", "."]), outputs.display()); } if !options.keep { println!("# {}", kubectl.planned(&["delete", "job", job, "--ignore-not-found", "--wait=false"])); } outcome.status = "dry-run".to_string(); Ok(()) } fn write_result(options: &Options, outcome: &Outcome) -> Result<(), String> { let json = outcome.to_json(); match &options.result { Some(path) => fs::write(path, format!("{json}\n")) .map_err(|err| format!("writing {}: {err}", path.display())), None => { println!("{json}"); Ok(()) } } } fn stringify(err: std::io::Error) -> String { err.to_string() } #[cfg(test)] mod tests { use super::*; #[test] fn result_json() { let outcome = Outcome { name: "ci 1".into(), job: "ci-1".into(), namespace: "default".into(), pod: Some("pod-1".into()), status: "failure".into(), steps: vec![StepResult { name: "test".into(), status: "failure".into(), exit_code: Some(1), duration_ms: 12, timed_out: false, }], log: PathBuf::from("ci.log"), outputs: None, job_deleted: true, dry_run: false, error: None, }; let json = outcome.to_json(); assert!(json.contains("\"name\":\"ci 1\"")); assert!(json.contains("\"exitCode\":1")); assert!(json.contains("\"outputs\":null")); assert!(json.contains("\"jobDeleted\":true")); } }