refactor(flight): enhance error handling during request serialization and deserialization

Author: dqhl76

src/common/exception/src/exception_into.rs

@@ -392,7 +392,10 @@ impl From<tonic::Status> for ErrorCode {
                     .set_span(serialized_error.span),
                 }
             }
-            _ => ErrorCode::Unimplemented(status.to_string()),
+            _ => {
+                let debug = format!("{:?}", status);
+                ErrorCode::Unimplemented(debug)
+            }
         }
     }
 }

src/query/service/src/servers/flight/flight_client.rs

@@ -14,6 +14,7 @@
 
 use std::str::FromStr;
 use std::sync::Arc;
+use std::time::Instant;
 
 use arrow_flight::Action;
 use arrow_flight::FlightData;
@@ -115,16 +116,55 @@ impl FlightClient {
             AsciiMetadataKey::from_str("secret").unwrap(),
             AsciiMetadataValue::from_str(&secret).unwrap(),
         );
+        request.metadata_mut().insert(
+            AsciiMetadataKey::from_str("x-request-id").unwrap(),
+            AsciiMetadataValue::from_str(&uuid).unwrap(),
+        );
+
+        let start = Instant::now();
+        debug!(
+            "[{}]FlightClient::do_action rpc start: path={:?}, timeout={}s",
+            &uuid, &path, timeout
+        );
+
+        let response = self.inner.do_action(request).await.map_err(|status| {
+            error!(
+                "[{}]FlightClient::do_action rpc failed: path={:?}, elapsed_ms={}, status={}",
+                &uuid,
+                &path,
+                start.elapsed().as_millis(),
+                status
+            );
+            ErrorCode::from(status)
+        })?;
 
-        let response = self.inner.do_action(request).await?;
+        debug!(
+            "[{}]FlightClient::do_action rpc headers received: path={:?}, elapsed_ms={}",
+            &uuid,
+            &path,
+            start.elapsed().as_millis()
+        );
 
-        match response.into_inner().message().await? {
+        let mut response_stream = response.into_inner();
+        debug!("[{}]FlightClient::wait first response message: path={:?}", &uuid, &path);
+        match response_stream.message().await.map_err(|status| {
+            error!(
+                "[{}]FlightClient::receive response failed: path={:?}, elapsed_ms={}, status={}",
+                &uuid,
+                &path,
+                start.elapsed().as_millis(),
+                status
+            );
+            ErrorCode::from(status)
+        })? {
             Some(response) => {
                 let response_type = std::any::type_name::<Res>();
                 let response_len = response.body.len();
                 debug!(
-                    "[{}]FlightClient::receive response: path={:?}",
-                    &uuid, &path
+                    "[{}]FlightClient::receive response: path={:?}, elapsed_ms={}",
+                    &uuid,
+                    &path,
+                    start.elapsed().as_millis()
                 );
                 let after_deserd = match catch_unwind(
                     || -> std::result::Result<Res, serde_json::Error> {

src/query/service/src/servers/flight/v1/flight_service.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::pin::Pin;
+use std::time::Instant;
 
 use arrow_flight::Action;
 use arrow_flight::ActionType;
@@ -167,7 +168,17 @@ impl FlightService for DatabendQueryFlightService {
     #[async_backtrace::framed]
     async fn do_action(&self, request: Request<Action>) -> Response<Self::DoActionStream> {
         let uuid = uuid::Uuid::new_v4().to_string();
-        debug!("[{}]FlightService::do_action", &uuid);
+        let request_id = request
+            .metadata()
+            .get("x-request-id")
+            .and_then(|value| value.to_str().ok())
+            .unwrap_or("unknown")
+            .to_string();
+        let start = Instant::now();
+        debug!(
+            "[{}]FlightService::do_action: request_id={}",
+            &uuid, &request_id
+        );
         let root = databend_common_tracing::start_trace_for_remote_request(func_path!(), &request);
 
         let secret = request.get_metadata("secret")?;
@@ -189,18 +200,27 @@ impl FlightService for DatabendQueryFlightService {
         {
             Err(cause) => {
                 error!(
-                    "[{}]flight do_action failed, node: {}, action: {}, body_len: {}, code: {}, error: {:?}",
+                    "[{}]flight do_action failed, request_id: {}, node: {}, action: {}, body_len: {}, elapsed_ms: {}, code: {}, error: {:?}",
                     uuid,
+                    request_id,
                     config.query.node_id,
                     action.r#type,
                     action.body.len(),
+                    start.elapsed().as_millis(),
                     cause.code(),
                     cause
                 );
                 Err(cause.into())
             }
             Ok(body) => {
-                debug!("[{}]FlightService: finish", &uuid);
+                debug!(
+                    "[{}]FlightService: finish, request_id: {}, action: {}, body_len: {}, elapsed_ms={}",
+                    &uuid,
+                    &request_id,
+                    action.r#type,
+                    body.len(),
+                    start.elapsed().as_millis()
+                );
                 Ok(RawResponse::new(
                     Box::pin(tokio_stream::once(Ok(FlightResult { body: body.into() })))
                         as FlightStream<FlightResult>,