fix(cluster): TPC-DS cluster timeout expired

.github/actions/build_linux/action.yml

@@ -107,6 +107,14 @@ runs:
         strip --strip-debug --remove-section=.comment --remove-section=.note ./target/${{ inputs.target }}/${{ env.BUILD_PROFILE }}/databend-query
         pushd ./target/${{ inputs.target }}/${{ env.BUILD_PROFILE }} && objcopy --add-gnu-debuglink databend-query.debug databend-query && popd
 
+    - name: Upload databend-query.debug (GitHub Artifact)
+      # Keep debug symbols in GitHub Artifacts for quick retrieval.
+      if: inputs.upload == 'true' && inputs.trim_debug == 'true'
+      uses: actions/upload-artifact@v4
+      with:
+        name: databend-query-debug-${{ inputs.target }}-${{ inputs.sha }}
+        path: ./target/${{ inputs.target }}/${{ env.BUILD_PROFILE }}/databend-query.debug
+
     # - name: Compress Binaries with UPX
     #   if: env.BUILD_PROFILE == 'debug'
     #   uses: crazy-max/ghaction-upx@v2

.github/workflows/reuse.sqllogic.yml

@@ -263,14 +263,16 @@ jobs:
       fail-fast: false
       matrix:
         tests:
-          - { dirs: "query", runner: "4c" }
-          - { dirs: "duckdb", runner: "4c" }
-          - { dirs: "crdb", runner: "2c", parallel: "2" }
-          - { dirs: "base", runner: "2c", parallel: "2" }
-          - { dirs: "ydb", runner: "2c" }
           - { dirs: "tpcds", runner: "4c" }
-          - { dirs: "tpch", runner: "2c" }
-          - { dirs: "cluster", runner: "2c" }
+          - { dirs: "tpcds", runner: "4c" }
+          - { dirs: "tpcds", runner: "4c" }
+          - { dirs: "tpcds", runner: "4c" }
+          - { dirs: "tpcds", runner: "4c" }
+          - { dirs: "tpcds", runner: "4c" }
+          - { dirs: "tpcds", runner: "4c" }
+          - { dirs: "tpcds", runner: "4c" }
+          - { dirs: "tpcds", runner: "4c" }
+          - { dirs: "tpcds", runner: "4c" }
         handler:
           - "hybrid"
           - "http"

scripts/ci/ci-collect-async-tasks-dump.sh

@@ -0,0 +1,36 @@
+#!/bin/bash
+# Copyright 2020-2021 The Databend Authors.
+# SPDX-License-Identifier: Apache-2.0.
+
+set -u
+
+OUTPUT_PATH=${1:-"./.databend/async_tasks_dump.log"}
+INTERVAL_SECONDS=${2:-1}
+ENDPOINTS=${3:-"http://127.0.0.1:8080/debug/async_tasks/dump,http://127.0.0.1:8082/debug/async_tasks/dump,http://127.0.0.1:8083/debug/async_tasks/dump"}
+
+mkdir -p "$(dirname "${OUTPUT_PATH}")"
+touch "${OUTPUT_PATH}"
+
+IFS=',' read -r -a ENDPOINT_LIST <<< "${ENDPOINTS}"
+
+{
+    echo "=== async_tasks_dump collector started at $(date -u +"%Y-%m-%dT%H:%M:%SZ") ==="
+    echo "interval_seconds=${INTERVAL_SECONDS}"
+    echo "endpoints=${ENDPOINTS}"
+} >> "${OUTPUT_PATH}"
+
+while true; do
+    TIMESTAMP="$(date -u +"%Y-%m-%dT%H:%M:%SZ")"
+    for endpoint in "${ENDPOINT_LIST[@]}"; do
+        echo "" >> "${OUTPUT_PATH}"
+        echo "===== ${TIMESTAMP} ${endpoint} =====" >> "${OUTPUT_PATH}"
+
+        curl --silent --show-error --max-time 3 "${endpoint}" >> "${OUTPUT_PATH}" 2>&1
+        CURL_EXIT_CODE=$?
+        if [ "${CURL_EXIT_CODE}" -ne 0 ]; then
+            echo "[collector] curl exit code: ${CURL_EXIT_CODE}" >> "${OUTPUT_PATH}"
+        fi
+    done
+
+    sleep "${INTERVAL_SECONDS}"
+done

scripts/ci/ci-run-sqllogic-tests-cluster.sh

@@ -9,6 +9,26 @@ export STORAGE_ALLOW_INSECURE=true
 echo "Starting Cluster databend-query"
 ./scripts/ci/deploy/databend-query-cluster-3-nodes.sh
 
+ASYNC_TASKS_DUMP_OUTPUT=${ASYNC_TASKS_DUMP_OUTPUT:-"./.databend/async_tasks_dump.log"}
+ASYNC_TASKS_DUMP_INTERVAL=${ASYNC_TASKS_DUMP_INTERVAL:-1}
+ASYNC_TASKS_DUMP_ENDPOINTS=${ASYNC_TASKS_DUMP_ENDPOINTS:-"http://127.0.0.1:8080/debug/async_tasks/dump,http://127.0.0.1:8082/debug/async_tasks/dump,http://127.0.0.1:8083/debug/async_tasks/dump"}
+
+echo "Starting async tasks dump collector"
+bash ./scripts/ci/ci-collect-async-tasks-dump.sh \
+	"${ASYNC_TASKS_DUMP_OUTPUT}" \
+	"${ASYNC_TASKS_DUMP_INTERVAL}" \
+	"${ASYNC_TASKS_DUMP_ENDPOINTS}" &
+ASYNC_TASKS_DUMP_PID=$!
+
+cleanup_async_tasks_dump_collector() {
+	if [ -n "${ASYNC_TASKS_DUMP_PID:-}" ] && kill -0 "${ASYNC_TASKS_DUMP_PID}" 2>/dev/null; then
+		echo "Stopping async tasks dump collector"
+		kill "${ASYNC_TASKS_DUMP_PID}" 2>/dev/null || true
+		wait "${ASYNC_TASKS_DUMP_PID}" 2>/dev/null || true
+	fi
+}
+trap cleanup_async_tasks_dump_collector EXIT
+
 export RUST_BACKTRACE=1
 
 TEST_HANDLERS=${TEST_HANDLERS:-"mysql,http"}

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/clusters/cluster.rs

@@ -213,7 +213,10 @@ impl ClusterHelper for Cluster {
                                 attempt += 1;
                                 sleep(Duration::from_secs(flight_params.retry_interval)).await;
                             }
-                            Err(e) => return Err(e),
+                            Err(e) => {
+                                error!("do_action failed: {:?}", e);
+                                return Err(e);
+                            }
                         }
                     }
                 }

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

@@ -14,13 +14,15 @@
 
 use std::str::FromStr;
 use std::sync::Arc;
+use std::time::Instant;
 
 use arrow_flight::Action;
 use arrow_flight::FlightData;
 use arrow_flight::Ticket;
 use arrow_flight::flight_service_client::FlightServiceClient;
 use async_channel::Receiver;
 use async_channel::Sender;
+use databend_common_base::runtime::catch_unwind;
 use databend_common_base::runtime::drop_guard;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
@@ -29,6 +31,8 @@ use fastrace::func_path;
 use fastrace::future::FutureExt;
 use futures::StreamExt;
 use futures_util::future::Either;
+use log::debug;
+use log::error;
 use serde::Deserialize;
 use serde::Serialize;
 use tokio::time::Duration;
@@ -47,6 +51,8 @@ pub struct FlightClient {
     inner: FlightServiceClient<Channel>,
 }
 
+const DO_GET_TIMEOUT_SECS: u64 = 60;
+
 // TODO: Integration testing required
 impl FlightClient {
     pub fn new(mut inner: FlightServiceClient<Channel>) -> FlightClient {
@@ -69,15 +75,34 @@ impl FlightClient {
         T: Serialize,
         Res: for<'a> Deserialize<'a>,
     {
-        let mut body = Vec::with_capacity(512);
-        let mut serializer = serde_json::Serializer::new(&mut body);
-        let serializer = serde_stacker::Serializer::new(&mut serializer);
-        message.serialize(serializer).map_err(|cause| {
-            ErrorCode::BadArguments(format!(
-                "Request payload serialize error while in {:?}, cause: {}",
-                path, cause
-            ))
-        })?;
+        let uuid = uuid::Uuid::new_v4().to_string();
+        debug!("[{}]FlightClient::do_action: path={:?}", &uuid, &path);
+        let message_type = std::any::type_name::<T>();
+        let body = match catch_unwind(|| -> std::result::Result<Vec<u8>, serde_json::Error> {
+            let mut body = Vec::with_capacity(512);
+            let mut serializer = serde_json::Serializer::new(&mut body);
+            let serializer = serde_stacker::Serializer::new(&mut serializer);
+            message.serialize(serializer)?;
+            Ok(body)
+        }) {
+            Ok(Ok(body)) => body,
+            Ok(Err(cause)) => {
+                error!("{:?}", &cause);
+                return Err(ErrorCode::BadArguments(format!(
+                    "Request payload serialize error while in {:?}, type: {}, cause: {}",
+                    path, message_type, cause
+                )));
+            }
+            Err(cause) => {
+                error!("{:?}", &cause);
+                return Err(cause.add_message_back(format!(
+                    "(while serializing flight action request: action={:?}, type={})",
+                    path, message_type
+                )));
+            }
+        };
+
+        debug!("[{}]FlightClient::serd finish: path={:?}", &uuid, &path);
 
         drop(message);
         let mut request =
@@ -91,21 +116,85 @@ 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 mut deserializer = serde_json::Deserializer::from_slice(&response.body);
-                deserializer.disable_recursion_limit();
-                let deserializer = serde_stacker::Deserializer::new(&mut deserializer);
-
-                Res::deserialize(deserializer).map_err(|cause| {
-                    ErrorCode::BadBytes(format!(
-                        "Response payload deserialize error while in {:?}, cause: {}",
-                        path, cause
-                    ))
-                })
+                let response_type = std::any::type_name::<Res>();
+                let response_len = response.body.len();
+                debug!(
+                    "[{}]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> {
+                        let mut deserializer = serde_json::Deserializer::from_slice(&response.body);
+                        deserializer.disable_recursion_limit();
+                        let deserializer = serde_stacker::Deserializer::new(&mut deserializer);
+                        Res::deserialize(deserializer)
+                    },
+                ) {
+                    Ok(Ok(res)) => Ok(res),
+                    Ok(Err(cause)) => {
+                        error!("{:?}", &cause);
+                        Err(ErrorCode::BadBytes(format!(
+                            "Response payload deserialize error while in {:?}, type: {}, len: {}, cause: {}",
+                            path, response_type, response_len, cause
+                        )))
+                    }
+                    Err(cause) => {
+                        error!("{:?}", &cause);
+                        Err(cause.add_message_back(format!(
+                        "(while deserializing flight action response: action={:?}, type={}, len={})",
+                        path, response_type, response_len
+                    )))
+                    }
+                };
+                debug!("[{}]FlightClient::deserd finish: path={:?}", &uuid, &path);
+                after_deserd
             }
             None => Err(ErrorCode::EmptyDataFromServer(format!(
                 "Can not receive data from flight server, action: {:?}",
@@ -197,10 +286,38 @@ impl FlightClient {
     }
 
     #[async_backtrace::framed]
-    async fn get_streaming(&mut self, request: Request<Ticket>) -> Result<Streaming<FlightData>> {
+    async fn get_streaming(
+        &mut self,
+        mut request: Request<Ticket>,
+    ) -> Result<Streaming<FlightData>> {
+        let request_type = request
+            .metadata()
+            .get("x-type")
+            .and_then(|value| value.to_str().ok())
+            .unwrap_or("unknown")
+            .to_string();
+
+        request.set_timeout(Duration::from_secs(DO_GET_TIMEOUT_SECS));
+        debug!(
+            "[flight] FlightClient::do_get start, x-type: {}, timeout_secs: {}",
+            request_type, DO_GET_TIMEOUT_SECS
+        );
+
         match self.inner.do_get(request).await {
-            Ok(res) => Ok(res.into_inner()),
-            Err(status) => Err(ErrorCode::from(status).add_message_back("(while in query flight)")),
+            Ok(res) => {
+                debug!(
+                    "[flight] FlightClient::do_get finish, x-type: {}",
+                    request_type
+                );
+                Ok(res.into_inner())
+            }
+            Err(status) => {
+                error!(
+                    "[flight] FlightClient::do_get failed, x-type: {}, status: {:?}",
+                    request_type, status
+                );
+                Err(ErrorCode::from(status).add_message_back("(while in query flight)"))
+            }
         }
     }
 }