fix(file source): Allow file pattern with curly braces in include directive

benches/files.rs

@@ -37,8 +37,6 @@ fn benchmark_files_without_partitions(c: &mut Criterion) {
                 let mut input = directory_str.to_owned();
                 input.push_str("/test.in");
 
-                let input = PathBuf::from(input);
-
                 let mut output = directory_str.to_owned();
                 output.push_str("/test.out");
 

lib/file-source/Cargo.toml

@@ -10,6 +10,7 @@ bytes = "0.5"
 crc = "1.8.1"
 futures = { version = "0.3", default-features = false, features = ["executor"] }
 glob = "0.3.0"
+globwalk = "0.8.1"
 scan_fmt = "0.2.5"
 tracing = "0.1.15"
 indexmap = {version = "1.5.1", features = ["serde-1"]}
@@ -21,8 +22,12 @@ serde = { version = "1.0.117", features = ["derive"] }
 serde_json = "1.0.33"
 chrono = { version = "0.4.19", features = ["serde"] }
 dashmap = "4.0.0"
+snafu = "0.6.10"
 
 [dev-dependencies]
 quickcheck = "0.9"
 rand = "0.7"
 tempfile = "3.1.0"
+
+[[test]]
+name = "glob"

lib/file-source/src/file_server.rs

@@ -12,6 +12,7 @@ use futures::{
     stream, Future, Sink, SinkExt,
 };
 use indexmap::IndexMap;
+use snafu::{ResultExt, Snafu};
 use std::{
     collections::{BTreeMap, HashSet},
     fs::{self, remove_file},
@@ -23,6 +24,19 @@ use tokio::time::delay_for;
 
 use crate::paths_provider::PathsProvider;
 
+#[derive(Debug, Snafu)]
+pub enum FileServerError<
+    PathsSource: std::error::Error + 'static,
+    ChannelClosedSource: std::error::Error + 'static,
+> {
+    #[snafu(display("File paths could not be retrieved."))]
+    PathsError { source: PathsSource },
+    #[snafu(display("Output channel closed."))]
+    ChannelClosedError { source: ChannelClosedSource },
+    #[snafu(display("Overflow ocurred when calculating timer."))]
+    TimeOverflowError,
+}
+
 /// `FileServer` is a Source which cooperatively schedules reads over files,
 /// converting the lines of said files into `LogLine` structures. As
 /// `FileServer` is intended to be useful across multiple operating systems with
@@ -72,7 +86,7 @@ where
         self,
         mut chans: C,
         shutdown: S,
-    ) -> Result<Shutdown, <C as Sink<Vec<(Bytes, String)>>>::Error>
+    ) -> Result<Shutdown, FileServerError<PP::Error, <C as Sink<Vec<(Bytes, String)>>>::Error>>
     where
         C: Sink<Vec<(Bytes, String)>> + Unpin,
         <C as Sink<Vec<(Bytes, String)>>>::Error: std::error::Error,
@@ -92,7 +106,7 @@ where
         let mut known_small_files = HashSet::new();
 
         let mut existing_files = Vec::new();
-        for path in self.paths_provider.paths().into_iter() {
+        for path in self.paths_provider.paths().context(PathsError)? {
             if let Some(file_id) = self.fingerprinter.get_fingerprint_or_log_error(
                 &path,
                 &mut fingerprint_buffer,
@@ -189,7 +203,9 @@ where
             let now_time = time::Instant::now();
             if next_glob_time <= now_time {
                 // Schedule the next glob time.
-                next_glob_time = now_time.checked_add(self.glob_minimum_cooldown).unwrap();
+                next_glob_time = now_time
+                    .checked_add(self.glob_minimum_cooldown)
+                    .ok_or(FileServerError::TimeOverflowError)?;
 
                 if stats.started_at.elapsed() > Duration::from_secs(1) {
                     stats.report();
@@ -204,7 +220,7 @@ where
                 for (_file_id, watcher) in &mut fp_map {
                     watcher.set_file_findable(false); // assume not findable until found
                 }
-                for path in self.paths_provider.paths().into_iter() {
+                for path in self.paths_provider.paths().context(PathsError)? {
                     if let Some(file_id) = self.fingerprinter.get_fingerprint_or_log_error(
                         &path,
                         &mut fingerprint_buffer,
@@ -342,14 +358,7 @@ where
             let start = time::Instant::now();
             let to_send = std::mem::take(&mut lines);
             let mut stream = stream::once(futures::future::ok(to_send));
-            let result = block_on(chans.send_all(&mut stream));
-            match result {
-                Ok(()) => {}
-                Err(error) => {
-                    error!(message = "Output channel closed.", error = ?error);
-                    return Err(error);
-                }
-            }
+            block_on(chans.send_all(&mut stream)).context(ChannelClosedError)?;
             stats.record("sending", start.elapsed());
 
             let start = time::Instant::now();

lib/file-source/src/paths_provider/glob.rs

@@ -1,11 +1,39 @@
 //! [`Glob`] paths provider.
 
 use super::PathsProvider;
-
 use glob::Pattern;
+use globwalk::glob;
+use snafu::{ResultExt, Snafu};
 use std::path::PathBuf;
 
-pub use glob::MatchOptions;
+#[derive(Debug, Snafu)]
+/// An error that arised either during parsing or execution of this glob.
+pub enum GlobError {
+    /// Include glob pattern could not be parsed.
+    #[snafu(display("Include glob pattern {} could not be parsed", glob))]
+    InvalidIncludePattern {
+        /// The glob string that produced the error.
+        glob: String,
+        /// The underlying error.
+        source: globwalk::GlobError,
+    },
+    /// Exclude glob pattern could not be parsed.
+    #[snafu(display("Exclude glob pattern {} could not be parsed", glob))]
+    InvalidExcludePattern {
+        /// The glob string that produced the error.
+        glob: String,
+        /// The underlying error.
+        source: glob::PatternError,
+    },
+    /// Failed while iterating on the glob.
+    #[snafu(display("Failed while iterating on the glob {}", glob))]
+    WalkError {
+        /// The glob string that produced the error.
+        glob: String,
+        /// The underlying error.
+        source: globwalk::WalkError,
+    },
+}
 
 /// A glob-based path provider.
 ///
@@ -14,54 +42,85 @@ pub use glob::MatchOptions;
 pub struct Glob {
     include_patterns: Vec<String>,
     exclude_patterns: Vec<Pattern>,
-    glob_match_options: MatchOptions,
 }
 
 impl Glob {
     /// Create a new [`Glob`].
     ///
-    /// Returns `None` if patterns aren't valid.
+    /// Returns `GlobError` if any of the patterns is not valid.
     pub fn new(
-        include_patterns: &[PathBuf],
-        exclude_patterns: &[PathBuf],
-        glob_match_options: MatchOptions,
-    ) -> Option<Self> {
-        let include_patterns = include_patterns
+        include_patterns: &[String],
+        exclude_patterns: &[String],
+    ) -> Result<Self, GlobError> {
+        // Validate include patterns. We can't parse the `GlobWalkers` and save them in our struct
+        // for later use because they are mutable iterators and don't implement the
+        // `std::clone::Clone` trait.
+        include_patterns
             .iter()
-            .map(|path| path.to_str().map(ToOwned::to_owned))
-            .collect::<Option<_>>()?;
+            .map(|include_pattern| -> Result<_, _> {
+                let glob = glob(include_pattern).with_context(|| InvalidIncludePattern {
+                    glob: include_pattern.to_owned(),
+                })?;
+
+                Ok(glob)
+            })
+            .collect::<Result<Vec<_>, _>>()?;
 
+        let include_patterns = include_patterns.to_owned();
+
+        // Validate exclude patterns.
         let exclude_patterns = exclude_patterns
             .iter()
-            .map(|path| path.to_str().map(|path| Pattern::new(path).ok()))
-            .flatten()
-            .collect::<Option<Vec<_>>>()?;
+            .map(|exclude_pattern| -> Result<_, _> {
+                let pattern =
+                    Pattern::new(exclude_pattern).with_context(|| InvalidExcludePattern {
+                        glob: exclude_pattern.to_owned(),
+                    })?;
+
+                Ok(pattern)
+            })
+            .collect::<Result<Vec<_>, _>>()?;
 
-        Some(Self {
+        Ok(Self {
             include_patterns,
             exclude_patterns,
-            glob_match_options,
         })
     }
 }
 
 impl PathsProvider for Glob {
     type IntoIter = Vec<PathBuf>;
+    type Error = GlobError;
 
-    fn paths(&self) -> Self::IntoIter {
-        self.include_patterns
-            .iter()
-            .flat_map(|include_pattern| {
-                glob::glob_with(include_pattern.as_str(), self.glob_match_options)
-                    .expect("failed to read glob pattern")
-                    .filter_map(|val| val.ok())
-            })
-            .filter(|candidate_path: &PathBuf| -> bool {
-                !self.exclude_patterns.iter().any(|exclude_pattern| {
-                    let candidate_path_str = candidate_path.to_str().unwrap();
-                    exclude_pattern.matches(candidate_path_str)
-                })
-            })
-            .collect()
+    fn paths(&self) -> Result<Self::IntoIter, Self::Error> {
+        let mut paths = Vec::new();
+
+        // Iterate over all include patterns, turn them into globs and walk them on the file system.
+        for include_pattern in &self.include_patterns {
+            let glob = glob(include_pattern).with_context(|| InvalidIncludePattern {
+                glob: include_pattern.to_owned(),
+            })?;
+
+            for dir_entry in glob {
+                let path = dir_entry
+                    .with_context(|| WalkError {
+                        glob: include_pattern.to_owned(),
+                    })?
+                    .into_path();
+
+                let is_excluded = path.to_str().map_or(false, |path| {
+                    self.exclude_patterns
+                        .iter()
+                        .any(|exclude_pattern| exclude_pattern.matches(path))
+                });
+
+                // Only include paths that does not match the list of our exclude patterns.
+                if !is_excluded {
+                    paths.push(path);
+                }
+            }
+        }
+
+        Ok(paths)
     }
 }

lib/file-source/src/paths_provider/mod.rs

@@ -27,7 +27,9 @@ pub mod glob;
 pub trait PathsProvider {
     /// Provides the iterator that returns paths.
     type IntoIter: IntoIterator<Item = PathBuf>;
+    /// Provides the error that can arise during iterator construction.
+    type Error: std::error::Error;
 
     /// Provides a set of paths.
-    fn paths(&self) -> Self::IntoIter;
+    fn paths(&self) -> Result<Self::IntoIter, Self::Error>;
 }