faucet_core/observability/
drift.rs

1//! Metric registration + emit helper for schema-drift handling (issue #194).
2//! `faucet_schema_drift_total` counts top-level columns detected as drifted on
3//! a page, labelled by the policy mode that handled the drift and the drift
4//! kind. Emitted from the pipeline loop where the per-page drift pass runs.
5
6use metrics::{counter, describe_counter};
7
8/// Register HELP text for the schema-drift metric. Idempotent — safe to call
9/// more than once. Invoked from `install_observability` so the description is
10/// present in `/metrics` from t=0, even before the first drift event. No-op
11/// when no recorder is installed.
12pub fn describe() {
13    describe_counter!(
14        "faucet_schema_drift_total",
15        "Top-level columns detected as drifted, by policy mode and drift kind."
16    );
17}
18
19/// Emit `faucet_schema_drift_total{pipeline,row,connector,mode,kind}` for a
20/// detected drift: `count` columns of `kind` handled under policy `mode`.
21/// No-op when `count == 0`.
22pub fn schema_drift(
23    pipeline: &str,
24    row: &str,
25    connector: &str,
26    mode: &str,
27    kind: &str,
28    count: u64,
29) {
30    if count == 0 {
31        return;
32    }
33    counter!(
34        "faucet_schema_drift_total",
35        "pipeline" => pipeline.to_string(),
36        "row" => row.to_string(),
37        "connector" => connector.to_string(),
38        "mode" => mode.to_string(),
39        "kind" => kind.to_string(),
40    )
41    .increment(count);
42}
43
44#[cfg(test)]
45mod tests {
46    use super::*;
47
48    #[test]
49    fn describe_is_callable_and_idempotent() {
50        // No recorder installed in this test → describe is a no-op; the call
51        // must not panic regardless, and repeat calls are safe.
52        describe();
53        describe();
54    }
55
56    #[test]
57    fn emit_does_not_panic_without_recorder() {
58        // Metric emits are no-ops when no recorder is installed; assert the
59        // helper stays panic-free.
60        schema_drift("p", "r", "postgres", "evolve", "added", 2);
61        // count == 0 short-circuits without emitting; must also not panic.
62        schema_drift("p", "r", "postgres", "ignore", "removed", 0);
63    }
64}
faucet_core/observability/drift.rs

faucet_core/observability/
drift.rs
