tyler/proxmox-backup
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

sync/pull: make namespace aware

Browse Source 
Allow pulling all groups from a certain source namespace, and
possibly sub namespaces until max-depth, into a target namespace.

If any sub-namespaces get pulled, they will be mapped relatively from
the source parent namespace to the target parent namespace.

Signed-off-by: Fabian Grünbichler <f.gruenbichler@proxmox.com>
Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
This commit is contained in:
Fabian Grünbichler
2022-04-29 13:46:14 +02:00
committed by Thomas Lamprecht
parent 31aa38b684
commit c06c1b4bd7
8 changed files with 582 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

449
src/server/pull.rs
View File

@ -1,5 +1,6 @@
//! Sync datastore from remote server
use std::cmp::min;
use std::collections::{HashMap, HashSet};
use std::convert::TryFrom;
use std::io::{Seek, SeekFrom};
@ -9,14 +10,16 @@ use std::time::SystemTime;
use anyhow::{bail, format_err, Error};
use http::StatusCode;
use pbs_config::CachedUserInfo;
use serde_json::json;
use proxmox_router::HttpError;
use proxmox_sys::task_log;
use pbs_api_types::{
Authid, BackupNamespace, GroupFilter, GroupListItem, Operation, RateLimitConfig, Remote,
SnapshotListItem,
Authid, BackupNamespace, DatastoreWithNamespace, GroupFilter, GroupListItem, NamespaceListItem,
Operation, RateLimitConfig, Remote, SnapshotListItem, MAX_NAMESPACE_DEPTH,
PRIV_DATASTORE_AUDIT, PRIV_DATASTORE_BACKUP, PRIV_DATASTORE_MODIFY,
};
use pbs_client::{
@ -43,10 +46,16 @@ pub struct PullParameters {
source: BackupRepository,
/// Local store that is pulled into
store: Arc<DataStore>,
/// Remote namespace
remote_ns: BackupNamespace,
/// Local namespace (anchor)
ns: BackupNamespace,
/// Owner of synced groups (needs to match local owner of pre-existing groups)
owner: Authid,
/// Whether to remove groups which exist locally, but not on the remote end
remove_vanished: bool,
/// How many levels of sub-namespaces to pull (0 == no recursion)
max_depth: usize,
/// Filters for reducing the pull scope
group_filter: Option<Vec<GroupFilter>>,
/// Rate limits for all transfers from `remote`
@ -60,15 +69,20 @@ impl PullParameters {
/// [BackupRepository] with `remote_store`.
pub fn new(
store: &str,
ns: BackupNamespace,
remote: &str,
remote_store: &str,
remote_ns: BackupNamespace,
owner: Authid,
remove_vanished: Option<bool>,
max_depth: usize,
group_filter: Option<Vec<GroupFilter>>,
limit: RateLimitConfig,
) -> Result<Self, Error> {
let store = DataStore::lookup_datastore(store, Some(Operation::Write))?;
let max_depth = min(max_depth, MAX_NAMESPACE_DEPTH - remote_ns.depth());
let (remote_config, _digest) = pbs_config::remote::config()?;
let remote: Remote = remote_config.lookup("remote", remote)?;
@ -83,10 +97,13 @@ impl PullParameters {
Ok(Self {
remote,
remote_ns,
ns,
source,
store,
owner,
remove_vanished,
max_depth,
group_filter,
limit,
})
@ -96,6 +113,14 @@ impl PullParameters {
pub async fn client(&self) -> Result<HttpClient, Error> {
crate::api2::config::remote::remote_client(&self.remote, Some(self.limit.clone())).await
}
/// Returns DatastoreWithNamespace with namespace (or local namespace anchor).
pub fn store_with_ns(&self, ns: BackupNamespace) -> DatastoreWithNamespace {
DatastoreWithNamespace {
store: self.store.name().to_string(),
ns,
}
}
}
async fn pull_index_chunks<I: IndexFile>(
@ -240,14 +265,12 @@ async fn pull_single_archive(
worker: &WorkerTask,
reader: &BackupReader,
chunk_reader: &mut RemoteChunkReader,
tgt_store: Arc<DataStore>,
snapshot: &pbs_api_types::BackupDir,
snapshot: &pbs_datastore::BackupDir,
archive_info: &FileInfo,
downloaded_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
) -> Result<(), Error> {
let archive_name = &archive_info.filename;
let mut path = tgt_store.base_path();
path.push(snapshot.to_string());
let mut path = snapshot.full_path();
path.push(archive_name);
let mut tmp_path = path.clone();
@ -274,7 +297,7 @@ async fn pull_single_archive(
pull_index_chunks(
worker,
chunk_reader.clone(),
tgt_store.clone(),
snapshot.datastore().clone(),
index,
downloaded_chunks,
)
@ -290,7 +313,7 @@ async fn pull_single_archive(
pull_index_chunks(
worker,
chunk_reader.clone(),
tgt_store.clone(),
snapshot.datastore().clone(),
index,
downloaded_chunks,
)
@ -347,18 +370,13 @@ async fn try_client_log_download(
async fn pull_snapshot(
worker: &WorkerTask,
reader: Arc<BackupReader>,
tgt_store: Arc<DataStore>,
snapshot: &pbs_api_types::BackupDir,
snapshot: &pbs_datastore::BackupDir,
downloaded_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
) -> Result<(), Error> {
let snapshot_relative_path = snapshot.to_string();
let mut manifest_name = tgt_store.base_path();
manifest_name.push(&snapshot_relative_path);
let mut manifest_name = snapshot.full_path();
manifest_name.push(MANIFEST_BLOB_NAME);
let mut client_log_name = tgt_store.base_path();
client_log_name.push(&snapshot_relative_path);
let mut client_log_name = snapshot.full_path();
client_log_name.push(CLIENT_LOG_BLOB_NAME);
let mut tmp_manifest_name = manifest_name.clone();
@ -424,8 +442,7 @@ async fn pull_snapshot(
let manifest = BackupManifest::try_from(tmp_manifest_blob)?;
for item in manifest.files() {
let mut path = tgt_store.base_path();
path.push(&snapshot_relative_path);
let mut path = snapshot.full_path();
path.push(&item.filename);
if path.exists() {
@ -474,7 +491,6 @@ async fn pull_snapshot(
worker,
&reader,
&mut chunk_reader,
tgt_store.clone(),
snapshot,
item,
downloaded_chunks.clone(),
@ -491,38 +507,37 @@ async fn pull_snapshot(
}
// cleanup - remove stale files
tgt_store.cleanup_backup_dir(snapshot, &manifest)?;
snapshot
.datastore()
.cleanup_backup_dir(snapshot, &manifest)?;
Ok(())
}
/// Pulls a `snapshot` into `tgt_store`, differentiating between new snapshots (removed on error)
/// and existing ones (kept even on error).
/// Pulls a `snapshot`, removing newly created ones on error, but keeping existing ones in any case.
///
/// The `reader` is configured to read from the remote / source namespace, while the `snapshot` is
/// pointing to the local datastore and target namespace.
async fn pull_snapshot_from(
worker: &WorkerTask,
reader: Arc<BackupReader>,
tgt_store: Arc<DataStore>,
snapshot: &pbs_api_types::BackupDir,
snapshot: &pbs_datastore::BackupDir,
downloaded_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
) -> Result<(), Error> {
// FIXME: Namespace support requires source AND target namespace
let ns = BackupNamespace::root();
let (_path, is_new, _snap_lock) = tgt_store.create_locked_backup_dir(&ns, snapshot)?;
let (_path, is_new, _snap_lock) = snapshot
.datastore()
.create_locked_backup_dir(snapshot.backup_ns(), snapshot.as_ref())?;
let snapshot_path = snapshot.to_string();
if is_new {
task_log!(worker, "sync snapshot {:?}", snapshot_path);
if let Err(err) = pull_snapshot(
worker,
reader,
tgt_store.clone(),
snapshot,
downloaded_chunks,
)
.await
{
if let Err(cleanup_err) = tgt_store.remove_backup_dir(&ns, snapshot, true) {
if let Err(err) = pull_snapshot(worker, reader, snapshot, downloaded_chunks).await {
if let Err(cleanup_err) = snapshot.datastore().remove_backup_dir(
snapshot.backup_ns(),
snapshot.as_ref(),
true,
) {
task_log!(worker, "cleanup error - {}", cleanup_err);
}
return Err(err);
@ -530,14 +545,7 @@ async fn pull_snapshot_from(
task_log!(worker, "sync snapshot {:?} done", snapshot_path);
} else {
task_log!(worker, "re-sync snapshot {:?}", snapshot_path);
pull_snapshot(
worker,
reader,
tgt_store.clone(),
snapshot,
downloaded_chunks,
)
.await?;
pull_snapshot(worker, reader, snapshot, downloaded_chunks).await?;
task_log!(worker, "re-sync snapshot {:?} done", snapshot_path);
}
@ -590,13 +598,18 @@ impl std::fmt::Display for SkipInfo {
/// Pulls a group according to `params`.
///
/// Pulling a group consists of the following steps:
/// - Query the list of snapshots available for this group on the remote, sort by snapshot time
/// - Query the list of snapshots available for this group in the source namespace on the remote
/// - Sort by snapshot time
/// - Get last snapshot timestamp on local datastore
/// - Iterate over list of snapshots
/// -- Recreate client/BackupReader
/// -- pull snapshot, unless it's not finished yet or older than last local snapshot
/// - (remove_vanished) list all local snapshots, remove those that don't exist on remote
///
/// Backwards-compat: if `source_ns` is [None], only the group type and ID will be sent to the
/// remote when querying snapshots. This allows us to interact with old remotes that don't have
/// namespace support yet.
///
/// Permission checks:
/// - remote snapshot access is checked by remote (twice: query and opening the backup reader)
/// - local group owner is already checked by pull_store
@ -605,21 +618,25 @@ async fn pull_group(
client: &HttpClient,
params: &PullParameters,
group: &pbs_api_types::BackupGroup,
remote_ns: BackupNamespace,
progress: &mut StoreProgress,
) -> Result<(), Error> {
// FIXME: Namespace support
let ns = BackupNamespace::root();
let path = format!(
"api2/json/admin/datastore/{}/snapshots",
params.source.store()
);
let args = json!({
let mut args = json!({
"backup-type": group.ty,
"backup-id": group.id,
});
if !remote_ns.is_root() {
args["backup-ns"] = serde_json::to_value(&remote_ns)?;
}
let target_ns = remote_ns.map_prefix(&params.remote_ns, &params.ns)?;
let mut result = client.get(&path, Some(args)).await?;
let mut list: Vec<SnapshotListItem> = serde_json::from_value(result["data"].take())?;
@ -629,7 +646,7 @@ async fn pull_group(
let fingerprint = client.fingerprint();
let last_sync = params.store.last_successful_backup(&ns, group)?;
let last_sync = params.store.last_successful_backup(&target_ns, group)?;
let mut remote_snapshots = std::collections::HashSet::new();
@ -684,20 +701,15 @@ async fn pull_group(
new_client,
None,
params.source.store(),
&ns,
&remote_ns,
&snapshot,
true,
)
.await?;
let result = pull_snapshot_from(
worker,
reader,
params.store.clone(),
&snapshot,
downloaded_chunks.clone(),
)
.await;
let snapshot = params.store.backup_dir(target_ns.clone(), snapshot)?;
let result = pull_snapshot_from(worker, reader, &snapshot, downloaded_chunks.clone()).await;
progress.done_snapshots = pos as u64 + 1;
task_log!(worker, "percentage done: {}", progress);
@ -706,7 +718,7 @@ async fn pull_group(
}
if params.remove_vanished {
let group = params.store.backup_group(ns.clone(), group.clone());
let group = params.store.backup_group(target_ns.clone(), group.clone());
let local_list = group.list_backups()?;
for info in local_list {
let backup_time = info.backup_dir.backup_time();
@ -728,7 +740,7 @@ async fn pull_group(
);
params
.store
.remove_backup_dir(&ns, info.backup_dir.as_ref(), false)?;
.remove_backup_dir(&target_ns, info.backup_dir.as_ref(), false)?;
}
}
@ -739,37 +751,290 @@ async fn pull_group(
Ok(())
}
async fn query_namespaces(
client: &HttpClient,
params: &PullParameters,
) -> Result<Vec<BackupNamespace>, Error> {
let path = format!(
"api2/json/admin/datastore/{}/namespace",
params.source.store()
);
let data = json!({
"max-depth": params.max_depth,
});
let mut result = client
.get(&path, Some(data))
.await
.map_err(|err| format_err!("Failed to retrieve namespaces from remote - {}", err))?;
let mut list: Vec<NamespaceListItem> = serde_json::from_value(result["data"].take())?;
// parents first
list.sort_unstable_by(|a, b| a.ns.name_len().cmp(&b.ns.name_len()));
Ok(list.iter().map(|item| item.ns.clone()).collect())
}
fn check_ns_privs(
store_with_ns: &DatastoreWithNamespace,
owner: &Authid,
privs: u64,
) -> Result<(), Error> {
let user_info = CachedUserInfo::new()?;
// TODO re-sync with API, maybe find common place?
let user_privs = user_info.lookup_privs(owner, &store_with_ns.acl_path());
if (user_privs & privs) == 0 {
bail!("no permission to modify parent/datastore.");
}
Ok(())
}
fn check_and_create_ns(
params: &PullParameters,
store_with_ns: &DatastoreWithNamespace,
) -> Result<bool, Error> {
let ns = &store_with_ns.ns;
let mut created = false;
if !ns.is_root() && !params.store.namespace_path(&ns).exists() {
let mut parent = ns.clone();
let name = parent.pop();
let parent = params.store_with_ns(parent);
if let Err(err) = check_ns_privs(&parent, &params.owner, PRIV_DATASTORE_MODIFY) {
bail!(
"Not allowed to create namespace {} - {}",
store_with_ns,
err,
);
}
if let Some(name) = name {
if let Err(err) = params.store.create_namespace(&parent.ns, name) {
bail!(
"sync namespace {} failed - namespace creation failed: {}",
&store_with_ns,
err
);
}
created = true;
} else {
bail!(
"sync namespace {} failed - namespace creation failed - couldn't determine parent namespace",
&store_with_ns,
);
}
}
// TODO re-sync with API, maybe find common place?
if let Err(err) = check_ns_privs(&store_with_ns, &params.owner, PRIV_DATASTORE_BACKUP) {
bail!("sync namespace {} failed - {}", &store_with_ns, err);
}
Ok(created)
}
fn check_and_remove_ns(params: &PullParameters, local_ns: &BackupNamespace) -> Result<bool, Error> {
let parent = local_ns.clone().parent();
check_ns_privs(
&params.store_with_ns(parent),
&params.owner,
PRIV_DATASTORE_MODIFY,
)?;
params.store.remove_namespace_recursive(local_ns)
}
fn check_and_remove_vanished_ns(
worker: &WorkerTask,
params: &PullParameters,
synced_ns: HashSet<BackupNamespace>,
) -> Result<bool, Error> {
let mut errors = false;
let user_info = CachedUserInfo::new()?;
let mut local_ns_list: Vec<BackupNamespace> = params
.store
.recursive_iter_backup_ns_ok(params.ns.clone(), Some(params.max_depth))?
.filter(|ns| {
let store_with_ns = params.store_with_ns(ns.clone());
let user_privs = user_info.lookup_privs(&params.owner, &store_with_ns.acl_path());
user_privs & (PRIV_DATASTORE_BACKUP | PRIV_DATASTORE_AUDIT) != 0
})
.collect();
// children first!
local_ns_list.sort_unstable_by_key(|b| std::cmp::Reverse(b.name_len()));
for local_ns in local_ns_list {
if local_ns == params.ns {
continue;
}
if synced_ns.contains(&local_ns) {
continue;
}
if local_ns.is_root() {
continue;
}
match check_and_remove_ns(params, &local_ns) {
Ok(true) => task_log!(worker, "Removed namespace {}", local_ns),
Ok(false) => task_log!(
worker,
"Did not remove namespace {} - protected snapshots remain",
local_ns
),
Err(err) => {
task_log!(worker, "Failed to remove namespace {} - {}", local_ns, err);
errors = true;
}
}
}
Ok(errors)
}
/// Pulls a store according to `params`.
///
/// Pulling a store consists of the following steps:
/// - Query list of groups on the remote
/// - Query list of namespaces on the remote
/// - Iterate list
/// -- create sub-NS if needed (and allowed)
/// -- attempt to pull each NS in turn
/// - (remove_vanished && max_depth > 0) remove sub-NS which are not or no longer available on the remote
///
/// Backwards compat: if the remote namespace is `/` and recursion is disabled, no namespace is
/// passed to the remote at all to allow pulling from remotes which have no notion of namespaces.
///
/// Permission checks:
/// - access to local datastore, namespace anchor and remote entry need to be checked at call site
/// - remote namespaces are filtered by remote
/// - creation and removal of sub-NS checked here
/// - access to sub-NS checked here
pub async fn pull_store(
worker: &WorkerTask,
client: &HttpClient,
params: &PullParameters,
) -> Result<(), Error> {
// explicit create shared lock to prevent GC on newly created chunks
let _shared_store_lock = params.store.try_shared_chunk_store_lock()?;
let namespaces = if params.remote_ns.is_root() && params.max_depth == 0 {
vec![params.remote_ns.clone()] // backwards compat - don't query remote namespaces!
} else {
query_namespaces(client, params).await?
};
let (mut groups, mut snapshots) = (0, 0);
let mut synced_ns = HashSet::with_capacity(namespaces.len());
let mut errors = false;
for namespace in namespaces {
let source_store_ns = DatastoreWithNamespace {
store: params.source.store().to_owned(),
ns: namespace.clone(),
};
let target_ns = namespace.map_prefix(&params.remote_ns, &params.ns)?;
let target_store_ns = params.store_with_ns(target_ns.clone());
task_log!(worker, "----");
task_log!(
worker,
"Syncing {} into {}",
source_store_ns,
target_store_ns
);
synced_ns.insert(target_ns.clone());
match check_and_create_ns(params, &target_store_ns) {
Ok(true) => task_log!(worker, "Created namespace {}", target_ns),
Ok(false) => {}
Err(err) => {
task_log!(
worker,
"Cannot sync {} into {} - {}",
source_store_ns,
target_store_ns,
err,
);
errors = true;
continue;
}
}
match pull_ns(worker, client, params, namespace.clone(), target_ns).await {
Ok((ns_progress, ns_errors)) => {
errors |= ns_errors;
if params.max_depth > 0 {
groups += ns_progress.done_groups;
snapshots += ns_progress.done_snapshots;
task_log!(
worker,
"Finished syncing namespace {}, current progress: {} groups, {} snapshots",
namespace,
groups,
snapshots,
);
}
}
Err(err) => {
errors = true;
task_log!(
worker,
"Encountered errors while syncing namespace {} - {}",
namespace,
err,
);
}
};
}
if params.remove_vanished {
errors |= check_and_remove_vanished_ns(worker, params, synced_ns)?;
}
if errors {
bail!("sync failed with some errors.");
}
Ok(())
}
/// Pulls a namespace according to `params`.
///
/// Pulling a namespace consists of the following steps:
/// - Query list of groups on the remote (in `source_ns`)
/// - Filter list according to configured group filters
/// - Iterate list and attempt to pull each group in turn
/// - (remove_vanished) remove groups with matching owner and matching the configured group filters which are
/// not or no longer available on the remote
///
/// Permission checks:
/// - access to local datastore and remote entry need to be checked at call site
/// - remote groups are filtered by remote
/// - remote namespaces are filtered by remote
/// - owner check for vanished groups done here
pub async fn pull_store(
pub async fn pull_ns(
worker: &WorkerTask,
client: &HttpClient,
params: &PullParameters,
) -> Result<(), Error> {
// FIXME: Namespace support requires source AND target namespace
let ns = BackupNamespace::root();
let local_ns = BackupNamespace::root();
// explicit create shared lock to prevent GC on newly created chunks
let _shared_store_lock = params.store.try_shared_chunk_store_lock()?;
// FIXME: Namespaces! AND: If we make this API call recurse down namespaces we need to do the
// same down in the `remove_vanished` case!
source_ns: BackupNamespace,
target_ns: BackupNamespace,
) -> Result<(StoreProgress, bool), Error> {
let path = format!("api2/json/admin/datastore/{}/groups", params.source.store());
let args = if !source_ns.is_root() {
Some(json!({
"backup-ns": source_ns,
}))
} else {
None
};
let mut result = client
.get(&path, None)
.get(&path, args)
.await
.map_err(|err| format_err!("Failed to retrieve backup groups from remote - {}", err))?;
@ -789,6 +1054,7 @@ pub async fn pull_store(
filters.iter().any(|filter| group.matches(filter))
};
// Get groups with target NS set
let list: Vec<pbs_api_types::BackupGroup> = list.into_iter().map(|item| item.backup).collect();
let list = if let Some(ref group_filter) = &params.group_filter {
@ -826,7 +1092,7 @@ pub async fn pull_store(
let (owner, _lock_guard) =
match params
.store
.create_locked_backup_group(&ns, &group, &params.owner)
.create_locked_backup_group(&target_ns, &group, &params.owner)
{
Ok(result) => result,
Err(err) => {
@ -852,7 +1118,16 @@ pub async fn pull_store(
owner
);
errors = true; // do not stop here, instead continue
} else if let Err(err) = pull_group(worker, client, params, &group, &mut progress).await {
} else if let Err(err) = pull_group(
worker,
client,
params,
&group,
source_ns.clone(),
&mut progress,
)
.await
{
task_log!(worker, "sync group {} failed - {}", &group, err,);
errors = true; // do not stop here, instead continue
}
@ -860,13 +1135,12 @@ pub async fn pull_store(
if params.remove_vanished {
let result: Result<(), Error> = proxmox_lang::try_block!({
// FIXME: See above comment about namespaces & recursion
for local_group in params.store.iter_backup_groups(Default::default())? {
for local_group in params.store.iter_backup_groups(target_ns.clone())? {
let local_group = local_group?;
if new_groups.contains(local_group.as_ref()) {
continue;
}
let owner = params.store.get_owner(&local_ns, &local_group.group())?;
let owner = params.store.get_owner(&target_ns, local_group.group())?;
if check_backup_owner(&owner, &params.owner).is_err() {
continue;
}
@ -881,7 +1155,10 @@ pub async fn pull_store(
local_group.backup_type(),
local_group.backup_id()
);
match params.store.remove_backup_group(&ns, local_group.as_ref()) {
match params
.store
.remove_backup_group(&target_ns, local_group.as_ref())
{
Ok(true) => {}
Ok(false) => {
task_log!(
@ -904,9 +1181,5 @@ pub async fn pull_store(
};
}
if errors {
bail!("sync failed with some errors.");
}
Ok(())
Ok((progress, errors))
}