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

ckb-bin/src/lib.rs

@@ -128,12 +128,22 @@ 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) = if cmd == cli::CMD_RUN {
+        SetupGuard::from_setup(&setup, &version, handle.clone(), is_silent_logging, false)?
+    } else {
+        SetupGuard::from_setup(&setup, &version, handle.clone(), is_silent_logging, true)?
+    };
 
     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.unwrap(),
+        ),
         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()),

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, Option<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,
+            },
+            Some(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, Option<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,
+            },
+            Some(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();

notify/src/lib.rs

@@ -5,7 +5,7 @@
 //! notifications about these events asynchronously.
 use ckb_app_config::NotifyConfig;
 use ckb_async_runtime::Handle;
-use ckb_logger::{debug, error, info, trace};
+use ckb_logger::{Level, debug, error, info, trace};
 use ckb_stop_handler::{CancellationToken, new_tokio_exit_rx};
 use ckb_types::packed::Byte32;
 use ckb_types::{
@@ -23,6 +23,15 @@ use tokio::time::timeout;
 
 pub use ckb_types::core::tx_pool::PoolTransactionEntry;
 
+/// A log entry containing the message and log level.
+#[derive(Clone, Debug)]
+pub struct LogEntry {
+    /// The log message.
+    pub message: String,
+    /// The log level.
+    pub level: Level,
+}
+
 /// Asynchronous request sent to the service.
 pub struct Request<A, R> {
     /// Oneshot channel for the service to send back the response.
@@ -106,6 +115,8 @@ pub struct NotifyController {
     reject_transaction_notifier: Sender<(PoolTransactionEntry, Reject)>,
     network_alert_register: NotifyRegister<Alert>,
     network_alert_notifier: Sender<Alert>,
+    log_register: NotifyRegister<LogEntry>,
+    log_notifier: Sender<LogEntry>,
     handle: Handle,
 }
 
@@ -120,6 +131,7 @@ pub struct NotifyService {
     proposed_transaction_subscribers: HashMap<String, Sender<PoolTransactionEntry>>,
     reject_transaction_subscribers: HashMap<String, Sender<(PoolTransactionEntry, Reject)>>,
     network_alert_subscribers: HashMap<String, Sender<Alert>>,
+    log_subscribers: HashMap<String, Sender<LogEntry>>,
     timeout: NotifyTimeout,
     handle: Handle,
 }
@@ -137,14 +149,15 @@ impl NotifyService {
             proposed_transaction_subscribers: HashMap::default(),
             reject_transaction_subscribers: HashMap::default(),
             network_alert_subscribers: HashMap::default(),
+            log_subscribers: HashMap::default(),
             timeout,
             handle,
         }
     }
 
     /// start background tokio spawned task.
     pub fn start(mut self) -> NotifyController {
-        let signal_receiver: CancellationToken = new_tokio_exit_rx();
+        let stop_token: CancellationToken = new_tokio_exit_rx();
         let handle = self.handle.clone();
 
         let (new_block_register, mut new_block_register_receiver) =
@@ -172,6 +185,10 @@ impl NotifyService {
             mpsc::channel(REGISTER_CHANNEL_SIZE);
         let (network_alert_sender, mut network_alert_receiver) = mpsc::channel(NOTIFY_CHANNEL_SIZE);
 
+        let (log_register, mut log_register_receiver) = mpsc::channel(REGISTER_CHANNEL_SIZE);
+        let (log_sender, mut log_receiver) = mpsc::channel(NOTIFY_CHANNEL_SIZE);
+
+        let stop_token_clone = stop_token;
         handle.spawn(async move {
             loop {
                 tokio::select! {
@@ -186,7 +203,9 @@ impl NotifyService {
                     Some(msg) = reject_transaction_receiver.recv() => { self.handle_notify_reject_transaction(msg) },
                     Some(msg) = network_alert_register_receiver.recv() => { self.handle_register_network_alert(msg) },
                     Some(msg) = network_alert_receiver.recv() => { self.handle_notify_network_alert(msg) },
-                    _ = signal_receiver.cancelled() => {
+                    Some(msg) = log_register_receiver.recv() => { self.handle_register_log(msg) },
+                    Some(msg) = log_receiver.recv() => { self.handle_notify_log(msg) },
+                    _ = stop_token_clone.cancelled() => {
                         info!("NotifyService received exit signal, exit now");
                         break;
                     }
@@ -207,6 +226,8 @@ impl NotifyService {
             reject_transaction_notifier: reject_transaction_sender,
             network_alert_register,
             network_alert_notifier: network_alert_sender,
+            log_register,
+            log_notifier: log_sender,
             handle,
         }
     }
@@ -415,6 +436,26 @@ impl NotifyService {
             });
         }
     }
+
+    fn handle_register_log(&mut self, msg: Request<String, Receiver<LogEntry>>) {
+        let Request {
+            responder,
+            arguments: name,
+        } = msg;
+        debug!("Register log {:?}", name);
+        let (sender, receiver) = mpsc::channel(NOTIFY_CHANNEL_SIZE);
+        self.log_subscribers.insert(name, sender);
+        let _ = responder.send(receiver);
+    }
+
+    fn handle_notify_log(&self, log_entry: LogEntry) {
+        for subscriber in self.log_subscribers.values() {
+            let log_entry = log_entry.clone();
+            let subscriber = subscriber.clone();
+            // Ignore failures
+            subscriber.try_send(log_entry).ok();
+        }
+    }
 }
 
 impl NotifyController {
@@ -528,4 +569,20 @@ impl NotifyController {
             }
         });
     }
+
+    /// Subscribes to log notifications with the given name.
+    ///
+    /// Returns a receiver channel that will receive log events.
+    pub async fn subscribe_log<S: ToString>(&self, name: S) -> Receiver<LogEntry> {
+        Request::call(&self.log_register, name.to_string())
+            .await
+            .expect("Subscribe log should be OK")
+    }
+
+    /// Notifies all subscribers of a log entry.
+    pub fn notify_log(&self, log_entry: LogEntry) {
+        let log_notifier = self.log_notifier.clone();
+        // Ignore failures
+        log_notifier.try_send(log_entry).ok();
+    }
 }