Skip to main content

decdn_cli/commands/
node_top.rs

1//! `decdn node top` — live view of node activity scraped from the
2//! daemon's loopback `/metrics` HTTP endpoint (issue #275).
3
4use 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
18/// Parse an `OpenMetrics` text body into `name -> value`.
19///
20/// Scope is deliberate: only label-free counters and gauges. Any
21/// line starting with `#` is skipped — this covers the
22/// `OpenMetrics` `HELP`/`TYPE`/`UNIT`/`EOF` directives the encoder
23/// emits, plus any other comments in future revisions. Lines whose name token
24/// contains `{` (label-bearing series) are skipped — `node top`
25/// only consumes the bare `decdn_*` series the daemon registers
26/// (`crates/node/src/metrics.rs`), none of which carry labels today,
27/// so silently ignoring labelled lines is the correct degradation
28/// if the metric surface grows them later.
29///
30/// Counter `_created` timestamp lines parse as floats; callers look
31/// up the names they want and ignore the rest.
32pub(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/// Selected fields of a single `/metrics` scrape, normalised to
54/// integer types so the renderer can format them without re-checking
55/// for fractional values from the `OpenMetrics` float wire type.
56#[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    /// Raw USDC held in accepted vouchers not yet redeemed on-chain, refreshed
65    /// once per redeemer self-tick (`decdn_unredeemed_usdc`).
66    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        // Saturating cast: gauges/counters are non-negative in
73        // practice. Negative or non-finite floats clamp to 0;
74        // values above `u64::MAX` clamp to `u64::MAX`. The negative
75        // branch is mostly defensive (a clock-skew gauge we don't
76        // read could go negative); the high-end clamp matters
77        // because `f64::INFINITY >= u64::MAX as f64` is true.
78        #[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/// Per-second deltas between two consecutive snapshots. Unit is
108/// implicit in the type name; the field-level comments make it
109/// re-readable at use sites where `SnapshotDelta` isn't visible
110/// (e.g. `format_rate_bytes(d.bytes)`).
111#[derive(Debug, Clone)]
112pub(crate) struct SnapshotDelta {
113    /// Cache hits per second.
114    pub hits: f64,
115    /// Cache misses per second.
116    pub misses: f64,
117    /// Cache bytes returned per second.
118    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        // Saturating subtraction in case the underlying counter
132        // went backwards between ticks — typically a daemon
133        // restart (we keep `prev` from before the restart and the
134        // first post-restart scrape returns lower values), but
135        // also any future per-instance counter reset. Showing 0/s
136        // for that one tick is more honest than a huge spike from
137        // the wraparound.
138        #[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/// Cumulative hit ratio over `(hits + misses)`. `None` when the
149/// denominator is 0 — distinct from `Some(0.0)` ("only misses so
150/// far") so the renderer prints a literal `n/a` rather than a
151/// misleading `0.0%`. Uses `saturating_add` for symmetry with
152/// `SnapshotDelta::between`'s arithmetic, even though u64 overflow
153/// from these counters is unreachable in practice.
154#[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/// Compact byte size for display: largest binary unit at which the
165/// value is < 1024, one decimal of precision. Mirrors how `du -h`
166/// renders sizes — operators are used to that. Strict binary units
167/// (KiB/MiB/...) so the displayed value matches the raw counter
168/// when divided by the obvious power of two.
169#[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
208/// Render a raw micro-USDC amount (6 decimals, the on-chain base unit) as a
209/// dollar figure, e.g. `12_345_678` → `$12.345678`. Emits all six fractional
210/// digits rather than rounding, so the operator reads back the exact base-unit
211/// value this formatter is handed. That value is itself exact only below 2^53
212/// micro-USDC (~$9e9, far past any realistic per-node total): the `/metrics`
213/// scrape parses series as `f64` in [`parse_openmetrics`], which cannot
214/// represent every larger `u64` — an upstream limit of the text exposition,
215/// not of this formatter.
216fn 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
241/// Render the live-view table to `w`. Writing to `&mut impl Write`
242/// rather than stdout makes the formatter unit-testable and matches
243/// the `write_peers_table` pattern used elsewhere in the CLI.
244pub(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    // Pending redemption: refreshed once per redeemer self-tick, so no
310    // meaningful /sec rate — the `-` placeholder matches the other gauges.
311    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
327/// Render a single snapshot as pretty JSON. The schema is hand-rolled
328/// so the public output shape (field names, ordering, `hit_rate` as
329/// a fraction in `[0, 1]` or null) is stable across internal
330/// `Snapshot` field reorderings.
331pub(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        // Raw micro-USDC (6 decimals) so downstream tooling gets the exact
341        // integer the gauge exports; the human table renders it as dollars.
342        "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
358/// Resolve the metrics URL. Mirrors `resolve_admin_url` in
359/// `crates/cli/src/commands/node.rs` so the operator's config-file
360/// experience for `node top` is identical to the other `node`
361/// subcommands. The host
362/// is hard-coded to `127.0.0.1` because the daemon's `metrics_bind`
363/// can be `0.0.0.0` / `::`, and a client that dials that goes
364/// nowhere; the operator overrides the host explicitly via
365/// `--metrics-url` if they exposed metrics on a non-loopback address.
366///
367/// Precedence: `--metrics-url` flag → `DECDN_METRICS_PORT` env (the same var
368/// the daemon reads, `common/src/cli/run.rs`) → `observability.metrics_port`
369/// from the config file → default 9090. The env step (#864) keeps an env-only
370/// deployment reachable instead of dialing the default and reporting the node
371/// is down.
372pub(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
390/// Build the loopback metrics URL from the raw `DECDN_METRICS_PORT` env value,
391/// or `None` when unset. Pure (takes the value rather than reading the
392/// environment) so the parse/validation logic is testable without mutating
393/// process-global state. A non-numeric value or `0` is an error rather than a
394/// silent fall-through — port 0 cannot be dialed and signals misconfiguration.
395fn 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            // Partial deserializer: a typo in some unrelated section
418            // (e.g. `[payments]`) shouldn't cost the operator the
419            // ability to point `node top` at the daemon. Mirrors
420            // `port_from_config_file` for `admin_port`.
421            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
446/// ANSI escape: clear screen + move cursor home. Used between live
447/// ticks so the table redraws in place rather than scrolling.
448const ANSI_CLEAR_HOME: &str = "\x1b[2J\x1b[H";
449
450/// Cap on consecutive failed scrapes before the loop bails. 30
451/// ticks at the default `--interval-ms 1000` is ≈30s on fast-fail
452/// (connection refused), comfortably longer than a normal
453/// `decdn-node` restart. With `--timeout-ms` defaulting to 5000
454/// every consecutive timeout extends the wall-clock cap by up to
455/// 5s, so a sustained timeout takes ≈30 ticks but ≈150s. Operators
456/// chasing a real flap will see the per-tick `eprintln` in the
457/// meantime.
458const MAX_CONSECUTIVE_SCRAPE_FAILURES: u32 = 30;
459
460/// Entry point dispatched from `node_dispatch`. Polls the daemon's
461/// `/metrics` endpoint every `--interval-ms`, parses it, and renders
462/// the table. Single-shot JSON mode bypasses the loop and exits
463/// after one fetch.
464pub 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    // `Skip` so a stalled scrape doesn't cause a burst of catch-up
488    // ticks once the daemon comes back.
489    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                // Bail immediately on errors that can't possibly
523                // become transient: reqwest builder class (URL
524                // parse / unsupported scheme) and most 4xx
525                // statuses. Retrying for 30s achieves nothing.
526                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
544/// Build the full `/metrics` URL from an operator-supplied base.
545/// Tolerates the natural mistakes — a trailing `/` on the base
546/// (`http://h:9090/`), a base already ending in `/metrics`
547/// (`http://h:9090/metrics`), and a query string or fragment on
548/// either (`http://h:9090?token=foo`) — without producing
549/// `//metrics`, `/metrics/metrics`, or splicing the path inside the
550/// query value. The resolver in this crate hands us a clean
551/// `http://127.0.0.1:{port}` so the public-facing exposure here is
552/// the `--metrics-url` / `DECDN_METRICS_URL` path.
553fn append_metrics_path(base: &str) -> String {
554    // Split off any query string or fragment first so we can
555    // manipulate the path independently. Without this split, an
556    // operator URL like `http://h:9090?token=foo` would have
557    // `/metrics` appended *inside* the query value.
558    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
571/// Recognise scrape errors that can't recover by retrying:
572/// - reqwest builder errors (URL parse failures, unsupported
573///   schemes, missing host) — the URL is fixed for the lifetime of
574///   the process, so retrying for 30 ticks is just noise.
575/// - 4xx HTTP status (except 408 Request Timeout / 429 Too Many
576///   Requests, which can resolve on their own) — the daemon is
577///   reachable but the path/auth is wrong. Tagged in
578///   `fetch_metrics` via the `PERMANENT_STATUS_TAG` sentinel so we
579///   can match without inspecting the bare `anyhow::Error` shape.
580///
581/// Note that this helper is the only thing standing between a
582/// misconfigured `--metrics-url` and an indefinite retry loop.
583/// The `reqwest::Error` is matched by walking the source chain
584/// because `fetch_metrics` wraps it with `with_context`; if that
585/// wrapping stops being a direct chain element (e.g. via an
586/// indirection that boxes through a different error type), the
587/// `is_builder()` arm silently stops firing — guard with the
588/// `is_permanent_scrape_error_recognises_*` tests.
589fn 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        // Permanent vs. transient: 4xx (except 408 Request Timeout
608        // and 429 Too Many Requests, both of which can resolve on
609        // their own) means the daemon is reachable but the URL
610        // path/auth is wrong. Retrying for 30 ticks won't fix that.
611        // Tagging the error so `is_permanent_scrape_error` picks
612        // it up and bails on the first failure rather than
613        // burning 30s of operator-watching-zeros.
614        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
629/// Sentinel context string used by `fetch_metrics` to mark a 4xx
630/// (except 408/429) so `is_permanent_scrape_error` can match it
631/// without re-classifying the bare `anyhow::Error` shape. A typed
632/// scrape-error enum would replace this; until a second permanent
633/// class shows up the sentinel is enough.
634const 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;