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 8A19C1FF0E1 for ; Thu, 27 Aug 2026 13:43:40 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id 6F34821651; Thu, 27 Aug 2026 13:43:15 +0200 (CEST) From: Lukas Wagner To: pdm-devel@lists.proxmox.com Subject: [PATCH datacenter-manager v3 12/21] connection: use client factory from PdmApplication handle Date: Thu, 27 Aug 2026 13:42:35 +0200 Message-ID: <20260827114244.424784-13-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: 1787830960990 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.576 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 T_FILL_THIS_FORM_SHORT 0.01 Fill in a short form with personal information Message-ID-Hash: 4R2QKT4XS7APIO3NEHNQYTIIFG2CWPZK X-Message-ID-Hash: 4R2QKT4XS7APIO3NEHNQYTIIFG2CWPZK 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: The `connection` module now does not store a handle to the client factory any more, instead the free-standing helpers access the client factory instead via context::pdm_application(). Long-term, the free-standing helpers should be removed anyway, and the client factory only be accessed through the app handle. The change in `connection` required some adaptations to test cases for the remote metric collection task. Signed-off-by: Lukas Wagner --- Notes: Changes since v2: - Use ProductConfig from proxmox-product-config instead of implementing it locally server/src/bin/proxmox-datacenter-api/main.rs | 2 +- server/src/connection.rs | 54 +++++++++---------- server/src/context/mod.rs | 3 -- server/src/metric_collection/mod.rs | 6 ++- .../remote_collection_task.rs | 47 ++++++++-------- 5 files changed, 55 insertions(+), 57 deletions(-) diff --git a/server/src/bin/proxmox-datacenter-api/main.rs b/server/src/bin/proxmox-datacenter-api/main.rs index 64326b9b..eac1585b 100644 --- a/server/src/bin/proxmox-datacenter-api/main.rs +++ b/server/src/bin/proxmox-datacenter-api/main.rs @@ -344,7 +344,7 @@ async fn run(app: PdmApplication, debug: bool) -> Result<(), Error> { }); start_task_scheduler(); - metric_collection::start_task()?; + metric_collection::start_task(app)?; tasks::remote_node_mapping::start_task(); resource_cache::start_task(); tasks::remote_tasks::start_task()?; diff --git a/server/src/connection.rs b/server/src/connection.rs index 140a1902..df122836 100644 --- a/server/src/connection.rs +++ b/server/src/connection.rs @@ -7,9 +7,9 @@ use std::collections::HashMap; use std::future::Future; use std::pin::{Pin, pin}; use std::sync::Arc; +use std::sync::LazyLock; use std::sync::Mutex as StdMutex; use std::sync::Once; -use std::sync::{LazyLock, OnceLock}; use std::time::{Duration, SystemTime}; use anyhow::{Error, bail, format_err}; @@ -25,11 +25,10 @@ use proxmox_time::epoch_i64; use pdm_api_types::remotes::{NodeUrl, Remote, RemoteType, TlsProbeOutcome}; use pve_api_types::client::PveClientImpl; +use crate::context; use crate::pbs_client::PbsClient; use crate::remote_cache::ConnectionState; -static INSTANCE: OnceLock> = OnceLock::new(); - /// Connection Info returned from [`prepare_connect_client`] struct ConnectInfo { prefix: String, @@ -404,19 +403,11 @@ impl ClientFactory for DefaultClientFactory { } } -fn instance() -> &'static (dyn ClientFactory + Send + Sync) { - // Not initializing the connection factory instance is - // entirely in our responsibility and not something we can recover from, - // so it should be okay to panic in this case. - INSTANCE - .get() - .expect("client factory instance not set") - .as_ref() -} - /// Create a new API client for PVE remotes pub fn make_pve_client(remote: &Remote) -> Result, Error> { - instance().make_pve_client(remote) + context::pdm_application() + .client_factory() + .make_pve_client(remote) } /// Create a new API client for PVE remotes, but for a specific endpoint @@ -424,21 +415,29 @@ pub fn make_pve_client_with_endpoint( remote: &Remote, target_endpoint: Option<&str>, ) -> Result, Error> { - instance().make_pve_client_with_endpoint(remote, target_endpoint) + context::pdm_application() + .client_factory() + .make_pve_client_with_endpoint(remote, target_endpoint) } /// Create a new API client for PVE remotes and try to make it connect to a specific *node*. pub fn make_pve_client_with_node(remote: &Remote, node: &str) -> Result, Error> { - instance().make_pve_client_with_node(remote, node) + context::pdm_application() + .client_factory() + .make_pve_client_with_node(remote, node) } /// Create a new API client for PBS remotes pub fn make_pbs_client(remote: &Remote) -> Result, Error> { - instance().make_pbs_client(remote) + context::pdm_application() + .client_factory() + .make_pbs_client(remote) } pub fn make_raw_client(remote: &Remote) -> Result, Error> { - instance().make_raw_client(remote) + context::pdm_application() + .client_factory() + .make_raw_client(remote) } /// Create a new API client for PVE remotes. @@ -451,7 +450,10 @@ pub fn make_raw_client(remote: &Remote) -> Result, Error> { /// /// Note: currently does not support two factor authentication. pub async fn make_pve_client_and_login(remote: &Remote) -> Result, Error> { - instance().make_pve_client_and_login(remote).await + context::pdm_application() + .client_factory() + .make_pve_client_and_login(remote) + .await } /// Create a new API client for PBS remotes. @@ -464,16 +466,10 @@ pub async fn make_pve_client_and_login(remote: &Remote) -> Result /// /// Note: currently does not support two factor authentication. pub async fn make_pbs_client_and_login(remote: &Remote) -> Result>, Error> { - instance().make_pbs_client_and_login(remote).await -} - -/// Initialize the [`ClientFactory`] instance. -/// -/// Will panic if the instance has already been set. -pub fn init(instance: Arc) { - if INSTANCE.set(instance).is_err() { - panic!("connection factory instance already set"); - } + context::pdm_application() + .client_factory() + .make_pbs_client_and_login(remote) + .await } /// In order to allow the [`MultiClient`] to check the cached reachability state of a client, we diff --git a/server/src/context/mod.rs b/server/src/context/mod.rs index 3e89bceb..f0853d65 100644 --- a/server/src/context/mod.rs +++ b/server/src/context/mod.rs @@ -35,9 +35,6 @@ pub fn init() -> Result { pdm_config::remotes::init(factory.make_remote_config()?); pdm_config::subscriptions::init(factory.make_subscription_key_config()?); - // FIXME: Rather let connection use an Application context object from here - connection::init(factory.make_client_factory()?); - Ok(app) } diff --git a/server/src/metric_collection/mod.rs b/server/src/metric_collection/mod.rs index 3d7a477d..15b01b9a 100644 --- a/server/src/metric_collection/mod.rs +++ b/server/src/metric_collection/mod.rs @@ -20,6 +20,7 @@ pub mod top_entities; use remote_collection_task::{ControlMsg, RemoteMetricCollectionTask}; use rrd_cache::RrdCache; +use crate::context::PdmApplication; use crate::metric_collection::local_collection_task::LocalMetricCollectionTask; const RRD_CACHE_BASEDIR: &str = concat!(PDM_STATE_DIR_M!(), "/rrdb"); @@ -39,7 +40,7 @@ pub fn init() -> Result<(), Error> { } /// Start the metric collection task. -pub fn start_task() -> Result<(), Error> { +pub fn start_task(app: PdmApplication) -> Result<(), Error> { let (metric_data_tx, metric_data_rx) = mpsc::channel(128); let cache = rrd_cache::get_cache(); @@ -57,7 +58,8 @@ pub fn start_task() -> Result<(), Error> { let metric_data_tx_clone = metric_data_tx.clone(); tokio::spawn(async move { let metric_collection_task_future = pin!(async move { - match RemoteMetricCollectionTask::new(metric_data_tx_clone, trigger_collection_rx) { + match RemoteMetricCollectionTask::new(app, metric_data_tx_clone, trigger_collection_rx) + { Ok(mut task) => task.run().await, Err(err) => log::error!("could not start metric collection task: {err}"), } diff --git a/server/src/metric_collection/remote_collection_task.rs b/server/src/metric_collection/remote_collection_task.rs index 9fe67371..57910d46 100644 --- a/server/src/metric_collection/remote_collection_task.rs +++ b/server/src/metric_collection/remote_collection_task.rs @@ -19,8 +19,9 @@ use proxmox_sys::fs::CreateOptions; use pdm_api_types::remotes::{Remote, RemoteType}; +use crate::context::PdmApplication; use crate::metric_collection::rrd_task::CollectionStats; -use crate::{connection, task_utils}; +use crate::task_utils; use super::{ rrd_task::{RrdStoreRequest, RrdStoreResult}, @@ -48,6 +49,7 @@ pub(super) enum ControlMsg { /// Task which periodically collects metrics from all remotes and stores /// them in the local metrics database. pub(super) struct RemoteMetricCollectionTask { + app: PdmApplication, state: MetricCollectionState, metric_data_tx: Sender, control_message_rx: Receiver, @@ -56,12 +58,14 @@ pub(super) struct RemoteMetricCollectionTask { impl RemoteMetricCollectionTask { /// Create a new metric collection task. pub(super) fn new( + app: PdmApplication, metric_data_tx: Sender, control_message_rx: Receiver, ) -> Result { let state = load_state()?; Ok(Self { + app, state, metric_data_tx, control_message_rx, @@ -234,9 +238,12 @@ impl RemoteMetricCollectionTask { // called on the semaphore. let permit = Arc::clone(&semaphore).acquire_owned().await.unwrap(); + let app_clone = self.app.clone(); + if let Some(remote) = remote_config.get(remote_name).cloned() { log::debug!("fetching remote '{}'", remote.id); handles.spawn(Self::fetch_single_remote( + app_clone, remote, status, self.metric_data_tx.clone(), @@ -294,6 +301,7 @@ impl RemoteMetricCollectionTask { /// Fetch a single remote. #[tracing::instrument(skip_all, fields(remote = remote.id), name = "metric_collection_task")] async fn fetch_single_remote( + app: PdmApplication, remote: Remote, mut status: RemoteStatus, sender: Sender, @@ -307,7 +315,7 @@ impl RemoteMetricCollectionTask { let res: Result = async { match remote.ty { RemoteType::Pve => { - let client = connection::make_pve_client(&remote)?; + let client = app.client_factory().make_pve_client(&remote)?; let metrics = client .cluster_metrics_export( Some(true), @@ -331,7 +339,7 @@ impl RemoteMetricCollectionTask { .await?; } RemoteType::Pbs => { - let client = connection::make_pbs_client(&remote)?; + let client = app.client_factory().make_pbs_client(&remote)?; let metrics = client .metrics(Some(true), Some(status.most_recent_datapoint)) .await?; @@ -386,8 +394,6 @@ pub(super) fn load_state() -> Result { #[cfg(test)] pub(super) mod tests { - use std::sync::Once; - use anyhow::bail; use http::StatusCode; @@ -397,6 +403,7 @@ pub(super) mod tests { use crate::{ connection::{ClientFactory, PveClient}, + context::ContextFactory, metric_collection::rrd_task::RrdStoreResult, pbs_client::PbsClient, test_support::temp::NamedTempFile, @@ -550,25 +557,18 @@ pub(super) mod tests { number_of_requests } - static START: Once = Once::new(); + const NOW: i64 = 1000; - fn test_init() -> i64 { - let now = 10000; - START.call_once(|| { - // TODO: the client factory is currently stored in a OnceLock - - // we can only set it from one test... Ideally we'd like to have the - // option to set it in every single test if needed - task/thread local? - connection::init(Arc::new(TestClientFactory { now })); - }); + struct TestContextFactory(); - now + impl ContextFactory for TestContextFactory { + fn make_client_factory(&self) -> Result, Error> { + Ok(Arc::new(TestClientFactory { now: NOW })) + } } #[tokio::test] async fn test_fetch_remotes_updates_state() { - // Arrange - let now = test_init(); - let (tx, rx) = tokio::sync::mpsc::channel(10); let handle = tokio::task::spawn(fake_rrd_task(rx)); @@ -579,7 +579,10 @@ pub(super) mod tests { let (_control_tx, control_rx) = tokio::sync::mpsc::channel(10); + let app = TestContextFactory().make_pdm_application().unwrap(); + let mut task = RemoteMetricCollectionTask { + app, state, metric_data_tx: tx, control_message_rx: control_rx, @@ -608,7 +611,7 @@ pub(super) mod tests { ); assert_eq!(status.last_collection, None); } else { - assert!(now - status.most_recent_datapoint <= 10); + assert!(NOW - status.most_recent_datapoint <= 10); assert!(status.error.is_none()); } } @@ -619,9 +622,6 @@ pub(super) mod tests { #[tokio::test] async fn test_fetch_overdue() { - // Arrange - test_init(); - let (tx, rx) = tokio::sync::mpsc::channel(10); let handle = tokio::task::spawn(fake_rrd_task(rx)); @@ -630,6 +630,8 @@ pub(super) mod tests { let state_file = NamedTempFile::new(get_create_options()).unwrap(); let mut state = MetricCollectionState::new(state_file.path().into(), get_create_options()); + let app = TestContextFactory().make_pdm_application().unwrap(); + let now = proxmox_time::epoch_i64(); // This one should be fetched @@ -652,6 +654,7 @@ pub(super) mod tests { let (_control_tx, control_rx) = tokio::sync::mpsc::channel(10); let mut task = RemoteMetricCollectionTask { + app, state, metric_data_tx: tx, control_message_rx: control_rx, -- 2.47.3