构建Labmda函数实现AWS资源自动标签

新钛云服 2024/08/28

本篇文章介绍了一种自动为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"
    ]
  }
}