From a67640db8305f7eadc0b19ca55c1c0aa6707029b Mon Sep 17 00:00:00 2001 From: yuyr Date: Tue, 30 Jun 2026 17:40:28 +0800 Subject: [PATCH] add rtr report metrics --- .../dashboards/ours-rp-rtr-overview.json | 666 +++++++++++++ scripts/soak/portable-soak.env.example | 5 + scripts/soak/run_24h_soak_with_metrics.sh | 18 +- src/tools/rpki_artifact_metrics.rs | 880 +++++++++++++++++- 4 files changed, 1561 insertions(+), 8 deletions(-) create mode 100644 monitor/grafana/dashboards/ours-rp-rtr-overview.json diff --git a/monitor/grafana/dashboards/ours-rp-rtr-overview.json b/monitor/grafana/dashboards/ours-rp-rtr-overview.json new file mode 100644 index 0000000..673acc1 --- /dev/null +++ b/monitor/grafana/dashboards/ours-rp-rtr-overview.json @@ -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 +} diff --git a/scripts/soak/portable-soak.env.example b/scripts/soak/portable-soak.env.example index 662daa1..88bd14d 100644 --- a/scripts/soak/portable-soak.env.example +++ b/scripts/soak/portable-soak.env.example @@ -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 diff --git a/scripts/soak/run_24h_soak_with_metrics.sh b/scripts/soak/run_24h_soak_with_metrics.sh index f23bfa0..fe59567 100755 --- a/scripts/soak/run_24h_soak_with_metrics.sh +++ b/scripts/soak/run_24h_soak_with_metrics.sh @@ -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,12 +146,17 @@ main() { free -h > "$LOG_ROOT/24h-free-before.txt" 2>&1 || true if is_true "$START_METRICS_SERVICE"; then - "$BIN_DIR/rpki_artifact_metrics" \ - --run-root "$RUN_ROOT" \ - --listen "$METRICS_LISTEN" \ - --poll-secs "$METRICS_POLL_SECS" \ - --instance "$METRICS_INSTANCE" \ - > "$LOG_ROOT/metrics.stdout" 2> "$LOG_ROOT/metrics.stderr" & + 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" \ + --poll-secs "$METRICS_POLL_SECS" \ + --instance "$METRICS_INSTANCE" \ + > "$LOG_ROOT/metrics.stdout" 2> "$LOG_ROOT/metrics.stderr" & METRICS_PID="$!" echo "$METRICS_PID" > "$LOG_ROOT/metrics.pid" fi diff --git a/src/tools/rpki_artifact_metrics.rs b/src/tools/rpki_artifact_metrics.rs index d9b13f1..d2060fa 100644 --- a/src/tools/rpki_artifact_metrics.rs +++ b/src/tools/rpki_artifact_metrics.rs @@ -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, cir: Option, ccr: Option, + 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, + parse_errors: Vec, + parse_errors_total: u64, + files: BTreeMap, + source: Option, + runtime: Option, + clients: Option, +} + +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, + latest_generated_at: Option, + latest_timestamp_seconds: Option, +} + +#[derive(Clone, Debug, Default, Serialize)] +#[serde(rename_all = "camelCase")] +struct RtrSourceMetrics { + file_path: String, + generated_at: Option, + generated_at_seconds: Option, + phase: Option, + refresh_status: Option, + refresh_changed: Option, + refresh_duration_seconds: Option, + refresh_consecutive_failures: Option, + refresh_last_success_at: Option, + refresh_last_success_seconds: Option, + ccr_file_size_bytes: Option, + data_quality: BTreeMap>, + slurm_filters: BTreeMap, + slurm_assertions: BTreeMap, + cache_availability: Option, + cache_versions: Vec, +} + +#[derive(Clone, Debug, Default, Serialize)] +#[serde(rename_all = "camelCase")] +struct RtrCacheVersion { + version: u64, + serial: Option, + snapshot: BTreeMap, + latest_delta_announced: BTreeMap, + latest_delta_withdrawn: BTreeMap, + delta_window_length: Option, +} + +#[derive(Clone, Debug, Default, Serialize)] +#[serde(rename_all = "camelCase")] +struct RtrRuntimeMetrics { + file_path: String, + generated_at: Option, + generated_at_seconds: Option, + phase: Option, + process_rss_bytes: Option, + service: Option, +} + +#[derive(Clone, Debug, Default, Serialize)] +#[serde(rename_all = "camelCase")] +struct RtrClientsMetrics { + file_path: String, + generated_at: Option, + generated_at_seconds: Option, + phase: Option, + service: Option, +} + +#[derive(Clone, Debug, Default, Serialize)] +#[serde(rename_all = "camelCase")] +struct RtrServiceMetrics { + started_at: Option, + uptime_seconds: Option, + active_connections: Option, + connections_by_transport: BTreeMap, + session_listeners: Option, + max_connections: Option, + connection_utilization: Option, +} + #[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 { - 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 { + 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 { 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 { + 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) -> 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, 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) -> Option { + let Ok(bytes) = fs::read(path) else { + errors.push(format!("read rtr report failed: {}", path.display())); + return None; + }; + match serde_json::from_slice::(&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::>() + }) + .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 { + 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> { + 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, @@ -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 { "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 { .or_else(|| current.as_u64().map(|v| v as f64)) } +fn json_bool(value: &Value, path: &[&str]) -> Option { + let mut current = value; + for key in path { + current = current.get(*key)?; + } + current.as_bool() +} + fn parse_rfc3339_to_unix(value: &str) -> Option { 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 { let rejected = vec![CirRejectedObject { object_uri: "rsync://repo.example/a/bad.roa".to_string(),