Coverage for lambda / cross-region-aggregator / handler.py: 100.00%
244 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2Cross-region aggregation through reachable, IAM-authenticated regional APIs.
4The centralized Lambda is intentionally not VPC-attached and cannot connect to
5private ALBs in other regions. It discovers each deterministic regional API
6Gateway stack with CloudFormation, signs each request with its execution-role
7credentials, and sends it over the AWS-managed HTTPS endpoint. The regional
8API's VPC Lambda then signs the backend request with the deployment HMAC key and
9uses private-root authenticated TLS to reach that region's internal ALB.
11Regional Endpoint Discovery:
12 ``TARGET_REGIONS`` identifies the required regional API stacks. Each stack
13 is named ``{PROJECT_NAME}-regional-api-{region}`` and exposes a
14 ``RegionalApiEndpoint`` output. Discovery fails closed if any configured
15 bridge is absent or invalid.
17Environment Variables:
18 PROJECT_NAME: Deployment prefix used in regional API stack names.
19 TARGET_REGIONS: JSON list of required workload regions.
20 AWS_URL_SUFFIX: CDK-resolved DNS suffix for the deployment partition.
22API Routes:
23 GET /api/v1/global/jobs - List jobs across all regions
24 GET /api/v1/global/health - Health status across all regions
25 GET /api/v1/global/status - Cluster status across all regions
26 DELETE /api/v1/global/jobs - Bulk delete across all regions
27"""
29import json
30import logging
31import os
32import re
33import time
34from concurrent.futures import ThreadPoolExecutor, as_completed
35from typing import Any
36from urllib.parse import urlencode, urlsplit
38import boto3
39import urllib3
40from botocore.auth import SigV4Auth
41from botocore.awsrequest import AWSRequest
43# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
44# Generated at (UTC): 2026-09-01T14:42:56Z
45# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc
46# Flowchart(s) generated from this file:
47# * ``lambda_handler`` -> ``diagrams/code_diagrams/lambda/cross-region-aggregator/handler.lambda_handler.html``
48# (PNG: ``diagrams/code_diagrams/lambda/cross-region-aggregator/handler.lambda_handler.png``)
49# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
50# <pyflowchart-code-diagram> END
53_LOGGER = logging.getLogger(__name__)
54_REGION_RE = re.compile(r"^[a-z]{2,4}(?:-[a-z0-9]+)+-[0-9]+$")
55_API_ID_RE = re.compile(r"^[a-z0-9]+$")
56_DNS_SUFFIX_RE = re.compile(
57 r"(?=.{1,253}\Z)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+"
58 r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?",
59 re.IGNORECASE,
60)
62# Regional API Gateway uses the AWS public trust chain. Certificate validation
63# remains mandatory; the private-root pool is used only by each VPC proxy's
64# subsequent ALB connection.
65http = urllib3.PoolManager(cert_reqs="CERT_REQUIRED")
67_cached_endpoints: dict[str, str] | None = None
68_endpoints_cache_time: float = 0
69_ENDPOINTS_CACHE_TTL = 300.0
70_ENDPOINTS_CACHE_MAX_STALE = 3_600.0
73def _configured_regions() -> list[str]:
74 """Return the validated, de-duplicated regional bridge list."""
75 try:
76 configured = json.loads(os.environ["TARGET_REGIONS"])
77 except (KeyError, json.JSONDecodeError) as exc:
78 raise RuntimeError("Regional API discovery is not configured") from exc
79 if not isinstance(configured, list) or not configured:
80 raise RuntimeError("Regional API discovery is not configured")
82 regions: list[str] = []
83 for value in configured:
84 if not isinstance(value, str) or _REGION_RE.fullmatch(value) is None:
85 raise RuntimeError("Regional API discovery contains an invalid region")
86 if value not in regions:
87 regions.append(value)
88 return regions
91def _aws_url_suffix() -> str:
92 """Return the CDK-resolved DNS suffix for this deployment partition."""
93 suffix = os.environ.get("AWS_URL_SUFFIX", "").strip().lower()
94 if _DNS_SUFFIX_RE.fullmatch(suffix) is None:
95 raise RuntimeError("The AWS URL suffix is not configured")
96 return suffix
99def _normalize_regional_api_url(value: Any, region: str) -> str:
100 """Validate one stack output as this region's execute-api ``prod`` URL."""
101 parsed = urlsplit(str(value or "").strip())
102 host = (parsed.hostname or "").lower()
103 api_id = host.split(".", 1)[0]
104 expected_host = f"{api_id}.execute-api.{region}.{_aws_url_suffix()}"
105 if (
106 parsed.scheme != "https"
107 or parsed.username is not None
108 or parsed.password is not None
109 or parsed.port not in (None, 443)
110 or host != expected_host
111 or _API_ID_RE.fullmatch(api_id) is None
112 or parsed.path.rstrip("/") != "/prod"
113 or parsed.query
114 or parsed.fragment
115 ):
116 raise RuntimeError(f"The regional API endpoint for {region} is invalid")
117 return f"https://{host}/prod"
120def get_regional_endpoints() -> dict[str, str]:
121 """Discover every required regional API Gateway endpoint via CloudFormation."""
122 global _cached_endpoints, _endpoints_cache_time
124 now = time.monotonic()
125 cache_age = now - _endpoints_cache_time
126 if _cached_endpoints is not None and cache_age < _ENDPOINTS_CACHE_TTL:
127 return _cached_endpoints
129 project_name = os.environ.get("PROJECT_NAME", "gco").strip()
130 if not project_name:
131 raise RuntimeError("Regional API discovery is not configured")
133 endpoints: dict[str, str] = {}
134 failed_regions: list[str] = []
135 for region in _configured_regions():
136 stack_name = f"{project_name}-regional-api-{region}"
137 try:
138 response = boto3.client("cloudformation", region_name=region).describe_stacks(
139 StackName=stack_name
140 )
141 stacks = response.get("Stacks", [])
142 if len(stacks) != 1:
143 raise RuntimeError("Regional API stack was not found")
144 outputs = {
145 output.get("OutputKey"): output.get("OutputValue")
146 for output in stacks[0].get("Outputs", [])
147 }
148 endpoints[region] = _normalize_regional_api_url(
149 outputs.get("RegionalApiEndpoint"), region
150 )
151 except Exception:
152 failed_regions.append(region)
153 _LOGGER.exception("Regional API discovery failed for %s", region)
155 if failed_regions:
156 if _cached_endpoints is not None and cache_age < _ENDPOINTS_CACHE_MAX_STALE:
157 _LOGGER.warning(
158 "Using bounded stale regional API discovery after failures in %s",
159 ", ".join(failed_regions),
160 )
161 return _cached_endpoints
162 raise RuntimeError("One or more regional API bridges are unavailable")
164 _cached_endpoints = endpoints
165 _endpoints_cache_time = time.monotonic()
166 return endpoints
169def _sigv4_headers(region: str, method: str, url: str, body: str | None) -> dict[str, str]:
170 """Sign one execute-api request with the Lambda execution-role credentials."""
171 credentials = boto3.Session().get_credentials()
172 if credentials is None:
173 raise RuntimeError("Regional API request credentials are unavailable")
174 get_frozen = getattr(credentials, "get_frozen_credentials", None)
175 signing_credentials = get_frozen() if callable(get_frozen) else credentials
176 request = AWSRequest(
177 method=method.upper(),
178 url=url,
179 data=(body or "").encode("utf-8"),
180 headers={"Content-Type": "application/json"},
181 )
182 SigV4Auth(signing_credentials, "execute-api", region).add_auth(request)
183 return {str(key): str(value) for key, value in request.headers.items()}
186def query_region(
187 region: str,
188 endpoint: str,
189 path: str,
190 method: str = "GET",
191 body: str | None = None,
192 query_params: dict[str, str] | None = None,
193) -> dict[str, Any]:
194 """Query one regional bridge over AWS-managed TLS with SigV4 authentication."""
195 try:
196 base_url = _normalize_regional_api_url(endpoint, region)
197 query_str = "?" + urlencode(query_params) if query_params else ""
198 url = f"{base_url}{path}{query_str}"
200 response = http.request(
201 method,
202 url,
203 headers=_sigv4_headers(region, method, url, body),
204 body=body.encode("utf-8") if body else None,
205 timeout=10.0,
206 )
208 if response.status == 200:
209 data: dict[str, Any] = json.loads(response.data.decode("utf-8"))
210 data["_region"] = region
211 data["_status"] = "success"
212 return data
213 if response.status == 503 and path == "/api/v1/health":
214 # The regional health API may deliberately report degraded state
215 # with 503 while still returning a useful authenticated JSON body.
216 try:
217 data = json.loads(response.data.decode("utf-8"))
218 data["_region"] = region
219 data["_status"] = "success"
220 return data
221 except json.JSONDecodeError, UnicodeDecodeError:
222 pass
223 return {
224 "_region": region,
225 "_status": "error",
226 "_error": f"HTTP {response.status}",
227 }
228 except Exception:
229 # Never expose credentials, certificate state, API identifiers, or
230 # network details through the aggregate API response.
231 _LOGGER.exception("Authenticated regional API request failed for %s", region)
232 return {
233 "_region": region,
234 "_status": "error",
235 "_error": "Authenticated regional API request failed",
236 }
239def _job_metadata_text(job: dict[str, Any], key: str) -> str:
240 """Return a string metadata field for total, deterministic ordering."""
241 metadata = job.get("metadata")
242 if not isinstance(metadata, dict):
243 return ""
244 value = metadata.get(key)
245 return value if isinstance(value, str) else ""
248def aggregate_jobs(
249 namespace: str | None = None,
250 status: str | None = None,
251 limit: int = 50,
252) -> dict[str, Any]:
253 """Aggregate jobs from all regions."""
254 endpoints = get_regional_endpoints()
256 query_params: dict[str, str] = {"limit": str(limit * 2)} # Get more per region, then trim
257 if namespace:
258 query_params["namespace"] = namespace
259 if status:
260 query_params["status"] = status
262 all_jobs: list[dict[str, Any]] = []
263 region_summaries: list[dict[str, Any]] = []
264 errors: list[dict[str, Any]] = []
266 # Query all regions in parallel
267 with ThreadPoolExecutor(max_workers=10) as executor:
268 futures = {
269 executor.submit(
270 query_region, region, endpoint, "/api/v1/jobs", "GET", None, query_params
271 ): region
272 for region, endpoint in endpoints.items()
273 }
275 for future in as_completed(futures):
276 region = futures[future]
277 try:
278 result = future.result()
279 if result.get("_status") == "success":
280 jobs = result.get("jobs", [])
281 # Add region to each job
282 for job in jobs:
283 job["_source_region"] = region
284 all_jobs.extend(jobs)
285 region_summaries.append(
286 {
287 "region": region,
288 "count": result.get("count", len(jobs)),
289 "total": result.get("total", len(jobs)),
290 }
291 )
292 else:
293 errors.append(
294 {
295 "region": region,
296 "error": result.get("_error", "Unknown error"),
297 }
298 )
299 except Exception:
300 _LOGGER.exception("Unexpected aggregate-jobs failure for %s", region)
301 errors.append({"region": region, "error": "Regional request failed"})
303 # Canonicalize presentation independently of thread completion order.
304 region_summaries.sort(key=lambda item: item["region"])
305 errors.sort(key=lambda item: item["region"])
306 all_jobs.sort(
307 key=lambda job: (
308 str(job.get("_source_region") or ""),
309 _job_metadata_text(job, "namespace"),
310 _job_metadata_text(job, "name"),
311 _job_metadata_text(job, "uid"),
312 )
313 )
314 # Stable second pass keeps the tie-breakers ascending within each timestamp.
315 all_jobs.sort(
316 key=lambda job: _job_metadata_text(job, "creationTimestamp"),
317 reverse=True,
318 )
320 # Trim to limit
321 all_jobs = all_jobs[:limit]
323 return {
324 "total": sum(r["total"] for r in region_summaries),
325 "count": len(all_jobs),
326 "limit": limit,
327 "regions_queried": len(endpoints),
328 "regions_successful": len(region_summaries),
329 "region_summaries": region_summaries,
330 "jobs": all_jobs,
331 "errors": errors if errors else None,
332 }
335def aggregate_metrics() -> dict[str, Any]:
336 """Aggregate cluster metrics from all regions."""
337 endpoints = get_regional_endpoints()
339 region_metrics: list[dict[str, Any]] = []
340 errors: list[dict[str, Any]] = []
342 with ThreadPoolExecutor(max_workers=10) as executor:
343 futures = {
344 executor.submit(query_region, region, endpoint, "/api/v1/status"): region
345 for region, endpoint in endpoints.items()
346 }
348 for future in as_completed(futures):
349 region = futures[future]
350 try:
351 result = future.result()
352 if result.get("_status") == "success":
353 region_metrics.append(
354 {
355 "region": region,
356 "cluster_id": result.get("cluster_id"),
357 "templates_count": result.get("templates_count", 0),
358 "webhooks_count": result.get("webhooks_count", 0),
359 "resource_limits": result.get("resource_limits", {}),
360 "allowed_namespaces": result.get("allowed_namespaces", []),
361 }
362 )
363 else:
364 errors.append(
365 {
366 "region": region,
367 "error": result.get("_error", "Unknown error"),
368 }
369 )
370 except Exception:
371 _LOGGER.exception("Unexpected aggregate-metrics failure for %s", region)
372 errors.append({"region": region, "error": "Regional request failed"})
374 region_metrics.sort(key=lambda item: item["region"])
375 errors.sort(key=lambda item: item["region"])
376 return {
377 "regions_queried": len(endpoints),
378 "regions_successful": len(region_metrics),
379 "regions": region_metrics,
380 "errors": errors if errors else None,
381 }
384def aggregate_health() -> dict[str, Any]:
385 """Aggregate health status from all regions."""
386 endpoints = get_regional_endpoints()
388 region_health: list[dict[str, Any]] = []
390 with ThreadPoolExecutor(max_workers=10) as executor:
391 futures = {
392 executor.submit(query_region, region, endpoint, "/api/v1/health"): region
393 for region, endpoint in endpoints.items()
394 }
396 for future in as_completed(futures):
397 region = futures[future]
398 try:
399 result = future.result()
400 if result.get("_status") == "success":
401 region_health.append(
402 {
403 "region": region,
404 "status": result.get("status", "unknown"),
405 "cluster_id": result.get("cluster_id"),
406 "kubernetes_api": result.get("kubernetes_api"),
407 }
408 )
409 else:
410 region_health.append(
411 {
412 "region": region,
413 "status": "unreachable",
414 "error": result.get("_error"),
415 }
416 )
417 except Exception:
418 _LOGGER.exception("Unexpected aggregate-health failure for %s", region)
419 region_health.append(
420 {
421 "region": region,
422 "status": "error",
423 "error": "Regional request failed",
424 }
425 )
427 region_health.sort(key=lambda item: item["region"])
428 healthy_count = sum(1 for r in region_health if r["status"] == "healthy")
429 overall_status = "healthy" if healthy_count == len(endpoints) else "degraded"
430 if healthy_count == 0:
431 overall_status = "unhealthy"
433 return {
434 "overall_status": overall_status,
435 "healthy_regions": healthy_count,
436 "total_regions": len(endpoints),
437 "regions": region_health,
438 }
441def bulk_delete_jobs(
442 namespace: str | None = None,
443 status: str | None = None,
444 older_than_days: int | None = None,
445 label_selector: str | None = None,
446 dry_run: bool = True,
447) -> dict[str, Any]:
448 """Bulk delete jobs across all regions."""
449 endpoints = get_regional_endpoints()
451 request_body: dict[str, Any] = {
452 "dry_run": dry_run,
453 }
454 if namespace:
455 request_body["namespace"] = namespace
456 if status:
457 request_body["status"] = status
458 if older_than_days:
459 request_body["older_than_days"] = older_than_days
460 if label_selector:
461 request_body["label_selector"] = label_selector
463 body_str = json.dumps(request_body)
465 region_results: list[dict[str, Any]] = []
466 errors: list[dict[str, Any]] = []
467 total_deleted = 0
468 total_matched = 0
470 with ThreadPoolExecutor(max_workers=10) as executor:
471 futures = {
472 executor.submit(
473 query_region, region, endpoint, "/api/v1/jobs", "DELETE", body_str
474 ): region
475 for region, endpoint in endpoints.items()
476 }
478 for future in as_completed(futures):
479 region = futures[future]
480 try:
481 result = future.result()
482 if result.get("_status") == "success":
483 region_results.append(
484 {
485 "region": region,
486 "matched": result.get("total_matched", 0),
487 "deleted": result.get("deleted_count", 0),
488 "failed": result.get("failed_count", 0),
489 }
490 )
491 total_matched += result.get("total_matched", 0)
492 total_deleted += result.get("deleted_count", 0)
493 else:
494 errors.append(
495 {
496 "region": region,
497 "error": result.get("_error", "Unknown error"),
498 }
499 )
500 except Exception:
501 _LOGGER.exception("Unexpected bulk-delete failure for %s", region)
502 errors.append({"region": region, "error": "Regional request failed"})
504 region_results.sort(key=lambda item: item["region"])
505 errors.sort(key=lambda item: item["region"])
506 return {
507 "dry_run": dry_run,
508 "total_matched": total_matched,
509 "total_deleted": total_deleted,
510 "regions_queried": len(endpoints),
511 "region_results": region_results,
512 "errors": errors if errors else None,
513 }
516def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
517 """
518 Handle cross-region aggregation requests.
520 Routes:
521 GET /global/jobs - List jobs across all regions
522 GET /global/health - Health status across all regions
523 GET /global/status - Cluster status across all regions
524 DELETE /global/jobs - Bulk delete across all regions
525 """
526 http_method = event.get("httpMethod", "GET")
527 path = event.get("path", "")
528 query_params = event.get("queryStringParameters") or {}
529 body = event.get("body")
531 try:
532 # Route to appropriate handler
533 if path == "/api/v1/global/jobs" and http_method == "GET":
534 result = aggregate_jobs(
535 namespace=query_params.get("namespace"),
536 status=query_params.get("status"),
537 limit=int(query_params.get("limit", "50")),
538 )
539 elif path == "/api/v1/global/jobs" and http_method == "DELETE":
540 body_data = json.loads(body) if body else {}
541 result = bulk_delete_jobs(
542 namespace=body_data.get("namespace"),
543 status=body_data.get("status"),
544 older_than_days=body_data.get("older_than_days"),
545 label_selector=body_data.get("label_selector"),
546 dry_run=body_data.get("dry_run", True),
547 )
548 elif path == "/api/v1/global/health":
549 result = aggregate_health()
550 elif path == "/api/v1/global/status":
551 result = aggregate_metrics()
552 else:
553 return {
554 "statusCode": 404,
555 "body": json.dumps({"error": "Not found", "path": path}),
556 }
558 return {
559 "statusCode": 200,
560 "headers": {"Content-Type": "application/json"},
561 "body": json.dumps(result),
562 }
564 except RuntimeError:
565 _LOGGER.exception("Regional aggregation bridge discovery failed")
566 return {
567 "statusCode": 503,
568 "body": json.dumps({"error": "Regional aggregation is temporarily unavailable"}),
569 }
570 except Exception:
571 _LOGGER.exception("Unhandled regional aggregation request failure")
572 return {
573 "statusCode": 500,
574 "body": json.dumps({"error": "Internal server error"}),
575 }