Pulumi IaC Infrastructure as Code

Pulumi IaC

Pulumi Infrastructure as Code Python TypeScript Go AWS Azure GCP Kubernetes State Management Stack Testing CI/CD Production Deployment

AWS Infrastructure

# === Pulumi AWS Infrastructure ===

# pip install pulumi pulumi-aws

# import pulumi
# import pulumi_aws as aws
#
# # VPC
# vpc = aws.ec2.Vpc("main-vpc",
#     cidr_block="10.0.0.0/16",
#     enable_dns_hostnames=True,
#     tags={"Name": "main-vpc", "Environment": pulumi.get_stack()})
#
# # Public Subnet
# public_subnet = aws.ec2.Subnet("public-subnet",
#     vpc_id=vpc.id,
#     cidr_block="10.0.1.0/24",
#     availability_zone="ap-southeast-1a",
#     map_public_ip_on_launch=True,
#     tags={"Name": "public-subnet"})
#
# # Security Group
# sg = aws.ec2.SecurityGroup("web-sg",
#     vpc_id=vpc.id,
#     ingress=[
#         {"protocol": "tcp", "from_port": 80, "to_port": 80, "cidr_blocks": ["0.0.0.0/0"]},
#         {"protocol": "tcp", "from_port": 443, "to_port": 443, "cidr_blocks": ["0.0.0.0/0"]},
#     ],
#     egress=[
#         {"protocol": "-1", "from_port": 0, "to_port": 0, "cidr_blocks": ["0.0.0.0/0"]},
#     ],
#     tags={"Name": "web-sg"})
#
# # EC2 Instance
# instance = aws.ec2.Instance("web-server",
#     instance_type="t3.micro",
#     ami="ami-0abcdef1234567890",
#     subnet_id=public_subnet.id,
#     vpc_security_group_ids=[sg.id],
#     tags={"Name": "web-server", "Environment": pulumi.get_stack()})
#
# # RDS
# db = aws.rds.Instance("app-db",
#     engine="postgres",
#     engine_version="15",
#     instance_class="db.t3.micro",
#     allocated_storage=20,
#     db_name="appdb",
#     username="admin",
#     password=pulumi.Config().require_secret("db_password"),
#     skip_final_snapshot=True,
#     tags={"Name": "app-db"})
#
# # Outputs
# pulumi.export("vpc_id", vpc.id)
# pulumi.export("instance_ip", instance.public_ip)
# pulumi.export("db_endpoint", db.endpoint)

from dataclasses import dataclass

@dataclass
class PulumiResource:
    resource: str
    provider: str
    properties: str
    depends_on: str
    stack_output: str

resources = [
    PulumiResource("VPC", "aws.ec2.Vpc", "cidr_block dns_hostnames tags", "None", "vpc_id"),
    PulumiResource("Subnet", "aws.ec2.Subnet", "vpc_id cidr_block az public_ip", "VPC", "subnet_id"),
    PulumiResource("Security Group", "aws.ec2.SecurityGroup", "vpc_id ingress egress", "VPC", "sg_id"),
    PulumiResource("EC2 Instance", "aws.ec2.Instance", "ami type subnet sg tags", "Subnet SG", "public_ip"),
    PulumiResource("RDS", "aws.rds.Instance", "engine class storage password", "Subnet SG", "endpoint"),
    PulumiResource("S3 Bucket", "aws.s3.Bucket", "acl versioning encryption", "None", "bucket_name"),
]

print("=== Pulumi Resources ===")
for r in resources:
    print(f"  [{r.resource}] Provider: {r.provider}")
    print(f"    Props: {r.properties}")
    print(f"    Depends: {r.depends_on} | Output: {r.stack_output}")

Testing

# === Pulumi Unit Testing ===

# import pulumi
# import pytest
#
# class MyMocks(pulumi.runtime.Mocks):
#     def new_resource(self, args):
#         return [args.name + "_id", args.inputs]
#     def call(self, args):
#         return {}
#
# pulumi.runtime.set_mocks(MyMocks())
#
# from infra import vpc, sg, instance
#
# @pulumi.runtime.test
# def test_vpc_cidr():
#     def check(cidr):
#         assert cidr == "10.0.0.0/16"
#     return vpc.cidr_block.apply(check)
#
# @pulumi.runtime.test
# def test_sg_no_ssh_public():
#     def check(ingress):
#         for rule in ingress:
#             if rule["from_port"] == 22:
#                 assert "0.0.0.0/0" not in rule.get("cidr_blocks", [])
#     return sg.ingress.apply(check)
#
# @pulumi.runtime.test
# def test_instance_has_tags():
#     def check(tags):
#         assert "Name" in tags
#         assert "Environment" in tags
#     return instance.tags.apply(check)

# Policy as Code
# from pulumi_policy import PolicyPack, ResourceValidationPolicy
#
# def no_public_s3(args, report_violation):
#     if args.resource_type == "aws:s3:Bucket":
#         acl = args.props.get("acl", "private")
#         if acl == "public-read":
#             report_violation("S3 Bucket must not be public-read")
#
# PolicyPack("security", policies=[
#     ResourceValidationPolicy(name="no-public-s3", validate=no_public_s3),
# ])

@dataclass
class TestCase:
    test_name: str
    test_type: str
    assertion: str
    tool: str

tests = [
    TestCase("VPC CIDR correct", "Unit", "cidr == 10.0.0.0/16", "pytest + mocks"),
    TestCase("No public SSH", "Policy", "port 22 not open to 0.0.0.0/0", "PolicyPack"),
    TestCase("Instance has tags", "Unit", "Name and Environment tags exist", "pytest + mocks"),
    TestCase("RDS encrypted", "Policy", "storage_encrypted == True", "PolicyPack"),
    TestCase("S3 not public", "Policy", "acl != public-read", "PolicyPack"),
    TestCase("Subnet in correct AZ", "Unit", "az == ap-southeast-1a", "pytest + mocks"),
    TestCase("Full stack deploy", "Integration", "All resources created", "pulumi up --stack test"),
    TestCase("Connectivity test", "Integration", "HTTP 200 from instance", "pytest + requests"),
]

print("\n=== Test Suite ===")
for t in tests:
    print(f"  [{t.test_type}] {t.test_name}")
    print(f"    Assert: {t.assertion} | Tool: {t.tool}")

CI/CD Pipeline

# === Pulumi CI/CD ===

# GitHub Actions
# name: Pulumi Deploy
# on:
#   push: {branches: [main]}
#   pull_request: {branches: [main]}
# jobs:
#   preview:
#     if: github.event_name == 'pull_request'
#     runs-on: ubuntu-latest
#     steps:
#       - uses: actions/checkout@v4
#       - uses: actions/setup-python@v5
#       - run: pip install -r requirements.txt
#       - uses: pulumi/actions@v5
#         with:
#           command: preview
#           stack-name: dev
#         env:
#           PULUMI_ACCESS_TOKEN: }
#
#   deploy:
#     if: github.ref == 'refs/heads/main'
#     runs-on: ubuntu-latest
#     steps:
#       - uses: actions/checkout@v4
#       - uses: pulumi/actions@v5
#         with:
#           command: up
#           stack-name: production
#         env:
#           PULUMI_ACCESS_TOKEN: }

@dataclass
class StackConfig:
    stack: str
    environment: str
    instance_type: str
    db_class: str
    replicas: int
    auto_deploy: bool

stacks = [
    StackConfig("dev", "Development", "t3.micro", "db.t3.micro", 1, True),
    StackConfig("staging", "Staging", "t3.small", "db.t3.small", 2, True),
    StackConfig("production", "Production", "t3.medium", "db.t3.medium", 3, False),
]

print("Stack Configurations:")
for s in stacks:
    auto = "Auto" if s.auto_deploy else "Manual approval"
    print(f"  [{s.stack}] {s.environment}")
    print(f"    EC2: {s.instance_type} | DB: {s.db_class} | Replicas: {s.replicas}")
    print(f"    Deploy: {auto}")

เคล็ดลับ

• Stack: ใช้ Stack แยก dev staging production
• Secret: ใช้ pulumi.Config().require_secret() สำหรับ Password
• Preview: รัน pulumi preview ทุกครั้งก่อน up
• Test: เขียน Unit Test และ Policy Test สำหรับทุก Resource
• Component: สร้าง ComponentResource สำหรับ Reusable Pattern

Pulumi คืออะไร

IaC Platform ภาษาโปรแกรมจริง Python TypeScript Go AWS Azure GCP Kubernetes State Management Pulumi Cloud Loop Condition Function Class

สรุป

Pulumi IaC Infrastructure as Code Python TypeScript AWS Azure GCP Stack Testing Policy CI/CD GitHub Actions Production Deployment