rpc: implement logs subscription

ckb-bin/Cargo.toml

@@ -40,6 +40,7 @@ ckb-async-runtime.workspace = true
 ckb-migrate.workspace = true
 ckb-launcher.workspace = true
 ckb-constant.workspace = true
+ckb-logger-config.workspace = true
 base64.workspace = true
 tempfile.workspace = true
 rayon.workspace = true
@@ -48,6 +49,7 @@ is-terminal.workspace = true
 fdlimit.workspace = true
 ckb-stop-handler.workspace = true
 tokio = { workspace = true, features = ["sync"] }
+time = { workspace = true }
 
 [target.'cfg(not(target_os="windows"))'.dependencies]
 daemonize-me = { version = "2" }

ckb-bin/src/lib.rs

@@ -12,12 +12,13 @@ mod tests;
 use ckb_app_config::ExitCode;
 use ckb_async_runtime::new_global_runtime;
 use ckb_build_info::Version;
-use ckb_logger::{debug, info};
+use ckb_logger::debug;
 use ckb_network::tokio;
 use clap::ArgMatches;
 use helper::raise_fd_limit;
 use setup::Setup;
 use setup_guard::SetupGuard;
+use time::OffsetDateTime;
 
 #[cfg(not(target_os = "windows"))]
 use colored::Colorize;
@@ -28,6 +29,26 @@ use subcommand::check_process;
 #[cfg(feature = "with_sentry")]
 pub(crate) const LOG_TARGET_SENTRY: &str = "sentry";
 
+/// Print a log-like message to stderr.
+/// Format: `YYYY-MM-DD HH:MM:SS.mmm +00:00 thread_name INFO module  message`
+fn log_println(message: &str) {
+    let now = OffsetDateTime::now_utc();
+    let thread = std::thread::current();
+    let thread_name = thread.name().unwrap_or("main");
+    eprintln!(
+        "{:04}-{:02}-{:02} {:02}:{:02}:{:02}.{:03} +00:00 {} INFO ckb_bin  {}",
+        now.year(),
+        now.month() as u8,
+        now.day(),
+        now.hour(),
+        now.minute(),
+        now.second(),
+        now.millisecond(),
+        thread_name,
+        message
+    );
+}
+
 /// The executable main entry.
 ///
 /// It returns `Ok` when the process exist normally, otherwise the `ExitCode` is converted to the
@@ -109,11 +130,11 @@ fn run_app_in_daemon(
 
     match daemon.start() {
         Ok(_) => {
-            info!("Success, daemonized ...");
+            ckb_logger::info!("Success, daemonized ...");
             run_app_inner(version, bin_name, cmd, matches)
         }
         Err(e) => {
-            info!("daemonize error: {}", e);
+            ckb_logger::info!("daemonize error: {}", e);
             Err(ExitCode::Failure)
         }
     }
@@ -128,12 +149,21 @@ fn run_app_inner(
     let is_silent_logging = is_silent_logging(cmd);
     let (mut handle, mut handle_stop_rx, _runtime) = new_global_runtime(None);
     let setup = Setup::from_matches(bin_name, cmd, matches)?;
-    let _guard = SetupGuard::from_setup(&setup, &version, handle.clone(), is_silent_logging)?;
+    // Disable logging here if the user is executing `ckb run`.
+    // Logs subscription of RPC service requires access to `struct Shared`,
+    // so logger of `ckb run` will be initialized in `subcommand::run`.
+    let (_guard, log_config) = SetupGuard::from_setup(
+        &setup,
+        &version,
+        handle.clone(),
+        is_silent_logging,
+        cmd != cli::CMD_RUN,
+    )?;
 
     raise_fd_limit();
 
     let ret = match cmd {
-        cli::CMD_RUN => subcommand::run(setup.run(matches)?, version, handle.clone()),
+        cli::CMD_RUN => subcommand::run(setup.run(matches)?, version, handle.clone(), log_config),
         cli::CMD_MINER => subcommand::miner(setup.miner(matches)?, handle.clone()),
         cli::CMD_REPLAY => subcommand::replay(setup.replay(matches)?, handle.clone()),
         cli::CMD_EXPORT => subcommand::export(setup.export(matches)?, handle.clone()),
@@ -150,9 +180,18 @@ fn run_app_inner(
         handle.drop_guard();
 
         tokio::task::block_in_place(|| {
-            info!("Waiting for all tokio tasks to exit...");
+            // Here we use `log_println` instead of `info!` because the Logger has already been
+            // shut down when `subcommand::run()` returned (LoggerGuard was dropped there).
+            //
+            // We cannot simply extend the LoggerGuard's lifetime to here because it would cause
+            // a deadlock: NotifyController (held by Logger's background thread) contains a clone
+            // of Handle, which holds a clone of `guard` (Sender). If LoggerGuard is not dropped
+            // before `handle_stop_rx.blocking_recv()`, the Logger thread won't terminate, and
+            // the Sender inside NotifyController won't be dropped, causing blocking_recv() to
+            // wait forever.
+            log_println("Waiting for all tokio tasks to exit...");
             handle_stop_rx.blocking_recv();
-            info!("All tokio tasks and threads have exited. CKB shutdown");
+            log_println("All tokio tasks and threads have exited. CKB shutdown");
         });
     }
 

ckb-bin/src/setup_guard.rs

@@ -6,10 +6,10 @@ use ckb_metrics_service::{self, Guard as MetricsInitGuard};
 
 use crate::setup::Setup;
 
-const CKB_LOG_ENV: &str = "CKB_LOG";
+pub const CKB_LOG_ENV: &str = "CKB_LOG";
 
 pub struct SetupGuard {
-    _logger_guard: LoggerInitGuard,
+    _logger_guard: Option<LoggerInitGuard>,
     #[cfg(feature = "with_sentry")]
     _sentry_guard: Option<sentry::ClientInitGuard>,
     _metrics_guard: MetricsInitGuard,
@@ -22,18 +22,23 @@ impl SetupGuard {
         version: &Version,
         async_handle: Handle,
         silent_logging: bool,
-    ) -> Result<Self, ExitCode> {
+        enable_logging: bool,
+    ) -> Result<(Self, ckb_logger_config::Config), ExitCode> {
         // Initialization of logger must do before sentry, since `logger::init()` and
         // `sentry_config::init()` both registers custom panic hooks, but `logger::init()`
         // replaces all hooks previously registered.
-        let logger_guard = if silent_logging {
-            ckb_logger_service::init_silent()?
+        let logger_guard = if enable_logging {
+            Some(if silent_logging {
+                ckb_logger_service::init_silent()?
+            } else {
+                let mut logger_config = setup.config.logger().to_owned();
+                if logger_config.emit_sentry_breadcrumbs.is_none() {
+                    logger_config.emit_sentry_breadcrumbs = Some(setup.is_sentry_enabled);
+                }
+                ckb_logger_service::init(Some(CKB_LOG_ENV), logger_config, None)?
+            })
         } else {
-            let mut logger_config = setup.config.logger().to_owned();
-            if logger_config.emit_sentry_breadcrumbs.is_none() {
-                logger_config.emit_sentry_breadcrumbs = Some(setup.is_sentry_enabled);
-            }
-            ckb_logger_service::init(Some(CKB_LOG_ENV), logger_config)?
+            None
         };
 
         let sentry_guard = if setup.is_sentry_enabled {
@@ -67,11 +72,14 @@ impl SetupGuard {
                 ExitCode::Config
             })?;
 
-        Ok(Self {
-            _logger_guard: logger_guard,
-            _sentry_guard: sentry_guard,
-            _metrics_guard: metrics_guard,
-        })
+        Ok((
+            Self {
+                _logger_guard: logger_guard,
+                _sentry_guard: sentry_guard,
+                _metrics_guard: metrics_guard,
+            },
+            setup.config.logger().to_owned(),
+        ))
     }
 
     #[cfg(not(feature = "with_sentry"))]
@@ -80,12 +88,18 @@ impl SetupGuard {
         _version: &Version,
         async_handle: Handle,
         silent_logging: bool,
-    ) -> Result<Self, ExitCode> {
-        let logger_guard = if silent_logging {
-            ckb_logger_service::init_silent()?
+        // For ckb run, logging can be disabled here, since it requires `Shared` to create a logger that will be used for `ckb run`
+        enable_logging: bool,
+    ) -> Result<(Self, ckb_logger_config::Config), ExitCode> {
+        let logger_guard = if enable_logging {
+            Some(if silent_logging {
+                ckb_logger_service::init_silent()?
+            } else {
+                let logger_config = setup.config.logger().to_owned();
+                ckb_logger_service::init(Some(CKB_LOG_ENV), logger_config, None)?
+            })
         } else {
-            let logger_config = setup.config.logger().to_owned();
-            ckb_logger_service::init(Some(CKB_LOG_ENV), logger_config)?
+            None
         };
 
         let metrics_config = setup.config.metrics().to_owned();
@@ -95,9 +109,12 @@ impl SetupGuard {
                 ExitCode::Config
             })?;
 
-        Ok(Self {
-            _logger_guard: logger_guard,
-            _metrics_guard: metrics_guard,
-        })
+        Ok((
+            Self {
+                _logger_guard: logger_guard,
+                _metrics_guard: metrics_guard,
+            },
+            setup.config.logger().to_owned(),
+        ))
     }
 }

ckb-bin/src/subcommand/run.rs

@@ -1,6 +1,7 @@
 use std::thread::available_parallelism;
 
 use crate::helper::deadlock_detection;
+use crate::setup_guard::CKB_LOG_ENV;
 use ckb_app_config::{ExitCode, RunArgs};
 use ckb_async_runtime::{Handle, new_global_runtime};
 use ckb_build_info::Version;
@@ -13,22 +14,33 @@ use ckb_stop_handler::{broadcast_exit_signals, wait_all_ckb_services_exit};
 
 use ckb_types::core::cell::setup_system_cell_cache;
 
-pub fn run(args: RunArgs, version: Version, async_handle: Handle) -> Result<(), ExitCode> {
+pub fn run(
+    args: RunArgs,
+    version: Version,
+    async_handle: Handle,
+    log_config: ckb_logger_config::Config,
+) -> Result<(), ExitCode> {
     check_default_db_options_exists(&args)?;
     deadlock_detection();
 
     let rpc_threads_num = calc_rpc_threads_num(&args);
-    info!("ckb version: {}", version);
-    info!("run rpc server with {} threads", rpc_threads_num);
     let (mut rpc_handle, _rpc_stop_rx, _runtime) = new_global_runtime(Some(rpc_threads_num));
-    let launcher = Launcher::new(args, version, async_handle, rpc_handle.clone());
+    let launcher = Launcher::new(args, version.clone(), async_handle, rpc_handle.clone());
 
     let block_assembler_config = launcher.sanitize_block_assembler_config()?;
     let miner_enable = block_assembler_config.is_some();
 
     launcher.check_indexer_config()?;
 
     let (shared, mut pack) = launcher.build_shared(block_assembler_config)?;
+    let _logger_guard = ckb_logger_service::init(
+        Some(CKB_LOG_ENV),
+        log_config,
+        Some(shared.notify_controller().clone()),
+    )?;
+
+    info!("ckb version: {}", version);
+    info!("run rpc server with {} threads", rpc_threads_num);
 
     // spawn freezer background process
     let _freezer = shared.spawn_freeze();