Add recover_status parser lustrefs_exporter

Cargo.toml

@@ -35,6 +35,7 @@ serde_json = "1"
 serde_yaml = "0.9"
 serial_test = "3.2"
 sysinfo = "0.29"
+test-case = "3.3"
 thiserror = "2"
 tokio = "1"
 tower = "0.5"

lustre-collector/Cargo.toml

@@ -19,6 +19,7 @@ tracing-subscriber.workspace = true
 [dev-dependencies]
 include_dir.workspace = true
 insta.workspace = true
+test-case.workspace = true
 tokio = { workspace = true, features = ["full"] }
 criterion = { workspace = true, features = ["html_reports", "async_tokio"] }
 sysinfo.workspace = true

lustre-collector/src/fixtures/valid/lustre-2.14.0_ddn212/2.14.0_ddn212_recovery.txt

@@ -0,0 +1,36 @@
+obdfilter.200NVX2-OST0000.recovery_status=
+status: COMPLETE
+recovery_start: 1761567697
+recovery_duration: 1
+completed_clients: 8/8
+replayed_requests: 0
+last_transno: 17184814233
+VBR: DISABLED
+IR: ENABLED
+obdfilter.200NVX2-OST0003.recovery_status=
+status: COMPLETE
+recovery_start: 1759494115
+recovery_duration: 15
+completed_clients: 8/8
+replayed_requests: 0
+last_transno: 12934942643
+VBR: DISABLED
+IR: DISABLED
+obdfilter.200NVX2-OST0004.recovery_status=
+status: COMPLETE
+recovery_start: 1759494115
+recovery_duration: 14
+completed_clients: 8/8
+replayed_requests: 0
+last_transno: 12934956643
+VBR: DISABLED
+IR: DISABLED
+obdfilter.200NVX2-OST0007.recovery_status=
+status: COMPLETE
+recovery_start: 1759494115
+recovery_duration: 14
+completed_clients: 8/8
+replayed_requests: 0
+last_transno: 12934943652
+VBR: DISABLED
+IR: DISABLED

lustre-collector/src/parser.rs

@@ -6,7 +6,7 @@ use crate::{
     ldlm, llite, mdd_parser,
     mds::{self, client_count_parser},
     mgs::mgs_parser,
-    nodemap, osd_parser, oss, quota, top_level_parser,
+    nodemap, osd_parser, oss, quota, recovery_status_parser, top_level_parser,
     types::Record,
 };
 use combine::{Parser, Stream, choice, error::ParseError, many};
@@ -24,6 +24,7 @@ pub fn params() -> Vec<String> {
         .chain(mdd_parser::params())
         .chain(quota::params())
         .chain(nodemap::params())
+        .chain(recovery_status_parser::params())
         .collect()
 }
 
@@ -44,6 +45,7 @@ where
         mdd_parser::parse().map(|x| vec![x]),
         quota::parse().map(|x| vec![x]),
         nodemap::parse().map(|x| vec![x]),
+        recovery_status_parser::parse(),
     )))
     .map(|xs: Vec<_>| xs.into_iter().flatten().collect())
 }
@@ -86,6 +88,7 @@ mod tests {
     test_fixtures!(test_lustre_2_14_0_ddn145_fixtures, "*ddn145*");
 
     test_fixtures!(test_lustre_2_14_0_ddn133_fixtures, "*ddn133*");
+    test_fixtures!(test_lustre_2_14_0_ddn212_fixtures, "*ddn212*");
 
     #[test]
     fn test_params() {

lustre-collector/src/recovery_status_parser.rs

@@ -2,6 +2,8 @@
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.
 
+use std::iter;
+
 use crate::{
     base_parsers::{digits, param, period, target, till_newline},
     types::{Param, Record, RecoveryStatus, Target, TargetStat, TargetStats, TargetVariant},
@@ -71,7 +73,8 @@ where
         })
 }
 
-fn clients_line<I>(x: &'static str) -> impl Parser<I, Output = u64>
+/// Parses a client line containing the completed and a optional value for the total count (e.g., "completed: 5/10").
+fn clients_line<I>(x: &'static str) -> impl Parser<I, Output = (u64, Option<u64>)>
 where
     I: Stream<Token = char>,
     I::Error: ParseError<I::Token, I::Range, I::Position>,
@@ -83,15 +86,18 @@ where
         optional((token('/'), digits())),
         optional(newline().map(drop).or(eof())),
     )
-        .map(|(_, _, x, _, _): (_, _, u64, _, _)| x)
+        .map(|(_, _, x, y, _): (_, _, u64, Option<(_, u64)>, _)| (x, y.map(|(_, v)| v)))
 }
 
 #[derive(Debug)]
 enum RecoveryStat {
-    Status(RecoveryStatus),
     Completed(u64),
     Connected(u64),
     Evicted(u64),
+    RecoveryDuration(u64),
+    Status(RecoveryStatus),
+    TimeRemaining(u64),
+    Total(Option<u64>),
 }
 
 pub struct StatName(pub String);
@@ -105,6 +111,20 @@ where
     many1(alpha_num().or(one_of("_-".chars()))).map(StatName)
 }
 
+/// Parses all the simple recovery stats for a target, which are just plain u64 values
+fn simple_client_stat<I>(
+    name: &'static str,
+    constructor: fn(u64) -> RecoveryStat,
+) -> impl Parser<I, Output = Vec<RecoveryStat>>
+where
+    I: Stream<Token = char>,
+    I::Error: ParseError<I::Token, I::Range, I::Position>,
+{
+    clients_line(name)
+        .skip(optional(newline()))
+        .map(move |(x, _)| vec![constructor(x)])
+}
+
 fn target_recovery_stats<I>() -> impl Parser<I, Output = Vec<RecoveryStat>>
 where
     I: Stream<Token = char>,
@@ -113,22 +133,23 @@ where
     many(choice((
         status_line()
             .skip(optional(newline()))
-            .map(RecoveryStat::Status)
-            .map(Some),
-        clients_line("completed_clients")
-            .skip(optional(newline()))
-            .map(RecoveryStat::Completed)
-            .map(Some),
+            .map(|x| vec![RecoveryStat::Status(x)]),
+        simple_client_stat("recovery_duration", RecoveryStat::RecoveryDuration),
+        simple_client_stat("completed_clients", RecoveryStat::Completed),
+        simple_client_stat("time_remaining", RecoveryStat::TimeRemaining),
+        simple_client_stat("evicted_clients", RecoveryStat::Evicted),
         clients_line("connected_clients")
             .skip(optional(newline()))
-            .map(RecoveryStat::Connected)
-            .map(Some),
-        clients_line("evicted_clients")
-            .skip(optional(newline()))
-            .map(RecoveryStat::Evicted)
-            .map(Some),
+            .map(|(x, y)| {
+                iter::once(RecoveryStat::Connected(x))
+                    .chain(
+                        y.map(|total| vec![RecoveryStat::Total(Some(total))])
+                            .unwrap_or_default(),
+                    )
+                    .collect()
+            }),
         // This will ignore line/field we don't care
-        attempt((stat_name(), token(':'), till_newline().skip(newline()))).map(|_| None),
+        attempt((stat_name(), token(':'), till_newline().skip(newline()))).map(|_| vec![]),
     )))
     .map(|xs: Vec<_>| xs.into_iter().flatten().collect())
 }
@@ -176,6 +197,28 @@ where
                             value: *value,
                         })
                     }
+                    RecoveryStat::RecoveryDuration(value) => {
+                        TargetStats::RecoveryDuration(TargetStat {
+                            kind,
+                            param: param.clone(),
+                            target: target.clone(),
+                            value: *value,
+                        })
+                    }
+                    RecoveryStat::TimeRemaining(value) => {
+                        TargetStats::RecoveryTimeRemaining(TargetStat {
+                            kind,
+                            param: param.clone(),
+                            target: target.clone(),
+                            value: *value,
+                        })
+                    }
+                    RecoveryStat::Total(value) => TargetStats::RecoveryTotalClients(TargetStat {
+                        kind,
+                        param: param.clone(),
+                        target: target.clone(),
+                        value: value.unwrap_or(0),
+                    }),
                 })
                 .collect()
         })
@@ -186,20 +229,19 @@ where
     I: Stream<Token = char>,
     I::Error: ParseError<I::Token, I::Range, I::Position>,
 {
-    many(
-        (
-            target_status(),
-            skip_until(attempt(ost_or_mdt().map(drop)).or(eof())),
-        )
-            .map(|(x, _)| x.into_iter().map(Record::Target).collect()),
+    (
+        target_status(),
+        skip_until(attempt(ost_or_mdt().map(drop)).or(eof())),
     )
-    .map(|x: Vec<Vec<Record>>| x.into_iter().flatten().collect())
+        .map(|(x, _)| x.into_iter().map(Record::Target).collect())
 }
 
 #[cfg(test)]
 mod tests {
-    use crate::recovery_status_parser::{clients_line, parse, target_recovery_stats};
+    use crate::parser::parse;
+    use crate::recovery_status_parser::{clients_line, target_recovery_stats};
     use combine::{Parser, parser::EasyParser, stream::position};
+    use test_case::test_case;
 
     #[test]
     fn test_multiple() {
@@ -235,14 +277,12 @@ mod tests {
         insta::assert_debug_snapshot!(records);
     }
 
-    #[test]
-    fn test_clients_line() {
-        let result = clients_line("completed_clients").parse("completed_clients: 3/7\n");
-        assert_eq!(result, Ok((3, "")));
-        let result = clients_line("connected_clients").parse("connected_clients: 3/7\n");
-        assert_eq!(result, Ok((3, "")));
-        let result = clients_line("completed_clients").parse("completed_clients: 3\n");
-        assert_eq!(result, Ok((3, "")));
+    #[test_case("completed_clients", "completed_clients: 3/7\n", (3, Some(7)); "completed clients with total")]
+    #[test_case("connected_clients", "connected_clients: 3/7\n", (3, Some(7)); "connected clients with total")]
+    #[test_case("completed_clients", "completed_clients: 3\n", (3, None); "completed clients without total")]
+    fn test_clients_line(field_name: &'static str, input: &str, expected: (u64, Option<u64>)) {
+        let result = clients_line(field_name).parse(input);
+        assert_eq!(result, Ok((expected, "")));
     }
 
     #[test]
@@ -259,7 +299,19 @@ IR: ENABLED
 
         let (records, _): (Vec<_>, _) = target_recovery_stats().parse(x).unwrap();
 
-        insta::assert_debug_snapshot!(records);
+        insta::assert_debug_snapshot!(records, @r"
+        [
+            Status(
+                Complete,
+            ),
+            RecoveryDuration(
+                150,
+            ),
+            Completed(
+                4,
+            ),
+        ]
+        ");
     }
 
     #[test]
@@ -275,6 +327,26 @@ completed_clients: 3
 
         let (records, _): (Vec<_>, _) = target_recovery_stats().parse(x).unwrap();
 
-        insta::assert_debug_snapshot!(records);
+        insta::assert_debug_snapshot!(records, @r"
+        [
+            Status(
+                Recovering,
+            ),
+            TimeRemaining(
+                119,
+            ),
+            Connected(
+                3,
+            ),
+            Total(
+                Some(
+                    7,
+                ),
+            ),
+            Completed(
+                3,
+            ),
+        ]
+        ");
     }
 }

lustre-collector/src/snapshots/lustre_collector__parser__tests__params.snap

@@ -67,4 +67,6 @@ expression: params()
     "qmt.*.{dt,md}-*.glb-grp",
     "nodemap.*.dt_stats",
     "nodemap.*.md_stats",
+    "obdfilter.*OST*.recovery_status",
+    "mdt.*MDT*.recovery_status",
 ]