基于 AWS DevOps Agent 构建 AI 驱动的运维分析系统

aws lambda create-function \
  --function-name devops-agent-trigger-investigation \
  --runtime python3.12 \
  --handler lambda_a.lambda_handler \
  --role "arn:aws:iam::${AWS_ACCOUNT_ID}:role/DevOpsAgentDemoLambdaRole" \
  --zip-file fileb://lambda/lambda_a.zip \
  --timeout 30 --memory-size 128 \
  --layers "${LAYER_ARN}" \
  --environment "Variables={DEVOPS_AGENT_SPACE_ID=${DEVOPS_AGENT_SPACE_ID}}"

import json
import os
import boto3

DEVOPS_AGENT_SPACE_ID = os.environ["DEVOPS_AGENT_SPACE_ID"]

def lambda_handler(event, context):
    detail = event.get("detail", {})
    state_value = detail.get("state", {}).get("value", "")

    # 仅在 ALARM 状态时触发调查
    if state_value != "ALARM":
        return {"statusCode": 200, "body": f"Skipped: state={state_value}"}

    # 提取告警信息
    alarm_name = detail.get("alarmName", "Unknown")
    reason = detail.get("state", {}).get("reason", "N/A")
    metrics = detail.get("configuration", {}).get("metrics", [])
    instance_id = ""
    if metrics:
        dims = metrics[0].get("metricStat", {}).get("metric", {}).get("dimensions", {})
        instance_id = dims.get("InstanceId", "")

    # 调用 DevOps Agent 创建调查任务
    client = boto3.client("devops-agent", region_name=os.environ.get("AWS_REGION", "us-west-2"))
    response = client.create_backlog_task(
        agentSpaceId=DEVOPS_AGENT_SPACE_ID,
        taskType="INVESTIGATION",           # 任务类型：调查
        title=f"Investigate: {alarm_name} - {instance_id}",
        priority="HIGH",                    # 优先级
        description=f"CloudWatch Alarm '{alarm_name}' triggered.\n"
                    f"EC2 Instance: {instance_id}\n"
                    f"Reason: {reason}\n"
                    f"Please investigate the root cause.",
    )

    task = response["task"]
    # 返回 taskId 和 executionId，调查已异步启动
    return {"statusCode": 200, "body": json.dumps({
        "taskId": task["taskId"],
        "executionId": task["executionId"],
        "status": task["status"],  # PENDING_START
    })}

aws lambda create-function \
  --function-name devops-agent-notify-feishu \
  --runtime python3.12 \
  --handler lambda_b.lambda_handler \
  --role "arn:aws:iam::${AWS_ACCOUNT_ID}:role/DevOpsAgentDemoLambdaRole" \
  --zip-file fileb://lambda/lambda_b.zip \
  --timeout 60 --memory-size 128 \
  --layers "${LAYER_ARN}" \
  --environment "Variables={DEVOPS_AGENT_SPACE_ID=${DEVOPS_AGENT_SPACE_ID},FEISHU_APP_ID=${FEISHU_APP_ID},FEISHU_APP_SECRET=${FEISHU_APP_SECRET},FEISHU_CHAT_ID=${FEISHU_CHAT_ID}}"

def get_investigation_summary(agent_space_id, execution_id):
    """从 DevOps Agent 获取调查摘要"""
    client = boto3.client("devops-agent", region_name="us-west-2")
    response = client.list_journal_records(
        agentSpaceId=agent_space_id,
        executionId=execution_id,
    )

    # 在 records 中找 investigation_summary_md 类型（Markdown 格式）
    for record in response.get("records", []):
        if record.get("recordType") == "investigation_summary_md":
            return record.get("content", "")

    return "No investigation summary available."


def lambda_handler(event, context):
    detail = event.get("detail", {})
    metadata = detail.get("metadata", {})
    agent_space_id = metadata["agent_space_id"]
    execution_id = metadata["execution_id"]

    # 获取调查摘要
    summary = get_investigation_summary(agent_space_id, execution_id)

    # 发送飞书通知（富文本消息）
    token = get_tenant_access_token()
    send_feishu_message(token, FEISHU_CHAT_ID, "post", {
        "zh_cn": {
            "title": "DevOps Agent Investigation Completed",
            "content": [
                [{"tag": "text", "text": f"Task: {metadata['task_id']}"}],
                [{"tag": "text", "text": summary}],
            ]
        }
    })