8  More test driven development

In this chapter, we are going to expand on the service in ECS we have already defined in the previous chapter and make it a load-balanced service that can automatically scale out and scale in, based on load. We are going to use a test-driven design (TDD) approach to figure out what we want to do here.

8.1 A recap

First, a quick re-cap what we have done so far:

• We are using AWS CDK to define an infrastructure, using Python
• This infrastructure uses an existing VPC (Virtual Private Cloud - network infrastructure)
• We run a container-based solution using AWS Elastic Container Service (ECS)
• We have an ECS task definition that defines the container properties to use
• We have an ECS service definition that uses this task definition, plus a desired state description, so ECS will make sure that our solution matches that desired state.

We can expose the container publicly and connect to the web application that we have used so far to test the infrastructure building blocks.

So far, we have three code files we have worked with:

• my-container-infra.py - the main program file
• containers.py - where most of our container handling logic lives
• containers_test.py - The test code we have for our container management logic

The code (in Python) to this point is below here:

my-container-infra.py

import os
import aws_cdk as cdk
from aws_cdk import (
    aws_ec2 as ec2,
)
import containers

app = cdk.App()
env = cdk.Environment(
    account=os.getenv("CDK_DEFAULT_ACCOUNT"), region=os.getenv("CDK_DEFAULT_REGION")
)
stack = cdk.Stack(app, "my-container-infra", env=env)

vpc = ec2.Vpc.from_lookup(stack, "vpc", is_default=True)

cluster = containers.add_cluster(stack, "my-test-cluster", vpc)

taskconfig: containers.TaskConfig = {
    "cpu": 512,
    "memory_limit_mib": 1024,
    "family": "webapp",
}
containerconfig: containers.ContainerConfig = {
    "image": "public.ecr.aws/aws-containers/hello-app-runner:latest",
}
taskdef = containers.add_task_definition_with_container(
    stack, f"taskdef-{taskconfig['family']}", taskconfig, containerconfig
)

containers.add_service(
    stack, f"service-{taskconfig['family']}", cluster, taskdef, 8000, 0, True
)

app.synth()

containers.py

from typing import Literal, TypedDict  # noqa
import constructs as cons
from aws_cdk import (
    aws_ec2 as ec2,
    aws_ecs as ecs,
    aws_logs as logs,
)


class TaskConfig(TypedDict):
    cpu: Literal[256, 512, 1024, 2048, 4096]
    memory_limit_mib: int
    family: str


class ContainerConfig(TypedDict):
    image: str


def add_task_definition_with_container(
    scope: cons.Construct,
    id: str,
    task_config: TaskConfig,
    container_config: ContainerConfig,
) -> ecs.FargateTaskDefinition:
    taskdef = ecs.FargateTaskDefinition(
        scope,
        id,
        cpu=task_config["cpu"],
        memory_limit_mib=task_config["memory_limit_mib"],
        family=task_config["family"],
    )

    logdriver = ecs.LogDrivers.aws_logs(
        stream_prefix=taskdef.family,
        log_retention=logs.RetentionDays.ONE_DAY,
    )
    image = ecs.ContainerImage.from_registry(container_config["image"])
    image_id = f"container-{_extract_image_name(container_config['image'])}"
    taskdef.add_container(image_id, image=image, logging=logdriver)

    return taskdef


def add_service(
    scope: cons.Construct,
    id: str,
    cluster: ecs.Cluster,
    taskdef: ecs.FargateTaskDefinition,
    port: int,
    desired_count: int,
    assign_public_ip: bool = False,
    service_name: str = None,
) -> ecs.FargateService:
    name = service_name if service_name else ""
    sg = ec2.SecurityGroup(
        scope,
        f"{id}-security-group",
        description=f"security group for service {name}",
        vpc=cluster.vpc,
    )
    sg.add_ingress_rule(ec2.Peer.any_ipv4(), ec2.Port.tcp(port))

    service = ecs.FargateService(
        scope,
        id,
        cluster=cluster,
        task_definition=taskdef,
        desired_count=desired_count,
        service_name=service_name,
        security_groups=[sg],
        circuit_breaker=ecs.DeploymentCircuitBreaker(
            rollback=True,
        ),
        assign_public_ip=assign_public_ip,
    )
    return service


def add_cluster(scope: cons.Construct, id: str, vpc: ec2.IVpc) -> ecs.Cluster:
    return ecs.Cluster(scope, id, vpc=vpc)


def _extract_image_name(image_ref):
    name_with_tag = image_ref.split("/")[-1]
    name = name_with_tag.split(":")[0]
    return name

containers_test.py

import aws_cdk as cdk
from aws_cdk import (
    aws_ec2 as ec2,
    aws_ecs as ecs,
    assertions,
)
import containers


def test_ecs_cluster_defined_with_existing_vpc():
    stack = cdk.Stack()
    vpc = ec2.Vpc(stack, "vpc")
    cluster = containers.add_cluster(stack, "my-test-cluster", vpc=vpc)

    template = assertions.Template.from_stack(stack)
    template.resource_count_is("AWS::ECS::Cluster", 1)
    assert cluster.vpc is vpc


def test_ecs_fargate_task_definition_defined():
    stack = cdk.Stack()
    cpuval = 512
    memval = 1024
    familyval = "test"
    taskcfg: containers.TaskConfig = {
        "cpu": cpuval,
        "memory_limit_mib": memval,
        "family": familyval,
    }
    image = "public.ecr.aws/aws-containers/hello-app-runner:latest"
    containercfg: containers.ContainerConfig = {"image": image}
    taskdef = containers.add_task_definition_with_container(
        stack, f"taskdef-{taskcfg['family']}", taskcfg, containercfg
    )

    assert taskdef.is_fargate_compatible
    assert taskdef in stack.node.children

    template = assertions.Template.from_stack(stack)
    template.resource_count_is("AWS::ECS::TaskDefinition", 1)
    template.has_resource_properties(
        "AWS::ECS::TaskDefinition",
        {
            "RequiresCompatibilities": ["FARGATE"],
            "Cpu": str(cpuval),
            "Memory": str(memval),
            "Family": familyval,
        },
    )


def test_container_definition_added_to_task_definition():
    stack = cdk.Stack()
    cpuval = 512
    memval = 1024
    familyval = "test"
    taskcfg: containers.TaskConfig = {
        "cpu": cpuval,
        "memory_limit_mib": memval,
        "family": familyval,
    }
    image_name = "public.ecr.aws/aws-containers/hello-app-runner:latest"
    containercfg: containers.ContainerConfig = {"image": image_name}

    taskdef = containers.add_task_definition_with_container(
        stack, "test-taskdef", taskcfg, containercfg
    )

    template = assertions.Template.from_stack(stack)
    containerdef: ecs.ContainerDefinition = taskdef.default_container  # type: ignore

    assert containerdef is not None
    assert containerdef.image_name == image_name

    template.has_resource_properties(
        "AWS::ECS::TaskDefinition",
        {
            "ContainerDefinitions": assertions.Match.array_with(
                [assertions.Match.object_like({"Image": image_name})]
            )
        },
    )


def test_fargate_service_created_with_only_mandatory_properties():
    stack = cdk.Stack()
    vpc = ec2.Vpc(stack, "vpc")
    cluster = containers.add_cluster(stack, "test-cluster", vpc=vpc)
    cpuval = 512
    memval = 1024
    familyval = "test"
    taskcfg: containers.TaskConfig = {
        "cpu": cpuval,
        "memory_limit_mib": memval,
        "family": familyval,
    }
    image_name = "public.ecr.aws/aws-containers/hello-app-runner:latest"
    containercfg: containers.ContainerConfig = {"image": image_name}

    taskdef = containers.add_task_definition_with_container(
        stack, "test-taskdef", taskcfg, containercfg
    )

    port = 80
    desired_count = 1

    service = containers.add_service(
        stack, "test-service", cluster, taskdef, port, desired_count
    )

    sg_capture = assertions.Capture()
    template = assertions.Template.from_stack(stack)

    assert service.cluster == cluster
    assert service.task_definition == taskdef

    template.resource_count_is("AWS::ECS::Service", 1)
    template.has_resource_properties(
        "AWS::ECS::Service",
        {
            "DesiredCount": desired_count,
            "LaunchType": "FARGATE",
            "NetworkConfiguration": assertions.Match.object_like(
                {
                    "AwsvpcConfiguration": assertions.Match.object_like(
                        {
                            "AssignPublicIp": "DISABLED",
                            "SecurityGroups": assertions.Match.array_with([sg_capture]),
                        }
                    )
                }
            ),
        },
    )

    template.has_resource_properties(
        "AWS::EC2::SecurityGroup",
        {
            "SecurityGroupIngress": assertions.Match.array_with(
                [
                    assertions.Match.object_like(
                        {"CidrIp": "0.0.0.0/0", "FromPort": port, "IpProtocol": "tcp"}
                    )
                ]
            )
        },
    )

8.2 What we will do now

In the previous chapter, there were some points about our design that we should address, which include:

• Specifying a port number to access the service through add_service is right now fine for a single container instance only. For multiple containers (desiredCount > 1) there would need to be a load balancer.
• The design opens for traffic from everywhere, regardless of whether it uses public or private IP addresses

We should also add some mechanisms to scale the solution based on some load characteristics.

The points above tie in with each other. If we put the container service behind a load balancer, we want to restrict access so that it only goes via the load balancer. We also want to have a suitable number of containers running, based on the load on the service.

So we have a few tasks:

• Add a load balancer in front of a service, when we set up a service in ECS
• Make sure access is restricted to go via load balancer
• Add scalability configuration for containers behind load balancer

We can break these up in more detailed steps later.

8.3 The trouble with default VPCs

Until now, our infrastructure experiments have assumed that there is a VPC in place, and we have also for simplicity used the default VPC, which is something most AWS accounts will have.

A problem with the default VPC is that there are only public subnets in that VPC. The reason for that is backward compatibility with the old EC2 Classic, back in those days when you did not need an explicit VPC to run your virtual machines in AWS. This is my guess anyway.

Either way, it is a shortcoming of this VPC. To enable us to use a different VPC with both private and public subnets, we will add a small code segment to optionally add a VPC in the stack. You can also point to an existing VPC that has both public and private subnets.

vpcname = app.node.try_get_context("vpcname")
if vpcname:
    vpc = ec2.Vpc.from_lookup(stack, "vpc", vpc_name=vpcname)
else:
    vpc = ec2.Vpc(stack, "vpc", vpc_name="my-vpc", nat_gateways=1, max_azs=2)

Option one is to provide the name of an existing VPC, and in that case, we use that. If not VPC name is provided, we create a new VPC.

The created VPC will have both public and private subnets, will use two availability zones and have a shared NAT gateway for outbound traffic from private subnets. This is strictly to keep costs down.

If you have an existing VPC to use, you can provide the name of that VPC through a command line option:

cdk synth -c vpcname=thenameofmyvpc

This approach to sending contextual data to the AWS CDK App can be quite useful. The key takeaway for us here is that we can have a VPC with both private and public subnets.

8.4 Starting from the tests - TDD a load-balanced service

Let us start by looking at our tests. We currently have one test for the add_service function, see below. There is a fair amount of set-up work here.

def test_fargate_service_created_with_only_mandatory_properties():
    stack = cdk.Stack()
    vpc = ec2.Vpc(stack, "vpc")
    cluster = containers.add_cluster(stack, "test-cluster", vpc=vpc)
    cpuval = 512
    memval = 1024
    familyval = "test"
    taskcfg: containers.TaskConfig = {
        "cpu": cpuval,
        "memory_limit_mib": memval,
        "family": familyval,
    }
    image_name = "public.ecr.aws/aws-containers/hello-app-runner:latest"
    containercfg: containers.ContainerConfig = {"image": image_name}

    taskdef = containers.add_task_definition_with_container(
        stack, "test-taskdef", taskcfg, containercfg
    )

    port = 80
    desired_count = 1

    service = containers.add_service(
        stack, "test-service", cluster, taskdef, port, desired_count
    )

    sg_capture = assertions.Capture()
    template = assertions.Template.from_stack(stack)

    assert service.cluster == cluster
    assert service.task_definition == taskdef

    template.resource_count_is("AWS::ECS::Service", 1)
    template.has_resource_properties(
        "AWS::ECS::Service",
        {
            "DesiredCount": desired_count,
            "LaunchType": "FARGATE",
            "NetworkConfiguration": assertions.Match.object_like(
                {
                    "AwsvpcConfiguration": assertions.Match.object_like(
                        {
                            "AssignPublicIp": "DISABLED",
                            "SecurityGroups": assertions.Match.array_with([sg_capture]),
                        }
                    )
                }
            ),
        },
    )

    template.has_resource_properties(
        "AWS::EC2::SecurityGroup",
        {
            "SecurityGroupIngress": assertions.Match.array_with(
                [
                    assertions.Match.object_like(
                        {"CidrIp": "0.0.0.0/0", "FromPort": port, "IpProtocol": "tcp"}
                    )
                ]
            )
        },
    )

My suspicion at this point is that we may need multiple tests for the service part. Each test would have will have had the same setup work to do. With that in mind, I think we could refactor the service test into a group of tests that all run the same setup. Later, if we add more service tests, most of the set-up work will be in place already. Luckily, the pytest test framework we use with our AWS CDK Python code has support for that, using fixtures. The refactored code looks like this:

@pytest.fixture
def service_test_input_data():
    stack = cdk.Stack()
    vpc = ec2.Vpc(stack, "vpc")
    cluster = containers.add_cluster(stack, "test-cluster", vpc=vpc)
    cpuval = 512
    memval = 1024
    familyval = 'test'
    taskcfg: containers.TaskConfig = {
        "cpu": cpuval,
        "memory_limit_mib": memval,
        "family": familyval,
    }
    image_name = "public.ecr.aws/aws-containers/hello-app-runner:latest"
    containercfg: containers.ContainerConfig = {"image": image_name}

    taskdef = containers.add_task_definition_with_container(
        stack, "test-taskdef", taskcfg, containercfg
    )
    return { "stack": stack, "cluster": cluster, "task_definition": taskdef}


def test_fargate_service_created_with_only_mandatory_properties(service_test_input_data):
    stack = service_test_input_data["stack"]
    cluster = service_test_input_data["cluster"]
    taskdef = service_test_input_data["task_definition"]

    port = 80
    desired_count = 1

    service = containers.add_service(
        stack, "test-service", cluster, taskdef, port, desired_count
    )

    sg_capture = assertions.Capture()
    template = assertions.Template.from_stack(stack)

    assert service.cluster == cluster
    assert service.task_definition == taskdef

    template.resource_count_is("AWS::ECS::Service", 1)
    template.has_resource_properties(
        "AWS::ECS::Service",
        {
            "DesiredCount": desired_count,
            "LaunchType": "FARGATE",
            "NetworkConfiguration": assertions.Match.object_like(
                {
                    "AwsvpcConfiguration": assertions.Match.object_like(
                        {
                            "AssignPublicIp": "DISABLED",
                            "SecurityGroups": assertions.Match.array_with([sg_capture]),
                        }
                    )
                }
            ),
        },
    )

    template.has_resource_properties(
        "AWS::EC2::SecurityGroup",
        {
            "SecurityGroupIngress": assertions.Match.array_with(
                [
                    assertions.Match.object_like(
                        {"CidrIp": "0.0.0.0/0", "FromPort": port, "IpProtocol": "tcp"}
                    )
                ]
            )
        },
    )

When you change the code, run uv run pytest to make sure that the tests still work.

Next, let us think about what we want to do. We want to add a load balancer in front of our container service, so that we can scale it to a suitable amount and set meaningful desired count larger than 1.

So we are going to practice some test-driven development to update our add_service() function.

Since we should have a load balanced service, it would make sense to test that we get a load balancer in place. The underlying CloudFormation would generate a resource called AWS::ElasticLoadBalancerV2::LoadBalancer. So we can add a test that there will be such a resource after we have called our add_service function.

How do we implement this? We are trying to keep things simple, and fortunately AWS CDK has a construct that fits what we need well, called ApplicationLoadBalancedFargateService. This should set up a load balancer and create the service behind it, which is just what we want. We already have a task definition for the container we want to run as a service and this construct can accept that as input, which is just great!

Another change we can make to our function besides the name change is the optional parameter assign_public_ip. This is not relevant for a load balancer, since we will not have a specific IP address for the load balancer. But the underlying purpose of that parameter is that we should be able to tell if the endpoint we expose should be public (available through the internet) or private (only reachable in our own network). So we can also change the name of this parameter to better reflect our intent, instead of what it should do. So we can rename it to use_public_endpoint instead, for example.

So now we have a new test and some new code to support this, but if we run uv run pytest now, we get a failure! Our new construct will create security groups under the hood for us, and there will be security groups for the load balancer itself, as well as for the targets it will redirect traffic to. So our previous test for a single security group is no longer valid. We might not even care exactly how many service groups are created, as long as the functionality they are supposed to handle is in place (only allow specific network traffic to selected resources). We can also reasonably assume that those who have built ApplicationLoadBalancedFargateService have tested out that code well enough that we should not test their implementation. So we can skip our test for a single security group we had before.

So we adjust the test and we also adjust the underlying implementation of add_service, since we do not need to care about managing the security groups ourselves, at this point in time at least.

Now if we run uv run pytest, we see the test succeed! Is that good enough? Actually, we should also be clear what ports to map, so we should also include some port mapping information. The load balancer port and the container ports may not be the same. This may be suitable to include in the container configuration for the container itself.

Finally, let us also make sure we have a test for the use_public_endpoint parameter. Here, one relatively simple way to check is if the CloudFormation load balancer resource has its scheme set to internal if use_public_endpoint is false, and it should be internet-facing if use_public_endpoint is true.

Now we have a new set of code, see below:

containers_test.py

import pytest
import aws_cdk as cdk
from aws_cdk import (
    aws_ec2 as ec2,
    aws_ecs as ecs,
    assertions,
)
import containers


def test_ecs_cluster_defined_with_existing_vpc():
    stack = cdk.Stack()
    vpc = ec2.Vpc(stack, "vpc")
    cluster = containers.add_cluster(stack, "my-test-cluster", vpc=vpc)

    template = assertions.Template.from_stack(stack)
    template.resource_count_is("AWS::ECS::Cluster", 1)
    assert cluster.vpc is vpc


def test_ecs_fargate_task_definition_defined():
    stack = cdk.Stack()
    cpuval = 512
    memval = 1024
    familyval = "test"
    taskcfg: containers.TaskConfig = {
        "cpu": cpuval,
        "memory_limit_mib": memval,
        "family": familyval,
    }
    image = "public.ecr.aws/aws-containers/hello-app-runner:latest"
    containercfg: containers.ContainerConfig = {"image": image, "tcp_ports": [8080]}
    taskdef = containers.add_task_definition_with_container(
        stack, f"taskdef-{taskcfg['family']}", taskcfg, containercfg
    )

    assert taskdef.is_fargate_compatible
    assert taskdef in stack.node.children

    template = assertions.Template.from_stack(stack)
    template.resource_count_is("AWS::ECS::TaskDefinition", 1)
    template.has_resource_properties(
        "AWS::ECS::TaskDefinition",
        {
            "RequiresCompatibilities": ["FARGATE"],
            "Cpu": str(cpuval),
            "Memory": str(memval),
            "Family": familyval,
        },
    )


def test_container_definition_added_to_task_definition():
    stack = cdk.Stack()
    cpuval = 512
    memval = 1024
    familyval = "test"
    taskcfg: containers.TaskConfig = {
        "cpu": cpuval,
        "memory_limit_mib": memval,
        "family": familyval,
    }
    image_name = "public.ecr.aws/aws-containers/hello-app-runner:latest"
    containercfg: containers.ContainerConfig = {"image": image_name, "tcp_ports": [8080]}

    taskdef = containers.add_task_definition_with_container(
        stack, "test-taskdef", taskcfg, containercfg
    )

    template = assertions.Template.from_stack(stack)
    containerdef: ecs.ContainerDefinition = taskdef.default_container  # type: ignore

    assert containerdef is not None
    assert containerdef.image_name == image_name

    template.has_resource_properties(
        "AWS::ECS::TaskDefinition",
        {
            "ContainerDefinitions": assertions.Match.array_with(
                [assertions.Match.object_like({"Image": image_name})]
            )
        },
    )

@pytest.fixture
def service_test_input_data():
    stack = cdk.Stack()
    vpc = ec2.Vpc(stack, "vpc")
    cluster = containers.add_cluster(stack, "test-cluster", vpc=vpc)
    cpuval = 512
    memval = 1024
    familyval = 'test'
    taskcfg: containers.TaskConfig = {
        "cpu": cpuval,
        "memory_limit_mib": memval,
        "family": familyval,
    }
    image_name = "public.ecr.aws/aws-containers/hello-app-runner:latest"
    containercfg: containers.ContainerConfig = {"image": image_name, "tcp_ports": [8080]}

    taskdef = containers.add_task_definition_with_container(
        stack, "test-taskdef", taskcfg, containercfg
    )
    return { "stack": stack, "cluster": cluster, "task_definition": taskdef}


def test_fargate_service_created_with_only_mandatory_properties(service_test_input_data):
    stack = service_test_input_data["stack"]
    cluster = service_test_input_data["cluster"]
    taskdef = service_test_input_data["task_definition"]

    port = 80
    desired_count = 1

    service = containers.add_service(
        stack, "test-service", cluster, taskdef, port, desired_count
    )

    sg_capture = assertions.Capture()
    template = assertions.Template.from_stack(stack)

    assert service.cluster == cluster
    assert service.task_definition == taskdef

    template.resource_count_is("AWS::ECS::Service", 1)
    template.has_resource_properties(
        "AWS::ECS::Service",
        {
            "DesiredCount": desired_count,
            "LaunchType": "FARGATE",
            "NetworkConfiguration": assertions.Match.object_like(
                {
                    "AwsvpcConfiguration": assertions.Match.object_like(
                        {
                            "AssignPublicIp": "DISABLED",
                            "SecurityGroups": assertions.Match.array_with([sg_capture]),
                        }
                    )
                }
            ),
        },
    )

    template.resource_count_is('AWS::ElasticLoadBalancingV2::LoadBalancer', 1)
    template.has_resource_properties('AWS::ElasticLoadBalancingV2::LoadBalancer', {
        'Type': 'application',
        'Scheme': 'internet-facing'
    })

    template.has_resource_properties(
        "AWS::EC2::SecurityGroup",
        {
            "SecurityGroupIngress": assertions.Match.array_with(
                [
                    assertions.Match.object_like(
                        {"CidrIp": "0.0.0.0/0", "FromPort": port, "IpProtocol": "tcp"}
                    )
                ]
            )
        },
    )

def test_fargate_service_created_without_public_access(service_test_input_data):
    stack = service_test_input_data["stack"]
    cluster = service_test_input_data["cluster"]
    taskdef = service_test_input_data["task_definition"]

    port = 80
    desired_count = 1
    containers.add_service(stack, 'test-service', cluster, taskdef, port, desired_count, False)

    template = assertions.Template.from_stack(stack)
    template.resource_count_is('AWS::ElasticLoadBalancingV2::LoadBalancer', 1)
    template.has_resource_properties('AWS::ElasticLoadBalancingV2::LoadBalancer', {
        'Type': 'application',
        'Scheme': 'internal'
    })

containers.py

from typing import Literal, TypedDict, List  # noqa
import constructs as cons
from aws_cdk import (
    aws_ec2 as ec2,
    aws_ecs as ecs,
    aws_ecs_patterns as ecspat,
    aws_logs as logs,
)


class TaskConfig(TypedDict):
    cpu: Literal[256, 512, 1024, 2048, 4096]
    memory_limit_mib: int
    family: str


class ContainerConfig(TypedDict):
    image: str
    tcp_ports: List[int]


def add_task_definition_with_container(
    scope: cons.Construct,
    id: str,
    task_config: TaskConfig,
    container_config: ContainerConfig,
) -> ecs.FargateTaskDefinition:
    taskdef = ecs.FargateTaskDefinition(
        scope,
        id,
        cpu=task_config["cpu"],
        memory_limit_mib=task_config["memory_limit_mib"],
        family=task_config["family"],
    )

    logdriver = ecs.LogDrivers.aws_logs(
        stream_prefix=taskdef.family,
        log_retention=logs.RetentionDays.ONE_DAY,
    )
    image = ecs.ContainerImage.from_registry(container_config["image"])
    image_id = f"container-{_extract_image_name(container_config['image'])}"
    containerdef = taskdef.add_container(image_id, image=image, logging=logdriver)

    for port in container_config["tcp_ports"]:
        containerdef.add_port_mappings(ecs.PortMapping(container_port=port, protocol=ecs.Protocol.TCP))

    return taskdef


def add_service(
    scope: cons.Construct,
    id: str,
    cluster: ecs.Cluster,
    taskdef: ecs.FargateTaskDefinition,
    port: int,
    desired_count: int,
    use_public_endpoint: bool = True,
    service_name: str | None = None,
) -> ecspat.ApplicationLoadBalancedFargateService:
    service = ecspat.ApplicationLoadBalancedFargateService(
        scope,
        id,
        cluster=cluster,
        task_definition=taskdef,
        listener_port=port,
        desired_count=desired_count,
        service_name=service_name,
        circuit_breaker=ecs.DeploymentCircuitBreaker(
            rollback=True,
        ),
        public_load_balancer=use_public_endpoint,
    )
    return service


def add_cluster(scope: cons.Construct, id: str, vpc: ec2.IVpc) -> ecs.Cluster:
    return ecs.Cluster(scope, id, vpc=vpc)


def _extract_image_name(image_ref):
    name_with_tag = image_ref.split("/")[-1]
    name = name_with_tag.split(":")[0]
    return name

my-container-infra.py

import os
import aws_cdk as cdk
from aws_cdk import (
    aws_ec2 as ec2,
)
import containers

app = cdk.App()
env = cdk.Environment(
    account=os.getenv("CDK_DEFAULT_ACCOUNT"), region=os.getenv("CDK_DEFAULT_REGION")
)
stack = cdk.Stack(app, "my-container-infra", env=env)

vpcname = app.node.try_get_context("vpcname")
if vpcname:
    vpc = ec2.Vpc.from_lookup(stack, "vpc", vpc_name=vpcname)
else:
    vpc = ec2.Vpc(stack, "vpc", vpc_name="my-vpc", nat_gateways=1, max_azs=2)

cluster = containers.add_cluster(stack, "my-test-cluster", vpc)

taskconfig: containers.TaskConfig = {
    "cpu": 512,
    "memory_limit_mib": 1024,
    "family": "webapp",
}
containerconfig: containers.ContainerConfig = {
    "image": "public.ecr.aws/aws-containers/hello-app-runner:latest",
    "tcp_ports": [8000],
}
taskdef = containers.add_task_definition_with_container(
    stack, f"taskdef-{taskconfig['family']}", taskconfig, containerconfig
)

containers.add_service(
    stack, f"service-{taskconfig['family']}", cluster, taskdef, 8000, 2, True
)

app.synth()

Run npm test and make sure all the tests pass. You can also to deploy the solution now, using cdk deploy. Expect the deployment to take some time, especially if you deploy everything from scratch and if you have it set up to create a new VPC.

You notice that when the deployment is completed, there will be two outputs representing the load balancer endpoint. This results from using ApplicationLoadBalancedFargateService, which adds these outputs by default to the stack.

8.5 A bit of auto-scaling

If we have a load balancer in front of our service to distribute the load among multiple containers, it would make sense to scale up and down how many containers will handle that load. Fortunately, after some investigation, we can see that the FargateService class has a function called autoScaleTaskCount, which allows us to set the minimum and maximum number of tasks. It also allows us to specify thresholds for CPU and memory to determine if the service should scale down or up the number of tasks running in the service.

There is no corresponding function on ApplicationLoadBalancedFargateService though, but we can access the underlying FargateService using a service property on ApplicationLoadBalancedFargateService.

Unfortunately, we cannot access all these scaling settings on FargateService once we have set them, so to verify that these settings have been set, we would resort to check CloudFormation resources. In this case, it is resources of the ApplicationAutoScaling that is used by CloudFormation.

Frankly, I am hesitant about the value of some of these tests, since this may be borderline to test the CDK implementation rather than testing our own logic. I kept them here more to be a learning experience. Even with these tests, we would want to do some actual integration tests to see that the real behaviour once deployed is what we expect. There is some experimental work in AWS CDK for building integration tests, but that is out of scope for this article.

For now, we can at least know that there are some settings in place for the auto-scaling, but more work would be needed to test its actual behaviour.

Extract from containers_test.py

def test_scaling_settings_for_service(service_test_input_data):
    stack = service_test_input_data['stack']
    cluster = service_test_input_data['cluster']
    taskdef = service_test_input_data['task_definition']
    port = 80
    desired_count = 2

    service = containers.add_service(stack, 'test-service', cluster, taskdef, port, desired_count, False)

    config = containers.ServiceScalingConfig(
        min_count=1,
        max_count=5,
        scale_cpu_target=containers.ScalingThreshold(percent=50),
        scale_memory_target=containers.ScalingThreshold(percent=50))
    containers.set_service_scaling(service=service.service, config=config)

    scale_resource = assertions.Capture()
    template = assertions.Template.from_stack(stack)
    template.resource_count_is('AWS::ApplicationAutoScaling::ScalableTarget', 1)
    template.has_resource_properties('AWS::ApplicationAutoScaling::ScalableTarget', {
        'MaxCapacity': config["max_count"],
        'MinCapacity': config["min_count"],
        'ResourceId': scale_resource,
        'ScalableDimension': 'ecs:service:DesiredCount',
        'ServiceNamespace': 'ecs'
    })

    template.resource_count_is('AWS::ApplicationAutoScaling::ScalingPolicy', 2)
    template.has_resource_properties('AWS::ApplicationAutoScaling::ScalingPolicy', {
        'PolicyType': 'TargetTrackingScaling',
        'TargetTrackingScalingPolicyConfiguration': assertions.Match.object_like({
            'PredefinedMetricSpecification': assertions.Match.object_equals({
                'PredefinedMetricType': 'ECSServiceAverageCPUUtilization'
            }),
            'TargetValue': config["scale_cpu_target"]["percent"]
        })
    })
    template.has_resource_properties('AWS::ApplicationAutoScaling::ScalingPolicy', {
        'PolicyType': 'TargetTrackingScaling',
        'TargetTrackingScalingPolicyConfiguration': assertions.Match.object_like({
            'PredefinedMetricSpecification': assertions.Match.object_equals({
                'PredefinedMetricType': 'ECSServiceAverageMemoryUtilization'
            }),
            'TargetValue': config["scale_memory_target"]["percent"]
        })
    })

Extract from containers.py

class ScalingThreshold(TypedDict):
    percent: float

class ServiceScalingConfig(TypedDict):
    min_count: int
    max_count: int
    scale_cpu_target: ScalingThreshold
    scale_memory_target: ScalingThreshold

def set_service_scaling(service: ecs.FargateService, config: ServiceScalingConfig):
    scaling = service.auto_scale_task_count(max_capacity=config["max_count"], min_capacity=config["min_count"])
    scaling.scale_on_cpu_utilization('CpuScaling', target_utilization_percent=config["scale_cpu_target"]["percent"])
    scaling.scale_on_memory_utilization('MemoryScaling', target_utilization_percent=config["scale_memory_target"]["percent"])

Extract from my-container-infra.py

class ScalingThreshold(TypedDict):
    percent: float

class ServiceScalingConfig(TypedDict):
    min_count: int
    max_count: int
    scale_cpu_target: ScalingThreshold
    scale_memory_target: ScalingThreshold

def set_service_scaling(service: ecs.FargateService, config: ServiceScalingConfig):
    scaling = service.auto_scale_task_count(max_capacity=config["max_count"], min_capacity=config["min_count"])
    scaling.scale_on_cpu_utilization('CpuScaling', target_utilization_percent=config["scale_cpu_target"]["percent"])
    scaling.scale_on_memory_utilization('MemoryScaling', target_utilization_percent=config["scale_memory_target"]["percent"])

You can run uv run pytest to make sure all tests pass, and run cdk deploy to see that it also deploys ok.

Congratulations, you have now set up something that is fairly similar to what we initially set up with AppRunner!

As you can see, this was definitely more details to handle, even though we had a fair amount of help from AWS CDK itself also.