#!/usr/bin/env python3 from __future__ import annotations from typing import Optional from functools import partial from constructs import Construct from aws_cdk.cx_api import EnvironmentPlaceholders from aws_cdk import ( DefaultStackSynthesizer, BootstraplessSynthesizer, DefaultStackSynthesizerProps, Stack, Stage, CfnOutput, RemovalPolicy, App, IStackSynthesizer, Duration ) from aws_cdk.aws_ssm import StringParameter from aws_cdk.aws_s3 import Bucket from aws_cdk.aws_ecr import Repository from aws_cdk.aws_sqs import Queue from aws_cdk.aws_s3_assets import Asset ASSEMBLY_PARTITION = EnvironmentPlaceholders.CURRENT_PARTITION ASSEMBLY_ACCOUNT = EnvironmentPlaceholders.CURRENT_ACCOUNT ASSEMBLY_REGION = EnvironmentPlaceholders.CURRENT_REGION class ServiceStack(Stack): def __init__( self, scope: Construct, construct_id: str, *, stack_name: Optional[str] = None, synthesizer: Optional[IStackSynthesizer] = None, **kwargs ): service = isinstance(scope, Service) and scope or Service.of(scope) if not service: raise ValueError(f'{type(self).__name__} {repr(construct_id)} requires Service scope') # Ensure physical stack_name conforms to company governance service_name = service.service_name if not stack_name: stack_name = construct_id if construct_id != 'Default' else '' stack_name = f'{service_name}-{stack_name}' if stack_name else service_name # Default to our custom synthesizer to implicitly bootstrap the service synthesizer = synthesizer or service.synthesizer super().__init__( scope, construct_id, stack_name=stack_name, synthesizer=synthesizer, **kwargs ) class Synthesizer(DefaultStackSynthesizer): def __init__( self, service: Service, *, bootstrap_name: Optional[str] = None, **kwargs ): self.service = service self.bootstrap_name = bootstrap_name or 'cdk' self.bootstrap_prefix = f'{service.service_name}-{self.bootstrap_name}' self.stack = None # Ensure the service specific bootstrap resource names conform to company governance service_role_arn = service.service_role_arn version_name = f'/{self.bootstrap_prefix}/version' file_assets_name = f'{self.bootstrap_prefix}-assets-{ASSEMBLY_ACCOUNT}' file_assets_prefix = f'{ASSEMBLY_REGION}/' image_assets_name = f'{self.bootstrap_prefix}-image-assets-{ASSEMBLY_ACCOUNT}/{ASSEMBLY_REGION}' self.synthesizer_properties = DefaultStackSynthesizerProps( cloud_formation_execution_role=service_role_arn, deploy_role_arn=service_role_arn, lookup_role_arn=service_role_arn, bootstrap_stack_version_ssm_parameter=version_name, file_asset_publishing_role_arn=service_role_arn, file_assets_bucket_name=file_assets_name, bucket_prefix=file_assets_prefix, image_asset_publishing_role_arn=service_role_arn, image_assets_repository_name=image_assets_name, **kwargs ) super().__init__(**self.synthesizer_properties._values) def bind(self, stack): super().bind(stack) self.stack = stack # Implicitly boostrap the service bootstrap = Bootstrap.of(self.service) or Bootstrap(self.service, 'Bootstrap') stack.add_dependency(bootstrap) # Service specific bootstrap resources # Inherits from ServiceStack to also conform to company governance class Bootstrap(ServiceStack): @classmethod def of(cls, scope: Construct): service = scope if isinstance(scope, Service) else Service.of(scope) bootstraps = [child for child in service.node.children if isinstance(child, cls)] if service else [] if len(bootstraps) > 1: raise TypeError(f'Unable to get {cls.__name__} construct') elif len(bootstraps) == 1: return bootstraps[0] def __init__( self, scope: Service, construct_id: str, *, description: Optional[str] = None, stack_name: Optional[str] = None, termination_protection: Optional[bool] = True, **kwargs ): synthesizer = scope.synthesizer bootstrap_prefix = synthesizer.bootstrap_prefix synthesizer_properties = synthesizer.synthesizer_properties cloud_formation_execution_role_arn = synthesizer_properties.cloud_formation_execution_role deploy_role_arn = synthesizer_properties.deploy_role_arn version_name = synthesizer_properties.bootstrap_stack_version_ssm_parameter file_assets_name = synthesizer_properties.file_assets_bucket_name file_assets_prefix = synthesizer_properties.bucket_prefix image_assets_name = synthesizer_properties.image_assets_repository_name description = description or f'{scope.service_name} bootstrap stack' stack_name = stack_name or bootstrap_prefix # Bootstrapless since this is the bootstrap stack synthesizer = BootstraplessSynthesizer( cloud_formation_execution_role_arn=cloud_formation_execution_role_arn, deploy_role_arn=deploy_role_arn ) super().__init__( scope, construct_id, description=description, stack_name=stack_name, termination_protection=termination_protection, synthesizer=synthesizer, **kwargs ) replace_placeholders = partial( EnvironmentPlaceholders.replace, partition=self.partition, account_id=self.account, region=self.region ) # Initialize version parameter resource self.version = StringParameter( self, 'CdkBootstrapVersion', parameter_name=replace_placeholders(version_name), string_value='14', simple_name=False ) CfnOutput(self, 'BootstrapVersion', value=self.version.string_value) # Initialize assets bucket resource self.file_assets = Bucket( self, 'StagingBucket', bucket_name=replace_placeholders(file_assets_name) ) self.file_assets_prefix = replace_placeholders(file_assets_prefix) CfnOutput(self, 'BucketName', value=self.file_assets.bucket_name) CfnOutput(self, 'BucketDomainName', value=self.file_assets.bucket_domain_name) # Initalize image assets repository resource self.image_assets = Repository( self, 'ContainerAssetsRepository', repository_name=replace_placeholders(image_assets_name), image_scan_on_push=True, removal_policy=RemovalPolicy.DESTROY ) CfnOutput(self, 'ImageRepositoryName', value=self.image_assets.repository_name) # Encapsulate service stacks into a stage # This puts the stacks at the same construct tree root which helps centralize service wide # management, context, environment, and company governance class Service(Stage): def __init__( self, scope: Construct, service_name: str, *, service_role_arn: str, **kwargs ): if not service_role_arn: raise TypeError(f'{type(self).__name__} required keyword-only argument is missing or None: \'service_role_arn\'') super().__init__(scope, service_name, **kwargs) self.service_name = service_name self.service_role_arn = service_role_arn @property def synthesizer(self): # create new synthesizer instance every time # stacks bind to their synthesizer instance, so synthesizers can't be reused across stacks return Synthesizer(self) app = App() # Some external role does need to exist to grease the wheels, but is highly dependant on what your deployment # process looks like. Our company governance, for example, dictates much of this, which often conflicts # with CDK assumptions. This external role could be broad, covering all permissions for all CDK actions, or # more specific to get the process started in which you would define and deploy the more specific roles in # the Bootstrap stack above. service_role_arn = app.node.try_get_context('service_role_arn') or 'arn:aws:iam::123456789012:role/service-role' service = Service(app, service_name='Service', service_role_arn=service_role_arn) stack = ServiceStack(service, 'Stack') queue = Queue(stack, 'Queue', visibility_timeout=Duration.seconds(300)) # Some random resource for example # Force an asset if necessary for example # However, DefaultStackSynthesizer inserts a bootstrap version requirement in the manifest anyway # asset = Asset(stack, 'Asset', path=__file__) app.synth()