[pbs-devel] [PATCH proxmox-backup RFC 04/10] datastore: separate functions into impl block
Hannes Laimer
h.laimer at proxmox.com
Tue Sep 3 14:33:55 CEST 2024
... based on whether they are reading/writing.
Signed-off-by: Hannes Laimer <h.laimer at proxmox.com>
---
pbs-datastore/src/datastore.rs | 1633 ++++++++++++++++----------------
1 file changed, 813 insertions(+), 820 deletions(-)
diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index be7767ff..70b31705 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -180,33 +180,178 @@ impl DataStore<WriteStore> {
}
Ok(store)
}
-}
+ /// Destroy a datastore. This requires that there are no active operations on the datastore.
+ ///
+ /// This is a synchronous operation and should be run in a worker-thread.
+ pub fn destroy(name: &str, destroy_data: bool) -> Result<(), Error> {
+ let config_lock = pbs_config::datastore::lock_config()?;
+
+ let (mut config, _digest) = pbs_config::datastore::config()?;
+ let mut datastore_config: DataStoreConfig = config.lookup("datastore", name)?;
+
+ datastore_config.set_maintenance_mode(Some(MaintenanceMode {
+ ty: MaintenanceType::Delete,
+ message: None,
+ }))?;
+
+ config.set_data(name, "datastore", &datastore_config)?;
+ pbs_config::datastore::save_config(&config)?;
+ drop(config_lock);
+
+ let (operations, _lock) = task_tracking::get_active_operations_locked(name)?;
+
+ if operations.read != 0 || operations.write != 0 {
+ bail!("datastore is currently in use");
+ }
+
+ let base = PathBuf::from(&datastore_config.path);
+
+ let mut ok = true;
+ if destroy_data {
+ let remove = |subdir, ok: &mut bool| {
+ if let Err(err) = std::fs::remove_dir_all(base.join(subdir)) {
+ if err.kind() != io::ErrorKind::NotFound {
+ warn!("failed to remove {subdir:?} subdirectory: {err}");
+ *ok = false;
+ }
+ }
+ };
+
+ info!("Deleting datastore data...");
+ remove("ns", &mut ok); // ns first
+ remove("ct", &mut ok);
+ remove("vm", &mut ok);
+ remove("host", &mut ok);
+
+ if ok {
+ if let Err(err) = std::fs::remove_file(base.join(".gc-status")) {
+ if err.kind() != io::ErrorKind::NotFound {
+ warn!("failed to remove .gc-status file: {err}");
+ ok = false;
+ }
+ }
+ }
+
+ // chunks get removed last and only if the backups were successfully deleted
+ if ok {
+ remove(".chunks", &mut ok);
+ }
+ }
+
+ // now the config
+ if ok {
+ info!("Removing datastore from config...");
+ let _lock = pbs_config::datastore::lock_config()?;
+ let _ = config.sections.remove(name);
+ pbs_config::datastore::save_config(&config)?;
+ }
+
+ // finally the lock & toplevel directory
+ if destroy_data {
+ if ok {
+ if let Err(err) = std::fs::remove_file(base.join(".lock")) {
+ if err.kind() != io::ErrorKind::NotFound {
+ warn!("failed to remove .lock file: {err}");
+ ok = false;
+ }
+ }
+ }
+
+ if ok {
+ info!("Finished deleting data.");
+
+ match std::fs::remove_dir(base) {
+ Ok(()) => info!("Removed empty datastore directory."),
+ Err(err) if err.kind() == io::ErrorKind::NotFound => {
+ // weird, but ok
+ }
+ Err(err) if err.is_errno(nix::errno::Errno::EBUSY) => {
+ warn!("Cannot delete datastore directory (is it a mount point?).")
+ }
+ Err(err) if err.is_errno(nix::errno::Errno::ENOTEMPTY) => {
+ warn!("Datastore directory not empty, not deleting.")
+ }
+ Err(err) => {
+ warn!("Failed to remove datastore directory: {err}");
+ }
+ }
+ } else {
+ info!("There were errors deleting data.");
+ }
+ }
+
+ 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);
+ }
+
+ Ok(())
+ }
+ /// removes all datastores that are not configured anymore
+ pub fn remove_unused_datastores() -> Result<(), Error> {
+ let (config, _digest) = pbs_config::datastore::config()?;
+
+ let mut map_read = DATASTORE_MAP_READ.lock().unwrap();
+ // removes all elements that are not in the config
+ map_read.retain(|key, _| config.sections.contains_key(key));
+
+ let mut map_write = DATASTORE_MAP_WRITE.lock().unwrap();
+ // removes all elements that are not in the config
+ map_write.retain(|key, _| config.sections.contains_key(key));
+ Ok(())
}
}
-impl DataStore {
- // This one just panics on everything
- #[doc(hidden)]
- pub(crate) unsafe fn new_test() -> Arc<Self> {
- Arc::new(Self {
- inner: unsafe { DataStoreImpl::new_test() },
- operation: None,
- })
+impl<T: CanRead + 'static> DataStore<T> {
+ /// Get a streaming iter over top-level backup groups of a datastore of a particular type,
+ /// filtered by `Ok` results
+ ///
+ /// The iterated item's result is already unwrapped, if it contained an error it will be
+ /// logged. Can be useful in iterator chain commands
+ pub fn iter_backup_type_ok(
+ self: &Arc<DataStore<T>>,
+ ns: BackupNamespace,
+ ty: BackupType,
+ ) -> Result<impl Iterator<Item = BackupGroup<T>> + 'static, Error> {
+ Ok(self.iter_backup_type(ns, ty)?.ok())
+ }
+
+ /// Get a streaming iter over top-level backup groups of a datatstore, filtered by Ok results
+ ///
+ /// The iterated item's result is already unwrapped, if it contained an error it will be
+ /// logged. Can be useful in iterator chain commands
+ pub fn iter_backup_groups_ok(
+ self: &Arc<DataStore<T>>,
+ ns: BackupNamespace,
+ ) -> Result<impl Iterator<Item = BackupGroup<T>> + 'static, Error> {
+ Ok(self.iter_backup_groups(ns)?.ok())
}
+}
- pub fn lookup_datastore(
+impl<T: CanRead> DataStore<T> {
+ fn open_datastore(
name: &str,
operation: Option<Operation>,
- ) -> Result<Arc<DataStore>, Error> {
+ cache_entry: Option<Arc<DataStoreImpl<T>>>,
+ ) -> Result<Arc<DataStore<T>>, Error> {
// Avoid TOCTOU between checking maintenance mode and updating active operation counter, as
// we use it to decide whether it is okay to delete the datastore.
- let _config_lock = pbs_config::datastore::lock_config()?;
-
// we could use the ConfigVersionCache's generation for staleness detection, but we load
// the config anyway -> just use digest, additional benefit: manual changes get detected
- let (config, digest) = pbs_config::datastore::config()?;
- let config: DataStoreConfig = config.lookup("datastore", name)?;
+ let (config, digest, _lock) = Self::read_config(name)?;
if let Some(maintenance_mode) = config.get_maintenance_mode() {
if let Err(error) = maintenance_mode.check(operation) {
@@ -214,11 +359,8 @@ impl DataStore {
}
}
- let mut datastore_cache = DATASTORE_MAP.lock().unwrap();
- let entry = datastore_cache.get(name);
-
// reuse chunk store so that we keep using the same process locker instance!
- let chunk_store = if let Some(datastore) = &entry {
+ let chunk_store = if let Some(datastore) = &cache_entry {
let last_digest = datastore.last_digest.as_ref();
if let Some(true) = last_digest.map(|last_digest| last_digest == &digest) {
if let Some(operation) = operation {
@@ -235,73 +377,59 @@ impl DataStore {
DatastoreTuning::API_SCHEMA
.parse_property_string(config.tuning.as_deref().unwrap_or(""))?,
)?;
- Arc::new(ChunkStore::open(
+ Arc::new(ChunkStore::<T>::open(
name,
&config.path,
tuning.sync_level.unwrap_or_default(),
)?)
};
- let datastore = DataStore::with_store_and_config(chunk_store, config, Some(digest))?;
-
- let datastore = Arc::new(datastore);
- datastore_cache.insert(name.to_string(), datastore.clone());
+ let datastore = Self::with_store_and_config(chunk_store, config, Some(digest))?;
if let Some(operation) = operation {
update_active_operations(name, operation, 1)?;
}
Ok(Arc::new(Self {
- inner: datastore,
+ inner: datastore.into(),
operation,
}))
}
+ fn with_store_and_config(
+ chunk_store: Arc<ChunkStore<T>>,
+ config: DataStoreConfig,
+ last_digest: Option<[u8; 32]>,
+ ) -> Result<DataStoreImpl<T>, Error> {
+ let mut gc_status_path = chunk_store.base_path();
+ gc_status_path.push(".gc-status");
- /// removes all datastores that are not configured anymore
- pub fn remove_unused_datastores() -> Result<(), Error> {
- let (config, _digest) = pbs_config::datastore::config()?;
-
- let mut map = DATASTORE_MAP.lock().unwrap();
- // removes all elements that are not in the config
- map.retain(|key, _| config.sections.contains_key(key));
- 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));
- }
+ let gc_status = if let Some(state) = file_read_optional_string(gc_status_path)? {
+ match serde_json::from_str(&state) {
+ Ok(state) => state,
+ Err(err) => {
+ log::error!("error reading gc-status: {}", err);
+ GarbageCollectionStatus::default()
+ }
+ }
+ } else {
+ GarbageCollectionStatus::default()
+ };
- Ok(())
- }
+ let tuning: DatastoreTuning = serde_json::from_value(
+ DatastoreTuning::API_SCHEMA
+ .parse_property_string(config.tuning.as_deref().unwrap_or(""))?,
+ )?;
- /// Open a raw database given a name and a path.
- ///
- /// # Safety
- /// See the safety section in `open_from_config`
- pub unsafe fn open_path(
- name: &str,
- path: impl AsRef<Path>,
- operation: Option<Operation>,
- ) -> Result<Arc<Self>, Error> {
- let path = path
- .as_ref()
- .to_str()
- .ok_or_else(|| format_err!("non-utf8 paths not supported"))?
- .to_owned();
- unsafe { Self::open_from_config(DataStoreConfig::new(name.to_owned(), path), operation) }
+ Ok(DataStoreImpl {
+ chunk_store,
+ gc_mutex: Mutex::new(()),
+ last_gc_status: Mutex::new(gc_status),
+ verify_new: config.verify_new.unwrap_or(false),
+ chunk_order: tuning.chunk_order.unwrap_or_default(),
+ last_digest,
+ sync_level: tuning.sync_level.unwrap_or_default(),
+ })
}
-
/// Open a datastore given a raw configuration.
///
/// # Safety
@@ -322,7 +450,7 @@ impl DataStore {
.parse_property_string(config.tuning.as_deref().unwrap_or(""))?,
)?;
let chunk_store =
- ChunkStore::open(&name, &config.path, tuning.sync_level.unwrap_or_default())?;
+ ChunkStore::<T>::open(&name, &config.path, tuning.sync_level.unwrap_or_default())?;
let inner = Arc::new(Self::with_store_and_config(
Arc::new(chunk_store),
config,
@@ -335,88 +463,24 @@ impl DataStore {
Ok(Arc::new(Self { inner, operation }))
}
+ pub fn get_chunk_iterator(
+ &self,
+ ) -> Result<
+ impl Iterator<Item = (Result<proxmox_sys::fs::ReadDirEntry, Error>, usize, bool)>,
+ Error,
+ > {
+ self.inner.chunk_store.get_chunk_iterator()
+ }
+ pub fn open_fixed_reader<P: AsRef<Path>>(
+ &self,
+ filename: P,
+ ) -> Result<FixedIndexReader, Error> {
+ let full_path = self.inner.chunk_store.relative_path(filename.as_ref());
- fn with_store_and_config(
- chunk_store: Arc<ChunkStore>,
- config: DataStoreConfig,
- last_digest: Option<[u8; 32]>,
- ) -> Result<DataStoreImpl, Error> {
- let mut gc_status_path = chunk_store.base_path();
- gc_status_path.push(".gc-status");
-
- let gc_status = if let Some(state) = file_read_optional_string(gc_status_path)? {
- match serde_json::from_str(&state) {
- Ok(state) => state,
- Err(err) => {
- log::error!("error reading gc-status: {}", err);
- GarbageCollectionStatus::default()
- }
- }
- } else {
- GarbageCollectionStatus::default()
- };
-
- let tuning: DatastoreTuning = serde_json::from_value(
- DatastoreTuning::API_SCHEMA
- .parse_property_string(config.tuning.as_deref().unwrap_or(""))?,
- )?;
-
- Ok(DataStoreImpl {
- chunk_store,
- gc_mutex: Mutex::new(()),
- last_gc_status: Mutex::new(gc_status),
- verify_new: config.verify_new.unwrap_or(false),
- chunk_order: tuning.chunk_order.unwrap_or_default(),
- last_digest,
- sync_level: tuning.sync_level.unwrap_or_default(),
- })
- }
-
- pub fn get_chunk_iterator(
- &self,
- ) -> Result<
- impl Iterator<Item = (Result<proxmox_sys::fs::ReadDirEntry, Error>, usize, bool)>,
- Error,
- > {
- self.inner.chunk_store.get_chunk_iterator()
- }
-
- pub fn create_fixed_writer<P: AsRef<Path>>(
- &self,
- filename: P,
- size: usize,
- chunk_size: usize,
- ) -> Result<FixedIndexWriter, Error> {
- let index = FixedIndexWriter::create(
- self.inner.chunk_store.clone(),
- filename.as_ref(),
- size,
- chunk_size,
- )?;
-
- Ok(index)
- }
-
- pub fn open_fixed_reader<P: AsRef<Path>>(
- &self,
- filename: P,
- ) -> Result<FixedIndexReader, Error> {
- let full_path = self.inner.chunk_store.relative_path(filename.as_ref());
-
- let index = FixedIndexReader::open(&full_path)?;
+ let index = FixedIndexReader::open(&full_path)?;
Ok(index)
}
-
- pub fn create_dynamic_writer<P: AsRef<Path>>(
- &self,
- filename: P,
- ) -> Result<DynamicIndexWriter, Error> {
- let index = DynamicIndexWriter::create(self.inner.chunk_store.clone(), filename.as_ref())?;
-
- Ok(index)
- }
-
pub fn open_dynamic_reader<P: AsRef<Path>>(
&self,
filename: P,
@@ -427,6 +491,22 @@ impl DataStore {
Ok(index)
}
+ /// Open a raw database given a name and a path.
+ ///
+ /// # Safety
+ /// See the safety section in `open_from_config`
+ pub unsafe fn open_path(
+ name: &str,
+ path: impl AsRef<Path>,
+ operation: Option<Operation>,
+ ) -> Result<Arc<Self>, Error> {
+ let path = path
+ .as_ref()
+ .to_str()
+ .ok_or_else(|| format_err!("non-utf8 paths not supported"))?
+ .to_owned();
+ unsafe { Self::open_from_config(DataStoreConfig::new(name.to_owned(), path), operation) }
+ }
pub fn open_index<P>(&self, filename: P) -> Result<Box<dyn IndexFile + Send>, Error>
where
@@ -466,121 +546,443 @@ impl DataStore {
Ok(())
}
-
- pub fn name(&self) -> &str {
- self.inner.chunk_store.name()
+ /// Returns if the given namespace exists on the datastore
+ pub fn namespace_exists(&self, ns: &BackupNamespace) -> bool {
+ let mut path = self.base_path();
+ path.push(ns.path());
+ path.exists()
}
+ /// Returns the time of the last successful backup
+ ///
+ /// Or None if there is no backup in the group (or the group dir does not exist).
+ pub fn last_successful_backup(
+ self: &Arc<Self>,
+ ns: &BackupNamespace,
+ backup_group: &pbs_api_types::BackupGroup,
+ ) -> Result<Option<i64>, Error> {
+ let backup_group = self.backup_group(ns.clone(), backup_group.clone());
- pub fn base_path(&self) -> PathBuf {
- self.inner.chunk_store.base_path()
- }
+ let group_path = backup_group.full_group_path();
- /// Returns the absolute path for a backup namespace on this datastore
- pub fn namespace_path(&self, ns: &BackupNamespace) -> PathBuf {
- let mut path = self.base_path();
- path.reserve(ns.path_len());
- for part in ns.components() {
- path.push("ns");
- path.push(part);
+ if group_path.exists() {
+ backup_group.last_successful_backup()
+ } else {
+ Ok(None)
}
- path
- }
-
- /// Returns the absolute path for a backup_type
- pub fn type_path(&self, ns: &BackupNamespace, backup_type: BackupType) -> PathBuf {
- let mut full_path = self.namespace_path(ns);
- full_path.push(backup_type.to_string());
- full_path
}
-
- /// Returns the absolute path for a backup_group
- pub fn group_path(
+ /// Returns the backup owner.
+ ///
+ /// The backup owner is the entity who first created the backup group.
+ pub fn get_owner(
&self,
ns: &BackupNamespace,
backup_group: &pbs_api_types::BackupGroup,
- ) -> PathBuf {
- let mut full_path = self.namespace_path(ns);
- full_path.push(backup_group.to_string());
- full_path
+ ) -> Result<Authid, Error> {
+ let full_path = self.owner_path(ns, backup_group);
+ let owner = proxmox_sys::fs::file_read_firstline(full_path)?;
+ owner
+ .trim_end() // remove trailing newline
+ .parse()
+ .map_err(|err| format_err!("parsing owner for {backup_group} failed: {err}"))
}
- /// Returns the absolute path for backup_dir
- pub fn snapshot_path(
+ pub fn owns_backup(
&self,
ns: &BackupNamespace,
- backup_dir: &pbs_api_types::BackupDir,
- ) -> PathBuf {
- let mut full_path = self.namespace_path(ns);
- full_path.push(backup_dir.to_string());
- full_path
- }
-
- /// Create a backup namespace.
- pub fn create_namespace(
- self: &Arc<Self>,
- parent: &BackupNamespace,
- name: String,
- ) -> Result<BackupNamespace, Error> {
- if !self.namespace_exists(parent) {
- bail!("cannot create new namespace, parent {parent} doesn't already exists");
- }
-
- // construct ns before mkdir to enforce max-depth and name validity
- let ns = BackupNamespace::from_parent_ns(parent, name)?;
-
- let mut ns_full_path = self.base_path();
- ns_full_path.push(ns.path());
-
- std::fs::create_dir_all(ns_full_path)?;
+ backup_group: &pbs_api_types::BackupGroup,
+ auth_id: &Authid,
+ ) -> Result<bool, Error> {
+ let owner = self.get_owner(ns, backup_group)?;
- Ok(ns)
+ Ok(check_backup_owner(&owner, auth_id).is_ok())
}
-
- /// Returns if the given namespace exists on the datastore
- pub fn namespace_exists(&self, ns: &BackupNamespace) -> bool {
- let mut path = self.base_path();
- path.push(ns.path());
- path.exists()
+ /// Get a streaming iter over single-level backup namespaces of a datatstore
+ ///
+ /// The iterated item is still a Result that can contain errors from rather unexptected FS or
+ /// parsing errors.
+ pub fn iter_backup_ns(
+ self: &Arc<DataStore<T>>,
+ ns: BackupNamespace,
+ ) -> Result<ListNamespaces<T>, Error> {
+ ListNamespaces::new(Arc::clone(self), ns)
}
- /// Remove all backup groups of a single namespace level but not the namespace itself.
- ///
- /// Does *not* descends into child-namespaces and doesn't remoes the namespace itself either.
+ /// Get a streaming iter over single-level backup namespaces of a datatstore, filtered by Ok
///
- /// Returns true if all the groups were removed, and false if some were protected.
- pub fn remove_namespace_groups(self: &Arc<Self>, ns: &BackupNamespace) -> Result<bool, Error> {
- // FIXME: locking? The single groups/snapshots are already protected, so may not be
- // necessary (depends on what we all allow to do with namespaces)
- log::info!("removing all groups in namespace {}:/{ns}", self.name());
+ /// The iterated item's result is already unwrapped, if it contained an error it will be
+ /// logged. Can be useful in iterator chain commands
+ pub fn iter_backup_ns_ok(
+ self: &Arc<DataStore<T>>,
+ ns: BackupNamespace,
+ ) -> Result<impl Iterator<Item = BackupNamespace>, Error> {
+ let this = Arc::clone(self);
+ Ok(
+ ListNamespaces::new(Arc::clone(self), ns)?.filter_map(move |ns| match ns {
+ Ok(ns) => Some(ns),
+ Err(err) => {
+ log::error!("list groups error on datastore {} - {}", this.name(), err);
+ None
+ }
+ }),
+ )
+ }
- let mut removed_all_groups = true;
+ /// Get a streaming iter over single-level backup namespaces of a datatstore
+ ///
+ /// The iterated item is still a Result that can contain errors from rather unexptected FS or
+ /// parsing errors.
+ pub fn recursive_iter_backup_ns(
+ self: &Arc<DataStore<T>>,
+ ns: BackupNamespace,
+ ) -> Result<ListNamespacesRecursive<T>, Error> {
+ ListNamespacesRecursive::new(Arc::clone(self), ns)
+ }
- for group in self.iter_backup_groups(ns.to_owned())? {
- let delete_stats = group?.destroy()?;
- removed_all_groups = removed_all_groups && delete_stats.all_removed();
+ /// Get a streaming iter over single-level backup namespaces of a datatstore, filtered by Ok
+ ///
+ /// The iterated item's result is already unwrapped, if it contained an error it will be
+ /// logged. Can be useful in iterator chain commands
+ pub fn recursive_iter_backup_ns_ok(
+ self: &Arc<DataStore<T>>,
+ ns: BackupNamespace,
+ max_depth: Option<usize>,
+ ) -> Result<impl Iterator<Item = BackupNamespace>, Error> {
+ let this = Arc::clone(self);
+ Ok(if let Some(depth) = max_depth {
+ ListNamespacesRecursive::<T>::new_max_depth(Arc::clone(self), ns, depth)?
+ } else {
+ ListNamespacesRecursive::<T>::new(Arc::clone(self), ns)?
}
-
- let base_file = std::fs::File::open(self.base_path())?;
- let base_fd = base_file.as_raw_fd();
- for ty in BackupType::iter() {
- let mut ty_dir = ns.path();
- ty_dir.push(ty.to_string());
- // best effort only, but we probably should log the error
- if let Err(err) = unlinkat(Some(base_fd), &ty_dir, UnlinkatFlags::RemoveDir) {
- if err != nix::errno::Errno::ENOENT {
- log::error!("failed to remove backup type {ty} in {ns} - {err}");
- }
+ .filter_map(move |ns| match ns {
+ Ok(ns) => Some(ns),
+ Err(err) => {
+ log::error!("list groups error on datastore {} - {}", this.name(), err);
+ None
}
- }
+ }))
+ }
- Ok(removed_all_groups)
+ /// Get a streaming iter over top-level backup groups of a datatstore of a particular type.
+ ///
+ /// The iterated item is still a Result that can contain errors from rather unexptected FS or
+ /// parsing errors.
+ pub fn iter_backup_type(
+ self: &Arc<DataStore<T>>,
+ ns: BackupNamespace,
+ ty: BackupType,
+ ) -> Result<ListGroupsType<T>, Error> {
+ ListGroupsType::new(Arc::clone(self), ns, ty)
}
- /// Remove a complete backup namespace optionally including all it's, and child namespaces',
- /// groups. If `removed_groups` is false this only prunes empty namespaces.
+ /// Get a streaming iter over top-level backup groups of a datatstore
///
- /// Returns true if everything requested, and false if some groups were protected or if some
- /// namespaces weren't empty even though all groups were deleted (race with new backup)
+ /// The iterated item is still a Result that can contain errors from rather unexptected FS or
+ /// parsing errors.
+ pub fn iter_backup_groups(
+ self: &Arc<Self>,
+ ns: BackupNamespace,
+ ) -> Result<ListGroups<T>, Error> {
+ ListGroups::new(Arc::clone(self), ns)
+ }
+
+ /// Get a in-memory vector for all top-level backup groups of a datatstore
+ ///
+ /// NOTE: using the iterator directly is most often more efficient w.r.t. memory usage
+ pub fn list_backup_groups(
+ self: &Arc<DataStore<T>>,
+ ns: BackupNamespace,
+ ) -> Result<Vec<BackupGroup<T>>, Error> {
+ ListGroups::new(Arc::clone(self), ns)?.collect()
+ }
+
+ pub fn list_images(&self) -> Result<Vec<PathBuf>, Error> {
+ let base = self.base_path();
+
+ let mut list = vec![];
+
+ use walkdir::WalkDir;
+
+ let walker = WalkDir::new(base).into_iter();
+
+ // make sure we skip .chunks (and other hidden files to keep it simple)
+ fn is_hidden(entry: &walkdir::DirEntry) -> bool {
+ entry
+ .file_name()
+ .to_str()
+ .map(|s| s.starts_with('.'))
+ .unwrap_or(false)
+ }
+ let handle_entry_err = |err: walkdir::Error| {
+ // first, extract the actual IO error and the affected path
+ let (inner, path) = match (err.io_error(), err.path()) {
+ (None, _) => return Ok(()), // not an IO-error
+ (Some(inner), Some(path)) => (inner, path),
+ (Some(inner), None) => bail!("unexpected error on datastore traversal: {inner}"),
+ };
+ if inner.kind() == io::ErrorKind::PermissionDenied {
+ if err.depth() <= 1 && path.ends_with("lost+found") {
+ // allow skipping of (root-only) ext4 fsck-directory on EPERM ..
+ return Ok(());
+ }
+ // .. but do not ignore EPERM in general, otherwise we might prune too many chunks.
+ // E.g., if users messed up with owner/perms on a rsync
+ bail!("cannot continue garbage-collection safely, permission denied on: {path:?}");
+ } else if inner.kind() == io::ErrorKind::NotFound {
+ log::info!("ignoring vanished file: {path:?}");
+ return Ok(());
+ } else {
+ bail!("unexpected error on datastore traversal: {inner} - {path:?}");
+ }
+ };
+ for entry in walker.filter_entry(|e| !is_hidden(e)) {
+ let path = match entry {
+ Ok(entry) => entry.into_path(),
+ Err(err) => {
+ handle_entry_err(err)?;
+ continue;
+ }
+ };
+ if let Ok(archive_type) = ArchiveType::from_path(&path) {
+ if archive_type == ArchiveType::FixedIndex
+ || archive_type == ArchiveType::DynamicIndex
+ {
+ list.push(path);
+ }
+ }
+ }
+
+ Ok(list)
+ }
+
+ pub fn last_gc_status(&self) -> GarbageCollectionStatus {
+ self.inner.last_gc_status.lock().unwrap().clone()
+ }
+
+ pub fn try_shared_chunk_store_lock(&self) -> Result<ProcessLockSharedGuard, Error> {
+ self.inner.chunk_store.try_shared_lock()
+ }
+ pub fn stat_chunk(&self, digest: &[u8; 32]) -> Result<std::fs::Metadata, Error> {
+ let (chunk_path, _digest_str) = self.inner.chunk_store.chunk_path(digest);
+ std::fs::metadata(chunk_path).map_err(Error::from)
+ }
+
+ pub fn load_chunk(&self, digest: &[u8; 32]) -> Result<DataBlob, Error> {
+ let (chunk_path, digest_str) = self.inner.chunk_store.chunk_path(digest);
+
+ proxmox_lang::try_block!({
+ let mut file = std::fs::File::open(&chunk_path)?;
+ DataBlob::load_from_reader(&mut file)
+ })
+ .map_err(|err| {
+ format_err!(
+ "store '{}', unable to load chunk '{}' - {}",
+ self.name(),
+ digest_str,
+ err,
+ )
+ })
+ }
+ /// returns a list of chunks sorted by their inode number on disk chunks that couldn't get
+ /// stat'ed are placed at the end of the list
+ pub fn get_chunks_in_order<F, A>(
+ &self,
+ index: &(dyn IndexFile + Send),
+ skip_chunk: F,
+ check_abort: A,
+ ) -> Result<Vec<(usize, u64)>, Error>
+ where
+ F: Fn(&[u8; 32]) -> bool,
+ A: Fn(usize) -> Result<(), Error>,
+ {
+ let index_count = index.index_count();
+ let mut chunk_list = Vec::with_capacity(index_count);
+ use std::os::unix::fs::MetadataExt;
+ for pos in 0..index_count {
+ check_abort(pos)?;
+
+ let info = index.chunk_info(pos).unwrap();
+
+ if skip_chunk(&info.digest) {
+ continue;
+ }
+
+ let ino = match self.inner.chunk_order {
+ ChunkOrder::Inode => {
+ match self.stat_chunk(&info.digest) {
+ Err(_) => u64::MAX, // could not stat, move to end of list
+ Ok(metadata) => metadata.ino(),
+ }
+ }
+ ChunkOrder::None => 0,
+ };
+
+ chunk_list.push((pos, ino));
+ }
+
+ match self.inner.chunk_order {
+ // sorting by inode improves data locality, which makes it lots faster on spinners
+ ChunkOrder::Inode => {
+ chunk_list.sort_unstable_by(|(_, ino_a), (_, ino_b)| ino_a.cmp(ino_b))
+ }
+ ChunkOrder::None => {}
+ }
+
+ Ok(chunk_list)
+ }
+
+ /// Open a backup group from this datastore.
+ pub fn backup_group(
+ self: &Arc<Self>,
+ ns: BackupNamespace,
+ group: pbs_api_types::BackupGroup,
+ ) -> BackupGroup<T> {
+ BackupGroup::new(Arc::clone(self), ns, group)
+ }
+
+ /// Open a backup group from this datastore.
+ pub fn backup_group_from_parts<D>(
+ self: &Arc<Self>,
+ ns: BackupNamespace,
+ ty: BackupType,
+ id: D,
+ ) -> BackupGroup<T>
+ where
+ D: Into<String>,
+ {
+ self.backup_group(ns, (ty, id.into()).into())
+ }
+
+ /*
+ /// Open a backup group from this datastore by backup group path such as `vm/100`.
+ ///
+ /// Convenience method for `store.backup_group(path.parse()?)`
+ pub fn backup_group_from_path(self: &Arc<Self>, path: &str) -> Result<BackupGroup, Error> {
+ todo!("split out the namespace");
+ }
+ */
+
+ /// Open a snapshot (backup directory) from this datastore.
+ pub fn backup_dir(
+ self: &Arc<Self>,
+ ns: BackupNamespace,
+ dir: pbs_api_types::BackupDir,
+ ) -> Result<BackupDir<T>, Error> {
+ BackupDir::with_group(self.backup_group(ns, dir.group), dir.time)
+ }
+
+ /// Open a snapshot (backup directory) from this datastore.
+ pub fn backup_dir_from_parts<D>(
+ self: &Arc<Self>,
+ ns: BackupNamespace,
+ ty: BackupType,
+ id: D,
+ time: i64,
+ ) -> Result<BackupDir<T>, Error>
+ where
+ D: Into<String>,
+ {
+ self.backup_dir(ns, (ty, id.into(), time).into())
+ }
+
+ /// Open a snapshot (backup directory) from this datastore with a cached rfc3339 time string.
+ pub fn backup_dir_with_rfc3339<D: Into<String>>(
+ self: &Arc<Self>,
+ group: BackupGroup<T>,
+ time_string: D,
+ ) -> Result<BackupDir<T>, Error> {
+ BackupDir::with_rfc3339(group, time_string.into())
+ }
+
+ /*
+ /// Open a snapshot (backup directory) from this datastore by a snapshot path.
+ pub fn backup_dir_from_path(self: &Arc<Self>, path: &str) -> Result<BackupDir, Error> {
+ todo!("split out the namespace");
+ }
+ */
+}
+
+impl<T: CanWrite> DataStore<T> {
+ pub fn create_fixed_writer<P: AsRef<Path>>(
+ &self,
+ filename: P,
+ size: usize,
+ chunk_size: usize,
+ ) -> Result<FixedIndexWriter<T>, Error> {
+ let index = FixedIndexWriter::create(
+ self.inner.chunk_store.clone(),
+ filename.as_ref(),
+ size,
+ chunk_size,
+ )?;
+
+ Ok(index)
+ }
+ pub fn create_dynamic_writer<P: AsRef<Path>>(
+ &self,
+ filename: P,
+ ) -> Result<DynamicIndexWriter<T>, Error> {
+ let index = DynamicIndexWriter::create(self.inner.chunk_store.clone(), filename.as_ref())?;
+
+ Ok(index)
+ }
+ /// Create a backup namespace.
+ pub fn create_namespace(
+ self: &Arc<Self>,
+ parent: &BackupNamespace,
+ name: String,
+ ) -> Result<BackupNamespace, Error> {
+ if !self.namespace_exists(parent) {
+ bail!("cannot create new namespace, parent {parent} doesn't already exists");
+ }
+
+ // construct ns before mkdir to enforce max-depth and name validity
+ let ns = BackupNamespace::from_parent_ns(parent, name)?;
+
+ let mut ns_full_path = self.base_path();
+ ns_full_path.push(ns.path());
+
+ std::fs::create_dir_all(ns_full_path)?;
+
+ Ok(ns)
+ }
+ /// Remove all backup groups of a single namespace level but not the namespace itself.
+ ///
+ /// Does *not* descends into child-namespaces and doesn't remoes the namespace itself either.
+ ///
+ /// Returns true if all the groups were removed, and false if some were protected.
+ pub fn remove_namespace_groups(self: &Arc<Self>, ns: &BackupNamespace) -> Result<bool, Error> {
+ // FIXME: locking? The single groups/snapshots are already protected, so may not be
+ // necessary (depends on what we all allow to do with namespaces)
+ log::info!("removing all groups in namespace {}:/{ns}", self.name());
+
+ let mut removed_all_groups = true;
+
+ for group in self.iter_backup_groups(ns.to_owned())? {
+ let delete_stats = group?.destroy()?;
+ removed_all_groups = removed_all_groups && delete_stats.all_removed();
+ }
+
+ let base_file = std::fs::File::open(self.base_path())?;
+ let base_fd = base_file.as_raw_fd();
+ for ty in BackupType::iter() {
+ let mut ty_dir = ns.path();
+ ty_dir.push(ty.to_string());
+ // best effort only, but we probably should log the error
+ if let Err(err) = unlinkat(Some(base_fd), &ty_dir, UnlinkatFlags::RemoveDir) {
+ if err != nix::errno::Errno::ENOENT {
+ log::error!("failed to remove backup type {ty} in {ns} - {err}");
+ }
+ }
+ }
+
+ Ok(removed_all_groups)
+ }
+
+ /// Remove a complete backup namespace optionally including all it's, and child namespaces',
+ /// groups. If `removed_groups` is false this only prunes empty namespaces.
+ ///
+ /// Returns true if everything requested, and false if some groups were protected or if some
+ /// namespaces weren't empty even though all groups were deleted (race with new backup)
pub fn remove_namespace_recursive(
self: &Arc<Self>,
ns: &BackupNamespace,
@@ -660,88 +1062,6 @@ impl DataStore {
backup_dir.destroy(force)
}
-
- /// Returns the time of the last successful backup
- ///
- /// Or None if there is no backup in the group (or the group dir does not exist).
- pub fn last_successful_backup(
- self: &Arc<Self>,
- ns: &BackupNamespace,
- backup_group: &pbs_api_types::BackupGroup,
- ) -> Result<Option<i64>, Error> {
- let backup_group = self.backup_group(ns.clone(), backup_group.clone());
-
- let group_path = backup_group.full_group_path();
-
- if group_path.exists() {
- backup_group.last_successful_backup()
- } else {
- Ok(None)
- }
- }
-
- /// Return the path of the 'owner' file.
- fn owner_path(&self, ns: &BackupNamespace, group: &pbs_api_types::BackupGroup) -> PathBuf {
- self.group_path(ns, group).join("owner")
- }
-
- /// Returns the backup owner.
- ///
- /// The backup owner is the entity who first created the backup group.
- pub fn get_owner(
- &self,
- ns: &BackupNamespace,
- backup_group: &pbs_api_types::BackupGroup,
- ) -> Result<Authid, Error> {
- let full_path = self.owner_path(ns, backup_group);
- let owner = proxmox_sys::fs::file_read_firstline(full_path)?;
- owner
- .trim_end() // remove trailing newline
- .parse()
- .map_err(|err| format_err!("parsing owner for {backup_group} failed: {err}"))
- }
-
- pub fn owns_backup(
- &self,
- ns: &BackupNamespace,
- backup_group: &pbs_api_types::BackupGroup,
- auth_id: &Authid,
- ) -> Result<bool, Error> {
- let owner = self.get_owner(ns, backup_group)?;
-
- Ok(check_backup_owner(&owner, auth_id).is_ok())
- }
-
- /// Set the backup owner.
- pub fn set_owner(
- &self,
- ns: &BackupNamespace,
- backup_group: &pbs_api_types::BackupGroup,
- auth_id: &Authid,
- force: bool,
- ) -> Result<(), Error> {
- let path = self.owner_path(ns, backup_group);
-
- let mut open_options = std::fs::OpenOptions::new();
- open_options.write(true);
- open_options.truncate(true);
-
- if force {
- open_options.create(true);
- } else {
- open_options.create_new(true);
- }
-
- let mut file = open_options
- .open(&path)
- .map_err(|err| format_err!("unable to create owner file {:?} - {}", path, err))?;
-
- writeln!(file, "{}", auth_id)
- .map_err(|err| format_err!("unable to write owner file {:?} - {}", path, err))?;
-
- Ok(())
- }
-
/// Create (if it does not already exists) and lock a backup group
///
/// And set the owner to 'userid'. If the group already exists, it returns the
@@ -784,224 +1104,71 @@ impl DataStore {
"another backup is already running",
)?;
let owner = self.get_owner(ns, backup_group)?; // just to be sure
- Ok((owner, guard))
- }
- Err(err) => bail!("unable to create backup group {:?} - {}", full_path, err),
- }
- }
-
- /// Creates a new backup snapshot inside a BackupGroup
- ///
- /// The BackupGroup directory needs to exist.
- pub fn create_locked_backup_dir(
- &self,
- ns: &BackupNamespace,
- backup_dir: &pbs_api_types::BackupDir,
- ) -> Result<(PathBuf, bool, DirLockGuard), Error> {
- let full_path = self.snapshot_path(ns, backup_dir);
- let relative_path = full_path.strip_prefix(self.base_path()).map_err(|err| {
- format_err!(
- "failed to produce correct path for backup {backup_dir} in namespace {ns}: {err}"
- )
- })?;
-
- let lock = || {
- lock_dir_noblock(
- &full_path,
- "snapshot",
- "internal error - tried creating snapshot that's already in use",
- )
- };
-
- match std::fs::create_dir(&full_path) {
- Ok(_) => Ok((relative_path.to_owned(), true, lock()?)),
- Err(ref e) if e.kind() == io::ErrorKind::AlreadyExists => {
- Ok((relative_path.to_owned(), false, lock()?))
- }
- Err(e) => Err(e.into()),
- }
- }
-
- /// Get a streaming iter over single-level backup namespaces of a datatstore
- ///
- /// The iterated item is still a Result that can contain errors from rather unexptected FS or
- /// parsing errors.
- pub fn iter_backup_ns(
- self: &Arc<DataStore>,
- ns: BackupNamespace,
- ) -> Result<ListNamespaces, Error> {
- ListNamespaces::new(Arc::clone(self), ns)
- }
-
- /// Get a streaming iter over single-level backup namespaces of a datatstore, filtered by Ok
- ///
- /// The iterated item's result is already unwrapped, if it contained an error it will be
- /// logged. Can be useful in iterator chain commands
- pub fn iter_backup_ns_ok(
- self: &Arc<DataStore>,
- ns: BackupNamespace,
- ) -> Result<impl Iterator<Item = BackupNamespace> + 'static, Error> {
- let this = Arc::clone(self);
- Ok(
- ListNamespaces::new(Arc::clone(self), ns)?.filter_map(move |ns| match ns {
- Ok(ns) => Some(ns),
- Err(err) => {
- log::error!("list groups error on datastore {} - {}", this.name(), err);
- None
- }
- }),
- )
- }
-
- /// Get a streaming iter over single-level backup namespaces of a datatstore
- ///
- /// The iterated item is still a Result that can contain errors from rather unexptected FS or
- /// parsing errors.
- pub fn recursive_iter_backup_ns(
- self: &Arc<DataStore>,
- ns: BackupNamespace,
- ) -> Result<ListNamespacesRecursive, Error> {
- ListNamespacesRecursive::new(Arc::clone(self), ns)
- }
-
- /// Get a streaming iter over single-level backup namespaces of a datatstore, filtered by Ok
- ///
- /// The iterated item's result is already unwrapped, if it contained an error it will be
- /// logged. Can be useful in iterator chain commands
- pub fn recursive_iter_backup_ns_ok(
- self: &Arc<DataStore>,
- ns: BackupNamespace,
- max_depth: Option<usize>,
- ) -> Result<impl Iterator<Item = BackupNamespace> + 'static, Error> {
- let this = Arc::clone(self);
- Ok(if let Some(depth) = max_depth {
- ListNamespacesRecursive::new_max_depth(Arc::clone(self), ns, depth)?
- } else {
- ListNamespacesRecursive::new(Arc::clone(self), ns)?
- }
- .filter_map(move |ns| match ns {
- Ok(ns) => Some(ns),
- Err(err) => {
- log::error!("list groups error on datastore {} - {}", this.name(), err);
- None
- }
- }))
- }
-
- /// Get a streaming iter over top-level backup groups of a datatstore of a particular type.
- ///
- /// The iterated item is still a Result that can contain errors from rather unexptected FS or
- /// parsing errors.
- pub fn iter_backup_type(
- self: &Arc<DataStore>,
- ns: BackupNamespace,
- ty: BackupType,
- ) -> Result<ListGroupsType, Error> {
- ListGroupsType::new(Arc::clone(self), ns, ty)
- }
-
- /// Get a streaming iter over top-level backup groups of a datastore of a particular type,
- /// filtered by `Ok` results
- ///
- /// The iterated item's result is already unwrapped, if it contained an error it will be
- /// logged. Can be useful in iterator chain commands
- pub fn iter_backup_type_ok(
- self: &Arc<DataStore>,
- ns: BackupNamespace,
- ty: BackupType,
- ) -> Result<impl Iterator<Item = BackupGroup> + 'static, Error> {
- Ok(self.iter_backup_type(ns, ty)?.ok())
- }
-
- /// Get a streaming iter over top-level backup groups of a datatstore
- ///
- /// The iterated item is still a Result that can contain errors from rather unexptected FS or
- /// parsing errors.
- pub fn iter_backup_groups(
- self: &Arc<DataStore>,
- ns: BackupNamespace,
- ) -> Result<ListGroups, Error> {
- ListGroups::new(Arc::clone(self), ns)
+ Ok((owner, guard))
+ }
+ Err(err) => bail!("unable to create backup group {:?} - {}", full_path, err),
+ }
}
- /// Get a streaming iter over top-level backup groups of a datatstore, filtered by Ok results
+ /// Creates a new backup snapshot inside a BackupGroup
///
- /// The iterated item's result is already unwrapped, if it contained an error it will be
- /// logged. Can be useful in iterator chain commands
- pub fn iter_backup_groups_ok(
- self: &Arc<DataStore>,
- ns: BackupNamespace,
- ) -> Result<impl Iterator<Item = BackupGroup> + 'static, Error> {
- Ok(self.iter_backup_groups(ns)?.ok())
- }
+ /// The BackupGroup directory needs to exist.
+ pub fn create_locked_backup_dir(
+ &self,
+ ns: &BackupNamespace,
+ backup_dir: &pbs_api_types::BackupDir,
+ ) -> Result<(PathBuf, bool, DirLockGuard), Error> {
+ let full_path = self.snapshot_path(ns, backup_dir);
+ let relative_path = full_path.strip_prefix(self.base_path()).map_err(|err| {
+ format_err!(
+ "failed to produce correct path for backup {backup_dir} in namespace {ns}: {err}"
+ )
+ })?;
- /// Get a in-memory vector for all top-level backup groups of a datatstore
- ///
- /// NOTE: using the iterator directly is most often more efficient w.r.t. memory usage
- pub fn list_backup_groups(
- self: &Arc<DataStore>,
- ns: BackupNamespace,
- ) -> Result<Vec<BackupGroup>, Error> {
- ListGroups::new(Arc::clone(self), ns)?.collect()
- }
+ let lock = || {
+ lock_dir_noblock(
+ &full_path,
+ "snapshot",
+ "internal error - tried creating snapshot that's already in use",
+ )
+ };
- pub fn list_images(&self) -> Result<Vec<PathBuf>, Error> {
- let base = self.base_path();
+ match std::fs::create_dir(&full_path) {
+ Ok(_) => Ok((relative_path.to_owned(), true, lock()?)),
+ Err(ref e) if e.kind() == io::ErrorKind::AlreadyExists => {
+ Ok((relative_path.to_owned(), false, lock()?))
+ }
+ Err(e) => Err(e.into()),
+ }
+ }
+ /// Set the backup owner.
+ pub fn set_owner(
+ &self,
+ ns: &BackupNamespace,
+ backup_group: &pbs_api_types::BackupGroup,
+ auth_id: &Authid,
+ force: bool,
+ ) -> Result<(), Error> {
+ let path = self.owner_path(ns, backup_group);
- let mut list = vec![];
+ let mut open_options = std::fs::OpenOptions::new();
+ open_options.write(true);
+ open_options.truncate(true);
- use walkdir::WalkDir;
+ if force {
+ open_options.create(true);
+ } else {
+ open_options.create_new(true);
+ }
- let walker = WalkDir::new(base).into_iter();
+ let mut file = open_options
+ .open(&path)
+ .map_err(|err| format_err!("unable to create owner file {:?} - {}", path, err))?;
- // make sure we skip .chunks (and other hidden files to keep it simple)
- fn is_hidden(entry: &walkdir::DirEntry) -> bool {
- entry
- .file_name()
- .to_str()
- .map(|s| s.starts_with('.'))
- .unwrap_or(false)
- }
- let handle_entry_err = |err: walkdir::Error| {
- // first, extract the actual IO error and the affected path
- let (inner, path) = match (err.io_error(), err.path()) {
- (None, _) => return Ok(()), // not an IO-error
- (Some(inner), Some(path)) => (inner, path),
- (Some(inner), None) => bail!("unexpected error on datastore traversal: {inner}"),
- };
- if inner.kind() == io::ErrorKind::PermissionDenied {
- if err.depth() <= 1 && path.ends_with("lost+found") {
- // allow skipping of (root-only) ext4 fsck-directory on EPERM ..
- return Ok(());
- }
- // .. but do not ignore EPERM in general, otherwise we might prune too many chunks.
- // E.g., if users messed up with owner/perms on a rsync
- bail!("cannot continue garbage-collection safely, permission denied on: {path:?}");
- } else if inner.kind() == io::ErrorKind::NotFound {
- log::info!("ignoring vanished file: {path:?}");
- return Ok(());
- } else {
- bail!("unexpected error on datastore traversal: {inner} - {path:?}");
- }
- };
- for entry in walker.filter_entry(|e| !is_hidden(e)) {
- let path = match entry {
- Ok(entry) => entry.into_path(),
- Err(err) => {
- handle_entry_err(err)?;
- continue;
- }
- };
- if let Ok(archive_type) = ArchiveType::from_path(&path) {
- if archive_type == ArchiveType::FixedIndex
- || archive_type == ArchiveType::DynamicIndex
- {
- list.push(path);
- }
- }
- }
+ writeln!(file, "{}", auth_id)
+ .map_err(|err| format_err!("unable to write owner file {:?} - {}", path, err))?;
- Ok(list)
+ Ok(())
}
// mark chunks used by ``index`` as used
@@ -1104,14 +1271,6 @@ impl DataStore {
Ok(())
}
- pub fn last_gc_status(&self) -> GarbageCollectionStatus {
- self.inner.last_gc_status.lock().unwrap().clone()
- }
-
- pub fn garbage_collection_running(&self) -> bool {
- self.inner.gc_mutex.try_lock().is_err()
- }
-
pub fn garbage_collection(
&self,
worker: &dyn WorkerTaskContext,
@@ -1194,218 +1353,69 @@ impl DataStore {
if gc_status.disk_chunks > 0 {
let avg_chunk = gc_status.disk_bytes / (gc_status.disk_chunks as u64);
info!("Average chunk size: {}", HumanByte::from(avg_chunk));
- }
-
- if let Ok(serialized) = serde_json::to_string(&gc_status) {
- let mut path = self.base_path();
- path.push(".gc-status");
-
- let backup_user = pbs_config::backup_user()?;
- let mode = nix::sys::stat::Mode::from_bits_truncate(0o0644);
- // set the correct owner/group/permissions while saving file
- // owner(rw) = backup, group(r)= backup
- let options = CreateOptions::new()
- .perm(mode)
- .owner(backup_user.uid)
- .group(backup_user.gid);
-
- // ignore errors
- let _ = replace_file(path, serialized.as_bytes(), options, false);
- }
-
- *self.inner.last_gc_status.lock().unwrap() = gc_status;
- } else {
- bail!("Start GC failed - (already running/locked)");
- }
-
- Ok(())
- }
-
- pub fn try_shared_chunk_store_lock(&self) -> Result<ProcessLockSharedGuard, Error> {
- self.inner.chunk_store.try_shared_lock()
- }
-
- pub fn chunk_path(&self, digest: &[u8; 32]) -> (PathBuf, String) {
- self.inner.chunk_store.chunk_path(digest)
- }
-
- pub fn cond_touch_chunk(&self, digest: &[u8; 32], assert_exists: bool) -> Result<bool, Error> {
- self.inner
- .chunk_store
- .cond_touch_chunk(digest, assert_exists)
- }
-
- pub fn insert_chunk(&self, chunk: &DataBlob, digest: &[u8; 32]) -> Result<(bool, u64), Error> {
- self.inner.chunk_store.insert_chunk(chunk, digest)
- }
-
- pub fn stat_chunk(&self, digest: &[u8; 32]) -> Result<std::fs::Metadata, Error> {
- let (chunk_path, _digest_str) = self.inner.chunk_store.chunk_path(digest);
- std::fs::metadata(chunk_path).map_err(Error::from)
- }
-
- pub fn load_chunk(&self, digest: &[u8; 32]) -> Result<DataBlob, Error> {
- let (chunk_path, digest_str) = self.inner.chunk_store.chunk_path(digest);
-
- proxmox_lang::try_block!({
- let mut file = std::fs::File::open(&chunk_path)?;
- DataBlob::load_from_reader(&mut file)
- })
- .map_err(|err| {
- format_err!(
- "store '{}', unable to load chunk '{}' - {}",
- self.name(),
- digest_str,
- err,
- )
- })
- }
-
- /// Updates the protection status of the specified snapshot.
- pub fn update_protection(&self, backup_dir: &BackupDir, protection: bool) -> Result<(), Error> {
- let full_path = backup_dir.full_path();
-
- if !full_path.exists() {
- bail!("snapshot {} does not exist!", backup_dir.dir());
- }
-
- let _guard = lock_dir_noblock(&full_path, "snapshot", "possibly running or in use")?;
-
- let protected_path = backup_dir.protected_file();
- if protection {
- std::fs::File::create(protected_path)
- .map_err(|err| format_err!("could not create protection file: {}", err))?;
- } else if let Err(err) = std::fs::remove_file(protected_path) {
- // ignore error for non-existing file
- if err.kind() != std::io::ErrorKind::NotFound {
- bail!("could not remove protection file: {}", err);
- }
- }
-
- Ok(())
- }
-
- pub fn verify_new(&self) -> bool {
- self.inner.verify_new
- }
-
- /// returns a list of chunks sorted by their inode number on disk chunks that couldn't get
- /// stat'ed are placed at the end of the list
- pub fn get_chunks_in_order<F, A>(
- &self,
- index: &(dyn IndexFile + Send),
- skip_chunk: F,
- check_abort: A,
- ) -> Result<Vec<(usize, u64)>, Error>
- where
- F: Fn(&[u8; 32]) -> bool,
- A: Fn(usize) -> Result<(), Error>,
- {
- let index_count = index.index_count();
- let mut chunk_list = Vec::with_capacity(index_count);
- use std::os::unix::fs::MetadataExt;
- for pos in 0..index_count {
- check_abort(pos)?;
-
- let info = index.chunk_info(pos).unwrap();
-
- if skip_chunk(&info.digest) {
- continue;
- }
-
- let ino = match self.inner.chunk_order {
- ChunkOrder::Inode => {
- match self.stat_chunk(&info.digest) {
- Err(_) => u64::MAX, // could not stat, move to end of list
- Ok(metadata) => metadata.ino(),
- }
- }
- ChunkOrder::None => 0,
- };
+ }
- chunk_list.push((pos, ino));
- }
+ if let Ok(serialized) = serde_json::to_string(&gc_status) {
+ let mut path = self.base_path();
+ path.push(".gc-status");
- match self.inner.chunk_order {
- // sorting by inode improves data locality, which makes it lots faster on spinners
- ChunkOrder::Inode => {
- chunk_list.sort_unstable_by(|(_, ino_a), (_, ino_b)| ino_a.cmp(ino_b))
+ let backup_user = pbs_config::backup_user()?;
+ let mode = nix::sys::stat::Mode::from_bits_truncate(0o0644);
+ // set the correct owner/group/permissions while saving file
+ // owner(rw) = backup, group(r)= backup
+ let options = CreateOptions::new()
+ .perm(mode)
+ .owner(backup_user.uid)
+ .group(backup_user.gid);
+
+ // ignore errors
+ let _ = replace_file(path, serialized.as_bytes(), options, false);
}
- ChunkOrder::None => {}
+
+ *self.inner.last_gc_status.lock().unwrap() = gc_status;
+ } else {
+ bail!("Start GC failed - (already running/locked)");
}
- Ok(chunk_list)
+ Ok(())
}
-
- /// Open a backup group from this datastore.
- pub fn backup_group(
- self: &Arc<Self>,
- ns: BackupNamespace,
- group: pbs_api_types::BackupGroup,
- ) -> BackupGroup {
- BackupGroup::new(Arc::clone(self), ns, group)
+ pub fn cond_touch_chunk(&self, digest: &[u8; 32], assert_exists: bool) -> Result<bool, Error> {
+ self.inner
+ .chunk_store
+ .cond_touch_chunk(digest, assert_exists)
}
- /// Open a backup group from this datastore.
- pub fn backup_group_from_parts<T>(
- self: &Arc<Self>,
- ns: BackupNamespace,
- ty: BackupType,
- id: T,
- ) -> BackupGroup
- where
- T: Into<String>,
- {
- self.backup_group(ns, (ty, id.into()).into())
+ pub fn insert_chunk(&self, chunk: &DataBlob, digest: &[u8; 32]) -> Result<(bool, u64), Error> {
+ self.inner.chunk_store.insert_chunk(chunk, digest)
}
- /*
- /// Open a backup group from this datastore by backup group path such as `vm/100`.
- ///
- /// Convenience method for `store.backup_group(path.parse()?)`
- pub fn backup_group_from_path(self: &Arc<Self>, path: &str) -> Result<BackupGroup, Error> {
- todo!("split out the namespace");
- }
- */
+ /// Updates the protection status of the specified snapshot.
+ pub fn update_protection(
+ &self,
+ backup_dir: &BackupDir<T>,
+ protection: bool,
+ ) -> Result<(), Error> {
+ let full_path = backup_dir.full_path();
- /// Open a snapshot (backup directory) from this datastore.
- pub fn backup_dir(
- self: &Arc<Self>,
- ns: BackupNamespace,
- dir: pbs_api_types::BackupDir,
- ) -> Result<BackupDir, Error> {
- BackupDir::with_group(self.backup_group(ns, dir.group), dir.time)
- }
+ if !full_path.exists() {
+ bail!("snapshot {} does not exist!", backup_dir.dir());
+ }
- /// Open a snapshot (backup directory) from this datastore.
- pub fn backup_dir_from_parts<T>(
- self: &Arc<Self>,
- ns: BackupNamespace,
- ty: BackupType,
- id: T,
- time: i64,
- ) -> Result<BackupDir, Error>
- where
- T: Into<String>,
- {
- self.backup_dir(ns, (ty, id.into(), time).into())
- }
+ let _guard = lock_dir_noblock(&full_path, "snapshot", "possibly running or in use")?;
- /// Open a snapshot (backup directory) from this datastore with a cached rfc3339 time string.
- pub fn backup_dir_with_rfc3339<T: Into<String>>(
- self: &Arc<Self>,
- group: BackupGroup,
- time_string: T,
- ) -> Result<BackupDir, Error> {
- BackupDir::with_rfc3339(group, time_string.into())
- }
+ let protected_path = backup_dir.protected_file();
+ if protection {
+ std::fs::File::create(protected_path)
+ .map_err(|err| format_err!("could not create protection file: {}", err))?;
+ } else if let Err(err) = std::fs::remove_file(protected_path) {
+ // ignore error for non-existing file
+ if err.kind() != std::io::ErrorKind::NotFound {
+ bail!("could not remove protection file: {}", err);
+ }
+ }
- /*
- /// Open a snapshot (backup directory) from this datastore by a snapshot path.
- pub fn backup_dir_from_path(self: &Arc<Self>, path: &str) -> Result<BackupDir, Error> {
- todo!("split out the namespace");
+ Ok(())
}
- */
/// Syncs the filesystem of the datastore if 'sync_level' is set to
/// [`DatastoreFSyncLevel::Filesystem`]. Uses syncfs(2).
@@ -1421,107 +1431,90 @@ impl DataStore {
}
Ok(())
}
+}
+impl<T> DataStore<T> {
+ #[doc(hidden)]
+ pub(crate) fn new_test() -> Arc<Self> {
+ Arc::new(Self {
+ inner: DataStoreImpl::new_test(),
+ operation: None,
+ })
+ }
- /// Destroy a datastore. This requires that there are no active operations on the datastore.
- ///
- /// This is a synchronous operation and should be run in a worker-thread.
- pub fn destroy(name: &str, destroy_data: bool) -> Result<(), Error> {
- let config_lock = pbs_config::datastore::lock_config()?;
-
- let (mut config, _digest) = pbs_config::datastore::config()?;
- let mut datastore_config: DataStoreConfig = config.lookup("datastore", name)?;
+ pub fn read_config(name: &str) -> Result<(DataStoreConfig, [u8; 32], BackupLockGuard), Error> {
+ // Avoid TOCTOU between checking maintenance mode and updating active operation counter, as
+ // we use it to decide whether it is okay to delete the datastore.
+ let lock = pbs_config::datastore::lock_config()?;
- datastore_config.set_maintenance_mode(Some(MaintenanceMode {
- ty: MaintenanceType::Delete,
- message: None,
- }))?;
+ // we could use the ConfigVersionCache's generation for staleness detection, but we load
+ // the config anyway -> just use digest, additional benefit: manual changes get detected
+ let (config, digest) = pbs_config::datastore::config()?;
+ let config: DataStoreConfig = config.lookup("datastore", name)?;
+ Ok((config, digest, lock))
+ }
- config.set_data(name, "datastore", &datastore_config)?;
- pbs_config::datastore::save_config(&config)?;
- drop(config_lock);
+ pub fn name(&self) -> &str {
+ self.inner.chunk_store.name()
+ }
- let (operations, _lock) = task_tracking::get_active_operations_locked(name)?;
+ pub fn base_path(&self) -> PathBuf {
+ self.inner.chunk_store.base_path()
+ }
- if operations.read != 0 || operations.write != 0 {
- bail!("datastore is currently in use");
+ /// Returns the absolute path for a backup namespace on this datastore
+ pub fn namespace_path(&self, ns: &BackupNamespace) -> PathBuf {
+ let mut path = self.base_path();
+ path.reserve(ns.path_len());
+ for part in ns.components() {
+ path.push("ns");
+ path.push(part);
}
+ path
+ }
- let base = PathBuf::from(&datastore_config.path);
-
- let mut ok = true;
- if destroy_data {
- let remove = |subdir, ok: &mut bool| {
- if let Err(err) = std::fs::remove_dir_all(base.join(subdir)) {
- if err.kind() != io::ErrorKind::NotFound {
- warn!("failed to remove {subdir:?} subdirectory: {err}");
- *ok = false;
- }
- }
- };
-
- info!("Deleting datastore data...");
- remove("ns", &mut ok); // ns first
- remove("ct", &mut ok);
- remove("vm", &mut ok);
- remove("host", &mut ok);
-
- if ok {
- if let Err(err) = std::fs::remove_file(base.join(".gc-status")) {
- if err.kind() != io::ErrorKind::NotFound {
- warn!("failed to remove .gc-status file: {err}");
- ok = false;
- }
- }
- }
+ /// Returns the absolute path for a backup_type
+ pub fn type_path(&self, ns: &BackupNamespace, backup_type: BackupType) -> PathBuf {
+ let mut full_path = self.namespace_path(ns);
+ full_path.push(backup_type.to_string());
+ full_path
+ }
- // chunks get removed last and only if the backups were successfully deleted
- if ok {
- remove(".chunks", &mut ok);
- }
- }
+ /// Returns the absolute path for a backup_group
+ pub fn group_path(
+ &self,
+ ns: &BackupNamespace,
+ backup_group: &pbs_api_types::BackupGroup,
+ ) -> PathBuf {
+ let mut full_path = self.namespace_path(ns);
+ full_path.push(backup_group.to_string());
+ full_path
+ }
- // now the config
- if ok {
- info!("Removing datastore from config...");
- let _lock = pbs_config::datastore::lock_config()?;
- let _ = config.sections.remove(name);
- pbs_config::datastore::save_config(&config)?;
- }
+ /// Returns the absolute path for backup_dir
+ pub fn snapshot_path(
+ &self,
+ ns: &BackupNamespace,
+ backup_dir: &pbs_api_types::BackupDir,
+ ) -> PathBuf {
+ let mut full_path = self.namespace_path(ns);
+ full_path.push(backup_dir.to_string());
+ full_path
+ }
- // finally the lock & toplevel directory
- if destroy_data {
- if ok {
- if let Err(err) = std::fs::remove_file(base.join(".lock")) {
- if err.kind() != io::ErrorKind::NotFound {
- warn!("failed to remove .lock file: {err}");
- ok = false;
- }
- }
- }
+ /// Return the path of the 'owner' file.
+ fn owner_path(&self, ns: &BackupNamespace, group: &pbs_api_types::BackupGroup) -> PathBuf {
+ self.group_path(ns, group).join("owner")
+ }
- if ok {
- info!("Finished deleting data.");
+ pub fn chunk_path(&self, digest: &[u8; 32]) -> (PathBuf, String) {
+ self.inner.chunk_store.chunk_path(digest)
+ }
- match std::fs::remove_dir(base) {
- Ok(()) => info!("Removed empty datastore directory."),
- Err(err) if err.kind() == io::ErrorKind::NotFound => {
- // weird, but ok
- }
- Err(err) if err.is_errno(nix::errno::Errno::EBUSY) => {
- warn!("Cannot delete datastore directory (is it a mount point?).")
- }
- Err(err) if err.is_errno(nix::errno::Errno::ENOTEMPTY) => {
- warn!("Datastore directory not empty, not deleting.")
- }
- Err(err) => {
- warn!("Failed to remove datastore directory: {err}");
- }
- }
- } else {
- info!("There were errors deleting data.");
- }
- }
+ pub fn verify_new(&self) -> bool {
+ self.inner.verify_new
+ }
- Ok(())
+ pub fn garbage_collection_running(&self) -> bool {
+ self.inner.gc_mutex.try_lock().is_err()
}
}
--
2.39.2
More information about the pbs-devel
mailing list