fix: prevent log table from logging its own logs (#17802)

* fix: prevent log table from logging its own logs

* add test

* docs: add comments for default stage name

* chore: refine more comments and add a config to control retention frequency

* chore: refine more comments and add a config to control retention frequency

* chore: refine more comments and add a config to control retention frequency

* chore: refine more comments and add a config to control retention frequency

* update config file

* handle cluster mode

* Revert "handle cluster mode"

This reverts commit 69cacfd.

* handle cluster mode

* rebase

Author: dqhl76

src/common/base/src/runtime/runtime_tracker.rs

@@ -103,6 +103,7 @@ pub struct TrackingPayload {
     pub profile: Option<Arc<Profile>>,
     pub mem_stat: Option<Arc<MemStat>>,
     pub metrics: Option<Arc<ScopedRegistry>>,
+    pub should_log: bool,
 }
 
 pub struct TrackingGuard {
@@ -163,6 +164,7 @@ impl ThreadTracker {
                 metrics: None,
                 mem_stat: None,
                 query_id: None,
+                should_log: true,
             },
         }
     }
@@ -243,6 +245,10 @@ impl ThreadTracker {
                 .map(|query_id| unsafe { &*(query_id as *const String) })
         })
     }
+
+    pub fn should_log() -> bool {
+        TRACKER.with(|tracker| tracker.borrow().payload.should_log)
+    }
 }
 
 pin_project! {

src/common/tracing/src/config.rs

@@ -342,14 +342,20 @@ pub struct PersistentLogConfig {
     pub stage_name: String,
     pub level: String,
     pub retention: usize,
+    pub retention_interval: usize,
 }
 
 impl Display for PersistentLogConfig {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(
             f,
-            "enabled={}, interval={}, stage_name={}, level={}, retention={}",
-            self.on, self.interval, self.stage_name, self.level, self.retention
+            "enabled={}, interval={}, stage_name={}, level={}, retention={}, retention_interval={}",
+            self.on,
+            self.interval,
+            self.stage_name,
+            self.level,
+            self.retention,
+            self.retention_interval
         )
     }
 }
@@ -359,9 +365,13 @@ impl Default for PersistentLogConfig {
         Self {
             on: false,
             interval: 2,
+            // The default value of stage name uses an uuid to avoid conflicts with existing stages
             stage_name: "log_1f93b76af0bd4b1d8e018667865fbc65".to_string(),
-            retention: 72,
             level: "WARN".to_string(),
+            // Data older than 72 hours will be deleted during retention tasks
+            retention: 72,
+            // Trigger the retention task every 24 hours
+            retention_interval: 24,
         }
     }
 }

src/common/tracing/src/filter.rs

@@ -0,0 +1,24 @@
+// 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 databend_common_base::runtime::ThreadTracker;
+use logforth::filter::CustomFilter;
+use logforth::filter::FilterResult;
+
+pub fn filter_by_thread_tracker() -> CustomFilter {
+    CustomFilter::new(|_metadata| match ThreadTracker::should_log() {
+        true => FilterResult::Neutral,
+        false => FilterResult::Reject,
+    })
+}

src/common/tracing/src/init.rs

@@ -31,6 +31,7 @@ use opendal::Operator;
 use opentelemetry_otlp::WithExportConfig;
 
 use crate::config::OTLPProtocol;
+use crate::filter::filter_by_thread_tracker;
 use crate::loggers::get_layout;
 use crate::loggers::new_rolling_file_appender;
 use crate::remote_log::RemoteLog;
@@ -219,6 +220,7 @@ pub fn init_logging(
 
         let dispatch = Dispatch::new()
             .filter(env_filter(&cfg.file.level))
+            .filter(filter_by_thread_tracker())
             .append(normal_log_file.with_layout(get_layout(&cfg.file.format)));
         logger = logger.dispatch(dispatch);
     }
@@ -227,6 +229,7 @@ pub fn init_logging(
     if cfg.stderr.on {
         let dispatch = Dispatch::new()
             .filter(env_filter(&cfg.stderr.level))
+            .filter(filter_by_thread_tracker())
             .append(
                 logforth::append::Stderr::default().with_layout(get_layout(&cfg.stderr.format)),
             );
@@ -253,6 +256,7 @@ pub fn init_logging(
             .expect("initialize opentelemetry logger");
         let dispatch = Dispatch::new()
             .filter(env_filter(&cfg.otlp.level))
+            .filter(filter_by_thread_tracker())
             .append(otel);
         logger = logger.dispatch(dispatch);
     }
@@ -261,6 +265,7 @@ pub fn init_logging(
     if cfg.tracing.on || cfg.structlog.on {
         let dispatch = Dispatch::new()
             .filter(env_filter(&cfg.tracing.capture_log_level))
+            .filter(filter_by_thread_tracker())
             .append(logforth::append::FastraceEvent::default());
         logger = logger.dispatch(dispatch);
     }
@@ -281,6 +286,7 @@ pub fn init_logging(
                     EnvFilterBuilder::new()
                         .filter(Some("databend::log::query"), LevelFilter::Trace),
                 ))
+                .filter(filter_by_thread_tracker())
                 .append(query_log_file.with_layout(get_layout("identical")));
             logger = logger.dispatch(dispatch);
         }
@@ -306,6 +312,7 @@ pub fn init_logging(
                     EnvFilterBuilder::new()
                         .filter(Some("databend::log::query"), LevelFilter::Trace),
                 ))
+                .filter(filter_by_thread_tracker())
                 .append(otel);
             logger = logger.dispatch(dispatch);
         }
@@ -327,6 +334,7 @@ pub fn init_logging(
                     EnvFilterBuilder::new()
                         .filter(Some("databend::log::profile"), LevelFilter::Trace),
                 ))
+                .filter(filter_by_thread_tracker())
                 .append(profile_log_file.with_layout(get_layout("identical")));
             logger = logger.dispatch(dispatch);
         }
@@ -352,6 +360,7 @@ pub fn init_logging(
                     EnvFilterBuilder::new()
                         .filter(Some("databend::log::profile"), LevelFilter::Trace),
                 ))
+                .filter(filter_by_thread_tracker())
                 .append(otel);
             logger = logger.dispatch(dispatch);
         }
@@ -372,6 +381,7 @@ pub fn init_logging(
                 EnvFilterBuilder::new()
                     .filter(Some("databend::log::structlog"), LevelFilter::Trace),
             ))
+            .filter(filter_by_thread_tracker())
             .append(structlog_log_file);
         logger = logger.dispatch(dispatch);
     }
@@ -400,6 +410,7 @@ pub fn init_logging(
             .filter(EnvFilter::new(
                 filter_builder.parse(&cfg.persistentlog.level),
             ))
+            .filter(filter_by_thread_tracker())
             .append(remote_log);
 
         logger = logger.dispatch(dispatch);

src/common/tracing/src/lib.rs

@@ -19,6 +19,7 @@
 
 mod config;
 mod crash_hook;
+mod filter;
 mod init;
 mod loggers;
 mod panic_hook;

src/query/config/src/config.rs

@@ -2556,6 +2556,9 @@ pub struct PersistentLogConfig {
     pub log_persistentlog_interval: usize,
 
     /// Specifies the name of the staging area that temporarily holds log data before it is finally copied into the table
+    ///
+    /// Note:
+    /// The default value uses an uuid to avoid conflicts with existing stages
     #[clap(
         long = "log-persistentlog-stage-name",
         value_name = "VALUE",
@@ -2564,7 +2567,17 @@ pub struct PersistentLogConfig {
     #[serde(rename = "stage_name")]
     pub log_persistentlog_stage_name: String,
 
-    /// Specifies how long the persistent log should be retained, in hours
+    /// Log level <DEBUG|INFO|WARN|ERROR>
+    #[clap(
+        long = "log-persistentlog-level",
+        value_name = "VALUE",
+        default_value = "WARN"
+    )]
+    #[serde(rename = "level")]
+    pub log_persistentlog_level: String,
+
+    /// The retention period (in hours) for persistent logs.
+    /// Data older than this period will be deleted during retention tasks.
     #[clap(
         long = "log-persistentlog-retention",
         value_name = "VALUE",
@@ -2573,14 +2586,15 @@ pub struct PersistentLogConfig {
     #[serde(rename = "retention")]
     pub log_persistentlog_retention: usize,
 
-    /// Log level <DEBUG|INFO|WARN|ERROR>
+    /// The interval (in hours) at which the retention process is triggered.
+    /// Specifies how often the retention task runs to clean up old data.
     #[clap(
-        long = "log-persistentlog-level",
+        long = "log-persistentlog-retention-interval",
         value_name = "VALUE",
-        default_value = "WARN"
+        default_value = "24"
     )]
-    #[serde(rename = "level")]
-    pub log_persistentlog_level: String,
+    #[serde(rename = "retention_interval")]
+    pub log_persistentlog_retention_interval: usize,
 }
 
 impl Default for PersistentLogConfig {
@@ -2599,6 +2613,7 @@ impl TryInto<InnerPersistentLogConfig> for PersistentLogConfig {
             stage_name: self.log_persistentlog_stage_name,
             level: self.log_persistentlog_level,
             retention: self.log_persistentlog_retention,
+            retention_interval: self.log_persistentlog_retention_interval,
         })
     }
 }
@@ -2611,6 +2626,7 @@ impl From<InnerPersistentLogConfig> for PersistentLogConfig {
             log_persistentlog_stage_name: inner.stage_name,
             log_persistentlog_level: inner.level,
             log_persistentlog_retention: inner.retention,
+            log_persistentlog_retention_interval: inner.retention_interval,
         }
     }
 }