From 30471ce2d783de5e5650c5801195b7cc1f08effb Mon Sep 17 00:00:00 2001 From: Luca Cominardi Date: Fri, 7 Aug 2026 10:07:01 +0200 Subject: [PATCH 1/2] Gate indexer readiness on control plane --- .../src/tests/no_cp_tests.rs | 57 +++++++++++++ quickwit/quickwit-proto/build.rs | 1 + .../quickwit/quickwit.control_plane.rs | 50 +++++++++++ quickwit/quickwit-serve/src/lib.rs | 82 ++++++++++++++++++- 4 files changed, 187 insertions(+), 3 deletions(-) diff --git a/quickwit/quickwit-integration-tests/src/tests/no_cp_tests.rs b/quickwit/quickwit-integration-tests/src/tests/no_cp_tests.rs index ccc8d2fa77f..5bed9b6a491 100644 --- a/quickwit/quickwit-integration-tests/src/tests/no_cp_tests.rs +++ b/quickwit/quickwit-integration-tests/src/tests/no_cp_tests.rs @@ -14,10 +14,14 @@ //! Tests for cluster configurations without a control plane. +use std::time::Duration; + +use quickwit_common::test_utils::wait_until_predicate; use quickwit_config::ConfigFormat; use quickwit_config::service::QuickwitService; use quickwit_rest_client::error::{ApiError, Error as RestClientError}; use quickwit_serve::SearchRequestQueryString; +use reqwest::StatusCode; use crate::test_utils::ClusterSandboxBuilder; @@ -70,6 +74,59 @@ async fn test_search_after_control_plane_shutdown() { sandbox.shutdown().await.unwrap(); } +#[tokio::test] +async fn test_indexer_readiness_decreases_after_control_plane_shutdown() { + initialize_tests(); + let mut sandbox = ClusterSandboxBuilder::default() + .add_node([QuickwitService::ControlPlane]) + .add_node([QuickwitService::Metastore]) + .add_node([QuickwitService::Indexer]) + .build_and_start() + .await; + let indexer_rest_addr = sandbox + .node_configs + .iter() + .find(|(_, services)| services.contains(&QuickwitService::Indexer)) + .map(|(config, _)| config.rest_config.listen_addr) + .unwrap(); + let indexer_readyz_url = format!("http://{indexer_rest_addr}/health/readyz"); + + let indexer_readyz_response = reqwest::get(&indexer_readyz_url).await.unwrap(); + assert_eq!(indexer_readyz_response.status(), StatusCode::OK); + assert_eq!(indexer_readyz_response.text().await.unwrap(), "true"); + + sandbox + .shutdown_services([QuickwitService::ControlPlane]) + .await + .unwrap(); + assert!( + sandbox + .rest_client(QuickwitService::Metastore) + .node_health() + .is_ready() + .await + .unwrap() + ); + + wait_until_predicate( + || { + let indexer_readyz_url = indexer_readyz_url.clone(); + async move { + match reqwest::get(&indexer_readyz_url).await { + Ok(response) => response.status() == StatusCode::SERVICE_UNAVAILABLE, + Err(_) => false, + } + } + }, + Duration::from_secs(10), + Duration::from_millis(100), + ) + .await + .unwrap(); + + sandbox.shutdown().await.unwrap(); +} + #[tokio::test] async fn test_searcher_and_metastore_without_control_plane() { initialize_tests(); diff --git a/quickwit/quickwit-proto/build.rs b/quickwit/quickwit-proto/build.rs index 5d51c8cd3f9..173e31f0c79 100644 --- a/quickwit/quickwit-proto/build.rs +++ b/quickwit/quickwit-proto/build.rs @@ -69,6 +69,7 @@ fn main() -> Result<(), Box> { .with_output_dir("src/codegen/quickwit") .with_result_type_path("crate::control_plane::ControlPlaneResult") .with_error_type_path("crate::control_plane::ControlPlaneError") + .generate_extra_service_methods() .run() .unwrap(); diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.control_plane.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.control_plane.rs index a2bcc2af275..d1d8c9d399a 100644 --- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.control_plane.rs +++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.control_plane.rs @@ -180,6 +180,8 @@ pub trait ControlPlaneService: std::fmt::Debug + Send + Sync + 'static { &self, request: super::metastore::PruneShardsRequest, ) -> crate::control_plane::ControlPlaneResult; + async fn check_connectivity(&self) -> anyhow::Result<()>; + fn endpoints(&self) -> Vec; } #[derive(Debug, Clone)] pub struct ControlPlaneServiceClient { @@ -362,6 +364,12 @@ impl ControlPlaneService for ControlPlaneServiceClient { ) -> crate::control_plane::ControlPlaneResult { self.inner.0.prune_shards(request).await } + async fn check_connectivity(&self) -> anyhow::Result<()> { + self.inner.0.check_connectivity().await + } + fn endpoints(&self) -> Vec { + self.inner.0.endpoints() + } } #[cfg(any(test, feature = "testsuite"))] pub mod mock_control_plane_service { @@ -450,6 +458,12 @@ pub mod mock_control_plane_service { > { self.inner.lock().await.prune_shards(request).await } + async fn check_connectivity(&self) -> anyhow::Result<()> { + self.inner.lock().await.check_connectivity().await + } + fn endpoints(&self) -> Vec { + futures::executor::block_on(self.inner.lock()).endpoints() + } } } pub type BoxFuture = std::pin::Pin< @@ -745,6 +759,12 @@ impl ControlPlaneService for ControlPlaneServiceTowerServiceStack { ) -> crate::control_plane::ControlPlaneResult { self.prune_shards_svc.clone().ready().await?.call(request).await } + async fn check_connectivity(&self) -> anyhow::Result<()> { + self.inner.0.check_connectivity().await + } + fn endpoints(&self) -> Vec { + self.inner.0.endpoints() + } } type CreateIndexLayer = quickwit_common::tower::BoxLayer< quickwit_common::tower::BoxService< @@ -1749,6 +1769,18 @@ where ) -> crate::control_plane::ControlPlaneResult { self.clone().call(request).await } + async fn check_connectivity(&self) -> anyhow::Result<()> { + if self.inner.is_disconnected() { + anyhow::bail!("actor `{}` is disconnected", self.inner.actor_instance_id()) + } + Ok(()) + } + fn endpoints(&self) -> Vec { + vec![ + quickwit_common::uri::Uri::from_str(& format!("actor://localhost/{}", self + .inner.actor_instance_id())).expect("URI should be valid") + ] + } } #[derive(Debug, Clone)] pub struct ControlPlaneServiceGrpcClientAdapter { @@ -1968,6 +2000,24 @@ where super::metastore::PruneShardsRequest::rpc_name(), )) } + async fn check_connectivity(&self) -> anyhow::Result<()> { + if self.connection_addrs_rx.borrow().is_empty() { + anyhow::bail!("no server currently available") + } + Ok(()) + } + fn endpoints(&self) -> Vec { + self.connection_addrs_rx + .borrow() + .iter() + .flat_map(|addr| quickwit_common::uri::Uri::from_str( + &format!( + "grpc://{addr}/{}.{}", "quickwit.control_plane", + "ControlPlaneService" + ), + )) + .collect() + } } #[derive(Debug)] pub struct ControlPlaneServiceGrpcServerAdapter { diff --git a/quickwit/quickwit-serve/src/lib.rs b/quickwit/quickwit-serve/src/lib.rs index cacafe31423..630283bac3f 100644 --- a/quickwit/quickwit-serve/src/lib.rs +++ b/quickwit/quickwit-serve/src/lib.rs @@ -101,7 +101,7 @@ use quickwit_metastore::{ }; use quickwit_opentelemetry::otlp::{OtlpGrpcLogsService, OtlpGrpcTracesService}; use quickwit_proto::compaction::CompactionPlannerServiceClient; -use quickwit_proto::control_plane::ControlPlaneServiceClient; +use quickwit_proto::control_plane::{ControlPlaneService, ControlPlaneServiceClient}; use quickwit_proto::indexing::{IndexingServiceClient, ShardPositionsUpdate}; use quickwit_proto::ingest::ingester::{ IngesterService, IngesterServiceClient, IngesterServiceTowerLayerStack, IngesterStatus, @@ -518,6 +518,10 @@ async fn shutdown_signal_handler( cluster: Cluster, ) -> HashMap { shutdown_signal.await; + #[cfg(any(test, feature = "testsuite"))] + cluster.leave().await; + #[cfg(not(any(test, feature = "testsuite")))] + cluster.set_self_node_readiness(false).await; if let Err(error) = notify_ingester_decommission(ingester_opt.as_ref()).await { error!("failed to initiate ingester decommission: {:?}", error); } @@ -1017,12 +1021,21 @@ pub async fn serve_quickwit( // Node readiness indicates that the server is ready to receive requests. // Thus readiness task is started once gRPC and REST servers are started. + let control_plane_readiness_opt = if quickwit_services + .node_config + .is_service_enabled(QuickwitService::Indexer) + { + Some(quickwit_services.control_plane_client.clone()) + } else { + None + }; spawn_named_task( node_readiness_reporting_task( cluster.clone(), primary_metastore_through_control_plane, read_replica_metastore_client_opt, ingester_opt.clone(), + control_plane_readiness_opt, grpc_readiness_signal_rx, rest_readiness_signal_rx, health_reporter, @@ -1581,11 +1594,13 @@ fn with_arg(arg: T) -> impl Filter, ingester_opt: Option, + control_plane_opt: Option, grpc_readiness_signal_rx: oneshot::Receiver<()>, rest_readiness_signal_rx: oneshot::Receiver<()>, health_reporter: HealthReporter, @@ -1650,7 +1665,28 @@ async fn node_readiness_reporting_task( } else { true }; - let is_ready = metastore_available && ingester_is_available; + let control_plane_is_available = if let Some(control_plane) = &control_plane_opt { + match control_plane.check_connectivity().await { + Ok(()) => { + debug!( + control_plane_endpoints=?control_plane.endpoints(), + "control plane service is available" + ); + true + } + Err(error) => { + debug!( + control_plane_endpoints=?control_plane.endpoints(), + error=?error, + "control plane service is unavailable" + ); + false + } + } + } else { + true + }; + let is_ready = metastore_available && ingester_is_available && control_plane_is_available; let new_node_ready = if is_ready { consecutive_readiness_failures = 0; true @@ -1670,6 +1706,14 @@ async fn node_readiness_reporting_task( "metastore unavailability caused node readiness to decrease" ); } + if !node_ready && !control_plane_is_available { + warn!( + control_plane_endpoints = ?control_plane_opt + .as_ref() + .map(ControlPlaneServiceClient::endpoints), + "control plane unavailability caused node readiness to decrease" + ); + } cluster.set_self_node_readiness(node_ready).await; let serving_status = if node_ready { @@ -1734,6 +1778,7 @@ mod tests { use anyhow::{bail, ensure}; use quickwit_cluster::{ChitchatTransport, ClusterNode, create_cluster_for_test}; + use quickwit_common::tower::{BalanceChannel, Change}; use quickwit_common::uri::Uri; use quickwit_common::{ServiceStream, assert_eventually}; use quickwit_config::SearcherConfig; @@ -1744,7 +1789,7 @@ mod tests { use quickwit_proto::types::{IndexUid, PipelineUid}; use quickwit_search::Job; use tokio::sync::watch; - use tonic::transport::{Channel, Server}; + use tonic::transport::{Channel, Endpoint, Server}; use tonic_health::pb::HealthCheckRequest; use tonic_health::pb::health_client::HealthClient; use tonic_health::server::health_reporter; @@ -1840,6 +1885,12 @@ mod tests { }); Ok(observation_stream) }); + let (control_plane_balance_channel, control_plane_change_tx) = BalanceChannel::new(); + let control_plane = ControlPlaneServiceClient::from_balance_channel( + control_plane_balance_channel, + ByteSize::mib(20), + None, + ); let (grpc_readiness_trigger_tx, grpc_readiness_signal_rx) = oneshot::channel(); let (rest_readiness_trigger_tx, rest_readiness_signal_rx) = oneshot::channel(); @@ -1869,6 +1920,7 @@ mod tests { mock_metastore, None, Some(mock_ingester), + Some(control_plane), grpc_readiness_signal_rx, rest_readiness_signal_rx, health_reporter, @@ -1885,6 +1937,14 @@ mod tests { metastore_readiness_tx.send(true).unwrap(); ingester_status_tx.send(IngesterStatus::Ready).unwrap(); + tokio::time::sleep(READINESS_REPORTING_INTERVAL * 3).await; + assert!(!cluster.is_self_node_ready().await); + + let control_plane_addr = "127.0.0.1:10000".parse().unwrap(); + let control_plane_channel = Endpoint::from_static("http://127.0.0.1:10000").connect_lazy(); + control_plane_change_tx + .send(Change::Insert(control_plane_addr, control_plane_channel)) + .unwrap(); assert_eventually!(cluster.is_self_node_ready().await); let request = tonic::Request::new(HealthCheckRequest::default()); @@ -1897,6 +1957,21 @@ mod tests { tokio::time::sleep(READINESS_REPORTING_INTERVAL * READINESS_FAILURE_THRESHOLD as u32).await; assert!(cluster.is_self_node_ready().await); + control_plane_change_tx + .send(Change::Remove(control_plane_addr)) + .unwrap(); + assert_eventually!(!cluster.is_self_node_ready().await); + + let request = tonic::Request::new(HealthCheckRequest::default()); + let response = health_client.check(request).await.unwrap().into_inner(); + assert_eq!(response.status(), ServingStatus::NotServing.into()); + + let control_plane_channel = Endpoint::from_static("http://127.0.0.1:10000").connect_lazy(); + control_plane_change_tx + .send(Change::Insert(control_plane_addr, control_plane_channel)) + .unwrap(); + assert_eventually!(cluster.is_self_node_ready().await); + metastore_readiness_tx.send(false).unwrap(); assert_eventually!(!cluster.is_self_node_ready().await); @@ -1926,6 +2001,7 @@ mod tests { primary_metastore, Some(replica_metastore), None::, + None, grpc_readiness_signal_rx, rest_readiness_signal_rx, health_reporter, From 3e99d49abf86a26de16d14450fec98a0fe7c1ccc Mon Sep 17 00:00:00 2001 From: Luca Cominardi Date: Fri, 7 Aug 2026 10:57:43 +0200 Subject: [PATCH 2/2] Order control plane readiness before ingester --- quickwit/quickwit-serve/src/lib.rs | 38 +++++++++++++++--------------- 1 file changed, 19 insertions(+), 19 deletions(-) diff --git a/quickwit/quickwit-serve/src/lib.rs b/quickwit/quickwit-serve/src/lib.rs index 630283bac3f..4adc985e779 100644 --- a/quickwit/quickwit-serve/src/lib.rs +++ b/quickwit/quickwit-serve/src/lib.rs @@ -1034,8 +1034,8 @@ pub async fn serve_quickwit( cluster.clone(), primary_metastore_through_control_plane, read_replica_metastore_client_opt, - ingester_opt.clone(), control_plane_readiness_opt, + ingester_opt.clone(), grpc_readiness_signal_rx, rest_readiness_signal_rx, health_reporter, @@ -1599,8 +1599,8 @@ async fn node_readiness_reporting_task( cluster: Cluster, primary_metastore: MetastoreServiceClient, read_replica_metastore_opt: Option, - ingester_opt: Option, control_plane_opt: Option, + ingester_opt: Option, grpc_readiness_signal_rx: oneshot::Receiver<()>, rest_readiness_signal_rx: oneshot::Receiver<()>, health_reporter: HealthReporter, @@ -1650,21 +1650,6 @@ async fn node_readiness_reporting_task( false } }; - let ingester_is_available = if let Some(ingester) = &ingester_opt { - match try_get_ingester_status(ingester).await { - Ok(status) => { - status == IngesterStatus::Initializing || status != IngesterStatus::Failed - } - Err(error) => { - // If we couldn't get the ingester status, it's not looking good, so we set the - // node to not ready. - error!(%error, "failed to get ingester status"); - false - } - } - } else { - true - }; let control_plane_is_available = if let Some(control_plane) = &control_plane_opt { match control_plane.check_connectivity().await { Ok(()) => { @@ -1686,6 +1671,21 @@ async fn node_readiness_reporting_task( } else { true }; + let ingester_is_available = if let Some(ingester) = &ingester_opt { + match try_get_ingester_status(ingester).await { + Ok(status) => { + status == IngesterStatus::Initializing || status != IngesterStatus::Failed + } + Err(error) => { + // If we couldn't get the ingester status, it's not looking good, so we set the + // node to not ready. + error!(%error, "failed to get ingester status"); + false + } + } + } else { + true + }; let is_ready = metastore_available && ingester_is_available && control_plane_is_available; let new_node_ready = if is_ready { consecutive_readiness_failures = 0; @@ -1919,8 +1919,8 @@ mod tests { cluster.clone(), mock_metastore, None, - Some(mock_ingester), Some(control_plane), + Some(mock_ingester), grpc_readiness_signal_rx, rest_readiness_signal_rx, health_reporter, @@ -2000,8 +2000,8 @@ mod tests { cluster.clone(), primary_metastore, Some(replica_metastore), - None::, None, + None::, grpc_readiness_signal_rx, rest_readiness_signal_rx, health_reporter,