faucet-cli 1.13.1

Config-driven CLI runner for faucet-stream pipelines (YAML / JSON, Meltano-style)
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

//! `faucet_serve_tenant_*`, `faucet_serve_connections` and
//! `faucet_serve_connect_flows_total` (#709). The `tenant` label is bounded by
//! the tenants an operator creates.

use crate::serve::state::ServerState;
use metrics::{counter, describe_counter, describe_gauge, gauge};

pub fn describe() {
    describe_counter!(
        "faucet_serve_tenant_runs_total",
        "Runs started for a tenant that finished, by tenant and outcome"
    );
    describe_counter!(
        "faucet_serve_tenant_limit_rejections_total",
        "Runs refused by a tenant limit, by tenant and limit"
    );
    describe_gauge!(
        "faucet_serve_connections",
        "Tenant connections by status (active / needs_reauth)"
    );
    describe_counter!(
        "faucet_serve_connect_flows_total",
        "Hosted OAuth connect flows, by provider and outcome"
    );
}

pub fn record_run(tenant: &str, outcome: &str) {
    counter!(
        "faucet_serve_tenant_runs_total",
        "tenant" => tenant.to_string(),
        "outcome" => outcome.to_string()
    )
    .increment(1);
}

pub fn record_limit_rejection(tenant: &str, limit: &'static str) {
    counter!(
        "faucet_serve_tenant_limit_rejections_total",
        "tenant" => tenant.to_string(),
        "limit" => limit
    )
    .increment(1);
}

pub fn record_connect_flow(provider: &str, outcome: &'static str) {
    counter!(
        "faucet_serve_connect_flows_total",
        "provider" => provider.to_string(),
        "outcome" => outcome
    )
    .increment(1);
}

/// Count connections by status across every tenant. Best-effort.
pub async fn refresh_connection_gauges(state: &ServerState) {
    let history = state.history();
    let Ok(tenants) = history.tenant_list().await else {
        return;
    };
    let (mut active, mut reauth) = (0u64, 0u64);
    for t in tenants {
        if let Ok(conns) = history.connection_list(&t.id).await {
            for c in conns {
                match c.status {
                    crate::serve::history::tenants::ConnectionStatus::Active => active += 1,
                    crate::serve::history::tenants::ConnectionStatus::NeedsReauth => reauth += 1,
                }
            }
        }
    }
    gauge!("faucet_serve_connections", "status" => "active").set(active as f64);
    gauge!("faucet_serve_connections", "status" => "needs_reauth").set(reauth as f64);
}