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, + + metric_level: MetricLevel, } impl BatchEnvironment { @@ -73,6 +75,7 @@ impl BatchEnvironment { client_pool: ComputeClientPoolRef, dml_manager: DmlManagerRef, source_metrics: Arc, + 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 { 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!(