[pdm-devel] [PATCH proxmox-datacenter-manager v3 09/26] metric collection: add tests for the fetch_remotes function

From: Lukas Wagner <l.wagner@proxmox.com>
To: pdm-devel@lists.proxmox.com
Subject: [pdm-devel] [PATCH proxmox-datacenter-manager v3 09/26] metric collection: add tests for the fetch_remotes function
Date: Wed, 16 Apr 2025 14:56:25 +0200	[thread overview]
Message-ID: <20250416125642.291552-10-l.wagner@proxmox.com> (raw)
In-Reply-To: <20250416125642.291552-1-l.wagner@proxmox.com>

Since we have the option to have mocked remote clients and a mocked
remote config, as well as due to the decoupling of the metric collection
and RRD task, it is now viable to write a unit test for one of the core
pieces of the metric collection system.

Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
Reviewed-by: Maximiliano Sandoval <m.sandoval@proxmox.com>
---
 .../src/metric_collection/collection_task.rs  | 227 ++++++++++++++++++
 server/src/metric_collection/state.rs         |  16 +-
 2 files changed, 231 insertions(+), 12 deletions(-)

diff --git a/server/src/metric_collection/collection_task.rs b/server/src/metric_collection/collection_task.rs
index 67953965..4331e9aa 100644
--- a/server/src/metric_collection/collection_task.rs
+++ b/server/src/metric_collection/collection_task.rs
@@ -370,3 +370,230 @@ pub(super) fn load_state() -> Result<MetricCollectionState, Error> {
         file_options,
     ))
 }
+
+#[cfg(test)]
+pub(super) mod tests {
+    use std::{collections::HashMap, sync::Once};
+
+    use anyhow::bail;
+    use http::StatusCode;
+
+    use pdm_api_types::Authid;
+    use pve_api_types::{ClusterMetrics, ClusterMetricsData};
+
+    use crate::{
+        connection::{ClientFactory, PveClient},
+        metric_collection::rrd_task::RrdStoreResult,
+        pbs_client::PbsClient,
+        test_support::temp::NamedTempFile,
+    };
+
+    use super::*;
+
+    pub(crate) fn get_create_options() -> CreateOptions {
+        CreateOptions::new()
+            .owner(nix::unistd::Uid::effective())
+            .group(nix::unistd::Gid::effective())
+            .perm(nix::sys::stat::Mode::from_bits_truncate(0o600))
+    }
+
+    struct TestClientFactory {
+        now: i64,
+    }
+
+    #[async_trait::async_trait]
+    impl ClientFactory for TestClientFactory {
+        fn make_pve_client(&self, remote: &Remote) -> Result<Arc<PveClient>, Error> {
+            Ok(Arc::new(TestPveClient {
+                fail: remote.id.contains("fail"),
+                now: self.now,
+            }))
+        }
+        /// Create a new API client for PVE remotes, but with a specific endpoint.
+        fn make_pve_client_with_endpoint(
+            &self,
+            _remote: &Remote,
+            _target_endpoint: Option<&str>,
+        ) -> Result<Arc<PveClient>, Error> {
+            bail!("not implemented")
+        }
+
+        fn make_pbs_client(&self, _remote: &Remote) -> Result<Box<PbsClient>, Error> {
+            bail!("not implemented")
+        }
+
+        async fn make_pve_client_and_login(
+            &self,
+            _remote: &Remote,
+        ) -> Result<Arc<PveClient>, Error> {
+            bail!("not implemented")
+        }
+
+        async fn make_pbs_client_and_login(
+            &self,
+            _remote: &Remote,
+        ) -> Result<Box<PbsClient>, Error> {
+            bail!("not implemented")
+        }
+    }
+
+    struct TestPveClient {
+        now: i64,
+        fail: bool,
+    }
+
+    #[async_trait::async_trait]
+    impl pve_api_types::client::PveClient for TestPveClient {
+        /// Retrieve metrics of the cluster.
+        async fn cluster_metrics_export(
+            &self,
+            _history: Option<bool>,
+            _local_only: Option<bool>,
+            start_time: Option<i64>,
+        ) -> Result<ClusterMetrics, proxmox_client::Error> {
+            if self.fail {
+                return Err(proxmox_client::Error::Api(
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    "internal server error".into(),
+                ));
+            }
+
+            let mut time = start_time.unwrap_or(0);
+            time = time.max(self.now - (30 * 60));
+            let mut data = Vec::new();
+
+            use pve_api_types::ClusterMetricsDataType::*;
+
+            while time < self.now {
+                let point = |id: &str, metric: &str, timestamp, ty| ClusterMetricsData {
+                    id: id.into(),
+                    metric: metric.into(),
+                    timestamp,
+                    ty,
+                    value: 0.1,
+                };
+
+                for i in 0..5 {
+                    let id = format!("node/node-{i}");
+                    data.push(point(&id, "cpu_current", time, Gauge));
+                }
+
+                // Advance time by 10 seconds
+                time += 10;
+            }
+
+            Ok(ClusterMetrics { data })
+        }
+    }
+
+    fn make_remote_config() -> SectionConfigData<Remote> {
+        let mut order = Vec::new();
+        let mut sections = HashMap::new();
+
+        for i in 0..4 {
+            let status = if i >= 2 { "fail" } else { "pass" };
+            let name = format!("pve-{i}-{status}");
+
+            sections.insert(
+                name.clone(),
+                Remote {
+                    ty: pdm_api_types::remotes::RemoteType::Pve,
+                    id: name.clone(),
+                    nodes: Vec::new(),
+                    authid: Authid::root_auth_id().clone(),
+                    token: "".into(),
+                    web_url: None,
+                },
+            );
+
+            order.push(name);
+        }
+
+        SectionConfigData { sections, order }
+    }
+
+    async fn fake_rrd_task(mut rx: Receiver<RrdStoreRequest>) -> u32 {
+        let mut number_of_requests = 0;
+
+        while let Some(request) = rx.recv().await {
+            number_of_requests += 1;
+
+            if let RrdStoreRequest::Pve {
+                metrics, channel, ..
+            } = request
+            {
+                let most_recent_timestamp =
+                    metrics.data.iter().map(|e| e.timestamp).max().unwrap_or(0);
+
+                let _ = channel.send(RrdStoreResult {
+                    most_recent_timestamp,
+                });
+            }
+        }
+
+        number_of_requests
+    }
+
+    static START: Once = Once::new();
+
+    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(Box::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));
+
+        let settings = CollectionSettings::default();
+        let config = make_remote_config();
+
+        let state_file = NamedTempFile::new(get_create_options()).unwrap();
+        let state = MetricCollectionState::new(state_file.path().into(), get_create_options());
+
+        let (_control_tx, control_rx) = tokio::sync::mpsc::channel(10);
+
+        let mut task = MetricCollectionTask {
+            state,
+            settings,
+            metric_data_tx: tx,
+            control_message_rx: control_rx,
+        };
+
+        // Act
+        task.fetch_remotes(&config, &config.order).await;
+
+        // Assert
+        for remote in config.order {
+            let status = task.state.get_status(&remote).unwrap();
+
+            // Our faked PVE client will return an error if the remote name contains
+            // 'fail'.
+            if remote.contains("fail") {
+                assert!(status
+                    .error
+                    .as_ref()
+                    .unwrap()
+                    .contains("internal server error"));
+                assert_eq!(status.last_collection, None);
+            } else {
+                assert!(now - status.most_recent_datapoint <= 10);
+                assert!(status.error.is_none());
+            }
+        }
+
+        drop(task);
+        assert_eq!(handle.await.unwrap(), 2);
+    }
+}
diff --git a/server/src/metric_collection/state.rs b/server/src/metric_collection/state.rs
index 5b04ea61..86375a65 100644
--- a/server/src/metric_collection/state.rs
+++ b/server/src/metric_collection/state.rs
@@ -87,23 +87,15 @@ impl MetricCollectionState {
 
 #[cfg(test)]
 mod tests {
-    use proxmox_sys::fs::CreateOptions;
+    use crate::metric_collection::collection_task::tests::get_create_options;
+    use crate::test_support::temp::NamedTempFile;
 
     use super::*;
 
-    use crate::test_support::temp::NamedTempFile;
-
-    fn get_options() -> CreateOptions {
-        CreateOptions::new()
-            .owner(nix::unistd::Uid::effective())
-            .group(nix::unistd::Gid::effective())
-            .perm(nix::sys::stat::Mode::from_bits_truncate(0o600))
-    }
-
     #[test]
     fn save_and_load() -> Result<(), Error> {
-        let file = NamedTempFile::new(get_options())?;
-        let options = get_options();
+        let file = NamedTempFile::new(get_create_options())?;
+        let options = get_create_options();
         let mut state = MetricCollectionState::new(file.path().into(), options.clone());
 
         state.set_status(
-- 
2.39.5



_______________________________________________
pdm-devel mailing list
pdm-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pdm-devel

