Coverage for lambda / cross-region-aggregator / handler.py: 100.00%
244 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +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-16T14:35:30Z
45# Generated from Git commit: a3141db05a743a382b008c3642b98ab968a5aa34
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):
80 raise RuntimeError("Regional API discovery is not configured")
81 # An empty list is a control-plane-only deployment (zero workload Regions):
82 # discovery finds no bridges and every aggregation answers with empty
83 # results rather than a 503.
85 regions: list[str] = []
86 for value in configured:
87 if not isinstance(value, str) or _REGION_RE.fullmatch(value) is None:
88 raise RuntimeError("Regional API discovery contains an invalid region")
89 if value not in regions:
90 regions.append(value)
91 return regions
94def _aws_url_suffix() -> str:
95 """Return the CDK-resolved DNS suffix for this deployment partition."""
96 suffix = os.environ.get("AWS_URL_SUFFIX", "").strip().lower()
97 if _DNS_SUFFIX_RE.fullmatch(suffix) is None:
98 raise RuntimeError("The AWS URL suffix is not configured")
99 return suffix
102def _normalize_regional_api_url(value: Any, region: str) -> str:
103 """Validate one stack output as this region's execute-api ``prod`` URL."""
104 parsed = urlsplit(str(value or "").strip())
105 host = (parsed.hostname or "").lower()
106 api_id = host.split(".", 1)[0]
107 expected_host = f"{api_id}.execute-api.{region}.{_aws_url_suffix()}"
108 if (
109 parsed.scheme != "https"
110 or parsed.username is not None
111 or parsed.password is not None
112 or parsed.port not in (None, 443)
113 or host != expected_host
114 or _API_ID_RE.fullmatch(api_id) is None
115 or parsed.path.rstrip("/") != "/prod"
116 or parsed.query
117 or parsed.fragment
118 ):
119 raise RuntimeError(f"The regional API endpoint for {region} is invalid")
120 return f"https://{host}/prod"
123def get_regional_endpoints() -> dict[str, str]:
124 """Discover every required regional API Gateway endpoint via CloudFormation."""
125 global _cached_endpoints, _endpoints_cache_time
127 now = time.monotonic()
128 cache_age = now - _endpoints_cache_time
129 if _cached_endpoints is not None and cache_age < _ENDPOINTS_CACHE_TTL:
130 return _cached_endpoints
132 project_name = os.environ.get("PROJECT_NAME", "gco").strip()
133 if not project_name:
134 raise RuntimeError("Regional API discovery is not configured")
136 endpoints: dict[str, str] = {}
137 failed_regions: list[str] = []
138 for region in _configured_regions():
139 stack_name = f"{project_name}-regional-api-{region}"
140 try:
141 response = boto3.client("cloudformation", region_name=region).describe_stacks(
142 StackName=stack_name
143 )
144 stacks = response.get("Stacks", [])
145 if len(stacks) != 1:
146 raise RuntimeError("Regional API stack was not found")
147 outputs = {
148 output.get("OutputKey"): output.get("OutputValue")
149 for output in stacks[0].get("Outputs", [])
150 }
151 endpoints[region] = _normalize_regional_api_url(
152 outputs.get("RegionalApiEndpoint"), region
153 )
154 except Exception:
155 failed_regions.append(region)
156 _LOGGER.exception("Regional API discovery failed for %s", region)
158 if failed_regions:
159 if _cached_endpoints is not None and cache_age < _ENDPOINTS_CACHE_MAX_STALE:
160 _LOGGER.warning(
161 "Using bounded stale regional API discovery after failures in %s",
162 ", ".join(failed_regions),
163 )
164 return _cached_endpoints
165 raise RuntimeError("One or more regional API bridges are unavailable")
167 _cached_endpoints = endpoints
168 _endpoints_cache_time = time.monotonic()
169 return endpoints
172def _sigv4_headers(region: str, method: str, url: str, body: str | None) -> dict[str, str]:
173 """Sign one execute-api request with the Lambda execution-role credentials."""
174 credentials = boto3.Session().get_credentials()
175 if credentials is None:
176 raise RuntimeError("Regional API request credentials are unavailable")
177 get_frozen = getattr(credentials, "get_frozen_credentials", None)
178 signing_credentials = get_frozen() if callable(get_frozen) else credentials
179 request = AWSRequest(
180 method=method.upper(),
181 url=url,
182 data=(body or "").encode("utf-8"),
183 headers={"Content-Type": "application/json"},
184 )
185 SigV4Auth(signing_credentials, "execute-api", region).add_auth(request)
186 return {str(key): str(value) for key, value in request.headers.items()}
189def query_region(
190 region: str,
191 endpoint: str,
192 path: str,
193 method: str = "GET",
194 body: str | None = None,
195 query_params: dict[str, str] | None = None,
196) -> dict[str, Any]:
197 """Query one regional bridge over AWS-managed TLS with SigV4 authentication."""
198 try:
199 base_url = _normalize_regional_api_url(endpoint, region)
200 query_str = "?" + urlencode(query_params) if query_params else ""
201 url = f"{base_url}{path}{query_str}"
203 response = http.request(
204 method,
205 url,
206 headers=_sigv4_headers(region, method, url, body),
207 body=body.encode("utf-8") if body else None,
208 timeout=10.0,
209 )
211 if response.status == 200:
212 data: dict[str, Any] = json.loads(response.data.decode("utf-8"))
213 data["_region"] = region
214 data["_status"] = "success"
215 return data
216 if response.status == 503 and path == "/api/v1/health":
217 # The regional health API may deliberately report degraded state
218 # with 503 while still returning a useful authenticated JSON body.
219 try:
220 data = json.loads(response.data.decode("utf-8"))
221 data["_region"] = region
222 data["_status"] = "success"
223 return data
224 except json.JSONDecodeError, UnicodeDecodeError:
225 pass
226 return {
227 "_region": region,
228 "_status": "error",
229 "_error": f"HTTP {response.status}",
230 }
231 except Exception:
232 # Never expose credentials, certificate state, API identifiers, or
233 # network details through the aggregate API response.
234 _LOGGER.exception("Authenticated regional API request failed for %s", region)
235 return {
236 "_region": region,
237 "_status": "error",
238 "_error": "Authenticated regional API request failed",
239 }
242def _job_metadata_text(job: dict[str, Any], key: str) -> str:
243 """Return a string metadata field for total, deterministic ordering."""
244 metadata = job.get("metadata")
245 if not isinstance(metadata, dict):
246 return ""
247 value = metadata.get(key)
248 return value if isinstance(value, str) else ""
251def aggregate_jobs(
252 namespace: str | None = None,
253 status: str | None = None,
254 limit: int = 50,
255) -> dict[str, Any]:
256 """Aggregate jobs from all regions."""
257 endpoints = get_regional_endpoints()
259 query_params: dict[str, str] = {"limit": str(limit * 2)} # Get more per region, then trim
260 if namespace:
261 query_params["namespace"] = namespace
262 if status:
263 query_params["status"] = status
265 all_jobs: list[dict[str, Any]] = []
266 region_summaries: list[dict[str, Any]] = []
267 errors: list[dict[str, Any]] = []
269 # Query all regions in parallel
270 with ThreadPoolExecutor(max_workers=10) as executor:
271 futures = {
272 executor.submit(
273 query_region, region, endpoint, "/api/v1/jobs", "GET", None, query_params
274 ): region
275 for region, endpoint in endpoints.items()
276 }
278 for future in as_completed(futures):
279 region = futures[future]
280 try:
281 result = future.result()
282 if result.get("_status") == "success":
283 jobs = result.get("jobs", [])
284 # Add region to each job
285 for job in jobs:
286 job["_source_region"] = region
287 all_jobs.extend(jobs)
288 region_summaries.append(
289 {
290 "region": region,
291 "count": result.get("count", len(jobs)),
292 "total": result.get("total", len(jobs)),
293 }
294 )
295 else:
296 errors.append(
297 {
298 "region": region,
299 "error": result.get("_error", "Unknown error"),
300 }
301 )
302 except Exception:
303 _LOGGER.exception("Unexpected aggregate-jobs failure for %s", region)
304 errors.append({"region": region, "error": "Regional request failed"})
306 # Canonicalize presentation independently of thread completion order.
307 region_summaries.sort(key=lambda item: item["region"])
308 errors.sort(key=lambda item: item["region"])
309 all_jobs.sort(
310 key=lambda job: (
311 str(job.get("_source_region") or ""),
312 _job_metadata_text(job, "namespace"),
313 _job_metadata_text(job, "name"),
314 _job_metadata_text(job, "uid"),
315 )
316 )
317 # Stable second pass keeps the tie-breakers ascending within each timestamp.
318 all_jobs.sort(
319 key=lambda job: _job_metadata_text(job, "creationTimestamp"),
320 reverse=True,
321 )
323 # Trim to limit
324 all_jobs = all_jobs[:limit]
326 return {
327 "total": sum(r["total"] for r in region_summaries),
328 "count": len(all_jobs),
329 "limit": limit,
330 "regions_queried": len(endpoints),
331 "regions_successful": len(region_summaries),
332 "region_summaries": region_summaries,
333 "jobs": all_jobs,
334 "errors": errors if errors else None,
335 }
338def aggregate_metrics() -> dict[str, Any]:
339 """Aggregate cluster metrics from all regions."""
340 endpoints = get_regional_endpoints()
342 region_metrics: list[dict[str, Any]] = []
343 errors: list[dict[str, Any]] = []
345 with ThreadPoolExecutor(max_workers=10) as executor:
346 futures = {
347 executor.submit(query_region, region, endpoint, "/api/v1/status"): region
348 for region, endpoint in endpoints.items()
349 }
351 for future in as_completed(futures):
352 region = futures[future]
353 try:
354 result = future.result()
355 if result.get("_status") == "success":
356 region_metrics.append(
357 {
358 "region": region,
359 "cluster_id": result.get("cluster_id"),
360 "templates_count": result.get("templates_count", 0),
361 "webhooks_count": result.get("webhooks_count", 0),
362 "resource_limits": result.get("resource_limits", {}),
363 "allowed_namespaces": result.get("allowed_namespaces", []),
364 }
365 )
366 else:
367 errors.append(
368 {
369 "region": region,
370 "error": result.get("_error", "Unknown error"),
371 }
372 )
373 except Exception:
374 _LOGGER.exception("Unexpected aggregate-metrics failure for %s", region)
375 errors.append({"region": region, "error": "Regional request failed"})
377 region_metrics.sort(key=lambda item: item["region"])
378 errors.sort(key=lambda item: item["region"])
379 return {
380 "regions_queried": len(endpoints),
381 "regions_successful": len(region_metrics),
382 "regions": region_metrics,
383 "errors": errors if errors else None,
384 }
387def aggregate_health() -> dict[str, Any]:
388 """Aggregate health status from all regions."""
389 endpoints = get_regional_endpoints()
391 region_health: list[dict[str, Any]] = []
393 with ThreadPoolExecutor(max_workers=10) as executor:
394 futures = {
395 executor.submit(query_region, region, endpoint, "/api/v1/health"): region
396 for region, endpoint in endpoints.items()
397 }
399 for future in as_completed(futures):
400 region = futures[future]
401 try:
402 result = future.result()
403 if result.get("_status") == "success":
404 region_health.append(
405 {
406 "region": region,
407 "status": result.get("status", "unknown"),
408 "cluster_id": result.get("cluster_id"),
409 "kubernetes_api": result.get("kubernetes_api"),
410 }
411 )
412 else:
413 region_health.append(
414 {
415 "region": region,
416 "status": "unreachable",
417 "error": result.get("_error"),
418 }
419 )
420 except Exception:
421 _LOGGER.exception("Unexpected aggregate-health failure for %s", region)
422 region_health.append(
423 {
424 "region": region,
425 "status": "error",
426 "error": "Regional request failed",
427 }
428 )
430 region_health.sort(key=lambda item: item["region"])
431 healthy_count = sum(1 for r in region_health if r["status"] == "healthy")
432 overall_status = "healthy" if healthy_count == len(endpoints) else "degraded"
433 if healthy_count == 0:
434 overall_status = "unhealthy"
436 return {
437 "overall_status": overall_status,
438 "healthy_regions": healthy_count,
439 "total_regions": len(endpoints),
440 "regions": region_health,
441 }
444def bulk_delete_jobs(
445 namespace: str | None = None,
446 status: str | None = None,
447 older_than_days: int | None = None,
448 label_selector: str | None = None,
449 dry_run: bool = True,
450) -> dict[str, Any]:
451 """Bulk delete jobs across all regions."""
452 endpoints = get_regional_endpoints()
454 request_body: dict[str, Any] = {
455 "dry_run": dry_run,
456 }
457 if namespace:
458 request_body["namespace"] = namespace
459 if status:
460 request_body["status"] = status
461 if older_than_days:
462 request_body["older_than_days"] = older_than_days
463 if label_selector:
464 request_body["label_selector"] = label_selector
466 body_str = json.dumps(request_body)
468 region_results: list[dict[str, Any]] = []
469 errors: list[dict[str, Any]] = []
470 total_deleted = 0
471 total_matched = 0
473 with ThreadPoolExecutor(max_workers=10) as executor:
474 futures = {
475 executor.submit(
476 query_region, region, endpoint, "/api/v1/jobs", "DELETE", body_str
477 ): region
478 for region, endpoint in endpoints.items()
479 }
481 for future in as_completed(futures):
482 region = futures[future]
483 try:
484 result = future.result()
485 if result.get("_status") == "success":
486 region_results.append(
487 {
488 "region": region,
489 "matched": result.get("total_matched", 0),
490 "deleted": result.get("deleted_count", 0),
491 "failed": result.get("failed_count", 0),
492 }
493 )
494 total_matched += result.get("total_matched", 0)
495 total_deleted += result.get("deleted_count", 0)
496 else:
497 errors.append(
498 {
499 "region": region,
500 "error": result.get("_error", "Unknown error"),
501 }
502 )
503 except Exception:
504 _LOGGER.exception("Unexpected bulk-delete failure for %s", region)
505 errors.append({"region": region, "error": "Regional request failed"})
507 region_results.sort(key=lambda item: item["region"])
508 errors.sort(key=lambda item: item["region"])
509 return {
510 "dry_run": dry_run,
511 "total_matched": total_matched,
512 "total_deleted": total_deleted,
513 "regions_queried": len(endpoints),
514 "region_results": region_results,
515 "errors": errors if errors else None,
516 }
519def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
520 """
521 Handle cross-region aggregation requests.
523 Routes:
524 GET /global/jobs - List jobs across all regions
525 GET /global/health - Health status across all regions
526 GET /global/status - Cluster status across all regions
527 DELETE /global/jobs - Bulk delete across all regions
528 """
529 http_method = event.get("httpMethod", "GET")
530 path = event.get("path", "")
531 query_params = event.get("queryStringParameters") or {}
532 body = event.get("body")
534 try:
535 # Route to appropriate handler
536 if path == "/api/v1/global/jobs" and http_method == "GET":
537 result = aggregate_jobs(
538 namespace=query_params.get("namespace"),
539 status=query_params.get("status"),
540 limit=int(query_params.get("limit", "50")),
541 )
542 elif path == "/api/v1/global/jobs" and http_method == "DELETE":
543 body_data = json.loads(body) if body else {}
544 result = bulk_delete_jobs(
545 namespace=body_data.get("namespace"),
546 status=body_data.get("status"),
547 older_than_days=body_data.get("older_than_days"),
548 label_selector=body_data.get("label_selector"),
549 dry_run=body_data.get("dry_run", True),
550 )
551 elif path == "/api/v1/global/health":
552 result = aggregate_health()
553 elif path == "/api/v1/global/status":
554 result = aggregate_metrics()
555 else:
556 return {
557 "statusCode": 404,
558 "body": json.dumps({"error": "Not found", "path": path}),
559 }
561 return {
562 "statusCode": 200,
563 "headers": {"Content-Type": "application/json"},
564 "body": json.dumps(result),
565 }
567 except RuntimeError:
568 _LOGGER.exception("Regional aggregation bridge discovery failed")
569 return {
570 "statusCode": 503,
571 "body": json.dumps({"error": "Regional aggregation is temporarily unavailable"}),
572 }
573 except Exception:
574 _LOGGER.exception("Unhandled regional aggregation request failure")
575 return {
576 "statusCode": 500,
577 "body": json.dumps({"error": "Internal server error"}),
578 }