背景
在雲計算日新月異的時代，企業對亞馬遜雲科技資源的使用正在呈現指數級增長。從最初的幾十個資源擴展到數百甚至上萬個，有效管理這些雲資源已成為企業 IT 運營中的關鍵挑戰。亞馬遜雲科技資源標籤（Tags）作為一種元數據機制，已發展成為現代雲資源管理策略的基石。
📢限時插播：Amazon Q Developer 來幫你做應用啦！
🌟10分鐘幫你構建智能番茄鍾應用，1小時搞定新功能拓展、測試優化、文檔注程和部署
⏩快快點擊進入《Agentic Al 幫你做應用 -- 從0到1打造自己的智能番茄鍾》實驗
免費體驗企業級 AI 開發工具的真實效果吧
構建無限，探索啓程!
亞馬遜雲科技標籤不僅是簡單的鍵值對標識，它們在企業雲環境中扮演着多重角色：
- 財務管控的支柱：使財務團隊能夠將雲支出精準歸類到特定業務單元、項目或環境，實現成本透明化
- 安全防線的基礎：允許安全團隊基於資源標籤實施精細的訪問控制策略，增強安全態勢
- 運維效率的催化劑：幫助運維團隊快速定位、篩選和組織管理相關資源，提升日常運維效率
- 自動化流程的驅動力：為自動化腳本和工作流提供目標資源的識別機制，加速 IT 流程自動化
然而，在實際實施過程中，標籤管理往往面臨諸多挑戰：手動操作的低效性，在亞馬遜雲科技控制枱中逐一添加標籤不僅耗時，還易出錯；標籤一致性的難題，不同團隊可能採用不同的命名規範，導致標籤混亂；歷史資源的標籤缺失，早期創建的資源往往沒有按照當前策略打標籤。
這些挑戰在大規模雲遷移項目中尤為突出。例如，在將數百個應用遷移到亞馬遜雲科技的過程中，確保所有資源均帶有“map-xxx”標籤對於跟蹤遷移進度、評估成功率以及後續資源管理至關重要。
本文將介紹一套自動化標籤管理解決方案，幫助組織全面掃描亞馬遜雲科技資源、識別未標記的資源並高效批量應用標籤。該方案不僅適用於遷移場景，也可應用於任何需要優化標籤策略的企業雲環境。
解決方案概述
我們設計的自動化標籤管理工具是一個全面而靈活的端到端解決方案，專為解決客户在亞馬遜雲科技環境中面臨的標籤管理挑戰而設計。該工具採用模塊化架構，包含三個核心組件。可以手動執行 python 代碼，也可以設置觸發器在資源新建立的時候調用代碼，通過觸發器自動打標籤，參考代碼鏈接（僅供參考，不要直接使用生產環境）：
1、服務配置生成
此組件能自動探索亞馬遜雲科技賬户中支持的服務和資源類型，並生成結構化配置文件。它利用 Amazon SDK 智能檢測已啓用的服務，定義各類資源類型、對應的 API 調用方法以及數據提取邏輯，為後續掃描過程奠定基礎。
2、資源掃描
該模塊根據生成的配置文件，高效地遍歷賬户中的所有區域和服務，精確識別缺少特定標籤的資源。採用多線程設計實現並行處理多個區域，大幅提升掃描效率，即使在大型環境中也能保持良好性能。
3、標籤應用
最後一個組件負責將指定標籤批量應用於已識別的資源。它利用 Amazon Resource Groups Tagging API 執行標籤操作，同時生成詳細的執行報告，包括成功、失敗和跳過操作的統計信息，確保標籤應用過程可追蹤。
4、自動化配置（可選）
可以設置觸發器在資源新建立的時候調用代碼，通過觸發器自動打標籤。當用户創建亞馬遜雲科技資源時，API 調用會被 CloudTrail 記錄並通過 EventBridge 觸發 Lambda 函數執行打標籤的邏輯，詳細可參考此鏈接。
服務支持（可以根據自己的需求擴展配置模板）：
- 計算服務：EC2、Lambda、ECS、EKS 等虛擬機和容器服務
- 存儲服務：S3、EBS、EFS 等對象存儲和文件系統
- 數據庫服務：RDS、DynamoDB、Redshift、DocumentDB 等關係型和非關係型數據庫
- 應用服務：SQS、SNS、SES 等應用集成組件
- 安全服務：IAM、KMS、Secrets Manager 等身份與加密管理
- 管理服務：CloudWatch、Systems Manager 等基礎設施管理工具
工作流程
該解決方案提供清晰而靈活的工作流程，如圖所示，可分為以下三個主要步驟：
- 服務配置生成：自動掃描亞馬遜雲科技賬户，創建服務和資源類型映射，識別可標記的資源
- 資源掃描：利用映射信息，識別缺少指定標籤的所有亞馬遜雲科技資源
- 標籤應用：為識別出的資源批量添加預定義標籤
整個流程既支持完全自動化執行，也可分步驟手動實施，允許管理員在應用標籤前審查掃描結果。這種設計適用於一次性標籤修正活動，也適合定期的合規性檢查。還可以通過將代碼改造為 Lambda 函數並結合 EventBridge 調度或者做成定時任務，實現完全自動化的持續標籤管理，細緻流程如下所示：
技術實現
1、服務配置生成
服務配置生成是解決方案的基礎環節。為了有效掃描和標記資源，系統需要了解每種亞馬遜雲科技服務的資源類型及其 API 交互方式。亞馬遜雲科技服務的異構性和複雜性（如不同的 API 結構、資源標識符格式、標籤支持差異）是該環節面臨的主要挑戰。
我們開發的智能腳本 generate_service_config.py 能夠：
- 自動識別賬户中啓用的亞馬遜雲科技服務
- 為每種服務定義資源類型、API 調用方法、ARN 模板和資源 ID 提取邏輯
- 將配置保存為 JSON 格式，供後續模塊使用
該腳本使用預定義的服務模板庫，涵蓋常見亞馬遜雲科技服務的特性，同時具備自適應能力，能根據實際環境動態調整配置。
代碼示例（核心配置生成邏輯）：
#!/usr/bin/env python3
import boto3
import json
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
# 服務模板示例 - EC2服務配置
SERVICE_TEMPLATES = {
'ec2': {
'resource_types': [
{
'type': 'instance',
'list_method': 'describe_instances',
'arn_template': 'arn:aws:ec2:{region}:{account_id}:instance/{id}',
'extract_ids': 'lambda data: [i["InstanceId"] for r in data.get("Reservations", []) for i in r.get("Instances", [])]'
},
{
'type': 'volume',
'list_method': 'describe_volumes',
'arn_template': 'arn:aws:ec2:{region}:{account_id}:volume/{id}',
'extract_ids': 'lambda data: [v["VolumeId"] for v in data.get("Volumes", [])]'
}
# ... 更多資源類型
]
}
# ... 更多服務配置
}
def discover_active_services(region):
"""發現指定區域中活躍的AWS服務"""
active_services = {}
for service_name in POTENTIAL_SERVICES:
try:
client = boto3.client(service_name, region_name=region)
# 檢查是否有實際資源
if service_name in SERVICE_TEMPLATES:
for resource_type in SERVICE_TEMPLATES[service_name]['resource_types']:
try:
list_method = resource_type['list_method']
response = getattr(client, list_method)()
# 提取資源ID
extract_ids = eval(resource_type['extract_ids'])
resource_ids = extract_ids(response)
if resource_ids:
active_services[service_name] = SERVICE_TEMPLATES[service_name]
logger.info(f"發現活躍服務: {service_name} ({len(resource_ids)} 個資源)")
break
except Exception as e:
logger.debug(f"檢查服務 {service_name} 時出錯: {str(e)}")
continue
except Exception as e:
logger.debug(f"服務 {service_name} 不可用: {str(e)}")
continue
return active_services資源掃描是解決方案的核心功能，檢查亞馬遜雲科技環境並識別未標記的資源。
掃描器利用 Amazon Resource Groups Tagging API 集中查詢資源標籤，再結合各服務的專用 API 進行資源枚舉，確保全面發現所有未標記資源。
代碼示例（核心掃描邏輯）：
def get_untagged_resources(account_id, regions=None, services=None, tag_key='xxx-xxxx', max_workers=10):
"""獲取未打指定標籤的資源"""
if regions is None:
regions = get_available_regions()
untagged_resources = {}
# 使用線程池並行處理多個區域
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_region = {
executor.submit(process_region, account_id, region, services, tag_key): region
for region in regions
}
for future in as_completed(future_to_region):
region = future_to_region[future]
try:
region_resources = future.result()
if region_resources:
untagged_resources[region] = region_resources
except Exception as e:
logger.error(f"處理區域 {region} 時出錯: {str(e)}")
return untagged_resources
def process_service(account_id, region, service_name, service_config, tagging, tag_key):
"""處理單個服務的資源掃描"""
service_resources = {}
try:
client = boto3.client(service_config['client'], region_name=region)
for resource_type_config in service_config['resource_types']:
# 獲取資源列表
paginator = client.get_paginator(resource_type_config['list_method'])
all_resources = []
for page in paginator.paginate():
extract_ids = resource_type_config['extract_ids']
if resource_type_config['arn_template'] is None:
# 直接返回ARN的服務
resource_arns = extract_ids(page)
else:
# 需要構建ARN的服務
resource_ids = extract_ids(page)
resource_arns = [
resource_type_config['arn_template'].format(
id=resource_id, region=region, account_id=account_id
) for resource_id in resource_ids
]
all_resources.extend(resource_arns)
# 批量檢查標籤
untagged = []
for i in range(0, len(all_resources), 20):
batch = all_resources[i:i+20]
try:
response = tagging.get_resources(ResourceARNList=batch)
tagged_resources = {
res['ResourceARN']: [tag['Key'] for tag in res.get('Tags', [])]
for res in response.get('ResourceTagMappingList', [])
}
for arn in batch:
if tag_key not in tagged_resources.get(arn, []):
untagged.append(arn)
except ClientError as e:
logger.warning(f"檢查標籤時出錯: {str(e)}")
continue
if untagged:
service_resources[resource_type_config['type']] = untagged
except Exception as e:
logger.error(f"處理服務 {service_name} 時出錯: {str(e)}")
return service_resources使用方法
該解決方案的使用流程直觀且靈活，適合不同規模的亞馬遜雲科技環境：
步驟 1：生成服務配置
python generate_service_config.py --regions us-east-1 us-west-2 --output aws_service_config.json --verbose此命令將掃描指定區域的亞馬遜雲科技服務，並生成詳細的服務配置文件。參數説明：
--regions：要掃描的亞馬遜雲科技區域列表--output：配置文件輸出路徑--verbose：啓用詳細日誌輸出
步驟 2：掃描未標記資源
python add_service_tag.py --tag-key xxx-xxxx --output untagged_resources.json --regions us-east-1 us-west-2此命令將在指定區域中掃描缺少”xxx-xxxx”標籤的資源，並將結果保存到 JSON 文件。參數説明：
--tag-key：要檢查的標籤鍵--output：掃描結果輸出文件--regions：要掃描的亞馬遜雲科技區域
步驟 3：應用標籤
python add_service_tag.py --input untagged_resources.json --tag-value xxxx-xx-xxx --apply此命令將為之前識別的資源應用指定值的標籤。參數説明：
--input：包含未標記資源的輸入文件--tag-value：要應用的標籤值--apply：確認實際應用標籤（不帶此參數則只進行試運行）
完整的工作流程可以從頭至尾一次執行，也可以分步驟進行，讓管理員有機會在各階段審查結果。
注意事項
權限要求
運行此工具的 IAM 用户或角色需具備以下權限：
tag:GetResources和tag:TagResources：用於標籤管理ec2:DescribeRegions：用於獲取區域列表- 各服務的資源列表權限（如
ec2:DescribeInstances）
建議創建專用的 IAM 角色，僅授予必要的最小權限集，可以在執行過程中根據需要自行調整。
性能優化
在大型亞馬遜雲科技環境中優化工具性能的建議：
- 限制掃描範圍：使用
--regions和--services參數縮小掃描範圍 - 調整並行度：根據系統能力修改
max_workers參數 - 分批處理：按區域或服務分別執行掃描和標籤應用
- 優化配置：從服務配置中移除不需要的服務或資源類型（修改預生成的標籤文件）
未來展望
MCP 協議集成：MCP 化改造，面向 AI 驅動的智能化治理
Model Context Protocol（MCP）是 Anthropic 推出的開放協議標準，專為連接 AI 大語言模型與外部系統而設計。MCP 的核心價值在於為 AI 應用提供標準化的工具調用接口，使得智能體能夠動態發現和調用各種外部服務能力。
在雲資源管理領域，MCP 協議的引入意味着：
- 自然語言驅動：用户可以用自然語言描述需求，由 AI 助手理解並執行相應的標籤管理操作
- 智能決策支持：AI 可以分析資源狀態，提供個性化的標籤策略建議
- 動態能力發現：AI 助手能夠自動發現和學習新的標籤管理功能
工具 MCP 化改造的價值
將現有標籤管理工具升級為 MCP 服務後，可以實現以下突破性能力，以下只是建議，非本文核心內容：
1、智能化操作體驗
用户: "標記所有在us-east-1區域、超過30天未使用的EC2實例為'candidate-for-termination'"
AI: 正在掃描us-east-1區域的EC2實例...
發現12個超過30天未使用的實例
正在應用標籤...
操作完成：成功標記12個實例2、上下文感知的智能建議
AI 助手可以基於歷史數據和最佳實踐，提供個性化建議：
- 分析未標記資源的成本影響
- 推薦基於業務場景的標籤策略
- 識別標籤不一致性並建議修復方案
MCP 服務實現
以下是將標籤管理工具封裝為 MCP 服務的核心實現，以下只是代碼示例，可根據自己的需求進行編寫：
from mcp import ClientSession, StdioServerTransport
import json
import asyncio
class AWSTaggingMCPServer:
"""AWS資源標籤管理MCP服務"""
def
__init__
(self):
self.name = "aws-tagging-service"
self.version = "1.0.0"
async def get_tool_definitions(self):
"""返回MCP工具定義"""
return {
"aws-resource-discovery": {
"name": "aws-resource-discovery",
"description": "發現和分析AWS資源的標籤狀態",
"parameters": {
"type": "object",
"properties": {
"regions": {
"type": "array",
"items": {"type": "string"},
"description": "要掃描的AWS區域列表",
"default": ["us-east-1"]
},
"services": {
"type": "array",
"items": {"type": "string"},
"description": "要掃描的AWS服務列表",
"default": ["ec2", "s3", "rds"]
},
"tag_key": {
"type": "string",
"description": "要檢查的標籤鍵",
"default": "Environment"
},
"analysis_type": {
"type": "string",
"enum": ["untagged", "inconsistent", "cost-analysis"],
"description": "分析類型",
"default": "untagged"
}
}
}
},
"aws-tag-application": {
"name": "aws-tag-application",
"description": "為AWS資源批量應用標籤",
"parameters": {
"type": "object",
"properties": {
"resource_arns": {
"type": "array",
"items": {"type": "string"},
"description": "要標記的資源ARN列表"
},
"tags": {
"type": "object",
"description": "要應用的標籤鍵值對",
"additionalProperties": {"type": "string"}
},
"dry_run": {
"type": "boolean",
"description": "是否為預演模式",
"default": true
}
},
"required": ["resource_arns", "tags"]
}
},
"aws-tag-compliance-check": {
"name": "aws-tag-compliance-check",
"description": "檢查資源標籤合規性並生成報告",
"parameters": {
"type": "object",
"properties": {
"compliance_policy": {
"type": "object",
"description": "標籤合規策略定義",
"properties": {
"required_tags": {
"type": "array",
"items": {"type": "string"}
},
"tag_patterns": {
"type": "object",
"additionalProperties": {"type": "string"}
}
}
},
"output_format": {
"type": "string",
"enum": ["json", "csv", "html"],
"default": "json"
}
}
}
}
}
async def execute_tool(self, tool_name: str, parameters: dict):
"""執行MCP工具調用"""
try:
if tool_name == "aws-resource-discovery":
return await self._discover_resources(parameters)
elif tool_name == "aws-tag-application":
return await self._apply_tags(parameters)
elif tool_name == "aws-tag-compliance-check":
return await self._check_compliance(parameters)
else:
return {"error": f"未知工具: {tool_name}"}
except Exception as e:
return {"error": f"執行工具時出錯: {str(e)}"}
async def _discover_resources(self, params):
"""發現和分析資源"""
from add_service_tag import get_untagged_resources, get_account_id
# 調用現有的掃描邏輯
account_id = get_account_id()
untagged = get_untagged_resources(
account_id=account_id,
regions=params.get('regions'),
services=params.get('services'),
tag_key=params.get('tag_key', 'Environment')
)
# 生成智能分析報告
total_resources = sum(
len(arns)
for region_data in untagged.values()
for service_data in region_data.values()
for arns in service_data.values()
)
analysis = {
"summary": {
"total_untagged": total_resources,
"regions_affected": len(untagged),
"services_affected": len({
service for region_data in untagged.values()
for service in region_data.keys()
})
},
"details": untagged,
"recommendations": self._generate_recommendations(untagged)
}
return analysis
async def _apply_tags(self, params):
"""應用標籤到資源"""
from add_service_tag import tag_resources
# 構造資源數據結構
resources_by_region = {}
for arn in params['resource_arns']:
region = self._extract_region_from_arn(arn)
if region not in resources_by_region:
resources_by_region[region] = {'batch': {'resources': []}}
resources_by_region[region]['batch']['resources'].append(arn)
# 應用標籤
results = {}
for region, region_data in resources_by_region.items():
region_results = tag_resources(
resources={region: region_data},
tag_key=list(params['tags'].keys())[0],
tag_value=list(params['tags'].values())[0],
dry_run=params.get('dry_run', True)
)
results[region] = region_results
return {
"operation": "tag_application",
"dry_run": params.get('dry_run', True),
"results": results,
"summary": {
"total_processed": len(params['resource_arns']),
"success_rate": self._calculate_success_rate(results)
}
}
def _generate_recommendations(self, untagged_resources):
"""生成智能建議"""
recommendations = []
# 分析EC2實例
ec2_count = sum(
len(region_data.get('ec2', {}).get('instance', []))
for region_data in untagged_resources.values()
)
if ec2_count > 0:
recommendations.append({
"type": "cost_optimization",
"priority": "high",
"message": f"發現{ec2_count}個未標記EC2實例，建議立即添加Environment和Owner標籤以進行成本歸屬"
})
# 分析S3存儲桶
s3_count = sum(
len(region_data.get('s3', {}).get('bucket', []))
for region_data in untagged_resources.values()
)
if s3_count > 0:
recommendations.append({
"type": "security",
"priority": "medium",
"message": f"發現{s3_count}個未標記S3存儲桶，建議添加Classification和DataOwner標籤以加強數據治理"
})
return recommendations
MCP服務啓動
async def main():
server = AWSTaggingMCPServer()
# 配置MCP傳輸層
transport = StdioServerTransport()
async with ClientSession(transport) as session:
# 註冊工具
tools = await server.get_tool_definitions()
await session.initialize(server.name, server.version, tools)
# 處理工具調用
async for request in session.request_stream():
if request.method == "tools/call":
result = await server.execute_tool(
request.params["name"],
request.params.get("arguments", {})
)
await session.send_result(request.id, result)
if
name
== "
__main__
":
asyncio.run(main())MCP 集成的實際應用場景
場景 1：智能成本分析（僅舉例）
可以配合 Amazon Q CLI，調用成本分析 MCP 服務（Cost Explorer MCP Server）完成成本分析工作，步驟如下：
- 安裝 Amazon Q CLI，安裝步驟見鏈接
- 安裝 MCP 服務，部署官方提供的成本分析 MCP 服務，部署説明見鏈接
- 啓動 MCP 服務後，通過 q chat 調用成本分析服務，如下所示：
🛠️ Using tool: get_today_date from mcp server awslabscost_explorer_mcp_server
⋮
● Running get_today_date with the param:
⋮ {
⋮ "arguments": {},
⋮ "name": "get_today_date"
⋮ }
Allow this action? Use 't' to trust (always allow) this tool for the session. [y/n/t]:分析後的結論如下（部分）：
## 成本可見性對比
### 當前狀態 (部分標記)
總成本: $34.34/月
可分配成本:
├── Training部門 (CostCenter): $11.39 (33.1%)
│ └── ResilienceWorkshop項目: $11.39
└── 未分配成本: $22.95 (66.9%) ❌
└── EKS工作負載: $22.95 (無法追蹤到具體項目)
### 完整標記後的預期狀態
總成本: $34.34/月
可分配成本:
├── Training部門 (CostCenter): $34.34 (100%)
│ ├── ResilienceWorkshop項目: $11.39 (33.1%)
│ └── EKS-Workshop項目: $22.95 (66.9%) ✅
└── 未分配成本: $0.00 (0%) ✅
## 立即行動的ROI
### 標記成本 vs 收益
• **一次性標記工作**: ~20分鐘
• **月度成本節省潛力**: $6.87-$11.45 (20-33%)
• **年度ROI**: 4,100-6,870%
### 具體節省機會
1. EKS節點優化: $6.87/月 (通過正確調度)
2. 自動關停: $4.58/月 (非工作時間)
3. 實例類型優化: $3.44/月 (基於使用模式)
結論: 投入~20分鐘完成標記工作，每月可節省$10-17，年度ROI超過6,000%，6,000%，
同時獲得完整成本可見性和控制力。場景 2：合規性自動檢查（僅舉例）
可以使用本博客提供的代改造成 MCP 服務，配合 Amazon Q CLI 做資源檢查：
用户: "檢查所有資源是否符合公司標籤策略"
AI: 執行合規性檢查...
策略要求: Environment、Owner、Project標籤必須存在
檢查結果:
✅ 合規資源: 1,234個 (78%)
❌ 不合規資源: 345個 (22%)
建議修復措施已生成，是否立即執行標籤修復?總結
亞馬遜雲科技資源標籤對於有效管理雲基礎設施至關重要，但隨着環境規模增長，手動管理標籤變得日益困難。我們開發的自動化標籤管理解決方案為企業提供了強大而靈活的工具，該解決方案適用於多種場景，包括雲遷移項目、成本分配優化等，還可以配合數據庫或者文件存儲做標籤分類，可為企業帶來顯著的運維效率提升和治理能力增強。
*前述特定亞馬遜雲科技生成式人工智能相關的服務目前在亞馬遜雲科技海外區域可用。亞馬遜雲科技中國區域相關雲服務由西雲數據和光環新網運營，具體信息以中國區域官網為準。
本篇作者
本期最新實驗為《Agentic AI 幫你做應用 —— 從0到1打造自己的智能番茄鍾》
✨ 自然語言玩轉命令行，10分鐘幫你構建應用，1小時搞定新功能拓展、測試優化、文檔註釋和部署
💪 免費體驗企業級 AI 開發工具，質量+安全全掌控
⏩️[點擊進入實驗] 即刻開啓 AI 開發之旅
構建無限, 探索啓程！
