Chapter 7: Deployment Strategies and Environment Management

Knowledge Summary

Environment Management Overview

In enterprise applications, it’s typically necessary to manage multiple environments: development (dev), testing (test), staging, and production (prod). CDK provides flexible mechanisms to manage differences across these environments.

graph TD
    A[Source Code] --> B[Build Stage]
    B --> C{Environment Selection}

    C --> D[Development Environment]
    C --> E[Test Environment]
    C --> F[Staging Environment]
    C --> G[Production Environment]

    D --> D1[dev-stack]
    E --> E1[test-stack]
    F --> F1[staging-stack]
    G --> G1[prod-stack]

    D1 --> D2[Basic Resources]
    E1 --> E2[Full Feature Testing]
    F1 --> F2[Production-Grade Configuration]
    G1 --> G2[High Availability Deployment]

    style D fill:#e8f5e8
    style E fill:#fff2cc
    style F fill:#fce4ec
    style G fill:#ffebee

Environment-Specific Configuration

# config/environments.py
from dataclasses import dataclass
from typing import Dict, List, Optional
import aws_cdk as cdk

@dataclass
class EnvironmentConfig:
    """Environment configuration data class"""
    name: str
    account: str
    region: str

    # Infrastructure configuration
    instance_count: int
    instance_type: str
    database_instance_type: str

    # Lambda configuration
    lambda_memory: int
    lambda_timeout: int

    # Network configuration
    vpc_cidr: str
    enable_nat_gateway: bool

    # Monitoring and logging
    enable_detailed_monitoring: bool
    log_retention_days: int

    # Security configuration
    enable_encryption: bool
    backup_retention_days: int

    # Tags
    tags: Dict[str, str]

class EnvironmentFactory:
    """Environment configuration factory"""

    @staticmethod
    def get_config(env_name: str) -> EnvironmentConfig:
        configs = {
            'dev': EnvironmentConfig(
                name='dev',
                account='123456789012',
                region='us-east-1',
                instance_count=1,
                instance_type='t3.micro',
                database_instance_type='db.t3.micro',
                lambda_memory=128,
                lambda_timeout=30,
                vpc_cidr='10.0.0.0/16',
                enable_nat_gateway=False,
                enable_detailed_monitoring=False,
                log_retention_days=7,
                enable_encryption=False,
                backup_retention_days=7,
                tags={
                    'Environment': 'dev',
                    'Project': 'web-app',
                    'Owner': 'dev-team'
                }
            ),
            'staging': EnvironmentConfig(
                name='staging',
                account='123456789012',
                region='us-east-1',
                instance_count=2,
                instance_type='t3.small',
                database_instance_type='db.t3.small',
                lambda_memory=256,
                lambda_timeout=60,
                vpc_cidr='10.1.0.0/16',
                enable_nat_gateway=True,
                enable_detailed_monitoring=True,
                log_retention_days=30,
                enable_encryption=True,
                backup_retention_days=14,
                tags={
                    'Environment': 'staging',
                    'Project': 'web-app',
                    'Owner': 'dev-team'
                }
            ),
            'prod': EnvironmentConfig(
                name='prod',
                account='987654321098',
                region='us-east-1',
                instance_count=3,
                instance_type='t3.medium',
                database_instance_type='db.r5.large',
                lambda_memory=512,
                lambda_timeout=120,
                vpc_cidr='10.2.0.0/16',
                enable_nat_gateway=True,
                enable_detailed_monitoring=True,
                log_retention_days=365,
                enable_encryption=True,
                backup_retention_days=30,
                tags={
                    'Environment': 'prod',
                    'Project': 'web-app',
                    'Owner': 'ops-team',
                    'CostCenter': 'engineering'
                }
            )
        }

        if env_name not in configs:
            raise ValueError(f"Unknown environment: {env_name}")

        return configs[env_name]

Parameterized Stack Design

Configuration-Driven Stack

# stacks/configurable_web_stack.py
import aws_cdk as cdk
from aws_cdk import (
    aws_ec2 as ec2,
    aws_rds as rds,
    aws_lambda as lambda_,
    aws_apigateway as apigateway,
    aws_s3 as s3,
    aws_cloudwatch as cloudwatch
)
from constructs import Construct
from config.environments import EnvironmentConfig

class ConfigurableWebStack(cdk.Stack):
    """Configurable Web Application Stack"""

    def __init__(self, scope: Construct, construct_id: str,
                 config: EnvironmentConfig, **kwargs) -> None:

        # Set environment
        env = cdk.Environment(
            account=config.account,
            region=config.region
        )
        super().__init__(scope, construct_id, env=env, **kwargs)

        self.config = config

        # Apply tags
        for key, value in config.tags.items():
            cdk.Tags.of(self).add(key, value)

        # Create VPC
        self.vpc = self._create_vpc()

        # Create database
        self.database = self._create_database()

        # Create Lambda function
        self.lambda_function = self._create_lambda()

        # Create API Gateway
        self.api = self._create_api_gateway()

        # Create S3 bucket
        self.bucket = self._create_s3_bucket()

        # Create monitoring
        if config.enable_detailed_monitoring:
            self._create_monitoring()

    def _create_vpc(self) -> ec2.Vpc:
        """Create VPC"""
        return ec2.Vpc(
            self,
            "VPC",
            ip_addresses=ec2.IpAddresses.cidr(self.config.vpc_cidr),
            max_azs=2,
            nat_gateways=1 if self.config.enable_nat_gateway else 0,
            subnet_configuration=[
                ec2.SubnetConfiguration(
                    cidr_mask=24,
                    name="Public",
                    subnet_type=ec2.SubnetType.PUBLIC
                ),
                ec2.SubnetConfiguration(
                    cidr_mask=24,
                    name="Private",
                    subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS if self.config.enable_nat_gateway
                              else ec2.SubnetType.PRIVATE_ISOLATED
                )
            ]
        )

    def _create_database(self) -> rds.DatabaseInstance:
        """Create RDS database"""
        # Database security group
        db_security_group = ec2.SecurityGroup(
            self,
            "DatabaseSecurityGroup",
            vpc=self.vpc,
            description="Security group for RDS database",
            allow_all_outbound=False
        )

        # Create database subnet group
        subnet_group = rds.SubnetGroup(
            self,
            "DatabaseSubnetGroup",
            description="Subnet group for RDS database",
            vpc=self.vpc,
            vpc_subnets=ec2.SubnetSelection(
                subnet_type=ec2.SubnetType.PRIVATE_ISOLATED
            )
        )

        return rds.DatabaseInstance(
            self,
            "Database",
            engine=rds.DatabaseInstanceEngine.postgres(
                version=rds.PostgresEngineVersion.VER_13_7
            ),
            instance_type=ec2.InstanceType(self.config.database_instance_type),
            vpc=self.vpc,
            subnet_group=subnet_group,
            security_groups=[db_security_group],
            database_name="webapp",
            credentials=rds.Credentials.from_generated_secret("dbuser"),
            storage_encrypted=self.config.enable_encryption,
            monitoring_interval=cdk.Duration.seconds(60) if self.config.enable_detailed_monitoring else None,
            backup_retention=cdk.Duration.days(self.config.backup_retention_days),
            deletion_protection=self.config.name == 'prod'
        )

    def _create_lambda(self) -> lambda_.Function:
        """Create Lambda function"""
        # Lambda security group
        lambda_security_group = ec2.SecurityGroup(
            self,
            "LambdaSecurityGroup",
            vpc=self.vpc,
            description="Security group for Lambda function"
        )

        # Allow Lambda to access database
        self.database.connections.allow_default_port_from(lambda_security_group)

        return lambda_.Function(
            self,
            "WebFunction",
            runtime=lambda_.Runtime.PYTHON_3_9,
            handler="app.handler",
            code=lambda_.Code.from_asset("lambda"),
            vpc=self.vpc,
            security_groups=[lambda_security_group],
            memory_size=self.config.lambda_memory,
            timeout=cdk.Duration.seconds(self.config.lambda_timeout),
            environment={
                "DB_HOST": self.database.instance_endpoint.hostname,
                "DB_PORT": self.database.instance_endpoint.port,
                "DB_NAME": "webapp",
                "ENVIRONMENT": self.config.name
            },
            log_retention=getattr(cdk.aws_logs.RetentionDays, f"_{self.config.log_retention_days}_DAYS")
        )

    def _create_api_gateway(self) -> apigateway.RestApi:
        """Create API Gateway"""
        api = apigateway.RestApi(
            self,
            "WebAPI",
            rest_api_name=f"{self.config.name}-web-api",
            description=f"Web API for {self.config.name} environment",
            deploy_options=apigateway.StageOptions(
                stage_name=self.config.name,
                throttling_rate_limit=100 if self.config.name != 'prod' else 1000,
                throttling_burst_limit=200 if self.config.name != 'prod' else 2000,
                logging_level=apigateway.MethodLoggingLevel.INFO if self.config.enable_detailed_monitoring
                             else apigateway.MethodLoggingLevel.ERROR
            )
        )

        # Create Lambda integration
        lambda_integration = apigateway.LambdaIntegration(
            self.lambda_function,
            request_templates={"application/json": '{ "statusCode": "200" }'}
        )

        # Add resources and methods
        api.root.add_method("ANY", lambda_integration)
        api.root.add_proxy(default_integration=lambda_integration)

        return api

    def _create_s3_bucket(self) -> s3.Bucket:
        """Create S3 bucket"""
        return s3.Bucket(
            self,
            "WebAssets",
            bucket_name=f"{self.config.name}-web-assets-{self.account}",
            versioned=True,
            encryption=s3.BucketEncryption.S3_MANAGED if self.config.enable_encryption
                      else s3.BucketEncryption.UNENCRYPTED,
            lifecycle_rules=[
                s3.LifecycleRule(
                    id="DeleteOldVersions",
                    enabled=True,
                    noncurrent_version_expiration=cdk.Duration.days(30)
                )
            ],
            removal_policy=cdk.RemovalPolicy.DESTROY if self.config.name != 'prod'
                          else cdk.RemovalPolicy.RETAIN
        )

    def _create_monitoring(self):
        """Create monitoring resources"""
        # Create CloudWatch Dashboard
        dashboard = cloudwatch.Dashboard(
            self,
            "WebAppDashboard",
            dashboard_name=f"{self.config.name}-webapp-dashboard"
        )

        # Lambda monitoring
        dashboard.add_widgets(
            cloudwatch.GraphWidget(
                title="Lambda Invocations",
                left=[self.lambda_function.metric_invocations()],
                right=[self.lambda_function.metric_errors()]
            ),
            cloudwatch.GraphWidget(
                title="Lambda Duration",
                left=[self.lambda_function.metric_duration()]
            )
        )

        # API Gateway monitoring
        dashboard.add_widgets(
            cloudwatch.GraphWidget(
                title="API Gateway Requests",
                left=[self.api.metric_count()],
                right=[self.api.metric_client_error(), self.api.metric_server_error()]
            )
        )

        # Database monitoring
        dashboard.add_widgets(
            cloudwatch.GraphWidget(
                title="Database Connections",
                left=[self.database.metric_database_connections()]
            )
        )

Application Entry Point

# app.py
#!/usr/bin/env python3
import os
import aws_cdk as cdk
from stacks.configurable_web_stack import ConfigurableWebStack
from config.environments import EnvironmentFactory

def main():
    """Main function"""
    app = cdk.App()

    # Get environment name from environment variable or context
    env_name = app.node.try_get_context("environment") or os.environ.get("ENVIRONMENT", "dev")

    # Get environment configuration
    config = EnvironmentFactory.get_config(env_name)

    # Create Stack
    ConfigurableWebStack(
        app,
        f"WebApp-{config.name}",
        config=config,
        description=f"Web application stack for {config.name} environment"
    )

    app.synth()

if __name__ == "__main__":
    main()

Implementing CI/CD with CDK Pipelines

Pipeline Stack

# stacks/pipeline_stack.py
import aws_cdk as cdk
from aws_cdk import (
    pipelines,
    aws_codecommit as codecommit,
    aws_codebuild as codebuild,
    aws_iam as iam
)
from constructs import Construct
from .configurable_web_stack import ConfigurableWebStack
from config.environments import EnvironmentFactory

class WebAppPipelineStack(cdk.Stack):
    """Web Application CI/CD Pipeline"""

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Create code repository
        repo = codecommit.Repository(
            self,
            "WebAppRepo",
            repository_name="web-app-cdk",
            description="Web application CDK code repository"
        )

        # Create Pipeline
        pipeline = pipelines.CodePipeline(
            self,
            "WebAppPipeline",
            pipeline_name="web-app-pipeline",
            synth=pipelines.ShellStep(
                "Synth",
                input=pipelines.CodePipelineSource.code_commit(repo, "main"),
                commands=[
                    "npm install -g aws-cdk",
                    "pip install -r requirements.txt",
                    "cdk synth"
                ]
            ),
            code_build_defaults=pipelines.CodeBuildOptions(
                build_environment=codebuild.BuildEnvironment(
                    build_image=codebuild.LinuxBuildImage.STANDARD_5_0,
                    compute_type=codebuild.ComputeType.SMALL
                ),
                role_policy=[
                    iam.PolicyStatement(
                        effect=iam.Effect.ALLOW,
                        actions=[
                            "ssm:GetParameter",
                            "ssm:GetParameters",
                            "kms:Decrypt"
                        ],
                        resources=["*"]
                    )
                ]
            )
        )

        # Development environment stage
        dev_stage = WebAppStage(self, "Dev", env_name="dev")
        dev_deployment = pipeline.add_stage(dev_stage)

        # Add development environment tests
        dev_deployment.add_post([
            pipelines.ShellStep(
                "DevIntegrationTests",
                commands=[
                    "pip install -r requirements-test.txt",
                    "pytest tests/integration/ -v"
                ],
                env_from_cfn_outputs={
                    "API_URL": dev_stage.api_url
                }
            )
        ])

        # Staging environment stage
        staging_stage = WebAppStage(self, "Staging", env_name="staging")
        staging_deployment = pipeline.add_stage(staging_stage)

        # Add staging environment tests
        staging_deployment.add_post([
            pipelines.ShellStep(
                "StagingLoadTests",
                commands=[
                    "pip install locust",
                    "locust --headless --users 10 --spawn-rate 2 --run-time 1m --host $API_URL"
                ],
                env_from_cfn_outputs={
                    "API_URL": staging_stage.api_url
                }
            )
        ])

        # Production environment stage (requires manual approval)
        prod_stage = WebAppStage(self, "Prod", env_name="prod")
        prod_deployment = pipeline.add_stage(
            prod_stage,
            pre=[
                pipelines.ManualApprovalStep("PromoteToProd")
            ]
        )

        # Post-deployment verification for production
        prod_deployment.add_post([
            pipelines.ShellStep(
                "ProdSmokeTests",
                commands=[
                    "curl -f $API_URL/health || exit 1",
                    "echo 'Production deployment successful'"
                ],
                env_from_cfn_outputs={
                    "API_URL": prod_stage.api_url
                }
            )
        ])

class WebAppStage(cdk.Stage):
    """Web Application Deployment Stage"""

    def __init__(self, scope: Construct, construct_id: str,
                 env_name: str, **kwargs) -> None:

        config = EnvironmentFactory.get_config(env_name)
        env = cdk.Environment(account=config.account, region=config.region)

        super().__init__(scope, construct_id, env=env, **kwargs)

        # Create Web App Stack
        web_stack = ConfigurableWebStack(
            self,
            f"WebApp-{env_name}",
            config=config
        )

        # Output API URL for subsequent stages
        self.api_url = cdk.CfnOutput(
            web_stack,
            "ApiUrl",
            value=web_stack.api.url,
            export_name=f"WebApp-{env_name}-ApiUrl"
        )

Deployment Configuration

# cdk.json
{
  "app": "python app.py",
  "watch": {
    "include": [
      "**"
    ],
    "exclude": [
      "README.md",
      "cdk*.json",
      "requirements*.txt",
      "source.bat",
      "**/__pycache__",
      "**/*.pyc"
    ]
  },
  "context": {
    "@aws-cdk/aws-lambda:recognizeLayerVersion": true,
    "@aws-cdk/core:checkSecretUsage": true,
    "@aws-cdk/core:target": "aws-cdk-lib",
    "@aws-cdk-containers/ecs-service-extensions:enableDefaultLogDriver": true,
    "@aws-cdk/aws-ec2:uniqueImdsv2TemplateName": true,
    "@aws-cdk/aws-ecs:arnFormatIncludesClusterName": true,
    "@aws-cdk/aws-iam:minimizePolicies": true,
    "@aws-cdk/core:validateSnapshotRemovalPolicy": true,
    "@aws-cdk/aws-codepipeline:crossAccountKeyAliasStackSafeResourceName": true,
    "@aws-cdk/aws-s3:createDefaultLoggingPolicy": true,
    "@aws-cdk/aws-sns-subscriptions:restrictSqsDescryption": true,
    "@aws-cdk/aws-apigateway:disableCloudWatchRole": true,
    "@aws-cdk/core:enablePartitionLiterals": true,
    "@aws-cdk/aws-events:eventsTargetQueueSameAccount": true,
    "@aws-cdk/aws-iam:standardizedServicePrincipals": true,
    "@aws-cdk/aws-ecs:disableExplicitDeploymentControllerForCircuitBreaker": true,
    "@aws-cdk/aws-iam:importedRoleStackSafeDefaultPolicyName": true,
    "@aws-cdk/aws-s3:serverAccessLogsUseBucketPolicy": true,
    "@aws-cdk/aws-route53-patters:useCertificate": true,
    "@aws-cdk/customresources:installLatestAwsSdkDefault": false,
    "@aws-cdk/aws-rds:databaseProxyUniqueResourceName": true,
    "@aws-cdk/aws-codedeploy:removeAlarmsFromDeploymentGroup": true,
    "@aws-cdk/aws-apigateway:authorizerChangeDeploymentLogicalId": true,
    "@aws-cdk/aws-ec2:launchTemplateDefaultUserData": true,
    "@aws-cdk/aws-secretsmanager:useAttachedSecretResourcePolicyForSecretTargetAttachments": true,
    "@aws-cdk/aws-redshift:columnId": true,
    "@aws-cdk/aws-stepfunctions-tasks:enableLoggingForLambdaInvoke": true,
    "@aws-cdk/aws-ec2:restrictDefaultSecurityGroup": true,
    "@aws-cdk/aws-apigateway:requestValidatorUniqueId": true,
    "@aws-cdk/aws-kms:aliasNameRef": true,
    "@aws-cdk/aws-autoscaling:generateLaunchTemplateInsteadOfLaunchConfig": true,
    "@aws-cdk/core:includePrefixInUniqueNameGeneration": true,
    "@aws-cdk/aws-efs:denyAnonymousAccess": true,
    "@aws-cdk/aws-opensearchservice:enableOpensearchMultiAzWithStandby": true,
    "@aws-cdk/aws-lambda-nodejs:useLatestRuntimeVersion": true,
    "@aws-cdk/aws-efs:mountTargetOrderInsensitiveLogicalId": true,
    "@aws-cdk/aws-rds:auroraClusterChangeScopeOfInstanceParameterGroupWithEachParameters": true,
    "@aws-cdk/aws-appsync:useArnForSourceApiAssociationIdentifier": true,
    "@aws-cdk/aws-rds:preventRenderingDeprecatedCredentials": true,
    "@aws-cdk/aws-codepipeline-actions:betaFeatures": true,
    "@aws-cdk/aws-redshift:createClusterSubnetGroupDefaultRolePermsToPublicSchema": true
  }
}

Advanced Deployment Strategies

Blue-Green Deployment

# stacks/blue_green_deployment.py
import aws_cdk as cdk
from aws_cdk import (
    aws_lambda as lambda_,
    aws_codedeploy as codedeploy,
    aws_apigateway as apigateway,
    aws_cloudwatch as cloudwatch,
    aws_iam as iam
)
from constructs import Construct

class BlueGreenDeploymentStack(cdk.Stack):
    """Blue-Green Deployment Stack"""

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Lambda function
        self.lambda_function = lambda_.Function(
            self,
            "WebFunction",
            runtime=lambda_.Runtime.PYTHON_3_9,
            handler="app.handler",
            code=lambda_.Code.from_asset("lambda"),
            memory_size=256,
            timeout=cdk.Duration.seconds(30)
        )

        # Lambda alias
        live_alias = lambda_.Alias(
            self,
            "LiveAlias",
            alias_name="live",
            version=self.lambda_function.current_version
        )

        # CodeDeploy application
        application = codedeploy.LambdaApplication(
            self,
            "DeploymentApplication",
            application_name="web-app-deployment"
        )

        # Deployment configuration
        deployment_config = codedeploy.LambdaDeploymentConfig.ALL_AT_ONCE_CODEDEPLOY_LAMBDA

        # If canary deployment is needed
        if self.node.try_get_context("canary_deployment"):
            deployment_config = codedeploy.LambdaDeploymentConfig.CANARY_10_PERCENT_30_MINUTES

        # Deployment group
        deployment_group = codedeploy.LambdaDeploymentGroup(
            self,
            "DeploymentGroup",
            application=application,
            alias=live_alias,
            deployment_config=deployment_config,
            # Auto-rollback configuration
            auto_rollback=codedeploy.AutoRollbackConfig(
                failed_deployment=True,
                stopped_deployment=True,
                deployment_in_alarm=True
            ),
            # CloudWatch alarms
            alarms=[
                cloudwatch.Alarm(
                    self,
                    "ErrorAlarm",
                    metric=self.lambda_function.metric_errors(),
                    threshold=5,
                    evaluation_periods=2,
                    datapoints_to_alarm=2
                ),
                cloudwatch.Alarm(
                    self,
                    "DurationAlarm",
                    metric=self.lambda_function.metric_duration(),
                    threshold=5000,  # 5 seconds
                    evaluation_periods=2,
                    datapoints_to_alarm=2
                )
            ]
        )

        # API Gateway pointing to alias
        api = apigateway.RestApi(
            self,
            "WebAPI",
            rest_api_name="web-api"
        )

        integration = apigateway.LambdaIntegration(live_alias)
        api.root.add_method("ANY", integration)
        api.root.add_proxy(default_integration=integration)

        # Outputs
        cdk.CfnOutput(
            self,
            "ApiUrl",
            value=api.url,
            description="API Gateway URL"
        )

        cdk.CfnOutput(
            self,
            "DeploymentGroup",
            value=deployment_group.deployment_group_name,
            description="CodeDeploy deployment group name"
        )

Canary Release

# stacks/canary_deployment.py
import aws_cdk as cdk
from aws_cdk import (
    aws_lambda as lambda_,
    aws_apigateway as apigateway,
    aws_cloudwatch as cloudwatch,
    aws_synthetics as synthetics,
    aws_iam as iam
)
from constructs import Construct

class CanaryDeploymentStack(cdk.Stack):
    """Canary Release Stack"""

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Production version Lambda
        prod_function = lambda_.Function(
            self,
            "ProdFunction",
            runtime=lambda_.Runtime.PYTHON_3_9,
            handler="app.handler",
            code=lambda_.Code.from_asset("lambda/prod"),
            memory_size=256,
            timeout=cdk.Duration.seconds(30),
            environment={
                "VERSION": "prod"
            }
        )

        # Canary version Lambda
        canary_function = lambda_.Function(
            self,
            "CanaryFunction",
            runtime=lambda_.Runtime.PYTHON_3_9,
            handler="app.handler",
            code=lambda_.Code.from_asset("lambda/canary"),
            memory_size=256,
            timeout=cdk.Duration.seconds(30),
            environment={
                "VERSION": "canary"
            }
        )

        # API Gateway with weighted routing
        api = apigateway.RestApi(
            self,
            "CanaryAPI",
            rest_api_name="canary-api",
            description="API with canary deployment"
        )

        # Create deployment stages
        prod_deployment = apigateway.Deployment(
            self,
            "ProdDeployment",
            api=api,
            description="Production deployment"
        )

        canary_deployment = apigateway.Deployment(
            self,
            "CanaryDeployment",
            api=api,
            description="Canary deployment"
        )

        # Production stage
        prod_stage = apigateway.Stage(
            self,
            "ProdStage",
            deployment=prod_deployment,
            stage_name="prod"
        )

        # Canary stage (10% traffic)
        canary_stage = apigateway.Stage(
            self,
            "CanaryStage",
            deployment=canary_deployment,
            stage_name="canary",
            canary_settings=apigateway.CanarySettings(
                percent_traffic=10,
                stage_variables={
                    "lambdaAlias": "canary"
                }
            )
        )

        # Lambda integrations
        prod_integration = apigateway.LambdaIntegration(prod_function)
        canary_integration = apigateway.LambdaIntegration(canary_function)

        # Add routes
        api.root.add_method("ANY", prod_integration)
        api.root.add_proxy(default_integration=prod_integration)

        # CloudWatch Synthetics monitoring
        canary_role = iam.Role(
            self,
            "CanaryRole",
            assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AWSLambdaBasicExecutionRole")
            ]
        )

        # Create Synthetics Canary for continuous monitoring
        synthetic_canary = synthetics.Canary(
            self,
            "ApiCanary",
            canary_name="web-api-canary",
            schedule=synthetics.Schedule.rate(cdk.Duration.minutes(5)),
            test=synthetics.Test.custom(
                code=synthetics.Code.from_inline("""
                const synthetics = require('Synthetics');
                const log = require('SyntheticsLogger');

                const checkApiHealth = async function () {
                    const url = process.env.API_URL + '/health';
                    const response = await synthetics.getPage(url);

                    if (response.status !== 200) {
                        throw new Error(`API health check failed with status: ${response.status}`);
                    }

                    log.info('API health check passed');
                };

                exports.handler = async () => {
                    return await synthetics.executeStep('checkApiHealth', checkApiHealth);
                };
                """),
                handler="index.handler"
            ),
            runtime=synthetics.Runtime.SYNTHETICS_NODEJS_PUPPETEER_3_1,
            environment_variables={
                "API_URL": api.url
            },
            role=canary_role
        )

        # CloudWatch alarms
        error_alarm = cloudwatch.Alarm(
            self,
            "CanaryErrorAlarm",
            alarm_name="canary-error-alarm",
            metric=canary_function.metric_errors(),
            threshold=3,
            evaluation_periods=2,
            datapoints_to_alarm=2,
            alarm_description="Canary deployment error rate too high"
        )

        # Auto-rollback Lambda (simplified version)
        rollback_function = lambda_.Function(
            self,
            "RollbackFunction",
            runtime=lambda_.Runtime.PYTHON_3_9,
            handler="rollback.handler",
            code=lambda_.Code.from_inline("""
import boto3
import json

def handler(event, context):
    # Simplified auto-rollback logic
    apigateway = boto3.client('apigateway')

    # Get alarm details
    alarm_data = json.loads(event['Records'][0]['Sns']['Message'])

    if alarm_data['NewStateValue'] == 'ALARM':
        # Execute rollback: remove canary deployment
        # Actual implementation requires more complex logic
        print("Triggering canary rollback...")

        # Should call API Gateway API to adjust traffic allocation
        # Or trigger CodeDeploy rollback

    return {
        'statusCode': 200,
        'body': json.dumps('Rollback handler executed')
    }
            """),
            timeout=cdk.Duration.seconds(60)
        )

        # Outputs
        cdk.CfnOutput(
            self,
            "ProdApiUrl",
            value=f"{api.url}prod/",
            description="Production API URL"
        )

        cdk.CfnOutput(
            self,
            "CanaryApiUrl",
            value=f"{api.url}canary/",
            description="Canary API URL"
        )

Deployment Scripts and Automation

Environment Deployment Script

# scripts/deploy.py
#!/usr/bin/env python3
"""
CDK Deployment Script
Supports multi-environment deployment and configuration validation
"""

import argparse
import boto3
import subprocess
import sys
import json
from typing import Dict, List, Optional

class CDKDeployer:
    """CDK Deployer"""

    def __init__(self, environment: str, region: str = "us-east-1"):
        self.environment = environment
        self.region = region
        self.session = boto3.Session(region_name=region)

    def validate_environment(self) -> bool:
        """Validate environment configuration"""
        try:
            # Validate AWS credentials
            sts = self.session.client('sts')
            identity = sts.get_caller_identity()
            print(f"Deployment account: {identity['Account']}")

            # Validate environment configuration file
            from config.environments import EnvironmentFactory
            config = EnvironmentFactory.get_config(self.environment)
            print(f"Environment configuration: {config.name} ({config.account})")

            return True

        except Exception as e:
            print(f"Environment validation failed: {e}")
            return False

    def bootstrap_if_needed(self) -> bool:
        """Execute Bootstrap if needed"""
        try:
            # Check if already bootstrapped
            cf = self.session.client('cloudformation')
            stacks = cf.list_stacks(
                StackStatusFilter=['CREATE_COMPLETE', 'UPDATE_COMPLETE']
            )

            bootstrap_exists = any(
                stack['StackName'].startswith('CDKToolkit')
                for stack in stacks['StackSummaries']
            )

            if not bootstrap_exists:
                print("Executing CDK Bootstrap...")
                result = subprocess.run([
                    'cdk', 'bootstrap',
                    '--region', self.region
                ], capture_output=True, text=True)

                if result.returncode != 0:
                    print(f"Bootstrap failed: {result.stderr}")
                    return False

                print("Bootstrap complete")
            else:
                print("Bootstrap already exists, skipping")

            return True

        except Exception as e:
            print(f"Bootstrap check failed: {e}")
            return False

    def synth(self) -> bool:
        """Synthesize CloudFormation template"""
        try:
            print("Synthesizing CDK application...")
            result = subprocess.run([
                'cdk', 'synth',
                '--context', f'environment={self.environment}'
            ], capture_output=True, text=True)

            if result.returncode != 0:
                print(f"Synthesis failed: {result.stderr}")
                return False

            print("Synthesis successful")
            return True

        except Exception as e:
            print(f"Synthesis exception: {e}")
            return False

    def diff(self) -> bool:
        """Show deployment differences"""
        try:
            print("Checking deployment differences...")
            result = subprocess.run([
                'cdk', 'diff',
                '--context', f'environment={self.environment}'
            ], capture_output=True, text=True)

            print(result.stdout)
            if result.stderr:
                print(f"Warning: {result.stderr}")

            return True

        except Exception as e:
            print(f"Diff check exception: {e}")
            return False

    def deploy(self, require_approval: bool = True) -> bool:
        """Deploy application"""
        try:
            print(f"Deploying to {self.environment} environment...")

            cmd = [
                'cdk', 'deploy',
                '--context', f'environment={self.environment}',
                '--all'
            ]

            if not require_approval:
                cmd.append('--require-approval=never')

            result = subprocess.run(cmd, text=True)

            if result.returncode != 0:
                print("Deployment failed")
                return False

            print("Deployment successful")
            return True

        except Exception as e:
            print(f"Deployment exception: {e}")
            return False

    def destroy(self, force: bool = False) -> bool:
        """Destroy application"""
        if not force:
            confirm = input(f"Are you sure you want to destroy {self.environment} environment? (yes/no): ")
            if confirm.lower() != 'yes':
                print("Destroy cancelled")
                return False

        try:
            print(f"Destroying {self.environment} environment...")
            result = subprocess.run([
                'cdk', 'destroy',
                '--context', f'environment={self.environment}',
                '--all',
                '--force'
            ], text=True)

            if result.returncode != 0:
                print("Destroy failed")
                return False

            print("Destroy successful")
            return True

        except Exception as e:
            print(f"Destroy exception: {e}")
            return False

def main():
    parser = argparse.ArgumentParser(description='CDK deployment tool')
    parser.add_argument('action', choices=['validate', 'synth', 'diff', 'deploy', 'destroy'])
    parser.add_argument('--environment', '-e', required=True,
                       choices=['dev', 'staging', 'prod'])
    parser.add_argument('--region', '-r', default='us-east-1')
    parser.add_argument('--no-approval', action='store_true',
                       help='No confirmation required during deployment')
    parser.add_argument('--force', action='store_true',
                       help='Force execution (for destroy)')

    args = parser.parse_args()

    deployer = CDKDeployer(args.environment, args.region)

    # Execute corresponding operation
    if args.action == 'validate':
        success = deployer.validate_environment()
    elif args.action == 'synth':
        success = deployer.synth()
    elif args.action == 'diff':
        success = deployer.diff()
    elif args.action == 'deploy':
        success = (deployer.validate_environment() and
                  deployer.bootstrap_if_needed() and
                  deployer.synth() and
                  deployer.deploy(not args.no_approval))
    elif args.action == 'destroy':
        success = deployer.destroy(args.force)

    sys.exit(0 if success else 1)

if __name__ == "__main__":
    main()

Makefile Automation

# Makefile
ENVIRONMENT ?= dev
REGION ?= us-east-1

.PHONY: help install test lint synth diff deploy destroy clean

help:			## Show help information
	@echo "Available commands:"
	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'

install:		## Install dependencies
	npm install -g aws-cdk
	pip install -r requirements.txt
	pip install -r requirements-dev.txt

test:			## Run tests
	pytest tests/ -v --cov=stacks

lint:			## Code linting
	flake8 stacks/ config/ tests/
	black --check stacks/ config/ tests/
	mypy stacks/ config/

format:			## Format code
	black stacks/ config/ tests/
	isort stacks/ config/ tests/

synth:			## Synthesize template
	python scripts/deploy.py synth --environment $(ENVIRONMENT)

diff:			## Show differences
	python scripts/deploy.py diff --environment $(ENVIRONMENT)

deploy:			## Deploy application
	python scripts/deploy.py deploy --environment $(ENVIRONMENT) --region $(REGION)

deploy-prod:		## Deploy to production environment
	@echo "Deploying to production environment, please confirm configuration..."
	python scripts/deploy.py deploy --environment prod --region $(REGION)

destroy:		## Destroy environment
	python scripts/deploy.py destroy --environment $(ENVIRONMENT) --region $(REGION)

clean:			## Clean temporary files
	find . -type f -name "*.pyc" -delete
	find . -type d -name "__pycache__" -delete
	rm -rf cdk.out/
	rm -rf .pytest_cache/
	rm -rf htmlcov/

validate-all:		## Validate all environments
	@for env in dev staging prod; do \
		echo "Validating $$env environment..."; \
		python scripts/deploy.py validate --environment $$env || exit 1; \
	done

bootstrap-all:		## Bootstrap all environments
	@for env in dev staging prod; do \
		echo "Bootstrapping $$env environment..."; \
		cdk bootstrap --context environment=$$env || exit 1; \
	done

Through this chapter, you should be able to design and implement enterprise-grade CDK deployment strategies, achieving secure, reliable, and scalable multi-environment infrastructure management.