diff --git a/src/cmd/Cargo.toml b/src/cmd/Cargo.toml index 2c925de57c8..0ef2f35b371 100644 --- a/src/cmd/Cargo.toml +++ b/src/cmd/Cargo.toml @@ -125,6 +125,7 @@ api.workspace = true client = { workspace = true, features = ["testing"] } common-test-util.workspace = true common-version.workspace = true +datanode = { workspace = true, features = ["testing"] } file-engine.workspace = true metric-engine.workspace = true mito2.workspace = true diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs index bc6a62cd4d5..76c95c26d32 100644 --- a/src/cmd/src/standalone.rs +++ b/src/cmd/src/standalone.rs @@ -48,8 +48,8 @@ use common_meta::wal_provider::{WalProvider, WalProviderRef, build_wal_provider} use common_options::plugin_options::StandaloneFlag; use common_procedure::ProcedureManagerRef; use common_query::prelude::set_default_prefix; -use common_telemetry::info; use common_telemetry::logging::{DEFAULT_LOGGING_DIR, TracingOptions}; +use common_telemetry::{info, warn}; use common_time::timezone::set_default_timezone; use common_version::{short_version, verbose_version}; use common_wal::config::DatanodeWalConfig; @@ -274,32 +274,53 @@ impl App for Instance { Ok(()) } + /// Stops every component in order. A failed step does not skip the later + /// ones; the first error is returned and the rest are logged. async fn stop(&mut self) -> Result<()> { - self.frontend - .shutdown() - .await - .context(error::ShutdownFrontendSnafu)?; + let mut first_error = None; + let mut record = |result: Result<()>| { + if let Err(err) = result { + if first_error.is_none() { + first_error = Some(err); + } else { + warn!(err; "Ignored a later shutdown error"); + } + } + }; - self.leader_services_controller - .stop( - self.procedure_manager.clone(), - self.datanode.region_server(), - ) - .await?; + record( + self.frontend + .shutdown() + .await + .context(error::ShutdownFrontendSnafu), + ); - self.datanode - .shutdown() - .await - .context(error::ShutdownDatanodeSnafu)?; + record( + self.leader_services_controller + .stop( + self.procedure_manager.clone(), + self.datanode.region_server(), + ) + .await, + ); - self.flownode - .shutdown() - .await - .context(error::ShutdownFlownodeSnafu)?; + record( + self.datanode + .shutdown() + .await + .context(error::ShutdownDatanodeSnafu), + ); + + record( + self.flownode + .shutdown() + .await + .context(error::ShutdownFlownodeSnafu), + ); info!("Datanode instance stopped."); - Ok(()) + first_error.map_or(Ok(()), Err) } } @@ -1067,10 +1088,89 @@ mod tests { use frontend::frontend::FrontendOptions; use object_store::config::{FileConfig, GcsConfig}; use servers::grpc::GrpcOptions; + use servers::server::ServerHandlers; + use store_api::logstore::LogStore; use super::*; use crate::options::GlobalOptions; + struct FailingLeaderServices; + + #[async_trait] + impl StandaloneLeaderServicesController for FailingLeaderServices { + async fn start(&self, _context: LeaderServicesContext) -> Result<()> { + Ok(()) + } + + async fn stop( + &self, + _procedure_manager: ProcedureManagerRef, + _region_server: RegionServer, + ) -> Result<()> { + error::IllegalConfigSnafu { + msg: "leader services failed to stop", + } + .fail() + } + } + + struct FailingServer; + + #[async_trait] + impl servers::server::Server for FailingServer { + async fn shutdown(&self) -> servers::error::Result<()> { + servers::error::InternalSnafu { + err_msg: "server failed to stop", + } + .fail() + } + + async fn start(&mut self, _listening: SocketAddr) -> servers::error::Result<()> { + Ok(()) + } + + fn name(&self) -> &str { + "failing" + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } + } + + #[tokio::test] + async fn test_stop_runs_every_step_and_returns_the_first_error() { + common_telemetry::init_default_ut_logging(); + let data_home = create_temp_dir("standalone-stop"); + let mut opts = StandaloneOptions { + wal: DatanodeWalConfig::ObjectStore(ObjectStoreWalConfig::default()), + ..Default::default() + }; + opts.storage.data_home = data_home.path().to_str().unwrap().to_string(); + let creator = InstanceCreator::default() + .with_leader_services_controller(Box::new(FailingLeaderServices)); + let (mut instance, _) = StartCommand::build_with(opts, vec![], creator) + .await + .unwrap(); + let log_store = instance.datanode.object_store_log_store().unwrap(); + // The datanode fails to stop its servers after the leader services failed. + let mut services = ServerHandlers::default(); + services.insert((Box::new(FailingServer), "127.0.0.1:0".parse().unwrap())); + services.start_all().await.unwrap(); + instance.datanode.setup_services(services); + + let err = instance.stop().await.unwrap_err(); + + assert!( + matches!(&err, error::Error::IllegalConfig { msg, .. } if msg == "leader services failed to stop"), + "unexpected error: {err:?}" + ); + assert!(matches!( + log_store.append_batch(vec![]).await, + Err(log_store::error::Error::ObjectStoreWalStopped { .. }) + )); + } + #[tokio::test] async fn test_build_standalone_wal_provider() { let kv_backend = Arc::new(MemoryKvBackend::new()) as KvBackendRef; diff --git a/src/datanode/src/datanode.rs b/src/datanode/src/datanode.rs index 8a80083aef3..fdac0597572 100644 --- a/src/datanode/src/datanode.rs +++ b/src/datanode/src/datanode.rs @@ -179,6 +179,13 @@ impl Datanode { self.region_server.clone() } + /// Returns the object store log store when the datanode runs on the object + /// store WAL. + #[cfg(any(test, feature = "testing"))] + pub fn object_store_log_store(&self) -> Option> { + self.object_store_log_store.clone() + } + pub fn plugins(&self) -> Plugins { self.plugins.clone() } diff --git a/src/log-store/src/metrics.rs b/src/log-store/src/metrics.rs index f12ad3e815a..824058d70af 100644 --- a/src/log-store/src/metrics.rs +++ b/src/log-store/src/metrics.rs @@ -89,4 +89,12 @@ lazy_static! { &[LOGSTORE_LABEL, PARTITION_LABEL], ) .unwrap(); + + /// Counter of segments a read of the object store logstore skipped because + /// they did not decode. + pub static ref METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL: IntCounter = register_int_counter!( + "greptime_logstore_object_store_wal_skipped_segments_total", + "object store logstore skipped corrupted segments total", + ) + .unwrap(); } diff --git a/src/log-store/src/object_store_wal.rs b/src/log-store/src/object_store_wal.rs index 2ec36b30463..d4bfc096393 100644 --- a/src/log-store/src/object_store_wal.rs +++ b/src/log-store/src/object_store_wal.rs @@ -50,7 +50,9 @@ //! Recovery lists the objects, reads and verifies only the header, trailer and //! footer of each, and indexes the footers of the chain in sequence order to //! rebuild the object catalog, which rejects a sequence it already holds. -//! Segments are read and checksummed only when a read decodes them. +//! Segments are read and checksummed only when a read decodes them; a segment +//! that does not decode is fetched once more and then skipped as a hole of its +//! region or fails the read, as `on_corrupted_segment` says. //! //! Entry ids are object-sequence-major, see [`entry_id`]: the high bits of an //! id name the object that holds the entry, the low bits its position among @@ -63,4 +65,4 @@ mod io; mod store; pub use batch::entry_id; -pub use store::ObjectStoreLogStore; +pub use store::{ObjectStoreLogStore, WalHole}; diff --git a/src/log-store/src/object_store_wal/store.rs b/src/log-store/src/object_store_wal/store.rs index 5a6454123dd..8750a61cc4b 100644 --- a/src/log-store/src/object_store_wal/store.rs +++ b/src/log-store/src/object_store_wal/store.rs @@ -23,8 +23,8 @@ use std::time::{Duration, Instant}; use async_stream::try_stream; use bytes::Bytes; -use common_telemetry::info; -use common_wal::config::object_store::{AckMode, ObjectStoreWalConfig}; +use common_telemetry::{info, warn}; +use common_wal::config::object_store::{AckMode, CorruptedSegmentAction, ObjectStoreWalConfig}; use futures::future::BoxFuture; use futures::stream::FuturesUnordered; use futures::{StreamExt, TryStreamExt}; @@ -46,6 +46,7 @@ use crate::error::{ StaleWalObjectSnafu, UnconfirmedWalEpochStartSnafu, WalObjectSequenceExhaustedSnafu, WalObjectSequenceUnsettledSnafu, }; +use crate::metrics::METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL; use crate::object_store_wal::batch::{OBJECT_SEQ_LIMIT, OpenBatch, entry_id, sequence_floor}; use crate::object_store_wal::catalog::ObjectCatalog; use crate::object_store_wal::format::{ @@ -87,13 +88,17 @@ const RECOVERY_TAIL_WINDOW: usize = 64 * 1024; /// indexed, so an acknowledged entry never has a missing predecessor; in the /// `enqueued` mode it returns on admission and the object is created in the /// background. While [`MAX_SEALED_BATCHES`] batches -/// wait to become durable no append is admitted. +/// wait to become durable no append is admitted. A segment that a read cannot +/// decode is handled as `on_corrupted_segment` says. pub struct ObjectStoreLogStore { prefix: String, ack_mode: AckMode, + on_corrupted_segment: CorruptedSegmentAction, io: Arc, catalog: Arc>, obsolete_entry_ids: ObsoleteEntryIds, + /// Segments that reads skipped, per region and by object sequence. + wal_holes: WalHoles, terminal_error: TerminalError, stopped: Arc, command_tx: mpsc::Sender, @@ -114,6 +119,20 @@ pub struct ObjectStoreLogStore { type ObsoleteEntryIds = Arc>>; +type WalHoles = Arc>>>; + +/// A segment a read skipped because it did not decode: the entries of one +/// region in one object are missing from every read of that region. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WalHole { + /// Key of the object that holds the segment. + pub path: String, + pub object_seq: u64, + /// Entry id range the footer records for the segment. + pub min_entry_id: EntryId, + pub max_entry_id: EntryId, +} + type TerminalError = Arc>>>; impl fmt::Debug for ObjectStoreLogStore { @@ -121,6 +140,7 @@ impl fmt::Debug for ObjectStoreLogStore { f.debug_struct("ObjectStoreLogStore") .field("prefix", &self.prefix) .field("ack_mode", &self.ack_mode) + .field("on_corrupted_segment", &self.on_corrupted_segment) .finish_non_exhaustive() } } @@ -260,9 +280,11 @@ impl ObjectStoreLogStore { Ok(Arc::new(Self { prefix, ack_mode: config.ack_mode, + on_corrupted_segment: config.on_corrupted_segment, io, catalog, obsolete_entry_ids, + wal_holes: WalHoles::default(), terminal_error, stopped, command_tx, @@ -297,6 +319,21 @@ impl ObjectStoreLogStore { catalog.region_max_entry_id(region_id).unwrap_or(0) } + /// Returns the segments of the provider's region that reads of this store + /// skipped because they did not decode, ordered by object sequence. Holes + /// are kept in memory only: a store opened later on the same prefix + /// records them again when its reads meet the segments. + pub fn wal_holes(&self, provider: &Provider) -> Result> { + let region_id = self.region_of(provider)?; + Ok(self + .wal_holes + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get(®ion_id) + .map(|holes| holes.values().cloned().collect()) + .unwrap_or_default()) + } + /// Returns the region of `provider`, which must select this store's prefix. fn region_of(&self, provider: &Provider) -> Result { let provider = @@ -477,6 +514,30 @@ impl ObjectStoreLogStore { pub fn begin_stop(&self) { self.stopped.store(true, Ordering::Release); } + + /// Returns the key of the object `object_seq` and the byte range the + /// segment of the provider's region occupies in it, so a test can damage + /// that segment alone, or `None` when the object holds no segment of the + /// region. + pub fn segment_location( + &self, + provider: &Provider, + object_seq: u64, + ) -> Result)>> { + let region_id = self.region_of(provider)?; + let catalog = self.catalog.read().unwrap_or_else(PoisonError::into_inner); + Ok(catalog + .objects_for_entry_range(region_id, 0, EntryId::MAX)? + .into_iter() + .find(|(seq, _)| *seq == object_seq) + .map(|(_, entry)| { + let start = entry.segment_offset; + ( + self.io.object_path(object_seq), + start..start + entry.segment_len, + ) + })) + } } #[async_trait::async_trait] @@ -534,6 +595,9 @@ impl LogStore for ObjectStoreLogStore { /// Reads the provider's region from `entry_id`, hiding obsolete entries. /// The catalog locates each segment, so a caller-supplied index is unnecessary. + /// A segment that does not decode fails the read or is skipped and recorded + /// as a hole, as `on_corrupted_segment` says; a failed fetch always fails + /// the read. async fn read( &self, provider: &Provider, @@ -568,15 +632,20 @@ impl LogStore for ObjectStoreLogStore { let io = self.io.clone(); let provider = provider.clone(); + let on_corrupted_segment = self.on_corrupted_segment; + let wal_holes = self.wal_holes.clone(); Ok(Box::pin(try_stream! { for (object_seq, footer_entry) in objects { - let bytes = io - .get_range(object_seq, footer_entry.segment_offset, footer_entry.segment_len) - .await?; - let records = decode_segment(&bytes, &footer_entry) - .with_context(|_| InvalidWalObjectSnafu { - path: io.object_path(object_seq), - })?; + let records = match fetch_segment(io.as_ref(), object_seq, &footer_entry).await { + Ok(records) => records, + Err(error @ Error::InvalidWalObject { .. }) + if on_corrupted_segment == CorruptedSegmentAction::Skip => + { + record_skipped_segment(&wal_holes, io.as_ref(), object_seq, &footer_entry, &error); + continue; + } + Err(error) => Err(error)?, + }; let entries = records .into_iter() .filter(|record| record.entry_id >= start_entry_id) @@ -1646,6 +1715,59 @@ fn shared(error: &Arc) -> Error { ObjectStoreWalSnafu.into_error(error.clone()) } +/// Fetches and decodes the segment `entry` describes in the object +/// `object_seq`. A segment that does not decode is fetched once more, since +/// the download rather than the object may be damaged; when it still does not +/// decode the error is returned as [`Error::InvalidWalObject`]. A failed fetch +/// is returned as is. +async fn fetch_segment( + io: &dyn WalObjectIo, + object_seq: u64, + entry: &FooterEntry, +) -> Result> { + let fetch = || io.get_range(object_seq, entry.segment_offset, entry.segment_len); + if let Ok(records) = decode_segment(&fetch().await?, entry) { + return Ok(records); + } + decode_segment(&fetch().await?, entry).with_context(|_| InvalidWalObjectSnafu { + path: io.object_path(object_seq), + }) +} + +/// Records the segment `entry` describes in the object `object_seq` as a hole +/// of its region, once per object, counts the skip and warns about it. +fn record_skipped_segment( + wal_holes: &WalHoles, + io: &dyn WalObjectIo, + object_seq: u64, + entry: &FooterEntry, + error: &Error, +) { + let hole = WalHole { + path: io.object_path(object_seq), + object_seq, + min_entry_id: entry.min_entry_id, + max_entry_id: entry.max_entry_id, + }; + warn!( + error; + "Skipped a corrupted WAL segment of region {}, object {} (sequence {}), entry ids {}..={}", + entry.region_id, + hole.path, + object_seq, + hole.min_entry_id, + hole.max_entry_id + ); + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.inc(); + wal_holes + .lock() + .unwrap_or_else(PoisonError::into_inner) + .entry(entry.region_id) + .or_default() + .entry(object_seq) + .or_insert(hole); +} + fn encode_batch(header: Header, entries: Vec) -> Result { let records = entries .into_iter() @@ -2071,6 +2193,10 @@ mod tests { const PREFIX: &str = "wal/datanodes/1/epochs/2"; const WAIT: Duration = Duration::from_secs(30); + /// Held by every test that reads the process-wide skipped segments + /// counter, so that tests running in one process do not interleave + /// their increments. + static SKIPPED_SEGMENTS: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); fn memory_store() -> ObjectStore { ObjectStore::new(Memory::default()).unwrap() @@ -2093,6 +2219,16 @@ mod tests { } } + fn on_corruption( + on_corrupted_segment: CorruptedSegmentAction, + config: ObjectStoreWalConfig, + ) -> ObjectStoreWalConfig { + ObjectStoreWalConfig { + on_corrupted_segment, + ..config + } + } + /// Every append reaches the size limit, so it is persisted on its own. fn eager() -> ObjectStoreWalConfig { config(Duration::from_secs(3600), 1) @@ -2472,6 +2608,18 @@ mod tests { (trailer, footer) } + /// Asserts that recovery over `object_store` rejects the object at `path` + /// for `reason`, whatever a read does with a corrupted segment. + async fn assert_recovery_rejects(object_store: &ObjectStore, path: &str, reason: &str) { + for on_corrupted_segment in [CorruptedSegmentAction::Skip, CorruptedSegmentAction::Fail] { + let config = on_corruption(on_corrupted_segment, eager()); + let error = ObjectStoreLogStore::try_new(object_store.clone(), &config, 1, 2) + .await + .unwrap_err(); + assert_invalid_object(&error, path, reason); + } + } + fn assert_invalid_object(error: &Error, path: &str, reason: &str) { match error { Error::InvalidWalObject { @@ -2553,10 +2701,7 @@ mod tests { let path = object_path(&object_store, 1); corrupt_object(&object_store, &path, *corrupt).await; - let error = ObjectStoreLogStore::try_new(object_store, &eager(), 1, 2) - .await - .unwrap_err(); - assert_invalid_object(&error, &path, reason); + assert_recovery_rejects(&object_store, &path, reason).await; } } @@ -2575,14 +2720,12 @@ mod tests { .unwrap(); io.put_if_absent(5, encoded.bytes).await.unwrap(); - let error = ObjectStoreLogStore::try_new(object_store, &eager(), 1, 2) - .await - .unwrap_err(); - assert_invalid_object( - &error, + assert_recovery_rejects( + &object_store, &io.object_path(5), "header sequence 0 does not match key sequence 5", - ); + ) + .await; } /// Overwrites the byte range of footer entry `index` and refreshes the @@ -2635,10 +2778,7 @@ mod tests { }) .await; - let error = ObjectStoreLogStore::try_new(object_store.clone(), &eager(), 1, 2) - .await - .unwrap_err(); - assert_invalid_object(&error, &path, reason); + assert_recovery_rejects(&object_store, &path, reason).await; let io = ObjectStoreIo::new(object_store, PREFIX).unwrap(); let error = recover_by_decoding(&io).await.unwrap_err(); assert_invalid_object(&error, &path, reason); @@ -2872,9 +3012,11 @@ mod tests { let store = ObjectStoreLogStore { prefix: PREFIX.to_string(), ack_mode: AckMode::Durable, + on_corrupted_segment: CorruptedSegmentAction::Skip, io: Arc::new(ObjectStoreIo::new(memory_store(), PREFIX).unwrap()), catalog: Arc::default(), obsolete_entry_ids: ObsoleteEntryIds::default(), + wal_holes: WalHoles::default(), terminal_error: Arc::default(), stopped: Arc::new(AtomicBool::new(false)), command_tx, @@ -3267,11 +3409,12 @@ mod tests { store.stop().await.unwrap(); } - #[tokio::test] - async fn test_store_corrupted_segment_fails_the_read_that_decodes_it() { - let object_store = memory_store(); + /// Writes object 0 with one entry of region 1 and one of region 2 and + /// object 1 with an entry of region 2, then flips a byte in the segment of + /// region 2 in object 0. Returns the key of object 0 and its footer. + async fn write_corrupted_segment(object_store: &ObjectStore) -> (String, Vec) { put_records( - &object_store, + object_store, 0, &[ Record { @@ -3287,19 +3430,93 @@ mod tests { ], ) .await; - put_object(&object_store, 1, region(2), &[10]).await; - let path = object_path(&object_store, 0); + put_object(object_store, 1, region(2), &[10]).await; + let path = object_path(object_store, 0); let bytes = object_store.read(&path).await.unwrap().to_vec(); let footer = footer_of(&bytes).1; let corrupt = &footer[1]; - corrupt_object(&object_store, &path, |bytes| { + assert_eq!(region(2), corrupt.region_id); + corrupt_object(object_store, &path, |bytes| { bytes[(corrupt.segment_offset + corrupt.segment_len - 1) as usize] ^= 1; }) .await; + (path, footer) + } + + #[tokio::test] + async fn test_store_skips_a_corrupted_segment_and_records_a_hole() { + let _serialized = SKIPPED_SEGMENTS.lock().await; + let object_store = memory_store(); + let (path, footer) = write_corrupted_segment(&object_store).await; let (io, reads) = RecordingIo::over(object_store); - let store = ObjectStoreLogStore::open(io, &eager(), PREFIX.to_string()) - .await - .unwrap(); + let store = open_over(io, &eager()).await; + assert_eq!(CorruptedSegmentAction::Skip, store.on_corrupted_segment); + let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get(); + + // Recovery indexes the object; a read of the other region and one of + // the later object of the damaged region are unaffected. + assert_eq!( + expected_entries(region(1), &[(1, "a1")]), + read_entries(&store, region(1), 0).await + ); + assert_eq!( + expected_entries(region(2), &[(10, "e10")]), + read_entries(&store, region(2), 10).await + ); + assert!(store.wal_holes(&provider(region(2))).unwrap().is_empty()); + assert_eq!( + skipped, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() + ); + + // A read across the damaged segment fetches it twice, continues with + // the next object and records the hole. + reads.lock().unwrap().clear(); + assert_eq!( + expected_entries(region(2), &[(10, "e10")]), + read_entries(&store, region(2), 0).await + ); + let corrupt = (0, footer[1].segment_offset, footer[1].segment_len); + assert_eq!(corrupt, reads.lock().unwrap()[0]); + assert_eq!(corrupt, reads.lock().unwrap()[1]); + let hole = WalHole { + path, + object_seq: 0, + min_entry_id: 1, + max_entry_id: 1, + }; + assert_eq!( + vec![hole.clone()], + store.wal_holes(&provider(region(2))).unwrap() + ); + assert!(store.wal_holes(&provider(region(1))).unwrap().is_empty()); + assert_eq!( + skipped + 1, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() + ); + + // Every read across the segment skips it again; the hole is one. + assert_eq!( + expected_entries(region(2), &[(10, "e10")]), + read_entries(&store, region(2), 0).await + ); + assert_eq!(vec![hole], store.wal_holes(&provider(region(2))).unwrap()); + assert_eq!( + skipped + 2, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() + ); + store.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_store_corrupted_segment_fails_the_read_that_decodes_it() { + let _serialized = SKIPPED_SEGMENTS.lock().await; + let object_store = memory_store(); + let (path, footer) = write_corrupted_segment(&object_store).await; + let (io, reads) = RecordingIo::over(object_store); + let config = on_corruption(CorruptedSegmentAction::Fail, eager()); + let store = open_over(io, &config).await; + let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get(); assert_eq!(1, latest(&store, region(1))); assert_eq!(10, latest(&store, region(2))); reads.lock().unwrap().clear(); @@ -3328,13 +3545,78 @@ mod tests { &path, &format!("segment of region {} checksum mismatch", region(2)), ); + let corrupt = (0, footer[1].segment_offset, footer[1].segment_len); + assert_eq!(vec![corrupt, corrupt], *reads.lock().unwrap()); + assert!(store.wal_holes(&provider(region(2))).unwrap().is_empty()); assert_eq!( - vec![(0, corrupt.segment_offset, corrupt.segment_len)], - *reads.lock().unwrap() + skipped, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() ); store.stop().await.unwrap(); } + #[tokio::test] + async fn test_store_reads_a_segment_whose_first_download_was_damaged() { + let _serialized = SKIPPED_SEGMENTS.lock().await; + for on_corrupted_segment in [CorruptedSegmentAction::Skip, CorruptedSegmentAction::Fail] { + let object_store = memory_store(); + put_object(&object_store, 0, region(1), &[1, 2]).await; + let (io, reads) = RecordingIo::over(object_store); + let config = on_corruption(on_corrupted_segment, eager()); + let store = open_over(io.clone(), &config).await; + let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get(); + + // The first download of the segment is damaged, the second is + // not: the read sees the entries and records no hole. + reads.lock().unwrap().clear(); + io.damage_next_range_read.store(true, Ordering::Relaxed); + assert_eq!( + expected_entries(region(1), &[(1, "e1"), (2, "e2")]), + read_entries(&store, region(1), 0).await + ); + assert_eq!(2, reads.lock().unwrap().len()); + assert!(store.wal_holes(&provider(region(1))).unwrap().is_empty()); + assert_eq!( + skipped, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() + ); + store.stop().await.unwrap(); + } + } + + #[tokio::test] + async fn test_store_failed_segment_fetch_fails_the_read() { + let _serialized = SKIPPED_SEGMENTS.lock().await; + for on_corrupted_segment in [CorruptedSegmentAction::Skip, CorruptedSegmentAction::Fail] { + let object_store = memory_store(); + put_object(&object_store, 0, region(1), &[1]).await; + let path = object_path(&object_store, 0); + let (io, _) = RecordingIo::over(object_store.clone()); + let config = on_corruption(on_corrupted_segment, eager()); + let store = open_over(io, &config).await; + let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get(); + + object_store.delete(&path).await.unwrap(); + let error = store + .read(&provider(region(1)), 0, None) + .await + .unwrap() + .try_collect::>() + .await + .unwrap_err(); + assert!( + matches!(&error, Error::WalObjectStore { path: actual, .. } if *actual == path), + "unexpected error: {error:?}" + ); + assert!(store.wal_holes(&provider(region(1))).unwrap().is_empty()); + assert_eq!( + skipped, + METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get() + ); + store.stop().await.unwrap(); + } + } + #[tokio::test] async fn test_store_assigns_ids_from_the_object_sequence() { let store = open(memory_store(), &manual()).await; @@ -5540,13 +5822,14 @@ mod tests { /// Object access that records every range read as (sequence, offset, /// length) and, on request, reports the next conditional create as failed - /// after it wrote the object, or fails it with a given error before it - /// writes. + /// after it wrote the object, fails it with a given error before it + /// writes, or damages the bytes the next range read returns. struct RecordingIo { inner: ObjectStoreIo, reads: RangeReads, fail_after_next_put: AtomicBool, fail_next_put: Mutex>, + damage_next_range_read: AtomicBool, } impl RecordingIo { @@ -5557,6 +5840,7 @@ mod tests { reads: reads.clone(), fail_after_next_put: AtomicBool::new(false), fail_next_put: Mutex::new(None), + damage_next_range_read: AtomicBool::new(false), }; (Arc::new(io), reads) } @@ -5601,7 +5885,15 @@ mod tests { async fn get_range(&self, object_seq: u64, offset: u64, len: u64) -> Result { self.reads.lock().unwrap().push((object_seq, offset, len)); - self.inner.get_range(object_seq, offset, len).await + let bytes = self.inner.get_range(object_seq, offset, len).await?; + if self.damage_next_range_read.swap(false, Ordering::Relaxed) { + let mut damaged = bytes.to_vec(); + if let Some(last) = damaged.last_mut() { + *last ^= 1; + } + return Ok(Bytes::from(damaged)); + } + Ok(bytes) } async fn list(&self) -> Result> { diff --git a/src/mito2/src/engine/object_store_wal_recovery_test.rs b/src/mito2/src/engine/object_store_wal_recovery_test.rs index 6017f745d2e..917ad614a07 100644 --- a/src/mito2/src/engine/object_store_wal_recovery_test.rs +++ b/src/mito2/src/engine/object_store_wal_recovery_test.rs @@ -30,10 +30,12 @@ use common_base::readable_size::ReadableSize; use common_error::ext::{BoxedError, ErrorExt}; use common_error::status_code::StatusCode; use common_recordbatch::RecordBatches; -use common_wal::config::object_store::{AckMode, ObjectStoreWalConfig, STANDALONE_GENERATION}; +use common_wal::config::object_store::{ + AckMode, CorruptedSegmentAction, ObjectStoreWalConfig, STANDALONE_GENERATION, +}; use common_wal::options::{ObjectStoreWalOptions, WAL_OPTIONS_KEY, WalOptions}; use log_store::ObjectStoreLogStore; -use log_store::object_store_wal::entry_id; +use log_store::object_store_wal::{WalHole, entry_id}; use object_store::ObjectStore; use object_store::services::Memory; use rstest::rstest; @@ -412,6 +414,111 @@ async fn test_reopen_after_crash_replays_durable_entries_once(#[case] ack_mode: assert_eq!(3, engine.get_region_statistic(REGION_A).unwrap().num_rows); } +#[rstest] +#[case(CorruptedSegmentAction::Skip)] +#[case(CorruptedSegmentAction::Fail)] +#[tokio::test] +async fn test_reopen_with_a_corrupted_segment( + #[case] on_corrupted_segment: CorruptedSegmentAction, +) { + let mut env = TestEnv::with_prefix("object-store-wal-corrupted-segment").await; + let object_store = memory_store(); + let store = open_store(&object_store, AckMode::Durable).await; + let engine = new_engine(&mut env, store.clone()).await; + let (table_dir_a, schema_a) = create_region(&engine, REGION_A, &[]).await; + let (table_dir_b, schema_b) = create_region(&engine, REGION_B, &[]).await; + + // Objects 1 and 2 each hold one entry of both regions; nothing is flushed. + let mut writer = SealedWriter::new(&store); + writer + .put_and_seal( + &engine, + vec![ + (REGION_A, rows(&schema_a, 0, 2)), + (REGION_B, rows(&schema_b, 0, 3)), + ], + ) + .await; + writer + .put_and_seal( + &engine, + vec![ + (REGION_A, rows(&schema_a, 2, 4)), + (REGION_B, rows(&schema_b, 3, 5)), + ], + ) + .await; + let rows_b = scan_rows(&engine, REGION_B).await; + engine.stop().await.unwrap(); + + // The segment of region A in object 1 is damaged. + let (path, segment) = store + .segment_location(&provider(REGION_A), 1) + .unwrap() + .unwrap(); + let mut bytes = object_store.read(&path).await.unwrap().to_vec(); + bytes[segment.end as usize - 1] ^= 1; + object_store.write(&path, bytes).await.unwrap(); + drop(engine); + drop(writer); + drop(store); + + let config = ObjectStoreWalConfig { + on_corrupted_segment, + ..store_config(AckMode::Durable) + }; + let store = + ObjectStoreLogStore::try_new(object_store.clone(), &config, 0, STANDALONE_GENERATION) + .await + .unwrap(); + let engine = new_engine(&mut env, store.clone()).await; + + // Region B replays both of its entries whatever the damage does to A. + open_region(&engine, REGION_B, &table_dir_b, &[]) + .await + .unwrap(); + assert_eq!(rows_b, scan_rows(&engine, REGION_B).await); + assert_eq!(5, engine.get_region_statistic(REGION_B).unwrap().num_rows); + assert!(store.wal_holes(&provider(REGION_B)).unwrap().is_empty()); + + let opened = open_region(&engine, REGION_A, &table_dir_a, &[]).await; + match on_corrupted_segment { + // Region A replays the entry of object 2 alone and records the + // segment of object 1 as a hole. + CorruptedSegmentAction::Skip => { + opened.unwrap(); + assert_eq!( + RegionRecoveryState { + last_entry_id: entry_id(2, 1), + memtable_rows: 2, + ..EMPTY_RECOVERY_STATE + }, + region_recovery_state(&engine, REGION_A).await + ); + assert_eq!(2, engine.get_region_statistic(REGION_A).unwrap().num_rows); + assert_eq!( + vec![WalHole { + path, + object_seq: 1, + min_entry_id: entry_id(1, 1), + max_entry_id: entry_id(1, 1), + }], + store.wal_holes(&provider(REGION_A)).unwrap() + ); + } + // The read fails, so the region does not open. + CorruptedSegmentAction::Fail => { + let message = opened.unwrap_err().output_msg(); + assert!( + message.contains("checksum mismatch"), + "unexpected error: {message}" + ); + assert!(!engine.is_region_exists(REGION_A)); + assert!(store.wal_holes(&provider(REGION_A)).unwrap().is_empty()); + } + } +} + #[tokio::test] async fn test_open_region_rejects_mismatched_wal_prefix() { let mut env = TestEnv::with_prefix("object-store-wal-prefix-mismatch").await; diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 3fa1d394489..5a36d495df8 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -39,14 +39,13 @@ use common_meta::procedure_executor::LocalProcedureExecutor; use common_meta::region_keeper::MemoryRegionKeeper; use common_meta::region_registry::LeaderRegionRegistry; use common_meta::sequence::SequenceBuilder; -use common_meta::wal_provider::build_wal_provider; use common_procedure::ProcedureManagerRef; use common_procedure::local::EventRecorderHandle; use common_procedure::options::ProcedureConfig; use common_telemetry::logging::SlowQueryOptions; use common_test_util::find_workspace_path; -use common_wal::config::{DatanodeWalConfig, MetasrvWalConfig}; -use datanode::datanode::DatanodeBuilder; +use common_wal::config::DatanodeWalConfig; +use datanode::datanode::{Datanode, DatanodeBuilder}; use flow::{FlownodeBuilder, FrontendClient, GrpcQueryHandlerWithBoxedError}; use frontend::frontend::Frontend; use frontend::instance::Instance; @@ -63,6 +62,8 @@ use crate::test_util::{self, StorageType, TestGuard, create_tmp_dir_and_datanode pub struct GreptimeDbStandalone { /// Storage engine for assertions across protocol and storage boundaries. pub mito_engine: mito2::engine::MitoEngine, + /// Owns the region server and the log store, so a test can shut them down. + pub datanode: Datanode, pub frontend: Arc, pub opts: StandaloneOptions, pub guard: TestGuard, @@ -81,7 +82,6 @@ impl GreptimeDbStandalone { pub struct GreptimeDbStandaloneBuilder { instance_name: String, datanode_wal_config: DatanodeWalConfig, - metasrv_wal_config: MetasrvWalConfig, store_providers: Option>, default_store: Option, plugin: Option, @@ -102,7 +102,6 @@ impl GreptimeDbStandaloneBuilder { plugin: None, default_store: None, datanode_wal_config: DatanodeWalConfig::default(), - metasrv_wal_config: MetasrvWalConfig::default(), // Enable slow query log with 1s threshold by default for integration tests. slow_query_options: SlowQueryOptions { enable: true, @@ -199,12 +198,6 @@ impl GreptimeDbStandaloneBuilder { self } - #[must_use] - pub fn with_metasrv_wal_config(mut self, metasrv_wal_config: MetasrvWalConfig) -> Self { - self.metasrv_wal_config = metasrv_wal_config; - self - } - pub async fn build_with( &self, kv_backend: KvBackendRef, @@ -281,10 +274,13 @@ impl GreptimeDbStandaloneBuilder { .step(10) .build(), ); - let kafka_options = opts.wal.clone().try_into().unwrap(); - let wal_provider = build_wal_provider(&kafka_options, kv_backend.clone()) - .await - .unwrap(); + let wal_provider = cmd::standalone::build_standalone_wal_provider( + &opts.wal, + opts.datanode_options().node_id, + kv_backend.clone(), + ) + .await + .unwrap(); let wal_provider = Arc::new(wal_provider); let table_metadata_allocator = Arc::new(TableMetadataAllocator::new( table_id_allocator, @@ -363,6 +359,7 @@ impl GreptimeDbStandaloneBuilder { GreptimeDbStandalone { mito_engine: datanode.region_server().mito_engine().unwrap(), + datanode, frontend: Arc::new(frontend), opts, guard, @@ -399,7 +396,7 @@ impl GreptimeDbStandaloneBuilder { storage: opts.storage, procedure: procedure_config, metadata_store: kv_backend_config, - wal: self.metasrv_wal_config.clone().into(), + wal: opts.wal, grpc: GrpcOptions::default().with_server_addr("127.0.0.1:4001"), slow_query: self.slow_query_options.clone(), event_recorder: self.event_recorder_options.clone(), diff --git a/tests-integration/src/tests/test_util.rs b/tests-integration/src/tests/test_util.rs index 5ddcd0748a9..7c99e89ee08 100644 --- a/tests-integration/src/tests/test_util.rs +++ b/tests-integration/src/tests/test_util.rs @@ -263,15 +263,8 @@ pub(crate) async fn standalone_with_kafka_wal() -> Option>(); let test_name = uuid::Uuid::new_v4().to_string(); - let builder = GreptimeDbStandaloneBuilder::new(&test_name) - .with_datanode_wal_config(DatanodeWalConfig::Kafka(DatanodeKafkaConfig { - connection: KafkaConnectionConfig { - broker_endpoints: endpoints.clone(), - ..Default::default() - }, - ..Default::default() - })) - .with_metasrv_wal_config(MetasrvWalConfig::Kafka(MetasrvKafkaConfig { + let builder = GreptimeDbStandaloneBuilder::new(&test_name).with_datanode_wal_config( + DatanodeWalConfig::Kafka(DatanodeKafkaConfig { connection: KafkaConnectionConfig { broker_endpoints: endpoints, ..Default::default() @@ -282,7 +275,8 @@ pub(crate) async fn standalone_with_kafka_wal() -> Option Output { + SqlQueryHandler::do_query(instance, sql, QueryContext::arc()) + .await + .remove(0) + .unwrap() +} + +async fn query_rows(standalone: &GreptimeDbStandalone) -> String { + execute_sql(standalone.fe_instance(), QUERY) + .await + .data + .pretty_print() + .await +} + +/// Returns the number of WAL objects under [`WAL_ROOT`] of the default store. +async fn count_wal_objects(standalone: &GreptimeDbStandalone) -> usize { + let store = match &standalone.opts.storage.store { + ObjectStoreConfig::File(_) => { + ObjectStore::new(Fs::default().root(&standalone.opts.storage.data_home)).unwrap() + } + ObjectStoreConfig::S3(s3) => ObjectStore::new(S3::from(&s3.connection)).unwrap(), + other => panic!("unexpected store {other:?}"), + }; + store + .list_with(&format!("{WAL_ROOT}/")) + .recursive(true) + .await + .unwrap() + .into_iter() + .filter(|entry| entry.metadata().is_file()) + .count() +} + +/// Stops the procedure manager, the region server and the WAL of the +/// instance. +async fn shutdown(standalone: &mut GreptimeDbStandalone) { + standalone.procedure_manager.stop().await.unwrap(); + standalone.datanode.shutdown().await.unwrap(); +} + +/// Shuts the instance down and builds it again on its metadata, data home and +/// object store, like a process restart. +async fn restart( + builder: &GreptimeDbStandaloneBuilder, + mut standalone: GreptimeDbStandalone, +) -> GreptimeDbStandalone { + shutdown(&mut standalone).await; + let GreptimeDbStandalone { + opts, + guard, + kv_backend, + .. + } = standalone; + + let (procedure_manager, event_recorder_handle) = + standalone::build_procedure_manager(kv_backend.clone(), ProcedureConfig::default()); + builder + .build_with( + kv_backend, + guard, + opts, + procedure_manager, + event_recorder_handle, + true, + ) + .await +} + +/// Writes batches that land in separate WAL objects, restarts before and +/// after a flush, and checks that the rows survive both restarts. +async fn run_restarts_on_object_store_wal(store_type: StorageType) { + common_telemetry::init_default_ut_logging(); + let wal_config = ObjectStoreWalConfig { + prefix: WAL_ROOT.to_string(), + ..Default::default() + }; + let builder = GreptimeDbStandaloneBuilder::new("object_store_wal") + .with_default_store_type(store_type) + .with_datanode_wal_config(DatanodeWalConfig::ObjectStore(wal_config)); + let standalone = builder.build().await; + + execute_sql( + standalone.fe_instance(), + r#" + CREATE TABLE cpu ( + hostname STRING PRIMARY KEY, + usage_user DOUBLE, + ts TIMESTAMP TIME INDEX + ) + "#, + ) + .await; + // Every insert returns once its entries are durable, so the batches land + // in separate WAL objects after the object that started the epoch. + for batch in 0..BATCHES { + let values = (0..ROWS_PER_BATCH) + .map(|row| { + let i = batch * ROWS_PER_BATCH + row; + format!( + "('host_{i}', {i}.0, {})", + 1_686_567_600_000 + i as i64 * 1000 + ) + }) + .collect::>() + .join(", "); + execute_sql( + standalone.fe_instance(), + &format!("INSERT INTO cpu VALUES {values}"), + ) + .await; + } + let expected = query_rows(&standalone).await; + assert_eq!(BATCHES * ROWS_PER_BATCH + 4, expected.lines().count()); + let objects = count_wal_objects(&standalone).await; + assert!( + objects > BATCHES, + "expected more than {BATCHES} WAL objects" + ); + // No Raft Engine log store is created as a fallback. + assert!( + !Path::new(&standalone.opts.storage.data_home) + .join("wal") + .exists() + ); + + // Nothing was flushed, so the rows come back from the WAL alone; the + // reopened store starts its epoch with another object. + let standalone = restart(&builder, standalone).await; + assert_eq!(expected, query_rows(&standalone).await); + assert!(count_wal_objects(&standalone).await > objects); + + // The flushed rows come back from the SST files. + execute_sql(standalone.fe_instance(), "ADMIN FLUSH_TABLE('cpu')").await; + let mut standalone = restart(&builder, standalone).await; + assert_eq!(expected, query_rows(&standalone).await); + shutdown(&mut standalone).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_standalone_object_store_wal_survives_restarts_on_file() { + run_restarts_on_object_store_wal(StorageType::File).await; +} + +/// Runs against the S3 bucket of the `GT_S3_*` environment variables, so it +/// is skipped unless they are set. +#[tokio::test(flavor = "multi_thread")] +async fn test_standalone_object_store_wal_survives_restarts_on_s3() { + if !StorageType::S3.test_on() { + return; + } + run_restarts_on_object_store_wal(StorageType::S3).await; +}