From 40c141e2b7a95154a370228fcef1836a186ff051 Mon Sep 17 00:00:00 2001 From: Nadav Gov-Ari Date: Tue, 11 Aug 2026 15:10:05 -0400 Subject: [PATCH 1/2] Remove departed ingesters from routing table --- .../src/actors/indexing_service.rs | 1 + quickwit/quickwit-ingest/src/ingest_v2/mod.rs | 9 ++++ .../quickwit-ingest/src/ingest_v2/router.rs | 27 +++++++++++- .../src/ingest_v2/routing_table.rs | 41 +++++++++++++++++++ quickwit/quickwit-serve/src/lib.rs | 28 ++++++++++--- 5 files changed, 99 insertions(+), 7 deletions(-) diff --git a/quickwit/quickwit-indexing/src/actors/indexing_service.rs b/quickwit/quickwit-indexing/src/actors/indexing_service.rs index a533d2789a0..46a29258e21 100644 --- a/quickwit/quickwit-indexing/src/actors/indexing_service.rs +++ b/quickwit/quickwit-indexing/src/actors/indexing_service.rs @@ -777,6 +777,7 @@ impl IndexingService { Ok(Some(merge_planner_mailbox)) } + //noinspection ALL /// For all Ingest V2 pipelines, assigns the set of shards they should be working on. /// This is done regardless of whether there has been a change in their shard list /// or not. diff --git a/quickwit/quickwit-ingest/src/ingest_v2/mod.rs b/quickwit/quickwit-ingest/src/ingest_v2/mod.rs index 68b3cc5f17a..7db2ea0024b 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/mod.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/mod.rs @@ -45,6 +45,7 @@ pub use broadcast::{ use bytes::buf::Writer; use bytes::{BufMut, BytesMut}; use bytesize::ByteSize; +use quickwit_common::pubsub::Event; use quickwit_common::tower::Pool; use quickwit_proto::ingest::ingester::{IngesterServiceClient, IngesterStatus}; use quickwit_proto::ingest::router::{IngestRequestV2, IngestSubrequest}; @@ -96,6 +97,14 @@ impl IngesterPoolEntry { pub type IngesterPool = Pool; +/// Published when an ingester leaves the cluster. +#[derive(Debug, Clone)] +pub struct IngesterDeparture { + pub node_id: NodeId, +} + +impl Event for IngesterDeparture {} + /// Identifies an ingester client, typically a source, for logging and debugging purposes. pub type ClientId = String; diff --git a/quickwit/quickwit-ingest/src/ingest_v2/router.rs b/quickwit/quickwit-ingest/src/ingest_v2/router.rs index ecf5f707642..b4322b7665a 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/router.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/router.rs @@ -48,7 +48,7 @@ use super::debouncing::{ use super::ingester::PERSIST_REQUEST_TIMEOUT; use super::routing_table::RoutingTable; use super::workbench::IngestWorkbench; -use super::{IngesterPool, pending_subrequests}; +use super::{IngesterDeparture, IngesterPool, pending_subrequests}; use crate::get_ingest_router_buffer_size; use crate::ingest_v2::metrics::{ INGEST_ATTEMPTS, INGEST_RESULT_CIRCUIT_BREAKER, INGEST_RESULT_INDEX_NOT_FOUND, @@ -151,7 +151,10 @@ impl IngestRouter { pub fn subscribe(&self) { let weak_router_state = WeakRouterState(Arc::downgrade(&self.state)); self.event_broker - .subscribe::(weak_router_state) + .subscribe::(weak_router_state.clone()) + .forever(); + self.event_broker + .subscribe::(weak_router_state) .forever(); } @@ -613,6 +616,26 @@ impl EventSubscriber for WeakRouterState { } } +/// Clears a departed ingester's routing entries. +#[async_trait] +impl EventSubscriber for WeakRouterState { + async fn handle_event(&mut self, departure: IngesterDeparture) { + let Some(state) = self.0.upgrade() else { + return; + }; + let mut state_guard = state.lock().await; + let num_entries = state_guard.routing_table.remove_node(&departure.node_id); + drop(state_guard); + + if num_entries > 0 { + info!( + node_id=%departure.node_id, + "removed ingester from routing table after it left the cluster" + ); + } + } +} + pub(super) struct PersistRequestSummary { pub leader_id: NodeId, pub subrequest_ids: Vec, diff --git a/quickwit/quickwit-ingest/src/ingest_v2/routing_table.rs b/quickwit/quickwit-ingest/src/ingest_v2/routing_table.rs index 973de0e582a..63cc61661d6 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/routing_table.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/routing_table.rs @@ -323,6 +323,18 @@ impl RoutingTable { } entry.seeded_from_cp = true; } + + /// Removes a node that left the cluster from every routing entry. + pub fn remove_node(&mut self, node_id: &NodeId) -> usize { + let mut num_entries = 0; + + for entry in self.table.values_mut() { + if entry.nodes.remove(node_id).is_some() { + num_entries += 1; + } + } + num_entries + } } #[cfg(test)] @@ -763,6 +775,35 @@ mod tests { assert!(entry.nodes.contains_key("node-4")); } + #[test] + fn test_remove_node() { + let mut table = RoutingTable::default(); + let index_uid = IndexUid::for_test("test-index", 0); + + table.apply_capacity_update( + NodeId::from_str("node-1"), + index_uid.clone(), + "test-source".into(), + 8, + 3, + ); + table.apply_capacity_update( + NodeId::from_str("node-2"), + index_uid, + "test-source".into(), + 6, + 2, + ); + assert_eq!(table.remove_node(&NodeId::from_str("node-1")), 1); + + let entry = table + .table + .get(&("test-index".to_string(), "test-source".to_string())) + .unwrap(); + assert!(!entry.nodes.contains_key("node-1")); + assert!(entry.nodes.contains_key("node-2")); + } + #[test] fn test_classify_az_locality() { let table = RoutingTable::new(Some("az-1".to_string())); diff --git a/quickwit/quickwit-serve/src/lib.rs b/quickwit/quickwit-serve/src/lib.rs index cacafe31423..eefca67f21f 100644 --- a/quickwit/quickwit-serve/src/lib.rs +++ b/quickwit/quickwit-serve/src/lib.rs @@ -88,11 +88,11 @@ use quickwit_indexing::actors::{IndexingService, MergeSchedulerService}; use quickwit_indexing::models::ShardPositionsService; use quickwit_indexing::{IndexingSplitCache, start_indexing_service}; use quickwit_ingest::{ - GetMemoryCapacity, IngestRequest, IngestRouter, IngestServiceClient, Ingester, IngesterPool, - IngesterPoolEntry, LocalShardsUpdate, get_idle_shard_timeout, notify_ingester_decommission, - setup_ingester_capacity_update_listener, setup_local_shards_update_listener, - start_ingest_api_service, try_get_ingester_status, wait_for_ingester_decommission, - wait_for_ingester_status, + GetMemoryCapacity, IngestRequest, IngestRouter, IngestServiceClient, Ingester, + IngesterDeparture, IngesterPool, IngesterPoolEntry, LocalShardsUpdate, get_idle_shard_timeout, + notify_ingester_decommission, setup_ingester_capacity_update_listener, + setup_local_shards_update_listener, start_ingest_api_service, try_get_ingester_status, + wait_for_ingester_decommission, wait_for_ingester_status, }; use quickwit_jaeger::JaegerService; use quickwit_janitor::{JanitorService, start_janitor_service}; @@ -1205,6 +1205,7 @@ async fn setup_ingest_v2( cluster.change_stream(), ingester_opt.clone(), ingester_pool, + event_broker.clone(), grpc_compression_encoding_opt, node_config.grpc_config.max_message_size, ); @@ -1215,11 +1216,13 @@ fn setup_ingester_pool( cluster_change_stream: ClusterChangeStream, ingester_opt: Option, ingester_pool: IngesterPool, + event_broker: EventBroker, grpc_compression_encoding_opt: Option, grpc_max_message_size: ByteSize, ) { let ingester_change_stream = cluster_change_stream.filter_map(move |cluster_change| { let ingester_opt_clone = ingester_opt.clone(); + let event_broker_clone = event_broker.clone(); Box::pin(async move { match cluster_change { ClusterChange::Add(node) if node.is_indexer() => { @@ -1254,6 +1257,9 @@ fn setup_ingester_pool( Some(change) } ClusterChange::Remove(node) if node.is_indexer() => { + event_broker_clone.publish(IngesterDeparture { + node_id: node.node_id.clone(), + }); let change = build_ingester_remove_change(&node); Some(change) } @@ -2164,10 +2170,17 @@ mod tests { let (cluster_change_stream, cluster_change_stream_tx) = ClusterChangeStream::new_unbounded(); let ingester_pool = IngesterPool::default(); + let event_broker = EventBroker::default(); + let departures: Arc>> = Arc::new(Mutex::new(Vec::new())); + let departures_clone = departures.clone(); + let _subscription = event_broker.subscribe(move |departure: IngesterDeparture| { + departures_clone.lock().unwrap().push(departure.node_id); + }); setup_ingester_pool( cluster_change_stream, None::, ingester_pool.clone(), + event_broker, None, ByteSize::mib(20), ); @@ -2249,6 +2262,11 @@ mod tests { tokio::time::sleep(Duration::from_millis(1)).await; assert!(ingester_pool.is_empty()); + // Routers are told to drop the departed node's routing entries. + assert_eq!( + *departures.lock().unwrap(), + vec![NodeId::from_str("test-ingester-node")] + ); } #[tokio::test] From 37268e24ef819050f479f70ad53165126b0f640e Mon Sep 17 00:00:00 2001 From: Nadav Gov-Ari Date: Tue, 11 Aug 2026 15:57:58 -0400 Subject: [PATCH 2/2] PR comments --- .../quickwit-indexing/src/actors/indexing_service.rs | 1 - quickwit/quickwit-ingest/src/ingest_v2/mod.rs | 4 ++-- quickwit/quickwit-ingest/src/ingest_v2/router.rs | 12 ++++++------ quickwit/quickwit-serve/src/lib.rs | 6 +++--- 4 files changed, 11 insertions(+), 12 deletions(-) diff --git a/quickwit/quickwit-indexing/src/actors/indexing_service.rs b/quickwit/quickwit-indexing/src/actors/indexing_service.rs index 46a29258e21..a533d2789a0 100644 --- a/quickwit/quickwit-indexing/src/actors/indexing_service.rs +++ b/quickwit/quickwit-indexing/src/actors/indexing_service.rs @@ -777,7 +777,6 @@ impl IndexingService { Ok(Some(merge_planner_mailbox)) } - //noinspection ALL /// For all Ingest V2 pipelines, assigns the set of shards they should be working on. /// This is done regardless of whether there has been a change in their shard list /// or not. diff --git a/quickwit/quickwit-ingest/src/ingest_v2/mod.rs b/quickwit/quickwit-ingest/src/ingest_v2/mod.rs index 7db2ea0024b..f366278be60 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/mod.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/mod.rs @@ -99,11 +99,11 @@ pub type IngesterPool = Pool; /// Published when an ingester leaves the cluster. #[derive(Debug, Clone)] -pub struct IngesterDeparture { +pub struct IngesterLeft { pub node_id: NodeId, } -impl Event for IngesterDeparture {} +impl Event for IngesterLeft {} /// Identifies an ingester client, typically a source, for logging and debugging purposes. pub type ClientId = String; diff --git a/quickwit/quickwit-ingest/src/ingest_v2/router.rs b/quickwit/quickwit-ingest/src/ingest_v2/router.rs index b4322b7665a..f7074879153 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/router.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/router.rs @@ -48,7 +48,7 @@ use super::debouncing::{ use super::ingester::PERSIST_REQUEST_TIMEOUT; use super::routing_table::RoutingTable; use super::workbench::IngestWorkbench; -use super::{IngesterDeparture, IngesterPool, pending_subrequests}; +use super::{IngesterLeft, IngesterPool, pending_subrequests}; use crate::get_ingest_router_buffer_size; use crate::ingest_v2::metrics::{ INGEST_ATTEMPTS, INGEST_RESULT_CIRCUIT_BREAKER, INGEST_RESULT_INDEX_NOT_FOUND, @@ -154,7 +154,7 @@ impl IngestRouter { .subscribe::(weak_router_state.clone()) .forever(); self.event_broker - .subscribe::(weak_router_state) + .subscribe::(weak_router_state) .forever(); } @@ -618,18 +618,18 @@ impl EventSubscriber for WeakRouterState { /// Clears a departed ingester's routing entries. #[async_trait] -impl EventSubscriber for WeakRouterState { - async fn handle_event(&mut self, departure: IngesterDeparture) { +impl EventSubscriber for WeakRouterState { + async fn handle_event(&mut self, departed_ingester: IngesterLeft) { let Some(state) = self.0.upgrade() else { return; }; let mut state_guard = state.lock().await; - let num_entries = state_guard.routing_table.remove_node(&departure.node_id); + let num_entries = state_guard.routing_table.remove_node(&departed_ingester.node_id); drop(state_guard); if num_entries > 0 { info!( - node_id=%departure.node_id, + node_id=%departed_ingester.node_id, "removed ingester from routing table after it left the cluster" ); } diff --git a/quickwit/quickwit-serve/src/lib.rs b/quickwit/quickwit-serve/src/lib.rs index eefca67f21f..46d45fd0e25 100644 --- a/quickwit/quickwit-serve/src/lib.rs +++ b/quickwit/quickwit-serve/src/lib.rs @@ -89,7 +89,7 @@ use quickwit_indexing::models::ShardPositionsService; use quickwit_indexing::{IndexingSplitCache, start_indexing_service}; use quickwit_ingest::{ GetMemoryCapacity, IngestRequest, IngestRouter, IngestServiceClient, Ingester, - IngesterDeparture, IngesterPool, IngesterPoolEntry, LocalShardsUpdate, get_idle_shard_timeout, + IngesterLeft, IngesterPool, IngesterPoolEntry, LocalShardsUpdate, get_idle_shard_timeout, notify_ingester_decommission, setup_ingester_capacity_update_listener, setup_local_shards_update_listener, start_ingest_api_service, try_get_ingester_status, wait_for_ingester_decommission, wait_for_ingester_status, @@ -1257,7 +1257,7 @@ fn setup_ingester_pool( Some(change) } ClusterChange::Remove(node) if node.is_indexer() => { - event_broker_clone.publish(IngesterDeparture { + event_broker_clone.publish(IngesterLeft { node_id: node.node_id.clone(), }); let change = build_ingester_remove_change(&node); @@ -2173,7 +2173,7 @@ mod tests { let event_broker = EventBroker::default(); let departures: Arc>> = Arc::new(Mutex::new(Vec::new())); let departures_clone = departures.clone(); - let _subscription = event_broker.subscribe(move |departure: IngesterDeparture| { + let _subscription = event_broker.subscribe(move |departure: IngesterLeft| { departures_clone.lock().unwrap().push(departure.node_id); }); setup_ingester_pool(