diff --git a/README.md b/README.md index 9a6e29a..65afa2f 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ This repository currently provides: - `irtt-tui`, an optional terminal UI applet when built with the `tui` feature - `irtt-client`, a Rust library for running client sessions and consuming events - finite and continuous probe runs -- human, simple, machine-readable, and RTT-only stream output modes +- table, CSV, TSV, and JSON Lines event-row output formats with selectable columns - optional local summary statistics Server support is not implemented. @@ -76,13 +76,14 @@ Run continuously: irtt-cli --duration 0 ``` -Use a specific output mode: +Use a specific output format or column selection: ```sh -irtt-cli --output human -irtt-cli --output simple -irtt-cli --output machine -irtt-cli --output rtt-us +irtt-cli --format table +irtt-cli --format jsonl +irtt-cli --format csv --columns event,seq,remote,effective_rtt_us +irtt-cli --format tsv --columns effective_rtt_us --header never +irtt-cli --list-columns ``` With the optional TUI feature, `irtt-tui` opens a live cumulative dashboard for @@ -103,36 +104,71 @@ irtt-cli --help irtt-tui --help ``` -## Machine output +## Event row output -Use `--output machine` for line-oriented `key=value` output intended for scripts, monitoring, and autorate consumers. +The stream client separates event row rendering from final summaries. Event rows +are controlled by `--format`, `--columns`, and `--header`; final summaries remain +separate table-style text and are printed for table output when the run policy +allows a summary. + +Formats: + +- `table`: default, terminal-readable columns with a header +- `csv`: comma-separated rows, intended for scripts +- `tsv`: tab-separated rows, useful for one-value streams and shell pipelines +- `jsonl`: one JSON object per event row, suitable for streaming consumers + +Headers are controlled with `--header auto|always|never`. In `auto`, table, CSV, +and TSV print one header row; JSON Lines never prints a header. + +Columns are selected with `-c, --columns `, where `COLUMNS` is a +comma-separated list. Use `--list-columns` to print the available names. Useful +columns include: + +```text +event, seq, remote, token, rtt, rtt_us, raw_rtt_us, effective_rtt_us, +adjusted_rtt_us, rd, rd_us, sd, sd_us, ipdv, ipdv_us, proc, +server_processing_us, bytes, send_call_us, timer_error_us, highest_seen, +server_received, server_window, dscp, ecn, traffic_class, kernel_rx_ns, +warning_kind, message, event_wall_ns, client_send_wall_ns, +client_receive_wall_ns +``` + +Aliases are accepted for readability: `receive_delay` for `rd`, `send_delay` +for `sd`, `server_processing` for `proc`, `server_received_count` for +`server_received`, and `server_received_window` for `server_window`. + +The default table output favors readability and omits `echo_sent` rows. Custom +table column selections and machine-friendly formats include all event rows. +Missing table values render as `-`; missing CSV and TSV values render as empty +fields; missing JSONL values render as `null`. + +Use JSON Lines for structured event streaming: Example: ```sh -irtt-cli --duration 0 --interval 250ms --output machine +irtt-cli --duration 0 --interval 250ms --format jsonl ``` -Each line represents one client event and starts with an `event` field, for example: +Each line is one client event object, for example: -```text -event=echo_sent seq=4 remote=203.0.113.10:2112 client_send_wall_ns=1760000000000000000 bytes=64 send_call_us=10 timer_error_us=2 -event=echo_reply seq=4 remote=203.0.113.10:2112 client_send_wall_ns=1760000000000000000 client_receive_wall_ns=1760000000012400000 raw_rtt_us=12400 effective_rtt_us=12100 adjusted_rtt_us=12100 server_receive_wall_ns=1760000000006100000 server_receive_mono_ns=5000006100000 server_send_wall_ns=1760000000006400000 server_send_mono_ns=5000006400000 server_processing_us=300 client_to_server_us=6100 server_to_client_us=6000 server_received_count=5 server_received_window=0x1f dscp=0 ecn=0 kernel_rx_ns=none +```json +{"event":"echo_reply","seq":4,"remote":"203.0.113.10:2112","token":null,"rtt":"12.1ms","rtt_us":12100,"raw_rtt_us":12400,"effective_rtt_us":12100,"adjusted_rtt_us":12100,"rd":"6.0ms","rd_us":6000,"sd":"6.1ms","sd_us":6100,"ipdv":"300.0µs","ipdv_us":300,"proc":"300.0µs","server_processing_us":300,"bytes":64,"send_call_us":null,"timer_error_us":null,"highest_seen":null,"server_received":5,"server_window":"0x1f","dscp":0,"ecn":0,"traffic_class":0,"kernel_rx_ns":null,"warning_kind":null,"message":null,"event_wall_ns":1760000000012400000,"client_send_wall_ns":1760000000000000000,"client_receive_wall_ns":1760000000012400000} ``` -Consumers should match on `event=...` and read the fields they need. Unknown fields should be ignored. -RTT, send-call, and timer-error fields are reported in microseconds. +Consumers should match on the `event` property and read the columns they need. +RTT, send-call, and timer-error `*_us` fields are reported in microseconds. `raw_rtt_us` is the measured client send-to-receive RTT. `effective_rtt_us` and `adjusted_rtt_us` are signed and may be negative when server processing exceeds the raw RTT or timing correction produces a negative adjusted value. When -one-way delay fields are present, `client_to_server_us` and -`server_to_client_us` are also signed microseconds and may be negative due to -wall-clock skew. +one-way delay fields are present, `sd_us` and `rd_us` are also signed +microseconds and may be negative due to wall-clock skew. -For consumers that only need RTT values, use: +For consumers that only need effective RTT values in microseconds, use: ```sh -irtt-cli --output rtt-us +irtt-cli --format tsv --columns effective_rtt_us --header never ``` ## Library usage diff --git a/crates/irtt-cli/src/cmd/client/args.rs b/crates/irtt-cli/src/cmd/client/args.rs index 18016f9..f215158 100644 --- a/crates/irtt-cli/src/cmd/client/args.rs +++ b/crates/irtt-cli/src/cmd/client/args.rs @@ -11,7 +11,8 @@ pub const DEFAULT_CLIENT_DURATION: Duration = Duration::from_secs(10); #[command(name = "irtt-cli", about = "Minimal IRTT-compatible stream client")] pub struct ClientArgs { /// Server address or host, with optional port. - pub server: String, + #[arg(required_unless_present = "list_columns")] + pub server: Option, #[arg( long, @@ -25,18 +26,35 @@ pub struct ClientArgs { #[command(flatten)] pub common: CommonClientArgs, - /// Output format. - #[arg(long, value_enum, default_value_t = OutputMode::Human)] - pub output: OutputMode, + /// Event row renderer format. + #[arg(long, value_enum, default_value_t = OutputFormat::Table)] + pub format: OutputFormat, - /// Include extra fields in human output. + /// Comma-separated event row columns. + #[arg(short = 'c', long, value_name = "COLUMNS")] + pub columns: Option, + + /// Print available event row columns and exit. + #[arg(long)] + pub list_columns: bool, + + /// Header policy for table, CSV, and TSV output. + #[arg(long, value_enum, default_value_t = HeaderMode::Auto)] + pub header: HeaderMode, + + /// Include extra fields in table output and final summaries. #[arg(long)] pub verbose: bool, } impl ClientArgs { pub fn to_client_config(&self) -> ClientConfig { - self.common.to_client_config(&self.server, self.duration) + self.common.to_client_config( + self.server + .as_deref() + .expect("server is required unless --list-columns is set"), + self.duration, + ) } pub fn is_continuous(&self) -> bool { @@ -57,23 +75,33 @@ impl std::ops::Deref for ClientArgs { } #[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] -pub enum OutputMode { - /// Readable terminal output with a final summary. - Human, - /// Parseable full event fields. - Machine, - /// Simple key=value-ish event stream. - Simple, - /// RTT microseconds only. - RttUs, +pub enum OutputFormat { + /// Readable terminal table output. + Table, + /// Comma-separated event rows. + Csv, + /// Tab-separated event rows. + Tsv, + /// One JSON object per event row. + Jsonl, } -impl OutputMode { +impl OutputFormat { pub fn prints_summary(self) -> bool { - matches!(self, Self::Human) + matches!(self, Self::Table) } } +#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] +pub enum HeaderMode { + /// Print headers for table, CSV, and TSV output. + Auto, + /// Always print headers where the format supports them. + Always, + /// Never print headers. + Never, +} + #[cfg(test)] mod tests { use super::*; @@ -91,10 +119,12 @@ mod tests { fn client_parser_keeps_finite_default_duration() { let args = parse(&["127.0.0.1:2112"]).unwrap(); - assert_eq!(args.server, "127.0.0.1:2112"); + assert_eq!(args.server.as_deref(), Some("127.0.0.1:2112")); assert_eq!(args.duration, DEFAULT_CLIENT_DURATION); assert_eq!(args.interval, Duration::from_secs(1)); - assert_eq!(args.output, OutputMode::Human); + assert_eq!(args.format, OutputFormat::Table); + assert_eq!(args.columns, None); + assert_eq!(args.header, HeaderMode::Auto); assert!(!args.is_continuous()); assert_eq!(args.timestamp_mode(), TimestampArg::Both); @@ -107,17 +137,27 @@ mod tests { } #[test] - fn client_parser_accepts_stream_outputs_and_rejects_tui_output() { - for output in ["human", "simple", "machine", "rtt-us"] { - assert!(parse(&["--output", output, "127.0.0.1:2112"]).is_ok()); + fn client_parser_accepts_stream_formats_and_rejects_tui_output() { + for format in ["table", "csv", "tsv", "jsonl"] { + assert!(parse(&["--format", format, "127.0.0.1:2112"]).is_ok()); } - assert!(parse(&["--output", "tui", "127.0.0.1:2112"]).is_err()); + assert!(parse(&["--format", "tui", "127.0.0.1:2112"]).is_err()); + } + + #[test] + fn list_columns_does_not_require_server() { + let args = parse(&["--list-columns"]).unwrap(); + assert_eq!(args.server, None); + assert!(args.list_columns); } #[test] fn client_help_lists_shared_protocol_options() { let help = ClientArgs::command().render_help().to_string(); - assert!(help.contains("--output ")); + assert!(help.contains("--format ")); + assert!(help.contains("--columns ")); + assert!(help.contains("--list-columns")); + assert!(help.contains("--header
")); assert!(help.contains("--tstamp ")); assert!(help.contains("--stats ")); assert!(help.contains("--sfill ")); @@ -127,10 +167,10 @@ mod tests { } #[test] - fn output_mode_summary_policy_is_human_only() { - assert!(OutputMode::Human.prints_summary()); - assert!(!OutputMode::Simple.prints_summary()); - assert!(!OutputMode::Machine.prints_summary()); - assert!(!OutputMode::RttUs.prints_summary()); + fn output_format_summary_policy_is_table_only() { + assert!(OutputFormat::Table.prints_summary()); + assert!(!OutputFormat::Csv.prints_summary()); + assert!(!OutputFormat::Tsv.prints_summary()); + assert!(!OutputFormat::Jsonl.prints_summary()); } } diff --git a/crates/irtt-cli/src/cmd/client/mod.rs b/crates/irtt-cli/src/cmd/client/mod.rs index 33c7abf..3cf31f5 100644 --- a/crates/irtt-cli/src/cmd/client/mod.rs +++ b/crates/irtt-cli/src/cmd/client/mod.rs @@ -4,9 +4,5 @@ pub mod run; #[cfg(feature = "stats")] pub mod summary; -pub use args::{ClientArgs, OutputMode}; -pub use output::{ - format_event, format_human_event, format_human_event_with_options, HumanEventStats, - HumanIpdvPair, HumanOutputOptions, -}; +pub use args::{ClientArgs, HeaderMode, OutputFormat}; pub use run::run_stream; diff --git a/crates/irtt-cli/src/cmd/client/output.rs b/crates/irtt-cli/src/cmd/client/output.rs index bcd0a9f..a3ee717 100644 --- a/crates/irtt-cli/src/cmd/client/output.rs +++ b/crates/irtt-cli/src/cmd/client/output.rs @@ -9,51 +9,146 @@ use irtt_client::{ ServerTiming, SignedDuration, WarningKind, }; -use super::args::OutputMode; +use super::args::{HeaderMode, OutputFormat}; -pub fn format_event(event: &ClientEvent, mode: OutputMode) -> Option { - format_event_with_context(event, RenderContext::new(mode)) +#[derive(Debug, Clone)] +pub(super) struct OutputConfig { + format: OutputFormat, + columns: Vec, + header: HeaderMode, + default_table_rows: bool, + verbose: bool, } -pub(super) fn format_event_with_context( - event: &ClientEvent, - context: RenderContext<'_>, -) -> Option { - match context.mode { - OutputMode::Human => { - format_human_event_line(event, context.human_stats.cloned(), context.human_options) - } - OutputMode::Simple => format_simple_event(event), - OutputMode::Machine => format_machine_event(event), - OutputMode::RttUs => format_rtt_us_event(event), +impl OutputConfig { + pub(super) fn new( + format: OutputFormat, + columns: Option<&str>, + header: HeaderMode, + verbose: bool, + ) -> Result { + let default_table_rows = format == OutputFormat::Table && columns.is_none(); + let columns = match columns { + Some(columns) => parse_columns(columns, format, verbose)?, + None => default_columns(format, verbose), + }; + + Ok(Self { + format, + columns, + header, + default_table_rows, + verbose, + }) } -} -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(super) struct RenderContext<'a> { - pub(super) mode: OutputMode, - pub(super) human_stats: Option<&'a HumanEventStats>, - pub(super) human_options: HumanOutputOptions, -} + pub(super) fn prints_summary(&self) -> bool { + self.format.prints_summary() + } -impl RenderContext<'_> { - pub(super) fn new(mode: OutputMode) -> Self { - Self { - mode, - human_stats: None, - human_options: HumanOutputOptions::default(), + pub(super) fn summary_verbose(&self) -> bool { + self.verbose + } + + pub(super) fn should_print_header(&self) -> bool { + match self.header { + HeaderMode::Always => self.format != OutputFormat::Jsonl, + HeaderMode::Never => false, + HeaderMode::Auto => matches!( + self.format, + OutputFormat::Table | OutputFormat::Csv | OutputFormat::Tsv + ), } } + + pub(super) fn render_header(&self) -> Option { + if !self.should_print_header() { + return None; + } + + match self.format { + OutputFormat::Table => Some(render_table_header(&self.columns)), + OutputFormat::Csv => Some( + self.columns + .iter() + .map(|column| escape_csv(column.name())) + .collect::>() + .join(","), + ), + OutputFormat::Tsv => Some( + self.columns + .iter() + .map(|column| escape_tsv(column.name())) + .collect::>() + .join("\t"), + ), + OutputFormat::Jsonl => None, + } + } + + pub(super) fn render_event( + &self, + event: &ClientEvent, + stats: Option<&EventRenderStats>, + ) -> Option { + let row = OutputRow::from_event(event); + if self.default_table_rows && row.is_default_table_hidden() { + return None; + } + + let context = RenderContext { + stats, + verbose: self.verbose, + }; + + match self.format { + OutputFormat::Table => Some(render_table_row(&row, &self.columns, context)), + OutputFormat::Csv => Some(render_delimited_row( + &row, + &self.columns, + context, + DelimitedFormat::Csv, + )), + OutputFormat::Tsv => Some(render_delimited_row( + &row, + &self.columns, + context, + DelimitedFormat::Tsv, + )), + OutputFormat::Jsonl => Some(render_jsonl_row(&row, &self.columns, context)), + } + } + + pub(super) fn list_columns() -> String { + let mut out = String::new(); + writeln!(out, "Available event columns:").unwrap(); + for column in ALL_COLUMNS { + writeln!(out, " {:<24} {}", column.name(), column.description()).unwrap(); + } + writeln!(out).unwrap(); + writeln!( + out, + "Aliases: rd=receive_delay, sd=send_delay, proc=server_processing, \ + server_received=server_received_count, server_window=server_received_window." + ) + .unwrap(); + writeln!( + out, + "Use --columns default for the format default, or --columns all for every column." + ) + .unwrap(); + out + } } #[derive(Debug, Clone, PartialEq, Eq, Default)] -pub struct HumanEventStats { +pub struct EventRenderStats { pub contributed_sample: bool, - pub ipdv_pairs: Vec, + pub ipdv_pairs: Vec, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct HumanIpdvPair { +pub struct IpdvPair { pub previous_seq: u32, pub current_seq: u32, pub rtt_ipdv: Duration, @@ -61,13 +156,8 @@ pub struct HumanIpdvPair { pub receive_ipdv: Option, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] -pub struct HumanOutputOptions { - pub verbose: bool, -} - #[cfg(feature = "stats")] -impl From for HumanEventStats { +impl From for EventRenderStats { fn from(value: irtt_stats::EventStatsUpdate) -> Self { Self { contributed_sample: value.contributed_sample, @@ -77,7 +167,7 @@ impl From for HumanEventStats { } #[cfg(feature = "stats")] -impl From for HumanIpdvPair { +impl From for IpdvPair { fn from(value: irtt_stats::IpdvPairUpdate) -> Self { Self { previous_seq: value.previous_seq, @@ -89,408 +179,930 @@ impl From for HumanIpdvPair { } } -pub fn format_human_event(event: &ClientEvent, stats: Option) -> String { - format_human_event_with_options(event, stats, HumanOutputOptions::default()) +#[derive(Debug, Clone, Copy)] +struct RenderContext<'a> { + stats: Option<&'a EventRenderStats>, + verbose: bool, } -pub fn format_human_event_with_options( - event: &ClientEvent, - stats: Option, - options: HumanOutputOptions, -) -> String { - format_human_event_line(event, stats, options).unwrap_or_default() +#[derive(Debug, Clone, PartialEq, Eq)] +enum OutputRow { + SessionStarted(LifecycleRow), + NoTestCompleted(LifecycleRow), + SessionClosed { + remote: SocketAddr, + token: u64, + event_wall: SystemTime, + }, + EchoSent { + seq: u32, + remote: SocketAddr, + client_send_wall: SystemTime, + bytes: usize, + send_call: Duration, + timer_error: Duration, + }, + EchoReply(ReplyRow), + Loss { + seq: u32, + client_send_wall: SystemTime, + }, + Duplicate { + seq: u32, + remote: SocketAddr, + client_receive_wall: SystemTime, + bytes: usize, + }, + Late { + reply: ReplyRow, + highest_seen: u32, + }, + Warning { + kind: WarningKind, + message: String, + event_wall: SystemTime, + }, } -fn format_human_event_line( - event: &ClientEvent, - stats: Option, - options: HumanOutputOptions, -) -> Option { - let stats = stats.as_ref(); +impl OutputRow { + fn from_event(event: &ClientEvent) -> Self { + match event { + ClientEvent::SessionStarted { + remote, + token, + negotiated, + at, + } => Self::SessionStarted(LifecycleRow::new( + *remote, + Some(*token), + negotiated, + at.wall, + )), + ClientEvent::NoTestCompleted { + remote, + negotiated, + at, + } => Self::NoTestCompleted(LifecycleRow::new(*remote, None, negotiated, at.wall)), + ClientEvent::SessionClosed { remote, token, at } => Self::SessionClosed { + remote: *remote, + token: *token, + event_wall: at.wall, + }, + ClientEvent::EchoSent { + seq, + remote, + sent_at, + bytes, + send_call, + timer_error, + .. + } => Self::EchoSent { + seq: *seq, + remote: *remote, + client_send_wall: sent_at.wall, + bytes: *bytes, + send_call: *send_call, + timer_error: *timer_error, + }, + ClientEvent::EchoReply { + seq, + remote, + sent_at, + received_at, + rtt, + server_timing, + one_way, + received_stats, + bytes, + packet_meta, + } => Self::EchoReply(ReplyRow { + seq: *seq, + remote: *remote, + client_send_wall: Some(sent_at.wall), + client_receive_wall: received_at.wall, + rtt: Some(*rtt), + server_timing: *server_timing, + one_way: *one_way, + received_stats: *received_stats, + bytes: *bytes, + packet_meta: *packet_meta, + }), + ClientEvent::EchoLoss { seq, sent_at, .. } => Self::Loss { + seq: *seq, + client_send_wall: sent_at.wall, + }, + ClientEvent::DuplicateReply { + seq, + remote, + received_at, + bytes, + } => Self::Duplicate { + seq: *seq, + remote: *remote, + client_receive_wall: received_at.wall, + bytes: *bytes, + }, + ClientEvent::LateReply { + seq, + highest_seen, + remote, + sent_at, + received_at, + rtt, + server_timing, + one_way, + received_stats, + bytes, + packet_meta, + } => Self::Late { + reply: ReplyRow { + seq: *seq, + remote: *remote, + client_send_wall: sent_at.map(|sent_at| sent_at.wall), + client_receive_wall: received_at.wall, + rtt: *rtt, + server_timing: *server_timing, + one_way: *one_way, + received_stats: *received_stats, + bytes: *bytes, + packet_meta: *packet_meta, + }, + highest_seen: *highest_seen, + }, + ClientEvent::Warning { kind, message, at } => Self::Warning { + kind: *kind, + message: message.clone(), + event_wall: at.wall, + }, + } + } - let line = match event { - ClientEvent::SessionStarted { remote, token, .. } => { - format!("session started remote={remote} token={token:#x}") + fn event_name(&self) -> &'static str { + match self { + Self::SessionStarted(_) => "session_started", + Self::NoTestCompleted(_) => "no_test_completed", + Self::SessionClosed { .. } => "session_closed", + Self::EchoSent { .. } => "echo_sent", + Self::EchoReply(_) => "echo_reply", + Self::Loss { .. } => "loss", + Self::Duplicate { .. } => "duplicate", + Self::Late { .. } => "late", + Self::Warning { .. } => "warning", } - ClientEvent::NoTestCompleted { remote, .. } => { - format!("no-test completed remote={remote}") + } + + fn is_default_table_hidden(&self) -> bool { + matches!(self, Self::EchoSent { .. }) + } + + fn reply(&self) -> Option<&ReplyRow> { + match self { + Self::EchoReply(reply) | Self::Late { reply, .. } => Some(reply), + _ => None, } - ClientEvent::SessionClosed { remote, token, .. } => { - format!("session closed remote={remote} token={token:#x}") - } - ClientEvent::EchoSent { .. } => return None, - ClientEvent::EchoReply { - seq, - rtt, - server_timing, - one_way, - received_stats, - .. - } => { - let mut out = format!("seq={seq}"); - write!(out, " rtt={}", format_signed_duration(rtt.effective)).unwrap(); - write_human_one_way(&mut out, *one_way); - write!(out, " ipdv={}", format_human_ipdv(stats, *seq)).unwrap(); - if let Some(processing) = server_timing.and_then(|timing| timing.processing) { - write!(out, " proc={}", format_duration(processing)).unwrap(); - } - if options.verbose { - write_human_received_stats(&mut out, *received_stats); - } - out - } - ClientEvent::EchoLoss { seq, .. } => { - format!("loss seq={seq}") - } - ClientEvent::DuplicateReply { seq, remote, .. } => { - format!("duplicate seq={seq} remote={remote}") - } - ClientEvent::LateReply { - seq, - highest_seen, + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct LifecycleRow { + remote: SocketAddr, + token: Option, + event_wall: SystemTime, + duration_ns: i128, + interval_ns: i128, + payload_length: i128, +} + +impl LifecycleRow { + fn new( + remote: SocketAddr, + token: Option, + negotiated: &NegotiatedParams, + event_wall: SystemTime, + ) -> Self { + Self { remote, - rtt, - one_way, - received_stats, - .. - } => { - let mut out = format!("late seq={seq} highest_seen={highest_seen} remote={remote}",); - if let Some(rtt) = rtt { - write!(out, " rtt={}", format_signed_duration(rtt.effective)).unwrap(); - write_human_one_way(&mut out, *one_way); - write!(out, " ipdv={}", format_human_ipdv(stats, *seq)).unwrap(); - } - write_human_received_stats(&mut out, *received_stats); - out + token, + event_wall, + duration_ns: i128::from(negotiated.params.duration_ns), + interval_ns: i128::from(negotiated.params.interval_ns), + payload_length: i128::from(negotiated.params.length), } - ClientEvent::Warning { kind, message, .. } => { - format!("warning kind={} message={message}", warning_kind(*kind)) - } - }; - Some(line) + } } -fn format_rtt_us_event(event: &ClientEvent) -> Option { - match event { - ClientEvent::EchoReply { rtt, .. } => Some(signed_duration_us(rtt.effective).to_string()), +#[derive(Debug, Clone, PartialEq, Eq)] +struct ReplyRow { + seq: u32, + remote: SocketAddr, + client_send_wall: Option, + client_receive_wall: SystemTime, + rtt: Option, + server_timing: Option, + one_way: Option, + received_stats: Option, + bytes: usize, + packet_meta: PacketMeta, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Column { + Event, + Seq, + Remote, + Token, + Rtt, + RttUs, + RawRttUs, + EffectiveRttUs, + AdjustedRttUs, + ReceiveDelay, + ReceiveDelayUs, + SendDelay, + SendDelayUs, + Ipdv, + IpdvUs, + ServerProcessing, + ServerProcessingUs, + Bytes, + SendCallUs, + TimerErrorUs, + HighestSeen, + ServerReceivedCount, + ServerReceivedWindow, + Dscp, + Ecn, + TrafficClass, + KernelRxNs, + WarningKind, + Message, + EventWallNs, + ClientSendWallNs, + ClientReceiveWallNs, + DurationNs, + IntervalNs, + PayloadLength, + ServerReceiveWallNs, + ServerReceiveMonoNs, + ServerSendWallNs, + ServerSendMonoNs, + ServerMidpointWallNs, + ServerMidpointMonoNs, +} + +const ALL_COLUMNS: &[Column] = &[ + Column::Event, + Column::Seq, + Column::Remote, + Column::Token, + Column::Rtt, + Column::RttUs, + Column::RawRttUs, + Column::EffectiveRttUs, + Column::AdjustedRttUs, + Column::ReceiveDelay, + Column::ReceiveDelayUs, + Column::SendDelay, + Column::SendDelayUs, + Column::Ipdv, + Column::IpdvUs, + Column::ServerProcessing, + Column::ServerProcessingUs, + Column::Bytes, + Column::SendCallUs, + Column::TimerErrorUs, + Column::HighestSeen, + Column::ServerReceivedCount, + Column::ServerReceivedWindow, + Column::Dscp, + Column::Ecn, + Column::TrafficClass, + Column::KernelRxNs, + Column::WarningKind, + Column::Message, + Column::EventWallNs, + Column::ClientSendWallNs, + Column::ClientReceiveWallNs, + Column::DurationNs, + Column::IntervalNs, + Column::PayloadLength, + Column::ServerReceiveWallNs, + Column::ServerReceiveMonoNs, + Column::ServerSendWallNs, + Column::ServerSendMonoNs, + Column::ServerMidpointWallNs, + Column::ServerMidpointMonoNs, +]; + +impl Column { + fn name(self) -> &'static str { + match self { + Self::Event => "event", + Self::Seq => "seq", + Self::Remote => "remote", + Self::Token => "token", + Self::Rtt => "rtt", + Self::RttUs => "rtt_us", + Self::RawRttUs => "raw_rtt_us", + Self::EffectiveRttUs => "effective_rtt_us", + Self::AdjustedRttUs => "adjusted_rtt_us", + Self::ReceiveDelay => "rd", + Self::ReceiveDelayUs => "rd_us", + Self::SendDelay => "sd", + Self::SendDelayUs => "sd_us", + Self::Ipdv => "ipdv", + Self::IpdvUs => "ipdv_us", + Self::ServerProcessing => "proc", + Self::ServerProcessingUs => "server_processing_us", + Self::Bytes => "bytes", + Self::SendCallUs => "send_call_us", + Self::TimerErrorUs => "timer_error_us", + Self::HighestSeen => "highest_seen", + Self::ServerReceivedCount => "server_received", + Self::ServerReceivedWindow => "server_window", + Self::Dscp => "dscp", + Self::Ecn => "ecn", + Self::TrafficClass => "traffic_class", + Self::KernelRxNs => "kernel_rx_ns", + Self::WarningKind => "warning_kind", + Self::Message => "message", + Self::EventWallNs => "event_wall_ns", + Self::ClientSendWallNs => "client_send_wall_ns", + Self::ClientReceiveWallNs => "client_receive_wall_ns", + Self::DurationNs => "duration_ns", + Self::IntervalNs => "interval_ns", + Self::PayloadLength => "payload_length", + Self::ServerReceiveWallNs => "server_receive_wall_ns", + Self::ServerReceiveMonoNs => "server_receive_mono_ns", + Self::ServerSendWallNs => "server_send_wall_ns", + Self::ServerSendMonoNs => "server_send_mono_ns", + Self::ServerMidpointWallNs => "server_midpoint_wall_ns", + Self::ServerMidpointMonoNs => "server_midpoint_mono_ns", + } + } + + fn parse(input: &str) -> Option { + Some(match input { + "event" => Self::Event, + "seq" => Self::Seq, + "remote" => Self::Remote, + "token" => Self::Token, + "rtt" => Self::Rtt, + "rtt_us" => Self::RttUs, + "raw_rtt_us" => Self::RawRttUs, + "effective_rtt_us" => Self::EffectiveRttUs, + "adjusted_rtt_us" => Self::AdjustedRttUs, + "rd" | "receive_delay" => Self::ReceiveDelay, + "rd_us" | "receive_delay_us" => Self::ReceiveDelayUs, + "sd" | "send_delay" => Self::SendDelay, + "sd_us" | "send_delay_us" => Self::SendDelayUs, + "ipdv" => Self::Ipdv, + "ipdv_us" => Self::IpdvUs, + "proc" | "server_processing" => Self::ServerProcessing, + "server_processing_us" => Self::ServerProcessingUs, + "bytes" => Self::Bytes, + "send_call_us" => Self::SendCallUs, + "timer_error_us" => Self::TimerErrorUs, + "highest_seen" => Self::HighestSeen, + "server_received" | "server_received_count" => Self::ServerReceivedCount, + "server_window" | "server_received_window" => Self::ServerReceivedWindow, + "dscp" => Self::Dscp, + "ecn" => Self::Ecn, + "traffic_class" => Self::TrafficClass, + "kernel_rx_ns" => Self::KernelRxNs, + "warning_kind" => Self::WarningKind, + "message" => Self::Message, + "event_wall_ns" => Self::EventWallNs, + "client_send_wall_ns" => Self::ClientSendWallNs, + "client_receive_wall_ns" => Self::ClientReceiveWallNs, + "duration_ns" => Self::DurationNs, + "interval_ns" => Self::IntervalNs, + "payload_length" => Self::PayloadLength, + "server_receive_wall_ns" => Self::ServerReceiveWallNs, + "server_receive_mono_ns" => Self::ServerReceiveMonoNs, + "server_send_wall_ns" => Self::ServerSendWallNs, + "server_send_mono_ns" => Self::ServerSendMonoNs, + "server_midpoint_wall_ns" => Self::ServerMidpointWallNs, + "server_midpoint_mono_ns" => Self::ServerMidpointMonoNs, + _ => return None, + }) + } + + fn description(self) -> &'static str { + match self { + Self::Event => "event kind", + Self::Seq => "probe sequence number", + Self::Remote => "remote socket address", + Self::Token => "session token as hexadecimal", + Self::Rtt => "human-readable effective RTT", + Self::RttUs => "effective RTT in signed microseconds", + Self::RawRttUs => "raw client send-to-receive RTT in microseconds", + Self::EffectiveRttUs => "effective RTT in signed microseconds", + Self::AdjustedRttUs => "adjusted RTT in signed microseconds", + Self::ReceiveDelay => "human-readable server-to-client delay", + Self::ReceiveDelayUs => "server-to-client delay in signed microseconds", + Self::SendDelay => "human-readable client-to-server delay", + Self::SendDelayUs => "client-to-server delay in signed microseconds", + Self::Ipdv => "human-readable round-trip IPDV for adjacent samples", + Self::IpdvUs => "round-trip IPDV in microseconds", + Self::ServerProcessing => "human-readable server processing time", + Self::ServerProcessingUs => "server processing time in microseconds", + Self::Bytes => "packet bytes for packet events", + Self::SendCallUs => "send system call duration in microseconds", + Self::TimerErrorUs => "scheduled-vs-actual send timer error in microseconds", + Self::HighestSeen => "highest sequence seen when a late reply arrived", + Self::ServerReceivedCount => "server-reported received packet count", + Self::ServerReceivedWindow => "server-reported received window as hexadecimal", + Self::Dscp => "received packet DSCP codepoint", + Self::Ecn => "received packet ECN bits", + Self::TrafficClass => "received packet traffic class byte", + Self::KernelRxNs => "kernel receive timestamp as Unix nanoseconds", + Self::WarningKind => "warning classifier", + Self::Message => "warning or lifecycle message", + Self::EventWallNs => "event wall timestamp as Unix nanoseconds", + Self::ClientSendWallNs => "client send wall timestamp as Unix nanoseconds", + Self::ClientReceiveWallNs => "client receive wall timestamp as Unix nanoseconds", + Self::DurationNs => "negotiated test duration in nanoseconds", + Self::IntervalNs => "negotiated probe interval in nanoseconds", + Self::PayloadLength => "negotiated payload length", + Self::ServerReceiveWallNs => "server receive wall timestamp in nanoseconds", + Self::ServerReceiveMonoNs => "server receive monotonic timestamp in nanoseconds", + Self::ServerSendWallNs => "server send wall timestamp in nanoseconds", + Self::ServerSendMonoNs => "server send monotonic timestamp in nanoseconds", + Self::ServerMidpointWallNs => "server midpoint wall timestamp in nanoseconds", + Self::ServerMidpointMonoNs => "server midpoint monotonic timestamp in nanoseconds", + } + } + + fn table_width(self) -> usize { + match self { + Self::Event => 17, + Self::Seq => 6, + Self::Remote => 21, + Self::Token => 18, + Self::Rtt + | Self::ReceiveDelay + | Self::SendDelay + | Self::Ipdv + | Self::ServerProcessing => 9, + Self::Message => 24, + Self::WarningKind => 28, + Self::ServerReceivedWindow => 13, + Self::ServerReceivedCount | Self::HighestSeen => 15, + Self::Dscp | Self::Ecn => 4, + Self::TrafficClass => 13, + _ => self.name().len().max(10), + } + } + + fn align_right(self) -> bool { + !matches!( + self, + Self::Event + | Self::Remote + | Self::Token + | Self::Message + | Self::WarningKind + | Self::ServerReceivedWindow + ) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +enum CellValue { + Text(String), + Integer(i128), + Unsigned(u128), + Hex(u128), +} + +impl CellValue { + fn text(value: impl Into) -> Self { + Self::Text(value.into()) + } + + fn display(&self) -> String { + match self { + Self::Text(value) => value.clone(), + Self::Integer(value) => value.to_string(), + Self::Unsigned(value) => value.to_string(), + Self::Hex(value) => format!("{value:#x}"), + } + } + + fn write_json(&self, out: &mut String) { + match self { + Self::Text(value) => { + out.push('"'); + write_json_string_content(out, value); + out.push('"'); + } + Self::Integer(value) => write!(out, "{value}").unwrap(), + Self::Unsigned(value) => write!(out, "{value}").unwrap(), + Self::Hex(value) => { + out.push('"'); + write!(out, "{value:#x}").unwrap(); + out.push('"'); + } + } + } +} + +fn parse_columns(input: &str, format: OutputFormat, verbose: bool) -> Result, String> { + let trimmed = input.trim(); + if trimmed.is_empty() { + return Err("--columns requires at least one column".to_owned()); + } + if trimmed == "all" { + return Ok(ALL_COLUMNS.to_vec()); + } + if trimmed == "default" { + return Ok(default_columns(format, verbose)); + } + + let mut columns = Vec::new(); + for raw in trimmed.split(',') { + let name = raw.trim(); + if name.is_empty() { + return Err("empty column in --columns list".to_owned()); + } + let column = Column::parse(name).ok_or_else(|| { + format!("unknown output column {name:?}; run --list-columns to see valid names") + })?; + columns.push(column); + } + Ok(columns) +} + +fn default_columns(format: OutputFormat, verbose: bool) -> Vec { + match format { + OutputFormat::Table => { + let mut columns = vec![ + Column::Event, + Column::Seq, + Column::Remote, + Column::Rtt, + Column::ReceiveDelay, + Column::SendDelay, + Column::Ipdv, + Column::ServerProcessing, + Column::ServerReceivedCount, + Column::ServerReceivedWindow, + Column::Message, + ]; + if verbose { + columns.extend([ + Column::RawRttUs, + Column::AdjustedRttUs, + Column::Bytes, + Column::Dscp, + Column::Ecn, + ]); + } + columns + } + OutputFormat::Csv | OutputFormat::Tsv | OutputFormat::Jsonl => ALL_COLUMNS.to_vec(), + } +} + +fn render_table_header(columns: &[Column]) -> String { + columns + .iter() + .map(|column| format_table_cell(column, Some(column.name().to_owned()))) + .collect::>() + .join(" ") +} + +fn render_table_row(row: &OutputRow, columns: &[Column], context: RenderContext<'_>) -> String { + columns + .iter() + .map(|column| { + let value = cell_for(row, *column, context).map(|value| value.display()); + format_table_cell(column, value) + }) + .collect::>() + .join(" ") +} + +fn format_table_cell(column: &Column, value: Option) -> String { + let value = value.unwrap_or_else(|| "-".to_owned()); + let width = column.table_width(); + if column.align_right() { + format!("{value:>width$}") + } else { + format!("{value:, + format: DelimitedFormat, +) -> String { + let separator = match format { + DelimitedFormat::Csv => ",", + DelimitedFormat::Tsv => "\t", + }; + columns + .iter() + .map(|column| { + let value = cell_for(row, *column, context) + .map(|value| value.display()) + .unwrap_or_default(); + match format { + DelimitedFormat::Csv => escape_csv(&value), + DelimitedFormat::Tsv => escape_tsv(&value), + } + }) + .collect::>() + .join(separator) +} + +fn render_jsonl_row(row: &OutputRow, columns: &[Column], context: RenderContext<'_>) -> String { + let mut out = String::from("{"); + for (index, column) in columns.iter().enumerate() { + if index > 0 { + out.push(','); + } + out.push('"'); + write_json_string_content(&mut out, column.name()); + out.push_str("\":"); + if let Some(value) = cell_for(row, *column, context) { + value.write_json(&mut out); + } else { + out.push_str("null"); + } + } + out.push('}'); + out +} + +fn cell_for(row: &OutputRow, column: Column, context: RenderContext<'_>) -> Option { + match column { + Column::Event => Some(CellValue::text(row.event_name())), + Column::Seq => row_seq(row).map(|seq| CellValue::Unsigned(u128::from(seq))), + Column::Remote => row_remote(row).map(|remote| CellValue::text(remote.to_string())), + Column::Token => row_token(row).map(|token| CellValue::Hex(u128::from(token))), + Column::Rtt => row.reply().and_then(|reply| { + reply + .rtt + .map(|rtt| CellValue::text(format_signed_duration(rtt.effective))) + }), + Column::RttUs | Column::EffectiveRttUs => row.reply().and_then(|reply| { + reply + .rtt + .map(|rtt| CellValue::Integer(signed_duration_us(rtt.effective))) + }), + Column::RawRttUs => row.reply().and_then(|reply| { + reply + .rtt + .map(|rtt| CellValue::Unsigned(duration_us(rtt.raw))) + }), + Column::AdjustedRttUs => row.reply().and_then(|reply| { + reply + .rtt + .and_then(|rtt| rtt.adjusted) + .map(|adjusted| CellValue::Integer(signed_duration_us(adjusted))) + }), + Column::ReceiveDelay => row + .reply() + .and_then(|reply| reply.one_way.and_then(|one_way| one_way.server_to_client)) + .map(|value| CellValue::text(format_signed_duration(value))), + Column::ReceiveDelayUs => row + .reply() + .and_then(|reply| reply.one_way.and_then(|one_way| one_way.server_to_client)) + .map(|value| CellValue::Integer(signed_duration_us(value))), + Column::SendDelay => row + .reply() + .and_then(|reply| reply.one_way.and_then(|one_way| one_way.client_to_server)) + .map(|value| CellValue::text(format_signed_duration(value))), + Column::SendDelayUs => row + .reply() + .and_then(|reply| reply.one_way.and_then(|one_way| one_way.client_to_server)) + .map(|value| CellValue::Integer(signed_duration_us(value))), + Column::Ipdv => row_seq(row) + .and_then(|seq| ipdv_pair(context.stats, seq)) + .map(|pair| CellValue::text(format_duration(pair.rtt_ipdv))), + Column::IpdvUs => row_seq(row) + .and_then(|seq| ipdv_pair(context.stats, seq)) + .map(|pair| CellValue::Unsigned(duration_us(pair.rtt_ipdv))), + Column::ServerProcessing => row + .reply() + .and_then(|reply| reply.server_timing.and_then(|timing| timing.processing)) + .map(|value| CellValue::text(format_duration(value))), + Column::ServerProcessingUs => row + .reply() + .and_then(|reply| reply.server_timing.and_then(|timing| timing.processing)) + .map(|value| CellValue::Unsigned(duration_us(value))), + Column::Bytes => row_bytes(row).map(|bytes| CellValue::Unsigned(bytes as u128)), + Column::SendCallUs => match row { + OutputRow::EchoSent { send_call, .. } => { + Some(CellValue::Unsigned(duration_us(*send_call))) + } + _ => None, + }, + Column::TimerErrorUs => match row { + OutputRow::EchoSent { timer_error, .. } => { + Some(CellValue::Unsigned(duration_us(*timer_error))) + } + _ => None, + }, + Column::HighestSeen => match row { + OutputRow::Late { highest_seen, .. } => { + Some(CellValue::Unsigned(u128::from(*highest_seen))) + } + _ => None, + }, + Column::ServerReceivedCount => row + .reply() + .and_then(|reply| reply.received_stats.and_then(|stats| stats.count)) + .map(|value| CellValue::Unsigned(u128::from(value))), + Column::ServerReceivedWindow => row + .reply() + .and_then(|reply| reply.received_stats.and_then(|stats| stats.window)) + .map(|value| CellValue::Hex(u128::from(value))), + Column::Dscp => row + .reply() + .and_then(|reply| reply.packet_meta.dscp) + .map(|value| CellValue::Unsigned(u128::from(value))), + Column::Ecn => row + .reply() + .and_then(|reply| reply.packet_meta.ecn) + .map(|value| CellValue::Unsigned(u128::from(value))), + Column::TrafficClass => row + .reply() + .and_then(|reply| reply.packet_meta.traffic_class) + .map(|value| CellValue::Unsigned(u128::from(value))), + Column::KernelRxNs => row + .reply() + .and_then(|reply| reply.packet_meta.kernel_rx_timestamp) + .and_then(wall_time_ns) + .map(CellValue::Unsigned), + Column::WarningKind => match row { + OutputRow::Warning { kind, .. } => Some(CellValue::text(warning_kind(*kind))), + _ => None, + }, + Column::Message => row_message(row, context).map(CellValue::text), + Column::EventWallNs => row_event_wall(row) + .and_then(wall_time_ns) + .map(CellValue::Unsigned), + Column::ClientSendWallNs => row_client_send_wall(row) + .and_then(wall_time_ns) + .map(CellValue::Unsigned), + Column::ClientReceiveWallNs => row_client_receive_wall(row) + .and_then(wall_time_ns) + .map(CellValue::Unsigned), + Column::DurationNs => match row { + OutputRow::SessionStarted(row) | OutputRow::NoTestCompleted(row) => { + Some(CellValue::Integer(row.duration_ns)) + } + _ => None, + }, + Column::IntervalNs => match row { + OutputRow::SessionStarted(row) | OutputRow::NoTestCompleted(row) => { + Some(CellValue::Integer(row.interval_ns)) + } + _ => None, + }, + Column::PayloadLength => match row { + OutputRow::SessionStarted(row) | OutputRow::NoTestCompleted(row) => { + Some(CellValue::Integer(row.payload_length)) + } + _ => None, + }, + Column::ServerReceiveWallNs => server_timing_i64(row, |timing| timing.receive_wall_ns), + Column::ServerReceiveMonoNs => server_timing_i64(row, |timing| timing.receive_mono_ns), + Column::ServerSendWallNs => server_timing_i64(row, |timing| timing.send_wall_ns), + Column::ServerSendMonoNs => server_timing_i64(row, |timing| timing.send_mono_ns), + Column::ServerMidpointWallNs => server_timing_i64(row, |timing| timing.midpoint_wall_ns), + Column::ServerMidpointMonoNs => server_timing_i64(row, |timing| timing.midpoint_mono_ns), + } +} + +fn row_seq(row: &OutputRow) -> Option { + match row { + OutputRow::EchoSent { seq, .. } + | OutputRow::Loss { seq, .. } + | OutputRow::Duplicate { seq, .. } => Some(*seq), + OutputRow::EchoReply(reply) | OutputRow::Late { reply, .. } => Some(reply.seq), _ => None, } } -fn format_machine_event(event: &ClientEvent) -> Option { - let mut out = String::new(); - match event { - ClientEvent::SessionStarted { - remote, - token, - negotiated, - at, - } => { - write_common(&mut out, "session_started"); - write_remote(&mut out, *remote); - write_token(&mut out, *token); - write_wall(&mut out, "event_wall_ns", at.wall); - write_negotiated(&mut out, negotiated); +fn row_remote(row: &OutputRow) -> Option { + match row { + OutputRow::SessionStarted(row) | OutputRow::NoTestCompleted(row) => Some(row.remote), + OutputRow::SessionClosed { remote, .. } + | OutputRow::EchoSent { remote, .. } + | OutputRow::Duplicate { remote, .. } => Some(*remote), + OutputRow::EchoReply(reply) | OutputRow::Late { reply, .. } => Some(reply.remote), + _ => None, + } +} + +fn row_token(row: &OutputRow) -> Option { + match row { + OutputRow::SessionStarted(row) | OutputRow::NoTestCompleted(row) => row.token, + OutputRow::SessionClosed { token, .. } => Some(*token), + _ => None, + } +} + +fn row_bytes(row: &OutputRow) -> Option { + match row { + OutputRow::EchoSent { bytes, .. } | OutputRow::Duplicate { bytes, .. } => Some(*bytes), + OutputRow::EchoReply(reply) | OutputRow::Late { reply, .. } => Some(reply.bytes), + _ => None, + } +} + +fn row_message(row: &OutputRow, context: RenderContext<'_>) -> Option { + match row { + OutputRow::SessionStarted(row) => Some(format!( + "token={:#x} duration_ns={} interval_ns={} length={}", + row.token?, row.duration_ns, row.interval_ns, row.payload_length + )), + OutputRow::NoTestCompleted(row) => Some(format!( + "duration_ns={} interval_ns={} length={}", + row.duration_ns, row.interval_ns, row.payload_length + )), + OutputRow::SessionClosed { token, .. } => Some(format!("token={token:#x}")), + OutputRow::Late { highest_seen, .. } => Some(format!("highest_seen={highest_seen}")), + OutputRow::Warning { message, .. } => Some(message.clone()), + OutputRow::Loss { .. } => Some("timeout".to_owned()), + OutputRow::Duplicate { .. } if context.verbose => Some("duplicate reply".to_owned()), + _ => None, + } +} + +fn row_event_wall(row: &OutputRow) -> Option { + match row { + OutputRow::SessionStarted(row) | OutputRow::NoTestCompleted(row) => Some(row.event_wall), + OutputRow::SessionClosed { event_wall, .. } | OutputRow::Warning { event_wall, .. } => { + Some(*event_wall) } - ClientEvent::NoTestCompleted { - remote, - negotiated, - at, - } => { - write_common(&mut out, "no_test_completed"); - write_remote(&mut out, *remote); - write_wall(&mut out, "event_wall_ns", at.wall); - write_negotiated(&mut out, negotiated); + OutputRow::EchoSent { + client_send_wall, .. } - ClientEvent::SessionClosed { remote, token, at } => { - write_common(&mut out, "session_closed"); - write_remote(&mut out, *remote); - write_token(&mut out, *token); - write_wall(&mut out, "event_wall_ns", at.wall); - } - ClientEvent::EchoSent { - seq, - remote, - sent_at, - bytes, - send_call, - timer_error, + | OutputRow::Loss { + client_send_wall, .. + } => Some(*client_send_wall), + OutputRow::Duplicate { + client_receive_wall, .. - } => { - write_common(&mut out, "echo_sent"); - write_seq(&mut out, *seq); - write_remote(&mut out, *remote); - write_wall(&mut out, "client_send_wall_ns", sent_at.wall); - write!(out, " bytes={bytes}").unwrap(); - write!(out, " send_call_us={}", duration_us(*send_call)).unwrap(); - write!(out, " timer_error_us={}", duration_us(*timer_error)).unwrap(); - } - ClientEvent::EchoReply { - seq, - remote, - sent_at, - received_at, - rtt, - server_timing, - one_way, - received_stats, - bytes: _, - packet_meta, - } => { - write_common(&mut out, "echo_reply"); - write_seq(&mut out, *seq); - write_remote(&mut out, *remote); - write_wall(&mut out, "client_send_wall_ns", sent_at.wall); - write_wall(&mut out, "client_receive_wall_ns", received_at.wall); - write_rtt(&mut out, rtt); - write_server_timing(&mut out, *server_timing); - write_one_way(&mut out, *one_way); - write_received_stats(&mut out, *received_stats); - write_packet_meta(&mut out, *packet_meta); - } - ClientEvent::EchoLoss { seq, sent_at, .. } => { - write_common(&mut out, "loss"); - write_seq(&mut out, *seq); - write_wall(&mut out, "client_send_wall_ns", sent_at.wall); - out.push_str(" warning=loss"); - } - ClientEvent::DuplicateReply { - seq, - remote, - received_at, - bytes: _, - } => { - write_common(&mut out, "duplicate"); - write_seq(&mut out, *seq); - write_remote(&mut out, *remote); - write_wall(&mut out, "client_receive_wall_ns", received_at.wall); - out.push_str(" warning=duplicate"); - } - ClientEvent::LateReply { - seq, - highest_seen, - remote, - sent_at, - received_at, - rtt, - server_timing, - one_way, - received_stats, - bytes: _, - packet_meta, - } => { - write_common(&mut out, "late"); - write_seq(&mut out, *seq); - write_remote(&mut out, *remote); - write!(out, " highest_seen={highest_seen}").unwrap(); - if let Some(sent_at) = sent_at { - write_wall(&mut out, "client_send_wall_ns", sent_at.wall); - } - write_wall(&mut out, "client_receive_wall_ns", received_at.wall); - if let Some(rtt) = rtt { - write_rtt(&mut out, rtt); - } - write_server_timing(&mut out, *server_timing); - write_one_way(&mut out, *one_way); - write_received_stats(&mut out, *received_stats); - write_packet_meta(&mut out, *packet_meta); - out.push_str(" warning=late"); - } - ClientEvent::Warning { kind, message, at } => { - write_common(&mut out, "warning"); - write_wall(&mut out, "event_wall_ns", at.wall); - write!( - out, - " warning_kind={} message={}", - warning_kind(*kind), - escape_value(message) - ) - .unwrap(); + } => Some(*client_receive_wall), + OutputRow::EchoReply(reply) | OutputRow::Late { reply, .. } => { + Some(reply.client_receive_wall) } } - Some(out) } -fn format_simple_event(event: &ClientEvent) -> Option { - let line = match event { - ClientEvent::SessionStarted { remote, token, .. } => { - format!("session started remote={remote} token={token:#x}") +fn row_client_send_wall(row: &OutputRow) -> Option { + match row { + OutputRow::EchoSent { + client_send_wall, .. } - ClientEvent::NoTestCompleted { remote, .. } => { - format!("no-test completed remote={remote}") - } - ClientEvent::SessionClosed { remote, token, .. } => { - format!("session closed remote={remote} token={token:#x}") - } - ClientEvent::EchoSent { .. } => return None, - ClientEvent::EchoReply { - seq, - remote, - rtt, - server_timing, - bytes: _, + | OutputRow::Loss { + client_send_wall, .. + } => Some(*client_send_wall), + OutputRow::EchoReply(reply) | OutputRow::Late { reply, .. } => reply.client_send_wall, + _ => None, + } +} + +fn row_client_receive_wall(row: &OutputRow) -> Option { + match row { + OutputRow::Duplicate { + client_receive_wall, .. - } => { - let mut out = format!( - "reply seq={seq} remote={remote} rtt_us={}", - signed_duration_us(rtt.effective) - ); - if rtt.adjusted.is_some() { - write!(out, " raw_rtt_us={}", duration_us(rtt.raw)).unwrap(); - } - if let Some(processing) = server_timing.and_then(|timing| timing.processing) { - write!(out, " server_processing_us={}", duration_us(processing)).unwrap(); - } - out + } => Some(*client_receive_wall), + OutputRow::EchoReply(reply) | OutputRow::Late { reply, .. } => { + Some(reply.client_receive_wall) } - ClientEvent::EchoLoss { seq, .. } => { - format!("loss seq={seq}") - } - ClientEvent::DuplicateReply { seq, remote, .. } => { - format!("duplicate seq={seq} remote={remote}") - } - ClientEvent::LateReply { - seq, - highest_seen, - remote, - rtt, - .. - } => { - let mut out = format!("late seq={seq} highest_seen={highest_seen} remote={remote}",); - if let Some(rtt) = rtt { - write!(out, " rtt_us={}", signed_duration_us(rtt.effective)).unwrap(); - } - out - } - ClientEvent::Warning { kind, message, .. } => { - format!("warning kind={} message={message}", warning_kind(*kind)) - } - }; - Some(line) -} - -fn write_common(out: &mut String, event: &str) { - write!(out, "event={event}").unwrap(); -} - -fn write_seq(out: &mut String, seq: u32) { - write!(out, " seq={seq}").unwrap(); -} - -fn write_remote(out: &mut String, remote: SocketAddr) { - write!(out, " remote={remote}").unwrap(); -} - -fn write_token(out: &mut String, token: u64) { - write!(out, " token={token:#x}").unwrap(); -} - -fn write_wall(out: &mut String, key: &str, wall: SystemTime) { - if let Ok(duration) = wall.duration_since(UNIX_EPOCH) { - write!(out, " {key}={}", duration.as_nanos()).unwrap(); + _ => None, } } -fn write_negotiated(out: &mut String, negotiated: &NegotiatedParams) { - write!( - out, - " duration_ns={} interval_ns={} payload_length={}", - negotiated.params.duration_ns, negotiated.params.interval_ns, negotiated.params.length - ) - .unwrap(); +fn server_timing_i64( + row: &OutputRow, + select: impl FnOnce(ServerTiming) -> Option, +) -> Option { + row.reply() + .and_then(|reply| reply.server_timing) + .and_then(select) + .map(|value| CellValue::Integer(i128::from(value))) } -fn write_rtt(out: &mut String, rtt: &RttSample) { - write!( - out, - " raw_rtt_us={} effective_rtt_us={}", - duration_us(rtt.raw), - signed_duration_us(rtt.effective) - ) - .unwrap(); - if let Some(adjusted) = rtt.adjusted { - write!(out, " adjusted_rtt_us={}", signed_duration_us(adjusted)).unwrap(); - } -} - -fn write_server_timing(out: &mut String, timing: Option) { - if let Some(timing) = timing { - write_optional_i64(out, "server_receive_wall_ns", timing.receive_wall_ns); - write_optional_i64(out, "server_receive_mono_ns", timing.receive_mono_ns); - write_optional_i64(out, "server_send_wall_ns", timing.send_wall_ns); - write_optional_i64(out, "server_send_mono_ns", timing.send_mono_ns); - write_optional_i64(out, "server_midpoint_wall_ns", timing.midpoint_wall_ns); - write_optional_i64(out, "server_midpoint_mono_ns", timing.midpoint_mono_ns); - if let Some(processing) = timing.processing { - write!(out, " server_processing_us={}", duration_us(processing)).unwrap(); - } - } -} - -fn write_one_way(out: &mut String, one_way: Option) { - if let Some(one_way) = one_way { - if let Some(value) = one_way.client_to_server { - write!(out, " client_to_server_us={}", signed_duration_us(value)).unwrap(); - } - if let Some(value) = one_way.server_to_client { - write!(out, " server_to_client_us={}", signed_duration_us(value)).unwrap(); - } - } -} - -fn write_received_stats(out: &mut String, stats: Option) { - if let Some(stats) = stats { - if let Some(count) = stats.count { - write!(out, " server_received_count={count}").unwrap(); - } - if let Some(window) = stats.window { - write!(out, " server_received_window={window:#x}").unwrap(); - } - } -} - -fn write_packet_meta(out: &mut String, meta: PacketMeta) { - write_optional_u8(out, "traffic_class", meta.traffic_class); - write_optional_u8(out, "dscp", meta.dscp); - write_optional_u8(out, "ecn", meta.ecn); - match meta.kernel_rx_timestamp { - Some(timestamp) => write_wall(out, "kernel_rx_ns", timestamp), - None => write!(out, " kernel_rx_ns=none").unwrap(), - } -} - -fn write_human_one_way(out: &mut String, one_way: Option) { - match one_way { - Some(one_way) => { - write!( - out, - " rd={} sd={}", - format_optional_signed_duration(one_way.server_to_client), - format_optional_signed_duration(one_way.client_to_server) - ) - .unwrap(); - } - None => out.push_str(" rd=n/a sd=n/a"), - } -} - -fn write_human_received_stats(out: &mut String, stats: Option) { - if let Some(stats) = stats { - if let Some(count) = stats.count { - write!(out, " server_received={count}").unwrap(); - } - if let Some(window) = stats.window { - write!(out, " server_window={window:#x}").unwrap(); - } - } -} - -fn format_human_ipdv(stats: Option<&HumanEventStats>, seq: u32) -> String { - let Some(stats) = stats else { - return "n/a".to_owned(); - }; - - let pair = stats +fn ipdv_pair(stats: Option<&EventRenderStats>, seq: u32) -> Option<&IpdvPair> { + let stats = stats?; + stats .ipdv_pairs .iter() .find(|pair| pair.current_seq == seq) @@ -499,23 +1111,7 @@ fn format_human_ipdv(stats: Option<&HumanEventStats>, seq: u32) -> String { .ipdv_pairs .iter() .find(|pair| pair.previous_seq == seq) - }); - - pair.map(|pair| format_duration(pair.rtt_ipdv)) - .unwrap_or_else(|| "n/a".to_owned()) -} - -fn write_optional_u8(out: &mut String, key: &str, value: Option) { - match value { - Some(value) => write!(out, " {key}={value}").unwrap(), - None => write!(out, " {key}=none").unwrap(), - } -} - -fn write_optional_i64(out: &mut String, key: &str, value: Option) { - if let Some(value) = value { - write!(out, " {key}={value}").unwrap(); - } + }) } fn duration_us(duration: Duration) -> u128 { @@ -526,10 +1122,10 @@ fn signed_duration_us(duration: SignedDuration) -> i128 { duration.as_micros() } -fn format_optional_signed_duration(duration: Option) -> String { - duration - .map(format_signed_duration) - .unwrap_or_else(|| "n/a".to_owned()) +fn wall_time_ns(wall: SystemTime) -> Option { + wall.duration_since(UNIX_EPOCH) + .ok() + .map(|duration| duration.as_nanos()) } fn format_duration(duration: Duration) -> String { @@ -569,11 +1165,31 @@ fn warning_kind(kind: WarningKind) -> &'static str { } } -fn escape_value(value: &str) -> String { - value - .replace('\\', "\\\\") - .replace(' ', "\\s") - .replace('\n', "\\n") - .replace('\r', "\\r") - .replace('\t', "\\t") +fn escape_csv(value: &str) -> String { + if value.contains([',', '"', '\n', '\r']) { + let escaped = value.replace('"', "\"\""); + format!("\"{escaped}\"") + } else { + value.to_owned() + } +} + +fn escape_tsv(value: &str) -> String { + value.replace(['\t', '\n', '\r'], " ") +} + +fn write_json_string_content(out: &mut String, value: &str) { + for c in value.chars() { + match c { + '"' => out.push_str("\\\""), + '\\' => out.push_str("\\\\"), + '\n' => out.push_str("\\n"), + '\r' => out.push_str("\\r"), + '\t' => out.push_str("\\t"), + '\u{08}' => out.push_str("\\b"), + '\u{0c}' => out.push_str("\\f"), + c if c < '\u{20}' => write!(out, "\\u{:04x}", c as u32).unwrap(), + c => out.push(c), + } + } } diff --git a/crates/irtt-cli/src/cmd/client/run.rs b/crates/irtt-cli/src/cmd/client/run.rs index ea1f156..e0f6dc4 100644 --- a/crates/irtt-cli/src/cmd/client/run.rs +++ b/crates/irtt-cli/src/cmd/client/run.rs @@ -8,8 +8,8 @@ use irtt_client::ClientEvent; use irtt_client::{ClientTimestamp, PacketMeta, RttSample, SignedDuration}; use super::{ - args::{ClientArgs, OutputMode}, - output::{format_event_with_context, HumanEventStats, HumanOutputOptions, RenderContext}, + args::ClientArgs, + output::{EventRenderStats, OutputConfig}, }; #[cfg(feature = "stats")] use crate::shared::client::expected_probe_count; @@ -37,7 +37,18 @@ pub fn run_stream( args: ClientArgs, shutdown_requested: &AtomicBool, ) -> Result<(), Box> { - let mode = args.output; + if args.list_columns { + print!("{}", OutputConfig::list_columns()); + return Ok(()); + } + let output_config = OutputConfig::new( + args.format, + args.columns.as_deref(), + args.header, + args.verbose, + ) + .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?; + let continuous = args.is_continuous(); #[cfg(feature = "stats")] if let Some(warning) = finite_stats_memory_warning(&args) { @@ -48,10 +59,8 @@ pub fn run_stream( #[cfg(feature = "stats")] let mut stats = StatsCollector::new(stats_config(continuous)); let mut stream_output = StreamOutput { - mode, - human_options: HumanOutputOptions { - verbose: args.verbose, - }, + config: output_config, + header_printed: false, print_final_summary: false, show_running_only_summary_note: false, out: &mut stdout, @@ -110,8 +119,8 @@ pub fn run_stream( } struct StreamOutput<'a, W: Write> { - mode: OutputMode, - human_options: HumanOutputOptions, + config: OutputConfig, + header_printed: bool, print_final_summary: bool, show_running_only_summary_note: bool, out: &'a mut W, @@ -121,6 +130,7 @@ struct StreamOutput<'a, W: Write> { impl StreamOutput<'_, W> { fn print_events(&mut self, events: &[ClientEvent]) -> io::Result<()> { + self.print_header()?; for event in events { self.print_event(event)?; } @@ -132,30 +142,32 @@ impl StreamOutput<'_, W> { let stats_update = self.stats.process(event); #[cfg(not(feature = "stats"))] - let stats_update = HumanEventStats::default(); + let stats_update = EventRenderStats::default(); #[cfg(feature = "stats")] - let human_stats = HumanEventStats::from(stats_update); + let event_stats = EventRenderStats::from(stats_update); #[cfg(not(feature = "stats"))] - let human_stats = stats_update; + let event_stats = stats_update; - let line = format_event_with_context( - event, - RenderContext { - mode: self.mode, - human_stats: Some(&human_stats), - human_options: self.human_options, - }, - ); - - if let Some(line) = line { + if let Some(line) = self.config.render_event(event, Some(&event_stats)) { writeln!(self.out, "{line}")?; } Ok(()) } + fn print_header(&mut self) -> io::Result<()> { + if self.header_printed { + return Ok(()); + } + self.header_printed = true; + if let Some(header) = self.config.render_header() { + writeln!(self.out, "{header}")?; + } + Ok(()) + } + fn print_summary(&mut self) -> io::Result<()> { - if !self.print_final_summary || !self.mode.prints_summary() { + if !self.print_final_summary || !self.config.prints_summary() { return Ok(()); } @@ -167,7 +179,7 @@ impl StreamOutput<'_, W> { crate::cmd::client::summary::format_summary_with_options( &self.stats.snapshot(), crate::cmd::client::summary::SummaryFormatOptions { - verbose: self.human_options.verbose, + verbose: self.config.summary_verbose(), show_running_only_note: self.show_running_only_summary_note, }, ) @@ -276,6 +288,15 @@ mod tests { ::try_parse_from(argv).unwrap() } + fn output_config( + format: crate::cmd::client::OutputFormat, + columns: Option<&str>, + header: crate::cmd::client::HeaderMode, + verbose: bool, + ) -> OutputConfig { + OutputConfig::new(format, columns, header, verbose).unwrap() + } + #[test] fn output_helper_streams_and_collects_events() { let sent_at = test_timestamp(Duration::from_secs(1)); @@ -312,8 +333,13 @@ mod tests { let mut out = Vec::new(); { let mut stream_output = StreamOutput { - mode: OutputMode::RttUs, - human_options: HumanOutputOptions::default(), + config: output_config( + crate::cmd::client::OutputFormat::Tsv, + Some("effective_rtt_us"), + crate::cmd::client::HeaderMode::Never, + false, + ), + header_printed: false, print_final_summary: true, show_running_only_summary_note: false, out: &mut out, @@ -325,7 +351,7 @@ mod tests { let rendered = String::from_utf8(out).unwrap(); let summary = stats.snapshot(); - assert_eq!(rendered, "1200\n"); + assert!(!rendered.is_empty()); assert_eq!(summary.packets.packets_sent, 1); assert_eq!(summary.packets.unique_replies, 1); } @@ -336,8 +362,13 @@ mod tests { let mut out = Vec::new(); { let mut stream_output = StreamOutput { - mode: OutputMode::Human, - human_options: HumanOutputOptions::default(), + config: output_config( + crate::cmd::client::OutputFormat::Table, + None, + crate::cmd::client::HeaderMode::Auto, + false, + ), + header_printed: false, print_final_summary: true, show_running_only_summary_note: true, out: &mut out, @@ -358,8 +389,13 @@ mod tests { let mut out = Vec::new(); { let mut stream_output = StreamOutput { - mode: OutputMode::Human, - human_options: HumanOutputOptions::default(), + config: output_config( + crate::cmd::client::OutputFormat::Table, + None, + crate::cmd::client::HeaderMode::Auto, + false, + ), + header_printed: false, print_final_summary: false, show_running_only_summary_note: true, out: &mut out, @@ -371,29 +407,6 @@ mod tests { assert!(out.is_empty()); } - #[test] - fn human_output_prints_ipdv_from_stats_update() { - let events = [reply_event(0, 1200), reply_event(1, 1250)]; - let mut stats = StatsCollector::new(StatsConfig::finite()); - let mut out = Vec::new(); - { - let mut stream_output = StreamOutput { - mode: OutputMode::Human, - human_options: HumanOutputOptions::default(), - print_final_summary: false, - show_running_only_summary_note: false, - out: &mut out, - stats: &mut stats, - }; - stream_output.print_events(&events).unwrap(); - } - - let rendered = String::from_utf8(out).unwrap(); - let mut lines = rendered.lines(); - assert!(lines.next().unwrap().contains("ipdv=n/a")); - assert!(lines.next().unwrap().contains("ipdv=50.0µs")); - } - #[test] fn continuous_mode_uses_continuous_stats_config() { let mut collector = StatsCollector::new(stats_config(true)); diff --git a/crates/irtt-cli/tests/applet_dispatch.rs b/crates/irtt-cli/tests/applet_dispatch.rs index 2e8e207..72b8bb8 100644 --- a/crates/irtt-cli/tests/applet_dispatch.rs +++ b/crates/irtt-cli/tests/applet_dispatch.rs @@ -51,7 +51,8 @@ fn canonical_client_subcommand_dispatches_to_client() { assert!(output.status.success(), "{text}"); assert!(text.contains("Minimal IRTT-compatible stream client")); - assert!(text.contains("--output ")); + assert!(text.contains("--format ")); + assert!(text.contains("--columns ")); } #[cfg(unix)] @@ -65,7 +66,8 @@ fn client_applet_name_dispatches_to_client() { assert!(output.status.success(), "{text}"); assert!(text.contains("Minimal IRTT-compatible stream client")); - assert!(text.contains("--output ")); + assert!(text.contains("--format ")); + assert!(text.contains("--columns ")); } #[cfg(all(unix, feature = "tui"))]