Introduce a non blocking file appender

Cargo.toml

@@ -11,5 +11,6 @@ members = [
  "tracing-macros",
  "tracing-subscriber",
  "tracing-serde",
+ "tracing-appender",
  "examples"
 ]

tracing-appender/Cargo.toml

@@ -0,0 +1,15 @@
+[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"

tracing-appender/src/inner.rs

@@ -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
+}

tracing-appender/src/lib.rs

@@ -0,0 +1,28 @@
+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.
+///
+/// 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)
+}

tracing-appender/src/non_blocking.rs

@@ -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),
+ }
+ }
+}