feat(ops): automate metrics edge controlled apply
This commit is contained in:
56
scripts/ops/apply_metrics_edge_policy.py
Executable file
56
scripts/ops/apply_metrics_edge_policy.py
Executable file
@@ -0,0 +1,56 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Check or apply the source-owned EwoooC metrics edge policy."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from services.metrics_edge_controlled_apply_service import ( # noqa: E402
|
||||
apply_metrics_edge_plan,
|
||||
build_metrics_edge_plan,
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--nginx-config", type=Path, required=True)
|
||||
parser.add_argument("--prometheus-config", type=Path, required=True)
|
||||
parser.add_argument("--nginx-fragment", type=Path, required=True)
|
||||
parser.add_argument("--prometheus-snippet", type=Path, required=True)
|
||||
parser.add_argument("--run-id", required=True)
|
||||
parser.add_argument("--work-item-id", default="SEC-P0-001")
|
||||
parser.add_argument("--apply", action="store_true")
|
||||
parser.add_argument("--canary-timeout", type=int, default=90)
|
||||
parser.add_argument("--receipt", type=Path)
|
||||
args = parser.parse_args()
|
||||
|
||||
plan = build_metrics_edge_plan(
|
||||
nginx_config=args.nginx_config,
|
||||
prometheus_config=args.prometheus_config,
|
||||
nginx_fragment=args.nginx_fragment,
|
||||
prometheus_snippet=args.prometheus_snippet,
|
||||
run_id=args.run_id,
|
||||
work_item_id=args.work_item_id,
|
||||
)
|
||||
payload = apply_metrics_edge_plan(
|
||||
plan,
|
||||
canary_timeout=args.canary_timeout,
|
||||
) if args.apply else {key: value for key, value in plan.items() if not key.startswith("_")}
|
||||
output = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)
|
||||
if args.receipt:
|
||||
args.receipt.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.receipt.write_text(f"{output}\n", encoding="utf-8")
|
||||
print(output)
|
||||
return 0 if payload.get("status") not in {"rolled_back", "rollback_degraded"} else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user