EX-10318: add metrics for osc.*.state.current_state

Add metrics for osc.*.state.current_state
For lustrefs-exporter, if current_state is FULL or IDLE, the metrics value is 1.
Or the metrics value is 0

Signed-off-by: Feng Lei <flei@ddn.com>

Author: flei-ddn

lustre-collector/src/error.rs

@@ -31,3 +31,13 @@ impl From<combine::stream::easy::Errors<char, &str, usize>> for LustreCollectorE
         LustreCollectorError::CombineEasyError(err.map_range(|_| ""))
     }
 }
+
+impl<'a> From<combine::stream::easy::Errors<char, &'a str, combine::stream::PointerOffset<str>>>
+    for LustreCollectorError
+{
+    fn from(
+        err: combine::stream::easy::Errors<char, &'a str, combine::stream::PointerOffset<str>>,
+    ) -> Self {
+        LustreCollectorError::CombineEasyError(err.map_range(|_| "").map_position(|_| 0))
+    }
+}

lustre-collector/src/lib.rs

@@ -26,12 +26,76 @@ mod top_level_parser;
 pub mod types;
 
 pub use crate::error::LustreCollectorError;
-use combine::parser::EasyParser;
+use combine::{EasyParser, Parser, parser::token::satisfy};
 pub use lnetctl_parser::{parse as parse_lnetctl_output, parse_lnetctl_stats};
 pub use node_stats_parsers::{parse_cpustats_output, parse_meminfo_output};
 use std::{io, str};
 pub use types::*;
 
+/// Normalize lctl output to valid YAML format
+/// - If a line ends with '=', replace '=' with ':'
+/// - For all other non-empty lines, prepend a space
+fn normalize_lctl_output<Input>() -> impl Parser<Input, Output = String>
+where
+    Input: combine::Stream<Token = char>,
+    Input::Error: combine::ParseError<Input::Token, Input::Range, Input::Position>,
+{
+    use combine::{
+        attempt, choice, many, many1,
+        parser::char::{char, newline},
+    };
+
+    let normalize_line = choice((
+        // Line ending with '=' -> replace with ':'
+        attempt(
+            (many1(satisfy(|c| c != '\n' && c != '=')), char('='))
+                .map(|(s, _): (String, _)| format!("{}:", s)),
+        ),
+        // Other non-empty line -> prepend a space
+        many1(satisfy(|c| c != '\n')).map(|s: String| format!(" {}", s)),
+    ));
+
+    many(normalize_line.skip(newline())).map(|lines: Vec<String>| lines.join("\n"))
+}
+
+pub fn parse_osc_state_output(
+    osc_state_output: &[u8],
+) -> Result<Vec<Record>, LustreCollectorError> {
+    let osc_state_str = str::from_utf8(osc_state_output)?;
+
+    // Preprocess the output using combine parser to convert lctl format to valid YAML:
+    // - If a line ends with '=', replace '=' with ':'
+    // - For non-empty lines that don't end with '=', add a space at the beginning
+    let (processed_str, _) = normalize_lctl_output()
+        .easy_parse(osc_state_str)
+        .map_err(LustreCollectorError::from)?;
+
+    let osc_states: OscStates = serde_yaml::from_str(&processed_str)?;
+
+    // Convert each OSC state to a ControllerStats record
+    let records: Vec<Record> = osc_states
+        .into_iter()
+        .map(|(key, value)| {
+            // Remove "osc." prefix and ".state" suffix from keys
+            let cleaned_key = key
+                .strip_prefix("osc.")
+                .unwrap_or(&key)
+                .strip_suffix(".state")
+                .unwrap_or(&key)
+                .to_string();
+
+            Record::Controller(ControllerStats::OscState(ControllerStat {
+                kind: ControllerVariant::Osc,
+                param: Param("state".to_string()),
+                controller: Controller(cleaned_key),
+                value,
+            }))
+        })
+        .collect();
+
+    Ok(records)
+}
+
 fn check_output(
     records: Vec<Record>,
     state: &str,

lustre-collector/src/main.rs

@@ -5,8 +5,8 @@
 use clap::{Arg, ValueEnum, value_parser};
 use lustre_collector::{
     error::LustreCollectorError, mgs::mgs_fs_parser, parse_lctl_output, parse_lnetctl_output,
-    parse_lnetctl_stats, parse_mgs_fs_output, parse_recovery_status_output, parser,
-    recovery_status_parser, types::Record,
+    parse_lnetctl_stats, parse_mgs_fs_output, parse_osc_state_output, parse_recovery_status_output,
+    parser, recovery_status_parser, types::Record,
 };
 use std::{
     fmt, panic,
@@ -81,6 +81,15 @@ fn get_lnetctl_stats_output() -> Result<Vec<u8>, LustreCollectorError> {
     Ok(r.stdout)
 }
 
+fn get_osc_state_output() -> Result<Vec<u8>, LustreCollectorError> {
+    let r = Command::new("lctl")
+        .arg("get_param")
+        .arg("osc.*.state")
+        .output()?;
+
+    Ok(r.stdout)
+}
+
 fn main() -> ExitCode {
     match run() {
         Ok(()) => ExitCode::SUCCESS,
@@ -144,6 +153,13 @@ fn run() -> Result<(), LustreCollectorError> {
             Ok(recovery_statuses)
         });
 
+    let osc_state_handle = thread::spawn(move || -> Result<Vec<Record>, LustreCollectorError> {
+        let osc_state_output = get_osc_state_output()?;
+        let osc_states = parse_osc_state_output(&osc_state_output)?;
+
+        Ok(osc_states)
+    });
+
     let lnetctl_net_show_output = Command::new("lnetctl")
         .args(["net", "show", "-v", "4"])
         .output()
@@ -172,10 +188,16 @@ fn run() -> Result<(), LustreCollectorError> {
         Err(e) => panic::resume_unwind(e),
     };
 
+    let mut osc_state_records = match osc_state_handle.join() {
+        Ok(r) => r.unwrap_or_default(),
+        Err(e) => panic::resume_unwind(e),
+    };
+
     lctl_record.append(&mut lnet_record);
     lctl_record.append(&mut mgs_fs_record);
     lctl_record.append(&mut recovery_status_records);
     lctl_record.append(&mut lnetctl_stats_record);
+    lctl_record.append(&mut osc_state_records);
 
     let x = match format {
         Format::Json => serde_json::to_string(&lctl_record)?,

lustre-collector/src/snapshots/lustre_collector__types__tests__osc_states_parse_with_cleaned_keys.snap

@@ -0,0 +1,54 @@
+---
+source: lustre-collector/src/types.rs
+expression: records
+---
+[
+    Controller(
+        OscState(
+            ControllerStat {
+                kind: Osc,
+                param: Param(
+                    "state",
+                ),
+                controller: Controller(
+                    "fs-OST0000-osc-MDT0000",
+                ),
+                value: OscState {
+                    current_state: "FULL",
+                },
+            },
+        ),
+    ),
+    Controller(
+        OscState(
+            ControllerStat {
+                kind: Osc,
+                param: Param(
+                    "state",
+                ),
+                controller: Controller(
+                    "fs-OST0000-osc-ffff8d639e4f0800",
+                ),
+                value: OscState {
+                    current_state: "CONNECTING",
+                },
+            },
+        ),
+    ),
+    Controller(
+        OscState(
+            ControllerStat {
+                kind: Osc,
+                param: Param(
+                    "state",
+                ),
+                controller: Controller(
+                    "fs-OST0001-osc-MDT0000",
+                ),
+                value: OscState {
+                    current_state: "IDLE",
+                },
+            },
+        ),
+    ),
+]

lustre-collector/src/types.rs

@@ -29,6 +29,18 @@ impl Deref for Target {
     }
 }
 
+#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
+/// The Lustre controller corresponding to these stats (e.g., OSC, MDC).
+pub struct Controller(pub String);
+
+impl Deref for Controller {
+    type Target = str;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 /// The name of the stat.
 pub struct Param(pub String);
@@ -324,6 +336,29 @@ impl Deref for TargetVariant {
     }
 }
 
+#[derive(PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize, Clone, Copy)]
+pub enum ControllerVariant {
+    Osc,
+}
+
+impl fmt::Display for ControllerVariant {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match *self {
+            ControllerVariant::Osc => write!(f, "OSC"),
+        }
+    }
+}
+
+impl Deref for ControllerVariant {
+    type Target = str;
+
+    fn deref(&self) -> &Self::Target {
+        match *self {
+            ControllerVariant::Osc => "OSC",
+        }
+    }
+}
+
 #[derive(PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
 /// Stats specific to a target.
 pub struct TargetStat<T> {
@@ -333,6 +368,15 @@ pub struct TargetStat<T> {
     pub value: T,
 }
 
+#[derive(PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
+/// Stats specific to a controller (e.g., OSC, MDC).
+pub struct ControllerStat<T> {
+    pub kind: ControllerVariant,
+    pub param: Param,
+    pub controller: Controller,
+    pub value: T,
+}
+
 #[derive(PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
 /// Stats from parsing `ost.OSS.<PARAM>.stats`
 pub struct OssStat {
@@ -532,6 +576,12 @@ pub enum TargetStats {
     QuotaStatsOsd(TargetStat<QuotaStatsOsd>),
 }
 
+/// The controller stats currently collected
+#[derive(PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
+pub enum ControllerStats {
+    OscState(ControllerStat<OscState>),
+}
+
 #[derive(PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
 pub enum LNetStats {
     SendCount(LNetStat<i64>),
@@ -555,6 +605,7 @@ pub enum Record {
     LustreService(LustreServiceStats),
     Node(NodeStats),
     Target(TargetStats),
+    Controller(ControllerStats),
 }
 
 #[derive(PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
@@ -602,9 +653,17 @@ pub enum QuotaKind {
     Prj,
 }
 
+#[derive(PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
+pub struct OscState {
+    pub current_state: String,
+}
+
+pub type OscStates = std::collections::HashMap<String, OscState>;
+
 #[cfg(test)]
 mod tests {
     use super::*;
+    use insta::assert_debug_snapshot;
     use std::convert::TryInto;
 
     #[test]
@@ -633,4 +692,79 @@ mod tests {
             Err(e) => panic!("Error occurred: {:?}", e),
         }
     }
+
+    #[test]
+    fn test_osc_states_parse() {
+        // Test data matching fixtures/osc_states.json
+        // state_history is present but will be ignored during parsing
+        let yaml_data = r#"
+osc.fs-OST0000-osc-MDT0000.state:
+  current_state: FULL
+  state_history:
+    - [1775627216, CONNECTING]
+    - [1775627216, FULL]
+osc.fs-OST0000-osc-ffff8d639e4f0800.state:
+  current_state: CONNECTING
+  state_history:
+    - [1775627490, CONNECTING]
+osc.fs-OST0001-osc-MDT0000.state:
+  current_state: IDLE
+  state_history:
+    - [1775627600, CONNECTING]
+    - [1775627600, IDLE]
+"#;
+
+        let osc_states: OscStates = serde_yaml::from_str(yaml_data).unwrap();
+
+        // Verify we have 3 entries
+        assert_eq!(osc_states.len(), 3);
+
+        // Verify original keys (before cleaning)
+        let state0 = osc_states.get("osc.fs-OST0000-osc-MDT0000.state").unwrap();
+        assert_eq!(state0.current_state, "FULL");
+
+        let state1 = osc_states
+            .get("osc.fs-OST0000-osc-ffff8d639e4f0800.state")
+            .unwrap();
+        assert_eq!(state1.current_state, "CONNECTING");
+
+        let state2 = osc_states.get("osc.fs-OST0001-osc-MDT0000.state").unwrap();
+        assert_eq!(state2.current_state, "IDLE");
+    }
+
+    #[test]
+    fn test_osc_states_parse_with_cleaned_keys() {
+        use crate::parse_osc_state_output;
+
+        // Test data matching fixtures/osc_states.json in raw lctl format (before sed transformation)
+        let raw_lctl_output = r#"osc.fs-OST0000-osc-MDT0000.state=
+current_state: FULL
+state_history:
+  - [1775627216, CONNECTING]
+  - [1775627216, FULL]
+osc.fs-OST0000-osc-ffff8d639e4f0800.state=
+current_state: CONNECTING
+state_history:
+  - [1775627490, CONNECTING]
+osc.fs-OST0001-osc-MDT0000.state=
+current_state: IDLE
+state_history:
+  - [1775627600, CONNECTING]
+  - [1775627600, IDLE]
+"#;
+
+        let mut records = parse_osc_state_output(raw_lctl_output.as_bytes()).unwrap();
+
+        // Sort records by controller name to ensure consistent ordering for snapshot testing
+        records.sort_by(|a, b| match (a, b) {
+            (
+                Record::Controller(ControllerStats::OscState(a_stat)),
+                Record::Controller(ControllerStats::OscState(b_stat)),
+            ) => a_stat.controller.0.cmp(&b_stat.controller.0),
+            _ => std::cmp::Ordering::Equal,
+        });
+
+        // Use snapshot testing to verify the complete output structure
+        assert_debug_snapshot!(records);
+    }
 }