feat: init support collecting meta node logs into history table (#18551)

* feat: support logging meta node to history table

* add test

* taplo

* clippy

* change config name

* minor fix

Author: dqhl76

Cargo.lock

@@ -35,6 +35,7 @@ databend-common-meta-semaphore = { workspace = true }
 databend-common-meta-sled-store = { workspace = true }
 databend-common-meta-store = { workspace = true }
 databend-common-meta-types = { workspace = true }
+databend-common-storage = { workspace = true }
 databend-common-tracing = { workspace = true }
 databend-common-version = { workspace = true }
 databend-meta = { workspace = true }
@@ -48,6 +49,7 @@ fastrace = { workspace = true }
 futures = { workspace = true }
 log = { workspace = true }
 mlua = { workspace = true }
+num_cpus = { workspace = true }
 rand = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

src/meta/binaries/Cargo.toml

@@ -12,14 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::BTreeMap;
 use std::ops::Deref;
 use std::sync::Arc;
 use std::time::Duration;
 
 use anyerror::AnyError;
+use databend_common_base::base::GlobalInstance;
 use databend_common_base::base::StopHandle;
 use databend_common_base::base::Stoppable;
+use databend_common_base::runtime::GlobalIORuntime;
 use databend_common_grpc::RpcClientConf;
 use databend_common_meta_raft_store::ondisk::OnDisk;
 use databend_common_meta_raft_store::ondisk::DATA_VERSION;
@@ -29,8 +30,9 @@ use databend_common_meta_types::node::Node;
 use databend_common_meta_types::Cmd;
 use databend_common_meta_types::LogEntry;
 use databend_common_meta_types::MetaAPIError;
-use databend_common_tracing::init_logging;
+use databend_common_storage::init_operator;
 use databend_common_tracing::set_panic_hook;
+use databend_common_tracing::GlobalLogger;
 use databend_common_version::DATABEND_COMMIT_VERSION;
 use databend_meta::api::GrpcServer;
 use databend_meta::api::HttpService;
@@ -61,18 +63,7 @@ pub async fn entry(conf: Config) -> anyhow::Result<()> {
 
     set_panic_hook(binary_version);
 
-    // app name format: node_id@cluster_id
-    let app_name_shuffle = format!(
-        "databend-meta-{}@{}",
-        conf.raft_config.id, conf.raft_config.cluster_name
-    );
-    let mut log_labels = BTreeMap::new();
-    log_labels.insert(
-        "cluster_name".to_string(),
-        conf.raft_config.cluster_name.clone(),
-    );
-    let guards = init_logging(&app_name_shuffle, &conf.log, log_labels);
-    Box::new(guards).leak();
+    init_logging_system(&conf).await?;
 
     info!("Databend Meta version: {}", METASRV_COMMIT_VERSION.as_str());
     info!(
@@ -137,6 +128,9 @@ pub async fn entry(conf: Config) -> anyhow::Result<()> {
         if conf.log.tracing.on {
             println!("    Tracing: {}", conf.log.tracing);
         }
+        if conf.log.history.on {
+            println!("    Storage: {}", conf.log.history.on);
+        }
         let r = &conf.raft_config;
         println!("Raft  Id: {}; Cluster: {}", r.id, r.cluster_name);
         println!("      Dir: {}", r.raft_dir);
@@ -380,6 +374,38 @@ async fn run_cmd(conf: &Config) -> bool {
     true
 }
 
+async fn init_logging_system(conf: &Config) -> anyhow::Result<()> {
+    let app_name = format!(
+        "databend-meta-{}@{}",
+        conf.raft_config.id, conf.raft_config.cluster_name
+    );
+
+    let log_labels = [
+        ("cluster_name", conf.raft_config.cluster_name.as_str()),
+        ("node_id", &conf.raft_config.id.to_string()),
+        ("cluster_id", conf.raft_config.cluster_name.as_str()),
+    ]
+    .into_iter()
+    .map(|(k, v)| (k.to_string(), v.to_string()))
+    .collect();
+
+    GlobalInstance::init_production();
+    GlobalLogger::init(&app_name, &conf.log, log_labels);
+
+    if conf.log.history.on {
+        GlobalIORuntime::init(num_cpus::get())?;
+
+        let params = conf.log.history.storage_params.as_ref().ok_or_else(|| {
+            anyhow::anyhow!("Log history is enabled but storage_params is not set")
+        })?;
+
+        let remote_log_op = init_operator(params).map_err(|e| anyhow::anyhow!(e))?;
+        GlobalLogger::instance().set_operator(remote_log_op).await;
+    }
+
+    Ok(())
+}
+
 fn pretty<T>(v: &T) -> Result<String, serde_json::Error>
 where T: serde::Serialize {
     serde_json::to_string_pretty(v)

src/meta/binaries/meta/entry.rs

@@ -23,6 +23,7 @@ async-trait = { workspace = true }
 backon = { workspace = true }
 clap = { workspace = true }
 databend-common-base = { workspace = true }
+databend-common-config = { workspace = true }
 databend-common-grpc = { workspace = true }
 databend-common-http = { workspace = true }
 databend-common-meta-api = { workspace = true }
@@ -33,6 +34,7 @@ databend-common-meta-sled-store = { workspace = true }
 databend-common-meta-stoerr = { workspace = true }
 databend-common-meta-types = { workspace = true }
 databend-common-metrics = { workspace = true }
+databend-common-storage = { workspace = true }
 databend-common-tracing = { workspace = true }
 databend-common-version = { workspace = true }
 deepsize = { workspace = true }

src/meta/service/Cargo.toml

@@ -17,12 +17,14 @@ use std::env;
 use clap::ArgAction;
 use clap::Args;
 use clap::Parser;
+use databend_common_config::StorageConfig;
 use databend_common_meta_raft_store::config::get_default_raft_advertise_host;
 use databend_common_meta_raft_store::config::RaftConfig as InnerRaftConfig;
 use databend_common_meta_types::MetaStartupError;
+use databend_common_storage::StorageConfig as InnerStorageConfig;
 use databend_common_tracing::Config as InnerLogConfig;
 use databend_common_tracing::FileConfig as InnerFileLogConfig;
-use databend_common_tracing::HistoryConfig;
+use databend_common_tracing::HistoryConfig as InnerLogHistoryConfig;
 use databend_common_tracing::OTLPConfig;
 use databend_common_tracing::ProfileLogConfig;
 use databend_common_tracing::QueryLogConfig;
@@ -422,6 +424,7 @@ impl Into<Config> for ConfigViaEnv {
                 stderr_level: self.metasrv_log_stderr_level,
                 stderr_format: self.metasrv_log_stderr_format,
             },
+            storage: StorageLogConfig::default(),
         };
 
         Config {
@@ -671,6 +674,9 @@ pub struct LogConfig {
 
     #[clap(flatten)]
     pub stderr: StderrLogConfig,
+
+    #[clap(flatten)]
+    pub storage: StorageLogConfig,
 }
 
 impl Default for LogConfig {
@@ -690,7 +696,7 @@ impl Into<InnerLogConfig> for LogConfig {
             profile: ProfileLogConfig::default(),
             structlog: StructLogConfig::default(),
             tracing: TracingConfig::default(),
-            history: HistoryConfig::default(),
+            history: self.storage.into(),
         }
     }
 }
@@ -700,6 +706,7 @@ impl From<InnerLogConfig> for LogConfig {
         Self {
             file: inner.file.into(),
             stderr: inner.stderr.into(),
+            storage: inner.history.into(),
         }
     }
 }
@@ -825,3 +832,93 @@ impl From<InnerStderrLogConfig> for StderrLogConfig {
         }
     }
 }
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Args)]
+#[serde(default)]
+pub struct StorageLogConfig {
+    #[clap(
+        long = "log-storage-on", value_name = "VALUE", default_value = "false", action = ArgAction::Set, num_args = 0..=1, require_equals = true, default_missing_value = "true"
+    )]
+    #[serde(rename = "on")]
+    pub log_storage_on: bool,
+
+    /// Specifies the interval in seconds for how often the log is flushed
+    #[clap(
+        long = "log-storage-interval",
+        value_name = "VALUE",
+        default_value = "2"
+    )]
+    #[serde(rename = "interval")]
+    pub log_storage_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-storage-stage-name",
+        value_name = "VALUE",
+        default_value = "log_1f93b76af0bd4b1d8e018667865fbc65"
+    )]
+    #[serde(rename = "stage_name")]
+    pub log_storage_stage_name: String,
+
+    /// Log level <DEBUG|INFO|WARN|ERROR>
+    #[clap(
+        long = "log-storage-level",
+        value_name = "VALUE",
+        default_value = "INFO"
+    )]
+    #[serde(rename = "level")]
+    pub log_storage_level: String,
+
+    /// Specify store the log into where
+    #[clap(skip)]
+    #[serde(rename = "params")]
+    pub log_storage_params: StorageConfig,
+}
+
+impl Default for StorageLogConfig {
+    fn default() -> Self {
+        StorageLogConfig {
+            log_storage_on: false,
+            log_storage_interval: 2,
+            log_storage_stage_name: "log_1f93b76af0bd4b1d8e018667865fbc65".to_string(),
+            log_storage_level: "INFO".to_string(),
+            log_storage_params: Default::default(),
+        }
+    }
+}
+
+#[allow(clippy::from_over_into)]
+impl Into<InnerLogHistoryConfig> for StorageLogConfig {
+    fn into(self) -> InnerLogHistoryConfig {
+        let storage_params: Option<InnerStorageConfig> =
+            Some(self.log_storage_params.try_into().unwrap_or_default());
+        InnerLogHistoryConfig {
+            on: self.log_storage_on,
+            interval: self.log_storage_interval,
+            stage_name: self.log_storage_stage_name,
+            level: self.log_storage_level,
+            storage_params: storage_params.map(|cfg| cfg.params),
+            ..Default::default()
+        }
+    }
+}
+
+impl From<InnerLogHistoryConfig> for StorageLogConfig {
+    fn from(value: InnerLogHistoryConfig) -> Self {
+        let inner_storage_config: Option<InnerStorageConfig> =
+            value.storage_params.map(|params| InnerStorageConfig {
+                params,
+                ..Default::default()
+            });
+        Self {
+            log_storage_on: value.on,
+            log_storage_interval: value.interval,
+            log_storage_stage_name: value.stage_name,
+            log_storage_level: value.level,
+            log_storage_params: inner_storage_config.map(Into::into).unwrap_or_default(),
+        }
+    }
+}

src/meta/service/src/configs/outer_v0.rs

@@ -0,0 +1,12 @@
+[log.storage]
+on = true
+
+[log.storage.params]
+type = "s3"
+
+[log.storage.params.s3]
+bucket = "testbucket"
+root = "admin"
+endpoint_url = "http://127.0.0.1:9900"
+access_key_id = "minioadmin"
+secret_access_key = "minioadmin"

tests/logging/history_table_meta.toml

@@ -2,6 +2,20 @@
 
 set -e
 
+echo "*************************************"
+echo "* Setting STORAGE_TYPE to S3.       *"
+echo "*                                   *"
+echo "* Please make sure that S3 backend  *"
+echo "* is ready, and configured properly.*"
+echo "*************************************"
+export STORAGE_TYPE=s3
+export STORAGE_S3_BUCKET=testbucket
+export STORAGE_S3_ROOT=admin
+export STORAGE_S3_ENDPOINT_URL=http://127.0.0.1:9900
+export STORAGE_S3_ACCESS_KEY_ID=minioadmin
+export STORAGE_S3_SECRET_ACCESS_KEY=minioadmin
+export STORAGE_ALLOW_INSECURE=true
+
 BUILD_PROFILE="${BUILD_PROFILE:-debug}"
 SCRIPT_PATH="$(cd "$(dirname "$0")" >/dev/null 2>&1 && pwd)"
 cd "$SCRIPT_PATH/../../" || exit
@@ -18,10 +32,17 @@ echo "Starting Databend Query cluster with 2 nodes enable history tables"
 for node in 1 2; do
     CONFIG_FILE="./scripts/ci/deploy/config/databend-query-node-${node}.toml"
 
-    echo "Appending history table config to node-${node}"
+    echo "Appending history table config to query node-${node}"
     cat ./tests/logging/history_table.toml >> "$CONFIG_FILE"
 done
 
+for node in 1 2 3; do
+    CONFIG_FILE="./scripts/ci/deploy/config/databend-meta-node-${node}.toml"
+
+    echo "Appending history table config to meta node-${node}"
+    cat ./tests/logging/history_table_meta.toml >> "$CONFIG_FILE"
+done
+
 # Start meta cluster (3 nodes - needed for HA)
 echo 'Start Meta service HA cluster(3 nodes)...'
 
@@ -66,6 +87,17 @@ echo "Running test queries to test inner history tables"
 
 response1=$(curl -s -u root: -XPOST "http://localhost:8000/v1/query" -H 'Content-Type: application/json' -d "{\"sql\": \"select * from system_history.log_history where query_id = '${drop_query_id}'\"}")
 
+meta_count_response=$(curl -s -u root: -XPOST "http://localhost:8000/v1/query" -H 'Content-Type: application/json' -d "{\"sql\": \"select count(*) from system_history.log_history where message like 'Databend Meta version%'\"}")
+
+meta_count_response_data=$(echo "$meta_count_response" | jq -r '.data')
+if [[ "$meta_count_response_data" != *"3"* ]]; then
+    echo "ERROR: meta node logs is empty"
+    echo "meta_count_response_data: $meta_count_response_data"
+    exit 1
+else
+    echo "✓ meta node logs are collected as expected"
+fi
+
 # **Internal -> External**: should reset
 
 echo "Add a node with external history table enabled"