Coverage for gco / services / health_monitor.py: 100.00%
350 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""
2Health Monitor Service for GCO (Global Capacity Orchestrator on AWS).
4This service monitors Kubernetes cluster resource utilization and reports
5health status for load balancer health checks and monitoring dashboards.
7Key Features:
8- Collects CPU, memory, and GPU utilization metrics from Kubernetes Metrics Server
9- Compares utilization against configurable thresholds
10- Reports health status (healthy/unhealthy) based on threshold violations
11- Caches metrics to reduce API calls to Kubernetes
13Environment Variables:
14 CLUSTER_NAME: Name of the EKS cluster being monitored
15 REGION: AWS region of the cluster
16 CPU_THRESHOLD: CPU utilization threshold percentage (default: 80, -1 to disable)
17 MEMORY_THRESHOLD: Memory utilization threshold percentage (default: 80, -1 to disable)
18 GPU_THRESHOLD: GPU utilization threshold percentage (default: 60, -1 to disable)
20Usage:
21 health_monitor = create_health_monitor_from_env()
22 status = await health_monitor.get_health_status()
23"""
25import asyncio
26import logging
27import os
28from datetime import UTC, datetime
29from typing import Any, Literal
31import boto3
32from botocore.config import Config
33from botocore.exceptions import ClientError
34from kubernetes import client, config
35from kubernetes.client.rest import ApiException
37from gco.k8s_api_versions import METRICS_API_GROUP, METRICS_API_VERSION
38from gco.models import HealthStatus, RequestedResources, ResourceThresholds, ResourceUtilization
39from gco.services.leader_lease import (
40 LEASE_MIN_DURATION_SECONDS,
41 LEASE_REQUEST_TIMEOUT,
42 LeaseIdentity,
43 try_acquire_lease,
44)
45from gco.services.structured_logging import configure_structured_logging
47# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
48# Generated at (UTC): 2026-10-04T13:42:40Z
49# Generated from Git commit: c8a2c18a0d540e23acd5eda86d3aab9db69e5c41
50# Flowchart(s) generated from this file:
51# * ``HealthMonitor.get_health_status`` -> ``diagrams/code_diagrams/gco/services/health_monitor.HealthMonitor_get_health_status.html``
52# (PNG: ``diagrams/code_diagrams/gco/services/health_monitor.HealthMonitor_get_health_status.png``)
53# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
54# <pyflowchart-code-diagram> END
57logging.basicConfig(
58 level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
59)
60logger = logging.getLogger(__name__)
62# Kept as module names for callers/tests that import them; the values live in
63# gco.services.leader_lease so every election shares one floor and timeout.
64_ALB_SYNC_LEASE_MIN_SECONDS = LEASE_MIN_DURATION_SECONDS
65_ALB_SYNC_K8S_TIMEOUT = LEASE_REQUEST_TIMEOUT
66_ALB_SYNC_SSM_CONFIG = Config(
67 connect_timeout=3,
68 read_timeout=10,
69 retries={"total_max_attempts": 2, "mode": "standard"},
70)
73class HealthMonitor:
74 """
75 Monitors Kubernetes cluster resource utilization and determines health status
77 Every Kubernetes read goes through ``asyncio.to_thread``: the official
78 client is synchronous, and a blocking call inside the poll task stalls the
79 whole event loop — the ``/healthz`` and ``/readyz`` the kubelet probes, and
80 the socket bind uvicorn performs right after the lifespan yields. With an
81 unreachable API server the client's connect timeout and retries add up to
82 minutes, which is exactly the startup-probe budget: the pod was killed for
83 "not starting" while it was really waiting on the API server (seen on kind,
84 where NetworkPolicy blocked the :6443 endpoint). Off the loop, the service
85 keeps serving its probes and its cached status regardless of how the
86 cluster reads are doing.
87 """
89 def __init__(self, cluster_id: str, region: str, thresholds: ResourceThresholds):
90 self.cluster_id = cluster_id
91 self.region = region
92 self.thresholds = thresholds
94 # Initialize Kubernetes clients
95 try:
96 # Try to load in-cluster config first (when running in pod)
97 config.load_incluster_config()
98 logger.info("Loaded in-cluster Kubernetes configuration")
99 except config.ConfigException:
100 try:
101 # Fall back to local kubeconfig (for development)
102 config.load_kube_config()
103 logger.info("Loaded local Kubernetes configuration")
104 except config.ConfigException as e:
105 logger.error(f"Failed to load Kubernetes configuration: {e}")
106 raise
108 self.core_v1 = client.CoreV1Api()
109 self.networking_v1 = client.NetworkingV1Api()
110 self.coordination_v1 = client.CoordinationV1Api()
111 self.metrics_v1beta1 = client.CustomObjectsApi()
113 # Timeout for Kubernetes API calls (seconds)
114 self._k8s_timeout = int(os.environ.get("K8S_API_TIMEOUT", "30"))
116 # Cache for metrics
117 self._last_metrics_time: datetime | None = None
118 self._cached_metrics: dict[str, Any] | None = None
119 self._cache_duration = 30 # seconds
121 # ALB hostname sync. Every replica keeps serving health endpoints, but
122 # only the holder of this Kubernetes Lease may perform the mutating SSM
123 # reconciliation. The Lease is pre-created by 02-rbac.yaml so RBAC can
124 # grant update on one exact resource instead of create on every Lease in
125 # gco-system.
126 self._last_alb_sync: datetime | None = None
127 self._alb_sync_interval = 300 # 5 minutes
128 self._alb_sync_lease_name = os.environ.get(
129 "ALB_SYNC_LEASE_NAME", "gco-health-monitor-alb-sync"
130 )
131 self._alb_sync_lease_namespace = os.environ.get("POD_NAMESPACE", "gco-system")
132 configured_lease_duration = int(os.environ.get("ALB_SYNC_LEASE_DURATION", "90"))
133 self._alb_sync_lease_duration = max(configured_lease_duration, _ALB_SYNC_LEASE_MIN_SECONDS)
134 if configured_lease_duration < _ALB_SYNC_LEASE_MIN_SECONDS:
135 logger.warning(
136 "ALB_SYNC_LEASE_DURATION=%s is too short; enforcing %s seconds",
137 configured_lease_duration,
138 _ALB_SYNC_LEASE_MIN_SECONDS,
139 )
140 self._alb_sync_holder = (
141 os.environ.get("POD_NAME")
142 or os.environ.get("HOSTNAME")
143 or f"health-monitor-{os.getpid()}"
144 )
146 async def get_cluster_metrics(self) -> tuple[ResourceUtilization, int, int, RequestedResources]:
147 """
148 Get current cluster resource utilization metrics
149 Returns: (ResourceUtilization, active_jobs_count, pending_pods_count, pending_requested_resources)
150 """
151 try:
152 # Get node metrics from metrics server
153 node_metrics = await self._get_node_metrics()
155 # Get pod metrics for active jobs count and pending pods
156 active_jobs, pending_pods = await self._get_pod_counts()
158 # Calculate cluster-wide utilization (each helper lists the nodes)
159 cpu_utilization = await asyncio.to_thread(self._calculate_cpu_utilization, node_metrics)
160 memory_utilization = await asyncio.to_thread(
161 self._calculate_memory_utilization, node_metrics
162 )
163 gpu_utilization = await self._calculate_gpu_utilization()
165 # Calculate resources requested by pending pods
166 pending_requested = await self._calculate_pending_requested_resources()
168 resource_utilization = ResourceUtilization(
169 cpu=cpu_utilization, memory=memory_utilization, gpu=gpu_utilization
170 )
172 logger.info(
173 f"Cluster metrics - CPU: {cpu_utilization:.1f}%, "
174 f"Memory: {memory_utilization:.1f}%, GPU: {gpu_utilization:.1f}%, "
175 f"Active Jobs: {active_jobs}, Pending Pods: {pending_pods}, "
176 f"Pending Requested CPU: {pending_requested.cpu_vcpus:.1f} vCPUs, "
177 f"Pending Requested Memory: {pending_requested.memory_gb:.1f} GB"
178 )
180 return resource_utilization, active_jobs, pending_pods, pending_requested
182 except Exception as e:
183 logger.error(f"Failed to get cluster metrics: {e}")
184 # Re-raise so get_health_status returns "unhealthy" instead of
185 # silently reporting 0% utilization (which looks healthy to GA).
186 raise
188 async def _get_node_metrics(self) -> dict[str, Any]:
189 """Get node metrics from Kubernetes metrics server"""
190 try:
191 # Check cache first
192 now = datetime.now(UTC)
193 if (
194 self._cached_metrics
195 and self._last_metrics_time
196 and (now - self._last_metrics_time).seconds < self._cache_duration
197 ):
198 return self._cached_metrics
200 # Fetch fresh metrics
201 node_metrics: dict[str, Any] = await asyncio.to_thread(
202 self.metrics_v1beta1.list_cluster_custom_object,
203 group=METRICS_API_GROUP,
204 version=METRICS_API_VERSION,
205 plural="nodes",
206 _request_timeout=self._k8s_timeout,
207 )
209 # Update cache
210 self._cached_metrics = node_metrics
211 self._last_metrics_time = now
213 return node_metrics
215 except ApiException as e:
216 logger.error(f"Failed to get node metrics: {e}")
217 # Invalidate cache so stale data isn't used on next call
218 self._cached_metrics = None
219 self._last_metrics_time = None
220 # Re-raise so get_cluster_metrics propagates the failure
221 # to get_health_status, which returns "unhealthy"
222 raise
224 def _calculate_cpu_utilization(self, node_metrics: dict[str, Any]) -> float:
225 """Calculate cluster-wide CPU utilization percentage"""
226 total_cpu_usage = 0.0
227 total_cpu_capacity = 0.0
229 try:
230 # Get node list for capacity information
231 nodes = self.core_v1.list_node(_request_timeout=self._k8s_timeout)
232 node_capacities = {}
234 for node in nodes.items:
235 node_name = node.metadata.name
236 cpu_capacity = node.status.allocatable.get("cpu", "0")
237 # Convert CPU capacity to millicores
238 if cpu_capacity.endswith("m"):
239 cpu_capacity_millicores = int(cpu_capacity[:-1])
240 else:
241 cpu_capacity_millicores = int(cpu_capacity) * 1000
242 node_capacities[node_name] = cpu_capacity_millicores
243 total_cpu_capacity += cpu_capacity_millicores
245 # Calculate usage from metrics
246 for item in node_metrics.get("items", []):
247 cpu_usage = item["usage"]["cpu"]
249 # Convert CPU usage to millicores
250 if cpu_usage.endswith("n"):
251 cpu_usage_millicores = int(cpu_usage[:-1]) / 1_000_000
252 elif cpu_usage.endswith("u"):
253 cpu_usage_millicores = int(cpu_usage[:-1]) / 1_000
254 elif cpu_usage.endswith("m"):
255 cpu_usage_millicores = int(cpu_usage[:-1])
256 else:
257 cpu_usage_millicores = int(cpu_usage) * 1000
259 total_cpu_usage += cpu_usage_millicores
261 if total_cpu_capacity > 0:
262 return (total_cpu_usage / total_cpu_capacity) * 100
264 except Exception as e:
265 logger.error(f"Error calculating CPU utilization: {e}")
267 return 0.0
269 def _calculate_memory_utilization(self, node_metrics: dict[str, Any]) -> float:
270 """Calculate cluster-wide memory utilization percentage"""
271 total_memory_usage = 0
272 total_memory_capacity = 0
274 try:
275 # Get node list for capacity information
276 nodes = self.core_v1.list_node(_request_timeout=self._k8s_timeout)
278 for node in nodes.items:
279 memory_capacity = node.status.allocatable.get("memory", "0")
280 # Convert memory capacity to bytes
281 memory_capacity_bytes = self._parse_memory_string(memory_capacity)
282 total_memory_capacity += memory_capacity_bytes
284 # Calculate usage from metrics
285 for item in node_metrics.get("items", []):
286 memory_usage = item["usage"]["memory"]
287 memory_usage_bytes = self._parse_memory_string(memory_usage)
288 total_memory_usage += memory_usage_bytes
290 if total_memory_capacity > 0:
291 return (total_memory_usage / total_memory_capacity) * 100
293 except Exception as e:
294 logger.error(f"Error calculating memory utilization: {e}")
296 return 0.0
298 def _parse_memory_string(self, memory_str: str) -> int:
299 """Parse Kubernetes memory string to bytes"""
300 if not memory_str:
301 return 0
303 memory_str = memory_str.strip()
305 # Handle different units
306 if memory_str.endswith("Ki"):
307 return int(memory_str[:-2]) * 1024
308 if memory_str.endswith("Mi"):
309 return int(memory_str[:-2]) * 1024 * 1024
310 if memory_str.endswith("Gi"):
311 return int(memory_str[:-2]) * 1024 * 1024 * 1024
312 if memory_str.endswith("Ti"):
313 return int(memory_str[:-2]) * 1024 * 1024 * 1024 * 1024
314 if memory_str.endswith("k"):
315 return int(memory_str[:-1]) * 1000
316 if memory_str.endswith("M"):
317 return int(memory_str[:-1]) * 1000 * 1000
318 if memory_str.endswith("G"):
319 return int(memory_str[:-1]) * 1000 * 1000 * 1000
320 return int(memory_str)
322 async def _calculate_gpu_utilization(self) -> float:
323 """Calculate cluster-wide GPU utilization percentage"""
324 try:
325 # Get pods with GPU requests
326 pods = await asyncio.to_thread(
327 self.core_v1.list_pod_for_all_namespaces,
328 _request_timeout=self._k8s_timeout,
329 )
331 total_gpu_requested = 0
332 total_gpu_capacity = 0
334 # Get node GPU capacity
335 nodes = await asyncio.to_thread(
336 self.core_v1.list_node, _request_timeout=self._k8s_timeout
337 )
338 for node in nodes.items:
339 gpu_capacity = node.status.allocatable.get("nvidia.com/gpu", "0")
340 total_gpu_capacity += int(gpu_capacity)
342 # Calculate GPU requests from running pods
343 for pod in pods.items:
344 if pod.status.phase == "Running":
345 for container in pod.spec.containers:
346 if container.resources and container.resources.requests:
347 gpu_request = container.resources.requests.get("nvidia.com/gpu", "0")
348 total_gpu_requested += int(gpu_request)
350 if total_gpu_capacity > 0:
351 return (total_gpu_requested / total_gpu_capacity) * 100
353 except Exception as e:
354 logger.error(f"Error calculating GPU utilization: {e}")
356 return 0.0
358 async def _get_active_jobs_count(self) -> int:
359 """Get count of active jobs in the cluster"""
360 try:
361 # Count running pods (excluding system pods)
362 pods = await asyncio.to_thread(
363 self.core_v1.list_pod_for_all_namespaces,
364 _request_timeout=self._k8s_timeout,
365 )
366 active_jobs = 0
368 for pod in pods.items:
369 # Skip system namespaces
370 if pod.metadata.namespace in ["kube-system", "kube-public", "kube-node-lease"]:
371 continue
373 # Count running pods as active jobs
374 if pod.status.phase == "Running":
375 active_jobs += 1
377 return active_jobs
379 except Exception as e:
380 logger.error(f"Error getting active jobs count: {e}")
381 return 0
383 async def _get_pod_counts(self) -> tuple[int, int]:
384 """Get count of active jobs and pending pods in the cluster"""
385 try:
386 pods = await asyncio.to_thread(
387 self.core_v1.list_pod_for_all_namespaces,
388 _request_timeout=self._k8s_timeout,
389 )
390 active_jobs = 0
391 pending_pods = 0
393 for pod in pods.items:
394 # Skip system namespaces
395 if pod.metadata.namespace in ["kube-system", "kube-public", "kube-node-lease"]:
396 continue
398 if pod.status.phase == "Running":
399 active_jobs += 1
400 elif pod.status.phase == "Pending":
401 pending_pods += 1
403 return active_jobs, pending_pods
405 except Exception as e:
406 logger.error(f"Error getting pod counts: {e}")
407 return 0, 0
409 async def _calculate_pending_requested_resources(self) -> RequestedResources:
410 """Calculate total resources requested by pending pods"""
411 try:
412 pods = await asyncio.to_thread(
413 self.core_v1.list_pod_for_all_namespaces,
414 _request_timeout=self._k8s_timeout,
415 )
416 total_cpu_millicores = 0.0
417 total_memory_bytes = 0
418 total_gpus = 0
420 for pod in pods.items:
421 # Skip system namespaces
422 if pod.metadata.namespace in ["kube-system", "kube-public", "kube-node-lease"]:
423 continue
425 # Only count pending pods
426 if pod.status.phase != "Pending":
427 continue
429 for container in pod.spec.containers:
430 if container.resources and container.resources.requests:
431 # CPU
432 cpu_request = container.resources.requests.get("cpu", "0")
433 if cpu_request.endswith("m"):
434 total_cpu_millicores += int(cpu_request[:-1])
435 elif cpu_request.endswith("n"):
436 total_cpu_millicores += int(cpu_request[:-1]) / 1_000_000
437 else:
438 total_cpu_millicores += float(cpu_request) * 1000
440 # Memory
441 memory_request = container.resources.requests.get("memory", "0")
442 total_memory_bytes += self._parse_memory_string(memory_request)
444 # GPUs
445 gpu_request = container.resources.requests.get("nvidia.com/gpu", "0")
446 total_gpus += int(gpu_request)
448 # Convert to vCPUs and GB
449 cpu_vcpus = total_cpu_millicores / 1000
450 memory_gb = total_memory_bytes / (1024 * 1024 * 1024)
452 return RequestedResources(cpu_vcpus=cpu_vcpus, memory_gb=memory_gb, gpus=total_gpus)
454 except Exception as e:
455 logger.error(f"Error calculating pending requested resources: {e}")
456 return RequestedResources(cpu_vcpus=0.0, memory_gb=0.0, gpus=0)
458 async def get_health_status(self) -> HealthStatus:
459 """
460 Get current health status of the cluster
461 """
462 try:
463 # Get current metrics
464 (
465 resource_utilization,
466 active_jobs,
467 pending_pods,
468 pending_requested,
469 ) = await self.get_cluster_metrics()
471 # Determine health status based on thresholds
472 # A threshold of -1 means that check is disabled
473 is_healthy = True
474 if not self.thresholds.is_disabled("cpu_threshold"):
475 is_healthy = (
476 is_healthy and resource_utilization.cpu <= self.thresholds.cpu_threshold
477 )
478 if not self.thresholds.is_disabled("memory_threshold"):
479 is_healthy = (
480 is_healthy and resource_utilization.memory <= self.thresholds.memory_threshold
481 )
482 if not self.thresholds.is_disabled("gpu_threshold"):
483 is_healthy = (
484 is_healthy and resource_utilization.gpu <= self.thresholds.gpu_threshold
485 )
486 if not self.thresholds.is_disabled("pending_pods_threshold"):
487 is_healthy = is_healthy and pending_pods <= self.thresholds.pending_pods_threshold
488 if not self.thresholds.is_disabled("pending_requested_cpu_vcpus"):
489 is_healthy = (
490 is_healthy
491 and pending_requested.cpu_vcpus <= self.thresholds.pending_requested_cpu_vcpus
492 )
493 if not self.thresholds.is_disabled("pending_requested_memory_gb"):
494 is_healthy = (
495 is_healthy
496 and pending_requested.memory_gb <= self.thresholds.pending_requested_memory_gb
497 )
498 if not self.thresholds.is_disabled("pending_requested_gpus"):
499 is_healthy = (
500 is_healthy and pending_requested.gpus <= self.thresholds.pending_requested_gpus
501 )
503 status: Literal["healthy", "unhealthy"] = "healthy" if is_healthy else "unhealthy"
505 # Generate status message
506 message = None
507 if not is_healthy:
508 violations = []
509 if (
510 not self.thresholds.is_disabled("cpu_threshold")
511 and resource_utilization.cpu > self.thresholds.cpu_threshold
512 ):
513 violations.append(
514 f"CPU: {resource_utilization.cpu:.1f}% > {self.thresholds.cpu_threshold}%"
515 )
516 if (
517 not self.thresholds.is_disabled("memory_threshold")
518 and resource_utilization.memory > self.thresholds.memory_threshold
519 ):
520 violations.append(
521 f"Memory: {resource_utilization.memory:.1f}% > {self.thresholds.memory_threshold}%"
522 )
523 if (
524 not self.thresholds.is_disabled("gpu_threshold")
525 and resource_utilization.gpu > self.thresholds.gpu_threshold
526 ):
527 violations.append(
528 f"GPU: {resource_utilization.gpu:.1f}% > {self.thresholds.gpu_threshold}%"
529 )
530 if (
531 not self.thresholds.is_disabled("pending_pods_threshold")
532 and pending_pods > self.thresholds.pending_pods_threshold
533 ):
534 violations.append(
535 f"Pending Pods: {pending_pods} > {self.thresholds.pending_pods_threshold}"
536 )
537 if (
538 not self.thresholds.is_disabled("pending_requested_cpu_vcpus")
539 and pending_requested.cpu_vcpus > self.thresholds.pending_requested_cpu_vcpus
540 ):
541 violations.append(
542 f"Pending CPU: {pending_requested.cpu_vcpus:.1f} vCPUs > {self.thresholds.pending_requested_cpu_vcpus} vCPUs"
543 )
544 if (
545 not self.thresholds.is_disabled("pending_requested_memory_gb")
546 and pending_requested.memory_gb > self.thresholds.pending_requested_memory_gb
547 ):
548 violations.append(
549 f"Pending Memory: {pending_requested.memory_gb:.1f} GB > {self.thresholds.pending_requested_memory_gb} GB"
550 )
551 if (
552 not self.thresholds.is_disabled("pending_requested_gpus")
553 and pending_requested.gpus > self.thresholds.pending_requested_gpus
554 ):
555 violations.append(
556 f"Pending GPUs: {pending_requested.gpus} > {self.thresholds.pending_requested_gpus}"
557 )
558 message = f"Threshold violations: {', '.join(violations)}"
560 health_status = HealthStatus(
561 cluster_id=self.cluster_id,
562 region=self.region,
563 timestamp=datetime.now(UTC),
564 status=status,
565 resource_utilization=resource_utilization,
566 thresholds=self.thresholds,
567 active_jobs=active_jobs,
568 pending_pods=pending_pods,
569 pending_requested=pending_requested,
570 message=message,
571 )
573 logger.info(f"Health status: {status} - {message or 'All thresholds within limits'}")
574 return health_status
576 except Exception as e:
577 logger.error(f"Error getting health status: {e}")
578 # Return unhealthy status on error
579 return HealthStatus(
580 cluster_id=self.cluster_id,
581 region=self.region,
582 timestamp=datetime.now(UTC),
583 status="unhealthy",
584 resource_utilization=ResourceUtilization(cpu=0.0, memory=0.0, gpu=0.0),
585 thresholds=self.thresholds,
586 active_jobs=0,
587 pending_pods=0,
588 pending_requested=RequestedResources(cpu_vcpus=0.0, memory_gb=0.0, gpus=0),
589 message=f"Health check error: {e!s}",
590 )
592 def _try_acquire_alb_sync_lease(self) -> bool:
593 """Acquire or renew the single-writer Lease for ALB self-healing.
595 Shared election rules live in :mod:`gco.services.leader_lease`:
596 optimistic ``replace`` (a racing writer gets HTTP 409), expired or
597 timestamp-less holders may be replaced, and every API or RBAC failure
598 returns ``False`` — losing self-healing is safer than allowing two
599 replicas to mutate the cross-region SSM parameter.
600 """
601 return try_acquire_lease(
602 self.coordination_v1,
603 LeaseIdentity(
604 name=self._alb_sync_lease_name,
605 namespace=self._alb_sync_lease_namespace,
606 holder=self._alb_sync_holder,
607 duration_seconds=self._alb_sync_lease_duration,
608 ),
609 label="ALB-sync",
610 request_timeout=_ALB_SYNC_K8S_TIMEOUT,
611 )
613 async def sync_alb_registration(self) -> None:
614 """Run ALB self-healing without blocking FastAPI's event loop."""
615 await asyncio.to_thread(self._sync_alb_registration)
617 def _sync_alb_registration(self) -> None:
618 """Ensure the SSM hostname matches the platform Gateway address.
620 Every replica renews or checks the leader Lease on each health loop;
621 only the leader performs reconciliation, at most once every 5 minutes.
622 A second optimistic Lease renewal immediately before ``PutParameter``
623 prevents a stale former leader from writing. The SSM client's bounded
624 retry/timeouts keep that write comfortably inside the Lease duration.
625 """
626 if not self._try_acquire_alb_sync_lease():
627 return
629 now = datetime.now(UTC)
630 if (
631 self._last_alb_sync
632 and (now - self._last_alb_sync).total_seconds() < self._alb_sync_interval
633 ):
634 return
636 self._last_alb_sync = now
638 try:
639 gateway = self.metrics_v1beta1.get_namespaced_custom_object(
640 group="gateway.networking.k8s.io",
641 version="v1",
642 namespace="gco-system",
643 plural="gateways",
644 name="gco-gateway",
645 _request_timeout=self._k8s_timeout,
646 )
647 addresses = gateway.get("status", {}).get("addresses", [])
648 current_hostname = next(
649 (
650 str(address.get("value", "")).strip()
651 for address in addresses
652 if isinstance(address, dict)
653 and address.get("type", "Hostname") == "Hostname"
654 and str(address.get("value", "")).strip()
655 ),
656 None,
657 )
658 if not current_hostname:
659 return
661 global_region = os.environ.get("GLOBAL_REGION", "us-east-2")
662 project_name = os.environ.get("PROJECT_NAME", "gco")
663 param_name = f"/{project_name}/alb-hostname-{self.region}"
664 ssm_client = boto3.client(
665 "ssm",
666 region_name=global_region,
667 config=_ALB_SYNC_SSM_CONFIG,
668 )
669 try:
670 response = ssm_client.get_parameter(Name=param_name)
671 stored_hostname = str(response["Parameter"]["Value"])
672 except ClientError as exc:
673 if exc.response.get("Error", {}).get("Code") != "ParameterNotFound":
674 raise
675 stored_hostname = None
677 if stored_hostname != current_hostname:
678 logger.warning(
679 "ALB hostname mismatch: SSM=%s, Gateway=%s. Updating SSM.",
680 stored_hostname,
681 current_hostname,
682 )
683 # Re-read and optimistically replace the Lease immediately
684 # before the only mutating AWS call. A 409 or a new live holder
685 # makes this replica fail closed.
686 if not self._try_acquire_alb_sync_lease():
687 logger.info("Lost ALB-sync Lease before SSM update; skipping mutation")
688 return
689 ssm_client.put_parameter(
690 Name=param_name,
691 Value=current_hostname,
692 Type="String",
693 Overwrite=True,
694 )
695 logger.info("Updated SSM parameter %s to %s", param_name, current_hostname)
697 except Exception as exc:
698 logger.warning("Gateway ALB sync check failed (non-fatal): %s", exc)
701def create_health_monitor_from_env() -> HealthMonitor:
702 """
703 Create HealthMonitor instance from environment variables
704 """
705 cluster_id = os.getenv("CLUSTER_NAME", "unknown-cluster")
706 region = os.getenv("REGION", "unknown-region")
708 # Load thresholds from environment (defaults match cdk.json)
709 cpu_threshold = int(os.getenv("CPU_THRESHOLD", "80"))
710 memory_threshold = int(os.getenv("MEMORY_THRESHOLD", "80"))
711 gpu_threshold = int(os.getenv("GPU_THRESHOLD", "60"))
712 pending_pods_threshold = int(os.getenv("PENDING_PODS_THRESHOLD", "10"))
713 pending_requested_cpu_vcpus = int(os.getenv("PENDING_REQUESTED_CPU_VCPUS", "100"))
714 pending_requested_memory_gb = int(os.getenv("PENDING_REQUESTED_MEMORY_GB", "200"))
715 pending_requested_gpus = int(os.getenv("PENDING_REQUESTED_GPUS", "8"))
717 thresholds = ResourceThresholds(
718 cpu_threshold=cpu_threshold,
719 memory_threshold=memory_threshold,
720 gpu_threshold=gpu_threshold,
721 pending_pods_threshold=pending_pods_threshold,
722 pending_requested_cpu_vcpus=pending_requested_cpu_vcpus,
723 pending_requested_memory_gb=pending_requested_memory_gb,
724 pending_requested_gpus=pending_requested_gpus,
725 )
727 return HealthMonitor(cluster_id, region, thresholds)
730async def main() -> None:
731 """
732 Main function for running the health monitor with webhook dispatcher.
734 This runs both the health monitoring loop and the webhook dispatcher
735 as concurrent tasks.
736 """
737 from gco.services.webhook_dispatcher import create_webhook_dispatcher_from_env
739 health_monitor = create_health_monitor_from_env()
741 # Enable structured JSON logging for CloudWatch Insights
742 configure_structured_logging(
743 service_name="health-monitor",
744 cluster_id=health_monitor.cluster_id,
745 region=health_monitor.region,
746 )
748 webhook_dispatcher = create_webhook_dispatcher_from_env()
750 # Start webhook dispatcher
751 await webhook_dispatcher.start()
752 logger.info("Webhook dispatcher started")
754 try:
755 while True:
756 try:
757 health_status = await health_monitor.get_health_status()
758 print(f"Health Status: {health_status.status}")
759 print(f"CPU: {health_status.resource_utilization.cpu:.1f}%")
760 print(f"Memory: {health_status.resource_utilization.memory:.1f}%")
761 print(f"GPU: {health_status.resource_utilization.gpu:.1f}%")
762 print(f"Active Jobs: {health_status.active_jobs}")
763 print(f"Pending Pods: {health_status.pending_pods}")
764 if health_status.pending_requested:
765 print(
766 f"Pending Requested CPU: {health_status.pending_requested.cpu_vcpus:.1f} vCPUs"
767 )
768 print(
769 f"Pending Requested Memory: {health_status.pending_requested.memory_gb:.1f} GB"
770 )
771 if health_status.message:
772 print(f"Message: {health_status.message}")
774 # Print webhook dispatcher metrics
775 webhook_metrics = webhook_dispatcher.get_metrics()
776 print(
777 f"Webhook Deliveries: {webhook_metrics['deliveries_total']} "
778 f"(success={webhook_metrics['deliveries_success']}, "
779 f"failed={webhook_metrics['deliveries_failed']})"
780 )
781 print("-" * 50)
783 await asyncio.sleep(30) # Check every 30 seconds
785 except KeyboardInterrupt:
786 raise
787 except Exception as e:
788 logger.error(f"Error in main loop: {e}")
789 await asyncio.sleep(10)
791 except KeyboardInterrupt:
792 logger.info("Health monitor stopped by user")
793 finally:
794 await webhook_dispatcher.stop()
795 logger.info("Webhook dispatcher stopped")
798if __name__ == "__main__":
799 asyncio.run(main())