Chain stream instead of merge

Cargo.toml

@@ -9,7 +9,7 @@ resolver = "2"
 authors = ["EMF Team <[email protected]>"]
 edition = "2021"
 license = "MIT"
-version = "0.9.1"
+version = "0.9.2"
 
 [workspace.lints.rust]
 unreachable_pub = "deny"

lustrefs-exporter/debian/changelog

@@ -1,4 +1,4 @@
-prometheus-lustrefs-exporter (0.9.1-0.1) unstable; urgency=medium
+prometheus-lustrefs-exporter (0.9.2-0.1) unstable; urgency=medium
 
   * New release.
 

lustrefs-exporter/lustrefs_exporter.spec

@@ -1,5 +1,5 @@
 Name:           lustrefs_exporter
-Version:        0.9.1
+Version:        0.9.2
 Release:        1%{?dist}
 Summary:        prometheus exporter for lustre
 License:        MIT

lustrefs-exporter/src/jobstats.rs

@@ -134,6 +134,9 @@ pub fn jobstats_stream<R: BufRead + std::marker::Send + 'static>(
     let x = tokio::task::spawn_blocking(move || {
         let mut state = State::Empty;
 
+        // Send a new line to avoid printing at the same level as the previous stats
+        _ = tx.blocking_send('\n'.to_compact_string());
+
         for line in f.lines() {
             let r = handle_line(&tx, line.map_err(Error::Io), state);
 
@@ -335,7 +338,7 @@ pub mod tests {
 
         fut.await.unwrap();
 
-        assert_eq!(cnt, 21_147_876);
+        assert_eq!(cnt, 21_147_876 + 1);
     }
 
     #[tokio::test(flavor = "multi_thread")]
@@ -354,7 +357,7 @@ pub mod tests {
 
         fut.await.unwrap();
 
-        assert_eq!(cnt, 5_310_036);
+        assert_eq!(cnt, 5_310_036 + 1);
     }
 
     #[tokio::test(flavor = "multi_thread")]
@@ -373,7 +376,7 @@ pub mod tests {
 
         fut.await.unwrap();
 
-        assert_eq!(cnt, 1_728);
+        assert_eq!(cnt, 1_728 + 1);
     }
 
     const JOBSTAT_JOB: &str = r#"
@@ -421,6 +424,7 @@ job_stats:{}"#,
              4 + // 4 metrics per write_bytes
              10) // 10 metrics for "getattr" | "setattr" | "punch" | "sync" | "destroy" | "create" | "statfs" | "get_info" | "set_info" | "quotactl"
              * 10
+                + 1
         );
     }
 
@@ -440,6 +444,25 @@ job_stats:{}"#,
 
         fut.await.unwrap();
 
-        assert_eq!(cnt, 108);
+        assert_eq!(cnt, 108 + 1);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn parse_2_14_0_164_jobstats() {
+        let f = File::open("fixtures/jobstats_only/2.14.0_164.txt").unwrap();
+
+        let f = BufReader::with_capacity(128 * 1_024, f);
+
+        let (fut, mut rx) = jobstats_stream(f);
+
+        let mut output = r#"previous_stat{foo="bar"} 0"#.to_string();
+
+        while let Some(x) = rx.recv().await {
+            output.push_str(x.as_str());
+        }
+
+        fut.await.unwrap();
+
+        insta::assert_snapshot!(output);
     }
 }