diff --git a/.changeset/kubernetes-executor.md b/.changeset/kubernetes-executor.md new file mode 100644 index 00000000..cf0159cf --- /dev/null +++ b/.changeset/kubernetes-executor.md @@ -0,0 +1,7 @@ +--- +default: minor +--- + +# A Kubernetes executor + +`knotless-run` now runs each workflow run as one Job on a tenant cluster by default; `--executor docker` is the explicit fallback. The checkout and shared steps are streamed in, each step runs through `kubectl exec` with its variables and secrets on stdin, outputs are streamed back, and the Job is deleted. The runner stays outside the cluster; the Job pod is what raises the tenant's scale demand. Caches stream in and out of the pod. Pods get a default-deny ingress NetworkPolicy, and a scoped ServiceAccount kubeconfig is supported. diff --git a/crates/knotless-run/src/executor.rs b/crates/knotless-run/src/executor.rs index 5d3311d6..125195ea 100644 --- a/crates/knotless-run/src/executor.rs +++ b/crates/knotless-run/src/executor.rs @@ -5,11 +5,13 @@ use anyhow::{Context, Result, bail}; use rustix::process::{Pid, Signal}; +use serde_json::{Value, json}; use std::collections::{BTreeSet, HashSet}; use std::path::{Path, PathBuf}; use std::process::Stdio; use std::sync::Mutex; use std::time::{Duration, Instant}; +use tokio::io::AsyncWriteExt; use tokio::process::Command; /// The runner's own variables a step sees, so it finds its tools and the @@ -32,6 +34,20 @@ pub enum Executor { /// `docker run --rm`, in the workflow's image or this one, within /// `limits`. Docker { image: String, limits: Limits }, + /// One Kubernetes Job per run, in a tenant cluster, with the checkout + /// streamed in and each step run through `kubectl exec`. The runner + /// stays outside the cluster: the Job's pod is what raises the + /// tenant's scale demand. + Kubernetes { + image: String, + limits: Limits, + kubeconfig: Option, + context: Option, + namespace: String, + kubectl: PathBuf, + service_account: Option, + node_selector: Vec<(String, String)>, + }, } /// What one run's container may take of the host, under docker: each @@ -87,6 +103,19 @@ pub enum Session { id: String, volumes: Vec, }, + /// A Job's pod, through this kubectl client: steps are `kubectl exec` + /// into it, and its outputs are streamed back when it stops. + Kubernetes { + kubectl: PathBuf, + kubeconfig: Option, + context: Option, + namespace: String, + job: String, + pod: String, + outputs: PathBuf, + vars: Vec<(String, String)>, + caches: Vec, + }, } /// What a step came to. @@ -130,6 +159,10 @@ pub struct Setup<'a> { /// Whether its code is vouched for by an owner the runner trusts, to /// run under [`Limits::trusted_runtime`]. pub vouched: bool, + /// How long the whole run may take, where that is limited. The docker + /// executor enforces it per step; a Kubernetes Job uses it as its + /// `activeDeadlineSeconds`. + pub run_timeout: Option, } /// The `--cpu-shares` a run behind the rest is given: docker makes it a @@ -181,33 +214,54 @@ impl Executor { image: image.to_string(), limits: Limits::default(), }), + "kubernetes" => Some(Executor::Kubernetes { + image: image.to_string(), + limits: Limits::default(), + kubeconfig: None, + context: None, + namespace: "default".into(), + kubectl: PathBuf::from("kubectl"), + service_account: None, + node_selector: Vec::new(), + }), _ => None, } } + /// Whether this executor keeps caches between runs. + pub fn uses_caches(&self) -> bool { + matches!(self, Executor::Docker { .. } | Executor::Kubernetes { .. }) + } + /// Where the steps find the `i`th shared step's repository, checked - /// out at `at` on the host: under docker, where it is mounted. + /// out at `at` on the host: under docker and kubernetes, where it is + /// streamed. pub fn step_dir(&self, i: usize, at: &Path) -> String { match self { Executor::Shell => at.to_string_lossy().into_owned(), - Executor::Docker { .. } => format!("{STEPS}/{i}"), + Executor::Docker { .. } | Executor::Kubernetes { .. } => format!("{STEPS}/{i}"), } } /// Starts what a run's steps run in, as `setup` says: under docker, a /// container whose start, an image pulled among it, is written to - /// `log` and may take `timeout`. + /// `log` and may take `timeout`; under kubernetes, a Job whose pod is + /// Ready. pub async fn start( &self, setup: &Setup<'_>, log: &std::fs::File, timeout: Duration, ) -> Result { - self.start_with(Path::new("docker"), setup, log, timeout) - .await + let client = match self { + Executor::Kubernetes { kubectl, .. } => kubectl.as_path(), + _ => Path::new("docker"), + }; + self.start_with(client, setup, log, timeout).await } - /// [`Executor::start`], with `docker` as the docker client. + /// [`Executor::start`], with `docker` as the docker client, or the + /// kubectl client a Kubernetes executor already names. pub async fn start_with( &self, docker: &Path, @@ -215,65 +269,657 @@ impl Executor { log: &std::fs::File, timeout: Duration, ) -> Result { - let Setup { - dir, - vars, - image, - caches, - outputs, - steps: _, - owner: _, - behind: _, - vouched: _, - } = *setup; - let mut vars = vars.to_vec(); - let Executor::Docker { - image: default, - limits, - } = self - else { - vars.push((OUTPUTS.into(), outputs.display().to_string())); - return Ok(Session::Shell { - dir: dir.to_path_buf(), - vars, - }); - }; - vars.push((OUTPUTS.into(), "/outputs".into())); - let image = image.unwrap_or(default); - let mut volumes = Vec::new(); - let mut made = Ok(()); - for cache in caches { - made = overlay(docker, cache, log).await; - if made.is_err() { - break; + match self { + Executor::Shell => { + let mut vars = setup.vars.to_vec(); + vars.push((OUTPUTS.into(), setup.outputs.display().to_string())); + Ok(Session::Shell { + dir: setup.dir.to_path_buf(), + vars, + }) } - volumes.push(cache.volume.clone()); - } - let started = match made { - Ok(()) => { - let start = Start { + Executor::Docker { image, limits } => { + start_docker(docker, setup, image, limits, log, timeout).await + } + Executor::Kubernetes { + image, + limits, + kubeconfig, + context, + namespace, + kubectl, + service_account, + node_selector, + } => { + let image = setup.image.unwrap_or(image); + let job = job_name(setup.owner.run); + let manifest = job_manifest( + &job, + setup, image, limits, - vars: &vars, + namespace, + service_account, + node_selector, timeout, - }; - container(docker, setup, start, log).await + ) + .to_string(); + let pod = start_kubernetes( + kubectl, kubeconfig, context, namespace, &job, &manifest, setup, log, timeout, + ) + .await?; + crate::telemetry::container("job started", &job, Some(image)); + let mut vars = setup.vars.to_vec(); + vars.push((OUTPUTS.into(), "/outputs".into())); + Ok(Session::Kubernetes { + kubectl: kubectl.clone(), + kubeconfig: kubeconfig.clone(), + context: context.clone(), + namespace: namespace.clone(), + job, + pod, + outputs: setup.outputs.to_path_buf(), + vars, + caches: setup.caches.to_vec(), + }) } - Err(err) => Err(err), - }; - // A container that did not start leaves no overlay mounted, for - // its layers to be removed. - if started.is_err() { - remove_volumes(docker, &volumes).await; } - Ok(Session::Docker { - docker: docker.to_path_buf(), - id: started?, - volumes, - }) } } +/// Starts the docker container a run's steps run in. +async fn start_docker( + docker: &Path, + setup: &Setup<'_>, + default: &str, + limits: &Limits, + log: &std::fs::File, + timeout: Duration, +) -> Result { + let mut vars = setup.vars.to_vec(); + vars.push((OUTPUTS.into(), "/outputs".into())); + let image = setup.image.unwrap_or(default); + let mut volumes = Vec::new(); + let mut made = Ok(()); + for cache in setup.caches { + made = overlay(docker, cache, log).await; + if made.is_err() { + break; + } + volumes.push(cache.volume.clone()); + } + let started = match made { + Ok(()) => { + let start = Start { + image, + limits, + vars: &vars, + timeout, + }; + container(docker, setup, start, log).await + } + Err(err) => Err(err), + }; + // A container that did not start leaves no overlay mounted, for + // its layers to be removed. + if started.is_err() { + remove_volumes(docker, &volumes).await; + } + Ok(Session::Docker { + docker: docker.to_path_buf(), + id: started?, + volumes, + }) +} + +/// The Kubernetes Job name for a run: a DNS label at most 63 characters, +/// with a hash suffix when the check key is longer. +fn job_name(run: &str) -> String { + let mut out = String::new(); + for ch in run.chars() { + let ch = ch.to_ascii_lowercase(); + if ch.is_ascii_alphanumeric() { + out.push(ch); + } else { + out.push('-'); + } + } + let out = out.trim_matches('-'); + let out = if out.is_empty() { + "run".to_string() + } else { + out.to_string() + }; + let mut name = format!("knotless-{out}"); + if name.len() > 63 { + let hash = fnv1a(name.as_bytes()); + name.truncate(54); + let mut cut = name.trim_end_matches('-').to_string(); + cut.push_str(&format!("-{:08x}", hash as u32)); + name = cut; + } + name +} + +/// A Kubernetes label value for `value`, which a DID or a check key is +/// not: lowercase alphanumerics, `-`, `_` and `.`, at most 63. +fn label_value(value: &str) -> String { + let mut out = String::new(); + for ch in value.chars() { + let ch = ch.to_ascii_lowercase(); + if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') { + out.push(ch); + } else { + out.push('-'); + } + } + let out = out.trim_matches(|c: char| !c.is_ascii_alphanumeric()); + let out = if out.is_empty() { "unknown" } else { out }; + if out.len() <= 63 { + out.to_string() + } else { + let hash = fnv1a(out.as_bytes()); + let mut cut = out[..54] + .trim_end_matches(|c: char| !c.is_ascii_alphanumeric()) + .to_string(); + cut.push_str(&format!("-{:08x}", hash as u32)); + cut + } +} + +fn fnv1a(bytes: &[u8]) -> u64 { + let mut hash: u64 = 0xcbf29ce484222325; + for byte in bytes { + hash ^= *byte as u64; + hash = hash.wrapping_mul(0x100000001b3); + } + hash +} + +/// The Job one run becomes: one pod, one container that stays alive for +/// each step's `kubectl exec`. +#[allow(clippy::too_many_arguments)] +fn job_manifest( + job: &str, + setup: &Setup<'_>, + image: &str, + limits: &Limits, + namespace: &str, + service_account: &Option, + node_selector: &[(String, String)], + timeout: Duration, +) -> Value { + let runner = label_value(setup.owner.runner); + let run = label_value(setup.owner.run); + let mut requests = serde_json::Map::new(); + if let Some(cpus) = &limits.cpus { + requests.insert("cpu".into(), json!(cpus)); + } + if let Some(memory) = limits.memory { + requests.insert("memory".into(), json!(memory.to_string())); + } + let mut container = json!({ + "name": "run", + "image": image, + "command": ["sh", "-c", "tail -f /dev/null"], + "workingDir": "/work", + "securityContext": {"allowPrivilegeEscalation": false}, + }); + if !requests.is_empty() { + container["resources"] = json!({"requests": requests}); + } + let mut pod = json!({ + "restartPolicy": "Never", + "automountServiceAccountToken": false, + "enableServiceLinks": false, + "containers": [container], + // One run's pod per node: the tenant node is a Firecracker guest + // and therefore the isolation boundary for untrusted pull code. + "affinity": {"podAntiAffinity": {"requiredDuringSchedulingIgnoredDuringExecution": [{ + "labelSelector": {"matchLabels": {RUNNER_LABEL: runner}}, + "topologyKey": "kubernetes.io/hostname" + }]}}, + }); + if let Some(service_account) = service_account { + pod["serviceAccountName"] = json!(service_account); + } + if !node_selector.is_empty() { + let selector: serde_json::Map = node_selector + .iter() + .map(|(k, v)| (k.clone(), json!(v))) + .collect(); + pod["nodeSelector"] = json!(selector); + } + let deadline = setup + .run_timeout + .unwrap_or_else(|| timeout.saturating_mul(4)) + .as_secs() + .max(1); + json!({ + "apiVersion": "batch/v1", + "kind": "Job", + "metadata": { + "name": job, + "namespace": namespace, + "labels": {RUNNER_LABEL: runner, RUN_LABEL: run}, + "annotations": {RUNNER_LABEL: setup.owner.runner, RUN_LABEL: setup.owner.run}, + }, + "spec": { + "backoffLimit": 0, + "activeDeadlineSeconds": deadline, + "ttlSecondsAfterFinished": 300, + "template": { + "metadata": {"labels": {RUNNER_LABEL: runner, RUN_LABEL: run}}, + "spec": pod, + }, + }, + }) +} + +/// The kubectl client for a Kubernetes executor. +fn kube_client( + kubectl: &Path, + kubeconfig: &Option, + context: &Option, + namespace: &str, +) -> Command { + let mut command = Command::new(kubectl); + if let Some(kubeconfig) = kubeconfig { + command.arg("--kubeconfig").arg(kubeconfig); + } + if let Some(context) = context { + command.arg("--context").arg(context); + } + command.args(["-n", namespace]); + command +} + +/// Starts a run's Job and streams its checkout in: its pod's name. +#[allow(clippy::too_many_arguments)] +async fn start_kubernetes( + kubectl: &Path, + kubeconfig: &Option, + context: &Option, + namespace: &str, + job: &str, + manifest: &str, + setup: &Setup<'_>, + log: &std::fs::File, + timeout: Duration, +) -> Result { + let started = async { + ensure_network_policy( + kubectl, + kubeconfig, + context, + namespace, + setup.owner.runner, + log, + ) + .await?; + apply_job(kubectl, kubeconfig, context, namespace, manifest, log).await?; + let wait = setup + .run_timeout + .unwrap_or_else(|| timeout.saturating_mul(4)) + .max(timeout); + wait_ready(kubectl, kubeconfig, context, namespace, job, wait, log).await?; + let pod = pod_name(kubectl, kubeconfig, context, namespace, job).await?; + crate::telemetry::container("pod ready", &pod, None); + mkdirs(kubectl, kubeconfig, context, namespace, &pod, log).await?; + tar_in( + kubectl, kubeconfig, context, namespace, &pod, "/work", setup.dir, log, + ) + .await?; + if let Some(steps) = setup.steps { + tar_in( + kubectl, kubeconfig, context, namespace, &pod, "/steps", steps, log, + ) + .await?; + } + for cache in setup.caches { + exec_ok( + kubectl, + kubeconfig, + context, + namespace, + &pod, + &["mkdir", "-p", &cache.path], + log, + ) + .await?; + tar_in( + kubectl, + kubeconfig, + context, + namespace, + &pod, + &cache.path, + &cache.lower, + log, + ) + .await?; + } + Ok::(pod) + } + .await; + if started.is_err() { + let _ = delete_job(kubectl, kubeconfig, context, namespace, job).await; + } + started +} + +/// Makes the runner's pods refuse every ingress, so one run's pod cannot +/// reach another. Egress is untouched: a step still reaches its image's +/// network. Named for the runner, so every run in the namespace shares it. +async fn ensure_network_policy( + kubectl: &Path, + kubeconfig: &Option, + context: &Option, + namespace: &str, + runner: &str, + log: &std::fs::File, +) -> Result<()> { + let runner = label_value(runner); + let mut name = format!("knotless-deny-{runner}"); + if name.len() > 63 { + let hash = fnv1a(name.as_bytes()); + name.truncate(54); + let mut cut = name.trim_end_matches('-').to_string(); + cut.push_str(&format!("-{:08x}", hash as u32)); + name = cut; + } + let policy = json!({ + "apiVersion": "networking.k8s.io/v1", + "kind": "NetworkPolicy", + "metadata": {"name": name, "namespace": namespace}, + "spec": { + "podSelector": {"matchLabels": {RUNNER_LABEL: runner}}, + "policyTypes": ["Ingress"], + "ingress": [], + }, + }) + .to_string(); + match apply_job(kubectl, kubeconfig, context, namespace, &policy, log).await { + Ok(()) => Ok(()), + Err(err) => { + // Concurrent runs make the same policy: AlreadyExists is fine + // when the named policy is there. + let found = kube_client(kubectl, kubeconfig, context, namespace) + .args(["get", "networkpolicy", &name]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .await + .map(|s| s.success()) + .unwrap_or(false); + if found { Ok(()) } else { Err(err) } + } + } +} + +async fn exec_ok( + kubectl: &Path, + kubeconfig: &Option, + context: &Option, + namespace: &str, + pod: &str, + args: &[&str], + log: &std::fs::File, +) -> Result<()> { + let status = kube_client(kubectl, kubeconfig, context, namespace) + .args(["exec", pod, "--"]) + .args(args) + .stdout(Stdio::null()) + .stderr(log.try_clone()?) + .status() + .await + .context("running kubectl exec")?; + if !status.success() { + bail!("kubectl exec {} failed: see the log", args.join(" ")); + } + Ok(()) +} + +async fn apply_job( + kubectl: &Path, + kubeconfig: &Option, + context: &Option, + namespace: &str, + manifest: &str, + log: &std::fs::File, +) -> Result<()> { + let mut command = kube_client(kubectl, kubeconfig, context, namespace); + command + .args(["apply", "-f", "-"]) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(log.try_clone()?); + let mut child = command.spawn().context("starting kubectl apply")?; + let mut stdin = child.stdin.take().context("kubectl apply has no stdin")?; + stdin + .write_all(manifest.as_bytes()) + .await + .context("writing the Job manifest")?; + drop(stdin); + let status = child.wait().await.context("waiting for kubectl apply")?; + if !status.success() { + bail!("the Job was not applied ({status:?}): see the log"); + } + Ok(()) +} + +async fn wait_ready( + kubectl: &Path, + kubeconfig: &Option, + context: &Option, + namespace: &str, + job: &str, + timeout: Duration, + log: &std::fs::File, +) -> Result<()> { + let selector = format!("job-name={job}"); + let timeout = format!("{}s", timeout.as_secs().max(1)); + let status = kube_client(kubectl, kubeconfig, context, namespace) + .args([ + "wait", + "--for=condition=Ready", + "pod", + "-l", + &selector, + "--timeout", + &timeout, + ]) + .stdout(Stdio::null()) + .stderr(log.try_clone()?) + .status() + .await + .context("waiting for the Job's pod")?; + if !status.success() { + bail!("the Job's pod did not become Ready: see the log"); + } + Ok(()) +} + +async fn pod_name( + kubectl: &Path, + kubeconfig: &Option, + context: &Option, + namespace: &str, + job: &str, +) -> Result { + let selector = format!("job-name={job}"); + let output = kube_client(kubectl, kubeconfig, context, namespace) + .args([ + "get", + "pods", + "-l", + &selector, + "-o", + "jsonpath={.items[0].metadata.name}", + ]) + .output() + .await + .context("looking for the Job's pod")?; + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + bail!("kubectl get pods failed: {}", stderr.trim()); + } + let pod = String::from_utf8_lossy(&output.stdout).trim().to_string(); + if pod.is_empty() { + bail!("the Job has no pod"); + } + Ok(pod) +} + +async fn mkdirs( + kubectl: &Path, + kubeconfig: &Option, + context: &Option, + namespace: &str, + pod: &str, + log: &std::fs::File, +) -> Result<()> { + let status = kube_client(kubectl, kubeconfig, context, namespace) + .args([ + "exec", pod, "--", "mkdir", "-p", "/work", "/steps", "/outputs", + ]) + .stdout(Stdio::null()) + .stderr(log.try_clone()?) + .status() + .await + .context("creating the run's directories")?; + if !status.success() { + bail!("creating /work, /steps and /outputs failed: see the log"); + } + Ok(()) +} + +#[allow(clippy::too_many_arguments)] +async fn tar_in( + kubectl: &Path, + kubeconfig: &Option, + context: &Option, + namespace: &str, + pod: &str, + dest: &str, + source: &Path, + log: &std::fs::File, +) -> Result<()> { + let mut tar = Command::new("tar") + .args(["-cf", "-", "-C"]) + .arg(source) + .arg(".") + .stdout(Stdio::piped()) + .stderr(log.try_clone()?) + .spawn() + .context("starting tar")?; + let mut command = kube_client(kubectl, kubeconfig, context, namespace); + command + .args(["exec", "-i", pod, "--", "tar", "-xf", "-", "-C", dest]) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(log.try_clone()?); + let mut child = command.spawn().context("starting kubectl exec tar")?; + let mut from = tar.stdout.take().context("tar has no stdout")?; + let mut to = child.stdin.take().context("kubectl exec has no stdin")?; + tokio::io::copy(&mut from, &mut to) + .await + .context("streaming the checkout")?; + drop(to); + let status = child.wait().await.context("waiting for kubectl exec tar")?; + let tar_status = tar.wait().await.context("waiting for tar")?; + if !status.success() { + bail!("streaming {} into {dest} failed", source.display()); + } + if !tar_status.success() { + bail!("tar reading {} failed", source.display()); + } + Ok(()) +} + +#[allow(clippy::too_many_arguments)] +async fn tar_out( + kubectl: &Path, + kubeconfig: &Option, + context: &Option, + namespace: &str, + pod: &str, + source: &str, + dest: &Path, + log: &std::fs::File, +) -> Result<()> { + let mut command = kube_client(kubectl, kubeconfig, context, namespace); + command + .args(["exec", pod, "--", "tar", "-cf", "-", "-C", source, "."]) + .stdout(Stdio::piped()) + .stderr(log.try_clone()?); + let mut child = command.spawn().context("starting kubectl exec tar")?; + let mut from = child.stdout.take().context("kubectl exec has no stdout")?; + let mut tar = Command::new("tar") + .args(["-xf", "-", "-C"]) + .arg(dest) + .stdin(Stdio::piped()) + .stderr(log.try_clone()?) + .spawn() + .context("starting tar")?; + let mut to = tar.stdin.take().context("tar has no stdin")?; + tokio::io::copy(&mut from, &mut to) + .await + .context("streaming the outputs")?; + drop(to); + let status = child.wait().await.context("waiting for kubectl exec tar")?; + let tar_status = tar.wait().await.context("waiting for tar")?; + if !status.success() { + bail!("streaming {source} out of the pod failed"); + } + if !tar_status.success() { + bail!("tar writing {} failed", dest.display()); + } + Ok(()) +} + +async fn delete_job( + kubectl: &Path, + kubeconfig: &Option, + context: &Option, + namespace: &str, + job: &str, +) -> Result<()> { + let status = kube_client(kubectl, kubeconfig, context, namespace) + .args(["delete", "job", job, "--ignore-not-found", "--wait=false"]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .await + .context("deleting the Job")?; + if !status.success() { + bail!("deleting the Job failed"); + } + crate::telemetry::container("job deleted", job, None); + Ok(()) +} + +/// The shell payload that gives a step its variables and secrets over +/// stdin, so they never reach argv, the pod spec or a file. +fn step_payload(script: &str, own: &[(String, String)]) -> Vec { + let mut payload = String::new(); + for (name, value) in own { + payload.push_str("export "); + payload.push_str(name); + payload.push('='); + payload.push_str(&shell_quote(value)); + payload.push('\n'); + } + payload.push_str("exec sh -ec "); + payload.push_str(&shell_quote(script)); + payload.push('\n'); + payload.into_bytes() +} + +fn shell_quote(value: &str) -> String { + format!("'{}'", value.replace('\'', "'\\''")) +} + /// Makes the overlay volume `cache`, writing what goes wrong to `log`. async fn overlay(docker: &Path, cache: &Cache, log: &std::fs::File) -> Result<()> { let layers = format!( @@ -487,6 +1133,17 @@ impl Executor { going: &Mutex>, work: Option<&Path>, ) { + if let Executor::Kubernetes { + kubeconfig, + context, + namespace, + kubectl, + .. + } = self + { + remove_left_kubernetes(kubectl, kubeconfig, context, namespace, runner, going).await; + return; + } if !matches!(self, Executor::Docker { .. }) { return; } @@ -525,6 +1182,47 @@ impl Executor { } } +/// Deletes the Jobs a dead runner left in the tenant cluster: those +/// labeled with its DID and not of a run still going, by its check key. +async fn remove_left_kubernetes( + kubectl: &Path, + kubeconfig: &Option, + context: &Option, + namespace: &str, + runner: &str, + going: &Mutex>, +) { + let selector = format!("{RUNNER_LABEL}={}", label_value(runner)); + let output = kube_client(kubectl, kubeconfig, context, namespace) + .args(["get", "jobs", "-l", &selector, "-o", "json"]) + .output() + .await; + let Ok(output) = output else { + return; + }; + if !output.status.success() { + return; + } + let Ok(value) = serde_json::from_slice::(&output.stdout) else { + return; + }; + let Some(items) = value["items"].as_array() else { + return; + }; + for item in items { + let Some(name) = item["metadata"]["name"].as_str() else { + continue; + }; + let run = item["metadata"]["annotations"][RUN_LABEL] + .as_str() + .unwrap_or_default(); + let live = going.lock().unwrap().contains(run); + if !live { + let _ = delete_job(kubectl, kubeconfig, context, namespace, name).await; + } + } +} + /// The lines `cmd` prints, or none if it cannot be run. async fn words(cmd: &mut Command) -> Vec { let listed = cmd.stderr(Stdio::null()).output().await; @@ -612,6 +1310,41 @@ impl Session { cmd.args([id, "sh", "-ec", script]); cmd } + Session::Kubernetes { + kubectl, + kubeconfig, + context, + namespace, + pod, + .. + } => { + // The payload, variables and secrets among it, is written + // to this process's stdin by [`Session::input`]: nothing + // reaches argv, the pod spec or a file. + let mut cmd = kube_client(kubectl, kubeconfig, context, namespace); + cmd.args(["exec", "-i", pod, "--", "sh", "-s"]); + cmd + } + } + } + + /// The stdin payload a step needs, where its executor has one: under + /// kubernetes, the run's variables and the step's own, then the + /// script. None under docker and shell, whose variables reach the + /// process through the client's environment. + pub fn input(&self, script: &str, own: &[(String, String)]) -> Option> { + match self { + Session::Kubernetes { vars, .. } => { + let mut vars = vars.clone(); + for (name, value) in own { + match vars.iter_mut().find(|(k, _)| k == name) { + Some((_, held)) => *held = value.clone(), + None => vars.push((name.clone(), value.clone())), + } + } + Some(step_payload(script, &vars)) + } + Session::Shell { .. } | Session::Docker { .. } => None, } } @@ -619,11 +1352,26 @@ impl Session { /// [`Session::end_left`] to tell which a step left: none under the /// shell executor, whose steps share the runner's host. pub async fn running(&self) -> Option> { - let Session::Docker { docker, id, .. } = self else { - return None; + let mut list = match self { + Session::Docker { docker, id, .. } => { + let mut list = client(docker, &[]); + list.args(["exec", id, "sh", "-c", RUNNING]); + list + } + Session::Kubernetes { + kubectl, + kubeconfig, + context, + namespace, + pod, + .. + } => { + let mut list = kube_client(kubectl, kubeconfig, context, namespace); + list.args(["exec", pod, "--", "sh", "-c", RUNNING]); + list + } + Session::Shell { .. } => return None, }; - let mut list = client(docker, &[]); - list.args(["exec", id, "sh", "-c", RUNNING]); let ids = words(&mut list).await; Some(ids.iter().filter_map(|p| p.parse().ok()).collect()) } @@ -638,7 +1386,7 @@ impl Session { step: &str, mut log: &std::fs::File, ) -> Result<()> { - let (Some(before), Session::Docker { docker, id, .. }) = (before, self) else { + let Some(before) = before else { return Ok(()); }; let now = self.running().await.unwrap_or_default(); @@ -646,13 +1394,34 @@ impl Session { if left.is_empty() { return Ok(()); } - let mut kill = client(docker, &[]); - kill.args(["exec", id, "kill", "-9"]).args(&left); - let _ = kill - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .status() - .await; + match self { + Session::Docker { docker, id, .. } => { + let mut kill = client(docker, &[]); + kill.args(["exec", id, "kill", "-9"]).args(&left); + let _ = kill + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .await; + } + Session::Kubernetes { + kubectl, + kubeconfig, + context, + namespace, + pod, + .. + } => { + let mut kill = kube_client(kubectl, kubeconfig, context, namespace); + kill.args(["exec", pod, "--", "kill", "-9"]).args(&left); + let _ = kill + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .await; + } + Session::Shell { .. } => return Ok(()), + } use std::io::Write; let n = left.len(); let said = format!("==> ended {n} of what {step} left running, as it was given secrets\n"); @@ -665,20 +1434,69 @@ impl Session { /// overlay volumes, which unmounts them, for its layers to be kept or /// removed. pub async fn stop(self) { - if let Session::Docker { - docker, - id, - volumes, - } = self - { - let _ = client(&docker, &[]) - .args(["rm", "-f", &id]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .status() - .await; - crate::telemetry::container("container stopped", &id, None); - remove_volumes(&docker, &volumes).await; + match self { + Session::Docker { + docker, + id, + volumes, + } => { + let _ = client(&docker, &[]) + .args(["rm", "-f", &id]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .await; + crate::telemetry::container("container stopped", &id, None); + remove_volumes(&docker, &volumes).await; + } + Session::Kubernetes { + kubectl, + kubeconfig, + context, + namespace, + job, + pod, + outputs, + caches, + .. + } => { + // The outputs and the caches live in the pod until it goes: + // stream them back, the caches into their run's upper + // layers for the lease to promote or keep. + let log = std::fs::OpenOptions::new() + .write(true) + .open("/dev/null") + .ok(); + if let Some(log) = &log { + for cache in &caches { + let _ = std::fs::create_dir_all(&cache.upper); + let _ = tar_out( + &kubectl, + &kubeconfig, + &context, + &namespace, + &pod, + &cache.path, + &cache.upper, + log, + ) + .await; + } + let _ = tar_out( + &kubectl, + &kubeconfig, + &context, + &namespace, + &pod, + "/outputs", + &outputs, + log, + ) + .await; + } + let _ = delete_job(&kubectl, &kubeconfig, &context, &namespace, &job).await; + } + Session::Shell { .. } => {} } } } @@ -688,17 +1506,30 @@ impl Session { /// everything it started. pub async fn run( mut cmd: Command, + input: Option>, log: &std::fs::File, timeout: Duration, cancelled: impl Future, ) -> Result { let started = Instant::now(); - cmd.stdin(Stdio::null()) - .stdout(log.try_clone()?) - .stderr(log.try_clone()?) - .kill_on_drop(true) - .process_group(0); + cmd.stdin(if input.is_some() { + Stdio::piped() + } else { + Stdio::null() + }) + .stdout(log.try_clone()?) + .stderr(log.try_clone()?) + .kill_on_drop(true) + .process_group(0); let mut child = cmd.spawn().context("starting a step")?; + if let Some(input) = input { + let mut stdin = child.stdin.take().context("the step has no stdin")?; + stdin + .write_all(&input) + .await + .context("writing the step's variables")?; + drop(stdin); + } // Its exit, or, killed, whether for being cancelled. let ended = tokio::select! { status = tokio::time::timeout(timeout, child.wait()) => status.map_err(|_| false), @@ -892,6 +1723,7 @@ mod tests { }, behind: false, vouched: false, + run_timeout: None, }; // A pull's, behind the rest for CPU. let behind = Setup { @@ -905,6 +1737,7 @@ mod tests { .unwrap(); let ran = run( session.command("echo ran", &[]), + None, &log, time, std::future::pending(), @@ -1067,6 +1900,7 @@ mod tests { }, behind: false, vouched: false, + run_timeout: None, }; let within = |network: &str| Executor::Docker { image: "rust:1".into(), @@ -1173,14 +2007,14 @@ mod tests { }; let cmd = shell.command("echo hi $WHO; echo err >&2; exit 3", &[]); let never = std::future::pending::<()>; - let exit = run(cmd, &log, Duration::from_secs(10), never()) + let exit = run(cmd, None, &log, Duration::from_secs(10), never()) .await .unwrap(); assert_eq!(exit.code, Some(3)); assert!(!exit.success()); // A child of the shell is killed with it. let cmd = shell.command("sleep 30 & sleep 30", &[]); - let exit = run(cmd, &log, Duration::from_millis(200), never()) + let exit = run(cmd, None, &log, Duration::from_millis(200), never()) .await .unwrap(); assert_eq!((exit.code, exit.cancelled), (None, false)); @@ -1188,7 +2022,9 @@ mod tests { // Cancelled, likewise. let cmd = shell.command("sleep 30 & sleep 30", &[]); let soon = tokio::time::sleep(Duration::from_millis(200)); - let exit = run(cmd, &log, Duration::from_secs(10), soon).await.unwrap(); + let exit = run(cmd, None, &log, Duration::from_secs(10), soon) + .await + .unwrap(); assert_eq!((exit.code, exit.cancelled), (None, true)); assert!(exit.elapsed < Duration::from_secs(5)); let text = std::fs::read_to_string(dir.path().join("log")).unwrap(); @@ -1196,9 +2032,386 @@ mod tests { let mut missing = shell.command("true", &[]); missing.current_dir(dir.path().join("gone")); assert!( - run(missing, &log, Duration::from_secs(1), never()) + run(missing, None, &log, Duration::from_secs(1), never()) .await .is_err() ); } + + fn fake_kubectl(dir: &Path) -> PathBuf { + let path = dir.join("kubectl"); + let calls = dir.join("calls"); + let jobs = dir.join("jobs.json"); + std::fs::write( + &jobs, + r#"{"items":[ + {"metadata":{"name":"old","annotations":{"dev.knotless.run":"old-run"}}}, + {"metadata":{"name":"live","annotations":{"dev.knotless.run":"live-run"}}} + ]}"#, + ) + .unwrap(); + let script = format!( + r#"#!/bin/sh +echo "$*" >> {calls} +case "$*" in + *" apply "*) cat >/dev/null ;; + *" get pods "*) printf 'pod-1' ;; + *" get jobs "*) cat {jobs} ;; + *" sh -s"*) cat >/dev/null ;; + *" sh -c for p in /proc"*) printf '1\n7\n9\nx\n' ;; + *" kill -9 "*) ;; + *" tar -xf "*) cat >/dev/null ;; +esac +exit 0 +"#, + calls = calls.display(), + jobs = jobs.display() + ); + std::fs::write(&path, script).unwrap(); + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap(); + path + } + + #[test] + fn kubernetes_names_and_payload() { + assert_eq!(job_name("k1"), "knotless-k1"); + assert!(job_name(&"a".repeat(80)).len() <= 63); + assert_eq!(label_value("did:plc:r"), "did-plc-r"); + let payload = + String::from_utf8(step_payload("echo $K", &[("K".into(), "v'x".into())])).unwrap(); + assert_eq!(payload, "export K='v'\\''x'\nexec sh -ec 'echo $K'\n"); + } + + #[test] + fn kubernetes_manifest_has_isolation_and_limits() { + let dir = tempfile::tempdir().unwrap(); + let limits = Limits { + cpus: Some("1".into()), + memory: Some(2 << 30), + ..Limits::default() + }; + let setup = Setup { + dir: dir.path(), + vars: &[], + image: None, + caches: &[], + outputs: dir.path(), + steps: None, + owner: Owner { + runner: "did:plc:r", + run: "k1", + }, + behind: false, + vouched: false, + run_timeout: Some(Duration::from_secs(60)), + }; + let value = job_manifest( + "knotless-k1", + &setup, + "busybox:1.36", + &limits, + "ci", + &Some("runner".into()), + &[("pool".into(), "ci".into())], + Duration::from_secs(10), + ); + assert_eq!(value["spec"]["activeDeadlineSeconds"], 60); + assert_eq!( + value["spec"]["template"]["spec"]["serviceAccountName"], + "runner" + ); + assert_eq!( + value["spec"]["template"]["spec"]["nodeSelector"]["pool"], + "ci" + ); + assert_eq!( + value["spec"]["template"]["spec"]["containers"][0]["resources"]["requests"]["cpu"], + "1" + ); + assert!(value["spec"]["template"]["spec"]["affinity"]["podAntiAffinity"] + ["requiredDuringSchedulingIgnoredDuringExecution"][0] + ["labelSelector"]["matchLabels"]["dev.knotless.runner"] + .is_string()); + } + + #[tokio::test] + async fn kubernetes_one_job_a_run() { + let dir = tempfile::tempdir().unwrap(); + let kubectl = fake_kubectl(dir.path()); + let log_path = dir.path().join("log"); + let log = std::fs::File::create(&log_path).unwrap(); + let src = dir.path().join("src"); + let steps = dir.path().join("steps"); + let outputs = dir.path().join("outputs"); + std::fs::create_dir_all(&src).unwrap(); + std::fs::create_dir_all(&steps).unwrap(); + std::fs::create_dir_all(&outputs).unwrap(); + std::fs::write(src.join("file"), "x").unwrap(); + let lower = dir.path().join("lower"); + std::fs::create_dir_all(&lower).unwrap(); + std::fs::write(lower.join("warm"), "x").unwrap(); + let caches = [Cache { + volume: "v".into(), + path: "/cache".into(), + run: "k1".into(), + lower: lower.clone(), + upper: dir.path().join("upper"), + work: dir.path().join("work"), + }]; + let setup = Setup { + dir: &src, + vars: &[], + image: None, + caches: &caches, + outputs: &outputs, + steps: Some(&steps), + owner: Owner { + runner: "did:plc:r", + run: "k1", + }, + behind: false, + vouched: false, + run_timeout: Some(Duration::from_secs(60)), + }; + let executor = Executor::Kubernetes { + image: "busybox:1.36".into(), + limits: Limits::default(), + kubeconfig: None, + context: None, + namespace: "ci".into(), + kubectl: kubectl.clone(), + service_account: None, + node_selector: Vec::new(), + }; + let session = executor + .start_with(&kubectl, &setup, &log, Duration::from_secs(10)) + .await + .unwrap(); + let Session::Kubernetes { pod, .. } = &session else { + panic!("not a kubernetes session"); + }; + assert_eq!(pod, "pod-1"); + let cmd = session.command("echo hi", &[]); + let input = session.input("echo hi", &[("K".into(), "v".into())]); + let exit = run( + cmd, + input, + &log, + Duration::from_secs(10), + std::future::pending(), + ) + .await + .unwrap(); + assert!(exit.success()); + let before = session.running().await.unwrap(); + assert_eq!(before, BTreeSet::from([1, 7, 9])); + session + .end_left(Some(&BTreeSet::from([1, 7])), "deploy", &log) + .await + .unwrap(); + session.stop().await; + let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap(); + for wanted in [ + "apply -f -", + "wait --for=condition=Ready", + "get pods", + "exec -i pod-1 -- tar -xf - -C /work", + "exec -i pod-1 -- tar -xf - -C /steps", + "exec pod-1 -- mkdir -p /cache", + "exec -i pod-1 -- tar -xf - -C /cache", + "exec -i pod-1 -- sh -s", + "exec pod-1 -- tar -cf - -C /cache .", + "exec pod-1 -- tar -cf - -C /outputs .", + "delete job knotless-k1", + ] { + assert!(calls.contains(wanted), "missing {wanted:?} in:\n{calls}"); + } + let said = std::fs::read_to_string(&log_path).unwrap(); + assert!( + said.contains("ended 1 of what deploy left running"), + "{said}" + ); + // Orphans are deleted, but a live run's Job is not. + let going = Mutex::new(HashSet::from(["live-run".to_string()])); + executor + .remove_left_with(&kubectl, "did:plc:r", &going, None) + .await; + let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap(); + assert!(calls.contains("delete job old"), "{calls}"); + assert!(!calls.contains("delete job live"), "{calls}"); + } + fn fake_kubectl_script(dir: &Path, body: &str) -> PathBuf { + let path = dir.join("kubectl"); + let calls = dir.join("calls"); + let script = format!( + r#"#!/bin/sh +echo "$*" >> {calls} +case "$*" in + *" apply "*) cat >/dev/null ;; + *" get pods "*) printf 'pod-1' ;; + *" get jobs "*) printf '{{"items":[]}}' ;; + *" sh -s"*) cat >/dev/null ;; + *" tar -xf "*) cat >/dev/null ;; +esac +{body} +exit 0 +"#, + calls = calls.display(), + ); + std::fs::write(&path, script).unwrap(); + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap(); + path + } + + fn kubernetes_setup( + dir: &Path, + kubectl: &Path, + ) -> ( + Executor, + std::path::PathBuf, + std::path::PathBuf, + std::path::PathBuf, + ) { + let src = dir.join("src"); + let outputs = dir.join("outputs"); + std::fs::create_dir_all(&src).unwrap(); + std::fs::create_dir_all(&outputs).unwrap(); + let executor = Executor::Kubernetes { + image: "busybox:1.36".into(), + limits: Limits::default(), + kubeconfig: None, + context: None, + namespace: "ci".into(), + kubectl: kubectl.to_path_buf(), + service_account: None, + node_selector: Vec::new(), + }; + (executor, src, outputs, dir.join("log")) + } + + #[tokio::test] + async fn kubernetes_a_pod_that_never_readies_removes_the_job() { + let dir = tempfile::tempdir().unwrap(); + let kubectl = fake_kubectl_script(dir.path(), "case \"$*\" in *' wait '*) exit 1;; esac"); + let (executor, src, outputs, log_path) = kubernetes_setup(dir.path(), &kubectl); + let log = std::fs::File::create(&log_path).unwrap(); + let setup = Setup { + dir: &src, + vars: &[], + image: None, + caches: &[], + outputs: &outputs, + steps: None, + owner: Owner { + runner: "did:r", + run: "k1", + }, + behind: false, + vouched: false, + run_timeout: Some(Duration::from_secs(60)), + }; + let err = executor + .start_with(&kubectl, &setup, &log, Duration::from_secs(10)) + .await + .unwrap_err(); + let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap_or_default(); + let said = std::fs::read_to_string(&log_path).unwrap_or_default(); + assert!( + err.to_string().contains("did not become Ready"), + "{err}\ncalls:\n{calls}\nlog:\n{said}" + ); + assert!(calls.contains("delete job knotless-k1"), "{calls}"); + } + + #[tokio::test] + async fn kubernetes_a_denied_policy_stops_the_run() { + let dir = tempfile::tempdir().unwrap(); + let kubectl = dir.path().join("kubectl"); + let calls = dir.path().join("calls"); + let script = format!( + r#"#!/bin/sh +echo "$*" >> {calls} +case "$*" in + *" apply "*) body=$(cat); case "$body" in *NetworkPolicy*) exit 1;; esac ;; + *" get networkpolicy "*) exit 1 ;; + *" get pods "*) printf 'pod-1' ;; + *" tar -xf "*) cat >/dev/null ;; +esac +exit 0 +"#, + calls = calls.display() + ); + std::fs::write(&kubectl, script).unwrap(); + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&kubectl, std::fs::Permissions::from_mode(0o755)).unwrap(); + let (executor, src, outputs, log_path) = kubernetes_setup(dir.path(), &kubectl); + let log = std::fs::File::create(&log_path).unwrap(); + let setup = Setup { + dir: &src, + vars: &[], + image: None, + caches: &[], + outputs: &outputs, + steps: None, + owner: Owner { + runner: "did:r", + run: "k1", + }, + behind: false, + vouched: false, + run_timeout: Some(Duration::from_secs(60)), + }; + let err = executor + .start_with(&kubectl, &setup, &log, Duration::from_secs(10)) + .await + .unwrap_err(); + let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap_or_default(); + assert!( + err.to_string().contains("not applied") || err.to_string().contains("not"), + "{err}\ncalls:\n{calls}" + ); + assert!(calls.contains("delete job knotless-k1"), "{calls}"); + } + + #[tokio::test] + async fn kubernetes_a_step_that_overruns_is_killed() { + let dir = tempfile::tempdir().unwrap(); + let kubectl = fake_kubectl_script(dir.path(), "case \"$*\" in *' sh -s'*) sleep 5;; esac"); + let (executor, src, outputs, log_path) = kubernetes_setup(dir.path(), &kubectl); + let log = std::fs::File::create(&log_path).unwrap(); + let setup = Setup { + dir: &src, + vars: &[], + image: None, + caches: &[], + outputs: &outputs, + steps: None, + owner: Owner { + runner: "did:r", + run: "k1", + }, + behind: false, + vouched: false, + run_timeout: Some(Duration::from_secs(60)), + }; + let session = executor + .start_with(&kubectl, &setup, &log, Duration::from_secs(10)) + .await + .unwrap(); + let cmd = session.command("sleep", &[]); + let input = session.input("sleep", &[]); + let exit = run( + cmd, + input, + &log, + Duration::from_millis(200), + std::future::pending(), + ) + .await + .unwrap(); + assert_eq!((exit.code, exit.cancelled), (None, false)); + session.stop().await; + } } diff --git a/crates/knotless-run/src/lib.rs b/crates/knotless-run/src/lib.rs index e6ec2846..2b4cbe54 100644 --- a/crates/knotless-run/src/lib.rs +++ b/crates/knotless-run/src/lib.rs @@ -126,7 +126,7 @@ use tokio::sync::{Notify, Semaphore}; use turns::{Class, Rank, Turns, Whose}; use workflow::{Superseded, Workflow}; -const USAGE: &str = "usage: knotless-run [--jetstream ] [--executor docker|shell] [--image ] [--secrets ] [--secrets-key ] [--state ] [--work ] [--step-timeout ] [--run-timeout ] [--cpus ] [--memory ] [--pids ] [--network ] [--runtime ] [--trusted-runtime ] [--trust ] [--resolv ] [--jobs ] [--pull-jobs ] [--owner-jobs ] [--allowance ] [--private-allowance ] [--cache-max ] [--disk-floor ] [--caches ] [--paused ] [--keep-checks ] [--sweep ] [--stop-wait ] [--poll ] [--pulls] [--pulls-only /] [--unmetered ] [--queue [--batch ] [--depth 1|2] [--notice ] [--review-wait ]] [--discover] [/...]"; +const USAGE: &str = "usage: knotless-run [--jetstream ] [--executor docker|kubernetes|shell] [--image ] [--kubeconfig ] [--context ] [--namespace ] [--kubectl ] [--service-account ] [--node-selector ] [--secrets ] [--secrets-key ] [--state ] [--work ] [--step-timeout ] [--run-timeout ] [--cpus ] [--memory ] [--pids ] [--network ] [--runtime ] [--trusted-runtime ] [--trust ] [--resolv ] [--jobs ] [--pull-jobs ] [--owner-jobs ] [--allowance ] [--private-allowance ] [--cache-max ] [--disk-floor ] [--caches ] [--paused ] [--keep-checks ] [--sweep ] [--stop-wait ] [--poll ] [--pulls] [--pulls-only /] [--unmetered ] [--queue [--batch ] [--depth 1|2] [--notice ] [--review-wait ]] [--discover] [/...]"; #[derive(Debug, Clone, PartialEq)] pub struct Options { @@ -379,9 +379,15 @@ fn secs(flag: &str, value: String) -> Result { pub fn parse_args(args: &[String]) -> Result { let mut opts = Options::new(Vec::new()); - let mut executor = "docker".to_string(); + let mut executor = "kubernetes".to_string(); let mut image = "debian:stable-slim".to_string(); let mut limits = executor::Limits::default(); + let mut kubeconfig: Option = None; + let mut kube_context: Option = None; + let mut namespace = "default".to_string(); + let mut kubectl = "kubectl".to_string(); + let mut service_account: Option = None; + let mut node_selector: Vec<(String, String)> = Vec::new(); let mut it = args.iter().skip(1); while let Some(flag) = it.next() { let mut value = || it.next().cloned().ok_or(format!("{flag} takes a value")); @@ -389,6 +395,21 @@ pub fn parse_args(args: &[String]) -> Result { "--jetstream" => opts.jetstream = value()?, "--executor" => executor = value()?, "--image" => image = value()?, + "--kubeconfig" => kubeconfig = Some(value()?.into()), + "--context" => kube_context = Some(value()?), + "--namespace" => namespace = value()?, + "--kubectl" => kubectl = value()?, + "--service-account" => service_account = Some(value()?), + "--node-selector" => { + let pair = value()?; + let (key, value) = pair + .split_once('=') + .ok_or("--node-selector takes key=value")?; + if key.is_empty() || value.is_empty() { + return Err("--node-selector takes key=value".into()); + } + node_selector.push((key.to_string(), value.to_string())); + } "--secrets" => opts.secrets = Some(value()?.into()), "--secrets-key" => opts.secrets_key = Some(value()?.into()), "--state" => opts.state = Some(value()?.into()), @@ -482,17 +503,42 @@ pub fn parse_args(args: &[String]) -> Result { repo => opts.repos.push(repo.trim_start_matches(SCHEME).to_string()), } } - opts.executor = Executor::parse(&executor, &image) - .ok_or(format!("--executor is docker or shell, not {executor}"))?; - match &mut opts.executor { - Executor::Docker { limits: held, .. } => *held = limits, - Executor::Shell if limits != executor::Limits::default() => { + if executor == "kubernetes" { + if limits.pids.is_some() + || limits.network.is_some() + || limits.runtime.is_some() + || limits.trusted_runtime.is_some() + || limits.resolv.is_some() + { return Err( - "--cpus, --memory, --pids, --network, --runtime, --trusted-runtime and --resolv hold a run's container: they take --executor docker" + "--pids, --network, --runtime, --trusted-runtime and --resolv hold a run's container: they take --executor docker; kubernetes takes --cpus and --memory" .into(), ); } - Executor::Shell => {} + opts.executor = Executor::Kubernetes { + image, + limits, + kubeconfig, + context: kube_context, + namespace, + kubectl: kubectl.into(), + service_account, + node_selector, + }; + } else { + opts.executor = Executor::parse(&executor, &image).ok_or(format!( + "--executor is docker, kubernetes or shell, not {executor}" + ))?; + match &mut opts.executor { + Executor::Docker { limits: held, .. } => *held = limits, + Executor::Shell if limits != executor::Limits::default() => { + return Err( + "--cpus, --memory, --pids, --network, --runtime, --trusted-runtime and --resolv hold a run's container: they take --executor docker" + .into(), + ); + } + Executor::Shell | Executor::Kubernetes { .. } => {} + } } if opts.repos.is_empty() && !opts.discover { return Err(format!( @@ -504,7 +550,7 @@ pub fn parse_args(args: &[String]) -> Result { } if (opts.pulls || opts.queue) && opts.executor == Executor::Shell { return Err( - "--pulls and --queue run code anyone can open a pull with: they take --executor docker" + "--pulls and --queue run code anyone can open a pull with: they take --executor docker or kubernetes" .into(), ); } @@ -1197,6 +1243,31 @@ struct Job { impl Runner { /// Logs in as the runner's account and finds each watched repository. pub async fn start(opts: Options, ctx: Ctx) -> Result<(Arc, Vec)> { + if let Executor::Kubernetes { + ref kubeconfig, + ref kubectl, + .. + } = opts.executor + { + let found = kubeconfig + .as_ref() + .map(|p| p.exists()) + .or_else(|| { + std::env::var_os("KUBECONFIG") + .map(|paths| std::env::split_paths(&paths).any(|p| p.exists())) + }) + .or_else(|| { + std::env::var_os("HOME") + .map(|home| Path::new(&home).join(".kube/config").exists()) + }) + .unwrap_or(false); + if !found { + bail!( + "--executor kubernetes needs --kubeconfig or KUBECONFIG; no kubeconfig was found" + ); + } + let _ = kubectl; + } if (opts.pulls || opts.discover) && ctx.env.appview.is_none() { bail!( "--pulls and --discover find pulls and repositories through an appview, and KNOTLESS_APPVIEW is set empty" @@ -3281,7 +3352,11 @@ impl Runner { let outputs = dir.path().join("outputs"); std::fs::create_dir(&outputs)?; let branch = own_branch(subject, wf); - let wanted = caches(&watch.id(), kept_apart(subject, wf), wf, branch); + let wanted = if self.opts.executor.uses_caches() { + caches(&watch.id(), kept_apart(subject, wf), wf, branch) + } else { + Vec::new() + }; let lease = self.caches.lease(&wanted)?; let caches = lease.mounts(); let trusts = self.opts.trust.contains(&watch.remote.did().to_string()); @@ -3300,6 +3375,7 @@ impl Runner { // branch, the merge queue's next batch among them. behind: !subject.first(), vouched: trusts && vouched_for(subject), + run_timeout: self.opts.run_timeout, }; let timeout = self.opts.step_timeout; let _tracked = Tracked::new(self, setup.owner.run); @@ -3583,10 +3659,11 @@ impl Runner { false => None, }; let cmd = session.command(&step.run, &own); + let input = session.input(&step.run, &own); // The run's task holds the sender to its end, so this ends only // when it is set. let cancel = cancelled.wait(); - let exit = executor::run(cmd, &log, limit, async { + let exit = executor::run(cmd, input, &log, limit, async { let _ = cancel.await; }) .await?; @@ -4967,11 +5044,22 @@ mod tests { } fn args(list: &str) -> Result { + // Most cases here predate the Kubernetes default and assert the + // docker path; name it unless the case names an executor. let mut all = vec!["knotless-run".to_string()]; + if !list.split_whitespace().any(|word| word == "--executor") { + all.extend(["--executor".to_string(), "docker".to_string()]); + } all.extend(list.split_whitespace().map(str::to_string)); parse_args(&all) } + #[test] + fn kubernetes_is_the_default() { + let opts = parse_args(&["knotless-run".to_string(), "a/b".to_string()]).unwrap(); + assert!(matches!(opts.executor, Executor::Kubernetes { .. })); + } + #[test] fn arguments() { let opts = args("alice.test/proj knotless://did:plc:b/other").unwrap(); @@ -5206,7 +5294,32 @@ mod tests { ); assert_eq!( args("--executor vm a/b").unwrap_err(), - "--executor is docker or shell, not vm" + "--executor is docker, kubernetes or shell, not vm" + ); + let kube = args( + "--executor kubernetes --kubeconfig /k --context c --namespace ci --kubectl /bin/k --service-account sa --node-selector pool=ci --cpus 1 --memory 2 a/b", + ) + .unwrap(); + assert_eq!( + kube.executor, + Executor::Kubernetes { + image: "debian:stable-slim".into(), + limits: executor::Limits { + cpus: Some("1".into()), + memory: Some(2 << 30), + ..executor::Limits::default() + }, + kubeconfig: Some(PathBuf::from("/k")), + context: Some("c".into()), + namespace: "ci".into(), + kubectl: PathBuf::from("/bin/k"), + service_account: Some("sa".into()), + node_selector: vec![("pool".into(), "ci".into())], + } + ); + assert_eq!( + args("--executor kubernetes --pids 1 a/b").unwrap_err(), + "--pids, --network, --runtime, --trusted-runtime and --resolv hold a run's container: they take --executor docker; kubernetes takes --cpus and --memory" ); assert!(args("--pulls a/b").unwrap().pulls); // One named --pulls-only is watched, for its pulls. @@ -5225,7 +5338,10 @@ mod tests { ); for shell in ["--pulls", "--queue"] { let err = args(&format!("{shell} --executor shell a/b")).unwrap_err(); - assert!(err.ends_with("they take --executor docker"), "{err}"); + assert!( + err.ends_with("they take --executor docker or kubernetes"), + "{err}" + ); } } diff --git a/crates/knotless-run/tests/kubernetes_live.rs b/crates/knotless-run/tests/kubernetes_live.rs new file mode 100644 index 00000000..6e31133b --- /dev/null +++ b/crates/knotless-run/tests/kubernetes_live.rs @@ -0,0 +1,115 @@ +//! A live run of the Kubernetes executor against a tenant cluster. +//! +//! Ignored by default. Run with `KNOTLESS_K8S_LIVE=1`, `KUBECONFIG` set, +//! and `cargo test -p knotless-run --test kubernetes_live -- --ignored`. + +use std::collections::HashSet; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; +use std::time::Duration; + +use knotless_run::executor::{self, Cache, Executor, Limits, Owner, Setup}; + +#[tokio::test] +#[ignore = "needs KNOTLESS_K8S_LIVE=1 and a tenant KUBECONFIG"] +async fn a_live_job_and_a_warm_cache() { + if std::env::var("KNOTLESS_K8S_LIVE").as_deref() != Ok("1") { + return; + } + let kubeconfig = std::env::var_os("KUBECONFIG").map(PathBuf::from); + let namespace = std::env::var("KNOTLESS_K8S_NAMESPACE").unwrap_or_else(|_| "default".into()); + let dir = tempfile::tempdir().unwrap(); + let src = dir.path().join("src"); + let outputs = dir.path().join("outputs"); + let outputs2 = dir.path().join("outputs2"); + let cache = dir.path().join("cache"); + for path in [&src, &outputs, &outputs2, &cache] { + std::fs::create_dir_all(path).unwrap(); + } + std::fs::write(src.join("checked-out"), "yes\n").unwrap(); + let caches = [Cache { + volume: "live".into(), + path: "/cache".into(), + run: "live".into(), + lower: cache.clone(), + upper: cache.clone(), + work: dir.path().join("work"), + }]; + let log_path = dir.path().join("log"); + let log = std::fs::File::create(&log_path).unwrap(); + let run = format!("live-{}", std::process::id()); + let executor = Executor::Kubernetes { + image: "busybox:1.36".into(), + limits: Limits { + cpus: Some("1".into()), + memory: Some(1 << 30), + ..Limits::default() + }, + kubeconfig, + context: None, + namespace, + kubectl: PathBuf::from("kubectl"), + service_account: None, + node_selector: Vec::new(), + }; + let script = "test -f checked-out && test -f /cache/warm \ + && echo warm > /outputs/ran || echo cold > /outputs/ran; \ + echo yes > /cache/warm"; + let first = Setup { + dir: &src, + vars: &[], + image: None, + caches: &caches, + outputs: &outputs, + steps: None, + owner: Owner { + runner: "did:plc:runner", + run: &run, + }, + behind: false, + vouched: false, + run_timeout: Some(Duration::from_secs(1800)), + }; + run_once(&executor, &first, &log, script).await; + assert_eq!( + std::fs::read_to_string(outputs.join("ran")).unwrap(), + "cold\n" + ); + assert!(cache.join("warm").is_file(), "cache was not streamed back"); + + let second = Setup { + outputs: &outputs2, + ..first + }; + run_once(&executor, &second, &log, script).await; + assert_eq!( + std::fs::read_to_string(outputs2.join("ran")).unwrap(), + "warm\n" + ); + + // The next start removes Jobs a dead runner left, but not one going. + let going = Mutex::new(HashSet::new()); + executor + .remove_left_with(Path::new("kubectl"), "did:plc:runner", &going, None) + .await; +} + +async fn run_once(executor: &Executor, setup: &Setup<'_>, log: &std::fs::File, script: &str) { + let session = executor + .start_with(Path::new("kubectl"), setup, log, Duration::from_secs(600)) + .await + .unwrap(); + let cmd = session.command(script, &[]); + let input = session.input(script, &[]); + let exit = executor::run( + cmd, + input, + log, + Duration::from_secs(300), + std::future::pending(), + ) + .await + .unwrap(); + assert!(exit.success(), "step failed: {exit:?}"); + session.stop().await; +} diff --git a/docs-site/astro.config.mjs b/docs-site/astro.config.mjs index 9b30bb41..21e2d130 100644 --- a/docs-site/astro.config.mjs +++ b/docs-site/astro.config.mjs @@ -48,6 +48,7 @@ export default defineConfig({ page('Agents', 'guide/agents'), page('Review agents', 'guide/reviews'), page('Running knotless-serve', 'guide/server'), + page('Running on Kubernetes', 'guide/kubernetes-runner'), page('Workflows', 'guide/workflows'), page('Sites', 'guide/sites'), ], diff --git a/docs/design/runner.md b/docs/design/runner.md index 30a58f22..56adaaa5 100644 --- a/docs/design/runner.md +++ b/docs/design/runner.md @@ -364,8 +364,8 @@ a collaborator, or whose handle does not resolve. So is a batch holding it. Its log says so, naming the accounts. Trust decides when code runs, not what it is given. Every pull and batch -run gets no secrets, trusted or not, and only the docker executor runs -them: `--pulls` and `--queue` refuse `--executor shell`. +run gets no secrets, trusted or not, and only the docker and kubernetes +executors run them: `--pulls` and `--queue` refuse `--executor shell`. The shell executor runs each step as the runner's own user, who can read the secrets file and the runner's session. Watch a repository under it @@ -606,8 +606,8 @@ mirror at the commit, the log, and an `outputs/` directory, all removed afterwards. The clone hard-links objects rather than using alternates, so git works inside a container that sees only the checkout. -Under the `docker` executor, the command line's default (the library's -`Options::new` defaults to `shell`), each run gets one container from +Under the `docker` executor, chosen explicitly or where a host has no +tenant cluster, each run gets one container from the workflow's `image`, or `--image`. The checkout is mounted at `/work`. Each step is a `docker exec` of `sh -ec