聊聊如何構(gòu)建Labmda函數(shù)實現(xiàn)AWS資源自動標(biāo)簽

作者：王暄植 2024-08-28 08:01:51

Lambda 將創(chuàng)建一個名為 Lambda-AutoTag-role-1n9jg0j9 的執(zhí)行角色，此角色具有將日志上傳到 Amazon CloudWatch Logs 的權(quán)限。

本篇文章介紹了一種自動為AWS資源打標(biāo)簽的解決方案，當(dāng)前支持以下資源的自動打標(biāo)簽:

EC2（Elastic Compute Cloud）
EBS（Elastic Block Store）
RDS（Relational Database Service）
S3 bucket
S3 object（需要開啟事件記錄，默認(rèn)不開啟）
VPC（Virtual Private Cloud）
ELB（Elastic Load Balancer）
Target Groups

程序邏輯

通過CloudTrail追蹤創(chuàng)建資源事件，EventBridge根據(jù)預(yù)設(shè)的規(guī)則判斷并觸發(fā)Lambda函數(shù)。Lambda 函數(shù)運行 Python代碼進行打標(biāo)簽操作。

整個過程實現(xiàn)自動化，無需人工干涉。這種解決方案不僅提高了資源管理的效率，還能幫助組織更好地進行成本管理和合規(guī)性檢查。

具體實現(xiàn)

Lambda

創(chuàng)建Lambda資源

將編寫好的Lambda函數(shù)部署到AWS Lambda服務(wù)中。配置EventBridge規(guī)則，使Lambda函數(shù)能夠響應(yīng)預(yù)定義的事件，如EC2實例啟動、S3對象創(chuàng)建等。

圖片

在Lambda函數(shù)首頁選擇創(chuàng)建函數(shù)；

圖片

Lambda 將創(chuàng)建一個名為 Lambda-AutoTag-role-1n9jg0j9 的執(zhí)行角色，此角色具有將日志上傳到 Amazon CloudWatch Logs 的權(quán)限。

設(shè)置角色權(quán)限

在角色列表中找到Lambda角色；

圖片

點擊權(quán)限策略；

圖片

點擊編輯角色權(quán)限；

圖片

添加如下權(quán)限；

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

用于根據(jù)事件名稱判斷具體觸發(fā)事件

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 桶打標(biāo)簽
        tools.create_tag_to_s3_bucket()


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


    elif eventname == 'CreateTargetGroup':
        # TargetGroup ELB目標(biāo)組
        tools.create_tag_to_target_groups()


    elif eventname == 'PutObject':
        # 給 S3 中 Object 打標(biāo)簽
        tools.create_tag_to_s3_object()




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


    logger.info("Success!")
    return True

autotag.py

提取資源實例ID，并進行具體的打標(biāo)簽操作

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}進行打標(biāo)")
            response = client.create_tags(Resources=ids,
                        Tags=self.Tags)
            logger.info(f"打標(biāo)完成! {respnotallow=}")




    def create_tag_to_dynamodb(self):
        client = boto3.client('dynamodb')
        resource_arn = self.detail['responseElements']['tableDescription']['tableArn']
        logger.info(f"對dynamodb實例列表: {resource_arn}進行打標(biāo)")
        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}進行打標(biāo)")
        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}進行打標(biāo)")
        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}進行打標(biāo)")
        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}進行打標(biāo)")
        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}進行打標(biāo)")
        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}進行打標(biāo)")
        vpc.create_tags(DryRun=False, Tags=self.Tags)

修改預(yù)設(shè)標(biāo)簽值

按需更改以下三個參數(shù)即可:

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

創(chuàng)建事件跟蹤（CloudTrail)

使用 AWS CloudTrail 來跟蹤資源創(chuàng)建事件。CloudTrail能夠記錄所有對AWS 賬戶的 API 調(diào)用，并生成日志文件。確保 CloudTrail 已經(jīng)啟用，并且記錄了所有必要的事件。重點在于將事件信息發(fā)送到CloudWatch。

進入 CloudTrail 首頁

圖片

選擇并創(chuàng)建跟蹤；

創(chuàng)建 EventBridge 資源

設(shè)置AWS EventBridge規(guī)則，根據(jù)CloudTrail捕獲的資源創(chuàng)建事件來觸發(fā)特定的Lambda函數(shù)。EventBridge可以根據(jù)事件模式匹配預(yù)設(shè)規(guī)則，從而準(zhǔn)確地選擇需要處理的事件。

在 CloudWatch 中點擊事件規(guī)則

圖片

創(chuàng)建規(guī)則

規(guī)則類型選擇具有事件模式的規(guī)則

事件模式

選擇自定義模式(JSON編輯器)

預(yù)設(shè)觸發(fā)規(guī)則

將以下自定義規(guī)則填入事件模式中

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"
    ]
  }
}