johnandersen777.bsky.social/knotless-runner-k8s

Switch repository

Loading…

//! 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<i32>,
    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<String>,
    pub status: String,
    pub steps: Vec<StepResult>,
    pub log: PathBuf,
    pub outputs: Option<PathBuf>,
    pub job_deleted: bool,
    pub dry_run: bool,
    pub error: Option<String>,
}

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<String> = 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<Outcome, String> {
    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<bool, String> {
    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", "<pod>", "--", "mkdir", "-p", "/work", "/outputs", "/steps"])
    );
    println!("# tar -cf - -C {} . | {}", options.workdir.display(), kubectl.planned(&["exec", "-i", "<pod>", "--", "tar", "-xf", "-", "-C", "/work"]));
    if let Some(steps_dir) = &options.steps_dir {
        println!("# tar -cf - -C {} . | {}", steps_dir.display(), kubectl.planned(&["exec", "-i", "<pod>", "--", "tar", "-xf", "-", "-C", "/steps"]));
    }
    for step in steps {
        println!("# {}", kubectl.planned(&["exec", "<pod>", "--", "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", "<pod>", "--", "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"));
    }
}