diff --git a/src/meta-srv/src/handler/failure_handler.rs b/src/meta-srv/src/handler/failure_handler.rs index 5f1448d3ab6..d1d287d13ff 100644 --- a/src/meta-srv/src/handler/failure_handler.rs +++ b/src/meta-srv/src/handler/failure_handler.rs @@ -52,9 +52,10 @@ impl HeartbeatHandler for RegionFailureHandler { return Ok(HandleControl::Continue); }; + // Never blocks: the region lease has already been decided earlier in the + // chain, so a busy supervisor must not withhold the heartbeat response. self.heartbeat_acceptor - .accept(DatanodeHeartbeat::from(stat)) - .await; + .accept(DatanodeHeartbeat::from(stat)); Ok(HandleControl::Continue) } @@ -62,6 +63,8 @@ impl HeartbeatHandler for RegionFailureHandler { #[cfg(test)] mod tests { + use std::time::Duration; + use api::v1::meta::HeartbeatRequest; use common_catalog::consts::default_engine; use common_meta::datanode::{RegionManifestInfo, RegionStat, Stat}; @@ -73,7 +76,45 @@ mod tests { use crate::handler::{HeartbeatAccumulator, HeartbeatHandler}; use crate::metasrv::builder::MetasrvBuilder; use crate::region::supervisor::tests::new_test_supervisor; - use crate::region::supervisor::{Event, HeartbeatAcceptor}; + use crate::region::supervisor::{DatanodeHeartbeat, Event, HeartbeatAcceptor}; + + fn new_region_stat(region_id: u64) -> RegionStat { + RegionStat { + id: RegionId::from_u64(region_id), + rcus: 0, + wcus: 0, + approximate_bytes: 0, + engine: default_engine().to_string(), + role: RegionRole::Follower, + num_rows: 0, + memtable_size: 0, + manifest_size: 0, + sst_size: 0, + sst_num: 0, + index_size: 0, + region_manifest: RegionManifestInfo::Mito { + manifest_version: 0, + flushed_entry_id: 0, + file_removed_cnt: 0, + }, + data_topic_latest_entry_id: 0, + metadata_topic_latest_entry_id: 0, + written_bytes: 0, + query_cpu_time: 0, + query_scanned_bytes: 0, + min_timestamp: None, + max_timestamp: None, + } + } + + fn new_stat() -> Stat { + Stat { + id: 42, + region_stats: vec![new_region_stat(1), new_region_stat(2), new_region_stat(3)], + timestamp_millis: 1000, + ..Default::default() + } + } #[tokio::test] async fn test_handle_heartbeat() { @@ -85,40 +126,7 @@ mod tests { let metasrv = builder.build().await.unwrap(); let mut ctx = metasrv.new_ctx(); let acc = &mut HeartbeatAccumulator::default(); - fn new_region_stat(region_id: u64) -> RegionStat { - RegionStat { - id: RegionId::from_u64(region_id), - rcus: 0, - wcus: 0, - approximate_bytes: 0, - engine: default_engine().to_string(), - role: RegionRole::Follower, - num_rows: 0, - memtable_size: 0, - manifest_size: 0, - sst_size: 0, - sst_num: 0, - index_size: 0, - region_manifest: RegionManifestInfo::Mito { - manifest_version: 0, - flushed_entry_id: 0, - file_removed_cnt: 0, - }, - data_topic_latest_entry_id: 0, - metadata_topic_latest_entry_id: 0, - written_bytes: 0, - query_cpu_time: 0, - query_scanned_bytes: 0, - min_timestamp: None, - max_timestamp: None, - } - } - acc.stat = Some(Stat { - id: 42, - region_stats: vec![new_region_stat(1), new_region_stat(2), new_region_stat(3)], - timestamp_millis: 1000, - ..Default::default() - }); + acc.stat = Some(new_stat()); handler.handle(req, &mut ctx, acc).await.unwrap(); let (tx, rx) = oneshot::channel(); @@ -126,4 +134,39 @@ mod tests { let detector = rx.await.unwrap(); assert_eq!(detector.iter().collect::>().len(), 3); } + + #[tokio::test] + async fn test_handle_heartbeat_does_not_block_on_full_supervisor_queue() { + // Regression test for #9419: a full queue used to park the handler. + // Keep the supervisor alive, or the queue reports Closed. + let (_supervisor, sender) = new_test_supervisor(); + // Built directly, so the supervisor stays undrained. + let handler = RegionFailureHandler { + heartbeat_acceptor: HeartbeatAcceptor::new(sender.clone()), + }; + + let stat = new_stat(); + let mut queued = 0; + while sender + .try_send(Event::HeartbeatArrived(DatanodeHeartbeat::from(&stat))) + .is_ok() + { + queued += 1; + } + assert_eq!(queued, sender.max_capacity()); + + let builder = MetasrvBuilder::new(); + let metasrv = builder.build().await.unwrap(); + let mut ctx = metasrv.new_ctx(); + let acc = &mut HeartbeatAccumulator::default(); + // Without a stat the handler returns before the acceptor. + acc.stat = Some(stat); + + let result = tokio::time::timeout( + Duration::from_secs(5), + handler.handle(&HeartbeatRequest::default(), &mut ctx, acc), + ) + .await; + assert!(result.is_ok()); + } } diff --git a/src/meta-srv/src/metrics.rs b/src/meta-srv/src/metrics.rs index 1cf32194302..c92e6f31fff 100644 --- a/src/meta-srv/src/metrics.rs +++ b/src/meta-srv/src/metrics.rs @@ -48,6 +48,9 @@ lazy_static! { /// Heartbeat received by metasrv. pub static ref METRIC_META_HEARTBEAT_RECV: IntCounterVec = register_int_counter_vec!("greptime_meta_heartbeat_recv", "heartbeats received by metasrv", &["pusher_key"]).unwrap(); + /// Heartbeats dropped by metasrv. + pub static ref METRIC_META_HEARTBEAT_DROPPED: IntCounterVec = + register_int_counter_vec!("greptime_meta_heartbeat_dropped", "heartbeats dropped by metasrv", &["reason"]).unwrap(); /// The migration execute histogram. pub static ref METRIC_META_REGION_MIGRATION_EXECUTE: HistogramVec = register_histogram_vec!("greptime_meta_region_migration_execute", "meta region migration execute", &["state"]).unwrap(); diff --git a/src/meta-srv/src/region/supervisor.rs b/src/meta-srv/src/region/supervisor.rs index 56a9fe4ee57..99c68ac23c8 100644 --- a/src/meta-srv/src/region/supervisor.rs +++ b/src/meta-srv/src/region/supervisor.rs @@ -35,6 +35,7 @@ use common_time::util::current_time_millis; use futures::{StreamExt, TryStreamExt}; use snafu::{ResultExt, ensure}; use store_api::storage::RegionId; +use tokio::sync::mpsc::error::TrySendError; use tokio::sync::mpsc::{Receiver, Sender}; use tokio::sync::oneshot; use tokio::time::{MissedTickBehavior, interval, interval_at}; @@ -43,6 +44,7 @@ use crate::discovery::utils::accept_ingest_workload; use crate::error::{self, Result}; use crate::failure_detector::PhiAccrualFailureDetectorOptions; use crate::metasrv::{RegionStatAwareSelectorRef, SelectTarget, SelectorContext, SelectorRef}; +use crate::metrics::METRIC_META_HEARTBEAT_DROPPED; use crate::procedure::region_migration::manager::{ RegionMigrationManagerRef, RegionMigrationTriggerReason, SubmitRegionMigrationTaskResult, }; @@ -356,10 +358,29 @@ impl HeartbeatAcceptor { Self { sender } } - /// Accepts heartbeats from datanodes. - pub(crate) async fn accept(&self, heartbeat: DatanodeHeartbeat) { - if let Err(err) = self.sender.send(Event::HeartbeatArrived(heartbeat)).await { - error!(err; "RegionSupervisor has stop receiving heartbeat."); + /// Accepts heartbeats from datanodes without waiting for queue capacity: a full queue + /// drops the heartbeat, counted in [`METRIC_META_HEARTBEAT_DROPPED`]. A dropped + /// heartbeat costs the failure detector one sample. + pub(crate) fn accept(&self, heartbeat: DatanodeHeartbeat) { + let datanode_id = heartbeat.datanode_id; + match self.sender.try_send(Event::HeartbeatArrived(heartbeat)) { + Ok(()) => {} + Err(TrySendError::Full(_)) => { + METRIC_META_HEARTBEAT_DROPPED + .with_label_values(&["full"]) + .inc(); + warn!( + "Dropping heartbeat because the region supervisor event queue is full, datanode_id: {}, queue_capacity: {}", + datanode_id, + self.sender.max_capacity() + ); + } + Err(TrySendError::Closed(err)) => { + METRIC_META_HEARTBEAT_DROPPED + .with_label_values(&["closed"]) + .inc(); + error!(err; "RegionSupervisor has stop receiving heartbeat."); + } } } } @@ -899,14 +920,15 @@ pub(crate) mod tests { use tokio::time::sleep; use super::RegionSupervisorSelector; + use crate::metrics::METRIC_META_HEARTBEAT_DROPPED; use crate::procedure::region_migration::RegionMigrationTriggerReason; use crate::procedure::region_migration::manager::{ RegionMigrationManager, SubmitRegionMigrationTaskResult, }; use crate::procedure::region_migration::test_util::TestingEnv; use crate::region::supervisor::{ - DatanodeHeartbeat, Event, RegionFailureDetectorControl, RegionSupervisor, - RegionSupervisorTicker, + DatanodeHeartbeat, Event, HeartbeatAcceptor, RegionFailureDetectorControl, + RegionSupervisor, RegionSupervisorTicker, }; use crate::selector::test_utils::{RandomNodeSelector, new_test_selector_context}; @@ -940,6 +962,14 @@ pub(crate) mod tests { ) } + fn filler_heartbeat() -> DatanodeHeartbeat { + DatanodeHeartbeat { + datanode_id: 1, + regions: vec![], + timestamp: 0, + } + } + #[tokio::test] async fn test_heartbeat() { let (mut supervisor, sender) = new_test_supervisor(); @@ -1372,4 +1402,39 @@ pub(crate) mod tests { assert!(supervisor.failure_detector.contains(&detecting_region)); assert!(supervisor.failover_counts.contains_key(&detecting_region)); } + + #[tokio::test] + async fn test_heartbeat_accept_does_not_block_on_full_queue() { + // Regression test for #9419: a full queue used to park the caller. + // The supervisor is never run, so the queue stays full. + let (mut supervisor, sender) = new_test_supervisor(); + + let mut queued = 0; + while sender + .try_send(Event::HeartbeatArrived(filler_heartbeat())) + .is_ok() + { + queued += 1; + } + assert_eq!(queued, sender.max_capacity()); + assert_eq!(sender.capacity(), 0); + + let acceptor = HeartbeatAcceptor::new(sender.clone()); + let dropped = || { + METRIC_META_HEARTBEAT_DROPPED + .with_label_values(&["full"]) + .get() + }; + let dropped_before = dropped(); + // The full queue drops the heartbeat and counts it. + acceptor.accept(filler_heartbeat()); + assert_eq!(dropped(), dropped_before + 1); + assert_eq!(sender.capacity(), 0); + + // Free one slot: the next heartbeat is accepted. + supervisor.receiver.try_recv().unwrap(); + acceptor.accept(filler_heartbeat()); + assert_eq!(dropped(), dropped_before + 1); + assert_eq!(sender.capacity(), 0); + } }