diff --git a/CHANGELOG.md b/CHANGELOG.md index abf521d92..b161737fd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,15 @@ ### New features * Python API `crash_limit` now accepts `"unlimited"` and `"never-restart"` (and `0` is rejected) +* Python API now allows to enable stdio streaming + +### Changes + +* Relative `cwd`, `stdout`, `stderr` and `stream` paths are now rejected in the Python API and in job files. + Relative `cwd`, `stdout` and `stderr` paths are also rejected in array jobs submitted via the CLI. + Use an absolute path or a path starting with `%{SUBMIT_DIR}` (or `%{CWD}` for `stdout`/`stderr`). + A relative `--stream` path in the CLI is resolved against the submit directory. +* The stream path cannot contain task-specific placeholders (`%{TASK_ID}`, `%{INSTANCE_ID}`, `%{CWD}`). ## v0.27.0 diff --git a/crates/hyperqueue/src/client/commands/outputlog.rs b/crates/hyperqueue/src/client/commands/outputlog.rs index c31ff31b3..8e2a195fa 100644 --- a/crates/hyperqueue/src/client/commands/outputlog.rs +++ b/crates/hyperqueue/src/client/commands/outputlog.rs @@ -7,7 +7,7 @@ use tako::JobId; #[derive(Parser)] pub struct OutputLogOpts { - /// Path of the log file + /// Path of the directory with streamed output (see `hq submit --stream`) path: PathBuf, /// Filter files for the given server instance diff --git a/crates/hyperqueue/src/client/commands/submit/command.rs b/crates/hyperqueue/src/client/commands/submit/command.rs index 53050b170..dc1c5e61a 100644 --- a/crates/hyperqueue/src/client/commands/submit/command.rs +++ b/crates/hyperqueue/src/client/commands/submit/command.rs @@ -1,6 +1,7 @@ use super::directives::parse_hq_directives; use crate::client::commands::duration_doc; use crate::client::commands::submit::directives::parse_hq_directives_from_file; +use crate::client::commands::submit::paths::{validate_stream_path, validate_task_paths}; use crate::client::commands::wait::{wait_for_jobs, wait_for_jobs_with_progress}; use crate::client::globalsettings::GlobalSettings; use crate::client::resources::{ @@ -316,7 +317,7 @@ pub struct SubmitJobTaskConfOpts { )] time_limit: Option, - /// Stream the output of tasks into the given log file + /// Stream the output of tasks into the given directory #[arg(long)] stream: Option, @@ -667,6 +668,8 @@ pub async fn submit_computation( let task_count = ids.id_count(); check_suspicious_options(&opts, task_count)?; + let is_array = + opts.conf.array.is_some() || opts.conf.each_line.is_some() || opts.conf.from_json.is_some(); let JobSubmitOpts { commands, @@ -765,6 +768,14 @@ pub async fn submit_computation( job_id, }; + // Tasks of an array share the paths, so relative paths could not be told apart. + // For a single task, relative paths are resolved against the submit dir. + // The stream path is shared by all tasks and it is always resolved against the submit dir. + if is_array { + validate_task_paths(&request.submit_desc)?; + } + validate_stream_path(&request.submit_desc)?; + send_submit_request( gsettings, session, @@ -961,7 +972,7 @@ fn warn_unknown_placeholders(opts: &JobSubmitOpts) { opts.conf.stderr.as_ref().and_then(|arg| arg.path()), "stderr path", ); - check(opts.conf.stream.as_deref(), "log path"); + check(opts.conf.stream.as_deref(), "stream path"); check(opts.conf.cwd.as_deref(), "working directory path"); } diff --git a/crates/hyperqueue/src/client/commands/submit/jobfile.rs b/crates/hyperqueue/src/client/commands/submit/jobfile.rs index 38072c7f4..0913319f7 100644 --- a/crates/hyperqueue/src/client/commands/submit/jobfile.rs +++ b/crates/hyperqueue/src/client/commands/submit/jobfile.rs @@ -5,6 +5,7 @@ use crate::client::commands::submit::defs::{ ArrayDef, JobDef, StdioDefFull, StdioDefInput, TaskDef, }; use crate::client::commands::submit::defs::{PinMode as PinModeDef, TaskConfigDef}; +use crate::client::commands::submit::paths::validate_submit_paths; use crate::client::globalsettings::GlobalSettings; use crate::common::arraydef::IntArray; use crate::common::utils::fs::get_current_dir; @@ -238,6 +239,7 @@ pub async fn submit_computation_from_job_file( })?)? }; let request = build_job_submit(jdef, opts.job)?; + validate_submit_paths(&request.submit_desc)?; send_submit_request(gsettings, session, request, false, false, None).await } diff --git a/crates/hyperqueue/src/client/commands/submit/mod.rs b/crates/hyperqueue/src/client/commands/submit/mod.rs index 132d31442..bfa75a714 100644 --- a/crates/hyperqueue/src/client/commands/submit/mod.rs +++ b/crates/hyperqueue/src/client/commands/submit/mod.rs @@ -2,6 +2,7 @@ pub mod command; pub mod defs; pub mod directives; mod jobfile; +pub mod paths; pub use command::SubmitJobTaskConfOpts; pub use command::{JobSubmitOpts, submit_computation}; diff --git a/crates/hyperqueue/src/client/commands/submit/paths.rs b/crates/hyperqueue/src/client/commands/submit/paths.rs new file mode 100644 index 000000000..853a1ce9f --- /dev/null +++ b/crates/hyperqueue/src/client/commands/submit/paths.rs @@ -0,0 +1,194 @@ +use crate::common::placeholders::{ + CWD_PLACEHOLDER, INSTANCE_ID_PLACEHOLDER, SUBMIT_DIR_PLACEHOLDER, StringPart, + TASK_ID_PLACEHOLDER, parse_resolvable_string, +}; +use crate::transfer::messages::{ + JobSubmitDescription, JobTaskDescription, TaskDescription, TaskKind, +}; +use std::path::Path; +use tako::program::StdioDef; + +const ABSOLUTE_PLACEHOLDERS: [&str; 1] = [SUBMIT_DIR_PLACEHOLDER]; +const ABSOLUTE_OUTPUT_PLACEHOLDERS: [&str; 2] = [SUBMIT_DIR_PLACEHOLDER, CWD_PLACEHOLDER]; +/// The stream path is shared by all tasks of a job, so it cannot depend on a task. +const TASK_PLACEHOLDERS: [&str; 3] = [ + TASK_ID_PLACEHOLDER, + INSTANCE_ID_PLACEHOLDER, + CWD_PLACEHOLDER, +]; + +fn check_path(path: &Path, name: &str, placeholders: &[&str]) -> anyhow::Result<()> { + if path.is_absolute() { + return Ok(()); + } + let starts_with_absolute_placeholder = path.to_str().is_some_and(|path| { + matches!( + parse_resolvable_string(path).first(), + Some(StringPart::Placeholder(placeholder)) if placeholders.contains(placeholder) + ) + }); + if starts_with_absolute_placeholder { + return Ok(()); + } + anyhow::bail!( + "Path `{}` in `{name}` is not absolute. Use an absolute path or a path that starts with {}.", + path.display(), + placeholders + .iter() + .map(|placeholder| format!("`%{{{placeholder}}}`")) + .collect::>() + .join(" or ") + ) +} + +fn check_stdio(stdio: &StdioDef, name: &str) -> anyhow::Result<()> { + match stdio { + StdioDef::File { path, .. } => check_path(path, name, &ABSOLUTE_OUTPUT_PLACEHOLDERS), + StdioDef::Null | StdioDef::Pipe => Ok(()), + } +} + +fn check_task(desc: &TaskDescription) -> anyhow::Result<()> { + match &desc.kind { + TaskKind::ExternalProgram(program) => { + let program = &program.program; + check_path(&program.cwd, "cwd", &ABSOLUTE_PLACEHOLDERS)?; + check_stdio(&program.stdout, "stdout")?; + check_stdio(&program.stderr, "stderr") + } + } +} + +/// Checks that all task paths in the submit description are absolute or start with a placeholder. +/// The stream path is not checked. +pub fn validate_task_paths(desc: &JobSubmitDescription) -> anyhow::Result<()> { + match &desc.task_desc { + JobTaskDescription::Array { task_desc, .. } => check_task(task_desc), + JobTaskDescription::Graph { tasks, .. } => { + for task in tasks { + check_task(&task.task_desc)?; + } + Ok(()) + } + } +} + +fn check_stream_placeholders(path: &Path) -> anyhow::Result<()> { + let Some(path_str) = path.to_str() else { + return Ok(()); + }; + for part in parse_resolvable_string(path_str) { + if let StringPart::Placeholder(placeholder) = part + && TASK_PLACEHOLDERS.contains(&placeholder) + { + anyhow::bail!( + "Path `{}` in `stream` contains placeholder `%{{{placeholder}}}` that depends on a task. \ + The stream path is shared by all tasks of a job.", + path.display() + ); + } + } + Ok(()) +} + +/// Checks that the stream path does not contain task-specific placeholders. +/// A relative stream path is allowed; it is resolved against the submit dir. +pub fn validate_stream_path(desc: &JobSubmitDescription) -> anyhow::Result<()> { + if let Some(path) = &desc.stream_path { + check_stream_placeholders(path)?; + } + Ok(()) +} + +/// Checks that all paths in the submit description (including the stream path) +/// are absolute or start with a placeholder. +pub fn validate_submit_paths(desc: &JobSubmitDescription) -> anyhow::Result<()> { + validate_task_paths(desc)?; + validate_stream_path(desc)?; + if let Some(path) = &desc.stream_path { + check_path(path, "stream", &ABSOLUTE_PLACEHOLDERS)?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::{ + ABSOLUTE_OUTPUT_PLACEHOLDERS, ABSOLUTE_PLACEHOLDERS, check_path, check_stream_placeholders, + }; + use std::path::Path; + + fn ok(path: &str, placeholders: &[&str]) -> bool { + check_path(Path::new(path), "cwd", placeholders).is_ok() + } + + #[test] + fn accept_absolute_paths_and_absolute_placeholders() { + for path in [ + "/", + "/foo/bar", + "%{SUBMIT_DIR}", + "%{SUBMIT_DIR}/%{TASK_ID}/x", + ] { + assert!(ok(path, &ABSOLUTE_PLACEHOLDERS), "{path}"); + } + assert!(ok("%{CWD}/%{TASK_ID}.out", &ABSOLUTE_OUTPUT_PLACEHOLDERS)); + } + + #[test] + fn reject_relative_paths() { + for path in [ + "", + "foo", + "foo/bar", + "./foo", + "../foo", + "a/%{SUBMIT_DIR}", + "%foo", + ] { + assert!(!ok(path, &ABSOLUTE_PLACEHOLDERS), "{path}"); + } + } + + #[test] + fn reject_placeholders_that_do_not_produce_an_absolute_path() { + for path in [ + "%{TASK_ID}/x", + "%{JOB_ID}", + "%{INSTANCE_ID}/x", + "%{SERVER_UID}/x", + "%{UNKNOWN}/x", + ] { + assert!(!ok(path, &ABSOLUTE_OUTPUT_PLACEHOLDERS), "{path}"); + } + } + + #[test] + fn cwd_placeholder_only_in_output_paths() { + assert!(!ok("%{CWD}/x", &ABSOLUTE_PLACEHOLDERS)); + assert!(ok("%{CWD}/x", &ABSOLUTE_OUTPUT_PLACEHOLDERS)); + } + + #[test] + fn stream_placeholders() { + for path in [ + "foo", + "/foo", + "%{SUBMIT_DIR}/foo", + "foo-%{JOB_ID}", + "%{SERVER_UID}/x", + ] { + assert!(check_stream_placeholders(Path::new(path)).is_ok(), "{path}"); + } + for path in [ + "foo-%{TASK_ID}", + "%{SUBMIT_DIR}/%{INSTANCE_ID}", + "%{CWD}/foo", + ] { + assert!( + check_stream_placeholders(Path::new(path)).is_err(), + "{path}" + ); + } + } +} diff --git a/crates/pyhq/python/hyperqueue/ffi/protocol.py b/crates/pyhq/python/hyperqueue/ffi/protocol.py index 72ee58eae..e737b7689 100644 --- a/crates/pyhq/python/hyperqueue/ffi/protocol.py +++ b/crates/pyhq/python/hyperqueue/ffi/protocol.py @@ -53,3 +53,4 @@ class JobDescription: name: Optional[str] tasks: List[TaskDescription] max_fails: Optional[int] + stream_path: Optional[str] = None diff --git a/crates/pyhq/python/hyperqueue/job.py b/crates/pyhq/python/hyperqueue/job.py index 68a7ac261..d730c75ba 100644 --- a/crates/pyhq/python/hyperqueue/job.py +++ b/crates/pyhq/python/hyperqueue/job.py @@ -1,16 +1,19 @@ import dataclasses -from pathlib import Path -from typing import Dict, List, Optional, Sequence, Union +from typing import Any, Dict, List, Optional, Sequence, Union from .common import GenericPath from .ffi import JobId, TaskId from .ffi.protocol import CrashLimit, JobDescription, ResourceRequest -from .output import Stdio, default_stderr, default_stdout +from .output import Stdio, StdioDef, default_stderr, default_stdout from .task.function import PythonFunction from .task.program import ExternalProgram, ProgramArgs from .task.task import EnvType, Task +# Marks stdout/stderr that were not specified by the user +DEFAULT_STDIO: Any = object() + + class Job: """ Represents a HQ job. @@ -22,20 +25,27 @@ def __init__( max_fails: Optional[int] = 1, default_env: Optional[EnvType] = None, name: Optional[str] = None, + stream: Optional[GenericPath] = None, ): """ - :param default_workdir: Default working directory for tasks. + :param default_workdir: Default working directory for tasks. It has to be an absolute path or + start with `%{SUBMIT_DIR}`, because it is resolved outside this process. :param max_fails: How many tasks can fail before the whole job will be cancelled. :param default_env: Environment variables that will be automatically set for each task in this job. :param name: Name of the job. + :param stream: Path to a directory into which stdout/stderr of all tasks is streamed + (equivalent of `hq submit --stream`). It has to be an absolute path or start with + `%{SUBMIT_DIR}`. Tasks that do not specify + stdout/stderr will stream them; `stdout=None` discards the output. """ self.tasks: List[Task] = [] self.task_map: Dict[TaskId, Task] = {} self.max_fails = max_fails - self.default_workdir = Path(default_workdir).resolve() if default_workdir is not None else default_workdir + self.default_workdir = default_workdir self.default_env = default_env or {} self.name = name + self.stream = stream def task_by_id(self, id: TaskId) -> Optional[Task]: """ @@ -49,8 +59,8 @@ def program( *, env: Optional[EnvType] = None, cwd: Optional[GenericPath] = None, - stdout: Optional[Stdio] = default_stdout(), - stderr: Optional[Stdio] = default_stderr(), + stdout: Optional[Stdio] = DEFAULT_STDIO, + stderr: Optional[Stdio] = DEFAULT_STDIO, stdin: Optional[Union[str, bytes]] = None, deps: Sequence[Task] = (), name: Optional[str] = None, @@ -64,8 +74,10 @@ def program( :param args: List of arguments will be executed. The arguments have to be strings. :param env: Environment variables passed to the executed command. - :param cwd: Working directory of the executed command. - :param stdout: Path to a file that will store the standard output of the executed command. + :param cwd: Working directory of the executed command (an absolute path or a path starting + with `%{SUBMIT_DIR}`). + :param stdout: Path to a file that will store the standard output of the executed command + (an absolute path or a path starting with `%{SUBMIT_DIR}` or `%{CWD}`). :param stderr: Path to a file that will store the standard error output of the executed command. :param stdin: If provided, these bytes will be passed as the standard input of the executed @@ -86,8 +98,8 @@ def program( env=merge_envs(self.default_env, env), cwd=cwd or self.default_workdir, dependencies=deps, - stdout=stdout, - stderr=stderr, + stdout=self._resolve_stdio(stdout, default_stdout), + stderr=self._resolve_stdio(stderr, default_stderr), stdin=stdin, name=name, task_dir=task_dir, @@ -106,8 +118,8 @@ def function( kwargs=None, env: Optional[EnvType] = None, cwd: Optional[GenericPath] = None, - stdout: Optional[Stdio] = default_stdout(), - stderr: Optional[Stdio] = default_stderr(), + stdout: Optional[Stdio] = DEFAULT_STDIO, + stderr: Optional[Stdio] = DEFAULT_STDIO, deps: Sequence[Task] = (), name: Optional[str] = None, priority: int = 0, @@ -120,8 +132,10 @@ def function( :param args: Positional arguments that will be passed to the Python function. :param kwargs: Keyword arguments that will be passed to the Python function. :param env: Environment variables passed to the executed command. - :param cwd: Working directory of the executed command. - :param stdout: Path to a file that will store the standard output of the executed command. + :param cwd: Working directory of the executed command (an absolute path or a path starting + with `%{SUBMIT_DIR}`). + :param stdout: Path to a file that will store the standard output of the executed command + (an absolute path or a path starting with `%{SUBMIT_DIR}` or `%{CWD}`). :param stderr: Path to a file that will store the standard error output of the executed command. :param deps: A sequence of dependencies that have to be completed first before this task @@ -140,8 +154,8 @@ def function( kwargs=kwargs, env=merge_envs(self.default_env, env), cwd=cwd or self.default_workdir, - stdout=stdout, - stderr=stderr, + stdout=self._resolve_stdio(stdout, default_stdout), + stderr=self._resolve_stdio(stderr, default_stderr), name=name, dependencies=deps, priority=priority, @@ -151,6 +165,12 @@ def function( self._add_task(task) return task + def _resolve_stdio(self, stdio, default): + if stdio is not DEFAULT_STDIO: + return stdio + # Streamed jobs send output of tasks to the stream, unless a path is given explicitly + return StdioDef.to_stream() if self.stream is not None else default() + def _add_task(self, task: Task): self.tasks.append(task) self.task_map[task.task_id] = task @@ -159,7 +179,12 @@ def _build(self, client) -> JobDescription: task_descriptions = [] for task in self.tasks: task_descriptions.append(task._build(client)) - return JobDescription(name=self.name, tasks=task_descriptions, max_fails=self.max_fails) + return JobDescription( + name=self.name, + tasks=task_descriptions, + max_fails=self.max_fails, + stream_path=self.stream, + ) @dataclasses.dataclass diff --git a/crates/pyhq/python/hyperqueue/output.py b/crates/pyhq/python/hyperqueue/output.py index 60a174b6f..ebdb39506 100644 --- a/crates/pyhq/python/hyperqueue/output.py +++ b/crates/pyhq/python/hyperqueue/output.py @@ -16,11 +16,17 @@ class StdioDef: path: Optional[GenericPath] on_close: FileOnCloseBehavior + # Send the output into the job stream instead of a file (the job has to define `stream`) + pipe: bool = False @staticmethod def from_path(path: GenericPath) -> "StdioDef": return StdioDef(path=path, on_close="none") + @staticmethod + def to_stream() -> "StdioDef": + return StdioDef(path=None, on_close="none", pipe=True) + @staticmethod def remove_if_finished(path: Optional[GenericPath] = None) -> "StdioDef": return StdioDef(path=path, on_close="rm-if-finished") diff --git a/crates/pyhq/src/client/job.rs b/crates/pyhq/src/client/job.rs index 9468f232c..38d520253 100644 --- a/crates/pyhq/src/client/job.rs +++ b/crates/pyhq/src/client/job.rs @@ -4,6 +4,7 @@ use crate::{ClientContextPtr, FromPyObject, PyJobId, PyTaskId, borrow_mut, run_f use hyperqueue::client::commands::submit::command::{ DEFAULT_STDERR_PATH, DEFAULT_STDOUT_PATH, parse_crash_limit, }; +use hyperqueue::client::commands::submit::paths::validate_submit_paths; use hyperqueue::client::commands::submit::resource_rq_map_to_vec; use hyperqueue::client::output::resolve_task_paths; use hyperqueue::client::resources::parse_allocation_request; @@ -57,6 +58,7 @@ pub struct ResourceRequestDescription { pub struct StdioDefInput { path: Option, on_close: FromPy, + pipe: bool, } #[derive(Debug, FromPyObject)] @@ -80,6 +82,7 @@ pub struct PyJobDescription { name: Option, tasks: Vec, max_fails: Option, + stream_path: Option, } pub fn submit_job_impl(py: Python, ctx: ClientContextPtr, job: PyJobDescription) -> PyResult { @@ -93,17 +96,20 @@ pub fn submit_job_impl(py: Python, ctx: ClientContextPtr, job: PyJobDescription) resource_rqs: resource_rq_map_to_vec(resource_map), }; + let submit_desc = JobSubmitDescription { + task_desc, + submit_dir, + stream_path: job.stream_path, + }; + validate_submit_paths(&submit_desc)?; + let message = FromClientMessage::Submit( SubmitRequest { job_desc: JobDescription { name: job.name.unwrap_or_else(|| "PyAPI job".to_string()), max_fails: job.max_fails, }, - submit_desc: JobSubmitDescription { - task_desc, - submit_dir, - stream_path: None, - }, + submit_desc, job_id: None, }, None, @@ -225,6 +231,17 @@ fn build_task_resources( Ok(*resource_map.entry(rqv).or_insert(new_id)) } +fn stdio_def_from_input(stdio: StdioDefInput, default_path: &str) -> StdioDef { + if stdio.pipe { + StdioDef::Pipe + } else { + StdioDef::File { + path: stdio.path.unwrap_or_else(|| PathBuf::from(default_path)), + on_close: stdio.on_close.extract(), + } + } +} + fn build_task_desc(desc: TaskDescription, submit_dir: &Path) -> anyhow::Result { let args = desc.args.into_iter().map(|arg| arg.into()).collect(); let env = desc @@ -234,17 +251,11 @@ fn build_task_desc(desc: TaskDescription, submit_dir: &Path) -> anyhow::Result ... You can use [placeholders](#placeholders) in the working directory path. +### Relative paths + +The paths of the working directory (`--cwd`), `--stdout`, `--stderr` and `--stream` are resolved on the server and on +workers, not by the process that submits the job. A relative path of a single task is resolved against the directory +from which the job was submitted. A relative `--stream` path is always resolved against the directory from which the +job was submitted, also for task arrays. + +For [task arrays](arrays.md), relative `--cwd`, `--stdout` and `--stderr` paths are rejected with an error. For +[job files](jobfile.md) and the Python API, all relative paths (including the stream path) are rejected with an error. +Use an absolute path or a path that starts with a placeholder that expands to a directory, e.g. +`%{SUBMIT_DIR}/out/%{TASK_ID}.stdout`. These are `%{SUBMIT_DIR}` for all four paths and also `%{CWD}` for `--stdout` +and `--stderr`. Other placeholders, such as `%{TASK_ID}`, do not make a path absolute, so `%{TASK_ID}/out.txt` is still +a relative path. + ### Output By default, each job will produce two files containing the standard output and standard error output, respectively. The diff --git a/docs/jobs/streaming.md b/docs/jobs/streaming.md index c159a6904..c1cc5f3a5 100644 --- a/docs/jobs/streaming.md +++ b/docs/jobs/streaming.md @@ -34,6 +34,13 @@ The command above will cause the `stdout` and `stderr` of all `10_000` tasks to !!! tip You can use selected [placeholders](jobs.md#placeholders) inside the stream directory path. + The stream directory is shared by all tasks of a job, so task-specific placeholders (`%{TASK_ID}`, + `%{INSTANCE_ID}`, `%{CWD}`) are not allowed. + +!!! note + + A relative stream directory passed to `hq submit` is resolved against the directory from which the job was submitted. + For job files and the Python API, the stream directory has to be an absolute path or start with `%{SUBMIT_DIR}`. See [relative paths](jobs.md#relative-paths). ### Partial redirection @@ -46,7 +53,7 @@ For example: ```bash # Redirecting stdout into a file, streaming stderr into `my-log` -$ hq submit --stream=my-log --stdout="stdout-%{TASK_ID}" ... +$ hq submit --stream=my-log --stdout="%{SUBMIT_DIR}/stdout-%{TASK_ID}" ... # Streaming stdout into `my-log`, disabling stderr $ hq submit --stream=my-log --stderr=none ... diff --git a/tests/pyapi/test_cluster.py b/tests/pyapi/test_cluster.py index 09cc899a3..713bed529 100644 --- a/tests/pyapi/test_cluster.py +++ b/tests/pyapi/test_cluster.py @@ -1,3 +1,5 @@ +import os + import pytest from hyperqueue.cluster import LocalCluster, WorkerConfig from hyperqueue.job import Job @@ -33,7 +35,7 @@ def test_cluster_add_worker(): cluster.start_worker() client = cluster.client() job = Job() - job.program(["echo", "hello"], stdout="out.txt") + job.program(["echo", "hello"], stdout=os.path.abspath("out.txt")) job_id = client.submit(job) client.wait_for_jobs([job_id]) diff --git a/tests/pyapi/test_function.py b/tests/pyapi/test_function.py index 6a14c48ca..c2bf64054 100644 --- a/tests/pyapi/test_function.py +++ b/tests/pyapi/test_function.py @@ -19,7 +19,7 @@ def test_submit_pyfunction(hq_env: HqEnv): def body(a, b, c): print(f"stdout {a} {b} {c}") - job.function(body, args=(1, 2), kwargs={"c": 3}, stdout="out", stderr="err") + job.function(body, args=(1, 2), kwargs={"c": 3}, stdout=os.path.abspath("out"), stderr=os.path.abspath("err")) submitted_job = client.submit(job) wait_for_job_state(hq_env, submitted_job.id, "FINISHED") @@ -38,7 +38,7 @@ def body(): print(os.environ.get("ABC")) job = Job() - job.function(body, stdout="out") + job.function(body, stdout=os.path.abspath("out")) submitted_job = client.submit(job) wait_for_job_state(hq_env, submitted_job.id, "FINISHED") check_file_contents("out", "xyz\n") @@ -50,7 +50,7 @@ def test_submit_pyfunction_fail(hq_env: HqEnv): def body(): raise Exception("MyException") - job.function(body, stderr="err") + job.function(body, stderr=os.path.abspath("err")) job_id = client.submit(job) client.wait_for_jobs([job_id], raise_on_error=False) errors = client.get_failed_tasks(job_id) diff --git a/tests/pyapi/test_job.py b/tests/pyapi/test_job.py index 34ab56d27..6950f6ac1 100644 --- a/tests/pyapi/test_job.py +++ b/tests/pyapi/test_job.py @@ -66,8 +66,8 @@ def test_submit_stdio(hq_env: HqEnv): (job, client) = prepare_job_client(hq_env) job.program( args=bash("echo Test1; cat -; >&2 echo Test2"), - stdout="out", - stderr="err", + stdout=os.path.abspath("out"), + stderr=os.path.abspath("err"), stdin=b"Hello\n", ) submitted_job = client.submit(job) @@ -98,8 +98,8 @@ def test_stdio_rm_if_finished(hq_env: HqEnv): (job, client) = prepare_job_client(hq_env) job.program( args="uname", - stdout=StdioDef.remove_if_finished(path="out"), - stderr=StdioDef.remove_if_finished(path="err"), + stdout=StdioDef.remove_if_finished(path=os.path.abspath("out")), + stderr=StdioDef.remove_if_finished(path=os.path.abspath("err")), ) submitted_job = client.submit(job) wait_for_job_state(hq_env, submitted_job.id, "FINISHED") @@ -336,3 +336,60 @@ def test_job_name(hq_env: HqEnv): client.submit(job) data = hq_env.command(["job", "info", "last", "--output-mode", "json"], as_json=True) assert data[0]["info"]["name"] == "Foo" + + +def test_job_stream(hq_env: HqEnv): + os.mkdir("mylog") + (job, client) = prepare_job_client(hq_env, stream="%{SUBMIT_DIR}/mylog") + job.program(args=bash("echo Hello; >&2 echo Err")) + submitted_job = client.submit(job) + wait_for_job_state(hq_env, submitted_job.id, "FINISHED") + + assert hq_env.command(["output-log", "mylog", "cat", "1", "stdout"]) == "Hello\n" + assert hq_env.command(["output-log", "mylog", "cat", "1", "stderr"]) == "Err\n" + + +def test_job_stream_disable_stdout(hq_env: HqEnv): + os.mkdir("mylog") + (job, client) = prepare_job_client(hq_env, stream="%{SUBMIT_DIR}/mylog") + job.program(args=bash("echo Hello; >&2 echo Err"), stdout=None) + submitted_job = client.submit(job) + wait_for_job_state(hq_env, submitted_job.id, "FINISHED") + + assert hq_env.command(["output-log", "mylog", "cat", "1", "stdout"]) == "" + assert hq_env.command(["output-log", "mylog", "cat", "1", "stderr"]) == "Err\n" + + +@pytest.mark.parametrize("path", ["foo", "foo/bar", "./foo", "../foo"]) +def test_relative_paths_are_rejected(hq_env: HqEnv, path: str): + (_, client) = prepare_job_client(hq_env) + + def submit(job_args=None, **program_args): + job = Job(**(job_args or {})) + job.program(args="uname", **program_args) + with pytest.raises(Exception, match="is not absolute"): + client.submit(job) + + submit(cwd=path) + submit(stdout=path) + submit(stderr=path) + submit(stdout=StdioDef.remove_if_finished(path=path)) + submit(job_args=dict(default_workdir=path)) + submit(job_args=dict(stream=path)) + + +def test_placeholder_without_directory_is_rejected(hq_env: HqEnv): + (job, client) = prepare_job_client(hq_env) + job.program(args="uname", stdout="%{TASK_ID}/out") + with pytest.raises(Exception, match="is not absolute"): + client.submit(job) + job = Job() + job.program(args="uname", cwd="%{CWD}/foo") + with pytest.raises(Exception, match="is not absolute"): + client.submit(job) + + +def test_absolute_and_placeholder_paths_are_accepted(hq_env: HqEnv): + (job, client) = prepare_job_client(hq_env, default_workdir="/tmp", stream="%{SUBMIT_DIR}/log") + job.program(args="uname", cwd="%{SUBMIT_DIR}", stdout="/tmp/out", stderr="%{CWD}/err") + client.submit(job) diff --git a/tests/test_job.py b/tests/test_job.py index 8398632fe..94dae8912 100644 --- a/tests/test_job.py +++ b/tests/test_job.py @@ -1442,3 +1442,29 @@ def test_fast_retracting(hq_env: HqEnv): time.sleep(3) out = hq_env.command(["--output-mode=json", "job", "info", "1"], as_json=True) assert sorted(t["id"] for t in out[0]["tasks"] if t["state"] == "running") == list(range(1, 9)) + + +@pytest.mark.parametrize("array_args", [["--array=1-3"], ["--array=1"], ["--each-line=lines"]]) +@pytest.mark.parametrize("option", ["--stdout", "--stderr", "--cwd"]) +def test_array_relative_path_is_error(hq_env: HqEnv, array_args, option): + hq_env.start_server() + Path("lines").write_text("a\nb\n") + hq_env.command( + ["submit", *array_args, f"{option}=foo", "--", "true"], + expect_fail="is not absolute", + ) + # Placeholders that do not produce a directory do not make the path absolute + hq_env.command( + ["submit", *array_args, f"{option}=%{{TASK_ID}}/foo", "--", "true"], + expect_fail="is not absolute", + ) + hq_env.command(["submit", *array_args, f"{option}=%{{SUBMIT_DIR}}/foo-%{{TASK_ID}}", "--", "true"]) + hq_env.command(["submit", *array_args, f"{option}={os.path.abspath('foo')}-%{{TASK_ID}}", "--", "true"]) + + +def test_single_task_relative_path_is_resolved_on_server(hq_env: HqEnv): + hq_env.start_server() + hq_env.start_worker() + hq_env.command(["submit", "--stdout=out", "--stderr=err", "--", "bash", "-c", "echo Hello"]) + wait_for_job_state(hq_env, 1, "FINISHED") + check_file_contents("out", "Hello\n") diff --git a/tests/test_jobfile.py b/tests/test_jobfile.py index 01d21cc06..5ccbdd8e9 100644 --- a/tests/test_jobfile.py +++ b/tests/test_jobfile.py @@ -1,5 +1,8 @@ import collections import os +from pathlib import Path + +import pytest from .conftest import HqEnv from .utils import wait_for_job_state @@ -33,8 +36,8 @@ def test_job_file_submit_maximal(hq_env: HqEnv, tmp_path): [[task]] id = 12 -stdout = "testout-%{TASK_ID}" -stderr = "testerr-%{TASK_ID}" +stdout = "%{SUBMIT_DIR}/testout-%{TASK_ID}" +stderr = "%{SUBMIT_DIR}/testerr-%{TASK_ID}" task_dir = true time_limit = "1m 10s" priority = -1 @@ -48,7 +51,7 @@ def test_job_file_submit_maximal(hq_env: HqEnv, tmp_path): [[task]] id = 200 pin = "taskset" -cwd = "test-200" +cwd = "%{SUBMIT_DIR}/test-200" command = ["bash", "-c", "echo test1"] [[task.request]] resources = { "cpus" = "4 compact!", "gpus" = 2 } @@ -431,7 +434,7 @@ def test_job_file_stream(hq_env: HqEnv, tmp_path): hq_env.start_worker() tmp_path.joinpath("job.toml").write_text( """ -stream = "output" +stream = "%{SUBMIT_DIR}/output" [[task]] id = 1 @@ -485,3 +488,18 @@ def test_job_variant_weights(hq_env: HqEnv, tmp_path): "Resources", "# Variant 1\ncpus: 2 compact\n[weight: 2]\n# Variant 2\ncpus: 1 compact\ngpus: 2 compact\n[weight: 0.5]", ) + + +@pytest.mark.parametrize("config", ['stdout = "foo"', 'stderr = "foo"', 'cwd = "foo"', 'stream = "foo"']) +def test_job_file_relative_path_is_error(hq_env: HqEnv, config): + hq_env.start_server() + Path("job.toml").write_text( + f""" +{config if config.startswith("stream") else ""} +[[task]] +id = 1 +command = ["true"] +{config if not config.startswith("stream") else ""} +""" + ) + hq_env.command(["job", "submit-file", "job.toml"], expect_fail="is not absolute") diff --git a/tests/test_placeholders.py b/tests/test_placeholders.py index 113b18e5b..d3f67b6a4 100644 --- a/tests/test_placeholders.py +++ b/tests/test_placeholders.py @@ -188,7 +188,7 @@ def test_server_uid_placeholder(hq_env: HqEnv, tmp_path): @pytest.mark.parametrize("channel", ("stdout", "stderr")) def test_warning_missing_placeholder_in_output(hq_env: HqEnv, channel: str): hq_env.start_server() - output = hq_env.command(["submit", "--array=1-4", f"--{channel}=foo", "uname"]) + output = hq_env.command(["submit", "--array=1-4", f"--{channel}=%{{SUBMIT_DIR}}/foo", "uname"]) assert ( f"You have submitted an array job, but the `{channel}` " + "path does not contain the task ID placeholder." in output @@ -204,7 +204,7 @@ def test_missing_placeholder_in_output_present_in_cwd(hq_env: HqEnv, channel: st "submit", "--array=1-4", "--cwd", - "task-%{TASK_ID}", + "%{SUBMIT_DIR}/task-%{TASK_ID}", f"--{channel}", "%{CWD}/foo", "uname", @@ -230,7 +230,7 @@ def test_unknown_placeholder(hq_env: HqEnv): "uname", ] ) - assert "Found unknown placeholder `FOO` in log path" in output + assert "Found unknown placeholder `FOO` in stream path" in output assert "Found unknown placeholders `BAR, BAZ` in stdout path" in output assert "Found unknown placeholder `TAS_ID` in stderr path" in output assert "Found unknown placeholder `BAR` in working directory path" in output diff --git a/tests/test_stream.py b/tests/test_stream.py index b7697f52b..27aaf4a91 100644 --- a/tests/test_stream.py +++ b/tests/test_stream.py @@ -1,6 +1,8 @@ import time import os +import pytest + from .conftest import HqEnv from .utils import wait_for_job_state from .utils.io import check_file_contents @@ -16,7 +18,7 @@ def test_stream_collision(hq_env: HqEnv): [ "submit", "--stream", - "mylog", + "%{SUBMIT_DIR}/mylog", "--array=1-20", "--", "bash", @@ -39,7 +41,7 @@ def test_stream_submit(hq_env: HqEnv): [ "submit", "--stream", - "mylog", + "%{SUBMIT_DIR}/mylog", "--array=1-20", "--", "bash", @@ -88,7 +90,7 @@ def test_stream_overlap(hq_env: HqEnv): [ "submit", "--stream", - "mylog", + "%{SUBMIT_DIR}/mylog", "--array=1-2", "--", "bash", @@ -115,7 +117,7 @@ def test_stream_big_output(hq_env: HqEnv): [ "submit", "--stream", - "mylog", + "%{SUBMIT_DIR}/mylog", "--array=1-2", "--", "python3", @@ -363,3 +365,25 @@ def test_stream_timeout(hq_env: HqEnv): wait_for_job_state(hq_env, 1, "FAILED") wait_until(lambda: hq_env.command(["output-log", "mylog", "show"]) == "1.0:0> Start\n") + + +def test_stream_relative_path(hq_env: HqEnv): + hq_env.start_server() + hq_env.start_worker() + os.mkdir("sub") + hq_env.command( + ["submit", "--stream", "mylog", "--array=1-2", "--", "bash", "-c", "echo Hello${HQ_TASK_ID}"], + cwd="sub", + ) + wait_for_job_state(hq_env, 1, "FINISHED") + assert hq_env.command(["output-log", "sub/mylog", "cat", "1", "stdout"]) == "Hello1\nHello2\n" + + +@pytest.mark.parametrize("placeholder", ["TASK_ID", "INSTANCE_ID", "CWD"]) +@pytest.mark.parametrize("array_args", [[], ["--array=1-2"]]) +def test_stream_task_placeholder_is_error(hq_env: HqEnv, placeholder, array_args): + hq_env.start_server() + hq_env.command( + ["submit", *array_args, f"--stream=mylog-%{{{placeholder}}}", "--", "true"], + expect_fail="depends on a task", + )