fix: reduce tokio runtime polling overhead

.github/workflows/build_and_push_images.yml

@@ -26,8 +26,8 @@ on:
         options:
           - release
           - profiling
-      enable_tokio_console:
-        description: "Build with tokio-console support"
+      enable_tracing:
+        description: "Build with tracing support"
         required: true
         type: choice
         default: "false"
@@ -51,8 +51,6 @@ jobs:
       CARGO_BUILD_PROFILE: ${{ inputs.build_profile == 'profiling' && '--profile profiling' || '--release' }}
       CARGO_TARGET_PROFILE: ${{ inputs.build_profile == 'profiling' && 'profiling' || 'release' }}
       PROFILING_RUSTFLAGS: ${{ inputs.build_profile == 'profiling' && '-C force-frame-pointers=yes' || '' }}
-      TOKIO_CONSOLE_FEATURES: ${{ inputs.enable_tokio_console == 'true' && ',tokio-console' || '' }}
-      TOKIO_CONSOLE_RUSTFLAGS: ${{ inputs.enable_tokio_console == 'true' && '--cfg tokio_unstable' || '' }}
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
@@ -68,13 +66,13 @@ jobs:
       - name: Build for x86_64
         if: contains(inputs.platforms, 'linux/amd64')
         run: |
-          RUSTFLAGS="${TOKIO_CONSOLE_RUSTFLAGS} ${PROFILING_RUSTFLAGS}" cross build --target x86_64-unknown-linux-gnu ${CARGO_BUILD_PROFILE} --features metrics${TOKIO_CONSOLE_FEATURES}
+          RUSTFLAGS="${{ inputs.enable_tracing == 'true' && '--cfg tokio_unstable' || '' }} ${PROFILING_RUSTFLAGS}" cross build --target x86_64-unknown-linux-gnu ${CARGO_BUILD_PROFILE} --features metrics${{ inputs.enable_tracing == 'true' && ',tracing' || '' }}
           cp target/x86_64-unknown-linux-gnu/${CARGO_TARGET_PROFILE}/dt-main amd64-unknown-linux-gnu-dt-main
 
       - name: Build for aarch64
         if: contains(inputs.platforms, 'linux/arm64')
         run: |
-          RUSTFLAGS="${TOKIO_CONSOLE_RUSTFLAGS} ${PROFILING_RUSTFLAGS}" cross build --target aarch64-unknown-linux-gnu ${CARGO_BUILD_PROFILE} --features metrics${TOKIO_CONSOLE_FEATURES}
+          RUSTFLAGS="${{ inputs.enable_tracing == 'true' && '--cfg tokio_unstable' || '' }} ${PROFILING_RUSTFLAGS}" cross build --target aarch64-unknown-linux-gnu ${CARGO_BUILD_PROFILE} --features metrics${{ inputs.enable_tracing == 'true' && ',tracing' || '' }}
           cp target/aarch64-unknown-linux-gnu/${CARGO_TARGET_PROFILE}/dt-main arm64-unknown-linux-gnu-dt-main
 
       - name: Set up docker buildx

Cargo.toml

@@ -37,6 +37,9 @@ postgres-openssl = {git = "https://github.com/apecloud/rust-postgres"}
 postgres-protocol = {git = "https://github.com/apecloud/rust-postgres"}
 postgres-types= {git = "https://github.com/apecloud/rust-postgres"}
 tokio = { version = "1.25.0", features = ["full"] }
+tokio-metrics = { version = "0.5.1", default-features = false }
+tokio-util = {version = "0.7.7", features = ["rt"]}
+tracing-subscriber = { version = "0.3.20", default-features = false, features = ["fmt", "registry", "std"] }
 bytes = "1.0"
 configparser = "3.0.2"
 async-mutex = "1.4.0"

dt-common/Cargo.toml

@@ -10,6 +10,7 @@ workspace = true
 
 [features]
 metrics = ["prometheus"]
+tracing = ["dep:console-subscriber", "dep:tokio-metrics", "dep:tonic", "dep:tonic-prost", "dep:tracing-subscriber", "tokio/tracing"]
 
 [dependencies]
 dotenv = { workspace = true }
@@ -29,6 +30,7 @@ serde_yaml = { workspace = true }
 regex = { workspace = true }
 nom = { workspace = true }
 tokio = { workspace = true }
+tokio-metrics = { workspace = true, optional = true }
 thiserror = { workspace = true }
 kafka = { workspace = true }
 apache-avro = {workspace = true}
@@ -50,3 +52,7 @@ actix-web = {workspace = true}
 prometheus = {version = "0.14.0", optional = true}
 urlencoding= {workspace = true}
 governor = { workspace = true }
+console-subscriber = { version = "=0.5.0", optional = true }
+tonic = { version = "=0.14.5", optional = true }
+tonic-prost = { version = "=0.14.5", optional = true }
+tracing-subscriber = { workspace = true, optional = true }

dt-common/src/config/mod.rs

@@ -20,6 +20,7 @@ pub mod s3_config;
 pub mod sinker_config;
 pub mod ssl_config;
 pub mod task_config;
+pub mod tracing_config;
 
 #[cfg(feature = "metrics")]
 pub mod metrics_config;

dt-common/src/config/task_config.rs

@@ -42,6 +42,7 @@ use super::{
     runtime_config::RuntimeConfig,
     s3_config::S3Config,
     sinker_config::{BasicSinkerConfig, SinkerConfig},
+    tracing_config::TracingConfig,
 };
 
 #[derive(Clone)]
@@ -61,6 +62,7 @@ pub struct TaskConfig {
     pub meta_center: Option<MetaCenterConfig>,
     pub data_marker: Option<DataMarkerConfig>,
     pub processor: Option<ProcessorConfig>,
+    pub tracing: TracingConfig,
     #[cfg(feature = "metrics")]
     pub metrics: MetricsConfig,
 }
@@ -83,6 +85,7 @@ const DATA_MARKER: &str = "data_marker";
 const PROCESSOR: &str = "processor";
 const CHECKER: &str = "checker";
 const META_CENTER: &str = "metacenter";
+const TRACING: &str = "tracing";
 // keys
 const CHECK_LOG_DIR: &str = "check_log_dir";
 const CHECK_LOG_FILE_SIZE: &str = "check_log_file_size";
@@ -124,6 +127,8 @@ const CDC_CHECK_LOG_INTERVAL_SECS: &str = "cdc_check_log_interval_secs";
 const SAMPLE_RATE: &str = "sample_rate";
 const IS_DIRECT_CONNECTION: &str = "is_direct_connection";
 const MONGO_REQUIRE_SHARD_KEY_FILTER: &str = "mongo_require_shard_key_filter";
+const TASK_SUMMARY_MODE: &str = "task_summary_mode";
+const OUTPUT_FORMAT: &str = "output_format";
 
 // default values
 pub const APE_DTS: &str = "APE_DTS";
@@ -257,6 +262,7 @@ impl TaskConfig {
             data_marker: Self::load_data_marker_config(&loader)?,
             processor: Self::load_processor_config(&loader)?,
             meta_center: Self::load_meta_center_config(&loader)?,
+            tracing: Self::load_tracing_config(&loader),
             #[cfg(feature = "metrics")]
             metrics: Self::load_metrics_config(&loader)?,
         })
@@ -1228,6 +1234,18 @@ impl TaskConfig {
         })
     }
 
+    fn load_tracing_config(loader: &IniLoader) -> TracingConfig {
+        let default = TracingConfig::default();
+        TracingConfig {
+            task_summary_mode: loader.get_with_default(
+                TRACING,
+                TASK_SUMMARY_MODE,
+                default.task_summary_mode,
+            ),
+            output_format: loader.get_with_default(TRACING, OUTPUT_FORMAT, default.output_format),
+        }
+    }
+
     fn load_snapshot_parallel_size(loader: &IniLoader) -> usize {
         if loader.contains(EXTRACTOR, PARALLEL_SIZE) {
             loader.get_with_default(EXTRACTOR, PARALLEL_SIZE, 1)
@@ -1445,6 +1463,7 @@ mod tests {
     use crate::config::parallelizer_config::{
         ChunkPartitionerRebalanceCost, ChunkPartitionerRebalanceStrategy,
     };
+    use crate::runtime_trace::{TaskSummaryMode, TraceOutputFormat};
 
     use super::{
         CheckMode, ExtractorConfig, ParallelType, SinkerConfig, TaskConfig, TaskKind, TaskType,
@@ -1513,6 +1532,49 @@ parallel_type=rdb_merge
         )
     }
 
+    fn basic_snapshot_config(extra_config: &str) -> String {
+        format!(
+            r#"[extractor]
+db_type=mysql
+extract_type=snapshot
+url=mysql://127.0.0.1:3306
+
+[sinker]
+db_type=mysql
+sink_type=write
+url=mysql://127.0.0.1:3307
+
+[parallelizer]
+parallel_type=rdb_merge
+
+{extra_config}
+"#
+        )
+    }
+
+    #[test]
+    fn tracing_config_defaults_to_marker_summary_mode() {
+        let config = load_temp_task_config(&basic_snapshot_config(""))
+            .expect("default tracing config should be valid");
+
+        assert_eq!(config.tracing.task_summary_mode, TaskSummaryMode::Marker);
+        assert_eq!(config.tracing.output_format, TraceOutputFormat::Plain);
+    }
+
+    #[test]
+    fn tracing_config_loads_marker_summary_mode_and_json_output() {
+        let config = load_temp_task_config(&basic_snapshot_config(
+            r#"[tracing]
+task_summary_mode=marker
+output_format=json
+"#,
+        ))
+        .expect("tracing config should be valid");
+
+        assert_eq!(config.tracing.task_summary_mode, TaskSummaryMode::Marker);
+        assert_eq!(config.tracing.output_format, TraceOutputFormat::Json);
+    }
+
     #[test]
     fn redis_empty_is_cluster_keeps_auto_detection() {
         let config = load_temp_task_config(

dt-common/src/config/tracing_config.rs

@@ -0,0 +1,7 @@
+use crate::runtime_trace::{TaskSummaryMode, TraceOutputFormat};
+
+#[derive(Clone, Debug, Default)]
+pub struct TracingConfig {
+    pub task_summary_mode: TaskSummaryMode,
+    pub output_format: TraceOutputFormat,
+}

dt-common/src/lib.rs

@@ -6,6 +6,7 @@ pub mod logger;
 pub mod meta;
 pub mod monitor;
 pub mod rdb_filter;
+pub mod runtime_trace;
 pub mod system_dbs;
 pub mod time_filter;
 pub mod utils;

dt-common/src/limiter/base_limiter.rs

@@ -1,6 +1,6 @@
 use async_trait::async_trait;
 
-#[derive(Clone)]
+#[derive(Clone, Copy)]
 pub enum UnitType {
     Bytes,
     Records,

dt-common/src/limiter/capacity_limiter.rs

@@ -30,6 +30,6 @@ impl Limiter for CapacityLimiter {
     }
 
     async fn get_unit_type(&self) -> UnitType {
-        self.unit_type.clone()
+        self.unit_type
     }
 }

dt-common/src/limiter/rate_limiter.rs

@@ -56,6 +56,6 @@ impl Limiter for RateLimiter {
     async fn release(&self, _n: u32) {}
 
     async fn get_unit_type(&self) -> UnitType {
-        self.unit_type.clone()
+        self.unit_type
     }
 }

dt-common/src/logger.rs

@@ -38,6 +38,11 @@ macro_rules! log_summary {
     ($($arg:tt)+) => (log::log!(target: "summary_logger", log::Level::Info, $($arg)+));
 }
 
+#[macro_export(local_inner_macros)]
+macro_rules! log_runtime_trace {
+    ($($arg:tt)+) => (log::log!(target: "runtime_trace_logger", log::Level::Info, $($arg)+));
+}
+
 #[macro_export(local_inner_macros)]
 macro_rules! log_error {
     ($($arg:tt)+) => (log::log!(target: "default_logger", log::Level::Error, $($arg)+))