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 2C4331FF0E1 for ; Thu, 27 Aug 2026 13:43:49 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id 9A13021670; Thu, 27 Aug 2026 13:43:16 +0200 (CEST) From: Lukas Wagner To: pdm-devel@lists.proxmox.com Subject: [PATCH datacenter-manager v3 14/21] server: migrate existing ParallelFetcher users to use PdmApplication Date: Thu, 27 Aug 2026 13:42:37 +0200 Message-ID: <20260827114244.424784-15-l.wagner@proxmox.com> X-Mailer: git-send-email 2.47.3 In-Reply-To: <20260827114244.424784-1-l.wagner@proxmox.com> References: <20260827114244.424784-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: 1787830961301 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.566 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: 54KUPWDV3Z5XTGFL3FVNXE4V2JFYIFBR X-Message-ID-Hash: 54KUPWDV3Z5XTGFL3FVNXE4V2JFYIFBR 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: ParallelFetcher now requires a PdmApplication in both it's `new` and `builder` constructions. The provided application context object is later provided via ParallelFetcherArgs to the closure. For now, this is mostly used to give access to the client factory. The mandatory argument should gently nudge the developer towards using the #[state] injected handle, with the option to use context::pdm_application() as fallback, if too much refactoring is needed on the spot. Introducing this new mandatory parameter led to a small cascade of changes in existing users of ParallelFetcher. These were changed to fully use `app` everywhere they can, such as when reading remote config. This makes the commit slightly bigger, but avoids any awkward 'in-between' states where only part of a module uses PdmApplication. Signed-off-by: Lukas Wagner --- Notes: Changes since v2: - Use new #[state] attribute - Let ParallelFetcher have PdmApplication, not only ClientFactory - Merged two previous commits into this one: https://lore.proxmox.com/pdm-devel/20260820145220.418032-14-l.wagner@proxmox.com/T/#u https://lore.proxmox.com/pdm-devel/DKX7UQ5AWO68.3SIX8MRGM180I@proxmox.com/T/#mb0e73f6906c23766b6075751f2056dc86beda8fc server/src/api/pve/firewall.rs | 25 +++++++++----- server/src/api/sdn/controllers.rs | 13 +++++--- server/src/api/sdn/vnets.rs | 14 +++++--- server/src/api/sdn/zones.rs | 13 +++++--- server/src/parallel_fetcher.rs | 44 ++++++++++++++++++++----- server/src/remote_tasks/refresh_task.rs | 4 +-- server/src/remote_updates.rs | 4 +-- 7 files changed, 83 insertions(+), 34 deletions(-) diff --git a/server/src/api/pve/firewall.rs b/server/src/api/pve/firewall.rs index d0f2d6d1..29fb3dce 100644 --- a/server/src/api/pve/firewall.rs +++ b/server/src/api/pve/firewall.rs @@ -19,6 +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::context::PdmApplication; use crate::parallel_fetcher::{ParallelFetcher, ParallelFetcherArgs}; // top-level firewall routers @@ -114,7 +115,10 @@ struct ClusterFirewallData { async fn fetch_cluster_firewall_data( args: ParallelFetcherArgs<()>, ) -> Result { - let pve = crate::connection::make_pve_client(args.remote())?; + let pve = args + .pdm_application() + .client_factory() + .make_pve_client(args.remote())?; let guests = match pve.cluster_resources(Some(ClusterResourceKind::Vm)).await { Ok(guests) => guests, @@ -204,7 +208,10 @@ async fn load_guests_firewall_status( async fn fetch_node_firewall_status( args: ParallelFetcherArgs, ) -> Result { - let pve = crate::connection::make_pve_client(args.remote())?; + let pve = args + .pdm_application() + .client_factory() + .make_pve_client(args.remote())?; let options_response = pve.node_firewall_options(args.node()); let rules_response = pve.list_node_firewall_rules(args.node()); @@ -246,8 +253,9 @@ async fn fetch_node_firewall_status( /// Get firewall status of all PVE remotes. pub async fn pve_firewall_status( _rpcenv: &mut dyn RpcEnvironment, + #[state] app: PdmApplication, ) -> Result, Error> { - let pve_remotes: Vec = RemoteIterator::new(pdm_config::remotes::instance())? + let pve_remotes: Vec = RemoteIterator::new(app.remote_config())? .remote_type(pdm_api_types::remotes::RemoteType::Pve) .into_remotes() .collect(); @@ -257,7 +265,7 @@ pub async fn pve_firewall_status( } // 1: fetch cluster-level data (status + guests) - let cluster_fetcher = ParallelFetcher::new(()); + let cluster_fetcher = ParallelFetcher::new(app.clone(), ()); let cluster_results = cluster_fetcher .do_for_all_remotes(pve_remotes.iter().cloned(), fetch_cluster_firewall_data) .await; @@ -275,7 +283,7 @@ pub async fn pve_firewall_status( guests: Arc::new(vec![]), }; - let node_fetcher = ParallelFetcher::new(context); + let node_fetcher = ParallelFetcher::new(app, context); let node_results = node_fetcher .do_for_all_remote_nodes(pve_remotes.iter().cloned(), move |mut args| { if let Some(guests) = guests_per_remote.get(&args.remote().id) { @@ -352,8 +360,9 @@ pub async fn cluster_firewall_options( pub async fn cluster_firewall_status( remote: String, _rpcenv: &mut dyn RpcEnvironment, + #[state] app: PdmApplication, ) -> Result { - let (remote_config, _) = pdm_config::remotes::config()?; + let (remote_config, _) = app.remote_config().read()?; let remote_obj = remote_config .into_iter() @@ -362,7 +371,7 @@ pub async fn cluster_firewall_status( .ok_or_else(|| anyhow::format_err!("Remote '{}' not found", remote))?; // 1: fetch cluster-level data (status + guests) - let cluster_fetcher = ParallelFetcher::new(()); + let cluster_fetcher = ParallelFetcher::new(app.clone(), ()); let cluster_results = cluster_fetcher .do_for_all_remotes( std::iter::once(remote_obj.clone()), @@ -390,7 +399,7 @@ pub async fn cluster_firewall_status( guests: Arc::new(guests), }; - let node_fetcher = ParallelFetcher::new(context); + let node_fetcher = ParallelFetcher::new(app, context); let node_results = node_fetcher .do_for_all_remote_nodes(std::iter::once(remote_obj), fetch_node_firewall_status) .await; diff --git a/server/src/api/sdn/controllers.rs b/server/src/api/sdn/controllers.rs index 41108674..5400c189 100644 --- a/server/src/api/sdn/controllers.rs +++ b/server/src/api/sdn/controllers.rs @@ -9,8 +9,8 @@ use proxmox_router::{Permission, Router, RpcEnvironment, http_bail}; use proxmox_schema::api; use pve_api_types::ListControllersType; -use crate::api::pve; use crate::api::remotes::RemoteIterator; +use crate::context::PdmApplication; use crate::parallel_fetcher::ParallelFetcher; pub const ROUTER: Router = Router::new().get(&API_METHOD_LIST_CONTROLLERS); @@ -63,6 +63,7 @@ pub async fn list_controllers( ty: Option, remotes: Option>, rpcenv: &mut dyn RpcEnvironment, + #[state] app: PdmApplication, ) -> Result, Error> { let user_info = CachedUserInfo::new()?; @@ -75,7 +76,7 @@ pub async fn list_controllers( http_bail!(FORBIDDEN, "user has no access to resources"); } - let mut iter = RemoteIterator::new(pdm_config::remotes::instance())? + let mut iter = RemoteIterator::new(app.remote_config())? .remote_type(RemoteType::Pve) .any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT); if let Some(ref filter) = remotes { @@ -83,11 +84,15 @@ pub async fn list_controllers( } let mut vnets = Vec::new(); - let fetcher = ParallelFetcher::new((pending, running, ty)); + + let fetcher = ParallelFetcher::new(app, (pending, running, ty)); let results = fetcher .do_for_all_remotes(iter.into_remotes(), async |args| { - Ok(pve::connect(args.remote())? + Ok(args + .pdm_application() + .client_factory() + .make_pve_client(args.remote())? .list_controllers(args.context().0, args.context().1, args.context().2) .await?) }) diff --git a/server/src/api/sdn/vnets.rs b/server/src/api/sdn/vnets.rs index 89017438..312d46b6 100644 --- a/server/src/api/sdn/vnets.rs +++ b/server/src/api/sdn/vnets.rs @@ -1,6 +1,7 @@ use std::collections::HashSet; use anyhow::{Context, Error}; + use pbs_api_types::REMOTE_ID_SCHEMA; use pdm_api_types::{ Authid, PRIV_RESOURCE_AUDIT, @@ -13,8 +14,7 @@ use proxmox_router::{Permission, Router, RpcEnvironment, http_bail}; use proxmox_schema::api; use pve_api_types::{CreateVnet, SdnVnetType}; -use crate::api::pve; -use crate::api::remotes::RemoteIterator; +use crate::{api::remotes::RemoteIterator, context::PdmApplication}; use crate::{parallel_fetcher::ParallelFetcher, sdn_client::LockedSdnClients}; pub const ROUTER: Router = Router::new() @@ -64,6 +64,7 @@ async fn list_vnets( running: Option, remotes: Option>, rpcenv: &mut dyn RpcEnvironment, + #[state] app: PdmApplication, ) -> Result, Error> { let user_info = CachedUserInfo::new()?; @@ -76,7 +77,7 @@ async fn list_vnets( http_bail!(FORBIDDEN, "user has no access to resources"); } - let mut iter = RemoteIterator::new(pdm_config::remotes::instance())? + let mut iter = RemoteIterator::new(app.remote_config())? .remote_type(RemoteType::Pve) .any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT); if let Some(ref filter) = remotes { @@ -84,11 +85,14 @@ async fn list_vnets( } let mut vnets = Vec::new(); - let fetcher = ParallelFetcher::new((pending, running)); + let fetcher = ParallelFetcher::new(app, (pending, running)); let results = fetcher .do_for_all_remotes(iter.into_remotes(), async |args| { - Ok(pve::connect(args.remote())? + Ok(args + .pdm_application() + .client_factory() + .make_pve_client(args.remote())? .list_vnets(args.context().0, args.context().1) .await?) }) diff --git a/server/src/api/sdn/zones.rs b/server/src/api/sdn/zones.rs index 5d8d8add..82366e81 100644 --- a/server/src/api/sdn/zones.rs +++ b/server/src/api/sdn/zones.rs @@ -14,8 +14,7 @@ use proxmox_router::{Permission, Router, RpcEnvironment, http_bail}; use proxmox_schema::api; use pve_api_types::{CreateZone, ListZonesType}; -use crate::api::pve; -use crate::api::remotes::RemoteIterator; +use crate::{api::remotes::RemoteIterator, context::PdmApplication}; use crate::{parallel_fetcher::ParallelFetcher, sdn_client::LockedSdnClients}; pub const ROUTER: Router = Router::new() @@ -70,6 +69,7 @@ pub async fn list_zones( ty: Option, remotes: Option>, rpcenv: &mut dyn RpcEnvironment, + #[state] app: PdmApplication, ) -> Result, Error> { let user_info = CachedUserInfo::new()?; @@ -82,7 +82,7 @@ pub async fn list_zones( http_bail!(FORBIDDEN, "user has no access to resources"); } - let mut iter = RemoteIterator::new(pdm_config::remotes::instance())? + let mut iter = RemoteIterator::new(app.remote_config())? .remote_type(RemoteType::Pve) .any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT); if let Some(ref filter) = remotes { @@ -90,11 +90,14 @@ pub async fn list_zones( } let mut vnets = Vec::new(); - let fetcher = ParallelFetcher::new((pending, running, ty)); + let fetcher = ParallelFetcher::new(app, (pending, running, ty)); let results = fetcher .do_for_all_remotes(iter.into_remotes(), async |args| { - Ok(pve::connect(args.remote())? + Ok(args + .pdm_application() + .client_factory() + .make_pve_client(args.remote())? .list_zones(args.context().0, args.context().1, args.context().2) .await?) }) diff --git a/server/src/parallel_fetcher.rs b/server/src/parallel_fetcher.rs index 512a04d9..b90f02e3 100644 --- a/server/src/parallel_fetcher.rs +++ b/server/src/parallel_fetcher.rs @@ -8,6 +8,7 @@ //! # //! # #[tokio::main] //! # async fn main() -> Result<(), Error> { +//! # let app = server::context::pdm_application(); //! # let remotes: Vec = Vec::new(); //! # //! async fn fetch_meaning( @@ -25,7 +26,7 @@ //! // This context can be passed to the function what is executed for every remote node. //! let context = (); //! -//! let fetcher = ParallelFetcher::builder(context) +//! let fetcher = ParallelFetcher::builder(app, context) //! .max_connections(10) //! .max_connections_per_remote(2) //! .build(); @@ -73,7 +74,7 @@ use pve_api_types::ClusterNodeIndexResponse; use pdm_api_types::remotes::{Remote, RemoteType}; -use crate::connection; +use crate::context::PdmApplication; /// Maximum number of parallel outgoing API requests. pub const DEFAULT_MAX_CONNECTIONS: usize = 20; @@ -229,15 +230,17 @@ impl NodeResponse { pub struct ParallelFetcherBuilder { max_connections: Option, max_connections_per_remote: Option, + pdm_application: PdmApplication, context: C, } impl ParallelFetcherBuilder { - fn new(context: C) -> Self { + fn new(pdm_application: PdmApplication, context: C) -> Self { Self { context, max_connections: None, max_connections_per_remote: None, + pdm_application, } } @@ -263,6 +266,7 @@ impl ParallelFetcherBuilder { .max_connections_per_remote .unwrap_or(DEFAULT_MAX_CONNECTIONS_PER_REMOTE), context: self.context, + pdm_application: self.pdm_application, } } } @@ -280,6 +284,8 @@ pub struct ParallelFetcherArgs { /// The node. This may be 'localhost' for PBS remotes or if using /// [`ParallelFetcher::do_for_all_remotes`]. node: String, + /// A handle to the [`PdmApplication`] context object. + pdm_application: PdmApplication, } impl ParallelFetcherArgs { @@ -305,24 +311,32 @@ impl ParallelFetcherArgs { pub fn context_mut(&mut self) -> &mut C { &mut self.context } + + /// Get a reference to [`PdmApplication`]. + /// + /// Use this for accessing configuration or constructing API clients. + pub fn pdm_application(&self) -> &PdmApplication { + &self.pdm_application + } } /// Helper for parallelizing API requests to multiple remotes/nodes. pub struct ParallelFetcher { max_connections: usize, max_connections_per_remote: usize, + pdm_application: PdmApplication, context: C, } impl ParallelFetcher { /// Create a [`ParallelFetcher`] with default settings. - pub fn new(context: C) -> Self { - Self::builder(context).build() + pub fn new(app: PdmApplication, context: C) -> Self { + Self::builder(app, context).build() } /// Create the builder for constructing a [`ParallelFetcher`] with custom settings. - pub fn builder(context: C) -> ParallelFetcherBuilder { - ParallelFetcherBuilder::new(context) + pub fn builder(app: PdmApplication, context: C) -> ParallelFetcherBuilder { + ParallelFetcherBuilder::new(app, context) } /// Invoke a function `func` for all nodes of a given list of remotes in parallel. @@ -344,13 +358,16 @@ impl ParallelFetcher { for remote in remotes { let semaphore = Arc::clone(&total_connections_semaphore); + let app = self.pdm_application.clone(); let f = func.clone(); + let future = Self::fetch_remote( remote, self.context.clone(), semaphore, f, self.max_connections_per_remote, + app, ); if let Some(log_context) = LogContext::current() { @@ -382,6 +399,7 @@ impl ParallelFetcher { semaphore: Arc, func: F, max_connections_per_remote: usize, + pdm_application: PdmApplication, ) -> RemoteResponse> where F: Fn(ParallelFetcherArgs) -> Ft + Clone + Send + 'static, @@ -397,8 +415,10 @@ impl ParallelFetcher { RemoteType::Pve => { let remote_clone = remote.clone(); + let app = &pdm_application; + let nodes = match async move { - let client = connection::make_pve_client(&remote_clone)?; + let client = app.client_factory().make_pve_client(&remote_clone)?; let nodes = client.list_nodes().await?; Ok::, Error>(nodes) @@ -433,12 +453,14 @@ impl ParallelFetcher { let remote_clone = remote.clone(); let node_name = node.node.clone(); let context_clone = context.clone(); + let app = pdm_application.clone(); let future = Self::fetch_node( func_clone, context_clone, remote_clone, node_name, + app, permit, Some(per_remote_connections_permit), ); @@ -467,6 +489,7 @@ impl ParallelFetcher { context, remote.clone(), "localhost".into(), + pdm_application, permit.unwrap(), // Always set to `Some` at this point None, ) @@ -490,6 +513,7 @@ impl ParallelFetcher { context: C, remote: Remote, node: String, + pdm_application: PdmApplication, _permit: OwnedSemaphorePermit, _per_remote_connections_permit: Option, ) -> NodeResponse @@ -504,6 +528,7 @@ impl ParallelFetcher { context, remote, node: node.clone(), + pdm_application, }; let result = func(parallel_fetcher_context).await; @@ -539,7 +564,9 @@ impl ParallelFetcher { let remote_type = remote.ty; let context = self.context.clone(); + let app = self.pdm_application.clone(); let func = func.clone(); + let future = async move { let permit = total_connections_semaphore.acquire_owned().await.unwrap(); @@ -551,6 +578,7 @@ impl ParallelFetcher { context, remote, "localhost".into(), + app, permit, None, ) diff --git a/server/src/remote_tasks/refresh_task.rs b/server/src/remote_tasks/refresh_task.rs index c1753114..7d729c22 100644 --- a/server/src/remote_tasks/refresh_task.rs +++ b/server/src/remote_tasks/refresh_task.rs @@ -9,7 +9,6 @@ use pdm_api_types::RemoteUpid; use pdm_api_types::remotes::{Remote, RemoteType}; use proxmox_section_config::typed::SectionConfigData; -use crate::api; use crate::connection; use crate::parallel_fetcher::{ParallelFetcher, ParallelFetcherArgs}; use crate::pbs_client; @@ -17,6 +16,7 @@ use crate::remote_tasks::{ KEEP_OLD_FILES, ROTATE_AFTER, task_cache::{GetTasks, NodeFetchSuccessMap, State, TaskCache, TaskCacheItem}, }; +use crate::{api, context}; /// Interval in seconds at which to fetch the newest tasks from remotes (if there is no tracked /// task for this remote). @@ -234,7 +234,7 @@ async fn fetch_remotes( remotes: Vec, cache_state: Arc, ) -> (Vec, NodeFetchSuccessMap) { - let fetcher = ParallelFetcher::builder(cache_state) + let fetcher = ParallelFetcher::builder(context::pdm_application(), cache_state) .max_connections(MAX_CONNECTIONS) .max_connections_per_remote(CONNECTIONS_PER_PVE_REMOTE) .build(); diff --git a/server/src/remote_updates.rs b/server/src/remote_updates.rs index bd648efd..890bbfa9 100644 --- a/server/src/remote_updates.rs +++ b/server/src/remote_updates.rs @@ -12,7 +12,7 @@ use pdm_api_types::remotes::{Remote, RemoteType}; use crate::namespaced_cache::CacheError; use crate::parallel_fetcher::ParallelFetcher; -use crate::{api_cache, connection}; +use crate::{api_cache, connection, context}; const OLD_CACHEFILE: &str = concat!(pdm_buildcfg::PDM_CACHE_DIR_M!(), "/remote-updates.json"); @@ -229,7 +229,7 @@ async fn update_cached_summary_for_node( /// delays its own entry. The final pass records whole-remote failures and prunes vanished /// remotes and nodes. pub async fn refresh_update_summary_cache(remotes: Vec) -> Result<(), Error> { - let fetcher = ParallelFetcher::new(()); + let fetcher = ParallelFetcher::new(context::pdm_application(), ()); let fetch_response = fetcher .do_for_all_remote_nodes(remotes.into_iter(), |args| async move { -- 2.47.3