pbs-datastore/src/snapshot_reader.rs

   1 use std::fs::File;
   2 use std::os::unix::io::{AsRawFd, FromRawFd};
   3 use std::path::Path;
   4 use std::sync::Arc;
   5
   6 use anyhow::{bail, Error};
   7 use nix::dir::Dir;
   8
   9 use proxmox_sys::fs::lock_dir_noblock_shared;
  10
  11 use pbs_api_types::{BackupNamespace, Operation};
  12
  13 use crate::backup_info::BackupDir;
  14 use crate::dynamic_index::DynamicIndexReader;
  15 use crate::fixed_index::FixedIndexReader;
  16 use crate::index::IndexFile;
  17 use crate::manifest::{archive_type, ArchiveType, CLIENT_LOG_BLOB_NAME, MANIFEST_BLOB_NAME};
  18 use crate::DataStore;
  19
  20 /// Helper to access the contents of a datastore backup snapshot
  21 ///
  22 /// This make it easy to iterate over all used chunks and files.
  23 pub struct SnapshotReader {
  24     snapshot: BackupDir,
  25     datastore_name: String,
  26     file_list: Vec<String>,
  27     locked_dir: Dir,
  28 }
  29
  30 impl SnapshotReader {
  31     /// Lock snapshot, reads the manifest and returns a new instance
  32     pub fn new(
  33         datastore: Arc<DataStore>,
  34         ns: BackupNamespace,
  35         snapshot: pbs_api_types::BackupDir,
  36     ) -> Result<Self, Error> {
  37         Self::new_do(datastore.backup_dir(ns, snapshot)?)
  38     }
  39
  40     pub(crate) fn new_do(snapshot: BackupDir) -> Result<Self, Error> {
  41         let datastore = snapshot.datastore();
  42         let snapshot_path = snapshot.full_path();
  43
  44         let locked_dir =
  45             lock_dir_noblock_shared(&snapshot_path, "snapshot", "locked by another operation")?;
  46
  47         let datastore_name = datastore.name().to_string();
  48
  49         let manifest = match snapshot.load_manifest() {
  50             Ok((manifest, _)) => manifest,
  51             Err(err) => {
  52                 bail!(
  53                     "manifest load error on datastore '{}' snapshot '{}' - {}",
  54                     datastore_name,
  55                     snapshot,
  56                     err
  57                 );
  58             }
  59         };
  60
  61         let mut client_log_path = snapshot_path;
  62         client_log_path.push(CLIENT_LOG_BLOB_NAME);
  63
  64         let mut file_list = Vec::new();
  65         file_list.push(MANIFEST_BLOB_NAME.to_string());
  66         for item in manifest.files() {
  67             file_list.push(item.filename.clone());
  68         }
  69         if client_log_path.exists() {
  70             file_list.push(CLIENT_LOG_BLOB_NAME.to_string());
  71         }
  72
  73         Ok(Self {
  74             snapshot,
  75             datastore_name,
  76             file_list,
  77             locked_dir,
  78         })
  79     }
  80
  81     /// Return the snapshot directory
  82     pub fn snapshot(&self) -> &BackupDir {
  83         &self.snapshot
  84     }
  85
  86     /// Return the datastore name
  87     pub fn datastore_name(&self) -> &str {
  88         &self.datastore_name
  89     }
  90
  91     /// Returns the list of files the snapshot refers to.
  92     pub fn file_list(&self) -> &Vec<String> {
  93         &self.file_list
  94     }
  95
  96     /// Opens a file inside the snapshot (using openat) for reading
  97     pub fn open_file(&self, filename: &str) -> Result<File, Error> {
  98         let raw_fd = nix::fcntl::openat(
  99             self.locked_dir.as_raw_fd(),
 100             Path::new(filename),
 101             nix::fcntl::OFlag::O_RDONLY,
 102             nix::sys::stat::Mode::empty(),
 103         )?;
 104         let file = unsafe { File::from_raw_fd(raw_fd) };
 105         Ok(file)
 106     }
 107
 108     /// Returns an iterator for all chunks not skipped by `skip_fn`.
 109     pub fn chunk_iterator<F: Fn(&[u8; 32]) -> bool>(
 110         &self,
 111         skip_fn: F,
 112     ) -> Result<SnapshotChunkIterator<F>, Error> {
 113         SnapshotChunkIterator::new(self, skip_fn)
 114     }
 115 }
 116
 117 /// Iterates over all chunks used by a backup snapshot
 118 ///
 119 /// Note: The iterator returns a `Result`, and the iterator state is
 120 /// undefined after the first error. So it make no sense to continue
 121 /// iteration after the first error.
 122 pub struct SnapshotChunkIterator<'a, F: Fn(&[u8; 32]) -> bool> {
 123     snapshot_reader: &'a SnapshotReader,
 124     todo_list: Vec<String>,
 125     skip_fn: F,
 126     current_index: Option<(Arc<Box<dyn IndexFile + Send>>, usize, Vec<(usize, u64)>)>,
 127 }
 128
 129 impl<'a, F: Fn(&[u8; 32]) -> bool> Iterator for SnapshotChunkIterator<'a, F> {
 130     type Item = Result<[u8; 32], Error>;
 131
 132     fn next(&mut self) -> Option<Self::Item> {
 133         proxmox_lang::try_block!({
 134             loop {
 135                 if self.current_index.is_none() {
 136                     if let Some(filename) = self.todo_list.pop() {
 137                         let file = self.snapshot_reader.open_file(&filename)?;
 138                         let index: Box<dyn IndexFile + Send> = match archive_type(&filename)? {
 139                             ArchiveType::FixedIndex => Box::new(FixedIndexReader::new(file)?),
 140                             ArchiveType::DynamicIndex => Box::new(DynamicIndexReader::new(file)?),
 141                             _ => bail!(
 142                                 "SnapshotChunkIterator: got unknown file type - internal error"
 143                             ),
 144                         };
 145
 146                         let datastore = DataStore::lookup_datastore(
 147                             self.snapshot_reader.datastore_name(),
 148                             Some(Operation::Read),
 149                         )?;
 150                         let order =
 151                             datastore.get_chunks_in_order(&index, &self.skip_fn, |_| Ok(()))?;
 152
 153                         self.current_index = Some((Arc::new(index), 0, order));
 154                     } else {
 155                         return Ok(None);
 156                     }
 157                 }
 158                 let (index, pos, list) = self.current_index.take().unwrap();
 159                 if pos < list.len() {
 160                     let (real_pos, _) = list[pos];
 161                     let digest = *index.index_digest(real_pos).unwrap();
 162                     self.current_index = Some((index, pos + 1, list));
 163                     return Ok(Some(digest));
 164                 } else {
 165                     // pop next index
 166                 }
 167             }
 168         })
 169         .transpose()
 170     }
 171 }
 172
 173 impl<'a, F: Fn(&[u8; 32]) -> bool> SnapshotChunkIterator<'a, F> {
 174     pub fn new(snapshot_reader: &'a SnapshotReader, skip_fn: F) -> Result<Self, Error> {
 175         let mut todo_list = Vec::new();
 176
 177         for filename in snapshot_reader.file_list() {
 178             match archive_type(filename)? {
 179                 ArchiveType::FixedIndex | ArchiveType::DynamicIndex => {
 180                     todo_list.push(filename.to_owned());
 181                 }
 182                 ArchiveType::Blob => { /* no chunks, do nothing */ }
 183             }
 184         }
 185
 186         Ok(Self {
 187             snapshot_reader,
 188             todo_list,
 189             current_index: None,
 190             skip_fn,
 191         })
 192     }
 193 }