diff --git a/src/handler/grpc/request/search.rs b/src/handler/grpc/request/search.rs index 443e9740b0..c2571c569a 100644 --- a/src/handler/grpc/request/search.rs +++ b/src/handler/grpc/request/search.rs @@ -20,6 +20,7 @@ use tracing_opentelemetry::OpenTelemetrySpanExt; use crate::handler::grpc::cluster_rpc::search_server::Search; use crate::handler::grpc::cluster_rpc::SearchRequest; use crate::handler::grpc::cluster_rpc::SearchResponse; +use crate::infra::errors; use crate::infra::metrics; use crate::service::search as SearchService; @@ -53,7 +54,10 @@ impl Search for Searcher { metrics::GRPC_INCOMING_REQUESTS .with_label_values(&["/_search", "500", &org_id, "", stream_type]) .inc(); - return Err(Status::internal(err.to_string())); + return Err(match err { + errors::Error::ErrorCode(code) => Status::internal(code.to_json()), + _ => Status::internal(err.to_string()), + }); } }; diff --git a/src/infra/errors.rs b/src/infra/errors.rs index 7c33b1abcc..eba33da422 100644 --- a/src/infra/errors.rs +++ b/src/infra/errors.rs @@ -84,6 +84,7 @@ impl ErrorCodes { ErrorCodes::SearchFieldHasNoCompatibleDataType(_) => 20007, } } + pub fn get_message(&self) -> String { match self { ErrorCodes::ServerInternalError(msg) => msg.to_owned(), @@ -104,6 +105,64 @@ impl ErrorCodes { } } } + + pub fn get_inner_message(&self) -> String { + match self { + ErrorCodes::ServerInternalError(msg) => msg.to_owned(), + ErrorCodes::SearchSQLNotValid(sql) => sql.to_owned(), + ErrorCodes::SearchStreamNotFound(stream) => stream.to_owned(), + ErrorCodes::FullTextSearchFieldNotFound => "".to_string(), + ErrorCodes::SearchFieldNotFound(field) => field.to_owned(), + ErrorCodes::SearchFunctionNotDefined(func) => func.to_owned(), + ErrorCodes::SearchParquetFileNotFound => "".to_string(), + ErrorCodes::SearchFieldHasNoCompatibleDataType(field) => field.to_owned(), + } + } + + pub fn to_json(&self) -> String { + format!( + r#"{{"code": {}, "message": "{}", "inner": "{}"}}"#, + self.get_code(), + self.get_message(), + self.get_inner_message() + ) + } + + pub fn from_json(json: &str) -> Result { + let val: serde_json::Value = match serde_json::from_str(json) { + Ok(val) => val, + Err(_) => return Ok(ErrorCodes::ServerInternalError(json.to_string())), + }; + let map = match val.as_object() { + Some(map) => map, + None => return Ok(ErrorCodes::ServerInternalError(json.to_string())), + }; + let code = match map.get("code") { + Some(code) => match code.as_i64() { + Some(code) => code as u16, + None => return Ok(ErrorCodes::ServerInternalError(json.to_string())), + }, + None => return Ok(ErrorCodes::ServerInternalError(json.to_string())), + }; + let message = match map.get("inner") { + Some(message) => match message { + serde_json::Value::String(message) => message.to_owned(), + _ => message.to_string(), + }, + None => return Ok(ErrorCodes::ServerInternalError(json.to_string())), + }; + match code { + 10001 => Ok(ErrorCodes::ServerInternalError(message)), + 20001 => Ok(ErrorCodes::SearchSQLNotValid(message)), + 20002 => Ok(ErrorCodes::SearchStreamNotFound(message)), + 20003 => Ok(ErrorCodes::FullTextSearchFieldNotFound), + 20004 => Ok(ErrorCodes::SearchFieldNotFound(message)), + 20005 => Ok(ErrorCodes::SearchFunctionNotDefined(message)), + 20006 => Ok(ErrorCodes::SearchParquetFileNotFound), + 20007 => Ok(ErrorCodes::SearchFieldHasNoCompatibleDataType(message)), + _ => Ok(ErrorCodes::ServerInternalError(json.to_string())), + } + } } #[cfg(test)] diff --git a/src/service/search.rs b/src/service/search.rs index 0e07891689..ff5edcf42a 100644 --- a/src/service/search.rs +++ b/src/service/search.rs @@ -350,7 +350,10 @@ async fn search_in_cluster(req: cluster_rpc::SearchRequest) -> Result = req.org_id.parse()?; + let org_id: MetadataValue<_> = match req.org_id.parse() { + Ok(org_id) => org_id, + Err(_) => return Err(Error::Message("invalid org_id".to_string())), + }; let mut request = tonic::Request::new(req); request.set_timeout(Duration::from_secs(CONFIG.grpc.timeout)); @@ -361,8 +364,23 @@ async fn search_in_cluster(req: cluster_rpc::SearchRequest) -> Result = credentials_str.parse()?; - let channel = Channel::from_shared(node_addr).unwrap().connect().await?; + let token: MetadataValue<_> = match credentials_str.parse() { + Ok(token) => token, + Err(_) => return Err(Error::Message("invalid token".to_string())), + }; + let channel = match Channel::from_shared(node_addr).unwrap().connect().await { + Ok(channel) => channel, + Err(err) => { + log::error!( + "[TRACE] cluster->search: node: {}, connect grpc err: {:?}", + node.id, + err + ); + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + "connect search node error".to_string(), + ))); + } + }; let mut client = cluster_rpc::search_client::SearchClient::with_interceptor( channel, move |mut req: Request<()>| { @@ -376,13 +394,20 @@ async fn search_in_cluster(req: cluster_rpc::SearchRequest) -> Result response.into_inner(), - Err(error) => { - return Err(anyhow::anyhow!( - "cluster search: node: {}, err: {:?}", + Ok(res) => res.into_inner(), + Err(err) => { + log::error!( + "[TRACE] cluster->search: node: {}, search grpc err: {:?}", node.id, - error - )); + err + ); + if err.code().eq(&tonic::Code::Internal) { + let err = ErrorCodes::from_json(err.message())?; + return Err(Error::ErrorCode(err)); + } + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + "search node error".to_string(), + ))); } }; @@ -420,9 +445,7 @@ async fn search_in_cluster(req: cluster_rpc::SearchRequest) -> Result