chore(query): add external server request profile

src/common/base/src/runtime/profile/profiles.rs

@@ -46,6 +46,7 @@ pub enum ProfileStatisticsName {
     RuntimeFilterPruneParts,
     MemoryUsage,
     ExternalServerRetryCount,
+    ExternalServerRequestCount,
 }
 
 #[derive(Clone, Hash, Eq, PartialEq, serde::Serialize, serde::Deserialize, Debug)]
@@ -251,6 +252,13 @@ pub fn get_statistics_desc() -> Arc<BTreeMap<ProfileStatisticsName, ProfileDesc>
                 unit: StatisticsUnit::Count,
                 plain_statistics: true,
             }),
+            (ProfileStatisticsName::ExternalServerRequestCount, ProfileDesc {
+                display_name: "external server request count",
+                desc: "The count of external server request times",
+                index: ProfileStatisticsName::ExternalServerRequestCount as usize,
+                unit: StatisticsUnit::Count,
+                plain_statistics: true,
+            }),
         ]))
     }).clone()
 }

src/common/metrics/src/lib.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![feature(duration_millis_float)]
 #![allow(clippy::uninlined_format_args)]
 #![allow(dead_code)]
 #![recursion_limit = "256"]
@@ -23,6 +24,7 @@ pub type VecLabels = Vec<(&'static str, String)>;
 
 pub use crate::metrics::cache;
 pub use crate::metrics::cluster;
+pub use crate::metrics::external_server;
 /// Metrics.
 pub use crate::metrics::http;
 pub use crate::metrics::interpreter;

src/common/metrics/src/metrics/external_server.rs

@@ -0,0 +1,35 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::LazyLock;
+use std::time::Duration;
+
+use databend_common_base::runtime::metrics::register_histogram_family_in_seconds;
+use databend_common_base::runtime::metrics::FamilyHistogram;
+
+use crate::VecLabels;
+
+const METRIC_REQUEST_EXTERNAL_DURATION: &str = "external_request_duration";
+
+static REQUEST_EXTERNAL_DURATION: LazyLock<FamilyHistogram<VecLabels>> =
+    LazyLock::new(|| register_histogram_family_in_seconds(METRIC_REQUEST_EXTERNAL_DURATION));
+
+const LABEL_FUNCTION_NAME: &str = "function_name";
+
+pub fn record_request_external_duration(function_name: String, duration: Duration) {
+    let labels = &vec![(LABEL_FUNCTION_NAME, function_name)];
+    REQUEST_EXTERNAL_DURATION
+        .get_or_create(labels)
+        .observe(duration.as_millis_f64());
+}

src/common/metrics/src/metrics/mod.rs

@@ -14,6 +14,7 @@
 
 pub mod cache;
 pub mod cluster;
+pub mod external_server;
 pub mod http;
 pub mod interpreter;
 pub mod lock;

src/query/service/src/interpreters/interpreter_explain.rs

@@ -189,7 +189,7 @@ impl Interpreter for ExplainInterpreter {
             ExplainKind::Pipeline => {
                 // todo:(JackTan25), we need to make all execute2() just do `build pipeline` work,
                 // don't take real actions. for now we fix #13657 like below.
-                let pipeline = match &self.plan {
+                let mut pipeline = match &self.plan {
                     Plan::Query { .. } | Plan::DataMutation { .. } => {
                         let interpter =
                             InterpreterFactory::get(self.ctx.clone(), &self.plan).await?;
@@ -198,6 +198,15 @@ impl Interpreter for ExplainInterpreter {
                     _ => PipelineBuildResult::create(),
                 };
 
+                // The explain pipeline does not require executing on_init and on_finished.
+                let _ = pipeline.main_pipeline.take_on_init();
+                let _ = pipeline.main_pipeline.take_on_finished();
+
+                for pipeline in &mut pipeline.sources_pipelines {
+                    let _ = pipeline.take_on_init();
+                    let _ = pipeline.take_on_finished();
+                }
+
                 Self::format_pipeline(&pipeline)
             }
 

src/query/service/src/pipelines/processors/transforms/transform_udf_server.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::sync::Arc;
+use std::time::Instant;
 
 use databend_common_base::runtime::profile::Profile;
 use databend_common_base::runtime::profile::ProfileStatisticsName;
@@ -26,6 +27,7 @@ use databend_common_expression::BlockEntry;
 use databend_common_expression::DataBlock;
 use databend_common_expression::DataField;
 use databend_common_expression::DataSchema;
+use databend_common_metrics::external_server::record_request_external_duration;
 use databend_common_pipeline_transforms::processors::AsyncRetry;
 use databend_common_pipeline_transforms::processors::AsyncRetryWrapper;
 use databend_common_pipeline_transforms::processors::AsyncTransform;
@@ -105,14 +107,19 @@ impl TransformUdfServer {
             .to_record_batch_with_dataschema(&data_schema)
             .map_err(|err| ErrorCode::from_string(format!("{err}")))?;
 
+        let instant = Instant::now();
         let mut client =
             UDFFlightClient::connect(server_addr, connect_timeout, request_timeout, 65536)
                 .await?
                 .with_tenant(ctx.get_tenant().tenant_name())?
                 .with_func_name(&func.func_name)?
                 .with_query_id(&ctx.get_id())?;
 
+        Profile::record_usize_profile(ProfileStatisticsName::ExternalServerRequestCount, 1);
         let result_batch = client.do_exchange(&func.func_name, input_batch).await?;
+
+        record_request_external_duration(func.func_name.clone(), instant.elapsed());
+
         let schema = DataSchema::try_from(&(*result_batch.schema()))?;
         let (result_block, result_schema) = DataBlock::from_record_batch(&schema, &result_batch)
             .map_err(|err| {