update inter-rp repo sync metrics

scripts/inter_rp/inter_rp_ours_routinator_exporter.py

@@ -155,15 +155,44 @@ def parse_prometheus_samples(text, metric_name):
             continue
         yield parse_prom_labels(match.group(1)), value
 
+def canonical_object_type(value):
+    value = (value or "").strip().lower()
+    if value in ("manifest", "mft"):
+        return "mft"
+    if value == "crl":
+        return "crl"
+    if value in ("certificate", "cert", "ca_cert", "router_cert", "ee_cert", "crt"):
+        return "crt"
+    if value == "roa":
+        return "roa"
+    if value == "aspa":
+        return "aspa"
+    return None
+
+def empty_object_counts():
+    return {"mft": 0, "crl": 0, "crt": 0, "roa": 0, "aspa": 0}
+
+def add_object_count(counts_by_uri, uri, object_type, value):
+    canonical = canonical_object_type(object_type)
+    if not uri or canonical is None:
+        return
+    counts = counts_by_uri.setdefault(uri, empty_object_counts())
+    counts[canonical] = counts.get(canonical, 0) + int(value)
+
+def object_count_labels(counts_by_uri, uri):
+    counts = counts_by_uri.get(uri, empty_object_counts())
+    return {key: str(int(counts.get(key, 0))) for key in ["mft", "crl", "crt", "roa", "aspa"]}
+
 def load_ours_repo_sets(errors):
     try:
         import urllib.request
         text = urllib.request.urlopen("http://127.0.0.1:9556/metrics", timeout=10).read().decode("utf-8", "replace")
     except Exception as exc:
         errors.append(f"ours-rp repo metrics: {exc}")
-        return {"total": set(), "available": set(), "failed": set(), "info": {}}
+        return {"total": set(), "available": set(), "failed": set(), "info": {}, "object_counts": {}}
     info = {}
     states = {}
+    object_counts = {}
     for labels, value in parse_prometheus_samples(text, "ours_rp_repository_info"):
         uri = labels.get("uri")
         if uri:
@@ -178,6 +207,15 @@ def load_ours_repo_sets(errors):
         if not uri:
             continue
         states.setdefault(uri, {})[labels.get("terminal_state", "unknown")] = value
+    for labels, value in parse_prometheus_samples(text, "ours_rp_repository_objects_by_type"):
+        uri = labels.get("uri")
+        if not uri:
+            repo_id = labels.get("repo_id")
+            for candidate_uri, candidate in info.items():
+                if candidate.get("repo_id") == repo_id:
+                    uri = candidate_uri
+                    break
+        add_object_count(object_counts, uri, labels.get("object_type"), value)
     total = set(info)
     available = set()
     failed = set()
@@ -188,22 +226,23 @@ def load_ours_repo_sets(errors):
             available.add(uri)
         else:
             failed.add(uri)
-    return {"total": total, "available": available, "failed": failed, "info": info}
+    return {"total": total, "available": available, "failed": failed, "info": info, "object_counts": object_counts}
 
 def load_routinator_repo_sets(errors):
     path = PEER_ROOT / "routinator" / "routinator-metrics.prom"
     if not path.exists():
         errors.append(f"routinator repo metrics: missing {path}")
-        return {"total": set(), "success": set(), "failed": set(), "duration": {}}
+        return {"total": set(), "success": set(), "failed": set(), "duration": {}, "object_counts": {}}
     try:
         text = path.read_text(encoding="utf-8", errors="replace")
     except Exception as exc:
         errors.append(f"routinator repo metrics: read {path}: {exc}")
-        return {"total": set(), "success": set(), "failed": set(), "duration": {}}
+        return {"total": set(), "success": set(), "failed": set(), "duration": {}, "object_counts": {}}
     total = set()
     success = set()
     failed = set()
     duration = {}
+    object_counts = {}
     for metric in ["routinator_rrdp_status", "routinator_rsync_status"]:
         for labels, value in parse_prometheus_samples(text, metric):
             uri = labels.get("uri")
@@ -220,7 +259,9 @@ def load_routinator_repo_sets(errors):
             uri = labels.get("uri")
             if uri:
                 duration[uri] = max(duration.get(uri, 0.0), value)
-    return {"total": total, "success": success, "failed": failed, "duration": duration}
+    for labels, value in parse_prometheus_samples(text, "routinator_repository_objects_total"):
+        add_object_count(object_counts, labels.get("uri"), labels.get("type"), value)
+    return {"total": total, "success": success, "failed": failed, "duration": duration, "object_counts": object_counts}
 
 def emit_repo_diff_metrics(out, errors):
     ours = load_ours_repo_sets(errors)
@@ -242,9 +283,18 @@ def emit_repo_diff_metrics(out, errors):
     out.append(metric_line("inter_rp_repo_sync_overlap_total", {**base, "class": "only_ours"}, len(only_ours)))
     out.append(metric_line("inter_rp_repo_sync_overlap_total", {**base, "class": "only_routinator"}, len(only_routinator)))
     out.append(metric_line("inter_rp_repo_sync_overlap_total", {**base, "class": "neither_available"}, len(neither)))
-    for diff_class, uris in [("only_ours", only_ours), ("only_routinator", only_routinator)]:
+    for diff_class, uris, counts_by_uri in [
+        ("only_ours", only_ours, ours["object_counts"]),
+        ("only_routinator", only_routinator, routinator["object_counts"]),
+    ]:
         for rank, uri in enumerate(uris[:50], start=1):
-            labels = {**base, "class": diff_class, "rank": rank, "uri": uri}
+            labels = {
+                **base,
+                "class": diff_class,
+                "rank": rank,
+                "uri": uri,
+                **object_count_labels(counts_by_uri, uri),
+            }
             if uri in routinator["duration"]:
                 labels["routinator_duration"] = f"{routinator['duration'][uri]:.3f}"
             out.append(metric_line("inter_rp_repo_sync_diff_info", labels, 1))