add metics for fragment cancel counts

Author: carolinchen-docker

be/src/runtime/fragment_mgr.cpp

@@ -100,6 +100,9 @@ DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(timeout_canceled_fragment_count, MetricUnit::
 bvar::LatencyRecorder g_fragmentmgr_prepare_latency("doris_FragmentMgr", "prepare");
 
 bvar::Adder<uint64_t> g_fragment_executing_count("fragment_executing_count");
+
+bvar::Adder<uint64_t> g_timeout_canceled_fragment_count("timeout_canceled_fragment_count");
+
 bvar::Status<uint64_t> g_fragment_last_active_time(
         "fragment_last_active_time", duration_cast<std::chrono::milliseconds>(
                                              std::chrono::system_clock::now().time_since_epoch())
@@ -882,6 +885,7 @@ Status FragmentMgr::exec_plan_fragment(const TPipelineFragmentParams& params,
     return Status::OK();
 }
 
+// fe->be 取消的调用
 void FragmentMgr::cancel_query(const TUniqueId query_id, const Status reason) {
     std::shared_ptr<QueryContext> query_ctx = nullptr;
     {
@@ -893,6 +897,13 @@ void FragmentMgr::cancel_query(const TUniqueId query_id, const Status reason) {
             return;
         }
     }
+
+    // cancel_query 有两部分发起者，
+    if (reason.is<ErrorCode::TIMEOUT>()) {
+        g_timeout_canceled_fragment_count << 1;
+        LOG(WARNING) << "Query " << print_id(query_id) << " it is cancel by be";
+    }
+    
     query_ctx->cancel(reason);
     _query_ctx_map.erase(query_id);
     LOG(INFO) << "Query " << print_id(query_id)

be/src/service/internal_service.cpp

@@ -646,6 +646,10 @@ void PInternalService::cancel_plan_fragment(google::protobuf::RpcController* /*c
             // to cancel status here.
             if (request->cancel_reason() == PPlanFragmentCancelReason::LIMIT_REACH) {
                 actual_cancel_status = Status::Error<ErrorCode::LIMIT_REACH>("limit reach");
+            } else if (request->has_cancel_reason() == PPlanFragmentCancelReason::TIMEOUT) {
+                actual_cancel_status = Status::Error<ErrorCode::TIMEOUT>(
+                        "cancelled by timeout, reason: {}", PPlanFragmentCancelReason_Name(
+                                request->cancel_reason()));
             } else {
                 // Use cancel reason as error message
                 actual_cancel_status = Status::InternalError(

fe/fe-core/src/main/java/org/apache/doris/common/Status.java

@@ -83,6 +83,10 @@ public void updateStatus(TStatusCode code, String errorMessage) {
         this.errorMsg = errorMessage;
     }
 
+    public void updateErrorCode(TStatusCode code) {
+        this.errorCode = code;
+    }
+
     public boolean ok() {
         return this.errorCode == TStatusCode.OK;
     }

fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java

@@ -1018,7 +1018,11 @@ protected Map<TNetworkAddress, List<Long>>  waitPipelineRpc(List<Pair<Long, Trip
                 if (exception != null && errMsg == null) {
                     errMsg = operation + " failed. " + exception.getMessage();
                 }
-                queryStatus.updateStatus(TStatusCode.INTERNAL_ERROR, errMsg);
+                if (code == TStatusCode.TIMEOUT) {
+                    queryStatus.updateStatus(TStatusCode.TIMEOUT, errMsg);
+                } else {
+                    queryStatus.updateStatus(TStatusCode.INTERNAL_ERROR, errMsg);
+                }
                 cancelInternal(queryStatus);
                 switch (code) {
                     case TIMEOUT:

fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java

@@ -1652,6 +1652,9 @@ public void executeAndSendResult(boolean isOutfileQuery, boolean isSendFields,
             Status internalErrorSt = new Status(TStatusCode.INTERNAL_ERROR,
                     "cancel fragment query_id:{} cause {}",
                     DebugUtil.printId(context.queryId()), e.getMessage());
+            if (e.getMessage().contains("query timeout")) {
+                internalErrorSt.updateErrorCode(TStatusCode.TIMEOUT);
+            }
             LOG.warn(internalErrorSt.getErrorMsg());
             coordBase.cancel(internalErrorSt);
             throw e;