From 1e42808a59c35aae297d9dfd1c166c09f0b99347 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Thu, 14 Mar 2024 18:21:37 +0100 Subject: [PATCH] [db-snapshot] implement meta_auto_snapshot_interval --- Cargo.lock | 1 + Cargo.nix | 3 +- src/garage/server.rs | 2 +- src/model/Cargo.toml | 1 + src/model/garage.rs | 19 +++++- src/model/lib.rs | 1 + src/model/snapshot.rs | 136 ++++++++++++++++++++++++++++++++++++++++++ src/util/config.rs | 4 ++ 8 files changed, 164 insertions(+), 3 deletions(-) create mode 100644 src/model/snapshot.rs diff --git a/Cargo.lock b/Cargo.lock index f9bf0c0a..573ff5ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1438,6 +1438,7 @@ dependencies = [ "garage_util", "hex", "opentelemetry", + "parse_duration", "rand", "serde", "serde_bytes", diff --git a/Cargo.nix b/Cargo.nix index 453d83bb..a3ac4c90 100644 --- a/Cargo.nix +++ b/Cargo.nix @@ -34,7 +34,7 @@ args@{ ignoreLockHash, }: let - nixifiedLockHash = "8112e20b0e356bed77a9769600c2b2952662ec8af9548eecf8a2d46fe8433189"; + nixifiedLockHash = "f99156ba9724d370b33258f076f078fefc945f0af79292b1a246bd48bef2a9b2"; workspaceSrc = if args.workspaceSrc == null then ./. else args.workspaceSrc; currentLockHash = builtins.hashFile "sha256" (workspaceSrc + /Cargo.lock); lockHashIgnored = if ignoreLockHash @@ -2093,6 +2093,7 @@ in garage_util = (rustPackages."unknown".garage_util."0.9.3" { inherit profileName; }).out; hex = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".hex."0.4.3" { inherit profileName; }).out; opentelemetry = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".opentelemetry."0.17.0" { inherit profileName; }).out; + parse_duration = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".parse_duration."2.1.1" { inherit profileName; }).out; rand = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".rand."0.8.5" { inherit profileName; }).out; serde = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".serde."1.0.196" { inherit profileName; }).out; serde_bytes = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".serde_bytes."0.11.14" { inherit profileName; }).out; diff --git a/src/garage/server.rs b/src/garage/server.rs index 6323f957..65bf34db 100644 --- a/src/garage/server.rs +++ b/src/garage/server.rs @@ -51,7 +51,7 @@ pub async fn run_server(config_file: PathBuf, secrets: Secrets) -> Result<(), Er let (background, await_background_done) = BackgroundRunner::new(watch_cancel.clone()); info!("Spawning Garage workers..."); - garage.spawn_workers(&background); + garage.spawn_workers(&background)?; if config.admin.trace_sink.is_some() { info!("Initialize tracing..."); diff --git a/src/model/Cargo.toml b/src/model/Cargo.toml index 2e5b047d..bde354b5 100644 --- a/src/model/Cargo.toml +++ b/src/model/Cargo.toml @@ -28,6 +28,7 @@ chrono.workspace = true err-derive.workspace = true hex.workspace = true base64.workspace = true +parse_duration.workspace = true tracing.workspace = true rand.workspace = true zstd.workspace = true diff --git a/src/model/garage.rs b/src/model/garage.rs index acf943f6..a6f60546 100644 --- a/src/model/garage.rs +++ b/src/model/garage.rs @@ -278,7 +278,7 @@ impl Garage { })) } - pub fn spawn_workers(self: &Arc, bg: &BackgroundRunner) { + pub fn spawn_workers(self: &Arc, bg: &BackgroundRunner) -> Result<(), Error> { self.block_manager.spawn_workers(bg); self.bucket_table.spawn_workers(bg); @@ -299,6 +299,23 @@ impl Garage { #[cfg(feature = "k2v")] self.k2v.spawn_workers(bg); + + if let Some(itv) = self.config.metadata_auto_snapshot_interval.as_deref() { + let interval = parse_duration::parse(itv) + .ok_or_message("Invalid `metadata_auto_snapshot_interval`")?; + if interval < std::time::Duration::from_secs(600) { + return Err(Error::Message( + "metadata_auto_snapshot_interval too small or negative".into(), + )); + } + + bg.spawn_worker(crate::snapshot::AutoSnapshotWorker::new( + self.clone(), + interval, + )); + } + + Ok(()) } pub fn bucket_helper(&self) -> helper::bucket::BucketHelper { diff --git a/src/model/lib.rs b/src/model/lib.rs index 4f20ea46..8ec338da 100644 --- a/src/model/lib.rs +++ b/src/model/lib.rs @@ -19,3 +19,4 @@ pub mod s3; pub mod garage; pub mod helper; pub mod migrate; +pub mod snapshot; diff --git a/src/model/snapshot.rs b/src/model/snapshot.rs new file mode 100644 index 00000000..36f9ec7d --- /dev/null +++ b/src/model/snapshot.rs @@ -0,0 +1,136 @@ +use std::fs; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +use async_trait::async_trait; +use rand::prelude::*; +use tokio::sync::watch; + +use garage_util::background::*; +use garage_util::error::*; + +use crate::garage::Garage; + +// The two most recent snapshots are kept +const KEEP_SNAPSHOTS: usize = 2; + +static SNAPSHOT_MUTEX: Mutex<()> = Mutex::new(()); + +// ================ snapshotting logic ===================== + +/// Run snashot_metadata in a blocking thread and async await on it +pub async fn async_snapshot_metadata(garage: &Arc) -> Result<(), Error> { + let garage = garage.clone(); + let worker = tokio::task::spawn_blocking(move || snapshot_metadata(&garage)); + worker.await.unwrap()?; + Ok(()) +} + +/// Take a snapshot of the metadata database, and erase older +/// snapshots if necessary. +/// This is not an async function, it should be spawned on a thread pool +pub fn snapshot_metadata(garage: &Garage) -> Result<(), Error> { + let lock = match SNAPSHOT_MUTEX.try_lock() { + Ok(lock) => lock, + Err(_) => { + return Err(Error::Message( + "Cannot acquire lock, another snapshot might be in progress".into(), + )) + } + }; + + let mut snapshots_dir = garage.config.metadata_dir.clone(); + snapshots_dir.push("snapshots"); + fs::create_dir_all(&snapshots_dir)?; + + let mut new_path = snapshots_dir.clone(); + new_path.push(chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true)); + + info!("Snapshotting metadata db to {}", new_path.display()); + garage.db.snapshot(&new_path)?; + info!("Metadata db snapshot finished"); + + if let Err(e) = cleanup_snapshots(&snapshots_dir) { + error!("Failed to do cleanup in snapshots directory: {}", e); + } + + drop(lock); + + Ok(()) +} + +fn cleanup_snapshots(snapshots_dir: &PathBuf) -> Result<(), Error> { + let mut snapshots = + fs::read_dir(&snapshots_dir)?.collect::, std::io::Error>>()?; + + snapshots.retain(|x| x.file_name().len() > 8); + snapshots.sort_by_key(|x| x.file_name()); + + for to_delete in snapshots.iter().rev().skip(KEEP_SNAPSHOTS) { + let path = snapshots_dir.join(to_delete.path()); + if to_delete.metadata()?.file_type().is_dir() { + for file in fs::read_dir(&path)? { + let file = file?; + if file.metadata()?.is_file() { + fs::remove_file(path.join(file.path()))?; + } + } + std::fs::remove_dir(&path)?; + } else { + std::fs::remove_file(&path)?; + } + } + Ok(()) +} + +// ================ auto snapshot worker ===================== + +pub struct AutoSnapshotWorker { + garage: Arc, + next_snapshot: Instant, + snapshot_interval: Duration, +} + +impl AutoSnapshotWorker { + pub(crate) fn new(garage: Arc, snapshot_interval: Duration) -> Self { + Self { + garage, + snapshot_interval, + next_snapshot: Instant::now() + (snapshot_interval / 2), + } + } +} + +#[async_trait] +impl Worker for AutoSnapshotWorker { + fn name(&self) -> String { + "Metadata snapshot worker".into() + } + fn status(&self) -> WorkerStatus { + WorkerStatus { + freeform: vec![format!( + "Next snapshot: {}", + (chrono::Utc::now() + (self.next_snapshot - Instant::now())).to_rfc3339() + )], + ..Default::default() + } + } + async fn work(&mut self, _must_exit: &mut watch::Receiver) -> Result { + if Instant::now() < self.next_snapshot { + return Ok(WorkerState::Idle); + } + + async_snapshot_metadata(&self.garage).await?; + + let rand_factor = 1f32 + thread_rng().gen::() / 5f32; + self.next_snapshot = Instant::now() + self.snapshot_interval.mul_f32(rand_factor); + + Ok(WorkerState::Idle) + } + async fn wait_for_work(&mut self) -> WorkerState { + tokio::time::sleep_until(self.next_snapshot.into()).await; + WorkerState::Busy + } +} diff --git a/src/util/config.rs b/src/util/config.rs index 7338a506..8ecbdfbb 100644 --- a/src/util/config.rs +++ b/src/util/config.rs @@ -27,6 +27,10 @@ pub struct Config { #[serde(default)] pub disable_scrub: bool, + /// Automatic snapshot interval for metadata + #[serde(default)] + pub metadata_auto_snapshot_interval: Option, + /// Size of data blocks to save to disk #[serde( deserialize_with = "deserialize_capacity",