增加监控导出容器编排镜像audit

tests/test_audit_k8s_images.py

@@ -0,0 +1,41 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from tools.audit_k8s_images import audit
+
+
+class AuditK8sImagesTest(unittest.TestCase):
+    def test_detects_inconsistent_exporter_images(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            deploy = root / "deployment" / "mx-exporter"
+            deploy.mkdir(parents=True)
+            (deploy / "a.yaml").write_text("image: repo/mx-exporter:v1\n", encoding="utf-8")
+            (deploy / "b.yaml").write_text("image: repo/mx-exporter:v2\n", encoding="utf-8")
+
+            report = audit(root)
+
+        self.assertEqual(report["mx_exporter_image_count"], 2)
+        self.assertIs(report["mx_exporter_image_consistent"], False)
+
+    def test_preserves_quoted_helm_template_images(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            deploy = root / "deployment" / "mx-exporter"
+            deploy.mkdir(parents=True)
+            (deploy / "values.yaml").write_text(
+                'image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}"\n',
+                encoding="utf-8",
+            )
+
+            report = audit(root)
+
+        self.assertEqual(
+            report["mx_exporter_images"],
+            ["{{ .Values.image.repository }}:{{ .Values.image.tag }}"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/audit_k8s_images.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""Audit Kubernetes and Helm manifests for mx-exporter image references."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+from pathlib import Path
+
+
+IMAGE_RE = re.compile(r"^\s*image:\s*['\"]?([^'\"\n]+?)['\"]?\s*(?:#.*)?$", re.MULTILINE)
+
+
+def collect_images(root: Path) -> list[dict[str, str]]:
+    images: list[dict[str, str]] = []
+    deploy_dir = root / "deployment"
+    if not deploy_dir.is_dir():
+        return images
+
+    for path in sorted(deploy_dir.rglob("*")):
+        if path.suffix not in {".yaml", ".yml"}:
+            continue
+        if not path.is_file():
+            continue
+        text = path.read_text(encoding="utf-8", errors="replace")
+        for match in IMAGE_RE.finditer(text):
+            images.append({"path": path.relative_to(root).as_posix(), "image": match.group(1)})
+    return images
+
+
+def audit(root: Path) -> dict[str, object]:
+    images = collect_images(root)
+    exporter_images = sorted({item["image"] for item in images if "mx-exporter" in item["path"].lower()})
+    return {
+        "image_count": len(images),
+        "images": images,
+        "mx_exporter_images": exporter_images,
+        "mx_exporter_image_count": len(exporter_images),
+        "mx_exporter_image_consistent": len(exporter_images) <= 1,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--root", type=Path, default=Path.cwd(), help="repository root")
+    parser.add_argument("--strict", action="store_true", help="return non-zero when mx-exporter images differ")
+    parser.add_argument("--output", type=Path, help="write audit JSON to this path")
+    args = parser.parse_args()
+
+    payload = audit(args.root)
+    text = json.dumps(payload, indent=2, ensure_ascii=False)
+    if args.output:
+        args.output.write_text(text + "\n", encoding="utf-8")
+    else:
+        print(text)
+    return 1 if args.strict and not payload["mx_exporter_image_consistent"] else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())