mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-01-04 12:22:55 +00:00
* feat/kill-process: ### Add Cancellation Support and Enhance Process Management - **Cancellation Handle Implementation**: Introduced `CancellationHandle` in `cancellation_handle.rs` to facilitate cancellation of futures and streams. - **Process Management Enhancements**: - Updated `ProcessManager` in `process_manager.rs` to support cancellable processes using `CancellableProcess`. - Added `kill_process` method for terminating processes. - **Stream Wrapper Update**: - Replaced `StreamWrapper` with `CancellableStreamWrapper` in `stream_wrapper.rs` and `instance.rs` to handle stream cancellation. - **Error Handling**: - Added `StreamCancelled` error variant in `error.rs` to handle stream cancellation scenarios. - **gRPC Handler Update**: - Added `kill_process` gRPC method in `frontend_grpc_handler.rs` to allow external process termination. - **Dependency Updates**: - Updated `Cargo.lock` and `Cargo.toml` to include `common-base` and `tokio-util`. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * feat/kill-process: **Enhancements and Bug Fixes** - **Dependency Update**: Updated `greptime-proto` dependency in `Cargo.lock` and `Cargo.toml` to a new revision. - **Error Handling Improvements**: - Modified error variants in `src/catalog/src/error.rs` and `src/common/frontend/src/error.rs` to improve error messages and handling. - Added `FrontendNotFound` error variant for better error specificity. - **Process Management Enhancements**: - Updated `ProcessManager` in `src/catalog/src/process_manager.rs` to include `kill_process` functionality with server address validation. - Enhanced `FrontendClient` trait in `src/common/frontend/src/selector.rs` to support `kill_process` requests. - **gRPC Handler Update**: - Refactored `FrontendGrpcHandler` in `src/servers/src/grpc/frontend_grpc_handler.rs` to handle `kill_process` requests asynchronously and return process status. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * feat/kill-process: ### Add Kill Process Functionality - **`Cargo.lock`, `Cargo.toml`**: Added `common-frontend` as a dependency. - **`server.rs`, `builder.rs`, `instance.rs`**: Updated `FrontendInvoker` and `FrontendBuilder` to support process management. - **`error.rs`**: Introduced `InvalidProcessId` error for handling invalid process IDs. - **`statement.rs`, `kill.rs`**: Implemented `execute_kill` method in `StatementExecutor` to handle the `KILL` statement. - **`parser.rs`, `statement.rs`**: Updated SQL parser to recognize and parse the `KILL` statement. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * feat/kill-process: ## Add Cancellation Support to Query Execution - **`process_manager.rs`**: Updated `CancellationHandle` initialization to use `default()` method. - **`cancellation_handle.rs`**: Implemented `Debug` trait for `CancellationHandle` and added `Cancellation` and `CancellableFuture` structs to support cancellable futures. - **`error.rs`**: Introduced `Cancelled` error variant to handle query cancellations. - **`instance.rs`**: Integrated `CancellableFuture` to manage query execution with cancellation support. - **`stream_wrapper.rs`**: Modified `CancellableStreamWrapper` to use the new `waker()` method for cancellation handling. - **`statement.rs`**: Added `#[allow(clippy::too_many_arguments)]` to `StatementExecutor::new` to suppress clippy warnings. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * feat/kill-process: - **Add `MetaClientMissing` Error**: Introduced a new error variant `MetaClientMissing` in `error.rs` to handle missing meta client scenarios. - **Refactor Cancellation Handling**: Merged `cancellation_handle.rs` into `cancellation.rs` and updated related logic in `process_manager.rs`, `instance.rs`, and `stream_wrapper.rs`. - **Enhance Process Management**: Improved process management logic in `process_manager.rs` to handle process cancellation more effectively. - **Update Tests**: Added and updated tests in `cancellation.rs` and `stream_wrapper.rs` to cover new cancellation logic and error handling. - **Cargo.toml Update**: Adjusted workspace settings in `Cargo.toml` for `common-frontend`. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * feat/kill-process: - **Add Tests for Process Management**: Introduced multiple async tests in `process_manager.rs` to verify query registration, deregistration, cancellation, and process killing functionalities. - **Update Error Message in SQL Parser**: Modified the expected error message in `parser.rs` to clarify the expected token as a "process id string literal". Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * feat/kill-process: ### Add Process Count Metrics to Catalog - **`metrics.rs`**: Introduced a new metric `PROCESS_LIST_COUNT` to track the count of running processes per catalog using `IntGaugeVec`. - **`process_manager.rs`**: Updated `CancellableProcess` to increment and decrement `PROCESS_LIST_COUNT` upon creation and destruction, respectively. Added a `Drop` implementation for `CancellableProcess` to handle metric updates. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * feat/kill-process: ### Fix process removal logic in `process_manager.rs` - Corrected the condition for removing an entry from the catalog in `ProcessManager` by using `o.get()` instead of `o.get_mut()`. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * feat/kill-process: - **Error Handling Improvements**: - Updated status codes for `Error::FrontendNotFound` and `Error::MetaClientMissing` to `StatusCode::Unexpected` in `src/catalog/src/error.rs`. - Changed `InvokeFrontend` error display message and status code in `src/common/frontend/src/error.rs`. - Added `ProcessManagerMissing` error in `src/operator/src/error.rs` and updated its handling in `src/operator/src/statement/kill.rs`. - **Process Management Enhancements**: - Added documentation for `ProcessManager` and `register_query` in `src/catalog/src/process_manager.rs`. - Modified `kill_process` response handling in `src/servers/src/grpc/frontend_grpc_handler.rs`. - **Cancellation Logic Update**: - Improved cancellation logic in `src/common/base/src/cancellation.rs` to use `compare_exchange` for atomic operations. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * feat/kill-process: ### Add Process Kill Count Metric and Refactor Cancellation Handle - **Metrics Update**: Added a new metric `PROCESS_KILL_COUNT` in `metrics.rs` to track the count of completed kill process requests per catalog. - **Refactor Cancellation Handle**: Renamed `cancellation_handler` to `cancellation_handle` across multiple files for consistency: - `process_manager.rs` - `instance.rs` - `stream_wrapper.rs` - **Process Management**: Updated process management logic in `process_manager.rs` to increment the `PROCESS_KILL_COUNT` metric upon successful process termination. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * feat/kill-process: Update metric description in `metrics.rs` - Changed the description of `PROCESS_KILL_COUNT` to reflect the count of killed processes instead of running processes in `metrics.rs`. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> * feat/kill-process: Update `greptime-proto` Dependency and Fix Response Field - **Updated Dependency**: Changed the `greptime-proto` Git revision in `Cargo.lock` and `Cargo.toml` to `f0913f1`. - **Code Fix**: Modified `frontend_grpc_handler.rs` to correct the response field from `found` to `success` in `KillProcessResponse`. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com> --------- Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
407 lines
11 KiB
Rust
407 lines
11 KiB
Rust
// Copyright 2023 Greptime Team
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
use std::any::Any;
|
|
use std::fmt::Debug;
|
|
|
|
use common_error::ext::{BoxedError, ErrorExt};
|
|
use common_error::status_code::StatusCode;
|
|
use common_macro::stack_trace_debug;
|
|
use common_query::error::datafusion_status_code;
|
|
use datafusion::error::DataFusionError;
|
|
use snafu::{Location, Snafu};
|
|
|
|
#[derive(Snafu)]
|
|
#[snafu(visibility(pub))]
|
|
#[stack_trace_debug]
|
|
pub enum Error {
|
|
#[snafu(display("Failed to list catalogs"))]
|
|
ListCatalogs {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
source: BoxedError,
|
|
},
|
|
|
|
#[snafu(display("Failed to list {}'s schemas", catalog))]
|
|
ListSchemas {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
catalog: String,
|
|
source: BoxedError,
|
|
},
|
|
|
|
#[snafu(display("Failed to list {}.{}'s tables", catalog, schema))]
|
|
ListTables {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
catalog: String,
|
|
schema: String,
|
|
source: BoxedError,
|
|
},
|
|
|
|
#[snafu(display("Failed to list nodes in cluster"))]
|
|
ListNodes {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
source: BoxedError,
|
|
},
|
|
|
|
#[snafu(display("Failed to region stats in cluster"))]
|
|
ListRegionStats {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
source: BoxedError,
|
|
},
|
|
|
|
#[snafu(display("Failed to list flow stats"))]
|
|
ListFlowStats {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
source: BoxedError,
|
|
},
|
|
|
|
#[snafu(display("Failed to list flows in catalog {catalog}"))]
|
|
ListFlows {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
catalog: String,
|
|
source: BoxedError,
|
|
},
|
|
|
|
#[snafu(display("Flow info not found: {flow_name} in catalog {catalog_name}"))]
|
|
FlowInfoNotFound {
|
|
flow_name: String,
|
|
catalog_name: String,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Can't convert value to json, input={input}"))]
|
|
Json {
|
|
input: String,
|
|
#[snafu(source)]
|
|
error: serde_json::error::Error,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Failed to get information extension client"))]
|
|
GetInformationExtension {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Failed to list procedures"))]
|
|
ListProcedures {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
source: BoxedError,
|
|
},
|
|
|
|
#[snafu(display("Procedure id not found"))]
|
|
ProcedureIdNotFound {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("convert proto data error"))]
|
|
ConvertProtoData {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
source: BoxedError,
|
|
},
|
|
|
|
#[snafu(display("Failed to create table, table info: {}", table_info))]
|
|
CreateTable {
|
|
table_info: String,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
source: table::error::Error,
|
|
},
|
|
|
|
#[snafu(display("Cannot find catalog by name: {}", catalog_name))]
|
|
CatalogNotFound {
|
|
catalog_name: String,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Cannot find schema {} in catalog {}", schema, catalog))]
|
|
SchemaNotFound {
|
|
catalog: String,
|
|
schema: String,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Table `{}` already exists", table))]
|
|
TableExists {
|
|
table: String,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Table not found: {}", table))]
|
|
TableNotExist {
|
|
table: String,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("View info not found: {}", name))]
|
|
ViewInfoNotFound {
|
|
name: String,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display(
|
|
"View plan columns changed from: {} to: {}",
|
|
origin_names,
|
|
actual_names
|
|
))]
|
|
ViewPlanColumnsChanged {
|
|
origin_names: String,
|
|
actual_names: String,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Partition manager not found, it's not expected."))]
|
|
PartitionManagerNotFound {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Failed to find table partitions"))]
|
|
FindPartitions { source: partition::error::Error },
|
|
|
|
#[snafu(display("Failed to find region routes"))]
|
|
FindRegionRoutes { source: partition::error::Error },
|
|
|
|
#[snafu(display("Failed to create recordbatch"))]
|
|
CreateRecordBatch {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
source: common_recordbatch::error::Error,
|
|
},
|
|
|
|
#[snafu(display("Internal error"))]
|
|
Internal {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
source: BoxedError,
|
|
},
|
|
|
|
#[snafu(display("Failed to upgrade weak catalog manager reference"))]
|
|
UpgradeWeakCatalogManagerRef {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Failed to decode logical plan for view: {}", name))]
|
|
DecodePlan {
|
|
name: String,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
source: common_query::error::Error,
|
|
},
|
|
|
|
#[snafu(display("Invalid table info in catalog"))]
|
|
InvalidTableInfoInCatalog {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
source: datatypes::error::Error,
|
|
},
|
|
|
|
#[snafu(display("Illegal access to catalog: {} and schema: {}", catalog, schema))]
|
|
QueryAccessDenied { catalog: String, schema: String },
|
|
|
|
#[snafu(display("DataFusion error"))]
|
|
Datafusion {
|
|
#[snafu(source)]
|
|
error: DataFusionError,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Failed to project view columns"))]
|
|
ProjectViewColumns {
|
|
#[snafu(source)]
|
|
error: DataFusionError,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Table metadata manager error"))]
|
|
TableMetadataManager {
|
|
source: common_meta::error::Error,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Failed to get table cache"))]
|
|
GetTableCache {
|
|
source: common_meta::error::Error,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Failed to get view info from cache"))]
|
|
GetViewCache {
|
|
source: common_meta::error::Error,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Cache not found: {name}"))]
|
|
CacheNotFound {
|
|
name: String,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Failed to cast the catalog manager"))]
|
|
CastManager {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Failed to invoke frontend services"))]
|
|
InvokeFrontend {
|
|
source: common_frontend::error::Error,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Meta client is not provided"))]
|
|
MetaClientMissing {
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
|
|
#[snafu(display("Failed to find frontend node: {}", addr))]
|
|
FrontendNotFound {
|
|
addr: String,
|
|
#[snafu(implicit)]
|
|
location: Location,
|
|
},
|
|
}
|
|
|
|
impl Error {
|
|
pub fn should_fail(&self) -> bool {
|
|
use Error::*;
|
|
|
|
matches!(
|
|
self,
|
|
GetViewCache { .. }
|
|
| ViewInfoNotFound { .. }
|
|
| DecodePlan { .. }
|
|
| ViewPlanColumnsChanged { .. }
|
|
| ProjectViewColumns { .. }
|
|
)
|
|
}
|
|
}
|
|
|
|
pub type Result<T> = std::result::Result<T, Error>;
|
|
|
|
impl ErrorExt for Error {
|
|
fn status_code(&self) -> StatusCode {
|
|
match self {
|
|
Error::SchemaNotFound { .. }
|
|
| Error::CatalogNotFound { .. }
|
|
| Error::FindPartitions { .. }
|
|
| Error::FindRegionRoutes { .. }
|
|
| Error::CacheNotFound { .. }
|
|
| Error::CastManager { .. }
|
|
| Error::Json { .. }
|
|
| Error::GetInformationExtension { .. }
|
|
| Error::PartitionManagerNotFound { .. }
|
|
| Error::ProcedureIdNotFound { .. } => StatusCode::Unexpected,
|
|
|
|
Error::ViewPlanColumnsChanged { .. } => StatusCode::InvalidArguments,
|
|
|
|
Error::ViewInfoNotFound { .. } => StatusCode::TableNotFound,
|
|
|
|
Error::FlowInfoNotFound { .. } => StatusCode::FlowNotFound,
|
|
|
|
Error::UpgradeWeakCatalogManagerRef { .. } => StatusCode::Internal,
|
|
|
|
Error::CreateRecordBatch { source, .. } => source.status_code(),
|
|
Error::TableExists { .. } => StatusCode::TableAlreadyExists,
|
|
Error::TableNotExist { .. } => StatusCode::TableNotFound,
|
|
Error::ListCatalogs { source, .. }
|
|
| Error::ListNodes { source, .. }
|
|
| Error::ListSchemas { source, .. }
|
|
| Error::ListTables { source, .. }
|
|
| Error::ListFlows { source, .. }
|
|
| Error::ListFlowStats { source, .. }
|
|
| Error::ListProcedures { source, .. }
|
|
| Error::ListRegionStats { source, .. }
|
|
| Error::ConvertProtoData { source, .. } => source.status_code(),
|
|
|
|
Error::CreateTable { source, .. } => source.status_code(),
|
|
|
|
Error::DecodePlan { source, .. } => source.status_code(),
|
|
Error::InvalidTableInfoInCatalog { source, .. } => source.status_code(),
|
|
|
|
Error::Internal { source, .. } => source.status_code(),
|
|
|
|
Error::QueryAccessDenied { .. } => StatusCode::AccessDenied,
|
|
Error::Datafusion { error, .. } => datafusion_status_code::<Self>(error, None),
|
|
Error::ProjectViewColumns { .. } => StatusCode::EngineExecuteQuery,
|
|
Error::TableMetadataManager { source, .. } => source.status_code(),
|
|
Error::GetViewCache { source, .. } | Error::GetTableCache { source, .. } => {
|
|
source.status_code()
|
|
}
|
|
Error::InvokeFrontend { source, .. } => source.status_code(),
|
|
Error::FrontendNotFound { .. } | Error::MetaClientMissing { .. } => {
|
|
StatusCode::Unexpected
|
|
}
|
|
}
|
|
}
|
|
|
|
fn as_any(&self) -> &dyn Any {
|
|
self
|
|
}
|
|
}
|
|
|
|
impl From<Error> for DataFusionError {
|
|
fn from(e: Error) -> Self {
|
|
DataFusionError::External(Box::new(e))
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use snafu::GenerateImplicitData;
|
|
|
|
use super::*;
|
|
|
|
#[test]
|
|
pub fn test_errors_to_datafusion_error() {
|
|
let e: DataFusionError = Error::TableExists {
|
|
table: "test_table".to_string(),
|
|
location: Location::generate(),
|
|
}
|
|
.into();
|
|
match e {
|
|
DataFusionError::External(_) => {}
|
|
_ => {
|
|
panic!("catalog error should be converted to DataFusionError::Internal")
|
|
}
|
|
}
|
|
}
|
|
}
|