path: root/src/runner/runc/run.rs

use std::{
	collections::HashMap,
	path::{Path, PathBuf},
	process,
};

use nix::mount::MsFlags;
use tee_readwrite::TeeWriter;

use crate::{
	paths, runner,
	types::*,
	util::{
		error::*,
		fs::{self, Mount},
		tar,
	},
};

use super::spec;

fn init_task(task: &runner::Task) -> Result<Mount> {
	let task_state_dir = paths::task_state_dir(&task.input_hash);
	let task_layer_dir = paths::join(&[&task_state_dir, paths::TASK_STATE_LAYER_SUBDIR]);
	let task_work_dir = paths::join(&[&task_state_dir, paths::TASK_STATE_WORK_SUBDIR]);

	fs::mkdir(&task_work_dir)?;
	fs::mkdir(&task_layer_dir)?;

	let task_tmp_dir = paths::task_tmp_dir(&task.input_hash);
	let mount_target = paths::join(&[&task_tmp_dir, paths::TASK_TMP_WORKDIR]);

	fs::mkdir(&mount_target)?;
	fs::mkdir(paths::join(&[
		&task_tmp_dir,
		paths::TASK_TMP_DEPENDS_SUBDIR,
	]))?;
	fs::mkdir(paths::join(&[&task_tmp_dir, paths::TASK_TMP_ROOTFS_SUBDIR]))?;

	let mount = if task.input.inherit.is_empty() {
		Mount::mount(
			task_layer_dir,
			&mount_target,
			"none",
			MsFlags::MS_BIND,
			None,
		)
		.with_context(|| format!("Failed to bind mount to {:?}", mount_target))?
	} else {
		let lower = task
			.input
			.inherit
			.iter()
			.rev()
			.map(|hash| {
				paths::join(&[&paths::task_state_dir(hash), paths::TASK_STATE_LAYER_SUBDIR])
			})
			.collect::<Vec<_>>()
			.join(":");
		let options = format!(
			"xino=off,index=off,metacopy=off,lowerdir={lower},upperdir={upper},workdir={work}",
			lower = lower,
			upper = task_layer_dir,
			work = task_work_dir
		);
		Mount::mount(
			"overlay",
			&mount_target,
			"overlay",
			MsFlags::empty(),
			Some(&options),
		)
		.with_context(|| format!("Failed to mount work overlay to {:?}", mount_target))?
	};

	Ok(mount)
}

fn init_task_rootfs(task: &runner::Task) -> Result<Mount> {
	let task_tmp_dir = paths::task_tmp_dir(&task.input_hash);
	let depends_dir = paths::join(&[&task_tmp_dir, paths::TASK_TMP_DEPENDS_SUBDIR]);
	let mount_target = paths::join(&[&task_tmp_dir, paths::TASK_TMP_ROOTFS_SUBDIR]);

	let lower = [&depends_dir, paths::ROOTFS_DIR].join(":");
	let options = format!(
		"xino=off,index=off,metacopy=off,lowerdir={lower}",
		lower = lower,
	);

	let mount = Mount::mount(
		"overlay",
		&mount_target,
		"overlay",
		MsFlags::empty(),
		Some(&options),
	)
	.with_context(|| format!("Failed to mount rootfs overlay to {:?}", mount_target))?;

	Ok(mount)
}

fn cleanup_task(task: &runner::Task) -> Result<()> {
	let task_tmp_dir = paths::task_tmp_dir(&task.input_hash);

	fs::ensure_removed(&task_tmp_dir).context("Failed to remove task tmp dir")?;

	let task_state_dir = paths::task_state_dir(&task.input_hash);
	let task_work_dir = paths::join(&[&task_state_dir, paths::TASK_STATE_WORK_SUBDIR]);
	fs::ensure_removed(&task_work_dir).context("Failed to remove overlayfs workdir")?;

	Ok(())
}

fn unpack_dependency<P1: AsRef<Path>, P2: AsRef<Path>>(filename: P1, dest: P2) -> Result<()> {
	let file = fs::open(filename.as_ref())?;
	tar::unpack(file, dest.as_ref()).with_context(|| {
		format!(
			"Failed to unpack {:?} to {:?}",
			filename.as_ref(),
			dest.as_ref()
		)
	})
}

fn unpack_dependencies(task: &runner::Task) -> Result<()> {
	let task_tmp_dir = paths::task_tmp_dir(&task.input_hash);
	let downloads_dir = paths::join(&[&task_tmp_dir, paths::TASK_TMP_DLDIR]);

	fs::mkdir(&downloads_dir)?;

	for dep in task.input.depends.values() {
		match dep {
			Dependency::Fetch { name, .. } => {
				fs::copy(
					paths::join(&[paths::DOWNLOADS_DIR, name]),
					paths::join(&[&downloads_dir, name]),
				)?;
			}
			Dependency::Task { output, path } => {
				unpack_dependency(
					paths::archive_filename(output),
					paths::join(&[&task_tmp_dir, paths::TASK_TMP_DEPENDS_SUBDIR, path]),
				)?;
			}
		}
	}

	Ok(())
}

fn collect_one_output(task: &runner::Task, path: &str) -> Result<Option<ArchiveHash>> {
	let source: PathBuf = [
		&paths::task_tmp_dir(&task.input_hash),
		paths::TASK_TMP_DESTDIR,
		path,
	]
	.iter()
	.collect();
	if !source.is_dir() {
		return Ok(None);
	}

	let filename = paths::archive_tmp_filename(&task.input_hash);

	let hash = (|| -> Result<ArchiveHash> {
		let file = fs::create(&filename)?;
		let hasher = ArchiveHasher::new();
		let writer = TeeWriter::new(file, hasher);

		let writer = tar::pack(writer, &source, ["."])?;
		let (file, hasher) = writer.into_inner();
		file.sync_all()?;
		drop(file);

		Ok(ArchiveHash(StringHash(hasher.finalize().into())))
	})()
	.with_context(|| format!("Failed to pack {:?} to {:?}", source, filename))?;

	fs::rename(filename, paths::archive_filename(&hash))?;

	Ok(Some(hash))
}

fn collect_output(task: &runner::Task) -> Result<HashMap<String, ArchiveHash>> {
	let mut ret = HashMap::new();

	for (name, path) in &task.input.output {
		if let Some(hash) = collect_one_output(task, path)? {
			ret.insert(name.clone(), hash);
		}
	}

	Ok(ret)
}

fn run_task(task: &runner::Task) -> Result<HashMap<String, ArchiveHash>> {
	let _workdir_mount = init_task(task).context("Failed to initialize task")?;
	unpack_dependencies(task).context("Failed to unpack dependencies")?;
	let _rootfs_mount = init_task_rootfs(task).context("Failed to initialize task rootfs")?;

	let task_tmp_dir = paths::task_tmp_dir(&task.input_hash);

	spec::generate_spec(task.input.command.as_str())
		.save(paths::join(&[&task_tmp_dir, "config.json"]))
		.map_err(Error::new)
		.context("Failed to save runtime config")?;

	{
		let log_filename = paths::task_log_filename(&task.input_hash);
		let log = fs::create(&log_filename)?;
		let log_stdout = log
			.try_clone()
			.context("Failed to duplicate log file descriptor")?;
		let log_stderr = log
			.try_clone()
			.context("Failed to duplicate log file descriptor")?;

		let status = process::Command::new("runc")
			.arg("--root")
			.arg(paths::TASK_TMP_RUNC_ROOT_SUBDIR)
			.arg("run")
			.arg(task.input_hash.to_string())
			.current_dir(task_tmp_dir)
			.stdin(process::Stdio::null())
			.stdout(log_stdout)
			.stderr(log_stderr)
			.status()
			.context("Failed to start container runtime")?;

		log.sync_all().context("Failed to write log output")?;

		if !status.success() {
			return Err(Error::new(format!(
				"Task failed: {}\nOutput: {}",
				status, log_filename
			)));
		}
	}

	collect_output(task).context("Failed to store output")
}

pub fn handle_task(task: runner::Task) -> Result<HashMap<String, ArchiveHash>> {
	println!("Starting task {} ({})", task.label, task.input_hash);

	let task_ret = run_task(&task);
	let cleanup_ret = cleanup_task(&task);

	let output = task_ret?;
	cleanup_ret.context("Failed to clean up after task")?;

	println!("Finished task {} ({})", task.label, task.input_hash);

	Ok(output)
}