0da9e4c82d
- Move flat service manifests into structured subdirs (apps/, bootstrap-jobs/, repair-jobs/, migration-jobs/, validation-jobs/, node-ops/, networking/) - Retire oneoffs/ directories across services - Remove oceanus cluster and its host roles; add aether cluster + terraform scaffolding - Reorganize scripts/ into ops/, render/, sync/, manual-tests/ - Add Makefile with render/validate/test/flux targets and repo-structure tests - Update flux-system application CRs to the new paths - Add hermes-automated-triage-24h-plan knowledge doc (+ comms mirror) - Refresh knowledge catalogs, dashboards, vmalert rules, quality contract Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
86 lines
2.6 KiB
Python
Executable File
86 lines
2.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Print a compact inventory of Flux Kustomization resources."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
def _iter_docs(path: Path) -> list[dict[str, Any]]:
|
|
docs: list[dict[str, Any]] = []
|
|
for raw in yaml.safe_load_all(path.read_text(encoding="utf-8")):
|
|
if isinstance(raw, dict):
|
|
docs.append(raw)
|
|
return docs
|
|
|
|
|
|
def _is_flux_kustomization(doc: dict[str, Any]) -> bool:
|
|
return (
|
|
doc.get("kind") == "Kustomization"
|
|
and str(doc.get("apiVersion") or "").startswith("kustomize.toolkit.fluxcd.io/")
|
|
)
|
|
|
|
|
|
def _depends_on(spec: dict[str, Any]) -> str:
|
|
names = []
|
|
for dep in spec.get("dependsOn") or []:
|
|
if isinstance(dep, dict) and dep.get("name"):
|
|
names.append(str(dep["name"]))
|
|
return ",".join(names)
|
|
|
|
|
|
def _row(doc: dict[str, Any], path: Path) -> dict[str, str]:
|
|
metadata = doc.get("metadata") or {}
|
|
spec = doc.get("spec") or {}
|
|
annotations = metadata.get("annotations") or {}
|
|
return {
|
|
"name": str(metadata.get("name") or ""),
|
|
"path": str(spec.get("path") or ""),
|
|
"namespace": str(spec.get("targetNamespace") or ""),
|
|
"depends": _depends_on(spec),
|
|
"prune": str(spec.get("prune", "")),
|
|
"wait": str(spec.get("wait", "")),
|
|
"suspend": str(spec.get("suspend", False)),
|
|
"reason": str(annotations.get("atlas.bstein.dev/suspend-reason") or ""),
|
|
"file": str(path),
|
|
}
|
|
|
|
|
|
def build_inventory(root: Path) -> list[dict[str, str]]:
|
|
rows: list[dict[str, str]] = []
|
|
for path in sorted(root.rglob("*.yaml")):
|
|
for doc in _iter_docs(path):
|
|
if _is_flux_kustomization(doc):
|
|
rows.append(_row(doc, path))
|
|
return sorted(rows, key=lambda item: item["name"])
|
|
|
|
|
|
def _print_table(rows: list[dict[str, str]]) -> None:
|
|
columns = ["name", "path", "namespace", "depends", "prune", "wait", "suspend", "reason"]
|
|
widths = {
|
|
column: max([len(column), *[len(row[column]) for row in rows]]) for column in columns
|
|
}
|
|
header = " ".join(column.ljust(widths[column]) for column in columns)
|
|
print(header)
|
|
print(" ".join("-" * widths[column] for column in columns))
|
|
for row in rows:
|
|
print(" ".join(row[column].ljust(widths[column]) for column in columns))
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("root", nargs="?", default="clusters/atlas/flux-system")
|
|
args = parser.parse_args()
|
|
|
|
rows = build_inventory(Path(args.root))
|
|
_print_table(rows)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|