AWS 云资源实战大全

VPC 完整网络架构

import pulumi
import pulumi_aws as aws

# ── VPC ──────────────────────────────
vpc = aws.ec2.Vpc("main",
    cidr_block="10.0.0.0/16",
    enable_dns_hostnames=True,
    enable_dns_support=True,
    tags={"Name": "main-vpc"},
)

azs = ["us-east-1a", "us-east-1b"]

# 公共子网
public_subnets = [
    aws.ec2.Subnet(f"public-{az}",
        vpc_id=vpc.id, cidr_block=f"10.0.{i}.0/24",
        availability_zone=az, map_public_ip_on_launch=True,
        tags={"Name": f"public-{az}"},
    ) for i, az in enumerate(azs)
]

# 私有子网
private_subnets = [
    aws.ec2.Subnet(f"private-{az}",
        vpc_id=vpc.id, cidr_block=f"10.0.{i + 10}.0/24",
        availability_zone=az,
        tags={"Name": f"private-{az}"},
    ) for i, az in enumerate(azs)
]

# Internet Gateway → 公共子网出口
igw = aws.ec2.InternetGateway("igw", vpc_id=vpc.id)

public_rt = aws.ec2.RouteTable("public-rt",
    vpc_id=vpc.id,
    routes=[aws.ec2.RouteTableRouteArgs(
        cidr_block="0.0.0.0/0", gateway_id=igw.id
    )],
)
for i, s in enumerate(public_subnets):
    aws.ec2.RouteTableAssociation(f"public-rta-{i}",
        subnet_id=s.id, route_table_id=public_rt.id)

# NAT Gateway → 私有子网出口（每个 AZ 一个）
eips = [aws.ec2.Eip(f"nat-eip-{i}", domain="vpc") for i in range(len(azs))]
nat_gws = [
    aws.ec2.NatGateway(f"nat-{i}",
        subnet_id=public_subnets[i].id,
        allocation_id=eips[i].id,
    ) for i in range(len(azs))
]
for i, (s, ng) in enumerate(zip(private_subnets, nat_gws)):
    rt = aws.ec2.RouteTable(f"private-rt-{i}",
        vpc_id=vpc.id,
        routes=[aws.ec2.RouteTableRouteArgs(
            cidr_block="0.0.0.0/0", nat_gateway_id=ng.id
        )],
    )
    aws.ec2.RouteTableAssociation(f"private-rta-{i}",
        subnet_id=s.id, route_table_id=rt.id)

ALB + Auto Scaling Group

# ALB（Application Load Balancer）
alb = aws.lb.LoadBalancer("web-alb",
    load_balancer_type="application",
    subnets=[s.id for s in public_subnets],
    security_groups=[web_sg.id],
    tags={"Name": "web-alb"},
)

# Target Group
tg = aws.lb.TargetGroup("web-tg",
    port=80, protocol="HTTP",
    vpc_id=vpc.id,
    health_check=aws.lb.TargetGroupHealthCheckArgs(
        path="/health", protocol="HTTP",
        healthy_threshold=2, unhealthy_threshold=3,
    ),
)

# Listener：HTTP → Target Group
listener = aws.lb.Listener("web-listener",
    load_balancer_arn=alb.arn,
    port=80, protocol="HTTP",
    default_actions=[aws.lb.ListenerDefaultActionArgs(
        type="forward", target_group_arn=tg.arn,
    )],
)

# Launch Template
lt = aws.ec2.LaunchTemplate("web-lt",
    image_id="ami-0c02fb55956c7d316",
    instance_type="t3.micro",
    vpc_security_group_ids=[web_sg.id],
    user_data=pulumi.Output.from_input(
        """#!/bin/bash
yum install -y nginx
systemctl start nginx"""
    ).apply(lambda s: __import__("base64").b64encode(s.encode()).decode()),
)

# Auto Scaling Group
asg = aws.autoscaling.Group("web-asg",
    launch_template=aws.autoscaling.GroupLaunchTemplateArgs(
        id=lt.id, version="$Latest"
    ),
    min_size=2, max_size=10, desired_capacity=2,
    vpc_zone_identifiers=[s.id for s in private_subnets],
    target_group_arns=[tg.arn],
    health_check_type="ELB",
    tags=[aws.autoscaling.GroupTagArgs(
        key="Name", value="web-server", propagate_at_launch=True,
    )],
)

S3 存储桶高级配置

# 带版本控制和生命周期的 S3 Bucket
bucket = aws.s3.BucketV2("data-bucket",
    tags={"Purpose": "data-storage"},
)

# 开启版本控制
aws.s3.BucketVersioningV2("data-versioning",
    bucket=bucket.id,
    versioning_configuration=aws.s3.BucketVersioningV2VersioningConfigurationArgs(
        status="Enabled"
    ),
)

# 生命周期规则（30天移到 IA，90天移到 Glacier）
aws.s3.BucketLifecycleConfigurationV2("data-lifecycle",
    bucket=bucket.id,
    rules=[aws.s3.BucketLifecycleConfigurationV2RuleArgs(
        id="archive",
        status="Enabled",
        transitions=[
            aws.s3.BucketLifecycleConfigurationV2RuleTransitionArgs(
                days=30, storage_class="STANDARD_IA"
            ),
            aws.s3.BucketLifecycleConfigurationV2RuleTransitionArgs(
                days=90, storage_class="GLACIER"
            ),
        ],
    )],
)

# 服务端加密（SSE-S3）
aws.s3.BucketServerSideEncryptionConfigurationV2("data-sse",
    bucket=bucket.id,
    rules=[aws.s3.BucketServerSideEncryptionConfigurationV2RuleArgs(
        apply_server_side_encryption_by_default=aws.s3.BucketServerSideEncryptionConfigurationV2RuleApplyServerSideEncryptionByDefaultArgs(
            sse_algorithm="AES256"
        )
    )],
)

Lambda + API Gateway Serverless

import json, pulumi, pulumi_aws as aws

# Lambda 执行角色
lambda_role = aws.iam.Role("lambda-role",
    assume_role_policy=json.dumps({
        "Version": "2012-10-17",
        "Statement": [{
            "Effect": "Allow",
            "Principal": {"Service": "lambda.amazonaws.com"},
            "Action": "sts:AssumeRole",
        }]
    })
)
aws.iam.RolePolicyAttachment("lambda-basic",
    role=lambda_role.name,
    policy_arn="arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole",
)

# Lambda 函数
fn = aws.lambda_.Function("api-handler",
    runtime=aws.lambda_.Runtime.PYTHON3D12,
    code=pulumi.FileArchive("./lambda"),   # 打包目录
    handler="handler.main",
    role=lambda_role.arn,
    environment=aws.lambda_.FunctionEnvironmentArgs(
        variables={"STAGE": "prod"}
    ),
    timeout=30,
    memory_size=256,
)

# API Gateway v2（HTTP API）
api = aws.apigatewayv2.Api("http-api",
    protocol_type="HTTP",
    cors_configuration=aws.apigatewayv2.ApiCorsConfigurationArgs(
        allow_origins=["*"],
        allow_methods=["GET", "POST"],
    ),
)

integration = aws.apigatewayv2.Integration("lambda-int",
    api_id=api.id,
    integration_type="AWS_PROXY",
    integration_uri=fn.invoke_arn,
)

aws.apigatewayv2.Route("default-route",
    api_id=api.id,
    route_key="$default",
    target=integration.id.apply(lambda id: f"integrations/{id}"),
)

aws.apigatewayv2.Stage("default-stage",
    api_id=api.id,
    name="$default",
    auto_deploy=True,
)

# 允许 API Gateway 调用 Lambda
aws.lambda_.Permission("apigw-invoke",
    action="lambda:InvokeFunction",
    function=fn.name,
    principal="apigateway.amazonaws.com",
    source_arn=api.execution_arn.apply(lambda arn: f"{arn}/*"),
)

pulumi.export("api_endpoint", api.api_endpoint)

实战：三层 Web 架构（ALB + EC2 + RDS）

# RDS 安全组（只允许来自 App 层的流量）
db_sg = aws.ec2.SecurityGroup("db-sg",
    vpc_id=vpc.id,
    ingress=[aws.ec2.SecurityGroupIngressArgs(
        from_port=5432, to_port=5432, protocol="tcp",
        security_groups=[web_sg.id],   # 只允许 web_sg 的流量
    )],
)

# RDS 子网组（数据库专用私有子网）
db_subnet_group = aws.rds.SubnetGroup("db-subnets",
    subnet_ids=[s.id for s in private_subnets],
)

# RDS PostgreSQL 实例
db = aws.rds.Instance("app-db",
    engine="postgres",
    engine_version="15.4",
    instance_class="db.r5.large",
    allocated_storage=100,
    max_allocated_storage=1000,  # 自动扩容上限
    storage_encrypted=True,
    db_name="appdb",
    username="admin",
    password=db_password,
    db_subnet_group_name=db_subnet_group.name,
    vpc_security_group_ids=[db_sg.id],
    multi_az=True,                  # 高可用
    backup_retention_period=7,
    deletion_protection=True,
    skip_final_snapshot=False,
    final_snapshot_identifier="app-db-final-snapshot",
    opts=pulumi.ResourceOptions(protect=True),
)

pulumi.export("alb_dns", alb.dns_name)
pulumi.export("db_endpoint", db.address)
pulumi.export("web_url", pulumi.Output.concat("http://", alb.dns_name))

本章小结