9/1/25About 10 min
Chapter 7: Deployment Strategies and Environment Management
Learning Objectives
- Master multi-environment deployment strategies and best practices for CDK
- Understand the lifecycle management of CDK applications
- Learn to implement CI/CD using CDK Pipelines
- Master configuration management and parameterized deployments
- Understand advanced deployment strategies like blue-green and canary deployments
Knowledge Point Summary
Environment Management Overview
In enterprise-level applications, it is common to manage multiple environments: development (dev), testing (test), staging, and production (prod). CDK provides flexible mechanisms to manage the differences between these environments.
Differentiated Environment 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 the 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 a 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 an 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 a 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 a 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 the 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 an 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 an 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 a 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
```python
# 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 variables 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 the Stack
ConfigurableWebStack(
app,
f"WebApp-{config.name}",
config=config,
description=f"Web application stack for {config.name} environment"
)
app.synth()
if __name__ == "__main__":
main()CDK Pipelines for CI/CD
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 a CodeCommit repository
repo = codecommit.Repository(
self,
"WebAppRepo",
repository_name="web-app-cdk",
description="Web application CDK code repository"
)
# Create the 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 tests for the development environment
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 tests for the staging environment
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 validation for the production environment
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 the Web App Stack
web_stack = ConfigurableWebStack(
self,
f"WebApp-{env_name}",
config=config
)
# Output the API URL for subsequent stages to use
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 progressive 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 points to the 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 Deployment
# 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 Deployment 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 a 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"
)
# Automatic Lambda rollback (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 automatic 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...")
# This should call the API Gateway API to adjust traffic allocation
# or trigger a 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 for 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 the 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 the application"""
if not force:
confirm = input(f"Are you sure you want to destroy the {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='Do not require approval for deployment')
parser.add_argument('--force', action='store_true',
help='Force execution (for destroy)')
args = parser.parse_args()
deployer = CDKDeployer(args.environment, args.region)
# Execute the corresponding action
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/ tests/
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
@echo "Deploying to production, 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; \
doneSummary of Deployment Best Practices
- Environment Isolation: Use different AWS accounts or strict naming conventions to isolate environments
- Configuration Management: Externalize environment-specific configurations to avoid hard-coding
- Progressive Deployment: Use canary or blue-green deployments to reduce risk
- Automated Testing: Integrate appropriate tests at each deployment stage
- Monitoring and Alarms: Set up comprehensive monitoring and automatic rollback mechanisms
- Permission Control: Follow the principle of least privilege and configure different access permissions for different environments
- Documentation Maintenance: Keep deployment documentation and configuration instructions up-to-date
By completing this chapter, you should be able to design and implement enterprise-grade CDK deployment strategies to achieve secure, reliable, and scalable multi-environment infrastructure management.