Coverage for gco / config / config_loader.py: 100.00%
786 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +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- inference_proxy: Shared inference TLS proxy CPU request and HPA target
16- manifest_processor: Manifest validation and resource limits
17- api_gateway: Throttling and logging configuration
18- tracing: OpenTelemetry -> AWS X-Ray tracing for the API services
19- tags: Common tags applied to all resources
21Usage:
22 config = ConfigLoader(app)
23 regions = config.get_regions()
24 cluster_config = config.get_cluster_config("us-east-1")
25"""
27from __future__ import annotations
29import copy
30import json
31import logging
32import math
33import re
34from collections.abc import Mapping
35from typing import Any, cast
37import boto3
38from aws_cdk import App
40from gco.argocd_config import ArgoCdConfigError, validate_argocd_config
41from gco.eks_capabilities_config import (
42 EKS_CAPABILITIES_CONTEXT_KEY,
43 EKS_CAPABILITIES_OVERRIDES_CONTEXT_KEY,
44 EksCapabilitiesConfigError,
45 merge_eks_capabilities_overrides,
46 parse_eks_capabilities_overrides,
47 validate_eks_capabilities_config,
48)
49from gco.inference_proxy_config import (
50 INFERENCE_PROXY_MAX_REPLICAS_DEFAULT,
51 INFERENCE_PROXY_MIN_REPLICAS_DEFAULT,
52 INFERENCE_PROXY_TLS_CPU_REQUEST_MILLICORES_DEFAULT,
53 INFERENCE_PROXY_TLS_CPU_TARGET_UTILIZATION_DEFAULT,
54)
55from gco.manifest_security_policy import validate_manifest_security_policy
56from gco.models import ClusterConfig, ResourceThresholds
57from gco.resource_governance import parse_k8s_quantity
58from gco.slurm_config import SlurmConfigError, validate_slurm_config
59from gco.stacks.constants import (
60 CONTROL_PLANE_ONLY_CONTEXT_KEY,
61 DEFAULT_MAX_REQUEST_BODY_BYTES,
62 control_plane_only_requested,
63 known_cloudformation_regions,
64 validated_deployment_partition,
65 validated_regional_deployment_regions,
66 validated_request_body_limit,
67)
69logger = logging.getLogger(__name__)
71#: ``manifest_processor.autoscaling`` defaults. Off: the API tier is I/O-bound
72#: (Kubernetes round trips), so CPU is a weak saturation signal; an operator who
73#: has measured a CPU-bound API turns it on and the HPA owns the replica count
74#: between ``manifest_processor.replicas`` and ``max_replicas``.
75_MANIFEST_PROCESSOR_AUTOSCALING_DEFAULTS: dict[str, Any] = {
76 "enabled": False,
77 "max_replicas": 6,
78 "cpu_target_utilization_percentage": 70,
79}
81#: The ``manifest-processor`` container's fixed requests in
82#: ``lambda/kubectl-applier-simple/manifests/31-manifest-processor.yaml``;
83#: ``manifest_processor.resource_limits`` may not go below them. Pinned to the
84#: manifest by ``tests/test_config_loader.py``.
85_MANIFEST_PROCESSOR_CONTAINER_REQUESTS: dict[str, str] = {
86 "cpu": "500m",
87 "memory": "1Gi",
88}
90#: Gateway VPC endpoints ``vpc_endpoints.gateway`` may name. Free, route-table
91#: based, same-region only.
92VPC_GATEWAY_ENDPOINT_SERVICES: tuple[str, ...] = ("s3", "dynamodb")
94#: Interface (PrivateLink) endpoints ``vpc_endpoints.interface`` may name,
95#: keyed the way the regional stack maps them onto
96#: ``ec2.InterfaceVpcEndpointAwsService``. Each costs per AZ-hour plus per GB,
97#: so the list is opt-in; together they keep every AWS API call the platform
98#: and its jobs make inside the VPC (the cross-region calls to the global
99#: region's tables and buckets still leave through the NAT gateways).
100VPC_INTERFACE_ENDPOINT_SERVICES: tuple[str, ...] = (
101 "sts",
102 "ecr.api",
103 "ecr.dkr",
104 "logs",
105 "monitoring",
106 "sqs",
107 "ssm",
108 "secretsmanager",
109 "kms",
110 "eks",
111 "elasticfilesystem",
112 "bedrock-runtime",
113 # OTLP span export from the traced API services (``tracing`` block).
114 "xray",
115)
117_VPC_ENDPOINTS_DEFAULTS: dict[str, list[str]] = {
118 "gateway": ["s3", "dynamodb"],
119 "interface": [],
120}
122#: CDK context key that force-enables optional infrastructure features for one
123#: deploy without touching cdk.json — the infrastructure sibling of the
124#: ``helm_enabled_overrides`` context handled in ``gco/stacks/regional_stack.py``.
125#: Used by validation harnesses (``gco examples validate``) whose preflight
126#: requires a clean worktree.
127FEATURE_OVERRIDE_CONTEXT_KEY = "feature_enabled_overrides"
129#: The cdk.json blocks whose ``enabled`` flag the override may force on. Kept
130#: deliberately narrow: each of these is a self-contained regional feature the
131#: examples exercise (Aurora pgvector, Valkey Serverless, FSx for Lustre).
132FEATURE_OVERRIDE_KEYS = frozenset({"aurora_pgvector", "valkey", "fsx_lustre", "vector_store"})
135def parse_feature_enabled_overrides(raw: object) -> frozenset[str]:
136 """Parse and validate the ``feature_enabled_overrides`` context value.
138 Accepts a comma-separated string (the only shape the CDK CLI can pass with
139 ``--context``) or a list of strings (cdk.json-style). Unknown names raise
140 at synth time with the valid list — identical semantics to
141 ``_parse_helm_enabled_overrides``.
142 """
143 if raw is None:
144 return frozenset()
145 if isinstance(raw, str):
146 names = [part.strip() for part in raw.split(",") if part.strip()]
147 elif isinstance(raw, list) and all(isinstance(part, str) for part in raw):
148 names = [part.strip() for part in raw if part.strip()]
149 else:
150 raise ConfigValidationError(
151 f"{FEATURE_OVERRIDE_CONTEXT_KEY} must be a comma-separated string or string list"
152 )
153 unknown = sorted(set(names) - FEATURE_OVERRIDE_KEYS)
154 if unknown:
155 valid = ", ".join(sorted(FEATURE_OVERRIDE_KEYS))
156 raise ConfigValidationError(
157 f"Unknown {FEATURE_OVERRIDE_CONTEXT_KEY} name(s): {', '.join(unknown)}. Valid: {valid}"
158 )
159 return frozenset(names)
162#: cdk.json context key for the OpenTelemetry -> AWS X-Ray tracing block.
163TRACING_CONTEXT_KEY = "tracing"
165#: Run-scoped CDK context key carrying a JSON object deep-merged over the
166#: ``tracing`` block for one deploy (``cdk deploy --context
167#: tracing_overrides='{"sample_ratio": 1.0}'``) — the tracing sibling of
168#: ``eks_capabilities_overrides``. The live release harness, whose preflight
169#: requires a clean worktree, records every trace of its run this way.
170TRACING_OVERRIDES_CONTEXT_KEY = "tracing_overrides"
172#: The ``tracing`` block with every knob at its default. Tracing is ON by
173#: default, like cluster observability: an absent block means the shipped
174#: posture, not "off". ``sample_ratio`` is the head-sampling probability of a
175#: new trace; ``enable_transaction_search`` lets each regional stack switch on
176#: CloudWatch Transaction Search, which the X-Ray OTLP endpoint requires, and
177#: only matters while ``enabled`` is true.
178TRACING_DEFAULTS: dict[str, Any] = {
179 "enabled": True,
180 "sample_ratio": 0.05,
181 "enable_transaction_search": True,
182}
185def _deep_merge(base: Mapping[str, Any], overlay: Mapping[str, Any]) -> dict[str, Any]:
186 """Return ``overlay`` recursively merged over a copy of ``base``."""
187 merged: dict[str, Any] = copy.deepcopy(dict(base))
188 for key, value in overlay.items():
189 if isinstance(value, Mapping) and isinstance(merged.get(key), Mapping):
190 merged[key] = _deep_merge(merged[key], value)
191 else:
192 merged[key] = copy.deepcopy(value)
193 return merged
196def parse_tracing_overrides(raw: object) -> dict[str, Any]:
197 """Parse the ``tracing_overrides`` context value into a mapping.
199 Accepts a JSON object string (the only shape ``cdk --context`` can carry)
200 or a mapping (cdk.json-style); ``None`` and blank strings mean no
201 overrides. Only the shape is checked here: the merged block is validated
202 as a whole, so an override that produces an invalid block fails exactly
203 like an invalid cdk.json does.
204 """
205 if raw is None:
206 return {}
207 if isinstance(raw, Mapping):
208 return copy.deepcopy(dict(raw))
209 if isinstance(raw, str):
210 if not raw.strip():
211 return {}
212 try:
213 parsed = json.loads(raw)
214 except json.JSONDecodeError as exc:
215 raise ConfigValidationError(
216 f"{TRACING_OVERRIDES_CONTEXT_KEY} must be a JSON object: {exc}"
217 ) from exc
218 if not isinstance(parsed, dict):
219 raise ConfigValidationError(
220 f"{TRACING_OVERRIDES_CONTEXT_KEY} must be a JSON object, got {type(parsed).__name__}"
221 )
222 return parsed
223 raise ConfigValidationError(
224 f"{TRACING_OVERRIDES_CONTEXT_KEY} must be a JSON object string or an object, "
225 f"got {type(raw).__name__}"
226 )
229def resolve_tracing_config(raw: object, overrides: object = None) -> dict[str, Any]:
230 """Return the validated ``tracing`` block with overrides and defaults applied.
232 Single source of truth shared by :meth:`ConfigLoader.get_tracing_config`
233 and the regional stack, which falls back to it for config doubles that
234 predate the knob. ``overrides`` is the raw ``tracing_overrides`` context
235 value. Types are checked exactly: the flags are rendered verbatim into
236 the ``GCO_TRACING_ENABLED`` env var and a quoted ``"false"`` in cdk.json
237 must not read as enabled, and ``sample_ratio`` must be a real number in
238 ``[0, 1]`` because the services treat it as a probability.
240 Raises:
241 ConfigValidationError: If the block or the overrides are malformed.
242 """
243 if raw is None:
244 block: Mapping[str, Any] = {}
245 elif isinstance(raw, Mapping):
246 block = raw
247 else:
248 raise ConfigValidationError(
249 f"{TRACING_CONTEXT_KEY} must be an object, got {type(raw).__name__}"
250 )
251 parsed_overrides = parse_tracing_overrides(overrides)
252 merged = _deep_merge(_deep_merge(TRACING_DEFAULTS, block), parsed_overrides)
254 unknown = sorted(str(key) for key in merged if key not in TRACING_DEFAULTS)
255 if unknown:
256 source = f" (after {TRACING_OVERRIDES_CONTEXT_KEY})" if parsed_overrides else ""
257 raise ConfigValidationError(
258 f"{TRACING_CONTEXT_KEY}{source} contains unknown key(s): {', '.join(unknown)}; "
259 f"allowed keys: {', '.join(sorted(TRACING_DEFAULTS))}"
260 )
261 for flag in ("enabled", "enable_transaction_search"):
262 if type(merged[flag]) is not bool:
263 raise ConfigValidationError(
264 f"{TRACING_CONTEXT_KEY}.{flag} must be a boolean, got {merged[flag]!r}"
265 )
266 ratio = merged["sample_ratio"]
267 if (
268 isinstance(ratio, bool)
269 or not isinstance(ratio, int | float)
270 or not math.isfinite(ratio)
271 or not 0 <= ratio <= 1
272 ):
273 raise ConfigValidationError(
274 f"{TRACING_CONTEXT_KEY}.sample_ratio must be a number from 0 to 1 (inclusive), "
275 f"got {ratio!r}"
276 )
277 return {
278 "enabled": merged["enabled"],
279 "sample_ratio": float(ratio),
280 "enable_transaction_search": merged["enable_transaction_search"],
281 }
284class ConfigValidationError(Exception):
285 """Raised when configuration validation fails."""
288class ConfigLoader:
289 """
290 Loads and validates configuration from CDK context (cdk.json)
291 """
293 # Keep the public class attribute for compatibility. Endpoint metadata
294 # covers every CloudFormation Region known to the installed AWS SDK; this
295 # is not a project-specific allowlist.
296 VALID_REGIONS = known_cloudformation_regions()
298 def __init__(self, app: App):
299 self.app = app
300 self._validate_configuration()
302 def _validate_configuration(self) -> None:
303 """Validate the entire configuration"""
304 # Check if we have any context at all (might be running outside CDK)
305 project_name = self.app.node.try_get_context("project_name")
306 if project_name is None:
307 # Running outside CDK context, skip validation
308 return
310 # Validate required fields exist
311 required_fields = [
312 "project_name",
313 "kubernetes_version",
314 "resource_thresholds",
315 ]
316 for field in required_fields:
317 if not self.app.node.try_get_context(field):
318 raise ConfigValidationError(f"Required configuration field '{field}' is missing")
320 # Validate project_name format before anything consumes it (#139).
321 self._validate_project_name()
323 # Check for deployment_regions
324 deployment_regions = self.app.node.try_get_context("deployment_regions")
325 if not isinstance(deployment_regions, dict) or not deployment_regions:
326 raise ConfigValidationError(
327 "Required configuration field 'deployment_regions' must be a non-empty object"
328 )
330 # Validate regions
331 self._validate_regions()
333 # Validate resource thresholds
334 self._validate_resource_thresholds()
336 # Validate Global Accelerator config
337 self._validate_global_accelerator_config()
339 # Validate deployment-local backend TLS rotation policy
340 self._validate_backend_tls_config()
342 # Validate inference proxy TLS sidecar autoscaling settings
343 self._validate_inference_proxy_config()
345 # Validate ALB config
346 self._validate_alb_config()
348 # Validate manifest processor config
349 self._validate_manifest_processor_config()
351 # Validate API Gateway config
352 self._validate_api_gateway_config()
354 # Validate EKS cluster config
355 self._validate_eks_cluster_config()
357 # Validate the opt-in EKS Capabilities block (ACK / kro)
358 self._validate_eks_capabilities_config()
359 # Validate the self-managed Argo CD block (helm.argocd)
360 self._validate_argocd_config()
361 # Validate the Slurm cluster block (helm.slurm)
362 self._validate_slurm_config()
364 # Validate analytics environment config (optional block)
365 self._validate_analytics_environment_config()
367 # Validate cluster observability config (optional block)
368 self._validate_cluster_observability_config()
370 # Validate cost monitoring config (optional block)
371 self._validate_cost_monitoring_config()
373 # Validate the OpenTelemetry -> X-Ray tracing block and its run-scoped
374 # overrides (optional; on by default)
375 self._validate_tracing_config()
377 # Validate historical capacity surface config (optional block)
378 self._validate_capacity_history_config()
380 # Validate mission-memory configuration (recall across mission sessions)
381 self._validate_mission_memory_config()
382 # Validate the vector-store configuration (global workload RAG corpus)
383 self._validate_vector_store_config()
385 #: Allowed ``project_name`` format (#139). ``project_name`` is the
386 #: deployment's unique prefix and flows into S3 bucket names, the Cognito
387 #: hosted-UI domain prefix, SSM parameter paths, IAM role names, and
388 #: CloudFormation export names. The tightest of those constraints is S3 /
389 #: Cognito naming (lowercase letters, digits, hyphens; must start with a
390 #: letter), so require: a leading lowercase letter followed by 1–30
391 #: lowercase letters, digits, or hyphens (total length 2–31).
392 PROJECT_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9-]{1,30}$")
394 def _validate_project_name(self) -> None:
395 """Validate ``project_name`` format so misconfigurations fail at synth.
397 ``project_name`` is documented as the deployment's unique identifier
398 and is used as the prefix for nearly every physical resource name. If
399 it contains characters that are illegal in S3 bucket names or Cognito
400 domain prefixes (uppercase, underscores, dots, leading digit, etc.),
401 ``cdk synth`` still succeeds but the deploy fails late with an opaque
402 AWS naming error. Catching it here turns that into an actionable
403 message up front.
404 """
405 project_name = self.app.node.try_get_context("project_name")
406 if not isinstance(project_name, str) or not self.PROJECT_NAME_PATTERN.match(project_name):
407 raise ConfigValidationError(
408 f"Invalid project_name {project_name!r}. It must match "
409 f"{self.PROJECT_NAME_PATTERN.pattern} (start with a lowercase letter, then "
410 "2–31 total characters of lowercase letters, digits, or hyphens). "
411 "project_name is the deployment prefix for S3 buckets, the Cognito "
412 "domain, SSM paths, and CloudFormation exports, so it must be a valid "
413 "lowercase DNS-style label."
414 )
416 def _validate_regions(self) -> None:
417 """Validate region configuration against the shared app/CLI contract."""
418 deployment_regions = self.get_deployment_regions()
419 try:
420 for field in ("global", "api_gateway", "monitoring"):
421 region = deployment_regions[field]
422 if not isinstance(region, str) or region not in self.VALID_REGIONS:
423 raise ValueError(
424 f"Invalid {field} region {region!r}; expected an AWS region with a "
425 "CloudFormation endpoint known to the installed SDK"
426 )
427 regional = validated_regional_deployment_regions(
428 deployment_regions["regional"],
429 known_regions=self.VALID_REGIONS,
430 )
431 validated_deployment_partition(
432 (
433 deployment_regions["global"],
434 deployment_regions["api_gateway"],
435 deployment_regions["monitoring"],
436 *regional,
437 )
438 )
439 except (RuntimeError, ValueError) as exc:
440 raise ConfigValidationError(str(exc)) from exc
442 def _validate_resource_thresholds(self) -> None:
443 """Validate resource threshold configuration"""
444 thresholds_config = self.app.node.try_get_context("resource_thresholds")
446 required_thresholds = ["cpu_threshold", "memory_threshold", "gpu_threshold"]
447 for threshold in required_thresholds:
448 if threshold not in thresholds_config:
449 raise ConfigValidationError(f"Missing threshold configuration: {threshold}")
451 value = thresholds_config[threshold]
452 if not isinstance(value, int) or (value != -1 and not 0 <= value <= 100):
453 raise ConfigValidationError(
454 f"{threshold} must be an integer between 0 and 100 (or -1 to disable), got {value}"
455 )
457 # Validate optional thresholds if present
458 for opt_threshold in [
459 "pending_pods_threshold",
460 "pending_requested_cpu_vcpus",
461 "pending_requested_memory_gb",
462 "pending_requested_gpus",
463 ]:
464 if opt_threshold in thresholds_config:
465 value = thresholds_config[opt_threshold]
466 if not isinstance(value, int) or (value != -1 and value < 0):
467 raise ConfigValidationError(
468 f"{opt_threshold} must be a non-negative integer (or -1 to disable), got {value}"
469 )
471 #: Health-check probe intervals Global Accelerator accepts. The API
472 #: constrains HealthCheckIntervalSeconds to exactly 10 or 30 seconds, so
473 #: any other value must fail at synth instead of at deploy.
474 GLOBAL_ACCELERATOR_HEALTH_CHECK_INTERVALS = frozenset({10, 30})
476 #: Traffic-dial controller modes. ``monitor`` computes and publishes
477 #: per-region dial decisions without mutating Global Accelerator;
478 #: ``enforce`` additionally applies them via UpdateEndpointGroup.
479 TRAFFIC_DIAL_MODES = frozenset({"monitor", "enforce"})
481 def _validate_global_accelerator_config(self) -> None:
482 """Validate the ``global_accelerator`` block in cdk.json.
484 The block is optional; absence means the shipped defaults apply.
485 Validation runs against the *merged* configuration so a partial block
486 is checked together with every default it kept:
488 - ``health_check_interval``: 10 or 30 — the only probe intervals the
489 Global Accelerator API accepts (UpdateEndpointGroup rejects others).
490 - ``health_check_threshold``: integer 1-10 (the API ThresholdCount
491 range).
492 - ``health_check_path``: must start with ``/``.
493 - ``client_affinity``: NONE or SOURCE_IP, case-insensitive.
494 - ``traffic_dial``: see :meth:`_validate_traffic_dial_config`.
496 The legacy ``health_check_grace_period`` and ``health_check_timeout``
497 keys are tolerated and ignored: Global Accelerator endpoint groups
498 have no such settings (the keys were validated but never consumed),
499 so their presence in an existing cdk.json must not fail synthesis.
501 ``name`` is intentionally optional: when omitted it defaults to
502 ``<project_name>-accelerator`` so a second deployment gets a
503 project-scoped name from the single ``project_name`` knob (#139).
504 """
505 ga_config = self.get_global_accelerator_config()
507 interval = ga_config["health_check_interval"]
508 if (
509 not isinstance(interval, int)
510 or isinstance(interval, bool)
511 or interval not in self.GLOBAL_ACCELERATOR_HEALTH_CHECK_INTERVALS
512 ):
513 raise ConfigValidationError(
514 "global_accelerator.health_check_interval must be one of "
515 f"{sorted(self.GLOBAL_ACCELERATOR_HEALTH_CHECK_INTERVALS)} — the only probe "
516 f"intervals the Global Accelerator API accepts — got {interval!r}"
517 )
519 threshold = ga_config["health_check_threshold"]
520 if (
521 not isinstance(threshold, int)
522 or isinstance(threshold, bool)
523 or not 1 <= threshold <= 10
524 ):
525 raise ConfigValidationError(
526 "global_accelerator.health_check_threshold must be an integer between "
527 f"1 and 10, got {threshold!r}"
528 )
530 path = ga_config["health_check_path"]
531 if not isinstance(path, str) or not path.startswith("/"):
532 raise ConfigValidationError("health_check_path must start with '/'")
534 allowed_affinity = {"NONE", "SOURCE_IP"}
535 affinity = ga_config["client_affinity"]
536 if not isinstance(affinity, str) or affinity.upper() not in allowed_affinity:
537 raise ConfigValidationError(
538 f"client_affinity must be one of {sorted(allowed_affinity)}, got {affinity!r}"
539 )
541 self._validate_traffic_dial_config(ga_config["traffic_dial"])
543 def _validate_traffic_dial_config(self, dial_config: Any) -> None:
544 """Validate the merged ``global_accelerator.traffic_dial`` sub-block.
546 Ranges mirror the Global Accelerator API and the controller contract:
548 - ``enabled``: bool (default False — the controller is opt-in).
549 - ``mode``: ``monitor`` or ``enforce``, case-insensitive.
550 - ``interval_minutes`` / ``lookback_minutes``: 1-1440.
551 - ``min_dial_percentage``: 0-100 (TrafficDialPercentage range); the
552 floor a degraded region can be dialed down to.
553 - ``max_step_percentage``: 1-100; the largest change one run applies.
554 - ``full_health_percentage``: 1-100; the healthy fraction (percent)
555 at or above which a region is restored toward 100.
556 """
557 if not isinstance(dial_config, dict):
558 raise ConfigValidationError(
559 "global_accelerator.traffic_dial must be a mapping, got "
560 f"{type(dial_config).__name__}: {dial_config!r}"
561 )
563 enabled = dial_config["enabled"]
564 if not isinstance(enabled, bool):
565 raise ConfigValidationError(
566 "global_accelerator.traffic_dial.enabled must be a bool, got "
567 f"{type(enabled).__name__}: {enabled!r}"
568 )
570 mode = dial_config["mode"]
571 if not isinstance(mode, str) or mode.lower() not in self.TRAFFIC_DIAL_MODES:
572 raise ConfigValidationError(
573 "global_accelerator.traffic_dial.mode must be one of "
574 f"{sorted(self.TRAFFIC_DIAL_MODES)}, got {mode!r}"
575 )
577 int_ranges = (
578 ("interval_minutes", 1, 1_440),
579 ("lookback_minutes", 1, 1_440),
580 ("min_dial_percentage", 0, 100),
581 ("max_step_percentage", 1, 100),
582 ("full_health_percentage", 1, 100),
583 )
584 for field, minimum, maximum in int_ranges:
585 value = dial_config[field]
586 if (
587 not isinstance(value, int)
588 or isinstance(value, bool)
589 or not minimum <= value <= maximum
590 ):
591 raise ConfigValidationError(
592 f"global_accelerator.traffic_dial.{field} must be an integer between "
593 f"{minimum} and {maximum}, got {value!r}"
594 )
596 def _validate_backend_tls_config(self) -> None:
597 """Validate private-root and leaf-certificate lifecycle settings."""
598 config = self.get_backend_tls_config()
599 ranges = {
600 "root_generation": (1, 1_000_000),
601 "root_validity_days": (365, 36_500),
602 "root_rotate_before_days": (30, 3_650),
603 "root_activation_delay_hours": (1, 168),
604 "root_overlap_days": (2, 365),
605 "leaf_validity_days": (2, 397),
606 "leaf_rotate_before_days": (1, 90),
607 "rotation_schedule_hours": (1, 24),
608 "trust_cache_ttl_seconds": (1, 3_600),
609 "trust_cache_max_stale_seconds": (1, 86_400),
610 }
611 for field, (minimum, maximum) in ranges.items():
612 value = config.get(field)
613 if type(value) is not int or not minimum <= value <= maximum:
614 raise ConfigValidationError(
615 f"backend_tls.{field} must be an integer between "
616 f"{minimum} and {maximum}, got {value!r}"
617 )
619 if config["root_rotate_before_days"] >= config["root_validity_days"]:
620 raise ConfigValidationError(
621 "backend_tls.root_rotate_before_days must be less than root_validity_days"
622 )
623 if config["leaf_rotate_before_days"] >= config["leaf_validity_days"]:
624 raise ConfigValidationError(
625 "backend_tls.leaf_rotate_before_days must be less than leaf_validity_days"
626 )
627 if config["root_validity_days"] <= config["leaf_validity_days"]:
628 raise ConfigValidationError(
629 "backend_tls.root_validity_days must exceed leaf_validity_days"
630 )
631 if config["root_overlap_days"] <= config["leaf_validity_days"]:
632 raise ConfigValidationError(
633 "backend_tls.root_overlap_days must exceed leaf_validity_days so old leaves "
634 "remain trusted throughout root rollover"
635 )
636 if config["trust_cache_max_stale_seconds"] < config["trust_cache_ttl_seconds"]:
637 raise ConfigValidationError(
638 "backend_tls.trust_cache_max_stale_seconds must be at least trust_cache_ttl_seconds"
639 )
640 if config["root_activation_delay_hours"] * 3_600 <= config["trust_cache_max_stale_seconds"]:
641 raise ConfigValidationError(
642 "backend_tls.root_activation_delay_hours must exceed the maximum stale trust "
643 "cache window so every proxy can observe a pending root before leaf rollover"
644 )
646 def _validate_inference_proxy_config(self) -> None:
647 """Validate the inference proxy TLS CPU request, HPA target and replica bounds."""
648 config = self.get_inference_proxy_config()
649 ranges = {
650 "tls_proxy_cpu_request_millicores": (1, 250),
651 "tls_proxy_cpu_target_utilization_percentage": (1, 100),
652 "min_replicas": (1, 50),
653 "max_replicas": (1, 100),
654 }
655 for field, (minimum, maximum) in ranges.items():
656 value = config[field]
657 if type(value) is not int or not minimum <= value <= maximum:
658 raise ConfigValidationError(
659 f"inference_proxy.{field} must be an integer between "
660 f"{minimum} and {maximum}, got {value!r}"
661 )
662 if config["max_replicas"] < config["min_replicas"]:
663 raise ConfigValidationError(
664 "inference_proxy.max_replicas must be at least inference_proxy.min_replicas, "
665 f"got {config['max_replicas']} < {config['min_replicas']}"
666 )
668 def _validate_alb_config(self) -> None:
669 """Validate ALB configuration"""
670 alb_config = self.app.node.try_get_context("alb_config")
671 if not alb_config:
672 raise ConfigValidationError("alb_config configuration is required")
674 required_fields = [
675 "health_check_interval",
676 "health_check_timeout",
677 "healthy_threshold",
678 "unhealthy_threshold",
679 ]
680 for field in required_fields:
681 if field not in alb_config:
682 raise ConfigValidationError(f"Missing alb_config configuration: {field}")
684 value = alb_config[field]
685 if not isinstance(value, int) or value <= 0:
686 raise ConfigValidationError(f"{field} must be a positive integer, got {value}")
688 def _validate_manifest_processor_config(self) -> None:
689 """Validate manifest processor configuration.
691 The manifest processor section in cdk.json holds service-specific
692 settings only. The shared validation policy (allowed_namespaces,
693 resource_quotas, trusted_registries, trusted_dockerhub_orgs,
694 manifest_security_policy, allowed_kinds) lives under
695 ``job_validation_policy`` because the queue_processor reads the
696 same values.
697 """
698 mp_config = self.app.node.try_get_context("manifest_processor")
699 if not mp_config:
700 raise ConfigValidationError("manifest_processor configuration is required")
702 required_fields = [
703 "image",
704 "replicas",
705 "resource_limits",
706 ]
707 for field in required_fields:
708 if field not in mp_config:
709 raise ConfigValidationError(f"Missing manifest_processor configuration: {field}")
711 # Validate replicas
712 if not isinstance(mp_config["replicas"], int) or mp_config["replicas"] <= 0:
713 raise ConfigValidationError("manifest_processor replicas must be a positive integer")
715 try:
716 validated_request_body_limit(
717 mp_config.get("max_request_body_bytes", DEFAULT_MAX_REQUEST_BODY_BYTES)
718 )
719 except ValueError as exc:
720 raise ConfigValidationError(f"manifest_processor.{exc}") from exc
722 validation_enabled = mp_config.get("validation_enabled", True)
723 if type(validation_enabled) is not bool:
724 raise ConfigValidationError("manifest_processor.validation_enabled must be a boolean")
726 # Validate the shared policy section separately so a misconfigured
727 # policy block surfaces a clear error pointing at the right key.
728 policy = self.app.node.try_get_context("job_validation_policy")
729 if policy is None:
730 raise ConfigValidationError(
731 "job_validation_policy configuration is required (shared between "
732 "manifest_processor and queue_processor)"
733 )
734 if not isinstance(policy, dict):
735 raise ConfigValidationError("job_validation_policy must be an object")
736 for policy_field in ("allowed_namespaces", "resource_quotas"):
737 if policy_field not in policy:
738 raise ConfigValidationError(
739 f"Missing job_validation_policy configuration: {policy_field}"
740 )
742 try:
743 validate_manifest_security_policy(policy.get("manifest_security_policy", {}))
744 except ValueError as exc:
745 raise ConfigValidationError(f"job_validation_policy.{exc}") from exc
747 require_toleration = policy.get("require_accelerator_toleration", True)
748 if type(require_toleration) is not bool:
749 raise ConfigValidationError(
750 "job_validation_policy.require_accelerator_toleration must be a boolean"
751 )
753 # Validate resource limits. They render verbatim into the Deployment's
754 # container limits, so they must be non-empty Kubernetes quantities no
755 # smaller than the requests fixed in 31-manifest-processor.yaml —
756 # Kubernetes rejects limit < request, and catching that here fails the
757 # synth instead of the applier halfway through a deploy.
758 resource_limits = mp_config["resource_limits"]
759 if "cpu" not in resource_limits or "memory" not in resource_limits:
760 raise ConfigValidationError(
761 "manifest_processor resource_limits must contain 'cpu' and 'memory'"
762 )
763 for resource, request in _MANIFEST_PROCESSOR_CONTAINER_REQUESTS.items():
764 quantity = resource_limits[resource]
765 try:
766 if not isinstance(quantity, str):
767 raise ValueError("must be a string")
768 parsed = parse_k8s_quantity(quantity)
769 except ValueError as exc:
770 raise ConfigValidationError(
771 f"manifest_processor.resource_limits.{resource} must be a Kubernetes "
772 f"quantity string such as '1000m' or '2Gi', got {quantity!r} ({exc})"
773 ) from exc
774 if parsed < parse_k8s_quantity(request):
775 raise ConfigValidationError(
776 f"manifest_processor.resource_limits.{resource} must be at least the "
777 f"container request of {request!r} "
778 f"(31-manifest-processor.yaml), got {quantity!r}"
779 )
781 # Validate the optional CPU autoscaler. Off by default: the API tier is
782 # I/O-bound, so CPU is a weak saturation signal; operators who have
783 # measured otherwise turn it on here and the HPA takes over the count.
784 autoscaling = mp_config.get("autoscaling")
785 if autoscaling is None:
786 autoscaling = {}
787 if not isinstance(autoscaling, dict):
788 raise ConfigValidationError("manifest_processor.autoscaling must be an object")
789 unknown_autoscaling = sorted(
790 str(key) for key in autoscaling if key not in _MANIFEST_PROCESSOR_AUTOSCALING_DEFAULTS
791 )
792 if unknown_autoscaling:
793 raise ConfigValidationError(
794 "manifest_processor.autoscaling contains unknown key(s): "
795 + ", ".join(unknown_autoscaling)
796 + "; allowed keys: "
797 + ", ".join(sorted(_MANIFEST_PROCESSOR_AUTOSCALING_DEFAULTS))
798 )
799 merged_autoscaling = {**_MANIFEST_PROCESSOR_AUTOSCALING_DEFAULTS, **autoscaling}
800 if type(merged_autoscaling["enabled"]) is not bool:
801 raise ConfigValidationError("manifest_processor.autoscaling.enabled must be a boolean")
802 for key, minimum, maximum in (
803 ("max_replicas", 1, 100),
804 ("cpu_target_utilization_percentage", 1, 100),
805 ):
806 value = merged_autoscaling[key]
807 if type(value) is not int or not minimum <= value <= maximum:
808 raise ConfigValidationError(
809 f"manifest_processor.autoscaling.{key} must be an integer between "
810 f"{minimum} and {maximum}, got {value!r}"
811 )
812 if merged_autoscaling["max_replicas"] < mp_config["replicas"]:
813 raise ConfigValidationError(
814 "manifest_processor.autoscaling.max_replicas must be at least "
815 f"manifest_processor.replicas, got {merged_autoscaling['max_replicas']} "
816 f"< {mp_config['replicas']}"
817 )
819 # Validate allowed namespaces (lives under job_validation_policy).
820 if not isinstance(policy["allowed_namespaces"], list):
821 raise ConfigValidationError("job_validation_policy.allowed_namespaces must be a list")
823 def _validate_api_gateway_config(self) -> None:
824 """Validate API Gateway configuration"""
825 api_gw_config = self.app.node.try_get_context("api_gateway")
826 if not api_gw_config:
827 raise ConfigValidationError("api_gateway configuration is required")
829 required_fields = [
830 "throttle_rate_limit",
831 "throttle_burst_limit",
832 "log_level",
833 "metrics_enabled",
834 "tracing_enabled",
835 ]
836 for field in required_fields:
837 if field not in api_gw_config:
838 raise ConfigValidationError(f"Missing api_gateway configuration: {field}")
840 # Validate throttle limits
841 throttle_rate = api_gw_config["throttle_rate_limit"]
842 throttle_burst = api_gw_config["throttle_burst_limit"]
844 if not isinstance(throttle_rate, int) or throttle_rate <= 0:
845 raise ConfigValidationError(
846 f"throttle_rate_limit must be a positive integer, got {throttle_rate}"
847 )
849 if not isinstance(throttle_burst, int) or throttle_burst <= 0:
850 raise ConfigValidationError(
851 f"throttle_burst_limit must be a positive integer, got {throttle_burst}"
852 )
854 if throttle_burst < throttle_rate:
855 raise ConfigValidationError(
856 "throttle_burst_limit should be greater than or equal to throttle_rate_limit"
857 )
859 # Validate log level
860 valid_log_levels = ["OFF", "ERROR", "INFO"]
861 log_level = api_gw_config["log_level"]
862 if log_level not in valid_log_levels:
863 raise ConfigValidationError(
864 f"log_level must be one of {valid_log_levels}, got {log_level}"
865 )
867 # Validate boolean flags
868 if not isinstance(api_gw_config["metrics_enabled"], bool):
869 raise ConfigValidationError("metrics_enabled must be a boolean")
871 if not isinstance(api_gw_config["tracing_enabled"], bool):
872 raise ConfigValidationError("tracing_enabled must be a boolean")
874 if "regional_api_enabled" in api_gw_config and not isinstance(
875 api_gw_config["regional_api_enabled"], bool
876 ):
877 raise ConfigValidationError("regional_api_enabled must be a boolean")
879 def _validate_eks_cluster_config(self) -> None:
880 """Validate EKS cluster configuration"""
881 eks_config = self.app.node.try_get_context("eks_cluster") or {}
883 # Validate endpoint_access if present
884 if "endpoint_access" in eks_config:
885 valid_access_modes = ["PRIVATE", "PUBLIC_AND_PRIVATE"]
886 if eks_config["endpoint_access"] not in valid_access_modes:
887 raise ConfigValidationError(
888 f"endpoint_access must be one of {valid_access_modes}, "
889 f"got {eks_config['endpoint_access']}"
890 )
891 # The NetworkPolicy enforcement switch renders straight into a
892 # kube-system ConfigMap value, so only a literal JSON boolean is
893 # accepted — a string "false" would be applied verbatim and enable
894 # nothing while reading as disabled.
895 if (
896 "network_policy_enforcement" in eks_config
897 and type(eks_config["network_policy_enforcement"]) is not bool
898 ):
899 raise ConfigValidationError(
900 "eks_cluster.network_policy_enforcement must be a boolean, got "
901 f"{eks_config['network_policy_enforcement']!r}"
902 )
903 self._validate_vpc_endpoints_config()
905 def _validate_vpc_endpoints_config(self) -> None:
906 """Validate the optional ``vpc_endpoints`` block.
908 ``gateway`` lists the free route-table endpoints (``s3``,
909 ``dynamodb``); ``interface`` lists PrivateLink endpoints by the
910 service key in :data:`VPC_INTERFACE_ENDPOINT_SERVICES`. Interface
911 endpoints bill per AZ-hour, so the list is explicit and unknown names
912 fail at synth instead of silently creating nothing.
913 """
914 raw = self.app.node.try_get_context("vpc_endpoints")
915 if raw is None:
916 return
917 if not isinstance(raw, dict):
918 raise ConfigValidationError("vpc_endpoints must be an object")
919 unknown = sorted(str(key) for key in raw if key not in _VPC_ENDPOINTS_DEFAULTS)
920 if unknown:
921 raise ConfigValidationError(
922 "vpc_endpoints contains unknown key(s): "
923 + ", ".join(unknown)
924 + "; allowed keys: "
925 + ", ".join(sorted(_VPC_ENDPOINTS_DEFAULTS))
926 )
927 for key, allowed in (
928 ("gateway", VPC_GATEWAY_ENDPOINT_SERVICES),
929 ("interface", VPC_INTERFACE_ENDPOINT_SERVICES),
930 ):
931 if key not in raw:
932 continue
933 value = raw[key]
934 if not isinstance(value, list) or not all(isinstance(item, str) for item in value):
935 raise ConfigValidationError(f"vpc_endpoints.{key} must be a list of strings")
936 unsupported = sorted(set(value) - set(allowed))
937 if unsupported:
938 raise ConfigValidationError(
939 f"vpc_endpoints.{key} contains unsupported service(s): "
940 + ", ".join(unsupported)
941 + "; supported: "
942 + ", ".join(sorted(allowed))
943 )
944 if len(set(value)) != len(value):
945 raise ConfigValidationError(f"vpc_endpoints.{key} lists a service twice")
947 def _validate_eks_capabilities_config(self) -> None:
948 """Validate the optional ``eks_capabilities`` block.
950 Every capability type is off by default, so an absent block is valid.
951 When present, the shape rules live in
952 :mod:`gco.eks_capabilities_config`; per-type ``regions`` subsets must
953 name regional deployment regions and every ACK role or policy entry
954 must be an ARN of the right shape. The run-scoped
955 ``eks_capabilities_overrides`` context (a JSON object) is merged in
956 first, so an override that would produce an invalid block fails the
957 same way an invalid cdk.json does.
958 """
959 raw = self._raw_eks_capabilities_config()
960 if raw is None:
961 return
962 try:
963 validate_eks_capabilities_config(raw, self.get_regions())
964 except EksCapabilitiesConfigError as exc:
965 raise ConfigValidationError(str(exc)) from exc
967 def _raw_eks_capabilities_config(self) -> object:
968 """The cdk.json ``eks_capabilities`` block with run-scoped overrides merged in."""
969 raw = self.app.node.try_get_context(EKS_CAPABILITIES_CONTEXT_KEY)
970 try:
971 overrides = parse_eks_capabilities_overrides(
972 self.app.node.try_get_context(EKS_CAPABILITIES_OVERRIDES_CONTEXT_KEY)
973 )
974 except EksCapabilitiesConfigError as exc:
975 raise ConfigValidationError(str(exc)) from exc
976 return merge_eks_capabilities_overrides(raw, overrides)
978 def _raw_argocd_config(self) -> object:
979 """The cdk.json ``helm.argocd`` block, or ``None`` when absent."""
980 helm = self.app.node.try_get_context("helm")
981 return helm.get("argocd") if isinstance(helm, dict) else None
983 def _validate_argocd_config(self) -> None:
984 """Validate the optional self-managed Argo CD block (``helm.argocd``).
986 Argo CD is off by default, so an absent block is valid. When present,
987 the shape rules live in :mod:`gco.argocd_config`: unknown keys fail by
988 name, the GitOps repository must be a Git URL admitted by
989 ``source_repos``, and the sync policy is ``manual`` or ``automated``.
990 """
991 try:
992 validate_argocd_config(self._raw_argocd_config())
993 except ArgoCdConfigError as exc:
994 raise ConfigValidationError(str(exc)) from exc
996 def get_argocd_config(self) -> dict[str, Any]:
997 """Return the ``helm.argocd`` block with defaults merged in.
999 Keys: ``enabled``, ``source_repos``, ``gitops`` ``{repo_url, revision,
1000 path, sync_policy}`` (see :data:`gco.argocd_config.ARGOCD_DEFAULTS`).
1001 Whether the chart actually installs also honors the run-scoped
1002 ``helm_enabled_overrides`` context, which the regional stack resolves.
1003 """
1004 try:
1005 return validate_argocd_config(self._raw_argocd_config())
1006 except ArgoCdConfigError as exc:
1007 raise ConfigValidationError(str(exc)) from exc
1009 def _raw_slurm_config(self) -> object:
1010 """The cdk.json ``helm.slurm`` block, or ``None`` when absent."""
1011 helm = self.app.node.try_get_context("helm")
1012 return helm.get("slurm") if isinstance(helm, dict) else None
1014 def _slurm_fsx_available_everywhere(self) -> bool:
1015 """True when every regional cluster gets an FSx for Lustre filesystem.
1017 ``helm.slurm.storage.scratch: fsx`` asks for the FSx claim on every
1018 Slurm worker and login pod, and FSx is a per-region toggle, so the
1019 strict setting is only valid when each region has it. (``auto``
1020 resolves per region in the regional stack and never needs this.)
1022 Consulted only when the block actually says ``scratch: fsx``: the FSx
1023 lookup validates the run-scoped ``feature_enabled_overrides`` as a
1024 side effect, and a Slurm block that never asked for FSx must not turn
1025 an unrelated override typo into a construction-time Slurm error.
1026 """
1027 raw = self._raw_slurm_config()
1028 storage = raw.get("storage") if isinstance(raw, dict) else None
1029 if not isinstance(storage, dict) or storage.get("scratch") != "fsx":
1030 return False
1031 regions = self.get_regions()
1032 return bool(regions) and all(
1033 bool(self.get_fsx_lustre_config(region)["enabled"]) for region in regions
1034 )
1036 def _validate_slurm_config(self) -> None:
1037 """Validate the optional Slurm cluster block (``helm.slurm``).
1039 Slurm is off by default, so an absent block is valid. When present,
1040 the shape rules live in :mod:`gco.slurm_config`: unknown keys fail by
1041 name, every partition must target a known node pool with a coherent
1042 accelerator request and fit the ``gco-jobs`` resource quota, exactly
1043 one enabled partition is the default, autoscaled partitions need
1044 ``cluster_observability`` (their metrics come through Prometheus), and
1045 ``storage.scratch: fsx`` needs FSx in every region.
1046 """
1047 try:
1048 validate_slurm_config(
1049 self._raw_slurm_config(),
1050 resource_quota=self.app.node.try_get_context("resource_quota") or {},
1051 observability_enabled=self.get_cluster_observability_enabled(),
1052 fsx_enabled=self._slurm_fsx_available_everywhere(),
1053 )
1054 except SlurmConfigError as exc:
1055 raise ConfigValidationError(str(exc)) from exc
1057 def get_slurm_config(self) -> dict[str, Any]:
1058 """Return the ``helm.slurm`` block with defaults merged in.
1060 Keys: ``enabled``, ``cluster_name``, ``controller``, ``restapi``,
1061 ``accounting``, ``metrics``, ``login``, ``autoscaling``,
1062 ``partitions``, ``storage``, ``slurm_conf``, ``slurmdbd_conf`` (see
1063 :data:`gco.slurm_config.SLURM_DEFAULTS`). Whether the charts actually
1064 install also honors the run-scoped ``helm_enabled_overrides`` context,
1065 which the regional stack resolves.
1066 """
1067 try:
1068 return validate_slurm_config(
1069 self._raw_slurm_config(),
1070 resource_quota=self.app.node.try_get_context("resource_quota") or {},
1071 observability_enabled=self.get_cluster_observability_enabled(),
1072 fsx_enabled=self._slurm_fsx_available_everywhere(),
1073 )
1074 except SlurmConfigError as exc:
1075 raise ConfigValidationError(str(exc)) from exc
1077 def get_eks_capabilities_config(self) -> dict[str, Any]:
1078 """Return the ``eks_capabilities`` block with defaults merged in.
1080 Keys: ``ack`` / ``kro``, each ``{enabled, regions, ...}``
1081 (see :data:`gco.eks_capabilities_config.EKS_CAPABILITIES_DEFAULTS`).
1082 Both are disabled by default; ``regions: []`` means every
1083 regional deployment region. Honors the run-scoped
1084 ``eks_capabilities_overrides`` context like the validator does.
1085 """
1086 try:
1087 return validate_eks_capabilities_config(self._raw_eks_capabilities_config(), None)
1088 except EksCapabilitiesConfigError as exc:
1089 raise ConfigValidationError(str(exc)) from exc
1091 def get_vpc_endpoints_config(self) -> dict[str, list[str]]:
1092 """Return the VPC endpoint selection with defaults merged in.
1094 Defaults: the two free gateway endpoints (S3 carries the platform's
1095 largest data path — models, datasets, checkpoints, MLflow artifacts,
1096 cost reports — off the NAT gateways' per-GB metering; DynamoDB serves
1097 single-region topologies) and no interface endpoints.
1098 """
1099 configured = self.app.node.try_get_context("vpc_endpoints") or {}
1100 return {
1101 "gateway": list(configured.get("gateway", _VPC_ENDPOINTS_DEFAULTS["gateway"])),
1102 "interface": list(configured.get("interface", _VPC_ENDPOINTS_DEFAULTS["interface"])),
1103 }
1105 def _validate_analytics_environment_config(self) -> None:
1106 """Validate the optional analytics_environment block in cdk.json.
1108 The block is entirely optional; absence means the feature is disabled
1109 and no validation is needed. When present, we validate:
1111 - ``enabled``: must be a bool if present (defaults to False via merge).
1112 - ``hyperpod.enabled``: must be a bool if present (defaults to False).
1113 - ``cognito.removal_policy`` and ``efs.removal_policy``: must be the
1114 literal strings ``"destroy"`` or ``"retain"`` (case sensitive — they
1115 are passed verbatim to CDK's ``RemovalPolicy`` lookup by the
1116 consumer).
1117 """
1118 analytics_ctx = self.app.node.try_get_context("analytics_environment")
1119 if not isinstance(analytics_ctx, dict):
1120 # Block is absent or malformed — defaults apply, nothing to validate.
1121 return
1123 # Top-level `enabled` must be a bool if provided.
1124 if "enabled" in analytics_ctx and not isinstance(analytics_ctx["enabled"], bool):
1125 raise ConfigValidationError(
1126 f"analytics_environment.enabled must be a bool, got "
1127 f"{type(analytics_ctx['enabled']).__name__}: {analytics_ctx['enabled']!r}"
1128 )
1130 # `hyperpod.enabled` must be a bool if the sub-block is a dict and
1131 # carries the key.
1132 hyperpod_ctx = analytics_ctx.get("hyperpod")
1133 if (
1134 isinstance(hyperpod_ctx, dict)
1135 and "enabled" in hyperpod_ctx
1136 and not isinstance(hyperpod_ctx["enabled"], bool)
1137 ):
1138 raise ConfigValidationError(
1139 f"analytics_environment.hyperpod.enabled must be a bool, got "
1140 f"{type(hyperpod_ctx['enabled']).__name__}: {hyperpod_ctx['enabled']!r}"
1141 )
1143 # `canvas.enabled` must be a bool if the sub-block is a dict and
1144 # carries the key. Mirrors the hyperpod validation above.
1145 canvas_ctx = analytics_ctx.get("canvas")
1146 if (
1147 isinstance(canvas_ctx, dict)
1148 and "enabled" in canvas_ctx
1149 and not isinstance(canvas_ctx["enabled"], bool)
1150 ):
1151 raise ConfigValidationError(
1152 f"analytics_environment.canvas.enabled must be a bool, got "
1153 f"{type(canvas_ctx['enabled']).__name__}: {canvas_ctx['enabled']!r}"
1154 )
1156 valid_removal_policies = {"destroy", "retain"}
1158 for sub_block in ("cognito", "efs"):
1159 sub_ctx = analytics_ctx.get(sub_block)
1160 if not isinstance(sub_ctx, dict):
1161 continue
1162 if "removal_policy" not in sub_ctx:
1163 continue
1164 removal_policy = sub_ctx["removal_policy"]
1165 if removal_policy not in valid_removal_policies:
1166 raise ConfigValidationError(
1167 f"analytics_environment.{sub_block}.removal_policy must be one of "
1168 f"{sorted(valid_removal_policies)}, got {removal_policy!r}"
1169 )
1171 def _validate_cluster_observability_config(self) -> None:
1172 """Validate the optional cluster_observability block in cdk.json.
1174 The block is entirely optional; absence means the on-by-default
1175 defaults apply and nothing needs validating. When present, we check:
1177 - ``enabled``: must be a bool if present (defaults to True via merge —
1178 in-cluster observability is on unless explicitly disabled).
1179 - ``grafana``/``prometheus``/``alertmanager`` sub-block ``persistence_size``
1180 and ``prometheus.retention``: must be non-empty strings if present
1181 (they are passed verbatim to Helm chart values as Kubernetes
1182 quantity / duration strings).
1183 - ``alertmanager.enabled``: must be a bool if present.
1184 """
1185 obs_ctx = self.app.node.try_get_context("cluster_observability")
1186 if not isinstance(obs_ctx, dict):
1187 # Block is absent or malformed — defaults apply, nothing to validate.
1188 return
1190 if "enabled" in obs_ctx and not isinstance(obs_ctx["enabled"], bool):
1191 raise ConfigValidationError(
1192 f"cluster_observability.enabled must be a bool, got "
1193 f"{type(obs_ctx['enabled']).__name__}: {obs_ctx['enabled']!r}"
1194 )
1196 # Non-empty-string checks for the size / retention knobs.
1197 string_fields = (
1198 ("grafana", "persistence_size"),
1199 ("prometheus", "persistence_size"),
1200 ("prometheus", "retention"),
1201 ("alertmanager", "persistence_size"),
1202 )
1203 for sub_block, field in string_fields:
1204 sub_ctx = obs_ctx.get(sub_block)
1205 if not isinstance(sub_ctx, dict) or field not in sub_ctx:
1206 continue
1207 value = sub_ctx[field]
1208 if not isinstance(value, str) or not value.strip():
1209 raise ConfigValidationError(
1210 f"cluster_observability.{sub_block}.{field} must be a non-empty "
1211 f"string, got {value!r}"
1212 )
1214 # `grafana.admin_password_rotation_schedule`: the cron for the in-cluster
1215 # CronJob that rotates the Grafana admin password. Validate the 5-field
1216 # cron shape so a typo fails at synth rather than yielding an
1217 # un-schedulable CronJob in every region.
1218 grafana_ctx = obs_ctx.get("grafana")
1219 if isinstance(grafana_ctx, dict) and "admin_password_rotation_schedule" in grafana_ctx:
1220 schedule = grafana_ctx["admin_password_rotation_schedule"]
1221 if not isinstance(schedule, str) or len(schedule.split()) != 5:
1222 raise ConfigValidationError(
1223 "cluster_observability.grafana.admin_password_rotation_schedule must be "
1224 f"a 5-field cron expression string, got {schedule!r}"
1225 )
1227 # `alertmanager.enabled` must be a bool if the sub-block carries it.
1228 alertmanager_ctx = obs_ctx.get("alertmanager")
1229 if (
1230 isinstance(alertmanager_ctx, dict)
1231 and "enabled" in alertmanager_ctx
1232 and not isinstance(alertmanager_ctx["enabled"], bool)
1233 ):
1234 raise ConfigValidationError(
1235 f"cluster_observability.alertmanager.enabled must be a bool, got "
1236 f"{type(alertmanager_ctx['enabled']).__name__}: "
1237 f"{alertmanager_ctx['enabled']!r}"
1238 )
1240 def _validate_cost_monitoring_config(self) -> None:
1241 """Validate the optional ``cost_monitoring`` block in cdk.json.
1243 The block is entirely optional; absence means the on-by-default
1244 defaults apply and nothing needs validating. When present, we check:
1246 - ``enabled``: must be a bool if present (defaults to True via merge).
1247 - ``reports.interval_minutes``: positive int between 5 and 1440 if
1248 present (the cost-monitor service's scheduled report cadence).
1249 - ``reports.retention_days`` /
1250 ``reports.transition_to_infrequent_access_days`` /
1251 ``athena.query_results_retention_days``: positive ints if present.
1252 - The IA transition must happen strictly before expiration, otherwise
1253 the S3 lifecycle configuration is rejected at deploy time — fail at
1254 synth instead.
1256 There is deliberately no cross-toggle error against
1257 ``cluster_observability``: cost monitoring's *effective* enablement is
1258 the conjunction of both toggles (see
1259 :meth:`get_cost_monitoring_enabled`), so disabling observability
1260 simply switches the cost pipeline off with it — ``gco monitoring
1261 disable`` must not break synthesis.
1262 """
1263 cost_ctx = self.app.node.try_get_context("cost_monitoring")
1264 if not isinstance(cost_ctx, dict):
1265 # Block absent or malformed — defaults apply, nothing to validate.
1266 return
1268 if "enabled" in cost_ctx and not isinstance(cost_ctx["enabled"], bool):
1269 raise ConfigValidationError(
1270 f"cost_monitoring.enabled must be a bool, got "
1271 f"{type(cost_ctx['enabled']).__name__}: {cost_ctx['enabled']!r}"
1272 )
1274 int_fields = (
1275 ("reports", "interval_minutes", 5, 1_440),
1276 ("reports", "retention_days", 1, 3_650),
1277 ("reports", "transition_to_infrequent_access_days", 30, 3_650),
1278 ("athena", "query_results_retention_days", 1, 3_650),
1279 )
1280 for sub_block, field, minimum, maximum in int_fields:
1281 sub_ctx = cost_ctx.get(sub_block)
1282 if not isinstance(sub_ctx, dict) or field not in sub_ctx:
1283 continue
1284 value = sub_ctx[field]
1285 if (
1286 not isinstance(value, int)
1287 or isinstance(value, bool)
1288 or not minimum <= value <= maximum
1289 ):
1290 raise ConfigValidationError(
1291 f"cost_monitoring.{sub_block}.{field} must be an integer between "
1292 f"{minimum} and {maximum}, got {value!r}"
1293 )
1295 merged = self.get_cost_monitoring_config()
1296 reports = merged["reports"]
1297 if reports["transition_to_infrequent_access_days"] >= reports["retention_days"]:
1298 raise ConfigValidationError(
1299 "cost_monitoring.reports.transition_to_infrequent_access_days "
1300 f"({reports['transition_to_infrequent_access_days']}) must be smaller than "
1301 f"cost_monitoring.reports.retention_days ({reports['retention_days']}); "
1302 "S3 rejects lifecycle rules that transition on or after expiration."
1303 )
1305 def _validate_capacity_history_config(self) -> None:
1306 """Validate the optional ``historical`` block in cdk.json.
1308 The block is entirely optional; absence means the historical capacity
1309 surface is disabled and no validation is needed. When present, types
1310 are validated so a typo fails fast at synth time:
1312 - ``enabled``: bool if present.
1313 - ``retention_days`` / ``poll_interval_minutes``: positive ints if present.
1314 - ``watch_instance_types`` / ``enabled_regions``: lists of strings if present.
1315 - every region in ``enabled_regions`` must be a known AWS region.
1316 - ``spot_score_target_capacities``: non-empty list of positive
1317 integers (booleans rejected), every value a member of the supported
1318 set exported by ``cli/capacity/history.py`` — metric fields are
1319 statically named, so an unsupported capacity has nowhere to land.
1320 """
1321 historical_ctx = self.app.node.try_get_context("historical")
1322 if not isinstance(historical_ctx, dict):
1323 return
1325 if "enabled" in historical_ctx and not isinstance(historical_ctx["enabled"], bool):
1326 raise ConfigValidationError(
1327 f"historical.enabled must be a bool, got "
1328 f"{type(historical_ctx['enabled']).__name__}: {historical_ctx['enabled']!r}"
1329 )
1331 for int_field in (
1332 "retention_days",
1333 "poll_interval_minutes",
1334 "capacity_block_duration_hours",
1335 ):
1336 if int_field not in historical_ctx:
1337 continue
1338 value = historical_ctx[int_field]
1339 if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
1340 raise ConfigValidationError(
1341 f"historical.{int_field} must be a positive integer, got {value!r}"
1342 )
1344 # The long-block probe duration may be 0 to disable the long probe
1345 # entirely, so it is validated as non-negative rather than positive.
1346 if "capacity_block_long_duration_hours" in historical_ctx:
1347 long_value = historical_ctx["capacity_block_long_duration_hours"]
1348 if not isinstance(long_value, int) or isinstance(long_value, bool) or long_value < 0:
1349 raise ConfigValidationError(
1350 "historical.capacity_block_long_duration_hours must be a non-negative "
1351 f"integer (0 disables the long probe), got {long_value!r}"
1352 )
1354 for list_field in ("watch_instance_types", "enabled_regions"):
1355 if list_field not in historical_ctx:
1356 continue
1357 value = historical_ctx[list_field]
1358 if not isinstance(value, list) or not all(isinstance(v, str) for v in value):
1359 raise ConfigValidationError(
1360 f"historical.{list_field} must be a list of strings, got {value!r}"
1361 )
1363 for region in historical_ctx.get("enabled_regions", []) or []:
1364 if region not in self.VALID_REGIONS:
1365 raise ConfigValidationError(
1366 f"historical.enabled_regions contains invalid region '{region}'. "
1367 f"Valid regions: {sorted(self.VALID_REGIONS)}"
1368 )
1370 if "spot_score_target_capacities" in historical_ctx:
1371 # Function-local import: cli/capacity/history.py owns the supported
1372 # set and the capacity->field naming rule, and imports nothing from
1373 # gco, so validation and storage cannot drift apart.
1374 from cli.capacity.history import SUPPORTED_SPOT_SCORE_TARGET_CAPACITIES
1376 capacities = historical_ctx["spot_score_target_capacities"]
1377 if (
1378 not isinstance(capacities, list)
1379 or not capacities
1380 or not all(
1381 isinstance(value, int) and not isinstance(value, bool) and value > 0
1382 for value in capacities
1383 )
1384 ):
1385 raise ConfigValidationError(
1386 "historical.spot_score_target_capacities must be a non-empty list "
1387 f"of positive integers, got {capacities!r}"
1388 )
1389 for value in capacities:
1390 if value not in SUPPORTED_SPOT_SCORE_TARGET_CAPACITIES:
1391 raise ConfigValidationError(
1392 "historical.spot_score_target_capacities contains unsupported "
1393 f"target capacity {value!r}. Supported target capacities: "
1394 f"{list(SUPPORTED_SPOT_SCORE_TARGET_CAPACITIES)} (each needs a "
1395 "statically declared metric field; see cli/capacity/history.py)"
1396 )
1398 #: Distance functions the DynamoDB vector-index API accepts. The choice is
1399 #: immutable after index creation, so a typo must fail at synth time.
1400 MISSION_MEMORY_DISTANCE_FUNCTIONS = frozenset({"COSINE", "DOT_PRODUCT", "EUCLIDEAN"})
1402 #: DynamoDB vector-index maximum dimensionality (service quota).
1403 MISSION_MEMORY_MAX_DIMENSIONS = 4096
1405 def _validate_mission_memory_config(self) -> None:
1406 """Validate the ``mission_memory`` block in cdk.json.
1408 The block is optional; absence means the shipped defaults apply
1409 (feature on). When present, types are validated so a typo fails fast
1410 at synth time — especially the one-way-door fields (``dimensions``,
1411 ``distance_function``) that cannot be corrected after the vector
1412 index exists:
1414 - ``enabled``: bool if present.
1415 - ``retention_days`` / ``top_k``: positive ints if present.
1416 - ``dimensions``: positive int <= 4096 if present.
1417 - ``distance_function``: one of COSINE / DOT_PRODUCT / EUCLIDEAN.
1418 """
1419 mission_memory_ctx = self.app.node.try_get_context("mission_memory")
1420 if not isinstance(mission_memory_ctx, dict):
1421 return
1423 if "enabled" in mission_memory_ctx and not isinstance(mission_memory_ctx["enabled"], bool):
1424 raise ConfigValidationError(
1425 f"mission_memory.enabled must be a bool, got "
1426 f"{type(mission_memory_ctx['enabled']).__name__}: "
1427 f"{mission_memory_ctx['enabled']!r}"
1428 )
1430 for int_field in ("retention_days", "top_k"):
1431 if int_field not in mission_memory_ctx:
1432 continue
1433 value = mission_memory_ctx[int_field]
1434 if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
1435 raise ConfigValidationError(
1436 f"mission_memory.{int_field} must be a positive integer, got {value!r}"
1437 )
1439 if "dimensions" in mission_memory_ctx:
1440 dimensions = mission_memory_ctx["dimensions"]
1441 if (
1442 not isinstance(dimensions, int)
1443 or isinstance(dimensions, bool)
1444 or dimensions <= 0
1445 or dimensions > self.MISSION_MEMORY_MAX_DIMENSIONS
1446 ):
1447 raise ConfigValidationError(
1448 "mission_memory.dimensions must be a positive integer <= "
1449 f"{self.MISSION_MEMORY_MAX_DIMENSIONS} (DynamoDB vector-index "
1450 f"limit), got {dimensions!r}. This is a one-way door: it is "
1451 "immutable after index creation and must match the "
1452 "bedrock.embedding_model_id output width."
1453 )
1455 if "distance_function" in mission_memory_ctx:
1456 distance = mission_memory_ctx["distance_function"]
1457 if (
1458 not isinstance(distance, str)
1459 or distance not in self.MISSION_MEMORY_DISTANCE_FUNCTIONS
1460 ):
1461 valid = ", ".join(sorted(self.MISSION_MEMORY_DISTANCE_FUNCTIONS))
1462 raise ConfigValidationError(
1463 f"mission_memory.distance_function must be one of {valid}, got "
1464 f"{distance!r}. This is immutable after index creation."
1465 )
1467 def _validate_vector_store_config(self) -> None:
1468 """Validate the optional ``vector_store`` block in cdk.json.
1470 The block is optional; absence means the feature stays off. When
1471 present, types are validated so a typo fails fast at synth time —
1472 especially the one-way-door fields (``dimensions``,
1473 ``distance_function``) that cannot be corrected after the vector
1474 index exists. The distance-function set and dimension ceiling reuse
1475 the mission-memory constants because both features target the same
1476 DynamoDB vector-index API limits:
1478 - ``enabled``: bool if present.
1479 - ``dimensions``: positive int <= 4096 if present.
1480 - ``distance_function``: one of COSINE / DOT_PRODUCT / EUCLIDEAN.
1481 - ``embedding_model_id``: non-empty string. Deliberately independent
1482 of ``bedrock.embedding_model_id`` (mission memory) — the two
1483 corpora may use different models.
1484 - ``replica_regions``: list of known regions, no duplicates, and
1485 never the global region (that is the table's primary).
1486 - ``corpus_prefix``: non-empty S3 key prefix ending in ``/``.
1487 """
1488 vector_store_ctx = self.app.node.try_get_context("vector_store")
1489 if not isinstance(vector_store_ctx, dict):
1490 return
1492 if "enabled" in vector_store_ctx and not isinstance(vector_store_ctx["enabled"], bool):
1493 raise ConfigValidationError(
1494 f"vector_store.enabled must be a bool, got "
1495 f"{type(vector_store_ctx['enabled']).__name__}: "
1496 f"{vector_store_ctx['enabled']!r}"
1497 )
1498 if "dimensions" in vector_store_ctx:
1499 dimensions = vector_store_ctx["dimensions"]
1500 if (
1501 not isinstance(dimensions, int)
1502 or isinstance(dimensions, bool)
1503 or dimensions <= 0
1504 or dimensions > self.MISSION_MEMORY_MAX_DIMENSIONS
1505 ):
1506 raise ConfigValidationError(
1507 "vector_store.dimensions must be a positive integer <= "
1508 f"{self.MISSION_MEMORY_MAX_DIMENSIONS} (DynamoDB vector-index "
1509 f"limit), got {dimensions!r}. This is a one-way door: it is "
1510 "immutable after index creation and must match the "
1511 "vector_store.embedding_model_id output width."
1512 )
1513 if "distance_function" in vector_store_ctx:
1514 distance = vector_store_ctx["distance_function"]
1515 if (
1516 not isinstance(distance, str)
1517 or distance not in self.MISSION_MEMORY_DISTANCE_FUNCTIONS
1518 ):
1519 valid = ", ".join(sorted(self.MISSION_MEMORY_DISTANCE_FUNCTIONS))
1520 raise ConfigValidationError(
1521 f"vector_store.distance_function must be one of {valid}, got "
1522 f"{distance!r}. This is immutable after index creation."
1523 )
1524 if "embedding_model_id" in vector_store_ctx:
1525 model_id = vector_store_ctx["embedding_model_id"]
1526 if not isinstance(model_id, str) or not model_id.strip():
1527 raise ConfigValidationError(
1528 f"vector_store.embedding_model_id must be a non-empty string, got {model_id!r}"
1529 )
1530 if "replica_regions" in vector_store_ctx:
1531 replica_regions = vector_store_ctx["replica_regions"]
1532 if not isinstance(replica_regions, list) or not all(
1533 isinstance(region, str) for region in replica_regions
1534 ):
1535 raise ConfigValidationError(
1536 f"vector_store.replica_regions must be a list of region strings, "
1537 f"got {replica_regions!r}"
1538 )
1539 if len(replica_regions) != len(set(replica_regions)):
1540 raise ConfigValidationError(
1541 f"vector_store.replica_regions contains duplicates: {replica_regions!r}"
1542 )
1543 global_region = self.get_global_region()
1544 for region in replica_regions:
1545 if region not in self.VALID_REGIONS:
1546 raise ConfigValidationError(
1547 f"vector_store.replica_regions contains invalid region "
1548 f"'{region}'. Valid regions: {sorted(self.VALID_REGIONS)}"
1549 )
1550 if region == global_region:
1551 raise ConfigValidationError(
1552 f"vector_store.replica_regions must not include the global "
1553 f"region '{global_region}': the primary table already lives "
1554 "there and a global table cannot replicate into its own region."
1555 )
1556 if "corpus_prefix" in vector_store_ctx:
1557 corpus_prefix = vector_store_ctx["corpus_prefix"]
1558 if (
1559 not isinstance(corpus_prefix, str)
1560 or not corpus_prefix.strip()
1561 or not corpus_prefix.endswith("/")
1562 or corpus_prefix.startswith("/")
1563 ):
1564 raise ConfigValidationError(
1565 "vector_store.corpus_prefix must be a non-empty S3 key prefix "
1566 f"ending in '/' (and not starting with '/'), got {corpus_prefix!r}"
1567 )
1569 def get_project_name(self) -> str:
1570 """Get project name from configuration"""
1571 return self.app.node.try_get_context("project_name") or "gco"
1573 def get_deployment_regions(self) -> dict[str, Any]:
1574 """Get deployment regions configuration.
1576 Returns a dict with:
1577 - global: Region for Global Accelerator and SSM parameters (default: us-east-2)
1578 - api_gateway: Region for API Gateway stack (default: us-east-2)
1579 - monitoring: Region for Monitoring stack (default: us-east-2)
1580 - regional: List of regions for EKS clusters (default: ["us-east-1"]).
1581 An explicitly empty list is valid and means a control-plane-only
1582 deployment: only the global, API Gateway, and monitoring stacks are
1583 synthesized until a workload Region is added back.
1585 Note: Global Accelerator is a global service but requires a "home" region
1586 for CloudFormation deployment. us-east-2 is used by default to keep
1587 global infrastructure separate from workload regions.
1588 """
1589 deployment_regions = self.app.node.try_get_context("deployment_regions") or {}
1591 regional = deployment_regions.get("regional", ["us-east-1"])
1592 if self.is_control_plane_only():
1593 regional = []
1594 return {
1595 "global": deployment_regions.get("global", "us-east-2"),
1596 "api_gateway": deployment_regions.get("api_gateway", "us-east-2"),
1597 "monitoring": deployment_regions.get("monitoring", "us-east-2"),
1598 "regional": regional,
1599 }
1601 def is_control_plane_only(self) -> bool:
1602 """Return whether this run synthesizes with zero workload Regions.
1604 True when ``--context gco:control-plane-only=true`` was passed (see
1605 :data:`gco.stacks.constants.CONTROL_PLANE_ONLY_CONTEXT_KEY`). The
1606 configured ``deployment_regions.regional`` list in ``cdk.json`` is
1607 left untouched; only this synthesis ignores it.
1608 """
1609 return control_plane_only_requested(
1610 self.app.node.try_get_context(CONTROL_PLANE_ONLY_CONTEXT_KEY)
1611 )
1613 def get_deployment_partition(self) -> str:
1614 """Return the one SDK partition shared by every configured Region."""
1615 deployment_regions = self.get_deployment_regions()
1616 regional = validated_regional_deployment_regions(
1617 deployment_regions["regional"],
1618 known_regions=self.VALID_REGIONS,
1619 )
1620 return validated_deployment_partition(
1621 (
1622 deployment_regions["global"],
1623 deployment_regions["api_gateway"],
1624 deployment_regions["monitoring"],
1625 *regional,
1626 )
1627 )
1629 def supports_global_accelerator(self) -> bool:
1630 """Return whether this partition exposes the Global Accelerator topology."""
1631 return self.get_deployment_partition() == "aws"
1633 def get_global_region(self) -> str:
1634 """Get the region for global resources and shared SSM parameters."""
1635 region = self.get_deployment_regions()["global"]
1636 return str(region)
1638 def get_api_gateway_region(self) -> str:
1639 """Get the region for API Gateway stack."""
1640 region = self.get_deployment_regions()["api_gateway"]
1641 return str(region)
1643 def get_monitoring_region(self) -> str:
1644 """Get the region for Monitoring stack."""
1645 region = self.get_deployment_regions()["monitoring"]
1646 return str(region)
1648 def get_regions(self) -> list[str]:
1649 """Get list of regions for EKS cluster deployment (empty when scaled to zero)."""
1650 deployment_regions = self.get_deployment_regions()
1651 regional = deployment_regions["regional"]
1652 return list(regional) if isinstance(regional, list) else [str(regional)]
1654 def get_kubernetes_version(self) -> str:
1655 """Get Kubernetes version from configuration"""
1656 return self.app.node.try_get_context("kubernetes_version") or "1.37"
1658 def get_resource_thresholds(self) -> ResourceThresholds:
1659 """Get resource thresholds configuration"""
1660 thresholds_config = self.app.node.try_get_context("resource_thresholds") or {
1661 "cpu_threshold": 80,
1662 "memory_threshold": 80,
1663 "gpu_threshold": -1,
1664 "pending_pods_threshold": 10,
1665 "pending_requested_cpu_vcpus": 100,
1666 "pending_requested_memory_gb": 200,
1667 "pending_requested_gpus": -1,
1668 }
1669 return ResourceThresholds(
1670 cpu_threshold=thresholds_config["cpu_threshold"],
1671 memory_threshold=thresholds_config["memory_threshold"],
1672 gpu_threshold=thresholds_config["gpu_threshold"],
1673 pending_pods_threshold=thresholds_config.get("pending_pods_threshold", 10),
1674 pending_requested_cpu_vcpus=thresholds_config.get("pending_requested_cpu_vcpus", 100),
1675 pending_requested_memory_gb=thresholds_config.get("pending_requested_memory_gb", 200),
1676 pending_requested_gpus=thresholds_config.get("pending_requested_gpus", 8),
1677 )
1679 def get_cluster_config(self, region: str) -> ClusterConfig:
1680 """Get complete cluster configuration for a region"""
1681 return ClusterConfig(
1682 region=region,
1683 cluster_name=f"{self.get_project_name()}-{region}",
1684 kubernetes_version=self.get_kubernetes_version(),
1685 addons=["metrics-server"],
1686 resource_thresholds=self.get_resource_thresholds(),
1687 )
1689 def get_global_accelerator_config(self) -> dict[str, Any]:
1690 """Get the merged Global Accelerator configuration.
1692 Returns the ``global_accelerator`` block from cdk.json layered on top
1693 of the defaults below. This is a real merge, not the historical
1694 all-or-nothing fallback: a partial block keeps every unspecified
1695 default instead of silently dropping it. The ``traffic_dial``
1696 sub-block is deep-merged so overriding a single dial knob does not
1697 wipe the sub-block's other defaults, mirroring
1698 ``get_cost_monitoring_config``.
1700 Keys:
1701 - name: accelerator name (default ``<project_name>-accelerator``)
1702 - health_check_interval: seconds between endpoint-group probes;
1703 the Global Accelerator API accepts only 10 or 30 (default 30)
1704 - health_check_threshold: consecutive probes before an endpoint
1705 flips healthy/unhealthy, 1-10 (default 3)
1706 - health_check_path: HTTPS path probed on each regional ALB
1707 (default ``/api/v1/health``)
1708 - client_affinity: ``NONE`` or ``SOURCE_IP`` (default ``NONE``)
1709 - traffic_dial: capacity-driven traffic-dial controller
1710 sub-block (default disabled, ``monitor`` mode; knob reference
1711 in :meth:`_validate_traffic_dial_config`)
1712 """
1713 default_config: dict[str, Any] = {
1714 "name": f"{self.get_project_name()}-accelerator",
1715 "health_check_interval": 30,
1716 "health_check_threshold": 3,
1717 "health_check_path": "/api/v1/health",
1718 "client_affinity": "NONE",
1719 "traffic_dial": {
1720 "enabled": False,
1721 "mode": "monitor",
1722 "interval_minutes": 5,
1723 "lookback_minutes": 15,
1724 "min_dial_percentage": 10,
1725 "max_step_percentage": 20,
1726 "full_health_percentage": 95,
1727 },
1728 }
1729 configured = self.app.node.try_get_context("global_accelerator") or {}
1730 if not isinstance(configured, dict):
1731 raise ConfigValidationError("global_accelerator must be a mapping")
1732 merged: dict[str, Any] = {**default_config, **configured}
1734 # Deep-merge the nested sub-block so a partial override does not drop
1735 # the other defaults in the same sub-block. A non-mapping override is
1736 # deliberately left in place for the validator to reject with a
1737 # precise message.
1738 override = configured.get("traffic_dial")
1739 if isinstance(override, dict):
1740 default_sub = cast(dict[str, Any], default_config["traffic_dial"])
1741 merged["traffic_dial"] = {**default_sub, **override}
1743 return merged
1745 def get_backend_tls_config(self) -> dict[str, Any]:
1746 """Return the mandatory deployment-local backend TLS lifecycle policy."""
1747 defaults = {
1748 "root_generation": 1,
1749 "root_validity_days": 3_650,
1750 "root_rotate_before_days": 180,
1751 "root_activation_delay_hours": 24,
1752 "root_overlap_days": 45,
1753 "leaf_validity_days": 30,
1754 "leaf_rotate_before_days": 10,
1755 "rotation_schedule_hours": 12,
1756 "trust_cache_ttl_seconds": 300,
1757 "trust_cache_max_stale_seconds": 3_600,
1758 }
1759 configured = self.app.node.try_get_context("backend_tls") or {}
1760 if not isinstance(configured, dict):
1761 raise ConfigValidationError("backend_tls must be a mapping")
1762 return {**defaults, **configured}
1764 def get_inference_proxy_config(self) -> dict[str, int]:
1765 """Return merged inference TLS proxy autoscaling settings.
1767 Omission is backward compatible and returns the shipped defaults. AWS
1768 CDK normalizes a top-level JSON ``null`` context value to omission, so
1769 ``None`` follows the same default-preserving contract.
1770 """
1771 defaults = {
1772 "tls_proxy_cpu_request_millicores": (
1773 INFERENCE_PROXY_TLS_CPU_REQUEST_MILLICORES_DEFAULT
1774 ),
1775 "tls_proxy_cpu_target_utilization_percentage": (
1776 INFERENCE_PROXY_TLS_CPU_TARGET_UTILIZATION_DEFAULT
1777 ),
1778 "min_replicas": INFERENCE_PROXY_MIN_REPLICAS_DEFAULT,
1779 "max_replicas": INFERENCE_PROXY_MAX_REPLICAS_DEFAULT,
1780 }
1781 configured = self.app.node.try_get_context("inference_proxy")
1782 if configured is None:
1783 return dict(defaults)
1784 if not isinstance(configured, dict):
1785 raise ConfigValidationError(
1786 "inference_proxy must be an object, got "
1787 f"{type(configured).__name__}: {configured!r}"
1788 )
1790 unknown = sorted(str(key) for key in configured if key not in defaults)
1791 if unknown:
1792 unknown_paths = ", ".join(f"inference_proxy.{key}" for key in unknown)
1793 allowed_paths = ", ".join(f"inference_proxy.{key}" for key in sorted(defaults))
1794 raise ConfigValidationError(
1795 f"inference_proxy contains unknown key(s): {unknown_paths}; "
1796 f"allowed keys: {allowed_paths}"
1797 )
1798 return {**defaults, **configured}
1800 def get_alb_config(self) -> dict[str, Any]:
1801 """Get ALB configuration"""
1802 return self.app.node.try_get_context("alb_config") or {
1803 "health_check_interval": 30,
1804 "health_check_timeout": 5,
1805 "healthy_threshold": 2,
1806 "unhealthy_threshold": 2,
1807 }
1809 def get_manifest_processor_config(self) -> dict[str, Any]:
1810 """Get manifest processor configuration.
1812 Merges three cdk.json sections into a single runtime config:
1814 - ``manifest_processor``: service-specific settings (replicas, image,
1815 resource_limits, allowed_namespaces, validation_enabled,
1816 max_request_body_bytes, yaml_max_depth)
1817 - ``job_validation_policy``: shared validation policy (resource_quotas,
1818 trusted_registries, trusted_dockerhub_orgs, manifest_security_policy,
1819 allowed_kinds). Pulled in verbatim so the REST path reads the same
1820 policy the SQS queue processor enforces.
1822 Note: The 'image' field is a placeholder default. In practice, the actual
1823 image is built from dockerfiles/Dockerfile.manifest-processor and pushed
1824 to ECR during CDK deployment. The {{MANIFEST_PROCESSOR_IMAGE}} placeholder
1825 in manifests is replaced with the ECR image URI.
1826 """
1827 default_config = {
1828 "image": "gco/manifest-processor:latest", # Placeholder, replaced by ECR image
1829 "replicas": 3,
1830 "resource_limits": {"cpu": "1000m", "memory": "2Gi"},
1831 "autoscaling": dict(_MANIFEST_PROCESSOR_AUTOSCALING_DEFAULTS),
1832 "validation_enabled": True,
1833 "max_request_body_bytes": DEFAULT_MAX_REQUEST_BODY_BYTES,
1834 "central_queue_worker_enabled": True,
1835 "central_queue_poll_interval_seconds": 10,
1836 "central_queue_batch_size": 5,
1837 "central_queue_reconcile_limit": 100,
1838 "central_queue_lease_seconds": 300,
1839 "central_queue_lease_renewal_seconds": 60,
1840 # allowed_namespaces, resource_quotas, trusted_registries,
1841 # trusted_dockerhub_orgs, manifest_security_policy, and
1842 # allowed_kinds are merged in below from job_validation_policy.
1843 "allowed_namespaces": ["gco-jobs"],
1844 "resource_quotas": {
1845 "max_cpu_per_manifest": "10",
1846 "max_memory_per_manifest": "32Gi",
1847 "max_gpu_per_manifest": 4,
1848 },
1849 "trusted_registries": [
1850 "docker.io",
1851 "gcr.io",
1852 "quay.io",
1853 "registry.k8s.io",
1854 "k8s.gcr.io",
1855 "public.ecr.aws",
1856 "nvcr.io",
1857 "gco",
1858 ],
1859 "trusted_dockerhub_orgs": [
1860 "nvidia",
1861 "pytorch",
1862 "rayproject",
1863 "tensorflow",
1864 "huggingface",
1865 "amazon",
1866 "bitnami",
1867 ],
1868 }
1869 context_config = self.app.node.try_get_context("manifest_processor") or {}
1871 # Merge in the shared job_validation_policy section. These keys apply
1872 # to BOTH the manifest processor and the queue processor; they live
1873 # in their own top-level cdk.json section so neither service "owns"
1874 # them. We flatten them into the manifest processor's runtime config
1875 # so service code keeps its existing attribute layout.
1876 shared_policy = self.app.node.try_get_context("job_validation_policy") or {}
1877 merged = {**default_config, **context_config, **shared_policy}
1878 # Nested block: a partial ``autoscaling`` object (or JSON null) keeps
1879 # the unspecified defaults instead of replacing the whole mapping.
1880 # ``_validate_config`` has already rejected anything that is not a
1881 # mapping or null.
1882 merged["autoscaling"] = {
1883 **_MANIFEST_PROCESSOR_AUTOSCALING_DEFAULTS,
1884 **(context_config.get("autoscaling") or {}),
1885 }
1887 enabled = merged.get("central_queue_worker_enabled")
1888 if not isinstance(enabled, bool):
1889 raise ConfigValidationError(
1890 "manifest_processor.central_queue_worker_enabled must be a boolean"
1891 )
1892 for key, minimum, maximum in (
1893 ("central_queue_poll_interval_seconds", 1, 300),
1894 ("central_queue_batch_size", 1, 20),
1895 ("central_queue_reconcile_limit", 1, 500),
1896 ("central_queue_lease_seconds", 30, 3600),
1897 ("central_queue_lease_renewal_seconds", 1, 300),
1898 ):
1899 value = merged.get(key)
1900 if type(value) is not int or not minimum <= value <= maximum:
1901 raise ConfigValidationError(
1902 f"manifest_processor.{key} must be an integer between {minimum} and {maximum}"
1903 )
1904 if (
1905 merged["central_queue_lease_renewal_seconds"] * 2
1906 > merged["central_queue_lease_seconds"]
1907 ):
1908 raise ConfigValidationError(
1909 "manifest_processor.central_queue_lease_renewal_seconds must be no more than "
1910 "half of central_queue_lease_seconds"
1911 )
1912 return merged
1914 def get_api_gateway_config(self) -> dict[str, Any]:
1915 """Get API Gateway configuration.
1917 Returns:
1918 API Gateway configuration dictionary with the following keys:
1919 - throttle_rate_limit: Requests per second limit
1920 - throttle_burst_limit: Burst capacity
1921 - log_level: CloudWatch logging level (OFF, ERROR, INFO)
1922 - metrics_enabled: Enable CloudWatch metrics
1923 - tracing_enabled: Enable X-Ray tracing
1924 - regional_api_enabled: In the commercial ``aws`` partition,
1925 permit direct same-account callers to use the always-deployed
1926 regional API bridges. Other partitions force this access on
1927 because the bridges are the supported workload ingress without
1928 Global Accelerator. Centralized aggregation always uses them.
1929 """
1930 default_config = {
1931 "throttle_rate_limit": 1000,
1932 "throttle_burst_limit": 2000,
1933 "log_level": "INFO",
1934 "metrics_enabled": True,
1935 "tracing_enabled": True,
1936 "regional_api_enabled": False,
1937 }
1938 return {**default_config, **(self.app.node.try_get_context("api_gateway") or {})}
1940 def get_eks_cluster_config(self) -> dict[str, Any]:
1941 """Get EKS cluster configuration.
1943 Returns:
1944 EKS cluster configuration dictionary with the following keys:
1945 - endpoint_access: EKS API endpoint access mode
1946 - "PRIVATE": API server only accessible from within VPC (default, most secure)
1947 - "PUBLIC_AND_PRIVATE": API server accessible from internet and VPC
1948 - public_access_cidrs: CIDR allowlist for the public endpoint when
1949 endpoint_access is PUBLIC_AND_PRIVATE. Empty (the default) means
1950 0.0.0.0/0, which synthesis calls out with a loud warning.
1951 - developer_access: list of EKS access entries to synthesize for
1952 human principals, each ``{principal_arn, scope, namespaces}``.
1953 scope defaults to "namespace" and namespaces to ["gco-jobs"];
1954 scope "cluster" grants AmazonEKSClusterAdminPolicy instead.
1955 Empty (the default) synthesizes exactly today's entries.
1956 - network_policy_enforcement: whether the Auto Mode network policy
1957 controller is switched on (default True). False keeps the
1958 NetworkPolicy objects but stops enforcing them.
1960 Note:
1961 PRIVATE endpoint is recommended for production. Job submission still works
1962 via API Gateway → Lambda (in VPC) or SQS queues. For kubectl access with
1963 PRIVATE endpoint, use `gco cluster tunnel` (SSM), a bastion host, or a VPN —
1964 and an access entry for your principal either way (`gco stacks access`).
1965 """
1966 default_config: dict[str, Any] = {
1967 "endpoint_access": "PRIVATE",
1968 "public_access_cidrs": [],
1969 "developer_access": [],
1970 # Switches on the Auto Mode network policy controller
1971 # (06-network-policy-controller.yaml) so 03-network-policies.yaml
1972 # is enforced rather than merely stored.
1973 "network_policy_enforcement": True,
1974 }
1975 return {**default_config, **(self.app.node.try_get_context("eks_cluster") or {})}
1977 def get_fsx_lustre_config(self, region: str | None = None) -> dict[str, Any]:
1978 """Get FSx for Lustre configuration.
1980 Args:
1981 region: Optional region to get config for. If provided, checks for
1982 region-specific overrides first.
1984 Returns:
1985 FSx configuration dictionary with the following keys:
1986 - enabled: Whether FSx is enabled
1987 - storage_capacity_gib: Storage capacity in GiB (min 1200)
1988 - deployment_type: SCRATCH_1, SCRATCH_2, PERSISTENT_1, PERSISTENT_2
1989 - file_system_type_version: Lustre version (2.12 or 2.15, default: 2.15)
1990 IMPORTANT: Use 2.15 for kernel 6.x compatibility (AL2023, Bottlerocket)
1991 - per_unit_storage_throughput: Throughput for PERSISTENT types
1992 - data_compression_type: LZ4 or NONE
1993 - import_path: S3 path for data import
1994 - export_path: S3 path for data export
1995 - auto_import_policy: NEW, NEW_CHANGED, NEW_CHANGED_DELETED
1996 - node_group: Node group configuration for FSx workloads
1997 - instance_types: List of instance types
1998 - min_size: Minimum nodes (default: 0)
1999 - max_size: Maximum nodes (default: 10)
2000 - desired_size: Desired nodes (default: 0, scales from zero)
2001 - ami_type: AMI type - one of:
2002 AL2023_X86_64_STANDARD (default), AL2023_ARM_64_STANDARD,
2003 AL2023_X86_64_NVIDIA, AL2023_ARM_64_NVIDIA, AL2023_X86_64_NEURON
2004 - capacity_type: ON_DEMAND (default) or SPOT
2005 - disk_size: Root disk size in GB (default: 100)
2006 - labels: Additional node labels (dict)
2007 """
2008 default_config = {
2009 "enabled": False,
2010 "storage_capacity_gib": 1200,
2011 "deployment_type": "SCRATCH_2",
2012 "file_system_type_version": "2.15", # Use 2.15 for kernel 6.x compatibility
2013 "per_unit_storage_throughput": 200,
2014 "data_compression_type": "LZ4",
2015 "import_path": None,
2016 "export_path": None,
2017 "auto_import_policy": "NEW_CHANGED_DELETED",
2018 "node_group": {
2019 "instance_types": ["m5.large", "m5.xlarge", "m6i.large", "m6i.xlarge"],
2020 "min_size": 0,
2021 "max_size": 10,
2022 "desired_size": 1,
2023 "ami_type": "AL2023_X86_64_STANDARD",
2024 "capacity_type": "ON_DEMAND",
2025 "disk_size": 100,
2026 "labels": {},
2027 },
2028 }
2030 # Get global FSx config
2031 global_ctx = self.app.node.try_get_context("fsx_lustre")
2032 global_config: dict[str, Any] = global_ctx if isinstance(global_ctx, dict) else {}
2033 merged_config: dict[str, Any] = {**default_config, **global_config}
2035 # Ensure node_group has all required fields with defaults
2036 if "node_group" in global_config:
2037 global_node_group = global_config["node_group"]
2038 if isinstance(global_node_group, dict):
2039 default_node_group = cast(dict[str, Any], default_config["node_group"])
2040 merged_config["node_group"] = {
2041 **default_node_group,
2042 **global_node_group,
2043 }
2045 # Check for region-specific override
2046 if region:
2047 region_overrides_ctx = self.app.node.try_get_context("fsx_lustre_regions")
2048 region_overrides: dict[str, Any] = (
2049 region_overrides_ctx if isinstance(region_overrides_ctx, dict) else {}
2050 )
2051 if region in region_overrides:
2052 region_config = region_overrides[region]
2053 if isinstance(region_config, dict):
2054 # Preserve the fully merged default/global node group before
2055 # the top-level regional overlay replaces that nested value.
2056 # A regional node_group is a patch, not a wholesale reset.
2057 existing_node_group = merged_config.get("node_group")
2058 merged_config = {**merged_config, **region_config}
2059 # Handle nested node_group override
2060 if "node_group" in region_config:
2061 region_node_group = region_config["node_group"]
2062 if isinstance(region_node_group, dict):
2063 if isinstance(existing_node_group, dict):
2064 base_node_group = existing_node_group
2065 else:
2066 base_node_group = cast(dict[str, Any], default_config["node_group"])
2067 merged_config["node_group"] = {
2068 **base_node_group,
2069 **region_node_group,
2070 }
2072 if self._feature_override_enabled("fsx_lustre"):
2073 merged_config["enabled"] = True
2074 return merged_config
2076 def _feature_override_enabled(self, feature_key: str) -> bool:
2077 """True when ``feature_enabled_overrides`` context forces this feature on."""
2078 overrides = parse_feature_enabled_overrides(
2079 self.app.node.try_get_context(FEATURE_OVERRIDE_CONTEXT_KEY)
2080 )
2081 return feature_key in overrides
2083 def get_valkey_config(self) -> dict[str, Any]:
2084 """Get Valkey Serverless cache configuration.
2086 Returns:
2087 Valkey configuration dictionary with the following keys:
2088 - enabled: Whether Valkey cache is enabled (default: False)
2089 - max_data_storage_gb: Maximum data storage in GB (default: 5)
2090 - max_ecpu_per_second: Maximum ECPUs per second (default: 5000)
2091 - snapshot_retention_limit: Daily snapshots to retain (default: 1)
2092 """
2093 default_config: dict[str, Any] = {
2094 "enabled": False,
2095 "max_data_storage_gb": 5,
2096 "max_ecpu_per_second": 5000,
2097 "snapshot_retention_limit": 1,
2098 }
2099 valkey_ctx = self.app.node.try_get_context("valkey")
2100 valkey_config: dict[str, Any] = valkey_ctx if isinstance(valkey_ctx, dict) else {}
2101 merged = {**default_config, **valkey_config}
2102 if self._feature_override_enabled("valkey"):
2103 merged["enabled"] = True
2104 return merged
2106 def get_aurora_pgvector_config(self) -> dict[str, Any]:
2107 """Get Aurora Serverless v2 + pgvector vector database configuration.
2109 Returns:
2110 Aurora pgvector configuration dictionary with the following keys:
2111 - enabled: Whether Aurora pgvector is enabled (default: False)
2112 - min_acu: Minimum Aurora Capacity Units (default: 0, scales to zero)
2113 - max_acu: Maximum Aurora Capacity Units (default: 16)
2114 - backup_retention_days: Number of days to retain automated backups (default: 7)
2115 - deletion_protection: Whether deletion protection is enabled (default: False)
2116 """
2117 default_config: dict[str, Any] = {
2118 "enabled": False,
2119 "min_acu": 0,
2120 "max_acu": 16,
2121 "backup_retention_days": 7,
2122 "deletion_protection": False,
2123 }
2124 aurora_ctx = self.app.node.try_get_context("aurora_pgvector")
2125 aurora_config: dict[str, Any] = aurora_ctx if isinstance(aurora_ctx, dict) else {}
2126 merged = {**default_config, **aurora_config}
2127 if self._feature_override_enabled("aurora_pgvector"):
2128 merged["enabled"] = True
2129 return merged
2131 def get_analytics_config(self) -> dict[str, Any]:
2132 """Get optional analytics environment configuration.
2134 Returns the fully-merged analytics_environment block from cdk.json
2135 layered on top of the defaults below. Sub-blocks (``hyperpod``,
2136 ``cognito``, ``efs``, ``studio``) are deep-merged so a user who
2137 overrides a single nested key (e.g. ``cognito.domain_prefix``) does
2138 not inadvertently wipe the sub-block's other defaults — mirroring the
2139 nested-merge pattern used by ``get_fsx_lustre_config`` for its
2140 ``node_group`` sub-block.
2142 Returns:
2143 Analytics configuration dictionary with the following keys:
2144 - enabled: Whether the analytics environment stack is deployed
2145 (default: False — the feature is off unless explicitly opted in)
2146 - hyperpod: SageMaker HyperPod integration sub-block
2147 - enabled: Whether to add the HyperPod IAM grants to
2148 SageMaker_Execution_Role (default: False)
2149 - canvas: SageMaker Canvas integration sub-block
2150 - enabled: Whether to enable the SageMaker Canvas app on
2151 the Studio domain and attach ``AmazonSageMakerCanvasFullAccess``
2152 to the SageMaker_Execution_Role (default: False)
2153 - cognito: Cognito user-pool sub-block
2154 - domain_prefix: UserPoolDomain prefix, or None to let the
2155 analytics stack derive one (default: None)
2156 - removal_policy: "destroy" (default) or "retain" — controls
2157 the Cognito pool's CloudFormation DeletionPolicy
2158 - efs: Studio_EFS sub-block
2159 - removal_policy: "destroy" (default) or "retain" — controls
2160 the Studio EFS file system's CloudFormation DeletionPolicy
2161 - studio: SageMaker Studio sub-block
2162 - user_profile_name_prefix: Optional prefix for per-user
2163 profile names, or None to use the Cognito username verbatim
2164 (default: None)
2165 """
2166 default_config: dict[str, Any] = {
2167 "enabled": False,
2168 "hyperpod": {"enabled": False},
2169 "canvas": {"enabled": False},
2170 "cognito": {"domain_prefix": None, "removal_policy": "destroy"},
2171 "efs": {"removal_policy": "destroy"},
2172 "studio": {"user_profile_name_prefix": None},
2173 }
2174 analytics_ctx = self.app.node.try_get_context("analytics_environment")
2175 analytics_config: dict[str, Any] = analytics_ctx if isinstance(analytics_ctx, dict) else {}
2176 merged_config: dict[str, Any] = {**default_config, **analytics_config}
2178 # Deep-merge each nested sub-block so a partial override does not
2179 # drop the other defaults in the same sub-block.
2180 for sub_block in ("hyperpod", "canvas", "cognito", "efs", "studio"):
2181 override = analytics_config.get(sub_block)
2182 if isinstance(override, dict):
2183 default_sub = cast(dict[str, Any], default_config[sub_block])
2184 merged_config[sub_block] = {**default_sub, **override}
2186 return merged_config
2188 def get_analytics_enabled(self) -> bool:
2189 """Return whether the analytics environment stack is enabled.
2191 Thin wrapper around ``get_analytics_config()["enabled"]`` to mirror
2192 the existing ``get_valkey_config`` / ``get_aurora_pgvector_config``
2193 access pattern without forcing every call site to index into the
2194 merged dict.
2195 """
2196 return bool(self.get_analytics_config()["enabled"])
2198 def get_cluster_observability_config(self) -> dict[str, Any]:
2199 """Get the in-cluster observability configuration.
2201 Returns the fully-merged cluster_observability block from cdk.json
2202 layered on top of the defaults below. Sub-blocks (``grafana``,
2203 ``prometheus``, ``alertmanager``) are deep-merged so a user who
2204 overrides a single nested key (e.g. ``prometheus.retention``) does not
2205 inadvertently wipe the sub-block's other defaults — mirroring the
2206 nested-merge pattern used by ``get_analytics_config``.
2208 Unlike most optional features, this one is **on by default**: a stock
2209 deployment installs kube-prometheus-stack on every regional cluster.
2210 Operators opt out by setting ``cluster_observability.enabled = false``.
2212 Returns:
2213 Cluster observability configuration dictionary with the keys:
2214 - enabled: Whether kube-prometheus-stack is installed per region
2215 (default: True)
2216 - grafana: Grafana sub-block
2217 - persistence_size: EBS PVC size for Grafana's user database
2218 and dashboards (default: "10Gi")
2219 - admin_user: Grafana admin username; the password is
2220 chart-generated in the <release>-grafana Secret, never
2221 authored here (default: "admin")
2222 - admin_password_rotation_schedule: 5-field cron for the
2223 in-cluster CronJob that rotates the chart-generated admin
2224 password (default: "0 4 1 * *", monthly)
2225 - prometheus: Prometheus sub-block
2226 - persistence_size: EBS PVC size for the Prometheus TSDB
2227 (default: "50Gi")
2228 - retention: Prometheus retention window (default: "15d")
2229 - alertmanager: Alertmanager sub-block
2230 - enabled: Whether Alertmanager is deployed (default: True)
2231 - persistence_size: EBS PVC size for Alertmanager (default: "5Gi")
2232 - mlflow: MLflow experiment-tracking sub-block
2233 - enabled: Whether the MLflow tracking server is installed per
2234 region (default: True). Effective only while observability
2235 itself is enabled — see ``get_mlflow_enabled``.
2236 - persistence_size: EBS PVC size for the tracking server's
2237 SQLite run-metadata store (default: "10Gi"); artifacts go
2238 to S3, not this volume
2239 """
2240 default_config: dict[str, Any] = {
2241 "enabled": True,
2242 "grafana": {
2243 "persistence_size": "10Gi",
2244 "admin_user": "admin",
2245 # Monthly (04:00 on the 1st) rotation of the chart-generated
2246 # Grafana admin password, run by an in-cluster CronJob.
2247 "admin_password_rotation_schedule": "0 4 1 * *", # nosec B105 # cron schedule, not a password
2248 },
2249 "prometheus": {"persistence_size": "50Gi", "retention": "15d"},
2250 "alertmanager": {"enabled": True, "persistence_size": "5Gi"},
2251 "mlflow": {"enabled": True, "persistence_size": "10Gi"},
2252 }
2253 obs_ctx = self.app.node.try_get_context("cluster_observability")
2254 obs_config: dict[str, Any] = obs_ctx if isinstance(obs_ctx, dict) else {}
2255 merged_config: dict[str, Any] = {**default_config, **obs_config}
2257 # Deep-merge each nested sub-block so a partial override does not
2258 # drop the other defaults in the same sub-block.
2259 for sub_block in ("grafana", "prometheus", "alertmanager", "mlflow"):
2260 override = obs_config.get(sub_block)
2261 if isinstance(override, dict):
2262 default_sub = cast(dict[str, Any], default_config[sub_block])
2263 merged_config[sub_block] = {**default_sub, **override}
2265 return merged_config
2267 def get_cluster_observability_enabled(self) -> bool:
2268 """Return whether in-cluster observability is enabled (default True).
2270 Thin wrapper around ``get_cluster_observability_config()["enabled"]``
2271 so call sites (the regional stack's chart-enable and value-override
2272 methods, the CLI) do not have to index into the merged dict.
2273 """
2274 return bool(self.get_cluster_observability_config()["enabled"])
2276 def get_mlflow_enabled(self) -> bool:
2277 """Return whether the MLflow tracking server is effectively enabled.
2279 The conjunction of ``cluster_observability.mlflow.enabled`` (default
2280 True) and ``cluster_observability.enabled`` (default True): MLflow
2281 installs into the ``monitoring`` namespace kube-prometheus-stack
2282 creates, stores run metadata on the observability gp3 StorageClass,
2283 and is reached through the same tunnel commands, so disabling
2284 observability switches the tracking server off with it rather than
2285 deploying it against missing storage — the same conjunction shape
2286 ``get_cost_monitoring_enabled`` uses for OpenCost.
2287 """
2288 obs = self.get_cluster_observability_config()
2289 return bool(obs["mlflow"]["enabled"]) and bool(obs["enabled"])
2291 def get_cost_monitoring_config(self) -> dict[str, Any]:
2292 """Get the cost monitoring configuration.
2294 Returns the fully-merged ``cost_monitoring`` block from cdk.json
2295 layered on top of the defaults below. Sub-blocks (``reports``,
2296 ``athena``) are deep-merged so a user who overrides a single nested
2297 key does not inadvertently wipe the sub-block's other defaults —
2298 mirroring the nested-merge pattern used by
2299 ``get_cluster_observability_config``.
2301 Like cluster observability, cost monitoring is **on by default**: a
2302 stock deployment installs OpenCost per region, provisions the cost
2303 report bucket + Athena analytics in the monitoring stack, and runs
2304 the cost-monitor service on every regional cluster. Operators opt out
2305 by setting ``cost_monitoring.enabled = false``.
2307 Returns:
2308 Cost monitoring configuration dictionary with the keys:
2309 - enabled: Whether the cost monitoring pipeline is deployed
2310 (default: True). Requires ``cluster_observability.enabled``.
2311 - reports: Cost report sub-block
2312 - interval_minutes: cadence of the cost-monitor service's
2313 scheduled Parquet reports (default: 60)
2314 - retention_days: S3 lifecycle expiration for report objects
2315 (default: 365)
2316 - transition_to_infrequent_access_days: S3 lifecycle transition
2317 to STANDARD_IA (default: 90; must be < retention_days)
2318 - athena: Athena analytics sub-block
2319 - query_results_retention_days: S3 lifecycle expiration for
2320 Athena query results written under ``athena-results/``
2321 (default: 30)
2322 """
2323 default_config: dict[str, Any] = {
2324 "enabled": True,
2325 "reports": {
2326 "interval_minutes": 60,
2327 "retention_days": 365,
2328 "transition_to_infrequent_access_days": 90,
2329 },
2330 "athena": {
2331 "query_results_retention_days": 30,
2332 },
2333 }
2334 cost_ctx = self.app.node.try_get_context("cost_monitoring")
2335 cost_config: dict[str, Any] = cost_ctx if isinstance(cost_ctx, dict) else {}
2336 merged_config: dict[str, Any] = {**default_config, **cost_config}
2338 # Deep-merge each nested sub-block so a partial override does not
2339 # drop the other defaults in the same sub-block.
2340 for sub_block in ("reports", "athena"):
2341 override = cost_config.get(sub_block)
2342 if isinstance(override, dict):
2343 default_sub = cast(dict[str, Any], default_config[sub_block])
2344 merged_config[sub_block] = {**default_sub, **override}
2346 return merged_config
2348 def get_cost_monitoring_enabled(self) -> bool:
2349 """Return whether the cost monitoring pipeline is effectively enabled.
2351 The conjunction of ``cost_monitoring.enabled`` (default True) and
2352 ``cluster_observability.enabled`` (default True): OpenCost reads its
2353 usage data from the in-cluster Prometheus, so disabling observability
2354 switches the whole cost pipeline off with it rather than deploying a
2355 pipeline with no data source (or failing synthesis for a user who
2356 only ran ``gco monitoring disable``). Call sites — the regional
2357 stack's chart-enable and image-build methods, the monitoring stack,
2358 the CLI — all gate on this one conjunction.
2359 """
2360 return bool(self.get_cost_monitoring_config()["enabled"]) and bool(
2361 self.get_cluster_observability_config()["enabled"]
2362 )
2364 def _validate_tracing_config(self) -> None:
2365 """Validate the optional ``tracing`` block and ``tracing_overrides``.
2367 Resolving the block is the validation: unknown keys, non-boolean
2368 flags, a ``sample_ratio`` outside ``[0, 1]`` and malformed overrides
2369 all raise :class:`ConfigValidationError` at synth instead of shipping
2370 a value the services would silently replace with their defaults.
2371 """
2372 self.get_tracing_config()
2374 def get_tracing_config(self) -> dict[str, Any]:
2375 """Return the validated ``tracing`` block with defaults merged in.
2377 Keys (see :data:`TRACING_DEFAULTS`):
2378 - enabled: render ``GCO_TRACING_ENABLED=true`` into the
2379 health-monitor, manifest-processor, inference-proxy and
2380 cost-monitor Deployments and grant their roles the X-Ray span
2381 export actions (default True)
2382 - sample_ratio: head-sampling probability of a new trace, a float
2383 in ``[0, 1]`` (default 0.05)
2384 - enable_transaction_search: let each regional stack switch on
2385 CloudWatch Transaction Search in its Region (default True;
2386 effective only while ``enabled`` — see
2387 :meth:`get_transaction_search_enabled`)
2389 Honors the run-scoped ``tracing_overrides`` context (a JSON object
2390 deep-merged over the block), which the live release harness uses to
2391 sample every trace of its run.
2392 """
2393 return resolve_tracing_config(
2394 self.app.node.try_get_context(TRACING_CONTEXT_KEY),
2395 self.app.node.try_get_context(TRACING_OVERRIDES_CONTEXT_KEY),
2396 )
2398 def get_tracing_enabled(self) -> bool:
2399 """Return whether the API services export traces (default True)."""
2400 return bool(self.get_tracing_config()["enabled"])
2402 def get_transaction_search_enabled(self) -> bool:
2403 """Return whether the regional stacks enable CloudWatch Transaction Search.
2405 The conjunction of ``tracing.enabled`` and
2406 ``tracing.enable_transaction_search``: Transaction Search only exists
2407 to receive GCO's spans, so a deployment with tracing off never touches
2408 the account-level setting.
2409 """
2410 config = self.get_tracing_config()
2411 return bool(config["enabled"]) and bool(config["enable_transaction_search"])
2413 def get_capacity_history_config(self) -> dict[str, Any]:
2414 """Get the optional historical capacity surface configuration.
2416 Returns the merged ``historical`` block from cdk.json layered on top of
2417 the defaults below. The feature is off unless ``historical.enabled`` is
2418 explicitly true, mirroring the analytics-environment opt-in pattern.
2420 Keys:
2421 - enabled: deploy the capacity poller stack + history table (default False)
2422 - retention_days: DynamoDB TTL window for snapshots (default 90)
2423 - poll_interval_minutes: EventBridge schedule cadence (default 15)
2424 - capacity_block_duration_hours: short Capacity Block probe duration
2425 the poller snapshots (default 24 = 1 day)
2426 - capacity_block_long_duration_hours: long Capacity Block probe
2427 duration in hours (default 1512 = 63 days); 0 disables the long
2428 probe and its ``capacity_blocks_long_*`` metrics
2429 - spot_score_target_capacities: Spot Placement Score target
2430 capacities the poller snapshots per instance pool (default
2431 [1, 10, 50]); a subset selector over the supported set exported
2432 by ``cli/capacity/history.py``, where capacity 1 keeps the
2433 original ``spot_score`` field and N > 1 writes ``spot_score_at_N``
2434 - watch_instance_types: instance types the poller snapshots
2435 - enabled_regions: regions to poll; empty means all deployed regions
2436 """
2437 default_config: dict[str, Any] = {
2438 "enabled": False,
2439 "retention_days": 90,
2440 "poll_interval_minutes": 15,
2441 "capacity_block_duration_hours": 24,
2442 "capacity_block_long_duration_hours": 63 * 24,
2443 "spot_score_target_capacities": [1, 10, 50],
2444 "watch_instance_types": [
2445 "g4dn.12xlarge",
2446 "g4dn.16xlarge",
2447 "g4dn.2xlarge",
2448 "g4dn.4xlarge",
2449 "g4dn.8xlarge",
2450 "g4dn.metal",
2451 "g4dn.xlarge",
2452 "g5.12xlarge",
2453 "g5.16xlarge",
2454 "g5.24xlarge",
2455 "g5.2xlarge",
2456 "g5.48xlarge",
2457 "g5.4xlarge",
2458 "g5.8xlarge",
2459 "g5.xlarge",
2460 "g5g.16xlarge",
2461 "g5g.2xlarge",
2462 "g5g.4xlarge",
2463 "g5g.8xlarge",
2464 "g5g.metal",
2465 "g5g.xlarge",
2466 "g6.12xlarge",
2467 "g6.16xlarge",
2468 "g6.24xlarge",
2469 "g6.2xlarge",
2470 "g6.48xlarge",
2471 "g6.4xlarge",
2472 "g6.8xlarge",
2473 "g6.xlarge",
2474 "g6e.12xlarge",
2475 "g6e.16xlarge",
2476 "g6e.24xlarge",
2477 "g6e.2xlarge",
2478 "g6e.48xlarge",
2479 "g6e.4xlarge",
2480 "g6e.8xlarge",
2481 "g6e.xlarge",
2482 "g6f.2xlarge",
2483 "g6f.4xlarge",
2484 "g6f.large",
2485 "g6f.xlarge",
2486 "g7.12xlarge",
2487 "g7.24xlarge",
2488 "g7.2xlarge",
2489 "g7.48xlarge",
2490 "g7.4xlarge",
2491 "g7.8xlarge",
2492 "g7e.12xlarge",
2493 "g7e.24xlarge",
2494 "g7e.2xlarge",
2495 "g7e.48xlarge",
2496 "g7e.4xlarge",
2497 "g7e.8xlarge",
2498 "gr6.4xlarge",
2499 "gr6.8xlarge",
2500 "gr6f.4xlarge",
2501 "inf1.24xlarge",
2502 "inf1.2xlarge",
2503 "inf1.6xlarge",
2504 "inf1.xlarge",
2505 "inf2.24xlarge",
2506 "inf2.48xlarge",
2507 "inf2.8xlarge",
2508 "inf2.xlarge",
2509 "p3dn.24xlarge",
2510 "p4d.24xlarge",
2511 "p4de.24xlarge",
2512 "p5.48xlarge",
2513 "p5.4xlarge",
2514 "p5e.48xlarge",
2515 "p5en.48xlarge",
2516 "p6-b200.48xlarge",
2517 "p6-b300.48xlarge",
2518 "trn1.2xlarge",
2519 "trn1.32xlarge",
2520 "trn1n.32xlarge",
2521 "trn2.3xlarge",
2522 "trn2.48xlarge",
2523 ],
2524 "enabled_regions": [],
2525 }
2526 historical_ctx = self.app.node.try_get_context("historical")
2527 historical_config = historical_ctx if isinstance(historical_ctx, dict) else {}
2528 return {**default_config, **historical_config}
2530 def get_capacity_history_enabled(self) -> bool:
2531 """Return whether the historical capacity surface is enabled."""
2532 return bool(self.get_capacity_history_config()["enabled"])
2534 def get_mission_memory_config(self) -> dict[str, Any]:
2535 """Get the mission-memory configuration (recall across mission sessions).
2537 Returns the merged ``mission_memory`` block from cdk.json layered on
2538 top of the defaults below. The feature is ON by default — memory is
2539 cheap (one small PAY_PER_REQUEST item plus one embedding call per
2540 completed mission) and silently missing recall is the worse failure
2541 mode; set ``mission_memory.enabled: false`` to opt out.
2543 Keys:
2544 - enabled: provision the mission-memory table + vector index and
2545 activate best-effort write/retrieval in the engine (default True)
2546 - retention_days: DynamoDB TTL window for memory items (default 365)
2547 - dimensions: embedding vector width (default 1024). ONE-WAY DOOR:
2548 immutable after index creation and must match the configured
2549 ``bedrock.embedding_model_id`` output width.
2550 - distance_function: vector distance metric (default COSINE);
2551 immutable after index creation.
2552 - top_k: similar past missions retrieved into the sampling prompt
2553 (default 3)
2554 """
2555 default_config: dict[str, Any] = {
2556 "enabled": True,
2557 "retention_days": 365,
2558 "dimensions": 1024,
2559 "distance_function": "COSINE",
2560 "top_k": 3,
2561 }
2562 mission_memory_ctx = self.app.node.try_get_context("mission_memory")
2563 mission_memory_config = mission_memory_ctx if isinstance(mission_memory_ctx, dict) else {}
2564 return {**default_config, **mission_memory_config}
2566 def get_mission_memory_enabled(self) -> bool:
2567 """Return whether mission memory is enabled."""
2568 return bool(self.get_mission_memory_config()["enabled"])
2570 def get_vector_store_config(self) -> dict[str, Any]:
2571 """Get the vector-store configuration (global workload RAG corpus).
2573 Returns the merged ``vector_store`` block from cdk.json layered on
2574 top of the defaults below. The feature is OFF by default — a
2575 replicated vector store carries real per-region storage and write
2576 cost, so it is an explicit opt-in like ``aurora_pgvector``; set
2577 ``vector_store.enabled: true`` to provision it.
2579 Keys:
2580 - enabled: provision the global vector-store table + index,
2581 the S3-triggered ingest pipeline, and the regional read wiring
2582 (default False)
2583 - dimensions: embedding vector width (default 1024). ONE-WAY
2584 DOOR: immutable after index creation and must match the
2585 configured ``embedding_model_id`` output width.
2586 - distance_function: vector distance metric (default COSINE);
2587 immutable after index creation.
2588 - embedding_model_id: Bedrock text-embedding model used by the
2589 ingest pipeline and query paths (default
2590 amazon.titan-embed-text-v2:0). Independent of
2591 ``bedrock.embedding_model_id`` on purpose. Changing it means
2592 re-ingesting the corpus: vectors from different models are
2593 not comparable.
2594 - replica_regions: regions to replicate the table into. Empty
2595 (the default) means "follow deployment_regions.regional",
2596 excluding the global region (the primary).
2597 - corpus_prefix: S3 key prefix on the Cluster_Shared_Bucket
2598 watched by the ingest pipeline (default vector-corpus/).
2599 """
2600 default_config: dict[str, Any] = {
2601 "enabled": False,
2602 "dimensions": 1024,
2603 "distance_function": "COSINE",
2604 "embedding_model_id": "amazon.titan-embed-text-v2:0",
2605 "replica_regions": [],
2606 "corpus_prefix": "vector-corpus/",
2607 }
2608 vector_store_ctx = self.app.node.try_get_context("vector_store")
2609 vector_store_config = vector_store_ctx if isinstance(vector_store_ctx, dict) else {}
2610 merged = {**default_config, **vector_store_config}
2611 if self._feature_override_enabled("vector_store"):
2612 merged["enabled"] = True
2613 return merged
2615 def get_vector_store_enabled(self) -> bool:
2616 """Return whether the vector store is enabled (default False)."""
2617 return bool(self.get_vector_store_config()["enabled"])
2619 def get_vector_store_replica_regions(self) -> list[str]:
2620 """Return the effective replica region list for the vector store.
2622 The configured ``replica_regions`` when non-empty, otherwise the
2623 regional deployment list — in both cases with the global region
2624 removed, because the primary table lives there and a global table
2625 cannot replicate into its own region. May legitimately be empty
2626 (single-region deployments get a single-region global table).
2627 """
2628 config = self.get_vector_store_config()
2629 configured = [str(region) for region in config["replica_regions"]]
2630 candidates = configured or self.get_regions()
2631 global_region = self.get_global_region()
2632 return [region for region in candidates if region != global_region]
2634 def get_tags(self) -> dict[str, str]:
2635 """Get common tags from configuration"""
2636 return self.app.node.try_get_context("tags") or {}
2638 def validate_region_availability(self, region: str) -> bool:
2639 """Validate that a region is available in the current AWS account"""
2640 try:
2641 ec2 = boto3.client("ec2", region_name=region)
2642 ec2.describe_regions(RegionNames=[region])
2643 return True
2644 except Exception as e:
2645 logger.debug("Region %s not available: %s", region, e)
2646 return False
2648 def get_available_regions(self) -> list[str]:
2649 """Get list of available AWS regions for the current account"""
2650 try:
2651 ec2 = boto3.client("ec2")
2652 response = ec2.describe_regions()
2653 return [region["RegionName"] for region in response["Regions"]]
2654 except Exception as e:
2655 logger.debug("Failed to list regions, using defaults: %s", e)
2656 return list(self.VALID_REGIONS)