faucet-cli 1.13.0

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
74
75
76
77
78
79
80
81
82
83
84
85

//! `GET /v1/usage` (#704) — the cost & usage report over the invocations
//! this server has recorded, aggregated by pipeline / row / dataset / sink /
//! day. Requires `UsageRead` (viewer+): it is a priced read of volumes the
//! catalog already shows.

use crate::commands::usage::{parse_when, report_currency};
use crate::serve::error::ServeError;
use crate::serve::rbac::AuthContext;
use crate::serve::state::ServerState;
use crate::usage::{GroupBy, UsageFilter, UsageRecord, UsageReport, aggregate};
use axum::Json;
use axum::extract::{Extension, Query, State};
use serde::{Deserialize, Serialize};

const DEFAULT_LIMIT: usize = 5_000;
const MAX_LIMIT: usize = 50_000;

/// `GET /v1/usage` query string.
#[derive(Debug, Deserialize)]
pub struct UsageQuery {
    /// RFC 3339 or `YYYY-MM-DD`; only invocations recorded at or after it.
    pub since: Option<String>,
    /// RFC 3339 or `YYYY-MM-DD`; only invocations recorded before it.
    pub until: Option<String>,
    /// Only this pipeline.
    pub pipeline: Option<String>,
    /// Only invocations run for this tenant (#709).
    pub tenant: Option<String>,
    /// `pipeline` (default) / `row` / `dataset` / `sink` / `day`.
    pub by: Option<String>,
    /// Most invocation records to read (newest first).
    pub limit: Option<usize>,
    /// Also return the raw records the report was built from.
    #[serde(default)]
    pub include_records: bool,
}

#[derive(Debug, Serialize)]
pub struct UsageResponse {
    pub report: UsageReport,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub records: Option<Vec<UsageRecord>>,
}

pub async fn list_usage(
    State(state): State<ServerState>,
    Extension(actor): Extension<AuthContext>,
    Query(q): Query<UsageQuery>,
) -> Result<Json<UsageResponse>, ServeError> {
    let by = match q.by.as_deref() {
        None => GroupBy::Pipeline,
        Some(s) => GroupBy::parse(s).ok_or_else(|| {
            ServeError::BadConfig(format!(
                "`by` must be one of pipeline, row, dataset, sink, day, tenant (got `{s}`)"
            ))
        })?,
    };
    let filter = UsageFilter {
        since: q
            .since
            .as_deref()
            .map(parse_when)
            .transpose()
            .map_err(|e| ServeError::BadConfig(format!("since: {e}")))?,
        until: q
            .until
            .as_deref()
            .map(parse_when)
            .transpose()
            .map_err(|e| ServeError::BadConfig(format!("until: {e}")))?,
        pipeline: q.pipeline,
        tenant: actor.tenant_filter(q.tenant)?,
        limit: q.limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT),
    };
    let records = state
        .history()
        .usage_list(&filter)
        .await
        .map_err(|e| ServeError::Internal(e.to_string()))?;
    let report = aggregate(&records, by, &report_currency(&records));
    Ok(Json(UsageResponse {
        report,
        records: q.include_records.then_some(records),
    }))
}