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/file_appender.rs

@@ -0,0 +1,125 @@
+use crate::inner::BufWriterFactory;
+use crate::worker::Worker;
+use crate::Rotation;
+use chrono::Utc;
+use crossbeam_channel::{bounded, Sender};
+use std::path::Path;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use std::{io, thread};
+use tracing_subscriber::fmt::MakeWriter;
+
+#[allow(dead_code)]
+pub struct FileAppender {
+ log_writer: FileWriter,
+ worker_thread: thread::JoinHandle<()>,
+ error_counter: Arc<AtomicU64>,
+}
+
+impl FileAppender {
+ pub fn new(log_directory: &str, log_filename_prefix: &str) -> Self {
+ FileAppenderBuilder::default()
+ .build(log_directory, log_filename_prefix)
+ .expect("Failed to create FileAppender")
+ }
+
+ fn create_appender(
+ sender: Sender<Vec<u8>>,
+ worker: Worker<BufWriterFactory>,
+ error_counter: Arc<AtomicU64>,
+ ) -> Self {
+ Self {
+ log_writer: FileWriter {
+ channel: sender,
+ error_counter: error_counter.clone(),
+ },
+ worker_thread: worker.worker_thread(),
+ error_counter,
+ }
+ }
+
+ pub fn builder() -> FileAppenderBuilder {
+ FileAppenderBuilder::default()
+ }
+
+ pub fn get_writer(self) -> FileWriter {
+ self.log_writer
+ }
+
+ pub fn get_error_counter(self) -> Arc<AtomicU64> {
+ self.error_counter.clone()
+ }
+}
+
+pub struct FileAppenderBuilder {
+ buffered_lines_limit: usize,
+ rotation: Rotation,
+}
+
+impl FileAppenderBuilder {
+ pub fn buffered_lines_limit(mut self, buffered_lines_limit: usize) -> FileAppenderBuilder {
+ self.buffered_lines_limit = buffered_lines_limit;
+ self
+ }
+
+ pub fn build(
+ self,
+ log_directory: impl AsRef<Path>,
+ log_filename_prefix: impl AsRef<Path>,
+ ) -> io::Result<FileAppender> {
+ let (sender, receiver) = bounded(self.buffered_lines_limit);
+
+ let worker = Worker::new(
+ receiver,
+ log_directory.as_ref(),
+ log_filename_prefix.as_ref(),
+ self.rotation,
+ BufWriterFactory {},
+ Utc::now(),
+ );
+ Ok(FileAppender::create_appender(
+ sender,
+ worker?,
+ Arc::new(AtomicU64::new(0)),
+ ))
+ }
+}
+
+impl Default for FileAppenderBuilder {
+ fn default() -> Self {
+ FileAppenderBuilder {
+ buffered_lines_limit: 100_000,
+ rotation: Rotation::HOURLY,
+ }
+ }
+}
+
+pub struct FileWriter {
+ channel: Sender<Vec<u8>>,
+ error_counter: Arc<AtomicU64>,
+}
+
+impl std::io::Write for FileWriter {
+ fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+ let buf_size = buf.len();
+ if self.channel.try_send(buf.to_vec()).is_err() {
+ self.error_counter.fetch_add(1, Ordering::Relaxed);
+ }
+ Ok(buf_size)
+ }
+
+ fn flush(&mut self) -> io::Result<()> {
+ Ok(())
+ }
+}
+
+impl MakeWriter for FileWriter {
+ type Writer = FileWriter;
+
+ fn make_writer(&self) -> Self::Writer {
+ FileWriter {
+ channel: self.channel.clone(),
+ error_counter: self.error_counter.clone(),
+ }
+ }
+}

tracing-appender/src/inner.rs

@@ -0,0 +1,130 @@
+use std::io::BufWriter;
+use std::io::Write;
+use std::{fs, io};
+
+use crate::Rotation;
+use chrono::prelude::*;
+use std::fmt::Debug;
+use std::fs::File;
+use std::fs::OpenOptions;
+use std::path::Path;
+
+pub(crate) trait WriterFactory: Debug + Send {
+ type W: Write + Debug + Send;
+
+ fn create_writer(&self, directory: &str, filename: &str) -> io::Result<Self::W>;
+}
+
+#[derive(Debug)]
+pub(crate) struct BufWriterFactory {}
+
+impl WriterFactory for BufWriterFactory {
+ type W = BufWriter<File>;
+
+ fn create_writer(&self, directory: &str, filename: &str) -> io::Result<BufWriter<File>> {
+ let filepath = Path::new(directory).join(filename);
+ Ok(BufWriter::new(open_file_create_parent_dirs(&filepath)?))
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct InnerAppender<F: WriterFactory + Send> {
+ log_directory: String,
+ log_filename_prefix: String,
+ writer: F::W,
+ writer_factory: F,
+ next_date: DateTime<Utc>,
+ rotation: Rotation,
+}
+
+impl<F: WriterFactory> InnerAppender<F> {
+ fn write_with_ts(&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 _ = self.refresh_writer(date);
+ self.writer.write(buf)
+ }
+}
+
+impl<F: WriterFactory> io::Write for InnerAppender<F> {
+ fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+ let now = Utc::now();
+ self.write_with_ts(buf, now)
+ }
+
+ fn flush(&mut self) -> io::Result<()> {
+ self.writer.flush()
+ }
+}
+
+impl<F: WriterFactory> InnerAppender<F> {
+ pub(crate) fn new(
+ log_directory: &Path,
+ log_filename_prefix: &Path,
+ rotation: Rotation,
+ writer_factory: F,
+ 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);
+
+ let mut appender = InnerAppender {
+ log_directory: log_directory.to_string(),
+ log_filename_prefix: log_filename_prefix.to_string(),
+ writer: writer_factory.create_writer(log_directory, &filename)?,
+ writer_factory,
+ next_date,
+ rotation,
+ };
+
+ appender
+ .write_with_ts(b"Init Application\n", now)
+ .and_then(|_| appender.flush().and(Ok(appender)))
+ }
+}
+
+impl<F: WriterFactory> InnerAppender<F> {
+ pub(crate) fn refresh_writer(&mut self, now: DateTime<Utc>) -> io::Result<()> {
+ 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 self
+ .writer_factory
+ .create_writer(&self.log_directory, &filename)
+ {
+ Ok(writer) => {
+ self.writer = writer;
+ Ok(())
+ }
+ Err(err) => {
+ eprintln!("Couldn't create writer for logs: {}", err);
+ Err(err)
+ }
+ }
+ } else {
+ Ok(())
+ }
+ }
+
+ pub(crate) fn should_rollover(&self, date: DateTime<Utc>) -> bool {
+ date >= self.next_date
+ }
+}
+
+// Open a file - if it throws any error, try creating the parent directory and then the file.
+fn open_file_create_parent_dirs(path: &Path) -> io::Result<File> {
+ let new_file = OpenOptions::new().append(true).create(true).open(path);
+ if new_file.is_err() {
+ if let Some(parent) = path.parent() {
+ fs::create_dir_all(parent)?;
+ return OpenOptions::new().append(true).create(true).open(path);
+ }
+ }
+
+ new_file
+}

tracing-appender/src/lib.rs

@@ -0,0 +1,55 @@
+use chrono::{DateTime, Datelike, TimeZone, Timelike, Utc};
+
+pub mod file_appender;
+
+mod inner;
+mod worker;
+
+#[derive(Clone, Eq, PartialEq, Debug)]
+pub struct Rotation(RotationKind);
+
+#[derive(Clone, Eq, PartialEq, Debug)]
+enum RotationKind {
+ Hourly,
+ Daily,
+ Never,
+}
+
+impl Rotation {
+ pub const HOURLY: Self = Self(RotationKind::Hourly);
+ pub const DAILY: Self = Self(RotationKind::Daily);
+ pub const NEVER: Self = Self(RotationKind::Never);
+
+ fn next_date(&self, current_date: &DateTime<Utc>) -> DateTime<Utc> {
+ let unrounded_next_date = match *self {
+ Rotation::HOURLY => *current_date + chrono::Duration::hours(1),
+ Rotation::DAILY => *current_date + chrono::Duration::days(1),
+ Rotation::NEVER => Utc.ymd(9999, 1, 1).and_hms(1, 0, 0),
+ };
+ self.round_date(&unrounded_next_date)
+ }
+
+ fn round_date(&self, date: &DateTime<Utc>) -> DateTime<Utc> {
+ match *self {
+ Rotation::HOURLY => {
+ Utc.ymd(date.year(), date.month(), date.day())
+ .and_hms(date.hour(), 0, 0)
+ }
+ Rotation::DAILY => Utc
+ .ymd(date.year(), date.month(), date.day())
+ .and_hms(0, 0, 0),
+ Rotation::NEVER => {
+ Utc.ymd(date.year(), date.month(), date.day())
+ .and_hms(date.hour(), 0, 0)
+ }
+ }
+ }
+
+ fn join_date(&self, filename: &str, date: &DateTime<Utc>) -> String {
+ match *self {
+ Rotation::HOURLY => format!("{}.{}", filename, date.format("%F-%H")),
+ Rotation::DAILY => format!("{}.{}", filename, date.format("%F")),
+ Rotation::NEVER => filename.to_string(),
+ }
+ }
+}