Pulumi single EC2 instance custom component
The Custom component creates a single AWS EC2 instance includes security group, IAM configuration, etc
from dataclasses import dataclass, field
from typing import Optional, Mapping, Sequence
import pulumi
import pulumi_aws as aws
class EbsVolumeConfig:
volume_id: pulumi.Input[str]
device_name: pulumi.Input[str]
force_detach: Optional[pulumi.Input[bool]] = False
skip_destroy: Optional[pulumi.Input[bool]] = True
stop_instance_before_detaching: Optional[pulumi.Input[bool]] = True
@dataclass
class Ec2InstanceArgs:
ami: pulumi.Input[str]
type: pulumi.Input[str]
key_pair_name: pulumi.Input[str]
name: pulumi.Input[str]
subnet_id: pulumi.Input[str]
vpc_id: pulumi.Input[str]
associate_public_ip_address: Optional[pulumi.Input[bool]] = None
create_eip: Optional[pulumi.Input[bool]] = False
disable_api_stop: Optional[pulumi.Input[bool]] = True
disable_api_termination: Optional[pulumi.Input[bool]] = True
ebs_optimized: Optional[pulumi.Input[bool]] = True
ebs_volumes: Optional[pulumi.Input[Mapping[str, pulumi.Input[EbsVolumeConfig]]]] = None
metadata_http_endpoint: Optional[pulumi.Input[str]] = "enabled"
metadata_http_tokens: Optional[pulumi.Input[str]] = "required"
metadata_http_put_response_hop_limit: Optional[pulumi.Input[int]] = 1
tags: Optional[pulumi.Input[Mapping[str, pulumi.Input[str]]]] = None
monitoring: Optional[pulumi.Input[bool]] = False
placement_group: Optional[pulumi.Input[str]] = None
private_ip: Optional[pulumi.Input[str]] = None
root_volume_kms_key_id: Optional[pulumi.Input[str]] = None
root_volume_size: Optional[pulumi.Input[int]] = 8
root_volume_tags: Optional[pulumi.Input[Mapping[str, pulumi.Input[str]]]] = None
root_volume_type: Optional[pulumi.Input[str]] = "gp3"
root_volume_delete_on_termination: Optional[pulumi.Input[bool]] = True
secondary_private_ips: Optional[pulumi.Input[Sequence[pulumi.Input[str]]]] = None
source_dest_check: Optional[pulumi.Input[bool]] = None
tenancy: Optional[pulumi.Input[str]] = "default"
user_data: Optional[pulumi.Input[str]] = None
user_data_base64: Optional[pulumi.Input[str]] = None
user_data_replace_on_change: Optional[pulumi.Input[bool]] = False
@dataclass
class Ec2IamRoleArgs:
assume_role_principals: Optional[pulumi.Input[Sequence[pulumi.Input[str]]]] = field(default_factory=lambda: ["ec2.amazonaws.com"])
create_instance_profile: Optional[pulumi.Input[bool]] = True
create_role: Optional[pulumi.Input[bool]] = True
description: Optional[pulumi.Input[str]] = None
name: Optional[pulumi.Input[str]] = None
tags: Optional[pulumi.Input[Mapping[str, pulumi.Input[str]]]] = None
@dataclass
class Ec2SecurityGroupArgs:
additional_ids: Optional[pulumi.Input[Sequence[pulumi.Input[str]]]] = None
create_security_group: Optional[pulumi.Input[bool]] = True
description: Optional[pulumi.Input[str]] = None
name: Optional[pulumi.Input[str]] = None
tags: Optional[pulumi.Input[Mapping[str, pulumi.Input[str]]]] = None
@dataclass
class Ec2DnsArgs:
create_record: Optional[pulumi.Input[bool]] = True
name: Optional[pulumi.Input[str]] = None
ttl: Optional[pulumi.Input[int]] = 300
type: Optional[pulumi.Input[str]] = "A"
use_private_ip: Optional[pulumi.Input[bool]] = True
zone_id: Optional[pulumi.Input[str]] = None
zone_domain: Optional[pulumi.Input[str]] = None
class Ec2Instance(pulumi.ComponentResource):
def __init__(self, name, instance_args: Ec2InstanceArgs, role_args: Ec2IamRoleArgs,
security_group_args: Ec2SecurityGroupArgs, dns_args: Ec2DnsArgs, opts=None):
super().__init__('chintai:components:Ec2Server', name, {}, opts)
if instance_args.root_volume_kms_key_id is None:
root_block_kms_key_id = aws.kms.get_key(key_id=aws.ebs.get_default_kms_key().key_arn).arn
else:
root_block_kms_key_id = instance_args.root_volume_kms_key_id
if instance_args.root_volume_tags is None:
root_volume_tags = instance_args.tags
else:
root_volume_tags = instance_args.root_volume_tags
if security_group_args.create_security_group:
if security_group_args.tags is None:
sg_tags = instance_args.tags
else:
sg_tags = security_group_args.tags
instance_sg = aws.ec2.SecurityGroup(
resource_name=f"{name}-sg",
name=security_group_args.name,
description=security_group_args.description if security_group_args.description else f"Security group for {name}",
vpc_id=instance_args.vpc_id,
tags=sg_tags,
)
else:
if security_group_args.name:
instance_sg = aws.ec2.get_security_group(name=security_group_args.name)
else:
raise Exception("Either is_create should be True, or name should be provided")
if role_args.create_role:
service_principal_assume_role_policy_statement = aws.iam.GetPolicyDocumentStatementArgs(
sid="AssumeRole",
actions=[
"sts:AssumeRole"
],
principals=[
aws.iam.GetPolicyDocumentStatementPrincipalArgs(
type="Service",
identifiers=role_args.assume_role_principals
)
],
)
iam_policy_document_assume_role = aws.iam.get_policy_document(
statements=[
service_principal_assume_role_policy_statement
]
)
instance_role = aws.iam.Role(
resource_name=f"{name}-role",
name=role_args.name,
description=role_args.description if role_args.description else f"Role for {name}",
assume_role_policy=iam_policy_document_assume_role.json,
tags=role_args.tags
)
else:
instance_role = aws.iam.get_role(role_args.name)
if role_args.create_instance_profile:
aws.iam.InstanceProfile(
resource_name=f"{name}-instance-profile",
name=role_args.name,
role=instance_role.name
)
instance = aws.ec2.Instance(
resource_name=name,
ami=instance_args.ami,
associate_public_ip_address=instance_args.associate_public_ip_address,
ebs_optimized=instance_args.ebs_optimized,
instance_type=instance_args.type,
user_data=instance_args.user_data,
user_data_base64=instance_args.user_data_base64,
iam_instance_profile=role_args.name,
key_name=instance_args.key_pair_name,
vpc_security_group_ids=[instance_sg.id] + (security_group_args.additional_ids if security_group_args.additional_ids is not None else []),
placement_group=instance_args.placement_group,
disable_api_stop=instance_args.disable_api_stop,
disable_api_termination=instance_args.disable_api_termination,
private_ip=instance_args.private_ip,
subnet_id=instance_args.subnet_id,
metadata_options=aws.ec2.InstanceMetadataOptionsArgs(
http_endpoint=instance_args.metadata_http_endpoint,
http_tokens=instance_args.metadata_http_tokens,
http_put_response_hop_limit=instance_args.metadata_http_put_response_hop_limit
),
monitoring=instance_args.monitoring,
root_block_device=aws.ec2.InstanceRootBlockDeviceArgs(
delete_on_termination=instance_args.root_volume_delete_on_termination,
encrypted=True,
volume_size=instance_args.root_volume_size,
volume_type=instance_args.root_volume_type,
tags={
"Name": f"{instance_args.name}-root-volume" if isinstance(instance_args.name, str) else instance_args.name.apply(
lambda value: f"{value}-root-volume")
} | root_volume_tags,
kms_key_id=root_block_kms_key_id,
),
tenancy=instance_args.tenancy,
credit_specification=aws.ec2.InstanceCreditSpecificationArgs(
cpu_credits="standard"
),
secondary_private_ips=instance_args.secondary_private_ips,
source_dest_check=instance_args.source_dest_check,
user_data_replace_on_change=instance_args.user_data_replace_on_change,
tags={
"Name": instance_args.name,
} | instance_args.tags
)
if instance_args.ebs_volumes is not None:
for volume_name, volume_config in instance_args.ebs_volumes.items():
aws.ec2.VolumeAttachment(
resource_name=f"{name}-volume-attach-{volume_name}",
instance_id=instance.id,
volume_id=volume_config.volume_id,
device_name=volume_config.device_name,
stop_instance_before_detaching=volume_config.stop_instance_before_detaching,
skip_destroy=volume_config.skip_destroy,
force_detach=volume_config.force_detach
)
if instance_args.create_eip:
eip = aws.ec2.Eip(
resource_name=f"{name}-eip",
vpc=True,
tags={
"Name": f"{instance_args.name}-eip",
} | instance_args.tags
)
aws.ec2.EipAssociation(
resource_name=f"{name}-eip-assoc",
instance_id=instance.id,
allocation_id=eip.id
)
if dns_args.create_record:
if dns_args.zone_id is not None:
dns_zone_id = dns_args.zone_id
elif dns_args.zone_domain is not None:
dns_zone_id = aws.route53.get_zone(name=dns_args.zone_domain).id
else:
raise Exception("Either zone_id or zone_name should be provided")
aws.route53.Record(
resource_name=f"aws-route53-record-instance-{dns_args.type}",
zone_id=dns_zone_id,
name=dns_args.name,
type=dns_args.type,
ttl=dns_args.ttl,
records=[
instance.private_ip if dns_args.use_private_ip else instance.public_ip
]
)
self.ami = instance.ami
self.arn = instance.arn
self.id = instance.id
self.private_ip = instance.private_ip
self.public_ip = instance.public_ip
self.secondary_private_ips = instance.secondary_private_ips
self.iam_role_arn = instance_role.arn
self.iam_role_id = instance_role.id
self.iam_role_name = instance_role.name
self.security_group_arn = instance_sg.arn
self.security_group_id = instance_sg.id
self.security_group_name = instance_sg.name
self.register_outputs({})