139 lines
4.1 KiB
Python
139 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Record one Vignette admin health sample into app.admin_health_event.
|
|
|
|
This is a one-shot synthetic monitor entrypoint. It reuses the same health
|
|
calculation as GET /admin/health, but does not require an interactive admin
|
|
browser session. Schedule it with scripts/install-health-sampler-task.ps1 on
|
|
Windows, or run it manually for a local proof.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
API_ROOT = REPO_ROOT / "apps" / "api"
|
|
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
if hasattr(sys.stderr, "reconfigure"):
|
|
sys.stderr.reconfigure(encoding="utf-8")
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
description="Collect one live Vignette health sample and persist it to app.admin_health_event."
|
|
)
|
|
parser.add_argument(
|
|
"--json",
|
|
action="store_true",
|
|
help="Print a compact JSON result instead of a short text summary.",
|
|
)
|
|
parser.add_argument(
|
|
"--allow-unrecorded",
|
|
action="store_true",
|
|
help="Exit 0 even when the health check ran but no DB event rows were recorded.",
|
|
)
|
|
return parser
|
|
|
|
|
|
def _prepare_imports() -> None:
|
|
os.chdir(API_ROOT)
|
|
api_root = str(API_ROOT)
|
|
if api_root not in sys.path:
|
|
sys.path.insert(0, api_root)
|
|
|
|
|
|
def _service_payload(service: Any) -> dict[str, Any]:
|
|
return {
|
|
"key": service.key,
|
|
"status": service.status,
|
|
"metric": service.metric,
|
|
"load": service.load,
|
|
}
|
|
|
|
|
|
async def _collect() -> dict[str, Any]:
|
|
_prepare_imports()
|
|
from app.auth_sessions import ensure_runtime_tables
|
|
from app.db import close_pool, init_pool
|
|
from app.engine_client import engine_client
|
|
from app.routes.admin import apply_engine_config_from_store, record_admin_health_sample
|
|
from app.services.voice import voice_service
|
|
|
|
await init_pool()
|
|
try:
|
|
await ensure_runtime_tables()
|
|
await apply_engine_config_from_store()
|
|
await engine_client.startup()
|
|
await voice_service.startup()
|
|
health, recorded_count = await record_admin_health_sample(principal=None)
|
|
finally:
|
|
await voice_service.shutdown()
|
|
await engine_client.shutdown()
|
|
await close_pool()
|
|
return {
|
|
"ok": recorded_count > 0,
|
|
"recorded_count": recorded_count,
|
|
"generated_at": time.time(),
|
|
"status": health.status,
|
|
"environment": health.environment,
|
|
"engine_mode": health.engine_mode,
|
|
"services": [_service_payload(service) for service in health.services],
|
|
}
|
|
|
|
|
|
def _error_payload(exc: BaseException) -> dict[str, Any]:
|
|
return {
|
|
"ok": False,
|
|
"recorded_count": 0,
|
|
"generated_at": time.time(),
|
|
"error_type": exc.__class__.__name__,
|
|
"error": str(exc),
|
|
}
|
|
|
|
|
|
def _print_result(result: dict[str, Any], *, as_json: bool) -> None:
|
|
if as_json:
|
|
print(json.dumps(result, ensure_ascii=False, separators=(",", ":")))
|
|
return
|
|
if result.get("ok"):
|
|
print(
|
|
"recorded {count} health events: status={status}, env={env}, engine={engine}".format(
|
|
count=result["recorded_count"],
|
|
status=result["status"],
|
|
env=result["environment"],
|
|
engine=result["engine_mode"],
|
|
)
|
|
)
|
|
else:
|
|
print(
|
|
"health sample not recorded: {error}".format(
|
|
error=result.get("error") or f"recorded_count={result.get('recorded_count', 0)}"
|
|
),
|
|
file=sys.stderr,
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
args = build_parser().parse_args()
|
|
try:
|
|
result = asyncio.run(_collect())
|
|
except Exception as exc: # noqa: BLE001 - CLI should report a compact failure.
|
|
result = _error_payload(exc)
|
|
_print_result(result, as_json=args.json)
|
|
if result.get("ok") or args.allow_unrecorded:
|
|
return 0
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|