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