From b02bc53cd8b442b709ecbe10bcfb97bad2751fc8 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 22 Jun 2026 13:16:27 +0700 Subject: [PATCH] feat(cli): introduce flexible event row output formats and column selection Replace fixed output modes with customizable table, CSV, TSV, and JSON Lines formats. This allows users to select specific data columns and control header visibility for better integration with scripts and monitoring tools. --- README.md | 76 +- crates/irtt-cli/src/cmd/client/args.rs | 98 +- crates/irtt-cli/src/cmd/client/mod.rs | 6 +- crates/irtt-cli/src/cmd/client/output.rs | 1490 +++++++++++++++------- crates/irtt-cli/src/cmd/client/run.rs | 121 +- crates/irtt-cli/tests/applet_dispatch.rs | 6 +- 6 files changed, 1250 insertions(+), 547 deletions(-) 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"))]