Coverage for gco / stacks / global_stack.py: 100.00%
350 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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)
42# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
43# Generated at (UTC): 2026-09-13T18:39:14Z
44# Generated from Git commit: c13ec54b216d5dd044916ef1e08705fb19e1dd0b
45# Flowchart(s) generated from this file:
46# * ``GCOGlobalStack.__init__`` -> ``diagrams/code_diagrams/gco/stacks/global_stack.GCOGlobalStack___init__.html``
47# (PNG: ``diagrams/code_diagrams/gco/stacks/global_stack.GCOGlobalStack___init__.png``)
48# * ``GCOGlobalStack._create_image_replication_rule`` -> ``diagrams/code_diagrams/gco/stacks/global_stack.GCOGlobalStack__create_image_replication_rule.html``
49# (PNG: ``diagrams/code_diagrams/gco/stacks/global_stack.GCOGlobalStack__create_image_replication_rule.png``)
50# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
51# <pyflowchart-code-diagram> END
54# Default values for the ``images`` cdk.json block. The defaults match the
55# documented retention posture: repos survive a stack destroy by default
56# (``retain``), non-empty repos block destroy unless the operator explicitly
57# flips ``empty_on_delete`` to true, lifecycle keeps the latest 20 tagged
58# images and expires untagged ones after 7 days, and replication is enabled
59# by default to every deployed region.
60_IMAGES_DEFAULT_REMOVAL_POLICY = "retain"
61_IMAGES_DEFAULT_EMPTY_ON_DELETE = False
62_IMAGES_DEFAULT_KEEP_TAGGED = 20
63_IMAGES_DEFAULT_EXPIRE_UNTAGGED_DAYS = 7
64_IMAGES_DEFAULT_REPLICATION_ENABLED = True
65_IMAGES_DEFAULT_REPLICATION_DESTINATIONS = "all_deployed_regions"
67_IMAGES_VALID_REMOVAL_POLICIES = ("retain", "destroy")
70def _parse_images_config(cdk_context: dict[str, Any] | None) -> dict[str, Any]:
71 """Parse the ``images`` block from cdk.json with defaults applied.
73 Returns a normalized dict shape that the rest of the global stack
74 can consume without re-parsing. Validates ``removal_policy`` against
75 the set ``{"retain", "destroy"}`` and ``replication.destinations``
76 against either the literal string ``"all_deployed_regions"`` or a
77 ``list[str]``.
79 Args:
80 cdk_context: The dict returned by ``self.node.try_get_context("images")``.
81 ``None`` (the key being absent) is equivalent to an empty dict.
83 Returns:
84 A dict with keys ``removal_policy``, ``empty_on_delete``,
85 ``lifecycle`` (with ``keep_tagged`` and ``expire_untagged_days``),
86 and ``replication`` (with ``enabled`` and ``destinations``).
87 """
88 raw = cdk_context or {}
90 removal_policy = raw.get("removal_policy", _IMAGES_DEFAULT_REMOVAL_POLICY)
91 if not isinstance(removal_policy, str) or removal_policy not in _IMAGES_VALID_REMOVAL_POLICIES:
92 raise ValueError(
93 f"images.removal_policy must be 'retain' or 'destroy', got {removal_policy!r}"
94 )
96 empty_on_delete = bool(raw.get("empty_on_delete", _IMAGES_DEFAULT_EMPTY_ON_DELETE))
98 lifecycle_raw = raw.get("lifecycle") or {}
99 if not isinstance(lifecycle_raw, dict):
100 raise ValueError(f"images.lifecycle must be a mapping, got {type(lifecycle_raw).__name__}")
101 keep_tagged = int(lifecycle_raw.get("keep_tagged", _IMAGES_DEFAULT_KEEP_TAGGED))
102 expire_untagged_days = int(
103 lifecycle_raw.get("expire_untagged_days", _IMAGES_DEFAULT_EXPIRE_UNTAGGED_DAYS)
104 )
106 replication_raw = raw.get("replication") or {}
107 if not isinstance(replication_raw, dict):
108 raise ValueError(
109 f"images.replication must be a mapping, got {type(replication_raw).__name__}"
110 )
111 replication_enabled = bool(replication_raw.get("enabled", _IMAGES_DEFAULT_REPLICATION_ENABLED))
112 destinations = replication_raw.get("destinations", _IMAGES_DEFAULT_REPLICATION_DESTINATIONS)
113 if isinstance(destinations, str):
114 if destinations != _IMAGES_DEFAULT_REPLICATION_DESTINATIONS:
115 raise ValueError(
116 "images.replication.destinations must be the string "
117 f"{_IMAGES_DEFAULT_REPLICATION_DESTINATIONS!r} or a list of region names, "
118 f"got {destinations!r}"
119 )
120 elif isinstance(destinations, list):
121 if not all(isinstance(item, str) for item in destinations):
122 raise ValueError(
123 "images.replication.destinations list must contain only region name strings"
124 )
125 else:
126 raise ValueError(
127 "images.replication.destinations must be the string "
128 f"{_IMAGES_DEFAULT_REPLICATION_DESTINATIONS!r} or a list of region names, "
129 f"got {type(destinations).__name__}"
130 )
132 return {
133 "removal_policy": removal_policy,
134 "empty_on_delete": empty_on_delete,
135 "lifecycle": {
136 "keep_tagged": keep_tagged,
137 "expire_untagged_days": expire_untagged_days,
138 },
139 "replication": {
140 "enabled": replication_enabled,
141 "destinations": destinations,
142 },
143 }
146class GCOGlobalStack(Stack):
147 """Global shared resources with optional AWS Global Accelerator.
149 This stack must be deployed before regional stacks. In the commercial
150 ``aws`` partition, regional stacks register their ALBs with endpoint groups
151 created here. Other partitions omit the accelerator topology.
153 Attributes:
154 accelerator: Optional Global Accelerator resource.
155 listener: Optional TCP/443 listener.
156 endpoint_groups: Region-to-endpoint-group mapping; empty without GA.
157 templates_table: DynamoDB table for job templates.
158 webhooks_table: DynamoDB table for webhooks.
159 missions_table: DynamoDB table for mission session state.
160 """
162 def __init__(
163 self, scope: Construct, construct_id: str, config: ConfigLoader, **kwargs: Any
164 ) -> None:
165 super().__init__(scope, construct_id, **kwargs)
167 self.config = config
168 self.regional_endpoints: dict[str, str] = {}
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_.Code.from_asset("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_.Code.from_asset("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 (design §0.1) 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_.Code.from_asset("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 add_regional_endpoint(self, region: str, alb_arn: str) -> None:
1502 """Add a regional ALB endpoint to the Global Accelerator.
1504 Note: Due to cross-region reference limitations in CDK, the actual endpoint
1505 registration is handled by a custom resource in the regional stack.
1506 This method stores the ARN for reference but doesn't directly register it.
1508 The regional stack should use the endpoint group ARN exported by this stack
1509 to register its ALB via an AwsCustomResource.
1510 """
1511 self.regional_endpoints[region] = alb_arn
1512 # Actual registration happens in regional stack via custom resource
1514 def get_accelerator_dns_name(self) -> str | None:
1515 """Return the Global Accelerator DNS name when this partition supports it."""
1516 return str(self.accelerator.dns_name) if self.accelerator is not None else None
1518 def get_accelerator_arn(self) -> str:
1519 """Get the Global Accelerator ARN."""
1520 if self.accelerator is None:
1521 raise RuntimeError("Global Accelerator is unavailable in this partition")
1522 return str(self.accelerator.accelerator_arn)
1524 def get_listener_arn(self) -> str:
1525 """Get the Global Accelerator Listener ARN."""
1526 if self.listener is None:
1527 raise RuntimeError("Global Accelerator is unavailable in this partition")
1528 return str(self.listener.listener_arn)
1530 def get_endpoint_group_arn(self, region: str) -> str:
1531 """Get the endpoint group ARN for a specific region"""
1532 if region in self.endpoint_groups:
1533 return str(self.endpoint_groups[region].endpoint_group_arn)
1534 raise ValueError(f"No endpoint group found for region: {region}")
1536 def _create_dynamodb_tables(self) -> None:
1537 """Create DynamoDB tables for templates, webhooks, jobs, inference endpoints, and missions."""
1538 project_name = self.config.get_project_name()
1540 # Job Templates table - stores reusable job templates
1541 self.templates_table = dynamodb.Table(
1542 self,
1543 "JobTemplatesTable",
1544 table_name=f"{project_name}-job-templates",
1545 partition_key=dynamodb.Attribute(
1546 name="template_name",
1547 type=dynamodb.AttributeType.STRING,
1548 ),
1549 billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
1550 removal_policy=RemovalPolicy.DESTROY,
1551 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
1552 point_in_time_recovery_enabled=True
1553 ),
1554 encryption=dynamodb.TableEncryption.AWS_MANAGED,
1555 )
1557 # Webhooks table - stores webhook registrations
1558 self.webhooks_table = dynamodb.Table(
1559 self,
1560 "WebhooksTable",
1561 table_name=f"{project_name}-webhooks",
1562 partition_key=dynamodb.Attribute(
1563 name="webhook_id",
1564 type=dynamodb.AttributeType.STRING,
1565 ),
1566 billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
1567 removal_policy=RemovalPolicy.DESTROY,
1568 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
1569 point_in_time_recovery_enabled=True
1570 ),
1571 encryption=dynamodb.TableEncryption.AWS_MANAGED,
1572 )
1574 # Add GSI for querying webhooks by namespace
1575 self.webhooks_table.add_global_secondary_index(
1576 index_name="namespace-index",
1577 partition_key=dynamodb.Attribute(
1578 name="namespace",
1579 type=dynamodb.AttributeType.STRING,
1580 ),
1581 projection_type=dynamodb.ProjectionType.ALL,
1582 )
1584 # Jobs table - centralized job tracking and queue
1585 # This enables global job submission with regional pickup
1586 self.jobs_table = dynamodb.Table(
1587 self,
1588 "JobsTable",
1589 table_name=f"{project_name}-jobs",
1590 partition_key=dynamodb.Attribute(
1591 name="job_id",
1592 type=dynamodb.AttributeType.STRING,
1593 ),
1594 billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
1595 removal_policy=RemovalPolicy.DESTROY,
1596 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
1597 point_in_time_recovery_enabled=True
1598 ),
1599 encryption=dynamodb.TableEncryption.AWS_MANAGED,
1600 time_to_live_attribute="ttl", # Auto-cleanup old completed jobs
1601 )
1603 # Legacy GSI retained for compatibility with existing deployments and
1604 # ad-hoc operational queries.
1605 self.jobs_table.add_global_secondary_index(
1606 index_name="region-status-index",
1607 partition_key=dynamodb.Attribute(
1608 name="target_region",
1609 type=dynamodb.AttributeType.STRING,
1610 ),
1611 sort_key=dynamodb.Attribute(
1612 name="status",
1613 type=dynamodb.AttributeType.STRING,
1614 ),
1615 projection_type=dynamodb.ProjectionType.ALL,
1616 )
1618 # ``work_sort`` is priority/FIFO for queued records and lease expiry for
1619 # claimed or applying records. This unified worker index is the only GSI
1620 # added by this release because DynamoDB permits only one GSI creation or
1621 # deletion per table update. Workers repeatedly backfill legacy rows
1622 # through the retained region-status-index during mixed-version rollouts.
1623 self.jobs_table.add_global_secondary_index(
1624 index_name="region-status-work-index",
1625 partition_key=dynamodb.Attribute(
1626 name="region_status",
1627 type=dynamodb.AttributeType.STRING,
1628 ),
1629 sort_key=dynamodb.Attribute(
1630 name="work_sort",
1631 type=dynamodb.AttributeType.STRING,
1632 ),
1633 projection_type=dynamodb.ProjectionType.ALL,
1634 )
1636 # GSI for querying jobs by namespace
1637 self.jobs_table.add_global_secondary_index(
1638 index_name="namespace-index",
1639 partition_key=dynamodb.Attribute(
1640 name="namespace",
1641 type=dynamodb.AttributeType.STRING,
1642 ),
1643 sort_key=dynamodb.Attribute(
1644 name="submitted_at",
1645 type=dynamodb.AttributeType.STRING,
1646 ),
1647 projection_type=dynamodb.ProjectionType.ALL,
1648 )
1650 # GSI for querying jobs by status globally
1651 self.jobs_table.add_global_secondary_index(
1652 index_name="status-index",
1653 partition_key=dynamodb.Attribute(
1654 name="status",
1655 type=dynamodb.AttributeType.STRING,
1656 ),
1657 sort_key=dynamodb.Attribute(
1658 name="submitted_at",
1659 type=dynamodb.AttributeType.STRING,
1660 ),
1661 projection_type=dynamodb.ProjectionType.ALL,
1662 )
1664 # Export table names and ARNs for regional stacks
1665 CfnOutput(
1666 self,
1667 "TemplatesTableName",
1668 value=self.templates_table.table_name,
1669 description="DynamoDB table name for job templates",
1670 export_name=f"{project_name}-templates-table-name",
1671 )
1673 CfnOutput(
1674 self,
1675 "TemplatesTableArn",
1676 value=self.templates_table.table_arn,
1677 description="DynamoDB table ARN for job templates",
1678 export_name=f"{project_name}-templates-table-arn",
1679 )
1681 CfnOutput(
1682 self,
1683 "WebhooksTableName",
1684 value=self.webhooks_table.table_name,
1685 description="DynamoDB table name for webhooks",
1686 export_name=f"{project_name}-webhooks-table-name",
1687 )
1689 CfnOutput(
1690 self,
1691 "WebhooksTableArn",
1692 value=self.webhooks_table.table_arn,
1693 description="DynamoDB table ARN for webhooks",
1694 export_name=f"{project_name}-webhooks-table-arn",
1695 )
1697 CfnOutput(
1698 self,
1699 "JobsTableName",
1700 value=self.jobs_table.table_name,
1701 description="DynamoDB table name for centralized job tracking",
1702 export_name=f"{project_name}-jobs-table-name",
1703 )
1705 CfnOutput(
1706 self,
1707 "JobsTableArn",
1708 value=self.jobs_table.table_arn,
1709 description="DynamoDB table ARN for centralized job tracking",
1710 export_name=f"{project_name}-jobs-table-arn",
1711 )
1713 # Inference Endpoints table - stores desired state for inference deployments
1714 # The inference_monitor in each regional cluster polls this table
1715 self.inference_endpoints_table = dynamodb.Table(
1716 self,
1717 "InferenceEndpointsTable",
1718 table_name=f"{project_name}-inference-endpoints",
1719 partition_key=dynamodb.Attribute(
1720 name="endpoint_name",
1721 type=dynamodb.AttributeType.STRING,
1722 ),
1723 billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
1724 removal_policy=RemovalPolicy.DESTROY,
1725 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
1726 point_in_time_recovery_enabled=True
1727 ),
1728 encryption=dynamodb.TableEncryption.AWS_MANAGED,
1729 )
1731 CfnOutput(
1732 self,
1733 "InferenceEndpointsTableName",
1734 value=self.inference_endpoints_table.table_name,
1735 description="DynamoDB table name for inference endpoint state",
1736 export_name=f"{project_name}-inference-endpoints-table-name",
1737 )
1739 CfnOutput(
1740 self,
1741 "InferenceEndpointsTableArn",
1742 value=self.inference_endpoints_table.table_arn,
1743 description="DynamoDB table ARN for inference endpoint state",
1744 export_name=f"{project_name}-inference-endpoints-table-arn",
1745 )
1747 # Missions table - persists goal-directed iteration session state
1748 # Partition by session_id; the status-index GSI supports paginated
1749 # listing by status (e.g. running, completed, terminated, failed).
1750 self.missions_table = dynamodb.Table(
1751 self,
1752 "MissionsTable",
1753 table_name=f"{project_name}-missions",
1754 partition_key=dynamodb.Attribute(
1755 name="session_id",
1756 type=dynamodb.AttributeType.STRING,
1757 ),
1758 billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
1759 removal_policy=RemovalPolicy.DESTROY,
1760 point_in_time_recovery_specification=dynamodb.PointInTimeRecoverySpecification(
1761 point_in_time_recovery_enabled=True
1762 ),
1763 encryption=dynamodb.TableEncryption.AWS_MANAGED,
1764 )
1766 # GSI for paginating sessions by status (sorted by creation time)
1767 self.missions_table.add_global_secondary_index(
1768 index_name="status-index",
1769 partition_key=dynamodb.Attribute(
1770 name="status",
1771 type=dynamodb.AttributeType.STRING,
1772 ),
1773 sort_key=dynamodb.Attribute(
1774 name="created_at",
1775 type=dynamodb.AttributeType.STRING,
1776 ),
1777 projection_type=dynamodb.ProjectionType.ALL,
1778 )
1780 CfnOutput(
1781 self,
1782 "MissionsTableName",
1783 value=self.missions_table.table_name,
1784 description="DynamoDB table name for mission session state",
1785 export_name=f"{project_name}-missions-table-name",
1786 )
1788 CfnOutput(
1789 self,
1790 "MissionsTableArn",
1791 value=self.missions_table.table_arn,
1792 description="DynamoDB table ARN for mission session state",
1793 export_name=f"{project_name}-missions-table-arn",
1794 )
1796 # Store table names in SSM for cross-region access
1797 ssm.StringParameter(
1798 self,
1799 "TemplatesTableNameParam",
1800 parameter_name=f"/{project_name}/templates-table-name",
1801 string_value=self.templates_table.table_name,
1802 description="DynamoDB table name for job templates",
1803 )
1805 ssm.StringParameter(
1806 self,
1807 "WebhooksTableNameParam",
1808 parameter_name=f"/{project_name}/webhooks-table-name",
1809 string_value=self.webhooks_table.table_name,
1810 description="DynamoDB table name for webhooks",
1811 )
1813 ssm.StringParameter(
1814 self,
1815 "JobsTableNameParam",
1816 parameter_name=f"/{project_name}/jobs-table-name",
1817 string_value=self.jobs_table.table_name,
1818 description="DynamoDB table name for centralized job tracking",
1819 )
1821 ssm.StringParameter(
1822 self,
1823 "InferenceEndpointsTableNameParam",
1824 parameter_name=f"/{project_name}/inference-endpoints-table-name",
1825 string_value=self.inference_endpoints_table.table_name,
1826 description="DynamoDB table name for inference endpoint state",
1827 )
1829 ssm.StringParameter(
1830 self,
1831 "MissionsTableNameParam",
1832 parameter_name=f"/{project_name}/missions-table-name",
1833 string_value=self.missions_table.table_name,
1834 description="DynamoDB table name for mission session state",
1835 )
1837 def _create_model_bucket(self) -> None:
1838 """Create S3 bucket for model weights.
1840 This bucket serves as the central model registry. Users upload model
1841 weights here once, and the inference_monitor's init containers sync
1842 them to each region's local EFS at pod startup.
1844 The bucket name is auto-generated by CDK to avoid naming collisions.
1845 It's exported via CfnOutput and SSM for CLI discovery.
1846 """
1847 project_name = self.config.get_project_name()
1849 # KMS key for model bucket encryption
1850 self.model_bucket_key = kms.Key(
1851 self,
1852 "ModelBucketKey",
1853 description="KMS key for GCO model weights bucket",
1854 enable_key_rotation=True,
1855 removal_policy=RemovalPolicy.DESTROY,
1856 )
1858 # Access logs bucket (required for compliance)
1859 # Retention is configurable via cdk.json context field `s3_access_logs.retention_days`
1860 # (default: 90 days). Logs older than the configured retention are expired.
1861 s3_access_logs_ctx = self.node.try_get_context("s3_access_logs") or {}
1862 access_logs_retention_days = int(s3_access_logs_ctx.get("retention_days", 90))
1864 self.model_bucket_access_logs = s3.Bucket(
1865 self,
1866 "ModelWeightsAccessLogsBucket",
1867 encryption=s3.BucketEncryption.S3_MANAGED,
1868 block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
1869 enforce_ssl=True,
1870 versioned=True,
1871 removal_policy=RemovalPolicy.DESTROY,
1872 auto_delete_objects=True,
1873 lifecycle_rules=[
1874 s3.LifecycleRule(
1875 id="ExpireAccessLogs",
1876 enabled=True,
1877 expiration=Duration.days(access_logs_retention_days),
1878 )
1879 ],
1880 )
1882 # Model weights bucket
1883 self.model_bucket = s3.Bucket(
1884 self,
1885 "ModelWeightsBucket",
1886 encryption=s3.BucketEncryption.KMS,
1887 encryption_key=self.model_bucket_key,
1888 bucket_key_enabled=True,
1889 block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
1890 enforce_ssl=True,
1891 versioned=True,
1892 removal_policy=RemovalPolicy.DESTROY,
1893 auto_delete_objects=True,
1894 server_access_logs_bucket=self.model_bucket_access_logs,
1895 server_access_logs_prefix="model-bucket-logs/",
1896 )
1898 # CDK-nag suppressions — only replication (not needed for model weights)
1899 from gco.stacks.nag_suppressions import acknowledge_nag_findings
1901 replication_reason = (
1902 "Model weights are user-uploaded artifacts that can be re-uploaded. "
1903 "Cross-region replication is not required; the inference_monitor "
1904 "syncs models from S3 to each region's EFS at pod startup."
1905 )
1907 acknowledge_nag_findings(
1908 self.model_bucket,
1909 [
1910 {
1911 "id": "HIPAA.Security-S3BucketReplicationEnabled",
1912 "reason": replication_reason,
1913 },
1914 {
1915 "id": "NIST.800.53.R5-S3BucketReplicationEnabled",
1916 "reason": replication_reason,
1917 },
1918 {
1919 "id": "PCI.DSS.321-S3BucketReplicationEnabled",
1920 "reason": replication_reason,
1921 },
1922 ],
1923 )
1925 logs_reason = "This is the server access logs destination bucket."
1926 acknowledge_nag_findings(
1927 self.model_bucket_access_logs,
1928 [
1929 {"id": "AwsSolutions-S1", "reason": logs_reason},
1930 {"id": "HIPAA.Security-S3BucketLoggingEnabled", "reason": logs_reason},
1931 {
1932 "id": "HIPAA.Security-S3BucketReplicationEnabled",
1933 "reason": "Access logs do not require replication.",
1934 },
1935 {
1936 "id": "HIPAA.Security-S3DefaultEncryptionKMS",
1937 "reason": "SSE-S3 is sufficient for access logs.",
1938 },
1939 {"id": "NIST.800.53.R5-S3BucketLoggingEnabled", "reason": logs_reason},
1940 {
1941 "id": "NIST.800.53.R5-S3BucketReplicationEnabled",
1942 "reason": "Access logs do not require replication.",
1943 },
1944 {
1945 "id": "NIST.800.53.R5-S3DefaultEncryptionKMS",
1946 "reason": "SSE-S3 is sufficient for access logs.",
1947 },
1948 {"id": "PCI.DSS.321-S3BucketLoggingEnabled", "reason": logs_reason},
1949 {
1950 "id": "PCI.DSS.321-S3BucketReplicationEnabled",
1951 "reason": "Access logs do not require replication.",
1952 },
1953 {
1954 "id": "PCI.DSS.321-S3DefaultEncryptionKMS",
1955 "reason": "SSE-S3 is sufficient for access logs.",
1956 },
1957 ],
1958 )
1960 CfnOutput(
1961 self,
1962 "ModelBucketName",
1963 value=self.model_bucket.bucket_name,
1964 description="S3 bucket for model weights",
1965 export_name=f"{project_name}-model-bucket-name",
1966 )
1968 CfnOutput(
1969 self,
1970 "ModelBucketArn",
1971 value=self.model_bucket.bucket_arn,
1972 description="S3 bucket ARN for model weights",
1973 export_name=f"{project_name}-model-bucket-arn",
1974 )
1976 ssm.StringParameter(
1977 self,
1978 "ModelBucketNameParam",
1979 parameter_name=f"/{project_name}/model-bucket-name",
1980 string_value=self.model_bucket.bucket_name,
1981 description="S3 bucket name for model weights",
1982 )
1984 def _create_backup_plan(self) -> None:
1985 """Create AWS Backup plan for DynamoDB tables.
1987 Creates a backup plan with:
1988 - Daily backups retained for 35 days
1989 - Weekly backups retained for 90 days
1990 - All DynamoDB tables added to the backup selection
1991 """
1992 # Create backup vault for storing backups
1993 self.backup_vault = backup.BackupVault(
1994 self,
1995 "DynamoDBBackupVault",
1996 removal_policy=RemovalPolicy.DESTROY,
1997 )
1999 # Create backup plan with daily and weekly rules
2000 self.backup_plan = backup.BackupPlan(
2001 self,
2002 "DynamoDBBackupPlan",
2003 backup_plan_rules=[
2004 # Daily backup - retained for 35 days
2005 backup.BackupPlanRule(
2006 rule_name="DailyBackup",
2007 backup_vault=self.backup_vault,
2008 schedule_expression=events.Schedule.cron(
2009 hour="3",
2010 minute="0",
2011 ),
2012 delete_after=Duration.days(35),
2013 enable_continuous_backup=True, # Enable PITR for DynamoDB
2014 ),
2015 # Weekly backup - retained for 90 days
2016 backup.BackupPlanRule(
2017 rule_name="WeeklyBackup",
2018 backup_vault=self.backup_vault,
2019 schedule_expression=events.Schedule.cron(
2020 hour="4",
2021 minute="0",
2022 week_day="SUN",
2023 ),
2024 delete_after=Duration.days(90),
2025 ),
2026 ],
2027 )
2029 # Add all DynamoDB tables to the backup selection
2030 self.backup_plan.add_selection(
2031 "DynamoDBTablesSelection",
2032 resources=[
2033 backup.BackupResource.from_dynamo_db_table(self.templates_table),
2034 backup.BackupResource.from_dynamo_db_table(self.webhooks_table),
2035 backup.BackupResource.from_dynamo_db_table(self.jobs_table),
2036 backup.BackupResource.from_dynamo_db_table(self.inference_endpoints_table),
2037 backup.BackupResource.from_dynamo_db_table(self.missions_table),
2038 ],
2039 )
2041 # Export backup plan ARN
2042 project_name = self.config.get_project_name()
2043 CfnOutput(
2044 self,
2045 "BackupPlanArn",
2046 value=self.backup_plan.backup_plan_arn,
2047 description="AWS Backup plan ARN for DynamoDB tables",
2048 export_name=f"{project_name}-backup-plan-arn",
2049 )
2051 CfnOutput(
2052 self,
2053 "BackupVaultArn",
2054 value=self.backup_vault.backup_vault_arn,
2055 description="AWS Backup vault ARN for DynamoDB backups",
2056 export_name=f"{project_name}-backup-vault-arn",
2057 )
2059 def _create_cluster_shared_kms_key(self) -> None:
2060 """Create the always-on customer-managed KMS key for ``Cluster_Shared_Bucket``.
2062 The key:
2063 - Enables automatic annual rotation.
2064 - Uses a 7-day pending window on destroy — the AWS minimum, matching the
2065 destroy-by-default iteration-loop posture of the analytics-environment
2066 feature while still providing a safety net against accidental deletion.
2067 - Uses ``RemovalPolicy.DESTROY`` so a ``cdk destroy gco-global`` cleans up
2068 the key without operator intervention (iteration-loop posture).
2069 - Grants encrypt/decrypt to the ``s3.amazonaws.com`` and
2070 ``logs.<region>.amazonaws.com`` service principals via the key policy
2071 so S3 server-side encryption and CloudWatch access-log delivery can use
2072 the key without role-side grants.
2074 The key is exposed as ``self.cluster_shared_kms_key`` for tests and for
2075 ``_create_cluster_shared_bucket`` to reference. Role-side usage grants
2076 (``kms:Decrypt`` / ``kms:GenerateDataKey``) are attached by downstream
2077 consumers: ``GCORegionalStack`` on the job-pod role (always-on)
2078 and ``GCOAnalyticsStack`` on the SageMaker execution role (conditional on
2079 the analytics toggle).
2080 """
2081 self.cluster_shared_kms_key = kms.Key(
2082 self,
2083 "ClusterSharedKmsKey",
2084 description=(
2085 "Customer-managed KMS key for the always-on Cluster_Shared_Bucket "
2086 "in GCOGlobalStack. Consumed by every regional EKS cluster and by "
2087 "GCOAnalyticsStack when analytics is enabled."
2088 ),
2089 enable_key_rotation=True,
2090 pending_window=Duration.days(7),
2091 removal_policy=RemovalPolicy.DESTROY,
2092 )
2094 # Key-policy grants for service principals that need to encrypt/decrypt
2095 # on behalf of the bucket (S3 server-side encryption) and the access-logs
2096 # bucket (CloudWatch Logs delivery). The actions match the standard
2097 # service-principal pattern used by cdk's default key policies.
2098 kms_actions = [
2099 "kms:Encrypt",
2100 "kms:Decrypt",
2101 "kms:ReEncrypt*",
2102 "kms:GenerateDataKey*",
2103 "kms:DescribeKey",
2104 ]
2106 self.cluster_shared_kms_key.add_to_resource_policy(
2107 iam.PolicyStatement(
2108 sid="AllowS3ServiceEncryptDecrypt",
2109 effect=iam.Effect.ALLOW,
2110 principals=[iam.ServicePrincipal("s3.amazonaws.com")],
2111 actions=kms_actions,
2112 resources=["*"],
2113 )
2114 )
2116 self.cluster_shared_kms_key.add_to_resource_policy(
2117 iam.PolicyStatement(
2118 sid="AllowCloudWatchLogsEncryptDecrypt",
2119 effect=iam.Effect.ALLOW,
2120 principals=[iam.ServicePrincipal("logs.amazonaws.com", region=self.region)],
2121 actions=kms_actions,
2122 resources=["*"],
2123 )
2124 )
2126 def _create_cluster_shared_bucket(self) -> None:
2127 """Create the always-on ``Cluster_Shared_Bucket`` and its access-logs bucket.
2129 Two buckets are created:
2131 1. ``cluster_shared_access_logs_bucket`` — dedicated S3 access-logs bucket
2132 used as ``server_access_logs_bucket`` for the primary bucket. Separate
2133 from ``model_bucket_access_logs`` so cluster-shared-bucket access logs
2134 are not commingled with model-bucket logs.
2135 2. ``cluster_shared_bucket`` — the primary bucket. Its physical name
2136 is CloudFormation-generated (``<stack>-clustersharedbucket…``): S3
2137 bucket names are a global namespace and a deleted name is not
2138 reliably reusable, so a fixed project/account/region name would
2139 make every destroy-and-redeploy a collision hazard. Consumers never
2140 reconstruct it — every regional stack and the analytics stack read
2141 the SSM parameters published under
2142 ``cluster_shared_ssm_parameter_prefix(project_name)``. KMS-encrypted
2143 with ``cluster_shared_kms_key``, block-public-access on, SSL
2144 enforced, versioned, destroy-on-teardown.
2146 An explicit ``Deny`` statement for ``aws:SecureTransport=false`` is added
2147 to the bucket policy independent of ``enforce_ssl=True`` so the deny is
2148 verifiable in the synthesized template (belt-and-suspenders).
2150 Grants on ``Cluster_Shared_Bucket`` are intentionally not added here —
2151 they live on downstream role policies (``GCORegionalStack`` on the
2152 job-pod role, ``GCOAnalyticsStack`` on the SageMaker execution role)
2153 rather than in this bucket's policy. The bucket policy contains zero
2154 ``Principal: "*"`` Allow statements.
2155 """
2156 # Retention for the access-logs bucket honors the same `s3_access_logs`
2157 # context field as the model-bucket access-logs bucket (default 90 days).
2158 s3_access_logs_ctx = self.node.try_get_context("s3_access_logs") or {}
2159 access_logs_retention_days = int(s3_access_logs_ctx.get("retention_days", 90))
2161 # Dedicated access-logs bucket for Cluster_Shared_Bucket. Encrypted with
2162 # the cluster-shared KMS key (the key policy grants the logs service
2163 # principal encrypt/decrypt). Kept separate from model_bucket_access_logs
2164 # so operators can reason about each bucket's logs independently. Matches
2165 # the LifecycleRule used on `model_bucket_access_logs` so retention is
2166 # consistent across the two log sinks.
2167 self.cluster_shared_access_logs_bucket = s3.Bucket(
2168 self,
2169 "ClusterSharedAccessLogsBucket",
2170 encryption=s3.BucketEncryption.KMS,
2171 encryption_key=self.cluster_shared_kms_key,
2172 block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
2173 enforce_ssl=True,
2174 versioned=True,
2175 removal_policy=RemovalPolicy.DESTROY,
2176 auto_delete_objects=True,
2177 lifecycle_rules=[
2178 s3.LifecycleRule(
2179 id="ExpireAccessLogs",
2180 enabled=True,
2181 expiration=Duration.days(access_logs_retention_days),
2182 )
2183 ],
2184 )
2186 # Primary Cluster_Shared_Bucket. No ``bucket_name``: the physical name
2187 # is CloudFormation-generated so a destroy-and-redeploy can never
2188 # collide in S3's global namespace (the model bucket has always worked
2189 # this way). Downstream grants use the ARN resolved from the SSM
2190 # parameters published below, never a reconstructed name.
2191 # `bucket_key_enabled=True` mirrors the model_bucket pattern to reduce
2192 # per-object KMS request costs.
2193 self.cluster_shared_bucket = s3.Bucket(
2194 self,
2195 "ClusterSharedBucket",
2196 encryption=s3.BucketEncryption.KMS,
2197 encryption_key=self.cluster_shared_kms_key,
2198 bucket_key_enabled=True,
2199 block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
2200 enforce_ssl=True,
2201 versioned=True,
2202 removal_policy=RemovalPolicy.DESTROY,
2203 auto_delete_objects=True,
2204 server_access_logs_bucket=self.cluster_shared_access_logs_bucket,
2205 server_access_logs_prefix="cluster-shared/",
2206 )
2208 # Explicit Deny for insecure transport. `enforce_ssl=True` already adds
2209 # an equivalent statement, but duplicating it here makes the deny
2210 # verifiable in the synthesized template under a known SID and satisfies
2211 # a belt-and-suspenders posture.
2212 self.cluster_shared_bucket.add_to_resource_policy(
2213 iam.PolicyStatement(
2214 sid="DenyInsecureTransport",
2215 effect=iam.Effect.DENY,
2216 principals=[iam.AnyPrincipal()],
2217 actions=["s3:*"],
2218 resources=[
2219 self.cluster_shared_bucket.bucket_arn,
2220 f"{self.cluster_shared_bucket.bucket_arn}/*",
2221 ],
2222 conditions={"Bool": {"aws:SecureTransport": "false"}},
2223 )
2224 )
2226 # CDK-nag suppressions — scoped per-resource at the construct site to
2227 # mirror the ``_create_model_bucket`` pattern (keeps the suppression
2228 # co-located with the construct it applies to, so the reason survives
2229 # refactors). Every suppression carries an explicit reason
2230 # string; no blanket ``Resource::*`` bypasses.
2231 from gco.stacks.nag_suppressions import acknowledge_nag_findings
2233 shared_replication_reason = (
2234 "Cluster_Shared_Bucket is a regional scratch sink; cluster jobs "
2235 "publish to it from a single region, and there is no durability "
2236 "requirement that warrants cross-region replication. Access logs "
2237 "do not require replication for the same reason."
2238 )
2240 acknowledge_nag_findings(
2241 self.cluster_shared_bucket,
2242 [
2243 {
2244 "id": "HIPAA.Security-S3BucketReplicationEnabled",
2245 "reason": shared_replication_reason,
2246 },
2247 {
2248 "id": "NIST.800.53.R5-S3BucketReplicationEnabled",
2249 "reason": shared_replication_reason,
2250 },
2251 {
2252 "id": "PCI.DSS.321-S3BucketReplicationEnabled",
2253 "reason": shared_replication_reason,
2254 },
2255 ],
2256 )
2258 access_logs_is_self_target_reason = (
2259 "This is the server access logs destination bucket for Cluster_Shared_Bucket."
2260 )
2261 acknowledge_nag_findings(
2262 self.cluster_shared_access_logs_bucket,
2263 [
2264 {
2265 "id": "AwsSolutions-S1",
2266 "reason": access_logs_is_self_target_reason,
2267 },
2268 {
2269 "id": "HIPAA.Security-S3BucketLoggingEnabled",
2270 "reason": access_logs_is_self_target_reason,
2271 },
2272 {
2273 "id": "NIST.800.53.R5-S3BucketLoggingEnabled",
2274 "reason": access_logs_is_self_target_reason,
2275 },
2276 {
2277 "id": "PCI.DSS.321-S3BucketLoggingEnabled",
2278 "reason": access_logs_is_self_target_reason,
2279 },
2280 {
2281 "id": "HIPAA.Security-S3BucketReplicationEnabled",
2282 "reason": shared_replication_reason,
2283 },
2284 {
2285 "id": "NIST.800.53.R5-S3BucketReplicationEnabled",
2286 "reason": shared_replication_reason,
2287 },
2288 {
2289 "id": "PCI.DSS.321-S3BucketReplicationEnabled",
2290 "reason": shared_replication_reason,
2291 },
2292 ],
2293 )
2295 def _publish_cluster_shared_bucket_ssm_params(self) -> None:
2296 """Publish the three ``/gco/cluster-shared-bucket/*`` SSM parameters.
2298 Writes:
2300 - ``/gco/cluster-shared-bucket/name`` — bucket name
2301 - ``/gco/cluster-shared-bucket/arn`` — bucket ARN
2302 - ``/gco/cluster-shared-bucket/region`` — bucket home region (global region)
2304 These parameters are the cross-region contract consumed by
2305 ``GCORegionalStack._resolve_cluster_shared_bucket_from_ssm`` (always) and by
2306 ``GCOAnalyticsStack._grant_sagemaker_role_on_cluster_shared_bucket``
2307 (conditional on the analytics toggle). The prefix from
2308 ``cluster_shared_ssm_parameter_prefix(project_name)`` is the single
2309 source of truth so the namespace can be renamed in exactly one place.
2311 Also emits four ``CfnOutput`` values for discoverability: the three SSM
2312 values plus the KMS key ARN. Export names follow the existing
2313 ``{project_name}-cluster-shared-{suffix}`` pattern used by the rest of
2314 this stack's outputs so operators can cross-reference them from peer
2315 stacks via ``Fn.import_value`` if needed (the primary cross-region
2316 contract remains SSM).
2317 """
2318 project_name = self.config.get_project_name()
2320 ssm.StringParameter(
2321 self,
2322 "ClusterSharedBucketNameParam",
2323 parameter_name=f"{cluster_shared_ssm_parameter_prefix(project_name)}/name",
2324 string_value=self.cluster_shared_bucket.bucket_name,
2325 description="Name of the always-on Cluster_Shared_Bucket (owned by GCOGlobalStack).",
2326 )
2328 ssm.StringParameter(
2329 self,
2330 "ClusterSharedBucketArnParam",
2331 parameter_name=f"{cluster_shared_ssm_parameter_prefix(project_name)}/arn",
2332 string_value=self.cluster_shared_bucket.bucket_arn,
2333 description="ARN of the always-on Cluster_Shared_Bucket (owned by GCOGlobalStack).",
2334 )
2336 ssm.StringParameter(
2337 self,
2338 "ClusterSharedBucketRegionParam",
2339 parameter_name=f"{cluster_shared_ssm_parameter_prefix(project_name)}/region",
2340 string_value=self.region,
2341 description="Home region of the always-on Cluster_Shared_Bucket (the global region).",
2342 )
2344 CfnOutput(
2345 self,
2346 "ClusterSharedBucketName",
2347 value=self.cluster_shared_bucket.bucket_name,
2348 description="Name of the always-on Cluster_Shared_Bucket.",
2349 export_name=f"{project_name}-cluster-shared-bucket-name",
2350 )
2352 CfnOutput(
2353 self,
2354 "ClusterSharedBucketArn",
2355 value=self.cluster_shared_bucket.bucket_arn,
2356 description="ARN of the always-on Cluster_Shared_Bucket.",
2357 export_name=f"{project_name}-cluster-shared-bucket-arn",
2358 )
2360 CfnOutput(
2361 self,
2362 "ClusterSharedBucketRegion",
2363 value=self.region,
2364 description="Home region of the always-on Cluster_Shared_Bucket.",
2365 export_name=f"{project_name}-cluster-shared-bucket-region",
2366 )
2368 CfnOutput(
2369 self,
2370 "ClusterSharedKmsKeyArn",
2371 value=self.cluster_shared_kms_key.key_arn,
2372 description="ARN of the always-on KMS key encrypting Cluster_Shared_Bucket.",
2373 export_name=f"{project_name}-cluster-shared-kms-key-arn",
2374 )
2376 def _resolve_replication_destinations(self, destinations: str | list[str]) -> list[str]:
2377 """Resolve the configured replication destinations into a region list.
2379 When ``destinations`` is the literal ``"all_deployed_regions"``, the
2380 list comes from ``self.config.get_regions()`` (the same source the
2381 rest of the stack uses for cross-region wiring). When it is an
2382 explicit list, it is returned as-is. The source region (the global
2383 stack's deploy region) is excluded — ECR replication is point-to-point
2384 and a self-referential destination is rejected by the API.
2385 """
2386 if isinstance(destinations, str):
2387 candidate_regions = list(self.config.get_regions())
2388 else:
2389 candidate_regions = list(destinations)
2390 return [region for region in candidate_regions if region != self.region]
2392 def _create_image_replication_rule(self) -> None:
2393 """Provision the ECR replication rule for ``gco/*`` repositories.
2395 When ``images.replication.enabled`` is True and at least one
2396 non-source destination resolves, creates one
2397 ``aws_ecr.CfnReplicationConfiguration`` rule with a single
2398 ``PREFIX_MATCH`` filter on ``gco/`` and one destination per resolved
2399 region. When replication is disabled or the destination list is
2400 empty (e.g. single-region deploy), no replication resource is
2401 provisioned and the method becomes a no-op.
2402 """
2403 if not self.images_config["replication"]["enabled"]:
2404 return
2406 destinations = self._resolve_replication_destinations(
2407 self.images_config["replication"]["destinations"]
2408 )
2409 if not destinations:
2410 return
2412 ecr.CfnReplicationConfiguration(
2413 self,
2414 "GcoImageReplicationConfig",
2415 replication_configuration=ecr.CfnReplicationConfiguration.ReplicationConfigurationProperty(
2416 rules=[
2417 ecr.CfnReplicationConfiguration.ReplicationRuleProperty(
2418 destinations=[
2419 ecr.CfnReplicationConfiguration.ReplicationDestinationProperty(
2420 region=region,
2421 registry_id=self.account,
2422 )
2423 for region in destinations
2424 ],
2425 repository_filters=[
2426 ecr.CfnReplicationConfiguration.RepositoryFilterProperty(
2427 # Replicate this deployment's own ECR namespace
2428 # (``<project_name>/*``) — ``gco/`` for the stock
2429 # project — so two deployments don't cross-replicate (#139).
2430 filter=f"{self.config.get_project_name()}/",
2431 filter_type="PREFIX_MATCH",
2432 )
2433 ],
2434 )
2435 ]
2436 ),
2437 )
2439 def _create_image_lookup_lambda(self) -> None:
2440 """Create the lookup-or-create custom resource Lambda for image repos.
2442 The Lambda implements the adopt-or-create pattern for ECR repos
2443 under the project's ``gco/*`` prefix. It is invoked at the time
2444 ``cli images init`` registers a new repo with the global stack via
2445 a ``CustomResource``; the function itself is provisioned here so
2446 the ARN is stable across deploys.
2448 The Lambda's IAM role grants read/write access to ECR repository
2449 APIs scoped to the project's prefix, plus the standard basic
2450 execution policy for CloudWatch Logs.
2451 """
2452 project_name = self.config.get_project_name()
2454 # IAM role for the Lambda — minimal ECR + CloudWatch Logs permissions.
2455 # ECR repository APIs scope by repository name, not ARN, so the
2456 # ``gco/*`` prefix scope is enforced via the ARN pattern in the
2457 # policy resource list.
2458 repo_arn = f"arn:{self.partition}:ecr:*:{self.account}:repository/{project_name}/*"
2460 self.image_lookup_lambda = lambda_.Function(
2461 self,
2462 "ImageLookupFunction",
2463 runtime=getattr(lambda_.Runtime, LAMBDA_PYTHON_RUNTIME),
2464 handler="handler.lambda_handler",
2465 code=lambda_.Code.from_asset("lambda/image-lookup"),
2466 timeout=Duration.minutes(5),
2467 description=(
2468 "Lookup-or-create custom resource handler for ECR "
2469 "repositories under the project's gco/* prefix."
2470 ),
2471 )
2473 assert self.image_lookup_lambda.role is not None
2474 self.image_lookup_lambda.role.add_to_principal_policy(
2475 iam.PolicyStatement(
2476 effect=iam.Effect.ALLOW,
2477 actions=[
2478 "ecr:DescribeRepositories",
2479 "ecr:CreateRepository",
2480 "ecr:DeleteRepository",
2481 "ecr:PutLifecyclePolicy",
2482 "ecr:GetLifecyclePolicy",
2483 "ecr:TagResource",
2484 "ecr:ListTagsForResource",
2485 "ecr:BatchDeleteImage",
2486 "ecr:DescribeImages",
2487 "ecr:ListImages",
2488 ],
2489 resources=[repo_arn],
2490 )
2491 )
2493 CfnOutput(
2494 self,
2495 "ImageLookupFunctionArn",
2496 value=self.image_lookup_lambda.function_arn,
2497 description=(
2498 "Lambda ARN for the lookup-or-create custom resource that "
2499 "manages ECR repositories under the gco/* prefix."
2500 ),
2501 export_name=f"{project_name}-image-lookup-function-arn",
2502 )
2504 # The ECR repository policy uses a partition-aware
2505 # ``arn:<partition>:ecr:*:<account>:repository/gco/*`` resource
2506 # which cdk-nag flags as ``AwsSolutions-IAM5`` because of the trailing
2507 # ``*``. The wildcard here is the documented IAM way to express
2508 # "every repository in this project's prefix", which is exactly the
2509 # blast radius we want for a Lambda whose contract is to manage
2510 # ECR repos under that prefix. Suppression is scoped to the specific
2511 # ARN pattern (and to all ECR Describe/Read action wildcards in
2512 # the policy below) rather than a blanket ``Resource::*`` bypass.
2513 #
2514 # cdk-nag reports this finding's account as the ``<AWS::AccountId>``
2515 # placeholder for an environment-agnostic synth, but as the concrete
2516 # account id for an environment-specific one (the ARN above is
2517 # hand-built from ``self.account``, which becomes a literal once the
2518 # stack has a resolved account). We author the ``appliesTo`` with the
2519 # placeholder; ``acknowledge_nag_findings`` additionally registers the
2520 # concrete-account rendering when the account is resolved, so the
2521 # acknowledgment matches in both cases.
2522 from gco.stacks.nag_suppressions import acknowledge_nag_findings
2524 acknowledge_nag_findings(
2525 self.image_lookup_lambda.role,
2526 [
2527 {
2528 "id": "AwsSolutions-IAM5",
2529 "reason": (
2530 "The ImageLookupFunction's contract is to look up "
2531 "or create any ECR repository under the project's "
2532 "``gco/*`` prefix. The ARN pattern "
2533 "``arn:<partition>:ecr:*:<account>:repository/gco/*`` is "
2534 "the documented IAM way to express that scope: it "
2535 "covers exactly the repositories the function is "
2536 "allowed to touch and nothing else."
2537 ),
2538 "appliesTo": [
2539 f"Resource::arn:<AWS::Partition>:ecr:*:<AWS::AccountId>:"
2540 f"repository/{project_name}/*",
2541 ],
2542 },
2543 ],
2544 )