Merge pull request #15 from seyeongkim-lab/feature/flow-view-and-concurrent-probes

Fix stream payload, probe concurrently, add a flow view

Author: xtrusia

.github/workflows/ci.yml

@@ -31,3 +31,24 @@ jobs:
         run: cargo build --verbose
       - name: Test
         run: cargo test --verbose
+
+  msrv:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Resolve declared MSRV
+        id: msrv
+        run: |
+          set -euo pipefail
+          version=$(sed -n 's/^rust-version = "\(.*\)"$/\1/p' Cargo.toml)
+          if [ -z "$version" ]; then
+            echo "rust-version is missing from Cargo.toml" >&2
+            exit 1
+          fi
+          echo "version=$version" >> "$GITHUB_OUTPUT"
+      - uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ steps.msrv.outputs.version }}
+      - uses: Swatinem/rust-cache@v2
+      - name: Build with the declared MSRV
+        run: cargo build --locked --verbose

CHANGELOG.md

@@ -6,6 +6,49 @@ Notable changes are documented here. The format follows
 
 ## [Unreleased]
 
+### Added
+
+- Added a flow view, opened with `m`, that shows the OSD, placement group, and
+  object behind the ops in the trace buffer. radostrace lines already name all
+  three, so the view needs no extra remote command; with only osdtrace running
+  it collapses to OSD and placement group. `o` orders it by op count or by
+  latency and `s` reverses the order. Each row also carries the mean op size,
+  which separates a slow heavy request from one that is slow for the little it
+  asks for. A read reports the length it requested rather than bytes returned,
+  so a client that always asks for 4MiB reports 4MiB whatever the object holds.
+- Insights now name the checks behind a `HEALTH_WARN` or `HEALTH_ERR` instead of
+  telling the operator to go run `ceph health detail`. `ceph -s` already carried
+  them, so this costs no extra remote command.
+- The OSD table shows the commit and apply latency from `ceph osd perf`, which
+  puts Ceph's own view of a slow OSD next to the eBPF numbers. The query is
+  optional, so a cluster that refuses it keeps the rest of its status.
+- The node table shows the share of the last 10s that a host spent stalled on IO,
+  read from `/proc/pressure/io`, and an insight fires past 5%. Unlike a device
+  utilization figure this needs no OSD to block device mapping to be meaningful.
+
+### Fixed
+
+- `x` now clears every captured trace source rather than osdtrace alone.
+- Node readiness no longer breaks on hosts without `ceph-osd` processes. The
+  remote OSD count fell back through `pgrep -c ... || echo 0`, which emitted two
+  lines because `pgrep -c` prints `0` and exits 1 on no match. The extra line
+  made the node stream payload invalid JSON on mon-only hosts.
+
+### Changed
+
+- The cluster status stream now runs its three admin queries at once instead of
+  one after another. On a four node microceph cluster the tick period dropped
+  from 2278ms to 1400ms at the default `refresh_secs = 1`, with the same payload.
+  `refresh_secs` is the pause between ticks, not the period, and the README now
+  says so.
+- `doctor` and `snapshot` now probe hosts concurrently instead of one at a time,
+  with at most 8 hosts in flight. Each host still sees one SSH connection at a
+  time and the doctor report keeps its previous order.
+- Raised the declared minimum supported Rust version to 1.88. The source uses
+  let chains, which are stable only from 1.88, so builds on 1.85 through 1.87
+  failed despite the previous `rust-version = "1.85"`. CI now builds against the
+  declared MSRV.
+
 ## [0.1.4] - 2026-07-07
 
 ### Added

Cargo.toml

@@ -2,7 +2,7 @@
 name = "cephlens"
 version = "0.1.4"
 edition = "2024"
-rust-version = "1.85"
+rust-version = "1.88.0"
 authors = ["cephlens contributors"]
 description = "A lab-first Ceph investigation TUI with SSH-based status and osdtrace views"
 readme = "README.md"

README.md

@@ -5,7 +5,7 @@
 <h1 align="center">CephLens</h1>
 
 [![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)
-![Rust](https://img.shields.io/badge/rust-1.85%2B-orange.svg)
+![Rust](https://img.shields.io/badge/rust-1.88%2B-orange.svg)
 [![CI](https://github.com/xtrusia/cephlens/actions/workflows/ci.yml/badge.svg)](https://github.com/xtrusia/cephlens/actions/workflows/ci.yml)
 
 An SSH-driven Ceph investigation TUI with live cluster status, per-node
@@ -50,8 +50,16 @@ In a source clone, use `cargo run -- init-config`, edit the generated file, then
 ## Features
 
 - Live cluster health, quorum, OSD counts, and IO throughput over a single SSH stream.
-- Per-node readiness: connection state, OSD ids, CPU and memory percent, and Ceph version/deployment.
+- Per-node readiness: connection state, OSD ids, CPU and memory percent, IO stall
+  share from `/proc/pressure/io`, and Ceph version/deployment.
+- Per-OSD commit and apply latency from `ceph osd perf` next to the eBPF trace
+  numbers, and health check names read straight out of `ceph -s`.
 - osdtrace eBPF latency tracing with per-OSD and per-PG breakdown of queue, BlueStore, and KV-commit latency.
+- A flow view of the OSD, placement group, and object behind the observed
+  ops, built from the trace lines already streaming. radostrace names the
+  object so the view has three levels; with only osdtrace it collapses to
+  OSD and placement group. Rows carry the mean op size next to the latency; a
+  read reports the length it requested, not the bytes returned.
 - No standing agent: no permanent daemon on the nodes; the osdtrace runner script removes itself on stop, quit, or TTL expiry. (The cephtrace tracer binaries you deploy do persist under `~/.cephlens/bin/`.)
 - Edit hosts and trace settings live in the TUI; changes apply to open SSH streams immediately.
 - Export recorded sessions as Markdown reports with the same diagnostic rules used by the TUI.
@@ -60,7 +68,8 @@ In a source clone, use `cargo run -- init-config`, edit the generated file, then
 
 Controller (where the TUI runs):
 
-- Rust 1.85+ (edition 2024) to build.
+- Rust 1.88+ (edition 2024) to build. The source uses let chains, which are
+  stable only from 1.88 onward.
 - An OpenSSH client on `PATH`, with every host reachable over non-interactive SSH (key-based, no password prompt). Windows 10/11 ship this as the optional OpenSSH Client feature; macOS and Linux include it by default.
 
 Ceph nodes:
@@ -216,6 +225,7 @@ admin host:
   sudo -n ceph -s --format json
   sudo -n ceph osd tree --format json
   sudo -n ceph osd df --format json
+  sudo -n ceph osd perf --format json
   sudo -n rados --version
 
 bench command:
@@ -281,6 +291,9 @@ p          run a probe readiness check
 c          edit config
 t/f/r      view osdtrace / kfstrace / radostrace; press again to start or stop (confirmed)
 a          start or stop all trace sources (confirmed)
+m          flow view: the osd -> pg -> object mapping behind the live ops
+o          flow view: order by op count or by latency
+s          flow view: reverse the order
 i          install osdtrace
 x          clear captured trace events
 ?          toggle the help overlay
@@ -312,9 +325,11 @@ The integrated trace panel can show osdtrace, kfstrace, or radostrace data. The
 On wide terminals the trace panel appears on the right; on tall terminals it
 appears below the dashboard.
 Live TUI mode keeps one SSH stream open for cluster status and one stream per
-host for node readiness. Each stream emits data once per second by default and
-the node table shows connection state (`live`, `dial`, `retry`, `error`), OSD
-ids, CPU percentage, and memory percentage.
+host for node readiness. `refresh_secs` is the pause between ticks, so the
+period an operator sees is that pause plus the time the remote queries take. On
+a four node microceph cluster the cluster stream ticks about every 1.4s at the
+default `refresh_secs = 1`. The node table shows connection state (`live`,
+`dial`, `retry`, `error`), OSD ids, CPU percentage, and memory percentage.
 When `trace_auto_start` is true, cephlens starts osdtrace runners as soon as the
 TUI opens. The default config keeps it false so an operator explicitly starts
 and stops tracing with `t`, `f`, `r`, or `a`.

src/app.rs

@@ -19,6 +19,7 @@ use serde_json::Value;
 use crate::{
     collect::{parse_cluster_summary, parse_osds, run_probe},
     editor::ConfigEditor,
+    flow::{FlowMetric, FlowSort},
     kfstrace::{KfsEvent, parse_kfs_event},
     model::{NodeSummary, Snapshot},
     radostrace::{RadosEvent, parse_rados_event},
@@ -177,6 +178,10 @@ pub(crate) struct App {
     pub(crate) trace_following: bool,
     pub(crate) trace_session: Option<String>,
     pub(crate) trace_source: TraceSource,
+    pub(crate) flow_view: bool,
+    pub(crate) flow_metric: FlowMetric,
+    pub(crate) flow_sort: FlowSort,
+    pub(crate) flow_scroll: usize,
     pub(crate) kfstrace_events: Vec<KfsEvent>,
     pub(crate) kfstrace_active: usize,
     pub(crate) kfstrace_stop: Arc<AtomicBool>,
@@ -605,14 +610,15 @@ fn handle_stream_payload(app: &mut App, id: &str, payload: &str) -> Result<()> {
         let df = value
             .pointer("/df")
             .ok_or_else(|| anyhow!("cluster stream missing df"))?;
+        let perf = value.pointer("/perf").filter(|perf| !perf.is_null());
         let snapshot = Snapshot {
             captured_at: Utc::now(),
             profile: app.profile.clone(),
             admin_host: app.admin_host.clone(),
             hosts: app.hosts.clone(),
             cluster: parse_cluster_summary(status),
             nodes: ordered_nodes(app),
-            osds: parse_osds(tree, df),
+            osds: parse_osds(tree, df, perf),
         };
         record_session_snapshot(app, &snapshot);
         app.snapshot = Some(snapshot);

src/collect.rs

@@ -6,27 +6,35 @@ use serde_json::Value;
 
 use crate::{
     config::ResolvedConfig,
-    model::{ClusterSummary, NodeSummary, OsdSummary, Snapshot},
+    model::{ClusterSummary, HealthCheck, NodeSummary, OsdSummary, Snapshot},
     ssh::ssh_capture,
     stream::NODE_FACTS_SNIPPET,
-    util::{ptr_f64, ptr_i64, ptr_str, ptr_u64, shell_quote},
+    util::{MAX_PARALLEL_HOSTS, map_parallel, ptr_f64, ptr_i64, ptr_str, ptr_u64, shell_quote},
 };
 
 pub(crate) fn collect_snapshot(cfg: &ResolvedConfig) -> Result<Snapshot> {
     let status_out = ssh_capture(&cfg.admin_host, "sudo -n ceph -s --format json")?;
     let tree_out = ssh_capture(&cfg.admin_host, "sudo -n ceph osd tree --format json")?;
     let df_out = ssh_capture(&cfg.admin_host, "sudo -n ceph osd df --format json")?;
+    // Latency is supplementary, so a cluster that refuses this query still gets
+    // a snapshot.
+    let perf_out = ssh_capture(&cfg.admin_host, "sudo -n ceph osd perf --format json").ok();
 
     let status: Value = serde_json::from_str(status_out.trim())
         .with_context(|| "failed to parse ceph status json")?;
     let tree: Value =
         serde_json::from_str(tree_out.trim()).with_context(|| "failed to parse osd tree json")?;
     let df: Value =
         serde_json::from_str(df_out.trim()).with_context(|| "failed to parse osd df json")?;
+    let perf = perf_out.and_then(|perf| serde_json::from_str::<Value>(perf.trim()).ok());
 
     let cluster = parse_cluster_summary(&status);
-    let osds = parse_osds(&tree, &df);
-    let nodes = cfg.hosts.iter().map(|host| collect_node(host)).collect();
+    let osds = parse_osds(&tree, &df, perf.as_ref());
+    let nodes = map_parallel(&cfg.hosts, MAX_PARALLEL_HOSTS, |host| collect_node(host))
+        .into_iter()
+        .zip(&cfg.hosts)
+        .map(|(node, host)| node.unwrap_or_else(|| node_worker_panicked(host)))
+        .collect();
 
     Ok(Snapshot {
         captured_at: Utc::now(),
@@ -91,10 +99,50 @@ pub(crate) fn parse_cluster_summary(status: &Value) -> ClusterSummary {
         read_ops_sec: ptr_u64(status, "/pgmap/read_op_per_sec"),
         write_ops_sec: ptr_u64(status, "/pgmap/write_op_per_sec"),
         pg_states,
+        health_checks: parse_health_checks(status),
     }
 }
 
-pub(crate) fn parse_osds(tree: &Value, df: &Value) -> Vec<OsdSummary> {
+fn parse_health_checks(status: &Value) -> Vec<HealthCheck> {
+    let Some(checks) = status.pointer("/health/checks").and_then(Value::as_object) else {
+        return Vec::new();
+    };
+    let mut parsed = checks
+        .iter()
+        .map(|(code, check)| HealthCheck {
+            code: code.clone(),
+            severity: ptr_str(check, "/severity"),
+            message: ptr_str(check, "/summary/message"),
+        })
+        .collect::<Vec<_>>();
+    parsed.sort_by(|left, right| left.code.cmp(&right.code));
+    parsed
+}
+
+/// Maps OSD id to the latency `ceph osd perf` reports. The payload is optional,
+/// so a cluster where the query fails keeps its status, tree, and df data.
+pub(crate) fn parse_osd_perf(perf: Option<&Value>) -> HashMap<i64, (u64, u64)> {
+    let mut by_osd = HashMap::new();
+    let Some(infos) = perf
+        .and_then(|perf| perf.pointer("/osdstats/osd_perf_infos"))
+        .and_then(Value::as_array)
+    else {
+        return by_osd;
+    };
+    for info in infos {
+        by_osd.insert(
+            ptr_i64(info, "/id"),
+            (
+                ptr_u64(info, "/perf_stats/commit_latency_ms"),
+                ptr_u64(info, "/perf_stats/apply_latency_ms"),
+            ),
+        );
+    }
+    by_osd
+}
+
+pub(crate) fn parse_osds(tree: &Value, df: &Value, perf: Option<&Value>) -> Vec<OsdSummary> {
+    let latency_by_osd = parse_osd_perf(perf);
     let mut host_by_osd = HashMap::new();
     let mut status_by_osd = HashMap::new();
 
@@ -120,6 +168,8 @@ pub(crate) fn parse_osds(tree: &Value, df: &Value) -> Vec<OsdSummary> {
     if let Some(nodes) = df.pointer("/nodes").and_then(Value::as_array) {
         for node in nodes {
             let id = ptr_i64(node, "/id");
+            let (commit_latency_ms, apply_latency_ms) =
+                latency_by_osd.get(&id).copied().unwrap_or_default();
             osds.push(OsdSummary {
                 id,
                 name: ptr_str(node, "/name"),
@@ -133,6 +183,8 @@ pub(crate) fn parse_osds(tree: &Value, df: &Value) -> Vec<OsdSummary> {
                 pgs: ptr_u64(node, "/pgs"),
                 used_kb: ptr_u64(node, "/kb_used"),
                 avail_kb: ptr_u64(node, "/kb_avail"),
+                commit_latency_ms,
+                apply_latency_ms,
             });
         }
     }
@@ -164,6 +216,8 @@ printf 'ceph_osd_processes=%s\n' "$count"
 printf 'osd_ids=%s\n' "$ids"
 printf 'cpu_percent=%s\n' "$cpu_pct"
 printf 'mem_percent=%s\n' "$mem_pct"
+printf 'io_stall_percent=%s\n' "$io_stall"
+printf 'cpu_stall_percent=%s\n' "$cpu_stall"
 "#,
         facts = NODE_FACTS_SNIPPET
     );
@@ -189,6 +243,14 @@ printf 'mem_percent=%s\n' "$mem_pct"
                     .get("mem_percent")
                     .and_then(|s| s.parse().ok())
                     .unwrap_or_default(),
+                io_stall_percent: map
+                    .get("io_stall_percent")
+                    .and_then(|s| s.parse().ok())
+                    .unwrap_or_default(),
+                cpu_stall_percent: map
+                    .get("cpu_stall_percent")
+                    .and_then(|s| s.parse().ok())
+                    .unwrap_or_default(),
                 error: None,
             }
         }
@@ -200,6 +262,14 @@ printf 'mem_percent=%s\n' "$mem_pct"
     }
 }
 
+fn node_worker_panicked(host: &str) -> NodeSummary {
+    NodeSummary {
+        host: host.to_owned(),
+        error: Some("node collection worker panicked".to_owned()),
+        ..NodeSummary::default()
+    }
+}
+
 pub(crate) fn run_bench(
     host: &str,
     seconds: u64,
@@ -311,6 +381,59 @@ fn parse_key_values(output: &str) -> HashMap<String, String> {
 mod tests {
     use super::*;
 
+    // Shapes taken from ceph 19.2.3 on a microceph cluster.
+    #[test]
+    fn cluster_summary_names_the_failing_health_checks() {
+        let status: Value = serde_json::from_str(
+            r#"{"health":{"status":"HEALTH_WARN","checks":{
+                 "OSD_NEARFULL":{"severity":"HEALTH_WARN","summary":{"message":"1 nearfull osd(s)"}},
+                 "MON_CLOCK_SKEW":{"severity":"HEALTH_WARN","summary":{"message":"clock skew detected"}}}}}"#,
+        )
+        .unwrap();
+
+        let cluster = parse_cluster_summary(&status);
+
+        let codes = cluster
+            .health_checks
+            .iter()
+            .map(|check| check.code.as_str())
+            .collect::<Vec<_>>();
+        assert_eq!(codes, vec!["MON_CLOCK_SKEW", "OSD_NEARFULL"]);
+        assert_eq!(cluster.health_checks[1].message, "1 nearfull osd(s)");
+    }
+
+    #[test]
+    fn healthy_cluster_reports_no_checks() {
+        let status: Value =
+            serde_json::from_str(r#"{"health":{"status":"HEALTH_OK","checks":{},"mutes":[]}}"#)
+                .unwrap();
+
+        assert!(parse_cluster_summary(&status).health_checks.is_empty());
+    }
+
+    #[test]
+    fn osd_latency_is_merged_by_id_and_optional() {
+        let tree: Value = serde_json::from_str(r#"{"nodes":[]}"#).unwrap();
+        let df: Value =
+            serde_json::from_str(r#"{"nodes":[{"id":1,"name":"osd.1"},{"id":2,"name":"osd.2"}]}"#)
+                .unwrap();
+        let perf: Value = serde_json::from_str(
+            r#"{"osdstats":{"osd_perf_infos":[
+                 {"id":2,"perf_stats":{"commit_latency_ms":7,"apply_latency_ms":3}}]}}"#,
+        )
+        .unwrap();
+
+        let merged = parse_osds(&tree, &df, Some(&perf));
+        assert_eq!(merged[0].commit_latency_ms, 0, "osd.1 has no perf entry");
+        assert_eq!(merged[1].commit_latency_ms, 7);
+        assert_eq!(merged[1].apply_latency_ms, 3);
+
+        // A cluster that refuses `ceph osd perf` still gets its OSD rows.
+        let without = parse_osds(&tree, &df, None);
+        assert_eq!(without.len(), 2);
+        assert_eq!(without[1].commit_latency_ms, 0);
+    }
+
     #[test]
     fn bench_command_cleans_up_unique_pool_on_exit() {
         let pool = bench_pool_name("20260716-120000", 42);