diff --git a/Cargo.lock b/Cargo.lock index 1ad77cf..de537c3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -102,15 +102,6 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" -[[package]] -name = "arbitrary" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3bc62ac97cc33321f50863d514c3bc38a453947a8f9e781137e47c7401020aed" -dependencies = [ - "derive_arbitrary", -] - [[package]] name = "atomic-waker" version = "1.1.2" @@ -719,17 +710,6 @@ dependencies = [ "serde_core", ] -[[package]] -name = "derive_arbitrary" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b034bd7d5f032402a2479444dcc6f74e36a03f31854d41680fb240ef682a1ac" -dependencies = [ - "proc-macro2", - "quote", - "syn 3.0.3", -] - [[package]] name = "derive_builder" version = "0.20.2" @@ -2417,9 +2397,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" [[package]] name = "rustls-webpki" -version = "0.103.14" +version = "0.103.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0527518605e68109d875e248ea259b6758801cf165e4b2c2733ae3b51f12535a" +checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281" dependencies = [ "aws-lc-rs", "ring", @@ -3091,8 +3071,9 @@ checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" [[package]] name = "tracel-artifact" -version = "0.11.0" -source = "git+https://github.com/tracel-ai/tracel?rev=f52ee9e5c76153abf2ca750c31427d44dd86cc0b#f52ee9e5c76153abf2ca750c31427d44dd86cc0b" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5242b42fceb3cef1141e3094a7a26b24b9310d8ae51df19c3b42ce5f11b70fd6" dependencies = [ "crossbeam", "reqwest", @@ -3104,7 +3085,7 @@ dependencies = [ [[package]] name = "tracel-cli" -version = "0.6.0" +version = "0.7.0" dependencies = [ "anyhow", "cargo_metadata", @@ -3136,9 +3117,9 @@ dependencies = [ [[package]] name = "tracel-client" -version = "0.22.0" +version = "0.24.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "456275c8a4a1fd65e147aac5e39a34de32070ad8efe03929cb7cf6104fe2f407" +checksum = "ed83b0a8272286e92d8badb361d1241b95c5f49734298acdc95bfee198f19ab9" dependencies = [ "directories", "reqwest", @@ -3155,8 +3136,9 @@ dependencies = [ [[package]] name = "tracel-job" -version = "0.11.0" -source = "git+https://github.com/tracel-ai/tracel?rev=f52ee9e5c76153abf2ca750c31427d44dd86cc0b#f52ee9e5c76153abf2ca750c31427d44dd86cc0b" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d78edfde9f0a780eadca19ca21913519271d2b0e5a6673107c0e20cffab5c55" dependencies = [ "clap", "clap_complete", @@ -3267,6 +3249,12 @@ dependencies = [ "webpki-roots 0.26.11", ] +[[package]] +name = "typed-path" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ce898823d8cd0f028662d2f349694b1af0a5cc9be83f1d213a00d3147445e1e" + [[package]] name = "typeid" version = "1.0.3" @@ -3338,7 +3326,7 @@ dependencies = [ "serde_json", "ureq-proto", "utf8-zero", - "webpki-roots 1.0.9", + "webpki-roots 1.0.11", ] [[package]] @@ -3548,9 +3536,9 @@ dependencies = [ [[package]] name = "webpki-root-certs" -version = "1.0.9" +version = "1.0.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b96554aa2acc8ccdb7e1c9a58a7a68dd5d13bccc69cd124cb09406db612a1c9b" +checksum = "e49d0e65abc00c142e8f9428da991da9f1a250c39be97459901eb12b9b49f62a" dependencies = [ "rustls-pki-types", ] @@ -3561,14 +3549,14 @@ version = "0.26.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" dependencies = [ - "webpki-roots 1.0.9", + "webpki-roots 1.0.11", ] [[package]] name = "webpki-roots" -version = "1.0.9" +version = "1.0.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" +checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171" dependencies = [ "rustls-pki-types", ] @@ -3757,7 +3745,7 @@ checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" [[package]] name = "xtask" -version = "0.6.0" +version = "0.7.0" dependencies = [ "strum 0.27.2", "tracel-xtask", @@ -3883,15 +3871,18 @@ dependencies = [ [[package]] name = "zip" -version = "6.0.0" +version = "9.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb2a05c7c36fde6c09b08576c9f7fb4cda705990f73b58fe011abf7dfb24168b" +checksum = "1156b39d142a34b999c2310a50455e0cb3760bf7efad0122f796d813bee26ccb" dependencies = [ - "arbitrary", "crc32fast", "flate2", + "getrandom 0.4.3", "indexmap 2.14.0", "memchr", + "nix", + "typed-path", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 08f8dee..a40291f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,7 @@ members = ["crates/*", "xtask"] [workspace.package] edition = "2024" -version = "0.6.0" +version = "0.7.0" readme = "README.md" license = "MIT OR Apache-2.0" rust-version = "1.97.0" @@ -36,7 +36,7 @@ webbrowser = "1.2.4" url = "2.5.8" cargo_metadata = { version = "0.20.0", features = ["builder"] } walkdir = "2" -zip = { version = "6.0.0", default-features = false, features = ["deflate-flate2-zlib-rs"] } +zip = { version = "9.1.3", default-features = false, features = ["deflate-flate2-zlib-rs"] } ignore = "0.4.30" chrono = "0.4.45" strum = { version = "0.27.2", features = ["derive"] } @@ -48,12 +48,12 @@ rstest = { version = "0.27.0", default-features = false } ## External Crate -tracel-client = "0.22.0" +tracel-client = "0.24.0" tracel-artifact = "0.12.0" tracel-job = "0.12.0" ## Crate -tracel-cli = { version = "0.6.0", path = "crates/tracel-cli" } +tracel-cli = { version = "0.7.0", path = "crates/tracel-cli" } ### For xtask crate ### tracel-xtask = { version = "=5.0.0" } diff --git a/README.md b/README.md index 862d0e3..fa7893e 100644 --- a/README.md +++ b/README.md @@ -123,9 +123,8 @@ is recorded on the Tracel Console when the workspace is linked and you're logged in, and on this machine otherwise. Ctrl-C stops the job. With `--remote`, the workspace is packaged as `tracel package` does and the job -is queued on a compute provider group once you confirm, since a job may incur -costs. `--follow` then shows its logs until it finishes. Without a job, -`tracel run` is `cargo run --release`. +is queued on a compute provider group once you confirm. `--follow` then shows +its logs until it finishes. Without a job, `tracel run` is `cargo run --release`. ```bash tracel run --list diff --git a/crates/tracel-cli/src/commands/jobs.rs b/crates/tracel-cli/src/commands/jobs.rs index b2e96d5..267b442 100644 --- a/crates/tracel-cli/src/commands/jobs.rs +++ b/crates/tracel-cli/src/commands/jobs.rs @@ -1,11 +1,10 @@ -use std::io::{self, Write}; +use std::io; use std::time::{Duration, Instant}; use clap::{Args, Subcommand}; -use serde::Serialize; use tracel_client::ClientError; use tracel_client::console::job::response::{ - ExecutionContextResponse, JobLogResponse, JobResponse, MoneyResponse, ProjectJobsResponse, + ExecutionContextResponse, JobResponse, ProjectJobsResponse, }; use crate::commands::login::get_client_and_login_if_needed; @@ -13,7 +12,7 @@ use crate::context::CliContext; use crate::error::{CliError, ErrorKind}; use crate::helpers::{ FinalStatus, final_status, follow_logs, job_invocation, missing_job, poll_job, read_job, - require_completed, resolve_namespace_project, + require_completed, resolve_namespace_project, show_logs, }; use crate::tools::tracel_config::TracelProject; use crate::ui::{ @@ -67,10 +66,10 @@ pub struct LogsArgs { /// Job number #[arg(value_parser = clap::value_parser!(i32).range(1..))] pub num: i32, - /// Read from this byte of the logs - #[arg(long, value_name = "BYTE")] - pub start: Option, - /// Keep reading until it finishes + /// Show only its last N lines, or start following with them (10 by default) + #[arg(long, value_name = "N", value_parser = clap::value_parser!(u32).range(1..))] + pub tail: Option, + /// Keep showing new lines until it finishes #[arg(long)] pub follow: bool, } @@ -147,25 +146,6 @@ impl Render for JobResponse { .field("Command", command(&self.job_command)) .field("Code version", short_digest(&self.job_project_version)); let details = match &self.execution_context { - Some(ExecutionContextResponse::Managed { - compute_provider_name, - memory_gb, - cpu_cores, - num_gpus, - lock_price_per_hour, - final_cost, - estimated_cost, - }) => details - .field( - "Compute provider", - format!("{compute_provider_name} (managed)"), - ) - .field("CPU cores", cpu_cores) - .field("Memory", format!("{memory_gb} GB")) - .optional("GPUs", *num_gpus) - .field("Price per hour", money(lock_price_per_hour)) - .optional("Estimated cost", estimated_cost.as_ref().map(money)) - .optional("Final cost", final_cost.as_ref().map(money)), Some(ExecutionContextResponse::SelfManaged { compute_provider_name, }) => details.field( @@ -208,40 +188,6 @@ fn state(job: &JobResponse) -> String { } } -fn money(money: &MoneyResponse) -> String { - format!("{} {}", money.amount, money.currency) -} - -/// A page of job `num`'s logs, closed by the command that reads the next one when there -/// is more. -#[derive(Serialize)] -#[serde(transparent)] -struct LogPage { - #[serde(skip)] - num: i32, - page: JobLogResponse, -} - -impl Render for LogPage { - fn render(&self, out: &mut Human<'_>) -> io::Result<()> { - let logs = &self.page.logs; - out.write_all(logs.as_bytes())?; - if !logs.is_empty() && !logs.ends_with('\n') { - writeln!(out)?; - } - Ok(()) - } - - fn summary(&self) -> Option { - self.page.has_more.then(|| { - format!( - "More: `tracel jobs logs {} --start {}`.", - self.num, self.page.end - ) - }) - } -} - fn cancel_error(error: ClientError, project: &TracelProject, num: i32) -> anyhow::Error { if error.is_conflict() { return CliError::new( @@ -262,20 +208,13 @@ pub fn handle_command(args: JobsArgs, context: CliContext) -> anyhow::Result Ok(read_job(&client()?, &project, args.num)?.into()), JobsCommands::Logs(args) if args.follow => { - let start = args.start.unwrap_or(0); - let job = follow_logs(&client()?, context.output(), &project, args.num, start)?; + let job = follow_logs(&client()?, context.output(), &project, args.num, args.tail)?; context.terminal().outro(&state(&job)); Ok(Outcome::streamed()) } JobsCommands::Logs(args) => { - let page = client()? - .get_job_logs(&project.namespace, &project.name, args.num, args.start) - .map_err(|error| missing_job(error, &project, args.num))?; - Ok(LogPage { - num: args.num, - page, - } - .into()) + show_logs(&client()?, context.output(), &project, args.num, args.tail)?; + Ok(Outcome::streamed()) } JobsCommands::Cancel(args) => Ok(Outcome::data(cancel(args, &context, &project)?)), JobsCommands::Wait(args) => Ok(Outcome::data(wait(args, &context, &project)?)), @@ -375,7 +314,6 @@ mod tests { use super::*; use crate::error::ErrorReport; - use crate::test_support::plain; fn job(status: &str) -> JobResponse { JobResponse { @@ -396,19 +334,6 @@ mod tests { } } - fn log_page(logs: &str, has_more: bool) -> LogPage { - LogPage { - num: 12, - page: JobLogResponse { - logs: logs.into(), - start: 0, - end: logs.len() as u64, - total_size: 4096, - has_more, - }, - } - } - #[test] fn given_a_completed_job_when_waiting_for_it_then_the_wait_passes() { let waited = waited(&job("completed"), None); @@ -482,44 +407,14 @@ mod tests { #[test] fn given_a_job_that_has_not_started_when_saying_how_long_it_ran_then_there_is_nothing_to_say() { - let queued = JobResponse { + let not_started = JobResponse { started_at: None, completed_at: None, - ..job("queued") + ..job("new") }; - let ran = ran(&queued); + let ran = ran(¬_started); assert_eq!(ran, None); } - - #[test] - fn given_logs_that_end_mid_line_when_writing_a_page_then_the_line_is_ended() { - let page = log_page("epoch 1\nepoch 2", false); - - let text = plain(&page); - - assert_eq!(text, "epoch 1\nepoch 2\n"); - } - - #[test] - fn given_a_page_with_more_logs_when_summing_it_up_then_it_says_how_to_read_on() { - let page = log_page("epoch 1\n", true); - - let summary = page.summary().unwrap(); - - assert!( - summary.contains("tracel jobs logs 12 --start 8"), - "{summary}" - ); - } - - #[test] - fn given_the_last_page_when_summing_it_up_then_there_is_no_summary() { - let page = log_page("epoch 1\n", false); - - let summary = page.summary(); - - assert_eq!(summary, None); - } } diff --git a/crates/tracel-cli/src/commands/run/remote.rs b/crates/tracel-cli/src/commands/run/remote.rs index 798698e..6cac28a 100644 --- a/crates/tracel-cli/src/commands/run/remote.rs +++ b/crates/tracel-cli/src/commands/run/remote.rs @@ -83,7 +83,7 @@ pub fn check( } /// Package the workspace as `tracel package` does, then queue it as a job, once the user -/// agrees, since a job may incur costs. +/// agrees. pub fn queue(run: RemoteRun, context: &CliContext) -> anyhow::Result { let terminal = context.terminal(); let group = run.group; @@ -113,7 +113,7 @@ pub fn queue(run: RemoteRun, context: &CliContext) -> anyhow::Result { let digest = short_digest(&package.digest).to_string(); if !run.yes { let confirmed = terminal.confirm( - &format!("Queue it on {group} with code version {digest}? It may incur costs."), + &format!("Queue it on {group} with code version {digest}?"), false, queueing(group), )?; diff --git a/crates/tracel-cli/src/helpers/jobs.rs b/crates/tracel-cli/src/helpers/jobs.rs index 686ec98..fa862ba 100644 --- a/crates/tracel-cli/src/helpers/jobs.rs +++ b/crates/tracel-cli/src/helpers/jobs.rs @@ -3,13 +3,18 @@ use std::borrow::Cow; use std::io::{self, Write}; +use std::thread; use std::time::{Duration, Instant}; use serde::Serialize; use serde_json::Value; -use tracel_client::ClientError; use tracel_client::console::Client; -use tracel_client::console::job::response::{JobLogResponse, JobResponse}; +use tracel_client::console::job::request::JobLogCursor; +use tracel_client::console::job::response::{ + JobLogLine, JobLogResponse, JobLogStreamMessage, JobResponse, +}; +use tracel_client::websocket::WebSocketError; +use tracel_client::{ClientError, WebSocketClient}; use super::{Resource, map_resource_error}; use crate::error::{CliError, ErrorKind}; @@ -19,6 +24,16 @@ use crate::ui::{Human, Output, Render}; /// How often a followed job is read again once its logs are caught up. const FOLLOW_INTERVAL: Duration = Duration::from_secs(2); +/// How many of its last lines following a running job starts with, unless told, as +/// `tail -f` does. +const LINES_ON_ATTACH: u32 = 10; + +/// How many lines one read of a job's log asks for when paging through it. +const LOG_PAGE_LINES: u32 = 1_000; + +/// How long following waits to receive again when the log's stream has nothing new. +const STREAM_PAUSE: Duration = Duration::from_millis(100); + /// Write the job command that gives the program `arguments` unchanged. /// /// The compute provider splits the command into words with shell quoting rules (quotes, @@ -144,83 +159,251 @@ pub fn poll_job( } } -/// An event of following a job's logs. As JSON, a page of logs gains `"type": "log"`. +/// An event of following a job's logs. As JSON, a line gains `"type": "log"`. #[derive(Serialize)] #[serde(tag = "type", rename_all = "snake_case")] enum LogEvent<'a> { - Log(&'a JobLogResponse), - End { - status: &'a str, - /// Whether the logs so far end inside a line, which the text then ends. - #[serde(skip)] - mid_line: bool, - }, + Log(&'a JobLogLine), + End { status: &'a str }, } impl Render for LogEvent<'_> { fn render(&self, out: &mut Human<'_>) -> io::Result<()> { match self { - Self::Log(page) => out.write_all(page.logs.as_bytes()), - Self::End { mid_line: true, .. } => writeln!(out), + Self::Log(line) => writeln!(out, "{}", line.message), Self::End { .. } => Ok(()), } } } -/// Find where the next read of the logs starts after `page`, read from `start`, and -/// whether more is already there to read. -fn next_read(start: u64, page: &JobLogResponse) -> anyhow::Result<(u64, bool)> { - let next = page.end.max(start); - if page.has_more && next == start { - anyhow::bail!("The Tracel Console has more logs to send, but not from where."); +/// Where reading a job's log starts: with this many of its last lines, or with its first. +#[derive(Clone, Copy)] +enum LogStart { + Last(u32), + First, +} + +/// Where a reader of a job's log reads next: at its start until a line is read, then after +/// the newest line. +struct LogFollower { + newest: Option, + start: LogStart, +} + +impl LogFollower { + fn new(start: LogStart) -> Self { + Self { + newest: None, + start, + } + } + + /// A job numbers its lines from 0, so its first page is the lines before + /// [`LOG_PAGE_LINES`]. + fn read_at(&self) -> (JobLogCursor, Option) { + match (self.newest, self.start) { + (Some(seq), _) => (JobLogCursor::After(seq), Some(LOG_PAGE_LINES)), + (None, LogStart::Last(lines)) => (JobLogCursor::Last, Some(lines)), + (None, LogStart::First) => ( + JobLogCursor::Before(u64::from(LOG_PAGE_LINES)), + Some(LOG_PAGE_LINES), + ), + } + } + + /// Take `page`, read at [`read_at`](Self::read_at), and return whether newer lines may + /// already be waiting. A log still empty is read again from its first line. + fn take(&mut self, page: &JobLogResponse) -> bool { + let more = match (self.newest, self.start) { + (Some(_), _) => page.has_more, + (None, LogStart::First) => !page.lines.is_empty(), + (None, LogStart::Last(_)) => false, + }; + self.start = LogStart::First; + if let Some(line) = page.lines.last() { + self.newest = Some(line.seq); + } + more + } + + /// Take `lines`, a frame of the log's stream, and return the ones that continue the log + /// and whether one was skipped, so that the log has to be read at [`read_at`](Self::read_at). + fn take_live<'a>(&mut self, lines: &'a [JobLogLine]) -> (&'a [JobLogLine], bool) { + let Some(newest) = self.newest else { + return (&[], true); + }; + let shown_before = lines.iter().take_while(|line| line.seq <= newest).count(); + let lines = &lines[shown_before..]; + let continued = lines + .iter() + .zip(newest + 1..) + .take_while(|(line, seq)| line.seq == *seq) + .count(); + if let Some(line) = lines[..continued].last() { + self.newest = Some(line.seq); + } + (&lines[..continued], continued < lines.len()) } - Ok((next, page.has_more)) } -/// Write job `num`'s logs from byte `start` until the job has finished, and return it as -/// last read. The job is read before its logs, so the logs read after a final status are -/// complete. -pub fn follow_logs( - client: &Client, - output: &Output, - project: &TracelProject, +/// A job's log being shown: its reads and, while it is followed, the stream that shows its +/// new lines between them until the stream fails. +struct JobLog<'a> { + client: &'a Client, + output: &'a Output, + project: &'a TracelProject, num: i32, - start: u64, -) -> anyhow::Result { - let mut start = start; - let mut mid_line = false; - let job = poll_job(client, project, num, FOLLOW_INTERVAL, None, |_| { + follower: LogFollower, + stream: Option, +} + +impl<'a> JobLog<'a> { + fn new( + client: &'a Client, + output: &'a Output, + project: &'a TracelProject, + num: i32, + start: LogStart, + ) -> Self { + Self { + client, + output, + project, + num, + follower: LogFollower::new(start), + stream: None, + } + } + + /// Read the job, then show the lines after the newest one, and return the job. + fn read(&mut self) -> anyhow::Result { + let job = read_job(self.client, self.project, self.num)?; + self.read_new_lines()?; + Ok(job) + } + + fn read_new_lines(&mut self) -> anyhow::Result<()> { loop { - let page = client - .get_job_logs(&project.namespace, &project.name, num, Some(start)) - .map_err(|error| missing_job(error, project, num))?; - if !page.logs.is_empty() { - output.event(&LogEvent::Log(&page))?; - mid_line = !page.logs.ends_with('\n'); - } - let more; - (start, more) = next_read(start, &page)?; + let (cursor, limit) = self.follower.read_at(); + let page = self + .client + .get_job_logs( + &self.project.namespace, + &self.project.name, + self.num, + cursor, + limit, + ) + .map_err(|error| missing_job(error, self.project, self.num))?; + let more = self.follower.take(&page); + self.show(&page.lines)?; if !more { return Ok(()); } } - })?; + } + + /// Without a stream, following reads the log alone. + fn open_stream(&mut self) { + self.stream = self + .client + .open_job_log_websocket(&self.project.namespace, &self.project.name, self.num) + .ok(); + } + + /// Show the lines the stream sends until `next_read`, or wait for it without a stream. + fn receive_until(&mut self, next_read: Instant) -> anyhow::Result<()> { + loop { + let left = next_read.saturating_duration_since(Instant::now()); + if left.is_zero() { + return Ok(()); + } + let Some(stream) = self.stream.as_mut() else { + thread::sleep(left); + return Ok(()); + }; + match stream.receive::() { + Ok(Some(JobLogStreamMessage::Lines(lines))) => { + let (shown, skipped) = self.follower.take_live(&lines); + self.show(shown)?; + if skipped { + self.read_new_lines()?; + } + } + Ok(None) => thread::sleep(STREAM_PAUSE.min(left)), + Err(WebSocketError::SerializationError(_)) => {} + Err(_) => self.stream = None, + } + } + } + + fn show(&self, lines: &[JobLogLine]) -> anyhow::Result<()> { + for line in lines { + self.output.event(&LogEvent::Log(line))?; + } + Ok(()) + } +} + +/// Write job `num`'s log up to its newest line: its last `tail` lines, or every line. +pub fn show_logs( + client: &Client, + output: &Output, + project: &TracelProject, + num: i32, + tail: Option, +) -> anyhow::Result<()> { + let start = tail.map_or(LogStart::First, LogStart::Last); + JobLog::new(client, output, project, num, start).read_new_lines() +} + +/// Write job `num`'s log lines from `start` until the job has finished, and return it as +/// last read. The job is read before its logs, so the lines read after a final status are +/// complete; between reads, the log's stream shows lines as soon as they are stored. +fn follow( + client: &Client, + output: &Output, + project: &TracelProject, + num: i32, + start: LogStart, +) -> anyhow::Result { + let mut log = JobLog::new(client, output, project, num, start); + let mut job = log.read()?; + if final_status(&job.status).is_none() { + log.open_stream(); + } + while final_status(&job.status).is_none() { + log.receive_until(Instant::now() + FOLLOW_INTERVAL)?; + job = log.read()?; + } output.event(&LogEvent::End { status: &job.status, - mid_line, })?; Ok(job) } -/// Follow job `num`'s logs from the start until it finishes, and fail unless it -/// completed. +/// Write job `num`'s last `tail` lines, 10 unless told, then its new ones until it has +/// finished, and return it as last read. +pub fn follow_logs( + client: &Client, + output: &Output, + project: &TracelProject, + num: i32, + tail: Option, +) -> anyhow::Result { + let start = LogStart::Last(tail.unwrap_or(LINES_ON_ATTACH)); + follow(client, output, project, num, start) +} + +/// Follow job `num`, queued just before, with every line it writes until it finishes, and +/// fail unless it completed. pub fn follow_job( client: &Client, output: &Output, project: &TracelProject, num: i32, ) -> anyhow::Result { - let job = follow_logs(client, output, project, num, 0)?; + let job = follow(client, output, project, num, LogStart::First)?; require_completed(&job)?; Ok(job) } @@ -230,8 +413,9 @@ mod tests { use rstest::rstest; use serde_json::json; + use tracel_client::console::job::response::JobLogStream; + use super::*; - use crate::test_support::plain; fn command(arguments: &[&str]) -> String { let arguments: Vec = arguments @@ -241,16 +425,30 @@ mod tests { job_command(&arguments) } - fn page(end: u64, has_more: bool) -> JobLogResponse { + fn line(seq: u64) -> JobLogLine { + JobLogLine { + seq, + stream: JobLogStream::Stdout, + emitted_at: "2026-10-15T12:00:00.000Z".into(), + message: format!("epoch {seq}"), + } + } + + fn page(seqs: std::ops::Range, has_more: bool) -> JobLogResponse { JobLogResponse { - logs: "epoch 1\n".into(), - start: 0, - end, - total_size: 4096, + lines: seqs.map(line).collect(), has_more, } } + fn frame(numbers: &[u64]) -> Vec { + numbers.iter().copied().map(line).collect() + } + + fn seqs(lines: &[JobLogLine]) -> Vec { + lines.iter().map(|line| line.seq).collect() + } + #[test] fn given_plain_arguments_when_writing_a_job_command_then_they_are_joined_unchanged() { let written = command(&["train", "--epochs", "10", "--lr=0.01", "$HOME", "*.rs"]); @@ -316,35 +514,144 @@ mod tests { } #[test] - fn given_a_page_when_reading_on_then_the_next_read_starts_where_it_ends() { - let next = next_read(0, &page(8, true)).unwrap(); + fn given_no_line_read_yet_when_following_then_it_reads_the_last_lines_it_attaches_with() { + let follower = LogFollower::new(LogStart::Last(LINES_ON_ATTACH)); - assert_eq!(next, (8, true)); + assert_eq!( + follower.read_at(), + (JobLogCursor::Last, Some(LINES_ON_ATTACH)) + ); } #[test] - fn given_a_page_that_ends_before_the_start_when_reading_on_then_the_start_is_kept() { - let next = next_read(16, &page(8, false)).unwrap(); + fn given_lines_read_on_attach_when_following_then_it_reads_a_page_after_the_newest() { + let mut follower = LogFollower::new(LogStart::Last(LINES_ON_ATTACH)); + + let more = follower.take(&page(15..25, true)); - assert_eq!(next, (16, false)); + assert!(!more); + assert_eq!( + follower.read_at(), + (JobLogCursor::After(24), Some(LOG_PAGE_LINES)) + ); } #[test] - fn given_more_logs_but_no_progress_when_reading_on_then_it_fails() { - let next = next_read(8, &page(8, true)); + fn given_a_follower_that_attached_to_an_empty_log_when_reading_again_then_it_reads_from_the_first_line() + { + let mut follower = LogFollower::new(LogStart::Last(LINES_ON_ATTACH)); + + follower.take(&page(0..0, false)); + + assert_eq!( + follower.read_at(), + ( + JobLogCursor::Before(u64::from(LOG_PAGE_LINES)), + Some(LOG_PAGE_LINES) + ) + ); + } - assert!(next.is_err()); + #[test] + fn given_no_line_read_yet_when_reading_from_the_first_line_then_it_reads_the_lines_before_the_first_page_ends() + { + let follower = LogFollower::new(LogStart::First); + + assert_eq!( + follower.read_at(), + ( + JobLogCursor::Before(u64::from(LOG_PAGE_LINES)), + Some(LOG_PAGE_LINES) + ) + ); } #[test] - fn given_logs_that_end_mid_line_when_the_job_ends_then_the_line_is_ended() { - let end = LogEvent::End { - status: "completed", - mid_line: true, - }; + fn given_a_first_page_with_lines_when_reading_from_the_first_line_then_it_reads_on_after_the_newest() + { + let mut follower = LogFollower::new(LogStart::First); + + let more = follower.take(&page(0..1000, false)); + + assert!(more); + assert_eq!( + follower.read_at(), + (JobLogCursor::After(999), Some(LOG_PAGE_LINES)) + ); + } + + #[test] + fn given_an_empty_log_when_reading_from_the_first_line_then_it_stops() { + let mut follower = LogFollower::new(LogStart::First); + + let more = follower.take(&page(0..0, false)); + + assert!(!more); + } + + #[test] + fn given_more_lines_after_the_newest_when_following_then_it_reads_on_at_once() { + let mut follower = LogFollower::new(LogStart::Last(LINES_ON_ATTACH)); + follower.take(&page(0..3, false)); + + let more = follower.take(&page(3..1003, true)); + + assert!(more); + assert_eq!( + follower.read_at(), + (JobLogCursor::After(1002), Some(LOG_PAGE_LINES)) + ); + } + + #[test] + fn given_a_frame_continuing_the_log_when_following_then_it_shows_and_the_cursor_advances() { + let mut follower = LogFollower::new(LogStart::Last(LINES_ON_ATTACH)); + follower.take(&page(0..3, false)); + let received = frame(&[3, 4, 5]); + + let (shown, skipped) = follower.take_live(&received); + + assert_eq!(seqs(shown), [3, 4, 5]); + assert!(!skipped); + assert_eq!(follower.read_at().0, JobLogCursor::After(5)); + } + + #[test] + fn given_a_frame_with_lines_already_shown_when_following_then_only_the_new_ones_show() { + let mut follower = LogFollower::new(LogStart::Last(LINES_ON_ATTACH)); + follower.take(&page(0..5, false)); + let received = frame(&[3, 4, 5, 6]); + + let (shown, skipped) = follower.take_live(&received); + + assert_eq!(seqs(shown), [5, 6]); + assert!(!skipped); + assert_eq!(follower.read_at().0, JobLogCursor::After(6)); + } + + #[test] + fn given_a_frame_that_skips_a_line_when_following_then_it_shows_what_continued_and_reports_the_gap() + { + let mut follower = LogFollower::new(LogStart::Last(LINES_ON_ATTACH)); + follower.take(&page(0..3, false)); + let received = frame(&[3, 4, 6, 7]); + + let (shown, skipped) = follower.take_live(&received); + + assert_eq!(seqs(shown), [3, 4]); + assert!(skipped); + assert_eq!(follower.read_at().0, JobLogCursor::After(4)); + } + + #[test] + fn given_no_line_read_yet_when_a_frame_arrives_then_it_reports_a_gap() { + let mut follower = LogFollower::new(LogStart::Last(LINES_ON_ATTACH)); + follower.take(&page(0..0, false)); + let received = frame(&[0, 1]); - let text = plain(&end); + let (shown, skipped) = follower.take_live(&received); - assert_eq!(text, "\n"); + assert!(shown.is_empty()); + assert!(skipped); } } diff --git a/crates/tracel-cli/src/helpers/mod.rs b/crates/tracel-cli/src/helpers/mod.rs index 3ab5797..8af1704 100644 --- a/crates/tracel-cli/src/helpers/mod.rs +++ b/crates/tracel-cli/src/helpers/mod.rs @@ -8,7 +8,7 @@ mod transfer; pub use download::{DownloadFile, DownloadResult, download_files, validate_rel_path}; pub use jobs::{ FinalStatus, final_status, follow_job, follow_logs, job_command, job_invocation, missing_job, - poll_job, read_job, require_completed, + poll_job, read_job, require_completed, show_logs, }; pub use model::ensure_model_exists; pub use project::{ diff --git a/crates/tracel-cli/src/tools/packager.rs b/crates/tracel-cli/src/tools/packager.rs index 4a79cc7..fa4101d 100644 --- a/crates/tracel-cli/src/tools/packager.rs +++ b/crates/tracel-cli/src/tools/packager.rs @@ -456,7 +456,7 @@ mod tests { let archive = zip::ZipArchive::new(std::io::Cursor::new(bytes)).unwrap(); assert_eq!( - archive.file_names().collect::>(), + archive.file_names().collect::, _>>().unwrap(), ["Cargo.toml", "src/main.rs"] ); } diff --git a/docs/cli.md b/docs/cli.md index 0f138b2..634e585 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -49,8 +49,9 @@ and the instructions a login asks you to follow. `tracel experiments logs --follow --json` writes one line of JSON per event instead: each log entry, with `"type":"log"` added, then `{"type":"end","running":false}` when the experiment has finished. -`tracel jobs logs --follow --json` does the same with each page of logs, then -`{"type":"end","status":""}`, and `tracel run --remote --follow --json` +`tracel jobs logs --json` writes each line of the job's log the same way +(`seq`, `stream`, `emitted_at`, `message`), and with `--follow` ends with +`{"type":"end","status":""}`. `tracel run --remote --follow --json` starts with the queued job, with `"type":"queued"` added, then writes the job's log events. An error still writes the failure line. @@ -211,13 +212,13 @@ time left. ## Jobs -- A job's status is `new`, `queued`, `running`, `pending_cancellation`, - `completed`, `failed` or `cancelled`. -- `jobs logs ` reads one page of the job's log file, and shows it as it - is, then the command that reads the next page when there is more. - `--start ` reads from a byte offset, and a page's `end` is the next - page's `start`. `--follow` reads until the job has finished. -- `jobs cancel ` cancels a new, queued or running job, after asking; +- A job's status is `new`, `running`, `pending_cancellation`, `completed`, + `failed` or `cancelled`. +- `jobs logs ` shows the job's whole log, as `kubectl logs` does, and + `--tail ` only its last `n` lines. `--follow` starts with the last 10 + lines, or `--tail` of them, as `tail -f` does, then shows each new line until + the job has finished. +- `jobs cancel ` cancels a new or running job, after asking; `--yes` skips the question. A running job stays `pending_cancellation` until its compute provider stops it. - `jobs wait ` checks the job every `--interval` seconds until it @@ -270,14 +271,13 @@ time left. arguments after `--`. They reach the program unchanged: the job command single-quotes each argument that is empty, starts with `#`, or contains a space, tab, newline, quote or backslash. -- A job may incur costs, so the command asks before uploading anything, naming - the group and the code version, and `--yes` skips the question. Without a - terminal it needs `--yes`, and fails with `CONFIRMATION_REQUIRED` before - packaging. +- The command asks before uploading anything, naming the group and the code + version, and `--yes` skips the question. Without a terminal it needs `--yes`, + and fails with `CONFIRMATION_REQUIRED` before packaging. - In human output, the command ends on the commands that follow the job's logs and wait for it. - `--dry-run` packages and shows the job without logging in, uploading or queuing. -- `--follow` follows the job's logs, then exits as `jobs wait` does. Without a - job number from the Tracel Console, it fails after queuing, and the job stays - queued. +- `--follow` shows every line the job writes, then exits as `jobs wait` does. + Without a job number from the Tracel Console, it fails after queuing, and the + job stays queued.