From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [45.144.208.40]) by lore.proxmox.com (Postfix) with ESMTPS id 5E15B1FF0A8 for ; Thu, 20 Aug 2026 16:53:08 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id 439E421620; Thu, 20 Aug 2026 16:52:49 +0200 (CEST) From: Lukas Wagner To: pdm-devel@lists.proxmox.com Subject: [PATCH datacenter-manager v2 11/20] parallel fetcher: pass arguments to closure in a single type Date: Thu, 20 Aug 2026 16:52:11 +0200 Message-ID: <20260820145220.418032-12-l.wagner@proxmox.com> X-Mailer: git-send-email 2.47.3 In-Reply-To: <20260820145220.418032-1-l.wagner@proxmox.com> References: <20260820145220.418032-1-l.wagner@proxmox.com> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1787237519167 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.704 Adjusted score from AWL reputation of From: address DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment (newer systems) RCVD_IN_DNSWL_MED -2.3 Sender listed at https://www.dnswl.org/, medium trust SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: 4LFEKFGH45CXADTMEBF6CSGNYGGHURWR X-Message-ID-Hash: 4LFEKFGH45CXADTMEBF6CSGNYGGHURWR X-MailFrom: l.wagner@proxmox.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; loop; banned-address; emergency; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header X-Mailman-Version: 3.3.10 Precedence: list List-Id: Proxmox Datacenter Manager development discussion List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: This makes it much easier to pass more data later, e.g. a client factory. Signed-off-by: Lukas Wagner --- server/src/api/pve/firewall.rs | 31 +++++++++----------- server/src/api/sdn/controllers.rs | 6 ++-- server/src/api/sdn/vnets.rs | 6 ++-- server/src/api/sdn/zones.rs | 6 ++-- server/src/parallel_fetcher.rs | 39 ++++++++++++++++++------- server/src/remote_tasks/refresh_task.rs | 23 +++++++-------- server/src/remote_updates.rs | 7 +++-- 7 files changed, 69 insertions(+), 49 deletions(-) diff --git a/server/src/api/pve/firewall.rs b/server/src/api/pve/firewall.rs index b381a14e..dc6c97b5 100644 --- a/server/src/api/pve/firewall.rs +++ b/server/src/api/pve/firewall.rs @@ -19,7 +19,7 @@ use pdm_api_types::{PRIV_RESOURCE_AUDIT, PRIV_RESOURCE_MODIFY, PRIV_SYS_MODIFY}; use super::{connect_to_remote_by_id, find_node_for_vm}; use crate::api::remotes::RemoteIterator; use crate::connection::PveClient; -use crate::parallel_fetcher::ParallelFetcher; +use crate::parallel_fetcher::{ParallelFetcher, ParallelFetcherArgs}; // top-level firewall routers pub const PVE_FW_ROUTER: Router = Router::new() @@ -112,11 +112,9 @@ struct ClusterFirewallData { } async fn fetch_cluster_firewall_data( - _context: (), - remote: Remote, - _node: String, // unused for cluster-level data + args: ParallelFetcherArgs<()>, ) -> Result { - let pve = crate::connection::make_pve_client(&remote)?; + let pve = crate::connection::make_pve_client(&args.remote)?; let guests = match pve.cluster_resources(Some(ClusterResourceKind::Vm)).await { Ok(guests) => guests, @@ -204,14 +202,12 @@ async fn load_guests_firewall_status( } async fn fetch_node_firewall_status( - context: FirewallFetchContext, - remote: Remote, - node: String, + args: ParallelFetcherArgs, ) -> Result { - let pve = crate::connection::make_pve_client(&remote)?; + let pve = crate::connection::make_pve_client(&args.remote)?; - let options_response = pve.node_firewall_options(&node); - let rules_response = pve.list_node_firewall_rules(&node); + let options_response = pve.node_firewall_options(&args.node); + let rules_response = pve.list_node_firewall_rules(&args.node); let enabled = options_response .await @@ -227,10 +223,11 @@ async fn fetch_node_firewall_status( _ => None, }; - let guests_status = load_guests_firewall_status(pve, node.clone(), &context.guests).await; + let guests_status = + load_guests_firewall_status(pve, args.node.clone(), &args.context.guests).await; Ok(NodeFirewallStatus { - node, + node: args.node, status, guests: guests_status, }) @@ -280,11 +277,11 @@ pub async fn pve_firewall_status( let node_fetcher = ParallelFetcher::new(context); let node_results = node_fetcher - .do_for_all_remote_nodes(pve_remotes.iter().cloned(), move |mut ctx, remote, node| { - if let Some(guests) = guests_per_remote.get(&remote.id) { - ctx.guests = guests.clone(); + .do_for_all_remote_nodes(pve_remotes.iter().cloned(), move |mut args| { + if let Some(guests) = guests_per_remote.get(&args.remote.id) { + args.context.guests = guests.clone(); } - fetch_node_firewall_status(ctx, remote, node) + fetch_node_firewall_status(args) }) .await; diff --git a/server/src/api/sdn/controllers.rs b/server/src/api/sdn/controllers.rs index 060fef72..049358cd 100644 --- a/server/src/api/sdn/controllers.rs +++ b/server/src/api/sdn/controllers.rs @@ -86,9 +86,9 @@ pub async fn list_controllers( let fetcher = ParallelFetcher::new((pending, running, ty)); let results = fetcher - .do_for_all_remotes(iter.into_remotes(), async |ctx, r, _| { - Ok(pve::connect(&r)? - .list_controllers(ctx.0, ctx.1, ctx.2) + .do_for_all_remotes(iter.into_remotes(), async |args| { + Ok(pve::connect(&args.remote)? + .list_controllers(args.context.0, args.context.1, args.context.2) .await?) }) .await; diff --git a/server/src/api/sdn/vnets.rs b/server/src/api/sdn/vnets.rs index 30d431bc..8e7ae52b 100644 --- a/server/src/api/sdn/vnets.rs +++ b/server/src/api/sdn/vnets.rs @@ -87,8 +87,10 @@ async fn list_vnets( let fetcher = ParallelFetcher::new((pending, running)); let results = fetcher - .do_for_all_remotes(iter.into_remotes(), async |ctx, r, _| { - Ok(pve::connect(&r)?.list_vnets(ctx.0, ctx.1).await?) + .do_for_all_remotes(iter.into_remotes(), async |args| { + Ok(pve::connect(&args.remote)? + .list_vnets(args.context.0, args.context.1) + .await?) }) .await; diff --git a/server/src/api/sdn/zones.rs b/server/src/api/sdn/zones.rs index d37dc46c..bb7a3822 100644 --- a/server/src/api/sdn/zones.rs +++ b/server/src/api/sdn/zones.rs @@ -93,8 +93,10 @@ pub async fn list_zones( let fetcher = ParallelFetcher::new((pending, running, ty)); let results = fetcher - .do_for_all_remotes(iter.into_remotes(), async |ctx, r, _| { - Ok(pve::connect(&r)?.list_zones(ctx.0, ctx.1, ctx.2).await?) + .do_for_all_remotes(iter.into_remotes(), async |args| { + Ok(pve::connect(&args.remote)? + .list_zones(args.context.0, args.context.1, args.context.2) + .await?) }) .await; diff --git a/server/src/parallel_fetcher.rs b/server/src/parallel_fetcher.rs index 8153f1fe..0819eb86 100644 --- a/server/src/parallel_fetcher.rs +++ b/server/src/parallel_fetcher.rs @@ -4,18 +4,16 @@ //! # use anyhow::Error; //! # //! # use pdm_api_types::remotes::{RemoteType, Remote}; -//! # use server::parallel_fetcher::ParallelFetcher; +//! # use server::parallel_fetcher::{ParallelFetcher, ParallelFetcherArgs}; //! # //! # #[tokio::main] //! # async fn main() -> Result<(), Error> { //! # let remotes: Vec = Vec::new(); //! # //! async fn fetch_meaning( -//! _context: (), -//! remote: Remote, -//! node: String, +//! args: ParallelFetcherArgs<()>, //! ) -> Result { -//! match remote.ty { +//! match args.remote.ty { //! RemoteType::Pve => { //! // Perform the API request here and return some result. //! Ok(42) @@ -269,6 +267,20 @@ impl ParallelFetcherBuilder { } } +#[non_exhaustive] +/// The argument type that is passed when calling the closure passed to +/// [`ParallelFetcher::do_for_all_remote_nodes`] or [`ParallelFetcher::do_for_all_remotes`]. +pub struct ParallelFetcherArgs { + /// The context provided in [`ParallelFetcherBuilder::new`] or + /// [`ParallelFetcher::new`]. + pub context: C, + /// The remote. + pub remote: Remote, + /// The node. This may be 'localhost' for PBS remotes or if using + /// [`ParallelFetcher::do_for_all_remotes`]. + pub node: String, +} + /// Helper for parallelizing API requests to multiple remotes/nodes. pub struct ParallelFetcher { max_connections: usize, @@ -295,7 +307,7 @@ impl ParallelFetcher { ) -> FetcherResponse> where A: Iterator, - F: Fn(C, Remote, String) -> Ft + Clone + Send + 'static, + F: Fn(ParallelFetcherArgs) -> Ft + Clone + Send + 'static, Ft: Future> + Send + 'static, T: Send + Debug + 'static, { @@ -346,7 +358,7 @@ impl ParallelFetcher { max_connections_per_remote: usize, ) -> RemoteResponse> where - F: Fn(C, Remote, String) -> Ft + Clone + Send + 'static, + F: Fn(ParallelFetcherArgs) -> Ft + Clone + Send + 'static, Ft: Future> + Send + 'static, T: Send + Debug + 'static, { @@ -456,12 +468,19 @@ impl ParallelFetcher { _per_remote_connections_permit: Option, ) -> NodeResponse where - F: Fn(C, Remote, String) -> Ft + Clone + Send + 'static, + F: Fn(ParallelFetcherArgs) -> Ft + Clone + Send + 'static, Ft: Future> + Send + 'static, T: Send + Debug + 'static, { let now = Instant::now(); - let result = func(context, remote.clone(), node.clone()).await; + + let parallel_fetcher_context = ParallelFetcherArgs { + context, + remote, + node: node.clone(), + }; + + let result = func(parallel_fetcher_context).await; let api_response_time = now.elapsed(); NodeResponse { @@ -479,7 +498,7 @@ impl ParallelFetcher { ) -> FetcherResponse> where A: Iterator, - F: Fn(C, Remote, String) -> Ft + Clone + Send + 'static, + F: Fn(ParallelFetcherArgs) -> Ft + Clone + Send + 'static, Ft: Future> + Send + 'static, T: Send + Debug + 'static, { diff --git a/server/src/remote_tasks/refresh_task.rs b/server/src/remote_tasks/refresh_task.rs index 668b63e4..c5d25284 100644 --- a/server/src/remote_tasks/refresh_task.rs +++ b/server/src/remote_tasks/refresh_task.rs @@ -11,7 +11,7 @@ use proxmox_section_config::typed::SectionConfigData; use crate::api; use crate::connection; -use crate::parallel_fetcher::ParallelFetcher; +use crate::parallel_fetcher::{ParallelFetcher, ParallelFetcherArgs}; use crate::pbs_client; use crate::remote_tasks::{ KEEP_OLD_FILES, ROTATE_AFTER, @@ -277,17 +277,16 @@ async fn fetch_remotes( } async fn fetch_tasks_from_single_node( - context: Arc, - remote: Remote, - node: String, + args: ParallelFetcherArgs>, ) -> Result, Error> { - let since = context - .cutoff_timestamp(&remote.id, &node) + let since = args + .context + .cutoff_timestamp(&args.remote.id, &args.node) .unwrap_or_else(|| { proxmox_time::epoch_i64() - (KEEP_OLD_FILES as u64 * ROTATE_AFTER) as i64 }); - match remote.ty { + match args.remote.ty { RemoteType::Pve => { let params = pve_api_types::ListTasks { source: Some(pve_api_types::ListTasksSource::All), @@ -297,13 +296,13 @@ async fn fetch_tasks_from_single_node( ..Default::default() }; - let client = connection::make_pve_client(&remote)?; + let client = connection::make_pve_client(&args.remote)?; let task_list = client - .get_task_list(&node, params) + .get_task_list(&args.node, params) .await? .into_iter() - .map(|task| map_pve_task(task, remote.id.clone())) + .map(|task| map_pve_task(task, args.remote.id.clone())) .collect(); Ok(task_list) @@ -315,13 +314,13 @@ async fn fetch_tasks_from_single_node( limit: Some(MAX_TASKS_TO_FETCH), }; - let client = connection::make_pbs_client(&remote)?; + let client = connection::make_pbs_client(&args.remote)?; let task_list = client .get_task_list(params) .await? .into_iter() - .map(|task| map_pbs_task(task, remote.id.clone())) + .map(|task| map_pbs_task(task, args.remote.id.clone())) .collect(); Ok(task_list) diff --git a/server/src/remote_updates.rs b/server/src/remote_updates.rs index 855d9507..83b5a68f 100644 --- a/server/src/remote_updates.rs +++ b/server/src/remote_updates.rs @@ -232,14 +232,15 @@ pub async fn refresh_update_summary_cache(remotes: Vec) -> Result<(), Er let fetcher = ParallelFetcher::new(()); let fetch_response = fetcher - .do_for_all_remote_nodes(remotes.into_iter(), |context, remote, node| async move { - let result = fetch_available_updates(context, remote.clone(), node.clone()).await; + .do_for_all_remote_nodes(remotes.into_iter(), |args| async move { + let result = fetch_available_updates((), args.remote.clone(), args.node.clone()).await; let summary = match &result { Ok(update_info) => update_info.into(), Err(err) => node_error_summary(err), }; - if let Err(err) = update_cached_summary_for_node(remote, node, summary).await { + if let Err(err) = update_cached_summary_for_node(args.remote, args.node, summary).await + { log::error!("could not update 'remote-updates' API cache entry: {err}"); } -- 2.47.3