From 3d2b73762a6275a4bb44400e750fea4aeda87478 Mon Sep 17 00:00:00 2001 From: "B. Naut" Date: Wed, 28 Oct 2026 03:47:17 +0000 Subject: [PATCH] fix(catalog): propagate frontend process-list failures (#9999) * fix(catalog): propagate frontend process-list failures Return the existing invocation error when a selected frontend cannot list processes, so a successful response cannot omit that peer. Signed-off-by: Byte-Naut * test(frontend): cover remote process-list errors and aggregation Exercise real metadata discovery and gRPC listeners for a missing management service and successful catalog-scoped aggregation. Signed-off-by: Byte-Naut --------- Signed-off-by: Byte-Naut --- src/catalog/src/process_manager.rs | 18 ++-- tests-integration/tests/grpc.rs | 165 ++++++++++++++++++++++++++++- 2 files changed, 170 insertions(+), 13 deletions(-) diff --git a/src/catalog/src/process_manager.rs b/src/catalog/src/process_manager.rs index 17df06d7648..c0af1f10a2f 100644 --- a/src/catalog/src/process_manager.rs +++ b/src/catalog/src/process_manager.rs @@ -25,7 +25,7 @@ use common_event_recorder::EventRecorderRef; use common_frontend::selector::{FrontendSelector, MetaClientSelector}; use common_frontend::slow_query_event::SlowQueryEvent; use common_telemetry::logging::SlowQueriesRecordType; -use common_telemetry::{debug, info, slow, warn}; +use common_telemetry::{debug, info, slow}; use common_time::util::current_time_millis; use meta_client::MetaClientRef; use promql_parser::parser::EvalStmt; @@ -166,6 +166,9 @@ impl ProcessManager { Ok(result) } + /// Lists running processes from every selected frontend and the local registry. + /// + /// Returns an error if a remote frontend cannot provide its process list. pub async fn list_all_processes( &self, catalog: Option<&str>, @@ -177,20 +180,13 @@ impl ProcessManager { .await .context(error::InvokeFrontendSnafu)?; for mut f in frontends { - let result = f + let resp = f .list_process(ListProcessRequest { catalog: catalog.unwrap_or_default().to_string(), }) .await - .context(error::InvokeFrontendSnafu); - match result { - Ok(resp) => { - processes.extend(resp.processes); - } - Err(e) => { - warn!(e; "Skipping failing node: {:?}", f) - } - } + .context(error::InvokeFrontendSnafu)?; + processes.extend(resp.processes); } } processes.extend(self.local_processes(catalog)?); diff --git a/tests-integration/tests/grpc.rs b/tests-integration/tests/grpc.rs index e9cfee7e9d4..818a940d76b 100644 --- a/tests-integration/tests/grpc.rs +++ b/tests-integration/tests/grpc.rs @@ -15,6 +15,7 @@ // Retain coverage of the legacy constructors, including TLS configuration. #![allow(deprecated)] +use std::net::SocketAddr; use std::sync::Arc; use api::v1::alter_table_expr::Kind; @@ -32,18 +33,23 @@ use api::v1::{ }; use auth::user_provider_from_option; use base64::prelude::{BASE64_STANDARD, Engine as _}; +use catalog::process_manager::{ProcessManager, ProcessManagerRef}; use client::{Client, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, Database, OutputData}; use common_catalog::consts::{DEFAULT_PRIVATE_SCHEMA_NAME, MITO_ENGINE}; use common_event_recorder::DEFAULT_EVENTS_TABLE_NAME; use common_frontend::slow_query_event::SLOW_QUERY_TABLE_NAME; use common_grpc::channel_manager::ClientTlsOption; use common_memory_manager::OnExhaustedPolicy; +use common_meta::cluster::{FrontendStatus, NodeInfo, NodeInfoKey, NodeStatus, Role}; use common_meta::datanode::REGION_STATS_HISTORY_TABLE_NAME; +use common_meta::peer::Peer; +use common_meta::rpc::store::PutRequest; use common_query::Output; use common_recordbatch::RecordBatches; -use common_runtime::Runtime; use common_runtime::runtime::{BuilderBuild, RuntimeTrait}; +use common_runtime::{Builder as RuntimeBuilder, Runtime}; use common_test_util::find_workspace_path; +use common_time::util; use datatypes::arrow::array::{ Array, ArrayRef, Float64Array, Float64Builder, Int32Array, ListBuilder, StringArray, StructArray, TimestampNanosecondArray, UInt8Array, UInt16Array, UInt32Array, UInt64Array, @@ -52,6 +58,8 @@ use datatypes::arrow::array::{ use datatypes::arrow::datatypes::{DataType, Field}; use datatypes::arrow::ipc::writer::StreamWriter; use datatypes::arrow::record_batch::RecordBatch as ArrowRecordBatch; +use meta_client::client::MetaClientBuilder; +use meta_srv::mocks::{self, MockInfo}; use otel_arrow_rust::otlp::metrics::MetricType as ArrowMetricType; use otel_arrow_rust::proto::opentelemetry::arrow::v1::arrow_metrics_service_client::ArrowMetricsServiceClient; use otel_arrow_rust::proto::opentelemetry::arrow::v1::{ @@ -60,8 +68,9 @@ use otel_arrow_rust::proto::opentelemetry::arrow::v1::{ use otel_arrow_rust::proto::opentelemetry::metrics::v1::AggregationTemporality; use otel_arrow_rust::schema::consts as arrow_consts; use rstest_reuse::apply; -use servers::grpc::GrpcServerConfig; use servers::grpc::builder::GrpcServerBuilder; +use servers::grpc::frontend_grpc_handler::FrontendGrpcHandler; +use servers::grpc::{GrpcServer, GrpcServerConfig}; use servers::http::prometheus::{ PromData, PromQueryResult, PromSampleValue, PromSeriesMatrix, PromSeriesVector, PrometheusJsonResponse, PrometheusResponse, @@ -538,6 +547,158 @@ pub async fn test_grpc_auth(store_type: StorageType) { let _ = fe_grpc_server.shutdown().await; } +async fn start_process_rpc_server(process_manager: Option) -> GrpcServer { + let runtime: Runtime = RuntimeBuilder::default() + .worker_threads(2) + .thread_name("process-rpc-test") + .build() + .unwrap(); + let mut builder = GrpcServerBuilder::new(GrpcServerConfig::default(), runtime); + if let Some(process_manager) = process_manager { + builder = builder.frontend_grpc_handler(FrontendGrpcHandler::new(process_manager)); + } + let mut server = builder.build(); + server + .start("127.0.0.1:0".parse::().unwrap()) + .await + .unwrap(); + server +} + +async fn register_process_rpc_peer(mock: &MockInfo, addr: String) { + let node_info = NodeInfo { + peer: Peer { id: 1, addr }, + last_activity_ts: util::current_time_millis(), + status: NodeStatus::Frontend(FrontendStatus::default()), + version: String::new(), + git_commit: String::new(), + start_time_ms: 0, + total_cpu_millicores: 0, + total_memory_bytes: 0, + cpu_usage_millicores: 0, + memory_usage_bytes: 0, + hostname: String::new(), + env_vars: Default::default(), + schedulable: None, + }; + mock.in_memory + .as_ref() + .unwrap() + .put(PutRequest { + key: Vec::from(&NodeInfoKey { + role: Role::Frontend, + node_id: 1, + }), + value: node_info.try_into().unwrap(), + prev_kv: false, + }) + .await + .unwrap(); +} + +async fn build_process_manager_with_remote_peer( + mock: &MockInfo, + remote_addr: String, +) -> ProcessManagerRef { + register_process_rpc_peer(mock, remote_addr).await; + let mut meta_client = MetaClientBuilder::frontend_default_options() + .channel_manager(mock.channel_manager.clone()) + .build(); + meta_client.start(&[&mock.server_addr]).await.unwrap(); + Arc::new(ProcessManager::new( + "local".to_string(), + Some(Arc::new(meta_client)), + )) +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_list_all_processes_reports_unimplemented_remote_service() { + let mock = mocks::mock_with_memstore().await; + let server = start_process_rpc_server(None).await; + let remote_addr = server.bind_addr().unwrap().to_string(); + let process_manager = build_process_manager_with_remote_peer(&mock, remote_addr).await; + let _local = process_manager.register_query( + "greptime".to_string(), + vec!["public".to_string()], + "local query".to_string(), + String::new(), + None, + None, + ); + + let result = process_manager.list_all_processes(Some("greptime")).await; + server.shutdown().await.unwrap(); + + assert!( + matches!( + &result, + Err(catalog::error::Error::InvokeFrontend { + source: common_frontend::error::Error::InvokeFrontend { error, .. }, + .. + }) if error.code() == tonic::Code::Unimplemented + ), + "a missing remote management service must not return an incomplete list: {result:?}" + ); + assert_eq!( + process_manager + .local_processes(Some("greptime")) + .unwrap() + .len(), + 1 + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_list_all_processes_collects_remote_and_local_catalog() { + let mock = mocks::mock_with_memstore().await; + let remote_manager = Arc::new(ProcessManager::new("remote".to_string(), None)); + let _selected = remote_manager.register_query( + "greptime".to_string(), + vec!["public".to_string()], + "selected remote query".to_string(), + String::new(), + None, + None, + ); + let _other = remote_manager.register_query( + "other".to_string(), + vec!["public".to_string()], + "other remote query".to_string(), + String::new(), + None, + None, + ); + let server = start_process_rpc_server(Some(remote_manager)).await; + let remote_addr = server.bind_addr().unwrap().to_string(); + let process_manager = build_process_manager_with_remote_peer(&mock, remote_addr).await; + let _local = process_manager.register_query( + "greptime".to_string(), + vec!["public".to_string()], + "selected local query".to_string(), + String::new(), + None, + None, + ); + + let processes = process_manager + .list_all_processes(Some("greptime")) + .await + .unwrap(); + server.shutdown().await.unwrap(); + + assert_eq!(processes.len(), 2); + assert!( + processes + .iter() + .any(|process| process.query == "selected local query") + ); + assert!( + processes + .iter() + .any(|process| process.query == "selected remote query") + ); +} + pub async fn test_otel_arrow_auth(store_type: StorageType) { let user_provider = user_provider_from_option("static_user_provider:cmd:greptime_user=greptime_pwd").unwrap();