From 13c69cdaee7b0883207d8ab774e94581cc015f53 Mon Sep 17 00:00:00 2001 From: discord9 Date: Mon, 14 Sep 2026 07:30:39 +0000 Subject: [PATCH] perf(gc): pack file reference exchange (#9009) * perf(gc): pack file reference exchange Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(gc): address packed reference review feedback Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(gc): stop without retry when maintenance is enabled Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(meta): avoid logging malformed mailbox payloads Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/common/meta/src/error.rs | 16 + src/common/meta/src/heartbeat/handler.rs | 99 ++++- src/common/meta/src/instruction.rs | 348 +++++++++++++++- src/datanode/src/heartbeat.rs | 17 +- src/datanode/src/heartbeat/handler.rs | 14 +- .../src/heartbeat/handler/file_ref.rs | 56 ++- .../src/heartbeat/handler/gc_worker.rs | 35 +- src/meta-srv/src/error.rs | 4 +- src/meta-srv/src/event/gc.rs | 3 + src/meta-srv/src/gc/ctx.rs | 6 + src/meta-srv/src/gc/procedure.rs | 387 +++++++++++++----- src/meta-srv/src/handler.rs | 49 ++- src/meta-srv/src/key.rs | 4 +- src/meta-srv/src/metasrv/builder.rs | 1 + src/meta-srv/src/service/heartbeat.rs | 15 +- src/store-api/src/storage/file.rs | 5 + tests-integration/src/tests/gc.rs | 1 + tests-integration/src/tests/gc/repart.rs | 3 + tests-integration/tests/repartition.rs | 1 + 19 files changed, 942 insertions(+), 122 deletions(-) diff --git a/src/common/meta/src/error.rs b/src/common/meta/src/error.rs index 3bd1475cb4..3f4868ffba 100644 --- a/src/common/meta/src/error.rs +++ b/src/common/meta/src/error.rs @@ -269,6 +269,20 @@ pub enum Error { error: prost::DecodeError, }, + #[snafu(display("Failed to decode packed file references"))] + DecodePackedFileRefs { + #[snafu(implicit)] + location: Location, + #[snafu(source)] + error: base64::DecodeError, + }, + + #[snafu(display("Invalid packed file references framing"))] + InvalidPackedFileRefs { + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Failed to encode object into json"))] EncodeJson { #[snafu(implicit)] @@ -1212,6 +1226,8 @@ impl ErrorExt for Error { | PayloadNotExist { .. } | ConvertRawKey { .. } | DecodeProto { .. } + | DecodePackedFileRefs { .. } + | InvalidPackedFileRefs { .. } | BuildTableMeta { .. } | TableRouteNotFound { .. } | TableRepartNotFound { .. } diff --git a/src/common/meta/src/heartbeat/handler.rs b/src/common/meta/src/heartbeat/handler.rs index ecc735083b..fb5f16653a 100644 --- a/src/common/meta/src/heartbeat/handler.rs +++ b/src/common/meta/src/heartbeat/handler.rs @@ -94,7 +94,21 @@ impl HeartbeatResponseHandlerExecutor for HandlerGroupExecutor { Ok(HandleControl::Done) => break, Ok(HandleControl::Continue) => {} Err(e) => { - error!(e;"Error while handling: {:?}", ctx.response); + let mailbox_message_id = ctx + .response + .mailbox_message + .as_ref() + .map(|message| message.id); + let json_payload_len = + ctx.response.mailbox_message.as_ref().and_then(|message| { + message.payload.as_ref().map(|payload| match payload { + api::v1::meta::mailbox_message::Payload::Json(json) => json.len(), + }) + }); + error!( + %e; + "Error while handling heartbeat response: mailbox_message_id={mailbox_message_id:?}, json_payload_len={json_payload_len:?}" + ); break; } } @@ -102,3 +116,86 @@ impl HeartbeatResponseHandlerExecutor for HandlerGroupExecutor { Ok(()) } } + +#[cfg(test)] +mod error_log_tests { + use std::fmt::Debug; + use std::sync::{Arc, Mutex}; + + use api::v1::meta::mailbox_message::Payload; + use api::v1::meta::{HeartbeatResponse, MailboxMessage}; + use common_telemetry::tracing::field::{Field, Visit}; + use common_telemetry::tracing::{Event, Subscriber}; + use common_telemetry::tracing_subscriber::layer::{Context, SubscriberExt}; + use common_telemetry::{tracing, tracing_subscriber}; + + use super::parse_mailbox_message::ParseMailboxMessageHandler; + use super::*; + use crate::heartbeat::mailbox::HeartbeatMailbox; + + #[derive(Clone, Default)] + struct LogCapture(Arc>>); + + impl tracing_subscriber::Layer for LogCapture + where + S: Subscriber, + { + fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { + let mut visitor = FieldVisitor::default(); + event.record(&mut visitor); + self.0.lock().unwrap().push(visitor.fields.join(", ")); + } + } + + #[derive(Default)] + struct FieldVisitor { + fields: Vec, + } + + impl Visit for FieldVisitor { + fn record_debug(&mut self, field: &Field, value: &dyn Debug) { + self.fields.push(format!("{}={value:?}", field.name())); + } + + fn record_str(&mut self, field: &Field, value: &str) { + self.fields.push(format!("{}={value:?}", field.name())); + } + } + + #[tokio::test(flavor = "current_thread")] + async fn test_handler_error_log_excludes_mailbox_payload() { + let payload_sentinel = "MALFORMED_PACKED_PAYLOAD_MUST_NOT_BE_LOGGED".repeat(256); + let payload = format!( + r#"{{"PackedGcRegions":{{"regions":[],"packed_file_refs_manifest":"{}"}}"#, + payload_sentinel + ); + let payload_len = payload.len(); + let capture = LogCapture::default(); + let subscriber = tracing_subscriber::registry().with(capture.clone()); + let _guard = tracing::subscriber::set_default(subscriber); + let (mailbox_tx, _) = tokio::sync::mpsc::channel(1); + let ctx = HeartbeatResponseHandlerContext::new( + Arc::new(HeartbeatMailbox::new(mailbox_tx)), + HeartbeatResponse { + mailbox_message: Some(MailboxMessage { + id: 42, + subject: "unsafe subject".to_string(), + to: "unsafe recipient".to_string(), + from: "unsafe sender".to_string(), + payload: Some(Payload::Json(payload)), + ..Default::default() + }), + ..Default::default() + }, + ); + let executor = HandlerGroupExecutor::new(vec![Arc::new(ParseMailboxMessageHandler)]); + + executor.handle(ctx).await.unwrap(); + + let logs = capture.0.lock().unwrap().join("\n"); + assert!(logs.contains("Error while handling heartbeat response")); + assert!(logs.contains("mailbox_message_id=Some(42)")); + assert!(logs.contains(&format!("json_payload_len=Some({payload_len})"))); + assert!(!logs.contains(&payload_sentinel)); + } +} diff --git a/src/common/meta/src/instruction.rs b/src/common/meta/src/instruction.rs index ba4f2e32ab..60ebe14e37 100644 --- a/src/common/meta/src/instruction.rs +++ b/src/common/meta/src/instruction.rs @@ -16,16 +16,19 @@ use std::collections::{HashMap, HashSet}; use std::fmt::{Display, Formatter}; use std::time::Duration; +use base64::Engine; use common_error::ext::{ErrorExt, RetryHint}; use common_error::status_code::StatusCode; use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use snafu::ResultExt as _; use store_api::region_engine::SyncRegionFromRequest; use store_api::region_request::{RegionFlushReason, RegionRequirements}; -use store_api::storage::{FileRefsManifest, GcReport, RegionId, RegionNumber}; +use store_api::storage::{FileId, FileRef, FileRefsManifest, GcReport, RegionId, RegionNumber}; use strum::Display; use table::metadata::TableId; use table::table_name::TableName; +use crate::error::{DecodePackedFileRefsSnafu, InvalidPackedFileRefsSnafu}; use crate::flow_name::FlowName; use crate::key::schema_name::SchemaName; use crate::key::{FlowId, FlowPartitionId}; @@ -579,6 +582,24 @@ impl Display for GetFileRefs { } } +/// Instruction to get file references using the packed manifest format. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct GetPackedFileRefs { + /// List of region IDs to get file references from active FileHandles (in-memory). + pub query_regions: Vec, + /// Mapping from the src region IDs (whose file references to look for) to + /// the dst region IDs (where to read the manifests). + /// Key: The source region IDs (where files originally came from). + /// Value: The set of destination region IDs (whose manifests need to be read). + pub related_regions: HashMap>, +} + +impl Display for GetPackedFileRefs { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "GetPackedFileRefs(region_ids={:?})", self.query_regions) + } +} + /// Instruction to trigger garbage collection for a region. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct GcRegions { @@ -590,6 +611,147 @@ pub struct GcRegions { pub full_file_listing: bool, } +/// Instruction to trigger garbage collection with a packed file-reference manifest. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct PackedGcRegions { + /// The region ID to perform GC on. + pub regions: Vec, + /// Packed manifest used by the packed-only GC protocol. + pub packed_file_refs_manifest: PackedFileRefsManifest, + /// Whether to perform a full file listing to find orphan files. + pub full_file_listing: bool, +} + +impl Display for PackedGcRegions { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + let file_refs_count = self.packed_file_refs_manifest.file_refs.len(); + write!( + f, + "PackedGcRegions(regions={:?}, file_refs_count={}, full_file_listing={})", + self.regions, file_refs_count, self.full_file_listing + ) + } +} + +/// A packed, JSON-compatible encoding of a file reference manifest. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)] +pub struct PackedFileRefsManifest { + /// Required packed file references for each region. + pub file_refs: HashMap, + #[serde(default)] + pub manifest_version: HashMap, + #[serde(default)] + pub cross_region_refs: HashMap>, +} + +/// Packed file references for one region, encoded with STANDARD Base64. +/// +/// `files` is required and contains 16-byte UUID records for references whose +/// `index_version` is `None`. `indexed` is required and contains 24-byte records +/// for references whose `index_version` is `Some`: a 16-byte UUID followed by +/// an 8-byte big-endian `u64` version. Decoding fails when either field is not +/// valid STANDARD Base64 or its decoded bytes do not align to the corresponding +/// record width. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)] +pub struct PackedRegionFileRefs { + /// Required STANDARD Base64-encoded 16-byte UUID records. + pub files: String, + /// Required STANDARD Base64-encoded 16-byte UUID and 8-byte version records. + pub indexed: String, +} + +impl PackedFileRefsManifest { + /// Packs a file reference manifest using required [`PackedRegionFileRefs`] + /// STANDARD Base64 fields. + /// + /// References whose `index_version` is `None` are encoded as 16-byte UUID + /// `files` records; references whose `index_version` is `Some` are encoded + /// as 24-byte `indexed` records: a 16-byte UUID followed by an 8-byte + /// big-endian `u64` version. + pub fn from_manifest(manifest: &FileRefsManifest) -> Self { + Self { + file_refs: manifest + .file_refs + .iter() + .map(|(region, refs)| (*region, PackedRegionFileRefs::from_refs(refs))) + .collect(), + manifest_version: manifest.manifest_version.clone(), + cross_region_refs: manifest.cross_region_refs.clone(), + } + } + + /// Decodes this packed representation into a file reference manifest. + /// + /// Returns a typed error for invalid STANDARD Base64 or malformed record + /// framing without retaining encoded payloads. + pub fn into_manifest(self) -> crate::error::Result { + let mut file_refs = HashMap::new(); + for (region, encoded) in self.file_refs { + file_refs.insert(region, encoded.into_refs(region)?); + } + Ok(FileRefsManifest { + file_refs, + manifest_version: self.manifest_version, + cross_region_refs: self.cross_region_refs, + }) + } +} + +impl PackedRegionFileRefs { + /// Packs file references as the documented STANDARD Base64 record streams. + pub fn from_refs(refs: &HashSet) -> Self { + let engine = base64::engine::general_purpose::STANDARD; + let mut files = Vec::new(); + let mut indexed = Vec::new(); + for file_ref in refs { + match file_ref.index_version { + None => files.extend_from_slice(file_ref.file_id.as_bytes()), + Some(version) => { + indexed.extend_from_slice(file_ref.file_id.as_bytes()); + indexed.extend_from_slice(&version.to_be_bytes()); + } + } + } + Self { + files: engine.encode(files), + indexed: engine.encode(indexed), + } + } + + /// Decodes the documented STANDARD Base64 record streams for `region`. + fn into_refs(self, region: RegionId) -> crate::error::Result> { + let engine = base64::engine::general_purpose::STANDARD; + let files = engine + .decode(self.files) + .context(DecodePackedFileRefsSnafu)?; + let indexed = engine + .decode(self.indexed) + .context(DecodePackedFileRefsSnafu)?; + if files.len() % 16 != 0 || indexed.len() % 24 != 0 { + return InvalidPackedFileRefsSnafu.fail(); + } + + let mut refs = HashSet::new(); + for bytes in files.chunks_exact(16) { + let mut id = [0; 16]; + id.copy_from_slice(bytes); + refs.insert(FileRef::new(region, FileId::from_bytes(id), None)); + } + for bytes in indexed.chunks_exact(24) { + let mut id = [0; 16]; + id.copy_from_slice(&bytes[..16]); + let mut version = [0; 8]; + version.copy_from_slice(&bytes[16..]); + refs.insert(FileRef::new( + region, + FileId::from_bytes(id), + Some(u64::from_be_bytes(version)), + )); + } + Ok(refs) + } +} + impl Display for GcRegions { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { write!( @@ -625,6 +787,26 @@ impl Display for GetFileRefsReply { } } +/// Reply for GetPackedFileRefs instruction. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct GetPackedFileRefsReply { + pub packed_file_refs_manifest: PackedFileRefsManifest, + pub success: bool, + pub error: Option, +} + +impl Display for GetPackedFileRefsReply { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!( + f, + "GetPackedFileRefsReply(success={}, file_refs_count={}, error={:?})", + self.success, + self.packed_file_refs_manifest.file_refs.len(), + self.error + ) + } +} + /// Reply for GC instruction. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct GcRegionsReply { @@ -827,8 +1009,12 @@ pub enum Instruction { FlushRegions(FlushRegions), /// Gets file references for regions. GetFileRefs(GetFileRefs), + /// Gets file references using the packed manifest format. + GetPackedFileRefs(GetPackedFileRefs), /// Triggers garbage collection for a region. GcRegions(GcRegions), + /// Triggers garbage collection with a packed file-reference manifest. + PackedGcRegions(PackedGcRegions), /// Temporary suspend serving reads or writes Suspend, /// Makes regions enter staging state. @@ -1111,6 +1297,7 @@ pub enum InstructionReply { DowngradeRegions(DowngradeRegionsReply), FlushRegions(FlushRegionReply), GetFileRefs(GetFileRefsReply), + GetPackedFileRefs(GetPackedFileRefsReply), GcRegions(GcRegionsReply), EnterStagingRegions(EnterStagingRegionsReply), SyncRegions(SyncRegionsReply), @@ -1131,6 +1318,9 @@ impl Display for InstructionReply { } Self::FlushRegions(reply) => write!(f, "InstructionReply::FlushRegions({})", reply), Self::GetFileRefs(reply) => write!(f, "InstructionReply::GetFileRefs({})", reply), + Self::GetPackedFileRefs(reply) => { + write!(f, "InstructionReply::GetPackedFileRefs({})", reply) + } Self::GcRegions(reply) => write!(f, "InstructionReply::GcRegion({})", reply), Self::EnterStagingRegions(reply) => { write!( @@ -1221,6 +1411,7 @@ impl InstructionReply { #[cfg(test)] mod tests { use std::collections::HashSet; + use std::error::Error as _; use common_error::mock::MockError; use common_wal::options::WalOptions; @@ -1771,6 +1962,161 @@ mod tests { } } + #[test] + fn test_legacy_gc_file_refs_compatibility() { + #[derive(Debug, Deserialize)] + struct LegacyGetFileRefs { + query_regions: Vec, + related_regions: HashMap>, + } + + #[derive(Debug, Deserialize)] + struct LegacyGetFileRefsReply { + file_refs_manifest: FileRefsManifest, + success: bool, + error: Option, + } + + #[derive(Debug, Deserialize)] + enum LegacyInstruction { + GetFileRefs(LegacyGetFileRefs), + } + + let get_file_refs = Instruction::GetFileRefs(GetFileRefs { + query_regions: vec![RegionId::new(7, 3)], + related_regions: HashMap::new(), + }); + let get_file_refs_json = serde_json::to_string(&get_file_refs).unwrap(); + let legacy_get_file_refs: LegacyInstruction = + serde_json::from_str(&get_file_refs_json).unwrap(); + let LegacyInstruction::GetFileRefs(legacy_get_file_refs) = legacy_get_file_refs; + assert_eq!(legacy_get_file_refs.query_regions.len(), 1); + assert!(legacy_get_file_refs.related_regions.is_empty()); + + let reply = GetFileRefsReply { + file_refs_manifest: FileRefsManifest::default(), + success: true, + error: None, + }; + let reply_json = serde_json::to_string(&reply).unwrap(); + let legacy_reply: LegacyGetFileRefsReply = serde_json::from_str(&reply_json).unwrap(); + let current_reply: GetFileRefsReply = serde_json::from_str(&reply_json).unwrap(); + assert!(legacy_reply.success); + assert!(legacy_reply.error.is_none()); + assert!(legacy_reply.file_refs_manifest.file_refs.is_empty()); + assert_eq!(current_reply, reply); + + let packed = Instruction::PackedGcRegions(PackedGcRegions { + regions: vec![], + packed_file_refs_manifest: PackedFileRefsManifest::default(), + full_file_listing: false, + }); + let packed_json = serde_json::to_string(&packed).unwrap(); + assert!(serde_json::from_str::(&packed_json).is_err()); + } + + #[test] + fn test_packed_gc_regions_round_trip_is_distinct() { + let instruction = Instruction::PackedGcRegions(PackedGcRegions { + regions: vec![RegionId::new(7, 3)], + packed_file_refs_manifest: PackedFileRefsManifest::default(), + full_file_listing: true, + }); + let serialized = serde_json::to_string(&instruction).unwrap(); + assert!(serialized.contains("PackedGcRegions")); + assert_eq!( + serde_json::from_str::(&serialized).unwrap(), + instruction + ); + } + + #[test] + fn test_packed_file_refs_manifest_round_trip_and_validation() { + let region = RegionId::new(7, 3); + let file_id = FileId::parse_str("00000000-0000-0000-0000-000000000001").unwrap(); + let mut manifest = FileRefsManifest::default(); + manifest.file_refs.insert( + region, + HashSet::from([ + FileRef::new(region, file_id, None), + FileRef::new(region, file_id, Some(0)), + FileRef::new(region, file_id, Some(u64::MAX)), + ]), + ); + manifest.manifest_version.insert(region, 42); + manifest + .cross_region_refs + .insert(region, HashSet::from([RegionId::new(7, 4)])); + let packed = PackedFileRefsManifest::from_manifest(&manifest); + assert_eq!(packed.clone().into_manifest().unwrap(), manifest); + let mut missing_files = serde_json::to_value(&packed).unwrap(); + missing_files + .get_mut("file_refs") + .and_then(serde_json::Value::as_object_mut) + .and_then(|file_refs| file_refs.values_mut().next()) + .and_then(serde_json::Value::as_object_mut) + .unwrap() + .remove("files"); + assert!(serde_json::from_value::(missing_files).is_err()); + + let mut missing_indexed = serde_json::to_value(&packed).unwrap(); + missing_indexed + .get_mut("file_refs") + .and_then(serde_json::Value::as_object_mut) + .and_then(|file_refs| file_refs.values_mut().next()) + .and_then(serde_json::Value::as_object_mut) + .unwrap() + .remove("indexed"); + assert!(serde_json::from_value::(missing_indexed).is_err()); + + let mut malformed = packed.clone(); + malformed.file_refs.get_mut(®ion).unwrap().indexed = "!".to_string(); + let err = malformed.into_manifest().unwrap_err(); + assert!(matches!( + &err, + crate::error::Error::DecodePackedFileRefs { .. } + )); + assert!( + err.source() + .is_some_and(|source| source.is::()) + ); + assert_eq!(err.status_code(), StatusCode::Unexpected); + assert_eq!(err.retry_hint(), RetryHint::NonRetryable); + + let mut malformed = packed; + malformed.file_refs.get_mut(®ion).unwrap().indexed = + base64::engine::general_purpose::STANDARD.encode([0; 1]); + let err = malformed.into_manifest().unwrap_err(); + assert!(matches!( + &err, + crate::error::Error::InvalidPackedFileRefs { .. } + )); + assert!(err.source().is_none()); + assert_eq!(err.status_code(), StatusCode::Unexpected); + assert_eq!(err.retry_hint(), RetryHint::NonRetryable); + } + + #[test] + fn test_packed_file_refs_is_smaller_than_legacy() { + let region = RegionId::new(7, 3); + let mut manifest = FileRefsManifest::default(); + manifest.file_refs.insert( + region, + (0..500) + .map(|i| { + let id = + FileId::parse_str(&format!("00000000-0000-0000-0000-{i:012}")).unwrap(); + FileRef::new(region, id, None) + }) + .collect(), + ); + let legacy = serde_json::to_vec(&manifest).unwrap().len(); + let packed = serde_json::to_vec(&PackedFileRefsManifest::from_manifest(&manifest)) + .unwrap() + .len(); + assert!(packed * 2 < legacy, "packed={packed}, legacy={legacy}"); + } + #[test] fn test_serialize_get_file_refs_instruction_reply() { let mut manifest = FileRefsManifest::default(); diff --git a/src/datanode/src/heartbeat.rs b/src/datanode/src/heartbeat.rs index 0e0c51624e..f8d9221272 100644 --- a/src/datanode/src/heartbeat.rs +++ b/src/datanode/src/heartbeat.rs @@ -18,6 +18,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use api::v1::meta::heartbeat_request::NodeWorkloads; +use api::v1::meta::mailbox_message::Payload; use api::v1::meta::{DatanodeWorkloads, HeartbeatRequest, NodeInfo, Peer, RegionRole, RegionStat}; use common_base::Plugins; use common_meta::cache_invalidator::CacheInvalidatorRef; @@ -146,7 +147,15 @@ impl HeartbeatTask { None }) { if let Some(msg) = res.mailbox_message.as_ref() { - info!("Received mailbox message: {msg:?}, meta_client id: {client_id:?}"); + let payload_len = msg + .payload + .as_ref() + .map(|Payload::Json(payload)| payload.len()) + .unwrap_or_default(); + info!( + message_id = msg.id, + payload_len, client_id, "Received mailbox message" + ); } if let Some(lease) = res.region_lease.as_ref() { metrics::LAST_RECEIVED_HEARTBEAT_ELAPSED @@ -193,7 +202,8 @@ impl HeartbeatTask { ctx: HeartbeatResponseHandlerContext, handler_executor: HeartbeatResponseHandlerExecutorRef, ) -> Result<()> { - trace!("Heartbeat response: {:?}", ctx.response); + let mailbox_message_id = ctx.response.mailbox_message.as_ref().map(|msg| msg.id); + trace!(?mailbox_message_id, "Handling heartbeat response"); handler_executor .handle(ctx) .await @@ -379,7 +389,8 @@ impl HeartbeatTask { .set(last_sent.elapsed().as_millis() as i64); // Resets the timer. last_sent = Instant::now(); - debug!("Sending heartbeat request: {:?}", req); + let mailbox_message_id = req.mailbox_message.as_ref().map(|msg| msg.id); + debug!(?mailbox_message_id, "Sending heartbeat request"); if let Err(e) = tx.send(req).await { error!(e; "Failed to send heartbeat to metasrv"); match Self::create_streams( diff --git a/src/datanode/src/heartbeat/handler.rs b/src/datanode/src/heartbeat/handler.rs index 71797592c3..2d18d56004 100644 --- a/src/datanode/src/heartbeat/handler.rs +++ b/src/datanode/src/heartbeat/handler.rs @@ -41,9 +41,9 @@ use crate::heartbeat::handler::apply_staging_manifest::ApplyStagingManifestsHand use crate::heartbeat::handler::close_region::CloseRegionsHandler; use crate::heartbeat::handler::downgrade_region::DowngradeRegionsHandler; use crate::heartbeat::handler::enter_staging::EnterStagingRegionsHandler; -use crate::heartbeat::handler::file_ref::GetFileRefsHandler; +use crate::heartbeat::handler::file_ref::{GetFileRefsHandler, GetPackedFileRefsHandler}; use crate::heartbeat::handler::flush_region::FlushRegionsHandler; -use crate::heartbeat::handler::gc_worker::GcRegionsHandler; +use crate::heartbeat::handler::gc_worker::{GcRegionsHandler, PackedGcRegionsHandler}; use crate::heartbeat::handler::open_region::OpenRegionsHandler; use crate::heartbeat::handler::remap_manifest::RemapManifestHandler; use crate::heartbeat::handler::sync_region::SyncRegionHandler; @@ -135,7 +135,11 @@ impl RegionHeartbeatResponseHandler { .into(), ))), Instruction::GetFileRefs(_) => Ok(Some(Box::new(GetFileRefsHandler.into()))), + Instruction::GetPackedFileRefs(_) => { + Ok(Some(Box::new(GetPackedFileRefsHandler.into()))) + } Instruction::GcRegions(_) => Ok(Some(Box::new(GcRegionsHandler.into()))), + Instruction::PackedGcRegions(_) => Ok(Some(Box::new(PackedGcRegionsHandler.into()))), Instruction::InvalidateCaches(_) => InvalidHeartbeatResponseSnafu.fail(), Instruction::Suspend => Ok(None), Instruction::EnterStagingRegions(_) => { @@ -159,7 +163,9 @@ pub enum InstructionHandlers { DowngradeRegions(DowngradeRegionsHandler), UpgradeRegions(UpgradeRegionsHandler), GetFileRefs(GetFileRefsHandler), + GetPackedFileRefs(GetPackedFileRefsHandler), GcRegions(GcRegionsHandler), + PackedGcRegions(PackedGcRegionsHandler), EnterStagingRegions(EnterStagingRegionsHandler), SyncRegions(SyncRegionHandler), RemapManifest(RemapManifestHandler), @@ -192,7 +198,9 @@ impl_from_handler!( DowngradeRegionsHandler => DowngradeRegions, UpgradeRegionsHandler => UpgradeRegions, GetFileRefsHandler => GetFileRefs, + GetPackedFileRefsHandler => GetPackedFileRefs, GcRegionsHandler => GcRegions, + PackedGcRegionsHandler => PackedGcRegions, EnterStagingRegionsHandler => EnterStagingRegions, SyncRegionHandler => SyncRegions, RemapManifestHandler => RemapManifest, @@ -240,7 +248,9 @@ dispatch_instr!( DowngradeRegions => DowngradeRegions, UpgradeRegions => UpgradeRegions, GetFileRefs => GetFileRefs, + GetPackedFileRefs => GetPackedFileRefs, GcRegions => GcRegions, + PackedGcRegions => PackedGcRegions, EnterStagingRegions => EnterStagingRegions, SyncRegions => SyncRegions, RemapManifest => RemapManifest, diff --git a/src/datanode/src/heartbeat/handler/file_ref.rs b/src/datanode/src/heartbeat/handler/file_ref.rs index ef9cbd4158..554f4119b0 100644 --- a/src/datanode/src/heartbeat/handler/file_ref.rs +++ b/src/datanode/src/heartbeat/handler/file_ref.rs @@ -13,7 +13,10 @@ // limitations under the License. use common_error::ext::ErrorExt; -use common_meta::instruction::{GetFileRefs, GetFileRefsReply, InstructionError, InstructionReply}; +use common_meta::instruction::{ + GetFileRefs, GetFileRefsReply, GetPackedFileRefs, GetPackedFileRefsReply, InstructionError, + InstructionReply, PackedFileRefsManifest, +}; use store_api::storage::FileRefsManifest; use crate::heartbeat::handler::{HandlerContext, InstructionHandler}; @@ -65,3 +68,54 @@ impl InstructionHandler for GetFileRefsHandler { } } } + +pub struct GetPackedFileRefsHandler; + +#[async_trait::async_trait] +impl InstructionHandler for GetPackedFileRefsHandler { + type Instruction = GetPackedFileRefs; + + async fn handle( + &self, + ctx: &HandlerContext, + get_file_refs: Self::Instruction, + ) -> Option { + let region_server = &ctx.region_server; + let Some(mito_engine) = region_server.mito_engine() else { + return Some(InstructionReply::GetPackedFileRefs( + GetPackedFileRefsReply { + packed_file_refs_manifest: PackedFileRefsManifest::default(), + success: false, + error: Some(InstructionError::legacy_internal_retryable( + "MitoEngine not found", + )), + }, + )); + }; + match mito_engine + .get_snapshot_of_file_refs(get_file_refs.query_regions, get_file_refs.related_regions) + .await + { + Ok(all_file_refs) => Some(InstructionReply::GetPackedFileRefs( + GetPackedFileRefsReply { + packed_file_refs_manifest: PackedFileRefsManifest::from_manifest( + &all_file_refs, + ), + success: true, + error: None, + }, + )), + Err(e) => Some(InstructionReply::GetPackedFileRefs( + GetPackedFileRefsReply { + packed_file_refs_manifest: PackedFileRefsManifest::default(), + success: false, + error: Some(InstructionError::new( + e.status_code(), + format!("Failed to get file refs: {}", e.output_msg()), + e.retry_hint(), + )), + }, + )), + } + } +} diff --git a/src/datanode/src/heartbeat/handler/gc_worker.rs b/src/datanode/src/heartbeat/handler/gc_worker.rs index b32a1c29ad..e63d21b6e5 100644 --- a/src/datanode/src/heartbeat/handler/gc_worker.rs +++ b/src/datanode/src/heartbeat/handler/gc_worker.rs @@ -15,7 +15,7 @@ use std::collections::{BTreeMap, HashMap}; use std::sync::Arc; -use common_meta::instruction::{GcRegions, GcRegionsReply, InstructionReply}; +use common_meta::instruction::{GcRegions, GcRegionsReply, InstructionReply, PackedGcRegions}; use common_meta::key::table_info::TableInfoManager; use common_meta::key::table_route::TableRouteManager; use common_telemetry::{debug, warn}; @@ -34,6 +34,7 @@ use crate::error::{GcMitoEngineSnafu, GetMetadataSnafu, Result, UnexpectedSnafu} use crate::heartbeat::handler::{HandlerContext, InstructionHandler}; pub struct GcRegionsHandler; +pub struct PackedGcRegionsHandler; #[async_trait::async_trait] impl InstructionHandler for GcRegionsHandler { @@ -67,7 +68,7 @@ impl InstructionHandler for GcRegionsHandler { .push(rid); } - let file_refs_manifest = gc_regions.file_refs_manifest.clone(); + let file_refs_manifest = gc_regions.file_refs_manifest; let full_file_listing = gc_regions.full_file_listing; let ctx_clone = ctx.clone(); @@ -138,6 +139,36 @@ impl InstructionHandler for GcRegionsHandler { } } +#[async_trait::async_trait] +impl InstructionHandler for PackedGcRegionsHandler { + type Instruction = PackedGcRegions; + + async fn handle( + &self, + ctx: &HandlerContext, + instruction: Self::Instruction, + ) -> Option { + let manifest = match instruction.packed_file_refs_manifest.into_manifest() { + Ok(manifest) => manifest, + Err(err) => { + return Some(InstructionReply::GcRegions(GcRegionsReply { + result: Err(common_meta::instruction::InstructionError::from_error(&err)), + })); + } + }; + GcRegionsHandler + .handle( + ctx, + GcRegions { + regions: instruction.regions, + file_refs_manifest: manifest, + full_file_listing: instruction.full_file_listing, + }, + ) + .await + } +} + impl GcRegionsHandler { /// Create a GC worker for the given table and region IDs. async fn create_gc_worker( diff --git a/src/meta-srv/src/error.rs b/src/meta-srv/src/error.rs index 53872d4817..70997ebe7d 100644 --- a/src/meta-srv/src/error.rs +++ b/src/meta-srv/src/error.rs @@ -368,9 +368,9 @@ pub enum Error { location: Location, }, - #[snafu(display("Failed to deserialize from json: {}", input))] + #[snafu(display("Failed to deserialize from json payload of length {}", input_len))] DeserializeFromJson { - input: String, + input_len: usize, #[snafu(source)] error: serde_json::error::Error, #[snafu(implicit)] diff --git a/src/meta-srv/src/event/gc.rs b/src/meta-srv/src/event/gc.rs index 7f679f1785..94d671a176 100644 --- a/src/meta-srv/src/event/gc.rs +++ b/src/meta-srv/src/event/gc.rs @@ -218,6 +218,7 @@ mod tests { use common_event_recorder::testing::assert_event_contract; use common_event_recorder::{EventTypeFilter, PersistentEventContext, TriggerReason}; use common_meta::key::TableMetadataManager; + use common_meta::key::runtime_switch::RuntimeSwitchManager; use common_meta::kv_backend::memory::MemoryKvBackend; use common_meta::sequence::SequenceBuilder; use common_procedure::{ @@ -588,11 +589,13 @@ mod tests { fn batch_gc_procedure() -> BatchGcProcedure { let kv_backend = Arc::new(MemoryKvBackend::new()); let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone())); + let runtime_switch_manager = Arc::new(RuntimeSwitchManager::new(kv_backend.clone())); let mailbox_sequence = SequenceBuilder::new("test_batch_gc_event", kv_backend).build(); let mailbox = MailboxContext::new(mailbox_sequence); BatchGcProcedure::new( mailbox.mailbox().clone(), table_metadata_manager, + runtime_switch_manager, "localhost".to_string(), vec![RegionId::new(1024, 1)], true, diff --git a/src/meta-srv/src/gc/ctx.rs b/src/meta-srv/src/gc/ctx.rs index 4b46a941e2..857f5e87cd 100644 --- a/src/meta-srv/src/gc/ctx.rs +++ b/src/meta-srv/src/gc/ctx.rs @@ -27,6 +27,7 @@ use common_meta::ddl_manager::DdlManagerRef; #[cfg(feature = "enterprise")] use common_meta::key::DroppedTableName; use common_meta::key::TableMetadataManagerRef; +use common_meta::key::runtime_switch::RuntimeSwitchManagerRef; use common_meta::key::table_repart::TableRepartValue; use common_meta::key::table_route::PhysicalTableRouteValue; #[cfg(feature = "enterprise")] @@ -155,6 +156,8 @@ pub(crate) struct DefaultGcSchedulerCtx { pub(crate) table_metadata_manager: TableMetadataManagerRef, /// Procedure manager. pub(crate) procedure_manager: ProcedureManagerRef, + /// Runtime switch manager used by recovered and newly submitted GC procedures. + pub(crate) runtime_switch_manager: RuntimeSwitchManagerRef, /// DDL manager used to submit the existing purge procedure. #[cfg(feature = "enterprise")] pub(crate) ddl_manager: DdlManagerRef, @@ -176,6 +179,7 @@ impl DefaultGcSchedulerCtx { pub fn try_new( table_metadata_manager: TableMetadataManagerRef, procedure_manager: ProcedureManagerRef, + runtime_switch_manager: RuntimeSwitchManagerRef, #[cfg(feature = "enterprise")] ddl_manager: DdlManagerRef, meta_peer_client: MetaPeerClientRef, mailbox: MailboxRef, @@ -184,6 +188,7 @@ impl DefaultGcSchedulerCtx { Ok(Self { table_metadata_manager, procedure_manager, + runtime_switch_manager, #[cfg(feature = "enterprise")] ddl_manager, #[cfg(feature = "enterprise")] @@ -320,6 +325,7 @@ impl DefaultGcSchedulerCtx { let procedure = BatchGcProcedure::new( self.mailbox.clone(), self.table_metadata_manager.clone(), + self.runtime_switch_manager.clone(), self.server_addr.clone(), region_ids.to_vec(), full_file_listing, diff --git a/src/meta-srv/src/gc/procedure.rs b/src/meta-srv/src/gc/procedure.rs index f9f597577a..db22f3ab00 100644 --- a/src/meta-srv/src/gc/procedure.rs +++ b/src/meta-srv/src/gc/procedure.rs @@ -18,8 +18,14 @@ use std::sync::Arc; use std::time::Duration; use api::v1::meta::MailboxMessage; -use common_meta::instruction::{self, GcRegions, GetFileRefs, GetFileRefsReply, InstructionReply}; +use common_error::ext::PlainError; +use common_error::status_code::StatusCode; +use common_meta::instruction::{ + self, GetPackedFileRefs, GetPackedFileRefsReply, InstructionReply, PackedFileRefsManifest, + PackedGcRegions, PackedRegionFileRefs, +}; use common_meta::key::TableMetadataManagerRef; +use common_meta::key::runtime_switch::RuntimeSwitchManagerRef; use common_meta::key::table_repart::TableRepartValue; use common_meta::key::table_route::PhysicalTableRouteValue; use common_meta::lock_key::{RegionLock, TableLock}; @@ -34,6 +40,7 @@ use common_telemetry::tracing::Instrument as _; use common_telemetry::tracing_context::TracingContext; use common_telemetry::{debug, error, info, warn}; use futures::future::join_all; +use futures::stream::{FuturesUnordered, StreamExt}; use itertools::Itertools as _; use serde::{Deserialize, Serialize}; use snafu::ResultExt as _; @@ -49,14 +56,14 @@ use crate::metrics::{METRIC_META_GC_DATANODE_CALLS_TOTAL, METRIC_META_GC_FAILED_ use crate::procedure::utils::{instruction_error_result, instruction_to_error}; use crate::service::mailbox::{Channel, MailboxReceiver, MailboxRef}; -async fn send_get_file_refs_inner( +async fn send_get_packed_file_refs_inner( mailbox: &MailboxRef, server_addr: &str, peer: &Peer, - instruction: GetFileRefs, + instruction: GetPackedFileRefs, timeout: Duration, ) -> Result { - let instruction = instruction::Instruction::GetFileRefs(instruction); + let instruction = instruction::Instruction::GetPackedFileRefs(instruction); let tracing_ctx = TracingContext::from_current_span(); let msg = MailboxMessage::json_message( &format!("Get file references: {}", instruction), @@ -75,25 +82,25 @@ async fn send_get_file_refs_inner( .await } -async fn recv_get_file_refs_reply( +async fn recv_get_packed_file_refs_reply( peer: &Peer, mailbox_rx: MailboxReceiver, -) -> Result { +) -> Result { let reply = match mailbox_rx.await { Ok(reply_msg) => HeartbeatMailbox::json_reply(&reply_msg)?, Err(e) => { error!( - e; "Failed to receive reply from datanode {} for GetFileRefs instruction", + e; "Failed to receive reply from datanode {} for GetPackedFileRefs instruction", peer, ); return Err(e); } }; - let InstructionReply::GetFileRefs(reply) = reply else { + let InstructionReply::GetPackedFileRefs(reply) = reply else { return error::UnexpectedInstructionReplySnafu { - mailbox_message: format!("{:?}", reply), - reason: "Unexpected reply of the GetFileRefs instruction", + mailbox_message: "unexpected instruction reply for GetPackedFileRefs".to_string(), + reason: "Unexpected reply of the GetPackedFileRefs instruction", } .fail(); }; @@ -104,12 +111,11 @@ async fn recv_get_file_refs_reply( async fn send_gc_regions_inner( mailbox: &MailboxRef, peer: &Peer, - gc_regions: &GcRegions, + instruction: instruction::Instruction, server_addr: &str, timeout: Duration, description: &str, ) -> Result { - let instruction = instruction::Instruction::GcRegions(gc_regions.clone()); let tracing_ctx = TracingContext::from_current_span(); let msg = MailboxMessage::json_message( &format!("{}: {}", description, instruction), @@ -128,9 +134,43 @@ async fn send_gc_regions_inner( .await } +fn scoped_gc_instruction( + regions: Vec, + full_manifest: &FileRefsManifest, + full_file_listing: bool, +) -> instruction::Instruction { + let packed_file_refs_manifest = PackedFileRefsManifest { + file_refs: regions + .iter() + .filter_map(|region| { + full_manifest + .file_refs + .get(region) + .map(|refs| (*region, PackedRegionFileRefs::from_refs(refs))) + }) + .collect(), + manifest_version: regions + .iter() + .filter_map(|region| { + full_manifest + .manifest_version + .get(region) + .map(|version| (*region, *version)) + }) + .collect(), + cross_region_refs: HashMap::new(), + }; + + instruction::Instruction::PackedGcRegions(PackedGcRegions { + regions, + packed_file_refs_manifest, + full_file_listing, + }) +} + async fn recv_gc_regions_reply( peer: &Peer, - gc_regions: &GcRegions, + regions: &[RegionId], description: &str, mailbox_rx: MailboxReceiver, ) -> Result { @@ -147,7 +187,7 @@ async fn recv_gc_regions_reply( let InstructionReply::GcRegions(reply) = reply else { return error::UnexpectedInstructionReplySnafu { - mailbox_message: format!("{:?}", reply), + mailbox_message: "unexpected instruction reply for GcRegions".to_string(), reason: "Unexpected reply of the GcRegions instruction", } .fail(); @@ -158,14 +198,16 @@ async fn recv_gc_regions_reply( Ok(report) => Ok(report), Err(e) => { error!( - e; "Datanode {} reported error during GC for regions {:?}", - peer, gc_regions + e; "Datanode {} reported error during GC for {} regions", + peer, regions.len() ); instruction_error_result( &e, format!( - "Datanode {} reported error during GC for regions {:?}: {}", - peer, gc_regions, e + "Datanode {} reported error during GC for {} regions: {}", + peer, + regions.len(), + e ), ) } @@ -177,6 +219,7 @@ async fn recv_gc_regions_reply( pub struct BatchGcProcedure { mailbox: MailboxRef, table_metadata_manager: TableMetadataManagerRef, + runtime_switch_manager: RuntimeSwitchManagerRef, data: BatchGcData, } @@ -217,9 +260,11 @@ pub enum State { impl BatchGcProcedure { pub const TYPE_NAME: &'static str = "metasrv-procedure::BatchGcProcedure"; + #[allow(clippy::too_many_arguments)] pub fn new( mailbox: MailboxRef, table_metadata_manager: TableMetadataManagerRef, + runtime_switch_manager: RuntimeSwitchManagerRef, server_addr: String, regions: Vec, full_file_listing: bool, @@ -229,6 +274,7 @@ impl BatchGcProcedure { Self { mailbox, table_metadata_manager, + runtime_switch_manager, data: BatchGcData { state: State::Start, server_addr, @@ -251,6 +297,7 @@ impl BatchGcProcedure { pub fn new_update_repartition_for_test( mailbox: MailboxRef, table_metadata_manager: TableMetadataManagerRef, + runtime_switch_manager: RuntimeSwitchManagerRef, server_addr: String, regions: Vec, file_refs: FileRefsManifest, @@ -259,6 +306,7 @@ impl BatchGcProcedure { Self { mailbox, table_metadata_manager, + runtime_switch_manager, data: BatchGcData { state: State::UpdateRepartition, server_addr, @@ -286,6 +334,22 @@ impl BatchGcProcedure { }) } + async fn check_maintenance_mode(&self) -> ProcedureResult<()> { + let enabled = self + .runtime_switch_manager + .maintenance_mode() + .await + .context(error::RuntimeSwitchManagerSnafu) + .map_err(ProcedureError::retry_later)?; + if enabled { + return Err(ProcedureError::external(PlainError::new( + "maintenance mode is enabled".to_string(), + StatusCode::IllegalState, + ))); + } + Ok(()) + } + fn merge_gc_report(&mut self, report: GcReport) { let accumulated = self.data.gc_report.get_or_insert_default(); // Deleted objects are cumulative, while these sets describe the latest outcome @@ -644,10 +708,10 @@ impl BatchGcProcedure { } } - // Send GetFileRefs instructions to each datanode + // Send packed GetFileRefs instructions to each datanode let mut all_file_refs: HashMap> = HashMap::new(); let mut all_manifest_versions = HashMap::new(); - let mut all_cross_region_refs = HashMap::new(); + let mut all_cross_region_refs: HashMap> = HashMap::new(); let mut peers = HashSet::new(); peers.extend(datanode2query_regions.keys().cloned()); @@ -655,33 +719,34 @@ impl BatchGcProcedure { let mailbox = &self.mailbox; let server_addr = &self.data.server_addr; - let mut tasks = Vec::new(); - + // Each future owns both send and receive, so a completed reply is merged and + // dropped immediately rather than retained in a second join_all buffer. + let mut tasks = FuturesUnordered::new(); for peer in peers { let regions = datanode2query_regions.remove(&peer).unwrap_or_default(); let related_regions_for_peer = datanode2related_regions.remove(&peer).unwrap_or_default(); - if regions.is_empty() && related_regions_for_peer.is_empty() { continue; } - tasks.push(async move { - let instruction = GetFileRefs { - query_regions: regions.clone(), - related_regions: related_regions_for_peer.clone(), + let instruction = GetPackedFileRefs { + query_regions: regions, + related_regions: related_regions_for_peer, }; - - let reply = - send_get_file_refs_inner(mailbox, server_addr, &peer, instruction, timeout) - .await; - - (peer, regions, related_regions_for_peer, reply) + let rx = send_get_packed_file_refs_inner( + mailbox, + server_addr, + &peer, + instruction, + timeout, + ) + .await?; + let reply = recv_get_packed_file_refs_reply(&peer, rx).await?; + Ok::<_, crate::error::Error>((peer, reply)) }); } - let mut recv_tasks = Vec::new(); - // store error to make sure metrics doesn't ignore other peers let mut first_error = None; let mut record_get_file_refs_error = |e| { METRIC_META_GC_DATANODE_CALLS_TOTAL @@ -691,80 +756,56 @@ impl BatchGcProcedure { first_error = Some(e); } }; - for (peer, regions, related_regions_for_peer, reply) in join_all(tasks).await { - match reply { - Ok(mailbox_rx) => { - recv_tasks.push(async move { - let reply = recv_get_file_refs_reply(&peer, mailbox_rx).await; - (peer, regions, related_regions_for_peer, reply) - }); - } - Err(e) => record_get_file_refs_error(e), - } - } - - let replies = join_all(recv_tasks).await; - - for (peer, regions, related_regions_for_peer, reply) in replies { - let reply = match reply { + while let Some(result) = tasks.next().await { + let (peer, reply) = match result { Ok(reply) => reply, Err(e) => { record_get_file_refs_error(e); continue; } }; - debug!( - "Got file references from datanode: {:?}, query_regions: {:?}, related_regions: {:?}, reply: {:?}", - peer, regions, related_regions_for_peer, reply - ); - if !reply.success { - METRIC_META_GC_DATANODE_CALLS_TOTAL - .with_label_values(&["get_file_refs", "error"]) - .inc(); let err = if let Some(error) = &reply.error { instruction_to_error( error, - format!( - "Failed to get file references from datanode {}: {:?}", - peer, error - ), + format!("Failed to get file references from datanode {peer}"), ) } else { error::UnexpectedSnafu { - violated: format!( - "Failed to get file references from datanode {}: {:?}", - peer, reply.error - ), + violated: + "Datanode returned an unsuccessful GetPackedFileRefs reply without an error" + .to_string(), } .build() }; record_get_file_refs_error(err); continue; } + let manifest = match reply.packed_file_refs_manifest.into_manifest() { + Ok(manifest) => manifest, + Err(err) => { + record_get_file_refs_error(error::Error::Other { + source: common_error::ext::BoxedError::new(err), + location: snafu::Location::new(file!(), line!(), 0), + }); + continue; + } + }; METRIC_META_GC_DATANODE_CALLS_TOTAL .with_label_values(&["get_file_refs", "success"]) .inc(); - - // Merge the file references from this datanode - for (region_id, file_refs) in reply.file_refs_manifest.file_refs { - all_file_refs - .entry(region_id) - .or_default() - .extend(file_refs); + for (region_id, refs) in manifest.file_refs { + all_file_refs.entry(region_id).or_default().extend(refs); } - - // region manifest version should be the smallest one among all peers, so outdated region can be detected - for (region_id, version) in reply.file_refs_manifest.manifest_version { + for (region_id, version) in manifest.manifest_version { let entry = all_manifest_versions.entry(region_id).or_insert(version); *entry = (*entry).min(version); } - - for (region_id, related_region_ids) in reply.file_refs_manifest.cross_region_refs { - let entry = all_cross_region_refs + for (region_id, related) in manifest.cross_region_refs { + all_cross_region_refs .entry(region_id) - .or_insert_with(HashSet::new); - entry.extend(related_region_ids); + .or_default() + .extend(related); } } @@ -813,28 +854,23 @@ impl BatchGcProcedure { let tasks = datanode2regions .into_iter() .map(|(peer, regions_for_peer)| { - let gc_regions = GcRegions { - regions: regions_for_peer.clone(), - // file_refs_manifest could be somewhere large. But still intentionally clone per datanode here: - // this path is admin-triggered or scheduler-triggered, peer count is expected to be bounded, and - // and abnormal manifest growth should be addressed at the source - file_refs_manifest: file_refs.clone(), - full_file_listing, - }; - let region_count = gc_regions.regions.len() as u64; + let region_count = regions_for_peer.len() as u64; + let regions = regions_for_peer.clone(); + let instruction = + scoped_gc_instruction(regions_for_peer, file_refs, full_file_listing); async move { let report = send_gc_regions_inner( mailbox, &peer, - &gc_regions, + instruction, server_addr, timeout, "Batch GC", ) .await; - (peer, gc_regions, region_count, report) + (peer, regions, region_count, report) } }); @@ -851,12 +887,12 @@ impl BatchGcProcedure { first_error = Some(e); } }; - for (peer, gc_regions, region_count, report) in join_all(tasks).await { + for (peer, regions, region_count, report) in join_all(tasks).await { match report { Ok(mailbox_rx) => { recv_tasks.push(async move { let report = - recv_gc_regions_reply(&peer, &gc_regions, "Batch GC", mailbox_rx).await; + recv_gc_regions_reply(&peer, ®ions, "Batch GC", mailbox_rx).await; (peer, region_count, report) }); } @@ -887,12 +923,12 @@ impl BatchGcProcedure { if need_retry.is_empty() { info!( - "GC report from datanode {}: successfully deleted files for regions {:?}", + "GC report from datanode {}: successfully deleted files for region IDs {:?}", peer, success ); } else { warn!( - "GC report from datanode {}: successfully deleted files for regions {:?}, need retry for regions {:?}", + "GC report from datanode {}: successfully deleted files for region IDs {:?}, need retry for region IDs {:?}", peer, success, need_retry ); } @@ -921,6 +957,8 @@ impl Procedure for BatchGcProcedure { } async fn execute(&mut self, ctx: &ProcedureContext) -> ProcedureResult { + self.check_maintenance_mode().await?; + match self.data.state { State::Start => { let _regions_span = common_telemetry::tracing::debug_span!( @@ -1122,11 +1160,15 @@ mod tests { use api::v1::meta::MailboxMessage; use api::v1::meta::mailbox_message::Payload; - use common_meta::instruction::{GcRegionsReply, InstructionReply}; + use common_meta::instruction::{GcRegionsReply, Instruction, InstructionReply}; use common_meta::key::TableMetadataManager; + use common_meta::key::runtime_switch::RuntimeSwitchManager; use common_meta::kv_backend::memory::MemoryKvBackend; + use common_meta::kv_backend::test_util::MockKvBackendBuilder; use common_meta::peer::Peer; use common_meta::sequence::SequenceBuilder; + use common_procedure::Context as ProcedureContext; + use common_procedure_test::MockContextProvider; use common_time::util::current_time_millis; use store_api::storage::FileId; use tokio::sync::mpsc; @@ -1135,6 +1177,38 @@ mod tests { use crate::procedure::test_util::{MailboxContext, send_mock_reply}; use crate::service::mailbox::Channel; + #[test] + fn test_scoped_gc_instruction_selects_and_scopes_manifest() { + let region = RegionId::new(7, 3); + let other = RegionId::new(7, 4); + let mut manifest = FileRefsManifest::default(); + manifest.file_refs.insert(region, HashSet::new()); + manifest.file_refs.insert(other, HashSet::new()); + manifest.manifest_version.insert(region, 42); + + let packed = scoped_gc_instruction(vec![region], &manifest, true); + let Instruction::PackedGcRegions(packed) = packed else { + panic!("expected packed instruction"); + }; + assert_eq!(packed.regions, vec![region]); + assert_eq!( + packed.packed_file_refs_manifest.manifest_version[®ion], + 42 + ); + assert!( + !packed + .packed_file_refs_manifest + .file_refs + .contains_key(&other) + ); + assert!( + packed + .packed_file_refs_manifest + .cross_region_refs + .is_empty() + ); + } + #[test] fn test_done_with_gc_report_keeps_report() { let region_id = RegionId::new(1024, 1); @@ -1215,6 +1289,7 @@ mod tests { let kv_backend = Arc::new(MemoryKvBackend::new()); let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone())); + let runtime_switch_manager = Arc::new(RuntimeSwitchManager::new(kv_backend.clone())); let mailbox_sequence = SequenceBuilder::new("test_batch_gc_partial_report", kv_backend).build(); let mut mailbox = MailboxContext::new(mailbox_sequence); @@ -1230,6 +1305,7 @@ mod tests { let mut procedure = BatchGcProcedure::new( mailbox.mailbox().clone(), table_metadata_manager, + runtime_switch_manager, "localhost".to_string(), vec![first_region, second_region], true, @@ -1265,11 +1341,15 @@ mod tests { fn batch_gc_procedure() -> BatchGcProcedure { let kv_backend = Arc::new(MemoryKvBackend::new()); let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone())); - let mailbox_sequence = SequenceBuilder::new("test_batch_gc_procedure", kv_backend).build(); + let runtime_switch_manager = Arc::new(RuntimeSwitchManager::new(kv_backend)); + let mailbox_sequence = + SequenceBuilder::new("test_batch_gc_procedure", Arc::new(MemoryKvBackend::new())) + .build(); let mailbox = MailboxContext::new(mailbox_sequence); BatchGcProcedure::new( mailbox.mailbox().clone(), table_metadata_manager, + runtime_switch_manager, "localhost".to_string(), vec![RegionId::new(1024, 1)], true, @@ -1277,4 +1357,113 @@ mod tests { HashMap::new(), ) } + + #[tokio::test] + async fn test_maintenance_mode_gates_every_gc_state() { + let states = [ + State::Start, + State::Acquiring, + State::Gcing, + State::UpdateRepartition, + ]; + + for state in states { + let kv_backend = Arc::new(MemoryKvBackend::new()); + let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone())); + let runtime_switch_manager = Arc::new(RuntimeSwitchManager::new(kv_backend.clone())); + runtime_switch_manager.set_maintenance_mode().await.unwrap(); + let mailbox_sequence = + SequenceBuilder::new("test_batch_gc_maintenance_gate", kv_backend).build(); + let mut mailbox = MailboxContext::new(mailbox_sequence); + let (tx, mut rx) = mpsc::channel(1); + mailbox + .insert_heartbeat_response_receiver(Channel::Datanode(1), tx) + .await; + let mut procedure = BatchGcProcedure::new( + mailbox.mailbox().clone(), + table_metadata_manager, + runtime_switch_manager, + "localhost".to_string(), + vec![RegionId::new(1024, 1)], + true, + Duration::from_secs(10), + HashMap::new(), + ); + procedure.data.state = state.clone(); + let dump_before = procedure.dump().unwrap(); + + let ctx = ProcedureContext { + procedure_id: common_procedure::ProcedureId::random(), + provider: Arc::new(MockContextProvider::default()), + event_context: None, + }; + let err = procedure.execute(&ctx).await.unwrap_err(); + + assert!(!err.is_retry_later()); + assert_eq!( + common_error::ext::ErrorExt::retry_hint(&err), + common_error::ext::RetryHint::NonRetryable + ); + assert_eq!(procedure.data.state, state); + assert_eq!(procedure.dump().unwrap(), dump_before); + if matches!(state, State::Acquiring | State::Gcing) { + assert!(rx.try_recv().is_err()); + } + } + } + + #[tokio::test] + async fn test_maintenance_mode_read_error_retries_without_advancing_state() { + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let failing_once_calls = calls.clone(); + let kv_backend = Arc::new( + MockKvBackendBuilder::default() + .range_fn(Arc::new(move |_| { + if failing_once_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 { + common_meta::error::UnexpectedSnafu { + err_msg: "maintenance read failed", + } + .fail() + } else { + Ok(common_meta::rpc::store::RangeResponse { + kvs: vec![], + more: false, + }) + } + })) + .build() + .unwrap(), + ); + let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone())); + let runtime_switch_manager = Arc::new(RuntimeSwitchManager::new(kv_backend.clone())); + let mailbox_sequence = + SequenceBuilder::new("test_batch_gc_maintenance_read_error", kv_backend).build(); + let mailbox = MailboxContext::new(mailbox_sequence); + let mut procedure = BatchGcProcedure::new( + mailbox.mailbox().clone(), + table_metadata_manager, + runtime_switch_manager, + "localhost".to_string(), + vec![RegionId::new(1024, 1)], + true, + Duration::from_secs(10), + HashMap::new(), + ); + let ctx = ProcedureContext { + procedure_id: common_procedure::ProcedureId::random(), + provider: Arc::new(MockContextProvider::default()), + event_context: None, + }; + let dump_before = procedure.dump().unwrap(); + + assert!(procedure.execute(&ctx).await.unwrap_err().is_retry_later()); + assert_eq!(procedure.data.state, State::Start); + assert_eq!(procedure.dump().unwrap(), dump_before); + + assert!(matches!( + procedure.execute(&ctx).await.unwrap(), + Status::Executing { .. } + )); + assert_eq!(procedure.data.state, State::Acquiring); + } } diff --git a/src/meta-srv/src/handler.rs b/src/meta-srv/src/handler.rs index 50b53f4970..320f26fae7 100644 --- a/src/meta-srv/src/handler.rs +++ b/src/meta-srv/src/handler.rs @@ -428,9 +428,11 @@ impl HeartbeatMailbox { .as_ref() .with_context(|| UnexpectedInstructionReplySnafu { mailbox_message: msg.to_string(), - reason: format!("empty payload, msg: {msg:?}"), + reason: "empty JSON payload".to_string(), })?; - serde_json::from_str(payload).context(DeserializeFromJsonSnafu { input: payload }) + serde_json::from_str(payload).context(DeserializeFromJsonSnafu { + input_len: payload.len(), + }) } /// Parses the [Instruction] from [MailboxMessage]. @@ -443,9 +445,11 @@ impl HeartbeatMailbox { .as_ref() .with_context(|| UnexpectedInstructionReplySnafu { mailbox_message: msg.to_string(), - reason: format!("empty payload, msg: {msg:?}"), + reason: "empty JSON payload".to_string(), })?; - serde_json::from_str(payload).context(DeserializeFromJsonSnafu { input: payload }) + serde_json::from_str(payload).context(DeserializeFromJsonSnafu { + input_len: payload.len(), + }) } pub fn create(pushers: Pushers, sequence: Sequence) -> MailboxRef { @@ -526,7 +530,12 @@ impl Mailbox for HeartbeatMailbox { msg.id = message_id; let pusher_id = ch.pusher_id(); - debug!("Sending mailbox message {msg:?} to {pusher_id}"); + let payload_len = msg + .payload + .as_ref() + .map(|Payload::Json(payload)| payload.len()) + .unwrap_or_default(); + debug!(message_id, payload_len, %pusher_id, "Sending mailbox message"); let (tx, rx) = oneshot::channel(); let _ = self.senders.insert(message_id, tx); @@ -548,7 +557,12 @@ impl Mailbox for HeartbeatMailbox { msg.id = message_id; let pusher_id = ch.pusher_id(); - debug!("Sending mailbox message {msg:?} to {pusher_id}"); + let payload_len = msg + .payload + .as_ref() + .map(|Payload::Json(payload)| payload.len()) + .unwrap_or_default(); + debug!(message_id, payload_len, %pusher_id, "Sending one-way mailbox message"); self.pushers.push(pusher_id, msg).await?; @@ -560,7 +574,18 @@ impl Mailbox for HeartbeatMailbox { } async fn on_recv(&self, id: MessageId, maybe_msg: Result) -> Result<()> { - debug!("Received mailbox message {maybe_msg:?}"); + let payload_len = maybe_msg + .as_ref() + .ok() + .and_then(|msg| msg.payload.as_ref()) + .map(|Payload::Json(payload)| payload.len()) + .unwrap_or_default(); + debug!( + message_id = id, + payload_len, + success = maybe_msg.is_ok(), + "Received mailbox message" + ); let _ = self.timeouts.remove(&id); @@ -568,7 +593,15 @@ impl Mailbox for HeartbeatMailbox { tx.send(maybe_msg) .map_err(|_| error::MailboxClosedSnafu { id }.build())?; } else if let Ok(finally_msg) = maybe_msg { - warn!("The response arrived too late: {finally_msg:?}"); + let payload_len = finally_msg + .payload + .as_ref() + .map(|Payload::Json(payload)| payload.len()) + .unwrap_or_default(); + warn!( + message_id = id, + payload_len, "The mailbox response arrived too late" + ); } Ok(()) diff --git a/src/meta-srv/src/key.rs b/src/meta-srv/src/key.rs index 803ee9f0c6..abdc92ff6d 100644 --- a/src/meta-srv/src/key.rs +++ b/src/meta-srv/src/key.rs @@ -87,7 +87,9 @@ impl FromStr for LeaseValue { type Err = error::Error; fn from_str(value: &str) -> crate::Result { - serde_json::from_str(value).context(error::DeserializeFromJsonSnafu { input: value }) + serde_json::from_str(value).context(error::DeserializeFromJsonSnafu { + input_len: value.len(), + }) } } diff --git a/src/meta-srv/src/metasrv/builder.rs b/src/meta-srv/src/metasrv/builder.rs index 75b72f9a11..b40f06e3cf 100644 --- a/src/meta-srv/src/metasrv/builder.rs +++ b/src/meta-srv/src/metasrv/builder.rs @@ -530,6 +530,7 @@ impl MetasrvBuilder { let gc_scheduler_ctx = DefaultGcSchedulerCtx::try_new( table_metadata_manager.clone(), procedure_manager.clone(), + runtime_switch_manager.clone(), #[cfg(feature = "enterprise")] ddl_manager.clone(), meta_peer_client.clone(), diff --git a/src/meta-srv/src/service/heartbeat.rs b/src/meta-srv/src/service/heartbeat.rs index 066d156047..5cdc5b5789 100644 --- a/src/meta-srv/src/service/heartbeat.rs +++ b/src/meta-srv/src/service/heartbeat.rs @@ -206,7 +206,13 @@ where /// Handles the incoming heartbeat request, and returns whether to continue the session. async fn handle_request(&mut self, req: HeartbeatRequest, is_handshake: bool) -> bool { - debug!("Receiving heartbeat request: {:?}", req); + let mailbox_message_id = req.mailbox_message.as_ref().map(|msg| msg.id); + debug!( + sender_id = %self.sender_id, + ?mailbox_message_id, + is_handshake, + "Receiving heartbeat request" + ); let sender_id = self.sender_id.to_string(); METRIC_META_HEARTBEAT_RECV @@ -224,7 +230,12 @@ where let is_not_leader = res.as_ref().is_ok_and(|r| r.is_not_leader()); - debug!("Sending heartbeat response: {:?}", res); + debug!( + sender_id = %self.sender_id, + success = res.is_ok(), + is_not_leader, + "Sending heartbeat response" + ); if self.tx.send(res).await.is_err() { info!( diff --git a/src/store-api/src/storage/file.rs b/src/store-api/src/storage/file.rs index 0dcfcdc8cd..b0b7614682 100644 --- a/src/store-api/src/storage/file.rs +++ b/src/store-api/src/storage/file.rs @@ -51,6 +51,11 @@ impl FileId { pub fn as_bytes(&self) -> &[u8] { self.0.as_bytes() } + + /// Constructs a file id from its packed 16-byte UUID representation. + pub fn from_bytes(bytes: [u8; 16]) -> FileId { + FileId(Uuid::from_bytes(bytes)) + } } impl From for Uuid { diff --git a/tests-integration/src/tests/gc.rs b/tests-integration/src/tests/gc.rs index badaef5f7a..2024222342 100644 --- a/tests-integration/src/tests/gc.rs +++ b/tests-integration/src/tests/gc.rs @@ -208,6 +208,7 @@ async fn test_gc_basic(store_type: &StorageType) { let procedure = BatchGcProcedure::new( metasrv.mailbox().clone(), metasrv.table_metadata_manager().clone(), + metasrv.runtime_switch_manager().clone(), metasrv.options().grpc.server_addr.clone(), regions.clone(), false, // full_file_listing diff --git a/tests-integration/src/tests/gc/repart.rs b/tests-integration/src/tests/gc/repart.rs index 7e7a3edd5a..93969dbb3b 100644 --- a/tests-integration/src/tests/gc/repart.rs +++ b/tests-integration/src/tests/gc/repart.rs @@ -90,6 +90,7 @@ CREATE TABLE test_cleanup_repartition ( let mut procedure = BatchGcProcedure::new_update_repartition_for_test( metasrv.mailbox().clone(), metasrv.table_metadata_manager().clone(), + metasrv.runtime_switch_manager().clone(), metasrv.options().grpc.server_addr.clone(), regions, manifest, @@ -183,6 +184,7 @@ async fn test_cleanup_region_repartition_preserve_uninvolved_entries() { let mut procedure = BatchGcProcedure::new_update_repartition_for_test( metasrv.mailbox().clone(), metasrv.table_metadata_manager().clone(), + metasrv.runtime_switch_manager().clone(), metasrv.options().grpc.server_addr.clone(), regions, manifest, @@ -267,6 +269,7 @@ async fn test_cleanup_region_repartition_remove_when_tmp_refs_empty() { let mut procedure = BatchGcProcedure::new_update_repartition_for_test( metasrv.mailbox().clone(), metasrv.table_metadata_manager().clone(), + metasrv.runtime_switch_manager().clone(), metasrv.options().grpc.server_addr.clone(), regions, manifest, diff --git a/tests-integration/tests/repartition.rs b/tests-integration/tests/repartition.rs index cf64f40aa0..d25efb9201 100644 --- a/tests-integration/tests/repartition.rs +++ b/tests-integration/tests/repartition.rs @@ -812,6 +812,7 @@ async fn trigger_table_gc(metasrv: &Arc, table_name: &str) { let procedure = BatchGcProcedure::new( metasrv.mailbox().clone(), metasrv.table_metadata_manager().clone(), + metasrv.runtime_switch_manager().clone(), metasrv.options().grpc.server_addr.clone(), region_ids.clone(), false, // full_file_listing