ECSBuilder

Bases: AbstractAWSResourceBuilder['ECSBuilder', EcsConfig]

AWS CDK builder for comprehensive ECS service infrastructure deployment.

Orchestrates the creation of a complete containerized application stack including ECS Fargate service, Application Load Balancer with SSL termination, CloudWatch logging, IAM roles with least-privilege permissions, Route53 DNS integration, and EFS file system mounting support for persistent storage.

Supports both Secrets Manager and Parameter Store for secure credential management, with cross-account resource sharing and organizational naming standards.

The builder creates: - ECS Cluster with Fargate capacity providers - Fargate service with configurable scaling and health checks - Application Load Balancer with HTTPS listeners - CloudWatch log groups for container logging - IAM roles for task execution and runtime permissions - Route53 A records for domain routing - EFS volumes and mount points for persistent storage - Integration with existing VPC and security groups

Source code in mare_aws_common_lib/builders/ecs_builder.py

class ECSBuilder(AbstractAWSResourceBuilder["ECSBuilder", EcsConfig]):
    """AWS CDK builder for comprehensive ECS service infrastructure deployment.

    Orchestrates the creation of a complete containerized application stack including
    ECS Fargate service, Application Load Balancer with SSL termination, CloudWatch
    logging, IAM roles with least-privilege permissions, Route53 DNS integration,
    and EFS file system mounting support for persistent storage.

    Supports both Secrets Manager and Parameter Store for secure credential management,
    with cross-account resource sharing and organizational naming standards.

    The builder creates:
    - ECS Cluster with Fargate capacity providers
    - Fargate service with configurable scaling and health checks
    - Application Load Balancer with HTTPS listeners
    - CloudWatch log groups for container logging
    - IAM roles for task execution and runtime permissions
    - Route53 A records for domain routing
    - EFS volumes and mount points for persistent storage
    - Integration with existing VPC and security groups
    """
    _resource_type: AWSResourceType = AWSResourceType.ECS

    def reset(self) -> None:
        """Reset builder state and clear configuration data.

        Initializes the builder for a new build cycle by clearing all
        configuration data and resetting internal state to default values.
        """
        super().reset()
        self._route53_config: Dict[str, Any] = {}
        self._app_config: Dict[str, Any] = {}
        self._imported_params_cache: Dict[str, CfnParameter] = {}

    def set_route53_config(self, config: Dict[str, Any]) -> 'ECSBuilder':
        """Configure Route53 DNS settings for the ECS service.

        Sets the hosted zone and domain configuration required for creating
        DNS records that route traffic to the Application Load Balancer.

        Args:
            config: Route53 configuration containing hosted_zone_id and domain_name

        Returns:
            Builder instance for method chaining
        """
        self._route53_config = config
        return self

    def set_app_parameters(self, config: Dict[str, Any]) -> 'ECSBuilder':
        """Configure application runtime parameters and secrets.

        Sets environment variables and secret references that will be injected
        into the ECS container at runtime, supporting both plain text environment
        variables and secure credential retrieval.

        Args:
            config: Application parameters configuration with environment and secrets

        Returns:
            Builder instance for method chaining
        """
        self._app_config = config
        return self

    def build(self, scope: Construct) -> elbv2.ApplicationLoadBalancer:
        """Build the complete ECS infrastructure stack.

        Creates and configures all AWS resources required for a production-ready
        containerized application including networking, compute, security, logging,
        DNS components and EFS storage with proper integration and dependencies.

        Args:
            scope: CDK construct scope for resource creation

        Returns:
            Application Load Balancer instance for the deployed service

        Raises:
            ValidationError: If configuration validation fails
            ValueError: If required configurations are missing
        """
        super().build()

        vpc: ec2.Vpc = self._application_helper.get_vpc("vpc-ecs", scope)
        cluster: ecs.Cluster = self._create_cluster(scope, vpc)
        task_execution_role: iam.Role = self._create_task_execution_role(scope)
        task_role: iam.Role = self._create_task_role(scope)

        log_group: logs.LogGroup = self._create_log_group(scope)

        task_definition: ecs.FargateTaskDefinition = self._create_task_definition(scope, task_execution_role, task_role)

        self._add_container_definition(scope, log_group, task_definition)

        # Create a load balancer
        alb_name: str = self._get_name_for_resource(
            f"{self._application_helper.get_target_env()}-{self._config.alb_name}",
            max_length=AWSResourceNameLength.ALB.value)

        alb_security_groups: List[ec2.SecurityGroup] = self._get_alb_security_groups(scope, vpc)

        alb: elbv2.ApplicationLoadBalancer = elbv2.ApplicationLoadBalancer(
            scope,
            self._get_cfn_logical_id(self._config.alb_name),
            load_balancer_name=alb_name,
            internet_facing=True,
            vpc=vpc,
            security_group=alb_security_groups[0]
        )

        if len(alb_security_groups) >= 2:
            for i in range(1, len(alb_security_groups)):
                alb.add_security_group(alb_security_groups[i])

        # Create an ECS service
        service_name: str = self._get_name_for_resource(
            f"{self._application_helper.get_target_env()}-{self._config.service_name}",
            max_length=AWSResourceNameLength.ECS.value)

        ecs_security_groups: List[ec2.SecurityGroup] = self._get_ecs_security_groups(scope, vpc)

        ecs_service: ecs.FargateService = ecs.FargateService(
            scope,
            self._get_cfn_logical_id(self._config.service_name),
            service_name=service_name,
            cluster=cluster,
            task_definition=task_definition,
            desired_count=self._config.desired_count,
            min_healthy_percent=100 if self._config.desired_count == 1 else 50,
            max_healthy_percent=200,
            assign_public_ip=False,
            security_groups=ecs_security_groups,
            vpc_subnets=ec2.SubnetSelection(
                subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS
            ),
            capacity_provider_strategies=[
                ecs.CapacityProviderStrategy(capacity_provider="FARGATE_SPOT", weight=2),
                ecs.CapacityProviderStrategy(capacity_provider="FARGATE", weight=1)
            ]
        )

        self._add_listener(scope,
                           alb,
                           ecs_service,
                           vpc,
                           name="app",
                           listening_port=443,
                           target_port=self._config.application_port,
                           ssl_certificate=self._get_ssl_certificate(scope))

        # create A record for alb
        self._create_alb_route53_a_record(scope, alb)

        # Store ALB arn in SSM in order to use it in the cloudfront builder
        self._application_helper.store_parameter(scope,
                                                 "ALB_arn",
                                                 alb.load_balancer_arn)

        return alb

    def _get_alb_security_groups(self, scope: Construct, vpc: ec2.IVpc) -> List[ec2.SecurityGroup]:
        """Retrieve security groups for Application Load Balancer configuration.

        Fetches the required security groups for ALB traffic management including
        CloudFront integration and private network access patterns.

        Args:
            scope: CDK construct scope for resource lookup
            vpc: VPC instance for security group context

        Returns:
            List of security groups for ALB attachment
        """
        ecs_security_groups: List[ec2.SecurityGroup] = []

        full_name, _ = ResourceNaming.get_name_for_resource(self._application_helper, AWSResourceType.SG, resource_name="own-cloudfront-alb-sg",
                                             max_length=AWSResourceNameLength.SEC_GROUP.value)
        ecs_security_groups.append(self._get_security_group_by_name(scope, vpc, full_name))

        if self._config.admin_active:
            full_name, _ = ResourceNaming.get_name_for_resource(self._application_helper, AWSResourceType.SG, resource_name="own-private-alb-sg",
                                                max_length=AWSResourceNameLength.SEC_GROUP.value)
            ecs_security_groups.append(self._get_security_group_by_name(scope, vpc, full_name))
        return ecs_security_groups

    def _get_ecs_security_groups(self, scope: Construct, vpc: ec2.IVpc) -> List[ec2.SecurityGroup] :
        """Retrieve security groups for ECS service configuration.

        Fetches the security groups required for ECS container networking
        and communication with other AWS services.

        Args:
            scope: CDK construct scope for resource lookup
            vpc: VPC instance for security group context

        Returns:
            List of security groups for ECS service attachment
        """
        ecs_security_groups: List[ec2.SecurityGroup] = []
        full_name, _ = ResourceNaming.get_name_for_resource(self._application_helper, AWSResourceType.SG, resource_name="own-ecs-sg",
                                             max_length=AWSResourceNameLength.SEC_GROUP.value)
        security_group: ec2.SecurityGroup = self._get_security_group_by_name(scope, vpc, full_name)
        ecs_security_groups.append(security_group)

        return ecs_security_groups

    def _get_security_group_by_name(self, scope: Construct, vpc: ec2.IVpc, sg_name: str) -> ec2.SecurityGroup:
        """Lookup and return an immutable security group reference.

        Retrieves a security group by name and returns an immutable reference
        to prevent accidental modification of existing security group rules.

        Args:
            scope: CDK construct scope for resource lookup
            vpc: VPC instance containing the security group
            sg_name: Name of the security group to lookup

        Returns:
            Immutable security group reference
        """
        mutable_security_group: ec2.SecurityGroup = ec2.SecurityGroup.from_lookup_by_name(scope, f"m-{sg_name}",
                                                                                          sg_name, vpc)
        immutable_security_group: ec2.SecurityGroup = ec2.SecurityGroup.from_security_group_id(scope, sg_name,
                                                                                               mutable_security_group.security_group_id,
                                                                                               mutable=False)
        return immutable_security_group

    def _add_listener(self, scope: Construct, alb: elbv2.ApplicationLoadBalancer, ecs_service: ecs.FargateService,
                      vpc: ec2.Vpc, name: str, listening_port: int, target_port: int,
                      ssl_certificate: acm.ICertificate = None):
        """Add HTTP/HTTPS listener and target group to the Application Load Balancer.

        Creates a listener configuration with health checks and forwards traffic
        to the ECS service through a target group. Supports both HTTP and HTTPS
        protocols with optional SSL certificate integration.

        Args:
            scope: CDK construct scope for resource creation
            alb: Application Load Balancer instance
            ecs_service: ECS Fargate service to receive traffic
            vpc: VPC instance for target group configuration
            name: Logical name for the listener and target group
            listening_port: Port on which the ALB listens for incoming traffic
            target_port: Port on which the container receives forwarded traffic
            ssl_certificate: Optional SSL certificate for HTTPS termination
        """
        target_group_name: str = self._get_name_for_resource(f"{self._application_helper.get_target_env()}-{name}-tg",
                                                             max_length=AWSResourceNameLength.ALB.value)
        health_check: elbv2.HealthCheck = None
        if self._config.health_check_path is not None:
            health_check = elbv2.HealthCheck(
                enabled=True,
                path=self._config.health_check_path, 
                healthy_http_codes="200",
                healthy_threshold_count=2,
                unhealthy_threshold_count=5,
                interval=Duration.seconds(30),
                timeout=Duration.seconds(10)
            )
        # else: 
        #     health_check = elbv2.HealthCheck(
        #         path="/", 
        #         healthy_http_codes="200,302"
        #     )

        target_group: elbv2.ApplicationTargetGroup = elbv2.ApplicationTargetGroup(
            scope, self._get_cfn_logical_id(f"{name}_target_group"),
            target_type=elbv2.TargetType.IP,
            protocol=elbv2.ApplicationProtocol.HTTP,
            port=target_port,
            vpc=vpc,
            health_check=health_check,
            target_group_name=target_group_name
        )

        # Add a listener to the load balancer
        certificates = None if ssl_certificate is None else [ssl_certificate]
        protocol = elbv2.ApplicationProtocol.HTTPS if listening_port in [443] else elbv2.ApplicationProtocol.HTTP
        app_listener = alb.add_listener(
            self._get_cfn_logical_id(f"{name}-listener"),
            port=listening_port,
            protocol=protocol,
            certificates=certificates,
            open=False,
            default_action=elbv2.ListenerAction.forward([target_group])
        )
        # Attach the ECS service to the target group
        target_group.add_target(ecs_service)

    def _get_ecr_repo(self, scope: Construct) -> ecr.Repository:
        """Retrieve ECR repository reference for container image deployment.

        Creates a reference to the existing ECR repository containing the
        application container images for deployment to ECS.

        Args:
            scope: CDK construct scope for resource reference

        Returns:
            ECR repository instance for container image access
        """
        ecr_repo_name = self._get_name_for_resource(self._config.ecr_repo_base_name, max_length=AWSResourceNameLength.ECR_REPOSITORY.value)
        account_id =self._application_helper.get_all_envs().get(self._config.ecr_repo_source_env).get("account_id")
        ecr_repo_arn = ArnUriHelper.build_ecr_repo_arn(self._application_helper, ecr_repo_name, account_id)
        ecr_repo = ecr.Repository.from_repository_attributes(
            scope, self._get_cfn_logical_id("app-ecr-repo"),
            repository_name=ecr_repo_name,
            repository_arn=ecr_repo_arn
        )
        return ecr_repo

    def _add_container_definition(self, scope: Construct, log_group: logs.LogGroup, 
                                  task_definition: ecs.FargateTaskDefinition) -> ecs.ContainerDefinition:
        """Add container definition to the ECS task with full configuration.

        Creates and configures the main application container including image
        reference, port mappings, logging, environment variables, environment files, secrets
        integration for runtime configuration management and EFS mount points for persistent storage.

        Args:
            scope: CDK construct scope for resource creation
            log_group: CloudWatch log group for container output
            task_definition: ECS task definition to receive the container

        Returns:
            Configured container definition instance
        """
        secrets = self._get_secrets(scope)

        build_env: Optional[Mapping[str, str]] = {
            "BUILD_TS": datetime.now().strftime("%Y%m%d%H%M%S")
        }

        health_check: ecs.HealthCheck = None
        if self._config.health_check_path is not None:
            cmd = self._config.health_check_cmd if self._config.health_check_cmd else "curl -f"
            health_check = ecs.HealthCheck(
                command=["CMD-SHELL", f"{cmd} http://localhost:{self._config.application_port}{self._config.health_check_path} || exit 1"],
                interval=Duration.seconds(30),
                timeout=Duration.seconds(10),
                retries=5,
                start_period=Duration.seconds(60)
            )

        image: ecs.ContainerImage = None
        if self._config.container_repo:
            image = ecs.ContainerImage.from_registry(self._config.container_repo)
        else:
            image = ecs.ContainerImage.from_ecr_repository(repository=self._get_ecr_repo(scope), tag=self._application_helper.get_app_version())

        # Prepare environment files if configured
        environment_files: Optional[List[ecs.EnvironmentFile]] = None
        if self._config.environment_files:
            environment_files = [
                ecs.EnvironmentFile.from_bucket(self._get_s3_bucket(scope, env_file.bucket_base_name), env_file.file_path)
                for env_file in self._config.environment_files
            ]

        # Add container to the task definition
        container_definition = task_definition.add_container(
            self._get_cfn_logical_id("app-ctnr"),
            container_name=f"{self._application_helper.get_project()}-app-ctnr",
            image=image,
            user=self._get_container_user(),
            logging=ecs.LogDriver.aws_logs(
                stream_prefix="YYYY-MM-DD",
                log_group=log_group
            ),
            secrets=secrets,
            environment=build_env,
            environment_files=environment_files,
            port_mappings=[
                ecs.PortMapping(
                    container_port=self._config.application_port,
                    protocol=ecs.Protocol.TCP
                )
            ],
            health_check=health_check
        )

        #container_definition.add_port_mappings(ecs.PortMapping(container_port=self._config.application_port))
        if self._config.admin_active:
            container_definition.add_port_mappings(
                ecs.PortMapping(container_port=self._config.admin_port))

        # Iterate over the elements of the "environment" list
        for key, value in self._config.parameters.environment.items():
            container_definition.add_environment(key, value)

        # Add EFS mount points if configured
        if hasattr(self._config, 'efs_config') and self._config.efs_config is not None:
            self._add_efs_mount_points(container_definition)

        return container_definition

    def _get_s3_bucket(self, scope: Construct, bucket_base_name: str) -> s3.IBucket:
        bucket_name = self._get_name_for_resource(f"{bucket_base_name}-{self._application_helper.get_from_env("account_id")}", 
                                                    max_length=AWSResourceNameLength.S3_BUCKET.value)

        return s3.Bucket.from_bucket_name(
            scope, 
            f"S3_ENV_FILES_BUCKET_{bucket_base_name.upper().replace("-", "_")}_{self._application_helper.get_project().upper()}",
            self._application_helper.get_parameter_value(scope, f"s3_bucket_{bucket_base_name}")
        )

    def _get_container_user(self) -> Optional[str]:
        """Determine the user the container should run as to match EFS access points.

        Returns the user:group specification that matches the EFS access point
        configuration to ensure proper file permissions and ownership.

        Returns:
            User specification in format "uid:gid" or None for default
        """
        # If EFS is configured, try to match the first access point's user
        if hasattr(self._config, 'efs_config') and self._config.efs_config is not None:
            # Check if there's an explicit container user configured
            if self._config.efs_config.container_user is not None:
                return self._config.efs_config.container_user

        # Default to None (use container's default user)
        return None

    def _get_secrets(self, scope: Construct) -> Dict[str, ecs.Secret]:
        """Build secrets configuration for container runtime injection.

        Processes the secrets configuration to create ECS-compatible secret
        references from both AWS Secrets Manager and Systems Manager Parameter
        Store sources for secure credential management.

        Args:
            scope: CDK construct scope for resource references

        Returns:
            Dictionary mapping environment variable names to ECS secret definitions
        """
        secrets = {}
        for secret_name, secret_cfg in self._config.parameters.secrets.items():
            if secret_cfg.value_from == SecretValueFrom.PARAM:
                ssm_param_name = self._application_helper._get_parameter_name_from_key(secret_cfg.name, with_leading_slash=True)
                ssm_param = ssm.StringParameter.from_string_parameter_name(
                    scope,
                    self._get_cfn_logical_id(f"{secret_name}-{secret_cfg.value_from}"),
                    ssm_param_name)
                secrets[secret_name] = ecs.Secret.from_ssm_parameter(ssm_param)
            elif secret_cfg.value_from == SecretValueFrom.SECRET:
                sm_secret: secretsmanager.ISecret = secretsmanager.Secret.from_secret_complete_arn(
                    scope,
                    self._get_cfn_logical_id(f"{secret_name}-{secret_cfg.value_from}"),
                    secret_cfg.arn)
                secrets[secret_name] = ecs.Secret.from_secrets_manager(sm_secret, secret_cfg.suffix)

        return secrets

    def _create_task_definition(self, scope: Construct, task_execution_role: iam.Role, task_role: iam.Role) -> ecs.FargateTaskDefinition:
        """Create ECS Fargate task definition with resource allocation and EFS volumes..

        Defines the task specification including CPU and memory allocation,
        IAM roles for execution and runtime permissions, task family naming
        for version management and deployment tracking, and EFS volume configuration.

        Args:
            scope: CDK construct scope for resource creation
            task_execution_role: IAM role for ECS task execution permissions
            task_role: IAM role for container runtime permissions

        Returns:
            Configured Fargate task definition
        """
        task_definition: ecs.FargateTaskDefinition = ecs.FargateTaskDefinition(
            scope, self._get_cfn_logical_id("task-def"),
            cpu=self._config.cpu,
            memory_limit_mib=self._config.memory_limit_mib,
            task_role=task_role,
            execution_role=task_execution_role,
            family=self._get_name_for_resource(f"{self._application_helper.get_target_env()}-be-taskdef",
                                               max_length=255)
        )

        # Add EFS volume if configured
        if hasattr(self._config, 'efs_config') and self._config.efs_config is not None:
            self._add_efs_volumes_to_task(scope, task_definition)

        return task_definition

    def _add_efs_volumes_to_task(self, scope: Construct, task_definition: ecs.FargateTaskDefinition) -> None:
        """Add EFS volumes to the task definition - supports multiple EFS systems.

        Creates EFS volume configurations for each configured EFS system,
        supporting both access point and root access patterns with transit encryption.

        Args:
            scope: CDK construct scope for resource references
            task_definition: Task definition to add volumes to
        """
        for volume_config in self._config.efs_config.volumes:
            # Get the EFS file system ID from the specified parameter
            efs_parameter_suffix = volume_config.efs_parameter_key_suffix.upper().replace("-", "_")
            efs_parameter_key = f"EFS_FILE_SYSTEM_ID_{efs_parameter_suffix}"
            efs_file_system_id = self._get_efs_parameter_value(scope, efs_parameter_key)

            if volume_config.ap_parameter_key_suffix is not None:
                # Use access point
                access_point_parameter_key = f"EFS_{efs_parameter_suffix}_ACCESS_POINT_ID_{volume_config.ap_parameter_key_suffix.upper().replace("-", "_")}"
                access_point_id = self._get_efs_parameter_value(scope, access_point_parameter_key)

                efs_volume_config = ecs.EfsVolumeConfiguration(
                    file_system_id=efs_file_system_id,
                    transit_encryption="ENABLED",
                    authorization_config=ecs.AuthorizationConfig(
                        access_point_id=access_point_id,
                        iam="ENABLED"
                    )
                )
            else:
                # Root access
                efs_volume_config = ecs.EfsVolumeConfiguration(
                    file_system_id=efs_file_system_id,
                    transit_encryption="ENABLED"
                )

            task_definition.add_volume(
                name=volume_config.volume_name,
                efs_volume_configuration=efs_volume_config
            )

    def _get_efs_parameter_value(self, scope: Construct, parameter_key: str) -> str:
        """Get EFS parameter value (file system ID or access point ID) from SSM.

        Retrieves parameter values from SSM with support for cross-account access
        in non-devops environments for proper resource sharing between stacks.

        Args:
            scope: CDK construct scope for parameter retrieval
            parameter_key: The SSM parameter key to retrieve

        Returns:
            Parameter value (EFS ID or access point ID)
        """
        if self._application_helper.get_target_env().upper() != Environment.DEVOPS.name:
            # For non-devops environments, use cross-account parameter access
            if parameter_key in self._imported_params_cache:
                return Token.as_string(self._imported_params_cache[parameter_key].value)

            consumer = CrossAccountResourceConsumer(self._application_helper)
            param = consumer.import_ssm_parameter(
                scope,
                parameter_key=parameter_key,
                producer_account_id=self._application_helper.get_from_env("account_id"),
                secret_env_suffix=self._application_helper.get_target_env(),
                logical_id_prefix=f"cross-account-{parameter_key.lower().replace('_', '-')}"
            )
            self._imported_params_cache[parameter_key] = param

            return Token.as_string(param.value)
        else:
            # For devops environment, direct parameter access
            return self._application_helper.get_parameter_value(scope, parameter_key)

    def _add_efs_mount_points(self, container_definition: ecs.ContainerDefinition) -> None:
        """Add EFS mount points to the container - supports multiple EFS systems.

        Configures container mount points for all configured EFS volumes,
        allowing containers to access persistent storage at specified paths.

        Args:
            container_definition: Container to add mount points to
        """
        mount_points = self._config.efs_config.mount_points

        for mount_point in mount_points:
            container_definition.add_mount_points(
                ecs.MountPoint(
                    source_volume=mount_point.volume_name,  # Must match volume name from task definition
                    container_path=mount_point.container_path,
                    read_only=mount_point.read_only
                )
            )

    def _create_log_group(self, scope: Construct) -> logs.LogGroup:
        """Create CloudWatch log group for container output collection.

        Sets up centralized logging for the ECS service with appropriate
        retention policies and naming conventions for log management and
        operational visibility.

        Args:
            scope: CDK construct scope for resource creation

        Returns:
            CloudWatch log group for ECS container logging
        """
        log_group: logs.LogGroup = logs.LogGroup(
            scope, self._get_cfn_logical_id("ecs-loggroup"),
            log_group_name=f"{self._application_helper.get_domain()}/{self._application_helper.get_project()}/ecs".lower(),
            retention=logs.RetentionDays.ONE_MONTH,
            removal_policy=RemovalPolicy.DESTROY
        )
        return log_group

    def _create_cluster(self, scope: Construct, vpc: ec2.Vpc) -> ecs.Cluster:
        """Create ECS cluster with Fargate capacity provider configuration.

        Establishes the ECS cluster infrastructure with Fargate capacity providers
        enabled for serverless container execution and automatic scaling capabilities.

        Args:
            scope: CDK construct scope for resource creation
            vpc: VPC instance for cluster networking

        Returns:
            Configured ECS cluster with Fargate support
        """
        cluster_name = self._get_name_for_resource(
            f"{self._application_helper.get_target_env()}-{self._config.cluster_name}",
            max_length=AWSResourceNameLength.ECS.value)
        cluster: ecs.Cluster = ecs.Cluster(
            scope,
            self._get_cfn_logical_id("ecs-cluster"),
            cluster_name=cluster_name,
            vpc=vpc,
            enable_fargate_capacity_providers=True
        )
        cluster.enable_fargate_capacity_providers()
        return cluster

    def _create_task_role(self, scope: Construct) -> iam.Role:
        """Create IAM role for ECS task runtime permissions.

        Creates an IAM role with comprehensive permissions for application runtime
        including RDS access, S3 operations, CloudWatch logging, Secrets Manager,
        Parameter Store integration and EFS access following least-privilege principles.

        Args:
            scope: CDK construct scope for resource creation

        Returns:
            IAM role with task runtime permissions
        """
        task_role_name: str = self._get_name_for_resource("task_role", max_length=AWSResourceNameLength.ROLE.value)

        # Create a task role
        task_role = iam.Role(
            scope, self._get_cfn_logical_id("task_role"),
            assumed_by=iam.ServicePrincipal("ecs-tasks.amazonaws.com"),
            description="Role for ECS task with permissions for RDS, S3, CloudWatch Logs, Secrets Manager, SSM Parameter Store and EFS",
            role_name=task_role_name
        )

        # Attach permissions for RDS
        task_role.add_to_policy(
            iam.PolicyStatement(
                actions=[
                    "rds:*"
                ],
                resources=["*"]
            )
        )

        if self._config.use_iam_authentication:
            rds_db_resource_id: str
            if self._application_helper.get_target_env().upper() != Environment.DEVOPS.name:
                consumer = CrossAccountResourceConsumer(self._application_helper)
                rds_resource_id_param = consumer.import_ssm_parameter(
                    scope,
                    parameter_key="RDS_DB_RESOURCE_ID",
                    producer_account_id=self._application_helper.get_from_env("account_id"),
                    secret_env_suffix=self._application_helper.get_target_env(),
                    logical_id_prefix="cross-account-rds-rid"
                )
                rds_db_resource_id = Token.as_string(rds_resource_id_param.value)
            else: 
                rds_db_resource_id = self._application_helper.get_parameter_value(scope, "RDS_DB_RESOURCE_ID")

            task_role.add_to_policy(
                iam.PolicyStatement(
                    actions=[
                        "rds-db:connect"
                    ],
                    resources=[
                        f"arn:aws:rds-db:{self._application_helper.get_from_common("region")}:{self._application_helper.get_from_env("account_id")}:dbuser:{rds_db_resource_id}/*"
                        # f"arn:aws:rds-db:{self._application_helper.get_from_common("region")}:{self._application_helper.get_from_env("account_id")}:dbuser:{rds_db_resource_id}/${{resolve:secretsmanager:{rds_app_user_secret.secret_arn}:SecretString:db_app_username}}"
                    ]
                )
            )

        # Attach permissions for S3
        task_role.add_to_policy(
            iam.PolicyStatement(
                actions=[
                    "s3:GetObject",
                    "s3:ListBucket",
                    "s3:PutObject"  # Add other S3 actions as needed
                ],
                resources=["*"]
            )
        )

        # Attach permissions for CloudWatch Logs
        task_role.add_to_policy(
            iam.PolicyStatement(
                actions=[
                    "logs:CreateLogGroup",
                    "logs:CreateLogStream",
                    "logs:PutLogEvents",
                    "logs:DescribeLogStreams"
                ],
                resources=["*"]
            )
        )

        # Attach permissions for Secrets Manager
        task_role.add_to_policy(
            iam.PolicyStatement(
                actions=[
                    "kms:*",
                    "secretsmanager:GetSecretValue",
                    "secretsmanager:DescribeSecret"  # Add other Secrets Manager actions as needed
                ],
                resources=["*"]
            )
        )

        # Attach permissions for SSM Parameter Store
        task_role.add_to_policy(
            iam.PolicyStatement(
                actions=[
                    "ssm:GetParameters",
                    "ssm:GetParameter"
                ],
                resources=["*"]
            )
        )

        # Add EFS permissions if EFS is configured
        if hasattr(self._config, 'efs_config') and self._config.efs_config is not None:
            task_role.add_to_policy(
                iam.PolicyStatement(
                    actions=[
                        "elasticfilesystem:ClientRootAccess",
                        "elasticfilesystem:ClientWrite", 
                        "elasticfilesystem:ClientMount"
                    ],
                    resources=["*"]
                )
            )

        return task_role

    def _create_task_execution_role(self, scope: Construct) -> iam.Role:
        """Create IAM role for ECS task execution permissions.

        Creates an IAM role with permissions required for ECS service to execute
        tasks including ECR image pulling, CloudWatch logging, and secrets retrieval
        during container startup and initialization.

        Args:
            scope: CDK construct scope for resource creation

        Returns:
            IAM role with task execution permissions
        """
        task_execution_role_name: str = self._get_name_for_resource("task_exec_role",
                                                                    max_length=AWSResourceNameLength.ROLE.value)
        task_execution_role: iam.Role = iam.Role(
            scope,
            self._get_cfn_logical_id("task_exec_role"),
            assumed_by=iam.ServicePrincipal("ecs-tasks.amazonaws.com"),
            description="Role for ECS task execution",
            role_name=task_execution_role_name
        )
        # Attach the managed policy for ECS task execution
        task_execution_role.add_managed_policy(
            iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AmazonECSTaskExecutionRolePolicy")
        )
        # Optionally, add additional permissions if needed
        task_execution_role.add_to_policy(
            iam.PolicyStatement(
                actions=[
                    "logs:CreateLogGroup",
                    "logs:CreateLogStream",
                    "logs:PutLogEvents",
                    "logs:DescribeLogStreams",
                    "secretsmanager:GetSecretValue",
                    "secretsmanager:DescribeSecret",
                    "kms:Decrypt",
                    "kms:GenerateDataKey",
                    "s3:GetObject",
                    "s3:GetBucketLocation",
                    "ssm:GetParameters",
                    "ssm:GetParameter",
                    "elasticfilesystem:ClientMount"
                ],
                resources=["*"]
            )
        )
        return task_execution_role

    def _get_efs_file_system_id(self, scope: Construct) -> str:
        """Retrieve EFS file system ID from SSM Parameter Store with cross-account support.

        Fetches the EFS file system identifier required for volume configuration in ECS
        task definitions. The method handles both local parameter access in DevOps
        environments and cross-account parameter sharing for production deployments.

        Args:
            scope: CDK construct scope for parameter reference creation

        Returns:
            EFS file system ID string in format 'fs-xxxxxxxxx' for volume configuration

        Raises:
            ParameterNotFound: If the EFS_FILE_SYSTEM_ID parameter doesn't exist
            PermissionError: If cross-account access permissions are insufficient
        """
        if self._application_helper.get_target_env().upper() != Environment.DEVOPS.name:
            consumer = CrossAccountResourceConsumer(self._application_helper)
            efs_param = consumer.import_ssm_parameter(
                scope,
                parameter_key="EFS_FILE_SYSTEM_ID",
                producer_account_id=self._application_helper.get_from_env("account_id"),
                secret_env_suffix=self._application_helper.get_target_env(),
                logical_id_prefix="cross-account-efs-fid"
            )
            return Token.as_string(efs_param.value)
        else:
            return self._application_helper.get_parameter_value(scope, "EFS_FILE_SYSTEM_ID")

    def _get_efs_access_point_id(self, scope: Construct, access_point_index: int) -> str:
        """Retrieve EFS access point ID from SSM Parameter Store with cross-account support.

        Fetches the EFS access point identifier for fine-grained access control in ECS
        task definitions. Access points enable path-based access restrictions, user/group
        enforcement, and creation permissions for enhanced security in multi-tenant scenarios.

        The method constructs parameter keys dynamically based on the access point index,
        supporting multiple access points per EFS file system. Parameter retrieval follows
        the same cross-account patterns as file system ID access.

        Args:
            scope: CDK construct scope for parameter reference creation
            access_point_index: Zero-based index of the access point (0, 1, 2, ...)

        Returns:
            EFS access point ID string in format 'fsap-xxxxxxxxx' for authorization config

        Raises:
            ParameterNotFound: If the access point parameter doesn't exist for the given index
            PermissionError: If cross-account access permissions are insufficient
        """
        parameter_key = f"EFS_ACCESS_POINT_{access_point_index}_ID"

        if self._application_helper.get_target_env().upper() != Environment.DEVOPS.name:
            consumer = CrossAccountResourceConsumer(self._application_helper)
            ap_param = consumer.import_ssm_parameter(
                scope,
                parameter_key=parameter_key,
                producer_account_id=self._application_helper.get_from_env("account_id"),
                secret_env_suffix=self._application_helper.get_target_env(),
                logical_id_prefix=f"cross-account-efs-ap-{access_point_index}"
            )
            return Token.as_string(ap_param.value)
        else:
            return self._application_helper.get_parameter_value(scope, parameter_key)


    def _get_route53_hosted_zone(self, scope: Construct) -> route53.HostedZone:
        """Retrieve Route53 hosted zone reference for DNS record management.

        Creates a reference to the existing Route53 hosted zone for creating
        DNS records that route traffic to the Application Load Balancer.

        Args:
            scope: CDK construct scope for resource reference

        Returns:
            Route53 hosted zone reference
        """
        return route53.HostedZone.from_hosted_zone_attributes(
            scope,
            self._get_cfn_logical_id("ecs-hosted-zone"),
            hosted_zone_id=self._config.route53.hosted_zone_id,
            zone_name=self._config.route53.domain_name
        )

    def _create_alb_route53_a_record(self, scope: Construct, target: elbv2.ApplicationLoadBalancer) -> None:
        """Create Route53 A record pointing to the Application Load Balancer.

        Creates an alias A record that routes DNS queries for the configured
        domain prefix to the Application Load Balancer for traffic distribution.

        Args:
            scope: CDK construct scope for resource creation
            target: Load balancer target for the DNS record
        """
        route53.ARecord(
            scope,
            "alb_alias_record",
            zone=self._get_route53_hosted_zone(scope),
            record_name=f"{self._config.alb_domain_prefix}.{self._config.route53.domain_name}",
            target=route53.RecordTarget.from_alias(targets.LoadBalancerTarget(target))
        )

    def _get_ssl_certificate(self, scope: Construct) -> acm.ICertificate:
        """Retrieve SSL certificate reference for HTTPS termination.

        Creates a reference to the existing ACM certificate for SSL/TLS
        termination at the Application Load Balancer level.

        Args:
            scope: CDK construct scope for resource reference

        Returns:
            ACM certificate reference for HTTPS configuration
        """
        ssl_certificate = acm.Certificate.from_certificate_arn(
            scope,
            self._get_cfn_logical_id(f"{self._application_helper.get_target_env()}-ssl-certificate"),
            certificate_arn=self._get_certificate_arn(self._config.ssl_certificate_id, "eu-west-1")
        )
        return ssl_certificate

    def _get_certificate_arn(self, certificate_id: str, region: str) -> str:
        """Build ACM certificate ARN from certificate ID and region.

        Constructs the full ARN for an ACM certificate using the certificate ID
        and specified region, accounting for cross-region certificate usage patterns.

        Args:
            certificate_id: ACM certificate identifier
            region: AWS region where the certificate is located

        Returns:
            Complete ACM certificate ARN
        """
        # Because the default region for this profile is eu-west-1, but, the certificate is in us-east-1, hence the use of the mocked region instead of {Aws.REGION}
        return f"arn:aws:acm:{region}:{Aws.ACCOUNT_ID}:certificate/{certificate_id}"

    def _set_config(self) -> None:
        """Validate and set the ECS configuration from builder inputs.

        Combines the builder configuration with Route53 and application parameters
        to create a validated EcsConfig instance, ensuring all required fields
        and validation rules are satisfied.

        Raises:
            ValidationError: If the combined configuration fails validation
        """
        try:
            self._config = EcsConfig(**{
                **self._builder_config,
                "route53": self._route53_config,
                "parameters": self._app_config
            })
        except ValidationError as e:
            self._log_validation_error(e, EcsConfig)
            raise

    def _control_consistency(self) -> None:
        """Validate builder state and configuration consistency.

        Performs pre-build validation to ensure all required configurations
        are present and the builder state is consistent for successful
        infrastructure deployment.

        Raises:
            ValueError: If Route53 configuration is missing
            ValidationError: If configuration validation fails
        """
        super()._control_consistency()

        if not self._route53_config:
            raise ValueError("Route 53 configuration must be set before building")

        self._set_config()

Attributes

_resource_type = AWSResourceType.ECS class-attribute instance-attribute

Functions

_add_container_definition(scope, log_group, task_definition)

Add container definition to the ECS task with full configuration.

Creates and configures the main application container including image reference, port mappings, logging, environment variables, environment files, secrets integration for runtime configuration management and EFS mount points for persistent storage.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource creation	required
log_group	LogGroup	CloudWatch log group for container output	required
task_definition	FargateTaskDefinition	ECS task definition to receive the container	required

Returns:

Type	Description
ContainerDefinition	Configured container definition instance

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _add_container_definition(self, scope: Construct, log_group: logs.LogGroup, 
                              task_definition: ecs.FargateTaskDefinition) -> ecs.ContainerDefinition:
    """Add container definition to the ECS task with full configuration.

    Creates and configures the main application container including image
    reference, port mappings, logging, environment variables, environment files, secrets
    integration for runtime configuration management and EFS mount points for persistent storage.

    Args:
        scope: CDK construct scope for resource creation
        log_group: CloudWatch log group for container output
        task_definition: ECS task definition to receive the container

    Returns:
        Configured container definition instance
    """
    secrets = self._get_secrets(scope)

    build_env: Optional[Mapping[str, str]] = {
        "BUILD_TS": datetime.now().strftime("%Y%m%d%H%M%S")
    }

    health_check: ecs.HealthCheck = None
    if self._config.health_check_path is not None:
        cmd = self._config.health_check_cmd if self._config.health_check_cmd else "curl -f"
        health_check = ecs.HealthCheck(
            command=["CMD-SHELL", f"{cmd} http://localhost:{self._config.application_port}{self._config.health_check_path} || exit 1"],
            interval=Duration.seconds(30),
            timeout=Duration.seconds(10),
            retries=5,
            start_period=Duration.seconds(60)
        )

    image: ecs.ContainerImage = None
    if self._config.container_repo:
        image = ecs.ContainerImage.from_registry(self._config.container_repo)
    else:
        image = ecs.ContainerImage.from_ecr_repository(repository=self._get_ecr_repo(scope), tag=self._application_helper.get_app_version())

    # Prepare environment files if configured
    environment_files: Optional[List[ecs.EnvironmentFile]] = None
    if self._config.environment_files:
        environment_files = [
            ecs.EnvironmentFile.from_bucket(self._get_s3_bucket(scope, env_file.bucket_base_name), env_file.file_path)
            for env_file in self._config.environment_files
        ]

    # Add container to the task definition
    container_definition = task_definition.add_container(
        self._get_cfn_logical_id("app-ctnr"),
        container_name=f"{self._application_helper.get_project()}-app-ctnr",
        image=image,
        user=self._get_container_user(),
        logging=ecs.LogDriver.aws_logs(
            stream_prefix="YYYY-MM-DD",
            log_group=log_group
        ),
        secrets=secrets,
        environment=build_env,
        environment_files=environment_files,
        port_mappings=[
            ecs.PortMapping(
                container_port=self._config.application_port,
                protocol=ecs.Protocol.TCP
            )
        ],
        health_check=health_check
    )

    #container_definition.add_port_mappings(ecs.PortMapping(container_port=self._config.application_port))
    if self._config.admin_active:
        container_definition.add_port_mappings(
            ecs.PortMapping(container_port=self._config.admin_port))

    # Iterate over the elements of the "environment" list
    for key, value in self._config.parameters.environment.items():
        container_definition.add_environment(key, value)

    # Add EFS mount points if configured
    if hasattr(self._config, 'efs_config') and self._config.efs_config is not None:
        self._add_efs_mount_points(container_definition)

    return container_definition

_add_efs_mount_points(container_definition)

Add EFS mount points to the container - supports multiple EFS systems.

Configures container mount points for all configured EFS volumes, allowing containers to access persistent storage at specified paths.

Parameters:

Name	Type	Description	Default
container_definition	ContainerDefinition	Container to add mount points to	required

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _add_efs_mount_points(self, container_definition: ecs.ContainerDefinition) -> None:
    """Add EFS mount points to the container - supports multiple EFS systems.

    Configures container mount points for all configured EFS volumes,
    allowing containers to access persistent storage at specified paths.

    Args:
        container_definition: Container to add mount points to
    """
    mount_points = self._config.efs_config.mount_points

    for mount_point in mount_points:
        container_definition.add_mount_points(
            ecs.MountPoint(
                source_volume=mount_point.volume_name,  # Must match volume name from task definition
                container_path=mount_point.container_path,
                read_only=mount_point.read_only
            )
        )

_add_efs_volumes_to_task(scope, task_definition)

Add EFS volumes to the task definition - supports multiple EFS systems.

Creates EFS volume configurations for each configured EFS system, supporting both access point and root access patterns with transit encryption.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource references	required
task_definition	FargateTaskDefinition	Task definition to add volumes to	required

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _add_efs_volumes_to_task(self, scope: Construct, task_definition: ecs.FargateTaskDefinition) -> None:
    """Add EFS volumes to the task definition - supports multiple EFS systems.

    Creates EFS volume configurations for each configured EFS system,
    supporting both access point and root access patterns with transit encryption.

    Args:
        scope: CDK construct scope for resource references
        task_definition: Task definition to add volumes to
    """
    for volume_config in self._config.efs_config.volumes:
        # Get the EFS file system ID from the specified parameter
        efs_parameter_suffix = volume_config.efs_parameter_key_suffix.upper().replace("-", "_")
        efs_parameter_key = f"EFS_FILE_SYSTEM_ID_{efs_parameter_suffix}"
        efs_file_system_id = self._get_efs_parameter_value(scope, efs_parameter_key)

        if volume_config.ap_parameter_key_suffix is not None:
            # Use access point
            access_point_parameter_key = f"EFS_{efs_parameter_suffix}_ACCESS_POINT_ID_{volume_config.ap_parameter_key_suffix.upper().replace("-", "_")}"
            access_point_id = self._get_efs_parameter_value(scope, access_point_parameter_key)

            efs_volume_config = ecs.EfsVolumeConfiguration(
                file_system_id=efs_file_system_id,
                transit_encryption="ENABLED",
                authorization_config=ecs.AuthorizationConfig(
                    access_point_id=access_point_id,
                    iam="ENABLED"
                )
            )
        else:
            # Root access
            efs_volume_config = ecs.EfsVolumeConfiguration(
                file_system_id=efs_file_system_id,
                transit_encryption="ENABLED"
            )

        task_definition.add_volume(
            name=volume_config.volume_name,
            efs_volume_configuration=efs_volume_config
        )

_add_listener(scope, alb, ecs_service, vpc, name, listening_port, target_port, ssl_certificate=None)

Add HTTP/HTTPS listener and target group to the Application Load Balancer.

Creates a listener configuration with health checks and forwards traffic to the ECS service through a target group. Supports both HTTP and HTTPS protocols with optional SSL certificate integration.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource creation	required
alb	ApplicationLoadBalancer	Application Load Balancer instance	required
ecs_service	FargateService	ECS Fargate service to receive traffic	required
vpc	Vpc	VPC instance for target group configuration	required
name	str	Logical name for the listener and target group	required
listening_port	int	Port on which the ALB listens for incoming traffic	required
target_port	int	Port on which the container receives forwarded traffic	required
ssl_certificate	ICertificate	Optional SSL certificate for HTTPS termination	None

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _add_listener(self, scope: Construct, alb: elbv2.ApplicationLoadBalancer, ecs_service: ecs.FargateService,
                  vpc: ec2.Vpc, name: str, listening_port: int, target_port: int,
                  ssl_certificate: acm.ICertificate = None):
    """Add HTTP/HTTPS listener and target group to the Application Load Balancer.

    Creates a listener configuration with health checks and forwards traffic
    to the ECS service through a target group. Supports both HTTP and HTTPS
    protocols with optional SSL certificate integration.

    Args:
        scope: CDK construct scope for resource creation
        alb: Application Load Balancer instance
        ecs_service: ECS Fargate service to receive traffic
        vpc: VPC instance for target group configuration
        name: Logical name for the listener and target group
        listening_port: Port on which the ALB listens for incoming traffic
        target_port: Port on which the container receives forwarded traffic
        ssl_certificate: Optional SSL certificate for HTTPS termination
    """
    target_group_name: str = self._get_name_for_resource(f"{self._application_helper.get_target_env()}-{name}-tg",
                                                         max_length=AWSResourceNameLength.ALB.value)
    health_check: elbv2.HealthCheck = None
    if self._config.health_check_path is not None:
        health_check = elbv2.HealthCheck(
            enabled=True,
            path=self._config.health_check_path, 
            healthy_http_codes="200",
            healthy_threshold_count=2,
            unhealthy_threshold_count=5,
            interval=Duration.seconds(30),
            timeout=Duration.seconds(10)
        )
    # else: 
    #     health_check = elbv2.HealthCheck(
    #         path="/", 
    #         healthy_http_codes="200,302"
    #     )

    target_group: elbv2.ApplicationTargetGroup = elbv2.ApplicationTargetGroup(
        scope, self._get_cfn_logical_id(f"{name}_target_group"),
        target_type=elbv2.TargetType.IP,
        protocol=elbv2.ApplicationProtocol.HTTP,
        port=target_port,
        vpc=vpc,
        health_check=health_check,
        target_group_name=target_group_name
    )

    # Add a listener to the load balancer
    certificates = None if ssl_certificate is None else [ssl_certificate]
    protocol = elbv2.ApplicationProtocol.HTTPS if listening_port in [443] else elbv2.ApplicationProtocol.HTTP
    app_listener = alb.add_listener(
        self._get_cfn_logical_id(f"{name}-listener"),
        port=listening_port,
        protocol=protocol,
        certificates=certificates,
        open=False,
        default_action=elbv2.ListenerAction.forward([target_group])
    )
    # Attach the ECS service to the target group
    target_group.add_target(ecs_service)

_control_consistency()

Validate builder state and configuration consistency.

Performs pre-build validation to ensure all required configurations are present and the builder state is consistent for successful infrastructure deployment.

Raises:

Type	Description
ValueError	If Route53 configuration is missing
ValidationError	If configuration validation fails

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _control_consistency(self) -> None:
    """Validate builder state and configuration consistency.

    Performs pre-build validation to ensure all required configurations
    are present and the builder state is consistent for successful
    infrastructure deployment.

    Raises:
        ValueError: If Route53 configuration is missing
        ValidationError: If configuration validation fails
    """
    super()._control_consistency()

    if not self._route53_config:
        raise ValueError("Route 53 configuration must be set before building")

    self._set_config()

_create_alb_route53_a_record(scope, target)

Create Route53 A record pointing to the Application Load Balancer.

Creates an alias A record that routes DNS queries for the configured domain prefix to the Application Load Balancer for traffic distribution.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource creation	required
target	ApplicationLoadBalancer	Load balancer target for the DNS record	required

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _create_alb_route53_a_record(self, scope: Construct, target: elbv2.ApplicationLoadBalancer) -> None:
    """Create Route53 A record pointing to the Application Load Balancer.

    Creates an alias A record that routes DNS queries for the configured
    domain prefix to the Application Load Balancer for traffic distribution.

    Args:
        scope: CDK construct scope for resource creation
        target: Load balancer target for the DNS record
    """
    route53.ARecord(
        scope,
        "alb_alias_record",
        zone=self._get_route53_hosted_zone(scope),
        record_name=f"{self._config.alb_domain_prefix}.{self._config.route53.domain_name}",
        target=route53.RecordTarget.from_alias(targets.LoadBalancerTarget(target))
    )

_create_cluster(scope, vpc)

Create ECS cluster with Fargate capacity provider configuration.

Establishes the ECS cluster infrastructure with Fargate capacity providers enabled for serverless container execution and automatic scaling capabilities.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource creation	required
vpc	Vpc	VPC instance for cluster networking	required

Returns:

Type	Description
Cluster	Configured ECS cluster with Fargate support

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _create_cluster(self, scope: Construct, vpc: ec2.Vpc) -> ecs.Cluster:
    """Create ECS cluster with Fargate capacity provider configuration.

    Establishes the ECS cluster infrastructure with Fargate capacity providers
    enabled for serverless container execution and automatic scaling capabilities.

    Args:
        scope: CDK construct scope for resource creation
        vpc: VPC instance for cluster networking

    Returns:
        Configured ECS cluster with Fargate support
    """
    cluster_name = self._get_name_for_resource(
        f"{self._application_helper.get_target_env()}-{self._config.cluster_name}",
        max_length=AWSResourceNameLength.ECS.value)
    cluster: ecs.Cluster = ecs.Cluster(
        scope,
        self._get_cfn_logical_id("ecs-cluster"),
        cluster_name=cluster_name,
        vpc=vpc,
        enable_fargate_capacity_providers=True
    )
    cluster.enable_fargate_capacity_providers()
    return cluster

_create_log_group(scope)

Create CloudWatch log group for container output collection.

Sets up centralized logging for the ECS service with appropriate retention policies and naming conventions for log management and operational visibility.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource creation	required

Returns:

Type	Description
LogGroup	CloudWatch log group for ECS container logging

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _create_log_group(self, scope: Construct) -> logs.LogGroup:
    """Create CloudWatch log group for container output collection.

    Sets up centralized logging for the ECS service with appropriate
    retention policies and naming conventions for log management and
    operational visibility.

    Args:
        scope: CDK construct scope for resource creation

    Returns:
        CloudWatch log group for ECS container logging
    """
    log_group: logs.LogGroup = logs.LogGroup(
        scope, self._get_cfn_logical_id("ecs-loggroup"),
        log_group_name=f"{self._application_helper.get_domain()}/{self._application_helper.get_project()}/ecs".lower(),
        retention=logs.RetentionDays.ONE_MONTH,
        removal_policy=RemovalPolicy.DESTROY
    )
    return log_group

_create_task_definition(scope, task_execution_role, task_role)

Create ECS Fargate task definition with resource allocation and EFS volumes..

Defines the task specification including CPU and memory allocation, IAM roles for execution and runtime permissions, task family naming for version management and deployment tracking, and EFS volume configuration.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource creation	required
task_execution_role	Role	IAM role for ECS task execution permissions	required
task_role	Role	IAM role for container runtime permissions	required

Returns:

Type	Description
FargateTaskDefinition	Configured Fargate task definition

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _create_task_definition(self, scope: Construct, task_execution_role: iam.Role, task_role: iam.Role) -> ecs.FargateTaskDefinition:
    """Create ECS Fargate task definition with resource allocation and EFS volumes..

    Defines the task specification including CPU and memory allocation,
    IAM roles for execution and runtime permissions, task family naming
    for version management and deployment tracking, and EFS volume configuration.

    Args:
        scope: CDK construct scope for resource creation
        task_execution_role: IAM role for ECS task execution permissions
        task_role: IAM role for container runtime permissions

    Returns:
        Configured Fargate task definition
    """
    task_definition: ecs.FargateTaskDefinition = ecs.FargateTaskDefinition(
        scope, self._get_cfn_logical_id("task-def"),
        cpu=self._config.cpu,
        memory_limit_mib=self._config.memory_limit_mib,
        task_role=task_role,
        execution_role=task_execution_role,
        family=self._get_name_for_resource(f"{self._application_helper.get_target_env()}-be-taskdef",
                                           max_length=255)
    )

    # Add EFS volume if configured
    if hasattr(self._config, 'efs_config') and self._config.efs_config is not None:
        self._add_efs_volumes_to_task(scope, task_definition)

    return task_definition

_create_task_execution_role(scope)

Create IAM role for ECS task execution permissions.

Creates an IAM role with permissions required for ECS service to execute tasks including ECR image pulling, CloudWatch logging, and secrets retrieval during container startup and initialization.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource creation	required

Returns:

Type	Description
Role	IAM role with task execution permissions

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _create_task_execution_role(self, scope: Construct) -> iam.Role:
    """Create IAM role for ECS task execution permissions.

    Creates an IAM role with permissions required for ECS service to execute
    tasks including ECR image pulling, CloudWatch logging, and secrets retrieval
    during container startup and initialization.

    Args:
        scope: CDK construct scope for resource creation

    Returns:
        IAM role with task execution permissions
    """
    task_execution_role_name: str = self._get_name_for_resource("task_exec_role",
                                                                max_length=AWSResourceNameLength.ROLE.value)
    task_execution_role: iam.Role = iam.Role(
        scope,
        self._get_cfn_logical_id("task_exec_role"),
        assumed_by=iam.ServicePrincipal("ecs-tasks.amazonaws.com"),
        description="Role for ECS task execution",
        role_name=task_execution_role_name
    )
    # Attach the managed policy for ECS task execution
    task_execution_role.add_managed_policy(
        iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AmazonECSTaskExecutionRolePolicy")
    )
    # Optionally, add additional permissions if needed
    task_execution_role.add_to_policy(
        iam.PolicyStatement(
            actions=[
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents",
                "logs:DescribeLogStreams",
                "secretsmanager:GetSecretValue",
                "secretsmanager:DescribeSecret",
                "kms:Decrypt",
                "kms:GenerateDataKey",
                "s3:GetObject",
                "s3:GetBucketLocation",
                "ssm:GetParameters",
                "ssm:GetParameter",
                "elasticfilesystem:ClientMount"
            ],
            resources=["*"]
        )
    )
    return task_execution_role

_create_task_role(scope)

Create IAM role for ECS task runtime permissions.

Creates an IAM role with comprehensive permissions for application runtime including RDS access, S3 operations, CloudWatch logging, Secrets Manager, Parameter Store integration and EFS access following least-privilege principles.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource creation	required

Returns:

Type	Description
Role	IAM role with task runtime permissions

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _create_task_role(self, scope: Construct) -> iam.Role:
    """Create IAM role for ECS task runtime permissions.

    Creates an IAM role with comprehensive permissions for application runtime
    including RDS access, S3 operations, CloudWatch logging, Secrets Manager,
    Parameter Store integration and EFS access following least-privilege principles.

    Args:
        scope: CDK construct scope for resource creation

    Returns:
        IAM role with task runtime permissions
    """
    task_role_name: str = self._get_name_for_resource("task_role", max_length=AWSResourceNameLength.ROLE.value)

    # Create a task role
    task_role = iam.Role(
        scope, self._get_cfn_logical_id("task_role"),
        assumed_by=iam.ServicePrincipal("ecs-tasks.amazonaws.com"),
        description="Role for ECS task with permissions for RDS, S3, CloudWatch Logs, Secrets Manager, SSM Parameter Store and EFS",
        role_name=task_role_name
    )

    # Attach permissions for RDS
    task_role.add_to_policy(
        iam.PolicyStatement(
            actions=[
                "rds:*"
            ],
            resources=["*"]
        )
    )

    if self._config.use_iam_authentication:
        rds_db_resource_id: str
        if self._application_helper.get_target_env().upper() != Environment.DEVOPS.name:
            consumer = CrossAccountResourceConsumer(self._application_helper)
            rds_resource_id_param = consumer.import_ssm_parameter(
                scope,
                parameter_key="RDS_DB_RESOURCE_ID",
                producer_account_id=self._application_helper.get_from_env("account_id"),
                secret_env_suffix=self._application_helper.get_target_env(),
                logical_id_prefix="cross-account-rds-rid"
            )
            rds_db_resource_id = Token.as_string(rds_resource_id_param.value)
        else: 
            rds_db_resource_id = self._application_helper.get_parameter_value(scope, "RDS_DB_RESOURCE_ID")

        task_role.add_to_policy(
            iam.PolicyStatement(
                actions=[
                    "rds-db:connect"
                ],
                resources=[
                    f"arn:aws:rds-db:{self._application_helper.get_from_common("region")}:{self._application_helper.get_from_env("account_id")}:dbuser:{rds_db_resource_id}/*"
                    # f"arn:aws:rds-db:{self._application_helper.get_from_common("region")}:{self._application_helper.get_from_env("account_id")}:dbuser:{rds_db_resource_id}/${{resolve:secretsmanager:{rds_app_user_secret.secret_arn}:SecretString:db_app_username}}"
                ]
            )
        )

    # Attach permissions for S3
    task_role.add_to_policy(
        iam.PolicyStatement(
            actions=[
                "s3:GetObject",
                "s3:ListBucket",
                "s3:PutObject"  # Add other S3 actions as needed
            ],
            resources=["*"]
        )
    )

    # Attach permissions for CloudWatch Logs
    task_role.add_to_policy(
        iam.PolicyStatement(
            actions=[
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents",
                "logs:DescribeLogStreams"
            ],
            resources=["*"]
        )
    )

    # Attach permissions for Secrets Manager
    task_role.add_to_policy(
        iam.PolicyStatement(
            actions=[
                "kms:*",
                "secretsmanager:GetSecretValue",
                "secretsmanager:DescribeSecret"  # Add other Secrets Manager actions as needed
            ],
            resources=["*"]
        )
    )

    # Attach permissions for SSM Parameter Store
    task_role.add_to_policy(
        iam.PolicyStatement(
            actions=[
                "ssm:GetParameters",
                "ssm:GetParameter"
            ],
            resources=["*"]
        )
    )

    # Add EFS permissions if EFS is configured
    if hasattr(self._config, 'efs_config') and self._config.efs_config is not None:
        task_role.add_to_policy(
            iam.PolicyStatement(
                actions=[
                    "elasticfilesystem:ClientRootAccess",
                    "elasticfilesystem:ClientWrite", 
                    "elasticfilesystem:ClientMount"
                ],
                resources=["*"]
            )
        )

    return task_role

_get_alb_security_groups(scope, vpc)

Retrieve security groups for Application Load Balancer configuration.

Fetches the required security groups for ALB traffic management including CloudFront integration and private network access patterns.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource lookup	required
vpc	IVpc	VPC instance for security group context	required

Returns:

Type	Description
List[SecurityGroup]	List of security groups for ALB attachment

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_alb_security_groups(self, scope: Construct, vpc: ec2.IVpc) -> List[ec2.SecurityGroup]:
    """Retrieve security groups for Application Load Balancer configuration.

    Fetches the required security groups for ALB traffic management including
    CloudFront integration and private network access patterns.

    Args:
        scope: CDK construct scope for resource lookup
        vpc: VPC instance for security group context

    Returns:
        List of security groups for ALB attachment
    """
    ecs_security_groups: List[ec2.SecurityGroup] = []

    full_name, _ = ResourceNaming.get_name_for_resource(self._application_helper, AWSResourceType.SG, resource_name="own-cloudfront-alb-sg",
                                         max_length=AWSResourceNameLength.SEC_GROUP.value)
    ecs_security_groups.append(self._get_security_group_by_name(scope, vpc, full_name))

    if self._config.admin_active:
        full_name, _ = ResourceNaming.get_name_for_resource(self._application_helper, AWSResourceType.SG, resource_name="own-private-alb-sg",
                                            max_length=AWSResourceNameLength.SEC_GROUP.value)
        ecs_security_groups.append(self._get_security_group_by_name(scope, vpc, full_name))
    return ecs_security_groups

_get_certificate_arn(certificate_id, region)

Build ACM certificate ARN from certificate ID and region.

Constructs the full ARN for an ACM certificate using the certificate ID and specified region, accounting for cross-region certificate usage patterns.

Parameters:

Name	Type	Description	Default
certificate_id	str	ACM certificate identifier	required
region	str	AWS region where the certificate is located	required

Returns:

Type	Description
str	Complete ACM certificate ARN

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_certificate_arn(self, certificate_id: str, region: str) -> str:
    """Build ACM certificate ARN from certificate ID and region.

    Constructs the full ARN for an ACM certificate using the certificate ID
    and specified region, accounting for cross-region certificate usage patterns.

    Args:
        certificate_id: ACM certificate identifier
        region: AWS region where the certificate is located

    Returns:
        Complete ACM certificate ARN
    """
    # Because the default region for this profile is eu-west-1, but, the certificate is in us-east-1, hence the use of the mocked region instead of {Aws.REGION}
    return f"arn:aws:acm:{region}:{Aws.ACCOUNT_ID}:certificate/{certificate_id}"

_get_container_user()

Determine the user the container should run as to match EFS access points.

Returns the user:group specification that matches the EFS access point configuration to ensure proper file permissions and ownership.

Returns:

Type	Description
Optional[str]	User specification in format "uid:gid" or None for default

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_container_user(self) -> Optional[str]:
    """Determine the user the container should run as to match EFS access points.

    Returns the user:group specification that matches the EFS access point
    configuration to ensure proper file permissions and ownership.

    Returns:
        User specification in format "uid:gid" or None for default
    """
    # If EFS is configured, try to match the first access point's user
    if hasattr(self._config, 'efs_config') and self._config.efs_config is not None:
        # Check if there's an explicit container user configured
        if self._config.efs_config.container_user is not None:
            return self._config.efs_config.container_user

    # Default to None (use container's default user)
    return None

_get_ecr_repo(scope)

Retrieve ECR repository reference for container image deployment.

Creates a reference to the existing ECR repository containing the application container images for deployment to ECS.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource reference	required

Returns:

Type	Description
Repository	ECR repository instance for container image access

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_ecr_repo(self, scope: Construct) -> ecr.Repository:
    """Retrieve ECR repository reference for container image deployment.

    Creates a reference to the existing ECR repository containing the
    application container images for deployment to ECS.

    Args:
        scope: CDK construct scope for resource reference

    Returns:
        ECR repository instance for container image access
    """
    ecr_repo_name = self._get_name_for_resource(self._config.ecr_repo_base_name, max_length=AWSResourceNameLength.ECR_REPOSITORY.value)
    account_id =self._application_helper.get_all_envs().get(self._config.ecr_repo_source_env).get("account_id")
    ecr_repo_arn = ArnUriHelper.build_ecr_repo_arn(self._application_helper, ecr_repo_name, account_id)
    ecr_repo = ecr.Repository.from_repository_attributes(
        scope, self._get_cfn_logical_id("app-ecr-repo"),
        repository_name=ecr_repo_name,
        repository_arn=ecr_repo_arn
    )
    return ecr_repo

_get_ecs_security_groups(scope, vpc)

Retrieve security groups for ECS service configuration.

Fetches the security groups required for ECS container networking and communication with other AWS services.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource lookup	required
vpc	IVpc	VPC instance for security group context	required

Returns:

Type	Description
List[SecurityGroup]	List of security groups for ECS service attachment

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_ecs_security_groups(self, scope: Construct, vpc: ec2.IVpc) -> List[ec2.SecurityGroup] :
    """Retrieve security groups for ECS service configuration.

    Fetches the security groups required for ECS container networking
    and communication with other AWS services.

    Args:
        scope: CDK construct scope for resource lookup
        vpc: VPC instance for security group context

    Returns:
        List of security groups for ECS service attachment
    """
    ecs_security_groups: List[ec2.SecurityGroup] = []
    full_name, _ = ResourceNaming.get_name_for_resource(self._application_helper, AWSResourceType.SG, resource_name="own-ecs-sg",
                                         max_length=AWSResourceNameLength.SEC_GROUP.value)
    security_group: ec2.SecurityGroup = self._get_security_group_by_name(scope, vpc, full_name)
    ecs_security_groups.append(security_group)

    return ecs_security_groups

_get_efs_access_point_id(scope, access_point_index)

Retrieve EFS access point ID from SSM Parameter Store with cross-account support.

Fetches the EFS access point identifier for fine-grained access control in ECS task definitions. Access points enable path-based access restrictions, user/group enforcement, and creation permissions for enhanced security in multi-tenant scenarios.

The method constructs parameter keys dynamically based on the access point index, supporting multiple access points per EFS file system. Parameter retrieval follows the same cross-account patterns as file system ID access.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for parameter reference creation	required
access_point_index	int	Zero-based index of the access point (0, 1, 2, ...)	required

Returns:

Type	Description
str	EFS access point ID string in format 'fsap-xxxxxxxxx' for authorization config

Raises:

Type	Description
ParameterNotFound	If the access point parameter doesn't exist for the given index
PermissionError	If cross-account access permissions are insufficient

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_efs_access_point_id(self, scope: Construct, access_point_index: int) -> str:
    """Retrieve EFS access point ID from SSM Parameter Store with cross-account support.

    Fetches the EFS access point identifier for fine-grained access control in ECS
    task definitions. Access points enable path-based access restrictions, user/group
    enforcement, and creation permissions for enhanced security in multi-tenant scenarios.

    The method constructs parameter keys dynamically based on the access point index,
    supporting multiple access points per EFS file system. Parameter retrieval follows
    the same cross-account patterns as file system ID access.

    Args:
        scope: CDK construct scope for parameter reference creation
        access_point_index: Zero-based index of the access point (0, 1, 2, ...)

    Returns:
        EFS access point ID string in format 'fsap-xxxxxxxxx' for authorization config

    Raises:
        ParameterNotFound: If the access point parameter doesn't exist for the given index
        PermissionError: If cross-account access permissions are insufficient
    """
    parameter_key = f"EFS_ACCESS_POINT_{access_point_index}_ID"

    if self._application_helper.get_target_env().upper() != Environment.DEVOPS.name:
        consumer = CrossAccountResourceConsumer(self._application_helper)
        ap_param = consumer.import_ssm_parameter(
            scope,
            parameter_key=parameter_key,
            producer_account_id=self._application_helper.get_from_env("account_id"),
            secret_env_suffix=self._application_helper.get_target_env(),
            logical_id_prefix=f"cross-account-efs-ap-{access_point_index}"
        )
        return Token.as_string(ap_param.value)
    else:
        return self._application_helper.get_parameter_value(scope, parameter_key)

_get_efs_file_system_id(scope)

Retrieve EFS file system ID from SSM Parameter Store with cross-account support.

Fetches the EFS file system identifier required for volume configuration in ECS task definitions. The method handles both local parameter access in DevOps environments and cross-account parameter sharing for production deployments.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for parameter reference creation	required

Returns:

Type	Description
str	EFS file system ID string in format 'fs-xxxxxxxxx' for volume configuration

Raises:

Type	Description
ParameterNotFound	If the EFS_FILE_SYSTEM_ID parameter doesn't exist
PermissionError	If cross-account access permissions are insufficient

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_efs_file_system_id(self, scope: Construct) -> str:
    """Retrieve EFS file system ID from SSM Parameter Store with cross-account support.

    Fetches the EFS file system identifier required for volume configuration in ECS
    task definitions. The method handles both local parameter access in DevOps
    environments and cross-account parameter sharing for production deployments.

    Args:
        scope: CDK construct scope for parameter reference creation

    Returns:
        EFS file system ID string in format 'fs-xxxxxxxxx' for volume configuration

    Raises:
        ParameterNotFound: If the EFS_FILE_SYSTEM_ID parameter doesn't exist
        PermissionError: If cross-account access permissions are insufficient
    """
    if self._application_helper.get_target_env().upper() != Environment.DEVOPS.name:
        consumer = CrossAccountResourceConsumer(self._application_helper)
        efs_param = consumer.import_ssm_parameter(
            scope,
            parameter_key="EFS_FILE_SYSTEM_ID",
            producer_account_id=self._application_helper.get_from_env("account_id"),
            secret_env_suffix=self._application_helper.get_target_env(),
            logical_id_prefix="cross-account-efs-fid"
        )
        return Token.as_string(efs_param.value)
    else:
        return self._application_helper.get_parameter_value(scope, "EFS_FILE_SYSTEM_ID")

_get_efs_parameter_value(scope, parameter_key)

Get EFS parameter value (file system ID or access point ID) from SSM.

Retrieves parameter values from SSM with support for cross-account access in non-devops environments for proper resource sharing between stacks.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for parameter retrieval	required
parameter_key	str	The SSM parameter key to retrieve	required

Returns:

Type	Description
str	Parameter value (EFS ID or access point ID)

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_efs_parameter_value(self, scope: Construct, parameter_key: str) -> str:
    """Get EFS parameter value (file system ID or access point ID) from SSM.

    Retrieves parameter values from SSM with support for cross-account access
    in non-devops environments for proper resource sharing between stacks.

    Args:
        scope: CDK construct scope for parameter retrieval
        parameter_key: The SSM parameter key to retrieve

    Returns:
        Parameter value (EFS ID or access point ID)
    """
    if self._application_helper.get_target_env().upper() != Environment.DEVOPS.name:
        # For non-devops environments, use cross-account parameter access
        if parameter_key in self._imported_params_cache:
            return Token.as_string(self._imported_params_cache[parameter_key].value)

        consumer = CrossAccountResourceConsumer(self._application_helper)
        param = consumer.import_ssm_parameter(
            scope,
            parameter_key=parameter_key,
            producer_account_id=self._application_helper.get_from_env("account_id"),
            secret_env_suffix=self._application_helper.get_target_env(),
            logical_id_prefix=f"cross-account-{parameter_key.lower().replace('_', '-')}"
        )
        self._imported_params_cache[parameter_key] = param

        return Token.as_string(param.value)
    else:
        # For devops environment, direct parameter access
        return self._application_helper.get_parameter_value(scope, parameter_key)

_get_route53_hosted_zone(scope)

Retrieve Route53 hosted zone reference for DNS record management.

Creates a reference to the existing Route53 hosted zone for creating DNS records that route traffic to the Application Load Balancer.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource reference	required

Returns:

Type	Description
HostedZone	Route53 hosted zone reference

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_route53_hosted_zone(self, scope: Construct) -> route53.HostedZone:
    """Retrieve Route53 hosted zone reference for DNS record management.

    Creates a reference to the existing Route53 hosted zone for creating
    DNS records that route traffic to the Application Load Balancer.

    Args:
        scope: CDK construct scope for resource reference

    Returns:
        Route53 hosted zone reference
    """
    return route53.HostedZone.from_hosted_zone_attributes(
        scope,
        self._get_cfn_logical_id("ecs-hosted-zone"),
        hosted_zone_id=self._config.route53.hosted_zone_id,
        zone_name=self._config.route53.domain_name
    )

_get_s3_bucket(scope, bucket_base_name)

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_s3_bucket(self, scope: Construct, bucket_base_name: str) -> s3.IBucket:
    bucket_name = self._get_name_for_resource(f"{bucket_base_name}-{self._application_helper.get_from_env("account_id")}", 
                                                max_length=AWSResourceNameLength.S3_BUCKET.value)

    return s3.Bucket.from_bucket_name(
        scope, 
        f"S3_ENV_FILES_BUCKET_{bucket_base_name.upper().replace("-", "_")}_{self._application_helper.get_project().upper()}",
        self._application_helper.get_parameter_value(scope, f"s3_bucket_{bucket_base_name}")
    )

_get_secrets(scope)

Build secrets configuration for container runtime injection.

Processes the secrets configuration to create ECS-compatible secret references from both AWS Secrets Manager and Systems Manager Parameter Store sources for secure credential management.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource references	required

Returns:

Type	Description
Dict[str, Secret]	Dictionary mapping environment variable names to ECS secret definitions

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_secrets(self, scope: Construct) -> Dict[str, ecs.Secret]:
    """Build secrets configuration for container runtime injection.

    Processes the secrets configuration to create ECS-compatible secret
    references from both AWS Secrets Manager and Systems Manager Parameter
    Store sources for secure credential management.

    Args:
        scope: CDK construct scope for resource references

    Returns:
        Dictionary mapping environment variable names to ECS secret definitions
    """
    secrets = {}
    for secret_name, secret_cfg in self._config.parameters.secrets.items():
        if secret_cfg.value_from == SecretValueFrom.PARAM:
            ssm_param_name = self._application_helper._get_parameter_name_from_key(secret_cfg.name, with_leading_slash=True)
            ssm_param = ssm.StringParameter.from_string_parameter_name(
                scope,
                self._get_cfn_logical_id(f"{secret_name}-{secret_cfg.value_from}"),
                ssm_param_name)
            secrets[secret_name] = ecs.Secret.from_ssm_parameter(ssm_param)
        elif secret_cfg.value_from == SecretValueFrom.SECRET:
            sm_secret: secretsmanager.ISecret = secretsmanager.Secret.from_secret_complete_arn(
                scope,
                self._get_cfn_logical_id(f"{secret_name}-{secret_cfg.value_from}"),
                secret_cfg.arn)
            secrets[secret_name] = ecs.Secret.from_secrets_manager(sm_secret, secret_cfg.suffix)

    return secrets

_get_security_group_by_name(scope, vpc, sg_name)

Lookup and return an immutable security group reference.

Retrieves a security group by name and returns an immutable reference to prevent accidental modification of existing security group rules.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource lookup	required
vpc	IVpc	VPC instance containing the security group	required
sg_name	str	Name of the security group to lookup	required

Returns:

Type	Description
SecurityGroup	Immutable security group reference

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_security_group_by_name(self, scope: Construct, vpc: ec2.IVpc, sg_name: str) -> ec2.SecurityGroup:
    """Lookup and return an immutable security group reference.

    Retrieves a security group by name and returns an immutable reference
    to prevent accidental modification of existing security group rules.

    Args:
        scope: CDK construct scope for resource lookup
        vpc: VPC instance containing the security group
        sg_name: Name of the security group to lookup

    Returns:
        Immutable security group reference
    """
    mutable_security_group: ec2.SecurityGroup = ec2.SecurityGroup.from_lookup_by_name(scope, f"m-{sg_name}",
                                                                                      sg_name, vpc)
    immutable_security_group: ec2.SecurityGroup = ec2.SecurityGroup.from_security_group_id(scope, sg_name,
                                                                                           mutable_security_group.security_group_id,
                                                                                           mutable=False)
    return immutable_security_group

_get_ssl_certificate(scope)

Retrieve SSL certificate reference for HTTPS termination.

Creates a reference to the existing ACM certificate for SSL/TLS termination at the Application Load Balancer level.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource reference	required

Returns:

Type	Description
ICertificate	ACM certificate reference for HTTPS configuration

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _get_ssl_certificate(self, scope: Construct) -> acm.ICertificate:
    """Retrieve SSL certificate reference for HTTPS termination.

    Creates a reference to the existing ACM certificate for SSL/TLS
    termination at the Application Load Balancer level.

    Args:
        scope: CDK construct scope for resource reference

    Returns:
        ACM certificate reference for HTTPS configuration
    """
    ssl_certificate = acm.Certificate.from_certificate_arn(
        scope,
        self._get_cfn_logical_id(f"{self._application_helper.get_target_env()}-ssl-certificate"),
        certificate_arn=self._get_certificate_arn(self._config.ssl_certificate_id, "eu-west-1")
    )
    return ssl_certificate

_set_config()

Validate and set the ECS configuration from builder inputs.

Combines the builder configuration with Route53 and application parameters to create a validated EcsConfig instance, ensuring all required fields and validation rules are satisfied.

Raises:

Type	Description
ValidationError	If the combined configuration fails validation

Source code in mare_aws_common_lib/builders/ecs_builder.py

def _set_config(self) -> None:
    """Validate and set the ECS configuration from builder inputs.

    Combines the builder configuration with Route53 and application parameters
    to create a validated EcsConfig instance, ensuring all required fields
    and validation rules are satisfied.

    Raises:
        ValidationError: If the combined configuration fails validation
    """
    try:
        self._config = EcsConfig(**{
            **self._builder_config,
            "route53": self._route53_config,
            "parameters": self._app_config
        })
    except ValidationError as e:
        self._log_validation_error(e, EcsConfig)
        raise

build(scope)

Build the complete ECS infrastructure stack.

Creates and configures all AWS resources required for a production-ready containerized application including networking, compute, security, logging, DNS components and EFS storage with proper integration and dependencies.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope for resource creation	required

Returns:

Type	Description
ApplicationLoadBalancer	Application Load Balancer instance for the deployed service

Raises:

Type	Description
ValidationError	If configuration validation fails
ValueError	If required configurations are missing

Source code in mare_aws_common_lib/builders/ecs_builder.py

def build(self, scope: Construct) -> elbv2.ApplicationLoadBalancer:
    """Build the complete ECS infrastructure stack.

    Creates and configures all AWS resources required for a production-ready
    containerized application including networking, compute, security, logging,
    DNS components and EFS storage with proper integration and dependencies.

    Args:
        scope: CDK construct scope for resource creation

    Returns:
        Application Load Balancer instance for the deployed service

    Raises:
        ValidationError: If configuration validation fails
        ValueError: If required configurations are missing
    """
    super().build()

    vpc: ec2.Vpc = self._application_helper.get_vpc("vpc-ecs", scope)
    cluster: ecs.Cluster = self._create_cluster(scope, vpc)
    task_execution_role: iam.Role = self._create_task_execution_role(scope)
    task_role: iam.Role = self._create_task_role(scope)

    log_group: logs.LogGroup = self._create_log_group(scope)

    task_definition: ecs.FargateTaskDefinition = self._create_task_definition(scope, task_execution_role, task_role)

    self._add_container_definition(scope, log_group, task_definition)

    # Create a load balancer
    alb_name: str = self._get_name_for_resource(
        f"{self._application_helper.get_target_env()}-{self._config.alb_name}",
        max_length=AWSResourceNameLength.ALB.value)

    alb_security_groups: List[ec2.SecurityGroup] = self._get_alb_security_groups(scope, vpc)

    alb: elbv2.ApplicationLoadBalancer = elbv2.ApplicationLoadBalancer(
        scope,
        self._get_cfn_logical_id(self._config.alb_name),
        load_balancer_name=alb_name,
        internet_facing=True,
        vpc=vpc,
        security_group=alb_security_groups[0]
    )

    if len(alb_security_groups) >= 2:
        for i in range(1, len(alb_security_groups)):
            alb.add_security_group(alb_security_groups[i])

    # Create an ECS service
    service_name: str = self._get_name_for_resource(
        f"{self._application_helper.get_target_env()}-{self._config.service_name}",
        max_length=AWSResourceNameLength.ECS.value)

    ecs_security_groups: List[ec2.SecurityGroup] = self._get_ecs_security_groups(scope, vpc)

    ecs_service: ecs.FargateService = ecs.FargateService(
        scope,
        self._get_cfn_logical_id(self._config.service_name),
        service_name=service_name,
        cluster=cluster,
        task_definition=task_definition,
        desired_count=self._config.desired_count,
        min_healthy_percent=100 if self._config.desired_count == 1 else 50,
        max_healthy_percent=200,
        assign_public_ip=False,
        security_groups=ecs_security_groups,
        vpc_subnets=ec2.SubnetSelection(
            subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS
        ),
        capacity_provider_strategies=[
            ecs.CapacityProviderStrategy(capacity_provider="FARGATE_SPOT", weight=2),
            ecs.CapacityProviderStrategy(capacity_provider="FARGATE", weight=1)
        ]
    )

    self._add_listener(scope,
                       alb,
                       ecs_service,
                       vpc,
                       name="app",
                       listening_port=443,
                       target_port=self._config.application_port,
                       ssl_certificate=self._get_ssl_certificate(scope))

    # create A record for alb
    self._create_alb_route53_a_record(scope, alb)

    # Store ALB arn in SSM in order to use it in the cloudfront builder
    self._application_helper.store_parameter(scope,
                                             "ALB_arn",
                                             alb.load_balancer_arn)

    return alb

reset()

Reset builder state and clear configuration data.

Initializes the builder for a new build cycle by clearing all configuration data and resetting internal state to default values.

Source code in mare_aws_common_lib/builders/ecs_builder.py

def reset(self) -> None:
    """Reset builder state and clear configuration data.

    Initializes the builder for a new build cycle by clearing all
    configuration data and resetting internal state to default values.
    """
    super().reset()
    self._route53_config: Dict[str, Any] = {}
    self._app_config: Dict[str, Any] = {}
    self._imported_params_cache: Dict[str, CfnParameter] = {}

set_app_parameters(config)

Configure application runtime parameters and secrets.

Sets environment variables and secret references that will be injected into the ECS container at runtime, supporting both plain text environment variables and secure credential retrieval.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Application parameters configuration with environment and secrets	required

Returns:

Type	Description
ECSBuilder	Builder instance for method chaining

Source code in mare_aws_common_lib/builders/ecs_builder.py

def set_app_parameters(self, config: Dict[str, Any]) -> 'ECSBuilder':
    """Configure application runtime parameters and secrets.

    Sets environment variables and secret references that will be injected
    into the ECS container at runtime, supporting both plain text environment
    variables and secure credential retrieval.

    Args:
        config: Application parameters configuration with environment and secrets

    Returns:
        Builder instance for method chaining
    """
    self._app_config = config
    return self

set_route53_config(config)

Configure Route53 DNS settings for the ECS service.

Sets the hosted zone and domain configuration required for creating DNS records that route traffic to the Application Load Balancer.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Route53 configuration containing hosted_zone_id and domain_name	required

Returns:

Type	Description
ECSBuilder	Builder instance for method chaining

Source code in mare_aws_common_lib/builders/ecs_builder.py

def set_route53_config(self, config: Dict[str, Any]) -> 'ECSBuilder':
    """Configure Route53 DNS settings for the ECS service.

    Sets the hosted zone and domain configuration required for creating
    DNS records that route traffic to the Application Load Balancer.

    Args:
        config: Route53 configuration containing hosted_zone_id and domain_name

    Returns:
        Builder instance for method chaining
    """
    self._route53_config = config
    return self