diff --git a/cron_jobs/direct_health_monitor.py b/cron_jobs/direct_health_monitor.py index beb92de2..092b7f14 100644 --- a/cron_jobs/direct_health_monitor.py +++ b/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): diff --git a/resource_server_async/clusters/__init__.py b/resource_server_async/clusters/__init__.py index 6daf9d0a..559a3140 100644 --- a/resource_server_async/clusters/__init__.py +++ b/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", +] diff --git a/resource_server_async/clusters/direct_api.py b/resource_server_async/clusters/direct_api.py index e9fcc182..b90934af 100644 --- a/resource_server_async/clusters/direct_api.py +++ b/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 diff --git a/resource_server_async/clusters/minerva.py b/resource_server_async/clusters/minerva.py new file mode 100644 index 00000000..f3e017d5 --- /dev/null +++ b/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) diff --git a/resource_server_async/endpoints/__init__.py b/resource_server_async/endpoints/__init__.py index 183840e1..f4f0237b 100644 --- a/resource_server_async/endpoints/__init__.py +++ b/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", ] diff --git a/resource_server_async/endpoints/direct_api.py b/resource_server_async/endpoints/direct_api.py index 0f6c70f5..8c2c701c 100644 --- a/resource_server_async/endpoints/direct_api.py +++ b/resource_server_async/endpoints/direct_api.py @@ -12,7 +12,7 @@ from resource_server_async.endpoints.endpoint import ( BaseEndpoint, ) -from resource_server_async.httpx_client import AsyncHttpClient +from resource_server_async.httpx_client import AsyncHttpClient, create_ssl_context from resource_server_async.streaming import create_streaming_response_headers from ..errors import EndpointError @@ -29,6 +29,11 @@ class DirectAPIEndpointConfig(BaseModel): api_url: str api_key_env_name: str api_request_timeout: int = 120 + 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 class StreamingState(TypedDict): @@ -68,6 +73,11 @@ def __init__( "Content-Type": "application/json", "Authorization": f"Bearer {os.environ.get(self.__config.api_key_env_name, None)}", }, + 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 @@ -204,7 +214,14 @@ async def __get_stream_chunks( # Create an async HTTPx client try: async with httpx.AsyncClient( - timeout=self.config.api_request_timeout + timeout=self.config.api_request_timeout, + verify=create_ssl_context( + 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, ) as client: # Create a streaming client async with client.stream( diff --git a/resource_server_async/endpoints/minerva.py b/resource_server_async/endpoints/minerva.py new file mode 100644 index 00000000..05332860 --- /dev/null +++ b/resource_server_async/endpoints/minerva.py @@ -0,0 +1,54 @@ +import logging +from typing import Any + +from resource_server_async.clusters.minerva import MinervaCluster +from resource_server_async.endpoints.direct_api import DirectAPIEndpoint + +from ..errors import EndpointError +from ..schemas.endpoints import ( + SubmitStreamingTaskResponse, + SubmitTaskResult, +) + +log = logging.getLogger(__name__) + + +class MinervaEndpoint(DirectAPIEndpoint): + """Minerva Direct API endpoint backed by login-node NGINX routes.""" + + async def check_endpoint_status(self) -> bool: + cluster = await MinervaCluster.load_adapter("minerva") + jobs = await cluster.get_jobs(None) + live_models: list[str] = [] + for running in jobs.running: + models = running.Models + if isinstance(models, str): + live_models.extend([model.strip() for model in models.split(",")]) + else: + live_models.extend(models) # type: ignore[unreachable] + + if self.model not in live_models: + raise EndpointError( + f"{self.model!r} is not currently live on Minerva.", status_code=503 + ) + return True + + async def submit_task(self, data: dict[str, Any]) -> SubmitTaskResult: + await self.check_endpoint_status() + api_request_data = {**data["model_params"]} + api_request_data["stream"] = False + api_request_data.pop("api_port", None) + log.info( + f"Making Minerva Direct API call for model {self.model} (stream=False)" + ) + return await super().submit_task(api_request_data) + + async def submit_streaming_task( + self, data: dict[str, Any] + ) -> SubmitStreamingTaskResponse: + await self.check_endpoint_status() + api_request_data = {**data["model_params"]} + api_request_data["stream"] = True + api_request_data.pop("api_port", None) + log.info(f"Making Minerva Direct API call for model {self.model} (stream=True)") + return await super().submit_streaming_task(api_request_data) diff --git a/resource_server_async/httpx_client.py b/resource_server_async/httpx_client.py index c3e7ea27..5db2d62e 100644 --- a/resource_server_async/httpx_client.py +++ b/resource_server_async/httpx_client.py @@ -1,18 +1,53 @@ +import ssl from typing import Any import httpx +def create_ssl_context( + ca_cert_path: str | None = None, + client_cert_path: str | None = None, + client_key_path: str | None = None, + check_hostname: bool = True, +) -> ssl.SSLContext | bool: + """Build an HTTPX verify context for optional mTLS configuration.""" + if not ca_cert_path and not client_cert_path and not client_key_path: + return True + + context = ssl.create_default_context(cafile=ca_cert_path) + context.check_hostname = check_hostname + if client_cert_path or client_key_path: + if not client_cert_path or not client_key_path: + raise ValueError( + "client_cert_path and client_key_path must be configured together" + ) + context.load_cert_chain(certfile=client_cert_path, keyfile=client_key_path) + return context + + class AsyncHttpClient: def __init__( self, timeout: float = 30.0, headers: dict[str, str] | None = None, + 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, ): if headers is None: headers = {"Content-Type": "application/json"} self.headers = headers - self._client = httpx.AsyncClient(timeout=timeout, headers=self.headers) + verify = create_ssl_context( + ca_cert_path=ca_cert_path, + client_cert_path=client_cert_path, + client_key_path=client_key_path, + check_hostname=check_hostname, + ) + self._client = httpx.AsyncClient( + timeout=timeout, headers=self.headers, verify=verify, trust_env=trust_env + ) async def get(self, url: str) -> Any: response = await self._client.get(url)