From ff70e09aa090a140729483783076e7938fad0d39 Mon Sep 17 00:00:00 2001 From: Jonathan Davies Date: Thu, 9 Feb 2023 13:48:48 +0000 Subject: util/forwarded_headers.rs: Generalized handle_forwarded_for_headers() here. --- src/util/forwarded_headers.rs | 63 +++++++++++++++++++++++++++++++++++++++++++ src/util/lib.rs | 1 + 2 files changed, 64 insertions(+) create mode 100644 src/util/forwarded_headers.rs (limited to 'src') diff --git a/src/util/forwarded_headers.rs b/src/util/forwarded_headers.rs new file mode 100644 index 00000000..6ae275aa --- /dev/null +++ b/src/util/forwarded_headers.rs @@ -0,0 +1,63 @@ +use http::{HeaderMap, HeaderValue}; +use std::net::IpAddr; +use std::str::FromStr; + +use crate::error::{Error, OkOrMessage}; + +pub fn handle_forwarded_for_headers(headers: &HeaderMap) -> Result { + let forwarded_for_header = headers + .get("x-forwarded-for") + .ok_or_message("X-Forwarded-For header not provided")?; + + let forwarded_for_ip_str = forwarded_for_header + .to_str() + .ok_or_message("Error parsing X-Forwarded-For header")?; + + let client_ip = IpAddr::from_str(&forwarded_for_ip_str) + .ok_or_message("Valid IP address not found in X-Forwarded-For header")?; + + Ok(client_ip.to_string()) +} + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn test_handle_forwarded_for_headers_ipv4_client() { + let mut test_headers = HeaderMap::new(); + test_headers.insert("X-Forwarded-For", "192.0.2.100".parse().unwrap()); + + if let Ok(forwarded_ip) = handle_forwarded_for_headers(&test_headers) { + assert_eq!(forwarded_ip, "192.0.2.100"); + } + } + + #[test] + fn test_handle_forwarded_for_headers_ipv6_client() { + let mut test_headers = HeaderMap::new(); + test_headers.insert("X-Forwarded-For", "2001:db8::f00d:cafe".parse().unwrap()); + + if let Ok(forwarded_ip) = handle_forwarded_for_headers(&test_headers) { + assert_eq!(forwarded_ip, "2001:db8::f00d:cafe"); + } + } + + #[test] + fn test_handle_forwarded_for_headers_invalid_ip() { + let mut test_headers = HeaderMap::new(); + test_headers.insert("X-Forwarded-For", "www.example.com".parse().unwrap()); + + let result = handle_forwarded_for_headers(&test_headers); + assert!(result.is_err()); + } + + #[test] + fn test_handle_forwarded_for_headers_missing() { + let mut test_headers = HeaderMap::new(); + test_headers.insert("Host", "www.deuxfleurs.fr".parse().unwrap()); + + let result = handle_forwarded_for_headers(&test_headers); + assert!(result.is_err()); + } +} diff --git a/src/util/lib.rs b/src/util/lib.rs index d35ca72f..c9110fb2 100644 --- a/src/util/lib.rs +++ b/src/util/lib.rs @@ -11,6 +11,7 @@ pub mod data; pub mod encode; pub mod error; pub mod formater; +pub mod forwarded_headers; pub mod metrics; pub mod migrate; pub mod persister; -- cgit v1.2.3 From b8123fb6cdf15ad21e5d254368c96f6f32ba8c3c Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Mon, 6 Mar 2023 11:38:49 +0100 Subject: Clearer error message when LMDB has oom error (fix #517) --- src/model/garage.rs | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) (limited to 'src') diff --git a/src/model/garage.rs b/src/model/garage.rs index 0a9ec608..3daa1b33 100644 --- a/src/model/garage.rs +++ b/src/model/garage.rs @@ -136,9 +136,16 @@ impl Garage { env_builder.flag(heed::flags::Flags::MdbNoSync); env_builder.flag(heed::flags::Flags::MdbNoMetaSync); } - let db = env_builder - .open(&db_path) - .ok_or_message("Unable to open LMDB DB")?; + let db = match env_builder.open(&db_path) { + Err(heed::Error::Io(e)) if e.kind() == std::io::ErrorKind::OutOfMemory => { + return Err(Error::Message( + "OutOfMemory error while trying to open LMDB database. This can happen \ + if your operating system is not allowing you to use sufficient virtual \ + memory address space. Please check that no limit is set (ulimit -v). \ + On 32-bit machines, you should probably switch to another database engine.".into())) + } + x => x.ok_or_message("Unable to open LMDB DB")?, + }; db::lmdb_adapter::LmdbDb::init(db) } #[cfg(not(feature = "lmdb"))] -- cgit v1.2.3 From e4e5196066a75ed9327a62d8082decb1e0c381e7 Mon Sep 17 00:00:00 2001 From: Jonathan Davies Date: Thu, 9 Feb 2023 13:49:43 +0000 Subject: api/generic_server.rs: Use new handle_forwarded_for_headers() function. --- src/api/generic_server.rs | 13 ++++--------- 1 file changed, 4 insertions(+), 9 deletions(-) (limited to 'src') diff --git a/src/api/generic_server.rs b/src/api/generic_server.rs index aa90868a..d0354d28 100644 --- a/src/api/generic_server.rs +++ b/src/api/generic_server.rs @@ -19,6 +19,7 @@ use opentelemetry::{ }; use garage_util::error::Error as GarageError; +use garage_util::forwarded_headers; use garage_util::metrics::{gen_trace_id, RecordDuration}; pub(crate) trait ApiEndpoint: Send + Sync + 'static { @@ -126,15 +127,9 @@ impl ApiServer { ) -> Result, GarageError> { let uri = req.uri().clone(); - let has_forwarded_for_header = req.headers().contains_key("x-forwarded-for"); - if has_forwarded_for_header { - let forwarded_for_ip_addr = &req - .headers() - .get("x-forwarded-for") - .expect("Could not parse X-Forwarded-For header") - .to_str() - .unwrap_or_default(); - + if let Ok(forwarded_for_ip_addr) = + forwarded_headers::handle_forwarded_for_headers(&req.headers()) + { info!( "{} (via {}) {} {}", forwarded_for_ip_addr, -- cgit v1.2.3 From 4e0fc3d6c9d764f2eecb7cf9e995580e734bdb2d Mon Sep 17 00:00:00 2001 From: Jonathan Davies Date: Thu, 9 Feb 2023 13:50:04 +0000 Subject: web/web_server.rs: Handle X-Forwarded-For here too. --- src/web/web_server.rs | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) (limited to 'src') diff --git a/src/web/web_server.rs b/src/web/web_server.rs index 5719da54..0c7edf23 100644 --- a/src/web/web_server.rs +++ b/src/web/web_server.rs @@ -29,6 +29,7 @@ use garage_model::garage::Garage; use garage_table::*; use garage_util::error::Error as GarageError; +use garage_util::forwarded_headers; use garage_util::metrics::{gen_trace_id, RecordDuration}; struct WebMetrics { @@ -104,7 +105,19 @@ impl WebServer { req: Request, addr: SocketAddr, ) -> Result, Infallible> { - info!("{} {} {}", addr, req.method(), req.uri()); + if let Ok(forwarded_for_ip_addr) = + forwarded_headers::handle_forwarded_for_headers(&req.headers()) + { + info!( + "{} (via {}) {} {}", + forwarded_for_ip_addr, + addr, + req.method(), + req.uri() + ); + } else { + info!("{} {} {}", addr, req.method(), req.uri()); + } // Lots of instrumentation let tracer = opentelemetry::global::tracer("garage"); -- cgit v1.2.3 From 53d09eb00f29fe5c7a8a14fa22ce6cfc64bb0b14 Mon Sep 17 00:00:00 2001 From: Jonathan Davies Date: Sat, 4 Mar 2023 16:16:10 +0000 Subject: block/repair.rs: Added function and time_next_run_scrub with a random element of 10 days to SCRUB_INTERVAL to help balance scrub load across cluster. --- src/block/repair.rs | 48 +++++++++++++++++++++++++++++++++++++----------- 1 file changed, 37 insertions(+), 11 deletions(-) (limited to 'src') diff --git a/src/block/repair.rs b/src/block/repair.rs index d4593dbf..10d46291 100644 --- a/src/block/repair.rs +++ b/src/block/repair.rs @@ -4,6 +4,7 @@ use std::sync::Arc; use std::time::Duration; use async_trait::async_trait; +use rand::Rng; use serde::{Deserialize, Serialize}; use tokio::fs; use tokio::select; @@ -19,8 +20,8 @@ use garage_util::tranquilizer::Tranquilizer; use crate::manager::*; -// Full scrub every 30 days -const SCRUB_INTERVAL: Duration = Duration::from_secs(3600 * 24 * 30); +// Full scrub every 25 days with a random element of 10 days mixed in below +const SCRUB_INTERVAL: Duration = Duration::from_secs(3600 * 24 * 25); // Scrub tranquility is initially set to 4, but can be changed in the CLI // and the updated version is persisted over Garage restarts const INITIAL_SCRUB_TRANQUILITY: u32 = 4; @@ -175,13 +176,30 @@ pub struct ScrubWorker { pub struct ScrubWorkerPersisted { pub tranquility: u32, pub(crate) time_last_complete_scrub: u64, + pub(crate) time_next_run_scrub: u64, pub(crate) corruptions_detected: u64, } + +fn randomize_next_scrub_run_time() -> u64 { + // Take SCRUB_INTERVAL and mix in a random interval of 10 days to attempt to + // balance scrub load across different cluster nodes. + + let next_run_timestamp = now_msec() + + SCRUB_INTERVAL + .saturating_add(Duration::from_secs( + rand::thread_rng().gen_range(0..3600 * 24 * 10), + )) + .as_millis() as u64; + + next_run_timestamp +} + impl garage_util::migrate::InitialFormat for ScrubWorkerPersisted {} impl Default for ScrubWorkerPersisted { fn default() -> Self { ScrubWorkerPersisted { time_last_complete_scrub: 0, + time_next_run_scrub: randomize_next_scrub_run_time(), tranquility: INITIAL_SCRUB_TRANQUILITY, corruptions_detected: 0, } @@ -279,12 +297,13 @@ impl Worker for ScrubWorker { } fn status(&self) -> WorkerStatus { - let (corruptions_detected, tranquility, time_last_complete_scrub) = + let (corruptions_detected, tranquility, time_last_complete_scrub, time_next_run_scrub) = self.persister.get_with(|p| { ( p.corruptions_detected, p.tranquility, p.time_last_complete_scrub, + p.time_next_run_scrub, ) }); @@ -302,10 +321,16 @@ impl Worker for ScrubWorker { s.freeform = vec![format!("Scrub paused, resumes at {}", msec_to_rfc3339(*rt))]; } ScrubWorkerState::Finished => { - s.freeform = vec![format!( - "Last scrub completed at {}", - msec_to_rfc3339(time_last_complete_scrub) - )]; + s.freeform = vec![ + format!( + "Last scrub completed at {}", + msec_to_rfc3339(time_last_complete_scrub), + ), + format!( + "Next scrub scheduled for {}", + msec_to_rfc3339(time_next_run_scrub) + ), + ]; } } s @@ -334,8 +359,10 @@ impl Worker for ScrubWorker { .tranquilizer .tranquilize_worker(self.persister.get_with(|p| p.tranquility))) } else { - self.persister - .set_with(|p| p.time_last_complete_scrub = now_msec())?; + self.persister.set_with(|p| { + p.time_last_complete_scrub = now_msec(); + p.time_next_run_scrub = randomize_next_scrub_run_time(); + })?; self.work = ScrubWorkerState::Finished; self.tranquilizer.clear(); Ok(WorkerState::Idle) @@ -350,8 +377,7 @@ impl Worker for ScrubWorker { ScrubWorkerState::Running(_) => return WorkerState::Busy, ScrubWorkerState::Paused(_, resume_time) => (*resume_time, ScrubWorkerCommand::Resume), ScrubWorkerState::Finished => ( - self.persister.get_with(|p| p.time_last_complete_scrub) - + SCRUB_INTERVAL.as_millis() as u64, + self.persister.get_with(|p| p.time_next_run_scrub), ScrubWorkerCommand::Start, ), }; -- cgit v1.2.3 From 148b66b843be7f79f15b0a3805f38a0c3e944214 Mon Sep 17 00:00:00 2001 From: Jonathan Davies Date: Mon, 6 Mar 2023 12:31:03 +0000 Subject: block/manager.rs: Display scrub-next-run. --- src/block/manager.rs | 3 +++ 1 file changed, 3 insertions(+) (limited to 'src') diff --git a/src/block/manager.rs b/src/block/manager.rs index 051a9f93..bfa66069 100644 --- a/src/block/manager.rs +++ b/src/block/manager.rs @@ -185,6 +185,9 @@ impl BlockManager { vars.register_ro(&self.scrub_persister, "scrub-last-completed", |p| { p.get_with(|x| msec_to_rfc3339(x.time_last_complete_scrub)) }); + vars.register_ro(&self.scrub_persister, "scrub-next-run", |p| { + p.get_with(|x| msec_to_rfc3339(x.time_next_run_scrub)) + }); vars.register_ro(&self.scrub_persister, "scrub-corruptions_detected", |p| { p.get_with(|x| x.corruptions_detected) }); -- cgit v1.2.3 From b70cc0a94087bfd70931ff6741299823b48ad291 Mon Sep 17 00:00:00 2001 From: Jonathan Davies Date: Thu, 9 Mar 2023 15:34:14 +0000 Subject: block/repair.rs: Added migration for ScrubWorkerPersisted's time_next_run_scrub. Fixes: #520. --- src/block/repair.rs | 54 +++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 44 insertions(+), 10 deletions(-) (limited to 'src') diff --git a/src/block/repair.rs b/src/block/repair.rs index 10d46291..006cc866 100644 --- a/src/block/repair.rs +++ b/src/block/repair.rs @@ -5,7 +5,6 @@ use std::time::Duration; use async_trait::async_trait; use rand::Rng; -use serde::{Deserialize, Serialize}; use tokio::fs; use tokio::select; use tokio::sync::mpsc; @@ -162,6 +161,50 @@ impl Worker for RepairWorker { // and whose parameter (esp. speed) can be controlled at runtime. // ---- ---- ---- +mod v081 { + use serde::{Deserialize, Serialize}; + + #[derive(Serialize, Deserialize)] + pub struct ScrubWorkerPersisted { + pub tranquility: u32, + pub(crate) time_last_complete_scrub: u64, + pub(crate) corruptions_detected: u64, + } + + impl garage_util::migrate::InitialFormat for ScrubWorkerPersisted {} +} + +mod v082 { + use serde::{Deserialize, Serialize}; + + use super::v081; + + #[derive(Serialize, Deserialize)] + pub struct ScrubWorkerPersisted { + pub tranquility: u32, + pub(crate) time_last_complete_scrub: u64, + pub(crate) time_next_run_scrub: u64, + pub(crate) corruptions_detected: u64, + } + + impl garage_util::migrate::Migrate for ScrubWorkerPersisted { + type Previous = v081::ScrubWorkerPersisted; + + fn migrate(old: v081::ScrubWorkerPersisted) -> ScrubWorkerPersisted { + use crate::repair::randomize_next_scrub_run_time; + + ScrubWorkerPersisted { + tranquility: old.tranquility, + time_last_complete_scrub: old.time_last_complete_scrub, + time_next_run_scrub: randomize_next_scrub_run_time(), + corruptions_detected: old.corruptions_detected, + } + } + } +} + +pub use v082::*; + pub struct ScrubWorker { manager: Arc, rx_cmd: mpsc::Receiver, @@ -172,14 +215,6 @@ pub struct ScrubWorker { persister: PersisterShared, } -#[derive(Serialize, Deserialize)] -pub struct ScrubWorkerPersisted { - pub tranquility: u32, - pub(crate) time_last_complete_scrub: u64, - pub(crate) time_next_run_scrub: u64, - pub(crate) corruptions_detected: u64, -} - fn randomize_next_scrub_run_time() -> u64 { // Take SCRUB_INTERVAL and mix in a random interval of 10 days to attempt to // balance scrub load across different cluster nodes. @@ -194,7 +229,6 @@ fn randomize_next_scrub_run_time() -> u64 { next_run_timestamp } -impl garage_util::migrate::InitialFormat for ScrubWorkerPersisted {} impl Default for ScrubWorkerPersisted { fn default() -> Self { ScrubWorkerPersisted { -- cgit v1.2.3 From 7b65dd24e2bacc20afa747dc2b1f3fb81249f688 Mon Sep 17 00:00:00 2001 From: Jonathan Davies Date: Thu, 9 Mar 2023 16:32:22 +0000 Subject: block/repair.rs: Added a timestamp argument to randomize_next_scrub_run_time(). --- src/block/repair.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) (limited to 'src') diff --git a/src/block/repair.rs b/src/block/repair.rs index 006cc866..5476bf8a 100644 --- a/src/block/repair.rs +++ b/src/block/repair.rs @@ -196,7 +196,7 @@ mod v082 { ScrubWorkerPersisted { tranquility: old.tranquility, time_last_complete_scrub: old.time_last_complete_scrub, - time_next_run_scrub: randomize_next_scrub_run_time(), + time_next_run_scrub: randomize_next_scrub_run_time(old.time_last_complete_scrub), corruptions_detected: old.corruptions_detected, } } @@ -215,11 +215,11 @@ pub struct ScrubWorker { persister: PersisterShared, } -fn randomize_next_scrub_run_time() -> u64 { +fn randomize_next_scrub_run_time(timestamp: u64) -> u64 { // Take SCRUB_INTERVAL and mix in a random interval of 10 days to attempt to // balance scrub load across different cluster nodes. - let next_run_timestamp = now_msec() + let next_run_timestamp = timestamp + SCRUB_INTERVAL .saturating_add(Duration::from_secs( rand::thread_rng().gen_range(0..3600 * 24 * 10), @@ -233,7 +233,7 @@ impl Default for ScrubWorkerPersisted { fn default() -> Self { ScrubWorkerPersisted { time_last_complete_scrub: 0, - time_next_run_scrub: randomize_next_scrub_run_time(), + time_next_run_scrub: randomize_next_scrub_run_time(now_msec()), tranquility: INITIAL_SCRUB_TRANQUILITY, corruptions_detected: 0, } @@ -395,7 +395,7 @@ impl Worker for ScrubWorker { } else { self.persister.set_with(|p| { p.time_last_complete_scrub = now_msec(); - p.time_next_run_scrub = randomize_next_scrub_run_time(); + p.time_next_run_scrub = randomize_next_scrub_run_time(now_msec()); })?; self.work = ScrubWorkerState::Finished; self.tranquilizer.clear(); -- cgit v1.2.3 From d218f475cbe6450f0402424e01b99c4630cb8a2f Mon Sep 17 00:00:00 2001 From: Jonathan Davies Date: Thu, 9 Mar 2023 17:08:47 +0000 Subject: block/manager.rs: Set defaults for scrub_persister. --- src/block/manager.rs | 1 + 1 file changed, 1 insertion(+) (limited to 'src') diff --git a/src/block/manager.rs b/src/block/manager.rs index bfa66069..26278974 100644 --- a/src/block/manager.rs +++ b/src/block/manager.rs @@ -152,6 +152,7 @@ impl BlockManager { tx_scrub_command: ArcSwapOption::new(None), }); block_manager.endpoint.set_handler(block_manager.clone()); + block_manager.scrub_persister.set_with(|_| ()).unwrap(); block_manager } -- cgit v1.2.3