Parse PSI cgroup v2 metrics (#1214)

* Parse PSI cgroup v2 metrics

This commit introduces the parsing of v2 PSI metrics into lading.
We special-case memory.pressure, cpu.pressure and io.pressure to
be parsed.

Signed-off-by: Brian L. Troutwine <brian.troutwine@datadoghq.com>

* changelog

Signed-off-by: Brian L. Troutwine <brian.troutwine@datadoghq.com>

* tests

Signed-off-by: Brian L. Troutwine <brian.troutwine@datadoghq.com>

* debug -> warn

Signed-off-by: Brian L. Troutwine <brian.troutwine@datadoghq.com>

---------

Signed-off-by: Brian L. Troutwine <brian.troutwine@datadoghq.com>

Author: blt

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+## Added
+- cgroup.v2 PSI metrics are now parsed. 
+
 ## [0.25.3]
 ## Changed
 - Various dependencies updated, notably `hyper` is now 1.x.

lading/src/observer/linux/cgroup/v2.rs

@@ -20,6 +20,8 @@ pub enum Error {
     ParseFloat(#[from] std::num::ParseFloatError),
     #[error("Cgroup v2 not found")]
     CgroupV2NotFound,
+    #[error("Parsing PSI error: {0}")]
+    ParsingPsi(String),
 }
 
 /// Determines the cgroup v2 path for a given PID.
@@ -70,8 +72,21 @@ pub(crate) async fn poll(file_path: &Path, labels: &[(String, String)]) -> Resul
 
                                     match fs::read_to_string(&file_path).await {
                                         Ok(content) => {
-                                            let content = content.trim();
+                                            if file_name == "memory.pressure"
+                                                || file_name == "io.pressure"
+                                                || file_name == "cpu.pressure"
+                                            {
+                                                if let Err(err) =
+                                                    parse_pressure(&content, &metric_prefix, labels)
+                                                {
+                                                    warn!("[{path}] Failed to parse PSI contents: {err:?}",
+                                                        path = file_path.to_string_lossy()
+                                                    );
+                                                }
+                                                continue;
+                                            }
 
+                                            let content = content.trim();
                                             // The format of cgroupv2 interface
                                             // files is defined here:
                                             // https://docs.kernel.org/admin-guide/cgroup-v2.html#interface-files
@@ -171,3 +186,132 @@ fn kv_pairs(
     }
     Ok(())
 }
+
+fn parse_pressure(content: &str, prefix: &str, labels: &[(String, String)]) -> Result<(), Error> {
+    for line in content.lines() {
+        parse_pressure_line(line, prefix, |metric: String, value: f64| {
+            gauge!(metric, labels).set(value);
+        })?;
+    }
+    Ok(())
+}
+
+fn parse_pressure_line<F>(line: &str, prefix: &str, mut f: F) -> Result<(), Error>
+where
+    F: FnMut(String, f64),
+{
+    // [some|full] avg10=FLOAT avg60=FLOAT avg300=FLOAT total=FLOAT
+    let mut parts = line.split_whitespace();
+    if let Some(category) = parts.next() {
+        for field in parts {
+            let Some((key, val)) = field.split_once('=') else {
+                return Err(Error::ParsingPsi(format!("Invalid psi field: {field}")));
+            };
+            // It might be that total is an integer but for the sake of
+            // simplicity we'll parse as f64. It has to become a float anyway
+            // when we write it out as a metric.
+            let value = val
+                .parse::<f64>()
+                .map_err(|err| Error::ParsingPsi(format!("{val} -> {err}")))?;
+
+            let metric_name = format!("{prefix}.{category}.{key}");
+            f(metric_name, value);
+        }
+    } else {
+        warn!("Unexpected blank category in psi file, skipping line: {line}");
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::parse_pressure_line;
+
+    #[test]
+    fn parse_pressure_line_multiple_fields() {
+        let line = "some avg10=0.42 avg60=1.0 total=42";
+        let prefix = "cgroup.v2.memory.pressure";
+
+        let mut results = Vec::new();
+        let res = parse_pressure_line(line, prefix, |metric, value| {
+            results.push((metric, value));
+        });
+
+        assert!(res.is_ok());
+        assert_eq!(results.len(), 3);
+
+        assert_eq!(
+            results[0],
+            (String::from("cgroup.v2.memory.pressure.some.avg10"), 0.42)
+        );
+        assert_eq!(
+            results[1],
+            (String::from("cgroup.v2.memory.pressure.some.avg60"), 1.0)
+        );
+        assert_eq!(
+            results[2],
+            (String::from("cgroup.v2.memory.pressure.some.total"), 42.0)
+        );
+    }
+
+    #[test]
+    fn parse_pressure_line_blank_line() {
+        let line = "";
+        let prefix = "cgroup.v2.memory.pressure";
+
+        let mut results = Vec::new();
+        let res = parse_pressure_line(line, prefix, |metric, value| {
+            results.push((metric, value));
+        });
+
+        assert!(res.is_ok());
+        assert!(results.is_empty());
+    }
+
+    #[test]
+    fn parse_pressure_line_incomplete() {
+        let line = "some";
+        let prefix = "cgroup.v2.memory.pressure";
+
+        let mut results = Vec::new();
+        let res = parse_pressure_line(line, prefix, |metric, value| {
+            results.push((metric, value));
+        });
+
+        assert!(res.is_ok());
+        assert!(results.is_empty());
+    }
+
+    #[test]
+    fn parse_pressure_line_malformed_field() {
+        let line = "some avg10=0.0 avg60?";
+        let prefix = "cgroup.v2.memory.pressure";
+
+        let mut results = Vec::new();
+        let res = parse_pressure_line(line, prefix, |metric, value| {
+            results.push((metric, value));
+        });
+
+        // Intentionally grab as many fields as possible
+        assert!(res.is_err());
+        assert_eq!(results.len(), 1);
+        assert_eq!(
+            results[0],
+            (String::from("cgroup.v2.memory.pressure.some.avg10"), 0.0)
+        );
+    }
+
+    #[test]
+    fn parse_pressure_line_invalid_value() {
+        let line = "some avg10=hello";
+        let prefix = "cgroup.v2.memory.pressure";
+
+        let mut results = Vec::new();
+        let res = parse_pressure_line(line, prefix, |metric, value| {
+            results.push((metric, value));
+        });
+
+        assert!(res.is_err());
+        assert!(results.is_empty());
+    }
+}