1use std::collections::HashMap;
5use std::io;
6use std::io::Write;
7use std::path::{Path, PathBuf};
8use std::time::{Duration, Instant};
9
10use anyhow::Context;
11use serde::Deserialize;
12
13use decdn_common::cli;
14use decdn_common::cli::ConfigPathSource;
15use decdn_common::cli::common::{default_config_path, expand_tilde};
16use decdn_common::config::DEFAULT_METRICS_PORT;
17
18pub(crate) fn parse_openmetrics(text: &str) -> HashMap<String, f64> {
33 let mut out = HashMap::new();
34 for line in text.lines() {
35 let trimmed = line.trim();
36 if trimmed.is_empty() || trimmed.starts_with('#') {
37 continue;
38 }
39 let Some((name, rest)) = trimmed.split_once(char::is_whitespace) else {
40 continue;
41 };
42 if name.contains('{') {
43 continue;
44 }
45 let value_token = rest.split_whitespace().next().unwrap_or("");
46 if let Ok(v) = value_token.parse::<f64>() {
47 out.insert(name.to_string(), v);
48 }
49 }
50 out
51}
52
53#[derive(Debug, Clone)]
57pub(crate) struct Snapshot {
58 pub uptime_seconds: u64,
59 pub active_connections: u64,
60 pub dispatch_in_flight: u64,
61 pub cache_hits: u64,
62 pub cache_misses: u64,
63 pub cache_bytes_returned: u64,
64 pub unredeemed_usdc: u64,
67 pub rpc_healthy: bool,
68}
69
70impl Snapshot {
71 pub(crate) fn from_metrics(m: &HashMap<String, f64>) -> Self {
72 #[allow(
79 clippy::cast_possible_truncation,
80 clippy::cast_sign_loss,
81 clippy::cast_precision_loss
82 )]
83 let u = |name: &str| -> u64 {
84 m.get(name).copied().map_or(0, |v| {
85 if v < 0.0 || !v.is_finite() {
86 0
87 } else if v >= u64::MAX as f64 {
88 u64::MAX
89 } else {
90 v as u64
91 }
92 })
93 };
94 Self {
95 uptime_seconds: u("decdn_node_uptime_seconds"),
96 active_connections: u("decdn_active_connections"),
97 dispatch_in_flight: u("decdn_dispatch_in_flight"),
98 cache_hits: u("decdn_cache_hits_total"),
99 cache_misses: u("decdn_cache_misses_total"),
100 cache_bytes_returned: u("decdn_cache_bytes_returned_total"),
101 unredeemed_usdc: u("decdn_unredeemed_usdc"),
102 rpc_healthy: m.get("decdn_rpc_healthy").copied().unwrap_or(0.0) >= 0.5,
103 }
104 }
105}
106
107#[derive(Debug, Clone)]
112pub(crate) struct SnapshotDelta {
113 pub hits: f64,
115 pub misses: f64,
117 pub bytes: f64,
119}
120
121impl SnapshotDelta {
122 pub(crate) fn between(prev: &Snapshot, now: &Snapshot, elapsed: Duration) -> Self {
123 let secs = elapsed.as_secs_f64();
124 if secs <= 0.0 {
125 return Self {
126 hits: 0.0,
127 misses: 0.0,
128 bytes: 0.0,
129 };
130 }
131 #[allow(clippy::cast_precision_loss)]
139 let d = |a: u64, b: u64| (a.saturating_sub(b)) as f64 / secs;
140 Self {
141 hits: d(now.cache_hits, prev.cache_hits),
142 misses: d(now.cache_misses, prev.cache_misses),
143 bytes: d(now.cache_bytes_returned, prev.cache_bytes_returned),
144 }
145 }
146}
147
148#[allow(clippy::cast_precision_loss)]
155pub(crate) fn hit_rate(hits: u64, misses: u64) -> Option<f64> {
156 let total = hits.saturating_add(misses);
157 if total == 0 {
158 None
159 } else {
160 Some(hits as f64 / total as f64)
161 }
162}
163
164#[allow(clippy::cast_precision_loss)]
170pub(crate) fn format_bytes(b: u64) -> String {
171 const KIB: u64 = 1024;
172 const MIB: u64 = KIB * 1024;
173 const GIB: u64 = MIB * 1024;
174 const TIB: u64 = GIB * 1024;
175 if b < KIB {
176 return format!("{b} B");
177 }
178 let (val, unit) = if b < MIB {
179 (b as f64 / KIB as f64, "KiB")
180 } else if b < GIB {
181 (b as f64 / MIB as f64, "MiB")
182 } else if b < TIB {
183 (b as f64 / GIB as f64, "GiB")
184 } else {
185 (b as f64 / TIB as f64, "TiB")
186 };
187 format!("{val:.1} {unit}")
188}
189
190#[allow(
191 clippy::cast_possible_truncation,
192 clippy::cast_sign_loss,
193 clippy::cast_precision_loss
194)]
195pub(crate) fn format_rate_bytes(per_sec: f64) -> String {
196 if per_sec <= 0.0 || !per_sec.is_finite() {
197 return "0 B/s".to_string();
198 }
199 let b = if per_sec >= u64::MAX as f64 {
200 u64::MAX
201 } else {
202 per_sec as u64
203 };
204 let s = format_bytes(b);
205 format!("{s}/s")
206}
207
208fn format_usdc(micro: u64) -> String {
217 let dollars = micro / 1_000_000;
218 let frac = micro % 1_000_000;
219 format!("${dollars}.{frac:06}")
220}
221
222fn format_uptime(secs: u64) -> String {
223 if secs < 60 {
224 return format!("{secs}s");
225 }
226 let m = secs / 60;
227 let s = secs % 60;
228 if m < 60 {
229 return format!("{m}m {s}s");
230 }
231 let h = m / 60;
232 let m = m % 60;
233 if h < 24 {
234 return format!("{h}h {m}m");
235 }
236 let d = h / 24;
237 let h = h % 24;
238 format!("{d}d {h}h")
239}
240
241pub(crate) fn write_top_table(
245 w: &mut impl io::Write,
246 metrics_url: &str,
247 now: &Snapshot,
248 prev_with_elapsed: Option<(&Snapshot, Duration)>,
249 refresh_interval: Duration,
250) -> io::Result<()> {
251 let rpc = if now.rpc_healthy { "ok" } else { "unhealthy" };
252 writeln!(
253 w,
254 "decdn node top — {metrics_url} uptime={uptime} rpc={rpc} \
255 active_connections={ac} dispatch_in_flight={inf}",
256 uptime = format_uptime(now.uptime_seconds),
257 ac = now.active_connections,
258 inf = now.dispatch_in_flight,
259 )?;
260 writeln!(w)?;
261 writeln!(w, "{:<35} {:>14} {:>12}", "METRIC", "VALUE", "/sec")?;
262
263 let delta = prev_with_elapsed.map(|(p, dt)| SnapshotDelta::between(p, now, dt));
264
265 writeln!(
266 w,
267 "{:<35} {:>14} {:>12}",
268 "active_connections", now.active_connections, "-"
269 )?;
270 writeln!(
271 w,
272 "{:<35} {:>14} {:>12}",
273 "dispatch_in_flight", now.dispatch_in_flight, "-"
274 )?;
275
276 let hits_rate = delta
277 .as_ref()
278 .map_or_else(|| "-".to_string(), |d| format!("{:.1}", d.hits));
279 let miss_rate = delta
280 .as_ref()
281 .map_or_else(|| "-".to_string(), |d| format!("{:.1}", d.misses));
282 let bytes_rate = delta
283 .as_ref()
284 .map_or_else(|| "-".to_string(), |d| format_rate_bytes(d.bytes));
285
286 writeln!(
287 w,
288 "{:<35} {:>14} {:>12}",
289 "cache_hits_total", now.cache_hits, hits_rate
290 )?;
291 writeln!(
292 w,
293 "{:<35} {:>14} {:>12}",
294 "cache_misses_total", now.cache_misses, miss_rate
295 )?;
296
297 let hit_rate_s = match hit_rate(now.cache_hits, now.cache_misses) {
298 Some(r) => format!("{:.1}%", r * 100.0),
299 None => "n/a".to_string(),
300 };
301 writeln!(w, "{:<35} {:>14} {:>12}", "cache_hit_rate", hit_rate_s, "-")?;
302 writeln!(
303 w,
304 "{:<35} {:>14} {:>12}",
305 "cache_bytes_returned_total",
306 format_bytes(now.cache_bytes_returned),
307 bytes_rate,
308 )?;
309 writeln!(
312 w,
313 "{:<35} {:>14} {:>12}",
314 "unredeemed_usdc",
315 format_usdc(now.unredeemed_usdc),
316 "-",
317 )?;
318 writeln!(w)?;
319 writeln!(
320 w,
321 "(refreshes every {:.1}s — Ctrl-C to exit)",
322 refresh_interval.as_secs_f64(),
323 )?;
324 Ok(())
325}
326
327pub(crate) fn render_json_snapshot(s: &Snapshot) -> anyhow::Result<String> {
332 let value = serde_json::json!({
333 "uptime_seconds": s.uptime_seconds,
334 "active_connections": s.active_connections,
335 "dispatch_in_flight": s.dispatch_in_flight,
336 "cache_hits_total": s.cache_hits,
337 "cache_misses_total": s.cache_misses,
338 "cache_bytes_returned_total": s.cache_bytes_returned,
339 "cache_hit_rate": hit_rate(s.cache_hits, s.cache_misses),
340 "unredeemed_usdc_micro": s.unredeemed_usdc,
343 "rpc_healthy": s.rpc_healthy,
344 });
345 serde_json::to_string_pretty(&value).context("encode top snapshot as JSON")
346}
347
348#[derive(Debug, Default, Deserialize)]
349struct MetricsPortConfig {
350 observability: Option<MetricsPortObservability>,
351}
352
353#[derive(Debug, Default, Deserialize)]
354struct MetricsPortObservability {
355 metrics_port: Option<u16>,
356}
357
358pub(crate) fn resolve_metrics_url(
373 flag: Option<&str>,
374 config_path: Option<&Path>,
375) -> anyhow::Result<String> {
376 if let Some(url) = flag {
377 return Ok(url.to_string());
378 }
379 if let Some(url) = metrics_url_from_env(std::env::var("DECDN_METRICS_PORT").ok())? {
380 return Ok(url);
381 }
382 let (resolved, source) = match config_path {
383 Some(p) => (Some(expand_tilde(p)), ConfigPathSource::Explicit),
384 None => (default_config_path(), ConfigPathSource::Default),
385 };
386 let port = port_from_config_file(resolved.as_deref(), source)?.unwrap_or(DEFAULT_METRICS_PORT);
387 Ok(format!("http://127.0.0.1:{port}"))
388}
389
390fn metrics_url_from_env(raw: Option<String>) -> anyhow::Result<Option<String>> {
396 let Some(raw) = raw else { return Ok(None) };
397 let port: u16 = raw
398 .trim()
399 .parse()
400 .map_err(|_| anyhow::anyhow!("DECDN_METRICS_PORT is not a valid port number: {raw:?}"))?;
401 if port == 0 {
402 anyhow::bail!(
403 "DECDN_METRICS_PORT=0 cannot be dialed; pass --metrics-url or set a non-zero port"
404 );
405 }
406 Ok(Some(format!("http://127.0.0.1:{port}")))
407}
408
409fn port_from_config_file(
410 path: Option<&Path>,
411 source: ConfigPathSource,
412) -> anyhow::Result<Option<u16>> {
413 let Some(path) = path else { return Ok(None) };
414 let path: PathBuf = path.to_path_buf();
415 match std::fs::read_to_string(&path) {
416 Ok(contents) => {
417 let parsed: MetricsPortConfig = toml::from_str(&contents)
422 .with_context(|| format!("failed to parse config file {}", path.display()))?;
423 match parsed.observability.and_then(|o| o.metrics_port) {
424 Some(0) => anyhow::bail!(
425 "config {} disables the metrics server (observability.metrics_port = 0); \
426 pass --metrics-url or enable the metrics port",
427 path.display(),
428 ),
429 other => Ok(other),
430 }
431 }
432 Err(err) => match (err.kind(), source) {
433 (io::ErrorKind::NotFound, ConfigPathSource::Default) => Ok(None),
434 (io::ErrorKind::PermissionDenied, _) => Err(anyhow::anyhow!(
435 "cannot read config file {}: permission denied; check file mode and ownership",
436 path.display()
437 )),
438 _ => Err(anyhow::anyhow!(
439 "failed to read config file {}: {err}",
440 path.display()
441 )),
442 },
443 }
444}
445
446const ANSI_CLEAR_HOME: &str = "\x1b[2J\x1b[H";
449
450const MAX_CONSECUTIVE_SCRAPE_FAILURES: u32 = 30;
459
460pub async fn run(args: &cli::TopArgs, global_config: Option<&Path>) -> anyhow::Result<()> {
465 anyhow::ensure!(args.timeout_ms > 0, "--timeout-ms must be > 0");
466 anyhow::ensure!(args.interval_ms > 0, "--interval-ms must be > 0");
467
468 let config_path = args.config.as_deref().or(global_config);
469 let url_base = resolve_metrics_url(args.metrics_url.as_deref(), config_path)?;
470 let metrics_url = append_metrics_path(&url_base);
471
472 let client = reqwest::Client::builder()
473 .timeout(Duration::from_millis(args.timeout_ms))
474 .build()
475 .context("failed to build HTTP client")?;
476
477 if args.json {
478 let body = fetch_metrics(&client, &metrics_url).await?;
479 let parsed = parse_openmetrics(&body);
480 let snap = Snapshot::from_metrics(&parsed);
481 println!("{}", render_json_snapshot(&snap)?);
482 return Ok(());
483 }
484
485 let interval = Duration::from_millis(args.interval_ms);
486 let mut ticker = tokio::time::interval(interval);
487 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
490
491 let mut prev: Option<(Snapshot, Instant)> = None;
492 let mut consecutive_failures: u32 = 0;
493
494 loop {
495 tokio::select! {
496 _ = ticker.tick() => {}
497 res = tokio::signal::ctrl_c() => {
498 res.context("failed to install ctrl_c handler")?;
499 return Ok(());
500 }
501 }
502
503 let now_at = Instant::now();
504 match fetch_metrics(&client, &metrics_url).await {
505 Ok(body) => {
506 consecutive_failures = 0;
507 let parsed = parse_openmetrics(&body);
508 let snap = Snapshot::from_metrics(&parsed);
509 let prev_pair = prev
510 .as_ref()
511 .map(|(s, t)| (s, now_at.saturating_duration_since(*t)));
512 let mut stdout = io::stdout().lock();
513 stdout
514 .write_all(ANSI_CLEAR_HOME.as_bytes())
515 .context("failed to write clear-screen escape")?;
516 write_top_table(&mut stdout, &metrics_url, &snap, prev_pair, interval)
517 .context("failed to write top table")?;
518 stdout.flush().context("failed to flush top table")?;
519 prev = Some((snap, now_at));
520 }
521 Err(err) => {
522 if is_permanent_scrape_error(&err) {
527 return Err(err.context(
528 "permanent scrape error (URL malformed, unsupported scheme, or persistent 4xx)",
529 ));
530 }
531 consecutive_failures += 1;
532 eprintln!("scrape failed: {err}");
533 if consecutive_failures >= MAX_CONSECUTIVE_SCRAPE_FAILURES {
534 return Err(err.context(format!(
535 "scrape failed {MAX_CONSECUTIVE_SCRAPE_FAILURES} times in a row; \
536 giving up — check the daemon and --metrics-url"
537 )));
538 }
539 }
540 }
541 }
542}
543
544fn append_metrics_path(base: &str) -> String {
554 let split_at = base.find(['?', '#']);
559 let (path_part, suffix) = match split_at {
560 Some(i) => (base.get(..i).unwrap_or(base), base.get(i..).unwrap_or("")),
561 None => (base, ""),
562 };
563 let stripped = path_part.trim_end_matches('/');
564 if stripped.ends_with("/metrics") {
565 format!("{stripped}{suffix}")
566 } else {
567 format!("{stripped}/metrics{suffix}")
568 }
569}
570
571fn is_permanent_scrape_error(err: &anyhow::Error) -> bool {
590 if err.chain().any(|e| e.to_string() == PERMANENT_STATUS_TAG) {
591 return true;
592 }
593 err.chain().any(|e| {
594 e.downcast_ref::<reqwest::Error>()
595 .is_some_and(reqwest::Error::is_builder)
596 })
597}
598
599async fn fetch_metrics(client: &reqwest::Client, url: &str) -> anyhow::Result<String> {
600 let resp = client
601 .get(url)
602 .send()
603 .await
604 .with_context(|| format!("GET {url} failed"))?;
605 let status = resp.status();
606 if !status.is_success() {
607 let is_permanent_status = status.is_client_error()
615 && status != reqwest::StatusCode::REQUEST_TIMEOUT
616 && status != reqwest::StatusCode::TOO_MANY_REQUESTS;
617 let err = anyhow::anyhow!("GET {url} returned HTTP {status}");
618 return if is_permanent_status {
619 Err(err.context(PERMANENT_STATUS_TAG))
620 } else {
621 Err(err)
622 };
623 }
624 resp.text()
625 .await
626 .with_context(|| format!("read body from {url}"))
627}
628
629const PERMANENT_STATUS_TAG: &str = "permanent HTTP status (will not retry)";
635
636#[cfg(test)]
637#[allow(
638 clippy::unwrap_used,
639 clippy::expect_used,
640 clippy::indexing_slicing,
641 clippy::panic
642)]
643mod parse_tests;
644
645#[cfg(test)]
646#[allow(
647 clippy::unwrap_used,
648 clippy::expect_used,
649 clippy::indexing_slicing,
650 clippy::panic,
651 clippy::float_cmp
652)]
653mod snapshot_tests;
654
655#[cfg(test)]
656#[allow(
657 clippy::unwrap_used,
658 clippy::expect_used,
659 clippy::indexing_slicing,
660 clippy::panic
661)]
662mod render_tests;
663
664#[cfg(test)]
665#[allow(
666 clippy::unwrap_used,
667 clippy::expect_used,
668 clippy::indexing_slicing,
669 clippy::panic
670)]
671mod json_tests;
672
673#[cfg(test)]
674#[allow(
675 clippy::unwrap_used,
676 clippy::expect_used,
677 clippy::indexing_slicing,
678 clippy::panic
679)]
680mod e2e_tests;
681
682#[cfg(test)]
683#[allow(
684 clippy::unwrap_used,
685 clippy::expect_used,
686 clippy::indexing_slicing,
687 clippy::panic
688)]
689mod resolve_tests;
690
691#[cfg(test)]
692#[allow(
693 clippy::unwrap_used,
694 clippy::expect_used,
695 clippy::indexing_slicing,
696 clippy::panic
697)]
698mod classifier_tests;
699
700#[cfg(test)]
701#[allow(
702 clippy::unwrap_used,
703 clippy::expect_used,
704 clippy::indexing_slicing,
705 clippy::panic
706)]
707mod loop_budget_tests;