Batch Constructs

AWS Batch infrastructure, compute environments, and job management.

batch

Modules

infrastructure

AWS Batch infrastructure constructs.

This module provides CDK constructs for creating and managing AWS Batch compute environments, job queues, and related infrastructure.

Classes

Batch

Batch(
    scope: Construct,
    id: str,
    env_base: EnvBase,
    vpc: IVpc,
    instance_role_name: str | None = None,
    instance_role_policy_statements: list[PolicyStatement]
    | None = None,
)

Bases: EnvBaseConstruct

AWS Batch infrastructure construct for creating multiple Batch environments.

This construct simplifies the creation of Batch Environments. It allows for the creation of multiple Batch Environments with different configurations and launch templates, but using the same instance role and security group.

Note

Instance Roles are created with managed policies commonly used by Batch jobs, including access to S3, Lambda, and DynamoDB.

Defines

• Batch Compute Environment (Spot and OnDemand)
• Instance Role
• Launch Template
• Queue(s)

Initialize Batch infrastructure construct.

Creates the shared infrastructure for Batch Environments. Has the ability to create multiple Batch Environments with different configurations.

Parameters:

Name	Type	Description	Default
scope	Construct	The construct scope.	required
id	str	The construct ID.	required
env_base	EnvBase	Environment base to use for resource naming.	required
vpc	IVpc	VPC to use for Batch infrastructure.	required
instance_role_name	Optional[str]	Name for the instance role. Defaults to None (auto-generated).	None
instance_role_policy_statements	Optional[List[PolicyStatement]]	Additional policy statements for the instance role. Defaults to None.	None

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/infrastructure.py

def __init__(
    self,
    scope: constructs.Construct,
    id: str,
    env_base: EnvBase,
    vpc: ec2.IVpc,
    instance_role_name: str | None = None,
    instance_role_policy_statements: list[iam.PolicyStatement] | None = None,
) -> None:
    """Initialize Batch infrastructure construct.

    Creates the shared infrastructure for Batch Environments.
    Has the ability to create multiple Batch Environments with different configurations.

    Args:
        scope (constructs.Construct): The construct scope.
        id (str): The construct ID.
        env_base (EnvBase): Environment base to use for resource naming.
        vpc (ec2.IVpc): VPC to use for Batch infrastructure.
        instance_role_name (Optional[str]): Name for the instance role.
            Defaults to None (auto-generated).
        instance_role_policy_statements (Optional[List[iam.PolicyStatement]]):
            Additional policy statements for the instance role.
            Defaults to None.
    """
    super().__init__(scope, id, env_base)
    self.vpc = vpc

    # ---------------------------------------------------------------------
    # Shared EC2 pieces between the Compute Environments
    #  - instance role/profile
    #  - security group
    #  - launch template
    # ---------------------------------------------------------------------
    self.instance_role = self.create_instance_role(
        role_name=instance_role_name,
        statements=instance_role_policy_statements,
    )
    self.instance_profile = self.create_instance_profile(self.instance_role.role_name)
    self.security_group = self.create_security_group()

    self._batch_environment_mapping: MutableMapping[str, BatchEnvironment] = {}

Attributes

environments property

environments: list[BatchEnvironment]

Get all Batch environments sorted by name.

Returns:

Type	Description
list[BatchEnvironment]	List of BatchEnvironment instances.

Functions

create_instance_role

create_instance_role(
    role_name: str | None = None,
    statements: list[PolicyStatement] | None = None,
) -> Role

Create an IAM role for Batch EC2 instances.

Parameters:

Name	Type	Description	Default
role_name	Optional[str]	Name for the role. Defaults to None.	None
statements	Optional[List[PolicyStatement]]	Additional policy statements to attach. Defaults to None.	None

Returns:

Type	Description
Role	The created IAM role.

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/infrastructure.py

def create_instance_role(
    self,
    role_name: str | None = None,
    statements: list[iam.PolicyStatement] | None = None,
) -> iam.Role:
    """Create an IAM role for Batch EC2 instances.

    Args:
        role_name (Optional[str]): Name for the role. Defaults to None.
        statements (Optional[List[iam.PolicyStatement]]): Additional policy
            statements to attach. Defaults to None.

    Returns:
        The created IAM role.
    """
    instance_role = iam.Role(
        self,
        self.get_child_id(self, "instance-role"),
        role_name=role_name,
        description="Role used by ec2 instance in batch compute environment",
        assumed_by=iam.ServicePrincipal("ec2.amazonaws.com"),  # type: ignore  # Interface not inferred
    )
    managed_policy_names = [
        "service-role/AmazonEC2ContainerServiceforEC2Role",
        "AmazonS3ReadOnlyAccess",
        "AmazonSSMManagedInstanceCore",
        "AmazonElasticFileSystemClientReadWriteAccess",
        "CloudWatchAgentServerPolicy",
    ]
    for mpn in managed_policy_names:
        instance_role.add_managed_policy(iam.ManagedPolicy.from_aws_managed_policy_name(mpn))
    # Used for EBS autoscaling.
    iam.Policy(
        self,
        "ebs-autoscale-policy",
        statements=[
            iam.PolicyStatement(
                actions=[
                    "ec2:AttachVolume",
                    "ec2:DescribeVolumeStatus",
                    "ec2:DescribeVolumes",
                    "ec2:ModifyInstanceAttribute",
                    "ec2:DescribeVolumeAttribute",
                    "ec2:CreateVolume",
                    "ec2:DeleteVolume",
                    "ec2:CreateTags",
                ],
                effect=iam.Effect.ALLOW,
                resources=["*"],
            )
        ],
        roles=[instance_role],  # type: ignore  # Role is not inferred as IRole
    )

    # Used for S3 / Lambda
    iam.Policy(
        self,
        "s3-ecs-env",
        statements=[
            # allow read access from all buckets
            iam.PolicyStatement(
                sid="AllReadObjectActions",
                actions=S3_READ_ONLY_ACCESS_ACTIONS,
                effect=iam.Effect.ALLOW,
                resources=[
                    "arn:aws:s3:::*",
                    "arn:aws:s3:::*/*",
                ],
            ),
            iam.PolicyStatement(
                sid="AllObjectActions",
                actions=[
                    "s3:*Object",
                    "s3:GetBucket*",
                    "s3:List*",
                ],
                effect=iam.Effect.ALLOW,
                resources=[
                    f"arn:aws:s3:::{self.env_base}-*",
                    f"arn:aws:s3:::{self.env_base}-*/*",
                ],
            ),
            iam.PolicyStatement(
                sid="AllowCallDescribeInstances",
                actions=["ecs:DescribeContainerInstances"],
                effect=iam.Effect.ALLOW,
                resources=["*"],
            ),
            batch_policy_statement(actions=BATCH_READ_ONLY_ACTIONS, env_base=self.env_base),
        ],
        roles=[instance_role],  # type: ignore  # Role is not inferred as IRole
    )

    # Custom policy
    if statements:
        iam.Policy(
            self,
            "custom-policy",
            statements=statements,
            roles=[instance_role],  # type: ignore  # Role is not inferred as IRole
        )

    return instance_role

create_instance_profile

create_instance_profile(
    instance_role_name: str,
) -> CfnInstanceProfile

Create an instance profile for the Batch instance role.

Parameters:

Name	Type	Description	Default
instance_role_name	str	Name of the IAM role to associate.	required

Returns:

Type	Description
CfnInstanceProfile	The created instance profile.

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/infrastructure.py

def create_instance_profile(self, instance_role_name: str) -> iam.CfnInstanceProfile:
    """Create an instance profile for the Batch instance role.

    Args:
        instance_role_name (str): Name of the IAM role to associate.

    Returns:
        The created instance profile.
    """
    return iam.CfnInstanceProfile(
        self,
        "instance-profile",
        roles=[instance_role_name],
    )

create_security_group

create_security_group() -> SecurityGroup

Create a security group for Batch instances.

Returns:

Type	Description
SecurityGroup	The created security group with all outbound and self-ingress allowed.

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/infrastructure.py

def create_security_group(self) -> ec2.SecurityGroup:
    """Create a security group for Batch instances.

    Returns:
        The created security group with all outbound and self-ingress allowed.
    """
    security_group = ec2.SecurityGroup(
        self,
        self.get_construct_id("batch", "sg"),
        vpc=self.vpc,
        allow_all_outbound=True,
        description=f"Batch instance security group for {self.env_base}",
    )
    security_group.add_ingress_rule(peer=security_group, connection=ec2.Port.all_traffic())
    return security_group

grant_instance_role_permissions

grant_instance_role_permissions(
    read_write_resources: Iterable[
        Bucket | FileSystem | IFileSystem
    ]
    | None = None,
    read_only_resources: Iterable[
        Bucket | FileSystem | IFileSystem
    ]
    | None = None,
) -> None

Grant the instance role permissions to access resources.

Parameters:

Name	Type	Description	Default
read_write_resources	Optional[Iterable[Union[Bucket, FileSystem, IFileSystem]]]	Resources to grant read/write access to.	None
read_only_resources	Optional[Iterable[Union[Bucket, FileSystem, IFileSystem]]]	Resources to grant read-only access to.	None

Raises:

Type	Description
ValueError	If an unsupported resource type is provided.

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/infrastructure.py

def grant_instance_role_permissions(
    self,
    read_write_resources: Iterable[s3.Bucket | efs.FileSystem | efs.IFileSystem] | None = None,
    read_only_resources: Iterable[s3.Bucket | efs.FileSystem | efs.IFileSystem] | None = None,
) -> None:
    """Grant the instance role permissions to access resources.

    Args:
        read_write_resources (Optional[Iterable[Union[s3.Bucket, efs.FileSystem, efs.IFileSystem]]]):
            Resources to grant read/write access to.
        read_only_resources (Optional[Iterable[Union[s3.Bucket, efs.FileSystem, efs.IFileSystem]]]):
            Resources to grant read-only access to.

    Raises:
        ValueError: If an unsupported resource type is provided.
    """  # noqa: E501
    for resource in read_write_resources or []:
        if isinstance(resource, s3.Bucket):
            resource.grant_read_write(self.instance_role)
        elif isinstance(resource, efs.FileSystem):
            grant_role_file_system_access(resource, self.instance_role, "rw")  # type: ignore  # Role is not inferred as IRole
        else:
            raise ValueError(f"Unsupported resource type: {resource}")

    for resource in read_only_resources or []:
        if isinstance(resource, s3.Bucket):
            resource.grant_read(self.instance_role)
        elif isinstance(resource, efs.FileSystem):
            grant_role_file_system_access(resource, self.instance_role, "r")  # type: ignore  # Role is not inferred as IRole
        else:
            raise ValueError(f"Unsupported resource type: {resource}")

setup_batch_environment

setup_batch_environment(
    descriptor: IBatchEnvironmentDescriptor,
    config: BatchEnvironmentConfig,
    launch_template_builder: IBatchLaunchTemplateBuilder
    | None = None,
) -> BatchEnvironment

Set up a new Batch environment.

Parameters:

Name	Type	Description	Default
descriptor	IBatchEnvironmentDescriptor	Descriptor defining the environment.	required
config	BatchEnvironmentConfig	Configuration for the environment.	required
launch_template_builder	Optional[IBatchLaunchTemplateBuilder]	Optional builder for creating a launch template.	None

Returns:

Type	Description
BatchEnvironment	The created BatchEnvironment construct.

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/infrastructure.py

def setup_batch_environment(
    self,
    descriptor: IBatchEnvironmentDescriptor,
    config: "BatchEnvironmentConfig",
    launch_template_builder: IBatchLaunchTemplateBuilder | None = None,
) -> "BatchEnvironment":
    """Set up a new Batch environment.

    Args:
        descriptor (IBatchEnvironmentDescriptor): Descriptor defining the environment.
        config (BatchEnvironmentConfig): Configuration for the environment.
        launch_template_builder (Optional[IBatchLaunchTemplateBuilder]): Optional
            builder for creating a launch template.

    Returns:
        The created BatchEnvironment construct.
    """
    if launch_template_builder:
        launch_template_builder.grant_instance_role_permissions(self.instance_role)

    # ---------------------------------------------------------------------
    # Batch Environments
    # ---------------------------------------------------------------------
    batch_env = BatchEnvironment(
        self,
        f"{descriptor.get_name()}-batch-env",
        env_base=self.env_base,
        descriptor=descriptor,
        config=config,
        vpc=self.vpc,
        instance_role=self.instance_role,  # type: ignore  # Role is not inferred as IRole
        security_group=self.security_group,
        launch_template_builder=launch_template_builder,
    )
    batch_env.node.add_dependency(self.instance_profile)
    self._batch_environment_mapping[descriptor.get_name()] = batch_env
    return batch_env

BatchEnvironmentConfig dataclass

BatchEnvironmentConfig(
    allocation_strategy: AllocationStrategy | None,
    instance_types: list[str | InstanceType] | None,
    use_public_subnets: bool,
    use_spot: bool,
    use_fargate: bool = False,
    maxv_cpus: int | None = DEFAULT_MAXV_CPUS,
    minv_cpus: int | None = DEFAULT_MINV_CPUS,
)

Configuration for a Batch compute environment.

Attributes:

Name	Type	Description
allocation_strategy	Optional[AllocationStrategy]	The allocation strategy for the compute environment.
instance_types	Optional[List[Union[str, InstanceType]]]	Instance types to use in the compute environment.
use_public_subnets	bool	Whether to use public subnets.
use_spot	bool	Whether to use spot instances.
use_fargate	bool	Whether to use Fargate. Defaults to False.
maxv_cpus	Optional[int]	Maximum vCPUs for the environment. Defaults to 10240.
minv_cpus	Optional[int]	Minimum vCPUs for the environment. Defaults to 0.

Attributes

spot_bid_percentage property

spot_bid_percentage: int | None

Get the spot bid percentage.

Returns:

Type	Description
int | None	75 for spot instances (25% discount), None for on-demand.

BatchEnvironment

BatchEnvironment(
    scope: Construct,
    id: str,
    env_base: EnvBase,
    descriptor: IBatchEnvironmentDescriptor,
    config: BatchEnvironmentConfig,
    vpc: IVpc,
    instance_role: IRole,
    security_group: SecurityGroup,
    launch_template_builder: IBatchLaunchTemplateBuilder
    | None = None,
)

Bases: EnvBaseConstruct

A single Batch compute environment with its job queue.

This construct creates a Batch compute environment and associated job queue. It supports both Fargate and EC2 compute environments, with customizable configurations and launch templates.

Initialize a BatchEnvironment construct.

Parameters:

Name	Type	Description	Default
scope	Construct	The construct scope.	required
id	str	The construct ID.	required
env_base	EnvBase	Environment base for resource naming.	required
descriptor	IBatchEnvironmentDescriptor	Environment descriptor.	required
config	BatchEnvironmentConfig	Environment configuration.	required
vpc	IVpc	VPC for the environment.	required
instance_role	IRole	IAM role for instances.	required
security_group	SecurityGroup	Security group for instances.	required
launch_template_builder	Optional[IBatchLaunchTemplateBuilder]	Optional launch template builder.	None

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/infrastructure.py

def __init__(
    self,
    scope: constructs.Construct,
    id: str,
    env_base: EnvBase,
    descriptor: IBatchEnvironmentDescriptor,
    config: BatchEnvironmentConfig,
    vpc: ec2.IVpc,
    instance_role: iam.IRole,
    security_group: ec2.SecurityGroup,
    launch_template_builder: IBatchLaunchTemplateBuilder | None = None,
):
    """Initialize a BatchEnvironment construct.

    Args:
        scope (constructs.Construct): The construct scope.
        id (str): The construct ID.
        env_base (EnvBase): Environment base for resource naming.
        descriptor (IBatchEnvironmentDescriptor): Environment descriptor.
        config (BatchEnvironmentConfig): Environment configuration.
        vpc (ec2.IVpc): VPC for the environment.
        instance_role (iam.IRole): IAM role for instances.
        security_group (ec2.SecurityGroup): Security group for instances.
        launch_template_builder (Optional[IBatchLaunchTemplateBuilder]):
            Optional launch template builder.
    """
    super().__init__(scope, id, env_base)
    self._descriptor = descriptor
    self.config = config
    self.vpc = vpc
    self.instance_role = instance_role
    self.security_group = security_group
    self.launch_template_builder = launch_template_builder

    # Set up max number of compute environments
    compute_env = batch.OrderedComputeEnvironment(
        compute_environment=self.create_compute_environment(), order=1
    )
    self.compute_environments = [compute_env]
    job_queue = batch.JobQueue(
        self,
        self.job_queue_name,
        job_queue_name=self.job_queue_name,
        compute_environments=self.compute_environments,
    )
    self.job_queue = job_queue

Attributes

descriptor property

descriptor: IBatchEnvironmentDescriptor

Get the environment descriptor.

Returns:

Type	Description
IBatchEnvironmentDescriptor	The batch environment descriptor.

job_queue_name property

job_queue_name: str

Get the job queue name.

Returns:

Type	Description
str	The name of the job queue.

instance_types property

instance_types: Sequence[InstanceType] | None

Get the configured instance types.

Returns:

Type	Description
Sequence[InstanceType] | None	List of instance types, or None if not configured.

vpc_subnets property

vpc_subnets: SubnetSelection | None

Get the VPC subnet selection.

Returns:

Type	Description
SubnetSelection | None	SubnetSelection for public subnets if configured, None otherwise.

launch_template_user_data_hash property

launch_template_user_data_hash: str | None

Get a hash of the launch template user data.

Returns:

Type	Description
str | None	SHA256 hash of user data, or None if no launch template.

compute_resource_tags property

compute_resource_tags: Mapping[str, str] | None

Get tags for compute resources.

Includes a hash of user data to force recreation when launch template changes.

Returns:

Type	Description
Mapping[str, str] | None	Dictionary of tags, or None for Fargate environments.

compute_resource_type property

compute_resource_type: Literal[
    "ON_DEMAND", "SPOT", "FARGATE", "FARGATE_SPOT"
]

Get the compute resource type.

Returns:

Type	Description
Literal['ON_DEMAND', 'SPOT', 'FARGATE', 'FARGATE_SPOT']	The compute resource type based on configuration.

Functions

launch_template

launch_template() -> LaunchTemplate | None

Get or create the launch template.

Returns:

Type	Description
LaunchTemplate | None	The launch template, or None for Fargate environments.

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/infrastructure.py

@cached_property
def launch_template(self) -> ec2.LaunchTemplate | None:
    """Get or create the launch template.

    Returns:
        The launch template, or None for Fargate environments.
    """
    launch_template = None
    if self.launch_template_builder and not self.config.use_fargate:
        launch_template = self.launch_template_builder.create_launch_template(
            self.descriptor, self.security_group
        )
    return launch_template

create_compute_environment

create_compute_environment() -> IComputeEnvironment

Create the compute environment.

Returns:

Type	Description
IComputeEnvironment	A Fargate or EC2 compute environment based on configuration.

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/infrastructure.py

def create_compute_environment(self) -> batch.IComputeEnvironment:
    """Create the compute environment.

    Returns:
        A Fargate or EC2 compute environment based on configuration.
    """
    ce: batch.FargateComputeEnvironment | batch.ManagedEc2EcsComputeEnvironment
    if self.config.use_fargate:
        ce = batch.FargateComputeEnvironment(
            self,
            f"{self.descriptor.get_name()}-ce",
            vpc=self.vpc,
            maxv_cpus=self.config.maxv_cpus,
            spot=self.config.use_spot,
            vpc_subnets=self.vpc_subnets,
        )
    else:
        ce = batch.ManagedEc2EcsComputeEnvironment(
            self,
            f"{self.descriptor.get_name()}-ce",
            vpc=self.vpc,
            allocation_strategy=self.config.allocation_strategy,
            maxv_cpus=self.config.maxv_cpus,
            minv_cpus=self.config.minv_cpus,
            instance_role=self.instance_role,
            launch_template=self.launch_template,
            instance_types=self.instance_types,
            # When instance types are explicitly configured, dont use optimal instance classes
            # because we want to control the instance types via configuration and launch
            # templates. If no instance types are specified, fall back to optimal selection.
            use_optimal_instance_classes=self.instance_types is None,
            spot=self.config.use_spot,
            spot_bid_percentage=self.config.spot_bid_percentage,
            vpc_subnets=self.vpc_subnets,
        )

    if tags := self.compute_resource_tags:
        for tag_key, tag_value in tags.items():
            ce.tags.set_tag(key=tag_key, value=tag_value)
    return ce

grant_file_system_access

grant_file_system_access(
    *file_systems: IFileSystem,
) -> None

Grant compute environments access to EFS file systems.

Parameters:

Name	Type	Description	Default
*file_systems	IFileSystem	Variable number of file systems to grant access to.	()

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/infrastructure.py

def grant_file_system_access(self, *file_systems: efs.IFileSystem) -> None:
    """Grant compute environments access to EFS file systems.

    Args:
        *file_systems (efs.IFileSystem): Variable number of file systems
            to grant access to.
    """
    for file_system in file_systems:
        for ce in self.job_queue.compute_environments:
            grant_connectable_file_system_access(
                file_system, cast(ec2.IConnectable, ce.compute_environment)
            )

Functions

launch_template

Classes

CloudWatchConfigBuilder dataclass

CloudWatchConfigBuilder(
    env_base: EnvBase, batch_env_name: str
)

Functions

to_json

to_json() -> dict[str, Any]

Builds a CW Agent config

https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-Agent-Configuration-File-Details.html

Returns:

Type	Description
dict[str, Any]	Dict[str, Any]: config

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/launch_template.py

def to_json(self) -> dict[str, Any]:
    """Builds a CW Agent config

    # https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-Agent-Configuration-File-Details.html

    Returns:
        Dict[str, Any]: config
    """
    return {
        "agent": {
            "metrics_collection_interval": 60,
            "logfile": "/opt/aws/amazon-cloudwatch-agent/logs/amazon-cloudwatch-agent.log",
        },
        "logs": self.get_logs_config(),
        "metrics": self.get_metrics_config(),
    }

get_logs_config

get_logs_config() -> dict[str, Any]

Generates CW Agent sub config for logs

https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-Agent-Configuration-File-Details.html#CloudWatch-Agent-Configuration-File-Logssection

Returns:

Type	Description
dict[str, Any]	Dict[str, Any]: CW logs config

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/launch_template.py

def get_logs_config(self) -> dict[str, Any]:
    """Generates CW Agent sub config for logs

    https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-Agent-Configuration-File-Details.html#CloudWatch-Agent-Configuration-File-Logssection

    Returns:
        Dict[str, Any]: CW logs config
    """
    env_base = self.env_base
    return {
        "logs_collected": {
            "files": {
                "collect_list": [
                    {
                        "file_path": "/opt/aws/amazon-cloudwatch-agent/logs/amazon-cloudwatch-agent.log",
                        "log_group_name": f"/aws/ecs/container-instance/{env_base}",
                        "log_stream_name": f"/aws/ecs/container-instance/{env_base}/{{instance_id}}/amazon-cloudwatch-agent.log",
                    },
                    {
                        "file_path": "/var/log/cloud-init.log",
                        "log_group_name": f"/aws/ecs/container-instance/{env_base}",
                        "log_stream_name": f"/aws/ecs/container-instance/{env_base}/{{instance_id}}/cloud-init.log",
                    },
                    {
                        "file_path": "/var/log/cloud-init-output.log",
                        "log_group_name": f"/aws/ecs/container-instance/{env_base}",
                        "log_stream_name": f"/aws/ecs/container-instance/{env_base}/{{instance_id}}/cloud-init-output.log",
                    },
                    {
                        "file_path": "/var/log/ecs/ecs-init.log",
                        "log_group_name": f"/aws/ecs/container-instance/{env_base}",
                        "log_stream_name": f"/aws/ecs/container-instance/{env_base}/{{instance_id}}/ecs-init.log",
                    },
                    {
                        "file_path": "/var/log/ecs/ecs-agent.log",
                        "log_group_name": f"/aws/ecs/container-instance/{env_base}",
                        "log_stream_name": f"/aws/ecs/container-instance/{env_base}/{{instance_id}}/ecs-agent.log",
                    },
                    {
                        "file_path": "/var/log/ecs/ecs-volume-plugin.log",
                        "log_group_name": f"/aws/ecs/container-instance/{env_base}",
                        "log_stream_name": f"/aws/ecs/container-instance/{env_base}/{{instance_id}}/ecs-volume-plugin.log",
                    },
                    {
                        "file_path": "/var/log/ebs-autoscale-install.log",
                        "log_group_name": f"/aws/ecs/container-instance/{env_base}",
                        "log_stream_name": f"/aws/ecs/container-instance/{env_base}/{{instance_id}}/ebs-autoscale-install.log",
                    },
                    {
                        "file_path": "/var/log/ebs-autoscale.log",
                        "log_group_name": f"/aws/ecs/container-instance/{env_base}",
                        "log_stream_name": f"/aws/ecs/container-instance/{env_base}/{{instance_id}}/ebs-autoscale.log",
                    },
                ]
            }
        }
    }

get_metrics_config

get_metrics_config() -> dict[str, Any]

Generates metrics config section of CW Agent config

https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-Agent-Configuration-File-Details.html#CloudWatch-Agent-Configuration-File-Metricssection

Returns:

Type	Description
dict[str, Any]	Dict[str, Any]: metrics config

Source code in src/aibs_informatics_cdk_lib/constructs_/batch/launch_template.py

def get_metrics_config(self) -> dict[str, Any]:
    """Generates metrics config section of CW Agent config

    https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-Agent-Configuration-File-Details.html#CloudWatch-Agent-Configuration-File-Metricssection

    Returns:
        Dict[str, Any]: metrics config
    """
    return {
        "namespace": self.metric_namespace,
        "metrics_collected": self._get_metrics_collected(),
        "append_dimensions": {
            "ImageId": "${aws:ImageId}",
            "InstanceId": "${aws:InstanceId}",
            "InstanceType": "${aws:InstanceType}",
            "AutoScalingGroupName": "${aws:AutoScalingGroupName}",
        },
        "aggregation_dimensions": [
            ["AutoScalingGroupName"],
            ["InstanceId", "InstanceType"],
            ["env_base", "batch_env_name"],
            [],
        ],
    }