First stab at log rotation

jormungandr-lib/src/interfaces/config/mempool.rs

@@ -18,8 +18,10 @@ pub struct Mempool {
     #[serde(default)]
     pub log_max_entries: LogMaxEntries,
     /// path to the persistent log of all incoming fragments
+    // FIXME: should be a struct like `persistent_log.dir`,
+    // as we may want to add more options like rotation policy later
     #[serde(default)]
-    pub persistent_log_path: Option<PathBuf>,
+    pub persistent_log_dir: Option<PathBuf>,
 }
 
 impl Default for PoolMaxEntries {
@@ -39,7 +41,7 @@ impl Default for Mempool {
         Mempool {
             pool_max_entries: PoolMaxEntries::default(),
             log_max_entries: LogMaxEntries::default(),
-            persistent_log_path: None,
+            persistent_log_dir: None,
         }
     }
 }

jormungandr/Cargo.toml

@@ -12,14 +12,6 @@ Midgard Serpent
 edition = "2018"
 
 [dependencies]
-arc-swap = "^1.1.0"
-async-trait = "=0.1.42" # https://github.com/dtolnay/async-trait/issues/144
-async-graphql = "2.5.1"
-async-graphql-warp = "2.6.0"
-base64 = "0.13.0"
-bincode = "1.3.3"
-bytes = "1.0"
-bech32 = "0.7"
 chain-addr = { git = "https://github.com/input-output-hk/chain-libs.git", branch = "master" }
 chain-core = { git = "https://github.com/input-output-hk/chain-libs.git", branch = "master" }
 chain-crypto = { git = "https://github.com/input-output-hk/chain-libs.git", branch = "master" }
@@ -30,11 +22,20 @@ chain-time      = { git = "https://github.com/input-output-hk/chain-libs.git", b
 chain-vote = { git = "https://github.com/input-output-hk/chain-libs.git", branch = "master" }
 cardano-legacy-address = { git = "https://github.com/input-output-hk/chain-libs.git", branch = "master" }
 imhamt = { git = "https://github.com/input-output-hk/chain-libs.git", branch = "master" }
+
+arc-swap = "^1.1.0"
+async-trait = "=0.1.42" # https://github.com/dtolnay/async-trait/issues/144
+async-graphql = "2.5.1"
+async-graphql-warp = "2.6.0"
+base64 = "0.13.0"
+bincode = "1.3.3"
+bytes = "1.0"
+bech32 = "0.7"
+chrono = "0.4"
 error-chain = "0.12"
 futures = "0.3.14"
 hex = "0.4"
 http = "0.2.2"
-humantime = "2.0"
 jormungandr-lib = { path = "../jormungandr-lib" }
 keynesis = "1.1"
 lazy_static = "1.4"

jormungandr/src/fragment/pool.rs

@@ -55,6 +55,10 @@ impl Pools {
         &mut self.logs
     }
 
+    pub fn set_persistent_log(&mut self, file: File) {
+        self.persistent_log = Some(file);
+    }
+
     /// Returns number of registered fragments
     pub async fn insert_and_propagate_all(
         &mut self,

jormungandr/src/fragment/process.rs

@@ -9,8 +9,11 @@ use crate::{
 };
 
 use std::collections::HashMap;
-use std::fs::File;
+use std::fs;
+use std::io;
+use std::path::PathBuf;
 
+use chrono::Utc;
 use thiserror::Error;
 use tokio_stream::StreamExt;
 use tracing::{span, Level};
@@ -26,6 +29,8 @@ pub struct Process {
 pub enum Error {
     #[error("transaction pool error")]
     Pool(#[from] crate::fragment::pool::Error),
+    #[error("failed to open persistent log file")]
+    PersistentLog(#[source] io::Error),
 }
 
 impl Process {
@@ -42,14 +47,30 @@ impl Process {
         }
     }
 
-    pub async fn start(
+    pub async fn start<P: Into<PathBuf>>(
         self,
         n_pools: usize,
         service_info: TokioServiceInfo,
         stats_counter: StatsCounter,
         mut input: MessageQueue<TransactionMsg>,
-        persistent_log: Option<File>,
+        persistent_log_dir: Option<P>,
     ) -> Result<(), Error> {
+        let persistent_log = match persistent_log_dir {
+            None => None,
+            Some(dir) => {
+                let mut path = dir.into();
+                let log_file_name = Utc::now().format("%Y-%m-%d_%H.log").to_string();
+                path.push(log_file_name);
+                let file = fs::OpenOptions::new()
+                    .append(true)
+                    .create(true)
+                    .read(false)
+                    .open(path)
+                    .map_err(Error::PersistentLog)?;
+                Some(file)
+            }
+        };
+
         let mut pool = Pools::new(
             self.pool_max_entries,
             n_pools,

jormungandr/src/main.rs

@@ -71,7 +71,6 @@ pub struct BootstrappedNode {
     rest_context: Option<rest::ContextLock>,
     services: Services,
     initial_peers: Vec<topology::Peer>,
-    persistent_fragment_log: Option<std::fs::File>,
     _logger_guards: Vec<WorkerGuard>,
 }
 
@@ -277,15 +276,15 @@ fn start_services(bootstrapped_node: BootstrappedNode) -> Result<(), start_up::E
             bootstrapped_node.settings.mempool.log_max_entries.into(),
             network_msgbox.clone(),
         );
-        let persistent_fragment_log = bootstrapped_node.persistent_fragment_log;
+        let fragment_log_dir = bootstrapped_node.settings.mempool.persistent_log_dir;
 
         services.spawn_try_future("fragment", move |info| {
             process.start(
                 n_pools,
                 info,
                 stats_counter,
                 fragment_queue,
-                persistent_fragment_log,
+                fragment_log_dir,
             )
         });
     };
@@ -355,7 +354,6 @@ fn bootstrap(initialized_node: InitializedNode) -> Result<BootstrappedNode, star
         rest_context,
         mut services,
         cancellation_token,
-        persistent_fragment_log,
         _logger_guards,
     } = initialized_node;
 
@@ -387,7 +385,6 @@ fn bootstrap(initialized_node: InitializedNode) -> Result<BootstrappedNode, star
         rest_context,
         services,
         initial_peers,
-        persistent_fragment_log,
         _logger_guards,
     })
 }
@@ -513,7 +510,6 @@ pub struct InitializedNode {
     pub rest_context: Option<rest::ContextLock>,
     pub services: Services,
     pub cancellation_token: CancellationToken,
-    pub persistent_fragment_log: Option<std::fs::File>,
     pub _logger_guards: Vec<WorkerGuard>,
 }
 
@@ -603,20 +599,6 @@ fn initialize_node() -> Result<InitializedNode, start_up::Error> {
 
     let settings = raw_settings.try_into_settings()?;
 
-    let persistent_fragment_log = settings
-        .mempool
-        .persistent_log_path
-        .as_ref()
-        .map(|path| {
-            std::fs::OpenOptions::new()
-                .append(true)
-                .create(true)
-                .read(false)
-                .open(path)
-        })
-        .transpose()
-        .map_err(start_up::Error::PersistentFragmentLog)?;
-
     let storage = start_up::prepare_storage(&settings)?;
     if exit_after_storage_setup {
         tracing::info!("Exiting after successful storage setup");
@@ -701,7 +683,6 @@ fn initialize_node() -> Result<InitializedNode, start_up::Error> {
         rest_context,
         services,
         cancellation_token,
-        persistent_fragment_log,
         _logger_guards,
     })
 }

testing/jormungandr-integration-tests/src/jormungandr/fragments.rs

@@ -31,7 +31,7 @@ pub fn send_all_fragments() {
             .with_mempool(Mempool {
                 pool_max_entries: 1_000_000usize.into(),
                 log_max_entries: 1_000_000usize.into(),
-                persistent_log_path: None,
+                persistent_log_dir: None,
             }),
     )
     .unwrap();

testing/jormungandr-integration-tests/src/jormungandr/transactions.rs

@@ -23,7 +23,7 @@ pub fn accounts_funds_are_updated_after_transaction() {
             .with_mempool(Mempool {
                 pool_max_entries: 1_000_000usize.into(),
                 log_max_entries: 1_000_000usize.into(),
-                persistent_log_path: None,
+                persistent_log_dir: None,
             }),
     )
     .unwrap();