enhancement(sources): Allow line aggregation to never timeout

docs/reference/components/sources.cue

@@ -48,10 +48,12 @@ components: sources: [Name=string]: {
 						type: string: examples: ["^[^\\s]", "\\\\$", "^(INFO|ERROR) ", "[^;]$"]
 					}
 					timeout_ms: {
-						description: "The maximum time to wait for the continuation. Once this timeout is reached, the buffered message is guaranteed to be flushed, even if incomplete."
-						required:    true
+						description: "The maximum time to wait for the continuation. Once this timeout is reached, the buffered message is guaranteed to be flushed, even if incomplete. Set to \"none\" for no timeout."
+						required:    false
+						common:      true
 						sort:        4
 						type: uint: {
+							default: 1_000
 							examples: [1_000, 600_000]
 							unit: "milliseconds"
 						}

src/line_agg.rs

@@ -60,7 +60,7 @@ pub struct Config {
     /// The maximum time to wait for the continuation. Once this timeout is
     /// reached, the buffered message is guaranteed to be flushed, even if
     /// incomplete.
-    pub timeout: Duration,
+    pub timeout: Option<Duration>,
 }
 
 impl Config {
@@ -70,7 +70,7 @@ impl Config {
         let start_pattern = marker;
         let condition_pattern = start_pattern.clone();
         let mode = Mode::HaltBefore;
-        let timeout = Duration::from_millis(timeout_ms);
+        let timeout = Some(Duration::from_millis(timeout_ms));
 
         Self {
             start_pattern,
@@ -351,8 +351,9 @@ where
                 if self.config.start_pattern.is_match(line.as_ref()) {
                     // It was indeed a new line we need to filter.
                     // Set the timeout and buffer this line.
-                    self.timeouts
-                        .insert(entry.key().clone(), self.config.timeout);
+                    if let Some(timeout) = self.config.timeout {
+                        self.timeouts.insert(entry.key().clone(), timeout);
+                    }
                     entry.insert(Aggregate::new(line, context));
                     None
                 } else {
@@ -419,7 +420,7 @@ mod tests {
             start_pattern: Regex::new("^[^\\s]").unwrap(),
             condition_pattern: Regex::new("^[\\s]+").unwrap(),
             mode: Mode::ContinueThrough,
-            timeout: Duration::from_millis(10),
+            timeout: Some(Duration::from_millis(10)),
         };
         let expected = vec![
             "some usual line",
@@ -450,7 +451,7 @@ mod tests {
             start_pattern: Regex::new("\\\\$").unwrap(),
             condition_pattern: Regex::new("\\\\$").unwrap(),
             mode: Mode::ContinuePast,
-            timeout: Duration::from_millis(10),
+            timeout: Some(Duration::from_millis(10)),
         };
         let expected = vec![
             "some usual line",
@@ -481,7 +482,7 @@ mod tests {
             start_pattern: Regex::new("").unwrap(),
             condition_pattern: Regex::new("^(INFO|ERROR) ").unwrap(),
             mode: Mode::HaltBefore,
-            timeout: Duration::from_millis(10),
+            timeout: Some(Duration::from_millis(10)),
         };
         let expected = vec![
             "INFO some usual line",
@@ -512,7 +513,7 @@ mod tests {
             start_pattern: Regex::new("[^;]$").unwrap(),
             condition_pattern: Regex::new(";$").unwrap(),
             mode: Mode::HaltWith,
-            timeout: Duration::from_millis(10),
+            timeout: Some(Duration::from_millis(10)),
         };
         let expected = vec![
             "some usual line;",
@@ -538,7 +539,7 @@ mod tests {
             start_pattern: Regex::new("^[^\\s]").unwrap(),
             condition_pattern: Regex::new("^[\\s]+at").unwrap(),
             mode: Mode::ContinueThrough,
-            timeout: Duration::from_millis(10),
+            timeout: Some(Duration::from_millis(10)),
         };
         let expected = vec![concat!(
             "java.lang.Exception\n",
@@ -560,7 +561,7 @@ mod tests {
             start_pattern: Regex::new("^[^\\s]").unwrap(),
             condition_pattern: Regex::new("^[\\s]+from").unwrap(),
             mode: Mode::ContinueThrough,
-            timeout: Duration::from_millis(10),
+            timeout: Some(Duration::from_millis(10)),
         };
         let expected = vec![concat!(
             "foobar.rb:6:in `/': divided by 0 (ZeroDivisionError)\n",
@@ -594,7 +595,7 @@ mod tests {
             start_pattern: Regex::new("^\\s").unwrap(),
             condition_pattern: Regex::new("^\\s").unwrap(),
             mode: Mode::ContinueThrough,
-            timeout: Duration::from_millis(10),
+            timeout: Some(Duration::from_millis(10)),
         };
         let expected = vec![
             "not merged 1",
@@ -632,7 +633,7 @@ mod tests {
             start_pattern: Regex::new("").unwrap(),
             condition_pattern: Regex::new("^START ").unwrap(),
             mode: Mode::HaltBefore,
-            timeout: Duration::from_millis(10),
+            timeout: Some(Duration::from_millis(10)),
         };
         let expected = vec![
             "part 0.1\npart 0.2",

src/sources/docker.rs

@@ -1493,7 +1493,7 @@ mod integration_tests {
                 start_pattern: "^[^\\s]".to_owned(),
                 condition_pattern: "^[\\s]+at".to_owned(),
                 mode: line_agg::Mode::ContinueThrough,
-                timeout_ms: 10,
+                timeout_ms: Some(10),
             }),
             ..DockerConfig::default()
         };

src/sources/file.rs

@@ -1186,7 +1186,7 @@ mod tests {
                 start_pattern: "INFO".to_owned(),
                 condition_pattern: "INFO".to_owned(),
                 mode: line_agg::Mode::HaltBefore,
-                timeout_ms: 25, // less than 50 in sleep()
+                timeout_ms: Some(25), // less than 50 in sleep()
             }),
             ..test_default_file_config(&dir)
         };

src/sources/util/multiline_config.rs

@@ -12,7 +12,12 @@ pub struct MultilineConfig {
     pub start_pattern: String,
     pub condition_pattern: String,
     pub mode: line_agg::Mode,
-    pub timeout_ms: u64,
+    #[serde(default = "default_timeout_ms", with = "optional_u64")]
+    pub timeout_ms: Option<u64>,
+}
+
+const fn default_timeout_ms() -> Option<u64> {
+    Some(1000)
 }
 
 impl TryFrom<&MultilineConfig> for line_agg::Config {
@@ -31,7 +36,7 @@ impl TryFrom<&MultilineConfig> for line_agg::Config {
         let condition_pattern = Regex::new(condition_pattern)
             .with_context(|| InvalidMultilineConditionPattern { condition_pattern })?;
         let mode = mode.clone();
-        let timeout = Duration::from_millis(*timeout_ms);
+        let timeout = timeout_ms.map(Duration::from_millis);
 
         Ok(Self {
             start_pattern,
@@ -63,3 +68,69 @@ pub enum Error {
         source: regex::Error,
     },
 }
+
+mod optional_u64 {
+    use serde::de::{self, Deserializer, Unexpected, Visitor};
+    use serde::ser::Serializer;
+
+    pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<Option<u64>, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        struct OptionalU64;
+
+        impl<'de> Visitor<'de> for OptionalU64 {
+            type Value = Option<u64>;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
+                formatter.write_str("number or \"none\"")
+            }
+
+            fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
+            where
+                E: de::Error,
+            {
+                match value {
+                    "none" => Ok(None),
+                    s => Err(de::Error::invalid_value(
+                        Unexpected::Str(s),
+                        &"number or \"none\"",
+                    )),
+                }
+            }
+
+            fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E>
+            where
+                E: de::Error,
+            {
+                if value > 0 {
+                    Ok(Some(value as u64))
+                } else {
+                    Err(de::Error::invalid_value(
+                        Unexpected::Signed(value),
+                        &"positive integer",
+                    ))
+                }
+            }
+
+            fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E>
+            where
+                E: de::Error,
+            {
+                Ok(Some(value))
+            }
+        }
+
+        deserializer.deserialize_any(OptionalU64)
+    }
+
+    pub(super) fn serialize<S>(value: &Option<u64>, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        match *value {
+            Some(v) => serializer.serialize_u64(v),
+            None => serializer.serialize_str("none"),
+        }
+    }
+}