Add log Rotation

Cargo.toml

@@ -19,7 +19,7 @@ num-derive = "0.3"
 slog = "2.5.2"
 slog-term = "2.5.0"
 slog-async = "2.4.0"
-
+chrono = "0.4"
 [dev-dependencies]
 criterion = "0.3.0"
 

src/logger.rs

@@ -12,17 +12,26 @@
 // limitations under the License.
 
 use crate::db::filename::{generate_filename, FileType};
+use crate::error::Result;
 use crate::storage::{File, Storage};
 
 use log::{LevelFilter, Log, Metadata, Record};
 use slog::{o, Drain, Level};
 
+use chrono::prelude::*;
 use std::sync::Mutex;
-
 /// A `slog` based logger which can be used with `log` crate
 ///
 /// See `slog` at https://github.com/slog-rs/slog
 /// See `log` at https://github.com/rust-lang/log
+///
+
+fn create_file<S: Storage>(storage: &S, dp_path: &str, timestamp: i64) -> Result<S::F> {
+    let new_path = generate_filename(dp_path, FileType::OldInfoLog, timestamp as u64);
+    storage.rename(dp_path, new_path.as_str())?;
+    storage.create(dp_path)
+}
+
 pub struct Logger {
     inner: slog::Logger,
     level: LevelFilter,
@@ -35,11 +44,11 @@ impl Logger {
     /// If `inner` is `None`
     ///     - In dev mode, use a std output
     ///     - In release mode, use a storage specific file with name `LOG`
-    pub fn new<S: Storage>(
+    pub fn new<S: Storage + Clone + 'static>(
         inner: Option<slog::Logger>,
         level: LevelFilter,
-        storage: &S,
-        db_path: &str,
+        storage: S,
+        db_path: String,
     ) -> Self {
         let inner = match inner {
             Some(l) => l,
@@ -52,12 +61,14 @@ impl Logger {
                 } else {
                     // Use a file `Log` to record all logs
                     // TODO: add file rotation
-                    let file = storage
-                        .create(generate_filename(db_path, FileType::InfoLog, 0).as_str())
-                        .unwrap();
-                    let drain = slog_async::Async::new(FileBasedDrain::new(file))
-                        .build()
-                        .fuse();
+                    let file =
+                        create_file(&storage, db_path.as_str(), Local::now().timestamp()).unwrap();
+                    let file_fn = move |path: String| {
+                        create_file(&storage, path.as_str(), Local::now().timestamp())
+                    };
+                    let drain = FileBasedDrain::new(file, db_path.clone(), file_fn)
+                        .add_rotator(RotatedFileBySize::new(0));
+                    let drain = slog_async::Async::new(drain).build().fuse();
                     slog::Logger::root(drain, o!())
                 }
             }
@@ -104,7 +115,6 @@ impl Log for Logger {
             }
         }
     }
-
     fn flush(&self) {}
 }
 
@@ -120,13 +130,44 @@ fn log_to_slog_level(level: log::Level) -> Level {
 
 struct FileBasedDrain<F: File> {
     inner: Mutex<F>,
+    rotators: Vec<Box<dyn Rotator>>,
+    dp_path: String,
+    new_file: Box<dyn Send + Fn(String) -> Result<F>>,
 }
 
 impl<F: File> FileBasedDrain<F> {
-    fn new(f: F) -> Self {
+    fn new<H>(f: F, path: String, new_file: H) -> Self
+    where
+        H: 'static + Send + Fn(String) -> Result<F>,
+    {
         FileBasedDrain {
+            dp_path: path.clone(),
             inner: Mutex::new(f),
+            rotators: vec![],
+            new_file: Box::new(new_file),
+        }
+    }
+
+    fn add_rotator<R: 'static + Rotator>(mut self, rotator: R) -> Self {
+        if rotator.is_enabled() {
+            self.rotators.push(Box::new(rotator));
+        }
+        for rotator in (&self).rotators.iter() {
+            rotator.prepare(&*self.inner.lock().unwrap()).unwrap();
+        }
+        self
+    }
+
+    fn flush(&self) -> Result<()> {
+        self.inner.lock().unwrap().flush()?;
+        let new_file = (self.new_file)(self.dp_path.clone()).unwrap();
+
+        let mut old_file = self.inner.lock().unwrap();
+        std::mem::replace(&mut *old_file, new_file);
+        for rotator in self.rotators.iter() {
+            rotator.on_rotate()?;
         }
+        return Ok(());
     }
 }
 
@@ -138,17 +179,79 @@ impl<F: File> Drain for FileBasedDrain<F> {
         &self,
         record: &slog::Record,
         values: &slog::OwnedKVList,
-    ) -> Result<Self::Ok, Self::Err> {
-        // Ignore errors here
-        let _ = self.inner.lock().unwrap().write(
-            format!(
-                "[{}] : {:?} \n {:?} \n",
-                record.level(),
-                record.msg(),
-                values
-            )
-            .as_bytes(),
+    ) -> std::result::Result<Self::Ok, Self::Err> {
+        let by = format!(
+            "[{}] : {:?} \n {:?} \n",
+            record.level(),
+            record.msg(),
+            values
         );
+        for rotator in self.rotators.iter() {
+            if rotator.should_rotate() {
+                self.flush().unwrap();
+                return Ok(());
+            }
+        }
+
+        for rotator in self.rotators.iter() {
+            rotator.on_write(by.as_bytes()).unwrap();
+        }
+        // Ignore errors here
+        let _ = self.inner.lock().unwrap().write(by.as_bytes());
+
+        Ok(())
+    }
+}
+
+trait Rotator: Send {
+    /// Check if the option is enabled in configuration.
+    /// Return true if the `rotator` is valid.
+    fn is_enabled(&self) -> bool;
+
+    /// Call by operator, initializes the states of rotators.
+    fn prepare(&self, file: &dyn File) -> Result<()>;
+
+    /// Return if the file need to be rotated.
+    fn should_rotate(&self) -> bool;
+
+    fn on_write(&self, buf: &[u8]) -> Result<()>;
+    // Call by operator, update rotators' state while the operator execute a rotation.
+    fn on_rotate(&self) -> Result<()>;
+}
+
+struct RotatedFileBySize {
+    rotation_size: u64,
+    file_size: Mutex<u64>,
+}
+
+impl RotatedFileBySize {
+    fn new(rotation_size: u64) -> Self {
+        RotatedFileBySize {
+            rotation_size,
+            file_size: Mutex::new(0),
+        }
+    }
+}
+
+impl Rotator for RotatedFileBySize {
+    fn is_enabled(&self) -> bool {
+        self.rotation_size != 0
+    }
+    fn prepare(&self, file: &dyn File) -> Result<()> {
+        *self.file_size.lock().unwrap() = file.len().unwrap();
+        Ok(())
+    }
+
+    fn should_rotate(&self) -> bool {
+        *self.file_size.lock().unwrap() > self.rotation_size
+    }
+    fn on_write(&self, buf: &[u8]) -> Result<()> {
+        *self.file_size.lock().unwrap() += buf.len() as u64;
+        Ok(())
+    }
+
+    fn on_rotate(&self) -> Result<()> {
+        *self.file_size.lock().unwrap() = 0;
         Ok(())
     }
 }
@@ -165,8 +268,9 @@ mod tests {
     #[test]
     fn test_default_logger() {
         let s = MemStorage::default();
+        // let s = &'static s;
         let db_path = "test";
-        let logger = Logger::new(None, LevelFilter::Debug, &s, db_path);
+        let logger = Logger::new(None, LevelFilter::Debug, s, db_path.to_string());
         // Ignore the error if the logger have been set
         let _ = log::set_logger(Box::leak(Box::new(logger)));
         log::set_max_level(LevelFilter::Debug);

src/options.rs

@@ -205,7 +205,7 @@ impl Options {
     }
 
     /// Initialize Options by limiting ranges of some flags, applying customized Logger and etc.
-    pub(crate) fn initialize<O: File + 'static, S: Storage<F = O>>(
+    pub(crate) fn initialize<O: File + 'static, S: Storage<F = O> + Clone + 'static>(
         &mut self,
         db_name: String,
         storage: &S,
@@ -225,9 +225,14 @@ impl Options {
     }
 
     #[allow(unused_must_use)]
-    fn apply_logger<S: Storage>(&mut self, storage: &S, db_path: &str) {
+    fn apply_logger<S: 'static + Storage + Clone>(&mut self, storage: &S, db_path: &str) {
         let user_logger = std::mem::replace(&mut self.logger, None);
-        let logger = Logger::new(user_logger, self.logger_level, storage, db_path);
+        let logger = Logger::new(
+            user_logger,
+            self.logger_level,
+            storage.clone(),
+            db_path.to_string(),
+        );
         let static_logger: &'static dyn Log = Box::leak(Box::new(logger));
         log::set_logger(static_logger);
         log::set_max_level(self.logger_level);

src/version/version_edit.rs

@@ -480,6 +480,7 @@ mod tests {
         let mut edit = VersionEdit::new(7);
         let filename = String::from("Hello");
         edit.set_comparator_name(filename);
+
         assert_eq!("Hello", edit.comparator_name.unwrap().as_str());
     }
 
@@ -488,6 +489,7 @@ mod tests {
         let mut edit = VersionEdit::new(7);
         let log_num = u64::max_value();
         edit.set_log_number(log_num);
+
         assert_eq!(edit.log_number.unwrap(), log_num);
     }