proxmox-rrd/src/cache.rs

   1 use std::fs::File;
   2 use std::path::{Path, PathBuf};
   3 use std::sync::{Arc, RwLock};
   4 use std::io::{BufRead, BufReader};
   5 use std::time::SystemTime;
   6 use std::thread::spawn;
   7 use std::os::unix::io::AsRawFd;
   8 use std::collections::BTreeSet;
   9
  10 use crossbeam_channel::{bounded, TryRecvError};
  11 use anyhow::{format_err, bail, Error};
  12
  13 use proxmox_sys::fs::{create_path, CreateOptions};
  14
  15 use crate::rrd::{DST, CF, RRD, RRA};
  16
  17 mod journal;
  18 use journal::*;
  19
  20 mod rrd_map;
  21 use rrd_map::*;
  22
  23 /// RRD cache - keep RRD data in RAM, but write updates to disk
  24 ///
  25 /// This cache is designed to run as single instance (no concurrent
  26 /// access from other processes).
  27 pub struct RRDCache {
  28     config: Arc<CacheConfig>,
  29     state: Arc<RwLock<JournalState>>,
  30     rrd_map: Arc<RwLock<RRDMap>>,
  31 }
  32
  33 pub(crate) struct CacheConfig {
  34     apply_interval: f64,
  35     basedir: PathBuf,
  36     file_options: CreateOptions,
  37     dir_options: CreateOptions,
  38 }
  39
  40
  41 impl RRDCache {
  42
  43     /// Creates a new instance
  44     ///
  45     /// `basedir`: All files are stored relative to this path.
  46     ///
  47     /// `file_options`: Files are created with this options.
  48     ///
  49     /// `dir_options`: Directories are created with this options.
  50     ///
  51     /// `apply_interval`: Commit journal after `apply_interval` seconds.
  52     ///
  53     /// `load_rrd_cb`; The callback function is used to load RRD files,
  54     /// and should return a newly generated RRD if the file does not
  55     /// exists (or is unreadable). This may generate RRDs with
  56     /// different configurations (dependent on `rel_path`).
  57     pub fn new<P: AsRef<Path>>(
  58         basedir: P,
  59         file_options: Option<CreateOptions>,
  60         dir_options: Option<CreateOptions>,
  61         apply_interval: f64,
  62         load_rrd_cb: fn(path: &Path, rel_path: &str, dst: DST) -> RRD,
  63     ) -> Result<Self, Error> {
  64         let basedir = basedir.as_ref().to_owned();
  65
  66         let file_options = file_options.unwrap_or_else(CreateOptions::new);
  67         let dir_options = dir_options.unwrap_or_else(CreateOptions::new);
  68
  69         create_path(&basedir, Some(dir_options.clone()), Some(dir_options.clone()))
  70             .map_err(|err: Error| format_err!("unable to create rrdb stat dir - {}", err))?;
  71
  72         let config = Arc::new(CacheConfig {
  73             basedir,
  74             file_options,
  75             dir_options,
  76             apply_interval,
  77         });
  78
  79         let state = JournalState::new(Arc::clone(&config))?;
  80         let rrd_map = RRDMap::new(Arc::clone(&config), load_rrd_cb);
  81
  82         Ok(Self {
  83             config: Arc::clone(&config),
  84             state: Arc::new(RwLock::new(state)),
  85             rrd_map: Arc::new(RwLock::new(rrd_map)),
  86         })
  87     }
  88
  89     /// Create a new RRD as used by the proxmox backup server
  90     ///
  91     /// It contains the following RRAs:
  92     ///
  93     /// * cf=average,r=60,n=1440 => 1day
  94     /// * cf=maximum,r=60,n=1440 => 1day
  95     /// * cf=average,r=30*60,n=1440 => 1month
  96     /// * cf=maximum,r=30*60,n=1440 => 1month
  97     /// * cf=average,r=6*3600,n=1440 => 1year
  98     /// * cf=maximum,r=6*3600,n=1440 => 1year
  99     /// * cf=average,r=7*86400,n=570 => 10years
 100     /// * cf=maximum,r=7*86400,n=570 => 10year
 101     ///
 102     /// The resultion data file size is about 80KB.
 103     pub fn create_proxmox_backup_default_rrd(dst: DST) -> RRD {
 104         let rra_list = vec![
 105             // 1 min * 1440 => 1 day
 106             RRA::new(CF::Average, 60, 1440),
 107             RRA::new(CF::Maximum, 60, 1440),
 108             // 30 min * 1440 => 30 days ~ 1 month
 109             RRA::new(CF::Average, 30 * 60, 1440),
 110             RRA::new(CF::Maximum, 30 * 60, 1440),
 111             // 6 h * 1440 => 360 days ~ 1 year
 112             RRA::new(CF::Average, 6 * 3600, 1440),
 113             RRA::new(CF::Maximum, 6 * 3600, 1440),
 114             // 1 week * 570 => 10 years
 115             RRA::new(CF::Average, 7 * 86400, 570),
 116             RRA::new(CF::Maximum, 7 * 86400, 570),
 117         ];
 118
 119         RRD::new(dst, rra_list)
 120     }
 121
 122     /// Sync the journal data to disk (using `fdatasync` syscall)
 123     pub fn sync_journal(&self) -> Result<(), Error> {
 124         self.state.read().unwrap().sync_journal()
 125     }
 126
 127     /// Apply and commit the journal. Should be used at server startup.
 128     pub fn apply_journal(&self) -> Result<bool, Error> {
 129         let config = Arc::clone(&self.config);
 130         let state = Arc::clone(&self.state);
 131         let rrd_map = Arc::clone(&self.rrd_map);
 132
 133
 134         let mut state_guard = self.state.write().unwrap();
 135         let journal_applied = state_guard.journal_applied;
 136
 137         if let Some(ref recv) = state_guard.apply_thread_result {
 138             match recv.try_recv() {
 139                 Ok(Ok(())) => {
 140                     // finished without errors, OK
 141                     state_guard.apply_thread_result = None;
 142                 }
 143                 Ok(Err(err)) => {
 144                     // finished with errors, log them
 145                     log::error!("{}", err);
 146                     state_guard.apply_thread_result = None;
 147                 }
 148                 Err(TryRecvError::Empty) => {
 149                     // still running
 150                     return Ok(journal_applied);
 151                 }
 152                 Err(TryRecvError::Disconnected) => {
 153                     // crashed, start again
 154                     log::error!("apply journal thread crashed - try again");
 155                     state_guard.apply_thread_result = None;
 156                 }
 157             }
 158         }
 159
 160         let now = proxmox_time::epoch_f64();
 161         let wants_commit = (now - state_guard.last_journal_flush) > self.config.apply_interval;
 162
 163         if journal_applied && !wants_commit { return Ok(journal_applied); }
 164
 165         state_guard.last_journal_flush = proxmox_time::epoch_f64();
 166
 167         let (sender, receiver) = bounded(1);
 168         state_guard.apply_thread_result = Some(receiver);
 169
 170         spawn(move || {
 171             let result = apply_and_commit_journal_thread(config, state, rrd_map, journal_applied)
 172                 .map_err(|err| err.to_string());
 173             sender.send(result).unwrap();
 174         });
 175
 176         Ok(journal_applied)
 177     }
 178
 179
 180     /// Update data in RAM and write file back to disk (journal)
 181     pub fn update_value(
 182         &self,
 183         rel_path: &str,
 184         time: f64,
 185         value: f64,
 186         dst: DST,
 187     ) -> Result<(), Error> {
 188
 189         let journal_applied = self.apply_journal()?;
 190
 191         self.state.write().unwrap()
 192             .append_journal_entry(time, value, dst, rel_path)?;
 193
 194         if journal_applied {
 195             self.rrd_map.write().unwrap().update(rel_path, time, value, dst, false)?;
 196         }
 197
 198         Ok(())
 199     }
 200
 201     /// Extract data from cached RRD
 202     ///
 203     /// `start`: Start time. If not sepecified, we simply extract 10 data points.
 204     ///
 205     /// `end`: End time. Default is to use the current time.
 206     pub fn extract_cached_data(
 207         &self,
 208         base: &str,
 209         name: &str,
 210         cf: CF,
 211         resolution: u64,
 212         start: Option<u64>,
 213         end: Option<u64>,
 214     ) -> Result<Option<(u64, u64, Vec<Option<f64>>)>, Error> {
 215         self.rrd_map.read().unwrap()
 216             .extract_cached_data(base, name, cf, resolution, start, end)
 217     }
 218 }
 219
 220
 221 fn apply_and_commit_journal_thread(
 222     config: Arc<CacheConfig>,
 223     state: Arc<RwLock<JournalState>>,
 224     rrd_map: Arc<RwLock<RRDMap>>,
 225     commit_only: bool,
 226 ) -> Result<(), Error> {
 227
 228     if commit_only {
 229         state.write().unwrap().rotate_journal()?; // start new journal, keep old one
 230     } else {
 231         let start_time = SystemTime::now();
 232         log::debug!("applying rrd journal");
 233
 234         match apply_journal_impl(Arc::clone(&state), Arc::clone(&rrd_map)) {
 235             Ok(entries) => {
 236                 let elapsed = start_time.elapsed().unwrap().as_secs_f64();
 237                 log::info!("applied rrd journal ({} entries in {:.3} seconds)", entries, elapsed);
 238             }
 239             Err(err) => bail!("apply rrd journal failed - {}", err),
 240         }
 241     }
 242
 243     let start_time = SystemTime::now();
 244     log::debug!("commit rrd journal");
 245
 246     match commit_journal_impl(config, state, rrd_map) {
 247         Ok(rrd_file_count) => {
 248             let elapsed = start_time.elapsed().unwrap().as_secs_f64();
 249             log::info!("rrd journal successfully committed ({} files in {:.3} seconds)",
 250                        rrd_file_count, elapsed);
 251         }
 252         Err(err) => bail!("rrd journal commit failed: {}", err),
 253     }
 254     Ok(())
 255 }
 256
 257 fn apply_journal_lines(
 258     state: Arc<RwLock<JournalState>>,
 259     rrd_map: Arc<RwLock<RRDMap>>,
 260     journal_name: &str, // used for logging
 261     reader: &mut BufReader<File>,
 262     lock_read_line: bool,
 263 ) -> Result<usize, Error> {
 264
 265     let mut linenr = 0;
 266
 267     loop {
 268         linenr += 1;
 269         let mut line = String::new();
 270         let len = if lock_read_line {
 271             let _lock = state.read().unwrap(); // make sure we read entire lines
 272             reader.read_line(&mut line)?
 273         } else {
 274             reader.read_line(&mut line)?
 275         };
 276
 277         if len == 0 { break; }
 278
 279         let entry: JournalEntry = match line.parse() {
 280             Ok(entry) => entry,
 281             Err(err) => {
 282                 log::warn!(
 283                     "unable to parse rrd journal '{}' line {} (skip) - {}",
 284                     journal_name, linenr, err,
 285                 );
 286                 continue; // skip unparsable lines
 287             }
 288         };
 289
 290         rrd_map.write().unwrap().update(&entry.rel_path, entry.time, entry.value, entry.dst, true)?;
 291     }
 292     Ok(linenr)
 293 }
 294
 295 fn apply_journal_impl(
 296     state: Arc<RwLock<JournalState>>,
 297     rrd_map: Arc<RwLock<RRDMap>>,
 298 ) -> Result<usize, Error> {
 299
 300     let mut lines = 0;
 301
 302     // Apply old journals first
 303     let journal_list = state.read().unwrap().list_old_journals()?;
 304
 305     for entry in journal_list {
 306         log::info!("apply old journal log {}", entry.name);
 307         let file = std::fs::OpenOptions::new().read(true).open(&entry.path)?;
 308         let mut reader = BufReader::new(file);
 309         lines += apply_journal_lines(
 310             Arc::clone(&state),
 311             Arc::clone(&rrd_map),
 312             &entry.name,
 313             &mut reader,
 314             false,
 315         )?;
 316     }
 317
 318     let mut journal = state.read().unwrap().open_journal_reader()?;
 319
 320     lines += apply_journal_lines(
 321         Arc::clone(&state),
 322         Arc::clone(&rrd_map),
 323         "rrd.journal",
 324         &mut journal,
 325         true,
 326     )?;
 327
 328     {
 329         let mut state_guard = state.write().unwrap(); // block other writers
 330
 331         lines += apply_journal_lines(
 332             Arc::clone(&state),
 333             Arc::clone(&rrd_map),
 334             "rrd.journal",
 335             &mut journal,
 336             false,
 337         )?;
 338
 339         state_guard.rotate_journal()?; // start new journal, keep old one
 340
 341         // We need to apply the journal only once, because further updates
 342         // are always directly applied.
 343         state_guard.journal_applied = true;
 344     }
 345
 346
 347     Ok(lines)
 348 }
 349
 350 fn fsync_file_or_dir(path: &Path) -> Result<(), Error> {
 351     let file = std::fs::File::open(path)?;
 352     nix::unistd::fsync(file.as_raw_fd())?;
 353     Ok(())
 354 }
 355
 356 pub(crate)fn fsync_file_and_parent(path: &Path) -> Result<(), Error> {
 357     let file = std::fs::File::open(path)?;
 358     nix::unistd::fsync(file.as_raw_fd())?;
 359     if let Some(parent) = path.parent() {
 360         fsync_file_or_dir(parent)?;
 361     }
 362     Ok(())
 363 }
 364
 365 fn rrd_parent_dir(basedir: &Path, rel_path: &str) -> PathBuf {
 366     let mut path = basedir.to_owned();
 367     let rel_path = Path::new(rel_path);
 368     if let Some(parent) = rel_path.parent() {
 369         path.push(parent);
 370     }
 371     path
 372 }
 373
 374 fn commit_journal_impl(
 375     config: Arc<CacheConfig>,
 376     state: Arc<RwLock<JournalState>>,
 377     rrd_map: Arc<RwLock<RRDMap>>,
 378 ) -> Result<usize, Error> {
 379
 380     let files = rrd_map.read().unwrap().file_list();
 381
 382     let mut rrd_file_count = 0;
 383     let mut errors = 0;
 384
 385     let mut dir_set = BTreeSet::new();
 386
 387     log::info!("write rrd data back to disk");
 388
 389     // save all RRDs - we only need a read lock here
 390     // Note: no fsync here (we do it afterwards)
 391     for rel_path in files.iter() {
 392         let parent_dir = rrd_parent_dir(&config.basedir, rel_path);
 393         dir_set.insert(parent_dir);
 394         rrd_file_count += 1;
 395         if let Err(err) = rrd_map.read().unwrap().flush_rrd_file(rel_path) {
 396             errors += 1;
 397             log::error!("unable to save rrd {}: {}", rel_path, err);
 398         }
 399     }
 400
 401     if errors != 0 {
 402         bail!("errors during rrd flush - unable to commit rrd journal");
 403     }
 404
 405     // Important: We fsync files after writing all data! This increase
 406     // the likelihood that files are already synced, so this is
 407     // much faster (although we need to re-open the files).
 408
 409     log::info!("starting rrd data sync");
 410
 411     for rel_path in files.iter() {
 412         let mut path = config.basedir.clone();
 413         path.push(&rel_path);
 414         fsync_file_or_dir(&path)
 415             .map_err(|err| format_err!("fsync rrd file {} failed - {}", rel_path, err))?;
 416     }
 417
 418     // also fsync directories
 419     for dir_path in dir_set {
 420         fsync_file_or_dir(&dir_path)
 421             .map_err(|err| format_err!("fsync rrd dir {:?} failed - {}", dir_path, err))?;
 422     }
 423
 424     // if everything went ok, remove the old journal files
 425     state.write().unwrap().remove_old_journals()?;
 426
 427     Ok(rrd_file_count)
 428 }