From d67014c545e9b0785f9e1463ae49c7ddc3ecef70 Mon Sep 17 00:00:00 2001 From: gspark Date: Mon, 19 Oct 2026 09:12:35 +0900 Subject: [PATCH 1/2] feat(airm): read ai-rm v0.1 AcceleratorProfile, airm_* metrics and placement Infra > Devices gains a read-only ai-rm panel, shown only when the cluster has an ai-rm source: - AcceleratorProfile (airm.ai-computing.io/v1alpha1): vendor, model, memory, peak TFLOPS with its source; the selected device is marked. - Prometheus airm_* series with rm-core v0.1 names/units; vendor-derived tok_s/req_s/power_w/accelerators shown as [vendor] when absent. The device PWR column prefers airm_power_w{scope="card"}. - ConfigMap airm-placement (LMD_AIRM_NS, default llm-d): decision + reason. Without the CRD, series and ConfigMap nothing changes and nothing warns. ai-rm.yaml declares protocol "0.1", metrics: []. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 3 + README.md | 3 + ai-rm.yaml | 3 + cluster.example.env | 3 + src/airm.rs | 585 ++++++++++++++++++++++++++++++++++++++++++++ src/bootstrap.rs | 1 + src/collect.rs | 37 +++ src/config.rs | 2 + src/main.rs | 2 + src/ui/mod.rs | 141 ++++++++++- src/ui/tests.rs | 32 +++ 11 files changed, 811 insertions(+), 1 deletion(-) create mode 100644 src/airm.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 34b58db..8743415 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,9 @@ ## [Unreleased] +### ai-rm protocol v0.1 (read-only, 2026-10-19) +- Infra ▸ Devices shows an ai-rm panel when the cluster has any ai-rm source: `AcceleratorProfile` (`airm.ai-computing.io/v1alpha1`) vendor·model·memory·peak with its source (the selected device's profile is marked), `airm_*` Prometheus series with rm-core v0.1 names and units (vendor-derived `tok_s`/`req_s`/`power_w`/`accelerators` shown as `[vendor]` when no airm series exists), and the decisions in ConfigMap `airm-placement` (`LMD_AIRM_NS`, default `llm-d`). The device PWR column prefers `airm_power_w{scope="card"}`. Without the CRD, series and ConfigMap the view is unchanged and nothing is warned. Write actions are unchanged. + ### Furiosa SDK 2026.4 readiness (2026-10-16) - The default furiosa-llm image for compile, serve and training Jobs is pinned to `furiosaai/furiosa-llm:2026.3.0` (`accel::furiosa::DEFAULT_IMAGE`) instead of `:latest`. 2026.4 removed `serve --tensor-parallel-size` and legacy artifacts, so a floating tag would break existing manifests on their next restart. - Furiosa serve manifests pass `--tensor-parallel-size` only to images before 2026.4; for 2026.4+ (and unversioned tags, which will become 2026.4) TP comes from the FXB artifact. Measured on one RNGD with Qwen3-4B-FP8: 2026.4 serves +27–29% tok/s over 2026.3 (106→137 c1, 1146→1459 c16), so moving to it is a matter of setting `LMD_COMPILE_IMAGE_FURIOSA` once a 2026.4 image is published. diff --git a/README.md b/README.md index 5f55e4d..929f4c2 100644 --- a/README.md +++ b/README.md @@ -236,6 +236,9 @@ python3 -m unittest discover -s tests -p 'test_*.py' # 설치 스크립트 | `LMD_PROM` | Prometheus `host:port` (평문 HTTP, NodePort 또는 port-forward) | | `LMD_NS` | 모델 서버·EPP가 있는 네임스페이스 (기본 `llm-serving`) | | `LMD_GRAFANA` | Grafana 기본 URL (`:graf` 명령) | +| `LMD_AIRM_NS` | ai-rm ConfigMap `airm-placement`의 네임스페이스 (기본 `llm-d`) | + +**ai-rm protocol v0.1 (읽기 전용).** Infra ▸ Devices 아래에 ai-rm 패널이 나온다: `AcceleratorProfile`(`airm.ai-computing.io/v1alpha1`)의 벤더·모델·메모리·peak(출처), Prometheus `airm_*` 지표(rm-core `metrics.py`의 이름·단위, 없으면 기존 vendor 값을 `[vendor]`로 표시), ConfigMap `airm-placement`의 배치 결정과 이유. 장치별 PWR 열은 `airm_power_w{scope="card"}`가 있으면 그 값을 쓴다. CRD·지표·ConfigMap이 모두 없으면 패널은 나오지 않는다. `airm-placement`의 각 data 키는 워크로드 이름, 값은 `{"accelerator_ids": [...], "kv_transport": "...", "reason": "..."}`(JSON 또는 YAML)이다. ## Contributing & license diff --git a/ai-rm.yaml b/ai-rm.yaml index a64e5ab..897ea0b 100644 --- a/ai-rm.yaml +++ b/ai-rm.yaml @@ -1,4 +1,5 @@ name: lmd-top +protocol: "0.1" layer: l5-ops kind: tool summary: llm-d 클러스터의 가속기·서빙·EPP 라우팅·SLO를 한 화면에서 보고 조정하는 터미널 관측·운영 콘솔(ai-rm 관측 콘솔) @@ -19,3 +20,5 @@ interfaces: provides: [Metrics, Placement] # --json 상태·KPI 관측, heterogeneous-capacity EPP 정책(contrib/epp-score-observer) uses: [Metrics, Accelerator, Workload, Placement] # Prometheus 지표, 가속기 팩·device plugin 리소스, serving/학습 Job, InferencePool/EPP kpi: [이종자원수, SLO, 전력효율] +adapters: [] # Rust; reads rm-core formats (AcceleratorProfile, airm_* series, airm-placement) without importing rm-core +metrics: [] # reads airm_* only, reports none diff --git a/cluster.example.env b/cluster.example.env index e7b8ad0..b24e92c 100644 --- a/cluster.example.env +++ b/cluster.example.env @@ -12,3 +12,6 @@ LMD_NS=llm-serving # Grafana base URL used for dashboard links LMD_GRAFANA=http://: + +# Namespace of the ai-rm ConfigMap airm-placement (read-only placement decisions) +LMD_AIRM_NS=llm-d diff --git a/src/airm.rs b/src/airm.rs new file mode 100644 index 0000000..6b92634 --- /dev/null +++ b/src/airm.rs @@ -0,0 +1,585 @@ +//! ai-rm protocol v0.1 — read-only consumer of the formats rm-core defines. +//! +//! lmd-top is Rust, so it does not import rm-core; it reads what rm-core writes: +//! - `AcceleratorProfile` (`airm.ai-computing.io/v1alpha1`, cluster-scoped) — one per device, +//! published by the node daemon (`rm_core.k8s.accelerator_profile`). +//! - Prometheus series `airm_` for the names in `rm_core.metrics.METRICS`. +//! - ConfigMap `airm-placement` — placement decisions (`rm_core.workload.PlacementDecision`). +//! +//! Every source is optional. A cluster without the CRD, the series or the ConfigMap shows exactly +//! what it showed before: absence is `None`/empty, never a warning. + +use crate::collect::{Accel, AccelKind}; +use crate::prom::Series; +use std::collections::BTreeMap; + +pub const PROFILE_RESOURCE: &str = "acceleratorprofiles.airm.ai-computing.io"; +pub const PLACEMENT_CONFIGMAP: &str = "airm-placement"; +pub const PROM_PREFIX: &str = "airm_"; + +/// `rm_core.metrics.METRICS` (v0.1): name → unit. Display uses exactly these units. +pub const METRICS: &[(&str, &str)] = &[ + ("ttft_p50_ms", "ms"), + ("ttft_p90_ms", "ms"), + ("ttft_p99_ms", "ms"), + ("tpot_p50_ms", "ms"), + ("tpot_p90_ms", "ms"), + ("tpot_p99_ms", "ms"), + ("e2e_p90_ms", "ms"), + ("tok_s", "tokens/s"), + ("req_s", "requests/s"), + ("power_w", "W"), + ("energy_j", "J"), + ("gflops_per_w", "GFLOPS/W"), + ("mfu_pct", "%"), + ("kv_transfer_ms", "ms"), + ("kv_transfer_gbps", "GB/s"), + ("accuracy_pct", "%"), + ("slo_attainment_pct", "%"), + ("accelerators", "count"), +]; + +/// `rm_core.metrics.TAG_KEYS` — the only labels shown next to a value. +pub const TAG_KEYS: &[&str] = &[ + "component", + "model", + "accelerator", + "vendor", + "scope", + "peak_source", + "dataset", + "workload", + "kv_transport", + "run_id", + "commit", +]; + +pub fn unit(name: &str) -> Option<&'static str> { + METRICS.iter().find(|(n, _)| *n == name).map(|(_, u)| *u) +} + +/// One `AcceleratorProfile` (spec only). Unknown values stay unknown: `peak_tflops` empty, +/// `power_limit_w` None. +#[derive(Clone, Debug, PartialEq)] +pub struct Profile { + pub id: String, // "/:" + pub node: String, + pub vendor: String, + pub model: String, + pub memory_bytes: u64, + pub memory_kind: String, + pub peak_tflops: Vec<(String, f64)>, // precision → dense TFLOPS + pub peak_source: String, + pub power_limit_w: Option, + pub interconnect: Vec, +} + +impl Profile { + /// Device index from the id (`node/vendor:3` → 3). + pub fn index(&self) -> Option { + self.id.rsplit_once(':')?.1.parse().ok() + } + + pub fn memory_gb(&self) -> f64 { + self.memory_bytes as f64 / 1e9 + } + + /// "bf16 512 · fp8 1024 TF (datasheet)" — empty when no peak is published. + pub fn peak_label(&self) -> String { + if self.peak_tflops.is_empty() { + return String::new(); + } + let peaks: Vec = self + .peak_tflops + .iter() + .map(|(p, v)| format!("{p} {}", trim_num(*v))) + .collect(); + let src = if self.peak_source.is_empty() { + "source ?".to_string() + } else { + self.peak_source.clone() + }; + format!("{} TF ({src})", peaks.join(" · ")) + } + + /// Does this profile describe that collected device? Matched on node, vendor family and + /// device index — the profile id is `/:`, lmd-top ids are `gpu0`, + /// `rbln0`, `npu0`. + pub fn matches(&self, a: &Accel) -> bool { + let family = match a.kind { + AccelKind::Gpu => "nvidia", + AccelKind::Rbln => "rebellions", + AccelKind::Rngd => "furiosa", + AccelKind::Other(_) => return false, + }; + let digits: String = + a.id.chars() + .rev() + .take_while(char::is_ascii_digit) + .collect(); + let index = digits.chars().rev().collect::().parse::().ok(); + self.vendor == family && self.node == a.node && index.is_some() && self.index() == index + } +} + +fn trim_num(v: f64) -> String { + if v.fract() == 0.0 { + format!("{v:.0}") + } else { + format!("{v}") + } +} + +/// Number written as a string (rm-core keeps exact values as strings) or as a JSON number. +fn num(v: &serde_json::Value) -> Option { + v.as_f64() + .or_else(|| v.as_str().and_then(|s| s.trim().parse().ok())) +} + +/// Parse one AcceleratorProfile object. Objects missing a required field are skipped, as +/// `rm_core.k8s.from_profile` would reject them. +pub fn parse_profile(obj: &serde_json::Value) -> Option { + let s = &obj["spec"]; + let text = |k: &str| s[k].as_str().map(str::to_string); + let memory_bytes = num(&s["memoryBytes"]).filter(|v| *v > 0.0)? as u64; + let mut peak_tflops: Vec<(String, f64)> = s["peakTflops"] + .as_object() + .map(|m| { + m.iter() + .filter_map(|(k, v)| num(v).filter(|v| *v > 0.0).map(|v| (k.clone(), v))) + .collect() + }) + .unwrap_or_default(); + peak_tflops.sort_by(|a, b| a.0.cmp(&b.0)); + Some(Profile { + id: text("id")?, + node: text("node")?, + vendor: text("vendor")?, + model: text("model")?, + memory_bytes, + memory_kind: text("memoryKind")?, + peak_tflops, + peak_source: text("peakSource").unwrap_or_default(), + power_limit_w: num(&s["powerLimitW"]), + interconnect: s["interconnect"] + .as_array() + .map(|a| { + a.iter() + .filter_map(|x| x.as_str().map(String::from)) + .collect() + }) + .unwrap_or_default(), + }) +} + +/// `kubectl get acceleratorprofiles -o json` → profiles, sorted by id. +pub fn parse_profiles(list: &serde_json::Value) -> Vec { + let mut out: Vec = list["items"] + .as_array() + .map(|items| items.iter().filter_map(parse_profile).collect()) + .unwrap_or_default(); + out.sort_by(|a, b| a.id.cmp(&b.id)); + out +} + +/// One placement decision: which accelerators a workload got, the KV path, and why. +#[derive(Clone, Debug, PartialEq)] +pub struct Decision { + pub workload: String, + pub accelerator_ids: Vec, + pub kv_transport: String, + pub reason: String, +} + +/// Parse ConfigMap `airm-placement`. Each data key is a workload; its value is a JSON (or YAML) +/// `PlacementDecision` — `{accelerator_ids, kv_transport, reason}` (camelCase accepted too). +/// An empty ConfigMap (the v0.1 llm-d-setup default) yields no decisions. A value that does +/// not parse is kept with the parse failure as its reason, so a bad write stays visible. +pub fn parse_placement(cm: &serde_json::Value) -> Vec { + let Some(data) = cm["data"].as_object() else { + return Vec::new(); + }; + let mut out: Vec = data + .iter() + .map(|(workload, raw)| { + let raw = raw.as_str().unwrap_or_default(); + let parsed: Option = serde_json::from_str(raw) + .ok() + .or_else(|| serde_yaml::from_str(raw).ok()) + .filter(serde_json::Value::is_object); + let Some(v) = parsed else { + return Decision { + workload: workload.clone(), + accelerator_ids: Vec::new(), + kv_transport: String::new(), + reason: "unparsable decision".into(), + }; + }; + let field = |snake: &str, camel: &str| { + v.get(snake) + .or_else(|| v.get(camel)) + .cloned() + .unwrap_or_default() + }; + Decision { + workload: workload.clone(), + accelerator_ids: field("accelerator_ids", "acceleratorIds") + .as_array() + .map(|a| { + a.iter() + .filter_map(|x| x.as_str().map(String::from)) + .collect() + }) + .unwrap_or_default(), + kv_transport: field("kv_transport", "kvTransport") + .as_str() + .unwrap_or_default() + .to_string(), + reason: field("reason", "reason") + .as_str() + .unwrap_or_default() + .to_string(), + } + }) + .collect(); + out.sort_by(|a, b| a.workload.cmp(&b.workload)); + out +} + +/// One `airm_*` sample, with only protocol tags kept. +#[derive(Clone, Debug, PartialEq)] +pub struct Sample { + pub name: String, // protocol name, without the airm_ prefix + pub value: f64, + pub tags: BTreeMap, +} + +impl Sample { + pub fn tag_label(&self) -> String { + self.tags + .iter() + .map(|(k, v)| format!("{k}={v}")) + .collect::>() + .join(",") + } +} + +/// Result of `{__name__=~"airm_.+"}` → protocol samples. Series whose name is not in the v0.1 +/// table are dropped (PROTOCOL §6: a metric exists only once rm-core lists it). +pub fn parse_metrics(series: Vec) -> Vec { + let mut out: Vec = series + .into_iter() + .filter(|s| s.value.is_finite()) + .filter_map(|s| { + let name = s.l("__name__").strip_prefix(PROM_PREFIX)?.to_string(); + unit(&name)?; + let tags = s + .labels + .iter() + .filter(|(k, _)| TAG_KEYS.contains(&k.as_str())) + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + Some(Sample { + name, + value: s.value, + tags, + }) + }) + .collect(); + out.sort_by(|a, b| { + a.name + .cmp(&b.name) + .then_with(|| a.tag_label().cmp(&b.tag_label())) + }); + out +} + +/// Everything lmd-top reads from ai-rm. `None` = source absent (no CRD / no ConfigMap). +#[derive(Clone, Default)] +pub struct Airm { + pub profiles: Option>, + pub metrics: Vec, + pub placement: Option>, +} + +impl Airm { + /// Is any ai-rm source present? When false the UI shows nothing ai-rm related. + pub fn present(&self) -> bool { + self.profiles.is_some() || !self.metrics.is_empty() || self.placement.is_some() + } + + pub fn profile_for(&self, a: &Accel) -> Option<&Profile> { + self.profiles.as_ref()?.iter().find(|p| p.matches(a)) + } + + /// Card power for a device from `airm_power_w{scope="card",accelerator=}`. + pub fn card_power(&self, a: &Accel) -> Option { + let id = &self.profile_for(a)?.id; + self.metrics + .iter() + .find(|s| { + s.name == "power_w" + && s.tags.get("scope").map(String::as_str) == Some("card") + && s.tags.get("accelerator") == Some(id) + }) + .map(|s| s.value) + } +} + +/// A row in the ai-rm metric panel: the airm value when reported, otherwise the vendor-derived +/// value lmd-top already collects (marked as such), otherwise absent. +#[derive(Clone, Debug, PartialEq)] +pub struct MetricRow { + pub name: &'static str, + pub unit: &'static str, + pub value: f64, + pub tags: String, + pub source: &'static str, // "airm" | "vendor" +} + +/// Rows for the panel: every reported airm sample, then a vendor fallback for each of +/// `tok_s`, `req_s`, `power_w`, `accelerators` that has no airm sample. Vendor latency +/// percentiles (p95) are not mapped — the protocol has p50/p90/p99 only, and relabelling a +/// p95 as one of those would misstate it. +pub fn metric_rows(airm: &Airm, snap: &crate::collect::Snapshot) -> Vec { + let mut rows: Vec = airm + .metrics + .iter() + .filter_map(|s| { + let (name, unit) = METRICS.iter().find(|(n, _)| *n == s.name)?; + Some(MetricRow { + name, + unit, + value: s.value, + tags: s.tag_label(), + source: "airm", + }) + }) + .collect(); + let reported = |n: &str| airm.metrics.iter().any(|s| s.name == n); + let mut fallback = |name: &'static str, value: f64, tags: String| { + if !reported(name) && value.is_finite() { + rows.push(MetricRow { + name, + unit: unit(name).unwrap_or(""), + value, + tags, + source: "vendor", + }); + } + }; + fallback("tok_s", snap.perf.tps, String::new()); + fallback("req_s", snap.perf.req_rate, String::new()); + let powers: Vec = snap + .accel + .iter() + .map(|a| a.power) + .filter(|p| p.is_finite()) + .collect(); + if !powers.is_empty() { + fallback("power_w", powers.iter().sum(), "scope=card(sum)".into()); + } + let mut counts: BTreeMap = BTreeMap::new(); + for a in &snap.accel { + *counts.entry(a.disp().to_string()).or_default() += 1; + } + for (model, n) in counts { + fallback("accelerators", n as f64, format!("model={model}")); + } + rows +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn profile_obj(id: &str, node: &str, vendor: &str) -> serde_json::Value { + // Shape of rm_core.k8s.accelerator_profile(); numbers are strings as rm-core writes them. + json!({ + "apiVersion": "airm.ai-computing.io/v1alpha1", + "kind": "AcceleratorProfile", + "metadata": {"name": id.replace('/', ".").replace(':', "-")}, + "spec": { + "id": id, "node": node, "vendor": vendor, "model": "RNGD", + "memoryBytes": 51539607552u64, "memoryKind": "hbm", + "peakTflops": {"bf16": "256", "fp8": "512.5"}, "peakSource": "datasheet", + "powerLimitW": "180", "interconnect": ["pcie5"] + } + }) + } + + fn accel(kind: AccelKind, id: &str, node: &str) -> Accel { + Accel { + kind, + model: String::new(), + id: id.into(), + node: node.into(), + util: 0.0, + mem_used_gb: f64::NAN, + mem_total_gb: f64::NAN, + temp: f64::NAN, + power: 42.0, + busy_model: String::new(), + alive: true, + throttle: f64::NAN, + unified_mem: false, + mem_bw: f64::NAN, + clock_mhz: f64::NAN, + mem_temp: f64::NAN, + energy_mj: f64::NAN, + } + } + + fn series(name: &str, value: f64, labels: &[(&str, &str)]) -> Series { + let mut l: BTreeMap = labels + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + l.insert("__name__".into(), name.into()); + Series { labels: l, value } + } + + #[test] + fn parses_profile_written_by_rm_core() { + let p = parse_profile(&profile_obj("node-a/furiosa:1", "node-a", "furiosa")).unwrap(); + assert_eq!(p.id, "node-a/furiosa:1"); + assert_eq!(p.index(), Some(1)); + assert_eq!(p.memory_bytes, 51539607552); + assert_eq!(p.power_limit_w, Some(180.0)); + assert_eq!( + p.peak_tflops, + vec![("bf16".into(), 256.0), ("fp8".into(), 512.5)] + ); + assert_eq!(p.peak_label(), "bf16 256 · fp8 512.5 TF (datasheet)"); + assert_eq!(p.interconnect, vec!["pcie5".to_string()]); + } + + #[test] + fn unknown_peak_and_power_stay_unknown() { + let mut obj = profile_obj("node-a/cpu:0", "node-a", "cpu"); + obj["spec"]["peakTflops"] = json!({}); + obj["spec"]["powerLimitW"] = serde_json::Value::Null; + let p = parse_profile(&obj).unwrap(); + assert!(p.peak_tflops.is_empty()); + assert_eq!(p.peak_label(), ""); + assert_eq!(p.power_limit_w, None); + } + + #[test] + fn profile_missing_required_field_is_skipped() { + let mut bad = profile_obj("node-a/furiosa:0", "node-a", "furiosa"); + bad["spec"].as_object_mut().unwrap().remove("memoryKind"); + let list = json!({"items": [bad, profile_obj("node-a/furiosa:0", "node-a", "furiosa")]}); + assert_eq!(parse_profiles(&list).len(), 1); + assert!(parse_profiles(&json!({"items": []})).is_empty()); + } + + #[test] + fn profile_matches_collected_device_by_node_vendor_index() { + let p = parse_profile(&profile_obj("node-a/furiosa:1", "node-a", "furiosa")).unwrap(); + assert!(p.matches(&accel(AccelKind::Rngd, "npu1", "node-a"))); + assert!(!p.matches(&accel(AccelKind::Rngd, "npu0", "node-a"))); + assert!(!p.matches(&accel(AccelKind::Rngd, "npu1", "node-b"))); + assert!(!p.matches(&accel(AccelKind::Rbln, "rbln1", "node-a"))); + } + + #[test] + fn airm_series_keep_protocol_names_units_and_tags() { + let got = parse_metrics(vec![ + series( + "airm_power_w", + 150.0, + &[ + ("scope", "card"), + ("accelerator", "node-a/furiosa:1"), + ("instance", "192.0.2.10:9400"), + ], + ), + series("airm_tok_s", 61.7, &[("model", "m")]), + series("airm_not_in_protocol", 1.0, &[]), + series("airm_ttft_p90_ms", f64::NAN, &[]), + ]); + assert_eq!(got.len(), 2); + assert_eq!(got[0].name, "power_w"); + assert_eq!(unit(&got[0].name), Some("W")); + // non-protocol labels (scrape address) are dropped + assert_eq!( + got[0].tag_label(), + "accelerator=node-a/furiosa:1,scope=card" + ); + assert_eq!(unit(&got[1].name), Some("tokens/s")); + } + + #[test] + fn card_power_prefers_airm_series_for_matching_device() { + let airm = Airm { + profiles: Some(vec![parse_profile(&profile_obj( + "node-a/furiosa:1", + "node-a", + "furiosa", + )) + .unwrap()]), + metrics: parse_metrics(vec![series( + "airm_power_w", + 150.0, + &[("scope", "card"), ("accelerator", "node-a/furiosa:1")], + )]), + placement: None, + }; + assert_eq!( + airm.card_power(&accel(AccelKind::Rngd, "npu1", "node-a")), + Some(150.0) + ); + assert_eq!( + airm.card_power(&accel(AccelKind::Rngd, "npu0", "node-a")), + None + ); + } + + #[test] + fn metric_rows_fall_back_to_vendor_values_only_when_airm_absent() { + let mut snap = crate::collect::Snapshot::default(); + snap.perf.tps = 66.8; + snap.accel = vec![accel(AccelKind::Rngd, "npu0", "node-a")]; + let airm = Airm { + metrics: parse_metrics(vec![series("airm_tok_s", 61.7, &[])]), + ..Airm::default() + }; + let rows = metric_rows(&airm, &snap); + let tok: Vec<_> = rows.iter().filter(|r| r.name == "tok_s").collect(); + assert_eq!(tok.len(), 1); + assert_eq!((tok[0].value, tok[0].source), (61.7, "airm")); + let power = rows.iter().find(|r| r.name == "power_w").unwrap(); + assert_eq!( + (power.value, power.source, power.unit), + (42.0, "vendor", "W") + ); + // req_rate is NaN in the snapshot → no row rather than a zero + assert!(rows.iter().all(|r| r.name != "req_s")); + } + + #[test] + fn placement_configmap_parses_decisions() { + let cm = json!({"data": { + "decode-llama": r#"{"accelerator_ids": ["node-a/furiosa:0", "node-a/furiosa:1"], "kv_transport": "p2p", "reason": "most free HBM"}"#, + "prefill-llama": "acceleratorIds: [node-b/nvidia:0]\nkvTransport: p2p\nreason: highest bf16 peak\n", + "broken": "{not json", + }}); + let d = parse_placement(&cm); + assert_eq!(d.len(), 3); + assert_eq!(d[0].workload, "broken"); + assert_eq!(d[0].reason, "unparsable decision"); + assert_eq!(d[1].accelerator_ids.len(), 2); + assert_eq!(d[1].kv_transport, "p2p"); + assert_eq!(d[1].reason, "most free HBM"); + assert_eq!(d[2].accelerator_ids, vec!["node-b/nvidia:0".to_string()]); + // the v0.1 default is an empty ConfigMap + assert!(parse_placement(&json!({"metadata": {"name": "airm-placement"}})).is_empty()); + } + + #[test] + fn nothing_present_by_default() { + assert!(!Airm::default().present()); + } +} diff --git a/src/bootstrap.rs b/src/bootstrap.rs index 28e8f5f..a45c694 100644 --- a/src/bootstrap.rs +++ b/src/bootstrap.rs @@ -109,6 +109,7 @@ mod tests { interval_full: 3, interval_fast: 1, theme: 3, + airm_ns: "llm-d".into(), } } #[test] diff --git a/src/collect.rs b/src/collect.rs index 8855516..cee6278 100644 --- a/src/collect.rs +++ b/src/collect.rs @@ -398,6 +398,7 @@ pub struct Snapshot { pub prom_ok: bool, // Prometheus 도달 가능 여부(false 면 "가속기 없음"이 아니라 연결 문제) pub setup: SetupProbe, // 플랫폼 부트스트랩 전제조건(Setup 뷰) pub pvcs: Vec, // 대상 ns 에 실존하는 PVC 이름들 — prefetch/compile 대상 검증에 사용(빈 벡터=미관측) + pub airm: crate::airm::Airm, // ai-rm v0.1 AcceleratorProfile·airm_* 지표·배치 결정(읽기 전용, 없으면 빈 값) } impl Snapshot { @@ -1598,6 +1599,7 @@ pub async fn collect(cfg: &Config) -> Snapshot { // ---------- kube: deployments / pods / routes / gateway / epp ---------- collect_kube(cfg, &mut snap, &vllm, &mut warn).await; + snap.airm = collect_airm(cfg, snap.prom_ok).await; // per-model perf: 런칭된 모든 모델(snap.models)을 seed 로 → 트래픽 없어도 표(–)에 표시. // Join deployment rows only through declared Service selector relationships. @@ -2278,6 +2280,41 @@ async fn collect_node_npu(nodes: &mut [NodeInfo]) { } /// k8s/llm-d 이벤트(최근 40) → snap.events. collect_kube 에서 분리된 leaf 파서. +/// ai-rm v0.1 sources (read-only). Each is optional: a missing CRD, ConfigMap or series is the +/// normal state of a cluster without ai-rm and produces no warning. +async fn collect_airm(cfg: &Config, prom_ok: bool) -> crate::airm::Airm { + use crate::airm; + let metrics_q = format!("{{__name__=~\"{}.+\"}}", airm::PROM_PREFIX); + let profile_args = ["get", airm::PROFILE_RESOURCE, "-o", "json"]; + let placement_args = [ + "get", + "configmap", + airm::PLACEMENT_CONFIGMAP, + "-n", + &cfg.airm_ns, + "--ignore-not-found", + "-o", + "json", + ]; + let (profiles, placement, metrics) = tokio::join!( + kube::get_json(&profile_args), + kube::get_json(&placement_args), + async { + if prom_ok { + prom::query(&cfg.prom, &metrics_q).await.ok() + } else { + None + } + }, + ); + airm::Airm { + profiles: profiles.ok().map(|v| airm::parse_profiles(&v)), + // --ignore-not-found prints nothing for an absent ConfigMap → JSON parse error → None. + placement: placement.ok().map(|v| airm::parse_placement(&v)), + metrics: metrics.map(airm::parse_metrics).unwrap_or_default(), + } +} + async fn collect_events(cfg: &Config, snap: &mut Snapshot, warn: &mut Vec) { let result = tracked_json( snap, diff --git a/src/config.rs b/src/config.rs index 8ed4eeb..82ddc2d 100644 --- a/src/config.rs +++ b/src/config.rs @@ -12,6 +12,7 @@ pub struct Config { pub interval_full: u64, // full collection interval (seconds) pub interval_fast: u64, // fast-tier (accelerator/node) interval (seconds) pub theme: usize, // startup theme: 0 default·1 high-contrast·2 colorblind·3 soft + pub airm_ns: String, // namespace of ai-rm ConfigMap `airm-placement` (read-only) } impl Default for Config { @@ -51,6 +52,7 @@ impl Default for Config { interval_full: u("LMD_INTERVAL", "interval_full", 3).max(1), interval_fast: u("LMD_FAST_INTERVAL", "interval_fast", 1).max(1), theme, + airm_ns: s("LMD_AIRM_NS", "airm_namespace", "llm-d"), } } } diff --git a/src/main.rs b/src/main.rs index e0ef0d3..5a1115c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,6 +5,7 @@ mod accel; mod advisor; mod agent; +mod airm; mod app; mod audit; mod bootstrap; @@ -142,6 +143,7 @@ ENVIRONMENT: LMD_PROM Prometheus host:port LMD_NS namespace (default: llm-serving) LMD_GRAFANA Grafana base URL (opened via the `:graf` palette command) + LMD_AIRM_NS namespace of the ai-rm ConfigMap airm-placement (default: llm-d) LMD_THEME startup theme: soft | default | high-contrast | colorblind LMD_AUDIT audit log path (default: ~/.config/lmd-top/audit.log) LMD_HISTORY compile/serving history path (default: ~/.config/lmd-top/history.jsonl) diff --git a/src/ui/mod.rs b/src/ui/mod.rs index 0bd40b3..ec91bd3 100644 --- a/src/ui/mod.rs +++ b/src/ui/mod.rs @@ -1109,6 +1109,17 @@ fn footer(f: &mut Frame, area: Rect, app: &App) { // ── Accel ────────────────────────────────────────────── fn view_accel(f: &mut Frame, area: Rect, app: &App) { + // ai-rm v0.1 panel only when the cluster has an ai-rm source; otherwise the view is unchanged. + let area = if app.snap.airm.present() { + let split = Layout::default() + .direction(Direction::Vertical) + .constraints([Constraint::Min(6), Constraint::Length(10)]) + .split(area); + airm_panel(f, split[1], app); + split[0] + } else { + area + }; let order = app.order(); let rows: Vec = order .iter() @@ -1167,7 +1178,14 @@ fn view_accel(f: &mut Frame, area: Rect, app: &App) { format!("{:.0}°C", Metric(a.temp)), Style::default().fg(temp_color(a.temp)), )), - cellw(format!("{:.0}W", Metric(a.power)), 5), + // airm_power_w{scope=card} for this device when reported, else the vendor reading. + cellw( + format!( + "{:.0}W", + Metric(app.snap.airm.card_power(a).unwrap_or(a.power)) + ), + 5, + ), Cell::from(Span::styled(trend, Style::default().fg(util_color(a.util)))), // 인라인 트렌드(all-smi식) Cell::from(Span::styled(model_cell, Style::default().fg(C_DIM()))), ]) @@ -1212,6 +1230,127 @@ fn view_accel(f: &mut Frame, area: Rect, app: &App) { ); } +/// ai-rm (read-only): AcceleratorProfiles, `airm_*` metrics (vendor fallback marked), placement. +fn airm_panel(f: &mut Frame, area: Rect, app: &App) { + let airm = &app.snap.airm; + let cols = Layout::default() + .direction(Direction::Horizontal) + .constraints([ + Constraint::Percentage(40), + Constraint::Percentage(32), + Constraint::Percentage(28), + ]) + .split(area); + let dim = Style::default().fg(C_DIM()); + let selected = app + .order() + .get(app.selected) + .and_then(|&i| app.snap.accel.get(i)) + .and_then(|a| airm.profile_for(a)) + .map(|p| p.id.clone()); + + let profiles: Vec = match &airm.profiles { + None => vec![Line::styled("AcceleratorProfile CRD not installed", dim)], + Some(ps) if ps.is_empty() => vec![Line::styled("no AcceleratorProfile published", dim)], + Some(ps) => ps + .iter() + .map(|p| { + let mark = if selected.as_deref() == Some(p.id.as_str()) { + "▶ " + } else { + " " + }; + let peak = p.peak_label(); + Line::from(vec![ + Span::raw(mark), + Span::styled(truncw(&p.id, 24), Style::default().fg(C_ACC())), + Span::raw(format!( + " {} {} {:.0}GB {}", + p.vendor, + p.model, + p.memory_gb(), + p.memory_kind + )), + Span::styled( + if peak.is_empty() { + " peak ?".to_string() + } else { + format!(" {peak}") + }, + dim, + ), + ]) + }) + .collect(), + }; + f.render_widget( + Paragraph::new(profiles).block(block("ai-rm · AcceleratorProfile")), + cols[0], + ); + + let rows = crate::airm::metric_rows(airm, &app.snap); + let metrics: Vec = if rows.is_empty() { + vec![Line::styled("no airm_* series", dim)] + } else { + rows.iter() + .map(|r| { + Line::from(vec![ + Span::raw(if r.unit == "count" { + format!("{} {:.0}", r.name, r.value) + } else { + format!("{} {:.1} {}", r.name, r.value, r.unit) + }), + Span::styled( + if r.tags.is_empty() { + format!(" [{}]", r.source) + } else { + format!(" {} [{}]", r.tags, r.source) + }, + dim, + ), + ]) + }) + .collect() + }; + f.render_widget( + Paragraph::new(metrics) + .wrap(Wrap { trim: true }) + .block(block("ai-rm · metrics")), + cols[1], + ); + + let decisions: Vec = match &airm.placement { + None => vec![Line::styled("no airm-placement ConfigMap", dim)], + Some(ds) if ds.is_empty() => vec![Line::styled("no placement decisions", dim)], + Some(ds) => ds + .iter() + .map(|d| { + let ids = if d.accelerator_ids.is_empty() { + "(none)".to_string() + } else { + d.accelerator_ids.join(",") + }; + let kv = if d.kv_transport.is_empty() { + String::new() + } else { + format!(" kv={}", d.kv_transport) + }; + Line::from(vec![ + Span::styled(d.workload.clone(), Style::default().fg(C_ACC())), + Span::raw(format!(" → {ids}{kv}")), + Span::styled(format!(" — {}", d.reason), dim), + ]) + }) + .collect(), + }; + f.render_widget( + Paragraph::new(decisions) + .wrap(Wrap { trim: true }) + .block(block("ai-rm · placement (read-only)")), + cols[2], + ); +} + // ── Models ───────────────────────────────────────────── const MODEL_COLS: [&str; 10] = [ "name", "engine", "accel", "ready", "run", "wait", "kv", "tps", "path", "status", diff --git a/src/ui/tests.rs b/src/ui/tests.rs index 2db8a2d..686dfa8 100644 --- a/src/ui/tests.rs +++ b/src/ui/tests.rs @@ -473,3 +473,35 @@ fn evicted_vendor_devices_are_listed_not_dropped() { "accelerator panel lists it:\n{text}" ); } + +#[test] +fn devices_view_shows_airm_panel_only_when_airm_present() { + let _guard = RENDER_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let mut app = missing_metrics(); + app.view = View::Accel; + let plain = render(&app, 160, 40); + assert!(!plain.contains("ai-rm"), "no ai-rm source → view unchanged"); + + let profiles = serde_json::json!({"items": [{"spec": { + "id": "node-a/nvidia:0", "node": "node-a", "vendor": "nvidia", "model": "H100", + "memoryBytes": 80000000000u64, "memoryKind": "hbm", + "peakTflops": {"bf16": "989"}, "peakSource": "datasheet" + }}]}); + app.snap.airm = crate::airm::Airm { + profiles: Some(crate::airm::parse_profiles(&profiles)), + metrics: Vec::new(), + placement: Some(crate::airm::parse_placement(&serde_json::json!({"data": { + "decode": r#"{"accelerator_ids": ["node-a/nvidia:0"], "reason": "only free device"}"# + }}))), + }; + let screen = render(&app, 160, 40); + assert!(screen.contains("ai-rm · AcceleratorProfile")); + assert!( + screen.contains("▶ node-a/nvidia:0"), + "selected device's profile is marked" + ); + assert!(screen.contains("bf16 989 TF (datasheet)")); + // no airm_* series → vendor-derived value, marked as such + assert!(screen.contains("accelerators 1 model=H100 [vendor]")); + assert!(screen.contains("only free device")); +} From bb688954a75a360e2c08ba29e49f9a0a16207a3c Mon Sep 17 00:00:00 2001 From: gspark-etri Date: Mon, 19 Oct 2026 11:19:24 +0900 Subject: [PATCH 2/2] feat(airm): prefer airm_tok_s for npu-lora training throughput npu-lora now reports training throughput as airm_tok_s{component="npu-lora",run_id=...} and keeps npu_lora_tokens_per_second as a deprecated alias until protocol 0.2. Read the shared name first and fall back to the old one. Co-Authored-By: Claude Opus 5.5 --- src/collect.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/collect.rs b/src/collect.rs index cee6278..f4e83c8 100644 --- a/src/collect.rs +++ b/src/collect.rs @@ -2562,6 +2562,12 @@ async fn collect_trainings(cfg: &Config, snap: &mut Snapshot, warn: &mut Vec) {