DataOps-platform/deployment/app/edge_gateway/healthcheck.py

"""Bounded container health probe for the edge runtime."""

from __future__ import annotations

import json
import os
from datetime import UTC, datetime, timedelta
from pathlib import Path


def main() -> int:
    path = Path(os.environ.get("EDGE_HEALTH_PATH", "/run/edge/tmp/health.json"))
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
        checked = datetime.fromisoformat(str(value["checked_at"]).replace("Z", "+00:00"))
        if (
            set(value) - {
                "checked_at", "crl_next_update", "gateway_id_digest", "last_cycle",
                "queue", "status",
            }
            or value["status"] not in {"healthy", "degraded"}
            or checked.tzinfo is None
            or datetime.now(UTC) - checked.astimezone(UTC) > timedelta(seconds=90)
        ):
            return 1
    except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError):
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())