Coverage for gco/services/health_monitor.py: 93.97%
389 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +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.models import HealthStatus, RequestedResources, ResourceThresholds, ResourceUtilization
38from gco.services.structured_logging import configure_structured_logging
40logging.basicConfig(
41 level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
42)
43logger = logging.getLogger(__name__)
45_ALB_SYNC_LEASE_MIN_SECONDS = 60
46_ALB_SYNC_K8S_TIMEOUT = (3, 10)
47_ALB_SYNC_SSM_CONFIG = Config(
48 connect_timeout=3,
49 read_timeout=10,
50 retries={"total_max_attempts": 2, "mode": "standard"},
51)
54class HealthMonitor:
55 """
56 Monitors Kubernetes cluster resource utilization and determines health status
57 """
59 def __init__(self, cluster_id: str, region: str, thresholds: ResourceThresholds):
60 self.cluster_id = cluster_id
61 self.region = region
62 self.thresholds = thresholds
64 # Initialize Kubernetes clients
65 try:
66 # Try to load in-cluster config first (when running in pod)
67 config.load_incluster_config()
68 logger.info("Loaded in-cluster Kubernetes configuration")
69 except config.ConfigException:
70 try:
71 # Fall back to local kubeconfig (for development)
72 config.load_kube_config()
73 logger.info("Loaded local Kubernetes configuration")
74 except config.ConfigException as e:
75 logger.error(f"Failed to load Kubernetes configuration: {e}")
76 raise
78 self.core_v1 = client.CoreV1Api()
79 self.networking_v1 = client.NetworkingV1Api()
80 self.coordination_v1 = client.CoordinationV1Api()
81 self.metrics_v1beta1 = client.CustomObjectsApi()
83 # Timeout for Kubernetes API calls (seconds)
84 self._k8s_timeout = int(os.environ.get("K8S_API_TIMEOUT", "30"))
86 # Cache for metrics
87 self._last_metrics_time: datetime | None = None
88 self._cached_metrics: dict[str, Any] | None = None
89 self._cache_duration = 30 # seconds
91 # ALB hostname sync. Every replica keeps serving health endpoints, but
92 # only the holder of this Kubernetes Lease may perform the mutating SSM
93 # reconciliation. The Lease is pre-created by 02-rbac.yaml so RBAC can
94 # grant update on one exact resource instead of create on every Lease in
95 # gco-system.
96 self._last_alb_sync: datetime | None = None
97 self._alb_sync_interval = 300 # 5 minutes
98 self._alb_sync_lease_name = os.environ.get(
99 "ALB_SYNC_LEASE_NAME", "gco-health-monitor-alb-sync"
100 )
101 self._alb_sync_lease_namespace = os.environ.get("POD_NAMESPACE", "gco-system")
102 configured_lease_duration = int(os.environ.get("ALB_SYNC_LEASE_DURATION", "90"))
103 self._alb_sync_lease_duration = max(configured_lease_duration, _ALB_SYNC_LEASE_MIN_SECONDS)
104 if configured_lease_duration < _ALB_SYNC_LEASE_MIN_SECONDS: 104 ↛ 105line 104 didn't jump to line 105 because the condition on line 104 was never true
105 logger.warning(
106 "ALB_SYNC_LEASE_DURATION=%s is too short; enforcing %s seconds",
107 configured_lease_duration,
108 _ALB_SYNC_LEASE_MIN_SECONDS,
109 )
110 self._alb_sync_holder = (
111 os.environ.get("POD_NAME")
112 or os.environ.get("HOSTNAME")
113 or f"health-monitor-{os.getpid()}"
114 )
116 async def get_cluster_metrics(self) -> tuple[ResourceUtilization, int, int, RequestedResources]:
117 """
118 Get current cluster resource utilization metrics
119 Returns: (ResourceUtilization, active_jobs_count, pending_pods_count, pending_requested_resources)
120 """
121 try:
122 # Get node metrics from metrics server
123 node_metrics = await self._get_node_metrics()
125 # Get pod metrics for active jobs count and pending pods
126 active_jobs, pending_pods = await self._get_pod_counts()
128 # Calculate cluster-wide utilization
129 cpu_utilization = self._calculate_cpu_utilization(node_metrics)
130 memory_utilization = self._calculate_memory_utilization(node_metrics)
131 gpu_utilization = await self._calculate_gpu_utilization()
133 # Calculate resources requested by pending pods
134 pending_requested = await self._calculate_pending_requested_resources()
136 resource_utilization = ResourceUtilization(
137 cpu=cpu_utilization, memory=memory_utilization, gpu=gpu_utilization
138 )
140 logger.info(
141 f"Cluster metrics - CPU: {cpu_utilization:.1f}%, "
142 f"Memory: {memory_utilization:.1f}%, GPU: {gpu_utilization:.1f}%, "
143 f"Active Jobs: {active_jobs}, Pending Pods: {pending_pods}, "
144 f"Pending Requested CPU: {pending_requested.cpu_vcpus:.1f} vCPUs, "
145 f"Pending Requested Memory: {pending_requested.memory_gb:.1f} GB"
146 )
148 return resource_utilization, active_jobs, pending_pods, pending_requested
150 except Exception as e:
151 logger.error(f"Failed to get cluster metrics: {e}")
152 # Re-raise so get_health_status returns "unhealthy" instead of
153 # silently reporting 0% utilization (which looks healthy to GA).
154 raise
156 async def _get_node_metrics(self) -> dict[str, Any]:
157 """Get node metrics from Kubernetes metrics server"""
158 try:
159 # Check cache first
160 now = datetime.now()
161 if (
162 self._cached_metrics
163 and self._last_metrics_time
164 and (now - self._last_metrics_time).seconds < self._cache_duration
165 ):
166 return self._cached_metrics
168 # Fetch fresh metrics
169 node_metrics: dict[str, Any] = self.metrics_v1beta1.list_cluster_custom_object(
170 group="metrics.k8s.io",
171 version="v1beta1",
172 plural="nodes",
173 _request_timeout=self._k8s_timeout,
174 )
176 # Update cache
177 self._cached_metrics = node_metrics
178 self._last_metrics_time = now
180 return node_metrics
182 except ApiException as e:
183 logger.error(f"Failed to get node metrics: {e}")
184 # Invalidate cache so stale data isn't used on next call
185 self._cached_metrics = None
186 self._last_metrics_time = None
187 # Re-raise so get_cluster_metrics propagates the failure
188 # to get_health_status, which returns "unhealthy"
189 raise
191 def _calculate_cpu_utilization(self, node_metrics: dict[str, Any]) -> float:
192 """Calculate cluster-wide CPU utilization percentage"""
193 total_cpu_usage = 0.0
194 total_cpu_capacity = 0.0
196 try:
197 # Get node list for capacity information
198 nodes = self.core_v1.list_node(_request_timeout=self._k8s_timeout)
199 node_capacities = {}
201 for node in nodes.items:
202 node_name = node.metadata.name
203 cpu_capacity = node.status.allocatable.get("cpu", "0")
204 # Convert CPU capacity to millicores
205 if cpu_capacity.endswith("m"):
206 cpu_capacity_millicores = int(cpu_capacity[:-1])
207 else:
208 cpu_capacity_millicores = int(cpu_capacity) * 1000
209 node_capacities[node_name] = cpu_capacity_millicores
210 total_cpu_capacity += cpu_capacity_millicores
212 # Calculate usage from metrics
213 for item in node_metrics.get("items", []):
214 cpu_usage = item["usage"]["cpu"]
216 # Convert CPU usage to millicores
217 if cpu_usage.endswith("n"):
218 cpu_usage_millicores = int(cpu_usage[:-1]) / 1_000_000
219 elif cpu_usage.endswith("u"):
220 cpu_usage_millicores = int(cpu_usage[:-1]) / 1_000
221 elif cpu_usage.endswith("m"):
222 cpu_usage_millicores = int(cpu_usage[:-1])
223 else:
224 cpu_usage_millicores = int(cpu_usage) * 1000
226 total_cpu_usage += cpu_usage_millicores
228 if total_cpu_capacity > 0:
229 return (total_cpu_usage / total_cpu_capacity) * 100
231 except Exception as e:
232 logger.error(f"Error calculating CPU utilization: {e}")
234 return 0.0
236 def _calculate_memory_utilization(self, node_metrics: dict[str, Any]) -> float:
237 """Calculate cluster-wide memory utilization percentage"""
238 total_memory_usage = 0
239 total_memory_capacity = 0
241 try:
242 # Get node list for capacity information
243 nodes = self.core_v1.list_node(_request_timeout=self._k8s_timeout)
245 for node in nodes.items:
246 memory_capacity = node.status.allocatable.get("memory", "0")
247 # Convert memory capacity to bytes
248 memory_capacity_bytes = self._parse_memory_string(memory_capacity)
249 total_memory_capacity += memory_capacity_bytes
251 # Calculate usage from metrics
252 for item in node_metrics.get("items", []):
253 memory_usage = item["usage"]["memory"]
254 memory_usage_bytes = self._parse_memory_string(memory_usage)
255 total_memory_usage += memory_usage_bytes
257 if total_memory_capacity > 0:
258 return (total_memory_usage / total_memory_capacity) * 100
260 except Exception as e:
261 logger.error(f"Error calculating memory utilization: {e}")
263 return 0.0
265 def _parse_memory_string(self, memory_str: str) -> int:
266 """Parse Kubernetes memory string to bytes"""
267 if not memory_str:
268 return 0
270 memory_str = memory_str.strip()
272 # Handle different units
273 if memory_str.endswith("Ki"):
274 return int(memory_str[:-2]) * 1024
275 if memory_str.endswith("Mi"):
276 return int(memory_str[:-2]) * 1024 * 1024
277 if memory_str.endswith("Gi"):
278 return int(memory_str[:-2]) * 1024 * 1024 * 1024
279 if memory_str.endswith("Ti"):
280 return int(memory_str[:-2]) * 1024 * 1024 * 1024 * 1024
281 if memory_str.endswith("k"):
282 return int(memory_str[:-1]) * 1000
283 if memory_str.endswith("M"):
284 return int(memory_str[:-1]) * 1000 * 1000
285 if memory_str.endswith("G"):
286 return int(memory_str[:-1]) * 1000 * 1000 * 1000
287 return int(memory_str)
289 async def _calculate_gpu_utilization(self) -> float:
290 """Calculate cluster-wide GPU utilization percentage"""
291 try:
292 # Get pods with GPU requests
293 pods = self.core_v1.list_pod_for_all_namespaces(
294 _request_timeout=self._k8s_timeout,
295 )
297 total_gpu_requested = 0
298 total_gpu_capacity = 0
300 # Get node GPU capacity
301 nodes = self.core_v1.list_node(_request_timeout=self._k8s_timeout)
302 for node in nodes.items:
303 gpu_capacity = node.status.allocatable.get("nvidia.com/gpu", "0")
304 total_gpu_capacity += int(gpu_capacity)
306 # Calculate GPU requests from running pods
307 for pod in pods.items:
308 if pod.status.phase == "Running":
309 for container in pod.spec.containers:
310 if container.resources and container.resources.requests:
311 gpu_request = container.resources.requests.get("nvidia.com/gpu", "0")
312 total_gpu_requested += int(gpu_request)
314 if total_gpu_capacity > 0:
315 return (total_gpu_requested / total_gpu_capacity) * 100
317 except Exception as e:
318 logger.error(f"Error calculating GPU utilization: {e}")
320 return 0.0
322 async def _get_active_jobs_count(self) -> int:
323 """Get count of active jobs in the cluster"""
324 try:
325 # Count running pods (excluding system pods)
326 pods = self.core_v1.list_pod_for_all_namespaces(
327 _request_timeout=self._k8s_timeout,
328 )
329 active_jobs = 0
331 for pod in pods.items:
332 # Skip system namespaces
333 if pod.metadata.namespace in ["kube-system", "kube-public", "kube-node-lease"]:
334 continue
336 # Count running pods as active jobs
337 if pod.status.phase == "Running":
338 active_jobs += 1
340 return active_jobs
342 except Exception as e:
343 logger.error(f"Error getting active jobs count: {e}")
344 return 0
346 async def _get_pod_counts(self) -> tuple[int, int]:
347 """Get count of active jobs and pending pods in the cluster"""
348 try:
349 pods = self.core_v1.list_pod_for_all_namespaces(
350 _request_timeout=self._k8s_timeout,
351 )
352 active_jobs = 0
353 pending_pods = 0
355 for pod in pods.items:
356 # Skip system namespaces
357 if pod.metadata.namespace in ["kube-system", "kube-public", "kube-node-lease"]:
358 continue
360 if pod.status.phase == "Running":
361 active_jobs += 1
362 elif pod.status.phase == "Pending": 362 ↛ 355line 362 didn't jump to line 355 because the condition on line 362 was always true
363 pending_pods += 1
365 return active_jobs, pending_pods
367 except Exception as e:
368 logger.error(f"Error getting pod counts: {e}")
369 return 0, 0
371 async def _calculate_pending_requested_resources(self) -> RequestedResources:
372 """Calculate total resources requested by pending pods"""
373 try:
374 pods = self.core_v1.list_pod_for_all_namespaces(
375 _request_timeout=self._k8s_timeout,
376 )
377 total_cpu_millicores = 0.0
378 total_memory_bytes = 0
379 total_gpus = 0
381 for pod in pods.items:
382 # Skip system namespaces
383 if pod.metadata.namespace in ["kube-system", "kube-public", "kube-node-lease"]:
384 continue
386 # Only count pending pods
387 if pod.status.phase != "Pending":
388 continue
390 for container in pod.spec.containers:
391 if container.resources and container.resources.requests: 391 ↛ 390line 391 didn't jump to line 390 because the condition on line 391 was always true
392 # CPU
393 cpu_request = container.resources.requests.get("cpu", "0")
394 if cpu_request.endswith("m"):
395 total_cpu_millicores += int(cpu_request[:-1])
396 elif cpu_request.endswith("n"):
397 total_cpu_millicores += int(cpu_request[:-1]) / 1_000_000
398 else:
399 total_cpu_millicores += float(cpu_request) * 1000
401 # Memory
402 memory_request = container.resources.requests.get("memory", "0")
403 total_memory_bytes += self._parse_memory_string(memory_request)
405 # GPUs
406 gpu_request = container.resources.requests.get("nvidia.com/gpu", "0")
407 total_gpus += int(gpu_request)
409 # Convert to vCPUs and GB
410 cpu_vcpus = total_cpu_millicores / 1000
411 memory_gb = total_memory_bytes / (1024 * 1024 * 1024)
413 return RequestedResources(cpu_vcpus=cpu_vcpus, memory_gb=memory_gb, gpus=total_gpus)
415 except Exception as e:
416 logger.error(f"Error calculating pending requested resources: {e}")
417 return RequestedResources(cpu_vcpus=0.0, memory_gb=0.0, gpus=0)
419 async def get_health_status(self) -> HealthStatus:
420 """
421 Get current health status of the cluster
422 """
423 try:
424 # Get current metrics
425 (
426 resource_utilization,
427 active_jobs,
428 pending_pods,
429 pending_requested,
430 ) = await self.get_cluster_metrics()
432 # Determine health status based on thresholds
433 # A threshold of -1 means that check is disabled
434 is_healthy = True
435 if not self.thresholds.is_disabled("cpu_threshold"): 435 ↛ 439line 435 didn't jump to line 439 because the condition on line 435 was always true
436 is_healthy = (
437 is_healthy and resource_utilization.cpu <= self.thresholds.cpu_threshold
438 )
439 if not self.thresholds.is_disabled("memory_threshold"): 439 ↛ 443line 439 didn't jump to line 443 because the condition on line 439 was always true
440 is_healthy = (
441 is_healthy and resource_utilization.memory <= self.thresholds.memory_threshold
442 )
443 if not self.thresholds.is_disabled("gpu_threshold"): 443 ↛ 447line 443 didn't jump to line 447 because the condition on line 443 was always true
444 is_healthy = (
445 is_healthy and resource_utilization.gpu <= self.thresholds.gpu_threshold
446 )
447 if not self.thresholds.is_disabled("pending_pods_threshold"): 447 ↛ 449line 447 didn't jump to line 449 because the condition on line 447 was always true
448 is_healthy = is_healthy and pending_pods <= self.thresholds.pending_pods_threshold
449 if not self.thresholds.is_disabled("pending_requested_cpu_vcpus"): 449 ↛ 454line 449 didn't jump to line 454 because the condition on line 449 was always true
450 is_healthy = (
451 is_healthy
452 and pending_requested.cpu_vcpus <= self.thresholds.pending_requested_cpu_vcpus
453 )
454 if not self.thresholds.is_disabled("pending_requested_memory_gb"): 454 ↛ 459line 454 didn't jump to line 459 because the condition on line 454 was always true
455 is_healthy = (
456 is_healthy
457 and pending_requested.memory_gb <= self.thresholds.pending_requested_memory_gb
458 )
459 if not self.thresholds.is_disabled("pending_requested_gpus"): 459 ↛ 464line 459 didn't jump to line 464 because the condition on line 459 was always true
460 is_healthy = (
461 is_healthy and pending_requested.gpus <= self.thresholds.pending_requested_gpus
462 )
464 status: Literal["healthy", "unhealthy"] = "healthy" if is_healthy else "unhealthy"
466 # Generate status message
467 message = None
468 if not is_healthy:
469 violations = []
470 if (
471 not self.thresholds.is_disabled("cpu_threshold")
472 and resource_utilization.cpu > self.thresholds.cpu_threshold
473 ):
474 violations.append(
475 f"CPU: {resource_utilization.cpu:.1f}% > {self.thresholds.cpu_threshold}%"
476 )
477 if (
478 not self.thresholds.is_disabled("memory_threshold")
479 and resource_utilization.memory > self.thresholds.memory_threshold
480 ):
481 violations.append(
482 f"Memory: {resource_utilization.memory:.1f}% > {self.thresholds.memory_threshold}%"
483 )
484 if (
485 not self.thresholds.is_disabled("gpu_threshold")
486 and resource_utilization.gpu > self.thresholds.gpu_threshold
487 ):
488 violations.append(
489 f"GPU: {resource_utilization.gpu:.1f}% > {self.thresholds.gpu_threshold}%"
490 )
491 if (
492 not self.thresholds.is_disabled("pending_pods_threshold")
493 and pending_pods > self.thresholds.pending_pods_threshold
494 ):
495 violations.append(
496 f"Pending Pods: {pending_pods} > {self.thresholds.pending_pods_threshold}"
497 )
498 if (
499 not self.thresholds.is_disabled("pending_requested_cpu_vcpus")
500 and pending_requested.cpu_vcpus > self.thresholds.pending_requested_cpu_vcpus
501 ):
502 violations.append(
503 f"Pending CPU: {pending_requested.cpu_vcpus:.1f} vCPUs > {self.thresholds.pending_requested_cpu_vcpus} vCPUs"
504 )
505 if (
506 not self.thresholds.is_disabled("pending_requested_memory_gb")
507 and pending_requested.memory_gb > self.thresholds.pending_requested_memory_gb
508 ):
509 violations.append(
510 f"Pending Memory: {pending_requested.memory_gb:.1f} GB > {self.thresholds.pending_requested_memory_gb} GB"
511 )
512 if (
513 not self.thresholds.is_disabled("pending_requested_gpus")
514 and pending_requested.gpus > self.thresholds.pending_requested_gpus
515 ):
516 violations.append(
517 f"Pending GPUs: {pending_requested.gpus} > {self.thresholds.pending_requested_gpus}"
518 )
519 message = f"Threshold violations: {', '.join(violations)}"
521 health_status = HealthStatus(
522 cluster_id=self.cluster_id,
523 region=self.region,
524 timestamp=datetime.now(),
525 status=status,
526 resource_utilization=resource_utilization,
527 thresholds=self.thresholds,
528 active_jobs=active_jobs,
529 pending_pods=pending_pods,
530 pending_requested=pending_requested,
531 message=message,
532 )
534 logger.info(f"Health status: {status} - {message or 'All thresholds within limits'}")
535 return health_status
537 except Exception as e:
538 logger.error(f"Error getting health status: {e}")
539 # Return unhealthy status on error
540 return HealthStatus(
541 cluster_id=self.cluster_id,
542 region=self.region,
543 timestamp=datetime.now(),
544 status="unhealthy",
545 resource_utilization=ResourceUtilization(cpu=0.0, memory=0.0, gpu=0.0),
546 thresholds=self.thresholds,
547 active_jobs=0,
548 pending_pods=0,
549 pending_requested=RequestedResources(cpu_vcpus=0.0, memory_gb=0.0, gpus=0),
550 message=f"Health check error: {e!s}",
551 )
553 def _try_acquire_alb_sync_lease(self) -> bool:
554 """Acquire or renew the single-writer Lease for ALB self-healing.
556 ``replace_namespaced_lease`` carries the resourceVersion returned by
557 the read, so Kubernetes rejects a racing writer with HTTP 409. API or
558 RBAC failures return ``False``: losing self-healing is safer than
559 allowing two replicas to mutate the cross-region SSM parameter.
560 """
561 observed_at = datetime.now(UTC)
563 try:
564 lease = self.coordination_v1.read_namespaced_lease(
565 self._alb_sync_lease_name,
566 self._alb_sync_lease_namespace,
567 _request_timeout=_ALB_SYNC_K8S_TIMEOUT,
568 )
569 spec = lease.spec
570 current_holder = spec.holder_identity
571 renew_time = spec.renew_time
572 lease_duration = spec.lease_duration_seconds or self._alb_sync_lease_duration
574 expired = False
575 if current_holder: 575 ↛ 586line 575 didn't jump to line 586 because the condition on line 575 was always true
576 if renew_time is None: 576 ↛ 580line 576 didn't jump to line 580 because the condition on line 576 was never true
577 # A holder without a renewal timestamp cannot prove it still
578 # owns the lease. Treat it as expired so the named Lease
579 # cannot remain wedged indefinitely.
580 expired = True
581 else:
582 if renew_time.tzinfo is None: 582 ↛ 583line 582 didn't jump to line 583 because the condition on line 582 was never true
583 renew_time = renew_time.replace(tzinfo=UTC)
584 expired = (observed_at - renew_time).total_seconds() >= lease_duration
586 if current_holder not in (None, "", self._alb_sync_holder) and not expired:
587 return False
589 acquiring = current_holder != self._alb_sync_holder
590 renewed_at = datetime.now(UTC)
591 if acquiring:
592 spec.holder_identity = self._alb_sync_holder
593 spec.acquire_time = renewed_at
594 spec.lease_transitions = (spec.lease_transitions or 0) + 1
595 spec.lease_duration_seconds = self._alb_sync_lease_duration
596 spec.renew_time = renewed_at
598 try:
599 self.coordination_v1.replace_namespaced_lease(
600 self._alb_sync_lease_name,
601 self._alb_sync_lease_namespace,
602 lease,
603 _request_timeout=_ALB_SYNC_K8S_TIMEOUT,
604 )
605 except ApiException as exc:
606 if exc.status == 409: 606 ↛ 609line 606 didn't jump to line 609 because the condition on line 606 was always true
607 logger.debug("Lost ALB-sync Lease race to another health-monitor replica")
608 return False
609 raise
611 if acquiring:
612 logger.info("Acquired ALB-sync leader Lease as %s", self._alb_sync_holder)
613 return True
615 except ApiException as exc:
616 if exc.status == 404: 616 ↛ 623line 616 didn't jump to line 623 because the condition on line 616 was always true
617 logger.warning(
618 "ALB-sync Lease %s/%s is missing; self-healing is disabled until it is restored",
619 self._alb_sync_lease_namespace,
620 self._alb_sync_lease_name,
621 )
622 else:
623 logger.warning("ALB-sync Lease check failed (non-fatal): %s", exc)
624 return False
625 except Exception as exc:
626 logger.warning("ALB-sync Lease check failed (non-fatal): %s", exc)
627 return False
629 async def sync_alb_registration(self) -> None:
630 """Run ALB self-healing without blocking FastAPI's event loop."""
631 await asyncio.to_thread(self._sync_alb_registration)
633 def _sync_alb_registration(self) -> None:
634 """Ensure the SSM hostname matches the platform Gateway address.
636 Every replica renews or checks the leader Lease on each health loop;
637 only the leader performs reconciliation, at most once every 5 minutes.
638 A second optimistic Lease renewal immediately before ``PutParameter``
639 prevents a stale former leader from writing. The SSM client's bounded
640 retry/timeouts keep that write comfortably inside the Lease duration.
641 """
642 if not self._try_acquire_alb_sync_lease():
643 return
645 now = datetime.now()
646 if (
647 self._last_alb_sync
648 and (now - self._last_alb_sync).total_seconds() < self._alb_sync_interval
649 ):
650 return
652 self._last_alb_sync = now
654 try:
655 gateway = self.metrics_v1beta1.get_namespaced_custom_object(
656 group="gateway.networking.k8s.io",
657 version="v1",
658 namespace="gco-system",
659 plural="gateways",
660 name="gco-gateway",
661 _request_timeout=self._k8s_timeout,
662 )
663 addresses = gateway.get("status", {}).get("addresses", [])
664 current_hostname = next(
665 (
666 str(address.get("value", "")).strip()
667 for address in addresses
668 if isinstance(address, dict)
669 and address.get("type", "Hostname") == "Hostname"
670 and str(address.get("value", "")).strip()
671 ),
672 None,
673 )
674 if not current_hostname:
675 return
677 global_region = os.environ.get("GLOBAL_REGION", "us-east-2")
678 project_name = os.environ.get("PROJECT_NAME", "gco")
679 param_name = f"/{project_name}/alb-hostname-{self.region}"
680 ssm_client = boto3.client(
681 "ssm",
682 region_name=global_region,
683 config=_ALB_SYNC_SSM_CONFIG,
684 )
685 try:
686 response = ssm_client.get_parameter(Name=param_name)
687 stored_hostname = str(response["Parameter"]["Value"])
688 except ClientError as exc:
689 if exc.response.get("Error", {}).get("Code") != "ParameterNotFound":
690 raise
691 stored_hostname = None
693 if stored_hostname != current_hostname:
694 logger.warning(
695 "ALB hostname mismatch: SSM=%s, Gateway=%s. Updating SSM.",
696 stored_hostname,
697 current_hostname,
698 )
699 # Re-read and optimistically replace the Lease immediately
700 # before the only mutating AWS call. A 409 or a new live holder
701 # makes this replica fail closed.
702 if not self._try_acquire_alb_sync_lease():
703 logger.info("Lost ALB-sync Lease before SSM update; skipping mutation")
704 return
705 ssm_client.put_parameter(
706 Name=param_name,
707 Value=current_hostname,
708 Type="String",
709 Overwrite=True,
710 )
711 logger.info("Updated SSM parameter %s to %s", param_name, current_hostname)
713 except Exception as exc:
714 logger.warning("Gateway ALB sync check failed (non-fatal): %s", exc)
717def create_health_monitor_from_env() -> HealthMonitor:
718 """
719 Create HealthMonitor instance from environment variables
720 """
721 cluster_id = os.getenv("CLUSTER_NAME", "unknown-cluster")
722 region = os.getenv("REGION", "unknown-region")
724 # Load thresholds from environment (defaults match cdk.json)
725 cpu_threshold = int(os.getenv("CPU_THRESHOLD", "80"))
726 memory_threshold = int(os.getenv("MEMORY_THRESHOLD", "80"))
727 gpu_threshold = int(os.getenv("GPU_THRESHOLD", "60"))
728 pending_pods_threshold = int(os.getenv("PENDING_PODS_THRESHOLD", "10"))
729 pending_requested_cpu_vcpus = int(os.getenv("PENDING_REQUESTED_CPU_VCPUS", "100"))
730 pending_requested_memory_gb = int(os.getenv("PENDING_REQUESTED_MEMORY_GB", "200"))
731 pending_requested_gpus = int(os.getenv("PENDING_REQUESTED_GPUS", "8"))
733 thresholds = ResourceThresholds(
734 cpu_threshold=cpu_threshold,
735 memory_threshold=memory_threshold,
736 gpu_threshold=gpu_threshold,
737 pending_pods_threshold=pending_pods_threshold,
738 pending_requested_cpu_vcpus=pending_requested_cpu_vcpus,
739 pending_requested_memory_gb=pending_requested_memory_gb,
740 pending_requested_gpus=pending_requested_gpus,
741 )
743 return HealthMonitor(cluster_id, region, thresholds)
746async def main() -> None:
747 """
748 Main function for running the health monitor with webhook dispatcher.
750 This runs both the health monitoring loop and the webhook dispatcher
751 as concurrent tasks.
752 """
753 from gco.services.webhook_dispatcher import create_webhook_dispatcher_from_env
755 health_monitor = create_health_monitor_from_env()
757 # Enable structured JSON logging for CloudWatch Insights
758 configure_structured_logging(
759 service_name="health-monitor",
760 cluster_id=health_monitor.cluster_id,
761 region=health_monitor.region,
762 )
764 webhook_dispatcher = create_webhook_dispatcher_from_env()
766 # Start webhook dispatcher
767 await webhook_dispatcher.start()
768 logger.info("Webhook dispatcher started")
770 try:
771 while True:
772 try:
773 health_status = await health_monitor.get_health_status()
774 print(f"Health Status: {health_status.status}")
775 print(f"CPU: {health_status.resource_utilization.cpu:.1f}%")
776 print(f"Memory: {health_status.resource_utilization.memory:.1f}%")
777 print(f"GPU: {health_status.resource_utilization.gpu:.1f}%")
778 print(f"Active Jobs: {health_status.active_jobs}")
779 print(f"Pending Pods: {health_status.pending_pods}")
780 if health_status.pending_requested:
781 print(
782 f"Pending Requested CPU: {health_status.pending_requested.cpu_vcpus:.1f} vCPUs"
783 )
784 print(
785 f"Pending Requested Memory: {health_status.pending_requested.memory_gb:.1f} GB"
786 )
787 if health_status.message:
788 print(f"Message: {health_status.message}")
790 # Print webhook dispatcher metrics
791 webhook_metrics = webhook_dispatcher.get_metrics()
792 print(
793 f"Webhook Deliveries: {webhook_metrics['deliveries_total']} "
794 f"(success={webhook_metrics['deliveries_success']}, "
795 f"failed={webhook_metrics['deliveries_failed']})"
796 )
797 print("-" * 50)
799 await asyncio.sleep(30) # Check every 30 seconds
801 except KeyboardInterrupt:
802 raise
803 except Exception as e:
804 logger.error(f"Error in main loop: {e}")
805 await asyncio.sleep(10)
807 except KeyboardInterrupt:
808 logger.info("Health monitor stopped by user")
809 finally:
810 await webhook_dispatcher.stop()
811 logger.info("Webhook dispatcher stopped")
814if __name__ == "__main__":
815 asyncio.run(main())