Coverage for gco / config / config_loader.py: 100.00%
664 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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- tags: Common tags applied to all resources
20Usage:
21 config = ConfigLoader(app)
22 regions = config.get_regions()
23 cluster_config = config.get_cluster_config("us-east-1")
24"""
26from __future__ import annotations
28import logging
29import re
30from typing import Any, cast
32import boto3
33from aws_cdk import App
35from gco.inference_proxy_config import (
36 INFERENCE_PROXY_MAX_REPLICAS_DEFAULT,
37 INFERENCE_PROXY_MIN_REPLICAS_DEFAULT,
38 INFERENCE_PROXY_TLS_CPU_REQUEST_MILLICORES_DEFAULT,
39 INFERENCE_PROXY_TLS_CPU_TARGET_UTILIZATION_DEFAULT,
40)
41from gco.manifest_security_policy import validate_manifest_security_policy
42from gco.models import ClusterConfig, ResourceThresholds
43from gco.resource_governance import parse_k8s_quantity
44from gco.stacks.constants import (
45 DEFAULT_MAX_REQUEST_BODY_BYTES,
46 known_cloudformation_regions,
47 validated_deployment_partition,
48 validated_regional_deployment_regions,
49 validated_request_body_limit,
50)
52logger = logging.getLogger(__name__)
54#: ``manifest_processor.autoscaling`` defaults. Off: the API tier is I/O-bound
55#: (Kubernetes round trips), so CPU is a weak saturation signal; an operator who
56#: has measured a CPU-bound API turns it on and the HPA owns the replica count
57#: between ``manifest_processor.replicas`` and ``max_replicas``.
58_MANIFEST_PROCESSOR_AUTOSCALING_DEFAULTS: dict[str, Any] = {
59 "enabled": False,
60 "max_replicas": 6,
61 "cpu_target_utilization_percentage": 70,
62}
64#: The ``manifest-processor`` container's fixed requests in
65#: ``lambda/kubectl-applier-simple/manifests/31-manifest-processor.yaml``;
66#: ``manifest_processor.resource_limits`` may not go below them. Pinned to the
67#: manifest by ``tests/test_config_loader.py``.
68_MANIFEST_PROCESSOR_CONTAINER_REQUESTS: dict[str, str] = {
69 "cpu": "500m",
70 "memory": "1Gi",
71}
73#: Gateway VPC endpoints ``vpc_endpoints.gateway`` may name. Free, route-table
74#: based, same-region only.
75VPC_GATEWAY_ENDPOINT_SERVICES: tuple[str, ...] = ("s3", "dynamodb")
77#: Interface (PrivateLink) endpoints ``vpc_endpoints.interface`` may name,
78#: keyed the way the regional stack maps them onto
79#: ``ec2.InterfaceVpcEndpointAwsService``. Each costs per AZ-hour plus per GB,
80#: so the list is opt-in; together they keep every AWS API call the platform
81#: and its jobs make inside the VPC (the cross-region calls to the global
82#: region's tables and buckets still leave through the NAT gateways).
83VPC_INTERFACE_ENDPOINT_SERVICES: tuple[str, ...] = (
84 "sts",
85 "ecr.api",
86 "ecr.dkr",
87 "logs",
88 "monitoring",
89 "sqs",
90 "ssm",
91 "secretsmanager",
92 "kms",
93 "eks",
94 "elasticfilesystem",
95 "bedrock-runtime",
96)
98_VPC_ENDPOINTS_DEFAULTS: dict[str, list[str]] = {
99 "gateway": ["s3", "dynamodb"],
100 "interface": [],
101}
103#: CDK context key that force-enables optional infrastructure features for one
104#: deploy without touching cdk.json — the infrastructure sibling of the
105#: ``helm_enabled_overrides`` context handled in ``gco/stacks/regional_stack.py``.
106#: Used by validation harnesses (``gco examples validate``) whose preflight
107#: requires a clean worktree.
108FEATURE_OVERRIDE_CONTEXT_KEY = "feature_enabled_overrides"
110#: The cdk.json blocks whose ``enabled`` flag the override may force on. Kept
111#: deliberately narrow: each of these is a self-contained regional feature the
112#: examples exercise (Aurora pgvector, Valkey Serverless, FSx for Lustre).
113FEATURE_OVERRIDE_KEYS = frozenset({"aurora_pgvector", "valkey", "fsx_lustre", "vector_store"})
116def parse_feature_enabled_overrides(raw: object) -> frozenset[str]:
117 """Parse and validate the ``feature_enabled_overrides`` context value.
119 Accepts a comma-separated string (the only shape the CDK CLI can pass with
120 ``--context``) or a list of strings (cdk.json-style). Unknown names raise
121 at synth time with the valid list — identical semantics to
122 ``_parse_helm_enabled_overrides``.
123 """
124 if raw is None:
125 return frozenset()
126 if isinstance(raw, str):
127 names = [part.strip() for part in raw.split(",") if part.strip()]
128 elif isinstance(raw, list) and all(isinstance(part, str) for part in raw):
129 names = [part.strip() for part in raw if part.strip()]
130 else:
131 raise ConfigValidationError(
132 f"{FEATURE_OVERRIDE_CONTEXT_KEY} must be a comma-separated string or string list"
133 )
134 unknown = sorted(set(names) - FEATURE_OVERRIDE_KEYS)
135 if unknown:
136 valid = ", ".join(sorted(FEATURE_OVERRIDE_KEYS))
137 raise ConfigValidationError(
138 f"Unknown {FEATURE_OVERRIDE_CONTEXT_KEY} name(s): {', '.join(unknown)}. Valid: {valid}"
139 )
140 return frozenset(names)
143class ConfigValidationError(Exception):
144 """Raised when configuration validation fails."""
146 pass
149class ConfigLoader:
150 """
151 Loads and validates configuration from CDK context (cdk.json)
152 """
154 # Keep the public class attribute for compatibility. Endpoint metadata
155 # covers every CloudFormation Region known to the installed AWS SDK; this
156 # is not a project-specific allowlist.
157 VALID_REGIONS = known_cloudformation_regions()
159 def __init__(self, app: App):
160 self.app = app
161 self._validate_configuration()
163 def _validate_configuration(self) -> None:
164 """Validate the entire configuration"""
165 # Check if we have any context at all (might be running outside CDK)
166 project_name = self.app.node.try_get_context("project_name")
167 if project_name is None:
168 # Running outside CDK context, skip validation
169 return
171 # Validate required fields exist
172 required_fields = [
173 "project_name",
174 "kubernetes_version",
175 "resource_thresholds",
176 ]
177 for field in required_fields:
178 if not self.app.node.try_get_context(field):
179 raise ConfigValidationError(f"Required configuration field '{field}' is missing")
181 # Validate project_name format before anything consumes it (#139).
182 self._validate_project_name()
184 # Check for deployment_regions
185 deployment_regions = self.app.node.try_get_context("deployment_regions")
186 if not isinstance(deployment_regions, dict) or not deployment_regions:
187 raise ConfigValidationError(
188 "Required configuration field 'deployment_regions' must be a non-empty object"
189 )
191 # Validate regions
192 self._validate_regions()
194 # Validate resource thresholds
195 self._validate_resource_thresholds()
197 # Validate Global Accelerator config
198 self._validate_global_accelerator_config()
200 # Validate deployment-local backend TLS rotation policy
201 self._validate_backend_tls_config()
203 # Validate inference proxy TLS sidecar autoscaling settings
204 self._validate_inference_proxy_config()
206 # Validate ALB config
207 self._validate_alb_config()
209 # Validate manifest processor config
210 self._validate_manifest_processor_config()
212 # Validate API Gateway config
213 self._validate_api_gateway_config()
215 # Validate EKS cluster config
216 self._validate_eks_cluster_config()
218 # Validate analytics environment config (optional block)
219 self._validate_analytics_environment_config()
221 # Validate cluster observability config (optional block)
222 self._validate_cluster_observability_config()
224 # Validate cost monitoring config (optional block)
225 self._validate_cost_monitoring_config()
227 # Validate historical capacity surface config (optional block)
228 self._validate_capacity_history_config()
230 # Validate mission-memory configuration (recall across mission sessions)
231 self._validate_mission_memory_config()
232 # Validate the vector-store configuration (global workload RAG corpus)
233 self._validate_vector_store_config()
235 #: Allowed ``project_name`` format (#139). ``project_name`` is the
236 #: deployment's unique prefix and flows into S3 bucket names, the Cognito
237 #: hosted-UI domain prefix, SSM parameter paths, IAM role names, and
238 #: CloudFormation export names. The tightest of those constraints is S3 /
239 #: Cognito naming (lowercase letters, digits, hyphens; must start with a
240 #: letter), so require: a leading lowercase letter followed by 1–30
241 #: lowercase letters, digits, or hyphens (total length 2–31).
242 PROJECT_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9-]{1,30}$")
244 def _validate_project_name(self) -> None:
245 """Validate ``project_name`` format so misconfigurations fail at synth.
247 ``project_name`` is documented as the deployment's unique identifier
248 and is used as the prefix for nearly every physical resource name. If
249 it contains characters that are illegal in S3 bucket names or Cognito
250 domain prefixes (uppercase, underscores, dots, leading digit, etc.),
251 ``cdk synth`` still succeeds but the deploy fails late with an opaque
252 AWS naming error. Catching it here turns that into an actionable
253 message up front.
254 """
255 project_name = self.app.node.try_get_context("project_name")
256 if not isinstance(project_name, str) or not self.PROJECT_NAME_PATTERN.match(project_name):
257 raise ConfigValidationError(
258 f"Invalid project_name {project_name!r}. It must match "
259 f"{self.PROJECT_NAME_PATTERN.pattern} (start with a lowercase letter, then "
260 "2–31 total characters of lowercase letters, digits, or hyphens). "
261 "project_name is the deployment prefix for S3 buckets, the Cognito "
262 "domain, SSM paths, and CloudFormation exports, so it must be a valid "
263 "lowercase DNS-style label."
264 )
266 def _validate_regions(self) -> None:
267 """Validate region configuration against the shared app/CLI contract."""
268 deployment_regions = self.get_deployment_regions()
269 try:
270 for field in ("global", "api_gateway", "monitoring"):
271 region = deployment_regions[field]
272 if not isinstance(region, str) or region not in self.VALID_REGIONS:
273 raise ValueError(
274 f"Invalid {field} region {region!r}; expected an AWS region with a "
275 "CloudFormation endpoint known to the installed SDK"
276 )
277 regional = validated_regional_deployment_regions(
278 deployment_regions["regional"],
279 known_regions=self.VALID_REGIONS,
280 )
281 validated_deployment_partition(
282 (
283 deployment_regions["global"],
284 deployment_regions["api_gateway"],
285 deployment_regions["monitoring"],
286 *regional,
287 )
288 )
289 except (RuntimeError, ValueError) as exc:
290 raise ConfigValidationError(str(exc)) from exc
292 def _validate_resource_thresholds(self) -> None:
293 """Validate resource threshold configuration"""
294 thresholds_config = self.app.node.try_get_context("resource_thresholds")
296 required_thresholds = ["cpu_threshold", "memory_threshold", "gpu_threshold"]
297 for threshold in required_thresholds:
298 if threshold not in thresholds_config:
299 raise ConfigValidationError(f"Missing threshold configuration: {threshold}")
301 value = thresholds_config[threshold]
302 if not isinstance(value, int) or (value != -1 and not 0 <= value <= 100):
303 raise ConfigValidationError(
304 f"{threshold} must be an integer between 0 and 100 (or -1 to disable), got {value}"
305 )
307 # Validate optional thresholds if present
308 for opt_threshold in [
309 "pending_pods_threshold",
310 "pending_requested_cpu_vcpus",
311 "pending_requested_memory_gb",
312 "pending_requested_gpus",
313 ]:
314 if opt_threshold in thresholds_config:
315 value = thresholds_config[opt_threshold]
316 if not isinstance(value, int) or (value != -1 and value < 0):
317 raise ConfigValidationError(
318 f"{opt_threshold} must be a non-negative integer (or -1 to disable), got {value}"
319 )
321 #: Health-check probe intervals Global Accelerator accepts. The API
322 #: constrains HealthCheckIntervalSeconds to exactly 10 or 30 seconds, so
323 #: any other value must fail at synth instead of at deploy.
324 GLOBAL_ACCELERATOR_HEALTH_CHECK_INTERVALS = frozenset({10, 30})
326 #: Traffic-dial controller modes. ``monitor`` computes and publishes
327 #: per-region dial decisions without mutating Global Accelerator;
328 #: ``enforce`` additionally applies them via UpdateEndpointGroup.
329 TRAFFIC_DIAL_MODES = frozenset({"monitor", "enforce"})
331 def _validate_global_accelerator_config(self) -> None:
332 """Validate the ``global_accelerator`` block in cdk.json.
334 The block is optional; absence means the shipped defaults apply.
335 Validation runs against the *merged* configuration so a partial block
336 is checked together with every default it kept:
338 - ``health_check_interval``: 10 or 30 — the only probe intervals the
339 Global Accelerator API accepts (UpdateEndpointGroup rejects others).
340 - ``health_check_threshold``: integer 1-10 (the API ThresholdCount
341 range).
342 - ``health_check_path``: must start with ``/``.
343 - ``client_affinity``: NONE or SOURCE_IP, case-insensitive.
344 - ``traffic_dial``: see :meth:`_validate_traffic_dial_config`.
346 The legacy ``health_check_grace_period`` and ``health_check_timeout``
347 keys are tolerated and ignored: Global Accelerator endpoint groups
348 have no such settings (the keys were validated but never consumed),
349 so their presence in an existing cdk.json must not fail synthesis.
351 ``name`` is intentionally optional: when omitted it defaults to
352 ``<project_name>-accelerator`` so a second deployment gets a
353 project-scoped name from the single ``project_name`` knob (#139).
354 """
355 ga_config = self.get_global_accelerator_config()
357 interval = ga_config["health_check_interval"]
358 if (
359 not isinstance(interval, int)
360 or isinstance(interval, bool)
361 or interval not in self.GLOBAL_ACCELERATOR_HEALTH_CHECK_INTERVALS
362 ):
363 raise ConfigValidationError(
364 "global_accelerator.health_check_interval must be one of "
365 f"{sorted(self.GLOBAL_ACCELERATOR_HEALTH_CHECK_INTERVALS)} — the only probe "
366 f"intervals the Global Accelerator API accepts — got {interval!r}"
367 )
369 threshold = ga_config["health_check_threshold"]
370 if (
371 not isinstance(threshold, int)
372 or isinstance(threshold, bool)
373 or not 1 <= threshold <= 10
374 ):
375 raise ConfigValidationError(
376 "global_accelerator.health_check_threshold must be an integer between "
377 f"1 and 10, got {threshold!r}"
378 )
380 path = ga_config["health_check_path"]
381 if not isinstance(path, str) or not path.startswith("/"):
382 raise ConfigValidationError("health_check_path must start with '/'")
384 allowed_affinity = {"NONE", "SOURCE_IP"}
385 affinity = ga_config["client_affinity"]
386 if not isinstance(affinity, str) or affinity.upper() not in allowed_affinity:
387 raise ConfigValidationError(
388 f"client_affinity must be one of {sorted(allowed_affinity)}, got {affinity!r}"
389 )
391 self._validate_traffic_dial_config(ga_config["traffic_dial"])
393 def _validate_traffic_dial_config(self, dial_config: Any) -> None:
394 """Validate the merged ``global_accelerator.traffic_dial`` sub-block.
396 Ranges mirror the Global Accelerator API and the controller contract:
398 - ``enabled``: bool (default False — the controller is opt-in).
399 - ``mode``: ``monitor`` or ``enforce``, case-insensitive.
400 - ``interval_minutes`` / ``lookback_minutes``: 1-1440.
401 - ``min_dial_percentage``: 0-100 (TrafficDialPercentage range); the
402 floor a degraded region can be dialed down to.
403 - ``max_step_percentage``: 1-100; the largest change one run applies.
404 - ``full_health_percentage``: 1-100; the healthy fraction (percent)
405 at or above which a region is restored toward 100.
406 """
407 if not isinstance(dial_config, dict):
408 raise ConfigValidationError(
409 "global_accelerator.traffic_dial must be a mapping, got "
410 f"{type(dial_config).__name__}: {dial_config!r}"
411 )
413 enabled = dial_config["enabled"]
414 if not isinstance(enabled, bool):
415 raise ConfigValidationError(
416 "global_accelerator.traffic_dial.enabled must be a bool, got "
417 f"{type(enabled).__name__}: {enabled!r}"
418 )
420 mode = dial_config["mode"]
421 if not isinstance(mode, str) or mode.lower() not in self.TRAFFIC_DIAL_MODES:
422 raise ConfigValidationError(
423 "global_accelerator.traffic_dial.mode must be one of "
424 f"{sorted(self.TRAFFIC_DIAL_MODES)}, got {mode!r}"
425 )
427 int_ranges = (
428 ("interval_minutes", 1, 1_440),
429 ("lookback_minutes", 1, 1_440),
430 ("min_dial_percentage", 0, 100),
431 ("max_step_percentage", 1, 100),
432 ("full_health_percentage", 1, 100),
433 )
434 for field, minimum, maximum in int_ranges:
435 value = dial_config[field]
436 if (
437 not isinstance(value, int)
438 or isinstance(value, bool)
439 or not minimum <= value <= maximum
440 ):
441 raise ConfigValidationError(
442 f"global_accelerator.traffic_dial.{field} must be an integer between "
443 f"{minimum} and {maximum}, got {value!r}"
444 )
446 def _validate_backend_tls_config(self) -> None:
447 """Validate private-root and leaf-certificate lifecycle settings."""
448 config = self.get_backend_tls_config()
449 ranges = {
450 "root_generation": (1, 1_000_000),
451 "root_validity_days": (365, 36_500),
452 "root_rotate_before_days": (30, 3_650),
453 "root_activation_delay_hours": (1, 168),
454 "root_overlap_days": (2, 365),
455 "leaf_validity_days": (2, 397),
456 "leaf_rotate_before_days": (1, 90),
457 "rotation_schedule_hours": (1, 24),
458 "trust_cache_ttl_seconds": (1, 3_600),
459 "trust_cache_max_stale_seconds": (1, 86_400),
460 }
461 for field, (minimum, maximum) in ranges.items():
462 value = config.get(field)
463 if type(value) is not int or not minimum <= value <= maximum:
464 raise ConfigValidationError(
465 f"backend_tls.{field} must be an integer between "
466 f"{minimum} and {maximum}, got {value!r}"
467 )
469 if config["root_rotate_before_days"] >= config["root_validity_days"]:
470 raise ConfigValidationError(
471 "backend_tls.root_rotate_before_days must be less than root_validity_days"
472 )
473 if config["leaf_rotate_before_days"] >= config["leaf_validity_days"]:
474 raise ConfigValidationError(
475 "backend_tls.leaf_rotate_before_days must be less than leaf_validity_days"
476 )
477 if config["root_validity_days"] <= config["leaf_validity_days"]:
478 raise ConfigValidationError(
479 "backend_tls.root_validity_days must exceed leaf_validity_days"
480 )
481 if config["root_overlap_days"] <= config["leaf_validity_days"]:
482 raise ConfigValidationError(
483 "backend_tls.root_overlap_days must exceed leaf_validity_days so old leaves "
484 "remain trusted throughout root rollover"
485 )
486 if config["trust_cache_max_stale_seconds"] < config["trust_cache_ttl_seconds"]:
487 raise ConfigValidationError(
488 "backend_tls.trust_cache_max_stale_seconds must be at least trust_cache_ttl_seconds"
489 )
490 if config["root_activation_delay_hours"] * 3_600 <= config["trust_cache_max_stale_seconds"]:
491 raise ConfigValidationError(
492 "backend_tls.root_activation_delay_hours must exceed the maximum stale trust "
493 "cache window so every proxy can observe a pending root before leaf rollover"
494 )
496 def _validate_inference_proxy_config(self) -> None:
497 """Validate the inference proxy TLS CPU request, HPA target and replica bounds."""
498 config = self.get_inference_proxy_config()
499 ranges = {
500 "tls_proxy_cpu_request_millicores": (1, 250),
501 "tls_proxy_cpu_target_utilization_percentage": (1, 100),
502 "min_replicas": (1, 50),
503 "max_replicas": (1, 100),
504 }
505 for field, (minimum, maximum) in ranges.items():
506 value = config[field]
507 if type(value) is not int or not minimum <= value <= maximum:
508 raise ConfigValidationError(
509 f"inference_proxy.{field} must be an integer between "
510 f"{minimum} and {maximum}, got {value!r}"
511 )
512 if config["max_replicas"] < config["min_replicas"]:
513 raise ConfigValidationError(
514 "inference_proxy.max_replicas must be at least inference_proxy.min_replicas, "
515 f"got {config['max_replicas']} < {config['min_replicas']}"
516 )
518 def _validate_alb_config(self) -> None:
519 """Validate ALB configuration"""
520 alb_config = self.app.node.try_get_context("alb_config")
521 if not alb_config:
522 raise ConfigValidationError("alb_config configuration is required")
524 required_fields = [
525 "health_check_interval",
526 "health_check_timeout",
527 "healthy_threshold",
528 "unhealthy_threshold",
529 ]
530 for field in required_fields:
531 if field not in alb_config:
532 raise ConfigValidationError(f"Missing alb_config configuration: {field}")
534 value = alb_config[field]
535 if not isinstance(value, int) or value <= 0:
536 raise ConfigValidationError(f"{field} must be a positive integer, got {value}")
538 def _validate_manifest_processor_config(self) -> None:
539 """Validate manifest processor configuration.
541 The manifest processor section in cdk.json holds service-specific
542 settings only. The shared validation policy (allowed_namespaces,
543 resource_quotas, trusted_registries, trusted_dockerhub_orgs,
544 manifest_security_policy, allowed_kinds) lives under
545 ``job_validation_policy`` because the queue_processor reads the
546 same values.
547 """
548 mp_config = self.app.node.try_get_context("manifest_processor")
549 if not mp_config:
550 raise ConfigValidationError("manifest_processor configuration is required")
552 required_fields = [
553 "image",
554 "replicas",
555 "resource_limits",
556 ]
557 for field in required_fields:
558 if field not in mp_config:
559 raise ConfigValidationError(f"Missing manifest_processor configuration: {field}")
561 # Validate replicas
562 if not isinstance(mp_config["replicas"], int) or mp_config["replicas"] <= 0:
563 raise ConfigValidationError("manifest_processor replicas must be a positive integer")
565 try:
566 validated_request_body_limit(
567 mp_config.get("max_request_body_bytes", DEFAULT_MAX_REQUEST_BODY_BYTES)
568 )
569 except ValueError as exc:
570 raise ConfigValidationError(f"manifest_processor.{exc}") from exc
572 validation_enabled = mp_config.get("validation_enabled", True)
573 if type(validation_enabled) is not bool:
574 raise ConfigValidationError("manifest_processor.validation_enabled must be a boolean")
576 # Validate the shared policy section separately so a misconfigured
577 # policy block surfaces a clear error pointing at the right key.
578 policy = self.app.node.try_get_context("job_validation_policy")
579 if policy is None:
580 raise ConfigValidationError(
581 "job_validation_policy configuration is required (shared between "
582 "manifest_processor and queue_processor)"
583 )
584 if not isinstance(policy, dict):
585 raise ConfigValidationError("job_validation_policy must be an object")
586 for policy_field in ("allowed_namespaces", "resource_quotas"):
587 if policy_field not in policy:
588 raise ConfigValidationError(
589 f"Missing job_validation_policy configuration: {policy_field}"
590 )
592 try:
593 validate_manifest_security_policy(policy.get("manifest_security_policy", {}))
594 except ValueError as exc:
595 raise ConfigValidationError(f"job_validation_policy.{exc}") from exc
597 require_toleration = policy.get("require_accelerator_toleration", True)
598 if type(require_toleration) is not bool:
599 raise ConfigValidationError(
600 "job_validation_policy.require_accelerator_toleration must be a boolean"
601 )
603 # Validate resource limits. They render verbatim into the Deployment's
604 # container limits, so they must be non-empty Kubernetes quantities no
605 # smaller than the requests fixed in 31-manifest-processor.yaml —
606 # Kubernetes rejects limit < request, and catching that here fails the
607 # synth instead of the applier halfway through a deploy.
608 resource_limits = mp_config["resource_limits"]
609 if "cpu" not in resource_limits or "memory" not in resource_limits:
610 raise ConfigValidationError(
611 "manifest_processor resource_limits must contain 'cpu' and 'memory'"
612 )
613 for resource, request in _MANIFEST_PROCESSOR_CONTAINER_REQUESTS.items():
614 quantity = resource_limits[resource]
615 try:
616 if not isinstance(quantity, str):
617 raise ValueError("must be a string")
618 parsed = parse_k8s_quantity(quantity)
619 except ValueError as exc:
620 raise ConfigValidationError(
621 f"manifest_processor.resource_limits.{resource} must be a Kubernetes "
622 f"quantity string such as '1000m' or '2Gi', got {quantity!r} ({exc})"
623 ) from exc
624 if parsed < parse_k8s_quantity(request):
625 raise ConfigValidationError(
626 f"manifest_processor.resource_limits.{resource} must be at least the "
627 f"container request of {request!r} "
628 f"(31-manifest-processor.yaml), got {quantity!r}"
629 )
631 # Validate the optional CPU autoscaler. Off by default: the API tier is
632 # I/O-bound, so CPU is a weak saturation signal; operators who have
633 # measured otherwise turn it on here and the HPA takes over the count.
634 autoscaling = mp_config.get("autoscaling")
635 if autoscaling is None:
636 autoscaling = {}
637 if not isinstance(autoscaling, dict):
638 raise ConfigValidationError("manifest_processor.autoscaling must be an object")
639 unknown_autoscaling = sorted(
640 str(key) for key in autoscaling if key not in _MANIFEST_PROCESSOR_AUTOSCALING_DEFAULTS
641 )
642 if unknown_autoscaling:
643 raise ConfigValidationError(
644 "manifest_processor.autoscaling contains unknown key(s): "
645 + ", ".join(unknown_autoscaling)
646 + "; allowed keys: "
647 + ", ".join(sorted(_MANIFEST_PROCESSOR_AUTOSCALING_DEFAULTS))
648 )
649 merged_autoscaling = {**_MANIFEST_PROCESSOR_AUTOSCALING_DEFAULTS, **autoscaling}
650 if type(merged_autoscaling["enabled"]) is not bool:
651 raise ConfigValidationError("manifest_processor.autoscaling.enabled must be a boolean")
652 for key, minimum, maximum in (
653 ("max_replicas", 1, 100),
654 ("cpu_target_utilization_percentage", 1, 100),
655 ):
656 value = merged_autoscaling[key]
657 if type(value) is not int or not minimum <= value <= maximum:
658 raise ConfigValidationError(
659 f"manifest_processor.autoscaling.{key} must be an integer between "
660 f"{minimum} and {maximum}, got {value!r}"
661 )
662 if merged_autoscaling["max_replicas"] < mp_config["replicas"]:
663 raise ConfigValidationError(
664 "manifest_processor.autoscaling.max_replicas must be at least "
665 f"manifest_processor.replicas, got {merged_autoscaling['max_replicas']} "
666 f"< {mp_config['replicas']}"
667 )
669 # Validate allowed namespaces (lives under job_validation_policy).
670 if not isinstance(policy["allowed_namespaces"], list):
671 raise ConfigValidationError("job_validation_policy.allowed_namespaces must be a list")
673 def _validate_api_gateway_config(self) -> None:
674 """Validate API Gateway configuration"""
675 api_gw_config = self.app.node.try_get_context("api_gateway")
676 if not api_gw_config:
677 raise ConfigValidationError("api_gateway configuration is required")
679 required_fields = [
680 "throttle_rate_limit",
681 "throttle_burst_limit",
682 "log_level",
683 "metrics_enabled",
684 "tracing_enabled",
685 ]
686 for field in required_fields:
687 if field not in api_gw_config:
688 raise ConfigValidationError(f"Missing api_gateway configuration: {field}")
690 # Validate throttle limits
691 throttle_rate = api_gw_config["throttle_rate_limit"]
692 throttle_burst = api_gw_config["throttle_burst_limit"]
694 if not isinstance(throttle_rate, int) or throttle_rate <= 0:
695 raise ConfigValidationError(
696 f"throttle_rate_limit must be a positive integer, got {throttle_rate}"
697 )
699 if not isinstance(throttle_burst, int) or throttle_burst <= 0:
700 raise ConfigValidationError(
701 f"throttle_burst_limit must be a positive integer, got {throttle_burst}"
702 )
704 if throttle_burst < throttle_rate:
705 raise ConfigValidationError(
706 "throttle_burst_limit should be greater than or equal to throttle_rate_limit"
707 )
709 # Validate log level
710 valid_log_levels = ["OFF", "ERROR", "INFO"]
711 log_level = api_gw_config["log_level"]
712 if log_level not in valid_log_levels:
713 raise ConfigValidationError(
714 f"log_level must be one of {valid_log_levels}, got {log_level}"
715 )
717 # Validate boolean flags
718 if not isinstance(api_gw_config["metrics_enabled"], bool):
719 raise ConfigValidationError("metrics_enabled must be a boolean")
721 if not isinstance(api_gw_config["tracing_enabled"], bool):
722 raise ConfigValidationError("tracing_enabled must be a boolean")
724 if "regional_api_enabled" in api_gw_config and not isinstance(
725 api_gw_config["regional_api_enabled"], bool
726 ):
727 raise ConfigValidationError("regional_api_enabled must be a boolean")
729 def _validate_eks_cluster_config(self) -> None:
730 """Validate EKS cluster configuration"""
731 eks_config = self.app.node.try_get_context("eks_cluster") or {}
733 # Validate endpoint_access if present
734 if "endpoint_access" in eks_config:
735 valid_access_modes = ["PRIVATE", "PUBLIC_AND_PRIVATE"]
736 if eks_config["endpoint_access"] not in valid_access_modes:
737 raise ConfigValidationError(
738 f"endpoint_access must be one of {valid_access_modes}, "
739 f"got {eks_config['endpoint_access']}"
740 )
741 # The NetworkPolicy enforcement switch renders straight into a
742 # kube-system ConfigMap value, so only a literal JSON boolean is
743 # accepted — a string "false" would be applied verbatim and enable
744 # nothing while reading as disabled.
745 if (
746 "network_policy_enforcement" in eks_config
747 and type(eks_config["network_policy_enforcement"]) is not bool
748 ):
749 raise ConfigValidationError(
750 "eks_cluster.network_policy_enforcement must be a boolean, got "
751 f"{eks_config['network_policy_enforcement']!r}"
752 )
753 self._validate_vpc_endpoints_config()
755 def _validate_vpc_endpoints_config(self) -> None:
756 """Validate the optional ``vpc_endpoints`` block.
758 ``gateway`` lists the free route-table endpoints (``s3``,
759 ``dynamodb``); ``interface`` lists PrivateLink endpoints by the
760 service key in :data:`VPC_INTERFACE_ENDPOINT_SERVICES`. Interface
761 endpoints bill per AZ-hour, so the list is explicit and unknown names
762 fail at synth instead of silently creating nothing.
763 """
764 raw = self.app.node.try_get_context("vpc_endpoints")
765 if raw is None:
766 return
767 if not isinstance(raw, dict):
768 raise ConfigValidationError("vpc_endpoints must be an object")
769 unknown = sorted(str(key) for key in raw if key not in _VPC_ENDPOINTS_DEFAULTS)
770 if unknown:
771 raise ConfigValidationError(
772 "vpc_endpoints contains unknown key(s): "
773 + ", ".join(unknown)
774 + "; allowed keys: "
775 + ", ".join(sorted(_VPC_ENDPOINTS_DEFAULTS))
776 )
777 for key, allowed in (
778 ("gateway", VPC_GATEWAY_ENDPOINT_SERVICES),
779 ("interface", VPC_INTERFACE_ENDPOINT_SERVICES),
780 ):
781 if key not in raw:
782 continue
783 value = raw[key]
784 if not isinstance(value, list) or not all(isinstance(item, str) for item in value):
785 raise ConfigValidationError(f"vpc_endpoints.{key} must be a list of strings")
786 unsupported = sorted(set(value) - set(allowed))
787 if unsupported:
788 raise ConfigValidationError(
789 f"vpc_endpoints.{key} contains unsupported service(s): "
790 + ", ".join(unsupported)
791 + "; supported: "
792 + ", ".join(sorted(allowed))
793 )
794 if len(set(value)) != len(value):
795 raise ConfigValidationError(f"vpc_endpoints.{key} lists a service twice")
797 def get_vpc_endpoints_config(self) -> dict[str, list[str]]:
798 """Return the VPC endpoint selection with defaults merged in.
800 Defaults: the two free gateway endpoints (S3 carries the platform's
801 largest data path — models, datasets, checkpoints, MLflow artifacts,
802 cost reports — off the NAT gateways' per-GB metering; DynamoDB serves
803 single-region topologies) and no interface endpoints.
804 """
805 configured = self.app.node.try_get_context("vpc_endpoints") or {}
806 return {
807 "gateway": list(configured.get("gateway", _VPC_ENDPOINTS_DEFAULTS["gateway"])),
808 "interface": list(configured.get("interface", _VPC_ENDPOINTS_DEFAULTS["interface"])),
809 }
811 def _validate_analytics_environment_config(self) -> None:
812 """Validate the optional analytics_environment block in cdk.json.
814 The block is entirely optional; absence means the feature is disabled
815 and no validation is needed. When present, we validate:
817 - ``enabled``: must be a bool if present (defaults to False via merge).
818 - ``hyperpod.enabled``: must be a bool if present (defaults to False).
819 - ``cognito.removal_policy`` and ``efs.removal_policy``: must be the
820 literal strings ``"destroy"`` or ``"retain"`` (case sensitive — they
821 are passed verbatim to CDK's ``RemovalPolicy`` lookup by the
822 consumer).
823 """
824 analytics_ctx = self.app.node.try_get_context("analytics_environment")
825 if not isinstance(analytics_ctx, dict):
826 # Block is absent or malformed — defaults apply, nothing to validate.
827 return
829 # Top-level `enabled` must be a bool if provided.
830 if "enabled" in analytics_ctx and not isinstance(analytics_ctx["enabled"], bool):
831 raise ConfigValidationError(
832 f"analytics_environment.enabled must be a bool, got "
833 f"{type(analytics_ctx['enabled']).__name__}: {analytics_ctx['enabled']!r}"
834 )
836 # `hyperpod.enabled` must be a bool if the sub-block is a dict and
837 # carries the key.
838 hyperpod_ctx = analytics_ctx.get("hyperpod")
839 if (
840 isinstance(hyperpod_ctx, dict)
841 and "enabled" in hyperpod_ctx
842 and not isinstance(hyperpod_ctx["enabled"], bool)
843 ):
844 raise ConfigValidationError(
845 f"analytics_environment.hyperpod.enabled must be a bool, got "
846 f"{type(hyperpod_ctx['enabled']).__name__}: {hyperpod_ctx['enabled']!r}"
847 )
849 # `canvas.enabled` must be a bool if the sub-block is a dict and
850 # carries the key. Mirrors the hyperpod validation above.
851 canvas_ctx = analytics_ctx.get("canvas")
852 if (
853 isinstance(canvas_ctx, dict)
854 and "enabled" in canvas_ctx
855 and not isinstance(canvas_ctx["enabled"], bool)
856 ):
857 raise ConfigValidationError(
858 f"analytics_environment.canvas.enabled must be a bool, got "
859 f"{type(canvas_ctx['enabled']).__name__}: {canvas_ctx['enabled']!r}"
860 )
862 valid_removal_policies = {"destroy", "retain"}
864 for sub_block in ("cognito", "efs"):
865 sub_ctx = analytics_ctx.get(sub_block)
866 if not isinstance(sub_ctx, dict):
867 continue
868 if "removal_policy" not in sub_ctx:
869 continue
870 removal_policy = sub_ctx["removal_policy"]
871 if removal_policy not in valid_removal_policies:
872 raise ConfigValidationError(
873 f"analytics_environment.{sub_block}.removal_policy must be one of "
874 f"{sorted(valid_removal_policies)}, got {removal_policy!r}"
875 )
877 def _validate_cluster_observability_config(self) -> None:
878 """Validate the optional cluster_observability block in cdk.json.
880 The block is entirely optional; absence means the on-by-default
881 defaults apply and nothing needs validating. When present, we check:
883 - ``enabled``: must be a bool if present (defaults to True via merge —
884 in-cluster observability is on unless explicitly disabled).
885 - ``grafana``/``prometheus``/``alertmanager`` sub-block ``persistence_size``
886 and ``prometheus.retention``: must be non-empty strings if present
887 (they are passed verbatim to Helm chart values as Kubernetes
888 quantity / duration strings).
889 - ``alertmanager.enabled``: must be a bool if present.
890 """
891 obs_ctx = self.app.node.try_get_context("cluster_observability")
892 if not isinstance(obs_ctx, dict):
893 # Block is absent or malformed — defaults apply, nothing to validate.
894 return
896 if "enabled" in obs_ctx and not isinstance(obs_ctx["enabled"], bool):
897 raise ConfigValidationError(
898 f"cluster_observability.enabled must be a bool, got "
899 f"{type(obs_ctx['enabled']).__name__}: {obs_ctx['enabled']!r}"
900 )
902 # Non-empty-string checks for the size / retention knobs.
903 string_fields = (
904 ("grafana", "persistence_size"),
905 ("prometheus", "persistence_size"),
906 ("prometheus", "retention"),
907 ("alertmanager", "persistence_size"),
908 )
909 for sub_block, field in string_fields:
910 sub_ctx = obs_ctx.get(sub_block)
911 if not isinstance(sub_ctx, dict) or field not in sub_ctx:
912 continue
913 value = sub_ctx[field]
914 if not isinstance(value, str) or not value.strip():
915 raise ConfigValidationError(
916 f"cluster_observability.{sub_block}.{field} must be a non-empty "
917 f"string, got {value!r}"
918 )
920 # `grafana.admin_password_rotation_schedule`: the cron for the in-cluster
921 # CronJob that rotates the Grafana admin password. Validate the 5-field
922 # cron shape so a typo fails at synth rather than yielding an
923 # un-schedulable CronJob in every region.
924 grafana_ctx = obs_ctx.get("grafana")
925 if isinstance(grafana_ctx, dict) and "admin_password_rotation_schedule" in grafana_ctx:
926 schedule = grafana_ctx["admin_password_rotation_schedule"]
927 if not isinstance(schedule, str) or len(schedule.split()) != 5:
928 raise ConfigValidationError(
929 "cluster_observability.grafana.admin_password_rotation_schedule must be "
930 f"a 5-field cron expression string, got {schedule!r}"
931 )
933 # `alertmanager.enabled` must be a bool if the sub-block carries it.
934 alertmanager_ctx = obs_ctx.get("alertmanager")
935 if (
936 isinstance(alertmanager_ctx, dict)
937 and "enabled" in alertmanager_ctx
938 and not isinstance(alertmanager_ctx["enabled"], bool)
939 ):
940 raise ConfigValidationError(
941 f"cluster_observability.alertmanager.enabled must be a bool, got "
942 f"{type(alertmanager_ctx['enabled']).__name__}: "
943 f"{alertmanager_ctx['enabled']!r}"
944 )
946 def _validate_cost_monitoring_config(self) -> None:
947 """Validate the optional ``cost_monitoring`` block in cdk.json.
949 The block is entirely optional; absence means the on-by-default
950 defaults apply and nothing needs validating. When present, we check:
952 - ``enabled``: must be a bool if present (defaults to True via merge).
953 - ``reports.interval_minutes``: positive int between 5 and 1440 if
954 present (the cost-monitor service's scheduled report cadence).
955 - ``reports.retention_days`` /
956 ``reports.transition_to_infrequent_access_days`` /
957 ``athena.query_results_retention_days``: positive ints if present.
958 - The IA transition must happen strictly before expiration, otherwise
959 the S3 lifecycle configuration is rejected at deploy time — fail at
960 synth instead.
962 There is deliberately no cross-toggle error against
963 ``cluster_observability``: cost monitoring's *effective* enablement is
964 the conjunction of both toggles (see
965 :meth:`get_cost_monitoring_enabled`), so disabling observability
966 simply switches the cost pipeline off with it — ``gco monitoring
967 disable`` must not break synthesis.
968 """
969 cost_ctx = self.app.node.try_get_context("cost_monitoring")
970 if not isinstance(cost_ctx, dict):
971 # Block absent or malformed — defaults apply, nothing to validate.
972 return
974 if "enabled" in cost_ctx and not isinstance(cost_ctx["enabled"], bool):
975 raise ConfigValidationError(
976 f"cost_monitoring.enabled must be a bool, got "
977 f"{type(cost_ctx['enabled']).__name__}: {cost_ctx['enabled']!r}"
978 )
980 int_fields = (
981 ("reports", "interval_minutes", 5, 1_440),
982 ("reports", "retention_days", 1, 3_650),
983 ("reports", "transition_to_infrequent_access_days", 30, 3_650),
984 ("athena", "query_results_retention_days", 1, 3_650),
985 )
986 for sub_block, field, minimum, maximum in int_fields:
987 sub_ctx = cost_ctx.get(sub_block)
988 if not isinstance(sub_ctx, dict) or field not in sub_ctx:
989 continue
990 value = sub_ctx[field]
991 if (
992 not isinstance(value, int)
993 or isinstance(value, bool)
994 or not minimum <= value <= maximum
995 ):
996 raise ConfigValidationError(
997 f"cost_monitoring.{sub_block}.{field} must be an integer between "
998 f"{minimum} and {maximum}, got {value!r}"
999 )
1001 merged = self.get_cost_monitoring_config()
1002 reports = merged["reports"]
1003 if reports["transition_to_infrequent_access_days"] >= reports["retention_days"]:
1004 raise ConfigValidationError(
1005 "cost_monitoring.reports.transition_to_infrequent_access_days "
1006 f"({reports['transition_to_infrequent_access_days']}) must be smaller than "
1007 f"cost_monitoring.reports.retention_days ({reports['retention_days']}); "
1008 "S3 rejects lifecycle rules that transition on or after expiration."
1009 )
1011 def _validate_capacity_history_config(self) -> None:
1012 """Validate the optional ``historical`` block in cdk.json.
1014 The block is entirely optional; absence means the historical capacity
1015 surface is disabled and no validation is needed. When present, types
1016 are validated so a typo fails fast at synth time:
1018 - ``enabled``: bool if present.
1019 - ``retention_days`` / ``poll_interval_minutes``: positive ints if present.
1020 - ``watch_instance_types`` / ``enabled_regions``: lists of strings if present.
1021 - every region in ``enabled_regions`` must be a known AWS region.
1022 - ``spot_score_target_capacities``: non-empty list of positive
1023 integers (booleans rejected), every value a member of the supported
1024 set exported by ``cli/capacity/history.py`` — metric fields are
1025 statically named, so an unsupported capacity has nowhere to land.
1026 """
1027 historical_ctx = self.app.node.try_get_context("historical")
1028 if not isinstance(historical_ctx, dict):
1029 return
1031 if "enabled" in historical_ctx and not isinstance(historical_ctx["enabled"], bool):
1032 raise ConfigValidationError(
1033 f"historical.enabled must be a bool, got "
1034 f"{type(historical_ctx['enabled']).__name__}: {historical_ctx['enabled']!r}"
1035 )
1037 for int_field in (
1038 "retention_days",
1039 "poll_interval_minutes",
1040 "capacity_block_duration_hours",
1041 ):
1042 if int_field not in historical_ctx:
1043 continue
1044 value = historical_ctx[int_field]
1045 if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
1046 raise ConfigValidationError(
1047 f"historical.{int_field} must be a positive integer, got {value!r}"
1048 )
1050 # The long-block probe duration may be 0 to disable the long probe
1051 # entirely, so it is validated as non-negative rather than positive.
1052 if "capacity_block_long_duration_hours" in historical_ctx:
1053 long_value = historical_ctx["capacity_block_long_duration_hours"]
1054 if not isinstance(long_value, int) or isinstance(long_value, bool) or long_value < 0:
1055 raise ConfigValidationError(
1056 "historical.capacity_block_long_duration_hours must be a non-negative "
1057 f"integer (0 disables the long probe), got {long_value!r}"
1058 )
1060 for list_field in ("watch_instance_types", "enabled_regions"):
1061 if list_field not in historical_ctx:
1062 continue
1063 value = historical_ctx[list_field]
1064 if not isinstance(value, list) or not all(isinstance(v, str) for v in value):
1065 raise ConfigValidationError(
1066 f"historical.{list_field} must be a list of strings, got {value!r}"
1067 )
1069 for region in historical_ctx.get("enabled_regions", []) or []:
1070 if region not in self.VALID_REGIONS:
1071 raise ConfigValidationError(
1072 f"historical.enabled_regions contains invalid region '{region}'. "
1073 f"Valid regions: {sorted(self.VALID_REGIONS)}"
1074 )
1076 if "spot_score_target_capacities" in historical_ctx:
1077 # Function-local import: cli/capacity/history.py owns the supported
1078 # set and the capacity->field naming rule, and imports nothing from
1079 # gco, so validation and storage cannot drift apart.
1080 from cli.capacity.history import SUPPORTED_SPOT_SCORE_TARGET_CAPACITIES
1082 capacities = historical_ctx["spot_score_target_capacities"]
1083 if (
1084 not isinstance(capacities, list)
1085 or not capacities
1086 or not all(
1087 isinstance(value, int) and not isinstance(value, bool) and value > 0
1088 for value in capacities
1089 )
1090 ):
1091 raise ConfigValidationError(
1092 "historical.spot_score_target_capacities must be a non-empty list "
1093 f"of positive integers, got {capacities!r}"
1094 )
1095 for value in capacities:
1096 if value not in SUPPORTED_SPOT_SCORE_TARGET_CAPACITIES:
1097 raise ConfigValidationError(
1098 "historical.spot_score_target_capacities contains unsupported "
1099 f"target capacity {value!r}. Supported target capacities: "
1100 f"{list(SUPPORTED_SPOT_SCORE_TARGET_CAPACITIES)} (each needs a "
1101 "statically declared metric field; see cli/capacity/history.py)"
1102 )
1104 #: Distance functions the DynamoDB vector-index API accepts. The choice is
1105 #: immutable after index creation, so a typo must fail at synth time.
1106 MISSION_MEMORY_DISTANCE_FUNCTIONS = frozenset({"COSINE", "DOT_PRODUCT", "EUCLIDEAN"})
1108 #: DynamoDB vector-index maximum dimensionality (service quota).
1109 MISSION_MEMORY_MAX_DIMENSIONS = 4096
1111 def _validate_mission_memory_config(self) -> None:
1112 """Validate the ``mission_memory`` block in cdk.json.
1114 The block is optional; absence means the shipped defaults apply
1115 (feature on). When present, types are validated so a typo fails fast
1116 at synth time — especially the one-way-door fields (``dimensions``,
1117 ``distance_function``) that cannot be corrected after the vector
1118 index exists:
1120 - ``enabled``: bool if present.
1121 - ``retention_days`` / ``top_k``: positive ints if present.
1122 - ``dimensions``: positive int <= 4096 if present.
1123 - ``distance_function``: one of COSINE / DOT_PRODUCT / EUCLIDEAN.
1124 """
1125 mission_memory_ctx = self.app.node.try_get_context("mission_memory")
1126 if not isinstance(mission_memory_ctx, dict):
1127 return
1129 if "enabled" in mission_memory_ctx and not isinstance(mission_memory_ctx["enabled"], bool):
1130 raise ConfigValidationError(
1131 f"mission_memory.enabled must be a bool, got "
1132 f"{type(mission_memory_ctx['enabled']).__name__}: "
1133 f"{mission_memory_ctx['enabled']!r}"
1134 )
1136 for int_field in ("retention_days", "top_k"):
1137 if int_field not in mission_memory_ctx:
1138 continue
1139 value = mission_memory_ctx[int_field]
1140 if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
1141 raise ConfigValidationError(
1142 f"mission_memory.{int_field} must be a positive integer, got {value!r}"
1143 )
1145 if "dimensions" in mission_memory_ctx:
1146 dimensions = mission_memory_ctx["dimensions"]
1147 if (
1148 not isinstance(dimensions, int)
1149 or isinstance(dimensions, bool)
1150 or dimensions <= 0
1151 or dimensions > self.MISSION_MEMORY_MAX_DIMENSIONS
1152 ):
1153 raise ConfigValidationError(
1154 "mission_memory.dimensions must be a positive integer <= "
1155 f"{self.MISSION_MEMORY_MAX_DIMENSIONS} (DynamoDB vector-index "
1156 f"limit), got {dimensions!r}. This is a one-way door: it is "
1157 "immutable after index creation and must match the "
1158 "bedrock.embedding_model_id output width."
1159 )
1161 if "distance_function" in mission_memory_ctx:
1162 distance = mission_memory_ctx["distance_function"]
1163 if (
1164 not isinstance(distance, str)
1165 or distance not in self.MISSION_MEMORY_DISTANCE_FUNCTIONS
1166 ):
1167 valid = ", ".join(sorted(self.MISSION_MEMORY_DISTANCE_FUNCTIONS))
1168 raise ConfigValidationError(
1169 f"mission_memory.distance_function must be one of {valid}, got "
1170 f"{distance!r}. This is immutable after index creation."
1171 )
1173 def _validate_vector_store_config(self) -> None:
1174 """Validate the optional ``vector_store`` block in cdk.json.
1176 The block is optional; absence means the feature stays off. When
1177 present, types are validated so a typo fails fast at synth time —
1178 especially the one-way-door fields (``dimensions``,
1179 ``distance_function``) that cannot be corrected after the vector
1180 index exists. The distance-function set and dimension ceiling reuse
1181 the mission-memory constants because both features target the same
1182 DynamoDB vector-index API limits:
1184 - ``enabled``: bool if present.
1185 - ``dimensions``: positive int <= 4096 if present.
1186 - ``distance_function``: one of COSINE / DOT_PRODUCT / EUCLIDEAN.
1187 - ``embedding_model_id``: non-empty string. Deliberately independent
1188 of ``bedrock.embedding_model_id`` (mission memory) — the two
1189 corpora may use different models.
1190 - ``replica_regions``: list of known regions, no duplicates, and
1191 never the global region (that is the table's primary).
1192 - ``corpus_prefix``: non-empty S3 key prefix ending in ``/``.
1193 """
1194 vector_store_ctx = self.app.node.try_get_context("vector_store")
1195 if not isinstance(vector_store_ctx, dict):
1196 return
1198 if "enabled" in vector_store_ctx and not isinstance(vector_store_ctx["enabled"], bool):
1199 raise ConfigValidationError(
1200 f"vector_store.enabled must be a bool, got "
1201 f"{type(vector_store_ctx['enabled']).__name__}: "
1202 f"{vector_store_ctx['enabled']!r}"
1203 )
1204 if "dimensions" in vector_store_ctx:
1205 dimensions = vector_store_ctx["dimensions"]
1206 if (
1207 not isinstance(dimensions, int)
1208 or isinstance(dimensions, bool)
1209 or dimensions <= 0
1210 or dimensions > self.MISSION_MEMORY_MAX_DIMENSIONS
1211 ):
1212 raise ConfigValidationError(
1213 "vector_store.dimensions must be a positive integer <= "
1214 f"{self.MISSION_MEMORY_MAX_DIMENSIONS} (DynamoDB vector-index "
1215 f"limit), got {dimensions!r}. This is a one-way door: it is "
1216 "immutable after index creation and must match the "
1217 "vector_store.embedding_model_id output width."
1218 )
1219 if "distance_function" in vector_store_ctx:
1220 distance = vector_store_ctx["distance_function"]
1221 if (
1222 not isinstance(distance, str)
1223 or distance not in self.MISSION_MEMORY_DISTANCE_FUNCTIONS
1224 ):
1225 valid = ", ".join(sorted(self.MISSION_MEMORY_DISTANCE_FUNCTIONS))
1226 raise ConfigValidationError(
1227 f"vector_store.distance_function must be one of {valid}, got "
1228 f"{distance!r}. This is immutable after index creation."
1229 )
1230 if "embedding_model_id" in vector_store_ctx:
1231 model_id = vector_store_ctx["embedding_model_id"]
1232 if not isinstance(model_id, str) or not model_id.strip():
1233 raise ConfigValidationError(
1234 f"vector_store.embedding_model_id must be a non-empty string, got {model_id!r}"
1235 )
1236 if "replica_regions" in vector_store_ctx:
1237 replica_regions = vector_store_ctx["replica_regions"]
1238 if not isinstance(replica_regions, list) or not all(
1239 isinstance(region, str) for region in replica_regions
1240 ):
1241 raise ConfigValidationError(
1242 f"vector_store.replica_regions must be a list of region strings, "
1243 f"got {replica_regions!r}"
1244 )
1245 if len(replica_regions) != len(set(replica_regions)):
1246 raise ConfigValidationError(
1247 f"vector_store.replica_regions contains duplicates: {replica_regions!r}"
1248 )
1249 global_region = self.get_global_region()
1250 for region in replica_regions:
1251 if region not in self.VALID_REGIONS:
1252 raise ConfigValidationError(
1253 f"vector_store.replica_regions contains invalid region "
1254 f"'{region}'. Valid regions: {sorted(self.VALID_REGIONS)}"
1255 )
1256 if region == global_region:
1257 raise ConfigValidationError(
1258 f"vector_store.replica_regions must not include the global "
1259 f"region '{global_region}': the primary table already lives "
1260 "there and a global table cannot replicate into its own region."
1261 )
1262 if "corpus_prefix" in vector_store_ctx:
1263 corpus_prefix = vector_store_ctx["corpus_prefix"]
1264 if (
1265 not isinstance(corpus_prefix, str)
1266 or not corpus_prefix.strip()
1267 or not corpus_prefix.endswith("/")
1268 or corpus_prefix.startswith("/")
1269 ):
1270 raise ConfigValidationError(
1271 "vector_store.corpus_prefix must be a non-empty S3 key prefix "
1272 f"ending in '/' (and not starting with '/'), got {corpus_prefix!r}"
1273 )
1275 def get_project_name(self) -> str:
1276 """Get project name from configuration"""
1277 return self.app.node.try_get_context("project_name") or "gco"
1279 def get_deployment_regions(self) -> dict[str, Any]:
1280 """Get deployment regions configuration.
1282 Returns a dict with:
1283 - global: Region for Global Accelerator and SSM parameters (default: us-east-2)
1284 - api_gateway: Region for API Gateway stack (default: us-east-2)
1285 - monitoring: Region for Monitoring stack (default: us-east-2)
1286 - regional: List of regions for EKS clusters (default: ["us-east-1"])
1288 Note: Global Accelerator is a global service but requires a "home" region
1289 for CloudFormation deployment. us-east-2 is used by default to keep
1290 global infrastructure separate from workload regions.
1291 """
1292 deployment_regions = self.app.node.try_get_context("deployment_regions") or {}
1294 return {
1295 "global": deployment_regions.get("global", "us-east-2"),
1296 "api_gateway": deployment_regions.get("api_gateway", "us-east-2"),
1297 "monitoring": deployment_regions.get("monitoring", "us-east-2"),
1298 "regional": deployment_regions.get("regional", ["us-east-1"]),
1299 }
1301 def get_deployment_partition(self) -> str:
1302 """Return the one SDK partition shared by every configured Region."""
1303 deployment_regions = self.get_deployment_regions()
1304 regional = validated_regional_deployment_regions(
1305 deployment_regions["regional"],
1306 known_regions=self.VALID_REGIONS,
1307 )
1308 return validated_deployment_partition(
1309 (
1310 deployment_regions["global"],
1311 deployment_regions["api_gateway"],
1312 deployment_regions["monitoring"],
1313 *regional,
1314 )
1315 )
1317 def supports_global_accelerator(self) -> bool:
1318 """Return whether this partition exposes the Global Accelerator topology."""
1319 return self.get_deployment_partition() == "aws"
1321 def get_global_region(self) -> str:
1322 """Get the region for global resources and shared SSM parameters."""
1323 region = self.get_deployment_regions()["global"]
1324 return str(region)
1326 def get_api_gateway_region(self) -> str:
1327 """Get the region for API Gateway stack."""
1328 region = self.get_deployment_regions()["api_gateway"]
1329 return str(region)
1331 def get_monitoring_region(self) -> str:
1332 """Get the region for Monitoring stack."""
1333 region = self.get_deployment_regions()["monitoring"]
1334 return str(region)
1336 def get_regions(self) -> list[str]:
1337 """Get list of regions for EKS cluster deployment."""
1338 deployment_regions = self.get_deployment_regions()
1339 regional = deployment_regions["regional"]
1340 return list(regional) if isinstance(regional, list) else [str(regional)]
1342 def get_kubernetes_version(self) -> str:
1343 """Get Kubernetes version from configuration"""
1344 return self.app.node.try_get_context("kubernetes_version") or "1.36"
1346 def get_resource_thresholds(self) -> ResourceThresholds:
1347 """Get resource thresholds configuration"""
1348 thresholds_config = self.app.node.try_get_context("resource_thresholds") or {
1349 "cpu_threshold": 80,
1350 "memory_threshold": 80,
1351 "gpu_threshold": -1,
1352 "pending_pods_threshold": 10,
1353 "pending_requested_cpu_vcpus": 100,
1354 "pending_requested_memory_gb": 200,
1355 "pending_requested_gpus": -1,
1356 }
1357 return ResourceThresholds(
1358 cpu_threshold=thresholds_config["cpu_threshold"],
1359 memory_threshold=thresholds_config["memory_threshold"],
1360 gpu_threshold=thresholds_config["gpu_threshold"],
1361 pending_pods_threshold=thresholds_config.get("pending_pods_threshold", 10),
1362 pending_requested_cpu_vcpus=thresholds_config.get("pending_requested_cpu_vcpus", 100),
1363 pending_requested_memory_gb=thresholds_config.get("pending_requested_memory_gb", 200),
1364 pending_requested_gpus=thresholds_config.get("pending_requested_gpus", 8),
1365 )
1367 def get_cluster_config(self, region: str) -> ClusterConfig:
1368 """Get complete cluster configuration for a region"""
1369 return ClusterConfig(
1370 region=region,
1371 cluster_name=f"{self.get_project_name()}-{region}",
1372 kubernetes_version=self.get_kubernetes_version(),
1373 addons=["metrics-server"],
1374 resource_thresholds=self.get_resource_thresholds(),
1375 )
1377 def get_global_accelerator_config(self) -> dict[str, Any]:
1378 """Get the merged Global Accelerator configuration.
1380 Returns the ``global_accelerator`` block from cdk.json layered on top
1381 of the defaults below. This is a real merge, not the historical
1382 all-or-nothing fallback: a partial block keeps every unspecified
1383 default instead of silently dropping it. The ``traffic_dial``
1384 sub-block is deep-merged so overriding a single dial knob does not
1385 wipe the sub-block's other defaults, mirroring
1386 ``get_cost_monitoring_config``.
1388 Keys:
1389 - name: accelerator name (default ``<project_name>-accelerator``)
1390 - health_check_interval: seconds between endpoint-group probes;
1391 the Global Accelerator API accepts only 10 or 30 (default 30)
1392 - health_check_threshold: consecutive probes before an endpoint
1393 flips healthy/unhealthy, 1-10 (default 3)
1394 - health_check_path: HTTPS path probed on each regional ALB
1395 (default ``/api/v1/health``)
1396 - client_affinity: ``NONE`` or ``SOURCE_IP`` (default ``NONE``)
1397 - traffic_dial: capacity-driven traffic-dial controller
1398 sub-block (default disabled, ``monitor`` mode; knob reference
1399 in :meth:`_validate_traffic_dial_config`)
1400 """
1401 default_config: dict[str, Any] = {
1402 "name": f"{self.get_project_name()}-accelerator",
1403 "health_check_interval": 30,
1404 "health_check_threshold": 3,
1405 "health_check_path": "/api/v1/health",
1406 "client_affinity": "NONE",
1407 "traffic_dial": {
1408 "enabled": False,
1409 "mode": "monitor",
1410 "interval_minutes": 5,
1411 "lookback_minutes": 15,
1412 "min_dial_percentage": 10,
1413 "max_step_percentage": 20,
1414 "full_health_percentage": 95,
1415 },
1416 }
1417 configured = self.app.node.try_get_context("global_accelerator") or {}
1418 if not isinstance(configured, dict):
1419 raise ConfigValidationError("global_accelerator must be a mapping")
1420 merged: dict[str, Any] = {**default_config, **configured}
1422 # Deep-merge the nested sub-block so a partial override does not drop
1423 # the other defaults in the same sub-block. A non-mapping override is
1424 # deliberately left in place for the validator to reject with a
1425 # precise message.
1426 override = configured.get("traffic_dial")
1427 if isinstance(override, dict):
1428 default_sub = cast(dict[str, Any], default_config["traffic_dial"])
1429 merged["traffic_dial"] = {**default_sub, **override}
1431 return merged
1433 def get_backend_tls_config(self) -> dict[str, Any]:
1434 """Return the mandatory deployment-local backend TLS lifecycle policy."""
1435 defaults = {
1436 "root_generation": 1,
1437 "root_validity_days": 3_650,
1438 "root_rotate_before_days": 180,
1439 "root_activation_delay_hours": 24,
1440 "root_overlap_days": 45,
1441 "leaf_validity_days": 30,
1442 "leaf_rotate_before_days": 10,
1443 "rotation_schedule_hours": 12,
1444 "trust_cache_ttl_seconds": 300,
1445 "trust_cache_max_stale_seconds": 3_600,
1446 }
1447 configured = self.app.node.try_get_context("backend_tls") or {}
1448 if not isinstance(configured, dict):
1449 raise ConfigValidationError("backend_tls must be a mapping")
1450 return {**defaults, **configured}
1452 def get_inference_proxy_config(self) -> dict[str, int]:
1453 """Return merged inference TLS proxy autoscaling settings.
1455 Omission is backward compatible and returns the shipped defaults. AWS
1456 CDK normalizes a top-level JSON ``null`` context value to omission, so
1457 ``None`` follows the same default-preserving contract.
1458 """
1459 defaults = {
1460 "tls_proxy_cpu_request_millicores": (
1461 INFERENCE_PROXY_TLS_CPU_REQUEST_MILLICORES_DEFAULT
1462 ),
1463 "tls_proxy_cpu_target_utilization_percentage": (
1464 INFERENCE_PROXY_TLS_CPU_TARGET_UTILIZATION_DEFAULT
1465 ),
1466 "min_replicas": INFERENCE_PROXY_MIN_REPLICAS_DEFAULT,
1467 "max_replicas": INFERENCE_PROXY_MAX_REPLICAS_DEFAULT,
1468 }
1469 configured = self.app.node.try_get_context("inference_proxy")
1470 if configured is None:
1471 return dict(defaults)
1472 if not isinstance(configured, dict):
1473 raise ConfigValidationError(
1474 "inference_proxy must be an object, got "
1475 f"{type(configured).__name__}: {configured!r}"
1476 )
1478 unknown = sorted(str(key) for key in configured if key not in defaults)
1479 if unknown:
1480 unknown_paths = ", ".join(f"inference_proxy.{key}" for key in unknown)
1481 allowed_paths = ", ".join(f"inference_proxy.{key}" for key in sorted(defaults))
1482 raise ConfigValidationError(
1483 f"inference_proxy contains unknown key(s): {unknown_paths}; "
1484 f"allowed keys: {allowed_paths}"
1485 )
1486 return {**defaults, **configured}
1488 def get_alb_config(self) -> dict[str, Any]:
1489 """Get ALB configuration"""
1490 return self.app.node.try_get_context("alb_config") or {
1491 "health_check_interval": 30,
1492 "health_check_timeout": 5,
1493 "healthy_threshold": 2,
1494 "unhealthy_threshold": 2,
1495 }
1497 def get_manifest_processor_config(self) -> dict[str, Any]:
1498 """Get manifest processor configuration.
1500 Merges three cdk.json sections into a single runtime config:
1502 - ``manifest_processor``: service-specific settings (replicas, image,
1503 resource_limits, allowed_namespaces, validation_enabled,
1504 max_request_body_bytes, yaml_max_depth)
1505 - ``job_validation_policy``: shared validation policy (resource_quotas,
1506 trusted_registries, trusted_dockerhub_orgs, manifest_security_policy,
1507 allowed_kinds). Pulled in verbatim so the REST path reads the same
1508 policy the SQS queue processor enforces.
1510 Note: The 'image' field is a placeholder default. In practice, the actual
1511 image is built from dockerfiles/manifest-processor-dockerfile and pushed
1512 to ECR during CDK deployment. The {{MANIFEST_PROCESSOR_IMAGE}} placeholder
1513 in manifests is replaced with the ECR image URI.
1514 """
1515 default_config = {
1516 "image": "gco/manifest-processor:latest", # Placeholder, replaced by ECR image
1517 "replicas": 3,
1518 "resource_limits": {"cpu": "1000m", "memory": "2Gi"},
1519 "autoscaling": dict(_MANIFEST_PROCESSOR_AUTOSCALING_DEFAULTS),
1520 "validation_enabled": True,
1521 "max_request_body_bytes": DEFAULT_MAX_REQUEST_BODY_BYTES,
1522 "central_queue_worker_enabled": True,
1523 "central_queue_poll_interval_seconds": 10,
1524 "central_queue_batch_size": 5,
1525 "central_queue_reconcile_limit": 100,
1526 "central_queue_lease_seconds": 300,
1527 "central_queue_lease_renewal_seconds": 60,
1528 # allowed_namespaces, resource_quotas, trusted_registries,
1529 # trusted_dockerhub_orgs, manifest_security_policy, and
1530 # allowed_kinds are merged in below from job_validation_policy.
1531 "allowed_namespaces": ["gco-jobs"],
1532 "resource_quotas": {
1533 "max_cpu_per_manifest": "10",
1534 "max_memory_per_manifest": "32Gi",
1535 "max_gpu_per_manifest": 4,
1536 },
1537 "trusted_registries": [
1538 "docker.io",
1539 "gcr.io",
1540 "quay.io",
1541 "registry.k8s.io",
1542 "k8s.gcr.io",
1543 "public.ecr.aws",
1544 "nvcr.io",
1545 "gco",
1546 ],
1547 "trusted_dockerhub_orgs": [
1548 "nvidia",
1549 "pytorch",
1550 "rayproject",
1551 "tensorflow",
1552 "huggingface",
1553 "amazon",
1554 "bitnami",
1555 ],
1556 }
1557 context_config = self.app.node.try_get_context("manifest_processor") or {}
1559 # Merge in the shared job_validation_policy section. These keys apply
1560 # to BOTH the manifest processor and the queue processor; they live
1561 # in their own top-level cdk.json section so neither service "owns"
1562 # them. We flatten them into the manifest processor's runtime config
1563 # so service code keeps its existing attribute layout.
1564 shared_policy = self.app.node.try_get_context("job_validation_policy") or {}
1565 merged = {**default_config, **context_config, **shared_policy}
1566 # Nested block: a partial ``autoscaling`` object (or JSON null) keeps
1567 # the unspecified defaults instead of replacing the whole mapping.
1568 # ``_validate_config`` has already rejected anything that is not a
1569 # mapping or null.
1570 merged["autoscaling"] = {
1571 **_MANIFEST_PROCESSOR_AUTOSCALING_DEFAULTS,
1572 **(context_config.get("autoscaling") or {}),
1573 }
1575 enabled = merged.get("central_queue_worker_enabled")
1576 if not isinstance(enabled, bool):
1577 raise ConfigValidationError(
1578 "manifest_processor.central_queue_worker_enabled must be a boolean"
1579 )
1580 for key, minimum, maximum in (
1581 ("central_queue_poll_interval_seconds", 1, 300),
1582 ("central_queue_batch_size", 1, 20),
1583 ("central_queue_reconcile_limit", 1, 500),
1584 ("central_queue_lease_seconds", 30, 3600),
1585 ("central_queue_lease_renewal_seconds", 1, 300),
1586 ):
1587 value = merged.get(key)
1588 if type(value) is not int or not minimum <= value <= maximum:
1589 raise ConfigValidationError(
1590 f"manifest_processor.{key} must be an integer between {minimum} and {maximum}"
1591 )
1592 if (
1593 merged["central_queue_lease_renewal_seconds"] * 2
1594 > merged["central_queue_lease_seconds"]
1595 ):
1596 raise ConfigValidationError(
1597 "manifest_processor.central_queue_lease_renewal_seconds must be no more than "
1598 "half of central_queue_lease_seconds"
1599 )
1600 return merged
1602 def get_api_gateway_config(self) -> dict[str, Any]:
1603 """Get API Gateway configuration.
1605 Returns:
1606 API Gateway configuration dictionary with the following keys:
1607 - throttle_rate_limit: Requests per second limit
1608 - throttle_burst_limit: Burst capacity
1609 - log_level: CloudWatch logging level (OFF, ERROR, INFO)
1610 - metrics_enabled: Enable CloudWatch metrics
1611 - tracing_enabled: Enable X-Ray tracing
1612 - regional_api_enabled: In the commercial ``aws`` partition,
1613 permit direct same-account callers to use the always-deployed
1614 regional API bridges. Other partitions force this access on
1615 because the bridges are the supported workload ingress without
1616 Global Accelerator. Centralized aggregation always uses them.
1617 """
1618 default_config = {
1619 "throttle_rate_limit": 1000,
1620 "throttle_burst_limit": 2000,
1621 "log_level": "INFO",
1622 "metrics_enabled": True,
1623 "tracing_enabled": True,
1624 "regional_api_enabled": False,
1625 }
1626 return {**default_config, **(self.app.node.try_get_context("api_gateway") or {})}
1628 def get_eks_cluster_config(self) -> dict[str, Any]:
1629 """Get EKS cluster configuration.
1631 Returns:
1632 EKS cluster configuration dictionary with the following keys:
1633 - endpoint_access: EKS API endpoint access mode
1634 - "PRIVATE": API server only accessible from within VPC (default, most secure)
1635 - "PUBLIC_AND_PRIVATE": API server accessible from internet and VPC
1636 - public_access_cidrs: CIDR allowlist for the public endpoint when
1637 endpoint_access is PUBLIC_AND_PRIVATE. Empty (the default) means
1638 0.0.0.0/0, which synthesis calls out with a loud warning.
1639 - developer_access: list of EKS access entries to synthesize for
1640 human principals, each ``{principal_arn, scope, namespaces}``.
1641 scope defaults to "namespace" and namespaces to ["gco-jobs"];
1642 scope "cluster" grants AmazonEKSClusterAdminPolicy instead.
1643 Empty (the default) synthesizes exactly today's entries.
1644 - network_policy_enforcement: whether the Auto Mode network policy
1645 controller is switched on (default True). False keeps the
1646 NetworkPolicy objects but stops enforcing them.
1648 Note:
1649 PRIVATE endpoint is recommended for production. Job submission still works
1650 via API Gateway → Lambda (in VPC) or SQS queues. For kubectl access with
1651 PRIVATE endpoint, use `gco cluster tunnel` (SSM), a bastion host, or a VPN —
1652 and an access entry for your principal either way (`gco stacks access`).
1653 """
1654 default_config: dict[str, Any] = {
1655 "endpoint_access": "PRIVATE",
1656 "public_access_cidrs": [],
1657 "developer_access": [],
1658 # Switches on the Auto Mode network policy controller
1659 # (06-network-policy-controller.yaml) so 03-network-policies.yaml
1660 # is enforced rather than merely stored.
1661 "network_policy_enforcement": True,
1662 }
1663 return {**default_config, **(self.app.node.try_get_context("eks_cluster") or {})}
1665 def get_fsx_lustre_config(self, region: str | None = None) -> dict[str, Any]:
1666 """Get FSx for Lustre configuration.
1668 Args:
1669 region: Optional region to get config for. If provided, checks for
1670 region-specific overrides first.
1672 Returns:
1673 FSx configuration dictionary with the following keys:
1674 - enabled: Whether FSx is enabled
1675 - storage_capacity_gib: Storage capacity in GiB (min 1200)
1676 - deployment_type: SCRATCH_1, SCRATCH_2, PERSISTENT_1, PERSISTENT_2
1677 - file_system_type_version: Lustre version (2.12 or 2.15, default: 2.15)
1678 IMPORTANT: Use 2.15 for kernel 6.x compatibility (AL2023, Bottlerocket)
1679 - per_unit_storage_throughput: Throughput for PERSISTENT types
1680 - data_compression_type: LZ4 or NONE
1681 - import_path: S3 path for data import
1682 - export_path: S3 path for data export
1683 - auto_import_policy: NEW, NEW_CHANGED, NEW_CHANGED_DELETED
1684 - node_group: Node group configuration for FSx workloads
1685 - instance_types: List of instance types
1686 - min_size: Minimum nodes (default: 0)
1687 - max_size: Maximum nodes (default: 10)
1688 - desired_size: Desired nodes (default: 0, scales from zero)
1689 - ami_type: AMI type - one of:
1690 AL2023_X86_64_STANDARD (default), AL2023_ARM_64_STANDARD,
1691 AL2023_X86_64_NVIDIA, AL2023_ARM_64_NVIDIA, AL2023_X86_64_NEURON
1692 - capacity_type: ON_DEMAND (default) or SPOT
1693 - disk_size: Root disk size in GB (default: 100)
1694 - labels: Additional node labels (dict)
1695 """
1696 default_config = {
1697 "enabled": False,
1698 "storage_capacity_gib": 1200,
1699 "deployment_type": "SCRATCH_2",
1700 "file_system_type_version": "2.15", # Use 2.15 for kernel 6.x compatibility
1701 "per_unit_storage_throughput": 200,
1702 "data_compression_type": "LZ4",
1703 "import_path": None,
1704 "export_path": None,
1705 "auto_import_policy": "NEW_CHANGED_DELETED",
1706 "node_group": {
1707 "instance_types": ["m5.large", "m5.xlarge", "m6i.large", "m6i.xlarge"],
1708 "min_size": 0,
1709 "max_size": 10,
1710 "desired_size": 1,
1711 "ami_type": "AL2023_X86_64_STANDARD",
1712 "capacity_type": "ON_DEMAND",
1713 "disk_size": 100,
1714 "labels": {},
1715 },
1716 }
1718 # Get global FSx config
1719 global_ctx = self.app.node.try_get_context("fsx_lustre")
1720 global_config: dict[str, Any] = global_ctx if isinstance(global_ctx, dict) else {}
1721 merged_config: dict[str, Any] = {**default_config, **global_config}
1723 # Ensure node_group has all required fields with defaults
1724 if "node_group" in global_config:
1725 global_node_group = global_config["node_group"]
1726 if isinstance(global_node_group, dict):
1727 default_node_group = cast(dict[str, Any], default_config["node_group"])
1728 merged_config["node_group"] = {
1729 **default_node_group,
1730 **global_node_group,
1731 }
1733 # Check for region-specific override
1734 if region:
1735 region_overrides_ctx = self.app.node.try_get_context("fsx_lustre_regions")
1736 region_overrides: dict[str, Any] = (
1737 region_overrides_ctx if isinstance(region_overrides_ctx, dict) else {}
1738 )
1739 if region in region_overrides:
1740 region_config = region_overrides[region]
1741 if isinstance(region_config, dict):
1742 # Preserve the fully merged default/global node group before
1743 # the top-level regional overlay replaces that nested value.
1744 # A regional node_group is a patch, not a wholesale reset.
1745 existing_node_group = merged_config.get("node_group")
1746 merged_config = {**merged_config, **region_config}
1747 # Handle nested node_group override
1748 if "node_group" in region_config:
1749 region_node_group = region_config["node_group"]
1750 if isinstance(region_node_group, dict):
1751 if isinstance(existing_node_group, dict):
1752 base_node_group = existing_node_group
1753 else:
1754 base_node_group = cast(dict[str, Any], default_config["node_group"])
1755 merged_config["node_group"] = {
1756 **base_node_group,
1757 **region_node_group,
1758 }
1760 if self._feature_override_enabled("fsx_lustre"):
1761 merged_config["enabled"] = True
1762 return merged_config
1764 def _feature_override_enabled(self, feature_key: str) -> bool:
1765 """True when ``feature_enabled_overrides`` context forces this feature on."""
1766 overrides = parse_feature_enabled_overrides(
1767 self.app.node.try_get_context(FEATURE_OVERRIDE_CONTEXT_KEY)
1768 )
1769 return feature_key in overrides
1771 def get_valkey_config(self) -> dict[str, Any]:
1772 """Get Valkey Serverless cache configuration.
1774 Returns:
1775 Valkey configuration dictionary with the following keys:
1776 - enabled: Whether Valkey cache is enabled (default: False)
1777 - max_data_storage_gb: Maximum data storage in GB (default: 5)
1778 - max_ecpu_per_second: Maximum ECPUs per second (default: 5000)
1779 - snapshot_retention_limit: Daily snapshots to retain (default: 1)
1780 """
1781 default_config: dict[str, Any] = {
1782 "enabled": False,
1783 "max_data_storage_gb": 5,
1784 "max_ecpu_per_second": 5000,
1785 "snapshot_retention_limit": 1,
1786 }
1787 valkey_ctx = self.app.node.try_get_context("valkey")
1788 valkey_config: dict[str, Any] = valkey_ctx if isinstance(valkey_ctx, dict) else {}
1789 merged = {**default_config, **valkey_config}
1790 if self._feature_override_enabled("valkey"):
1791 merged["enabled"] = True
1792 return merged
1794 def get_aurora_pgvector_config(self) -> dict[str, Any]:
1795 """Get Aurora Serverless v2 + pgvector vector database configuration.
1797 Returns:
1798 Aurora pgvector configuration dictionary with the following keys:
1799 - enabled: Whether Aurora pgvector is enabled (default: False)
1800 - min_acu: Minimum Aurora Capacity Units (default: 0, scales to zero)
1801 - max_acu: Maximum Aurora Capacity Units (default: 16)
1802 - backup_retention_days: Number of days to retain automated backups (default: 7)
1803 - deletion_protection: Whether deletion protection is enabled (default: False)
1804 """
1805 default_config: dict[str, Any] = {
1806 "enabled": False,
1807 "min_acu": 0,
1808 "max_acu": 16,
1809 "backup_retention_days": 7,
1810 "deletion_protection": False,
1811 }
1812 aurora_ctx = self.app.node.try_get_context("aurora_pgvector")
1813 aurora_config: dict[str, Any] = aurora_ctx if isinstance(aurora_ctx, dict) else {}
1814 merged = {**default_config, **aurora_config}
1815 if self._feature_override_enabled("aurora_pgvector"):
1816 merged["enabled"] = True
1817 return merged
1819 def get_analytics_config(self) -> dict[str, Any]:
1820 """Get optional analytics environment configuration.
1822 Returns the fully-merged analytics_environment block from cdk.json
1823 layered on top of the defaults below. Sub-blocks (``hyperpod``,
1824 ``cognito``, ``efs``, ``studio``) are deep-merged so a user who
1825 overrides a single nested key (e.g. ``cognito.domain_prefix``) does
1826 not inadvertently wipe the sub-block's other defaults — mirroring the
1827 nested-merge pattern used by ``get_fsx_lustre_config`` for its
1828 ``node_group`` sub-block.
1830 Returns:
1831 Analytics configuration dictionary with the following keys:
1832 - enabled: Whether the analytics environment stack is deployed
1833 (default: False — the feature is off unless explicitly opted in)
1834 - hyperpod: SageMaker HyperPod integration sub-block
1835 - enabled: Whether to add the HyperPod IAM grants to
1836 SageMaker_Execution_Role (default: False)
1837 - canvas: SageMaker Canvas integration sub-block
1838 - enabled: Whether to enable the SageMaker Canvas app on
1839 the Studio domain and attach ``AmazonSageMakerCanvasFullAccess``
1840 to the SageMaker_Execution_Role (default: False)
1841 - cognito: Cognito user-pool sub-block
1842 - domain_prefix: UserPoolDomain prefix, or None to let the
1843 analytics stack derive one (default: None)
1844 - removal_policy: "destroy" (default) or "retain" — controls
1845 the Cognito pool's CloudFormation DeletionPolicy
1846 - efs: Studio_EFS sub-block
1847 - removal_policy: "destroy" (default) or "retain" — controls
1848 the Studio EFS file system's CloudFormation DeletionPolicy
1849 - studio: SageMaker Studio sub-block
1850 - user_profile_name_prefix: Optional prefix for per-user
1851 profile names, or None to use the Cognito username verbatim
1852 (default: None)
1853 """
1854 default_config: dict[str, Any] = {
1855 "enabled": False,
1856 "hyperpod": {"enabled": False},
1857 "canvas": {"enabled": False},
1858 "cognito": {"domain_prefix": None, "removal_policy": "destroy"},
1859 "efs": {"removal_policy": "destroy"},
1860 "studio": {"user_profile_name_prefix": None},
1861 }
1862 analytics_ctx = self.app.node.try_get_context("analytics_environment")
1863 analytics_config: dict[str, Any] = analytics_ctx if isinstance(analytics_ctx, dict) else {}
1864 merged_config: dict[str, Any] = {**default_config, **analytics_config}
1866 # Deep-merge each nested sub-block so a partial override does not
1867 # drop the other defaults in the same sub-block.
1868 for sub_block in ("hyperpod", "canvas", "cognito", "efs", "studio"):
1869 override = analytics_config.get(sub_block)
1870 if isinstance(override, dict):
1871 default_sub = cast(dict[str, Any], default_config[sub_block])
1872 merged_config[sub_block] = {**default_sub, **override}
1874 return merged_config
1876 def get_analytics_enabled(self) -> bool:
1877 """Return whether the analytics environment stack is enabled.
1879 Thin wrapper around ``get_analytics_config()["enabled"]`` to mirror
1880 the existing ``get_valkey_config`` / ``get_aurora_pgvector_config``
1881 access pattern without forcing every call site to index into the
1882 merged dict.
1883 """
1884 return bool(self.get_analytics_config()["enabled"])
1886 def get_cluster_observability_config(self) -> dict[str, Any]:
1887 """Get the in-cluster observability configuration.
1889 Returns the fully-merged cluster_observability block from cdk.json
1890 layered on top of the defaults below. Sub-blocks (``grafana``,
1891 ``prometheus``, ``alertmanager``) are deep-merged so a user who
1892 overrides a single nested key (e.g. ``prometheus.retention``) does not
1893 inadvertently wipe the sub-block's other defaults — mirroring the
1894 nested-merge pattern used by ``get_analytics_config``.
1896 Unlike most optional features, this one is **on by default**: a stock
1897 deployment installs kube-prometheus-stack on every regional cluster.
1898 Operators opt out by setting ``cluster_observability.enabled = false``.
1900 Returns:
1901 Cluster observability configuration dictionary with the keys:
1902 - enabled: Whether kube-prometheus-stack is installed per region
1903 (default: True)
1904 - grafana: Grafana sub-block
1905 - persistence_size: EBS PVC size for Grafana's user database
1906 and dashboards (default: "10Gi")
1907 - admin_user: Grafana admin username; the password is
1908 chart-generated in the <release>-grafana Secret, never
1909 authored here (default: "admin")
1910 - admin_password_rotation_schedule: 5-field cron for the
1911 in-cluster CronJob that rotates the chart-generated admin
1912 password (default: "0 4 1 * *", monthly)
1913 - prometheus: Prometheus sub-block
1914 - persistence_size: EBS PVC size for the Prometheus TSDB
1915 (default: "50Gi")
1916 - retention: Prometheus retention window (default: "15d")
1917 - alertmanager: Alertmanager sub-block
1918 - enabled: Whether Alertmanager is deployed (default: True)
1919 - persistence_size: EBS PVC size for Alertmanager (default: "5Gi")
1920 - mlflow: MLflow experiment-tracking sub-block
1921 - enabled: Whether the MLflow tracking server is installed per
1922 region (default: True). Effective only while observability
1923 itself is enabled — see ``get_mlflow_enabled``.
1924 - persistence_size: EBS PVC size for the tracking server's
1925 SQLite run-metadata store (default: "10Gi"); artifacts go
1926 to S3, not this volume
1927 """
1928 default_config: dict[str, Any] = {
1929 "enabled": True,
1930 "grafana": {
1931 "persistence_size": "10Gi",
1932 "admin_user": "admin",
1933 # Monthly (04:00 on the 1st) rotation of the chart-generated
1934 # Grafana admin password, run by an in-cluster CronJob.
1935 "admin_password_rotation_schedule": "0 4 1 * *",
1936 },
1937 "prometheus": {"persistence_size": "50Gi", "retention": "15d"},
1938 "alertmanager": {"enabled": True, "persistence_size": "5Gi"},
1939 "mlflow": {"enabled": True, "persistence_size": "10Gi"},
1940 }
1941 obs_ctx = self.app.node.try_get_context("cluster_observability")
1942 obs_config: dict[str, Any] = obs_ctx if isinstance(obs_ctx, dict) else {}
1943 merged_config: dict[str, Any] = {**default_config, **obs_config}
1945 # Deep-merge each nested sub-block so a partial override does not
1946 # drop the other defaults in the same sub-block.
1947 for sub_block in ("grafana", "prometheus", "alertmanager", "mlflow"):
1948 override = obs_config.get(sub_block)
1949 if isinstance(override, dict):
1950 default_sub = cast(dict[str, Any], default_config[sub_block])
1951 merged_config[sub_block] = {**default_sub, **override}
1953 return merged_config
1955 def get_cluster_observability_enabled(self) -> bool:
1956 """Return whether in-cluster observability is enabled (default True).
1958 Thin wrapper around ``get_cluster_observability_config()["enabled"]``
1959 so call sites (the regional stack's chart-enable and value-override
1960 methods, the CLI) do not have to index into the merged dict.
1961 """
1962 return bool(self.get_cluster_observability_config()["enabled"])
1964 def get_mlflow_enabled(self) -> bool:
1965 """Return whether the MLflow tracking server is effectively enabled.
1967 The conjunction of ``cluster_observability.mlflow.enabled`` (default
1968 True) and ``cluster_observability.enabled`` (default True): MLflow
1969 installs into the ``monitoring`` namespace kube-prometheus-stack
1970 creates, stores run metadata on the observability gp3 StorageClass,
1971 and is reached through the same tunnel commands, so disabling
1972 observability switches the tracking server off with it rather than
1973 deploying it against missing storage — the same conjunction shape
1974 ``get_cost_monitoring_enabled`` uses for OpenCost.
1975 """
1976 obs = self.get_cluster_observability_config()
1977 return bool(obs["mlflow"]["enabled"]) and bool(obs["enabled"])
1979 def get_cost_monitoring_config(self) -> dict[str, Any]:
1980 """Get the cost monitoring configuration.
1982 Returns the fully-merged ``cost_monitoring`` block from cdk.json
1983 layered on top of the defaults below. Sub-blocks (``reports``,
1984 ``athena``) are deep-merged so a user who overrides a single nested
1985 key does not inadvertently wipe the sub-block's other defaults —
1986 mirroring the nested-merge pattern used by
1987 ``get_cluster_observability_config``.
1989 Like cluster observability, cost monitoring is **on by default**: a
1990 stock deployment installs OpenCost per region, provisions the cost
1991 report bucket + Athena analytics in the monitoring stack, and runs
1992 the cost-monitor service on every regional cluster. Operators opt out
1993 by setting ``cost_monitoring.enabled = false``.
1995 Returns:
1996 Cost monitoring configuration dictionary with the keys:
1997 - enabled: Whether the cost monitoring pipeline is deployed
1998 (default: True). Requires ``cluster_observability.enabled``.
1999 - reports: Cost report sub-block
2000 - interval_minutes: cadence of the cost-monitor service's
2001 scheduled Parquet reports (default: 60)
2002 - retention_days: S3 lifecycle expiration for report objects
2003 (default: 365)
2004 - transition_to_infrequent_access_days: S3 lifecycle transition
2005 to STANDARD_IA (default: 90; must be < retention_days)
2006 - athena: Athena analytics sub-block
2007 - query_results_retention_days: S3 lifecycle expiration for
2008 Athena query results written under ``athena-results/``
2009 (default: 30)
2010 """
2011 default_config: dict[str, Any] = {
2012 "enabled": True,
2013 "reports": {
2014 "interval_minutes": 60,
2015 "retention_days": 365,
2016 "transition_to_infrequent_access_days": 90,
2017 },
2018 "athena": {
2019 "query_results_retention_days": 30,
2020 },
2021 }
2022 cost_ctx = self.app.node.try_get_context("cost_monitoring")
2023 cost_config: dict[str, Any] = cost_ctx if isinstance(cost_ctx, dict) else {}
2024 merged_config: dict[str, Any] = {**default_config, **cost_config}
2026 # Deep-merge each nested sub-block so a partial override does not
2027 # drop the other defaults in the same sub-block.
2028 for sub_block in ("reports", "athena"):
2029 override = cost_config.get(sub_block)
2030 if isinstance(override, dict):
2031 default_sub = cast(dict[str, Any], default_config[sub_block])
2032 merged_config[sub_block] = {**default_sub, **override}
2034 return merged_config
2036 def get_cost_monitoring_enabled(self) -> bool:
2037 """Return whether the cost monitoring pipeline is effectively enabled.
2039 The conjunction of ``cost_monitoring.enabled`` (default True) and
2040 ``cluster_observability.enabled`` (default True): OpenCost reads its
2041 usage data from the in-cluster Prometheus, so disabling observability
2042 switches the whole cost pipeline off with it rather than deploying a
2043 pipeline with no data source (or failing synthesis for a user who
2044 only ran ``gco monitoring disable``). Call sites — the regional
2045 stack's chart-enable and image-build methods, the monitoring stack,
2046 the CLI — all gate on this one conjunction.
2047 """
2048 return bool(self.get_cost_monitoring_config()["enabled"]) and bool(
2049 self.get_cluster_observability_config()["enabled"]
2050 )
2052 def get_capacity_history_config(self) -> dict[str, Any]:
2053 """Get the optional historical capacity surface configuration.
2055 Returns the merged ``historical`` block from cdk.json layered on top of
2056 the defaults below. The feature is off unless ``historical.enabled`` is
2057 explicitly true, mirroring the analytics-environment opt-in pattern.
2059 Keys:
2060 - enabled: deploy the capacity poller stack + history table (default False)
2061 - retention_days: DynamoDB TTL window for snapshots (default 90)
2062 - poll_interval_minutes: EventBridge schedule cadence (default 15)
2063 - capacity_block_duration_hours: short Capacity Block probe duration
2064 the poller snapshots (default 24 = 1 day)
2065 - capacity_block_long_duration_hours: long Capacity Block probe
2066 duration in hours (default 1512 = 63 days); 0 disables the long
2067 probe and its ``capacity_blocks_long_*`` metrics
2068 - spot_score_target_capacities: Spot Placement Score target
2069 capacities the poller snapshots per instance pool (default
2070 [1, 10, 50]); a subset selector over the supported set exported
2071 by ``cli/capacity/history.py``, where capacity 1 keeps the
2072 original ``spot_score`` field and N > 1 writes ``spot_score_at_N``
2073 - watch_instance_types: instance types the poller snapshots
2074 - enabled_regions: regions to poll; empty means all deployed regions
2075 """
2076 default_config: dict[str, Any] = {
2077 "enabled": False,
2078 "retention_days": 90,
2079 "poll_interval_minutes": 15,
2080 "capacity_block_duration_hours": 24,
2081 "capacity_block_long_duration_hours": 63 * 24,
2082 "spot_score_target_capacities": [1, 10, 50],
2083 "watch_instance_types": [
2084 "g4dn.12xlarge",
2085 "g4dn.16xlarge",
2086 "g4dn.2xlarge",
2087 "g4dn.4xlarge",
2088 "g4dn.8xlarge",
2089 "g4dn.metal",
2090 "g4dn.xlarge",
2091 "g5.12xlarge",
2092 "g5.16xlarge",
2093 "g5.24xlarge",
2094 "g5.2xlarge",
2095 "g5.48xlarge",
2096 "g5.4xlarge",
2097 "g5.8xlarge",
2098 "g5.xlarge",
2099 "g5g.16xlarge",
2100 "g5g.2xlarge",
2101 "g5g.4xlarge",
2102 "g5g.8xlarge",
2103 "g5g.metal",
2104 "g5g.xlarge",
2105 "g6.12xlarge",
2106 "g6.16xlarge",
2107 "g6.24xlarge",
2108 "g6.2xlarge",
2109 "g6.48xlarge",
2110 "g6.4xlarge",
2111 "g6.8xlarge",
2112 "g6.xlarge",
2113 "g6e.12xlarge",
2114 "g6e.16xlarge",
2115 "g6e.24xlarge",
2116 "g6e.2xlarge",
2117 "g6e.48xlarge",
2118 "g6e.4xlarge",
2119 "g6e.8xlarge",
2120 "g6e.xlarge",
2121 "g6f.2xlarge",
2122 "g6f.4xlarge",
2123 "g6f.large",
2124 "g6f.xlarge",
2125 "g7.12xlarge",
2126 "g7.24xlarge",
2127 "g7.2xlarge",
2128 "g7.48xlarge",
2129 "g7.4xlarge",
2130 "g7.8xlarge",
2131 "g7e.12xlarge",
2132 "g7e.24xlarge",
2133 "g7e.2xlarge",
2134 "g7e.48xlarge",
2135 "g7e.4xlarge",
2136 "g7e.8xlarge",
2137 "gr6.4xlarge",
2138 "gr6.8xlarge",
2139 "gr6f.4xlarge",
2140 "inf1.24xlarge",
2141 "inf1.2xlarge",
2142 "inf1.6xlarge",
2143 "inf1.xlarge",
2144 "inf2.24xlarge",
2145 "inf2.48xlarge",
2146 "inf2.8xlarge",
2147 "inf2.xlarge",
2148 "p3dn.24xlarge",
2149 "p4d.24xlarge",
2150 "p4de.24xlarge",
2151 "p5.48xlarge",
2152 "p5.4xlarge",
2153 "p5e.48xlarge",
2154 "p5en.48xlarge",
2155 "p6-b200.48xlarge",
2156 "p6-b300.48xlarge",
2157 "trn1.2xlarge",
2158 "trn1.32xlarge",
2159 "trn1n.32xlarge",
2160 "trn2.3xlarge",
2161 "trn2.48xlarge",
2162 ],
2163 "enabled_regions": [],
2164 }
2165 historical_ctx = self.app.node.try_get_context("historical")
2166 historical_config = historical_ctx if isinstance(historical_ctx, dict) else {}
2167 return {**default_config, **historical_config}
2169 def get_capacity_history_enabled(self) -> bool:
2170 """Return whether the historical capacity surface is enabled."""
2171 return bool(self.get_capacity_history_config()["enabled"])
2173 def get_mission_memory_config(self) -> dict[str, Any]:
2174 """Get the mission-memory configuration (recall across mission sessions).
2176 Returns the merged ``mission_memory`` block from cdk.json layered on
2177 top of the defaults below. The feature is ON by default — memory is
2178 cheap (one small PAY_PER_REQUEST item plus one embedding call per
2179 completed mission) and silently missing recall is the worse failure
2180 mode; set ``mission_memory.enabled: false`` to opt out.
2182 Keys:
2183 - enabled: provision the mission-memory table + vector index and
2184 activate best-effort write/retrieval in the engine (default True)
2185 - retention_days: DynamoDB TTL window for memory items (default 365)
2186 - dimensions: embedding vector width (default 1024). ONE-WAY DOOR:
2187 immutable after index creation and must match the configured
2188 ``bedrock.embedding_model_id`` output width.
2189 - distance_function: vector distance metric (default COSINE);
2190 immutable after index creation.
2191 - top_k: similar past missions retrieved into the sampling prompt
2192 (default 3)
2193 """
2194 default_config: dict[str, Any] = {
2195 "enabled": True,
2196 "retention_days": 365,
2197 "dimensions": 1024,
2198 "distance_function": "COSINE",
2199 "top_k": 3,
2200 }
2201 mission_memory_ctx = self.app.node.try_get_context("mission_memory")
2202 mission_memory_config = mission_memory_ctx if isinstance(mission_memory_ctx, dict) else {}
2203 return {**default_config, **mission_memory_config}
2205 def get_mission_memory_enabled(self) -> bool:
2206 """Return whether mission memory is enabled."""
2207 return bool(self.get_mission_memory_config()["enabled"])
2209 def get_vector_store_config(self) -> dict[str, Any]:
2210 """Get the vector-store configuration (global workload RAG corpus).
2212 Returns the merged ``vector_store`` block from cdk.json layered on
2213 top of the defaults below. The feature is OFF by default — a
2214 replicated vector store carries real per-region storage and write
2215 cost, so it is an explicit opt-in like ``aurora_pgvector``; set
2216 ``vector_store.enabled: true`` to provision it.
2218 Keys:
2219 - enabled: provision the global vector-store table + index,
2220 the S3-triggered ingest pipeline, and the regional read wiring
2221 (default False)
2222 - dimensions: embedding vector width (default 1024). ONE-WAY
2223 DOOR: immutable after index creation and must match the
2224 configured ``embedding_model_id`` output width.
2225 - distance_function: vector distance metric (default COSINE);
2226 immutable after index creation.
2227 - embedding_model_id: Bedrock text-embedding model used by the
2228 ingest pipeline and query paths (default
2229 amazon.titan-embed-text-v2:0). Independent of
2230 ``bedrock.embedding_model_id`` on purpose. Changing it means
2231 re-ingesting the corpus: vectors from different models are
2232 not comparable.
2233 - replica_regions: regions to replicate the table into. Empty
2234 (the default) means "follow deployment_regions.regional",
2235 excluding the global region (the primary).
2236 - corpus_prefix: S3 key prefix on the Cluster_Shared_Bucket
2237 watched by the ingest pipeline (default vector-corpus/).
2238 """
2239 default_config: dict[str, Any] = {
2240 "enabled": False,
2241 "dimensions": 1024,
2242 "distance_function": "COSINE",
2243 "embedding_model_id": "amazon.titan-embed-text-v2:0",
2244 "replica_regions": [],
2245 "corpus_prefix": "vector-corpus/",
2246 }
2247 vector_store_ctx = self.app.node.try_get_context("vector_store")
2248 vector_store_config = vector_store_ctx if isinstance(vector_store_ctx, dict) else {}
2249 merged = {**default_config, **vector_store_config}
2250 if self._feature_override_enabled("vector_store"):
2251 merged["enabled"] = True
2252 return merged
2254 def get_vector_store_enabled(self) -> bool:
2255 """Return whether the vector store is enabled (default False)."""
2256 return bool(self.get_vector_store_config()["enabled"])
2258 def get_vector_store_replica_regions(self) -> list[str]:
2259 """Return the effective replica region list for the vector store.
2261 The configured ``replica_regions`` when non-empty, otherwise the
2262 regional deployment list — in both cases with the global region
2263 removed, because the primary table lives there and a global table
2264 cannot replicate into its own region. May legitimately be empty
2265 (single-region deployments get a single-region global table).
2266 """
2267 config = self.get_vector_store_config()
2268 configured = [str(region) for region in config["replica_regions"]]
2269 candidates = configured or self.get_regions()
2270 global_region = self.get_global_region()
2271 return [region for region in candidates if region != global_region]
2273 def get_tags(self) -> dict[str, str]:
2274 """Get common tags from configuration"""
2275 return self.app.node.try_get_context("tags") or {}
2277 def validate_region_availability(self, region: str) -> bool:
2278 """Validate that a region is available in the current AWS account"""
2279 try:
2280 ec2 = boto3.client("ec2", region_name=region)
2281 ec2.describe_regions(RegionNames=[region])
2282 return True
2283 except Exception as e:
2284 logger.debug("Region %s not available: %s", region, e)
2285 return False
2287 def get_available_regions(self) -> list[str]:
2288 """Get list of available AWS regions for the current account"""
2289 try:
2290 ec2 = boto3.client("ec2")
2291 response = ec2.describe_regions()
2292 return [region["RegionName"] for region in response["Regions"]]
2293 except Exception as e:
2294 logger.debug("Failed to list regions, using defaults: %s", e)
2295 return list(self.VALID_REGIONS)