Merge pull request #94 from Lochnair/refactor/shared-target-preparation

refactor(cli): centralize managed target preparation
This commit was merged in pull request #94.
This commit is contained in:
Nils Andreas Svee authored and GitHub committed 2026-08-15 10:05:43 +07:00
commit 006add3648
8 files changed
+347 -128

No files matched your search

+30 -38
View File
@@ -1,12 +1,10 @@
use std::time::Duration; use std::time::Duration;
use clap::{Parser, ValueEnum};
use irtt_client::ClientConfig;
use crate::shared::client::{ use crate::shared::client::{
parse_labelled_target, parse_test_duration, prepare_managed_targets, target_specs, parse_labelled_target, parse_test_duration, prepare_managed_run, CommonClientArgs,
CommonClientArgs, GroupPacingArg, LabelledTargetArg, PreparedTarget, TargetSpec, TimestampArg, GroupPacingArg, LabelledTargetArg, ManagedRunSetup, TargetSelection, TimestampArg,
}; };
use clap::{Parser, ValueEnum};
pub const DEFAULT_CLIENT_DURATION: Duration = Duration::from_secs(10); pub const DEFAULT_CLIENT_DURATION: Duration = Duration::from_secs(10);
@@ -91,11 +89,19 @@ pub struct ClientArgs {
} }
impl ClientArgs { impl ClientArgs {
pub fn to_client_config(&self) -> ClientConfig { /// Validate the selected targets and prepare the managed run.
self.common.to_client_config( ///
self.primary_target_addr() /// Targets are only required here. `--list-columns` returns before this is
.expect("target is required unless --list-columns is set"), /// called, so listing columns still needs no target.
pub fn prepare(&self) -> Result<ManagedRunSetup, String> {
prepare_managed_run(
&self.common,
self.duration, self.duration,
TargetSelection {
positional: &self.targets,
labelled: &self.labelled_targets,
pacing: self.pacing,
},
) )
} }
@@ -106,22 +112,6 @@ impl ClientArgs {
pub fn timestamp_mode(&self) -> TimestampArg { pub fn timestamp_mode(&self) -> TimestampArg {
self.common.tstamp self.common.tstamp
} }
pub fn target_specs(&self) -> Result<Vec<CliTargetSpec>, String> {
target_specs(&self.targets, &self.labelled_targets)
}
pub fn managed_targets(&self) -> Result<Vec<CliManagedTarget>, String> {
prepare_managed_targets(self.target_specs()?)
}
fn primary_target_addr(&self) -> Option<&str> {
self.targets.first().map(String::as_str).or_else(|| {
self.labelled_targets
.first()
.map(|target| target.addr.as_str())
})
}
} }
impl std::ops::Deref for ClientArgs { impl std::ops::Deref for ClientArgs {
@@ -144,9 +134,6 @@ pub enum OutputFormat {
Jsonl, Jsonl,
} }
pub type CliTargetSpec = TargetSpec;
pub type CliManagedTarget = PreparedTarget;
impl OutputFormat { impl OutputFormat {
pub fn prints_summary(self) -> bool { pub fn prints_summary(self) -> bool {
matches!(self, Self::Table) matches!(self, Self::Table)
@@ -191,7 +178,7 @@ mod tests {
assert!(!args.is_continuous()); assert!(!args.is_continuous());
assert_eq!(args.timestamp_mode(), TimestampArg::Both); assert_eq!(args.timestamp_mode(), TimestampArg::Both);
let config = args.to_client_config(); let config = args.prepare().unwrap().client;
assert_eq!(config.duration, Some(DEFAULT_CLIENT_DURATION)); assert_eq!(config.duration, Some(DEFAULT_CLIENT_DURATION));
assert_eq!(config.received_stats, ReceivedStats::Both); assert_eq!(config.received_stats, ReceivedStats::Both);
assert_eq!(config.stamp_at, StampAt::Both); assert_eq!(config.stamp_at, StampAt::Both);
@@ -212,17 +199,22 @@ mod tests {
let args = parse(&["--list-columns"]).unwrap(); let args = parse(&["--list-columns"]).unwrap();
assert!(args.targets.is_empty()); assert!(args.targets.is_empty());
assert!(args.list_columns); assert!(args.list_columns);
// The parser accepts no target; preparation is where a target becomes
// mandatory, and the stream command returns before preparing when
// listing columns.
let err = args.prepare().unwrap_err();
assert!(err.contains("--list-columns"));
} }
#[test] #[test]
fn multiple_positional_targets_parse_as_targets() { fn multiple_positional_targets_parse_as_targets() {
let args = parse(&["host-a:2112", "host-b:2112"]).unwrap(); let args = parse(&["host-a:2112", "host-b:2112"]).unwrap();
let specs = args.target_specs().unwrap(); let specs = args.prepare().unwrap().targets;
assert_eq!(specs[0].label, "host-a:2112"); assert_eq!(specs[0].label, "host-a:2112");
assert_eq!(specs[0].addr, "host-a:2112"); assert_eq!(specs[0].managed.server_addr, "host-a:2112");
assert_eq!(specs[1].label, "host-b:2112"); assert_eq!(specs[1].label, "host-b:2112");
assert_eq!(specs[1].addr, "host-b:2112"); assert_eq!(specs[1].managed.server_addr, "host-b:2112");
} }
#[test] #[test]
@@ -234,18 +226,18 @@ mod tests {
"sg=sg.example.com:2112", "sg=sg.example.com:2112",
]) ])
.unwrap(); .unwrap();
let specs = args.target_specs().unwrap(); let specs = args.prepare().unwrap().targets;
assert_eq!(specs[0].label, "ams"); assert_eq!(specs[0].label, "ams");
assert_eq!(specs[0].addr, "ams.example.com:2112"); assert_eq!(specs[0].managed.server_addr, "ams.example.com:2112");
assert_eq!(specs[1].label, "sg"); assert_eq!(specs[1].label, "sg");
assert_eq!(specs[1].addr, "sg.example.com:2112"); assert_eq!(specs[1].managed.server_addr, "sg.example.com:2112");
} }
#[test] #[test]
fn duplicate_labels_are_rejected() { fn duplicate_labels_are_rejected() {
let args = parse(&["host-a:2112", "--target", "host-a:2112=host-b:2112"]).unwrap(); let args = parse(&["host-a:2112", "--target", "host-a:2112=host-b:2112"]).unwrap();
let err = args.target_specs().unwrap_err(); let err = args.prepare().unwrap_err();
assert!(err.contains("duplicate target label")); assert!(err.contains("duplicate target label"));
} }
@@ -253,7 +245,7 @@ mod tests {
#[test] #[test]
fn duplicate_positional_target_strings_get_stable_suffixes() { fn duplicate_positional_target_strings_get_stable_suffixes() {
let args = parse(&["host-a:2112", "host-a:2112"]).unwrap(); let args = parse(&["host-a:2112", "host-a:2112"]).unwrap();
let specs = args.target_specs().unwrap(); let specs = args.prepare().unwrap().targets;
assert_eq!(specs[0].label, "host-a:2112"); assert_eq!(specs[0].label, "host-a:2112");
assert_eq!(specs[1].label, "host-a:2112#2"); assert_eq!(specs[1].label, "host-a:2112#2");
@@ -262,7 +254,7 @@ mod tests {
#[test] #[test]
fn duplicate_target_endpoints_are_allowed() { fn duplicate_target_endpoints_are_allowed() {
let args = parse(&["127.0.0.1:2112", "127.0.0.1"]).unwrap(); let args = parse(&["127.0.0.1:2112", "127.0.0.1"]).unwrap();
assert_eq!(args.managed_targets().unwrap().len(), 2); assert_eq!(args.prepare().unwrap().target_count(), 2);
} }
#[test] #[test]
+10 -23
View File
@@ -9,9 +9,8 @@ use std::{
use irtt_client::{ use irtt_client::{
managed::{ managed::{
BlockingManagedClient, ManagedClientConfig, ManagedCompletionPolicy, ManagedEndReason, BlockingManagedClient, ManagedEndReason, ManagedEvent, ManagedEventSubscription,
ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError, ManagedTargetEndReason, ManagedEventTryRecvError, ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance,
ManagedTargetOutcome, TargetInstance,
}, },
ClientEvent, ClientEvent,
}; };
@@ -44,7 +43,6 @@ const FINITE_STATS_MEMORY_STRONG_WARNING_BYTES: u64 = 512 * MIB;
const FINITE_STATS_MEMORY_VERY_STRONG_WARNING_BYTES: u64 = GIB; const FINITE_STATS_MEMORY_VERY_STRONG_WARNING_BYTES: u64 = GIB;
const MANAGED_EVENT_WAIT_SLICE: Duration = Duration::from_millis(20); const MANAGED_EVENT_WAIT_SLICE: Duration = Duration::from_millis(20);
const MANAGED_EVENT_CAPACITY: usize = 16_384;
pub fn run_stream( pub fn run_stream(
args: ClientArgs, args: ClientArgs,
@@ -54,10 +52,10 @@ pub fn run_stream(
print!("{}", OutputConfig::list_columns()); print!("{}", OutputConfig::list_columns());
return Ok(()); return Ok(());
} }
let targets = args let setup = args
.managed_targets() .prepare()
.map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?; .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?;
let multi_target = targets.len() > 1; let multi_target = setup.is_multi_target();
let output_config = OutputConfig::new( let output_config = OutputConfig::new(
args.format, args.format,
args.columns.as_deref(), args.columns.as_deref(),
@@ -67,7 +65,7 @@ pub fn run_stream(
) )
.map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?; .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?;
let continuous = args.is_continuous(); let continuous = args.is_continuous();
let target_count = targets.len(); let target_count = setup.target_count();
if let Some(warning) = finite_stats_memory_warning(&args, target_count) { if let Some(warning) = finite_stats_memory_warning(&args, target_count) {
eprintln!("{warning}"); eprintln!("{warning}");
@@ -75,21 +73,9 @@ pub fn run_stream(
if is_shutdown_requested(shutdown_requested) { if is_shutdown_requested(shutdown_requested) {
return Ok(()); return Ok(());
} }
let config = ManagedClientConfig {
client: args.to_client_config(),
pacing: args.pacing.into(),
completion: ManagedCompletionPolicy::FinishWhenQuiescent,
event_capacity: MANAGED_EVENT_CAPACITY,
outcome_history_limit: target_count,
max_live_target_generations: target_count,
..ManagedClientConfig::default()
};
let (owner, mut events) = BlockingManagedClient::start_with_subscription( let (owner, mut events) = BlockingManagedClient::start_with_subscription(
config, setup.managed_config(),
targets setup.managed_targets(),
.iter()
.map(|target| target.managed.clone())
.collect(),
)?; )?;
let handle = owner.handle(); let handle = owner.handle();
let mut stdout = io::LineWriter::new(io::stdout().lock()); let mut stdout = io::LineWriter::new(io::stdout().lock());
@@ -101,7 +87,8 @@ pub fn run_stream(
out: &mut stdout, out: &mut stdout,
}; };
let mut stats = targets let mut stats = setup
.targets
.iter() .iter()
.map(|target| { .map(|target| {
( (
+30 -40
View File
@@ -1,13 +1,12 @@
use std::time::Duration; use std::time::Duration;
use clap::Parser; use clap::Parser;
use irtt_client::ClientConfig;
#[cfg(test)] #[cfg(test)]
use crate::shared::client::TimestampArg; use crate::shared::client::TimestampArg;
use crate::shared::client::{ use crate::shared::client::{
parse_labelled_target, parse_test_duration, prepare_managed_targets, target_specs, parse_labelled_target, parse_test_duration, prepare_managed_run, CommonClientArgs,
CommonClientArgs, GroupPacingArg, LabelledTargetArg, PreparedTarget, TargetSpec, GroupPacingArg, LabelledTargetArg, ManagedRunSetup, TargetSelection,
}; };
pub const DEFAULT_TUI_DURATION: Duration = Duration::ZERO; pub const DEFAULT_TUI_DURATION: Duration = Duration::ZERO;
@@ -56,11 +55,19 @@ pub struct TuiArgs {
} }
impl TuiArgs { impl TuiArgs {
pub fn to_client_config(&self) -> ClientConfig { /// Validate the selected targets and prepare the managed run.
self.common.to_client_config( ///
self.primary_target_addr() /// The TUI always requires a target: the parser rejects an empty target
.expect("at least one target is required"), /// list, and this rejects a target set that cannot be labelled uniquely.
pub fn prepare(&self) -> Result<ManagedRunSetup, String> {
prepare_managed_run(
&self.common,
self.duration, self.duration,
TargetSelection {
positional: &self.targets,
labelled: &self.labelled_targets,
pacing: self.pacing,
},
) )
} }
@@ -68,31 +75,12 @@ impl TuiArgs {
self.duration == Duration::ZERO self.duration == Duration::ZERO
} }
pub fn target_specs(&self) -> Result<Vec<TuiTargetSpec>, String> {
target_specs(&self.targets, &self.labelled_targets)
}
pub fn managed_targets(&self) -> Result<Vec<TuiManagedTarget>, String> {
prepare_managed_targets(self.target_specs()?)
}
#[cfg(test)] #[cfg(test)]
pub fn timestamp_mode(&self) -> TimestampArg { pub fn timestamp_mode(&self) -> TimestampArg {
self.common.tstamp self.common.tstamp
} }
fn primary_target_addr(&self) -> Option<&str> {
self.targets.first().map(String::as_str).or_else(|| {
self.labelled_targets
.first()
.map(|target| target.addr.as_str())
})
}
} }
pub type TuiTargetSpec = TargetSpec;
pub type TuiManagedTarget = PreparedTarget;
impl std::ops::Deref for TuiArgs { impl std::ops::Deref for TuiArgs {
type Target = CommonClientArgs; type Target = CommonClientArgs;
@@ -132,12 +120,12 @@ mod tests {
assert_eq!(args.pacing, GroupPacingArg::Staggered); assert_eq!(args.pacing, GroupPacingArg::Staggered);
assert_eq!(args.duration, DEFAULT_TUI_DURATION); assert_eq!(args.duration, DEFAULT_TUI_DURATION);
assert!(args.is_continuous()); assert!(args.is_continuous());
assert_eq!(args.to_client_config().duration, None); assert_eq!(args.prepare().unwrap().client.duration, None);
let finite = parse(&["--duration", "30s", "127.0.0.1:2112"]).unwrap(); let finite = parse(&["--duration", "30s", "127.0.0.1:2112"]).unwrap();
assert_eq!(finite.duration, Duration::from_secs(30)); assert_eq!(finite.duration, Duration::from_secs(30));
assert_eq!( assert_eq!(
finite.to_client_config().duration, finite.prepare().unwrap().client.duration,
Some(Duration::from_secs(30)) Some(Duration::from_secs(30))
); );
@@ -149,12 +137,12 @@ mod tests {
#[test] #[test]
fn multiple_positional_targets_parse() { fn multiple_positional_targets_parse() {
let args = parse(&["host-a:2112", "host-b:2112"]).unwrap(); let args = parse(&["host-a:2112", "host-b:2112"]).unwrap();
let specs = args.target_specs().unwrap(); let specs = args.prepare().unwrap().targets;
assert_eq!(specs[0].label, "host-a:2112"); assert_eq!(specs[0].label, "host-a:2112");
assert_eq!(specs[0].addr, "host-a:2112"); assert_eq!(specs[0].managed.server_addr, "host-a:2112");
assert_eq!(specs[1].label, "host-b:2112"); assert_eq!(specs[1].label, "host-b:2112");
assert_eq!(specs[1].addr, "host-b:2112"); assert_eq!(specs[1].managed.server_addr, "host-b:2112");
} }
#[test] #[test]
@@ -166,23 +154,26 @@ mod tests {
"sg=sg.example.com:2112", "sg=sg.example.com:2112",
]) ])
.unwrap(); .unwrap();
let specs = args.target_specs().unwrap(); let specs = args.prepare().unwrap().targets;
assert_eq!(specs[0].label, "ams"); assert_eq!(specs[0].label, "ams");
assert_eq!(specs[0].addr, "ams.example.com:2112"); assert_eq!(specs[0].managed.server_addr, "ams.example.com:2112");
assert_eq!(specs[1].label, "sg"); assert_eq!(specs[1].label, "sg");
assert_eq!(specs[1].addr, "sg.example.com:2112"); assert_eq!(specs[1].managed.server_addr, "sg.example.com:2112");
} }
#[test] #[test]
fn at_least_one_target_is_required() { fn at_least_one_target_is_required() {
// The TUI has no target-free mode, so this is rejected at parse time
// rather than during preparation.
assert!(parse(&[]).is_err()); assert!(parse(&[]).is_err());
assert!(parse(&["--target", "ams=ams.example.com:2112"]).is_ok());
} }
#[test] #[test]
fn duplicate_labels_are_rejected() { fn duplicate_labels_are_rejected() {
let args = parse(&["host-a:2112", "--target", "host-a:2112=host-b:2112"]).unwrap(); let args = parse(&["host-a:2112", "--target", "host-a:2112=host-b:2112"]).unwrap();
let err = args.target_specs().unwrap_err(); let err = args.prepare().unwrap_err();
assert!(err.contains("duplicate target label")); assert!(err.contains("duplicate target label"));
} }
@@ -190,7 +181,7 @@ mod tests {
#[test] #[test]
fn duplicate_positional_target_strings_get_stable_suffixes() { fn duplicate_positional_target_strings_get_stable_suffixes() {
let args = parse(&["host-a:2112", "host-a:2112"]).unwrap(); let args = parse(&["host-a:2112", "host-a:2112"]).unwrap();
let specs = args.target_specs().unwrap(); let specs = args.prepare().unwrap().targets;
assert_eq!(specs[0].label, "host-a:2112"); assert_eq!(specs[0].label, "host-a:2112");
assert_eq!(specs[1].label, "host-a:2112#2"); assert_eq!(specs[1].label, "host-a:2112#2");
@@ -199,7 +190,7 @@ mod tests {
#[test] #[test]
fn duplicate_target_endpoints_are_allowed() { fn duplicate_target_endpoints_are_allowed() {
let args = parse(&["127.0.0.1:2112", "127.0.0.1"]).unwrap(); let args = parse(&["127.0.0.1:2112", "127.0.0.1"]).unwrap();
assert_eq!(args.managed_targets().unwrap().len(), 2); assert_eq!(args.prepare().unwrap().target_count(), 2);
} }
#[test] #[test]
@@ -272,11 +263,10 @@ mod tests {
"--loose", "--loose",
"127.0.0.1:2112", "127.0.0.1:2112",
]; ];
let tui = parse(&tui_args).unwrap().to_client_config(); let tui = parse(&tui_args).unwrap().prepare().unwrap().client;
let common = parse_shared(&shared).unwrap().common; let common = parse_shared(&shared).unwrap().common;
let shared = common.to_client_config("127.0.0.1:2112", Duration::from_secs(30)); let shared = common.to_client_config(Duration::from_secs(30));
assert_eq!(shared.server_addr, tui.server_addr);
assert_eq!(shared.duration, tui.duration); assert_eq!(shared.duration, tui.duration);
assert_eq!(shared.interval, tui.interval); assert_eq!(shared.interval, tui.interval);
assert_eq!(shared.length, tui.length); assert_eq!(shared.length, tui.length);
+8 -22
View File
@@ -19,27 +19,26 @@ use crate::{
}; };
use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers}; use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers};
use irtt_client::managed::{ use irtt_client::managed::{
BlockingManagedClient, ManagedClientConfig, ManagedCompletionPolicy, ManagedEndReason, BlockingManagedClient, ManagedEndReason, ManagedEvent, ManagedEventSubscription,
ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError, TargetInstance, ManagedEventTryRecvError, TargetInstance,
}; };
const RENDER_INTERVAL: Duration = Duration::from_millis(250); const RENDER_INTERVAL: Duration = Duration::from_millis(250);
const TUI_WAIT_SLICE: Duration = Duration::from_millis(20); const TUI_WAIT_SLICE: Duration = Duration::from_millis(20);
const MANAGED_EVENT_CAPACITY: usize = 16_384;
const INPUT_EVENT_WORK_BUDGET: usize = 128; const INPUT_EVENT_WORK_BUDGET: usize = 128;
pub fn run_tui( pub fn run_tui(
args: TuiArgs, args: TuiArgs,
shutdown_requested: &AtomicBool, shutdown_requested: &AtomicBool,
) -> Result<(), Box<dyn std::error::Error>> { ) -> Result<(), Box<dyn std::error::Error>> {
let targets = args let setup = args
.managed_targets() .prepare()
.map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?; .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?;
let continuous = args.is_continuous(); let continuous = args.is_continuous();
let mut terminal = TuiTerminal::enter()?; let mut terminal = TuiTerminal::enter()?;
let mut state = TuiState::with_target_labels( let mut state = TuiState::with_target_labels(
TuiConfig::from_args(&args), TuiConfig::from_args(&args, &setup),
targets.iter().map(|target| target.label.clone()), setup.targets.iter().map(|target| target.label.clone()),
); );
let mut next_render = Instant::now(); let mut next_render = Instant::now();
if is_shutdown_requested(shutdown_requested) { if is_shutdown_requested(shutdown_requested) {
@@ -47,22 +46,9 @@ pub fn run_tui(
} }
state.set_status(TuiStatus::Opening); state.set_status(TuiStatus::Opening);
render_if_due(&mut terminal, &state, &mut next_render, true)?; render_if_due(&mut terminal, &state, &mut next_render, true)?;
let target_count = targets.len();
let config = ManagedClientConfig {
client: args.to_client_config(),
pacing: args.pacing.into(),
completion: ManagedCompletionPolicy::FinishWhenQuiescent,
event_capacity: MANAGED_EVENT_CAPACITY,
outcome_history_limit: target_count,
max_live_target_generations: target_count,
..ManagedClientConfig::default()
};
let (owner, mut events) = match BlockingManagedClient::start_with_subscription( let (owner, mut events) = match BlockingManagedClient::start_with_subscription(
config, setup.managed_config(),
targets setup.managed_targets(),
.iter()
.map(|target| target.managed.clone())
.collect(),
) { ) {
Ok(value) => value, Ok(value) => value,
Err(error) => { Err(error) => {
+3 -3
View File
@@ -30,7 +30,7 @@ use crate::{
}, },
tui::args::TuiArgs, tui::args::TuiArgs,
}, },
shared::client::{expected_probe_count, GroupPacingArg}, shared::client::{expected_probe_count, GroupPacingArg, ManagedRunSetup},
}; };
const HISTORY_LIMIT: usize = 100_000; const HISTORY_LIMIT: usize = 100_000;
@@ -656,11 +656,11 @@ pub(super) struct TuiConfig {
} }
impl TuiConfig { impl TuiConfig {
pub(super) fn from_args(args: &TuiArgs) -> Self { pub(super) fn from_args(args: &TuiArgs, setup: &ManagedRunSetup) -> Self {
Self { Self {
interval: args.interval, interval: args.interval,
duration: (!args.is_continuous()).then_some(args.duration), duration: (!args.is_continuous()).then_some(args.duration),
timeout: args.to_client_config().probe_timeout, timeout: setup.client.probe_timeout,
target_probes: (!args.is_continuous()) target_probes: (!args.is_continuous())
.then(|| expected_probe_count(args.duration, args.interval)), .then(|| expected_probe_count(args.duration, args.interval)),
pacing: args.pacing, pacing: args.pacing,
+6 -2
View File
@@ -7,9 +7,13 @@ use super::args::CommonClientArgs;
pub const DEFAULT_RECV_TIMEOUT: Duration = Duration::from_millis(20); pub const DEFAULT_RECV_TIMEOUT: Duration = Duration::from_millis(20);
impl CommonClientArgs { impl CommonClientArgs {
pub fn to_client_config(&self, server: &str, duration: Duration) -> ClientConfig { /// Build the shared client configuration template for a managed run.
///
/// The template carries no server address: the managed driver replaces it
/// per target from that target's configuration, so deriving one here from
/// an arbitrary target would be dead data.
pub fn to_client_config(&self, duration: Duration) -> ClientConfig {
ClientConfig { ClientConfig {
server_addr: server.to_owned(),
duration: (!duration.is_zero()).then_some(duration), duration: (!duration.is_zero()).then_some(duration),
interval: self.interval, interval: self.interval,
length: self.length, length: self.length,
+2
View File
@@ -1,5 +1,6 @@
pub mod args; pub mod args;
pub mod config; pub mod config;
pub mod prepare;
pub mod session; pub mod session;
pub mod targets; pub mod targets;
@@ -8,6 +9,7 @@ pub use args::{
ClockArg, CommonClientArgs, ReceivedStatsArg, TimestampArg, ClockArg, CommonClientArgs, ReceivedStatsArg, TimestampArg,
}; };
pub use config::{expected_probe_count, DEFAULT_RECV_TIMEOUT}; pub use config::{expected_probe_count, DEFAULT_RECV_TIMEOUT};
pub use prepare::{prepare_managed_run, ManagedRunSetup, TargetSelection, MANAGED_EVENT_CAPACITY};
pub use session::is_shutdown_requested; pub use session::is_shutdown_requested;
pub use targets::{ pub use targets::{
parse_labelled_target, prepare_managed_targets, target_specs, GroupPacingArg, parse_labelled_target, prepare_managed_targets, target_specs, GroupPacingArg,
@@ -0,0 +1,258 @@
//! One preparation step from parsed arguments to a runnable managed setup.
//!
//! Both client applets accept the same target and pacing arguments and then
//! have to turn them into the same three things: a validated target set, one
//! shared [`ClientConfig`] template, and a [`ManagedPacing`]. Doing that
//! separately in each applet is how the CLI ended up deriving its client
//! configuration from an arbitrary "primary" target whose address the managed
//! driver immediately replaced per target.
use std::time::Duration;
use irtt_client::{
managed::{ManagedClientConfig, ManagedCompletionPolicy, ManagedPacing},
ClientConfig,
};
use super::{
args::CommonClientArgs,
targets::{prepare_managed_targets, target_specs, GroupPacingArg, LabelledTargetArg},
PreparedTarget,
};
/// Capacity of the lossy managed presentation event channel.
pub const MANAGED_EVENT_CAPACITY: usize = 16_384;
/// Target and pacing arguments common to the client applets.
///
/// Duration is deliberately absent: the applets disagree about what a duration
/// means and what its default is, so each keeps its own.
#[derive(Debug, Clone, Copy)]
pub struct TargetSelection<'a> {
pub positional: &'a [String],
pub labelled: &'a [LabelledTargetArg],
pub pacing: GroupPacingArg,
}
/// Everything a managed run needs, prepared and validated together.
#[derive(Debug, Clone)]
pub struct ManagedRunSetup {
/// Validated targets, in the order the user gave them.
pub targets: Vec<PreparedTarget>,
/// Shared client configuration template.
///
/// The managed driver supplies each target's address and authentication
/// from its [`ManagedTargetConfig`](irtt_client::managed::ManagedTargetConfig),
/// so this template carries no target of its own.
pub client: ClientConfig,
/// Send coordination across active targets.
pub pacing: ManagedPacing,
}
impl ManagedRunSetup {
/// Number of validated targets.
pub fn target_count(&self) -> usize {
self.targets.len()
}
/// Whether this run drives more than one target.
pub fn is_multi_target(&self) -> bool {
self.targets.len() > 1
}
/// Managed target configurations, in argument order.
pub fn managed_targets(&self) -> Vec<irtt_client::managed::ManagedTargetConfig> {
self.targets
.iter()
.map(|target| target.managed.clone())
.collect()
}
/// Managed driver configuration for this run.
pub fn managed_config(&self) -> ManagedClientConfig {
let target_count = self.target_count();
ManagedClientConfig {
client: self.client.clone(),
pacing: self.pacing,
completion: ManagedCompletionPolicy::FinishWhenQuiescent,
event_capacity: MANAGED_EVENT_CAPACITY,
outcome_history_limit: target_count,
max_live_target_generations: target_count,
..ManagedClientConfig::default()
}
}
}
/// Validate the selected targets and build the shared run setup.
///
/// Targets are validated before any configuration is built, so a bad target set
/// fails before the caller can act on a half-prepared run.
pub fn prepare_managed_run(
common: &CommonClientArgs,
duration: Duration,
selection: TargetSelection<'_>,
) -> Result<ManagedRunSetup, String> {
let specs = target_specs(selection.positional, selection.labelled)?;
let targets = prepare_managed_targets(specs)?;
Ok(ManagedRunSetup {
targets,
client: common.to_client_config(duration),
pacing: selection.pacing.into(),
})
}
#[cfg(test)]
mod tests {
use clap::Parser;
use super::*;
#[derive(Parser)]
struct CommonOnlyArgs {
#[command(flatten)]
common: CommonClientArgs,
}
fn common(args: &[&str]) -> CommonClientArgs {
let mut argv = vec!["common-only"];
argv.extend_from_slice(args);
CommonOnlyArgs::try_parse_from(argv).unwrap().common
}
fn prepare(
positional: &[&str],
labelled: &[LabelledTargetArg],
pacing: GroupPacingArg,
duration: Duration,
) -> Result<ManagedRunSetup, String> {
let positional: Vec<String> = positional.iter().map(|value| (*value).to_owned()).collect();
prepare_managed_run(
&common(&[]),
duration,
TargetSelection {
positional: &positional,
labelled,
pacing,
},
)
}
#[test]
fn a_single_target_prepares_one_managed_target_and_a_finite_duration() {
let setup = prepare(
&["127.0.0.1:2112"],
&[],
GroupPacingArg::Staggered,
Duration::from_secs(30),
)
.unwrap();
assert_eq!(setup.target_count(), 1);
assert!(!setup.is_multi_target());
assert_eq!(setup.targets[0].label, "127.0.0.1:2112");
assert_eq!(setup.targets[0].managed.server_addr, "127.0.0.1:2112");
assert_eq!(setup.client.duration, Some(Duration::from_secs(30)));
assert_eq!(setup.pacing, ManagedPacing::Staggered);
}
#[test]
fn a_zero_duration_prepares_a_continuous_client_config() {
let setup = prepare(
&["127.0.0.1:2112"],
&[],
GroupPacingArg::Staggered,
Duration::ZERO,
)
.unwrap();
assert_eq!(setup.client.duration, None);
}
#[test]
fn multiple_targets_keep_argument_order_and_share_one_client_config() {
let labelled = vec![LabelledTargetArg {
label: "sg".to_owned(),
addr: "sg.example.test:2112".to_owned(),
}];
let setup = prepare(
&["host-a:2112", "host-b:2112"],
&labelled,
GroupPacingArg::Burst,
Duration::from_secs(10),
)
.unwrap();
assert!(setup.is_multi_target());
let labels: Vec<_> = setup
.targets
.iter()
.map(|target| target.label.as_str())
.collect();
assert_eq!(labels, ["host-a:2112", "host-b:2112", "sg"]);
assert_eq!(setup.targets[2].managed.server_addr, "sg.example.test:2112");
assert_eq!(setup.pacing, ManagedPacing::Burst);
let config = setup.managed_config();
assert_eq!(config.outcome_history_limit, 3);
assert_eq!(config.max_live_target_generations, 3);
assert_eq!(config.event_capacity, MANAGED_EVENT_CAPACITY);
assert_eq!(config.pacing, ManagedPacing::Burst);
assert_eq!(config.client, setup.client);
assert_eq!(setup.managed_targets().len(), 3);
}
#[test]
fn the_shared_client_config_carries_no_target_of_its_own() {
// The managed driver replaces the address, and the CLI's key, per
// target, so the template must not pin either to one target.
let positional = vec!["host-a:2112".to_owned(), "host-b:2112".to_owned()];
let setup = prepare_managed_run(
&common(&["--hmac", "secret"]),
Duration::from_secs(10),
TargetSelection {
positional: &positional,
labelled: &[],
pacing: GroupPacingArg::Staggered,
},
)
.unwrap();
assert_eq!(
setup.client.server_addr,
ClientConfig::default().server_addr
);
assert_eq!(setup.client.hmac_key, Some(b"secret".to_vec()));
// Per-target authentication is not exposed on the command line, so no
// target carries an override of the shared key.
assert!(setup
.targets
.iter()
.all(|target| target.managed.auth.is_none()));
}
#[test]
fn an_empty_target_set_is_rejected_with_the_list_columns_hint() {
let err =
prepare(&[], &[], GroupPacingArg::Staggered, Duration::from_secs(10)).unwrap_err();
assert!(err.contains("at least one target is required"));
assert!(err.contains("--list-columns"));
}
#[test]
fn a_duplicate_label_is_rejected_before_any_configuration_is_built() {
let labelled = vec![LabelledTargetArg {
label: "host-a:2112".to_owned(),
addr: "host-b:2112".to_owned(),
}];
let err = prepare(
&["host-a:2112"],
&labelled,
GroupPacingArg::Staggered,
Duration::from_secs(10),
)
.unwrap_err();
assert!(err.contains("duplicate target label"));
}
}