NASP/rpki
75
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add rtr report metrics

Browse Source
This commit is contained in:
yuyr 2026-06-30 17:40:28 +08:00
parent d5270308d6
commit a67640db83
4 changed files with 1561 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

666
monitor/grafana/dashboards/ours-rp-rtr-overview.json Normal file
View File

@ -0,0 +1,666 @@
{
"annotations": {
"list": []
},
"editable": true,
"fiscalYearStartMonth": 0,
"graphTooltip": 0,
"id": null,
"links": [],
"liveNow": false,
"panels": [
{
"id": 1,
"title": "RTR Metrics Enabled",
"type": "stat",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 4,
"w": 4,
"x": 0,
"y": 0
},
"fieldConfig": {
"defaults": {
"unit": "short",
"decimals": 0
},
"overrides": []
},
"options": {
"colorMode": "value",
"graphMode": "area",
"justifyMode": "auto",
"orientation": "auto",
"reduceOptions": {
"calcs": [
"lastNotNull"
],
"fields": "",
"values": false
},
"textMode": "auto",
"wideLayout": true
},
"targets": [
{
"expr": "max(ours_rp_rtr_metrics_enabled)",
"legendFormat": "enabled",
"refId": "A",
"instant": true
}
]
},
{
"id": 2,
"title": "Refresh Success",
"type": "stat",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 4,
"w": 5,
"x": 4,
"y": 0
},
"fieldConfig": {
"defaults": {
"unit": "short",
"decimals": 0
},
"overrides": []
},
"options": {
"colorMode": "value",
"graphMode": "area",
"justifyMode": "auto",
"orientation": "auto",
"reduceOptions": {
"calcs": [
"lastNotNull"
],
"fields": "",
"values": false
},
"textMode": "auto",
"wideLayout": true
},
"targets": [
{
"expr": "max(ours_rp_rtr_source_refresh_status{status=\"success\"})",
"legendFormat": "success",
"refId": "A",
"instant": true
}
]
},
{
"id": 3,
"title": "Consecutive Failures",
"type": "stat",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 4,
"w": 5,
"x": 9,
"y": 0
},
"fieldConfig": {
"defaults": {
"unit": "short",
"decimals": 0
},
"overrides": []
},
"options": {
"colorMode": "value",
"graphMode": "area",
"justifyMode": "auto",
"orientation": "auto",
"reduceOptions": {
"calcs": [
"lastNotNull"
],
"fields": "",
"values": false
},
"textMode": "auto",
"wideLayout": true
},
"targets": [
{
"expr": "max(ours_rp_rtr_source_refresh_consecutive_failures)",
"legendFormat": "failures",
"refId": "A",
"instant": true
}
]
},
{
"id": 4,
"title": "Last Success Age",
"type": "stat",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 4,
"w": 5,
"x": 14,
"y": 0
},
"fieldConfig": {
"defaults": {
"unit": "s",
"decimals": 0,
"min": 0
},
"overrides": []
},
"options": {
"colorMode": "value",
"graphMode": "area",
"justifyMode": "auto",
"orientation": "auto",
"reduceOptions": {
"calcs": [
"lastNotNull"
],
"fields": "",
"values": false
},
"textMode": "auto",
"wideLayout": true
},
"targets": [
{
"expr": "max(ours_rp_rtr_source_last_success_age_seconds)",
"legendFormat": "age",
"refId": "A",
"instant": true
}
]
},
{
"id": 5,
"title": "RTR RSS",
"type": "stat",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 4,
"w": 5,
"x": 19,
"y": 0
},
"fieldConfig": {
"defaults": {
"unit": "bytes",
"decimals": 0,
"min": 0
},
"overrides": []
},
"options": {
"colorMode": "value",
"graphMode": "area",
"justifyMode": "auto",
"orientation": "auto",
"reduceOptions": {
"calcs": [
"lastNotNull"
],
"fields": "",
"values": false
},
"textMode": "auto",
"wideLayout": true
},
"targets": [
{
"expr": "max(ours_rp_rtr_process_rss_bytes)",
"legendFormat": "rss",
"refId": "A",
"instant": true
}
]
},
{
"id": 6,
"title": "Data Quality Totals",
"type": "timeseries",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 8,
"w": 12,
"x": 0,
"y": 4
},
"fieldConfig": {
"defaults": {
"unit": "short",
"min": 0
},
"overrides": []
},
"options": {
"legend": {
"calcs": [
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"mode": "single",
"sort": "none"
}
},
"targets": [
{
"expr": "ours_rp_rtr_data_quality_items{stage=~\"ccr_input|before_slurm|after_slurm\",type=\"total\"}",
"legendFormat": "{{stage}} total",
"refId": "A"
},
{
"expr": "ours_rp_rtr_data_quality_items{stage=\"after_slurm\",type=\"vrp\"}",
"legendFormat": "after_slurm vrp",
"refId": "B"
},
{
"expr": "ours_rp_rtr_data_quality_items{stage=\"after_slurm\",type=\"aspa\"}",
"legendFormat": "after_slurm aspa",
"refId": "C"
}
]
},
{
"id": 7,
"title": "SLURM Filters / Assertions",
"type": "timeseries",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 8,
"w": 12,
"x": 12,
"y": 4
},
"fieldConfig": {
"defaults": {
"unit": "short",
"min": 0
},
"overrides": []
},
"options": {
"legend": {
"calcs": [
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"mode": "single",
"sort": "none"
}
},
"targets": [
{
"expr": "ours_rp_rtr_slurm_filters",
"legendFormat": "filter {{type}}",
"refId": "A"
},
{
"expr": "ours_rp_rtr_slurm_assertions",
"legendFormat": "assert {{type}}",
"refId": "B"
}
]
},
{
"id": 8,
"title": "Cache Ready / Delta Window",
"type": "timeseries",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 8,
"w": 8,
"x": 0,
"y": 12
},
"fieldConfig": {
"defaults": {
"unit": "short",
"min": 0
},
"overrides": []
},
"options": {
"legend": {
"calcs": [
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"mode": "single",
"sort": "none"
}
},
"targets": [
{
"expr": "ours_rp_rtr_cache_ready",
"legendFormat": "ready",
"refId": "A"
},
{
"expr": "ours_rp_rtr_cache_delta_window_length",
"legendFormat": "v{{version}} length",
"refId": "B"
}
]
},
{
"id": 9,
"title": "Cache Snapshot Items",
"type": "timeseries",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 8,
"w": 8,
"x": 8,
"y": 12
},
"fieldConfig": {
"defaults": {
"unit": "short",
"min": 0
},
"overrides": []
},
"options": {
"legend": {
"calcs": [
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"mode": "single",
"sort": "none"
}
},
"targets": [
{
"expr": "ours_rp_rtr_cache_snapshot_items",
"legendFormat": "v{{version}} {{type}}",
"refId": "A"
}
]
},
{
"id": 10,
"title": "Latest Delta Items",
"type": "timeseries",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 8,
"w": 8,
"x": 16,
"y": 12
},
"fieldConfig": {
"defaults": {
"unit": "short",
"min": 0
},
"overrides": []
},
"options": {
"legend": {
"calcs": [
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"mode": "single",
"sort": "none"
}
},
"targets": [
{
"expr": "ours_rp_rtr_cache_delta_items",
"legendFormat": "v{{version}} {{direction}} {{type}}",
"refId": "A"
}
]
},
{
"id": 11,
"title": "Connections",
"type": "timeseries",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 8,
"w": 8,
"x": 0,
"y": 20
},
"fieldConfig": {
"defaults": {
"unit": "short",
"min": 0
},
"overrides": []
},
"options": {
"legend": {
"calcs": [
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"mode": "single",
"sort": "none"
}
},
"targets": [
{
"expr": "ours_rp_rtr_active_connections",
"legendFormat": "active",
"refId": "A"
},
{
"expr": "ours_rp_rtr_connections",
"legendFormat": "{{transport}}",
"refId": "B"
}
]
},
{
"id": 12,
"title": "Connection Utilization / Max",
"type": "timeseries",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 8,
"w": 8,
"x": 8,
"y": 20
},
"fieldConfig": {
"defaults": {
"min": 0
},
"overrides": [
{
"matcher": {
"id": "byRegexp",
"options": ".*utilization.*"
},
"properties": [
{
"id": "unit",
"value": "percentunit"
}
]
},
{
"matcher": {
"id": "byRegexp",
"options": ".*max.*"
},
"properties": [
{
"id": "unit",
"value": "short"
}
]
}
]
},
"options": {
"legend": {
"calcs": [
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"mode": "single",
"sort": "none"
}
},
"targets": [
{
"expr": "ours_rp_rtr_connection_utilization",
"legendFormat": "utilization",
"refId": "A"
},
{
"expr": "ours_rp_rtr_max_connections",
"legendFormat": "max",
"refId": "B"
}
]
},
{
"id": 13,
"title": "Report Age",
"type": "timeseries",
"datasource": {
"type": "prometheus",
"uid": "Prometheus"
},
"gridPos": {
"h": 8,
"w": 8,
"x": 16,
"y": 20
},
"fieldConfig": {
"defaults": {
"unit": "s",
"min": 0
},
"overrides": []
},
"options": {
"legend": {
"calcs": [
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"mode": "single",
"sort": "none"
}
},
"targets": [
{
"expr": "ours_rp_rtr_source_report_age_seconds",
"legendFormat": "source",
"refId": "A"
},
{
"expr": "ours_rp_rtr_runtime_report_age_seconds",
"legendFormat": "runtime",
"refId": "B"
},
{
"expr": "ours_rp_rtr_clients_report_age_seconds",
"legendFormat": "clients",
"refId": "C"
}
]
}
],
"refresh": "10s",
"schemaVersion": 40,
"tags": [
"rpki",
"inter-rp",
"routinator"
],
"templating": {
"list": []
},
"time": {
"from": "now-6h",
"to": "now"
},
"timezone": "browser",
"title": "RTR Service Overview",
"uid": "ours-rp-rtr-overview",
"version": 1
}

5
scripts/soak/portable-soak.env.example
View File

@ -115,6 +115,11 @@ METRICS_LISTEN=0.0.0.0:9556
METRICS_POLL_SECS=5
METRICS_INSTANCE=remote231-24h
# 可选：显式开启外部 RTR report JSON 指标采集。
# 为空或未设置时不采集；设置后应指向包含
# rtr-source-*.json / rtr-runtime-*.json / rtr-clients-*.json 的目录。
RPKI_METRICS_RTR_REPORT_DIR=
# 是否启动 query service。启动后默认只索引启动后新完成的 run，避免补历史 run 影响 RP 性能。
START_QUERY_SERVICE=0

6
scripts/soak/run_24h_soak_with_metrics.sh
View File

@ -31,6 +31,7 @@ STOP_METRICS_SERVICE_ON_EXIT="${STOP_METRICS_SERVICE_ON_EXIT:-0}"
METRICS_LISTEN="${METRICS_LISTEN:-0.0.0.0:9556}"
METRICS_POLL_SECS="${METRICS_POLL_SECS:-5}"
METRICS_INSTANCE="${METRICS_INSTANCE:-remote231-24h}"
RPKI_METRICS_RTR_REPORT_DIR="${RPKI_METRICS_RTR_REPORT_DIR:-}"
START_QUERY_SERVICE="${START_QUERY_SERVICE:-0}"
STOP_QUERY_SERVICE_ON_EXIT="${STOP_QUERY_SERVICE_ON_EXIT:-0}"
QUERY_LISTEN="${QUERY_LISTEN:-0.0.0.0:9560}"
@ -145,6 +146,11 @@ main() {
free -h > "$LOG_ROOT/24h-free-before.txt" 2>&1 || true
if is_true "$START_METRICS_SERVICE"; then
metrics_env=()
if [[ -n "$RPKI_METRICS_RTR_REPORT_DIR" ]]; then
metrics_env+=(RPKI_METRICS_RTR_REPORT_DIR="$RPKI_METRICS_RTR_REPORT_DIR")
fi
env "${metrics_env[@]}" \
"$BIN_DIR/rpki_artifact_metrics" \
--run-root "$RUN_ROOT" \
--listen "$METRICS_LISTEN" \

880
src/tools/rpki_artifact_metrics.rs
View File

@ -15,6 +15,7 @@ use sha2::{Digest, Sha256};
const LARGE_PP_OBJECT_THRESHOLDS: &[u64] = &[10, 50, 100, 500, 1000, 5000, 10000, 50000];
const PP_SYNC_SECONDS_BUCKETS: &[f64] = &[0.1, 0.5, 1.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0];
const RTR_COUNT_TYPES: &[&str] = &["total", "vrp", "router_key", "aspa"];
#[derive(Clone, Debug, PartialEq, Eq)]
struct Args {
@ -160,6 +161,7 @@ struct MetricsSnapshot {
top_pp_by_sync_duration: Vec<TopPublicationPoint>,
cir: Option<CirMetrics>,
ccr: Option<CcrMetrics>,
rtr: RtrMetrics,
}
#[derive(Clone, Debug, Default, Serialize)]
@ -173,6 +175,106 @@ struct ServiceMetrics {
runs_root: String,
}
#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrMetrics {
enabled: bool,
report_dir: Option<String>,
parse_errors: Vec<String>,
parse_errors_total: u64,
files: BTreeMap<String, RtrFileScan>,
source: Option<RtrSourceMetrics>,
runtime: Option<RtrRuntimeMetrics>,
clients: Option<RtrClientsMetrics>,
}
impl RtrMetrics {
fn preferred_service(&self) -> Option<&RtrServiceMetrics> {
self.runtime
.as_ref()
.and_then(|runtime| runtime.service.as_ref())
.or_else(|| {
self.clients
.as_ref()
.and_then(|clients| clients.service.as_ref())
})
}
}
#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrFileScan {
seen: u64,
latest_path: Option<String>,
latest_generated_at: Option<String>,
latest_timestamp_seconds: Option<f64>,
}
#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrSourceMetrics {
file_path: String,
generated_at: Option<String>,
generated_at_seconds: Option<f64>,
phase: Option<String>,
refresh_status: Option<String>,
refresh_changed: Option<bool>,
refresh_duration_seconds: Option<f64>,
refresh_consecutive_failures: Option<u64>,
refresh_last_success_at: Option<String>,
refresh_last_success_seconds: Option<f64>,
ccr_file_size_bytes: Option<u64>,
data_quality: BTreeMap<String, BTreeMap<String, u64>>,
slurm_filters: BTreeMap<String, u64>,
slurm_assertions: BTreeMap<String, u64>,
cache_availability: Option<String>,
cache_versions: Vec<RtrCacheVersion>,
}
#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrCacheVersion {
version: u64,
serial: Option<u64>,
snapshot: BTreeMap<String, u64>,
latest_delta_announced: BTreeMap<String, u64>,
latest_delta_withdrawn: BTreeMap<String, u64>,
delta_window_length: Option<u64>,
}
#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrRuntimeMetrics {
file_path: String,
generated_at: Option<String>,
generated_at_seconds: Option<f64>,
phase: Option<String>,
process_rss_bytes: Option<u64>,
service: Option<RtrServiceMetrics>,
}
#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrClientsMetrics {
file_path: String,
generated_at: Option<String>,
generated_at_seconds: Option<f64>,
phase: Option<String>,
service: Option<RtrServiceMetrics>,
}
#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RtrServiceMetrics {
started_at: Option<String>,
uptime_seconds: Option<u64>,
active_connections: Option<u64>,
connections_by_transport: BTreeMap<String, u64>,
session_listeners: Option<u64>,
max_connections: Option<u64>,
connection_utilization: Option<f64>,
}
#[derive(Clone, Debug, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct RunScanSummary {
@ -391,13 +493,29 @@ struct RunRecord {
}
fn scan_run_root(input_root: &Path, instance: &str) -> Result<MetricsSnapshot, String> {
scan_run_root_incremental(input_root, instance, None)
let rtr_report_dir = configured_rtr_report_dir();
scan_run_root_incremental_with_rtr_dir(input_root, instance, None, rtr_report_dir.as_deref())
}
fn scan_run_root_incremental(
input_root: &Path,
instance: &str,
previous: Option<&MetricsSnapshot>,
) -> Result<MetricsSnapshot, String> {
let rtr_report_dir = configured_rtr_report_dir();
scan_run_root_incremental_with_rtr_dir(
input_root,
instance,
previous,
rtr_report_dir.as_deref(),
)
}
fn scan_run_root_incremental_with_rtr_dir(
input_root: &Path,
instance: &str,
previous: Option<&MetricsSnapshot>,
rtr_report_dir: Option<&Path>,
) -> Result<MetricsSnapshot, String> {
let started = Instant::now();
let runs_root = resolve_runs_root(input_root);
@ -451,9 +569,11 @@ fn scan_run_root_incremental(
}
}
snapshot.rtr = scan_rtr_reports(rtr_report_dir, previous.map(|snapshot| &snapshot.rtr));
snapshot.service.last_scan_timestamp_seconds = unix_now_seconds();
snapshot.service.last_scan_duration_seconds = started.elapsed().as_secs_f64();
snapshot.service.last_reload_success = snapshot.service.parse_errors.is_empty();
snapshot.service.last_reload_success =
snapshot.service.parse_errors.is_empty() && snapshot.rtr.parse_errors.is_empty();
Ok(snapshot)
}
@ -502,6 +622,291 @@ fn resolve_runs_root(input_root: &Path) -> PathBuf {
}
}
fn configured_rtr_report_dir() -> Option<PathBuf> {
std::env::var("RPKI_METRICS_RTR_REPORT_DIR")
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.map(PathBuf::from)
}
fn scan_rtr_reports(report_dir: Option<&Path>, previous: Option<&RtrMetrics>) -> RtrMetrics {
let mut metrics = RtrMetrics {
enabled: report_dir.is_some(),
report_dir: report_dir.map(|path| path.display().to_string()),
parse_errors_total: previous.map(|prev| prev.parse_errors_total).unwrap_or(0),
..RtrMetrics::default()
};
let Some(report_dir) = report_dir else {
return metrics;
};
if !report_dir.is_dir() {
metrics.parse_errors.push(format!(
"rtr report dir is not a directory: {}",
report_dir.display()
));
metrics.parse_errors_total = metrics
.parse_errors_total
.saturating_add(metrics.parse_errors.len() as u64);
return metrics;
}
metrics.files.insert(
"source".to_string(),
scan_rtr_kind(
report_dir,
&["rtr-source-*.json"],
&mut metrics.parse_errors,
),
);
metrics.files.insert(
"runtime".to_string(),
scan_rtr_kind(
report_dir,
&["rtr-runtime-*.json"],
&mut metrics.parse_errors,
),
);
metrics.files.insert(
"clients".to_string(),
scan_rtr_kind(
report_dir,
&["rtr-clients-*.json", "rtr-client-*.json"],
&mut metrics.parse_errors,
),
);
if let Some(file) = metrics
.files
.get("source")
.and_then(|scan| scan.latest_path.as_deref())
{
if let Some(value) = read_rtr_json(&PathBuf::from(file), &mut metrics.parse_errors) {
metrics.source = Some(parse_rtr_source_metrics(&PathBuf::from(file), &value));
if let Some(source) = metrics.source.as_ref() {
if let Some(scan) = metrics.files.get_mut("source") {
scan.latest_generated_at = source.generated_at.clone();
scan.latest_timestamp_seconds = source.generated_at_seconds;
}
}
}
}
if let Some(file) = metrics
.files
.get("runtime")
.and_then(|scan| scan.latest_path.as_deref())
{
if let Some(value) = read_rtr_json(&PathBuf::from(file), &mut metrics.parse_errors) {
metrics.runtime = Some(parse_rtr_runtime_metrics(&PathBuf::from(file), &value));
if let Some(runtime) = metrics.runtime.as_ref() {
if let Some(scan) = metrics.files.get_mut("runtime") {
scan.latest_generated_at = runtime.generated_at.clone();
scan.latest_timestamp_seconds = runtime.generated_at_seconds;
}
}
}
}
if let Some(file) = metrics
.files
.get("clients")
.and_then(|scan| scan.latest_path.as_deref())
{
if let Some(value) = read_rtr_json(&PathBuf::from(file), &mut metrics.parse_errors) {
metrics.clients = Some(parse_rtr_clients_metrics(&PathBuf::from(file), &value));
if let Some(clients) = metrics.clients.as_ref() {
if let Some(scan) = metrics.files.get_mut("clients") {
scan.latest_generated_at = clients.generated_at.clone();
scan.latest_timestamp_seconds = clients.generated_at_seconds;
}
}
}
}
metrics.parse_errors_total = metrics
.parse_errors_total
.saturating_add(metrics.parse_errors.len() as u64);
metrics
}
fn scan_rtr_kind(report_dir: &Path, patterns: &[&str], errors: &mut Vec<String>) -> RtrFileScan {
let mut names = BTreeSet::new();
for pattern in patterns {
match glob_rtr_pattern(report_dir, pattern) {
Ok(found) => {
for path in found {
names.insert(path);
}
}
Err(err) => errors.push(err),
}
}
let latest_path = names.iter().next_back().cloned();
RtrFileScan {
seen: names.len() as u64,
latest_path: latest_path.map(|path| path.display().to_string()),
latest_generated_at: None,
latest_timestamp_seconds: None,
}
}
fn glob_rtr_pattern(report_dir: &Path, pattern: &str) -> Result<Vec<PathBuf>, String> {
let Some((prefix, suffix)) = pattern.split_once('*') else {
return Err(format!("unsupported RTR report pattern: {pattern}"));
};
let mut matches = Vec::new();
let entries = fs::read_dir(report_dir)
.map_err(|e| format!("read rtr report dir failed: {}: {e}", report_dir.display()))?;
for entry in entries {
let entry = entry.map_err(|e| format!("read rtr report entry failed: {e}"))?;
let path = entry.path();
if !path.is_file() {
continue;
}
let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
continue;
};
if name.starts_with(prefix) && name.ends_with(suffix) {
matches.push(path);
}
}
matches.sort();
Ok(matches)
}
fn read_rtr_json(path: &Path, errors: &mut Vec<String>) -> Option<Value> {
let Ok(bytes) = fs::read(path) else {
errors.push(format!("read rtr report failed: {}", path.display()));
return None;
};
match serde_json::from_slice::<Value>(&bytes) {
Ok(value) => Some(value),
Err(err) => {
errors.push(format!(
"parse rtr report failed: {}: {err}",
path.display()
));
None
}
}
}
fn parse_rtr_source_metrics(path: &Path, value: &Value) -> RtrSourceMetrics {
let generated_at = json_str(value, &["generated_at"]).map(|value| value.to_string());
let generated_at_seconds = generated_at.as_deref().and_then(parse_rfc3339_to_unix);
let mut data_quality = BTreeMap::new();
for stage in ["ccr_input", "invalid", "before_slurm", "after_slurm"] {
if let Some(stage_metrics) =
json_counts_map(value.pointer(&format!("/data_quality/{stage}")))
{
data_quality.insert(stage.to_string(), stage_metrics);
}
}
let cache_versions = value
.pointer("/cache/versions")
.and_then(|value| value.as_array())
.map(|versions| {
versions
.iter()
.map(|version| RtrCacheVersion {
version: json_u64(version, &["version"]).unwrap_or(0),
serial: json_u64(version, &["serial"]),
snapshot: json_counts_map(version.get("snapshot")).unwrap_or_default(),
latest_delta_announced: json_counts_map(
version.pointer("/latest_delta/announced"),
)
.unwrap_or_default(),
latest_delta_withdrawn: json_counts_map(
version.pointer("/latest_delta/withdrawn"),
)
.unwrap_or_default(),
delta_window_length: json_u64(version, &["delta_window", "length"]),
})
.collect::<Vec<_>>()
})
.unwrap_or_default();
RtrSourceMetrics {
file_path: path.display().to_string(),
generated_at,
generated_at_seconds,
phase: json_str(value, &["phase"]).map(|value| value.to_string()),
refresh_status: json_str(value, &["refresh", "status"]).map(|value| value.to_string()),
refresh_changed: json_bool(value, &["refresh", "changed"]),
refresh_duration_seconds: json_u64(value, &["refresh", "duration_ms"])
.map(|value| value as f64 / 1000.0),
refresh_consecutive_failures: json_u64(value, &["refresh", "consecutive_failures"]),
refresh_last_success_at: json_str(value, &["refresh", "last_success_at"])
.map(|value| value.to_string()),
refresh_last_success_seconds: json_str(value, &["refresh", "last_success_at"])
.and_then(parse_rfc3339_to_unix),
ccr_file_size_bytes: json_u64(value, &["source", "ccr_file_size_bytes"]),
data_quality,
slurm_filters: json_counts_map(value.pointer("/data_quality/slurm_filters"))
.unwrap_or_default(),
slurm_assertions: json_counts_map(value.pointer("/data_quality/slurm_assertions"))
.unwrap_or_default(),
cache_availability: json_str(value, &["cache", "availability"])
.map(|value| value.to_string()),
cache_versions,
}
}
fn parse_rtr_runtime_metrics(path: &Path, value: &Value) -> RtrRuntimeMetrics {
let generated_at = json_str(value, &["generated_at"]).map(|value| value.to_string());
let generated_at_seconds = generated_at.as_deref().and_then(parse_rfc3339_to_unix);
RtrRuntimeMetrics {
file_path: path.display().to_string(),
generated_at,
generated_at_seconds,
phase: json_str(value, &["phase"]).map(|value| value.to_string()),
process_rss_bytes: json_u64(value, &["process", "rss_mib"])
.map(|value| value.saturating_mul(1024 * 1024)),
service: parse_rtr_service_metrics(value),
}
}
fn parse_rtr_clients_metrics(path: &Path, value: &Value) -> RtrClientsMetrics {
let generated_at = json_str(value, &["generated_at"]).map(|value| value.to_string());
let generated_at_seconds = generated_at.as_deref().and_then(parse_rfc3339_to_unix);
RtrClientsMetrics {
file_path: path.display().to_string(),
generated_at,
generated_at_seconds,
phase: json_str(value, &["phase"]).map(|value| value.to_string()),
service: parse_rtr_service_metrics(value),
}
}
fn parse_rtr_service_metrics(value: &Value) -> Option<RtrServiceMetrics> {
let service = value.get("service")?;
Some(RtrServiceMetrics {
started_at: json_str(service, &["started_at"]).map(|value| value.to_string()),
uptime_seconds: json_u64(service, &["uptime_seconds"]),
active_connections: json_u64(service, &["active_connections"]),
connections_by_transport: json_counts_map(service.get("connections_by_transport"))
.unwrap_or_default(),
session_listeners: json_u64(service, &["session_listeners"]),
max_connections: json_u64(service, &["max_connections"]),
connection_utilization: json_f64(service, &["connection_utilization"]),
})
}
fn json_counts_map(value: Option<&Value>) -> Option<BTreeMap<String, u64>> {
let mut out = BTreeMap::new();
let object = value?.as_object()?;
for (key, value) in object {
if let Some(count) = value.as_u64() {
out.insert(key.clone(), count);
}
}
Some(out)
}
fn collect_run_records(
runs_root: &Path,
errors: &mut Vec<String>,
@ -1229,9 +1634,298 @@ fn render_metrics(snapshot: &MetricsSnapshot) -> String {
if let Some(ccr) = snapshot.ccr.as_ref() {
render_ccr_metrics(&mut writer, instance, ccr);
}
render_rtr_metrics(&mut writer, instance, snapshot);
out
}
fn render_rtr_metrics(writer: &mut PromWriter<'_>, instance: &str, snapshot: &MetricsSnapshot) {
let rtr = &snapshot.rtr;
let scan_now = snapshot.service.last_scan_timestamp_seconds;
writer.gauge(
"ours_rp_rtr_metrics_enabled",
"Whether RTR report collection is enabled by environment variable",
&[label("instance", instance)],
bool_value(rtr.enabled),
);
writer.counter(
"ours_rp_rtr_metrics_parse_errors_total",
"Total RTR report parse errors observed by the artifact metrics service",
&[label("instance", instance)],
rtr.parse_errors_total as f64,
);
writer.gauge(
"ours_rp_rtr_metrics_parse_errors",
"Current RTR report parse error count in the latest scan",
&[label("instance", instance)],
rtr.parse_errors.len() as f64,
);
for kind in ["source", "runtime", "clients"] {
let scan = rtr.files.get(kind).cloned().unwrap_or_default();
writer.gauge(
"ours_rp_rtr_report_files_seen",
"RTR report files seen by kind",
&[label("instance", instance), label("kind", kind)],
scan.seen as f64,
);
if let Some(timestamp) = scan.latest_timestamp_seconds {
writer.gauge(
"ours_rp_rtr_report_latest_timestamp_seconds",
"Latest RTR report generated timestamp by kind",
&[label("instance", instance), label("kind", kind)],
timestamp,
);
writer.gauge(
"ours_rp_rtr_report_age_seconds",
"Age of the latest RTR report by kind at last scan time",
&[label("instance", instance), label("kind", kind)],
(scan_now - timestamp).max(0.0),
);
}
}
if let Some(source) = rtr.source.as_ref() {
if let Some(status) = source.refresh_status.as_deref() {
writer.gauge(
"ours_rp_rtr_source_refresh_status",
"Current RTR source refresh status",
&[label("instance", instance), label("status", status)],
1.0,
);
}
if let Some(value) = source.refresh_changed {
writer.gauge(
"ours_rp_rtr_source_refresh_changed",
"Whether the latest RTR source refresh changed data",
&[label("instance", instance)],
bool_value(value),
);
}
if let Some(value) = source.refresh_duration_seconds {
writer.gauge(
"ours_rp_rtr_source_refresh_duration_seconds",
"Latest RTR source refresh duration",
&[label("instance", instance)],
value,
);
}
if let Some(value) = source.refresh_consecutive_failures {
writer.gauge(
"ours_rp_rtr_source_refresh_consecutive_failures",
"Consecutive RTR source refresh failures",
&[label("instance", instance)],
value as f64,
);
}
if let Some(value) = source.refresh_last_success_seconds {
writer.gauge(
"ours_rp_rtr_source_last_success_age_seconds",
"Age of the last successful RTR source refresh at last scan time",
&[label("instance", instance)],
(scan_now - value).max(0.0),
);
}
if let Some(value) = source.generated_at_seconds {
writer.gauge(
"ours_rp_rtr_source_report_age_seconds",
"Age of the latest RTR source report at last scan time",
&[label("instance", instance)],
(scan_now - value).max(0.0),
);
}
if let Some(value) = source.ccr_file_size_bytes {
writer.gauge(
"ours_rp_rtr_source_ccr_file_size_bytes",
"RTR source CCR input file size",
&[label("instance", instance)],
value as f64,
);
}
for (stage, counts) in &source.data_quality {
for (kind, count) in counts {
writer.gauge(
"ours_rp_rtr_data_quality_items",
"RTR data quality item counts by stage and type",
&[
label("instance", instance),
label("stage", stage),
label("type", kind),
],
*count as f64,
);
}
}
for (kind, count) in &source.slurm_filters {
writer.gauge(
"ours_rp_rtr_slurm_filters",
"RTR SLURM filtered item counts by type",
&[label("instance", instance), label("type", kind)],
*count as f64,
);
}
for (kind, count) in &source.slurm_assertions {
writer.gauge(
"ours_rp_rtr_slurm_assertions",
"RTR SLURM asserted item counts by type",
&[label("instance", instance), label("type", kind)],
*count as f64,
);
}
if let Some(availability) = source.cache_availability.as_deref() {
writer.gauge(
"ours_rp_rtr_cache_ready",
"Whether RTR cache availability is ready",
&[label("instance", instance)],
bool_value(availability == "ready"),
);
}
for version in &source.cache_versions {
let version_label = version.version.to_string();
if let Some(serial) = version.serial {
writer.gauge(
"ours_rp_rtr_cache_version_serial",
"RTR cache version serial by version",
&[
label("instance", instance),
label("version", &version_label),
],
serial as f64,
);
}
if let Some(length) = version.delta_window_length {
writer.gauge(
"ours_rp_rtr_cache_delta_window_length",
"RTR cache delta window length by version",
&[
label("instance", instance),
label("version", &version_label),
],
length as f64,
);
}
for kind in RTR_COUNT_TYPES {
writer.gauge(
"ours_rp_rtr_cache_snapshot_items",
"RTR cache snapshot item counts by version and type",
&[
label("instance", instance),
label("version", &version_label),
label("type", kind),
],
version.snapshot.get(*kind).copied().unwrap_or(0) as f64,
);
}
for (direction, counts) in [
("announced", &version.latest_delta_announced),
("withdrawn", &version.latest_delta_withdrawn),
] {
for kind in RTR_COUNT_TYPES {
writer.gauge(
"ours_rp_rtr_cache_delta_items",
"RTR cache latest delta item counts by version, direction and type",
&[
label("instance", instance),
label("version", &version_label),
label("direction", direction),
label("type", kind),
],
counts.get(*kind).copied().unwrap_or(0) as f64,
);
}
}
}
}
if let Some(runtime) = rtr.runtime.as_ref() {
if let Some(value) = runtime.process_rss_bytes {
writer.gauge(
"ours_rp_rtr_process_rss_bytes",
"RTR process resident set size",
&[label("instance", instance)],
value as f64,
);
}
}
if let Some(runtime_age) = rtr
.runtime
.as_ref()
.and_then(|runtime| runtime.generated_at_seconds)
.map(|timestamp| (scan_now - timestamp).max(0.0))
{
writer.gauge(
"ours_rp_rtr_runtime_report_age_seconds",
"Age of the latest RTR runtime report at last scan time",
&[label("instance", instance)],
runtime_age,
);
}
if let Some(clients_age) = rtr
.clients
.as_ref()
.and_then(|clients| clients.generated_at_seconds)
.map(|timestamp| (scan_now - timestamp).max(0.0))
{
writer.gauge(
"ours_rp_rtr_clients_report_age_seconds",
"Age of the latest RTR clients report at last scan time",
&[label("instance", instance)],
clients_age,
);
}
if let Some(service) = rtr.preferred_service() {
if let Some(value) = service.uptime_seconds {
writer.gauge(
"ours_rp_rtr_service_uptime_seconds",
"RTR service uptime",
&[label("instance", instance)],
value as f64,
);
}
if let Some(value) = service.active_connections {
writer.gauge(
"ours_rp_rtr_active_connections",
"RTR active connections",
&[label("instance", instance)],
value as f64,
);
}
for (transport, count) in &service.connections_by_transport {
writer.gauge(
"ours_rp_rtr_connections",
"RTR connections by transport",
&[label("instance", instance), label("transport", transport)],
*count as f64,
);
}
if let Some(value) = service.session_listeners {
writer.gauge(
"ours_rp_rtr_session_listeners",
"RTR session listeners",
&[label("instance", instance)],
value as f64,
);
}
if let Some(value) = service.max_connections {
writer.gauge(
"ours_rp_rtr_max_connections",
"RTR maximum configured connections",
&[label("instance", instance)],
value as f64,
);
}
if let Some(value) = service.connection_utilization {
writer.gauge(
"ours_rp_rtr_connection_utilization",
"RTR connection utilization ratio",
&[label("instance", instance)],
value,
);
}
}
}
fn render_latest_metrics(writer: &mut PromWriter<'_>, instance: &str, latest: &LatestRunMetrics) {
writer.gauge(
"ours_rp_run_sequence",
@ -1831,6 +2525,7 @@ fn render_status_json(snapshot: &MetricsSnapshot) -> Result<String, String> {
"latestRun": snapshot.latest_run,
"cir": snapshot.cir,
"ccr": snapshot.ccr,
"rtr": snapshot.rtr,
"topRepositoriesBySyncDuration": snapshot.top_repos_by_sync_duration,
"topPublicationPointsByObjectCount": snapshot.top_pp_by_object_count,
"topPublicationPointsBySyncDuration": snapshot.top_pp_by_sync_duration,
@ -2144,6 +2839,14 @@ fn json_f64(value: &Value, path: &[&str]) -> Option<f64> {
.or_else(|| current.as_u64().map(|v| v as f64))
}
fn json_bool(value: &Value, path: &[&str]) -> Option<bool> {
let mut current = value;
for key in path {
current = current.get(*key)?;
}
current.as_bool()
}
fn parse_rfc3339_to_unix(value: &str) -> Option<f64> {
time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339)
.ok()
@ -2371,6 +3074,179 @@ mod tests {
assert!(snapshot.latest_run.is_none());
}
#[test]
fn scan_fixture_exports_rtr_metrics_when_env_enabled() {
let td = TempDir::new().expect("tempdir");
let runs = td.path().join("runs");
fs::create_dir_all(&runs).expect("create runs");
let rtr = td.path().join("rtr");
fs::create_dir_all(&rtr).expect("create rtr");
fs::write(
rtr.join("rtr-source-20260625142349411798736.json"),
r#"{
"schema_version": 2,
"generated_at": "2026-06-25T14:23:49.411798736+08:00",
"phase": "refresh_skipped_unchanged",
"source": { "ccr_file_size_bytes": 21763268 },
"refresh": {
"last_success_at": "2026-06-25T14:23:49.411053688+08:00",
"status": "success",
"changed": false,
"duration_ms": 33,
"consecutive_failures": 0
},
"data_quality": {
"ccr_input": { "total": 833492, "vrp": 832054, "router_key": 0, "aspa": 1438 },
"invalid": { "total": 0, "vrp": 0, "router_key": 0, "aspa": 0 },
"before_slurm": { "total": 833492, "vrp": 832054, "router_key": 0, "aspa": 1438 },
"after_slurm": { "total": 828563, "vrp": 827126, "router_key": 0, "aspa": 1437 },
"slurm_filters": { "prefix": 1, "router_key": 0, "aspa": 1 },
"slurm_assertions": { "prefix": 0, "router_key": 0, "aspa": 0 }
},
"cache": {
"availability": "ready",
"versions": [
{
"version": 2,
"serial": 9,
"snapshot": { "total": 828563, "vrp": 827126, "router_key": 0, "aspa": 1437 },
"latest_delta": {
"announced": { "total": 1, "vrp": 1, "router_key": 0, "aspa": 0 },
"withdrawn": { "total": 1, "vrp": 1, "router_key": 0, "aspa": 0 }
},
"delta_window": { "length": 4 }
}
]
}
}"#,
)
.expect("source");
fs::write(
rtr.join("rtr-runtime-20260625142610957948453.json"),
r#"{
"schema_version": 2,
"generated_at": "2026-06-25T14:26:10.957948453+08:00",
"phase": "runtime_periodic",
"service": {
"started_at": "2026-06-24T11:51:08.599427589+08:00",
"uptime_seconds": 95714,
"active_connections": 0,
"connections_by_transport": { "tcp": 0, "tls": 0, "ssh": 0 },
"session_listeners": 0,
"max_connections": 512,
"connection_utilization": 0.0
},
"process": { "rss_mib": 1285 }
}"#,
)
.expect("runtime");
fs::write(
rtr.join("rtr-clients-20260624115123282266076.json"),
r#"{
"schema_version": 2,
"generated_at": "2026-06-24T11:51:23.282266076+08:00",
"phase": "startup",
"service": {
"started_at": "2026-06-24T11:51:08.599427589+08:00",
"uptime_seconds": 14,
"active_connections": 0,
"connections_by_transport": { "tcp": 0, "tls": 0, "ssh": 0 },
"session_listeners": 0,
"max_connections": 512,
"connection_utilization": 0.0
}
}"#,
)
.expect("clients");
let snapshot = scan_run_root_incremental_with_rtr_dir(td.path(), "test", None, Some(&rtr))
.expect("scan");
assert!(snapshot.rtr.enabled);
assert_eq!(snapshot.rtr.files["source"].seen, 1);
assert_eq!(
snapshot
.rtr
.source
.as_ref()
.unwrap()
.refresh_status
.as_deref(),
Some("success")
);
assert_eq!(
snapshot.rtr.source.as_ref().unwrap().data_quality["after_slurm"]["aspa"],
1437
);
assert_eq!(
snapshot.rtr.runtime.as_ref().unwrap().process_rss_bytes,
Some(1285 * 1024 * 1024)
);
assert_eq!(
snapshot
.rtr
.preferred_service()
.and_then(|service| service.max_connections),
Some(512)
);
let metrics = render_metrics(&snapshot);
assert!(metrics.contains("ours_rp_rtr_metrics_enabled"));
assert!(metrics.contains(r#"ours_rp_rtr_cache_ready{instance="test"} 1"#));
assert!(metrics.contains(
r#"ours_rp_rtr_data_quality_items{instance="test",stage="after_slurm",type="aspa"} 1437"#
));
assert!(metrics.contains(
r#"ours_rp_rtr_cache_snapshot_items{instance="test",version="2",type="vrp"} 827126"#
));
assert!(metrics.contains(
r#"ours_rp_rtr_cache_delta_items{instance="test",version="2",direction="withdrawn",type="total"} 1"#
));
assert!(metrics.contains(r#"ours_rp_rtr_process_rss_bytes{instance="test"} 1347420160"#));
assert!(metrics.contains(r#"ours_rp_rtr_max_connections{instance="test"} 512"#));
assert!(metrics.contains(r#"ours_rp_rtr_connections{instance="test",transport="ssh"} 0"#));
let status = render_status_json(&snapshot).expect("status");
assert!(status.contains(r#""enabled": true"#));
assert!(status.contains(r#""processRssBytes": 1347420160"#));
}
#[test]
fn rtr_bad_json_and_missing_fields_do_not_crash() {
let td = TempDir::new().expect("tempdir");
let runs = td.path().join("runs");
fs::create_dir_all(&runs).expect("create runs");
let rtr = td.path().join("rtr");
fs::create_dir_all(&rtr).expect("create rtr");
fs::write(rtr.join("rtr-source-20260625000000000000000.json"), "{bad").expect("bad");
fs::write(
rtr.join("rtr-runtime-20260625000000000000000.json"),
r#"{"generated_at":"2026-06-25T14:26:10+08:00","process":{}}"#,
)
.expect("runtime");
fs::write(
rtr.join("rtr-clients-20260625000000000000000.json"),
r#"{"generated_at":"2026-06-25T14:26:10+08:00"}"#,
)
.expect("clients");
let snapshot = scan_run_root_incremental_with_rtr_dir(td.path(), "test", None, Some(&rtr))
.expect("scan");
assert!(snapshot.rtr.enabled);
assert_eq!(snapshot.rtr.files["source"].seen, 1);
assert!(snapshot.rtr.source.is_none());
assert!(snapshot.rtr.runtime.is_some());
assert!(snapshot.rtr.clients.is_some());
assert!(!snapshot.rtr.parse_errors.is_empty());
let metrics = render_metrics(&snapshot);
assert!(metrics.contains("ours_rp_rtr_metrics_parse_errors_total"));
assert!(metrics.contains(r#"ours_rp_rtr_metrics_enabled{instance="test"} 1"#));
}
fn sample_cir() -> Vec<u8> {
let rejected = vec![CirRejectedObject {
object_uri: "rsync://repo.example/a/bad.roa".to_string(),