Coverage for gco / stacks / global_stack.py: 100.00%
348 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""
2Global shared-resources stack with optional commercial-partition routing.
4This stack always creates partition-wide shared state: SSM registries, DynamoDB
5tables, ECR/S3 resources, backups, and optional capacity history. In the
6commercial ``aws`` partition it also creates AWS Global Accelerator, its
7TCP/443 listener, and one endpoint group per workload region. Other AWS
8partitions omit those unavailable resources and use regional IAM APIs.
10Regional ALB registration is performed separately by each regional stack only
11when the accelerator topology exists.
12"""
14import json
15from typing import Any
17from aws_cdk import (
18 CfnOutput,
19 Duration,
20 Fn,
21 RemovalPolicy,
22 Stack,
23)
24from aws_cdk import aws_backup as backup
25from aws_cdk import aws_dynamodb as dynamodb
26from aws_cdk import aws_ecr as ecr
27from aws_cdk import aws_events as events
28from aws_cdk import aws_globalaccelerator as ga
29from aws_cdk import aws_iam as iam
30from aws_cdk import aws_kms as kms
31from aws_cdk import aws_lambda as lambda_
32from aws_cdk import aws_s3 as s3
33from aws_cdk import aws_ssm as ssm
34from constructs import Construct
36from gco.config.config_loader import ConfigLoader
37from gco.stacks.constants import (
38 LAMBDA_PYTHON_RUNTIME,
39 cluster_shared_ssm_parameter_prefix,
40)
41from gco.stacks.lambda_assets import lambda_source_code
43# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
44# Generated at (UTC): 2026-10-16T16:57:50Z
45# Generated from Git commit: 3458f76ef1541be718c00458b8ecc8f7c770c6ad
46# Flowchart(s) generated from this file:
47# * ``GCOGlobalStack.__init__`` -> ``diagrams/code_diagrams/gco/stacks/global_stack.GCOGlobalStack___init__.html``
48# (PNG: ``diagrams/code_diagrams/gco/stacks/global_stack.GCOGlobalStack___init__.png``)
49# * ``GCOGlobalStack._create_image_replication_rule`` -> ``diagrams/code_diagrams/gco/stacks/global_stack.GCOGlobalStack__create_image_replication_rule.html``
50# (PNG: ``diagrams/code_diagrams/gco/stacks/global_stack.GCOGlobalStack__create_image_replication_rule.png``)
51# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
52# <pyflowchart-code-diagram> END
55# Default values for the ``images`` cdk.json block. The defaults match the
56# documented retention posture: repos survive a stack destroy by default
57# (``retain``), non-empty repos block destroy unless the operator explicitly
58# flips ``empty_on_delete`` to true, lifecycle keeps the latest 20 tagged
59# images and expires untagged ones after 7 days, and replication is enabled
60# by default to every deployed region.
61_IMAGES_DEFAULT_REMOVAL_POLICY = "retain"
62_IMAGES_DEFAULT_EMPTY_ON_DELETE = False
63_IMAGES_DEFAULT_KEEP_TAGGED = 20
64_IMAGES_DEFAULT_EXPIRE_UNTAGGED_DAYS = 7
65_IMAGES_DEFAULT_REPLICATION_ENABLED = True
66_IMAGES_DEFAULT_REPLICATION_DESTINATIONS = "all_deployed_regions"
68_IMAGES_VALID_REMOVAL_POLICIES = ("retain", "destroy")
71def _parse_images_config(cdk_context: dict[str, Any] | None) -> dict[str, Any]:
72 """Parse the ``images`` block from cdk.json with defaults applied.
74 Returns a normalized dict shape that the rest of the global stack
75 can consume without re-parsing. Validates ``removal_policy`` against
76 the set ``{"retain", "destroy"}`` and ``replication.destinations``
77 against either the literal string ``"all_deployed_regions"`` or a
78 ``list[str]``.
80 Args:
81 cdk_context: The dict returned by ``self.node.try_get_context("images")``.
82 ``None`` (the key being absent) is equivalent to an empty dict.
84 Returns:
85 A dict with keys ``removal_policy``, ``empty_on_delete``,
86 ``lifecycle`` (with ``keep_tagged`` and ``expire_untagged_days``),
87 and ``replication`` (with ``enabled`` and ``destinations``).
88 """
89 raw = cdk_context or {}
91 removal_policy = raw.get("removal_policy", _IMAGES_DEFAULT_REMOVAL_POLICY)
92 if not isinstance(removal_policy, str) or removal_policy not in _IMAGES_VALID_REMOVAL_POLICIES:
93 raise ValueError(
94 f"images.removal_policy must be 'retain' or 'destroy', got {removal_policy!r}"
95 )
97 empty_on_delete = bool(raw.get("empty_on_delete", _IMAGES_DEFAULT_EMPTY_ON_DELETE))
99 lifecycle_raw = raw.get("lifecycle") or {}
100 if not isinstance(lifecycle_raw, dict):
101 raise ValueError(f"images.lifecycle must be a mapping, got {type(lifecycle_raw).__name__}")
102 keep_tagged = int(lifecycle_raw.get("keep_tagged", _IMAGES_DEFAULT_KEEP_TAGGED))
103 expire_untagged_days = int(
104 lifecycle_raw.get("expire_untagged_days", _IMAGES_DEFAULT_EXPIRE_UNTAGGED_DAYS)
105 )
107 replication_raw = raw.get("replication") or {}
108 if not isinstance(replication_raw, dict):
109 raise ValueError(
110 f"images.replication must be a mapping, got {type(replication_raw).__name__}"
111 )
112 replication_enabled = bool(replication_raw.get("enabled", _IMAGES_DEFAULT_REPLICATION_ENABLED))
113 destinations = replication_raw.get("destinations", _IMAGES_DEFAULT_REPLICATION_DESTINATIONS)
114 if isinstance(destinations, str):
115 if destinations != _IMAGES_DEFAULT_REPLICATION_DESTINATIONS:
116 raise ValueError(
117 "images.replication.destinations must be the string "
118 f"{_IMAGES_DEFAULT_REPLICATION_DESTINATIONS!r} or a list of region names, "
119 f"got {destinations!r}"
120 )
121 elif isinstance(destinations, list):
122 if not all(isinstance(item, str) for item in destinations):
123 raise ValueError(
124 "images.replication.destinations list must contain only region name strings"
125 )
126 else:
127 raise ValueError(
128 "images.replication.destinations must be the string "
129 f"{_IMAGES_DEFAULT_REPLICATION_DESTINATIONS!r} or a list of region names, "
130 f"got {type(destinations).__name__}"
131 )
133 return {
134 "removal_policy": removal_policy,
135 "empty_on_delete": empty_on_delete,
136 "lifecycle": {
137 "keep_tagged": keep_tagged,
138 "expire_untagged_days": expire_untagged_days,
139 },
140 "replication": {
141 "enabled": replication_enabled,
142 "destinations": destinations,
143 },
144 }
147class GCOGlobalStack(Stack):
148 """Global shared resources with optional AWS Global Accelerator.
150 This stack must be deployed before regional stacks. In the commercial
151 ``aws`` partition, regional stacks register their ALBs with endpoint groups
152 created here. Other partitions omit the accelerator topology.
154 Attributes:
155 accelerator: Optional Global Accelerator resource.
156 listener: Optional TCP/443 listener.
157 endpoint_groups: Region-to-endpoint-group mapping; empty without GA.
158 templates_table: DynamoDB table for job templates.
159 webhooks_table: DynamoDB table for webhooks.
160 missions_table: DynamoDB table for mission session state.
161 """
163 def __init__(
164 self, scope: Construct, construct_id: str, config: ConfigLoader, **kwargs: Any
165 ) -> None:
166 super().__init__(scope, construct_id, **kwargs)
168 self.config = config
169 self.endpoint_groups: dict[str, ga.EndpointGroup] = {}
170 supports_global_accelerator = getattr(config, "supports_global_accelerator", None)
171 self.global_accelerator_enabled = (
172 bool(supports_global_accelerator()) if callable(supports_global_accelerator) else True
173 )
175 ga_config = self.config.get_global_accelerator_config()
177 # Store the accelerator name for reference by other stacks. Defaults
178 # to ``<project_name>-accelerator`` when not pinned in cdk.json so a
179 # second deployment gets its own project-scoped name (#139); an explicit
180 # ``global_accelerator.name`` still overrides.
181 self.accelerator_name = (
182 ga_config.get("name") or f"{self.config.get_project_name()}-accelerator"
183 )
185 # Create DynamoDB tables for templates and webhooks
186 self._create_dynamodb_tables()
188 # Create S3 bucket for model weights
189 self._create_model_bucket()
191 # Create always-on Cluster_Shared_Bucket + KMS key + SSM parameters.
192 # These run unconditionally (no feature toggle) — they are consumed by
193 # every Regional_Stack and, when analytics is enabled, by GCOAnalyticsStack.
194 self._create_cluster_shared_kms_key()
195 self._create_cluster_shared_bucket()
196 self._publish_cluster_shared_bucket_ssm_params()
198 # Create AWS Backup plan for DynamoDB tables
199 self._create_backup_plan()
201 # Container image registry — parses the cdk.json ``images`` block,
202 # provisions the optional ECR replication rule for ``gco/*`` repos,
203 # and creates the lookup-or-create custom resource Lambda that
204 # ``cli images init`` will invoke per-repo on demand. The Lambda
205 # construct is created here regardless of replication settings so
206 # the function ARN is available for downstream invocations.
207 self.images_config = _parse_images_config(self.node.try_get_context("images"))
208 self._create_image_replication_rule()
209 self._create_image_lookup_lambda()
211 # Optional Historical Capacity Surface add-on (gated by historical.enabled
212 # in cdk.json). Folded into the global stack rather than a separate stack
213 # so it reuses the global DynamoDB/encryption conventions.
214 if self.config.get_capacity_history_enabled():
215 self._create_capacity_poller()
217 # Mission memory add-on (gated by mission_memory.enabled in cdk.json,
218 # ON by default): a DynamoDB table with a vector index over embedded
219 # mission directives, giving the mission engine recall across sessions.
220 if self.config.get_mission_memory_enabled():
221 self._create_mission_memory()
223 # Vector store add-on (gated by vector_store.enabled in cdk.json,
224 # OFF by default): a globally replicated DynamoDB table with a vector
225 # index over an S3-ingested document corpus, giving workloads in every
226 # deployment region local-latency semantic search. The ingest Lambda
227 # watches the cluster-shared bucket's corpus prefix and writes
228 # embedded chunks to the table's primary replica.
229 if self.config.get_vector_store_enabled():
230 self._create_vector_store()
231 self._create_vector_ingest()
233 # Global Accelerator is available only in the commercial ``aws``
234 # partition. Other coherent AWS partitions retain all shared resources
235 # and use their IAM-authenticated regional API bridges directly rather
236 # than synthesizing an unavailable global service.
237 self.accelerator: ga.Accelerator | None = None
238 self.listener: ga.Listener | None = None
239 self.accelerator_id: str | None = None
240 if self.global_accelerator_enabled:
241 # There is deliberately no port-80 listener, so the backend path
242 # cannot downgrade from authenticated TLS.
243 self.accelerator = ga.Accelerator(
244 self,
245 "GCOAccelerator",
246 accelerator_name=self.accelerator_name,
247 enabled=True,
248 )
249 self.accelerator_id = Fn.select(
250 1,
251 Fn.split("/", self.accelerator.accelerator_arn),
252 )
253 self.listener = self.accelerator.add_listener(
254 "GCOListener",
255 port_ranges=[ga.PortRange(from_port=443, to_port=443)],
256 protocol=ga.ConnectionProtocol.TCP,
257 client_affinity=self._resolve_client_affinity(ga_config),
258 )
259 for region in self.config.get_regions():
260 self._create_endpoint_group(region)
261 self._create_outputs()
263 # Optional capacity-driven traffic-dial controller, gated by
264 # global_accelerator.traffic_dial.enabled (OFF by default). It
265 # lives inside the accelerator guard on purpose: outside the
266 # commercial partition there is nothing to dial.
267 traffic_dial_config = ga_config.get("traffic_dial") or {}
268 if bool(traffic_dial_config.get("enabled", False)):
269 self._create_traffic_dial_controller(ga_config)
271 # Apply cdk-nag suppressions
272 self._apply_nag_suppressions()
274 @staticmethod
275 def _resolve_client_affinity(ga_config: dict[str, Any]) -> ga.ClientAffinity:
276 """Map the ``client_affinity`` cdk.json knob to a CDK enum.
278 Global Accelerator supports two client-affinity modes:
280 - ``NONE`` (default): each new connection may be routed to any
281 healthy endpoint, maximising even load distribution.
282 - ``SOURCE_IP``: connections from the same source IP are pinned to
283 the same endpoint, which is useful for workloads that keep
284 per-client state on a single region.
286 The value is validated up front by
287 ``ConfigLoader._validate_global_accelerator_config`` so an unknown
288 string never reaches this point; the fallback to ``NONE`` keeps the
289 stack synthesizable even when the key is omitted entirely.
290 """
291 affinity = str(ga_config.get("client_affinity", "NONE")).upper()
292 mapping = {
293 "NONE": ga.ClientAffinity.NONE,
294 "SOURCE_IP": ga.ClientAffinity.SOURCE_IP,
295 }
296 return mapping.get(affinity, ga.ClientAffinity.NONE)
298 def _create_traffic_dial_controller(self, ga_config: dict[str, Any]) -> None:
299 """Create the optional capacity-driven traffic-dial controller.
301 An EventBridge-scheduled Lambda (commercial ``aws`` partition only,
302 opt-in via ``global_accelerator.traffic_dial.enabled``) that reads
303 each region's ``GCO/HealthMonitor`` ``ClusterHealthy`` signal and
304 converges every endpoint group's TrafficDialPercentage toward the
305 observed health of its region. ``monitor`` mode only publishes
306 decisions (``GCO/TrafficDial`` CloudWatch metrics plus the
307 ``/{project}/traffic-dial/state`` SSM parameter); ``enforce`` mode
308 additionally applies them via UpdateEndpointGroup — and only
309 ``enforce`` mode is granted that IAM action.
311 The safety rails live in the handler: a configurable dial floor,
312 per-run step limiting, a last-healthy-region guard that never leaves
313 every endpoint group dialed below 100, a hold on missing telemetry
314 (absent metrics must never look like ideal health), a skip while the
315 accelerator is mid-deployment, and per-region manual overrides
316 (``gco capacity traffic-dial set``) that the controller respects.
318 This is deliberately a single writer next to the accelerator it
319 manages: regional writers would need leader election and new
320 cross-region IAM to mutate one global resource.
321 """
322 from aws_cdk import aws_events_targets as events_targets
323 from aws_cdk import aws_sqs as sqs
325 from gco.stacks.nag_suppressions import acknowledge_nag_findings
327 if self.listener is None:
328 raise RuntimeError("Traffic-dial controller requires the Global Accelerator listener")
330 project_name = self.config.get_project_name()
331 dial_config = ga_config.get("traffic_dial") or {}
332 mode = str(dial_config.get("mode", "monitor")).lower()
333 interval_minutes = int(dial_config.get("interval_minutes", 5))
334 lookback_minutes = int(dial_config.get("lookback_minutes", 15))
335 min_dial_percentage = int(dial_config.get("min_dial_percentage", 10))
336 max_step_percentage = int(dial_config.get("max_step_percentage", 20))
337 full_health_percentage = int(dial_config.get("full_health_percentage", 95))
339 controller_role = iam.Role(
340 self,
341 "TrafficDialControllerRole",
342 assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
343 managed_policies=[
344 iam.ManagedPolicy.from_aws_managed_policy_name(
345 "service-role/AWSLambdaBasicExecutionRole"
346 )
347 ],
348 )
350 # Global Accelerator APIs do not support resource-level IAM scoping.
351 # UpdateEndpointGroup — the only mutating action — is granted solely
352 # in enforce mode; monitor mode stays read-only against GA.
353 ga_actions = [
354 "globalaccelerator:DescribeAccelerator",
355 "globalaccelerator:ListEndpointGroups",
356 ]
357 if mode == "enforce":
358 ga_actions.append("globalaccelerator:UpdateEndpointGroup")
359 controller_role.add_to_policy(
360 iam.PolicyStatement(
361 effect=iam.Effect.ALLOW,
362 actions=sorted(ga_actions),
363 resources=["*"],
364 )
365 )
366 # Cross-region read of each cluster's ClusterHealthy metric.
367 # GetMetricData does not support resource-level permissions.
368 controller_role.add_to_policy(
369 iam.PolicyStatement(
370 effect=iam.Effect.ALLOW,
371 actions=["cloudwatch:GetMetricData"],
372 resources=["*"],
373 )
374 )
375 controller_role.add_to_policy(
376 iam.PolicyStatement(
377 effect=iam.Effect.ALLOW,
378 actions=["cloudwatch:PutMetricData"],
379 resources=["*"],
380 conditions={"StringEquals": {"cloudwatch:namespace": "GCO/TrafficDial"}},
381 )
382 )
383 # State publication plus the manual-override parameters written by
384 # `gco capacity traffic-dial set`, scoped to this project's dial tree.
385 controller_role.add_to_policy(
386 iam.PolicyStatement(
387 effect=iam.Effect.ALLOW,
388 actions=["ssm:GetParameter", "ssm:GetParametersByPath", "ssm:PutParameter"],
389 resources=[
390 f"arn:{self.partition}:ssm:{self.region}:{self.account}:"
391 f"parameter/{project_name}/traffic-dial",
392 f"arn:{self.partition}:ssm:{self.region}:{self.account}:"
393 f"parameter/{project_name}/traffic-dial/*",
394 ],
395 )
396 )
398 self.traffic_dial_lambda = lambda_.Function(
399 self,
400 "TrafficDialControllerFunction",
401 runtime=getattr(lambda_.Runtime, LAMBDA_PYTHON_RUNTIME),
402 handler="handler.lambda_handler",
403 code=lambda_source_code("lambda/traffic-dial-controller"),
404 timeout=Duration.minutes(5),
405 memory_size=256,
406 role=controller_role,
407 environment={
408 "LISTENER_ARN": self.listener.listener_arn,
409 "PROJECT_NAME": project_name,
410 "MODE": mode,
411 "REGIONS": ",".join(self.config.get_regions()),
412 "LOOKBACK_MINUTES": str(lookback_minutes),
413 "MIN_DIAL_PERCENTAGE": str(min_dial_percentage),
414 "MAX_STEP_PERCENTAGE": str(max_step_percentage),
415 "FULL_HEALTH_PERCENTAGE": str(full_health_percentage),
416 },
417 tracing=lambda_.Tracing.ACTIVE,
418 description=(
419 "Capacity-driven Global Accelerator traffic-dial controller "
420 f"({mode} mode): converges per-region TrafficDialPercentage "
421 "toward each cluster's ClusterHealthy signal."
422 ),
423 )
425 dial_dlq = sqs.Queue(
426 self,
427 "TrafficDialRuleDlq",
428 retention_period=Duration.days(14),
429 enforce_ssl=True,
430 encryption=sqs.QueueEncryption.SQS_MANAGED,
431 removal_policy=RemovalPolicy.DESTROY,
432 )
433 events.Rule(
434 self,
435 "TrafficDialSchedule",
436 description=(
437 f"Traffic-dial controller for {project_name} "
438 f"({mode} mode, every {interval_minutes} min)"
439 ),
440 schedule=events.Schedule.rate(Duration.minutes(interval_minutes)),
441 targets=[
442 events_targets.LambdaFunction(
443 self.traffic_dial_lambda, dead_letter_queue=dial_dlq, retry_attempts=2
444 )
445 ],
446 )
448 acknowledge_nag_findings(
449 controller_role,
450 [
451 {
452 "id": "AwsSolutions-IAM4",
453 "reason": (
454 "AWSLambdaBasicExecutionRole provides the standard CloudWatch "
455 "Logs permissions every Lambda needs."
456 ),
457 },
458 {
459 "id": "AwsSolutions-IAM5",
460 "reason": (
461 "The Global Accelerator Describe/List/Update APIs and CloudWatch "
462 "GetMetricData do not support resource-level permissions and "
463 "require a wildcard resource; PutMetricData is namespace-"
464 "conditioned and the SSM wildcard is scoped to this project's "
465 "traffic-dial parameter tree."
466 ),
467 "appliesTo": [
468 "Resource::*",
469 f"Resource::arn:<AWS::Partition>:ssm:{self.region}:"
470 f"<AWS::AccountId>:parameter/{project_name}/traffic-dial/*",
471 ],
472 },
473 ],
474 )
475 acknowledge_nag_findings(
476 dial_dlq,
477 [
478 {
479 "id": "AwsSolutions-SQS3",
480 "reason": (
481 "This queue is the dead-letter queue for the TrafficDialSchedule "
482 "EventBridge rule; a DLQ for a DLQ is circular."
483 ),
484 },
485 {
486 "id": "Serverless-SQSRedrivePolicy",
487 "reason": (
488 "This queue is itself the dead-letter queue for the "
489 "TrafficDialSchedule EventBridge rule, so it does not need its "
490 "own redrive policy; a DLQ for a DLQ is circular."
491 ),
492 },
493 ],
494 )
496 def _create_capacity_poller(self) -> None:
497 """Create the optional Historical Capacity Surface add-on.
499 This is an optional add-on to the global stack (not a separate stack),
500 gated by ``historical.enabled`` in cdk.json. It provisions a DynamoDB
501 time-series table plus an EventBridge-scheduled Lambda that snapshots
502 capacity signals (spot score, spot price, AZ coverage, capacity-block
503 availability) for the watched instance types across the enabled regions,
504 reusing this stack's DynamoDB/encryption conventions.
505 """
506 from aws_cdk import aws_events_targets as events_targets
507 from aws_cdk import aws_sqs as sqs
509 # Function-local on purpose: the capacity->metric-field naming rule and
510 # the Spot Placement Score pool catalog each have exactly one source of
511 # truth (cli/capacity/history.py and scripts/accelerator_catalog.py),
512 # and the poller Lambda cannot import either, so this stack serializes
513 # them into its environment at synth time. Synthesis always runs from a
514 # repository checkout (the CDK app and Lambda assets require one), but
515 # the installed wheel ships gco.stacks without scripts/, so a
516 # module-level import would break mere importability of this module.
517 from cli.capacity.history import metric_field_for_target_capacity
518 from gco.stacks.nag_suppressions import acknowledge_nag_findings
519 from scripts.accelerator_catalog import INSTANCE_POOLS
521 project_name = self.config.get_project_name()
522 historical = self.config.get_capacity_history_config()
523 retention_days = int(historical["retention_days"])
524 poll_interval_minutes = int(historical["poll_interval_minutes"])
525 watch_instance_types = list(historical["watch_instance_types"])
526 enabled_regions = list(historical["enabled_regions"]) or self.config.get_regions()
527 # Capacity Block probe durations. ``get_capacity_history_config`` always
528 # supplies both keys (defaults merged in), so index directly like the
529 # sibling fields above.
530 block_duration_hours = int(historical["capacity_block_duration_hours"])
531 long_block_duration_hours = int(historical["capacity_block_long_duration_hours"])
532 # Serialized with compact separators: Lambda caps the whole environment
533 # at 4 KB, and the pool catalog plus the watch list are the two big
534 # values (~2.9 KB total today). ``spot_score_target_capacities`` is
535 # validated against the supported set at config load, so the naming
536 # function cannot raise here for a config that passed validation.
537 spot_score_target_capacities_env = json.dumps(
538 [
539 {
540 "target_capacity": capacity,
541 "metric_field": metric_field_for_target_capacity(capacity),
542 }
543 for capacity in historical["spot_score_target_capacities"]
544 ],
545 separators=(",", ":"),
546 )
547 instance_pools_env = json.dumps(
548 [{"name": pool.name, "members": list(pool.members)} for pool in INSTANCE_POOLS],
549 separators=(",", ":"),
550 )
552 self.capacity_history_table = dynamodb.Table(
553 self,
554 "CapacityHistoryTable",
555 table_name=f"{project_name}-capacity-history",
556 partition_key=dynamodb.Attribute(name="pk", type=dynamodb.AttributeType.STRING),
557 sort_key=dynamodb.Attribute(name="sk", type=dynamodb.AttributeType.STRING),
558 billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
559 removal_policy=RemovalPolicy.DESTROY,
560 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
561 point_in_time_recovery_enabled=True
562 ),
563 encryption=dynamodb.TableEncryption.AWS_MANAGED,
564 time_to_live_attribute="ttl",
565 )
566 self.capacity_history_table.add_global_secondary_index(
567 index_name="by-timestamp",
568 partition_key=dynamodb.Attribute(
569 name="instance_type", type=dynamodb.AttributeType.STRING
570 ),
571 sort_key=dynamodb.Attribute(name="sk", type=dynamodb.AttributeType.STRING),
572 projection_type=dynamodb.ProjectionType.ALL,
573 )
575 poller_role = iam.Role(
576 self,
577 "CapacityPollerRole",
578 assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
579 managed_policies=[
580 iam.ManagedPolicy.from_aws_managed_policy_name(
581 "service-role/AWSLambdaBasicExecutionRole"
582 )
583 ],
584 )
585 poller_role.add_to_policy(
586 iam.PolicyStatement(
587 effect=iam.Effect.ALLOW,
588 actions=["dynamodb:PutItem", "dynamodb:BatchWriteItem"],
589 resources=[
590 self.capacity_history_table.table_arn,
591 f"{self.capacity_history_table.table_arn}/index/*",
592 ],
593 )
594 )
595 poller_role.add_to_policy(
596 iam.PolicyStatement(
597 effect=iam.Effect.ALLOW,
598 actions=[
599 "ec2:DescribeSpotPriceHistory",
600 "ec2:GetSpotPlacementScores",
601 "ec2:DescribeCapacityBlockOfferings",
602 "ec2:DescribeCapacityReservations",
603 "ec2:DescribeAvailabilityZones",
604 # The poller's region-enablement pre-check uses
605 # DescribeRegions(AllRegions=True, RegionNames=[region])
606 # from the Lambda's default-Region client. Discovered live:
607 # without this grant the probe fails with
608 # UnauthorizedOperation for every configured Region.
609 "ec2:DescribeRegions",
610 ],
611 resources=["*"],
612 )
613 )
615 self.capacity_poller_lambda = lambda_.Function(
616 self,
617 "CapacityPollerFunction",
618 runtime=getattr(lambda_.Runtime, LAMBDA_PYTHON_RUNTIME),
619 handler="handler.lambda_handler",
620 code=lambda_source_code("lambda/capacity-poller"),
621 timeout=Duration.minutes(14),
622 memory_size=256,
623 role=poller_role,
624 environment={
625 "CAPACITY_HISTORY_TABLE_NAME": self.capacity_history_table.table_name,
626 "WATCH_INSTANCE_TYPES": ",".join(watch_instance_types),
627 "ENABLED_REGIONS": ",".join(enabled_regions),
628 "CAPACITY_HISTORY_RETENTION_DAYS": str(retention_days),
629 "CAPACITY_BLOCK_DURATION_HOURS": str(block_duration_hours),
630 "CAPACITY_BLOCK_LONG_DURATION_HOURS": str(long_block_duration_hours),
631 "SPOT_SCORE_TARGET_CAPACITIES": spot_score_target_capacities_env,
632 "INSTANCE_POOLS": instance_pools_env,
633 },
634 tracing=lambda_.Tracing.ACTIVE,
635 description=(
636 "Historical Capacity Surface poller (optional global-stack add-on): "
637 "snapshots capacity signals into the capacity-history table."
638 ),
639 )
641 poller_dlq = sqs.Queue(
642 self,
643 "CapacityPollerRuleDlq",
644 retention_period=Duration.days(14),
645 enforce_ssl=True,
646 encryption=sqs.QueueEncryption.SQS_MANAGED,
647 removal_policy=RemovalPolicy.DESTROY,
648 )
649 events.Rule(
650 self,
651 "CapacityPollerSchedule",
652 description=(
653 f"Capacity poller for {project_name} history surface "
654 f"(every {poll_interval_minutes} min)"
655 ),
656 schedule=events.Schedule.rate(Duration.minutes(poll_interval_minutes)),
657 targets=[
658 events_targets.LambdaFunction(
659 self.capacity_poller_lambda, dead_letter_queue=poller_dlq, retry_attempts=2
660 )
661 ],
662 )
664 ssm.StringParameter(
665 self,
666 "CapacityHistoryTableNameParam",
667 parameter_name=f"/{project_name}/capacity-history-table-name",
668 string_value=self.capacity_history_table.table_name,
669 description="DynamoDB table name for historical capacity snapshots",
670 )
671 CfnOutput(
672 self,
673 "CapacityHistoryTableName",
674 value=self.capacity_history_table.table_name,
675 description="DynamoDB table name for historical capacity snapshots",
676 export_name=f"{project_name}-capacity-history-table-name",
677 )
678 CfnOutput(
679 self,
680 "CapacityHistoryTableArn",
681 value=self.capacity_history_table.table_arn,
682 description="DynamoDB table ARN for historical capacity snapshots",
683 export_name=f"{project_name}-capacity-history-table-arn",
684 )
686 acknowledge_nag_findings(
687 poller_role,
688 [
689 {
690 "id": "AwsSolutions-IAM4",
691 "reason": (
692 "AWSLambdaBasicExecutionRole provides the standard CloudWatch "
693 "Logs permissions every Lambda needs."
694 ),
695 },
696 {
697 "id": "AwsSolutions-IAM5",
698 "reason": (
699 "The EC2 capacity describe/get APIs (DescribeSpotPriceHistory, "
700 "GetSpotPlacementScores, DescribeCapacityBlockOfferings, "
701 "DescribeCapacityReservations, DescribeAvailabilityZones) do not "
702 "support resource-level permissions and require a wildcard "
703 "resource. The DynamoDB index wildcard is scoped to this table's "
704 "own indexes."
705 ),
706 "appliesTo": [
707 "Resource::*",
708 "Resource::<CapacityHistoryTable506A0FBA.Arn>/index/*",
709 ],
710 },
711 ],
712 )
713 acknowledge_nag_findings(
714 poller_dlq,
715 [
716 {
717 "id": "AwsSolutions-SQS3",
718 "reason": (
719 "This queue is the dead-letter queue for the "
720 "CapacityPollerSchedule EventBridge rule; a DLQ for a DLQ is "
721 "circular."
722 ),
723 },
724 {
725 "id": "Serverless-SQSRedrivePolicy",
726 "reason": (
727 "This queue is itself the dead-letter queue for the "
728 "CapacityPollerSchedule EventBridge rule, so it does not need "
729 "its own redrive policy; a DLQ for a DLQ is circular."
730 ),
731 },
732 ],
733 )
734 acknowledge_nag_findings(
735 self.capacity_history_table,
736 [
737 {
738 "id": "HIPAA.Security-DynamoDBInBackupPlan",
739 "reason": (
740 "The capacity-history table holds ephemeral, reconstructable "
741 "telemetry snapshots with a TTL (default 90 days) and "
742 "point-in-time recovery enabled. The poller re-collects this "
743 "data continuously, so an AWS Backup plan is unnecessary for "
744 "this optional add-on."
745 ),
746 },
747 {
748 "id": "NIST.800.53.R5-DynamoDBInBackupPlan",
749 "reason": (
750 "The capacity-history table holds ephemeral, reconstructable "
751 "telemetry snapshots with a TTL (default 90 days) and "
752 "point-in-time recovery enabled. The poller re-collects this "
753 "data continuously, so an AWS Backup plan is unnecessary for "
754 "this optional add-on."
755 ),
756 },
757 ],
758 )
760 def _create_mission_memory(self) -> None:
761 """Create the mission-memory table and its directive vector index.
763 Gated by ``mission_memory.enabled`` in cdk.json (ON by default).
764 Completed missions write one memory item (the operator directive, its
765 embedding, and the model-written lessons / recommended follow-ups);
766 later missions retrieve the most similar past directives into their
767 sampling prompts. Memory is shared institutional memory: items are
768 keyed by session only, so every operator in the account sees every
769 operator's mission lessons.
771 The vector index CANNOT be expressed declaratively — CloudFormation's
772 ``AWS::DynamoDB::Table`` has no vector-index property at all, and CDK
773 has no L1/L2 support — so it is created by an ``AwsCustomResource``
774 calling the DynamoDB ``UpdateTable`` control-plane API against the
775 freshly created (and therefore empty — no backfill wait needed)
776 table. ``UpdateTable`` returns before the index reaches ACTIVE;
777 runtime consumers tolerate a not-yet-ready index as part of their
778 best-effort posture. Index parameters (dimensions, distance function,
779 the INCLUDE projection) are one-way doors: immutable after creation.
780 """
781 from aws_cdk import custom_resources as cr
783 project_name = self.config.get_project_name()
784 memory_config = self.config.get_mission_memory_config()
785 dimensions = int(memory_config["dimensions"])
786 distance_function = str(memory_config["distance_function"])
787 index_name = "directive-embedding-index"
789 self.mission_memory_table = dynamodb.Table(
790 self,
791 "MissionMemoryTable",
792 table_name=f"{project_name}-mission-memory",
793 partition_key=dynamodb.Attribute(
794 name="session_id",
795 type=dynamodb.AttributeType.STRING,
796 ),
797 # Vector indexes require on-demand billing.
798 billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
799 removal_policy=RemovalPolicy.DESTROY,
800 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
801 point_in_time_recovery_enabled=True
802 ),
803 encryption=dynamodb.TableEncryption.AWS_MANAGED,
804 time_to_live_attribute="ttl",
805 )
807 # Shared, pre-created execution role for the singleton
808 # AwsCustomResource Lambda; this feature contributes its own
809 # table-scoped statement (see _vector_index_custom_resource_role).
810 index_role = self._vector_index_custom_resource_role()
811 index_role.add_to_policy(
812 iam.PolicyStatement(
813 effect=iam.Effect.ALLOW,
814 actions=["dynamodb:UpdateTable", "dynamodb:DescribeTable"],
815 resources=[self.mission_memory_table.table_arn],
816 )
817 )
819 # CreateVectorIndexAction is FLAT — IndexName, VectorAttribute
820 # ({AttributeName}), Dimensions, DistanceFunction, SearchSchema,
821 # Projection all sit at the same level. Verified against the live
822 # service (its validation names create.vectorAttribute /
823 # create.dimensions / create.distanceFunction as the required
824 # members) and against botocore's UpdateTable model; an earlier
825 # nested "VectorConfiguration" draft shape deployed as nulls and
826 # failed create. tests/test_mission_memory_stack.py pins this
827 # payload against the botocore model.
828 vector_index_updates = [
829 {
830 "Create": {
831 "IndexName": index_name,
832 "VectorAttribute": {"AttributeName": "directive_embedding"},
833 "Dimensions": dimensions,
834 "DistanceFunction": distance_function,
835 "SearchSchema": [
836 {
837 "AttributeName": "final_verdict",
838 "SearchSchemaElementType": "INLINE_FILTER",
839 }
840 ],
841 "Projection": {
842 "ProjectionType": "INCLUDE",
843 "NonKeyAttributes": [
844 "directive",
845 "lessons",
846 "recommended_followups",
847 "final_verdict",
848 "verdict_reason",
849 "iteration_count",
850 "completed_at",
851 ],
852 },
853 }
854 }
855 ]
856 vector_index = cr.AwsCustomResource(
857 self,
858 "MissionMemoryVectorIndex",
859 on_create=cr.AwsSdkCall(
860 service="DynamoDB",
861 action="updateTable",
862 parameters={
863 "TableName": self.mission_memory_table.table_name,
864 # Any attribute an index references must be declared in
865 # AttributeDefinitions, and for an index added through
866 # UpdateTable the declaration rides in the same call —
867 # the standard add-a-GSI pattern, verified live: without
868 # it the service rejects the SearchSchema with "One
869 # element in SearchSchema is not defined in attribute
870 # definitions". It cannot ride on CreateTable instead;
871 # CreateTable rejects definitions unused by key schemas.
872 "AttributeDefinitions": [
873 {"AttributeName": "final_verdict", "AttributeType": "S"}
874 ],
875 "VectorIndexUpdates": vector_index_updates,
876 },
877 physical_resource_id=cr.PhysicalResourceId.of(
878 f"{project_name}-mission-memory-vector-index"
879 ),
880 ),
881 # DELIBERATELY NO on_delete, matching the vector store (see the
882 # long comment on VectorStoreIndex for the live incident). The
883 # table carries RemovalPolicy.DESTROY, so teardown deletes it and
884 # the index goes with it — an explicit index delete only buys an
885 # UPDATING window for the next resource to trip over. This table
886 # is single-region today, so the replica-delete deadlock that
887 # wedged the vector store cannot fire here; the hazard is latent,
888 # not absent, and it would become live the day this table gains a
889 # replica. Keeping both call sites identical means the fix cannot
890 # be half-applied. tests/test_mission_memory_stack.py pins the
891 # absence.
892 # The Lambda behind AwsCustomResource ships the runtime's bundled
893 # AWS SDK for JavaScript, which lags the API models: a bundled SDK
894 # that predates vector indexes silently DROPS the unknown
895 # VectorIndexUpdates member at serialization, and DynamoDB then
896 # rejects the bare UpdateTable with "At least one of
897 # ProvisionedThroughput, BillingMode, ... is required". boto3
898 # having the API on the developer's machine says nothing about this Lambda's
899 # SDK. install_latest_aws_sdk fetches a current SDK on first
900 # invocation so the member survives. Verified live: the bundled
901 # SDK reproduces the failure, the installed SDK creates the index.
902 # The floating fetch is bounded risk: it runs only on stack
903 # create/delete/update (never on a data path), and an npm failure
904 # falls back to the bundled SDK with a logged warning. Revisit
905 # once the bundled runtime SDK knows VectorIndexUpdates (drop the
906 # flag for determinism), or if Phase 2 multiplies these custom
907 # resources (a shared pinned provider Lambda becomes worth it).
908 install_latest_aws_sdk=True,
909 role=index_role,
910 )
911 # The table must be ACTIVE before UpdateTable can add an index.
912 vector_index.node.add_dependency(self.mission_memory_table)
913 vector_index.node.add_dependency(index_role)
915 # SSM is the established runtime/cross-region discovery contract;
916 # MissionMemoryStore resolves both names lazily the same way
917 # DynamoDBBackend resolves the missions table.
918 ssm.StringParameter(
919 self,
920 "MissionMemoryTableNameParam",
921 parameter_name=f"/{project_name}/mission-memory-table-name",
922 string_value=self.mission_memory_table.table_name,
923 description="DynamoDB table name for mission memory items",
924 )
925 ssm.StringParameter(
926 self,
927 "MissionMemoryIndexNameParam",
928 parameter_name=f"/{project_name}/mission-memory-index-name",
929 string_value=index_name,
930 description="Vector index name over embedded mission directives",
931 )
932 CfnOutput(
933 self,
934 "MissionMemoryTableName",
935 value=self.mission_memory_table.table_name,
936 description="DynamoDB table name for mission memory items",
937 export_name=f"{project_name}-mission-memory-table-name",
938 )
939 CfnOutput(
940 self,
941 "MissionMemoryTableArn",
942 value=self.mission_memory_table.table_arn,
943 description="DynamoDB table ARN for mission memory items",
944 export_name=f"{project_name}-mission-memory-table-arn",
945 )
947 # Backup coverage: join the existing DynamoDB backup plan rather than
948 # acknowledging DynamoDBInBackupPlan twice. The plan is created by
949 # _create_backup_plan earlier in __init__, so the selection exists.
950 self.backup_plan.add_selection(
951 "MissionMemoryTableSelection",
952 resources=[backup.BackupResource.from_dynamo_db_table(self.mission_memory_table)],
953 )
955 def _vector_index_custom_resource_role(self) -> iam.Role:
956 """Return the execution role shared by every vector-index custom resource.
958 ``cr.AwsCustomResource`` Lambdas are a per-stack SINGLETON: every
959 instance shares one provider function, and that function executes
960 with the role of whichever instance is constructed *first*.
961 Live-earned on the first vector-store deploy: with mission memory
962 enabled, the vector-store index's ``UpdateTable`` ran under the
963 mission-memory role and was denied on the vector-store table. One
964 shared role — created lazily by the first feature to need it, with
965 each feature adding only its own table-scoped statements — makes
966 the composition explicit and order-independent, while a deployment
967 with a single feature enabled still carries only that feature's
968 grants. Pre-created (rather than CDK's ``policy=`` auto-role) for
969 the same IAM-propagation reason as the regional stack's shared
970 AwsCustomResource role.
971 """
972 from gco.stacks.nag_suppressions import acknowledge_nag_findings
974 existing: iam.Role | None = getattr(self, "_index_custom_resource_role", None)
975 if existing is not None:
976 return existing
978 role = iam.Role(
979 self,
980 "VectorIndexCustomResourceRole",
981 assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
982 description=(
983 "Shared execution role for the singleton AwsCustomResource "
984 "Lambda that creates DynamoDB vector indexes (mission memory "
985 "and the vector store); each enabled feature contributes only "
986 "its own table-scoped UpdateTable/DescribeTable statement."
987 ),
988 managed_policies=[
989 iam.ManagedPolicy.from_aws_managed_policy_name(
990 "service-role/AWSLambdaBasicExecutionRole"
991 )
992 ],
993 )
994 acknowledge_nag_findings(
995 role,
996 [
997 {
998 "id": "AwsSolutions-IAM4",
999 "reason": (
1000 "AWSLambdaBasicExecutionRole provides the standard CloudWatch "
1001 "Logs permissions every Lambda needs."
1002 ),
1003 },
1004 ],
1005 )
1006 self._index_custom_resource_role = role
1007 return role
1009 def _create_vector_store(self) -> None:
1010 """Create the globally replicated vector-store table and its index.
1012 Gated by ``vector_store.enabled`` in cdk.json (OFF by default — a
1013 replicated table carries real per-region storage and write cost).
1014 The S3-triggered ingest pipeline writes embedded document chunks to
1015 the primary table in this (global) region; DynamoDB global-table
1016 replication fans the items — and the vector index definition —
1017 out to every replica, so workloads query their own region.
1019 The table is a ``TableV2`` (``AWS::DynamoDB::GlobalTable``,
1020 2019.11.21 replication): replicas come from
1021 ``vector_store.replica_regions`` when set, otherwise they follow
1022 ``deployment_regions.regional`` minus this region (the primary is
1023 implicit and a global table cannot replicate into its own region).
1025 The vector index rides the same ``AwsCustomResource`` shape that
1026 mission memory live-earned, because CloudFormation still cannot
1027 express vector indexes (see ``_create_mission_memory``): flat
1028 ``CreateVectorIndexAction``, the INLINE_FILTER attribute declared in
1029 the same ``UpdateTable`` call, ``install_latest_aws_sdk`` so the
1030 member survives serialization, and delete-path error swallowing so
1031 teardown never wedges. Live spike findings for the global-table
1032 variant: the primary accepts the call with ACTIVE replicas, the
1033 index definition propagates to replicas on its own, and the index
1034 takes several minutes to reach ACTIVE — ``UpdateTable`` returns at
1035 call acceptance, so deploys are not blocked, but queries answer
1036 ValidationException until then (the CLI's unavailable-hint covers
1037 it). Index parameters (dimensions, distance function, the INCLUDE
1038 projection) are one-way doors: immutable after creation.
1039 """
1040 from aws_cdk import custom_resources as cr
1042 project_name = self.config.get_project_name()
1043 store_config = self.config.get_vector_store_config()
1044 dimensions = int(store_config["dimensions"])
1045 distance_function = str(store_config["distance_function"])
1046 replica_regions = self.config.get_vector_store_replica_regions()
1047 index_name = "corpus-embedding-index"
1049 self.vector_store_table = dynamodb.TableV2(
1050 self,
1051 "VectorStoreTable",
1052 table_name=f"{project_name}-vector-store",
1053 partition_key=dynamodb.Attribute(
1054 name="doc_id",
1055 type=dynamodb.AttributeType.STRING,
1056 ),
1057 # Vector indexes require on-demand billing; a bursty ingest-then-
1058 # query corpus fits it anyway.
1059 billing=dynamodb.Billing.on_demand(),
1060 replicas=[dynamodb.ReplicaTableProps(region=region) for region in replica_regions],
1061 removal_policy=RemovalPolicy.DESTROY,
1062 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
1063 point_in_time_recovery_enabled=True
1064 ),
1065 encryption=dynamodb.TableEncryptionV2.aws_managed_key(),
1066 )
1068 # Shared, pre-created execution role for the singleton
1069 # AwsCustomResource Lambda; this feature contributes its own
1070 # table-scoped statement (see _vector_index_custom_resource_role,
1071 # which documents the live-earned singleton-role failure).
1072 index_role = self._vector_index_custom_resource_role()
1073 index_role.add_to_policy(
1074 iam.PolicyStatement(
1075 effect=iam.Effect.ALLOW,
1076 actions=["dynamodb:UpdateTable", "dynamodb:DescribeTable"],
1077 resources=[self.vector_store_table.table_arn],
1078 )
1079 )
1081 vector_index_updates = [
1082 {
1083 "Create": {
1084 "IndexName": index_name,
1085 "VectorAttribute": {"AttributeName": "embedding"},
1086 "Dimensions": dimensions,
1087 "DistanceFunction": distance_function,
1088 "SearchSchema": [
1089 {
1090 "AttributeName": "source",
1091 "SearchSchemaElementType": "INLINE_FILTER",
1092 }
1093 ],
1094 "Projection": {
1095 "ProjectionType": "INCLUDE",
1096 "NonKeyAttributes": [
1097 "text",
1098 "source",
1099 "chunk_index",
1100 "title",
1101 "embedding_model_id",
1102 ],
1103 },
1104 }
1105 }
1106 ]
1107 vector_index = cr.AwsCustomResource(
1108 self,
1109 "VectorStoreIndex",
1110 on_create=cr.AwsSdkCall(
1111 service="DynamoDB",
1112 action="updateTable",
1113 parameters={
1114 "TableName": self.vector_store_table.table_name,
1115 # The INLINE_FILTER attribute must be declared in the same
1116 # UpdateTable call (live-verified on the mission-memory
1117 # index; re-verified against a 2019.11.21 global table in
1118 # the Phase 2 spike).
1119 "AttributeDefinitions": [{"AttributeName": "source", "AttributeType": "S"}],
1120 "VectorIndexUpdates": vector_index_updates,
1121 },
1122 physical_resource_id=cr.PhysicalResourceId.of(f"{project_name}-vector-store-index"),
1123 ),
1124 # DELIBERATELY NO on_delete. Deleting the index at teardown is
1125 # both unnecessary and actively harmful here.
1126 #
1127 # Unnecessary: the table carries RemovalPolicy.DESTROY, so stack
1128 # teardown deletes the table, and deleting a table removes its
1129 # indexes with it. There is no path where this custom resource
1130 # goes away while the table survives.
1131 #
1132 # Harmful: UpdateTable{VectorIndexUpdates:[Delete]} returns as
1133 # soon as the call is ACCEPTED and parks the table in UPDATING
1134 # for as long as the service needs. CFN then moves to the next
1135 # resource in reverse-dependency order — the GlobalTable — whose
1136 # first act is UpdateTable{ReplicaUpdates:[Delete <replica>]}.
1137 # That call requires the table to be ACTIVE, so it fails with
1138 # ResourceInUseException, and CFN retries it on a fixed interval
1139 # with no backoff and no give-up. Caught live 2026-08-14: index
1140 # delete accepted at 17:50:23Z, replica delete first refused at
1141 # 17:50:31Z, ~130 consecutive refusals, table still UPDATING
1142 # (index still ACTIVE on the primary, absent on the replica)
1143 # 2.5h later, gco-global DELETE_FAILED, and every manual
1144 # recovery path — update-table replica delete, delete-table in
1145 # either region — refused as well. The prior mitigation here
1146 # (ignore_error_codes_matching on this call) could not help: the
1147 # failure surfaces in CFN's own GlobalTable handler, not in this
1148 # custom resource. Removing the call removes the UPDATING window
1149 # altogether, which is the only reliable fix available to an
1150 # AwsCustomResource (it has no isComplete poller to wait for
1151 # ACTIVE). tests/test_vector_store_stack.py pins the absence.
1152 # Same live-earned requirement as mission memory: the runtime's
1153 # bundled SDK silently drops the VectorIndexUpdates member.
1154 install_latest_aws_sdk=True,
1155 role=index_role,
1156 )
1157 # The table (and its replicas) must be ACTIVE before UpdateTable can
1158 # add an index; the spike confirmed the call is accepted the moment
1159 # the GlobalTable resource completes.
1160 vector_index.node.add_dependency(self.vector_store_table)
1161 vector_index.node.add_dependency(index_role)
1163 # SSM is the established runtime/cross-region discovery contract.
1164 # Regional readers resolve the SSM parameters from the global region
1165 # and then query their LOCAL replica.
1166 ssm.StringParameter(
1167 self,
1168 "VectorStoreTableNameParam",
1169 parameter_name=f"/{project_name}/vector-store-table-name",
1170 string_value=self.vector_store_table.table_name,
1171 description="DynamoDB global-table name for the workload vector store",
1172 )
1173 ssm.StringParameter(
1174 self,
1175 "VectorStoreIndexNameParam",
1176 parameter_name=f"/{project_name}/vector-store-index-name",
1177 string_value=index_name,
1178 description="Vector index name over embedded corpus documents",
1179 )
1180 CfnOutput(
1181 self,
1182 "VectorStoreTableName",
1183 value=self.vector_store_table.table_name,
1184 description="DynamoDB global-table name for the workload vector store",
1185 export_name=f"{project_name}-vector-store-table-name",
1186 )
1187 CfnOutput(
1188 self,
1189 "VectorStoreTableArn",
1190 value=self.vector_store_table.table_arn,
1191 description="DynamoDB global-table ARN for the workload vector store",
1192 export_name=f"{project_name}-vector-store-table-arn",
1193 )
1195 # Backup coverage: join the existing DynamoDB backup plan (covers the
1196 # primary replica; corpus data is re-derivable from S3 by re-ingest,
1197 # so replicas need no independent backups).
1198 self.backup_plan.add_selection(
1199 "VectorStoreTableSelection",
1200 resources=[backup.BackupResource.from_dynamo_db_table(self.vector_store_table)],
1201 )
1203 def _create_vector_ingest(self) -> None:
1204 """Create the S3-triggered ingest Lambda for the vector store.
1206 Objects dropped under ``vector_store.corpus_prefix`` on the always-on
1207 cluster-shared bucket invoke the ``lambda/vector-ingest`` handler
1208 asynchronously; it chunks, embeds (Bedrock), and writes items to the
1209 vector-store table created by :meth:`_create_vector_store`. Uploading
1210 a corpus is therefore a plain S3 write — ``gco vector ingest`` wraps
1211 it, but any S3 client works.
1213 The bucket notification is additive: it synthesizes a separate
1214 ``Custom::S3BucketNotifications`` resource (plus CDK's singleton
1215 handler Lambda), so the disabled path leaves the cluster-shared
1216 bucket's own template byte-identical. The execution role carries the
1217 pipeline's write-only identity — object reads under the corpus
1218 prefix, ``PutItem`` on the table, and ``InvokeModel`` on the
1219 configured embedding model; the read path (``SearchVectors``)
1220 belongs to the regional workload role and the CLI, never to ingest.
1221 """
1222 from aws_cdk import aws_s3_notifications as s3_notifications
1223 from aws_cdk import aws_sqs as sqs
1225 from gco.stacks.nag_suppressions import acknowledge_nag_findings
1227 store_config = self.config.get_vector_store_config()
1228 corpus_prefix = str(store_config["corpus_prefix"])
1229 embedding_model_id = str(store_config["embedding_model_id"])
1230 dimensions = int(store_config["dimensions"])
1232 ingest_role = iam.Role(
1233 self,
1234 "VectorIngestRole",
1235 assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
1236 description=(
1237 "Execution role for the vector-store ingest Lambda: corpus-prefix "
1238 "object reads, vector-store PutItem, and embedding-model InvokeModel."
1239 ),
1240 managed_policies=[
1241 iam.ManagedPolicy.from_aws_managed_policy_name(
1242 "service-role/AWSLambdaBasicExecutionRole"
1243 )
1244 ],
1245 )
1246 ingest_role.add_to_policy(
1247 iam.PolicyStatement(
1248 effect=iam.Effect.ALLOW,
1249 actions=["s3:GetObject"],
1250 resources=[f"{self.cluster_shared_bucket.bucket_arn}/{corpus_prefix}*"],
1251 )
1252 )
1253 # The bucket is KMS-encrypted with the cluster-shared key; GetObject
1254 # needs Decrypt on it.
1255 ingest_role.add_to_policy(
1256 iam.PolicyStatement(
1257 effect=iam.Effect.ALLOW,
1258 actions=["kms:Decrypt"],
1259 resources=[self.cluster_shared_kms_key.key_arn],
1260 )
1261 )
1262 ingest_role.add_to_policy(
1263 iam.PolicyStatement(
1264 effect=iam.Effect.ALLOW,
1265 actions=["dynamodb:PutItem"],
1266 resources=[self.vector_store_table.table_arn],
1267 )
1268 )
1269 # The embedding model is a foundation model (account-less ARN) in
1270 # this (global) region — the same region the table's primary replica
1271 # and this Lambda live in.
1272 ingest_role.add_to_policy(
1273 iam.PolicyStatement(
1274 effect=iam.Effect.ALLOW,
1275 actions=["bedrock:InvokeModel"],
1276 resources=[
1277 f"arn:{self.partition}:bedrock:{self.region}::foundation-model/"
1278 f"{embedding_model_id}"
1279 ],
1280 )
1281 )
1283 # Async-invoke failures (after Lambda's built-in retries) land here
1284 # rather than vanishing; the DLQ message carries the original S3
1285 # event for replay.
1286 ingest_dlq = sqs.Queue(
1287 self,
1288 "VectorIngestDlq",
1289 retention_period=Duration.days(14),
1290 enforce_ssl=True,
1291 encryption=sqs.QueueEncryption.SQS_MANAGED,
1292 removal_policy=RemovalPolicy.DESTROY,
1293 )
1295 self.vector_ingest_lambda = lambda_.Function(
1296 self,
1297 "VectorIngestFunction",
1298 runtime=getattr(lambda_.Runtime, LAMBDA_PYTHON_RUNTIME),
1299 handler="handler.lambda_handler",
1300 code=lambda_source_code("lambda/vector-ingest"),
1301 timeout=Duration.minutes(5),
1302 memory_size=512,
1303 role=ingest_role,
1304 dead_letter_queue=ingest_dlq,
1305 environment={
1306 "VECTOR_STORE_TABLE_NAME": self.vector_store_table.table_name,
1307 "EMBEDDING_MODEL_ID": embedding_model_id,
1308 "EMBEDDING_DIMENSIONS": str(dimensions),
1309 "CORPUS_PREFIX": corpus_prefix,
1310 },
1311 tracing=lambda_.Tracing.ACTIVE,
1312 description=(
1313 "Vector-store ingest (opt-in global-stack add-on): chunks and embeds "
1314 "corpus objects from the cluster-shared bucket into the vector store."
1315 ),
1316 )
1318 self.cluster_shared_bucket.add_event_notification(
1319 s3.EventType.OBJECT_CREATED,
1320 s3_notifications.LambdaDestination(self.vector_ingest_lambda),
1321 s3.NotificationKeyFilter(prefix=corpus_prefix),
1322 )
1324 acknowledge_nag_findings(
1325 ingest_role,
1326 [
1327 {
1328 "id": "AwsSolutions-IAM4",
1329 "reason": (
1330 "AWSLambdaBasicExecutionRole provides the standard CloudWatch "
1331 "Logs permissions every Lambda needs."
1332 ),
1333 },
1334 {
1335 "id": "AwsSolutions-IAM5",
1336 "reason": (
1337 "The object-read grant is scoped to the vector corpus prefix "
1338 "of the cluster-shared bucket; ingest must read every object "
1339 "dropped under it, and S3 object grants require a key wildcard."
1340 ),
1341 "appliesTo": [
1342 f"Resource::<ClusterSharedBucket45D6691E.Arn>/{corpus_prefix}*",
1343 ],
1344 },
1345 {
1346 "id": "AwsSolutions-IAM5",
1347 "reason": (
1348 "Active X-Ray tracing requires xray:PutTraceSegments and "
1349 "xray:PutTelemetryRecords on Resource::* because those APIs do not "
1350 "support resource-level IAM constraints."
1351 ),
1352 "appliesTo": ["Resource::*"],
1353 },
1354 ],
1355 )
1356 acknowledge_nag_findings(
1357 ingest_dlq,
1358 [
1359 {
1360 "id": "AwsSolutions-SQS3",
1361 "reason": (
1362 "This queue is the dead-letter queue for the ingest Lambda's "
1363 "async invocations; a DLQ for a DLQ is circular."
1364 ),
1365 },
1366 {
1367 "id": "Serverless-SQSRedrivePolicy",
1368 "reason": (
1369 "This queue is itself the dead-letter queue for the ingest "
1370 "Lambda's async invocations, so it does not need its own "
1371 "redrive policy; a DLQ for a DLQ is circular."
1372 ),
1373 },
1374 ],
1375 )
1376 # CDK's bucket-notification wiring synthesizes a singleton handler
1377 # Lambda whose auto-generated role needs s3:PutBucketNotification on
1378 # every bucket it manages — the API supports no resource scoping.
1379 notifications_handler = self.node.try_find_child(
1380 "BucketNotificationsHandler050a0587b7544547bf325f094a3db834"
1381 )
1382 if notifications_handler is not None:
1383 acknowledge_nag_findings(
1384 notifications_handler,
1385 [
1386 {
1387 "id": "AwsSolutions-IAM4",
1388 "reason": (
1389 "CDK's singleton S3 bucket-notifications handler attaches "
1390 "AWSLambdaBasicExecutionRole for CloudWatch Logs."
1391 ),
1392 },
1393 {
1394 "id": "AwsSolutions-IAM5",
1395 "reason": (
1396 "s3:PutBucketNotification supports no resource-level "
1397 "scoping; CDK's singleton notifications handler requires "
1398 "the wildcard to manage bucket notification configuration."
1399 ),
1400 "appliesTo": ["Resource::*"],
1401 },
1402 ],
1403 )
1405 def _create_outputs(self) -> None:
1406 """Create CloudFormation outputs for cross-stack references."""
1407 if self.accelerator is None or self.listener is None:
1408 raise RuntimeError("Global Accelerator outputs require the commercial AWS partition")
1409 project_name = self.config.get_project_name()
1411 CfnOutput(
1412 self,
1413 "GlobalAcceleratorDnsName",
1414 value=self.accelerator.dns_name,
1415 description="Global Accelerator DNS name for global endpoint",
1416 export_name=f"{project_name}-global-accelerator-dns",
1417 )
1419 CfnOutput(
1420 self,
1421 "GlobalAcceleratorArn",
1422 value=self.accelerator.accelerator_arn,
1423 description="Global Accelerator ARN",
1424 export_name=f"{project_name}-global-accelerator-arn",
1425 )
1427 CfnOutput(
1428 self,
1429 "GlobalAcceleratorListenerArn",
1430 value=self.listener.listener_arn,
1431 description="Global Accelerator Listener ARN",
1432 export_name=f"{project_name}-global-accelerator-listener-arn",
1433 )
1435 def _apply_nag_suppressions(self) -> None:
1436 """Apply cdk-nag suppressions for this stack."""
1437 from gco.stacks.nag_suppressions import apply_all_suppressions
1439 apply_all_suppressions(
1440 self, stack_type="global", project_name=self.config.get_project_name()
1441 )
1443 def _create_endpoint_group(self, region: str) -> None:
1444 """
1445 Create an endpoint group for a specific region.
1447 Configures an HTTPS/443 health-check contract matching the ALB's only
1448 listener. Global Accelerator derives ALB endpoint health from the ALB
1449 target groups, but keeping the endpoint-group settings aligned prevents
1450 an accidental plaintext fallback if the endpoint type changes later.
1452 Also stores the endpoint group ARN in SSM Parameter Store for
1453 cross-region access by regional stacks.
1455 Args:
1456 region: AWS region name (e.g., 'us-east-1')
1457 """
1458 if self.listener is None:
1459 raise RuntimeError(
1460 "Global Accelerator endpoint groups are unavailable in this partition"
1461 )
1462 project_name = self.config.get_project_name()
1463 region_id = region.replace("-", "").title()
1464 ga_config = self.config.get_global_accelerator_config()
1466 # Keep the endpoint-group contract aligned with the HTTPS-only ALB.
1467 # For ALB endpoints GA uses target-group health rather than actively
1468 # applying these probe settings, but 443/HTTPS remains the safe default
1469 # if an endpoint type is changed in a future deployment.
1470 endpoint_group = self.listener.add_endpoint_group(
1471 f"EndpointGroup{region_id}",
1472 region=region,
1473 health_check_port=443,
1474 health_check_protocol=ga.HealthCheckProtocol.HTTPS,
1475 health_check_path=ga_config.get("health_check_path", "/api/v1/health"),
1476 health_check_interval=Duration.seconds(ga_config.get("health_check_interval", 30)),
1477 health_check_threshold=ga_config.get("health_check_threshold", 3),
1478 )
1480 self.endpoint_groups[region] = endpoint_group
1482 # Export endpoint group ARN for regional stacks
1483 CfnOutput(
1484 self,
1485 f"EndpointGroup{region_id}Arn",
1486 value=endpoint_group.endpoint_group_arn,
1487 description=f"Endpoint group ARN for {region}",
1488 export_name=f"{project_name}-endpoint-group-{region}-arn",
1489 )
1491 # Store endpoint group ARN in SSM Parameter Store for cross-region access
1492 # Regional stacks read this to register their ALBs with Global Accelerator
1493 ssm.StringParameter(
1494 self,
1495 f"EndpointGroup{region_id}ArnParam",
1496 parameter_name=f"/{project_name}/endpoint-group-{region}-arn",
1497 string_value=endpoint_group.endpoint_group_arn,
1498 description=f"Global Accelerator endpoint group ARN for {region}",
1499 )
1501 def get_accelerator_dns_name(self) -> str | None:
1502 """Return the Global Accelerator DNS name when this partition supports it."""
1503 return str(self.accelerator.dns_name) if self.accelerator is not None else None
1505 def get_accelerator_arn(self) -> str:
1506 """Get the Global Accelerator ARN."""
1507 if self.accelerator is None:
1508 raise RuntimeError("Global Accelerator is unavailable in this partition")
1509 return str(self.accelerator.accelerator_arn)
1511 def get_listener_arn(self) -> str:
1512 """Get the Global Accelerator Listener ARN."""
1513 if self.listener is None:
1514 raise RuntimeError("Global Accelerator is unavailable in this partition")
1515 return str(self.listener.listener_arn)
1517 def get_endpoint_group_arn(self, region: str) -> str:
1518 """Get the endpoint group ARN for a specific region"""
1519 if region in self.endpoint_groups:
1520 return str(self.endpoint_groups[region].endpoint_group_arn)
1521 raise ValueError(f"No endpoint group found for region: {region}")
1523 def _create_dynamodb_tables(self) -> None:
1524 """Create DynamoDB tables for templates, webhooks, jobs, inference endpoints, and missions."""
1525 project_name = self.config.get_project_name()
1527 # Job Templates table - stores reusable job templates
1528 self.templates_table = dynamodb.Table(
1529 self,
1530 "JobTemplatesTable",
1531 table_name=f"{project_name}-job-templates",
1532 partition_key=dynamodb.Attribute(
1533 name="template_name",
1534 type=dynamodb.AttributeType.STRING,
1535 ),
1536 billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
1537 removal_policy=RemovalPolicy.DESTROY,
1538 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
1539 point_in_time_recovery_enabled=True
1540 ),
1541 encryption=dynamodb.TableEncryption.AWS_MANAGED,
1542 )
1544 # Webhooks table - stores webhook registrations
1545 self.webhooks_table = dynamodb.Table(
1546 self,
1547 "WebhooksTable",
1548 table_name=f"{project_name}-webhooks",
1549 partition_key=dynamodb.Attribute(
1550 name="webhook_id",
1551 type=dynamodb.AttributeType.STRING,
1552 ),
1553 billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
1554 removal_policy=RemovalPolicy.DESTROY,
1555 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
1556 point_in_time_recovery_enabled=True
1557 ),
1558 encryption=dynamodb.TableEncryption.AWS_MANAGED,
1559 )
1561 # Add GSI for querying webhooks by namespace
1562 self.webhooks_table.add_global_secondary_index(
1563 index_name="namespace-index",
1564 partition_key=dynamodb.Attribute(
1565 name="namespace",
1566 type=dynamodb.AttributeType.STRING,
1567 ),
1568 projection_type=dynamodb.ProjectionType.ALL,
1569 )
1571 # Jobs table - centralized job tracking and queue
1572 # This enables global job submission with regional pickup
1573 self.jobs_table = dynamodb.Table(
1574 self,
1575 "JobsTable",
1576 table_name=f"{project_name}-jobs",
1577 partition_key=dynamodb.Attribute(
1578 name="job_id",
1579 type=dynamodb.AttributeType.STRING,
1580 ),
1581 billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
1582 removal_policy=RemovalPolicy.DESTROY,
1583 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
1584 point_in_time_recovery_enabled=True
1585 ),
1586 encryption=dynamodb.TableEncryption.AWS_MANAGED,
1587 time_to_live_attribute="ttl", # Auto-cleanup old completed jobs
1588 )
1590 # Legacy GSI retained for compatibility with existing deployments and
1591 # ad-hoc operational queries.
1592 self.jobs_table.add_global_secondary_index(
1593 index_name="region-status-index",
1594 partition_key=dynamodb.Attribute(
1595 name="target_region",
1596 type=dynamodb.AttributeType.STRING,
1597 ),
1598 sort_key=dynamodb.Attribute(
1599 name="status",
1600 type=dynamodb.AttributeType.STRING,
1601 ),
1602 projection_type=dynamodb.ProjectionType.ALL,
1603 )
1605 # ``work_sort`` is priority/FIFO for queued records and lease expiry for
1606 # claimed or applying records. This unified worker index is the only GSI
1607 # added by this release because DynamoDB permits only one GSI creation or
1608 # deletion per table update. Workers repeatedly backfill legacy rows
1609 # through the retained region-status-index during mixed-version rollouts.
1610 self.jobs_table.add_global_secondary_index(
1611 index_name="region-status-work-index",
1612 partition_key=dynamodb.Attribute(
1613 name="region_status",
1614 type=dynamodb.AttributeType.STRING,
1615 ),
1616 sort_key=dynamodb.Attribute(
1617 name="work_sort",
1618 type=dynamodb.AttributeType.STRING,
1619 ),
1620 projection_type=dynamodb.ProjectionType.ALL,
1621 )
1623 # GSI for querying jobs by namespace
1624 self.jobs_table.add_global_secondary_index(
1625 index_name="namespace-index",
1626 partition_key=dynamodb.Attribute(
1627 name="namespace",
1628 type=dynamodb.AttributeType.STRING,
1629 ),
1630 sort_key=dynamodb.Attribute(
1631 name="submitted_at",
1632 type=dynamodb.AttributeType.STRING,
1633 ),
1634 projection_type=dynamodb.ProjectionType.ALL,
1635 )
1637 # GSI for querying jobs by status globally
1638 self.jobs_table.add_global_secondary_index(
1639 index_name="status-index",
1640 partition_key=dynamodb.Attribute(
1641 name="status",
1642 type=dynamodb.AttributeType.STRING,
1643 ),
1644 sort_key=dynamodb.Attribute(
1645 name="submitted_at",
1646 type=dynamodb.AttributeType.STRING,
1647 ),
1648 projection_type=dynamodb.ProjectionType.ALL,
1649 )
1651 # Export table names and ARNs for regional stacks
1652 CfnOutput(
1653 self,
1654 "TemplatesTableName",
1655 value=self.templates_table.table_name,
1656 description="DynamoDB table name for job templates",
1657 export_name=f"{project_name}-templates-table-name",
1658 )
1660 CfnOutput(
1661 self,
1662 "TemplatesTableArn",
1663 value=self.templates_table.table_arn,
1664 description="DynamoDB table ARN for job templates",
1665 export_name=f"{project_name}-templates-table-arn",
1666 )
1668 CfnOutput(
1669 self,
1670 "WebhooksTableName",
1671 value=self.webhooks_table.table_name,
1672 description="DynamoDB table name for webhooks",
1673 export_name=f"{project_name}-webhooks-table-name",
1674 )
1676 CfnOutput(
1677 self,
1678 "WebhooksTableArn",
1679 value=self.webhooks_table.table_arn,
1680 description="DynamoDB table ARN for webhooks",
1681 export_name=f"{project_name}-webhooks-table-arn",
1682 )
1684 CfnOutput(
1685 self,
1686 "JobsTableName",
1687 value=self.jobs_table.table_name,
1688 description="DynamoDB table name for centralized job tracking",
1689 export_name=f"{project_name}-jobs-table-name",
1690 )
1692 CfnOutput(
1693 self,
1694 "JobsTableArn",
1695 value=self.jobs_table.table_arn,
1696 description="DynamoDB table ARN for centralized job tracking",
1697 export_name=f"{project_name}-jobs-table-arn",
1698 )
1700 # Inference Endpoints table - stores desired state for inference deployments
1701 # The inference_monitor in each regional cluster polls this table
1702 self.inference_endpoints_table = dynamodb.Table(
1703 self,
1704 "InferenceEndpointsTable",
1705 table_name=f"{project_name}-inference-endpoints",
1706 partition_key=dynamodb.Attribute(
1707 name="endpoint_name",
1708 type=dynamodb.AttributeType.STRING,
1709 ),
1710 billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
1711 removal_policy=RemovalPolicy.DESTROY,
1712 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
1713 point_in_time_recovery_enabled=True
1714 ),
1715 encryption=dynamodb.TableEncryption.AWS_MANAGED,
1716 )
1718 CfnOutput(
1719 self,
1720 "InferenceEndpointsTableName",
1721 value=self.inference_endpoints_table.table_name,
1722 description="DynamoDB table name for inference endpoint state",
1723 export_name=f"{project_name}-inference-endpoints-table-name",
1724 )
1726 CfnOutput(
1727 self,
1728 "InferenceEndpointsTableArn",
1729 value=self.inference_endpoints_table.table_arn,
1730 description="DynamoDB table ARN for inference endpoint state",
1731 export_name=f"{project_name}-inference-endpoints-table-arn",
1732 )
1734 # Missions table - persists goal-directed iteration session state
1735 # Partition by session_id; the status-index GSI supports paginated
1736 # listing by status (e.g. running, completed, terminated, failed).
1737 self.missions_table = dynamodb.Table(
1738 self,
1739 "MissionsTable",
1740 table_name=f"{project_name}-missions",
1741 partition_key=dynamodb.Attribute(
1742 name="session_id",
1743 type=dynamodb.AttributeType.STRING,
1744 ),
1745 billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
1746 removal_policy=RemovalPolicy.DESTROY,
1747 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
1748 point_in_time_recovery_enabled=True
1749 ),
1750 encryption=dynamodb.TableEncryption.AWS_MANAGED,
1751 )
1753 # GSI for paginating sessions by status (sorted by creation time)
1754 self.missions_table.add_global_secondary_index(
1755 index_name="status-index",
1756 partition_key=dynamodb.Attribute(
1757 name="status",
1758 type=dynamodb.AttributeType.STRING,
1759 ),
1760 sort_key=dynamodb.Attribute(
1761 name="created_at",
1762 type=dynamodb.AttributeType.STRING,
1763 ),
1764 projection_type=dynamodb.ProjectionType.ALL,
1765 )
1767 CfnOutput(
1768 self,
1769 "MissionsTableName",
1770 value=self.missions_table.table_name,
1771 description="DynamoDB table name for mission session state",
1772 export_name=f"{project_name}-missions-table-name",
1773 )
1775 CfnOutput(
1776 self,
1777 "MissionsTableArn",
1778 value=self.missions_table.table_arn,
1779 description="DynamoDB table ARN for mission session state",
1780 export_name=f"{project_name}-missions-table-arn",
1781 )
1783 # Store table names in SSM for cross-region access
1784 ssm.StringParameter(
1785 self,
1786 "TemplatesTableNameParam",
1787 parameter_name=f"/{project_name}/templates-table-name",
1788 string_value=self.templates_table.table_name,
1789 description="DynamoDB table name for job templates",
1790 )
1792 ssm.StringParameter(
1793 self,
1794 "WebhooksTableNameParam",
1795 parameter_name=f"/{project_name}/webhooks-table-name",
1796 string_value=self.webhooks_table.table_name,
1797 description="DynamoDB table name for webhooks",
1798 )
1800 ssm.StringParameter(
1801 self,
1802 "JobsTableNameParam",
1803 parameter_name=f"/{project_name}/jobs-table-name",
1804 string_value=self.jobs_table.table_name,
1805 description="DynamoDB table name for centralized job tracking",
1806 )
1808 ssm.StringParameter(
1809 self,
1810 "InferenceEndpointsTableNameParam",
1811 parameter_name=f"/{project_name}/inference-endpoints-table-name",
1812 string_value=self.inference_endpoints_table.table_name,
1813 description="DynamoDB table name for inference endpoint state",
1814 )
1816 ssm.StringParameter(
1817 self,
1818 "MissionsTableNameParam",
1819 parameter_name=f"/{project_name}/missions-table-name",
1820 string_value=self.missions_table.table_name,
1821 description="DynamoDB table name for mission session state",
1822 )
1824 def _create_model_bucket(self) -> None:
1825 """Create S3 bucket for model weights.
1827 This bucket serves as the central model registry. Users upload model
1828 weights here once, and the inference_monitor's init containers sync
1829 them to each region's local EFS at pod startup.
1831 The bucket name is auto-generated by CDK to avoid naming collisions.
1832 It's exported via CfnOutput and SSM for CLI discovery.
1833 """
1834 project_name = self.config.get_project_name()
1836 # KMS key for model bucket encryption
1837 self.model_bucket_key = kms.Key(
1838 self,
1839 "ModelBucketKey",
1840 description="KMS key for GCO model weights bucket",
1841 enable_key_rotation=True,
1842 removal_policy=RemovalPolicy.DESTROY,
1843 )
1845 # Access logs bucket (required for compliance)
1846 # Retention is configurable via cdk.json context field `s3_access_logs.retention_days`
1847 # (default: 90 days). Logs older than the configured retention are expired.
1848 s3_access_logs_ctx = self.node.try_get_context("s3_access_logs") or {}
1849 access_logs_retention_days = int(s3_access_logs_ctx.get("retention_days", 90))
1851 self.model_bucket_access_logs = s3.Bucket(
1852 self,
1853 "ModelWeightsAccessLogsBucket",
1854 encryption=s3.BucketEncryption.S3_MANAGED,
1855 block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
1856 enforce_ssl=True,
1857 versioned=True,
1858 removal_policy=RemovalPolicy.DESTROY,
1859 auto_delete_objects=True,
1860 lifecycle_rules=[
1861 s3.LifecycleRule(
1862 id="ExpireAccessLogs",
1863 enabled=True,
1864 expiration=Duration.days(access_logs_retention_days),
1865 )
1866 ],
1867 )
1869 # Model weights bucket
1870 self.model_bucket = s3.Bucket(
1871 self,
1872 "ModelWeightsBucket",
1873 encryption=s3.BucketEncryption.KMS,
1874 encryption_key=self.model_bucket_key,
1875 bucket_key_enabled=True,
1876 block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
1877 enforce_ssl=True,
1878 versioned=True,
1879 removal_policy=RemovalPolicy.DESTROY,
1880 auto_delete_objects=True,
1881 server_access_logs_bucket=self.model_bucket_access_logs,
1882 server_access_logs_prefix="model-bucket-logs/",
1883 )
1885 # CDK-nag suppressions — only replication (not needed for model weights)
1886 from gco.stacks.nag_suppressions import acknowledge_nag_findings
1888 replication_reason = (
1889 "Model weights are user-uploaded artifacts that can be re-uploaded. "
1890 "Cross-region replication is not required; the inference_monitor "
1891 "syncs models from S3 to each region's EFS at pod startup."
1892 )
1894 acknowledge_nag_findings(
1895 self.model_bucket,
1896 [
1897 {
1898 "id": "HIPAA.Security-S3BucketReplicationEnabled",
1899 "reason": replication_reason,
1900 },
1901 {
1902 "id": "NIST.800.53.R5-S3BucketReplicationEnabled",
1903 "reason": replication_reason,
1904 },
1905 {
1906 "id": "PCI.DSS.321-S3BucketReplicationEnabled",
1907 "reason": replication_reason,
1908 },
1909 ],
1910 )
1912 logs_reason = "This is the server access logs destination bucket."
1913 acknowledge_nag_findings(
1914 self.model_bucket_access_logs,
1915 [
1916 {"id": "AwsSolutions-S1", "reason": logs_reason},
1917 {"id": "HIPAA.Security-S3BucketLoggingEnabled", "reason": logs_reason},
1918 {
1919 "id": "HIPAA.Security-S3BucketReplicationEnabled",
1920 "reason": "Access logs do not require replication.",
1921 },
1922 {
1923 "id": "HIPAA.Security-S3DefaultEncryptionKMS",
1924 "reason": "SSE-S3 is sufficient for access logs.",
1925 },
1926 {"id": "NIST.800.53.R5-S3BucketLoggingEnabled", "reason": logs_reason},
1927 {
1928 "id": "NIST.800.53.R5-S3BucketReplicationEnabled",
1929 "reason": "Access logs do not require replication.",
1930 },
1931 {
1932 "id": "NIST.800.53.R5-S3DefaultEncryptionKMS",
1933 "reason": "SSE-S3 is sufficient for access logs.",
1934 },
1935 {"id": "PCI.DSS.321-S3BucketLoggingEnabled", "reason": logs_reason},
1936 {
1937 "id": "PCI.DSS.321-S3BucketReplicationEnabled",
1938 "reason": "Access logs do not require replication.",
1939 },
1940 {
1941 "id": "PCI.DSS.321-S3DefaultEncryptionKMS",
1942 "reason": "SSE-S3 is sufficient for access logs.",
1943 },
1944 ],
1945 )
1947 CfnOutput(
1948 self,
1949 "ModelBucketName",
1950 value=self.model_bucket.bucket_name,
1951 description="S3 bucket for model weights",
1952 export_name=f"{project_name}-model-bucket-name",
1953 )
1955 CfnOutput(
1956 self,
1957 "ModelBucketArn",
1958 value=self.model_bucket.bucket_arn,
1959 description="S3 bucket ARN for model weights",
1960 export_name=f"{project_name}-model-bucket-arn",
1961 )
1963 ssm.StringParameter(
1964 self,
1965 "ModelBucketNameParam",
1966 parameter_name=f"/{project_name}/model-bucket-name",
1967 string_value=self.model_bucket.bucket_name,
1968 description="S3 bucket name for model weights",
1969 )
1971 def _create_backup_plan(self) -> None:
1972 """Create AWS Backup plan for DynamoDB tables.
1974 Creates a backup plan with:
1975 - Daily backups retained for 35 days
1976 - Weekly backups retained for 90 days
1977 - All DynamoDB tables added to the backup selection
1978 """
1979 # Create backup vault for storing backups
1980 self.backup_vault = backup.BackupVault(
1981 self,
1982 "DynamoDBBackupVault",
1983 removal_policy=RemovalPolicy.DESTROY,
1984 )
1986 # Create backup plan with daily and weekly rules
1987 self.backup_plan = backup.BackupPlan(
1988 self,
1989 "DynamoDBBackupPlan",
1990 backup_plan_rules=[
1991 # Daily backup - retained for 35 days
1992 backup.BackupPlanRule(
1993 rule_name="DailyBackup",
1994 backup_vault=self.backup_vault,
1995 schedule_expression=events.Schedule.cron(
1996 hour="3",
1997 minute="0",
1998 ),
1999 delete_after=Duration.days(35),
2000 enable_continuous_backup=True, # Enable PITR for DynamoDB
2001 ),
2002 # Weekly backup - retained for 90 days
2003 backup.BackupPlanRule(
2004 rule_name="WeeklyBackup",
2005 backup_vault=self.backup_vault,
2006 schedule_expression=events.Schedule.cron(
2007 hour="4",
2008 minute="0",
2009 week_day="SUN",
2010 ),
2011 delete_after=Duration.days(90),
2012 ),
2013 ],
2014 )
2016 # Add all DynamoDB tables to the backup selection
2017 self.backup_plan.add_selection(
2018 "DynamoDBTablesSelection",
2019 resources=[
2020 backup.BackupResource.from_dynamo_db_table(self.templates_table),
2021 backup.BackupResource.from_dynamo_db_table(self.webhooks_table),
2022 backup.BackupResource.from_dynamo_db_table(self.jobs_table),
2023 backup.BackupResource.from_dynamo_db_table(self.inference_endpoints_table),
2024 backup.BackupResource.from_dynamo_db_table(self.missions_table),
2025 ],
2026 )
2028 # Export backup plan ARN
2029 project_name = self.config.get_project_name()
2030 CfnOutput(
2031 self,
2032 "BackupPlanArn",
2033 value=self.backup_plan.backup_plan_arn,
2034 description="AWS Backup plan ARN for DynamoDB tables",
2035 export_name=f"{project_name}-backup-plan-arn",
2036 )
2038 CfnOutput(
2039 self,
2040 "BackupVaultArn",
2041 value=self.backup_vault.backup_vault_arn,
2042 description="AWS Backup vault ARN for DynamoDB backups",
2043 export_name=f"{project_name}-backup-vault-arn",
2044 )
2046 def _create_cluster_shared_kms_key(self) -> None:
2047 """Create the always-on customer-managed KMS key for ``Cluster_Shared_Bucket``.
2049 The key:
2050 - Enables automatic annual rotation.
2051 - Uses a 7-day pending window on destroy — the AWS minimum, matching the
2052 destroy-by-default iteration-loop posture of the analytics-environment
2053 feature while still providing a safety net against accidental deletion.
2054 - Uses ``RemovalPolicy.DESTROY`` so a ``cdk destroy gco-global`` cleans up
2055 the key without operator intervention (iteration-loop posture).
2056 - Grants encrypt/decrypt to the ``s3.amazonaws.com`` and
2057 ``logs.<region>.amazonaws.com`` service principals via the key policy
2058 so S3 server-side encryption and CloudWatch access-log delivery can use
2059 the key without role-side grants.
2061 The key is exposed as ``self.cluster_shared_kms_key`` for tests and for
2062 ``_create_cluster_shared_bucket`` to reference. Role-side usage grants
2063 (``kms:Decrypt`` / ``kms:GenerateDataKey``) are attached by downstream
2064 consumers: ``GCORegionalStack`` on the job-pod role (always-on)
2065 and ``GCOAnalyticsStack`` on the SageMaker execution role (conditional on
2066 the analytics toggle).
2067 """
2068 self.cluster_shared_kms_key = kms.Key(
2069 self,
2070 "ClusterSharedKmsKey",
2071 description=(
2072 "Customer-managed KMS key for the always-on Cluster_Shared_Bucket "
2073 "in GCOGlobalStack. Consumed by every regional EKS cluster and by "
2074 "GCOAnalyticsStack when analytics is enabled."
2075 ),
2076 enable_key_rotation=True,
2077 pending_window=Duration.days(7),
2078 removal_policy=RemovalPolicy.DESTROY,
2079 )
2081 # Key-policy grants for service principals that need to encrypt/decrypt
2082 # on behalf of the bucket (S3 server-side encryption) and the access-logs
2083 # bucket (CloudWatch Logs delivery). The actions match the standard
2084 # service-principal pattern used by cdk's default key policies.
2085 kms_actions = [
2086 "kms:Encrypt",
2087 "kms:Decrypt",
2088 "kms:ReEncrypt*",
2089 "kms:GenerateDataKey*",
2090 "kms:DescribeKey",
2091 ]
2093 self.cluster_shared_kms_key.add_to_resource_policy(
2094 iam.PolicyStatement(
2095 sid="AllowS3ServiceEncryptDecrypt",
2096 effect=iam.Effect.ALLOW,
2097 principals=[iam.ServicePrincipal("s3.amazonaws.com")],
2098 actions=kms_actions,
2099 resources=["*"],
2100 )
2101 )
2103 self.cluster_shared_kms_key.add_to_resource_policy(
2104 iam.PolicyStatement(
2105 sid="AllowCloudWatchLogsEncryptDecrypt",
2106 effect=iam.Effect.ALLOW,
2107 principals=[iam.ServicePrincipal("logs.amazonaws.com", region=self.region)],
2108 actions=kms_actions,
2109 resources=["*"],
2110 )
2111 )
2113 def _create_cluster_shared_bucket(self) -> None:
2114 """Create the always-on ``Cluster_Shared_Bucket`` and its access-logs bucket.
2116 Two buckets are created:
2118 1. ``cluster_shared_access_logs_bucket`` — dedicated S3 access-logs bucket
2119 used as ``server_access_logs_bucket`` for the primary bucket. Separate
2120 from ``model_bucket_access_logs`` so cluster-shared-bucket access logs
2121 are not commingled with model-bucket logs.
2122 2. ``cluster_shared_bucket`` — the primary bucket. Its physical name
2123 is CloudFormation-generated (``<stack>-clustersharedbucket…``): S3
2124 bucket names are a global namespace and a deleted name is not
2125 reliably reusable, so a fixed project/account/region name would
2126 make every destroy-and-redeploy a collision hazard. Consumers never
2127 reconstruct it — every regional stack and the analytics stack read
2128 the SSM parameters published under
2129 ``cluster_shared_ssm_parameter_prefix(project_name)``. KMS-encrypted
2130 with ``cluster_shared_kms_key``, block-public-access on, SSL
2131 enforced, versioned, destroy-on-teardown.
2133 An explicit ``Deny`` statement for ``aws:SecureTransport=false`` is added
2134 to the bucket policy independent of ``enforce_ssl=True`` so the deny is
2135 verifiable in the synthesized template (belt-and-suspenders).
2137 Grants on ``Cluster_Shared_Bucket`` are intentionally not added here —
2138 they live on downstream role policies (``GCORegionalStack`` on the
2139 job-pod role, ``GCOAnalyticsStack`` on the SageMaker execution role)
2140 rather than in this bucket's policy. The bucket policy contains zero
2141 ``Principal: "*"`` Allow statements.
2142 """
2143 # Retention for the access-logs bucket honors the same `s3_access_logs`
2144 # context field as the model-bucket access-logs bucket (default 90 days).
2145 s3_access_logs_ctx = self.node.try_get_context("s3_access_logs") or {}
2146 access_logs_retention_days = int(s3_access_logs_ctx.get("retention_days", 90))
2148 # Dedicated access-logs bucket for Cluster_Shared_Bucket. Encrypted with
2149 # the cluster-shared KMS key (the key policy grants the logs service
2150 # principal encrypt/decrypt). Kept separate from model_bucket_access_logs
2151 # so operators can reason about each bucket's logs independently. Matches
2152 # the LifecycleRule used on `model_bucket_access_logs` so retention is
2153 # consistent across the two log sinks.
2154 self.cluster_shared_access_logs_bucket = s3.Bucket(
2155 self,
2156 "ClusterSharedAccessLogsBucket",
2157 encryption=s3.BucketEncryption.KMS,
2158 encryption_key=self.cluster_shared_kms_key,
2159 block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
2160 enforce_ssl=True,
2161 versioned=True,
2162 removal_policy=RemovalPolicy.DESTROY,
2163 auto_delete_objects=True,
2164 lifecycle_rules=[
2165 s3.LifecycleRule(
2166 id="ExpireAccessLogs",
2167 enabled=True,
2168 expiration=Duration.days(access_logs_retention_days),
2169 )
2170 ],
2171 )
2173 # Primary Cluster_Shared_Bucket. No ``bucket_name``: the physical name
2174 # is CloudFormation-generated so a destroy-and-redeploy can never
2175 # collide in S3's global namespace (the model bucket has always worked
2176 # this way). Downstream grants use the ARN resolved from the SSM
2177 # parameters published below, never a reconstructed name.
2178 # `bucket_key_enabled=True` mirrors the model_bucket pattern to reduce
2179 # per-object KMS request costs.
2180 self.cluster_shared_bucket = s3.Bucket(
2181 self,
2182 "ClusterSharedBucket",
2183 encryption=s3.BucketEncryption.KMS,
2184 encryption_key=self.cluster_shared_kms_key,
2185 bucket_key_enabled=True,
2186 block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
2187 enforce_ssl=True,
2188 versioned=True,
2189 removal_policy=RemovalPolicy.DESTROY,
2190 auto_delete_objects=True,
2191 server_access_logs_bucket=self.cluster_shared_access_logs_bucket,
2192 server_access_logs_prefix="cluster-shared/",
2193 )
2195 # Explicit Deny for insecure transport. `enforce_ssl=True` already adds
2196 # an equivalent statement, but duplicating it here makes the deny
2197 # verifiable in the synthesized template under a known SID and satisfies
2198 # a belt-and-suspenders posture.
2199 self.cluster_shared_bucket.add_to_resource_policy(
2200 iam.PolicyStatement(
2201 sid="DenyInsecureTransport",
2202 effect=iam.Effect.DENY,
2203 principals=[iam.AnyPrincipal()],
2204 actions=["s3:*"],
2205 resources=[
2206 self.cluster_shared_bucket.bucket_arn,
2207 f"{self.cluster_shared_bucket.bucket_arn}/*",
2208 ],
2209 conditions={"Bool": {"aws:SecureTransport": "false"}},
2210 )
2211 )
2213 # CDK-nag suppressions — scoped per-resource at the construct site to
2214 # mirror the ``_create_model_bucket`` pattern (keeps the suppression
2215 # co-located with the construct it applies to, so the reason survives
2216 # refactors). Every suppression carries an explicit reason
2217 # string; no blanket ``Resource::*`` bypasses.
2218 from gco.stacks.nag_suppressions import acknowledge_nag_findings
2220 shared_replication_reason = (
2221 "Cluster_Shared_Bucket is a regional scratch sink; cluster jobs "
2222 "publish to it from a single region, and there is no durability "
2223 "requirement that warrants cross-region replication. Access logs "
2224 "do not require replication for the same reason."
2225 )
2227 acknowledge_nag_findings(
2228 self.cluster_shared_bucket,
2229 [
2230 {
2231 "id": "HIPAA.Security-S3BucketReplicationEnabled",
2232 "reason": shared_replication_reason,
2233 },
2234 {
2235 "id": "NIST.800.53.R5-S3BucketReplicationEnabled",
2236 "reason": shared_replication_reason,
2237 },
2238 {
2239 "id": "PCI.DSS.321-S3BucketReplicationEnabled",
2240 "reason": shared_replication_reason,
2241 },
2242 ],
2243 )
2245 access_logs_is_self_target_reason = (
2246 "This is the server access logs destination bucket for Cluster_Shared_Bucket."
2247 )
2248 acknowledge_nag_findings(
2249 self.cluster_shared_access_logs_bucket,
2250 [
2251 {
2252 "id": "AwsSolutions-S1",
2253 "reason": access_logs_is_self_target_reason,
2254 },
2255 {
2256 "id": "HIPAA.Security-S3BucketLoggingEnabled",
2257 "reason": access_logs_is_self_target_reason,
2258 },
2259 {
2260 "id": "NIST.800.53.R5-S3BucketLoggingEnabled",
2261 "reason": access_logs_is_self_target_reason,
2262 },
2263 {
2264 "id": "PCI.DSS.321-S3BucketLoggingEnabled",
2265 "reason": access_logs_is_self_target_reason,
2266 },
2267 {
2268 "id": "HIPAA.Security-S3BucketReplicationEnabled",
2269 "reason": shared_replication_reason,
2270 },
2271 {
2272 "id": "NIST.800.53.R5-S3BucketReplicationEnabled",
2273 "reason": shared_replication_reason,
2274 },
2275 {
2276 "id": "PCI.DSS.321-S3BucketReplicationEnabled",
2277 "reason": shared_replication_reason,
2278 },
2279 ],
2280 )
2282 def _publish_cluster_shared_bucket_ssm_params(self) -> None:
2283 """Publish the three ``/gco/cluster-shared-bucket/*`` SSM parameters.
2285 Writes:
2287 - ``/gco/cluster-shared-bucket/name`` — bucket name
2288 - ``/gco/cluster-shared-bucket/arn`` — bucket ARN
2289 - ``/gco/cluster-shared-bucket/region`` — bucket home region (global region)
2291 These parameters are the cross-region contract consumed by
2292 ``GCORegionalStack._resolve_cluster_shared_bucket_from_ssm`` (always) and by
2293 ``GCOAnalyticsStack._grant_sagemaker_role_on_cluster_shared_bucket``
2294 (conditional on the analytics toggle). The prefix from
2295 ``cluster_shared_ssm_parameter_prefix(project_name)`` is the single
2296 source of truth so the namespace can be renamed in exactly one place.
2298 Also emits four ``CfnOutput`` values for discoverability: the three SSM
2299 values plus the KMS key ARN. Export names follow the existing
2300 ``{project_name}-cluster-shared-{suffix}`` pattern used by the rest of
2301 this stack's outputs so operators can cross-reference them from peer
2302 stacks via ``Fn.import_value`` if needed (the primary cross-region
2303 contract remains SSM).
2304 """
2305 project_name = self.config.get_project_name()
2307 ssm.StringParameter(
2308 self,
2309 "ClusterSharedBucketNameParam",
2310 parameter_name=f"{cluster_shared_ssm_parameter_prefix(project_name)}/name",
2311 string_value=self.cluster_shared_bucket.bucket_name,
2312 description="Name of the always-on Cluster_Shared_Bucket (owned by GCOGlobalStack).",
2313 )
2315 ssm.StringParameter(
2316 self,
2317 "ClusterSharedBucketArnParam",
2318 parameter_name=f"{cluster_shared_ssm_parameter_prefix(project_name)}/arn",
2319 string_value=self.cluster_shared_bucket.bucket_arn,
2320 description="ARN of the always-on Cluster_Shared_Bucket (owned by GCOGlobalStack).",
2321 )
2323 ssm.StringParameter(
2324 self,
2325 "ClusterSharedBucketRegionParam",
2326 parameter_name=f"{cluster_shared_ssm_parameter_prefix(project_name)}/region",
2327 string_value=self.region,
2328 description="Home region of the always-on Cluster_Shared_Bucket (the global region).",
2329 )
2331 CfnOutput(
2332 self,
2333 "ClusterSharedBucketName",
2334 value=self.cluster_shared_bucket.bucket_name,
2335 description="Name of the always-on Cluster_Shared_Bucket.",
2336 export_name=f"{project_name}-cluster-shared-bucket-name",
2337 )
2339 CfnOutput(
2340 self,
2341 "ClusterSharedBucketArn",
2342 value=self.cluster_shared_bucket.bucket_arn,
2343 description="ARN of the always-on Cluster_Shared_Bucket.",
2344 export_name=f"{project_name}-cluster-shared-bucket-arn",
2345 )
2347 CfnOutput(
2348 self,
2349 "ClusterSharedBucketRegion",
2350 value=self.region,
2351 description="Home region of the always-on Cluster_Shared_Bucket.",
2352 export_name=f"{project_name}-cluster-shared-bucket-region",
2353 )
2355 CfnOutput(
2356 self,
2357 "ClusterSharedKmsKeyArn",
2358 value=self.cluster_shared_kms_key.key_arn,
2359 description="ARN of the always-on KMS key encrypting Cluster_Shared_Bucket.",
2360 export_name=f"{project_name}-cluster-shared-kms-key-arn",
2361 )
2363 def _resolve_replication_destinations(self, destinations: str | list[str]) -> list[str]:
2364 """Resolve the configured replication destinations into a region list.
2366 When ``destinations`` is the literal ``"all_deployed_regions"``, the
2367 list comes from ``self.config.get_regions()`` (the same source the
2368 rest of the stack uses for cross-region wiring). When it is an
2369 explicit list, it is returned as-is. The source region (the global
2370 stack's deploy region) is excluded — ECR replication is point-to-point
2371 and a self-referential destination is rejected by the API.
2372 """
2373 if isinstance(destinations, str):
2374 candidate_regions = list(self.config.get_regions())
2375 else:
2376 candidate_regions = list(destinations)
2377 return [region for region in candidate_regions if region != self.region]
2379 def _create_image_replication_rule(self) -> None:
2380 """Provision the ECR replication rule for ``gco/*`` repositories.
2382 When ``images.replication.enabled`` is True and at least one
2383 non-source destination resolves, creates one
2384 ``aws_ecr.CfnReplicationConfiguration`` rule with a single
2385 ``PREFIX_MATCH`` filter on ``gco/`` and one destination per resolved
2386 region. When replication is disabled or the destination list is
2387 empty (e.g. single-region deploy), no replication resource is
2388 provisioned and the method becomes a no-op.
2389 """
2390 if not self.images_config["replication"]["enabled"]:
2391 return
2393 destinations = self._resolve_replication_destinations(
2394 self.images_config["replication"]["destinations"]
2395 )
2396 if not destinations:
2397 return
2399 ecr.CfnReplicationConfiguration(
2400 self,
2401 "GcoImageReplicationConfig",
2402 replication_configuration=ecr.CfnReplicationConfiguration.ReplicationConfigurationProperty(
2403 rules=[
2404 ecr.CfnReplicationConfiguration.ReplicationRuleProperty(
2405 destinations=[
2406 ecr.CfnReplicationConfiguration.ReplicationDestinationProperty(
2407 region=region,
2408 registry_id=self.account,
2409 )
2410 for region in destinations
2411 ],
2412 repository_filters=[
2413 ecr.CfnReplicationConfiguration.RepositoryFilterProperty(
2414 # Replicate this deployment's own ECR namespace
2415 # (``<project_name>/*``) — ``gco/`` for the stock
2416 # project — so two deployments don't cross-replicate (#139).
2417 filter=f"{self.config.get_project_name()}/",
2418 filter_type="PREFIX_MATCH",
2419 )
2420 ],
2421 )
2422 ]
2423 ),
2424 )
2426 def _create_image_lookup_lambda(self) -> None:
2427 """Create the lookup-or-create custom resource Lambda for image repos.
2429 The Lambda implements the adopt-or-create pattern for ECR repos
2430 under the project's ``gco/*`` prefix. It is invoked at the time
2431 ``cli images init`` registers a new repo with the global stack via
2432 a ``CustomResource``; the function itself is provisioned here so
2433 the ARN is stable across deploys.
2435 The Lambda's IAM role grants read/write access to ECR repository
2436 APIs scoped to the project's prefix, plus the standard basic
2437 execution policy for CloudWatch Logs.
2438 """
2439 project_name = self.config.get_project_name()
2441 # IAM role for the Lambda — minimal ECR + CloudWatch Logs permissions.
2442 # ECR repository APIs scope by repository name, not ARN, so the
2443 # ``gco/*`` prefix scope is enforced via the ARN pattern in the
2444 # policy resource list.
2445 repo_arn = f"arn:{self.partition}:ecr:*:{self.account}:repository/{project_name}/*"
2447 self.image_lookup_lambda = lambda_.Function(
2448 self,
2449 "ImageLookupFunction",
2450 runtime=getattr(lambda_.Runtime, LAMBDA_PYTHON_RUNTIME),
2451 handler="handler.lambda_handler",
2452 code=lambda_source_code("lambda/image-lookup"),
2453 timeout=Duration.minutes(5),
2454 description=(
2455 "Lookup-or-create custom resource handler for ECR "
2456 "repositories under the project's gco/* prefix."
2457 ),
2458 )
2460 assert self.image_lookup_lambda.role is not None
2461 self.image_lookup_lambda.role.add_to_principal_policy(
2462 iam.PolicyStatement(
2463 effect=iam.Effect.ALLOW,
2464 actions=[
2465 "ecr:DescribeRepositories",
2466 "ecr:CreateRepository",
2467 "ecr:DeleteRepository",
2468 "ecr:PutLifecyclePolicy",
2469 "ecr:GetLifecyclePolicy",
2470 "ecr:TagResource",
2471 "ecr:ListTagsForResource",
2472 "ecr:BatchDeleteImage",
2473 "ecr:DescribeImages",
2474 "ecr:ListImages",
2475 ],
2476 resources=[repo_arn],
2477 )
2478 )
2480 CfnOutput(
2481 self,
2482 "ImageLookupFunctionArn",
2483 value=self.image_lookup_lambda.function_arn,
2484 description=(
2485 "Lambda ARN for the lookup-or-create custom resource that "
2486 "manages ECR repositories under the gco/* prefix."
2487 ),
2488 export_name=f"{project_name}-image-lookup-function-arn",
2489 )
2491 # The ECR repository policy uses a partition-aware
2492 # ``arn:<partition>:ecr:*:<account>:repository/gco/*`` resource
2493 # which cdk-nag flags as ``AwsSolutions-IAM5`` because of the trailing
2494 # ``*``. The wildcard here is the documented IAM way to express
2495 # "every repository in this project's prefix", which is exactly the
2496 # blast radius we want for a Lambda whose contract is to manage
2497 # ECR repos under that prefix. Suppression is scoped to the specific
2498 # ARN pattern (and to all ECR Describe/Read action wildcards in
2499 # the policy below) rather than a blanket ``Resource::*`` bypass.
2500 #
2501 # cdk-nag reports this finding's account as the ``<AWS::AccountId>``
2502 # placeholder for an environment-agnostic synth, but as the concrete
2503 # account id for an environment-specific one (the ARN above is
2504 # hand-built from ``self.account``, which becomes a literal once the
2505 # stack has a resolved account). We author the ``appliesTo`` with the
2506 # placeholder; ``acknowledge_nag_findings`` additionally registers the
2507 # concrete-account rendering when the account is resolved, so the
2508 # acknowledgment matches in both cases.
2509 from gco.stacks.nag_suppressions import acknowledge_nag_findings
2511 acknowledge_nag_findings(
2512 self.image_lookup_lambda.role,
2513 [
2514 {
2515 "id": "AwsSolutions-IAM5",
2516 "reason": (
2517 "The ImageLookupFunction's contract is to look up "
2518 "or create any ECR repository under the project's "
2519 "``gco/*`` prefix. The ARN pattern "
2520 "``arn:<partition>:ecr:*:<account>:repository/gco/*`` is "
2521 "the documented IAM way to express that scope: it "
2522 "covers exactly the repositories the function is "
2523 "allowed to touch and nothing else."
2524 ),
2525 "appliesTo": [
2526 f"Resource::arn:<AWS::Partition>:ecr:*:<AWS::AccountId>:"
2527 f"repository/{project_name}/*",
2528 ],
2529 },
2530 ],
2531 )