构建Labmda函数实现AWS资源自动标签|key|lambda|object|spiderlinebreak|代码|触发器

新钛云服已累计为您分享809篇技术干货

本篇文章介绍了一种自动为AWS资源打标签的解决方案，当前支持以下资源的自动打标签:

EC2（Elastic Compute Cloud）
EBS（Elastic Block Store）
RDS（Relational Database Service）
S3 bucket
S3 object（需要开启事件记录，默认不开启）
VPC（Virtual Private Cloud）
ELB（Elastic Load Balancer）
Target Groups

程序逻辑

通过CloudTrail追踪创建资源事件，EventBridge根据预设的规则判断并触发Lambda函数。Lambda 函数运行 Python代码进行打标签操作。

整个过程实现自动化，无需人工干涉。这种解决方案不仅提高了资源管理的效率，还能帮助组织更好地进行成本管理和合规性检查。

具体实现

Lambda

创建Lambda资源

将编写好的Lambda函数部署到AWS Lambda服务中。配置EventBridge规则，使Lambda函数能够响应预定义的事件，如EC2实例启动、S3对象创建等。

在Lambda函数首页选择创建函数

Lambda 将创建一个名为 Lambda-AutoTag-role-1n9jg0j9 的执行角色，此角色具有将日志上传到 Amazon CloudWatch Logs 的权限。

设置角色权限

在角色列表中找到Lambda角色

点击权限策略

点击编辑角色权限

添加如下权限

JSON
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": [
"ec2:Describe*",
"ec2:CreateTags*",
"rds:AddTagsToResource",
"elasticloadbalancing:AddTags",
"s3:PutObjectTagging",
"s3:PutBucketTagging",
"logs:CreateLogGroup",
"logs:PutLogEvents",
"logs:CreateLogStream"
],
"Resource": [
"*"
]
}
]
}

添加代码到Lambda

在代码源中新增两个python文件

lambda_function.py

用于根据事件名称判断具体触发事件

Python
from __future__ import print_function
import logging
from autotag import AutoTag

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def lambda_handler(event, context):
logger.info(f"{event=}")
detail = event['detail']
eventname = detail['eventName']
logger.info(f"{eventname=}")


tools = AutoTag(event)
if eventname == 'RunInstances':
# EC2
tools.create_tag_to_ec2()

elif eventname == 'CreateDBInstance':
# RDS
tools.create_tag_to_rds()

elif eventname == 'CreateVpc':
# VPC
tools.create_tag_to_vpc()

elif eventname == 'CreateBucket':
# 给 S3 桶打标签
tools.create_tag_to_s3_bucket()

elif eventname == 'CreateLoadBalancer':
# ELB
tools.create_tag_to_elb()

elif eventname == 'CreateTargetGroup':
# TargetGroup ELB目标组
tools.create_tag_to_target_groups()

elif eventname == 'PutObject':
# 给 S3 中 Object 打标签
tools.create_tag_to_s3_object()


else:
logger.error(f'{eventname=} 未定义的操作')
return False

logger.info("Success!")
    return True

autotag.py

提取资源实例ID，并进行具体的打标签操作

Python
import logging
import boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

class AutoTag:
def __init__(self, event):
self.detail = event['detail']
self.Tag_Key = 'map-migrated'
self.Tag_Value = 'migX1EPYP8W1L'
self.Tags = [{'Key': self.Tag_Key, 'Value': self.Tag_Value}]

def create_tag_to_ec2(self):
ids = []
client = boto3.resource('ec2')

items = self.detail['responseElements']['instancesSet']['items']
for item in items:
ids.append(item['instanceId'])
logger.info(ids)
base = client.instances.filter(InstanceIds=ids)
for instance in base:
for vol in instance.volumes.all():
ids.append(vol.id)
for eni in instance.network_interfaces:
ids.append(eni.id)
if ids:
logger.info(f"对ec2实例列表: {ids}进行打标")
response = client.create_tags(Resources=ids,
Tags=self.Tags)
logger.info(f"打标完成! {response=}")


def create_tag_to_dynamodb(self):
client = boto3.client('dynamodb')
resource_arn = self.detail['responseElements']['tableDescription']['tableArn']
logger.info(f"对dynamodb实例列表: {resource_arn}进行打标")
client.tag_resource(ResourceArn=resource_arn, Tags=self.Tags)


def create_tag_to_lambda(self):
# Lambda 的实际 API 与文档中并不一致, 其组成为 AIP 名字+版本
client = boto3.client('lambda')

function_arn = self.detail['responseElements']['functionArn']
logger.info(f"对lambda实例列表: {function_arn}进行打标")
client.tag_resource(Resource=function_arn, Tags=self.Tags)

def create_tag_to_rds(self):
client = boto3.client('rds')
resource_arn = self.detail['responseElements']['dBInstanceArn']
logger.info(f"对rds实例列表: {resource_arn}进行打标")
client.add_tags_to_resource(ResourceName=resource_arn, Tags=self.Tags)

def create_tag_to_elb(self):
client = boto3.client('elbv2')
resource_arn = self.detail['responseElements']['loadBalancers'][0]['loadBalancerArn']
ids = []
ids.append(resource_arn)
logger.info(f"对elb实例列表: {ids}进行打标")
client.add_tags(ResourceArns=ids, Tags=self.Tags)

def create_tag_to_target_groups(self):
client = boto3.client('elbv2')
resource_arn = self.detail['responseElements']['targetGroups'][0]['targetGroupArn']
ids = []
ids.append(resource_arn)
logger.info(f"对elb实例列表: {ids}进行打标")
client.add_tags(ResourceArns=ids, Tags=self.Tags)

def create_tag_to_s3_object(self):
s3 = boto3.client("s3")
bucket_name = self.detail['requestParameters']['bucketName']
object_name = self.detail['requestParameters']['key']

tags = [{'Key': 'Owner', 'Value': self.user}, {'Key': 'PrincipalId', 'Value': self.principal}]
s3.put_object_tagging(Bucket=bucket_name, Key=object_name, Tagging={'TagSet': tags})

def create_tag_to_s3_bucket(self):
s3 = boto3.client("s3")
bucket_name = self.detail['requestParameters']['bucketName']
logger.info(f"对s3_bucket: {bucket_name}进行打标")
s3.put_bucket_tagging(Bucket=bucket_name, Tagging={'TagSet': self.Tags})

def create_tag_to_vpc(self):
vpc_id = self.detail['responseElements']['vpc']['vpcId']

ec2 = boto3.resource('ec2')
vpc = ec2.Vpc(vpc_id)
logger.info(f"对vpc: {vpc}进行打标")
        vpc.create_tags(DryRun=False, Tags=self.Tags)

修改预设标签值

按需更改以下三个参数即可:

Python
AutoTag.Tag_Key = 'map-migrated'
AutoTag.Tag_Value = 'migX1EPYP8W1L'
AutoTag.Tags = [{'Key': self.Tag_Key, 'Value': self.Tag_Value}]

创建事件跟踪（CloudTrail)

使用 AWS CloudTrail 来跟踪资源创建事件。CloudTrail能够记录所有对AWS 账户的 API 调用，并生成日志文件。确保 CloudTrail 已经启用，并且记录了所有必要的事件。重点在于将事件信息发送到CloudWatch。

进入 CloudTrail 首页

选择并创建跟踪

创建 EventBridge 资源

设置AWS EventBridge规则，根据CloudTrail捕获的资源创建事件来触发特定的Lambda函数。EventBridge可以根据事件模式匹配预设规则，从而准确地选择需要处理的事件。

在 CloudWatch 中点击事件规则

创建规则

规则类型选择具有事件模式的规则

事件模式

选择自定义模式(JSON编辑器)

预设触发规则

将以下自定义规则填入事件模式中

JSON
{
"source": [
"aws.ec2",
"aws.s3",
"aws.rds",
"aws.vpn",
"aws.elasticloadbalancing"
],
"detail-type": ["AWS API Call via CloudTrail"],
"detail": {
"eventSource": [
"ec2.amazonaws.com",
"s3.amazonaws.com",
"rds.amazonaws.com",
"vpn.amazonaws.com",
"elasticloadbalancing.amazonaws.com"
],
"eventName": [
"RunInstances",
"CreateBucket",
"CreateDBInstance",
"CreateVpc",
"CreateLoadBalancer",
"CreateTargetGroup",
"PutObject"
]
}
}

预设目标动作

选择刚创建的Lambda函数