zenkey-fleet 0.14.0

Fleet engine for keyspace-v2 Zenoh tooling: disciplined fan-in queries, liveliness roster, registry-slice sets, schema-aware decode, live key-tree monitoring — the shared core of zenctl and zengui
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
//! The Prometheus text exposition of an [`ExportSnapshot`] (#228, RFC 13
//! §3 *Exporter obligations*).
//!
//! A pure function of the snapshot, so the text a scraper reads and the
//! `--format json` document can never disagree — and deterministic in every
//! byte, so two scrapes with no traffic between them are identical (the
//! snapshot's `taken_at` is deliberately not here).
//!
//! **Names and units come from the registry, never from the leaf.** A
//! series is `zenkey_subject_<producer>_<literal chunks>` with the declared
//! `unit` normalised into the conventional suffix and `_total` appended for a
//! declared `counter`; a `{var}` chunk becomes a label named by its declared
//! name. The one thing this module does *not* do is guess: a subject with no
//! `unit` gets no suffix and a subject with no `kind` is `untyped`.
//!
//! What the exposition refuses is recorded in `zenctl export --help`, not
//! here — histograms, summaries, remote write and push are tool decisions
//! this chapter records rather than makes.

use std::collections::BTreeMap;
use std::fmt::Write as _;

use crate::report::{ExportSnapshot, SeriesRow, SeriesState};

/// The metric name for a registry subject: `zenkey_subject_<producer>_<literal
/// chunks>[_<unit>][_total]`.
///
/// `unit` and `kind` are the registry's tokens (`kind` is `counter`,
/// `gauge`, `bool`, `text` or a token this build does not know). A suffix
/// already spelled by the leaf's last chunk (`rx_bytes` with `unit =
/// "bytes"`, `messages_total` with `kind = "counter"`) is not doubled — that
/// is a comparison against the declaration, not a sniff of the leaf.
pub fn metric_name(
    producer: &str,
    pattern: &str,
    unit: Option<&str>,
    kind: Option<&str>,
) -> String {
    let mut name = String::from("zenkey_subject_");
    name.push_str(&sanitize(producer));
    for chunk in pattern.split('/') {
        if chunk.starts_with('{') {
            continue;
        }
        name.push('_');
        name.push_str(&sanitize(chunk));
    }
    if let Some(unit) = unit {
        let suffix = unit_suffix(unit);
        if !suffix.is_empty() && !name.ends_with(&format!("_{suffix}")) {
            name.push('_');
            name.push_str(&suffix);
        }
    }
    if kind == Some("counter") && !name.ends_with("_total") {
        name.push_str("_total");
    }
    name
}

/// The registry `unit` as a Prometheus suffix: the base-unit spellings the
/// convention uses, anything else verbatim (sanitised).
pub fn unit_suffix(unit: &str) -> String {
    match unit {
        "ms" => "milliseconds".into(),
        "us" => "microseconds".into(),
        "ns" => "nanoseconds".into(),
        "s" => "seconds".into(),
        "bytes" | "B" => "bytes".into(),
        "percent" | "%" => "percent".into(),
        "ratio" => "ratio".into(),
        other => sanitize(other),
    }
}

/// The `# TYPE` a declared kind earns: a counter is a counter, a gauge or
/// a bool is a gauge, anything else — including no declaration — is
/// `untyped`, which is Prometheus's spelling of *not asked*.
pub fn prom_type(kind: Option<&str>) -> &'static str {
    match kind {
        Some("counter") => "counter",
        Some("gauge" | "bool") => "gauge",
        _ => "untyped",
    }
}

/// A metric or label name: `[a-zA-Z0-9_]`, everything else `_`.
pub fn sanitize(s: &str) -> String {
    s.chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '_' {
                c
            } else {
                '_'
            }
        })
        .collect()
}

/// A label value, escaped as the text format requires.
fn escape_label(v: &str) -> String {
    let mut out = String::with_capacity(v.len());
    for c in v.chars() {
        match c {
            '\\' => out.push_str("\\\\"),
            '"' => out.push_str("\\\""),
            '\n' => out.push_str("\\n"),
            c => out.push(c),
        }
    }
    out
}

/// A `# HELP` line's text, escaped.
fn escape_help(v: &str) -> String {
    let mut out = String::with_capacity(v.len());
    for c in v.chars() {
        match c {
            '\\' => out.push_str("\\\\"),
            '\n' => out.push_str("\\n"),
            c => out.push(c),
        }
    }
    out
}

/// A sample value in the text format's spelling: `+Inf`, `-Inf`, `NaN`,
/// and no exponent for anything finite.
fn number(v: f64) -> String {
    if v.is_nan() {
        "NaN".into()
    } else if v == f64::INFINITY {
        "+Inf".into()
    } else if v == f64::NEG_INFINITY {
        "-Inf".into()
    } else {
        format!("{v}")
    }
}

/// The fixed label names every subject series carries; a `{var}` whose
/// declared name collides with one is exposed as `var_<name>`.
const FIXED_LABELS: [&str; 6] = ["origin", "producer", "class", "subject", "field", "state"];

fn labels_of(row: &SeriesRow) -> Vec<(String, String)> {
    let mut labels = vec![
        ("origin".to_string(), row.origin.clone()),
        ("producer".to_string(), row.producer.clone()),
        ("class".to_string(), row.class.clone()),
        ("subject".to_string(), row.subject.clone()),
    ];
    for (name, value) in &row.labels {
        let mut name = sanitize(name);
        if FIXED_LABELS.contains(&name.as_str()) {
            name = format!("var_{name}");
        }
        labels.push((name, value.clone()));
    }
    if let Some(field) = &row.field {
        labels.push(("field".to_string(), field.clone()));
    }
    labels
}

fn label_set(labels: &[(String, String)]) -> String {
    let mut out = String::from("{");
    for (i, (k, v)) in labels.iter().enumerate() {
        if i > 0 {
            out.push(',');
        }
        let _ = write!(out, "{k}=\"{}\"", escape_label(v));
    }
    out.push('}');
    out
}

/// One family: `# HELP`, `# TYPE`, then its samples, in the order given.
struct Family {
    name: String,
    help: String,
    kind: &'static str,
    lines: Vec<String>,
}

impl Family {
    fn new(name: impl Into<String>, help: impl Into<String>, kind: &'static str) -> Family {
        Family {
            name: name.into(),
            help: help.into(),
            kind,
            lines: Vec::new(),
        }
    }

    fn sample(&mut self, labels: &[(String, String)], value: impl Into<f64>) {
        let value: f64 = value.into();
        if labels.is_empty() {
            self.lines.push(format!("{} {}", self.name, number(value)));
        } else {
            self.lines.push(format!(
                "{}{} {}",
                self.name,
                label_set(labels),
                number(value)
            ));
        }
    }

    fn write(&self, out: &mut String) {
        let _ = writeln!(out, "# HELP {} {}", self.name, escape_help(&self.help));
        let _ = writeln!(out, "# TYPE {} {}", self.name, self.kind);
        for line in &self.lines {
            out.push_str(line);
            out.push('\n');
        }
    }
}

fn l(pairs: &[(&str, &str)]) -> Vec<(String, String)> {
    pairs
        .iter()
        .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
        .collect()
}

/// The whole exposition, `text/plain; version=0.0.4`.
pub fn exposition(s: &ExportSnapshot) -> String {
    let mut fams: Vec<Family> = Vec::new();

    // ── the observer's own bounds (O6) — four populations, never summed ──
    let mut f = Family::new(
        "zenkey_observer_dropped_total",
        "samples this observer missed while behind; while this moves every value below is a lower bound (RFC 13 §3 O6)",
        "counter",
    );
    f.sample(&[], s.observer.dropped as f64);
    fams.push(f);

    let mut f = Family::new(
        "zenkey_observer_evicted_total",
        "what the observer chose to forget at its bounds, by population — keys at the stats-table bound, retained samples at the byte budget, retained samples aged out, keys unwatched (never sum these; RFC 13 §3 O6)",
        "counter",
    );
    f.sample(
        &l(&[("population", "keys")]),
        s.observer.evicted_keys as f64,
    );
    f.sample(
        &l(&[("population", "retained_bytes")]),
        s.observer.evicted_bytes as f64,
    );
    f.sample(
        &l(&[("population", "retained_age")]),
        s.observer.expired as f64,
    );
    f.sample(
        &l(&[("population", "unwatched")]),
        s.observer.unwatched as f64,
    );
    fams.push(f);

    let mut f = Family::new(
        "zenkey_observer_coalesced_total",
        "samples folded into a newer one between two scrapes — only the newest value per series is exposed (RFC 13 §3 O6)",
        "counter",
    );
    f.sample(&[], s.observer.coalesced as f64);
    fams.push(f);

    let mut f = Family::new(
        "zenkey_observer_unstamped_total",
        "samples that carried no HLC timestamp — counted, never defaulted to arrival",
        "counter",
    );
    f.sample(&[], s.observer.unstamped as f64);
    fams.push(f);

    // ── the contract: declared versus observed ──
    let mut f = Family::new(
        "zenkey_qos_judged_total",
        "samples whose subject declares a QoS profile this build knows",
        "counter",
    );
    f.sample(&[], s.contract.qos_judged as f64);
    fams.push(f);

    let mut f = Family::new(
        "zenkey_qos_mismatch_total",
        "samples that did not ride their declared QoS profile, by declared subject (RFC 04 §3)",
        "counter",
    );
    for row in &s.contract.qos_mismatch_by_subject {
        f.sample(
            &l(&[("producer", &row.producer), ("subject", &row.subject)]),
            row.n as f64,
        );
    }
    fams.push(f);

    let mut f = Family::new(
        "zenkey_payload_verdict_total",
        "payload verdicts as three counted populations — valid, invalid, not_validated — never a ratio (RFC 13 §3); without --validate everything is not_validated",
        "counter",
    );
    f.sample(&l(&[("verdict", "valid")]), s.contract.payload_valid as f64);
    f.sample(
        &l(&[("verdict", "invalid")]),
        s.contract.payload_invalid as f64,
    );
    f.sample(
        &l(&[("verdict", "not_validated")]),
        s.contract.payload_not_validated as f64,
    );
    fams.push(f);

    // ── the doctor: not asked is a state of its own ──
    let mut info = Family::new(
        "zenkey_doctor_info",
        "whether the doctor was asked to run (--doctor-every): not_asked or ran",
        "gauge",
    );
    let mut findings = Family::new(
        "zenkey_doctor_finding",
        "one series per finding of the last doctor run, by check id and severity",
        "gauge",
    );
    let mut last_run = None;
    match s.doctor.as_option() {
        None => {
            info.sample(&l(&[("state", "not_asked")]), 1.0);
        }
        Some(d) => {
            info.sample(&l(&[("state", "ran")]), 1.0);
            let mut last = Family::new(
                "zenkey_doctor_last_run_timestamp_seconds",
                "when the last doctor run finished, unix seconds",
                "gauge",
            );
            last.sample(&[], d.ran_at_unix_s as f64);
            last_run = Some(last);
            for f in &d.findings {
                let sev = serde_json::to_value(f.severity)
                    .ok()
                    .and_then(|v| v.as_str().map(str::to_string))
                    .unwrap_or_default();
                findings.sample(
                    &l(&[
                        ("check_id", f.check.as_str()),
                        ("severity", &sev),
                        ("subject", &f.subject),
                    ]),
                    1.0,
                );
            }
        }
    }
    fams.push(info);
    fams.extend(last_run);
    fams.push(findings);

    // ── scope and provenance (O5) ──
    let mut f = Family::new(
        "zenkey_scope_info",
        "one series per selector watched; `excluded` names the planes a wildcard cannot reach (RFC 13 §3 O5, RFC 03 §4 D2)",
        "gauge",
    );
    let excluded = s.excluded.join(",");
    for scope in &s.scopes {
        f.sample(&l(&[("selector", scope), ("excluded", &excluded)]), 1.0);
    }
    fams.push(f);

    let mut f = Family::new(
        "zenkey_observer_started_timestamp_seconds",
        "when this observer started watching, unix seconds — a claim about anything earlier is unobservable",
        "gauge",
    );
    f.sample(&[], s.started_at_unix_s as f64);
    fams.push(f);

    let mut f = Family::new(
        "zenkey_registry_info",
        "the contract every subject series derives from: loaded (with the producer count) or not_loaded, in which case no key refines and every key is unregistered",
        "gauge",
    );
    match s.registry.as_option() {
        Some(r) => f.sample(
            &l(&[("state", "loaded"), ("producers", &r.producers.to_string())]),
            1.0,
        ),
        None => f.sample(&l(&[("state", "not_loaded"), ("producers", "")]), 1.0),
    }
    fams.push(f);

    let mut f = Family::new(
        "zenkey_series_suppressed_total",
        "samples that produced no series, by reason: cardinality (over the declared budget), max_series, fields (per-subject field cap), text, non_numeric, undecodable, unparsed",
        "counter",
    );
    for reason in SUPPRESSION_REASONS {
        let n = s.suppressed.get(reason).copied().unwrap_or(0);
        f.sample(&l(&[("reason", reason)]), n as f64);
    }
    for (reason, n) in &s.suppressed {
        if !SUPPRESSION_REASONS.contains(&reason.as_str()) {
            f.sample(&l(&[("reason", reason)]), *n as f64);
        }
    }
    fams.push(f);

    let mut f = Family::new(
        "zenkey_unregistered_keys",
        "distinct keys the registry does not declare — counted, never exported (RFC 13 §3 O4)",
        "gauge",
    );
    f.sample(&[], s.unregistered_keys as f64);
    fams.push(f);

    let mut f = Family::new(
        "zenkey_series_count",
        "series the ledger holds, against its --max-series bound",
        "gauge",
    );
    f.sample(&[], s.series.len() as f64);
    fams.push(f);

    // ── the subject series, grouped by metric name ──
    let mut subjects: BTreeMap<&str, Family> = BTreeMap::new();
    let mut last_seen = Family::new(
        "zenkey_key_last_seen_timestamp_seconds",
        "when the newest sample of a series arrived, unix seconds on the observer's clock — constant between scrapes",
        "gauge",
    );
    let mut state = Family::new(
        "zenkey_series_state",
        "why a series is or is not current: live, quiet (a state subject past its declared ttl_s), evicted (forgotten at the observer's bound), origin_down (the producer's alive token went), retired (a tombstone) — a stopped series keeps this line and loses its value (RFC 13 §3)",
        "gauge",
    );
    let mut exposed = Family::new(
        "zenkey_series_drop_exposed_total",
        "scrape intervals in which this series was fed while zenkey_observer_dropped_total moved — its value may not have been the newest",
        "counter",
    );
    for row in &s.series {
        let labels = labels_of(row);
        if row.state.exposes_value()
            && let Some(v) = row.value
        {
            let fam = subjects.entry(row.name.as_str()).or_insert_with(|| {
                Family::new(
                    row.name.clone(),
                    format!(
                        "registry subject `{}` `{}`{}{} — name and unit from the declaration, not the leaf; may lag while zenkey_observer_dropped_total moves",
                        row.producer,
                        row.subject,
                        row.unit
                            .as_deref()
                            .map(|u| format!(", unit {u}"))
                            .unwrap_or_default(),
                        row.kind
                            .as_deref()
                            .map(|k| format!(", kind {k}"))
                            .unwrap_or_else(|| ", kind undeclared".into()),
                    ),
                    prom_type(row.kind.as_deref()),
                )
            });
            fam.sample(&labels, v);
        }
        last_seen.sample(&labels, row.last_seen_unix_s as f64);
        let mut with_state = labels.clone();
        with_state.push(("state".to_string(), row.state.as_str().to_string()));
        state.sample(&with_state, 1.0);
        exposed.sample(&labels, row.drop_exposed as f64);
    }
    for fam in subjects.into_values() {
        fams.push(fam);
    }
    fams.push(last_seen);
    fams.push(state);
    fams.push(exposed);

    let mut out = String::new();
    for fam in &fams {
        fam.write(&mut out);
    }
    out
}

/// The suppression reasons the ledger spells, so every one is exposed at
/// zero and a reason a scraper alerts on cannot be absent.
pub const SUPPRESSION_REASONS: [&str; 7] = [
    "cardinality",
    "max_series",
    "fields",
    "text",
    "non_numeric",
    "undecodable",
    "unparsed",
];

/// The states, for a consumer that wants the vocabulary without a snapshot.
pub const SERIES_STATES: [SeriesState; 5] = [
    SeriesState::Live,
    SeriesState::Quiet,
    SeriesState::Evicted,
    SeriesState::OriginDown,
    SeriesState::Retired,
];

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn names_come_from_the_declaration_and_are_never_doubled() {
        assert_eq!(
            metric_name("sysinfo", "cpu/usage", Some("percent"), Some("gauge")),
            "zenkey_subject_sysinfo_cpu_usage_percent"
        );
        assert_eq!(
            metric_name(
                "netlink",
                "iface/{iface}/rx_bytes",
                Some("bytes"),
                Some("counter")
            ),
            "zenkey_subject_netlink_iface_rx_bytes_total"
        );
        assert_eq!(
            metric_name(
                "logs",
                "by_unit/{unit}/messages_total",
                None,
                Some("counter")
            ),
            "zenkey_subject_logs_by_unit_messages_total"
        );
        assert_eq!(
            metric_name("tc-gui", "latency/p95", Some("ms"), None),
            "zenkey_subject_tc_gui_latency_p95_milliseconds"
        );
        assert_eq!(prom_type(None), "untyped");
        assert_eq!(prom_type(Some("bool")), "gauge");
        assert_eq!(prom_type(Some("histogram")), "untyped");
    }

    #[test]
    fn label_values_are_escaped_and_numbers_spell_the_specials() {
        assert_eq!(escape_label("a\"b\\c\nd"), "a\\\"b\\\\c\\nd");
        assert_eq!(number(f64::INFINITY), "+Inf");
        assert_eq!(number(f64::NEG_INFINITY), "-Inf");
        assert_eq!(number(f64::NAN), "NaN");
        assert_eq!(number(12.5), "12.5");
        assert_eq!(number(3.0), "3");
    }
}