Coverage for gco/config/config_loader.py: 94.99%
425 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""
2Configuration loader for GCO (Global Capacity Orchestrator on AWS).
4This module loads and validates configuration from CDK context (cdk.json).
5It provides type-safe access to all configuration values with sensible defaults
6and comprehensive validation.
8Configuration Sections:
9- project_name: Unique identifier for the deployment
10- regions: List of AWS regions to deploy to
11- kubernetes_version: EKS Kubernetes version
12- resource_thresholds: CPU/memory/GPU utilization thresholds
13- global_accelerator: Global Accelerator settings
14- alb_config: Application Load Balancer health check settings
15- manifest_processor: Manifest validation and resource limits
16- api_gateway: Throttling and logging configuration
17- tags: Common tags applied to all resources
19Usage:
20 config = ConfigLoader(app)
21 regions = config.get_regions()
22 cluster_config = config.get_cluster_config("us-east-1")
23"""
25from __future__ import annotations
27import logging
28import re
29from typing import Any, cast
31import boto3
32from aws_cdk import App
34from gco.models import ClusterConfig, ResourceThresholds
35from gco.stacks.constants import (
36 DEFAULT_MAX_REQUEST_BODY_BYTES,
37 known_cloudformation_regions,
38 validated_deployment_partition,
39 validated_regional_deployment_regions,
40 validated_request_body_limit,
41)
43logger = logging.getLogger(__name__)
46class ConfigValidationError(Exception):
47 """Raised when configuration validation fails."""
49 pass
52class ConfigLoader:
53 """
54 Loads and validates configuration from CDK context (cdk.json)
55 """
57 # Keep the public class attribute for compatibility. Endpoint metadata
58 # covers every CloudFormation Region known to the installed AWS SDK; this
59 # is not a project-specific allowlist.
60 VALID_REGIONS = known_cloudformation_regions()
62 def __init__(self, app: App):
63 self.app = app
64 self._validate_configuration()
66 def _validate_configuration(self) -> None:
67 """Validate the entire configuration"""
68 # Check if we have any context at all (might be running outside CDK)
69 project_name = self.app.node.try_get_context("project_name")
70 if project_name is None:
71 # Running outside CDK context, skip validation
72 return
74 # Validate required fields exist
75 required_fields = [
76 "project_name",
77 "kubernetes_version",
78 "resource_thresholds",
79 ]
80 for field in required_fields:
81 if not self.app.node.try_get_context(field):
82 raise ConfigValidationError(f"Required configuration field '{field}' is missing")
84 # Validate project_name format before anything consumes it (#139).
85 self._validate_project_name()
87 # Check for deployment_regions
88 deployment_regions = self.app.node.try_get_context("deployment_regions")
89 if not isinstance(deployment_regions, dict) or not deployment_regions:
90 raise ConfigValidationError(
91 "Required configuration field 'deployment_regions' must be a non-empty object"
92 )
94 # Validate regions
95 self._validate_regions()
97 # Validate resource thresholds
98 self._validate_resource_thresholds()
100 # Validate Global Accelerator config
101 self._validate_global_accelerator_config()
103 # Validate deployment-local backend TLS rotation policy
104 self._validate_backend_tls_config()
106 # Validate ALB config
107 self._validate_alb_config()
109 # Validate manifest processor config
110 self._validate_manifest_processor_config()
112 # Validate API Gateway config
113 self._validate_api_gateway_config()
115 # Validate EKS cluster config
116 self._validate_eks_cluster_config()
118 # Validate analytics environment config (optional block)
119 self._validate_analytics_environment_config()
121 # Validate cluster observability config (optional block)
122 self._validate_cluster_observability_config()
124 # Validate cost monitoring config (optional block)
125 self._validate_cost_monitoring_config()
127 # Validate historical capacity surface config (optional block)
128 self._validate_capacity_history_config()
130 #: Allowed ``project_name`` format (#139). ``project_name`` is the
131 #: deployment's unique prefix and flows into S3 bucket names, the Cognito
132 #: hosted-UI domain prefix, SSM parameter paths, IAM role names, and
133 #: CloudFormation export names. The tightest of those constraints is S3 /
134 #: Cognito naming (lowercase letters, digits, hyphens; must start with a
135 #: letter), so require: a leading lowercase letter followed by 1–30
136 #: lowercase letters, digits, or hyphens (total length 2–31).
137 PROJECT_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9-]{1,30}$")
139 def _validate_project_name(self) -> None:
140 """Validate ``project_name`` format so misconfigurations fail at synth.
142 ``project_name`` is documented as the deployment's unique identifier
143 and is used as the prefix for nearly every physical resource name. If
144 it contains characters that are illegal in S3 bucket names or Cognito
145 domain prefixes (uppercase, underscores, dots, leading digit, etc.),
146 ``cdk synth`` still succeeds but the deploy fails late with an opaque
147 AWS naming error. Catching it here turns that into an actionable
148 message up front.
149 """
150 project_name = self.app.node.try_get_context("project_name")
151 if not isinstance(project_name, str) or not self.PROJECT_NAME_PATTERN.match(project_name): 151 ↛ 152line 151 didn't jump to line 152 because the condition on line 151 was never true
152 raise ConfigValidationError(
153 f"Invalid project_name {project_name!r}. It must match "
154 f"{self.PROJECT_NAME_PATTERN.pattern} (start with a lowercase letter, then "
155 "2–31 total characters of lowercase letters, digits, or hyphens). "
156 "project_name is the deployment prefix for S3 buckets, the Cognito "
157 "domain, SSM paths, and CloudFormation exports, so it must be a valid "
158 "lowercase DNS-style label."
159 )
161 def _validate_regions(self) -> None:
162 """Validate region configuration against the shared app/CLI contract."""
163 deployment_regions = self.get_deployment_regions()
164 try:
165 for field in ("global", "api_gateway", "monitoring"):
166 region = deployment_regions[field]
167 if not isinstance(region, str) or region not in self.VALID_REGIONS:
168 raise ValueError(
169 f"Invalid {field} region {region!r}; expected an AWS region with a "
170 "CloudFormation endpoint known to the installed SDK"
171 )
172 regional = validated_regional_deployment_regions(
173 deployment_regions["regional"],
174 known_regions=self.VALID_REGIONS,
175 )
176 validated_deployment_partition(
177 (
178 deployment_regions["global"],
179 deployment_regions["api_gateway"],
180 deployment_regions["monitoring"],
181 *regional,
182 )
183 )
184 except (RuntimeError, ValueError) as exc:
185 raise ConfigValidationError(str(exc)) from exc
187 def _validate_resource_thresholds(self) -> None:
188 """Validate resource threshold configuration"""
189 thresholds_config = self.app.node.try_get_context("resource_thresholds")
191 required_thresholds = ["cpu_threshold", "memory_threshold", "gpu_threshold"]
192 for threshold in required_thresholds:
193 if threshold not in thresholds_config:
194 raise ConfigValidationError(f"Missing threshold configuration: {threshold}")
196 value = thresholds_config[threshold]
197 if not isinstance(value, int) or (value != -1 and not 0 <= value <= 100):
198 raise ConfigValidationError(
199 f"{threshold} must be an integer between 0 and 100 (or -1 to disable), got {value}"
200 )
202 # Validate optional thresholds if present
203 for opt_threshold in [
204 "pending_pods_threshold",
205 "pending_requested_cpu_vcpus",
206 "pending_requested_memory_gb",
207 "pending_requested_gpus",
208 ]:
209 if opt_threshold in thresholds_config:
210 value = thresholds_config[opt_threshold]
211 if not isinstance(value, int) or (value != -1 and value < 0):
212 raise ConfigValidationError(
213 f"{opt_threshold} must be a non-negative integer (or -1 to disable), got {value}"
214 )
216 def _validate_global_accelerator_config(self) -> None:
217 """Validate Global Accelerator configuration"""
218 ga_config = self.app.node.try_get_context("global_accelerator")
219 if not ga_config:
220 raise ConfigValidationError("global_accelerator configuration is required")
222 # ``name`` is intentionally optional: when omitted it defaults to
223 # ``<project_name>-accelerator`` (see get_global_accelerator_config /
224 # GCOGlobalStack) so a second deployment gets a project-scoped name
225 # from the single ``project_name`` knob (#139).
226 required_fields = [
227 "health_check_grace_period",
228 "health_check_interval",
229 "health_check_timeout",
230 "health_check_path",
231 ]
232 for field in required_fields:
233 if field not in ga_config:
234 raise ConfigValidationError(f"Missing global_accelerator configuration: {field}")
236 # Validate timing values
237 for field in ["health_check_grace_period", "health_check_interval", "health_check_timeout"]:
238 value = ga_config[field]
239 if not isinstance(value, int) or value <= 0:
240 raise ConfigValidationError(f"{field} must be a positive integer, got {value}")
242 # Validate health check path
243 if not ga_config["health_check_path"].startswith("/"):
244 raise ConfigValidationError("health_check_path must start with '/'")
246 # Validate optional client affinity. Omitting the key is allowed and
247 # defaults to "NONE" in get_global_accelerator_config().
248 if "client_affinity" in ga_config:
249 allowed_affinity = {"NONE", "SOURCE_IP"}
250 value = ga_config["client_affinity"]
251 if not isinstance(value, str) or value.upper() not in allowed_affinity:
252 raise ConfigValidationError(
253 f"client_affinity must be one of {sorted(allowed_affinity)}, got {value!r}"
254 )
256 def _validate_backend_tls_config(self) -> None:
257 """Validate private-root and leaf-certificate lifecycle settings."""
258 config = self.get_backend_tls_config()
259 ranges = {
260 "root_generation": (1, 1_000_000),
261 "root_validity_days": (365, 36_500),
262 "root_rotate_before_days": (30, 3_650),
263 "root_activation_delay_hours": (1, 168),
264 "root_overlap_days": (2, 365),
265 "leaf_validity_days": (2, 397),
266 "leaf_rotate_before_days": (1, 90),
267 "rotation_schedule_hours": (1, 24),
268 "trust_cache_ttl_seconds": (1, 3_600),
269 "trust_cache_max_stale_seconds": (1, 86_400),
270 }
271 for field, (minimum, maximum) in ranges.items():
272 value = config.get(field)
273 if type(value) is not int or not minimum <= value <= maximum: 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true
274 raise ConfigValidationError(
275 f"backend_tls.{field} must be an integer between "
276 f"{minimum} and {maximum}, got {value!r}"
277 )
279 if config["root_rotate_before_days"] >= config["root_validity_days"]: 279 ↛ 280line 279 didn't jump to line 280 because the condition on line 279 was never true
280 raise ConfigValidationError(
281 "backend_tls.root_rotate_before_days must be less than root_validity_days"
282 )
283 if config["leaf_rotate_before_days"] >= config["leaf_validity_days"]: 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true
284 raise ConfigValidationError(
285 "backend_tls.leaf_rotate_before_days must be less than leaf_validity_days"
286 )
287 if config["root_validity_days"] <= config["leaf_validity_days"]: 287 ↛ 288line 287 didn't jump to line 288 because the condition on line 287 was never true
288 raise ConfigValidationError(
289 "backend_tls.root_validity_days must exceed leaf_validity_days"
290 )
291 if config["root_overlap_days"] <= config["leaf_validity_days"]: 291 ↛ 292line 291 didn't jump to line 292 because the condition on line 291 was never true
292 raise ConfigValidationError(
293 "backend_tls.root_overlap_days must exceed leaf_validity_days so old leaves "
294 "remain trusted throughout root rollover"
295 )
296 if config["trust_cache_max_stale_seconds"] < config["trust_cache_ttl_seconds"]: 296 ↛ 297line 296 didn't jump to line 297 because the condition on line 296 was never true
297 raise ConfigValidationError(
298 "backend_tls.trust_cache_max_stale_seconds must be at least trust_cache_ttl_seconds"
299 )
300 if config["root_activation_delay_hours"] * 3_600 <= config["trust_cache_max_stale_seconds"]: 300 ↛ 301line 300 didn't jump to line 301 because the condition on line 300 was never true
301 raise ConfigValidationError(
302 "backend_tls.root_activation_delay_hours must exceed the maximum stale trust "
303 "cache window so every proxy can observe a pending root before leaf rollover"
304 )
306 def _validate_alb_config(self) -> None:
307 """Validate ALB configuration"""
308 alb_config = self.app.node.try_get_context("alb_config")
309 if not alb_config:
310 raise ConfigValidationError("alb_config configuration is required")
312 required_fields = [
313 "health_check_interval",
314 "health_check_timeout",
315 "healthy_threshold",
316 "unhealthy_threshold",
317 ]
318 for field in required_fields:
319 if field not in alb_config:
320 raise ConfigValidationError(f"Missing alb_config configuration: {field}")
322 value = alb_config[field]
323 if not isinstance(value, int) or value <= 0:
324 raise ConfigValidationError(f"{field} must be a positive integer, got {value}")
326 def _validate_manifest_processor_config(self) -> None:
327 """Validate manifest processor configuration.
329 The manifest processor section in cdk.json holds service-specific
330 settings only. The shared validation policy (allowed_namespaces,
331 resource_quotas, trusted_registries, trusted_dockerhub_orgs,
332 manifest_security_policy, allowed_kinds) lives under
333 ``job_validation_policy`` because the queue_processor reads the
334 same values.
335 """
336 mp_config = self.app.node.try_get_context("manifest_processor")
337 if not mp_config:
338 raise ConfigValidationError("manifest_processor configuration is required")
340 required_fields = [
341 "image",
342 "replicas",
343 "resource_limits",
344 ]
345 for field in required_fields:
346 if field not in mp_config:
347 raise ConfigValidationError(f"Missing manifest_processor configuration: {field}")
349 # Validate replicas
350 if not isinstance(mp_config["replicas"], int) or mp_config["replicas"] <= 0:
351 raise ConfigValidationError("manifest_processor replicas must be a positive integer")
353 try:
354 validated_request_body_limit(
355 mp_config.get("max_request_body_bytes", DEFAULT_MAX_REQUEST_BODY_BYTES)
356 )
357 except ValueError as exc:
358 raise ConfigValidationError(f"manifest_processor.{exc}") from exc
360 # Validate the shared policy section separately so a misconfigured
361 # policy block surfaces a clear error pointing at the right key.
362 policy = self.app.node.try_get_context("job_validation_policy")
363 if policy is None:
364 raise ConfigValidationError(
365 "job_validation_policy configuration is required (shared between "
366 "manifest_processor and queue_processor)"
367 )
368 for policy_field in ("allowed_namespaces", "resource_quotas"):
369 if policy_field not in policy:
370 raise ConfigValidationError(
371 f"Missing job_validation_policy configuration: {policy_field}"
372 )
374 # Validate resource limits
375 resource_limits = mp_config["resource_limits"]
376 if "cpu" not in resource_limits or "memory" not in resource_limits:
377 raise ConfigValidationError(
378 "manifest_processor resource_limits must contain 'cpu' and 'memory'"
379 )
381 # Validate allowed namespaces (lives under job_validation_policy).
382 if not isinstance(policy["allowed_namespaces"], list):
383 raise ConfigValidationError("job_validation_policy.allowed_namespaces must be a list")
385 def _validate_api_gateway_config(self) -> None:
386 """Validate API Gateway configuration"""
387 api_gw_config = self.app.node.try_get_context("api_gateway")
388 if not api_gw_config:
389 raise ConfigValidationError("api_gateway configuration is required")
391 required_fields = [
392 "throttle_rate_limit",
393 "throttle_burst_limit",
394 "log_level",
395 "metrics_enabled",
396 "tracing_enabled",
397 ]
398 for field in required_fields:
399 if field not in api_gw_config:
400 raise ConfigValidationError(f"Missing api_gateway configuration: {field}")
402 # Validate throttle limits
403 throttle_rate = api_gw_config["throttle_rate_limit"]
404 throttle_burst = api_gw_config["throttle_burst_limit"]
406 if not isinstance(throttle_rate, int) or throttle_rate <= 0:
407 raise ConfigValidationError(
408 f"throttle_rate_limit must be a positive integer, got {throttle_rate}"
409 )
411 if not isinstance(throttle_burst, int) or throttle_burst <= 0:
412 raise ConfigValidationError(
413 f"throttle_burst_limit must be a positive integer, got {throttle_burst}"
414 )
416 if throttle_burst < throttle_rate:
417 raise ConfigValidationError(
418 "throttle_burst_limit should be greater than or equal to throttle_rate_limit"
419 )
421 # Validate log level
422 valid_log_levels = ["OFF", "ERROR", "INFO"]
423 log_level = api_gw_config["log_level"]
424 if log_level not in valid_log_levels:
425 raise ConfigValidationError(
426 f"log_level must be one of {valid_log_levels}, got {log_level}"
427 )
429 # Validate boolean flags
430 if not isinstance(api_gw_config["metrics_enabled"], bool):
431 raise ConfigValidationError("metrics_enabled must be a boolean")
433 if not isinstance(api_gw_config["tracing_enabled"], bool):
434 raise ConfigValidationError("tracing_enabled must be a boolean")
436 if "regional_api_enabled" in api_gw_config and not isinstance( 436 ↛ 439line 436 didn't jump to line 439 because the condition on line 436 was never true
437 api_gw_config["regional_api_enabled"], bool
438 ):
439 raise ConfigValidationError("regional_api_enabled must be a boolean")
441 def _validate_eks_cluster_config(self) -> None:
442 """Validate EKS cluster configuration"""
443 eks_config = self.app.node.try_get_context("eks_cluster") or {}
445 # Validate endpoint_access if present
446 if "endpoint_access" in eks_config:
447 valid_access_modes = ["PRIVATE", "PUBLIC_AND_PRIVATE"]
448 if eks_config["endpoint_access"] not in valid_access_modes:
449 raise ConfigValidationError(
450 f"endpoint_access must be one of {valid_access_modes}, "
451 f"got {eks_config['endpoint_access']}"
452 )
454 def _validate_analytics_environment_config(self) -> None:
455 """Validate the optional analytics_environment block in cdk.json.
457 The block is entirely optional; absence means the feature is disabled
458 and no validation is needed. When present, we validate:
460 - ``enabled``: must be a bool if present (defaults to False via merge).
461 - ``hyperpod.enabled``: must be a bool if present (defaults to False).
462 - ``cognito.removal_policy`` and ``efs.removal_policy``: must be the
463 literal strings ``"destroy"`` or ``"retain"`` (case sensitive — they
464 are passed verbatim to CDK's ``RemovalPolicy`` lookup by the
465 consumer).
466 """
467 analytics_ctx = self.app.node.try_get_context("analytics_environment")
468 if not isinstance(analytics_ctx, dict):
469 # Block is absent or malformed — defaults apply, nothing to validate.
470 return
472 # Top-level `enabled` must be a bool if provided.
473 if "enabled" in analytics_ctx and not isinstance(analytics_ctx["enabled"], bool):
474 raise ConfigValidationError(
475 f"analytics_environment.enabled must be a bool, got "
476 f"{type(analytics_ctx['enabled']).__name__}: {analytics_ctx['enabled']!r}"
477 )
479 # `hyperpod.enabled` must be a bool if the sub-block is a dict and
480 # carries the key.
481 hyperpod_ctx = analytics_ctx.get("hyperpod")
482 if (
483 isinstance(hyperpod_ctx, dict)
484 and "enabled" in hyperpod_ctx
485 and not isinstance(hyperpod_ctx["enabled"], bool)
486 ):
487 raise ConfigValidationError(
488 f"analytics_environment.hyperpod.enabled must be a bool, got "
489 f"{type(hyperpod_ctx['enabled']).__name__}: {hyperpod_ctx['enabled']!r}"
490 )
492 # `canvas.enabled` must be a bool if the sub-block is a dict and
493 # carries the key. Mirrors the hyperpod validation above.
494 canvas_ctx = analytics_ctx.get("canvas")
495 if (
496 isinstance(canvas_ctx, dict)
497 and "enabled" in canvas_ctx
498 and not isinstance(canvas_ctx["enabled"], bool)
499 ):
500 raise ConfigValidationError(
501 f"analytics_environment.canvas.enabled must be a bool, got "
502 f"{type(canvas_ctx['enabled']).__name__}: {canvas_ctx['enabled']!r}"
503 )
505 valid_removal_policies = {"destroy", "retain"}
507 for sub_block in ("cognito", "efs"):
508 sub_ctx = analytics_ctx.get(sub_block)
509 if not isinstance(sub_ctx, dict):
510 continue
511 if "removal_policy" not in sub_ctx:
512 continue
513 removal_policy = sub_ctx["removal_policy"]
514 if removal_policy not in valid_removal_policies:
515 raise ConfigValidationError(
516 f"analytics_environment.{sub_block}.removal_policy must be one of "
517 f"{sorted(valid_removal_policies)}, got {removal_policy!r}"
518 )
520 def _validate_cluster_observability_config(self) -> None:
521 """Validate the optional cluster_observability block in cdk.json.
523 The block is entirely optional; absence means the on-by-default
524 defaults apply and nothing needs validating. When present, we check:
526 - ``enabled``: must be a bool if present (defaults to True via merge —
527 in-cluster observability is on unless explicitly disabled).
528 - ``grafana``/``prometheus``/``alertmanager`` sub-block ``persistence_size``
529 and ``prometheus.retention``: must be non-empty strings if present
530 (they are passed verbatim to Helm chart values as Kubernetes
531 quantity / duration strings).
532 - ``alertmanager.enabled``: must be a bool if present.
533 """
534 obs_ctx = self.app.node.try_get_context("cluster_observability")
535 if not isinstance(obs_ctx, dict):
536 # Block is absent or malformed — defaults apply, nothing to validate.
537 return
539 if "enabled" in obs_ctx and not isinstance(obs_ctx["enabled"], bool):
540 raise ConfigValidationError(
541 f"cluster_observability.enabled must be a bool, got "
542 f"{type(obs_ctx['enabled']).__name__}: {obs_ctx['enabled']!r}"
543 )
545 # Non-empty-string checks for the size / retention knobs.
546 string_fields = (
547 ("grafana", "persistence_size"),
548 ("prometheus", "persistence_size"),
549 ("prometheus", "retention"),
550 ("alertmanager", "persistence_size"),
551 )
552 for sub_block, field in string_fields:
553 sub_ctx = obs_ctx.get(sub_block)
554 if not isinstance(sub_ctx, dict) or field not in sub_ctx:
555 continue
556 value = sub_ctx[field]
557 if not isinstance(value, str) or not value.strip():
558 raise ConfigValidationError(
559 f"cluster_observability.{sub_block}.{field} must be a non-empty "
560 f"string, got {value!r}"
561 )
563 # `grafana.admin_password_rotation_schedule`: the cron for the in-cluster
564 # CronJob that rotates the Grafana admin password. Validate the 5-field
565 # cron shape so a typo fails at synth rather than yielding an
566 # un-schedulable CronJob in every region.
567 grafana_ctx = obs_ctx.get("grafana")
568 if isinstance(grafana_ctx, dict) and "admin_password_rotation_schedule" in grafana_ctx:
569 schedule = grafana_ctx["admin_password_rotation_schedule"]
570 if not isinstance(schedule, str) or len(schedule.split()) != 5:
571 raise ConfigValidationError(
572 "cluster_observability.grafana.admin_password_rotation_schedule must be "
573 f"a 5-field cron expression string, got {schedule!r}"
574 )
576 # `alertmanager.enabled` must be a bool if the sub-block carries it.
577 alertmanager_ctx = obs_ctx.get("alertmanager")
578 if (
579 isinstance(alertmanager_ctx, dict)
580 and "enabled" in alertmanager_ctx
581 and not isinstance(alertmanager_ctx["enabled"], bool)
582 ):
583 raise ConfigValidationError(
584 f"cluster_observability.alertmanager.enabled must be a bool, got "
585 f"{type(alertmanager_ctx['enabled']).__name__}: "
586 f"{alertmanager_ctx['enabled']!r}"
587 )
589 def _validate_cost_monitoring_config(self) -> None:
590 """Validate the optional ``cost_monitoring`` block in cdk.json.
592 The block is entirely optional; absence means the on-by-default
593 defaults apply and nothing needs validating. When present, we check:
595 - ``enabled``: must be a bool if present (defaults to True via merge).
596 - ``reports.interval_minutes``: positive int between 5 and 1440 if
597 present (the cost-monitor service's scheduled report cadence).
598 - ``reports.retention_days`` /
599 ``reports.transition_to_infrequent_access_days`` /
600 ``athena.query_results_retention_days``: positive ints if present.
601 - The IA transition must happen strictly before expiration, otherwise
602 the S3 lifecycle configuration is rejected at deploy time — fail at
603 synth instead.
605 There is deliberately no cross-toggle error against
606 ``cluster_observability``: cost monitoring's *effective* enablement is
607 the conjunction of both toggles (see
608 :meth:`get_cost_monitoring_enabled`), so disabling observability
609 simply switches the cost pipeline off with it — ``gco monitoring
610 disable`` must not break synthesis.
611 """
612 cost_ctx = self.app.node.try_get_context("cost_monitoring")
613 if not isinstance(cost_ctx, dict):
614 # Block absent or malformed — defaults apply, nothing to validate.
615 return
617 if "enabled" in cost_ctx and not isinstance(cost_ctx["enabled"], bool):
618 raise ConfigValidationError(
619 f"cost_monitoring.enabled must be a bool, got "
620 f"{type(cost_ctx['enabled']).__name__}: {cost_ctx['enabled']!r}"
621 )
623 int_fields = (
624 ("reports", "interval_minutes", 5, 1_440),
625 ("reports", "retention_days", 1, 3_650),
626 ("reports", "transition_to_infrequent_access_days", 30, 3_650),
627 ("athena", "query_results_retention_days", 1, 3_650),
628 )
629 for sub_block, field, minimum, maximum in int_fields:
630 sub_ctx = cost_ctx.get(sub_block)
631 if not isinstance(sub_ctx, dict) or field not in sub_ctx:
632 continue
633 value = sub_ctx[field]
634 if (
635 not isinstance(value, int)
636 or isinstance(value, bool)
637 or not minimum <= value <= maximum
638 ):
639 raise ConfigValidationError(
640 f"cost_monitoring.{sub_block}.{field} must be an integer between "
641 f"{minimum} and {maximum}, got {value!r}"
642 )
644 merged = self.get_cost_monitoring_config()
645 reports = merged["reports"]
646 if reports["transition_to_infrequent_access_days"] >= reports["retention_days"]:
647 raise ConfigValidationError(
648 "cost_monitoring.reports.transition_to_infrequent_access_days "
649 f"({reports['transition_to_infrequent_access_days']}) must be smaller than "
650 f"cost_monitoring.reports.retention_days ({reports['retention_days']}); "
651 "S3 rejects lifecycle rules that transition on or after expiration."
652 )
654 def _validate_capacity_history_config(self) -> None:
655 """Validate the optional ``historical`` block in cdk.json.
657 The block is entirely optional; absence means the historical capacity
658 surface is disabled and no validation is needed. When present, types
659 are validated so a typo fails fast at synth time:
661 - ``enabled``: bool if present.
662 - ``retention_days`` / ``poll_interval_minutes``: positive ints if present.
663 - ``watch_instance_types`` / ``enabled_regions``: lists of strings if present.
664 - every region in ``enabled_regions`` must be a known AWS region.
665 """
666 historical_ctx = self.app.node.try_get_context("historical")
667 if not isinstance(historical_ctx, dict):
668 return
670 if "enabled" in historical_ctx and not isinstance(historical_ctx["enabled"], bool):
671 raise ConfigValidationError(
672 f"historical.enabled must be a bool, got "
673 f"{type(historical_ctx['enabled']).__name__}: {historical_ctx['enabled']!r}"
674 )
676 for int_field in (
677 "retention_days",
678 "poll_interval_minutes",
679 "capacity_block_duration_hours",
680 ):
681 if int_field not in historical_ctx:
682 continue
683 value = historical_ctx[int_field]
684 if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
685 raise ConfigValidationError(
686 f"historical.{int_field} must be a positive integer, got {value!r}"
687 )
689 # The long-block probe duration may be 0 to disable the long probe
690 # entirely, so it is validated as non-negative rather than positive.
691 if "capacity_block_long_duration_hours" in historical_ctx:
692 long_value = historical_ctx["capacity_block_long_duration_hours"]
693 if not isinstance(long_value, int) or isinstance(long_value, bool) or long_value < 0:
694 raise ConfigValidationError(
695 "historical.capacity_block_long_duration_hours must be a non-negative "
696 f"integer (0 disables the long probe), got {long_value!r}"
697 )
699 for list_field in ("watch_instance_types", "enabled_regions"):
700 if list_field not in historical_ctx:
701 continue
702 value = historical_ctx[list_field]
703 if not isinstance(value, list) or not all(isinstance(v, str) for v in value):
704 raise ConfigValidationError(
705 f"historical.{list_field} must be a list of strings, got {value!r}"
706 )
708 for region in historical_ctx.get("enabled_regions", []) or []:
709 if region not in self.VALID_REGIONS: 709 ↛ 708line 709 didn't jump to line 708 because the condition on line 709 was always true
710 raise ConfigValidationError(
711 f"historical.enabled_regions contains invalid region '{region}'. "
712 f"Valid regions: {sorted(self.VALID_REGIONS)}"
713 )
715 def get_project_name(self) -> str:
716 """Get project name from configuration"""
717 return self.app.node.try_get_context("project_name") or "gco"
719 def get_deployment_regions(self) -> dict[str, Any]:
720 """Get deployment regions configuration.
722 Returns a dict with:
723 - global: Region for Global Accelerator and SSM parameters (default: us-east-2)
724 - api_gateway: Region for API Gateway stack (default: us-east-2)
725 - monitoring: Region for Monitoring stack (default: us-east-2)
726 - regional: List of regions for EKS clusters (default: ["us-east-1"])
728 Note: Global Accelerator is a global service but requires a "home" region
729 for CloudFormation deployment. us-east-2 is used by default to keep
730 global infrastructure separate from workload regions.
731 """
732 deployment_regions = self.app.node.try_get_context("deployment_regions") or {}
734 return {
735 "global": deployment_regions.get("global", "us-east-2"),
736 "api_gateway": deployment_regions.get("api_gateway", "us-east-2"),
737 "monitoring": deployment_regions.get("monitoring", "us-east-2"),
738 "regional": deployment_regions.get("regional", ["us-east-1"]),
739 }
741 def get_deployment_partition(self) -> str:
742 """Return the one SDK partition shared by every configured Region."""
743 deployment_regions = self.get_deployment_regions()
744 regional = validated_regional_deployment_regions(
745 deployment_regions["regional"],
746 known_regions=self.VALID_REGIONS,
747 )
748 return validated_deployment_partition(
749 (
750 deployment_regions["global"],
751 deployment_regions["api_gateway"],
752 deployment_regions["monitoring"],
753 *regional,
754 )
755 )
757 def supports_global_accelerator(self) -> bool:
758 """Return whether this partition exposes the Global Accelerator topology."""
759 return self.get_deployment_partition() == "aws"
761 def get_global_region(self) -> str:
762 """Get the region for global resources and shared SSM parameters."""
763 region = self.get_deployment_regions()["global"]
764 return str(region)
766 def get_api_gateway_region(self) -> str:
767 """Get the region for API Gateway stack."""
768 region = self.get_deployment_regions()["api_gateway"]
769 return str(region)
771 def get_monitoring_region(self) -> str:
772 """Get the region for Monitoring stack."""
773 region = self.get_deployment_regions()["monitoring"]
774 return str(region)
776 def get_regions(self) -> list[str]:
777 """Get list of regions for EKS cluster deployment."""
778 deployment_regions = self.get_deployment_regions()
779 regional = deployment_regions["regional"]
780 return list(regional) if isinstance(regional, list) else [str(regional)]
782 def get_kubernetes_version(self) -> str:
783 """Get Kubernetes version from configuration"""
784 return self.app.node.try_get_context("kubernetes_version") or "1.36"
786 def get_resource_thresholds(self) -> ResourceThresholds:
787 """Get resource thresholds configuration"""
788 thresholds_config = self.app.node.try_get_context("resource_thresholds") or {
789 "cpu_threshold": 80,
790 "memory_threshold": 80,
791 "gpu_threshold": -1,
792 "pending_pods_threshold": 10,
793 "pending_requested_cpu_vcpus": 100,
794 "pending_requested_memory_gb": 200,
795 "pending_requested_gpus": -1,
796 }
797 return ResourceThresholds(
798 cpu_threshold=thresholds_config["cpu_threshold"],
799 memory_threshold=thresholds_config["memory_threshold"],
800 gpu_threshold=thresholds_config["gpu_threshold"],
801 pending_pods_threshold=thresholds_config.get("pending_pods_threshold", 10),
802 pending_requested_cpu_vcpus=thresholds_config.get("pending_requested_cpu_vcpus", 100),
803 pending_requested_memory_gb=thresholds_config.get("pending_requested_memory_gb", 200),
804 pending_requested_gpus=thresholds_config.get("pending_requested_gpus", 8),
805 )
807 def get_cluster_config(self, region: str) -> ClusterConfig:
808 """Get complete cluster configuration for a region"""
809 return ClusterConfig(
810 region=region,
811 cluster_name=f"{self.get_project_name()}-{region}",
812 kubernetes_version=self.get_kubernetes_version(),
813 addons=["metrics-server"],
814 resource_thresholds=self.get_resource_thresholds(),
815 )
817 def get_global_accelerator_config(self) -> dict[str, Any]:
818 """Get Global Accelerator configuration"""
819 return self.app.node.try_get_context("global_accelerator") or {
820 "name": f"{self.get_project_name()}-accelerator",
821 "health_check_grace_period": 30,
822 "health_check_interval": 30,
823 "health_check_timeout": 5,
824 "health_check_path": "/api/v1/health",
825 "client_affinity": "NONE",
826 }
828 def get_backend_tls_config(self) -> dict[str, Any]:
829 """Return the mandatory deployment-local backend TLS lifecycle policy."""
830 defaults = {
831 "root_generation": 1,
832 "root_validity_days": 3_650,
833 "root_rotate_before_days": 180,
834 "root_activation_delay_hours": 24,
835 "root_overlap_days": 45,
836 "leaf_validity_days": 30,
837 "leaf_rotate_before_days": 10,
838 "rotation_schedule_hours": 12,
839 "trust_cache_ttl_seconds": 300,
840 "trust_cache_max_stale_seconds": 3_600,
841 }
842 configured = self.app.node.try_get_context("backend_tls") or {}
843 if not isinstance(configured, dict): 843 ↛ 844line 843 didn't jump to line 844 because the condition on line 843 was never true
844 raise ConfigValidationError("backend_tls must be a mapping")
845 return {**defaults, **configured}
847 def get_alb_config(self) -> dict[str, Any]:
848 """Get ALB configuration"""
849 return self.app.node.try_get_context("alb_config") or {
850 "health_check_interval": 30,
851 "health_check_timeout": 5,
852 "healthy_threshold": 2,
853 "unhealthy_threshold": 2,
854 }
856 def get_manifest_processor_config(self) -> dict[str, Any]:
857 """Get manifest processor configuration.
859 Merges three cdk.json sections into a single runtime config:
861 - ``manifest_processor``: service-specific settings (replicas, image,
862 resource_limits, allowed_namespaces, validation_enabled,
863 max_request_body_bytes, yaml_max_depth)
864 - ``job_validation_policy``: shared validation policy (resource_quotas,
865 trusted_registries, trusted_dockerhub_orgs, manifest_security_policy,
866 allowed_kinds). Pulled in verbatim so the REST path reads the same
867 policy the SQS queue processor enforces.
869 Note: The 'image' field is a placeholder default. In practice, the actual
870 image is built from dockerfiles/manifest-processor-dockerfile and pushed
871 to ECR during CDK deployment. The {{MANIFEST_PROCESSOR_IMAGE}} placeholder
872 in manifests is replaced with the ECR image URI.
873 """
874 default_config = {
875 "image": "gco/manifest-processor:latest", # Placeholder, replaced by ECR image
876 "replicas": 3,
877 "resource_limits": {"cpu": "1000m", "memory": "2Gi"},
878 "validation_enabled": True,
879 "max_request_body_bytes": DEFAULT_MAX_REQUEST_BODY_BYTES,
880 "central_queue_worker_enabled": True,
881 "central_queue_poll_interval_seconds": 10,
882 "central_queue_batch_size": 5,
883 "central_queue_reconcile_limit": 100,
884 "central_queue_lease_seconds": 300,
885 "central_queue_lease_renewal_seconds": 60,
886 # allowed_namespaces, resource_quotas, trusted_registries,
887 # trusted_dockerhub_orgs, manifest_security_policy, and
888 # allowed_kinds are merged in below from job_validation_policy.
889 "allowed_namespaces": ["gco-jobs"],
890 "resource_quotas": {
891 "max_cpu_per_manifest": "10",
892 "max_memory_per_manifest": "32Gi",
893 "max_gpu_per_manifest": 4,
894 },
895 "trusted_registries": [
896 "docker.io",
897 "gcr.io",
898 "quay.io",
899 "registry.k8s.io",
900 "k8s.gcr.io",
901 "public.ecr.aws",
902 "nvcr.io",
903 "gco",
904 ],
905 "trusted_dockerhub_orgs": [
906 "nvidia",
907 "pytorch",
908 "rayproject",
909 "tensorflow",
910 "huggingface",
911 "amazon",
912 "bitnami",
913 ],
914 }
915 context_config = self.app.node.try_get_context("manifest_processor") or {}
917 # Merge in the shared job_validation_policy section. These keys apply
918 # to BOTH the manifest processor and the queue processor; they live
919 # in their own top-level cdk.json section so neither service "owns"
920 # them. We flatten them into the manifest processor's runtime config
921 # so service code keeps its existing attribute layout.
922 shared_policy = self.app.node.try_get_context("job_validation_policy") or {}
923 merged = {**default_config, **context_config, **shared_policy}
925 enabled = merged.get("central_queue_worker_enabled")
926 if not isinstance(enabled, bool): 926 ↛ 927line 926 didn't jump to line 927 because the condition on line 926 was never true
927 raise ConfigValidationError(
928 "manifest_processor.central_queue_worker_enabled must be a boolean"
929 )
930 for key, minimum, maximum in (
931 ("central_queue_poll_interval_seconds", 1, 300),
932 ("central_queue_batch_size", 1, 20),
933 ("central_queue_reconcile_limit", 1, 500),
934 ("central_queue_lease_seconds", 30, 3600),
935 ("central_queue_lease_renewal_seconds", 1, 300),
936 ):
937 value = merged.get(key)
938 if type(value) is not int or not minimum <= value <= maximum: 938 ↛ 939line 938 didn't jump to line 939 because the condition on line 938 was never true
939 raise ConfigValidationError(
940 f"manifest_processor.{key} must be an integer between {minimum} and {maximum}"
941 )
942 if ( 942 ↛ 946line 942 didn't jump to line 946 because the condition on line 942 was never true
943 merged["central_queue_lease_renewal_seconds"] * 2
944 > merged["central_queue_lease_seconds"]
945 ):
946 raise ConfigValidationError(
947 "manifest_processor.central_queue_lease_renewal_seconds must be no more than "
948 "half of central_queue_lease_seconds"
949 )
950 return merged
952 def get_api_gateway_config(self) -> dict[str, Any]:
953 """Get API Gateway configuration.
955 Returns:
956 API Gateway configuration dictionary with the following keys:
957 - throttle_rate_limit: Requests per second limit
958 - throttle_burst_limit: Burst capacity
959 - log_level: CloudWatch logging level (OFF, ERROR, INFO)
960 - metrics_enabled: Enable CloudWatch metrics
961 - tracing_enabled: Enable X-Ray tracing
962 - regional_api_enabled: In the commercial ``aws`` partition,
963 permit direct same-account callers to use the always-deployed
964 regional API bridges. Other partitions force this access on
965 because the bridges are the supported workload ingress without
966 Global Accelerator. Centralized aggregation always uses them.
967 """
968 default_config = {
969 "throttle_rate_limit": 1000,
970 "throttle_burst_limit": 2000,
971 "log_level": "INFO",
972 "metrics_enabled": True,
973 "tracing_enabled": True,
974 "regional_api_enabled": False,
975 }
976 return {**default_config, **(self.app.node.try_get_context("api_gateway") or {})}
978 def get_eks_cluster_config(self) -> dict[str, Any]:
979 """Get EKS cluster configuration.
981 Returns:
982 EKS cluster configuration dictionary with the following keys:
983 - endpoint_access: EKS API endpoint access mode
984 - "PRIVATE": API server only accessible from within VPC (default, most secure)
985 - "PUBLIC_AND_PRIVATE": API server accessible from internet and VPC
987 Note:
988 PRIVATE endpoint is recommended for production. Job submission still works
989 via API Gateway → Lambda (in VPC) or SQS queues. For kubectl access with
990 PRIVATE endpoint, use a bastion host, VPN, or AWS SSM Session Manager.
991 """
992 default_config = {
993 "endpoint_access": "PRIVATE",
994 }
995 return {**default_config, **(self.app.node.try_get_context("eks_cluster") or {})}
997 def get_fsx_lustre_config(self, region: str | None = None) -> dict[str, Any]:
998 """Get FSx for Lustre configuration.
1000 Args:
1001 region: Optional region to get config for. If provided, checks for
1002 region-specific overrides first.
1004 Returns:
1005 FSx configuration dictionary with the following keys:
1006 - enabled: Whether FSx is enabled
1007 - storage_capacity_gib: Storage capacity in GiB (min 1200)
1008 - deployment_type: SCRATCH_1, SCRATCH_2, PERSISTENT_1, PERSISTENT_2
1009 - file_system_type_version: Lustre version (2.12 or 2.15, default: 2.15)
1010 IMPORTANT: Use 2.15 for kernel 6.x compatibility (AL2023, Bottlerocket)
1011 - per_unit_storage_throughput: Throughput for PERSISTENT types
1012 - data_compression_type: LZ4 or NONE
1013 - import_path: S3 path for data import
1014 - export_path: S3 path for data export
1015 - auto_import_policy: NEW, NEW_CHANGED, NEW_CHANGED_DELETED
1016 - node_group: Node group configuration for FSx workloads
1017 - instance_types: List of instance types
1018 - min_size: Minimum nodes (default: 0)
1019 - max_size: Maximum nodes (default: 10)
1020 - desired_size: Desired nodes (default: 0, scales from zero)
1021 - ami_type: AMI type - one of:
1022 AL2023_X86_64_STANDARD (default), AL2023_ARM_64_STANDARD,
1023 AL2023_X86_64_NVIDIA, AL2023_ARM_64_NVIDIA, AL2023_X86_64_NEURON
1024 - capacity_type: ON_DEMAND (default) or SPOT
1025 - disk_size: Root disk size in GB (default: 100)
1026 - labels: Additional node labels (dict)
1027 """
1028 default_config = {
1029 "enabled": False,
1030 "storage_capacity_gib": 1200,
1031 "deployment_type": "SCRATCH_2",
1032 "file_system_type_version": "2.15", # Use 2.15 for kernel 6.x compatibility
1033 "per_unit_storage_throughput": 200,
1034 "data_compression_type": "LZ4",
1035 "import_path": None,
1036 "export_path": None,
1037 "auto_import_policy": "NEW_CHANGED_DELETED",
1038 "node_group": {
1039 "instance_types": ["m5.large", "m5.xlarge", "m6i.large", "m6i.xlarge"],
1040 "min_size": 0,
1041 "max_size": 10,
1042 "desired_size": 1,
1043 "ami_type": "AL2023_X86_64_STANDARD",
1044 "capacity_type": "ON_DEMAND",
1045 "disk_size": 100,
1046 "labels": {},
1047 },
1048 }
1050 # Get global FSx config
1051 global_ctx = self.app.node.try_get_context("fsx_lustre")
1052 global_config: dict[str, Any] = global_ctx if isinstance(global_ctx, dict) else {}
1053 merged_config: dict[str, Any] = {**default_config, **global_config}
1055 # Ensure node_group has all required fields with defaults
1056 if "node_group" in global_config:
1057 global_node_group = global_config["node_group"]
1058 if isinstance(global_node_group, dict): 1058 ↛ 1066line 1058 didn't jump to line 1066 because the condition on line 1058 was always true
1059 default_node_group = cast(dict[str, Any], default_config["node_group"])
1060 merged_config["node_group"] = {
1061 **default_node_group,
1062 **global_node_group,
1063 }
1065 # Check for region-specific override
1066 if region:
1067 region_overrides_ctx = self.app.node.try_get_context("fsx_lustre_regions")
1068 region_overrides: dict[str, Any] = (
1069 region_overrides_ctx if isinstance(region_overrides_ctx, dict) else {}
1070 )
1071 if region in region_overrides:
1072 region_config = region_overrides[region]
1073 if isinstance(region_config, dict): 1073 ↛ 1089line 1073 didn't jump to line 1089 because the condition on line 1073 was always true
1074 merged_config = {**merged_config, **region_config}
1075 # Handle nested node_group override
1076 if "node_group" in region_config:
1077 region_node_group = region_config["node_group"]
1078 if isinstance(region_node_group, dict): 1078 ↛ 1089line 1078 didn't jump to line 1089 because the condition on line 1078 was always true
1079 existing_node_group = merged_config.get("node_group")
1080 if isinstance(existing_node_group, dict): 1080 ↛ 1083line 1080 didn't jump to line 1083 because the condition on line 1080 was always true
1081 base_node_group = existing_node_group
1082 else:
1083 base_node_group = cast(dict[str, Any], default_config["node_group"])
1084 merged_config["node_group"] = {
1085 **base_node_group,
1086 **region_node_group,
1087 }
1089 return merged_config
1091 def get_valkey_config(self) -> dict[str, Any]:
1092 """Get Valkey Serverless cache configuration.
1094 Returns:
1095 Valkey configuration dictionary with the following keys:
1096 - enabled: Whether Valkey cache is enabled (default: False)
1097 - max_data_storage_gb: Maximum data storage in GB (default: 5)
1098 - max_ecpu_per_second: Maximum ECPUs per second (default: 5000)
1099 - snapshot_retention_limit: Daily snapshots to retain (default: 1)
1100 """
1101 default_config: dict[str, Any] = {
1102 "enabled": False,
1103 "max_data_storage_gb": 5,
1104 "max_ecpu_per_second": 5000,
1105 "snapshot_retention_limit": 1,
1106 }
1107 valkey_ctx = self.app.node.try_get_context("valkey")
1108 valkey_config: dict[str, Any] = valkey_ctx if isinstance(valkey_ctx, dict) else {}
1109 return {**default_config, **valkey_config}
1111 def get_aurora_pgvector_config(self) -> dict[str, Any]:
1112 """Get Aurora Serverless v2 + pgvector vector database configuration.
1114 Returns:
1115 Aurora pgvector configuration dictionary with the following keys:
1116 - enabled: Whether Aurora pgvector is enabled (default: False)
1117 - min_acu: Minimum Aurora Capacity Units (default: 0, scales to zero)
1118 - max_acu: Maximum Aurora Capacity Units (default: 16)
1119 - backup_retention_days: Number of days to retain automated backups (default: 7)
1120 - deletion_protection: Whether deletion protection is enabled (default: False)
1121 """
1122 default_config: dict[str, Any] = {
1123 "enabled": False,
1124 "min_acu": 0,
1125 "max_acu": 16,
1126 "backup_retention_days": 7,
1127 "deletion_protection": False,
1128 }
1129 aurora_ctx = self.app.node.try_get_context("aurora_pgvector")
1130 aurora_config: dict[str, Any] = aurora_ctx if isinstance(aurora_ctx, dict) else {}
1131 return {**default_config, **aurora_config}
1133 def get_analytics_config(self) -> dict[str, Any]:
1134 """Get optional analytics environment configuration.
1136 Returns the fully-merged analytics_environment block from cdk.json
1137 layered on top of the defaults below. Sub-blocks (``hyperpod``,
1138 ``cognito``, ``efs``, ``studio``) are deep-merged so a user who
1139 overrides a single nested key (e.g. ``cognito.domain_prefix``) does
1140 not inadvertently wipe the sub-block's other defaults — mirroring the
1141 nested-merge pattern used by ``get_fsx_lustre_config`` for its
1142 ``node_group`` sub-block.
1144 Returns:
1145 Analytics configuration dictionary with the following keys:
1146 - enabled: Whether the analytics environment stack is deployed
1147 (default: False — the feature is off unless explicitly opted in)
1148 - hyperpod: SageMaker HyperPod integration sub-block
1149 - enabled: Whether to add the HyperPod IAM grants to
1150 SageMaker_Execution_Role (default: False)
1151 - canvas: SageMaker Canvas integration sub-block
1152 - enabled: Whether to enable the SageMaker Canvas app on
1153 the Studio domain and attach ``AmazonSageMakerCanvasFullAccess``
1154 to the SageMaker_Execution_Role (default: False)
1155 - cognito: Cognito user-pool sub-block
1156 - domain_prefix: UserPoolDomain prefix, or None to let the
1157 analytics stack derive one (default: None)
1158 - removal_policy: "destroy" (default) or "retain" — controls
1159 the Cognito pool's CloudFormation DeletionPolicy
1160 - efs: Studio_EFS sub-block
1161 - removal_policy: "destroy" (default) or "retain" — controls
1162 the Studio EFS file system's CloudFormation DeletionPolicy
1163 - studio: SageMaker Studio sub-block
1164 - user_profile_name_prefix: Optional prefix for per-user
1165 profile names, or None to use the Cognito username verbatim
1166 (default: None)
1167 """
1168 default_config: dict[str, Any] = {
1169 "enabled": False,
1170 "hyperpod": {"enabled": False},
1171 "canvas": {"enabled": False},
1172 "cognito": {"domain_prefix": None, "removal_policy": "destroy"},
1173 "efs": {"removal_policy": "destroy"},
1174 "studio": {"user_profile_name_prefix": None},
1175 }
1176 analytics_ctx = self.app.node.try_get_context("analytics_environment")
1177 analytics_config: dict[str, Any] = analytics_ctx if isinstance(analytics_ctx, dict) else {}
1178 merged_config: dict[str, Any] = {**default_config, **analytics_config}
1180 # Deep-merge each nested sub-block so a partial override does not
1181 # drop the other defaults in the same sub-block.
1182 for sub_block in ("hyperpod", "canvas", "cognito", "efs", "studio"):
1183 override = analytics_config.get(sub_block)
1184 if isinstance(override, dict):
1185 default_sub = cast(dict[str, Any], default_config[sub_block])
1186 merged_config[sub_block] = {**default_sub, **override}
1188 return merged_config
1190 def get_analytics_enabled(self) -> bool:
1191 """Return whether the analytics environment stack is enabled.
1193 Thin wrapper around ``get_analytics_config()["enabled"]`` to mirror
1194 the existing ``get_valkey_config`` / ``get_aurora_pgvector_config``
1195 access pattern without forcing every call site to index into the
1196 merged dict.
1197 """
1198 return bool(self.get_analytics_config()["enabled"])
1200 def get_cluster_observability_config(self) -> dict[str, Any]:
1201 """Get the in-cluster observability configuration.
1203 Returns the fully-merged cluster_observability block from cdk.json
1204 layered on top of the defaults below. Sub-blocks (``grafana``,
1205 ``prometheus``, ``alertmanager``) are deep-merged so a user who
1206 overrides a single nested key (e.g. ``prometheus.retention``) does not
1207 inadvertently wipe the sub-block's other defaults — mirroring the
1208 nested-merge pattern used by ``get_analytics_config``.
1210 Unlike most optional features, this one is **on by default**: a stock
1211 deployment installs kube-prometheus-stack on every regional cluster.
1212 Operators opt out by setting ``cluster_observability.enabled = false``.
1214 Returns:
1215 Cluster observability configuration dictionary with the keys:
1216 - enabled: Whether kube-prometheus-stack is installed per region
1217 (default: True)
1218 - grafana: Grafana sub-block
1219 - persistence_size: EBS PVC size for Grafana's user database
1220 and dashboards (default: "10Gi")
1221 - admin_user: Grafana admin username; the password is
1222 chart-generated in the <release>-grafana Secret, never
1223 authored here (default: "admin")
1224 - admin_password_rotation_schedule: 5-field cron for the
1225 in-cluster CronJob that rotates the chart-generated admin
1226 password (default: "0 4 1 * *", monthly)
1227 - prometheus: Prometheus sub-block
1228 - persistence_size: EBS PVC size for the Prometheus TSDB
1229 (default: "50Gi")
1230 - retention: Prometheus retention window (default: "15d")
1231 - alertmanager: Alertmanager sub-block
1232 - enabled: Whether Alertmanager is deployed (default: True)
1233 - persistence_size: EBS PVC size for Alertmanager (default: "5Gi")
1234 """
1235 default_config: dict[str, Any] = {
1236 "enabled": True,
1237 "grafana": {
1238 "persistence_size": "10Gi",
1239 "admin_user": "admin",
1240 # Monthly (04:00 on the 1st) rotation of the chart-generated
1241 # Grafana admin password, run by an in-cluster CronJob.
1242 "admin_password_rotation_schedule": "0 4 1 * *",
1243 },
1244 "prometheus": {"persistence_size": "50Gi", "retention": "15d"},
1245 "alertmanager": {"enabled": True, "persistence_size": "5Gi"},
1246 }
1247 obs_ctx = self.app.node.try_get_context("cluster_observability")
1248 obs_config: dict[str, Any] = obs_ctx if isinstance(obs_ctx, dict) else {}
1249 merged_config: dict[str, Any] = {**default_config, **obs_config}
1251 # Deep-merge each nested sub-block so a partial override does not
1252 # drop the other defaults in the same sub-block.
1253 for sub_block in ("grafana", "prometheus", "alertmanager"):
1254 override = obs_config.get(sub_block)
1255 if isinstance(override, dict):
1256 default_sub = cast(dict[str, Any], default_config[sub_block])
1257 merged_config[sub_block] = {**default_sub, **override}
1259 return merged_config
1261 def get_cluster_observability_enabled(self) -> bool:
1262 """Return whether in-cluster observability is enabled (default True).
1264 Thin wrapper around ``get_cluster_observability_config()["enabled"]``
1265 so call sites (the regional stack's chart-enable and value-override
1266 methods, the CLI) do not have to index into the merged dict.
1267 """
1268 return bool(self.get_cluster_observability_config()["enabled"])
1270 def get_cost_monitoring_config(self) -> dict[str, Any]:
1271 """Get the cost monitoring configuration.
1273 Returns the fully-merged ``cost_monitoring`` block from cdk.json
1274 layered on top of the defaults below. Sub-blocks (``reports``,
1275 ``athena``) are deep-merged so a user who overrides a single nested
1276 key does not inadvertently wipe the sub-block's other defaults —
1277 mirroring the nested-merge pattern used by
1278 ``get_cluster_observability_config``.
1280 Like cluster observability, cost monitoring is **on by default**: a
1281 stock deployment installs OpenCost per region, provisions the cost
1282 report bucket + Athena analytics in the monitoring stack, and runs
1283 the cost-monitor service on every regional cluster. Operators opt out
1284 by setting ``cost_monitoring.enabled = false``.
1286 Returns:
1287 Cost monitoring configuration dictionary with the keys:
1288 - enabled: Whether the cost monitoring pipeline is deployed
1289 (default: True). Requires ``cluster_observability.enabled``.
1290 - reports: Cost report sub-block
1291 - interval_minutes: cadence of the cost-monitor service's
1292 scheduled Parquet reports (default: 60)
1293 - retention_days: S3 lifecycle expiration for report objects
1294 (default: 365)
1295 - transition_to_infrequent_access_days: S3 lifecycle transition
1296 to STANDARD_IA (default: 90; must be < retention_days)
1297 - athena: Athena analytics sub-block
1298 - query_results_retention_days: S3 lifecycle expiration for
1299 Athena query results written under ``athena-results/``
1300 (default: 30)
1301 """
1302 default_config: dict[str, Any] = {
1303 "enabled": True,
1304 "reports": {
1305 "interval_minutes": 60,
1306 "retention_days": 365,
1307 "transition_to_infrequent_access_days": 90,
1308 },
1309 "athena": {
1310 "query_results_retention_days": 30,
1311 },
1312 }
1313 cost_ctx = self.app.node.try_get_context("cost_monitoring")
1314 cost_config: dict[str, Any] = cost_ctx if isinstance(cost_ctx, dict) else {}
1315 merged_config: dict[str, Any] = {**default_config, **cost_config}
1317 # Deep-merge each nested sub-block so a partial override does not
1318 # drop the other defaults in the same sub-block.
1319 for sub_block in ("reports", "athena"):
1320 override = cost_config.get(sub_block)
1321 if isinstance(override, dict):
1322 default_sub = cast(dict[str, Any], default_config[sub_block])
1323 merged_config[sub_block] = {**default_sub, **override}
1325 return merged_config
1327 def get_cost_monitoring_enabled(self) -> bool:
1328 """Return whether the cost monitoring pipeline is effectively enabled.
1330 The conjunction of ``cost_monitoring.enabled`` (default True) and
1331 ``cluster_observability.enabled`` (default True): OpenCost reads its
1332 usage data from the in-cluster Prometheus, so disabling observability
1333 switches the whole cost pipeline off with it rather than deploying a
1334 pipeline with no data source (or failing synthesis for a user who
1335 only ran ``gco monitoring disable``). Call sites — the regional
1336 stack's chart-enable and image-build methods, the monitoring stack,
1337 the CLI — all gate on this one conjunction.
1338 """
1339 return bool(self.get_cost_monitoring_config()["enabled"]) and bool(
1340 self.get_cluster_observability_config()["enabled"]
1341 )
1343 def get_capacity_history_config(self) -> dict[str, Any]:
1344 """Get the optional historical capacity surface configuration.
1346 Returns the merged ``historical`` block from cdk.json layered on top of
1347 the defaults below. The feature is off unless ``historical.enabled`` is
1348 explicitly true, mirroring the analytics-environment opt-in pattern.
1350 Keys:
1351 - enabled: deploy the capacity poller stack + history table (default False)
1352 - retention_days: DynamoDB TTL window for snapshots (default 90)
1353 - poll_interval_minutes: EventBridge schedule cadence (default 15)
1354 - capacity_block_duration_hours: short Capacity Block probe duration
1355 the poller snapshots (default 24 = 1 day)
1356 - capacity_block_long_duration_hours: long Capacity Block probe
1357 duration in hours (default 1512 = 63 days); 0 disables the long
1358 probe and its ``capacity_blocks_long_*`` metrics
1359 - watch_instance_types: instance types the poller snapshots
1360 - enabled_regions: regions to poll; empty means all deployed regions
1361 """
1362 default_config: dict[str, Any] = {
1363 "enabled": False,
1364 "retention_days": 90,
1365 "poll_interval_minutes": 15,
1366 "capacity_block_duration_hours": 24,
1367 "capacity_block_long_duration_hours": 63 * 24,
1368 "watch_instance_types": [
1369 "g4dn.12xlarge",
1370 "g4dn.16xlarge",
1371 "g4dn.2xlarge",
1372 "g4dn.4xlarge",
1373 "g4dn.8xlarge",
1374 "g4dn.metal",
1375 "g4dn.xlarge",
1376 "g5.12xlarge",
1377 "g5.16xlarge",
1378 "g5.24xlarge",
1379 "g5.2xlarge",
1380 "g5.48xlarge",
1381 "g5.4xlarge",
1382 "g5.8xlarge",
1383 "g5.xlarge",
1384 "g5g.16xlarge",
1385 "g5g.2xlarge",
1386 "g5g.4xlarge",
1387 "g5g.8xlarge",
1388 "g5g.metal",
1389 "g5g.xlarge",
1390 "g6.12xlarge",
1391 "g6.16xlarge",
1392 "g6.24xlarge",
1393 "g6.2xlarge",
1394 "g6.48xlarge",
1395 "g6.4xlarge",
1396 "g6.8xlarge",
1397 "g6.xlarge",
1398 "g6e.12xlarge",
1399 "g6e.16xlarge",
1400 "g6e.24xlarge",
1401 "g6e.2xlarge",
1402 "g6e.48xlarge",
1403 "g6e.4xlarge",
1404 "g6e.8xlarge",
1405 "g6e.xlarge",
1406 "g6f.2xlarge",
1407 "g6f.4xlarge",
1408 "g6f.large",
1409 "g6f.xlarge",
1410 "g7.12xlarge",
1411 "g7.24xlarge",
1412 "g7.2xlarge",
1413 "g7.48xlarge",
1414 "g7.4xlarge",
1415 "g7.8xlarge",
1416 "g7e.12xlarge",
1417 "g7e.24xlarge",
1418 "g7e.2xlarge",
1419 "g7e.48xlarge",
1420 "g7e.4xlarge",
1421 "g7e.8xlarge",
1422 "gr6.4xlarge",
1423 "gr6.8xlarge",
1424 "gr6f.4xlarge",
1425 "inf1.24xlarge",
1426 "inf1.2xlarge",
1427 "inf1.6xlarge",
1428 "inf1.xlarge",
1429 "inf2.24xlarge",
1430 "inf2.48xlarge",
1431 "inf2.8xlarge",
1432 "inf2.xlarge",
1433 "p3dn.24xlarge",
1434 "p4d.24xlarge",
1435 "p4de.24xlarge",
1436 "p5.48xlarge",
1437 "p5.4xlarge",
1438 "p5e.48xlarge",
1439 "p5en.48xlarge",
1440 "p6-b200.48xlarge",
1441 "p6-b300.48xlarge",
1442 "trn1.2xlarge",
1443 "trn1.32xlarge",
1444 "trn1n.32xlarge",
1445 "trn2.3xlarge",
1446 "trn2.48xlarge",
1447 ],
1448 "enabled_regions": [],
1449 }
1450 historical_ctx = self.app.node.try_get_context("historical")
1451 historical_config = historical_ctx if isinstance(historical_ctx, dict) else {}
1452 return {**default_config, **historical_config}
1454 def get_capacity_history_enabled(self) -> bool:
1455 """Return whether the historical capacity surface is enabled."""
1456 return bool(self.get_capacity_history_config()["enabled"])
1458 def get_tags(self) -> dict[str, str]:
1459 """Get common tags from configuration"""
1460 return self.app.node.try_get_context("tags") or {}
1462 def validate_region_availability(self, region: str) -> bool:
1463 """Validate that a region is available in the current AWS account"""
1464 try:
1465 ec2 = boto3.client("ec2", region_name=region)
1466 ec2.describe_regions(RegionNames=[region])
1467 return True
1468 except Exception as e:
1469 logger.debug("Region %s not available: %s", region, e)
1470 return False
1472 def get_available_regions(self) -> list[str]:
1473 """Get list of available AWS regions for the current account"""
1474 try:
1475 ec2 = boto3.client("ec2")
1476 response = ec2.describe_regions()
1477 return [region["RegionName"] for region in response["Regions"]]
1478 except Exception as e:
1479 logger.debug("Failed to list regions, using defaults: %s", e)
1480 return list(self.VALID_REGIONS)