[pdm-devel] [PATCH datacenter-manager v2 2/3] server: introduce ApiCache abstraction

[Dominik Csapak <d.csapak@proxmox.com>]

Abstract our resources/subscription cache mechanism away into a struct
that takes care of the max-age logic and fetching/inserting.

This can be useful to have in general, and we don't need to copy it
every time.

We still have to use a LazyLock at the usage site and give the correct
Type, but we can deduplicate most of the access/update logic.

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
changes from v1:
* add new CacheUpdateResult struct for the 'update_or_get_cache'
  method
* rename like Lukas suggested
* combine impl blocks

 server/src/api/resources.rs                  | 142 +++----------------
 server/src/api_cache.rs                      | 128 +++++++++++++++++
 server/src/lib.rs                            |   1 +
 server/src/metric_collection/top_entities.rs |   2 +-
 4 files changed, 150 insertions(+), 123 deletions(-)
 create mode 100644 server/src/api_cache.rs

diff --git a/server/src/api/resources.rs b/server/src/api/resources.rs
index c640e4a2..0df9c781 100644
--- a/server/src/api/resources.rs
+++ b/server/src/api/resources.rs
@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 use std::str::FromStr;
-use std::sync::{LazyLock, RwLock};
+use std::sync::LazyLock;
 
 use anyhow::{bail, format_err, Error};
 use futures::future::join_all;
@@ -28,6 +28,7 @@ use proxmox_sortable_macro::sortable;
 use proxmox_subscription::SubscriptionStatus;
 use pve_api_types::{ClusterResource, ClusterResourceType};
 
+use crate::api_cache::ApiCache;
 use crate::connection;
 use crate::metric_collection::top_entities;
 
@@ -521,14 +522,8 @@ async fn get_top_entities(
     Ok(res)
 }
 
-#[derive(Clone)]
-struct CachedSubscriptionState {
-    node_info: HashMap<String, Option<NodeSubscriptionInfo>>,
-    timestamp: i64,
-}
-
-static SUBSCRIPTION_CACHE: LazyLock<RwLock<HashMap<String, CachedSubscriptionState>>> =
-    LazyLock::new(|| RwLock::new(HashMap::new()));
+static SUBSCRIPTION_CACHE: LazyLock<ApiCache<HashMap<String, Option<NodeSubscriptionInfo>>>> =
+    LazyLock::new(ApiCache::new);
 
 /// Get the subscription state for a given remote.
 ///
@@ -538,63 +533,12 @@ async fn get_subscription_info_for_remote(
     remote: &Remote,
     max_age: u64,
 ) -> Result<HashMap<String, Option<NodeSubscriptionInfo>>, Error> {
-    if let Some(cached_subscription) = get_cached_subscription_info(&remote.id, max_age) {
-        Ok(cached_subscription.node_info)
-    } else {
-        let node_info = fetch_remote_subscription_info(remote).await?;
-        let now = proxmox_time::epoch_i64();
-        update_cached_subscription_info(&remote.id, &node_info, now);
-        Ok(node_info)
-    }
-}
-
-fn get_cached_subscription_info(remote: &str, max_age: u64) -> Option<CachedSubscriptionState> {
-    let cache = SUBSCRIPTION_CACHE
-        .read()
-        .expect("subscription mutex poisoned");
-
-    if let Some(cached_subscription) = cache.get(remote) {
-        let now = proxmox_time::epoch_i64();
-        let diff = now - cached_subscription.timestamp;
-
-        if diff > max_age as i64 || diff < 0 {
-            // value is too old or from the future
-            None
-        } else {
-            Some(cached_subscription.clone())
-        }
-    } else {
-        None
-    }
-}
-
-/// Update cached subscription data.
-///
-/// If the cache already contains more recent data we don't insert the passed resources.
-fn update_cached_subscription_info(
-    remote: &str,
-    node_info: &HashMap<String, Option<NodeSubscriptionInfo>>,
-    now: i64,
-) {
-    // there is no good way to recover from this, so panicking should be fine
-    let mut cache = SUBSCRIPTION_CACHE
-        .write()
-        .expect("subscription mutex poisoned");
-
-    if let Some(cached_resource) = cache.get(remote) {
-        // skip updating if the data is new enough
-        if cached_resource.timestamp >= now {
-            return;
-        }
-    }
-
-    cache.insert(
-        remote.into(),
-        CachedSubscriptionState {
-            node_info: node_info.clone(),
-            timestamp: now,
-        },
-    );
+    let (data, _) = SUBSCRIPTION_CACHE
+        .get_value_or_update(&remote.id, max_age as i64, |_| async move {
+            fetch_remote_subscription_info(remote).await
+        })
+        .await?;
+    Ok(data)
 }
 
 /// Maps a list of node subscription infos into a single [`RemoteSubscriptionState`]
@@ -687,14 +631,7 @@ async fn fetch_remote_subscription_info(
     Ok(list)
 }
 
-#[derive(Clone)]
-pub struct CachedResources {
-    pub resources: Vec<Resource>,
-    pub timestamp: i64,
-}
-
-static CACHE: LazyLock<RwLock<HashMap<String, CachedResources>>> =
-    LazyLock::new(|| RwLock::new(HashMap::new()));
+static CACHE: LazyLock<ApiCache<Vec<Resource>>> = LazyLock::new(ApiCache::new);
 
 /// Get resources for a given remote.
 ///
@@ -702,57 +639,18 @@ static CACHE: LazyLock<RwLock<HashMap<String, CachedResources>>> =
 /// instead of calling out to the remote.
 async fn get_resources_for_remote(remote: &Remote, max_age: u64) -> Result<Vec<Resource>, Error> {
     let remote_name = remote.id.to_owned();
-    if let Some(cached_resource) = get_cached_resources(&remote_name, max_age) {
-        Ok(cached_resource.resources)
-    } else {
-        let resources = fetch_remote_resource(remote).await?;
-        let now = proxmox_time::epoch_i64();
-        update_cached_resources(&remote_name, &resources, now);
-        Ok(resources)
-    }
+    let (data, _) = CACHE
+        .get_value_or_update(&remote_name, max_age as i64, |_| async move {
+            fetch_remote_resource(remote).await
+        })
+        .await?;
+    Ok(data)
 }
 
 /// Read cached resource data from the cache
-pub fn get_cached_resources(remote: &str, max_age: u64) -> Option<CachedResources> {
-    // there is no good way to recover from this, so panicking should be fine
-    let cache = CACHE.read().expect("mutex poisoned");
-
-    if let Some(cached_resource) = cache.get(remote) {
-        let now = proxmox_time::epoch_i64();
-        let diff = now - cached_resource.timestamp;
-
-        if diff > max_age as i64 || diff < 0 {
-            // value is too old or from the future
-            None
-        } else {
-            Some(cached_resource.clone())
-        }
-    } else {
-        None
-    }
-}
-
-/// Update cached resource data.
-///
-/// If the cache already contains more recent data we don't insert the passed resources.
-fn update_cached_resources(remote: &str, resources: &[Resource], now: i64) {
-    // there is no good way to recover from this, so panicking should be fine
-    let mut cache = CACHE.write().expect("mutex poisoned");
-
-    if let Some(cached_resource) = cache.get(remote) {
-        // skip updating if existing data is newer
-        if cached_resource.timestamp >= now {
-            return;
-        }
-    }
-
-    cache.insert(
-        remote.into(),
-        CachedResources {
-            timestamp: now,
-            resources: resources.into(),
-        },
-    );
+pub fn get_cached_resources(remote: &str, max_age: u64) -> Option<Vec<Resource>> {
+    let (data, _) = CACHE.get_value(remote, max_age as i64)?;
+    Some(data)
 }
 
 /// Fetch remote resources and map to pdm-native data types.
diff --git a/server/src/api_cache.rs b/server/src/api_cache.rs
new file mode 100644
index 00000000..64b4ec04
--- /dev/null
+++ b/server/src/api_cache.rs
@@ -0,0 +1,128 @@
+use std::{collections::HashMap, future::Future, sync::RwLock};
+
+use anyhow::Error;
+
+pub struct CacheUpdateResult<T> {
+    pub data: T,
+    pub cache: bool,
+    pub cached_timestamp: i64,
+    pub error: Option<Error>,
+}
+
+/// Holds the given data per remote with a timestamp indicating the last update
+pub struct ApiCache<CacheType> {
+    data: RwLock<HashMap<String, (i64, CacheType)>>,
+}
+
+impl<CacheType> ApiCache<CacheType> {
+    /// Creates a new instance of `ApiCache`
+    pub fn new() -> Self {
+        Self {
+            data: RwLock::new(HashMap::new()),
+        }
+    }
+
+    fn update_cache(&self, remote: &str, data: CacheType, now: i64) {
+        // there is no good way to recover from this, so panicking should be fine
+        let mut cache = self.data.write().expect("cache mutex poisoned");
+
+        if let Some((timestamp, _)) = cache.get(remote) {
+            // skip updating if existing data is newer
+            if *timestamp >= now {
+                return;
+            }
+        }
+
+        cache.insert(remote.into(), (now, data));
+    }
+}
+
+impl<CacheType> Default for ApiCache<CacheType> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<CacheType> ApiCache<CacheType>
+where
+    CacheType: Clone,
+{
+    /// Returns the cached value and the cached timestamp for the given remote if it's not older than `max_age`
+    pub fn get_value(&self, remote: &str, max_age: i64) -> Option<(CacheType, i64)> {
+        let cache = self.data.read().expect("cache mutex poisoned");
+        if let Some((timestamp, cached_data)) = cache.get(remote) {
+            let now = proxmox_time::epoch_i64();
+            let diff = now - timestamp;
+            if diff > max_age || diff < 0 {
+                // value is too old or from the future
+                None
+            } else {
+                Some((cached_data.clone(), *timestamp))
+            }
+        } else {
+            None
+        }
+    }
+
+    /// Returns data for the given remote that is not older than `max_age` either
+    /// from the cache or updated via the given `update_func`
+    pub async fn get_value_or_update<S, F, R>(
+        &self,
+        remote: &str,
+        max_age: i64,
+        update_func: F,
+    ) -> Result<(CacheType, i64), Error>
+    where
+        S: Into<CacheType>,
+        F: Fn(&str) -> R,
+        R: Future<Output = Result<S, Error>>,
+    {
+        if let Some(cached_data) = self.get_value(remote, max_age) {
+            Ok(cached_data)
+        } else {
+            let data = update_func(remote).await?;
+            let now = proxmox_time::epoch_i64();
+
+            let data: CacheType = data.into();
+            self.update_cache(remote, data.clone(), now);
+            Ok((data, now))
+        }
+    }
+
+    /// Returns data for the given remote. Tries to update the value (and cache).
+    /// If that's not possible tries to load any cached value.
+    pub async fn update_or_get_cache<S, F, R>(
+        &self,
+        remote: &str,
+        update_func: F,
+    ) -> Result<CacheUpdateResult<CacheType>, Error>
+    where
+        S: Into<CacheType>,
+        F: Fn(&str) -> R,
+        R: Future<Output = Result<S, Error>>,
+    {
+        match update_func(remote).await {
+            Ok(data) => {
+                let data: CacheType = data.into();
+                let now = proxmox_time::epoch_i64();
+
+                self.update_cache(remote, data.clone(), now);
+                Ok(CacheUpdateResult {
+                    data,
+                    cache: false,
+                    cached_timestamp: now,
+                    error: None,
+                })
+            }
+            Err(err) => match self.get_value(remote, i64::MAX) {
+                Some((data, timestamp)) => Ok(CacheUpdateResult {
+                    data,
+                    cache: true,
+                    cached_timestamp: timestamp,
+                    error: Some(err),
+                }),
+                None => Err(err.context("cache was empty and could not update")),
+            },
+        }
+    }
+}
diff --git a/server/src/lib.rs b/server/src/lib.rs
index 964807eb..1f8508c9 100644
--- a/server/src/lib.rs
+++ b/server/src/lib.rs
@@ -2,6 +2,7 @@
 
 pub mod acl;
 pub mod api;
+pub mod api_cache;
 pub mod auth;
 pub mod context;
 pub mod env;
diff --git a/server/src/metric_collection/top_entities.rs b/server/src/metric_collection/top_entities.rs
index ea121eef..715d0eeb 100644
--- a/server/src/metric_collection/top_entities.rs
+++ b/server/src/metric_collection/top_entities.rs
@@ -50,7 +50,7 @@ pub fn calculate_top(
         if let Some(data) =
             crate::api::resources::get_cached_resources(remote_name, i64::MAX as u64)
         {
-            for res in data.resources {
+            for res in data {
                 let id = res.id().to_string();
                 let name = format!("pve/{remote_name}/{id}");
                 match &res {
-- 
2.47.3



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