refactor: remove re-export from logging (#3865)

* refactor: remove re-export from logging Signed-off-by: Ruihang Xia <[email protected]> * fix merge problem Signed-off-by: Ruihang Xia <[email protected]> * run formatter Signed-off-by: Ruihang Xia <[email protected]> --------- Signed-off-by: Ruihang Xia <[email protected]>
GreptimeTeam · May 6, 2024 · 5303537 · 5303537
1 parent 573c19b
commit 5303537
Show file tree

Hide file tree

Showing 38 changed files with 112 additions and 120 deletions.
diff --git a/src/catalog/src/information_schema/cluster_info.rs b/src/catalog/src/information_schema/cluster_info.rs
@@ -24,7 +24,7 @@ use common_meta::peer::Peer;
 use common_query::physical_plan::TaskContext;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
-use common_telemetry::logging::warn;
+use common_telemetry::warn;
 use common_time::timestamp::Timestamp;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;

diff --git a/src/cmd/src/cli/repl.rs b/src/cmd/src/cli/repl.rs
@@ -26,7 +26,7 @@ use common_error::ext::ErrorExt;
 use common_meta::cache_invalidator::MultiCacheInvalidator;
 use common_query::Output;
 use common_recordbatch::RecordBatches;
-use common_telemetry::logging;
+use common_telemetry::debug;
 use either::Either;
 use meta_client::client::MetaClientBuilder;
 use query::datafusion::DatafusionQueryEngine;
@@ -78,7 +78,7 @@ impl Repl {
 
             let history_file = history_file();
             if let Err(e) = rl.load_history(&history_file) {
-                logging::debug!(
+                debug!(
                     "failed to load history file on {}, error: {e}",
                     history_file.display()
                 );
@@ -225,7 +225,7 @@ impl Drop for Repl {
         if self.rl.helper().is_some() {
             let history_file = history_file();
             if let Err(e) = self.rl.save_history(&history_file) {
-                logging::debug!(
+                debug!(
                     "failed to save history file on {}, error: {e}",
                     history_file.display()
                 );

diff --git a/src/cmd/src/datanode.rs b/src/cmd/src/datanode.rs
@@ -18,7 +18,7 @@ use std::time::Duration;
 use async_trait::async_trait;
 use catalog::kvbackend::MetaKvBackend;
 use clap::Parser;
-use common_telemetry::{info, logging};
+use common_telemetry::info;
 use common_wal::config::DatanodeWalConfig;
 use datanode::config::DatanodeOptions;
 use datanode::datanode::{Datanode, DatanodeBuilder};
@@ -210,8 +210,8 @@ impl StartCommand {
             .await
             .context(StartDatanodeSnafu)?;
 
-        logging::info!("Datanode start command: {:#?}", self);
-        logging::info!("Datanode options: {:#?}", opts);
+        info!("Datanode start command: {:#?}", self);
+        info!("Datanode options: {:#?}", opts);
 
         let node_id = opts
             .node_id

diff --git a/src/cmd/src/frontend.rs b/src/cmd/src/frontend.rs
@@ -22,7 +22,7 @@ use client::client_manager::DatanodeClients;
 use common_meta::cache_invalidator::MultiCacheInvalidator;
 use common_meta::heartbeat::handler::parse_mailbox_message::ParseMailboxMessageHandler;
 use common_meta::heartbeat::handler::HandlerGroupExecutor;
-use common_telemetry::logging;
+use common_telemetry::info;
 use common_time::timezone::set_default_timezone;
 use frontend::frontend::FrontendOptions;
 use frontend::heartbeat::handler::invalidate_table_cache::InvalidateTableCacheHandler;
@@ -219,8 +219,8 @@ impl StartCommand {
             .await
             .context(StartFrontendSnafu)?;
 
-        logging::info!("Frontend start command: {:#?}", self);
-        logging::info!("Frontend options: {:#?}", opts);
+        info!("Frontend start command: {:#?}", self);
+        info!("Frontend options: {:#?}", opts);
 
         set_default_timezone(opts.default_timezone.as_deref()).context(InitTimezoneSnafu)?;
 

diff --git a/src/cmd/src/metasrv.rs b/src/cmd/src/metasrv.rs
@@ -16,7 +16,7 @@ use std::time::Duration;
 
 use async_trait::async_trait;
 use clap::Parser;
-use common_telemetry::logging;
+use common_telemetry::info;
 use meta_srv::bootstrap::MetasrvInstance;
 use meta_srv::metasrv::MetasrvOptions;
 use snafu::ResultExt;
@@ -198,8 +198,8 @@ impl StartCommand {
             .await
             .context(StartMetaServerSnafu)?;
 
-        logging::info!("Metasrv start command: {:#?}", self);
-        logging::info!("Metasrv options: {:#?}", opts);
+        info!("Metasrv start command: {:#?}", self);
+        info!("Metasrv options: {:#?}", opts);
 
         let builder = meta_srv::bootstrap::metasrv_builder(&opts, plugins.clone(), None)
             .await

diff --git a/src/common/function/src/table/migrate_region.rs b/src/common/function/src/table/migrate_region.rs
@@ -20,7 +20,7 @@ use common_meta::rpc::procedure::MigrateRegionRequest;
 use common_query::error::Error::ThreadJoin;
 use common_query::error::{InvalidFuncArgsSnafu, MissingProcedureServiceHandlerSnafu, Result};
 use common_query::prelude::{Signature, TypeSignature, Volatility};
-use common_telemetry::logging::error;
+use common_telemetry::error;
 use datatypes::data_type::DataType;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::value::{Value, ValueRef};

diff --git a/src/common/procedure/src/local.rs b/src/common/procedure/src/local.rs
@@ -24,7 +24,7 @@ use async_trait::async_trait;
 use backon::ExponentialBuilder;
 use common_runtime::{RepeatedTask, TaskFunction};
 use common_telemetry::tracing_context::{FutureExt, TracingContext};
-use common_telemetry::{info, logging, tracing};
+use common_telemetry::{error, info, tracing};
 use snafu::{ensure, ResultExt};
 use tokio::sync::watch::{self, Receiver, Sender};
 use tokio::sync::{Mutex as TokioMutex, Notify};
@@ -244,20 +244,18 @@ impl ManagerContext {
     ) -> Option<LoadedProcedure> {
         let loaders = self.loaders.lock().unwrap();
         let loader = loaders.get(&message.type_name).or_else(|| {
-            logging::error!(
+            error!(
                 "Loader not found, procedure_id: {}, type_name: {}",
-                procedure_id,
-                message.type_name
+                procedure_id, message.type_name
             );
             None
         })?;
 
         let procedure = loader(&message.data)
             .map_err(|e| {
-                logging::error!(
+                error!(
                     "Failed to load procedure data, key: {}, source: {:?}",
-                    procedure_id,
-                    e
+                    procedure_id, e
                 );
                 e
             })
@@ -496,7 +494,7 @@ impl LocalManager {
                     continue;
                 };
 
-                logging::info!(
+                info!(
                     "Recover root procedure {}-{}, step: {}",
                     loaded_procedure.procedure.type_name(),
                     procedure_id,
@@ -521,15 +519,15 @@ impl LocalManager {
                     loaded_procedure.step,
                     loaded_procedure.procedure,
                 ) {
-                    logging::error!(e; "Failed to recover procedure {}", procedure_id);
+                    error!(e; "Failed to recover procedure {}", procedure_id);
                 }
             }
         }
     }
 
     /// Recovers unfinished procedures and reruns them.
     async fn recover(&self) -> Result<()> {
-        logging::info!("LocalManager start to recover");
+        info!("LocalManager start to recover");
         let recover_start = Instant::now();
 
         let (messages, rollback_messages, finished_ids) =
@@ -539,19 +537,19 @@ impl LocalManager {
         self.submit_recovered_messages(messages, InitProcedureState::Running);
 
         if !finished_ids.is_empty() {
-            logging::info!(
+            info!(
                 "LocalManager try to clean finished procedures, num: {}",
                 finished_ids.len()
             );
 
             for procedure_id in finished_ids {
                 if let Err(e) = self.procedure_store.delete_procedure(procedure_id).await {
-                    logging::error!(e; "Failed to delete procedure {}", procedure_id);
+                    error!(e; "Failed to delete procedure {}", procedure_id);
                 }
             }
         }
 
-        logging::info!(
+        info!(
             "LocalManager finish recovery, cost: {}ms",
             recover_start.elapsed().as_millis()
         );

diff --git a/src/common/procedure/src/local/runner.rs b/src/common/procedure/src/local/runner.rs
@@ -16,7 +16,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use backon::{BackoffBuilder, ExponentialBuilder};
-use common_telemetry::logging;
+use common_telemetry::{debug, error, info};
 use tokio::time;
 
 use super::rwlock::OwnedKeyRwLockGuard;
@@ -54,7 +54,7 @@ impl ProcedureGuard {
 impl Drop for ProcedureGuard {
     fn drop(&mut self) {
         if !self.finish {
-            logging::error!("Procedure {} exits unexpectedly", self.meta.id);
+            error!("Procedure {} exits unexpectedly", self.meta.id);
 
             // Set state to failed. This is useful in test as runtime may not abort when the runner task panics.
             // See https://github.com/tokio-rs/tokio/issues/2002 .
@@ -104,7 +104,7 @@ impl Runner {
         // Ensure we can update the procedure state.
         let mut guard = ProcedureGuard::new(self.meta.clone(), self.manager_ctx.clone());
 
-        logging::info!(
+        info!(
             "Runner {}-{} starts",
             self.procedure.type_name(),
             self.meta.id
@@ -149,7 +149,7 @@ impl Runner {
 
             for id in procedure_ids {
                 if let Err(e) = self.store.delete_procedure(id).await {
-                    logging::error!(
+                    error!(
                         e;
                         "Runner {}-{} failed to delete procedure {}",
                         self.procedure.type_name(),
@@ -160,7 +160,7 @@ impl Runner {
             }
         }
 
-        logging::info!(
+        info!(
             "Runner {}-{} exits",
             self.procedure.type_name(),
             self.meta.id
@@ -260,7 +260,7 @@ impl Runner {
             ProcedureState::Running | ProcedureState::Retrying { .. } => {
                 match self.procedure.execute(ctx).await {
                     Ok(status) => {
-                        logging::debug!(
+                        debug!(
                             "Execute procedure {}-{} once, status: {:?}, need_persist: {}",
                             self.procedure.type_name(),
                             self.meta.id,
@@ -299,7 +299,7 @@ impl Runner {
                         }
                     }
                     Err(e) => {
-                        logging::error!(
+                        error!(
                             e;
                             "Failed to execute procedure {}-{}, retry: {}",
                             self.procedure.type_name(),
@@ -394,7 +394,7 @@ impl Runner {
 
     /// Extend the retry time to wait for the next retry.
     async fn wait_on_err(&mut self, d: Duration, i: u64) {
-        logging::info!(
+        info!(
             "Procedure {}-{} retry for the {} times after {} millis",
             self.procedure.type_name(),
             self.meta.id,
@@ -407,7 +407,7 @@ impl Runner {
     async fn on_suspended(&mut self, subprocedures: Vec<ProcedureWithId>) {
         let has_child = !subprocedures.is_empty();
         for subprocedure in subprocedures {
-            logging::info!(
+            info!(
                 "Procedure {}-{} submit subprocedure {}-{}",
                 self.procedure.type_name(),
                 self.meta.id,
@@ -422,7 +422,7 @@ impl Runner {
             );
         }
 
-        logging::info!(
+        info!(
             "Procedure {}-{} is waiting for subprocedures",
             self.procedure.type_name(),
             self.meta.id,
@@ -432,7 +432,7 @@ impl Runner {
         if has_child {
             self.meta.child_notify.notified().await;
 
-            logging::info!(
+            info!(
                 "Procedure {}-{} is waked up",
                 self.procedure.type_name(),
                 self.meta.id,
@@ -454,7 +454,7 @@ impl Runner {
             )
             .await
             .map_err(|e| {
-                logging::error!(
+                error!(
                     e; "Failed to persist procedure {}-{}",
                     self.procedure.type_name(),
                     self.meta.id
@@ -470,7 +470,7 @@ impl Runner {
             .commit_procedure(self.meta.id, self.step)
             .await
             .map_err(|e| {
-                logging::error!(
+                error!(
                     e; "Failed to commit procedure {}-{}",
                     self.procedure.type_name(),
                     self.meta.id
@@ -496,7 +496,7 @@ impl Runner {
             .rollback_procedure(self.meta.id, message)
             .await
             .map_err(|e| {
-                logging::error!(
+                error!(
                     e; "Failed to write rollback key for procedure {}-{}",
                     self.procedure.type_name(),
                     self.meta.id
@@ -509,7 +509,7 @@ impl Runner {
 
     fn done(&self, output: Option<Output>) {
         // TODO(yingwen): Add files to remove list.
-        logging::info!(
+        info!(
             "Procedure {}-{} done",
             self.procedure.type_name(),
             self.meta.id,

diff --git a/src/common/procedure/src/store.rs b/src/common/procedure/src/store.rs
@@ -15,7 +15,7 @@
 use std::collections::HashMap;
 use std::fmt;
 
-use common_telemetry::logging;
+use common_telemetry::{debug, error, info, warn};
 use futures::TryStreamExt;
 use serde::{Deserialize, Serialize};
 use snafu::ResultExt;
@@ -65,7 +65,7 @@ impl ProcedureStore {
     /// Creates a new [ProcedureStore] from specific [StateStoreRef].
     pub(crate) fn new(parent_path: &str, store: StateStoreRef) -> ProcedureStore {
         let proc_path = format!("{}{PROC_PATH}", parent_path);
-        logging::info!("The procedure state store path is: {}", &proc_path);
+        info!("The procedure state store path is: {}", &proc_path);
         ProcedureStore { proc_path, store }
     }
 
@@ -154,7 +154,7 @@ impl ProcedureStore {
         while let Some((key_set, _)) = key_values.try_next().await? {
             let key = key_set.key();
             let Some(curr_key) = ParsedKey::parse_str(&self.proc_path, key) else {
-                logging::warn!("Unknown key while deleting procedures, key: {}", key);
+                warn!("Unknown key while deleting procedures, key: {}", key);
                 continue;
             };
             if curr_key.key_type == KeyType::Step {
@@ -165,11 +165,9 @@ impl ProcedureStore {
             }
         }
 
-        logging::debug!(
+        debug!(
             "Delete keys for procedure {}, step_keys: {:?}, finish_keys: {:?}",
-            procedure_id,
-            step_keys,
-            finish_keys
+            procedure_id, step_keys, finish_keys
         );
         // We delete all step keys first.
         self.store.batch_delete(step_keys.as_slice()).await?;
@@ -203,7 +201,7 @@ impl ProcedureStore {
         while let Some((key_set, value)) = key_values.try_next().await? {
             let key = key_set.key();
             let Some(curr_key) = ParsedKey::parse_str(&self.proc_path, key) else {
-                logging::warn!("Unknown key while loading procedures, key: {}", key);
+                warn!("Unknown key while loading procedures, key: {}", key);
                 continue;
             };
 
@@ -251,7 +249,7 @@ impl ProcedureStore {
         serde_json::from_slice(value)
             .map_err(|e| {
                 // `e` doesn't impl ErrorExt so we print it as normal error.
-                logging::error!("Failed to parse value, key: {:?}, source: {:?}", key, e);
+                error!("Failed to parse value, key: {:?}, source: {:?}", key, e);
                 e
             })
             .ok()