9/1/25About 13 min
Chapter 12: CDK Ecosystem and Extensions
Learning Objectives
- Understand the composition and development of the CDK ecosystem
- Master the use and integration of third-party CDK constructs
- Learn to create and publish custom CDK constructs
- Understand the integration of CDK with other IaC tools
- Master the development of CDK plugins and extensions
- Learn about CDK community best practices and contribution methods
CDK Ecosystem Overview
AWS CDK has a rich ecosystem that includes official constructs, community contributions, third-party integrations, and more.
Third-party Construct Integration
Popular Community Constructs
# Example project using community constructs
# requirements.txt
"""
aws-cdk-lib>=2.50.0
constructs>=10.0.0
# Community constructs
cdk-dynamo-table-viewer>=4.0.0
cdk-spa-deploy>=2.0.0
cdk-datadog>=1.0.0
cdk-github>=1.0.0
cdk-fargate-patterns>=2.0.0
"""
# stacks/community_constructs_stack.py
import aws_cdk as cdk
from constructs import Construct
# Third-party construct imports
try:
from cdk_dynamo_table_viewer import TableViewer
from cdk_spa_deploy import SPADeploy
from datadog_cdk_constructs import Datadog
from cdk_github import GitHubRepository
except ImportError as e:
print(f"Some community constructs are not installed: {e}")
print("Please run: pip install cdk-dynamo-table-viewer cdk-spa-deploy")
class CommunityConstructsStack(cdk.Stack):
"""Example Stack demonstrating the use of community constructs"""
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# 1. DynamoDB Table Viewer - visualize DynamoDB tables
self._setup_dynamo_viewer()
# 2. SPA Deploy - single-page application deployment
self._setup_spa_deploy()
# 3. Datadog integration
self._setup_datadog_integration()
# 4. GitHub integration
self._setup_github_integration()
# 5. Example of using a custom community construct
self._demonstrate_pattern_usage()
def _setup_dynamo_viewer(self):
"""Set up the DynamoDB table viewer"""
# Create a DynamoDB table
from aws_cdk import aws_dynamodb as dynamodb
users_table = dynamodb.Table(
self,
"UsersTable",
table_name="community-users-table",
partition_key=dynamodb.Attribute(
name="userId",
type=dynamodb.AttributeType.STRING
),
billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
removal_policy=cdk.RemovalPolicy.DESTROY
)
# Create a table viewer using a community construct
if 'TableViewer' in globals():
table_viewer = TableViewer(
self,
"UsersTableViewer",
table=users_table,
title="Users Table Viewer",
sort_by="-userId"
)
cdk.CfnOutput(
self,
"TableViewerUrl",
value=table_viewer.endpoint,
description="DynamoDB Table Viewer URL"
)
def _setup_spa_deploy(self):
"""Set up SPA deployment"""
if 'SPADeploy' in globals():
spa_deploy = SPADeploy(
self,
"SPADeploy",
source_path="./frontend/build", # Build output directory for React/Vue, etc.
# Custom domain (optional)
custom_domain_name="app.example.com" if self.node.try_get_context("domain_name") else None,
# Enable HTTPS redirect
redirect_https=True,
# Enable IPv6
ipv6_enabled=True,
# Error configurations
error_configurations=[
{
"error_code": 404,
"response_code": 200,
"response_page_path": "/index.html"
}
]
)
cdk.CfnOutput(
self,
"SPAUrl",
value=spa_deploy.cloudfront_distribution.distribution_domain_name,
description="SPA CloudFront URL"
)
def _setup_datadog_integration(self):
"""Set up Datadog monitoring integration"""
if 'Datadog' in globals():
# Datadog API key (should be stored in Secrets Manager)
datadog_api_key = cdk.SecretValue.secrets_manager("datadog/api-key")
datadog_integration = Datadog(
self,
"DatadogIntegration",
api_key=datadog_api_key,
# Monitored service
service="community-app",
env="production",
version="1.0.0",
# Enabled monitoring features
enable_logs_monitoring=True,
enable_enhanced_metrics=True,
enable_tracing=True,
# Tags
tags=[
"team:engineering",
"project:community-app",
"environment:prod"
]
)
# Custom dashboard
datadog_integration.create_dashboard({
"title": "Community App Dashboard",
"widgets": [
{
"type": "timeseries",
"title": "API Requests",
"query": "sum:aws.apigateway.count{service:community-app}.as_count()"
},
{
"type": "query_value",
"title": "Error Rate",
"query": "sum:aws.apigateway.5xx{service:community-app}.as_count()"
}
]
})
def _setup_github_integration(self):
"""Set up GitHub integration"""
if 'GitHubRepository' in globals():
# Create a GitHub repository (if it doesn't exist)
github_repo = GitHubRepository(
self,
"CommunityAppRepo",
repository_name="community-app-infrastructure",
description="Infrastructure as Code for Community App",
# Repository configuration
private=True,
auto_init=True,
gitignore_template="Python",
license_template="MIT",
# Branch protection rules
branch_protection_rules=[
{
"pattern": "main",
"required_status_checks": {
"strict": True,
"contexts": ["ci/build", "ci/test"]
},
"required_pull_request_reviews": {
"required_approving_review_count": 2,
"dismiss_stale_reviews": True
},
"enforce_admins": False,
"restrictions": None
}
]
)
# GitHub Actions workflow
github_repo.add_workflow(
name="ci",
workflow={
"name": "CI/CD Pipeline",
"on": {
"push": {"branches": ["main", "develop"]},
"pull_request": {"branches": ["main"]}
},
"jobs": {
"test": {
"runs-on": "ubuntu-latest",
"steps": [
{"uses": "actions/checkout@v3"},
{
"uses": "actions/setup-python@v4",
"with": {"python-version": "3.9"}
},
{"run": "pip install -r requirements.txt"},
{"run": "pytest tests/"},
{"run": "cdk synth"}
]
},
"deploy": {
"needs": "test",
"runs-on": "ubuntu-latest",
"if": "github.ref == 'refs/heads/main'",
"steps": [
{"uses": "actions/checkout@v3"},
{"run": "cdk deploy --require-approval never"}
]
}
}
}
)
def _demonstrate_pattern_usage(self):
"""Demonstrate the use of common patterns"""
# Create complex patterns using a combination of constructs
from aws_cdk import (
aws_lambda as lambda_,
aws_apigateway as apigateway,
aws_dynamodb as dynamodb,
aws_s3 as s3,
aws_cloudfront as cloudfront
)
# Serverless API pattern
api_lambda = lambda_.Function(
self,
"APIFunction",
runtime=lambda_.Runtime.PYTHON_3_9,
handler="api.handler",
code=lambda_.Code.from_asset("lambda/api")
)
api = apigateway.RestApi(
self,
"CommunityAPI",
rest_api_name="community-api"
)
api.root.add_proxy(
default_integration=apigateway.LambdaIntegration(api_lambda)
)
# Static assets + dynamic API pattern
assets_bucket = s3.Bucket(
self,
"AssetsBucket",
bucket_name=f"community-assets-{cdk.Aws.ACCOUNT_ID}",
public_read_access=False
)
distribution = cloudfront.CloudFrontWebDistribution(
self,
"CommunityDistribution",
origin_configs=[
# Static assets source
cloudfront.SourceConfiguration(
s3_origin_source=cloudfront.S3OriginConfig(
s3_bucket_source=assets_bucket
),
behaviors=[
cloudfront.Behavior(
path_pattern="/static/*",
is_default_behavior=False
)
]
),
# Dynamic content source
cloudfront.SourceConfiguration(
custom_origin_source=cloudfront.CustomOriginConfig(
domain_name=api.url.replace("https://", "").replace("/", "")
),
behaviors=[
cloudfront.Behavior(
path_pattern="/api/*",
is_default_behavior=False,
allowed_methods=cloudfront.CloudFrontAllowedMethods.ALL
)
]
)
],
default_root_object="index.html"
)Custom Construct Development
Creating a Reusable Construct
# constructs/multi_tier_web_app.py
import aws_cdk as cdk
from aws_cdk import (
aws_ec2 as ec2,
aws_rds as rds,
aws_elasticloadbalancingv2 as elbv2,
aws_autoscaling as autoscaling,
aws_s3 as s3,
aws_cloudfront as cloudfront
)
from constructs import Construct
from typing import Optional, List, Dict
import json
class MultiTierWebAppProps:
"""Properties for the multi-tier web application construct"""
def __init__(self,
app_name: str,
environment: str,
vpc_cidr: str = "10.0.0.0/16",
instance_type: str = "t3.medium",
min_capacity: int = 2,
max_capacity: int = 10,
database_instance_type: str = "db.t3.micro",
enable_https: bool = True,
domain_name: Optional[str] = None,
**kwargs):
self.app_name = app_name
self.environment = environment
self.vpc_cidr = vpc_cidr
self.instance_type = instance_type
self.min_capacity = min_capacity
self.max_capacity = max_capacity
self.database_instance_type = database_instance_type
self.enable_https = enable_https
self.domain_name = domain_name
# Merge any additional properties
for key, value in kwargs.items():
setattr(self, key, value)
class MultiTierWebApp(Construct):
"""Multi-tier web application construct - a reusable complete web application architecture"""
def __init__(self, scope: Construct, construct_id: str,
props: MultiTierWebAppProps, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
self.props = props
# Store references to created resources
self.vpc: Optional[ec2.Vpc] = None
self.database: Optional[rds.DatabaseInstance] = None
self.load_balancer: Optional[elbv2.ApplicationLoadBalancer] = None
self.auto_scaling_group: Optional[autoscaling.AutoScalingGroup] = None
self.cdn: Optional[cloudfront.CloudFrontWebDistribution] = None
# Create the infrastructure layers
self._create_networking()
self._create_database_layer()
self._create_application_layer()
self._create_load_balancer()
self._create_cdn()
# Output important information
self._create_outputs()
def _create_networking(self):
"""Create the network layer"""
self.vpc = ec2.Vpc(
self,
"VPC",
ip_addresses=ec2.IpAddresses.cidr(self.props.vpc_cidr),
max_azs=3,
nat_gateways=2,
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
),
ec2.SubnetConfiguration(
cidr_mask=28,
name="Database",
subnet_type=ec2.SubnetType.PRIVATE_ISOLATED
)
]
)
# Security groups
self.web_security_group = ec2.SecurityGroup(
self,
"WebSecurityGroup",
vpc=self.vpc,
description=f"Security group for {self.props.app_name} web servers",
allow_all_outbound=True
)
self.database_security_group = ec2.SecurityGroup(
self,
"DatabaseSecurityGroup",
vpc=self.vpc,
description=f"Security group for {self.props.app_name} database",
allow_all_outbound=False
)
# Allow the web tier to access the database
self.database_security_group.add_ingress_rule(
peer=self.web_security_group,
connection=ec2.Port.tcp(5432),
description="Allow web servers to access database"
)
def _create_database_layer(self):
"""Create the database layer"""
# Database subnet group
db_subnet_group = rds.SubnetGroup(
self,
"DatabaseSubnetGroup",
description=f"Subnet group for {self.props.app_name} database",
vpc=self.vpc,
vpc_subnets=ec2.SubnetSelection(
subnet_type=ec2.SubnetType.PRIVATE_ISOLATED
)
)
# Database credentials
db_credentials = rds.DatabaseSecret(
self,
"DatabaseCredentials",
username=f"{self.props.app_name}admin"
)
# RDS instance
self.database = rds.DatabaseInstance(
self,
"Database",
engine=rds.DatabaseInstanceEngine.postgres(
version=rds.PostgresEngineVersion.VER_14_9
),
instance_type=ec2.InstanceType(self.props.database_instance_type),
vpc=self.vpc,
subnet_group=db_subnet_group,
security_groups=[self.database_security_group],
credentials=rds.Credentials.from_secret(db_credentials),
database_name=self.props.app_name.replace("-", "_"),
backup_retention=cdk.Duration.days(7),
deletion_protection=self.props.environment == "prod",
multi_az=self.props.environment == "prod"
)
def _create_application_layer(self):
"""Create the application layer"""
# User data script
user_data = ec2.UserData.for_linux()
user_data.add_commands(
"yum update -y",
"yum install -y python3 python3-pip",
f"echo 'APP_NAME={self.props.app_name}' > /etc/environment",
f"echo 'ENVIRONMENT={self.props.environment}' >> /etc/environment",
f"echo 'DB_HOST={self.database.instance_endpoint.hostname}' >> /etc/environment",
"systemctl enable amazon-cloudwatch-agent",
"systemctl start amazon-cloudwatch-agent"
)
# Launch template
launch_template = ec2.LaunchTemplate(
self,
"LaunchTemplate",
instance_type=ec2.InstanceType(self.props.instance_type),
machine_image=ec2.AmazonLinuxImage(
generation=ec2.AmazonLinuxGeneration.AMAZON_LINUX_2
),
security_group=self.web_security_group,
user_data=user_data,
role=self._create_instance_role()
)
# Auto Scaling Group
self.auto_scaling_group = autoscaling.AutoScalingGroup(
self,
"AutoScalingGroup",
vpc=self.vpc,
launch_template=launch_template,
min_capacity=self.props.min_capacity,
max_capacity=self.props.max_capacity,
desired_capacity=self.props.min_capacity,
vpc_subnets=ec2.SubnetSelection(
subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS
),
health_check=autoscaling.HealthCheck.elb(
grace_period=cdk.Duration.minutes(5)
),
update_policy=autoscaling.UpdatePolicy.rolling_update(
min_instances_in_service=1,
max_batch_size=2
)
)
# CPU-based scaling policy
self.auto_scaling_group.scale_on_cpu_utilization(
"CPUScaling",
target_utilization_percent=70,
scale_in_cooldown=cdk.Duration.minutes(5),
scale_out_cooldown=cdk.Duration.minutes(2)
)
def _create_instance_role(self):
"""Create an EC2 instance role"""
from aws_cdk import aws_iam as iam
role = iam.Role(
self,
"InstanceRole",
assumed_by=iam.ServicePrincipal("ec2.amazonaws.com"),
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name("CloudWatchAgentServerPolicy"),
iam.ManagedPolicy.from_aws_managed_policy_name("AmazonSSMManagedInstanceCore")
]
)
# Allow access to the database secret
role.add_to_policy(
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=["secretsmanager:GetSecretValue"],
resources=[self.database.secret.secret_arn]
)
)
return role
def _create_load_balancer(self):
"""Create a load balancer"""
# Application Load Balancer
self.load_balancer = elbv2.ApplicationLoadBalancer(
self,
"LoadBalancer",
vpc=self.vpc,
internet_facing=True,
vpc_subnets=ec2.SubnetSelection(
subnet_type=ec2.SubnetType.PUBLIC
)
)
# Target group
target_group = elbv2.ApplicationTargetGroup(
self,
"TargetGroup",
port=80,
protocol=elbv2.ApplicationProtocol.HTTP,
targets=[self.auto_scaling_group],
vpc=self.vpc,
health_check=elbv2.HealthCheck(
enabled=True,
healthy_http_codes="200",
interval=cdk.Duration.seconds(30),
path="/health",
timeout=cdk.Duration.seconds(5)
)
)
# HTTP listener
http_listener = self.load_balancer.add_listener(
"HTTPListener",
port=80,
protocol=elbv2.ApplicationProtocol.HTTP,
default_target_groups=[target_group]
)
# HTTPS listener (if enabled)
if self.props.enable_https and self.props.domain_name:
from aws_cdk import aws_certificatemanager as acm
certificate = acm.Certificate(
self,
"Certificate",
domain_name=self.props.domain_name,
validation=acm.CertificateValidation.from_dns()
)
https_listener = self.load_balancer.add_listener(
"HTTPSListener",
port=443,
protocol=elbv2.ApplicationProtocol.HTTPS,
certificates=[certificate],
default_target_groups=[target_group]
)
# HTTP to HTTPS redirect
http_listener.add_action(
"HTTPSRedirect",
action=elbv2.ListenerAction.redirect(
protocol="HTTPS",
port="443",
permanent=True
)
)
# Allow the ALB to access the web servers
self.web_security_group.add_ingress_rule(
peer=ec2.Peer.any_ipv4(),
connection=ec2.Port.tcp(80),
description="Allow HTTP from ALB"
)
def _create_cdn(self):
"""Create a CDN"""
# Static assets bucket
static_assets_bucket = s3.Bucket(
self,
"StaticAssetsBucket",
bucket_name=f"{self.props.app_name}-static-{cdk.Aws.ACCOUNT_ID}",
public_read_access=False,
removal_policy=cdk.RemovalPolicy.DESTROY
)
# CloudFront distribution
self.cdn = cloudfront.CloudFrontWebDistribution(
self,
"CDN",
origin_configs=[
# Static assets source
cloudfront.SourceConfiguration(
s3_origin_source=cloudfront.S3OriginConfig(
s3_bucket_source=static_assets_bucket,
origin_access_identity=cloudfront.OriginAccessIdentity(
self, "OAI"
)
),
behaviors=[
cloudfront.Behavior(
path_pattern="/static/*",
is_default_behavior=False,
compress=True,
cache_policy=cloudfront.CachePolicy.CACHING_OPTIMIZED
)
]
),
# Dynamic content source
cloudfront.SourceConfiguration(
custom_origin_source=cloudfront.CustomOriginConfig(
domain_name=self.load_balancer.load_balancer_dns_name,
http_port=80,
https_port=443 if self.props.enable_https else None,
origin_protocol_policy=cloudfront.OriginProtocolPolicy.HTTP_ONLY if not self.props.enable_https else cloudfront.OriginProtocolPolicy.HTTPS_ONLY
),
behaviors=[
cloudfront.Behavior(
path_pattern="/api/*",
is_default_behavior=False,
cache_policy=cloudfront.CachePolicy.CACHING_DISABLED,
allowed_methods=cloudfront.CloudFrontAllowedMethods.ALL
),
cloudfront.Behavior(
is_default_behavior=True,
cache_policy=cloudfront.CachePolicy.CACHING_OPTIMIZED
)
]
)
],
price_class=cloudfront.PriceClass.PRICE_CLASS_100,
enable_ipv6=True
)
def _create_outputs(self):
"""Create outputs"""
cdk.CfnOutput(
self,
"LoadBalancerDNS",
value=self.load_balancer.load_balancer_dns_name,
description=f"Load Balancer DNS for {self.props.app_name}"
)
cdk.CfnOutput(
self,
"CDNDomainName",
value=self.cdn.distribution_domain_name,
description=f"CloudFront distribution domain for {self.props.app_name}"
)
cdk.CfnOutput(
self,
"DatabaseEndpoint",
value=self.database.instance_endpoint.hostname,
description=f"Database endpoint for {self.props.app_name}"
)
cdk.CfnOutput(
self,
"VPCId",
value=self.vpc.vpc_id,
description=f"VPC ID for {self.props.app_name}"
)
def add_monitoring(self, alert_email: str):
"""Add monitoring and alerting"""
from aws_cdk import aws_cloudwatch as cloudwatch
from aws_cdk import aws_sns as sns
from aws_cdk import aws_sns_subscriptions as subscriptions
# SNS topic
alert_topic = sns.Topic(
self,
"AlertTopic",
topic_name=f"{self.props.app_name}-alerts"
)
alert_topic.add_subscription(
subscriptions.EmailSubscription(alert_email)
)
# CloudWatch alarms
# High CPU utilization alarm
cloudwatch.Alarm(
self,
"HighCPUAlarm",
alarm_name=f"{self.props.app_name}-high-cpu",
metric=self.auto_scaling_group.metric_cpu_utilization(),
threshold=80,
evaluation_periods=3
).add_alarm_action(
cloudwatch.SnsAction(alert_topic)
)
# ALB error rate alarm
cloudwatch.Alarm(
self,
"HighErrorRateAlarm",
alarm_name=f"{self.props.app_name}-high-error-rate",
metric=self.load_balancer.metric_http_code_elb(
code=elbv2.HttpCodeElb.ELB_5XX_COUNT
),
threshold=10,
evaluation_periods=2
).add_alarm_action(
cloudwatch.SnsAction(alert_topic)
)
# Database connections alarm
cloudwatch.Alarm(
self,
"DatabaseConnectionsAlarm",
alarm_name=f"{self.props.app_name}-db-connections",
metric=self.database.metric_database_connections(),
threshold=80,
evaluation_periods=2
).add_alarm_action(
cloudwatch.SnsAction(alert_topic)
)
def enable_backup(self, retention_days: int = 30):
"""Enable backups"""
from aws_cdk import aws_backup as backup
# Backup vault
backup_vault = backup.BackupVault(
self,
"BackupVault",
backup_vault_name=f"{self.props.app_name}-backup-vault"
)
# Backup plan
backup_plan = backup.BackupPlan(
self,
"BackupPlan",
backup_plan_name=f"{self.props.app_name}-backup-plan",
backup_plan_rules=[
backup.BackupPlanRule(
rule_name="DailyBackups",
backup_vault=backup_vault,
schedule_expression=backup.Schedule.cron(
hour="2",
minute="0"
),
delete_after=cdk.Duration.days(retention_days)
)
]
)
# Add the database to the backup selection
backup_plan.add_selection(
"DatabaseBackupSelection",
resources=[
backup.BackupResource.from_rds_database_instance(self.database)
]
)Construct Publishing and Distribution
# Construct package publishing configuration
# setup.py
from setuptools import setup, find_packages
setup(
name="multi-tier-web-app-cdk",
version="1.0.0",
description="A reusable CDK construct for multi-tier web applications",
long_description=open("README.md").read(),
long_description_content_type="text/markdown",
author="Your Name",
author_email="your.email@example.com",
url="https://github.com/yourusername/multi-tier-web-app-cdk",
packages=find_packages(exclude=["tests*"]),
install_requires=[
"aws-cdk-lib>=2.50.0",
"constructs>=10.0.0"
],
python_requires=">=3.7",
classifiers=[
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: System :: Systems Administration",
"Typing :: Typed"
],
keywords="aws cdk construct web-app infrastructure",
project_urls={
"Bug Reports": "https://github.com/yourusername/multi-tier-web-app-cdk/issues",
"Source": "https://github.com/yourusername/multi-tier-web-app-cdk",
"Documentation": "https://multi-tier-web-app-cdk.readthedocs.io/"
}
)# .github/workflows/publish.yml
name: Publish to PyPI
on:
release:
types: [published]
jobs:
build-and-publish:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.9'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install build twine
pip install -r requirements.txt
- name: Run tests
run: |
pip install pytest
pytest tests/
- name: Build package
run: python -m build
- name: Publish to PyPI
env:
TWINE_USERNAME: __token__
TWINE_PASSWORD: ${{ secrets.PYPI_API_TOKEN }}
run: twine upload dist/*
- name: Publish to Constructs Hub
run: |
# Automatically submit to Constructs Hub
echo "Package published, will appear in Constructs Hub"CDK Tool Integration
VS Code Plugin Development
// Example VS Code CDK plugin
// package.json
{
"name": "aws-cdk-helper",
"displayName": "AWS CDK Helper",
"description": "Helper tools for AWS CDK development",
"version": "1.0.0",
"engines": {
"vscode": "^1.60.0"
},
"categories": ["Other"],
"activationEvents": [
"onLanguage:python",
"onLanguage:typescript",
"workspaceContains:**/cdk.json"
],
"main": "./out/extension.js",
"contributes": {
"commands": [
{
"command": "cdk-helper.synthesize",
"title": "CDK: Synthesize"
},
{
"command": "cdk-helper.deploy",
"title": "CDK: Deploy"
},
{
"command": "cdk-helper.diff",
"title": "CDK: Diff"
}
],
"menus": {
"explorer/context": [
{
"command": "cdk-helper.synthesize",
"when": "resourceFilename == cdk.json",
"group": "navigation"
}
]
},
"configuration": {
"title": "AWS CDK Helper",
"properties": {
"cdk-helper.defaultProfile": {
"type": "string",
"default": "default",
"description": "Default AWS profile to use"
}
}
}
},
"scripts": {
"vscode:prepublish": "npm run compile",
"compile": "tsc -p ./",
"watch": "tsc -watch -p ./'
},
"devDependencies": {
"@types/vscode": "^1.60.0",
"@types/node": "16.x",
"typescript": "^4.4.4"
}
}// src/extension.ts
import * as vscode from 'vscode';
import { exec } from 'child_process';
import * as path from 'path';
export function activate(context: vscode.ExtensionContext) {
console.log('AWS CDK Helper is now active!');
// Register commands
let synthesizeCommand = vscode.commands.registerCommand('cdk-helper.synthesize', () => {
synthesizeCDK();
});
let deployCommand = vscode.commands.registerCommand('cdk-helper.deploy', () => {
deployCDK();
});
let diffCommand = vscode.commands.registerCommand('cdk-helper.diff', () => {
diffCDK();
});
context.subscriptions.push(synthesizeCommand, deployCommand, diffCommand);
// Status bar item
let statusBarItem = vscode.window.createStatusBarItem(vscode.StatusBarAlignment.Left);
statusBarItem.text = "$(cloud) CDK";
statusBarItem.command = 'cdk-helper.synthesize';
statusBarItem.show();
context.subscriptions.push(statusBarItem);
// File watcher
let watcher = vscode.workspace.createFileSystemWatcher('**/*.py');
watcher.onDidChange(() => {
validateCDKCode();
});
context.subscriptions.push(watcher);
}
function synthesizeCDK() {
const terminal = vscode.window.createTerminal('CDK Synthesize');
terminal.show();
terminal.sendText('cdk synth');
}
function deployCDK() {
vscode.window.showInformationMessage('Deploying CDK stack...', 'Confirm', 'Cancel')
.then(selection => {
if (selection === 'Confirm') {
const terminal = vscode.window.createTerminal('CDK Deploy');
terminal.show();
terminal.sendText('cdk deploy');
}
});
}
function diffCDK() {
const terminal = vscode.window.createTerminal('CDK Diff');
terminal.show();
terminal.sendText('cdk diff');
}
function validateCDKCode() {
// Simple code validation logic
const activeEditor = vscode.window.activeTextEditor;
if (activeEditor) {
const document = activeEditor.document;
if (document.languageId === 'python') {
// Check for common CDK patterns
const text = document.getText();
if (text.includes('aws_cdk') && text.includes('Stack')) {
vscode.window.showInformationMessage('CDK Stack detected!');
}
}
}
}
export function deactivate() {}CDK Testing Tools
# tests/test_framework.py
import json
import pytest
from aws_cdk import App, assertions
from constructs.multi_tier_web_app import MultiTierWebApp, MultiTierWebAppProps
class CDKTestFramework:
"""CDK Testing Framework"""
def __init__(self):
self.app = App()
self.test_results = []
def create_test_stack(self, stack_name: str, props: MultiTierWebAppProps):
"""Create a test Stack"""
return MultiTierWebApp(self.app, stack_name, props)
def assert_resource_count(self, stack, resource_type: str, expected_count: int):
"""Assert resource count"""
template = assertions.Template.from_stack(stack)
template.resource_count_is(resource_type, expected_count)
self.test_results.append({
"test": "resource_count",
"resource_type": resource_type,
"expected": expected_count,
"status": "passed"
})
def assert_resource_properties(self, stack, resource_type: str, properties: dict):
"""Assert resource properties"""
template = assertions.Template.from_stack(stack)
template.has_resource_properties(resource_type, properties)
self.test_results.append({
"test": "resource_properties",
"resource_type": resource_type,
"properties": properties,
"status": "passed"
})
def assert_security_best_practices(self, stack):
"""Assert security best practices"""
template = assertions.Template.from_stack(stack)
# Check if S3 buckets have encryption enabled
template.has_resource_properties("AWS::S3::Bucket", {
"BucketEncryption": assertions.Match.object_like({
"ServerSideEncryptionConfiguration": assertions.Match.any_value()
})
})
# Check if the database has encryption enabled
template.has_resource_properties("AWS::RDS::DBInstance", {
"StorageEncrypted": True
})
# Check security group rules
template.has_resource_properties("AWS::EC2::SecurityGroup", {
"SecurityGroupIngress": assertions.Match.array_with([
assertions.Match.object_like({
"IpProtocol": "tcp",
"FromPort": assertions.Match.any_value(),
"ToPort": assertions.Match.any_value()
})
])
})
self.test_results.append({
"test": "security_best_practices",
"status": "passed"
})
def generate_test_report(self, output_file: str = "test_report.json"):
"""Generate a test report"""
report = {
"timestamp": json.dumps(datetime.now().isoformat()),
"total_tests": len(self.test_results),
"passed_tests": len([r for r in self.test_results if r["status"] == "passed"]),
"failed_tests": len([r for r in self.test_results if r["status"] == "failed"]),
"test_details": self.test_results
}
with open(output_file, 'w') as f:
json.dump(report, f, indent=2)
return report
# Example of using the test framework
def test_multi_tier_web_app():
"""Test the multi-tier web application construct"""
# Create the test framework
test_framework = CDKTestFramework()
# Create test properties
props = MultiTierWebAppProps(
app_name="test-app",
environment="test",
instance_type="t3.micro",
min_capacity=1,
max_capacity=3
)
# Create the test Stack
stack = test_framework.create_test_stack("TestStack", props)
# Execute tests
test_framework.assert_resource_count(stack, "AWS::EC2::VPC", 1)
test_framework.assert_resource_count(stack, "AWS::RDS::DBInstance", 1)
test_framework.assert_resource_count(stack, "AWS::ElasticLoadBalancingV2::LoadBalancer", 1)
test_framework.assert_resource_count(stack, "AWS::AutoScaling::AutoScalingGroup", 1)
# Security tests
test_framework.assert_security_best_practices(stack)
# Generate the test report
report = test_framework.generate_test_report()
print(f"Tests complete: {report['passed_tests']}/{report['total_tests']} passed")
if __name__ == "__main__":
test_multi_tier_web_app()CDK CLI Extension
# cdk_extensions/cli_plugin.py
import click
import subprocess
import json
import os
from typing import List, Dict
@click.group()
def cdk_ext():
"""CDK extended command-line tool"""
pass
@cdk_ext.command()
@click.option('--stack-name', help='Stack name to analyze')
@click.option('--output', default='analysis.json', help='Output file name')
def analyze(stack_name: str, output: str):
"""Analyze the cost and security of a CDK Stack"""
click.echo(f"Analyzing Stack: {stack_name}")
# Get the Stack template
result = subprocess.run(['cdk', 'synth', stack_name],
capture_output=True, text=True)
if result.returncode != 0:
click.echo(f"Error: {result.stderr}", err=True)
return
template = json.loads(result.stdout)
# Analyze cost
cost_analysis = analyze_cost(template)
# Analyze security
security_analysis = analyze_security(template)
# Generate a report
analysis_report = {
"stack_name": stack_name,
"cost_analysis": cost_analysis,
"security_analysis": security_analysis,
"recommendations": generate_recommendations(cost_analysis, security_analysis)
}
with open(output, 'w') as f:
json.dump(analysis_report, f, indent=2)
click.echo(f"Analysis report saved to: {output}")
def analyze_cost(template: Dict) -> Dict:
"""Cost analysis"""
resources = template.get('Resources', {})
cost_estimate = 0
resource_costs = {}
for resource_id, resource in resources.items():
resource_type = resource.get('Type')
properties = resource.get('Properties', {})
# Simplified cost estimation logic
if resource_type == 'AWS::EC2::Instance':
instance_type = properties.get('InstanceType', 't3.micro')
cost_estimate += get_ec2_cost(instance_type) * 24 * 30 # Monthly cost
resource_costs[resource_id] = {
"type": resource_type,
"monthly_cost": get_ec2_cost(instance_type) * 24 * 30
}
elif resource_type == 'AWS::RDS::DBInstance':
instance_class = properties.get('DBInstanceClass', 'db.t3.micro')
cost_estimate += get_rds_cost(instance_class) * 24 * 30
resource_costs[resource_id] = {
"type": resource_type,
"monthly_cost": get_rds_cost(instance_class) * 24 * 30
}
return {
"total_monthly_estimate": round(cost_estimate, 2),
"resource_breakdown": resource_costs,
"top_cost_drivers": sorted(
resource_costs.items(),
key=lambda x: x[1]["monthly_cost"],
reverse=True
)[:5]
}
def analyze_security(template: Dict) -> Dict:
"""Security analysis"""
resources = template.get('Resources', {})
security_issues = []
security_score = 100
for resource_id, resource in resources.items():
resource_type = resource.get('Type')
properties = resource.get('Properties', {})
# S3 bucket security checks
if resource_type == 'AWS::S3::Bucket':
if not properties.get('BucketEncryption'):
security_issues.append({
"resource": resource_id,
"issue": "S3 bucket encryption not enabled",
"severity": "high",
"recommendation": "Enable server-side encryption"
})
security_score -= 15
if properties.get('PublicAccessBlockConfiguration', {}).get('BlockPublicAcls') != True:
security_issues.append({
"resource": resource_id,
"issue": "S3 bucket allows public access",
"severity": "critical",
"recommendation": "Block all public access"
})
security_score -= 25
# RDS security checks
elif resource_type == 'AWS::RDS::DBInstance':
if not properties.get('StorageEncrypted'):
security_issues.append({
"resource": resource_id,
"issue": "RDS instance storage not encrypted",
"severity": "high",
"recommendation": "Enable storage encryption"
})
security_score -= 15
# Security group checks
elif resource_type == 'AWS::EC2::SecurityGroup':
ingress_rules = properties.get('SecurityGroupIngress', [])
for rule in ingress_rules:
if rule.get('CidrIp') == '0.0.0.0/0' and rule.get('FromPort') != 80 and rule.get('FromPort') != 443:
security_issues.append({
"resource": resource_id,
"issue": f"Security group allows unrestricted access on port {rule.get('FromPort')}",
"severity": "high",
"recommendation": "Restrict access to specific IP ranges"
})
security_score -= 10
return {
"security_score": max(0, security_score),
"issues_found": len(security_issues),
"issues": security_issues,
"compliance_status": "compliant" if security_score >= 80 else "non-compliant"
}
def generate_recommendations(cost_analysis: Dict, security_analysis: Dict) -> List[Dict]:
"""Generate optimization recommendations"""
recommendations = []
# Cost optimization recommendations
if cost_analysis["total_monthly_estimate"] > 1000:
recommendations.append({
"category": "cost",
"priority": "high",
"title": "High monthly cost detected",
"description": f"Estimated monthly cost: ${cost_analysis['total_monthly_estimate']}",
"actions": [
"Consider using Reserved Instances for consistent workloads",
"Review instance types and downsize if possible",
"Implement auto-scaling to optimize resource usage"
]
})
# Security recommendations
if security_analysis["security_score"] < 80:
recommendations.append({
"category": "security",
"priority": "critical",
"title": "Security improvements needed",
"description": f"Security score: {security_analysis['security_score']}/100",
"actions": [
"Address all critical and high severity security issues",
"Enable encryption for all data at rest",
"Review and restrict security group rules"
]
})
return recommendations
def get_ec2_cost(instance_type: str) -> float:
"""Get the hourly cost of an EC2 instance (simplified)"""
costs = {
't3.micro': 0.0104,
't3.small': 0.0208,
't3.medium': 0.0416,
't3.large': 0.0832,
'm5.large': 0.096,
'm5.xlarge': 0.192
}
return costs.get(instance_type, 0.0416) # Default to t3.medium price
def get_rds_cost(instance_class: str) -> float:
"""Get the hourly cost of an RDS instance (simplified)"""
costs = {
'db.t3.micro': 0.017,
'db.t3.small': 0.034,
'db.t3.medium': 0.068,
'db.r5.large': 0.24,
'db.r5.xlarge': 0.48
}
return costs.get(instance_class, 0.068) # Default to db.t3.medium price
@cdk_ext.command()
@click.option('--stack-name', help='Stack name to optimize')
def optimize(stack_name: str):
"""Optimize a CDK Stack configuration"""
click.echo(f"Optimizing Stack: {stack_name}")
# Run the analysis
result = subprocess.run(['python', '-m', 'cdk_extensions.cli_plugin',
'analyze', '--stack-name', stack_name],
capture_output=True, text=True)
if result.returncode != 0:
click.echo(f"Error running analysis: {result.stderr}", err=True)
return
# Read the analysis results
with open('analysis.json', 'r') as f:
analysis = json.load(f)
# Generate optimization recommendations
recommendations = analysis.get('recommendations', [])
click.echo("\nOptimization Recommendations:")
click.echo("=" * 50)
for i, rec in enumerate(recommendations, 1):
click.echo(f"{i}. {rec['title']} ({rec['priority']})")
click.echo(f" {rec['description']}")
click.echo(" Recommended actions:")
for action in rec['actions']:
click.echo(f" - {action}")
click.echo()
@cdk_ext.command()
@click.option('--template-name', help='Template name to create')
@click.option('--language', default='python', help='Programming language')
def create_template(template_name: str, language: str):
"""Create a CDK project template"""
click.echo(f"Creating template: {template_name} ({language})")
# Template directory structure
templates = {
'web-app': {
'description': 'Multi-tier web application',
'files': [
'app.py',
'requirements.txt',
'cdk.json',
'stacks/__init__.py',
'stacks/web_app_stack.py'
]
},
'serverless-api': {
'description': 'Serverless API with Lambda and API Gateway',
'files': [
'app.py',
'requirements.txt',
'cdk.json',
'stacks/__init__.py',
'stacks/api_stack.py',
'lambda/api.py'
]
},
'data-pipeline': {
'description': 'Data processing pipeline',
'files': [
'app.py',
'requirements.txt',
'cdk.json',
'stacks/__init__.py',
'stacks/pipeline_stack.py'
]
}
}
if template_name not in templates:
click.echo(f"Unknown template: {template_name}")
click.echo(f"Available templates: {list(templates.keys())}")
return
template_info = templates[template_name]
# Create the project directory
project_dir = f"{template_name}-project"
os.makedirs(project_dir, exist_ok=True)
# Create the files
for file_path in template_info['files']:
full_path = os.path.join(project_dir, file_path)
os.makedirs(os.path.dirname(full_path), exist_ok=True)
# Generate file content (simplified)
content = generate_file_content(file_path, template_name, language)
with open(full_path, 'w') as f:
f.write(content)
click.echo(f"Template project created: {project_dir}")
click.echo(f"Description: {template_info['description']}")
def generate_file_content(file_path: str, template_name: str, language: str) -> str:
"""Generate template file content"""
if file_path == 'app.py':
return f'''#!/usr/bin/env python3
import aws_cdk as cdk
from stacks.{template_name.replace('-', '_')}_stack import {template_name.replace('-', ' ').title().replace(' ', '')}Stack
app = cdk.App()
{template_name.replace('-', ' ').title().replace(' ', '')}Stack(app, "{template_name.replace('-', ' ').title().replace(' ', '')}Stack")
app.synth()
'''
elif file_path == 'requirements.txt':
return '''aws-cdk-lib>=2.50.0
constructs>=10.0.0
'''
elif file_path == 'cdk.json':
return '''{
"app": "python app.py",
"watch": {
"include": ["**"],
"exclude": ["README.md", "cdk*.json", "requirements*.txt", "**/__pycache__", "**/*.pyc"]
},
"context": {
"@aws-cdk/aws-lambda:recognizeLayerVersion": true,
"@aws-cdk/core:checkSecretUsage": true
}
}
'''
elif 'stack.py' in file_path:
stack_class = template_name.replace('-', ' ').title().replace(' ', '') + 'Stack'
return f'''import aws_cdk as cdk
from aws_cdk import aws_s3 as s3
from constructs import Construct
class {stack_class}(cdk.Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# TODO: Add your resources here
bucket = s3.Bucket(self, "MyBucket")
cdk.CfnOutput(self, "BucketName", value=bucket.bucket_name)
'''
else:
return f'# {file_path}\n# TODO: Add implementation\n'
if __name__ == '__main__':
cdk_ext()Summary of CDK Ecosystem Best Practices
- Construct Reuse: Prioritize using mature community constructs to avoid reinventing the wheel
- Modular Design: Decompose complex infrastructure into reusable constructs
- Versioning: Establish a clear versioning strategy for custom constructs
- Thorough Documentation: Provide detailed documentation and examples for constructs
- Test Coverage: Ensure constructs have adequate unit and integration tests
- Community Contribution: Actively participate in the open-source community to share and improve constructs
- Tool Integration: Use and develop tools to improve development efficiency
- Standardization: Establish construct development standards for your team or organization
- Monitoring and Tracking: Track the usage and performance of constructs
- Continuous Improvement: Continuously optimize construct design based on user feedback
By completing this chapter, you should be able to fully leverage the rich resources of the CDK ecosystem, create your own reusable constructs, and contribute to the community. The power of CDK lies not only in its technical capabilities but also in its active community and rich ecosystem.
Conclusion
Congratulations on completing the full AWS CDK course! From basic concepts to advanced hands-on projects, from performance optimization to security best practices, you have now mastered the comprehensive skills to build modern cloud infrastructure using CDK.
As a revolutionary Infrastructure as Code tool, CDK is changing the way we build and manage cloud infrastructure. We hope this course will help you succeed in your cloud-native architecture and DevOps practices.
Remember, learning is a continuous process. As AWS services continue to evolve and the CDK ecosystem becomes richer, maintaining a passion for learning and practice, and staying closely connected with the community, will keep you competitive in this rapidly developing field.
We wish you the best on your journey with cloud infrastructure!