refactor: add ResourceStat

Signed-off-by: zyy17 <[email protected]>

Author: zyy17

Cargo.lock

@@ -30,6 +30,7 @@ use common_meta::heartbeat::handler::invalidate_table_cache::InvalidateCacheHand
 use common_meta::heartbeat::handler::parse_mailbox_message::ParseMailboxMessageHandler;
 use common_meta::key::TableMetadataManager;
 use common_meta::key::flow::FlowMetadataManager;
+use common_stat::ResourceStatImpl;
 use common_telemetry::info;
 use common_telemetry::logging::{DEFAULT_LOGGING_DIR, TracingOptions};
 use common_version::{short_version, verbose_version};
@@ -372,11 +373,15 @@ impl StartCommand {
             Arc::new(InvalidateCacheHandler::new(layered_cache_registry.clone())),
         ]);
 
+        let mut resource_stat = ResourceStatImpl::default();
+        resource_stat.start_collect_cpu_usage();
+
         let heartbeat_task = flow::heartbeat::HeartbeatTask::new(
             &opts,
             meta_client.clone(),
             opts.heartbeat.clone(),
             Arc::new(executor),
+            Arc::new(resource_stat),
         );
 
         let flow_metadata_manager = Arc::new(FlowMetadataManager::new(cached_meta_backend.clone()));

src/cmd/src/flownode.rs

@@ -30,6 +30,7 @@ use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder};
 use common_meta::heartbeat::handler::HandlerGroupExecutor;
 use common_meta::heartbeat::handler::invalidate_table_cache::InvalidateCacheHandler;
 use common_meta::heartbeat::handler::parse_mailbox_message::ParseMailboxMessageHandler;
+use common_stat::ResourceStatImpl;
 use common_telemetry::info;
 use common_telemetry::logging::{DEFAULT_LOGGING_DIR, TracingOptions};
 use common_time::timezone::set_default_timezone;
@@ -421,11 +422,15 @@ impl StartCommand {
             Arc::new(InvalidateCacheHandler::new(layered_cache_registry.clone())),
         ]);
 
+        let mut resource_stat = ResourceStatImpl::default();
+        resource_stat.start_collect_cpu_usage();
+
         let heartbeat_task = HeartbeatTask::new(
             &opts,
             meta_client.clone(),
             opts.heartbeat.clone(),
             Arc::new(executor),
+            Arc::new(resource_stat),
         );
         let heartbeat_task = Some(heartbeat_task);
 

src/cmd/src/frontend.rs

@@ -11,7 +11,6 @@ workspace = true
 common-base.workspace = true
 common-error.workspace = true
 common-macro.workspace = true
-common-stat.workspace = true
 config.workspace = true
 humantime-serde.workspace = true
 object-store.workspace = true

src/common/config/Cargo.toml

@@ -14,7 +14,6 @@
 
 pub mod config;
 pub mod error;
-pub mod utils;
 
 use std::time::Duration;
 

src/common/config/src/lib.rs

@@ -6,11 +6,14 @@ license.workspace = true
 
 [dependencies]
 common-base.workspace = true
+common-runtime.workspace = true
+common-telemetry.workspace = true
 lazy_static.workspace = true
 nix.workspace = true
 num_cpus.workspace = true
 prometheus.workspace = true
 sysinfo.workspace = true
+tokio.workspace = true
 
 [lints]
 workspace = true

src/common/config/src/utils.rs

@@ -145,16 +145,16 @@ pub fn get_cpu_usage_from_cgroups() -> Option<i64> {
     fields[1].trim().parse::<i64>().ok()
 }
 
-/// Calculate the cpu usage in millicores from cgroups filesystem.
-///
-/// - Return `0` if the current cpu usage is equal to the last cpu usage or the interval is 0.
-pub fn calculate_cpu_usage(
+// Calculate the cpu usage in millicores from cgroups filesystem.
+//
+// - Return `0` if the current cpu usage is equal to the last cpu usage or the interval is 0.
+pub(crate) fn calculate_cpu_usage(
     current_cpu_usage_usecs: i64,
     last_cpu_usage_usecs: i64,
     interval_milliseconds: i64,
 ) -> i64 {
     let diff = current_cpu_usage_usecs - last_cpu_usage_usecs;
-    if diff != 0 && interval_milliseconds != 0 {
+    if diff > 0 && interval_milliseconds > 0 {
         ((diff as f64 / interval_milliseconds as f64).round() as i64).max(1)
     } else {
         0

src/common/stat/Cargo.toml

@@ -13,66 +13,7 @@
 // limitations under the License.
 
 mod cgroups;
+mod resource;
 
 pub use cgroups::*;
-use common_base::readable_size::ReadableSize;
-use sysinfo::System;
-
-/// Get the total CPU in millicores.
-pub fn get_total_cpu_millicores() -> i64 {
-    // Get CPU limit from cgroups filesystem.
-    if let Some(cgroup_cpu_limit) = get_cpu_limit_from_cgroups() {
-        cgroup_cpu_limit
-    } else {
-        // Get total CPU cores from host system.
-        num_cpus::get() as i64 * 1000
-    }
-}
-
-/// Get the total memory in bytes.
-pub fn get_total_memory_bytes() -> i64 {
-    // Get memory limit from cgroups filesystem.
-    if let Some(cgroup_memory_limit) = get_memory_limit_from_cgroups() {
-        cgroup_memory_limit
-    } else {
-        // Get total memory from host system.
-        if sysinfo::IS_SUPPORTED_SYSTEM {
-            let mut sys_info = System::new();
-            sys_info.refresh_memory();
-            sys_info.total_memory() as i64
-        } else {
-            // If the system is not supported, return -1.
-            -1
-        }
-    }
-}
-
-/// Get the total CPU cores. The result will be rounded to the nearest integer.
-/// For example, if the total CPU is 1.5 cores(1500 millicores), the result will be 2.
-pub fn get_total_cpu_cores() -> usize {
-    ((get_total_cpu_millicores() as f64) / 1000.0).round() as usize
-}
-
-/// Get the total memory in readable size.
-pub fn get_total_memory_readable() -> Option<ReadableSize> {
-    if get_total_memory_bytes() > 0 {
-        Some(ReadableSize(get_total_memory_bytes() as u64))
-    } else {
-        None
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_get_total_cpu_cores() {
-        assert!(get_total_cpu_cores() > 0);
-    }
-
-    #[test]
-    fn test_get_total_memory_readable() {
-        assert!(get_total_memory_readable().unwrap() > ReadableSize::mb(0));
-    }
-}
+pub use resource::*;