datastore: remove datastore from internal cache based on maintenance mode

We keep a DataStore cache, so ChunkStore's and lock files are kept by
the proxy process and don't have to be reopened every time. However,
for specific maintenance modes, e.g. 'offline', our process should not
keep file in that datastore open. This clears the cache entry of a
datastore if it is in a specific maintanance mode and the last task
finished, which also drops any files still open by the process.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
Reviewed-by: Gabriel Goller <g.goller@proxmox.com>
Tested-by: Gabriel Goller <g.goller@proxmox.com>

pbs-api-types/src/maintenance.rs

@@ -77,6 +77,12 @@ pub struct MaintenanceMode {
 }
 
 impl MaintenanceMode {
+    /// Used for deciding whether the datastore is cleared from the internal cache after the last
+    /// task finishes, so all open files are closed.
+    pub fn is_offline(&self) -> bool {
+        self.ty == MaintenanceType::Offline
+    }
+
     pub fn check(&self, operation: Option<Operation>) -> Result<(), Error> {
         if self.ty == MaintenanceType::Delete {
             bail!("datastore is being deleted");

pbs-datastore/src/datastore.rs

@@ -104,8 +104,26 @@ impl Clone for DataStore {
 impl Drop for DataStore {
     fn drop(&mut self) {
         if let Some(operation) = self.operation {
-            if let Err(e) = update_active_operations(self.name(), operation, -1) {
-                log::error!("could not update active operations - {}", e);
+            let mut last_task = false;
+            match update_active_operations(self.name(), operation, -1) {
+                Err(e) => log::error!("could not update active operations - {}", e),
+                Ok(updated_operations) => {
+                    last_task = updated_operations.read + updated_operations.write == 0;
+                }
+            }
+
+            // remove datastore from cache iff
+            //  - last task finished, and
+            //  - datastore is in a maintenance mode that mandates it
+            let remove_from_cache = last_task
+                && pbs_config::datastore::config()
+                    .and_then(|(s, _)| s.lookup::<DataStoreConfig>("datastore", self.name()))
+                    .map_or(false, |c| {
+                        c.get_maintenance_mode().map_or(false, |m| m.is_offline())
+                    });
+
+            if remove_from_cache {
+                DATASTORE_MAP.lock().unwrap().remove(self.name());
             }
         }
     }
@@ -193,6 +211,24 @@ impl DataStore {
         Ok(())
     }
 
+    /// trigger clearing cache entry based on maintenance mode. Entry will only
+    /// be cleared iff there is no other task running, if there is, the end of the
+    /// last running task will trigger the clearing of the cache entry.
+    pub fn update_datastore_cache(name: &str) -> Result<(), Error> {
+        let (config, _digest) = pbs_config::datastore::config()?;
+        let datastore: DataStoreConfig = config.lookup("datastore", name)?;
+        if datastore
+            .get_maintenance_mode()
+            .map_or(false, |m| m.is_offline())
+        {
+            // the datastore drop handler does the checking if tasks are running and clears the
+            // cache entry, so we just have to trigger it here
+            let _ = DataStore::lookup_datastore(name, Some(Operation::Lookup));
+        }
+
+        Ok(())
+    }
+
     /// Open a raw database given a name and a path.
     ///
     /// # Safety

pbs-datastore/src/task_tracking.rs

@@ -91,15 +91,23 @@ pub fn get_active_operations_locked(
     Ok((data, lock.unwrap()))
 }
 
-pub fn update_active_operations(name: &str, operation: Operation, count: i64) -> Result<(), Error> {
+pub fn update_active_operations(
+    name: &str,
+    operation: Operation,
+    count: i64,
+) -> Result<ActiveOperationStats, Error> {
     let path = PathBuf::from(format!("{}/{}", crate::ACTIVE_OPERATIONS_DIR, name));
 
     let (_lock, options) = open_lock_file(name)?;
 
     let pid = std::process::id();
     let starttime = procfs::PidStat::read_from_pid(Pid::from_raw(pid as pid_t))?.starttime;
-    let mut updated = false;
 
+    let mut updated_active_operations = match operation {
+        Operation::Read => ActiveOperationStats { read: 1, write: 0 },
+        Operation::Write => ActiveOperationStats { read: 0, write: 1 },
+        Operation::Lookup => ActiveOperationStats { read: 0, write: 0 },
+    };
     let mut updated_tasks: Vec<TaskOperations> = match file_read_optional_string(&path)? {
         Some(data) => serde_json::from_str::<Vec<TaskOperations>>(&data)?
             .iter_mut()
@@ -108,12 +116,12 @@ pub fn update_active_operations(name: &str, operation: Operation, count: i64) ->
                     Some(stat) if pid == task.pid && stat.starttime != task.starttime => None,
                     Some(_) => {
                         if pid == task.pid {
-                            updated = true;
                             match operation {
                                 Operation::Read => task.active_operations.read += count,
                                 Operation::Write => task.active_operations.write += count,
                                 Operation::Lookup => (), // no IO must happen there
                             };
+                            updated_active_operations = task.active_operations;
                         }
                         Some(task.clone())
                     }
@@ -124,15 +132,11 @@ pub fn update_active_operations(name: &str, operation: Operation, count: i64) ->
         None => Vec::new(),
     };
 
-    if !updated {
+    if updated_tasks.is_empty() {
         updated_tasks.push(TaskOperations {
             pid,
             starttime,
-            active_operations: match operation {
-                Operation::Read => ActiveOperationStats { read: 1, write: 0 },
-                Operation::Write => ActiveOperationStats { read: 0, write: 1 },
-                Operation::Lookup => ActiveOperationStats { read: 0, write: 0 },
-            },
+            active_operations: updated_active_operations,
         })
     }
     replace_file(
@@ -141,4 +145,5 @@ pub fn update_active_operations(name: &str, operation: Operation, count: i64) ->
         options,
         false,
     )
+    .map(|_| updated_active_operations)
 }

src/api2/config/datastore.rs

@@ -389,7 +389,9 @@ pub fn update_datastore(
         data.tuning = update.tuning;
     }
 
+    let mut maintenance_mode_changed = false;
     if update.maintenance_mode.is_some() {
+        maintenance_mode_changed = data.maintenance_mode != update.maintenance_mode;
         data.maintenance_mode = update.maintenance_mode;
     }
 
@@ -403,6 +405,25 @@ pub fn update_datastore(
         jobstate::update_job_last_run_time("garbage_collection", &name)?;
     }
 
+    // tell the proxy it might have to clear a cache entry
+    if maintenance_mode_changed {
+        tokio::spawn(async move {
+            if let Ok(proxy_pid) =
+                proxmox_rest_server::read_pid(pbs_buildcfg::PROXMOX_BACKUP_PROXY_PID_FN)
+            {
+                let sock = proxmox_rest_server::ctrl_sock_from_pid(proxy_pid);
+                let _ = proxmox_rest_server::send_raw_command(
+                    sock,
+                    &format!(
+                        "{{\"command\":\"update-datastore-cache\",\"args\":\"{}\"}}\n",
+                        &name
+                    ),
+                )
+                .await;
+            }
+        });
+    }
+
     Ok(())
 }
 

src/bin/proxmox-backup-proxy.rs

@@ -289,6 +289,16 @@ async fn run() -> Result<(), Error> {
         Ok(Value::Null)
     })?;
 
+    // clear cache entry for datastore that is in a specific maintenance mode
+    command_sock.register_command("update-datastore-cache".to_string(), |value| {
+        if let Some(name) = value.and_then(Value::as_str) {
+            if let Err(err) = DataStore::update_datastore_cache(name) {
+                log::error!("could not trigger update datastore cache: {err}");
+            }
+        }
+        Ok(Value::Null)
+    })?;
+
     let connections = proxmox_rest_server::connection::AcceptBuilder::new()
         .debug(debug)
         .rate_limiter_lookup(Arc::new(lookup_rate_limiter))