Chapter 8: Real-world Project: Full-Stack Web Application Deployment
Haiyue
55min
Chapter 8: Real-world Project: Full-Stack Web Application Deployment
Learning Objectives
- Build a complete full-stack web application infrastructure
- Master front-end static asset hosting and CDN configuration
- Implement a serverless architecture for back-end API services
- Configure database and caching layers
- Implement user authentication and authorization systems
- Add monitoring, logging, and alarming features
Project Overview
We will build a modern full-stack web application, including:
🔄 正在渲染 Mermaid 图表...
Project Architecture Features
- Front-end: React SPA + TypeScript, deployed to S3 + CloudFront
- Back-end: Lambda + API Gateway, serverless architecture
- Database: RDS PostgreSQL primary + ElastiCache Redis cache
- Authentication: AWS Cognito User Pool
- Storage: S3 file storage + Lambda image processing
- Monitoring: CloudWatch + SNS alarms
Infrastructure Layer Implementation
Network Infrastructure
# stacks/network_stack.py
import aws_cdk as cdk
from aws_cdk import (
aws_ec2 as ec2,
aws_logs as logs
)
from constructs import Construct
class NetworkStack(cdk.Stack):
"""Network Infrastructure Stack"""
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# VPC
self.vpc = ec2.Vpc(
self,
"WebAppVpc",
ip_addresses=ec2.IpAddresses.cidr("10.0.0.0/16"),
max_azs=3,
nat_gateways=2, # High availability
subnet_configuration=[
# Public subnets - ALB, NAT Gateway
ec2.SubnetConfiguration(
cidr_mask=24,
name="PublicSubnet",
subnet_type=ec2.SubnetType.PUBLIC
),
# Private subnets - Lambda, RDS
ec2.SubnetConfiguration(
cidr_mask=24,
name="PrivateSubnet",
subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS
),
# Isolated subnets - Database
ec2.SubnetConfiguration(
cidr_mask=28,
name="DatabaseSubnet",
subnet_type=ec2.SubnetType.PRIVATE_ISOLATED
)
],
# VPC Flow Logs
flow_logs={
"FlowLog": ec2.FlowLogOptions(
traffic_type=ec2.FlowLogTrafficType.ALL,
destination=ec2.FlowLogDestination.to_cloud_watch_logs(
logs.LogGroup(
self,
"VpcFlowLogGroup",
retention=logs.RetentionDays.ONE_MONTH
)
)
)
}
)
# VPC Endpoints - Reduce NAT Gateway costs
self.vpc.add_gateway_endpoint(
"S3Endpoint",
service=ec2.GatewayVpcEndpointAwsService.S3
)
self.vpc.add_gateway_endpoint(
"DynamoDbEndpoint",
service=ec2.GatewayVpcEndpointAwsService.DYNAMODB
)
# Interface Endpoints
self.vpc.add_interface_endpoint(
"LambdaEndpoint",
service=ec2.InterfaceVpcEndpointAwsService.LAMBDA
)
# Security groups
self.lambda_security_group = ec2.SecurityGroup(
self,
"LambdaSecurityGroup",
vpc=self.vpc,
description="Security group for Lambda functions",
allow_all_outbound=True
)
self.database_security_group = ec2.SecurityGroup(
self,
"DatabaseSecurityGroup",
vpc=self.vpc,
description="Security group for RDS database",
allow_all_outbound=False
)
self.cache_security_group = ec2.SecurityGroup(
self,
"CacheSecurityGroup",
vpc=self.vpc,
description="Security group for ElastiCache",
allow_all_outbound=False
)
# Security group rules
self.database_security_group.add_ingress_rule(
peer=self.lambda_security_group,
connection=ec2.Port.tcp(5432),
description="Allow Lambda to access PostgreSQL"
)
self.cache_security_group.add_ingress_rule(
peer=self.lambda_security_group,
connection=ec2.Port.tcp(6379),
description="Allow Lambda to access Redis"
)
# Outputs
cdk.CfnOutput(self, "VpcId", value=self.vpc.vpc_id)
cdk.CfnOutput(self, "VpcCidr", value=self.vpc.vpc_cidr_block)Data Storage Layer
# stacks/data_stack.py
import aws_cdk as cdk
from aws_cdk import (
aws_rds as rds,
aws_elasticache as elasticache,
aws_s3 as s3,
aws_s3_notifications as s3n,
aws_lambda as lambda_,
aws_iam as iam,
aws_secretsmanager as secrets
)
from constructs import Construct
from .network_stack import NetworkStack
class DataStack(cdk.Stack):
"""Data Storage Stack"""
def __init__(self, scope: Construct, construct_id: str,
network_stack: NetworkStack, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
self.network_stack = network_stack
# Database credentials
self.db_credentials = rds.DatabaseSecret(
self,
"DatabaseSecret",
username="webapp_user",
secret_name="webapp-db-credentials"
)
# RDS Subnet Group
db_subnet_group = rds.SubnetGroup(
self,
"DatabaseSubnetGroup",
description="Subnet group for RDS database",
vpc=network_stack.vpc,
vpc_subnets=ec2.SubnetSelection(
subnet_type=ec2.SubnetType.PRIVATE_ISOLATED
)
)
# RDS PostgreSQL Database
self.database = rds.DatabaseInstance(
self,
"PostgreSQLDatabase",
engine=rds.DatabaseInstanceEngine.postgres(
version=rds.PostgresEngineVersion.VER_14_9
),
instance_type=ec2.InstanceType.of(
ec2.InstanceClass.BURSTABLE3,
ec2.InstanceSize.MICRO
),
vpc=network_stack.vpc,
subnet_group=db_subnet_group,
security_groups=[network_stack.database_security_group],
credentials=rds.Credentials.from_secret(self.db_credentials),
database_name="webapp",
allocated_storage=20,
max_allocated_storage=100,
storage_encrypted=True,
multi_az=False, # Set to True for high availability
backup_retention=cdk.Duration.days(7),
delete_automated_backups=True,
deletion_protection=False, # Should be True in production
monitoring_interval=cdk.Duration.seconds(60),
performance_insights_enabled=True,
cloudwatch_logs_exports=["postgresql"],
# Parameter group configuration
parameter_group=rds.ParameterGroup(
self,
"DatabaseParameterGroup",
engine=rds.DatabaseInstanceEngine.postgres(
version=rds.PostgresEngineVersion.VER_14_9
),
parameters={
"shared_preload_libraries": "pg_stat_statements",
"log_statement": "all",
"log_min_duration_statement": "1000" # Log queries longer than 1 second
}
)
)
# ElastiCache Subnet Group
cache_subnet_group = elasticache.CfnSubnetGroup(
self,
"CacheSubnetGroup",
description="Subnet group for ElastiCache",
subnet_ids=network_stack.vpc.select_subnets(
subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS
).subnet_ids
)
# ElastiCache Redis
self.cache_cluster = elasticache.CfnCacheCluster(
self,
"RedisCache",
cache_node_type="cache.t3.micro",
engine="redis",
num_cache_nodes=1,
cache_subnet_group_name=cache_subnet_group.ref,
vpc_security_group_ids=[network_stack.cache_security_group.security_group_id],
# Redis configuration
engine_version="7.0",
port=6379,
# Snapshots and backups
snapshot_retention_limit=5,
snapshot_window="03:00-05:00",
preferred_maintenance_window="sun:05:00-sun:07:00",
# Notifications
notification_topic_arn=None # SNS topic can be added
)
# S3 Bucket - File Storage
self.file_bucket = s3.Bucket(
self,
"FileBucket",
bucket_name=f"webapp-files-{self.account}-{self.region}",
versioned=True,
encryption=s3.BucketEncryption.S3_MANAGED,
# CORS configuration
cors=[
s3.CorsRule(
allowed_methods=[s3.HttpMethods.GET, s3.HttpMethods.POST, s3.HttpMethods.PUT],
allowed_origins=["*"], # Production should restrict domains
allowed_headers=["*"],
max_age=3000
)
],
# Lifecycle rules
lifecycle_rules=[
s3.LifecycleRule(
id="DeleteIncompleteMultipartUploads",
enabled=True,
abort_incomplete_multipart_upload_after=cdk.Duration.days(1)
),
s3.LifecycleRule(
id="TransitionToIA",
enabled=True,
transitions=[
s3.Transition(
storage_class=s3.StorageClass.INFREQUENT_ACCESS,
transition_after=cdk.Duration.days(30)
),
s3.Transition(
storage_class=s3.StorageClass.GLACIER,
transition_after=cdk.Duration.days(90)
)
]
)
],
removal_policy=cdk.RemovalPolicy.DESTROY # Should be RETAIN in production
)
# Image Processing Lambda
self.image_processor = lambda_.Function(
self,
"ImageProcessor",
runtime=lambda_.Runtime.PYTHON_3_9,
handler="image_processor.handler",
code=lambda_.Code.from_asset("lambda/image_processor"),
timeout=cdk.Duration.seconds(30),
memory_size=1024, # Image processing requires more memory
environment={
"DESTINATION_BUCKET": self.file_bucket.bucket_name
}
)
# S3 Event Notifications
self.file_bucket.add_event_notification(
s3.EventType.OBJECT_CREATED,
s3n.LambdaDestination(self.image_processor),
s3.NotificationKeyFilter(prefix="uploads/", suffix=".jpg")
)
self.file_bucket.add_event_notification(
s3.EventType.OBJECT_CREATED,
s3n.LambdaDestination(self.image_processor),
s3.NotificationKeyFilter(prefix="uploads/", suffix=".png")
)
# Grant Lambda access to S3
self.file_bucket.grant_read_write(self.image_processor)
# Outputs
cdk.CfnOutput(self, "DatabaseEndpoint", value=self.database.instance_endpoint.hostname)
cdk.CfnOutput(self, "DatabasePort", value=self.database.instance_endpoint.port)
cdk.CfnOutput(self, "DatabaseSecretArn", value=self.db_credentials.secret_arn)
cdk.CfnOutput(self, "CacheEndpoint", value=self.cache_cluster.attr_cache_nodes[0].address)
cdk.CfnOutput(self, "FileBucketName", value=self.file_bucket.bucket_name)Authentication Service Layer
# stacks/auth_stack.py
import aws_cdk as cdk
from aws_cdk import (
aws_cognito as cognito,
aws_lambda as lambda_,
aws_iam as iam,
aws_logs as logs
)
from constructs import Construct
class AuthStack(cdk.Stack):
"""Authentication Service Stack"""
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# Lambda Trigger - Post-confirmation user registration
self.post_confirmation_trigger = lambda_.Function(
self,
"PostConfirmationTrigger",
runtime=lambda_.Runtime.PYTHON_3_9,
handler="post_confirmation.handler",
code=lambda_.Code.from_asset("lambda/auth_triggers"),
timeout=cdk.Duration.seconds(30),
log_retention=logs.RetentionDays.ONE_WEEK
)
# Lambda Trigger - Custom authentication flow
self.pre_authentication_trigger = lambda_.Function(
self,
"PreAuthenticationTrigger",
runtime=lambda_.Runtime.PYTHON_3_9,
handler="pre_authentication.handler",
code=lambda_.Code.from_asset("lambda/auth_triggers"),
timeout=cdk.Duration.seconds(30),
log_retention=logs.RetentionDays.ONE_WEEK
)
# Cognito User Pool
self.user_pool = cognito.UserPool(
self,
"WebAppUserPool",
user_pool_name="webapp-users",
# Sign-up configuration
sign_in_aliases=cognito.SignInAliases(
username=True,
email=True
),
# User attributes
standard_attributes=cognito.StandardAttributes(
email=cognito.StandardAttribute(
required=True,
mutable=True
),
given_name=cognito.StandardAttribute(
required=True,
mutable=True
),
family_name=cognito.StandardAttribute(
required=True,
mutable=True
)
),
# Custom attributes
custom_attributes={
"role": cognito.StringAttribute(min_len=1, max_len=50, mutable=True),
"organization": cognito.StringAttribute(min_len=1, max_len=100, mutable=True)
},
# Password policy
password_policy=cognito.PasswordPolicy(
min_length=8,
require_lowercase=True,
require_uppercase=True,
require_digits=True,
require_symbols=True
),
# Account recovery
account_recovery=cognito.AccountRecovery.EMAIL_ONLY,
# Email verification
email_verification_subject="Verify Your WebApp Account",
email_verification_body="Thank you for signing up for WebApp! Please click the link to verify your email: {##Verify Email##}",
# User invitation
user_invitation=cognito.UserInvitationConfig(
email_subject="Welcome to WebApp",
email_body="You have been invited to WebApp. Username: {username}, Temporary password: {####}",
sms_message="Your WebApp username: {username}, Temporary password: {####}"
),
# Lambda triggers
lambda_triggers=cognito.UserPoolTriggers(
post_confirmation=self.post_confirmation_trigger,
pre_authentication=self.pre_authentication_trigger
),
# MFA configuration
mfa=cognito.Mfa.OPTIONAL,
mfa_second_factor=cognito.MfaSecondFactor(
sms=True,
otp=True # Time-based One Time Password
),
# Device tracking
device_tracking=cognito.DeviceTracking(
challenge_required_on_new_device=True,
device_only_remembered_on_user_prompt=True
),
removal_policy=cdk.RemovalPolicy.DESTROY # Should be RETAIN in production
)
# User Pool Domain
self.user_pool_domain = cognito.UserPoolDomain(
self,
"WebAppUserPoolDomain",
user_pool=self.user_pool,
cognito_domain=cognito.CognitoDomainOptions(
domain_prefix="webapp-auth-domain" # Must be globally unique
)
)
# User Pool Client - Web Application
self.web_client = cognito.UserPoolClient(
self,
"WebAppClient",
user_pool=self.user_pool,
user_pool_client_name="webapp-web-client",
# OAuth configuration
o_auth=cognito.OAuthSettings(
flows=cognito.OAuthFlows(
authorization_code_grant=True,
implicit_code_grant=True
),
scopes=[
cognito.OAuthScope.EMAIL,
cognito.OAuthScope.OPENID,
cognito.OAuthScope.PROFILE
],
callback_urls=[
"http://localhost:3000/auth/callback", # Development environment
"https://webapp.example.com/auth/callback" # Production environment
],
logout_urls=[
"http://localhost:3000/auth/logout",
"https://webapp.example.com/auth/logout"
]
),
# Authentication flows
auth_flows=cognito.AuthFlow(
user_password=True,
user_srp=True
),
# Token validity
access_token_validity=cdk.Duration.hours(1),
id_token_validity=cdk.Duration.hours(1),
refresh_token_validity=cdk.Duration.days(30),
# Prevent user existence enumeration attacks
prevent_user_existence_errors=True
)
# User Pool Client - Mobile Application
self.mobile_client = cognito.UserPoolClient(
self,
"WebAppMobileClient",
user_pool=self.user_pool,
user_pool_client_name="webapp-mobile-client",
generate_secret=True, # Mobile applications need client secrets
auth_flows=cognito.AuthFlow(
user_password=True,
user_srp=True,
custom=True # Supports custom authentication flows
),
# Token validity (mobile apps can have longer)
access_token_validity=cdk.Duration.hours(8),
id_token_validity=cdk.Duration.hours(8),
refresh_token_validity=cdk.Duration.days(60)
)
# Identity Pool - for AWS resource access
self.identity_pool = cognito.CfnIdentityPool(
self,
"WebAppIdentityPool",
identity_pool_name="webapp_identity_pool",
allow_unauthenticated_identities=False,
cognito_identity_providers=[
cognito.CfnIdentityPool.CognitoIdentityProviderProperty(
client_id=self.web_client.user_pool_client_id,
provider_name=self.user_pool.user_pool_provider_name
)
]
)
# Identity Pool Roles
# Authenticated user role
self.authenticated_role = iam.Role(
self,
"AuthenticatedRole",
assumed_by=iam.FederatedPrincipal(
"cognito-identity.amazonaws.com",
{
"StringEquals": {
"cognito-identity.amazonaws.com:aud": self.identity_pool.ref
},
"ForAnyValue:StringLike": {
"cognito-identity.amazonaws.com:amr": "authenticated"
}
},
"sts:AssumeRoleWithWebIdentity"
)
)
# Add basic permissions for authenticated users
self.authenticated_role.add_to_policy(
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"cognito-sync:*",
"cognito-identity:*"
],
resources=["*"]
)
)
# S3 access permissions (based on user identity)
self.authenticated_role.add_to_policy(
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"s3:GetObject",
"s3:PutObject",
"s3:DeleteObject"
],
resources=[
f"arn:aws:s3:::webapp-files-{self.account}-{self.region}/users/${{cognito-identity.amazonaws.com:sub}}/*"
]
)
)
# Identity Pool role mapping
cognito.CfnIdentityPoolRoleAttachment(
self,
"IdentityPoolRoleAttachment",
identity_pool_id=self.identity_pool.ref,
roles={
"authenticated": self.authenticated_role.role_arn
}
)
# Outputs
cdk.CfnOutput(self, "UserPoolId", value=self.user_pool.user_pool_id)
cdk.CfnOutput(self, "UserPoolArn", value=self.user_pool.user_pool_arn)
cdk.CfnOutput(self, "WebClientId", value=self.web_client.user_pool_client_id)
cdk.CfnOutput(self, "MobileClientId", value=self.mobile_client.user_pool_client_id)
cdk.CfnOutput(self, "IdentityPoolId", value=self.identity_pool.ref)
cdk.CfnOutput(self, "UserPoolDomainUrl",
value=f"https://{self.user_pool_domain.domain_name}.auth.{self.region}.amazoncognito.com")API Service Layer Implementation
Lambda Function Layer
# stacks/api_stack.py
import aws_cdk as cdk
from aws_cdk import (
aws_lambda as lambda_,
aws_apigateway as apigateway,
aws_iam as iam,
aws_logs as logs,
aws_ssm as ssm
)
from constructs import Construct
from .network_stack import NetworkStack
from .data_stack import DataStack
from .auth_stack import AuthStack
class ApiStack(cdk.Stack):
"""API Service Stack"""
def __init__(self, scope: Construct, construct_id: str,
network_stack: NetworkStack,
data_stack: DataStack,
auth_stack: AuthStack, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
self.network_stack = network_stack
self.data_stack = data_stack
self.auth_stack = auth_stack
# Lambda Layer - Shared libraries
self.shared_layer = lambda_.LayerVersion(
self,
"SharedLayer",
code=lambda_.Code.from_asset("lambda/layers/shared"),
compatible_runtimes=[lambda_.Runtime.PYTHON_3_9],
description="Shared libraries for Lambda functions"
)
# Lambda Function Environment Variables
common_environment = {
"DB_SECRET_ARN": data_stack.db_credentials.secret_arn,
"REDIS_ENDPOINT": data_stack.cache_cluster.attr_cache_nodes[0].address,
"REDIS_PORT": "6379",
"FILE_BUCKET": data_stack.file_bucket.bucket_name,
"USER_POOL_ID": auth_stack.user_pool.user_pool_id,
"LOG_LEVEL": "INFO"
}
# User Management Lambda
self.user_function = lambda_.Function(
self,
"UserFunction",
runtime=lambda_.Runtime.PYTHON_3_9,
handler="user.handler",
code=lambda_.Code.from_asset("lambda/api/user"),
layers=[self.shared_layer],
vpc=network_stack.vpc,
security_groups=[network_stack.lambda_security_group],
timeout=cdk.Duration.seconds(30),
memory_size=256,
environment=common_environment,
log_retention=logs.RetentionDays.ONE_MONTH
)
# File Management Lambda
self.file_function = lambda_.Function(
self,
"FileFunction",
runtime=lambda_.Runtime.PYTHON_3_9,
handler="file.handler",
code=lambda_.Code.from_asset("lambda/api/file"),
layers=[self.shared_layer],
vpc=network_stack.vpc,
security_groups=[network_stack.lambda_security_group],
timeout=cdk.Duration.seconds(30),
memory_size=512, # File processing requires more memory
environment=common_environment,
log_retention=logs.RetentionDays.ONE_MONTH
)
# Business Logic Lambda
self.business_function = lambda_.Function(
self,
"BusinessFunction",
runtime=lambda_.Runtime.PYTHON_3_9,
handler="business.handler",
code=lambda_.Code.from_asset("lambda/api/business"),
layers=[self.shared_layer],
vpc=network_stack.vpc,
security_groups=[network_stack.lambda_security_group],
timeout=cdk.Duration.seconds(60),
memory_size=512,
environment=common_environment,
log_retention=logs.RetentionDays.ONE_MONTH
)
# Lambda Authorizer
self.authorizer_function = lambda_.Function(
self,
"AuthorizerFunction",
runtime=lambda_.Runtime.PYTHON_3_9,
handler="authorizer.handler",
code=lambda_.Code.from_asset("lambda/authorizer"),
layers=[self.shared_layer],
timeout=cdk.Duration.seconds(30),
memory_size=256,
environment={
"USER_POOL_ID": auth_stack.user_pool.user_pool_id,
"APP_CLIENT_ID": auth_stack.web_client.user_pool_client_id
},
log_retention=logs.RetentionDays.ONE_MONTH
)
# Grant Lambda access to database secrets
data_stack.db_credentials.grant_read(self.user_function)
data_stack.db_credentials.grant_read(self.file_function)
data_stack.db_credentials.grant_read(self.business_function)
# Grant Lambda access to S3
data_stack.file_bucket.grant_read_write(self.file_function)
# Grant Lambda access to Cognito
self.user_function.add_to_role_policy(
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"cognito-idp:AdminGetUser",
"cognito-idp:AdminUpdateUserAttributes",
"cognito-idp:ListUsers"
],
resources=[auth_stack.user_pool.user_pool_arn]
)
)
# API Gateway
self.api = apigateway.RestApi(
self,
"WebAppAPI",
rest_api_name="webapp-api",
description="Web Application REST API",
# CORS configuration
default_cors_preflight_options=apigateway.CorsOptions(
allow_origins=apigateway.Cors.ALL_ORIGINS, # Production should restrict domains
allow_methods=apigateway.Cors.ALL_METHODS,
allow_headers=["Content-Type", "X-Amz-Date", "Authorization", "X-Api-Key"]
),
# API Key configuration
api_key_source_type=apigateway.ApiKeySourceType.HEADER,
# Deployment options
deploy_options=apigateway.StageOptions(
stage_name="v1",
throttling_rate_limit=1000,
throttling_burst_limit=2000,
logging_level=apigateway.MethodLoggingLevel.INFO,
data_trace_enabled=True,
metrics_enabled=True
)
)
# Lambda Authorizer
self.lambda_authorizer = apigateway.TokenAuthorizer(
self,
"LambdaAuthorizer",
handler=self.authorizer_function,
token_source=apigateway.IdentitySource.header("Authorization"),
results_cache_ttl=cdk.Duration.minutes(5)
)
# API Resources and Methods
# /users resource
users_resource = self.api.root.add_resource("users")
users_resource.add_method(
"GET",
apigateway.LambdaIntegration(self.user_function),
authorizer=self.lambda_authorizer,
api_key_required=True
)
users_resource.add_method(
"POST",
apigateway.LambdaIntegration(self.user_function),
authorizer=self.lambda_authorizer,
api_key_required=True
)
# /users/{id} resource
user_resource = users_resource.add_resource("{id}")
user_resource.add_method(
"GET",
apigateway.LambdaIntegration(self.user_function),
authorizer=self.lambda_authorizer,
api_key_required=True
)
user_resource.add_method(
"PUT",
apigateway.LambdaIntegration(self.user_function),
authorizer=self.lambda_authorizer,
api_key_required=True
)
user_resource.add_method(
"DELETE",
apigateway.LambdaIntegration(self.user_function),
authorizer=self.lambda_authorizer,
api_key_required=True
)
# /files resource
files_resource = self.api.root.add_resource("files")
files_resource.add_method(
"POST",
apigateway.LambdaIntegration(self.file_function),
authorizer=self.lambda_authorizer,
api_key_required=True
)
files_resource.add_method(
"GET",
apigateway.LambdaIntegration(self.file_function),
authorizer=self.lambda_authorizer,
api_key_required=True
)
# /business resource
business_resource = self.api.root.add_resource("business")
business_resource.add_method(
"POST",
apigateway.LambdaIntegration(self.business_function),
authorizer=self.lambda_authorizer,
api_key_required=True
)
# API Key and Usage Plan
self.api_key = apigateway.ApiKey(
self,
"WebAppApiKey",
api_key_name="webapp-api-key",
description="API key for web application"
)
self.usage_plan = apigateway.UsagePlan(
self,
"WebAppUsagePlan",
name="webapp-usage-plan",
description="Usage plan for web application",
api_stages=[
apigateway.UsagePlanPerApiStage(
api=self.api,
stage=self.api.deployment_stage
)
],
throttle=apigateway.ThrottleSettings(
rate_limit=1000,
burst_limit=2000
),
quota=apigateway.QuotaSettings(
limit=10000,
period=apigateway.Period.DAY
)
)
self.usage_plan.add_api_key(self.api_key)
# Outputs
cdk.CfnOutput(self, "ApiUrl", value=self.api.url)
cdk.CfnOutput(self, "ApiId", value=self.api.rest_api_id)
cdk.CfnOutput(self, "ApiKeyId", value=self.api_key.key_id)Front-End Infrastructure
Static Site Hosting
# stacks/frontend_stack.py
import aws_cdk as cdk
from aws_cdk import (
aws_s3 as s3,
aws_s3_deployment as s3_deployment,
aws_cloudfront as cloudfront,
aws_cloudfront_origins as origins,
aws_certificatemanager as acm,
aws_route53 as route53,
aws_route53_targets as targets,
aws_iam as iam
)
from constructs import Construct
class FrontendStack(cdk.Stack):
"""Front-End Infrastructure Stack"""
def __init__(self, scope: Construct, construct_id: str,
domain_name: str = None, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
self.domain_name = domain_name
# S3 Bucket - Static website hosting
self.website_bucket = s3.Bucket(
self,
"WebsiteBucket",
bucket_name=f"webapp-frontend-{self.account}-{self.region}",
website_index_document="index.html",
website_error_document="error.html",
public_read_access=False, # Accessed via CloudFront
block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
removal_policy=cdk.RemovalPolicy.DESTROY
)
# Origin Access Identity - Restrict S3 access
self.origin_access_identity = cloudfront.OriginAccessIdentity(
self,
"WebsiteOAI",
comment="OAI for webapp frontend"
)
# Grant CloudFront access to S3
self.website_bucket.add_to_resource_policy(
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
principals=[self.origin_access_identity.grant_principal],
actions=["s3:GetObject"],
resources=[f"{self.website_bucket.bucket_arn}/*"]
)
)
# SSL Certificate (if domain name is provided)
certificate = None
if domain_name:
certificate = acm.Certificate(
self,
"WebsiteCertificate",
domain_name=domain_name,
subject_alternative_names=[f"www.{domain_name}"],
validation=acm.CertificateValidation.from_dns()
)
# CloudFront Distribution
self.distribution = cloudfront.Distribution(
self,
"WebsiteDistribution",
default_behavior=cloudfront.BehaviorOptions(
origin=origins.S3Origin(
self.website_bucket,
origin_access_identity=self.origin_access_identity
),
viewer_protocol_policy=cloudfront.ViewerProtocolPolicy.REDIRECT_TO_HTTPS,
allowed_methods=cloudfront.AllowedMethods.ALLOW_GET_HEAD,
cached_methods=cloudfront.CachedMethods.CACHE_GET_HEAD,
cache_policy=cloudfront.CachePolicy.CACHING_OPTIMIZED,
compress=True
),
additional_behaviors={
# API path - No caching
"/api/*": cloudfront.BehaviorOptions(
origin=origins.HttpOrigin("api.example.com"), # Replace with actual API domain
viewer_protocol_policy=cloudfront.ViewerProtocolPolicy.HTTPS_ONLY,
allowed_methods=cloudfront.AllowedMethods.ALLOW_ALL,
cache_policy=cloudfront.CachePolicy.CACHING_DISABLED,
origin_request_policy=cloudfront.OriginRequestPolicy.CORS_S3_ORIGIN
),
# Static assets - Long-term caching
"/static/*": cloudfront.BehaviorOptions(
origin=origins.S3Origin(
self.website_bucket,
origin_access_identity=self.origin_access_identity
),
viewer_protocol_policy=cloudfront.ViewerProtocolPolicy.HTTPS_ONLY,
cache_policy=cloudfront.CachePolicy.CACHING_OPTIMIZED_FOR_UNCOMPRESSED_OBJECTS,
compress=True
)
},
# Error page configuration
error_responses=[
cloudfront.ErrorResponse(
http_status=404,
response_http_status=200,
response_page_path="/index.html",
ttl=cdk.Duration.minutes(5)
),
cloudfront.ErrorResponse(
http_status=403,
response_http_status=200,
response_page_path="/index.html",
ttl=cdk.Duration.minutes(5)
)
],
# Geo restriction
geo_restriction=cloudfront.GeoRestriction.allowlist("US", "CA", "GB", "DE", "JP", "CN"),
# Price class
price_class=cloudfront.PriceClass.PRICE_CLASS_100,
# Enable IPv6
enable_ipv6=True,
# SSL Configuration
certificate=certificate,
domain_names=[domain_name, f"www.{domain_name}"] if domain_name else None,
# Default root object
default_root_object="index.html"
)
# Route 53 Configuration (if domain name is provided)
if domain_name:
# Get hosted zone
hosted_zone = route53.HostedZone.from_lookup(
self,
"HostedZone",
domain_name=domain_name
)
# A record pointing to CloudFront
route53.ARecord(
self,
"WebsiteARecord",
zone=hosted_zone,
record_name=domain_name,
target=route53.RecordTarget.from_alias(
targets.CloudFrontTarget(self.distribution)
)
)
# www subdomain
route53.ARecord(
self,
"WebsiteWWWARecord",
zone=hosted_zone,
record_name=f"www.{domain_name}",
target=route53.RecordTarget.from_alias(
targets.CloudFrontTarget(self.distribution)
)
)
# Deploy front-end application
s3_deployment.BucketDeployment(
self,
"DeployWebsite",
sources=[s3_deployment.Source.asset("frontend/build")],
destination_bucket=self.website_bucket,
distribution=self.distribution,
distribution_paths=["/*"], # Invalidate all cache
# Retain existing files
prune=False
)
# Outputs
cdk.CfnOutput(
self,
"WebsiteUrl",
value=f"https://{self.distribution.distribution_domain_name}",
description="CloudFront distribution URL"
)
if domain_name:
cdk.CfnOutput(
self,
"CustomDomainUrl",
value=f"https://{domain_name}",
description="Custom domain URL"
)
cdk.CfnOutput(self, "BucketName", value=self.website_bucket.bucket_name)
cdk.CfnOutput(self, "DistributionId", value=self.distribution.distribution_id)Monitoring and Alerting
# stacks/monitoring_stack.py
import aws_cdk as cdk
from aws_cdk import (
aws_cloudwatch as cloudwatch,
aws_sns as sns,
aws_sns_subscriptions as subscriptions,
aws_logs as logs,
aws_logs_destinations as destinations
)
from constructs import Construct
from .api_stack import ApiStack
from .data_stack import DataStack
from .frontend_stack import FrontendStack
class MonitoringStack(cdk.Stack):
"""Monitoring and Alerting Stack"""
def __init__(self, scope: Construct, construct_id: str,
api_stack: ApiStack,
data_stack: DataStack,
frontend_stack: FrontendStack,
alert_email: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# SNS Topic - Alert notifications
self.alert_topic = sns.Topic(
self,
"AlertTopic",
topic_name="webapp-alerts",
display_name="WebApp Alerts"
)
# Add email subscription
self.alert_topic.add_subscription(
subscriptions.EmailSubscription(alert_email)
)
# CloudWatch Dashboard
self.dashboard = cloudwatch.Dashboard(
self,
"WebAppDashboard",
dashboard_name="WebApp-Monitoring",
period_override=cloudwatch.PeriodOverride.AUTO
)
# API Gateway Monitoring
self._create_api_monitoring(api_stack)
# Lambda Monitoring
self._create_lambda_monitoring(api_stack)
# Database Monitoring
self._create_database_monitoring(data_stack)
# CloudFront Monitoring
self._create_cloudfront_monitoring(frontend_stack)
# Custom Business Metrics
self._create_business_monitoring()
def _create_api_monitoring(self, api_stack: ApiStack):
"""API Gateway Monitoring"""
# API Gateway Metrics
api_requests = api_stack.api.metric_count()
api_4xx_errors = api_stack.api.metric_client_error()
api_5xx_errors = api_stack.api.metric_server_error()
api_latency = api_stack.api.metric_latency()
# Add to dashboard
self.dashboard.add_widgets(
cloudwatch.GraphWidget(
title="API Gateway - Requests",
left=[api_requests],
right=[api_4xx_errors, api_5xx_errors],
period=cdk.Duration.minutes(5),
width=12,
height=6
),
cloudwatch.GraphWidget(
title="API Gateway - Latency",
left=[api_latency],
period=cdk.Duration.minutes(5),
width=12,
height=6
)
)
# Alarms
# High error rate alarm
cloudwatch.Alarm(
self,
"ApiHighErrorRateAlarm",
alarm_name="API-High-Error-Rate",
alarm_description="API Gateway error rate is too high",
metric=api_5xx_errors,
threshold=10,
evaluation_periods=2,
datapoints_to_alarm=2,
treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING
).add_alarm_action(
cloudwatch.SnsAction(self.alert_topic)
)
# High latency alarm
cloudwatch.Alarm(
self,
"ApiHighLatencyAlarm",
alarm_name="API-High-Latency",
alarm_description="API Gateway latency is too high",
metric=api_latency,
threshold=5000, # 5 seconds
evaluation_periods=3,
datapoints_to_alarm=2
).add_alarm_action(
cloudwatch.SnsAction(self.alert_topic)
)
def _create_lambda_monitoring(self, api_stack: ApiStack):
"""Lambda Function Monitoring"""
lambda_functions = [
("User Function", api_stack.user_function),
("File Function", api_stack.file_function),
("Business Function", api_stack.business_function)
]
for name, func in lambda_functions:
# Lambda Metrics
invocations = func.metric_invocations()
errors = func.metric_errors()
duration = func.metric_duration()
throttles = func.metric_throttles()
# Add to dashboard
self.dashboard.add_widgets(
cloudwatch.GraphWidget(
title=f"{name} - Metrics",
left=[invocations, throttles],
right=[errors],
period=cdk.Duration.minutes(5),
width=12,
height=6
),
cloudwatch.GraphWidget(
title=f"{name} - Duration",
left=[duration],
period=cdk.Duration.minutes(5),
width=12,
height=6
)
)
# Error rate alarm
cloudwatch.Alarm(
self,
f"{name.replace(' ', '')}ErrorAlarm",
alarm_name=f"Lambda-{name.replace(' ', '-')}-Errors",
alarm_description=f"{name} error rate is too high",
metric=errors,
threshold=5,
evaluation_periods=2,
datapoints_to_alarm=2
).add_alarm_action(
cloudwatch.SnsAction(self.alert_topic)
)
# Execution duration alarm
cloudwatch.Alarm(
self,
f"{name.replace(' ', '')}DurationAlarm",
alarm_name=f"Lambda-{name.replace(' ', '-')}-Duration",
alarm_description=f"{name} execution duration is too high",
metric=duration,
threshold=10000, # 10 seconds
evaluation_periods=3,
datapoints_to_alarm=2
).add_alarm_action(
cloudwatch.SnsAction(self.alert_topic)
)
def _create_database_monitoring(self, data_stack: DataStack):
"""Database Monitoring"""
# RDS Metrics
db_connections = data_stack.database.metric_database_connections()
cpu_utilization = data_stack.database.metric_cpu_utilization()
free_storage_space = data_stack.database.metric_free_storage_space()
# Add to dashboard
self.dashboard.add_widgets(
cloudwatch.GraphWidget(
title="RDS - Database Connections",
left=[db_connections],
period=cdk.Duration.minutes(5),
width=12,
height=6
),
cloudwatch.GraphWidget(
title="RDS - CPU Utilization",
left=[cpu_utilization],
period=cdk.Duration.minutes(5),
width=12,
height=6
),
cloudwatch.GraphWidget(
title="RDS - Free Storage Space",
left=[free_storage_space],
period=cdk.Duration.minutes(5),
width=12,
height=6
)
)
# Database Alarms
# High CPU utilization
cloudwatch.Alarm(
self,
"DatabaseHighCPUAlarm",
alarm_name="Database-High-CPU",
alarm_description="Database CPU utilization is too high",
metric=cpu_utilization,
threshold=80,
evaluation_periods=3,
datapoints_to_alarm=2
).add_alarm_action(
cloudwatch.SnsAction(self.alert_topic)
)
# Low storage space
cloudwatch.Alarm(
self,
"DatabaseLowStorageAlarm",
alarm_name="Database-Low-Storage",
alarm_description="Database free storage space is running low",
metric=free_storage_space,
threshold=1000000000, # 1GB in bytes
comparison_operator=cloudwatch.ComparisonOperator.LESS_THAN_THRESHOLD,
evaluation_periods=2,
datapoints_to_alarm=2
).add_alarm_action(
cloudwatch.SnsAction(self.alert_topic)
)
def _create_cloudfront_monitoring(self, frontend_stack: FrontendStack):
"""CloudFront Monitoring"""
# CloudFront Metrics
requests = cloudwatch.Metric(
namespace="AWS/CloudFront",
metric_name="Requests",
dimensions_map={
"DistributionId": frontend_stack.distribution.distribution_id
}
)
bytes_downloaded = cloudwatch.Metric(
namespace="AWS/CloudFront",
metric_name="BytesDownloaded",
dimensions_map={
"DistributionId": frontend_stack.distribution.distribution_id
}
)
# Add to dashboard
self.dashboard.add_widgets(
cloudwatch.GraphWidget(
title="CloudFront - Requests",
left=[requests],
period=cdk.Duration.minutes(5),
width=12,
height=6
),
cloudwatch.GraphWidget(
title="CloudFront - Bytes Downloaded",
left=[bytes_downloaded],
period=cdk.Duration.minutes(5),
width=12,
height=6
)
)
def _create_business_monitoring(self):
"""Business Metrics Monitoring"""
# Custom metric - User registrations
user_registrations = cloudwatch.Metric(
namespace="WebApp/Business",
metric_name="UserRegistrations",
statistic="Sum"
)
# Custom metric - File uploads
file_uploads = cloudwatch.Metric(
namespace="WebApp/Business",
metric_name="FileUploads",
statistic="Sum"
)
# Add to dashboard
self.dashboard.add_widgets(
cloudwatch.GraphWidget(
title="Business Metrics",
left=[user_registrations],
right=[file_uploads],
period=cdk.Duration.hours(1),
width=12,
height=6
)
)
# Outputs
cdk.CfnOutput(self, "DashboardUrl",
value=f"https://console.aws.amazon.com/cloudwatch/home?region={self.region}#dashboards:name={self.dashboard.dashboard_name}")
cdk.CfnOutput(self, "AlertTopicArn", value=self.alert_topic.topic_arn)Application Integration
Main Application
# app.py
#!/usr/bin/env python3
import aws_cdk as cdk
from stacks.network_stack import NetworkStack
from stacks.data_stack import DataStack
from stacks.auth_stack import AuthStack
from stacks.api_stack import ApiStack
from stacks.frontend_stack import FrontendStack
from stacks.monitoring_stack import MonitoringStack
def main():
app = cdk.App()
# Get configuration from context
environment = app.node.try_get_context("environment") or "dev"
domain_name = app.node.try_get_context("domain_name")
alert_email = app.node.try_get_context("alert_email") or "admin@example.com"
# Common tags
tags = {
"Project": "WebApp",
"Environment": environment,
"Owner": "DevTeam",
"ManagedBy": "CDK"
}
# Apply tags to all resources
for key, value in tags.items():
cdk.Tags.of(app).add(key, value)
# Network Infrastructure
network_stack = NetworkStack(
app,
f"WebApp-Network-{environment}",
stack_name=f"webapp-network-{environment}"
)
# Data Storage
data_stack = DataStack(
app,
f"WebApp-Data-{environment}",
network_stack=network_stack,
stack_name=f"webapp-data-{environment}"
)
# Authentication Service
auth_stack = AuthStack(
app,
f"WebApp-Auth-{environment}",
stack_name=f"webapp-auth-{environment}"
)
# API Service
api_stack = ApiStack(
app,
f"WebApp-API-{environment}",
network_stack=network_stack,
data_stack=data_stack,
auth_stack=auth_stack,
stack_name=f"webapp-api-{environment}"
)
# Frontend Hosting
frontend_stack = FrontendStack(
app,
f"WebApp-Frontend-{environment}",
domain_name=domain_name,
stack_name=f"webapp-frontend-{environment}"
)
# Monitoring and Alerting
monitoring_stack = MonitoringStack(
app,
f"WebApp-Monitoring-{environment}",
api_stack=api_stack,
data_stack=data_stack,
frontend_stack=frontend_stack,
alert_email=alert_email,
stack_name=f"webapp-monitoring-{environment}"
)
# Stack dependencies
data_stack.add_dependency(network_stack)
api_stack.add_dependency(data_stack)
api_stack.add_dependency(auth_stack)
monitoring_stack.add_dependency(api_stack)
monitoring_stack.add_dependency(data_stack)
monitoring_stack.add_dependency(frontend_stack)
app.synth()
if __name__ == "__main__":
main()Deployment and Usage Guide
Deployment Steps
# 1. Install dependencies
pip install -r requirements.txt
npm install -g aws-cdk
# 2. Configure AWS credentials
aws configure
# 3. Bootstrap CDK (if not already done)
cdk bootstrap
# 4. Deploy infrastructure
cdk deploy --all --require-approval never
# 5. Build and deploy frontend
cd frontend
npm run build
cd ..
cdk deploy WebApp-Frontend-devConfiguration File Example
# cdk.context.json
{
"environment": "dev",
"domain_name": "webapp.example.com",
"alert_email": "admin@example.com"
}Project Best Practices Summary
- Modular Design: Separate different functionalities into independent Stacks.
- Security First: Use VPC, security groups, and IAM roles with the principle of least privilege.
- Scalability: Utilize serverless services like Lambda, API Gateway.
- Comprehensive Monitoring: Full-fledged monitoring, logging, and alarming system.
- Cost Optimization: Judicious use of VPC Endpoints, appropriate instance sizes.
- High Availability: Multi-AZ deployments, automatic failover.
- DevOps Integration: Support for CI/CD and Infrastructure as Code.
Through this hands-on project, you should be able to build a complete, production-grade full-stack web application infrastructure, covering all core components of a modern cloud-native application.