[PATCH proxmox-backup v3 03/10] api/datastore: refactor datastore creation helper logic

[Christian Ebner <c.ebner@proxmox.com>]

Move most of the logic from the datastore creation worker task helper
to the datastore implementation itself.

This has the advantage of better encapsulation, reducing the
interdependencies.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-datastore/src/datastore.rs | 139 +++++++++++++++++++++++++++++++--
 src/api2/config/datastore.rs   | 123 ++---------------------------
 2 files changed, 140 insertions(+), 122 deletions(-)

diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index e2d1ae67c..99adfde8a 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -18,8 +18,8 @@ use tracing::{info, warn};
 
 use proxmox_human_byte::HumanByte;
 use proxmox_s3_client::{
-    RequestCounterThresholds, S3Client, S3ClientConf, S3ClientOptions, S3ObjectKey, S3PathPrefix,
-    S3RateLimiterOptions, S3RequestCounterConfig, SharedRequestCounters,
+    RequestCounterThresholds, S3_HTTP_REQUEST_TIMEOUT, S3Client, S3ClientConf, S3ClientOptions,
+    S3ObjectKey, S3PathPrefix, S3RateLimiterOptions, S3RequestCounterConfig, SharedRequestCounters,
 };
 use proxmox_schema::ApiType;
 
@@ -32,9 +32,9 @@ use proxmox_worker_task::WorkerTaskContext;
 
 use pbs_api_types::{
     ArchiveType, Authid, BackupGroupDeleteStats, BackupNamespace, BackupType, ChunkOrder,
-    DataStoreConfig, DatastoreBackendConfig, DatastoreBackendType, GarbageCollectionCacheStats,
-    GarbageCollectionStatus, MAX_NAMESPACE_DEPTH, MaintenanceMode, MaintenanceType, Operation,
-    S3Statistics, UPID,
+    DataStoreConfig, DatastoreBackendConfig, DatastoreBackendType, DatastoreTuning,
+    GarbageCollectionCacheStats, GarbageCollectionStatus, MAX_NAMESPACE_DEPTH, MaintenanceMode,
+    MaintenanceType, Operation, S3Statistics, UPID,
 };
 use pbs_config::s3::S3_CFG_TYPE_ID;
 use pbs_config::{BackupLockGuard, ConfigVersionCache};
@@ -392,6 +392,17 @@ impl DatastoreThreadSettings {
     }
 }
 
+#[derive(Default, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "kebab-case")]
+/// S3 in-use marker content
+///
+/// Allows to de-/serialize the contents of the in-use object contents stored on S3 backends to
+/// protect against access by multiple PBS instances.
+struct InUseContent {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    hostname: Option<String>,
+}
+
 /// Returns the parsed datastore config (`datastore.cfg`) and its
 /// generation.
 ///
@@ -468,6 +479,124 @@ impl DataStore {
         })
     }
 
+    /// Create or reuse the chunk store from given datastore configuration
+    pub fn construct_chunk_store(
+        store_config: &mut DataStoreConfig,
+        tuning: &DatastoreTuning,
+        reuse_existing: bool,
+        overwrite_in_use: bool,
+    ) -> Result<(), Error> {
+        let (backend_type, backend_s3_client) =
+            match DataStore::s3_client_and_backend_from_datastore_config(store_config)? {
+                (backend_type, Some(s3_client)) => {
+                    if !overwrite_in_use {
+                        let object_key = S3ObjectKey::try_from(S3_DATASTORE_IN_USE_MARKER)
+                            .context("failed to generate s3 object key")?;
+                        if let Some(response) = proxmox_async::runtime::block_on(
+                            s3_client.get_object(object_key.clone()),
+                        )
+                        .context("failed to get in-use marker from bucket")?
+                        {
+                            let content =
+                                proxmox_async::runtime::block_on(response.content.collect())
+                                    .unwrap_or_default();
+                            let content =
+                                String::from_utf8(content.to_bytes().to_vec()).unwrap_or_default();
+                            let in_use: InUseContent =
+                                serde_json::from_str(&content).unwrap_or_default();
+                            if let Some(hostname) = in_use.hostname {
+                                bail!(
+                                    "Bucket already contains datastore in use by host {hostname}"
+                                );
+                            } else {
+                                bail!("Bucket already contains datastore in use");
+                            }
+                        }
+                    }
+
+                    (backend_type, Some(Arc::new(s3_client)))
+                }
+                (backend_type, None) => (backend_type, None),
+            };
+
+        let path: PathBuf = store_config.absolute_path().into();
+
+        let chunk_store = if reuse_existing && backend_type == DatastoreBackendType::Filesystem {
+            ChunkStore::verify_chunkstore(&path).and_then(|_| {
+                // Must be the only instance accessing and locking the chunk store,
+                // dropping will close all other locks from this process on the lockfile as well.
+                ChunkStore::open(
+                    &store_config.name,
+                    &path,
+                    tuning.sync_level.unwrap_or_default(),
+                )
+            })?
+        } else {
+            if !reuse_existing && backend_type == DatastoreBackendType::Filesystem {
+                if let Ok(dir) = std::fs::read_dir(&path) {
+                    for file in dir {
+                        let name = file?.file_name();
+                        let name = name.to_str();
+                        if !name.is_some_and(|name| name.starts_with('.') || name == "lost+found") {
+                            bail!("datastore path not empty");
+                        }
+                    }
+                }
+            }
+            if reuse_existing && backend_type == DatastoreBackendType::S3 {
+                let chunks_path = path.join(".chunks");
+                if let Err(err) = std::fs::remove_dir_all(&chunks_path) {
+                    if err.kind() != std::io::ErrorKind::NotFound {
+                        return Err(err).with_context(|| {
+                            format!("failed to remove pre-existing chunks in {chunks_path:?}")
+                        });
+                    }
+                }
+                // starting out in maintenance mode s3-refresh,
+                // so no other operation will start until done with that.
+                store_config.set_maintenance_mode(Some(MaintenanceMode {
+                    ty: MaintenanceType::S3Refresh,
+                    message: None,
+                }))?;
+            }
+            let backup_user = pbs_config::backup_user()?;
+            ChunkStore::create(
+                &store_config.name,
+                path.clone(),
+                backup_user.uid,
+                backup_user.gid,
+                tuning.sync_level.unwrap_or_default(),
+            )?
+        };
+
+        if let Some(ref s3_client) = backend_s3_client {
+            let object_key = S3ObjectKey::try_from(S3_DATASTORE_IN_USE_MARKER)
+                .context("failed to generate s3 object key")?;
+            let content = serde_json::to_string(&InUseContent {
+                hostname: Some(proxmox_sys::nodename().to_string()),
+            })
+            .context("failed to encode hostname")?;
+            proxmox_async::runtime::block_on(s3_client.put_object(
+                object_key,
+                hyper::body::Bytes::from(content).into(),
+                Some(S3_HTTP_REQUEST_TIMEOUT),
+                true,
+            ))
+            .context("failed to upload in-use marker for datastore")?;
+        }
+
+        if tuning.gc_atime_safety_check.unwrap_or(true) {
+            chunk_store
+                .check_fs_atime_updates(true, backend_s3_client)
+                .context("access time safety check failed")?;
+            info!("Access time update check successful.");
+        } else {
+            info!("Access time update check skipped.");
+        }
+
+        Ok(())
+    }
+
     /// Get the backend for this datastore based on it's configuration
     pub fn backend(&self) -> Result<DatastoreBackend, Error> {
         let backend = match self.backend_type() {
diff --git a/src/api2/config/datastore.rs b/src/api2/config/datastore.rs
index fae289e70..86bf9dbf1 100644
--- a/src/api2/config/datastore.rs
+++ b/src/api2/config/datastore.rs
@@ -1,11 +1,9 @@
 use std::path::{Path, PathBuf};
-use std::sync::Arc;
 
 use ::serde::{Deserialize, Serialize};
 use anyhow::{Context, Error, bail, format_err};
-use http_body_util::BodyExt;
 use serde_json::Value;
-use tracing::{info, warn};
+use tracing::warn;
 
 use proxmox_router::{Permission, Router, RpcEnvironment, RpcEnvironmentType, http_bail};
 use proxmox_schema::{ApiType, api, param_bail};
@@ -14,12 +12,11 @@ use proxmox_uuid::Uuid;
 
 use pbs_api_types::{
     Authid, DATASTORE_SCHEMA, DataStoreConfig, DataStoreConfigUpdater, DatastoreBackendType,
-    DatastoreNotify, KeepOptions, MaintenanceMode, MaintenanceType, PRIV_DATASTORE_ALLOCATE,
-    PRIV_DATASTORE_AUDIT, PRIV_DATASTORE_MODIFY, PRIV_SYS_MODIFY, PROXMOX_CONFIG_DIGEST_SCHEMA,
-    PruneJobConfig, PruneJobOptions, UPID_SCHEMA,
+    DatastoreNotify, KeepOptions, MaintenanceMode, PRIV_DATASTORE_ALLOCATE, PRIV_DATASTORE_AUDIT,
+    PRIV_DATASTORE_MODIFY, PRIV_SYS_MODIFY, PROXMOX_CONFIG_DIGEST_SCHEMA, PruneJobConfig,
+    PruneJobOptions, UPID_SCHEMA,
 };
 use pbs_config::BackupLockGuard;
-use pbs_datastore::chunk_store::ChunkStore;
 
 use crate::api2::admin::datastore::do_mount_device;
 use crate::api2::admin::prune::list_prune_jobs;
@@ -31,20 +28,12 @@ use crate::api2::config::tape_backup_job::{delete_tape_backup_job, list_tape_bac
 use crate::api2::config::verify::delete_verification_job;
 use pbs_config::CachedUserInfo;
 
-use pbs_datastore::{DataStore, S3_DATASTORE_IN_USE_MARKER, get_datastore_mount_status};
+use pbs_datastore::{DataStore, get_datastore_mount_status};
 use proxmox_rest_server::WorkerTask;
-use proxmox_s3_client::{S3_HTTP_REQUEST_TIMEOUT, S3ObjectKey};
 
 use crate::server::jobstate;
 use crate::tools::disks::unmount_by_mountpoint;
 
-#[derive(Default, serde::Deserialize, serde::Serialize)]
-#[serde(rename_all = "kebab-case")]
-struct InUseContent {
-    #[serde(skip_serializing_if = "Option::is_none")]
-    hostname: Option<String>,
-}
-
 #[api(
     input: {
         properties: {},
@@ -130,107 +119,7 @@ pub(crate) fn do_create_datastore(
         UnmountGuard::new(None)
     };
 
-    let (backend_type, backend_s3_client) =
-        match DataStore::s3_client_and_backend_from_datastore_config(&datastore)? {
-            (backend_type, Some(s3_client)) => {
-                if !overwrite_in_use {
-                    let object_key = S3ObjectKey::try_from(S3_DATASTORE_IN_USE_MARKER)
-                        .context("failed to generate s3 object key")?;
-                    if let Some(response) =
-                        proxmox_async::runtime::block_on(s3_client.get_object(object_key.clone()))
-                            .context("failed to get in-use marker from bucket")?
-                    {
-                        let content = proxmox_async::runtime::block_on(response.content.collect())
-                            .unwrap_or_default();
-                        let content =
-                            String::from_utf8(content.to_bytes().to_vec()).unwrap_or_default();
-                        let in_use: InUseContent =
-                            serde_json::from_str(&content).unwrap_or_default();
-                        if let Some(hostname) = in_use.hostname {
-                            bail!("Bucket already contains datastore in use by host {hostname}");
-                        } else {
-                            bail!("Bucket already contains datastore in use");
-                        }
-                    }
-                }
-
-                (backend_type, Some(Arc::new(s3_client)))
-            }
-            (backend_type, None) => (backend_type, None),
-        };
-
-    let chunk_store = if reuse_datastore && backend_type == DatastoreBackendType::Filesystem {
-        ChunkStore::verify_chunkstore(&path).and_then(|_| {
-            // Must be the only instance accessing and locking the chunk store,
-            // dropping will close all other locks from this process on the lockfile as well.
-            ChunkStore::open(
-                &datastore.name,
-                &path,
-                tuning.sync_level.unwrap_or_default(),
-            )
-        })?
-    } else {
-        if !reuse_datastore && backend_type == DatastoreBackendType::Filesystem {
-            if let Ok(dir) = std::fs::read_dir(&path) {
-                for file in dir {
-                    let name = file?.file_name();
-                    let name = name.to_str();
-                    if !name.is_some_and(|name| name.starts_with('.') || name == "lost+found") {
-                        bail!("datastore path not empty");
-                    }
-                }
-            }
-        }
-        if reuse_datastore && backend_type == DatastoreBackendType::S3 {
-            let chunks_path = path.join(".chunks");
-            if let Err(err) = std::fs::remove_dir_all(&chunks_path) {
-                if err.kind() != std::io::ErrorKind::NotFound {
-                    return Err(err).with_context(|| {
-                        format!("failed to remove pre-existing chunks in {chunks_path:?}")
-                    });
-                }
-            }
-            // starting out in maintenance mode s3-refresh,
-            // so no other operation will start until done with that.
-            datastore.set_maintenance_mode(Some(MaintenanceMode {
-                ty: MaintenanceType::S3Refresh,
-                message: None,
-            }))?;
-        }
-        let backup_user = pbs_config::backup_user()?;
-        ChunkStore::create(
-            &datastore.name,
-            path.clone(),
-            backup_user.uid,
-            backup_user.gid,
-            tuning.sync_level.unwrap_or_default(),
-        )?
-    };
-
-    if let Some(ref s3_client) = backend_s3_client {
-        let object_key = S3ObjectKey::try_from(S3_DATASTORE_IN_USE_MARKER)
-            .context("failed to generate s3 object key")?;
-        let content = serde_json::to_string(&InUseContent {
-            hostname: Some(proxmox_sys::nodename().to_string()),
-        })
-        .context("failed to encode hostname")?;
-        proxmox_async::runtime::block_on(s3_client.put_object(
-            object_key,
-            hyper::body::Bytes::from(content).into(),
-            Some(S3_HTTP_REQUEST_TIMEOUT),
-            true,
-        ))
-        .context("failed to upload in-use marker for datastore")?;
-    }
-
-    if tuning.gc_atime_safety_check.unwrap_or(true) {
-        chunk_store
-            .check_fs_atime_updates(true, backend_s3_client)
-            .context("access time safety check failed")?;
-        info!("Access time update check successful.");
-    } else {
-        info!("Access time update check skipped.");
-    }
+    DataStore::construct_chunk_store(&mut datastore, &tuning, reuse_datastore, overwrite_in_use)?;
 
     config.set_data(&datastore.name, "datastore", &datastore)?;
 
-- 
2.47.3