SiamCafe · Blog
Pulumi IaC Infrastructure as Code — จัดการ
บทความ

Pulumi IaC Infrastructure as Code — จัดการ

เผยแพร่ 28 พฤษภาคม 2569

Pulumi IaC

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

FeaturePulumiTerraformCDKCrossplane
LanguagePython TS Go C#HCLPython TS JavaYAML CRD
StatePulumi Cloud/S3TF Cloud/S3CloudFormationK8s etcd
Testingpytest jestTerratestpytest jestK8s test
Multi-cloudดีมากดีมากAWS onlyดี
Learningง่าย (dev)ปานกลางปานกลางสูง (K8s)
เหมาะกับDeveloper-firstOps/PlatformAWS-heavyK8s-native

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

บทความที่เกี่ยวข้อง