From 2de20e2a87a759870a262113c67060761874b571 Mon Sep 17 00:00:00 2001
From: Renjie Liu <liurenjie2008@gmail.com>
Date: Thu, 28 Mar 2024 19:41:39 +0800
Subject: [PATCH] fix: Disable batch query metrics by default (#15993)

Co-authored-by: Eric Fu <eric@singularity-data.com>
---
 src/batch/src/task/context.rs | 47 +++++++++++++++++++++--------------
 src/batch/src/task/env.rs     | 11 +++++++-
 src/compute/src/server.rs     |  1 +
 3 files changed, 39 insertions(+), 20 deletions(-)

diff --git a/src/batch/src/task/context.rs b/src/batch/src/task/context.rs
index 70fab761e656..45212b7ce8c6 100644
--- a/src/batch/src/task/context.rs
+++ b/src/batch/src/task/context.rs
@@ -15,7 +15,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
 use risingwave_common::catalog::SysCatalogReaderRef;
-use risingwave_common::config::BatchConfig;
+use risingwave_common::config::{BatchConfig, MetricLevel};
 use risingwave_common::memory::MemoryContext;
 use risingwave_common::util::addr::{is_local_address, HostAddr};
 use risingwave_connector::source::monitor::SourceMetrics;
@@ -164,24 +164,33 @@ impl ComputeNodeContext {
     }
 
     pub fn new(env: BatchEnvironment, task_id: TaskId) -> Self {
-        let batch_mem_context = env.task_manager().memory_context_ref();
-
-        let batch_metrics = Arc::new(BatchMetricsWithTaskLabelsInner::new(
-            env.task_manager().metrics(),
-            env.task_metrics(),
-            env.executor_metrics(),
-            task_id,
-        ));
-        let mem_context = MemoryContext::new(
-            Some(batch_mem_context),
-            batch_metrics.task_mem_usage.clone(),
-        );
-        Self {
-            env,
-            batch_metrics: Some(batch_metrics),
-            cur_mem_val: Arc::new(0.into()),
-            last_mem_val: Arc::new(0.into()),
-            mem_context,
+        if env.metric_level() >= MetricLevel::Debug {
+            let batch_mem_context = env.task_manager().memory_context_ref();
+            let batch_metrics = Arc::new(BatchMetricsWithTaskLabelsInner::new(
+                env.task_manager().metrics(),
+                env.task_metrics(),
+                env.executor_metrics(),
+                task_id,
+            ));
+            let mem_context = MemoryContext::new(
+                Some(batch_mem_context),
+                batch_metrics.task_mem_usage.clone(),
+            );
+            Self {
+                env,
+                batch_metrics: Some(batch_metrics),
+                cur_mem_val: Arc::new(0.into()),
+                last_mem_val: Arc::new(0.into()),
+                mem_context,
+            }
+        } else {
+            Self {
+                env,
+                batch_metrics: None,
+                cur_mem_val: Arc::new(0.into()),
+                last_mem_val: Arc::new(0.into()),
+                mem_context: MemoryContext::none(),
+            }
         }
     }
 
diff --git a/src/batch/src/task/env.rs b/src/batch/src/task/env.rs
index 2099adad4da6..58631cb9563a 100644
--- a/src/batch/src/task/env.rs
+++ b/src/batch/src/task/env.rs
@@ -14,7 +14,7 @@
 
 use std::sync::Arc;
 
-use risingwave_common::config::BatchConfig;
+use risingwave_common::config::{BatchConfig, MetricLevel};
 use risingwave_common::util::addr::HostAddr;
 use risingwave_common::util::worker_util::WorkerNodeId;
 use risingwave_connector::source::monitor::SourceMetrics;
@@ -58,6 +58,8 @@ pub struct BatchEnvironment {
 
     /// Metrics for source.
     source_metrics: Arc<SourceMetrics>,
+
+    metric_level: MetricLevel,
 }
 
 impl BatchEnvironment {
@@ -73,6 +75,7 @@ impl BatchEnvironment {
         client_pool: ComputeClientPoolRef,
         dml_manager: DmlManagerRef,
         source_metrics: Arc<SourceMetrics>,
+        metric_level: MetricLevel,
     ) -> Self {
         BatchEnvironment {
             server_addr,
@@ -85,6 +88,7 @@ impl BatchEnvironment {
             client_pool,
             dml_manager,
             source_metrics,
+            metric_level,
         }
     }
 
@@ -111,6 +115,7 @@ impl BatchEnvironment {
             dml_manager: Arc::new(DmlManager::for_test()),
             source_metrics: Arc::new(SourceMetrics::default()),
             executor_metrics: Arc::new(BatchExecutorMetrics::for_test()),
+            metric_level: MetricLevel::Debug,
         }
     }
 
@@ -157,4 +162,8 @@ impl BatchEnvironment {
     pub fn source_metrics(&self) -> Arc<SourceMetrics> {
         self.source_metrics.clone()
     }
+
+    pub fn metric_level(&self) -> MetricLevel {
+        self.metric_level
+    }
 }
diff --git a/src/compute/src/server.rs b/src/compute/src/server.rs
index f9cd24162f56..7119df4926b9 100644
--- a/src/compute/src/server.rs
+++ b/src/compute/src/server.rs
@@ -331,6 +331,7 @@ pub async fn compute_node_serve(
         client_pool,
         dml_mgr.clone(),
         source_metrics.clone(),
+        config.server.metrics_level,
     );
 
     info!(