titan-iac/services/hermes/model-gate-configmap.yaml

# services/hermes/model-gate-configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: hermes-model-gate
  namespace: hermes
data:
  model_gate.py: |
    #!/usr/bin/env python3
    """Fail-closed proxy that admits local inference only while Hermes owns titan-24."""

    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
    import json
    import os
    from pathlib import Path
    import ssl
    import threading
    import time
    from urllib.error import HTTPError, URLError
    from urllib.request import Request, urlopen


    LISTEN_HOST = os.environ.get("LISTEN_HOST", "0.0.0.0")
    LISTEN_PORT = int(os.environ.get("LISTEN_PORT", "8080"))
    UPSTREAM_URL = os.environ.get("UPSTREAM_URL", "http://hermes-ollama.hermes.svc.cluster.local:11434").rstrip("/")
    LEASE_NAMESPACE = os.environ.get("LEASE_NAMESPACE", "hermes")
    LEASE_NAME = os.environ.get("LEASE_NAME", "titan-24-gpu-owner")
    CACHE_TTL_SEC = float(os.environ.get("LEASE_CACHE_TTL_SEC", "1"))
    API_HOST = os.environ.get("KUBERNETES_SERVICE_HOST", "kubernetes.default.svc")
    API_PORT = os.environ.get("KUBERNETES_SERVICE_PORT_HTTPS", "443")
    TOKEN_PATH = Path("/var/run/secrets/kubernetes.io/serviceaccount/token")
    CA_PATH = Path("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt")
    LEASE_URL = (
        f"https://{API_HOST}:{API_PORT}/apis/coordination.k8s.io/v1/"
        f"namespaces/{LEASE_NAMESPACE}/leases/{LEASE_NAME}"
    )

    _cache_lock = threading.Lock()
    _cached_owner = "unavailable"
    _cached_at = 0.0


    def _lease_owner() -> str:
      """Return the current owner, failing closed when Kubernetes is unavailable."""

      global _cached_at, _cached_owner
      now = time.monotonic()
      with _cache_lock:
        if now - _cached_at < CACHE_TTL_SEC:
          return _cached_owner
        try:
          token = TOKEN_PATH.read_text(encoding="utf-8").strip()
          request = Request(LEASE_URL, headers={"Authorization": f"Bearer {token}"})
          context = ssl.create_default_context(cafile=str(CA_PATH))
          with urlopen(request, timeout=3, context=context) as response:
            payload = json.load(response)
          owner = str((payload.get("spec") or {}).get("holderIdentity") or "unavailable").strip()
        except Exception:
          owner = "unavailable"
        _cached_owner = owner
        _cached_at = now
        return owner


    class Handler(BaseHTTPRequestHandler):
      """Proxy local model traffic while exposing health and ownership status."""

      protocol_version = "HTTP/1.1"

      def _json(self, status: int, payload: dict) -> None:
        body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

      def _local_allowed(self) -> tuple[bool, str]:
        owner = _lease_owner()
        return owner == "hermes", owner

      def _proxy(self) -> None:
        allowed, owner = self._local_allowed()
        if not allowed:
          self._json(
              503,
              {
                  "error": {
                      "message": f"local GPU inference unavailable while titan-24 owner is {owner}",
                      "type": "server_error",
                  },
                  "gpu_owner": owner,
                  "fallback_required": True,
              },
          )
          return

        length = int(self.headers.get("Content-Length", "0") or 0)
        body = self.rfile.read(length) if length else None
        headers = {"Content-Type": self.headers.get("Content-Type", "application/json")}
        if self.headers.get("Accept"):
          headers["Accept"] = self.headers["Accept"]
        request = Request(f"{UPSTREAM_URL}{self.path}", data=body, headers=headers, method=self.command)
        try:
          response = urlopen(request, timeout=1800)
        except HTTPError as exc:
          response = exc
        except (TimeoutError, URLError) as exc:
          self._json(503, {"error": {"message": f"local model upstream unavailable: {exc}", "type": "server_error"}})
          return

        self.send_response(response.status)
        content_type = response.headers.get("Content-Type")
        if content_type:
          self.send_header("Content-Type", content_type)
        content_length = response.headers.get("Content-Length")
        if content_length:
          self.send_header("Content-Length", content_length)
        else:
          self.send_header("Connection", "close")
          self.close_connection = True
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        while True:
          chunk = response.read(65536)
          if not chunk:
            break
          self.wfile.write(chunk)
          self.wfile.flush()
        response.close()

      def do_GET(self) -> None:
        if self.path == "/healthz":
          self._json(200, {"status": "ok"})
          return
        if self.path == "/gate/status":
          allowed, owner = self._local_allowed()
          self._json(200, {"gpu_owner": owner, "local_inference_allowed": allowed})
          return
        self._proxy()

      def do_POST(self) -> None:
        self._proxy()

      def log_message(self, format_string: str, *args) -> None:
        print(f"model-gate {self.address_string()} {format_string % args}", flush=True)


    if __name__ == "__main__":
      ThreadingHTTPServer((LISTEN_HOST, LISTEN_PORT), Handler).serve_forever()