feat: add support for storing logs in table (#17598)

* feat: add support for storing logs in table

* chore: make clippy happy and refine structure

* chore: add rentention time, fix unit test

* chore: clippy

* chore: add buffer limit to prevent log flooding

* fix vacuum

* chore: add level config

* test: add more test for persistent log

* fix: sender should use `blocking_send`

* fix: unit test

* chore: add `retention_frequency` to config

* fixup

* fixup

* taplo

* test: init add logic test

* refactor: make append operation lightly

* feat: add different target and change schema of table

* fixup

* fixup

* fixup

* chore: add a new column to resolve json problem and apply review suggestion

* move retention logic to next PR

* fixup

Author: dqhl76

Cargo.lock

@@ -278,6 +278,7 @@ cidr = { version = "0.2.2" }
 clap = { version = "4.4.2", features = ["derive"] }
 codeq = { version = "0.5.2" }
 comfy-table = "7"
+concurrent-queue = "2.5.0"
 convert_case = "0.6.0"
 cookie = "0.18.1"
 crc32fast = "1.3.2"

Cargo.toml

@@ -65,6 +65,10 @@ otlp_protocol = "http"
 pkey1 = "pvalue1"
 pkey2 = "pvalue2"
 
+[log.persistentlog]
+on = true
+level = "INFO"
+
 [meta]
 endpoints = ["0.0.0.0:9191"]
 username = "root"

scripts/ci/deploy/config/databend-query-node-otlp-logs.toml

@@ -33,6 +33,7 @@ use databend_common_storage::DataOperator;
 use databend_common_tracing::set_panic_hook;
 use databend_query::clusters::ClusterDiscovery;
 use databend_query::local;
+use databend_query::persistent_log::GlobalPersistentLog;
 use databend_query::servers::admin::AdminService;
 use databend_query::servers::flight::FlightService;
 use databend_query::servers::metrics::MetricService;
@@ -281,6 +282,10 @@ pub async fn start_services(conf: &InnerConfig) -> Result<(), MainError> {
     if conf.log.structlog.on {
         println!("    structlog: {}", conf.log.structlog);
     }
+    if conf.log.persistentlog.on {
+        GlobalPersistentLog::instance().initialized();
+        println!("    persistentlog: {}", conf.log.persistentlog);
+    }
 
     println!();
     println!(

src/binaries/query/entry.rs

@@ -7,8 +7,13 @@ publish = { workspace = true }
 edition = { workspace = true }
 
 [dependencies]
+anyhow = { workspace = true }
+arrow-array = { workspace = true }
+arrow-schema = { workspace = true }
+async-channel = { workspace = true }
 backtrace = { workspace = true, features = ["std", "serialize-serde"] }
 chrono = { workspace = true }
+concurrent-queue = { workspace = true }
 databend-common-base = { workspace = true }
 databend-common-exception = { workspace = true }
 defer = { workspace = true }
@@ -18,9 +23,11 @@ itertools = { workspace = true }
 libc = { workspace = true }
 log = { workspace = true }
 logforth = { workspace = true }
+opendal = { workspace = true }
 opentelemetry = { workspace = true }
 opentelemetry-otlp = { workspace = true, features = ["reqwest-client"] }
 opentelemetry_sdk = { workspace = true }
+parquet = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 tonic = { workspace = true }

src/common/tracing/Cargo.toml

@@ -31,6 +31,7 @@ pub struct Config {
     pub profile: ProfileLogConfig,
     pub structlog: StructLogConfig,
     pub tracing: TracingConfig,
+    pub persistentlog: PersistentLogConfig,
 }
 
 impl Config {
@@ -340,3 +341,34 @@ impl Default for OTLPEndpointConfig {
         }
     }
 }
+
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
+pub struct PersistentLogConfig {
+    pub on: bool,
+    pub interval: usize,
+    pub stage_name: String,
+    pub level: String,
+    pub retention: 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
+        )
+    }
+}
+
+impl Default for PersistentLogConfig {
+    fn default() -> Self {
+        Self {
+            on: false,
+            interval: 2,
+            stage_name: "log_1f93b76af0bd4b1d8e018667865fbc65".to_string(),
+            retention: 72,
+            level: "WARN".to_string(),
+        }
+    }
+}

src/common/tracing/src/config.rs

@@ -14,9 +14,11 @@
 
 use std::borrow::Cow;
 use std::collections::BTreeMap;
+use std::sync::Arc;
 use std::time::Duration;
 
 use databend_common_base::base::tokio;
+use databend_common_base::base::tokio::sync::RwLock;
 use databend_common_base::base::GlobalInstance;
 use databend_common_base::runtime::Thread;
 use fastrace::prelude::*;
@@ -25,24 +27,54 @@ use logforth::filter::env::EnvFilterBuilder;
 use logforth::filter::EnvFilter;
 use logforth::Dispatch;
 use logforth::Logger;
+use opendal::Operator;
 use opentelemetry_otlp::WithExportConfig;
 
 use crate::config::OTLPProtocol;
 use crate::loggers::get_layout;
 use crate::loggers::new_rolling_file_appender;
+use crate::remote_log::RemoteLog;
 use crate::structlog::StructLogReporter;
 use crate::Config;
 
 const HEADER_TRACE_PARENT: &str = "traceparent";
 
 pub struct GlobalLogger {
     _drop_guards: Vec<Box<dyn Send + Sync + 'static>>,
+    pub remote_log_operator: RwLock<Option<Operator>>,
 }
 
 impl GlobalLogger {
     pub fn init(name: &str, cfg: &Config, labels: BTreeMap<String, String>) {
         let _drop_guards = init_logging(name, cfg, labels);
-        GlobalInstance::set(Self { _drop_guards });
+
+        // GlobalLogger is initialized before DataOperator, so set the operator to None first
+        let remote_log_operator = RwLock::new(None);
+
+        let instance = Arc::new(Self {
+            _drop_guards,
+            remote_log_operator,
+        });
+        GlobalInstance::set(instance);
+    }
+
+    pub fn instance() -> Arc<GlobalLogger> {
+        GlobalInstance::get()
+    }
+
+    // Get the operator for remote log when it is ready.
+    pub(crate) async fn get_operator(&self) -> Option<Operator> {
+        let operator = self.remote_log_operator.read().await;
+        if let Some(operator) = operator.as_ref() {
+            return Some(operator.clone());
+        }
+        None
+    }
+
+    // Set the operator for remote log, this should be only called once
+    pub async fn set_operator(&self, operator: Operator) {
+        let mut remote_log_operator = self.remote_log_operator.write().await;
+        *remote_log_operator = Some(operator);
     }
 }
 
@@ -100,7 +132,6 @@ pub fn init_logging(
             }
         ),
     };
-
     // initialize tracing a reporter
     if cfg.tracing.on {
         let endpoint = cfg.tracing.otlp.endpoint.clone();
@@ -345,6 +376,36 @@ pub fn init_logging(
         logger = logger.dispatch(dispatch);
     }
 
+    if cfg.persistentlog.on {
+        let (remote_log, flush_guard) =
+            RemoteLog::new(&labels, cfg).expect("initialize remote logger");
+
+        let mut filter_builder =
+            EnvFilterBuilder::new().filter(Some("databend::log::structlog"), LevelFilter::Off);
+
+        if cfg.profile.on && !cfg.profile.dir.is_empty() {
+            filter_builder =
+                filter_builder.filter(Some("databend::log::profile"), LevelFilter::Trace);
+        } else {
+            filter_builder =
+                filter_builder.filter(Some("databend::log::profile"), LevelFilter::Off);
+        }
+        if cfg.query.on && !cfg.query.dir.is_empty() {
+            filter_builder =
+                filter_builder.filter(Some("databend::log::query"), LevelFilter::Trace);
+        } else {
+            filter_builder = filter_builder.filter(Some("databend::log::query"), LevelFilter::Off);
+        }
+        let dispatch = Dispatch::new()
+            .filter(EnvFilter::new(
+                filter_builder.parse(&cfg.persistentlog.level),
+            ))
+            .append(remote_log);
+
+        logger = logger.dispatch(dispatch);
+        _drop_guards.push(flush_guard);
+    }
+
     // set global logger
     if logger.apply().is_err() {
         eprintln!("logger has already been set");

src/common/tracing/src/init.rs

@@ -22,6 +22,7 @@ mod crash_hook;
 mod init;
 mod loggers;
 mod panic_hook;
+mod remote_log;
 mod structlog;
 
 pub use crash_hook::pipe_file;
@@ -32,6 +33,7 @@ pub use crate::config::FileConfig;
 pub use crate::config::OTLPConfig;
 pub use crate::config::OTLPEndpointConfig;
 pub use crate::config::OTLPProtocol;
+pub use crate::config::PersistentLogConfig;
 pub use crate::config::ProfileLogConfig;
 pub use crate::config::QueryLogConfig;
 pub use crate::config::StderrConfig;
@@ -45,6 +47,11 @@ pub use crate::init::start_trace_for_remote_request;
 pub use crate::init::GlobalLogger;
 pub use crate::panic_hook::log_panic;
 pub use crate::panic_hook::set_panic_hook;
+pub use crate::remote_log::convert_to_batch;
+pub use crate::remote_log::LogBuffer as RemoteLogBuffer;
+pub use crate::remote_log::RemoteLog;
+pub use crate::remote_log::RemoteLogElement;
+pub use crate::remote_log::RemoteLogGuard;
 pub use crate::structlog::DummyReporter;
 pub use crate::structlog::StructLogReporter;