Add an HTTP API endpoint + UI widgets to stream logs (#49)

* Added JSON logs to Desktop

* Move logging config into librqbit for reuse

* Log printer now available in both Desktop and Web UI

* Fix JS type error

crates/librqbit/src/api.rs

@@ -19,7 +19,8 @@ use crate::{
     torrent_state::{
         peer::stats::snapshot::{PeerStatsFilter, PeerStatsSnapshot},
         ManagedTorrentHandle,
-    }, log_subscriber::LineBroadcast,
+    },
+    tracing_subscriber_config_utils::LineBroadcast,
 };
 
 pub use crate::torrent_state::stats::{LiveStats, TorrentStats};
@@ -39,12 +40,12 @@ impl Api {
     pub fn new(
         session: Arc<Session>,
         rust_log_reload_tx: Option<UnboundedSender<String>>,
-        line_broadcast: Option<LineBroadcast>
+        line_broadcast: Option<LineBroadcast>,
     ) -> Self {
         Self {
             session,
             rust_log_reload_tx,
-            line_broadcast
+            line_broadcast,
         }
     }
 

crates/librqbit/src/http_api.rs

@@ -262,12 +262,12 @@ impl HttpApi {
         };
 
         let app = app
-        .layer(cors_layer)
+            .layer(cors_layer)
             .layer(tower_http::trace::TraceLayer::new_for_http())
             .with_state(state)
             .into_make_service();
 
-        info!("starting HTTP server on {}", addr);
+        info!(%addr, "starting HTTP server");
 
         use tokio::net::TcpListener;
         let listener = TcpListener::bind(&addr)

crates/librqbit/src/lib.rs

@@ -29,12 +29,12 @@ mod dht_utils;
 mod file_ops;
 pub mod http_api;
 pub mod http_api_client;
-pub mod log_subscriber;
 mod peer_connection;
 mod peer_info_reader;
 mod session;
 mod spawn_utils;
 mod torrent_state;
+pub mod tracing_subscriber_config_utils;
 mod tracker_comms;
 mod type_aliases;
 

crates/librqbit/src/log_subscriber.rs

@@ -1,47 +0,0 @@
-use std::io::LineWriter;
-
-use bytes::Bytes;
-use tracing_subscriber::fmt::MakeWriter;
-
-pub struct Subscriber {
-    tx: tokio::sync::broadcast::Sender<Bytes>,
-}
-
-pub struct Writer {
-    tx: tokio::sync::broadcast::Sender<Bytes>,
-}
-
-pub type LineBroadcast = tokio::sync::broadcast::Sender<Bytes>;
-
-impl Subscriber {
-    pub fn new() -> (Self, LineBroadcast) {
-        let (tx, _) = tokio::sync::broadcast::channel(100);
-        (Self { tx: tx.clone() }, tx)
-    }
-}
-
-impl<'a> MakeWriter<'a> for Subscriber {
-    type Writer = LineWriter<Writer>;
-
-    fn make_writer(&self) -> Self::Writer {
-        LineWriter::new(Writer {
-            tx: self.tx.clone(),
-        })
-    }
-}
-
-impl std::io::Write for Writer {
-    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
-        let len = buf.len();
-        if self.tx.receiver_count() == 0 {
-            return Ok(len);
-        }
-        let arc = buf.to_vec().into();
-        let _ = self.tx.send(arc);
-        Ok(len)
-    }
-
-    fn flush(&mut self) -> std::io::Result<()> {
-        Ok(())
-    }
-}

crates/librqbit/src/session.rs

@@ -725,7 +725,7 @@ impl Session {
 
         std::fs::rename(&tmp_filename, &self.persistence_filename)
             .context("error renaming persistence file")?;
-        debug!("wrote persistence to {:?}", &self.persistence_filename);
+        trace!(filename=?self.persistence_filename, "wrote persistence");
         Ok(())
     }
 
@@ -779,7 +779,7 @@ impl Session {
                     })
                     .collect();
 
-                debug!("querying DHT for {:?}", info_hash);
+                debug!(?info_hash, "querying DHT");
                 let (info, dht_rx, initial_peers) = match read_metainfo_from_peer_receiver(
                     self.peer_id,
                     info_hash,
@@ -794,7 +794,7 @@ impl Session {
                         anyhow::bail!("DHT died, no way to discover torrent metainfo")
                     }
                 };
-                debug!("received result from DHT: {:?}", info);
+                debug!(?info, "received result from DHT");
                 (
                     info_hash,
                     info,
@@ -828,7 +828,7 @@ impl Session {
 
                 let dht_rx = match self.dht.as_ref() {
                     Some(dht) if !opts.paused && !opts.list_only => {
-                        debug!("reading peers for {:?} from DHT", torrent.info_hash);
+                        debug!(info_hash=?torrent.info_hash, "reading peers from DHT");
                         Some(dht.get_peers(torrent.info_hash, announce_port)?)
                     }
                     _ => None,
@@ -911,7 +911,7 @@ impl Session {
                                 continue;
                             }
                             if !list_only {
-                                info!("Will download {:?}", filename);
+                                info!(?filename, "will download");
                             }
                         }
                         Ok(Some(only_files))
@@ -1043,14 +1043,14 @@ impl Session {
         match (paused, delete_files) {
             (Err(e), true) => Err(e).context("torrent deleted, but could not delete files"),
             (Err(e), false) => {
-                warn!("could not delete torrent files: {:?}", e);
+                warn!(error=?e, "could not delete torrent files");
                 Ok(())
             }
             (Ok(Some(paused)), true) => {
                 drop(paused.files);
                 for file in paused.filenames {
                     if let Err(e) = std::fs::remove_file(&file) {
-                        warn!("could not delete file {:?}: {:?}", file, e);
+                        warn!(?file, error=?e, "could not delete file");
                     }
                 }
                 Ok(())

crates/librqbit/src/tracing_subscriber_config_utils.rs

@@ -0,0 +1,142 @@
+use std::io::LineWriter;
+
+use anyhow::Context;
+use bytes::Bytes;
+use librqbit_core::spawn_utils::spawn;
+use tracing::error_span;
+use tracing_subscriber::fmt::MakeWriter;
+
+struct Subscriber {
+    tx: tokio::sync::broadcast::Sender<Bytes>,
+}
+
+struct Writer {
+    tx: tokio::sync::broadcast::Sender<Bytes>,
+}
+
+pub type LineBroadcast = tokio::sync::broadcast::Sender<Bytes>;
+
+impl Subscriber {
+    pub fn new() -> (Self, LineBroadcast) {
+        let (tx, _) = tokio::sync::broadcast::channel(100);
+        (Self { tx: tx.clone() }, tx)
+    }
+}
+
+impl<'a> MakeWriter<'a> for Subscriber {
+    type Writer = LineWriter<Writer>;
+
+    fn make_writer(&self) -> Self::Writer {
+        LineWriter::new(Writer {
+            tx: self.tx.clone(),
+        })
+    }
+}
+
+impl std::io::Write for Writer {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        let len = buf.len();
+        if self.tx.receiver_count() == 0 {
+            return Ok(len);
+        }
+        let arc = buf.to_vec().into();
+        let _ = self.tx.send(arc);
+        Ok(len)
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+pub struct InitLoggingOptions<'a> {
+    pub default_rust_log_value: Option<&'a str>,
+    pub log_file: Option<&'a str>,
+    pub log_file_rust_log: Option<&'a str>,
+}
+
+pub struct InitLoggingResult {
+    pub rust_log_reload_tx: tokio::sync::mpsc::UnboundedSender<String>,
+    pub line_broadcast: LineBroadcast,
+}
+
+pub fn init_logging(opts: InitLoggingOptions) -> anyhow::Result<InitLoggingResult> {
+    let stderr_filter = EnvFilter::builder()
+        .with_default_directive(
+            opts.default_rust_log_value
+                .unwrap_or("info")
+                .parse()
+                .context("can't parse provided rust_log value")?,
+        )
+        .from_env()
+        .context("invalid RUST_LOG value")?;
+
+    let (stderr_filter, reload_stderr_filter) =
+        tracing_subscriber::reload::Layer::new(stderr_filter);
+
+    use tracing_subscriber::{fmt, prelude::*, EnvFilter};
+
+    let (line_sub, line_broadcast) = Subscriber::new();
+
+    let layered = tracing_subscriber::registry()
+        // Stderr logging layer.
+        .with(fmt::layer().with_filter(stderr_filter))
+        // HTTP API log broadcast layer.
+        .with(
+            fmt::layer()
+                .with_ansi(false)
+                .fmt_fields(tracing_subscriber::fmt::format::JsonFields::new())
+                .event_format(fmt::format().with_ansi(false).json())
+                .with_writer(line_sub)
+                .with_filter(EnvFilter::builder().parse("info,librqbit=debug").unwrap()),
+        );
+    if let Some(log_file) = &opts.log_file {
+        let log_file = log_file.to_string();
+        let log_file = move || {
+            LineWriter::new(
+                std::fs::OpenOptions::new()
+                    .create(true)
+                    .append(true)
+                    .write(true)
+                    .open(&log_file)
+                    .with_context(|| format!("error opening log file {:?}", log_file))
+                    .unwrap(),
+            )
+        };
+        layered
+            .with(
+                fmt::layer()
+                    .with_ansi(false)
+                    .with_writer(log_file)
+                    .with_filter(
+                        EnvFilter::builder()
+                            .parse(opts.log_file_rust_log.unwrap_or("info,librqbit=debug"))
+                            .context("can't parse log-file-rust-log")?,
+                    ),
+            )
+            .try_init()
+            .context("can't init logging")?;
+    } else {
+        layered.try_init().context("can't init logging")?;
+    }
+
+    let (reload_tx, mut reload_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
+    spawn(error_span!("fmt_filter_reloader"), async move {
+        while let Some(rust_log) = reload_rx.recv().await {
+            let stderr_env_filter = match EnvFilter::builder().parse(&rust_log) {
+                Ok(f) => f,
+                Err(e) => {
+                    eprintln!("can't parse env filter {:?}: {:#?}", rust_log, e);
+                    continue;
+                }
+            };
+            eprintln!("setting RUST_LOG to {:?}", rust_log);
+            let _ = reload_stderr_filter.reload(stderr_env_filter);
+        }
+        Ok(())
+    });
+    Ok(InitLoggingResult {
+        rust_log_reload_tx: reload_tx,
+        line_broadcast,
+    })
+}