[store]: refactor: upgrade raft to 0.6.2-alpha.10

common/tracing/src/lib.rs

@@ -16,7 +16,23 @@ mod logging;
 mod tracing_to_jaeger;
 
 pub use logging::init_default_tracing;
+pub use logging::init_global_tracing;
+pub use logging::init_tracing;
 pub use logging::init_tracing_with_file;
 pub use tracing;
 pub use tracing_to_jaeger::extract_remote_span_as_parent;
 pub use tracing_to_jaeger::inject_span_to_tonic_request;
+
+#[macro_export]
+macro_rules! func_name {
+    () => {{
+        fn f() {}
+        fn type_name_of<T>(_: T) -> &'static str {
+            std::any::type_name::<T>()
+        }
+        let name = type_name_of(f);
+        let n = &name[..name.len() - 3];
+        let nn = n.replace("::{{closure}}", "");
+        nn
+    }};
+}

common/tracing/src/logging.rs

@@ -13,10 +13,13 @@
 // limitations under the License.
 
 use std::env;
+use std::fs::OpenOptions;
+use std::path::Path;
 use std::sync::Once;
 
 use opentelemetry::global;
 use opentelemetry::sdk::propagation::TraceContextPropagator;
+use tracing::Subscriber;
 use tracing_appender::non_blocking::WorkerGuard;
 use tracing_appender::rolling::RollingFileAppender;
 use tracing_appender::rolling::Rotation;
@@ -28,6 +31,8 @@ use tracing_subscriber::prelude::*;
 use tracing_subscriber::registry::Registry;
 use tracing_subscriber::EnvFilter;
 
+use crate::tracing::subscriber::DefaultGuard;
+
 /// Write logs to stdout.
 pub fn init_default_tracing() {
     static START: Once = Once::new();
@@ -54,12 +59,25 @@ fn init_tracing_stdout() {
     let fmt_layer = Layer::default()
         .with_thread_ids(true)
         .with_thread_names(true)
-        .pretty()
-        .with_ansi(true)
+        // .pretty()
+        .with_ansi(false)
         .with_span_events(fmt::format::FmtSpan::FULL);
 
+    let subscriber = Registry::default()
+        .with(EnvFilter::from_default_env())
+        .with(fmt_layer)
+        .with(jaeger_layer());
+
+    tracing::subscriber::set_global_default(subscriber)
+        .expect("error setting global tracing subscriber");
+}
+
+fn jaeger_layer<
+    S: tracing::Subscriber + for<'span> tracing_subscriber::registry::LookupSpan<'span>,
+>() -> Option<impl tracing_subscriber::layer::Layer<S>> {
     let fuse_jaeger = env::var("FUSE_JAEGER").unwrap_or_else(|_| "".to_string());
-    let ot_layer = if !fuse_jaeger.is_empty() {
+
+    if !fuse_jaeger.is_empty() {
         global::set_text_map_propagator(TraceContextPropagator::new());
 
         let tracer = opentelemetry_jaeger::new_pipeline()
@@ -71,15 +89,7 @@ fn init_tracing_stdout() {
         Some(ot_layer)
     } else {
         None
-    };
-
-    let subscriber = Registry::default()
-        .with(EnvFilter::from_default_env())
-        .with(fmt_layer)
-        .with(ot_layer);
-
-    tracing::subscriber::set_global_default(subscriber)
-        .expect("error setting global tracing subscriber");
+    }
 }
 
 /// Write logs to file and rotation by HOUR.
@@ -100,8 +110,68 @@ pub fn init_tracing_with_file(app_name: &str, dir: &str, level: &str) -> Vec<Wor
         .with(stdout_logging_layer)
         .with(JsonStorageLayer)
         .with(file_logging_layer);
+
     tracing::subscriber::set_global_default(subscriber)
         .expect("error setting global tracing subscriber");
 
     guards
 }
+
+/// Creates a tracing/logging subscriber that is valid until the guards are dropped.
+/// The format layer logging span/event in plain text, without color, one event per line.
+/// This is useful in a unit test.
+pub fn init_tracing(app_name: &str, dir: &str) -> (WorkerGuard, DefaultGuard) {
+    let (g, sub) = init_file_subscriber(app_name, dir);
+
+    let subs_guard = tracing::subscriber::set_default(sub);
+
+    tracing::info!("initialized tracing");
+    (g, subs_guard)
+}
+
+/// Creates a global tracing/logging subscriber which saves events in one log file.
+pub fn init_global_tracing(app_name: &str, dir: &str) -> WorkerGuard {
+    let (g, sub) = init_file_subscriber(app_name, dir);
+    tracing::subscriber::set_global_default(sub).expect("error setting global tracing subscriber");
+
+    tracing::info!("initialized global tracing");
+    g
+}
+
+/// Create a file based tracing/logging subscriber.
+/// A guard must be held during using the logging.
+/// The format layer logging span/event in plain text, without color, one event per line.
+/// Optionally it adds a layer to send to opentelemetry if env var `FUSE_JAEGER` is present.
+pub fn init_file_subscriber(app_name: &str, dir: &str) -> (WorkerGuard, impl Subscriber) {
+    let path_str = dir.to_string() + "/" + app_name;
+    let path: &Path = path_str.as_ref();
+
+    let mut open_options = OpenOptions::new();
+    open_options.append(true).create(true);
+
+    let mut open_res = open_options.open(path);
+    if open_res.is_err() {
+        if let Some(parent) = path.parent() {
+            std::fs::create_dir_all(parent).unwrap();
+            open_res = open_options.open(path);
+        }
+    }
+
+    let f = open_res.unwrap();
+
+    let (writer, writer_guard) = tracing_appender::non_blocking(f);
+
+    let f_layer = Layer::new()
+        .with_writer(writer)
+        .with_thread_ids(true)
+        .with_thread_names(true)
+        .with_ansi(false)
+        .with_span_events(fmt::format::FmtSpan::FULL);
+
+    let subscriber = Registry::default()
+        .with(EnvFilter::from_default_env())
+        .with(f_layer)
+        .with(jaeger_layer());
+
+    (writer_guard, subscriber)
+}

store/Cargo.toml

@@ -37,7 +37,7 @@ common-tracing = {path = "../common/tracing"}
 
 # Crates.io dependencies
 anyhow = "1.0.43"
-async-raft = { git = "https://github.com/datafuse-extras/async-raft", tag = "v0.6.2-alpha.9" }
+async-raft = { git = "https://github.com/datafuse-extras/async-raft", tag = "v0.6.2-alpha.10" }
 async-trait = "0.1"
 byteorder = "1.1.0"
 env_logger = "0.9"
@@ -62,6 +62,7 @@ tempfile = "3.2.0"
 thiserror = "1.0.26"
 threadpool = "1.8.1"
 tokio-stream = "0.1"
+tracing-appender = "0.1.2"
 tonic = { version = "0.4.3", features = ["tls"]}
 
 sha2 = "0.9.5"

store/src/api/rpc/flight_service_test.rs

@@ -47,7 +47,7 @@ async fn test_flight_restart() -> anyhow::Result<()> {
     // - restart
     // - Test read the db and read the table.
 
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
 
     let (mut tc, addr) = crate::tests::start_store_server().await?;
 
@@ -165,7 +165,7 @@ async fn test_flight_restart() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_flight_create_database() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
 
     // 1. Service starts.
     let (_tc, addr) = crate::tests::start_store_server().await?;
@@ -231,7 +231,7 @@ async fn test_flight_create_database() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_flight_create_get_table() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
     use std::sync::Arc;
 
     use common_datavalues::DataField;
@@ -358,7 +358,7 @@ async fn test_flight_create_get_table() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_flight_drop_table() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
     use std::sync::Arc;
 
     use common_datavalues::DataField;
@@ -463,7 +463,7 @@ async fn test_flight_drop_table() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_do_append() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
 
     use std::sync::Arc;
 
@@ -537,7 +537,7 @@ async fn test_do_append() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_scan_partition() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
     use std::sync::Arc;
 
     use common_datavalues::prelude::*;
@@ -631,7 +631,7 @@ async fn test_scan_partition() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_flight_generic_kv_mget() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
     {
         let span = tracing::span!(tracing::Level::INFO, "test_flight_generic_kv_list");
         let _ent = span.enter();
@@ -678,7 +678,7 @@ async fn test_flight_generic_kv_mget() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_flight_generic_kv_list() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
     {
         let span = tracing::span!(tracing::Level::INFO, "test_flight_generic_kv_list");
         let _ent = span.enter();
@@ -725,7 +725,7 @@ async fn test_flight_generic_kv_list() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_flight_generic_kv_delete() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
     {
         let span = tracing::span!(tracing::Level::INFO, "test_flight_generic_kv_list");
         let _ent = span.enter();
@@ -792,7 +792,7 @@ async fn test_flight_generic_kv_delete() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_flight_generic_kv_update() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
     {
         let span = tracing::span!(tracing::Level::INFO, "test_flight_generic_kv_list");
         let _ent = span.enter();
@@ -900,7 +900,7 @@ async fn test_flight_generic_kv_timeout() -> anyhow::Result<()> {
     // - Test list expired and non-expired.
     // - Test update with a new expire value.
 
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
     {
         let span = tracing::span!(tracing::Level::INFO, "test_flight_generic_kv_timeout");
         let _ent = span.enter();
@@ -1012,7 +1012,7 @@ async fn test_flight_generic_kv_timeout() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_flight_generic_kv() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
 
     {
         let span = tracing::span!(tracing::Level::INFO, "test_flight_generic_kv");
@@ -1087,7 +1087,7 @@ async fn test_flight_generic_kv() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_flight_get_database_meta_empty_db() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
     let (_tc, addr) = crate::tests::start_store_server().await?;
     let mut client = StoreClient::try_create(addr.as_str(), "root", "xxx").await?;
 
@@ -1100,7 +1100,7 @@ async fn test_flight_get_database_meta_empty_db() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_flight_get_database_meta_ddl_db() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
     let (_tc, addr) = crate::tests::start_store_server().await?;
     let mut client = StoreClient::try_create(addr.as_str(), "root", "xxx").await?;
 
@@ -1161,7 +1161,7 @@ async fn test_flight_get_database_meta_ddl_db() -> anyhow::Result<()> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_flight_get_database_meta_ddl_table() -> anyhow::Result<()> {
-    init_store_unittest();
+    let _log_guards = init_store_unittest(&common_tracing::func_name!());
     let (_tc, addr) = crate::tests::start_store_server().await?;
     let mut client = StoreClient::try_create(addr.as_str(), "root", "xxx").await?;
 

store/src/configs/config.rs

@@ -41,6 +41,10 @@ lazy_static! {
 
 #[derive(Clone, Debug, serde::Deserialize, PartialEq, StructOpt, StructOptToml)]
 pub struct Config {
+    /// Identify a config. This is only meant to make debugging easier with more than one Config involved.
+    #[structopt(long, default_value = "")]
+    pub config_id: String,
+
     #[structopt(long, env = "STORE_LOG_LEVEL", default_value = "INFO")]
     pub log_level: String,