fix: query errorcode in cluster mode (#328)

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()),
+                });
             }
         };
 

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<ErrorCodes> {
+        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)]

src/service/search.rs

@@ -350,7 +350,10 @@ async fn search_in_cluster(req: cluster_rpc::SearchRequest) -> Result<Response,
         let credentials_str = credentials.clone();
         let task = tokio::task::spawn(
             async move {
-                let org_id: MetadataValue<_> = 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<Response,
                     )
                 });
 
-                let token: MetadataValue<_> = 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,
                     .send_compressed(CompressionEncoding::Gzip)
                     .accept_compressed(CompressionEncoding::Gzip);
                 let response: cluster_rpc::SearchResponse = match client.search(request).await {
-                    Ok(response) => 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<Response,
                 if let Err(e) = locker.unlock().await {
                     log::error!("search in cluster unlock error: {}", e);
                 }
-                return Err(Error::ErrorCode(ErrorCodes::ServerInternalError(
-                    err.to_string(),
-                )));
+                return Err(err);
             }
         }
     }