[PATCH datacenter-manager v3 12/21] connection: use client factory from PdmApplication handle

[Lukas Wagner <l.wagner@proxmox.com>]

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 <l.wagner@proxmox.com>
---

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<Arc<dyn ClientFactory + Send + Sync>> = 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<Arc<PveClient>, 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<Arc<PveClient>, 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<Arc<PveClient>, 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<Box<PbsClient>, Error> {
-    instance().make_pbs_client(remote)
+    context::pdm_application()
+        .client_factory()
+        .make_pbs_client(remote)
 }
 
 pub fn make_raw_client(remote: &Remote) -> Result<Box<Client>, 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<Box<Client>, Error> {
 ///
 /// Note: currently does not support two factor authentication.
 pub async fn make_pve_client_and_login(remote: &Remote) -> Result<Arc<PveClient>, 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<Arc<PveClient>
 ///
 /// Note: currently does not support two factor authentication.
 pub async fn make_pbs_client_and_login(remote: &Remote) -> Result<Box<PbsClient<Client>>, 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<dyn ClientFactory + Send + Sync>) {
-    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<PdmApplication, Error> {
     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<RrdStoreRequest>,
     control_message_rx: Receiver<ControlMsg>,
@@ -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<RrdStoreRequest>,
         control_message_rx: Receiver<ControlMsg>,
     ) -> Result<Self, Error> {
         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<RrdStoreRequest>,
@@ -307,7 +315,7 @@ impl RemoteMetricCollectionTask {
         let res: Result<RrdStoreResult, Error> = 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<MetricCollectionState, Error> {
 
 #[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<Arc<dyn ClientFactory + Send + Sync>, 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