otto/prometheus: add --format prometheus to checkup output

README.md

@@ -63,6 +63,26 @@ otto cluster checkup --ceph_report_json=report.json --format json
 stdout stays valid JSON). The document has the shape
 `{"summary": {...}, "sections": [...]}`.
 
+### Monitoring integration
+
+If you already run Prometheus + node_exporter on your Ceph hosts (the standard
+monitoring stack), `--format prometheus` emits the checkup result in Prometheus
+text exposition format, ready for node_exporter's
+[textfile collector](https://github.com/prometheus/node_exporter#textfile-collector).
+Point a cron job at the collector directory and checkup regressions show up in
+your existing dashboards and alerts:
+
+```bash
+# /etc/cron.d/otto-checkup — refresh metrics every 15 minutes
+*/15 * * * * root otto cluster checkup --format prometheus > /var/lib/node_exporter/otto.prom.$$ && mv /var/lib/node_exporter/otto.prom.$$ /var/lib/node_exporter/otto.prom
+```
+
+The write-then-rename keeps node_exporter from ever reading a half-written file.
+Metrics exported: `otto_checkup_score`, `otto_checkup_max_score`,
+`otto_checkup_section_score{section=...}`, and
+`otto_checkup_check_status{section=...,check=...}` (0=PASS, 1=WARN, 2=UNKNOWN,
+3=FAIL).
+
 ## Requirements
 
 - Python 3.11+

otto/src/clyso/ceph/otto/__init__.py

@@ -17,6 +17,7 @@
 from clyso.ceph.ai.data import CephData
 from clyso.ceph.ai.pg import add_command_pg
 from clyso.ceph.otto.upmap import add_command_upmap_remapped
+from clyso.ceph.otto.prometheus import render_prometheus
 from clyso.__version__ import __version__
 from clyso.ceph.ai.cephfs import add_command_cephfs
 from clyso.ceph.ai.rgw import add_command_rgw
@@ -221,15 +222,18 @@ def subcommand_checkup(args: argparse.Namespace) -> None:
     else:
         warnings.append("Config dump analysis skipped - using static report")
 
+    machine_format = args.format in ("json", "prometheus")
     if args.verbose or args.summary:
-        warning_stream = sys.stderr if args.format == "json" else None
+        warning_stream = sys.stderr if machine_format else None
         for warning in warnings:
             print(f"Warning: {warning}", file=warning_stream)
 
     result = generate_result(ceph_data=data)
 
     if args.format == "json":
         print(result.dump())
+    elif args.format == "prometheus":
+        print(render_prometheus(result.dump()), end="")
     elif args.summary:
         compact_result_summary(result.dump())
     elif args.verbose:
@@ -526,10 +530,11 @@ def main():
     parser_checkup.add_argument("--verbose", action="store_true", help="Verbose output")
     parser_checkup.add_argument(
         "--format",
-        choices=["text", "json"],
+        choices=["text", "json", "prometheus"],
         default="text",
-        help="Output format: 'text' (default, honors --summary/--verbose) "
-        "or 'json' (machine-readable, full result document)",
+        help="Output format: 'text' (default, honors --summary/--verbose), "
+        "'json' (machine-readable, full result document), or 'prometheus' "
+        "(text exposition format for node_exporter's textfile collector)",
     )
     parser_checkup.set_defaults(func=subcommand_checkup)
 

otto/src/clyso/ceph/otto/prometheus.py

@@ -0,0 +1,56 @@
+# Copyright (C) 2025 Clyso
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+import json
+
+# Check result -> numeric gauge value, ordered by severity
+# (PASS < WARN < UNKNOWN < FAIL) so higher values mean "worse" and dashboards
+# can alert on a threshold.
+STATUS_VALUE = {"PASS": 0, "WARN": 1, "UNKNOWN": 2, "FAIL": 3}
+
+
+def _escape_label(value: str) -> str:
+    """Escape a Prometheus label value (backslash, double-quote, newline)."""
+    return value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
+
+
+def render_prometheus(result: str) -> str:
+    """Render a checkup result JSON string as Prometheus text exposition."""
+    data = json.loads(result)
+    summary = data["summary"]
+    sections = data["sections"]
+
+    lines: list[str] = []
+
+    lines.append("# HELP otto_checkup_score Overall checkup score")
+    lines.append("# TYPE otto_checkup_score gauge")
+    lines.append(f"otto_checkup_score {summary['score']}")
+
+    lines.append("# HELP otto_checkup_max_score Maximum possible checkup score")
+    lines.append("# TYPE otto_checkup_max_score gauge")
+    lines.append(f"otto_checkup_max_score {summary['max_score']}")
+
+    lines.append("# HELP otto_checkup_section_score Per-section checkup score")
+    lines.append("# TYPE otto_checkup_section_score gauge")
+    for section in sections:
+        label = _escape_label(section["id"])
+        lines.append(
+            f'otto_checkup_section_score{{section="{label}"}} {section["score"]}'
+        )
+
+    lines.append(
+        "# HELP otto_checkup_check_status "
+        "Check status (0=PASS, 1=WARN, 2=UNKNOWN, 3=FAIL)"
+    )
+    lines.append("# TYPE otto_checkup_check_status gauge")
+    for section in sections:
+        section_label = _escape_label(section["id"])
+        for check in section["checks"]:
+            check_label = _escape_label(check["id"])
+            value = STATUS_VALUE[check["result"]]
+            lines.append(
+                f'otto_checkup_check_status{{section="{section_label}",'
+                f'check="{check_label}"}} {value}'
+            )
+
+    return "\n".join(lines) + "\n"

tests/test_checkup_prometheus.py

@@ -0,0 +1,97 @@
+# Copyright (C) 2025 Clyso
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+import json
+import re
+import subprocess
+import unittest
+
+from clyso.ceph.otto.prometheus import STATUS_VALUE, render_prometheus
+
+# Every non-comment line must be `name`, optional `{labels}`, a space, a number.
+_LINE_RE = re.compile(r"^[a-z_]+(\{[^}]*\})? -?[0-9.]+$")
+
+
+class RenderPrometheusTest(unittest.TestCase):
+    def setUp(self):
+        with open("tests/otto.json") as f:
+            self.result = json.load(f)
+        self.out = render_prometheus(json.dumps(self.result))
+
+    def test_metric_families_and_completeness(self):
+        for type_line in (
+            "# TYPE otto_checkup_score gauge",
+            "# TYPE otto_checkup_max_score gauge",
+            "# TYPE otto_checkup_section_score gauge",
+            "# TYPE otto_checkup_check_status gauge",
+        ):
+            self.assertIn(type_line, self.out)
+
+        # One check_status line per check in the input.
+        n_checks = sum(len(s["checks"]) for s in self.result["sections"])
+        n_status_lines = sum(
+            1
+            for line in self.out.splitlines()
+            if line.startswith("otto_checkup_check_status{")
+        )
+        self.assertEqual(n_status_lines, n_checks)
+
+        # Spot-check two status values: Cluster/Health is WARN (1);
+        # Version/Check for Known Issues is FAIL (3).
+        self.assertIn(
+            'otto_checkup_check_status{section="Cluster",check="Health"} '
+            f"{STATUS_VALUE['WARN']}",
+            self.out,
+        )
+        self.assertIn(
+            'otto_checkup_check_status{section="Version",'
+            'check="Check for Known Issues in Running Version"} '
+            f"{STATUS_VALUE['FAIL']}",
+            self.out,
+        )
+
+    def test_label_escaping(self):
+        result = {
+            "summary": {"score": 0.0, "max_score": 1, "grade": "F"},
+            "sections": [
+                {
+                    "id": 'Quo"te\\back',
+                    "score": 0.0,
+                    "max_score": 1,
+                    "grade": "F",
+                    "checks": [{"id": 'a"b\\c', "result": "FAIL"}],
+                }
+            ],
+        }
+        out = render_prometheus(json.dumps(result))
+        self.assertIn(r'section="Quo\"te\\back"', out)
+        self.assertIn(r'check="a\"b\\c"', out)
+
+    def test_exposition_format_shape(self):
+        self.assertTrue(self.out.endswith("\n"))
+        for line in self.out.splitlines():
+            if line.startswith("#") or not line:
+                continue
+            self.assertRegex(line, _LINE_RE)
+
+
+class CliPrometheusTest(unittest.TestCase):
+    def test_cli_format_prometheus(self):
+        process = subprocess.Popen(
+            [  # noqa: S607
+                "otto",
+                "checkup",
+                "--ceph_report_json=tests/report.pacific.json",
+                "--format=prometheus",
+            ],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        stdout_output, _ = process.communicate()
+        self.assertEqual(process.returncode, 0)
+        first_line = stdout_output.decode().splitlines()[0]
+        self.assertEqual(first_line, "# HELP otto_checkup_score Overall checkup score")
+
+
+if __name__ == "__main__":
+    unittest.main()