Add Minerva Direct API support

cron_jobs/direct_health_monitor.py

@@ -65,8 +65,9 @@
 from resource_server_async.clusters import (
     BaseCluster,  # noqa: E402
     MetisCluster,
+    MinervaCluster,
 )
-from resource_server_async.endpoints import MetisEndpoint
+from resource_server_async.endpoints import MetisEndpoint, MinervaEndpoint
 from resource_server_async.errors import BaseError
 from resource_server_async.models import (
     AuthService,
@@ -126,6 +127,7 @@ def configure_logging(log_file: str | None = None) -> logging.Logger:
 GATEWAY_HEALTH_TIMEOUT = int(os.getenv("HEALTH_MONITOR_GATEWAY_TIMEOUT", 5))
 GLOBUS_HEALTH_TIMEOUT = int(os.getenv("HEALTH_MONITOR_GLOBUS_TIMEOUT", 30))
 METIS_HEALTH_TIMEOUT = int(os.getenv("HEALTH_MONITOR_METIS_TIMEOUT", 15))
+MINERVA_HEALTH_TIMEOUT = int(os.getenv("HEALTH_MONITOR_MINERVA_TIMEOUT", 15))
 
 FULL_REPORT_FREQUENCY_HOURS = int(os.getenv("HEALTH_MONITOR_FULL_REPORT_HOURS", 24))
 
@@ -632,6 +634,141 @@ async def check_metis_models() -> list[HealthRecord]:
     return records
 
 
+async def extract_minerva_models() -> list[str]:
+    """Flatten Minerva status structure into a list of live model names."""
+
+    minerva = await MinervaCluster.load_adapter("minerva")
+    jobs = await minerva.get_jobs(None)
+
+    return [model.strip() for j in jobs.running for model in j.Models.split(",")]
+
+
+async def check_minerva_models() -> list[HealthRecord]:
+    """Run mTLS route checks for active Minerva models."""
+
+    records: list[HealthRecord] = []
+
+    try:
+        models = await extract_minerva_models()
+    except Exception as e:
+        records.append(
+            HealthRecord(
+                component="Minerva",
+                cluster="minerva",
+                status=HealthStatus.FAILED,
+                detail=str(e),
+            )
+        )
+        return records
+
+    if not models:
+        records.append(
+            HealthRecord(
+                component="Minerva",
+                cluster="minerva",
+                status=HealthStatus.IDLE,
+                detail="No live models returned by Minerva status",
+            )
+        )
+        return records
+
+    for model_name in models:
+        try:
+            endpoint = await MinervaEndpoint.load_adapter("minerva", "api", model_name)
+        except Exception as e:
+            records.append(
+                HealthRecord(
+                    component=model_name,
+                    cluster="minerva",
+                    status=HealthStatus.FAILED,
+                    detail=f"Failed to load endpoint: {e}",
+                )
+            )
+            continue
+
+        url = f"{endpoint.config.api_url.rstrip('/')}/models"
+        log.info("Calling Minerva route check: model=%s url=%s", model_name, url)
+
+        start = time.monotonic()
+        try:
+            payload = await asyncio.wait_for(
+                endpoint.httpx_client.get(url), timeout=MINERVA_HEALTH_TIMEOUT
+            )
+        except httpx.TimeoutException:
+            elapsed = time.monotonic() - start
+            records.append(
+                HealthRecord(
+                    component=model_name,
+                    cluster="minerva",
+                    status=HealthStatus.FAILED,
+                    detail=f"Timeout calling {url}",
+                    elapsed=elapsed,
+                )
+            )
+            continue
+        except httpx.HTTPStatusError as e:
+            elapsed = time.monotonic() - start
+            records.append(
+                HealthRecord(
+                    component=model_name,
+                    cluster="minerva",
+                    status=HealthStatus.FAILED,
+                    detail=f"HTTP {e.response.status_code}: {e.response.text[:256]}",
+                    elapsed=elapsed,
+                )
+            )
+            continue
+        except (httpx.HTTPError, asyncio.TimeoutError) as e:
+            elapsed = time.monotonic() - start
+            records.append(
+                HealthRecord(
+                    component=model_name,
+                    cluster="minerva",
+                    status=HealthStatus.FAILED,
+                    detail=str(e),
+                    elapsed=elapsed,
+                )
+            )
+            continue
+        except Exception as e:
+            elapsed = time.monotonic() - start
+            records.append(
+                HealthRecord(
+                    component=model_name,
+                    cluster="minerva",
+                    status=HealthStatus.FAILED,
+                    detail=f"Unexpected error: {e}",
+                    elapsed=elapsed,
+                )
+            )
+            continue
+
+        elapsed = time.monotonic() - start
+        status = HealthStatus.HEALTHY
+        detail = "ok"
+        if elapsed > SLOW_THRESHOLD_SECONDS:
+            status = HealthStatus.SLOW
+            detail = f"slow: elapsed={format_duration(elapsed)}"
+        else:
+            detail = f"ok (elapsed={format_duration(elapsed)})"
+
+        if not isinstance(payload, dict) or "data" not in payload:
+            status = HealthStatus.FAILED
+            detail = f"Unexpected /models payload: {payload!r}"
+
+        records.append(
+            HealthRecord(
+                component=model_name,
+                cluster="minerva",
+                status=status,
+                detail=detail,
+                elapsed=elapsed,
+            )
+        )
+
+    return records
+
+
 async def check_gateway_health() -> HealthRecord:
     """Check resource_server /health"""
 
@@ -841,6 +978,7 @@ async def run_monitor() -> list[HealthRecord]:
     results = await asyncio.gather(
         check_sophia_models(),
         check_metis_models(),
+        check_minerva_models(),
         asyncio.gather(
             check_gateway_health(),
             check_redis_health(),
@@ -850,7 +988,7 @@ async def run_monitor() -> list[HealthRecord]:
         return_exceptions=True,
     )
 
-    cluster_labels = ["sophia", "metis", "vm"]
+    cluster_labels = ["sophia", "metis", "minerva", "vm"]
     records: list[HealthRecord] = []
     for cluster_name, result in zip(cluster_labels, results):
         if isinstance(result, Exception):

resource_server_async/clusters/__init__.py

@@ -2,5 +2,12 @@
 from .direct_api import DirectAPICluster
 from .globus_compute import GlobusComputeCluster
 from .metis import MetisCluster
+from .minerva import MinervaCluster
 
-__all__ = ["BaseCluster", "GlobusComputeCluster", "DirectAPICluster", "MetisCluster"]
+__all__ = [
+    "BaseCluster",
+    "GlobusComputeCluster",
+    "DirectAPICluster",
+    "MetisCluster",
+    "MinervaCluster",
+]

resource_server_async/clusters/direct_api.py

@@ -14,6 +14,11 @@
 class ClusterConfig(BaseModel):
     status_url: str
     api_request_timeout: int = 10
+    ca_cert_path: str | None = None
+    client_cert_path: str | None = None
+    client_key_path: str | None = None
+    check_hostname: bool = True
+    trust_env: bool = True
 
 
 # Direct API implementation of a BaseCluster
@@ -38,6 +43,11 @@ def __init__(
         # Create HTTPx async client
         self.__httpx_client = AsyncHttpClient(
             timeout=self.__config.api_request_timeout,
+            ca_cert_path=self.__config.ca_cert_path,
+            client_cert_path=self.__config.client_cert_path,
+            client_key_path=self.__config.client_key_path,
+            check_hostname=self.__config.check_hostname,
+            trust_env=self.__config.trust_env,
         )
 
         # Initialize the rest of the common attributes

resource_server_async/clusters/minerva.py

@@ -0,0 +1,152 @@
+import logging
+from typing import Any, List, override
+
+from httpx import HTTPError, TimeoutException
+
+from resource_server_async.cache import cache_item_async, get_item_from_cache_async
+from resource_server_async.clusters.direct_api import DirectAPICluster
+
+from ..errors import GetJobsError
+from ..schemas.clusters import JobInfo, JobsByStatus
+from ..schemas.structured_logs import UserPydantic
+
+log = logging.getLogger(__name__)
+
+
+class MinervaCluster(DirectAPICluster):
+    """Minerva Direct API cluster backed by the login-node status endpoint."""
+
+    def __init__(
+        self,
+        id: str,
+        cluster_name: str,
+        cluster_adapter: str,
+        frameworks: List[str],
+        openai_endpoints: List[str],
+        config: dict[str, Any],
+        allowed_globus_groups: List[str] = [],
+        allowed_domains: List[str] = [],
+    ):
+        super().__init__(
+            id,
+            cluster_name,
+            cluster_adapter,
+            frameworks,
+            openai_endpoints,
+            config=config,
+            allowed_globus_groups=allowed_globus_groups,
+            allowed_domains=allowed_domains,
+        )
+
+    @override
+    async def get_jobs(self, _auth: UserPydantic | None) -> JobsByStatus:
+        cache_key = f"{self.cluster_name}_status_response"
+
+        cached_result: JobsByStatus | None = await get_item_from_cache_async(cache_key)
+        if cached_result is not None:
+            return cached_result
+
+        status = await self._fetch_minerva_status()
+        formatted = self._format_status(status)
+
+        try:
+            await cache_item_async(cache_key, formatted, ttl=60)
+        except Exception as e:
+            log.warning("Failed to cache %s: %s", cache_key, e)
+
+        return formatted
+
+    async def _fetch_minerva_status(self) -> Any:
+        try:
+            return await self.httpx_client.get(self.config.status_url)
+        except TimeoutException:
+            raise GetJobsError(
+                f"Timeout calling {self.config.status_url!r}", status_code=504
+            )
+        except HTTPError as e:
+            raise GetJobsError(
+                f"Unexpected error calling {self.config.status_url!r}: {e}"
+            )
+
+    def _format_status(self, status: Any) -> JobsByStatus:
+        if not isinstance(status, dict):
+            raise GetJobsError("Unexpected response type from Minerva status URL")
+
+        if isinstance(status.get("models"), list):
+            model_entries: list[dict[str, Any]] = [
+                {"status": "Live", "model": model}
+                for model in status["models"]
+                if isinstance(model, str)
+            ]
+        else:
+            model_entries = []
+            for model_info in status.values():
+                if not isinstance(model_info, dict):
+                    raise GetJobsError(
+                        "Unexpected model entry type from Minerva status URL"
+                    )
+                model_entries.append(model_info)
+
+        formatted = JobsByStatus()
+        formatted.cluster_status = {
+            "cluster": self.cluster_name,
+            "total_models": len(model_entries),
+            "live_models": 0,
+            "stopped_models": 0,
+        }
+
+        for model_info in model_entries:
+            raw_status = str(model_info.get("status", "Unknown"))
+            model_name = str(model_info.get("model") or "").strip()
+            if not model_name:
+                continue
+
+            model_version = str(model_info.get("model_version") or "")
+            description = str(
+                model_info.get("description")
+                or self._build_description(model_name, model_version, model_info)
+            )
+            normalized_status = (
+                "running" if raw_status.lower() == "live" else raw_status.lower()
+            )
+
+            job_entry = JobInfo(
+                **{
+                    "Models": model_name,
+                    "Framework": "api",
+                    "Cluster": self.cluster_name,
+                    "Model Status": normalized_status,
+                    "Description": description,
+                    "Model Version": model_version,
+                }
+            )
+
+            if raw_status.lower() == "live":
+                formatted.running.append(job_entry)
+                formatted.cluster_status["live_models"] += 1
+            elif raw_status.lower() == "stopped":
+                formatted.stopped.append(job_entry)
+                formatted.cluster_status["stopped_models"] += 1
+            else:
+                formatted.queued.append(job_entry)
+
+        return formatted
+
+    def _build_description(
+        self, model_name: str, model_version: str, model_info: dict[str, Any]
+    ) -> str:
+        route_prefix = model_info.get("route_prefix")
+        pbs_job_id = model_info.get("pbs_job_id")
+        node = model_info.get("node")
+
+        parts = [model_version or model_name, "on Minerva"]
+        safe_details = []
+        if isinstance(route_prefix, str) and route_prefix.startswith("/models/"):
+            safe_details.append(route_prefix)
+        if isinstance(pbs_job_id, str):
+            safe_details.append(f"job {pbs_job_id}")
+        if isinstance(node, str):
+            safe_details.append(node)
+        if safe_details:
+            parts.append(f"({', '.join(safe_details)})")
+        return " ".join(parts)

resource_server_async/endpoints/__init__.py

@@ -2,10 +2,12 @@
 from .endpoint import BaseEndpoint
 from .globus_compute import GlobusComputeEndpoint
 from .metis import MetisEndpoint
+from .minerva import MinervaEndpoint
 
 __all__ = [
     "BaseEndpoint",
     "GlobusComputeEndpoint",
     "DirectAPIEndpoint",
     "MetisEndpoint",
+    "MinervaEndpoint",
 ]