構建Labmda函式實現AWS資源自動標籤

新鈦雲服已累計為您分享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

用於根據事件名稱判斷具體觸發事件

Pythonfrom __future__ import print_functionimport loggingfrom autotag import AutoTaglogger = 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，並進行具體的打標籤操作

Pythonimport loggingimport boto3logger = 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)

修改預設標籤值

按需更改以下三個引數即可:

PythonAutoTag.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函式