tokio-rs · hawkw · Apr 14, 2020 · Apr 7, 2020 · Apr 8, 2020 · Apr 8, 2020
@@ -11,5 +11,6 @@ members = [
     "tracing-macros",
     "tracing-subscriber",
     "tracing-serde",
+    "tracing-appender",
     "examples"
 ]
@@ -0,0 +1,18 @@
+[package]
+name = "tracing-appender"
+version = "0.1.0"
+authors = ["Zeki Sherif <zekshi@amazon.com>"]
+edition = "2018"
+repository = "https://github.com/tokio-rs/tracing"
+homepage = "https://tokio.rs"
+description = """
+Provides an `appender` that writes to a file
+"""
+
+[dependencies]
+tracing-subscriber = {path = "../tracing-subscriber", version = "0.2.4"}
+crossbeam-channel = "0.4.2"
+chrono = "0.4.11"
+
+[dev-dependencies]
+tracing = { path = "../tracing", version = "0.1" }
@@ -0,0 +1,96 @@
+use std::io::{BufWriter, Write};
+use std::{fs, io};
+
+use crate::rolling::Rotation;
+use chrono::prelude::*;
+use std::fmt::Debug;
+use std::fs::{File, OpenOptions};
+use std::path::Path;
+
+#[derive(Debug)]
+pub(crate) struct InnerAppender {
+    log_directory: String,
+    log_filename_prefix: String,
+    writer: BufWriter<File>,
+    next_date: DateTime<Utc>,
+    rotation: Rotation,
+}
+
+impl io::Write for InnerAppender {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        let now = Utc::now();
+        self.write_timestamped(buf, now)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.writer.flush()
+    }
+}
+
+impl InnerAppender {
+    pub(crate) fn new(
+        log_directory: &Path,
+        log_filename_prefix: &Path,
+        rotation: Rotation,
+        now: DateTime<Utc>,
+    ) -> io::Result<Self> {
+        let log_directory = log_directory.to_str().unwrap();
+        let log_filename_prefix = log_filename_prefix.to_str().unwrap();
+
+        let filename = rotation.join_date(log_filename_prefix, &now);
+        let next_date = rotation.next_date(&now);
+
+        Ok(InnerAppender {
+            log_directory: log_directory.to_string(),
+            log_filename_prefix: log_filename_prefix.to_string(),
+            writer: create_writer(log_directory, &filename)?,
+            next_date,
+            rotation,
+        })
+    }
+
+    fn write_timestamped(&mut self, buf: &[u8], date: DateTime<Utc>) -> io::Result<usize> {
+        // Even if refresh_writer fails, we still have the original writer. Ignore errors
+        // and proceed with the write.
+        let buf_len = buf.len();
+        self.refresh_writer(date);
+        self.writer.write_all(buf).map(|_| buf_len)
+    }
+
+    fn refresh_writer(&mut self, now: DateTime<Utc>) {
+        if self.should_rollover(now) {
+            let filename = self.rotation.join_date(&self.log_filename_prefix, &now);
+
+            self.next_date = self.rotation.next_date(&now);
+
+            match create_writer(&self.log_directory, &filename) {
+                Ok(writer) => self.writer = writer,
+                Err(err) => eprintln!("Couldn't create writer for logs: {}", err),
+            }
+        }
+    }
+
+    fn should_rollover(&self, date: DateTime<Utc>) -> bool {
+        date >= self.next_date
+    }
+}
+
+fn create_writer(directory: &str, filename: &str) -> io::Result<BufWriter<File>> {
+    let file_path = Path::new(directory).join(filename);
+    Ok(BufWriter::new(open_file_create_parent_dirs(&file_path)?))
+}
+
+fn open_file_create_parent_dirs(path: &Path) -> io::Result<File> {
+    let mut open_options = OpenOptions::new();
+    open_options.append(true).create(true);
+
+    let new_file = open_options.open(path);
+    if new_file.is_err() {
+        if let Some(parent) = path.parent() {
+            fs::create_dir_all(parent)?;
+            return open_options.open(path);
+        }
+    }
+
+    new_file
+}
@@ -0,0 +1,49 @@
+use crate::non_blocking::{NonBlocking, WorkerGuard};
+
+use std::io::Write;
+
+mod inner;
+pub mod non_blocking;
+pub mod rolling;
+mod worker;
+
+/// Creates a non-blocking, off-thread writer.
+///
+/// This spawns a dedicated worker thread which is responsible for writing log
+/// lines to the provided writer. When a line is written using the returned
+/// `NonBlocking` struct's `make_writer` method, it will be enqueued to be
+/// written by the worker thread.
+///
+/// The queue has a fixed capacity, and if it becomes full, any logs written
+/// to it will be dropped until capacity is once again available. This may
+/// occur if logs are consistently produced faster than the worker thread can
+/// output them. The queue capacity and behavior when full (i.e., whether to
+/// drop logs or to exert backpressure to slow down senders) can be configured
+/// using [`NonBlockingBuilder::default()`]. This function simply returns the default
+/// configuration &mdash; it is equivalent to
+///
+/// ```rust
+/// # use tracing_appender::non_blocking::{NonBlocking, WorkerGuard};
+/// # fn doc() -> (NonBlocking, WorkerGuard) {
+/// tracing_appender::non_blocking::NonBlocking::new(std::io::stdout())
+/// # }
+/// ```
+/// [`NonBlocking::builder()`]: /non_blocking/struct.NonBlocking.html#method.builder
+///
+/// Note that the `WorkerGuard` returned by `non_blocking` _must_ be assigned to a binding that
+/// is not `_`, as `_` will result in the `WorkerGuard` being dropped immediately.
+///
+/// # Examples
+/// ``` rust
+///
+/// # fn docs() {
+/// let (non_blocking, _guard) = tracing_appender::non_blocking(std::io::stdout());
+/// let subscriber = tracing_subscriber::fmt().with_writer(non_blocking);
+/// tracing::subscriber::with_default(subscriber.finish(), || {
+///    tracing::event!(tracing::Level::INFO, "Hello");
+/// });
+/// # }
+/// ```
+pub fn non_blocking<T: Write + Send + Sync + 'static>(writer: T) -> (NonBlocking, WorkerGuard) {
+    NonBlocking::new(writer)
+}
@@ -0,0 +1,146 @@
+use crate::worker::Worker;
+use crossbeam_channel::{bounded, Sender};
+use std::io;
+use std::io::Write;
+use std::sync::atomic::Ordering;
+use std::sync::atomic::{AtomicBool, AtomicU64};
+use std::sync::Arc;
+use std::thread::JoinHandle;
+use tracing_subscriber::fmt::MakeWriter;
+
+pub const DEFAULT_BUFFERED_LINES_LIMIT: usize = 128_000;
+
+#[derive(Debug)]
+pub struct WorkerGuard {
+    guard: Option<JoinHandle<()>>,
+    shutdown_signal: Arc<AtomicBool>,
+}
+
+#[derive(Clone, Debug)]
+pub struct NonBlocking {
+    error_counter: Arc<AtomicU64>,
+    channel: Sender<Vec<u8>>,
+    is_lossy: bool,
+}
+
+impl NonBlocking {
+    pub fn new<T: Write + Send + Sync + 'static>(writer: T) -> (NonBlocking, WorkerGuard) {
+        NonBlockingBuilder::default().finish(writer)
+    }
+
+    fn create<T: Write + Send + Sync + 'static>(
+        writer: T,
+        buffered_lines_limit: usize,
+        is_lossy: bool,
+    ) -> (NonBlocking, WorkerGuard) {
+        let (sender, receiver) = bounded(buffered_lines_limit);
+        let shutdown_signal = Arc::new(AtomicBool::new(false));
+
+        let worker = Worker::new(receiver, writer, shutdown_signal.clone());
+        let worker_guard = WorkerGuard::new(worker.worker_thread(), shutdown_signal);
+
+        (
+            Self {
+                channel: sender,
+                error_counter: Arc::new(AtomicU64::new(0)),
+                is_lossy,
+            },
+            worker_guard,
+        )
+    }
+
+    pub fn error_counter(&self) -> Arc<AtomicU64> {
+        self.error_counter.clone()
+    }
+}
+
+#[derive(Debug)]
+pub struct NonBlockingBuilder {
+    buffered_lines_limit: usize,
+    is_lossy: bool,
+}
+
+impl NonBlockingBuilder {
+    pub fn buffered_lines_limit(mut self, buffered_lines_limit: usize) -> NonBlockingBuilder {
+        self.buffered_lines_limit = buffered_lines_limit;
+        self
+    }
+
+    pub fn lossy(mut self, is_lossy: bool) -> NonBlockingBuilder {
+        self.is_lossy = is_lossy;
+        self
+    }
+
+    pub fn finish<T: Write + Send + Sync + 'static>(self, writer: T) -> (NonBlocking, WorkerGuard) {
+        NonBlocking::create(writer, self.buffered_lines_limit, self.is_lossy)
+    }
+}
+
+impl Default for NonBlockingBuilder {
+    fn default() -> Self {
+        NonBlockingBuilder {
+            buffered_lines_limit: DEFAULT_BUFFERED_LINES_LIMIT,
+            is_lossy: true,
+        }
+    }
+}
+
+impl std::io::Write for NonBlocking {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        let buf_size = buf.len();
+        if self.is_lossy {
+            if self.channel.try_send(buf.to_vec()).is_err() {
+                self.error_counter.fetch_add(1, Ordering::Relaxed);
+            }
+        } else {
+            return match self.channel.send(buf.to_vec()) {
+                Ok(_) => Ok(buf_size),
+                Err(_) => Err(io::Error::from(io::ErrorKind::Other)),
+            };
+        }
+        Ok(buf_size)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+
+    #[inline]
+    fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
+        self.write(buf).map(|_| ())
+    }
+}
+
+impl MakeWriter for NonBlocking {
+    type Writer = NonBlocking;
+
+    fn make_writer(&self) -> Self::Writer {
+        self.clone()
+    }
+}
+
+impl WorkerGuard {
+    fn new(handle: JoinHandle<()>, shutdown_signal: Arc<AtomicBool>) -> Self {
+        WorkerGuard {
+            guard: Some(handle),
+            shutdown_signal,
+        }
+    }
+
+    fn stop(&mut self) -> std::thread::Result<()> {
+        match self.guard.take() {
+            Some(handle) => handle.join(),
+            None => Ok(()),
+        }
+    }
+}
+
+impl Drop for WorkerGuard {
+    fn drop(&mut self) {
+        self.shutdown_signal.store(true, Ordering::Relaxed);
+        match self.stop() {
+            Ok(_) => (),
+            Err(e) => println!("Failed to join worker thread. Error: {:?}", e),
+        }
+    }
+}